//! Experimental permissioned-data XRPC handlers. //! //! This module is intentionally feature-flagged at dispatch and should remain //! the only HTTP entry point for `com.atproto.space.*` and //! `com.atproto.simplespace.*` behavior while the upstream protocol is still //! experimental. const std = @import("std"); const clock = @import("../core/clock.zig"); const auth = @import("../auth/tokens.zig"); const config = @import("../core/config.zig"); const http_api = @import("../http/api.zig"); const cbor_json = @import("../internal/cbor_json.zig"); const permissioned = @import("../internal/permissioned_data.zig"); const client_attestation = @import("../internal/client_attestation.zig"); const space_signature = @import("../internal/space_signature.zig"); const scopes = @import("../internal/scopes.zig"); const space_uris = @import("../internal/space_uri.zig"); const store = @import("../storage/store.zig"); const zat = @import("zat"); const http = std.http; pub fn dispatch(request: *http_api.Request) !void { if (!config.permissionedData()) { return http_api.xrpcError( request, .not_implemented, "MethodNotImplemented", "Permissioned data is experimental and disabled on this PDS", ); } const method = xrpcMethod(request.url.raw) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid XRPC path"); }; if (std.mem.eql(u8, method, "com.atproto.simplespace.getSpace")) return getSimpleSpace(request); // pre-alpha name for the same query; routable until clients move. if (std.mem.eql(u8, method, "com.atproto.space.getSpace")) return getSimpleSpace(request); if (std.mem.eql(u8, method, "com.atproto.space.listSpaces")) return listSpaces(request); if (std.mem.eql(u8, method, "com.atproto.space.listRepos")) return listRepos(request); if (std.mem.eql(u8, method, "com.atproto.space.getDelegationToken")) return getDelegationToken(request); if (std.mem.eql(u8, method, "com.atproto.space.createRecord")) return createRecord(request); if (std.mem.eql(u8, method, "com.atproto.space.putRecord")) return putRecord(request); if (std.mem.eql(u8, method, "com.atproto.space.deleteRecord")) return deleteRecord(request); if (std.mem.eql(u8, method, "com.atproto.space.applyWrites")) return applyWrites(request); if (std.mem.eql(u8, method, "com.atproto.space.getRecord")) return getRecord(request); if (std.mem.eql(u8, method, "com.atproto.space.listRecords")) return listRecords(request); if (std.mem.eql(u8, method, "com.atproto.space.getBlob")) return getBlob(request); if (std.mem.eql(u8, method, "com.atproto.space.listBlobs")) return listBlobs(request); if (std.mem.eql(u8, method, "com.atproto.space.getLatestCommit")) return getLatestCommit(request); // newer name for the same operation in the atpint/bulleted stack; both // stay routable while the draft settles on one. if (std.mem.eql(u8, method, "com.atproto.space.getRepoState")) return getLatestCommit(request); if (std.mem.eql(u8, method, "com.atproto.space.getRepo")) return getRepo(request); if (std.mem.eql(u8, method, "com.atproto.space.listRepoOps")) return listRepoOps(request); if (std.mem.eql(u8, method, "com.atproto.space.registerNotify")) return registerNotify(request); if (std.mem.eql(u8, method, "com.atproto.space.unregisterNotify")) return unregisterNotify(request); if (std.mem.eql(u8, method, "com.atproto.space.getSpaceCredential")) return getSpaceCredential(request); if (std.mem.eql(u8, method, "com.atproto.space.notifyCredentialRevoked")) return notifyCredentialRevoked(request); if (std.mem.eql(u8, method, "com.atproto.space.notifyWrite")) return notifyWrite(request); if (std.mem.eql(u8, method, "com.atproto.space.notifySpaceDeleted")) return notifySpaceDeleted(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.createSpace")) return createSimpleSpace(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.updateSpace")) return updateSimpleSpace(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.deleteSpace")) return deleteSimpleSpace(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.putMember")) return putSimpleSpaceMember(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.removeMember")) return removeSimpleSpaceMember(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.listMembers")) return listSimpleSpaceMembers(request); if (std.mem.eql(u8, method, "com.atproto.simplespace.checkUserAccess")) return checkSimpleSpaceUserAccess(request); return http_api.xrpcError( request, .not_implemented, "MethodNotImplemented", "Permissioned data is enabled but this method is not implemented yet", ); } fn createSimpleSpace(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed = parseBody(request, allocator, 64 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed.value; const authority = auth_ctx.account.did; const space_type = zat.json.getString(input, "spaceType") orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing spaceType"); }; const skey = zat.json.getString(input, "skey") orelse try store.generateRkey(allocator); try requireSpaceScope(request, auth_ctx.oauth_scope, space_type, authority, skey, .manage_create, null); if (jsonField(input, "policy") != null) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "policy was replaced by readPolicy and writePolicy"); const read_value = jsonField(input, "readPolicy") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing readPolicy"); const read_policy = policyFromLex(read_value) catch |err| return rejectPolicy(request, err); const write_value = jsonField(input, "writePolicy") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing writePolicy"); const write_policy = policyFromLex(write_value) catch |err| return rejectPolicy(request, err); const app_access_value = jsonField(input, "appAccess") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing appAccess"); const app_access_json = appAccessFromLex(allocator, app_access_value) catch |err| return rejectAppAccess(request, err); const space = store.createSpace(allocator, .{ .actor_did = authority, .authority_did = authority, .space_type = space_type, .skey = skey, .is_authority = true, .read_managing_app = read_policy.managing_app, .read_policy = read_policy.policy, .write_managing_app = write_policy.managing_app, .write_policy = write_policy.policy, .app_access_json = app_access_json, }) catch |err| switch (err) { error.InvalidRepoPath => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid authority"), error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidType", "Invalid space type"), error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid skey"), error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid simplespace config"), error.InvalidRefreshSession => return http_api.xrpcError(request, .bad_request, "SpaceAlreadyExists", "Space already exists"), else => return err, }; const body = try std.fmt.allocPrint(allocator, "{{\"uri\":{f}}}", .{std.json.fmt(space.uri, .{})}); return http_api.json(request, .ok, body); } fn getSimpleSpace(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space_uri = http_api.queryParam(request.url.raw, "space", &space_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; const parsed = parseSpaceUri(space_uri) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); }; if (http_api.requireBearerAccess(request, allocator)) |auth_ctx| { if (!std.mem.eql(u8, parsed.authority_did, auth_ctx.account.did)) { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found on this authority host"); } try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, .read, null); } else |err| { if (err == error.ExpiredToken or spaceToken(request) == null) return http_api.bearerErrorResponse(request, err); _ = requireSpaceCredential(request, allocator, space_uri, null) catch return; } const space = (try store.getSpace(allocator, parsed.authority_did, space_uri)) orelse { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); }; if (!space.is_authority or space.deleted_at != null) { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); } return http_api.json(request, .ok, try simpleSpaceViewJson(allocator, space)); } fn listSpaces(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; var did_buf: [256]u8 = undefined; const maybe_did = http_api.queryParam(request.url.raw, "did", &did_buf); if (maybe_did) |did| { if (zat.Did.parse(did) == null) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid did"); } var type_buf: [320]u8 = undefined; const maybe_type = http_api.queryParam(request.url.raw, "spaceType", &type_buf); if (maybe_type) |space_type| { if (zat.Nsid.parse(space_type) == null) return http_api.xrpcError(request, .bad_request, "InvalidType", "Invalid space type"); } if (auth_ctx.oauth_scope) |scope_text| { if (!scopes.spaceAllows(scope_text, .read, maybe_type orelse "*", maybe_did orelse "*", "*", null)) { return http_api.xrpcError(request, .forbidden, "InsufficientScope", "OAuth token does not grant the requested space operation"); } } var cursor_buf: [1024]u8 = undefined; const maybe_cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); const spaces = try store.listSpaces(allocator, auth_ctx.account.did, maybe_did, maybe_type, maybe_cursor, http_api.queryLimit(request.url.raw, 50)); var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"spaces\":["); for (spaces, 0..) |space, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print("{{\"uri\":{f},\"isOwner\":{}}}", .{ std.json.fmt(space.uri, .{}), space.is_authority }); } try out.writer.writeByte(']'); if (spaces.len > 0) { try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(spaces[spaces.len - 1].uri, .{})}); } try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn listRepos(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; _ = requireSpaceCredential(request, allocator, space, null) catch return; var cursor_buf: [256]u8 = undefined; const maybe_cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); const limit = pageLimit(request.url.raw, 100, 1000); const repos = store.listSpaceWriters(allocator, space, maybe_cursor, limit) catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), else => return err, }; var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"repos\":["); for (repos, 0..) |repo, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print("{{\"did\":{f},\"repoRev\":", .{std.json.fmt(repo.repo_did, .{})}); try out.writer.print("{f},\"hash\":", .{std.json.fmt(repo.rev, .{})}); try writeAtJsonBytes(&out.writer, allocator, repo.hash); try out.writer.print(",\"spaceRev\":{f}}}", .{std.json.fmt(repo.space_rev, .{})}); } try out.writer.writeByte(']'); if (repos.len > 0) try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(repos[repos.len - 1].space_rev, .{})}); try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn putSimpleSpaceMember(request: *http_api.Request) !void { return mutateSimpleSpaceMember(request, .put); } fn removeSimpleSpaceMember(request: *http_api.Request) !void { return mutateSimpleSpaceMember(request, .remove); } const MemberMutation = enum { put, remove }; fn mutateSimpleSpaceMember(request: *http_api.Request, mutation: MemberMutation) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const member = zat.json.getString(input, "did") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did"); try requireSimpleSpaceAuthority(request, allocator, auth_ctx, space, .manage_update); const read = if (mutation == .put) requiredBool(input, "read") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "read must be a boolean") else false; const write = if (mutation == .put) requiredBool(input, "write") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "write must be a boolean") else false; (switch (mutation) { .put => store.putSimpleSpaceMember(space, member, read, write), .remove => store.removeSimpleSpaceMember(space, member), }) catch |err| switch (err) { error.InvalidRepoPath => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid DID"), error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Cannot remove the space authority"), else => return err, }; return http_api.json(request, .ok, "{}"); } fn listSimpleSpaceMembers(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); try requireSimpleSpaceAuthority(request, allocator, auth_ctx, space, .read_self); var cursor_buf: [256]u8 = undefined; const maybe_cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); const members = try store.listSimpleSpaceMembers(allocator, space, maybe_cursor, pageLimit(request.url.raw, 100, 1000)); var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"members\":["); for (members, 0..) |member, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print("{{\"did\":{f},\"read\":{},\"write\":{}}}", .{ std.json.fmt(member.did, .{}), member.read, member.write }); } try out.writer.writeByte(']'); if (members.len > 0) try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(members[members.len - 1].did, .{})}); try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn checkSimpleSpaceUserAccess(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); var user_buf: [256]u8 = undefined; const user = http_api.queryParam(request.url.raw, "user", &user_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing user"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); if (zat.Did.parse(user) == null) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid user DID"); const service = requireServiceAuth(request, allocator, "com.atproto.simplespace.checkUserAccess") catch return; if (!std.mem.eql(u8, service.issuer_did, parsed.authority_did)) { return http_api.xrpcError(request, .unauthorized, "InvalidToken", "JWT issuer must be the space authority DID"); } if (!serviceAudienceMatchesPds(service.audience)) { return http_api.xrpcError(request, .unauthorized, "BadJwtAudience", "JWT audience does not identify this service"); } var access_buf: [16]u8 = undefined; const access = http_api.queryParam(request.url.raw, "access", &access_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing access"); if (!std.mem.eql(u8, access, "read") and !std.mem.eql(u8, access, "write")) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "access must be read or write"); return http_api.json(request, .ok, "{\"authorized\":false}"); } fn updateSimpleSpace(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed_body = parseBody(request, allocator, 64 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); if (!std.mem.eql(u8, parsed.authority_did, auth_ctx.account.did)) return http_api.xrpcError(request, .forbidden, "NotSpaceOwner", "Not the space owner"); try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, .manage_update, null); const existing = (try store.getSpace(allocator, auth_ctx.account.did, space)) orelse return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); if (!existing.is_authority) return http_api.xrpcError(request, .forbidden, "NotSpaceOwner", "Not the space owner"); if (existing.deleted_at != null) return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); if (jsonField(input, "policy") != null) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "policy was replaced by readPolicy and writePolicy"); const maybe_read: ?store.SimpleSpacePolicy = if (jsonField(input, "readPolicy")) |value| policyFromLex(value) catch |err| return rejectPolicy(request, err) else null; const maybe_write: ?store.SimpleSpacePolicy = if (jsonField(input, "writePolicy")) |value| policyFromLex(value) catch |err| return rejectPolicy(request, err) else null; const maybe_app_access: ?[]const u8 = if (jsonField(input, "appAccess")) |value| appAccessFromLex(allocator, value) catch |err| return rejectAppAccess(request, err) else null; store.updateSimpleSpaceConfig(space, maybe_read, maybe_write, maybe_app_access) catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid simplespace config"), else => return err, }; return http_api.json(request, .ok, "{}"); } fn deleteSimpleSpace(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const space = zat.json.getString(parsed_body.value, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); if (!std.mem.eql(u8, parsed.authority_did, auth_ctx.account.did)) return http_api.xrpcError(request, .forbidden, "NotSpaceOwner", "Not the space owner"); try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, .manage_delete, null); const existing = try store.getSpace(allocator, auth_ctx.account.did, space); if (existing == null) return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); if (!existing.?.is_authority) return http_api.xrpcError(request, .forbidden, "NotSpaceOwner", "Not the space owner"); if (existing.?.deleted_at != null) return http_api.json(request, .ok, "{}"); const recipients = try store.listNotificationRecipients(allocator, space); try store.markSpaceDeleted(auth_ctx.account.did, space); try store.purgeAuthoritySpaceData(space, auth_ctx.account.did); fireNotifySpaceDeleted(allocator, auth_ctx.account, space, recipients) catch {}; return http_api.json(request, .ok, "{}"); } fn getDelegationToken(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, .read, null); var keypair = store.signingKeypair(auth_ctx.account.did) catch return http_api.xrpcError(request, .not_found, "RepoNotFound", "Signing key not found"); const token = try permissioned.createDelegationToken(allocator, store.currentIo(), auth_ctx.account.did, parsed.authority_did, space, &keypair); return http_api.json(request, .ok, try std.fmt.allocPrint(allocator, "{{\"token\":{f}}}", .{std.json.fmt(token, .{})})); } fn createRecord(request: *http_api.Request) !void { return writeSpaceRecord(request, .create); } fn putRecord(request: *http_api.Request) !void { return writeSpaceRecord(request, .put); } const WriteMode = enum { create, put }; fn deleteRecord(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed_body = parseBody(request, allocator, 64 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const repo = zat.json.getString(input, "repo") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); if (!std.mem.eql(u8, repo, auth_ctx.account.did)) return http_api.xrpcError(request, .forbidden, "InvalidRepo", "repo must match authenticated user"); const collection = zat.json.getString(input, "collection") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); const rkey = zat.json.getString(input, "rkey") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"); requireSpaceAccess(request, allocator, auth_ctx, space, .delete, collection) catch return; store.deleteSpaceRecord(allocator, space, repo, collection, rkey) catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.MissingRecord => {}, error.InvalidRefreshSession => return http_api.xrpcError(request, .bad_request, "RecordAlreadyExists", "Record already exists"), else => return err, }; const state = try store.getSpaceRepoState(allocator, space, repo); fireNotifyWriteForState(allocator, auth_ctx.account, space, repo, state) catch {}; return http_api.json(request, .ok, "{}"); } fn applyWrites(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed_body = parseBody(request, allocator, 512 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const repo = zat.json.getString(input, "repo") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); if (!std.mem.eql(u8, repo, auth_ctx.account.did)) return http_api.xrpcError(request, .forbidden, "InvalidRepo", "repo must match authenticated user"); const writes_value = switch (input) { .object => |object| object.get("writes") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing writes"), else => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid body"), }; if (writes_value != .array) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "writes must be an array"); var ops: std.ArrayList(store.SpaceWriteOp) = .empty; for (writes_value.array.items) |write| { const write_type = zat.json.getString(write, "$type") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "write missing $type"); const collection = zat.json.getString(write, "collection") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "write missing collection"); if (std.mem.endsWith(u8, write_type, "#create")) { requireSpaceAccess(request, allocator, auth_ctx, space, .create, collection) catch return; const rkey = zat.json.getString(write, "rkey") orelse try store.generateRkey(allocator); const value = switch (write) { .object => |object| object.get("value") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "create missing value"), else => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid write"), }; const prepared = try prepareSpaceRecord(request, allocator, collection, rkey, value); try ops.append(allocator, .{ .create = .{ .collection = collection, .rkey = rkey, .prepared = prepared } }); } else if (std.mem.endsWith(u8, write_type, "#update")) { requireSpaceAccess(request, allocator, auth_ctx, space, .update, collection) catch return; const rkey = zat.json.getString(write, "rkey") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "update missing rkey"); const value = switch (write) { .object => |object| object.get("value") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "update missing value"), else => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid write"), }; const prepared = try prepareSpaceRecord(request, allocator, collection, rkey, value); try ops.append(allocator, .{ .update = .{ .collection = collection, .rkey = rkey, .prepared = prepared } }); } else if (std.mem.endsWith(u8, write_type, "#delete")) { requireSpaceAccess(request, allocator, auth_ctx, space, .delete, collection) catch return; const rkey = zat.json.getString(write, "rkey") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "delete missing rkey"); try ops.append(allocator, .{ .delete = .{ .collection = collection, .rkey = rkey } }); } else { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Unknown write type"); } } const results = store.applySpaceWrites(allocator, space, repo, ops.items) catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.MissingRecord => return http_api.xrpcError(request, .not_found, "RecordNotFound", "Record not found"), error.InvalidRefreshSession => return http_api.xrpcError(request, .bad_request, "RecordAlreadyExists", "Record already exists"), else => return err, }; const state = try store.getSpaceRepoState(allocator, space, repo); fireNotifyWriteForState(allocator, auth_ctx.account, space, repo, state) catch {}; var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"results\":["); for (results, 0..) |result, idx| { if (idx != 0) try out.writer.writeByte(','); switch (result) { .create => |record| try out.writer.print("{{\"$type\":\"com.atproto.space.applyWrites#createResult\",\"uri\":{f},\"cid\":{f}}}", .{ std.json.fmt(try record.uri(allocator), .{}), std.json.fmt(record.cid, .{}) }), .update => |record| try out.writer.print("{{\"$type\":\"com.atproto.space.applyWrites#updateResult\",\"uri\":{f},\"cid\":{f}}}", .{ std.json.fmt(try record.uri(allocator), .{}), std.json.fmt(record.cid, .{}) }), .delete => try out.writer.writeAll("{\"$type\":\"com.atproto.space.applyWrites#deleteResult\"}"), } } try out.writer.writeAll("]}"); return http_api.json(request, .ok, out.written()); } fn writeSpaceRecord(request: *http_api.Request, mode: WriteMode) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const auth_ctx = requireAccount(request, allocator) catch return; const parsed = parseBody(request, allocator, 256 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed.value; const space = zat.json.getString(input, "space") orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; const repo = zat.json.getString(input, "repo") orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); }; if (!std.mem.eql(u8, repo, auth_ctx.account.did)) { return http_api.xrpcError(request, .forbidden, "InvalidRepo", "repo must match authenticated user"); } const collection = zat.json.getString(input, "collection") orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); }; switch (mode) { .create => requireSpaceAccess(request, allocator, auth_ctx, space, .create, collection) catch return, .put => { requireSpaceAccess(request, allocator, auth_ctx, space, .create, collection) catch return; requireSpaceAccess(request, allocator, auth_ctx, space, .update, collection) catch return; }, } const rkey = zat.json.getString(input, "rkey") orelse switch (mode) { .create => try store.generateRkey(allocator), .put => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"), }; const record = recordBodyValue(input) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing record"); }; const prepared = store.prepareRecordValue(allocator, collection, rkey, record) catch |err| switch (err) { error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid record type"), error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), else => return err, }; const stored = switch (mode) { .create => store.createSpaceRecord(allocator, space, repo, collection, rkey, prepared), .put => store.putSpaceRecord(allocator, space, repo, collection, rkey, prepared), } catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.InvalidRefreshSession => return http_api.xrpcError(request, .bad_request, "RecordAlreadyExists", "Record already exists"), error.MissingRecord => return http_api.xrpcError(request, .bad_request, "RecordNotFound", "Record not found"), else => return err, }; const state = try store.getSpaceRepoState(allocator, space, repo); fireNotifyWriteForState(allocator, auth_ctx.account, space, repo, state) catch {}; const uri = try stored.uri(allocator); const body = try std.fmt.allocPrint( allocator, "{{\"uri\":{f},\"cid\":{f}}}", .{ std.json.fmt(uri, .{}), std.json.fmt(stored.cid, .{}) }, ); return http_api.json(request, .ok, body); } fn getRecord(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const params = recordParams(request, allocator) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; _ = requireReadAccess(request, allocator, params.space, params.repo, params.collection) catch return; const record = (try store.getSpaceRecord(allocator, params.space, params.repo, params.collection, params.rkey)) orelse { return http_api.xrpcError(request, .not_found, "RecordNotFound", "Record not found"); }; const uri = try record.uri(allocator); const body = try std.fmt.allocPrint( allocator, "{{\"uri\":{f},\"cid\":{f},\"value\":{s}}}", .{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}), record.value_json }, ); return http_api.json(request, .ok, body); } fn listRecords(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); }; var collection_buf: [320]u8 = undefined; const maybe_collection = http_api.queryParam(request.url.raw, "collection", &collection_buf); _ = requireReadAccess(request, allocator, space, repo, maybe_collection) catch return; var cursor_buf: [1024]u8 = undefined; const maybe_cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); var reverse_buf: [8]u8 = undefined; const reverse = if (http_api.queryParam(request.url.raw, "reverse", &reverse_buf)) |value| std.mem.eql(u8, value, "true") else false; var exclude_values_buf: [8]u8 = undefined; const exclude_values = if (http_api.queryParam(request.url.raw, "excludeValues", &exclude_values_buf)) |value| std.mem.eql(u8, value, "true") else false; const records = try store.listSpaceRecords( allocator, space, repo, maybe_collection, maybe_cursor, reverse, http_api.queryLimit(request.url.raw, 50), !exclude_values, ); var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"records\":["); for (records, 0..) |record, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print( "{{\"collection\":{f},\"rkey\":{f},\"cid\":{f}", .{ std.json.fmt(record.collection, .{}), std.json.fmt(record.rkey, .{}), std.json.fmt(record.cid, .{}) }, ); if (record.value_json) |value_json| { try out.writer.print(",\"value\":{s}", .{value_json}); } try out.writer.writeByte('}'); } try out.writer.writeByte(']'); if (records.len > 0) { const last = records[records.len - 1]; try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(try std.fmt.allocPrint(allocator, "{s}/{s}", .{ last.collection, last.rkey }), .{})}); } try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn getBlob(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); }; var cid_buf: [256]u8 = undefined; const cid = http_api.queryParam(request.url.raw, "cid", &cid_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing cid"); }; _ = requireReadAccess(request, allocator, space, repo, null) catch return; if (!try store.permissionedBlobReferenced(space, repo, cid)) { return http_api.xrpcError(request, .not_found, "BlobNotFound", "Blob not found"); } const blob = store.getBlob(allocator, repo, cid) orelse { return http_api.xrpcError(request, .not_found, "BlobNotFound", "Blob not found"); }; return writeBlobResponse(request, allocator, cid, blob); } fn listBlobs(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); }; var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse { return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); }; _ = requireReadAccess(request, allocator, space, repo, null) catch return; var since_buf: [128]u8 = undefined; const since = http_api.queryParam(request.url.raw, "since", &since_buf); if (since) |rev| { if (zat.Tid.parse(rev) == null) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid since revision"); } var cursor_buf: [256]u8 = undefined; const cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); const limit = listBlobsLimit(request) catch return; const cids = try store.listPermissionedBlobs(allocator, space, repo, since, cursor, limit); var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"cids\":["); for (cids, 0..) |cid, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print("{f}", .{std.json.fmt(cid, .{})}); } try out.writer.writeByte(']'); if (cids.len == limit and cids.len > 0) { try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(cids[cids.len - 1], .{})}); } try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn getLatestCommit(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); _ = requireReadAccess(request, allocator, space, repo, null) catch return; const state = try store.getSpaceRepoState(allocator, space, repo); return writeSignedState(request, allocator, space, repo, repo, state); } fn getRepo(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); _ = requireReadAccess(request, allocator, space, repo, null) catch return; if ((try store.findAccount(allocator, repo)) == null) { return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); } switch (try store.accountStatus(repo)) { .active => {}, .takendown => return http_api.xrpcError(request, .forbidden, "RepoTakendown", "Repo has been taken down"), .suspended => return http_api.xrpcError(request, .forbidden, "RepoSuspended", "Repo is suspended"), .deactivated => return http_api.xrpcError(request, .forbidden, "RepoDeactivated", "Repo is deactivated"), .deleted => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"), } const state = try store.getSpaceRepoState(allocator, space, repo); const set_hash = state.set_hash orelse return http_api.xrpcError(request, .not_found, "RepoNotFound", "Permissioned repo not found"); const rev = state.rev orelse return http_api.xrpcError(request, .not_found, "RepoNotFound", "Permissioned repo not found"); var keypair = try store.signingKeypair(repo); const commit = try permissioned.createCommit(allocator, store.currentIo(), set_hash, .{ .space = space, .author = repo, .rev = rev, }, &keypair); const records = try store.loadSpaceRepoBlocks(allocator, space, repo); const body = try permissioned.serializeRepoCar(allocator, commit, records); const headers = [_]http.Header{ .{ .name = "content-type", .value = "application/vnd.ipld.car" }, .{ .name = "access-control-allow-origin", .value = "*" }, .{ .name = "access-control-allow-private-network", .value = "true" }, .{ .name = "connection", .value = "close" }, }; return http_api.respond(request, .ok, body, &headers); } fn listRepoOps(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); var space_buf: [1024]u8 = undefined; const space = http_api.queryParam(request.url.raw, "space", &space_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); var repo_buf: [256]u8 = undefined; const repo = http_api.queryParam(request.url.raw, "repo", &repo_buf) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); _ = requireReadAccess(request, allocator, space, repo, null) catch return; var since_buf: [128]u8 = undefined; const since = http_api.queryParam(request.url.raw, "since", &since_buf); var exclude_values_buf: [8]u8 = undefined; const exclude_values = if (http_api.queryParam(request.url.raw, "excludeValues", &exclude_values_buf)) |value| std.mem.eql(u8, value, "true") else false; var cursor_buf: [160]u8 = undefined; const cursor: ?store.OplogCursor = if (http_api.queryParam(request.url.raw, "cursor", &cursor_buf)) |raw| parseOplogCursor(raw) orelse return http_api.xrpcError(request, .bad_request, "MalformedCursor", "Malformed cursor") else null; const limit = pageLimit(request.url.raw, 100, 1000); const ops = try store.listSpaceRecordOplog(allocator, space, repo, since, cursor, limit, !exclude_values); const state = try store.getSpaceRepoState(allocator, space, repo); var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"ops\":["); for (ops, 0..) |op, idx| { if (idx != 0) try out.writer.writeByte(','); try out.writer.print( "{{\"rev\":{f},\"collection\":{f},\"rkey\":{f},\"cid\":", .{ std.json.fmt(op.rev, .{}), std.json.fmt(op.collection, .{}), std.json.fmt(op.rkey, .{}) }, ); if (op.cid) |cid_value| try out.writer.print("{f}", .{std.json.fmt(cid_value, .{})}) else try out.writer.writeAll("null"); try out.writer.writeAll(",\"prev\":"); if (op.prev) |prev_value| try out.writer.print("{f}", .{std.json.fmt(prev_value, .{})}) else try out.writer.writeAll("null"); if (op.value_json) |value_json| try out.writer.print(",\"value\":{s}", .{value_json}); try out.writer.writeByte('}'); } try out.writer.writeByte(']'); if (ops.len == limit) { const last = ops[ops.len - 1]; try out.writer.print(",\"cursor\":\"{s}/{d}\"", .{ last.rev, last.idx }); } else if (try signedCommitJson(allocator, space, repo, repo, state)) |commit_json| { try out.writer.print(",\"commit\":{s}", .{commit_json}); } try out.writer.writeByte('}'); return http_api.json(request, .ok, out.written()); } fn registerNotify(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const service = zat.json.getString(input, "service") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing service"); _ = requireSpaceCredential(request, allocator, space, null) catch return; const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); if ((try store.findAccount(allocator, parsed.authority_did)) == null) { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); } const existing = try store.getSpace(allocator, parsed.authority_did, space); if (existing == null or !existing.?.is_authority or existing.?.deleted_at != null) { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); } const endpoint = serviceEndpointForId(allocator, service) catch return http_api.xrpcError(request, .bad_request, "ServiceNotResolvable", "Could not resolve notification service"); const expires_at = unixNow() + 24 * 60 * 60; try store.registerSpaceNotification(space, service, endpoint, expires_at); return http_api.json(request, .ok, try std.fmt.allocPrint( allocator, "{{\"expiresAt\":{f}}}", .{std.json.fmt(try datetimeFromUnix(allocator, expires_at), .{})}, )); } fn unregisterNotify(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const service = zat.json.getString(input, "service") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing service"); _ = requireSpaceCredential(request, allocator, space, null) catch return; const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); const existing = try store.getSpace(allocator, parsed.authority_did, space); if (existing == null or !existing.?.is_authority or existing.?.deleted_at != null) { return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); } try store.unregisterSpaceNotification(space, service); return http_api.json(request, .ok, "{}"); } fn getSpaceCredential(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const token = bearerToken(request) orelse return http_api.xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"); const holder_key = space_signature.verifyRequest(allocator, request, null, null) catch { return http_api.xrpcError(request, .unauthorized, "BadSpaceSignature", "A valid HTTP signature is required to obtain a space credential"); }; const requester_did = permissioned.unverifiedStringClaim(allocator, token, "iss") catch { return http_api.xrpcError(request, .unauthorized, "InvalidDelegationToken", "Invalid delegation token"); }; const public_key = signingKeyForDid(allocator, requester_did) catch { return http_api.xrpcError(request, .bad_gateway, "DidResolutionFailed", "could not resolve delegation token issuer did"); }; const delegation = permissioned.verifyDelegationToken(allocator, token, public_key) catch { return http_api.xrpcError(request, .unauthorized, "InvalidDelegationToken", "Invalid delegation token"); }; const parsed_body = parseBody(request, allocator, 32 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); if (!std.mem.eql(u8, delegation.space, space)) return http_api.xrpcError(request, .bad_request, "InvalidDelegationToken", "Delegation token subject does not match requested space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); if (!std.mem.eql(u8, delegation.authority_did, parsed.authority_did)) return http_api.xrpcError(request, .bad_request, "InvalidDelegationToken", "Delegation token audience does not match the space authority"); const config_row = (try store.getSpace(allocator, parsed.authority_did, space)) orelse return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); if (config_row.deleted_at != null) { return http_api.xrpcError(request, .bad_request, "SpaceDeleted", "Space has been deleted"); } const attestation_claims: ?client_attestation.Claims = if (zat.json.getString(input, "clientAttestation")) |attestation| blk: { const audience = try std.fmt.allocPrint(allocator, "{s}#atproto_space_host", .{parsed.authority_did}); break :blk client_attestation.verify(allocator, store.currentIo(), attestation, audience) catch { return http_api.xrpcError(request, .unauthorized, "InvalidClientAttestation", "Client attestation verification failed"); }; } else null; const attested_client_id = if (attestation_claims) |claims| claims.client_id else null; if (!spaceAllowsClient(allocator, config_row, attested_client_id)) { return http_api.xrpcError(request, .forbidden, "AppNotAuthorized", "This space requires a verified client attestation"); } if (!try spacePolicyAllowsRequester(allocator, config_row, delegation.requester_did, .read, attested_client_id)) { return http_api.xrpcError(request, .forbidden, "UserNotAuthorized", "The space host did not grant access to this requester"); } const attestation_replay: ?store.SpaceReplayToken = if (attestation_claims) |claims| .{ .jti = claims.jti, .expires_at = claims.exp } else null; if (!try store.consumeSpaceCredentialExchange( .{ .jti = delegation.jti, .expires_at = delegation.exp }, attestation_replay, )) { return http_api.xrpcError(request, .unauthorized, "InvalidDelegationToken", "Credential exchange token has already been used"); } var keypair = store.signingKeypair(parsed.authority_did) catch return http_api.xrpcError(request, .not_found, "RepoNotFound", "Authority signing key not found"); const credential = try permissioned.createSpaceCredential(allocator, store.currentIo(), parsed.authority_did, space, holder_key, attested_client_id, &keypair); return http_api.json(request, .ok, try std.fmt.allocPrint(allocator, "{{\"credential\":{f}}}", .{std.json.fmt(credential, .{})})); } fn notifyCredentialRevoked(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const service = requireServiceAuth(request, allocator, "com.atproto.space.notifyCredentialRevoked") catch return; const token = bearerToken(request).?; const issuer = try permissioned.unverifiedStringClaim(allocator, token, "iss"); const lxm = try permissioned.unverifiedStringClaim(allocator, token, "lxm"); if (!std.mem.eql(u8, issuer, service.issuer_did) or !std.mem.eql(u8, lxm, "com.atproto.space.notifyCredentialRevoked")) return http_api.xrpcError(request, .unauthorized, "InvalidToken", "Expected exact authority and revocation method"); const body = parseBody(request, allocator, 32 * 1024) catch return; const space = zat.json.getString(body.value, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space"); if (!std.mem.eql(u8, service.issuer_did, parsed.authority_did) or try store.findAccount(allocator, service.audience) == null) return http_api.xrpcError(request, .unauthorized, "BadJwtAudience", "Expected space authority and local repo audience"); const credentials = zat.json.getPath(body.value, "credentials") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing credentials"); if (credentials != .array or credentials.array.items.len == 0 or credentials.array.items.len > 100) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Expected 1 to 100 credential IDs"); var ids: [100][]const u8 = undefined; for (credentials.array.items, 0..) |value, i| { if (value != .string or value.string.len == 0) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid credential ID"); ids[i] = value.string; } try store.revokeSpaceCredentials(space, ids[0..credentials.array.items.len]); return http_api.json(request, .ok, "{}"); } fn notifyWrite(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const input = parsed_body.value; const space = zat.json.getString(input, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const repo = zat.json.getString(input, "repo") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); const rev = zat.json.getString(input, "repoRev") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing repoRev"); const hash = parseAtJsonBytes(allocator, switch (input) { .object => |object| object.get("hash") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing hash"), else => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid body"), }) catch return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid hash"); if (hash.len != permissioned.commit_hash_bytes) return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid hash"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); const service = requireServiceAuth(request, allocator, "com.atproto.space.notifyWrite") catch return; if (!std.mem.eql(u8, service.issuer_did, repo)) return http_api.xrpcError(request, .unauthorized, "InvalidToken", "JWT issuer must be the writer repo DID"); const audience = try std.fmt.allocPrint(allocator, "{s}#atproto_space_host", .{parsed.authority_did}); if (!std.mem.eql(u8, service.audience, parsed.authority_did) and !std.mem.eql(u8, service.audience, audience)) return http_api.xrpcError(request, .unauthorized, "BadJwtAudience", "JWT audience must be the space host"); processSpaceNotification(allocator, space, repo, rev, hash) catch |err| switch (err) { error.RepoNotFound => return http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"), error.WriterNotAuthorized => return http_api.xrpcError(request, .forbidden, "UserNotAuthorized", "Writer is not authorized"), error.FutureRev => return http_api.xrpcError(request, .bad_request, "FutureRev", "Repo revision is too far in the future"), error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid repo revision"), else => return err, }; return http_api.json(request, .ok, "{}"); } fn notifySpaceDeleted(request: *http_api.Request) !void { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); defer arena.deinit(); const allocator = arena.allocator(); const parsed_body = parseBody(request, allocator, 16 * 1024) catch |err| switch (err) { error.HandledResponse => return, else => return err, }; const space = zat.json.getString(parsed_body.value, "space") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing space"); const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); const service = requireServiceAuth(request, allocator, "com.atproto.space.notifySpaceDeleted") catch return; if (!std.mem.eql(u8, service.issuer_did, parsed.authority_did)) return http_api.xrpcError(request, .unauthorized, "InvalidToken", "JWT issuer must be the space DID"); if (zat.Did.parse(service.audience) == null and zat.Handle.parse(service.audience) == null) return http_api.json(request, .ok, "{}"); if ((store.findAccount(allocator, service.audience) catch null) == null) return http_api.json(request, .ok, "{}"); if ((try store.getSpace(allocator, service.audience, space)) == null) return http_api.json(request, .ok, "{}"); try store.markSpaceDeleted(service.audience, space); return http_api.json(request, .ok, "{}"); } const RecordParams = struct { space: []const u8, repo: []const u8, collection: []const u8, rkey: []const u8, }; fn recordParams(request: *http_api.Request, allocator: std.mem.Allocator) !RecordParams { var space_buf: [1024]u8 = undefined; var repo_buf: [256]u8 = undefined; var collection_buf: [320]u8 = undefined; var rkey_buf: [512]u8 = undefined; return .{ .space = try requiredQueryParam(request, allocator, "space", &space_buf, "Missing space"), .repo = try requiredQueryParam(request, allocator, "repo", &repo_buf, "Missing repo"), .collection = try requiredQueryParam(request, allocator, "collection", &collection_buf, "Missing collection"), .rkey = try requiredQueryParam(request, allocator, "rkey", &rkey_buf, "Missing rkey"), }; } fn requiredQueryParam( request: *http_api.Request, allocator: std.mem.Allocator, name: []const u8, buf: []u8, message: []const u8, ) ![]const u8 { const value = http_api.queryParam(request.url.raw, name, buf) orelse { try http_api.xrpcError(request, .bad_request, "InvalidRequest", message); return error.HandledResponse; }; return allocator.dupe(u8, value); } fn writeBlobResponse(request: *http_api.Request, allocator: std.mem.Allocator, cid: []const u8, blob: store.BlobRecord) !void { const range = http_api.headerValue(request, "range"); const selected = parseRange(range, blob.data.len) catch { const headers = [_]http.Header{ .{ .name = "content-range", .value = try std.fmt.allocPrint(allocator, "bytes */{d}", .{blob.data.len}) }, .{ .name = "accept-ranges", .value = "bytes" }, .{ .name = "access-control-allow-origin", .value = "*" }, }; return http_api.respond(request, .range_not_satisfiable, "", &headers); }; const start = selected.start; const end_inclusive = selected.end_inclusive; const body = blob.data[start .. end_inclusive + 1]; const status: http.Status = if (range == null) .ok else .partial_content; const content_range = try std.fmt.allocPrint(allocator, "bytes {d}-{d}/{d}", .{ start, end_inclusive, blob.data.len }); const disposition = try std.fmt.allocPrint(allocator, "attachment; filename=\"{s}\"", .{cid}); const headers = [_]http.Header{ .{ .name = "content-type", .value = if (blob.mime_type.len > 0) blob.mime_type else "application/octet-stream" }, .{ .name = "accept-ranges", .value = "bytes" }, .{ .name = "content-range", .value = content_range }, .{ .name = "x-content-type-options", .value = "nosniff" }, .{ .name = "content-disposition", .value = disposition }, .{ .name = "content-security-policy", .value = "default-src 'none'; sandbox" }, .{ .name = "access-control-allow-origin", .value = "*" }, .{ .name = "access-control-allow-private-network", .value = "true" }, }; return http_api.respond(request, status, if (request.method == .HEAD) "" else body, &headers); } fn parseRange(maybe_range: ?[]const u8, size: usize) !struct { start: usize, end_inclusive: usize } { if (size == 0) return error.InvalidRange; const raw = maybe_range orelse return .{ .start = 0, .end_inclusive = size - 1 }; if (!std.mem.startsWith(u8, raw, "bytes=")) return error.InvalidRange; const spec = raw["bytes=".len..]; const dash = std.mem.indexOfScalar(u8, spec, '-') orelse return error.InvalidRange; if (dash == 0) { const suffix = try std.fmt.parseInt(usize, spec[1..], 10); if (suffix == 0) return error.InvalidRange; const start = if (suffix >= size) 0 else size - suffix; return .{ .start = start, .end_inclusive = size - 1 }; } const start = try std.fmt.parseInt(usize, spec[0..dash], 10); const end = if (dash + 1 < spec.len) try std.fmt.parseInt(usize, spec[dash + 1 ..], 10) else size - 1; if (start >= size or end < start) return error.InvalidRange; return .{ .start = start, .end_inclusive = @min(end, size - 1) }; } fn bearerToken(request: *http_api.Request) ?[]const u8 { const raw_header = http_api.headerValue(request, "authorization") orelse return null; if (!std.ascii.startsWithIgnoreCase(raw_header, "bearer ")) return null; return std.mem.trim(u8, raw_header["bearer ".len..], " \t"); } fn spaceToken(request: *http_api.Request) ?[]const u8 { const raw_header = http_api.headerValue(request, "authorization") orelse return null; if (!std.ascii.startsWithIgnoreCase(raw_header, "atproto-space ")) return null; return std.mem.trim(u8, raw_header["atproto-space ".len..], " \t"); } fn signingKeyForDid(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { return signingKeyForDidKid(allocator, did, "#atproto"); } fn signingKeyForDidKid(allocator: std.mem.Allocator, did: []const u8, kid: []const u8) ![]const u8 { if (try store.findAccount(allocator, did) != null) { if (!std.mem.eql(u8, kid, "#atproto")) return error.MissingSigningKey; var keypair = try store.signingKeypair(did); const did_key = try keypair.did(allocator); defer allocator.free(did_key); if (!std.mem.startsWith(u8, did_key, "did:key:")) return error.InvalidSigningKey; return allocator.dupe(u8, did_key["did:key:".len..]); } const parsed = zat.Did.parse(did) orelse return error.InvalidDid; var resolver = zat.DidResolver.init(store.currentIo(), allocator); defer resolver.deinit(); var doc = try resolver.resolve(parsed); defer doc.deinit(); const absolute_kid = if (std.mem.startsWith(u8, kid, "#")) try std.fmt.allocPrint(allocator, "{s}{s}", .{ did, kid }) else try allocator.dupe(u8, kid); defer allocator.free(absolute_kid); for (doc.verification_methods) |method| { if (std.mem.eql(u8, method.id, kid) or std.mem.eql(u8, method.id, absolute_kid)) { if (!std.mem.eql(u8, method.controller, did)) return error.InvalidSigningKey; return allocator.dupe(u8, method.public_key_multibase); } } return error.MissingSigningKey; } const ServiceAuth = struct { issuer_did: []const u8, audience: []const u8, }; fn requireServiceAuth(request: *http_api.Request, allocator: std.mem.Allocator, method: []const u8) !ServiceAuth { const token = bearerToken(request) orelse { try http_api.xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"); return error.HandledResponse; }; var jwt = zat.Jwt.parse(allocator, token) catch { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); return error.HandledResponse; }; defer jwt.deinit(); const issuer_did = issuerDid(jwt.payload.iss); if (zat.Did.parse(issuer_did) == null or jwt.isExpired(store.currentIo())) { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); return error.HandledResponse; } const lxm = jwt.payload.lxm orelse { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); return error.HandledResponse; }; if (!std.mem.eql(u8, lxm, method) and !std.mem.eql(u8, lxm, "*")) { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "JWT lxm mismatch"); return error.HandledResponse; } const public_key = signingKeyForDid(allocator, issuer_did) catch { try http_api.xrpcError(request, .bad_gateway, "DidResolutionFailed", "could not resolve jwt issuer did"); return error.HandledResponse; }; jwt.verify(public_key) catch { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); return error.HandledResponse; }; return .{ .issuer_did = try allocator.dupe(u8, issuer_did), .audience = try allocator.dupe(u8, jwt.payload.aud), }; } fn issuerDid(iss: []const u8) []const u8 { return if (std.mem.indexOfScalar(u8, iss, '#')) |idx| iss[0..idx] else iss; } fn serviceAudienceMatchesPds(aud: []const u8) bool { if (std.mem.eql(u8, aud, config.serverDid())) return true; if (!std.mem.startsWith(u8, aud, config.serverDid())) return false; return std.mem.eql(u8, aud[config.serverDid().len..], "#atproto_pds"); } fn spaceHostEndpointForDid(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { const parsed = zat.Did.parse(did) orelse return error.InvalidDid; var resolver = zat.DidResolver.init(store.currentIo(), allocator); defer resolver.deinit(); var doc = try resolver.resolve(parsed); defer doc.deinit(); for (doc.services) |service| { if (std.mem.endsWith(u8, service.id, "#atproto_space_host")) { return allocator.dupe(u8, service.service_endpoint); } } return allocator.dupe(u8, doc.pdsEndpoint() orelse return error.MissingSpaceHostEndpoint); } fn serviceEndpointForId(allocator: std.mem.Allocator, service_id: []const u8) ![]const u8 { const fragment_index = std.mem.indexOfScalar(u8, service_id, '#'); const did_text = if (fragment_index) |idx| service_id[0..idx] else service_id; const fragment = if (fragment_index) |idx| service_id[idx..] else null; const did = zat.Did.parse(did_text) orelse return error.InvalidServiceId; var resolver = zat.DidResolver.init(store.currentIo(), allocator); defer resolver.deinit(); var doc = try resolver.resolve(did); defer doc.deinit(); if (fragment) |service_fragment| { for (doc.services) |service| { if (std.mem.eql(u8, service.id, service_id) or std.mem.eql(u8, service.id, service_fragment)) { return allocator.dupe(u8, service.service_endpoint); } } return error.MissingServiceEndpoint; } return allocator.dupe(u8, doc.pdsEndpoint() orelse return error.MissingServiceEndpoint); } fn managingAppAccessUrl( allocator: std.mem.Allocator, endpoint: []const u8, space: []const u8, requester_did: []const u8, access: store.SpaceAccess, client_id: ?[]const u8, ) ![]const u8 { const encoded_space = try percentEncode(allocator, space); const encoded_user = try percentEncode(allocator, requester_did); return if (if (access == .read) client_id else null) |id| try std.fmt.allocPrint( allocator, "{s}/xrpc/com.atproto.simplespace.checkUserAccess?space={s}&user={s}&access={s}&clientId={s}", .{ endpoint, encoded_space, encoded_user, @tagName(access), try percentEncode(allocator, id) }, ) else try std.fmt.allocPrint( allocator, "{s}/xrpc/com.atproto.simplespace.checkUserAccess?space={s}&user={s}&access={s}", .{ endpoint, encoded_space, encoded_user, @tagName(access) }, ); } fn managingAppAllowsRequester( allocator: std.mem.Allocator, space: store.SpaceConfig, requester_did: []const u8, access: store.SpaceAccess, client_id: ?[]const u8, ) !bool { const managing_app = (if (access == .read) space.read_managing_app else space.write_managing_app) orelse return false; const endpoint = serviceEndpointForId(allocator, managing_app) catch return false; const parsed = parseSpaceUri(space.uri) orelse return false; const authority = (try store.findAccount(allocator, parsed.authority_did)) orelse return false; var keypair = try store.signingKeypair(authority.did); const token = try auth.createServiceJwtWithKeypair( allocator, authority, managing_app, "com.atproto.simplespace.checkUserAccess", null, &keypair, ); const authorization = try std.fmt.allocPrint(allocator, "Bearer {s}", .{token}); const url = try managingAppAccessUrl(allocator, endpoint, space.uri, requester_did, access, client_id); var transport = zat.HttpTransport.init(store.currentIo(), allocator); defer transport.deinit(); var result = transport.fetch(.{ .url = url, .authorization = authorization, .accept = "application/json", .max_response_size = 16 * 1024, .on_redirect = .refuse, }) catch return false; defer result.deinit(allocator); if (result.status != .ok) return false; const response = std.json.parseFromSlice(std.json.Value, allocator, result.body, .{}) catch return false; defer response.deinit(); return switch (response.value) { .object => |object| switch (object.get("authorized") orelse return false) { .bool => |authorized| authorized, else => false, }, else => false, }; } fn postSignedXrpc( allocator: std.mem.Allocator, account: auth.Account, endpoint: []const u8, audience: []const u8, method: []const u8, body: []const u8, ) !void { var keypair = try store.signingKeypair(account.did); const token = try auth.createServiceJwtWithKeypair(allocator, account, audience, method, null, &keypair); const authorization = try std.fmt.allocPrint(allocator, "Bearer {s}", .{token}); const url = try std.fmt.allocPrint(allocator, "{s}/xrpc/{s}", .{ endpoint, method }); var transport = zat.HttpTransport.init(store.currentIo(), allocator); defer transport.deinit(); const result = try transport.fetch(.{ .url = url, .method = .POST, .payload = body, .content_type = "application/json", .authorization = authorization, .max_response_size = 64 * 1024, .on_redirect = .refuse, }); defer allocator.free(result.body); if (@backingInt(result.status) >= 500 or result.status == .too_many_requests or result.status == .request_timeout) return error.RemoteServerError; if (@backingInt(result.status) < 200 or @backingInt(result.status) >= 300) return error.RemotePermanentError; } fn processSpaceNotification(allocator: std.mem.Allocator, space: []const u8, repo: []const u8, rev: []const u8, hash: []const u8) !void { const parsed = parseSpaceUri(space) orelse return error.InvalidSpaceUri; const authority = (try store.findAccount(allocator, parsed.authority_did)) orelse return error.RepoNotFound; const row = (try store.getSpace(allocator, parsed.authority_did, space)) orelse return error.RepoNotFound; if (!row.is_authority or row.deleted_at != null) return error.RepoNotFound; if (!try spacePolicyAllowsRequester(allocator, row, repo, .write, null)) return error.WriterNotAuthorized; const update = (try store.recordSpaceWriter(allocator, space, repo, rev, hash)) orelse return; const previous = if (update.prev_space_rev) |value| try std.fmt.allocPrint(allocator, ",\"prevSpaceRev\":{f}", .{std.json.fmt(value, .{})}) else ""; const body = try std.fmt.allocPrint(allocator, "{{\"space\":{f},\"repo\":{f},\"repoRev\":{f},\"hash\":{s},\"spaceRev\":{f}{s}}}", .{ std.json.fmt(space, .{}), std.json.fmt(repo, .{}), std.json.fmt(rev, .{}), try atJsonBytes(allocator, hash), std.json.fmt(update.space_rev, .{}), previous }); for (try store.listNotificationRecipients(allocator, space)) |recipient| { postSignedXrpc(allocator, authority, recipient.service_endpoint, recipient.service_id, "com.atproto.space.notifyWrite", body) catch {}; } } fn deliverSpaceNotification(allocator: std.mem.Allocator, account: auth.Account, space: []const u8, repo: []const u8, rev: []const u8, hash: []const u8) !void { const parsed = parseSpaceUri(space) orelse return error.InvalidSpaceUri; if (try store.findAccount(allocator, parsed.authority_did) != null) { return processSpaceNotification(allocator, space, repo, rev, hash); } const endpoint = try spaceHostEndpointForDid(allocator, parsed.authority_did); const audience = try std.fmt.allocPrint(allocator, "{s}#atproto_space_host", .{parsed.authority_did}); const body = try std.fmt.allocPrint(allocator, "{{\"space\":{f},\"repo\":{f},\"repoRev\":{f},\"hash\":{s}}}", .{ std.json.fmt(space, .{}), std.json.fmt(repo, .{}), std.json.fmt(rev, .{}), try atJsonBytes(allocator, hash) }); try postSignedXrpc(allocator, account, endpoint, audience, "com.atproto.space.notifyWrite", body); } fn permanentNotificationError(err: anyerror) bool { return err == error.RemotePermanentError or err == error.WriterNotAuthorized or err == error.RepoNotFound or err == error.InvalidRecordType or err == error.FutureRev; } fn fireNotifyWriteForState(allocator: std.mem.Allocator, account: auth.Account, space: []const u8, repo: []const u8, state: store.SpaceState) !void { const rev = state.rev orelse return; const set_hash = try permissioned.LtHash.fromBytes(state.set_hash orelse return); deliverSpaceNotification(allocator, account, space, repo, rev, &set_hash.digest()) catch |err| { if (!permanentNotificationError(err)) return err; }; try store.finishSpaceNotification(space, repo, rev); } pub fn startNotificationWorker() !void { const thread = try std.Thread.spawn(.{}, notificationWorker, .{}); thread.detach(); } fn notificationWorker() void { while (true) { var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); const a = arena.allocator(); const pending = store.claimSpaceNotification(a) catch null; if (pending) |item| { deliverPendingNotification(a, item) catch {}; } arena.deinit(); if (pending == null) store.currentIo().sleep(std.Io.Duration.fromSeconds(1), .awake) catch {}; } } fn deliverPendingNotification(allocator: std.mem.Allocator, item: store.PendingSpaceNotification) !void { const account = (try store.findAccount(allocator, item.repo)) orelse { return store.finishSpaceNotification(item.space, item.repo, item.rev); }; deliverSpaceNotification(allocator, account, item.space, item.repo, item.rev, item.hash) catch |err| { if (!permanentNotificationError(err)) return store.retrySpaceNotification(item); }; try store.finishSpaceNotification(item.space, item.repo, item.rev); } fn fireNotifySpaceDeleted(allocator: std.mem.Allocator, account: auth.Account, space: []const u8, recipients: []const store.CredentialRecipient) !void { const body = try std.fmt.allocPrint(allocator, "{{\"space\":{f}}}", .{std.json.fmt(space, .{})}); for (recipients) |recipient| { postSignedXrpc(allocator, account, recipient.service_endpoint, recipient.service_id, "com.atproto.space.notifySpaceDeleted", body) catch {}; } } fn unixNow() i64 { return clock.nowSeconds(); } fn datetimeFromUnix(allocator: std.mem.Allocator, seconds: i64) ![]const u8 { const epoch_seconds = std.time.epoch.EpochSeconds{ .secs = @intCast(seconds) }; const day_seconds = epoch_seconds.getDaySeconds(); const year_day = epoch_seconds.getEpochDay().calculateYearDay(); const month_day = year_day.calculateMonthDay(); return std.fmt.allocPrint( allocator, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.000Z", .{ year_day.year, month_day.month.numeric(), month_day.day_index + 1, day_seconds.getHoursIntoDay(), day_seconds.getMinutesIntoHour(), day_seconds.getSecondsIntoMinute() }, ); } fn requireAccount(request: *http_api.Request, allocator: std.mem.Allocator) !http_api.BearerAccount { return http_api.requireBearerAccess(request, allocator) catch |err| { try http_api.bearerErrorResponse(request, err); return error.HandledResponse; }; } fn requireSpaceScope( request: *http_api.Request, maybe_scope: ?[]const u8, space_type: []const u8, did: []const u8, skey: []const u8, action: scopes.SpaceAction, collection: ?[]const u8, ) !void { if (maybe_scope == null) return; if (scopes.spaceAllows(maybe_scope, action, space_type, did, skey, collection)) return; return http_api.xrpcError(request, .forbidden, "InsufficientScope", "OAuth token does not grant the requested space operation"); } fn requireSpaceAccess( request: *http_api.Request, allocator: std.mem.Allocator, auth_ctx: http_api.BearerAccount, space: []const u8, action: scopes.SpaceAction, collection: ?[]const u8, ) !void { const parsed = parseSpaceUri(space) orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, action, collection); if (try store.getSpace(allocator, auth_ctx.account.did, space)) |visible| { if (visible.deleted_at == null) return; try http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space has been deleted"); return error.HandledResponse; } // A writer's PDS needs no prior knowledge of a space: it keeps the writer's // own repo, and the authority decides who counts as a writer when // notifyWrite arrives. First write materializes the local repo. _ = store.createSpace(allocator, .{ .actor_did = auth_ctx.account.did, .authority_did = parsed.authority_did, .space_type = parsed.space_type, .skey = parsed.skey, .is_authority = false, .read_managing_app = null, .read_policy = "member-list", .write_managing_app = null, .write_policy = "member-list", .app_access_json = "{\"type\":\"open\"}", }) catch { try http_api.xrpcError(request, .forbidden, "NotPermitted", "Not permitted to write in this space"); return error.HandledResponse; }; } fn requireSimpleSpaceAuthority( request: *http_api.Request, allocator: std.mem.Allocator, auth_ctx: http_api.BearerAccount, space: []const u8, action: scopes.SpaceAction, ) !void { const parsed = parseSpaceUri(space) orelse { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); return error.HandledResponse; }; if (!std.mem.eql(u8, parsed.authority_did, auth_ctx.account.did)) { try http_api.xrpcError(request, .forbidden, "NotSpaceOwner", "Not the space owner"); return error.HandledResponse; } try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, action, null); const existing = (try store.getSpace(allocator, auth_ctx.account.did, space)) orelse { try http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); return error.HandledResponse; }; if (!existing.is_authority or existing.deleted_at != null) { try http_api.xrpcError(request, .not_found, "SpaceNotFound", "Space not found"); return error.HandledResponse; } } const ReadAuth = union(enum) { bearer: http_api.BearerAccount, space_credential: permissioned.SpaceCredential, }; fn requireReadAccess(request: *http_api.Request, allocator: std.mem.Allocator, space: []const u8, repo: []const u8, collection: ?[]const u8) !ReadAuth { if (zat.Did.parse(repo) == null) { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid repo DID"); return error.HandledResponse; } const repo_status = store.accountStatus(repo) catch |err| switch (err) { error.RepoNotFound => { try http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); return error.HandledResponse; }, else => return err, }; if (!repo_status.isActive()) { switch (repo_status) { .active => unreachable, .takendown => try http_api.xrpcError(request, .bad_request, "RepoTakendown", "Repo has been taken down"), .suspended => try http_api.xrpcError(request, .bad_request, "RepoSuspended", "Repo is suspended"), .deactivated => try http_api.xrpcError(request, .bad_request, "RepoDeactivated", "Repo is deactivated"), .deleted => try http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"), } return error.HandledResponse; } const parsed = parseSpaceUri(space) orelse { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); return error.HandledResponse; }; if (http_api.requireBearerAccess(request, allocator)) |auth_ctx| { if (!std.mem.eql(u8, repo, auth_ctx.account.did)) { try http_api.xrpcError(request, .forbidden, "NotPermitted", "OAuth credentials can only read the holder's own permissioned repo"); return error.HandledResponse; } try requireSpaceScope(request, auth_ctx.oauth_scope, parsed.space_type, parsed.authority_did, parsed.skey, .read_self, collection); return .{ .bearer = auth_ctx }; } else |err| if (err == error.ExpiredToken or spaceToken(request) == null) { try http_api.bearerErrorResponse(request, err); return error.HandledResponse; } return .{ .space_credential = try requireSpaceCredential(request, allocator, space, repo) }; } fn listBlobsLimit(request: *http_api.Request) !usize { var buf: [16]u8 = undefined; const raw = http_api.queryParam(request.url.raw, "limit", &buf) orelse return 500; const limit = std.fmt.parseInt(usize, raw, 10) catch { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid limit"); return error.HandledResponse; }; if (limit < 1 or limit > 1000) { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "limit must be between 1 and 1000"); return error.HandledResponse; } return limit; } fn requireSpaceCredential(request: *http_api.Request, allocator: std.mem.Allocator, space: []const u8, repo: ?[]const u8) !permissioned.SpaceCredential { const parsed = parseSpaceUri(space) orelse { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid space URI"); return error.HandledResponse; }; const token = spaceToken(request) orelse { try http_api.xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"); return error.HandledResponse; }; const issuer = permissioned.unverifiedStringClaim(allocator, token, "iss") catch { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"); return error.HandledResponse; }; if (!std.mem.eql(u8, issuer, parsed.authority_did)) { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Space credential issuer must be the space authority"); return error.HandledResponse; } const kid = permissioned.unverifiedHeaderString(allocator, token, "kid") catch { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid space credential"); return error.HandledResponse; }; if (!std.mem.eql(u8, kid, "#atproto") and !std.mem.eql(u8, kid, "#atproto_space")) { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid space credential signing key"); return error.HandledResponse; } const public_key = signingKeyForDidKid(allocator, issuer, kid) catch { try http_api.xrpcError(request, .bad_gateway, "DidResolutionFailed", "could not resolve credential issuer did"); return error.HandledResponse; }; const credential = permissioned.verifySpaceCredential(allocator, token, public_key) catch { try http_api.xrpcError(request, .unauthorized, "InvalidToken", "Invalid space credential"); return error.HandledResponse; }; if (!std.mem.eql(u8, credential.space, space)) { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Credential space mismatch"); return error.HandledResponse; } if (try store.spaceCredentialRevoked(space, credential.jti)) { try http_api.xrpcError(request, .unauthorized, "CredentialRevoked", "Space credential has been revoked"); return error.HandledResponse; } _ = space_signature.verifyRequest(allocator, request, credential.holder_key, repo orelse parsed.authority_did) catch |err| { try http_api.xrpcError(request, .unauthorized, if (err == error.BadSpaceAudience) "BadSpaceAudience" else "BadSpaceSignature", "Invalid space credential signature or audience"); return error.HandledResponse; }; return credential; } fn prepareSpaceRecord( request: *http_api.Request, allocator: std.mem.Allocator, collection: []const u8, rkey: []const u8, value: std.json.Value, ) !store.PreparedRecord { return store.prepareRecordValue(allocator, collection, rkey, value) catch |err| switch (err) { error.InvalidCollection => { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"); return error.HandledResponse; }, error.InvalidRecordKey => { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"); return error.HandledResponse; }, error.InvalidRecordType => { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid record type"); return error.HandledResponse; }, error.InvalidDagCbor => { try http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"); return error.HandledResponse; }, else => return err, }; } fn writeSignedState( request: *http_api.Request, allocator: std.mem.Allocator, space: []const u8, signer_did: []const u8, user_did: []const u8, state: store.SpaceState, ) !void { if (try signedCommitJson(allocator, space, signer_did, user_did, state)) |commit_json| { return http_api.json(request, .ok, try std.fmt.allocPrint(allocator, "{{\"commit\":{s}}}", .{commit_json})); } return http_api.json(request, .ok, "{}"); } fn signedCommitJson( allocator: std.mem.Allocator, space: []const u8, signer_did: []const u8, user_did: []const u8, state: store.SpaceState, ) !?[]const u8 { const set_hash = state.set_hash orelse return null; const rev = state.rev orelse return null; _ = parseSpaceUri(space) orelse return error.InvalidSpaceUri; var keypair = try store.signingKeypair(signer_did); const commit = try permissioned.createCommit(allocator, store.currentIo(), set_hash, .{ .space = space, .author = user_did, .rev = rev, }, &keypair); return try permissioned.signedCommitJson(allocator, commit); } fn parseBody(request: *http_api.Request, allocator: std.mem.Allocator, max_len: usize) !std.json.Parsed(std.json.Value) { const body = http_api.readBodyAlloc(request, allocator, max_len) catch |err| switch (err) { error.BodyTooLarge => { try http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "Request body too large"); return error.HandledResponse; }, else => return err, }; return http_api.parseJsonBody(request, allocator, body); } fn recordBodyValue(value: std.json.Value) ?std.json.Value { return switch (value) { .object => |object| object.get("record"), else => null, }; } fn valueBool(value: std.json.Value, key: []const u8) ?bool { return switch (value) { .object => |object| switch (object.get(key) orelse return null) { .bool => |boolean| boolean, else => null, }, else => null, }; } fn spaceAllowsClient(allocator: std.mem.Allocator, space: store.SpaceConfig, client_id: ?[]const u8) bool { const parsed = std.json.parseFromSlice(std.json.Value, allocator, space.app_access_json, .{}) catch return false; defer parsed.deinit(); const object = switch (parsed.value) { .object => |object| object, else => return false, }; const kind = switch (object.get("type") orelse return false) { .string => |value| value, else => return false, }; if (std.mem.eql(u8, kind, "open")) return true; if (std.mem.eql(u8, kind, "allowList")) { const actual_client_id = client_id orelse return false; const allowed = object.get("allowed") orelse return false; if (allowed != .array) return false; for (allowed.array.items) |item| { if (item == .string and std.mem.eql(u8, item.string, actual_client_id)) return true; } } return false; } fn spacePolicyAllowsRequester( allocator: std.mem.Allocator, space: store.SpaceConfig, requester_did: []const u8, access: store.SpaceAccess, client_id: ?[]const u8, ) !bool { // The authority is the only party that can reconfigure the space, so it must // never be able to lock itself out — and reaching its own data must not // depend on a managing app being reachable. Matches the reference PDS // (SimpleSpaceManager.authorizeUser). if (std.mem.eql(u8, requester_did, space.authority_did)) return true; const policy = if (access == .read) space.read_policy else space.write_policy; if (std.mem.eql(u8, policy, "public")) return true; if (std.mem.eql(u8, policy, "member-list")) return store.simpleSpaceMemberAllows(space.uri, requester_did, access); if (std.mem.eql(u8, policy, "managing-app")) return managingAppAllowsRequester(allocator, space, requester_did, access, client_id); return false; } fn percentEncode(allocator: std.mem.Allocator, value: []const u8) ![]const u8 { var out: std.Io.Writer.Allocating = .init(allocator); for (value) |c| { if (std.ascii.isAlphanumeric(c) or c == '-' or c == '_' or c == '.' or c == '~') { try out.writer.writeByte(c); } else { try out.writer.print("%{X:0>2}", .{c}); } } return out.toOwnedSlice(); } fn atJsonBytes(allocator: std.mem.Allocator, bytes: []const u8) ![]const u8 { const encoded = try allocator.alloc(u8, std.base64.standard.Encoder.calcSize(bytes.len)); _ = std.base64.standard.Encoder.encode(encoded, bytes); return std.fmt.allocPrint(allocator, "{{\"$bytes\":{f}}}", .{std.json.fmt(encoded, .{})}); } fn writeAtJsonBytes(writer: *std.Io.Writer, allocator: std.mem.Allocator, bytes: []const u8) !void { try writer.writeAll(try atJsonBytes(allocator, bytes)); } fn parseAtJsonBytes(allocator: std.mem.Allocator, value: std.json.Value) ![]const u8 { const encoded = zat.json.getString(value, "$bytes") orelse return error.InvalidBytes; return cbor_json.decodeAtBase64(allocator, encoded); } fn requiredBool(value: std.json.Value, key: []const u8) ?bool { const field = jsonField(value, key) orelse return null; return if (field == .bool) field.bool else null; } fn jsonField(value: std.json.Value, key: []const u8) ?std.json.Value { return switch (value) { .object => |object| object.get(key), else => null, }; } fn pageLimit(target: []const u8, default: usize, max: usize) usize { const requested = http_api.queryLimit(target, default); return @min(if (requested == 0) default else requested, max); } const public_policy_type = "com.atproto.simplespace.defs#publicPolicy"; const member_list_policy_type = "com.atproto.simplespace.defs#memberListPolicy"; const managing_app_policy_type = "com.atproto.simplespace.defs#managingAppPolicy"; const open_app_access_type = "com.atproto.simplespace.defs#open"; const allow_list_app_access_type = "com.atproto.simplespace.defs#allowList"; const PolicyError = error{ UnsupportedPolicy, InvalidRequest }; fn policyFromLex(value: std.json.Value) PolicyError!store.SimpleSpacePolicy { if (value != .object) return error.InvalidRequest; const kind = zat.json.getString(value, "$type") orelse return error.InvalidRequest; if (std.mem.eql(u8, kind, public_policy_type)) return .{ .policy = "public", .managing_app = null }; if (std.mem.eql(u8, kind, member_list_policy_type)) return .{ .policy = "member-list", .managing_app = null }; if (std.mem.eql(u8, kind, managing_app_policy_type)) { const managing_app = zat.json.getString(value, "managingApp") orelse return error.InvalidRequest; if (!std.mem.startsWith(u8, managing_app, "did:")) return error.UnsupportedPolicy; return .{ .policy = "managing-app", .managing_app = managing_app }; } return error.UnsupportedPolicy; } fn rejectPolicy(request: *http_api.Request, err: PolicyError) !void { return switch (err) { error.UnsupportedPolicy => http_api.xrpcError(request, .bad_request, "UnsupportedPolicy", "The requested policy is not one this host implements"), error.InvalidRequest => http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid policy"), }; } const AppAccessError = error{ UnsupportedAppAccess, InvalidRequest, OutOfMemory }; fn appAccessFromLex(allocator: std.mem.Allocator, value: std.json.Value) AppAccessError![]const u8 { if (value != .object) return error.InvalidRequest; const kind = zat.json.getString(value, "$type") orelse return error.InvalidRequest; if (std.mem.eql(u8, kind, open_app_access_type)) return allocator.dupe(u8, "{\"type\":\"open\"}"); if (!std.mem.eql(u8, kind, allow_list_app_access_type)) return error.UnsupportedAppAccess; const allowed = jsonField(value, "allowed") orelse return error.InvalidRequest; if (allowed != .array) return error.InvalidRequest; for (allowed.array.items) |item| { if (item != .string) return error.InvalidRequest; } var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); out.writer.writeAll("{\"type\":\"allowList\",\"allowed\":[") catch return error.OutOfMemory; for (allowed.array.items, 0..) |item, idx| { if (idx != 0) out.writer.writeByte(',') catch return error.OutOfMemory; out.writer.print("{f}", .{std.json.fmt(item.string, .{})}) catch return error.OutOfMemory; } out.writer.writeAll("]}") catch return error.OutOfMemory; return out.toOwnedSlice(); } fn rejectAppAccess(request: *http_api.Request, err: AppAccessError) !void { return switch (err) { error.UnsupportedAppAccess => http_api.xrpcError(request, .bad_request, "UnsupportedAppAccess", "The requested appAccess variant is not one this host implements"), error.InvalidRequest => http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid appAccess"), error.OutOfMemory => error.OutOfMemory, }; } fn lexAppAccessJson(allocator: std.mem.Allocator, raw: []const u8) ![]const u8 { const parsed = try std.json.parseFromSlice(std.json.Value, allocator, raw, .{}); const kind = zat.json.getString(parsed.value, "type") orelse return error.InvalidRecordType; if (std.mem.eql(u8, kind, "open")) return allocator.dupe(u8, "{\"$type\":\"" ++ open_app_access_type ++ "\"}"); if (!std.mem.eql(u8, kind, "allowList")) return error.InvalidRecordType; const allowed = jsonField(parsed.value, "allowed") orelse return error.InvalidRecordType; if (allowed != .array) return error.InvalidRecordType; var out: std.Io.Writer.Allocating = .init(allocator); defer out.deinit(); try out.writer.writeAll("{\"$type\":\"" ++ allow_list_app_access_type ++ "\",\"allowed\":["); for (allowed.array.items, 0..) |item, idx| { if (item != .string) return error.InvalidRecordType; if (idx != 0) try out.writer.writeByte(','); try out.writer.print("{f}", .{std.json.fmt(item.string, .{})}); } try out.writer.writeAll("]}"); return out.toOwnedSlice(); } fn lexPolicyJson(allocator: std.mem.Allocator, policy: []const u8, managing_app: ?[]const u8) ![]const u8 { if (std.mem.eql(u8, policy, "public")) return allocator.dupe(u8, "{\"$type\":\"" ++ public_policy_type ++ "\"}"); if (std.mem.eql(u8, policy, "managing-app")) { return std.fmt.allocPrint(allocator, "{{\"$type\":\"" ++ managing_app_policy_type ++ "\",\"managingApp\":{f}}}", .{std.json.fmt(managing_app orelse "", .{})}); } return allocator.dupe(u8, "{\"$type\":\"" ++ member_list_policy_type ++ "\"}"); } fn simpleSpaceViewJson(allocator: std.mem.Allocator, space: store.SpaceConfig) ![]const u8 { return std.fmt.allocPrint(allocator, "{{\"uri\":{f},\"readPolicy\":{s},\"writePolicy\":{s},\"appAccess\":{s}}}", .{ std.json.fmt(space.uri, .{}), try lexPolicyJson(allocator, space.read_policy, space.read_managing_app), try lexPolicyJson(allocator, space.write_policy, space.write_managing_app), try lexAppAccessJson(allocator, space.app_access_json), }); } fn parseOplogCursor(raw: []const u8) ?store.OplogCursor { const slash = std.mem.indexOfScalar(u8, raw, '/') orelse return null; const rev = raw[0..slash]; if (rev.len == 0) return null; const idx = std.fmt.parseInt(i64, raw[slash + 1 ..], 10) catch return null; return .{ .rev = rev, .idx = idx }; } const parseSpaceUri = space_uris.SpaceUri.parse; fn xrpcMethod(target: []const u8) ?[]const u8 { const prefix = "/xrpc/"; if (!std.mem.startsWith(u8, target, prefix)) return null; const rest = target[prefix.len..]; const end = std.mem.indexOfScalar(u8, rest, '?') orelse rest.len; return rest[0..end]; } test "parses permissioned space uris" { const parsed = parseSpaceUri("at://did:plc:abc/space/fm.plyr.privateMedia/self").?; try std.testing.expectEqualStrings("did:plc:abc", parsed.authority_did); try std.testing.expectEqualStrings("fm.plyr.privateMedia", parsed.space_type); try std.testing.expectEqualStrings("self", parsed.skey); } test "AT JSON bytes decode with and without padding" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const expected = [_]u8{ 1, 2, 3, 4 }; for ([_][]const u8{ "{\"$bytes\":\"AQIDBA==\"}", "{\"$bytes\":\"AQIDBA\"}" }) |text| { const parsed = try std.json.parseFromSlice(std.json.Value, allocator, text, .{}); defer parsed.deinit(); try std.testing.expectEqualSlices(u8, &expected, try parseAtJsonBytes(allocator, parsed.value)); } } test "permissioned wire hashes use AT JSON bytes" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const expected = [_]u8{ 0, 1, 2, 253, 254, 255 }; const encoded = try atJsonBytes(allocator, &expected); const parsed = try std.json.parseFromSlice(std.json.Value, allocator, encoded, .{}); defer parsed.deinit(); const actual = try parseAtJsonBytes(allocator, parsed.value); try std.testing.expectEqualSlices(u8, &expected, actual); } test "simplespace policy unions round-trip through the stored shape" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const managing = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#managingAppPolicy\",\"managingApp\":\"did:web:bulletin.my#bulletin\"}", .{}); const parsed = try policyFromLex(managing.value); try std.testing.expectEqualStrings("managing-app", parsed.policy); try std.testing.expectEqualStrings("did:web:bulletin.my#bulletin", parsed.managing_app.?); try std.testing.expectEqualStrings( "{\"$type\":\"com.atproto.simplespace.defs#managingAppPolicy\",\"managingApp\":\"did:web:bulletin.my#bulletin\"}", try lexPolicyJson(allocator, parsed.policy, parsed.managing_app), ); const public = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#publicPolicy\"}", .{}); try std.testing.expectEqualStrings("public", (try policyFromLex(public.value)).policy); const members = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#memberListPolicy\"}", .{}); try std.testing.expectEqualStrings("member-list", (try policyFromLex(members.value)).policy); const not_a_did = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#managingAppPolicy\",\"managingApp\":\"https://bulletin.my\"}", .{}); try std.testing.expectError(error.UnsupportedPolicy, policyFromLex(not_a_did.value)); const unknown = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#somethingElse\"}", .{}); try std.testing.expectError(error.UnsupportedPolicy, policyFromLex(unknown.value)); const legacy_string = try std.json.parseFromSlice(std.json.Value, allocator, "\"member-list\"", .{}); try std.testing.expectError(error.InvalidRequest, policyFromLex(legacy_string.value)); } test "simplespace appAccess unions round-trip through the stored shape" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const open = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#open\"}", .{}); const stored_open = try appAccessFromLex(allocator, open.value); try std.testing.expectEqualStrings("{\"type\":\"open\"}", stored_open); try std.testing.expectEqualStrings("{\"$type\":\"com.atproto.simplespace.defs#open\"}", try lexAppAccessJson(allocator, stored_open)); const allow = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#allowList\",\"allowed\":[\"https://a.example/c.json\"]}", .{}); const stored_allow = try appAccessFromLex(allocator, allow.value); try std.testing.expectEqualStrings("{\"type\":\"allowList\",\"allowed\":[\"https://a.example/c.json\"]}", stored_allow); try std.testing.expectEqualStrings( "{\"$type\":\"com.atproto.simplespace.defs#allowList\",\"allowed\":[\"https://a.example/c.json\"]}", try lexAppAccessJson(allocator, stored_allow), ); const unknown = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#denyList\"}", .{}); try std.testing.expectError(error.UnsupportedAppAccess, appAccessFromLex(allocator, unknown.value)); const missing_allowed = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"com.atproto.simplespace.defs#allowList\"}", .{}); try std.testing.expectError(error.InvalidRequest, appAccessFromLex(allocator, missing_allowed.value)); } test "simplespace getSpace view matches the alpha lexicon" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const view = try simpleSpaceViewJson(allocator, .{ .uri = "at://did:plc:abc/space/my.bulletin.board/self", .authority_did = "did:plc:abc", .space_type = "my.bulletin.board", .skey = "self", .read_managing_app = "did:web:bulletin.my#bulletin", .read_policy = "managing-app", .write_managing_app = "did:web:bulletin.my#bulletin", .write_policy = "managing-app", .app_access_json = "{\"type\":\"open\"}", .is_authority = true, .deleted_at = null, }); try std.testing.expectEqualStrings( "{\"uri\":\"at://did:plc:abc/space/my.bulletin.board/self\",\"readPolicy\":{\"$type\":\"com.atproto.simplespace.defs#managingAppPolicy\",\"managingApp\":\"did:web:bulletin.my#bulletin\"},\"writePolicy\":{\"$type\":\"com.atproto.simplespace.defs#managingAppPolicy\",\"managingApp\":\"did:web:bulletin.my#bulletin\"},\"appAccess\":{\"$type\":\"com.atproto.simplespace.defs#open\"}}", view, ); } test "the space authority is always authorized for its own space" { const managing_app_space: store.SpaceConfig = .{ .uri = "at://did:plc:auth/space/my.bulletin.board/self", .authority_did = "did:plc:auth", .space_type = "my.bulletin.board", .skey = "self", .read_managing_app = "did:web:unreachable.invalid#bulletin", .read_policy = "managing-app", .write_managing_app = "did:web:unreachable.invalid#bulletin", .write_policy = "managing-app", .app_access_json = "{\"type\":\"open\"}", .is_authority = true, .deleted_at = null, }; // Returns before the managing-app branch would try to resolve the (here // deliberately unreachable) service, so no allocator or store is touched. try std.testing.expect(try spacePolicyAllowsRequester(std.testing.allocator, managing_app_space, "did:plc:auth", .read, null)); } test "listRepoOps cursors are rev/idx" { const cursor = parseOplogCursor("3lx7abcdefg22/4").?; try std.testing.expectEqualStrings("3lx7abcdefg22", cursor.rev); try std.testing.expectEqual(@as(i64, 4), cursor.idx); try std.testing.expect(parseOplogCursor("3lx7abcdefg22") == null); try std.testing.expect(parseOplogCursor("/4") == null); try std.testing.expect(parseOplogCursor("3lx7abcdefg22/x") == null); } test "space access selects independent read and write policies" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); try store.init(std.Options.debug_io, ":memory:"); defer store.close(); const owner = try store.createAccount(a, "access.test", "access@test.com", "password", "did:plc:access", true); var space = try store.createSpace(a, .{ .actor_did = owner.did, .authority_did = owner.did, .space_type = "test.access.space", .skey = "self", .is_authority = true, .read_policy = "public", .write_policy = "member-list", .read_managing_app = null, .write_managing_app = null, .app_access_json = "{\"type\":\"open\"}", }); try std.testing.expect(try spacePolicyAllowsRequester(a, space, "did:plc:member", .read, null)); try std.testing.expect(!try spacePolicyAllowsRequester(a, space, "did:plc:member", .write, null)); space.read_policy = "member-list"; space.write_policy = "public"; try std.testing.expect(!try spacePolicyAllowsRequester(a, space, "did:plc:member", .read, null)); try std.testing.expect(try spacePolicyAllowsRequester(a, space, "did:plc:member", .write, null)); space.write_policy = "member-list"; try store.putSimpleSpaceMember(space.uri, "did:plc:member", false, true); try std.testing.expect(!try spacePolicyAllowsRequester(a, space, "did:plc:member", .read, null)); try std.testing.expect(try spacePolicyAllowsRequester(a, space, "did:plc:member", .write, null)); try std.testing.expect(try spacePolicyAllowsRequester(a, space, owner.did, .read, null)); try std.testing.expect(try spacePolicyAllowsRequester(a, space, owner.did, .write, null)); } test "managing app checks include access and omit client identity for writes" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const prefix = "https://manager.test/xrpc/com.atproto.simplespace.checkUserAccess?space=space&user=user&access="; try std.testing.expectEqualStrings(prefix ++ "read&clientId=https%3A%2F%2Fclient.test", try managingAppAccessUrl(a, "https://manager.test", "space", "user", .read, "https://client.test")); try std.testing.expectEqualStrings(prefix ++ "write", try managingAppAccessUrl(a, "https://manager.test", "space", "user", .write, "https://client.test")); try std.testing.expectEqualStrings(prefix ++ "read", try managingAppAccessUrl(a, "https://manager.test", "space", "user", .read, null)); }