From 562c25d0ff0c84d9676300a737c03d9b6eae21a6 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 7 Jun 2026 20:27:19 -0500 Subject: [PATCH] refactor: extract util/ and atproto/ packages from sprawling src root MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit slurper.zig was carrying ~160 lines of generic parsing/network primitives (hostname validation, SSRF checks, raw HTTP header parsing) that had nothing to do with crawl orchestration. several helpers were also copy-pasted across modules — getenv/parseEnvInt in 4 files, timestamp variants in 5+. centralize into two packages, mirroring tigerbeetle's stdx and ghostty's thematic-package conventions: - src/util/ — std extensions (env, time), flat namespace, no domain knowledge - src/atproto/ — relay domain logic (hostname parsing+validation, host_check describeServer/SSRF/relay-loop detection) folds the duplicate hostname parser (slurper + validator each had one) into a single atproto.extractHostFromUrl/validateHostname. slurper drops to pure orchestration. call sites keep old names via file-local aliases, so no churn. docs/design.md gains a "code organization" section defining where new helpers go. .claude/ harness state added to .gitignore. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 3 + build.zig | 2 + docs/design.md | 19 +++ src/api/admin.zig | 28 +---- src/atproto/host_check.zig | 126 +++++++++++++++++++ src/atproto/hostname.zig | 157 ++++++++++++++++++++++++ src/atproto/main.zig | 16 +++ src/broadcaster.zig | 7 +- src/event_log.zig | 7 +- src/frame_worker.zig | 6 +- src/host_ops.zig | 6 +- src/main.zig | 15 +-- src/slurper.zig | 241 +------------------------------------ src/subscriber.zig | 22 +--- src/util/env.zig | 18 +++ src/util/time.zig | 49 ++++++++ src/util/util.zig | 20 +++ src/validator.zig | 41 +------ 18 files changed, 443 insertions(+), 340 deletions(-) create mode 100644 src/atproto/host_check.zig create mode 100644 src/atproto/hostname.zig create mode 100644 src/atproto/main.zig create mode 100644 src/util/env.zig create mode 100644 src/util/time.zig create mode 100644 src/util/util.zig diff --git a/.gitignore b/.gitignore index f5053b9..f2045df 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,6 @@ data/ # local scratch (content migrated to relay repo ops-changelog) TODO.md EXPERIMENTS.md + +# claude code harness state (session locks, local config) +.claude/ diff --git a/build.zig b/build.zig index 71aa86b..3711111 100644 --- a/build.zig +++ b/build.zig @@ -78,6 +78,8 @@ pub fn build(b: *std.Build) void { "src/backfill.zig", "src/thread_pool.zig", "src/frame_worker.zig", + "src/util/util.zig", + "src/atproto/main.zig", }; inline for (test_files) |file| { const test_mod = b.createModule(.{ diff --git a/docs/design.md b/docs/design.md index 612db8f..37797fe 100644 --- a/docs/design.md +++ b/docs/design.md @@ -3,6 +3,25 @@ an AT Protocol relay that crawls PDS instances directly, validates commit signatures, and rebroadcasts to downstream consumers over WebSocket. +## code organization + +`src/` root holds the domain modules (one per pipeline stage: `slurper`, +`subscriber`, `frame_worker`, `validator`, `broadcaster`, `event_log`, …). +shared helpers live in two packages so they don't get re-defined per module: + +- `src/util/` — extensions to the standard library: things that could have + been in `std` but aren't (`env`, `time`). flat namespace (`util.timestamp`, + not `util.time.timestamp`); no domain knowledge. add here when a helper is + generic and would make sense in any zig project. +- `src/atproto/` — AT Protocol / relay domain logic shared across modules + (`hostname` parsing + validation, `host_check` describeServer/SSRF/relay-loop + detection). these encode relay *policy* (mirroring indigo), so they are not + `util`. each package's `main.zig` doc-comment defines its scope — that scope + is what decides where a new helper goes. + +if a helper is being copy-pasted into a third module, that's the signal to +move it into one of these two packages rather than re-declaring it. + ## data flow ``` diff --git a/src/api/admin.zig b/src/api/admin.zig index 7dce94b..fba9397 100644 --- a/src/api/admin.zig +++ b/src/api/admin.zig @@ -15,8 +15,11 @@ const event_log_mod = @import("../event_log.zig"); const backfill_mod = @import("../backfill.zig"); const cleaner_mod = @import("../cleaner.zig"); const resync_mod = @import("../resync.zig"); +const util = @import("../util/util.zig"); const log = std.log.scoped(.relay); +const getenv = util.getenv; +const formatTimestamp = util.formatTimestamp; const HttpContext = router.HttpContext; const DbRequest = event_log_mod.DbRequest; @@ -506,28 +509,3 @@ fn buildAccountFrame(allocator: std.mem.Allocator, did: []const u8) ?[]const u8 return frame; } - -fn formatTimestamp(buf: *[24]u8) []const u8 { - var tp: std.c.timespec = undefined; - _ = std.c.clock_gettime(.REALTIME, &tp); - const ts: u64 = @intCast(tp.sec); - const es = std.time.epoch.EpochSeconds{ .secs = ts }; - const day = es.getEpochDay(); - const yd = day.calculateYearDay(); - const md = yd.calculateMonthDay(); - const ds = es.getDaySeconds(); - - return std.fmt.bufPrint(buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}Z", .{ - yd.year, - @as(u32, @intFromEnum(md.month)) + 1, - @as(u32, md.day_index) + 1, - ds.getHoursIntoDay(), - ds.getMinutesIntoHour(), - ds.getSecondsIntoMinute(), - }) catch "1970-01-01T00:00:00Z"; -} - -fn getenv(key: [*:0]const u8) ?[]const u8 { - const ptr = std.c.getenv(key) orelse return null; - return std.mem.sliceTo(ptr, 0); -} diff --git a/src/atproto/host_check.zig b/src/atproto/host_check.zig new file mode 100644 index 0000000..a1a1c07 --- /dev/null +++ b/src/atproto/host_check.zig @@ -0,0 +1,126 @@ +//! host_check — liveness & safety checks for a candidate PDS host. +//! +//! mirrors the Go relay's host_checker.go (describeServer probe), ssrf.go +//! (private-IP rejection), and slurper.go (relay-loop detection via the +//! Server header). relay policy, so it lives under atproto/. + +const std = @import("std"); +const Io = std.Io; +const http = std.http; +const Allocator = std.mem.Allocator; +const hostname_mod = @import("hostname.zig"); + +const HostValidationError = hostname_mod.HostValidationError; +const log = std.log.scoped(.relay); + +/// check that a host is a real PDS by calling describeServer. +/// also checks the Server header for relay-loop detection. +/// Go relay: host_checker.go CheckHost + slurper.go Server header check. +pub fn checkHost(allocator: Allocator, host: []const u8, io: Io) HostValidationError!void { + // SSRF protection: reject private IPs before making any request + try rejectPrivateHost(allocator, host); + var url_buf: [512]u8 = undefined; + const url = std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.server.describeServer", .{host}) catch return error.HostUnreachable; + + var client: http.Client = .{ .allocator = allocator, .io = io }; + defer client.deinit(); + + const uri = std.Uri.parse(url) catch return error.HostUnreachable; + var req = client.request(.GET, uri, .{}) catch return error.HostUnreachable; + defer req.deinit(); + req.sendBodiless() catch return error.HostUnreachable; + + var redirect_buf: [2048]u8 = undefined; + const response = req.receiveHead(&redirect_buf) catch return error.HostUnreachable; + + if (response.head.status != .ok) return error.NotAPds; + + // relay loop detection: check Server header for "atproto-relay" + // Go relay: slurper.go — auto-bans hosts whose Server header contains "atproto-relay" + if (findHeaderInRaw(response.head.bytes, "server")) |server_val| { + if (std.mem.indexOf(u8, server_val, "atproto-relay") != null) { + return error.IsARelay; + } + } +} + +/// SSRF protection: resolve hostname and reject private/reserved IP ranges. +/// Go relay: ssrf.go PublicOnlyTransport — rejects 10/8, 172.16/12, 192.168/16, 127/8, link-local. +pub fn rejectPrivateHost(allocator: Allocator, host: []const u8) HostValidationError!void { + // null-terminate hostname for getaddrinfo + const hostname_z = allocator.dupeZ(u8, host) catch return error.HostUnreachable; + defer allocator.free(hostname_z); + + var hints: std.c.addrinfo = .{ + .flags = .{}, + .family = std.c.AF.UNSPEC, + .socktype = std.c.SOCK.STREAM, + .protocol = 0, + .addrlen = 0, + .addr = null, + .canonname = null, + .next = null, + }; + + var res: ?*std.c.addrinfo = null; + const rc = std.c.getaddrinfo(hostname_z, "443", &hints, &res); + if (@intFromEnum(rc) != 0 or res == null) return error.HostUnreachable; + defer std.c.freeaddrinfo(res.?); + + // check all resolved addresses — reject if ANY is private + var cur = res; + var found_any = false; + while (cur) |node| : (cur = node.next) { + found_any = true; + if (node.family == std.c.AF.INET) { + const sa: *const std.c.sockaddr.in = @ptrCast(@alignCast(node.addr.?)); + const bytes: [4]u8 = @bitCast(sa.addr); + if (bytes[0] == 10 or // 10.0.0.0/8 + (bytes[0] == 172 and (bytes[1] & 0xf0) == 16) or // 172.16.0.0/12 + (bytes[0] == 192 and bytes[1] == 168) or // 192.168.0.0/16 + bytes[0] == 127 or // 127.0.0.0/8 + bytes[0] == 0 or // 0.0.0.0/8 + (bytes[0] == 169 and bytes[1] == 254)) // 169.254.0.0/16 link-local + { + log.warn("SSRF: {s} resolves to private IP {d}.{d}.{d}.{d}", .{ host, bytes[0], bytes[1], bytes[2], bytes[3] }); + return error.HostUnreachable; + } + } + // allow IPv6 for now (could add RFC 4193 check later) + } + if (!found_any) return error.HostUnreachable; +} + +/// search raw HTTP headers for a header by name (case-insensitive). +/// returns the trimmed value, or null if not found. +pub fn findHeaderInRaw(raw: []const u8, name: []const u8) ?[]const u8 { + var it = std.mem.splitSequence(u8, raw, "\r\n"); + _ = it.next(); // skip status line + while (it.next()) |line| { + if (line.len == 0) break; + const colon = std.mem.indexOfScalar(u8, line, ':') orelse continue; + const key = std.mem.trim(u8, line[0..colon], " "); + if (key.len != name.len) continue; + // case-insensitive compare + var match = true; + for (key, name) |a, b| { + if (std.ascii.toLower(a) != std.ascii.toLower(b)) { + match = false; + break; + } + } + if (match) { + return std.mem.trim(u8, line[colon + 1 ..], " "); + } + } + return null; +} + +// --- tests --- + +test "findHeaderInRaw is case-insensitive and trims" { + const raw = "HTTP/1.1 200 OK\r\nServer: atproto-relay/1.0\r\nContent-Type: application/json\r\n\r\n"; + try std.testing.expectEqualStrings("atproto-relay/1.0", findHeaderInRaw(raw, "server").?); + try std.testing.expectEqualStrings("application/json", findHeaderInRaw(raw, "content-type").?); + try std.testing.expect(findHeaderInRaw(raw, "x-missing") == null); +} diff --git a/src/atproto/hostname.zig b/src/atproto/hostname.zig new file mode 100644 index 0000000..164e55d --- /dev/null +++ b/src/atproto/hostname.zig @@ -0,0 +1,157 @@ +//! hostname — relay host parsing & validation. +//! +//! mirrors indigo relay's ParseHostname rules (reject IPs, ports, localhost, +//! malformed DNS names). this is relay policy, not generic DNS validation, +//! which is why it lives under atproto/ rather than util/. + +const std = @import("std"); +const Allocator = std.mem.Allocator; + +pub const HostValidationError = error{ + EmptyHostname, + InvalidCharacter, + InvalidLabel, + TooFewLabels, + LooksLikeIpAddress, + PortNotAllowed, + LocalhostNotAllowed, + DomainBanned, + HostUnreachable, + NotAPds, + IsARelay, +}; + +/// validate and normalize a hostname for crawling. +/// rejects IPs, ports, localhost, invalid DNS names. +/// returns lowercased hostname on success (caller owns the allocation). +pub fn validateHostname(allocator: Allocator, raw: []const u8) HostValidationError![]u8 { + if (raw.len == 0) return error.EmptyHostname; + + // strip scheme if present + var hostname = raw; + for ([_][]const u8{ "https://", "http://", "wss://", "ws://" }) |scheme| { + if (hostname.len > scheme.len and std.ascii.startsWithIgnoreCase(hostname, scheme)) { + hostname = hostname[scheme.len..]; + break; + } + } + + // strip trailing slash and path + if (std.mem.indexOfScalar(u8, hostname, '/')) |i| { + hostname = hostname[0..i]; + } + + // reject ports (Go relay rejects non-localhost ports) + if (std.mem.indexOfScalar(u8, hostname, ':')) |_| { + return error.PortNotAllowed; + } + + // reject localhost + if (std.ascii.eqlIgnoreCase(hostname, "localhost")) { + return error.LocalhostNotAllowed; + } + + // validate characters and split into labels + var label_count: usize = 0; + var all_labels_numeric = true; + var it = std.mem.splitScalar(u8, hostname, '.'); + while (it.next()) |label| { + if (label.len == 0 or label.len > 63) return error.InvalidLabel; + // labels must be alphanumeric with hyphens, no leading/trailing hyphens + if (label[0] == '-' or label[label.len - 1] == '-') return error.InvalidLabel; + var is_numeric = true; + for (label) |c| { + if (!std.ascii.isAlphanumeric(c) and c != '-') return error.InvalidCharacter; + if (!std.ascii.isDigit(c)) is_numeric = false; + } + if (!is_numeric) all_labels_numeric = false; + label_count += 1; + } + + // need at least 2 labels (e.g. "pds.example.com", "bsky.network") + if (label_count < 2) return error.TooFewLabels; + + // all-numeric labels = IP address (e.g. "192.168.1.1") + if (all_labels_numeric) return error.LooksLikeIpAddress; + + // lowercase normalize + const result = allocator.alloc(u8, hostname.len) catch return error.EmptyHostname; + for (hostname, 0..) |c, i| { + result[i] = std.ascii.toLower(c); + } + return result; +} + +/// extract hostname from a URL like "https://pds.example.com" or +/// "https://pds.example.com:443/path". borrows from `url` — no allocation. +pub fn extractHostFromUrl(url: []const u8) ?[]const u8 { + // strip scheme + var rest = url; + if (std.mem.startsWith(u8, rest, "https://")) { + rest = rest["https://".len..]; + } else if (std.mem.startsWith(u8, rest, "http://")) { + rest = rest["http://".len..]; + } + // strip path + if (std.mem.indexOfScalar(u8, rest, '/')) |i| { + rest = rest[0..i]; + } + // strip port + if (std.mem.indexOfScalar(u8, rest, ':')) |i| { + rest = rest[0..i]; + } + if (rest.len == 0) return null; + return rest; +} + +// --- tests --- + +test "validateHostname accepts valid PDS hostnames" { + const alloc = std.testing.allocator; + + const h1 = try validateHostname(alloc, "pds.example.com"); + defer alloc.free(h1); + try std.testing.expectEqualStrings("pds.example.com", h1); + + const h2 = try validateHostname(alloc, "bsky.network"); + defer alloc.free(h2); + try std.testing.expectEqualStrings("bsky.network", h2); + + // lowercases + const h3 = try validateHostname(alloc, "PDS.Example.COM"); + defer alloc.free(h3); + try std.testing.expectEqualStrings("pds.example.com", h3); + + // strips scheme + const h4 = try validateHostname(alloc, "https://pds.example.com"); + defer alloc.free(h4); + try std.testing.expectEqualStrings("pds.example.com", h4); + + // strips path + const h5 = try validateHostname(alloc, "pds.example.com/some/path"); + defer alloc.free(h5); + try std.testing.expectEqualStrings("pds.example.com", h5); +} + +test "validateHostname rejects invalid hostnames" { + const alloc = std.testing.allocator; + + try std.testing.expectError(error.EmptyHostname, validateHostname(alloc, "")); + try std.testing.expectError(error.LocalhostNotAllowed, validateHostname(alloc, "localhost")); + try std.testing.expectError(error.TooFewLabels, validateHostname(alloc, "intranet")); + try std.testing.expectError(error.LooksLikeIpAddress, validateHostname(alloc, "192.168.1.1")); + try std.testing.expectError(error.LooksLikeIpAddress, validateHostname(alloc, "10.0.0.1")); + try std.testing.expectError(error.PortNotAllowed, validateHostname(alloc, "pds.example.com:443")); + try std.testing.expectError(error.InvalidCharacter, validateHostname(alloc, "pds.exam ple.com")); + try std.testing.expectError(error.InvalidCharacter, validateHostname(alloc, "pds.exam_ple.com")); + try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "-pds.example.com")); + try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "pds-.example.com")); + try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "pds..example.com")); +} + +test "extractHostFromUrl strips scheme, path, and port" { + try std.testing.expectEqualStrings("pds.example.com", extractHostFromUrl("https://pds.example.com").?); + try std.testing.expectEqualStrings("pds.example.com", extractHostFromUrl("https://pds.example.com:443/x").?); + try std.testing.expectEqualStrings("pds.example.com", extractHostFromUrl("http://pds.example.com/path").?); + try std.testing.expect(extractHostFromUrl("https://") == null); +} diff --git a/src/atproto/main.zig b/src/atproto/main.zig new file mode 100644 index 0000000..b03744c --- /dev/null +++ b/src/atproto/main.zig @@ -0,0 +1,16 @@ +//! atproto — AT Protocol / relay domain logic shared across modules. +//! host parsing and policy live here (not util/) because they encode relay +//! rules, mirroring indigo's relay rather than generic networking. + +pub const hostname = @import("hostname.zig"); +pub const host_check = @import("host_check.zig"); + +pub const HostValidationError = hostname.HostValidationError; +pub const validateHostname = hostname.validateHostname; +pub const extractHostFromUrl = hostname.extractHostFromUrl; +pub const checkHost = host_check.checkHost; +pub const rejectPrivateHost = host_check.rejectPrivateHost; + +test { + @import("std").testing.refAllDecls(@This()); +} diff --git a/src/broadcaster.zig b/src/broadcaster.zig index 707c6ce..e684466 100644 --- a/src/broadcaster.zig +++ b/src/broadcaster.zig @@ -16,9 +16,12 @@ const event_log_mod = @import("event_log.zig"); const builtin = @import("builtin"); const build_options = @import("build_options"); +const util = @import("util/util.zig"); const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); +const timestamp = util.timestamp; + // --- stats --- pub const Stats = struct { @@ -1479,10 +1482,6 @@ pub fn formatStatsResponse(stats: *const Stats, buf: []u8, io: Io) []const u8 { }) catch ""; } -fn timestamp(io: Io) i64 { - return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s)); -} - // --- tests --- var test_shutdown: std.atomic.Value(bool) = .{ .raw = false }; diff --git a/src/event_log.zig b/src/event_log.zig index 09396bf..66362e2 100644 --- a/src/event_log.zig +++ b/src/event_log.zig @@ -14,7 +14,9 @@ const std = @import("std"); const Io = std.Io; const pg = @import("pg"); const lru = @import("lru.zig"); +const util = @import("util/util.zig"); +const getenv = util.getenv; const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); @@ -1469,11 +1471,6 @@ test "header is little-endian" { try std.testing.expectEqual(@as(u8, 0x01), buf[9]); } -fn getenv(key: [*:0]const u8) ?[]const u8 { - const ptr = std.c.getenv(key) orelse return null; - return std.mem.sliceTo(ptr, 0); -} - test "scanForLastSeq stops at a torn tail" { // regression for the seq-rewind incident (docs/incident-2026-05-31-seq-rewind.md): // an interrupted flush leaves a partial final record. recovery must stop at the diff --git a/src/frame_worker.zig b/src/frame_worker.zig index 76f44a1..78a5a0c 100644 --- a/src/frame_worker.zig +++ b/src/frame_worker.zig @@ -16,13 +16,11 @@ const event_log_mod = @import("event_log.zig"); const collection_index_mod = @import("collection_index.zig"); const resync_mod = @import("resync.zig"); const thread_pool = @import("thread_pool.zig"); +const util = @import("util/util.zig"); const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); - -fn microTimestamp(io: Io) i64 { - return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_us)); -} +const microTimestamp = util.microTimestamp; pub const FrameWork = struct { data: []u8, // raw frame bytes (heap-duped by reader, freed by worker) diff --git a/src/host_ops.zig b/src/host_ops.zig index 0dd1448..c3ad961 100644 --- a/src/host_ops.zig +++ b/src/host_ops.zig @@ -14,9 +14,11 @@ const std = @import("std"); const event_log_mod = @import("event_log.zig"); const broadcaster = @import("broadcaster.zig"); +const util = @import("util/util.zig"); const Io = std.Io; const log = std.log.scoped(.relay); +const timestamp = util.timestamp; // --------------------------------------------------------------------------- // cursor coalescing map @@ -335,8 +337,4 @@ pub const HostOpsQueue = struct { log.warn("host_ops: persist #account takedown failed: {s}", .{@errorName(err)}); } } - - fn timestamp(io: Io) i64 { - return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s)); - } }; diff --git a/src/main.zig b/src/main.zig index b971aac..22e01b0 100644 --- a/src/main.zig +++ b/src/main.zig @@ -37,7 +37,11 @@ const cleaner_mod = @import("cleaner.zig"); const resync_mod = @import("resync.zig"); const host_ops_mod = @import("host_ops.zig"); const api = @import("api.zig"); +const util = @import("util/util.zig"); const build_options = @import("build_options"); + +const getenv = util.getenv; +const parseEnvInt = util.parseEnvInt; const malloc_trim: ?*const fn (pad: usize) callconv(.c) c_int = if (builtin.os.tag == .linux) @extern(*const fn (pad: usize) callconv(.c) c_int, .{ .name = "malloc_trim" }) else @@ -509,17 +513,6 @@ fn normalizeSeedHost(raw: []const u8) []const u8 { return raw; } -/// libc getenv — std.posix.getenv removed in 0.16 -fn getenv(key: [*:0]const u8) ?[]const u8 { - const ptr = std.c.getenv(key) orelse return null; - return std.mem.sliceTo(ptr, 0); -} - -fn parseEnvInt(comptime T: type, key: [*:0]const u8, default: T) T { - const val = getenv(key) orelse return default; - return std.fmt.parseInt(T, val, 10) catch default; -} - test "normalizeSeedHost" { // normal hostnames pass through try std.testing.expectEqualStrings("bsky.network", normalizeSeedHost("bsky.network")); diff --git a/src/slurper.zig b/src/slurper.zig index 9c3e99f..032a815 100644 --- a/src/slurper.zig +++ b/src/slurper.zig @@ -21,10 +21,15 @@ const collection_index_mod = @import("collection_index.zig"); const resync_mod = @import("resync.zig"); const frame_worker_mod = @import("frame_worker.zig"); const host_ops_mod = @import("host_ops.zig"); +const atproto = @import("atproto/main.zig"); const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); +pub const HostValidationError = atproto.HostValidationError; +pub const validateHostname = atproto.validateHostname; +const checkHost = atproto.checkHost; + pub const Options = struct { seed_host: []const u8 = "bsky.network", max_message_size: usize = 5 * 1024 * 1024, @@ -36,187 +41,6 @@ pub const Options = struct { startup_batch_size: u16 = 50, }; -// --- host validation --- -// mirrors indigo relay's ParseHostname + CheckHost + relay loop detection - -pub const HostValidationError = error{ - EmptyHostname, - InvalidCharacter, - InvalidLabel, - TooFewLabels, - LooksLikeIpAddress, - PortNotAllowed, - LocalhostNotAllowed, - DomainBanned, - HostUnreachable, - NotAPds, - IsARelay, -}; - -/// validate and normalize a hostname for crawling. -/// rejects IPs, ports, localhost, invalid DNS names. -/// returns lowercased hostname on success. -pub fn validateHostname(allocator: Allocator, raw: []const u8) HostValidationError![]u8 { - if (raw.len == 0) return error.EmptyHostname; - - // strip scheme if present - var hostname = raw; - for ([_][]const u8{ "https://", "http://", "wss://", "ws://" }) |scheme| { - if (hostname.len > scheme.len and std.ascii.startsWithIgnoreCase(hostname, scheme)) { - hostname = hostname[scheme.len..]; - break; - } - } - - // strip trailing slash and path - if (std.mem.indexOfScalar(u8, hostname, '/')) |i| { - hostname = hostname[0..i]; - } - - // reject ports (Go relay rejects non-localhost ports) - if (std.mem.indexOfScalar(u8, hostname, ':')) |_| { - return error.PortNotAllowed; - } - - // reject localhost - if (std.ascii.eqlIgnoreCase(hostname, "localhost")) { - return error.LocalhostNotAllowed; - } - - // validate characters and split into labels - var label_count: usize = 0; - var all_labels_numeric = true; - var it = std.mem.splitScalar(u8, hostname, '.'); - while (it.next()) |label| { - if (label.len == 0 or label.len > 63) return error.InvalidLabel; - // labels must be alphanumeric with hyphens, no leading/trailing hyphens - if (label[0] == '-' or label[label.len - 1] == '-') return error.InvalidLabel; - var is_numeric = true; - for (label) |c| { - if (!std.ascii.isAlphanumeric(c) and c != '-') return error.InvalidCharacter; - if (!std.ascii.isDigit(c)) is_numeric = false; - } - if (!is_numeric) all_labels_numeric = false; - label_count += 1; - } - - // need at least 2 labels (e.g. "pds.example.com", "bsky.network") - if (label_count < 2) return error.TooFewLabels; - - // all-numeric labels = IP address (e.g. "192.168.1.1") - if (all_labels_numeric) return error.LooksLikeIpAddress; - - // lowercase normalize - const result = allocator.alloc(u8, hostname.len) catch return error.EmptyHostname; - for (hostname, 0..) |c, i| { - result[i] = std.ascii.toLower(c); - } - return result; -} - -/// SSRF protection: resolve hostname and reject private/reserved IP ranges. -/// Go relay: ssrf.go PublicOnlyTransport — rejects 10/8, 172.16/12, 192.168/16, 127/8, link-local. -fn rejectPrivateHost(allocator: Allocator, hostname: []const u8) HostValidationError!void { - // null-terminate hostname for getaddrinfo - const hostname_z = allocator.dupeZ(u8, hostname) catch return error.HostUnreachable; - defer allocator.free(hostname_z); - - var hints: std.c.addrinfo = .{ - .flags = .{}, - .family = std.c.AF.UNSPEC, - .socktype = std.c.SOCK.STREAM, - .protocol = 0, - .addrlen = 0, - .addr = null, - .canonname = null, - .next = null, - }; - - var res: ?*std.c.addrinfo = null; - const rc = std.c.getaddrinfo(hostname_z, "443", &hints, &res); - if (@intFromEnum(rc) != 0 or res == null) return error.HostUnreachable; - defer std.c.freeaddrinfo(res.?); - - // check all resolved addresses — reject if ANY is private - var cur = res; - var found_any = false; - while (cur) |node| : (cur = node.next) { - found_any = true; - if (node.family == std.c.AF.INET) { - const sa: *const std.c.sockaddr.in = @ptrCast(@alignCast(node.addr.?)); - const bytes: [4]u8 = @bitCast(sa.addr); - if (bytes[0] == 10 or // 10.0.0.0/8 - (bytes[0] == 172 and (bytes[1] & 0xf0) == 16) or // 172.16.0.0/12 - (bytes[0] == 192 and bytes[1] == 168) or // 192.168.0.0/16 - bytes[0] == 127 or // 127.0.0.0/8 - bytes[0] == 0 or // 0.0.0.0/8 - (bytes[0] == 169 and bytes[1] == 254)) // 169.254.0.0/16 link-local - { - log.warn("SSRF: {s} resolves to private IP {d}.{d}.{d}.{d}", .{ hostname, bytes[0], bytes[1], bytes[2], bytes[3] }); - return error.HostUnreachable; - } - } - // allow IPv6 for now (could add RFC 4193 check later) - } - if (!found_any) return error.HostUnreachable; -} - -/// check that a host is a real PDS by calling describeServer. -/// also checks Server header for relay loop detection. -/// Go relay: host_checker.go CheckHost + slurper.go Server header check. -fn checkHost(allocator: Allocator, hostname: []const u8, io: Io) HostValidationError!void { - // SSRF protection: reject private IPs before making any request - rejectPrivateHost(allocator, hostname) catch |err| return err; - var url_buf: [512]u8 = undefined; - const url = std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.server.describeServer", .{hostname}) catch return error.HostUnreachable; - - var client: http.Client = .{ .allocator = allocator, .io = io }; - defer client.deinit(); - - const uri = std.Uri.parse(url) catch return error.HostUnreachable; - var req = client.request(.GET, uri, .{}) catch return error.HostUnreachable; - defer req.deinit(); - req.sendBodiless() catch return error.HostUnreachable; - - var redirect_buf: [2048]u8 = undefined; - const response = req.receiveHead(&redirect_buf) catch return error.HostUnreachable; - - if (response.head.status != .ok) return error.NotAPds; - - // relay loop detection: check Server header for "atproto-relay" - // Go relay: slurper.go — auto-bans hosts whose Server header contains "atproto-relay" - if (findHeaderInRaw(response.head.bytes, "server")) |server_val| { - if (std.mem.indexOf(u8, server_val, "atproto-relay") != null) { - return error.IsARelay; - } - } -} - -/// search raw HTTP headers for a header by name (case-insensitive). -/// returns the trimmed value, or null if not found. -fn findHeaderInRaw(raw: []const u8, name: []const u8) ?[]const u8 { - var it = std.mem.splitSequence(u8, raw, "\r\n"); - _ = it.next(); // skip status line - while (it.next()) |line| { - if (line.len == 0) break; - const colon = std.mem.indexOfScalar(u8, line, ':') orelse continue; - const key = std.mem.trim(u8, line[0..colon], " "); - if (key.len != name.len) continue; - // case-insensitive compare - var match = true; - for (key, name) |a, b| { - if (std.ascii.toLower(a) != std.ascii.toLower(b)) { - match = false; - break; - } - } - if (match) { - return std.mem.trim(u8, line[colon + 1 ..], " "); - } - } - return null; -} - const WorkerEntry = struct { future: Io.Future(void), subscriber: *subscriber_mod.Subscriber, @@ -772,58 +596,3 @@ pub const Slurper = struct { if (self.ca_bundle) |*b| b.deinit(self.allocator); } }; - -// --- tests --- - -test "validateHostname accepts valid PDS hostnames" { - const alloc = std.testing.allocator; - - // basic valid hostnames - const h1 = try validateHostname(alloc, "pds.example.com"); - defer alloc.free(h1); - try std.testing.expectEqualStrings("pds.example.com", h1); - - // two labels minimum - const h2 = try validateHostname(alloc, "bsky.network"); - defer alloc.free(h2); - try std.testing.expectEqualStrings("bsky.network", h2); - - // lowercases - const h3 = try validateHostname(alloc, "PDS.Example.COM"); - defer alloc.free(h3); - try std.testing.expectEqualStrings("pds.example.com", h3); - - // strips scheme - const h4 = try validateHostname(alloc, "https://pds.example.com"); - defer alloc.free(h4); - try std.testing.expectEqualStrings("pds.example.com", h4); - - // strips path - const h5 = try validateHostname(alloc, "pds.example.com/some/path"); - defer alloc.free(h5); - try std.testing.expectEqualStrings("pds.example.com", h5); -} - -test "validateHostname rejects invalid hostnames" { - const alloc = std.testing.allocator; - - // empty - try std.testing.expectError(error.EmptyHostname, validateHostname(alloc, "")); - // localhost - try std.testing.expectError(error.LocalhostNotAllowed, validateHostname(alloc, "localhost")); - // single label (non-localhost) - try std.testing.expectError(error.TooFewLabels, validateHostname(alloc, "intranet")); - // IP address - try std.testing.expectError(error.LooksLikeIpAddress, validateHostname(alloc, "192.168.1.1")); - try std.testing.expectError(error.LooksLikeIpAddress, validateHostname(alloc, "10.0.0.1")); - // port - try std.testing.expectError(error.PortNotAllowed, validateHostname(alloc, "pds.example.com:443")); - // invalid characters - try std.testing.expectError(error.InvalidCharacter, validateHostname(alloc, "pds.exam ple.com")); - try std.testing.expectError(error.InvalidCharacter, validateHostname(alloc, "pds.exam_ple.com")); - // leading/trailing hyphens - try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "-pds.example.com")); - try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "pds-.example.com")); - // empty label - try std.testing.expectError(error.InvalidLabel, validateHostname(alloc, "pds..example.com")); -} diff --git a/src/subscriber.zig b/src/subscriber.zig index 36b7a7b..33911ed 100644 --- a/src/subscriber.zig +++ b/src/subscriber.zig @@ -16,11 +16,17 @@ const collection_index_mod = @import("collection_index.zig"); const resync_mod = @import("resync.zig"); const frame_worker_mod = @import("frame_worker.zig"); const host_ops_mod = @import("host_ops.zig"); +const util = @import("util/util.zig"); const Allocator = std.mem.Allocator; const Io = std.Io; const log = std.log.scoped(.relay); +const timestamp = util.timestamp; +const milliTimestamp = util.milliTimestamp; +const microTimestamp = util.microTimestamp; +const nanoTimestamp = util.nanoTimestamp; + const max_consecutive_failures = 15; const cursor_flush_interval_sec = 4; // flush cursor to DB every N seconds (Go relay: 4s) const ping_interval_sec = 30; // keepalive ping interval (Go relay: 30s) @@ -69,22 +75,6 @@ pub const Options = struct { ca_bundle: ?std.crypto.Certificate.Bundle = null, }; -fn timestamp(io: Io) i64 { - return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s)); -} - -fn milliTimestamp(io: Io) i64 { - return Io.Timestamp.now(io, .real).toMilliseconds(); -} - -fn microTimestamp(io: Io) i64 { - return Io.Timestamp.now(io, .real).toMicroseconds(); -} - -fn nanoTimestamp(io: Io) i96 { - return Io.Timestamp.now(io, .real).toNanoseconds(); -} - /// simple sliding window rate limiter — tracks event counts per second/hour/day. /// Sliding window rate limiter (same algorithm as Go relay's github.com/RussellLuo/slidingwindow). /// Uses millisecond timestamps for sub-second precision (critical for the 1-second window). diff --git a/src/util/env.zig b/src/util/env.zig new file mode 100644 index 0000000..0b4f850 --- /dev/null +++ b/src/util/env.zig @@ -0,0 +1,18 @@ +//! env — process environment helpers. +//! +//! std.posix.getenv was removed in zig 0.16; this wraps libc getenv so the +//! four call sites that needed it don't each re-declare the extern shim. + +const std = @import("std"); + +/// libc getenv — returns the value as a slice, or null if unset. +pub fn getenv(key: [*:0]const u8) ?[]const u8 { + const ptr = std.c.getenv(key) orelse return null; + return std.mem.sliceTo(ptr, 0); +} + +/// parse an integer-valued env var, falling back to `default` if unset or unparseable. +pub fn parseEnvInt(comptime T: type, key: [*:0]const u8, default: T) T { + const val = getenv(key) orelse return default; + return std.fmt.parseInt(T, val, 10) catch default; +} diff --git a/src/util/time.zig b/src/util/time.zig new file mode 100644 index 0000000..908e3ff --- /dev/null +++ b/src/util/time.zig @@ -0,0 +1,49 @@ +//! time — clock helpers wrapping the Io timestamp source. +//! +//! every reader thread / fiber needs "what time is it" in some unit; these +//! were re-implemented in five modules before being centralized here. + +const std = @import("std"); +const Io = std.Io; + +/// seconds since the unix epoch. +pub fn timestamp(io: Io) i64 { + return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s)); +} + +/// milliseconds since the unix epoch. +pub fn milliTimestamp(io: Io) i64 { + return Io.Timestamp.now(io, .real).toMilliseconds(); +} + +/// microseconds since the unix epoch. +pub fn microTimestamp(io: Io) i64 { + return Io.Timestamp.now(io, .real).toMicroseconds(); +} + +/// nanoseconds since the unix epoch. +pub fn nanoTimestamp(io: Io) i96 { + return Io.Timestamp.now(io, .real).toNanoseconds(); +} + +/// format the current wall-clock time as RFC-3339 (e.g. "2026-06-07T12:34:56Z"). +/// writes into a caller-provided 24-byte buffer and returns the slice. +pub fn formatTimestamp(buf: *[24]u8) []const u8 { + var tp: std.c.timespec = undefined; + _ = std.c.clock_gettime(.REALTIME, &tp); + const ts: u64 = @intCast(tp.sec); + const es = std.time.epoch.EpochSeconds{ .secs = ts }; + const day = es.getEpochDay(); + const yd = day.calculateYearDay(); + const md = yd.calculateMonthDay(); + const ds = es.getDaySeconds(); + + return std.fmt.bufPrint(buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}Z", .{ + yd.year, + @as(u32, @intFromEnum(md.month)) + 1, + @as(u32, md.day_index) + 1, + ds.getHoursIntoDay(), + ds.getMinutesIntoHour(), + ds.getSecondsIntoMinute(), + }) catch "1970-01-01T00:00:00Z"; +} diff --git a/src/util/util.zig b/src/util/util.zig new file mode 100644 index 0000000..c6a9b7b --- /dev/null +++ b/src/util/util.zig @@ -0,0 +1,20 @@ +//! util — extensions to the standard library: things that could have been in +//! std, but aren't. namespacing is deliberately flat (util.getenv, not +//! util.env.getenv) — directness over scalability; hierarchy can come later. + +pub const env = @import("env.zig"); +pub const time = @import("time.zig"); + +// flattened leaves — the common case +pub const getenv = env.getenv; +pub const parseEnvInt = env.parseEnvInt; + +pub const timestamp = time.timestamp; +pub const milliTimestamp = time.milliTimestamp; +pub const microTimestamp = time.microTimestamp; +pub const nanoTimestamp = time.nanoTimestamp; +pub const formatTimestamp = time.formatTimestamp; + +test { + @import("std").testing.refAllDecls(@This()); +} diff --git a/src/validator.zig b/src/validator.zig index 49118e9..d4f108e 100644 --- a/src/validator.zig +++ b/src/validator.zig @@ -11,10 +11,16 @@ const zat = @import("zat"); const broadcaster = @import("broadcaster.zig"); const event_log_mod = @import("event_log.zig"); const lru = @import("lru.zig"); +const util = @import("util/util.zig"); +const atproto = @import("atproto/main.zig"); const Allocator = std.mem.Allocator; const log = std.log.scoped(.relay); +const parseEnvInt = util.parseEnvInt; +const timestamp = util.timestamp; +const extractHostFromUrl = atproto.extractHostFromUrl; + /// decoded and cached signing key for a DID const CachedKey = struct { key_type: zat.multicodec.KeyType, @@ -635,41 +641,6 @@ pub const Validator = struct { } }; -/// extract hostname from a URL like "https://pds.example.com" or "https://pds.example.com:443/path" -pub fn extractHostFromUrl(url: []const u8) ?[]const u8 { - // strip scheme - var rest = url; - if (std.mem.startsWith(u8, rest, "https://")) { - rest = rest["https://".len..]; - } else if (std.mem.startsWith(u8, rest, "http://")) { - rest = rest["http://".len..]; - } - // strip path - if (std.mem.indexOfScalar(u8, rest, '/')) |i| { - rest = rest[0..i]; - } - // strip port - if (std.mem.indexOfScalar(u8, rest, ':')) |i| { - rest = rest[0..i]; - } - if (rest.len == 0) return null; - return rest; -} - -fn getenv(key: [*:0]const u8) ?[]const u8 { - const ptr = std.c.getenv(key) orelse return null; - return std.mem.sliceTo(ptr, 0); -} - -fn parseEnvInt(comptime T: type, key: [*:0]const u8, default: T) T { - const val = getenv(key) orelse return default; - return std.fmt.parseInt(T, val, 10) catch default; -} - -fn timestamp(io: Io) i64 { - return @intCast(@divFloor(Io.Timestamp.now(io, .real).nanoseconds, std.time.ns_per_s)); -} - // --- tests --- test "validateCommit skips on cache miss" { -- 2.51.2