// Copyright (C) 2023-2024 Lightpanda (Selecy SAS) // // Francis Bouvier // Pierre Tachoire // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as // published by the Free Software Foundation, either version 3 of the // License, or (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . const std = @import("std"); const builtin = @import("builtin"); const net = std.net; const posix = std.posix; const Allocator = std.mem.Allocator; const ArenaAllocator = std.heap.ArenaAllocator; const log = @import("log.zig"); const App = @import("app.zig").App; const CDP = @import("cdp/cdp.zig").CDP; const TimeoutCheck = std.time.ns_per_ms * 100; const MAX_HTTP_REQUEST_SIZE = 4096; // max message size // +14 for max websocket payload overhead // +140 for the max control packet that might be interleaved in a message const MAX_MESSAGE_SIZE = 512 * 1024 + 14 + 140; pub const Server = struct { app: *App, allocator: Allocator, client: ?posix.socket_t, listener: ?posix.socket_t, json_version_response: []const u8, pub fn init(app: *App, address: net.Address) !Server { const allocator = app.allocator; const json_version_response = try buildJSONVersionResponse(allocator, address); errdefer allocator.free(json_version_response); return .{ .app = app, .client = null, .listener = null, .allocator = allocator, .json_version_response = json_version_response, }; } pub fn deinit(self: *Server) void { self.allocator.free(self.json_version_response); if (self.listener) |listener| { posix.close(listener); } } pub fn run(self: *Server, address: net.Address, timeout_ms: i32) !void { const flags = posix.SOCK.STREAM | posix.SOCK.CLOEXEC; const listener = try posix.socket(address.any.family, flags, posix.IPPROTO.TCP); self.listener = listener; try posix.setsockopt(listener, posix.SOL.SOCKET, posix.SO.REUSEADDR, &std.mem.toBytes(@as(c_int, 1))); if (@hasDecl(posix.TCP, "NODELAY")) { try posix.setsockopt(listener, posix.IPPROTO.TCP, posix.TCP.NODELAY, &std.mem.toBytes(@as(c_int, 1))); } try posix.bind(listener, &address.any, address.getOsSockLen()); try posix.listen(listener, 1); log.info(.app, "server running", .{ .address = address }); while (true) { const socket = posix.accept(listener, null, null, posix.SOCK.NONBLOCK) catch |err| { log.err(.app, "CDP accept", .{ .err = err }); std.time.sleep(std.time.ns_per_s); continue; }; self.client = socket; defer if (self.client) |s| { posix.close(s); self.client = null; }; if (log.enabled(.app, .info)) { var client_address: std.net.Address = undefined; var socklen: posix.socklen_t = @sizeOf(net.Address); try std.posix.getsockname(socket, &client_address.any, &socklen); log.info(.app, "client connected", .{ .ip = client_address }); } self.readLoop(socket, timeout_ms) catch |err| { log.err(.app, "CDP client loop", .{ .err = err }); }; } } fn readLoop(self: *Server, socket: posix.socket_t, timeout_ms: i32) !void { // This shouldn't be necessary, but the Client is HUGE (> 512KB) because // it has a large read buffer. I don't know why, but v8 crashes if this // is on the stack (and I assume it's related to its size). const client = try self.allocator.create(Client); defer self.allocator.destroy(client); client.* = try Client.init(socket, self); defer client.deinit(); var last_message = timestamp(); var http = &self.app.http; while (true) { if (http.poll(20, socket)) { const n = posix.read(socket, client.readBuf()) catch |err| { log.warn(.app, "CDP read", .{ .err = err }); return; }; if (n == 0) { log.info(.app, "CDP disconnect", .{}); return; } const more = client.processData(n) catch false; if (!more) { return; } last_message = timestamp(); } else if (timestamp() - last_message > timeout_ms) { log.info(.app, "CDP timeout", .{}); return; } // We have 3 types of "events": // - Incoming CDP messages // - Network events from the browser // - Timeouts from the browser // The call to http.poll above handles the first two (which is why // we pass the client socket to it). But browser timeouts aren't // hooked into that. So we need to go to the browser page (if there // is one), and ask it to process any pending events. That action // doesn't starve #2 (Network events from the browser), because // page.wait() handles that too. But it does starve #1 (Incoming CDP // messages). The good news is that, while the Page is mostly // unaware of CDP, it will only block if it actually has something to // do AND it knows if we're waiting on an intercept request, and will // eagerly return control here in those cases. if (client.mode == .cdp) { client.mode.cdp.pageWait(); } } } }; pub const Client = struct { // The client is initially serving HTTP requests but, under normal circumstances // should eventually be upgraded to a websocket connections mode: union(enum) { http: void, cdp: CDP, }, server: *Server, reader: Reader(true), socket: posix.socket_t, socket_flags: usize, send_arena: ArenaAllocator, const EMPTY_PONG = [_]u8{ 138, 0 }; // CLOSE, 2 length, code const CLOSE_NORMAL = [_]u8{ 136, 2, 3, 232 }; // code: 1000 const CLOSE_TOO_BIG = [_]u8{ 136, 2, 3, 241 }; // 1009 const CLOSE_PROTOCOL_ERROR = [_]u8{ 136, 2, 3, 234 }; //code: 1002 // "private-use" close codes must be from 4000-49999 const CLOSE_TIMEOUT = [_]u8{ 136, 2, 15, 160 }; // code: 4000 fn init(socket: posix.socket_t, server: *Server) !Client { const socket_flags = try posix.fcntl(socket, posix.F.GETFL, 0); const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true })); // we expect the socket to come to us as nonblocking std.debug.assert(socket_flags & nonblocking == nonblocking); var reader = try Reader(true).init(server.allocator); errdefer reader.deinit(); return .{ .socket = socket, .server = server, .reader = reader, .mode = .{ .http = {} }, .socket_flags = socket_flags, .send_arena = ArenaAllocator.init(server.allocator), }; } fn deinit(self: *Client) void { switch (self.mode) { .cdp => |*cdp| cdp.deinit(), .http => {}, } self.reader.deinit(); self.send_arena.deinit(); } fn readBuf(self: *Client) []u8 { return self.reader.readBuf(); } fn processData(self: *Client, len: usize) !bool { self.reader.len += len; switch (self.mode) { .cdp => |*cdp| return self.processWebsocketMessage(cdp), .http => return self.processHTTPRequest(), } } fn processHTTPRequest(self: *Client) !bool { std.debug.assert(self.reader.pos == 0); const request = self.reader.buf[0..self.reader.len]; if (request.len > MAX_HTTP_REQUEST_SIZE) { self.writeHTTPErrorResponse(413, "Request too large"); return error.RequestTooLarge; } // we're only expecting [body-less] GET requests. if (std.mem.endsWith(u8, request, "\r\n\r\n") == false) { // we need more data, put any more data here return true; } // the next incoming data can go to the front of our buffer defer self.reader.len = 0; return self.handleHTTPRequest(request) catch |err| { switch (err) { error.NotFound => self.writeHTTPErrorResponse(404, "Not found"), error.InvalidRequest => self.writeHTTPErrorResponse(400, "Invalid request"), error.InvalidProtocol => self.writeHTTPErrorResponse(400, "Invalid HTTP protocol"), error.MissingHeaders => self.writeHTTPErrorResponse(400, "Missing required header"), error.InvalidUpgradeHeader => self.writeHTTPErrorResponse(400, "Unsupported upgrade type"), error.InvalidVersionHeader => self.writeHTTPErrorResponse(400, "Invalid websocket version"), error.InvalidConnectionHeader => self.writeHTTPErrorResponse(400, "Invalid connection header"), else => { log.err(.app, "server 500", .{ .err = err, .req = request[0..@min(100, request.len)] }); self.writeHTTPErrorResponse(500, "Internal Server Error"); }, } return err; }; } fn handleHTTPRequest(self: *Client, request: []u8) !bool { if (request.len < 18) { // 18 is [generously] the smallest acceptable HTTP request return error.InvalidRequest; } if (std.mem.eql(u8, request[0..4], "GET ") == false) { return error.NotFound; } const url_end = std.mem.indexOfScalarPos(u8, request, 4, ' ') orelse { return error.InvalidRequest; }; const url = request[4..url_end]; if (std.mem.eql(u8, url, "/")) { try self.upgradeConnection(request); return true; } if (std.mem.eql(u8, url, "/json/version")) { try self.send(self.server.json_version_response); // Chromedp (a Go driver) does an http request to /json/version // then to / (websocket upgrade) using a different connection. // Since we only allow 1 connection at a time, the 2nd one (the // websocket upgrade) blocks until the first one times out. // We can avoid that by closing the connection. json_version_response // has a Connection: Close header too. try posix.shutdown(self.socket, .recv); return false; } return error.NotFound; } fn upgradeConnection(self: *Client, request: []u8) !void { // our caller already confirmed that we have a trailing \r\n\r\n const request_line_end = std.mem.indexOfScalar(u8, request, '\r') orelse unreachable; const request_line = request[0..request_line_end]; if (!std.ascii.endsWithIgnoreCase(request_line, "http/1.1")) { return error.InvalidProtocol; } // we need to extract the sec-websocket-key value var key: []const u8 = ""; // we need to make sure that we got all the necessary headers + values var required_headers: u8 = 0; // can't std.mem.split because it forces the iterated value to be const // (we could @constCast...) var buf = request[request_line_end + 2 ..]; while (buf.len > 4) { const index = std.mem.indexOfScalar(u8, buf, '\r') orelse unreachable; const separator = std.mem.indexOfScalar(u8, buf[0..index], ':') orelse return error.InvalidRequest; const name = std.mem.trim(u8, toLower(buf[0..separator]), &std.ascii.whitespace); const value = std.mem.trim(u8, buf[(separator + 1)..index], &std.ascii.whitespace); if (std.mem.eql(u8, name, "upgrade")) { if (!std.ascii.eqlIgnoreCase("websocket", value)) { return error.InvalidUpgradeHeader; } required_headers |= 1; } else if (std.mem.eql(u8, name, "sec-websocket-version")) { if (value.len != 2 or value[0] != '1' or value[1] != '3') { return error.InvalidVersionHeader; } required_headers |= 2; } else if (std.mem.eql(u8, name, "connection")) { // find if connection header has upgrade in it, example header: // Connection: keep-alive, Upgrade if (std.ascii.indexOfIgnoreCase(value, "upgrade") == null) { return error.InvalidConnectionHeader; } required_headers |= 4; } else if (std.mem.eql(u8, name, "sec-websocket-key")) { key = value; required_headers |= 8; } const next = index + 2; buf = buf[next..]; } if (required_headers != 15) { return error.MissingHeaders; } // our caller has already made sure this request ended in \r\n\r\n // so it isn't something we need to check again const allocator = self.send_arena.allocator(); const response = blk: { // Response to an ugprade request is always this, with // the Sec-Websocket-Accept value a spacial sha1 hash of the // request "sec-websocket-version" and a magic value. const template = "HTTP/1.1 101 Switching Protocols\r\n" ++ "Upgrade: websocket\r\n" ++ "Connection: upgrade\r\n" ++ "Sec-Websocket-Accept: 0000000000000000000000000000\r\n\r\n"; // The response will be sent via the IO Loop and thus has to have its // own lifetime. const res = try allocator.dupe(u8, template); // magic response const key_pos = res.len - 32; var h: [20]u8 = undefined; var hasher = std.crypto.hash.Sha1.init(.{}); hasher.update(key); // websocket spec always used this value hasher.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11"); hasher.final(&h); _ = std.base64.standard.Encoder.encode(res[key_pos .. key_pos + 28], h[0..]); break :blk res; }; self.mode = .{ .cdp = try CDP.init(self.server.app, self) }; return self.send(response); } fn writeHTTPErrorResponse(self: *Client, comptime status: u16, comptime body: []const u8) void { const response = std.fmt.comptimePrint( "HTTP/1.1 {d} \r\nConnection: Close\r\nContent-Length: {d}\r\n\r\n{s}", .{ status, body.len, body }, ); // we're going to close this connection anyways, swallowing any // error seems safe self.send(response) catch {}; } fn processWebsocketMessage(self: *Client, cdp: *CDP) !bool { var reader = &self.reader; while (true) { const msg = reader.next() catch |err| { switch (err) { error.TooLarge => self.send(&CLOSE_TOO_BIG) catch {}, error.NotMasked => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.ReservedFlags => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.InvalidMessageType => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.ControlTooLarge => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.InvalidContinuation => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.NestedFragementation => self.send(&CLOSE_PROTOCOL_ERROR) catch {}, error.OutOfMemory => {}, // don't borther trying to send an error in this case } return err; } orelse break; switch (msg.type) { .pong => {}, .ping => try self.sendPong(msg.data), .close => { self.send(&CLOSE_NORMAL) catch {}; return false; }, .text, .binary => if (cdp.handleMessage(msg.data) == false) { return false; }, } if (msg.cleanup_fragment) { reader.cleanup(); } } // We might have read part of the next message. Our reader potentially // has to move data around in its buffer to make space. reader.compact(); return true; } fn sendPong(self: *Client, data: []const u8) !void { if (data.len == 0) { return self.send(&EMPTY_PONG); } var header_buf: [10]u8 = undefined; const header = websocketHeader(&header_buf, .pong, data.len); const allocator = self.send_arena.allocator(); var framed = try allocator.alloc(u8, header.len + data.len); @memcpy(framed[0..header.len], header); @memcpy(framed[header.len..], data); return self.send(framed); } // called by CDP // Websocket frames have a variable lenght header. For server-client, // it could be anywhere from 2 to 10 bytes. Our IO.Loop doesn't have // writev, so we need to get creative. We'll JSON serialize to a // buffer, where the first 10 bytes are reserved. We can then backfill // the header and send the slice. pub fn sendJSON(self: *Client, message: anytype, opts: std.json.StringifyOptions) !void { const allocator = self.send_arena.allocator(); var buf: std.ArrayListUnmanaged(u8) = .{}; try buf.ensureTotalCapacity(allocator, 512); // reserve space for the maximum possible header buf.appendSliceAssumeCapacity(&.{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }); try std.json.stringify(message, opts, buf.writer(allocator)); const framed = fillWebsocketHeader(buf); return self.send(framed); } pub fn sendJSONRaw( self: *Client, buf: std.ArrayListUnmanaged(u8), ) !void { // Dangerous API!. We assume the caller has reserved the first 10 // bytes in `buf`. const framed = fillWebsocketHeader(buf); return self.send(framed); } fn send(self: *Client, data: []const u8) !void { var pos: usize = 0; var changed_to_blocking: bool = false; defer _ = self.send_arena.reset(.{ .retain_with_limit = 1024 * 32 }); defer if (changed_to_blocking) { // We had to change our socket to blocking me to get our write out // We need to change it back to non-blocking. _ = posix.fcntl(self.socket, posix.F.SETFL, self.socket_flags) catch |err| { log.err(.app, "CDP restore nonblocking", .{ .err = err }); }; }; LOOP: while (pos < data.len) { const written = posix.write(self.socket, data[pos..]) catch |err| switch (err) { error.WouldBlock => { // self.socket is nonblocking, because we don't want to block // reads. But our life is a lot easier if we block writes, // largely, because we don't have to maintain a queue of pending // writes (which would each need their own allocations). So // if we get a WouldBlock error, we'll switch the socket to // blocking and switch it back to non-blocking after the write // is complete. Doesn't seem particularly efficiently, but // this should virtually never happen. std.debug.assert(changed_to_blocking == false); log.debug(.app, "CDP write would block", .{}); changed_to_blocking = true; _ = try posix.fcntl(self.socket, posix.F.SETFL, self.socket_flags & ~@as(u32, @bitCast(posix.O{ .NONBLOCK = true }))); continue :LOOP; }, else => return err, }; if (written == 0) { return error.Closed; } pos += written; } } }; // WebSocket message reader. Given websocket message, acts as an iterator that // can return zero or more Messages. When next returns null, any incomplete // message will remain in reader.data fn Reader(comptime EXPECT_MASK: bool) type { return struct { allocator: Allocator, // position in buf of the start of the next message pos: usize = 0, // position in buf up until where we have valid data // (any new reads must be placed after this) len: usize = 0, // we add 140 to allow 1 control message (ping/pong/close) to be // fragmented into a normal message. buf: []u8, fragments: ?Fragments = null, const Self = @This(); fn init(allocator: Allocator) !Self { const buf = try allocator.alloc(u8, 16 * 1024); return .{ .buf = buf, .allocator = allocator, }; } fn deinit(self: *Self) void { self.cleanup(); self.allocator.free(self.buf); } fn cleanup(self: *Self) void { if (self.fragments) |*f| { f.message.deinit(self.allocator); self.fragments = null; } } fn readBuf(self: *Self) []u8 { // We might have read a partial http or websocket message. // Subsequent reads must read from where we left off. return self.buf[self.len..]; } fn next(self: *Self) !?Message { LOOP: while (true) { var buf = self.buf[self.pos..self.len]; const length_of_len, const message_len = extractLengths(buf) orelse { // we don't have enough bytes return null; }; const byte1 = buf[0]; if (byte1 & 112 != 0) { return error.ReservedFlags; } if (comptime EXPECT_MASK) { if (buf[1] & 128 != 128) { // client -> server messages _must_ be masked return error.NotMasked; } } else if (buf[1] & 128 != 0) { // server -> client are never masked return error.Masked; } var is_control = false; var is_continuation = false; var message_type: Message.Type = undefined; switch (byte1 & 15) { 0 => is_continuation = true, 1 => message_type = .text, 2 => message_type = .binary, 8 => { is_control = true; message_type = .close; }, 9 => { is_control = true; message_type = .ping; }, 10 => { is_control = true; message_type = .pong; }, else => return error.InvalidMessageType, } if (is_control) { if (message_len > 125) { return error.ControlTooLarge; } } else if (message_len > MAX_MESSAGE_SIZE) { return error.TooLarge; } else if (message_len > self.buf.len) { const len = self.buf.len; self.buf = try growBuffer(self.allocator, self.buf, message_len); buf = self.buf[0..len]; // we need more data return null; } else if (buf.len < message_len) { // we need more data return null; } // prefix + length_of_len + mask const header_len = 2 + length_of_len + if (comptime EXPECT_MASK) 4 else 0; const payload = buf[header_len..message_len]; if (comptime EXPECT_MASK) { mask(buf[header_len - 4 .. header_len], payload); } // whatever happens after this, we know where the next message starts self.pos += message_len; const fin = byte1 & 128 == 128; if (is_continuation) { const fragments = &(self.fragments orelse return error.InvalidContinuation); if (fragments.message.items.len + message_len > MAX_MESSAGE_SIZE) { return error.TooLarge; } try fragments.message.appendSlice(self.allocator, payload); if (fin == false) { // maybe we have more parts of the message waiting continue :LOOP; } // this continuation is done! return .{ .type = fragments.type, .data = fragments.message.items, .cleanup_fragment = true, }; } const can_be_fragmented = message_type == .text or message_type == .binary; if (self.fragments != null and can_be_fragmented) { // if this isn't a continuation, then we can't have fragments return error.NestedFragementation; } if (fin == false) { if (can_be_fragmented == false) { return error.InvalidContinuation; } // not continuation, and not fin. It has to be the first message // in a fragmented message. var fragments = Fragments{ .message = .{}, .type = message_type }; try fragments.message.appendSlice(self.allocator, payload); self.fragments = fragments; continue :LOOP; } return .{ .data = payload, .type = message_type, .cleanup_fragment = false, }; } } fn extractLengths(buf: []const u8) ?struct { usize, usize } { if (buf.len < 2) { return null; } const length_of_len: usize = switch (buf[1] & 127) { 126 => 2, 127 => 8, else => 0, }; if (buf.len < length_of_len + 2) { // we definitely don't have enough buf yet return null; } const message_len = switch (length_of_len) { 2 => @as(u16, @intCast(buf[3])) | @as(u16, @intCast(buf[2])) << 8, 8 => @as(u64, @intCast(buf[9])) | @as(u64, @intCast(buf[8])) << 8 | @as(u64, @intCast(buf[7])) << 16 | @as(u64, @intCast(buf[6])) << 24 | @as(u64, @intCast(buf[5])) << 32 | @as(u64, @intCast(buf[4])) << 40 | @as(u64, @intCast(buf[3])) << 48 | @as(u64, @intCast(buf[2])) << 56, else => buf[1] & 127, } + length_of_len + 2 + if (comptime EXPECT_MASK) 4 else 0; // +2 for header prefix, +4 for mask; return .{ length_of_len, message_len }; } // This is called after we've processed complete websocket messages (this // only applies to websocket messages). // There are three cases: // 1 - We don't have any incomplete data (for a subsequent message) in buf. // This is the easier to handle, we can set pos & len to 0. // 2 - We have part of the next message, but we know it'll fit in the // remaining buf. We don't need to do anything // 3 - We have part of the next message, but either it won't fight into the // remaining buffer, or we don't know (because we don't have enough // of the header to tell the length). We need to "compact" the buffer fn compact(self: *Self) void { const pos = self.pos; const len = self.len; std.debug.assert(pos <= len); // how many (if any) partial bytes do we have const partial_bytes = len - pos; if (partial_bytes == 0) { // We have no partial bytes. Setting these to 0 ensures that we // get the best utilization of our buffer self.pos = 0; self.len = 0; return; } const partial = self.buf[pos..len]; // If we have enough bytes of the next message to tell its length // we'll be able to figure out whether we need to do anything or not. if (extractLengths(partial)) |length_meta| { const next_message_len = length_meta.@"1"; // if this isn't true, then we have a full message and it // should have been processed. std.debug.assert(next_message_len > partial_bytes); const missing_bytes = next_message_len - partial_bytes; const free_space = self.buf.len - len; if (missing_bytes < free_space) { // we have enough space in our buffer, as is, return; } } // We're here because we either don't have enough bytes of the next // message, or we know that it won't fit in our buffer as-is. std.mem.copyForwards(u8, self.buf, partial); self.pos = 0; self.len = partial_bytes; } }; } fn growBuffer(allocator: Allocator, buf: []u8, required_capacity: usize) ![]u8 { // from std.ArrayList var new_capacity = buf.len; while (true) { new_capacity +|= new_capacity / 2 + 8; if (new_capacity >= required_capacity) break; } log.debug(.app, "CDP buffer growth", .{ .from = buf.len, .to = new_capacity }); if (allocator.resize(buf, new_capacity)) { return buf.ptr[0..new_capacity]; } const new_buffer = try allocator.alloc(u8, new_capacity); @memcpy(new_buffer[0..buf.len], buf); allocator.free(buf); return new_buffer; } const Fragments = struct { type: Message.Type, message: std.ArrayListUnmanaged(u8), }; const Message = struct { type: Type, data: []const u8, cleanup_fragment: bool, const Type = enum { text, binary, close, ping, pong, }; }; // These are the only websocket types that we're currently sending const OpCode = enum(u8) { text = 128 | 1, close = 128 | 8, pong = 128 | 10, }; fn fillWebsocketHeader(buf: std.ArrayListUnmanaged(u8)) []const u8 { // can't use buf[0..10] here, because the header length // is variable. If it's just 2 bytes, for example, we need the // framed message to be: // h1, h2, data // If we use buf[0..10], we'd get: // h1, h2, 0, 0, 0, 0, 0, 0, 0, 0, data var header_buf: [10]u8 = undefined; // -10 because we reserved 10 bytes for the header above const header = websocketHeader(&header_buf, .text, buf.items.len - 10); const start = 10 - header.len; const message = buf.items; @memcpy(message[start..10], header); return message[start..]; } // makes the assumption that our caller reserved the first // 10 bytes for the header fn websocketHeader(buf: []u8, op_code: OpCode, payload_len: usize) []const u8 { std.debug.assert(buf.len == 10); const len = payload_len; buf[0] = 128 | @intFromEnum(op_code); // fin | opcode if (len <= 125) { buf[1] = @intCast(len); return buf[0..2]; } if (len < 65536) { buf[1] = 126; buf[2] = @intCast((len >> 8) & 0xFF); buf[3] = @intCast(len & 0xFF); return buf[0..4]; } buf[1] = 127; buf[2] = 0; buf[3] = 0; buf[4] = 0; buf[5] = 0; buf[6] = @intCast((len >> 24) & 0xFF); buf[7] = @intCast((len >> 16) & 0xFF); buf[8] = @intCast((len >> 8) & 0xFF); buf[9] = @intCast(len & 0xFF); return buf[0..10]; } // Utils // -------- fn buildJSONVersionResponse( allocator: Allocator, address: net.Address, ) ![]const u8 { const body_format = "{{\"webSocketDebuggerUrl\": \"ws://{}/\"}}"; const body_len = std.fmt.count(body_format, .{address}); // We send a Connection: Close (and actually close the connection) // because chromedp (Go driver) sends a request to /json/version and then // does an upgrade request, on a different connection. Since we only allow // 1 connection at a time, the upgrade connection doesn't proceed until we // timeout the /json/version. So, instead of waiting for that, we just // always close HTTP requests. const response_format = "HTTP/1.1 200 OK\r\n" ++ "Content-Length: {d}\r\n" ++ "Connection: Close\r\n" ++ "Content-Type: application/json; charset=UTF-8\r\n\r\n" ++ body_format; return try std.fmt.allocPrint(allocator, response_format, .{ body_len, address }); } fn timestamp() u32 { const ts = std.posix.clock_gettime(std.posix.CLOCK.MONOTONIC) catch unreachable; return @intCast(ts.sec); } // In-place string lowercase fn toLower(str: []u8) []u8 { for (str, 0..) |c, i| { str[i] = std.ascii.toLower(c); } return str; } // Zig is in a weird backend transition right now. Need to determine if // SIMD is even available. const backend_supports_vectors = switch (builtin.zig_backend) { .stage2_llvm, .stage2_c => true, else => false, }; // Websocket messages from client->server are masked using a 4 byte XOR mask fn mask(m: []const u8, payload: []u8) void { var data = payload; if (!comptime backend_supports_vectors) return simpleMask(m, data); const vector_size = std.simd.suggestVectorLength(u8) orelse @sizeOf(usize); if (data.len >= vector_size) { const mask_vector = std.simd.repeat(vector_size, @as(@Vector(4, u8), m[0..4].*)); while (data.len >= vector_size) { const slice = data[0..vector_size]; const masked_data_slice: @Vector(vector_size, u8) = slice.*; slice.* = masked_data_slice ^ mask_vector; data = data[vector_size..]; } } simpleMask(m, data); } // Used when SIMD isn't available, or for any remaining part of the message // which is too small to effectively use SIMD. fn simpleMask(m: []const u8, payload: []u8) void { for (payload, 0..) |b, i| { payload[i] = b ^ m[i & 3]; } } const testing = std.testing; test "server: buildJSONVersionResponse" { const address = try net.Address.parseIp4("127.0.0.1", 9001); const res = try buildJSONVersionResponse(testing.allocator, address); defer testing.allocator.free(res); try testing.expectEqualStrings("HTTP/1.1 200 OK\r\n" ++ "Content-Length: 48\r\n" ++ "Connection: Close\r\n" ++ "Content-Type: application/json; charset=UTF-8\r\n\r\n" ++ "{\"webSocketDebuggerUrl\": \"ws://127.0.0.1:9001/\"}", res); } test "Client: http invalid request" { var c = try createTestClient(); defer c.deinit(); const res = try c.httpRequest("GET /over/9000 HTTP/1.1\r\n" ++ "Header: " ++ ("a" ** 4100) ++ "\r\n\r\n"); try testing.expectEqualStrings("HTTP/1.1 413 \r\n" ++ "Connection: Close\r\n" ++ "Content-Length: 17\r\n\r\n" ++ "Request too large", res); } test "Client: http invalid handshake" { try assertHTTPError( 400, "Invalid request", "\r\n\r\n", ); try assertHTTPError( 404, "Not found", "GET /over/9000 HTTP/1.1\r\n\r\n", ); try assertHTTPError( 404, "Not found", "POST / HTTP/1.1\r\n\r\n", ); try assertHTTPError( 400, "Invalid HTTP protocol", "GET / HTTP/1.0\r\n\r\n", ); try assertHTTPError( 400, "Missing required header", "GET / HTTP/1.1\r\n\r\n", ); try assertHTTPError( 400, "Missing required header", "GET / HTTP/1.1\r\nConnection: upgrade\r\n\r\n", ); try assertHTTPError( 400, "Missing required header", "GET / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: websocket\r\n\r\n", ); try assertHTTPError( 400, "Missing required header", "GET / HTTP/1.1\r\nConnection: upgrade\r\nUpgrade: websocket\r\nsec-websocket-version:13\r\n\r\n", ); } test "Client: http valid handshake" { var c = try createTestClient(); defer c.deinit(); const request = "GET / HTTP/1.1\r\n" ++ "Connection: upgrade\r\n" ++ "Upgrade: websocket\r\n" ++ "sec-websocket-version:13\r\n" ++ "sec-websocket-key: this is my key\r\n" ++ "Custom: Header-Value\r\n\r\n"; const res = try c.httpRequest(request); try testing.expectEqualStrings("HTTP/1.1 101 Switching Protocols\r\n" ++ "Upgrade: websocket\r\n" ++ "Connection: upgrade\r\n" ++ "Sec-Websocket-Accept: flzHu2DevQ2dSCSVqKSii5e9C2o=\r\n\r\n", res); } test "Client: read invalid websocket message" { // 131 = 128 (fin) | 3 where 3 isn't a valid type try assertWebSocketError( 1002, &.{ 131, 128, 'm', 'a', 's', 'k' }, ); for ([_]u8{ 16, 32, 64 }) |rsv| { // none of the reserve flags should be set try assertWebSocketError( 1002, &.{ rsv, 128, 'm', 'a', 's', 'k' }, ); // as a bitmask try assertWebSocketError( 1002, &.{ rsv + 4, 128, 'm', 'a', 's', 'k' }, ); } // client->server messages must be masked try assertWebSocketError( 1002, &.{ 129, 1, 'a' }, ); // control types (ping/ping/close) can't be > 125 bytes for ([_]u8{ 136, 137, 138 }) |op| { try assertWebSocketError( 1002, &.{ op, 254, 1, 1 }, ); } // length of message is 0000 0810, i.e: 1024 * 512 + 265 try assertWebSocketError(1009, &.{ 129, 255, 0, 0, 0, 0, 0, 8, 1, 0, 'm', 'a', 's', 'k' }); // continuation type message must come after a normal message // even when not a fin frame try assertWebSocketError( 1002, &.{ 0, 129, 'm', 'a', 's', 'k', 'd' }, ); // continuation type message must come after a normal message // even as a fin frame try assertWebSocketError( 1002, &.{ 128, 129, 'm', 'a', 's', 'k', 'd' }, ); // text (non-fin) - text (non-fin) try assertWebSocketError( 1002, &.{ 1, 129, 'm', 'a', 's', 'k', 'd', 1, 128, 'k', 's', 'a', 'm' }, ); // text (non-fin) - text (fin) should always been continuation after non-fin try assertWebSocketError( 1002, &.{ 1, 129, 'm', 'a', 's', 'k', 'd', 129, 128, 'k', 's', 'a', 'm' }, ); // close must be fin try assertWebSocketError( 1002, &.{ 8, 129, 'm', 'a', 's', 'k', 'd', }, ); // ping must be fin try assertWebSocketError( 1002, &.{ 9, 129, 'm', 'a', 's', 'k', 'd', }, ); // pong must be fin try assertWebSocketError( 1002, &.{ 10, 129, 'm', 'a', 's', 'k', 'd', }, ); } test "Client: ping reply" { try assertWebSocketMessage( // fin | pong, len &.{ 138, 0 }, // fin | ping, masked | len, 4-byte mask &.{ 137, 128, 0, 0, 0, 0 }, ); try assertWebSocketMessage( // fin | pong, len, payload &.{ 138, 5, 100, 96, 97, 109, 104 }, // fin | ping, masked | len, 4-byte mask, 5 byte payload &.{ 137, 133, 0, 5, 7, 10, 100, 101, 102, 103, 104 }, ); } test "Client: close message" { try assertWebSocketMessage( // fin | close, len, close code (normal) &.{ 136, 2, 3, 232 }, // fin | close, masked | len, 4-byte mask &.{ 136, 128, 0, 0, 0, 0 }, ); } test "server: mask" { var buf: [4000]u8 = undefined; const messages = [_][]const u8{ "1234", "1234" ** 99, "1234" ** 999 }; for (messages) |message| { // we need the message to be mutable since mask operates in-place const payload = buf[0..message.len]; @memcpy(payload, message); mask(&.{ 1, 2, 200, 240 }, payload); try testing.expectEqual(false, std.mem.eql(u8, payload, message)); mask(&.{ 1, 2, 200, 240 }, payload); try testing.expectEqual(true, std.mem.eql(u8, payload, message)); } } test "server: 404" { var c = try createTestClient(); defer c.deinit(); const res = try c.httpRequest("GET /unknown HTTP/1.1\r\n\r\n"); try testing.expectEqualStrings("HTTP/1.1 404 \r\n" ++ "Connection: Close\r\n" ++ "Content-Length: 9\r\n\r\n" ++ "Not found", res); } test "server: get /json/version" { const expected_response = "HTTP/1.1 200 OK\r\n" ++ "Content-Length: 48\r\n" ++ "Connection: Close\r\n" ++ "Content-Type: application/json; charset=UTF-8\r\n\r\n" ++ "{\"webSocketDebuggerUrl\": \"ws://127.0.0.1:9583/\"}"; { // twice on the same connection var c = try createTestClient(); defer c.deinit(); const res1 = try c.httpRequest("GET /json/version HTTP/1.1\r\n\r\n"); try testing.expectEqualStrings(expected_response, res1); } { // again on a new connection var c = try createTestClient(); defer c.deinit(); const res1 = try c.httpRequest("GET /json/version HTTP/1.1\r\n\r\n"); try testing.expectEqualStrings(expected_response, res1); } } fn assertHTTPError( comptime expected_status: u16, comptime expected_body: []const u8, input: []const u8, ) !void { var c = try createTestClient(); defer c.deinit(); const res = try c.httpRequest(input); const expected_response = std.fmt.comptimePrint( "HTTP/1.1 {d} \r\nConnection: Close\r\nContent-Length: {d}\r\n\r\n{s}", .{ expected_status, expected_body.len, expected_body }, ); try testing.expectEqualStrings(expected_response, res); } fn assertWebSocketError(close_code: u16, input: []const u8) !void { var c = try createTestClient(); defer c.deinit(); try c.handshake(); try c.stream.writeAll(input); const msg = try c.readWebsocketMessage() orelse return error.NoMessage; defer if (msg.cleanup_fragment) { c.reader.cleanup(); }; try testing.expectEqual(.close, msg.type); try testing.expectEqual(2, msg.data.len); try testing.expectEqual(close_code, std.mem.readInt(u16, msg.data[0..2], .big)); } fn assertWebSocketMessage(expected: []const u8, input: []const u8) !void { var c = try createTestClient(); defer c.deinit(); try c.handshake(); try c.stream.writeAll(input); const msg = try c.readWebsocketMessage() orelse return error.NoMessage; defer if (msg.cleanup_fragment) { c.reader.cleanup(); }; const actual = c.reader.buf[0 .. msg.data.len + 2]; try testing.expectEqualSlices(u8, expected, actual); } const MockCDP = struct { messages: std.ArrayListUnmanaged([]const u8) = .{}, allocator: Allocator = testing.allocator, fn init(_: Allocator, client: anytype) MockCDP { _ = client; return .{}; } fn deinit(self: *MockCDP) void { const allocator = self.allocator; for (self.messages.items) |msg| { allocator.free(msg); } self.messages.deinit(allocator); } fn handleMessage(self: *MockCDP, message: []const u8) bool { const owned = self.allocator.dupe(u8, message) catch unreachable; self.messages.append(self.allocator, owned) catch unreachable; return true; } }; fn createTestClient() !TestClient { const address = std.net.Address.initIp4([_]u8{ 127, 0, 0, 1 }, 9583); const stream = try std.net.tcpConnectToAddress(address); const timeout = std.mem.toBytes(posix.timeval{ .sec = 2, .usec = 0, }); try posix.setsockopt(stream.handle, posix.SOL.SOCKET, posix.SO.RCVTIMEO, &timeout); try posix.setsockopt(stream.handle, posix.SOL.SOCKET, posix.SO.SNDTIMEO, &timeout); return .{ .stream = stream, .reader = .{ .allocator = testing.allocator, .buf = try testing.allocator.alloc(u8, 1024 * 16), }, }; } const TestClient = struct { stream: std.net.Stream, buf: [1024]u8 = undefined, reader: Reader(false), fn deinit(self: *TestClient) void { self.stream.close(); self.reader.deinit(); } fn httpRequest(self: *TestClient, req: []const u8) ![]const u8 { try self.stream.writeAll(req); var pos: usize = 0; var total_length: ?usize = null; while (true) { pos += try self.stream.read(self.buf[pos..]); if (pos == 0) { return error.NoMoreData; } const response = self.buf[0..pos]; if (total_length == null) { const header_end = std.mem.indexOf(u8, response, "\r\n\r\n") orelse continue; const header = response[0 .. header_end + 4]; const cl = blk: { const cl_header = "Content-Length: "; const start = (std.mem.indexOf(u8, header, cl_header) orelse { break :blk 0; }) + cl_header.len; const end = std.mem.indexOfScalarPos(u8, header, start, '\r') orelse { return error.InvalidContentLength; }; break :blk std.fmt.parseInt(usize, header[start..end], 10) catch { return error.InvalidContentLength; }; }; total_length = cl + header.len; } if (total_length) |tl| { if (pos == tl) { return response; } if (pos > tl) { return error.DataExceedsContentLength; } } } } fn handshake(self: *TestClient) !void { const request = "GET / HTTP/1.1\r\n" ++ "Connection: upgrade\r\n" ++ "Upgrade: websocket\r\n" ++ "sec-websocket-version:13\r\n" ++ "sec-websocket-key: this is my key\r\n" ++ "Custom: Header-Value\r\n\r\n"; const res = try self.httpRequest(request); try testing.expectEqualStrings("HTTP/1.1 101 Switching Protocols\r\n" ++ "Upgrade: websocket\r\n" ++ "Connection: upgrade\r\n" ++ "Sec-Websocket-Accept: flzHu2DevQ2dSCSVqKSii5e9C2o=\r\n\r\n", res); } fn readWebsocketMessage(self: *TestClient) !?Message { while (true) { const n = try self.stream.read(self.reader.readBuf()); if (n == 0) { return error.Closed; } self.reader.len += n; if (try self.reader.next()) |msg| { return msg; } } } };