atproto pds in zig
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368const std = @import("std");const auth = @import("../auth/tokens.zig");const config = @import("../core/config.zig");const log = @import("../core/log.zig");const http_api = @import("../http/api.zig");const scopes = @import("../internal/scopes.zig");const store = @import("../storage/store.zig");const zat = @import("zat");
const http = std.http;const proxy_transport_allocator = std.heap.smp_allocator;
threadlocal var proxy_transport_ready = false;threadlocal var proxy_transport_state: zat.HttpTransport = undefined;
pub fn xrpcProxy(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator();
if (request.method != .GET and request.method != .POST) { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "XRPC proxy supports GET and POST"); }
const method = xrpcMethod(request.url.raw) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid XRPC path"); }; if (isProtectedMethod(method)) { return http_api.xrpcError(request, .bad_request, "InvalidToken", "Cannot proxy protected method"); } const proxy_to = http_api.headerValue(request, "atproto-proxy") orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing atproto-proxy header"); }; log.debug("xrpc proxy start method={s} target={s} proxy_to={s}\n", .{ method, request.url.raw, proxy_to }); const service = resolveProxyService(allocator, proxy_to) catch |err| switch (err) { error.InvalidProxyHeader => { log.err("xrpc proxy invalid header method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid atproto-proxy header"); }, error.UnsupportedDidMethod => { log.err("xrpc proxy unsupported did method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Unsupported proxy DID method"); }, error.ServiceNotFound => { log.err("xrpc proxy service not found method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Proxy service not found"); }, error.DidResolutionFailed => { log.err("xrpc proxy did resolution failed method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .bad_gateway, "UpstreamFailure", "Could not resolve proxy DID"); }, else => |e| { log.err("xrpc proxy resolve failed method={s} proxy_to={s} err={s}\n", .{ method, proxy_to, @errorName(e) }); return http_api.xrpcError(request, .bad_gateway, "UpstreamFailure", "Could not resolve proxy service"); }, };
const auth_ctx = http_api.requireBearerAccess(request, allocator) catch |err| switch (err) { error.AuthRequired => { log.debug("xrpc proxy auth missing method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"); }, error.InvalidToken => { log.err("xrpc proxy auth invalid method={s} proxy_to={s}\n", .{ method, proxy_to }); return http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); }, }; const scope_aud = try std.fmt.allocPrint(allocator, "{s}#{s}", .{ service.did, service.service_id }); if (!scopes.rpcAllows(auth_ctx.oauth_scope, scope_aud, method)) { log.debug("xrpc proxy insufficient_scope method={s} account={s} aud={s}\n", .{ method, auth_ctx.account.did, scope_aud }); return http_api.xrpcError(request, .forbidden, "InsufficientScope", "Insufficient scope"); } const account = auth_ctx.account; log.debug("xrpc proxy auth ok method={s} account={s} service_did={s} endpoint={s}\n", .{ method, account.did, service.did, service.endpoint }); const upstream_url = try std.fmt.allocPrint(allocator, "{s}{s}", .{ service.endpoint, request.url.raw });
var extra_buf: [6]std.http.Header = undefined; var extra_count: usize = 0; inline for (.{ "accept-language", "atproto-accept-labelers", "atproto-content-labelers", "x-bsky-topics", "dpop" }) |name| { if (http_api.headerValue(request, name)) |value| { extra_buf[extra_count] = .{ .name = name, .value = try allocator.dupe(u8, value) }; extra_count += 1; } }
const payload = if (request.method == .POST) try http_api.readBodyAlloc(request, allocator, 1024 * 1024) else null;
var keypair = store.signingKeypair(account.did) catch { log.err("xrpc proxy signing key missing method={s} account={s}\n", .{ method, account.did }); return http_api.xrpcError(request, .bad_gateway, "UpstreamFailure", "Could not load account signing key"); }; const service_token = try auth.createServiceJwtWithKeypair(allocator, account, service.did, method, null, &keypair); const authorization = try std.fmt.allocPrint(allocator, "Bearer {s}", .{service_token});
log.debug("xrpc proxy fetch method={s} upstream={s} extra_headers={d}\n", .{ method, upstream_url, extra_count }); const fetch_options: zat.HttpTransport.FetchOptions = .{ .url = upstream_url, .method = http_api.toStdMethod(request.method), .payload = payload, .authorization = authorization, .accept = "application/json", .extra_headers = extra_buf[0..extra_count], .max_response_size = 10 * 1024 * 1024, }; const result = proxyFetch(fetch_options) catch |err| { log.err("xrpc proxy fetch failed method={s} upstream={s} err={s}\n", .{ method, upstream_url, @errorName(err) }); return http_api.xrpcError(request, .bad_gateway, "UpstreamFailure", "Failed to reach proxied service"); }; defer proxy_transport_allocator.free(result.body); log.debug("xrpc proxy upstream method={s} status={d} body_len={d}\n", .{ method, @intFromEnum(result.status), result.body.len }); if (result.body.len == 0) return http_api.empty(request, result.status); return http_api.json(request, result.status, result.body);}
fn proxyFetch(options: zat.HttpTransport.FetchOptions) !zat.HttpTransport.FetchResult { return proxyTransport().fetch(options) catch |err| switch (err) { error.HttpConnectionClosing => { var transport = zat.HttpTransport.init(store.currentIo(), proxy_transport_allocator); transport.keep_alive = false; defer transport.deinit(); return transport.fetch(options); }, else => return err, };}
fn proxyTransport() *zat.HttpTransport { if (!proxy_transport_ready) { proxy_transport_state = zat.HttpTransport.init(store.currentIo(), proxy_transport_allocator); proxy_transport_ready = true; } return &proxy_transport_state;}
pub fn shouldProxy(request: *const http_api.Request) bool { if (request.method != .GET and request.method != .POST) return false; if (http_api.headerValue(request, "atproto-proxy") == null) return false; const method = xrpcMethod(request.url.raw) orelse return false; return !isProtectedMethod(method);}
const ProxyService = struct { did: []const u8, service_id: []const u8, endpoint: []const u8,};
fn resolveProxyService(allocator: std.mem.Allocator, proxy_to: []const u8) !ProxyService { const hash = std.mem.indexOfScalar(u8, proxy_to, '#') orelse return error.InvalidProxyHeader; if (hash == 0 or hash + 1 >= proxy_to.len) return error.InvalidProxyHeader; const did = proxy_to[0..hash]; const service_id = proxy_to[hash + 1 ..]; if (try configuredProxyService(allocator, did, service_id)) |service| return service; const did_doc = try fetchDidDocument(allocator, did); const service_ref = try std.fmt.allocPrint(allocator, "#{s}", .{service_id}); const endpoint = serviceEndpoint(did_doc.value, service_ref) orelse return error.ServiceNotFound; return .{ .did = try allocator.dupe(u8, did), .service_id = try allocator.dupe(u8, service_id), .endpoint = try allocator.dupe(u8, std.mem.trim(u8, endpoint, "/")), };}
fn configuredProxyService(allocator: std.mem.Allocator, did: []const u8, service_id: []const u8) !?ProxyService { if (!std.mem.eql(u8, service_id, config.proxyServiceId())) return null; if (!std.mem.eql(u8, did, config.proxyServiceDid())) return null; return .{ .did = try allocator.dupe(u8, did), .service_id = try allocator.dupe(u8, service_id), .endpoint = try allocator.dupe(u8, config.proxyServiceUrl()), };}
fn fetchDidDocument(allocator: std.mem.Allocator, did: []const u8) !std.json.Parsed(std.json.Value) { const url = if (std.mem.startsWith(u8, did, "did:plc:")) try std.fmt.allocPrint(allocator, "https://plc.directory/{s}", .{did}) else if (std.mem.startsWith(u8, did, "did:web:")) try didWebUrl(allocator, did) else return error.UnsupportedDidMethod;
var transport = zat.HttpTransport.init(store.currentIo(), allocator); defer transport.deinit(); const result = try transport.fetch(.{ .url = url, .method = .GET, .accept = "application/json", .max_response_size = 256 * 1024, }); const status_code = @intFromEnum(result.status); if (status_code < 200 or status_code >= 300) return error.DidResolutionFailed; return std.json.parseFromSlice(std.json.Value, allocator, result.body, .{});}
fn didWebUrl(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { const body = did["did:web:".len..]; const decoded_buf = try allocator.alloc(u8, body.len); const decoded = try percentDecodeDidWeb(body, decoded_buf); const path = try allocator.dupe(u8, decoded); std.mem.replaceScalar(u8, path, ':', '/'); if (std.mem.indexOfScalar(u8, path, '/') == null) { return std.fmt.allocPrint(allocator, "https://{s}/.well-known/did.json", .{path}); } return std.fmt.allocPrint(allocator, "https://{s}/did.json", .{path});}
fn percentDecodeDidWeb(input: []const u8, out: []u8) ![]const u8 { var write: usize = 0; var read: usize = 0; while (read < input.len) { switch (input[read]) { '%' => { if (read + 2 >= input.len) return error.InvalidPercentEncoding; out[write] = try std.fmt.parseInt(u8, input[read + 1 .. read + 3], 16); read += 3; }, else => |c| { out[write] = c; read += 1; }, } write += 1; } return out[0..write];}
fn serviceEndpoint(did_doc: std.json.Value, service_id: []const u8) ?[]const u8 { const services = zat.json.getArray(did_doc, "service") orelse return null; for (services) |service_value| { if (zat.json.getString(service_value, "id")) |id| { if (std.mem.eql(u8, id, service_id)) { return zat.json.getString(service_value, "serviceEndpoint"); } } } return null;}
fn xrpcMethod(target: []const u8) ?[]const u8 { if (!std.mem.startsWith(u8, target, "/xrpc/")) return null; const start = "/xrpc/".len; const end = std.mem.indexOfScalarPos(u8, target, start, '?') orelse target.len; return target[start..end];}
fn isProtectedMethod(method: []const u8) bool { const protected = [_][]const u8{ "app.bsky.actor.getPreferences", "app.bsky.actor.putPreferences", "com.atproto.admin.deleteAccount", "com.atproto.admin.disableAccountInvites", "com.atproto.admin.disableInviteCodes", "com.atproto.admin.enableAccountInvites", "com.atproto.admin.getAccountInfo", "com.atproto.admin.getAccountInfos", "com.atproto.admin.getInviteCodes", "com.atproto.admin.getSubjectStatus", "com.atproto.admin.searchAccounts", "com.atproto.admin.sendEmail", "com.atproto.admin.updateAccountEmail", "com.atproto.admin.updateAccountHandle", "com.atproto.admin.updateAccountPassword", "com.atproto.admin.updateSubjectStatus", "com.atproto.identity.getRecommendedDidCredentials", "com.atproto.identity.requestPlcOperationSignature", "com.atproto.identity.signPlcOperation", "com.atproto.identity.submitPlcOperation", "com.atproto.identity.updateHandle", "com.atproto.repo.applyWrites", "com.atproto.repo.createRecord", "com.atproto.repo.deleteRecord", "com.atproto.repo.importRepo", "com.atproto.repo.putRecord", "com.atproto.repo.uploadBlob", "com.atproto.server.activateAccount", "com.atproto.server.checkAccountStatus", "com.atproto.server.confirmEmail", "com.atproto.server.confirmSignup", "com.atproto.server.createAccount", "com.atproto.server.createAppPassword", "com.atproto.server.createInviteCode", "com.atproto.server.createInviteCodes", "com.atproto.server.createSession", "com.atproto.server.createTotpSecret", "com.atproto.server.deactivateAccount", "com.atproto.server.deleteAccount", "com.atproto.server.deletePasskey", "com.atproto.server.deleteSession", "com.atproto.server.describeServer", "com.atproto.server.disableTotp", "com.atproto.server.enableTotp", "com.atproto.server.finishPasskeyRegistration", "com.atproto.server.getAccountInviteCodes", "com.atproto.server.getServiceAuth", "com.atproto.server.getSession", "com.atproto.server.getTotpStatus", "com.atproto.server.listAppPasswords", "com.atproto.server.listPasskeys", "com.atproto.server.refreshSession", "com.atproto.server.regenerateBackupCodes", "com.atproto.server.requestAccountDelete", "com.atproto.server.requestEmailConfirmation", "com.atproto.server.requestEmailUpdate", "com.atproto.server.requestPasswordReset", "com.atproto.server.resendMigrationVerification", "com.atproto.server.resendVerification", "com.atproto.server.reserveSigningKey", "com.atproto.server.resetPassword", "com.atproto.server.revokeAppPassword", "com.atproto.server.startPasskeyRegistration", "com.atproto.server.updateEmail", "com.atproto.server.updatePasskey", "com.atproto.server.verifyMigrationEmail", "com.atproto.sync.getBlob", "com.atproto.sync.getBlocks", "com.atproto.sync.getCheckout", "com.atproto.sync.getHead", "com.atproto.sync.getLatestCommit", "com.atproto.sync.getRecord", "com.atproto.sync.getRepo", "com.atproto.sync.getRepoStatus", "com.atproto.sync.listBlobs", "com.atproto.sync.listRepos", "com.atproto.sync.notifyOfUpdate", "com.atproto.sync.requestCrawl", "com.atproto.sync.subscribeRepos", "com.atproto.temp.checkSignupQueue", "com.atproto.temp.dereferenceScope", }; for (protected) |item| { if (std.ascii.eqlIgnoreCase(method, item)) return true; } return false;}
test "proxy protected methods are explicit" { try std.testing.expect(isProtectedMethod("com.atproto.repo.applyWrites")); try std.testing.expect(isProtectedMethod("com.atproto.sync.subscribeRepos")); try std.testing.expect(!isProtectedMethod("com.atproto.identity.resolveHandle")); try std.testing.expect(!isProtectedMethod("com.atproto.repo.getRecord")); try std.testing.expect(!isProtectedMethod("com.atproto.repo.listRecords"));}
test "configured proxy service skips did resolution" { const before_did = config.proxyServiceDid(); const before_id = config.proxyServiceId(); const before_url = config.proxyServiceUrl(); defer config.setProxyServiceDid(before_did); defer config.setProxyServiceId(before_id); defer config.setProxyServiceUrl(before_url); config.setProxyServiceDid("did:web:api.bsky.app"); config.setProxyServiceId("bsky_appview"); config.setProxyServiceUrl("https://api.bsky.app");
const service = (try resolveProxyService(std.testing.allocator, "did:web:api.bsky.app#bsky_appview")); defer std.testing.allocator.free(service.did); defer std.testing.allocator.free(service.service_id); defer std.testing.allocator.free(service.endpoint);
try std.testing.expectEqualStrings("did:web:api.bsky.app", service.did); try std.testing.expectEqualStrings("bsky_appview", service.service_id); try std.testing.expectEqualStrings("https://api.bsky.app", service.endpoint);}