Something went wrong. Try again.
websocket
Something went wrong. Try again.
54 kB · 1441 lines
Zig
at zig-0.17
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442const std = @import("std");const proto = @import("../proto.zig");const net_io = @import("../net_io.zig");const buffer = @import("../buffer.zig");
const ascii = std.ascii;const Io = std.Io;const net = Io.net;const posix = std.posix;const tls = std.crypto.tls;const log = std.log.scoped(.websocket);
const Reader = proto.Reader;const Allocator = std.mem.Allocator;const Bundle = std.crypto.Certificate.Bundle;const CompressionOpts = @import("../websocket.zig").Compression;const ServerHandshake = @import("../server/handshake.zig").Handshake;
fn milliTimestamp(io: Io) i64 { const ts = Io.Timestamp.now(io, .real); return @intCast(@divTrunc(ts.nanoseconds, std.time.ns_per_ms));}
fn ReadLoopHandler(comptime T: type) type { const info = @typeInfo(T);
switch (info) { .@"struct" => |struct_info| { if (struct_info.is_tuple) @compileError("readLoop: handler does not support tuples.");
return T; }, .pointer => |ptr_info| { switch (ptr_info.size) { .one => return ReadLoopHandler(ptr_info.child), else => @compileError("readLoop: handler does not support Slice, C and Many pointers."), } }, else => @compileError("readLoop: expected handler to be a struct or pointer to a struct but found '" ++ @tagName(info) ++ "'"), }}
pub const Client = struct { io: Io, stream: Stream, _reader: Reader, _closed: bool, _compression_opts: ?CompressionOpts, _compression: ?Client.Compression = null,
/// populated when handshake() fails with error.InvalidHandshakeResponse /// because the server answered the upgrade with a non-101 HTTP response. /// servers reject upgrades pre-connection with meaningful statuses and /// JSON error envelopes (e.g. jetstream's 400 CursorTooOld); without /// this the caller only sees the bare error and cannot distinguish a /// terminal rejection from a transient one. null when the handshake /// failed for any other reason (malformed response, timeout, transport). handshake_failure: ?HandshakeFailure = null,
// Serializes writes from concurrent tasks (ping loop, auto-pong, close). // Matches server-side Conn.lock pattern. _write_lock: Io.Mutex = .init,
// When creating a client, we can either be given a BufferProvider or create // one ourselves. If we create it ourselves (in init), we "own" it and must // free it on deinit. (The reference to the buffer provider is already in the // reader, no need to hold another reference in the client). _own_bp: bool,
// For advanced cases, a custom masking function can be provided. Masking // is a security feature that only really makes sense in the browser. If you // aren't running websockets in the browser AND you control both the client // and the server, you could get a performance boost by not masking. _mask_fn: *const fn (Io) [4]u8,
pub const Config = struct { port: u16, host: []const u8, tls: bool = false, max_size: usize = 65536, buffer_size: usize = 4096, ca_bundle: ?Bundle = null, mask_fn: ?*const fn (Io) [4]u8 = null, buffer_provider: ?*buffer.Provider = null, compression: ?CompressionOpts = null, };
pub const HandshakeFailure = struct { status: u16, body_len: usize = 0, body_buf: [1024]u8 = undefined,
pub fn body(self: *const HandshakeFailure) []const u8 { return self.body_buf[0..self.body_len]; } };
pub const HandshakeOpts = struct { timeout_ms: u32 = 10000, headers: ?[]const u8 = null, };
const Compression = struct { allocator: Allocator, retain_writer: bool, write_treshold: usize, writer: std.Io.Writer.Allocating, };
pub fn init(io: Io, allocator: Allocator, config: Config) !Client { if (config.compression != null) { log.err("Compression is disabled as part of the 0.15 upgrade. I do hope to re-enable it soon.", .{}); return error.InvalidConfiguraion; }
// 0.16: networking via Io.net.HostName const host_name = try net.HostName.init(config.host); // 0.16: connect requires mode option (stream vs datagram) const net_stream = try host_name.connect(io, config.port, .{ .mode = .stream });
var tls_client: ?*TLSClient = null; if (config.tls) { tls_client = try TLSClient.init(allocator, io, net_stream, &config); } const stream = Stream.init(io, net_stream, tls_client);
var own_bp = false; var buffer_provider: *buffer.Provider = undefined;
// If a buffer_provider is provided, we'll use that. // If it isn't, we need to create one which also means we now "own" it // and we're responsible for cleaning it up if (config.buffer_provider) |shared_bp| { buffer_provider = shared_bp; } else { own_bp = true; buffer_provider = try allocator.create(buffer.Provider); errdefer allocator.destroy(buffer_provider); buffer_provider.* = try buffer.Provider.init(allocator, .{ .size = 0, .count = 0, .max = config.max_size, }); }
errdefer if (own_bp) { buffer_provider.deinit(); allocator.destroy(buffer_provider); };
const reader_buf = try buffer_provider.allocator.alloc(u8, config.buffer_size); errdefer buffer_provider.allocator.free(reader_buf);
return .{ .io = io, .stream = stream, ._closed = false, ._own_bp = own_bp, ._mask_fn = config.mask_fn orelse generateMask, ._compression_opts = null, //TODO: ZIG 0.15 ._reader = Reader.init(reader_buf, buffer_provider, null), }; }
// 0.16: Alternative init that accepts a pre-existing stream. // Supports TLS if config.tls is set (uses config.host for SNI). pub fn initWithStream(io: Io, allocator: Allocator, net_stream: net.Stream, config: Config) !Client { var tls_client: ?*TLSClient = null; if (config.tls) { tls_client = try TLSClient.init(allocator, io, net_stream, &config); } const stream = Stream.init(io, net_stream, tls_client);
var own_bp = false; var buffer_provider: *buffer.Provider = undefined;
if (config.buffer_provider) |shared_bp| { buffer_provider = shared_bp; } else { own_bp = true; buffer_provider = try allocator.create(buffer.Provider); errdefer allocator.destroy(buffer_provider); buffer_provider.* = try buffer.Provider.init(allocator, .{ .size = 0, .count = 0, .max = config.max_size, }); }
errdefer if (own_bp) { buffer_provider.deinit(); allocator.destroy(buffer_provider); };
const reader_buf = try buffer_provider.allocator.alloc(u8, config.buffer_size); errdefer buffer_provider.allocator.free(reader_buf);
return .{ .io = io, .stream = stream, ._closed = false, ._own_bp = own_bp, ._mask_fn = config.mask_fn orelse generateMask, ._compression_opts = null, ._reader = Reader.init(reader_buf, buffer_provider, null), }; }
pub fn deinit(self: *Client) void { self.closeStream();
const larger_buffer_provider = self._reader.large_buffer_provider; const allocator = larger_buffer_provider.allocator; allocator.free(self._reader.static);
self._reader.deinit();
if (self._own_bp) { larger_buffer_provider.deinit(); allocator.destroy(larger_buffer_provider); } }
pub fn handshake(self: *Client, path: []const u8, opts: HandshakeOpts) !void { const stream = &self.stream; errdefer self.closeStream();
// we've already setup our reader, and the reader has a static buffer // we might as well use it! const buf = self._reader.static; const key = blk: { const bin_key = generateKey(self.io); var encoded_key: [24]u8 = undefined; break :blk std.base64.standard.Encoder.encode(&encoded_key, &bin_key); };
try sendHandshake(path, key, buf, &opts, self._compression_opts != null, stream);
self.handshake_failure = null; const res = try HandShakeReply.read(buf, key, &opts, self._compression_opts != null, stream, &self.handshake_failure); errdefer self.close(.{ .code = 1001 }) catch unreachable;
// Set up compression with agreed-on parameters if (res.compression) { try self.setupCompression(); }
// We might have read more than handshake response. If so, readHandshakeReply // has positioned the extra data at the start of the buffer, but we need // to set the length. self._reader.pos = res.over_read; }
fn setupCompression(self: *Client) !void { std.debug.assert(self._compression_opts != null); self._reader.allow_compressed = true;
const allocator = self._reader.large_buffer_provider.allocator; const config = self._compression_opts.?; self._compression = .{ .allocator = allocator, .write_treshold = config.write_threshold.?, .retain_writer = config.retain_write_buffer, .writer = std.Io.Writer.Allocating.init(allocator), }; }
pub fn readLoop(self: *Client, handler: anytype) !void { const Handler = ReadLoopHandler(@TypeOf(handler)); var reader = &self._reader;
defer if (comptime std.meta.hasFn(Handler, "close")) { handler.close(); };
// block until we have data try self.readTimeout(0);
while (true) { const message = self.read() catch |err| switch (err) { error.Closed => return, else => return err, } orelse unreachable;
const message_type = message.type; defer reader.done(message_type);
switch (message_type) { .text, .binary => { switch (comptime @typeInfo(@TypeOf(Handler.serverMessage)).@"fn".param_types.len) { 2 => try handler.serverMessage(message.data), 3 => try handler.serverMessage(message.data, if (message_type == .text) .text else .binary), else => @compileError(@typeName(Handler) ++ ".serverMessage must accept 2 or 3 parameters"), } }, .ping => if (comptime std.meta.hasFn(Handler, "serverPing")) { try handler.serverPing(message.data); } else { // @constCast is safe because we know message.data points to // reader.buffer.buf, which we own and which can be mutated try self.writeFrame(.pong, @constCast(message.data)); }, .close => { if (comptime std.meta.hasFn(Handler, "serverClose")) { try handler.serverClose(message.data); } else { self.close(.{}) catch unreachable; } return; }, .pong => if (comptime std.meta.hasFn(Handler, "serverPong")) { try handler.serverPong(message.data); }, } } }
pub const HeartbeatConfig = struct { /// ping interval in milliseconds. readTimeout is set to this value. /// when no data arrives within the interval, a ping is sent. interval_ms: u32 = 30_000, /// close connection after this many consecutive intervals with no data or pong. max_failures: u32 = 4, };
pub fn readLoopWithHeartbeat(self: *Client, handler: anytype, heartbeat: HeartbeatConfig) !void { const Handler = ReadLoopHandler(@TypeOf(handler)); var reader = &self._reader;
defer if (comptime std.meta.hasFn(Handler, "close")) { handler.close(); };
try self.readTimeout(heartbeat.interval_ms);
var pending_pings: u32 = 0;
while (true) { const message = self.read() catch |err| switch (err) { error.Closed => return, else => return err, } orelse { // timeout — no data in interval pending_pings += 1; if (pending_pings >= heartbeat.max_failures) { self.close(.{}) catch {}; return error.Closed; } self.writePing(&.{}) catch { self.close(.{}) catch {}; return error.Closed; }; continue; };
// any received frame proves liveness pending_pings = 0;
const message_type = message.type; defer reader.done(message_type);
switch (message_type) { .text, .binary => { switch (comptime @typeInfo(@TypeOf(Handler.serverMessage)).@"fn".param_types.len) { 2 => try handler.serverMessage(message.data), 3 => try handler.serverMessage(message.data, if (message_type == .text) .text else .binary), else => @compileError(@typeName(Handler) ++ ".serverMessage must accept 2 or 3 parameters"), } }, .ping => if (comptime std.meta.hasFn(Handler, "serverPing")) { try handler.serverPing(message.data); } else { try self.writeFrame(.pong, @constCast(message.data)); }, .close => { if (comptime std.meta.hasFn(Handler, "serverClose")) { try handler.serverClose(message.data); } else { self.close(.{}) catch unreachable; } return; }, .pong => if (comptime std.meta.hasFn(Handler, "serverPong")) { try handler.serverPong(message.data); }, } } }
pub fn read(self: *Client) !?proto.Message { var reader = &self._reader; const stream = &self.stream;
while (true) { // try to read a message from our buffer first, before trying to // get more data from the socket. const has_more, const message = reader.read() catch |err| { self.close(.{ .code = 1002 }) catch unreachable; return err; } orelse { // 0.16: Io vtable error set changed reader.fill(stream) catch |err| { // Check for timeout/would-block type errors if (err == error.Canceled or err == error.WouldBlock) return null; // Check for connection closed errors if (err == error.Closed or err == error.ConnectionResetByPeer or err == error.NotOpenForReading) { @atomicStore(bool, &self._closed, true, .monotonic); return error.Closed; } self.close(.{ .code = 1002 }) catch unreachable; return err; }; continue; };
_ = has_more; return message; } }
pub fn done(self: *Client, message: proto.Message) void { self._reader.done(message.type); }
pub fn readLoopInNewThread(self: *Client, h: anytype) !std.Thread { return std.Thread.spawn(.{}, readLoopOwnedThread, .{ self, h }); }
fn readLoopOwnedThread(self: *Client, h: anytype) void { self.readLoop(h) catch {}; }
pub fn writeTimeout(self: *const Client, ms: u32) !void { return self.stream.writeTimeout(ms); }
pub fn readTimeout(self: *const Client, ms: u32) !void { return self.stream.readTimeout(ms); }
pub fn write(self: *Client, data: []u8) !void { return self.writeFrame(.text, data); }
pub fn writeText(self: *Client, data: []u8) !void { return self.writeFrame(.text, data); }
pub fn writeBin(self: *Client, data: []u8) !void { return self.writeFrame(.binary, data); }
pub fn writePing(self: *Client, data: []u8) !void { return self.writeFrame(.ping, data); }
pub fn writePong(self: *Client, data: []u8) !void { return self.writeFrame(.pong, data); }
const CloseOpts = struct { code: ?u16 = null, reason: []const u8 = "", };
pub fn close(self: *Client, opts: CloseOpts) !void { if (@atomicRmw(bool, &self._closed, .Xchg, true, .monotonic) == true) { // already closed return; }
// Hold the write lock across BOTH the close frame and the stream // teardown. stream.close() frees the TLS client, so a concurrent // writer (a keepalive ping task, say) that was queued on the lock // when close ran must observe error.Closed on acquiring it — not // write through freed TLS state. Releasing the lock between the // close frame and stream.close() left exactly that window, and a // production ping fiber died in it (GPF, 2026-08-27). self._write_lock.lockUncancelable(self.io); defer self._write_lock.unlock(self.io); defer self.stream.close();
const code = opts.code orelse { self.writeFrameLocked(.close, "") catch {}; return; };
const reason = opts.reason; if (reason.len > 123) { return error.ReasonTooLong; }
var buf: [125]u8 = undefined; buf[0] = @intCast((code >> 8) & 0xFF); buf[1] = @intCast(code & 0xFF);
const end = 2 + reason.len; @memcpy(buf[2..end], reason); self.writeFrameLocked(.close, buf[0..end]) catch {}; }
pub fn writeFrame(self: *Client, op_code: proto.OpCode, data: []u8) !void { // Serialize writes — concurrent ping/pong/close must not interleave // frames. The closed check must happen under the lock: close() tears // the stream (and TLS state) down while holding it, so a writer that // slept through a close must fail here rather than touch the stream. self._write_lock.lockUncancelable(self.io); defer self._write_lock.unlock(self.io); if (@atomicLoad(bool, &self._closed, .monotonic)) { return error.Closed; } return self.writeFrameLocked(op_code, data); }
/// the write lock must be held (and _closed checked, except by close /// itself) before calling. fn writeFrameLocked(self: *Client, op_code: proto.OpCode, data: []u8) !void { const payload = data; const compressed = false;
// maximum possible prefix length. op_code + length_type + 8byte length + 4 byte mask var buf: [14]u8 = undefined; const header = proto.writeFrameHeader(&buf, op_code, payload.len, compressed);
const header_len = header.len; const header_end = header.len + 4; // for the mask
buf[1] |= 128; // indicate that the payload is masked
const mask = self._mask_fn(self.io); @memcpy(buf[header_len..header_end], &mask);
if (payload.len > 0) { proto.mask(&mask, payload); }
try self.stream.writeAll(buf[0..header_end]); if (payload.len > 0) { try self.stream.writeAll(payload); } }
pub fn isClosed(self: *const Client) bool { return @atomicLoad(bool, &self._closed, .monotonic); }
fn closeStream(self: *Client) void { if (@atomicRmw(bool, &self._closed, .Xchg, true, .monotonic) == false) { // same lock discipline as close(): never free the stream while a // writer holds (or is queued on) the write lock. self._write_lock.lockUncancelable(self.io); defer self._write_lock.unlock(self.io); self.stream.close(); } }};
// wraps a net.Stream and optional a tls.Clientpub const Stream = struct { io: Io, stream: net.Stream, tls_client: ?*TLSClient = null,
pub fn init(io: Io, stream: net.Stream, tls_client: ?*TLSClient) Stream { return .{ .io = io, .stream = stream, .tls_client = tls_client, }; }
pub fn close(self: *Stream) void { if (self.tls_client) |tls_client| { self.stream.shutdown(self.io, .both) catch {}; tls_client.deinit(); } self.stream.close(self.io); }
pub fn read(self: *Stream, buf: []u8) !usize { if (self.tls_client) |tls_client| { var w: std.Io.Writer = .fixed(buf); while (true) { const n = try tls_client.client.reader.stream(&w, .limited(buf.len)); if (n != 0) { return n; } } } var bufs = [_][]u8{buf}; return net_io.read(self.io, self.stream.socket.handle, &bufs) catch |err| { return switch (err) { error.ConnectionResetByPeer => error.ConnectionResetByPeer, error.ConnectionTimedOut => error.WouldBlock, else => error.Unexpected, }; }; }
pub fn writeAll(self: *Stream, data: []const u8) !void { if (self.tls_client) |tls_client| { try tls_client.client.writer.writeAll(data); try tls_client.client.writer.flush(); try tls_client.stream_writer.interface.flush(); return; } var remaining = data; while (remaining.len > 0) { const n = net_io.write(self.io, self.stream.socket.handle, remaining) catch |err| { return switch (err) { error.ConnectionResetByPeer => error.ConnectionResetByPeer, else => error.Unexpected, }; }; if (n == 0) return error.Unexpected; remaining = remaining[n..]; } }
const zero_timeout = std.mem.toBytes(posix.timeval{ .sec = 0, .usec = 0 }); pub fn writeTimeout(self: *const Stream, ms: u32) !void { return self.setTimeout(posix.SO.SNDTIMEO, ms); }
pub fn readTimeout(self: *const Stream, ms: u32) !void { return self.setTimeout(posix.SO.RCVTIMEO, ms); }
fn setTimeout(self: *const Stream, opt_name: u32, ms: u32) !void { if (ms == 0) { return self.setsockopt(opt_name, &zero_timeout); }
const timeout = std.mem.toBytes(posix.timeval{ .sec = @intCast(@divTrunc(ms, 1000)), .usec = @intCast(@mod(ms, 1000) * 1000), }); return self.setsockopt(opt_name, &timeout); }
pub fn setsockopt(self: *const Stream, opt_name: u32, value: []const u8) !void { return setConnSockOpt(self.stream.socket.handle, posix.SOL.SOCKET, opt_name, value); }};
/// setsockopt for a *connection* socket.////// std.posix.setsockopt maps BADF/NOTSOCK/INVAL/FAULT to `unreachable`/// ("always a race condition"). On a listening socket that holds. On a/// connection socket it does not: the peer can reset, or another thread can/// close the fd, between connect and the timeout call. `unreachable` cannot be/// caught, so that race aborts the whole process instead of surfacing as an/// error the caller's reconnect path would handle — observed as a panic inside/// the client handshake when an upstream dropped the connection.////// FAULT is deliberately NOT swallowed: a bad option pointer is our own bug,/// not a peer's doing, and hiding it would trade one silent failure for/// another.////// The timeouts this backs are advisory: if the socket really is gone, the/// following read or write reports it properly. So swallow exactly the arms the/// stdlib calls impossible and keep every other error meaningful.////// The server side took the same approach in `setSockOptBestEffort`/// (src/server/server.zig).fn setConnSockOpt(fd: posix.socket_t, level: i32, optname: u32, opt: []const u8) !void { switch (posix.errno(posix.system.setsockopt(fd, level, optname, opt.ptr, @intCast(opt.len)))) { .SUCCESS => {}, // The socket died under us. Not our failure to report. .BADF, .NOTSOCK, .INVAL => {}, .DOM => return error.TimeoutTooBig, .ISCONN => return error.AlreadyConnected, .NOPROTOOPT => return error.InvalidProtocolOption, .NOMEM, .NOBUFS => return error.SystemResources, .PERM => return error.PermissionDenied, .NODEV => return error.NoDevice, .OPNOTSUPP => return error.OperationUnsupported, else => |err| return posix.unexpectedErrno(err), }}
const TLSClient = struct { client: tls.Client, stream: net.Stream, stream_writer: net.Stream.Writer, stream_reader: net.Stream.Reader, arena: std.heap.ArenaAllocator,
fn init(allocator: Allocator, io: Io, stream: net.Stream, config: *const Client.Config) !*TLSClient { var arena = std.heap.ArenaAllocator.init(allocator); errdefer arena.deinit();
const aa = arena.allocator();
const bundle_ptr = try aa.create(Bundle); if (config.ca_bundle) |ca| { bundle_ptr.* = ca; } else { bundle_ptr.* = .{ .map = .empty, .bytes = .empty }; try bundle_ptr.rescan(aa, io, Io.Timestamp.zero); }
const rwlock = try aa.create(std.Io.RwLock); rwlock.* = std.Io.RwLock.init;
// The TLS input and output have to be max_ciphertext_record_len each. // It isn't clear to me how big the un-encrypted reader and writer // need to be. I would think 0, but that will fail an assertion. I // don't think that it's right that we need 4 buffers, but apparently // we do. Until i figure this out, using 4 x max_ciphertext_record_len // seems like the only safe choice. const buf_len = std.crypto.tls.max_ciphertext_record_len; var buf = try aa.alloc(u8, buf_len * 4);
const self = try aa.create(TLSClient); self.* = .{ .stream = stream, .arena = arena, .client = undefined, .stream_writer = stream.writer(io, buf.ptr[0..buf_len][0..buf_len]), .stream_reader = stream.reader(io, buf.ptr[buf_len .. 2 * buf_len][0..buf_len]), };
var entropy: [tls.Client.Options.entropy_len]u8 = undefined; io.random(&entropy); self.client = try tls.Client.init( &self.stream_reader.interface, &self.stream_writer.interface, .{ .ca = .{ .bundle = .{ .gpa = aa, .io = io, .lock = rwlock, .bundle = bundle_ptr } }, .host = .{ .explicit = config.host }, .read_buffer = buf.ptr[2 * buf_len .. 3 * buf_len][0..buf_len], .write_buffer = buf.ptr[3 * buf_len .. 4 * buf_len][0..buf_len], .entropy = &entropy, .realtime_now = Io.Timestamp.now(io, .real), }, );
return self; }
fn deinit(self: *TLSClient) void { _ = self.client.end() catch {}; self.arena.deinit(); }};
fn generateKey(io: Io) [16]u8 { if (comptime @import("builtin").is_test) { return [16]u8{ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16 }; } // 0.16: io.random() fills a buffer var key: [16]u8 = undefined; io.random(&key); return key;}
fn generateMask(io: Io) [4]u8 { // 0.16: io.random() fills a buffer var mask: [4]u8 = undefined; io.random(&mask); return mask;}
fn sendHandshake(path: []const u8, key: []const u8, buf: []u8, opts: *const Client.HandshakeOpts, compression: bool, stream: anytype) !void { @memcpy(buf[0..4], "GET "); var pos: usize = 4; var end = pos + path.len;
{ @memcpy(buf[pos..end], path); pos = end; }
{ const headers = " HTTP/1.1\r\ncontent-length: 0\r\nupgrade: websocket\r\nsec-websocket-version: 13\r\nconnection: upgrade\r\nsec-websocket-key: "; end = pos + headers.len; @memcpy(buf[pos..end], headers);
pos = end; end = pos + key.len; @memcpy(buf[pos..end], key); }
if (compression) { // NOTE: client_max_window_bits is unsupported const permessage_deflate = "\r\nSec-WebSocket-Extensions: permessage-deflate; server_no_context_takeover; client_no_context_takeover"; pos = end; end = pos + permessage_deflate.len; @memcpy(buf[pos..end], permessage_deflate); }
{ pos = end; end = pos + 2; @memcpy(buf[pos..end], "\r\n"); pos = end; }
if (opts.headers) |extra_headers| { end = pos + extra_headers.len; @memcpy(buf[pos..end], extra_headers); pos = end; if (!std.mem.endsWith(u8, extra_headers, "\r\n")) { buf[pos] = '\r'; buf[pos + 1] = '\n'; pos += 2; } } buf[pos] = '\r'; buf[pos + 1] = '\n';
try stream.writeTimeout(opts.timeout_ms); try stream.writeAll(buf[0 .. pos + 2]); try stream.writeTimeout(0);}
const HandShakeReply = struct { compression: bool, over_read: usize,
/// On a non-101 status line, read the rest of the rejection (bounded by /// buf and the already-armed read timeout) and capture the status plus /// up to body_buf.len bytes of body. Returns null when the status line /// is too short/unparseable — that is a malformed response, not a /// rejection. Content-Length, when present and sane, bounds the body /// read exactly; otherwise reading stops at close, timeout, or the cap. fn captureRejection(status_line: []const u8, buf: []u8, initial_pos: usize, stream: anytype) ?Client.HandshakeFailure { if (status_line.len < 12 or status_line[8] != ' ') return null; const status = std.fmt.parseInt(u16, status_line[9..12], 10) catch return null; var f: Client.HandshakeFailure = .{ .status = status };
var pos = initial_pos; const body_start = while (true) { if (std.mem.indexOf(u8, buf[0..pos], "\r\n\r\n")) |i| break i + 4; if (pos == buf.len) return f; const n = stream.read(buf[pos..]) catch return f; if (n == 0) return f; pos += n; };
// Only read further when Content-Length promises the bytes exist: a // speculative read on an open connection blocks until the handshake // timeout (and SO_RCVTIMEO's EAGAIN panics under Io.Threaded in // debug). Without Content-Length, the body is whatever arrived with // the headers. if (findContentLength(buf[0..body_start])) |cl| { const want = @min(cl, f.body_buf.len); while (pos - body_start < want and pos < buf.len) { const n = stream.read(buf[pos..]) catch break; if (n == 0) break; pos += n; } } const len = @min(pos - body_start, f.body_buf.len); @memcpy(f.body_buf[0..len], buf[body_start .. body_start + len]); f.body_len = len; return f; }
fn findContentLength(headers: []const u8) ?usize { var it = std.mem.splitSequence(u8, headers, "\r\n"); while (it.next()) |line| { const colon = std.mem.indexOfScalar(u8, line, ':') orelse continue; if (!ascii.eqlIgnoreCase(std.mem.trim(u8, line[0..colon], " "), "content-length")) continue; return std.fmt.parseInt(usize, std.mem.trim(u8, line[colon + 1 ..], " "), 10) catch null; } return null; }
fn read(buf: []u8, key: []const u8, opts: *const Client.HandshakeOpts, compression: bool, stream: anytype, failure: *?Client.HandshakeFailure) !HandShakeReply { const timeout_ms = opts.timeout_ms; const deadline = milliTimestamp(stream.io) + timeout_ms; try stream.readTimeout(timeout_ms);
var pos: usize = 0; var line_start: usize = 0; var complete_response: u8 = 0; var server_compression: bool = false;
while (true) { // 0.16: using libc recv, WouldBlock indicates timeout const n = stream.read(buf[pos..]) catch |err| switch (err) { error.WouldBlock => return error.Timeout, else => return err, }; if (n == 0) { return error.ConnectionClosed; }
pos += n; while (std.mem.indexOfScalar(u8, buf[line_start..pos], '\r')) |relative_end| { if (relative_end == 0) { if (complete_response != 15) { return error.InvalidHandshakeResponse; } // TCP can split the terminating CRLF — if the trailing \n // hasn't arrived yet, pos == line_start + 1 and the over_read // subtraction below would underflow. break for more data. if (line_start + 2 > pos) break; const over_read = pos - (line_start + 2); std.mem.copyForwards(u8, buf[0..over_read], buf[line_start + 2 .. pos]); try stream.readTimeout(0); return .{ .over_read = over_read, .compression = server_compression, }; }
const line_end = line_start + relative_end; const line = buf[line_start..line_end];
// the next line starts where this line ends, skip over the \r\n. // TCP can split mid-CRLF — if \n hasn't arrived yet, break to // the outer read loop for more data. line_start = line_end + 2; if (line_start > pos) break;
if (complete_response == 0) { if (!ascii.startsWithIgnoreCase(line, "HTTP/1.1 101 ")) { // A well-formed non-101 response is a server-side // rejection with meaning: capture the status and a // bounded body so the caller can act on it (see // Client.handshake_failure), then fail as before. failure.* = captureRejection(line, buf, pos, stream); return error.InvalidHandshakeResponse; } complete_response |= 1; continue; }
for (line, 0..) |b, i| { // find the colon and lowercase the header while we're iterating if ('A' <= b and b <= 'Z') { line[i] = b + 32; continue; }
if (b != ':') { continue; }
switch (i) { 7 => if (std.mem.eql(u8, line[0..i], "upgrade")) { if (!ascii.eqlIgnoreCase(std.mem.trim(u8, line[i + 1 ..], &ascii.whitespace), "websocket")) { return error.InvalidUpgradeHeader; } complete_response |= 2; }, 10 => if (std.mem.eql(u8, line[0..i], "connection")) { if (!ascii.eqlIgnoreCase(std.mem.trim(u8, line[i + 1 ..], &ascii.whitespace), "upgrade")) { return error.InvalidConnectionHeader; } complete_response |= 4; }, 20 => if (std.mem.eql(u8, line[0..i], "sec-websocket-accept")) { var h: [20]u8 = undefined; { var hasher = std.crypto.hash.Sha1.init(.{}); hasher.update(key); hasher.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11"); hasher.final(&h); }
var encoded_buf: [28]u8 = undefined; const sec_hash = std.base64.standard.Encoder.encode(&encoded_buf, &h); const header_value = std.mem.trim(u8, line[i + 1 ..], &ascii.whitespace);
if (!std.mem.eql(u8, header_value, sec_hash)) { return error.InvalidWebsocketAcceptHeader; } complete_response |= 8; }, 24 => if (std.mem.eql(u8, line[0..i], "sec-websocket-extensions")) { if (try parseExtension(line[i + 1 ..])) |sc| { if (!compression) { // server is saying compression, but we didn't ask for it. return error.InvalidExtensionHeader; } if (!sc.client_no_context_takeover or !sc.server_no_context_takeover) { // as of Zig 0.15, we no longer support context takeover // We told the server this, it should have respected it. return error.InvalidExtensionHeader; }
server_compression = true; } }, else => {}, // some other header we don't care about } } }
if (milliTimestamp(stream.io) > deadline) { return error.Timeout; }
if (pos == buf.len) { return error.ResponseTooLarge; } } }
pub fn parseExtension(value: []const u8) !?ServerHandshake.Compression { var deflate = false; var client_max_bits: u8 = 15; var client_no_context_takeover = false; var server_no_context_takeover = false;
var it = std.mem.splitScalar(u8, value, ';'); while (it.next()) |param_| { const param = std.mem.trim(u8, param_, &ascii.whitespace); if (std.mem.eql(u8, param, "permessage-deflate")) { deflate = true; continue; } if (std.mem.eql(u8, param, "client_no_context_takeover")) { client_no_context_takeover = true; continue; } if (std.mem.eql(u8, param, "server_no_context_takeover")) { server_no_context_takeover = true; continue; } const client_max_window_bits = "client_max_window_bits="; if (std.mem.startsWith(u8, param, client_max_window_bits)) { client_max_bits = std.fmt.parseInt(u8, param[client_max_window_bits.len..], 10) catch { return error.InvalidCompressionServerMaxBits; }; } } if (deflate == false) { return null; }
if (client_max_bits != 15) { // We don't offer client window, so if the server asks for one, that's an error return error.InvalidExtensionHeader; }
return .{ .client_no_context_takeover = client_no_context_takeover, .server_no_context_takeover = server_no_context_takeover, }; }};
const t = @import("../t.zig");test "Client: handshake" { { // empty response var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); }
{ // invalid websocket response var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 200 OK\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); // a well-formed non-101 is a rejection: status captured, empty body try t.expectEqual(@as(u16, 200), client.handshake_failure.?.status); try t.expectEqual(@as(usize, 0), client.handshake_failure.?.body().len); }
{ // pre-upgrade rejection with a JSON error envelope (the jetstream // subscribeEvents contract: 400 CursorTooOld etc.) — the caller // needs the status and body to pick terminal vs transient handling var pair = t.SocketPair.init(.{}); defer pair.deinit(); const body = "{\"error\":\"CursorTooOld\",\"message\":\"below floor\"}"; var response_buf: [256]u8 = undefined; const response = std.fmt.bufPrint( &response_buf, "HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {d}\r\n\r\n{s}", .{ body.len, body }, ) catch unreachable; try pair.clientWriteAll(response);
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); const failure = client.handshake_failure.?; try t.expectEqual(@as(u16, 400), failure.status); try t.expectString(body, failure.body()); }
{ // missing upgrade header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); }
{ // wrong upgrade header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nUpgrade: nope\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidUpgradeHeader, client.handshake("/", .{})); }
{ // missing connection header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nUpgrade: websocket\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); }
{ // wrong connection header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nupgrade: WebSocket\r\nConnection: something\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidConnectionHeader, client.handshake("/", .{})); }
{ // missing Sec-Websocket-Accept header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nUpgrade: websocket\r\nConnection: upgrade\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidHandshakeResponse, client.handshake("/", .{})); }
{ // wrong Sec-Websocket-Accept header var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nupgrade: WebSocket\r\nConnection: UPGRADE\r\nSec-Websocket-Accept: hack\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try t.expectError(error.InvalidWebsocketAcceptHeader, client.handshake("/", .{})); }
{ // ok for successful var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nupgrade: WebSocket\r\nConnection: UPGRADE\r\nSec-Websocket-Accept: C/0nmHhBztSRGR1CwL6Tf4ZjwpY=\r\n\r\n");
var client = testClient(&pair); defer client.deinit(); try client.handshake("/", .{}); try t.expectEqual(0, client._reader.pos); }
{ // ok for successful, with overread var pair = t.SocketPair.init(.{}); defer pair.deinit(); try pair.clientWriteAll("HTTP/1.1 101 Switching Protocol\r\nupgrade: WebSocket\r\nConnection: UPGRADE\r\nSec-Websocket-Accept: C/0nmHhBztSRGR1CwL6Tf4ZjwpY=\r\n\r\nSome Random Data Which is Part Of the Next Message");
var client = testClient(&pair); defer client.deinit(); try client.handshake("/", .{}); try t.expectEqual(50, client._reader.pos); }}
test "Client: handshake with terminating CRLF split across reads" { // regression: when TCP delivers the final \r of the blank-line CRLF as the // last byte of a read and the \n arrives in a later read, the end-of-headers // branch computed `over_read = pos - (line_start + 2)` while pos == line_start // + 1, underflowing usize → integer-overflow panic. parser must wait for \n. const io = std.Options.debug_io;
// generateKey() is deterministic under test ({1..16}); this accept matches it. const bin_key = [16]u8{ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16 }; var encoded_key: [24]u8 = undefined; const key = std.base64.standard.Encoder.encode(&encoded_key, &bin_key);
const headers = "HTTP/1.1 101 Switching Protocol\r\nupgrade: WebSocket\r\nConnection: UPGRADE\r\nSec-Websocket-Accept: C/0nmHhBztSRGR1CwL6Tf4ZjwpY=\r\n"; const trailing = "Some Random Data Which is Part Of the Next Message";
const ChunkStream = struct { io: Io, chunks: []const []const u8, idx: usize = 0, fn read(self: *@This(), buf: []u8) !usize { if (self.idx >= self.chunks.len) return 0; const c = self.chunks[self.idx]; std.debug.assert(c.len <= buf.len); @memcpy(buf[0..c.len], c); self.idx += 1; return c.len; } fn readTimeout(self: *const @This(), ms: u32) !void { _ = self; _ = ms; } };
// first read ends on the terminating \r; the \n (and over-read) arrive next. var chunks = [_][]const u8{ headers ++ "\r", "\n" ++ trailing }; var stream = ChunkStream{ .io = io, .chunks = &chunks };
var buf: [4096]u8 = undefined; const opts = Client.HandshakeOpts{}; var failure: ?Client.HandshakeFailure = null; const res = try HandShakeReply.read(&buf, key, &opts, false, &stream, &failure); try t.expectEqual(trailing.len, res.over_read); try t.expectSlice(u8, trailing, buf[0..res.over_read]);}
test "Client: setting a timeout on a dead socket does not abort the process" { // regression: std.posix.setsockopt maps BADF/NOTSOCK/INVAL/FAULT to // `unreachable`. Those are reachable on a *connection* socket — the peer // resets, or another thread closes the fd, between connect and the timeout // call. `unreachable` cannot be caught, so an upstream dropping the // connection aborted the whole process from inside the handshake instead of // returning an error the caller could reconnect on. const io = std.Options.debug_io; const stream = try testConnectedStream(io, "127.0.0.1", 9292); var dead = Stream.init(io, stream, null); // Close underneath the Stream: every later setsockopt sees EBADF, which is // precisely the race the stdlib declares impossible. stream.close(io);
// Before the fix each of these panicked rather than returning. try dead.readTimeout(5_000); try dead.writeTimeout(5_000); try dead.readTimeout(0);}
test "Client: writeFrame after close returns error.Closed, never touches the stream" { // regression: close() released the write lock between sending the close // frame and stream.close(), so a concurrent writer queued on the lock // (zlay's keepalive pinger) resumed into a freed TLS client — a GPF in // production on 2026-08-27. writers must observe _closed under the lock. var pair = t.SocketPair.init(.{}); defer pair.deinit();
var client = testClient(&pair); defer client.deinit();
var payload = [_]u8{ 'h', 'i' }; try client.writeFrame(.text, &payload);
try client.close(.{ .code = 4000 }); try t.expectError(error.Closed, client.writePing(&payload)); try t.expectError(error.Closed, client.writeFrame(.text, &payload)); // idempotent close still fine after the lock-scoped teardown try client.close(.{});}
test "Client: write/read" { const io = std.Options.debug_io; const stream = try testConnectedStream(io, "127.0.0.1", 9292); var client = try Client.initWithStream(io, t.allocator, stream, .{ .port = 9292, .host = "127.0.0.1", }); defer client.deinit();
try client.handshake("/", .{ .timeout_ms = 1000, });
var buf = [_]u8{ 'o', 'v', 'e', 'r' }; try client.write(&buf); try client.readTimeout(1000);
const message = (try client.read()) orelse unreachable; try t.expectEqual(.text, message.type); try t.expectString("9000", message.data);
client.close(.{}) catch unreachable;}
test "Client: close with code" { const io = std.Options.debug_io; const stream = try testConnectedStream(io, "127.0.0.1", 9292); var client = try Client.initWithStream(io, t.allocator, stream, .{ .port = 9292, .host = "127.0.0.1", }); defer client.deinit();
try client.handshake("/", .{ .timeout_ms = 1000, });
client.close(.{ .code = 4002 }) catch unreachable;}
test "Client: with code and reason" { const io = std.Options.debug_io; const stream = try testConnectedStream(io, "127.0.0.1", 9292); var client = try Client.initWithStream(io, t.allocator, stream, .{ .port = 9292, .host = "127.0.0.1", }); defer client.deinit();
try client.handshake("/", .{ .timeout_ms = 1000, });
client.close(.{ .code = 4002, .reason = "goodbye" }) catch unreachable;}
test "Client: Handler" { var h = try ClientHandler.init(t.allocator); defer h.deinit();
var buf: [6]u8 = undefined; { @memcpy(buf[0..3], "dyn"); try h.client.write(buf[0..3]); }
{ @memcpy(buf[0..4], "ping"); try h.client.write(buf[0..4]); }
{ @memcpy(buf[0..4], "pong"); try h.client.write(buf[0..4]); }
{ @memcpy(buf[0..6], "close1"); try h.client.write(buf[0..6]); }
try h.client.readLoop(&h);
// if pong is true then ping and message have to be true // because each asserts the previous try t.expectEqual(true, h.pong); try t.expectEqual(true, h.closed);}
fn testClient(pair: *t.SocketPair) Client { pair.server_taken = true; const stream = pair.server; const io = std.Options.debug_io; const bp = t.allocator.create(buffer.Provider) catch unreachable; bp.* = buffer.Provider.init(t.allocator, .{ .count = 0, .size = 0, .max = 4096 }) catch unreachable;
const reader_buf = bp.allocator.alloc(u8, 1024) catch unreachable;
return .{ .io = io, ._closed = false, ._own_bp = true, ._mask_fn = generateMask, ._compression_opts = null, .stream = .{ .io = io, .stream = stream }, ._reader = Reader.init(reader_buf, bp, null), };}
fn testConnectedStream(io: Io, host: []const u8, port: u16) !net.Stream { const addr = try net.IpAddress.parse(host, port); return net.IpAddress.connect(&addr, io, .{ .mode = .stream });}
const ClientHandler = struct { ping: bool = false, pong: bool = false, closed: bool = false, message: bool = false, client: Client,
fn init(allocator: Allocator) !ClientHandler { const io = std.Options.debug_io; const stream = try testConnectedStream(io, "127.0.0.1", 9292); var client = try Client.initWithStream(io, allocator, stream, .{ .port = 9292, .host = "127.0.0.1", }); errdefer client.deinit();
try client.handshake("/", .{ .timeout_ms = 1000, });
return .{ .client = client, }; }
fn deinit(self: *ClientHandler) void { self.client.deinit(); }
pub fn serverMessage(self: *ClientHandler, data: []u8, tpe: proto.Message.TextType) !void { try t.expectEqual(.text, tpe); try t.expectString("over 9000!", data); self.message = true; }
pub fn serverPing(self: *ClientHandler, data: []u8) !void { try t.expectEqual(true, self.message); try t.expectString("a-ping", data); self.ping = true; }
pub fn serverPong(self: *ClientHandler, data: []u8) !void { try t.expectEqual(true, self.ping); try t.expectString("a-pong", data); self.pong = true; }
pub fn close(self: *ClientHandler) void { self.client.close(.{}) catch unreachable; self.closed = true; }};