Implement HTTP server support O_o

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())
         {

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<void(http_message_t *)> 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<void(const http_message_t*)> response_callback;
+    std::function<void(http_message_t*)> response_callback;
 
     int request_timeout = 0;
     bool ssl = false;
@@ -76,7 +84,7 @@ struct http_co_t
     bool want_streaming;
     bool keepalive;
 
-    std::vector<std::function<void()>> keepalive_queue;
+    std::vector<http_call_t> 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<void(const http_message_t *response)> response_callback);
+        const http_options_t & options, std::function<void(http_message_t *response)> 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<void(const http_message_t *msg)> response_callback)
+    const http_options_t & options, std::function<void(http_message_t *msg)> 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<void(const http_message_t *response)> response_callback)
+    const http_options_t & options, std::function<void(http_message_t *response)> 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<void(http_message_t *msg)> 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<void(const http_message_t*)> cb;
+    std::function<void(http_message_t*)> 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<void(const http_message_t *response)> response_callback)
+    const http_options_t & options, std::function<void(http_message_t *response)> 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]()
+        keepalive_queue.emplace_back((http_call_t){ host, request, options, std::move(response_callback) });
-        {
-            this->send_request(host, request, options, 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)
+        if (state == HTTP_CO_REQ_HDR_RECEIVED)
-            close_connection();
+        {
+            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];
+        auto next = std::move(keepalive_queue[0]);
-        keepalive_queue.erase(keepalive_queue.begin(), keepalive_queue.begin()+1);
+        keepalive_queue.erase(keepalive_queue.begin());
-        next();
+        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 (!is_request)
-    if (status_text)
     {
-        parsed->status_line = status_text;
+        sscanf(status_line.c_str(), "HTTP/1.%d %d %ms", &http_version, &parsed->status_code, &status_text);
-        // %ms = allocate a buffer
+        if (status_text)
-        free(status_text);
+        {
-        status_text = NULL;
+            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)

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<void(const http_message_t *msg)> on_message);
+    const http_options_t & options, std::function<void(http_message_t *msg)> on_message);
 void http_request(http_co_t *handler, const std::string & host, const std::string & request,
-    const http_options_t & options, std::function<void(const http_message_t *response)> response_callback);
+    const http_options_t & options, std::function<void(http_message_t *response)> 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<void(http_message_t *msg)> 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