Compare commits

...

6 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
19 changed files with 1103 additions and 46 deletions
+2 -1
View File
@@ -1,6 +1,6 @@
# ArtifactAPI # 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 ## Quick Start
@@ -33,6 +33,7 @@ API: `http://localhost:8000` | Frontend: `http://localhost:5173`
| `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` | | `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) | | `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, 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 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
@@ -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"}, {"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"}, {"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"}, {"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"},
} }
+103
View File
@@ -14,6 +14,8 @@ import (
"strings" "strings"
"testing" "testing"
"time" "time"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
) )
// TestVirtualPyPIMerge uploads different packages to two pypi locals and // 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) 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)
}
}
+13 -1
View File
@@ -1,6 +1,7 @@
package v1 package v1
import ( import (
"bytes"
"errors" "errors"
"fmt" "fmt"
"io" "io"
@@ -105,13 +106,24 @@ func (h *ProxyHandler) handleProxy(w http.ResponseWriter, r *http.Request) {
} }
defer result.Reader.Close() 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("Content-Type", result.ContentType)
w.Header().Set("X-Artifact-Source", result.Source) w.Header().Set("X-Artifact-Source", result.Source)
if result.Size > 0 { if result.Size > 0 {
w.Header().Set("X-Artifact-Size", fmt.Sprintf("%d", result.Size)) w.Header().Set("X-Artifact-Size", fmt.Sprintf("%d", result.Size))
} }
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
io.Copy(w, result.Reader) io.Copy(w, body)
} }
func (h *ProxyHandler) handleProxyHead(w http.ResponseWriter, r *http.Request) { func (h *ProxyHandler) handleProxyHead(w http.ResponseWriter, r *http.Request) {
+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 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 // RemoteMetadataStore is the persistence surface a RemoteServer needs to cache
// and read the metadata it derives per upstream asset. *database.DB satisfies it. // and read the metadata it derives per upstream asset. *database.DB satisfies it.
type RemoteMetadataStore interface { type RemoteMetadataStore interface {
+1
View File
@@ -29,6 +29,7 @@ import (
_ "git.unkin.net/unkin/artifactapi/internal/provider/goproxy" _ "git.unkin.net/unkin/artifactapi/internal/provider/goproxy"
_ "git.unkin.net/unkin/artifactapi/internal/provider/helm" _ "git.unkin.net/unkin/artifactapi/internal/provider/helm"
_ "git.unkin.net/unkin/artifactapi/internal/provider/npm" _ "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/puppet"
_ "git.unkin.net/unkin/artifactapi/internal/provider/pypi" _ "git.unkin.net/unkin/artifactapi/internal/provider/pypi"
"git.unkin.net/unkin/artifactapi/internal/provider/rpm" "git.unkin.net/unkin/artifactapi/internal/provider/rpm"
+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"
"net/http/httptest" "net/http/httptest"
"net/url" "net/url"
"path"
"slices" "slices"
"strings" "strings"
"sync" "sync"
@@ -26,32 +27,51 @@ type Engine struct {
proxyEngine *proxy.Engine proxyEngine *proxy.Engine
getRemote func(context.Context, string) (*models.Remote, error) getRemote func(context.Context, string) (*models.Remote, error)
rpmMember func(context.Context, string) (*RPMMember, error) rpmMember func(context.Context, string) (*RPMMember, error)
debMember func(ctx context.Context, name, base string) (*DebMember, error)
mergeTTL time.Duration mergeTTL time.Duration
sf singleflight.Group sf singleflight.Group
mu sync.Mutex mu sync.Mutex
rpm map[string]*rpmGen gens map[string]*repoGen
} }
// rpmGen holds a virtual's current merge plus the previous one, so a client // mergedRepo is a merged index document plus the files it references.
// holding the previous repomd can still fetch its content-hashed data files. type mergedRepo interface {
type rpmGen struct { index() []byte
cur, prev *RPMRepo files() map[string][]byte
at time.Time }
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 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 { func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine {
e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL} e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL}
e.rpmMember = e.fetchRPMMember e.rpmMember = e.fetchRPMMember
e.debMember = e.fetchDebMember
return e 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 { switch virt.PackageType {
case models.PackageRPM:
return e.fetchRPM(ctx, virt, path) return e.fetchRPM(ctx, virt, path)
case models.PackageDeb:
return e.fetchDeb(ctx, virt, path)
} }
merger, err := GetMerger(virt.PackageType) merger, err := GetMerger(virt.PackageType)
@@ -170,11 +190,22 @@ var (
ErrBadPath = errors.New("invalid path") ErrBadPath = errors.New("invalid path")
) )
// MemberRedirect maps an rpm virtual package path (<member>/<href>, as written // MemberRedirect maps an rpm or deb virtual package path (<member>/<href>, as
// by MergeRPM) to the owning member's route. ok is false for any other path; // written by MergeRPM/MergeDeb) to the owning member's route. ok is false for
// ErrBadPath means the path is absolute or escapes the member. // 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) { 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 return "", false, nil
} }
name, rest, found := strings.Cut(path, "/") 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/") { if !strings.HasPrefix(path, "repodata/") {
return nil, "", ErrNotFound return nil, "", ErrNotFound
} }
key := "rpm\x00" + virt.Name
if path != "repodata/repomd.xml" { 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 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 { if err != nil {
return nil, "", err return nil, "", err
} }
if path == "repodata/repomd.xml" { 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 body, "application/gzip", nil
} }
return nil, "", ErrNotFound return nil, "", ErrNotFound
} }
// mergedRPM returns the virtual's merge, reusing it for mergeTTL and // merged returns the merge cached under key, reusing it for mergeTTL and
// collapsing concurrent merges of the same virtual into one. // collapsing concurrent merges of the same key into one.
// ponytail: per-replica cache; behind a non-sticky LB a data request landing on // 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 // another replica re-merges and 404s if a member changed in between. Move to the
// shared redis cache if that shows up. // 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() e.mu.Lock()
g := e.rpm[virt.Name] g := e.gens[key]
if g != nil && time.Since(g.at) < e.mergeTTL { if g != nil && time.Since(g.at) < e.mergeTTL {
g.used = time.Now()
e.mu.Unlock() e.mu.Unlock()
return g.cur, nil return g.cur, nil
} }
e.mu.Unlock() e.mu.Unlock()
v, err, _ := e.sf.Do(virt.Name, func() (any, error) { v, err, _ := e.sf.Do(key, func() (any, error) {
repo, err := e.mergeRPM(context.WithoutCancel(ctx), virt) repo, err := merge(context.WithoutCancel(ctx))
if err != nil { if err != nil {
return nil, err return nil, err
} }
e.mu.Lock() e.mu.Lock()
defer e.mu.Unlock() defer e.mu.Unlock()
if e.rpm == nil { if e.gens == nil {
e.rpm = map[string]*rpmGen{} e.gens = map[string]*repoGen{}
} }
g := e.rpm[virt.Name] g := e.gens[key]
switch { switch {
case g == nil: case g == nil:
e.rpm[virt.Name] = &rpmGen{cur: repo, at: time.Now()} e.gens[key] = &repoGen{cur: repo, at: time.Now(), used: time.Now()}
case string(g.cur.Repomd) == string(repo.Repomd): case string(g.cur.index()) == string(repo.index()):
g.at = time.Now() g.at, g.used = time.Now(), time.Now()
repo = g.cur repo = g.cur
default: 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 return repo, nil
}) })
if err != nil { if err != nil {
return nil, err return nil, err
} }
return v.(*RPMRepo), nil return v.(mergedRepo), nil
} }
func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) { 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)) errs := make([]error, len(virt.Members))
var wg sync.WaitGroup var wg sync.WaitGroup
for i, name := range virt.Members { for i, name := range virt.Members {
wg.Add(1) wg.Add(1)
go func() { go func() {
defer wg.Done() defer wg.Done()
m, err := e.rpmMember(ctx, name) m, err := fetch(name)
if err != nil { if err != nil {
errs[i] = fmt.Errorf("member %q: %w", name, err) errs[i] = fmt.Errorf("member %q: %w", name, err)
return return
@@ -299,26 +347,101 @@ func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, e
if err := errors.Join(errs...); err != nil { if err := errors.Join(errs...); err != nil {
return nil, fmt.Errorf("virtual %q: %v", virt.Name, err) return nil, fmt.Errorf("virtual %q: %v", virt.Name, err)
} }
return members, nil
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) { 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() e.mu.Lock()
defer e.mu.Unlock() 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 { if g == nil {
return nil, false return nil, false
} }
for _, r := range []*RPMRepo{g.cur, g.prev} { for _, r := range []mergedRepo{g.cur, g.prev} {
if r == nil { if r == nil {
continue continue
} }
if body, ok := r.Files[path]; ok { if body, ok := r.files()[path]; ok {
g.used = time.Now()
return body, true return body, true
} }
} }
+211
View File
@@ -5,6 +5,7 @@ import (
"errors" "errors"
"fmt" "fmt"
"regexp" "regexp"
"strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
"testing" "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")
}
}
+2
View File
@@ -17,6 +17,7 @@ const (
PackageTerraform PackageType = "terraform" PackageTerraform PackageType = "terraform"
PackageGoProxy PackageType = "goproxy" PackageGoProxy PackageType = "goproxy"
PackageCargo PackageType = "cargo" PackageCargo PackageType = "cargo"
PackageNuGet PackageType = "nuget"
PackageGitHubRPM PackageType = "github_rpm" PackageGitHubRPM PackageType = "github_rpm"
PackageGitHubDeb PackageType = "github_deb" PackageGitHubDeb PackageType = "github_deb"
PackageGitHubAlpine PackageType = "github_alpine" PackageGitHubAlpine PackageType = "github_alpine"
@@ -35,6 +36,7 @@ var validPackageTypes = map[PackageType]bool{
PackageTerraform: true, PackageTerraform: true,
PackageGoProxy: true, PackageGoProxy: true,
PackageCargo: true, PackageCargo: true,
PackageNuGet: true,
PackageGitHubRPM: true, PackageGitHubRPM: true,
PackageGitHubDeb: true, PackageGitHubDeb: true,
PackageGitHubAlpine: true, PackageGitHubAlpine: true,
+1
View File
@@ -19,6 +19,7 @@ func TestPackageTypeValid(t *testing.T) {
models.PackageTerraform, models.PackageTerraform,
models.PackageGoProxy, models.PackageGoProxy,
models.PackageCargo, models.PackageCargo,
models.PackageNuGet,
models.PackageGitHubRPM, models.PackageGitHubRPM,
models.PackageGitHubDeb, models.PackageGitHubDeb,
models.PackageGitHubAlpine, 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': case 'puppet':
return [ return [
{ {
+1
View File
@@ -18,6 +18,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
terraform: 'blue', terraform: 'blue',
goproxy: 'green', goproxy: 'green',
cargo: 'yellow', cargo: 'yellow',
nuget: 'blue',
}; };
export function Locals() { export function Locals() {
+1
View File
@@ -18,6 +18,7 @@ const typeColors: Record<string, 'blue' | 'green' | 'yellow' | 'red' | 'default'
terraform: 'blue', terraform: 'blue',
goproxy: 'green', goproxy: 'green',
cargo: 'yellow', cargo: 'yellow',
nuget: 'blue',
}; };
export function Remotes() { export function Remotes() {