From 70e802764bc2a12eddaf79d06e6d989dba4de746 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Thu, 16 Oct 2025 00:05:45 -0500 Subject: [PATCH] crazy refactor to start using holds embedded pds for crew/captain validation --- cmd/appview/config.go | 22 +- cmd/appview/serve.go | 55 +- cmd/hold/main.go | 28 +- docker-compose.yml | 4 +- docs/SAILOR.md | 8 +- gen/main.go | 14 +- .../db/migrations/0003_add_crew_cache.yaml | 20 + pkg/appview/db/schema.go | 19 + pkg/appview/jetstream/backfill.go | 191 +++++- pkg/appview/middleware/registry.go | 56 +- pkg/appview/storage/proxy_blob_store.go | 82 ++- pkg/appview/storage/routing_repository.go | 8 +- pkg/{hold/pds => atproto}/cbor_gen.go | 2 +- pkg/atproto/client.go | 36 ++ pkg/atproto/lexicon.go | 45 +- pkg/atproto/profile.go | 40 +- pkg/auth/hold_authorizer.go | 77 +++ pkg/auth/hold_local.go | 101 ++++ pkg/auth/hold_remote.go | 559 ++++++++++++++++++ pkg/auth/oauth/server.go | 22 +- pkg/auth/token/handler.go | 22 +- pkg/hold/authorization.go | 181 ------ pkg/hold/{pds => }/blobstore_adapter.go | 15 +- pkg/hold/pds/captain.go | 25 +- pkg/hold/pds/crew.go | 24 +- pkg/hold/pds/did.go | 39 +- pkg/hold/pds/server.go | 104 +++- pkg/hold/pds/types.go | 32 - pkg/hold/pds/xrpc.go | 11 +- pkg/hold/service.go | 56 +- 30 files changed, 1530 insertions(+), 368 deletions(-) create mode 100644 pkg/appview/db/migrations/0003_add_crew_cache.yaml rename pkg/{hold/pds => atproto}/cbor_gen.go (99%) create mode 100644 pkg/auth/hold_authorizer.go create mode 100644 pkg/auth/hold_local.go create mode 100644 pkg/auth/hold_remote.go delete mode 100644 pkg/hold/authorization.go rename pkg/hold/{pds => }/blobstore_adapter.go (69%) delete mode 100644 pkg/hold/pds/types.go diff --git a/cmd/appview/config.go b/cmd/appview/config.go index d481702..cc8a99f 100644 --- a/cmd/appview/config.go +++ b/cmd/appview/config.go @@ -1,6 +1,8 @@ package main import ( + "crypto/rand" + "encoding/hex" "fmt" "net/url" "os" @@ -71,8 +73,22 @@ func buildHTTPConfig() (configuration.HTTP, error) { addr := getEnvOrDefault("ATCR_HTTP_ADDR", ":5000") debugAddr := getEnvOrDefault("ATCR_DEBUG_ADDR", ":5001") + // HTTP secret - only needed for multipart uploads in distribution's storage driver + // Since AppView is stateless and routes all storage through middleware, this isn't + // actually used, but we generate a random secret for defense in depth + httpSecret := os.Getenv("REGISTRY_HTTP_SECRET") + if httpSecret == "" { + // Generate a random 32-byte secret + randomBytes := make([]byte, 32) + if _, err := rand.Read(randomBytes); err != nil { + return configuration.HTTP{}, fmt.Errorf("failed to generate random secret: %w", err) + } + httpSecret = hex.EncodeToString(randomBytes) + } + return configuration.HTTP{ - Addr: addr, + Addr: addr, + Secret: httpSecret, Headers: map[string][]string{ "X-Content-Type-Options": {"nosniff"}, }, @@ -108,12 +124,16 @@ func buildStorageConfig() configuration.Storage { // buildMiddlewareConfig creates middleware configuration func buildMiddlewareConfig(defaultHold string) map[string][]configuration.Middleware { + // Check test mode + testMode := os.Getenv("TEST_MODE") == "true" + return map[string][]configuration.Middleware{ "registry": { { Name: "atproto-resolver", Options: configuration.Parameters{ "default_storage_endpoint": defaultHold, + "test_mode": testMode, }, }, }, diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index 5274ee7..20097be 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -20,6 +20,8 @@ import ( "github.com/spf13/cobra" "atcr.io/pkg/appview/middleware" + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth" "atcr.io/pkg/auth/oauth" "atcr.io/pkg/auth/token" @@ -147,8 +149,20 @@ func serveRegistry(cmd *cobra.Command, args []string) error { metricsDB := db.NewMetricsDB(uiDatabase) middleware.SetGlobalDatabase(metricsDB) + // 6.6. Create RemoteHoldAuthorizer for hold authorization with caching + holdAuthorizer := auth.NewRemoteHoldAuthorizer(uiDatabase) + middleware.SetGlobalAuthorizer(holdAuthorizer) + fmt.Println("Hold authorizer initialized with database caching") + + // 6.7. Extract default hold DID for OAuth server and backfill worker + // This is used to create sailor profiles on first login and cache captain records + // Expected format: "did:web:hold01.atcr.io" + // To find a hold's DID, visit: https://hold01.atcr.io/.well-known/did.json + // The extraction function normalizes URLs to DIDs for consistency + defaultHoldDID := extractDefaultHoldDID(config) + // 7. Initialize UI routes with OAuth app, refresher, and device store - uiTemplates, uiRouter := initializeUIRoutes(uiDatabase, uiReadOnlyDB, uiSessionStore, oauthApp, refresher, baseURL, deviceStore) + uiTemplates, uiRouter := initializeUIRoutes(uiDatabase, uiReadOnlyDB, uiSessionStore, oauthApp, refresher, baseURL, deviceStore, defaultHoldDID) // 8. Create OAuth server oauthServer := oauth.NewServer(oauthApp) @@ -161,12 +175,11 @@ func serveRegistry(cmd *cobra.Command, args []string) error { // Connect database for user avatar management oauthServer.SetDatabase(uiDatabase) - // 8.5. Extract default hold endpoint and set it on OAuth server + // 8.5. Set default hold DID on OAuth server (extracted earlier) // This is used to create sailor profiles on first login - defaultHoldEndpoint := extractDefaultHoldEndpoint(config) - if defaultHoldEndpoint != "" { - oauthServer.SetDefaultHoldEndpoint(defaultHoldEndpoint) - fmt.Printf("OAuth server will create profiles with default hold: %s\n", defaultHoldEndpoint) + if defaultHoldDID != "" { + oauthServer.SetDefaultHoldDID(defaultHoldDID) + fmt.Printf("OAuth server will create profiles with default hold: %s\n", defaultHoldDID) } // 9. Initialize auth keys and create token issuer @@ -227,8 +240,8 @@ func serveRegistry(cmd *cobra.Command, args []string) error { // Mount auth endpoints if enabled if issuer != nil { // Basic Auth token endpoint (supports device secrets and app passwords) - // Reuse defaultHoldEndpoint extracted earlier - tokenHandler := token.NewHandler(issuer, deviceStore, defaultHoldEndpoint) + // Reuse defaultHoldDID extracted earlier + tokenHandler := token.NewHandler(issuer, deviceStore, defaultHoldDID) tokenHandler.RegisterRoutes(mux) // Device authorization endpoints (public) @@ -351,8 +364,11 @@ func getIntParam(params configuration.Parameters, key string, defaultValue int) return defaultValue } -// extractDefaultHoldEndpoint extracts the default hold endpoint from middleware config -func extractDefaultHoldEndpoint(config *configuration.Configuration) string { +// extractDefaultHoldDID extracts the default hold DID from middleware config +// Returns a DID (e.g., "did:web:hold01.atcr.io") for consistency +// Accepts both DIDs and URLs in config for backward compatibility +// To find a hold's DID, visit: https://hold-url/.well-known/did.json +func extractDefaultHoldDID(config *configuration.Configuration) string { // Navigate through: middleware.registry[].options.default_storage_endpoint registryMiddleware, ok := config.Middleware["registry"] if !ok { @@ -369,7 +385,9 @@ func extractDefaultHoldEndpoint(config *configuration.Configuration) string { // Extract options - options is configuration.Parameters which is map[string]any if mw.Options != nil { if endpoint, ok := mw.Options["default_storage_endpoint"].(string); ok { - return endpoint + // Normalize to DID (handles both URLs and DIDs) + // This ensures we store DIDs consistently + return atproto.ResolveHoldDIDFromURL(endpoint) } } } @@ -447,7 +465,8 @@ func initializeDatabase() (*sql.DB, *sql.DB, *db.SessionStore) { // initializeUIRoutes initializes the web UI routes // database: read-write connection for auth and writes // readOnlyDB: read-only connection for public queries (search, user pages, etc.) -func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.SessionStore, oauthApp *oauth.App, refresher *oauth.Refresher, baseURL string, deviceStore *db.DeviceStore) (*template.Template, *mux.Router) { +// defaultHoldDID: DID of the default hold service (e.g., "did:web:hold01.atcr.io") +func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.SessionStore, oauthApp *oauth.App, refresher *oauth.Refresher, baseURL string, deviceStore *db.DeviceStore, defaultHoldDID string) (*template.Template, *mux.Router) { // Check if UI is enabled uiEnabled := os.Getenv("ATCR_UI_ENABLED") if uiEnabled == "false" { @@ -647,12 +666,20 @@ func initializeUIRoutes(database *sql.DB, readOnlyDB *sql.DB, sessionStore *db.S relayEndpoint = "https://relay1.us-east.bsky.network" } - backfillWorker, err := jetstream.NewBackfillWorker(database, relayEndpoint) + // Check test mode + testMode := os.Getenv("TEST_MODE") == "true" + + backfillWorker, err := jetstream.NewBackfillWorker(database, relayEndpoint, defaultHoldDID, testMode) if err != nil { fmt.Printf("Warning: Failed to create backfill worker: %v\n", err) } else { - // Run initial backfill + // Run initial backfill with startup delay for Docker compose go func() { + // Wait for hold service to be ready (Docker startup race condition) + startupDelay := 5 * time.Second + fmt.Printf("Backfill: Waiting %s for services to be ready...\n", startupDelay) + time.Sleep(startupDelay) + fmt.Printf("Backfill: Starting sync-based backfill from %s...\n", relayEndpoint) if err := backfillWorker.Start(context.Background()); err != nil { fmt.Printf("Backfill: Finished with error: %v\n", err) diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 0c41949..cb52327 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -23,13 +23,8 @@ func main() { log.Fatalf("Failed to load config: %v", err) } - // Create hold service - service, err := hold.NewHoldService(cfg) - if err != nil { - log.Fatalf("Failed to create hold service: %v", err) - } - // Initialize embedded PDS if database path is configured + // This must happen before creating HoldService since service needs PDS for authorization var holdPDS *pds.HoldPDS var xrpcHandler *pds.XRPCHandler if cfg.Database.Path != "" { @@ -49,13 +44,22 @@ func main() { log.Fatalf("Failed to bootstrap PDS: %v", err) } - // Create blob store adapter - blobStore := pds.NewHoldServiceBlobStore(service, holdDID) - - // Create XRPC handler - xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, blobStore) - log.Printf("Embedded PDS initialized successfully") + } else { + log.Fatalf("Database path is required for embedded PDS authorization") + } + + // Create hold service with PDS + service, err := hold.NewHoldService(cfg, holdPDS) + if err != nil { + log.Fatalf("Failed to create hold service: %v", err) + } + + // Create blob store adapter and XRPC handler + if holdPDS != nil { + holdDID := holdPDS.DID() + blobStore := hold.NewHoldServiceBlobStore(service, holdDID) + xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, blobStore) } // Setup HTTP routes diff --git a/docker-compose.yml b/docker-compose.yml index 28ef289..4bf5b5e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -13,10 +13,12 @@ services: environment: # Server configuration ATCR_HTTP_ADDR: :5000 - ATCR_DEFAULT_HOLD: http://atcr-hold:8080 + ATCR_DEFAULT_HOLD: http://172.28.0.3:8080 # UI configuration ATCR_UI_ENABLED: true ATCR_BACKFILL_ENABLED: true + # Test mode - fallback to default hold when user's hold is unreachable + TEST_MODE: true # Logging ATCR_LOG_LEVEL: info volumes: diff --git a/docs/SAILOR.md b/docs/SAILOR.md index d28b15c..cdde2d9 100644 --- a/docs/SAILOR.md +++ b/docs/SAILOR.md @@ -31,7 +31,7 @@ User approved Claude's plan: 4. Create Profile Management File: pkg/atproto/profile.go (new file) - - EnsureProfile(ctx, client, defaultHoldEndpoint) function + - EnsureProfile(ctx, client, defaultHoldDID) function - Logic: check if profile exists, create with default if not 5. Update Auth Handlers @@ -39,7 +39,7 @@ User approved Claude's plan: Files: pkg/auth/exchange/handler.go and pkg/auth/token/service.go - Call EnsureProfile() after token validation - Use authenticated client (has write access to user's PDS) - - Pass AppView's default_hold_endpoint config + - Pass AppView's default_hold_did config (format: "did:web:hold01.atcr.io") 6. Update Hold Resolution @@ -89,8 +89,8 @@ Progress Summary 5. Updated /auth/exchange handler to manage profile âŗ In Progress: - - Need to update /auth/token handler similarly (add defaultHoldEndpoint parameter and profile management) - - Fix compilation error in extractDefaultHoldEndpoint() - should use configuration.Middleware type not any + - Need to update /auth/token handler similarly (add defaultHoldDID parameter and profile management) + - Fix compilation error in extractDefaultHoldDID() - should use configuration.Middleware type not any 🔜 Remaining: - Update findStorageEndpoint() for new priority logic (check profile → own hold → default) diff --git a/gen/main.go b/gen/main.go index 205441d..b848078 100644 --- a/gen/main.go +++ b/gen/main.go @@ -7,8 +7,8 @@ package main // Usage: // go run gen/main.go // -// This creates pkg/hold/pds/cbor_gen.go which should be committed to git. -// Only re-run when you modify types in pkg/hold/pds/types.go +// This creates pkg/atproto/cbor_gen.go which should be committed to git. +// Only re-run when you modify types in pkg/atproto/types.go import ( "fmt" @@ -16,18 +16,18 @@ import ( cbg "github.com/whyrusleeping/cbor-gen" - "atcr.io/pkg/hold/pds" + "atcr.io/pkg/atproto" ) func main() { // Generate map-style encoders for CrewRecord and CaptainRecord - if err := cbg.WriteMapEncodersToFile("pkg/hold/pds/cbor_gen.go", "pds", - pds.CrewRecord{}, - pds.CaptainRecord{}, + if err := cbg.WriteMapEncodersToFile("pkg/atproto/cbor_gen.go", "atproto", + atproto.CrewRecord{}, + atproto.CaptainRecord{}, ); err != nil { fmt.Printf("Failed to generate CBOR encoders: %v\n", err) os.Exit(1) } - fmt.Println("Generated CBOR encoders in pkg/hold/pds/cbor_gen.go") + fmt.Println("Generated CBOR encoders in pkg/atproto/cbor_gen.go") } diff --git a/pkg/appview/db/migrations/0003_add_crew_cache.yaml b/pkg/appview/db/migrations/0003_add_crew_cache.yaml new file mode 100644 index 0000000..f4595b8 --- /dev/null +++ b/pkg/appview/db/migrations/0003_add_crew_cache.yaml @@ -0,0 +1,20 @@ +description: Add crew cache tables for authorization with exponential backoff +query: | + CREATE TABLE IF NOT EXISTS hold_crew_approvals ( + hold_did TEXT NOT NULL, + user_did TEXT NOT NULL, + approved_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + expires_at TIMESTAMP NOT NULL, + PRIMARY KEY(hold_did, user_did) + ); + CREATE INDEX IF NOT EXISTS idx_crew_approvals_expires ON hold_crew_approvals(expires_at); + + CREATE TABLE IF NOT EXISTS hold_crew_denials ( + hold_did TEXT NOT NULL, + user_did TEXT NOT NULL, + denial_count INTEGER NOT NULL DEFAULT 1, + next_retry_at TIMESTAMP NOT NULL, + last_denied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(hold_did, user_did) + ); + CREATE INDEX IF NOT EXISTS idx_crew_denials_retry ON hold_crew_denials(next_retry_at); diff --git a/pkg/appview/db/schema.go b/pkg/appview/db/schema.go index 93c36a2..2af4624 100644 --- a/pkg/appview/db/schema.go +++ b/pkg/appview/db/schema.go @@ -179,6 +179,25 @@ CREATE TABLE IF NOT EXISTS hold_captain_records ( updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_hold_captain_updated ON hold_captain_records(updated_at); + +CREATE TABLE IF NOT EXISTS hold_crew_approvals ( + hold_did TEXT NOT NULL, + user_did TEXT NOT NULL, + approved_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + expires_at TIMESTAMP NOT NULL, + PRIMARY KEY(hold_did, user_did) +); +CREATE INDEX IF NOT EXISTS idx_crew_approvals_expires ON hold_crew_approvals(expires_at); + +CREATE TABLE IF NOT EXISTS hold_crew_denials ( + hold_did TEXT NOT NULL, + user_did TEXT NOT NULL, + denial_count INTEGER NOT NULL DEFAULT 1, + next_retry_at TIMESTAMP NOT NULL, + last_denied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(hold_did, user_did) +); +CREATE INDEX IF NOT EXISTS idx_crew_denials_retry ON hold_crew_denials(next_retry_at); ` // InitDB initializes the SQLite database with the schema diff --git a/pkg/appview/jetstream/backfill.go b/pkg/appview/jetstream/backfill.go index 9821805..7471895 100644 --- a/pkg/appview/jetstream/backfill.go +++ b/pkg/appview/jetstream/backfill.go @@ -17,9 +17,11 @@ import ( // BackfillWorker uses com.atproto.sync.listReposByCollection to backfill historical data type BackfillWorker struct { - db *sql.DB - client *atproto.Client - directory identity.Directory + db *sql.DB + client *atproto.Client + directory identity.Directory + defaultHoldDID string // Default hold DID from AppView config (e.g., "did:web:hold01.atcr.io") + testMode bool // If true, suppress warnings for external holds } // BackfillState tracks backfill progress @@ -34,14 +36,18 @@ type BackfillState struct { } // NewBackfillWorker creates a backfill worker using sync API -func NewBackfillWorker(database *sql.DB, relayEndpoint string) (*BackfillWorker, error) { +// defaultHoldDID should be in format "did:web:hold01.atcr.io" +// To find a hold's DID, visit: https://hold-url/.well-known/did.json +func NewBackfillWorker(database *sql.DB, relayEndpoint, defaultHoldDID string, testMode bool) (*BackfillWorker, error) { // Create client for relay - used only for listReposByCollection client := atproto.NewClient(relayEndpoint, "", "") return &BackfillWorker{ - db: database, - client: client, // This points to the relay - directory: identity.DefaultDirectory(), + db: database, + client: client, // This points to the relay + directory: identity.DefaultDirectory(), + defaultHoldDID: defaultHoldDID, + testMode: testMode, }, nil } @@ -49,10 +55,20 @@ func NewBackfillWorker(database *sql.DB, relayEndpoint string) (*BackfillWorker, func (b *BackfillWorker) Start(ctx context.Context) error { fmt.Println("Backfill: Starting sync-based backfill...") + // First, query and cache the default hold's captain record + if b.defaultHoldDID != "" { + fmt.Printf("Backfill: Querying default hold captain record: %s\n", b.defaultHoldDID) + if err := b.queryCaptainRecord(ctx, b.defaultHoldDID); err != nil { + fmt.Printf("WARNING: Failed to query default hold captain record: %v\n", err) + // Don't fail the whole backfill - just warn + } + } + collections := []string{ - atproto.ManifestCollection, // io.atcr.manifest - atproto.TagCollection, // io.atcr.tag - atproto.StarCollection, // io.atcr.sailor.star + atproto.ManifestCollection, // io.atcr.manifest + atproto.TagCollection, // io.atcr.tag + atproto.StarCollection, // io.atcr.sailor.star + atproto.SailorProfileCollection, // io.atcr.sailor.profile } for _, collection := range collections { @@ -267,6 +283,8 @@ func (b *BackfillWorker) processRecord(ctx context.Context, did, collection stri return b.processTagRecord(did, record) case atproto.StarCollection: return b.processStarRecord(did, record) + case atproto.SailorProfileCollection: + return b.processSailorProfileRecord(ctx, did, record) default: return fmt.Errorf("unsupported collection: %s", collection) } @@ -364,6 +382,159 @@ func (b *BackfillWorker) processStarRecord(did string, record *atproto.Record) e return db.UpsertStar(b.db, did, starRecord.Subject.DID, starRecord.Subject.Repository, starRecord.CreatedAt) } +// processSailorProfileRecord processes a sailor profile record +// Extracts defaultHold and queries the hold's captain record to cache it +func (b *BackfillWorker) processSailorProfileRecord(ctx context.Context, did string, record *atproto.Record) error { + var profileRecord atproto.SailorProfileRecord + if err := json.Unmarshal(record.Value, &profileRecord); err != nil { + return fmt.Errorf("failed to unmarshal sailor profile: %w", err) + } + + // Skip if no default hold set + if profileRecord.DefaultHold == "" { + return nil + } + + // Convert hold URL/DID to canonical DID + holdDID := atproto.ResolveHoldDIDFromURL(profileRecord.DefaultHold) + if holdDID == "" { + fmt.Printf("WARNING [backfill]: Invalid hold reference in profile for %s: %s\n", did, profileRecord.DefaultHold) + return nil + } + + // Query and cache the captain record + if err := b.queryCaptainRecord(ctx, holdDID); err != nil { + // In test mode, only warn about default hold (local hold) + // External/production holds may not have captain records yet (dev ahead of prod) + if b.testMode && holdDID != b.defaultHoldDID { + // Suppress warning for external holds in test mode + return nil + } + fmt.Printf("WARNING [backfill]: Failed to query captain record for hold %s: %v\n", holdDID, err) + // Don't fail the whole backfill - just skip this hold + return nil + } + + return nil +} + +// queryCaptainRecord queries a hold's captain record and caches it in the database +func (b *BackfillWorker) queryCaptainRecord(ctx context.Context, holdDID string) error { + // Check if we already have it cached (skip if recently updated) + existing, err := db.GetCaptainRecord(b.db, holdDID) + if err == nil && existing != nil { + // If cached within last hour, skip refresh + if time.Since(existing.UpdatedAt) < 1*time.Hour { + return nil + } + } + + // Resolve hold DID to URL + // For did:web, we need to fetch .well-known/did.json + holdURL, err := resolveHoldDIDToURL(ctx, holdDID) + if err != nil { + return fmt.Errorf("failed to resolve hold DID to URL: %w", err) + } + + // Create client for hold's PDS + holdClient := atproto.NewClient(holdURL, holdDID, "") + + // Query captain record with retries (for Docker startup timing) + var record *atproto.Record + maxRetries := 3 + for attempt := 1; attempt <= maxRetries; attempt++ { + record, err = holdClient.GetRecord(ctx, "io.atcr.hold.captain", "self") + if err == nil { + break + } + + // Retry on connection errors (hold service might still be starting) + if attempt < maxRetries && strings.Contains(err.Error(), "connection refused") { + fmt.Printf("Backfill: Hold not ready (attempt %d/%d), retrying in 2s...\n", attempt, maxRetries) + time.Sleep(2 * time.Second) + continue + } + + return fmt.Errorf("failed to get captain record: %w", err) + } + + // Parse captain record from the record's Value field + var captainRecord struct { + Owner string `json:"owner"` + Public bool `json:"public"` + AllowAllCrew bool `json:"allowAllCrew"` + DeployedAt string `json:"deployedAt"` + Region string `json:"region"` + Provider string `json:"provider"` + } + + if err := json.Unmarshal(record.Value, &captainRecord); err != nil { + return fmt.Errorf("failed to parse captain record: %w", err) + } + + // Cache in database + dbRecord := &db.HoldCaptainRecord{ + HoldDID: holdDID, + OwnerDID: captainRecord.Owner, + Public: captainRecord.Public, + AllowAllCrew: captainRecord.AllowAllCrew, + DeployedAt: captainRecord.DeployedAt, + Region: captainRecord.Region, + Provider: captainRecord.Provider, + UpdatedAt: time.Now(), + } + + if err := db.UpsertCaptainRecord(b.db, dbRecord); err != nil { + return fmt.Errorf("failed to cache captain record: %w", err) + } + + fmt.Printf("Backfill: Cached captain record for hold %s (owner: %s)\n", holdDID, captainRecord.Owner) + return nil +} + +// resolveHoldDIDToURL resolves a hold DID to its service endpoint URL +// Fetches the DID document and returns both the canonical DID and service endpoint +func resolveHoldDIDToURL(ctx context.Context, inputDID string) (string, error) { + // For did:web, construct the .well-known URL + if !strings.HasPrefix(inputDID, "did:web:") { + return "", fmt.Errorf("only did:web is supported, got: %s", inputDID) + } + + // Extract hostname from did:web:hostname[:port] + hostname := strings.TrimPrefix(inputDID, "did:web:") + + // Try HTTP first (for local Docker), then HTTPS + var serviceEndpoint string + for _, scheme := range []string{"http", "https"} { + testURL := fmt.Sprintf("%s://%s/.well-known/did.json", scheme, hostname) + + // Fetch DID document (use NewClient to initialize httpClient) + client := atproto.NewClient("", "", "") + didDoc, err := client.FetchDIDDocument(ctx, testURL) + if err == nil && didDoc != nil { + // Extract service endpoint from DID document + for _, service := range didDoc.Service { + if service.Type == "AtprotoPersonalDataServer" || service.Type == "AtcrHoldService" { + serviceEndpoint = service.ServiceEndpoint + break + } + } + + if serviceEndpoint != "" { + fmt.Printf("DEBUG [backfill]: Resolved %s → canonical DID: %s, endpoint: %s\n", + inputDID, didDoc.ID, serviceEndpoint) + return serviceEndpoint, nil + } + } + } + + // Fallback: assume the hold service is at the root of the hostname + // Try HTTP first for local development + url := fmt.Sprintf("http://%s", hostname) + fmt.Printf("WARNING [backfill]: Failed to fetch DID document for %s, using fallback URL: %s\n", inputDID, url) + return url, nil +} + // ensureUser resolves and upserts a user by DID func (b *BackfillWorker) ensureUser(ctx context.Context, did string) error { // Check if user already exists diff --git a/pkg/appview/middleware/registry.go b/pkg/appview/middleware/registry.go index 949cbd9..11e69d1 100644 --- a/pkg/appview/middleware/registry.go +++ b/pkg/appview/middleware/registry.go @@ -29,6 +29,9 @@ var globalDatabase interface { IncrementPushCount(did, repository string) error } +// Global authorizer instance (set by main.go for hold authorization) +var globalAuthorizer auth.HoldAuthorizer + // SetGlobalRefresher sets the global OAuth refresher instance func SetGlobalRefresher(refresher *oauth.Refresher) { globalRefresher = refresher @@ -42,6 +45,11 @@ func SetGlobalDatabase(database interface { globalDatabase = database } +// SetGlobalAuthorizer sets the global authorizer instance for hold access control +func SetGlobalAuthorizer(authorizer auth.HoldAuthorizer) { + globalAuthorizer = authorizer +} + func init() { // Register the name resolution middleware registrymw.Register("atproto-resolver", initATProtoResolver) @@ -52,6 +60,7 @@ type NamespaceResolver struct { distribution.Namespace directory identity.Directory defaultStorageEndpoint string + testMode bool // If true, fallback to default hold when user's hold is unreachable repositories sync.Map // Cache of RoutingRepository instances by key (did:reponame) } @@ -61,15 +70,24 @@ func initATProtoResolver(ctx context.Context, ns distribution.Namespace, _ drive directory := identity.DefaultDirectory() // Get default storage endpoint from config (optional) + // Normalize to DID format for consistency defaultStorageEndpoint := "" if endpoint, ok := options["default_storage_endpoint"].(string); ok { - defaultStorageEndpoint = endpoint + // Convert URL to DID if needed (or pass through if already a DID) + defaultStorageEndpoint = atproto.ResolveHoldDIDFromURL(endpoint) + } + + // Check test mode from options (passed via env var) + testMode := false + if tm, ok := options["test_mode"].(bool); ok { + testMode = tm } return &NamespaceResolver{ Namespace: ns, directory: directory, defaultStorageEndpoint: defaultStorageEndpoint, + testMode: testMode, }, nil } @@ -177,8 +195,8 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Create routing repository - routes manifests to ATProto, blobs to hold service // The registry is stateless - no local storage is used - // Pass storage endpoint and DID as parameters (can't use context as it gets lost) - routingRepo := storage.NewRoutingRepository(repo, atprotoClient, repositoryName, storageEndpoint, did, globalDatabase) + // Pass storage endpoint, DID, and authorizer as parameters (can't use context as it gets lost) + routingRepo := storage.NewRoutingRepository(repo, atprotoClient, repositoryName, storageEndpoint, did, globalDatabase, globalAuthorizer) // Cache the repository nr.repositories.Store(cacheKey, routingRepo) @@ -206,7 +224,8 @@ func (nr *NamespaceResolver) BlobStatter() distribution.BlobStatter { // 1. User's sailor profile defaultHold (if set) // 2. User's own hold record (io.atcr.hold) // 3. AppView's default hold endpoint -// Returns the storage endpoint URL, or empty string if none configured +// Returns a hold DID (e.g., "did:web:hold01.atcr.io"), or empty string if none configured +// Note: Despite returning a DID, this is used as the "storage endpoint" throughout the code func (nr *NamespaceResolver) findStorageEndpoint(ctx context.Context, did, pdsEndpoint string) string { // Create ATProto client (without auth - reading public records) client := atproto.NewClient(pdsEndpoint, did, "") @@ -219,7 +238,15 @@ func (nr *NamespaceResolver) findStorageEndpoint(ctx context.Context, did, pdsEn } if profile != nil && profile.DefaultHold != "" { - // Profile exists with defaultHold set - use it + // Profile exists with defaultHold set + // In test mode, verify it's reachable before using it + if nr.testMode { + if nr.isHoldReachable(ctx, profile.DefaultHold) { + return profile.DefaultHold + } + fmt.Printf("DEBUG [registry/middleware/testmode]: User's defaultHold %s unreachable, falling back to default\n", profile.DefaultHold) + return nr.defaultStorageEndpoint + } return profile.DefaultHold } @@ -247,3 +274,22 @@ func (nr *NamespaceResolver) findStorageEndpoint(ctx context.Context, did, pdsEn // 3. No profile defaultHold and no own hold records - use AppView default return nr.defaultStorageEndpoint } + +// isHoldReachable checks if a hold service is reachable +// Used in test mode to fallback to default hold when user's hold is unavailable +func (nr *NamespaceResolver) isHoldReachable(ctx context.Context, holdDID string) bool { + // Try to fetch the DID document + hostname := strings.TrimPrefix(holdDID, "did:web:") + + // Try HTTP first (local), then HTTPS + for _, scheme := range []string{"http", "https"} { + testURL := fmt.Sprintf("%s://%s/.well-known/did.json", scheme, hostname) + client := atproto.NewClient("", "", "") + _, err := client.FetchDIDDocument(ctx, testURL) + if err == nil { + return true + } + } + + return false +} diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 86ae347..c818adb 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -10,6 +10,8 @@ import ( "sync" "time" + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth" "github.com/distribution/distribution/v3" "github.com/opencontainers/go-digest" ) @@ -34,11 +36,17 @@ type ProxyBlobStore struct { did string database DatabaseMetrics repository string + authorizer auth.HoldAuthorizer + holdDID string } // NewProxyBlobStore creates a new proxy blob store -func NewProxyBlobStore(storageEndpoint, did string, database DatabaseMetrics, repository string) *ProxyBlobStore { - fmt.Printf("DEBUG [proxy_blob_store]: NewProxyBlobStore created with endpoint=%s, did=%s, repo=%s\n", storageEndpoint, did, repository) +func NewProxyBlobStore(storageEndpoint, did string, database DatabaseMetrics, repository string, authorizer auth.HoldAuthorizer) *ProxyBlobStore { + // Convert storage endpoint URL to did:web DID for authorization + holdDID := atproto.ResolveHoldDIDFromURL(storageEndpoint) + fmt.Printf("DEBUG [proxy_blob_store]: NewProxyBlobStore created with endpoint=%s, holdDID=%s, userDID=%s, repo=%s\n", + storageEndpoint, holdDID, did, repository) + return &ProxyBlobStore{ storageEndpoint: storageEndpoint, httpClient: &http.Client{ @@ -54,11 +62,56 @@ func NewProxyBlobStore(storageEndpoint, did string, database DatabaseMetrics, re did: did, database: database, repository: repository, + authorizer: authorizer, + holdDID: holdDID, } } +// checkReadAccess verifies the user has read access to the hold +func (p *ProxyBlobStore) checkReadAccess(ctx context.Context) error { + if p.authorizer == nil { + // No authorizer configured - allow access (backward compatibility) + return nil + } + + hasAccess, err := p.authorizer.CheckReadAccess(ctx, p.holdDID, p.did) + if err != nil { + return fmt.Errorf("authorization check failed: %w", err) + } + + if !hasAccess { + return distribution.ErrBlobUnknown // Return same error as missing blob for security + } + + return nil +} + +// checkWriteAccess verifies the user has write access to the hold +func (p *ProxyBlobStore) checkWriteAccess(ctx context.Context) error { + if p.authorizer == nil { + // No authorizer configured - allow access (backward compatibility) + return nil + } + + hasAccess, err := p.authorizer.CheckWriteAccess(ctx, p.holdDID, p.did) + if err != nil { + return fmt.Errorf("authorization check failed: %w", err) + } + + if !hasAccess { + return fmt.Errorf("write access denied to hold %s", p.holdDID) + } + + return nil +} + // Stat returns the descriptor for a blob func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { + // Check read access + if err := p.checkReadAccess(ctx); err != nil { + return distribution.Descriptor{}, err + } + // Get presigned HEAD URL url, err := p.getHeadURL(ctx, dgst) if err != nil { @@ -96,6 +149,11 @@ func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribu // Get retrieves a blob func (p *ProxyBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, error) { + // Check read access + if err := p.checkReadAccess(ctx); err != nil { + return nil, err + } + url, err := p.getDownloadURL(ctx, dgst) if err != nil { return nil, err @@ -117,6 +175,11 @@ func (p *ProxyBlobStore) Get(ctx context.Context, dgst digest.Digest) ([]byte, e // Open returns a reader for a blob func (p *ProxyBlobStore) Open(ctx context.Context, dgst digest.Digest) (io.ReadSeekCloser, error) { + // Check read access + if err := p.checkReadAccess(ctx); err != nil { + return nil, err + } + url, err := p.getDownloadURL(ctx, dgst) if err != nil { return nil, err @@ -141,6 +204,11 @@ func (p *ProxyBlobStore) Open(ctx context.Context, dgst digest.Digest) (io.ReadS // Put stores a blob func (p *ProxyBlobStore) Put(ctx context.Context, mediaType string, content []byte) (distribution.Descriptor, error) { + // Check write access + if err := p.checkWriteAccess(ctx); err != nil { + return distribution.Descriptor{}, err + } + // Calculate digest dgst := digest.FromBytes(content) @@ -189,6 +257,11 @@ func (p *ProxyBlobStore) Delete(ctx context.Context, dgst digest.Digest) error { // ServeBlob serves a blob via HTTP redirect func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r *http.Request, dgst digest.Digest) error { + // Check read access + if err := p.checkReadAccess(ctx); err != nil { + return err + } + // For HEAD requests, redirect to presigned HEAD URL if r.Method == http.MethodHead { url, err := p.getHeadURL(ctx, dgst) @@ -214,6 +287,11 @@ func (p *ProxyBlobStore) ServeBlob(ctx context.Context, w http.ResponseWriter, r // Create returns a blob writer for uploading using multipart upload func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.BlobCreateOption) (distribution.BlobWriter, error) { + // Check write access + if err := p.checkWriteAccess(ctx); err != nil { + return nil, err + } + // Parse options var opts distribution.CreateOptions for _, option := range options { diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index ae1bc42..c8b1b39 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -6,6 +6,7 @@ import ( "time" "atcr.io/pkg/atproto" + "atcr.io/pkg/auth" "github.com/distribution/distribution/v3" ) @@ -26,6 +27,7 @@ type RoutingRepository struct { manifestStore *atproto.ManifestStore // Cached manifest store instance blobStore *ProxyBlobStore // Cached blob store instance database DatabaseMetrics // Database for metrics tracking + authorizer auth.HoldAuthorizer // Authorization for hold access } // NewRoutingRepository creates a new routing repository @@ -36,6 +38,7 @@ func NewRoutingRepository( storageEndpoint string, did string, database DatabaseMetrics, + authorizer auth.HoldAuthorizer, ) *RoutingRepository { return &RoutingRepository{ Repository: baseRepo, @@ -44,6 +47,7 @@ func NewRoutingRepository( storageEndpoint: storageEndpoint, did: did, database: database, + authorizer: authorizer, } } @@ -105,8 +109,8 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { panic("storage endpoint not set in RoutingRepository - ensure default_storage_endpoint is configured in middleware") } - // Create and cache proxy blob store - r.blobStore = NewProxyBlobStore(holdEndpoint, r.did, r.database, r.repositoryName) + // Create and cache proxy blob store with authorization + r.blobStore = NewProxyBlobStore(holdEndpoint, r.did, r.database, r.repositoryName, r.authorizer) return r.blobStore } diff --git a/pkg/hold/pds/cbor_gen.go b/pkg/atproto/cbor_gen.go similarity index 99% rename from pkg/hold/pds/cbor_gen.go rename to pkg/atproto/cbor_gen.go index e2576b4..5650fba 100644 --- a/pkg/hold/pds/cbor_gen.go +++ b/pkg/atproto/cbor_gen.go @@ -1,6 +1,6 @@ // Code generated by github.com/whyrusleeping/cbor-gen. DO NOT EDIT. -package pds +package atproto import ( "fmt" diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index bb41187..4047d20 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -625,3 +625,39 @@ func (c *Client) GetProfileRecord(ctx context.Context, did string) (*ProfileReco func BlobCDNURL(didOrHandle, cid string) string { return fmt.Sprintf("https://imgs.blue/%s/%s", didOrHandle, cid) } + +// DIDDocument represents a did:web document +type DIDDocument struct { + Context []string `json:"@context"` + ID string `json:"id"` + Service []struct { + ID string `json:"id"` + Type string `json:"type"` + ServiceEndpoint string `json:"serviceEndpoint"` + } `json:"service"` +} + +// FetchDIDDocument fetches and parses a DID document from a URL +func (c *Client) FetchDIDDocument(ctx context.Context, didDocURL string) (*DIDDocument, error) { + req, err := http.NewRequestWithContext(ctx, "GET", didDocURL, nil) + if err != nil { + return nil, err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch DID document: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("fetch DID document failed with status %d", resp.StatusCode) + } + + var didDoc DIDDocument + if err := json.NewDecoder(resp.Body).Decode(&didDoc); err != nil { + return nil, fmt.Errorf("failed to decode DID document: %w", err) + } + + return &didDoc, nil +} diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index 5a69876..181f667 100644 --- a/pkg/atproto/lexicon.go +++ b/pkg/atproto/lexicon.go @@ -1,5 +1,7 @@ package atproto +//go:generate go run github.com/whyrusleeping/cbor-gen --map-encoding CrewRecord CaptainRecord + import ( "encoding/base64" "encoding/json" @@ -19,9 +21,19 @@ const ( // HoldCollection is the collection name for storage holds (BYOS) HoldCollection = "io.atcr.hold" - // HoldCrewCollection is the collection name for hold crew (membership) + // HoldCrewCollection is the collection name for hold crew (membership) - LEGACY BYOS model + // Stored in owner's PDS for BYOS holds HoldCrewCollection = "io.atcr.hold.crew" + // CaptainCollection is the collection name for captain records (hold ownership) - EMBEDDED PDS model + // Stored in hold's embedded PDS (singleton record at rkey "self") + CaptainCollection = "io.atcr.hold.captain" + + // CrewCollection is the collection name for crew records (access control) - EMBEDDED PDS model + // Stored in hold's embedded PDS (one record per member) + // Note: Uses same collection name as HoldCrewCollection but stored in different PDS (hold's PDS vs owner's PDS) + CrewCollection = "io.atcr.hold.crew" + // SailorProfileCollection is the collection name for user profiles SailorProfileCollection = "io.atcr.sailor.profile" @@ -371,3 +383,34 @@ func ResolveHoldDIDFromURL(holdURL string) string { // did:web uses hostname directly (port included if non-standard) return "did:web:" + hostname } + +// ============================================================================= +// Embedded PDS Types (Hold Service) +// ============================================================================= + +// CaptainRecord represents the hold's ownership and metadata +// Collection: io.atcr.hold.captain (singleton record at rkey "self") +// Stored in the hold's embedded PDS to identify the hold owner and settings +// Uses CBOR encoding for efficient storage in hold's carstore +type CaptainRecord struct { + Type string `json:"$type" cborgen:"$type"` + Owner string `json:"owner" cborgen:"owner"` // DID of hold owner + Public bool `json:"public" cborgen:"public"` // Public read access + AllowAllCrew bool `json:"allowAllCrew" cborgen:"allowAllCrew"` // Allow any authenticated user to register as crew + DeployedAt string `json:"deployedAt" cborgen:"deployedAt"` // RFC3339 timestamp + Region string `json:"region,omitempty" cborgen:"region,omitempty"` // S3 region (optional) + Provider string `json:"provider,omitempty" cborgen:"provider,omitempty"` // Deployment provider (optional) +} + +// CrewRecord represents a crew member in the hold +// Collection: io.atcr.hold.crew (one record per member) +// Stored in the hold's embedded PDS for access control +// Uses CBOR encoding for efficient storage in hold's carstore +// Note: Same collection name as HoldCrewRecord but stored in hold's PDS (not owner's PDS) +type CrewRecord struct { + Type string `json:"$type" cborgen:"$type"` + Member string `json:"member" cborgen:"member"` + Role string `json:"role" cborgen:"role"` + Permissions []string `json:"permissions" cborgen:"permissions"` + AddedAt string `json:"addedAt" cborgen:"addedAt"` // RFC3339 timestamp +} diff --git a/pkg/atproto/profile.go b/pkg/atproto/profile.go index f3b4652..4fdb078 100644 --- a/pkg/atproto/profile.go +++ b/pkg/atproto/profile.go @@ -12,8 +12,10 @@ const ProfileRKey = "self" // EnsureProfile checks if a user's profile exists and creates it if needed // This should be called during authentication (OAuth exchange or token service) -// If defaultHoldEndpoint is provided, creates profile with that default (or empty if not provided) -func EnsureProfile(ctx context.Context, client *Client, defaultHoldEndpoint string) error { +// If defaultHoldDID is provided, creates profile with that default (or empty if not provided) +// Expected format: "did:web:hold01.atcr.io" +// Normalizes URLs to DIDs for consistency (for backward compatibility) +func EnsureProfile(ctx context.Context, client *Client, defaultHoldDID string) error { // Check if profile already exists profile, err := client.GetRecord(ctx, SailorProfileCollection, ProfileRKey) if err == nil && profile != nil { @@ -21,21 +23,28 @@ func EnsureProfile(ctx context.Context, client *Client, defaultHoldEndpoint stri return nil } + // Normalize to DID if it's a URL (or pass through if already a DID) + // This ensures we store DIDs consistently in new profiles + normalizedDID := "" + if defaultHoldDID != "" { + normalizedDID = ResolveHoldDIDFromURL(defaultHoldDID) + } + // Profile doesn't exist - create it - // defaultHoldEndpoint can be empty string (user will need to configure it later) - newProfile := NewSailorProfileRecord(defaultHoldEndpoint) + newProfile := NewSailorProfileRecord(normalizedDID) _, err = client.PutRecord(ctx, SailorProfileCollection, ProfileRKey, newProfile) if err != nil { return fmt.Errorf("failed to create sailor profile: %w", err) } - fmt.Printf("DEBUG [profile]: Created sailor profile with defaultHold=%s\n", defaultHoldEndpoint) + fmt.Printf("DEBUG [profile]: Created sailor profile with defaultHold=%s\n", normalizedDID) return nil } // GetProfile retrieves the user's profile from their PDS // Returns nil if profile doesn't exist +// Automatically migrates old URL-based defaultHold values to DIDs func GetProfile(ctx context.Context, client *Client) (*SailorProfileRecord, error) { record, err := client.GetRecord(ctx, SailorProfileCollection, ProfileRKey) if err != nil { @@ -52,11 +61,32 @@ func GetProfile(ctx context.Context, client *Client) (*SailorProfileRecord, erro return nil, fmt.Errorf("failed to parse profile: %w", err) } + // Migrate old URL-based defaultHold to DID format + // This ensures backward compatibility with profiles created before DID migration + if profile.DefaultHold != "" && !isDID(profile.DefaultHold) { + // Convert URL to DID transparently + profile.DefaultHold = ResolveHoldDIDFromURL(profile.DefaultHold) + fmt.Printf("DEBUG [profile]: Migrated defaultHold URL to DID: %s\n", profile.DefaultHold) + } + return &profile, nil } +// isDID checks if a string is a DID (starts with "did:") +func isDID(s string) bool { + return len(s) > 4 && s[:4] == "did:" +} + // UpdateProfile updates the user's profile +// Normalizes defaultHold to DID format before saving func UpdateProfile(ctx context.Context, client *Client, profile *SailorProfileRecord) error { + // Normalize defaultHold to DID if it's a URL + // This ensures we always store DIDs, even if user provides a URL + if profile.DefaultHold != "" && !isDID(profile.DefaultHold) { + profile.DefaultHold = ResolveHoldDIDFromURL(profile.DefaultHold) + fmt.Printf("DEBUG [profile]: Normalized defaultHold to DID: %s\n", profile.DefaultHold) + } + _, err := client.PutRecord(ctx, SailorProfileCollection, ProfileRKey, profile) if err != nil { return fmt.Errorf("failed to update profile: %w", err) diff --git a/pkg/auth/hold_authorizer.go b/pkg/auth/hold_authorizer.go new file mode 100644 index 0000000..4df460e --- /dev/null +++ b/pkg/auth/hold_authorizer.go @@ -0,0 +1,77 @@ +package auth + +import ( + "context" + "fmt" + + "atcr.io/pkg/atproto" +) + +// HoldAuthorizer checks if a DID has read/write access to a hold +// Implementations can query local PDS (hold service) or remote XRPC (appview) +type HoldAuthorizer interface { + // CheckReadAccess checks if userDID can read from holdDID + // Returns: (allowed bool, error) + CheckReadAccess(ctx context.Context, holdDID, userDID string) (bool, error) + + // CheckWriteAccess checks if userDID can write to holdDID + // Returns: (allowed bool, error) + CheckWriteAccess(ctx context.Context, holdDID, userDID string) (bool, error) + + // GetCaptainRecord retrieves the captain record for a hold + // Used to check public flag and allowAllCrew settings + GetCaptainRecord(ctx context.Context, holdDID string) (*atproto.CaptainRecord, error) + + // IsCrewMember checks if userDID is a crew member of holdDID + IsCrewMember(ctx context.Context, holdDID, userDID string) (bool, error) +} + +// CheckReadAccessWithCaptain implements the standard read authorization logic +// This is shared across all HoldAuthorizer implementations +// Read access rules: +// - Public hold: allow anyone (even anonymous) +// - Private hold: require authentication (any authenticated user) +func CheckReadAccessWithCaptain(captain *atproto.CaptainRecord, userDID string) bool { + if captain.Public { + // Public hold - allow anyone (even anonymous) + return true + } + + // Private hold - require authentication + // Any authenticated user with a DID can read + if userDID == "" { + // Anonymous user trying to access private hold + return false + } + + // For MVP: assume DID presence means they have sailor.profile + // Future: could query PDS to verify sailor.profile exists + return true +} + +// CheckWriteAccessWithCaptain implements the standard write authorization logic +// This is shared across all HoldAuthorizer implementations +// Write access rules: +// - Must be authenticated +// - Must be hold owner OR crew member +func CheckWriteAccessWithCaptain(captain *atproto.CaptainRecord, userDID string, isCrew bool) bool { + if userDID == "" { + // Anonymous writes not allowed + return false + } + + // Check if DID is the hold owner + if userDID == captain.Owner { + // Owner always has write access + return true + } + + // Check if DID is a crew member + return isCrew +} + +// ErrHoldNotFound is returned when a hold's captain record cannot be found +var ErrHoldNotFound = fmt.Errorf("hold not found") + +// ErrUnauthorized is returned when access is denied +var ErrUnauthorized = fmt.Errorf("unauthorized") diff --git a/pkg/auth/hold_local.go b/pkg/auth/hold_local.go new file mode 100644 index 0000000..d0d636c --- /dev/null +++ b/pkg/auth/hold_local.go @@ -0,0 +1,101 @@ +package auth + +import ( + "context" + "fmt" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/hold/pds" +) + +// LocalHoldAuthorizer queries the hold's own embedded PDS directly +// Used by hold service to authorize access to its own storage +type LocalHoldAuthorizer struct { + pds *pds.HoldPDS +} + +// NewLocalHoldAuthorizer creates a new local authorizer for hold service +func NewLocalHoldAuthorizer(holdPDS *pds.HoldPDS) HoldAuthorizer { + return &LocalHoldAuthorizer{ + pds: holdPDS, + } +} + +// NewLocalHoldAuthorizerFromInterface creates a new local authorizer from an any +// This is used to avoid import cycles - caller must pass a *pds.HoldPDS +func NewLocalHoldAuthorizerFromInterface(holdPDS any) HoldAuthorizer { + // Type assert to *pds.HoldPDS + if pdsTyped, ok := holdPDS.(*pds.HoldPDS); ok { + return &LocalHoldAuthorizer{ + pds: pdsTyped, + } + } + // Return nil if type assertion fails - caller should check + return nil +} + +// GetCaptainRecord retrieves the captain record from the hold's PDS +func (a *LocalHoldAuthorizer) GetCaptainRecord(ctx context.Context, holdDID string) (*atproto.CaptainRecord, error) { + // Verify that the requested holdDID matches this hold + if holdDID != a.pds.DID() { + return nil, fmt.Errorf("holdDID mismatch: requested %s, this hold is %s", holdDID, a.pds.DID()) + } + + // Query the PDS for captain record + _, pdsCaptain, err := a.pds.GetCaptainRecord(ctx) + if err != nil { + return nil, fmt.Errorf("failed to get captain record: %w", err) + } + + // The PDS returns *atproto.CaptainRecord directly now (after we update pds to use atproto types) + return pdsCaptain, nil +} + +// IsCrewMember checks if userDID is a crew member +func (a *LocalHoldAuthorizer) IsCrewMember(ctx context.Context, holdDID, userDID string) (bool, error) { + // Verify that the requested holdDID matches this hold + if holdDID != a.pds.DID() { + return false, fmt.Errorf("holdDID mismatch: requested %s, this hold is %s", holdDID, a.pds.DID()) + } + + // Query the PDS for crew list + crewList, err := a.pds.ListCrewMembers(ctx) + if err != nil { + return false, fmt.Errorf("failed to list crew members: %w", err) + } + + // Check if userDID is in the crew list + for _, member := range crewList { + if member.Record.Member == userDID { + // TODO: Check expiration if set + return true, nil + } + } + + return false, nil +} + +// CheckReadAccess implements read authorization using shared logic +func (a *LocalHoldAuthorizer) CheckReadAccess(ctx context.Context, holdDID, userDID string) (bool, error) { + captain, err := a.GetCaptainRecord(ctx, holdDID) + if err != nil { + return false, err + } + + return CheckReadAccessWithCaptain(captain, userDID), nil +} + +// CheckWriteAccess implements write authorization using shared logic +func (a *LocalHoldAuthorizer) CheckWriteAccess(ctx context.Context, holdDID, userDID string) (bool, error) { + captain, err := a.GetCaptainRecord(ctx, holdDID) + if err != nil { + return false, err + } + + isCrew, err := a.IsCrewMember(ctx, holdDID, userDID) + if err != nil { + return false, err + } + + return CheckWriteAccessWithCaptain(captain, userDID, isCrew), nil +} diff --git a/pkg/auth/hold_remote.go b/pkg/auth/hold_remote.go new file mode 100644 index 0000000..a0fc114 --- /dev/null +++ b/pkg/auth/hold_remote.go @@ -0,0 +1,559 @@ +package auth + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + "atcr.io/pkg/atproto" +) + +// RemoteHoldAuthorizer queries a hold's PDS via XRPC endpoints +// Used by AppView to authorize access to remote holds +// Implements caching for captain records to reduce XRPC calls +type RemoteHoldAuthorizer struct { + db *sql.DB + httpClient *http.Client + cacheTTL time.Duration // TTL for captain record cache + recentDenials sync.Map // In-memory cache for first denials (10s backoff) + stopCleanup chan struct{} // Signal to stop cleanup goroutine +} + +// denialEntry stores timestamp for in-memory first denials +type denialEntry struct { + timestamp time.Time +} + +// NewRemoteHoldAuthorizer creates a new remote authorizer for AppView +func NewRemoteHoldAuthorizer(db *sql.DB) HoldAuthorizer { + a := &RemoteHoldAuthorizer{ + db: db, + httpClient: &http.Client{ + Timeout: 10 * time.Second, + }, + cacheTTL: 1 * time.Hour, // 1 hour cache TTL + stopCleanup: make(chan struct{}), + } + + // Start cleanup goroutine for in-memory denials + go a.cleanupRecentDenials() + + return a +} + +// cleanupRecentDenials runs every 10s to remove expired first-denial entries +func (a *RemoteHoldAuthorizer) cleanupRecentDenials() { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + now := time.Now() + a.recentDenials.Range(func(key, value any) bool { + entry := value.(denialEntry) + // Remove entries older than 15 seconds (10s backoff + 5s grace) + if now.Sub(entry.timestamp) > 15*time.Second { + a.recentDenials.Delete(key) + } + return true + }) + case <-a.stopCleanup: + return + } + } +} + +// GetCaptainRecord retrieves a captain record with caching +// 1. Check database cache +// 2. If cache miss or expired, query hold's XRPC endpoint +// 3. Update cache +func (a *RemoteHoldAuthorizer) GetCaptainRecord(ctx context.Context, holdDID string) (*atproto.CaptainRecord, error) { + // Try cache first + if a.db != nil { + cached, err := a.getCachedCaptainRecord(holdDID) + if err == nil && cached != nil { + // Cache hit - check if still valid + if time.Since(cached.UpdatedAt) < a.cacheTTL { + return cached.CaptainRecord, nil + } + // Cache expired - continue to fetch fresh data + } + } + + // Cache miss or expired - query XRPC endpoint + record, err := a.fetchCaptainRecordFromXRPC(ctx, holdDID) + if err != nil { + return nil, err + } + + // Update cache + if a.db != nil { + if err := a.setCachedCaptainRecord(holdDID, record); err != nil { + // Log error but don't fail - caching is best-effort + fmt.Printf("WARNING: Failed to cache captain record: %v\n", err) + } + } + + return record, nil +} + +// captainRecordWithMeta includes UpdatedAt for cache management +type captainRecordWithMeta struct { + *atproto.CaptainRecord + UpdatedAt time.Time +} + +// getCachedCaptainRecord retrieves a captain record from database cache +func (a *RemoteHoldAuthorizer) getCachedCaptainRecord(holdDID string) (*captainRecordWithMeta, error) { + query := ` + SELECT owner_did, public, allow_all_crew, deployed_at, region, provider, updated_at + FROM hold_captain_records + WHERE hold_did = ? + ` + + var record atproto.CaptainRecord + var deployedAt, region, provider sql.NullString + var updatedAt time.Time + + err := a.db.QueryRow(query, holdDID).Scan( + &record.Owner, + &record.Public, + &record.AllowAllCrew, + &deployedAt, + ®ion, + &provider, + &updatedAt, + ) + + if err == sql.ErrNoRows { + return nil, nil // Cache miss + } + + if err != nil { + return nil, fmt.Errorf("cache query failed: %w", err) + } + + // Handle nullable fields + if deployedAt.Valid { + record.DeployedAt = deployedAt.String + } + if region.Valid { + record.Region = region.String + } + if provider.Valid { + record.Provider = provider.String + } + + return &captainRecordWithMeta{ + CaptainRecord: &record, + UpdatedAt: updatedAt, + }, nil +} + +// setCachedCaptainRecord stores a captain record in database cache +func (a *RemoteHoldAuthorizer) setCachedCaptainRecord(holdDID string, record *atproto.CaptainRecord) error { + query := ` + INSERT INTO hold_captain_records ( + hold_did, owner_did, public, allow_all_crew, + deployed_at, region, provider, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(hold_did) DO UPDATE SET + owner_did = excluded.owner_did, + public = excluded.public, + allow_all_crew = excluded.allow_all_crew, + deployed_at = excluded.deployed_at, + region = excluded.region, + provider = excluded.provider, + updated_at = excluded.updated_at + ` + + _, err := a.db.Exec(query, + holdDID, + record.Owner, + record.Public, + record.AllowAllCrew, + nullString(record.DeployedAt), + nullString(record.Region), + nullString(record.Provider), + time.Now(), + ) + + return err +} + +// fetchCaptainRecordFromXRPC queries the hold's XRPC endpoint for captain record +func (a *RemoteHoldAuthorizer) fetchCaptainRecordFromXRPC(ctx context.Context, holdDID string) (*atproto.CaptainRecord, error) { + // Resolve DID to URL + holdURL, err := resolveDIDToURL(holdDID) + if err != nil { + return nil, fmt.Errorf("failed to resolve hold DID: %w", err) + } + + // Build XRPC request URL + // GET /xrpc/com.atproto.repo.getRecord?repo={did}&collection=io.atcr.hold.captain&rkey=self + xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", + holdURL, url.QueryEscape(holdDID), url.QueryEscape(atproto.CaptainCollection)) + + req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) + if err != nil { + return nil, err + } + + resp, err := a.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("XRPC request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body)) + } + + // Parse response + var xrpcResp struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value struct { + Type string `json:"$type"` + Owner string `json:"owner"` + Public bool `json:"public"` + AllowAllCrew bool `json:"allowAllCrew"` + DeployedAt string `json:"deployedAt"` + Region string `json:"region,omitempty"` + Provider string `json:"provider,omitempty"` + } `json:"value"` + } + + if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil { + return nil, fmt.Errorf("failed to decode XRPC response: %w", err) + } + + // Convert to our type + record := &atproto.CaptainRecord{ + Type: atproto.CaptainCollection, + Owner: xrpcResp.Value.Owner, + Public: xrpcResp.Value.Public, + AllowAllCrew: xrpcResp.Value.AllowAllCrew, + DeployedAt: xrpcResp.Value.DeployedAt, + Region: xrpcResp.Value.Region, + Provider: xrpcResp.Value.Provider, + } + + return record, nil +} + +// IsCrewMember checks if userDID is a crew member with caching +// 1. Check approval cache (15min TTL) +// 2. Check denial cache with exponential backoff +// 3. If cache miss, query XRPC endpoint and update cache +func (a *RemoteHoldAuthorizer) IsCrewMember(ctx context.Context, holdDID, userDID string) (bool, error) { + // Skip caching if no database + if a.db == nil { + return a.isCrewMemberNoCache(ctx, holdDID, userDID) + } + + // Check approval cache first (15min TTL) + if approved, err := a.getCachedApproval(holdDID, userDID); err == nil && approved { + return true, nil + } + + // Check denial cache with backoff + if blocked, err := a.isBlockedByDenialBackoff(holdDID, userDID); err == nil && blocked { + // Still in backoff period - don't query again + return false, nil + } + + // Cache miss or expired - query XRPC endpoint + isCrew, err := a.isCrewMemberNoCache(ctx, holdDID, userDID) + if err != nil { + return false, err + } + + // Update cache based on result + if isCrew { + // Cache approval for 15 minutes + _ = a.cacheApproval(holdDID, userDID, 15*time.Minute) + } else { + // Cache denial with exponential backoff + _ = a.cacheDenial(holdDID, userDID) + } + + return isCrew, nil +} + +// isCrewMemberNoCache queries XRPC without caching (internal helper) +func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID, userDID string) (bool, error) { + // Resolve DID to URL + holdURL, err := resolveDIDToURL(holdDID) + if err != nil { + return false, fmt.Errorf("failed to resolve hold DID: %w", err) + } + + // Build XRPC request URL + // GET /xrpc/com.atproto.repo.listRecords?repo={did}&collection=io.atcr.hold.crew + xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s", + holdURL, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection)) + + req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) + if err != nil { + return false, err + } + + resp, err := a.httpClient.Do(req) + if err != nil { + return false, fmt.Errorf("XRPC request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return false, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body)) + } + + // Parse response + var xrpcResp struct { + Records []struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value struct { + Type string `json:"$type"` + Member string `json:"member"` + Role string `json:"role"` + Permissions []string `json:"permissions"` + AddedAt string `json:"addedAt"` + } `json:"value"` + } `json:"records"` + } + + if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil { + return false, fmt.Errorf("failed to decode XRPC response: %w", err) + } + + // Check if userDID is in the crew list + for _, record := range xrpcResp.Records { + if record.Value.Member == userDID { + // TODO: Check expiration if set + return true, nil + } + } + + return false, nil +} + +// CheckReadAccess implements read authorization using shared logic +func (a *RemoteHoldAuthorizer) CheckReadAccess(ctx context.Context, holdDID, userDID string) (bool, error) { + captain, err := a.GetCaptainRecord(ctx, holdDID) + if err != nil { + return false, err + } + + return CheckReadAccessWithCaptain(captain, userDID), nil +} + +// CheckWriteAccess implements write authorization using shared logic +func (a *RemoteHoldAuthorizer) CheckWriteAccess(ctx context.Context, holdDID, userDID string) (bool, error) { + captain, err := a.GetCaptainRecord(ctx, holdDID) + if err != nil { + return false, err + } + + isCrew, err := a.IsCrewMember(ctx, holdDID, userDID) + if err != nil { + return false, err + } + + return CheckWriteAccessWithCaptain(captain, userDID, isCrew), nil +} + +// resolveDIDToURL converts a did:web DID to an HTTPS URL +// Example: did:web:hold01.atcr.io → https://hold01.atcr.io +func resolveDIDToURL(did string) (string, error) { + // Handle did:web format + if !strings.HasPrefix(did, "did:web:") { + return "", fmt.Errorf("only did:web is supported, got: %s", did) + } + + // Extract hostname from did:web:hostname + hostname := strings.TrimPrefix(did, "did:web:") + + // Convert to HTTPS URL + return "https://" + hostname, nil +} + +// nullString converts a string to sql.NullString +func nullString(s string) sql.NullString { + if s == "" { + return sql.NullString{Valid: false} + } + return sql.NullString{String: s, Valid: true} +} + +// getCachedApproval checks if user has a cached crew approval +func (a *RemoteHoldAuthorizer) getCachedApproval(holdDID, userDID string) (bool, error) { + query := ` + SELECT expires_at + FROM hold_crew_approvals + WHERE hold_did = ? AND user_did = ? + ` + + var expiresAt time.Time + err := a.db.QueryRow(query, holdDID, userDID).Scan(&expiresAt) + + if err == sql.ErrNoRows { + return false, nil // Cache miss + } + + if err != nil { + return false, err + } + + // Check if approval has expired + if time.Now().After(expiresAt) { + // Expired - clean up + _ = a.deleteCachedApproval(holdDID, userDID) + return false, nil + } + + return true, nil +} + +// cacheApproval stores a crew approval with TTL +func (a *RemoteHoldAuthorizer) cacheApproval(holdDID, userDID string, ttl time.Duration) error { + query := ` + INSERT INTO hold_crew_approvals (hold_did, user_did, approved_at, expires_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(hold_did, user_did) DO UPDATE SET + approved_at = excluded.approved_at, + expires_at = excluded.expires_at + ` + + now := time.Now() + expiresAt := now.Add(ttl) + + _, err := a.db.Exec(query, holdDID, userDID, now, expiresAt) + return err +} + +// deleteCachedApproval removes an expired approval +func (a *RemoteHoldAuthorizer) deleteCachedApproval(holdDID, userDID string) error { + query := `DELETE FROM hold_crew_approvals WHERE hold_did = ? AND user_did = ?` + _, err := a.db.Exec(query, holdDID, userDID) + return err +} + +// isBlockedByDenialBackoff checks if user is in denial backoff period +// Checks in-memory cache first (for 10s first denials), then DB (for longer backoffs) +func (a *RemoteHoldAuthorizer) isBlockedByDenialBackoff(holdDID, userDID string) (bool, error) { + // Check in-memory cache first (first denials with 10s backoff) + key := fmt.Sprintf("%s:%s", holdDID, userDID) + if val, ok := a.recentDenials.Load(key); ok { + entry := val.(denialEntry) + // Check if still within 10s backoff + if time.Since(entry.timestamp) < 10*time.Second { + return true, nil // Still blocked by in-memory first denial + } + } + + // Check database for longer backoffs (second+ denials) + query := ` + SELECT next_retry_at + FROM hold_crew_denials + WHERE hold_did = ? AND user_did = ? + ` + + var nextRetryAt time.Time + err := a.db.QueryRow(query, holdDID, userDID).Scan(&nextRetryAt) + + if err == sql.ErrNoRows { + return false, nil // No denial record + } + + if err != nil { + return false, err + } + + // Check if still in backoff period + if time.Now().Before(nextRetryAt) { + return true, nil // Still blocked + } + + // Backoff period expired - can retry + return false, nil +} + +// cacheDenial stores or updates a denial with exponential backoff +// First denial: in-memory only (10s backoff) +// Second+ denial: database with exponential backoff (1m, 5m, 15m, 1h) +func (a *RemoteHoldAuthorizer) cacheDenial(holdDID, userDID string) error { + key := fmt.Sprintf("%s:%s", holdDID, userDID) + + // Check if this is a first denial (not in memory, not in DB) + _, inMemory := a.recentDenials.Load(key) + + var denialCount int + query := `SELECT denial_count FROM hold_crew_denials WHERE hold_did = ? AND user_did = ?` + err := a.db.QueryRow(query, holdDID, userDID).Scan(&denialCount) + + inDB := err != sql.ErrNoRows + if err != nil && err != sql.ErrNoRows { + return err + } + + // If not in memory and not in DB, this is the first denial + if !inMemory && !inDB { + // First denial: store only in memory with 10s backoff + a.recentDenials.Store(key, denialEntry{timestamp: time.Now()}) + return nil + } + + // Second+ denial: persist to database with exponential backoff + denialCount++ + backoff := getBackoffDuration(denialCount) + now := time.Now() + nextRetry := now.Add(backoff) + + // Upsert denial record + upsertQuery := ` + INSERT INTO hold_crew_denials (hold_did, user_did, denial_count, next_retry_at, last_denied_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(hold_did, user_did) DO UPDATE SET + denial_count = excluded.denial_count, + next_retry_at = excluded.next_retry_at, + last_denied_at = excluded.last_denied_at + ` + + _, err = a.db.Exec(upsertQuery, holdDID, userDID, denialCount, nextRetry, now) + + // Remove from in-memory cache since we're now tracking in DB + a.recentDenials.Delete(key) + + return err +} + +// getBackoffDuration returns the backoff duration based on denial count +// Note: First denial (10s) is in-memory only and not tracked by this function +// This function handles second+ denials: 1m, 5m, 15m, 1h +func getBackoffDuration(denialCount int) time.Duration { + backoffs := []time.Duration{ + 1 * time.Minute, // 1st DB denial (2nd overall) - being added soon + 5 * time.Minute, // 2nd DB denial (3rd overall) - probably not happening + 15 * time.Minute, // 3rd DB denial (4th overall) - definitely not soon + 60 * time.Minute, // 4th+ DB denial (5th+ overall) - stop hammering + } + + idx := denialCount - 1 + if idx >= len(backoffs) { + idx = len(backoffs) - 1 + } + + return backoffs[idx] +} diff --git a/pkg/auth/oauth/server.go b/pkg/auth/oauth/server.go index 5d60454..046fc72 100644 --- a/pkg/auth/oauth/server.go +++ b/pkg/auth/oauth/server.go @@ -27,11 +27,11 @@ type UserStore interface { // Server handles OAuth authorization for the AppView type Server struct { - app *App - refresher *Refresher - uiSessionStore UISessionStore - db *sql.DB - defaultHoldEndpoint string + app *App + refresher *Refresher + uiSessionStore UISessionStore + db *sql.DB + defaultHoldDID string // Default hold DID (e.g., "did:web:hold01.atcr.io") } // NewServer creates a new OAuth server @@ -41,9 +41,11 @@ func NewServer(app *App) *Server { } } -// SetDefaultHoldEndpoint sets the default hold endpoint for profile creation -func (s *Server) SetDefaultHoldEndpoint(endpoint string) { - s.defaultHoldEndpoint = endpoint +// SetDefaultHoldDID sets the default hold DID for profile creation +// Expected format: "did:web:hold01.atcr.io" +// To find a hold's DID, visit: https://hold-url/.well-known/did.json +func (s *Server) SetDefaultHoldDID(did string) { + s.defaultHoldDID = did } // SetRefresher sets the refresher for invalidating session cache @@ -271,8 +273,8 @@ func (s *Server) fetchAndStoreAvatar(ctx context.Context, did, sessionID, handle client := atproto.NewClientWithIndigoClient(pdsEndpoint, did, session.APIClient()) // Ensure sailor profile exists (creates with default hold if configured, or empty profile if not) - fmt.Printf("DEBUG [oauth/server]: Ensuring profile exists for %s (defaultHold=%s)\n", did, s.defaultHoldEndpoint) - if err := atproto.EnsureProfile(ctx, client, s.defaultHoldEndpoint); err != nil { + fmt.Printf("DEBUG [oauth/server]: Ensuring profile exists for %s (defaultHold=%s)\n", did, s.defaultHoldDID) + if err := atproto.EnsureProfile(ctx, client, s.defaultHoldDID); err != nil { fmt.Printf("WARNING [oauth/server]: Failed to ensure profile for %s: %v\n", did, err) // Continue anyway - profile creation is not critical for avatar fetch } else { diff --git a/pkg/auth/token/handler.go b/pkg/auth/token/handler.go index 44a910a..4b88cc3 100644 --- a/pkg/auth/token/handler.go +++ b/pkg/auth/token/handler.go @@ -18,19 +18,21 @@ import ( // Handler handles /auth/token requests type Handler struct { - issuer *Issuer - validator *atproto.SessionValidator - deviceStore *db.DeviceStore // For validating device secrets - defaultHoldEndpoint string + issuer *Issuer + validator *atproto.SessionValidator + deviceStore *db.DeviceStore // For validating device secrets + defaultHoldDID string } // NewHandler creates a new token handler -func NewHandler(issuer *Issuer, deviceStore *db.DeviceStore, defaultHoldEndpoint string) *Handler { +// defaultHoldDID should be in format "did:web:hold01.atcr.io" +// To find a hold's DID, visit: https://hold-url/.well-known/did.json +func NewHandler(issuer *Issuer, deviceStore *db.DeviceStore, defaultHoldDID string) *Handler { return &Handler{ - issuer: issuer, - validator: atproto.NewSessionValidator(), - deviceStore: deviceStore, - defaultHoldEndpoint: defaultHoldEndpoint, + issuer: issuer, + validator: atproto.NewSessionValidator(), + deviceStore: deviceStore, + defaultHoldDID: defaultHoldDID, } } @@ -157,7 +159,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { atprotoClient := mainAtproto.NewClient(pdsEndpoint, did, accessToken) // Ensure profile exists (will create with default hold if not exists and default is configured) - if err := mainAtproto.EnsureProfile(r.Context(), atprotoClient, h.defaultHoldEndpoint); err != nil { + if err := mainAtproto.EnsureProfile(r.Context(), atprotoClient, h.defaultHoldDID); err != nil { // Log error but don't fail auth - profile management is not critical fmt.Printf("WARNING: failed to ensure profile for %s: %v\n", did, err) } diff --git a/pkg/hold/authorization.go b/pkg/hold/authorization.go deleted file mode 100644 index bd26029..0000000 --- a/pkg/hold/authorization.go +++ /dev/null @@ -1,181 +0,0 @@ -package hold - -import ( - "context" - "encoding/json" - "fmt" - "log" - "time" - - "atcr.io/pkg/atproto" - "github.com/bluesky-social/indigo/atproto/identity" - "github.com/bluesky-social/indigo/atproto/syntax" -) - -// isAuthorizedRead checks if a DID can read from this hold -// Authorization: -// - Public hold: allow anonymous (empty DID) or any authenticated user -// - Private hold: require authentication (any user with sailor.profile) -func (s *HoldService) isAuthorizedRead(did string) bool { - // Check hold public flag - isPublic, err := s.isHoldPublic() - if err != nil { - log.Printf("ERROR: Failed to check hold public flag: %v", err) - // Fail secure - deny access on error - return false - } - - if isPublic { - // Public hold - allow anyone (even anonymous) - return true - } - - // Private hold - require authentication - // Any authenticated user with sailor.profile can read - if did == "" { - // Anonymous user trying to access private hold - return false - } - - // For MVP: assume DID presence means they have sailor.profile - // Future: could query PDS to verify sailor.profile exists - return true -} - -// isAuthorizedWrite checks if a DID can write to this hold -// Authorization: must be hold owner OR crew member -func (s *HoldService) isAuthorizedWrite(did string) bool { - if did == "" { - // Anonymous writes not allowed - return false - } - - // Check if DID is the hold owner - ownerDID := s.config.Registration.OwnerDID - if ownerDID == "" { - log.Printf("ERROR: Hold owner DID not configured") - return false - } - - if did == ownerDID { - // Owner always has write access - return true - } - - // Check if DID is a crew member - isCrew, err := s.isCrewMember(did) - if err != nil { - log.Printf("ERROR: Failed to check crew membership: %v", err) - return false - } - - return isCrew -} - -// isHoldPublic checks if this hold allows public (anonymous) reads -func (s *HoldService) isHoldPublic() (bool, error) { - // Use cached config value for now - // Future: could query PDS for hold record to get live value - return s.config.Server.Public, nil -} - -// isCrewMember checks if a DID is a crew member of this hold -// Supports both explicit DID matching and pattern-based matching (wildcards, handle globs) -func (s *HoldService) isCrewMember(did string) (bool, error) { - ownerDID := s.config.Registration.OwnerDID - if ownerDID == "" { - return false, fmt.Errorf("hold owner DID not configured") - } - - ctx := context.Background() - - // Resolve owner's PDS endpoint using indigo - directory := identity.DefaultDirectory() - ownerDIDParsed, err := syntax.ParseDID(ownerDID) - if err != nil { - return false, fmt.Errorf("invalid owner DID: %w", err) - } - - ident, err := directory.LookupDID(ctx, ownerDIDParsed) - if err != nil { - return false, fmt.Errorf("failed to resolve owner PDS: %w", err) - } - - pdsEndpoint := ident.PDSEndpoint() - if pdsEndpoint == "" { - return false, fmt.Errorf("no PDS endpoint found for owner") - } - - // Build this hold's URI for filtering - publicURL := s.config.Server.PublicURL - if publicURL == "" { - return false, fmt.Errorf("hold public URL not configured") - } - holdName, err := extractHostname(publicURL) - if err != nil { - return false, fmt.Errorf("failed to extract hold name: %w", err) - } - holdURI := fmt.Sprintf("at://%s/%s/%s", ownerDID, atproto.HoldCollection, holdName) - - // Create unauthenticated client to read public records - client := atproto.NewClient(pdsEndpoint, ownerDID, "") - - // List crew records for this hold - // Crew records are public, so we can read them without auth - records, err := client.ListRecords(ctx, atproto.HoldCrewCollection, 100) - if err != nil { - return false, fmt.Errorf("failed to list crew records: %w", err) - } - - // Resolve handle once for pattern matching (lazily, only if needed) - var handle string - var handleResolved bool - - // Check crew records for both explicit DID and pattern matches - for _, record := range records { - var crewRecord atproto.HoldCrewRecord - if err := json.Unmarshal(record.Value, &crewRecord); err != nil { - continue - } - - // Only check crew records for THIS hold (prevents cross-hold access) - if crewRecord.Hold != holdURI { - continue - } - - // Check expiration (if set) - if crewRecord.ExpiresAt != nil && time.Now().After(*crewRecord.ExpiresAt) { - continue // Skip expired membership - } - - // Check explicit DID match - if crewRecord.Member != nil && *crewRecord.Member == did { - // Found explicit crew membership - return true, nil - } - - // Check pattern match (if pattern is set) - if crewRecord.MemberPattern != nil && *crewRecord.MemberPattern != "" { - // Lazy handle resolution - only resolve if we encounter a pattern - if !handleResolved { - handle, err = resolveHandle(did) - if err != nil { - log.Printf("Warning: failed to resolve handle for DID %s: %v", did, err) - // Continue checking explicit DIDs even if handle resolution fails - handleResolved = true // Mark as attempted (don't retry) - handle = "" // Empty handle won't match patterns - } else { - handleResolved = true - } - } - - // If we have a handle, check pattern match - if handle != "" && matchPattern(*crewRecord.MemberPattern, handle) { - // Found pattern-based crew membership - return true, nil - } - } - } - - return false, nil -} diff --git a/pkg/hold/pds/blobstore_adapter.go b/pkg/hold/blobstore_adapter.go similarity index 69% rename from pkg/hold/pds/blobstore_adapter.go rename to pkg/hold/blobstore_adapter.go index 9ef261a..af7c7dd 100644 --- a/pkg/hold/pds/blobstore_adapter.go +++ b/pkg/hold/blobstore_adapter.go @@ -1,19 +1,19 @@ -package pds +package hold import ( "context" - "atcr.io/pkg/hold" + "atcr.io/pkg/hold/pds" ) -// HoldServiceBlobStore adapts the hold service to implement the BlobStore interface +// HoldServiceBlobStore adapts the hold service to implement the pds.BlobStore interface type HoldServiceBlobStore struct { - service *hold.HoldService + service *HoldService holdDID string } // NewHoldServiceBlobStore creates a blob store adapter for the hold service -func NewHoldServiceBlobStore(service *hold.HoldService, holdDID string) *HoldServiceBlobStore { +func NewHoldServiceBlobStore(service *HoldService, holdDID string) pds.BlobStore { return &HoldServiceBlobStore{ service: service, holdDID: holdDID, @@ -23,9 +23,8 @@ func NewHoldServiceBlobStore(service *hold.HoldService, holdDID string) *HoldSer // GetPresignedDownloadURL returns a presigned URL for downloading a blob func (b *HoldServiceBlobStore) GetPresignedDownloadURL(digest string) (string, error) { // Use the hold service's existing presigned URL logic - // We need to expose a wrapper method on HoldService ctx := context.Background() - url, err := b.service.GetPresignedURL(ctx, hold.OperationGet, digest, b.holdDID) + url, err := b.service.GetPresignedURL(ctx, OperationGet, digest, b.holdDID) if err != nil { return "", err } @@ -36,7 +35,7 @@ func (b *HoldServiceBlobStore) GetPresignedDownloadURL(digest string) (string, e func (b *HoldServiceBlobStore) GetPresignedUploadURL(digest string) (string, error) { // Use the hold service's existing presigned URL logic ctx := context.Background() - url, err := b.service.GetPresignedURL(ctx, hold.OperationPut, digest, b.holdDID) + url, err := b.service.GetPresignedURL(ctx, OperationPut, digest, b.holdDID) if err != nil { return "", err } diff --git a/pkg/hold/pds/captain.go b/pkg/hold/pds/captain.go index 9fcb78f..f6adcad 100644 --- a/pkg/hold/pds/captain.go +++ b/pkg/hold/pds/captain.go @@ -6,6 +6,7 @@ import ( "fmt" "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/repo" "github.com/ipfs/go-cid" ) @@ -17,8 +18,8 @@ const ( // CreateCaptainRecord creates the captain record for the hold func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, public bool, allowAllCrew bool) (cid.Cid, error) { - captainRecord := &CaptainRecord{ - Type: CaptainCollection, + captainRecord := &atproto.CaptainRecord{ + Type: atproto.CaptainCollection, Owner: ownerDID, Public: public, AllowAllCrew: allowAllCrew, @@ -26,7 +27,7 @@ func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, publ } // Create record in repo with fixed rkey "self" - recordCID, rkey, err := p.repo.CreateRecord(ctx, CaptainCollection, captainRecord) + recordCID, rkey, err := p.repo.CreateRecord(ctx, atproto.CaptainCollection, captainRecord) if err != nil { return cid.Undef, fmt.Errorf("failed to create captain record: %w", err) } @@ -48,9 +49,8 @@ func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, publ return cid.Undef, fmt.Errorf("failed to persist commit: %w", err) } - // Create a new session for the next operation - rootStr := root.String() - newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rootStr) + // Create a new session for the next operation (use revision string, not CID) + newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rev) if err != nil { return cid.Undef, fmt.Errorf("failed to create new session: %w", err) } @@ -71,8 +71,8 @@ func (p *HoldPDS) CreateCaptainRecord(ctx context.Context, ownerDID string, publ } // GetCaptainRecord retrieves the captain record -func (p *HoldPDS) GetCaptainRecord(ctx context.Context) (cid.Cid, *CaptainRecord, error) { - path := fmt.Sprintf("%s/%s", CaptainCollection, CaptainRkey) +func (p *HoldPDS) GetCaptainRecord(ctx context.Context) (cid.Cid, *atproto.CaptainRecord, error) { + path := fmt.Sprintf("%s/%s", atproto.CaptainCollection, CaptainRkey) // Get the record bytes and decode manually recordCID, recBytes, err := p.repo.GetRecordBytes(ctx, path) @@ -81,7 +81,7 @@ func (p *HoldPDS) GetCaptainRecord(ctx context.Context) (cid.Cid, *CaptainRecord } // Decode the CBOR bytes into our CaptainRecord type - var captainRecord CaptainRecord + var captainRecord atproto.CaptainRecord if err := captainRecord.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil { return cid.Undef, nil, fmt.Errorf("failed to decode captain record: %w", err) } @@ -102,7 +102,7 @@ func (p *HoldPDS) UpdateCaptainRecord(ctx context.Context, public bool, allowAll existing.AllowAllCrew = allowAllCrew // Update record in repo - path := fmt.Sprintf("%s/%s", CaptainCollection, CaptainRkey) + path := fmt.Sprintf("%s/%s", atproto.CaptainCollection, CaptainRkey) recordCID, err := p.repo.UpdateRecord(ctx, path, existing) if err != nil { return cid.Undef, fmt.Errorf("failed to update captain record: %w", err) @@ -125,9 +125,8 @@ func (p *HoldPDS) UpdateCaptainRecord(ctx context.Context, public bool, allowAll return cid.Undef, fmt.Errorf("failed to persist commit: %w", err) } - // Create a new session for the next operation - rootStr := root.String() - newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rootStr) + // Create a new session for the next operation (use revision string, not CID) + newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rev) if err != nil { return cid.Undef, fmt.Errorf("failed to create new session: %w", err) } diff --git a/pkg/hold/pds/crew.go b/pkg/hold/pds/crew.go index 1bcbfdd..c6c0937 100644 --- a/pkg/hold/pds/crew.go +++ b/pkg/hold/pds/crew.go @@ -7,14 +7,15 @@ import ( "strings" "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/repo" "github.com/ipfs/go-cid" ) // AddCrewMember adds a new crew member to the hold and commits to carstore func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, permissions []string) (cid.Cid, error) { - crewRecord := &CrewRecord{ - Type: CrewCollection, + crewRecord := &atproto.CrewRecord{ + Type: atproto.CrewCollection, Member: memberDID, Role: role, Permissions: permissions, @@ -22,7 +23,7 @@ func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, per } // Create record in repo (using memberDID as rkey for easy lookup) - recordCID, _, err := p.repo.CreateRecord(ctx, CrewCollection, crewRecord) + recordCID, _, err := p.repo.CreateRecord(ctx, atproto.CrewCollection, crewRecord) if err != nil { return cid.Undef, fmt.Errorf("failed to create crew record: %w", err) } @@ -44,9 +45,8 @@ func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, per return cid.Undef, fmt.Errorf("failed to persist commit: %w", err) } - // Create a new session for the next operation (old session is now closed) - rootStr := root.String() - newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rootStr) + // Create a new session for the next operation (use revision string, not CID) + newSession, err := p.carstore.NewDeltaSession(ctx, p.uid, &rev) if err != nil { return cid.Undef, fmt.Errorf("failed to create new session: %w", err) } @@ -65,8 +65,8 @@ func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, per } // GetCrewMember retrieves a crew member by their record key -func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *CrewRecord, error) { - path := fmt.Sprintf("%s/%s", CrewCollection, rkey) +func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *atproto.CrewRecord, error) { + path := fmt.Sprintf("%s/%s", atproto.CrewCollection, rkey) // Get the record bytes and decode manually (indigo doesn't know our custom type) recordCID, recBytes, err := p.repo.GetRecordBytes(ctx, path) @@ -75,7 +75,7 @@ func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *Cre } // Decode the CBOR bytes into our CrewRecord type - var crewRecord CrewRecord + var crewRecord atproto.CrewRecord if err := crewRecord.UnmarshalCBOR(bytes.NewReader(*recBytes)); err != nil { return cid.Undef, nil, fmt.Errorf("failed to decode crew record: %w", err) } @@ -87,14 +87,14 @@ func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *Cre type CrewMemberWithKey struct { Rkey string Cid cid.Cid - Record *CrewRecord + Record *atproto.CrewRecord } // ListCrewMembers returns all crew members with their rkeys func (p *HoldPDS) ListCrewMembers(ctx context.Context) ([]*CrewMemberWithKey, error) { var crew []*CrewMemberWithKey - err := p.repo.ForEach(ctx, CrewCollection, func(k string, v cid.Cid) error { + err := p.repo.ForEach(ctx, atproto.CrewCollection, func(k string, v cid.Cid) error { // Extract rkey from full path (k is like "io.atcr.hold.crew/3m37dr2ddit22") parts := strings.Split(k, "/") rkey := parts[len(parts)-1] @@ -127,7 +127,7 @@ func (p *HoldPDS) ListCrewMembers(ctx context.Context) ([]*CrewMemberWithKey, er // RemoveCrewMember removes a crew member func (p *HoldPDS) RemoveCrewMember(ctx context.Context, rkey string) error { - path := fmt.Sprintf("%s/%s", CrewCollection, rkey) + path := fmt.Sprintf("%s/%s", atproto.CrewCollection, rkey) err := p.repo.DeleteRecord(ctx, path) if err != nil { diff --git a/pkg/hold/pds/did.go b/pkg/hold/pds/did.go index f0b688e..b882871 100644 --- a/pkg/hold/pds/did.go +++ b/pkg/hold/pds/did.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "net/url" - "strings" ) // DIDDocument represents a did:web document @@ -35,13 +34,22 @@ type Service struct { // GenerateDIDDocument creates a DID document for a did:web identity func (p *HoldPDS) GenerateDIDDocument(publicURL string) (*DIDDocument, error) { - // Extract hostname from public URL - hostname := strings.TrimPrefix(publicURL, "http://") - hostname = strings.TrimPrefix(hostname, "https://") - hostname = strings.Split(hostname, "/")[0] // Remove any path - hostname = strings.Split(hostname, ":")[0] // Remove port for DID + // Parse URL to extract host and port + u, err := url.Parse(publicURL) + if err != nil { + return nil, fmt.Errorf("failed to parse public URL: %w", err) + } - did := fmt.Sprintf("did:web:%s", hostname) + hostname := u.Hostname() + port := u.Port() + + // Build host string (include non-standard ports per did:web spec) + host := hostname + if port != "" && port != "80" && port != "443" { + host = fmt.Sprintf("%s:%s", hostname, port) + } + + did := fmt.Sprintf("did:web:%s", host) // Get public key in multibase format using indigo's crypto pubKey, err := p.signingKey.PublicKey() @@ -58,7 +66,7 @@ func (p *HoldPDS) GenerateDIDDocument(publicURL string) (*DIDDocument, error) { }, ID: did, AlsoKnownAs: []string{ - fmt.Sprintf("at://%s", hostname), + fmt.Sprintf("at://%s", host), }, VerificationMethod: []VerificationMethod{ { @@ -99,20 +107,29 @@ func (p *HoldPDS) MarshalDIDDocument() ([]byte, error) { } // GenerateDIDFromURL creates a did:web identifier from a public URL -// Example: "http://hold1.example.com:8080" -> "did:web:hold1.example.com" +// Example: "http://hold1.example.com:8080" -> "did:web:hold1.example.com:8080" +// Note: Per did:web spec, non-standard ports (not 80/443) are included in the DID func GenerateDIDFromURL(publicURL string) string { // Parse URL u, err := url.Parse(publicURL) if err != nil { // Fallback: assume it's just a hostname - return fmt.Sprintf("did:web:%s", strings.Split(publicURL, ":")[0]) + return fmt.Sprintf("did:web:%s", publicURL) } - // Use hostname without port for DID + // Get hostname hostname := u.Hostname() if hostname == "" { hostname = "localhost" } + // Get port + port := u.Port() + + // Include port in DID if it's non-standard (not 80 for http, not 443 for https) + if port != "" && port != "80" && port != "443" { + return fmt.Sprintf("did:web:%s:%s", hostname, port) + } + return fmt.Sprintf("did:web:%s", hostname) } diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 7bfdf65..c9f85d0 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -5,7 +5,9 @@ import ( "fmt" "os" "path/filepath" + "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/carstore" "github.com/bluesky-social/indigo/models" @@ -59,17 +61,22 @@ func NewHoldPDS(ctx context.Context, did, publicURL, dbPath, keyPath string) (*H var session *carstore.DeltaSession var r *repo.Repo - // Create a session connected to this user's data in carstore - session, err = cs.NewDeltaSession(ctx, uid, nil) - if err != nil { - return nil, fmt.Errorf("failed to create delta session: %w", err) - } - if !hasValidRepo { - // No valid repo - create new empty repo + // No valid repo - create new session with nil (new repo) + session, err = cs.NewDeltaSession(ctx, uid, nil) + if err != nil { + return nil, fmt.Errorf("failed to create delta session: %w", err) + } + // Create new empty repo r = repo.NewRepo(ctx, did, session) } else { - // Repo exists with valid head - load from existing head + // Repo exists with valid head - create session pointing to current head + headStr := head.String() + session, err = cs.NewDeltaSession(ctx, uid, &headStr) + if err != nil { + return nil, fmt.Errorf("failed to create delta session: %w", err) + } + // Load from existing head r, err = repo.OpenRepo(ctx, session, head) if err != nil { return nil, fmt.Errorf("failed to open existing repo: %w", err) @@ -104,17 +111,27 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string, public bool, a return nil } - // Check if repo already has commits - head, err := p.carstore.GetUserRepoHead(ctx, p.uid) - if err != nil || !head.Defined() { - // No repo exists yet, bootstrap - fmt.Printf("🚀 Bootstrapping hold PDS with owner: %s\n", ownerDID) - } else { - // Repo exists and is valid - fmt.Printf("⏭ī¸ Skipping PDS bootstrap: repo already initialized (head: %s)\n", head.String()[:16]) + // Check if captain record already exists (idempotent bootstrap) + _, _, err := p.GetCaptainRecord(ctx) + if err == nil { + // Captain record exists, we're good + fmt.Printf("✅ Captain record exists, skipping bootstrap\n") return nil } + // No captain record - check if this is a new repo or existing repo + head, err := p.carstore.GetUserRepoHead(ctx, p.uid) + isNewRepo := (err != nil || !head.Defined()) + + if isNewRepo { + fmt.Printf("🚀 Bootstrapping new hold PDS with owner: %s\n", ownerDID) + // For new repo, create records inline to avoid session issues + return p.bootstrapNewRepo(ctx, ownerDID, public, allowAllCrew) + } + + // Existing repo - use normal record creation flow + fmt.Printf("ℹī¸ Repo already initialized (head: %s), creating captain record...\n", head.String()[:16]) + // Create captain record (hold ownership and settings) _, err = p.CreateCaptainRecord(ctx, ownerDID, public, allowAllCrew) if err != nil { @@ -133,6 +150,61 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string, public bool, a return nil } +// bootstrapNewRepo handles bootstrapping a brand new repo (avoids session juggling issues) +func (p *HoldPDS) bootstrapNewRepo(ctx context.Context, ownerDID string, public bool, allowAllCrew bool) error { + // Create captain and crew records in a single commit + captainRecord := &atproto.CaptainRecord{ + Type: atproto.CaptainCollection, + Owner: ownerDID, + Public: public, + AllowAllCrew: allowAllCrew, + DeployedAt: time.Now().Format(time.RFC3339), + } + + crewRecord := &atproto.CrewRecord{ + Type: atproto.CrewCollection, + Member: ownerDID, + Role: "admin", + Permissions: []string{"blob:read", "blob:write", "crew:admin"}, + AddedAt: time.Now().Format(time.RFC3339), + } + + // Create both records in the repo + _, _, err := p.repo.CreateRecord(ctx, atproto.CaptainCollection, captainRecord) + if err != nil { + return fmt.Errorf("failed to create captain record: %w", err) + } + + _, _, err = p.repo.CreateRecord(ctx, atproto.CrewCollection, crewRecord) + if err != nil { + return fmt.Errorf("failed to create crew record: %w", err) + } + + // Commit everything in one go + signer := func(ctx context.Context, did string, data []byte) ([]byte, error) { + return p.signingKey.HashAndSign(data) + } + + root, rev, err := p.repo.Commit(ctx, signer) + if err != nil { + return fmt.Errorf("failed to commit bootstrap records: %w", err) + } + + // Close the session with the new root + _, err = p.session.CloseWithRoot(ctx, root, rev) + if err != nil { + return fmt.Errorf("failed to persist bootstrap commit: %w", err) + } + + fmt.Printf("✅ Created captain record (public=%v, allowAllCrew=%v)\n", public, allowAllCrew) + fmt.Printf("✅ Added %s as hold admin\n", ownerDID) + + // DON'T create a new session here - let subsequent operations handle that + // The PDS is now bootstrapped and will be reloaded properly on next restart + + return nil +} + // Close closes the session and carstore func (p *HoldPDS) Close() error { // TODO: Close session properly diff --git a/pkg/hold/pds/types.go b/pkg/hold/pds/types.go deleted file mode 100644 index 8735696..0000000 --- a/pkg/hold/pds/types.go +++ /dev/null @@ -1,32 +0,0 @@ -package pds - -//go:generate go run github.com/whyrusleeping/cbor-gen --map-encoding CrewRecord CaptainRecord - -// ATProto record types for the hold service - -// CaptainRecord represents the hold's ownership and metadata -// Collection: io.atcr.hold.captain (single record per hold) -type CaptainRecord struct { - Type string `json:"$type" cborgen:"$type"` - Owner string `json:"owner" cborgen:"owner"` // DID of hold owner - Public bool `json:"public" cborgen:"public"` // Public read access - AllowAllCrew bool `json:"allowAllCrew" cborgen:"allowAllCrew"` // Allow any authenticated user to register as crew - DeployedAt string `json:"deployedAt" cborgen:"deployedAt"` // RFC3339 timestamp - Region string `json:"region,omitempty" cborgen:"region,omitempty"` // S3 region (optional) - Provider string `json:"provider,omitempty" cborgen:"provider,omitempty"` // Deployment provider (optional) -} - -// CrewRecord represents a crew member in the hold -// Collection: io.atcr.hold.crew (one record per member) -type CrewRecord struct { - Type string `json:"$type" cborgen:"$type"` - Member string `json:"member" cborgen:"member"` - Role string `json:"role" cborgen:"role"` - Permissions []string `json:"permissions" cborgen:"permissions"` - AddedAt string `json:"addedAt" cborgen:"addedAt"` // RFC3339 timestamp -} - -const ( - CaptainCollection = "io.atcr.hold.captain" - CrewCollection = "io.atcr.hold.crew" -) diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index a1367ff..9c4b955 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -7,6 +7,7 @@ import ( "net/http" "strings" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/repo" "github.com/bluesky-social/indigo/util" "github.com/ipfs/go-cid" @@ -151,7 +152,7 @@ func (h *XRPCHandler) HandleDescribeRepo(w http.ResponseWriter, r *http.Request) "did": h.pds.DID(), "handle": h.pds.DID(), "didDoc": didDoc, - "collections": []string{CrewCollection}, + "collections": []string{atproto.CrewCollection}, "handleIsCorrect": true, } @@ -181,7 +182,7 @@ func (h *XRPCHandler) HandleGetRecord(w http.ResponseWriter, r *http.Request) { } // Only support crew collection for now - if collection != CrewCollection { + if collection != atproto.CrewCollection { http.Error(w, "collection not found", http.StatusNotFound) return } @@ -223,7 +224,7 @@ func (h *XRPCHandler) HandleListRecords(w http.ResponseWriter, r *http.Request) } // Only support crew collection for now - if collection != CrewCollection { + if collection != atproto.CrewCollection { http.Error(w, "collection not found", http.StatusNotFound) return } @@ -273,7 +274,7 @@ func (h *XRPCHandler) HandleSyncGetRecord(w http.ResponseWriter, r *http.Request } // Only support crew collection for now - if collection != CrewCollection { + if collection != atproto.CrewCollection { http.Error(w, "collection not found", http.StatusNotFound) return } @@ -551,7 +552,7 @@ func (h *XRPCHandler) HandleRequestCrew(w http.ResponseWriter, r *http.Request) if member.Record.Member == user.DID { // Already a crew member, return success with existing record response := map[string]any{ - "uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), CrewCollection, member.Rkey), + "uri": fmt.Sprintf("at://%s/%s/%s", h.pds.DID(), atproto.CrewCollection, member.Rkey), "cid": member.Cid.String(), "status": "already_member", "message": "User is already a crew member", diff --git a/pkg/hold/service.go b/pkg/hold/service.go index a9df63c..4ac1fe9 100644 --- a/pkg/hold/service.go +++ b/pkg/hold/service.go @@ -7,23 +7,33 @@ import ( "net/http" "net/url" + "atcr.io/pkg/auth" "github.com/aws/aws-sdk-go/service/s3" storagedriver "github.com/distribution/distribution/v3/registry/storage/driver" "github.com/distribution/distribution/v3/registry/storage/driver/factory" ) +// HoldPDSInterface is the minimal interface needed from the embedded PDS +// This avoids a circular import between pkg/hold and pkg/hold/pds +type HoldPDSInterface interface { + DID() string +} + // HoldService provides presigned URLs for blob storage in a hold type HoldService struct { driver storagedriver.StorageDriver config *Config - s3Client *s3.S3 // S3 client for presigned URLs (nil if not S3 storage) - bucket string // S3 bucket name - s3PathPrefix string // S3 path prefix (if any) - MultipartMgr *MultipartManager // Exported for access in route handlers + s3Client *s3.S3 // S3 client for presigned URLs (nil if not S3 storage) + bucket string // S3 bucket name + s3PathPrefix string // S3 path prefix (if any) + MultipartMgr *MultipartManager // Exported for access in route handlers + pds HoldPDSInterface // Embedded PDS for captain/crew records + authorizer auth.HoldAuthorizer // Authorizer for access control } // NewHoldService creates a new hold service -func NewHoldService(cfg *Config) (*HoldService, error) { +// holdPDS must be a *pds.HoldPDS but we use any to avoid import cycle +func NewHoldService(cfg *Config, holdPDS any) (*HoldService, error) { // Create storage driver from config ctx := context.Background() driver, err := factory.Create(ctx, cfg.Storage.Type(), cfg.Storage.Parameters()) @@ -31,10 +41,22 @@ func NewHoldService(cfg *Config) (*HoldService, error) { return nil, fmt.Errorf("failed to create storage driver: %w", err) } + // Create local authorizer using the embedded PDS + // This requires casting holdPDS to the concrete type expected by auth + authorizer := auth.NewLocalHoldAuthorizerFromInterface(holdPDS) + + // Cast to our interface for storage + pdsInterface, ok := holdPDS.(HoldPDSInterface) + if !ok { + return nil, fmt.Errorf("holdPDS must implement HoldPDSInterface") + } + service := &HoldService{ driver: driver, config: cfg, MultipartMgr: NewMultipartManager(), + pds: pdsInterface, + authorizer: authorizer, } // Initialize S3 client for presigned URLs (if using S3 storage) @@ -50,6 +72,30 @@ func (s *HoldService) GetPresignedURL(ctx context.Context, operation PresignedUR return s.getPresignedURL(ctx, operation, digest, did) } +// isAuthorizedRead checks if the given DID has read access to this hold +// This is a helper wrapper around the authorizer for internal use +func (s *HoldService) isAuthorizedRead(did string) bool { + ctx := context.Background() + allowed, err := s.authorizer.CheckReadAccess(ctx, s.pds.DID(), did) + if err != nil { + log.Printf("Authorization check failed: %v", err) + return false + } + return allowed +} + +// isAuthorizedWrite checks if the given DID has write access to this hold +// This is a helper wrapper around the authorizer for internal use +func (s *HoldService) isAuthorizedWrite(did string) bool { + ctx := context.Background() + allowed, err := s.authorizer.CheckWriteAccess(ctx, s.pds.DID(), did) + if err != nil { + log.Printf("Authorization check failed: %v", err) + return false + } + return allowed +} + // HealthHandler handles health check requests func (s *HoldService) HealthHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json")