From 5d52007104e91cd91b067f147a4bea7daa598484 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Tue, 21 Oct 2025 09:21:51 -0500 Subject: [PATCH] general bug fixes --- .tangled/workflows/tests.yml | 4 +- cmd/appview/serve.go | 5 + cmd/oauth-helper/main.go | 5 +- pkg/appview/config.go | 7 +- pkg/appview/db/session_store.go | 18 + pkg/appview/middleware/registry.go | 110 ++++- pkg/appview/storage/context.go | 1 + pkg/appview/storage/proxy_blob_store.go | 123 +----- pkg/appview/storage/proxy_blob_store_test.go | 304 ++++++------- pkg/appview/storage/routing_repository.go | 36 +- pkg/atproto/client.go | 28 +- pkg/atproto/client_test.go | 10 +- pkg/atproto/endpoints.go | 42 ++ pkg/atproto/manifest_store_test.go | 2 +- pkg/auth/hold_remote.go | 8 +- pkg/auth/oauth/refresher.go | 32 +- pkg/auth/oauth/server.go | 4 +- pkg/auth/{atproto => }/session.go | 5 +- pkg/auth/token/cache.go | 169 ++++++++ pkg/auth/token/handler.go | 5 +- pkg/hold/oci/xrpc_test.go | 37 +- pkg/hold/pds/auth.go | 3 +- pkg/hold/pds/auth_test.go | 3 +- pkg/hold/pds/xrpc.go | 30 +- pkg/hold/pds/xrpc_test.go | 120 +++--- test-e2e.sh | 432 +++++++++++++++++++ 26 files changed, 1096 insertions(+), 447 deletions(-) rename pkg/auth/{atproto => }/session.go (98%) create mode 100644 pkg/auth/token/cache.go create mode 100755 test-e2e.sh diff --git a/.tangled/workflows/tests.yml b/.tangled/workflows/tests.yml index 9b04e59..46ca750 100644 --- a/.tangled/workflows/tests.yml +++ b/.tangled/workflows/tests.yml @@ -8,10 +8,12 @@ engine: "nixery" dependencies: nixpkgs: - - git + - gcc - go steps: - name: Run Tests + environment: + CGO_ENABLED: 1 command: | go test -cover ./... \ No newline at end of file diff --git a/cmd/appview/serve.go b/cmd/appview/serve.go index 0d1efa9..c87dc68 100644 --- a/cmd/appview/serve.go +++ b/cmd/appview/serve.go @@ -151,6 +151,11 @@ func serveRegistry(cmd *cobra.Command, args []string) error { // Create oauth token refresher refresher := oauth.NewRefresher(oauthApp) + // Wire up UI session store to refresher so it can invalidate UI sessions on OAuth failures + if uiSessionStore != nil { + refresher.SetUISessionStore(uiSessionStore) + } + // Set global refresher for middleware middleware.SetGlobalRefresher(refresher) diff --git a/cmd/oauth-helper/main.go b/cmd/oauth-helper/main.go index 791cf94..f232a9b 100644 --- a/cmd/oauth-helper/main.go +++ b/cmd/oauth-helper/main.go @@ -11,6 +11,7 @@ import ( "os" "time" + "atcr.io/pkg/atproto" "atcr.io/pkg/auth/oauth" indigo_oauth "github.com/bluesky-social/indigo/atproto/auth/oauth" ) @@ -108,8 +109,8 @@ func main() { // Generate DPoP proof for deleteRecord endpoint if all params provided if *repo != "" && *rkey != "" { - deleteURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.deleteRecord?repo=%s&collection=%s&rkey=%s", - *holdURL, *repo, *collection, *rkey) + deleteURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&rkey=%s", + *holdURL, atproto.RepoDeleteRecord, *repo, *collection, *rkey) dpopProof, err := generateDPoPProof(result.Session, "POST", deleteURL) if err != nil { diff --git a/pkg/appview/config.go b/pkg/appview/config.go index 61b78df..8d0615e 100644 --- a/pkg/appview/config.go +++ b/pkg/appview/config.go @@ -41,7 +41,7 @@ func LoadConfigFromEnv() (*configuration.Configuration, error) { config.Middleware = buildMiddlewareConfig(defaultHoldDID) // Auth - baseURL := getBaseURL(httpConfig.Addr) + baseURL := GetBaseURL(httpConfig.Addr) authConfig, err := buildAuthConfig(baseURL) if err != nil { return nil, fmt.Errorf("failed to build auth config: %w", err) @@ -200,11 +200,6 @@ func GetBaseURL(httpAddr string) string { return fmt.Sprintf("http://%s", httpAddr) } -// getBaseURL is the internal version used by buildAuthConfig -func getBaseURL(httpAddr string) string { - return GetBaseURL(httpAddr) -} - // getServiceName extracts service name from base URL or uses env var func getServiceName(baseURL string) string { // Check env var first diff --git a/pkg/appview/db/session_store.go b/pkg/appview/db/session_store.go index f3f7a63..42d92ab 100644 --- a/pkg/appview/db/session_store.go +++ b/pkg/appview/db/session_store.go @@ -128,6 +128,24 @@ func (s *SessionStore) Delete(id string) { } } +// DeleteByDID removes all sessions for a given DID +// This is useful when OAuth refresh fails and we need to force re-authentication +func (s *SessionStore) DeleteByDID(did string) { + result, err := s.db.Exec(` + DELETE FROM ui_sessions WHERE did = ? + `, did) + + if err != nil { + fmt.Printf("Warning: Failed to delete sessions for DID %s: %v\n", did, err) + return + } + + deleted, _ := result.RowsAffected() + if deleted > 0 { + fmt.Printf("Deleted %d UI session(s) for DID %s due to OAuth failure\n", deleted, did) + } +} + // Cleanup removes expired sessions func (s *SessionStore) Cleanup() { result, err := s.db.Exec(` diff --git a/pkg/appview/middleware/registry.go b/pkg/appview/middleware/registry.go index eeeebf5..7e58b43 100644 --- a/pkg/appview/middleware/registry.go +++ b/pkg/appview/middleware/registry.go @@ -4,12 +4,17 @@ import ( "context" "encoding/json" "fmt" + "io" + "net/http" + "net/url" "strings" "sync" + "time" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/distribution/distribution/v3" + "github.com/distribution/distribution/v3/registry/api/errcode" registrymw "github.com/distribution/distribution/v3/registry/middleware/registry" "github.com/distribution/distribution/v3/registry/storage/driver" "github.com/distribution/reference" @@ -18,6 +23,7 @@ import ( "atcr.io/pkg/atproto" "atcr.io/pkg/auth" "atcr.io/pkg/auth/oauth" + "atcr.io/pkg/auth/token" ) // Global variables for initialization only @@ -140,6 +146,102 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name } ctx = context.WithValue(ctx, "hold.did", holdDID) + // Get service token for hold authentication + // Check cache first to avoid unnecessary PDS calls on every request + var serviceToken string + if nr.refresher != nil { + cachedToken, expiresAt := token.GetServiceToken(did, holdDID) + + // Use cached token if it exists and has > 10s remaining + if cachedToken != "" && time.Until(expiresAt) > 10*time.Second { + fmt.Printf("DEBUG [registry/middleware]: Using cached service token for DID=%s (expires in %v)\n", + did, time.Until(expiresAt).Round(time.Second)) + serviceToken = cachedToken + } else { + // Cache miss or expiring soon - validate OAuth and get new service token + if cachedToken == "" { + fmt.Printf("DEBUG [registry/middleware]: Cache miss, fetching service token for DID=%s\n", did) + } else { + fmt.Printf("DEBUG [registry/middleware]: Token expiring soon, proactively renewing for DID=%s\n", did) + } + + session, err := nr.refresher.GetSession(ctx, did) + if err != nil { + // OAuth session unavailable - fail fast with proper auth error + nr.refresher.InvalidateSession(did) + token.InvalidateServiceToken(did, holdDID) + fmt.Printf("ERROR [registry/middleware]: Failed to get OAuth session for DID=%s: %v\n", did, err) + fmt.Printf("ERROR [registry/middleware]: User needs to re-authenticate via credential helper\n") + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session expired - please re-authenticate") + } + + // Call com.atproto.server.getServiceAuth on the user's PDS + // Request 5-minute expiry (PDS may grant less) + // exp must be absolute Unix timestamp, not relative duration + expiryTime := time.Now().Unix() + 300 // 5 minutes from now + serviceAuthURL := fmt.Sprintf("%s%s?aud=%s&lxm=%s&exp=%d", + pdsEndpoint, + atproto.ServerGetServiceAuth, + url.QueryEscape(holdDID), + url.QueryEscape("com.atproto.repo.getRecord"), + expiryTime, + ) + + req, err := http.NewRequestWithContext(ctx, "GET", serviceAuthURL, nil) + if err != nil { + fmt.Printf("ERROR [registry/middleware]: Failed to create service auth request: %v\n", err) + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session validation failed") + } + + // Use OAuth session to authenticate to PDS (with DPoP) + resp, err := session.DoWithAuth(session.Client, req, "com.atproto.server.getServiceAuth") + if err != nil { + // Invalidate session on auth errors (may indicate corrupted session or expired tokens) + nr.refresher.InvalidateSession(did) + token.InvalidateServiceToken(did, holdDID) + fmt.Printf("ERROR [registry/middleware]: OAuth validation failed for DID=%s: %v\n", did, err) + fmt.Printf("ERROR [registry/middleware]: User needs to re-authenticate via credential helper\n") + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session expired - please re-authenticate") + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + // Invalidate session on auth failures + bodyBytes, _ := io.ReadAll(resp.Body) + nr.refresher.InvalidateSession(did) + token.InvalidateServiceToken(did, holdDID) + fmt.Printf("ERROR [registry/middleware]: OAuth validation failed for DID=%s: status %d, body: %s\n", + did, resp.StatusCode, string(bodyBytes)) + fmt.Printf("ERROR [registry/middleware]: User needs to re-authenticate via credential helper\n") + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session expired - please re-authenticate") + } + + // Parse response to get service token + var result struct { + Token string `json:"token"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + fmt.Printf("ERROR [registry/middleware]: Failed to decode service auth response: %v\n", err) + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session validation failed") + } + + if result.Token == "" { + fmt.Printf("ERROR [registry/middleware]: Empty token in service auth response\n") + return nil, errcode.ErrorCodeUnauthorized.WithDetail("OAuth session validation failed") + } + + serviceToken = result.Token + + // Cache the token (parses JWT to extract actual expiry) + if err := token.SetServiceToken(did, holdDID, serviceToken); err != nil { + fmt.Printf("WARN [registry/middleware]: Failed to cache service token: %v\n", err) + // Non-fatal - we have the token, just won't be cached + } + + fmt.Printf("DEBUG [registry/middleware]: OAuth validation succeeded for DID=%s\n", did) + } + } + // Create a new reference with identity/image format // Use the identity (or DID) as the namespace to ensure canonical format // This transforms: evan.jarrett.net/debian -> evan.jarrett.net/debian (keeps full path) @@ -192,9 +294,12 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Cache key is DID + repository name cacheKey := did + ":" + repositoryName - // Check cache first + // Check cache first and update service token if cached, ok := nr.repositories.Load(cacheKey); ok { - return cached.(*storage.RoutingRepository), nil + cachedRepo := cached.(*storage.RoutingRepository) + // Always update the service token even for cached repos (token may have been renewed) + cachedRepo.Ctx.ServiceToken = serviceToken + return cachedRepo, nil } // Create routing repository - routes manifests to ATProto, blobs to hold service @@ -205,6 +310,7 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name HoldDID: holdDID, PDSEndpoint: pdsEndpoint, Repository: repositoryName, + ServiceToken: serviceToken, // Cached service token from middleware validation ATProtoClient: atprotoClient, Database: nr.database, Authorizer: nr.authorizer, diff --git a/pkg/appview/storage/context.go b/pkg/appview/storage/context.go index 52dd412..1421e37 100644 --- a/pkg/appview/storage/context.go +++ b/pkg/appview/storage/context.go @@ -20,6 +20,7 @@ type RegistryContext struct { 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 // Shared services (same for all requests) diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index b9e931e..3e334d4 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -7,13 +7,13 @@ import ( "fmt" "io" "net/http" - "net/url" "sync" "time" "atcr.io/pkg/appview" "atcr.io/pkg/atproto" "github.com/distribution/distribution/v3" + "github.com/distribution/distribution/v3/registry/api/errcode" "github.com/opencontainers/go-digest" ) @@ -30,20 +30,6 @@ var ( globalUploadsMu sync.RWMutex ) -// Service token cache entry -type serviceTokenEntry struct { - token string - expiresAt time.Time -} - -// Global service token cache (shared across all ProxyBlobStore instances) -// Cache key: "userDID:holdDID" -// Tokens are valid for 60 seconds from PDS, we cache for 50 seconds to be safe -var ( - globalServiceTokens = make(map[string]*serviceTokenEntry) - globalServiceTokensMu sync.RWMutex -) - // ProxyBlobStore proxies blob requests to an external storage service type ProxyBlobStore struct { ctx *RegistryContext // All context and services @@ -75,96 +61,19 @@ func NewProxyBlobStore(ctx *RegistryContext) *ProxyBlobStore { } } -// getServiceToken gets a service token for the hold service from the user's PDS -// Uses com.atproto.server." endpoint -// Tokens are cached for 50 seconds (they're valid for 60 seconds from PDS) -func (p *ProxyBlobStore) getServiceToken(ctx context.Context) (string, error) { - // Check cache first - cacheKey := p.ctx.DID + ":" + p.ctx.HoldDID - globalServiceTokensMu.RLock() - entry, exists := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() - - if exists && time.Now().Before(entry.expiresAt) { - fmt.Printf("DEBUG [proxy_blob_store]: Using cached service token for %s\n", cacheKey) - return entry.token, nil - } - - // No valid cached token, request a new one from PDS - if p.ctx.Refresher == nil { - return "", fmt.Errorf("no OAuth refresher available for service token request") - } - - session, err := p.ctx.Refresher.GetSession(ctx, p.ctx.DID) - if err != nil { - return "", fmt.Errorf("failed to get OAuth session: %w", err) - } - - // Call com.atproto.server.getServiceAuth on the user's PDS - // Include lxm (lexicon scope) and exp (expiration) parameters - pdsURL := p.ctx.PDSEndpoint - serviceAuthURL := fmt.Sprintf("%s/xrpc/com.atproto.server.getServiceAuth?aud=%s&lxm=%s", - pdsURL, - url.QueryEscape(p.ctx.HoldDID), - url.QueryEscape("com.atproto.repo.getRecord"), - ) - - req, err := http.NewRequestWithContext(ctx, "GET", serviceAuthURL, nil) - if err != nil { - return "", fmt.Errorf("failed to create service auth request: %w", err) - } - - // Use OAuth session to authenticate to PDS (with DPoP) - resp, err := session.DoWithAuth(session.Client, req, "com.atproto.server.getServiceAuth") - if err != nil { - return "", fmt.Errorf("failed to call getServiceAuth: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - bodyBytes, _ := io.ReadAll(resp.Body) - return "", fmt.Errorf("getServiceAuth failed: status %d, body: %s", resp.StatusCode, string(bodyBytes)) - } - - // Parse response - var result struct { - Token string `json:"token"` - } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { - return "", fmt.Errorf("failed to decode service auth response: %w", err) - } - - if result.Token == "" { - return "", fmt.Errorf("empty token in service auth response") - } - - fmt.Printf("DEBUG [proxy_blob_store]: Got new service token for %s (length=%d)\n", cacheKey, len(result.Token)) - - // Cache the token (expires in 50 seconds) - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: result.Token, - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() - - return result.Token, nil -} - // doAuthenticatedRequest performs an HTTP request with service token authentication -// Gets a service token from the user's PDS and uses it to authenticate to the hold service +// Uses the service token from middleware to authenticate requests to the hold service func (p *ProxyBlobStore) doAuthenticatedRequest(ctx context.Context, req *http.Request) (*http.Response, error) { - // Get service token for the hold service - serviceToken, err := p.getServiceToken(ctx) - if err != nil { - fmt.Printf("DEBUG [proxy_blob_store]: Failed to get service token for DID=%s: %v, will attempt without auth\n", p.ctx.DID, err) - // Fall back to non-authenticated request - return p.httpClient.Do(req) + // Use service token that middleware already validated and cached + // Middleware fails fast with HTTP 401 if OAuth session is invalid + if p.ctx.ServiceToken == "" { + // Should never happen - middleware validates OAuth before handlers run + fmt.Printf("ERROR [proxy_blob_store]: No service token in context for DID=%s\n", p.ctx.DID) + return nil, fmt.Errorf("no service token available (middleware should have validated)") } // Add Bearer token to Authorization header - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", serviceToken)) - fmt.Printf("DEBUG [proxy_blob_store]: Using service token for hold service request, DID=%s\n", p.ctx.DID) + req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", p.ctx.ServiceToken)) return p.httpClient.Do(req) } @@ -408,11 +317,9 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo // Start multipart upload via hold service uploadID, err := p.startMultipartUpload(ctx, tempDigest) if err != nil { - return nil, fmt.Errorf("failed to start multipart upload: %w", err) + return nil, err } - fmt.Printf(" Started multipart upload: uploadID=%s\n", uploadID) - writer := &ProxyBlobWriter{ store: p, options: opts, @@ -452,8 +359,8 @@ func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation string, // Use XRPC endpoint: /xrpc/com.atproto.sync.getBlob?did={userDID}&cid={digest} // The 'did' parameter is the USER's DID (whose blob we're fetching), not the hold service DID // Per migration doc: hold accepts OCI digest directly as cid parameter (checks for sha256: prefix) - xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s&method=%s", - p.holdURL, p.ctx.DID, dgst.String(), operation) + xrpcURL := fmt.Sprintf("%s%s?did=%s&cid=%s&method=%s", + p.holdURL, atproto.SyncGetBlob, p.ctx.DID, dgst.String(), operation) req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) if err != nil { @@ -462,7 +369,11 @@ func (p *ProxyBlobStore) getPresignedURL(ctx context.Context, operation string, resp, err := p.doAuthenticatedRequest(ctx, req) if err != nil { - return "", fmt.Errorf("failed to call hold service: %w", err) + // Don't wrap errcode errors - return them directly + if _, ok := err.(errcode.Error); ok { + return "", err + } + return "", fmt.Errorf("failed to get presigned URL: %w", err) } defer resp.Body.Close() diff --git a/pkg/appview/storage/proxy_blob_store_test.go b/pkg/appview/storage/proxy_blob_store_test.go index 6fa689b..9691621 100644 --- a/pkg/appview/storage/proxy_blob_store_test.go +++ b/pkg/appview/storage/proxy_blob_store_test.go @@ -2,7 +2,9 @@ package storage import ( "context" + "encoding/base64" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -10,112 +12,97 @@ import ( "time" "atcr.io/pkg/atproto" + "atcr.io/pkg/auth/token" "github.com/opencontainers/go-digest" ) // TestGetServiceToken_CachingLogic tests the token caching mechanism func TestGetServiceToken_CachingLogic(t *testing.T) { - // Clear cache before test - globalServiceTokensMu.Lock() - globalServiceTokens = make(map[string]*serviceTokenEntry) - globalServiceTokensMu.Unlock() + userDID := "did:plc:test" + holdDID := "did:web:hold.example.com" - // Test 1: Empty cache - cacheKey := "did:plc:test:did:web:hold.example.com" - globalServiceTokensMu.RLock() - _, exists := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() - - if exists { + // Test 1: Empty cache - invalidate any existing token + token.InvalidateServiceToken(userDID, holdDID) + cachedToken, _ := token.GetServiceToken(userDID, holdDID) + if cachedToken != "" { t.Error("Expected empty cache at start") } // Test 2: Insert token into cache - testToken := "test-token-12345" - expiresAt := time.Now().Add(50 * time.Second) + // Create a JWT-like token with exp claim for testing + // Format: header.payload.signature where payload has exp claim + testPayload := fmt.Sprintf(`{"exp":%d}`, time.Now().Add(50*time.Second).Unix()) + testToken := "eyJhbGciOiJIUzI1NiJ9." + base64URLEncode(testPayload) + ".signature" - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: testToken, - expiresAt: expiresAt, + err := token.SetServiceToken(userDID, holdDID, testToken) + if err != nil { + t.Fatalf("Failed to set service token: %v", err) } - globalServiceTokensMu.Unlock() // Test 3: Retrieve from cache - globalServiceTokensMu.RLock() - entry, exists := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() - - if !exists { + cachedToken, expiresAt := token.GetServiceToken(userDID, holdDID) + if cachedToken == "" { t.Fatal("Expected token to be in cache") } - if entry.token != testToken { - t.Errorf("Expected token %s, got %s", testToken, entry.token) + if cachedToken != testToken { + t.Errorf("Expected token %s, got %s", testToken, cachedToken) } - if time.Now().After(entry.expiresAt) { + if time.Now().After(expiresAt) { t.Error("Expected token to not be expired") } - // Test 4: Expired token - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: "expired-token", - expiresAt: time.Now().Add(-1 * time.Hour), - } - globalServiceTokensMu.Unlock() + // Test 4: Expired token - GetServiceToken automatically removes it + expiredPayload := fmt.Sprintf(`{"exp":%d}`, time.Now().Add(-1*time.Hour).Unix()) + expiredToken := "eyJhbGciOiJIUzI1NiJ9." + base64URLEncode(expiredPayload) + ".signature" + token.SetServiceToken(userDID, holdDID, expiredToken) - globalServiceTokensMu.RLock() - expiredEntry := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() - - if !time.Now().After(expiredEntry.expiresAt) { - t.Error("Expected token to be expired") + // GetServiceToken should return empty string for expired token + cachedToken, _ = token.GetServiceToken(userDID, holdDID) + if cachedToken != "" { + t.Error("Expected expired token to be removed from cache") } } -// TestGetServiceToken_NoRefresher tests that getServiceToken returns error when refresher is nil -func TestGetServiceToken_NoRefresher(t *testing.T) { +// base64URLEncode helper for creating test JWT tokens +func base64URLEncode(data string) string { + return strings.TrimRight(base64.URLEncoding.EncodeToString([]byte(data)), "=") +} + +// TestServiceToken_EmptyInContext tests that operations fail when service token is missing +func TestServiceToken_EmptyInContext(t *testing.T) { ctx := &RegistryContext{ - DID: "did:plc:test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", - Refresher: nil, // No refresher + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "", // No service token (middleware didn't set it) + Refresher: nil, } store := NewProxyBlobStore(ctx) - // Clear cache to force token fetch attempt - globalServiceTokensMu.Lock() - delete(globalServiceTokens, "did:plc:test:did:web:hold.example.com") - globalServiceTokensMu.Unlock() + // Try a write operation that requires authentication + testDigest := digest.FromString("test-content") + _, err := store.Stat(context.Background(), testDigest) - _, err := store.getServiceToken(context.Background()) + // Should fail because no service token is available if err == nil { - t.Error("Expected error when refresher is nil") + t.Error("Expected error when service token is empty") } - if !strings.Contains(err.Error(), "no OAuth refresher") { - t.Errorf("Expected error about no OAuth refresher, got: %v", err) + // Error should indicate authentication issue + if !strings.Contains(err.Error(), "UNAUTHORIZED") && !strings.Contains(err.Error(), "authentication") { + t.Logf("Got error (acceptable): %v", err) } } // TestDoAuthenticatedRequest_BearerTokenInjection tests that Bearer tokens are added to requests func TestDoAuthenticatedRequest_BearerTokenInjection(t *testing.T) { - // This test verifies the Bearer token injection logic when a token is cached + // This test verifies the Bearer token injection logic - // Setup: Create a cached token - testToken := "cached-bearer-token-xyz" - cacheKey := "did:plc:bearer-test:did:web:hold.example.com" - - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: testToken, - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() + testToken := "test-bearer-token-xyz" // Create a test server to verify the Authorization header var receivedAuthHeader string @@ -125,13 +112,14 @@ func TestDoAuthenticatedRequest_BearerTokenInjection(t *testing.T) { })) defer testServer.Close() - // Create ProxyBlobStore with cached token + // Create ProxyBlobStore with service token in context (set by middleware) ctx := &RegistryContext{ - DID: "did:plc:bearer-test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", - Refresher: nil, // Will use cached token, so refresher not needed + DID: "did:plc:bearer-test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: testToken, // Service token from middleware + Refresher: nil, } store := NewProxyBlobStore(ctx) @@ -156,15 +144,9 @@ func TestDoAuthenticatedRequest_BearerTokenInjection(t *testing.T) { } } -// TestDoAuthenticatedRequest_FallbackWhenTokenUnavailable tests fallback to non-auth -func TestDoAuthenticatedRequest_FallbackWhenTokenUnavailable(t *testing.T) { - // Clear cache - cacheKey := "did:plc:fallback:did:web:hold.example.com" - globalServiceTokensMu.Lock() - delete(globalServiceTokens, cacheKey) - globalServiceTokensMu.Unlock() - - // Create test server +// TestDoAuthenticatedRequest_ErrorWhenTokenUnavailable tests that authentication failures return proper errors +func TestDoAuthenticatedRequest_ErrorWhenTokenUnavailable(t *testing.T) { + // Create test server (should not be called since auth fails first) called := false testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { called = true @@ -172,13 +154,14 @@ func TestDoAuthenticatedRequest_FallbackWhenTokenUnavailable(t *testing.T) { })) defer testServer.Close() - // Create ProxyBlobStore without refresher (will fail to get token and fall back) + // Create ProxyBlobStore without service token (middleware didn't set it) ctx := &RegistryContext{ - DID: "did:plc:fallback", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", - Refresher: nil, // No refresher = can't get token + DID: "did:plc:fallback", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "", // No service token + Refresher: nil, } store := NewProxyBlobStore(ctx) @@ -189,19 +172,23 @@ func TestDoAuthenticatedRequest_FallbackWhenTokenUnavailable(t *testing.T) { t.Fatalf("Failed to create request: %v", err) } - // Do authenticated request - should fall back to non-auth + // Do authenticated request - should fail when no service token resp, err := store.doAuthenticatedRequest(context.Background(), req) - if err != nil { - t.Fatalf("doAuthenticatedRequest should not fail even without token: %v", err) + if err == nil { + t.Fatal("Expected doAuthenticatedRequest to fail when no service token is available") } - defer resp.Body.Close() - - if !called { - t.Error("Expected request to be made despite missing token") + if resp != nil { + resp.Body.Close() } - if resp.StatusCode != http.StatusOK { - t.Errorf("Expected status 200, got %d", resp.StatusCode) + // Verify error indicates authentication/authorization issue + errStr := err.Error() + if !strings.Contains(errStr, "service token") && !strings.Contains(errStr, "UNAUTHORIZED") { + t.Errorf("Expected service token or unauthorized error, got: %v", err) + } + + if called { + t.Error("Expected request to NOT be made when authentication fails") } } @@ -241,34 +228,25 @@ func TestResolveHoldURL(t *testing.T) { // TestServiceTokenCacheExpiry tests that expired cached tokens are not used func TestServiceTokenCacheExpiry(t *testing.T) { - cacheKey := "did:plc:expiry:did:web:hold.example.com" + userDID := "did:plc:expiry" + holdDID := "did:web:hold.example.com" // Insert expired token - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: "expired-token", - expiresAt: time.Now().Add(-1 * time.Hour), // Expired 1 hour ago - } - globalServiceTokensMu.Unlock() + expiredPayload := fmt.Sprintf(`{"exp":%d}`, time.Now().Add(-1*time.Hour).Unix()) + expiredToken := "eyJhbGciOiJIUzI1NiJ9." + base64URLEncode(expiredPayload) + ".signature" + token.SetServiceToken(userDID, holdDID, expiredToken) - // Check that it's expired - globalServiceTokensMu.RLock() - entry := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() + // GetServiceToken should automatically remove expired tokens + cachedToken, expiresAt := token.GetServiceToken(userDID, holdDID) - if entry == nil { - t.Fatal("Expected token entry to exist") + // Should return empty string for expired token + if cachedToken != "" { + t.Error("Expected GetServiceToken to return empty string for expired token") } - if !time.Now().After(entry.expiresAt) { - t.Error("Expected token to be expired") - } - - // The getServiceToken function would check time.Now().Before(entry.expiresAt) - // and this would return false for an expired token, causing it to fetch a new one - shouldUseCache := time.Now().Before(entry.expiresAt) - if shouldUseCache { - t.Error("Expected expired token to not be used from cache") + // expiresAt should be zero time for expired/missing tokens + if !expiresAt.IsZero() { + t.Error("Expected zero time for expired token") } } @@ -327,22 +305,18 @@ func TestNewProxyBlobStore(t *testing.T) { // Benchmark for token cache access func BenchmarkServiceTokenCacheAccess(b *testing.B) { - cacheKey := "did:plc:bench:did:web:hold.example.com" + userDID := "did:plc:bench" + holdDID := "did:web:hold.example.com" - globalServiceTokensMu.Lock() - globalServiceTokens[cacheKey] = &serviceTokenEntry{ - token: "benchmark-token", - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() + testPayload := fmt.Sprintf(`{"exp":%d}`, time.Now().Add(50*time.Second).Unix()) + testTokenStr := "eyJhbGciOiJIUzI1NiJ9." + base64URLEncode(testPayload) + ".signature" + token.SetServiceToken(userDID, holdDID, testTokenStr) b.ResetTimer() for i := 0; i < b.N; i++ { - globalServiceTokensMu.RLock() - entry, exists := globalServiceTokens[cacheKey] - globalServiceTokensMu.RUnlock() + cachedToken, expiresAt := token.GetServiceToken(userDID, holdDID) - if !exists || time.Now().After(entry.expiresAt) { + if cachedToken == "" || time.Now().After(expiresAt) { b.Error("Cache miss in benchmark") } } @@ -374,22 +348,15 @@ func TestCompleteMultipartUpload_JSONFormat(t *testing.T) { // Create store with mocked hold URL ctx := &RegistryContext{ - DID: "did:plc:test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "test-service-token", // Service token from middleware } store := NewProxyBlobStore(ctx) store.holdURL = holdServer.URL - // Setup token cache to avoid auth errors - globalServiceTokensMu.Lock() - globalServiceTokens["did:plc:test:did:web:hold.example.com"] = &serviceTokenEntry{ - token: "test-token", - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() - // Call completeMultipartUpload parts := []CompletedPart{ {PartNumber: 1, ETag: "etag-1"}, @@ -476,24 +443,17 @@ func TestGet_UsesPresignedURLDirectly(t *testing.T) { })) defer holdServer.Close() - // Create store + // Create store with service token in context ctx := &RegistryContext{ - DID: "did:plc:test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "test-service-token", // Service token from middleware } store := NewProxyBlobStore(ctx) store.holdURL = holdServer.URL - // Setup token cache - globalServiceTokensMu.Lock() - globalServiceTokens["did:plc:test:did:web:hold.example.com"] = &serviceTokenEntry{ - token: "test-token", - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() - // Call Get() dgst := digest.FromBytes(blobData) retrieved, err := store.Get(context.Background(), dgst) @@ -544,24 +504,17 @@ func TestOpen_UsesPresignedURLDirectly(t *testing.T) { })) defer holdServer.Close() - // Create store + // Create store with service token in context ctx := &RegistryContext{ - DID: "did:plc:test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "test-service-token", // Service token from middleware } store := NewProxyBlobStore(ctx) store.holdURL = holdServer.URL - // Setup token cache - globalServiceTokensMu.Lock() - globalServiceTokens["did:plc:test:did:web:hold.example.com"] = &serviceTokenEntry{ - token: "test-token", - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() - // Call Open() dgst := digest.FromBytes(blobData) reader, err := store.Open(context.Background(), dgst) @@ -636,24 +589,17 @@ func TestMultipartEndpoints_CorrectURLs(t *testing.T) { })) defer holdServer.Close() - // Create store + // Create store with service token in context ctx := &RegistryContext{ - DID: "did:plc:test", - HoldDID: "did:web:hold.example.com", - PDSEndpoint: "https://pds.example.com", - Repository: "test-repo", + DID: "did:plc:test", + HoldDID: "did:web:hold.example.com", + PDSEndpoint: "https://pds.example.com", + Repository: "test-repo", + ServiceToken: "test-service-token", // Service token from middleware } store := NewProxyBlobStore(ctx) store.holdURL = holdServer.URL - // Setup token cache - globalServiceTokensMu.Lock() - globalServiceTokens["did:plc:test:did:web:hold.example.com"] = &serviceTokenEntry{ - token: "test-token", - expiresAt: time.Now().Add(50 * time.Second), - } - globalServiceTokensMu.Unlock() - // Call the function _ = tt.testFunc(store) // Ignore error, we just care about the URL diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index 7978bd3..fcf4e92 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -13,7 +13,7 @@ import ( // The registry (AppView) is stateless and NEVER stores blobs locally type RoutingRepository struct { distribution.Repository - ctx *RegistryContext // All context and services + Ctx *RegistryContext // All context and services (exported for token updates) manifestStore *atproto.ManifestStore // Cached manifest store instance blobStore *ProxyBlobStore // Cached blob store instance } @@ -22,7 +22,7 @@ type RoutingRepository struct { func NewRoutingRepository(baseRepo distribution.Repository, ctx *RegistryContext) *RoutingRepository { return &RoutingRepository{ Repository: baseRepo, - ctx: ctx, + Ctx: ctx, } } @@ -36,13 +36,13 @@ func (r *RoutingRepository) Manifests(ctx context.Context, options ...distributi // ManifestStore needs both DID and URL for backward compat (legacy holdEndpoint field) // For now, pass holdDID twice (will be cleaned up in manifest_store.go later) r.manifestStore = atproto.NewManifestStore( - r.ctx.ATProtoClient, - r.ctx.Repository, - r.ctx.HoldDID, - r.ctx.HoldDID, - r.ctx.DID, + r.Ctx.ATProtoClient, + r.Ctx.Repository, + r.Ctx.HoldDID, + r.Ctx.HoldDID, + r.Ctx.DID, blobStore, - r.ctx.Database, + r.Ctx.Database, ) } @@ -52,9 +52,9 @@ func (r *RoutingRepository) Manifests(ctx context.Context, options ...distributi time.Sleep(100 * time.Millisecond) // Brief delay to let manifest fetch complete if holdDID := r.manifestStore.GetLastFetchedHoldDID(); holdDID != "" { // Cache for 10 minutes - should cover typical pull operations - GetGlobalHoldCache().Set(r.ctx.DID, r.ctx.Repository, holdDID, 10*time.Minute) + GetGlobalHoldCache().Set(r.Ctx.DID, r.Ctx.Repository, holdDID, 10*time.Minute) fmt.Printf("DEBUG [storage/routing]: Cached hold DID: did=%s, repo=%s, hold=%s\n", - r.ctx.DID, r.ctx.Repository, holdDID) + r.Ctx.DID, r.Ctx.Repository, holdDID) } }() @@ -67,23 +67,23 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { // Return cached blob store if available if r.blobStore != nil { fmt.Printf("DEBUG [storage/blobs]: Returning cached blob store for did=%s, repo=%s\n", - r.ctx.DID, r.ctx.Repository) + r.Ctx.DID, r.Ctx.Repository) return r.blobStore } // For pull operations, check if we have a cached hold DID from a recent manifest fetch // This ensures blobs are fetched from the hold recorded in the manifest, not re-discovered - holdDID := r.ctx.HoldDID // Default to discovery-based DID + holdDID := r.Ctx.HoldDID // Default to discovery-based DID - if cachedHoldDID, ok := GetGlobalHoldCache().Get(r.ctx.DID, r.ctx.Repository); ok { + if cachedHoldDID, ok := GetGlobalHoldCache().Get(r.Ctx.DID, r.Ctx.Repository); ok { // Use cached hold DID from manifest holdDID = cachedHoldDID fmt.Printf("DEBUG [storage/blobs]: Using cached hold from manifest: did=%s, repo=%s, hold=%s\n", - r.ctx.DID, r.ctx.Repository, cachedHoldDID) + r.Ctx.DID, r.Ctx.Repository, cachedHoldDID) } else { // No cached hold, use discovery-based DID (for push or first pull) fmt.Printf("DEBUG [storage/blobs]: Using discovery-based hold: did=%s, repo=%s, hold=%s\n", - r.ctx.DID, r.ctx.Repository, holdDID) + r.Ctx.DID, r.Ctx.Repository, holdDID) } if holdDID == "" { @@ -92,15 +92,15 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { } // Update context with the correct hold DID (may be cached or discovered) - r.ctx.HoldDID = holdDID + r.Ctx.HoldDID = holdDID // Create and cache proxy blob store - r.blobStore = NewProxyBlobStore(r.ctx) + r.blobStore = NewProxyBlobStore(r.Ctx) return r.blobStore } // Tags returns the tag service // Tags are stored in ATProto as io.atcr.tag records func (r *RoutingRepository) Tags(ctx context.Context) distribution.TagService { - return atproto.NewTagStore(r.ctx.ATProtoClient, r.ctx.Repository) + return atproto.NewTagStore(r.Ctx.ATProtoClient, r.Ctx.Repository) } diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index 0d461c7..2c0e143 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -83,7 +83,7 @@ func (c *Client) PutRecord(ctx context.Context, collection, rkey string, record return nil, fmt.Errorf("failed to marshal record: %w", err) } - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.putRecord", c.pdsEndpoint) + url := fmt.Sprintf("%s%s", c.pdsEndpoint, RepoPutRecord) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) if err != nil { return nil, err @@ -137,8 +137,8 @@ func (c *Client) GetRecord(ctx context.Context, collection, rkey string) (*Recor } // Basic Auth (app passwords) - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s", - c.pdsEndpoint, c.did, collection, rkey) + url := fmt.Sprintf("%s%s?repo=%s&collection=%s&rkey=%s", + c.pdsEndpoint, RepoGetRecord, c.did, collection, rkey) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { @@ -203,7 +203,7 @@ func (c *Client) DeleteRecord(ctx context.Context, collection, rkey string) erro return fmt.Errorf("failed to marshal delete request: %w", err) } - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.deleteRecord", c.pdsEndpoint) + url := fmt.Sprintf("%s%s", c.pdsEndpoint, RepoDeleteRecord) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) if err != nil { return err @@ -228,8 +228,8 @@ func (c *Client) DeleteRecord(ctx context.Context, collection, rkey string) erro // ListRecords lists records in a collection func (c *Client) ListRecords(ctx context.Context, collection string, limit int) ([]Record, error) { - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=%d", - c.pdsEndpoint, c.did, collection, limit) + url := fmt.Sprintf("%s%s?repo=%s&collection=%s&limit=%d", + c.pdsEndpoint, RepoListRecords, c.did, collection, limit) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { @@ -301,7 +301,7 @@ func (c *Client) UploadBlob(ctx context.Context, data []byte, mimeType string) ( } // Basic Auth (app passwords) - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.uploadBlob", c.pdsEndpoint) + url := fmt.Sprintf("%s%s", c.pdsEndpoint, RepoUploadBlob) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(data)) if err != nil { return nil, err @@ -333,8 +333,8 @@ func (c *Client) UploadBlob(ctx context.Context, data []byte, mimeType string) ( // GetBlob downloads a blob by its CID from the PDS func (c *Client) GetBlob(ctx context.Context, cid string) ([]byte, error) { - url := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", - c.pdsEndpoint, c.did, cid) + url := fmt.Sprintf("%s%s?did=%s&cid=%s", + c.pdsEndpoint, SyncGetBlob, c.did, cid) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { @@ -405,7 +405,7 @@ type RepoRef struct { // This is a network-wide query, not limited to a single PDS func (c *Client) ListReposByCollection(ctx context.Context, collection string, limit int, cursor string) (*ListReposByCollectionResult, error) { // Build URL with query parameters - url := fmt.Sprintf("%s/xrpc/com.atproto.sync.listReposByCollection?collection=%s", c.pdsEndpoint, collection) + url := fmt.Sprintf("%s%s?collection=%s", c.pdsEndpoint, SyncListReposByCollection, collection) if limit > 0 { url += fmt.Sprintf("&limit=%d", limit) @@ -447,8 +447,8 @@ func (c *Client) ListReposByCollection(ctx context.Context, collection string, l // ListRecordsForRepo lists records in a collection for a specific repo (DID) // This differs from ListRecords which uses the client's DID func (c *Client) ListRecordsForRepo(ctx context.Context, repoDID, collection string, limit int, cursor string) ([]Record, string, error) { - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s", - c.pdsEndpoint, repoDID, collection) + url := fmt.Sprintf("%s%s?repo=%s&collection=%s", + c.pdsEndpoint, RepoListRecords, repoDID, collection) if limit > 0 { url += fmt.Sprintf("&limit=%d", limit) @@ -583,8 +583,8 @@ func (c *Client) GetProfileRecord(ctx context.Context, did string) (*ProfileReco } // Basic Auth (app passwords) - url := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=app.bsky.actor.profile&rkey=self", - c.pdsEndpoint, did) + url := fmt.Sprintf("%s%s?repo=%s&collection=app.bsky.actor.profile&rkey=self", + c.pdsEndpoint, RepoGetRecord, did) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { diff --git a/pkg/atproto/client_test.go b/pkg/atproto/client_test.go index 20c2dd3..2023509 100644 --- a/pkg/atproto/client_test.go +++ b/pkg/atproto/client_test.go @@ -81,7 +81,7 @@ func TestPutRecord(t *testing.T) { } // Verify path - expectedPath := "/xrpc/com.atproto.repo.putRecord" + expectedPath := RepoPutRecord if r.URL.Path != expectedPath { t.Errorf("Path = %v, want %v", r.URL.Path, expectedPath) } @@ -198,7 +198,7 @@ func TestGetRecord(t *testing.T) { } // Verify path - expectedPath := "/xrpc/com.atproto.repo.getRecord" + expectedPath := RepoGetRecord if r.URL.Path != expectedPath { t.Errorf("Path = %v, want %v", r.URL.Path, expectedPath) } @@ -284,7 +284,7 @@ func TestDeleteRecord(t *testing.T) { } // Verify path - expectedPath := "/xrpc/com.atproto.repo.deleteRecord" + expectedPath := RepoDeleteRecord if r.URL.Path != expectedPath { t.Errorf("Path = %v, want %v", r.URL.Path, expectedPath) } @@ -378,8 +378,8 @@ func TestUploadBlob(t *testing.T) { t.Errorf("Method = %v, want POST", r.Method) } - if r.URL.Path != "/xrpc/com.atproto.repo.uploadBlob" { - t.Errorf("Path = %v, want /xrpc/com.atproto.repo.uploadBlob", r.URL.Path) + if r.URL.Path != RepoUploadBlob { + t.Errorf("Path = %v, want %s", r.URL.Path, RepoUploadBlob) } if r.Header.Get("Content-Type") != mimeType { diff --git a/pkg/atproto/endpoints.go b/pkg/atproto/endpoints.go index 47ac082..4574b3b 100644 --- a/pkg/atproto/endpoints.go +++ b/pkg/atproto/endpoints.go @@ -70,11 +70,23 @@ const ( // Response: CAR file (application/vnd.ipld.car) SyncGetRepo = "/xrpc/com.atproto.sync.getRepo" + // SyncGetRecord retrieves a specific record as part of a repository sync. + // Method: GET + // Query: did={did}&collection={collection}&rkey={key} + // Response: Record data + SyncGetRecord = "/xrpc/com.atproto.sync.getRecord" + // SyncListRepos lists all repositories on a PDS. // Method: GET // Response: {"repos": [{...}]} SyncListRepos = "/xrpc/com.atproto.sync.listRepos" + // SyncListReposByCollection lists all repositories that have records in a specific collection. + // Method: GET + // Query: collection={collection}&limit={limit}&cursor={cursor} + // Response: {"repos": [{"did": "..."}], "cursor": "..."} + SyncListReposByCollection = "/xrpc/com.atproto.sync.listReposByCollection" + // SyncSubscribeRepos subscribes to real-time repository events via WebSocket. // Method: GET (WebSocket upgrade) // Response: Stream of #commit events @@ -101,6 +113,18 @@ const ( // Method: GET // Response: {"did": "...", "availableUserDomains": [...]} ServerDescribeServer = "/xrpc/com.atproto.server.describeServer" + + // ServerCreateSession creates a new session with identifier and password. + // Method: POST + // Request: {"identifier": "...", "password": "..."} + // Response: {"accessJwt": "...", "refreshJwt": "...", "did": "...", "handle": "..."} + ServerCreateSession = "/xrpc/com.atproto.server.createSession" + + // ServerGetSession validates a session and returns the current session info. + // Method: GET + // Headers: Authorization (Bearer or DPoP), DPoP (if using DPoP) + // Response: {"did": "...", "handle": "..."} + ServerGetSession = "/xrpc/com.atproto.server.getSession" ) // ATProto repo endpoints (com.atproto.repo.*) @@ -113,6 +137,24 @@ const ( // Response: {"did": "...", "handle": "...", "collections": [...]} RepoDescribeRepo = "/xrpc/com.atproto.repo.describeRepo" + // RepoPutRecord creates or updates a record in a repository. + // Method: POST + // Request: {"repo": "...", "collection": "...", "rkey": "...", "record": {...}} + // Response: {"uri": "...", "cid": "..."} + RepoPutRecord = "/xrpc/com.atproto.repo.putRecord" + + // RepoGetRecord retrieves a record from a repository. + // Method: GET + // Query: repo={did}&collection={collection}&rkey={key} + // Response: {"uri": "...", "cid": "...", "value": {...}} + RepoGetRecord = "/xrpc/com.atproto.repo.getRecord" + + // RepoListRecords lists records in a collection. + // Method: GET + // Query: repo={did}&collection={collection}&limit={limit}&cursor={cursor} + // Response: {"records": [...], "cursor": "..."} + RepoListRecords = "/xrpc/com.atproto.repo.listRecords" + // RepoDeleteRecord deletes a record from a repository. // Method: POST // Query: repo={did}&collection={collection}&rkey={key} diff --git a/pkg/atproto/manifest_store_test.go b/pkg/atproto/manifest_store_test.go index 0b1c713..188cb78 100644 --- a/pkg/atproto/manifest_store_test.go +++ b/pkg/atproto/manifest_store_test.go @@ -442,7 +442,7 @@ func TestExtractConfigLabels_NoLabels(t *testing.T) { } // Should return empty map (or nil) - if labels != nil && len(labels) != 0 { + if len(labels) != 0 { t.Errorf("extractConfigLabels() should return empty/nil for config without labels, got %v", labels) } } diff --git a/pkg/auth/hold_remote.go b/pkg/auth/hold_remote.go index 6f4f57e..ab87c14 100644 --- a/pkg/auth/hold_remote.go +++ b/pkg/auth/hold_remote.go @@ -201,8 +201,8 @@ func (a *RemoteHoldAuthorizer) fetchCaptainRecordFromXRPC(ctx context.Context, h // Build XRPC request URL // GET /xrpc/com.atproto.repo.getRecord?repo={did}&collection=io.atcr.hold.captain&rkey=self - xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", - holdURL, url.QueryEscape(holdDID), url.QueryEscape(atproto.CaptainCollection)) + xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&rkey=self", + holdURL, atproto.RepoGetRecord, url.QueryEscape(holdDID), url.QueryEscape(atproto.CaptainCollection)) req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) if err != nil { @@ -302,8 +302,8 @@ func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID, // Build XRPC request URL // GET /xrpc/com.atproto.repo.listRecords?repo={did}&collection=io.atcr.hold.crew - xrpcURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s", - holdURL, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection)) + xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s", + holdURL, atproto.RepoListRecords, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection)) req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) if err != nil { diff --git a/pkg/auth/oauth/refresher.go b/pkg/auth/oauth/refresher.go index bc57ba5..ec281f8 100644 --- a/pkg/auth/oauth/refresher.go +++ b/pkg/auth/oauth/refresher.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sync" + "time" "github.com/bluesky-social/indigo/atproto/auth/oauth" "github.com/bluesky-social/indigo/atproto/syntax" @@ -15,13 +16,21 @@ type SessionCache struct { SessionID string } +// UISessionStore interface for managing UI sessions +// Shared between refresher and server +type UISessionStore interface { + Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error) + DeleteByDID(did string) +} + // Refresher manages OAuth sessions and token refresh for AppView type Refresher struct { - app *App - sessions map[string]*SessionCache // Key: DID string - mu sync.RWMutex - refreshLocks map[string]*sync.Mutex // Per-DID locks for refresh operations - refreshLockMu sync.Mutex // Protects refreshLocks map + app *App + sessions map[string]*SessionCache // Key: DID string + mu sync.RWMutex + refreshLocks map[string]*sync.Mutex // Per-DID locks for refresh operations + refreshLockMu sync.Mutex // Protects refreshLocks map + uiSessionStore UISessionStore // For invalidating UI sessions on OAuth failures } // NewRefresher creates a new session refresher @@ -33,6 +42,11 @@ func NewRefresher(app *App) *Refresher { } } +// SetUISessionStore sets the UI session store for invalidating sessions on OAuth failures +func (r *Refresher) SetUISessionStore(store UISessionStore) { + r.uiSessionStore = store +} + // GetSession gets a fresh OAuth session for a DID // Returns cached session if still valid, otherwise resumes from store func (r *Refresher) GetSession(ctx context.Context, did string) (*oauth.ClientSession, error) { @@ -115,11 +129,17 @@ func (r *Refresher) resumeSession(ctx context.Context, did string) (*oauth.Clien } // InvalidateSession removes a cached session for a DID -// This is useful when a new OAuth flow creates a fresh session +// This is useful when a new OAuth flow creates a fresh session or when OAuth refresh fails +// Also invalidates any UI sessions for this DID to force re-authentication func (r *Refresher) InvalidateSession(did string) { r.mu.Lock() delete(r.sessions, did) r.mu.Unlock() + + // Also delete UI sessions to force user to re-authenticate + if r.uiSessionStore != nil { + r.uiSessionStore.DeleteByDID(did) + } } // GetSessionID returns the sessionID for a cached session diff --git a/pkg/auth/oauth/server.go b/pkg/auth/oauth/server.go index 4082934..bee6fc3 100644 --- a/pkg/auth/oauth/server.go +++ b/pkg/auth/oauth/server.go @@ -16,9 +16,7 @@ import ( ) // UISessionStore is the interface for UI session management -type UISessionStore interface { - Create(did, handle, pdsEndpoint string, duration time.Duration) (string, error) -} +// UISessionStore is defined in refresher.go to avoid duplication // UserStore is the interface for user management type UserStore interface { diff --git a/pkg/auth/atproto/session.go b/pkg/auth/session.go similarity index 98% rename from pkg/auth/atproto/session.go rename to pkg/auth/session.go index 40feedc..f5b2419 100644 --- a/pkg/auth/atproto/session.go +++ b/pkg/auth/session.go @@ -1,4 +1,4 @@ -package atproto +package auth import ( "bytes" @@ -11,6 +11,7 @@ import ( "net/http" "sync" "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" @@ -143,7 +144,7 @@ func (v *SessionValidator) createSession(ctx context.Context, pdsEndpoint, ident return nil, fmt.Errorf("failed to marshal request: %w", err) } - url := fmt.Sprintf("%s/xrpc/com.atproto.server.createSession", pdsEndpoint) + url := fmt.Sprintf("%s%s", pdsEndpoint, atproto.ServerCreateSession) fmt.Printf("DEBUG [atproto/session]: POST %s\n", url) req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewReader(body)) diff --git a/pkg/auth/token/cache.go b/pkg/auth/token/cache.go new file mode 100644 index 0000000..58e95f6 --- /dev/null +++ b/pkg/auth/token/cache.go @@ -0,0 +1,169 @@ +package token + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "strings" + "sync" + "time" +) + +// serviceTokenEntry represents a cached service token +type serviceTokenEntry struct { + token string + expiresAt time.Time +} + +// Global cache for service tokens (DID:HoldDID -> token) +// Service tokens are JWTs issued by a user's PDS to authorize AppView to act on their behalf +// when communicating with hold services. These tokens are scoped to specific holds and have +// limited lifetime (typically 60s, can request up to 5min). +var ( + globalServiceTokens = make(map[string]*serviceTokenEntry) + globalServiceTokensMu sync.RWMutex +) + +// GetServiceToken retrieves a cached service token for the given DID and hold DID +// Returns empty string if no valid cached token exists +func GetServiceToken(did, holdDID string) (token string, expiresAt time.Time) { + cacheKey := did + ":" + holdDID + + globalServiceTokensMu.RLock() + entry, exists := globalServiceTokens[cacheKey] + globalServiceTokensMu.RUnlock() + + if !exists { + return "", time.Time{} + } + + // Check if token is still valid + if time.Now().After(entry.expiresAt) { + // Token expired, remove from cache + globalServiceTokensMu.Lock() + delete(globalServiceTokens, cacheKey) + globalServiceTokensMu.Unlock() + return "", time.Time{} + } + + return entry.token, entry.expiresAt +} + +// SetServiceToken stores a service token in the cache +// Automatically parses the JWT to extract the expiry time +// Applies a 10-second safety margin (cache expires 10s before actual JWT expiry) +func SetServiceToken(did, holdDID, token string) error { + cacheKey := did + ":" + holdDID + + // Parse JWT to extract expiry (don't verify signature - we trust the PDS) + expiry, err := parseJWTExpiry(token) + if err != nil { + // If parsing fails, use default 50s TTL (conservative fallback) + fmt.Printf("WARN [token/cache]: Failed to parse JWT expiry, using default 50s: %v\n", err) + expiry = time.Now().Add(50 * time.Second) + } else { + // Apply 10s safety margin to avoid using nearly-expired tokens + expiry = expiry.Add(-10 * time.Second) + } + + globalServiceTokensMu.Lock() + globalServiceTokens[cacheKey] = &serviceTokenEntry{ + token: token, + expiresAt: expiry, + } + globalServiceTokensMu.Unlock() + + fmt.Printf("DEBUG [token/cache]: Cached service token for %s (expires in %v)\n", + cacheKey, time.Until(expiry).Round(time.Second)) + + return nil +} + +// parseJWTExpiry extracts the expiry time from a JWT without verifying the signature +// We trust tokens from the user's PDS, so signature verification isn't needed here +// Manually decodes the JWT payload to avoid algorithm compatibility issues +func parseJWTExpiry(tokenString string) (time.Time, error) { + // JWT format: header.payload.signature + parts := strings.Split(tokenString, ".") + if len(parts) != 3 { + return time.Time{}, fmt.Errorf("invalid JWT format: expected 3 parts, got %d", len(parts)) + } + + // Decode the payload (second part) + payload, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + return time.Time{}, fmt.Errorf("failed to decode JWT payload: %w", err) + } + + // Parse the JSON payload + var claims struct { + Exp int64 `json:"exp"` + } + if err := json.Unmarshal(payload, &claims); err != nil { + return time.Time{}, fmt.Errorf("failed to parse JWT claims: %w", err) + } + + if claims.Exp == 0 { + return time.Time{}, fmt.Errorf("JWT missing exp claim") + } + + return time.Unix(claims.Exp, 0), nil +} + +// InvalidateServiceToken removes a service token from the cache +// Used when we detect that a token is invalid or the user's session has expired +func InvalidateServiceToken(did, holdDID string) { + cacheKey := did + ":" + holdDID + + globalServiceTokensMu.Lock() + delete(globalServiceTokens, cacheKey) + globalServiceTokensMu.Unlock() + + fmt.Printf("DEBUG [token/cache]: Invalidated service token for %s\n", cacheKey) +} + +// GetCacheStats returns statistics about the service token cache for debugging +func GetCacheStats() map[string]interface{} { + globalServiceTokensMu.RLock() + defer globalServiceTokensMu.RUnlock() + + validCount := 0 + expiredCount := 0 + now := time.Now() + + for _, entry := range globalServiceTokens { + if now.Before(entry.expiresAt) { + validCount++ + } else { + expiredCount++ + } + } + + return map[string]interface{}{ + "total_entries": len(globalServiceTokens), + "valid_tokens": validCount, + "expired_tokens": expiredCount, + } +} + +// CleanExpiredTokens removes expired tokens from the cache +// Can be called periodically to prevent unbounded growth (though expired tokens +// are also removed lazily on access) +func CleanExpiredTokens() { + globalServiceTokensMu.Lock() + defer globalServiceTokensMu.Unlock() + + now := time.Now() + removed := 0 + + for key, entry := range globalServiceTokens { + if now.After(entry.expiresAt) { + delete(globalServiceTokens, key) + removed++ + } + } + + if removed > 0 { + fmt.Printf("DEBUG [token/cache]: Cleaned %d expired service tokens\n", removed) + } +} diff --git a/pkg/auth/token/handler.go b/pkg/auth/token/handler.go index 736b63b..5fdaf18 100644 --- a/pkg/auth/token/handler.go +++ b/pkg/auth/token/handler.go @@ -13,13 +13,12 @@ import ( "atcr.io/pkg/appview/db" mainAtproto "atcr.io/pkg/atproto" "atcr.io/pkg/auth" - "atcr.io/pkg/auth/atproto" ) // Handler handles /auth/token requests type Handler struct { issuer *Issuer - validator *atproto.SessionValidator + validator *auth.SessionValidator deviceStore *db.DeviceStore // For validating device secrets defaultHoldDID string } @@ -30,7 +29,7 @@ type Handler struct { func NewHandler(issuer *Issuer, deviceStore *db.DeviceStore, defaultHoldDID string) *Handler { return &Handler{ issuer: issuer, - validator: atproto.NewSessionValidator(), + validator: auth.NewSessionValidator(), deviceStore: deviceStore, defaultHoldDID: defaultHoldDID, } diff --git a/pkg/hold/oci/xrpc_test.go b/pkg/hold/oci/xrpc_test.go index 7f0d318..42b7c99 100644 --- a/pkg/hold/oci/xrpc_test.go +++ b/pkg/hold/oci/xrpc_test.go @@ -12,6 +12,7 @@ import ( "testing" "atcr.io/pkg/hold/pds" + "atcr.io/pkg/atproto" "atcr.io/pkg/s3" "github.com/distribution/distribution/v3/registry/storage/driver/factory" _ "github.com/distribution/distribution/v3/registry/storage/driver/filesystem" @@ -109,7 +110,7 @@ func decodeJSONResponse(t *testing.T, w *httptest.ResponseRecorder, v any) { func TestHandleInitiateUpload_Success(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{ + req := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{ "digest": "sha256:abc123", }) addMockAuth(req) @@ -133,7 +134,7 @@ func TestHandleInitiateUpload_Success(t *testing.T) { func TestHandleInitiateUpload_MissingDigest(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{}) + req := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{}) addMockAuth(req) w := httptest.NewRecorder() @@ -154,7 +155,7 @@ func TestHandleGetPartUploadUrl_Buffered(t *testing.T) { handler, _ := setupTestOCIHandler(t) // First, initiate an upload - initReq := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{ + initReq := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{ "digest": "sha256:abc123", }) addMockAuth(initReq) @@ -166,7 +167,7 @@ func TestHandleGetPartUploadUrl_Buffered(t *testing.T) { uploadID := initResp["uploadId"].(string) // Now get part upload URL - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.getPartUploadUrl", map[string]any{ + req := makeJSONRequest("POST", atproto.HoldGetPartUploadUrl, map[string]any{ "uploadId": uploadID, "partNumber": 1, }) @@ -194,7 +195,7 @@ func TestHandleGetPartUploadUrl_Buffered(t *testing.T) { func TestHandleGetPartUploadUrl_InvalidSession(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.getPartUploadUrl", map[string]any{ + req := makeJSONRequest("POST", atproto.HoldGetPartUploadUrl, map[string]any{ "uploadId": "invalid-upload-id", "partNumber": 1, }) @@ -222,7 +223,7 @@ func TestHandleGetPartUploadUrl_MissingParams(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.getPartUploadUrl", tt.body) + req := makeJSONRequest("POST", atproto.HoldGetPartUploadUrl, tt.body) addMockAuth(req) w := httptest.NewRecorder() @@ -241,7 +242,7 @@ func TestHandleUploadPart_Success(t *testing.T) { handler, _ := setupTestOCIHandler(t) // Initiate upload - initReq := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{ + initReq := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{ "digest": "sha256:abc123", }) addMockAuth(initReq) @@ -254,7 +255,7 @@ func TestHandleUploadPart_Success(t *testing.T) { // Upload a part partData := []byte("test part data") - req := httptest.NewRequest("PUT", "/xrpc/io.atcr.hold.uploadPart", bytes.NewReader(partData)) + req := httptest.NewRequest("PUT", atproto.HoldUploadPart, bytes.NewReader(partData)) req.Header.Set("X-Upload-Id", uploadID) req.Header.Set("X-Part-Number", "1") addMockAuth(req) @@ -292,7 +293,7 @@ func TestHandleUploadPart_MissingHeaders(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := httptest.NewRequest("PUT", "/xrpc/io.atcr.hold.uploadPart", bytes.NewReader([]byte("data"))) + req := httptest.NewRequest("PUT", atproto.HoldUploadPart, bytes.NewReader([]byte("data"))) if tt.uploadID != "" { req.Header.Set("X-Upload-Id", tt.uploadID) } @@ -314,7 +315,7 @@ func TestHandleUploadPart_MissingHeaders(t *testing.T) { func TestHandleUploadPart_InvalidPartNumber(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := httptest.NewRequest("PUT", "/xrpc/io.atcr.hold.uploadPart", bytes.NewReader([]byte("data"))) + req := httptest.NewRequest("PUT", atproto.HoldUploadPart, bytes.NewReader([]byte("data"))) req.Header.Set("X-Upload-Id", "test-id") req.Header.Set("X-Part-Number", "not-a-number") addMockAuth(req) @@ -333,7 +334,7 @@ func TestHandleCompleteUpload_BufferedMode(t *testing.T) { handler, _ := setupTestOCIHandler(t) // Initiate upload - initReq := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{ + initReq := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{ "digest": "sha256:abc123", }) addMockAuth(initReq) @@ -355,7 +356,7 @@ func TestHandleCompleteUpload_BufferedMode(t *testing.T) { var partInfos []map[string]any for _, p := range parts { - req := httptest.NewRequest("PUT", "/xrpc/io.atcr.hold.uploadPart", bytes.NewReader([]byte(p.data))) + req := httptest.NewRequest("PUT", atproto.HoldUploadPart, bytes.NewReader([]byte(p.data))) req.Header.Set("X-Upload-Id", uploadID) req.Header.Set("X-Part-Number", strconv.Itoa(p.number)) addMockAuth(req) @@ -373,7 +374,7 @@ func TestHandleCompleteUpload_BufferedMode(t *testing.T) { } // Complete upload - completeReq := makeJSONRequest("POST", "/xrpc/io.atcr.hold.completeUpload", map[string]any{ + completeReq := makeJSONRequest("POST", atproto.HoldCompleteUpload, map[string]any{ "uploadId": uploadID, "digest": "sha256:finaldigest123", "parts": partInfos, @@ -401,7 +402,7 @@ func TestHandleCompleteUpload_BufferedMode(t *testing.T) { func TestHandleCompleteUpload_MissingParts(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.completeUpload", map[string]any{ + req := makeJSONRequest("POST", atproto.HoldCompleteUpload, map[string]any{ "uploadId": "test-id", "digest": "sha256:test", "parts": []any{}, @@ -419,7 +420,7 @@ func TestHandleCompleteUpload_MissingParts(t *testing.T) { func TestHandleCompleteUpload_InvalidSession(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.completeUpload", map[string]any{ + req := makeJSONRequest("POST", atproto.HoldCompleteUpload, map[string]any{ "uploadId": "invalid-upload-id", "digest": "sha256:test", "parts": []any{ @@ -442,7 +443,7 @@ func TestHandleAbortUpload_Success(t *testing.T) { handler, _ := setupTestOCIHandler(t) // Initiate upload - initReq := makeJSONRequest("POST", "/xrpc/io.atcr.hold.initiateUpload", map[string]string{ + initReq := makeJSONRequest("POST", atproto.HoldInitiateUpload, map[string]string{ "digest": "sha256:abc123", }) addMockAuth(initReq) @@ -454,7 +455,7 @@ func TestHandleAbortUpload_Success(t *testing.T) { uploadID := initResp["uploadId"].(string) // Abort upload - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.abortUpload", map[string]string{ + req := makeJSONRequest("POST", atproto.HoldAbortUpload, map[string]string{ "uploadId": uploadID, }) addMockAuth(req) @@ -477,7 +478,7 @@ func TestHandleAbortUpload_Success(t *testing.T) { func TestHandleAbortUpload_InvalidSession(t *testing.T) { handler, _ := setupTestOCIHandler(t) - req := makeJSONRequest("POST", "/xrpc/io.atcr.hold.abortUpload", map[string]string{ + req := makeJSONRequest("POST", atproto.HoldAbortUpload, map[string]string{ "uploadId": "invalid-upload-id", }) addMockAuth(req) diff --git a/pkg/hold/pds/auth.go b/pkg/hold/pds/auth.go index b02fcc6..a6d9889 100644 --- a/pkg/hold/pds/auth.go +++ b/pkg/hold/pds/auth.go @@ -11,6 +11,7 @@ import ( "strings" "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" @@ -154,7 +155,7 @@ type SessionResponse struct { // The httpClient parameter is optional and defaults to http.DefaultClient if nil. func validateTokenWithPDS(ctx context.Context, pdsURL, accessToken, dpopProof string, httpClient HTTPClient) (*SessionResponse, error) { // Call com.atproto.server.getSession with DPoP headers - url := fmt.Sprintf("%s/xrpc/com.atproto.server.getSession", strings.TrimSuffix(pdsURL, "/")) + url := fmt.Sprintf("%s%s", strings.TrimSuffix(pdsURL, "/"), atproto.ServerGetSession) req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { diff --git a/pkg/hold/pds/auth_test.go b/pkg/hold/pds/auth_test.go index cf67974..1eb05e8 100644 --- a/pkg/hold/pds/auth_test.go +++ b/pkg/hold/pds/auth_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/atproto/atcrypto" "github.com/bluesky-social/indigo/atproto/auth/oauth" ) @@ -24,7 +25,7 @@ type mockPDSClient struct{} func (m *mockPDSClient) Do(req *http.Request) (*http.Response, error) { // Verify request is for getSession endpoint - if !strings.Contains(req.URL.Path, "/xrpc/com.atproto.server.getSession") { + if !strings.Contains(req.URL.Path, atproto.ServerGetSession) { return &http.Response{ StatusCode: http.StatusNotFound, Body: http.NoBody, diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 74f9778..2f7ab82 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -138,18 +138,18 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { // Health and server info r.Get("/xrpc/_health", h.HandleHealth) - r.Get("/xrpc/com.atproto.server.describeServer", h.HandleDescribeServer) + r.Get(atproto.ServerDescribeServer, h.HandleDescribeServer) // Repository metadata - r.Get("/xrpc/com.atproto.repo.describeRepo", h.HandleDescribeRepo) - r.Get("/xrpc/com.atproto.repo.getRecord", h.HandleGetRecord) - r.Get("/xrpc/com.atproto.repo.listRecords", h.HandleListRecords) + r.Get(atproto.RepoDescribeRepo, h.HandleDescribeRepo) + r.Get(atproto.RepoGetRecord, h.HandleGetRecord) + r.Get(atproto.RepoListRecords, h.HandleListRecords) // Sync endpoints - r.Get("/xrpc/com.atproto.sync.listRepos", h.HandleListRepos) - r.Get("/xrpc/com.atproto.sync.getRecord", h.HandleSyncGetRecord) - r.Get("/xrpc/com.atproto.sync.getRepo", h.HandleGetRepo) - r.Get("/xrpc/com.atproto.sync.subscribeRepos", h.HandleSubscribeRepos) + r.Get(atproto.SyncListRepos, h.HandleListRepos) + r.Get(atproto.SyncGetRecord, h.HandleSyncGetRecord) + r.Get(atproto.SyncGetRepo, h.HandleGetRepo) + r.Get(atproto.SyncSubscribeRepos, h.HandleSubscribeRepos) // DID document and handle resolution r.Get("/.well-known/did.json", h.HandleDIDDocument) @@ -161,8 +161,8 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { r.Group(func(r chi.Router) { r.Use(h.corsMiddleware) - r.Get("/xrpc/com.atproto.sync.getBlob", h.HandleGetBlob) - r.Head("/xrpc/com.atproto.sync.getBlob", h.HandleGetBlob) + r.Get(atproto.SyncGetBlob, h.HandleGetBlob) + r.Head(atproto.SyncGetBlob, h.HandleGetBlob) }) // Write endpoints (CORS + owner/crew admin auth) @@ -170,8 +170,8 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { r.Use(h.corsMiddleware) r.Use(h.requireOwnerOrCrewAdmin) - r.Post("/xrpc/com.atproto.repo.deleteRecord", h.HandleDeleteRecord) - r.Post("/xrpc/com.atproto.repo.uploadBlob", h.HandleUploadBlob) + r.Post(atproto.RepoDeleteRecord, h.HandleDeleteRecord) + r.Post(atproto.RepoUploadBlob, h.HandleUploadBlob) }) // Auth-only endpoints (CORS + DPoP auth) @@ -179,7 +179,7 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { r.Use(h.corsMiddleware) r.Use(h.requireAuth) - r.Post("/xrpc/io.atcr.hold.requestCrew", h.HandleRequestCrew) + r.Post(atproto.HoldRequestCrew, h.HandleRequestCrew) }) } @@ -1136,8 +1136,8 @@ func getProxyURL(publicURL string, digest, did string, operation string) string if operation == http.MethodGet || operation == http.MethodHead { // Generate hold DID from public URL using shared function holdDID := atproto.ResolveHoldDIDFromURL(publicURL) - return fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", - publicURL, holdDID, digest) + return fmt.Sprintf("%s%s?did=%s&cid=%s", + publicURL, atproto.SyncGetBlob, holdDID, digest) } // For PUT operations, proxy fallback is not supported with XRPC diff --git a/pkg/hold/pds/xrpc_test.go b/pkg/hold/pds/xrpc_test.go index 0947e7b..7df61a4 100644 --- a/pkg/hold/pds/xrpc_test.go +++ b/pkg/hold/pds/xrpc_test.go @@ -172,7 +172,7 @@ func TestHandleHealth_MethodNotAllowed(t *testing.T) { func TestHandleDescribeServer(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.server.describeServer", nil) + req := makeXRPCGetRequest(atproto.ServerDescribeServer, nil) w := httptest.NewRecorder() handler.HandleDescribeServer(w, req) @@ -210,7 +210,7 @@ func TestHandleDescribeRepo(t *testing.T) { handler, _ := setupTestXRPCHandler(t) holdDID := "did:web:hold.example.com" - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.describeRepo", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoDescribeRepo, map[string]string{ "repo": holdDID, }) w := httptest.NewRecorder() @@ -249,7 +249,7 @@ func TestHandleDescribeRepo(t *testing.T) { func TestHandleDescribeRepo_MissingRepo(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.describeRepo", nil) + req := makeXRPCGetRequest(atproto.RepoDescribeRepo, nil) w := httptest.NewRecorder() handler.HandleDescribeRepo(w, req) @@ -263,7 +263,7 @@ func TestHandleDescribeRepo_MissingRepo(t *testing.T) { func TestHandleDescribeRepo_InvalidRepo(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.describeRepo", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoDescribeRepo, map[string]string{ "repo": "did:plc:wrongdid", }) w := httptest.NewRecorder() @@ -284,7 +284,7 @@ func TestHandleGetRecord(t *testing.T) { holdDID := "did:web:hold.example.com" // Get the captain record that was created during bootstrap - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoGetRecord, map[string]string{ "repo": holdDID, "collection": atproto.CaptainCollection, "rkey": CaptainRkey, @@ -330,7 +330,7 @@ func TestHandleGetRecord(t *testing.T) { crewRkey := crew[0].Rkey - req = makeXRPCGetRequest("/xrpc/com.atproto.repo.getRecord", map[string]string{ + req = makeXRPCGetRequest(atproto.RepoGetRecord, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, "rkey": crewRkey, @@ -379,7 +379,7 @@ func TestHandleGetRecord_MissingParameters(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.getRecord", tt.params) + req := makeXRPCGetRequest(atproto.RepoGetRecord, tt.params) w := httptest.NewRecorder() handler.HandleGetRecord(w, req) @@ -396,7 +396,7 @@ func TestHandleGetRecord_RecordNotFound(t *testing.T) { handler, _ := setupTestXRPCHandler(t) holdDID := "did:web:hold.example.com" - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoGetRecord, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, "rkey": "nonexistent", @@ -414,7 +414,7 @@ func TestHandleGetRecord_RecordNotFound(t *testing.T) { func TestHandleGetRecord_InvalidRepo(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoGetRecord, map[string]string{ "repo": "did:plc:wrongdid", "collection": atproto.CaptainCollection, "rkey": CaptainRkey, @@ -452,7 +452,7 @@ func TestHandleListRecords(t *testing.T) { } // Test listing crew records - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, }) @@ -512,7 +512,7 @@ func TestHandleListRecords_Pagination(t *testing.T) { } // Test with limit=2 - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, "limit": "2", @@ -538,7 +538,7 @@ func TestHandleListRecords_Pagination(t *testing.T) { t.Error("Expected cursor in response when there are more records") } else { // Test pagination with cursor - req2 := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req2 := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, "limit": "2", @@ -576,7 +576,7 @@ func TestHandleListRecords_Reverse(t *testing.T) { } // Get normal order - req1 := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req1 := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, }) @@ -586,7 +586,7 @@ func TestHandleListRecords_Reverse(t *testing.T) { records1 := result1["records"].([]any) // Get reverse order - req2 := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req2 := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": holdDID, "collection": atproto.CrewCollection, "reverse": "true", @@ -627,7 +627,7 @@ func TestHandleListRecords_InvalidLimit(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": "did:web:hold.example.com", "collection": atproto.CrewCollection, "limit": tt.limit, @@ -657,7 +657,7 @@ func TestHandleListRecords_EmptyCollection(t *testing.T) { } // Query a collection that has no records yet - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", map[string]string{ + req := makeXRPCGetRequest(atproto.RepoListRecords, map[string]string{ "repo": "did:web:hold.example.com", "collection": atproto.CrewCollection, }) @@ -698,7 +698,7 @@ func TestHandleListRecords_MissingParameters(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCGetRequest("/xrpc/com.atproto.repo.listRecords", tt.params) + req := makeXRPCGetRequest(atproto.RepoListRecords, tt.params) w := httptest.NewRecorder() handler.HandleListRecords(w, req) @@ -739,7 +739,7 @@ func TestHandleDeleteRecord(t *testing.T) { "rkey": rkey, } - req := makeXRPCPostRequest("/xrpc/com.atproto.repo.deleteRecord", body) + req := makeXRPCPostRequest(atproto.RepoDeleteRecord, body) // Add DPoP authentication - owner has admin permission to delete crew ownerDID := "did:plc:testowner123" @@ -780,7 +780,7 @@ func TestHandleDeleteRecord(t *testing.T) { func TestHandleDeleteRecord_InvalidJSON(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.deleteRecord", bytes.NewReader([]byte("invalid json"))) + req := httptest.NewRequest(http.MethodPost, atproto.RepoDeleteRecord, bytes.NewReader([]byte("invalid json"))) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() @@ -821,7 +821,7 @@ func TestHandleDeleteRecord_MissingParameters(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCPostRequest("/xrpc/com.atproto.repo.deleteRecord", tt.body) + req := makeXRPCPostRequest(atproto.RepoDeleteRecord, tt.body) w := httptest.NewRecorder() handler.HandleDeleteRecord(w, req) @@ -848,7 +848,7 @@ func TestHandleListRepos(t *testing.T) { handler, _ := setupTestXRPCHandler(t) holdDID := "did:web:hold.example.com" - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.listRepos", nil) + req := makeXRPCGetRequest(atproto.SyncListRepos, nil) w := httptest.NewRecorder() handler.HandleListRepos(w, req) @@ -902,7 +902,7 @@ func TestHandleListRepos_EmptyRepo(t *testing.T) { // setupTestPDS creates the PDS/database but doesn't initialize the repo // Check if implementation returns repos before initialization - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.listRepos", nil) + req := makeXRPCGetRequest(atproto.SyncListRepos, nil) w := httptest.NewRecorder() handler.HandleListRepos(w, req) @@ -922,7 +922,7 @@ func TestHandleListRepos_EmptyRepo(t *testing.T) { t.Fatalf("Failed to initialize repo: %v", err) } - req = makeXRPCGetRequest("/xrpc/com.atproto.sync.listRepos", nil) + req = makeXRPCGetRequest(atproto.SyncListRepos, nil) w = httptest.NewRecorder() handler.HandleListRepos(w, req) @@ -953,7 +953,7 @@ func TestHandleSyncGetRecord(t *testing.T) { holdDID := "did:web:hold.example.com" // Get the captain record as CAR file - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetRecord, map[string]string{ "did": holdDID, "collection": atproto.CaptainCollection, "rkey": CaptainRkey, @@ -1001,7 +1001,7 @@ func TestHandleSyncGetRecord_MissingParameters(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRecord", tt.params) + req := makeXRPCGetRequest(atproto.SyncGetRecord, tt.params) w := httptest.NewRecorder() handler.HandleSyncGetRecord(w, req) @@ -1018,7 +1018,7 @@ func TestHandleSyncGetRecord_MissingParameters(t *testing.T) { func TestHandleSyncGetRecord_RecordNotFound(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetRecord, map[string]string{ "did": "did:web:hold.example.com", "collection": atproto.CrewCollection, "rkey": "nonexistent", @@ -1037,7 +1037,7 @@ func TestHandleSyncGetRecord_RecordNotFound(t *testing.T) { func TestHandleSyncGetRecord_InvalidDID(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRecord", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetRecord, map[string]string{ "did": "did:plc:wrongdid", "collection": atproto.CaptainCollection, "rkey": CaptainRkey, @@ -1060,7 +1060,7 @@ func TestHandleGetRepo(t *testing.T) { holdDID := "did:web:hold.example.com" // Get full repo as CAR file - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRepo", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetRepo, map[string]string{ "did": holdDID, }) w := httptest.NewRecorder() @@ -1081,7 +1081,7 @@ func TestHandleGetRepo(t *testing.T) { func TestHandleGetRepo_MissingDID(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRepo", nil) + req := makeXRPCGetRequest(atproto.SyncGetRepo, nil) w := httptest.NewRecorder() handler.HandleGetRepo(w, req) @@ -1096,7 +1096,7 @@ func TestHandleGetRepo_MissingDID(t *testing.T) { func TestHandleGetRepo_InvalidDID(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRepo", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetRepo, map[string]string{ "did": "did:plc:wrongdid", }) w := httptest.NewRecorder() @@ -1115,7 +1115,7 @@ func TestHandleGetRepo_WithSince(t *testing.T) { holdDID := "did:web:hold.example.com" // Get current rev to use as 'since' - req1 := makeXRPCGetRequest("/xrpc/com.atproto.sync.listRepos", nil) + req1 := makeXRPCGetRequest(atproto.SyncListRepos, nil) w1 := httptest.NewRecorder() handler.HandleListRepos(w1, req1) result := assertJSONResponse(t, w1, http.StatusOK) @@ -1132,7 +1132,7 @@ func TestHandleGetRepo_WithSince(t *testing.T) { } // Get repo diff since that rev - req2 := makeXRPCGetRequest("/xrpc/com.atproto.sync.getRepo", map[string]string{ + req2 := makeXRPCGetRequest(atproto.SyncGetRepo, map[string]string{ "did": holdDID, "since": rev, }) @@ -1173,7 +1173,7 @@ func TestHandleRequestCrew(t *testing.T) { "permissions": []string{"blob:read"}, } - req := makeXRPCPostRequest("/xrpc/io.atcr.hold.requestCrew", body) + req := makeXRPCPostRequest(atproto.HoldRequestCrew, body) w := httptest.NewRecorder() // Note: This will fail auth because we're not providing DPoP tokens @@ -1216,7 +1216,7 @@ func TestHandleRequestCrew_AllowAllCrewDisabled(t *testing.T) { "permissions": []string{"blob:read"}, } - req := makeXRPCPostRequest("/xrpc/io.atcr.hold.requestCrew", body) + req := makeXRPCPostRequest(atproto.HoldRequestCrew, body) w := httptest.NewRecorder() handler.HandleRequestCrew(w, req) @@ -1232,7 +1232,7 @@ func TestHandleRequestCrew_AllowAllCrewDisabled(t *testing.T) { func TestHandleRequestCrew_InvalidJSON(t *testing.T) { handler, _ := setupTestXRPCHandler(t) - req := httptest.NewRequest(http.MethodPost, "/xrpc/io.atcr.hold.requestCrew", bytes.NewReader([]byte("invalid json"))) + req := httptest.NewRequest(http.MethodPost, atproto.HoldRequestCrew, bytes.NewReader([]byte("invalid json"))) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() @@ -1391,7 +1391,7 @@ func TestHandleUploadBlob(t *testing.T) { blobData := []byte("Hello, ATProto!") // Test standard single blob upload (POST with raw bytes) - req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader(blobData)) + req := httptest.NewRequest(http.MethodPost, atproto.RepoUploadBlob, bytes.NewReader(blobData)) req.Header.Set("Content-Type", "application/octet-stream") // Add DPoP authentication - owner has admin permission for blob upload @@ -1447,7 +1447,7 @@ func TestHandleUploadBlob_EmptyBody(t *testing.T) { handler, _, _ := setupTestXRPCHandlerWithBlobs(t) // Empty blob should succeed (edge case) - req := httptest.NewRequest(http.MethodPost, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte{})) + req := httptest.NewRequest(http.MethodPost, atproto.RepoUploadBlob, bytes.NewReader([]byte{})) req.Header.Set("Content-Type", "application/octet-stream") // Add DPoP authentication @@ -1481,7 +1481,7 @@ func TestHandleUploadBlob_MethodNotAllowed(t *testing.T) { handler, _, _ := setupTestXRPCHandlerWithBlobs(t) // GET is not allowed for upload (only POST) - req := httptest.NewRequest(http.MethodGet, "/xrpc/com.atproto.repo.uploadBlob", bytes.NewReader([]byte("test"))) + req := httptest.NewRequest(http.MethodGet, atproto.RepoUploadBlob, bytes.NewReader([]byte("test"))) w := httptest.NewRecorder() handler.HandleUploadBlob(w, req) @@ -1509,7 +1509,7 @@ func TestHandleGetBlob(t *testing.T) { holdDID := "did:web:hold.example.com" cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": holdDID, "cid": cid, }) @@ -1548,7 +1548,7 @@ func TestHandleGetBlob_SHA256Digest(t *testing.T) { holdDID := "did:web:hold.example.com" digest := "sha256:abc123def456" // OCI digest format - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": holdDID, "cid": digest, }) @@ -1584,7 +1584,7 @@ func TestHandleGetBlob_HeadMethod(t *testing.T) { cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" // Use HEAD instead of GET - req := httptest.NewRequest(http.MethodHead, "/xrpc/com.atproto.sync.getBlob?did="+holdDID+"&cid="+cid, nil) + req := httptest.NewRequest(http.MethodHead, atproto.SyncGetBlob+"?did="+holdDID+"&cid="+cid, nil) w := httptest.NewRecorder() handler.HandleGetBlob(w, req) @@ -1641,7 +1641,7 @@ func TestHandleGetBlob_MissingParameters(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", tt.params) + req := makeXRPCGetRequest(atproto.SyncGetBlob, tt.params) w := httptest.NewRecorder() handler.HandleGetBlob(w, req) @@ -1658,7 +1658,7 @@ func TestHandleGetBlob_MissingParameters(t *testing.T) { func TestHandleGetBlob_InvalidDID(t *testing.T) { handler, _, _ := setupTestXRPCHandlerWithBlobs(t) - req := makeXRPCGetRequest("/xrpc/com.atproto.sync.getBlob", map[string]string{ + req := makeXRPCGetRequest(atproto.SyncGetBlob, map[string]string{ "did": "did:plc:wrongdid", "cid": "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke", }) @@ -1690,7 +1690,7 @@ func TestHandleGetBlob_CORSHeaders(t *testing.T) { holdDID := "did:web:hold.example.com" cid := "bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke" - url := fmt.Sprintf("/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", holdDID, cid) + url := fmt.Sprintf("%s?did=%s&cid=%s", atproto.SyncGetBlob, holdDID, cid) // Test GET request req := httptest.NewRequest(http.MethodGet, url, nil) @@ -1738,8 +1738,8 @@ func TestCORSMiddleware(t *testing.T) { method string }{ {"health endpoint", "/xrpc/_health", "GET"}, - {"describe server", "/xrpc/com.atproto.server.describeServer", "GET"}, - {"get blob", "/xrpc/com.atproto.sync.getBlob?did=test&cid=test", "GET"}, + {"describe server", atproto.ServerDescribeServer, "GET"}, + {"get blob", atproto.SyncGetBlob + "?did=test&cid=test", "GET"}, } for _, tt := range tests { @@ -1783,7 +1783,7 @@ func TestRequireOwnerOrCrewAdmin_Authorized(t *testing.T) { } body, _ := json.Marshal(input) - req := httptest.NewRequest("POST", "/xrpc/com.atproto.repo.deleteRecord", bytes.NewReader(body)) + req := httptest.NewRequest("POST", atproto.RepoDeleteRecord, bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") if err := dpopHelper.AddDPoPToRequest(req); err != nil { @@ -1819,7 +1819,7 @@ func TestRequireOwnerOrCrewAdmin_Unauthorized(t *testing.T) { } body, _ := json.Marshal(input) - req := httptest.NewRequest("POST", "/xrpc/com.atproto.repo.deleteRecord", bytes.NewReader(body)) + req := httptest.NewRequest("POST", atproto.RepoDeleteRecord, bytes.NewReader(body)) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() @@ -1844,7 +1844,7 @@ func TestRequireAuth_ValidDPoP(t *testing.T) { t.Fatalf("Failed to create DPoP helper: %v", err) } - req := httptest.NewRequest("POST", "/xrpc/io.atcr.hold.requestCrew", bytes.NewReader([]byte("{}"))) + req := httptest.NewRequest("POST", atproto.HoldRequestCrew, bytes.NewReader([]byte("{}"))) req.Header.Set("Content-Type", "application/json") if err := dpopHelper.AddDPoPToRequest(req); err != nil { @@ -1868,7 +1868,7 @@ func TestRequireAuth_MissingAuth(t *testing.T) { handler.RegisterHandlers(r) // requestCrew requires auth, but we send no auth - req := httptest.NewRequest("POST", "/xrpc/io.atcr.hold.requestCrew", bytes.NewReader([]byte("{}"))) + req := httptest.NewRequest("POST", atproto.HoldRequestCrew, bytes.NewReader([]byte("{}"))) req.Header.Set("Content-Type", "application/json") w := httptest.NewRecorder() @@ -1893,9 +1893,9 @@ func TestPublicRoutes_NoAuthRequired(t *testing.T) { method string }{ {"health check", "/xrpc/_health", "GET"}, - {"describe server", "/xrpc/com.atproto.server.describeServer", "GET"}, - {"describe repo", "/xrpc/com.atproto.repo.describeRepo?repo=" + handler.pds.DID(), "GET"}, - {"list repos", "/xrpc/com.atproto.sync.listRepos", "GET"}, + {"describe server", atproto.ServerDescribeServer, "GET"}, + {"describe repo", atproto.RepoDescribeRepo + "?repo=" + handler.pds.DID(), "GET"}, + {"list repos", atproto.SyncListRepos, "GET"}, {"did document", "/.well-known/did.json", "GET"}, {"atproto did", "/.well-known/atproto-did", "GET"}, } @@ -1923,7 +1923,7 @@ func TestBlobReadRoutes_ConditionalAuth(t *testing.T) { handler.RegisterHandlers(r) // getBlob should work without auth if captain.public = true - req := httptest.NewRequest("GET", "/xrpc/com.atproto.sync.getBlob?did="+handler.pds.DID()+"&cid=test123", nil) + req := httptest.NewRequest("GET", atproto.SyncGetBlob+"?did="+handler.pds.DID()+"&cid=test123", nil) w := httptest.NewRecorder() r.ServeHTTP(w, req) @@ -1946,8 +1946,8 @@ func TestWriteRoutes_RequireAdmin(t *testing.T) { path string body string }{ - {"delete record", "/xrpc/com.atproto.repo.deleteRecord", `{"repo":"test","collection":"test","rkey":"test"}`}, - {"upload blob", "/xrpc/com.atproto.repo.uploadBlob", "blob data"}, + {"delete record", atproto.RepoDeleteRecord, `{"repo":"test","collection":"test","rkey":"test"}`}, + {"upload blob", atproto.RepoUploadBlob, "blob data"}, } for _, tt := range tests { @@ -1976,8 +1976,8 @@ func TestRouteMethodEnforcement_GET(t *testing.T) { // GET-only routes should reject POST tests := []string{ "/xrpc/_health", - "/xrpc/com.atproto.server.describeServer", - "/xrpc/com.atproto.sync.listRepos", + atproto.ServerDescribeServer, + atproto.SyncListRepos, } for _, path := range tests { @@ -2003,8 +2003,8 @@ func TestRouteMethodEnforcement_POST(t *testing.T) { // POST-only routes should reject GET tests := []string{ - "/xrpc/com.atproto.repo.deleteRecord", - "/xrpc/io.atcr.hold.requestCrew", + atproto.RepoDeleteRecord, + atproto.HoldRequestCrew, } for _, path := range tests { diff --git a/test-e2e.sh b/test-e2e.sh new file mode 100755 index 0000000..a6662d2 --- /dev/null +++ b/test-e2e.sh @@ -0,0 +1,432 @@ +#!/bin/bash +# +# ATCR End-to-End Test Script +# Tests single-arch and multi-arch image push/pull flows +# +# Usage: +# ./test-e2e.sh # Run full test suite +# ./test-e2e.sh --multi # Skip to multi-arch test only +# REGISTRY=localhost:5000 ./test-e2e.sh # Custom registry +# NAMESPACE=myuser ./test-e2e.sh # Custom namespace +# VERBOSE=0 ./test-e2e.sh # Hide docker output +# +# To see bash command execution, edit this file and uncomment 'set -x' below +# + +set -e + +# Verbose mode - shows docker command output +# Set VERBOSE=0 to hide docker output: VERBOSE=0 ./test-e2e.sh +VERBOSE="${VERBOSE:-1}" + +# Bash trace mode - shows every bash command as it executes +# Uncomment the line below to see bash commands as they execute +# set -x + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +NC='\033[0m' # No Color + +# Configuration +REGISTRY="${REGISTRY:-127.0.0.1:5000}" +NAMESPACE="${NAMESPACE:-evan.jarrett.net}" +IMAGE_NAME="test-image" +TAG="latest" +MULTI_ARCH_TAG="multiarch" + +# Full image references +SINGLE_ARCH_IMAGE="${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}:${TAG}" +MULTI_ARCH_IMAGE="${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}:${MULTI_ARCH_TAG}" +AMD64_IMAGE="${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}:${TAG}-amd64" +ARM64_IMAGE="${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}:${TAG}-arm64" + +# Temporary directory for test images +BUILD_DIR=$(mktemp -d) + +# Cleanup function +cleanup() { + log_info "Cleaning up..." + rm -rf ${BUILD_DIR} + + # Clean up any dangling manifest lists + docker manifest rm ${MULTI_ARCH_IMAGE} 2>/dev/null || true +} + +trap cleanup EXIT + +# Logging functions +log_info() { + echo -e "${GREEN}[INFO]${NC} $1" +} + +log_error() { + echo -e "${RED}[ERROR]${NC} $1" +} + +log_warn() { + echo -e "${YELLOW}[WARN]${NC} $1" +} + +log_step() { + echo -e "\n${GREEN}==>${NC} $1" +} + +log_cmd() { + echo -e "${BLUE}[CMD]${NC} $*" + if [ "$VERBOSE" = "1" ]; then + "$@" + else + "$@" > /dev/null 2>&1 + fi +} + +# Check if docker compose services are running +check_compose_services() { + log_step "Checking docker compose services..." + + if ! docker compose ps | grep -q "Up"; then + log_warn "Some services may not be running. Starting services..." + docker compose up -d + sleep 5 + fi + + # Check for errors in logs + log_info "Checking for errors in service logs..." + if docker compose logs --tail=50 | grep -i "error" | grep -v "level=error msg=\"error processing" | grep -v "test"; then + log_warn "Found some errors in logs (may be normal)" + else + log_info "No critical errors found in recent logs" + fi +} + +# Build a simple test image +build_single_arch_image() { + log_step "Building single-arch test image..." + + cat > ${BUILD_DIR}/Dockerfile < /test.txt +RUN echo "Architecture: \$(uname -m)" >> /test.txt + +# Add a simple script +RUN echo '#!/bin/sh' > /test.sh && \\ + echo 'echo "Test image running successfully!"' >> /test.sh && \\ + echo 'cat /test.txt' >> /test.sh && \\ + chmod +x /test.sh + +CMD ["/test.sh"] +EOF + + log_cmd docker build -t ${SINGLE_ARCH_IMAGE} ${BUILD_DIR} + log_info "Built single-arch image: ${SINGLE_ARCH_IMAGE}" +} + +# Build multi-arch images using docker manifest create (old school!) +build_multi_arch_images() { + log_step "Building multi-arch images (amd64 and arm64) using docker manifest..." + + # Create Dockerfile for multi-arch builds + cat > ${BUILD_DIR}/Dockerfile.multiarch < /test.txt +RUN echo "Target OS: \${TARGETOS}" >> /test.txt +RUN echo "Target Architecture: \${TARGETARCH}" >> /test.txt +RUN echo "Built at: \$(date)" >> /test.txt + +# Add a simple script +RUN echo '#!/bin/sh' > /test.sh && \\ + echo 'echo "Multi-arch test image running!"' >> /test.sh && \\ + echo 'cat /test.txt' >> /test.sh && \\ + chmod +x /test.sh + +CMD ["/test.sh"] +EOF + + # Build amd64 image + log_info "Building amd64 image..." + log_cmd docker build \ + --platform linux/amd64 \ + --build-arg TARGETARCH=amd64 \ + --build-arg TARGETOS=linux \ + -t ${AMD64_IMAGE} \ + -f ${BUILD_DIR}/Dockerfile.multiarch \ + ${BUILD_DIR} + + # Push amd64 image + log_info "Pushing amd64 image..." + echo -e "${BLUE}[CMD]${NC} docker push ${AMD64_IMAGE}" + if ! docker push ${AMD64_IMAGE}; then + log_error "Failed to push amd64 image" + return 1 + fi + + # Build arm64 image + log_info "Building arm64 image..." + log_cmd docker build \ + --platform linux/arm64 \ + --build-arg TARGETARCH=arm64 \ + --build-arg TARGETOS=linux \ + -t ${ARM64_IMAGE} \ + -f ${BUILD_DIR}/Dockerfile.multiarch \ + ${BUILD_DIR} + + # Push arm64 image + log_info "Pushing arm64 image..." + echo -e "${BLUE}[CMD]${NC} docker push ${ARM64_IMAGE}" + if ! docker push ${ARM64_IMAGE}; then + log_error "Failed to push arm64 image" + return 1 + fi + + # Create manifest list + log_info "Creating multi-arch manifest list..." + echo -e "${BLUE}[CMD]${NC} docker manifest create --insecure ${MULTI_ARCH_IMAGE} ${AMD64_IMAGE} ${ARM64_IMAGE}" + docker manifest create --insecure ${MULTI_ARCH_IMAGE} \ + ${AMD64_IMAGE} \ + ${ARM64_IMAGE} + + # Annotate manifests with platform info + log_info "Annotating manifest with platform info..." + docker manifest annotate ${MULTI_ARCH_IMAGE} ${AMD64_IMAGE} \ + --os linux --arch amd64 + docker manifest annotate ${MULTI_ARCH_IMAGE} ${ARM64_IMAGE} \ + --os linux --arch arm64 + + # Push the manifest list + log_info "Pushing multi-arch manifest list..." + echo -e "${BLUE}[CMD]${NC} docker manifest push --insecure ${MULTI_ARCH_IMAGE}" + docker manifest push --insecure ${MULTI_ARCH_IMAGE} + + log_info "Multi-arch manifest created and pushed: ${MULTI_ARCH_IMAGE}" +} + +# Push single-arch image and verify digest +push_single_arch_image() { + log_step "Pushing single-arch image..." + + echo -e "${BLUE}[CMD]${NC} docker push ${SINGLE_ARCH_IMAGE}" + local push_output + push_output=$(docker push ${SINGLE_ARCH_IMAGE} 2>&1 | tee /dev/tty) + + # Extract and verify digest + local digest + digest=$(echo "$push_output" | grep -oP 'digest: \K[a-z0-9:]+' | tail -1) + + if [ -z "$digest" ]; then + log_error "Failed to get digest from push output" + return 1 + fi + + log_info "Pushed with digest: ${digest}" + + # Verify we can reference by digest + log_info "Verifying digest reference..." + echo -e "${BLUE}[CMD]${NC} docker manifest inspect --insecure ${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}@${digest}" + docker manifest inspect --insecure "${REGISTRY}/${NAMESPACE}/${IMAGE_NAME}@${digest}" + log_info "Digest verification successful!" +} + +# Verify multi-arch manifest +verify_multi_arch_manifest() { + log_step "Verifying multi-arch manifest..." + + echo -e "${BLUE}[CMD]${NC} docker manifest inspect --insecure ${MULTI_ARCH_IMAGE}" + local manifest + manifest=$(docker manifest inspect --insecure ${MULTI_ARCH_IMAGE} | tee /dev/tty) + + # Check for both architectures (check for "amd64" and "arm64" in platform.architecture) + if echo "$manifest" | grep -q '"architecture": "amd64"'; then + log_info "Found linux/amd64 manifest" + else + log_error "Missing linux/amd64 manifest" + return 1 + fi + + if echo "$manifest" | grep -q '"architecture": "arm64"'; then + log_info "Found linux/arm64 manifest" + else + log_error "Missing linux/arm64 manifest" + return 1 + fi + + # Extract digest + local digest + digest=$(echo "$manifest" | jq -r '.manifests[0].digest' 2>/dev/null || echo "$manifest" | grep -oP 'sha256:[a-f0-9]+' | head -1) + + if [ -n "$digest" ]; then + log_info "Multi-arch manifest digest: ${digest}" + fi +} + +# Remove local images +cleanup_local_images() { + log_step "Removing local images..." + + echo -e "${BLUE}[CMD]${NC} docker rmi ${SINGLE_ARCH_IMAGE}" + docker rmi ${SINGLE_ARCH_IMAGE} || true + + echo -e "${BLUE}[CMD]${NC} docker rmi ${AMD64_IMAGE}" + docker rmi ${AMD64_IMAGE} || true + + echo -e "${BLUE}[CMD]${NC} docker rmi ${ARM64_IMAGE}" + docker rmi ${ARM64_IMAGE} || true + + echo -e "${BLUE}[CMD]${NC} docker rmi ${MULTI_ARCH_IMAGE}" + docker rmi ${MULTI_ARCH_IMAGE} || true + + # Clean up manifest list + docker manifest rm ${MULTI_ARCH_IMAGE} 2>/dev/null || true + + # Also remove any cached layers + log_info "Pruning dangling images..." + log_cmd docker image prune -f + + log_info "Local images removed" +} + +# Pull images back +pull_images() { + log_step "Pulling images back from registry..." + + # Pull single-arch image + log_info "Pulling single-arch image..." + echo -e "${BLUE}[CMD]${NC} docker pull ${SINGLE_ARCH_IMAGE}" + if docker pull ${SINGLE_ARCH_IMAGE}; then + log_info "Successfully pulled: ${SINGLE_ARCH_IMAGE}" + else + log_error "Failed to pull single-arch image" + return 1 + fi + + # Pull multi-arch image + log_info "Pulling multi-arch image..." + echo -e "${BLUE}[CMD]${NC} docker pull ${MULTI_ARCH_IMAGE}" + if docker pull ${MULTI_ARCH_IMAGE}; then + log_info "Successfully pulled: ${MULTI_ARCH_IMAGE}" + else + log_error "Failed to pull multi-arch image" + return 1 + fi +} + +# Test running the images +test_images() { + log_step "Testing pulled images..." + + # Test single-arch image + log_info "Running single-arch image..." + log_cmd docker run --rm ${SINGLE_ARCH_IMAGE} + + # Test multi-arch image + log_info "Running multi-arch image..." + log_cmd docker run --rm ${MULTI_ARCH_IMAGE} + + log_info "All images ran successfully!" +} + +# Check for errors in compose logs after operations +check_compose_logs() { + log_step "Checking compose logs for errors..." + + local error_count + error_count=$(docker compose logs --tail=100 | grep -i "error" | grep -v "level=error msg=\"error processing" | grep -v "test" | wc -l) + + if [ "$error_count" -gt 0 ]; then + log_warn "Found ${error_count} error messages in logs:" + docker compose logs --tail=100 | grep -i "error" | grep -v "level=error msg=\"error processing" | grep -v "test" | tail -10 + else + log_info "No errors found in compose logs" + fi +} + +# Multi-arch only test flow +multi_arch_only() { + log_step "Starting ATCR multi-arch test (skipping single-arch)" + log_info "Registry: ${REGISTRY}" + log_info "Namespace: ${NAMESPACE}" + log_info "Build directory: ${BUILD_DIR}" + log_info "Verbose mode: ${VERBOSE} (set VERBOSE=0 to hide docker output)" + + # Check prerequisites + if ! command -v docker &> /dev/null; then + log_error "Docker is not installed" + exit 1 + fi + + if ! command -v jq &> /dev/null; then + log_warn "jq is not installed - some checks may be limited" + fi + + # Run multi-arch test steps only + check_compose_services + build_multi_arch_images + verify_multi_arch_manifest + cleanup_local_images + pull_images + log_info "Running multi-arch image..." + log_cmd docker run --rm ${MULTI_ARCH_IMAGE} + check_compose_logs + + log_step "Multi-arch test completed successfully!" + echo -e "${GREEN}========================================${NC}" + echo -e "${GREEN}Multi-arch test passed!${NC}" + echo -e "${GREEN}========================================${NC}" +} + +# Main test flow +main() { + log_step "Starting ATCR end-to-end test" + log_info "Registry: ${REGISTRY}" + log_info "Namespace: ${NAMESPACE}" + log_info "Build directory: ${BUILD_DIR}" + log_info "Verbose mode: ${VERBOSE} (set VERBOSE=0 to hide docker output)" + + # Check prerequisites + if ! command -v docker &> /dev/null; then + log_error "Docker is not installed" + exit 1 + fi + + if ! command -v jq &> /dev/null; then + log_warn "jq is not installed - some checks may be limited" + fi + + # Run test steps + check_compose_services + build_single_arch_image + push_single_arch_image + build_multi_arch_images + verify_multi_arch_manifest + cleanup_local_images + pull_images + test_images + check_compose_logs + + log_step "End-to-end test completed successfully!" + echo -e "${GREEN}========================================${NC}" + echo -e "${GREEN}All tests passed!${NC}" + echo -e "${GREEN}========================================${NC}" +} + +# Parse arguments +if [ "$1" = "--multi" ]; then + multi_arch_only +else + main "$@" +fi