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

serve: Improve cdp/bidi server

Browse Source 
Significant rework of the CDP/BiDi server. There are two main changes:

1 - poll replaced with EPoll/Kqueue (1)
2 - make http serving a first class citizen

The change from poll -> epoll/kqueue isn't performance driven, it's just about
tighter code. Both epoll and kqueue let you associate arbitrary data with a
socket, so we don't need to keep arrays in sync in order to associate a socket
with a CDP by index. They both provide some event/notification mechanism, which
is cleaner than the pipe required by poll.

The poll -> epoll/kqueue change could almost have been mechanical. Making HTTP
a first class citizen is the more significant of the two changes

In `main`, a new connection always spawns a thread and, until does its own
little read loop until the connection is upgraded. This is not efficient, it
uses up a connection slot, and it's inconsistent with the final WebSocket
connection which _is_ polled off the main loop. Using up a slot means that
keepalive isn't possible, else HTTP connections would quickly use up all
available slots/threads.

This commit parses and serves HTTP requests on the main thread (safe
because none of the processing is blocking). The approach is better streamlined
for HTTP requests which never upgrade (/metrics, WebDriver) without causing
any performance overhead for those that do. It simplifies some things (e.g. an
"http" socket or a "websocket" socket is monitored and read in a similar manner
(on the main loop)). It makes other things more complicated; the flow is no
longer accept -> spawn -> upgrade -> websocket loop. It's loop -> accept -> loop
-> process -> (http | ws).

This is built ontop of the BiDi branch because (a) WebDriver is what needs
better  HTTP support and (b) some of the more mechanical changes already exist
in that branch (e.g. src/cdp/, src/server.zig -> src/server/*)

(1) kqueue landing in 2 commits from now on this branch.
This commit is contained in:
Karl Seguin committed 2026-09-03 06:47:49 +08:00
1 parent 91187e9483
commit e6241c1918
11 files changed
+2155 -1686
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/Metrics.zig
+6 -2
View File
@@ -20,8 +20,10 @@ 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: CounterEnum("reason", enum { first_request, idle }) = .{},
serve_connections: CounterEnum("driver", Driver) = .{},
serve_connection_limit: Counter = .{},
serve_active_connections: GaugeEnum("driver", Driver) = .{},
@@ -92,8 +94,10 @@ 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 exceeding a deadline; first_request never completed a request, idle is a keepalive connection that went quiet",
.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/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/Connection.zig → src/server/Link.zig
+85 -86
View File
@@ -21,44 +21,43 @@ 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 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;
pub const Connection = @This();
// 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();
// 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),
protocol: Driver.Protocol,
reader: WS.Reader,
send_arena: ArenaAllocator,
pub fn init(
self: *Connection,
self: *Link,
app: *App,
socket: posix.socket_t,
protocol: Protocol,
protocol: Driver.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 nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
lp.assert(socket_flags & nonblocking == nonblocking, "Link.init blocking", .{});
}
const config = app.config;
@@ -75,19 +74,17 @@ pub fn init(
};
}
pub fn deinit(self: *Connection) void {
pub fn deinit(self: *Link) void {
self.reader.deinit();
self.send_arena.deinit();
}
pub fn send(self: *Connection, data: []const u8) !void {
pub fn send(self: *Link, 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 });
};
@@ -96,15 +93,12 @@ pub fn send(self: *Connection, data: []const u8) !void {
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", .{});
// The socket is nonblocking so loop reads never stall. Writes
// are simpler if we can block: no per-connection pending-write
// queue with its own allocations. On WouldBlock we flip the
// socket to blocking for this write and flip it back after.
// Should virtually never happen.
lp.assert(changed_to_blocking == false, "Link 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;
@@ -119,7 +113,7 @@ pub fn send(self: *Connection, data: []const u8) !void {
}
}
pub fn sendPong(self: *Connection, data: []const u8) !void {
pub fn sendPong(self: *Link, data: []const u8) !void {
if (data.len == 0) {
return self.send(&WS.EMPTY_PONG);
}
@@ -133,61 +127,74 @@ pub fn sendPong(self: *Connection, data: []const u8) !void {
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 {
// 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);
// 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`.
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 fn feed(self: *Connection, data: []const u8) !bool {
var remaining = data;
while (remaining.len > 0) {
// we copy what will fit into our read buffer
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 {
var pushed = false;
var remaining = budget;
while (remaining > 0) {
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;
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;
}
remaining = remaining[used..];
}
return true;
return .{ .keep = true, .pushed = pushed };
}
// Framing-only iteration over received bytes. Will process as many messages
// as are buffered.
fn processMessages(self: *Connection) !bool {
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),
.ping, .text, .binary => try self.handleMessage(msg, pushed),
.close => blk: {
_ = try self.handleMessage(msg);
_ = try self.handleMessage(msg, pushed);
break :blk false;
},
};
@@ -195,52 +202,46 @@ fn processMessages(self: *Connection) !bool {
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 {
fn handleMessage(self: *Link, msg: WS.Message, pushed: *bool) !bool {
switch (msg.type) {
.text, .binary => return switch (self.protocol) {
.cdp => self.pushCdp(msg.data),
.bidi => self.pushBiDi(msg.data),
.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 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?
// 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(),
@@ -248,27 +249,25 @@ fn pushCdp(self: *Connection, bytes: []const u8) !bool {
.{ .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,
} });
self.inbox.push(arena, .{ .cdp = .{ .raw = raw, .input = input } });
pushed.* = true;
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 {
// 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;
}
pub fn shutdown(self: *Connection) void {
// Called from the worker (Driver.shutdown) to break the loop's read.
pub fn shutdown(self: *Link) void {
sys_net.shutdown(self.socket, .recv) catch {};
}
src/server/Server.zig
+987 -930
View File
File diff suppressed because it is too large. Load diff
src/network/WS.zig → src/server/WS.zig
+10 -7
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
@@ -537,7 +540,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.
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
+934
View File
@@ -0,0 +1,934 @@
// 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,
// Whether at least one request has been answered. A deadline miss before
// that is a client that connected and never spoke; after, an idle keepalive.
served: bool,
// 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
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,
};
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.served = false;
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,
.served = false,
.pending = null,
.state = .header,
.buffer = try .init(allocator, 4096),
};
return conn;
}
fn destroy(self: *Pool, conn: *Connection) void {
conn.deinit();
self.allocator.destroy(conn);
}
};
};
// How long a keepalive connection may sit without a complete request before
// we close it.
const IDLE_TIMEOUT_MS = 10_000;
pub const FIRST_TIMEOUT_MS = 5_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.
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 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 => {
log.warn(.serve, "serve error", .{ .err = err });
return;
},
};
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.served = true;
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
+65 -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,70 @@ 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,
}
}
}
fn errnoError(e: posix.E) anyerror {
return switch (e) {
.AGAIN => error.WouldBlock,