nxdns/src/web/server_integration_test.zig

//! Loopback tests for `server.zig` and `router.zig`.
//!
//! This lives in its own file because it needs `@import("build_options")`,
//! which only exists when the compilation is driven by build.zig. The body is
//! compiled by every `zig build test` run, so it cannot rot, and skips at run
//! time unless `-Dintegration` is passed.
//!
//! Hermetic: one listener and one or two clients on 127.0.0.1, handlers that
//! touch nothing but the request. No stream read in 0.16.0 takes a timeout, so
//! the whole client side of each test runs as one task raced against a budget
//! and nothing can hang.

const std = @import("std");
const build_options = @import("build_options");
const net = std.Io.net;

const http_util = @import("http_util.zig");
const router = @import("router.zig");
const server = @import("server.zig");

const testing = std.testing;

const budget: std.Io.Clock.Duration = .{ .raw = .fromSeconds(5), .clock = .awake };

/// Long enough that a loopback round trip cannot lose to scheduling, short
/// enough that the cancellation test stays quick.
const settle: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(200), .clock = .awake };

fn okHandler(
    state: *server.WebState,
    io: std.Io,
    request: *http_util.Request,
) http_util.HandlerError!void {
    _ = state;
    _ = io;
    return http_util.respondBytes(request, .ok, "pong", http_util.content_type_text, &.{});
}

/// Echoes the body length back, so a test can prove the body arrived whole and
/// that the cap fires before a handler ever sees an oversize one.
fn echoLengthHandler(
    state: *server.WebState,
    io: std.Io,
    request: *http_util.Request,
) http_util.HandlerError!void {
    _ = state;
    _ = io;
    const body = http_util.readBody(request) catch |err| switch (err) {
        error.TooLarge => return http_util.respondError(request, .payload_too_large, "body too large"),
        error.OutOfMemory => return error.OutOfMemory,
        error.ReadFailed => return error.WriteFailed,
        error.WriteFailed => return error.WriteFailed,
        error.HttpExpectationFailed => return error.HttpExpectationFailed,
    };
    var buf: [32]u8 = undefined;
    const text = std.fmt.bufPrint(&buf, "{d}", .{body.len}) catch unreachable;
    return http_util.respondBytes(request, .ok, text, http_util.content_type_text, &.{});
}

/// Answers with the decoded query value, proving the router hands handlers a
/// target copy that survives the head being invalidated.
fn echoDomainHandler(
    state: *server.WebState,
    io: std.Io,
    request: *http_util.Request,
) http_util.HandlerError!void {
    _ = state;
    _ = io;
    var buf: [http_util.max_query_value_len]u8 = undefined;
    const value = http_util.queryValue(request.query, "domain", &buf) catch {
        return http_util.respondError(request, .bad_request, "bad query");
    } orelse "";
    return http_util.respondBytes(request, .ok, value, http_util.content_type_text, &.{});
}

/// Reads the body first and only then looks at the path, which is exactly the
/// order that would break without the head copy (ruling 25).
fn bodyThenPathHandler(
    state: *server.WebState,
    io: std.Io,
    request: *http_util.Request,
) http_util.HandlerError!void {
    _ = state;
    _ = io;
    _ = http_util.readBody(request) catch return error.WriteFailed;
    var buf: [64]u8 = undefined;
    var writer: std.Io.Writer = .fixed(&buf);
    writer.print("{s}|{?d}", .{ request.raw_path, request.id }) catch unreachable;
    return http_util.respondBytes(request, .ok, writer.buffered(), http_util.content_type_text, &.{});
}

const test_routes = [_]router.RouteInfo{
    .{ .method = .GET, .pattern = "/api/health", .auth = .open, .policy = .read, .handler = okHandler, .rate_limit = .exempt },
    .{ .method = .GET, .pattern = "/api/groups", .auth = .session, .policy = .read, .handler = okHandler },
    .{ .method = .POST, .pattern = "/api/groups", .auth = .session, .policy = .config_write, .handler = echoLengthHandler },
    .{ .method = .PUT, .pattern = "/api/groups/{id}", .auth = .session, .policy = .config_write, .handler = bodyThenPathHandler },
    .{ .method = .GET, .pattern = "/api/lookup", .auth = .open, .policy = .read, .handler = echoDomainHandler },
};

