From e6cd05a259943a2f92c58152a7746ac942cf86a7 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 17 Aug 2026 14:45:17 -0500 Subject: [PATCH] archive key ring: named, metered, file-revocable keys beside the fleet key upstream OSS has no archive auth (only the single import token); Bluesky meters hosted archives in a proprietary gateway (2 MB/s/key default). the ring approximates that gateway in-process: name:key[:mbps] lines, per-key token buckets charged on served bytes (429 on deficit, which the SDK already retries), revocation by deleting a line (mtime reload, no restart), fail-closed on an unreadable file. fleet key unchanged and unmetered. motivation: invite third-party archive drains (23 MB/s measured server ceiling) without letting one consumer contend with compaction on the ~200 MB/s volume, and revoke per-consumer without rotating the fleet credential. Co-Authored-By: Claude Fable 5 --- docs/configuration.md | 21 +++ docs/semantic-parity.md | 17 ++ src/internal/runtime/cli.zig | 4 + src/internal/runtime/environment.zig | 3 +- src/internal/serve/api_keys.zig | 260 +++++++++++++++++++++++++++ src/internal/serve/server.zig | 6 + src/internal/serve/xrpcapi.zig | 105 ++++++++++- src/main.zig | 10 ++ 8 files changed, 422 insertions(+), 4 deletions(-) create mode 100644 src/internal/serve/api_keys.zig diff --git a/docs/configuration.md b/docs/configuration.md index 163f174..92c8187 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -279,6 +279,27 @@ the reverse proxy; Stream serves plain HTTP internally. Empty, missing, and bad credentials all produce the same 401. Submitted paths are canonicalized through symlinks and must resolve to regular files inside the configured directory. +### archive key ring + +`--archive-api-keys-file=` (`JETSTREAM_ARCHIVE_API_KEYS_FILE`) adds +named, metered archive credentials beside the fleet key: + +``` +# name:key[:mbps] mbps omitted or 0 = unmetered +evelyn:some-long-random-secret:8 +ci:another-secret +``` + +- keys authorize the archive endpoints exactly like `--archive-api-key` + (which stays valid and unmetered) +- `mbps` is a per-key token bucket (4s burst) charged on served bytes; + a key in deficit gets `429 RateLimitExceeded` until it refills — the + SDK's fetch retry already backs off on 429 +- **revocation = delete the line.** the file's mtime is rechecked at most + once per second on the request path; no restart, no signal +- an unreadable/missing file revokes every ring key (fail closed) and + logs a warning; the fleet key is unaffected + ## OpenTelemetry Stream installs the same process-wide tracing provider as pinned Jetstream V2. diff --git a/docs/semantic-parity.md b/docs/semantic-parity.md index 7537a52..7db8dcf 100644 --- a/docs/semantic-parity.md +++ b/docs/semantic-parity.md @@ -200,3 +200,20 @@ The admission command must: The previous `4d163e8` receipt remains historical evidence for that artifact. It is not a current admission and must not be copied forward. + +## archive key ring (2026-08-17, extension) + +upstream OSS has no archive auth at all — its only bearer surface is the +single timestamp-import token (xrpcapi/auth.go). Bluesky's hosted +instances gate and meter archives in a proprietary gateway in front +(2 MB/s per key by default; the gateway exists to adjust limits per key — +alex.bsky.team, 2026-08-17). stream's single `--archive-api-key` was +already an extension mirroring that hosted behavior; the key ring +(`--archive-api-keys-file`, serve/api_keys.zig) extends it into an +in-process approximation of the gateway: named keys (`name:key[:mbps]` +lines), per-key token-bucket byte budgets (429 + retry semantics the SDK +already honors), and revocation by deleting a line (mtime-based reload, +no restart). the fleet key remains valid and unmetered. rationale: +metering protects compaction/live-delivery from uncapped archive drains +on a ~200 MB/s volume; revocation avoids rotating the fleet credential +per consumer. diff --git a/src/internal/runtime/cli.zig b/src/internal/runtime/cli.zig index c34a051..a365187 100644 --- a/src/internal/runtime/cli.zig +++ b/src/internal/runtime/cli.zig @@ -205,6 +205,7 @@ pub const ServeOptions = struct { /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13: /// live tail and dictionary open, archive behind a token). archive_api_key: []const u8 = "", + archive_api_keys_file: []const u8 = "", timestamp_import_dir: []const u8 = "", pub const overrides = .{ @@ -469,6 +470,7 @@ pub const ServeConfig = struct { /// hosted Bluesky instances' edge behavior (Jetstream v2, 2026-08-13: /// live tail and dictionary open, archive behind a token). archive_api_key: []const u8 = "", + archive_api_keys_file: []const u8 = "", timestamp_import_dir_arg: ?[]const u8 = null, repo_action_rate_limits: bool = true, store_fault_prefix: ?[]const u8 = null, @@ -677,6 +679,8 @@ pub fn parseServe( cfg.plan_config.whole_segment_threshold = try std.fmt.parseFloat(f64, arg["--plan-whole-segment-threshold=".len..]); } else if (std.mem.startsWith(u8, arg, "--archive-api-key=")) { cfg.archive_api_key = arg["--archive-api-key=".len..]; + } else if (std.mem.startsWith(u8, arg, "--archive-api-keys-file=")) { + cfg.archive_api_keys_file = arg["--archive-api-keys-file=".len..]; } else if (std.mem.startsWith(u8, arg, "--timestamp-import-token=")) { cfg.timestamp_import_token = arg["--timestamp-import-token=".len..]; } else if (std.mem.startsWith(u8, arg, "--timestamp-import-dir=")) { diff --git a/src/internal/runtime/environment.zig b/src/internal/runtime/environment.zig index 824d4b2..4d96fee 100644 --- a/src/internal/runtime/environment.zig +++ b/src/internal/runtime/environment.zig @@ -50,6 +50,7 @@ pub const mappings = [_]Mapping{ .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_TOKEN", .flag = "timestamp-import-token" }, .{ .env = "JETSTREAM_TIMESTAMP_IMPORT_DIR", .flag = "timestamp-import-dir" }, .{ .env = "JETSTREAM_ARCHIVE_API_KEY", .flag = "archive-api-key" }, + .{ .env = "JETSTREAM_ARCHIVE_API_KEYS_FILE", .flag = "archive-api-keys-file" }, }; pub fn processEntries(allocator: std.mem.Allocator) ![]const []const u8 { @@ -133,7 +134,7 @@ fn isKnown(key: []const u8) bool { } test "unknown Jetstream variables are sorted, deduplicated, and scoped" { - try std.testing.expectEqual(@as(usize, 39), mappings.len); + try std.testing.expectEqual(@as(usize, 40), mappings.len); const entries = [_][]const u8{ "JETSTREAM_ZZZ=1", "JETSTREAM_ADDR=127.0.0.1:0", diff --git a/src/internal/serve/api_keys.zig b/src/internal/serve/api_keys.zig new file mode 100644 index 0000000..ea6ff9c --- /dev/null +++ b/src/internal/serve/api_keys.zig @@ -0,0 +1,260 @@ +//! named, revocable archive API keys with per-key byte-rate budgets. +//! +//! extension beyond upstream: the OSS server has NO archive auth at all +//! (its only bearer surface is the single timestamp-import token, +//! xrpcapi/auth.go); Bluesky gates and meters hosted archives in a +//! proprietary gateway in front — 2 MB/s per key by default +//! (alex.bsky.team, 2026-08-17), with the gateway existing precisely to +//! adjust per-key limits. this ring approximates that gateway in-process: +//! named keys, per-key byte budgets, and revocation without rotating the +//! fleet credential. recorded in docs/semantic-parity.md. +//! +//! the key file is one entry per line: +//! +//! name:key[:mbps] +//! +//! `#` comments and blank lines are ignored; a missing or 0 mbps means +//! unmetered. revocation is deleting the line — the file's mtime is +//! re-checked (at most once per second) on the request path, so no restart +//! or signal is needed. the legacy single fleet key (--archive-api-key) +//! keeps working beside the ring, unmetered. +//! +//! metering is a token bucket in bytes, charged AFTER each response (the +//! bucket may go negative; the next request is refused with 429 until the +//! deficit refills). burst capacity is 4 seconds of rate. presented keys +//! are compared as sha256 digests in constant time, same as authorized(). + +const std = @import("std"); + +const Io = std.Io; +const Allocator = std.mem.Allocator; +const log = std.log.scoped(.api_keys); + +pub const burst_seconds = 4; +/// stat the key file at most this often +const reload_check_interval_us: i64 = std.time.us_per_s; + +pub const Entry = struct { + name: []const u8, + key_hash: [32]u8, + /// bytes per second; 0 = unmetered + rate_bps: u64, + /// token bucket level; negative = in deficit + bucket_bytes: i64, + last_refill_us: i64, +}; + +pub const KeyRing = struct { + allocator: Allocator, + path: []const u8, + mutex: Io.Mutex = .init, + entries: std.ArrayList(Entry) = .empty, + arena: std.heap.ArenaAllocator, + loaded_mtime_us: i64 = std.math.minInt(i64), + last_check_us: i64 = std.math.minInt(i64), + + pub fn init(allocator: Allocator, path: []const u8) KeyRing { + return .{ + .allocator = allocator, + .path = path, + .arena = std.heap.ArenaAllocator.init(allocator), + }; + } + + pub fn deinit(self: *KeyRing) void { + self.entries.deinit(self.allocator); + self.arena.deinit(); + self.* = undefined; + } + + /// resolve a bearer header to a live key entry. reloads the file when + /// its mtime changed (checked at most once per second), so adding or + /// deleting a line takes effect without a restart. returns null for + /// missing/unknown/revoked keys. + pub fn authorize(self: *KeyRing, io: Io, authorization: ?[]const u8, now_us: i64) ?*Entry { + const raw = authorization orelse return null; + const prefix = "Bearer "; + if (raw.len <= prefix.len or !std.ascii.eqlIgnoreCase(raw[0..prefix.len], prefix)) return null; + var presented_hash: [32]u8 = undefined; + std.crypto.hash.sha2.Sha256.hash(raw[prefix.len..], &presented_hash, .{}); + + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + self.maybeReload(io, now_us); + for (self.entries.items) |*entry| { + if (std.crypto.timing_safe.eql([32]u8, entry.key_hash, presented_hash)) return entry; + } + return null; + } + + /// deficit check before serving. returns seconds to wait when the + /// key's bucket is negative (serve a 429), null when clear to serve. + pub fn precheck(self: *KeyRing, io: Io, entry: *Entry, now_us: i64) ?u64 { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + if (entry.rate_bps == 0) return null; + refill(entry, now_us); + if (entry.bucket_bytes >= 0) return null; + const deficit: u64 = @intCast(-entry.bucket_bytes); + return @max(1, deficit / entry.rate_bps); + } + + /// charge served bytes to the key after the response. the bucket may + /// go negative — one oversized response is always allowed, and the + /// deficit gates the next one. + pub fn charge(self: *KeyRing, io: Io, entry: *Entry, bytes: u64, now_us: i64) void { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + if (entry.rate_bps == 0) return; + refill(entry, now_us); + entry.bucket_bytes -= @intCast(@min(bytes, std.math.maxInt(i64))); + } + + fn refill(entry: *Entry, now_us: i64) void { + const elapsed_us = now_us - entry.last_refill_us; + entry.last_refill_us = now_us; + if (elapsed_us <= 0) return; + const cap: i64 = @intCast(entry.rate_bps * burst_seconds); + const gained: i64 = @intCast(@min( + (@as(u128, @intCast(elapsed_us)) * entry.rate_bps) / std.time.us_per_s, + @as(u128, @intCast(std.math.maxInt(i64))), + )); + entry.bucket_bytes = @min(entry.bucket_bytes +| gained, cap); + } + + fn maybeReload(self: *KeyRing, io: Io, now_us: i64) void { + if (now_us -| self.last_check_us < reload_check_interval_us) return; + self.last_check_us = now_us; + const cwd = Io.Dir.cwd(); + var file = cwd.openFile(io, self.path, .{}) catch |err| { + // a missing/unreadable file revokes everything — fail closed, + // loudly + if (self.entries.items.len > 0) { + log.warn("key file {s} unreadable ({s}); all ring keys revoked", .{ self.path, @errorName(err) }); + self.entries.clearRetainingCapacity(); + self.loaded_mtime_us = std.math.minInt(i64); + } + return; + }; + defer file.close(io); + const stat = file.stat(io) catch return; + const mtime_us: i64 = stat.mtime.toMicroseconds(); + if (mtime_us == self.loaded_mtime_us) return; + + const bytes = cwd.readFileAlloc(io, self.path, self.allocator, .limited(1 << 20)) catch |err| { + log.warn("key file {s} read failed: {s}", .{ self.path, @errorName(err) }); + return; + }; + defer self.allocator.free(bytes); + self.loadFromSlice(bytes) catch |err| { + log.warn("key file {s} parse failed: {s}; keeping previous ring", .{ self.path, @errorName(err) }); + return; + }; + self.loaded_mtime_us = mtime_us; + log.info("archive key ring loaded: {d} keys from {s}", .{ self.entries.items.len, self.path }); + } + + /// parse `name:key[:mbps]` lines, replacing the ring. buckets of keys + /// that survive the reload keep their level (revoke-and-re-add resets). + pub fn loadFromSlice(self: *KeyRing, bytes: []const u8) !void { + var fresh: std.ArrayList(Entry) = .empty; + errdefer fresh.deinit(self.allocator); + var next_arena = std.heap.ArenaAllocator.init(self.allocator); + errdefer next_arena.deinit(); + + var lines = std.mem.tokenizeScalar(u8, bytes, '\n'); + while (lines.next()) |raw_line| { + const line = std.mem.trim(u8, raw_line, " \t\r"); + if (line.len == 0 or line[0] == '#') continue; + var parts = std.mem.splitScalar(u8, line, ':'); + const name = parts.next() orelse return error.MalformedKeyLine; + const key = parts.next() orelse return error.MalformedKeyLine; + if (name.len == 0 or key.len == 0) return error.MalformedKeyLine; + const rate_bps: u64 = if (parts.next()) |mbps_str| + (std.fmt.parseInt(u64, mbps_str, 10) catch return error.MalformedKeyLine) * 1000 * 1000 + else + 0; + var hash: [32]u8 = undefined; + std.crypto.hash.sha2.Sha256.hash(key, &hash, .{}); + var entry: Entry = .{ + .name = try next_arena.allocator().dupe(u8, name), + .key_hash = hash, + .rate_bps = rate_bps, + .bucket_bytes = @intCast(rate_bps * burst_seconds), + .last_refill_us = 0, + }; + for (self.entries.items) |old| { + if (std.crypto.timing_safe.eql([32]u8, old.key_hash, hash)) { + entry.bucket_bytes = old.bucket_bytes; + entry.last_refill_us = old.last_refill_us; + break; + } + } + try fresh.append(self.allocator, entry); + } + self.entries.deinit(self.allocator); + self.entries = fresh; + self.arena.deinit(); + self.arena = next_arena; + } +}; + +// === tests === + +const testing = std.testing; + +test "loadFromSlice: names, rates, comments, revocation, bucket carry-over" { + var ring = KeyRing.init(testing.allocator, "/nonexistent"); + defer ring.deinit(); + + try ring.loadFromSlice( + \\# fleet consumers + \\evelyn:sekrit-key-1:8 + \\ci:other-key + ); + try testing.expectEqual(@as(usize, 2), ring.entries.items.len); + try testing.expectEqualStrings("evelyn", ring.entries.items[0].name); + try testing.expectEqual(@as(u64, 8_000_000), ring.entries.items[0].rate_bps); + try testing.expectEqual(@as(u64, 0), ring.entries.items[1].rate_bps); + + // drain some budget, then reload with the key surviving: level carries + ring.entries.items[0].bucket_bytes = -5_000_000; + try ring.loadFromSlice("evelyn:sekrit-key-1:8\n"); + try testing.expectEqual(@as(usize, 1), ring.entries.items.len); + try testing.expectEqual(@as(i64, -5_000_000), ring.entries.items[0].bucket_bytes); + + // revocation: the line is gone, the key is gone + try ring.loadFromSlice("ci:other-key\n"); + try testing.expectEqual(@as(usize, 1), ring.entries.items.len); + try testing.expectEqualStrings("ci", ring.entries.items[0].name); + + try testing.expectError(error.MalformedKeyLine, ring.loadFromSlice("no-key-here\n")); +} + +test "bucket: burst allowed, deficit 429s, refill clears it" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var ring = KeyRing.init(testing.allocator, "/nonexistent"); + defer ring.deinit(); + try ring.loadFromSlice("e:k:2\n"); // 2 MB/s, 8 MB burst + const entry = &ring.entries.items[0]; + entry.last_refill_us = 0; + + // fresh bucket: clear to serve, then a 300 MB drain lands us in deficit + try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000)); + ring.charge(io, entry, 300_000_000, 1_000_000); + const wait = ring.precheck(io, entry, 1_000_001) orelse return error.ExpectedDeficit; + // ~292 MB deficit at 2 MB/s ≈ 146s + try testing.expect(wait > 100 and wait < 200); + + // after enough wall clock the bucket refills to burst and serves again + try testing.expectEqual(@as(?u64, null), ring.precheck(io, entry, 1_000_000 + 200 * std.time.us_per_s)); + + // unmetered key never blocks + try ring.loadFromSlice("open:k2\n"); + const open = &ring.entries.items[0]; + ring.charge(io, open, 1 << 40, 5); + try testing.expectEqual(@as(?u64, null), ring.precheck(io, open, 6)); +} diff --git a/src/internal/serve/server.zig b/src/internal/serve/server.zig index 48b8388..52d4582 100644 --- a/src/internal/serve/server.zig +++ b/src/internal/serve/server.zig @@ -1281,6 +1281,12 @@ const XrpcResponder = struct { modified_s: ?i64 = null, cache_max_age_s: u64 = 0, + /// bytes actually served on this request (0 until a 200 body went out); + /// the api key ring charges per-key budgets from this + pub fn servedBytes(self: *const XrpcResponder) u64 { + return self.observation.served_bytes; + } + pub fn archiveModified(self: *XrpcResponder, modified_s: i64) void { // net/http treats the Unix epoch as an unspecified modification time. self.modified_s = if (modified_s == 0) null else modified_s; diff --git a/src/internal/serve/xrpcapi.zig b/src/internal/serve/xrpcapi.zig index 3cf6183..7a42d56 100644 --- a/src/internal/serve/xrpcapi.zig +++ b/src/internal/serve/xrpcapi.zig @@ -11,6 +11,7 @@ //! contract, including multipart/byteranges and mixed overlap handling. const std = @import("std"); +const api_keys = @import("api_keys.zig"); const archive_mod = @import("../storage/archive.zig"); const segment = @import("../storage/segment.zig"); const import_manager = @import("../timestamp/manager.zig"); @@ -55,6 +56,10 @@ pub const Api = struct { /// Empty = open archive — the operator's opt-in, since gating an /// already-public surface must be a deliberate flip, not a default. archive_api_key: []const u8 = "", + /// named, revocable, per-key-metered archive keys (api_keys.zig). + /// checked when the fleet key above does not match; null = ring + /// disabled and the single key is the only credential. + key_ring: ?*api_keys.KeyRing = null, /// Keep one manifest planning pass in flight so public callers cannot /// multiply CPU and response allocation with concurrent requests. plan_mutex: Io.Mutex = .init, @@ -95,9 +100,26 @@ pub const Api = struct { std.mem.eql(u8, method, "listSegments") or std.mem.eql(u8, method, "getSegment") or std.mem.eql(u8, method, "getBlock"); - if (archive_method and self.archive_api_key.len > 0 and !authorized(self.archive_api_key, authorization)) { - respond.unauthorized(); - return true; + var metered: ?*api_keys.Entry = null; + if (archive_method) { + const fleet_ok = self.archive_api_key.len > 0 and authorized(self.archive_api_key, authorization); + const gated = self.archive_api_key.len > 0 or self.key_ring != null; + if (gated and !fleet_ok) { + const ring = self.key_ring orelse { + respond.unauthorized(); + return true; + }; + const now_us = nowUs(self.io); + metered = ring.authorize(self.io, authorization, now_us) orelse { + respond.unauthorized(); + return true; + }; + if (ring.precheck(self.io, metered.?, now_us)) |_| { + // the SDK's fetchWithRetry backs off on 429 (1s/2s/4s) + respond.err(429, "RateLimitExceeded", "per-key byte budget exhausted; retry shortly"); + return true; + } + } } if (std.mem.eql(u8, method, "planSnapshot")) { if (!is_post) { @@ -105,6 +127,7 @@ pub const Api = struct { return true; } planSnapshot(self, respond, body); + self.chargeMetered(respond, metered); return true; } if (std.mem.eql(u8, method, "planBackfill")) { @@ -128,9 +151,24 @@ pub const Api = struct { } else { respond.err(501, "MethodNotImplemented", method); } + self.chargeMetered(respond, metered); return true; } + /// charge a metered key for the bytes the responder actually served. + /// post-hoc by design: the bucket may go negative and gates the NEXT + /// request, so one oversized response is always allowed through. + fn chargeMetered(self: *Api, respond: anytype, metered: ?*api_keys.Entry) void { + const entry = metered orelse return; + const ring = self.key_ring orelse return; + const bytes: u64 = if (comptime @hasDecl(@TypeOf(respond.*), "servedBytes")) respond.servedBytes() else 0; + ring.charge(self.io, entry, bytes, nowUs(self.io)); + } + + fn nowUs(io: Io) i64 { + return @intCast(@divTrunc(Io.Timestamp.now(io, .awake).nanoseconds, std.time.ns_per_us)); + } + fn getZstdDictionary(_: *Api, respond: anytype, query: []const u8, if_none_match: ?[]const u8) void { if (queryParam(query, "id")) |raw| { const parsed = std.fmt.parseInt(i64, raw, 10) catch @@ -1653,6 +1691,67 @@ test "archive api key gates plan/list/segment/block and nothing else" { try testing.expect(open.code != 401); } +test "key ring: named keys authorize, meter with 429, and revoke on reload" { + const testing = std.testing; + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(io, &path_buffer); + var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]); + defer archive.deinit(); + + // a real key file: the ring reloads on mtime, so revocation below is + // exercised the way an operator does it — by editing the file + var key_path_buf: [Io.Dir.max_path_bytes + 32]u8 = undefined; + const key_path = try std.fmt.bufPrint(&key_path_buf, "{s}/keys.conf", .{path_buffer[0..path_len]}); + { + var f = try tmp.dir.createFile(io, "keys.conf", .{}); + defer f.close(io); + try f.writeStreamingAll(io, "consumer:ring-key:1\n"); + } + var ring = api_keys.KeyRing.init(testing.allocator, key_path); + defer ring.deinit(); + + var api: Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive, .archive_api_key = "fleet", .key_ring = &ring }; + + // ring key authorizes where the fleet key would too + var ok: PlanResponder = .{}; + _ = api.handle(&ok, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{}); + try testing.expect(ok.code != 401 and ok.code != 429); + + // unknown key still 401s + var bad: PlanResponder = .{}; + _ = api.handle(&bad, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer nope", .{}); + try testing.expectEqual(@as(u16, 401), bad.code); + + // drive the key into deficit: the next request is a 429; the fleet + // key is unmetered and keeps working + ring.charge(io, &ring.entries.items[0], 500_000_000, Api.nowUs(io)); + var throttled: PlanResponder = .{}; + _ = api.handle(&throttled, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{}); + try testing.expectEqual(@as(u16, 429), throttled.code); + var fleet: PlanResponder = .{}; + _ = api.handle(&fleet, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer fleet", .{}); + try testing.expect(fleet.code != 429 and fleet.code != 401); + + // revocation: rewrite the file without the key; the next authorize + // reloads (mtime changed, check throttle bypassed via last_check) + { + var f = try tmp.dir.createFile(io, "keys.conf", .{ .truncate = true }); + defer f.close(io); + try f.writeStreamingAll(io, "other:different-key\n"); + } + ring.last_check_us = std.math.minInt(i64); + ring.loaded_mtime_us = std.math.minInt(i64); + var revoked: PlanResponder = .{}; + _ = api.handle(&revoked, "GET", "/xrpc/network.bsky.jetstream.listSegments", "", "", "Bearer ring-key", .{}); + try testing.expectEqual(@as(u16, 401), revoked.code); +} + test "planBackfill never answers 200 with a short plan under allocation failure" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); diff --git a/src/main.zig b/src/main.zig index 9093afb..31595ff 100644 --- a/src/main.zig +++ b/src/main.zig @@ -14,6 +14,7 @@ const cursor_mod = @import("internal/ingest/cursor.zig"); const ingest = @import("internal/ingest/ingest.zig"); const verify = @import("internal/ingest/verify.zig"); const xrpcapi = @import("internal/serve/xrpcapi.zig"); +const api_keys_mod = @import("internal/serve/api_keys.zig"); const metrics = @import("internal/runtime/metrics.zig"); const meta_store = @import("internal/storage/meta_store.zig"); const pipeline_mod = @import("internal/ingest/pipeline.zig"); @@ -721,6 +722,15 @@ pub fn main(init: std.process.Init.Minimal) !void { xrpc_api.import = &import_manager; xrpc_api.import_token = cfg.timestamp_import_token; xrpc_api.archive_api_key = cfg.archive_api_key; + var key_ring: ?api_keys_mod.KeyRing = if (cfg.archive_api_keys_file.len > 0) + api_keys_mod.KeyRing.init(allocator, cfg.archive_api_keys_file) + else + null; + defer if (key_ring) |*ring| ring.deinit(); + if (key_ring) |*ring| { + xrpc_api.key_ring = ring; + log.info("archive key ring enabled: {s} (revocation = edit the file)", .{cfg.archive_api_keys_file}); + } // Killed getRepo downloads can leave incomplete scratch files. Bootstrap // owns backfill/, while steady repair has a separate root so a legitimate -- 2.51.2