diff --git a/cmd/credential-helper/main.go b/cmd/credential-helper/main.go index 88d8784..15bf9f4 100644 --- a/cmd/credential-helper/main.go +++ b/cmd/credential-helper/main.go @@ -85,7 +85,7 @@ func handleGet() { // will validate the session token and issue a registry JWT creds := Credentials{ ServerURL: serverURL, - Username: "oauth2", // Signals token-based auth to Docker + Username: "oauth2", // Signals token-based auth to Docker Secret: session.SessionToken, // Return session token directly } @@ -280,4 +280,3 @@ func exchangeSessionForRegistryToken(sessionToken, appViewURL string) (string, e } return result.AccessToken, nil } - diff --git a/cmd/hold/main.go b/cmd/hold/main.go index 9333877..9155acc 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -142,7 +142,7 @@ func (s *HoldService) HandleGetPresignedURL(w http.ResponseWriter, r *http.Reque // For now, construct direct URL to blob // In production, this would use driver-specific presigned URLs - url, err := s.getDownloadURL(ctx, req.Digest) + url, err := s.getDownloadURL(ctx, req.Digest, req.DID) if err != nil { http.Error(w, fmt.Sprintf("failed to generate URL: %v", err), http.StatusInternalServerError) return @@ -258,10 +258,50 @@ func (s *HoldService) HandleProxyGet(w http.ResponseWriter, r *http.Request) { w.Write(content) } +// HandleMove moves a blob from one path to another +// POST /move?from={path}&to={digest}&did={did} +func (s *HoldService) HandleMove(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + fromPath := r.URL.Query().Get("from") + toDigest := r.URL.Query().Get("to") + did := r.URL.Query().Get("did") + + if fromPath == "" || toDigest == "" { + http.Error(w, "missing from or to parameter", http.StatusBadRequest) + return + } + + // Authorize WRITE access + if !s.isAuthorizedWrite(did) { + if did == "" { + http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized) + } else { + http.Error(w, "forbidden: write access denied", http.StatusForbidden) + } + return + } + + ctx := r.Context() + sourcePath := blobPath(fromPath) + destPath := blobPath(toDigest) + + // Try to move using driver's Move operation + if err := s.driver.Move(ctx, sourcePath, destPath); err != nil { + log.Printf("HandleMove: failed to move blob: %v", err) + http.Error(w, fmt.Sprintf("failed to move blob: %v", err), http.StatusInternalServerError) + return + } + + log.Printf("HandleMove: successfully moved blob from=%s to=%s", fromPath, toDigest) + w.WriteHeader(http.StatusOK) +} + // HandleProxyPut proxies a blob upload through the service func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) { - log.Printf("HandleProxyPut: method=%s, path=%s, query=%s", r.Method, r.URL.Path, r.URL.RawQuery) - if r.Method != http.MethodPut { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return @@ -278,41 +318,45 @@ func (s *HoldService) HandleProxyPut(w http.ResponseWriter, r *http.Request) { did = r.Header.Get("X-ATCR-DID") } - log.Printf("HandleProxyPut: digest=%s, did=%s", digest, did) - // Authorize WRITE access - authorized := s.isAuthorizedWrite(did) - log.Printf("HandleProxyPut: authorization check: did=%s, authorized=%v", did, authorized) - if !authorized { + if !s.isAuthorizedWrite(did) { if did == "" { - log.Printf("HandleProxyPut: rejecting - no DID provided") http.Error(w, "unauthorized: authentication required", http.StatusUnauthorized) } else { - log.Printf("HandleProxyPut: rejecting - DID not authorized for write") http.Error(w, "forbidden: write access denied", http.StatusForbidden) } return } - // Write blob to storage + // Stream blob to storage (no buffering) ctx := r.Context() path := blobPath(digest) - content, err := io.ReadAll(r.Body) + // Create writer for streaming + writer, err := s.driver.Writer(ctx, path, false) if err != nil { - log.Printf("HandleProxyPut: failed to read body: %v", err) - http.Error(w, "failed to read body", http.StatusBadRequest) + log.Printf("HandleProxyPut: failed to create writer: %v", err) + http.Error(w, "failed to create writer", http.StatusInternalServerError) return } - log.Printf("HandleProxyPut: writing blob to path=%s, size=%d bytes", path, len(content)) - if err := s.driver.PutContent(ctx, path, content); err != nil { - log.Printf("HandleProxyPut: failed to store blob: %v", err) - http.Error(w, "failed to store blob", http.StatusInternalServerError) + // Stream directly from request body to storage + written, err := io.Copy(writer, r.Body) + if err != nil { + writer.Cancel(ctx) + log.Printf("HandleProxyPut: failed to write blob: %v", err) + http.Error(w, "failed to write blob", http.StatusInternalServerError) return } - log.Printf("HandleProxyPut: successfully stored blob digest=%s, size=%d", digest, len(content)) + // Commit the write + if err := writer.Commit(ctx); err != nil { + log.Printf("HandleProxyPut: failed to commit blob: %v", err) + http.Error(w, "failed to commit blob", http.StatusInternalServerError) + return + } + + log.Printf("HandleProxyPut: successfully stored blob path=%s, size=%d", digest, written) w.WriteHeader(http.StatusCreated) } @@ -426,7 +470,7 @@ func (s *HoldService) isCrewMember(did string) (bool, error) { } // getDownloadURL generates a download URL for a blob -func (s *HoldService) getDownloadURL(ctx context.Context, digest string) (string, error) { +func (s *HoldService) getDownloadURL(ctx context.Context, digest string, did string) (string, error) { // Check if blob exists path := blobPath(digest) _, err := s.driver.Stat(ctx, path) @@ -435,8 +479,8 @@ func (s *HoldService) getDownloadURL(ctx context.Context, digest string) (string } // For drivers that support presigned URLs (S3), use those - // For now, return a proxy URL through this service - return fmt.Sprintf("%s/blobs/%s", s.config.Server.PublicURL, digest), nil + // For now, return a proxy URL through this service with DID for authorization + return fmt.Sprintf("%s/blobs/%s?did=%s", s.config.Server.PublicURL, digest, did), nil } // getUploadURL generates an upload URL for a blob @@ -563,6 +607,7 @@ func main() { mux.HandleFunc("/register", service.HandleRegister) mux.HandleFunc("/get-presigned-url", service.HandleGetPresignedURL) mux.HandleFunc("/put-presigned-url", service.HandlePutPresignedURL) + mux.HandleFunc("/move", service.HandleMove) // OAuth client metadata endpoint for ATProto OAuth clientID := cfg.Server.PublicURL + "/client-metadata.json" @@ -697,11 +742,16 @@ func getEnvOrDefault(key, defaultValue string) string { return defaultValue } -// blobPath converts a digest (e.g., "sha256:abc123...") to a storage path +// blobPath converts a digest (e.g., "sha256:abc123...") or temp path to a storage path // Distribution stores blobs as: /docker/registry/v2/blobs/{algorithm}/{xx}/{hash}/data // where xx is the first 2 characters of the hash for directory sharding // NOTE: Path must start with / for filesystem driver func blobPath(digest string) string { + // Handle temp paths (start with uploads/temp-) + if strings.HasPrefix(digest, "uploads/temp-") { + return fmt.Sprintf("/docker/registry/v2/%s/data", digest) + } + // Split digest into algorithm and hash parts := strings.SplitN(digest, ":", 2) if len(parts) != 2 { diff --git a/cmd/registry/serve.go b/cmd/registry/serve.go index 052176b..8ba9ca9 100644 --- a/cmd/registry/serve.go +++ b/cmd/registry/serve.go @@ -112,7 +112,7 @@ func serveRegistry(cmd *cobra.Command, args []string) error { clientIDConfig := oauth.ClientIDConfig{ BaseURL: baseURL, CallbackPath: "/auth/oauth/callback", - Scopes: []string{"atproto"}, + Scopes: oauth.GetDefaultScopes(), } clientID, redirectURI := clientIDConfig.MakeClientID() @@ -133,6 +133,8 @@ func serveRegistry(cmd *cobra.Command, args []string) error { // 8. Create OAuth server oauthServer := oauth.NewServer(refreshStorage, sessionManager, baseURL) + // Connect server to refresher for cache invalidation + oauthServer.SetRefresher(refresher) // 9. Initialize auth keys and create token issuer var issuer *token.Issuer diff --git a/config/config.yml b/config/config.yml index 2bfa34c..64a77c5 100644 --- a/config/config.yml +++ b/config/config.yml @@ -38,6 +38,7 @@ auth: realm: http://127.0.0.1:5000/auth/token service: atcr.io issuer: atcr.io + expiration: 1800 # 30 minutes (in seconds) # Certificate bundle for validating JWTs rootcertbundle: /var/lib/atcr/auth/private-key.crt diff --git a/docker-compose.yml b/docker-compose.yml index f104384..1056171 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,5 +1,5 @@ services: - registry: + atcr-registry: build: context: . dockerfile: Dockerfile @@ -24,7 +24,7 @@ services: # - OAuth tokens -> Persistent volume (atcr-tokens) # Future: Add read_only: true for production deployments - hold: + atcr-hold: environment: HOLD_PUBLIC_URL: http://172.28.0.3:8080 HOLD_OWNER: did:plc:pddp4xt5lgnv2qsegbzzs4xg diff --git a/pkg/atproto/client.go b/pkg/atproto/client.go index bff2b8e..22d0c0c 100644 --- a/pkg/atproto/client.go +++ b/pkg/atproto/client.go @@ -231,7 +231,15 @@ func (c *Client) UploadBlob(ctx context.Context, data []byte, mimeType string) ( return nil, err } - req.Header.Set("Authorization", c.authHeader()) + // Only set Authorization header if we have an access token + if c.accessToken != "" { + authHeader := c.authHeader() + fmt.Printf("DEBUG [atproto/client]: UploadBlob Authorization header: %q (useDPoP=%v, token_length=%d)\n", authHeader, c.useDPoP, len(c.accessToken)) + req.Header.Set("Authorization", authHeader) + } else { + fmt.Printf("DEBUG [atproto/client]: UploadBlob: No access token available, sending unauthenticated request\n") + return nil, fmt.Errorf("no access token available for authenticated PDS operation - please complete OAuth flow at: http://127.0.0.1:5000/auth/oauth/authorize?handle=") + } req.Header.Set("Content-Type", mimeType) resp, err := c.httpClient.Do(req) diff --git a/pkg/auth/oauth/client.go b/pkg/auth/oauth/client.go index 7f82f91..c7b3a40 100644 --- a/pkg/auth/oauth/client.go +++ b/pkg/auth/oauth/client.go @@ -10,7 +10,7 @@ import ( "fmt" "net/http" - atprotoclient "atcr.io/pkg/atproto" + "atcr.io/pkg/atproto" "authelia.com/client/oauth2" ) @@ -19,7 +19,7 @@ type Client struct { config *oauth2.Config dpopKey *ecdsa.PrivateKey dpopTransport *DPoPTransport - resolver *atprotoclient.Resolver + resolver *atproto.Resolver clientID string redirectURI string metadata *AuthServerMetadata @@ -36,7 +36,7 @@ func NewClient(clientID, redirectURI string) (*Client, error) { return &Client{ dpopKey: dpopKey, dpopTransport: NewDPoPTransport(http.DefaultTransport, dpopKey), - resolver: atprotoclient.NewResolver(), + resolver: atproto.NewResolver(), clientID: clientID, redirectURI: redirectURI, }, nil @@ -70,7 +70,7 @@ func (c *Client) InitializeForHandle(ctx context.Context, handle string) error { PushedAuthURL: metadata.PushedAuthorizationRequestEndpoint, }, RedirectURL: c.redirectURI, - Scopes: []string{"atproto"}, + Scopes: GetDefaultScopes(), } return nil diff --git a/pkg/auth/oauth/client_id.go b/pkg/auth/oauth/client_id.go index 96050da..3d12b5a 100644 --- a/pkg/auth/oauth/client_id.go +++ b/pkg/auth/oauth/client_id.go @@ -27,9 +27,9 @@ func IsLocalhostURL(urlStr string) bool { // ClientIDConfig helps construct appropriate client IDs for different environments type ClientIDConfig struct { - BaseURL string // Base URL (e.g., "http://127.0.0.1:8888" or "https://example.com") + BaseURL string // Base URL (e.g., "http://127.0.0.1:8888" or "https://example.com") CallbackPath string // Callback path (e.g., "/oauth/callback") - Scopes []string // OAuth scopes + Scopes []string // OAuth scopes } // MakeClientID creates the appropriate client ID based on the environment diff --git a/pkg/auth/oauth/flow.go b/pkg/auth/oauth/flow.go index 3c545ff..d4ce686 100644 --- a/pkg/auth/oauth/flow.go +++ b/pkg/auth/oauth/flow.go @@ -24,8 +24,9 @@ type FlowResult struct { // RunInteractiveFlow executes an interactive OAuth authorization code flow // The setupCallback function is called TWICE: -// 1. First with authURL="" to start the server (before PAR) -// 2. Then with the actual authURL to display it to the user (after PAR) +// 1. First with authURL="" to start the server (before PAR) +// 2. Then with the actual authURL to display it to the user (after PAR) +// // This two-phase approach ensures the server is running before PAR tries to fetch client metadata func RunInteractiveFlow(ctx context.Context, cfg InteractiveFlowConfig, setupCallback func(authURL string, handler *CallbackHandler, metadata *ClientMetadata) error) (*FlowResult, error) { diff --git a/pkg/auth/oauth/refresher.go b/pkg/auth/oauth/refresher.go index 62e2bc8..deed2ef 100644 --- a/pkg/auth/oauth/refresher.go +++ b/pkg/auth/oauth/refresher.go @@ -15,16 +15,19 @@ import ( type AccessTokenEntry struct { Token string DPoPKey *ecdsa.PrivateKey + Transport *DPoPTransport // Cache the transport to preserve nonce across requests ExpiresAt time.Time } // Refresher manages OAuth token refresh for AppView type Refresher struct { - storage *RefreshTokenStorage - accessTokens map[string]*AccessTokenEntry - mu sync.RWMutex - clientID string - redirectURI string + storage *RefreshTokenStorage + accessTokens map[string]*AccessTokenEntry + mu sync.RWMutex + refreshLocks map[string]*sync.Mutex // Per-DID locks for refresh operations + refreshLockMu sync.Mutex // Protects refreshLocks map + clientID string + redirectURI string } // NewRefresher creates a new token refresher @@ -32,6 +35,7 @@ func NewRefresher(storage *RefreshTokenStorage, clientID, redirectURI string) *R return &Refresher{ storage: storage, accessTokens: make(map[string]*AccessTokenEntry), + refreshLocks: make(map[string]*sync.Mutex), clientID: clientID, redirectURI: redirectURI, } @@ -39,33 +43,59 @@ func NewRefresher(storage *RefreshTokenStorage, clientID, redirectURI string) *R // GetAccessToken gets a fresh access token for a DID // Returns cached token if still valid, otherwise refreshes -func (r *Refresher) GetAccessToken(ctx context.Context, did string) (string, *ecdsa.PrivateKey, error) { - // Check cache first +// Returns: accessToken, dpopKey, dpopTransport, error +func (r *Refresher) GetAccessToken(ctx context.Context, did string) (string, *ecdsa.PrivateKey, *DPoPTransport, error) { + // Check cache first (fast path) r.mu.RLock() entry, ok := r.accessTokens[did] r.mu.RUnlock() if ok && time.Now().Before(entry.ExpiresAt) { // Token still valid - return entry.Token, entry.DPoPKey, nil + return entry.Token, entry.DPoPKey, entry.Transport, nil } - // Token expired or not cached, refresh it + // Token expired or not cached, need to refresh + // Get or create per-DID lock to prevent concurrent refreshes + r.refreshLockMu.Lock() + didLock, ok := r.refreshLocks[did] + if !ok { + didLock = &sync.Mutex{} + r.refreshLocks[did] = didLock + } + r.refreshLockMu.Unlock() + + // Acquire DID-specific lock + didLock.Lock() + defer didLock.Unlock() + + // Double-check cache after acquiring lock (another goroutine might have refreshed) + r.mu.RLock() + entry, ok = r.accessTokens[did] + r.mu.RUnlock() + + if ok && time.Now().Before(entry.ExpiresAt) { + // Token was refreshed while we waited for the lock + return entry.Token, entry.DPoPKey, entry.Transport, nil + } + + // Actually refresh the token return r.RefreshToken(ctx, did) } // RefreshToken forces a token refresh for a DID -func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecdsa.PrivateKey, error) { +// Returns: accessToken, dpopKey, dpopTransport, error +func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecdsa.PrivateKey, *DPoPTransport, error) { // Get stored refresh token entry, err := r.storage.Get(did) if err != nil { - return "", nil, fmt.Errorf("failed to get stored refresh token: %w", err) + return "", nil, nil, fmt.Errorf("failed to get stored refresh token: %w", err) } // Parse DPoP key dpopKey, err := r.storage.GetDPoPKey(did) if err != nil { - return "", nil, fmt.Errorf("failed to get DPoP key: %w", err) + return "", nil, nil, fmt.Errorf("failed to get DPoP key: %w", err) } // Create OAuth client with DPoP transport @@ -75,7 +105,7 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds // Discover PDS OAuth metadata metadata, err := DiscoverAuthServer(ctx, entry.PDS) if err != nil { - return "", nil, fmt.Errorf("failed to discover auth server: %w", err) + return "", nil, nil, fmt.Errorf("failed to discover auth server: %w", err) } // Configure OAuth2 client @@ -87,7 +117,7 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds PushedAuthURL: metadata.PushedAuthorizationRequestEndpoint, }, RedirectURL: r.redirectURI, - Scopes: []string{"atproto"}, + Scopes: GetDefaultScopes(), } // Create context with custom HTTP client @@ -98,7 +128,7 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds RefreshToken: entry.RefreshToken, }).Token() if err != nil { - return "", nil, fmt.Errorf("failed to refresh token: %w", err) + return "", nil, nil, fmt.Errorf("failed to refresh token: %w", err) } // Update last refresh timestamp @@ -116,7 +146,10 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds } } - // Cache the access token + // Set access token on transport for "ath" claim in future DPoP proofs + dpopTransport.SetAccessToken(token.AccessToken) + + // Cache the access token and transport // Expire 1 minute early to avoid edge cases expiresAt := token.Expiry.Add(-1 * time.Minute) @@ -124,11 +157,20 @@ func (r *Refresher) RefreshToken(ctx context.Context, did string) (string, *ecds r.accessTokens[did] = &AccessTokenEntry{ Token: token.AccessToken, DPoPKey: dpopKey, + Transport: dpopTransport, // Cache transport to preserve nonce ExpiresAt: expiresAt, } r.mu.Unlock() - return token.AccessToken, dpopKey, nil + return token.AccessToken, dpopKey, dpopTransport, nil +} + +// InvalidateAccessToken removes a cached access token for a DID +// This is useful when a new refresh token is obtained (e.g., after re-authorization) +func (r *Refresher) InvalidateAccessToken(did string) { + r.mu.Lock() + delete(r.accessTokens, did) + r.mu.Unlock() } // RevokeToken removes stored refresh token and cached access token diff --git a/pkg/auth/oauth/scopes.go b/pkg/auth/oauth/scopes.go new file mode 100644 index 0000000..84633f9 --- /dev/null +++ b/pkg/auth/oauth/scopes.go @@ -0,0 +1,22 @@ +package oauth + +import ( + "fmt" + + "atcr.io/pkg/atproto" +) + +// GetDefaultScopes returns the default OAuth scopes for ATCR registry operations +func GetDefaultScopes() []string { + return []string{ + "atproto", + "transition:generic.full", + "blob:application/vnd.docker.distribution.manifest.v2+json", + "blob:application/vnd.docker.image.rootfs.diff.tar.gzip", + "blob:application/vnd.docker.container.image.v1+json", + fmt.Sprintf("repo:%s?action=create", atproto.ManifestCollection), + fmt.Sprintf("repo:%s?action=update", atproto.ManifestCollection), + fmt.Sprintf("repo:%s?action=create", atproto.TagCollection), + fmt.Sprintf("repo:%s?action=update", atproto.TagCollection), + } +} diff --git a/pkg/auth/oauth/server.go b/pkg/auth/oauth/server.go index a2c1365..41f41fd 100644 --- a/pkg/auth/oauth/server.go +++ b/pkg/auth/oauth/server.go @@ -17,25 +17,26 @@ import ( // Server handles OAuth authorization for the AppView type Server struct { - storage *RefreshTokenStorage - sessionManager *session.Manager - resolver *atproto.Resolver - clientID string - redirectURI string - baseURL string - states map[string]*OAuthState - statesMu sync.RWMutex + storage *RefreshTokenStorage + sessionManager *session.Manager + resolver *atproto.Resolver + refresher *Refresher + clientID string + redirectURI string + baseURL string + states map[string]*OAuthState + statesMu sync.RWMutex } // OAuthState tracks an in-progress OAuth flow type OAuthState struct { - State string - Handle string - DID string - PDSEndpoint string - CodeVerifier string - DPoPKey *ecdsa.PrivateKey - CreatedAt time.Time + State string + Handle string + DID string + PDSEndpoint string + CodeVerifier string + DPoPKey *ecdsa.PrivateKey + CreatedAt time.Time } // NewServer creates a new OAuth server @@ -44,7 +45,7 @@ func NewServer(storage *RefreshTokenStorage, sessionManager *session.Manager, ba cfg := ClientIDConfig{ BaseURL: baseURL, CallbackPath: "/auth/oauth/callback", - Scopes: []string{"atproto"}, + Scopes: GetDefaultScopes(), } clientID, redirectURI := cfg.MakeClientID() @@ -52,6 +53,7 @@ func NewServer(storage *RefreshTokenStorage, sessionManager *session.Manager, ba storage: storage, sessionManager: sessionManager, resolver: atproto.NewResolver(), + refresher: nil, // Will be set via SetRefresher() clientID: clientID, redirectURI: redirectURI, baseURL: baseURL, @@ -59,6 +61,11 @@ func NewServer(storage *RefreshTokenStorage, sessionManager *session.Manager, ba } } +// SetRefresher sets the refresher for invalidating access token cache +func (s *Server) SetRefresher(refresher *Refresher) { + s.refresher = refresher +} + // ServeAuthorize handles GET /auth/oauth/authorize func (s *Server) ServeAuthorize(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { @@ -190,7 +197,7 @@ func (s *Server) exchangeCodeForSession(ctx context.Context, code string, state PushedAuthURL: metadata.PushedAuthorizationRequestEndpoint, }, RedirectURL: s.redirectURI, - Scopes: []string{"atproto"}, + Scopes: GetDefaultScopes(), } // Create context with custom HTTP client @@ -222,6 +229,12 @@ func (s *Server) exchangeCodeForSession(ctx context.Context, code string, state return "", fmt.Errorf("failed to store refresh token: %w", err) } + // Invalidate cached access token (if any) since we have a new refresh token with new scopes + if s.refresher != nil { + s.refresher.InvalidateAccessToken(state.DID) + fmt.Printf("DEBUG [oauth/server]: Invalidated cached access token for DID=%s after storing new refresh token\n", state.DID) + } + // Create session token for credential helper sessionToken, err := s.sessionManager.Create(state.DID, state.Handle) if err != nil { diff --git a/pkg/auth/oauth/transport.go b/pkg/auth/oauth/transport.go index ea1dd8e..71c8011 100644 --- a/pkg/auth/oauth/transport.go +++ b/pkg/auth/oauth/transport.go @@ -123,6 +123,11 @@ func (t *DPoPTransport) addDPoPHeader(req *http.Request) error { // Add DPoP header req.Header.Set("DPoP", proofString) + proofPreview := proofString + if len(proofPreview) > 50 { + proofPreview = proofPreview[:50] + } + fmt.Printf("DEBUG [oauth/transport]: Added DPoP proof for %s %s (proof_length=%d, first_50=%q)\n", req.Method, req.URL.String(), len(proofString), proofPreview) return nil } diff --git a/pkg/middleware/registry.go b/pkg/middleware/registry.go index 8650858..03ac972 100644 --- a/pkg/middleware/registry.go +++ b/pkg/middleware/registry.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "sync" "github.com/distribution/distribution/v3" registrymw "github.com/distribution/distribution/v3/registry/middleware/registry" @@ -35,6 +36,7 @@ type NamespaceResolver struct { distribution.Namespace resolver *atproto.Resolver defaultStorageEndpoint string + repositories sync.Map // Cache of RoutingRepository instances by key (did:reponame) } // initATProtoResolver initializes the name resolution middleware @@ -115,11 +117,10 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name if globalRefresher != nil { // Try OAuth flow first - accessToken, dpopKey, err := globalRefresher.GetAccessToken(ctx, did) + accessToken, dpopKey, dpopTransport, err := globalRefresher.GetAccessToken(ctx, did) if err == nil { - // OAuth token available - create client with DPoP support - fmt.Printf("DEBUG [registry/middleware]: Using OAuth access token for DID=%s\n", did) - dpopTransport := oauth.NewDPoPTransport(nil, dpopKey) + // OAuth token available - use cached DPoP transport (preserves nonce) + fmt.Printf("DEBUG [registry/middleware]: Using OAuth access token for DID=%s (length=%d, first_20=%q)\n", did, len(accessToken), accessToken[:min(20, len(accessToken))]) atprotoClient = atproto.NewClientWithDPoP(pdsEndpoint, did, accessToken, dpopKey, dpopTransport) } else { fmt.Printf("DEBUG [registry/middleware]: OAuth refresh failed for DID=%s: %v, falling back to Basic Auth\n", did, err) @@ -143,13 +144,25 @@ func (nr *NamespaceResolver) Repository(ctx context.Context, name reference.Name // Example: "evan.jarrett.net/debian" -> store as "debian" repositoryName := imageName - fmt.Printf("DEBUG [registry/middleware]: Creating RoutingRepository for image=%s (ATProto repo name)\n", repositoryName) + // Cache key is DID + repository name + cacheKey := did + ":" + repositoryName + + // Check cache first + if cached, ok := nr.repositories.Load(cacheKey); ok { + fmt.Printf("DEBUG [registry/middleware]: Using cached RoutingRepository for %s\n", cacheKey) + return cached.(*storage.RoutingRepository), nil + } + + fmt.Printf("DEBUG [registry/middleware]: Creating new RoutingRepository for image=%s (ATProto repo name)\n", repositoryName) // Create routing repository - routes manifests to ATProto, blobs to hold service // The registry is stateless - no local storage is used // Pass storage endpoint and DID as parameters (can't use context as it gets lost) routingRepo := storage.NewRoutingRepository(repo, atprotoClient, repositoryName, storageEndpoint, did) + // Cache the repository + nr.repositories.Store(cacheKey, routingRepo) + return routingRepo, nil } diff --git a/pkg/storage/proxy_blob_store.go b/pkg/storage/proxy_blob_store.go index cd8d1b9..9a2cd14 100644 --- a/pkg/storage/proxy_blob_store.go +++ b/pkg/storage/proxy_blob_store.go @@ -14,6 +14,12 @@ import ( "github.com/opencontainers/go-digest" ) +const ( + // maxChunkSize is the maximum buffer size before flushing to hold service + // Matches S3's minimum multipart upload size + maxChunkSize = 5 * 1024 * 1024 // 5MB +) + // Global upload tracking (shared across all ProxyBlobStore instances) // This is necessary because distribution creates new repository/blob store instances per request var ( @@ -35,6 +41,13 @@ func NewProxyBlobStore(storageEndpoint, did string) *ProxyBlobStore { storageEndpoint: storageEndpoint, httpClient: &http.Client{ Timeout: 5 * time.Minute, // Timeout for presigned URL requests and uploads + Transport: &http.Transport{ + DisableKeepAlives: false, // Re-enable keep-alive + MaxIdleConns: 100, + MaxIdleConnsPerHost: 100, + MaxConnsPerHost: 0, // unlimited + IdleConnTimeout: 90 * time.Second, + }, }, did: did, } @@ -42,19 +55,33 @@ func NewProxyBlobStore(storageEndpoint, did string) *ProxyBlobStore { // Stat returns the descriptor for a blob func (p *ProxyBlobStore) Stat(ctx context.Context, dgst digest.Digest) (distribution.Descriptor, error) { - // For simplicity, we'll just check if we can get a download URL - // In production, you'd want a dedicated stat endpoint - url, err := p.getDownloadURL(ctx, dgst) + // Quick HEAD request to hold service to check if blob exists + url := fmt.Sprintf("%s/blobs/%s?did=%s", p.storageEndpoint, dgst.String(), p.did) + req, err := http.NewRequestWithContext(ctx, "HEAD", url, nil) if err != nil { return distribution.Descriptor{}, distribution.ErrBlobUnknown } - // We don't have size info from the storage service - // Return a minimal descriptor + resp, err := p.httpClient.Do(req) + if err != nil { + return distribution.Descriptor{}, distribution.ErrBlobUnknown + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return distribution.Descriptor{}, distribution.ErrBlobUnknown + } + + // Return a minimal descriptor with size from Content-Length if available + size := int64(0) + if contentLength := resp.Header.Get("Content-Length"); contentLength != "" { + fmt.Sscanf(contentLength, "%d", &size) + } + return distribution.Descriptor{ Digest: dgst, + Size: size, MediaType: "application/octet-stream", - URLs: []string{url}, }, nil } @@ -167,13 +194,21 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo } } - // Create proxy blob writer + // Create pipe for streaming upload + pipeReader, pipeWriter := io.Pipe() + uploadErr := make(chan error, 1) + digestChan := make(chan string, 1) + + // Create writer writer := &ProxyBlobWriter{ - store: p, - ctx: ctx, - options: opts, - id: fmt.Sprintf("upload-%d", time.Now().UnixNano()), - startedAt: time.Now(), + store: p, + options: opts, + pipeWriter: pipeWriter, + pipeReader: pipeReader, + digestChan: digestChan, + uploadErr: uploadErr, + id: fmt.Sprintf("upload-%d", time.Now().UnixNano()), + startedAt: time.Now(), } // Store in global uploads map for resume support @@ -181,6 +216,65 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo globalUploads[writer.id] = writer globalUploadsMu.Unlock() + // Start background goroutine that streams to temp location immediately + go func() { + defer pipeReader.Close() + + // Stream to temp location immediately to avoid pipe deadlock + tempPath := fmt.Sprintf("uploads/temp-%s", writer.id) // No leading slash + url := fmt.Sprintf("%s/blobs/%s?did=%s", p.storageEndpoint, tempPath, p.did) + + fmt.Printf("DEBUG [goroutine]: Starting upload to temp: url=%s\n", url) + + // Use context with timeout to prevent hanging forever + uploadCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + + req, err := http.NewRequestWithContext(uploadCtx, "PUT", url, pipeReader) + if err != nil { + fmt.Printf("DEBUG [goroutine]: Failed to create request: %v\n", err) + // Consume digest channel even on error + <-digestChan + uploadErr <- fmt.Errorf("failed to create request: %w", err) + return + } + req.Header.Set("Content-Type", "application/octet-stream") + + fmt.Printf("DEBUG [goroutine]: Sending PUT request...\n") + // Stream to temp location (this will block until all data is written) + resp, err := p.httpClient.Do(req) + if err != nil { + fmt.Printf("DEBUG [goroutine]: PUT failed: %v\n", err) + <-digestChan + uploadErr <- fmt.Errorf("failed to upload to temp: %w", err) + return + } + defer resp.Body.Close() + + fmt.Printf("DEBUG [goroutine]: Got response status=%d\n", resp.StatusCode) + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + bodyBytes, _ := io.ReadAll(resp.Body) + fmt.Printf("DEBUG [goroutine]: Upload failed with status %d, body=%s\n", resp.StatusCode, string(bodyBytes)) + <-digestChan + uploadErr <- fmt.Errorf("upload to temp failed: status %d, body: %s", resp.StatusCode, string(bodyBytes)) + return + } + + fmt.Printf("DEBUG [goroutine]: Upload to temp succeeded, waiting for digest...\n") + // Upload to temp succeeded, now wait for digest from Commit() + digest, ok := <-digestChan + if !ok { + uploadErr <- fmt.Errorf("upload cancelled after streaming to temp") + return + } + + fmt.Printf("DEBUG [goroutine]: Got digest=%s, signaling completion\n", digest) + // Store digest for Commit() to use in move operation + writer.finalDigest = digest + uploadErr <- nil + }() + return writer, nil } @@ -195,6 +289,7 @@ func (p *ProxyBlobStore) Resume(ctx context.Context, id string) (distribution.Bl return nil, distribution.ErrBlobUploadUnknown } + // With streaming, no flush needed - just return the writer return writer, nil } @@ -283,14 +378,17 @@ func (p *ProxyBlobStore) getUploadURL(ctx context.Context, dgst digest.Digest, s // ProxyBlobWriter implements distribution.BlobWriter for proxy uploads type ProxyBlobWriter struct { - store *ProxyBlobStore - ctx context.Context - options distribution.CreateOptions - buffer bytes.Buffer - size int64 - closed bool - id string - startedAt time.Time + store *ProxyBlobStore + options distribution.CreateOptions + pipeWriter *io.PipeWriter // Streams directly to hold service + pipeReader *io.PipeReader + digestChan chan string // Sends digest to upload goroutine + uploadErr chan error // Receives upload result from goroutine + finalDigest string // Final digest for move operation + size int64 + closed bool + id string // Distribution's upload ID + startedAt time.Time } // ID returns the upload ID @@ -304,13 +402,22 @@ func (w *ProxyBlobWriter) StartedAt() time.Time { } // Write writes data to the upload +// Streams directly to hold service via pipe func (w *ProxyBlobWriter) Write(p []byte) (int, error) { if w.closed { return 0, fmt.Errorf("writer closed") } - n, err := w.buffer.Write(p) + + // Write to pipe - streams immediately to hold service + n, err := w.pipeWriter.Write(p) + if err != nil { + // If write fails (client disconnected), close pipe to unblock goroutine + w.pipeWriter.CloseWithError(err) + return n, err + } w.size += int64(n) - return n, err + + return n, nil } // ReadFrom reads from a reader @@ -318,9 +425,29 @@ func (w *ProxyBlobWriter) ReadFrom(r io.Reader) (int64, error) { if w.closed { return 0, fmt.Errorf("writer closed") } - n, err := w.buffer.ReadFrom(r) - w.size += n - return n, err + + // Read in chunks and flush when needed + buf := make([]byte, 32*1024) // 32KB read buffer + var total int64 + + for { + nr, err := r.Read(buf) + if nr > 0 { + nw, werr := w.Write(buf[:nr]) + total += int64(nw) + if werr != nil { + return total, werr + } + } + if err == io.EOF { + break + } + if err != nil { + return total, err + } + } + + return total, nil } // Size returns the current size @@ -340,41 +467,48 @@ func (w *ProxyBlobWriter) Commit(ctx context.Context, desc distribution.Descript delete(globalUploads, w.id) globalUploadsMu.Unlock() - // Upload the buffered content - content := w.buffer.Bytes() - dgst := digest.FromBytes(content) - - // Verify digest matches - if desc.Digest != "" && dgst != desc.Digest { - return distribution.Descriptor{}, fmt.Errorf("digest mismatch") + // Close pipe to signal EOF to upload goroutine + if err := w.pipeWriter.Close(); err != nil { + return distribution.Descriptor{}, fmt.Errorf("failed to close pipe: %w", err) } - // Get upload URL - url, err := w.store.getUploadURL(ctx, dgst, int64(len(content))) + // Send digest to upload goroutine (it's waiting after temp upload completes) + w.digestChan <- desc.Digest.String() + close(w.digestChan) + + // Wait for upload goroutine to complete + if err := <-w.uploadErr; err != nil { + return distribution.Descriptor{}, fmt.Errorf("upload to temp failed: %w", err) + } + + // Now move temp → final location + tempPath := fmt.Sprintf("uploads/temp-%s", w.id) // No leading slash + finalPath := desc.Digest.String() + + moveURL := fmt.Sprintf("%s/move?from=%s&to=%s&did=%s", + w.store.storageEndpoint, tempPath, finalPath, w.store.did) + + req, err := http.NewRequestWithContext(context.Background(), "POST", moveURL, nil) if err != nil { - return distribution.Descriptor{}, err + return distribution.Descriptor{}, fmt.Errorf("failed to create move request: %w", err) } - // Upload - req, err := http.NewRequestWithContext(ctx, "PUT", url, bytes.NewReader(content)) - if err != nil { - return distribution.Descriptor{}, err - } - req.Header.Set("Content-Type", "application/octet-stream") - resp, err := w.store.httpClient.Do(req) if err != nil { - return distribution.Descriptor{}, err + return distribution.Descriptor{}, fmt.Errorf("failed to move blob: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { - return distribution.Descriptor{}, fmt.Errorf("upload failed: status %d", resp.StatusCode) + bodyBytes, _ := io.ReadAll(resp.Body) + return distribution.Descriptor{}, fmt.Errorf("move blob failed: status %d, body: %s", resp.StatusCode, string(bodyBytes)) } + fmt.Printf("DEBUG [proxy_blob_store]: Committed upload: digest=%s, size=%d (moved from temp)\n", desc.Digest, w.size) + return distribution.Descriptor{ - Digest: dgst, - Size: int64(len(content)), + Digest: desc.Digest, + Size: w.size, MediaType: desc.MediaType, }, nil } @@ -388,15 +522,26 @@ func (w *ProxyBlobWriter) Cancel(ctx context.Context) error { delete(globalUploads, w.id) globalUploadsMu.Unlock() + // Close digest channel without sending digest + close(w.digestChan) + + // Close pipe with error to stop streaming + if w.pipeWriter != nil { + w.pipeWriter.CloseWithError(fmt.Errorf("upload cancelled")) + } + + // Wait for goroutine to finish + <-w.uploadErr + + fmt.Printf("DEBUG [proxy_blob_store]: Cancelled upload: id=%s\n", w.id) return nil } // Close closes the writer -// NOTE: For resumable uploads, we don't mark as closed here -// Distribution calls Close() after each PATCH, but the upload may continue -// Only Commit() and Cancel() actually finalize the upload +// Just returns - streaming continues via pipe func (w *ProxyBlobWriter) Close() error { - // Don't set w.closed = true here - allow resuming + // Don't close pipe here - that happens in Commit() or Cancel() + // Don't set w.closed = true - allow resuming for next PATCH return nil } diff --git a/pkg/storage/routing_repository.go b/pkg/storage/routing_repository.go index 6aaf1f8..bbf4242 100644 --- a/pkg/storage/routing_repository.go +++ b/pkg/storage/routing_repository.go @@ -18,6 +18,7 @@ type RoutingRepository struct { storageEndpoint string // Hold service endpoint for blobs (from discovery for push) did string // User's DID for authorization manifestStore *atproto.ManifestStore // Cached manifest store instance + blobStore *ProxyBlobStore // Cached blob store instance } // NewRoutingRepository creates a new routing repository @@ -62,6 +63,13 @@ func (r *RoutingRepository) Manifests(ctx context.Context, options ...distributi // Blobs returns a proxy blob store that routes to external hold service // The registry (AppView) NEVER stores blobs locally - all blobs go through hold service 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) + return r.blobStore + } + // For pull operations, check if we have a cached hold endpoint from a recent manifest fetch // This ensures blobs are fetched from the hold recorded in the manifest, not re-discovered holdEndpoint := r.storageEndpoint // Default to discovery-based endpoint @@ -82,8 +90,9 @@ func (r *RoutingRepository) Blobs(ctx context.Context) distribution.BlobStore { panic("storage endpoint not set in RoutingRepository - ensure default_storage_endpoint is configured in middleware") } - // Always use proxy blob store - routes to external hold service - return NewProxyBlobStore(holdEndpoint, r.did) + // Create and cache proxy blob store + r.blobStore = NewProxyBlobStore(holdEndpoint, r.did) + return r.blobStore } // Tags returns the tag service