fn denyAll(state: *server.WebState, io: std.Io, request: *const http_util.Request) bool {
    _ = state;
    _ = io;
    _ = request;
    return false;
}

fn alwaysLimited(state: *server.WebState, io: std.Io, request: *const http_util.Request) server.LimitVerdict {
    _ = state;
    _ = io;
    _ = request;
    return .{ .allowed = false, .retry_after_s = 42 };
}

fn testState(gpa: std.mem.Allocator) server.WebState {
    return .{
        .gpa = gpa,
        .routes = &test_routes,
        .check_auth = server.allowAll,
        .check_limit = server.neverLimit,
    };
}

const Outcome = union(enum) {
    work: anyerror!void,
    expiry: std.Io.Cancelable!void,
};

fn expire(io: std.Io, duration: std.Io.Clock.Duration) std.Io.Cancelable!void {
    return duration.sleep(io);
}

/// Runs the client side under a budget so a server that never answers fails the
/// test instead of hanging the run.
fn bounded(io: std.Io, comptime f: anytype, args: std.meta.ArgsTuple(@TypeOf(f))) !void {
    var outcomes: [2]Outcome = undefined;
    var race: std.Io.Select(Outcome) = .init(io, &outcomes);
    defer race.cancelDiscard();

    try race.concurrent(.work, f, args);
    try race.concurrent(.expiry, expire, .{ io, budget });

    switch (try race.await()) {
        .work => |result| return result,
        .expiry => |result| {
            try result;
            return error.TestTimedOut;
        },
    }
}

/// One open connection with a reader and a writer, which is all these tests
/// need of an HTTP client.
const Conn = struct {
    stream: net.Stream,
    reader: net.Stream.Reader,
    writer: net.Stream.Writer,
    read_buf: [8192]u8 = undefined,
    write_buf: [4096]u8 = undefined,
    /// Header lines are copied here because each `takeDelimiterInclusive`
    /// invalidates the previous line's slice into the read buffer.
    head_buf: [4096]u8 = undefined,

    fn connect(self: *Conn, io: std.Io, address: net.IpAddress) !void {
        self.stream = try address.connect(io, .{ .mode = .stream });
        self.reader = self.stream.reader(io, &self.read_buf);
        self.writer = self.stream.writer(io, &self.write_buf);
    }

    fn close(self: *Conn, io: std.Io) void {
        self.stream.close(io);
    }

    fn send(self: *Conn, request: []const u8) !void {
        try self.writer.interface.writeAll(request);
        try self.writer.interface.flush();
    }

    /// Reads one response: head to the blank line, then exactly
    /// `content-length` bytes. Every response these tests provoke carries one.
    fn receive(self: *Conn, out: []u8) !Response {
        var head_len: usize = 0;
        while (true) {
            const raw = try self.reader.interface.takeDelimiterInclusive('\n');
            const line = std.mem.trimEnd(u8, raw, "\r\n");
            if (line.len == 0) break;
            if (head_len + line.len + 1 > self.head_buf.len) return error.TestHeadTooLarge;
            @memcpy(self.head_buf[head_len..][0..line.len], line);
            head_len += line.len;
            self.head_buf[head_len] = '\n';
            head_len += 1;
        }
        const head = self.head_buf[0..head_len];
        const status = try parseStatus(head);
        const length = try contentLength(head);
        if (length > out.len) return error.TestResponseTooLarge;
        const body = out[0..length];
        try self.reader.interface.readSliceAll(body);
        return .{ .status = status, .head = head, .body = body };
    }
};

const Response = struct {
    status: u16,
    /// Borrows the connection's read buffer; valid until the next receive.
    head: []const u8,
    body: []const u8,

    fn header(self: Response, name: []const u8) ?[]const u8 {
        var lines = std.mem.splitScalar(u8, self.head, '\n');
        _ = lines.next();
        while (lines.next()) |line| {
            const colon = std.mem.findScalar(u8, line, ':') orelse continue;
            if (!std.ascii.eqlIgnoreCase(std.mem.trim(u8, line[0..colon], " "), name)) continue;
            return std.mem.trim(u8, line[colon + 1 ..], " ");
        }
        return null;
    }
};

