Compare commits

..

7 Commits

Author SHA1 Message Date
unkin-agent 1837f6ef8c Add rpm virtual repositories (#131)
ci/woodpecker/tag/docker Pipeline failed
Virtual repos only merge helm and pypi, so several rpm repos (e.g. many github_rpm remotes) cannot be served as one yum repo.

- merge member primary/filelists/other into one repodata set; member order wins duplicate NEVRAs
- read member repodata through local, github_rpm and proxied remote paths
- prefix package locations (incl. xml:base under a member upstream) with the member name and 302 them to the member route
- reject absolute and dot-segment member paths; escape the redirect and keep its query
- return 502 when any member's repodata is unavailable
- reuse each virtual's merge for 60s behind singleflight; serve data files from its current and previous merge (per replica)
- add merger/engine unit tests and a dockerised dnf e2e case

Reviewed-on: #131
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-09 22:07:20 +11:00
unkin-agent 80368986bb Add cargo sparse registry remote type (#132)
Rust crate builds (rpmbuilder CI) fetch from crates.io directly because artifactapi has no Cargo remote type.

- add `cargo` package type proxying the sparse registry protocol (RFC 2789)
- synthesize `config.json` so `dl` points crate downloads back at the remote
- treat index files as mutable and `crates/*/*.crate` as immutable, fetched from static.crates.io for crates.io
- add a `~/.cargo/config.toml` usage snippet to the UI and a cargo e2e caching case

Reviewed-on: #132
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-09 21:44:17 +11:00
unkin-agent bda762b10e Fix objects test compile after Routes evictor change (#130)
ci/woodpecker/tag/docker Pipeline was successful
ObjectsHandler.Routes now takes an Evictor, but TestObjectsListPrefix still calls it with no arguments, so the api/v2 test package fails to compile on master.

- pass the existing fakeEvictor to Routes in objects_test.go

Reviewed-on: #130
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-04 22:11:19 +11:00
unkin-agent a71d126239 Filter remote object listings by prefix (#129)
`GET /api/v2/remotes/{name}/objects` dropped the `prefix` query parameter, so a filtered listing returned the first page of every artifact in the remote.

- add a `prefix` argument to `ListArtifacts` that matches path prefixes
- pass the `prefix` query parameter through from the objects list handler

Reviewed-on: #129
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-04 22:07:21 +11:00
unkin-agent 6a08539a78 Evict remote objects from every cache layer (#128)
Evicting a remote object only deleted its Postgres row. The S3 index and the Redis TTL/ETag keys survived, so stale mirror metadata such as EPEL repodata kept being served.

- evict the artifact row, S3 index and Redis keys via `Engine.Evict`
- evict a directory with `<dir>/*`; other wildcards return 400, unknown remotes 404
- wait on the per-path fetch lock for single-path evicts; return 503 on timeout or a Redis error
- wildcard evicts take no lock; a fetch already in flight may re-cache its path
- return S3 list errors from `DeletePrefix` instead of deleting nothing

Reviewed-on: #128
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-04 22:07:14 +11:00
unkin-agent 0b159dad90 ci: move Go steps to the estate-built gobuilder image (#127)
Packer-built almalinux9-gobuilder is retiring. Moves test and
pre-commit steps to artifactapi docker-internal/gobuilder 0.1.2-alma9
(estate CA, go1.26.7, go-cache-plugin baked in).

- Repoint both steps at the new image, quoted (has a colon)
- Drop curl/sha256sum bootstrap for go-cache-plugin; it's on PATH now
- Set GOCACHEPROG as a static env var instead of a conditional export
- Verified build/test still pass with unreachable/invalid S3 creds —
  best-effort degrade is a plugin runtime property, unaffected

Reviewed-on: #127
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-03 00:26:02 +10:00
unkin-agent 7a4f4054cc ci: back Go compiles with the shared S3 build cache (#126)
Every pipeline run recompiled the whole tree from scratch; the shared gocache S3 bucket already holds those objects.

- Bootstrap the go-cache-plugin GOCACHEPROG in the test and pre-commit steps, keyed `ci-artifactapi`.
- Best-effort: a failed plugin fetch leaves GOCACHEPROG unset, and S3 errors degrade to cache misses, so an outage never fails a build.
- Run tests on the gobuilder image, which trusts the internal CA the S3 endpoint presents.

Container builds still compile uncached; plumbing the cache into the Dockerfile is a follow-up.

Reviewed-on: #126
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-09-27 17:33:19 +10:00
32 changed files with 2020 additions and 20 deletions
+15 -1
View File
@@ -3,14 +3,28 @@ when:
steps: steps:
- name: pre-commit - name: pre-commit
image: git.unkin.net/unkin/almalinux9-gobuilder:20260606 # gobuilder trusts the internal CA, which the S3 build cache endpoint needs.
# go-cache-plugin is baked into the image; S3 errors degrade to cache misses.
image: "artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9"
commands: commands:
- uvx pre-commit run --all-files - uvx pre-commit run --all-files
environment: environment:
# golib lives on Gitea; skip the public proxy/sum db. # golib lives on Gitea; skip the public proxy/sum db.
GOPRIVATE: git.unkin.net GOPRIVATE: git.unkin.net
GOCACHEPROG: "go-cache-plugin --cache-dir=/tmp/gocache"
GOCACHE_S3_BUCKET: gocache
# Explicit region skips a GetBucketLocation probe RGW handles poorly.
GOCACHE_S3_REGION: us-east-1
GOCACHE_S3_ENDPOINT_URL: "https://s3.ceph.unkin.net"
GOCACHE_S3_PATH_STYLE: "true"
GOCACHE_KEY_PREFIX: ci-artifactapi
AWS_ACCESS_KEY_ID:
from_secret: GOCACHE_AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY:
from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
backend_options: backend_options:
kubernetes: kubernetes:
serviceAccountName: default
resources: resources:
requests: requests:
memory: 512Mi memory: 512Mi
+24 -1
View File
@@ -3,9 +3,32 @@ when:
steps: steps:
- name: test - name: test
image: golang:1.25 # gobuilder trusts the internal CA, which the S3 build cache endpoint needs.
# go-cache-plugin is baked into the image; S3 errors degrade to cache misses.
image: "artifactapi.k8s.syd1.au.unkin.net/docker-internal/gobuilder:0.1.2-alma9"
commands: commands:
- go test -race -count=1 ./pkg/... ./internal/... - go test -race -count=1 ./pkg/... ./internal/...
environment: environment:
# golib lives on Gitea; skip the public proxy/sum db. # golib lives on Gitea; skip the public proxy/sum db.
GOPRIVATE: git.unkin.net GOPRIVATE: git.unkin.net
GOCACHEPROG: "go-cache-plugin --cache-dir=/tmp/gocache"
GOCACHE_S3_BUCKET: gocache
# Explicit region skips a GetBucketLocation probe RGW handles poorly.
GOCACHE_S3_REGION: us-east-1
GOCACHE_S3_ENDPOINT_URL: "https://s3.ceph.unkin.net"
GOCACHE_S3_PATH_STYLE: "true"
GOCACHE_KEY_PREFIX: ci-artifactapi
AWS_ACCESS_KEY_ID:
from_secret: GOCACHE_AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY:
from_secret: GOCACHE_AWS_SECRET_ACCESS_KEY
backend_options:
kubernetes:
serviceAccountName: default
resources:
requests:
memory: 1Gi
cpu: 1
limits:
memory: 4Gi
cpu: 2
+2 -1
View File
@@ -1,6 +1,6 @@
# ArtifactAPI # ArtifactAPI
Caching proxy for package repositories. Single Go binary, 10 package types, content-addressable storage, managed by Terraform. Caching proxy for package repositories. Single Go binary, 11 package types, content-addressable storage, managed by Terraform.
## Quick Start ## Quick Start
@@ -32,6 +32,7 @@ API: `http://localhost:8000` | Frontend: `http://localhost:5173`
| `puppet` | `v3/modules/*`, `v3/releases*` | `.tar.gz` | | `puppet` | `v3/modules/*`, `v3/releases*` | `.tar.gz` |
| `terraform` | `*/versions` | `*/download/*/*` | | `terraform` | `*/versions` | `*/download/*/*` |
| `goproxy` | `@v/list`, `@latest` | `.info`, `.mod`, `.zip` | | `goproxy` | `@v/list`, `@latest` | `.info`, `.mod`, `.zip` |
| `cargo` | sparse index files (`config.json` synthesized) | `crates/*/*.crate` |
| `github_rpm` | `repodata/*` (synthesized) | `.rpm` (redirected) | | `github_rpm` | `repodata/*` (synthesized) | `.rpm` (redirected) |
Providers classify paths automatically. Users only configure what to proxy and TTLs. Providers classify paths automatically. Users only configure what to proxy and TTLs.
+1 -1
View File
@@ -23,7 +23,7 @@ already-running stack.
- **Repository lifecycle** — add / change / delete for remote, local and virtual repos. - **Repository lifecycle** — add / change / delete for remote, local and virtual repos.
- **Caching** — one immutable artifact per remote package type (generic, docker, - **Caching** — one immutable artifact per remote package type (generic, docker,
helm, pypi, npm, rpm, alpine, puppet, terraform, goproxy) proxied through the helm, pypi, npm, rpm, alpine, puppet, terraform, goproxy, cargo) proxied through the
mock upstream: first fetch `X-Artifact-Source: remote`, second `cache`, bytes mock upstream: first fetch `X-Artifact-Source: remote`, second `cache`, bytes
verified against the origin fixture. verified against the origin fixture.
- **Local uploads** — generic (upload/download), pypi (wheel + generated `simple/` - **Local uploads** — generic (upload/download), pypi (wheel + generated `simple/`
+1
View File
@@ -29,6 +29,7 @@ func TestCachingPerProvider(t *testing.T) {
{"alpine", "alpine/x86_64/testpkg-1.0-r0.apk", "alpine/x86_64/testpkg-1.0-r0.apk"}, {"alpine", "alpine/x86_64/testpkg-1.0-r0.apk", "alpine/x86_64/testpkg-1.0-r0.apk"},
{"puppet", "puppet-releases/author-mod-1.0.0.tar.gz", "puppet-releases/author-mod-1.0.0.tar.gz"}, {"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"}, {"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"},
{"terraform", "hashicorp/aws/download/pkg.zip", "v1/providers/hashicorp/aws/download/pkg.zip"}, {"terraform", "hashicorp/aws/download/pkg.zip", "v1/providers/hashicorp/aws/download/pkg.zip"},
{"docker", "library/testimg/blobs/blobdata", "v2/library/testimg/blobs/blobdata"}, {"docker", "library/testimg/blobs/blobdata", "v2/library/testimg/blobs/blobdata"},
} }
+84
View File
@@ -3,9 +3,17 @@
package e2edocker package e2edocker
import ( import (
"bytes"
"compress/gzip"
"encoding/xml"
"fmt"
"io"
"net/http" "net/http"
"os"
"os/exec"
"strings" "strings"
"testing" "testing"
"time"
) )
// TestVirtualPyPIMerge uploads different packages to two pypi locals and // TestVirtualPyPIMerge uploads different packages to two pypi locals and
@@ -52,3 +60,79 @@ func TestVirtualHelmMerge(t *testing.T) {
t.Fatalf("merged helm index missing a member chart (want alpha and beta): %s", s) t.Fatalf("merged helm index missing a member chart (want alpha and beta): %s", s)
} }
} }
// TestVirtualRPMMerge merges a remote rpm repo and a local rpm repo carrying the
// same package: the first member wins the duplicate, its package download
// routes through the virtual, and a real dnf can consume the merged repo.
func TestVirtualRPMMerge(t *testing.T) {
createRepo(t, `{"name":"vrpm-remote","package_type":"rpm","repo_type":"remote","base_url":"`+mockUpstream()+`/rpm-mirror","stale_on_error":true}`)
createRepo(t, `{"name":"vrpm-local","package_type":"rpm","repo_type":"local"}`)
defer deleteRepo(t, "vrpm-remote")
defer deleteRepo(t, "vrpm-local")
pkg := fixtureBytes(t, "rpmrepo/Packages/e2e-testpkg-1.0-1.noarch.rpm")
uploadFile(t, "vrpm-local", "e2e-testpkg-1.0-1.noarch.rpm", pkg, "application/x-rpm")
createVirtual(t, `{"name":"vrpm","package_type":"rpm","members":["vrpm-remote","vrpm-local"]}`)
defer deleteVirtual(t, "vrpm")
resp, body := getEventually(t, api("/api/v1/virtual/vrpm/repodata/repomd.xml"), 15*time.Second)
if resp.StatusCode != http.StatusOK {
t.Fatalf("virtual repomd.xml: status %d: %s", resp.StatusCode, body)
}
var md struct {
Data []struct {
Type string `xml:"type,attr"`
Location struct {
Href string `xml:"href,attr"`
} `xml:"location"`
} `xml:"data"`
}
if err := xml.Unmarshal(body, &md); err != nil {
t.Fatalf("parse repomd.xml: %v\n%s", err, body)
}
var primary []byte
for _, d := range md.Data {
if d.Type == "primary" {
resp, gz := doRequest(t, http.MethodGet, api("/api/v1/virtual/vrpm/"+d.Location.Href), nil, "")
if resp.StatusCode != http.StatusOK {
t.Fatalf("primary: status %d", resp.StatusCode)
}
r, err := gzip.NewReader(bytes.NewReader(gz))
if err != nil {
t.Fatal(err)
}
primary, _ = io.ReadAll(r)
}
}
href := "vrpm-remote/Packages/e2e-testpkg-1.0-1.noarch.rpm"
if strings.Count(string(primary), "<name>e2e-testpkg</name>") != 1 || !strings.Contains(string(primary), `href="`+href+`"`) {
t.Fatalf("merged primary should hold one e2e-testpkg owned by the first member:\n%s", primary)
}
resp, got := doRequest(t, http.MethodGet, api("/api/v1/virtual/vrpm/"+href), nil, "")
if resp.StatusCode != http.StatusOK || !bytes.Equal(got, pkg) {
t.Fatalf("package via virtual: status %d, %d bytes (want %d)", resp.StatusCode, len(got), len(pkg))
}
network := os.Getenv("COMPOSE_NETWORK")
internal := os.Getenv("ARTIFACTAPI_INTERNAL")
if network == "" || internal == "" {
t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; skipping real dnf")
}
if _, err := exec.LookPath("docker"); err != nil {
t.Skip("docker not available on the test host")
}
repoConf := fmt.Sprintf("[vrpm]\nname=vrpm\nbaseurl=%s/api/v1/virtual/vrpm/\nenabled=1\ngpgcheck=0\nmetadata_expire=0\n", strings.TrimRight(internal, "/"))
script := "set -euo pipefail; " +
"printf '%s' \"$REPO\" > /etc/yum.repos.d/vrpm.repo; " +
"dnf -y --disablerepo='*' --enablerepo=vrpm makecache; " +
"dnf -y --disablerepo='*' --enablerepo=vrpm list --available; " +
"dnf -y --disablerepo='*' --enablerepo=vrpm install e2e-testpkg; " +
"rpm -q e2e-testpkg"
out, err := exec.Command("docker", "run", "--rm", "--network", network, "-e", "REPO="+repoConf,
"rockylinux:9", "bash", "-c", script).CombinedOutput()
if err != nil {
t.Fatalf("real dnf against rpm virtual failed: %v\n%s", err, out)
}
}
+1 -1
View File
@@ -18,6 +18,7 @@ require (
github.com/testcontainers/testcontainers-go/modules/redis v0.42.0 github.com/testcontainers/testcontainers-go/modules/redis v0.42.0
github.com/ulikunitz/xz v0.5.16 github.com/ulikunitz/xz v0.5.16
golang.org/x/crypto v0.54.0 golang.org/x/crypto v0.54.0
golang.org/x/sync v0.22.0
golang.org/x/time v0.15.0 golang.org/x/time v0.15.0
gopkg.in/yaml.v3 v3.0.1 gopkg.in/yaml.v3 v3.0.1
) )
@@ -101,7 +102,6 @@ require (
go.uber.org/atomic v1.11.0 // indirect go.uber.org/atomic v1.11.0 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect
golang.org/x/net v0.56.0 // indirect golang.org/x/net v0.56.0 // indirect
golang.org/x/sync v0.22.0 // indirect
golang.org/x/sys v0.47.0 // indirect golang.org/x/sys v0.47.0 // indirect
golang.org/x/text v0.40.0 // indirect golang.org/x/text v0.40.0 // indirect
gopkg.in/ini.v1 v1.67.2 // indirect gopkg.in/ini.v1 v1.67.2 // indirect
+14
View File
@@ -162,7 +162,21 @@ func (h *ProxyHandler) handleVirtual(w http.ResponseWriter, r *http.Request) {
proxyBaseURL := fmt.Sprintf("%s://%s", scheme(r), r.Host) proxyBaseURL := fmt.Sprintf("%s://%s", scheme(r), r.Host)
loc, ok, err := h.virtualEngine.MemberRedirect(r.Context(), *virt, path, r.URL.RawQuery, proxyBaseURL)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if ok {
http.Redirect(w, r, loc, http.StatusFound)
return
}
body, contentType, err := h.virtualEngine.Fetch(r.Context(), *virt, path, proxyBaseURL) body, contentType, err := h.virtualEngine.Fetch(r.Context(), *virt, path, proxyBaseURL)
if errors.Is(err, virtual.ErrNotFound) {
http.Error(w, "not found", http.StatusNotFound)
return
}
if err != nil { if err != nil {
slog.Error("virtual fetch failed", "virtual", virtualName, "path", path, "error", err) slog.Error("virtual fetch failed", "virtual", virtualName, "path", path, "error", err)
http.Error(w, "bad gateway", http.StatusBadGateway) http.Error(w, "bad gateway", http.StatusBadGateway)
+21 -5
View File
@@ -1,6 +1,8 @@
package v2 package v2
import ( import (
"context"
"errors"
"fmt" "fmt"
"net/http" "net/http"
"strconv" "strconv"
@@ -8,8 +10,14 @@ import (
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
"git.unkin.net/unkin/artifactapi/internal/database" "git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/internal/proxy"
) )
// Evictor drops a remote path from every cache layer.
type Evictor interface {
Evict(ctx context.Context, remoteName, path string) error
}
type ObjectsHandler struct { type ObjectsHandler struct {
db *database.DB db *database.DB
} }
@@ -18,10 +26,13 @@ func NewObjectsHandler(db *database.DB) *ObjectsHandler {
return &ObjectsHandler{db: db} return &ObjectsHandler{db: db}
} }
func (h *ObjectsHandler) Routes() chi.Router { // Routes lists and evicts objects for remote repos; evictor serves the DELETE.
func (h *ObjectsHandler) Routes(evictor Evictor) chi.Router {
r := chi.NewRouter() r := chi.NewRouter()
r.Get("/", h.list) r.Get("/", h.list)
r.Delete("/*", h.evict) r.Delete("/*", func(w http.ResponseWriter, r *http.Request) {
evict(w, r, evictor)
})
return r return r
} }
@@ -51,7 +62,7 @@ func (h *ObjectsHandler) list(w http.ResponseWriter, r *http.Request) {
remoteName := chi.URLParam(r, "name") remoteName := chi.URLParam(r, "name")
limit, offset := pageBounds(r) limit, offset := pageBounds(r)
artifacts, err := h.db.ListArtifacts(r.Context(), remoteName, limit, offset) artifacts, err := h.db.ListArtifacts(r.Context(), remoteName, r.URL.Query().Get("prefix"), limit, offset)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return return
@@ -82,11 +93,16 @@ func (h *ObjectsHandler) evictLocal(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent) w.WriteHeader(http.StatusNoContent)
} }
func (h *ObjectsHandler) evict(w http.ResponseWriter, r *http.Request) { func evict(w http.ResponseWriter, r *http.Request, evictor Evictor) {
remoteName := chi.URLParam(r, "name") remoteName := chi.URLParam(r, "name")
path := chi.URLParam(r, "*") path := chi.URLParam(r, "*")
if err := h.db.DeleteArtifact(r.Context(), remoteName, path); err != nil { if err := evictor.Evict(r.Context(), remoteName, path); err != nil {
var proxyErr *proxy.ProxyError
if errors.As(err, &proxyErr) {
http.Error(w, proxyErr.Message, proxyErr.Status)
return
}
http.Error(w, fmt.Sprintf("evict failed: %v", err), http.StatusInternalServerError) http.Error(w, fmt.Sprintf("evict failed: %v", err), http.StatusInternalServerError)
return return
} }
+59
View File
@@ -0,0 +1,59 @@
package v2
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"testing"
"github.com/go-chi/chi/v5"
"git.unkin.net/unkin/artifactapi/internal/proxy"
)
type fakeEvictor struct {
remote, path string
err error
}
func (f *fakeEvictor) Evict(_ context.Context, remote, path string) error {
f.remote, f.path = remote, path
return f.err
}
func deleteObject(ev Evictor, path string) int {
router := chi.NewRouter()
router.Route("/remotes/{name}/objects", func(r chi.Router) {
r.Delete("/*", NewObjectsHandler(nil).Routes(ev).ServeHTTP)
})
w := httptest.NewRecorder()
router.ServeHTTP(w, httptest.NewRequest("DELETE", "/remotes/epel/objects/"+path, nil))
return w.Code
}
func TestRemoteEvictDelegatesToEvictor(t *testing.T) {
for _, path := range []string{"8/Everything/x86_64/repodata/repomd.xml", "8/Everything/x86_64/repodata/*"} {
ev := &fakeEvictor{}
if code := deleteObject(ev, path); code != 204 || ev.remote != "epel" || ev.path != path {
t.Errorf("DELETE %s: code=%d evicted=%q/%q", path, code, ev.remote, ev.path)
}
}
}
func TestRemoteEvictMapsErrorStatus(t *testing.T) {
for name, tc := range map[string]struct {
err error
want int
}{
"bad wildcard": {&proxy.ProxyError{Status: http.StatusBadRequest, Message: "wildcard evict must be <dir>/*"}, http.StatusBadRequest},
"unknown remote": {&proxy.ProxyError{Status: http.StatusNotFound, Message: "remote not found"}, http.StatusNotFound},
"lock busy": {fmt.Errorf("wrapped: %w", &proxy.ProxyError{Status: http.StatusServiceUnavailable, Message: "retry"}), http.StatusServiceUnavailable},
"other error": {errors.New("db down"), http.StatusInternalServerError},
} {
if code := deleteObject(&fakeEvictor{err: tc.err}, "x"); code != tc.want {
t.Errorf("%s: code = %d, want %d", name, code, tc.want)
}
}
}
+68
View File
@@ -0,0 +1,68 @@
package v2
import (
"context"
"encoding/json"
"net/http/httptest"
"testing"
"github.com/go-chi/chi/v5"
"git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// TestObjectsListPrefix verifies the remote objects listing passes ?prefix=
// through to the database filter.
func TestObjectsListPrefix(t *testing.T) {
if testDSN == "" {
t.Skip("Docker unavailable")
}
ctx := context.Background()
db, err := database.New(testDSN)
if err != nil {
t.Fatal(err)
}
defer db.Close()
const remote = "generic-objs-prefix"
if err := db.CreateRemote(ctx, &models.Remote{
Name: remote, PackageType: models.PackageGeneric, RepoType: models.RepoTypeRemote,
BaseURL: "https://example.com", MutableTTL: 3600,
}); err != nil {
t.Fatal(err)
}
const hash = "sha256:bb22"
if err := db.UpsertBlob(ctx, hash, "blobs/bb/22", 10, "text/plain"); err != nil {
t.Fatal(err)
}
for _, p := range []string{"a/one.txt", "b/two.txt"} {
if err := db.UpsertArtifact(ctx, remote, p, hash, ""); err != nil {
t.Fatal(err)
}
}
router := chi.NewRouter()
router.Mount("/remotes/{name}/objects", NewObjectsHandler(db).Routes(&fakeEvictor{}))
list := func(query string) []models.Artifact {
t.Helper()
w := httptest.NewRecorder()
router.ServeHTTP(w, httptest.NewRequest("GET", "/remotes/"+remote+"/objects"+query, nil))
if w.Code != 200 {
t.Fatalf("list%s = %d, want 200", query, w.Code)
}
var got []models.Artifact
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
t.Fatalf("decode: %v", err)
}
return got
}
if got := list(""); len(got) != 2 {
t.Fatalf("unfiltered listing returned %d objects, want 2", len(got))
}
if got := list("?prefix=b/"); len(got) != 1 || got[0].Path != "b/two.txt" {
t.Fatalf("prefix=b/ listing = %+v, want only b/two.txt", got)
}
}
+69
View File
@@ -131,3 +131,72 @@ func TestFlushRemote(t *testing.T) {
t.Error("expected keys flushed") t.Error("expected keys flushed")
} }
} }
func setPathKeys(t *testing.T, remote string, paths ...string) {
t.Helper()
ctx := context.Background()
for _, p := range paths {
if err := testRedis.SetTTL(ctx, remote, p, time.Minute); err != nil {
t.Fatal(err)
}
if err := testRedis.SetETag(ctx, remote, p, `"e"`, time.Minute); err != nil {
t.Fatal(err)
}
}
}
func pathKeysExist(t *testing.T, remote, path string) (ttl, etag bool) {
t.Helper()
ctx := context.Background()
ttl, _ = testRedis.CheckTTL(ctx, remote, path)
e, _ := testRedis.GetETag(ctx, remote, path)
return ttl, e != ""
}
func TestForgetPath(t *testing.T) {
requireRedis(t)
const meta = `repo/a*b?[c]\d.xml`
setPathKeys(t, "fp", meta, "repo/aXb.xml")
setPathKeys(t, "fp-other", meta)
if err := testRedis.ForgetPath(context.Background(), "fp", meta); err != nil {
t.Fatal(err)
}
if ttl, etag := pathKeysExist(t, "fp", meta); ttl || etag {
t.Errorf("forgotten path keys remain: ttl=%v etag=%v", ttl, etag)
}
for _, k := range [][2]string{{"fp", "repo/aXb.xml"}, {"fp-other", meta}} {
if ttl, etag := pathKeysExist(t, k[0], k[1]); !ttl || !etag {
t.Errorf("%s:%s lost keys: ttl=%v etag=%v", k[0], k[1], ttl, etag)
}
}
}
func TestForgetPrefix(t *testing.T) {
requireRedis(t)
const prefix = `r*[1]?\/`
under := []string{prefix + "repomd.xml", prefix + "sub/x.rpm"}
// Each would match the prefix if its glob metacharacters were left unescaped.
globMatches := []string{`rX11/repomd.xml`, `r[1]?\/x`, `r*1Z/x`}
setPathKeys(t, "fx", append(under, globMatches...)...)
setPathKeys(t, "fx-other", under...)
if err := testRedis.ForgetPrefix(context.Background(), "fx", prefix); err != nil {
t.Fatal(err)
}
for _, p := range under {
if ttl, etag := pathKeysExist(t, "fx", p); ttl || etag {
t.Errorf("%s keys remain: ttl=%v etag=%v", p, ttl, etag)
}
}
for _, p := range globMatches {
if ttl, etag := pathKeysExist(t, "fx", p); !ttl || !etag {
t.Errorf("%s outside prefix lost keys: ttl=%v etag=%v", p, ttl, etag)
}
}
for _, p := range under {
if ttl, etag := pathKeysExist(t, "fx-other", p); !ttl || !etag {
t.Errorf("other remote %s lost keys: ttl=%v etag=%v", p, ttl, etag)
}
}
}
+25
View File
@@ -3,6 +3,7 @@ package cache
import ( import (
"context" "context"
"fmt" "fmt"
"strings"
"time" "time"
"github.com/redis/go-redis/v9" "github.com/redis/go-redis/v9"
@@ -115,3 +116,27 @@ func (r *Redis) FlushRemote(ctx context.Context, remote string) error {
} }
return iter.Err() return iter.Err()
} }
// ForgetPath drops the freshness and ETag keys of one cached path.
func (r *Redis) ForgetPath(ctx context.Context, remote, path string) error {
return r.client.Del(ctx, fmt.Sprintf("ttl:%s:%s", remote, path), fmt.Sprintf("etag:%s:%s", remote, path)).Err()
}
// ForgetPrefix drops the freshness and ETag keys of every cached path under prefix.
func (r *Redis) ForgetPrefix(ctx context.Context, remote, prefix string) error {
glob := globEscaper.Replace(remote + ":" + prefix)
for _, kind := range []string{"ttl:", "etag:"} {
iter := r.client.Scan(ctx, 0, kind+glob+"*", 100).Iterator()
for iter.Next(ctx) {
if err := r.client.Del(ctx, iter.Val()).Err(); err != nil {
return err
}
}
if err := iter.Err(); err != nil {
return err
}
}
return nil
}
var globEscaper = strings.NewReplacer(`\`, `\\`, `*`, `\*`, `?`, `\?`, `[`, `\[`, `]`, `\]`)
+12 -4
View File
@@ -65,7 +65,8 @@ func (db *DB) TouchArtifactAccess(ctx context.Context, remoteName, path string)
return err return err
} }
func (db *DB) ListArtifacts(ctx context.Context, remoteName string, limit, offset int) ([]models.Artifact, error) { // ListArtifacts pages a remote's artifacts whose path starts with prefix ("" lists all).
func (db *DB) ListArtifacts(ctx context.Context, remoteName, prefix string, limit, offset int) ([]models.Artifact, error) {
rows, err := db.Pool.Query(ctx, ` rows, err := db.Pool.Query(ctx, `
SELECT a.id, a.remote_name, a.path, a.content_hash, a.upstream_etag, SELECT a.id, a.remote_name, a.path, a.content_hash, a.upstream_etag,
a.upstream_last_modified, a.first_seen_at, a.last_fetched_at, a.upstream_last_modified, a.first_seen_at, a.last_fetched_at,
@@ -73,10 +74,10 @@ func (db *DB) ListArtifacts(ctx context.Context, remoteName string, limit, offse
b.size_bytes, b.content_type b.size_bytes, b.content_type
FROM artifacts a FROM artifacts a
JOIN blobs b ON a.content_hash = b.content_hash JOIN blobs b ON a.content_hash = b.content_hash
WHERE a.remote_name = $1 WHERE a.remote_name = $1 AND left(a.path, length($2)) = $2
ORDER BY a.path ORDER BY a.path
LIMIT $2 OFFSET $3 LIMIT $3 OFFSET $4
`, remoteName, limit, offset) `, remoteName, prefix, limit, offset)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -103,6 +104,13 @@ func (db *DB) DeleteArtifact(ctx context.Context, remoteName, path string) error
return err return err
} }
// DeleteArtifactsByPrefix removes every artifact row of a remote whose path
// starts with prefix.
func (db *DB) DeleteArtifactsByPrefix(ctx context.Context, remoteName, prefix string) error {
_, err := db.Pool.Exec(ctx, `DELETE FROM artifacts WHERE remote_name = $1 AND left(path, length($2)) = $2`, remoteName, prefix)
return err
}
func (db *DB) InsertAccessLog(ctx context.Context, remoteName, path string, cacheHit bool, sizeBytes int64, upstreamMS int, clientIP string) error { func (db *DB) InsertAccessLog(ctx context.Context, remoteName, path string, cacheHit bool, sizeBytes int64, upstreamMS int, clientIP string) error {
_, err := db.Pool.Exec(ctx, ` _, err := db.Pool.Exec(ctx, `
INSERT INTO access_log (remote_name, path, cache_hit, size_bytes, upstream_ms, client_ip) INSERT INTO access_log (remote_name, path, cache_hit, size_bytes, upstream_ms, client_ip)
+49 -3
View File
@@ -168,10 +168,17 @@ func TestArtifactsAndBlobs(t *testing.T) {
if err := testDB.TouchArtifactAccess(ctx(), "r-art", "path/a.txt"); err != nil { if err := testDB.TouchArtifactAccess(ctx(), "r-art", "path/a.txt"); err != nil {
t.Fatal(err) t.Fatal(err)
} }
arts, err := testDB.ListArtifacts(ctx(), "r-art", 10, 0) if err := testDB.UpsertArtifact(ctx(), "r-art", "other/b.txt", hash, ""); err != nil {
if err != nil || len(arts) != 1 { t.Fatal(err)
}
arts, err := testDB.ListArtifacts(ctx(), "r-art", "", 10, 0)
if err != nil || len(arts) != 2 {
t.Fatalf("list artifacts: %v %v", len(arts), err) t.Fatalf("list artifacts: %v %v", len(arts), err)
} }
arts, err = testDB.ListArtifacts(ctx(), "r-art", "path/", 10, 0)
if err != nil || len(arts) != 1 || arts[0].Path != "path/a.txt" {
t.Fatalf("list artifacts with prefix: %+v %v", arts, err)
}
if err := testDB.InsertAccessLog(ctx(), "r-art", "path/a.txt", true, 10, 5, "1.2.3.4"); err != nil { if err := testDB.InsertAccessLog(ctx(), "r-art", "path/a.txt", true, 10, 5, "1.2.3.4"); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -188,6 +195,45 @@ func TestArtifactsAndBlobs(t *testing.T) {
} }
} }
func TestListArtifactsPrefix(t *testing.T) {
requireDB(t)
seedRemote(t, "r-prefix")
seedBlob(t, "prefixhash")
for _, p := range []string{"a%b/y", "a1b/y", "a_b/x", "aXb/x", "pkg/1", "pkg/2", "pkg/3", "pkgx/4"} {
if err := testDB.UpsertArtifact(ctx(), "r-prefix", p, "sha256:prefixhash", ""); err != nil {
t.Fatal(err)
}
}
paths := func(prefix string, limit, offset int) []string {
t.Helper()
arts, err := testDB.ListArtifacts(ctx(), "r-prefix", prefix, limit, offset)
if err != nil {
t.Fatalf("list %q: %v", prefix, err)
}
out := make([]string, len(arts))
for i, a := range arts {
out[i] = a.Path
}
return out
}
// LIKE wildcards in the prefix must match literally.
if got := paths("a_b/", 10, 0); len(got) != 1 || got[0] != "a_b/x" {
t.Fatalf("prefix a_b/ = %v, want [a_b/x]", got)
}
if got := paths("a%", 10, 0); len(got) != 1 || got[0] != "a%b/y" {
t.Fatalf("prefix a%% = %v, want [a%%b/y]", got)
}
// limit/offset page the filtered set, not the whole remote.
if got := paths("pkg/", 2, 0); len(got) != 2 || got[0] != "pkg/1" || got[1] != "pkg/2" {
t.Fatalf("prefix pkg/ page 1 = %v, want [pkg/1 pkg/2]", got)
}
if got := paths("pkg/", 2, 2); len(got) != 1 || got[0] != "pkg/3" {
t.Fatalf("prefix pkg/ page 2 = %v, want [pkg/3]", got)
}
}
func TestOrphanAndColdCleanup(t *testing.T) { func TestOrphanAndColdCleanup(t *testing.T) {
requireDB(t) requireDB(t)
seedBlob(t, "orphanhash") seedBlob(t, "orphanhash")
@@ -320,7 +366,7 @@ func TestDatabaseErrorPaths(t *testing.T) {
if _, err := bad.ListVirtuals(ctx); err == nil { if _, err := bad.ListVirtuals(ctx); err == nil {
t.Error("ListVirtuals should error") t.Error("ListVirtuals should error")
} }
if _, err := bad.ListArtifacts(ctx, "r", 10, 0); err == nil { if _, err := bad.ListArtifacts(ctx, "r", "", 10, 0); err == nil {
t.Error("ListArtifacts should error") t.Error("ListArtifacts should error")
} }
if _, err := bad.ListLocalFiles(ctx, "r", 10, 0); err == nil { if _, err := bad.ListLocalFiles(ctx, "r", 10, 0); err == nil {
+92
View File
@@ -0,0 +1,92 @@
// Package cargo proxies a Cargo sparse registry (RFC 2789): mutable index
// files plus immutable .crate downloads, with config.json synthesized so
// cargo fetches crates back through artifactapi.
package cargo
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{})
}
const (
crateDir = "crates/"
cratesIOIndex = "index.crates.io"
cratesIODownload = "https://static.crates.io"
)
type Provider struct{}
func (p *Provider) Type() models.PackageType { return models.PackageCargo }
func (p *Provider) Classify(path string) provider.Mutability {
if isCrate(path) {
return provider.Immutable
}
return provider.Mutable
}
func (p *Provider) ContentType(path string) string {
if isCrate(path) {
return "application/gzip"
}
if path == "config.json" {
return "application/json"
}
return "text/plain"
}
// UpstreamURL maps crates/{crate}/{crate}-{version}.crate to the download host
// and everything else to the index. crates.io splits the two across hosts;
// any other upstream is expected to serve both.
func (p *Provider) UpstreamURL(remote models.Remote, path string) string {
path = strings.TrimLeft(path, "/")
base := strings.TrimRight(remote.BaseURL, "/")
if isCrate(path) {
if u, err := url.Parse(base); err == nil && u.Host == cratesIOIndex {
base = cratesIODownload
}
}
return base + "/" + path
}
func (p *Provider) RewriteResponse(_ []byte, _ models.Remote, _ string) ([]byte, error) {
return nil, nil
}
func (p *Provider) AuthHeaders(_ context.Context, remote models.Remote) (http.Header, error) {
return auth.BasicHeaders(remote), nil
}
// ServeRemote answers config.json itself: dl must name the requesting host,
// which a cached upstream body cannot.
func (p *Provider) ServeRemote(w http.ResponseWriter, r *http.Request, remote models.Remote, path, proxyBaseURL string, _ provider.RemoteMetadataStore) bool {
if strings.TrimLeft(path, "/") != "config.json" {
return false
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(Config(remote, proxyBaseURL))
return true
}
// Config builds the sparse registry config.json pointing downloads at proxyBaseURL.
func Config(remote models.Remote, proxyBaseURL string) map[string]string {
return map[string]string{
"dl": strings.TrimRight(proxyBaseURL, "/") + "/api/v1/remote/" + remote.Name + "/" + crateDir + "{crate}/{crate}-{version}.crate",
}
}
func isCrate(path string) bool {
path = strings.TrimLeft(path, "/")
return strings.HasPrefix(path, crateDir) && strings.HasSuffix(path, ".crate")
}
+75
View File
@@ -0,0 +1,75 @@
package cargo
import (
"encoding/json"
"net/http/httptest"
"testing"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
func TestClassify(t *testing.T) {
p := &Provider{}
tests := []struct {
path string
want provider.Mutability
}{
{"crates/serde/serde-1.0.200.crate", provider.Immutable},
{"se/rd/serde", provider.Mutable},
{"1/a", provider.Mutable},
{"3/a/anyhow", provider.Mutable},
{"config.json", provider.Mutable},
{"cr/at/crates", 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{}
cratesIO := models.Remote{BaseURL: "https://index.crates.io/"}
mirror := models.Remote{BaseURL: "http://mirror.example"}
tests := []struct {
remote models.Remote
path, want string
}{
{cratesIO, "se/rd/serde", "https://index.crates.io/se/rd/serde"},
{cratesIO, "crates/serde/serde-1.0.200.crate", "https://static.crates.io/crates/serde/serde-1.0.200.crate"},
{mirror, "crates/serde/serde-1.0.200.crate", "http://mirror.example/crates/serde/serde-1.0.200.crate"},
{mirror, "/2/ab", "http://mirror.example/2/ab"},
}
for _, tt := range tests {
if got := p.UpstreamURL(tt.remote, tt.path); got != tt.want {
t.Errorf("UpstreamURL(%q, %q) = %q, want %q", tt.remote.BaseURL, tt.path, got, tt.want)
}
}
}
func TestServeRemoteConfig(t *testing.T) {
p := &Provider{}
remote := models.Remote{Name: "crates-io", BaseURL: "https://index.crates.io"}
w := httptest.NewRecorder()
if !p.ServeRemote(w, httptest.NewRequest("GET", "/", nil), remote, "config.json", "https://aa.example/", nil) {
t.Fatal("config.json not served")
}
var cfg map[string]string
if err := json.Unmarshal(w.Body.Bytes(), &cfg); err != nil {
t.Fatal(err)
}
want := "https://aa.example/api/v1/remote/crates-io/crates/{crate}/{crate}-{version}.crate"
if cfg["dl"] != want {
t.Errorf("dl = %q, want %q", cfg["dl"], want)
}
if _, ok := cfg["api"]; ok {
t.Error("api must be omitted: publishing through the proxy is unsupported")
}
if p.ServeRemote(httptest.NewRecorder(), httptest.NewRequest("GET", "/", nil), remote, "se/rd/serde", "https://aa.example", nil) {
t.Error("index files must fall through to the proxy engine")
}
}
+71
View File
@@ -16,6 +16,8 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/jackc/pgx/v5"
"git.unkin.net/unkin/artifactapi/internal/cache" "git.unkin.net/unkin/artifactapi/internal/cache"
"git.unkin.net/unkin/artifactapi/internal/database" "git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/internal/provider" "git.unkin.net/unkin/artifactapi/internal/provider"
@@ -47,6 +49,8 @@ type Engine struct {
// mirror strategy to prefer the mirror currently handling the fewest // mirror strategy to prefer the mirror currently handling the fewest
// requests. Per-replica and approximate, which is fine. // requests. Per-replica and approximate, which is fine.
inflight sync.Map inflight sync.Map
// evictLockWait bounds how long Evict waits on a held fetch lock.
evictLockWait time.Duration
} }
func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine { func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine {
@@ -57,6 +61,8 @@ func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine {
cas: storage.NewCAS(s), cas: storage.NewCAS(s),
circuit: NewCircuitBreaker(c), circuit: NewCircuitBreaker(c),
accessLog: make(chan database.AccessLogEntry, accessLogBufferSize), accessLog: make(chan database.AccessLogEntry, accessLogBufferSize),
evictLockWait: fetchLockTTL,
} }
go e.runAccessLogWriter() go e.runAccessLogWriter()
return e return e
@@ -208,6 +214,71 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
return result, nil return result, nil
} }
// Evict drops path from every cache layer (artifact row, index object, Redis
// freshness and ETag keys) so the next request refetches from upstream. A
// trailing "/*" evicts every path under that directory.
func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
prefix, wildcard := strings.CutSuffix(path, "*")
if wildcard && !strings.HasSuffix(prefix, "/") {
return &ProxyError{Status: http.StatusBadRequest, Message: "wildcard evict must be <dir>/*"}
}
if _, err := e.db.GetRemote(ctx, remoteName); errors.Is(err, pgx.ErrNoRows) {
return &ProxyError{Status: http.StatusNotFound, Message: fmt.Sprintf("remote %q not found", remoteName)}
} else if err != nil {
return fmt.Errorf("get remote: %w", err)
}
if !wildcard {
if err := e.waitForLock(ctx, remoteName, path); err != nil {
return err
}
defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }()
if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil {
return fmt.Errorf("delete artifact: %w", err)
}
if err := e.store.Delete(ctx, storage.IndexKey(remoteName, path)); err != nil {
return fmt.Errorf("delete index: %w", err)
}
return e.cache.ForgetPath(ctx, remoteName, path)
}
// ponytail: no lock for wildcards; a Fetch already in flight under the
// prefix can re-cache its path after the evict. Per-path locks over a
// directory would close it if that ever matters.
if err := e.db.DeleteArtifactsByPrefix(ctx, remoteName, prefix); err != nil {
return fmt.Errorf("delete artifacts: %w", err)
}
if err := e.store.DeletePrefix(ctx, storage.IndexKey(remoteName, prefix)); err != nil {
return fmt.Errorf("delete indexes: %w", err)
}
return e.cache.ForgetPrefix(ctx, remoteName, prefix)
}
// waitForLock takes the per-path fetch lock so an in-flight Fetch cannot
// re-set TTL/ETag keys after an evict. It fails with a 503 when the lock
// cannot be taken within evictLockWait or Redis errors.
func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) error {
deadline := time.Now().Add(e.evictLockWait)
for {
ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL)
if ok {
return nil
}
if ctx.Err() != nil {
return ctx.Err()
}
if err != nil {
return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)}
}
if time.Now().After(deadline) {
return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"}
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(50 * time.Millisecond):
}
}
}
// HeadResult carries artifact metadata for a HEAD request. There is no body. // HeadResult carries artifact metadata for a HEAD request. There is no body.
type HeadResult struct { type HeadResult struct {
ContentType string ContentType string
+271
View File
@@ -0,0 +1,271 @@
package proxy
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
_ "git.unkin.net/unkin/artifactapi/internal/provider/rpm"
"git.unkin.net/unkin/artifactapi/internal/storage"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// changingUpstream serves every path with the current revision, as a mirror
// does after a sync replaces its repodata. It answers 304 to a matching
// If-None-Match and counts conditional requests.
func changingUpstream(t *testing.T) (*httptest.Server, *atomic.Value, *atomic.Int32) {
t.Helper()
var rev atomic.Value
var conditional atomic.Int32
rev.Store("rev1")
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
v := rev.Load().(string)
etag := `"` + v + `"`
w.Header().Set("ETag", etag)
if inm := r.Header.Get("If-None-Match"); inm != "" {
conditional.Add(1)
if inm == etag {
w.WriteHeader(http.StatusNotModified)
return
}
}
_, _ = w.Write([]byte(v + ":" + r.URL.Path))
}))
t.Cleanup(srv.Close)
return srv, &rev, &conditional
}
func fetchBody(t *testing.T, r models.Remote, path string) string {
t.Helper()
res, err := testEngine.Fetch(context.Background(), r, path, prov(t, models.PackageRPM))
if err != nil {
t.Fatalf("fetch %s: %v", path, err)
}
return readAll(t, res)
}
func rpmRemote(t *testing.T, name, baseURL string) models.Remote {
return seed(t, models.Remote{Name: name, PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: baseURL, MutableTTL: 7200, CheckMutable: true})
}
// cached reports which cache layers hold path: artifact row, index object,
// Redis TTL key, Redis ETag key.
type cached struct{ row, index, ttl, etag bool }
// Mutable indexes live in the S3 index; immutable blobs get an artifact row.
var (
indexCached = cached{index: true, ttl: true, etag: true}
blobCached = cached{row: true, ttl: true, etag: true}
)
func layers(t *testing.T, remote, path string) cached {
t.Helper()
ctx := context.Background()
var c cached
_, err := testDB.GetArtifact(ctx, remote, path)
c.row = err == nil
c.index, err = testEngine.store.Exists(ctx, storage.IndexKey(remote, path))
if err != nil {
t.Fatalf("stat index %s: %v", path, err)
}
c.ttl, _ = testCache.CheckTTL(ctx, remote, path)
etag, _ := testCache.GetETag(ctx, remote, path)
c.etag = etag != ""
return c
}
func TestEvictMutableIndexRefetches(t *testing.T) {
requireStack(t)
srv, rev, conditional := changingUpstream(t)
r := rpmRemote(t, "evict-idx", srv.URL)
const path = "8/Everything/x86_64/repodata/repomd.xml"
if got := fetchBody(t, r, path); got != "rev1:/"+path {
t.Fatalf("initial fetch = %q", got)
}
if c := layers(t, r.Name, path); c != indexCached {
t.Fatalf("before evict = %+v, want %+v", c, indexCached)
}
rev.Store("rev2")
if got := fetchBody(t, r, path); got != "rev1:/"+path {
t.Fatalf("within TTL = %q, want cached rev1", got)
}
if err := testEngine.Evict(context.Background(), r.Name, path); err != nil {
t.Fatalf("evict: %v", err)
}
if c := layers(t, r.Name, path); c != (cached{}) {
t.Fatalf("after evict = %+v, want every layer gone", c)
}
conditional.Store(0)
if got := fetchBody(t, r, path); got != "rev2:/"+path {
t.Fatalf("after evict = %q, want rev2", got)
}
if n := conditional.Load(); n != 0 {
t.Errorf("post-evict fetch revalidated with the evicted ETag (%d conditional requests)", n)
}
}
func TestEvictImmutableBlobDropsRow(t *testing.T) {
requireStack(t)
srv, _, _ := changingUpstream(t)
r := rpmRemote(t, "evict-blob", srv.URL)
const path = "8/Everything/x86_64/Packages/a/a-1.0-1.el8.x86_64.rpm"
fetchBody(t, r, path)
if c := layers(t, r.Name, path); c != blobCached {
t.Fatalf("before evict = %+v, want %+v", c, blobCached)
}
if err := testEngine.Evict(context.Background(), r.Name, path); err != nil {
t.Fatalf("evict: %v", err)
}
if c := layers(t, r.Name, path); c != (cached{}) {
t.Fatalf("after evict = %+v, want every layer gone", c)
}
}
func TestEvictRejectsNonDirectoryWildcard(t *testing.T) {
requireStack(t)
srv, _, _ := changingUpstream(t)
r := rpmRemote(t, "evict-bare", srv.URL)
const path = "8/Everything/x86_64/repodata/repomd.xml"
fetchBody(t, r, path)
for _, bad := range []string{"*", "8*", "8/Every*"} {
var pe *ProxyError
if err := testEngine.Evict(context.Background(), r.Name, bad); !errors.As(err, &pe) || pe.Status != http.StatusBadRequest {
t.Errorf("evict %s = %v, want 400", bad, err)
}
}
if c := layers(t, r.Name, path); c != (indexCached) {
t.Errorf("rejected wildcard evicted layers: %+v", c)
}
}
func TestEvictUnknownRemote(t *testing.T) {
requireStack(t)
var pe *ProxyError
if err := testEngine.Evict(context.Background(), "evict-no-such-remote", "a/b"); !errors.As(err, &pe) || pe.Status != http.StatusNotFound {
t.Fatalf("evict unknown remote = %v, want 404", err)
}
}
func holdLock(t *testing.T, remote, path string) {
t.Helper()
ctx := context.Background()
if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil {
t.Fatalf("acquire: %v %v", ok, err)
}
t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) })
}
func TestEvictWaitsForFetchLock(t *testing.T) {
requireStack(t)
srv, _, _ := changingUpstream(t)
r := rpmRemote(t, "evict-lock", srv.URL)
ctx := context.Background()
const path = "8/Everything/x86_64/repodata/repomd.xml"
holdLock(t, r.Name, path)
done := make(chan error, 1)
go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
select {
case err := <-done:
t.Fatalf("evict returned while fetch lock held: %v", err)
case <-time.After(200 * time.Millisecond):
}
_ = testCache.ReleaseLock(ctx, r.Name, path)
if err := <-done; err != nil {
t.Fatalf("evict: %v", err)
}
ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
if !ok || err != nil {
t.Fatalf("lock still held after evict: %v %v", ok, err)
}
}
func TestEvictCancelledWhileWaitingDeletesNothing(t *testing.T) {
requireStack(t)
srv, _, _ := changingUpstream(t)
r := rpmRemote(t, "evict-cancel", srv.URL)
const path = "8/Everything/x86_64/repodata/repomd.xml"
fetchBody(t, r, path)
holdLock(t, r.Name, path)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
time.Sleep(100 * time.Millisecond)
cancel()
select {
case err := <-done:
if err == nil {
t.Fatal("cancelled evict returned nil")
}
case <-time.After(2 * time.Second):
t.Fatal("cancelled evict did not return")
}
if c := layers(t, r.Name, path); c != (indexCached) {
t.Errorf("cancelled evict deleted layers: %+v", c)
}
}
func TestEvictLockTimeoutIs503(t *testing.T) {
requireStack(t)
srv, _, _ := changingUpstream(t)
r := rpmRemote(t, "evict-timeout", srv.URL)
const path = "8/Everything/x86_64/repodata/repomd.xml"
fetchBody(t, r, path)
holdLock(t, r.Name, path)
prev := testEngine.evictLockWait
testEngine.evictLockWait = 100 * time.Millisecond
t.Cleanup(func() { testEngine.evictLockWait = prev })
var pe *ProxyError
if err := testEngine.Evict(context.Background(), r.Name, path); !errors.As(err, &pe) || pe.Status != http.StatusServiceUnavailable {
t.Fatalf("evict = %v, want 503", err)
}
if c := layers(t, r.Name, path); c != (indexCached) {
t.Errorf("timed-out evict deleted layers: %+v", c)
}
}
func TestEvictWildcardClearsPrefixOnly(t *testing.T) {
requireStack(t)
srv, rev, _ := changingUpstream(t)
r := rpmRemote(t, "evict-wild", srv.URL)
const (
repomd = "8/Everything/x86_64/repodata/repomd.xml"
rpm = "8/Everything/x86_64/Packages/a/a-1.0-1.el8.x86_64.rpm"
other = "9/Everything/x86_64/repodata/repomd.xml"
)
for _, p := range []string{repomd, rpm, other} {
fetchBody(t, r, p)
}
if c := layers(t, r.Name, rpm); c != blobCached {
t.Fatalf("%s before evict = %+v, want %+v", rpm, c, blobCached)
}
rev.Store("rev2")
if err := testEngine.Evict(context.Background(), r.Name, "8/Everything/x86_64/*"); err != nil {
t.Fatalf("evict: %v", err)
}
for _, p := range []string{repomd, rpm} {
if c := layers(t, r.Name, p); c != (cached{}) {
t.Errorf("%s after evict = %+v, want every layer gone", p, c)
}
}
if c := layers(t, r.Name, other); c != (indexCached) {
t.Errorf("%s outside prefix = %+v, want %+v", other, c, indexCached)
}
if got := fetchBody(t, r, repomd); got != "rev2:/"+repomd {
t.Errorf("index under prefix = %q, want rev2", got)
}
if got := fetchBody(t, r, rpm); got != "rev2:/"+rpm {
t.Errorf("artifact under prefix = %q, want rev2", got)
}
if got := fetchBody(t, r, other); got != "rev1:/"+other {
t.Errorf("path outside prefix = %q, want cached rev1", got)
}
}
+3 -2
View File
@@ -21,6 +21,7 @@ import (
"git.unkin.net/unkin/artifactapi/internal/gc" "git.unkin.net/unkin/artifactapi/internal/gc"
"git.unkin.net/unkin/artifactapi/internal/githubauth" "git.unkin.net/unkin/artifactapi/internal/githubauth"
"git.unkin.net/unkin/artifactapi/internal/provider/alpine" "git.unkin.net/unkin/artifactapi/internal/provider/alpine"
_ "git.unkin.net/unkin/artifactapi/internal/provider/cargo"
"git.unkin.net/unkin/artifactapi/internal/provider/deb" "git.unkin.net/unkin/artifactapi/internal/provider/deb"
_ "git.unkin.net/unkin/artifactapi/internal/provider/docker" _ "git.unkin.net/unkin/artifactapi/internal/provider/docker"
_ "git.unkin.net/unkin/artifactapi/internal/provider/generic" _ "git.unkin.net/unkin/artifactapi/internal/provider/generic"
@@ -196,8 +197,8 @@ func (s *Server) routes() chi.Router {
r.Route("/remotes/{name}/objects", func(r chi.Router) { r.Route("/remotes/{name}/objects", func(r chi.Router) {
objHandler := v2.NewObjectsHandler(s.db) objHandler := v2.NewObjectsHandler(s.db)
r.Get("/", objHandler.Routes().ServeHTTP) r.Get("/", objHandler.Routes(s.engine).ServeHTTP)
r.Delete("/*", objHandler.Routes().ServeHTTP) r.Delete("/*", objHandler.Routes(s.engine).ServeHTTP)
}) })
r.Route("/locals/{name}/objects", func(r chi.Router) { r.Route("/locals/{name}/objects", func(r chi.Router) {
+24
View File
@@ -99,6 +99,30 @@ func (s *S3) Stat(ctx context.Context, key string) (*minio.ObjectInfo, error) {
return &info, nil return &info, nil
} }
// DeletePrefix removes every object whose key starts with prefix. It lists
// first so a failed list returns its error instead of deleting nothing.
func (s *S3) DeletePrefix(ctx context.Context, prefix string) error {
var objs []minio.ObjectInfo
for obj := range s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) {
if obj.Err != nil {
return obj.Err
}
objs = append(objs, obj)
}
ch := make(chan minio.ObjectInfo, len(objs))
for _, obj := range objs {
ch <- obj
}
close(ch)
var err error
for res := range s.client.RemoveObjects(ctx, s.bucket, ch, minio.RemoveObjectsOptions{}) {
if res.Err != nil && err == nil {
err = res.Err
}
}
return err
}
// ListStaleObjects returns keys under prefix last modified before cutoff. Used // ListStaleObjects returns keys under prefix last modified before cutoff. Used
// by the GC to reap abandoned staging objects (e.g. cancelled docker pushes). // by the GC to reap abandoned staging objects (e.g. cancelled docker pushes).
func (s *S3) ListStaleObjects(ctx context.Context, prefix string, cutoff time.Time) ([]string, error) { func (s *S3) ListStaleObjects(ctx context.Context, prefix string, cutoff time.Time) ([]string, error) {
+27
View File
@@ -4,11 +4,15 @@ import (
"bytes" "bytes"
"context" "context"
"io" "io"
"net/http"
"net/http/httptest"
"os" "os"
"strings" "strings"
"testing" "testing"
"time" "time"
"github.com/minio/minio-go/v7"
"git.unkin.net/unkin/artifactapi/internal/testsupport" "git.unkin.net/unkin/artifactapi/internal/testsupport"
) )
@@ -158,3 +162,26 @@ func TestCASStore(t *testing.T) {
t.Errorf("stored content mismatch: %q", got) t.Errorf("stored content mismatch: %q", got)
} }
} }
// The fake endpoint fails every list but accepts every delete, as an S3 that
// tolerates deleting an empty key would.
func TestDeletePrefixReturnsListError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodPost {
_, _ = io.WriteString(w, `<DeleteResult></DeleteResult>`)
return
}
w.WriteHeader(http.StatusInternalServerError)
_, _ = io.WriteString(w, `<Error><Code>InternalError</Code><Message>list failed</Message></Error>`)
}))
defer srv.Close()
client, err := minio.New(strings.TrimPrefix(srv.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1})
if err != nil {
t.Fatal(err)
}
s := &S3{client: client, bucket: "bucket"}
err = s.DeletePrefix(context.Background(), "indexes/r/")
if err == nil {
t.Fatal("DeletePrefix on a failing list returned nil")
}
}
+262 -1
View File
@@ -2,27 +2,58 @@ package virtual
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"io" "io"
"log/slog" "log/slog"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"strings"
"sync" "sync"
"time"
"git.unkin.net/unkin/artifactapi/internal/database" "git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/internal/provider" "git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/internal/proxy" "git.unkin.net/unkin/artifactapi/internal/proxy"
"git.unkin.net/unkin/artifactapi/pkg/models" "git.unkin.net/unkin/artifactapi/pkg/models"
"golang.org/x/sync/singleflight"
) )
type Engine struct { type Engine struct {
db *database.DB db *database.DB
proxyEngine *proxy.Engine proxyEngine *proxy.Engine
getRemote func(context.Context, string) (*models.Remote, error)
rpmMember func(context.Context, string) (*RPMMember, error)
mergeTTL time.Duration
sf singleflight.Group
mu sync.Mutex
rpm map[string]*rpmGen
} }
// 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
}
const rpmMergeTTL = 60 * time.Second
func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine { func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
return &Engine{db: db, proxyEngine: proxyEngine} e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL}
e.rpmMember = e.fetchRPMMember
return e
} }
func (e *Engine) Fetch(ctx context.Context, virt models.Virtual, path string, proxyBaseURL string) ([]byte, string, error) { func (e *Engine) Fetch(ctx context.Context, virt models.Virtual, path string, proxyBaseURL string) ([]byte, string, error) {
if virt.PackageType == models.PackageRPM {
return e.fetchRPM(ctx, virt, path)
}
merger, err := GetMerger(virt.PackageType) merger, err := GetMerger(virt.PackageType)
if err != nil { if err != nil {
return nil, "", fmt.Errorf("unsupported virtual type %q: %w", virt.PackageType, err) return nil, "", fmt.Errorf("unsupported virtual type %q: %w", virt.PackageType, err)
@@ -133,3 +164,233 @@ func (e *Engine) fetchLocalIndex(ctx context.Context, remote models.Remote, path
return indexer.GenerateLocalIndex(ctx, e.db, remote.Name, path) return indexer.GenerateLocalIndex(ctx, e.db, remote.Name, path)
} }
var (
ErrNotFound = errors.New("not found")
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.
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/") {
return "", false, nil
}
name, rest, found := strings.Cut(path, "/")
if !found || !slices.Contains(virt.Members, name) {
return "", false, nil
}
segs, err := memberSegments(rest)
if err != nil {
return "", false, err
}
remote, err := e.getRemote(ctx, name)
if err != nil {
return "", false, nil
}
loc := fmt.Sprintf("%s/api/v1/%s/%s/%s", strings.TrimRight(proxyBaseURL, "/"), remote.RepoType, url.PathEscape(name), strings.Join(segs, "/"))
if rawQuery != "" {
loc += "?" + rawQuery
}
return loc, true, nil
}
// memberSegments decodes rest and returns its path-escaped segments, rejecting
// absolute paths and any "." or ".." segment.
func memberSegments(rest string) ([]string, error) {
dec, err := url.PathUnescape(rest)
if err != nil || dec == "" || strings.HasPrefix(dec, "/") {
return nil, ErrBadPath
}
segs := strings.Split(dec, "/")
for i, s := range segs {
if s == "." || s == ".." || strings.Contains(s, "\\") {
return nil, ErrBadPath
}
segs[i] = url.PathEscape(s)
}
return segs, nil
}
func (e *Engine) fetchRPM(ctx context.Context, virt models.Virtual, path string) ([]byte, string, error) {
if !strings.HasPrefix(path, "repodata/") {
return nil, "", ErrNotFound
}
if path != "repodata/repomd.xml" {
if body, ok := e.cachedRPMFile(virt.Name, path); ok {
return body, "application/gzip", nil
}
}
repo, err := e.mergedRPM(ctx, virt)
if err != nil {
return nil, "", err
}
if path == "repodata/repomd.xml" {
return repo.Repomd, "application/xml", nil
}
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.
// 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) {
e.mu.Lock()
g := e.rpm[virt.Name]
if g != nil && time.Since(g.at) < e.mergeTTL {
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)
if err != nil {
return nil, err
}
e.mu.Lock()
defer e.mu.Unlock()
if e.rpm == nil {
e.rpm = map[string]*rpmGen{}
}
g := e.rpm[virt.Name]
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()
repo = g.cur
default:
g.prev, g.cur, g.at = g.cur, repo, time.Now()
}
return repo, nil
})
if err != nil {
return nil, err
}
return v.(*RPMRepo), nil
}
func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
members := make([]RPMMember, 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)
if err != nil {
errs[i] = fmt.Errorf("member %q: %w", name, err)
return
}
members[i] = *m
}()
}
wg.Wait()
// %v, not %w: a member failure is a 502 whatever its cause, never a 404.
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
}
func (e *Engine) cachedRPMFile(virt, path string) ([]byte, bool) {
e.mu.Lock()
defer e.mu.Unlock()
g := e.rpm[virt]
if g == nil {
return nil, false
}
for _, r := range []*RPMRepo{g.cur, g.prev} {
if r == nil {
continue
}
if body, ok := r.Files[path]; ok {
return body, true
}
}
return nil, false
}
func (e *Engine) fetchRPMMember(ctx context.Context, name string) (*RPMMember, error) {
remote, err := e.getRemote(ctx, name)
if err != nil {
return nil, fmt.Errorf("remote %q: %w", name, err)
}
repomd, err := e.fetchMemberPath(ctx, *remote, "repodata/repomd.xml")
if err != nil {
return nil, err
}
locs, ts, err := parseRepomd(repomd)
if err != nil {
return nil, err
}
m := &RPMMember{RemoteName: name, Bases: remote.UpstreamPool(), Timestamp: ts, Data: map[string][]byte{}}
for t, href := range locs {
raw, err := e.fetchMemberPath(ctx, *remote, href)
if err != nil {
return nil, err
}
if m.Data[t], err = decompress(href, raw); err != nil {
return nil, fmt.Errorf("decompress %s: %w", href, err)
}
}
return m, nil
}
// fetchMemberPath reads one path from a member the same way its own route
// would serve it: local index, synthesized remote (e.g. github_rpm), or proxy.
func (e *Engine) fetchMemberPath(ctx context.Context, remote models.Remote, path string) ([]byte, error) {
prov, err := provider.Get(remote.PackageType)
if err != nil {
return nil, fmt.Errorf("provider %q: %w", remote.PackageType, err)
}
var serve func(w http.ResponseWriter, r *http.Request) bool
if remote.RepoType == models.RepoTypeLocal {
indexer, ok := prov.(provider.LocalIndexer)
if !ok {
return nil, fmt.Errorf("provider %q does not serve a local index", remote.PackageType)
}
serve = func(w http.ResponseWriter, r *http.Request) bool {
return indexer.ServeLocalIndex(w, r, e.db, remote.Name, path)
}
} else if rs, ok := prov.(provider.RemoteServer); ok {
serve = func(w http.ResponseWriter, r *http.Request) bool {
return rs.ServeRemote(w, r, remote, path, "", e.db)
}
}
if serve != nil {
rec := httptest.NewRecorder()
if serve(rec, httptest.NewRequestWithContext(ctx, http.MethodGet, "/"+path, nil)) {
if rec.Code != http.StatusOK {
return nil, fmt.Errorf("%s/%s: status %d", remote.Name, path, rec.Code)
}
return rec.Body.Bytes(), nil
}
if remote.RepoType == models.RepoTypeLocal {
return nil, fmt.Errorf("%s/%s: %w", remote.Name, path, ErrNotFound)
}
}
res, err := e.proxyEngine.Fetch(ctx, remote, path, prov)
if err != nil {
return nil, fmt.Errorf("fetch %s/%s: %w", remote.Name, path, err)
}
defer res.Reader.Close()
return io.ReadAll(res.Reader)
}
+158
View File
@@ -0,0 +1,158 @@
package virtual
import (
"context"
"errors"
"fmt"
"regexp"
"sync"
"sync/atomic"
"testing"
"time"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
func fakeEngine(members map[string]*RPMMember) *Engine {
return &Engine{
getRemote: func(_ context.Context, name string) (*models.Remote, error) {
return &models.Remote{Name: name, RepoType: models.RepoTypeRemote}, nil
},
rpmMember: func(_ context.Context, name string) (*RPMMember, error) {
if m := members[name]; m != nil {
return m, nil
}
return nil, errors.New("upstream down")
},
}
}
func rpmVirt(members ...string) models.Virtual {
return models.Virtual{Name: "v", PackageType: models.PackageRPM, Members: members}
}
func TestFetchRPMFailsClosed(t *testing.T) {
e := fakeEngine(map[string]*RPMMember{
"a": {RemoteName: "a", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}},
})
_, _, err := e.Fetch(context.Background(), rpmVirt("a", "b"), "repodata/repomd.xml", "")
if err == nil || errors.Is(err, ErrNotFound) {
t.Fatalf("want upstream error with a member down, got %v", err)
}
}
func TestFetchRPMDataSurvivesMemberChange(t *testing.T) {
m := &RPMMember{RemoteName: "a", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}}
e := fakeEngine(map[string]*RPMMember{"a": m})
virt := rpmVirt("a")
repomd, _, err := e.Fetch(context.Background(), virt, "repodata/repomd.xml", "")
if err != nil {
t.Fatal(err)
}
m.Data = map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "2", "bbb", "foo2.rpm"))}
href := regexp.MustCompile(`repodata/[0-9a-f]+-primary\.xml\.gz`).Find(repomd)
body, _, err := e.Fetch(context.Background(), virt, string(href), "")
if err != nil {
t.Fatalf("data from previous repomd must stay resolvable: %v", err)
}
if got := string(gunzip(t, body)); !regexp.MustCompile(`ver="1"`).MatchString(got) {
t.Fatalf("served wrong primary:\n%s", got)
}
newMD, _, err := e.Fetch(context.Background(), virt, "repodata/repomd.xml", "")
if err != nil || string(newMD) == string(repomd) {
t.Fatalf("repomd should reflect the member change (err %v)", err)
}
if _, _, err := e.Fetch(context.Background(), virt, string(href), ""); err != nil {
t.Fatalf("previous generation must survive one re-merge: %v", err)
}
}
func TestFetchRPMLocalMemberWithoutRepodataIsUpstreamError(t *testing.T) {
e := fakeEngine(map[string]*RPMMember{
"a": {RemoteName: "a", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}},
})
e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) {
if name == "local" {
return nil, fmt.Errorf("local/repodata/repomd.xml: %w", ErrNotFound)
}
return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}}, nil
}
_, _, err := e.Fetch(context.Background(), rpmVirt("a", "local"), "repodata/repomd.xml", "")
if err == nil || errors.Is(err, ErrNotFound) {
t.Fatalf("member failure must not surface as not-found, got %v", err)
}
}
func TestFetchRPMMergeIsShared(t *testing.T) {
var calls atomic.Int32
e := fakeEngine(nil)
e.mergeTTL = time.Minute
e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) {
calls.Add(1)
time.Sleep(20 * time.Millisecond)
return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}}, nil
}
var wg sync.WaitGroup
for range 10 {
wg.Add(1)
go func() {
defer wg.Done()
if _, _, err := e.Fetch(context.Background(), rpmVirt("a"), "repodata/repomd.xml", ""); err != nil {
t.Error(err)
}
}()
}
wg.Wait()
if _, _, err := e.Fetch(context.Background(), rpmVirt("a"), "repodata/repomd.xml", ""); err != nil {
t.Fatal(err)
}
if n := calls.Load(); n != 1 {
t.Fatalf("member fetched %d times, want 1", n)
}
}
func TestFetchRPMDataScopedToVirtual(t *testing.T) {
e := fakeEngine(map[string]*RPMMember{
"a": {RemoteName: "a", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", "aaa", "foo.rpm"))}},
"b": {RemoteName: "b", Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("bar", "1", "bbb", "bar.rpm"))}},
})
virtA := models.Virtual{Name: "va", PackageType: models.PackageRPM, Members: []string{"a"}}
virtB := models.Virtual{Name: "vb", PackageType: models.PackageRPM, Members: []string{"b"}}
repomd, _, err := e.Fetch(context.Background(), virtA, "repodata/repomd.xml", "")
if err != nil {
t.Fatal(err)
}
href := regexp.MustCompile(`repodata/[0-9a-f]+-primary\.xml\.gz`).Find(repomd)
if _, _, err := e.Fetch(context.Background(), virtB, string(href), ""); !errors.Is(err, ErrNotFound) {
t.Fatalf("virtual vb served va's data file (err %v)", err)
}
}
func TestMemberRedirect(t *testing.T) {
e := fakeEngine(nil)
virt := rpmVirt("gh", "other")
for _, tc := range []struct {
path, query, want string
ok bool
err error
}{
{path: "gh/o/r/releases/download/v1/c++-1.rpm", query: "x=1", ok: true,
want: "http://h/api/v1/remote/gh/o/r/releases/download/v1/c++-1.rpm?x=1"},
{path: "gh/a%20b/c%3Fd.rpm", ok: true, want: "http://h/api/v1/remote/gh/a%20b/c%3Fd.rpm"},
{path: "gh/../../remote/other/x", err: ErrBadPath},
{path: "gh/%2e%2e/%2e%2e/remote/other/x", err: ErrBadPath},
{path: "gh/a/./b", err: ErrBadPath},
{path: "gh//etc/passwd", err: ErrBadPath},
{path: "repodata/repomd.xml"},
{path: "nope/x.rpm"},
} {
got, ok, err := e.MemberRedirect(context.Background(), virt, tc.path, tc.query, "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)
}
}
}
+298
View File
@@ -0,0 +1,298 @@
package virtual
import (
"bytes"
"compress/bzip2"
"compress/gzip"
"crypto/sha256"
"encoding/hex"
"encoding/xml"
"errors"
"fmt"
"io"
"regexp"
"strings"
"github.com/klauspost/compress/zstd"
"github.com/ulikunitz/xz"
)
var rpmDataTypes = []string{"primary", "filelists", "other"}
var rpmRoots = map[string]string{
"primary": `<metadata xmlns="http://linux.duke.edu/metadata/common" xmlns:rpm="http://linux.duke.edu/metadata/rpm" packages="%d">`,
"filelists": `<filelists xmlns="http://linux.duke.edu/metadata/filelists" packages="%d">`,
"other": `<otherdata xmlns="http://linux.duke.edu/metadata/other" packages="%d">`,
}
var rpmRootClose = map[string]string{"primary": "</metadata>", "filelists": "</filelists>", "other": "</otherdata>"}
var locationRe = regexp.MustCompile(`<location\b[^>]*/>`)
// RPMMember is one member repo's decompressed primary/filelists/other XML,
// keyed by repomd data type. Missing types contribute no packages.
type RPMMember struct {
RemoteName string
Bases []string
Timestamp int64
Data map[string][]byte
}
// RPMRepo is a merged yum repo: repomd.xml plus the files it references, keyed
// by their path under the repo root.
type RPMRepo struct {
Repomd []byte
Files map[string][]byte
}
type repomdDoc struct {
Revision string `xml:"revision"`
Data []repomdData `xml:"data"`
}
type repomdData struct {
Type string `xml:"type,attr"`
Location struct {
Href string `xml:"href,attr"`
} `xml:"location"`
Timestamp int64 `xml:"timestamp"`
}
type rpmPkgVersion struct {
Epoch string `xml:"epoch,attr"`
Ver string `xml:"ver,attr"`
Rel string `xml:"rel,attr"`
}
type primaryPkg struct {
Name string `xml:"name"`
Arch string `xml:"arch"`
Version rpmPkgVersion `xml:"version"`
Checksum string `xml:"checksum"`
Location struct {
Href string `xml:"href,attr"`
Base string `xml:"http://www.w3.org/XML/1998/namespace base,attr"`
} `xml:"location"`
}
type pkgidPkg struct {
PkgID string `xml:"pkgid,attr"`
}
// parseRepomd returns the member's primary/filelists/other locations and its
// newest data timestamp.
func parseRepomd(body []byte) (map[string]string, int64, error) {
var doc repomdDoc
if err := xml.Unmarshal(body, &doc); err != nil {
return nil, 0, fmt.Errorf("parse repomd.xml: %w", err)
}
locs := map[string]string{}
var ts int64
for _, d := range doc.Data {
for _, t := range rpmDataTypes {
if d.Type == t && d.Location.Href != "" {
locs[t] = d.Location.Href
ts = max(ts, d.Timestamp)
}
}
}
if locs["primary"] == "" {
return nil, 0, errors.New("repomd.xml has no primary data")
}
return locs, ts, nil
}
func decompress(href string, body []byte) ([]byte, error) {
var r io.Reader
switch {
case strings.HasSuffix(href, ".gz"):
gz, err := gzip.NewReader(bytes.NewReader(body))
if err != nil {
return nil, err
}
r = gz
case strings.HasSuffix(href, ".xz"):
x, err := xz.NewReader(bytes.NewReader(body))
if err != nil {
return nil, err
}
r = x
case strings.HasSuffix(href, ".zst"):
z, err := zstd.NewReader(bytes.NewReader(body))
if err != nil {
return nil, err
}
defer z.Close()
r = z
case strings.HasSuffix(href, ".bz2"):
r = bzip2.NewReader(bytes.NewReader(body))
default:
return body, nil
}
return io.ReadAll(r)
}
// splitPackages returns the raw bytes of each top-level <package> element.
func splitPackages(doc []byte) ([][]byte, error) {
d := xml.NewDecoder(bytes.NewReader(doc))
var pkgs [][]byte
depth := 0
for {
start := d.InputOffset()
tok, err := d.Token()
if err == io.EOF {
return pkgs, nil
}
if err != nil {
return nil, err
}
switch t := tok.(type) {
case xml.StartElement:
if depth == 1 && t.Name.Local == "package" {
if err := d.Skip(); err != nil {
return nil, err
}
pkgs = append(pkgs, doc[start:d.InputOffset()])
continue
}
depth++
case xml.EndElement:
depth--
}
}
}
// memberHref returns href relative to the member root. An xml:base under one of
// the member's upstream bases is folded into the href; any other xml:base is a
// host the member can't serve, so ok is false and the location is left as is.
func memberHref(m RPMMember, base, href string) (string, bool) {
href = strings.TrimLeft(href, "/")
if base == "" {
return href, true
}
base = strings.TrimRight(base, "/") + "/"
for _, b := range m.Bases {
b = strings.TrimRight(b, "/") + "/"
if strings.HasPrefix(base, b) {
return strings.TrimPrefix(base, b) + href, true
}
}
return "", false
}
func xmlEscape(s string) string {
var b bytes.Buffer
_ = xml.EscapeText(&b, []byte(s))
return b.String()
}
// MergeRPM merges members into one repo. Members are in priority order: the
// first member to carry a NEVRA wins. Package locations are prefixed with the
// owning member's name so the virtual can route downloads back to it.
func MergeRPM(members []RPMMember) (*RPMRepo, error) {
kept := map[string]bool{}
seen := map[string]bool{}
out := map[string][][]byte{}
for i, m := range members {
pkgs, err := splitPackages(m.Data["primary"])
if err != nil {
return nil, fmt.Errorf("member %q primary: %w", m.RemoteName, err)
}
for _, raw := range pkgs {
var p primaryPkg
if err := xml.Unmarshal(raw, &p); err != nil {
return nil, fmt.Errorf("member %q primary package: %w", m.RemoteName, err)
}
epoch := p.Version.Epoch
if epoch == "" {
epoch = "0"
}
nevra := fmt.Sprintf("%s-%s:%s-%s.%s", p.Name, epoch, p.Version.Ver, p.Version.Rel, p.Arch)
if seen[nevra] {
continue
}
seen[nevra] = true
kept[fmt.Sprintf("%d/%s", i, p.Checksum)] = true
if href, ok := memberHref(m, p.Location.Base, p.Location.Href); ok {
loc := `<location href="` + xmlEscape(m.RemoteName+"/"+href) + `"/>`
raw = locationRe.ReplaceAll(raw, []byte(loc))
}
out["primary"] = append(out["primary"], raw)
}
}
for _, t := range rpmDataTypes[1:] {
emitted := map[string]bool{}
for i, m := range members {
pkgs, err := splitPackages(m.Data[t])
if err != nil {
return nil, fmt.Errorf("member %q %s: %w", m.RemoteName, t, err)
}
for _, raw := range pkgs {
var p pkgidPkg
if err := xml.Unmarshal(raw, &p); err != nil {
return nil, fmt.Errorf("member %q %s package: %w", m.RemoteName, t, err)
}
key := fmt.Sprintf("%d/%s", i, p.PkgID)
if kept[key] && !emitted[key] {
emitted[key] = true
out[t] = append(out[t], raw)
}
}
}
}
var ts int64
for _, m := range members {
ts = max(ts, m.Timestamp)
}
repo := &RPMRepo{Files: map[string][]byte{}}
var md bytes.Buffer
md.WriteString(xml.Header)
md.WriteString(`<repomd xmlns="http://linux.duke.edu/metadata/repo" xmlns:rpm="http://linux.duke.edu/metadata/rpm">` + "\n")
fmt.Fprintf(&md, " <revision>%d</revision>\n", ts)
for _, t := range rpmDataTypes {
var doc bytes.Buffer
doc.WriteString(xml.Header)
fmt.Fprintf(&doc, rpmRoots[t]+"\n", len(out[t]))
for _, raw := range out[t] {
doc.Write(raw)
doc.WriteString("\n")
}
doc.WriteString(rpmRootClose[t] + "\n")
gz := gzipDeterministic(doc.Bytes())
sum := sha256Hex(gz)
href := fmt.Sprintf("repodata/%s-%s.xml.gz", sum, t)
repo.Files[href] = gz
fmt.Fprintf(&md, " <data type=\"%s\">\n", t)
fmt.Fprintf(&md, " <checksum type=\"sha256\">%s</checksum>\n", sum)
fmt.Fprintf(&md, " <open-checksum type=\"sha256\">%s</open-checksum>\n", sha256Hex(doc.Bytes()))
fmt.Fprintf(&md, " <location href=\"%s\"/>\n", href)
fmt.Fprintf(&md, " <timestamp>%d</timestamp>\n", ts)
fmt.Fprintf(&md, " <size>%d</size>\n", len(gz))
fmt.Fprintf(&md, " <open-size>%d</open-size>\n", doc.Len())
md.WriteString(" </data>\n")
}
md.WriteString("</repomd>\n")
repo.Repomd = md.Bytes()
return repo, nil
}
func gzipDeterministic(data []byte) []byte {
var buf bytes.Buffer
gz := gzip.NewWriter(&buf)
gz.Header = gzip.Header{OS: 255}
_, _ = gz.Write(data)
_ = gz.Close()
return buf.Bytes()
}
func sha256Hex(data []byte) string {
h := sha256.Sum256(data)
return hex.EncodeToString(h[:])
}
+274
View File
@@ -0,0 +1,274 @@
package virtual
import (
"bytes"
"compress/gzip"
"crypto/sha256"
"encoding/hex"
"encoding/xml"
"fmt"
"io"
"strings"
"testing"
"github.com/klauspost/compress/zstd"
"github.com/ulikunitz/xz"
)
func primaryXML(pkgs ...string) []byte {
return []byte(`<?xml version="1.0" encoding="UTF-8"?>
<metadata xmlns="http://linux.duke.edu/metadata/common" xmlns:rpm="http://linux.duke.edu/metadata/rpm" packages="` + fmt.Sprint(len(pkgs)) + `">
` + strings.Join(pkgs, "\n") + `
</metadata>`)
}
func primaryPkgXML(name, ver, pkgid, href string) string {
return `<package type="rpm">
<name>` + name + `</name>
<arch>x86_64</arch>
<version epoch="0" ver="` + ver + `" rel="1"/>
<checksum type="sha256" pkgid="YES">` + pkgid + `</checksum>
<location href="` + href + `"/>
<format>
<rpm:provides><rpm:entry name="` + name + `"/></rpm:provides>
</format>
</package>`
}
func filelistsXML(pkgids ...string) []byte {
var b strings.Builder
b.WriteString(`<?xml version="1.0"?><filelists xmlns="http://linux.duke.edu/metadata/filelists" packages="1">`)
for _, id := range pkgids {
b.WriteString(`<package pkgid="` + id + `" name="x" arch="x86_64"><version epoch="0" ver="1" rel="1"/><file>/usr/bin/` + id + `</file></package>`)
}
b.WriteString(`</filelists>`)
return []byte(b.String())
}
func gunzip(t *testing.T, b []byte) []byte {
t.Helper()
r, err := gzip.NewReader(bytes.NewReader(b))
if err != nil {
t.Fatal(err)
}
out, err := io.ReadAll(r)
if err != nil {
t.Fatal(err)
}
return out
}
func mergedFile(t *testing.T, repo *RPMRepo, dtype string) []byte {
t.Helper()
for href, body := range repo.Files {
if strings.HasSuffix(href, "-"+dtype+".xml.gz") {
return gunzip(t, body)
}
}
t.Fatalf("no %s file in merged repo", dtype)
return nil
}
func TestMergeRPMDedupePriority(t *testing.T) {
members := []RPMMember{
{RemoteName: "first", Timestamp: 100, Data: map[string][]byte{
"primary": primaryXML(primaryPkgXML("foo", "1.0", "aaa", "Packages/foo-1.0.rpm")),
"filelists": filelistsXML("aaa"),
}},
{RemoteName: "second", Timestamp: 200, Data: map[string][]byte{
"primary": primaryXML(
primaryPkgXML("foo", "1.0", "bbb", "Packages/foo-1.0-rebuilt.rpm"),
primaryPkgXML("bar", "2.0", "ccc", "Packages/bar-2.0.rpm"),
),
"filelists": filelistsXML("bbb", "ccc"),
}},
}
repo, err := MergeRPM(members)
if err != nil {
t.Fatal(err)
}
primary := string(mergedFile(t, repo, "primary"))
if strings.Count(primary, "<name>foo</name>") != 1 {
t.Fatalf("duplicate NEVRA not deduped:\n%s", primary)
}
if !strings.Contains(primary, `href="first/Packages/foo-1.0.rpm"`) || strings.Contains(primary, "foo-1.0-rebuilt") {
t.Fatalf("first member should win duplicate NEVRA:\n%s", primary)
}
if !strings.Contains(primary, `href="second/Packages/bar-2.0.rpm"`) {
t.Fatalf("unique package from second member missing:\n%s", primary)
}
if !strings.Contains(primary, `packages="2"`) {
t.Fatalf("package count wrong:\n%s", primary)
}
filelists := string(mergedFile(t, repo, "filelists"))
if !strings.Contains(filelists, "/usr/bin/aaa") || !strings.Contains(filelists, "/usr/bin/ccc") || strings.Contains(filelists, "/usr/bin/bbb") {
t.Fatalf("filelists must follow the winning primary entries:\n%s", filelists)
}
other := string(mergedFile(t, repo, "other"))
if !strings.Contains(other, `packages="0"`) {
t.Fatalf("members without other data should yield an empty other.xml:\n%s", other)
}
if !strings.Contains(string(repo.Repomd), "<revision>200</revision>") {
t.Fatalf("revision should be the newest member timestamp:\n%s", repo.Repomd)
}
}
func TestMergeRPMOutputIsWellFormed(t *testing.T) {
repo, err := MergeRPM([]RPMMember{{RemoteName: "a", Data: map[string][]byte{
"primary": primaryXML(primaryPkgXML("foo", "1.0", "aaa", "Packages/foo.rpm")),
}}})
if err != nil {
t.Fatal(err)
}
var doc struct {
XMLName xml.Name
Packages []struct {
Name string `xml:"name"`
Provides []struct {
Name string `xml:"name,attr"`
} `xml:"format>provides>entry"`
} `xml:"package"`
}
if err := xml.Unmarshal(mergedFile(t, repo, "primary"), &doc); err != nil {
t.Fatal(err)
}
if doc.XMLName.Space != "http://linux.duke.edu/metadata/common" || len(doc.Packages) != 1 {
t.Fatalf("unexpected merged primary: %+v", doc)
}
if len(doc.Packages[0].Provides) != 1 || doc.Packages[0].Provides[0].Name != "foo" {
t.Fatalf("rpm: namespaced format lost: %+v", doc.Packages[0])
}
}
func TestMergeRPMHrefRewriting(t *testing.T) {
based := func(name, base string) string {
return `<package type="rpm"><name>` + name + `</name><arch>noarch</arch><version epoch="0" ver="1" rel="1"/>` +
`<checksum type="sha256" pkgid="YES">` + name + `</checksum><location xml:base="` + base + `" href="Packages/` + name + `.rpm"/></package>`
}
repo, err := MergeRPM([]RPMMember{{RemoteName: "gh", Bases: []string{"https://up.example/el9", "https://mirror.example/el9/"}, Data: map[string][]byte{
"primary": primaryXML(
primaryPkgXML("foo", "1.0", "aaa", "storytold/photocraft/releases/download/v1.0/foo&amp;bar.rpm"),
primaryPkgXML("lead", "1.0", "bbb", "/Packages/lead.rpm"),
based("root", "https://up.example/el9/"),
based("sub", "https://mirror.example/el9/extra"),
based("ext", "https://other.example/"),
),
}}})
if err != nil {
t.Fatal(err)
}
primary := string(mergedFile(t, repo, "primary"))
for _, want := range []string{
`<location href="gh/storytold/photocraft/releases/download/v1.0/foo&amp;bar.rpm"/>`,
`<location href="gh/Packages/lead.rpm"/>`,
`<location href="gh/Packages/root.rpm"/>`,
`<location href="gh/extra/Packages/sub.rpm"/>`,
`<location xml:base="https://other.example/" href="Packages/ext.rpm"/>`,
} {
if !strings.Contains(primary, want) {
t.Errorf("missing %s in:\n%s", want, primary)
}
}
}
func TestMergeRPMChecksums(t *testing.T) {
repo, err := MergeRPM([]RPMMember{{RemoteName: "a", Timestamp: 42, Data: map[string][]byte{
"primary": primaryXML(primaryPkgXML("foo", "1.0", "aaa", "Packages/foo.rpm")),
}}})
if err != nil {
t.Fatal(err)
}
var md struct {
Data []struct {
Type string `xml:"type,attr"`
Checksum string `xml:"checksum"`
OpenChecksum string `xml:"open-checksum"`
Location struct {
Href string `xml:"href,attr"`
} `xml:"location"`
Timestamp int64 `xml:"timestamp"`
Size int `xml:"size"`
OpenSize int `xml:"open-size"`
} `xml:"data"`
}
if err := xml.Unmarshal(repo.Repomd, &md); err != nil {
t.Fatal(err)
}
if len(md.Data) != 3 {
t.Fatalf("want primary/filelists/other, got %d entries", len(md.Data))
}
for _, d := range md.Data {
body, ok := repo.Files[d.Location.Href]
if !ok {
t.Fatalf("%s: location %q not served", d.Type, d.Location.Href)
}
sum := sha256.Sum256(body)
if hex.EncodeToString(sum[:]) != d.Checksum || len(body) != d.Size {
t.Errorf("%s: checksum/size do not match served bytes", d.Type)
}
open := gunzip(t, body)
osum := sha256.Sum256(open)
if hex.EncodeToString(osum[:]) != d.OpenChecksum || len(open) != d.OpenSize {
t.Errorf("%s: open-checksum/open-size do not match decompressed bytes", d.Type)
}
if d.Timestamp != 42 {
t.Errorf("%s: timestamp %d, want 42", d.Type, d.Timestamp)
}
}
again, _ := MergeRPM([]RPMMember{{RemoteName: "a", Timestamp: 42, Data: map[string][]byte{
"primary": primaryXML(primaryPkgXML("foo", "1.0", "aaa", "Packages/foo.rpm")),
}}})
if !bytes.Equal(repo.Repomd, again.Repomd) {
t.Error("merge must be deterministic so repomd checksums stay valid across requests")
}
}
func TestParseRepomd(t *testing.T) {
locs, ts, err := parseRepomd([]byte(`<repomd xmlns="http://linux.duke.edu/metadata/repo">
<data type="primary"><location href="repodata/p-primary.xml.zst"/><timestamp>10</timestamp></data>
<data type="primary_db"><location href="repodata/p.sqlite.bz2"/><timestamp>99</timestamp></data>
<data type="other"><location href="repodata/o-other.xml.gz"/><timestamp>20</timestamp></data>
</repomd>`))
if err != nil {
t.Fatal(err)
}
if locs["primary"] != "repodata/p-primary.xml.zst" || locs["other"] != "repodata/o-other.xml.gz" || len(locs) != 2 {
t.Fatalf("unexpected locations: %v", locs)
}
if ts != 20 {
t.Fatalf("timestamp %d, want 20 (sqlite entries ignored)", ts)
}
if _, _, err := parseRepomd([]byte(`<repomd><data type="other"><location href="x"/></data></repomd>`)); err == nil {
t.Fatal("repomd without primary should error")
}
}
func TestDecompress(t *testing.T) {
plain := []byte("<metadata/>")
var xzBuf bytes.Buffer
xw, _ := xz.NewWriter(&xzBuf)
_, _ = xw.Write(plain)
_ = xw.Close()
zw, _ := zstd.NewWriter(nil)
zst := zw.EncodeAll(plain, nil)
for href, body := range map[string][]byte{
"p.xml": plain,
"p.xml.gz": gzipDeterministic(plain),
"p.xml.xz": xzBuf.Bytes(),
"p.xml.zst": zst,
} {
got, err := decompress(href, body)
if err != nil || !bytes.Equal(got, plain) {
t.Errorf("%s: got %q, %v", href, got, err)
}
}
}
+2
View File
@@ -16,6 +16,7 @@ const (
PackagePuppet PackageType = "puppet" PackagePuppet PackageType = "puppet"
PackageTerraform PackageType = "terraform" PackageTerraform PackageType = "terraform"
PackageGoProxy PackageType = "goproxy" PackageGoProxy PackageType = "goproxy"
PackageCargo PackageType = "cargo"
PackageGitHubRPM PackageType = "github_rpm" PackageGitHubRPM PackageType = "github_rpm"
PackageGitHubDeb PackageType = "github_deb" PackageGitHubDeb PackageType = "github_deb"
PackageGitHubAlpine PackageType = "github_alpine" PackageGitHubAlpine PackageType = "github_alpine"
@@ -33,6 +34,7 @@ var validPackageTypes = map[PackageType]bool{
PackagePuppet: true, PackagePuppet: true,
PackageTerraform: true, PackageTerraform: true,
PackageGoProxy: true, PackageGoProxy: true,
PackageCargo: true,
PackageGitHubRPM: true, PackageGitHubRPM: true,
PackageGitHubDeb: true, PackageGitHubDeb: true,
PackageGitHubAlpine: true, PackageGitHubAlpine: true,
+1
View File
@@ -18,6 +18,7 @@ func TestPackageTypeValid(t *testing.T) {
models.PackagePuppet, models.PackagePuppet,
models.PackageTerraform, models.PackageTerraform,
models.PackageGoProxy, models.PackageGoProxy,
models.PackageCargo,
models.PackageGitHubRPM, models.PackageGitHubRPM,
models.PackageGitHubDeb, models.PackageGitHubDeb,
models.PackageGitHubAlpine, models.PackageGitHubAlpine,
+15
View File
@@ -228,6 +228,21 @@ go mod download`,
}, },
]; ];
case 'cargo':
return [
{
title: 'Point crates.io at this sparse registry',
language: 'toml',
code: `# ~/.cargo/config.toml
[source.crates-io]
replace-with = "${name}"
[registries.${name}]
index = "sparse+${proxy}/"`,
note: 'config.json is served by artifactapi so crate downloads also go through this remote.',
},
];
case 'puppet': case 'puppet':
return [ return [
{ {
+1
View File
@@ -17,6 +17,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
puppet: 'yellow', puppet: 'yellow',
terraform: 'blue', terraform: 'blue',
goproxy: 'green', goproxy: 'green',
cargo: 'yellow',
}; };
export function Locals() { export function Locals() {
+1
View File
@@ -17,6 +17,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
puppet: 'yellow', puppet: 'yellow',
terraform: 'blue', terraform: 'blue',
goproxy: 'green', goproxy: 'green',
cargo: 'yellow',
}; };
export function Remotes() { export function Remotes() {