lib/http/src/websocket.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3
4 const log = std.log.scoped(.http_ws);
5 const connection_mod = @import("connection.zig");
6 const Connection = connection_mod.Connection;
7 const http_message = @import("message.zig");
8 const Request = http_message.Request;
9 const sys = @import("sys").net;
10 const time = @import("sys").time;
11
12 const maximum_frame_header_bytes: usize = 14;
13 const maximum_control_payload_bytes: usize = 125;
14 const maximum_protocol_payload_bytes: u128 = (@as(u128, 1) << 63) - 1;
15 const mask_vector_bytes: usize = 16;
16 const MaskVector = @Vector(mask_vector_bytes, u8);
17
18 pub const FrameError = error{
19 IncompleteFrame,
20 InvalidOpcode,
21 InvalidFrame,
22 OutputTooSmall,
23 PayloadTooLarge,
24 MaskRequired,
25 };
26
27 pub const HandshakeError = error{
28 InvalidMethod,
29 InvalidHttpVersion,
30 InvalidUpgrade,
31 MissingVersion,
32 UnsupportedVersion,
33 MissingKey,
34 InvalidKey,
35 };
36
37 pub const Limits = struct {
38 frame_payload_bytes: usize,
39 message_payload_bytes: usize,
40 };
41
42 pub const Capacity = struct {
43 frame_payload_bytes: usize,
44 message_payload_bytes: usize,
45 frame_bytes: usize,
46 storage_bytes: usize,
47
48 pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {
49 if (@as(u128, limits.frame_payload_bytes) > maximum_protocol_payload_bytes) {
50 return error.CapacityOverflow;
51 }
52 const frame_bytes = try alloc_phase.capacity.add(
53 usize,
54 maximum_frame_header_bytes,
55 limits.frame_payload_bytes,
56 );
57 const storage_bytes = try alloc_phase.capacity.add(
58 usize,
59 frame_bytes,
60 limits.message_payload_bytes,
61 );
62 return .{
63 .frame_payload_bytes = limits.frame_payload_bytes,
64 .message_payload_bytes = limits.message_payload_bytes,
65 .frame_bytes = frame_bytes,
66 .storage_bytes = storage_bytes,
67 };
68 }
69 };
70
71 pub const StorageExhaustion = error{
72 FramePayloadCapacityExceeded,
73 MessagePayloadCapacityExceeded,
74 };
75
76 const StorageLimits = Limits;
77 const StorageCapacity = Capacity;
78
79 pub const Storage = struct {
80 phase: alloc_phase.capacity.Phase,
81 capacity: StorageCapacity,
82 bytes: []u8,
83
84 pub const Limits: type = StorageLimits;
85 pub const Capacity: type = StorageCapacity;
86 pub const Exhaustion: type = StorageExhaustion;
87 pub const InitError = std.mem.Allocator.Error || error{CapacityOverflow};
88
89 pub const claim: alloc_phase.capacity.Declaration = .{
90 .source = .{
91 .id = "http.websocket_storage",
92 .kind = .phase_static,
93 .limit_source = .caller,
94 .storage = .{
95 .covered = &.{
96 .{
97 .id = "one_websocket_wire_frame_and_fragmented_message_byte_region",
98 .lifetime = .steady,
99 .detail = "one WebSocket wire-frame and fragmented-message byte region",
100 },
101 },
102 .excluded = &.{
103 "Connection unread bytes and router session objects",
104 "caller-owned outbound payloads and handler-retained message copies",
105 "socket and kernel queues, protocol callbacks, and application effects",
106 },
107 },
108 .capacity = .{
109 .inputs = &.{
110 alloc_phase.capacity.bindInput(StorageLimits, "frame_payload_bytes", "frame_payload_bytes"),
111 alloc_phase.capacity.bindInput(StorageLimits, "message_payload_bytes", "message_payload_bytes"),
112 },
113 .type_selectors = &.{},
114 .nodes = &.{
115 .{ .constant = 14 },
116 .{ .input = 0 },
117 .{ .input = 1 },
118 .{ .add = .{ .left = 0, .right = 1 } },
119 .{ .add = .{ .left = 3, .right = 2 } },
120 },
121 .assertions = &.{.{
122 .scope = .closure_total,
123 .measure = .retained,
124 .relation = .exact,
125 .expression = 4,
126 }},
127 },
128 .overload = .{
129 .kind = .reject_before_mutation,
130 .detail = "oversize admission preserves prior fragmented-message bytes",
131 },
132 .risks = .{
133 .transitive = .{
134 .status = .open,
135 .detail = "router handlers can retain borrowed messages or allocate independently",
136 },
137 .foreign = .{
138 .status = .open,
139 .detail = "frame I/O crosses socket, kernel, scheduler, and peer storage",
140 },
141 },
142 .obligations = &.{
143 .{ .key = "http_websocket_capacity", .role = .capacity_model },
144 .{ .key = "http_websocket_oom_retry", .role = .custom },
145 .{ .key = "http_websocket_sealed", .role = .overload },
146 .{ .key = "http_websocket_fragmented", .role = .custom },
147 .{ .key = "http_websocket_upgrade", .role = .custom },
148 .{ .key = "http_websocket_semantics", .role = .custom },
149 },
150 },
151 .bindings = .{
152 .owner = @This(),
153 .seal = .{
154 .family = alloc_phase.capacity.selector(@This().activate),
155 .premise = .{
156 .class = .checked_semantic_fact,
157 .authority = .checker,
158 },
159 },
160 .teardown = .{
161 .family = alloc_phase.capacity.selector(@This().deinit),
162 .premise = .{
163 .class = .checked_semantic_fact,
164 .authority = .checker,
165 },
166 },
167 },
168 };
169
170 pub fn init(allocator: std.mem.Allocator, limits: StorageLimits) InitError!Storage {
171 const capacity = try StorageCapacity.derive(limits);
172 const bytes = if (capacity.storage_bytes == 0)
173 @as([]u8, &.{})
174 else
175 try allocator.alloc(u8, capacity.storage_bytes);
176 return .{
177 .phase = .initialization,
178 .capacity = capacity,
179 .bytes = bytes,
180 };
181 }
182
183 pub fn activate(self: *Storage) void {
184 std.debug.assert(self.phase == .initialization);
185 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
186 self.phase = .steady;
187 }
188
189 pub fn frame(self: *Storage, payload_bytes: usize) Exhaustion![]u8 {
190 std.debug.assert(self.phase != .teardown);
191 if (payload_bytes > self.capacity.frame_payload_bytes) {
192 return error.FramePayloadCapacityExceeded;
193 }
194 return self.bytes[0 .. maximum_frame_header_bytes + payload_bytes];
195 }
196
197 pub fn message(self: *Storage, payload_bytes: usize) Exhaustion![]u8 {
198 std.debug.assert(self.phase != .teardown);
199 if (payload_bytes > self.capacity.message_payload_bytes) {
200 return error.MessagePayloadCapacityExceeded;
201 }
202 const start = self.capacity.frame_bytes;
203 return self.bytes[start..][0..payload_bytes];
204 }
205
206 pub fn deinit(self: *Storage, allocator: std.mem.Allocator) void {
207 std.debug.assert(self.phase != .teardown);
208 std.debug.assert(self.bytes.len == self.capacity.storage_bytes);
209 self.phase = .teardown;
210 if (self.bytes.len != 0) allocator.free(self.bytes);
211 self.bytes = &.{};
212 }
213 };
214
215 comptime {
216 alloc_phase.capacity.requireAllocatorRejectingOwnerShape(Storage);
217 }
218
219 pub const Opcode = enum(u4) {
220 continuation = 0x0,
221 text = 0x1,
222 binary = 0x2,
223 close = 0x8,
224 ping = 0x9,
225 pong = 0xA,
226
227 pub fn fromWire(value: u8) FrameError!Opcode {
228 return switch (value) {
229 0x0 => .continuation,
230 0x1 => .text,
231 0x2 => .binary,
232 0x8 => .close,
233 0x9 => .ping,
234 0xA => .pong,
235 else => FrameError.InvalidOpcode,
236 };
237 }
238
239 pub fn isControl(self: Opcode) bool {
240 return @backingInt(self) >= 0x8;
241 }
242 };
243
244 pub const FrameParseResult = struct {
245 frame: Frame,
246 consumed: usize,
247 };
248
249 inline fn maskPayload(source: []const u8, target: []u8, mask: [4]u8) void {
250 std.debug.assert(source.len == target.len);
251 var index: usize = 0;
252 if (source.len >= mask_vector_bytes) {
253 const vector_mask: MaskVector = .{
254 mask[0], mask[1], mask[2], mask[3],
255 mask[0], mask[1], mask[2], mask[3],
256 mask[0], mask[1], mask[2], mask[3],
257 mask[0], mask[1], mask[2], mask[3],
258 };
259 while (source.len - index >= mask_vector_bytes) : (index += mask_vector_bytes) {
260 const input: MaskVector = @bitCast(source[index..][0..mask_vector_bytes].*);
261 target[index..][0..mask_vector_bytes].* = @bitCast(input ^ vector_mask);
262 }
263 }
264 for (source[index..], target[index..], 0..) |input, *output, tail_index| {
265 output.* = input ^ mask[tail_index % mask.len];
266 }
267 }
268
269 pub const Frame = struct {
270 fin: bool,
271 opcode: Opcode,
272 mask: ?[4]u8,
273 payload: []const u8,
274
275 pub fn parse(data: []u8, maximum_payload_bytes: usize) FrameError!FrameParseResult {
276 if (data.len < 2) return FrameError.IncompleteFrame;
277 if ((data[0] & 0x70) != 0) return FrameError.InvalidFrame;
278
279 const fin = (data[0] & 0x80) != 0;
280 const opcode = try Opcode.fromWire(data[0] & 0x0F);
281
282 const masked = (data[1] & 0x80) != 0;
283 const len7 = data[1] & 0x7F;
284
285 var header_len: usize = 2;
286 var payload_len: usize = undefined;
287
288 if (len7 < 126) {
289 payload_len = len7;
290 } else if (len7 == 126) {
291 if (data.len < 4) return FrameError.IncompleteFrame;
292 payload_len = std.mem.readInt(u16, data[2..4], .big);
293 if (payload_len < 126) return FrameError.InvalidFrame;
294 header_len = 4;
295 } else {
296 if (data.len < 10) return FrameError.IncompleteFrame;
297 const len64 = std.mem.readInt(u64, data[2..10], .big);
298 if (len64 < 65536) return FrameError.InvalidFrame;
299 if (@as(u128, len64) > maximum_protocol_payload_bytes) {
300 return FrameError.InvalidFrame;
301 }
302 if (@as(u128, len64) > @as(u128, std.math.maxInt(usize))) {
303 return FrameError.PayloadTooLarge;
304 }
305 payload_len = @intCast(len64);
306 header_len = 10;
307 }
308
309 try validateFrameHeader(fin, opcode, payload_len);
310 if (payload_len > maximum_payload_bytes) return FrameError.PayloadTooLarge;
311
312 const mask_key_offset = header_len;
313 if (masked) {
314 header_len += 4;
315 }
316
317 const total_len = std.math.add(usize, header_len, payload_len) catch {
318 return FrameError.PayloadTooLarge;
319 };
320 if (data.len < total_len) return FrameError.IncompleteFrame;
321
322 const mask: ?[4]u8 = if (masked)
323 data[mask_key_offset..][0..4].*
324 else
325 null;
326
327 const payload = data[header_len..total_len];
328 if (mask) |m| {
329 maskPayload(payload, payload, m);
330 }
331
332 return .{
333 .frame = .{
334 .fin = fin,
335 .opcode = opcode,
336 .mask = mask,
337 .payload = payload,
338 },
339 .consumed = total_len,
340 };
341 }
342
343 pub fn serializedLength(
344 self: *const Frame,
345 maximum_payload_bytes: usize,
346 ) FrameError!usize {
347 const header_bytes = try self.headerLength(maximum_payload_bytes);
348 return std.math.add(usize, header_bytes, self.payload.len) catch {
349 return FrameError.PayloadTooLarge;
350 };
351 }
352
353 pub fn serializeInto(
354 self: *const Frame,
355 output: []u8,
356 maximum_payload_bytes: usize,
357 ) FrameError![]u8 {
358 const total_bytes = try self.serializedLength(maximum_payload_bytes);
359 if (output.len < total_bytes) return FrameError.OutputTooSmall;
360 const target = output[0..total_bytes];
361 const header = try self.writeHeader(target, maximum_payload_bytes);
362 const payload = target[header.len..];
363 if (self.mask) |mask| {
364 maskPayload(self.payload, payload, mask);
365 } else {
366 @memcpy(payload, self.payload);
367 }
368 return target;
369 }
370
371 pub fn serializeHeaderInto(
372 self: *const Frame,
373 output: []u8,
374 maximum_payload_bytes: usize,
375 ) FrameError![]u8 {
376 return try self.writeHeader(
377 output,
378 maximum_payload_bytes,
379 );
380 }
381
382 pub fn parseServer(data: []u8, maximum_payload_bytes: usize) FrameError!FrameParseResult {
383 if (data.len < 2) return FrameError.IncompleteFrame;
384 if ((data[1] & 0x80) == 0) return FrameError.MaskRequired;
385
386 return parse(data, maximum_payload_bytes);
387 }
388
389 fn headerLength(self: *const Frame, maximum_payload_bytes: usize) FrameError!usize {
390 const payload_bytes = self.payload.len;
391 try validateFrameHeader(self.fin, self.opcode, payload_bytes);
392 if (payload_bytes > maximum_payload_bytes) return FrameError.PayloadTooLarge;
393 if (@as(u128, payload_bytes) > maximum_protocol_payload_bytes) {
394 return FrameError.PayloadTooLarge;
395 }
396 const length_bytes: usize = if (payload_bytes < 126)
397 0
398 else if (payload_bytes <= std.math.maxInt(u16))
399 2
400 else
401 8;
402 return 2 + length_bytes + @as(usize, if (self.mask == null) 0 else 4);
403 }
404
405 fn writeHeader(
406 self: *const Frame,
407 output: []u8,
408 maximum_payload_bytes: usize,
409 ) FrameError![]u8 {
410 const header_bytes = try self.headerLength(maximum_payload_bytes);
411 if (output.len < header_bytes) return FrameError.OutputTooSmall;
412 const target = output[0..header_bytes];
413 target[0] = @as(u8, if (self.fin) 0x80 else 0) | @backingInt(self.opcode);
414
415 const mask_bit: u8 = if (self.mask == null) 0 else 0x80;
416 var index: usize = 1;
417 if (self.payload.len < 126) {
418 target[index] = mask_bit | @as(u8, @intCast(self.payload.len));
419 index += 1;
420 } else if (self.payload.len <= std.math.maxInt(u16)) {
421 target[index] = mask_bit | 126;
422 index += 1;
423 std.mem.writeInt(u16, target[index..][0..2], @intCast(self.payload.len), .big);
424 index += 2;
425 } else {
426 target[index] = mask_bit | 127;
427 index += 1;
428 std.mem.writeInt(u64, target[index..][0..8], @intCast(self.payload.len), .big);
429 index += 8;
430 }
431 if (self.mask) |mask| {
432 @memcpy(target[index..][0..mask.len], &mask);
433 index += mask.len;
434 }
435 std.debug.assert(index == target.len);
436 return target;
437 }
438 };
439
440 fn validateFrameHeader(fin: bool, opcode: Opcode, payload_len: usize) FrameError!void {
441 if (opcode.isControl() and (!fin or payload_len > maximum_control_payload_bytes)) {
442 return FrameError.InvalidFrame;
443 }
444 }
445
446 const FrameLengthEncoding = enum {
447 shortest,
448 extended16,
449 extended64,
450 };
451
452 fn rawFrameBytes(
453 output: []u8,
454 fin: bool,
455 opcode_value: u8,
456 mask: ?[4]u8,
457 payload: []const u8,
458 length_encoding: FrameLengthEncoding,
459 ) FrameError![]u8 {
460 const length_bytes: usize = switch (length_encoding) {
461 .shortest => 0,
462 .extended16 => 2,
463 .extended64 => 8,
464 };
465 const header_bytes = 2 + length_bytes + @as(usize, if (mask == null) 0 else 4);
466 const total_bytes = std.math.add(usize, header_bytes, payload.len) catch {
467 return FrameError.PayloadTooLarge;
468 };
469 if (output.len < total_bytes) return FrameError.OutputTooSmall;
470 switch (length_encoding) {
471 .shortest => if (payload.len > 125) return FrameError.PayloadTooLarge,
472 .extended16 => if (payload.len > std.math.maxInt(u16)) {
473 return FrameError.PayloadTooLarge;
474 },
475 .extended64 => if (@as(u128, payload.len) > maximum_protocol_payload_bytes) {
476 return FrameError.PayloadTooLarge;
477 },
478 }
479 const bytes = output[0..total_bytes];
480 bytes[0] = @as(u8, if (fin) 0x80 else 0) | opcode_value;
481 const mask_bit: u8 = if (mask != null) 0x80 else 0;
482 var index: usize = 1;
483 switch (length_encoding) {
484 .shortest => {
485 bytes[index] = mask_bit | @as(u8, @intCast(payload.len));
486 index += 1;
487 },
488 .extended16 => {
489 bytes[index] = mask_bit | 126;
490 index += 1;
491 std.mem.writeInt(u16, bytes[index..][0..2], @intCast(payload.len), .big);
492 index += 2;
493 },
494 .extended64 => {
495 bytes[index] = mask_bit | 127;
496 index += 1;
497 std.mem.writeInt(u64, bytes[index..][0..8], @intCast(payload.len), .big);
498 index += 8;
499 },
500 }
501
502 if (mask) |value| {
503 @memcpy(bytes[index..][0..value.len], &value);
504 index += value.len;
505 for (payload, bytes[index..], 0..) |source, *target, payload_index| {
506 target.* = source ^ value[payload_index % value.len];
507 }
508 } else {
509 @memcpy(bytes[index..], payload);
510 }
511 return bytes;
512 }
513
514 fn expectFrameParseError(
515 expected: FrameError,
516 data: []u8,
517 ) !void {
518 const result = Frame.parse(data, std.math.maxInt(usize));
519 if (result) |_| {
520 return error.MalformedFrameAccepted;
521 } else |err| {
522 try std.testing.expectEqual(expected, err);
523 }
524 }
525
526 pub const WebSocketState = enum {
527 open,
528 closing,
529 closed,
530 };
531
532 pub const WebSocketCloseCode = struct {
533 pub const normal: u16 = 1000;
534 pub const going_away: u16 = 1001;
535 pub const protocol_error: u16 = 1002;
536 pub const unsupported_data: u16 = 1003;
537 pub const no_status: u16 = 1005;
538 pub const abnormal: u16 = 1006;
539 pub const invalid_payload: u16 = 1007;
540 pub const policy_violation: u16 = 1008;
541 pub const message_too_big: u16 = 1009;
542 pub const mandatory_extension: u16 = 1010;
543 pub const internal_error: u16 = 1011;
544 pub const tls_handshake: u16 = 1015;
545 };
546
547 pub const WebSocketClientHandshake = struct {
548 key: []const u8,
549
550 pub fn accept(self: WebSocketClientHandshake) [28]u8 {
551 return accept_key(self.key);
552 }
553 };
554
555 pub const WebSocketMessage = struct {
556 opcode: Opcode,
557 payload: []const u8,
558 };
559
560 const SessionLimits = Limits;
561 const SessionCapacity = Capacity;
562 const SessionStorage = Storage;
563
564 pub const WebSocket = struct {
565 connection: *Connection,
566 state: WebSocketState,
567 storage: SessionStorage,
568 read_length: usize = 0,
569 pending_consumed: usize = 0,
570 fragment_opcode: ?Opcode,
571 fragment_length: usize = 0,
572 fragment_returned: bool = false,
573
574 pub const Limits: type = SessionLimits;
575 pub const Capacity: type = SessionCapacity;
576 pub const Storage: type = SessionStorage;
577
578 pub fn init(
579 allocator: std.mem.Allocator,
580 limits: SessionLimits,
581 connection: *Connection,
582 ) SessionStorage.InitError!WebSocket {
583 return .{
584 .connection = connection,
585 .state = .open,
586 .storage = try SessionStorage.init(allocator, limits),
587 .fragment_opcode = null,
588 };
589 }
590
591 pub fn activate(self: *WebSocket) void {
592 self.storage.activate();
593 }
594
595 pub fn deinit(self: *WebSocket, allocator: std.mem.Allocator) void {
596 self.storage.deinit(allocator);
597 self.* = undefined;
598 }
599
600 pub fn capacity(self: *const WebSocket) SessionCapacity {
601 return self.storage.capacity;
602 }
603
604 pub fn validateClientHandshake(req: *const Request) HandshakeError!WebSocketClientHandshake {
605 if (req.method != .GET) return HandshakeError.InvalidMethod;
606 if (req.version != .http_1_1) return HandshakeError.InvalidHttpVersion;
607 if (!req.isWebSocketUpgrade()) return HandshakeError.InvalidUpgrade;
608
609 const version = req.headers.get("Sec-WebSocket-Version") orelse return HandshakeError.MissingVersion;
610 if (!std.mem.eql(u8, version, "13")) return HandshakeError.UnsupportedVersion;
611
612 const key = req.getWebSocketKey() orelse return HandshakeError.MissingKey;
613 try validateClientKey(key);
614
615 return .{ .key = key };
616 }
617
618 pub fn validateClientKey(key: []const u8) HandshakeError!void {
619 const decoded_len = std.base64.standard.Decoder.calcSizeForSlice(key) catch return HandshakeError.InvalidKey;
620 if (decoded_len != 16) return HandshakeError.InvalidKey;
621
622 var decoded: [16]u8 = undefined;
623 std.base64.standard.Decoder.decode(&decoded, key) catch return HandshakeError.InvalidKey;
624 }
625
626 pub fn handshake(key: []const u8) HandshakeError![28]u8 {
627 try validateClientKey(key);
628 return accept_key(key);
629 }
630
631 pub fn sendText(self: *WebSocket, data: []const u8) !void {
632 try self.sendFrame(.{
633 .fin = true,
634 .opcode = .text,
635 .mask = null,
636 .payload = data,
637 });
638 }
639
640 pub fn sendBinary(self: *WebSocket, data: []const u8) !void {
641 try self.sendFrame(.{
642 .fin = true,
643 .opcode = .binary,
644 .mask = null,
645 .payload = data,
646 });
647 }
648
649 pub fn sendPing(self: *WebSocket, data: []const u8) !void {
650 try self.sendFrame(.{
651 .fin = true,
652 .opcode = .ping,
653 .mask = null,
654 .payload = data,
655 });
656 }
657
658 pub fn sendPong(self: *WebSocket, data: []const u8) !void {
659 try self.sendFrame(.{
660 .fin = true,
661 .opcode = .pong,
662 .mask = null,
663 .payload = data,
664 });
665 }
666
667 pub fn close(self: *WebSocket, code: u16, reason: []const u8) !void {
668 if (self.state == .closed) return;
669 const payload_len = std.math.add(usize, 2, reason.len) catch {
670 return FrameError.InvalidFrame;
671 };
672 if (payload_len > maximum_control_payload_bytes) return FrameError.InvalidFrame;
673
674 var payload: [maximum_control_payload_bytes]u8 = undefined;
675 std.mem.writeInt(u16, payload[0..2], code, .big);
676 @memcpy(payload[2..][0..reason.len], reason);
677
678 try self.sendFrame(.{
679 .fin = true,
680 .opcode = .close,
681 .mask = null,
682 .payload = payload[0..payload_len],
683 });
684
685 self.state = .closing;
686 }
687
688 pub fn receive(self: *WebSocket) !?WebSocketMessage {
689 std.debug.assert(self.storage.phase == .steady);
690 self.releaseBorrowedMessage();
691 while (self.state != .closed) {
692 const buffered_length = self.read_length;
693 if (try self.processBufferedFrame()) |message| return message;
694 if (self.state == .closed) return null;
695 if (self.read_length != buffered_length) continue;
696 try self.readFrameBytes();
697 }
698 return null;
699 }
700
701 fn processBufferedFrame(self: *WebSocket) !?WebSocketMessage {
702 if (self.read_length == 0) return null;
703 const frame_region = self.frameRegion();
704 const parsed = Frame.parseServer(
705 frame_region[0..self.read_length],
706 self.storage.capacity.frame_payload_bytes,
707 ) catch |err| switch (err) {
708 error.IncompleteFrame => return null,
709 error.MaskRequired, error.InvalidOpcode, error.InvalidFrame => {
710 self.rejectProtocol();
711 return null;
712 },
713 error.PayloadTooLarge => {
714 self.rejectMessageTooBig();
715 return null;
716 },
717 else => return err,
718 };
719 if (parsed.frame.opcode.isControl()) {
720 try self.processControlFrame(parsed.frame, parsed.consumed);
721 return null;
722 }
723 return self.processDataFrame(parsed.frame, parsed.consumed);
724 }
725
726 fn processControlFrame(self: *WebSocket, frame: Frame, consumed: usize) !void {
727 defer self.consumeFrame(consumed);
728 switch (frame.opcode) {
729 .ping => try self.sendPong(frame.payload),
730 .pong => {},
731 .close => {
732 if (self.state != .closing) self.replyToClose(frame.payload);
733 self.state = .closed;
734 self.connection.markClosing();
735 },
736 else => unreachable,
737 }
738 }
739
740 fn replyToClose(self: *WebSocket, payload: []const u8) void {
741 if (payload.len >= 2) {
742 const code = std.mem.readInt(u16, payload[0..2], .big);
743 self.close(code, payload[2..]) catch |err| {
744 log.warn("failed to send close response: {s}", .{@errorName(err)});
745 };
746 } else {
747 self.close(WebSocketCloseCode.normal, "") catch |err| {
748 log.warn("failed to send close response: {s}", .{@errorName(err)});
749 };
750 }
751 }
752
753 fn processDataFrame(self: *WebSocket, frame: Frame, consumed: usize) ?WebSocketMessage {
754 if (frame.fin) return self.finishDataFrame(frame, consumed);
755 if (frame.opcode == .continuation) {
756 if (self.fragment_opcode == null) {
757 self.rejectProtocol();
758 return null;
759 }
760 } else if (self.fragment_opcode == null) {
761 self.fragment_opcode = frame.opcode;
762 self.fragment_length = 0;
763 } else {
764 self.rejectProtocol();
765 return null;
766 }
767 if (!self.retainFragment(frame.payload)) return null;
768 self.consumeFrame(consumed);
769 return null;
770 }
771
772 fn finishDataFrame(self: *WebSocket, frame: Frame, consumed: usize) ?WebSocketMessage {
773 if (self.fragment_opcode) |opcode| {
774 if (frame.opcode != .continuation) {
775 self.rejectProtocol();
776 return null;
777 }
778 if (!self.retainFragment(frame.payload)) return null;
779 self.consumeFrame(consumed);
780 self.fragment_opcode = null;
781 self.fragment_returned = true;
782 return .{
783 .opcode = opcode,
784 .payload = self.messageRegion()[0..self.fragment_length],
785 };
786 }
787 if (frame.opcode == .continuation) {
788 self.rejectProtocol();
789 return null;
790 }
791 if (frame.payload.len > self.storage.capacity.message_payload_bytes) {
792 self.rejectMessageTooBig();
793 return null;
794 }
795 self.pending_consumed = consumed;
796 return .{ .opcode = frame.opcode, .payload = frame.payload };
797 }
798
799 fn retainFragment(self: *WebSocket, payload: []const u8) bool {
800 if (!fragmentLengthAllowed(
801 self.fragment_length,
802 payload.len,
803 self.storage.capacity.message_payload_bytes,
804 )) {
805 self.rejectMessageTooBig();
806 return false;
807 }
808 self.appendFragment(payload);
809 return true;
810 }
811
812 fn readFrameBytes(self: *WebSocket) !void {
813 const frame_region = self.frameRegion();
814 if (self.read_length == frame_region.len) {
815 self.rejectMessageTooBig();
816 return;
817 }
818 const read = self.connection.read(
819 frame_region[self.read_length..],
820 ) catch |err| switch (err) {
821 error.WouldBlock => return error.WouldBlock,
822 error.ConnectionClosed => {
823 self.state = .closed;
824 self.connection.markClosing();
825 return;
826 },
827 else => return err,
828 };
829 if (read == 0) {
830 self.state = .closed;
831 self.connection.markClosing();
832 return;
833 }
834 self.read_length += read;
835 }
836
837 fn sendFrame(self: *WebSocket, frame: Frame) !void {
838 std.debug.assert(self.storage.phase == .steady);
839 std.debug.assert(frame.mask == null);
840 var header: [maximum_frame_header_bytes]u8 = undefined;
841 const encoded = try frame.writeHeader(&header, std.math.maxInt(usize));
842 try self.connection.write(encoded);
843 if (frame.payload.len != 0) try self.connection.write(frame.payload);
844 }
845
846 fn frameRegion(self: *WebSocket) []u8 {
847 return self.storage.frame(self.storage.capacity.frame_payload_bytes) catch unreachable;
848 }
849
850 fn messageRegion(self: *WebSocket) []u8 {
851 return self.storage.message(self.storage.capacity.message_payload_bytes) catch unreachable;
852 }
853
854 fn appendFragment(self: *WebSocket, payload: []const u8) void {
855 const message = self.messageRegion();
856 @memcpy(message[self.fragment_length..][0..payload.len], payload);
857 self.fragment_length += payload.len;
858 }
859
860 fn consumeFrame(self: *WebSocket, consumed: usize) void {
861 std.debug.assert(consumed <= self.read_length);
862 const remaining = self.read_length - consumed;
863 const frame = self.frameRegion();
864 std.mem.copyForwards(u8, frame[0..remaining], frame[consumed..self.read_length]);
865 self.read_length = remaining;
866 }
867
868 fn releaseBorrowedMessage(self: *WebSocket) void {
869 if (self.pending_consumed != 0) {
870 self.consumeFrame(self.pending_consumed);
871 self.pending_consumed = 0;
872 }
873 if (self.fragment_returned) {
874 self.fragment_length = 0;
875 self.fragment_returned = false;
876 }
877 }
878
879 fn rejectProtocol(self: *WebSocket) void {
880 self.close(WebSocketCloseCode.protocol_error, "Invalid WebSocket frame") catch |err| {
881 log.warn("failed to send protocol error close: {s}", .{@errorName(err)});
882 };
883 self.state = .closed;
884 self.connection.markClosing();
885 }
886
887 fn rejectMessageTooBig(self: *WebSocket) void {
888 self.close(WebSocketCloseCode.message_too_big, "Message too big") catch |close_err| {
889 log.warn("failed to send message-too-big close: {s}", .{@errorName(close_err)});
890 };
891 self.state = .closed;
892 self.connection.markClosing();
893 }
894 };
895
896 fn fragmentLengthAllowed(current: usize, incoming: usize, maximum: usize) bool {
897 return current <= maximum and incoming <= maximum - current;
898 }
899
900 fn accept_key(key: []const u8) [28]u8 {
901 const guid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
902 var hasher = std.crypto.hash.Sha1.init(.{});
903 hasher.update(key);
904 hasher.update(guid);
905 const hash = hasher.finalResult();
906
907 var accept: [28]u8 = undefined;
908 _ = std.base64.standard.Encoder.encode(&accept, &hash);
909 return accept;
910 }
911
912 fn independentCapacity(limits: Limits) error{CapacityOverflow}!Capacity {
913 const frame_bytes = @as(u128, 2 + 8 + 4) + limits.frame_payload_bytes;
914 const storage_bytes = frame_bytes + limits.message_payload_bytes;
915 if (frame_bytes > std.math.maxInt(usize)) return error.CapacityOverflow;
916 if (storage_bytes > std.math.maxInt(usize)) return error.CapacityOverflow;
917 return .{
918 .frame_payload_bytes = limits.frame_payload_bytes,
919 .message_payload_bytes = limits.message_payload_bytes,
920 .frame_bytes = @intCast(frame_bytes),
921 .storage_bytes = @intCast(storage_bytes),
922 };
923 }
924
925 test "WebSocket capacity matches independent protocol arithmetic" {
926 comptime {
927 @stardustClaim(
928 @import("alloc_phase").capacity.witness(Storage, "http_websocket_capacity"),
929 null,
930 null,
931 null,
932 null,
933 null,
934 null,
935 );
936 }
937
938 for (0..4097) |frame_payload_bytes| {
939 const limits = Limits{
940 .frame_payload_bytes = frame_payload_bytes,
941 .message_payload_bytes = 4096 - @min(frame_payload_bytes, 4096),
942 };
943 try std.testing.expectEqual(
944 try independentCapacity(limits),
945 try Capacity.derive(limits),
946 );
947 }
948
949 try std.testing.expectError(
950 error.CapacityOverflow,
951 Capacity.derive(.{
952 .frame_payload_bytes = 0,
953 .message_payload_bytes = std.math.maxInt(usize),
954 }),
955 );
956 if (@sizeOf(usize) >= @sizeOf(u64)) {
957 const maximum_protocol_payload: usize = @intCast(maximum_protocol_payload_bytes);
958 _ = try Capacity.derive(.{
959 .frame_payload_bytes = maximum_protocol_payload,
960 .message_payload_bytes = 0,
961 });
962 try std.testing.expectError(
963 error.CapacityOverflow,
964 Capacity.derive(.{
965 .frame_payload_bytes = maximum_protocol_payload + 1,
966 .message_payload_bytes = 0,
967 }),
968 );
969 }
970 }
971
972 fn checkStorageInitFailures(allocator: std.mem.Allocator) !void {
973 var storage = try Storage.init(allocator, .{
974 .frame_payload_bytes = 1024,
975 .message_payload_bytes = 4096,
976 });
977 storage.deinit(allocator);
978 }
979
980 test "WebSocket storage retries after every allocation failure" {
981 comptime {
982 @stardustClaim(
983 @import("alloc_phase").capacity.witness(Storage, "http_websocket_oom_retry"),
984 null,
985 null,
986 null,
987 null,
988 null,
989 null,
990 );
991 }
992
993 try std.testing.checkAllAllocationFailures(
994 std.testing.allocator,
995 checkStorageInitFailures,
996 .{},
997 );
998
999 var storage = try Storage.init(std.testing.allocator, .{
1000 .frame_payload_bytes = 1024,
1001 .message_payload_bytes = 4096,
1002 });
1003 defer storage.deinit(std.testing.allocator);
1004 storage.activate();
1005 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, storage.phase);
1006 }
1007
1008 test "WebSocket storage seals before frame admission" {
1009 comptime {
1010 @stardustClaim(
1011 @import("alloc_phase").capacity.witness(Storage, "http_websocket_sealed"),
1012 null,
1013 null,
1014 null,
1015 null,
1016 null,
1017 null,
1018 );
1019 }
1020
1021 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
1022 var storage = Storage.init(
1023 phase_allocator.initializationAllocator(),
1024 .{ .frame_payload_bytes = 32, .message_payload_bytes = 64 },
1025 ) catch |err| {
1026 phase_allocator.abortInitialization();
1027 phase_allocator.deinit();
1028 return err;
1029 };
1030 errdefer {
1031 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
1032 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
1033 if (storage.phase != .teardown) storage.deinit(phase_allocator.teardownAllocator());
1034 phase_allocator.deinit();
1035 }
1036
1037 const pointer = storage.bytes.ptr;
1038 const capacity = storage.capacity;
1039 @memset(storage.bytes, 0xa5);
1040 phase_allocator.seal();
1041 storage.activate();
1042
1043 try std.testing.expectError(
1044 error.FramePayloadCapacityExceeded,
1045 storage.frame(storage.capacity.frame_payload_bytes + 1),
1046 );
1047 try std.testing.expectError(
1048 error.MessagePayloadCapacityExceeded,
1049 storage.message(storage.capacity.message_payload_bytes + 1),
1050 );
1051 var short = [_]u8{0xa5};
1052 const frame = Frame{ .fin = true, .opcode = .text, .mask = null, .payload = "x" };
1053 try std.testing.expectError(error.OutputTooSmall, frame.serializeInto(&short, 1));
1054 for (storage.bytes) |byte| try std.testing.expectEqual(@as(u8, 0xa5), byte);
1055 try std.testing.expectEqual(@as(u8, 0xa5), short[0]);
1056 try std.testing.expect(storage.bytes.ptr == pointer);
1057 try std.testing.expectEqual(capacity, storage.capacity);
1058 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
1059
1060 phase_allocator.beginTeardown();
1061 storage.deinit(phase_allocator.teardownAllocator());
1062 phase_allocator.deinit();
1063 }
1064
1065 test "WebSocket masking matches scalar oracle through vector boundaries" {
1066 const maximum_payload_bytes = 4_096;
1067 const mask = [4]u8{ 0x12, 0x34, 0x56, 0x78 };
1068 var source: [maximum_payload_bytes]u8 = undefined;
1069 for (&source, 0..) |*byte, index| byte.* = @truncate(index *% 131 +% 17);
1070 var expected: [maximum_payload_bytes]u8 = undefined;
1071 var out_of_place: [maximum_payload_bytes]u8 = undefined;
1072 var in_place: [maximum_payload_bytes]u8 = undefined;
1073
1074 var payload_bytes: usize = 0;
1075 while (payload_bytes <= maximum_payload_bytes) : (payload_bytes += 1) {
1076 for (source[0..payload_bytes], expected[0..payload_bytes], 0..) |input, *output, index| {
1077 output.* = input ^ mask[index % mask.len];
1078 }
1079 maskPayload(source[0..payload_bytes], out_of_place[0..payload_bytes], mask);
1080 try std.testing.expectEqualSlices(
1081 u8,
1082 expected[0..payload_bytes],
1083 out_of_place[0..payload_bytes],
1084 );
1085
1086 @memcpy(in_place[0..payload_bytes], source[0..payload_bytes]);
1087 maskPayload(in_place[0..payload_bytes], in_place[0..payload_bytes], mask);
1088 try std.testing.expectEqualSlices(
1089 u8,
1090 expected[0..payload_bytes],
1091 in_place[0..payload_bytes],
1092 );
1093 }
1094 }
1095
1096 test "Frame.serialize: small text frame unmasked" {
1097 const frame = Frame{
1098 .fin = true,
1099 .opcode = .text,
1100 .mask = null,
1101 .payload = "Hello",
1102 };
1103
1104 var output: [32]u8 = undefined;
1105 const data = try frame.serializeInto(&output, 32);
1106
1107 try std.testing.expectEqual(@as(usize, 7), data.len);
1108 try std.testing.expectEqual(@as(u8, 0x81), data[0]);
1109 try std.testing.expectEqual(@as(u8, 0x05), data[1]);
1110 try std.testing.expectEqualStrings("Hello", data[2..]);
1111 }
1112
1113 test "Frame.serialize: small frame masked" {
1114 const mask = [4]u8{ 0x12, 0x34, 0x56, 0x78 };
1115 const frame = Frame{
1116 .fin = true,
1117 .opcode = .text,
1118 .mask = mask,
1119 .payload = "Hi",
1120 };
1121
1122 var output: [32]u8 = undefined;
1123 const data = try frame.serializeInto(&output, 32);
1124
1125 try std.testing.expectEqual(@as(usize, 8), data.len);
1126 try std.testing.expectEqual(@as(u8, 0x81), data[0]);
1127 try std.testing.expectEqual(@as(u8, 0x82), data[1]);
1128 try std.testing.expectEqualSlices(u8, &mask, data[2..6]);
1129 try std.testing.expectEqual(@as(u8, 0x48 ^ 0x12), data[6]);
1130 try std.testing.expectEqual(@as(u8, 0x69 ^ 0x34), data[7]);
1131 }
1132
1133 test "Frame.serialize: 16-bit extended length" {
1134 var payload: [200]u8 = undefined;
1135 @memset(&payload, 'A');
1136
1137 const frame = Frame{
1138 .fin = true,
1139 .opcode = .binary,
1140 .mask = null,
1141 .payload = &payload,
1142 };
1143
1144 var output: [256]u8 = undefined;
1145 const data = try frame.serializeInto(&output, 256);
1146
1147 try std.testing.expectEqual(@as(usize, 204), data.len);
1148 try std.testing.expectEqual(@as(u8, 0x82), data[0]);
1149 try std.testing.expectEqual(@as(u8, 126), data[1]);
1150 try std.testing.expectEqual(@as(u8, 0x00), data[2]);
1151 try std.testing.expectEqual(@as(u8, 200), data[3]);
1152 }
1153
1154 test "Frame.serialize: continuation frame" {
1155 const frame = Frame{
1156 .fin = false,
1157 .opcode = .continuation,
1158 .mask = null,
1159 .payload = "part",
1160 };
1161
1162 var output: [32]u8 = undefined;
1163 const data = try frame.serializeInto(&output, 32);
1164
1165 try std.testing.expectEqual(@as(u8, 0x00), data[0]);
1166 }
1167
1168 test "Frame.serialize: ping frame" {
1169 const frame = Frame{
1170 .fin = true,
1171 .opcode = .ping,
1172 .mask = null,
1173 .payload = "",
1174 };
1175
1176 var output: [32]u8 = undefined;
1177 const data = try frame.serializeInto(&output, 32);
1178
1179 try std.testing.expectEqual(@as(u8, 0x89), data[0]);
1180 try std.testing.expectEqual(@as(u8, 0x00), data[1]);
1181 }
1182
1183 test "Frame.parse: simple text frame unmasked" {
1184 var data = [_]u8{ 0x81, 0x05, 'H', 'e', 'l', 'l', 'o' };
1185
1186 const result = try Frame.parse(&data, data.len);
1187
1188 try std.testing.expect(result.frame.fin);
1189 try std.testing.expectEqual(Opcode.text, result.frame.opcode);
1190 try std.testing.expectEqual(@as(?[4]u8, null), result.frame.mask);
1191 try std.testing.expectEqualStrings("Hello", result.frame.payload);
1192 try std.testing.expectEqual(@as(usize, 7), result.consumed);
1193 }
1194
1195 test "Frame.parse: masked frame" {
1196 const mask = [4]u8{ 0x12, 0x34, 0x56, 0x78 };
1197 var data = [_]u8{ 0x81, 0x82, 0x12, 0x34, 0x56, 0x78, 0x48 ^ 0x12, 0x69 ^ 0x34 };
1198
1199 const result = try Frame.parse(&data, data.len);
1200
1201 try std.testing.expect(result.frame.fin);
1202 try std.testing.expectEqual(Opcode.text, result.frame.opcode);
1203 try std.testing.expectEqual(mask, result.frame.mask.?);
1204 try std.testing.expectEqualStrings("Hi", result.frame.payload);
1205 }
1206
1207 test "Frame.parse: 16-bit extended length" {
1208 var data: [204]u8 = undefined;
1209 data[0] = 0x82;
1210 data[1] = 126;
1211 data[2] = 0x00;
1212 data[3] = 200;
1213 @memset(data[4..], 'A');
1214
1215 const result = try Frame.parse(&data, data.len);
1216
1217 try std.testing.expectEqual(@as(usize, 200), result.frame.payload.len);
1218 try std.testing.expectEqual(@as(usize, 204), result.consumed);
1219 }
1220
1221 test "Frame.parse: incomplete frame header" {
1222 var data = [_]u8{0x81};
1223 const result = Frame.parse(&data, std.math.maxInt(usize));
1224 try std.testing.expectError(FrameError.IncompleteFrame, result);
1225 }
1226
1227 test "Frame.parse: incomplete frame payload" {
1228 var data = [_]u8{ 0x81, 0x0A, 'H', 'e', 'l' };
1229 const result = Frame.parse(&data, std.math.maxInt(usize));
1230 try std.testing.expectError(FrameError.IncompleteFrame, result);
1231 }
1232
1233 test "Frame.serialize/parse round-trip" {
1234 const original = Frame{
1235 .fin = true,
1236 .opcode = .text,
1237 .mask = null,
1238 .payload = "Round trip test!",
1239 };
1240
1241 var output: [64]u8 = undefined;
1242 const serialized = try original.serializeInto(&output, 64);
1243 const result = try Frame.parse(serialized, serialized.len);
1244
1245 try std.testing.expectEqual(original.fin, result.frame.fin);
1246 try std.testing.expectEqual(original.opcode, result.frame.opcode);
1247 try std.testing.expectEqualStrings(original.payload, result.frame.payload);
1248 }
1249
1250 test "Frame.serialize/parse round-trip masked" {
1251 comptime {
1252 @stardustClaim(
1253 @import("alloc_phase").capacity.witness(Storage, "http_websocket_semantics"),
1254 null,
1255 null,
1256 null,
1257 null,
1258 null,
1259 null,
1260 );
1261 }
1262
1263 const mask = [4]u8{ 0xAB, 0xCD, 0xEF, 0x12 };
1264 const original = Frame{
1265 .fin = true,
1266 .opcode = .binary,
1267 .mask = mask,
1268 .payload = "Masked data",
1269 };
1270
1271 var output: [64]u8 = undefined;
1272 const serialized = try original.serializeInto(&output, 64);
1273 const result = try Frame.parse(serialized, serialized.len);
1274
1275 try std.testing.expectEqualStrings(original.payload, result.frame.payload);
1276 }
1277
1278 test "Frame.serializeHeaderInto matches serialized prefix" {
1279 const testing = std.testing;
1280 const frame = Frame{
1281 .fin = false,
1282 .opcode = .binary,
1283 .mask = null,
1284 .payload = "payload",
1285 };
1286 var header_storage: [maximum_frame_header_bytes]u8 =
1287 undefined;
1288 var frame_storage: [maximum_frame_header_bytes + "payload".len]u8 =
1289 undefined;
1290 const header = try frame.serializeHeaderInto(
1291 &header_storage,
1292 "payload".len,
1293 );
1294 const serialized = try frame.serializeInto(
1295 &frame_storage,
1296 "payload".len,
1297 );
1298 try testing.expectEqualSlices(
1299 u8,
1300 serialized[0..header.len],
1301 header,
1302 );
1303 try testing.expectEqualStrings(
1304 "payload",
1305 serialized[header.len..],
1306 );
1307 }
1308
1309 test "Frame.parse: close frame with code and reason" {
1310 var data = [_]u8{ 0x88, 0x05, 0x03, 0xE8, 'b', 'y', 'e' };
1311
1312 const result = try Frame.parse(&data, data.len);
1313
1314 try std.testing.expectEqual(Opcode.close, result.frame.opcode);
1315 const code = std.mem.readInt(u16, result.frame.payload[0..2], .big);
1316 try std.testing.expectEqual(@as(u16, 1000), code);
1317 try std.testing.expectEqualStrings("bye", result.frame.payload[2..]);
1318 }
1319
1320 test "Frame.parse: pong frame" {
1321 var data = [_]u8{ 0x8A, 0x04, 'p', 'o', 'n', 'g' };
1322
1323 const result = try Frame.parse(&data, data.len);
1324
1325 try std.testing.expectEqual(Opcode.pong, result.frame.opcode);
1326 try std.testing.expectEqualStrings("pong", result.frame.payload);
1327 }
1328
1329 test "Opcode.isControl" {
1330 try std.testing.expect(!Opcode.continuation.isControl());
1331 try std.testing.expect(!Opcode.text.isControl());
1332 try std.testing.expect(!Opcode.binary.isControl());
1333 try std.testing.expect(Opcode.close.isControl());
1334 try std.testing.expect(Opcode.ping.isControl());
1335 try std.testing.expect(Opcode.pong.isControl());
1336 }
1337
1338 test "WebSocket.handshake computes correct accept key" {
1339 const key = "dGhlIHNhbXBsZSBub25jZQ==";
1340 const accept = try WebSocket.handshake(key);
1341 try std.testing.expectEqualStrings("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", &accept);
1342 }
1343
1344 test "WebSocket.handshake rejects invalid client key" {
1345 try std.testing.expectError(HandshakeError.InvalidKey, WebSocket.handshake("AAAA"));
1346 }
1347
1348 const TestParsedRequest = struct {
1349 storage: http_message.RequestStorage,
1350 result: http_message.RequestParseResult,
1351
1352 fn init(data: []const u8) !TestParsedRequest {
1353 var storage = try http_message.RequestStorage.init(std.testing.allocator, .{
1354 .request_count = 1,
1355 .header_count_per_request = http_message.default_request_header_count,
1356 .header_line_bytes = http_message.default_request_header_line_bytes,
1357 .body_bytes_per_request = http_message.default_request_body_bytes,
1358 });
1359 errdefer storage.deinit(std.testing.allocator);
1360 storage.activate();
1361 return .{
1362 .result = try Request.parse(try storage.request(0), data),
1363 .storage = storage,
1364 };
1365 }
1366
1367 fn deinit(self: *TestParsedRequest) void {
1368 self.storage.deinit(std.testing.allocator);
1369 }
1370 };
1371
1372 test "WebSocket.validateClientHandshake accepts valid request" {
1373 const data =
1374 "GET /ws HTTP/1.1\r\n" ++
1375 "Host: example.test\r\n" ++
1376 "Upgrade: websocket\r\n" ++
1377 "Connection: keep-alive, Upgrade\r\n" ++
1378 "Sec-WebSocket-Version: 13\r\n" ++
1379 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" ++
1380 "\r\n";
1381
1382 var parsed = try TestParsedRequest.init(data);
1383 defer parsed.deinit();
1384 var req = parsed.result.request;
1385
1386 const client_handshake = try WebSocket.validateClientHandshake(&req);
1387 const accept = client_handshake.accept();
1388 try std.testing.expectEqualStrings("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", &accept);
1389 }
1390
1391 test "WebSocket.validateClientHandshake rejects unsupported version" {
1392 const data =
1393 "GET /ws HTTP/1.1\r\n" ++
1394 "Upgrade: websocket\r\n" ++
1395 "Connection: upgrade\r\n" ++
1396 "Sec-WebSocket-Version: 12\r\n" ++
1397 "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" ++
1398 "\r\n";
1399
1400 var parsed = try TestParsedRequest.init(data);
1401 defer parsed.deinit();
1402 var req = parsed.result.request;
1403
1404 try std.testing.expectError(HandshakeError.UnsupportedVersion, WebSocket.validateClientHandshake(&req));
1405 }
1406
1407 test "WebSocket.validateClientHandshake rejects invalid key" {
1408 const data =
1409 "GET /ws HTTP/1.1\r\n" ++
1410 "Upgrade: websocket\r\n" ++
1411 "Connection: upgrade\r\n" ++
1412 "Sec-WebSocket-Version: 13\r\n" ++
1413 "Sec-WebSocket-Key: AAAA\r\n" ++
1414 "\r\n";
1415
1416 var parsed = try TestParsedRequest.init(data);
1417 defer parsed.deinit();
1418 var req = parsed.result.request;
1419
1420 try std.testing.expectError(HandshakeError.InvalidKey, WebSocket.validateClientHandshake(&req));
1421 }
1422
1423 test "Frame.parseServer: rejects unmasked frames (RFC 6455 Section 5.1)" {
1424 var data = [_]u8{ 0x81, 0x05, 'H', 'e', 'l', 'l', 'o' };
1425 const result = Frame.parseServer(&data, data.len);
1426 try std.testing.expectError(FrameError.MaskRequired, result);
1427 }
1428
1429 test "Frame.parseServer: accepts masked frames" {
1430 var data = [_]u8{ 0x81, 0x82, 0x12, 0x34, 0x56, 0x78, 0x48 ^ 0x12, 0x69 ^ 0x34 };
1431 const result = try Frame.parseServer(&data, data.len);
1432
1433 try std.testing.expectEqualStrings("Hi", result.frame.payload);
1434 }
1435
1436 fn sendTestBytes(socket: sys.Socket, bytes: []const u8) !void {
1437 var sent: usize = 0;
1438 while (sent < bytes.len) {
1439 const written = try sys.send(socket, bytes[sent..], 0);
1440 if (written == 0) return error.ConnectionClosed;
1441 sent += written;
1442 }
1443 }
1444
1445 fn readTestBytes(socket: sys.Socket, bytes: []u8) !void {
1446 var received: usize = 0;
1447 while (received < bytes.len) {
1448 const count = try sys.recv(socket, bytes[received..], 0);
1449 if (count == 0) return error.ConnectionClosed;
1450 received += count;
1451 }
1452 }
1453
1454 const TestConnection = struct {
1455 storage: connection_mod.InputStorage,
1456 connection: Connection,
1457
1458 fn init(socket: sys.Socket, id: usize, input_bytes: usize) !TestConnection {
1459 var storage = try connection_mod.InputStorage.init(
1460 std.testing.allocator,
1461 .{ .connection_count = 1, .bytes_per_connection = input_bytes },
1462 );
1463 errdefer storage.deinit(std.testing.allocator);
1464 storage.activate();
1465 return .{
1466 .storage = storage,
1467 .connection = Connection.init(id, socket, try storage.connection(0), .{
1468 .headers = &.{},
1469 .body = &.{},
1470 .header_line_bytes = 0,
1471 }, .{
1472 .headers = &.{},
1473 .head = &.{},
1474 }, .system(), try time.bootNow()),
1475 };
1476 }
1477
1478 fn deinit(self: *TestConnection) void {
1479 self.connection.deinit();
1480 self.storage.deinit(std.testing.allocator);
1481 }
1482 };
1483
1484 fn checkFragmentedMessageAtLimit() !void {
1485 var first_bytes: [32]u8 = undefined;
1486 const first = try rawFrameBytes(
1487 &first_bytes,
1488 false,
1489 @backingInt(Opcode.text),
1490 .{ 1, 2, 3, 4 },
1491 "abc",
1492 .shortest,
1493 );
1494 var final_bytes: [32]u8 = undefined;
1495 const final = try rawFrameBytes(
1496 &final_bytes,
1497 true,
1498 @backingInt(Opcode.continuation),
1499 .{ 5, 6, 7, 8 },
1500 "def",
1501 .shortest,
1502 );
1503 const sockets = try sys.socketPairUnixStream();
1504 defer sys.close(sockets[1]);
1505 var owned_connection = try TestConnection.init(sockets[0], 1, 18);
1506 defer owned_connection.deinit();
1507 const connection = &owned_connection.connection;
1508 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
1509 var websocket = WebSocket.init(
1510 phase_allocator.initializationAllocator(),
1511 .{ .frame_payload_bytes = 4, .message_payload_bytes = 6 },
1512 connection,
1513 ) catch |err| {
1514 phase_allocator.abortInitialization();
1515 phase_allocator.deinit();
1516 return err;
1517 };
1518 errdefer {
1519 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
1520 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
1521 if (websocket.storage.phase != .teardown) {
1522 websocket.deinit(phase_allocator.teardownAllocator());
1523 }
1524 phase_allocator.deinit();
1525 }
1526
1527 try connection.retainInput(first);
1528 try connection.retainInput(final);
1529 const pointer = websocket.storage.bytes.ptr;
1530 const capacity = websocket.capacity();
1531 phase_allocator.seal();
1532 websocket.activate();
1533
1534 const received = (try websocket.receive()).?;
1535 try std.testing.expectEqual(Opcode.text, received.opcode);
1536 try std.testing.expectEqualStrings("abcdef", received.payload);
1537 try std.testing.expect(received.payload.ptr == websocket.messageRegion().ptr);
1538 try websocket.sendText("ok");
1539 try std.testing.expectEqualStrings("abcdef", received.payload);
1540
1541 var outbound: [4]u8 = undefined;
1542 try readTestBytes(sockets[1], &outbound);
1543 const parsed = try Frame.parse(&outbound, 2);
1544 try std.testing.expectEqual(Opcode.text, parsed.frame.opcode);
1545 try std.testing.expectEqualStrings("ok", parsed.frame.payload);
1546 try std.testing.expect(websocket.storage.bytes.ptr == pointer);
1547 try std.testing.expectEqual(capacity, websocket.capacity());
1548 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
1549
1550 phase_allocator.beginTeardown();
1551 websocket.deinit(phase_allocator.teardownAllocator());
1552 phase_allocator.deinit();
1553 }
1554
1555 fn checkFragmentedMessageOverLimit() !void {
1556 var first_bytes: [32]u8 = undefined;
1557 const first = try rawFrameBytes(
1558 &first_bytes,
1559 false,
1560 @backingInt(Opcode.binary),
1561 .{ 1, 2, 3, 4 },
1562 "abc",
1563 .shortest,
1564 );
1565 var final_bytes: [32]u8 = undefined;
1566 const final = try rawFrameBytes(
1567 &final_bytes,
1568 true,
1569 @backingInt(Opcode.continuation),
1570 .{ 5, 6, 7, 8 },
1571 "def",
1572 .shortest,
1573 );
1574
1575 const sockets = try sys.socketPairUnixStream();
1576 defer sys.close(sockets[1]);
1577 var owned_connection = try TestConnection.init(sockets[0], 2, 18);
1578 defer owned_connection.deinit();
1579 const connection = &owned_connection.connection;
1580
1581 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
1582 var websocket = WebSocket.init(
1583 phase_allocator.initializationAllocator(),
1584 .{ .frame_payload_bytes = 3, .message_payload_bytes = 5 },
1585 connection,
1586 ) catch |err| {
1587 phase_allocator.abortInitialization();
1588 phase_allocator.deinit();
1589 return err;
1590 };
1591 errdefer {
1592 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
1593 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
1594 if (websocket.storage.phase != .teardown) {
1595 websocket.deinit(phase_allocator.teardownAllocator());
1596 }
1597 phase_allocator.deinit();
1598 }
1599
1600 @memset(websocket.messageRegion(), 0xa5);
1601 try sendTestBytes(sockets[1], first);
1602 try sendTestBytes(sockets[1], final);
1603 const pointer = websocket.storage.bytes.ptr;
1604 const capacity = websocket.capacity();
1605 phase_allocator.seal();
1606 websocket.activate();
1607
1608 try std.testing.expect((try websocket.receive()) == null);
1609 try std.testing.expectEqual(WebSocketState.closed, websocket.state);
1610 try std.testing.expectEqualStrings("abc", websocket.messageRegion()[0..3]);
1611 for (websocket.messageRegion()[3..]) |byte| {
1612 try std.testing.expectEqual(@as(u8, 0xa5), byte);
1613 }
1614 try std.testing.expect(websocket.storage.bytes.ptr == pointer);
1615 try std.testing.expectEqual(capacity, websocket.capacity());
1616 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
1617
1618 phase_allocator.beginTeardown();
1619 websocket.deinit(phase_allocator.teardownAllocator());
1620 phase_allocator.deinit();
1621 }
1622
1623 test "WebSocket fragmented messages preserve bounded borrowed storage" {
1624 comptime {
1625 @stardustClaim(
1626 @import("alloc_phase").capacity.witness(Storage, "http_websocket_fragmented"),
1627 null,
1628 null,
1629 null,
1630 null,
1631 null,
1632 null,
1633 );
1634 }
1635
1636 try checkFragmentedMessageAtLimit();
1637 try checkFragmentedMessageOverLimit();
1638 }
1639
1640 test "WebSocket.receive closes invalid client frame as protocol error" {
1641 const sockets = try sys.socketPairUnixStream();
1642 defer sys.close(sockets[1]);
1643
1644 var owned_connection = try TestConnection.init(sockets[0], 1, 32);
1645 defer owned_connection.deinit();
1646 const conn = &owned_connection.connection;
1647
1648 var ws = try WebSocket.init(std.testing.allocator, .{
1649 .frame_payload_bytes = 128,
1650 .message_payload_bytes = 128,
1651 }, conn);
1652 defer ws.deinit(std.testing.allocator);
1653 ws.activate();
1654
1655 var raw_bytes: [32]u8 = undefined;
1656 const raw = try rawFrameBytes(
1657 &raw_bytes,
1658 true,
1659 0x3,
1660 [4]u8{ 0x12, 0x34, 0x56, 0x78 },
1661 "",
1662 .shortest,
1663 );
1664
1665 _ = try sys.send(sockets[1], raw, 0);
1666
1667 const message = try ws.receive();
1668 try std.testing.expect(message == null);
1669 try std.testing.expectEqual(WebSocketState.closed, ws.state);
1670
1671 var close_buf: [128]u8 = undefined;
1672 const close_len = try sys.recv(sockets[1], &close_buf, 0);
1673 const close_result = try Frame.parse(close_buf[0..close_len], close_buf.len);
1674
1675 try std.testing.expectEqual(Opcode.close, close_result.frame.opcode);
1676 try std.testing.expect(close_result.frame.payload.len >= 2);
1677 try std.testing.expectEqual(
1678 WebSocketCloseCode.protocol_error,
1679 std.mem.readInt(u16, close_result.frame.payload[0..2], .big),
1680 );
1681 }
1682
1683 test "Frame.parse: rejects reserved opcode" {
1684 var data = [_]u8{ 0x83, 0x00 };
1685 try expectFrameParseError(FrameError.InvalidOpcode, &data);
1686 }
1687
1688 test "Frame.parse: rejects reserved bits without extensions" {
1689 var data = [_]u8{ 0xC1, 0x00 };
1690 try expectFrameParseError(FrameError.InvalidFrame, &data);
1691 }
1692
1693 test "Frame.parse: rejects fragmented control frame" {
1694 var data = [_]u8{ 0x09, 0x00 };
1695 try expectFrameParseError(FrameError.InvalidFrame, &data);
1696 }
1697
1698 test "WebSocket bounds aggregate fragmented message size" {
1699 try std.testing.expect(fragmentLengthAllowed(15, 1, 16));
1700 try std.testing.expect(!fragmentLengthAllowed(16, 1, 16));
1701 try std.testing.expect(!fragmentLengthAllowed(17, 0, 16));
1702 }
1703
1704 test "Frame.parse: rejects oversized control frame" {
1705 var data = [_]u8{ 0x89, 126, 0, 126 };
1706 try expectFrameParseError(FrameError.InvalidFrame, &data);
1707 }
1708
1709 test "Frame.parse: rejects noncanonical extended length" {
1710 var data = [_]u8{ 0x82, 126, 0, 5, 'H', 'e', 'l', 'l', 'o' };
1711 try expectFrameParseError(FrameError.InvalidFrame, &data);
1712 }