mirror/browser
1
0 Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2026-09-17 08:27:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3329 from lightpanda-io/better-server

Browse Source 
Better server
This commit is contained in:
Karl Seguin authored and GitHub committed 2026-09-03 07:27:33 +08:00
2 parents 37f31afec5 c189d72adf
commit 658d350404
14 files changed
+3031 -1865
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Inbox.zig
+69 -1
View File
@@ -37,6 +37,10 @@ const Inbox = @This();
mutex: std.Io.Mutex = .init,
queue: DoublyLinkedList = .{},
// Payload bytes sitting in the queue. Used to disconnect a client if we've
// fallen too far behind (largely to protect against a misbehaving client)
queued_bytes: usize = 0,
// One-way latch, set by the worker's drainInbox the first time it
// observes a .disconnect (or .close) and never cleared. Ensures that, on
// multiple drains, the terminated state is preserved / communicated. This is
@@ -51,6 +55,13 @@ pub fn deinit(self: *Inbox) void {
const msg: *Message = @fieldParentPtr("node", node);
msg.deinit();
}
self.queued_bytes = 0;
}
pub fn queuedBytes(self: *Inbox) usize {
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
return self.queued_bytes;
}
pub fn push(self: *Inbox, arena: *lp.Arena, payload: Message.Payload) void {
@@ -61,6 +72,7 @@ pub fn push(self: *Inbox, arena: *lp.Arena, payload: Message.Payload) void {
msg.* = .{ .payload = payload, .arena = arena };
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
self.queued_bytes += payload.size();
self.queue.append(&msg.node);
}
@@ -68,7 +80,9 @@ pub fn pop(self: *Inbox) ?*Message {
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
const node = self.queue.popFirst() orelse return null;
return @fieldParentPtr("node", node);
const msg: *Message = @fieldParentPtr("node", node);
self.queued_bytes -= msg.payload.size();
return msg;
}
// Peek for a message matching `predicate` without removing it. Used by
@@ -99,6 +113,7 @@ pub fn popIf(self: *Inbox, predicate: *const fn (*Message) bool) ?*Message {
const msg: *Message = @fieldParentPtr("node", node);
if (predicate(msg)) {
self.queue.remove(node);
self.queued_bytes -= msg.payload.size();
return msg;
}
}
@@ -141,6 +156,14 @@ pub const Message = struct {
// pushes this on peer EOF, fatal WS framing error, or
// (now) JSON parse failure.
disconnect: ?anyerror,
pub fn size(self: Payload) usize {
return switch (self) {
.cdp => |c| c.raw.len,
.bidi, .ping => |b| b.len,
.close, .disconnect => 0,
};
}
};
pub const Cdp = struct {
@@ -346,3 +369,48 @@ test "Inbox: popIf picks first match in FIFO order" {
defer m.deinit();
try testing.expectEqual("first", m.payload.ping);
}
test "Inbox: queued bytes track the payloads" {
const arena_pool = &testing.test_app.arena_pool;
var inbox = Inbox{};
defer inbox.deinit();
try testing.expectEqual(0, inbox.queuedBytes());
{
const arena = try arena_pool.acquire(.tiny, "inbox test");
inbox.push(arena, .{ .ping = try arena.dupe(u8, "12345") });
}
try testing.expectEqual(5, inbox.queuedBytes());
{
// control payloads are free; only what the peer sends counts
const arena = try arena_pool.acquire(.tiny, "inbox test");
inbox.push(arena, .{ .disconnect = null });
}
try testing.expectEqual(5, inbox.queuedBytes());
{
const arena = try arena_pool.acquire(.tiny, "inbox test");
inbox.push(arena, .{ .bidi = try arena.dupe(u8, "abc") });
}
try testing.expectEqual(8, inbox.queuedBytes());
// popIf cherry-picks out of the middle, and has to pay the same toll
{
const m = inbox.popIf(struct {
fn f(msg: *Message) bool {
return msg.payload == .bidi;
}
}.f).?;
defer m.deinit();
}
try testing.expectEqual(5, inbox.queuedBytes());
{
const m = inbox.pop().?;
defer m.deinit();
}
try testing.expectEqual(0, inbox.queuedBytes());
}
src/Metrics.zig
+8 -2
View File
@@ -20,8 +20,11 @@ const std = @import("std");
const lp = @import("lightpanda");
const Metrics = @This();
const Driver = @import("server/Handshake.zig").Driver;
const Driver = @import("server/Driver.zig").Protocol;
serve_http_requests: CounterEnum("status", @import("network/http.zig").StatusCategory) = .{},
serve_http_evictions: Counter = .{},
serve_inbox_backlog: Counter = .{},
serve_connections: CounterEnum("driver", Driver) = .{},
serve_connection_limit: Counter = .{},
serve_active_connections: GaugeEnum("driver", Driver) = .{},
@@ -92,8 +95,11 @@ robots_access: CounterEnum("result", enum { allow, deny }) = .{},
// Emitted as each metric's "# HELP" line. A field without an entry is a
// compile error.
const help = .{
.serve_http_requests = "HTTP responses sent, by status category (includes the pre-parse 400/413 rejections)",
.serve_http_evictions = "HTTP connections closed for sitting past their deadline without completing a request",
.serve_inbox_backlog = "Websocket connections closed for queueing more unprocessed messages than the worker could drain",
.serve_connections = "Websocket connections accepted, by driver protocol",
.serve_connection_limit = "Connections rejected because --cdp-max-connections was reached (counted before the handshake, so no driver label)",
.serve_connection_limit = "Accepts deferred because the connection budget was full: the listener pauses until a slot frees (counted before any handshake, so no driver label)",
.serve_active_connections = "Currently connected clients, by driver protocol",
.serve_commands = "Commands dispatched, by driver protocol",
.serve_unknown_commands = "Commands rejected for an unknown domain, module or method, by driver protocol",
src/log.zig
+1
View File
@@ -37,6 +37,7 @@ pub const Scope = enum {
note,
not_implemented,
scheduler,
serve,
storage,
telemetry,
unknown_prop,
src/server/Connection.zig
-274
View File
@@ -1,274 +0,0 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
// 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 <https://www.gnu.org/licenses/>.
const std = @import("std");
const lp = @import("lightpanda");
const App = @import("../App.zig");
const Inbox = @import("../Inbox.zig");
const WS = @import("../network/WS.zig");
const sys_net = @import("../sys/net.zig");
const ArenaPool = @import("../ArenaPool.zig");
const CDP = @import("cdp/CDP.zig");
const log = lp.log;
const posix = std.posix;
const ArenaAllocator = std.heap.ArenaAllocator;
pub const Connection = @This();
// is .starting until server.track is called
const State = enum { starting, live };
const Protocol = enum { cdp, bidi };
// reference to http_client.inbox
inbox: *Inbox,
arena_pool: *ArenaPool,
socket: posix.socket_t,
socket_flags: usize,
state: State = .starting,
protocol: Protocol,
reader: WS.Reader(true),
send_arena: ArenaAllocator,
pub fn init(
self: *Connection,
app: *App,
socket: posix.socket_t,
protocol: Protocol,
inbox: *Inbox,
) !void {
const socket_flags = try sys_net.fcntl(socket, posix.F.GETFL, 0);
const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
if (lp.IS_TEST == false) {
lp.assert(socket_flags & nonblocking == nonblocking, "Connection.init blocking", .{});
}
const config = app.config;
const allocator = app.allocator;
self.* = .{
.inbox = inbox,
.socket = socket,
.protocol = protocol,
.arena_pool = &app.arena_pool,
.socket_flags = socket_flags,
.reader = try .init(allocator, config.cdpMaxMessageSize()),
.send_arena = ArenaAllocator.init(allocator),
};
}
pub fn deinit(self: *Connection) void {
self.reader.deinit();
self.send_arena.deinit();
}
pub fn send(self: *Connection, 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 mode to get our write out
// We need to change it back to non-blocking.
_ = sys_net.fcntl(self.socket, posix.F.SETFL, self.socket_flags) catch |err| {
log.err(.app, "ws restore nonblocking", .{ .err = err });
};
};
LOOP: while (pos < data.len) {
const written = sys_net.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.
lp.assert(changed_to_blocking == false, "Connection.double block", .{});
changed_to_blocking = true;
_ = try sys_net.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;
}
}
pub fn sendPong(self: *Connection, data: []const u8) !void {
if (data.len == 0) {
return self.send(&WS.EMPTY_PONG);
}
var header_buf: [10]u8 = undefined;
const header = WS.frameHeader(&header_buf, .pong, data.len);
const allocator = self.send_arena.allocator();
const framed = try allocator.alloc(u8, header.len + data.len);
@memcpy(framed[0..header.len], header);
@memcpy(framed[header.len..], data);
return self.send(framed);
}
// Websocket frames have a variable length 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: *Connection, message: anytype, opts: std.json.Stringify.Options) !void {
const allocator = self.send_arena.allocator();
var aw = try std.Io.Writer.Allocating.initCapacity(allocator, 512);
// reserve space for the maximum possible header
try aw.writer.writeAll(&[_]u8{0} ** 10);
try std.json.Stringify.value(message, opts, &aw.writer);
const framed = WS.fillHeader(aw.toArrayList());
return self.send(framed);
}
pub fn sendJSONRaw(self: *Connection, buf: std.ArrayList(u8)) !void {
// Dangerous API!. We assume the caller has reserved the first 10
// bytes in `buf`.
const framed = WS.fillHeader(buf);
return self.send(framed);
}
pub fn feed(self: *Connection, data: []const u8) !bool {
var remaining = data;
while (remaining.len > 0) {
// we copy what will fit into our read buffer
const dst = self.reader.readBuf();
const used = @min(remaining.len, dst.len);
@memcpy(dst[0..used], remaining[0..used]);
self.reader.len += used;
// If we copied 1+ valid messages, this will process it.
if ((try self.processMessages()) == false) {
return false;
}
remaining = remaining[used..];
}
return true;
}
// Framing-only iteration over received bytes. Will process as many messages
// as are buffered.
fn processMessages(self: *Connection) !bool {
var reader = &self.reader;
while (true) {
const msg = (try reader.next()) orelse break;
const keep = switch (msg.type) {
.pong => true,
.ping, .text, .binary => try self.handleMessage(msg),
.close => blk: {
_ = try self.handleMessage(msg);
break :blk false;
},
};
if (msg.cleanup_fragment) {
reader.cleanup();
}
if (!keep) {
return false;
}
}
// 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 handleMessage(self: *Connection, msg: WS.Message) !bool {
switch (msg.type) {
.text, .binary => return switch (self.protocol) {
.cdp => self.pushCdp(msg.data),
.bidi => self.pushBiDi(msg.data),
},
.ping => {
const arena = try self.arena_pool.acquire(.tiny, "ws ping");
errdefer arena.release();
self.inbox.push(arena, .{ .ping = try arena.dupe(u8, msg.data) });
return true;
},
.close => {
const arena = try self.arena_pool.acquire(.tiny, "ws close");
self.inbox.push(arena, .close);
return true;
},
.pong => unreachable, // processMessages skips pong
}
}
// Parse a CDP JSON frame on the Network thread and push it onto the
// inbox already-parsed. The consumer's allowlist check works on
// `input.method` directly (no substring matching against raw JSON),
// and the worker doesn't re-parse on dispatch. On parse failure we
// push `.disconnect(error.InvalidJSON)` so the worker tears down —
// treated the same way as a fatal WS framing error.
fn pushCdp(self: *Connection, bytes: []const u8) !bool {
// TODO: is it worth trying to pad this for the cost overhead of parsing?
const arena = try self.arena_pool.acquire(bytes.len, "cdp data");
errdefer arena.release();
const raw = try arena.dupe(u8, bytes);
const input = std.json.parseFromSliceLeaky(
CDP.InputMessage,
arena.allocator(),
raw,
.{ .ignore_unknown_fields = true },
) catch {
self.inbox.push(arena, .{ .disconnect = error.InvalidJSON });
return false;
};
self.inbox.push(arena, .{ .cdp = .{
.raw = raw,
.input = input,
} });
return true;
}
// BiDi frames are pushed raw; the worker parses them. Unlike CDP there's
// no allowlist that needs the method name on this thread yet — when BiDi
// grows request interception, this is where that parse would go.
fn pushBiDi(self: *Connection, bytes: []const u8) !bool {
const arena = try self.arena_pool.acquire(bytes.len, "bidi data");
errdefer arena.release();
self.inbox.push(arena, .{ .bidi = try arena.dupe(u8, bytes) });
return true;
}
pub fn shutdown(self: *Connection) void {
sys_net.shutdown(self.socket, .recv) catch {};
}
src/server/Driver.zig
+50 -39
View File
@@ -19,63 +19,64 @@
const std = @import("std");
const lp = @import("lightpanda");
const WS = @import("../network/WS.zig");
const Inbox = @import("../Inbox.zig");
const CDP = @import("cdp/CDP.zig");
const Server = @import("Server.zig");
const BiDi = @import("bidi/BiDi.zig");
const Connection = @import("Connection.zig");
const Browser = @import("../browser/Browser.zig");
const Session = @import("../browser/Session.zig");
const WS = @import("WS.zig");
const Link = @import("Link.zig");
const CDP = @import("cdp/CDP.zig");
const BiDi = @import("bidi/BiDi.zig");
const log = lp.log;
// Parts of the driver are owned by the server run loop, parts are owned by
// Parts of the driver are owned by the server loop, parts are owned by
// the worker thread. The run loop reads messages and pushes to the inbox,
// the worker mostly just writes to the socket.
//
// What every protocol has - a connection, a browser, a link to the network
// thread - lives here rather than behind `impl`, so the shared paths are plain
// field access. Only what genuinely differs switches on `impl`.
const Driver = @This();
pub const Impl = union(enum) {
// Doubles as the metrics label
pub const Protocol = enum { cdp, bidi };
pub const Impl = union(Protocol) {
cdp: *CDP,
bidi: *BiDi,
};
impl: Impl,
conn: *Connection,
// every implementation has this
conn: *Link,
browser: *Browser,
link: *Server.Link,
// The protocol's log scope, so shared code still logs as .cdp / .bidi.
scope: log.Scope,
// Called from CDP.init / BiDi.init, where conn, link and browser are all
// still undefined: we only take their addresses, which the impl's own
// allocation already fixed.
// Called from CDP.init / BiDi.init, where conn and browser are both still
// undefined: we only take their addresses, which the impl's own allocation
// already fixed.
pub fn init(impl: Impl) Driver {
return switch (impl) {
// The tag names line up with the log scopes of the same name.
inline else => |d, tag| .{
.impl = impl,
.conn = &d.conn,
.link = &d.link,
.browser = &d.browser,
.scope = @field(log.Scope, @tagName(tag)),
.scope = @field(log.Scope, @tagName(tag)), // The tag names line up with the log scopes of the same name.
},
};
}
// Server run loop. Received data, driver returns false to signal it should
// disconnect.
pub fn onData(self: *const Driver, data: []const u8) anyerror!bool {
return self.conn.feed(data);
// server loop. The socket is readable, drain up to budget bytes
pub fn onReadable(self: *const Driver, budget: usize) anyerror!bool {
const read = try self.conn.readAvailable(budget);
if (read.pushed) {
self.wakeup();
}
return read.keep;
}
// Server run loop. Called when it drops the link unsolicited (peer EOF, ...)
// server loop. Called when it drops the link unsolicited (peer EOF, ...)
pub fn onLinkDisconnect(self: *const Driver, err: ?anyerror) void {
const arena = self.browser.arena_pool.acquire(.tiny, "driver disconnect") catch |e| switch (e) {
error.OutOfMemory => @panic("OOM"),
@@ -84,6 +85,23 @@ pub fn onLinkDisconnect(self: *const Driver, err: ?anyerror) void {
// when tick() discovers the terminatePending flag is set.
self.browser.http_client.inbox.push(arena, .{ .disconnect = err });
self.browser.env.requestTerminate();
self.wakeup();
}
// server loop. We used to send a nice WS close frame here but (a) it isn't strictly
// required and (b) we'd have to protect against an interleaved write from
// the worker thread.
pub fn shutdown(self: *const Driver) void {
self.browser.env.terminate();
self.conn.shutdown();
}
// a server-processed call (onReadable, onLinkDisconnect) wants to signal the
// worker that there's data in its inbox waiting to be processed.
fn wakeup(self: *const Driver) void {
self.browser.http_client.handles.wakeup() catch |err| {
log.err(self.scope, "wakeup", .{ .err = err });
};
}
// Worker thread. We're processing messages from the inbox.
@@ -136,12 +154,16 @@ pub fn run(self: *const Driver) void {
// One iteration of the worker loop. Returns false to disconnect.
fn tick(self: *const Driver) !bool {
if (self.browser.env.terminatePending()) {
// Maybe something bad happened (e.g. watchdog) or maybe the client
// just disconnected. Check the inbox to see if there's a disconnect
// message and, if so, it'll handle it directly.
// Our own requestTerminate from onLinkDisconnect: the peer is gone or
// sent garbage. Report it with its own close code, nothing to warn
// about. Pops close/disconnect only: nothing else may be dispatched
// in a shutting-down state.
self.browser.http_client.drainTerminal() catch |err| switch (err) {
error.ClientDisconnected => return false,
};
// Anything else means someone decided this browser must die (e.g.
// shutdown, or the heap limit was reached).
log.warn(self.scope, "closing connection", .{ .reason = "pending terminate" });
// The worker thread is the sole writer of this socket, so sending
// the close frame here can't interleave with another write.
@@ -195,14 +217,3 @@ fn pageWait(self: *const Driver) ?PageWait {
},
}
}
// signal handler thread
pub fn shutdown(self: *const Driver) void {
if (self.conn.state == .live) {
self.browser.env.terminate();
// We use to send a nice WS close frame here but (a) it isn't
// strictly required and (b) we'd have to protect against an interleaved
// write from the worker thread.
}
self.conn.shutdown();
}
src/server/Handshake.zig
-551
View File
@@ -1,551 +0,0 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
// 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 <https://www.gnu.org/licenses/>.
// The pre-upgrade HTTP phase of a connection. Owns the socket until it
// either serves a plain HTTP request (/json/*, /metrics) and closes, or
// completes a websocket upgrade — at which point the request path decides
// which protocol driver the connection is handed to.
const std = @import("std");
const lp = @import("lightpanda");
const App = @import("../App.zig");
const sys_net = @import("../sys/net.zig");
const uuidv4 = @import("../id.zig").uuidv4;
const header_parser = @import("../network/header_parser.zig");
const bidi_session = @import("bidi/session.zig");
const log = lp.log;
const posix = std.posix;
const Handshake = @This();
pub const Driver = enum { cdp, bidi };
// Which driver an upgraded socket is handed to.
pub const Route = union(Driver) {
cdp,
bidi: ?[36]u8, // the sessionId
};
// Which route families are served.
pub const Protocols = struct {
cdp: bool = false,
webdriver: bool = false,
};
// What every handshake on a server needs; built once by Server.
pub const Options = struct {
protocols: Protocols,
bidi_session_url: []const u8,
json_version_response: []const u8,
};
app: *App,
len: usize = 0,
socket: posix.socket_t,
// cdpMaxHTTPMessageSize is a u14, so this covers any configured limit.
buf: [std.math.maxInt(u14) + 1]u8 = undefined,
options: *const Options,
const Result = union(enum) {
more,
close,
upgrade: Route,
};
// Runs the HTTP phase to completion. Returns the route to hand the
// upgraded socket to, or null if the connection is done (plain HTTP
// request served, error, timeout or disconnect).
pub fn run(app: *App, socket: posix.socket_t, options: *const Options) ?Route {
var self = Handshake{
.app = app,
.socket = socket,
.options = options,
};
while (true) {
var pfds = [_]posix.pollfd{.{
.fd = self.socket,
.events = posix.POLL.IN,
.revents = 0,
}};
const n = posix.poll(&pfds, 5000) catch return null;
if (n == 0) {
log.info(.cdp, "handshake timeout", .{});
return null;
}
const read_bytes = posix.read(self.socket, self.buf[self.len..]) catch |err| {
log.warn(.cdp, "handshake read", .{ .err = err });
return null;
};
if (read_bytes == 0) {
log.info(.cdp, "handshake disconnect", .{});
return null;
}
self.len += read_bytes;
const result = self.processHttpRequest() catch return null;
switch (result) {
.more => continue,
.close => return null,
.upgrade => |route| return route,
}
}
}
fn processHttpRequest(self: *Handshake) !Result {
const request = self.buf[0..self.len];
if (request.len > self.app.config.cdpMaxHTTPMessageSize()) {
log.warn(.cdp, "message too big", .{ .type = "HTTP", .len = request.len, .hint = "See the --cdp-max-http-message-size <bytes>" });
self.sendHttpError(413, "Request too large");
return error.RequestTooLarge;
}
// Wait for the whole header block; put any more data here.
const head_len = (std.mem.indexOf(u8, request, "\r\n\r\n") orelse return .more) + 4;
return self.handleHttpRequest(request, head_len) catch |err| {
switch (err) {
error.NotFound => self.sendHttpError(404, "Not found"),
error.ForbiddenOrigin => self.sendHttpError(403, "Origin not allowed"),
error.ForbiddenHost => self.sendHttpError(403, "Host not allowed"),
error.InvalidRequest => self.sendHttpError(400, "Invalid request"),
error.InvalidProtocol => self.sendHttpError(400, "Invalid HTTP protocol"),
error.MissingHeaders => self.sendHttpError(400, "Missing required header"),
error.InvalidUpgradeHeader => self.sendHttpError(400, "Unsupported upgrade type"),
error.InvalidVersionHeader => self.sendHttpError(400, "Invalid websocket version"),
error.InvalidConnectionHeader => self.sendHttpError(400, "Invalid connection header"),
else => {
log.err(.app, "server 500", .{ .err = err, .req = request[0..@min(100, request.len)] });
self.sendHttpError(500, "Internal Server Error");
},
}
return err;
};
}
fn handleHttpRequest(self: *Handshake, request: []u8, head_len: usize) !Result {
if (request.len < 18) {
// 18 is [generously] the smallest acceptable HTTP request
return error.InvalidRequest;
}
// The classic WebDriver session bootstrap is the only thing with a body.
if (std.mem.startsWith(u8, request, "POST ") or std.mem.startsWith(u8, request, "DELETE ")) {
if (!self.options.protocols.webdriver) {
return error.NotFound;
}
return self.handleWebDriverRequest(request, head_len);
}
if (std.mem.eql(u8, request[0..4], "GET ") == false) {
return error.NotFound;
}
// Everything else is a body-less GET: the header block is the request.
if (head_len != request.len) {
return .more;
}
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, "/metrics") and self.app.config.metricsEndpointEnabled()) {
try self.sendMetrics();
self.shutdown();
return .close;
}
if (self.options.protocols.webdriver) {
if (std.mem.eql(u8, url, "/session")) {
// /session is the path Firefox advertises its BiDi endpoint on
try self.upgrade(request);
return .{ .upgrade = .{ .bidi = null } };
}
if (std.mem.startsWith(u8, url, "/session/") and url.len == "/session/".len + 36) {
// The URL a POST /session handed out; the session id is the suffix.
var session_id: [36]u8 = undefined;
@memcpy(&session_id, url["/session/".len..]);
try self.upgrade(request);
return .{ .upgrade = .{ .bidi = session_id } };
}
if (std.mem.eql(u8, url, "/status")) {
// WebDriver's discovery endpoint; `ready` is whether a new session
// can be created, which the bootstrap never refuses.
return self.sendWebDriver("200 OK", .{ .ready = true, .message = "" });
}
}
if (!self.options.protocols.cdp) {
return error.NotFound;
}
if (std.mem.eql(u8, url, "/")) {
try self.upgrade(request);
return .{ .upgrade = .cdp };
}
if (std.mem.eql(u8, url, "/json/version") or std.mem.eql(u8, url, "/json/version/")) {
try self.send(self.options.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.
self.shutdown();
return .close;
}
if (std.mem.eql(u8, url, "/json/list") or std.mem.eql(u8, url, "/json/list/") or
std.mem.eql(u8, url, "/json") or std.mem.eql(u8, url, "/json/"))
{
try self.send(empty_json_list_response);
self.shutdown();
return .close;
}
if (std.mem.eql(u8, url, "/json/protocol") or std.mem.eql(u8, url, "/json/protocol/")) {
try self.send(protocol_response);
self.shutdown();
return .close;
}
return error.NotFound;
}
// TODO: Temporary solution that provides the bare minimum for Selenium to
// connect. Serve a few of the (classic) WebDriver HTTP API. It's obvious that
// Handshake.zig needs to become a more generic HTTP server/router, but that
// can be done after the experimental BiDi code lands.
fn handleWebDriverRequest(self: *Handshake, request: []const u8, head_len: usize) !Result {
// A malformed request line or header maps to a 400 in processHttpRequest.
const method, const path, _, var header_iterator = header_parser.parseRequest(request) catch {
return error.InvalidProtocol;
};
var content_length: usize = 0;
while (header_iterator.next() catch return error.InvalidRequest) |header| {
if (std.ascii.eqlIgnoreCase(header.key, "content-length")) {
content_length = std.fmt.parseInt(usize, header.value, 10) catch return error.InvalidRequest;
}
}
const total_len = head_len + content_length;
if (request.len < total_len) {
return .more;
}
if (request.len > total_len) {
return error.InvalidRequest;
}
const body = request[head_len..total_len];
switch (method) {
.post => if (std.mem.eql(u8, path, "/session")) {
return self.newSession(body);
},
.delete => if (std.mem.startsWith(u8, path, "/session/")) {
return self.sendWebDriver("200 OK", null);
},
else => {},
}
return error.NotFound;
}
fn newSession(self: *Handshake, body: []const u8) !Result {
const allocator = self.app.allocator;
const Capability = struct { webSocketUrl: ?bool = null };
const parsed = std.json.parseFromSlice(struct {
capabilities: ?struct {
alwaysMatch: ?Capability = null,
firstMatch: ?[]const Capability = null,
} = null,
}, allocator, body, .{ .ignore_unknown_fields = true }) catch {
return self.sendWebDriver("400 Bad Request", .{
.@"error" = "invalid argument",
.message = "invalid JSON body",
.stacktrace = "",
});
};
defer parsed.deinit();
// Without the capability the client intends to drive the session over
// HTTP, which this server doesn't serve: tell it now rather than 404
// its first real command.
if (!requestsWebSocketUrl(parsed.value.capabilities)) {
return self.sendWebDriver("500 Internal Server Error", .{
.@"error" = "session not created",
.message = "only WebDriver BiDi sessions are supported; request the webSocketUrl capability",
.stacktrace = "",
});
}
var session_id: [36]u8 = undefined;
uuidv4(&session_id);
const url = try std.fmt.allocPrint(allocator, "{s}{s}", .{ self.options.bidi_session_url, &session_id });
defer allocator.free(url);
return self.sendWebDriver("200 OK", .{
.sessionId = &session_id,
.capabilities = bidi_session.Capabilities{
.userAgent = self.app.config.http_headers.user_agent,
.webSocketUrl = url,
},
});
}
fn requestsWebSocketUrl(capabilities: anytype) bool {
const caps = capabilities orelse return false;
if (caps.alwaysMatch) |always| {
if (always.webSocketUrl == true) {
return true;
}
}
for (caps.firstMatch orelse &.{}) |first| {
if (first.webSocketUrl == true) {
return true;
}
}
return false;
}
// Answers a classic WebDriver request with {"value": value} and closes.
fn sendWebDriver(self: *Handshake, comptime status: []const u8, value: anytype) !Result {
const allocator = self.app.allocator;
var aw = try std.Io.Writer.Allocating.initCapacity(allocator, 512);
defer aw.deinit();
try std.json.Stringify.value(.{ .value = value }, .{}, &aw.writer);
const body = aw.written();
const response = try std.fmt.allocPrint(allocator, "HTTP/1.1 " ++ status ++ "\r\n" ++
"Content-Length: {d}\r\n" ++
"Connection: Close\r\n" ++
"Content-Type: application/json; charset=UTF-8\r\n\r\n" ++
"{s}", .{ body.len, body });
defer allocator.free(response);
try self.send(response);
self.shutdown();
return .close;
}
fn upgrade(self: *Handshake, request: []u8) !void {
// We need to make sure that we got all the necessary headers + values;
// a bit per required header.
const FOUND_UPGRADE: u8 = 1 << 0; // Upgrade: websocket
const FOUND_VERSION: u8 = 1 << 1; // Sec-WebSocket-Version: 13
const FOUND_CONNECTION: u8 = 1 << 2; // Connection: upgrade
const FOUND_KEY: u8 = 1 << 3; // Sec-WebSocket-Key
const FOUND_ALL = FOUND_UPGRADE | FOUND_VERSION | FOUND_CONNECTION | FOUND_KEY;
// A malformed request line maps to a 400 in processHttpRequest.
const method, _, const version, var header_iterator = header_parser.parseRequest(request) catch {
return error.InvalidProtocol;
};
if (method != .get or version != .@"1.1") {
return error.InvalidProtocol;
}
var found_headers: u8 = 0;
// We need to extract the `Sec-WebSocket-Key` value.
var sec_websocket_key: []const u8 = "";
// A malformed header maps to a 400 in processHttpRequest.
while (header_iterator.next() catch return error.InvalidRequest) |header| {
const key = header.key;
const value = header.value;
// Header names are case-insensitive; `Header.parse` keeps their
// original casing.
if (std.ascii.eqlIgnoreCase(key, "upgrade")) {
if (!std.ascii.eqlIgnoreCase("websocket", value)) {
return error.InvalidUpgradeHeader;
}
found_headers |= FOUND_UPGRADE;
} else if (std.ascii.eqlIgnoreCase(key, "sec-websocket-version")) {
if (value.len != 2 or value[0] != '1' or value[1] != '3') {
return error.InvalidVersionHeader;
}
found_headers |= FOUND_VERSION;
} else if (std.ascii.eqlIgnoreCase(key, "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;
}
found_headers |= FOUND_CONNECTION;
} else if (std.ascii.eqlIgnoreCase(key, "sec-websocket-key")) {
sec_websocket_key = value;
found_headers |= FOUND_KEY;
} else if (std.ascii.eqlIgnoreCase(key, "origin")) {
// Only a browser sends `Origin`, and a browser has no business
// driving CDP: whatever page sent this is cross-origin to us by
// definition, including one served from loopback itself. Scripted
// clients (Puppeteer, Playwright, chromedp, ...) never send it.
log.warn(.cdp, "rejected websocket origin", .{
.origin = value[0..@min(value.len, 64)],
});
return error.ForbiddenOrigin;
} else if (std.ascii.eqlIgnoreCase(key, "host")) {
const host = value;
const is_allowed = blk: {
// allow literal localhost
if (std.mem.startsWith(u8, host, "localhost:")) {
break :blk true;
}
_ = std.Io.net.IpAddress.parseLiteral(host) catch break :blk false;
break :blk true;
};
// Defense in depth against DNS rebinding: an IP literal is the only
// thing that can legitimately reach us, because no name has to be
// resolved to produce one. The one name we accept is
// `localhost:<port>`, which browsers hardwire to loopback without
// any DNS lookup. Any other name means something answered a DNS
// lookup with our address, which is exactly what a rebinding
// attack looks like. A request without a Host header isn't from a
// browser, so it can't be the vector.
if (!is_allowed) {
log.warn(.cdp, "rejected websocket host", .{
.host = host[0..@min(host.len, 64)],
.hint = "connect to the CDP endpoint by IP address or localhost",
});
return error.ForbiddenHost;
}
}
}
// Check if we've received all related headers.
if (found_headers != FOUND_ALL) {
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
// Response to an upgrade request is always this, with the
// Sec-Websocket-Accept value a special sha1 hash of the request
// "sec-websocket-key" 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";
var res: [template.len]u8 = template.*;
const key_pos = res.len - 32;
var h: [20]u8 = undefined;
var hasher = std.crypto.hash.Sha1.init(.{});
hasher.update(sec_websocket_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..]);
return self.send(&res);
}
fn sendMetrics(self: *Handshake) !void {
const allocator = self.app.allocator;
var aw = try std.Io.Writer.Allocating.initCapacity(allocator, 4096);
defer aw.deinit();
lp.metrics.write(&aw.writer);
const body = aw.written();
const response = try std.fmt.allocPrint(allocator, "HTTP/1.1 200 OK\r\n" ++
"Content-Length: {d}\r\n" ++
"Connection: Close\r\n" ++
"Content-Type: text/plain; version=0.0.4; charset=utf-8\r\n\r\n" ++
"{s}", .{ body.len, body });
defer allocator.free(response);
try self.send(response);
}
fn sendHttpError(self: *Handshake, 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 {};
}
// The socket is non-blocking (reads must never block once the network
// thread owns them), but our responses are small one-shot writes, so on
// WouldBlock we just wait for writability rather than queueing.
fn send(self: *Handshake, data: []const u8) !void {
var pos: usize = 0;
while (pos < data.len) {
const written = sys_net.write(self.socket, data[pos..]) catch |err| switch (err) {
error.WouldBlock => {
var pfds = [_]posix.pollfd{.{
.fd = self.socket,
.events = posix.POLL.OUT,
.revents = 0,
}};
const n = try posix.poll(&pfds, 5000);
if (n == 0) {
return error.Timeout;
}
continue;
},
else => return err,
};
if (written == 0) {
return error.Closed;
}
pos += written;
}
}
fn shutdown(self: *Handshake) void {
sys_net.shutdown(self.socket, .recv) catch {};
}
const empty_json_list_response =
"HTTP/1.1 200 OK\r\n" ++
"Content-Length: 2\r\n" ++
"Connection: Close\r\n" ++
"Content-Type: application/json; charset=UTF-8\r\n\r\n" ++
"[]";
const protocol_json = @embedFile("../data/protocol.json");
const protocol_response = std.fmt.comptimePrint(
"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",
.{protocol_json.len},
) ++ protocol_json;
src/server/Link.zig
+380
View File
@@ -0,0 +1,380 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
// 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 <https://www.gnu.org/licenses/>.
const std = @import("std");
const lp = @import("lightpanda");
const App = @import("../App.zig");
const Inbox = @import("../Inbox.zig");
const ArenaPool = @import("../ArenaPool.zig");
const sys_net = @import("../sys/net.zig");
const WS = @import("WS.zig");
const CDP = @import("cdp/CDP.zig");
const Driver = @import("Driver.zig");
const log = lp.log;
const posix = std.posix;
const ArenaAllocator = std.heap.ArenaAllocator;
// The worker's end of an upgraded connection (the loop's is Server.WebSocket).
// Reads/framing happen on the server run loop (readAvailable → inbox); the worker
// thread is the sole writer (send*). The two sides touch disjoint state
// (reader+inbox vs send_arena+socket write) so no lock is needed beyond the
// inbox's own.
const Link = @This();
// A send that hits WouldBlock waits at most this long for the peer to take
// more bytes.
const SEND_TIMEOUT_MS = 5_000;
// The loop reads as fast as it can. The worker _can_ be slow to process its
// inbox (e.g. stuck in a syncRequest). Still, we want _some_ limit on how much
// data is queued. This is 32 * the configured max message size. Which should
// be plenty for a well-behaving client.
const INBOX_BACKLOG_MESSAGES = 32;
inbox: *Inbox,
arena_pool: *ArenaPool,
socket: posix.socket_t,
protocol: Driver.Protocol,
reader: WS.Reader,
send_arena: ArenaAllocator,
send_timeout_ms: i32,
max_inbox_backlog: usize,
pub fn init(
self: *Link,
app: *App,
socket: posix.socket_t,
protocol: Driver.Protocol,
inbox: *Inbox,
) !void {
if (lp.IS_TEST == false) {
const socket_flags = try sys_net.fcntl(socket, posix.F.GETFL, 0);
const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
lp.assert(socket_flags & nonblocking == nonblocking, "Link.init blocking", .{});
}
const config = app.config;
const allocator = app.allocator;
self.* = .{
.inbox = inbox,
.socket = socket,
.protocol = protocol,
.arena_pool = &app.arena_pool,
.reader = try .init(allocator, config.cdpMaxMessageSize()),
.send_arena = ArenaAllocator.init(allocator),
.send_timeout_ms = SEND_TIMEOUT_MS,
.max_inbox_backlog = @as(usize, config.cdpMaxMessageSize()) * INBOX_BACKLOG_MESSAGES,
};
}
pub fn deinit(self: *Link) void {
self.reader.deinit();
self.send_arena.deinit();
}
pub fn send(self: *Link, data: []const u8) !void {
var pos: usize = 0;
const socket = self.socket;
defer _ = self.send_arena.reset(.{ .retain_with_limit = 1024 * 32 });
while (pos < data.len) {
const written = sys_net.write(socket, data[pos..]) catch |err| switch (err) {
// The socket is nonblocking so loop reads never stall. Writes are
// simpler if they can wait: no per-connection pending-write queue
// with its own allocations. Waiting is done with poll rather than
// by flipping the fd to blocking: O_NONBLOCK lives on the open
// file description, so a flip would reach the loop's reads too.
// Should virtually never happen.
error.WouldBlock => {
// The socket is nonblocking so that the main read loop doesn't
// block. But we don't want to make writes truly async, because
// then we'd need to allocate the message and hook that back into
// the main thread, so...we'll just pull until the we can write
// or we hit our send timeout
var fds = [_]posix.pollfd{.{ .fd = socket, .events = posix.POLL.OUT, .revents = 0 }};
if ((try posix.poll(&fds, self.send_timeout_ms)) == 0) {
return error.Timeout;
}
continue;
},
// a signal landed mid-write; nothing was written
error.Interrupted => continue,
else => return err,
};
if (written == 0) {
return error.Closed;
}
pos += written;
}
}
pub fn sendPong(self: *Link, data: []const u8) !void {
if (data.len == 0) {
return self.send(&WS.EMPTY_PONG);
}
var header_buf: [10]u8 = undefined;
const header = WS.frameHeader(&header_buf, .pong, data.len);
const allocator = self.send_arena.allocator();
const framed = try allocator.alloc(u8, header.len + data.len);
@memcpy(framed[0..header.len], header);
@memcpy(framed[header.len..], data);
return self.send(framed);
}
// Websocket frames have a variable-length header (2-10 bytes server->client).
// We serialize into a buffer whose first 10 bytes are reserved, then
// backfill the header right-aligned and send the slice.
pub fn sendJSON(self: *Link, message: anytype, opts: std.json.Stringify.Options) !void {
const allocator = self.send_arena.allocator();
var aw = try std.Io.Writer.Allocating.initCapacity(allocator, 512);
try aw.writer.writeAll(&[_]u8{0} ** 10);
try std.json.Stringify.value(message, opts, &aw.writer);
const framed = WS.fillHeader(aw.toArrayList());
return self.send(framed);
}
pub fn sendJSONRaw(self: *Link, buf: std.ArrayList(u8)) !void {
// Dangerous API! Assumes the caller reserved the first 10 bytes in buf.
const framed = WS.fillHeader(buf);
return self.send(framed);
}
pub const Read = struct {
// false once a close frame was consumed: stop reading, the worker
// replies and disconnects itself
keep: bool,
// at least one frame landed in the inbox
pushed: bool,
};
// Server loop. The socket is readable
pub fn readAvailable(self: *Link, budget: usize) !Read {
if (self.inbox.queuedBytes() >= self.max_inbox_backlog) {
lp.metrics.serve_inbox_backlog.incr();
return error.InboxBacklog;
}
var pushed = false;
var remaining = budget;
while (remaining > 0) {
const dst = self.reader.readBuf();
if (dst.len == 0) {
// a partial message already fills the buffer
return error.TooLarge;
}
const want = dst[0..@min(dst.len, remaining)];
const n = posix.read(self.socket, want) catch |err| switch (err) {
error.WouldBlock => break,
else => return err,
};
if (n == 0) {
return error.Closed;
}
self.reader.len += n;
if ((try self.processMessages(&pushed)) == false) {
return .{ .keep = false, .pushed = pushed };
}
remaining -= n;
if (n < want.len) {
// a short read: the socket is (very likely) drained
break;
}
}
return .{ .keep = true, .pushed = pushed };
}
fn processMessages(self: *Link, pushed: *bool) !bool {
var reader = &self.reader;
while (true) {
const msg = (try reader.next()) orelse break;
const keep = switch (msg.type) {
.pong => true,
.ping, .text, .binary => try self.handleMessage(msg, pushed),
.close => blk: {
_ = try self.handleMessage(msg, pushed);
break :blk false;
},
};
if (msg.cleanup_fragment) {
reader.cleanup();
}
if (!keep) {
return false;
}
}
reader.compact();
return true;
}
fn handleMessage(self: *Link, msg: WS.Message, pushed: *bool) !bool {
switch (msg.type) {
.text, .binary => return switch (self.protocol) {
.cdp => self.pushCdp(msg.data, pushed),
.bidi => self.pushBiDi(msg.data, pushed),
},
.ping => {
const arena = try self.arena_pool.acquire(.tiny, "ws ping");
errdefer arena.release();
self.inbox.push(arena, .{ .ping = try arena.dupe(u8, msg.data) });
pushed.* = true;
return true;
},
.close => {
const arena = try self.arena_pool.acquire(.tiny, "ws close");
self.inbox.push(arena, .close);
pushed.* = true;
return true;
},
.pong => unreachable, // processMessages skips pong
}
}
// Parse a CDP JSON frame on the run loop and push it already-parsed: the
// consumer's allowlist works on input.method directly and the worker
// doesn't re-parse. On parse failure push .disconnect(InvalidJSON) so the
// worker tears down, same as a fatal framing error.
fn pushCdp(self: *Link, bytes: []const u8, pushed: *bool) !bool {
const arena = try self.arena_pool.acquire(bytes.len, "cdp data");
errdefer arena.release();
const raw = try arena.dupe(u8, bytes);
const input = std.json.parseFromSliceLeaky(
CDP.InputMessage,
arena.allocator(),
raw,
.{ .ignore_unknown_fields = true },
) catch {
self.inbox.push(arena, .{ .disconnect = error.InvalidJSON });
pushed.* = true;
return false;
};
self.inbox.push(arena, .{ .cdp = .{ .raw = raw, .input = input } });
pushed.* = true;
return true;
}
// BiDi frames are pushed raw; the worker parses them.
fn pushBiDi(self: *Link, bytes: []const u8, pushed: *bool) !bool {
const arena = try self.arena_pool.acquire(bytes.len, "bidi data");
errdefer arena.release();
self.inbox.push(arena, .{ .bidi = try arena.dupe(u8, bytes) });
pushed.* = true;
return true;
}
// Server loop, closing only the read side. The worker can still send a message
// (e.g. a close frame).
pub fn shutdown(self: *Link) void {
sys_net.shutdown(self.socket, .recv) catch {};
}
const testing = @import("../testing.zig");
test "link: send gives up when the peer stops reading" {
var pair: [2]posix.socket_t = undefined;
if (std.c.socketpair(posix.AF.LOCAL, posix.SOCK.STREAM, 0, &pair) != 0) {
return error.SocketPairFailed;
}
defer sys_net.close(pair[0]);
defer sys_net.close(pair[1]);
const small = std.mem.toBytes(@as(c_int, 4096));
try posix.setsockopt(pair[0], posix.SOL.SOCKET, posix.SO.RCVBUF, &small);
try posix.setsockopt(pair[1], posix.SOL.SOCKET, posix.SO.SNDBUF, &small);
const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
const flags = try sys_net.fcntl(pair[1], posix.F.GETFL, 0);
_ = try sys_net.fcntl(pair[1], posix.F.SETFL, flags | nonblocking);
var inbox: Inbox = .{};
defer inbox.deinit();
var link: Link = undefined;
try link.init(testing.test_app, pair[1], .cdp, &inbox);
defer link.deinit();
// shorten the wait so the test doesn't sit out the real one
link.send_timeout_ms = 50;
const payload = try testing.allocator.alloc(u8, 1024 * 1024);
defer testing.allocator.free(payload);
@memset(payload, 'a');
// Nobody drains pair[0]. send() waits for writability to finish the
// write; unbounded, that parks the worker forever and, because shutdown
// only half-closes the read side, hangs the whole process on SIGINT.
try testing.expectError(error.Timeout, link.send(payload));
// and the run loop's reads share the fd: it must still be non-blocking
try testing.expectEqual(flags | nonblocking, try sys_net.fcntl(pair[1], posix.F.GETFL, 0));
}
test "link: stops reading once the worker's inbox backs up" {
var pair: [2]posix.socket_t = undefined;
if (std.c.socketpair(posix.AF.LOCAL, posix.SOCK.STREAM, 0, &pair) != 0) {
return error.SocketPairFailed;
}
defer sys_net.close(pair[0]);
defer sys_net.close(pair[1]);
const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
const flags = try sys_net.fcntl(pair[1], posix.F.GETFL, 0);
_ = try sys_net.fcntl(pair[1], posix.F.SETFL, flags | nonblocking);
var inbox: Inbox = .{};
defer inbox.deinit();
var link: Link = undefined;
try link.init(testing.test_app, pair[1], .cdp, &inbox);
defer link.deinit();
// a ceiling below one max-size message would refuse what was configured
try testing.expect(link.max_inbox_backlog > testing.test_app.config.cdpMaxMessageSize());
// an empty inbox reads normally (nothing pending, so nothing pushed)
const read = try link.readAvailable(1024);
try testing.expectEqual(true, read.keep);
try testing.expectEqual(false, read.pushed);
// a worker that has fallen this far behind isn't going to catch up
{
const arena = try testing.test_app.arena_pool.acquire(link.max_inbox_backlog, "backlog test");
const payload = try arena.allocator().alloc(u8, link.max_inbox_backlog);
inbox.push(arena, .{ .bidi = payload });
}
try testing.expectEqual(link.max_inbox_backlog, inbox.queuedBytes());
try testing.expectError(error.InboxBacklog, link.readAvailable(1024));
// and it recovers once the worker drains
{
const msg = inbox.pop().?;
defer msg.deinit();
}
try testing.expectEqual(0, inbox.queuedBytes());
_ = try link.readAvailable(1024);
}
src/server/Server.zig
+1441 -917
View File
File diff suppressed because it is too large. Load diff
src/network/WS.zig → src/server/WS.zig
+42 -9
View File
@@ -108,13 +108,18 @@ pub fn fillHeader(buf: std.ArrayList(u8)) []const u8 {
const RECLAIM_TO = 256 * 1024;
const RECLAIM_AFTER = 8;
// 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
pub fn Reader(comptime EXPECT_MASK: bool) type {
pub const Reader = ReaderM(true);
pub const ReaderNoMask = ReaderM(false);
// WebSocket and HTTP aware reader. EXPECT_MASK is always true, (since this is
// only used to read server mesages) except for testing, where we setup test
// clients.
fn ReaderM(comptime EXPECT_MASK: bool) type {
return struct {
allocator: Allocator,
buf: []u8,
// position in buf of the start of the next message
pos: usize = 0,
@@ -124,8 +129,6 @@ pub fn Reader(comptime EXPECT_MASK: bool) type {
max_message_size: usize,
buf: []u8,
fragments: ?Fragments = null,
// consecutive messages we've received which fit i RECLAIM_TO
@@ -220,7 +223,9 @@ pub fn Reader(comptime EXPECT_MASK: bool) type {
buf = self.buf[0..len];
// we need more data
return null;
} else if (buf.len < message_len) {
}
if (buf.len < message_len) {
// we need more data
return null;
}
@@ -393,7 +398,7 @@ pub fn Reader(comptime EXPECT_MASK: bool) type {
// don't need to narrow it first; unrecognized errors return null.
pub fn errorReply(err: anyerror) ?[]const u8 {
return switch (err) {
error.TooLarge => &CLOSE_TOO_BIG,
error.TooLarge, error.InboxBacklog => &CLOSE_TOO_BIG,
error.Masked,
error.NotMasked,
error.ReservedFlags,
@@ -537,7 +542,7 @@ fn feedAndDrain(reader: anytype, frame: []const u8) !void {
test "reader: reclaims buffer after a run of small messages" {
const allocator = testing.allocator;
var reader = try Reader(false).init(allocator, 4 * 1024 * 1024);
var reader = try ReaderNoMask.init(allocator, 4 * 1024 * 1024);
defer reader.deinit();
// A large message forces the buffer to grow well past RECLAIM_TO.
@@ -577,3 +582,31 @@ test "reader: reclaims buffer after a run of small messages" {
try testing.expect(reader.buf.len > RECLAIM_TO);
try testing.expectEqual(@as(usize, 0), reader.small_message_streak);
}
test "reader: control frame arriving in pieces" {
const allocator = testing.allocator;
var reader = try ReaderNoMask.init(allocator, 1024 * 1024);
defer reader.deinit();
// A ping with a 114 byte payload; only the header and 50 bytes of it
// have arrived. The control branch skips the "is the whole frame here"
// check that the data branches do, so next() used to slice past len.
var frame: [2 + 114]u8 = undefined;
frame[0] = 128 | 9; // FIN + ping
frame[1] = 114;
@memset(frame[2..], 'a');
const partial = frame[0 .. 2 + 50];
@memcpy(reader.readBuf()[0..partial.len], partial);
reader.len += partial.len;
try testing.expectEqual(@as(?Message, null), try reader.next());
// the rest arrives
const rest = frame[2 + 50 ..];
@memcpy(reader.readBuf()[0..rest.len], rest);
reader.len += rest.len;
const msg = (try reader.next()) orelse return error.NoMessage;
try testing.expectEqual(.ping, msg.type);
try testing.expectEqual(114, msg.data.len);
}
src/server/bidi/BiDi.zig
+4 -17
View File
@@ -25,11 +25,10 @@ const Server = @import("../Server.zig");
const Browser = @import("../../browser/Browser.zig");
const Session = @import("../../browser/Session.zig");
const Notification = @import("../../Notification.zig");
const NodeRegistry = @import("../../NodeRegistry.zig");
const Link = @import("../Link.zig");
const Driver = @import("../Driver.zig");
const Connection = @import("../Connection.zig");
const script = @import("script.zig");
const remote_value = @import("remote_value.zig");
@@ -40,11 +39,7 @@ const Allocator = std.mem.Allocator;
const BiDi = @This();
app: *App,
conn: Connection,
// Server run-loop read-side handle for the socket. Server registers it
// after the handshake and unregisters before teardown; see CDP.zig.
link: Server.Link,
conn: Link,
// Re-used arena for processing a message. Works because we strictly process
// one message at a time.
@@ -93,7 +88,6 @@ pub fn init(self: *BiDi, app: *App, socket: posix.socket_t, session_id: ?[36]u8)
const allocator = app.allocator;
self.* = .{
.app = app,
.link = undefined,
.conn = undefined,
.browser = undefined,
.user_context = undefined,
@@ -105,21 +99,14 @@ pub fn init(self: *BiDi, app: *App, socket: posix.socket_t, session_id: ?[36]u8)
.session_arena = std.heap.ArenaAllocator.init(allocator),
};
const driver: Driver = .init(.{ .bidi = self });
const driver = Driver.init(.{ .bidi = self });
try self.browser.init(app, .{}, driver);
errdefer self.browser.deinit();
const http_client = &self.browser.http_client;
try self.conn.init(app, socket, .bidi, &http_client.inbox);
try self.conn.init(app, socket, .bidi, &self.browser.http_client.inbox);
errdefer self.conn.deinit();
self.link = .{
.driver = driver,
.state = .live,
.socket = socket,
.handles = http_client.handles,
};
self.notification = try Notification.init(allocator);
errdefer self.notification.deinit();
src/server/cdp/CDP.zig
+13 -31
View File
@@ -23,27 +23,27 @@ const App = @import("../../App.zig");
const Inbox = @import("../../Inbox.zig");
const Notification = @import("../../Notification.zig");
const WS = @import("../../network/WS.zig");
const http = @import("../../network/http.zig");
const Server = @import("../Server.zig");
const HttpClient = @import("../../network/HttpClient.zig");
const js = @import("../../browser/js/js.zig");
const Browser = @import("../../browser/Browser.zig");
const Session = @import("../../browser/Session.zig");
const Frame = @import("../../browser/Frame.zig");
const Page = @import("../../browser/Page.zig");
const Mime = @import("../../browser/Mime.zig");
const Frame = @import("../../browser/Frame.zig");
const Browser = @import("../../browser/Browser.zig");
const Session = @import("../../browser/Session.zig");
const Element = @import("../../browser/webapi/Element.zig");
const Label = @import("../../browser/webapi/element/html/Label.zig");
const Connection = @import("../Connection.zig");
const WS = @import("../WS.zig");
const Link = @import("../Link.zig");
const Server = @import("../Server.zig");
const Driver = @import("../Driver.zig");
const Incrementing = @import("id.zig").Incrementing;
const fetch = @import("domains/fetch.zig");
const network_domain = @import("domains/network.zig");
const js = lp.js;
const log = lp.log;
const json = std.json;
const posix = std.posix;
@@ -62,15 +62,10 @@ pub const InvocationIdGen = Incrementing(u32, "INV");
const CDP = @This();
app: *App,
conn: Connection,
conn: Link,
browser: Browser,
allocator: Allocator,
// Server run-loop read-side handle for the CDP socket. Populated in
// init; Server.serve calls registerLink(&cdp.link) after the
// worker-side handshake completes, and unregisterLink before teardown.
link: Server.Link,
// when true, any target creation must be attached.
target_auto_attach: bool = false,
@@ -103,16 +98,11 @@ browser_context_arena: std.heap.ArenaAllocator,
// Files handed out as IO stream handles (Page.printToPDF ReturnAsStream).
streams: @import("domains/io.zig").Streams,
pub fn init(
self: *CDP,
app: *App,
socket: posix.socket_t,
) !void {
pub fn init(self: *CDP, app: *App, socket: posix.socket_t) !void {
const allocator = app.allocator;
self.* = .{
.app = app,
.link = undefined,
.conn = undefined,
.browser = undefined,
.allocator = allocator,
@@ -124,20 +114,12 @@ pub fn init(
.streams = .{ .allocator = allocator },
};
const driver: Driver = .init(.{ .cdp = self });
const driver = Driver.init(.{ .cdp = self });
try self.browser.init(app, .{ .env = .{ .with_inspector = true } }, driver);
const http_client = &self.browser.http_client;
errdefer self.browser.deinit();
try self.conn.init(app, socket, .cdp, &http_client.inbox);
errdefer self.conn.deinit();
self.link = .{
.driver = driver,
.state = .live,
.socket = socket,
.handles = http_client.handles,
};
try self.conn.init(app, socket, .cdp, &self.browser.http_client.inbox);
}
pub fn deinit(self: *CDP) void {
@@ -1355,7 +1337,7 @@ pub const Command = struct {
// When we parse a JSON message from the client, this is the structure
// we always expect. Parsed on the Network thread inside
// Connection.handleMessage; the slices reference the raw JSON bytes
// Link.handleMessage; the slices reference the raw JSON bytes
// (or arena allocations for fields that needed unescaping). Both
// outlive the InputMessage for the inbox message's lifetime.
pub const InputMessage = struct {
src/server/http.zig
+929
View File
@@ -0,0 +1,929 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
// 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 <https://www.gnu.org/licenses/>.
const std = @import("std");
const lp = @import("lightpanda");
const App = @import("../App.zig");
const sys_net = @import("../sys/net.zig");
const header_parser = @import("../network/header_parser.zig");
const statusCategory = @import("../network/http.zig").statusCategory;
const Server = @import("Server.zig");
const Driver = @import("Driver.zig");
const bidi_session = @import("bidi/session.zig");
const uuidv4 = @import("../id.zig").uuidv4;
const log = lp.log;
const posix = std.posix;
const Allocator = std.mem.Allocator;
// A client connection in its http phase: loop-owned, pooled.
pub const Connection = struct {
state: State,
buffer: Buffer,
socket: posix.socket_t,
address: sys_net.IpAddress,
node: std.DoublyLinkedList.Node,
// When a keepalive (or just connected) connection should be closed
deadline: u64,
// Response that couldn't be sent without blocking. Socket will switch to
// "write-mode" until it's drained.
pending: ?Writing,
pub const Writing = struct {
pos: usize, // how ,uch of Data we've already written
data: Data,
keepalive: bool,
pub const Data = union(enum) {
// copied out of the server's scratch buffer; freed once written
owned: []const u8,
// lives as long as the server; referenced, never freed
static: []const u8,
};
pub fn remaining(self: *const Writing) []const u8 {
return switch (self.data) {
inline else => |d| d[self.pos..],
};
}
pub fn deinit(self: *const Writing, allocator: Allocator) void {
switch (self.data) {
.static => {},
.owned => |owned| allocator.free(owned),
}
}
};
pub fn deinit(self: *Connection) void {
self.buffer.deinit();
}
// True if the request is in keepalive state and thus is a candidate to be
// closed if we need its slot for a new connection.
pub fn isIdle(self: *const Connection) bool {
if (self.pending != null or self.buffer.len != 0) {
// has a pending write, or has extra data to read
return false;
}
return self.state == .header;
}
pub const Request = struct {
method: Method,
// origin-form, query string stripped, always starts with '/'
path: []const u8,
keepalive: bool,
body: []const u8,
// The raw request head (request line + headers, through the final
// CRLF CRLF); a slice into the read buffer. Upgrade handlers re-parse
// it for the WebSocket headers.
head: []const u8,
// Filled in by the router for /session/{id}[/...] routes; points
// into the read buffer like path does.
session_id: ?*const [36]u8 = null,
};
pub const Method = enum {
GET,
POST,
PUT,
DELETE,
};
pub const State = union(enum) {
header: void, // still parsing the header
request: Request,
pub fn parseHeader(self: *State, data: []u8) !bool {
const header_index = std.mem.indexOf(u8, data, "\r\n\r\n") orelse {
return false;
};
// include the last line's \r\n so every line, including the request
// line of a header-less request, is terminated
const header = data[0 .. header_index + 2];
const method, const path, const keepalive, const line_1_end = try parseRequestLine(header);
_ = line_1_end;
const body_start = header_index + 4;
const total = body_start + try contentLength(header);
if (data.len < total) {
// the body is still arriving
return false;
}
// A WebSocket upgrade may be pipelined with its first frames, but every
// client we care about waits for the 101 first. Anything past the
// declared body is unsupported (and rejects pipelining).
if (data.len != total) {
return error.BodyNotSupported;
}
self.* = .{ .request = .{
.method = method,
.path = path,
.keepalive = keepalive,
.body = data[body_start..total],
.head = data[0..body_start],
} };
return true;
}
// The classic WebDriver bootstrap (POST /session) is the only thing
// that sends a body; everything else is 0.
fn contentLength(header: []const u8) !usize {
const key = "\r\ncontent-length:";
const at = std.ascii.indexOfIgnoreCase(header, key) orelse return 0;
const start = at + key.len;
const end = std.mem.indexOfPos(u8, header, start, "\r\n") orelse return error.InvalidHeader;
const value = std.mem.trim(u8, header[start..end], " \t");
return std.fmt.parseInt(usize, value, 10) catch error.InvalidHeader;
}
fn parseRequestLine(header: []const u8) !struct { Method, []const u8, bool, usize } {
const l1 = std.mem.indexOfScalar(u8, header, '\r') orelse return error.InvalidHeader;
if (l1 == header.len) {
return error.InvalidHeader;
}
if (header[l1 + 1] != '\n') {
return error.InvalidHeader;
}
var it = std.mem.tokenizeScalar(u8, header[0..l1], ' ');
const method = std.meta.stringToEnum(Method, it.next() orelse return error.InvalidHeader) orelse return error.InvalidHTTPMethod;
// Only the origin-form request-target is accepted; nothing we serve
// reads the query string, so it's dropped here.
const target = it.next() orelse return error.InvalidHeader;
if (target[0] != '/') {
return error.InvalidHeader;
}
const path = target[0 .. std.mem.indexOfScalar(u8, target, '?') orelse target.len];
const protocol = it.next() orelse return error.InvalidHeader;
const keepalive = std.mem.indexOf(u8, protocol, "1.0") == null;
return .{ method, path, keepalive, l1 };
}
};
const Buffer = struct {
buf: []u8,
// position in buf up until where we have valid data
len: usize,
allocator: Allocator,
fn init(allocator: Allocator, size: usize) !Buffer {
return .{
.len = 0,
.buf = try allocator.alloc(u8, size),
.allocator = allocator,
};
}
fn deinit(self: *const Buffer) void {
self.allocator.free(self.buf);
}
pub fn read(self: *Buffer, socket: posix.socket_t) ![]u8 {
const len = self.len;
if (len == self.buf.len) {
return error.RequestTooLarge;
}
const n = try posix.read(socket, self.buf[len..]);
if (n == 0) {
return error.ConnectionClosed;
}
const total = len + n;
self.len = total;
return self.buf[0..total];
}
};
pub const Pool = struct {
allocator: Allocator,
free: std.DoublyLinkedList,
live: usize, // acquired and not yet released
retain: usize, // min # to keep
free_count: usize, // # of connections available in free
buffer_size: usize, // --cdp-max-http-message-size
pub fn init(app: *App) !Pool {
const retain = app.config.maxConnections();
var self = Pool{
.live = 0,
.free = .{},
.free_count = 0,
.retain = retain,
.allocator = app.allocator,
.buffer_size = app.config.cdpMaxHTTPMessageSize(),
};
errdefer self.deinit();
for (0..retain) |_| {
const conn = try self.create();
self.free.append(&conn.node);
self.free_count += 1;
}
return self;
}
// Every live connection must have been released (the server disconnects
// them all on deinit).
pub fn deinit(self: *Pool) void {
lp.assert(self.live == 0, "Connection.Pool.deinit live", .{ .live = self.live });
while (self.free.popFirst()) |node| {
const conn: *Connection = @fieldParentPtr("node", node);
self.destroy(conn);
}
}
pub fn acquire(self: *Pool) !*Connection {
const conn = blk: {
if (self.free.popFirst()) |node| {
self.free_count -= 1;
break :blk @as(*Connection, @fieldParentPtr("node", node));
}
break :blk try self.create();
};
self.live += 1;
return conn;
}
pub fn release(self: *Pool, conn: *Connection) void {
self.live -= 1;
if (self.free_count == self.retain) {
return self.destroy(conn);
}
conn.node = .{};
conn.socket = -1;
conn.address = .{ .ip4 = .unspecified(0) };
conn.deadline = 0;
conn.pending = null;
conn.buffer.len = 0;
conn.state = .header;
self.free.prepend(&conn.node);
self.free_count += 1;
}
fn create(self: *Pool) !*Connection {
const allocator = self.allocator;
const conn = try allocator.create(Connection);
errdefer allocator.destroy(conn);
conn.* = .{
.node = .{},
.socket = -1,
.address = .{ .ip4 = .unspecified(0) },
.deadline = 0,
.pending = null,
.state = .header,
.buffer = try .init(allocator, self.buffer_size),
};
return conn;
}
fn destroy(self: *Pool, conn: *Connection) void {
conn.deinit();
self.allocator.destroy(conn);
}
};
};
// How long a connection may sit without completing a request before we close it.
pub const IDLE_TIMEOUT_MS = 10_000;
pub fn processEvent(server: *Server, conn: *Connection, rw: Server.IOEvent.ReadWrite, now: u64) void {
if (conn.pending != null) {
// registered for OUT only; a hangup shows up as a write error
if (rw.writable or rw.hangup) {
flush(server, conn, now);
}
return;
}
if (rw.readable) {
const keepalive = processHTTP(server, conn, now) catch |err| blk: {
writeError(conn, err);
break :blk false;
};
if (keepalive == false) {
disconnect(server, conn);
}
// else: the socket is level-triggered and stays registered; the
// deadline was refreshed by processHTTP when the response went out
} else if (rw.hangup) {
disconnect(server, conn);
}
}
// Continues a write that previously hit WouldBlock.
fn flush(server: *Server, conn: *Connection, now: u64) void {
const pending = &conn.pending.?;
const remaining = pending.remaining();
const n = write(conn.socket, remaining) catch |err| {
log.debug(.serve, "flush", .{ .err = err });
return disconnect(server, conn);
};
if (n < remaining.len) {
// hit a WouldBlock
pending.pos += n;
return;
}
// write is complete
const keepalive = pending.keepalive;
pending.deinit(server.app.allocator);
conn.pending = null;
if (keepalive == false) {
return disconnect(server, conn);
}
server.io_engine.waitReadable(conn) catch |err| {
log.err(.serve, "wait readable", .{ .err = err });
return disconnect(server, conn);
};
touch(server, conn, now);
}
fn processHTTP(server: *Server, conn: *Connection, now: u64) !bool {
const http = &conn.state;
while (true) {
switch (http.*) {
.header => {
const data = try conn.buffer.read(conn.socket);
if (try http.parseHeader(data) == false) {
// don't have a complete header yet
return true;
}
if (comptime lp.IS_DEBUG) {
// we do have a complete header, the state must have transitioned
// to .request
std.debug.assert(http.* == .request);
}
},
.request => |*req| {
if (try serveHTTP(server, conn, req) == .upgraded) {
// The fd moved to a WebSocket (and out of server.http); all
// that's left of this Connection is to recycle it.
// upgradeConnection already took it out of http_connections.
recycle(server, conn);
return true;
}
// req lives in http.*; read what we need before resetting it
const keepalive = req.keepalive;
http.* = .header;
conn.buffer.len = 0;
if (conn.pending != null) {
// We got a WouldBlock and now have a pending write. The
// connection stays alive until we flush it. After the write
// if flushed, we'll apply the keepalive result.
return true;
}
if (keepalive == false) {
return false;
}
touch(server, conn, now);
return true;
},
}
}
}
// Error responses use a minimal, uniform shape: no reason phrase, an explicit
// Connection: Close, and no Content-Type. errorResponse builds it at comptime.
const invalid_request_response = errorResponse(400, "Invalid request");
const invalid_protocol_response = errorResponse(400, "Invalid HTTP protocol");
const missing_header_response = errorResponse(400, "Missing required header");
const forbidden_origin_response = errorResponse(403, "Origin not allowed");
const forbidden_host_response = errorResponse(403, "Host not allowed");
const request_too_large_response = errorResponse(413, "Request too large");
const not_found_response = errorResponse(404, "Not found");
const method_not_allowed_response = errorResponse(405, "Method not allowed");
const service_unavailable_response = errorResponse(503, "Too many connections");
const internal_error_response = errorResponse(500, "Internal server error");
const empty_json_list_response = staticResponse(.{ .status = "200 OK", .body = "[]", .content_type = "application/json; charset=UTF-8" });
// WebDriver's discovery endpoint; `ready` is whether a new session can be
// created, which the bootstrap never refuses.
const status_response = staticResponse(.{ .status = "200 OK", .body = "{\"value\":{\"ready\":true,\"message\":\"\"}}", .content_type = "application/json; charset=UTF-8" });
const delete_session_response = staticResponse(.{ .status = "200 OK", .body = "{\"value\":null}", .content_type = "application/json; charset=UTF-8" });
const protocol_response = staticResponse(.{ .status = "200 OK", .body = @embedFile("../data/protocol.json"), .content_type = "application/json; charset=UTF-8" });
const Served = enum {
responded,
upgraded,
};
const Route = struct {
method: Connection.Method,
// exact match against the normalized path
path: []const u8,
gate: Gate = .none,
handler: *const fn (*Server, *Connection, *Connection.Request) anyerror!Served,
// A closed gate makes the route invisible (404), not forbidden.
const Gate = enum {
none,
cdp,
webdriver,
metrics,
};
};
const routes = [_]Route{
.{ .method = .GET, .path = "/", .gate = .cdp, .handler = upgradeCDP },
.{ .method = .GET, .path = "/metrics", .gate = .metrics, .handler = serveMetrics },
.{ .method = .GET, .path = "/json/version", .gate = .cdp, .handler = serveJSONVersion },
.{ .method = .GET, .path = "/json/list", .gate = .cdp, .handler = serveJSONList },
.{ .method = .GET, .path = "/json", .gate = .cdp, .handler = serveJSONList },
.{ .method = .GET, .path = "/json/protocol", .gate = .cdp, .handler = serveJSONProtocol },
// /session is the path Firefox advertises its BiDi endpoint on
.{ .method = .GET, .path = "/session", .gate = .webdriver, .handler = upgradeBiDi },
.{ .method = .POST, .path = "/session", .gate = .webdriver, .handler = newSession },
.{ .method = .GET, .path = "/status", .gate = .webdriver, .handler = serveStatus },
};
const session_routes = [_]Route{
.{ .method = .GET, .path = "", .handler = upgradeBiDi },
.{ .method = .DELETE, .path = "", .handler = deleteSession },
};
// Routes under /session/{id}; path is what follows the id ("" for the
// session itself). The classic command surface goes here.
const SESSION_PREFIX = "/session/";
const SESSION_ID_LEN = 36;
fn serveHTTP(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
var path = req.path;
if (path.len > 1 and path[path.len - 1] == '/') {
path = path[0 .. path.len - 1];
}
if (std.mem.startsWith(u8, path, SESSION_PREFIX) and path.len >= SESSION_PREFIX.len + SESSION_ID_LEN) {
if (!server.protocols.webdriver) {
return serveNotFound(server, conn, req);
}
const tail = path[SESSION_PREFIX.len + SESSION_ID_LEN ..];
if (tail.len != 0 and tail[0] != '/') {
return serveNotFound(server, conn, req);
}
req.session_id = path[SESSION_PREFIX.len..][0..SESSION_ID_LEN];
return dispatch(server, &session_routes, conn, req, tail);
}
return dispatch(server, &routes, conn, req, path);
}
fn dispatch(server: *Server, comptime table: []const Route, conn: *Connection, req: *Connection.Request, path: []const u8) !Served {
var path_matched = false;
inline for (table) |route| {
if (std.mem.eql(u8, route.path, path) and gateOpen(server, route.gate)) {
if (route.method == req.method) {
return route.handler(server, conn, req);
}
path_matched = true;
}
}
if (path_matched) {
return serveMethodNotAllowed(server, conn, req);
}
return serveNotFound(server, conn, req);
}
// Best effort, connection is being closed. A partial write ends up as a partial
// write: no pending, no retry.
fn writeError(conn: *Connection, err: anyerror) void {
const response: []const u8 = switch (err) {
error.ConnectionClosed, error.ConnectionResetByPeer, error.BrokenPipe => return,
error.InvalidHeader, error.InvalidHTTPMethod, error.BodyNotSupported => invalid_request_response,
error.RequestTooLarge => request_too_large_response,
else => blk: {
log.warn(.serve, "serve error", .{ .err = err });
break :blk internal_error_response;
},
};
recordResponse(response);
_ = write(conn.socket, response) catch {};
}
// Every response starts with the status line our two builders emit, so the
// category is read straight off the bytes rather than threaded through.
fn recordResponse(response: []const u8) void {
const prefix = "HTTP/1.1 ";
lp.assert(std.mem.startsWith(u8, response, prefix), "Server.recordResponse status line", .{});
const status = std.fmt.parseInt(u16, response[prefix.len..][0..3], 10) catch 0;
lp.metrics.serve_http_requests.incr(statusCategory(status));
}
const Response = union(enum) {
// lives as long as the server; a queued remainder references it
static: []const u8,
// lives in server.scratch until the next response; a queued remainder is copied
dynamic: []const u8,
};
// Can do a partial write
fn write(socket: posix.socket_t, data: []const u8) !usize {
var pos: usize = 0;
while (pos < data.len) {
const n = sys_net.write(socket, data[pos..]) catch |err| switch (err) {
error.WouldBlock => break,
error.Interrupted => continue,
else => return err,
};
pos += n;
}
return pos;
}
// Dynamic responses are built in server.scratch with room for the header
// reserved up front; once the body length is known the header is written
// right-aligned against it (the same trick as WS.fillHeader).
const HEADER_RESERVE = 192;
fn beginBody(server: *Server) !*std.Io.Writer {
server.scratch.clearRetainingCapacity();
try server.scratch.writer.splatByteAll(0, HEADER_RESERVE);
return &server.scratch.writer;
}
fn serveDynamicHTTPResponse(server: *Server, conn: *Connection, req: *const Connection.Request, comptime status: []const u8, comptime content_type: []const u8) !Served {
const header_format = "HTTP/1.1 " ++ status ++ "\r\n" ++
"Content-Length: {d}\r\n" ++
"Content-Type: " ++ content_type ++ "\r\n\r\n";
// a usize prints as at most 20 digits
comptime std.debug.assert(header_format.len + 20 <= HEADER_RESERVE);
const buf = server.scratch.written();
var header_buf: [HEADER_RESERVE]u8 = undefined;
const header = std.fmt.bufPrint(&header_buf, header_format, .{buf.len - HEADER_RESERVE}) catch unreachable;
const start = HEADER_RESERVE - header.len;
@memcpy(buf[start..HEADER_RESERVE], header);
return serveHTTPResponse(server, conn, req, .{ .dynamic = buf[start..] });
}
fn errorResponse(comptime status: u16, comptime body: []const u8) []const u8 {
return std.fmt.comptimePrint(
"HTTP/1.1 {d} \r\nConnection: Close\r\nContent-Length: {d}\r\n\r\n{s}",
.{ status, body.len, body },
);
}
fn staticResponse(comptime opts: struct {
status: []const u8,
body: []const u8,
content_type: []const u8 = "text/plain",
close: bool = false,
}) []const u8 {
return std.fmt.comptimePrint("HTTP/1.1 " ++ opts.status ++ "\r\n" ++
"Content-Length: {d}\r\n" ++
(if (opts.close) "Connection: Close\r\n" else "") ++
"Content-Type: " ++ opts.content_type ++ "\r\n\r\n", .{opts.body.len}) ++ opts.body;
}
fn gateOpen(server: *const Server, gate: Route.Gate) bool {
return switch (gate) {
.none => true,
.cdp => server.protocols.cdp,
.webdriver => server.protocols.webdriver,
.metrics => server.app.config.metricsEndpointEnabled(),
};
}
fn upgradeCDP(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return upgrade(server, conn, req, .cdp, null);
}
fn serveJSONVersion(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = server.json_version_response });
}
fn serveJSONList(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = empty_json_list_response });
}
fn serveJSONProtocol(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = protocol_response });
}
fn serveMetrics(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
const writer = try beginBody(server);
lp.metrics.write(writer);
return serveDynamicHTTPResponse(server, conn, req, "200 OK", "text/plain; version=0.0.4; charset=utf-8");
}
fn serveStatus(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = status_response });
}
// req.session_id is null for GET /session, set for GET /session/{id}
fn upgradeBiDi(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
const session_id: ?[36]u8 = if (req.session_id) |s| s.* else null;
return upgrade(server, conn, req, .bidi, session_id);
}
// What Selenium does before it speaks BiDi: a classic POST /session that
// hands back the websocket URL of a session that already exists.
fn newSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
const allocator = server.app.allocator;
const Capability = struct { webSocketUrl: ?bool = null };
const parsed = std.json.parseFromSlice(struct {
capabilities: ?struct {
alwaysMatch: ?Capability = null,
firstMatch: ?[]const Capability = null,
} = null,
}, allocator, req.body, .{ .ignore_unknown_fields = true }) catch {
return serveWebDriver(server, conn, req, "400 Bad Request", .{
.@"error" = "invalid argument",
.message = "invalid JSON body",
.stacktrace = "",
});
};
defer parsed.deinit();
// Without the capability the client intends to drive the session over
// HTTP, which this server doesn't serve: tell it now rather than 404
// its first real command.
if (!requestsWebSocketUrl(parsed.value.capabilities)) {
return serveWebDriver(server, conn, req, "500 Internal Server Error", .{
.@"error" = "session not created",
.message = "only WebDriver BiDi sessions are supported; request the webSocketUrl capability",
.stacktrace = "",
});
}
var session_id: [36]u8 = undefined;
uuidv4(&session_id);
const url = try std.fmt.allocPrint(allocator, "{s}{s}", .{ server.bidi_session_url, &session_id });
defer allocator.free(url);
return serveWebDriver(server, conn, req, "200 OK", .{
.sessionId = &session_id,
.capabilities = bidi_session.Capabilities{
.userAgent = server.app.config.http_headers.user_agent,
.webSocketUrl = url,
},
});
}
fn requestsWebSocketUrl(capabilities: anytype) bool {
const caps = capabilities orelse return false;
if (caps.alwaysMatch) |always| {
if (always.webSocketUrl == true) {
return true;
}
}
for (caps.firstMatch orelse &.{}) |first| {
if (first.webSocketUrl == true) {
return true;
}
}
return false;
}
// Answers a classic WebDriver request with {"value": value}.
fn serveWebDriver(server: *Server, conn: *Connection, req: *const Connection.Request, comptime status: []const u8, value: anytype) !Served {
const writer = try beginBody(server);
try std.json.Stringify.value(.{ .value = value }, .{}, writer);
return serveDynamicHTTPResponse(server, conn, req, status, "application/json; charset=UTF-8");
}
fn deleteSession(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = delete_session_response });
}
fn serveNotFound(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = not_found_response });
}
fn serveMethodNotAllowed(server: *Server, conn: *Connection, req: *Connection.Request) !Served {
return serveHTTPResponse(server, conn, req, .{ .static = method_not_allowed_response });
}
// Writes what the socket will take now. Anything left is queued on the
// connection, which switches to waiting for writability.
fn serveHTTPResponse(server: *Server, conn: *Connection, req: *const Connection.Request, response: Response) !Served {
const data = switch (response) {
inline else => |d| d,
};
recordResponse(data);
const n = try write(conn.socket, data);
if (n == data.len) {
return .responded;
}
lp.assert(conn.pending == null, "Server.send pending", .{});
conn.pending = .{
.pos = 0,
.keepalive = req.keepalive,
.data = switch (response) {
.static => .{ .static = data[n..] },
.dynamic => .{ .owned = try server.app.allocator.dupe(u8, data[n..]) },
},
};
// on failure the caller disconnects, which frees pending
try server.io_engine.waitWritable(conn);
return .responded;
}
// HTTP-phase teardown. Websockets tear down via releaseWorker.
pub fn disconnect(server: *Server, conn: *Connection) void {
server.io_engine.remove(conn.socket);
sys_net.close(conn.socket);
if (conn.pending) |*pending| {
pending.deinit(server.app.allocator);
conn.pending = null;
}
server.http_connections.remove(&conn.node);
recycle(server, conn);
}
// Return a connection to the pool; a slot in the fd budget is free.
fn recycle(server: *Server, conn: *Connection) void {
server.http_connection_pool.release(conn);
server.slotFreed();
}
fn touch(server: *Server, conn: *Connection, now: u64) void {
conn.deadline = now + IDLE_TIMEOUT_MS;
const node = &conn.node;
if (server.http_connections.last == node) {
return;
}
server.http_connections.remove(&conn.node);
server.http_connections.append(&conn.node);
}
pub fn buildJSONVersionResponse(app: *const App, port: u16) ![]const u8 {
const host = app.config.advertiseHost();
if (app.config.bindIsWildcard()) {
// Serve is bound to INADDR_ANY but no --advertise-host was given;
// advertiseHost() falls back to 127.0.0.1 so clients can still
// connect locally. Surface the trade-off so users running
// outside the same host know they have to opt in.
log.note(.cdp, "advertising loopback for wildcard bind", .{
.message = "--host is a wildcard (0.0.0.0 / ::) without --advertise-host; clients on other hosts will need --advertise-host to reach the CDP endpoint",
});
}
const body_format =
"{{" ++
"\"Browser\": \"Lightpanda/1.0\", " ++
"\"Protocol-Version\": \"1.3\", " ++
"\"User-Agent\": \"Lightpanda/1.0\", " ++
"\"Lightpanda-Version\": \"" ++ lp.build_config.version ++ "\", " ++
"\"webSocketDebuggerUrl\": \"ws://{s}:{d}/\"" ++
"}}";
const body_len = std.fmt.count(body_format, .{ host, port });
const response_format =
"HTTP/1.1 200 OK\r\n" ++
"Content-Length: {d}\r\n" ++
"Content-Type: application/json; charset=UTF-8\r\n\r\n" ++
body_format;
return try std.fmt.allocPrint(app.allocator, response_format, .{ body_len, host, port });
}
// Shared upgrade path: validate the WebSocket headers, write the 101, park the
// fd, and spawn the worker that will build the driver and attach it.
fn upgrade(server: *Server, conn: *Connection, req: *Connection.Request, protocol: Driver.Protocol, session_id: ?[36]u8) !Served {
if (server.websocket_pool.isFull()) {
lp.metrics.serve_connection_limit.incr();
return serveHTTPResponse(server, conn, req, .{ .static = service_unavailable_response });
}
var accept_buf: [28]u8 = undefined;
const accept_key = webSocketAccept(req.head, &accept_buf) catch |err| {
const response: []const u8 = switch (err) {
error.ForbiddenOrigin => forbidden_origin_response,
error.ForbiddenHost => forbidden_host_response,
error.InvalidProtocol => invalid_protocol_response,
error.MissingHeader => missing_header_response,
else => invalid_request_response,
};
return serveHTTPResponse(server, conn, req, .{ .static = response });
};
// The 101 is ~129 bytes into an empty send buffer, so a single write
// always completes; a partial write here means the peer is already gone.
var response_buf: [160]u8 = undefined;
const response = std.fmt.bufPrint(&response_buf, "HTTP/1.1 101 Switching Protocols\r\n" ++
"Upgrade: websocket\r\n" ++
"Connection: upgrade\r\n" ++
"Sec-Websocket-Accept: {s}\r\n\r\n", .{accept_key}) catch unreachable;
const n = write(conn.socket, response) catch return error.ConnectionClosed;
if (n != response.len) {
return error.ConnectionClosed;
}
server.upgradeConnection(conn, protocol, session_id);
return .upgraded;
}
// Validate an incoming WebSocket upgrade request head and, on success, write
// the Sec-WebSocket-Accept value into `out`. Mirrors the origin/host defenses
// from the old Handshake path.
fn webSocketAccept(head: []const u8, out: *[28]u8) ![]const u8 {
const FOUND_UPGRADE: u8 = 1 << 0;
const FOUND_VERSION: u8 = 1 << 1;
const FOUND_CONNECTION: u8 = 1 << 2;
const FOUND_KEY: u8 = 1 << 3;
const FOUND_ALL = FOUND_UPGRADE | FOUND_VERSION | FOUND_CONNECTION | FOUND_KEY;
const method, _, const version, var it = header_parser.parseRequest(head) catch return error.InvalidRequest;
if (method != .get or version != .@"1.1") {
return error.InvalidProtocol;
}
var found: u8 = 0;
var key: []const u8 = "";
while (it.next() catch return error.InvalidRequest) |h| {
if (std.ascii.eqlIgnoreCase(h.key, "upgrade")) {
if (!std.ascii.eqlIgnoreCase("websocket", h.value)) return error.MissingHeader;
found |= FOUND_UPGRADE;
} else if (std.ascii.eqlIgnoreCase(h.key, "sec-websocket-version")) {
if (h.value.len != 2 or h.value[0] != '1' or h.value[1] != '3') return error.MissingHeader;
found |= FOUND_VERSION;
} else if (std.ascii.eqlIgnoreCase(h.key, "connection")) {
if (std.ascii.indexOfIgnoreCase(h.value, "upgrade") == null) return error.MissingHeader;
found |= FOUND_CONNECTION;
} else if (std.ascii.eqlIgnoreCase(h.key, "sec-websocket-key")) {
key = h.value;
found |= FOUND_KEY;
} else if (std.ascii.eqlIgnoreCase(h.key, "origin")) {
// Only a browser sends Origin, and a browser has no business
// driving CDP/BiDi: it's cross-origin to us by definition.
log.warn(.serve, "rejected websocket origin", .{ .origin = h.value[0..@min(h.value.len, 64)] });
return error.ForbiddenOrigin;
} else if (std.ascii.eqlIgnoreCase(h.key, "host")) {
// Defense in depth against DNS rebinding: only an IP literal can
// legitimately reach us (no name resolution involved). The one
// name we accept is `localhost:<port>`, which browsers hardwire
// to loopback without a lookup.
if (!std.mem.startsWith(u8, h.value, "localhost:")) {
_ = std.Io.net.IpAddress.parseLiteral(h.value) catch {
log.warn(.serve, "rejected websocket host", .{ .host = h.value[0..@min(h.value.len, 64)] });
return error.ForbiddenHost;
};
}
}
}
if (found != FOUND_ALL) {
return error.MissingHeader;
}
var sha: [20]u8 = undefined;
var hasher = std.crypto.hash.Sha1.init(.{});
hasher.update(key);
hasher.update("258EAFA5-E914-47DA-95CA-C5AB0DC85B11");
hasher.final(&sha);
_ = std.base64.standard.Encoder.encode(out, &sha);
return out;
}
src/sys/net.zig
+93 -23
View File
@@ -1,7 +1,7 @@
// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
// Pierres Tachoire <pierre@lightpanda.io>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
@@ -180,28 +180,6 @@ pub fn getsockname(sock: socket_t, addr: *posix.sockaddr, len: *posix.socklen_t)
}
}
/// pipe2 semantics; flags applied via fcntl since macOS has no pipe2.
pub fn pipe2(flags: struct { NONBLOCK: bool = false, CLOEXEC: bool = false }) ![2]posix.fd_t {
var fds: [2]posix.fd_t = undefined;
const rc = c.pipe(&fds);
if (rc != 0) {
return errnoError(c.errno(rc));
}
errdefer for (fds) |fd| {
_ = c.close(fd);
};
for (fds) |fd| {
if (flags.NONBLOCK) {
const fl = try fcntl(fd, posix.F.GETFL, 0);
_ = try fcntl(fd, posix.F.SETFL, fl | @as(u32, @bitCast(posix.O{ .NONBLOCK = true })));
}
if (flags.CLOEXEC) {
_ = try fcntl(fd, posix.F.SETFD, posix.FD_CLOEXEC);
}
}
return fds;
}
pub fn connect(addr: *const IpAddress) !socket_t {
const sock = try socket(family(addr), posix.SOCK.STREAM, posix.IPPROTO.TCP);
errdefer _ = c.close(sock);
@@ -242,6 +220,98 @@ pub fn fcntl(fd: posix.fd_t, cmd: i32, arg: usize) !usize {
return @intCast(rc);
}
pub fn close(fd: posix.fd_t) void {
switch (c.errno(c.close(fd))) {
.BADF => unreachable, // Always a race condition.
.INTR => {}, // This is still a success. See https://github.com/ziglang/zig/issues/2425
else => {},
}
}
pub fn epoll_create1(flags: u32) !i32 {
const rc = c.epoll_create1(flags);
return switch (c.errno(rc)) {
.SUCCESS => return @intCast(rc),
.INVAL => unreachable,
.MFILE => error.ProcessFdQuotaExceeded,
.NFILE => error.SystemFdQuotaExceeded,
.NOMEM => error.SystemResources,
else => error.Unexpected,
};
}
pub fn eventfd(initval: u32, flags: u32) !i32 {
const rc = c.eventfd(initval, flags);
return switch (c.errno(rc)) {
.SUCCESS => @intCast(rc),
.INVAL => unreachable, // invalid parameters
.MFILE => error.ProcessFdQuotaExceeded,
.NFILE => error.SystemFdQuotaExceeded,
.NODEV => error.SystemResources,
.NOMEM => error.SystemResources,
else => error.Unexpected,
};
}
pub fn epoll_ctl(epfd: i32, op: u32, fd: i32, event: ?*c.epoll_event) !void {
const rc = c.epoll_ctl(epfd, op, fd, event);
return switch (c.errno(rc)) {
.SUCCESS => {},
.BADF => unreachable, // always a race condition if this happens
.EXIST => error.FileDescriptorAlreadyPresentInSet,
.INVAL => unreachable,
.LOOP => error.OperationCausesCircularLoop,
.NOENT => error.FileDescriptorNotRegistered,
.NOMEM => error.SystemResources,
.NOSPC => error.UserResourceLimitReached,
.PERM => error.FileDescriptorIncompatibleWithEpoll,
else => error.Unexpected,
};
}
pub fn epoll_wait(epfd: i32, events: []c.epoll_event, timeout: i32) usize {
while (true) {
// TODO get rid of the @intCast
const rc = c.epoll_wait(epfd, events.ptr, @intCast(events.len), timeout);
switch (posix.errno(rc)) {
.SUCCESS => return @intCast(rc),
.INTR => continue,
.BADF => unreachable,
.FAULT => unreachable,
.INVAL => unreachable,
else => unreachable,
}
}
}
pub fn kqueue() !i32 {
const rc = c.kqueue();
return switch (c.errno(rc)) {
.SUCCESS => @intCast(rc),
.MFILE => error.ProcessFdQuotaExceeded,
.NFILE => error.SystemFdQuotaExceeded,
else => error.Unexpected,
};
}
pub fn kevent(kq: i32, changes: []const c.Kevent, events: []c.Kevent, timeout: ?*const c.timespec) !usize {
while (true) {
const rc = c.kevent(kq, changes.ptr, @intCast(changes.len), events.ptr, @intCast(events.len), timeout);
switch (c.errno(rc)) {
.SUCCESS => return @intCast(rc),
.INTR => continue,
.BADF => unreachable, // always a race condition if this happens
.FAULT => unreachable,
.INVAL => unreachable,
.ACCES => return error.AccessDenied,
.NOENT => return error.EventNotFound,
.NOMEM => return error.SystemResources,
.SRCH => return error.ProcessNotFound,
else => return error.Unexpected,
}
}
}
fn errnoError(e: posix.E) anyerror {
return switch (e) {
.AGAIN => error.WouldBlock,
src/testing.zig
+1 -1
View File
@@ -524,7 +524,7 @@ var test_ws_server_thread: ?std.Thread = null;
var sse_flag = std.atomic.Value(bool).init(false);
var sse_reconnect_hits = std.atomic.Value(usize).init(0);
var test_config: Config = undefined;
pub var test_config: Config = undefined;
test "tests:beforeAll" {
log.opts.level = .warn;