Watch
1
0
Fork
You've already forked pkg-proxy
1
mirror of https://github.com/git-pkgs/proxy.git synced 2026-08-23 12:24:57 -04:00

Compare commits

...
Author SHA1 Message Date
Andrew Nesbitt
52a02c3e7d
Preserve direct-serve redirects for blob HEAD requests 2026-08-13 07:21:51 +01:00
Andrew Nesbitt
7c074e3564 container: cache manifests for offline pulls 2026-08-13 07:08:25 +01:00
12 changed files with 881 additions and 222 deletions

View file

@ -353,6 +353,7 @@ Eviction can be implemented as:
- Fresh data - new versions visible immediately - Fresh data - new versions visible immediately
- Metadata is small, upstream fetch is fast - 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 - 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?** **Why stream artifacts?**
- Memory efficient - don't load large files into RAM - Memory efficient - don't load large files into RAM

View file

@ -248,6 +248,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. 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 ```yaml
cache_metadata: true cache_metadata: true
``` ```

View file

@ -30,8 +30,12 @@ func TestUpsertAndGetMetadataCache(t *testing.T) {
StoragePath: "_metadata/npm/lodash/metadata", StoragePath: "_metadata/npm/lodash/metadata",
ETag: sql.NullString{String: `"abc123"`, Valid: true}, ETag: sql.NullString{String: `"abc123"`, Valid: true},
ContentType: sql.NullString{String: "application/json", Valid: true}, ContentType: sql.NullString{String: "application/json", Valid: true},
Size: sql.NullInt64{Int64: 1024, Valid: true}, ContentDigest: sql.NullString{
FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, String: "sha256:0123456789abcdef",
Valid: true,
},
Size: sql.NullInt64{Int64: 1024, Valid: true},
FetchedAt: sql.NullTime{Time: time.Now(), Valid: true},
} }
err := db.UpsertMetadataCache(entry) err := db.UpsertMetadataCache(entry)
@ -62,6 +66,9 @@ func TestUpsertAndGetMetadataCache(t *testing.T) {
if !got.ContentType.Valid || got.ContentType.String != "application/json" { if !got.ContentType.Valid || got.ContentType.String != "application/json" {
t.Errorf("content_type = %v, want %q", got.ContentType, "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 { if !got.Size.Valid || got.Size.Int64 != 1024 {
t.Errorf("size = %v, want 1024", got.Size) 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") 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)
}
}

View file

@ -894,7 +894,7 @@ func (db *DB) GetMetadataCache(ecosystem, name string) (*MetadataCacheEntry, err
var entry MetadataCacheEntry var entry MetadataCacheEntry
query := db.Rebind(` query := db.Rebind(`
SELECT id, ecosystem, name, storage_path, etag, content_type, 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 = ? FROM metadata_cache WHERE ecosystem = ? AND name = ?
`) `)
err := db.Get(&entry, query, ecosystem, name) err := db.Get(&entry, query, ecosystem, name)
@ -914,12 +914,13 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error {
if db.dialect == DialectPostgres { if db.dialect == DialectPostgres {
query = ` query = `
INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, content_type, INSERT INTO metadata_cache (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)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT(ecosystem, name) DO UPDATE SET ON CONFLICT(ecosystem, name) DO UPDATE SET
storage_path = EXCLUDED.storage_path, storage_path = EXCLUDED.storage_path,
etag = EXCLUDED.etag, etag = EXCLUDED.etag,
content_type = EXCLUDED.content_type, content_type = EXCLUDED.content_type,
content_digest = EXCLUDED.content_digest,
size = EXCLUDED.size, size = EXCLUDED.size,
last_modified = EXCLUDED.last_modified, last_modified = EXCLUDED.last_modified,
fetched_at = EXCLUDED.fetched_at, fetched_at = EXCLUDED.fetched_at,
@ -928,12 +929,13 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error {
} else { } else {
query = ` query = `
INSERT INTO metadata_cache (ecosystem, name, storage_path, etag, content_type, INSERT INTO metadata_cache (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)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ecosystem, name) DO UPDATE SET ON CONFLICT(ecosystem, name) DO UPDATE SET
storage_path = excluded.storage_path, storage_path = excluded.storage_path,
etag = excluded.etag, etag = excluded.etag,
content_type = excluded.content_type, content_type = excluded.content_type,
content_digest = excluded.content_digest,
size = excluded.size, size = excluded.size,
last_modified = excluded.last_modified, last_modified = excluded.last_modified,
fetched_at = excluded.fetched_at, fetched_at = excluded.fetched_at,
@ -943,7 +945,7 @@ func (db *DB) UpsertMetadataCache(entry *MetadataCacheEntry) error {
_, err := db.Exec(query, _, err := db.Exec(query,
entry.Ecosystem, entry.Name, entry.StoragePath, entry.ETag, 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 { if err != nil {
return fmt.Errorf("upserting metadata cache: %w", err) return fmt.Errorf("upserting metadata cache: %w", err)

View file

@ -102,6 +102,7 @@ CREATE TABLE IF NOT EXISTS metadata_cache (
storage_path TEXT NOT NULL, storage_path TEXT NOT NULL,
etag TEXT, etag TEXT,
content_type TEXT, content_type TEXT,
content_digest TEXT,
size INTEGER, size INTEGER,
last_modified DATETIME, last_modified DATETIME,
fetched_at DATETIME, fetched_at DATETIME,
@ -202,6 +203,7 @@ CREATE TABLE IF NOT EXISTS metadata_cache (
storage_path TEXT NOT NULL, storage_path TEXT NOT NULL,
etag TEXT, etag TEXT,
content_type TEXT, content_type TEXT,
content_digest TEXT,
size BIGINT, size BIGINT,
last_modified TIMESTAMP, last_modified TIMESTAMP,
fetched_at TIMESTAMP, fetched_at TIMESTAMP,
@ -359,6 +361,7 @@ var migrations = []migration{
{"003_ensure_artifacts_table", migrateEnsureArtifactsTable}, {"003_ensure_artifacts_table", migrateEnsureArtifactsTable},
{"004_ensure_vulnerabilities_table", migrateEnsureVulnerabilitiesTable}, {"004_ensure_vulnerabilities_table", migrateEnsureVulnerabilitiesTable},
{"005_ensure_metadata_cache_table", migrateEnsureMetadataCacheTable}, {"005_ensure_metadata_cache_table", migrateEnsureMetadataCacheTable},
{"006_add_metadata_content_digest", migrateAddMetadataContentDigest},
} }
// isTableNotFound returns true if the error indicates a missing table. // isTableNotFound returns true if the error indicates a missing table.
@ -581,6 +584,20 @@ func migrateEnsureMetadataCacheTable(db *DB) error {
return db.EnsureMetadataCacheTable() 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. // EnsureMetadataCacheTable creates the metadata_cache table if it doesn't exist.
func (db *DB) EnsureMetadataCacheTable() error { func (db *DB) EnsureMetadataCacheTable() error {
has, err := db.HasTable("metadata_cache") has, err := db.HasTable("metadata_cache")
@ -601,6 +618,7 @@ func (db *DB) EnsureMetadataCacheTable() error {
storage_path TEXT NOT NULL, storage_path TEXT NOT NULL,
etag TEXT, etag TEXT,
content_type TEXT, content_type TEXT,
content_digest TEXT,
size BIGINT, size BIGINT,
last_modified TIMESTAMP, last_modified TIMESTAMP,
fetched_at TIMESTAMP, fetched_at TIMESTAMP,
@ -618,6 +636,7 @@ func (db *DB) EnsureMetadataCacheTable() error {
storage_path TEXT NOT NULL, storage_path TEXT NOT NULL,
etag TEXT, etag TEXT,
content_type TEXT, content_type TEXT,
content_digest TEXT,
size INTEGER, size INTEGER,
last_modified DATETIME, last_modified DATETIME,
fetched_at DATETIME, fetched_at DATETIME,

View file

@ -78,17 +78,18 @@ func (a *Artifact) IsCached() bool {
// MetadataCacheEntry represents a cached metadata blob for offline serving. // MetadataCacheEntry represents a cached metadata blob for offline serving.
type MetadataCacheEntry struct { type MetadataCacheEntry struct {
ID int64 `db:"id" json:"id"` ID int64 `db:"id" json:"id"`
Ecosystem string `db:"ecosystem" json:"ecosystem"` Ecosystem string `db:"ecosystem" json:"ecosystem"`
Name string `db:"name" json:"name"` Name string `db:"name" json:"name"`
StoragePath string `db:"storage_path" json:"storage_path"` StoragePath string `db:"storage_path" json:"storage_path"`
ETag sql.NullString `db:"etag" json:"etag,omitempty"` ETag sql.NullString `db:"etag" json:"etag,omitempty"`
ContentType sql.NullString `db:"content_type" json:"content_type,omitempty"` ContentType sql.NullString `db:"content_type" json:"content_type,omitempty"`
Size sql.NullInt64 `db:"size" json:"size,omitempty"` ContentDigest sql.NullString `db:"content_digest" json:"content_digest,omitempty"`
LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"` Size sql.NullInt64 `db:"size" json:"size,omitempty"`
FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"` LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"`
CreatedAt time.Time `db:"created_at" json:"created_at"` FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"` CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
} }
// Vulnerability represents a cached vulnerability record. // Vulnerability represents a cached vulnerability record.

