mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-08-01 01:58:09 -04:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bc08021536 | ||
|
|
0ee37c7521 | ||
|
|
f8f475a65d |
17 changed files with 1451 additions and 222 deletions
|
|
@ -96,7 +96,8 @@ upstream:
|
||||||
cargo_download: "https://static.crates.io/crates"
|
cargo_download: "https://static.crates.io/crates"
|
||||||
|
|
||||||
# Authentication for upstream registries
|
# 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.
|
# Values can reference environment variables using ${VAR_NAME} syntax.
|
||||||
#
|
#
|
||||||
# Supported auth types:
|
# Supported auth types:
|
||||||
|
|
|
||||||
|
|
@ -240,6 +240,8 @@ Fetches artifacts from upstream registries.
|
||||||
- Exponential backoff retry on 429 (rate limit) and 5xx errors
|
- Exponential backoff retry on 429 (rate limit) and 5xx errors
|
||||||
- Returns streaming reader (doesn't load into memory)
|
- Returns streaming reader (doesn't load into memory)
|
||||||
- Configurable user-agent
|
- 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:**
|
**Resolver:**
|
||||||
- Determines download URL for a package/version
|
- Determines download URL for a package/version
|
||||||
|
|
@ -351,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
|
||||||
|
|
|
||||||
|
|
@ -123,7 +123,9 @@ upstream:
|
||||||
|
|
||||||
## Authentication
|
## 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
|
### Bearer Token
|
||||||
|
|
||||||
|
|
@ -172,7 +174,7 @@ upstream:
|
||||||
|
|
||||||
### URL Matching
|
### 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
|
```yaml
|
||||||
upstream:
|
upstream:
|
||||||
|
|
@ -244,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.
|
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
|
||||||
```
|
```
|
||||||
|
|
|
||||||
|
|
@ -274,23 +274,29 @@ type UpstreamConfig struct {
|
||||||
CargoDownload string `json:"cargo_download" yaml:"cargo_download"`
|
CargoDownload string `json:"cargo_download" yaml:"cargo_download"`
|
||||||
|
|
||||||
// Auth configures authentication for upstream registries.
|
// 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.
|
// Example: "https://npm.pkg.github.com" matches all requests to that host.
|
||||||
Auth map[string]AuthConfig `json:"auth" yaml:"auth"`
|
Auth map[string]AuthConfig `json:"auth" yaml:"auth"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// AuthForURL returns the auth config that matches the given URL.
|
// 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 {
|
func (u *UpstreamConfig) AuthForURL(url string) *AuthConfig {
|
||||||
if u.Auth == nil {
|
if u.Auth == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
target, err := parseAuthURL(url)
|
||||||
|
if err != nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
var bestMatch *AuthConfig
|
var bestMatch *AuthConfig
|
||||||
var bestLen int
|
var bestLen int
|
||||||
|
|
||||||
for pattern, auth := range u.Auth {
|
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
|
a := auth // copy to avoid loop variable capture
|
||||||
bestMatch = &a
|
bestMatch = &a
|
||||||
bestLen = len(pattern)
|
bestLen = len(pattern)
|
||||||
|
|
@ -300,6 +306,45 @@ func (u *UpstreamConfig) AuthForURL(url string) *AuthConfig {
|
||||||
return bestMatch
|
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.
|
// AuthConfig configures authentication for an upstream registry.
|
||||||
type AuthConfig struct {
|
type AuthConfig struct {
|
||||||
// Type is the authentication type: "bearer", "basic", or "header".
|
// Type is the authentication type: "bearer", "basic", or "header".
|
||||||
|
|
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -30,6 +30,10 @@ 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},
|
||||||
|
ContentDigest: sql.NullString{
|
||||||
|
String: "sha256:0123456789abcdef",
|
||||||
|
Valid: true,
|
||||||
|
},
|
||||||
Size: sql.NullInt64{Int64: 1024, Valid: true},
|
Size: sql.NullInt64{Int64: 1024, Valid: true},
|
||||||
FetchedAt: sql.NullTime{Time: time.Now(), Valid: true},
|
FetchedAt: sql.NullTime{Time: time.Now(), Valid: true},
|
||||||
}
|
}
|
||||||
|
|
@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -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)
|
||||||
|
|
|
||||||
|
|
@ -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,
|
||||||
|
|
|
||||||
|
|
@ -84,6 +84,7 @@ type MetadataCacheEntry struct {
|
||||||
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"`
|
||||||
|
ContentDigest sql.NullString `db:"content_digest" json:"content_digest,omitempty"`
|
||||||
Size sql.NullInt64 `db:"size" json:"size,omitempty"`
|
Size sql.NullInt64 `db:"size" json:"size,omitempty"`
|
||||||
LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"`
|
LastModified sql.NullTime `db:"last_modified" json:"last_modified,omitempty"`
|
||||||
FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"`
|
FetchedAt sql.NullTime `db:"fetched_at" json:"fetched_at,omitempty"`
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,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
|
||||||
|
|
@ -23,7 +22,6 @@ const (
|
||||||
type ContainerHandler struct {
|
type ContainerHandler struct {
|
||||||
proxy *Proxy
|
proxy *Proxy
|
||||||
registryURL string
|
registryURL string
|
||||||
authURL string
|
|
||||||
proxyURL string
|
proxyURL string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -32,7 +30,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, "/"),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -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)
|
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")
|
||||||
|
if r.Method == http.MethodHead {
|
||||||
|
serveArtifactHead(w, cached)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ServeArtifact(w, 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 {
|
||||||
|
|
@ -127,8 +131,31 @@ func (h *ContainerHandler) handleBlobDownload(w http.ResponseWriter, r *http.Req
|
||||||
ServeArtifact(w, result)
|
ServeArtifact(w, result)
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleManifest proxies manifest requests to upstream.
|
func serveArtifactHead(w http.ResponseWriter, result *CacheResult) {
|
||||||
// Manifests change when tags are updated, so we proxy these directly.
|
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()
|
||||||
|
}
|
||||||
|
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}
|
// 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 {
|
||||||
|
|
@ -143,57 +170,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.
|
||||||
|
|
@ -209,13 +186,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
|
||||||
|
|
@ -227,8 +197,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")
|
||||||
|
|
@ -241,45 +209,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)
|
||||||
|
|
@ -288,8 +219,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")
|
||||||
|
|
|
||||||
251
internal/handler/container_manifest.go
Normal file
251
internal/handler/container_manifest.go
Normal 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[:])
|
||||||
|
}
|
||||||
|
|
@ -1,16 +1,14 @@
|
||||||
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"
|
||||||
|
|
||||||
"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 +133,387 @@ 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
|
||||||
|
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")
|
w.Header().Set("Content-Type", "application/json")
|
||||||
_ = json.NewEncoder(w).Encode(map[string]string{"token": "test-token-123"})
|
_ = 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()
|
w := httptest.NewRecorder()
|
||||||
handler.ServeHTTP(w, req)
|
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()
|
||||||
|
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",
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f *mockFetcherWithHeaders) Fetch(ctx context.Context, url string) (*fetch.Artifact, error) {
|
req := httptest.NewRequest(http.MethodHead, "/library/nginx/blobs/"+digest, nil)
|
||||||
return f.FetchWithHeaders(ctx, url, 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) FetchWithHeaders(ctx context.Context, url string, headers http.Header) (*fetch.Artifact, error) {
|
func TestContainerHandler_BlobHead_DirectServeRedirects(t *testing.T) {
|
||||||
return f.fetchFn(ctx, url, headers)
|
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",
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f *mockFetcherWithHeaders) Head(_ context.Context, _ string) (int64, string, error) {
|
req := httptest.NewRequest(http.MethodHead, "/library/nginx/blobs/"+digest, nil)
|
||||||
return 0, "", 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"}`
|
||||||
|
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 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 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) {
|
||||||
|
|
|
||||||
|
|
@ -127,18 +127,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)
|
||||||
|
|
@ -766,18 +773,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)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,6 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"io"
|
"io"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
@ -16,6 +15,7 @@ import (
|
||||||
|
|
||||||
"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"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -151,7 +151,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,
|
||||||
}
|
}
|
||||||
|
|
@ -159,7 +159,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,
|
||||||
|
|
|
||||||
418
internal/httpclient/transport.go
Normal file
418
internal/httpclient/transport.go
Normal file
|
|
@ -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()
|
||||||
|
}
|
||||||
151
internal/httpclient/transport_test.go
Normal file
151
internal/httpclient/transport_test.go
Normal file
|
|
@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -58,17 +58,19 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/git-pkgs/cooldown"
|
||||||
swaggerdoc "github.com/git-pkgs/proxy/docs/swagger"
|
swaggerdoc "github.com/git-pkgs/proxy/docs/swagger"
|
||||||
"github.com/git-pkgs/proxy/internal/config"
|
"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/database"
|
||||||
"github.com/git-pkgs/proxy/internal/enrichment"
|
"github.com/git-pkgs/proxy/internal/enrichment"
|
||||||
"github.com/git-pkgs/proxy/internal/handler"
|
"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/metrics"
|
||||||
"github.com/git-pkgs/proxy/internal/mirror"
|
"github.com/git-pkgs/proxy/internal/mirror"
|
||||||
"github.com/git-pkgs/proxy/internal/storage"
|
"github.com/git-pkgs/proxy/internal/storage"
|
||||||
"github.com/git-pkgs/purl"
|
"github.com/git-pkgs/purl"
|
||||||
"github.com/git-pkgs/registries/fetch"
|
"github.com/git-pkgs/registries/fetch"
|
||||||
|
"github.com/git-pkgs/registries/safehttp"
|
||||||
"github.com/git-pkgs/spdx"
|
"github.com/git-pkgs/spdx"
|
||||||
"github.com/go-chi/chi/v5"
|
"github.com/go-chi/chi/v5"
|
||||||
"github.com/go-chi/chi/v5/middleware"
|
"github.com/go-chi/chi/v5/middleware"
|
||||||
|
|
@ -156,8 +158,18 @@ func New(cfg *config.Config, logger *slog.Logger) (*Server, error) {
|
||||||
|
|
||||||
// Start starts the HTTP server.
|
// Start starts the HTTP server.
|
||||||
func (s *Server) Start() error {
|
func (s *Server) Start() error {
|
||||||
// Create shared components with circuit breaker
|
// Use one authentication-aware transport for metadata and artifacts so
|
||||||
baseFetcher := fetch.NewFetcher(fetch.WithAuthFunc(s.authForURL))
|
// 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)
|
fetcher := fetch.NewCircuitBreakerFetcher(baseFetcher)
|
||||||
resolver := fetch.NewResolver()
|
resolver := fetch.NewResolver()
|
||||||
cd := &cooldown.Config{
|
cd := &cooldown.Config{
|
||||||
|
|
@ -166,7 +178,7 @@ func (s *Server) Start() error {
|
||||||
Packages: s.cfg.Cooldown.Packages,
|
Packages: s.cfg.Cooldown.Packages,
|
||||||
}
|
}
|
||||||
proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger)
|
proxy := handler.NewProxy(s.db, s.storage, fetcher, resolver, s.logger)
|
||||||
proxy.HTTPClient.Timeout = s.cfg.ParseHTTPTimeout()
|
proxy.HTTPClient = &metadataClient
|
||||||
proxy.Cooldown = cd
|
proxy.Cooldown = cd
|
||||||
proxy.CacheMetadata = s.cfg.CacheMetadata
|
proxy.CacheMetadata = s.cfg.CacheMetadata
|
||||||
proxy.MetadataTTL = s.cfg.ParseMetadataTTL()
|
proxy.MetadataTTL = s.cfg.ParseMetadataTTL()
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue