lib/http/src/client/runtime.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const sys_root = @import("sys");
4 const sys = sys_root.net;
5 const tls = sys_root.tls;
6 const Allocator = std.mem.Allocator;
7 const Io = std.Io;
8 const client_operation = @import("operation/root.zig");
9 const client_response = @import("response.zig");
10 const client_stream = @import("stream/root.zig");
11 const client_websocket = @import("websocket/root.zig");
12 const http = @import("../root.zig");
13
14 const transport = @import("root.zig").transport;
15 const SocketReader = transport.SocketReader;
16 const SocketWriter = transport.SocketWriter;
17 const TlsConn = transport.TlsConn;
18
19 const Header = client_response.Header;
20 const ClientOperation = client_operation.ClientOperation;
21 const ClientResponse = client_response.ClientResponse;
22 const OperationScratch = client_operation.Scratch;
23 const PreparedOperation = client_operation.Prepared;
24 const ResponseScratch = client_response.Scratch;
25 const StreamScratch = client_stream.Scratch;
26 const WebSocketScratch = client_websocket.Scratch;
27 const WebSocketMessage = http.WebSocketMessage;
28
29 pub const StreamHandlerVTable = client_stream.HandlerVTable;
30 pub const StreamHandler = client_stream.Handler;
31
32 pub const ClientError = client_operation.Error || client_response.Error || error{
33 ConnectionFailed,
34 TunnelUnsupported,
35 TlsFailed,
36 WriteFailed,
37 OutOfMemory,
38 };
39
40 pub const default_timeout_ms: u32 = 300_000;
41
42 const PersistentConnection = union(enum) {
43 plain: *PlainConn,
44 tls: *TlsConn,
45 };
46
47 pub const Client = struct {
48 allocator: Allocator,
49 ca_bundle: tls.Bundle = .empty,
50 ca_bundle_lock: tls.BundleLock = .init,
51 certs_loaded: bool = false,
52 timeout_ms: u32 = default_timeout_ms,
53 keep_alive: bool = false,
54 tcp_no_delay: bool = false,
55 persistent: ?PersistentConnection = null,
56 persistent_host: [client_operation.default_host_bytes]u8 =
57 undefined,
58 persistent_host_len: u16 = 0,
59 persistent_port: u16 = 0,
60
61 /// Builds a client that holds `allocator` to back its certificate authority
62 /// bundle and retain an open connection for subsequent requests. Building a
63 /// client touches the network in no way. The certificate authority bundle
64 /// is loaded from the local operating system store on the first TLS request
65 /// that requires it rather than during initialization. Calling `deinit`
66 /// closes any retained connection and frees the bundle.
67 pub fn init(allocator: Allocator) Client {
68 return .{ .allocator = allocator };
69 }
70
71 pub fn deinit(self: *Client) void {
72 self.dropPersistent(true);
73 if (self.certs_loaded) self.ca_bundle.deinit(self.allocator);
74 }
75
76 fn ensureCerts(self: *Client) !void {
77 if (!self.certs_loaded) {
78 tls.loadSystemBundle(&self.ca_bundle, self.allocator) catch return error.TlsFailed;
79 self.certs_loaded = true;
80 }
81 }
82
83 pub fn post(
84 self: *Client,
85 operation_scratch: OperationScratch,
86 response_scratch: ResponseScratch,
87 url: []const u8,
88 headers: []const Header,
89 body: []const u8,
90 ) ClientError!ClientResponse {
91 return self.request(operation_scratch, response_scratch, "POST", url, headers, body);
92 }
93
94 pub fn postStream(
95 self: *Client,
96 operation_scratch: OperationScratch,
97 stream_scratch: StreamScratch,
98 url: []const u8,
99 headers: []const Header,
100 body: []const u8,
101 handler: StreamHandler,
102 ) anyerror!u16 {
103 return self.requestStream(
104 operation_scratch,
105 stream_scratch,
106 "POST",
107 url,
108 headers,
109 body,
110 handler,
111 );
112 }
113
114 pub fn get(
115 self: *Client,
116 operation_scratch: OperationScratch,
117 response_scratch: ResponseScratch,
118 url: []const u8,
119 headers: []const Header,
120 ) ClientError!ClientResponse {
121 return self.request(operation_scratch, response_scratch, "GET", url, headers, "");
122 }
123
124 pub fn getStream(
125 self: *Client,
126 operation_scratch: OperationScratch,
127 stream_scratch: StreamScratch,
128 url: []const u8,
129 headers: []const Header,
130 handler: StreamHandler,
131 ) anyerror!u16 {
132 return self.requestStream(
133 operation_scratch,
134 stream_scratch,
135 "GET",
136 url,
137 headers,
138 "",
139 handler,
140 );
141 }
142
143 /// Sends one request and returns the response read from the server.
144 /// Returned headers and body slices point directly into `response_scratch`.
145 /// This backing storage must remain alive and unchanged while the response
146 /// is in use, with subsequent parses into that storage serving as one point
147 /// of invalidation.
148 ///
149 /// Parsing the URL into target fields and checking the read and write
150 /// buffers for a plain request occur before opening any socket, so failures
151 /// during these preparation checks leave the network untouched.
152 ///
153 /// A request over TLS loads the local operating system certificate
154 /// authority bundle on the first call requiring it. When keep-alive is
155 /// enabled, the request reuses an existing connection already open to that
156 /// host and port.
157 pub fn request(
158 self: *Client,
159 operation_scratch: OperationScratch,
160 response_scratch: ResponseScratch,
161 method: []const u8,
162 url: []const u8,
163 headers: []const Header,
164 body: []const u8,
165 ) ClientError!ClientResponse {
166 try rejectTunnelMethod(method);
167 const prepared = try ClientOperation.prepare(operation_scratch, url);
168 if (!prepared.is_tls) try ClientOperation.validatePlain(operation_scratch);
169 const alloc = self.allocator;
170 if (prepared.is_tls) try self.ensureCerts();
171
172 if (self.keep_alive) {
173 return self.reusableRequest(
174 response_scratch,
175 prepared,
176 method,
177 headers,
178 body,
179 );
180 }
181 self.dropPersistent(true);
182 const stream = connectToHost(
183 prepared.host,
184 prepared.port,
185 self.timeout_ms,
186 self.tcp_no_delay,
187 ) catch return error.ConnectionFailed;
188 if (prepared.is_tls) {
189 return self.doTlsRequest(
190 alloc,
191 response_scratch,
192 stream,
193 prepared.host,
194 prepared.target,
195 method,
196 headers,
197 body,
198 );
199 } else {
200 return doPlainRequest(
201 operation_scratch,
202 response_scratch,
203 stream,
204 prepared.host,
205 prepared.target,
206 method,
207 headers,
208 body,
209 );
210 }
211 }
212
213 pub fn openWebSocket(
214 self: *Client,
215 operation_scratch: OperationScratch,
216 websocket_scratch: WebSocketScratch,
217 url: []const u8,
218 ) anyerror!ClientWebSocket {
219 const prepared = try ClientOperation.prepare(
220 operation_scratch,
221 url,
222 );
223 if (!prepared.is_tls) {
224 try ClientOperation.validatePlain(
225 operation_scratch,
226 );
227 }
228 if (prepared.is_tls) try self.ensureCerts();
229 const socket = connectToHost(
230 prepared.host,
231 prepared.port,
232 self.timeout_ms,
233 self.tcp_no_delay,
234 ) catch return error.ConnectionFailed;
235 const connection: PersistentConnection =
236 if (prepared.is_tls)
237 .{ .tls = try self.createTlsPersistent(
238 socket,
239 prepared.host,
240 ) }
241 else
242 .{ .plain = try PlainConn.create(
243 self.allocator,
244 socket,
245 ) };
246 var owned = true;
247 errdefer if (owned) {
248 destroyConnection(
249 self.allocator,
250 connection,
251 false,
252 );
253 };
254 try clientWebSocketHandshake(
255 connection,
256 prepared,
257 );
258 const result = try ClientWebSocket.init(
259 self.allocator,
260 connection,
261 websocket_scratch,
262 );
263 owned = false;
264 return result;
265 }
266
267 pub fn requestStream(
268 self: *Client,
269 operation_scratch: OperationScratch,
270 stream_scratch: StreamScratch,
271 method: []const u8,
272 url: []const u8,
273 headers: []const Header,
274 body: []const u8,
275 handler: StreamHandler,
276 ) anyerror!u16 {
277 try rejectTunnelMethod(method);
278 const prepared = try ClientOperation.prepare(operation_scratch, url);
279 if (!prepared.is_tls) try ClientOperation.validatePlain(operation_scratch);
280 const alloc = self.allocator;
281 if (prepared.is_tls) try self.ensureCerts();
282
283 const stream = connectToHost(
284 prepared.host,
285 prepared.port,
286 self.timeout_ms,
287 self.tcp_no_delay,
288 ) catch return error.ConnectionFailed;
289
290 if (prepared.is_tls) {
291 return self.doTlsRequestStream(
292 alloc,
293 stream_scratch,
294 stream,
295 prepared.host,
296 prepared.target,
297 method,
298 headers,
299 body,
300 handler,
301 );
302 } else {
303 return doPlainRequestStream(
304 operation_scratch,
305 stream_scratch,
306 stream,
307 prepared.host,
308 prepared.target,
309 method,
310 headers,
311 body,
312 handler,
313 );
314 }
315 }
316
317 fn reusableRequest(
318 self: *Client,
319 response_scratch: ResponseScratch,
320 prepared: PreparedOperation,
321 method: []const u8,
322 headers: []const Header,
323 body: []const u8,
324 ) ClientError!ClientResponse {
325 try self.ensurePersistent(prepared);
326 const response = self.performReusableRequest(
327 response_scratch,
328 prepared.host,
329 prepared.target,
330 method,
331 headers,
332 body,
333 ) catch |err| {
334 self.dropPersistent(false);
335 return err;
336 };
337 if (!response.connection_reusable) {
338 self.dropPersistent(true);
339 }
340 return response;
341 }
342
343 fn ensurePersistent(
344 self: *Client,
345 prepared: PreparedOperation,
346 ) ClientError!void {
347 if (self.persistentMatches(prepared)) return;
348 self.dropPersistent(true);
349 if (prepared.host.len > self.persistent_host.len) {
350 return error.ClientHostCapacityExceeded;
351 }
352 const socket = connectToHost(
353 prepared.host,
354 prepared.port,
355 self.timeout_ms,
356 self.tcp_no_delay,
357 ) catch return error.ConnectionFailed;
358 const connection: PersistentConnection =
359 if (prepared.is_tls)
360 .{ .tls = try self.createTlsPersistent(
361 socket,
362 prepared.host,
363 ) }
364 else
365 .{ .plain = try PlainConn.create(
366 self.allocator,
367 socket,
368 ) };
369 self.persistent = connection;
370 @memcpy(
371 self.persistent_host[0..prepared.host.len],
372 prepared.host,
373 );
374 self.persistent_host_len =
375 @intCast(prepared.host.len);
376 self.persistent_port = prepared.port;
377 }
378
379 fn createTlsPersistent(
380 self: *Client,
381 socket: sys.Socket,
382 host: []const u8,
383 ) ClientError!*TlsConn {
384 const conn = self.allocator.create(TlsConn) catch {
385 sys.close(socket);
386 return error.OutOfMemory;
387 };
388 errdefer self.allocator.destroy(conn);
389 conn.initAt(socket);
390 errdefer conn.deinit(false);
391 try self.initializeTls(conn, host);
392 return conn;
393 }
394
395 fn initializeTls(
396 self: *Client,
397 conn: *TlsConn,
398 host: []const u8,
399 ) ClientError!void {
400 var entropy: [tls.Client.Options.entropy_len]u8 =
401 undefined;
402 tls.fillEntropy(&entropy) catch
403 return error.TlsFailed;
404 const now = tls.realtimeNow();
405 conn.tls_client = tls.Client.init(
406 &conn.socket_reader.reader,
407 &conn.socket_writer.writer,
408 .{
409 .host = .{ .explicit = host },
410 .ca = tls.bundleAuthority(
411 self.allocator,
412 &self.ca_bundle_lock,
413 &self.ca_bundle,
414 ),
415 .read_buffer = &conn.tls_read_buf,
416 .write_buffer = &conn.tls_write_buf,
417 .entropy = &entropy,
418 .realtime_now = now,
419 .allow_truncation_attacks = true,
420 },
421 ) catch return error.TlsFailed;
422 }
423
424 fn performReusableRequest(
425 self: *Client,
426 response_scratch: ResponseScratch,
427 host: []const u8,
428 target: []const u8,
429 method: []const u8,
430 headers: []const Header,
431 body: []const u8,
432 ) ClientError!ClientResponse {
433 const persistent = self.persistent orelse
434 return error.ConnectionFailed;
435 return switch (persistent) {
436 .plain => |conn| blk: {
437 ClientOperation.writeReusableRequest(
438 &conn.socket_writer.writer,
439 method,
440 host,
441 target,
442 headers,
443 body,
444 ) catch return error.WriteFailed;
445 conn.socket_writer.writer.flush() catch
446 return error.WriteFailed;
447 break :blk try client_response.read(
448 response_scratch,
449 &conn.socket_reader.reader,
450 method,
451 );
452 },
453 .tls => |conn| blk: {
454 ClientOperation.writeReusableRequest(
455 &conn.tls_client.?.writer,
456 method,
457 host,
458 target,
459 headers,
460 body,
461 ) catch return error.WriteFailed;
462 conn.tls_client.?.writer.flush() catch
463 return error.WriteFailed;
464 conn.socket_writer.writer.flush() catch
465 return error.WriteFailed;
466 break :blk try client_response.read(
467 response_scratch,
468 &conn.tls_client.?.reader,
469 method,
470 );
471 },
472 };
473 }
474
475 fn persistentMatches(
476 self: *const Client,
477 prepared: PreparedOperation,
478 ) bool {
479 const persistent = self.persistent orelse
480 return false;
481 const is_tls = switch (persistent) {
482 .plain => false,
483 .tls => true,
484 };
485 return is_tls == prepared.is_tls and
486 self.persistent_port == prepared.port and
487 std.mem.eql(
488 u8,
489 self.persistent_host[0..self.persistent_host_len],
490 prepared.host,
491 );
492 }
493
494 fn dropPersistent(
495 self: *Client,
496 notify: bool,
497 ) void {
498 if (self.persistent) |persistent| {
499 switch (persistent) {
500 .plain => |conn| conn.destroy(
501 self.allocator,
502 ),
503 .tls => |conn| {
504 conn.deinit(notify);
505 self.allocator.destroy(conn);
506 },
507 }
508 self.persistent = null;
509 }
510 self.persistent_host_len = 0;
511 self.persistent_port = 0;
512 }
513
514 fn rejectTunnelMethod(method: []const u8) error{TunnelUnsupported}!void {
515 if (isTunnelMethod(method)) return error.TunnelUnsupported;
516 }
517
518 fn isTunnelMethod(method: []const u8) bool {
519 return std.mem.eql(u8, method, "CONNECT");
520 }
521
522 fn doTlsRequest(
523 self: *Client,
524 alloc: Allocator,
525 response_scratch: ResponseScratch,
526 socket: sys.Socket,
527 host: []const u8,
528 path: []const u8,
529 method: []const u8,
530 headers: []const Header,
531 body: []const u8,
532 ) ClientError!ClientResponse {
533 std.debug.assert(!isTunnelMethod(method));
534 const conn = alloc.create(TlsConn) catch return error.OutOfMemory;
535 var send_close_notify = false;
536 defer {
537 conn.deinit(send_close_notify);
538 alloc.destroy(conn);
539 }
540
541 conn.initAt(socket);
542
543 try self.initializeTls(conn, host);
544
545 ClientOperation.writeRequest(
546 &conn.tls_client.?.writer,
547 method,
548 host,
549 path,
550 headers,
551 body,
552 ) catch return error.WriteFailed;
553 conn.tls_client.?.writer.flush() catch return error.WriteFailed;
554 conn.socket_writer.writer.flush() catch return error.WriteFailed;
555
556 const parsed = try client_response.read(
557 response_scratch,
558 &conn.tls_client.?.reader,
559 method,
560 );
561 send_close_notify = true;
562 return parsed;
563 }
564
565 fn doTlsRequestStream(
566 self: *Client,
567 alloc: Allocator,
568 stream_scratch: StreamScratch,
569 socket: sys.Socket,
570 host: []const u8,
571 path: []const u8,
572 method: []const u8,
573 headers: []const Header,
574 body: []const u8,
575 handler: StreamHandler,
576 ) anyerror!u16 {
577 std.debug.assert(!isTunnelMethod(method));
578 const conn = alloc.create(TlsConn) catch return error.OutOfMemory;
579 var send_close_notify = false;
580 defer {
581 conn.deinit(send_close_notify);
582 alloc.destroy(conn);
583 }
584
585 conn.initAt(socket);
586 try self.initializeTls(conn, host);
587
588 ClientOperation.writeRequest(
589 &conn.tls_client.?.writer,
590 method,
591 host,
592 path,
593 headers,
594 body,
595 ) catch return error.WriteFailed;
596 conn.tls_client.?.writer.flush() catch return error.WriteFailed;
597 conn.socket_writer.writer.flush() catch return error.WriteFailed;
598
599 const status = try client_stream.ClientStream.readForMethod(
600 stream_scratch,
601 &conn.tls_client.?.reader,
602 method,
603 handler,
604 );
605 send_close_notify = true;
606 return status;
607 }
608
609 fn doPlainRequest(
610 operation_scratch: OperationScratch,
611 response_scratch: ResponseScratch,
612 socket: sys.Socket,
613 host: []const u8,
614 path: []const u8,
615 method: []const u8,
616 headers: []const Header,
617 body: []const u8,
618 ) ClientError!ClientResponse {
619 std.debug.assert(!isTunnelMethod(method));
620 defer sys.close(socket);
621
622 var socket_writer = SocketWriter.init(socket, operation_scratch.plain_write);
623 ClientOperation.writeRequest(
624 &socket_writer.writer,
625 method,
626 host,
627 path,
628 headers,
629 body,
630 ) catch return error.WriteFailed;
631 socket_writer.writer.flush() catch return error.WriteFailed;
632
633 var socket_reader = SocketReader.init(socket, operation_scratch.plain_read);
634 return client_response.read(response_scratch, &socket_reader.reader, method);
635 }
636
637 fn doPlainRequestStream(
638 operation_scratch: OperationScratch,
639 stream_scratch: StreamScratch,
640 socket: sys.Socket,
641 host: []const u8,
642 path: []const u8,
643 method: []const u8,
644 headers: []const Header,
645 body: []const u8,
646 handler: StreamHandler,
647 ) anyerror!u16 {
648 std.debug.assert(!isTunnelMethod(method));
649 defer sys.close(socket);
650
651 var socket_writer = SocketWriter.init(socket, operation_scratch.plain_write);
652 ClientOperation.writeRequest(
653 &socket_writer.writer,
654 method,
655 host,
656 path,
657 headers,
658 body,
659 ) catch return error.WriteFailed;
660 socket_writer.writer.flush() catch return error.WriteFailed;
661
662 var socket_reader = SocketReader.init(socket, operation_scratch.plain_read);
663 return client_stream.ClientStream.readForMethod(
664 stream_scratch,
665 &socket_reader.reader,
666 method,
667 handler,
668 );
669 }
670 };
671
672 const PlainConn = struct {
673 read_storage: [client_operation.default_plain_read_bytes]u8,
674 write_storage: [client_operation.default_plain_write_bytes]u8,
675 socket_reader: SocketReader,
676 socket_writer: SocketWriter,
677 socket: sys.Socket,
678
679 fn create(
680 allocator: Allocator,
681 socket: sys.Socket,
682 ) ClientError!*PlainConn {
683 const self = allocator.create(PlainConn) catch {
684 sys.close(socket);
685 return error.OutOfMemory;
686 };
687 self.socket = socket;
688 self.socket_reader = SocketReader.init(
689 socket,
690 &self.read_storage,
691 );
692 self.socket_writer = SocketWriter.init(
693 socket,
694 &self.write_storage,
695 );
696 return self;
697 }
698
699 fn destroy(
700 self: *PlainConn,
701 allocator: Allocator,
702 ) void {
703 sys.close(self.socket);
704 allocator.destroy(self);
705 }
706 };
707
708 const client_websocket_head_bytes_max: usize = 4096;
709 const client_websocket_header_count_max: usize = 16;
710 pub const ClientWebSocket = struct {
711 allocator: Allocator,
712 connection: PersistentConnection,
713 read_storage: []u8,
714 write_storage: []u8,
715 frame_payload_bytes: usize,
716 read_length: usize = 0,
717 pending_consumed: usize = 0,
718 open: bool = true,
719
720 fn init(
721 allocator: Allocator,
722 connection: PersistentConnection,
723 scratch: WebSocketScratch,
724 ) !ClientWebSocket {
725 const expected = try alloc_phase.capacity.add(
726 usize,
727 client_websocket.frame_header_bytes,
728 scratch.frame_payload_bytes,
729 );
730 if (scratch.read.len != expected or
731 scratch.write.len != expected)
732 {
733 return error.InvalidWebSocketScratch;
734 }
735 return .{
736 .allocator = allocator,
737 .connection = connection,
738 .read_storage = scratch.read,
739 .write_storage = scratch.write,
740 .frame_payload_bytes = scratch.frame_payload_bytes,
741 };
742 }
743
744 pub fn deinit(self: *ClientWebSocket) void {
745 destroyConnection(
746 self.allocator,
747 self.connection,
748 true,
749 );
750 std.crypto.secureZero(u8, self.read_storage);
751 std.crypto.secureZero(u8, self.write_storage);
752 self.* = undefined;
753 }
754
755 pub fn pollReadable(
756 self: *ClientWebSocket,
757 timeout_ms: i32,
758 ) !bool {
759 if (timeout_ms < 0) return error.InvalidTimeout;
760 if (self.read_length > self.pending_consumed or
761 connectionBuffered(self.connection))
762 {
763 return true;
764 }
765 return try sys.pollReadable(
766 connectionSocket(self.connection),
767 timeout_ms,
768 );
769 }
770
771 pub fn sendBinary(
772 self: *ClientWebSocket,
773 payload: []const u8,
774 ) !void {
775 try self.sendFrame(.binary, payload);
776 }
777
778 pub fn receive(
779 self: *ClientWebSocket,
780 ) !?WebSocketMessage {
781 self.releaseBorrowed();
782 while (self.open) {
783 if (self.read_length == 0) {
784 try self.readFrame();
785 }
786 const parsed = http.Frame.parse(
787 self.wire()[0..self.read_length],
788 self.frame_payload_bytes,
789 ) catch |err| return err;
790 if (parsed.frame.mask != null) {
791 return error.MaskedServerFrame;
792 }
793 if (!parsed.frame.fin or
794 parsed.frame.opcode == .continuation)
795 {
796 return error.FragmentedServerMessage;
797 }
798 switch (parsed.frame.opcode) {
799 .binary, .text => {
800 self.pending_consumed = parsed.consumed;
801 return .{
802 .opcode = parsed.frame.opcode,
803 .payload = parsed.frame.payload,
804 };
805 },
806 .ping => {
807 try self.sendFrame(
808 .pong,
809 parsed.frame.payload,
810 );
811 self.consume(parsed.consumed);
812 },
813 .pong => self.consume(parsed.consumed),
814 .close => {
815 self.sendFrame(
816 .close,
817 parsed.frame.payload,
818 ) catch {};
819 self.consume(parsed.consumed);
820 self.open = false;
821 return null;
822 },
823 .continuation => unreachable,
824 }
825 }
826 return null;
827 }
828
829 pub fn close(self: *ClientWebSocket) !void {
830 if (!self.open) return;
831 try self.sendFrame(.close, &.{});
832 self.open = false;
833 }
834
835 fn sendFrame(
836 self: *ClientWebSocket,
837 opcode: http.Opcode,
838 payload: []const u8,
839 ) !void {
840 if (!self.open and opcode != .close) {
841 return error.ConnectionClosed;
842 }
843 var mask: [4]u8 = undefined;
844 try sys_root.random.secureBytes(&mask);
845 const encoded = try (http.Frame{
846 .fin = true,
847 .opcode = opcode,
848 .mask = mask,
849 .payload = payload,
850 }).serializeInto(
851 self.output(),
852 self.frame_payload_bytes,
853 );
854 defer std.crypto.secureZero(u8, encoded);
855 const writer = connectionWriter(
856 self.connection,
857 );
858 try writer.writeAll(encoded);
859 try flushConnection(self.connection);
860 }
861
862 fn readFrame(self: *ClientWebSocket) !void {
863 const reader = connectionReader(self.connection);
864 const target = self.wire();
865 try readClientWebSocketExact(
866 reader,
867 target[0..2],
868 );
869 if ((target[1] & 0x80) != 0) {
870 return error.MaskedServerFrame;
871 }
872 const short_length = target[1] & 0x7f;
873 var header_bytes: usize = 2;
874 const payload_bytes: usize = switch (short_length) {
875 0...125 => short_length,
876 126 => length: {
877 try readClientWebSocketExact(
878 reader,
879 target[2..4],
880 );
881 header_bytes = 4;
882 break :length std.mem.readInt(
883 u16,
884 target[2..4],
885 .big,
886 );
887 },
888 127 => length: {
889 try readClientWebSocketExact(
890 reader,
891 target[2..10],
892 );
893 header_bytes = 10;
894 const wide = std.mem.readInt(
895 u64,
896 target[2..10],
897 .big,
898 );
899 if (wide > self.frame_payload_bytes or
900 wide > std.math.maxInt(usize))
901 {
902 return error.PayloadTooLarge;
903 }
904 break :length @intCast(wide);
905 },
906 else => unreachable,
907 };
908 if (payload_bytes > self.frame_payload_bytes) {
909 return error.PayloadTooLarge;
910 }
911 const total = std.math.add(
912 usize,
913 header_bytes,
914 payload_bytes,
915 ) catch return error.PayloadTooLarge;
916 if (total > target.len) {
917 return error.FrameCapacity;
918 }
919 try readClientWebSocketExact(
920 reader,
921 target[header_bytes..total],
922 );
923 self.read_length = total;
924 }
925
926 fn releaseBorrowed(self: *ClientWebSocket) void {
927 if (self.pending_consumed == 0) return;
928 self.consume(self.pending_consumed);
929 self.pending_consumed = 0;
930 }
931
932 fn consume(
933 self: *ClientWebSocket,
934 consumed: usize,
935 ) void {
936 const remaining = self.read_length - consumed;
937 const previous_length = self.read_length;
938 std.mem.copyForwards(
939 u8,
940 self.wire()[0..remaining],
941 self.wire()[consumed..self.read_length],
942 );
943 std.crypto.secureZero(
944 u8,
945 self.wire()[remaining..previous_length],
946 );
947 self.read_length = remaining;
948 }
949
950 fn wire(self: *ClientWebSocket) []u8 {
951 return self.read_storage;
952 }
953
954 fn output(self: *ClientWebSocket) []u8 {
955 return self.write_storage;
956 }
957 };
958
959 fn readClientWebSocketExact(
960 reader: *std.Io.Reader,
961 target: []u8,
962 ) !void {
963 reader.readSliceAll(target) catch |err|
964 return switch (err) {
965 error.EndOfStream => error.ConnectionClosed,
966 error.ReadFailed => error.ReadFailed,
967 };
968 }
969
970 fn clientWebSocketHandshake(
971 connection: PersistentConnection,
972 prepared: PreparedOperation,
973 ) !void {
974 var entropy: [16]u8 = undefined;
975 try sys_root.random.secureBytes(&entropy);
976 defer std.crypto.secureZero(u8, &entropy);
977 var key_buffer: [24]u8 = undefined;
978 const key = std.base64.standard.Encoder.encode(
979 &key_buffer,
980 &entropy,
981 );
982 const expected_accept = try http.WebSocket.handshake(
983 key,
984 );
985 const writer = connectionWriter(connection);
986 try writer.writeAll("GET ");
987 try writer.writeAll(prepared.target);
988 try writer.writeAll(" HTTP/1.1\r\nHost: ");
989 try writer.writeAll(prepared.host);
990 const default_port: u16 =
991 if (prepared.is_tls) 443 else 80;
992 if (prepared.port != default_port) {
993 try writer.print(":{d}", .{prepared.port});
994 }
995 try writer.writeAll(
996 "\r\nUpgrade: websocket\r\n" ++
997 "Connection: Upgrade\r\n" ++
998 "Sec-WebSocket-Key: ",
999 );
1000 try writer.writeAll(key);
1001 try writer.writeAll(
1002 "\r\nSec-WebSocket-Version: 13\r\n\r\n",
1003 );
1004 try flushConnection(connection);
1005
1006 var headers: [
1007 client_websocket_header_count_max
1008 ]Header = undefined;
1009 var head: [client_websocket_head_bytes_max]u8 =
1010 undefined;
1011 const response = try client_response.read(
1012 .{
1013 .headers = &headers,
1014 .head = &head,
1015 .body = &.{},
1016 },
1017 connectionReader(connection),
1018 "GET",
1019 );
1020 try validateClientWebSocketResponse(
1021 response,
1022 &expected_accept,
1023 );
1024 }
1025
1026 fn validateClientWebSocketResponse(
1027 response: ClientResponse,
1028 expected_accept: []const u8,
1029 ) !void {
1030 if (response.status != 101) {
1031 return error.UpgradeRejected;
1032 }
1033 const upgrade = response.header("Upgrade") orelse
1034 return error.MissingUpgradeHeader;
1035 if (!containsHeaderToken(upgrade, "websocket")) {
1036 return error.InvalidUpgradeHeader;
1037 }
1038 const connection_header =
1039 response.header("Connection") orelse
1040 return error.MissingConnectionHeader;
1041 if (!containsHeaderToken(
1042 connection_header,
1043 "upgrade",
1044 )) {
1045 return error.InvalidConnectionHeader;
1046 }
1047 const accepted =
1048 response.header("Sec-WebSocket-Accept") orelse
1049 return error.MissingUpgradeAccept;
1050 if (!std.mem.eql(
1051 u8,
1052 accepted,
1053 expected_accept,
1054 )) {
1055 return error.InvalidUpgradeAccept;
1056 }
1057 }
1058
1059 fn containsHeaderToken(
1060 value: []const u8,
1061 expected: []const u8,
1062 ) bool {
1063 var tokens = std.mem.splitScalar(u8, value, ',');
1064 while (tokens.next()) |token| {
1065 if (std.ascii.eqlIgnoreCase(
1066 std.mem.trim(u8, token, " \t"),
1067 expected,
1068 )) return true;
1069 }
1070 return false;
1071 }
1072
1073 fn connectionReader(
1074 connection: PersistentConnection,
1075 ) *std.Io.Reader {
1076 return switch (connection) {
1077 .plain => |conn| &conn.socket_reader.reader,
1078 .tls => |conn| &conn.tls_client.?.reader,
1079 };
1080 }
1081
1082 fn connectionWriter(
1083 connection: PersistentConnection,
1084 ) *std.Io.Writer {
1085 return switch (connection) {
1086 .plain => |conn| &conn.socket_writer.writer,
1087 .tls => |conn| &conn.tls_client.?.writer,
1088 };
1089 }
1090
1091 fn flushConnection(
1092 connection: PersistentConnection,
1093 ) !void {
1094 switch (connection) {
1095 .plain => |conn| {
1096 try conn.socket_writer.writer.flush();
1097 },
1098 .tls => |conn| {
1099 try conn.tls_client.?.writer.flush();
1100 try conn.socket_writer.writer.flush();
1101 },
1102 }
1103 }
1104
1105 fn connectionSocket(
1106 connection: PersistentConnection,
1107 ) sys.Socket {
1108 return switch (connection) {
1109 .plain => |conn| conn.socket,
1110 .tls => |conn| conn.socket,
1111 };
1112 }
1113
1114 fn connectionBuffered(
1115 connection: PersistentConnection,
1116 ) bool {
1117 return switch (connection) {
1118 .plain => |conn| conn.socket_reader.reader.bufferedLen() != 0,
1119 .tls => |conn| conn.tls_client.?.reader.bufferedLen() != 0 or
1120 conn.socket_reader.reader.bufferedLen() != 0,
1121 };
1122 }
1123
1124 fn destroyConnection(
1125 allocator: Allocator,
1126 connection: PersistentConnection,
1127 notify: bool,
1128 ) void {
1129 switch (connection) {
1130 .plain => |conn| conn.destroy(allocator),
1131 .tls => |conn| {
1132 conn.deinit(notify);
1133 allocator.destroy(conn);
1134 },
1135 }
1136 }
1137
1138 fn connectToHost(
1139 host: []const u8,
1140 port: u16,
1141 timeout_ms: u32,
1142 tcp_no_delay: bool,
1143 ) !sys.Socket {
1144 return connectToResolvedHost(
1145 host,
1146 port,
1147 timeout_ms,
1148 tcp_no_delay,
1149 );
1150 }
1151
1152 fn connectToResolvedHost(
1153 host: []const u8,
1154 port: u16,
1155 timeout_ms: u32,
1156 tcp_no_delay: bool,
1157 ) !sys.Socket {
1158 const address = try sys.resolveIpAddressForHost(host, port, .{});
1159 const socket = try sys.tcpStreamSocketForAddress(address, .{ .close_on_exec = true });
1160 errdefer sys.close(socket);
1161 try sys.connectIpAddress(socket, address);
1162 if (tcp_no_delay) try sys.setTcpNoDelay(socket);
1163 try applySocketTimeout(socket, timeout_ms);
1164 return socket;
1165 }
1166
1167 fn applySocketTimeout(socket: sys.Socket, timeout_ms: u32) !void {
1168 if (timeout_ms == 0) return;
1169 sys.setReadTimeout(socket, timeout_ms) catch |err| switch (err) {
1170 error.UnsupportedPlatform => return,
1171 else => return err,
1172 };
1173 sys.setWriteTimeout(socket, timeout_ms) catch |err| switch (err) {
1174 error.UnsupportedPlatform => return,
1175 else => return err,
1176 };
1177 }
1178
1179 fn testSocketPair() ![2]sys.Socket {
1180 return sys.socketPairUnixStream();
1181 }
1182
1183 test "SocketWriter writes data to peer" {
1184 const sockets = try testSocketPair();
1185 defer sys.close(sockets[0]);
1186 defer sys.close(sockets[1]);
1187
1188 var writer_buf: [32]u8 = undefined;
1189 var socket_writer = SocketWriter.init(sockets[0], &writer_buf);
1190 try socket_writer.writer.writeAll("hello ");
1191 try socket_writer.writer.writeAll("world");
1192 try socket_writer.writer.flush();
1193
1194 var recv_buf: [64]u8 = undefined;
1195 var total: usize = 0;
1196 const expected = "hello world";
1197 while (total < expected.len) {
1198 const n = sys.recv(sockets[1], recv_buf[total..], 0) catch unreachable;
1199 if (n == 0) break;
1200 total += n;
1201 }
1202 try std.testing.expectEqual(expected.len, total);
1203 try std.testing.expectEqualStrings(expected, recv_buf[0..total]);
1204 }
1205
1206 test "SocketWriter handles randomized chunking" {
1207 const sockets = try testSocketPair();
1208 defer sys.close(sockets[0]);
1209 defer sys.close(sockets[1]);
1210
1211 var writer_buf: [64]u8 = undefined;
1212 var socket_writer = SocketWriter.init(sockets[0], &writer_buf);
1213
1214 var rng = std.Random.DefaultPrng.init(0x5eedd00d);
1215 const random = rng.random();
1216
1217 var expected: [2048]u8 = undefined;
1218 var expected_len: usize = 0;
1219
1220 var i: usize = 0;
1221 while (i < 50) : (i += 1) {
1222 const len = random.intRangeAtMost(usize, 1, 32);
1223 var j: usize = 0;
1224 while (j < len) : (j += 1) {
1225 expected[expected_len + j] = random.int(u8);
1226 }
1227 try socket_writer.writer.writeAll(expected[expected_len .. expected_len + len]);
1228 expected_len += len;
1229 }
1230
1231 try socket_writer.writer.flush();
1232
1233 const recv = try std.testing.allocator.alloc(u8, expected_len);
1234 defer std.testing.allocator.free(recv);
1235
1236 var total: usize = 0;
1237 while (total < expected_len) {
1238 const n = sys.recv(sockets[1], recv[total..], 0) catch unreachable;
1239 if (n == 0) break;
1240 total += n;
1241 }
1242
1243 try std.testing.expectEqual(expected_len, total);
1244 try std.testing.expectEqualSlices(u8, expected[0..expected_len], recv[0..expected_len]);
1245 }
1246
1247 test "SocketWriter returns WriteFailed on broken pipe" {
1248 const sockets = try testSocketPair();
1249 defer sys.close(sockets[0]);
1250 sys.close(sockets[1]);
1251
1252 var writer_buf: [16]u8 = undefined;
1253 var socket_writer = SocketWriter.init(sockets[0], &writer_buf);
1254 try socket_writer.writer.writeAll("boom");
1255 const err = socket_writer.writer.flush();
1256 try std.testing.expectError(error.WriteFailed, err);
1257 try std.testing.expect(socket_writer.err != null);
1258 }
1259
1260 test "SocketReader zero-length readVec appends to buffer" {
1261 const sockets = try testSocketPair();
1262 defer sys.close(sockets[0]);
1263 defer sys.close(sockets[1]);
1264
1265 var read_buf: [4]u8 = undefined;
1266 var socket_reader = SocketReader.init(sockets[0], &read_buf);
1267 var zero_vec = [_][]u8{""};
1268
1269 try std.testing.expectEqual(@as(usize, 2), try sys.sendNoSignal(sockets[1], "ab"));
1270 try std.testing.expectEqual(@as(usize, 0), try socket_reader.reader.vtable.readVec(&socket_reader.reader, &zero_vec));
1271 try std.testing.expectEqual(@as(usize, 0), socket_reader.reader.seek);
1272 try std.testing.expectEqual(@as(usize, 2), socket_reader.reader.end);
1273 try std.testing.expectEqualStrings("ab", socket_reader.reader.buffer[0..socket_reader.reader.end]);
1274
1275 try std.testing.expectEqual(@as(usize, 2), try sys.sendNoSignal(sockets[1], "cd"));
1276 try std.testing.expectEqual(@as(usize, 0), try socket_reader.reader.vtable.readVec(&socket_reader.reader, &zero_vec));
1277 try std.testing.expectEqual(@as(usize, 0), socket_reader.reader.seek);
1278 try std.testing.expectEqual(@as(usize, 4), socket_reader.reader.end);
1279 try std.testing.expectEqualStrings("abcd", socket_reader.reader.buffer[0..socket_reader.reader.end]);
1280 }
1281
1282 const ClientWebSocketFixture = struct {
1283 const frame_payload_bytes: usize = 256;
1284 sockets: [2]sys.Socket,
1285 read_storage: [
1286 client_websocket.frame_header_bytes +
1287 frame_payload_bytes
1288 ]u8 = undefined,
1289 write_storage: [
1290 client_websocket.frame_header_bytes +
1291 frame_payload_bytes
1292 ]u8 = undefined,
1293 socket: ClientWebSocket,
1294
1295 fn init(self: *ClientWebSocketFixture) !void {
1296 self.sockets = try testSocketPair();
1297 var peer_owned = true;
1298 errdefer if (peer_owned) sys.close(self.sockets[1]);
1299 const connection: PersistentConnection = .{
1300 .plain = try PlainConn.create(
1301 std.testing.allocator,
1302 self.sockets[0],
1303 ),
1304 };
1305 var connection_owned = true;
1306 errdefer if (connection_owned) {
1307 destroyConnection(
1308 std.testing.allocator,
1309 connection,
1310 false,
1311 );
1312 };
1313 self.socket = try ClientWebSocket.init(
1314 std.testing.allocator,
1315 connection,
1316 .{
1317 .read = &self.read_storage,
1318 .write = &self.write_storage,
1319 .frame_payload_bytes = frame_payload_bytes,
1320 },
1321 );
1322 connection_owned = false;
1323 peer_owned = false;
1324 }
1325
1326 fn deinit(self: *ClientWebSocketFixture) void {
1327 self.socket.deinit();
1328 sys.close(self.sockets[1]);
1329 self.* = undefined;
1330 }
1331 };
1332
1333 fn sendServerFrame(
1334 peer: sys.Socket,
1335 output: []u8,
1336 payload: []const u8,
1337 ) !void {
1338 const encoded = try (http.Frame{
1339 .fin = true,
1340 .opcode = .binary,
1341 .mask = null,
1342 .payload = payload,
1343 }).serializeInto(
1344 output,
1345 ClientWebSocketFixture.frame_payload_bytes,
1346 );
1347 try std.testing.expectEqual(
1348 encoded.len,
1349 try sys.sendNoSignal(peer, encoded),
1350 );
1351 }
1352
1353 test "Client WebSocket exchanges bounded masked frames" {
1354 comptime {
1355 @stardustClaim(
1356 @import("alloc_phase").capacity.witness(@import("./websocket/root.zig").ClientWebsocketStorage, "http_client_websocket_network_overload"),
1357 null,
1358 null,
1359 null,
1360 null,
1361 null,
1362 null,
1363 );
1364 }
1365 comptime {
1366 @stardustClaim(
1367 @import("alloc_phase").capacity.witness(@import("./websocket/root.zig").ClientWebsocketStorage, "http_client_websocket_network_transitive_risk"),
1368 null,
1369 null,
1370 null,
1371 null,
1372 null,
1373 null,
1374 );
1375 }
1376 comptime {
1377 @stardustClaim(
1378 @import("alloc_phase").capacity.witness(@import("./websocket/root.zig").ClientWebsocketStorage, "http_client_websocket_network_foreign_risk"),
1379 null,
1380 null,
1381 null,
1382 null,
1383 null,
1384 null,
1385 );
1386 }
1387
1388 var fixture: ClientWebSocketFixture = undefined;
1389 try fixture.init();
1390 defer fixture.deinit();
1391 var server_wire: [
1392 client_websocket.frame_header_bytes +
1393 ClientWebSocketFixture.frame_payload_bytes
1394 ]u8 = undefined;
1395 try sendServerFrame(
1396 fixture.sockets[1],
1397 &server_wire,
1398 &.{},
1399 );
1400 const empty_message = (try fixture.socket.receive()) orelse
1401 return error.ExpectedWebSocketMessage;
1402 try std.testing.expectEqual(
1403 http.Opcode.binary,
1404 empty_message.opcode,
1405 );
1406 try std.testing.expectEqual(@as(usize, 0), empty_message.payload.len);
1407
1408 var extended_payload: [130]u8 = undefined;
1409 for (&extended_payload, 0..) |*byte, index| {
1410 byte.* = @intCast(index);
1411 }
1412 try sendServerFrame(
1413 fixture.sockets[1],
1414 &server_wire,
1415 &extended_payload,
1416 );
1417 const extended_message =
1418 (try fixture.socket.receive()) orelse
1419 return error.ExpectedWebSocketMessage;
1420 try std.testing.expectEqualSlices(
1421 u8,
1422 &extended_payload,
1423 extended_message.payload,
1424 );
1425
1426 try fixture.socket.sendBinary("bounded client frame");
1427 var client_wire: [64]u8 = undefined;
1428 const client_bytes = try sys.recv(
1429 fixture.sockets[1],
1430 &client_wire,
1431 0,
1432 );
1433 const parsed = try http.Frame.parse(
1434 client_wire[0..client_bytes],
1435 ClientWebSocketFixture.frame_payload_bytes,
1436 );
1437 try std.testing.expect(parsed.frame.mask != null);
1438 try std.testing.expectEqual(
1439 http.Opcode.binary,
1440 parsed.frame.opcode,
1441 );
1442 try std.testing.expectEqualStrings(
1443 "bounded client frame",
1444 parsed.frame.payload,
1445 );
1446 }
1447
1448 const TestResponseBuffer = struct {
1449 headers: [16]Header = undefined,
1450 head: [1024]u8 = undefined,
1451 body: [4096]u8 = undefined,
1452
1453 fn scratch(self: *TestResponseBuffer) ResponseScratch {
1454 return .{ .headers = &self.headers, .head = &self.head, .body = &self.body };
1455 }
1456 };
1457
1458 const TestOperationBuffer = struct {
1459 host: [64]u8 = undefined,
1460 target: [128]u8 = undefined,
1461 plain_read: [7]u8 = undefined,
1462 plain_write: [11]u8 = undefined,
1463
1464 fn scratch(self: *TestOperationBuffer) OperationScratch {
1465 return .{
1466 .host = &self.host,
1467 .target = &self.target,
1468 .plain_read = &self.plain_read,
1469 .plain_write = &self.plain_write,
1470 };
1471 }
1472 };
1473
1474 const NetworkStreamHandler = struct {
1475 head_count: usize = 0,
1476
1477 fn handler(self: *NetworkStreamHandler) StreamHandler {
1478 return .{ .ctx = self, .vtable = &vtable };
1479 }
1480
1481 fn onHead(context: *anyopaque, _: u16, _: []const Header) anyerror!void {
1482 const self: *NetworkStreamHandler = @ptrCast(@alignCast(context));
1483 self.head_count += 1;
1484 }
1485
1486 fn onChunk(_: *anyopaque, _: []const u8) anyerror!void {}
1487
1488 const vtable = StreamHandlerVTable{
1489 .onHead = &onHead,
1490 .onChunk = &onChunk,
1491 };
1492 };
1493
1494 fn emptyResponseScratch() ResponseScratch {
1495 return .{ .headers = &.{}, .head = &.{}, .body = &.{} };
1496 }
1497
1498 fn emptyOperationScratch() OperationScratch {
1499 return .{ .host = &.{}, .target = &.{}, .plain_read = &.{}, .plain_write = &.{} };
1500 }
1501
1502 test "Client plain request uses caller operation windows" {
1503 comptime {
1504 @stardustClaim(
1505 @import("alloc_phase").capacity.witness(@import("./operation/root.zig").ClientOperationStorage, "http_client_operation_network_overload"),
1506 null,
1507 null,
1508 null,
1509 null,
1510 null,
1511 null,
1512 );
1513 }
1514 comptime {
1515 @stardustClaim(
1516 @import("alloc_phase").capacity.witness(@import("./operation/root.zig").ClientOperationStorage, "http_client_operation_network_foreign_risk"),
1517 null,
1518 null,
1519 null,
1520 null,
1521 null,
1522 null,
1523 );
1524 }
1525
1526 const sockets = try testSocketPair();
1527 errdefer sys.close(sockets[0]);
1528 defer sys.close(sockets[1]);
1529
1530 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
1531 try std.testing.expectEqual(raw.len, try sys.sendNoSignal(sockets[1], raw));
1532
1533 var response_buffer = TestResponseBuffer{};
1534 var operation_buffer = TestOperationBuffer{};
1535 const parsed = try Client.doPlainRequest(
1536 operation_buffer.scratch(),
1537 response_buffer.scratch(),
1538 sockets[0],
1539 "localhost",
1540 "/",
1541 "GET",
1542 &.{},
1543 "",
1544 );
1545
1546 try std.testing.expectEqual(@as(u16, 200), parsed.status);
1547 try std.testing.expectEqualStrings("Hello", parsed.body);
1548 }
1549
1550 test "Client stream network reads enforce configured storage" {
1551 comptime {
1552 @stardustClaim(
1553 @import("alloc_phase").capacity.witness(@import("./stream/root.zig").ClientStreamStorage, "http_client_stream_network"),
1554 null,
1555 null,
1556 null,
1557 null,
1558 null,
1559 null,
1560 );
1561 }
1562
1563 const sockets = try testSocketPair();
1564 errdefer sys.close(sockets[0]);
1565 defer sys.close(sockets[1]);
1566
1567 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
1568 try std.testing.expectEqual(raw.len, try sys.sendNoSignal(sockets[1], raw));
1569
1570 var headers: [1]Header = undefined;
1571 var head: [8]u8 = undefined;
1572 var operation_buffer = TestOperationBuffer{};
1573 var handler = NetworkStreamHandler{};
1574 try std.testing.expectError(
1575 error.StreamHeadCapacityExceeded,
1576 Client.doPlainRequestStream(
1577 operation_buffer.scratch(),
1578 .{ .headers = &headers, .head = &head },
1579 sockets[0],
1580 "localhost",
1581 "/",
1582 "GET",
1583 &.{},
1584 "",
1585 handler.handler(),
1586 ),
1587 );
1588 try std.testing.expectEqual(@as(usize, 0), handler.head_count);
1589 }
1590
1591 test "Client.doPlainRequest respects socket read timeout" {
1592 const sockets = try testSocketPair();
1593 defer sys.close(sockets[1]);
1594 errdefer sys.close(sockets[0]);
1595 try applySocketTimeout(sockets[0], 10);
1596 var response_buffer = TestResponseBuffer{};
1597 var operation_buffer = TestOperationBuffer{};
1598 try std.testing.expectError(
1599 error.ReadFailed,
1600 Client.doPlainRequest(
1601 operation_buffer.scratch(),
1602 response_buffer.scratch(),
1603 sockets[0],
1604 "localhost",
1605 "/",
1606 "GET",
1607 &.{},
1608 "",
1609 ),
1610 );
1611 }
1612
1613 test "Client.doPlainRequest completes HEAD response at headers" {
1614 const sockets = try testSocketPair();
1615 errdefer sys.close(sockets[0]);
1616 defer sys.close(sockets[1]);
1617 try applySocketTimeout(sockets[0], 10);
1618
1619 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\n";
1620 try std.testing.expectEqual(raw.len, try sys.sendNoSignal(sockets[1], raw));
1621
1622 var response_buffer = TestResponseBuffer{};
1623 var operation_buffer = TestOperationBuffer{};
1624 const parsed = try Client.doPlainRequest(
1625 operation_buffer.scratch(),
1626 response_buffer.scratch(),
1627 sockets[0],
1628 "localhost",
1629 "/",
1630 "HEAD",
1631 &.{},
1632 "",
1633 );
1634
1635 try std.testing.expectEqual(@as(u16, 200), parsed.status);
1636 try std.testing.expectEqualStrings("5", parsed.header("Content-Length").?);
1637 try std.testing.expectEqual(@as(usize, 0), parsed.body.len);
1638 }
1639
1640 test "Client response retains headers in caller storage" {
1641 const sockets = try testSocketPair();
1642 errdefer sys.close(sockets[0]);
1643 defer sys.close(sockets[1]);
1644
1645 const raw = "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nSet-Cookie: sid=abc\r\nContent-Length: 5\r\n\r\nHello";
1646 try std.testing.expectEqual(raw.len, try sys.sendNoSignal(sockets[1], raw));
1647
1648 var response_buffer = TestResponseBuffer{};
1649 var operation_buffer = TestOperationBuffer{};
1650 const parsed = try Client.doPlainRequest(
1651 operation_buffer.scratch(),
1652 response_buffer.scratch(),
1653 sockets[0],
1654 "localhost",
1655 "/",
1656 "GET",
1657 &.{},
1658 "",
1659 );
1660
1661 try std.testing.expectEqualStrings("text/html", parsed.header("content-type").?);
1662 try std.testing.expectEqualStrings("sid=abc", parsed.header("Set-Cookie").?);
1663 }
1664
1665 test "Client full-response reads enforce configured storage" {
1666 comptime {
1667 @stardustClaim(
1668 @import("alloc_phase").capacity.witness(@import("./root.zig").ClientResponseStorage, "http_client_response_network_overload"),
1669 null,
1670 null,
1671 null,
1672 null,
1673 null,
1674 null,
1675 );
1676 }
1677 comptime {
1678 @stardustClaim(
1679 @import("alloc_phase").capacity.witness(@import("./root.zig").ClientResponseStorage, "http_client_response_network_foreign_risk"),
1680 null,
1681 null,
1682 null,
1683 null,
1684 null,
1685 null,
1686 );
1687 }
1688
1689 const sockets = try testSocketPair();
1690 errdefer sys.close(sockets[0]);
1691 defer sys.close(sockets[1]);
1692
1693 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHello";
1694 try std.testing.expectEqual(raw.len, try sys.sendNoSignal(sockets[1], raw));
1695
1696 var headers: [1]Header = undefined;
1697 var head: [128]u8 = undefined;
1698 var body: [4]u8 = undefined;
1699 var operation_buffer = TestOperationBuffer{};
1700 try std.testing.expectError(
1701 error.ResponseBodyCapacityExceeded,
1702 Client.doPlainRequest(
1703 operation_buffer.scratch(),
1704 .{ .headers = &headers, .head = &head, .body = &body },
1705 sockets[0],
1706 "localhost",
1707 "/",
1708 "GET",
1709 &.{},
1710 "",
1711 ),
1712 );
1713 }
1714
1715 const testing = std.testing;
1716
1717 test "Client.request: rejects unsupported scheme" {
1718 var client = Client.init(testing.allocator);
1719 defer client.deinit();
1720 try testing.expectError(
1721 error.UnsupportedScheme,
1722 client.request(
1723 emptyOperationScratch(),
1724 emptyResponseScratch(),
1725 "GET",
1726 "ftp://example.com/path",
1727 &.{},
1728 "",
1729 ),
1730 );
1731 }
1732
1733 test "Client high-level requests reject tunnel ownership before preparation" {
1734 var client = Client.init(testing.allocator);
1735 defer client.deinit();
1736
1737 try testing.expectError(
1738 error.TunnelUnsupported,
1739 client.request(
1740 emptyOperationScratch(),
1741 emptyResponseScratch(),
1742 "CONNECT",
1743 "http://example.test/",
1744 &.{},
1745 "",
1746 ),
1747 );
1748
1749 var stream_handler = NetworkStreamHandler{};
1750 try testing.expectError(
1751 error.TunnelUnsupported,
1752 client.requestStream(
1753 emptyOperationScratch(),
1754 .{ .headers = &.{}, .head = &.{} },
1755 "CONNECT",
1756 "http://example.test/",
1757 &.{},
1758 "",
1759 stream_handler.handler(),
1760 ),
1761 );
1762 try testing.expectEqual(@as(usize, 0), stream_handler.head_count);
1763 try Client.rejectTunnelMethod("connect");
1764 }
1765
1766 test "Client request rejects operation capacity before network effects" {
1767 comptime {
1768 @stardustClaim(
1769 @import("alloc_phase").capacity.witness(@import("./operation/root.zig").ClientOperationStorage, "http_client_operation_atomic"),
1770 null,
1771 null,
1772 null,
1773 null,
1774 null,
1775 null,
1776 );
1777 }
1778
1779 var client = Client.init(testing.allocator);
1780 defer client.deinit();
1781 var short_host: [11]u8 = undefined;
1782 var full_host: [12]u8 = undefined;
1783 var short_target: [1]u8 = undefined;
1784 var full_target: [2]u8 = undefined;
1785 var byte: [1]u8 = undefined;
1786
1787 try testing.expectError(
1788 error.ClientHostCapacityExceeded,
1789 client.request(.{
1790 .host = &short_host,
1791 .target = &full_target,
1792 .plain_read = &byte,
1793 .plain_write = &byte,
1794 }, emptyResponseScratch(), "GET", "http://example.test/x", &.{}, ""),
1795 );
1796 try testing.expectError(
1797 error.ClientTargetCapacityExceeded,
1798 client.request(.{
1799 .host = &full_host,
1800 .target = &short_target,
1801 .plain_read = &byte,
1802 .plain_write = &byte,
1803 }, emptyResponseScratch(), "GET", "http://example.test/x", &.{}, ""),
1804 );
1805 try testing.expectError(
1806 error.ClientPlainReadCapacityExceeded,
1807 client.request(.{
1808 .host = &full_host,
1809 .target = &full_target,
1810 .plain_read = &.{},
1811 .plain_write = &byte,
1812 }, emptyResponseScratch(), "GET", "http://example.test/x", &.{}, ""),
1813 );
1814 try testing.expectError(
1815 error.ClientPlainWriteCapacityExceeded,
1816 client.request(.{
1817 .host = &full_host,
1818 .target = &full_target,
1819 .plain_read = &byte,
1820 .plain_write = &.{},
1821 }, emptyResponseScratch(), "GET", "http://example.test/x", &.{}, ""),
1822 );
1823 }
1824
1825 test "ClientResponse.parse: simple 200" {
1826 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 11\r\n\r\nHello World";
1827 var response_buffer = TestResponseBuffer{};
1828 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1829
1830 try testing.expectEqual(@as(u16, 200), resp.status);
1831 try testing.expectEqualStrings("Hello World", resp.body);
1832 try testing.expect(resp.connection_reusable);
1833 }
1834
1835 test "ClientResponse persistence follows version framing and Connection" {
1836 var response_buffer = TestResponseBuffer{};
1837 const closed = try ClientResponse.parse(
1838 response_buffer.scratch(),
1839 "HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n",
1840 );
1841 try testing.expect(!closed.connection_reusable);
1842 const legacy = try ClientResponse.parse(
1843 response_buffer.scratch(),
1844 "HTTP/1.0 204 No Content\r\n\r\n",
1845 );
1846 try testing.expect(!legacy.connection_reusable);
1847 const legacy_kept = try ClientResponse.parse(
1848 response_buffer.scratch(),
1849 "HTTP/1.0 204 No Content\r\nConnection: keep-alive\r\n\r\n",
1850 );
1851 try testing.expect(legacy_kept.connection_reusable);
1852 }
1853
1854 test "ClientResponse.parseForMethod: HEAD preserves headers and omits body" {
1855 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 11\r\n\r\n";
1856 var response_buffer = TestResponseBuffer{};
1857 const resp = try ClientResponse.parseForMethod(
1858 response_buffer.scratch(),
1859 raw,
1860 "HEAD",
1861 );
1862
1863 try testing.expectEqual(@as(u16, 200), resp.status);
1864 try testing.expectEqualStrings("11", resp.header("Content-Length").?);
1865 try testing.expectEqual(@as(usize, 0), resp.body.len);
1866 }
1867
1868 test "ClientResponse.parse: 401 with JSON body" {
1869 const raw = "HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: 24\r\n\r\n{\"error\":\"unauthorized\"}";
1870 var response_buffer = TestResponseBuffer{};
1871 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1872
1873 try testing.expectEqual(@as(u16, 401), resp.status);
1874 try testing.expectEqualStrings("{\"error\":\"unauthorized\"}", resp.body);
1875 }
1876
1877 test "ClientResponse.parse: rejects truncated content-length body" {
1878 const raw = "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nHel";
1879 var response_buffer = TestResponseBuffer{};
1880 try testing.expectError(
1881 error.MalformedResponse,
1882 ClientResponse.parse(response_buffer.scratch(), raw),
1883 );
1884 }
1885
1886 test "ClientResponse.parse: no-body statuses preserve headers and omit body" {
1887 var response_buffer = TestResponseBuffer{};
1888 const no_content = try ClientResponse.parse(
1889 response_buffer.scratch(),
1890 "HTTP/1.1 204 No Content\r\nContent-Length: 5\r\n\r\nHello",
1891 );
1892 try testing.expectEqual(@as(u16, 204), no_content.status);
1893 try testing.expectEqualStrings("5", no_content.header("Content-Length").?);
1894 try testing.expectEqual(@as(usize, 0), no_content.body.len);
1895
1896 const not_modified = try ClientResponse.parse(
1897 response_buffer.scratch(),
1898 "HTTP/1.1 304 Not Modified\r\nContent-Length: 5\r\n\r\nHello",
1899 );
1900 try testing.expectEqual(@as(u16, 304), not_modified.status);
1901 try testing.expectEqualStrings("5", not_modified.header("Content-Length").?);
1902 try testing.expectEqual(@as(usize, 0), not_modified.body.len);
1903 }
1904
1905 test "ClientResponse.parse: truncated chunked body errors" {
1906 const raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n";
1907 var response_buffer = TestResponseBuffer{};
1908 try testing.expectError(
1909 error.MalformedResponse,
1910 ClientResponse.parse(response_buffer.scratch(), raw),
1911 );
1912 }
1913
1914 test "ClientResponse.parse: chunked transfer encoding" {
1915 const raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n6\r\n World\r\n0\r\n\r\n";
1916 var response_buffer = TestResponseBuffer{};
1917 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1918
1919 try testing.expectEqual(@as(u16, 200), resp.status);
1920 try testing.expectEqualStrings("Hello World", resp.body);
1921 }
1922
1923 test "ClientResponse.parse: chunked trailers" {
1924 const raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\nDigest: sha-256=abc123\r\nX-Trace: bench\r\n\r\n";
1925 var response_buffer = TestResponseBuffer{};
1926 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1927
1928 try testing.expectEqual(@as(u16, 200), resp.status);
1929 try testing.expectEqualStrings("Hello", resp.body);
1930 }
1931
1932 test "ClientResponse.parse: transfer encoding token is case-insensitive" {
1933 const raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: gzip, Chunked\r\n\r\n5\r\nHello\r\n0\r\n\r\n";
1934 var response_buffer = TestResponseBuffer{};
1935 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1936
1937 try testing.expectEqual(@as(u16, 200), resp.status);
1938 try testing.expectEqualStrings("Hello", resp.body);
1939 }
1940
1941 test "ClientResponse.parse: empty chunked body" {
1942 const raw = "HTTP/1.1 204 No Content\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n";
1943 var response_buffer = TestResponseBuffer{};
1944 const resp = try ClientResponse.parse(response_buffer.scratch(), raw);
1945
1946 try testing.expectEqual(@as(u16, 204), resp.status);
1947 try testing.expectEqual(@as(usize, 0), resp.body.len);
1948 }