View file

@ -12,7 +12,6 @@ import (
const ( const (
dockerHubRegistry = "https://registry-1.docker.io" dockerHubRegistry = "https://registry-1.docker.io"
dockerHubAuth = "https://auth.docker.io"
blobMatchCount = 3 // full match + name + digest blobMatchCount = 3 // full match + name + digest
manifestMatchCount = 3 // full match + name + reference manifestMatchCount = 3 // full match + name + reference
tagsListMatchCount = 2 // full match + name tagsListMatchCount = 2 // full match + name
@ -24,7 +23,6 @@ const (
type ContainerHandler struct { type ContainerHandler struct {
proxy *Proxy proxy *Proxy
registryURL string registryURL string
authURL string
proxyURL string proxyURL string
} }
@ -33,7 +31,6 @@ func NewContainerHandler(proxy *Proxy, proxyURL string) *ContainerHandler {
return &ContainerHandler{ return &ContainerHandler{
proxy: proxy, proxy: proxy,
registryURL: dockerHubRegistry, registryURL: dockerHubRegistry,
authURL: dockerHubAuth,
proxyURL: strings.TrimSuffix(proxyURL, "/"), proxyURL: strings.TrimSuffix(proxyURL, "/"),
} }
} }
@ -90,31 +87,34 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req
h.proxy.Logger.Info("container blob request", "name", name, "digest", digest) h.proxy.Logger.Info("container blob request", "name", name, "digest", digest)
// Get auth token for upstream filename := digest
token, err := h.getAuthToken(r.Context(), name, "pull") cached, err := h.proxy.GetCachedArtifact(r.Context(), "oci", name, digest, filename)
if err != nil { if err != nil {
h.proxy.Logger.Error("failed to get auth token", "error", err) h.proxy.Logger.Error("failed to check blob cache", "error", err)
h.containerError(w, http.StatusUnauthorized, "UNAUTHORIZED", "failed to authenticate") 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")
serveArtifact(w, r.Method, cached)
return return
} }
// For HEAD requests, just proxy to upstream // For HEAD requests, just proxy to upstream
if r.Method == http.MethodHead { if r.Method == http.MethodHead {
h.proxyBlobHead(w, r, name, digest, token) h.proxyBlobHead(w, r, name, digest)
return return
} }
// Try to get from cache, or fetch from upstream with auth // Try to get from cache, or fetch from the authentication-aware upstream client.
filename := digest result, err := h.proxy.GetOrFetchArtifactFromURL(
headers := http.Header{"Authorization": {"Bearer " + token}}
result, err := h.proxy.GetOrFetchArtifactFromURLWithHeaders(
r.Context(), r.Context(),
"oci", "oci",
name, name,
digest, // use digest as version digest, // use digest as version
filename, filename,
fmt.Sprintf("%s/v2/%s/blobs/%s", h.registryURL, name, digest), fmt.Sprintf("%s/v2/%s/blobs/%s", h.registryURL, name, digest),
headers,
) )
if err != nil { if err != nil {
@ -132,8 +132,7 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req
ServeArtifact(w, result) ServeArtifact(w, result)
} }
// handleManifest proxies manifest requests to upstream. // handleManifest serves immutable manifests from cache and revalidates mutable tags.
// Manifests change when tags are updated, so we proxy these directly.
// Path format: {name}/manifests/{reference} // Path format: {name}/manifests/{reference}
func (h *ContainerHandler) handleManifest(w http.ResponseWriter, r *http.Request, path string) { func (h *ContainerHandler) handleManifest(w http.ResponseWriter, r *http.Request, path string) {
if r.Method != http.MethodGet && r.Method != http.MethodHead { if r.Method != http.MethodGet && r.Method != http.MethodHead {
@ -148,57 +147,7 @@ func (h *ContainerHandler) handleManifest(w http.ResponseWriter, r *http.Request
} }
h.proxy.Logger.Info("container manifest request", "name", name, "reference", reference) h.proxy.Logger.Info("container manifest request", "name", name, "reference", reference)
h.serveManifest(w, r, name, 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)
} }
// handleTagsList proxies tag list requests to upstream. // handleTagsList proxies tag list requests to upstream.
@ -214,13 +163,6 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request
return 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) upstreamURL := fmt.Sprintf("%s/v2/%s/tags/list", h.registryURL, name)
if r.URL.RawQuery != "" { if r.URL.RawQuery != "" {
upstreamURL += "?" + r.URL.RawQuery upstreamURL += "?" + r.URL.RawQuery
@ -232,8 +174,6 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request
return return
} }
req.Header.Set("Authorization", "Bearer "+token)
resp, err := h.proxy.HTTPClient.Do(req) resp, err := h.proxy.HTTPClient.Do(req)
if err != nil { if err != nil {
h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream")
@ -246,45 +186,8 @@ func (h *ContainerHandler) handleTagsList(w http.ResponseWriter, r *http.Request
_, _ = io.Copy(w, resp.Body) _, _ = 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. // 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) upstreamURL := fmt.Sprintf("%s/v2/%s/blobs/%s", h.registryURL, name, digest)
req, err := http.NewRequestWithContext(r.Context(), http.MethodHead, upstreamURL, nil) req, err := http.NewRequestWithContext(r.Context(), http.MethodHead, upstreamURL, nil)
@ -293,8 +196,6 @@ func (h *ContainerHandler) proxyBlobHead(w http.ResponseWriter, r *http.Request,
return return
} }
req.Header.Set("Authorization", "Bearer "+token)
resp, err := h.proxy.HTTPClient.Do(req) resp, err := h.proxy.HTTPClient.Do(req)
if err != nil { if err != nil {
h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream") h.containerError(w, http.StatusBadGateway, "INTERNAL_ERROR", "failed to fetch from upstream")

View file

@ -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[:])
}

View file

@ -1,16 +1,15 @@
package handler package handler
import ( import (
"bytes"
"context"
"encoding/json" "encoding/json"
"io" "io"
"log/slog"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strconv"
"testing" "testing"
"time"
"github.com/git-pkgs/proxy/internal/database" upstreamhttp "github.com/git-pkgs/proxy/internal/httpclient"
"github.com/git-pkgs/registries/fetch" "github.com/git-pkgs/registries/fetch"
) )
@ -135,90 +134,521 @@ func TestContainerHandler_parseTagsListPath(t *testing.T) {
} }
} }
func TestContainerHandler_BlobDownload_CachesWithAuth(t *testing.T) { func TestContainerHandler_BlobDownload_DiscoversBearerChallenge(t *testing.T) {
// Set up a mock auth server that returns a token digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd"
authServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { registryRequests := 0
w.Header().Set("Content-Type", "application/json") tokenRequests := 0
_ = json.NewEncoder(w).Encode(map[string]string{"token": "test-token-123"}) 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 proxy, _, _, _ := setupTestProxy(t)
var capturedHeaders http.Header authTransport := upstreamhttp.NewTransport(http.DefaultTransport, nil)
mf := &mockFetcherWithHeaders{ client := &http.Client{Transport: authTransport}
fetchFn: func(_ context.Context, _ string, headers http.Header) (*fetch.Artifact, error) { artifactFetcher := fetch.NewFetcher(
capturedHeaders = headers fetch.WithHTTPClient(client),
return &fetch.Artifact{ fetch.WithMaxRetries(0),
Body: io.NopCloser(bytes.NewReader([]byte("blob-content"))), )
Size: 12, t.Cleanup(func() { _ = artifactFetcher.Close() })
ContentType: "application/octet-stream", proxy.Fetcher = artifactFetcher
}, nil proxy.HTTPClient = client
},
}
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{},
}
h := &ContainerHandler{ h := &ContainerHandler{
proxy: proxy, proxy: proxy,
registryURL: "https://registry-1.docker.io", registryURL: upstream.URL,
authURL: authServer.URL,
proxyURL: "http://localhost:8080", proxyURL: "http://localhost:8080",
} }
handler := h.Routes() for range 2 {
req := httptest.NewRequest(http.MethodGet, "/library/nginx/blobs/sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd", nil) 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_CachedImagePullSurvivesRegistryAndTokenOutages(t *testing.T) {
digest := "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd"
manifest := `{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json"}`
blob := "cached image blob"
registryAvailable := true
tokenAvailable := true
registryRequests := 0
tokenRequests := 0
tokenServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
tokenRequests++
if !tokenAvailable {
http.Error(w, "token service unavailable", http.StatusServiceUnavailable)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"token": "discovered-token",
"expires_in": 3600,
})
}))
defer tokenServer.Close()
registryServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
registryRequests++
if !registryAvailable {
http.Error(w, "registry unavailable", http.StatusServiceUnavailable)
return
}
if r.Header.Get("Authorization") != "Bearer discovered-token" {
w.Header().Set("WWW-Authenticate", `Bearer realm="`+tokenServer.URL+`",service="registry.test",scope="repository:library/nginx:pull"`)
http.Error(w, "authentication required", http.StatusUnauthorized)
return
}
switch r.URL.Path {
case "/v2/library/nginx/manifests/latest":
w.Header().Set("Content-Type", "application/vnd.oci.image.manifest.v1+json")
w.Header().Set("Docker-Content-Digest", digest)
_, _ = io.WriteString(w, manifest)
case "/v2/library/nginx/blobs/" + digest:
w.Header().Set("Content-Type", "application/octet-stream")
_, _ = io.WriteString(w, blob)
default:
http.NotFound(w, r)
}
}))
defer registryServer.Close()
warmProxy, db, store, _ := setupTestProxy(t)
warmClient := &http.Client{Transport: upstreamhttp.NewTransport(http.DefaultTransport, nil)}
warmFetcher := fetch.NewFetcher(
fetch.WithHTTPClient(warmClient),
fetch.WithMaxRetries(0),
)
t.Cleanup(func() { _ = warmFetcher.Close() })
warmProxy.Fetcher = warmFetcher
warmProxy.HTTPClient = warmClient
warmProxy.MetadataTTL = time.Hour
warmHandler := (&ContainerHandler{
proxy: warmProxy,
registryURL: registryServer.URL,
proxyURL: "http://localhost:8080",
}).Routes()
for _, request := range []struct {
path string
body string
}{
{path: "/library/nginx/manifests/latest", body: manifest},
{path: "/library/nginx/blobs/" + digest, body: blob},
} {
response := httptest.NewRecorder()
warmHandler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, request.path, nil))
if response.Code != http.StatusOK {
t.Fatalf("warming %s: status = %d, want %d; body: %s", request.path, response.Code, http.StatusOK, response.Body.String())
}
if got := response.Body.String(); got != request.body {
t.Fatalf("warming %s: body = %q, want %q", request.path, got, request.body)
}
}
warmRegistryRequests := registryRequests
warmTokenRequests := tokenRequests
registryAvailable = false
tokenAvailable = false
offlineClient := &http.Client{Transport: upstreamhttp.NewTransport(http.DefaultTransport, nil)}
offlineFetcher := fetch.NewFetcher(
fetch.WithHTTPClient(offlineClient),
fetch.WithMaxRetries(0),
)
t.Cleanup(func() { _ = offlineFetcher.Close() })
offlineProxy := NewProxy(db, store, offlineFetcher, fetch.NewResolver(), warmProxy.Logger)
offlineProxy.HTTPClient = offlineClient
offlineProxy.MetadataTTL = time.Hour
offlineHandler := (&ContainerHandler{
proxy: offlineProxy,
registryURL: registryServer.URL,
proxyURL: "http://localhost:8080",
}).Routes()
for _, request := range []struct {
name string
path string
body string
}{
{name: "tag manifest", path: "/library/nginx/manifests/latest", body: manifest},
{name: "digest manifest", path: "/library/nginx/manifests/" + digest, body: manifest},
{name: "blob", path: "/library/nginx/blobs/" + digest, body: blob},
} {
t.Run(request.name, func(t *testing.T) {
response := httptest.NewRecorder()
offlineHandler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, request.path, nil))
if response.Code != http.StatusOK {
t.Fatalf("status = %d, want %d; body: %s", response.Code, http.StatusOK, response.Body.String())
}
if got := response.Body.String(); got != request.body {
t.Errorf("body = %q, want %q", got, request.body)
}
if got := response.Header().Get("Docker-Content-Digest"); got != digest {
t.Errorf("Docker-Content-Digest = %q, want %q", got, digest)
}
})
}
if registryRequests != warmRegistryRequests {
t.Errorf("offline registry requests = %d, want 0", registryRequests-warmRegistryRequests)
}
if tokenRequests != warmTokenRequests {
t.Errorf("offline token requests = %d, want 0", tokenRequests-warmTokenRequests)
}
}
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() w := httptest.NewRecorder()
handler.ServeHTTP(w, req) h.Routes().ServeHTTP(w, req)
if w.Code != http.StatusOK { 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())
} }
if got := w.Body.String(); got != "cached blob" {
// Verify auth header was passed to the fetcher t.Errorf("body = %q, want %q", got, "cached blob")
if capturedHeaders == nil {
t.Fatal("expected headers to be passed to fetcher, got nil")
} }
auth := capturedHeaders.Get("Authorization") if upstreamRequests != 0 {
if auth != "Bearer test-token-123" { t.Errorf("upstream requests = %d, want 0", upstreamRequests)
t.Errorf("Authorization = %q, want %q", auth, "Bearer test-token-123")
} }
if fetcher.fetchCalled {
// Verify response headers t.Error("fetcher should not be called on cache hit")
if got := w.Header().Get("Docker-Content-Digest"); got != "sha256:abc123def456abc123def456abc123def456abc123def456abc123def456abcd" {
t.Errorf("Docker-Content-Digest = %q, want digest", got)
} }
} }
// mockFetcherWithHeaders captures headers passed to FetchWithHeaders. func TestContainerHandler_BlobHead_CacheHitSkipsUpstreamAndAuth(t *testing.T) {
type mockFetcherWithHeaders struct { proxy, db, store, fetcher := setupTestProxy(t)
fetchFn func(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) 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) { func TestContainerHandler_BlobHead_DirectServeRedirects(t *testing.T) {
return f.FetchWithHeaders(ctx, url, nil) 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 (f *mockFetcherWithHeaders) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) { func TestContainerHandler_ManifestByDigest_CacheHitSkipsUpstream(t *testing.T) {
return f.fetchFn(ctx, url, headers) 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) Head(_ context.Context, _ string) (int64, string, error) { func TestContainerHandler_ManifestByTag_UsesStaleCacheOnUpstreamFailure(t *testing.T) {
return 0, "", nil 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 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) { func TestContainerHandler_Routes_VersionCheck(t *testing.T) {

View file

@ -136,18 +136,25 @@ type CacheResult struct {
// GetOrFetchArtifact retrieves an artifact from cache or fetches from upstream. // GetOrFetchArtifact retrieves an artifact from cache or fetches from upstream.
func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) { func (p *Proxy) GetOrFetchArtifact(ctx context.Context, ecosystem, name, version, filename string) (*CacheResult, error) {
pkgPURL := purl.MakePURLString(ecosystem, name, "") if cached, err := p.GetCachedArtifact(ctx, ecosystem, name, version, filename); err != nil {
versionPURL := purl.MakePURLString(ecosystem, name, version)
if cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename); err != nil {
return nil, err return nil, err
} else if cached != nil { } else if cached != nil {
return 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) 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. // 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) { func (p *Proxy) checkCache(ctx context.Context, pkgPURL, versionPURL, filename string) (*CacheResult, error) {
pkg, err := p.DB.GetPackageByPURL(pkgPURL) pkg, err := p.DB.GetPackageByPURL(pkgPURL)
@ -363,6 +370,10 @@ func (p *Proxy) updateCacheDB(ecosystem, name, filename, pkgPURL, versionPURL, u
// ServeArtifact writes a CacheResult to an HTTP response. // ServeArtifact writes a CacheResult to an HTTP response.
func ServeArtifact(w http.ResponseWriter, result *CacheResult) { func ServeArtifact(w http.ResponseWriter, result *CacheResult) {
serveArtifact(w, http.MethodGet, result)
}
func serveArtifact(w http.ResponseWriter, method string, result *CacheResult) {
if result.RedirectURL != "" { if result.RedirectURL != "" {
if result.Hash != "" { if result.Hash != "" {
w.Header().Set("ETag", fmt.Sprintf(`"%s"`, result.Hash)) w.Header().Set("ETag", fmt.Sprintf(`"%s"`, result.Hash))
@ -372,12 +383,14 @@ func ServeArtifact(w http.ResponseWriter, result *CacheResult) {
return return
} }
defer func() { _ = result.Reader.Close() }() if result.Reader != nil {
defer func() { _ = result.Reader.Close() }()
}
if result.ContentType != "" { if result.ContentType != "" {
w.Header().Set("Content-Type", result.ContentType) w.Header().Set("Content-Type", result.ContentType)
} }
if result.Size > 0 { if result.Size > 0 || (method == http.MethodHead && result.Size == 0) {
w.Header().Set("Content-Length", fmt.Sprintf("%d", result.Size)) w.Header().Set("Content-Length", fmt.Sprintf("%d", result.Size))
} }
if result.Hash != "" { if result.Hash != "" {
@ -385,7 +398,9 @@ func ServeArtifact(w http.ResponseWriter, result *CacheResult) {
} }
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
_, _ = io.Copy(w, result.Reader) if method != http.MethodHead && result.Reader != nil {
_, _ = io.Copy(w, result.Reader)
}
} }
// ProxyUpstream forwards a request to an upstream URL without caching. // ProxyUpstream forwards a request to an upstream URL without caching.
@ -807,18 +822,16 @@ func (p *Proxy) GetOrFetchArtifactFromURL(ctx context.Context, ecosystem, name,
} }
// GetOrFetchArtifactFromURLWithHeaders retrieves an artifact from cache or fetches from a URL // GetOrFetchArtifactFromURLWithHeaders retrieves an artifact from cache or fetches from a URL
// with additional HTTP headers. This is needed for registries that require authentication // with additional request-specific HTTP headers.
// (e.g. Docker Hub requires a Bearer token even for public images).
func (p *Proxy) GetOrFetchArtifactFromURLWithHeaders(ctx context.Context, ecosystem, name, version, filename, downloadURL string, headers http.Header) (*CacheResult, error) { func (p *Proxy) GetOrFetchArtifactFromURLWithHeaders(ctx context.Context, ecosystem, name, version, filename, downloadURL string, headers http.Header) (*CacheResult, error) {
pkgPURL := purl.MakePURLString(ecosystem, name, "") if cached, err := p.GetCachedArtifact(ctx, ecosystem, name, version, filename); err != nil {
versionPURL := purl.MakePURLString(ecosystem, name, version)
if cached, err := p.checkCache(ctx, pkgPURL, versionPURL, filename); err != nil {
return nil, err return nil, err
} else if cached != nil { } else if cached != nil {
return 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) return p.fetchAndCacheFromURL(ctx, ecosystem, name, version, filename, pkgPURL, versionPURL, downloadURL, headers)
} }

View file

@ -5,7 +5,6 @@ import (
"context" "context"
"database/sql" "database/sql"
"errors" "errors"
"fmt"
"io" "io"
"log/slog" "log/slog"
"net/http" "net/http"
@ -17,6 +16,7 @@ import (
"github.com/git-pkgs/proxy/internal/config" "github.com/git-pkgs/proxy/internal/config"
"github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/database"
"github.com/git-pkgs/proxy/internal/storage" "github.com/git-pkgs/proxy/internal/storage"
"github.com/git-pkgs/purl"
"github.com/git-pkgs/registries/fetch" "github.com/git-pkgs/registries/fetch"
) )
@ -152,7 +152,7 @@ func seedPackage(t *testing.T, db *database.DB, store *mockStorage, ecosystem, n
t.Helper() t.Helper()
pkg := &database.Package{ pkg := &database.Package{
PURL: fmt.Sprintf("pkg:%s/%s", ecosystem, name), PURL: purl.MakePURLString(ecosystem, name, ""),
Ecosystem: ecosystem, Ecosystem: ecosystem,
Name: name, Name: name,
} }
@ -160,7 +160,7 @@ func seedPackage(t *testing.T, db *database.DB, store *mockStorage, ecosystem, n
t.Fatalf("failed to upsert package: %v", err) 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{ ver := &database.Version{
PURL: versionPURL, PURL: versionPURL,
PackagePURL: pkg.PURL, PackagePURL: pkg.PURL,

View file

@ -1,7 +1,6 @@
package handler package handler
import ( import (
"context"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings" "strings"
@ -117,23 +116,12 @@ func TestComposerDownloadUpstreamNotFoundReturns404(t *testing.T) {
} }
func TestContainerBlobUpstreamNotFoundReturns404(t *testing.T) { func TestContainerBlobUpstreamNotFoundReturns404(t *testing.T) {
authServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { proxy, _, _, fetcher := setupTestProxy(t)
w.Header().Set("Content-Type", "application/json") fetcher.fetchErr = fetch.ErrNotFound
_, _ = w.Write([]byte(`{"token": "test-token-123"}`))
}))
defer authServer.Close()
proxy, _, _, _ := setupTestProxy(t)
proxy.Fetcher = &mockFetcherWithHeaders{
fetchFn: func(_ context.Context, _ string, _ http.Header) (*fetch.Artifact, error) {
return nil, fetch.ErrNotFound
},
}
h := &ContainerHandler{ h := &ContainerHandler{
proxy: proxy, proxy: proxy,
registryURL: "https://registry-1.docker.io", registryURL: "https://registry-1.docker.io",
authURL: authServer.URL,
proxyURL: "http://localhost:8080", proxyURL: "http://localhost:8080",
} }