| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "websocket-internal.h" |
| 4 | |
| 5 | // -------------------------------------------------------------------------------------------------------------------- |
| 6 | // writing to the socket |
| 7 | |
| 8 | // Actually write data to the client socket |
| 9 | ssize_t websocket_write_data(WS_CLIENT *wsc) { |
| 10 | internal_fatal(wsc->wth->tid != gettid_cached(), "Function %s() should only be used by the websocket thread", __FUNCTION__ ); |
| 11 | |
| 12 | worker_is_busy(WORKERS_WEBSOCKET_SOCK_SEND); |
| 13 | |
| 14 | if (!wsc->out_buffer.data || wsc->sock.fd < 0) |
| 15 | return -1; |
| 16 | |
| 17 | ssize_t bytes_written = 0; |
| 18 | |
| 19 | // Let cbuffer_next_unsafe determine if there's data to write |
| 20 | // This correctly handles the circular buffer wrap-around cases |
| 21 | |
| 22 | // Get data to write from circular buffer |
| 23 | char *data; |
| 24 | size_t data_length = cbuffer_next_unsafe(&wsc->out_buffer, &data); |
| 25 | if (data_length == 0) |
| 26 | goto done; |
| 27 | |
| 28 | // Dump the data being written for debugging |
| 29 | websocket_dump_debug(wsc, data, data_length, "TX SOCK %zu bytes", data_length); |
| 30 | |
| 31 | // In the websocket thread we want non-blocking behavior |
| 32 | // Use nd_sock_write with a single retry |
| 33 | bytes_written = nd_sock_write(&wsc->sock, data, data_length, 1); // 1 retry for non-blocking write |
| 34 | |
| 35 | if (bytes_written < 0) { |
| 36 | websocket_error(wsc, "Failed to write to client: %s", strerror(errno)); |
| 37 | goto done; |
| 38 | } |
| 39 | |
| 40 | // Remove written bytes from circular buffer |
| 41 | if (bytes_written > 0) |
| 42 | cbuffer_remove_unsafe(&wsc->out_buffer, bytes_written); |
| 43 | |
| 44 | done: |
| 45 | websocket_thread_update_client_poll_flags(wsc); |
| 46 | return bytes_written; |
| 47 | } |
| 48 | |
| 49 | // -------------------------------------------------------------------------------------------------------------------- |
| 50 | |
| 51 | static inline size_t select_header_size(size_t payload_len) { |
| 52 | if (payload_len < 126) |
| 53 | return 2; |
| 54 | else if (payload_len <= 65535) |
| 55 | return 4; |
| 56 | else |
| 57 | return 10; |
| 58 | } |
| 59 | |
| 60 | /** |
| 61 | * @brief Create and send a WebSocket frame |
| 62 | * |
| 63 | * Creates a WebSocket frame with the given payload and sends it. |
| 64 | * This function directly creates a frame with the specified FIN and RSV1 bits. |
| 65 | * |
| 66 | * @param wsc WebSocket client |
| 67 | * @param payload Payload data to send |
| 68 | * @param payload_len Length of the payload |
| 69 | * @param opcode WebSocket opcode (text, binary, continuation, etc.) |
| 70 | * @param compressed Whether the payload is already compressed (RSV1 bit) |
| 71 | * @param final Whether this is the final frame in a message (FIN bit) |
| 72 | * @return Number of bytes sent or -1 on error |
| 73 | */ |
| 74 | static int websocket_protocol_send_frame( |
| 75 | WS_CLIENT *wsc, const char *payload, size_t payload_len, |
| 76 | WEBSOCKET_OPCODE opcode, bool compressed, bool final) { |
| 77 | |
| 78 | if(!wsc) |
| 79 | return -1; |
| 80 | |
| 81 | const char *disconnect_msg = ""; |
| 82 | |
| 83 | if (wsc->sock.fd < 0) { |
| 84 | disconnect_msg = "Client not connected"; |
| 85 | goto abnormal_disconnect; |
| 86 | } |
| 87 | |
| 88 | // Validate parameters based on WebSocket protocol |
| 89 | if (websocket_frame_is_control_opcode(opcode) && payload_len > 125) { |
| 90 | disconnect_msg = "Control frame payload too large"; |
| 91 | goto abnormal_disconnect; |
| 92 | } |
| 93 | |
| 94 | // Control frames must not be fragmented |
| 95 | if (websocket_frame_is_control_opcode(opcode) && !final) { |
| 96 | disconnect_msg = "Control frames cannot be fragmented"; |
| 97 | goto abnormal_disconnect; |
| 98 | } |
| 99 | |
| 100 | // Control frames cannot be compressed |
| 101 | if (websocket_frame_is_control_opcode(opcode) && compressed) { |
| 102 | disconnect_msg = "Control frames cannot be compressed"; |
| 103 | goto abnormal_disconnect; |
| 104 | } |
| 105 | |
| 106 | // For compressed payloads, we already did the compression outside this function |
| 107 | // so we don't need to compress again, just use the provided payload directly |
| 108 | size_t final_payload_len = payload_len; |
| 109 | |
| 110 | // Determine header size based on payload length |
| 111 | size_t header_size = select_header_size(final_payload_len); |
| 112 | |
| 113 | // Calculate frame size |
| 114 | size_t frame_size = header_size + final_payload_len; |
| 115 | |
| 116 | // Reserve space in the circular buffer for the entire frame |
| 117 | unsigned char *header_dst = (unsigned char *)cbuffer_reserve_unsafe(&wsc->out_buffer, frame_size); |
| 118 | if (!header_dst) { |
| 119 | disconnect_msg = "Buffer full - too much outgoing data"; |
| 120 | goto abnormal_disconnect; |
| 121 | } |
| 122 | |
| 123 | // The payload will be written directly after the header in our reserved buffer |
| 124 | char *payload_dst = (char *)(header_dst + header_size); |
| 125 | |
| 126 | // Copy payload data to our buffer |
| 127 | if (payload && payload_len > 0) { |
| 128 | memcpy(payload_dst, payload, payload_len); |
| 129 | } |
| 130 | |
| 131 | // Write the header |
| 132 | // First byte: FIN bit, RSV1 bit (compression), and opcode |
| 133 | // Only set FIN bit if this is the final frame |
| 134 | // Only set RSV1 if this frame uses compression |
| 135 | header_dst[0] = (final ? 0x80 : 0) | (compressed ? 0x40 : 0) | (opcode & 0x0F); |
| 136 | |
| 137 | // Write payload length with the appropriate format |
| 138 | switch(header_size) { |
| 139 | case 2: |
| 140 | header_dst[1] = final_payload_len & 0x7F; |
| 141 | break; |
| 142 | |
| 143 | case 4: |
| 144 | header_dst[1] = 126; |
| 145 | header_dst[2] = (final_payload_len >> 8) & 0xFF; |
| 146 | header_dst[3] = final_payload_len & 0xFF; |
| 147 | break; |
| 148 | |
| 149 | case 10: |
| 150 | header_dst[1] = 127; |
| 151 | header_dst[2] = (final_payload_len >> 56) & 0xFF; |
| 152 | header_dst[3] = (final_payload_len >> 48) & 0xFF; |
| 153 | header_dst[4] = (final_payload_len >> 40) & 0xFF; |
| 154 | header_dst[5] = (final_payload_len >> 32) & 0xFF; |
| 155 | header_dst[6] = (final_payload_len >> 24) & 0xFF; |
| 156 | header_dst[7] = (final_payload_len >> 16) & 0xFF; |
| 157 | header_dst[8] = (final_payload_len >> 8) & 0xFF; |
| 158 | header_dst[9] = final_payload_len & 0xFF; |
| 159 | break; |
| 160 | |
| 161 | default: |
| 162 | // impossible case - added to avoid compiler warning |
| 163 | disconnect_msg = "Invalid header size"; |
| 164 | goto abnormal_disconnect; |
| 165 | } |
| 166 | |
| 167 | // Commit the final frame size (header + payload) |
| 168 | size_t final_frame_size = header_size + final_payload_len; |
| 169 | cbuffer_commit_reserved_unsafe(&wsc->out_buffer, final_frame_size); |
| 170 | |
| 171 | #ifdef NETDATA_INTERNAL_CHECKS |
| 172 | // Log frame being sent with detailed format matching the received frame logging |
| 173 | WEBSOCKET_FRAME_HEADER header; |
| 174 | if(!websocket_protocol_parse_header_from_buffer((const char *)header_dst, header_size, &header)) { |
| 175 | disconnect_msg = "Failed to parse the header we generated"; |
| 176 | goto abnormal_disconnect; |
| 177 | } |
| 178 | |
| 179 | websocket_debug(wsc, |
| 180 | "TX FRAME: OPCODE=0x%x (%s), FIN=%s, RSV1=%d, RSV2=%d, RSV3=%d, MASK=%s, LEN=%d, " |
| 181 | "PAYLOAD_LEN=%zu, HEADER_SIZE=%zu, FRAME_SIZE=%zu, MASK=%02x%02x%02x%02x", |
| 182 | header.opcode, |
| 183 | WEBSOCKET_OPCODE_2str(opcode), |
| 184 | header.fin ? "True" : "False", header.rsv1, header.rsv2, header.rsv3, |
| 185 | header.mask ? "True" : "False", header.len, |
| 186 | header.payload_length, header.header_size, header.frame_size, |
| 187 | header.mask_key[0], header.mask_key[1], header.mask_key[2], header.mask_key[3]); |
| 188 | #endif |
| 189 | |
| 190 | // Make sure the client's poll flags include WRITE |
| 191 | websocket_thread_update_client_poll_flags(wsc); |
| 192 | |
| 193 | return (int)final_frame_size; |
| 194 | |
| 195 | abnormal_disconnect: |
| 196 | websocket_error(wsc, "triggering abnormal disconnect: %s", disconnect_msg); |
| 197 | websocket_thread_send_command(wsc->wth, WEBSOCKET_THREAD_CMD_REMOVE_CLIENT, wsc->id); |
| 198 | return -1; |
| 199 | |
| 200 | //graceful_disconnect: |
| 201 | // // the current implementation does not support graceful disconnect - so we do an abnormal one |
| 202 | // websocket_error(wsc, "triggering graceful disconnect: %s", disconnect_msg); |
| 203 | // websocket_thread_send_command(wsc->wth, WEBSOCKET_THREAD_CMD_REMOVE_CLIENT, wsc->id); |
| 204 | // return -1; |
| 205 | } |
| 206 | |
| 207 | /** |
| 208 | * @brief Send a potentially large payload with automatic fragmentation |
| 209 | * |
| 210 | * This function handles large message fragmentation to ensure browser compatibility. |
| 211 | * It compresses the message if requested and then splits it into fragments |
| 212 | * that are smaller than WS_MAX_OUTGOING_FRAME_SIZE. |
| 213 | * |
| 214 | * @param wsc WebSocket client |
| 215 | * @param payload Payload data to send |
| 216 | * @param payload_len Length of the payload |
| 217 | * @param opcode WebSocket opcode (text, binary) |
| 218 | * @param use_compression Whether to attempt compression |
| 219 | * @return Total number of bytes sent or -1 on error |
| 220 | */ |
| 221 | int websocket_protocol_send_payload( |
| 222 | WS_CLIENT *wsc, const char *payload, size_t payload_len, |
| 223 | WEBSOCKET_OPCODE opcode, bool use_compression) { |
| 224 | |
| 225 | if (!wsc || wsc->sock.fd < 0) |
| 226 | return -1; |
| 227 | |
| 228 | // Control frames should never use this function as they can't be fragmented |
| 229 | if (websocket_frame_is_control_opcode(opcode) || !payload || !payload_len) { |
| 230 | return websocket_protocol_send_frame(wsc, payload, payload_len, opcode, false, true); |
| 231 | } |
| 232 | |
| 233 | // Attempt compression if requested and conditions are met |
| 234 | bool compressed = false; |
| 235 | const char *data_to_send = payload; |
| 236 | size_t data_len = payload_len; |
| 237 | |
| 238 | if (use_compression) { |
| 239 | compressed = websocket_client_compress_message(wsc, payload, payload_len); |
| 240 | if (compressed) { |
| 241 | data_to_send = wsb_data(&wsc->c_payload); |
| 242 | data_len = wsb_length(&wsc->c_payload); |
| 243 | |
| 244 | websocket_debug(wsc, "Using compressed payload for transmission (%zu -> %zu bytes)", |
| 245 | payload_len, data_len); |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // Check if we need to fragment the message |
| 250 | if (data_len <= wsc->max_outbound_frame_size) { |
| 251 | // Small enough to send in a single frame |
| 252 | return websocket_protocol_send_frame(wsc, data_to_send, data_len, opcode, compressed, true); |
| 253 | } |
| 254 | |
| 255 | // We need to fragment the message |
| 256 | websocket_debug(wsc, "Fragmenting large message (%zu bytes) into frames of max %zu bytes", |
| 257 | data_len, wsc->max_outbound_frame_size); |
| 258 | |
| 259 | size_t total_sent = 0; |
| 260 | size_t bytes_remaining = data_len; |
| 261 | size_t offset = 0; |
| 262 | bool first_frame = true; |
| 263 | |
| 264 | // Send fragments |
| 265 | while (bytes_remaining > 0) { |
| 266 | // Determine size for this fragment |
| 267 | size_t fragment_size = MIN(bytes_remaining, wsc->max_outbound_frame_size); |
| 268 | bool is_final = (fragment_size == bytes_remaining); |
| 269 | |
| 270 | // First frame uses original opcode, subsequent frames use continuation |
| 271 | WEBSOCKET_OPCODE frame_opcode = first_frame ? opcode : WS_OPCODE_CONTINUATION; |
| 272 | |
| 273 | // Compression flag is only set on the first frame |
| 274 | bool frame_compressed = compressed && first_frame; |
| 275 | |
| 276 | // Send this fragment |
| 277 | int result = websocket_protocol_send_frame( |
| 278 | wsc, |
| 279 | data_to_send + offset, |
| 280 | fragment_size, |
| 281 | frame_opcode, |
| 282 | frame_compressed, |
| 283 | is_final |
| 284 | ); |
| 285 | |
| 286 | if (result < 0) { |
| 287 | // Failed to send frame |
| 288 | websocket_error(wsc, "Failed to send message fragment at offset %zu", offset); |
| 289 | return -1; |
| 290 | } |
| 291 | |
| 292 | // Update counters for next iteration |
| 293 | total_sent += result; |
| 294 | offset += fragment_size; |
| 295 | bytes_remaining -= fragment_size; |
| 296 | first_frame = false; |
| 297 | } |
| 298 | |
| 299 | websocket_debug(wsc, "Successfully sent fragmented message in multiple frames, total bytes: %zu", total_sent); |
| 300 | return total_sent; |
| 301 | } |
| 302 | |
| 303 | /** |
| 304 | * @brief Send a text message with automatic fragmentation |
| 305 | * |
| 306 | * This function sends a WebSocket text message with automatic compression and fragmentation. |
| 307 | * |
| 308 | * @param wsc WebSocket client |
| 309 | * @param text Text to send |
| 310 | * @return Number of bytes sent or -1 on error |
| 311 | */ |
| 312 | int websocket_protocol_send_text(WS_CLIENT *wsc, const char *text) { |
| 313 | if (!wsc) |
| 314 | return -1; |
| 315 | |
| 316 | size_t text_len = strlen(text); |
| 317 | |
| 318 | websocket_debug(wsc, "Sending text message, length=%zu", text_len); |
| 319 | |
| 320 | // Dump text message for debugging |
| 321 | websocket_dump_debug(wsc, text, text_len, "TX TEXT MSG"); |
| 322 | |
| 323 | // Enable compression for text messages by default, with automatic fragmentation |
| 324 | return websocket_protocol_send_payload(wsc, text, text_len, WS_OPCODE_TEXT, true); |
| 325 | } |
| 326 | |
| 327 | /** |
| 328 | * @brief Send a binary message with automatic fragmentation |
| 329 | * |
| 330 | * This function sends a WebSocket binary message with automatic compression and fragmentation. |
| 331 | * |
| 332 | * @param wsc WebSocket client |
| 333 | * @param data Binary data to send |
| 334 | * @param length Length of the binary data |
| 335 | * @return Number of bytes sent or -1 on error |
| 336 | */ |
| 337 | int websocket_protocol_send_binary(WS_CLIENT *wsc, const void *data, size_t length) { |
| 338 | if (!wsc) |
| 339 | return -1; |
| 340 | |
| 341 | websocket_debug(wsc, "Sending binary message, length=%zu", length); |
| 342 | |
| 343 | // Dump binary message for debugging |
| 344 | websocket_dump_debug(wsc, data, length, "TX BIN MSG"); |
| 345 | |
| 346 | // Enable compression for binary messages by default, with automatic fragmentation |
| 347 | return websocket_protocol_send_payload(wsc, data, length, WS_OPCODE_BINARY, true); |
| 348 | } |
| 349 | |
| 350 | // Send a close frame |
| 351 | int websocket_protocol_send_close(WS_CLIENT *wsc, WEBSOCKET_CLOSE_CODE code, const char *reason) { |
| 352 | if (!wsc || wsc->sock.fd < 0) |
| 353 | return -1; |
| 354 | |
| 355 | // Only send a close frame if we're in a valid state to do so |
| 356 | // Per RFC 6455: An endpoint MUST NOT send any more data frames after sending a Close frame |
| 357 | // CLOSING_CLIENT means we already responded to a client's close and shouldn't send another |
| 358 | if (wsc->state == WS_STATE_CLOSED || |
| 359 | wsc->state == WS_STATE_CLOSING_SERVER || |
| 360 | wsc->state == WS_STATE_CLOSING_CLIENT) |
| 361 | return -1; |
| 362 | |
| 363 | // Validate close code |
| 364 | if (!websocket_validate_close_code((uint16_t)code)) { |
| 365 | websocket_error(wsc, "Invalid close code: %d (%s)", code, WEBSOCKET_CLOSE_CODE_2str(code)); |
| 366 | code = WS_CLOSE_PROTOCOL_ERROR; |
| 367 | reason = "Invalid close code"; |
| 368 | } |
| 369 | |
| 370 | // Prepare close payload: 2-byte code + optional reason text |
| 371 | size_t reason_len = reason ? strlen(reason) : 0; |
| 372 | |
| 373 | // Control frames max size is 125 bytes |
| 374 | if (reason_len > 123) { |
| 375 | websocket_error(wsc, "Close frame payload too large: %zu bytes (max 123)", reason_len); |
| 376 | reason_len = 123; // Truncate reason to fit |
| 377 | } |
| 378 | |
| 379 | // Control frames are capped at 125 bytes, so a fixed stack buffer is sufficient. |
| 380 | size_t payload_len = 2 + reason_len; |
| 381 | char payload[125]; |
| 382 | |
| 383 | // Set status code in network byte order (big-endian) |
| 384 | uint16_t code_value = (uint16_t)code; |
| 385 | payload[0] = (code_value >> 8) & 0xFF; |
| 386 | payload[1] = code & 0xFF; |
| 387 | |
| 388 | // Add reason if provided (truncate if necessary) |
| 389 | if (reason && reason_len > 0) |
| 390 | memcpy(payload + 2, reason, reason_len); |
| 391 | |
| 392 | // Call the close handler if registered - this is used to inject a message on close if needed |
| 393 | if(wsc->on_close) |
| 394 | wsc->on_close(wsc, WS_CLOSE_GOING_AWAY, reason); |
| 395 | |
| 396 | // Send close frame (never compressed, always final) |
| 397 | int result = websocket_protocol_send_frame(wsc, payload, payload_len, WS_OPCODE_CLOSE, false, true); |
| 398 | |
| 399 | return result; |
| 400 | } |
| 401 | |
| 402 | // Send a ping frame |
| 403 | int websocket_protocol_send_ping(WS_CLIENT *wsc, const char *data, size_t length) { |
| 404 | if (!wsc) |
| 405 | return -1; |
| 406 | |
| 407 | // Control frames max size is 125 bytes |
| 408 | if (length > 125) { |
| 409 | websocket_error(wsc, "Ping frame payload too large: %zu bytes (max: 125)", |
| 410 | length); |
| 411 | return -1; |
| 412 | } |
| 413 | |
| 414 | // If no data provided, use empty ping |
| 415 | if (!data || length == 0) { |
| 416 | data = ""; |
| 417 | length = 0; |
| 418 | } |
| 419 | |
| 420 | // Send ping frame (never compressed, always final) |
| 421 | return websocket_protocol_send_frame(wsc, data, length, WS_OPCODE_PING, false, true); |
| 422 | } |
| 423 | |
| 424 | // Send a pong frame |
| 425 | int websocket_protocol_send_pong(WS_CLIENT *wsc, const char *data, size_t length) { |
| 426 | if (!wsc) |
| 427 | return -1; |
| 428 | |
| 429 | // Control frames max size is 125 bytes |
| 430 | if (length > 125) { |
| 431 | websocket_error(wsc, "Pong frame payload too large: %zu bytes (max: 125)", |
| 432 | length); |
| 433 | return -1; |
| 434 | } |
| 435 | |
| 436 | // If no data provided, use empty pong |
| 437 | if (!data || length == 0) { |
| 438 | data = ""; |
| 439 | length = 0; |
| 440 | } |
| 441 | |
| 442 | // Send pong frame (never compressed, always final) |
| 443 | return websocket_protocol_send_frame(wsc, data, length, WS_OPCODE_PONG, false, true); |
| 444 | } |