diff options
| -rw-r--r-- | demo-json-rpc-server.c | 271 |
1 files changed, 162 insertions, 109 deletions
diff --git a/demo-json-rpc-server.c b/demo-json-rpc-server.c index 1ae1bad..60a7173 100644 --- a/demo-json-rpc-server.c +++ b/demo-json-rpc-server.c @@ -540,9 +540,9 @@ enum ws_handler_state { WS_HANDLER_CONNECTING, ///< Parsing HTTP WS_HANDLER_OPEN, ///< Parsing WebSockets frames - WS_HANDLER_CLOSING, ///< Closing the connection - WS_HANDLER_ALMOST_DEAD, ///< Closing connection after failure - WS_HANDLER_CLOSED ///< Dead + WS_HANDLER_CLOSING, ///< Partial closure by us + WS_HANDLER_FLUSHING, ///< Just waiting for client EOF + WS_HANDLER_CLOSED ///< Dead, both sides closed }; struct ws_handler @@ -584,6 +584,7 @@ struct ws_handler // TODO: void (*on_handshake) (protocols) that will allow the user // to choose any sub-protocol, if the client has provided any. // This may render "on_connected" unnecessary. + // Should also enable failing the handshake. /// Called after successfuly connecting (handshake complete) bool (*on_connected) (struct ws_handler *); @@ -626,36 +627,41 @@ static void ws_handler_close (struct ws_handler *self, enum ws_status close_code, const char *reason, size_t len) { + hard_assert (self->state == WS_HANDLER_OPEN); + struct str payload = str_make (); str_pack_u16 (&payload, close_code); // XXX: maybe accept a null-terminated string on input? Has to be UTF-8 a/w str_append_data (&payload, reason, len); ws_handler_send_control (self, WS_OPCODE_CLOSE, payload.str, payload.len); - - // Close initiated by us; the reason is null-terminated within `payload' - if (self->on_close) - self->on_close (self, close_code, payload.str + 2); + self->close_cb (self, true /* half_close */); self->state = WS_HANDLER_CLOSING; str_free (&payload); } -static void -ws_handler_fail (struct ws_handler *self, enum ws_status close_code) +static bool +ws_handler_fail_connection (struct ws_handler *self, enum ws_status close_code) { - ws_handler_close (self, close_code, NULL, 0); - self->state = WS_HANDLER_ALMOST_DEAD; + hard_assert (self->state == WS_HANDLER_OPEN + || self->state == WS_HANDLER_CLOSING); + + if (self->state == WS_HANDLER_OPEN) + ws_handler_close (self, close_code, NULL, 0); - // TODO: set the close timer, ignore all further incoming input (either set - // some flag for the case that we're in the middle of ws_handler_push(), - // and/or add a mechanism to stop the caller from polling the socket for - // reads). - // TODO: make sure we don't send pings after the close + self->state = WS_HANDLER_FLUSHING; + if (self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, ""); + + ev_timer_stop (EV_DEFAULT_ &self->ping_timer); + ev_timer_set (&self->close_timeout_watcher, self->close_timeout, 0.); + ev_timer_start (EV_DEFAULT_ &self->close_timeout_watcher); + return false; } // TODO: add support for fragmented responses static void -ws_handler_send (struct ws_handler *self, +ws_handler_send_frame (struct ws_handler *self, enum ws_opcode opcode, const void *data, size_t len) { if (!soft_assert (self->state == WS_HANDLER_OPEN)) @@ -697,24 +703,26 @@ ws_handler_on_frame_header (void *user_data, const struct ws_parser *parser) || (!ws_is_control_frame (parser->opcode) && (self->expecting_continuation && parser->opcode != WS_OPCODE_CONT)) || parser->payload_len >= 0x8000000000000000ULL) - ws_handler_fail (self, WS_STATUS_PROTOCOL_ERROR); - else if (parser->payload_len > self->max_payload_len) - ws_handler_fail (self, WS_STATUS_MESSAGE_TOO_BIG); - else - return true; - return false; + return ws_handler_fail_connection (self, WS_STATUS_PROTOCOL_ERROR); + + if (parser->payload_len > self->max_payload_len + || (self->expecting_continuation && + self->message_data.len + parser->payload_len > self->max_payload_len)) + return ws_handler_fail_connection (self, WS_STATUS_MESSAGE_TOO_BIG); + return true; } static bool -ws_handler_on_protocol_close +ws_handler_on_control_close (struct ws_handler *self, const struct ws_parser *parser) { + hard_assert (self->state == WS_HANDLER_OPEN + || self->state == WS_HANDLER_CLOSING); struct msg_unpacker unpacker = msg_unpacker_make (parser->input.str, parser->payload_len); char *reason = NULL; uint16_t close_code = WS_STATUS_NO_STATUS_RECEIVED; - if (parser->payload_len >= 2) { (void) msg_unpacker_u16 (&unpacker, &close_code); @@ -723,17 +731,29 @@ ws_handler_on_protocol_close else reason = xstrdup (""); - if (self->state != WS_HANDLER_CLOSING) + if (close_code < 1000 || close_code > 4999) + // XXX: invalid close code: maybe we should fail the connection instead + close_code = WS_STATUS_PROTOCOL_ERROR; + + if (self->state == WS_HANDLER_OPEN) { // Close initiated by the client + // FIXME: not sending the potentially different close_code ws_handler_send_control (self, WS_OPCODE_CLOSE, parser->input.str, parser->payload_len); + + self->state = WS_HANDLER_FLUSHING; if (self->on_close) self->on_close (self, close_code, reason); } + else + self->state = WS_HANDLER_FLUSHING; free (reason); - self->state = WS_HANDLER_ALMOST_DEAD; + + ev_timer_stop (EV_DEFAULT_ &self->ping_timer); + ev_timer_set (&self->close_timeout_watcher, self->close_timeout, 0.); + ev_timer_start (EV_DEFAULT_ &self->close_timeout_watcher); return true; } @@ -744,21 +764,18 @@ ws_handler_on_control_frame switch (parser->opcode) { case WS_OPCODE_CLOSE: - return ws_handler_on_protocol_close (self, parser); + return ws_handler_on_control_close (self, parser); case WS_OPCODE_PING: ws_handler_send_control (self, WS_OPCODE_PONG, parser->input.str, parser->payload_len); break; case WS_OPCODE_PONG: - // XXX: maybe we should check the payload + // TODO: check the payload self->received_pong = true; break; default: // Unknown control frame - ws_handler_fail (self, WS_STATUS_PROTOCOL_ERROR); - // FIXME: we shouldn't close the connection right away; - // also check other places - return false; + return ws_handler_fail_connection (self, WS_STATUS_PROTOCOL_ERROR); } return true; } @@ -769,29 +786,19 @@ ws_handler_on_frame (void *user_data, const struct ws_parser *parser) struct ws_handler *self = user_data; if (ws_is_control_frame (parser->opcode)) return ws_handler_on_control_frame (self, parser); - - // TODO: do this rather in "on_frame_header" - if (self->message_data.len + parser->payload_len > self->max_payload_len) - { - ws_handler_fail (self, WS_STATUS_MESSAGE_TOO_BIG); - return false; - } - if (!self->expecting_continuation) self->message_opcode = parser->opcode; str_append_data (&self->message_data, parser->input.str, parser->payload_len); - self->expecting_continuation = !parser->is_fin; - - if (!parser->is_fin) + if ((self->expecting_continuation = !parser->is_fin)) return true; if (self->message_opcode == WS_OPCODE_TEXT && !utf8_validate (self->message_data.str, self->message_data.len)) { - ws_handler_fail (self, WS_STATUS_INVALID_PAYLOAD_DATA); - return false; + return ws_handler_fail_connection + (self, WS_STATUS_INVALID_PAYLOAD_DATA); } bool result = true; @@ -799,6 +806,8 @@ ws_handler_on_frame (void *user_data, const struct ws_parser *parser) result = self->on_message (self, self->message_opcode, self->message_data.str, self->message_data.len); str_reset (&self->message_data); + // TODO: if (!result), either replace this with a state check, + // or make sure to change the state return result; } @@ -810,11 +819,10 @@ ws_handler_on_ping_timer (EV_P_ ev_timer *watcher, int revents) struct ws_handler *self = watcher->data; if (!self->received_pong) - { - // TODO: close/fail the connection? - } + ws_handler_fail_connection (self, 4000); else { + // TODO: be an annoying server and send a nonce in the data ws_handler_send_control (self, WS_OPCODE_PING, NULL, 0); ev_timer_again (EV_A_ watcher); } @@ -823,20 +831,38 @@ ws_handler_on_ping_timer (EV_P_ ev_timer *watcher, int revents) static void ws_handler_on_close_timeout (EV_P_ ev_timer *watcher, int revents) { + (void) loop; (void) revents; struct ws_handler *self = watcher->data; - // TODO: anything else to do here? Invalidate our state? - if (self->close_cb) - self->close_cb (self, false /* half_close */); + hard_assert (self->state == WS_HANDLER_OPEN + || self->state == WS_HANDLER_CLOSING); + + if (self->state == WS_HANDLER_CLOSING + && self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, "close timeout"); + + self->state = WS_HANDLER_CLOSED; + self->close_cb (self, false /* half_close */); } static void ws_handler_on_handshake_timeout (EV_P_ ev_timer *watcher, int revents) { + (void) loop; (void) revents; struct ws_handler *self = watcher->data; - // TODO + + // XXX: this is a no-op, since this currently doesn't even call shutdown + // immediately but postpones it until later + self->close_cb (self, true /* half_close */); + self->state = WS_HANDLER_FLUSHING; + + if (self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, "handshake timeout"); + + self->state = WS_HANDLER_CLOSED; + self->close_cb (self, false /* half_close */); } static void @@ -991,6 +1017,7 @@ ws_handler_on_url (http_parser *parser, const char *at, size_t len) #define HTTP_400_BAD_REQUEST "400 Bad Request" #define HTTP_405_METHOD_NOT_ALLOWED "405 Method Not Allowed" #define HTTP_417_EXPECTATION_FAILED "407 Expectation Failed" +#define HTTP_426_UPGRADE_REQUIRED "426 Upgrade Required" #define HTTP_505_VERSION_NOT_SUPPORTED "505 HTTP Version Not Supported" static void @@ -1024,43 +1051,47 @@ ws_handler_http_responsev (struct ws_handler *self, str_free (&response); } -static void -ws_handler_http_response (struct ws_handler *self, const char *status, ...) +static bool +ws_handler_fail_handshake (struct ws_handler *self, const char *status, ...) { - struct strv v = strv_make (); - va_list ap; va_start (ap, status); const char *s; + struct strv v = strv_make (); while ((s = va_arg (ap, const char *))) strv_append (&v, s); va_end (ap); - ws_handler_http_responsev (self, status, v.vector); strv_free (&v); + + self->close_cb (self, true /* half_close */); + self->state = WS_HANDLER_FLUSHING; + + if (self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, status); + return false; } -#define FAIL_HANDSHAKE(status, ...) \ - BLOCK_START \ - self->state = WS_HANDLER_ALMOST_DEAD; \ - ws_handler_http_response (self, (status), __VA_ARGS__); \ - return false; \ - BLOCK_END +#define FAIL_HANDSHAKE(...) \ + return ws_handler_fail_handshake (self, __VA_ARGS__, NULL) static bool ws_handler_finish_handshake (struct ws_handler *self) { - // XXX: we probably shouldn't use 505 to reject the minor version but w/e - if (self->hp.http_major != 1 || self->hp.http_minor < 1) - FAIL_HANDSHAKE (HTTP_505_VERSION_NOT_SUPPORTED, NULL); if (self->hp.method != HTTP_GET) - FAIL_HANDSHAKE (HTTP_405_METHOD_NOT_ALLOWED, "Allow: GET", NULL); + FAIL_HANDSHAKE (HTTP_405_METHOD_NOT_ALLOWED, "Allow: GET"); + + // Technically, it must be /at least/ 1.1 but no other 1.x version of HTTP + // is going to happen and 2.x is entirely incompatible + // XXX: we probably shouldn't use 505 to reject the minor version but w/e + if (self->hp.http_major != 1 || self->hp.http_minor != 1) + FAIL_HANDSHAKE (HTTP_505_VERSION_NOT_SUPPORTED); // Your expectations are way too high if (str_map_find (&self->headers, "Expect")) - FAIL_HANDSHAKE (HTTP_417_EXPECTATION_FAILED, NULL); + FAIL_HANDSHAKE (HTTP_417_EXPECTATION_FAILED); // Reject URLs specifying the schema and host; we're not parsing that // TODO: actually do parse this and let our user decide if it matches @@ -1068,11 +1099,11 @@ ws_handler_finish_handshake (struct ws_handler *self) if (http_parser_parse_url (self->url.str, self->url.len, false, &url) || (url.field_set & (1 << UF_SCHEMA | 1 << UF_HOST | 1 << UF_PORT)) || !str_map_find (&self->headers, "Host")) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); const char *connection = str_map_find (&self->headers, "Connection"); if (!connection || strcasecmp_ascii (connection, "Upgrade")) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); // Check if we can actually upgrade the protocol to WebSockets const char *upgrade = str_map_find (&self->headers, "Upgrade"); @@ -1088,7 +1119,8 @@ ws_handler_finish_handshake (struct ws_handler *self) http_protocol_destroy (iter); } if (!can_upgrade) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + FAIL_HANDSHAKE (HTTP_426_UPGRADE_REQUIRED, + "Upgrade: websocket", SEC_WS_VERSION ": 13"); // Okay, we're finally past the basic HTTP/1.1 stuff const char *key = str_map_find (&self->headers, SEC_WS_KEY); @@ -1098,19 +1130,17 @@ ws_handler_finish_handshake (struct ws_handler *self) const char *extensions = str_map_find (&self->headers, SEC_WS_EXTENSIONS); */ - if (!key) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + if (!version) + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); + if (strcmp (version, "13")) + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, SEC_WS_VERSION ": 13"); struct str tmp = str_make (); - bool key_is_valid = base64_decode (key, false, &tmp) && tmp.len == 16; + bool key_is_valid = key + && base64_decode (key, false, &tmp) && tmp.len == 16; str_free (&tmp); if (!key_is_valid) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); - - if (!version) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); - if (strcmp (version, "13")) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, SEC_WS_VERSION ": 13", NULL); + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); struct strv fields = strv_make (); strv_append_args (&fields, @@ -1130,6 +1160,7 @@ ws_handler_finish_handshake (struct ws_handler *self) strv_free (&fields); + self->state = WS_HANDLER_OPEN; ev_timer_init (&self->ping_timer, ws_handler_on_ping_timer, self->ping_interval, 0); ev_timer_start (EV_DEFAULT_ &self->ping_timer); @@ -1141,40 +1172,62 @@ ws_handler_finish_handshake (struct ws_handler *self) static void ws_handler_start (struct ws_handler *self) { + hard_assert (self->state == WS_HANDLER_CONNECTING); + ev_timer_set (&self->handshake_timeout_watcher, self->handshake_timeout, 0.); ev_timer_start (EV_DEFAULT_ &self->handshake_timeout_watcher); } -/// Push data to the WebSocket handler. "len == 0" means EOF. +// The client should normally never close the connection, assume that it's +// either received an EOF from our side, or that it doesn't care about our data +// anymore, having called close() already static bool -ws_handler_push (struct ws_handler *self, const void *data, size_t len) +ws_handler_push_eof (struct ws_handler *self) { - // TODO: make sure all timers are stopped appropriately - - if (!len) + switch (self->state) { + case WS_HANDLER_CONNECTING: ev_timer_stop (EV_DEFAULT_ &self->handshake_timeout_watcher); - if (self->state == WS_HANDLER_OPEN) - { - if (self->on_close) - self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, ""); - } - else - { - // TODO: anything to do besides just closing the connection? - } - + self->state = WS_HANDLER_FLUSHING; + if (self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, "unexpected EOF"); + break; + case WS_HANDLER_OPEN: + ev_timer_stop (EV_DEFAULT_ &self->ping_timer); + // Fall-through + case WS_HANDLER_CLOSING: self->state = WS_HANDLER_CLOSED; - return false; + if (self->on_close) + self->on_close (self, WS_STATUS_ABNORMAL_CLOSURE, ""); + // Fall-through + case WS_HANDLER_FLUSHING: + ev_timer_stop (EV_DEFAULT_ &self->close_timeout_watcher); + break; + default: + soft_assert(self->state != WS_HANDLER_CLOSED); } + self->state = WS_HANDLER_CLOSED; + return false; +} + +/// Push data to the WebSocket handler. "len == 0" means EOF. +/// You are expected to close the connection and dispose of the handler +/// when the function returns false. +static bool +ws_handler_push (struct ws_handler *self, const void *data, size_t len) +{ + if (!len) + return ws_handler_push_eof (self); - if (self->state == WS_HANDLER_ALMOST_DEAD) + if (self->state == WS_HANDLER_FLUSHING) // We're waiting for an EOF from the client, must not process data return true; + if (self->state != WS_HANDLER_CONNECTING) - return ws_parser_push (&self->parser, data, len); + return soft_assert (self->state != WS_HANDLER_CLOSED) + && ws_parser_push (&self->parser, data, len); // The handshake hasn't been done yet, process HTTP headers static const http_parser_settings http_settings = @@ -1185,8 +1238,8 @@ ws_handler_push (struct ws_handler *self, const void *data, size_t len) .on_url = ws_handler_on_url, }; - size_t n_parsed = http_parser_execute (&self->hp, - &http_settings, data, len); + size_t n_parsed = + http_parser_execute (&self->hp, &http_settings, data, len); if (self->hp.upgrade) { @@ -1195,12 +1248,10 @@ ws_handler_push (struct ws_handler *self, const void *data, size_t len) // The handshake hasn't been finished, yet there is more data // to be processed after the headers already if (len - n_parsed) - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); if (!ws_handler_finish_handshake (self)) return false; - - self->state = WS_HANDLER_OPEN; if (self->on_connected) return self->on_connected (self); return true; @@ -1217,7 +1268,7 @@ ws_handler_push (struct ws_handler *self, const void *data, size_t len) print_debug ("WS handshake failed: %s", http_errno_description (err)); - FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST, NULL); + FAIL_HANDSHAKE (HTTP_400_BAD_REQUEST); } return true; } @@ -2319,15 +2370,15 @@ client_ws_on_message (struct ws_handler *handler, FIND_CONTAINER (self, handler, struct client_ws, handler); if (type != WS_OPCODE_TEXT) { - ws_handler_fail (&self->handler, WS_STATUS_UNSUPPORTED_DATA); - return false; + return ws_handler_fail_connection + (&self->handler, WS_STATUS_UNSUPPORTED_DATA); } struct server_context *ctx = ev_userdata (EV_DEFAULT); struct str response = str_make (); process_json_rpc (ctx, data, len, &response); if (response.len) - ws_handler_send (&self->handler, + ws_handler_send_frame (&self->handler, WS_OPCODE_TEXT, response.str, response.len); str_free (&response); return true; @@ -2353,6 +2404,7 @@ static bool client_ws_push (struct client *client, const void *data, size_t len) { FIND_CONTAINER (self, client, struct client_ws, client); + // client_close() will correctly destroy the client on EOF return ws_handler_push (&self->handler, data, len); } @@ -2361,7 +2413,8 @@ client_ws_shutdown (struct client *client) { FIND_CONTAINER (self, client, struct client_ws, client); if (self->handler.state == WS_HANDLER_CONNECTING) - ; // TODO: abort the connection immediately + // No on_close, no problem + client_destroy (&self->client); else if (self->handler.state == WS_HANDLER_OPEN) ws_handler_close (&self->handler, WS_STATUS_GOING_AWAY, NULL, 0); } |