From f8f475a65d2a6555a81d1f69779dd8a0f9926ae5 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Mon, 13 Jul 2026 16:17:52 -0700 Subject: [PATCH 1/3] upstream: apply authentication through shared transport --- config.example.yaml | 3 +- docs/architecture.md | 2 + docs/configuration.md | 6 +- internal/config/config.go | 51 +++- internal/config/config_test.go | 42 +++ internal/httpclient/transport.go | 418 ++++++++++++++++++++++++++ internal/httpclient/transport_test.go | 151 ++++++++++ internal/server/server.go | 32 +- 8 files changed, 689 insertions(+), 16 deletions(-) create mode 100644 internal/httpclient/transport.go create mode 100644 internal/httpclient/transport_test.go diff --git a/config.example.yaml b/config.example.yaml index d1ed9a1..2124e48 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -96,7 +96,8 @@ upstream: cargo_download: "https://static.crates.io/crates" # Authentication for upstream registries - # Keys are URL prefixes matched against request URLs. + # Keys are absolute URL scopes. Scheme, host, effective port, and path + # segment boundaries must match; the longest matching scope wins. # Values can reference environment variables using ${VAR_NAME} syntax. # # Supported auth types: diff --git a/docs/architecture.md b/docs/architecture.md index f04d548..cf8b0e2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -240,6 +240,8 @@ Fetches artifacts from upstream registries. - Exponential backoff retry on 429 (rate limit) and 5xx errors - Returns streaming reader (doesn't load into memory) - Configurable user-agent +- Shares an authentication-aware transport with metadata requests so URL-scoped credentials apply consistently +- Discovers and caches scoped OCI Bearer tokens from registry challenges **Resolver:** - Determines download URL for a package/version diff --git a/docs/configuration.md b/docs/configuration.md index dcdec24..02d426f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -123,7 +123,9 @@ upstream: ## Authentication -Configure authentication for private upstream registries. Auth is matched by URL prefix, and credentials can reference environment variables using `${VAR_NAME}` syntax. +Configure authentication for private upstream registries. The same authentication-aware client is used for metadata and artifact downloads, and credentials can reference environment variables using `${VAR_NAME}` syntax. + +OCI registries that return a Bearer challenge from a `/v2/{repository}/…` endpoint are handled automatically. The proxy discovers the token realm from `WWW-Authenticate`, applies any configured credentials for the token URL, and reuses the scoped token until shortly before it expires. ### Bearer Token @@ -172,7 +174,7 @@ upstream: ### URL Matching -Auth configs are matched by URL prefix. The longest matching prefix wins, so you can configure different credentials for different paths: +Auth keys must be absolute URLs. Matching compares the scheme, host, effective port, and path-segment prefix, preventing credentials for `registry.example.com` from being sent to a lookalike host such as `registry.example.com.evil.test`. The longest matching scope wins, so you can configure different credentials for different paths: ```yaml upstream: diff --git a/internal/config/config.go b/internal/config/config.go index 2d275c7..a54d50c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -274,23 +274,29 @@ type UpstreamConfig struct { CargoDownload string `json:"cargo_download" yaml:"cargo_download"` // Auth configures authentication for upstream registries. - // Keys are URL prefixes that are matched against request URLs. + // Keys are absolute URL scopes matched by scheme, host, effective port, + // and path-segment prefix. // Example: "https://npm.pkg.github.com" matches all requests to that host. Auth map[string]AuthConfig `json:"auth" yaml:"auth"` } // AuthForURL returns the auth config that matches the given URL. -// Matches are based on URL prefix - the longest matching prefix wins. +// The longest matching URL scope wins. func (u *UpstreamConfig) AuthForURL(url string) *AuthConfig { if u.Auth == nil { return nil } + target, err := parseAuthURL(url) + if err != nil { + return nil + } var bestMatch *AuthConfig var bestLen int for pattern, auth := range u.Auth { - if strings.HasPrefix(url, pattern) && len(pattern) > bestLen { + configured, err := parseAuthURL(pattern) + if err == nil && authURLMatches(configured, target) && len(pattern) > bestLen { a := auth // copy to avoid loop variable capture bestMatch = &a bestLen = len(pattern) @@ -300,6 +306,45 @@ func (u *UpstreamConfig) AuthForURL(url string) *AuthConfig { return bestMatch } +func parseAuthURL(value string) (*url.URL, error) { + parsed, err := url.Parse(value) + if err != nil || !parsed.IsAbs() || parsed.Hostname() == "" || parsed.Opaque != "" { + return nil, fmt.Errorf("invalid authentication URL") + } + return parsed, nil +} + +func authURLMatches(configured, target *url.URL) bool { + if !strings.EqualFold(configured.Scheme, target.Scheme) || + !strings.EqualFold(configured.Hostname(), target.Hostname()) || + authURLPort(configured) != authURLPort(target) { + return false + } + if configured.RawQuery != "" && configured.RawQuery != target.RawQuery { + return false + } + + configuredPath := strings.TrimSuffix(configured.EscapedPath(), "/") + if configuredPath == "" { + return true + } + targetPath := strings.TrimSuffix(target.EscapedPath(), "/") + return targetPath == configuredPath || strings.HasPrefix(targetPath, configuredPath+"/") +} + +func authURLPort(value *url.URL) string { + if port := value.Port(); port != "" { + return port + } + if strings.EqualFold(value.Scheme, "https") { + return "443" + } + if strings.EqualFold(value.Scheme, "http") { + return "80" + } + return "" +} + // AuthConfig configures authentication for an upstream registry. type AuthConfig struct { // Type is the authentication type: "bearer", "basic", or "header". diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9c34023..63520a2 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -760,3 +760,45 @@ func TestDatabaseConfigString(t *testing.T) { } } } + +func TestUpstreamAuthForURLMatchesURLComponents(t *testing.T) { + registryAuth := AuthConfig{Type: "bearer", Token: "registry-token"} + privateAuth := AuthConfig{Type: "bearer", Token: "private-token"} + config := UpstreamConfig{Auth: map[string]AuthConfig{ + "https://registry.example.com": registryAuth, + "https://registry.example.com/private": privateAuth, + }} + + tests := []struct { + name string + url string + wantToken string + }{ + {name: "registry root", url: "https://registry.example.com/package", wantToken: "registry-token"}, + {name: "host is case insensitive", url: "https://REGISTRY.EXAMPLE.COM/package", wantToken: "registry-token"}, + {name: "longest path match", url: "https://registry.example.com/private/package", wantToken: "private-token"}, + {name: "exact path match", url: "https://registry.example.com/private", wantToken: "private-token"}, + {name: "path segment boundary", url: "https://registry.example.com/private-other/package", wantToken: "registry-token"}, + {name: "lookalike host rejected", url: "https://registry.example.com.evil.test/package"}, + {name: "different scheme rejected", url: "http://registry.example.com/package"}, + {name: "different port rejected", url: "https://registry.example.com:8443/package"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + auth := config.AuthForURL(tt.url) + if tt.wantToken == "" { + if auth != nil { + t.Fatalf("AuthForURL() = %+v, want nil", auth) + } + return + } + if auth == nil { + t.Fatal("AuthForURL() = nil, want authentication") + } + if auth.Token != tt.wantToken { + t.Errorf("token = %q, want %q", auth.Token, tt.wantToken) + } + }) + } +} diff --git a/internal/httpclient/transport.go b/internal/httpclient/transport.go new file mode 100644 index 0000000..1cdb457 --- /dev/null +++ b/internal/httpclient/transport.go @@ -0,0 +1,418 @@ +// Package httpclient provides authentication-aware HTTP transports for upstream requests. +package httpclient + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" +) + +const ( + defaultTokenLifetime = 60 * time.Second + tokenExpirySkew = 5 * time.Second + maxTokenResponseSize = 1 << 20 + shortTokenSkewDivisor = 10 +) + +// AuthFunc returns a configured authentication header for a URL. +type AuthFunc func(url string) (headerName, headerValue string) + +// Transport adds configured authentication and follows OCI Bearer challenges. +type Transport struct { + base http.RoundTripper + authForURL AuthFunc + + mu sync.Mutex + tokens map[string]cachedToken + challenges map[string]bearerChallenge +} + +type cachedToken struct { + value string + expiresAt time.Time +} + +type bearerChallenge struct { + realm string + service string + scopes []string +} + +type tokenResponse struct { + Token string `json:"token"` + AccessToken string `json:"access_token"` + ExpiresIn int64 `json:"expires_in"` + IssuedAt string `json:"issued_at"` +} + +// NewTransport creates an authentication-aware transport around base. +func NewTransport(base http.RoundTripper, authForURL AuthFunc) *Transport { + if base == nil { + base = http.DefaultTransport + } + return &Transport{ + base: base, + authForURL: authForURL, + tokens: make(map[string]cachedToken), + challenges: make(map[string]bearerChallenge), + } +} + +// RoundTrip implements http.RoundTripper. +func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) { + outbound := cloneRequest(req) + t.applyAuthentication(outbound, req.Header.Get("Authorization") != "") + + resp, err := t.base.RoundTrip(outbound) + if err != nil || resp.StatusCode != http.StatusUnauthorized { + return resp, err + } + if registryProtectionSpace(req.URL) == "" { + return resp, nil + } + + challenge, ok := parseBearerChallenge(resp.Header.Values("WWW-Authenticate")) + if !ok || !canReplay(req) { + return resp, nil + } + + drainAndClose(resp.Body) + token, err := t.token(req.Context(), challenge) + if err != nil { + return nil, fmt.Errorf("registry authentication: %w", err) + } + t.rememberChallenge(req.URL, challenge) + + retry, err := cloneRequestForRetry(req) + if err != nil { + return nil, err + } + t.applyConfiguredAuthentication(retry) + retry.Header.Set("Authorization", "Bearer "+token) + return t.base.RoundTrip(retry) +} + +func (t *Transport) applyAuthentication(req *http.Request, hasExplicitAuthorization bool) { + t.applyConfiguredAuthentication(req) + if hasExplicitAuthorization { + return + } + if token := t.cachedTokenForRequest(req.URL); token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } +} + +func (t *Transport) applyConfiguredAuthentication(req *http.Request) { + if t.authForURL == nil { + return + } + name, value := t.authForURL(req.URL.String()) + if name != "" && value != "" && req.Header.Get(name) == "" { + req.Header.Set(name, value) + } +} + +func (t *Transport) token(ctx context.Context, challenge bearerChallenge) (string, error) { + key := challenge.key() + if token := t.cachedToken(key); token != "" { + return token, nil + } + + token, expiresAt, err := t.fetchToken(ctx, challenge) + if err != nil { + return "", err + } + + t.mu.Lock() + t.tokens[key] = cachedToken{value: token, expiresAt: expiresAt} + t.mu.Unlock() + return token, nil +} + +func (t *Transport) fetchToken(ctx context.Context, challenge bearerChallenge) (string, time.Time, error) { + tokenURL, err := url.Parse(challenge.realm) + if err != nil || !tokenURL.IsAbs() || (tokenURL.Scheme != "https" && tokenURL.Scheme != "http") { + return "", time.Time{}, fmt.Errorf("invalid token realm %q", challenge.realm) + } + + query := tokenURL.Query() + if challenge.service != "" { + query.Set("service", challenge.service) + } + for _, scope := range challenge.scopes { + query.Add("scope", scope) + } + query.Set("client_id", "git-pkgs-proxy") + tokenURL.RawQuery = query.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, tokenURL.String(), nil) + if err != nil { + return "", time.Time{}, err + } + + client := &http.Client{Transport: configuredTransport{parent: t}} + resp, err := client.Do(req) + if err != nil { + return "", time.Time{}, fmt.Errorf("requesting token: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxTokenResponseSize)) + return "", time.Time{}, fmt.Errorf("token service returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var payload tokenResponse + if err := json.NewDecoder(io.LimitReader(resp.Body, maxTokenResponseSize)).Decode(&payload); err != nil { + return "", time.Time{}, fmt.Errorf("decoding token response: %w", err) + } + token := payload.Token + if token == "" { + token = payload.AccessToken + } + if token == "" { + return "", time.Time{}, fmt.Errorf("token response did not contain a token") + } + + issuedAt := time.Now() + if payload.IssuedAt != "" { + if parsed, parseErr := time.Parse(time.RFC3339, payload.IssuedAt); parseErr == nil { + issuedAt = parsed + } + } + lifetime := time.Duration(payload.ExpiresIn) * time.Second + if lifetime <= 0 { + lifetime = defaultTokenLifetime + } + expiresAt := issuedAt.Add(lifetime).Add(-expirySkew(lifetime)) + return token, expiresAt, nil +} + +type configuredTransport struct { + parent *Transport +} + +func (t configuredTransport) RoundTrip(req *http.Request) (*http.Response, error) { + outbound := cloneRequest(req) + t.parent.applyConfiguredAuthentication(outbound) + return t.parent.base.RoundTrip(outbound) +} + +func (t *Transport) cachedTokenForRequest(requestURL *url.URL) string { + space := registryProtectionSpace(requestURL) + if space == "" { + return "" + } + + t.mu.Lock() + challenge, ok := t.challenges[space] + t.mu.Unlock() + if !ok { + return "" + } + return t.cachedToken(challenge.key()) +} + +func (t *Transport) cachedToken(key string) string { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + + token, ok := t.tokens[key] + if !ok { + return "" + } + if !now.Before(token.expiresAt) { + delete(t.tokens, key) + return "" + } + return token.value +} + +func (t *Transport) rememberChallenge(requestURL *url.URL, challenge bearerChallenge) { + space := registryProtectionSpace(requestURL) + if space == "" { + return + } + t.mu.Lock() + t.challenges[space] = challenge + t.mu.Unlock() +} + +func (c bearerChallenge) key() string { + return c.realm + "\x00" + c.service + "\x00" + strings.Join(c.scopes, "\x00") +} + +func registryProtectionSpace(u *url.URL) string { + const registryPrefix = "/v2/" + if u == nil || !strings.HasPrefix(u.Path, registryPrefix) { + return "" + } + rest := strings.TrimPrefix(u.Path, registryPrefix) + end := len(rest) + for _, marker := range []string{"/blobs/", "/manifests/", "/tags/", "/referrers/"} { + if index := strings.Index(rest, marker); index >= 0 && index < end { + end = index + } + } + if end == len(rest) || end == 0 { + return "" + } + return u.Scheme + "://" + u.Host + registryPrefix + rest[:end] +} + +func parseBearerChallenge(values []string) (bearerChallenge, bool) { + for _, value := range values { + params, ok := bearerParameters(value) + if !ok || params["realm"] == "" { + continue + } + challenge := bearerChallenge{ + realm: params["realm"], + service: params["service"], + } + if scope := params["scope"]; scope != "" { + challenge.scopes = append(challenge.scopes, scope) + } + return challenge, true + } + return bearerChallenge{}, false +} + +func bearerParameters(value string) (map[string]string, bool) { + start := findAuthScheme(value, "Bearer") + if start < 0 { + return nil, false + } + rest := value[start+len("Bearer"):] + params := make(map[string]string) + for { + rest = strings.TrimLeft(rest, " \t,") + if rest == "" { + break + } + + keyEnd := strings.IndexAny(rest, "= \t,") + if keyEnd <= 0 { + break + } + key := strings.ToLower(rest[:keyEnd]) + rest = strings.TrimLeft(rest[keyEnd:], " \t") + if rest == "" || rest[0] != '=' { + break + } + rest = strings.TrimLeft(rest[1:], " \t") + + parsed, remaining, ok := parseAuthValue(rest) + if !ok { + return nil, false + } + params[key] = parsed + rest = remaining + } + return params, true +} + +func findAuthScheme(value, scheme string) int { + inQuote := false + escaped := false + for index := 0; index+len(scheme) <= len(value); index++ { + char := value[index] + if escaped { + escaped = false + continue + } + if char == '\\' && inQuote { + escaped = true + continue + } + if char == '"' { + inQuote = !inQuote + continue + } + if inQuote || !strings.EqualFold(value[index:index+len(scheme)], scheme) { + continue + } + beforeOK := index == 0 || value[index-1] == ',' || value[index-1] == ' ' || value[index-1] == '\t' + after := index + len(scheme) + afterOK := after < len(value) && (value[after] == ' ' || value[after] == '\t') + if beforeOK && afterOK { + return index + } + } + return -1 +} + +func parseAuthValue(value string) (parsed, remaining string, ok bool) { + if value == "" { + return "", "", false + } + if value[0] != '"' { + end := strings.IndexAny(value, " \t,") + if end < 0 { + return value, "", true + } + return value[:end], value[end:], end > 0 + } + + var builder strings.Builder + escaped := false + for index := 1; index < len(value); index++ { + char := value[index] + if escaped { + builder.WriteByte(char) + escaped = false + continue + } + if char == '\\' { + escaped = true + continue + } + if char == '"' { + return builder.String(), value[index+1:], true + } + builder.WriteByte(char) + } + return "", "", false +} + +func cloneRequest(req *http.Request) *http.Request { + clone := req.Clone(req.Context()) + clone.Header = req.Header.Clone() + return clone +} + +func canReplay(req *http.Request) bool { + return req.Body == nil || req.GetBody != nil +} + +func cloneRequestForRetry(req *http.Request) (*http.Request, error) { + clone := cloneRequest(req) + if req.Body == nil { + return clone, nil + } + body, err := req.GetBody() + if err != nil { + return nil, fmt.Errorf("replaying authenticated request: %w", err) + } + clone.Body = body + return clone, nil +} + +func expirySkew(lifetime time.Duration) time.Duration { + if lifetime < tokenExpirySkew*2 { + return lifetime / shortTokenSkewDivisor + } + return tokenExpirySkew +} + +func drainAndClose(body io.ReadCloser) { + _, _ = io.Copy(io.Discard, io.LimitReader(body, maxTokenResponseSize)) + _ = body.Close() +} diff --git a/internal/httpclient/transport_test.go b/internal/httpclient/transport_test.go new file mode 100644 index 0000000..027a378 --- /dev/null +++ b/internal/httpclient/transport_test.go @@ -0,0 +1,151 @@ +package httpclient + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestTransportFollowsBearerChallengeAndCachesToken(t *testing.T) { + var registryRequests int + var tokenRequests int + var server *httptest.Server + + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/token": + tokenRequests++ + if got := r.URL.Query().Get("service"); got != "registry.test" { + t.Errorf("service = %q, want %q", got, "registry.test") + } + if got := r.URL.Query().Get("scope"); got != "repository:library/test:pull" { + t.Errorf("scope = %q, want %q", got, "repository:library/test:pull") + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"token":"registry-token","expires_in":3600}`) + case "/v2/library/test/blobs/sha256:first", "/v2/library/test/blobs/sha256:second": + registryRequests++ + if r.Header.Get("Authorization") != "Bearer registry-token" { + w.Header().Set("WWW-Authenticate", `Bearer realm="`+server.URL+`/token",service="registry.test",scope="repository:library/test:pull"`) + http.Error(w, "authentication required", http.StatusUnauthorized) + return + } + _, _ = io.WriteString(w, "blob") + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + client := &http.Client{Transport: NewTransport(http.DefaultTransport, nil)} + for _, digest := range []string{"sha256:first", "sha256:second"} { + resp, err := client.Get(server.URL + "/v2/library/test/blobs/" + digest) + if err != nil { + t.Fatalf("GET %s: %v", digest, err) + } + body, readErr := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if readErr != nil { + t.Fatalf("read %s response: %v", digest, readErr) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET %s status = %d, want %d", digest, resp.StatusCode, http.StatusOK) + } + if string(body) != "blob" { + t.Errorf("GET %s body = %q, want %q", digest, body, "blob") + } + } + + if tokenRequests != 1 { + t.Errorf("token requests = %d, want 1", tokenRequests) + } + if registryRequests != 3 { + t.Errorf("registry requests = %d, want 3", registryRequests) + } +} + +func TestTransportAddsConfiguredAuthentication(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("X-Registry-Token"); got != "configured-token" { + t.Errorf("X-Registry-Token = %q, want %q", got, "configured-token") + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + authForURL := func(url string) (string, string) { + if strings.HasPrefix(url, server.URL) { + return "X-Registry-Token", "configured-token" + } + return "", "" + } + client := &http.Client{Transport: NewTransport(http.DefaultTransport, authForURL)} + + resp, err := client.Get(server.URL + "/metadata") + if err != nil { + t.Fatalf("GET metadata: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusNoContent) + } +} + +func TestTransportPreservesExplicitAuthentication(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "Bearer explicit-token" { + t.Errorf("Authorization = %q, want %q", got, "Bearer explicit-token") + } + w.WriteHeader(http.StatusNoContent) + })) + defer server.Close() + + authForURL := func(string) (string, string) { + return "Authorization", "Bearer configured-token" + } + client := &http.Client{Transport: NewTransport(http.DefaultTransport, authForURL)} + req, err := http.NewRequest(http.MethodGet, server.URL+"/artifact", nil) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Authorization", "Bearer explicit-token") + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("GET artifact: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusNoContent) + } +} + +func TestTransportDoesNotFollowBearerChallengeOutsideOCIRegistry(t *testing.T) { + tokenRequests := 0 + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/token" { + tokenRequests++ + _, _ = io.WriteString(w, `{"token":"unexpected"}`) + return + } + w.Header().Set("WWW-Authenticate", `Bearer realm="`+server.URL+`/token"`) + http.Error(w, "authentication required", http.StatusUnauthorized) + })) + defer server.Close() + + client := &http.Client{Transport: NewTransport(http.DefaultTransport, nil)} + resp, err := client.Get(server.URL + "/api/packages") + if err != nil { + t.Fatalf("GET API: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusUnauthorized { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusUnauthorized) + } + if tokenRequests != 0 { + t.Errorf("token requests = %d, want 0", tokenRequests) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 5aac4db..b763278 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -58,17 +58,19 @@ import ( "strings" "time" + "github.com/git-pkgs/cooldown" swaggerdoc "github.com/git-pkgs/proxy/docs/swagger" "github.com/git-pkgs/proxy/internal/config" - "github.com/git-pkgs/cooldown" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/enrichment" "github.com/git-pkgs/proxy/internal/handler" + upstreamhttp "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/proxy/internal/mirror" "github.com/git-pkgs/proxy/internal/storage" "github.com/git-pkgs/purl" "github.com/git-pkgs/registries/fetch" + "github.com/git-pkgs/registries/safehttp" "github.com/git-pkgs/spdx" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" @@ -84,12 +86,12 @@ const ( // Server is the main proxy server. type Server struct { - cfg *config.Config - db *database.DB - storage storage.Storage - logger *slog.Logger - http *http.Server - templates *Templates + cfg *config.Config + db *database.DB + storage storage.Storage + logger *slog.Logger + http *http.Server + templates *Templates cancel context.CancelFunc healthCache *healthCache } @@ -156,8 +158,18 @@ func New(cfg *config.Config, logger *slog.Logger) (*Server, error) { // Start starts the HTTP server. func (s *Server) Start() error { - // Create shared components with circuit breaker - baseFetcher := fetch.NewFetcher(fetch.WithAuthFunc(s.authForURL)) + // Use one authentication-aware transport for metadata and artifacts so + // configured credentials and cached OCI challenges apply consistently. + safeClient := safehttp.New(nil, safehttp.Options{}) + authTransport := upstreamhttp.NewTransport(safeClient.Transport, upstreamhttp.AuthFunc(s.authForURL)) + metadataClient := *safeClient + metadataClient.Timeout = s.cfg.ParseHTTPTimeout() + metadataClient.Transport = authTransport + artifactClient := metadataClient + artifactClient.Timeout = serverWriteTimeout + + // Create shared components with circuit breaker. + baseFetcher := fetch.NewFetcher(fetch.WithHTTPClient(&artifactClient)) fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher) resolver := fetch.NewResolver() cd := &cooldown.Config{ @@ -166,7 +178,7 @@ func (s *Server) Start() error { Packages: s.cfg.Cooldown.Packages, } proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger) - proxy.HTTPClient.Timeout = s.cfg.ParseHTTPTimeout() + proxy.HTTPClient = &metadataClient proxy.Cooldown = cd proxy.CacheMetadata = s.cfg.CacheMetadata proxy.MetadataTTL = s.cfg.ParseMetadataTTL() From 0ee37c7521e492f03af7a25e66dcc676a920ac4b Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Mon, 13 Jul 2026 16:24:36 -0700 Subject: [PATCH 2/3] container: cache manifests for offline pulls --- docs/architecture.md | 1 + docs/configuration.md | 2 + internal/database/metadata_cache_test.go | 55 +++- internal/database/queries.go | 14 +- internal/database/schema.go | 19 ++ internal/database/types.go | 23 +- internal/handler/container.go | 151 +++------ internal/handler/container_manifest.go | 251 +++++++++++++++ internal/handler/container_test.go | 391 +++++++++++++++++++---- internal/handler/handler.go | 25 +- internal/handler/handler_test.go | 6 +- 11 files changed, 726 insertions(+), 212 deletions(-) create mode 100644 internal/handler/container_manifest.go diff --git a/docs/architecture.md b/docs/architecture.md index cf8b0e2..6d9bfda 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -353,6 +353,7 @@ Eviction can be implemented as: - Fresh data - new versions visible immediately - Metadata is small, upstream fetch is fast - Set `cache_metadata: true` or use the mirror command to enable metadata caching for offline use via the `metadata_cache` table +- OCI manifests are the exception: they are cached automatically so previously fetched images remain pullable when the registry or token service is unavailable **Why stream artifacts?** - Memory efficient - don't load large files into RAM diff --git a/docs/configuration.md b/docs/configuration.md index 02d426f..c73dc2c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -246,6 +246,8 @@ Note: Hex cooldown requires disabling registry signature verification since the By default the proxy fetches metadata fresh from upstream on every request. Enable `cache_metadata` to store metadata responses in the database and storage backend for offline fallback. When upstream is unreachable, the proxy serves the last cached copy. ETag-based revalidation avoids re-downloading unchanged metadata. +OCI manifests are always cached because cached image blobs cannot be pulled without their manifests. Digest-addressed manifests are immutable and served directly from cache. Tag-addressed manifests follow `metadata_ttl`, revalidate when stale, and fall back to the last cached response when the registry is unavailable. + ```yaml cache_metadata: true ``` diff --git a/internal/database/metadata_cache_test.go b/internal/database/metadata_cache_test.go index 5701816..09dcba3 100644 --- a/internal/database/metadata_cache_test.go +++ b/internal/database/metadata_cache_test.go @@ -30,8 +30,12 @@ func TestUpsertAndGetMetadataCache(t *testing.T) { StoragePath: "_metadata/npm/lodash/metadata", ETag: sql.NullString{String: `"abc123"`, Valid: true}, ContentType: sql.NullString{String: "application/json", Valid: true}, - Size: sql.NullInt64{Int64: 1024, Valid: true}, - FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, + ContentDigest: sql.NullString{ + String: "sha256:0123456789abcdef", + Valid: true, + }, + Size: sql.NullInt64{Int64: 1024, Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, } err := db.UpsertMetadataCache(entry) @@ -62,6 +66,9 @@ func TestUpsertAndGetMetadataCache(t *testing.T) { if !got.ContentType.Valid || got.ContentType.String != "application/json" { t.Errorf("content_type = %v, want %q", got.ContentType, "application/json") } + if !got.ContentDigest.Valid || got.ContentDigest.String != "sha256:0123456789abcdef" { + t.Errorf("content_digest = %v, want %q", got.ContentDigest, "sha256:0123456789abcdef") + } if !got.Size.Valid || got.Size.Int64 != 1024 { t.Errorf("size = %v, want 1024", got.Size) } @@ -178,3 +185,47 @@ func TestMetadataCacheTableCreatedByMigration(t *testing.T) { t.Error("metadata_cache table should exist after migration") } } + +func TestMetadataCacheContentDigestMigrationPreservesExistingRows(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "test.db") + db, err := Create(dbPath) + if err != nil { + t.Fatalf("Create failed: %v", err) + } + defer func() { _ = db.Close() }() + + if _, err := db.Exec("ALTER TABLE metadata_cache DROP COLUMN content_digest"); err != nil { + t.Fatalf("dropping content_digest: %v", err) + } + if _, err := db.Exec("DELETE FROM migrations WHERE name = ?", "006_add_metadata_content_digest"); err != nil { + t.Fatalf("resetting digest migration: %v", err) + } + if _, err := db.Exec(` + INSERT INTO metadata_cache (ecosystem, name, storage_path, content_type, size, fetched_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `, "oci-manifest", "cache-key", "_metadata/oci-manifest/cache-key/metadata", "application/json", 2, time.Now(), time.Now(), time.Now()); err != nil { + t.Fatalf("inserting legacy cache row: %v", err) + } + + if err := db.MigrateSchema(); err != nil { + t.Fatalf("MigrateSchema() error = %v", err) + } + hasDigest, err := db.HasColumn("metadata_cache", "content_digest") + if err != nil { + t.Fatalf("HasColumn() error = %v", err) + } + if !hasDigest { + t.Fatal("metadata_cache.content_digest was not added") + } + + entry, err := db.GetMetadataCache("oci-manifest", "cache-key") + if err != nil { + t.Fatalf("GetMetadataCache() error = %v", err) + } + if entry == nil || entry.StoragePath != "_metadata/oci-manifest/cache-key/metadata" { + t.Fatalf("existing metadata cache row was not preserved: %#v", entry) + } + if entry.ContentDigest.Valid { + t.Errorf("legacy content digest = %q, want NULL", entry.ContentDigest.String) + } +} diff --git a/internal/database/queries.go b/internal/database/queries.go index 5d95596..f8b76fe 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -894,7 +894,7 @@ func (db *DB) GetMetadataCache(ecosystem, name string) (*MetadataCacheEntry, err var entry MetadataCacheEntry query := db.Rebind(` SELECT id, ecosystem, name, storage_path, etag, content_type, - size, last_modified, fetched_at, created_at, updated_at + content_digest, size, last_modified, fetched_at, created_at, updated_at FROM metadata_cache WHERE ecosystem = ? AND name = ? `) err := db.Get(&entry, query, ecosystem, name) @@ -914,12 +914,13 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { if db.dialect == DialectPostgres { query = ` INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, content_type, - size, last_modified, fetched_at, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) + content_digest, size, last_modified, fetched_at, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT(ecosystem, name) DO UPDATE SET storage_path = EXCLUDED.storage_path, etag = EXCLUDED.etag, content_type = EXCLUDED.content_type, + content_digest = EXCLUDED.content_digest, size = EXCLUDED.size, last_modified = EXCLUDED.last_modified, fetched_at = EXCLUDED.fetched_at, @@ -928,12 +929,13 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { } else { query = ` INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, content_type, - size, last_modified, fetched_at, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + content_digest, size, last_modified, fetched_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(ecosystem, name) DO UPDATE SET storage_path = excluded.storage_path, etag = excluded.etag, content_type = excluded.content_type, + content_digest = excluded.content_digest, size = excluded.size, last_modified = excluded.last_modified, fetched_at = excluded.fetched_at, @@ -943,7 +945,7 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error { _, err := db.Exec(query, entry.Ecosystem, entry.Name, entry.StoragePath, entry.ETag, - entry.ContentType, entry.Size, entry.LastModified, entry.FetchedAt, now, now, + entry.ContentType, entry.ContentDigest, entry.Size, entry.LastModified, entry.FetchedAt, now, now, ) if err != nil { return fmt.Errorf("upserting metadata cache: %w", err) diff --git a/internal/database/schema.go b/internal/database/schema.go index c8d8d1e..c73877d 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -102,6 +102,7 @@ CREATE TABLE IF NOT EXISTS metadata_cache ( storage_path TEXT NOT NULL, etag TEXT, content_type TEXT, + content_digest TEXT, size INTEGER, last_modified DATETIME, fetched_at DATETIME, @@ -202,6 +203,7 @@ CREATE TABLE IF NOT EXISTS metadata_cache ( storage_path TEXT NOT NULL, etag TEXT, content_type TEXT, + content_digest TEXT, size BIGINT, last_modified TIMESTAMP, fetched_at TIMESTAMP, @@ -359,6 +361,7 @@ var migrations = []migration{ {"003_ensure_artifacts_table", migrateEnsureArtifactsTable}, {"004_ensure_vulnerabilities_table", migrateEnsureVulnerabilitiesTable}, {"005_ensure_metadata_cache_table", migrateEnsureMetadataCacheTable}, + {"006_add_metadata_content_digest", migrateAddMetadataContentDigest}, } // isTableNotFound returns true if the error indicates a missing table. @@ -581,6 +584,20 @@ func migrateEnsureMetadataCacheTable(db *DB) error { return db.EnsureMetadataCacheTable() } +func migrateAddMetadataContentDigest(db *DB) error { + hasColumn, err := db.HasColumn("metadata_cache", "content_digest") + if err != nil { + return fmt.Errorf("checking metadata_cache content_digest column: %w", err) + } + if hasColumn { + return nil + } + if _, err := db.Exec("ALTER TABLE metadata_cache ADD COLUMN content_digest TEXT"); err != nil { + return fmt.Errorf("adding metadata_cache content_digest column: %w", err) + } + return nil +} + // EnsureMetadataCacheTable creates the metadata_cache table if it doesn't exist. func (db *DB) EnsureMetadataCacheTable() error { has, err := db.HasTable("metadata_cache") @@ -601,6 +618,7 @@ func (db *DB) EnsureMetadataCacheTable() error { storage_path TEXT NOT NULL, etag TEXT, content_type TEXT, + content_digest TEXT, size BIGINT, last_modified TIMESTAMP, fetched_at TIMESTAMP, @@ -618,6 +636,7 @@ func (db *DB) EnsureMetadataCacheTable() error { storage_path TEXT NOT NULL, etag TEXT, content_type TEXT, + content_digest TEXT, size INTEGER, last_modified DATETIME, fetched_at DATETIME, diff --git a/internal/database/types.go b/internal/database/types.go index 47dc47e..7128f12 100644 --- a/internal/database/types.go +++ b/internal/database/types.go @@ -78,17 +78,18 @@ func (a *Artifact) IsCached() bool { // MetadataCacheEntry represents a cached metadata blob for offline serving. type MetadataCacheEntry struct { - ID int64 `db:"id" json:"id"` - Ecosystem string `db:"ecosystem" json:"ecosystem"` - Name string `db:"name" json:"name"` - StoragePath string `db:"storage_path" json:"storage_path"` - ETag sql.NullString `db:"etag" json:"etag,omitempty"` - ContentType sql.NullString `db:"content_type" json:"content_type,omitempty"` - Size sql.NullInt64 `db:"size" json:"size,omitempty"` - LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"` - FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"` - CreatedAt time.Time `db:"created_at" json:"created_at"` - UpdatedAt time.Time `db:"updated_at" json:"updated_at"` + ID int64 `db:"id" json:"id"` + Ecosystem string `db:"ecosystem" json:"ecosystem"` + Name string `db:"name" json:"name"` + StoragePath string `db:"storage_path" json:"storage_path"` + ETag sql.NullString `db:"etag" json:"etag,omitempty"` + ContentType sql.NullString `db:"content_type" json:"content_type,omitempty"` + ContentDigest sql.NullString `db:"content_digest" json:"content_digest,omitempty"` + Size sql.NullInt64 `db:"size" json:"size,omitempty"` + LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"` + FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"` + CreatedAt time.Time `db:"created_at" json:"created_at"` + UpdatedAt time.Time `db:"updated_at" json:"updated_at"` } // Vulnerability represents a cached vulnerability record. diff --git a/internal/handler/container.go b/internal/handler/container.go index 8ba5e97..89169aa 100644 --- a/internal/handler/container.go +++ b/internal/handler/container.go @@ -11,7 +11,6 @@ import ( const ( dockerHubRegistry = "https://registry-1.docker.io" - dockerHubAuth = "https://auth.docker.io" blobMatchCount = 3 // full match + name + digest manifestMatchCount = 3 // full match + name + reference tagsListMatchCount = 2 // full match + name @@ -23,7 +22,6 @@ const ( type ContainerHandler struct { proxy *Proxy registryURL string - authURL string proxyURL string } @@ -32,7 +30,6 @@ func NewContainerHandler(proxy *Proxy, proxyURL string) *ContainerHandler { return &ContainerHandler{ proxy: proxy, registryURL: dockerHubRegistry, - authURL: dockerHubAuth, proxyURL: strings.TrimSuffix(proxyURL, "/"), } } @@ -89,31 +86,38 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req h.proxy.Logger.Info("container blob request", "name", name, "digest", digest) - // Get auth token for upstream - token, err := h.getAuthToken(r.Context(), name, "pull") + filename := digest + cached, err := h.proxy.GetCachedArtifact(r.Context(), "oci", name, digest, filename) if err != nil { - h.proxy.Logger.Error("failed to get auth token", "error", err) - h.containerError(w, http.StatusUnauthorized, "UNAUTHORIZED", "failed to authenticate") + h.proxy.Logger.Error("failed to check blob cache", "error", err) + h.containerError(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to check blob cache") + return + } + if cached != nil { + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("Content-Type", "application/octet-stream") + if r.Method == http.MethodHead { + serveArtifactHead(w, cached) + return + } + ServeArtifact(w, cached) return } // For HEAD requests, just proxy to upstream if r.Method == http.MethodHead { - h.proxyBlobHead(w, r, name, digest, token) + h.proxyBlobHead(w, r, name, digest) return } - // Try to get from cache, or fetch from upstream with auth - filename := digest - headers := http.Header{"Authorization": {"Bearer " + token}} - result, err := h.proxy.GetOrFetchArtifactFromURLWithHeaders( + // Try to get from cache, or fetch from the authentication-aware upstream client. + result, err := h.proxy.GetOrFetchArtifactFromURL( r.Context(), "oci", name, digest, // use digest as version filename, fmt.Sprintf("%s/v2/%s/blobs/%s", h.registryURL, name, digest), - headers, ) if err != nil { @@ -127,8 +131,23 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req ServeArtifact(w, result) } -// handleManifest proxies manifest requests to upstream. -// Manifests change when tags are updated, so we proxy these directly. +func serveArtifactHead(w http.ResponseWriter, result *CacheResult) { + if result.Reader != nil { + _ = result.Reader.Close() + } + if result.ContentType != "" { + w.Header().Set("Content-Type", result.ContentType) + } + if result.Size >= 0 { + w.Header().Set("Content-Length", fmt.Sprintf("%d", result.Size)) + } + if result.Hash != "" { + w.Header().Set("ETag", fmt.Sprintf(`"%s"`, result.Hash)) + } + w.WriteHeader(http.StatusOK) +} + +// handleManifest serves immutable manifests from cache and revalidates mutable tags. // Path format: {name}/manifests/{reference} func (h *ContainerHandler) handleManifest(w http.ResponseWriter, r *http.Request, path string) { if r.Method != http.MethodGet && r.Method != http.MethodHead { @@ -143,57 +162,7 @@ func (h *ContainerHandler) handleManifest(w http.ResponseWriter, r *http.Request } h.proxy.Logger.Info("container manifest request", "name", name, "reference", reference) - - // Get auth token - token, err := h.getAuthToken(r.Context(), name, "pull") - if err != nil { - h.proxy.Logger.Error("failed to get auth token", "error", err) - h.containerError(w, http.StatusUnauthorized, "UNAUTHORIZED", "failed to authenticate") - return - } - - // Proxy to upstream - upstreamURL := fmt.Sprintf("%s/v2/%s/manifests/%s", h.registryURL, name, reference) - - req, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, nil) - if err != nil { - h.containerError(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to create request") - return - } - - req.Header.Set("Authorization", "Bearer "+token) - - // Forward Accept header for content negotiation - if accept := r.Header.Get("Accept"); accept != "" { - req.Header.Set("Accept", accept) - } else { - // Default accept headers for manifests - req.Header.Set("Accept", strings.Join([]string{ - "application/vnd.oci.image.manifest.v1+json", - "application/vnd.oci.image.index.v1+json", - "application/vnd.docker.distribution.manifest.v2+json", - "application/vnd.docker.distribution.manifest.list.v2+json", - "application/vnd.docker.distribution.manifest.v1+prettyjws", - }, ", ")) - } - - resp, err := h.proxy.HTTPClient.Do(req) - if err != nil { - h.proxy.Logger.Error("failed to fetch manifest", "error", err) - h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") - return - } - defer func() { _ = resp.Body.Close() }() - - // Copy relevant headers - for _, header := range []string{"Content-Type", "Content-Length", "Docker-Content-Digest", "ETag"} { - if v := resp.Header.Get(header); v != "" { - w.Header().Set(header, v) - } - } - - w.WriteHeader(resp.StatusCode) - _, _ = io.Copy(w, resp.Body) + h.serveManifest(w, r, name, reference) } // handleTagsList proxies tag list requests to upstream. @@ -209,13 +178,6 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request return } - // Get auth token - token, err := h.getAuthToken(r.Context(), name, "pull") - if err != nil { - h.containerError(w, http.StatusUnauthorized, "UNAUTHORIZED", "failed to authenticate") - return - } - upstreamURL := fmt.Sprintf("%s/v2/%s/tags/list", h.registryURL, name) if r.URL.RawQuery != "" { upstreamURL += "?" + r.URL.RawQuery @@ -227,8 +189,6 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request return } - req.Header.Set("Authorization", "Bearer "+token) - resp, err := h.proxy.HTTPClient.Do(req) if err != nil { h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") @@ -241,45 +201,8 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request _, _ = io.Copy(w, resp.Body) } -// getAuthToken gets a bearer token for the specified repository. -// Docker Hub requires auth even for public images. -func (h *ContainerHandler) getAuthToken(_ interface{ Done() <-chan struct{} }, repository, action string) (string, error) { - // For Docker Hub: https://auth.docker.io/token?service=registry.docker.io&scope=repository:{repo}:pull - authURL := fmt.Sprintf("%s/token?service=registry.docker.io&scope=repository:%s:%s", - h.authURL, repository, action) - - req, err := http.NewRequest(http.MethodGet, authURL, nil) - if err != nil { - return "", err - } - - resp, err := h.proxy.HTTPClient.Do(req) - if err != nil { - return "", err - } - defer func() { _ = resp.Body.Close() }() - - if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("auth failed with status %d", resp.StatusCode) - } - - var tokenResp struct { - Token string `json:"token"` - AccessToken string `json:"access_token"` - } - - if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil { - return "", err - } - - if tokenResp.Token != "" { - return tokenResp.Token, nil - } - return tokenResp.AccessToken, nil -} - // proxyBlobHead handles HEAD requests for blobs. -func (h *ContainerHandler) proxyBlobHead(w http.ResponseWriter, r *http.Request, name, digest, token string) { +func (h *ContainerHandler) proxyBlobHead(w http.ResponseWriter, r *http.Request, name, digest string) { upstreamURL := fmt.Sprintf("%s/v2/%s/blobs/%s", h.registryURL, name, digest) req, err := http.NewRequestWithContext(r.Context(), http.MethodHead, upstreamURL, nil) @@ -288,8 +211,6 @@ func (h *ContainerHandler) proxyBlobHead(w http.ResponseWriter, r *http.Request, return } - req.Header.Set("Authorization", "Bearer "+token) - resp, err := h.proxy.HTTPClient.Do(req) if err != nil { h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go new file mode 100644 index 0000000..245ced4 --- /dev/null +++ b/internal/handler/container_manifest.go @@ -0,0 +1,251 @@ +package handler + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "fmt" + "io" + "net/http" + "regexp" + "strconv" + "strings" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +const ( + containerManifestCacheEcosystem = "oci-manifest" + containerStaleWarning = `110 - "Response is Stale"` +) + +var manifestDigestReferencePattern = regexp.MustCompile(`^[a-z0-9]+:[a-f0-9]+$`) + +type cachedContainerManifest struct { + body []byte + contentType string + contentDigest string + etag string + size int64 + fetchedAt time.Time +} + +func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, name, reference string) { + accept := containerManifestAccept(r) + cacheKey := h.containerManifestCacheKey(name, reference, accept) + cached, err := h.loadContainerManifest(r.Context(), cacheKey) + if err != nil { + h.proxy.Logger.Warn("failed to read cached container manifest", "error", err) + cached = nil + } + + immutable := manifestDigestReferencePattern.MatchString(reference) + if cached != nil && (immutable || h.containerManifestFresh(cached)) { + writeContainerManifest(w, r.Method, cached, false) + return + } + + upstreamURL := fmt.Sprintf("%s/v2/%s/manifests/%s", h.registryURL, name, reference) + req, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, nil) + if err != nil { + h.containerError(w, http.StatusInternalServerError, "INTERNAL_ERROR", "failed to create request") + return + } + req.Header.Set("Accept", accept) + if cached != nil && cached.etag != "" { + req.Header.Set("If-None-Match", cached.etag) + } + + resp, err := h.proxy.HTTPClient.Do(req) + if err != nil { + h.serveStaleManifestOrError(w, r, cached, err) + return + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode == http.StatusNotModified && cached != nil { + cached.fetchedAt = time.Now() + if err := h.storeContainerManifest(r.Context(), cacheKey, cached); err != nil { + h.proxy.Logger.Warn("failed to refresh cached container manifest", "error", err) + } + writeContainerManifest(w, r.Method, cached, false) + return + } + if resp.StatusCode != http.StatusOK { + if cached != nil && shouldServeStaleManifest(resp.StatusCode) { + writeContainerManifest(w, r.Method, cached, true) + return + } + copyContainerManifestHeaders(w.Header(), resp.Header) + w.WriteHeader(resp.StatusCode) + _, _ = io.Copy(w, resp.Body) + return + } + + if r.Method == http.MethodHead { + copyContainerManifestHeaders(w.Header(), resp.Header) + w.WriteHeader(http.StatusOK) + return + } + + body, err := h.proxy.ReadMetadata(resp.Body) + if err != nil { + h.serveStaleManifestOrError(w, r, cached, fmt.Errorf("reading manifest: %w", err)) + return + } + manifest := &cachedContainerManifest{ + body: body, + contentType: resp.Header.Get("Content-Type"), + contentDigest: resp.Header.Get("Docker-Content-Digest"), + etag: resp.Header.Get("ETag"), + size: int64(len(body)), + fetchedAt: time.Now(), + } + if manifest.contentDigest == "" { + manifest.contentDigest = sha256Digest(body) + } + if err := h.storeContainerManifest(r.Context(), cacheKey, manifest); err != nil { + h.proxy.Logger.Warn("failed to cache container manifest", "error", err) + } + if manifest.contentDigest != reference && manifestDigestReferencePattern.MatchString(manifest.contentDigest) { + digestKey := h.containerManifestCacheKey(name, manifest.contentDigest, accept) + if err := h.storeContainerManifest(r.Context(), digestKey, manifest); err != nil { + h.proxy.Logger.Warn("failed to cache container manifest by digest", "error", err) + } + } + writeContainerManifest(w, r.Method, manifest, false) +} + +func (h *ContainerHandler) serveStaleManifestOrError(w http.ResponseWriter, r *http.Request, cached *cachedContainerManifest, err error) { + if cached != nil { + h.proxy.Logger.Warn("upstream manifest fetch failed, serving stale cache", "error", err) + writeContainerManifest(w, r.Method, cached, true) + return + } + h.proxy.Logger.Error("failed to fetch manifest", "error", err) + h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") +} + +func (h *ContainerHandler) containerManifestFresh(manifest *cachedContainerManifest) bool { + return h.proxy.MetadataTTL > 0 && !manifest.fetchedAt.IsZero() && time.Since(manifest.fetchedAt) < h.proxy.MetadataTTL +} + +func (h *ContainerHandler) containerManifestCacheKey(name, reference, accept string) string { + identity := strings.Join([]string{h.registryURL, name, reference, accept}, "\x00") + sum := sha256.Sum256([]byte(identity)) + return hex.EncodeToString(sum[:]) +} + +func (h *ContainerHandler) loadContainerManifest(ctx context.Context, cacheKey string) (*cachedContainerManifest, error) { + if h.proxy.DB == nil || h.proxy.Storage == nil { + return nil, nil + } + entry, err := h.proxy.DB.GetMetadataCache(containerManifestCacheEcosystem, cacheKey) + if err != nil || entry == nil { + return nil, err + } + reader, err := h.proxy.Storage.Open(ctx, entry.StoragePath) + if err != nil { + return nil, nil + } + defer func() { _ = reader.Close() }() + body, err := h.proxy.ReadMetadata(reader) + if err != nil { + return nil, err + } + + manifest := &cachedContainerManifest{body: body, size: int64(len(body))} + if entry.ContentType.Valid { + manifest.contentType = entry.ContentType.String + } + if entry.ContentDigest.Valid { + manifest.contentDigest = entry.ContentDigest.String + } else { + manifest.contentDigest = sha256Digest(body) + } + if entry.ETag.Valid { + manifest.etag = entry.ETag.String + } + if entry.Size.Valid { + manifest.size = entry.Size.Int64 + } + if entry.FetchedAt.Valid { + manifest.fetchedAt = entry.FetchedAt.Time + } + return manifest, nil +} + +func (h *ContainerHandler) storeContainerManifest(ctx context.Context, cacheKey string, manifest *cachedContainerManifest) error { + if h.proxy.DB == nil || h.proxy.Storage == nil { + return nil + } + storagePath := metadataStoragePath(containerManifestCacheEcosystem, cacheKey) + size, _, err := h.proxy.Storage.Store(ctx, storagePath, bytes.NewReader(manifest.body)) + if err != nil { + return fmt.Errorf("storing manifest: %w", err) + } + manifest.size = size + return h.proxy.DB.UpsertMetadataCache(&database.MetadataCacheEntry{ + Ecosystem: containerManifestCacheEcosystem, + Name: cacheKey, + StoragePath: storagePath, + ETag: sql.NullString{String: manifest.etag, Valid: manifest.etag != ""}, + ContentType: sql.NullString{String: manifest.contentType, Valid: manifest.contentType != ""}, + ContentDigest: sql.NullString{String: manifest.contentDigest, Valid: manifest.contentDigest != ""}, + Size: sql.NullInt64{Int64: size, Valid: true}, + FetchedAt: sql.NullTime{Time: manifest.fetchedAt, Valid: !manifest.fetchedAt.IsZero()}, + }) +} + +func writeContainerManifest(w http.ResponseWriter, method string, manifest *cachedContainerManifest, stale bool) { + if manifest.contentType != "" { + w.Header().Set("Content-Type", manifest.contentType) + } + w.Header().Set("Content-Length", strconv.FormatInt(manifest.size, 10)) + if manifest.contentDigest != "" { + w.Header().Set("Docker-Content-Digest", manifest.contentDigest) + } + if manifest.etag != "" { + w.Header().Set("ETag", manifest.etag) + } + if stale { + w.Header().Set("Warning", containerStaleWarning) + } + w.WriteHeader(http.StatusOK) + if method != http.MethodHead { + _, _ = w.Write(manifest.body) + } +} + +func containerManifestAccept(r *http.Request) string { + if accept := r.Header.Get("Accept"); accept != "" { + return accept + } + return strings.Join([]string{ + "application/vnd.oci.image.manifest.v1+json", + "application/vnd.oci.image.index.v1+json", + "application/vnd.docker.distribution.manifest.v2+json", + "application/vnd.docker.distribution.manifest.list.v2+json", + "application/vnd.docker.distribution.manifest.v1+prettyjws", + }, ", ") +} + +func copyContainerManifestHeaders(destination, source http.Header) { + for _, header := range []string{"Content-Type", "Content-Length", "Docker-Content-Digest", "ETag", "WWW-Authenticate"} { + if value := source.Get(header); value != "" { + destination.Set(header, value) + } + } +} + +func shouldServeStaleManifest(status int) bool { + return status == http.StatusTooManyRequests || status >= http.StatusInternalServerError +} + +func sha256Digest(body []byte) string { + digest := sha256.Sum256(body) + return "sha256:" + hex.EncodeToString(digest[:]) +} diff --git a/internal/handler/container_test.go b/internal/handler/container_test.go index 853059e..40ba53a 100644 --- a/internal/handler/container_test.go +++ b/internal/handler/container_test.go @@ -1,16 +1,14 @@ package handler import ( - "bytes" - "context" "encoding/json" "io" - "log/slog" "net/http" "net/http/httptest" + "strconv" "testing" - "github.com/git-pkgs/proxy/internal/database" + upstreamhttp "github.com/git-pkgs/proxy/internal/httpclient" "github.com/git-pkgs/registries/fetch" ) @@ -135,90 +133,353 @@ func TestContainerHandler_parseTagsListPath(t *testing.T) { } } -func TestContainerHandler_BlobDownload_CachesWithAuth(t *testing.T) { - // Set up a mock auth server that returns a token - authServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - _ = json.NewEncoder(w).Encode(map[string]string{"token": "test-token-123"}) +func TestContainerHandler_BlobDownload_DiscoversBearerChallenge(t *testing.T) { + digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" + registryRequests := 0 + tokenRequests := 0 + var upstream *httptest.Server + upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/token": + tokenRequests++ + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "token": "discovered-token", + "expires_in": 3600, + }) + case "/v2/library/nginx/blobs/" + digest: + registryRequests++ + if r.Header.Get("Authorization") != "Bearer discovered-token" { + w.Header().Set("WWW-Authenticate", `Bearer realm="`+upstream.URL+`/token",service="registry.test",scope="repository:library/nginx:pull"`) + http.Error(w, "authentication required", http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = io.WriteString(w, "upstream blob") + default: + http.NotFound(w, r) + } })) - defer authServer.Close() + defer upstream.Close() - // Set up mock fetcher that captures headers - var capturedHeaders http.Header - mf := &mockFetcherWithHeaders{ - fetchFn: func(_ context.Context, _ string, headers http.Header) (*fetch.Artifact, error) { - capturedHeaders = headers - return &fetch.Artifact{ - Body: io.NopCloser(bytes.NewReader([]byte("blob-content"))), - Size: 12, - ContentType: "application/octet-stream", - }, nil - }, - } - - dir := t.TempDir() - db, err := database.Create(dir + "/test.db") - if err != nil { - t.Fatalf("failed to create test database: %v", err) - } - t.Cleanup(func() { _ = db.Close() }) - - store := newMockStorage() - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - proxy := &Proxy{ - DB: db, - Storage: store, - Fetcher: mf, - Logger: logger, - HTTPClient: &http.Client{}, - } + proxy, _, _, _ := setupTestProxy(t) + authTransport := upstreamhttp.NewTransport(http.DefaultTransport, nil) + client := &http.Client{Transport: authTransport} + artifactFetcher := fetch.NewFetcher( + fetch.WithHTTPClient(client), + fetch.WithMaxRetries(0), + ) + t.Cleanup(func() { _ = artifactFetcher.Close() }) + proxy.Fetcher = artifactFetcher + proxy.HTTPClient = client h := &ContainerHandler{ proxy: proxy, - registryURL: "https://registry-1.docker.io", - authURL: authServer.URL, + registryURL: upstream.URL, proxyURL: "http://localhost:8080", } - handler := h.Routes() - req := httptest.NewRequest(http.MethodGet, "/library/nginx/blobs/sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd", nil) + for range 2 { + req := httptest.NewRequest(http.MethodGet, "/library/nginx/blobs/"+digest, nil) + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body: %s", w.Code, http.StatusOK, w.Body.String()) + } + if got := w.Body.String(); got != "upstream blob" { + t.Errorf("body = %q, want %q", got, "upstream blob") + } + } + + if tokenRequests != 1 { + t.Errorf("token requests = %d, want 1", tokenRequests) + } + if registryRequests != 2 { + t.Errorf("registry requests = %d, want 2", registryRequests) + } +} + +func TestContainerHandler_BlobDownload_CacheHitSkipsAuth(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" + seedPackage(t, db, store, "oci", "library/nginx", digest, digest, "cached blob") + + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamRequests++ + http.Error(w, "upstream unavailable", http.StatusServiceUnavailable) + })) + defer upstream.Close() + + h := &ContainerHandler{ + proxy: proxy, + registryURL: upstream.URL, + proxyURL: "http://localhost:8080", + } + + req := httptest.NewRequest(http.MethodGet, "/library/nginx/blobs/"+digest, nil) w := httptest.NewRecorder() - handler.ServeHTTP(w, req) + h.Routes().ServeHTTP(w, req) if w.Code != http.StatusOK { - t.Errorf("got status %d, want %d; body: %s", w.Code, http.StatusOK, w.Body.String()) + t.Fatalf("status = %d, want %d; body: %s", w.Code, http.StatusOK, w.Body.String()) } - - // Verify auth header was passed to the fetcher - if capturedHeaders == nil { - t.Fatal("expected headers to be passed to fetcher, got nil") + if got := w.Body.String(); got != "cached blob" { + t.Errorf("body = %q, want %q", got, "cached blob") } - auth := capturedHeaders.Get("Authorization") - if auth != "Bearer test-token-123" { - t.Errorf("Authorization = %q, want %q", auth, "Bearer test-token-123") + if upstreamRequests != 0 { + t.Errorf("upstream requests = %d, want 0", upstreamRequests) } - - // Verify response headers - if got := w.Header().Get("Docker-Content-Digest"); got != "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" { - t.Errorf("Docker-Content-Digest = %q, want digest", got) + if fetcher.fetchCalled { + t.Error("fetcher should not be called on cache hit") } } -// mockFetcherWithHeaders captures headers passed to FetchWithHeaders. -type mockFetcherWithHeaders struct { - fetchFn func(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) +func TestContainerHandler_BlobHead_CacheHitSkipsUpstreamAndAuth(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" + seedPackage(t, db, store, "oci", "library/nginx", digest, digest, "cached blob") + + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamRequests++ + http.Error(w, "upstream unavailable", http.StatusServiceUnavailable) + })) + defer upstream.Close() + proxy.HTTPClient = upstream.Client() + + h := &ContainerHandler{ + proxy: proxy, + registryURL: upstream.URL, + proxyURL: "http://localhost:8080", + } + + req := httptest.NewRequest(http.MethodHead, "/library/nginx/blobs/"+digest, nil) + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body: %s", w.Code, http.StatusOK, w.Body.String()) + } + if got := w.Header().Get("Docker-Content-Digest"); got != digest { + t.Errorf("Docker-Content-Digest = %q, want %q", got, digest) + } + if got := w.Header().Get("Content-Length"); got != "11" { + t.Errorf("Content-Length = %q, want %q", got, "11") + } + if w.Body.Len() != 0 { + t.Errorf("HEAD response body length = %d, want 0", w.Body.Len()) + } + if upstreamRequests != 0 { + t.Errorf("upstream requests = %d, want 0", upstreamRequests) + } + if fetcher.fetchCalled { + t.Error("fetcher should not be called on cache hit") + } } -func (f *mockFetcherWithHeaders) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) { - return f.FetchWithHeaders(ctx, url, nil) +func TestContainerHandler_ManifestByDigest_CacheHitSkipsUpstream(t *testing.T) { + digest := "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}` + upstreamAvailable := true + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamRequests++ + if !upstreamAvailable { + http.Error(w, "upstream unavailable", http.StatusServiceUnavailable) + return + } + if r.URL.Path != "/v2/library/nginx/manifests/"+digest { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json") + w.Header().Set("Docker-Content-Digest", digest) + w.Header().Set("ETag", `"manifest-etag"`) + if r.Method != http.MethodHead { + _, _ = io.WriteString(w, manifest) + } + })) + defer upstream.Close() + + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + h := &ContainerHandler{proxy: proxy, registryURL: upstream.URL, proxyURL: "http://localhost:8080"} + + first := httptest.NewRecorder() + h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/"+digest, nil)) + if first.Code != http.StatusOK { + t.Fatalf("initial status = %d, want %d; body: %s", first.Code, http.StatusOK, first.Body.String()) + } + if first.Body.String() != manifest { + t.Fatalf("initial body = %q, want %q", first.Body.String(), manifest) + } + + upstreamAvailable = false + second := httptest.NewRecorder() + h.Routes().ServeHTTP(second, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/"+digest, nil)) + if second.Code != http.StatusOK { + t.Fatalf("cached status = %d, want %d; body: %s", second.Code, http.StatusOK, second.Body.String()) + } + if second.Body.String() != manifest { + t.Errorf("cached body = %q, want %q", second.Body.String(), manifest) + } + if got := second.Header().Get("Docker-Content-Digest"); got != digest { + t.Errorf("cached Docker-Content-Digest = %q, want %q", got, digest) + } + + head := httptest.NewRecorder() + h.Routes().ServeHTTP(head, httptest.NewRequest(http.MethodHead, "/library/nginx/manifests/"+digest, nil)) + if head.Code != http.StatusOK { + t.Fatalf("cached HEAD status = %d, want %d", head.Code, http.StatusOK) + } + wantLength := strconv.Itoa(len(manifest)) + if got := head.Header().Get("Content-Length"); got != wantLength { + t.Errorf("cached HEAD Content-Length = %q, want %q", got, wantLength) + } + if head.Body.Len() != 0 { + t.Errorf("cached HEAD body length = %d, want 0", head.Body.Len()) + } + if upstreamRequests != 1 { + t.Errorf("upstream requests = %d, want 1", upstreamRequests) + } } -func (f *mockFetcherWithHeaders) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) { - return f.fetchFn(ctx, url, headers) +func TestContainerHandler_ManifestByTag_UsesStaleCacheOnUpstreamFailure(t *testing.T) { + digest := "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json"}` + upstreamAvailable := true + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamRequests++ + if !upstreamAvailable { + http.Error(w, "upstream unavailable", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json") + w.Header().Set("Docker-Content-Digest", digest) + _, _ = io.WriteString(w, manifest) + })) + defer upstream.Close() + + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + proxy.MetadataTTL = 0 + h := &ContainerHandler{proxy: proxy, registryURL: upstream.URL, proxyURL: "http://localhost:8080"} + + first := httptest.NewRecorder() + h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/latest", nil)) + if first.Code != http.StatusOK { + t.Fatalf("initial status = %d, want %d; body: %s", first.Code, http.StatusOK, first.Body.String()) + } + + upstreamAvailable = false + second := httptest.NewRecorder() + h.Routes().ServeHTTP(second, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/latest", nil)) + if second.Code != http.StatusOK { + t.Fatalf("stale status = %d, want %d; body: %s", second.Code, http.StatusOK, second.Body.String()) + } + if second.Body.String() != manifest { + t.Errorf("stale body = %q, want %q", second.Body.String(), manifest) + } + if got := second.Header().Get("Warning"); got != `110 - "Response is Stale"` { + t.Errorf("Warning = %q, want stale warning", got) + } + if got := second.Header().Get("Docker-Content-Digest"); got != digest { + t.Errorf("stale Docker-Content-Digest = %q, want %q", got, digest) + } + if upstreamRequests != 2 { + t.Errorf("upstream requests = %d, want 2", upstreamRequests) + } } -func (f *mockFetcherWithHeaders) Head(_ context.Context, _ string) (int64, string, error) { - return 0, "", nil +func TestContainerHandler_ManifestByTag_CachesDigestAlias(t *testing.T) { + digest := "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}` + upstreamAvailable := true + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamRequests++ + if !upstreamAvailable { + http.Error(w, "upstream unavailable", http.StatusServiceUnavailable) + return + } + if r.URL.Path != "/v2/library/nginx/manifests/latest" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json") + w.Header().Set("Docker-Content-Digest", digest) + _, _ = io.WriteString(w, manifest) + })) + defer upstream.Close() + + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + h := &ContainerHandler{proxy: proxy, registryURL: upstream.URL, proxyURL: "http://localhost:8080"} + + first := httptest.NewRecorder() + h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/latest", nil)) + if first.Code != http.StatusOK { + t.Fatalf("tag status = %d, want %d; body: %s", first.Code, http.StatusOK, first.Body.String()) + } + + upstreamAvailable = false + byDigest := httptest.NewRecorder() + h.Routes().ServeHTTP(byDigest, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/"+digest, nil)) + if byDigest.Code != http.StatusOK { + t.Fatalf("digest status = %d, want %d; body: %s", byDigest.Code, http.StatusOK, byDigest.Body.String()) + } + if byDigest.Body.String() != manifest { + t.Errorf("digest body = %q, want %q", byDigest.Body.String(), manifest) + } + if got := byDigest.Header().Get("Docker-Content-Digest"); got != digest { + t.Errorf("Docker-Content-Digest = %q, want %q", got, digest) + } + if upstreamRequests != 1 { + t.Errorf("upstream requests = %d, want 1", upstreamRequests) + } +} + +func TestContainerHandler_ManifestByTag_StaleHeadChecksUpstream(t *testing.T) { + oldDigest := "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd" + newDigest := "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee" + currentDigest := oldDigest + upstreamRequests := 0 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamRequests++ + w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json") + w.Header().Set("Docker-Content-Digest", currentDigest) + w.Header().Set("ETag", `"`+currentDigest+`"`) + if r.Method != http.MethodHead { + _, _ = io.WriteString(w, `{"schemaVersion":2}`) + } + })) + defer upstream.Close() + + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + proxy.MetadataTTL = 0 + h := &ContainerHandler{proxy: proxy, registryURL: upstream.URL, proxyURL: "http://localhost:8080"} + + first := httptest.NewRecorder() + h.Routes().ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/library/nginx/manifests/latest", nil)) + if first.Code != http.StatusOK { + t.Fatalf("initial status = %d, want %d", first.Code, http.StatusOK) + } + + currentDigest = newDigest + head := httptest.NewRecorder() + h.Routes().ServeHTTP(head, httptest.NewRequest(http.MethodHead, "/library/nginx/manifests/latest", nil)) + if head.Code != http.StatusOK { + t.Fatalf("HEAD status = %d, want %d", head.Code, http.StatusOK) + } + if got := head.Header().Get("Docker-Content-Digest"); got != newDigest { + t.Errorf("Docker-Content-Digest = %q, want %q", got, newDigest) + } + if upstreamRequests != 2 { + t.Errorf("upstream requests = %d, want 2", upstreamRequests) + } } func TestContainerHandler_Routes_VersionCheck(t *testing.T) { diff --git a/internal/handler/handler.go b/internal/handler/handler.go index d06ca83..91adc93 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -127,18 +127,25 @@ type CacheResult struct { // GetOrFetchArtifact retrieves an artifact from cache or fetches from upstream. func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) { - pkgPURL := purl.MakePURLString(ecosystem, name, "") - versionPURL := purl.MakePURLString(ecosystem, name, version) - - if cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename); err != nil { + if cached, err := p.GetCachedArtifact(ctx, ecosystem, name, version, filename); err != nil { return nil, err } else if cached != nil { return cached, nil } + pkgPURL := purl.MakePURLString(ecosystem, name, "") + versionPURL := purl.MakePURLString(ecosystem, name, version) return p.fetchAndCache(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL) } +// GetCachedArtifact retrieves an artifact from cache without contacting an upstream. +// It returns nil when no usable cache entry exists. +func (p *Proxy) GetCachedArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) { + pkgPURL := purl.MakePURLString(ecosystem, name, "") + versionPURL := purl.MakePURLString(ecosystem, name, version) + return p.checkCache(ctx, pkgPURL, versionPURL, filename) +} + // checkCache looks up an artifact in the cache. Returns nil if not cached. func (p *Proxy) checkCache(ctx context.Context, pkgPURL, versionPURL, filename string) (*CacheResult, error) { pkg, err := p.DB.GetPackageByPURL(pkgPURL) @@ -766,18 +773,16 @@ func (p *Proxy) GetOrFetchArtifactFromURL(ctx context.Context, ecosystem, name, } // GetOrFetchArtifactFromURLWithHeaders retrieves an artifact from cache or fetches from a URL -// with additional HTTP headers. This is needed for registries that require authentication -// (e.g. Docker Hub requires a Bearer token even for public images). +// with additional request-specific HTTP headers. func (p *Proxy) GetOrFetchArtifactFromURLWithHeaders(ctx context.Context, ecosystem, name, version, filename, downloadURL string, headers http.Header) (*CacheResult, error) { - pkgPURL := purl.MakePURLString(ecosystem, name, "") - versionPURL := purl.MakePURLString(ecosystem, name, version) - - if cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename); err != nil { + if cached, err := p.GetCachedArtifact(ctx, ecosystem, name, version, filename); err != nil { return nil, err } else if cached != nil { return cached, nil } + pkgPURL := purl.MakePURLString(ecosystem, name, "") + versionPURL := purl.MakePURLString(ecosystem, name, version) return p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers) } diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index bbcab72..bb2a837 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -5,7 +5,6 @@ import ( "context" "database/sql" "errors" - "fmt" "io" "log/slog" "net/http" @@ -16,6 +15,7 @@ import ( "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/purl" "github.com/git-pkgs/registries/fetch" ) @@ -151,7 +151,7 @@ func seedPackage(t *testing.T, db *database.DB, store *mockStorage, ecosystem, n t.Helper() pkg := &database.Package{ - PURL: fmt.Sprintf("pkg:%s/%s", ecosystem, name), + PURL: purl.MakePURLString(ecosystem, name, ""), Ecosystem: ecosystem, Name: name, } @@ -159,7 +159,7 @@ func seedPackage(t *testing.T, db *database.DB, store *mockStorage, ecosystem, n t.Fatalf("failed to upsert package: %v", err) } - versionPURL := fmt.Sprintf("pkg:%s/%s@%s", ecosystem, name, version) + versionPURL := purl.MakePURLString(ecosystem, name, version) ver := &database.Version{ PURL: versionPURL, PackagePURL: pkg.PURL, From bc0802153657320a047d6b9b92a58072fefcb2c2 Mon Sep 17 00:00:00 2001 From: Andrew Nesbitt Date: Mon, 13 Jul 2026 17:20:27 -0700 Subject: [PATCH 3/3] Preserve direct-serve redirects for blob HEAD requests --- internal/handler/container.go | 8 +++++++ internal/handler/container_test.go | 34 ++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/internal/handler/container.go b/internal/handler/container.go index 89169aa..3ef2fa1 100644 --- a/internal/handler/container.go +++ b/internal/handler/container.go @@ -132,6 +132,14 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req } func serveArtifactHead(w http.ResponseWriter, result *CacheResult) { + if result.RedirectURL != "" { + if result.Hash != "" { + w.Header().Set("ETag", fmt.Sprintf(`"%s"`, result.Hash)) + } + w.Header().Set("Location", result.RedirectURL) + w.WriteHeader(http.StatusFound) + return + } if result.Reader != nil { _ = result.Reader.Close() } diff --git a/internal/handler/container_test.go b/internal/handler/container_test.go index 40ba53a..404b12a 100644 --- a/internal/handler/container_test.go +++ b/internal/handler/container_test.go @@ -278,6 +278,40 @@ func TestContainerHandler_BlobHead_CacheHitSkipsUpstreamAndAuth(t *testing.T) { } } +func TestContainerHandler_BlobHead_DirectServeRedirects(t *testing.T) { + proxy, db, store, fetcher := setupTestProxy(t) + digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" + seedPackage(t, db, store, "oci", "library/nginx", digest, digest, "cached blob") + store.signedURL = "https://storage.example.test/cached-blob?signature=test" + proxy.DirectServe = true + + h := &ContainerHandler{ + proxy: proxy, + registryURL: "https://registry.example.test", + proxyURL: "http://localhost:8080", + } + + req := httptest.NewRequest(http.MethodHead, "/library/nginx/blobs/"+digest, nil) + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, req) + + if w.Code != http.StatusFound { + t.Fatalf("status = %d, want %d", w.Code, http.StatusFound) + } + if got := w.Header().Get("Location"); got != store.signedURL { + t.Errorf("Location = %q, want %q", got, store.signedURL) + } + if got := w.Header().Get("ETag"); got != `"abc123"` { + t.Errorf("ETag = %q, want %q", got, `"abc123"`) + } + if w.Body.Len() != 0 { + t.Errorf("HEAD response body length = %d, want 0", w.Body.Len()) + } + if fetcher.fetchCalled { + t.Error("fetcher should not be called on cache hit") + } +} + func TestContainerHandler_ManifestByDigest_CacheHitSkipsUpstream(t *testing.T) { digest := "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}`