Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fdbf322ca8 | |||
| 09ab49ef42 | |||
| d9f9fd3bca | |||
| b111c2e57f |
@@ -279,20 +279,6 @@ content-addressable store (deduplicated by digest, reaped by GC once
|
||||
unreferenced); tags are mutable references and re-pushing a tag moves it. Blob
|
||||
uploads support both the monolithic and chunked (`POST`/`PATCH`/`PUT`) flows.
|
||||
|
||||
### Local npm registry
|
||||
|
||||
A local `npm` repo is a real npm registry: `npm publish` (scoped and unscoped)
|
||||
and `npm install` work against `/api/v1/local/{name}/`. npm requires an
|
||||
`_authToken` for the registry before it will publish; any value works.
|
||||
|
||||
```ini
|
||||
# .npmrc
|
||||
registry=https://artifactapi.k8s.syd1.au.unkin.net/api/v1/local/npm-internal/
|
||||
//artifactapi.k8s.syd1.au.unkin.net/api/v1/local/npm-internal/:_authToken=unused
|
||||
```
|
||||
|
||||
Versions are immutable; dist-tags sent with a publish move to that version.
|
||||
|
||||
## Access Control
|
||||
|
||||
| Field | Default | Behaviour |
|
||||
@@ -311,7 +297,6 @@ No patterns + no blocklist = open proxy. Provider handles mutability classificat
|
||||
```
|
||||
GET /api/v1/remote/{name}/{path} Proxy/cache artifact
|
||||
GET /api/v1/virtual/{name}/{path} Virtual repo (merged index)
|
||||
PUT /api/v1/local/{name}/{package} npm publish (local npm repos)
|
||||
GET /v2/{name}/{path} Docker Registry v2
|
||||
```
|
||||
|
||||
@@ -386,8 +371,6 @@ additive and idempotent for the same reason; a test enforces it.
|
||||
| `GITHUB_APP_INSTALLATION_ID` | | GitHub App installation id |
|
||||
| `GITHUB_APP_PRIVATE_KEY` | | GitHub App private key, inline PEM |
|
||||
| `GITHUB_APP_PRIVATE_KEY_PATH` | | GitHub App private key, file path (alternative to inline PEM) |
|
||||
| `ARTIFACTAPI_OIDC_ISSUER` | | OIDC issuer URL (e.g. Vault `https://vault.service.consul:8200/v1/identity/oidc`). When set, `POST /api/v2/remotes`, `POST /api/v2/virtuals` and `PUT`/`DELETE` on `/api/v2/{remotes,virtuals}/{name}` require `Authorization: Bearer <jwt>` verified against the issuer's JWKS. Uploads, object/file deletes, probe, reads and `/v2` stay open. Unset = no auth |
|
||||
| `ARTIFACTAPI_OIDC_AUDIENCE` | | Comma-separated allowed `aud` values (e.g. the Vault OIDC role's client_id). Required when `ARTIFACTAPI_OIDC_ISSUER` is set |
|
||||
|
||||
## Development
|
||||
|
||||
|
||||
@@ -27,9 +27,7 @@ already-running stack.
|
||||
mock upstream: first fetch `X-Artifact-Source: remote`, second `cache`, bytes
|
||||
verified against the origin fixture.
|
||||
- **Local uploads** — generic (upload/download), pypi (wheel + generated `simple/`
|
||||
index), rpm (real package + **automatic repodata** generation), npm (stock
|
||||
`node:22` client: `npm publish` unscoped + scoped, `npm install` + `require()`,
|
||||
second version, republish rejected).
|
||||
index), rpm (real package + **automatic repodata** generation).
|
||||
- **Virtual repositories** — pypi simple-index merge and helm `index.yaml` merge
|
||||
across two members.
|
||||
- **Mirrorlist** — an rpm remote with a `mirrorlist` of extra upstream mirrors
|
||||
|
||||
@@ -1,62 +0,0 @@
|
||||
//go:build dockere2e
|
||||
|
||||
package e2edocker
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestLocalNPMPublishInstall drives a stock npm client against a local npm
|
||||
// repo: publish unscoped and scoped packages, install and require() them in a
|
||||
// fresh project, publish a second version, and reject a republish.
|
||||
func TestLocalNPMPublishInstall(t *testing.T) {
|
||||
network := os.Getenv("COMPOSE_NETWORK")
|
||||
internal := os.Getenv("ARTIFACTAPI_INTERNAL")
|
||||
if network == "" || internal == "" {
|
||||
t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; run via scripts/docker-e2e.sh")
|
||||
}
|
||||
if _, err := exec.LookPath("docker"); err != nil {
|
||||
t.Skip("docker not available on the test host")
|
||||
}
|
||||
|
||||
name := fmt.Sprintf("e2e-npm-local-%d", time.Now().UnixNano())
|
||||
createRepo(t, fmt.Sprintf(`{"name":%q,"package_type":"npm","repo_type":"local"}`, name))
|
||||
defer deleteRepo(t, name)
|
||||
|
||||
reg := strings.TrimRight(internal, "/") + "/api/v1/local/" + name + "/"
|
||||
script := `set -euo pipefail
|
||||
printf 'registry=%s\n%s:_authToken=unused\n' "$REG" "${REG#http:}" > ~/.npmrc
|
||||
mk() { d="/w/src/$1-$2"; mkdir -p "$d"; cd "$d"
|
||||
printf '{"name":"%s","version":"%s","main":"index.js"}' "$1" "$2" > package.json
|
||||
echo "module.exports = '$1@$2';" > index.js; }
|
||||
mk e2e-npm-plain 1.0.0 && npm publish
|
||||
mk @e2e/npm-scoped 1.0.0 && npm publish
|
||||
mk e2e-npm-plain 1.0.0
|
||||
if npm publish 2>/dev/null; then echo "republish accepted"; exit 1; fi
|
||||
mkdir -p /w/app && cd /w/app && npm init -y >/dev/null
|
||||
npm install --no-audit --cache /tmp/fresh-cache e2e-npm-plain @e2e/npm-scoped
|
||||
node -e "console.log('REQ', require('e2e-npm-plain'), require('@e2e/npm-scoped'))"
|
||||
mk e2e-npm-plain 1.1.0 && npm publish
|
||||
echo "VERSIONS $(npm view e2e-npm-plain versions --json | tr -d ' \n')"
|
||||
echo "LATEST $(npm view e2e-npm-plain dist-tags.latest)"`
|
||||
|
||||
out, err := exec.Command("docker", "run", "--rm", "--network", network,
|
||||
"-e", "REG="+reg, "node:22", "bash", "-c", script).CombinedOutput()
|
||||
if err != nil {
|
||||
t.Fatalf("npm client run failed: %v\n%s", err, out)
|
||||
}
|
||||
for _, want := range []string{
|
||||
"REQ e2e-npm-plain@1.0.0 @e2e/npm-scoped@1.0.0",
|
||||
`VERSIONS ["1.0.0","1.1.0"]`,
|
||||
"LATEST 1.1.0",
|
||||
} {
|
||||
if !strings.Contains(string(out), want) {
|
||||
t.Fatalf("missing %q in npm output:\n%s", want, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -7,9 +7,7 @@ require (
|
||||
github.com/cavaliergopher/rpm v1.3.0
|
||||
github.com/charmbracelet/bubbletea v1.3.10
|
||||
github.com/charmbracelet/lipgloss v1.1.0
|
||||
github.com/coreos/go-oidc/v3 v3.21.0
|
||||
github.com/go-chi/chi/v5 v5.3.0
|
||||
github.com/go-jose/go-jose/v4 v4.1.4
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/jackc/pgx/v5 v5.10.0
|
||||
github.com/klauspost/compress v1.19.2
|
||||
@@ -104,7 +102,6 @@ require (
|
||||
go.uber.org/atomic v1.11.0 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/net v0.56.0 // indirect
|
||||
golang.org/x/oauth2 v0.36.0 // indirect
|
||||
golang.org/x/sys v0.47.0 // indirect
|
||||
golang.org/x/text v0.40.0 // indirect
|
||||
gopkg.in/ini.v1 v1.67.2 // indirect
|
||||
|
||||
@@ -44,8 +44,6 @@ github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
|
||||
github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
|
||||
github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A=
|
||||
github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw=
|
||||
github.com/coreos/go-oidc/v3 v3.21.0 h1:wZo4Q9Pum8dYEj0eMUPrqR+kvuGkeUplbLpNCkBqoWM=
|
||||
github.com/coreos/go-oidc/v3 v3.21.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4=
|
||||
github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA=
|
||||
github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
|
||||
github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s=
|
||||
@@ -69,8 +67,6 @@ github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeO
|
||||
github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE=
|
||||
github.com/go-chi/chi/v5 v5.3.0 h1:halUjDxhshgXHMrao5bB8eNBXo/rnzwr8m5m36glehM=
|
||||
github.com/go-chi/chi/v5 v5.3.0/go.mod h1:R+tYY2hNuVUUjxoPtqUdgBqevM9s9njzkTLutVsOCto=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
|
||||
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
|
||||
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
|
||||
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
@@ -229,8 +225,6 @@ golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM
|
||||
golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
|
||||
golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
|
||||
golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
|
||||
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
|
||||
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
|
||||
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
|
||||
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
||||
@@ -1,286 +0,0 @@
|
||||
package v1
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha1"
|
||||
"crypto/sha512"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/database"
|
||||
"git.unkin.net/unkin/artifactapi/internal/storage"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
// Local npm repos speak the npm registry publish/read protocol. Per package:
|
||||
// <pkg>/-/<base>-<version>.tgz tarball (immutable)
|
||||
// <pkg>/-/<base>-<version>.json version manifest (immutable)
|
||||
// <pkg>/-/dist-tags/<tag> version string (mutable)
|
||||
|
||||
var (
|
||||
npmNameRe = regexp.MustCompile(`^(@[a-z0-9][a-z0-9._~-]*/)?[a-z0-9][a-z0-9._~-]*$`)
|
||||
npmRefRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+_-]*$`)
|
||||
)
|
||||
|
||||
func (h *ProxyHandler) localNPMRemote(r *http.Request, name string) (*models.Remote, bool) {
|
||||
remote, err := h.db.GetRemote(r.Context(), name)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return remote, remote.RepoType == models.RepoTypeLocal && remote.PackageType == models.PackageNPM
|
||||
}
|
||||
|
||||
// npmPackageName returns the package a packument/publish path addresses, or ""
|
||||
// when the path is something else (e.g. a tarball under <pkg>/-/).
|
||||
func npmPackageName(raw string) string {
|
||||
p, err := url.PathUnescape(raw)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
p = strings.TrimSuffix(p, "/")
|
||||
if !npmNameRe.MatchString(p) {
|
||||
return ""
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
type npmRelease struct {
|
||||
version string
|
||||
manifest []byte
|
||||
tarball []byte
|
||||
tags []string
|
||||
}
|
||||
|
||||
func parseNPMPublish(body io.Reader, pkg string) (*npmRelease, error) {
|
||||
var doc struct {
|
||||
Name string `json:"name"`
|
||||
DistTags map[string]string `json:"dist-tags"`
|
||||
Versions map[string]json.RawMessage `json:"versions"`
|
||||
Attachments map[string]struct {
|
||||
Data string `json:"data"`
|
||||
} `json:"_attachments"`
|
||||
}
|
||||
if err := json.NewDecoder(body).Decode(&doc); err != nil {
|
||||
return nil, fmt.Errorf("invalid publish body: %w", err)
|
||||
}
|
||||
if doc.Name != pkg {
|
||||
return nil, fmt.Errorf("body name %q does not match %q", doc.Name, pkg)
|
||||
}
|
||||
if len(doc.Versions) != 1 || len(doc.Attachments) != 1 {
|
||||
return nil, errors.New("publish must carry exactly one version and one tarball")
|
||||
}
|
||||
|
||||
rel := &npmRelease{}
|
||||
for v, m := range doc.Versions {
|
||||
rel.version, rel.manifest = v, m
|
||||
}
|
||||
if !npmRefRe.MatchString(rel.version) {
|
||||
return nil, fmt.Errorf("invalid version %q", rel.version)
|
||||
}
|
||||
var meta struct {
|
||||
Name string `json:"name"`
|
||||
Version string `json:"version"`
|
||||
Dist struct {
|
||||
Integrity string `json:"integrity"`
|
||||
Shasum string `json:"shasum"`
|
||||
} `json:"dist"`
|
||||
}
|
||||
if err := json.Unmarshal(rel.manifest, &meta); err != nil {
|
||||
return nil, fmt.Errorf("invalid version manifest: %w", err)
|
||||
}
|
||||
if meta.Name != pkg || meta.Version != rel.version {
|
||||
return nil, errors.New("version manifest name/version mismatch")
|
||||
}
|
||||
|
||||
for _, a := range doc.Attachments {
|
||||
tgz, err := base64.StdEncoding.DecodeString(a.Data)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid tarball encoding: %w", err)
|
||||
}
|
||||
rel.tarball = tgz
|
||||
}
|
||||
if err := verifyNPMDist(rel.tarball, meta.Dist.Integrity, meta.Dist.Shasum); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for tag, v := range doc.DistTags {
|
||||
if !npmRefRe.MatchString(tag) || v != rel.version {
|
||||
return nil, fmt.Errorf("dist-tag %q must point at the published version", tag)
|
||||
}
|
||||
rel.tags = append(rel.tags, tag)
|
||||
}
|
||||
return rel, nil
|
||||
}
|
||||
|
||||
func verifyNPMDist(tgz []byte, integrity, shasum string) error {
|
||||
if b64, ok := strings.CutPrefix(integrity, "sha512-"); ok {
|
||||
sum := sha512.Sum512(tgz)
|
||||
if base64.StdEncoding.EncodeToString(sum[:]) != b64 {
|
||||
return errors.New("tarball does not match dist.integrity")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if shasum != "" {
|
||||
sum := sha1.Sum(tgz)
|
||||
if hex.EncodeToString(sum[:]) != shasum {
|
||||
return errors.New("tarball does not match dist.shasum")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return errors.New("version manifest carries no sha512 integrity or shasum")
|
||||
}
|
||||
|
||||
func npmError(w http.ResponseWriter, status int, msg string) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
json.NewEncoder(w).Encode(map[string]string{"error": msg})
|
||||
}
|
||||
|
||||
func (h *ProxyHandler) handleLocalPut(w http.ResponseWriter, r *http.Request) {
|
||||
remote, ok := h.localNPMRemote(r, chi.URLParam(r, "localName"))
|
||||
if !ok {
|
||||
npmError(w, http.StatusMethodNotAllowed, "publish is only supported on local npm repositories")
|
||||
return
|
||||
}
|
||||
pkg := npmPackageName(chi.URLParam(r, "*"))
|
||||
if pkg == "" {
|
||||
npmError(w, http.StatusBadRequest, "invalid package name")
|
||||
return
|
||||
}
|
||||
rel, err := parseNPMPublish(r.Body, pkg)
|
||||
if err != nil {
|
||||
npmError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
base := pkg[strings.LastIndex(pkg, "/")+1:]
|
||||
stem := pkg + "/-/" + base + "-" + rel.version
|
||||
// The tarball's immutable row is the version's uniqueness gate.
|
||||
if err := h.npmStore(r.Context(), remote.Name, stem+".tgz", rel.tarball, "application/octet-stream", false); err != nil {
|
||||
if errors.Is(err, database.ErrAlreadyExists) {
|
||||
npmError(w, http.StatusConflict, fmt.Sprintf("%s@%s already exists; overwrites are not allowed", pkg, rel.version))
|
||||
return
|
||||
}
|
||||
npmError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if err := h.npmStore(r.Context(), remote.Name, stem+".json", rel.manifest, "application/json", false); err != nil {
|
||||
npmError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
for _, tag := range rel.tags {
|
||||
if err := h.npmStore(r.Context(), remote.Name, pkg+"/-/dist-tags/"+tag, []byte(rel.version), "text/plain", true); err != nil {
|
||||
npmError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
slog.Info("local npm package published", "repo", remote.Name, "package", pkg, "version", rel.version)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
json.NewEncoder(w).Encode(map[string]any{"ok": true, "id": pkg})
|
||||
}
|
||||
|
||||
func (h *ProxyHandler) npmStore(ctx context.Context, repo, path string, data []byte, contentType string, mutable bool) error {
|
||||
result, err := h.cas.Store(ctx, bytes.NewReader(data), contentType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := h.db.UpsertBlob(ctx, result.ContentHash, result.S3Key, result.SizeBytes, contentType); err != nil {
|
||||
return err
|
||||
}
|
||||
if mutable {
|
||||
return h.db.UpsertLocalFile(ctx, repo, path, result.ContentHash)
|
||||
}
|
||||
return h.db.CreateLocalFile(ctx, repo, path, result.ContentHash)
|
||||
}
|
||||
|
||||
func (h *ProxyHandler) readLocalBlob(ctx context.Context, contentHash string) ([]byte, error) {
|
||||
reader, _, err := h.store.Download(ctx, storage.BlobKey(strings.TrimPrefix(contentHash, "sha256:")))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer reader.Close()
|
||||
return io.ReadAll(reader)
|
||||
}
|
||||
|
||||
func (h *ProxyHandler) serveNPMPackument(w http.ResponseWriter, r *http.Request, remote *models.Remote, pkg string) {
|
||||
files, err := h.db.ListLocalFilesByPrefix(r.Context(), remote.Name, pkg+"/-/")
|
||||
if err != nil {
|
||||
npmError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
read := func(hash string) ([]byte, error) { return h.readLocalBlob(r.Context(), hash) }
|
||||
tarballBase := fmt.Sprintf("%s://%s/api/v1/local/%s/", scheme(r), r.Host, remote.Name)
|
||||
doc, err := buildNPMPackument(pkg, files, read, tarballBase)
|
||||
if err != nil {
|
||||
npmError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
if doc == nil {
|
||||
npmError(w, http.StatusNotFound, "not found")
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set("X-Artifact-Source", "local")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
json.NewEncoder(w).Encode(doc)
|
||||
}
|
||||
|
||||
// buildNPMPackument returns nil when pkg has no published versions.
|
||||
func buildNPMPackument(pkg string, files []database.LocalFile, read func(hash string) ([]byte, error), tarballBase string) (map[string]any, error) {
|
||||
prefix := pkg + "/-/"
|
||||
versions := map[string]any{}
|
||||
tags := map[string]string{}
|
||||
for _, f := range files {
|
||||
// LIKE treats '_' as a wildcard, so the prefix match can over-select.
|
||||
rest, ok := strings.CutPrefix(f.FilePath, prefix)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if tag, isTag := strings.CutPrefix(rest, "dist-tags/"); isTag {
|
||||
v, err := read(f.ContentHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tags[tag] = string(v)
|
||||
continue
|
||||
}
|
||||
stem, isManifest := strings.CutSuffix(rest, ".json")
|
||||
if !isManifest {
|
||||
continue
|
||||
}
|
||||
raw, err := read(f.ContentHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
return nil, fmt.Errorf("manifest %s: %w", f.FilePath, err)
|
||||
}
|
||||
dist, _ := m["dist"].(map[string]any)
|
||||
if dist == nil {
|
||||
dist = map[string]any{}
|
||||
m["dist"] = dist
|
||||
}
|
||||
dist["tarball"] = tarballBase + prefix + stem + ".tgz"
|
||||
v, _ := m["version"].(string)
|
||||
versions[v] = m
|
||||
}
|
||||
if len(versions) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return map[string]any{"_id": pkg, "name": pkg, "dist-tags": tags, "versions": versions}, nil
|
||||
}
|
||||
@@ -1,162 +0,0 @@
|
||||
package v1
|
||||
|
||||
import (
|
||||
"crypto/sha1"
|
||||
"crypto/sha512"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/database"
|
||||
)
|
||||
|
||||
func TestNPMPackageName(t *testing.T) {
|
||||
tests := map[string]string{
|
||||
"left-pad": "left-pad",
|
||||
"left-pad/": "left-pad",
|
||||
"@acme%2fwidget": "@acme/widget",
|
||||
"@acme%2Fwidget": "@acme/widget",
|
||||
"@acme/widget": "@acme/widget",
|
||||
"left-pad/-/left-pad.tgz": "",
|
||||
"@acme/widget/-/w-1.tgz": "",
|
||||
"a/b": "",
|
||||
"../etc": "",
|
||||
"Upper": "",
|
||||
"": "",
|
||||
}
|
||||
for in, want := range tests {
|
||||
if got := npmPackageName(in); got != want {
|
||||
t.Errorf("npmPackageName(%q) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type publishOpts struct {
|
||||
bodyName, manifestName, version, integrity, shasum string
|
||||
tags map[string]string
|
||||
extraVersion bool
|
||||
}
|
||||
|
||||
func publishBody(t *testing.T, pkg string, tgz []byte, o publishOpts) string {
|
||||
t.Helper()
|
||||
sha := sha512.Sum512(tgz)
|
||||
sh1 := sha1.Sum(tgz)
|
||||
if o.bodyName == "" {
|
||||
o.bodyName = pkg
|
||||
}
|
||||
if o.manifestName == "" {
|
||||
o.manifestName = pkg
|
||||
}
|
||||
if o.version == "" {
|
||||
o.version = "1.0.0"
|
||||
}
|
||||
if o.integrity == "" {
|
||||
o.integrity = "sha512-" + base64.StdEncoding.EncodeToString(sha[:])
|
||||
}
|
||||
if o.shasum == "" {
|
||||
o.shasum = hex.EncodeToString(sh1[:])
|
||||
}
|
||||
if o.tags == nil {
|
||||
o.tags = map[string]string{"latest": o.version}
|
||||
}
|
||||
versions := map[string]any{o.version: map[string]any{
|
||||
"name": o.manifestName, "version": o.version, "main": "index.js",
|
||||
"dist": map[string]any{"integrity": o.integrity, "shasum": o.shasum, "tarball": "http://elsewhere/x.tgz"},
|
||||
}}
|
||||
if o.extraVersion {
|
||||
versions["2.0.0"] = map[string]any{"name": pkg, "version": "2.0.0"}
|
||||
}
|
||||
b, _ := json.Marshal(map[string]any{
|
||||
"_id": pkg, "name": o.bodyName, "dist-tags": o.tags, "versions": versions,
|
||||
"_attachments": map[string]any{pkg + "-" + o.version + ".tgz": map[string]any{
|
||||
"content_type": "application/octet-stream", "data": base64.StdEncoding.EncodeToString(tgz), "length": len(tgz),
|
||||
}},
|
||||
})
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func TestParseNPMPublish(t *testing.T) {
|
||||
tgz := []byte("fake tarball bytes")
|
||||
|
||||
rel, err := parseNPMPublish(strings.NewReader(publishBody(t, "@acme/widget", tgz, publishOpts{tags: map[string]string{"latest": "1.0.0", "beta": "1.0.0"}})), "@acme/widget")
|
||||
if err != nil {
|
||||
t.Fatalf("valid scoped publish rejected: %v", err)
|
||||
}
|
||||
if rel.version != "1.0.0" || string(rel.tarball) != string(tgz) || len(rel.tags) != 2 {
|
||||
t.Fatalf("unexpected release: %+v", rel)
|
||||
}
|
||||
|
||||
if _, err := parseNPMPublish(strings.NewReader(publishBody(t, "left-pad", tgz, publishOpts{integrity: "none", shasum: hex.EncodeToString(make([]byte, 20))})), "left-pad"); err == nil {
|
||||
t.Fatal("shasum fallback should reject a mismatched tarball")
|
||||
}
|
||||
if _, err := parseNPMPublish(strings.NewReader(publishBody(t, "left-pad", tgz, publishOpts{integrity: "none"})), "left-pad"); err != nil {
|
||||
t.Fatalf("matching shasum fallback rejected: %v", err)
|
||||
}
|
||||
|
||||
bad := map[string]publishOpts{
|
||||
"body name mismatch": {bodyName: "other"},
|
||||
"manifest name mismatch": {manifestName: "other"},
|
||||
"integrity mismatch": {integrity: "sha512-AAAA"},
|
||||
"path-like version": {version: "../1.0.0"},
|
||||
"tag at another version": {tags: map[string]string{"latest": "9.9.9"}},
|
||||
"path-like tag": {tags: map[string]string{"../x": "1.0.0"}},
|
||||
"multiple versions at once": {extraVersion: true},
|
||||
}
|
||||
for name, o := range bad {
|
||||
if _, err := parseNPMPublish(strings.NewReader(publishBody(t, "left-pad", tgz, o)), "left-pad"); err == nil {
|
||||
t.Errorf("%s: expected rejection", name)
|
||||
}
|
||||
}
|
||||
if _, err := parseNPMPublish(strings.NewReader(`{"name":"left-pad","versions":{}}`), "left-pad"); err == nil {
|
||||
t.Error("deprecate-style body without attachments should be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildNPMPackument(t *testing.T) {
|
||||
blobs := map[string]string{
|
||||
"h1": `{"name":"my_pkg","version":"1.0.0","dist":{"shasum":"aa","tarball":"http://elsewhere/x.tgz"}}`,
|
||||
"h2": `{"name":"my_pkg","version":"1.1.0"}`,
|
||||
"h3": "1.1.0",
|
||||
"hx": `{"name":"myXpkg","version":"6.6.6"}`,
|
||||
}
|
||||
read := func(h string) ([]byte, error) {
|
||||
b, ok := blobs[h]
|
||||
if !ok {
|
||||
return nil, errors.New("missing blob")
|
||||
}
|
||||
return []byte(b), nil
|
||||
}
|
||||
files := []database.LocalFile{
|
||||
{FilePath: "my_pkg/-/dist-tags/latest", ContentHash: "h3"},
|
||||
{FilePath: "my_pkg/-/my_pkg-1.0.0.json", ContentHash: "h1"},
|
||||
{FilePath: "my_pkg/-/my_pkg-1.0.0.tgz", ContentHash: "nope"},
|
||||
{FilePath: "my_pkg/-/my_pkg-1.1.0.json", ContentHash: "h2"},
|
||||
{FilePath: "myXpkg/-/myXpkg-6.6.6.json", ContentHash: "hx"},
|
||||
}
|
||||
|
||||
doc, err := buildNPMPackument("my_pkg", files, read, "http://reg/api/v1/local/npm/")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tags := doc["dist-tags"].(map[string]string); tags["latest"] != "1.1.0" {
|
||||
t.Errorf("dist-tags = %v", tags)
|
||||
}
|
||||
versions := doc["versions"].(map[string]any)
|
||||
if len(versions) != 2 {
|
||||
t.Fatalf("versions = %v (LIKE over-match must be dropped)", versions)
|
||||
}
|
||||
v1 := versions["1.0.0"].(map[string]any)["dist"].(map[string]any)
|
||||
if v1["tarball"] != "http://reg/api/v1/local/npm/my_pkg/-/my_pkg-1.0.0.tgz" || v1["shasum"] != "aa" {
|
||||
t.Errorf("1.0.0 dist = %v", v1)
|
||||
}
|
||||
if versions["1.1.0"].(map[string]any)["dist"].(map[string]any)["tarball"] != "http://reg/api/v1/local/npm/my_pkg/-/my_pkg-1.1.0.tgz" {
|
||||
t.Error("1.1.0 missing rewritten tarball")
|
||||
}
|
||||
|
||||
if doc, err := buildNPMPackument("absent", files, read, "x/"); err != nil || doc != nil {
|
||||
t.Errorf("absent package: doc=%v err=%v", doc, err)
|
||||
}
|
||||
}
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
"git.unkin.net/unkin/artifactapi/internal/proxy"
|
||||
"git.unkin.net/unkin/artifactapi/internal/storage"
|
||||
"git.unkin.net/unkin/artifactapi/internal/virtual"
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
type ProxyHandler struct {
|
||||
@@ -42,7 +41,6 @@ func (h *ProxyHandler) Routes() chi.Router {
|
||||
r := chi.NewRouter()
|
||||
r.Get("/remote/{remoteName}/*", h.handleProxy)
|
||||
r.Get("/local/{localName}/*", h.handleLocal)
|
||||
r.Put("/local/{localName}/*", h.handleLocalPut)
|
||||
r.Get("/virtual/{virtualName}/*", h.handleVirtual)
|
||||
return r
|
||||
}
|
||||
@@ -201,13 +199,6 @@ func (h *ProxyHandler) handleLocal(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
if remote.PackageType == models.PackageNPM {
|
||||
if pkg := npmPackageName(path); pkg != "" {
|
||||
h.serveNPMPackument(w, r, remote, pkg)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
prov, _ := provider.Get(remote.PackageType)
|
||||
if indexer, ok := prov.(provider.LocalIndexer); ok {
|
||||
if indexer.ServeLocalIndex(w, r, h.db, remote.Name, path) {
|
||||
|
||||
@@ -1,158 +0,0 @@
|
||||
package auth
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/go-oidc/v3/oidc"
|
||||
)
|
||||
|
||||
// BearerOIDC verifies `Authorization: Bearer <jwt>` against an OIDC issuer's
|
||||
// JWKS. Discovery is lazy so startup never depends on the issuer being
|
||||
// reachable; a failed discovery is not retried until the backoff elapses.
|
||||
type BearerOIDC struct {
|
||||
issuer string
|
||||
audiences []string
|
||||
timeout time.Duration
|
||||
backoff time.Duration
|
||||
discover func(context.Context, string) (*oidc.Provider, error)
|
||||
|
||||
mu sync.Mutex
|
||||
verifier *oidc.IDTokenVerifier
|
||||
inflight chan struct{}
|
||||
failedAt time.Time
|
||||
}
|
||||
|
||||
var errIssuerUnavailable = errors.New("oidc issuer unavailable")
|
||||
|
||||
// NewBearerOIDC returns nil when issuer is empty, meaning auth is disabled.
|
||||
func NewBearerOIDC(issuer string, audiences []string) *BearerOIDC {
|
||||
if issuer == "" {
|
||||
return nil
|
||||
}
|
||||
return &BearerOIDC{issuer: issuer, audiences: audiences, timeout: 10 * time.Second, backoff: 15 * time.Second, discover: oidc.NewProvider}
|
||||
}
|
||||
|
||||
// getVerifier runs at most one discovery at a time without holding the lock
|
||||
// across network I/O; concurrent callers wait for it, and only callers within
|
||||
// the post-failure backoff fail fast.
|
||||
func (b *BearerOIDC) getVerifier(ctx context.Context) (*oidc.IDTokenVerifier, error) {
|
||||
b.mu.Lock()
|
||||
if v := b.verifier; v != nil {
|
||||
b.mu.Unlock()
|
||||
return v, nil
|
||||
}
|
||||
if done := b.inflight; done != nil {
|
||||
b.mu.Unlock()
|
||||
select {
|
||||
case <-done:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.verifier == nil {
|
||||
return nil, errIssuerUnavailable
|
||||
}
|
||||
return b.verifier, nil
|
||||
}
|
||||
if time.Since(b.failedAt) < b.backoff {
|
||||
b.mu.Unlock()
|
||||
return nil, errIssuerUnavailable
|
||||
}
|
||||
done := make(chan struct{})
|
||||
b.inflight = done
|
||||
b.mu.Unlock()
|
||||
|
||||
v, err := b.discoverOnce(done)
|
||||
if err != nil {
|
||||
slog.Error("oidc discovery failed", "issuer", b.issuer, "error", err)
|
||||
return nil, err
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// discoverOnce releases inflight and waiters on every exit, a panic included,
|
||||
// recording failedAt unless discovery succeeded.
|
||||
func (b *BearerOIDC) discoverOnce(done chan struct{}) (v *oidc.IDTokenVerifier, err error) {
|
||||
defer func() {
|
||||
b.mu.Lock()
|
||||
b.inflight = nil
|
||||
if v == nil {
|
||||
b.failedAt = time.Now()
|
||||
}
|
||||
b.verifier = v
|
||||
b.mu.Unlock()
|
||||
close(done)
|
||||
}()
|
||||
// The provider keeps this context and client for JWKS refreshes, so the
|
||||
// deadline lives on the client rather than the context.
|
||||
client := &http.Client{Timeout: b.timeout}
|
||||
p, err := b.discover(oidc.ClientContext(context.Background(), client), b.issuer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.Verifier(&oidc.Config{SkipClientIDCheck: true}), nil
|
||||
}
|
||||
|
||||
// Require gates every request through next on a valid token. A nil receiver
|
||||
// passes everything through.
|
||||
func (b *BearerOIDC) Require(next http.Handler) http.Handler {
|
||||
if b == nil {
|
||||
return next
|
||||
}
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
scheme, raw, _ := strings.Cut(strings.TrimSpace(r.Header.Get("Authorization")), " ")
|
||||
raw = strings.TrimSpace(raw)
|
||||
if !strings.EqualFold(scheme, "Bearer") || raw == "" {
|
||||
unauthorized(w, "missing bearer token")
|
||||
return
|
||||
}
|
||||
v, err := b.getVerifier(r.Context())
|
||||
if err != nil {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
_, _ = w.Write([]byte(`{"error":"token issuer unavailable"}`))
|
||||
return
|
||||
}
|
||||
tok, err := v.Verify(r.Context(), raw)
|
||||
if err != nil {
|
||||
slog.Info("bearer token rejected", "error", err)
|
||||
unauthorized(w, "invalid token")
|
||||
return
|
||||
}
|
||||
if len(b.audiences) > 0 && !slices.ContainsFunc(tok.Audience, func(a string) bool { return slices.Contains(b.audiences, a) }) {
|
||||
slog.Info("bearer token rejected", "error", "audience not allowed", "aud", tok.Audience)
|
||||
unauthorized(w, "invalid token")
|
||||
return
|
||||
}
|
||||
slog.Info("bearer token accepted", "sub", tok.Subject, "method", r.Method, "path", r.URL.Path)
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
// RequireWrites gates only mutating methods.
|
||||
func (b *BearerOIDC) RequireWrites(next http.Handler) http.Handler {
|
||||
gated := b.Require(next)
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.Method {
|
||||
case http.MethodGet, http.MethodHead, http.MethodOptions:
|
||||
next.ServeHTTP(w, r)
|
||||
default:
|
||||
gated.ServeHTTP(w, r)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func unauthorized(w http.ResponseWriter, msg string) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.Header().Set("WWW-Authenticate", "Bearer")
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
_, _ = w.Write([]byte(`{"error":"` + msg + `"}`))
|
||||
}
|
||||
@@ -1,255 +0,0 @@
|
||||
package auth
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/rsa"
|
||||
"encoding/base64"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/go-oidc/v3/oidc"
|
||||
"github.com/go-jose/go-jose/v4"
|
||||
"github.com/go-jose/go-jose/v4/jwt"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/testsupport"
|
||||
)
|
||||
|
||||
func serve(h http.Handler, method, token string) int {
|
||||
if token == "" {
|
||||
return serveAuth(h, method, "")
|
||||
}
|
||||
return serveAuth(h, method, "Bearer "+token)
|
||||
}
|
||||
|
||||
func serveAuth(h http.Handler, method, authz string) int {
|
||||
r := httptest.NewRequest(method, "/x", nil)
|
||||
if authz != "" {
|
||||
r.Header.Set("Authorization", authz)
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, r)
|
||||
return w.Code
|
||||
}
|
||||
|
||||
func claims(iss string) jwt.Claims {
|
||||
now := time.Now()
|
||||
return jwt.Claims{
|
||||
Issuer: iss,
|
||||
Subject: "ci",
|
||||
Audience: jwt.Audience{"artifactapi"},
|
||||
IssuedAt: jwt.NewNumericDate(now.Add(-time.Minute)),
|
||||
Expiry: jwt.NewNumericDate(now.Add(time.Hour)),
|
||||
}
|
||||
}
|
||||
|
||||
func hs256Token(t *testing.T, c jwt.Claims) string {
|
||||
sig, err := jose.NewSigner(jose.SigningKey{Algorithm: jose.HS256, Key: []byte("0123456789abcdef0123456789abcdef")},
|
||||
(&jose.SignerOptions{}).WithType("JWT").WithHeader("kid", "k1"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tok, err := jwt.Signed(sig).Claims(c).Serialize()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return tok
|
||||
}
|
||||
|
||||
func noneToken(iss string) string {
|
||||
enc := base64.RawURLEncoding.EncodeToString
|
||||
return enc([]byte(`{"alg":"none","typ":"JWT","kid":"k1"}`)) + "." +
|
||||
enc([]byte(`{"iss":"`+iss+`","sub":"ci","aud":"artifactapi","exp":`+strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10)+`}`)) + "."
|
||||
}
|
||||
|
||||
var ok = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) })
|
||||
|
||||
func TestBearerOIDC(t *testing.T) {
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
defer iss.Close()
|
||||
h := NewBearerOIDC(iss.URL, []string{"other", "artifactapi"}).Require(ok)
|
||||
|
||||
otherKey, _ := rsa.GenerateKey(rand.Reader, 2048)
|
||||
cases := []struct {
|
||||
name string
|
||||
token string
|
||||
want int
|
||||
}{
|
||||
{"valid", iss.Token(nil, "artifactapi", "ci", time.Hour), http.StatusNoContent},
|
||||
{"missing", "", http.StatusUnauthorized},
|
||||
{"garbage", "not.a.jwt", http.StatusUnauthorized},
|
||||
{"bad signature", iss.Token(otherKey, "artifactapi", "ci", time.Hour), http.StatusUnauthorized},
|
||||
{"wrong aud", iss.Token(nil, "nope", "ci", time.Hour), http.StatusUnauthorized},
|
||||
{"expired", iss.Token(nil, "artifactapi", "ci", -time.Hour), http.StatusUnauthorized},
|
||||
{"wrong iss", iss.Sign(nil, claims("https://evil.example")), http.StatusUnauthorized},
|
||||
{"hs256", hs256Token(t, claims(iss.URL)), http.StatusUnauthorized},
|
||||
{"alg none", noneToken(iss.URL), http.StatusUnauthorized},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := serve(h, http.MethodPost, c.token); got != c.want {
|
||||
t.Errorf("%s: status %d, want %d", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
|
||||
valid := iss.Token(nil, "artifactapi", "ci", time.Hour)
|
||||
headers := []struct {
|
||||
name string
|
||||
authz string
|
||||
want int
|
||||
}{
|
||||
{"lowercase scheme", "bearer " + valid, http.StatusNoContent},
|
||||
{"extra whitespace", " Bearer " + valid + " ", http.StatusNoContent},
|
||||
{"empty value", "Bearer ", http.StatusUnauthorized},
|
||||
{"scheme only", "Bearer", http.StatusUnauthorized},
|
||||
{"basic scheme", "Basic " + valid, http.StatusUnauthorized},
|
||||
}
|
||||
for _, c := range headers {
|
||||
if got := serveAuth(h, http.MethodPost, c.authz); got != c.want {
|
||||
t.Errorf("%s: status %d, want %d", c.name, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCHangingIssuerBacksOff(t *testing.T) {
|
||||
var hits atomic.Int32
|
||||
release := make(chan struct{})
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hits.Add(1)
|
||||
select {
|
||||
case <-r.Context().Done():
|
||||
case <-release:
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
defer close(release)
|
||||
|
||||
b := NewBearerOIDC(srv.URL, nil)
|
||||
b.timeout = 100 * time.Millisecond
|
||||
b.backoff = time.Hour
|
||||
h := b.Require(ok)
|
||||
|
||||
start := time.Now()
|
||||
if got := serve(h, http.MethodPost, "x.y.z"); got != http.StatusServiceUnavailable {
|
||||
t.Fatalf("hanging issuer: status %d, want 503", got)
|
||||
}
|
||||
if d := time.Since(start); d > 2*time.Second {
|
||||
t.Fatalf("hanging issuer: took %s, timeout not applied", d)
|
||||
}
|
||||
if got := serve(h, http.MethodPost, "x.y.z"); got != http.StatusServiceUnavailable {
|
||||
t.Fatalf("during backoff: status %d, want 503", got)
|
||||
}
|
||||
if n := hits.Load(); n != 1 {
|
||||
t.Fatalf("issuer hit %d times, want 1", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCPanickingDiscoveryBacksOff(t *testing.T) {
|
||||
b := NewBearerOIDC("http://issuer.invalid", nil)
|
||||
b.backoff = time.Hour
|
||||
b.discover = func(context.Context, string) (*oidc.Provider, error) { panic("boom") }
|
||||
h := b.Require(ok)
|
||||
|
||||
func() {
|
||||
defer func() { _ = recover() }()
|
||||
serve(h, http.MethodPost, "x.y.z")
|
||||
}()
|
||||
|
||||
got := make(chan int, 1)
|
||||
go func() { got <- serve(h, http.MethodPost, "x.y.z") }()
|
||||
select {
|
||||
case code := <-got:
|
||||
if code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("after panic: status %d, want 503", code)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("after panic: request hung on stale inflight discovery")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCConcurrentFirstRequestsWaitForDiscovery(t *testing.T) {
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
defer iss.Close()
|
||||
var hits atomic.Int32
|
||||
inner := iss.Config.Handler
|
||||
iss.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/.well-known/openid-configuration" {
|
||||
hits.Add(1)
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
inner.ServeHTTP(w, r)
|
||||
})
|
||||
|
||||
h := NewBearerOIDC(iss.URL, nil).Require(ok)
|
||||
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
|
||||
var wg sync.WaitGroup
|
||||
codes := make([]int, 20)
|
||||
for i := range codes {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
codes[i] = serve(h, http.MethodPost, tok)
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
for i, c := range codes {
|
||||
if c != http.StatusNoContent {
|
||||
t.Errorf("request %d: status %d, want 204", i, c)
|
||||
}
|
||||
}
|
||||
if n := hits.Load(); n != 1 {
|
||||
t.Fatalf("discovery hit %d times, want 1", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCRetriesAfterBackoff(t *testing.T) {
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
defer iss.Close()
|
||||
b := NewBearerOIDC(iss.URL, nil)
|
||||
b.failedAt = time.Now()
|
||||
b.backoff = 50 * time.Millisecond
|
||||
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
|
||||
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusServiceUnavailable {
|
||||
t.Fatalf("within backoff: status %d, want 503", got)
|
||||
}
|
||||
time.Sleep(60 * time.Millisecond)
|
||||
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusNoContent {
|
||||
t.Fatalf("after backoff: status %d, want 204", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCIssuerDown(t *testing.T) {
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
b := NewBearerOIDC(iss.URL, []string{"artifactapi"})
|
||||
tok := iss.Token(nil, "artifactapi", "ci", time.Hour)
|
||||
iss.Close()
|
||||
if got := serve(b.Require(ok), http.MethodPost, tok); got != http.StatusServiceUnavailable {
|
||||
t.Errorf("issuer down: status %d, want 503", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBearerOIDCDisabled(t *testing.T) {
|
||||
if got := serve(NewBearerOIDC("", nil).RequireWrites(ok), http.MethodDelete, ""); got != http.StatusNoContent {
|
||||
t.Errorf("disabled: status %d, want 204", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireWritesLeavesReadsOpen(t *testing.T) {
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
defer iss.Close()
|
||||
h := NewBearerOIDC(iss.URL, []string{"artifactapi"}).RequireWrites(ok)
|
||||
for _, m := range []string{http.MethodGet, http.MethodHead, http.MethodOptions} {
|
||||
if got := serve(h, m, ""); got != http.StatusNoContent {
|
||||
t.Errorf("%s: status %d, want 204", m, got)
|
||||
}
|
||||
}
|
||||
for _, m := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} {
|
||||
if got := serve(h, m, ""); got != http.StatusUnauthorized {
|
||||
t.Errorf("%s: status %d, want 401", m, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
@@ -57,10 +56,6 @@ type Config struct {
|
||||
GitHubAppInstallationID string
|
||||
GitHubAppPrivateKey string
|
||||
GitHubAppPrivateKeyPath string
|
||||
|
||||
// Bearer-JWT auth on configuration writes; empty issuer disables it.
|
||||
OIDCIssuer string
|
||||
OIDCAudiences []string
|
||||
}
|
||||
|
||||
func (c *Config) DatabaseDSN() string {
|
||||
@@ -95,17 +90,6 @@ func Load() (*Config, error) {
|
||||
return nil, fmt.Errorf("invalid GITHUB_SYNC_POLL_INTERVAL: %w", err)
|
||||
}
|
||||
|
||||
oidcIssuer := getenv("ARTIFACTAPI_OIDC_ISSUER", "")
|
||||
var oidcAudiences []string
|
||||
for _, a := range strings.Split(getenv("ARTIFACTAPI_OIDC_AUDIENCE", ""), ",") {
|
||||
if a = strings.TrimSpace(a); a != "" {
|
||||
oidcAudiences = append(oidcAudiences, a)
|
||||
}
|
||||
}
|
||||
if oidcIssuer != "" && len(oidcAudiences) == 0 {
|
||||
return nil, fmt.Errorf("ARTIFACTAPI_OIDC_AUDIENCE is required when ARTIFACTAPI_OIDC_ISSUER is set")
|
||||
}
|
||||
|
||||
cfg := &Config{
|
||||
ListenAddr: getenv("LISTEN_ADDR", ":8000"),
|
||||
|
||||
@@ -139,9 +123,6 @@ func Load() (*Config, error) {
|
||||
GitHubAppInstallationID: getenv("GITHUB_APP_INSTALLATION_ID", ""),
|
||||
GitHubAppPrivateKey: getenv("GITHUB_APP_PRIVATE_KEY", ""),
|
||||
GitHubAppPrivateKeyPath: getenv("GITHUB_APP_PRIVATE_KEY_PATH", ""),
|
||||
|
||||
OIDCIssuer: oidcIssuer,
|
||||
OIDCAudiences: oidcAudiences,
|
||||
}
|
||||
|
||||
return cfg, nil
|
||||
|
||||
@@ -64,20 +64,3 @@ func TestLoadInvalidPort(t *testing.T) {
|
||||
t.Error("expected error for invalid DBPORT")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadOIDC(t *testing.T) {
|
||||
t.Setenv("ARTIFACTAPI_OIDC_ISSUER", "https://vault.example/v1/identity/oidc")
|
||||
t.Setenv("ARTIFACTAPI_OIDC_AUDIENCE", " a, ,b ")
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
if len(cfg.OIDCAudiences) != 2 || cfg.OIDCAudiences[0] != "a" || cfg.OIDCAudiences[1] != "b" {
|
||||
t.Errorf("audiences = %q", cfg.OIDCAudiences)
|
||||
}
|
||||
|
||||
t.Setenv("ARTIFACTAPI_OIDC_AUDIENCE", "")
|
||||
if _, err := Load(); err == nil {
|
||||
t.Error("expected error for issuer without audience")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -207,3 +207,72 @@ func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
|
||||
t.Fatal("claimed inside freshness window after a successful sync")
|
||||
}
|
||||
}
|
||||
|
||||
// Deleting a remote drops its sync state, so a recreated remote of the same name
|
||||
// starts clean instead of inheriting the old ETag, sync time and retry state.
|
||||
func TestGitHubSyncStateDeletedWithRemote(t *testing.T) {
|
||||
requireDB(t)
|
||||
for _, tc := range []struct {
|
||||
table string
|
||||
pt models.PackageType
|
||||
claim func(name string) (bool, string, error)
|
||||
release func(name string, res provider.SyncResult) error
|
||||
}{
|
||||
{"github_rpm_sync_state", models.PackageGitHubRPM,
|
||||
func(n string) (bool, string, error) {
|
||||
return testDB.ClaimGitHubSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
|
||||
},
|
||||
func(n string, res provider.SyncResult) error {
|
||||
return testDB.ReleaseGitHubSyncLease(ctx(), n, "r1", res)
|
||||
}},
|
||||
{"github_deb_sync_state", models.PackageGitHubDeb,
|
||||
func(n string) (bool, string, error) {
|
||||
return testDB.ClaimGitHubDebSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
|
||||
},
|
||||
func(n string, res provider.SyncResult) error {
|
||||
return testDB.ReleaseGitHubDebSyncLease(ctx(), n, "r1", res)
|
||||
}},
|
||||
{"github_alpine_sync_state", models.PackageGitHubAlpine,
|
||||
func(n string) (bool, string, error) {
|
||||
return testDB.ClaimGitHubAlpineSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
|
||||
},
|
||||
func(n string, res provider.SyncResult) error {
|
||||
return testDB.ReleaseGitHubAlpineSyncLease(ctx(), n, "r1", res)
|
||||
}},
|
||||
} {
|
||||
t.Run(tc.table, func(t *testing.T) {
|
||||
name := "gh-recreate-" + time.Now().Format("150405.000000")
|
||||
create := func() {
|
||||
t.Helper()
|
||||
if err := testDB.CreateRemote(ctx(), &models.Remote{
|
||||
Name: name, PackageType: tc.pt, RepoType: models.RepoTypeRemote,
|
||||
BaseURL: "https://api.github.com/repos/acme/tools", ReleasesRemote: "github", MutableTTL: 3600,
|
||||
}); err != nil {
|
||||
t.Fatalf("create remote: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
create()
|
||||
if ok, _, err := tc.claim(name); err != nil || !ok {
|
||||
t.Fatalf("claim: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if err := tc.release(name, provider.SyncResult{Etag: `"old"`}); err != nil {
|
||||
t.Fatalf("release: %v", err)
|
||||
}
|
||||
|
||||
if err := testDB.DeleteRemote(ctx(), name); err != nil {
|
||||
t.Fatalf("delete remote: %v", err)
|
||||
}
|
||||
var rows int
|
||||
if err := testDB.Pool.QueryRow(ctx(), `SELECT count(*) FROM `+tc.table+` WHERE remote_name = $1`, name).Scan(&rows); err != nil || rows != 0 {
|
||||
t.Fatalf("sync state left after delete: rows=%d err=%v", rows, err)
|
||||
}
|
||||
|
||||
create()
|
||||
ok, etag, err := tc.claim(name)
|
||||
if err != nil || !ok || etag != "" {
|
||||
t.Fatalf("recreated remote claim ok=%v etag=%q err=%v, want a fresh claim with no etag", ok, etag, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -372,6 +372,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
return newEtag, false, err
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan so the ETag is not advanced and
|
||||
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
|
||||
var failed []error
|
||||
seen := map[string]bool{}
|
||||
for _, rel := range releases {
|
||||
if rel.Draft {
|
||||
@@ -400,10 +403,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
meta, err := p.deriveAsset(ctx, remote, asset, fp)
|
||||
if err != nil {
|
||||
slog.Warn("github_alpine: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
if !errors.Is(err, provider.ErrInvalidPackage) {
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := inserter.InsertAlpineMetadata(ctx, meta); err != nil {
|
||||
slog.Error("github_alpine: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
continue
|
||||
}
|
||||
slog.Info("github_alpine: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch)
|
||||
@@ -415,7 +422,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
_ = deleter.DeleteAlpineMetadata(ctx, remote.Name, fp)
|
||||
}
|
||||
}
|
||||
return newEtag, true, nil
|
||||
return newEtag, true, errors.Join(failed...)
|
||||
}
|
||||
|
||||
type ghRelease struct {
|
||||
@@ -496,7 +503,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
|
||||
return nil, err
|
||||
}
|
||||
if meta.Name == "" || meta.Arch == "" {
|
||||
return nil, errors.New(".PKGINFO missing pkgname/arch")
|
||||
return nil, fmt.Errorf("%w: .PKGINFO missing pkgname/arch", provider.ErrInvalidPackage)
|
||||
}
|
||||
|
||||
meta.RepoName = remote.Name
|
||||
@@ -526,13 +533,13 @@ func (p *GitHubProvider) fetchPkginfo(ctx context.Context, remote models.Remote,
|
||||
}
|
||||
meta, complete, perr := pkginfoFromPrefix(body)
|
||||
if perr != nil {
|
||||
return nil, fmt.Errorf("parse apk .PKGINFO: %w", perr)
|
||||
return nil, fmt.Errorf("%w: parse apk .PKGINFO: %w", provider.ErrInvalidPackage, perr)
|
||||
}
|
||||
if complete {
|
||||
return meta, nil
|
||||
}
|
||||
if full || n >= p.headerMax {
|
||||
return nil, fmt.Errorf(".PKGINFO not found within %d bytes of %s", n, downloadURL)
|
||||
return nil, fmt.Errorf("%w: .PKGINFO not found within %d bytes of %s", provider.ErrInvalidPackage, n, downloadURL)
|
||||
}
|
||||
n *= 2
|
||||
if n > p.headerMax {
|
||||
@@ -594,7 +601,7 @@ func pkginfoFromPrefix(prefix []byte) (meta *provider.AlpineMetadata, complete b
|
||||
}
|
||||
|
||||
// rangeGet returns the first n bytes of downloadURL. full is true when the
|
||||
// response body was shorter than n (i.e. we already have the whole object).
|
||||
// response holds the whole object.
|
||||
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
|
||||
if err != nil {
|
||||
@@ -616,15 +623,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
||||
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode)
|
||||
return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
full := int64(len(body)) < n
|
||||
return body, full, nil
|
||||
return provider.ReadPrefix(resp, n)
|
||||
}
|
||||
|
||||
// assetPath is the package's location relative to github.com — the path the
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -88,15 +89,17 @@ type githubFixture struct {
|
||||
notModHit int
|
||||
releaseAuth string
|
||||
assetAuth string
|
||||
assetFail map[string]int // asset filename -> downloads left to answer 500
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
t.Helper()
|
||||
f := &githubFixture{
|
||||
apkBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
apkBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
assetFail: map[string]int{},
|
||||
}
|
||||
f.apkBytes["demo-1.2.3-r0.apk"] = testsupport.MinimalApk("demo", "1.2.3-r0", "x86_64")
|
||||
|
||||
@@ -146,6 +149,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
}
|
||||
rng := r.Header.Get("Range")
|
||||
f.mu.Lock()
|
||||
if f.assetFail[name] > 0 {
|
||||
f.assetFail[name]--
|
||||
f.mu.Unlock()
|
||||
http.Error(w, "boom", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
f.assetAuth = r.Header.Get("Authorization")
|
||||
if rng != "" {
|
||||
f.rangeHit[name]++
|
||||
@@ -495,3 +504,13 @@ func readAPKIndex(t *testing.T, gzBytes []byte) string {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGitHubDeriveAssetMissingPkgnameIsInvalid(t *testing.T) {
|
||||
fx := newGitHubFixture(t, true)
|
||||
fx.apkBytes["nameless-1-r0.apk"] = testsupport.MinimalApk("", "1-r0", "x86_64")
|
||||
asset := ghAsset{BrowserDownloadURL: fx.srv.URL + "/acme/tools/releases/download/v1.2.3/nameless-1-r0.apk"}
|
||||
_, err := newTestProvider().deriveAsset(context.Background(), fx.remote(), asset, "fp")
|
||||
if !errors.Is(err, provider.ErrInvalidPackage) {
|
||||
t.Fatalf("want ErrInvalidPackage, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
|
||||
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
|
||||
}
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
|
||||
// is scheduled, and the retry against the unchanged release list derives it. An
|
||||
// invalid package is skipped instead, since no retry can fix it.
|
||||
func TestSyncerFailedAssetRetry(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
corrupt bool
|
||||
wantRetry bool
|
||||
}{
|
||||
{name: "transient", wantRetry: true},
|
||||
{name: "invalid package", corrupt: true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
fx := newGitHubFixture(t, true)
|
||||
fx.etag = `"v1"`
|
||||
if tc.corrupt {
|
||||
fx.apkBytes["other-9-r0.apk"] = []byte("not a package")
|
||||
} else {
|
||||
fx.apkBytes["other-9-r0.apk"] = testsupport.MinimalApk("other", "9-r0", "aarch64")
|
||||
fx.assetFail["other-9-r0.apk"] = 1
|
||||
}
|
||||
store := newFakeSyncStore()
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
remote := fx.remote()
|
||||
bg := context.Background()
|
||||
|
||||
s.process(bg, syncJob{remote: remote, prime: true})
|
||||
_, pending := store.retryAt[remote.Name]
|
||||
if pending != tc.wantRetry {
|
||||
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
|
||||
}
|
||||
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
|
||||
t.Fatalf("etag saved=%v after first scan", saved)
|
||||
}
|
||||
if rows, _ := store.ListAlpineMetadataEntries(bg, remote.Name); len(rows) != 1 {
|
||||
t.Fatalf("healthy asset not served: %d rows", len(rows))
|
||||
}
|
||||
if !tc.wantRetry {
|
||||
return
|
||||
}
|
||||
|
||||
store.mu.Lock()
|
||||
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
|
||||
store.mu.Unlock()
|
||||
s.process(bg, syncJob{remote: remote})
|
||||
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
|
||||
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
|
||||
}
|
||||
if rows, _ := store.ListAlpineMetadataEntries(bg, remote.Name); len(rows) != 2 {
|
||||
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -333,6 +333,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
return newEtag, false, err
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan so the ETag is not advanced and
|
||||
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
|
||||
var failed []error
|
||||
seen := map[string]bool{}
|
||||
for _, rel := range releases {
|
||||
if rel.Draft {
|
||||
@@ -361,10 +364,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
meta, err := p.deriveAsset(ctx, remote, asset, fp)
|
||||
if err != nil {
|
||||
slog.Warn("github_deb: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
if !errors.Is(err, provider.ErrInvalidPackage) {
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := store.InsertDebMetadata(ctx, meta); err != nil {
|
||||
slog.Error("github_deb: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
continue
|
||||
}
|
||||
slog.Info("github_deb: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Architecture)
|
||||
@@ -376,7 +383,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
_ = store.DeleteDebMetadata(ctx, remote.Name, fp)
|
||||
}
|
||||
}
|
||||
return newEtag, true, nil
|
||||
return newEtag, true, errors.Join(failed...)
|
||||
}
|
||||
|
||||
type ghRelease struct {
|
||||
@@ -468,7 +475,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
|
||||
Size: asset.Size,
|
||||
}
|
||||
if meta.Name == "" {
|
||||
return nil, errors.New("control missing Package field")
|
||||
return nil, fmt.Errorf("%w: control missing Package field", provider.ErrInvalidPackage)
|
||||
}
|
||||
|
||||
// The Packages SHA256 must be the sha256 of the whole .deb. Prefer GitHub's
|
||||
@@ -503,13 +510,13 @@ func (p *GitHubProvider) fetchControl(ctx context.Context, remote models.Remote,
|
||||
}
|
||||
control, complete, perr := controlFromPrefix(body)
|
||||
if perr != nil {
|
||||
return "", fmt.Errorf("parse deb control: %w", perr)
|
||||
return "", fmt.Errorf("%w: parse deb control: %w", provider.ErrInvalidPackage, perr)
|
||||
}
|
||||
if complete {
|
||||
return control, nil
|
||||
}
|
||||
if full || n >= p.headerMax {
|
||||
return "", fmt.Errorf("control.tar not found within %d bytes of %s", n, downloadURL)
|
||||
return "", fmt.Errorf("%w: control.tar not found within %d bytes of %s", provider.ErrInvalidPackage, n, downloadURL)
|
||||
}
|
||||
n *= 2
|
||||
if n > p.headerMax {
|
||||
@@ -567,7 +574,7 @@ func controlFromPrefix(prefix []byte) (control string, complete bool, err error)
|
||||
}
|
||||
|
||||
// rangeGet returns the first n bytes of downloadURL. full is true when the
|
||||
// response body was shorter than n (i.e. we already have the whole object).
|
||||
// response holds the whole object.
|
||||
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
|
||||
if err != nil {
|
||||
@@ -589,15 +596,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
||||
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode)
|
||||
return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
full := int64(len(body)) < n
|
||||
return body, full, nil
|
||||
return provider.ReadPrefix(resp, n)
|
||||
}
|
||||
|
||||
func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) {
|
||||
@@ -620,7 +621,7 @@ func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("GET %s: status %d", downloadURL, resp.StatusCode)
|
||||
return "", fmt.Errorf("GET: %w", provider.AssetStatusError(downloadURL, resp))
|
||||
}
|
||||
|
||||
h := sha256.New()
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -81,15 +82,17 @@ type githubFixture struct {
|
||||
notModHit int
|
||||
releaseAuth string
|
||||
assetAuth string
|
||||
assetFail map[string]int // asset filename -> downloads left to answer 500
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
t.Helper()
|
||||
f := &githubFixture{
|
||||
debBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
debBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
assetFail: map[string]int{},
|
||||
}
|
||||
f.debBytes["demo_1.2-3_amd64.deb"] = testsupport.MinimalDeb("demo", "1.2-3", "amd64")
|
||||
|
||||
@@ -139,6 +142,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
}
|
||||
rng := r.Header.Get("Range")
|
||||
f.mu.Lock()
|
||||
if f.assetFail[name] > 0 {
|
||||
f.assetFail[name]--
|
||||
f.mu.Unlock()
|
||||
http.Error(w, "boom", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
f.assetAuth = r.Header.Get("Authorization")
|
||||
if rng != "" {
|
||||
f.rangeHit[name]++
|
||||
@@ -460,3 +469,13 @@ func TestGitHubAssetPatternFilter(t *testing.T) {
|
||||
t.Fatalf("pattern filter failed, rows=%+v", rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGitHubDeriveAssetMissingPackageIsInvalid(t *testing.T) {
|
||||
fx := newGitHubFixture(t, true)
|
||||
fx.debBytes["nameless_1_amd64.deb"] = testsupport.MinimalDeb("", "1", "amd64")
|
||||
asset := ghAsset{BrowserDownloadURL: fx.srv.URL + "/acme/tools/releases/download/v1.2-3/nameless_1_amd64.deb", Digest: "sha256:00"}
|
||||
_, err := newTestProvider().deriveAsset(context.Background(), fx.remote(), asset, "fp")
|
||||
if !errors.Is(err, provider.ErrInvalidPackage) {
|
||||
t.Fatalf("want ErrInvalidPackage, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
|
||||
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
|
||||
}
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
|
||||
// is scheduled, and the retry against the unchanged release list derives it. An
|
||||
// invalid package is skipped instead, since no retry can fix it.
|
||||
func TestSyncerFailedAssetRetry(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
corrupt bool
|
||||
wantRetry bool
|
||||
}{
|
||||
{name: "transient", wantRetry: true},
|
||||
{name: "invalid package", corrupt: true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
fx := newGitHubFixture(t, true)
|
||||
fx.etag = `"v1"`
|
||||
if tc.corrupt {
|
||||
fx.debBytes["other_9-9_arm64.deb"] = []byte("not a package")
|
||||
} else {
|
||||
fx.debBytes["other_9-9_arm64.deb"] = testsupport.MinimalDeb("other", "9-9", "arm64")
|
||||
fx.assetFail["other_9-9_arm64.deb"] = 1
|
||||
}
|
||||
store := newFakeSyncStore()
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
remote := fx.remote()
|
||||
bg := context.Background()
|
||||
|
||||
s.process(bg, syncJob{remote: remote, prime: true})
|
||||
_, pending := store.retryAt[remote.Name]
|
||||
if pending != tc.wantRetry {
|
||||
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
|
||||
}
|
||||
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
|
||||
t.Fatalf("etag saved=%v after first scan", saved)
|
||||
}
|
||||
if rows, _ := store.ListDebMetadataEntries(bg, remote.Name); len(rows) != 1 {
|
||||
t.Fatalf("healthy asset not served: %d rows", len(rows))
|
||||
}
|
||||
if !tc.wantRetry {
|
||||
return
|
||||
}
|
||||
|
||||
store.mu.Lock()
|
||||
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
|
||||
store.mu.Unlock()
|
||||
s.process(bg, syncJob{remote: remote})
|
||||
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
|
||||
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
|
||||
}
|
||||
if rows, _ := store.ListDebMetadataEntries(bg, remote.Name); len(rows) != 2 {
|
||||
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -343,6 +343,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
return newEtag, false, err
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan so the ETag is not advanced and
|
||||
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
|
||||
var failed []error
|
||||
seen := map[string]bool{}
|
||||
for _, rel := range releases {
|
||||
if rel.Draft {
|
||||
@@ -373,10 +376,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
meta, err := p.deriveAsset(ctx, remote, asset, fp)
|
||||
if err != nil {
|
||||
slog.Warn("github_rpm: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
if !errors.Is(err, provider.ErrInvalidPackage) {
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := store.InsertRPMMetadata(ctx, meta); err != nil {
|
||||
slog.Error("github_rpm: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
|
||||
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
|
||||
continue
|
||||
}
|
||||
slog.Info("github_rpm: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch)
|
||||
@@ -388,7 +395,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
|
||||
_ = store.DeleteRPMMetadata(ctx, remote.Name, fp)
|
||||
}
|
||||
}
|
||||
return newEtag, true, nil
|
||||
return newEtag, true, errors.Join(failed...)
|
||||
}
|
||||
|
||||
type ghRelease struct {
|
||||
@@ -569,12 +576,12 @@ func (p *GitHubProvider) fetchHeader(ctx context.Context, remote models.Remote,
|
||||
}
|
||||
continue
|
||||
}
|
||||
return nil, fmt.Errorf("parse rpm header: %w", perr)
|
||||
return nil, fmt.Errorf("%w: parse rpm header: %w", provider.ErrInvalidPackage, perr)
|
||||
}
|
||||
}
|
||||
|
||||
// rangeGet returns the first n bytes of downloadURL. full is true when the
|
||||
// response body was shorter than n (i.e. we already have the whole object).
|
||||
// response holds the whole object.
|
||||
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
|
||||
if err != nil {
|
||||
@@ -596,15 +603,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
||||
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode)
|
||||
return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
full := int64(len(body)) < n
|
||||
return body, full, nil
|
||||
return provider.ReadPrefix(resp, n)
|
||||
}
|
||||
|
||||
func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) {
|
||||
@@ -627,7 +628,7 @@ func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("GET %s: status %d", downloadURL, resp.StatusCode)
|
||||
return "", fmt.Errorf("GET: %w", provider.AssetStatusError(downloadURL, resp))
|
||||
}
|
||||
|
||||
h := sha256.New()
|
||||
|
||||
@@ -1,11 +1,13 @@
|
||||
package rpm
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -24,15 +26,22 @@ import (
|
||||
// fakeStore is an in-memory provider.RemoteMetadataStore keyed by file_path,
|
||||
// mirroring the (repo_name, file_path) uniqueness of the real table.
|
||||
type fakeStore struct {
|
||||
mu sync.Mutex
|
||||
rows map[string]provider.RPMMetadata
|
||||
mu sync.Mutex
|
||||
rows map[string]provider.RPMMetadata
|
||||
insertFail map[string]int // package name -> inserts left to fail
|
||||
}
|
||||
|
||||
func newFakeStore() *fakeStore { return &fakeStore{rows: map[string]provider.RPMMetadata{}} }
|
||||
func newFakeStore() *fakeStore {
|
||||
return &fakeStore{rows: map[string]provider.RPMMetadata{}, insertFail: map[string]int{}}
|
||||
}
|
||||
|
||||
func (f *fakeStore) InsertRPMMetadata(_ context.Context, m *provider.RPMMetadata) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if f.insertFail[m.Name] > 0 {
|
||||
f.insertFail[m.Name]--
|
||||
return errors.New("insert failed")
|
||||
}
|
||||
if _, ok := f.rows[m.FilePath]; ok {
|
||||
return nil // ON CONFLICT DO NOTHING
|
||||
}
|
||||
@@ -79,15 +88,20 @@ type githubFixture struct {
|
||||
releaseAuth string // Authorization header seen on the last releases request
|
||||
assetAuth string // Authorization header seen on the last asset request
|
||||
failStatus int // when set, the releases list answers this status (rate-limit style)
|
||||
assetFail map[string]int // asset filename -> downloads left to answer failCode
|
||||
failCode int // status for assetFail downloads; 0 = 500
|
||||
shortRead map[string]int // asset filename -> ranged GETs left to close early
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
t.Helper()
|
||||
f := &githubFixture{
|
||||
rpmBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
rpmBytes: map[string][]byte{},
|
||||
rangeHit: map[string]int{},
|
||||
fullHit: map[string]int{},
|
||||
assetFail: map[string]int{},
|
||||
shortRead: map[string]int{},
|
||||
}
|
||||
f.rpmBytes["demo-1.2-3.x86_64.rpm"] = testsupport.MinimalRPM("demo", "1.2", "3", "x86_64")
|
||||
|
||||
@@ -145,6 +159,17 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
}
|
||||
rng := r.Header.Get("Range")
|
||||
f.mu.Lock()
|
||||
if f.assetFail[name] > 0 {
|
||||
f.assetFail[name]--
|
||||
code := cmp.Or(f.failCode, http.StatusInternalServerError)
|
||||
f.mu.Unlock()
|
||||
http.Error(w, "boom", code)
|
||||
return
|
||||
}
|
||||
short := rng != "" && f.shortRead[name] > 0
|
||||
if short {
|
||||
f.shortRead[name]--
|
||||
}
|
||||
f.assetAuth = r.Header.Get("Authorization")
|
||||
if rng != "" {
|
||||
f.rangeHit[name]++
|
||||
@@ -165,6 +190,9 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
|
||||
end = len(body) - 1
|
||||
}
|
||||
w.Header().Set("Content-Range", fmt.Sprintf("bytes 0-%d/%d", end, len(body)))
|
||||
if short {
|
||||
end /= 2
|
||||
}
|
||||
w.Header().Set("Content-Length", strconv.Itoa(end+1))
|
||||
w.WriteHeader(http.StatusPartialContent)
|
||||
w.Write(body[:end+1])
|
||||
|
||||
@@ -399,3 +399,74 @@ func TestSyncerFailedScanRetriesAfterBackoff(t *testing.T) {
|
||||
t.Fatalf("recovery did not refresh metadata: %d rows", len(rows))
|
||||
}
|
||||
}
|
||||
|
||||
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
|
||||
// is scheduled, and the retry against the unchanged release list derives it. An
|
||||
// invalid package is skipped instead, since no retry can fix it.
|
||||
func TestSyncerFailedAssetRetry(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
corrupt bool
|
||||
failCode int
|
||||
short bool
|
||||
insert bool
|
||||
wantRetry bool
|
||||
}{
|
||||
{name: "transient", wantRetry: true},
|
||||
{name: "rate limited", failCode: http.StatusForbidden, wantRetry: true},
|
||||
{name: "short read", short: true, wantRetry: true},
|
||||
{name: "insert failure", insert: true, wantRetry: true},
|
||||
{name: "invalid package", corrupt: true},
|
||||
{name: "asset not found", failCode: http.StatusNotFound},
|
||||
{name: "asset gone", failCode: http.StatusGone},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
const other = "other-9-9.aarch64.rpm"
|
||||
fx := newGitHubFixture(t, true)
|
||||
fx.etag = `"v1"`
|
||||
fx.rpmBytes[other] = testsupport.MinimalRPM("other", "9", "9", "aarch64")
|
||||
store := newFakeSyncStore()
|
||||
switch {
|
||||
case tc.corrupt:
|
||||
fx.rpmBytes[other] = []byte("not a package")
|
||||
case tc.short:
|
||||
fx.shortRead[other] = 1
|
||||
case tc.insert:
|
||||
store.insertFail["other"] = 1
|
||||
case tc.failCode != 0 && !tc.wantRetry:
|
||||
fx.failCode, fx.assetFail[other] = tc.failCode, 1<<30
|
||||
default:
|
||||
fx.failCode, fx.assetFail[other] = tc.failCode, 1
|
||||
}
|
||||
s := newSyncer(store, newTestProvider(), testSyncConfig())
|
||||
remote := fx.remote()
|
||||
bg := context.Background()
|
||||
|
||||
s.process(bg, syncJob{remote: remote, prime: true})
|
||||
_, pending := store.retryAt[remote.Name]
|
||||
if pending != tc.wantRetry {
|
||||
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
|
||||
}
|
||||
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
|
||||
t.Fatalf("etag saved=%v after first scan", saved)
|
||||
}
|
||||
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 1 {
|
||||
t.Fatalf("healthy asset not served: %d rows", len(rows))
|
||||
}
|
||||
if !tc.wantRetry {
|
||||
return
|
||||
}
|
||||
|
||||
store.mu.Lock()
|
||||
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
|
||||
store.mu.Unlock()
|
||||
s.process(bg, syncJob{remote: remote})
|
||||
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
|
||||
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
|
||||
}
|
||||
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 2 {
|
||||
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package provider
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -43,6 +45,49 @@ func NewUpstreamStatusError(url string, resp *http.Response) *UpstreamStatusErro
|
||||
return e
|
||||
}
|
||||
|
||||
// ErrInvalidPackage marks an asset whose bytes are not a valid package. It is
|
||||
// permanent, so a scan skips the asset instead of failing and retrying.
|
||||
var ErrInvalidPackage = errors.New("invalid package")
|
||||
|
||||
// AssetStatusError classifies a non-success asset download. A missing or
|
||||
// unsatisfiable asset is permanent; anything else (403/429 rate limits, 5xx) is
|
||||
// transient.
|
||||
func AssetStatusError(url string, resp *http.Response) error {
|
||||
switch resp.StatusCode {
|
||||
case http.StatusNotFound, http.StatusGone, http.StatusRequestedRangeNotSatisfiable:
|
||||
return fmt.Errorf("%w: %s: status %d", ErrInvalidPackage, url, resp.StatusCode)
|
||||
}
|
||||
return NewUpstreamStatusError(url, resp)
|
||||
}
|
||||
|
||||
// ReadPrefix reads up to n bytes of a 200 or 206 asset response. full reports
|
||||
// that the body is the whole object (per Content-Length or the Content-Range
|
||||
// total, or a clean EOF before n when the total is unknown); a body ending before n bytes that is not the whole object is a
|
||||
// transient short read.
|
||||
func ReadPrefix(resp *http.Response, n int64) (body []byte, full bool, err error) {
|
||||
body, err = io.ReadAll(io.LimitReader(resp.Body, n))
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
total := resp.ContentLength
|
||||
unknown := resp.StatusCode == http.StatusOK && total == -1
|
||||
if resp.StatusCode == http.StatusPartialContent {
|
||||
total = -1
|
||||
if _, t, ok := strings.Cut(resp.Header.Get("Content-Range"), "/"); ok {
|
||||
if v, perr := strconv.ParseInt(t, 10, 64); perr == nil {
|
||||
total = v
|
||||
}
|
||||
unknown = t == "*"
|
||||
}
|
||||
}
|
||||
// net/http errors on a premature close, so a clean EOF before n with an unknown total is the whole object.
|
||||
full = total == int64(len(body)) || (unknown && int64(len(body)) < n)
|
||||
if !full && int64(len(body)) < n {
|
||||
return nil, false, fmt.Errorf("short read: got %d of %d bytes", len(body), n)
|
||||
}
|
||||
return body, full, nil
|
||||
}
|
||||
|
||||
// SyncResult is a background scan's outcome, recorded when its sync lease is
|
||||
// released. A failed scan keeps the prior sync time and ETag and schedules a
|
||||
// retry after Backoff, doubled per consecutive failure up to MaxBackoff, and
|
||||
|
||||
@@ -3,7 +3,9 @@ package provider
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -61,3 +63,76 @@ func TestNewSyncResult(t *testing.T) {
|
||||
t.Fatalf("wrapped hint not clamped to ttl: retry in %v", until)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAssetStatusErrorClassification(t *testing.T) {
|
||||
for status, permanent := range map[int]bool{
|
||||
http.StatusNotFound: true, http.StatusGone: true, http.StatusRequestedRangeNotSatisfiable: true,
|
||||
http.StatusForbidden: false, http.StatusTooManyRequests: false, http.StatusInternalServerError: false, http.StatusBadGateway: false,
|
||||
} {
|
||||
err := AssetStatusError("u", &http.Response{StatusCode: status, Header: http.Header{}})
|
||||
if errors.Is(err, ErrInvalidPackage) != permanent {
|
||||
t.Errorf("status %d: permanent=%v, want %v", status, !permanent, permanent)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadPrefix(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
status int
|
||||
cl int64
|
||||
crange string
|
||||
body string
|
||||
n int64
|
||||
wantFull bool
|
||||
wantErr bool
|
||||
}{
|
||||
{name: "200 whole object", status: 200, cl: 5, body: "hello", n: 32, wantFull: true},
|
||||
{name: "200 larger than n", status: 200, cl: 5, body: "hello", n: 4},
|
||||
{name: "200 chunked shorter than n", status: 200, cl: -1, body: "hello", n: 32, wantFull: true},
|
||||
{name: "200 chunked larger than n", status: 200, cl: -1, body: "hello", n: 4},
|
||||
{name: "206 whole object", status: 206, crange: "bytes 0-4/5", body: "hello", n: 32, wantFull: true},
|
||||
{name: "206 prefix", status: 206, crange: "bytes 0-3/100", body: "hell", n: 4},
|
||||
{name: "206 short read", status: 206, crange: "bytes 0-31/100", body: "hello", n: 32, wantErr: true},
|
||||
{name: "206 no content-range short", status: 206, body: "hello", n: 32, wantErr: true},
|
||||
{name: "206 unknown total shorter than n", status: 206, cl: -1, crange: "bytes 0-4/*", body: "hello", n: 32, wantFull: true},
|
||||
{name: "206 unknown total prefix", status: 206, cl: -1, crange: "bytes 0-3/*", body: "hell", n: 4},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
if tc.cl > 0 {
|
||||
w.Header().Set("Content-Length", strconv.FormatInt(tc.cl, 10))
|
||||
}
|
||||
if tc.crange != "" {
|
||||
w.Header().Set("Content-Range", tc.crange)
|
||||
}
|
||||
w.WriteHeader(tc.status)
|
||||
_, _ = io.WriteString(w, tc.body)
|
||||
w.(http.Flusher).Flush()
|
||||
}))
|
||||
defer srv.Close()
|
||||
resp, err := http.Get(srv.URL)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if tc.cl == -1 && resp.ContentLength != -1 {
|
||||
t.Fatalf("response not chunked: ContentLength=%d", resp.ContentLength)
|
||||
}
|
||||
body, full, err := ReadPrefix(resp, tc.n)
|
||||
if (err != nil) != tc.wantErr {
|
||||
t.Fatalf("err=%v, wantErr=%v", err, tc.wantErr)
|
||||
}
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrInvalidPackage) {
|
||||
t.Fatalf("short read must be transient: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if full != tc.wantFull || int64(len(body)) > tc.n {
|
||||
t.Fatalf("full=%v len=%d, want full=%v", full, len(body), tc.wantFull)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,7 +12,7 @@ func cors(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*")
|
||||
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
|
||||
w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization")
|
||||
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
|
||||
if r.Method == http.MethodOptions {
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
return
|
||||
|
||||
@@ -1,80 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/internal/testsupport"
|
||||
)
|
||||
|
||||
func TestServerOIDCWriteAuth(t *testing.T) {
|
||||
requireStack(t)
|
||||
iss := testsupport.NewOIDCIssuer()
|
||||
defer iss.Close()
|
||||
|
||||
cfg := *testSrv.cfg
|
||||
cfg.OIDCIssuer = iss.URL
|
||||
cfg.OIDCAudiences = []string{"artifactapi"}
|
||||
s := *testSrv
|
||||
s.cfg = &cfg
|
||||
ts := httptest.NewServer(s.routes())
|
||||
defer ts.Close()
|
||||
|
||||
good := iss.Token(nil, "artifactapi", "ci", time.Hour)
|
||||
do := func(method, path, body, token string) int {
|
||||
t.Helper()
|
||||
rq, _ := http.NewRequest(method, ts.URL+path, strings.NewReader(body))
|
||||
rq.Header.Set("Content-Type", "application/json")
|
||||
if token != "" {
|
||||
rq.Header.Set("Authorization", "Bearer "+token)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(rq)
|
||||
if err != nil {
|
||||
t.Fatalf("%s %s: %v", method, path, err)
|
||||
}
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
_ = resp.Body.Close()
|
||||
return resp.StatusCode
|
||||
}
|
||||
expect := func(want int, method, path, body, token string) {
|
||||
t.Helper()
|
||||
if got := do(method, path, body, token); got != want {
|
||||
t.Errorf("%s %s: status %d, want %d", method, path, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
remote := `{"name":"srv-oidc","package_type":"generic","repo_type":"local"}`
|
||||
virtual := `{"name":"srv-oidc-v","package_type":"generic","members":["srv-oidc"]}`
|
||||
|
||||
expect(401, "POST", "/api/v2/remotes", remote, "")
|
||||
expect(401, "POST", "/api/v2/remotes", remote, iss.Token(nil, "other", "ci", time.Hour))
|
||||
expect(201, "POST", "/api/v2/remotes", remote, good)
|
||||
expect(401, "PUT", "/api/v2/remotes/srv-oidc", `{"package_type":"generic"}`, "")
|
||||
expect(401, "DELETE", "/api/v2/remotes/srv-oidc", "", "")
|
||||
expect(401, "POST", "/api/v2/virtuals", virtual, "")
|
||||
expect(201, "POST", "/api/v2/virtuals", virtual, good)
|
||||
expect(401, "PUT", "/api/v2/virtuals/srv-oidc-v", virtual, "")
|
||||
expect(401, "DELETE", "/api/v2/virtuals/srv-oidc-v", "", "")
|
||||
|
||||
expect(200, "GET", "/api/v2/remotes", "", "")
|
||||
expect(200, "GET", "/api/v2/remotes/srv-oidc", "", "")
|
||||
expect(200, "GET", "/api/v2/virtuals/srv-oidc-v", "", "")
|
||||
expect(201, "PUT", "/api/v2/remotes/srv-oidc/files/a.bin", "payload", "")
|
||||
expect(200, "GET", "/api/v2/remotes/srv-oidc/files/a.bin", "", "")
|
||||
if got := do("DELETE", "/api/v2/remotes/srv-oidc/files/a.bin", "", ""); got >= 400 {
|
||||
t.Errorf("delete file: status %d", got)
|
||||
}
|
||||
if got := do("POST", "/api/v2/probe", `{}`, ""); got == 401 {
|
||||
t.Error("probe gated")
|
||||
}
|
||||
if got := do("GET", "/v2/", "", ""); got == 401 {
|
||||
t.Error("docker /v2 gated")
|
||||
}
|
||||
|
||||
expect(204, "DELETE", "/api/v2/virtuals/srv-oidc-v", "", good)
|
||||
expect(204, "DELETE", "/api/v2/remotes/srv-oidc", "", good)
|
||||
}
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
tfregistry "git.unkin.net/unkin/artifactapi/internal/api/terraform"
|
||||
v1 "git.unkin.net/unkin/artifactapi/internal/api/v1"
|
||||
v2 "git.unkin.net/unkin/artifactapi/internal/api/v2"
|
||||
"git.unkin.net/unkin/artifactapi/internal/auth"
|
||||
"git.unkin.net/unkin/artifactapi/internal/cache"
|
||||
"git.unkin.net/unkin/artifactapi/internal/config"
|
||||
"git.unkin.net/unkin/artifactapi/internal/database"
|
||||
@@ -187,11 +186,10 @@ func (s *Server) routes() chi.Router {
|
||||
statsHandler := v2.NewStatsHandler(s.db)
|
||||
eventsHandler := v2.NewEventsHandler()
|
||||
probeHandler := v2.NewProbeHandler(s.engine, s.db)
|
||||
writeAuth := auth.NewBearerOIDC(s.cfg.OIDCIssuer, s.cfg.OIDCAudiences)
|
||||
|
||||
r.Route("/api/v2", func(r chi.Router) {
|
||||
r.Mount("/remotes", writeAuth.RequireWrites(remotesHandler.Routes()))
|
||||
r.Mount("/virtuals", writeAuth.RequireWrites(virtualsHandler.Routes()))
|
||||
r.Mount("/remotes", remotesHandler.Routes())
|
||||
r.Mount("/virtuals", virtualsHandler.Routes())
|
||||
r.Mount("/health", healthHandler.Routes())
|
||||
r.Mount("/stats", statsHandler.Routes())
|
||||
r.Mount("/events", eventsHandler.Routes())
|
||||
|
||||
@@ -1,73 +0,0 @@
|
||||
package testsupport
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"crypto/rsa"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"time"
|
||||
|
||||
"github.com/go-jose/go-jose/v4"
|
||||
"github.com/go-jose/go-jose/v4/jwt"
|
||||
)
|
||||
|
||||
// OIDCIssuer is a minimal OIDC issuer serving discovery and JWKS, signing
|
||||
// RS256 tokens with a generated key.
|
||||
type OIDCIssuer struct {
|
||||
*httptest.Server
|
||||
key *rsa.PrivateKey
|
||||
}
|
||||
|
||||
func NewOIDCIssuer() *OIDCIssuer {
|
||||
key, err := rsa.GenerateKey(rand.Reader, 2048)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
iss := &OIDCIssuer{key: key}
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, r *http.Request) {
|
||||
_ = json.NewEncoder(w).Encode(map[string]any{
|
||||
"issuer": iss.URL,
|
||||
"jwks_uri": iss.URL + "/keys",
|
||||
"id_token_signing_alg_values_supported": []string{"RS256"},
|
||||
})
|
||||
})
|
||||
mux.HandleFunc("/keys", func(w http.ResponseWriter, r *http.Request) {
|
||||
_ = json.NewEncoder(w).Encode(jose.JSONWebKeySet{Keys: []jose.JSONWebKey{
|
||||
{Key: &key.PublicKey, KeyID: "k1", Algorithm: "RS256", Use: "sig"},
|
||||
}})
|
||||
})
|
||||
iss.Server = httptest.NewServer(mux)
|
||||
return iss
|
||||
}
|
||||
|
||||
// Token signs a token for aud and sub. Pass a nil key to use the issuer's own;
|
||||
// expiry is relative to now.
|
||||
func (o *OIDCIssuer) Token(key *rsa.PrivateKey, aud, sub string, expiry time.Duration) string {
|
||||
now := time.Now()
|
||||
return o.Sign(key, jwt.Claims{
|
||||
Issuer: o.URL,
|
||||
Subject: sub,
|
||||
Audience: jwt.Audience{aud},
|
||||
IssuedAt: jwt.NewNumericDate(now.Add(-time.Minute)),
|
||||
Expiry: jwt.NewNumericDate(now.Add(expiry)),
|
||||
})
|
||||
}
|
||||
|
||||
// Sign signs arbitrary claims with RS256. Pass a nil key to use the issuer's own.
|
||||
func (o *OIDCIssuer) Sign(key *rsa.PrivateKey, claims jwt.Claims) string {
|
||||
if key == nil {
|
||||
key = o.key
|
||||
}
|
||||
sig, err := jose.NewSigner(jose.SigningKey{Algorithm: jose.RS256, Key: key},
|
||||
(&jose.SignerOptions{}).WithType("JWT").WithHeader("kid", "k1"))
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
tok, err := jwt.Signed(sig).Claims(claims).Serialize()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return tok
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Sync state is dropped with its remote, so a recreated remote of the same name
|
||||
-- starts with no inherited ETag or timing. NOT VALID skips checking rows left by
|
||||
-- remotes deleted before this migration, which would otherwise fail it.
|
||||
ALTER TABLE github_rpm_sync_state DROP CONSTRAINT IF EXISTS github_rpm_sync_state_remote_fk;
|
||||
ALTER TABLE github_rpm_sync_state ADD CONSTRAINT github_rpm_sync_state_remote_fk
|
||||
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
|
||||
ALTER TABLE github_deb_sync_state DROP CONSTRAINT IF EXISTS github_deb_sync_state_remote_fk;
|
||||
ALTER TABLE github_deb_sync_state ADD CONSTRAINT github_deb_sync_state_remote_fk
|
||||
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
|
||||
ALTER TABLE github_alpine_sync_state DROP CONSTRAINT IF EXISTS github_alpine_sync_state_remote_fk;
|
||||
ALTER TABLE github_alpine_sync_state ADD CONSTRAINT github_alpine_sync_state_remote_fk
|
||||
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
|
||||
@@ -101,27 +101,6 @@ pip install <package>`,
|
||||
];
|
||||
|
||||
case 'npm':
|
||||
if (isLocal) {
|
||||
const reg = `${url}/api/v1/local/${name}/`;
|
||||
return [
|
||||
{
|
||||
title: 'Configure .npmrc (registry + publish token)',
|
||||
language: 'bash',
|
||||
code: `cat >> .npmrc <<'EOF'
|
||||
registry=${reg}
|
||||
//${host}/api/v1/local/${name}/:_authToken=unused
|
||||
EOF`,
|
||||
note: 'npm refuses to publish without an _authToken for the registry; any value works. For a scope only, use @<scope>:registry=… instead of registry=.',
|
||||
},
|
||||
{
|
||||
title: 'Publish and install',
|
||||
language: 'bash',
|
||||
code: `npm publish
|
||||
npm install <package>`,
|
||||
note: 'Versions are immutable: republishing an existing version is rejected. The dist-tags sent with the publish (latest, or --tag) move to the new version.',
|
||||
},
|
||||
];
|
||||
}
|
||||
return [
|
||||
{
|
||||
title: 'Point npm at this registry',
|
||||
|
||||
Reference in New Issue
Block a user