mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-01 15:56:58 +00:00
fix issues pulling other users images. fix labels taking priority over annotations. fix various auth errors
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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.**
|
||||
+127
-104
@@ -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)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"},
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
+70
-27
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user