fn parseStatus(head: []const u8) !u16 {
    const first_space = std.mem.findScalar(u8, head, ' ') orelse return error.TestBadResponse;
    const rest = head[first_space + 1 ..];
    const second_space = std.mem.findScalar(u8, rest, ' ') orelse rest.len;
    return std.fmt.parseInt(u16, rest[0..second_space], 10) catch error.TestBadResponse;
}

fn contentLength(head: []const u8) !usize {
    var lines = std.mem.splitScalar(u8, head, '\n');
    while (lines.next()) |line| {
        const colon = std.mem.findScalar(u8, line, ':') orelse continue;
        if (!std.ascii.eqlIgnoreCase(std.mem.trim(u8, line[0..colon], " "), "content-length")) continue;
        return std.fmt.parseInt(usize, std.mem.trim(u8, line[colon + 1 ..], " "), 10) catch error.TestBadResponse;
    }
    return error.TestNoContentLength;
}

fn get(path: []const u8, buf: []u8) []const u8 {
    return std.fmt.bufPrint(buf, "GET {s} HTTP/1.1\r\nhost: t\r\n\r\n", .{path}) catch unreachable;
}

/// The listener's counters, read once after `f` finished and before the
/// listener is torn down. Plain values, because they are a report of a run that
/// is over: the shared core's counters and the web listener's own one land in
/// the same struct here.
const Counters = struct {
    connections: u64,
    rejected_at_capacity: u64,
    rejected_at_shutdown: u64,
    accept_errors: u64,
    connection_errors: u64,
    requests: u64,
};

/// Starts a listener on 127.0.0.1:0 with `state` and runs `f` against it under
/// the budget, then shuts the listener down through the drain path.
fn withServer(
    gpa: std.mem.Allocator,
    io: std.Io,
    state: *server.WebState,
    max_connections: u16,
    comptime f: anytype,
    extra: anytype,
) !Counters {
    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
    var web = try server.Server.listen(gpa, io, listen_address, state, .{ .max_connections = max_connections });
    const address = web.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, server.Server.serve, .{ &web, io });

    const result = bounded(io, f, .{ io, address } ++ extra);

    const stats: Counters = .{
        .connections = web.core.stats.connections.load(.monotonic),
        .rejected_at_capacity = web.core.stats.rejected_at_capacity.load(.monotonic),
        .rejected_at_shutdown = web.core.stats.rejected_at_shutdown.load(.monotonic),
        .accept_errors = web.core.stats.accept_errors.load(.monotonic),
        .connection_errors = web.core.stats.connection_errors.load(.monotonic),
        .requests = web.stats.requests.load(.monotonic),
    };

    web.deinit(io);
    group.await(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };

    try result;
    return stats;
}

fn twoRequestsOnOneConnection(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var body_buf: [256]u8 = undefined;
    for (0..2) |_| {
        var request_buf: [128]u8 = undefined;
        try conn.send(get("/api/health", &request_buf));
        const response = try conn.receive(&body_buf);
        try testing.expectEqual(@as(u16, 200), response.status);
        try testing.expectEqualStrings("pong", response.body);
    }
}

test "one connection carries two requests" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    const stats = try withServer(gpa, io, &state, 4, twoRequestsOnOneConnection, .{});

    // One accept for two requests is the whole point of keep-alive.
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 2), stats.requests);
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);
}

fn routingMatrix(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var body_buf: [512]u8 = undefined;
    var request_buf: [256]u8 = undefined;

    try conn.send(get("/api/nope", &request_buf));
    var response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 404), response.status);
    try testing.expectEqualStrings("{\"error\":\"not found\"}", response.body);

    try conn.send("DELETE /api/groups HTTP/1.1\r\nhost: t\r\n\r\n");
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 405), response.status);
    try testing.expectEqualStrings("GET, POST", response.header("allow").?);

    // '+' is a space, %2E is a literal dot: both survive the round trip.
    try conn.send(get("/api/lookup?domain=a+b%2Ecom", &request_buf));
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);
    try testing.expectEqualStrings("a b.com", response.body);

    // A truncated escape is a 400, not a value with a stray percent in it.
    try conn.send(get("/api/lookup?domain=abc%2", &request_buf));
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 400), response.status);

    // A path deeper than the segment budget is refused before matching.
    try conn.send(get("/1/2/3/4/5/6/7/8/9", &request_buf));
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 400), response.status);
}

