Compare commits

..

9 Commits

Author SHA1 Message Date
unkin-agent 841b9f1547 fix vacuous e2e dedupe assertion
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
2026-10-17 12:56:42 +11:00
unkin-agent ed6216d78b Serve deb virtual indexes by hash and bound cached suites
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
2026-10-17 12:46:14 +11:00
unkin-agent eca686b7bb Merge remote-tracking branch 'origin/benvin/virtual-shared-cache' into benvin/virtual-deb
# Conflicts:
#	internal/virtual/engine.go
#	internal/virtual/engine_test.go
2026-10-17 12:32:38 +11:00
unkin-agent ceae176089 Share the virtual merge cache across index types
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
2026-10-17 12:31:49 +11:00
unkin-agent bc68eb75fc Add nuget v3 remote type (#141)
.NET builds restore from nuget.org directly because artifactapi has no NuGet remote type.

- add `nuget` package type proxying a NuGet v3 feed; `index.json` maps to the upstream service index
- add a `ServeRewriter` provider hook so cached JSON has upstream-origin URLs pointed back at the requesting host per request
- treat `.nupkg`/`.nuspec` as immutable and service, registration and flat container indexes as mutable
- add a `nuget.config` usage snippet to the UI and a nuget e2e caching case

Reviewed-on: #141
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-17 12:19:51 +11:00
unkin-agent 78d24f7752 Add deb virtual repositories
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful
- merge member Packages indexes (flat and dists/<suite>) into one apt repo with a generated unsigned Release
- prefix Filename with the member name and 302 downloads to the member route
- share the rpm merge cache, singleflight and fail-closed member fetch
2026-10-16 23:06:56 +11:00
unkin-agent 9161558334 Add optional OIDC bearer auth on remote/virtual config writes (#136)
ci/woodpecker/tag/docker Pipeline was successful
Remote and virtual configuration writes are unauthenticated, so anything that can reach the API can repoint or delete repos. Vault identity OIDC tokens give callers a verifiable identity.

- add `ARTIFACTAPI_OIDC_ISSUER`/`ARTIFACTAPI_OIDC_AUDIENCE`; unset issuer keeps auth off
- require a JWKS-verified bearer token on POST/PUT/DELETE of `/api/v2/remotes` and `/api/v2/virtuals`, 401 otherwise
- leave uploads, object deletes, probe, reads and `/v2` open
- discover the issuer lazily with a 10s client timeout; fail fast with 503 for 15s after a failed discovery
- log the verified `sub`; allow `Authorization` in CORS

Reviewed-on: #136
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-16 22:08:20 +11:00
unkin-agent 492607a164 Retry failed GitHub release scans with backoff (#133)
ci/woodpecker/tag/docker Pipeline was successful
Releasing a GitHub sync lease always advanced `last_synced_at`, even after a failed scan (e.g. a rate-limit 403). A remote that failed once waited a full `mutable_ttl` before retrying, and a cold remote kept returning 503 until then.

- record scan outcomes in one shared lease helper for github_rpm/deb/alpine
- keep `last_synced_at` and the ETag on failure; retry from 60s with exponential backoff, capped at min(10m, ttl/4)
- honour `Retry-After` / `X-RateLimit-Reset`, clamped to `mutable_ttl`
- add `sync_failures` / `next_retry_at` columns (migration 0002)

Reviewed-on: #133
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-09 23:15:55 +11:00
unkin-agent 1780b3d77c Push release images to artifactapi docker-internal (#134)
Gitea's registry rejects blob HEADs with 401, so the v3.13.0 tag pipeline cannot push. Release images move to the in-cluster artifactapi docker-internal registry, which trusts the runner without credentials.

- push docker-api to docker-internal/artifactapi and docker-web to docker-internal/artifactapi-ui
- use the CA-baked docker-internal/plugin-docker-buildx image
- drop the droneci credentials
- set kubernetes resources and service account on both steps

Reviewed-on: #134
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-09 23:08:22 +11:00
47 changed files with 2224 additions and 195 deletions
+26 -12
View File
@@ -4,31 +4,45 @@ when:
steps:
- name: docker-api
image: woodpeckerci/plugin-docker-buildx
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
settings:
registry: git.unkin.net
repo: git.unkin.net/unkin/artifactapi
registry: artifactapi.k8s.syd1.au.unkin.net
repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/artifactapi
build_args:
VERSION: ${CI_COMMIT_TAG}
username: droneci
password:
from_secret: DRONECI_PASSWORD
tags:
- ${CI_COMMIT_TAG}
- latest
backend_options:
kubernetes:
serviceAccountName: default
resources:
requests:
memory: 1Gi
cpu: 1
limits:
memory: 4Gi
cpu: 2
- name: docker-web
image: woodpeckerci/plugin-docker-buildx
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
settings:
registry: git.unkin.net
repo: git.unkin.net/unkin/artifactapi-ui
registry: artifactapi.k8s.syd1.au.unkin.net
repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/artifactapi-ui
dockerfile: ui/Dockerfile.ui
context: ui
build_args:
BASE_PATH: /ui
username: droneci
password:
from_secret: DRONECI_PASSWORD
tags:
- ${CI_COMMIT_TAG}
- latest
backend_options:
kubernetes:
serviceAccountName: default
resources:
requests:
memory: 1Gi
cpu: 1
limits:
memory: 4Gi
cpu: 2
+4 -1
View File
@@ -1,6 +1,6 @@
# ArtifactAPI
Caching proxy for package repositories. Single Go binary, 11 package types, content-addressable storage, managed by Terraform.
Caching proxy for package repositories. Single Go binary, 12 package types, content-addressable storage, managed by Terraform.
## Quick Start
@@ -33,6 +33,7 @@ API: `http://localhost:8000` | Frontend: `http://localhost:5173`
| `terraform` | `*/versions` | `*/download/*/*` |
| `goproxy` | `@v/list`, `@latest` | `.info`, `.mod`, `.zip` |
| `cargo` | sparse index files (`config.json` synthesized) | `crates/*/*.crate` |
| `nuget` | `index.json` (service index), registration and flat container indexes | `.nupkg`, `.nuspec` |
| `github_rpm` | `repodata/*` (synthesized) | `.rpm` (redirected) |
Providers classify paths automatically. Users only configure what to proxy and TTLs.
@@ -371,6 +372,8 @@ additive and idempotent for the same reason; a test enforces it.
| `GITHUB_APP_INSTALLATION_ID` | | GitHub App installation id |
| `GITHUB_APP_PRIVATE_KEY` | | GitHub App private key, inline PEM |
| `GITHUB_APP_PRIVATE_KEY_PATH` | | GitHub App private key, file path (alternative to inline PEM) |
| `ARTIFACTAPI_OIDC_ISSUER` | | OIDC issuer URL (e.g. Vault `https://vault.service.consul:8200/v1/identity/oidc`). When set, `POST /api/v2/remotes`, `POST /api/v2/virtuals` and `PUT`/`DELETE` on `/api/v2/{remotes,virtuals}/{name}` require `Authorization: Bearer <jwt>` verified against the issuer's JWKS. Uploads, object/file deletes, probe, reads and `/v2` stay open. Unset = no auth |
| `ARTIFACTAPI_OIDC_AUDIENCE` | | Comma-separated allowed `aud` values (e.g. the Vault OIDC role's client_id). Required when `ARTIFACTAPI_OIDC_ISSUER` is set |
## Development
+1 -1
View File
@@ -23,7 +23,7 @@ already-running stack.
- **Repository lifecycle** — add / change / delete for remote, local and virtual repos.
- **Caching** — one immutable artifact per remote package type (generic, docker,
helm, pypi, npm, rpm, alpine, puppet, terraform, goproxy, cargo) proxied through the
helm, pypi, npm, rpm, alpine, puppet, terraform, goproxy, cargo, nuget) proxied through the
mock upstream: first fetch `X-Artifact-Source: remote`, second `cache`, bytes
verified against the origin fixture.
- **Local uploads** — generic (upload/download), pypi (wheel + generated `simple/`
+1
View File
@@ -30,6 +30,7 @@ func TestCachingPerProvider(t *testing.T) {
{"puppet", "puppet-releases/author-mod-1.0.0.tar.gz", "puppet-releases/author-mod-1.0.0.tar.gz"},
{"goproxy", "goproxy/example.com/mod/@v/v1.0.0.zip", "goproxy/example.com/mod/@v/v1.0.0.zip"},
{"cargo", "crates/mycrate/mycrate-1.0.0.crate", "crates/mycrate/mycrate-1.0.0.crate"},
{"nuget", "nuget/e2epkg/1.0.0/e2epkg.1.0.0.nupkg", "nuget/e2epkg/1.0.0/e2epkg.1.0.0.nupkg"},
{"terraform", "hashicorp/aws/download/pkg.zip", "v1/providers/hashicorp/aws/download/pkg.zip"},
{"docker", "library/testimg/blobs/blobdata", "v2/library/testimg/blobs/blobdata"},
}
+103
View File
@@ -14,6 +14,8 @@ import (
"strings"
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
)
// TestVirtualPyPIMerge uploads different packages to two pypi locals and
@@ -136,3 +138,104 @@ func TestVirtualRPMMerge(t *testing.T) {
t.Fatalf("real dnf against rpm virtual failed: %v\n%s", err, out)
}
}
// TestVirtualDebMerge merges two local deb repos carrying the same package: the
// first member wins the duplicate, and a real apt sees both members' packages
// and installs through the virtual.
func TestVirtualDebMerge(t *testing.T) {
createRepo(t, `{"name":"vdeb-a","package_type":"deb","repo_type":"local"}`)
createRepo(t, `{"name":"vdeb-b","package_type":"deb","repo_type":"local"}`)
defer deleteRepo(t, "vdeb-a")
defer deleteRepo(t, "vdeb-b")
uploadFile(t, "vdeb-a", "vdeb-one_1.0.0_amd64.deb", testsupport.MinimalDeb("vdeb-one", "1.0.0", "amd64"), "application/vnd.debian.binary-package")
uploadFile(t, "vdeb-b", "vdeb-one_1.0.0_amd64.deb", testsupport.MinimalDeb("vdeb-one", "1.0.0", "amd64"), "application/vnd.debian.binary-package")
uploadFile(t, "vdeb-b", "vdeb-two_2.0.0_amd64.deb", testsupport.MinimalDeb("vdeb-two", "2.0.0", "amd64"), "application/vnd.debian.binary-package")
// Local indexes regenerate asynchronously; merge only once vdeb-b lists both.
deadline := time.Now().Add(15 * time.Second)
for {
resp, body := doRequest(t, http.MethodGet, api("/api/v1/local/vdeb-b/Packages"), nil, "")
if resp.StatusCode == http.StatusOK && strings.Contains(string(body), "vdeb-two") || time.Now().After(deadline) {
break
}
time.Sleep(500 * time.Millisecond)
}
createVirtual(t, `{"name":"vdeb","package_type":"deb","members":["vdeb-a","vdeb-b"]}`)
defer deleteVirtual(t, "vdeb")
_, body := doRequest(t, http.MethodGet, api("/api/v1/virtual/vdeb/Packages"), nil, "")
pk := string(body)
if strings.Count(pk, "Package: vdeb-one\n") != 1 || !strings.Contains(pk, "Filename: vdeb-a/pool/vdeb-one_1.0.0_amd64.deb") ||
!strings.Contains(pk, "Filename: vdeb-b/pool/vdeb-two_2.0.0_amd64.deb") {
t.Fatalf("merged Packages should hold vdeb-one from the first member and vdeb-two:\n%s", pk)
}
network := os.Getenv("COMPOSE_NETWORK")
internal := os.Getenv("ARTIFACTAPI_INTERNAL")
if network == "" || internal == "" {
t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; skipping real apt")
}
if _, err := exec.LookPath("docker"); err != nil {
t.Skip("docker not available on the test host")
}
script := "set -euo pipefail; " +
"rm -f /etc/apt/sources.list.d/*; " +
"echo \"deb [trusted=yes] $URL ./\" > /etc/apt/sources.list.d/vdeb.list; " +
"apt-get update; " +
"apt-cache policy vdeb-one vdeb-two | tee /dev/stderr | grep -c virtual/vdeb | grep -qx 2; " +
"apt-get install -y vdeb-one vdeb-two; " +
"dpkg -s vdeb-one vdeb-two"
out, err := exec.Command("docker", "run", "--rm", "--network", network,
"-e", "URL="+strings.TrimRight(internal, "/")+"/api/v1/virtual/vdeb/",
"debian:13", "bash", "-c", script).CombinedOutput()
if err != nil {
t.Fatalf("real apt against deb virtual failed: %v\n%s", err, out)
}
}
// TestVirtualDebSuite merges two proxied deb.debian.org remotes and drives a
// real apt through the virtual's dists/<suite> layout (by-hash indexes).
func TestVirtualDebSuite(t *testing.T) {
network := os.Getenv("COMPOSE_NETWORK")
internal := os.Getenv("ARTIFACTAPI_INTERNAL")
if network == "" || internal == "" {
t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; skipping real apt")
}
if _, err := exec.LookPath("docker"); err != nil {
t.Skip("docker not available on the test host")
}
for _, n := range []string{"vdebs-a", "vdebs-b"} {
createRepo(t, `{"name":"`+n+`","package_type":"deb","repo_type":"remote","base_url":"http://deb.debian.org/debian"}`)
defer deleteRepo(t, n)
}
createVirtual(t, `{"name":"vdebs","package_type":"deb","members":["vdebs-a","vdebs-b"]}`)
defer deleteVirtual(t, "vdebs")
resp, rel := doRequest(t, http.MethodGet, api("/api/v1/virtual/vdebs/dists/trixie-updates/Release"), nil, "")
if resp.StatusCode != http.StatusOK || !strings.Contains(string(rel), "Acquire-By-Hash: yes\n") ||
!strings.Contains(string(rel), " main/binary-amd64/Packages.gz\n") {
t.Fatalf("merged suite Release: status %d:\n%s", resp.StatusCode, rel)
}
script := "set -euo pipefail; " +
"rm -f /etc/apt/sources.list.d/*; " +
"echo \"deb [trusted=yes] $URL trixie-updates main\" > /etc/apt/sources.list.d/vdebs.list; " +
"apt-get update; " +
"idx=/tmp/avail; apt-cache dumpavail > $idx; " +
"grep -q '^Filename: vdebs-a/pool/' $idx; if grep -q '^Filename: vdebs-b/' $idx; then echo 'vdebs-b not deduped'; exit 1; fi; " +
"pkg=$(awk '/^Package:/{p=$2} /^Size:/{if(!m||$2<m){m=$2;b=p}} END{print b}' $idx); " +
"cd /tmp && apt-get download $pkg && ls -l /tmp/*.deb"
out, err := exec.Command("docker", "run", "--rm", "--network", network,
"-e", "URL="+strings.TrimRight(internal, "/")+"/api/v1/virtual/vdebs/",
"debian:13", "bash", "-c", script).CombinedOutput()
if err != nil {
t.Fatalf("real apt against deb virtual suite failed: %v\n%s", err, out)
}
t.Logf("%s", out)
resp, _ = doRequest(t, http.MethodGet, api("/api/v1/remote/vdebs-a/dists/trixie-updates/main/binary-amd64/Packages.xz"), nil, "")
if src := resp.Header.Get("X-Artifact-Source"); resp.StatusCode != http.StatusOK || src != "cache" {
t.Fatalf("member index the virtual merged should be cached: status %d source %q", resp.StatusCode, src)
}
}
+3
View File
@@ -7,7 +7,9 @@ require (
github.com/cavaliergopher/rpm v1.3.0
github.com/charmbracelet/bubbletea v1.3.10
github.com/charmbracelet/lipgloss v1.1.0
github.com/coreos/go-oidc/v3 v3.21.0
github.com/go-chi/chi/v5 v5.3.0
github.com/go-jose/go-jose/v4 v4.1.4
github.com/google/uuid v1.6.0
github.com/jackc/pgx/v5 v5.10.0
github.com/klauspost/compress v1.19.2
@@ -102,6 +104,7 @@ require (
go.uber.org/atomic v1.11.0 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect
golang.org/x/net v0.56.0 // indirect
golang.org/x/oauth2 v0.36.0 // indirect
golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect
gopkg.in/ini.v1 v1.67.2 // indirect
+6
View File
@@ -44,6 +44,8 @@ github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A=
github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw=
github.com/coreos/go-oidc/v3 v3.21.0 h1:wZo4Q9Pum8dYEj0eMUPrqR+kvuGkeUplbLpNCkBqoWM=
github.com/coreos/go-oidc/v3 v3.21.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4=
github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA=
github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
@@ -67,6 +69,8 @@ github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeO
github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE=
github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -225,6 +229,8 @@ golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM
golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+13 -1
View File
@@ -1,6 +1,7 @@
package v1
import (
"bytes"
"errors"
"fmt"
"io"
@@ -105,13 +106,24 @@ func (h *ProxyHandler) handleProxy(w http.ResponseWriter, r *http.Request) {
}
defer result.Reader.Close()
var body io.Reader = result.Reader
if sr, ok := prov.(provider.ServeRewriter); ok && prov.Classify(path) == provider.Mutable {
raw, err := io.ReadAll(result.Reader)
if err != nil {
http.Error(w, "bad gateway", http.StatusBadGateway)
return
}
raw = sr.RewriteServed(raw, *remote, fmt.Sprintf("%s://%s", scheme(r), r.Host))
body, result.Size = bytes.NewReader(raw), int64(len(raw))
}
w.Header().Set("Content-Type", result.ContentType)
w.Header().Set("X-Artifact-Source", result.Source)
if result.Size > 0 {
w.Header().Set("X-Artifact-Size", fmt.Sprintf("%d", result.Size))
}
w.WriteHeader(http.StatusOK)
io.Copy(w, result.Reader)
io.Copy(w, body)
}
func (h *ProxyHandler) handleProxyHead(w http.ResponseWriter, r *http.Request) {
+158
View File
@@ -0,0 +1,158 @@
package auth
import (
"context"
"errors"
"log/slog"
"net/http"
"slices"
"strings"
"sync"
"time"
"github.com/coreos/go-oidc/v3/oidc"
)
// BearerOIDC verifies `Authorization: Bearer <jwt>` against an OIDC issuer's
// JWKS. Discovery is lazy so startup never depends on the issuer being
// reachable; a failed discovery is not retried until the backoff elapses.
type BearerOIDC struct {
issuer string
audiences []string
timeout time.Duration
backoff time.Duration
discover func(context.Context, string) (*oidc.Provider, error)
mu sync.Mutex
verifier *oidc.IDTokenVerifier
inflight chan struct{}
failedAt time.Time
}
var errIssuerUnavailable = errors.New("oidc issuer unavailable")
// NewBearerOIDC returns nil when issuer is empty, meaning auth is disabled.
func NewBearerOIDC(issuer string, audiences []string) *BearerOIDC {
if issuer == "" {
return nil
}
return &BearerOIDC{issuer: issuer, audiences: audiences, timeout: 10 * time.Second, backoff: 15 * time.Second, discover: oidc.NewProvider}
}
// getVerifier runs at most one discovery at a time without holding the lock
// across network I/O; concurrent callers wait for it, and only callers within
// the post-failure backoff fail fast.
func (b *BearerOIDC) getVerifier(ctx context.Context) (*oidc.IDTokenVerifier, error) {
b.mu.Lock()
if v := b.verifier; v != nil {
b.mu.Unlock()
return v, nil
}
if done := b.inflight; done != nil {
b.mu.Unlock()
select {
case <-done:
case <-ctx.Done():
return nil, ctx.Err()
}
b.mu.Lock()
defer b.mu.Unlock()
if b.verifier == nil {
return nil, errIssuerUnavailable
}
return b.verifier, nil
}
if time.Since(b.failedAt) < b.backoff {
b.mu.Unlock()
return nil, errIssuerUnavailable
}
done := make(chan struct{})
b.inflight = done
b.mu.Unlock()
v, err := b.discoverOnce(done)
if err != nil {
slog.Error("oidc discovery failed", "issuer", b.issuer, "error", err)
return nil, err
}
return v, nil
}
// discoverOnce releases inflight and waiters on every exit, a panic included,
// recording failedAt unless discovery succeeded.
func (b *BearerOIDC) discoverOnce(done chan struct{}) (v *oidc.IDTokenVerifier, err error) {
defer func() {
b.mu.Lock()
b.inflight = nil
if v == nil {
b.failedAt = time.Now()
}
b.verifier = v
b.mu.Unlock()
close(done)
}()
// The provider keeps this context and client for JWKS refreshes, so the
// deadline lives on the client rather than the context.
client := &http.Client{Timeout: b.timeout}
p, err := b.discover(oidc.ClientContext(context.Background(), client), b.issuer)
if err != nil {
return nil, err
}
return p.Verifier(&oidc.Config{SkipClientIDCheck: true}), nil
}
// Require gates every request through next on a valid token. A nil receiver
// passes everything through.
func (b *BearerOIDC) Require(next http.Handler) http.Handler {
if b == nil {
return next
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
scheme, raw, _ := strings.Cut(strings.TrimSpace(r.Header.Get("Authorization")), " ")
raw = strings.TrimSpace(raw)
if !strings.EqualFold(scheme, "Bearer") || raw == "" {
unauthorized(w, "missing bearer token")
return
}
v, err := b.getVerifier(r.Context())
if err != nil {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusServiceUnavailable)
_, _ = w.Write([]byte(`{"error":"token issuer unavailable"}`))
return
}
tok, err := v.Verify(r.Context(), raw)
if err != nil {
slog.Info("bearer token rejected", "error", err)
unauthorized(w, "invalid token")
return
}
if len(b.audiences) > 0 && !slices.ContainsFunc(tok.Audience, func(a string) bool { return slices.Contains(b.audiences, a) }) {
slog.Info("bearer token rejected", "error", "audience not allowed", "aud", tok.Audience)
unauthorized(w, "invalid token")
return
}
slog.Info("bearer token accepted", "sub", tok.Subject, "method", r.Method, "path", r.URL.Path)
next.ServeHTTP(w, r)
})
}
// RequireWrites gates only mutating methods.
func (b *BearerOIDC) RequireWrites(next http.Handler) http.Handler {
gated := b.Require(next)
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet, http.MethodHead, http.MethodOptions:
next.ServeHTTP(w, r)
default:
gated.ServeHTTP(w, r)
}
})
}
func unauthorized(w http.ResponseWriter, msg string) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("WWW-Authenticate", "Bearer")
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"error":"` + msg + `"}`))
}
+255
View File
@@ -0,0 +1,255 @@
package auth
import (
"context"
"crypto/rand"
"crypto/rsa"
"encoding/base64"
"net/http"
"net/http/httptest"
"strconv"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/coreos/go-oidc/v3/oidc"
"github.com/go-jose/go-jose/v4"
"github.com/go-jose/go-jose/v4/jwt"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
)
func serve(h http.Handler, method, token string) int {
if token == "" {
return serveAuth(h, method, "")
}
return serveAuth(h, method, "Bearer "+token)
}
func serveAuth(h http.Handler, method, authz string) int {
r := httptest.NewRequest(method, "/x", nil)
if authz != "" {
r.Header.Set("Authorization", authz)
}
w := httptest.NewRecorder()
h.ServeHTTP(w, r)
return w.Code
}
func claims(iss string) jwt.Claims {
now := time.Now()
return jwt.Claims{
Issuer: iss,
Subject: "ci",
Audience: jwt.Audience{"artifactapi"},
IssuedAt: jwt.NewNumericDate(now.Add(-time.Minute)),
Expiry: jwt.NewNumericDate(now.Add(time.Hour)),
}
}
func hs256Token(t *testing.T, c jwt.Claims) string {
sig, err := jose.NewSigner(jose.SigningKey{Algorithm: jose.HS256, Key: []byte("0123456789abcdef0123456789abcdef")},
(&jose.SignerOptions{}).WithType("JWT").WithHeader("kid", "k1"))
if err != nil {
t.Fatal(err)
}
tok, err := jwt.Signed(sig).Claims(c).Serialize()
if err != nil {
t.Fatal(err)
}
return tok
}
func noneToken(iss string) string {
enc := base64.RawURLEncoding.EncodeToString
return enc([]byte(`{"alg":"none","typ":"JWT","kid":"k1"}`)) + "." +
enc([]byte(`{"iss":"`+iss+`","sub":"ci","aud":"artifactapi","exp":`+strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10)+`}`)) + "."
}
var ok = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) })
func TestBearerOIDC(t *testing.T) {
iss := testsupport.NewOIDCIssuer()
defer iss.Close()
h := NewBearerOIDC(iss.URL, []string{"other", "artifactapi"}).Require(ok)
otherKey, _ := rsa.GenerateKey(rand.Reader, 2048)
cases := []struct {
name string
token string
want int
}{
{"valid", iss.Token(nil, "artifactapi", "ci", time.Hour), http.StatusNoContent},
{"missing", "", http.StatusUnauthorized},
{"garbage", "not.a.jwt", http.StatusUnauthorized},
{"bad signature", iss.Token(otherKey, "artifactapi", "ci", time.Hour), http.StatusUnauthorized},
{"wrong aud", iss.Token(nil, "nope", "ci", time.Hour), http.StatusUnauthorized},
{"expired", iss.Token(nil, "artifactapi", "ci", -time.Hour), http.StatusUnauthorized},
{"wrong iss", iss.Sign(nil, claims("https://evil.example")), http.StatusUnauthorized},
{"hs256", hs256Token(t, claims(iss.URL)), http.StatusUnauthorized},
{"alg none", noneToken(iss.URL), http.StatusUnauthorized},
}
for _, c := range cases {
if got := serve(h, http.MethodPost, c.token); got != c.want {
t.Errorf("%s: status %d, want %d", c.name, got, c.want)
}
}
valid := iss.Token(nil, "artifactapi", "ci", time.Hour)
headers := []struct {
name string
authz string
want int
}{
{"lowercase scheme", "bearer " + valid, http.StatusNoContent},
{"extra whitespace", " Bearer " + valid + " ", http.StatusNoContent},
{"empty value", "Bearer ", http.StatusUnauthorized},
{"scheme only", "Bearer", http.StatusUnauthorized},
{"basic scheme", "Basic " + valid, http.StatusUnauthorized},
}
for _, c := range headers {
if got := serveAuth(h, http.MethodPost, c.authz); got != c.want {
t.Errorf("%s: status %d, want %d", c.name, got, c.want)
}
}
}
func TestBearerOIDCHangingIssuerBacksOff(t *testing.T) {
var hits atomic.Int32
release := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
hits.Add(1)
select {
case <-r.Context().Done():
case <-release:
}
}))
defer srv.Close()
defer close(release)
b := NewBearerOIDC(srv.URL, nil)
b.timeout = 100 * time.Millisecond
b.backoff = time.Hour
h := b.Require(ok)
start := time.Now()
if got := serve(h, http.MethodPost, "x.y.z"); got != http.StatusServiceUnavailable {
t.Fatalf("hanging issuer: status %d, want 503", got)
}
if d := time.Since(start); d > 2*time.Second {
t.Fatalf("hanging issuer: took %s, timeout not applied", d)
}
if got := serve(h, http.MethodPost, "x.y.z"); got != http.StatusServiceUnavailable {
t.Fatalf("during backoff: status %d, want 503", got)
}
if n := hits.Load(); n != 1 {
t.Fatalf("issuer hit %d times, want 1", n)
}
}
func TestBearerOIDCPanickingDiscoveryBacksOff(t *testing.T) {
b := NewBearerOIDC("http://issuer.invalid", nil)
b.backoff = time.Hour
b.discover = func(context.Context, string) (*oidc.Provider, error) { panic("boom") }
h := b.Require(ok)
func() {
defer func() { _ = recover() }()
serve(h, http.MethodPost, "x.y.z")
}()
got := make(chan int, 1)
go func() { got <- serve(h, http.MethodPost, "x.y.z") }()
select {
case code := <-got:
if code != http.StatusServiceUnavailable {
t.Fatalf("after panic: status %d, want 503", code)
}
case <-time.After(2 * time.Second):
t.Fatal("after panic: request hung on stale inflight discovery")
}
}
func TestBearerOIDCConcurrentFirstRequestsWaitForDiscovery(t *testing.T) {
iss := testsupport.NewOIDCIssuer()
defer iss.Close()
var hits atomic.Int32
inner := iss.Config.Handler
iss.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/.well-known/openid-configuration" {
hits.Add(1)
time.Sleep(200 * time.Millisecond)
}
inner.ServeHTTP(w, r)
})
h := NewBearerOIDC(iss.URL, nil).Require(ok)
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
var wg sync.WaitGroup
codes := make([]int, 20)
for i := range codes {
wg.Add(1)
go func() {
defer wg.Done()
codes[i] = serve(h, http.MethodPost, tok)
}()
}
wg.Wait()
for i, c := range codes {
if c != http.StatusNoContent {
t.Errorf("request %d: status %d, want 204", i, c)
}
}
if n := hits.Load(); n != 1 {
t.Fatalf("discovery hit %d times, want 1", n)
}
}
func TestBearerOIDCRetriesAfterBackoff(t *testing.T) {
iss := testsupport.NewOIDCIssuer()
defer iss.Close()
b := NewBearerOIDC(iss.URL, nil)
b.failedAt = time.Now()
b.backoff = 50 * time.Millisecond
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusServiceUnavailable {
t.Fatalf("within backoff: status %d, want 503", got)
}
time.Sleep(60 * time.Millisecond)
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusNoContent {
t.Fatalf("after backoff: status %d, want 204", got)
}
}
func TestBearerOIDCIssuerDown(t *testing.T) {
iss := testsupport.NewOIDCIssuer()
b := NewBearerOIDC(iss.URL, []string{"artifactapi"})
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
iss.Close()
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusServiceUnavailable {
t.Errorf("issuer down: status %d, want 503", got)
}
}
func TestBearerOIDCDisabled(t *testing.T) {
if got := serve(NewBearerOIDC("", nil).RequireWrites(ok), http.MethodDelete, ""); got != http.StatusNoContent {
t.Errorf("disabled: status %d, want 204", got)
}
}
func TestRequireWritesLeavesReadsOpen(t *testing.T) {
iss := testsupport.NewOIDCIssuer()
defer iss.Close()
h := NewBearerOIDC(iss.URL, []string{"artifactapi"}).RequireWrites(ok)
for _, m := range []string{http.MethodGet, http.MethodHead, http.MethodOptions} {
if got := serve(h, m, ""); got != http.StatusNoContent {
t.Errorf("%s: status %d, want 204", m, got)
}
}
for _, m := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} {
if got := serve(h, m, ""); got != http.StatusUnauthorized {
t.Errorf("%s: status %d, want 401", m, got)
}
}
}
+19
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"os"
"strconv"
"strings"
)
type Config struct {
@@ -56,6 +57,10 @@ type Config struct {
GitHubAppInstallationID string
GitHubAppPrivateKey string
GitHubAppPrivateKeyPath string
// Bearer-JWT auth on configuration writes; empty issuer disables it.
OIDCIssuer string
OIDCAudiences []string
}
func (c *Config) DatabaseDSN() string {
@@ -90,6 +95,17 @@ func Load() (*Config, error) {
return nil, fmt.Errorf("invalid GITHUB_SYNC_POLL_INTERVAL: %w", err)
}
oidcIssuer := getenv("ARTIFACTAPI_OIDC_ISSUER", "")
var oidcAudiences []string
for _, a := range strings.Split(getenv("ARTIFACTAPI_OIDC_AUDIENCE", ""), ",") {
if a = strings.TrimSpace(a); a != "" {
oidcAudiences = append(oidcAudiences, a)
}
}
if oidcIssuer != "" && len(oidcAudiences) == 0 {
return nil, fmt.Errorf("ARTIFACTAPI_OIDC_AUDIENCE is required when ARTIFACTAPI_OIDC_ISSUER is set")
}
cfg := &Config{
ListenAddr: getenv("LISTEN_ADDR", ":8000"),
@@ -123,6 +139,9 @@ func Load() (*Config, error) {
GitHubAppInstallationID: getenv("GITHUB_APP_INSTALLATION_ID", ""),
GitHubAppPrivateKey: getenv("GITHUB_APP_PRIVATE_KEY", ""),
GitHubAppPrivateKeyPath: getenv("GITHUB_APP_PRIVATE_KEY_PATH", ""),
OIDCIssuer: oidcIssuer,
OIDCAudiences: oidcAudiences,
}
return cfg, nil
+17
View File
@@ -64,3 +64,20 @@ func TestLoadInvalidPort(t *testing.T) {
t.Error("expected error for invalid DBPORT")
}
}
func TestLoadOIDC(t *testing.T) {
t.Setenv("ARTIFACTAPI_OIDC_ISSUER", "https://vault.example/v1/identity/oidc")
t.Setenv("ARTIFACTAPI_OIDC_AUDIENCE", " a, ,b ")
cfg, err := Load()
if err != nil {
t.Fatalf("load: %v", err)
}
if len(cfg.OIDCAudiences) != 2 || cfg.OIDCAudiences[0] != "a" || cfg.OIDCAudiences[1] != "b" {
t.Errorf("audiences = %q", cfg.OIDCAudiences)
}
t.Setenv("ARTIFACTAPI_OIDC_AUDIENCE", "")
if _, err := Load(); err == nil {
t.Error("expected error for issuer without audience")
}
}
+8 -38
View File
@@ -2,11 +2,9 @@ package database
import (
"context"
"errors"
"time"
"github.com/jackc/pgx/v5"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -30,42 +28,14 @@ func (db *DB) ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, err
return remotes, rows.Err()
}
// ClaimGitHubAlpineSyncLease atomically claims the per-remote sync lease. It
// succeeds only when the remote is due (never synced, or synced longer than
// freshness ago) and no live lease is held by another replica. A zero freshness
// (prime scans) ignores the recency gate. The returned etag is the stored
// releases-list ETag, shared across replicas.
// ClaimGitHubAlpineSyncLease atomically claims the per-remote github_alpine sync
// lease. See claimSyncLease.
func (db *DB) ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
row := db.Pool.QueryRow(ctx, `
INSERT INTO github_alpine_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
VALUES ($1, $2, now() + make_interval(secs => $4))
ON CONFLICT (remote_name) DO UPDATE
SET sync_lease_owner = $2,
sync_lease_expires = now() + make_interval(secs => $4)
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
RETURNING s.etag
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
var etag string
if err := row.Scan(&etag); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return false, "", nil
}
return false, "", err
}
return true, etag, nil
return db.claimSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, freshness, lease)
}
// ReleaseGitHubAlpineSyncLease records the completed scan and frees the lease.
// Only the owning replica may release; last_synced_at advances so the next poll
// waits a full freshness window, and etag is persisted for the next conditional
// request.
func (db *DB) ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
_, err := db.Pool.Exec(ctx, `
UPDATE github_alpine_sync_state
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
WHERE remote_name = $1 AND sync_lease_owner = $2
`, remoteName, owner, syncedAt, etag)
return err
// ReleaseGitHubAlpineSyncLease records a github_alpine scan outcome and frees the
// lease. See releaseSyncLease.
func (db *DB) ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
return db.releaseSyncLease(ctx, "github_alpine_sync_state", remoteName, owner, res)
}
+8 -37
View File
@@ -2,11 +2,9 @@ package database
import (
"context"
"errors"
"time"
"github.com/jackc/pgx/v5"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -30,41 +28,14 @@ func (db *DB) ListGitHubDebRemotes(ctx context.Context) ([]models.Remote, error)
return remotes, rows.Err()
}
// ClaimGitHubDebSyncLease atomically claims the per-remote sync lease. It
// succeeds only when the remote is due (never synced, or synced longer than
// freshness ago) and no live lease is held by another replica. A zero freshness
// (prime scans) ignores the recency gate. The returned etag is the stored
// releases-list ETag, shared across replicas.
// ClaimGitHubDebSyncLease atomically claims the per-remote github_deb sync
// lease. See claimSyncLease.
func (db *DB) ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
row := db.Pool.QueryRow(ctx, `
INSERT INTO github_deb_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
VALUES ($1, $2, now() + make_interval(secs => $4))
ON CONFLICT (remote_name) DO UPDATE
SET sync_lease_owner = $2,
sync_lease_expires = now() + make_interval(secs => $4)
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
RETURNING s.etag
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
var etag string
if err := row.Scan(&etag); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return false, "", nil
}
return false, "", err
}
return true, etag, nil
return db.claimSyncLease(ctx, "github_deb_sync_state", remoteName, owner, freshness, lease)
}
// ReleaseGitHubDebSyncLease records the completed scan and frees the lease. Only
// the owning replica may release; last_synced_at advances so the next poll waits
// a full freshness window, and etag is persisted for the next conditional request.
func (db *DB) ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
_, err := db.Pool.Exec(ctx, `
UPDATE github_deb_sync_state
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
WHERE remote_name = $1 AND sync_lease_owner = $2
`, remoteName, owner, syncedAt, etag)
return err
// ReleaseGitHubDebSyncLease records a github_deb scan outcome and frees the
// lease. See releaseSyncLease.
func (db *DB) ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
return db.releaseSyncLease(ctx, "github_deb_sync_state", remoteName, owner, res)
}
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -44,7 +45,7 @@ func TestGitHubDebSyncLease(t *testing.T) {
t.Fatal("replica-2 claimed while replica-1 holds the lease")
}
if err := testDB.ReleaseGitHubDebSyncLease(ctx(), name, "replica-1", `"etag-1"`, time.Now()); err != nil {
if err := testDB.ReleaseGitHubDebSyncLease(ctx(), name, "replica-1", provider.SyncResult{Etag: `"etag-1"`}); err != nil {
t.Fatalf("release: %v", err)
}
+45 -17
View File
@@ -7,6 +7,8 @@ import (
"github.com/jackc/pgx/v5"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -30,22 +32,35 @@ func (db *DB) ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
return remotes, rows.Err()
}
// ClaimGitHubSyncLease atomically claims the per-remote sync lease. It succeeds
// (claimed=true) only when the remote is due — never synced, or synced longer
// than freshness ago — and no live lease is held by another replica. This bounds
// total GitHub load to roughly one scan per freshness window regardless of how
// many replicas poll. The returned etag is the stored releases-list ETag, shared
// across replicas so a conditional request can short-circuit an unchanged repo.
// A zero freshness (used for prime scans) ignores the recency gate and claims
// whenever no live lease is held.
// ClaimGitHubSyncLease atomically claims the per-remote github_rpm sync lease.
// See claimSyncLease.
func (db *DB) ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
return db.claimSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, freshness, lease)
}
// ReleaseGitHubSyncLease records a github_rpm scan outcome and frees the lease.
// See releaseSyncLease.
func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error {
return db.releaseSyncLease(ctx, "github_rpm_sync_state", remoteName, owner, res)
}
// claimSyncLease atomically claims a per-remote sync lease in table. It succeeds
// (claimed=true) only when the remote is due and no live lease is held by
// another replica. Due means: a pending retry after a failed scan has reached
// next_retry_at, or, with no retry pending, the remote was never synced or was
// synced longer than freshness ago. A zero freshness (prime scans) ignores the
// recency gate but still honours a pending retry's backoff. The returned etag is
// the stored releases-list ETag, shared across replicas so a conditional request
// can short-circuit an unchanged repo.
func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
row := db.Pool.QueryRow(ctx, `
INSERT INTO github_rpm_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
INSERT INTO `+table+` AS s (remote_name, sync_lease_owner, sync_lease_expires)
VALUES ($1, $2, now() + make_interval(secs => $4))
ON CONFLICT (remote_name) DO UPDATE
SET sync_lease_owner = $2,
sync_lease_expires = now() + make_interval(secs => $4)
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
WHERE (CASE WHEN s.next_retry_at IS NOT NULL THEN s.next_retry_at <= now()
ELSE s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3) END)
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
RETURNING s.etag
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
@@ -60,14 +75,27 @@ func (db *DB) ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string
return true, etag, nil
}
// ReleaseGitHubSyncLease records the completed scan and frees the lease. Only the
// owning replica may release; last_synced_at advances so the next poll waits a
// full freshness window, and etag is persisted for the next conditional request.
func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
// releaseSyncLease records a scan outcome and frees the lease; only the owning
// replica may release. Success advances last_synced_at, persists the etag and
// clears any retry. Failure leaves last_synced_at and etag untouched (the last
// good metadata keeps serving) and schedules next_retry_at with exponential
// backoff from the consecutive-failure count, never before res.RetryAt.
func (db *DB) releaseSyncLease(ctx context.Context, table, remoteName, owner string, res provider.SyncResult) error {
var retryAt *time.Time
if !res.RetryAt.IsZero() {
retryAt = &res.RetryAt
}
_, err := db.Pool.Exec(ctx, `
UPDATE github_rpm_sync_state
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
UPDATE `+table+`
SET last_synced_at = CASE WHEN $3 THEN last_synced_at ELSE now() END,
etag = CASE WHEN $3 THEN etag ELSE $4 END,
sync_failures = CASE WHEN $3 THEN sync_failures + 1 ELSE 0 END,
next_retry_at = CASE WHEN $3 THEN GREATEST(
now() + make_interval(secs => LEAST($5 * power(2, LEAST(sync_failures, 20)), $6)),
$7::timestamptz)
END,
sync_lease_owner = '', sync_lease_expires = NULL
WHERE remote_name = $1 AND sync_lease_owner = $2
`, remoteName, owner, syncedAt, etag)
`, remoteName, owner, res.Failed, res.Etag, res.Backoff.Seconds(), res.MaxBackoff.Seconds(), retryAt)
return err
}
+115 -1
View File
@@ -4,6 +4,7 @@ import (
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
@@ -47,7 +48,7 @@ func TestGitHubSyncLease(t *testing.T) {
}
// Replica 1 finishes: record the sync and persist an etag.
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "replica-1", `"etag-1"`, time.Now()); err != nil {
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "replica-1", provider.SyncResult{Etag: `"etag-1"`}); err != nil {
t.Fatalf("release: %v", err)
}
@@ -93,3 +94,116 @@ func TestListGitHubRPMRemotes(t *testing.T) {
t.Fatalf("seeded remote %q not returned", name)
}
}
// A failed scan must not count as a sync: the etag and last_synced_at stay put,
// every claim (prime included) is held off until the backoff elapses, the delay
// doubles per consecutive failure up to the cap, an upstream retry hint pushes
// it later, and a success clears the retry state.
func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
requireDB(t)
name := "gh-retry-" + time.Now().Format("150405.000000")
seedGitHubRPMRemote(t, name)
const lease = 15 * time.Minute
freshness := time.Hour
claim := func(f time.Duration) (bool, string) {
t.Helper()
ok, etag, err := testDB.ClaimGitHubSyncLease(ctx(), name, "r1", f, lease)
if err != nil {
t.Fatalf("claim: %v", err)
}
return ok, etag
}
release := func(res provider.SyncResult) {
t.Helper()
if err := testDB.ReleaseGitHubSyncLease(ctx(), name, "r1", res); err != nil {
t.Fatalf("release: %v", err)
}
}
state := func() (failures int, retryIn time.Duration, synced *time.Time, etag string) {
t.Helper()
var next *time.Time
var dbNow time.Time
if err := testDB.Pool.QueryRow(ctx(), `SELECT sync_failures, next_retry_at, last_synced_at, etag, now() FROM github_rpm_sync_state WHERE remote_name = $1`, name).
Scan(&failures, &next, &synced, &etag, &dbNow); err != nil {
t.Fatalf("read state: %v", err)
}
if next != nil {
retryIn = next.Sub(dbNow)
}
return
}
elapse := func() {
t.Helper()
if _, err := testDB.Pool.Exec(ctx(), `UPDATE github_rpm_sync_state SET next_retry_at = now() - interval '1 second' WHERE remote_name = $1`, name); err != nil {
t.Fatalf("elapse backoff: %v", err)
}
}
failed := provider.SyncResult{Failed: true, Etag: `"ignored"`, Backoff: time.Minute, MaxBackoff: 3 * time.Minute}
// Last good sync.
if ok, _ := claim(freshness); !ok {
t.Fatal("first claim failed")
}
release(provider.SyncResult{Etag: `"good"`})
_, _, goodSynced, _ := state()
// First failure: 60s backoff, sync time and etag untouched.
if ok, _ := claim(0); !ok {
t.Fatal("prime claim failed")
}
release(failed)
n, in, synced, etag := state()
if n != 1 || in < 55*time.Second || in > 65*time.Second {
t.Fatalf("after 1st failure: failures=%d retry in %v, want 1 and ~60s", n, in)
}
if etag != `"good"` || synced == nil || !synced.Equal(*goodSynced) {
t.Fatalf("failed scan overwrote sync state: etag=%q synced=%v", etag, synced)
}
if ok, _ := claim(0); ok {
t.Fatal("prime claimed inside the retry backoff")
}
// Backoff elapsed: claimable despite being inside the freshness window,
// and the second failure doubles the delay.
elapse()
if ok, _ := claim(freshness); !ok {
t.Fatal("periodic claim refused after backoff elapsed")
}
release(failed)
if n, in, _, _ := state(); n != 2 || in < 115*time.Second || in > 125*time.Second {
t.Fatalf("after 2nd failure: failures=%d retry in %v, want 2 and ~120s", n, in)
}
// Third failure caps at MaxBackoff (3m, not 4m).
elapse()
claim(freshness)
release(failed)
if n, in, _, _ := state(); n != 3 || in < 175*time.Second || in > 185*time.Second {
t.Fatalf("after 3rd failure: failures=%d retry in %v, want 3 and ~180s", n, in)
}
// An upstream rate-limit reset later than the backoff wins.
elapse()
claim(freshness)
hinted := failed
hinted.RetryAt = time.Now().Add(20 * time.Minute)
release(hinted)
if _, in, _, _ := state(); in < 19*time.Minute || in > 21*time.Minute {
t.Fatalf("retry hint ignored: retry in %v, want ~20m", in)
}
// Recovery: success persists the new etag and clears the retry, so the
// normal freshness window applies again.
elapse()
if ok, etag := claim(freshness); !ok || etag != `"good"` {
t.Fatalf("recovery claim ok=%v etag=%q, want last good etag", ok, etag)
}
release(provider.SyncResult{Etag: `"new"`})
if n, in, _, etag := state(); n != 0 || in != 0 || etag != `"new"` {
t.Fatalf("after success: failures=%d retry in %v etag=%q", n, in, etag)
}
if ok, _ := claim(freshness); ok {
t.Fatal("claimed inside freshness window after a successful sync")
}
}
+1 -1
View File
@@ -470,7 +470,7 @@ func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote
return nil, "", false, err
}
if resp.StatusCode != http.StatusOK {
return nil, "", false, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
return nil, "", false, provider.NewUpstreamStatusError("github releases API "+u, resp)
}
if page == 1 {
newEtag = respEtag
+7 -9
View File
@@ -28,7 +28,7 @@ type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error
ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
// SyncConfig tunes the shared syncer. Zero values fall back to safe defaults.
@@ -189,10 +189,11 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
s.mu.Unlock()
}()
freshness := time.Duration(job.remote.MutableTTL) * time.Second
if freshness <= 0 {
freshness = defaultSyncFreshness
ttl := time.Duration(job.remote.MutableTTL) * time.Second
if ttl <= 0 {
ttl = defaultSyncFreshness
}
freshness := ttl
if job.prime {
freshness = 0
}
@@ -210,16 +211,13 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
defer cancel()
newEtag, changed, scanErr := s.prov.scanWithState(scanCtx, job.remote, s.store, etag)
releaseEtag := etag
if scanErr == nil {
releaseEtag = newEtag
} else {
if scanErr != nil {
slog.Error("github_alpine syncer: scan failed", "remote", job.remote.Name, "error", scanErr)
}
relCtx, relCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer relCancel()
if err := s.store.ReleaseGitHubAlpineSyncLease(relCtx, job.remote.Name, s.owner, releaseEtag, time.Now()); err != nil {
if err := s.store.ReleaseGitHubAlpineSyncLease(relCtx, job.remote.Name, s.owner, provider.NewSyncResult(newEtag, scanErr, ttl)); err != nil {
slog.Warn("github_alpine syncer: release lease", "remote", job.remote.Name, "error", err)
}
+13 -3
View File
@@ -27,6 +27,7 @@ type fakeSyncStore struct {
leaseExp map[string]time.Time
lastSynced map[string]time.Time
etags map[string]string
retryAt map[string]time.Time
}
func newFakeSyncStore() *fakeSyncStore {
@@ -36,6 +37,7 @@ func newFakeSyncStore() *fakeSyncStore {
leaseExp: map[string]time.Time{},
lastSynced: map[string]time.Time{},
etags: map[string]string{},
retryAt: map[string]time.Time{},
}
}
@@ -52,6 +54,9 @@ func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owne
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
leaseOK := !hasExp || exp.Before(now)
if freshOK && leaseOK {
f.leaseOwner[name] = owner
@@ -61,14 +66,19 @@ func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owne
return false, "", nil
}
func (f *fakeSyncStore) ReleaseGitHubAlpineSyncLease(_ context.Context, name, owner, etag string, syncedAt time.Time) error {
func (f *fakeSyncStore) ReleaseGitHubAlpineSyncLease(_ context.Context, name, owner string, res provider.SyncResult) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.leaseOwner[name] != owner {
return nil
}
f.lastSynced[name] = syncedAt
f.etags[name] = etag
if res.Failed {
f.retryAt[name] = time.Now().Add(res.Backoff)
} else {
f.lastSynced[name] = time.Now()
f.etags[name] = res.Etag
delete(f.retryAt, name)
}
delete(f.leaseOwner, name)
delete(f.leaseExp, name)
return nil
+1 -1
View File
@@ -431,7 +431,7 @@ func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote
return nil, "", false, err
}
if resp.StatusCode != http.StatusOK {
return nil, "", false, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
return nil, "", false, provider.NewUpstreamStatusError("github releases API "+u, resp)
}
if page == 1 {
newEtag = respEtag
+7 -9
View File
@@ -28,7 +28,7 @@ type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubDebRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubDebSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error
ReleaseGitHubDebSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
// SyncConfig tunes the shared syncer. Zero values fall back to safe defaults.
@@ -189,10 +189,11 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
s.mu.Unlock()
}()
freshness := time.Duration(job.remote.MutableTTL) * time.Second
if freshness <= 0 {
freshness = defaultSyncFreshness
ttl := time.Duration(job.remote.MutableTTL) * time.Second
if ttl <= 0 {
ttl = defaultSyncFreshness
}
freshness := ttl
if job.prime {
freshness = 0
}
@@ -210,16 +211,13 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
defer cancel()
newEtag, changed, scanErr := s.prov.scanWithState(scanCtx, job.remote, s.store, etag)
releaseEtag := etag
if scanErr == nil {
releaseEtag = newEtag
} else {
if scanErr != nil {
slog.Error("github_deb syncer: scan failed", "remote", job.remote.Name, "error", scanErr)
}
relCtx, relCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer relCancel()
if err := s.store.ReleaseGitHubDebSyncLease(relCtx, job.remote.Name, s.owner, releaseEtag, time.Now()); err != nil {
if err := s.store.ReleaseGitHubDebSyncLease(relCtx, job.remote.Name, s.owner, provider.NewSyncResult(newEtag, scanErr, ttl)); err != nil {
slog.Warn("github_deb syncer: release lease", "remote", job.remote.Name, "error", err)
}
+13 -3
View File
@@ -27,6 +27,7 @@ type fakeSyncStore struct {
leaseExp map[string]time.Time
lastSynced map[string]time.Time
etags map[string]string
retryAt map[string]time.Time
}
func newFakeSyncStore() *fakeSyncStore {
@@ -36,6 +37,7 @@ func newFakeSyncStore() *fakeSyncStore {
leaseExp: map[string]time.Time{},
lastSynced: map[string]time.Time{},
etags: map[string]string{},
retryAt: map[string]time.Time{},
}
}
@@ -52,6 +54,9 @@ func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner s
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
leaseOK := !hasExp || exp.Before(now)
if freshOK && leaseOK {
f.leaseOwner[name] = owner
@@ -61,14 +66,19 @@ func (f *fakeSyncStore) ClaimGitHubDebSyncLease(_ context.Context, name, owner s
return false, "", nil
}
func (f *fakeSyncStore) ReleaseGitHubDebSyncLease(_ context.Context, name, owner, etag string, syncedAt time.Time) error {
func (f *fakeSyncStore) ReleaseGitHubDebSyncLease(_ context.Context, name, owner string, res provider.SyncResult) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.leaseOwner[name] != owner {
return nil
}
f.lastSynced[name] = syncedAt
f.etags[name] = etag
if res.Failed {
f.retryAt[name] = time.Now().Add(res.Backoff)
} else {
f.lastSynced[name] = time.Now()
f.etags[name] = res.Etag
delete(f.retryAt, name)
}
delete(f.leaseOwner, name)
delete(f.leaseExp, name)
return nil
+85
View File
@@ -0,0 +1,85 @@
// Package nuget proxies a NuGet v3 feed: the service index, registration pages
// and flat container indexes are mutable, .nupkg/.nuspec downloads immutable.
package nuget
import (
"context"
"encoding/json"
"net/http"
"net/url"
"strings"
"git.unkin.net/unkin/artifactapi/internal/auth"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
func init() {
provider.Register(&Provider{})
}
// serviceIndex is the path clients use as their source URL; it maps to base_url.
const serviceIndex = "index.json"
type Provider struct{}
func (p *Provider) Type() models.PackageType { return models.PackageNuGet }
func (p *Provider) Classify(path string) provider.Mutability {
if isPackage(path) {
return provider.Immutable
}
return provider.Mutable
}
func (p *Provider) ContentType(path string) string {
switch {
case strings.HasSuffix(path, ".nupkg"):
return "application/octet-stream"
case strings.HasSuffix(path, ".nuspec"):
return "application/xml"
}
return "application/json"
}
// UpstreamURL maps index.json to base_url (the upstream service index) and any
// other path to the same path on the base_url's origin, where nuget.org serves
// v3-flatcontainer/ and v3/registration*/.
func (p *Provider) UpstreamURL(remote models.Remote, path string) string {
path = strings.TrimLeft(path, "/")
if path == serviceIndex {
return remote.BaseURL
}
return origin(remote.BaseURL) + "/" + path
}
func (p *Provider) RewriteResponse(_ []byte, _ models.Remote, _ string) ([]byte, error) {
return nil, nil
}
// RewriteServed points upstream-origin URLs in cached JSON back at this remote.
// It runs per request because the URLs must name the requesting host.
func (p *Provider) RewriteServed(body []byte, remote models.Remote, proxyBaseURL string) []byte {
from := origin(remote.BaseURL)
if from == "" || !json.Valid(body) {
return body
}
to := strings.TrimRight(proxyBaseURL, "/") + "/api/v1/remote/" + remote.Name
return []byte(strings.ReplaceAll(string(body), from, to))
}
func (p *Provider) AuthHeaders(_ context.Context, remote models.Remote) (http.Header, error) {
return auth.BasicHeaders(remote), nil
}
func origin(baseURL string) string {
u, err := url.Parse(baseURL)
if err != nil || u.Host == "" {
return ""
}
return u.Scheme + "://" + u.Host
}
func isPackage(path string) bool {
return strings.HasSuffix(path, ".nupkg") || strings.HasSuffix(path, ".nuspec")
}
+68
View File
@@ -0,0 +1,68 @@
package nuget
import (
"testing"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
var _ provider.ServeRewriter = (*Provider)(nil)
var nugetOrg = models.Remote{Name: "nuget-org", BaseURL: "https://api.nuget.org/v3/index.json"}
func TestClassify(t *testing.T) {
p := &Provider{}
tests := []struct {
path string
want provider.Mutability
}{
{"v3-flatcontainer/newtonsoft.json/13.0.3/newtonsoft.json.13.0.3.nupkg", provider.Immutable},
{"v3-flatcontainer/newtonsoft.json/13.0.3/newtonsoft.json.nuspec", provider.Immutable},
{"v3-flatcontainer/newtonsoft.json/index.json", provider.Mutable},
{"v3/registration5-gz-semver2/newtonsoft.json/index.json", provider.Mutable},
{"index.json", provider.Mutable},
}
for _, tt := range tests {
if got := p.Classify(tt.path); got != tt.want {
t.Errorf("Classify(%q) = %v, want %v", tt.path, got, tt.want)
}
}
}
func TestUpstreamURL(t *testing.T) {
p := &Provider{}
tests := []struct {
path, want string
}{
{"index.json", "https://api.nuget.org/v3/index.json"},
{"/index.json", "https://api.nuget.org/v3/index.json"},
{"v3-flatcontainer/serilog/index.json", "https://api.nuget.org/v3-flatcontainer/serilog/index.json"},
{"v3/registration5-gz-semver2/serilog/index.json", "https://api.nuget.org/v3/registration5-gz-semver2/serilog/index.json"},
}
for _, tt := range tests {
if got := p.UpstreamURL(nugetOrg, tt.path); got != tt.want {
t.Errorf("UpstreamURL(%q) = %q, want %q", tt.path, got, tt.want)
}
}
}
func TestRewriteServed(t *testing.T) {
p := &Provider{}
body := `{"resources":[` +
`{"@id":"https://api.nuget.org/v3-flatcontainer/","@type":"PackageBaseAddress/3.0.0"},` +
`{"@id":"https://azuresearch-usnc.nuget.org/query","@type":"SearchQueryService"}]}`
want := `{"resources":[` +
`{"@id":"http://aa.example:8000/api/v1/remote/nuget-org/v3-flatcontainer/","@type":"PackageBaseAddress/3.0.0"},` +
`{"@id":"https://azuresearch-usnc.nuget.org/query","@type":"SearchQueryService"}]}`
if got := string(p.RewriteServed([]byte(body), nugetOrg, "http://aa.example:8000/")); got != want {
t.Errorf("RewriteServed =\n%s\nwant\n%s", got, want)
}
if got := string(p.RewriteServed([]byte(body), models.Remote{Name: "x"}, "http://aa.example")); got != body {
t.Error("remote without a parseable base_url must pass the body through")
}
cert := "\x30\x82https://api.nuget.org"
if got := string(p.RewriteServed([]byte(cert), nugetOrg, "http://aa.example")); got != cert {
t.Error("non-JSON bodies must pass through untouched")
}
}
+7
View File
@@ -75,6 +75,13 @@ type RemoteServer interface {
ServeRemote(w http.ResponseWriter, r *http.Request, remote models.Remote, path, proxyBaseURL string, store RemoteMetadataStore) bool
}
// ServeRewriter rewrites a cached mutable remote response per request, for
// indexes whose URLs must name the requesting host and so cannot be rewritten
// before caching.
type ServeRewriter interface {
RewriteServed(body []byte, remote models.Remote, proxyBaseURL string) []byte
}
// RemoteMetadataStore is the persistence surface a RemoteServer needs to cache
// and read the metadata it derives per upstream asset. *database.DB satisfies it.
type RemoteMetadataStore interface {
+1 -1
View File
@@ -447,7 +447,7 @@ func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote
return nil, "", false, err
}
if resp.StatusCode != http.StatusOK {
return nil, "", false, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
return nil, "", false, provider.NewUpstreamStatusError("github releases API "+u, resp)
}
if page == 1 {
newEtag = respEtag
+9
View File
@@ -78,6 +78,7 @@ type githubFixture struct {
notModHit int // releases-list requests answered 304
releaseAuth string // Authorization header seen on the last releases request
assetAuth string // Authorization header seen on the last asset request
failStatus int // when set, the releases list answers this status (rate-limit style)
mu sync.Mutex
}
@@ -100,6 +101,14 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
f.mu.Lock()
f.releasesHit++
f.releaseAuth = r.Header.Get("Authorization")
if f.failStatus != 0 {
status := f.failStatus
f.mu.Unlock()
w.Header().Set("X-RateLimit-Remaining", "0")
w.Header().Set("X-RateLimit-Reset", strconv.FormatInt(time.Now().Add(30*time.Second).Unix(), 10))
http.Error(w, "API rate limit exceeded", status)
return
}
etag := f.etag
if etag != "" && r.Header.Get("If-None-Match") == etag {
f.notModHit++
+8 -10
View File
@@ -35,7 +35,7 @@ type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubRPMRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error
ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner string, res provider.SyncResult) error
}
// SyncConfig tunes the shared syncer. Zero values fall back to safe defaults.
@@ -205,10 +205,11 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
s.mu.Unlock()
}()
freshness := time.Duration(job.remote.MutableTTL) * time.Second
if freshness <= 0 {
freshness = defaultSyncFreshness
ttl := time.Duration(job.remote.MutableTTL) * time.Second
if ttl <= 0 {
ttl = defaultSyncFreshness
}
freshness := ttl
if job.prime {
freshness = 0 // prime ignores the recency gate but still respects a live lease
}
@@ -226,18 +227,15 @@ func (s *Syncer) process(ctx context.Context, job syncJob) {
defer cancel()
newEtag, changed, scanErr := s.prov.scanWithState(scanCtx, job.remote, s.store, etag)
releaseEtag := etag
if scanErr == nil {
releaseEtag = newEtag
} else {
if scanErr != nil {
slog.Error("github_rpm syncer: scan failed", "remote", job.remote.Name, "error", scanErr)
}
// Release on a detached context so a clean shutdown mid-scan still frees the
// lease and advances last_synced_at (otherwise it simply expires).
// lease and records the outcome (otherwise the lease simply expires).
relCtx, relCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer relCancel()
if err := s.store.ReleaseGitHubSyncLease(relCtx, job.remote.Name, s.owner, releaseEtag, time.Now()); err != nil {
if err := s.store.ReleaseGitHubSyncLease(relCtx, job.remote.Name, s.owner, provider.NewSyncResult(newEtag, scanErr, ttl)); err != nil {
slog.Warn("github_rpm syncer: release lease", "remote", job.remote.Name, "error", err)
}
+92 -3
View File
@@ -27,6 +27,8 @@ type fakeSyncStore struct {
leaseExp map[string]time.Time
lastSynced map[string]time.Time
etags map[string]string
retryAt map[string]time.Time
lastResult provider.SyncResult
}
func newFakeSyncStore() *fakeSyncStore {
@@ -36,6 +38,7 @@ func newFakeSyncStore() *fakeSyncStore {
leaseExp: map[string]time.Time{},
lastSynced: map[string]time.Time{},
etags: map[string]string{},
retryAt: map[string]time.Time{},
}
}
@@ -52,6 +55,9 @@ func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner stri
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
if ra, pending := f.retryAt[name]; pending {
freshOK = !now.Before(ra)
}
leaseOK := !hasExp || exp.Before(now)
if freshOK && leaseOK {
f.leaseOwner[name] = owner
@@ -61,14 +67,20 @@ func (f *fakeSyncStore) ClaimGitHubSyncLease(_ context.Context, name, owner stri
return false, "", nil
}
func (f *fakeSyncStore) ReleaseGitHubSyncLease(_ context.Context, name, owner, etag string, syncedAt time.Time) error {
func (f *fakeSyncStore) ReleaseGitHubSyncLease(_ context.Context, name, owner string, res provider.SyncResult) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.leaseOwner[name] != owner {
return nil
}
f.lastSynced[name] = syncedAt
f.etags[name] = etag
f.lastResult = res
if res.Failed {
f.retryAt[name] = time.Now().Add(res.Backoff)
} else {
f.lastSynced[name] = time.Now()
f.etags[name] = res.Etag
delete(f.retryAt, name)
}
delete(f.leaseOwner, name)
delete(f.leaseExp, name)
return nil
@@ -310,3 +322,80 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
}
}
// A rate-limited scan is not recorded as a sync: the last good repodata keeps
// serving, the retry honours the upstream reset hint and backoff, and once the
// upstream recovers the next poll after the backoff re-syncs without waiting
// out mutable_ttl.
func TestSyncerFailedScanRetriesAfterBackoff(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
store := newFakeSyncStore()
p := newTestProvider()
s := newSyncer(store, p, testSyncConfig())
remote := fx.remote()
bg := context.Background()
s.process(bg, syncJob{remote: remote, prime: true})
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 1 {
t.Fatalf("prime did not derive: %d rows", len(rows))
}
// mutable_ttl elapses while GitHub is rate-limiting.
fx.mu.Lock()
fx.failStatus = http.StatusForbidden
fx.etag = `"v2"`
fx.mu.Unlock()
store.mu.Lock()
store.lastSynced[remote.Name] = time.Now().Add(-2 * time.Hour)
store.mu.Unlock()
s.process(bg, syncJob{remote: remote})
res := store.lastResult
if !res.Failed {
t.Fatal("403 scan was recorded as a successful sync")
}
if until := time.Until(res.RetryAt); until < 20*time.Second || until > 40*time.Second {
t.Fatalf("retry hint from X-RateLimit-Reset not carried: retry in %v", until)
}
if res.Backoff != time.Minute || res.MaxBackoff >= time.Duration(remote.MutableTTL)*time.Second {
t.Fatalf("backoff %v cap %v, want 1m first retry capped below mutable_ttl", res.Backoff, res.MaxBackoff)
}
if store.etags[remote.Name] != `"v1"` {
t.Fatalf("failed scan replaced the etag: %q", store.etags[remote.Name])
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-rpm/repodata/repomd.xml", nil)
p.ServeRemote(rec, req, remote, "repodata/repomd.xml", "https://x", store)
if rec.Code != http.StatusOK {
t.Fatalf("last good repodata not served during failure: %d %s", rec.Code, rec.Body.String())
}
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 1 {
t.Fatalf("failed scan dropped cached metadata: %d rows", len(rows))
}
// Inside the backoff nothing is retried.
hits := fx.releasesHit
s.process(bg, syncJob{remote: remote})
if fx.releasesHit != hits {
t.Fatal("scan retried inside the backoff window")
}
// Upstream recovers and the backoff elapses: the next poll re-syncs.
fx.mu.Lock()
fx.failStatus = 0
fx.rpmBytes["other-9-9.aarch64.rpm"] = testsupport.MinimalRPM("other", "9", "9", "aarch64")
fx.mu.Unlock()
store.mu.Lock()
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
store.mu.Unlock()
s.process(bg, syncJob{remote: remote})
if store.lastResult.Failed || store.etags[remote.Name] != `"v2"` {
t.Fatalf("recovery scan not recorded: failed=%v etag=%q", store.lastResult.Failed, store.etags[remote.Name])
}
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 2 {
t.Fatalf("recovery did not refresh metadata: %d rows", len(rows))
}
}
+75
View File
@@ -0,0 +1,75 @@
package provider
import (
"errors"
"fmt"
"net/http"
"strconv"
"time"
)
const (
syncRetryBase = time.Minute
syncRetryMax = 10 * time.Minute
)
// UpstreamStatusError is a non-success upstream response. RetryAt is the
// upstream's own retry hint (Retry-After, or X-RateLimit-Reset once the quota
// is exhausted); zero when it gave none.
type UpstreamStatusError struct {
URL string
Status int
RetryAt time.Time
}
func (e *UpstreamStatusError) Error() string {
return fmt.Sprintf("%s: status %d", e.URL, e.Status)
}
// NewUpstreamStatusError wraps a non-success response, capturing its retry hint.
func NewUpstreamStatusError(url string, resp *http.Response) *UpstreamStatusError {
e := &UpstreamStatusError{URL: url, Status: resp.StatusCode}
if ra := resp.Header.Get("Retry-After"); ra != "" {
if secs, err := strconv.Atoi(ra); err == nil {
e.RetryAt = time.Now().Add(time.Duration(secs) * time.Second)
} else if t, err := http.ParseTime(ra); err == nil {
e.RetryAt = t
}
} else if resp.Header.Get("X-RateLimit-Remaining") == "0" {
if reset, err := strconv.ParseInt(resp.Header.Get("X-RateLimit-Reset"), 10, 64); err == nil {
e.RetryAt = time.Unix(reset, 0)
}
}
return e
}
// SyncResult is a background scan's outcome, recorded when its sync lease is
// released. A failed scan keeps the prior sync time and ETag and schedules a
// retry after Backoff, doubled per consecutive failure up to MaxBackoff, and
// never earlier than RetryAt.
type SyncResult struct {
Etag string
Failed bool
RetryAt time.Time
Backoff time.Duration
MaxBackoff time.Duration
}
// NewSyncResult builds the result for a scan against a remote with the given
// mutable_ttl. The retry cap stays well below ttl, and an upstream hint is
// clamped to ttl so a bogus reset can never stall the remote longer than a
// normal sync interval would.
func NewSyncResult(etag string, scanErr error, ttl time.Duration) SyncResult {
if scanErr == nil {
return SyncResult{Etag: etag}
}
res := SyncResult{Failed: true, Backoff: syncRetryBase, MaxBackoff: min(syncRetryMax, max(syncRetryBase, ttl/4))}
var se *UpstreamStatusError
if errors.As(scanErr, &se) && !se.RetryAt.IsZero() {
res.RetryAt = se.RetryAt
if limit := time.Now().Add(ttl); res.RetryAt.After(limit) {
res.RetryAt = limit
}
}
return res
}
+63
View File
@@ -0,0 +1,63 @@
package provider
import (
"errors"
"fmt"
"net/http"
"strconv"
"testing"
"time"
)
func TestNewUpstreamStatusErrorRetryHint(t *testing.T) {
reset := time.Now().Add(15 * time.Minute).Truncate(time.Second)
cases := []struct {
name string
header map[string]string
want time.Duration // 0 = no hint
}{
{"retry-after seconds", map[string]string{"Retry-After": "90"}, 90 * time.Second},
{"retry-after date", map[string]string{"Retry-After": reset.UTC().Format(http.TimeFormat)}, 15 * time.Minute},
{"quota exhausted", map[string]string{"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": strconv.FormatInt(reset.Unix(), 10)}, 15 * time.Minute},
{"quota left is not a hint", map[string]string{"X-RateLimit-Remaining": "12", "X-RateLimit-Reset": strconv.FormatInt(reset.Unix(), 10)}, 0},
{"no headers", nil, 0},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
resp := &http.Response{StatusCode: http.StatusForbidden, Header: http.Header{}}
for k, v := range c.header {
resp.Header.Set(k, v)
}
e := NewUpstreamStatusError("u", resp)
if c.want == 0 {
if !e.RetryAt.IsZero() {
t.Fatalf("unexpected hint %v", e.RetryAt)
}
return
}
if d := time.Until(e.RetryAt) - c.want; d < -2*time.Second || d > 2*time.Second {
t.Fatalf("retry in %v, want ~%v", time.Until(e.RetryAt), c.want)
}
})
}
}
func TestNewSyncResult(t *testing.T) {
if r := NewSyncResult(`"e"`, nil, time.Hour); r.Failed || r.Etag != `"e"` {
t.Fatalf("success result = %+v", r)
}
r := NewSyncResult(`"e"`, errors.New("boom"), time.Hour)
if !r.Failed || r.Backoff != time.Minute || r.MaxBackoff != 10*time.Minute || !r.RetryAt.IsZero() {
t.Fatalf("plain failure = %+v, want 1m backoff capped at 10m, no hint", r)
}
if r := NewSyncResult("", errors.New("boom"), 5*time.Minute); r.MaxBackoff != time.Minute+15*time.Second {
t.Fatalf("short ttl cap = %v, want ttl/4", r.MaxBackoff)
}
far := &UpstreamStatusError{Status: 403, RetryAt: time.Now().Add(3 * time.Hour)}
r = NewSyncResult("", fmt.Errorf("scan: %w", far), time.Hour)
if until := time.Until(r.RetryAt); until > time.Hour || until < 59*time.Minute {
t.Fatalf("wrapped hint not clamped to ttl: retry in %v", until)
}
}
+1 -1
View File
@@ -12,7 +12,7 @@ func cors(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusNoContent)
return
+80
View File
@@ -0,0 +1,80 @@
package server
import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
)
func TestServerOIDCWriteAuth(t *testing.T) {
requireStack(t)
iss := testsupport.NewOIDCIssuer()
defer iss.Close()
cfg := *testSrv.cfg
cfg.OIDCIssuer = iss.URL
cfg.OIDCAudiences = []string{"artifactapi"}
s := *testSrv
s.cfg = &cfg
ts := httptest.NewServer(s.routes())
defer ts.Close()
good := iss.Token(nil, "artifactapi", "ci", time.Hour)
do := func(method, path, body, token string) int {
t.Helper()
rq, _ := http.NewRequest(method, ts.URL+path, strings.NewReader(body))
rq.Header.Set("Content-Type", "application/json")
if token != "" {
rq.Header.Set("Authorization", "Bearer "+token)
}
resp, err := http.DefaultClient.Do(rq)
if err != nil {
t.Fatalf("%s %s: %v", method, path, err)
}
_, _ = io.Copy(io.Discard, resp.Body)
_ = resp.Body.Close()
return resp.StatusCode
}
expect := func(want int, method, path, body, token string) {
t.Helper()
if got := do(method, path, body, token); got != want {
t.Errorf("%s %s: status %d, want %d", method, path, got, want)
}
}
remote := `{"name":"srv-oidc","package_type":"generic","repo_type":"local"}`
virtual := `{"name":"srv-oidc-v","package_type":"generic","members":["srv-oidc"]}`
expect(401, "POST", "/api/v2/remotes", remote, "")
expect(401, "POST", "/api/v2/remotes", remote, iss.Token(nil, "other", "ci", time.Hour))
expect(201, "POST", "/api/v2/remotes", remote, good)
expect(401, "PUT", "/api/v2/remotes/srv-oidc", `{"package_type":"generic"}`, "")
expect(401, "DELETE", "/api/v2/remotes/srv-oidc", "", "")
expect(401, "POST", "/api/v2/virtuals", virtual, "")
expect(201, "POST", "/api/v2/virtuals", virtual, good)
expect(401, "PUT", "/api/v2/virtuals/srv-oidc-v", virtual, "")
expect(401, "DELETE", "/api/v2/virtuals/srv-oidc-v", "", "")
expect(200, "GET", "/api/v2/remotes", "", "")
expect(200, "GET", "/api/v2/remotes/srv-oidc", "", "")
expect(200, "GET", "/api/v2/virtuals/srv-oidc-v", "", "")
expect(201, "PUT", "/api/v2/remotes/srv-oidc/files/a.bin", "payload", "")
expect(200, "GET", "/api/v2/remotes/srv-oidc/files/a.bin", "", "")
if got := do("DELETE", "/api/v2/remotes/srv-oidc/files/a.bin", "", ""); got >= 400 {
t.Errorf("delete file: status %d", got)
}
if got := do("POST", "/api/v2/probe", `{}`, ""); got == 401 {
t.Error("probe gated")
}
if got := do("GET", "/v2/", "", ""); got == 401 {
t.Error("docker /v2 gated")
}
expect(204, "DELETE", "/api/v2/virtuals/srv-oidc-v", "", good)
expect(204, "DELETE", "/api/v2/remotes/srv-oidc", "", good)
}
+5 -2
View File
@@ -15,6 +15,7 @@ import (
tfregistry "git.unkin.net/unkin/artifactapi/internal/api/terraform"
v1 "git.unkin.net/unkin/artifactapi/internal/api/v1"
v2 "git.unkin.net/unkin/artifactapi/internal/api/v2"
"git.unkin.net/unkin/artifactapi/internal/auth"
"git.unkin.net/unkin/artifactapi/internal/cache"
"git.unkin.net/unkin/artifactapi/internal/config"
"git.unkin.net/unkin/artifactapi/internal/database"
@@ -28,6 +29,7 @@ import (
_ "git.unkin.net/unkin/artifactapi/internal/provider/goproxy"
_ "git.unkin.net/unkin/artifactapi/internal/provider/helm"
_ "git.unkin.net/unkin/artifactapi/internal/provider/npm"
_ "git.unkin.net/unkin/artifactapi/internal/provider/nuget"
_ "git.unkin.net/unkin/artifactapi/internal/provider/puppet"
_ "git.unkin.net/unkin/artifactapi/internal/provider/pypi"
"git.unkin.net/unkin/artifactapi/internal/provider/rpm"
@@ -186,10 +188,11 @@ func (s *Server) routes() chi.Router {
statsHandler := v2.NewStatsHandler(s.db)
eventsHandler := v2.NewEventsHandler()
probeHandler := v2.NewProbeHandler(s.engine, s.db)
writeAuth := auth.NewBearerOIDC(s.cfg.OIDCIssuer, s.cfg.OIDCAudiences)
r.Route("/api/v2", func(r chi.Router) {
r.Mount("/remotes", remotesHandler.Routes())
r.Mount("/virtuals", virtualsHandler.Routes())
r.Mount("/remotes", writeAuth.RequireWrites(remotesHandler.Routes()))
r.Mount("/virtuals", writeAuth.RequireWrites(virtualsHandler.Routes()))
r.Mount("/health", healthHandler.Routes())
r.Mount("/stats", statsHandler.Routes())
r.Mount("/events", eventsHandler.Routes())
+73
View File
@@ -0,0 +1,73 @@
package testsupport
import (
"crypto/rand"
"crypto/rsa"
"encoding/json"
"net/http"
"net/http/httptest"
"time"
"github.com/go-jose/go-jose/v4"
"github.com/go-jose/go-jose/v4/jwt"
)
// OIDCIssuer is a minimal OIDC issuer serving discovery and JWKS, signing
// RS256 tokens with a generated key.
type OIDCIssuer struct {
*httptest.Server
key *rsa.PrivateKey
}
func NewOIDCIssuer() *OIDCIssuer {
key, err := rsa.GenerateKey(rand.Reader, 2048)
if err != nil {
panic(err)
}
iss := &OIDCIssuer{key: key}
mux := http.NewServeMux()
mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(map[string]any{
"issuer": iss.URL,
"jwks_uri": iss.URL + "/keys",
"id_token_signing_alg_values_supported": []string{"RS256"},
})
})
mux.HandleFunc("/keys", func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(jose.JSONWebKeySet{Keys: []jose.JSONWebKey{
{Key: &key.PublicKey, KeyID: "k1", Algorithm: "RS256", Use: "sig"},
}})
})
iss.Server = httptest.NewServer(mux)
return iss
}
// Token signs a token for aud and sub. Pass a nil key to use the issuer's own;
// expiry is relative to now.
func (o *OIDCIssuer) Token(key *rsa.PrivateKey, aud, sub string, expiry time.Duration) string {
now := time.Now()
return o.Sign(key, jwt.Claims{
Issuer: o.URL,
Subject: sub,
Audience: jwt.Audience{aud},
IssuedAt: jwt.NewNumericDate(now.Add(-time.Minute)),
Expiry: jwt.NewNumericDate(now.Add(expiry)),
})
}
// Sign signs arbitrary claims with RS256. Pass a nil key to use the issuer's own.
func (o *OIDCIssuer) Sign(key *rsa.PrivateKey, claims jwt.Claims) string {
if key == nil {
key = o.key
}
sig, err := jose.NewSigner(jose.SigningKey{Algorithm: jose.RS256, Key: key},
(&jose.SignerOptions{}).WithType("JWT").WithHeader("kid", "k1"))
if err != nil {
panic(err)
}
tok, err := jwt.Signed(sig).Claims(claims).Serialize()
if err != nil {
panic(err)
}
return tok
}
+245
View File
@@ -0,0 +1,245 @@
package virtual
import (
"bytes"
"crypto/md5"
"encoding/hex"
"fmt"
"path"
"regexp"
"slices"
"strings"
"time"
)
// debIndexRe matches the binary Packages indexes a virtual merges: the flat
// repo's root Packages and a suite's <component>/binary-<arch>/Packages.
var debIndexRe = regexp.MustCompile(`^([^/]+/binary-[^/]+/)?Packages(\.[a-z0-9]+)?$`)
// debByHashRe matches an index requested by its SHA256 (Acquire-By-Hash).
var debByHashRe = regexp.MustCompile(`^([^/]+/binary-[^/]+/)?by-hash/SHA256/[0-9a-f]{64}$`)
// debCompressions is the order a member's index variants are fetched in.
var debCompressions = []string{".xz", ".gz", ".zst", ".bz2", ""}
// DebMember is one member's Release plus its decompressed Packages indexes,
// keyed by index directory relative to the Release ("" for a flat repo).
type DebMember struct {
RemoteName string
Release []byte
Packages map[string][]byte
}
// DebRepo is a merged apt repo: Release plus the indexes it lists, keyed by
// their path relative to the Release.
type DebRepo struct {
Release []byte
Files map[string][]byte
}
type debRelease struct {
fields map[string]string
files []string
}
// parseDebRelease reads a Release file's single-line fields and the names
// listed in its checksum sections.
func parseDebRelease(body []byte) debRelease {
rel := debRelease{fields: map[string]string{}}
inSums := false
seen := map[string]bool{}
for line := range strings.SplitSeq(string(body), "\n") {
line = strings.TrimRight(line, "\r")
if line == "" {
continue
}
if line[0] == ' ' || line[0] == '\t' {
if f := strings.Fields(line); inSums && len(f) == 3 && !seen[f[2]] {
seen[f[2]] = true
rel.files = append(rel.files, f[2])
}
continue
}
k, v, _ := strings.Cut(line, ":")
v = strings.TrimSpace(v)
inSums = v == "" && (k == "MD5Sum" || strings.HasPrefix(k, "SHA"))
if _, ok := rel.fields[k]; !ok {
rel.fields[k] = v
}
}
return rel
}
// debIndexFiles returns, per index directory, the variant of its Packages file
// to fetch from a member's Release listing.
func debIndexFiles(rel debRelease) map[string]string {
listed := map[string]bool{}
for _, f := range rel.files {
listed[f] = true
}
out := map[string]string{}
for _, f := range rel.files {
m := debIndexRe.FindStringSubmatch(f)
if m == nil || out[m[1]] != "" {
continue
}
for _, c := range debCompressions {
if listed[m[1]+"Packages"+c] {
out[m[1]] = m[1] + "Packages" + c
break
}
}
}
return out
}
func splitStanzas(body []byte) []string {
var out []string
for s := range strings.SplitSeq(strings.ReplaceAll(string(body), "\r\n", "\n"), "\n\n") {
if s = strings.Trim(s, "\n"); s != "" {
out = append(out, s)
}
}
return out
}
func stanzaField(stanza, key string) string {
for line := range strings.SplitSeq(stanza, "\n") {
if k, v, ok := strings.Cut(line, ":"); ok && k == key {
return strings.TrimSpace(v)
}
}
return ""
}
func unionFields(dst []string, field string) []string {
for f := range strings.FieldsSeq(field) {
if !slices.Contains(dst, f) {
dst = append(dst, f)
}
}
return dst
}
func parseDebDate(s string) (time.Time, bool) {
for _, layout := range []string{time.RFC1123Z, time.RFC1123} {
if t, err := time.Parse(layout, s); err == nil {
return t, true
}
}
return time.Time{}, false
}
// MergeDeb merges members into one apt repo. Members are in priority order: the
// first member to carry a (Package, Version, Architecture) in an index wins;
// stanzas missing any of those are always kept. Filename is prefixed with the
// owning member's name so the virtual can route downloads back to it.
func MergeDeb(members []DebMember) *DebRepo {
var suite, codename string
var arches, comps, dirs []string
date := time.Unix(0, 0)
for _, m := range members {
rel := parseDebRelease(m.Release)
if suite == "" {
suite, codename = rel.fields["Suite"], rel.fields["Codename"]
}
arches = unionFields(arches, rel.fields["Architectures"])
comps = unionFields(comps, rel.fields["Components"])
if t, ok := parseDebDate(rel.fields["Date"]); ok && t.After(date) {
date = t
}
for d := range m.Packages {
if !slices.Contains(dirs, d) {
dirs = append(dirs, d)
}
}
}
slices.Sort(dirs)
repo := &DebRepo{Files: map[string][]byte{}}
for _, d := range dirs {
seen := map[string]bool{}
var b bytes.Buffer
for _, m := range members {
for _, s := range splitStanzas(m.Packages[d]) {
pkg, ver, arch := stanzaField(s, "Package"), stanzaField(s, "Version"), stanzaField(s, "Architecture")
if pkg != "" && ver != "" && arch != "" {
key := pkg + "\x00" + ver + "\x00" + arch
if seen[key] {
continue
}
seen[key] = true
}
if fn := stanzaField(s, "Filename"); fn != "" {
s = rewriteFilename(s, m.RemoteName+"/"+strings.TrimPrefix(path.Clean("/"+fn), "/"))
}
b.WriteString(s)
b.WriteString("\n\n")
}
}
gz := gzipDeterministic(b.Bytes())
repo.Files[d+"Packages"] = b.Bytes()
repo.Files[d+"Packages.gz"] = gz
repo.Files[d+"by-hash/SHA256/"+sha256Hex(b.Bytes())] = b.Bytes()
repo.Files[d+"by-hash/SHA256/"+sha256Hex(gz)] = gz
}
var r bytes.Buffer
if suite != "" {
fmt.Fprintf(&r, "Suite: %s\n", suite)
}
if codename != "" {
fmt.Fprintf(&r, "Codename: %s\n", codename)
}
fmt.Fprintf(&r, "Date: %s\n", date.UTC().Format(time.RFC1123Z))
fmt.Fprintf(&r, "Architectures: %s\n", strings.Join(arches, " "))
if len(comps) > 0 {
fmt.Fprintf(&r, "Components: %s\n", strings.Join(comps, " "))
}
r.WriteString("Acquire-By-Hash: yes\n")
for _, sec := range []struct {
name string
sum func([]byte) string
}{{"MD5Sum", md5Hex}, {"SHA256", sha256Hex}} {
r.WriteString(sec.name + ":\n")
for _, d := range dirs {
for _, n := range []string{d + "Packages", d + "Packages.gz"} {
fmt.Fprintf(&r, " %s %d %s\n", sec.sum(repo.Files[n]), len(repo.Files[n]), n)
}
}
}
repo.Release = r.Bytes()
return repo
}
func rewriteFilename(stanza, fn string) string {
lines := strings.Split(stanza, "\n")
for i, l := range lines {
if strings.HasPrefix(l, "Filename:") {
lines[i] = "Filename: " + fn
break
}
}
return strings.Join(lines, "\n")
}
// debIndexPath splits an index request into its Release directory and the file
// under it. ok is false for anything a deb virtual does not merge.
func debIndexPath(p string) (base, name string, ok bool) {
if rest, found := strings.CutPrefix(p, "dists/"); found {
suite, n, found := strings.Cut(rest, "/")
if !found || suite == "" {
return "", "", false
}
base, p = "dists/"+suite+"/", n
}
if p != "Release" && !debIndexRe.MatchString(p) && !debByHashRe.MatchString(p) {
return "", "", false
}
return base, p, true
}
func md5Hex(data []byte) string {
h := md5.Sum(data)
return hex.EncodeToString(h[:])
}
+179
View File
@@ -0,0 +1,179 @@
package virtual
import (
"bytes"
"crypto/md5"
"encoding/hex"
"fmt"
"maps"
"path"
"slices"
"strings"
"testing"
)
func debStanza(pkg, ver, arch, filename string) string {
return fmt.Sprintf("Package: %s\nVersion: %s\nArchitecture: %s\nDescription: test\n multi-line body\nFilename: %s\nSize: 1\n", pkg, ver, arch, filename)
}
func flatDebMember(name, date string, stanzas ...string) DebMember {
rel := "Date: " + date + "\nArchitectures: amd64\nAcquire-By-Hash: no\nSHA256:\n x 1 Packages\n x 1 Packages.gz\n"
return DebMember{RemoteName: name, Release: []byte(rel), Packages: map[string][]byte{"": []byte(strings.Join(stanzas, "\n"))}}
}
func TestMergeDebDedupesByMemberOrder(t *testing.T) {
repo := MergeDeb([]DebMember{
flatDebMember("a", "Mon, 05 Oct 2026 10:00:00 +0000",
debStanza("foo", "1.0", "amd64", "pool/foo_1.0_amd64.deb"),
debStanza("bar", "2.0", "amd64", "/pool/bar.deb"),
debStanza("dot", "1.0", "amd64", "./dot.deb"),
"Package: noversion\nFilename: nv1.deb\n",
"Package: noversion\nFilename: nv2.deb\n"),
flatDebMember("b", "Tue, 06 Oct 2026 10:00:00 UTC",
debStanza("foo", "1.0", "amd64", "pool/foo-from-b.deb"),
debStanza("foo", "1.0", "arm64", "pool/foo_arm64.deb"),
debStanza("foo", "1.1", "amd64", "o/r/releases/download/v1/foo.deb")),
})
pk := string(repo.Files["Packages"])
stanzas := splitStanzas(repo.Files["Packages"])
if len(stanzas) != 7 {
t.Fatalf("want 7 stanzas (one duplicate dropped), got %d:\n%s", len(stanzas), pk)
}
for _, want := range []string{
"Filename: a/pool/foo_1.0_amd64.deb",
"Filename: a/pool/bar.deb",
"Filename: a/dot.deb",
"Filename: a/nv1.deb",
"Filename: a/nv2.deb",
"Filename: b/pool/foo_arm64.deb",
"Filename: b/o/r/releases/download/v1/foo.deb",
" multi-line body",
} {
if !strings.Contains(pk, want) {
t.Errorf("merged Packages missing %q:\n%s", want, pk)
}
}
if strings.Contains(pk, "foo-from-b") {
t.Fatalf("second member's duplicate must lose:\n%s", pk)
}
if !bytes.Equal(gunzip(t, repo.Files["Packages.gz"]), repo.Files["Packages"]) {
t.Fatal("Packages.gz does not match Packages")
}
if !strings.Contains(string(repo.Release), "Date: Tue, 06 Oct 2026 10:00:00 +0000\n") {
t.Fatalf("Date should be the newest member date:\n%s", repo.Release)
}
}
func TestMergeDebReleaseHashesFiles(t *testing.T) {
repo := MergeDeb([]DebMember{flatDebMember("a", "", debStanza("foo", "1", "amd64", "pool/foo.deb"))})
rel := parseDebRelease(repo.Release)
if got := slices.Sorted(slices.Values(rel.files)); !slices.Equal(got, []string{"Packages", "Packages.gz"}) {
t.Fatalf("Release lists %v", got)
}
inSHA := false
checked := 0
for line := range strings.SplitSeq(string(repo.Release), "\n") {
if !strings.HasPrefix(line, " ") {
inSHA = line == "SHA256:"
continue
}
var sum, name string
var size int
if _, err := fmt.Sscan(line, &sum, &size, &name); err != nil {
t.Fatal(err)
}
body := repo.Files[name]
want := sha256Hex(body)
if !inSHA {
h := md5.Sum(body)
want = hex.EncodeToString(h[:])
}
if sum != want || size != len(body) {
t.Errorf("%s: Release has %s %d, file is %s %d", name, sum, size, want, len(body))
}
checked++
}
if checked != 4 {
t.Fatalf("checked %d Release entries, want 4:\n%s", checked, repo.Release)
}
if !bytes.Equal(MergeDeb([]DebMember{flatDebMember("a", "", debStanza("foo", "1", "amd64", "pool/foo.deb"))}).Release, repo.Release) {
t.Fatal("merge must be deterministic")
}
}
func TestMergeDebSuiteLayout(t *testing.T) {
rel := func(comps string) []byte {
return []byte("Suite: stable\nCodename: trixie\nArchitectures: amd64 arm64\nComponents: " + comps + "\nSHA256:\n x 1 main/binary-amd64/Packages.xz\n")
}
repo := MergeDeb([]DebMember{
{RemoteName: "deb1", Release: rel("main"), Packages: map[string][]byte{"main/binary-amd64/": []byte(debStanza("foo", "1", "amd64", "pool/main/f/foo.deb"))}},
{RemoteName: "deb2", Release: rel("main contrib"), Packages: map[string][]byte{
"main/binary-amd64/": []byte(debStanza("bar", "1", "amd64", "pool/main/b/bar.deb")),
"contrib/binary-amd64/": []byte(debStanza("baz", "1", "amd64", "pool/contrib/b/baz.deb")),
}},
})
for _, want := range []string{"Suite: stable\n", "Codename: trixie\n", "Components: main contrib\n", "Architectures: amd64 arm64\n"} {
if !strings.Contains(string(repo.Release), want) {
t.Errorf("Release missing %q:\n%s", want, repo.Release)
}
}
want := []string{"contrib/binary-amd64/Packages", "contrib/binary-amd64/Packages.gz", "main/binary-amd64/Packages", "main/binary-amd64/Packages.gz"}
if got := slices.Sorted(slices.Values(parseDebRelease(repo.Release).files)); !slices.Equal(got, want) {
t.Fatalf("Release lists %v, want %v", got, want)
}
if !strings.Contains(string(repo.Release), "Acquire-By-Hash: yes\n") {
t.Fatalf("Release must advertise by-hash:\n%s", repo.Release)
}
for _, n := range want {
byHash := path.Dir(n) + "/by-hash/SHA256/" + sha256Hex(repo.Files[n])
if !bytes.Equal(repo.Files[byHash], repo.Files[n]) {
t.Errorf("%s not served at %s", n, byHash)
}
}
if len(repo.Files) != 2*len(want) {
t.Fatalf("files %v, want each index plus its by-hash copy", slices.Sorted(maps.Keys(repo.Files)))
}
if n := len(splitStanzas(repo.Files["main/binary-amd64/Packages"])); n != 2 {
t.Fatalf("main/binary-amd64 should merge both members, got %d stanzas", n)
}
}
func TestDebIndexFiles(t *testing.T) {
rel := parseDebRelease([]byte("Origin: Debian\nMD5Sum:\n a 1 main/binary-amd64/Packages\n a 1 main/binary-amd64/Packages.gz\nSHA256:\n" +
" a 1 main/binary-amd64/Packages\n a 1 main/binary-amd64/Packages.xz\n a 1 main/binary-amd64/Packages.gz\n" +
" a 1 main/debian-installer/binary-amd64/Packages.xz\n a 1 main/i18n/Translation-en.xz\n a 1 contrib/binary-all/Packages\n"))
got := debIndexFiles(rel)
want := map[string]string{"main/binary-amd64/": "main/binary-amd64/Packages.xz", "contrib/binary-all/": "contrib/binary-all/Packages"}
if !maps.Equal(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
if flat := debIndexFiles(parseDebRelease([]byte("SHA256:\n a 1 Packages\n a 1 Packages.gz\n"))); flat[""] != "Packages.gz" {
t.Fatalf("flat repo should fetch Packages.gz, got %v", flat)
}
}
func TestDebIndexPath(t *testing.T) {
for _, tc := range []struct {
in, base, name string
ok bool
}{
{"Release", "", "Release", true},
{"Packages.gz", "", "Packages.gz", true},
{"dists/trixie/Release", "dists/trixie/", "Release", true},
{"dists/trixie/main/binary-amd64/Packages", "dists/trixie/", "main/binary-amd64/Packages", true},
{"by-hash/SHA256/abababababababababababababababababababababababababababababababab", "", "by-hash/SHA256/abababababababababababababababababababababababababababababababab", true},
{"dists/trixie/main/binary-amd64/by-hash/SHA256/abababababababababababababababababababababababababababababababab", "dists/trixie/", "main/binary-amd64/by-hash/SHA256/abababababababababababababababababababababababababababababababab", true},
{"dists/trixie/main/binary-amd64/by-hash/MD5Sum/abababababababababababababababab", "", "", false},
{"InRelease", "", "", false},
{"Release.gpg", "", "", false},
{"dists/trixie/InRelease", "", "", false},
{"dists/trixie/main/i18n/Translation-en", "", "", false},
{"dists/Release", "", "", false},
{"pool/foo.deb", "", "", false},
} {
base, name, ok := debIndexPath(tc.in)
if base != tc.base || name != tc.name || ok != tc.ok {
t.Errorf("%s: got (%q, %q, %v), want (%q, %q, %v)", tc.in, base, name, ok, tc.base, tc.name, tc.ok)
}
}
}
+166 -43
View File
@@ -9,6 +9,7 @@ import (
"net/http"
"net/http/httptest"
"net/url"
"path"
"slices"
"strings"
"sync"
@@ -26,32 +27,51 @@ type Engine struct {
proxyEngine *proxy.Engine
getRemote func(context.Context, string) (*models.Remote, error)
rpmMember func(context.Context, string) (*RPMMember, error)
debMember func(ctx context.Context, name, base string) (*DebMember, error)
mergeTTL time.Duration
sf singleflight.Group
mu sync.Mutex
rpm map[string]*rpmGen
sf singleflight.Group
mu sync.Mutex
gens map[string]*repoGen
}
// rpmGen holds a virtual's current merge plus the previous one, so a client
// holding the previous repomd can still fetch its content-hashed data files.
type rpmGen struct {
cur, prev *RPMRepo
at time.Time
// mergedRepo is a merged index document plus the files it references.
type mergedRepo interface {
index() []byte
files() map[string][]byte
}
func (r *RPMRepo) index() []byte { return r.Repomd }
func (r *RPMRepo) files() map[string][]byte { return r.Files }
func (r *DebRepo) index() []byte { return r.Release }
func (r *DebRepo) files() map[string][]byte { return r.Files }
// repoGen holds a virtual's current merge plus the previous one, so a client
// holding the previous index can still fetch the files it references.
type repoGen struct {
cur, prev mergedRepo
at, used time.Time
}
const rpmMergeTTL = 60 * time.Second
// maxDebSuites bounds the client-chosen suites cached per deb virtual.
const maxDebSuites = 32
func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL}
e.rpmMember = e.fetchRPMMember
e.debMember = e.fetchDebMember
return e
}
func (e *Engine) Fetch(ctx context.Context, virt models.Virtual, path string, proxyBaseURL string) ([]byte, string, error) {
if virt.PackageType == models.PackageRPM {
switch virt.PackageType {
case models.PackageRPM:
return e.fetchRPM(ctx, virt, path)
case models.PackageDeb:
return e.fetchDeb(ctx, virt, path)
}
merger, err := GetMerger(virt.PackageType)
@@ -170,11 +190,22 @@ var (
ErrBadPath = errors.New("invalid path")
)
// MemberRedirect maps an rpm virtual package path (<member>/<href>, as written
// by MergeRPM) to the owning member's route. ok is false for any other path;
// ErrBadPath means the path is absolute or escapes the member.
// MemberRedirect maps an rpm or deb virtual package path (<member>/<href>, as
// written by MergeRPM/MergeDeb) to the owning member's route. ok is false for
// any other path; ErrBadPath means the path is absolute or escapes the member.
func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path, rawQuery, proxyBaseURL string) (string, bool, error) {
if virt.PackageType != models.PackageRPM || strings.HasPrefix(path, "repodata/") {
switch virt.PackageType {
case models.PackageRPM:
if strings.HasPrefix(path, "repodata/") {
return "", false, nil
}
case models.PackageDeb:
// apt requests a flat repo's files under its "./" dist verbatim.
path = strings.TrimPrefix(path, "./")
if strings.HasPrefix(path, "dists/") {
return "", false, nil
}
default:
return "", false, nil
}
name, rest, found := strings.Cut(path, "/")
@@ -217,76 +248,93 @@ func (e *Engine) fetchRPM(ctx context.Context, virt models.Virtual, path string)
if !strings.HasPrefix(path, "repodata/") {
return nil, "", ErrNotFound
}
key := "rpm\x00" + virt.Name
if path != "repodata/repomd.xml" {
if body, ok := e.cachedRPMFile(virt.Name, path); ok {
if body, ok := e.cachedFile(key, path); ok {
return body, "application/gzip", nil
}
}
repo, err := e.mergedRPM(ctx, virt)
repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
return e.mergeRPM(ctx, virt)
})
if err != nil {
return nil, "", err
}
if path == "repodata/repomd.xml" {
return repo.Repomd, "application/xml", nil
return repo.index(), "application/xml", nil
}
if body, ok := repo.Files[path]; ok {
if body, ok := repo.files()[path]; ok {
return body, "application/gzip", nil
}
return nil, "", ErrNotFound
}
// mergedRPM returns the virtual's merge, reusing it for mergeTTL and
// collapsing concurrent merges of the same virtual into one.
// merged returns the merge cached under key, reusing it for mergeTTL and
// collapsing concurrent merges of the same key into one.
// ponytail: per-replica cache; behind a non-sticky LB a data request landing on
// another replica re-merges and 404s if a member changed in between. Move to the
// shared redis cache if that shows up.
func (e *Engine) mergedRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
func (e *Engine) merged(ctx context.Context, key string, merge func(context.Context) (mergedRepo, error)) (mergedRepo, error) {
e.mu.Lock()
g := e.rpm[virt.Name]
g := e.gens[key]
if g != nil && time.Since(g.at) < e.mergeTTL {
g.used = time.Now()
e.mu.Unlock()
return g.cur, nil
}
e.mu.Unlock()
v, err, _ := e.sf.Do(virt.Name, func() (any, error) {
repo, err := e.mergeRPM(context.WithoutCancel(ctx), virt)
v, err, _ := e.sf.Do(key, func() (any, error) {
repo, err := merge(context.WithoutCancel(ctx))
if err != nil {
return nil, err
}
e.mu.Lock()
defer e.mu.Unlock()
if e.rpm == nil {
e.rpm = map[string]*rpmGen{}
if e.gens == nil {
e.gens = map[string]*repoGen{}
}
g := e.rpm[virt.Name]
g := e.gens[key]
switch {
case g == nil:
e.rpm[virt.Name] = &rpmGen{cur: repo, at: time.Now()}
case string(g.cur.Repomd) == string(repo.Repomd):
g.at = time.Now()
e.gens[key] = &repoGen{cur: repo, at: time.Now(), used: time.Now()}
case string(g.cur.index()) == string(repo.index()):
g.at, g.used = time.Now(), time.Now()
repo = g.cur
default:
g.prev, g.cur, g.at = g.cur, repo, time.Now()
g.prev, g.cur, g.at, g.used = g.cur, repo, time.Now(), time.Now()
}
return repo, nil
})
if err != nil {
return nil, err
}
return v.(*RPMRepo), nil
return v.(mergedRepo), nil
}
func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
members := make([]RPMMember, len(virt.Members))
members, err := fetchMembers(virt, func(name string) (*RPMMember, error) { return e.rpmMember(ctx, name) })
if err != nil {
return nil, err
}
repo, err := MergeRPM(members)
if err != nil {
return nil, fmt.Errorf("merge rpm repodata: %w", err)
}
return repo, nil
}
// fetchMembers reads every member in parallel, failing if any member fails.
func fetchMembers[T any](virt models.Virtual, fetch func(name string) (*T, error)) ([]T, error) {
members := make([]T, len(virt.Members))
errs := make([]error, len(virt.Members))
var wg sync.WaitGroup
for i, name := range virt.Members {
wg.Add(1)
go func() {
defer wg.Done()
m, err := e.rpmMember(ctx, name)
m, err := fetch(name)
if err != nil {
errs[i] = fmt.Errorf("member %q: %w", name, err)
return
@@ -299,26 +347,101 @@ func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, e
if err := errors.Join(errs...); err != nil {
return nil, fmt.Errorf("virtual %q: %v", virt.Name, err)
}
repo, err := MergeRPM(members)
if err != nil {
return nil, fmt.Errorf("merge rpm repodata: %w", err)
}
return repo, nil
return members, nil
}
func (e *Engine) cachedRPMFile(virt, path string) ([]byte, bool) {
func (e *Engine) fetchDeb(ctx context.Context, virt models.Virtual, reqPath string) ([]byte, string, error) {
base, name, ok := debIndexPath(strings.TrimPrefix(path.Clean("/"+reqPath), "/"))
if !ok {
return nil, "", ErrNotFound
}
key := "deb\x00" + virt.Name + "\x00" + base
if name != "Release" {
if body, ok := e.cachedFile(key, name); ok {
return body, debContentType(name), nil
}
}
repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
members, err := fetchMembers(virt, func(name string) (*DebMember, error) { return e.debMember(ctx, name, base) })
if err != nil {
return nil, err
}
return MergeDeb(members), nil
})
if err != nil {
return nil, "", err
}
e.evictLRU("deb\x00"+virt.Name+"\x00", maxDebSuites)
if name == "Release" {
return repo.index(), "text/plain", nil
}
if body, ok := repo.files()[name]; ok {
return body, debContentType(name), nil
}
return nil, "", ErrNotFound
}
func debContentType(name string) string {
if strings.HasSuffix(name, ".gz") {
return "application/gzip"
}
return "text/plain"
}
// fetchDebMember reads a member's Release under base and each binary Packages
// index it lists.
func (e *Engine) fetchDebMember(ctx context.Context, name, base string) (*DebMember, error) {
remote, err := e.getRemote(ctx, name)
if err != nil {
return nil, fmt.Errorf("remote %q: %w", name, err)
}
rel, err := e.fetchMemberPath(ctx, *remote, base+"Release")
if err != nil {
return nil, err
}
m := &DebMember{RemoteName: name, Release: rel, Packages: map[string][]byte{}}
for dir, file := range debIndexFiles(parseDebRelease(rel)) {
raw, err := e.fetchMemberPath(ctx, *remote, base+file)
if err != nil {
return nil, err
}
if m.Packages[dir], err = decompress(file, raw); err != nil {
return nil, fmt.Errorf("decompress %s: %w", file, err)
}
}
return m, nil
}
// evictLRU drops the least recently used merges under prefix beyond keep.
func (e *Engine) evictLRU(prefix string, keep int) {
e.mu.Lock()
defer e.mu.Unlock()
g := e.rpm[virt]
var keys []string
for k := range e.gens {
if strings.HasPrefix(k, prefix) {
keys = append(keys, k)
}
}
slices.SortFunc(keys, func(a, b string) int { return e.gens[b].used.Compare(e.gens[a].used) })
for _, k := range keys[min(len(keys), keep):] {
delete(e.gens, k)
}
}
func (e *Engine) cachedFile(key, path string) ([]byte, bool) {
e.mu.Lock()
defer e.mu.Unlock()
g := e.gens[key]
if g == nil {
return nil, false
}
for _, r := range []*RPMRepo{g.cur, g.prev} {
for _, r := range []mergedRepo{g.cur, g.prev} {
if r == nil {
continue
}
if body, ok := r.Files[path]; ok {
if body, ok := r.files()[path]; ok {
g.used = time.Now()
return body, true
}
}
+211
View File
@@ -5,6 +5,7 @@ import (
"errors"
"fmt"
"regexp"
"strings"
"sync"
"sync/atomic"
"testing"
@@ -156,3 +157,213 @@ func TestMemberRedirect(t *testing.T) {
}
}
}
func debEngine(members map[string]*DebMember) *Engine {
e := fakeEngine(nil)
e.debMember = func(_ context.Context, name, base string) (*DebMember, error) {
if m := members[name]; m != nil && base == "" {
return m, nil
}
return nil, errors.New("upstream down")
}
return e
}
func debVirt(members ...string) models.Virtual {
return models.Virtual{Name: "v", PackageType: models.PackageDeb, Members: members}
}
func TestFetchDebFailsClosed(t *testing.T) {
a := flatDebMember("a", "", debStanza("foo", "1", "amd64", "pool/foo.deb"))
e := debEngine(map[string]*DebMember{"a": &a})
for _, p := range []string{"Release", "./Packages.gz", "dists/trixie/Release"} {
if _, _, err := e.Fetch(context.Background(), debVirt("a", "b"), p, ""); err == nil || errors.Is(err, ErrNotFound) {
t.Errorf("%s: want upstream error with a member down, got %v", p, err)
}
}
for _, p := range []string{"InRelease", "./Release.gpg", "Packages.xz"} {
if _, _, err := e.Fetch(context.Background(), debVirt("a"), p, ""); !errors.Is(err, ErrNotFound) {
t.Errorf("%s: want not found, got %v", p, err)
}
}
}
func TestFetchDebRemergesAfterTTL(t *testing.T) {
a := flatDebMember("a", "", debStanza("foo", "1", "amd64", "pool/foo_1.deb"))
e := debEngine(map[string]*DebMember{"a": &a})
virt := debVirt("a")
rel, _, err := e.Fetch(context.Background(), virt, "./Release", "")
if err != nil {
t.Fatal(err)
}
a.Packages[""] = []byte(debStanza("foo", "2", "amd64", "pool/foo_2.deb"))
e.mergeTTL = 0
newRel, _, err := e.Fetch(context.Background(), virt, "Release", "")
if err != nil || string(newRel) == string(rel) {
t.Fatalf("Release should reflect the member change (err %v)", err)
}
pk, ct, err := e.Fetch(context.Background(), virt, "./Packages", "")
if err != nil || ct != "text/plain" || !strings.Contains(string(pk), "Version: 2") {
t.Fatalf("Packages should be the current merge (ct %q, err %v):\n%s", ct, err, pk)
}
}
func TestFetchDebMergeIsSharedPerSuite(t *testing.T) {
var calls atomic.Int32
e := fakeEngine(nil)
e.mergeTTL = time.Minute
e.debMember = func(_ context.Context, name, base string) (*DebMember, error) {
calls.Add(1)
m := flatDebMember(name, "", debStanza("foo", "1", "amd64", "pool/foo.deb"))
return &m, nil
}
for _, p := range []string{"Release", "Packages", "Packages.gz", "./Release", "dists/a/Release", "dists/b/Release"} {
if _, _, err := e.Fetch(context.Background(), debVirt("a"), p, ""); err != nil && !errors.Is(err, ErrNotFound) {
t.Fatal(err)
}
}
if n := calls.Load(); n != 3 {
t.Fatalf("member fetched %d times, want once per Release dir (3)", n)
}
}
func TestMemberRedirectDeb(t *testing.T) {
e := fakeEngine(nil)
virt := debVirt("gh", "loc")
for _, tc := range []struct {
path, want string
ok bool
err error
}{
{path: "gh/o/r/releases/download/v1/foo_1.0_amd64.deb", ok: true, want: "http://h/api/v1/remote/gh/o/r/releases/download/v1/foo_1.0_amd64.deb"},
{path: "./gh/pool/a+b.deb", ok: true, want: "http://h/api/v1/remote/gh/pool/a+b.deb"},
{path: "gh/../loc/x.deb", err: ErrBadPath},
{path: "dists/trixie/Release"},
{path: "Packages"},
} {
got, ok, err := e.MemberRedirect(context.Background(), virt, tc.path, "", "http://h/")
if got != tc.want || ok != tc.ok || !errors.Is(err, tc.err) {
t.Errorf("%s: got (%q, %v, %v), want (%q, %v, %v)", tc.path, got, ok, err, tc.want, tc.ok, tc.err)
}
}
}
type fakeRepo struct {
idx string
fs map[string][]byte
}
func (r *fakeRepo) index() []byte { return []byte(r.idx) }
func (r *fakeRepo) files() map[string][]byte { return r.fs }
func genRepo(gen string) func(context.Context) (mergedRepo, error) {
return func(context.Context) (mergedRepo, error) {
return &fakeRepo{idx: gen, fs: map[string][]byte{gen + ".gz": []byte(gen), "fixed": []byte(gen)}}, nil
}
}
func TestMergedRetainsPreviousGeneration(t *testing.T) {
e := fakeEngine(nil)
for _, gen := range []string{"g1", "g2", "g3"} {
if _, err := e.merged(context.Background(), "k", genRepo(gen)); err != nil {
t.Fatal(err)
}
}
for path, want := range map[string]string{"g3.gz": "g3", "g2.gz": "g2", "fixed": "g3"} {
if body, ok := e.cachedFile("k", path); !ok || string(body) != want {
t.Errorf("%s: got (%q, %v), want %q", path, body, ok, want)
}
}
if _, ok := e.cachedFile("k", "g1.gz"); ok {
t.Error("generation two merges back must be dropped")
}
}
func TestMergedReusesWithinTTLAndIsolatesKeys(t *testing.T) {
e := fakeEngine(nil)
e.mergeTTL = time.Minute
for _, step := range []struct{ key, gen, want string }{
{"a", "a1", "a1"}, {"b", "b1", "b1"}, {"a", "a2", "a1"},
} {
repo, err := e.merged(context.Background(), step.key, genRepo(step.gen))
if err != nil || string(repo.index()) != step.want {
t.Fatalf("%s: got %v (err %v), want %s", step.key, repo, err, step.want)
}
}
if _, ok := e.cachedFile("b", "a1.gz"); ok {
t.Error("key b served key a's file")
}
if _, ok := e.cachedFile("c", "a1.gz"); ok {
t.Error("unknown key served a file")
}
}
func suiteDebMember(name, version string) *DebMember {
return &DebMember{RemoteName: name, Release: []byte("Suite: s\nSHA256:\n x 1 main/binary-amd64/Packages.gz\n"),
Packages: map[string][]byte{"main/binary-amd64/": []byte(debStanza("foo", version, "amd64", "pool/foo_"+version+".deb"))}}
}
func TestFetchDebByHashSurvivesRemerge(t *testing.T) {
version := "1"
e := fakeEngine(nil)
e.debMember = func(_ context.Context, name, _ string) (*DebMember, error) { return suiteDebMember(name, version), nil }
virt := debVirt("a")
fetch := func(p string) []byte {
body, _, err := e.Fetch(context.Background(), virt, p, "")
if err != nil {
t.Fatalf("%s: %v", p, err)
}
return body
}
byHash := func(rel []byte) string {
m := regexp.MustCompile(`(?m)^SHA256:\n(?: .*\n)*? ([0-9a-f]{64}) \d+ main/binary-amd64/Packages.gz$`).FindSubmatch(rel)
if m == nil {
t.Fatalf("no Packages.gz SHA256 in Release:\n%s", rel)
}
return "dists/trixie/main/binary-amd64/by-hash/SHA256/" + string(m[1])
}
oldRel := fetch("dists/trixie/Release")
version, e.mergeTTL = "2", 0
if newRel := fetch("dists/trixie/Release"); string(newRel) == string(oldRel) {
t.Fatal("Release should reflect the member change")
}
if got := string(gunzip(t, fetch(byHash(oldRel)))); !strings.Contains(got, "Version: 1\n") {
t.Fatalf("by-hash from the previous Release served the wrong index:\n%s", got)
}
if got := string(fetch("dists/trixie/main/binary-amd64/Packages")); !strings.Contains(got, "Version: 2\n") {
t.Fatalf("fixed-name Packages should be the current merge:\n%s", got)
}
}
func TestFetchDebBoundsSuitesPerVirtual(t *testing.T) {
e := fakeEngine(nil)
e.mergeTTL = time.Minute
e.debMember = func(_ context.Context, name, _ string) (*DebMember, error) { return suiteDebMember(name, "1"), nil }
other := models.Virtual{Name: "other", PackageType: models.PackageDeb, Members: []string{"a"}}
if _, _, err := e.Fetch(context.Background(), other, "dists/keep/Release", ""); err != nil {
t.Fatal(err)
}
for i := range maxDebSuites + 5 {
if _, _, err := e.Fetch(context.Background(), debVirt("a"), fmt.Sprintf("dists/s%d/Release", i), ""); err != nil {
t.Fatal(err)
}
time.Sleep(time.Microsecond)
}
n := 0
for k := range e.gens {
if strings.HasPrefix(k, "deb\x00v\x00") {
n++
}
}
if n != maxDebSuites {
t.Fatalf("virtual v caches %d suites, want %d", n, maxDebSuites)
}
if e.gens["deb\x00v\x00dists/s0/"] != nil || e.gens["deb\x00v\x00dists/s4/"] != nil {
t.Fatal("least recently used suites should be evicted first")
}
if e.gens["deb\x00other\x00dists/keep/"] == nil {
t.Fatal("another virtual's suites must not be evicted")
}
}
+9
View File
@@ -0,0 +1,9 @@
-- Failed GitHub release scans retry with backoff instead of waiting out a full
-- mutable_ttl: sync_failures counts consecutive failures, next_retry_at gates
-- the next claim while a retry is pending.
ALTER TABLE github_rpm_sync_state ADD COLUMN IF NOT EXISTS sync_failures INT NOT NULL DEFAULT 0;
ALTER TABLE github_rpm_sync_state ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ;
ALTER TABLE github_deb_sync_state ADD COLUMN IF NOT EXISTS sync_failures INT NOT NULL DEFAULT 0;
ALTER TABLE github_deb_sync_state ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ;
ALTER TABLE github_alpine_sync_state ADD COLUMN IF NOT EXISTS sync_failures INT NOT NULL DEFAULT 0;
ALTER TABLE github_alpine_sync_state ADD COLUMN IF NOT EXISTS next_retry_at TIMESTAMPTZ;
+2
View File
@@ -17,6 +17,7 @@ const (
PackageTerraform PackageType = "terraform"
PackageGoProxy PackageType = "goproxy"
PackageCargo PackageType = "cargo"
PackageNuGet PackageType = "nuget"
PackageGitHubRPM PackageType = "github_rpm"
PackageGitHubDeb PackageType = "github_deb"
PackageGitHubAlpine PackageType = "github_alpine"
@@ -35,6 +36,7 @@ var validPackageTypes = map[PackageType]bool{
PackageTerraform: true,
PackageGoProxy: true,
PackageCargo: true,
PackageNuGet: true,
PackageGitHubRPM: true,
PackageGitHubDeb: true,
PackageGitHubAlpine: true,
+1
View File
@@ -19,6 +19,7 @@ func TestPackageTypeValid(t *testing.T) {
models.PackageTerraform,
models.PackageGoProxy,
models.PackageCargo,
models.PackageNuGet,
models.PackageGitHubRPM,
models.PackageGitHubDeb,
models.PackageGitHubAlpine,
+16
View File
@@ -243,6 +243,22 @@ index = "sparse+${proxy}/"`,
},
];
case 'nuget':
return [
{
title: 'Use this feed instead of nuget.org',
language: 'xml',
code: `<!-- nuget.config -->
<configuration>
<packageSources>
<clear />
<add key="${name}" value="${proxy}/index.json" protocolVersion="3" />
</packageSources>
</configuration>`,
note: 'Serve artifactapi over HTTPS: NuGet rejects the repository signatures resource over plain HTTP.',
},
];
case 'puppet':
return [
{
+1
View File
@@ -18,6 +18,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
terraform: 'blue',
goproxy: 'green',
cargo: 'yellow',
nuget: 'blue',
};
export function Locals() {
+1
View File
@@ -18,6 +18,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
terraform: 'blue',
goproxy: 'green',
cargo: 'yellow',
nuget: 'blue',
};
export function Remotes() {