diff --git a/pkg/appview/middleware/registry.go b/pkg/appview/middleware/registry.go index 3bbd193..1ba1ff7 100644 --- a/pkg/appview/middleware/registry.go +++ b/pkg/appview/middleware/registry.go @@ -20,32 +20,29 @@ import ( "atcr.io/pkg/auth/oauth" ) -// Global refresher instance (set by main.go) -var globalRefresher *oauth.Refresher +// Global variables for initialization only +// These are set by main.go during startup and copied into NamespaceResolver instances. +// After initialization, request handling uses the NamespaceResolver's instance fields. +var ( + globalRefresher *oauth.Refresher + globalDatabase storage.DatabaseMetrics + globalAuthorizer auth.HoldAuthorizer +) -// Global database instance (set by main.go for pull tracking) -var globalDatabase interface { - IncrementPullCount(did, repository string) error - IncrementPushCount(did, repository string) error -} - -// Global authorizer instance (set by main.go for hold authorization) -var globalAuthorizer auth.HoldAuthorizer - -// SetGlobalRefresher sets the global OAuth refresher instance +// SetGlobalRefresher sets the OAuth refresher instance during initialization +// Must be called before the registry starts serving requests func SetGlobalRefresher(refresher *oauth.Refresher) { globalRefresher = refresher } -// SetGlobalDatabase sets the global database instance for metrics tracking -func SetGlobalDatabase(database interface { - IncrementPullCount(did, repository string) error - IncrementPushCount(did, repository string) error -}) { +// SetGlobalDatabase sets the database instance during initialization +// Must be called before the registry starts serving requests +func SetGlobalDatabase(database storage.DatabaseMetrics) { globalDatabase = database } -// SetGlobalAuthorizer sets the global authorizer instance for hold access control +// SetGlobalAuthorizer sets the authorizer instance during initialization +// Must be called before the registry starts serving requests func SetGlobalAuthorizer(authorizer auth.HoldAuthorizer) { globalAuthorizer = authorizer } @@ -59,9 +56,12 @@ func init() { type NamespaceResolver struct { distribution.Namespace directory identity.Directory - defaultHoldDID string // Default hold DID (e.g., "did:web:hold01.atcr.io") - testMode bool // If true, fallback to default hold when user's hold is unreachable - repositories sync.Map // Cache of RoutingRepository instances by key (did:reponame) + defaultHoldDID string // Default hold DID (e.g., "did:web:hold01.atcr.io") + testMode bool // If true, fallback to default hold when user's hold is unreachable + repositories sync.Map // Cache of RoutingRepository instances by key (did:reponame) + refresher *oauth.Refresher // OAuth session manager (copied from global on init) + database storage.DatabaseMetrics // Metrics database (copied from global on init) + authorizer auth.HoldAuthorizer // Hold authorization (copied from global on init) } // initATProtoResolver initializes the name resolution middleware @@ -82,11 +82,16 @@ func initATProtoResolver(ctx context.Context, ns distribution.Namespace, _ drive testMode = tm } + // Copy shared services from globals into the instance + // This avoids accessing globals during request handling return &NamespaceResolver{ Namespace: ns, directory: directory, defaultHoldDID: defaultHoldDID, testMode: testMode, + refresher: globalRefresher, + database: globalDatabase, + authorizer: globalAuthorizer, }, nil } @@ -155,9 +160,9 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Fall back to Basic Auth token cache (for users who used app passwords) var atprotoClient *atproto.Client - if globalRefresher != nil { + if nr.refresher != nil { // Try OAuth flow first - session, err := globalRefresher.GetSession(ctx, did) + session, err := nr.refresher.GetSession(ctx, did) if err == nil { // OAuth session available - use indigo's API client (handles DPoP automatically) apiClient := session.APIClient() @@ -194,8 +199,18 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Create routing repository - routes manifests to ATProto, blobs to hold service // The registry is stateless - no local storage is used - // Pass hold DID, user DID, authorizer, and refresher as parameters (can't use context as it gets lost) - routingRepo := storage.NewRoutingRepository(repo, atprotoClient, repositoryName, holdDID, did, globalDatabase, globalAuthorizer, globalRefresher) + // Bundle all context into a single RegistryContext struct + registryCtx := &storage.RegistryContext{ + DID: did, + HoldDID: holdDID, + PDSEndpoint: pdsEndpoint, + Repository: repositoryName, + ATProtoClient: atprotoClient, + Database: nr.database, + Authorizer: nr.authorizer, + Refresher: nr.refresher, + } + routingRepo := storage.NewRoutingRepository(repo, registryCtx) // Cache the repository nr.repositories.Store(cacheKey, routingRepo) diff --git a/pkg/appview/storage/context.go b/pkg/appview/storage/context.go new file mode 100644 index 0000000..fe4514b --- /dev/null +++ b/pkg/appview/storage/context.go @@ -0,0 +1,29 @@ +package storage + +import ( + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth" + "atcr.io/pkg/auth/oauth" +) + +// DatabaseMetrics interface for tracking pull/push counts +type DatabaseMetrics interface { + IncrementPullCount(did, repository string) error + IncrementPushCount(did, repository string) error +} + +// RegistryContext bundles all the context needed for registry operations +// This includes both per-request data (DID, hold) and shared services +type RegistryContext struct { + // Per-request identity and routing information + DID string // User's DID (e.g., "did:plc:abc123") + 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") + ATProtoClient *atproto.Client // Authenticated ATProto client for this user + + // Shared services (same for all requests) + Database DatabaseMetrics // Metrics tracking database + Authorizer auth.HoldAuthorizer // Hold access authorization + Refresher *oauth.Refresher // OAuth session manager +} diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index 19e7d29..439ed4e 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -11,8 +11,6 @@ import ( "sync" "time" - "atcr.io/pkg/auth" - "atcr.io/pkg/auth/oauth" "github.com/distribution/distribution/v3" "github.com/opencontainers/go-digest" ) @@ -32,26 +30,21 @@ var ( // ProxyBlobStore proxies blob requests to an external storage service type ProxyBlobStore struct { - holdDID string // Hold DID (e.g., "did:web:hold01.atcr.io") - holdURL string // Resolved HTTP URL for XRPC requests + ctx *RegistryContext // All context and services + holdURL string // Resolved HTTP URL for XRPC requests httpClient *http.Client - did string - database DatabaseMetrics - repository string - authorizer auth.HoldAuthorizer - refresher *oauth.Refresher // OAuth refresher for authenticating to hold service } // NewProxyBlobStore creates a new proxy blob store -func NewProxyBlobStore(holdDID, did string, database DatabaseMetrics, repository string, authorizer auth.HoldAuthorizer, refresher *oauth.Refresher) *ProxyBlobStore { +func NewProxyBlobStore(ctx *RegistryContext) *ProxyBlobStore { // Resolve DID to URL once at construction time - holdURL := resolveHoldURL(holdDID) + holdURL := resolveHoldURL(ctx.HoldDID) fmt.Printf("DEBUG [proxy_blob_store]: NewProxyBlobStore created with holdDID=%s, holdURL=%s, userDID=%s, repo=%s\n", - holdDID, holdURL, did, repository) + ctx.HoldDID, holdURL, ctx.DID, ctx.Repository) return &ProxyBlobStore{ - holdDID: holdDID, + ctx: ctx, holdURL: holdURL, httpClient: &http.Client{ Timeout: 5 * time.Minute, // Timeout for presigned URL requests and uploads @@ -63,11 +56,6 @@ func NewProxyBlobStore(holdDID, did string, database DatabaseMetrics, repository IdleConnTimeout: 90 * time.Second, }, }, - did: did, - database: database, - repository: repository, - authorizer: authorizer, - refresher: refresher, } } @@ -76,13 +64,13 @@ func NewProxyBlobStore(holdDID, did string, database DatabaseMetrics, repository // Otherwise, uses the default httpClient without authentication func (p *ProxyBlobStore) doAuthenticatedRequest(ctx context.Context, req *http.Request) (*http.Response, error) { // Try to get OAuth session for DPoP authentication - if p.refresher != nil { - session, err := p.refresher.GetSession(ctx, p.did) + if p.ctx.Refresher != nil { + session, err := p.ctx.Refresher.GetSession(ctx, p.ctx.DID) if err != nil { - fmt.Printf("DEBUG [proxy_blob_store]: Failed to get OAuth session for DID=%s: %v, will attempt without auth\n", p.did, err) + fmt.Printf("DEBUG [proxy_blob_store]: Failed to get OAuth session for DID=%s: %v, will attempt without auth\n", p.ctx.DID, err) } else { // Use session's DoWithAuth method (adds Authorization + DPoP headers) - fmt.Printf("DEBUG [proxy_blob_store]: Using OAuth session for hold service request, DID=%s\n", p.did) + fmt.Printf("DEBUG [proxy_blob_store]: Using OAuth session for hold service request, DID=%s\n", p.ctx.DID) // The endpoint parameter is not used for DPoP signing, just token refresh validation // For hold service XRPC requests, we can pass "com.atproto.repo.uploadBlob" return session.DoWithAuth(session.Client, req, "com.atproto.repo.uploadBlob") @@ -93,6 +81,36 @@ func (p *ProxyBlobStore) doAuthenticatedRequest(ctx context.Context, req *http.R return p.httpClient.Do(req) } +// checkReadAccess validates that the user has read access to blobs in this hold +func (p *ProxyBlobStore) checkReadAccess(ctx context.Context) error { + if p.ctx.Authorizer == nil { + return nil // No authorization check if authorizer not configured + } + allowed, err := p.ctx.Authorizer.CheckReadAccess(ctx, p.ctx.HoldDID, p.ctx.DID) + if err != nil { + return fmt.Errorf("authorization check failed: %w", err) + } + if !allowed { + return distribution.ErrBlobUnknown // Return same error as missing blob for security + } + return nil +} + +// checkWriteAccess validates that the user has write access to blobs in this hold +func (p *ProxyBlobStore) checkWriteAccess(ctx context.Context) error { + if p.ctx.Authorizer == nil { + return nil // No authorization check if authorizer not configured + } + allowed, err := p.ctx.Authorizer.CheckWriteAccess(ctx, p.ctx.HoldDID, p.ctx.DID) + if err != nil { + return fmt.Errorf("authorization check failed: %w", err) + } + if !allowed { + return fmt.Errorf("write access denied to hold %s", p.ctx.HoldDID) + } + return nil +} + // resolveHoldURL converts a hold DID to an HTTP URL for XRPC requests // did:web:hold01.atcr.io → https://hold01.atcr.io // did:web:172.28.0.3:8080 → http://172.28.0.3:8080 @@ -110,44 +128,6 @@ func resolveHoldURL(holdDID string) string { return "https://" + hostname } -// checkReadAccess verifies the user has read access to the hold -func (p *ProxyBlobStore) checkReadAccess(ctx context.Context) error { - if p.authorizer == nil { - // No authorizer configured - allow access (backward compatibility) - return nil - } - - hasAccess, err := p.authorizer.CheckReadAccess(ctx, p.holdDID, p.did) - if err != nil { - return fmt.Errorf("authorization check failed: %w", err) - } - - if !hasAccess { - return distribution.ErrBlobUnknown // Return same error as missing blob for security - } - - return nil -} - -// checkWriteAccess verifies the user has write access to the hold -func (p *ProxyBlobStore) checkWriteAccess(ctx context.Context) error { - if p.authorizer == nil { - // No authorizer configured - allow access (backward compatibility) - return nil - } - - hasAccess, err := p.authorizer.CheckWriteAccess(ctx, p.holdDID, p.did) - if err != nil { - return fmt.Errorf("authorization check failed: %w", err) - } - - if !hasAccess { - return fmt.Errorf("write access denied to hold %s", p.holdDID) - } - - return nil -} - // Stat returns the descriptor for a blob func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { // Check read access @@ -390,7 +370,7 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) // Use XRPC endpoint: GET /xrpc/com.atproto.sync.getBlob?did={holdDID}&cid={digest} // Per migration doc: hold accepts OCI digest directly as cid parameter (checks for sha256: prefix) url := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", - p.holdURL, p.holdDID, dgst.String()) + p.holdURL, p.ctx.HoldDID, dgst.String()) return url, nil } @@ -399,7 +379,7 @@ func (p *ProxyBlobStore) getDownloadURL(ctx context.Context, dgst digest.Digest) func (p *ProxyBlobStore) getHeadURL(ctx context.Context, dgst digest.Digest) (string, error) { // Same as GET - hold service handles HEAD method on getBlob endpoint url := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s", - p.holdURL, p.holdDID, dgst.String()) + p.holdURL, p.ctx.HoldDID, dgst.String()) return url, nil } diff --git a/pkg/appview/storage/routing_repository.go b/pkg/appview/storage/routing_repository.go index e9aeb31..7978bd3 100644 --- a/pkg/appview/storage/routing_repository.go +++ b/pkg/appview/storage/routing_repository.go @@ -6,52 +6,23 @@ import ( "time" "atcr.io/pkg/atproto" - "atcr.io/pkg/auth" - "atcr.io/pkg/auth/oauth" "github.com/distribution/distribution/v3" ) -// DatabaseMetrics interface for tracking pull/push counts -type DatabaseMetrics interface { - IncrementPullCount(did, repository string) error - IncrementPushCount(did, repository string) error -} - // RoutingRepository routes manifests to ATProto and blobs to external hold service // The registry (AppView) is stateless and NEVER stores blobs locally type RoutingRepository struct { distribution.Repository - atprotoClient *atproto.Client - repositoryName string - holdDID string // Hold service DID for blobs (from discovery for push), e.g., "did:web:hold01.atcr.io" - did string // User's DID for authorization - manifestStore *atproto.ManifestStore // Cached manifest store instance - blobStore *ProxyBlobStore // Cached blob store instance - database DatabaseMetrics // Database for metrics tracking - authorizer auth.HoldAuthorizer // Authorization for hold access - refresher *oauth.Refresher // OAuth refresher for authenticating to hold service + ctx *RegistryContext // All context and services + manifestStore *atproto.ManifestStore // Cached manifest store instance + blobStore *ProxyBlobStore // Cached blob store instance } // NewRoutingRepository creates a new routing repository -func NewRoutingRepository( - baseRepo distribution.Repository, - atprotoClient *atproto.Client, - repoName string, - holdDID string, - did string, - database DatabaseMetrics, - authorizer auth.HoldAuthorizer, - refresher *oauth.Refresher, -) *RoutingRepository { +func NewRoutingRepository(baseRepo distribution.Repository, ctx *RegistryContext) *RoutingRepository { return &RoutingRepository{ - Repository: baseRepo, - atprotoClient: atprotoClient, - repositoryName: repoName, - holdDID: holdDID, - did: did, - database: database, - authorizer: authorizer, - refresher: refresher, + Repository: baseRepo, + ctx: ctx, } } @@ -64,7 +35,15 @@ 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.atprotoClient, r.repositoryName, r.holdDID, r.holdDID, r.did, blobStore, r.database) + r.manifestStore = atproto.NewManifestStore( + r.ctx.ATProtoClient, + r.ctx.Repository, + r.ctx.HoldDID, + r.ctx.HoldDID, + r.ctx.DID, + blobStore, + r.ctx.Database, + ) } // After any manifest operation, cache the hold DID for blob fetches @@ -73,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.did, r.repositoryName, 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.did, r.repositoryName, holdDID) + r.ctx.DID, r.ctx.Repository, holdDID) } }() @@ -88,37 +67,40 @@ 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.did, r.repositoryName) + 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.holdDID // Default to discovery-based DID + holdDID := r.ctx.HoldDID // Default to discovery-based DID - if cachedHoldDID, ok := GetGlobalHoldCache().Get(r.did, r.repositoryName); 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.did, r.repositoryName, 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.did, r.repositoryName, holdDID) + r.ctx.DID, r.ctx.Repository, holdDID) } if holdDID == "" { // This should never happen if middleware is configured correctly - panic("hold DID not set in RoutingRepository - ensure default_hold_did is configured in middleware") + panic("hold DID not set in RegistryContext - ensure default_hold_did is configured in middleware") } - // Create and cache proxy blob store with authorization and OAuth refresher - r.blobStore = NewProxyBlobStore(holdDID, r.did, r.database, r.repositoryName, r.authorizer, r.refresher) + // Update context with the correct hold DID (may be cached or discovered) + r.ctx.HoldDID = holdDID + + // Create and cache proxy blob store + 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.atprotoClient, r.repositoryName) + return atproto.NewTagStore(r.ctx.ATProtoClient, r.ctx.Repository) }