test "routing answers 404, 405 with allow, and rejects malformed targets" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    const stats = try withServer(gpa, io, &state, 4, routingMatrix, .{});
    try testing.expectEqual(@as(u64, 5), stats.requests);
}

fn postBody(io: std.Io, address: net.IpAddress, length: usize, expected_status: u16) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var head_buf: [128]u8 = undefined;
    const head = try std.fmt.bufPrint(
        &head_buf,
        "POST /api/groups HTTP/1.1\r\nhost: t\r\ncontent-length: {d}\r\n\r\n",
        .{length},
    );
    try conn.writer.interface.writeAll(head);

    const chunk = [_]u8{'x'} ** 4096;
    var sent: usize = 0;
    while (sent < length) {
        const n = @min(chunk.len, length - sent);
        // A refused body ends the connection, so the tail of a rejected write
        // is expected to fail; the response is what the test reads.
        conn.writer.interface.writeAll(chunk[0..n]) catch break;
        sent += n;
    }
    conn.writer.interface.flush() catch {};

    var body_buf: [256]u8 = undefined;
    const response = try conn.receive(&body_buf);
    try testing.expectEqual(expected_status, response.status);
}

test "a body inside the cap is delivered whole" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    _ = try withServer(gpa, io, &state, 4, postBody, .{ @as(usize, 64 * 1024), @as(u16, 200) });
}

test "a body over the cap is 413, not a buffered megabyte" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    _ = try withServer(
        gpa,
        io,
        &state,
        4,
        postBody,
        .{ http_util.max_body_bytes + 1, @as(u16, 413) },
    );
}

fn postWithoutLength(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var body_buf: [256]u8 = undefined;
    try conn.send("POST /api/groups HTTP/1.1\r\nhost: t\r\n\r\n");
    const response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);
    try testing.expectEqualStrings("0", response.body);

    // A fresh connection proves the listener outlived the request; before the
    // head normalization it died on http/Server.zig:631's assert.
    var second: Conn = undefined;
    try second.connect(io, address);
    defer second.close(io);

    var request_buf: [128]u8 = undefined;
    try second.send(get("/api/health", &request_buf));
    const again = try second.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), again.status);
}

test "a POST with no content-length and no transfer-encoding is an empty body, not a crash" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    const stats = try withServer(gpa, io, &state, 4, postWithoutLength, .{});
    try testing.expectEqual(@as(u64, 0), stats.connection_errors);
}

fn bodyThenTarget(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    try conn.send("PUT /api/groups/17 HTTP/1.1\r\nhost: t\r\ncontent-length: 4\r\n\r\nabcd");

    var body_buf: [128]u8 = undefined;
    const response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);
    try testing.expectEqualStrings("/api/groups/17|17", response.body);
}

test "the target survives a body read" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    _ = try withServer(gpa, io, &state, 4, bodyThenTarget, .{});
}

fn refusedOverCapacity(io: std.Io, address: net.IpAddress) anyerror!void {
    // Hold the only slot with an idle keep-alive connection, so the second
    // client meets a full table rather than a race.
    var held: Conn = undefined;
    try held.connect(io, address);
    defer held.close(io);

    var request_buf: [128]u8 = undefined;
    var body_buf: [256]u8 = undefined;
    try held.send(get("/api/health", &request_buf));
    const first = try held.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), first.status);

    var overflow: Conn = undefined;
    try overflow.connect(io, address);
    defer overflow.close(io);

    try overflow.send(get("/api/health", &request_buf));
    const refused = try overflow.receive(&body_buf);
    try testing.expectEqual(@as(u16, 503), refused.status);
    try testing.expectEqualStrings("{\"error\":\"too many connections\"}", refused.body);
}

