From 157191b770d4d2b56db8a6dd91fbcb8bf09f6d63 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Fri, 30 Jan 2026 00:48:56 +0300 Subject: [PATCH] Implement HTTP server support O_o --- src/client/etcd_state_client.cpp | 8 +- src/client/http_client.cpp | 222 ++++++++++++++++++++++++------- src/client/http_client.h | 14 +- 3 files changed, 191 insertions(+), 53 deletions(-) diff --git a/src/client/etcd_state_client.cpp b/src/client/etcd_state_client.cpp index dadc6479..9e6dd3f6 100644 --- a/src/client/etcd_state_client.cpp +++ b/src/client/etcd_state_client.cpp @@ -82,7 +82,7 @@ http_context_t *etcd_state_client_t::get_http_ctx() if (!http_ctx) { std::string error; - http_ctx = http_context_init(etcd_client_cert, etcd_client_key, etcd_ca, error); + http_ctx = http_context_init(etcd_client_cert, etcd_client_key, etcd_ca, true, error); if (!http_ctx) { fprintf(stderr, "Failed to initialize HTTP context: %s\n", error.c_str()); @@ -112,7 +112,7 @@ void etcd_state_client_t::etcd_call_oneshot(std::string etcd_address, std::strin "Connection: close\r\n" "\r\n"+req; auto http_cli = http_init(tfd, get_http_ctx()); - auto cb = [http_cli, callback](const http_message_t *response) + auto cb = [http_cli, callback](http_message_t *response) { std::string err; json11::Json data; @@ -152,7 +152,7 @@ void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int t "\r\n"+req; retries--; auto cb = [this, api, payload, timeout, retries, interval, callback, - cur_addr = selected_etcd_address](const http_message_t *response) + cur_addr = selected_etcd_address](http_message_t *response) { std::string err; json11::Json data; @@ -386,7 +386,7 @@ void etcd_state_client_t::start_etcd_watcher() else http_close(etcd_watch_ws); open_websocket(etcd_watch_ws, etcd_address, etcd_api_path+"/watch", { .timeout = etcd_slow_timeout, .ssl = ssl }, - [this, cur_addr = selected_etcd_address](const http_message_t *msg) + [this, cur_addr = selected_etcd_address](http_message_t *msg) { if (msg->body.length()) { diff --git a/src/client/http_client.cpp b/src/client/http_client.cpp index 2f94d449..ec82f484 100644 --- a/src/client/http_client.cpp +++ b/src/client/http_client.cpp @@ -32,7 +32,7 @@ static std::string ws_format_frame(int type, uint64_t size); static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res); -static void parse_http_headers(std::string & res, http_message_t *parsed); +static void parse_http_headers(std::string & res, http_message_t *parsed, bool is_request); struct http_context_t { @@ -56,6 +56,14 @@ struct http_context_t } }; +struct http_call_t +{ + std::string host; + std::string request; + http_options_t options; + std::function cb; +}; + struct http_co_t { http_context_t *ctx = NULL; @@ -65,7 +73,7 @@ struct http_co_t #endif timerfd_manager_t *tfd; - std::function response_callback; + std::function response_callback; int request_timeout = 0; bool ssl = false; @@ -76,7 +84,7 @@ struct http_co_t bool want_streaming; bool keepalive; - std::vector> keepalive_queue; + std::vector keepalive_queue; int state = 0; std::string connected_host; @@ -112,8 +120,9 @@ struct http_co_t void on_ssl_error(int res); #endif void post_message(uint8_t type, const std::string & msg); + void reply(const std::string & msg); void send_request(const std::string & host, const std::string & request, - const http_options_t & options, std::function response_callback); + const http_options_t & options, std::function response_callback); }; #define HTTP_CO_CLOSED 0 @@ -124,10 +133,14 @@ struct http_co_t #define HTTP_CO_WEBSOCKET 5 #define HTTP_CO_CHUNKED 6 #define HTTP_CO_KEEPALIVE 7 +#define HTTP_CO_SERVER 8 +#define HTTP_CO_REQ_HDR_RECEIVED 9 +#define HTTP_CO_REQUEST_RECEIVED 10 #define DEFAULT_TIMEOUT 5000 -http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, const std::string & ssl_ca, std::string & error) +http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, + const std::string & ssl_ca, bool verify_peer, std::string & error) { http_context_t *ctx = new http_context_t; #ifdef WITH_OPENSSL @@ -138,7 +151,7 @@ http_context_t* http_context_init(const std::string & ssl_cert, const std::strin ctx->ssl_ctx = ssl_ctx; if (!ssl_ctx) goto init_err; - SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER, NULL); + SSL_CTX_set_verify(ssl_ctx, verify_peer ? SSL_VERIFY_PEER : SSL_VERIFY_NONE, NULL); if (!SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_2_VERSION)) goto init_err; if ((ssl_ca != "") @@ -172,7 +185,7 @@ http_co_t *http_init(timerfd_manager_t *tfd, http_context_t *ctx) } void open_websocket(http_co_t *handler, const std::string & host, const std::string & path, - const http_options_t & options, std::function response_callback) + const http_options_t & options, std::function response_callback) { if (handler->state == HTTP_CO_KEEPALIVE && (handler->connected_host != host || handler->ssl != options.ssl)) handler->close_connection(); @@ -200,15 +213,39 @@ void open_websocket(http_co_t *handler, const std::string & host, const std::str } void http_request(http_co_t *handler, const std::string & host, const std::string & request, - const http_options_t & options, std::function response_callback) + const http_options_t & options, std::function response_callback) { handler->send_request(host, request, options, response_callback); } +void http_serve(http_co_t *handler, int peer_fd, const http_options_t & options, std::function request_callback) +{ + if (handler->state != HTTP_CO_SERVER || handler->peer_fd != peer_fd) + handler->close_connection(); + handler->host = ""; + handler->request_timeout = options.timeout < 0 ? -1 : (options.timeout == 0 ? DEFAULT_TIMEOUT : options.timeout); + handler->want_streaming = false; + handler->keepalive = false; + handler->ssl = options.ssl; + handler->request = ""; + handler->response_callback = request_callback; + handler->ws_outbox = ""; + handler->response = ""; + handler->sent = 0; + handler->parsed = {}; + handler->peer_fd = peer_fd; + handler->state = HTTP_CO_SERVER; + handler->tfd->set_fd_handler(peer_fd, false, [handler](int peer_fd, int epoll_events) + { + handler->epoll_events |= epoll_events; + handler->handle_events(); + }); +} + void http_co_t::run_cb_and_clear() { parsed.eof = true; - std::function cb; + std::function cb; cb.swap(response_callback); // Call callback after clearing it because otherwise we may hit reenterability problems if (cb != NULL) @@ -217,7 +254,7 @@ void http_co_t::run_cb_and_clear() } void http_co_t::send_request(const std::string & host, const std::string & request, - const http_options_t & options, std::function response_callback) + const http_options_t & options, std::function response_callback) { stackin(); if (state == HTTP_CO_WEBSOCKET) @@ -227,10 +264,7 @@ void http_co_t::send_request(const std::string & host, const std::string & reque } else if (state != HTTP_CO_KEEPALIVE && state != HTTP_CO_CLOSED) { - keepalive_queue.push_back([this, host, request, options, response_callback]() - { - this->send_request(host, request, options, response_callback); - }); + keepalive_queue.emplace_back((http_call_t){ host, request, options, std::move(response_callback) }); stackout(); return; } @@ -273,7 +307,7 @@ void http_co_t::send_request(const std::string & host, const std::string & reque else { close_connection(); - parsed = { .error = "HTTP request timed out" }; + parsed = { .error = "HTTP request timed out", .status_code = ETIMEDOUT }; run_cb_and_clear(); } stackout(); @@ -287,6 +321,11 @@ void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg handler->post_message(type, msg); } +void http_reply(http_co_t *handler, const std::string & reply) +{ + handler->reply(reply); +} + void http_co_t::post_message(uint8_t type, const std::string & msg) { stackin(); @@ -296,7 +335,8 @@ void http_co_t::post_message(uint8_t type, const std::string & msg) request += msg; submit_send(); } - else if (state == HTTP_CO_KEEPALIVE || state == HTTP_CO_CHUNKED) + else if (state == HTTP_CO_KEEPALIVE || state == HTTP_CO_CHUNKED || + state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED) { throw std::runtime_error("Attempt to send websocket message on a regular HTTP connection"); } @@ -308,6 +348,18 @@ void http_co_t::post_message(uint8_t type, const std::string & msg) stackout(); } +void http_co_t::reply(const std::string & reply) +{ + stackin(); + if (state != HTTP_CO_REQUEST_RECEIVED) + { + throw std::runtime_error("Attempt to send HTTP response in invalid connection state"); + } + request += reply; + submit_send(); + stackout(); +} + void http_destroy(http_co_t *handler) { handler->end(); @@ -392,7 +444,7 @@ void http_co_t::start_ws_connection() if (state != HTTP_CO_WEBSOCKET) { close_connection(); - parsed = { .error = "Websocket connection timed out" }; + parsed = { .error = "Websocket connection timed out", .status_code = ETIMEDOUT }; run_cb_and_clear(); } stackout(); @@ -408,7 +460,7 @@ void http_co_t::start_connection() if (!string_to_addr(host.c_str(), 1, 80, &addr)) { close_connection(); - parsed = { .error = "Invalid address: "+host }; + parsed = { .error = "Invalid address: "+host, .status_code = EINVAL }; run_cb_and_clear(); stackout(); return; @@ -417,7 +469,7 @@ void http_co_t::start_connection() if (peer_fd < 0) { close_connection(); - parsed = { .error = std::string("socket: ")+strerror(errno) }; + parsed = { .error = std::string("socket: ")+strerror(errno), .status_code = errno }; run_cb_and_clear(); stackout(); return; @@ -466,7 +518,7 @@ init_err: if (r < 0 && errno != EINPROGRESS) { close_connection(); - parsed = { .error = std::string("connect: ")+strerror(errno) }; + parsed = { .error = std::string("connect: ")+strerror(errno), .status_code = errno }; run_cb_and_clear(); stackout(); return; @@ -501,6 +553,8 @@ void http_co_t::handle_events() { if (state == HTTP_CO_HEADERS_RECEIVED) std::swap(parsed.body, response); + else if (state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED) + parsed = { .error = "client has disconnected normally" }; close_connection(); run_cb_and_clear(); break; @@ -522,7 +576,7 @@ void http_co_t::handle_connect_result() if (result != 0) { close_connection(); - parsed = { .error = std::string("connect: ")+strerror(result) }; + parsed = { .error = std::string("connect: ")+strerror(result), .status_code = result }; run_cb_and_clear(); stackout(); return; @@ -587,13 +641,33 @@ again: else if (res < 0) { close_connection(); - parsed = { .error = std::string("sendmsg: ")+strerror(errno) }; + parsed = { .error = std::string("sendmsg: ")+strerror(errno), .status_code = errno }; run_cb_and_clear(); stackout(); return; } sent += res; - if (state == HTTP_CO_SENDING_REQUEST) + if (state == HTTP_CO_REQUEST_RECEIVED) + { + if (sent >= request.size()) + { + if (!keepalive) + { + close_connection(); + parsed = { .error = "connection is not keep-alive" }; + run_cb_and_clear(); + stackout(); + return; + } + state = HTTP_CO_SERVER; + request = ""; + sent = 0; + } + else + goto again; + handle_read(); + } + else if (state == HTTP_CO_SENDING_REQUEST) { if (sent >= request.size()) state = HTTP_CO_REQUEST_SENT; @@ -664,7 +738,7 @@ again: { // Timeout happened and there is no data to read close_connection(); - parsed = { .error = "HTTP request timed out" }; + parsed = { .error = "HTTP request timed out", .status_code = ETIMEDOUT }; run_cb_and_clear(); } } @@ -679,9 +753,11 @@ again: epoll_events = epoll_events & ~EPOLLIN; if (state == HTTP_CO_HEADERS_RECEIVED) std::swap(parsed.body, response); - close_connection(); if (res < 0) - parsed = { .error = std::string("recvmsg: ")+strerror(-res) }; + parsed = { .error = std::string("recvmsg: ")+strerror(-res), .status_code = (int)-res }; + else if (state == HTTP_CO_SERVER || state == HTTP_CO_REQ_HDR_RECEIVED || state == HTTP_CO_REQUEST_RECEIVED) + parsed = { .error = "client has disconnected normally" }; + close_connection(); run_cb_and_clear(); } else @@ -703,7 +779,7 @@ void http_co_t::on_ssl_error(int res) parsed = { .error = "peer closed the SSL connection" }; } else - parsed = { .error = std::string("SSL error: ")+ERR_error_string(ERR_get_error(), NULL) }; + parsed = { .error = std::string("SSL error: ")+ERR_error_string(ERR_get_error(), NULL), .status_code = EIO }; run_cb_and_clear(); } @@ -730,7 +806,7 @@ bool http_co_t::do_ssl_handshake(bool init_send) else { int errcode = ERR_get_error(); - parsed = { .error = ERR_error_string(errcode, NULL) }; + parsed = { .error = ERR_error_string(errcode, NULL), .status_code = EIO }; close_connection(); run_cb_and_clear(); return false; @@ -743,7 +819,10 @@ bool http_co_t::do_ssl_handshake(bool init_send) bool http_co_t::handle_read() { stackin(); - if (state == HTTP_CO_REQUEST_SENT) + if (state == HTTP_CO_REQUEST_RECEIVED) + { + } + else if (state == HTTP_CO_REQUEST_SENT) { int pos = response.find("\r\n\r\n"); if (pos >= 0) @@ -755,7 +834,7 @@ bool http_co_t::handle_read() timeout_id = -1; } state = HTTP_CO_HEADERS_RECEIVED; - parse_http_headers(response, &parsed); + parse_http_headers(response, &parsed, false); if (parsed.status_code == 101 && parsed.headers.find("sec-websocket-accept") != parsed.headers.end() && parsed.headers["upgrade"] == "websocket" && @@ -779,7 +858,7 @@ bool http_co_t::handle_read() { // Sorry, unsupported response close_connection(); - parsed = { .error = "Response has neither Connection: close, nor Transfer-Encoding: chunked nor Content-Length headers" }; + parsed = { .error = "Response has neither Connection: close, nor Transfer-Encoding: chunked nor Content-Length headers", .status_code = EINVAL }; run_cb_and_clear(); stackout(); return false; @@ -791,14 +870,57 @@ bool http_co_t::handle_read() } } } - if (state == HTTP_CO_HEADERS_RECEIVED && target_response_size > 0 && response.size() >= target_response_size) + else if (state == HTTP_CO_SERVER) + { + int pos = response.find("\r\n\r\n"); + if (pos >= 0) + { + if (timeout_id >= 0) + { + // Timeout is cleared when headers are received + tfd->clear_timer(timeout_id); + timeout_id = -1; + } + state = HTTP_CO_REQ_HDR_RECEIVED; + parse_http_headers(response, &parsed, true); + auto conn_it = parsed.headers.find("connection"); + keepalive = (conn_it != parsed.headers.end() && conn_it->second == "keep-alive"); + auto enc_it = parsed.headers.find("transfer-encoding"); + if (enc_it != parsed.headers.end()) + { + // Sorry, unsupported request + close_connection(); + parsed = { .error = "Chunked requests are not supported", .status_code = EINVAL }; + run_cb_and_clear(); + stackout(); + return false; + } + auto len_it = parsed.headers.find("content-length"); + target_response_size = stoull_full(len_it != parsed.headers.end() ? len_it->second : ""); + if (!target_response_size) + { + state = HTTP_CO_REQUEST_RECEIVED; + response_callback(&parsed); + } + } + } + if ((state == HTTP_CO_HEADERS_RECEIVED || state == HTTP_CO_REQ_HDR_RECEIVED) && + target_response_size > 0 && response.size() >= target_response_size) { std::swap(parsed.body, response); - if (!keepalive) - close_connection(); + if (state == HTTP_CO_REQ_HDR_RECEIVED) + { + state = HTTP_CO_REQUEST_RECEIVED; + response_callback(&parsed); + } else - state = HTTP_CO_KEEPALIVE; - run_cb_and_clear(); + { + if (!keepalive) + close_connection(); + else + state = HTTP_CO_KEEPALIVE; + run_cb_and_clear(); + } } else if (state == HTTP_CO_CHUNKED && response.size() > 0) { @@ -861,26 +983,34 @@ void http_co_t::next_request() { if (keepalive_queue.size() > 0) { - auto next = keepalive_queue[0]; - keepalive_queue.erase(keepalive_queue.begin(), keepalive_queue.begin()+1); - next(); + auto next = std::move(keepalive_queue[0]); + keepalive_queue.erase(keepalive_queue.begin()); + send_request(next.host, next.request, next.options, next.cb); } } -static void parse_http_headers(std::string & res, http_message_t *parsed) +static void parse_http_headers(std::string & res, http_message_t *parsed, bool is_request) { int pos = res.find("\r\n"); pos = pos < 0 ? res.length() : pos+2; std::string status_line = res.substr(0, pos); int http_version; char *status_text = NULL; - sscanf(status_line.c_str(), "HTTP/1.%d %d %ms", &http_version, &parsed->status_code, &status_text); - if (status_text) + if (!is_request) { - parsed->status_line = status_text; - // %ms = allocate a buffer - free(status_text); - status_text = NULL; + sscanf(status_line.c_str(), "HTTP/1.%d %d %ms", &http_version, &parsed->status_code, &status_text); + if (status_text) + { + parsed->status_line = status_text; + // %ms = allocate a buffer + free(status_text); + status_text = NULL; + } + } + else + { + // Should be GET/POST / HTTP/1.1 + parsed->status_line = status_line; } int prev = pos; while ((pos = res.find("\r\n", prev)) >= prev) diff --git a/src/client/http_client.h b/src/client/http_client.h index c36d2c23..cfcb516d 100644 --- a/src/client/http_client.h +++ b/src/client/http_client.h @@ -17,6 +17,8 @@ class timerfd_manager_t; +#pragma GCC visibility push(default) + struct http_options_t { int timeout; @@ -44,13 +46,19 @@ struct http_message_t // Opened websocket or keepalive HTTP connection struct http_co_t; -http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, const std::string & ssl_ca, std::string & error); +http_context_t* http_context_init(const std::string & ssl_cert, const std::string & ssl_key, + const std::string & ssl_ca, bool verify_peer, std::string & error); void http_context_destroy(http_context_t *ctx); http_co_t* http_init(timerfd_manager_t *tfd, http_context_t *ctx = NULL); void open_websocket(http_co_t *handler, const std::string & host, const std::string & path, - const http_options_t & options, std::function on_message); + const http_options_t & options, std::function on_message); void http_request(http_co_t *handler, const std::string & host, const std::string & request, - const http_options_t & options, std::function response_callback); + const http_options_t & options, std::function response_callback); void http_post_message(http_co_t *handler, uint8_t type, const std::string & msg); +void http_serve(http_co_t *handler, int peer_fd, const http_options_t & options, + std::function request_callback); +void http_reply(http_co_t *handler, const std::string & reply); void http_close(http_co_t *co); void http_destroy(http_co_t *co); + +#pragma GCC visibility pop