external/ton
1
0
Fork 0
mirror of https://github.com/ton-blockchain/ton synced 2025-03-09 15:40:10 +00:00
Code Issues Releases Wiki Activity

HTTP tunnel in rldp-http-proxy

Browse source
This commit is contained in:
SpyCheese 2022-06-27 19:38:11 +03:00
parent d11580dfb3
commit c55b6f84a5
7 changed files with 432 additions and 100 deletions
Download patch file Download diff file Expand all files Collapse all files

35
http/http-connection.cpp
View file

@ -58,9 +58,7 @@ void HttpConnection::loop() {
}
TRY_STATUS(buffered_fd_.flush_write());
if (writing_payload_ && buffered_fd_.left_unwritten() < fd_high_watermark()) {
auto w = buffered_fd_.left_unwritten();
continue_payload_write();
written = buffered_fd_.left_unwritten() > w;
written = continue_payload_write();
}
if (close_after_write_ && !writing_payload_ && !buffered_fd_.left_unwritten()) {
LOG(INFO) << "close after write";
@ -120,7 +118,7 @@ void HttpConnection::write_payload(std::shared_ptr<HttpPayload> payload) {
class Cb : public HttpPayload::Callback {
public:
Cb(td::actor::ActorId<HttpConnection> conn) : conn_(conn) {
Cb(td::actor::ActorId<HttpConnection> conn, size_t watermark) : conn_(conn), watermark_(watermark) {
}
void run(size_t ready_bytes) override {
if (!reached_ && ready_bytes >= watermark_) {
@ -135,23 +133,23 @@ void HttpConnection::write_payload(std::shared_ptr<HttpPayload> payload) {
}
private:
size_t watermark_ = chunk_size();
bool reached_ = false;
td::actor::ActorId<HttpConnection> conn_;
size_t watermark_;
bool reached_ = false;
};
writing_payload_->add_callback(std::make_unique<Cb>(actor_id(this)));
writing_payload_->add_callback(std::make_unique<Cb>(
actor_id(this), writing_payload_->payload_type() == HttpPayload::PayloadType::pt_tunnel ? 1 : chunk_size()));
continue_payload_write();
}
void HttpConnection::continue_payload_write() {
bool HttpConnection::continue_payload_write() {
if (!writing_payload_) {
return;
return false;
}
if (writing_payload_->is_error()) {
stop();
return;
return false;
}
auto t = writing_payload_->payload_type();
@ -159,19 +157,24 @@ void HttpConnection::continue_payload_write() {
t = HttpPayload::PayloadType::pt_chunked;
}
bool wrote = false;
while (!writing_payload_->written()) {
if (buffered_fd_.left_unwritten() > fd_high_watermark()) {
return;
return wrote;
}
if (!writing_payload_->parse_completed() && writing_payload_->ready_bytes() < chunk_size()) {
return;
bool is_tunnel = writing_payload_->payload_type() == HttpPayload::PayloadType::pt_tunnel;
if (!is_tunnel && !writing_payload_->parse_completed() && writing_payload_->ready_bytes() < chunk_size()) {
return wrote;
}
writing_payload_->store_http(buffered_fd_.output_buffer(), chunk_size(), t);
if (is_tunnel && writing_payload_->ready_bytes() == 0) {
return wrote;
}
wrote |= writing_payload_->store_http(buffered_fd_.output_buffer(), chunk_size(), t);
}
if (writing_payload_->parse_completed() && writing_payload_->written()) {
payload_written();
return;
}
return wrote;
}
td::Status HttpConnection::read_payload(HttpResponse *response) {

4
http/http-connection.h
View file

@ -65,9 +65,7 @@ class HttpConnection : public td::actor::Actor, public td::ObserverBase {
void send_request(std::unique_ptr<HttpRequest> request, std::shared_ptr<HttpPayload> payload);
void send_response(std::unique_ptr<HttpResponse> response, std::shared_ptr<HttpPayload> payload);
void write_payload(std::shared_ptr<HttpPayload> payload);
void continue_payload_write();
td::Status receive_request();
td::Status receive_response();
bool continue_payload_write();
td::Status read_payload(HttpRequest *request);
td::Status read_payload(HttpResponse *response);
td::Status read_payload(std::shared_ptr<HttpPayload> payload);

3
http/http-inbound-connection.h
View file

@ -35,7 +35,8 @@ class HttpInboundConnection : public HttpConnection {
td::Status receive_eof() override {
found_eof_ = true;
if (reading_payload_) {
if (reading_payload_->payload_type() != HttpPayload::PayloadType::pt_eof) {
if (reading_payload_->payload_type() != HttpPayload::PayloadType::pt_eof &&
reading_payload_->payload_type() != HttpPayload::PayloadType::pt_tunnel) {
return td::Status::Error("unexpected EOF");
} else {
reading_payload_->complete_parse();

3
http/http-outbound-connection.h
View file

@ -44,7 +44,8 @@ class HttpOutboundConnection : public HttpConnection {
td::Status receive_eof() override {
found_eof_ = true;
if (reading_payload_) {
if (reading_payload_->payload_type() != HttpPayload::PayloadType::pt_eof) {
if (reading_payload_->payload_type() != HttpPayload::PayloadType::pt_eof &&
reading_payload_->payload_type() != HttpPayload::PayloadType::pt_tunnel) {
return td::Status::Error("unexpected EOF");
} else {
LOG(INFO) << "stopping (EOF payload)";

55
http/http.cpp
View file

@ -164,6 +164,8 @@ td::Result<std::shared_ptr<HttpPayload>> HttpRequest::create_empty_payload() {
if (!need_payload()) {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_empty);
} else if (method_ == "CONNECT") {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_tunnel, low_watermark(), high_watermark());
} else if (found_content_length_) {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_content_length, low_watermark(), high_watermark(),
content_length_);
@ -175,7 +177,7 @@ td::Result<std::shared_ptr<HttpPayload>> HttpRequest::create_empty_payload() {
}
bool HttpRequest::need_payload() const {
return found_content_length_ || found_transfer_encoding_;
return found_content_length_ || found_transfer_encoding_ || method_ == "CONNECT";
}
td::Status HttpRequest::add_header(HttpHeader header) {
@ -284,7 +286,8 @@ td::Status HttpPayload::parse(td::ChainBufferReader &input) {
case PayloadType::pt_empty:
UNREACHABLE();
case PayloadType::pt_eof:
cur_chunk_size_ = 1 << 30;
case PayloadType::pt_tunnel:
cur_chunk_size_ = 1LL << 60;
break;
case PayloadType::pt_chunked:
state_ = ParseState::reading_crlf;
@ -480,17 +483,18 @@ void HttpPayload::run_callbacks() {
}
}
void HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, HttpPayload::PayloadType store_type) {
bool HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, HttpPayload::PayloadType store_type) {
if (store_type == PayloadType::pt_empty) {
return;
return false;
}
slice_gc();
bool wrote = false;
while (chunks_.size() > 0 && max_size > 0) {
auto cur_state = state_.load(std::memory_order_consume);
auto s = get_slice(max_size);
if (s.size() == 0) {
if (cur_state != ParseState::reading_trailer && cur_state != ParseState::completed) {
return;
return wrote;
} else {
break;
}
@ -500,28 +504,33 @@ void HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
if (store_type == PayloadType::pt_chunked) {
char buf[64];
::sprintf(buf, "%lx\r\n", s.size());
output.append(td::Slice(buf, strlen(buf)));
auto slice = td::Slice(buf, strlen(buf));
wrote |= !slice.empty();
output.append(slice);
}
wrote |= !s.empty();
output.append(std::move(s));
if (store_type == PayloadType::pt_chunked) {
output.append(td::Slice("\r\n", 2));
wrote = true;
}
}
if (chunks_.size() != 0) {
return;
return wrote;
}
if (!written_zero_chunk_) {
if (store_type == PayloadType::pt_chunked) {
output.append(td::Slice("0\r\n", 3));
wrote = true;
}
written_zero_chunk_ = true;
}
if (store_type != PayloadType::pt_chunked) {
written_trailer_ = true;
return;
return wrote;
}
while (max_size > 0) {
@ -529,15 +538,16 @@ void HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
HttpHeader h = get_header();
if (h.empty()) {
if (cur_state != ParseState::completed) {
return;
return wrote;
} else {
break;
}
}
auto s = h.size();
h.store_http(output);
wrote = true;
if (max_size <= s) {
return;
return wrote;
}
max_size -= s;
}
@ -545,7 +555,9 @@ void HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
if (!written_trailer_) {
output.append(td::Slice("\r\n", 2));
written_trailer_ = true;
wrote = true;
}
return wrote;
}
tl_object_ptr<ton_api::http_payloadPart> HttpPayload::store_tl(size_t max_size) {
@ -729,17 +741,18 @@ td::Result<std::unique_ptr<HttpResponse>> HttpResponse::parse(std::unique_ptr<Ht
}
HttpResponse::HttpResponse(std::string proto_version, td::uint32 code, std::string reason, bool force_no_payload,
bool keep_alive)
bool keep_alive, bool is_tunnel)
: proto_version_(std::move(proto_version))
, code_(code)
, reason_(std::move(reason))
, force_no_payload_(force_no_payload)
, force_no_keep_alive_(!keep_alive) {
, force_no_keep_alive_(!keep_alive)
, is_tunnel_(is_tunnel) {
}
td::Result<std::unique_ptr<HttpResponse>> HttpResponse::create(std::string proto_version, td::uint32 code,
std::string reason, bool force_no_payload,
bool keep_alive) {
bool keep_alive, bool is_tunnel) {
if (proto_version != "HTTP/1.0" && proto_version != "HTTP/1.1") {
return td::Status::Error(PSTRING() << "unsupported http version '" << proto_version << "'");
}
@ -749,7 +762,7 @@ td::Result<std::unique_ptr<HttpResponse>> HttpResponse::create(std::string proto
}
return std::make_unique<HttpResponse>(std::move(proto_version), code, std::move(reason), force_no_payload,
keep_alive);
keep_alive, is_tunnel);
}
td::Status HttpResponse::complete_parse_header() {
@ -767,6 +780,8 @@ td::Result<std::shared_ptr<HttpPayload>> HttpResponse::create_empty_payload() {
if (!need_payload()) {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_empty);
} else if (is_tunnel_) {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_tunnel, low_watermark(), high_watermark());
} else if (found_content_length_) {
return std::make_shared<HttpPayload>(HttpPayload::PayloadType::pt_content_length, low_watermark(), high_watermark(),
content_length_);
@ -828,10 +843,12 @@ void HttpResponse::store_http(td::ChainBufferWriter &output) {
for (auto &x : options_) {
x.store_http(output);
}
if (keep_alive_) {
HttpHeader{"Connection", "Keep-Alive"}.store_http(output);
} else {
HttpHeader{"Connection", "Close"}.store_http(output);
if (!is_tunnel_) {
if (keep_alive_) {
HttpHeader{"Connection", "Keep-Alive"}.store_http(output);
} else {
HttpHeader{"Connection", "Close"}.store_http(output);
}
}
output.append(td::Slice("\r\n", 2));
}
@ -893,7 +910,9 @@ void answer_error(HttpStatusCode code, std::string reason,
}
auto response = HttpResponse::create("HTTP/1.0", code, reason, false, false).move_as_ok();
response->add_header(HttpHeader{"Content-Length", "0"});
response->complete_parse_header();
auto payload = response->create_empty_payload().move_as_ok();
payload->complete_parse();
CHECK(payload->parse_completed());
promise.set_value(std::make_pair(std::move(response), std::move(payload)));
}

19
http/http.h
View file

@ -64,7 +64,7 @@ td::Result<HttpHeader> get_header(std::string line);
class HttpPayload {
public:
enum class PayloadType { pt_empty, pt_eof, pt_chunked, pt_content_length };
enum class PayloadType { pt_empty, pt_eof, pt_chunked, pt_content_length, pt_tunnel };
HttpPayload(PayloadType t, size_t low_watermark, size_t high_watermark, td::uint64 size)
: type_(t), low_watermark_(low_watermark), high_watermark_(high_watermark), cur_chunk_size_(size) {
CHECK(t == PayloadType::pt_content_length);
@ -75,17 +75,15 @@ class HttpPayload {
CHECK(t != PayloadType::pt_content_length);
CHECK(t != PayloadType::pt_empty);
switch (t) {
case PayloadType::pt_empty:
UNREACHABLE();
case PayloadType::pt_eof:
case PayloadType::pt_tunnel:
state_ = ParseState::reading_chunk_data;
break;
case PayloadType::pt_chunked:
state_ = ParseState::reading_chunk_header;
break;
case PayloadType::pt_content_length:
state_ = ParseState::reading_chunk_data;
break;
default:
UNREACHABLE();
}
}
HttpPayload(PayloadType t) : type_(t) {
@ -136,7 +134,7 @@ class HttpPayload {
void slice_gc();
HttpHeader get_header();
void store_http(td::ChainBufferWriter &output, size_t max_size, HttpPayload::PayloadType store_type);
bool store_http(td::ChainBufferWriter &output, size_t max_size, HttpPayload::PayloadType store_type);
tl_object_ptr<ton_api::http_payloadPart> store_tl(size_t max_size);
bool written() const {
@ -267,9 +265,11 @@ class HttpResponse {
}
static td::Result<std::unique_ptr<HttpResponse>> create(std::string proto_version, td::uint32 code,
std::string reason, bool force_no_payload, bool keep_alive);
std::string reason, bool force_no_payload, bool keep_alive,
bool is_tunnel = false);
HttpResponse(std::string proto_version, td::uint32 code, std::string reason, bool force_no_payload, bool keep_alive);
HttpResponse(std::string proto_version, td::uint32 code, std::string reason, bool force_no_payload, bool keep_alive,
bool is_tunnel = false);
bool check_parse_header_completed() const;
bool keep_alive() const {
@ -323,6 +323,7 @@ class HttpResponse {
bool keep_alive_ = false;
std::vector<HttpHeader> options_;
bool is_tunnel_ = false;
};
void answer_error(HttpStatusCode code, std::string reason,