test "a connection over the cap is told 503, not silently dropped" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    const stats = try withServer(gpa, io, &state, 1, refusedOverCapacity, .{});
    try testing.expectEqual(@as(u64, 1), stats.connections);
    try testing.expectEqual(@as(u64, 1), stats.rejected_at_capacity);
}

fn deniedAndLimited(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var request_buf: [128]u8 = undefined;
    var body_buf: [256]u8 = undefined;

    // The limiter runs before authentication, so a limited request is 429 even
    // though the same request would also have failed the session check.
    try conn.send(get("/api/groups", &request_buf));
    var response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 429), response.status);
    try testing.expectEqualStrings("42", response.header("retry-after").?);

    // Ruling 19: the monitoring endpoints are exempt and answer normally.
    try conn.send(get("/api/health", &request_buf));
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);
}

test "the limiter and the session check are applied in that order" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    state.check_auth = denyAll;
    state.check_limit = alwaysLimited;
    _ = try withServer(gpa, io, &state, 4, deniedAndLimited, .{});
}

fn unauthenticated(io: std.Io, address: net.IpAddress) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var request_buf: [128]u8 = undefined;
    var body_buf: [256]u8 = undefined;

    try conn.send(get("/api/groups", &request_buf));
    var response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 401), response.status);

    // An open route stays reachable so the SPA shell can show a login form.
    try conn.send(get("/api/health", &request_buf));
    response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);
}

test "a session route without a session is 401 and an open route is not" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    state.check_auth = denyAll;
    _ = try withServer(gpa, io, &state, 4, unauthenticated, .{});
}

/// Opens a connection, answers one request on it, and then leaves it idle and
/// open — the shape a browser tab holds, and the one that must not be able to
/// stall shutdown.
/// Returns plain `void`, not an error union: a group task must be coercible to
/// `Cancelable!void`, so the outcome travels in `failed` instead.
fn holdIdleConnection(io: std.Io, address: net.IpAddress, opened: *std.Io.Event, failed: *bool) void {
    holdIdleConnectionInner(io, address, opened) catch {
        failed.* = true;
        opened.set(io);
    };
}

fn holdIdleConnectionInner(io: std.Io, address: net.IpAddress, opened: *std.Io.Event) anyerror!void {
    var conn: Conn = undefined;
    try conn.connect(io, address);
    defer conn.close(io);

    var request_buf: [128]u8 = undefined;
    var body_buf: [256]u8 = undefined;
    try conn.send(get("/api/health", &request_buf));
    const response = try conn.receive(&body_buf);
    try testing.expectEqual(@as(u16, 200), response.status);

    opened.set(io);
    // Nothing more is sent. The connection sits in `receiveHead`, which is
    // where cancellation has to reach it.
    settle.sleep(io) catch {};
}

test "cancellation returns promptly with an idle keep-alive connection open" {
    if (!build_options.integration) return error.SkipZigTest;

    const gpa = testing.allocator;
    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var state = testState(gpa);
    const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
    var web = try server.Server.listen(gpa, io, listen_address, &state, .{ .max_connections = 4 });
    const address = web.boundAddress();

    var group: std.Io.Group = .init;
    try group.concurrent(io, server.Server.serve, .{ &web, io });

    var opened: std.Io.Event = .unset;
    var failed = false;
    var client: std.Io.Group = .init;
    try client.concurrent(io, holdIdleConnection, .{ io, address, &opened, &failed });
    opened.wait(io) catch |err| switch (err) {
        error.Canceled => unreachable,
    };
    try testing.expect(!failed);

    // The listener task is canceled with a client parked in `receiveHead`. If
    // the cancel path awaited the connection group instead of canceling it,
    // this would block until the client hung up, which the budget below would
    // catch as a failure.
    const start = std.Io.Clock.awake.now(io);
    group.cancel(io);
    const elapsed = start.durationTo(std.Io.Clock.awake.now(io));

    client.cancel(io);
    web.deinit(io);

    try testing.expect(elapsed.toMilliseconds() < settle.raw.toMilliseconds());
}