From 2a60a47fd571f5bd84c55375896103c27ee77860 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 23 Dec 2025 16:20:52 -0600 Subject: [PATCH] fix issues pulling other users images. fix labels taking priority over annotations. fix various auth errors --- CLAUDE.md | 37 +- docs/VALKEY_MIGRATION.md | 399 ++++++++++++++++++ pkg/appview/middleware/registry.go | 231 +++++----- pkg/appview/storage/context.go | 20 +- pkg/appview/storage/manifest_store.go | 16 +- pkg/appview/storage/routing_repository.go | 4 +- .../storage/routing_repository_test.go | 38 +- pkg/auth/token/claims.go | 19 + pkg/hold/pds/auth.go | 97 +++-- pkg/hold/pds/auth_test.go | 110 +++++ 10 files changed, 806 insertions(+), 165 deletions(-) create mode 100644 docs/VALKEY_MIGRATION.md diff --git a/CLAUDE.md b/CLAUDE.md index 4b2333d..016ef28 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -475,12 +475,47 @@ Lightweight standalone service for BYOS (Bring Your Own Storage) with embedded P Read access: - **Public hold** (`HOLD_PUBLIC=true`): Anonymous + all authenticated users -- **Private hold** (`HOLD_PUBLIC=false`): Requires authentication + crew membership with blob:read permission +- **Private hold** (`HOLD_PUBLIC=false`): Requires authentication + crew membership with blob:read OR blob:write permission +- **Note:** `blob:write` implicitly grants `blob:read` access (can't push without pulling) Write access: - Hold owner OR crew members with blob:write permission - Verified via `io.atcr.hold.crew` records in hold's embedded PDS +**Permission Matrix:** + +| User Type | Public Read | Private Read | Write | Crew Admin | +|-----------|-------------|--------------|-------|------------| +| Anonymous | Yes | No | No | No | +| Owner (captain) | Yes | Yes | Yes | Yes (implied) | +| Crew (blob:read only) | Yes | Yes | No | No | +| Crew (blob:write only) | Yes | Yes* | Yes | No | +| Crew (blob:read + blob:write) | Yes | Yes | Yes | No | +| Crew (crew:admin) | Yes | Yes | Yes | Yes | +| Authenticated non-crew | Yes | No | No | No | + +*`blob:write` implicitly grants `blob:read` access + +**Authorization Error Format:** + +All authorization failures use consistent structured errors (`pkg/hold/pds/auth.go`): +``` +access denied for [action]: [reason] (required: [permission(s)]) +``` + +Examples: +- `access denied for blob:read: user is not a crew member (required: blob:read or blob:write)` +- `access denied for blob:write: crew member lacks permission (required: blob:write)` +- `access denied for crew:admin: user is not a crew member (required: crew:admin)` + +**Shared Error Constants** (`pkg/hold/pds/auth.go`): +- `ErrMissingAuthHeader` - Missing Authorization header +- `ErrInvalidAuthFormat` - Invalid Authorization header format +- `ErrInvalidAuthScheme` - Invalid scheme (expected Bearer or DPoP) +- `ErrInvalidJWTFormat` - Malformed JWT +- `ErrMissingISSClaim` / `ErrMissingSubClaim` - Missing JWT claims +- `ErrTokenExpired` - Token has expired + **Embedded PDS Endpoints** (`pkg/hold/pds/xrpc.go`): Standard ATProto sync endpoints: diff --git a/docs/VALKEY_MIGRATION.md b/docs/VALKEY_MIGRATION.md new file mode 100644 index 0000000..b6679a6 --- /dev/null +++ b/docs/VALKEY_MIGRATION.md @@ -0,0 +1,399 @@ +# Analysis: AppView SQL Database Usage + +## Overview + +The AppView uses SQLite with 19 tables. The key finding: **most data is a cache of ATProto records** that could theoretically be rebuilt from users' PDS instances. + +## Data Categories + +### 1. MUST PERSIST (Local State Only) + +These tables contain data that **cannot be reconstructed** from external sources: + +| Table | Purpose | Why It Must Persist | +|-------|---------|---------------------| +| `oauth_sessions` | OAuth tokens | Refresh tokens are stateful; losing them = users must re-auth | +| `ui_sessions` | Web browser sessions | Session continuity for logged-in users | +| `devices` | Approved devices + bcrypt secrets | User authorization decisions; secrets are one-way hashed | +| `pending_device_auth` | In-flight auth flows | Short-lived (10min) but critical during auth | +| `oauth_auth_requests` | OAuth flow state | Short-lived but required for auth completion | +| `repository_stats` | Pull/push counts | **Locally tracked metrics** - not stored in ATProto | + +### 2. CACHED FROM PDS (Rebuildable) + +These tables are essentially a **read-through cache** of ATProto data: + +| Table | Source | ATProto Collection | +|-------|--------|-------------------| +| `users` | User's PDS profile | `app.bsky.actor.profile` + DID document | +| `manifests` | User's PDS | `io.atcr.manifest` records | +| `tags` | User's PDS | `io.atcr.tag` records | +| `layers` | Derived from manifests | Parsed from manifest content | +| `manifest_references` | Derived from manifest lists | Parsed from multi-arch manifests | +| `repository_annotations` | Manifest config blob | OCI annotations from config | +| `repo_pages` | User's PDS | `io.atcr.repo.page` records | +| `stars` | User's PDS | `io.atcr.sailor.star` records (synced via Jetstream) | +| `hold_captain_records` | Hold's embedded PDS | `io.atcr.hold.captain` records | +| `hold_crew_approvals` | Hold's embedded PDS | `io.atcr.hold.crew` records | +| `hold_crew_denials` | Local authorization cache | Could re-check on demand | + +### 3. OPERATIONAL + +| Table | Purpose | +|-------|---------| +| `schema_migrations` | Migration tracking | +| `firehose_cursor` | Jetstream position (can restart from 0) | + +## Key Insights + +### What's Actually Unique to AppView? + +1. **Authentication state** - OAuth sessions, devices, UI sessions +2. **Engagement metrics** - Pull/push counts (locally tracked, not in ATProto) + +### What Could Be Eliminated? + +If ATCR fully embraced the ATProto model: + +1. **`users`** - Query PDS on demand (with caching) +2. **`manifests`, `tags`, `layers`** - Query PDS on demand (with caching) +3. **`repository_annotations`** - Fetch manifest config on demand +4. **`repo_pages`** - Query PDS on demand +5. **`hold_*` tables** - Query hold's PDS on demand + +### Trade-offs + +**Current approach (heavy caching):** +- Fast queries for UI (search, browse, stats) +- Offline resilience (PDS down doesn't break UI) +- Complex sync logic (Jetstream consumer, backfill) +- State can diverge from source of truth + +**Lighter approach (query on demand):** +- Always fresh data +- Simpler codebase (no sync) +- Slower queries (network round-trips) +- Depends on PDS availability + +## Current Limitation: No Cache-Miss Queries + +**Finding:** There's no "query PDS on cache miss" logic. Users/manifests only enter the DB via: +1. OAuth login (user authenticates) +2. Jetstream events (firehose activity) + +**Problem:** If someone visits `atcr.io/alice/myapp` before alice is indexed → 404 + +**Where this happens:** +- `pkg/appview/handlers/repository.go:50-53`: If `db.GetUserByDID()` returns nil → 404 +- No fallback to `atproto.Client.ListRecords()` or similar + +**This matters for Valkey migration:** If cache is ephemeral and restarts clear it, you need cache-miss logic to repopulate on demand. Otherwise: +- Restart Valkey → all users/manifests gone +- Wait for Jetstream to re-index OR implement cache-miss queries + +**Cache-miss implementation design:** + +Existing code to reuse: `pkg/appview/jetstream/processor.go:43-97` (`EnsureUser`) + +```go +// New: pkg/appview/cache/loader.go + +type Loader struct { + cache Cache // Valkey interface + client *atproto.Client +} + +// GetUser with cache-miss fallback +func (l *Loader) GetUser(ctx context.Context, did string) (*User, error) { + // 1. Try cache + if user := l.cache.GetUser(did); user != nil { + return user, nil + } + + // 2. Cache miss - resolve identity (already queries network) + _, handle, pdsEndpoint, err := atproto.ResolveIdentity(ctx, did) + if err != nil { + return nil, err // User doesn't exist in network + } + + // 3. Fetch profile for avatar + client := atproto.NewClient(pdsEndpoint, "", "") + profile, _ := client.GetProfileRecord(ctx, did) + avatarURL := "" + if profile != nil && profile.Avatar != nil { + avatarURL = atproto.BlobCDNURL(did, profile.Avatar.Ref.Link) + } + + // 4. Cache and return + user := &User{DID: did, Handle: handle, PDSEndpoint: pdsEndpoint, Avatar: avatarURL} + l.cache.SetUser(user, 1*time.Hour) + return user, nil +} + +// GetManifestsForRepo with cache-miss fallback +func (l *Loader) GetManifestsForRepo(ctx context.Context, did, repo string) ([]Manifest, error) { + cacheKey := fmt.Sprintf("manifests:%s:%s", did, repo) + + // 1. Try cache + if cached := l.cache.Get(cacheKey); cached != nil { + return cached.([]Manifest), nil + } + + // 2. Cache miss - get user's PDS endpoint + user, err := l.GetUser(ctx, did) + if err != nil { + return nil, err + } + + // 3. Query PDS for manifests + client := atproto.NewClient(user.PDSEndpoint, "", "") + records, _, err := client.ListRecordsForRepo(ctx, did, atproto.ManifestCollection, 100, "") + if err != nil { + return nil, err + } + + // 4. Filter by repository and parse + var manifests []Manifest + for _, rec := range records { + var m atproto.ManifestRecord + if err := json.Unmarshal(rec.Value, &m); err != nil { + continue + } + if m.Repository == repo { + manifests = append(manifests, convertManifest(m)) + } + } + + // 5. Cache and return + l.cache.Set(cacheKey, manifests, 10*time.Minute) + return manifests, nil +} +``` + +**Handler changes:** +```go +// Before (repository.go:45-53): +owner, err := db.GetUserByDID(h.DB, did) +if owner == nil { + RenderNotFound(w, r, h.Templates, h.RegistryURL) + return +} + +// After: +owner, err := h.Loader.GetUser(r.Context(), did) +if err != nil { + RenderNotFound(w, r, h.Templates, h.RegistryURL) + return +} +``` + +**Performance considerations:** +- Cache hit: ~1ms (Valkey lookup) +- Cache miss: ~200-500ms (PDS round-trip) +- First request after restart: slower but correct +- Jetstream still useful for proactive warming + +--- + +## Proposed Architecture: Valkey + ATProto + +### Goal +Replace SQLite with Valkey (Redis-compatible) for ephemeral state, push remaining persistent data to ATProto. + +### What goes to Valkey (ephemeral, TTL-based) + +| Current Table | Valkey Key Pattern | TTL | Notes | +|---------------|-------------------|-----|-------| +| `oauth_sessions` | `oauth:{did}:{session_id}` | 90 days | Lost on restart = re-auth | +| `ui_sessions` | `ui:{session_id}` | Session duration | Lost on restart = re-login | +| `oauth_auth_requests` | `authreq:{state}` | 10 min | In-flight flows | +| `pending_device_auth` | `pending:{device_code}` | 10 min | In-flight flows | +| `firehose_cursor` | `cursor:jetstream` | None | Can restart from 0 | +| All PDS cache tables | `cache:{collection}:{did}:{rkey}` | 10-60 min | Query PDS on miss | + +**Benefits:** +- Multi-instance ready (shared Valkey) +- No schema migrations +- Natural TTL expiry +- Simpler code (no SQL) + +### What could become ATProto records + +| Current Table | Proposed Collection | Where Stored | Open Questions | +|---------------|---------------------|--------------|----------------| +| `devices` | `io.atcr.sailor.device` | User's PDS | Privacy: IP, user-agent sensitive? | +| `repository_stats` | `io.atcr.repo.stats` | Hold's PDS or User's PDS | Who owns the stats? | + +**Devices → Valkey:** +- Move current device table to Valkey +- Key: `device:{did}:{device_id}` → `{name, secret_hash, ip, user_agent, created_at, last_used}` +- TTL: Long (1 year?) or no expiry +- Device list: `devices:{did}` → Set of device IDs +- Secret validation works the same, just different backend + +**Service auth exploration (future):** +The challenge with pure ATProto service auth is the AppView still needs the user's OAuth session to write manifests to their PDS. The current flow: +1. User authenticates via OAuth → AppView gets OAuth tokens +2. AppView issues registry JWT to credential helper +3. Credential helper presents JWT on each push/pull +4. AppView uses OAuth session to write to user's PDS + +Service auth could work for the hold side (AppView → Hold), but not for the user's OAuth session. + +**Repository stats → Hold's PDS:** + +**Challenge discovered:** The hold's `getBlob` endpoint only receives `did` + `cid`, not the repository name. + +Current flow (`proxy_blob_store.go:358-362`): +```go +xrpcURL := fmt.Sprintf("%s%s?did=%s&cid=%s&method=%s", + p.holdURL, atproto.SyncGetBlob, p.ctx.DID, dgst.String(), operation) +``` + +**Implementation options:** + +**Option A: Add repository parameter to getBlob (recommended)** +```go +// Modified AppView call: +xrpcURL := fmt.Sprintf("%s%s?did=%s&cid=%s&method=%s&repo=%s", + p.holdURL, atproto.SyncGetBlob, p.ctx.DID, dgst.String(), operation, p.ctx.Repository) +``` + +```go +// Modified hold handler (xrpc.go:969): +func (h *XRPCHandler) HandleGetBlob(w http.ResponseWriter, r *http.Request) { + did := r.URL.Query().Get("did") + cidOrDigest := r.URL.Query().Get("cid") + repo := r.URL.Query().Get("repo") // NEW + + // ... existing blob handling ... + + // Increment stats if repo provided + if repo != "" { + go h.pds.IncrementPullCount(did, repo) // Async, non-blocking + } +} +``` + +**Stats record structure:** +``` +Collection: io.atcr.hold.stats +Rkey: base64(did:repository) // Deterministic, unique + +{ + "$type": "io.atcr.hold.stats", + "did": "did:plc:alice123", + "repository": "myapp", + "pullCount": 1542, + "pushCount": 47, + "lastPull": "2025-01-15T...", + "lastPush": "2025-01-10T...", + "createdAt": "2025-01-01T..." +} +``` + +**Hold-side implementation:** +```go +// New file: pkg/hold/pds/stats.go + +func (p *HoldPDS) IncrementPullCount(ctx context.Context, did, repo string) error { + rkey := statsRecordKey(did, repo) + + // Get or create stats record + stats, err := p.GetStatsRecord(ctx, rkey) + if err != nil || stats == nil { + stats = &atproto.StatsRecord{ + Type: atproto.StatsCollection, + DID: did, + Repository: repo, + PullCount: 0, + PushCount: 0, + CreatedAt: time.Now(), + } + } + + // Increment and update + stats.PullCount++ + stats.LastPull = time.Now() + + _, err = p.repomgr.UpdateRecord(ctx, p.uid, atproto.StatsCollection, rkey, stats) + return err +} +``` + +**Query endpoint (new XRPC):** +``` +GET /xrpc/io.atcr.hold.getStats?did={userDID}&repo={repository} +→ Returns JSON: { pullCount, pushCount, lastPull, lastPush } + +GET /xrpc/io.atcr.hold.listStats?did={userDID} +→ Returns all stats for a user across all repos on this hold +``` + +**AppView aggregation:** +```go +func (l *Loader) GetAggregatedStats(ctx context.Context, did, repo string) (*Stats, error) { + // 1. Get all holds that have served this repo + holdDIDs, _ := l.cache.GetHoldDIDsForRepo(did, repo) + + // 2. Query each hold for stats + var total Stats + for _, holdDID := range holdDIDs { + holdURL := resolveHoldDID(holdDID) + stats, _ := queryHoldStats(ctx, holdURL, did, repo) + total.PullCount += stats.PullCount + total.PushCount += stats.PushCount + } + + return &total, nil +} +``` + +**Files to modify:** +- `pkg/atproto/lexicon.go` - Add `StatsCollection` + `StatsRecord` +- `pkg/hold/pds/stats.go` - New file for stats operations +- `pkg/hold/pds/xrpc.go` - Add `repo` param to getBlob, add stats endpoints +- `pkg/appview/storage/proxy_blob_store.go` - Pass repository to getBlob +- `pkg/appview/cache/loader.go` - Aggregation logic + +### Migration Path + +**Phase 1: Add Valkey infrastructure** +- Add Valkey client to AppView +- Create store interfaces that abstract SQLite vs Valkey +- Dual-write OAuth sessions to both + +**Phase 2: Migrate sessions to Valkey** +- OAuth sessions, UI sessions, auth requests, pending device auth +- Remove SQLite session tables +- Test: restart AppView, users get logged out (acceptable) + +**Phase 3: Migrate devices to Valkey** +- Move device store to Valkey +- Same data structure, different backend +- Consider device expiry policy + +**Phase 4: Implement hold-side stats** +- Add `io.atcr.hold.stats` collection to hold's embedded PDS +- Hold increments stats on blob access +- Add XRPC endpoint: `io.atcr.hold.getStats` + +**Phase 5: AppView stats aggregation** +- Track holdDids per repo in Valkey cache +- Query holds for stats, aggregate +- Cache aggregated stats with TTL + +**Phase 6: Remove SQLite (optional)** +- Keep SQLite as optional cache layer for UI queries +- Or: Query PDS on demand with Valkey caching +- Jetstream still useful for real-time updates + +## Summary Table + +| Category | Tables | % of Schema | Truly Persistent? | +|----------|--------|-------------|-------------------| +| Auth & Sessions + Metrics | 6 | 32% | Yes | +| PDS Cache | 11 | 58% | No (rebuildable) | +| Operational | 2 | 10% | No | + +**~58% of the database is cached ATProto data that could be rebuilt from PDSes.** diff --git a/pkg/appview/middleware/registry.go b/pkg/appview/middleware/registry.go index c12957b..8ddb08b 100644 --- a/pkg/appview/middleware/registry.go +++ b/pkg/appview/middleware/registry.go @@ -29,6 +29,9 @@ const holdDIDKey contextKey = "hold.did" // authMethodKey is the context key for storing auth method from JWT const authMethodKey contextKey = "auth.method" +// pullerDIDKey is the context key for storing the authenticated user's DID from JWT +const pullerDIDKey contextKey = "puller.did" + // validationCacheEntry stores a validated service token with expiration type validationCacheEntry struct { serviceToken string @@ -302,83 +305,97 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Get service token for hold authentication (only if authenticated) // Use validation cache to prevent concurrent requests from racing on OAuth/DPoP // Route based on auth method from JWT token + // IMPORTANT: Use PULLER's DID/PDS for service token, not owner's! + // The puller (authenticated user) needs to authenticate to the hold service. var serviceToken string authMethod, _ := ctx.Value(authMethodKey).(string) + pullerDID, _ := ctx.Value(pullerDIDKey).(string) + var pullerPDSEndpoint string // Only fetch service token if user is authenticated // Unauthenticated requests (like /v2/ ping) should not trigger token fetching - if authMethod != "" { - // Create cache key: "did:holdDID" - cacheKey := fmt.Sprintf("%s:%s", did, holdDID) + if authMethod != "" && pullerDID != "" { + // Resolve puller's PDS endpoint for service token request + _, _, pullerPDSEndpoint, err = atproto.ResolveIdentity(ctx, pullerDID) + if err != nil { + slog.Warn("Failed to resolve puller's PDS, falling back to anonymous access", + "component", "registry/middleware", + "pullerDID", pullerDID, + "error", err) + // Continue without service token - hold will decide if anonymous access is allowed + } else { + // Create cache key: "pullerDID:holdDID" + cacheKey := fmt.Sprintf("%s:%s", pullerDID, holdDID) - // Fetch service token through validation cache - // This ensures only ONE request per DID:holdDID pair fetches the token - // Concurrent requests will wait for the first request to complete - var fetchErr error - serviceToken, fetchErr = nr.validationCache.getOrFetch(ctx, cacheKey, func() (string, error) { - if authMethod == token.AuthMethodAppPassword { - // App-password flow: use Bearer token authentication - slog.Debug("Using app-password flow for service token", - "component", "registry/middleware", - "did", did, - "cacheKey", cacheKey) - - token, err := token.GetOrFetchServiceTokenWithAppPassword(ctx, did, holdDID, pdsEndpoint) - if err != nil { - slog.Error("Failed to get service token with app-password", + // Fetch service token through validation cache + // This ensures only ONE request per pullerDID:holdDID pair fetches the token + // Concurrent requests will wait for the first request to complete + var fetchErr error + serviceToken, fetchErr = nr.validationCache.getOrFetch(ctx, cacheKey, func() (string, error) { + if authMethod == token.AuthMethodAppPassword { + // App-password flow: use Bearer token authentication + slog.Debug("Using app-password flow for service token", "component", "registry/middleware", - "did", did, - "holdDID", holdDID, - "pdsEndpoint", pdsEndpoint, - "error", err) - return "", err - } - return token, nil - } else if nr.refresher != nil { - // OAuth flow: use DPoP authentication - slog.Debug("Using OAuth flow for service token", - "component", "registry/middleware", - "did", did, - "cacheKey", cacheKey) + "pullerDID", pullerDID, + "cacheKey", cacheKey) - token, err := token.GetOrFetchServiceToken(ctx, nr.refresher, did, holdDID, pdsEndpoint) - if err != nil { - slog.Error("Failed to get service token with OAuth", + token, err := token.GetOrFetchServiceTokenWithAppPassword(ctx, pullerDID, holdDID, pullerPDSEndpoint) + if err != nil { + slog.Error("Failed to get service token with app-password", + "component", "registry/middleware", + "pullerDID", pullerDID, + "holdDID", holdDID, + "pullerPDSEndpoint", pullerPDSEndpoint, + "error", err) + return "", err + } + return token, nil + } else if nr.refresher != nil { + // OAuth flow: use DPoP authentication + slog.Debug("Using OAuth flow for service token", "component", "registry/middleware", - "did", did, - "holdDID", holdDID, - "pdsEndpoint", pdsEndpoint, - "error", err) - return "", err + "pullerDID", pullerDID, + "cacheKey", cacheKey) + + token, err := token.GetOrFetchServiceToken(ctx, nr.refresher, pullerDID, holdDID, pullerPDSEndpoint) + if err != nil { + slog.Error("Failed to get service token with OAuth", + "component", "registry/middleware", + "pullerDID", pullerDID, + "holdDID", holdDID, + "pullerPDSEndpoint", pullerPDSEndpoint, + "error", err) + return "", err + } + return token, nil } - return token, nil - } - return "", fmt.Errorf("no authentication method available") - }) + return "", fmt.Errorf("no authentication method available") + }) - // Handle errors from cached fetch - if fetchErr != nil { - errMsg := fetchErr.Error() + // Handle errors from cached fetch + if fetchErr != nil { + errMsg := fetchErr.Error() - // Check for app-password specific errors - if authMethod == token.AuthMethodAppPassword { - if strings.Contains(errMsg, "expired or invalid") || strings.Contains(errMsg, "no app-password") { - return nil, nr.authErrorMessage("App-password authentication failed. Please re-authenticate with: docker login") + // Check for app-password specific errors + if authMethod == token.AuthMethodAppPassword { + if strings.Contains(errMsg, "expired or invalid") || strings.Contains(errMsg, "no app-password") { + return nil, nr.authErrorMessage("App-password authentication failed. Please re-authenticate with: docker login") + } } - } - // Check for OAuth specific errors - if strings.Contains(errMsg, "OAuth session") || strings.Contains(errMsg, "OAuth validation") { - return nil, nr.authErrorMessage("OAuth session expired or invalidated by PDS. Your session has been cleared") - } + // Check for OAuth specific errors + if strings.Contains(errMsg, "OAuth session") || strings.Contains(errMsg, "OAuth validation") { + return nil, nr.authErrorMessage("OAuth session expired or invalidated by PDS. Your session has been cleared") + } - // Generic service token error - return nil, nr.authErrorMessage(fmt.Sprintf("Failed to obtain storage credentials: %v", fetchErr)) + // Generic service token error + return nil, nr.authErrorMessage(fmt.Sprintf("Failed to obtain storage credentials: %v", fetchErr)) + } } } else { slog.Debug("Skipping service token fetch for unauthenticated request", "component", "registry/middleware", - "did", did) + "ownerDID", did) } // Create a new reference with identity/image format @@ -396,37 +413,24 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name return nil, err } - // Get access token for PDS operations - // Use auth method from JWT to determine client type: - // - OAuth users: use session provider (DPoP-enabled) - // - App-password users: use Basic Auth token cache + // Create ATProto client for manifest/tag operations + // Pulls: ATProto records are public, no auth needed + // Pushes: Need auth, but puller must be owner anyway var atprotoClient *atproto.Client - if authMethod == token.AuthMethodOAuth && nr.refresher != nil { - // OAuth flow: use session provider for locked OAuth sessions - // This prevents DPoP nonce race conditions during concurrent layer uploads - slog.Debug("Creating ATProto client with OAuth session provider", - "component", "registry/middleware", - "did", did, - "authMethod", authMethod) - atprotoClient = atproto.NewClientWithSessionProvider(pdsEndpoint, did, nr.refresher) - } else { - // App-password flow (or fallback): use Basic Auth token cache - accessToken, ok := auth.GetGlobalTokenCache().Get(did) - if !ok { - slog.Debug("No cached access token found for app-password auth", - "component", "registry/middleware", - "did", did, - "authMethod", authMethod) - accessToken = "" // Will fail on manifest push, but let it try + if pullerDID == did { + // Puller is owner - may need auth for pushes + if authMethod == token.AuthMethodOAuth && nr.refresher != nil { + atprotoClient = atproto.NewClientWithSessionProvider(pdsEndpoint, did, nr.refresher) + } else if authMethod == token.AuthMethodAppPassword { + accessToken, _ := auth.GetGlobalTokenCache().Get(did) + atprotoClient = atproto.NewClient(pdsEndpoint, did, accessToken) } else { - slog.Debug("Creating ATProto client with app-password", - "component", "registry/middleware", - "did", did, - "authMethod", authMethod, - "token_length", len(accessToken)) + atprotoClient = atproto.NewClient(pdsEndpoint, did, "") } - atprotoClient = atproto.NewClient(pdsEndpoint, did, accessToken) + } else { + // Puller != owner - reads only, no auth needed + atprotoClient = atproto.NewClient(pdsEndpoint, did, "") } // IMPORTANT: Use only the image name (not identity/image) for ATProto storage @@ -449,18 +453,20 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // 3. The refresher already caches sessions efficiently (in-memory + DB) // 4. Caching the repository with a stale ATProtoClient causes refresh token errors registryCtx := &storage.RegistryContext{ - DID: did, - Handle: handle, - HoldDID: holdDID, - PDSEndpoint: pdsEndpoint, - Repository: repositoryName, - ServiceToken: serviceToken, // Cached service token from middleware validation - ATProtoClient: atprotoClient, - AuthMethod: authMethod, // Auth method from JWT token - Database: nr.database, - Authorizer: nr.authorizer, - Refresher: nr.refresher, - ReadmeFetcher: nr.readmeFetcher, + DID: did, + Handle: handle, + HoldDID: holdDID, + PDSEndpoint: pdsEndpoint, + Repository: repositoryName, + ServiceToken: serviceToken, // Cached service token from puller's PDS + ATProtoClient: atprotoClient, + AuthMethod: authMethod, // Auth method from JWT token + PullerDID: pullerDID, // Authenticated user making the request + PullerPDSEndpoint: pullerPDSEndpoint, // Puller's PDS for service token refresh + Database: nr.database, + Authorizer: nr.authorizer, + Refresher: nr.refresher, + ReadmeFetcher: nr.readmeFetcher, } return storage.NewRoutingRepository(repo, registryCtx), nil @@ -533,10 +539,17 @@ func (nr *NamespaceResolver) isHoldReachable(ctx context.Context, holdDID string return false } -// ExtractAuthMethod is an HTTP middleware that extracts the auth method from the JWT Authorization header -// and stores it in the request context for later use by the registry middleware +// ExtractAuthMethod is an HTTP middleware that extracts the auth method and puller DID from the JWT Authorization header +// and stores them in the request context for later use by the registry middleware. +// Also stores the HTTP method for routing decisions (GET/HEAD = pull, PUT/POST = push). func ExtractAuthMethod(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + + // Store HTTP method in context for routing decisions + // This is used by routing_repository.go to distinguish pull (GET/HEAD) from push (PUT/POST) + ctx = context.WithValue(ctx, "http.request.method", r.Method) + // Extract Authorization header authHeader := r.Header.Get("Authorization") if authHeader != "" { @@ -549,15 +562,25 @@ func ExtractAuthMethod(next http.Handler) http.Handler { authMethod := token.ExtractAuthMethod(tokenString) if authMethod != "" { // Store in context for registry middleware - ctx := context.WithValue(r.Context(), authMethodKey, authMethod) - r = r.WithContext(ctx) - slog.Debug("Extracted auth method from JWT", - "component", "registry/middleware", - "authMethod", authMethod) + ctx = context.WithValue(ctx, authMethodKey, authMethod) } + + // Extract puller DID (Subject) from JWT + // This is the authenticated user's DID, used for service token requests + pullerDID := token.ExtractSubject(tokenString) + if pullerDID != "" { + ctx = context.WithValue(ctx, pullerDIDKey, pullerDID) + } + + slog.Debug("Extracted auth info from JWT", + "component", "registry/middleware", + "authMethod", authMethod, + "pullerDID", pullerDID, + "httpMethod", r.Method) } } + r = r.WithContext(ctx) next.ServeHTTP(w, r) }) } diff --git a/pkg/appview/storage/context.go b/pkg/appview/storage/context.go index 5921615..e30d504 100644 --- a/pkg/appview/storage/context.go +++ b/pkg/appview/storage/context.go @@ -18,14 +18,18 @@ type DatabaseMetrics interface { // This includes both per-request data (DID, hold) and shared services type RegistryContext struct { // Per-request identity and routing information - DID string // User's DID (e.g., "did:plc:abc123") - Handle string // User's handle (e.g., "alice.bsky.social") - HoldDID string // Hold service DID (e.g., "did:web:hold01.atcr.io") - PDSEndpoint string // User's PDS endpoint URL - Repository string // Image repository name (e.g., "debian") - ServiceToken string // Service token for hold authentication (cached by middleware) - ATProtoClient *atproto.Client // Authenticated ATProto client for this user - AuthMethod string // Auth method used ("oauth" or "app_password") + // Owner = the user whose repository is being accessed + // Puller = the authenticated user making the request (from JWT Subject) + DID string // Owner's DID - whose repo is being accessed (e.g., "did:plc:abc123") + Handle string // Owner's handle (e.g., "alice.bsky.social") + HoldDID string // Hold service DID (e.g., "did:web:hold01.atcr.io") + PDSEndpoint string // Owner's PDS endpoint URL + Repository string // Image repository name (e.g., "debian") + ServiceToken string // Service token for hold authentication (from puller's PDS) + ATProtoClient *atproto.Client // Authenticated ATProto client for the owner + AuthMethod string // Auth method used ("oauth" or "app_password") + PullerDID string // Puller's DID - who is making the request (from JWT Subject) + PullerPDSEndpoint string // Puller's PDS endpoint URL // Shared services (same for all requests) Database DatabaseMetrics // Metrics tracking database diff --git a/pkg/appview/storage/manifest_store.go b/pkg/appview/storage/manifest_store.go index 2a4d2bb..ebfc702 100644 --- a/pkg/appview/storage/manifest_store.go +++ b/pkg/appview/storage/manifest_store.go @@ -8,7 +8,6 @@ import ( "fmt" "io" "log/slog" - "maps" "net/http" "strings" "sync" @@ -180,16 +179,23 @@ func (s *ManifestStore) Put(ctx context.Context, manifest distribution.Manifest, if err != nil { // Log error but don't fail the push - labels are optional slog.Warn("Failed to extract config labels", "error", err) - } else { + } else if len(labels) > 0 { // Initialize annotations map if needed if manifestRecord.Annotations == nil { manifestRecord.Annotations = make(map[string]string) } - // Copy labels to annotations (Dockerfile LABELs → manifest annotations) - maps.Copy(manifestRecord.Annotations, labels) + // Copy labels to annotations as fallback + // Only set label values for keys NOT already in manifest annotations + // This ensures explicit annotations take precedence over Dockerfile LABELs + // (which may be inherited from base images) + for key, value := range labels { + if _, exists := manifestRecord.Annotations[key]; !exists { + manifestRecord.Annotations[key] = value + } + } - slog.Debug("Extracted labels from config blob", "count", len(labels)) + slog.Debug("Merged labels from config blob", "labelsCount", len(labels), "annotationsCount", len(manifestRecord.Annotations)) } } diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index 70cf324..b990d3a 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -64,13 +64,13 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { return blobStore } - // Determine if this is a pull (GET) or push (PUT/POST/HEAD/etc) operation + // Determine if this is a pull (GET/HEAD) or push (PUT/POST/etc) operation // Pull operations use the historical hold DID from the database (blobs are where they were pushed) // Push operations use the discovery-based hold DID from user's profile/default // This allows users to change their default hold and have new pushes go there isPull := false if method, ok := ctx.Value("http.request.method").(string); ok { - isPull = method == "GET" + isPull = method == "GET" || method == "HEAD" } holdDID := r.Ctx.HoldDID // Default to discovery-based DID diff --git a/pkg/appview/storage/routing_repository_test.go b/pkg/appview/storage/routing_repository_test.go index 99bc99d..a32e8cb 100644 --- a/pkg/appview/storage/routing_repository_test.go +++ b/pkg/appview/storage/routing_repository_test.go @@ -109,28 +109,30 @@ func TestRoutingRepository_ManifestStoreCaching(t *testing.T) { assert.NotNil(t, repo.manifestStore) } -// TestRoutingRepository_Blobs_PullUsesDatabase tests that GET (pull) uses database hold DID +// TestRoutingRepository_Blobs_PullUsesDatabase tests that GET and HEAD (pull) use database hold DID func TestRoutingRepository_Blobs_PullUsesDatabase(t *testing.T) { dbHoldDID := "did:web:database.hold.io" discoveryHoldDID := "did:web:discovery.hold.io" - ctx := &RegistryContext{ - DID: "did:plc:test123", - Repository: "myapp", - HoldDID: discoveryHoldDID, // Discovery-based hold (should be overridden for pull) - ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:test123", ""), - Database: &mockDatabase{holdDID: dbHoldDID}, + // Test both GET and HEAD as pull operations + for _, method := range []string{"GET", "HEAD"} { + // Reset context for each test + ctx := &RegistryContext{ + DID: "did:plc:test123", + Repository: "myapp-" + method, // Unique repo to avoid caching + HoldDID: discoveryHoldDID, + ATProtoClient: atproto.NewClient("https://pds.example.com", "did:plc:test123", ""), + Database: &mockDatabase{holdDID: dbHoldDID}, + } + repo := NewRoutingRepository(nil, ctx) + + pullCtx := context.WithValue(context.Background(), "http.request.method", method) + blobStore := repo.Blobs(pullCtx) + + assert.NotNil(t, blobStore) + // Verify the hold DID was updated to use the database value for pull + assert.Equal(t, dbHoldDID, repo.Ctx.HoldDID, "pull (%s) should use database hold DID", method) } - - repo := NewRoutingRepository(nil, ctx) - - // Create context with GET method (pull operation) - pullCtx := context.WithValue(context.Background(), "http.request.method", "GET") - blobStore := repo.Blobs(pullCtx) - - assert.NotNil(t, blobStore) - // Verify the hold DID was updated to use the database value for pull - assert.Equal(t, dbHoldDID, repo.Ctx.HoldDID, "pull (GET) should use database hold DID") } // TestRoutingRepository_Blobs_PushUsesDiscovery tests that push operations use discovery hold DID @@ -144,7 +146,7 @@ func TestRoutingRepository_Blobs_PushUsesDiscovery(t *testing.T) { }{ {"PUT", "PUT"}, {"POST", "POST"}, - {"HEAD", "HEAD"}, + // HEAD is now treated as pull (like GET) - see TestRoutingRepository_Blobs_Pull {"PATCH", "PATCH"}, {"DELETE", "DELETE"}, } diff --git a/pkg/auth/token/claims.go b/pkg/auth/token/claims.go index 66fdc03..3838345 100644 --- a/pkg/auth/token/claims.go +++ b/pkg/auth/token/claims.go @@ -56,3 +56,22 @@ func ExtractAuthMethod(tokenString string) string { return claims.AuthMethod } + +// ExtractSubject parses a JWT token string and extracts the Subject claim (the user's DID) +// Returns the subject or empty string if not found or token is invalid +// This does NOT validate the token - it only parses it to extract the claim +func ExtractSubject(tokenString string) string { + // Parse token without validation (we only need the claims, validation is done by distribution library) + parser := jwt.NewParser(jwt.WithoutClaimsValidation()) + token, _, err := parser.ParseUnverified(tokenString, &Claims{}) + if err != nil { + return "" // Invalid token format + } + + claims, ok := token.Claims.(*Claims) + if !ok { + return "" // Wrong claims type + } + + return claims.Subject +} diff --git a/pkg/hold/pds/auth.go b/pkg/hold/pds/auth.go index 49551e7..017fd75 100644 --- a/pkg/hold/pds/auth.go +++ b/pkg/hold/pds/auth.go @@ -4,6 +4,7 @@ import ( "context" "encoding/base64" "encoding/json" + "errors" "fmt" "io" "log/slog" @@ -18,6 +19,44 @@ import ( "github.com/golang-jwt/jwt/v5" ) +// Authentication errors +var ( + ErrMissingAuthHeader = errors.New("missing Authorization header") + ErrInvalidAuthFormat = errors.New("invalid Authorization header format") + ErrInvalidAuthScheme = errors.New("invalid authorization scheme: expected 'Bearer' or 'DPoP'") + ErrMissingToken = errors.New("missing token") + ErrMissingDPoPHeader = errors.New("missing DPoP header") +) + +// JWT validation errors +var ( + ErrInvalidJWTFormat = errors.New("invalid JWT format: expected header.payload.signature") + ErrMissingISSClaim = errors.New("missing 'iss' claim in token") + ErrMissingSubClaim = errors.New("missing 'sub' claim in token") + ErrTokenExpired = errors.New("token has expired") +) + +// AuthError provides structured authorization error information +type AuthError struct { + Action string // The action being attempted: "blob:read", "blob:write", "crew:admin" + Reason string // Why access was denied + Required []string // What permission(s) would grant access +} + +func (e *AuthError) Error() string { + return fmt.Sprintf("access denied for %s: %s (required: %s)", + e.Action, e.Reason, strings.Join(e.Required, " or ")) +} + +// NewAuthError creates a new AuthError +func NewAuthError(action, reason string, required ...string) *AuthError { + return &AuthError{ + Action: action, + Reason: reason, + Required: required, + } +} + // HTTPClient interface allows injecting a custom HTTP client for testing type HTTPClient interface { Do(*http.Request) (*http.Response, error) @@ -44,13 +83,13 @@ func ValidateDPoPRequest(r *http.Request, httpClient HTTPClient) (*ValidatedUser // Extract Authorization header authHeader := r.Header.Get("Authorization") if authHeader == "" { - return nil, fmt.Errorf("missing Authorization header") + return nil, ErrMissingAuthHeader } // Check for DPoP authorization scheme parts := strings.SplitN(authHeader, " ", 2) if len(parts) != 2 { - return nil, fmt.Errorf("invalid Authorization header format") + return nil, ErrInvalidAuthFormat } if parts[0] != "DPoP" { @@ -59,13 +98,13 @@ func ValidateDPoPRequest(r *http.Request, httpClient HTTPClient) (*ValidatedUser accessToken := parts[1] if accessToken == "" { - return nil, fmt.Errorf("missing access token") + return nil, ErrMissingToken } // Extract DPoP header dpopProof := r.Header.Get("DPoP") if dpopProof == "" { - return nil, fmt.Errorf("missing DPoP header") + return nil, ErrMissingDPoPHeader } // TODO: We could verify the DPoP proof locally (signature, HTM, HTU, etc.) @@ -109,7 +148,7 @@ func extractDIDFromToken(token string) (string, string, error) { // JWT format: header.payload.signature parts := strings.Split(token, ".") if len(parts) != 3 { - return "", "", fmt.Errorf("invalid JWT format") + return "", "", ErrInvalidJWTFormat } // Decode payload (base64url) @@ -129,11 +168,11 @@ func extractDIDFromToken(token string) (string, string, error) { } if claims.Sub == "" { - return "", "", fmt.Errorf("missing sub claim (DID)") + return "", "", ErrMissingSubClaim } if claims.Iss == "" { - return "", "", fmt.Errorf("missing iss claim (PDS)") + return "", "", ErrMissingISSClaim } return claims.Sub, claims.Iss, nil @@ -216,7 +255,7 @@ func ValidateOwnerOrCrewAdmin(r *http.Request, pds *HoldPDS, httpClient HTTPClie return nil, fmt.Errorf("DPoP authentication failed: %w", err) } } else { - return nil, fmt.Errorf("missing or invalid Authorization header (expected Bearer or DPoP)") + return nil, ErrInvalidAuthScheme } // Get captain record to check owner @@ -243,12 +282,12 @@ func ValidateOwnerOrCrewAdmin(r *http.Request, pds *HoldPDS, httpClient HTTPClie return user, nil } // User is crew but doesn't have admin permission - return nil, fmt.Errorf("crew member lacks required 'crew:admin' permission") + return nil, NewAuthError("crew:admin", "crew member lacks permission", "crew:admin") } } // User is neither owner nor authorized crew - return nil, fmt.Errorf("user is not authorized (must be hold owner or crew admin)") + return nil, NewAuthError("crew:admin", "user is not a crew member", "crew:admin") } // ValidateBlobWriteAccess validates that the request has valid authentication @@ -276,7 +315,7 @@ func ValidateBlobWriteAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClien return nil, fmt.Errorf("DPoP authentication failed: %w", err) } } else { - return nil, fmt.Errorf("missing or invalid Authorization header (expected Bearer or DPoP)") + return nil, ErrInvalidAuthScheme } // Get captain record to check owner and public settings @@ -303,17 +342,18 @@ func ValidateBlobWriteAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClien return user, nil } // User is crew but doesn't have write permission - return nil, fmt.Errorf("crew member lacks required 'blob:write' permission") + return nil, NewAuthError("blob:write", "crew member lacks permission", "blob:write") } } // User is neither owner nor authorized crew - return nil, fmt.Errorf("user is not authorized for blob write (must be hold owner or crew with blob:write permission)") + return nil, NewAuthError("blob:write", "user is not a crew member", "blob:write") } // ValidateBlobReadAccess validates that the request has read access to blobs // If captain.public = true: No auth required (returns nil user to indicate public access) -// If captain.public = false: Requires valid DPoP + OAuth and (captain OR crew with blob:read permission). +// If captain.public = false: Requires valid DPoP + OAuth and (captain OR crew with blob:read or blob:write permission). +// Note: blob:write implicitly grants blob:read access. // The httpClient parameter is optional and defaults to http.DefaultClient if nil. func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient) (*ValidatedUser, error) { // Get captain record to check public setting @@ -344,7 +384,7 @@ func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient return nil, fmt.Errorf("DPoP authentication failed: %w", err) } } else { - return nil, fmt.Errorf("missing or invalid Authorization header (expected Bearer or DPoP)") + return nil, ErrInvalidAuthScheme } // Check if user is the owner (always has read access) @@ -352,7 +392,8 @@ func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient return user, nil } - // Check if user is crew with blob:read permission + // Check if user is crew with blob:read or blob:write permission + // Note: blob:write implicitly grants blob:read access crew, err := pds.ListCrewMembers(r.Context()) if err != nil { return nil, fmt.Errorf("failed to check crew membership: %w", err) @@ -360,17 +401,19 @@ func ValidateBlobReadAccess(r *http.Request, pds *HoldPDS, httpClient HTTPClient for _, member := range crew { if member.Record.Member == user.DID { - // Check if this crew member has blob:read permission - if slices.Contains(member.Record.Permissions, "blob:read") { + // Check if this crew member has blob:read or blob:write permission + // blob:write implicitly grants read access (can't push without pulling) + if slices.Contains(member.Record.Permissions, "blob:read") || + slices.Contains(member.Record.Permissions, "blob:write") { return user, nil } - // User is crew but doesn't have read permission - return nil, fmt.Errorf("crew member lacks required 'blob:read' permission") + // User is crew but doesn't have read or write permission + return nil, NewAuthError("blob:read", "crew member lacks permission", "blob:read", "blob:write") } } // User is neither owner nor authorized crew - return nil, fmt.Errorf("user is not authorized for blob read (must be hold owner or crew with blob:read permission)") + return nil, NewAuthError("blob:read", "user is not a crew member", "blob:read", "blob:write") } // ServiceTokenClaims represents the claims in a service token JWT @@ -385,13 +428,13 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient // Extract Authorization header authHeader := r.Header.Get("Authorization") if authHeader == "" { - return nil, fmt.Errorf("missing Authorization header") + return nil, ErrMissingAuthHeader } // Check for Bearer authorization scheme parts := strings.SplitN(authHeader, " ", 2) if len(parts) != 2 { - return nil, fmt.Errorf("invalid Authorization header format") + return nil, ErrInvalidAuthFormat } if parts[0] != "Bearer" { @@ -400,7 +443,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient tokenString := parts[1] if tokenString == "" { - return nil, fmt.Errorf("missing token") + return nil, ErrMissingToken } slog.Debug("Validating service token", "holdDID", holdDID) @@ -409,7 +452,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient // Split token: header.payload.signature tokenParts := strings.Split(tokenString, ".") if len(tokenParts) != 3 { - return nil, fmt.Errorf("invalid JWT format") + return nil, ErrInvalidJWTFormat } // Decode payload (second part) to extract claims @@ -427,7 +470,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient // Get issuer (user DID) issuerDID := claims.Issuer if issuerDID == "" { - return nil, fmt.Errorf("missing iss claim") + return nil, ErrMissingISSClaim } // Verify audience matches this hold service @@ -445,7 +488,7 @@ func ValidateServiceToken(r *http.Request, holdDID string, httpClient HTTPClient return nil, fmt.Errorf("failed to get expiration: %w", err) } if exp != nil && time.Now().After(exp.Time) { - return nil, fmt.Errorf("token has expired") + return nil, ErrTokenExpired } // Verify JWT signature using ATProto's secp256k1 crypto diff --git a/pkg/hold/pds/auth_test.go b/pkg/hold/pds/auth_test.go index 95f6c54..e688168 100644 --- a/pkg/hold/pds/auth_test.go +++ b/pkg/hold/pds/auth_test.go @@ -771,6 +771,116 @@ func TestValidateBlobReadAccess_PrivateHold(t *testing.T) { } } +// TestValidateBlobReadAccess_BlobWriteImpliesRead tests that blob:write grants read access +func TestValidateBlobReadAccess_BlobWriteImpliesRead(t *testing.T) { + ownerDID := "did:plc:owner123" + + pds, ctx := setupTestPDSWithBootstrap(t, ownerDID, false, false) + + // Verify captain record has public=false (private hold) + _, captain, err := pds.GetCaptainRecord(ctx) + if err != nil { + t.Fatalf("Failed to get captain record: %v", err) + } + + if captain.Public { + t.Error("Expected public=false for captain record") + } + + // Add crew member with ONLY blob:write permission (no blob:read) + writerDID := "did:plc:writer123" + _, err = pds.AddCrewMember(ctx, writerDID, "writer", []string{"blob:write"}) + if err != nil { + t.Fatalf("Failed to add crew writer: %v", err) + } + + mockClient := &mockPDSClient{} + + // Test writer (has only blob:write permission) can read + t.Run("crew with blob:write can read", func(t *testing.T) { + dpopHelper, err := NewDPoPTestHelper(writerDID, "https://test-pds.example.com") + if err != nil { + t.Fatalf("Failed to create DPoP helper: %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + if err := dpopHelper.AddDPoPToRequest(req); err != nil { + t.Fatalf("Failed to add DPoP to request: %v", err) + } + + // This should SUCCEED because blob:write implies blob:read + user, err := ValidateBlobReadAccess(req, pds, mockClient) + if err != nil { + t.Errorf("Expected blob:write to grant read access, got error: %v", err) + } + + if user == nil { + t.Error("Expected user to be returned for valid read access") + } else if user.DID != writerDID { + t.Errorf("Expected user DID %s, got %s", writerDID, user.DID) + } + }) + + // Also verify that crew with only blob:read still works + t.Run("crew with blob:read can read", func(t *testing.T) { + readerDID := "did:plc:reader123" + _, err = pds.AddCrewMember(ctx, readerDID, "reader", []string{"blob:read"}) + if err != nil { + t.Fatalf("Failed to add crew reader: %v", err) + } + + dpopHelper, err := NewDPoPTestHelper(readerDID, "https://test-pds.example.com") + if err != nil { + t.Fatalf("Failed to create DPoP helper: %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + if err := dpopHelper.AddDPoPToRequest(req); err != nil { + t.Fatalf("Failed to add DPoP to request: %v", err) + } + + user, err := ValidateBlobReadAccess(req, pds, mockClient) + if err != nil { + t.Errorf("Expected blob:read to grant read access, got error: %v", err) + } + + if user == nil { + t.Error("Expected user to be returned for valid read access") + } else if user.DID != readerDID { + t.Errorf("Expected user DID %s, got %s", readerDID, user.DID) + } + }) + + // Verify crew with neither permission cannot read + t.Run("crew without read or write cannot read", func(t *testing.T) { + noPermDID := "did:plc:noperm123" + _, err = pds.AddCrewMember(ctx, noPermDID, "noperm", []string{"crew:admin"}) + if err != nil { + t.Fatalf("Failed to add crew member: %v", err) + } + + dpopHelper, err := NewDPoPTestHelper(noPermDID, "https://test-pds.example.com") + if err != nil { + t.Fatalf("Failed to create DPoP helper: %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/test", nil) + if err := dpopHelper.AddDPoPToRequest(req); err != nil { + t.Fatalf("Failed to add DPoP to request: %v", err) + } + + _, err = ValidateBlobReadAccess(req, pds, mockClient) + if err == nil { + t.Error("Expected error for crew without read or write permission") + } + + // Verify error message format + if !strings.Contains(err.Error(), "access denied for blob:read") { + t.Errorf("Expected structured error message, got: %v", err) + } + }) +} + // TestValidateOwnerOrCrewAdmin tests admin permission checking func TestValidateOwnerOrCrewAdmin(t *testing.T) { ownerDID := "did:plc:owner123"