enforce remote quarantine for fresh immutable artifacts
- refuse uncached immutable paths published under quarantine_days ago with a JSON 403 - take publish time from Last-Modified, else a Redis first-seen key - let goproxy virtual probes skip a quarantining member - default quarantine_days to 3 on create and keep it on update - document quarantine in caching.md
This commit is contained in:
@@ -325,12 +325,12 @@ func (e *Engine) Head(ctx context.Context, remote models.Remote, path string, pr
|
||||
return &HeadResult{ContentType: info.ContentType, Size: info.Size, Source: "cache"}, nil
|
||||
}
|
||||
|
||||
return e.headUpstream(ctx, remote, path, prov)
|
||||
return e.headUpstream(ctx, remote, path, prov, class)
|
||||
}
|
||||
|
||||
// headUpstream issues an upstream HEAD, load-balancing across the remote's base
|
||||
// URLs and failing over to the next mirror on a network error or 5xx.
|
||||
func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
|
||||
func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification) (*HeadResult, error) {
|
||||
order := e.baseURLAttemptOrder(remote)
|
||||
if len(order) == 0 {
|
||||
return nil, &ProxyError{Status: http.StatusBadGateway, Message: "no upstream base_url configured"}
|
||||
@@ -338,7 +338,7 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st
|
||||
var lastErr error
|
||||
for i, url := range order {
|
||||
ctr := e.beginAttempt(remote, url)
|
||||
result, err := e.headUpstreamOnce(ctx, withBaseURL(remote, url), path, prov)
|
||||
result, err := e.headUpstreamOnce(ctx, withBaseURL(remote, url), path, prov, class)
|
||||
endAttempt(ctr)
|
||||
if err == nil {
|
||||
return result, nil
|
||||
@@ -353,7 +353,7 @@ func (e *Engine) headUpstream(ctx context.Context, remote models.Remote, path st
|
||||
return nil, lastErr
|
||||
}
|
||||
|
||||
func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider) (*HeadResult, error) {
|
||||
func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, path string, prov provider.Provider, class Classification) (*HeadResult, error) {
|
||||
url := prov.UpstreamURL(remote, path)
|
||||
|
||||
authHeaders, err := prov.AuthHeaders(ctx, remote)
|
||||
@@ -400,6 +400,11 @@ func (e *Engine) headUpstreamOnce(ctx context.Context, remote models.Remote, pat
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, &ProxyError{Status: resp.StatusCode, Message: fmt.Sprintf("upstream returned %d", resp.StatusCode)}
|
||||
}
|
||||
if class == ClassImmutable {
|
||||
if err := e.checkQuarantine(ctx, remote, path, resp.Header); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
contentType := prov.ContentType(path)
|
||||
if ct := resp.Header.Get("Content-Type"); ct != "" {
|
||||
@@ -527,6 +532,11 @@ func (e *Engine) fetchFromUpstreamOnce(ctx context.Context, remote models.Remote
|
||||
}, nil
|
||||
}
|
||||
|
||||
if err := e.checkQuarantine(ctx, remote, path, resp.Header); err != nil {
|
||||
_ = resp.Body.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Immutable blobs are streamed through the content-addressable store
|
||||
// (tempfile -> sha256 -> S3) so arbitrarily large artifacts never sit
|
||||
// fully in memory. Immutable content is never rewritten in the proxy path.
|
||||
@@ -663,6 +673,35 @@ func (e *Engine) checkUpstreamOnce(ctx context.Context, remote models.Remote, pa
|
||||
return resp.StatusCode == http.StatusNotModified, nil
|
||||
}
|
||||
|
||||
// checkQuarantine refuses an uncached immutable artifact published less than
|
||||
// quarantine_days ago. The publish time is the upstream Last-Modified, else
|
||||
// the first time this path was seen.
|
||||
func (e *Engine) checkQuarantine(ctx context.Context, remote models.Remote, path string, upstream http.Header) error {
|
||||
if !remote.QuarantineEnabled || remote.QuarantineDays <= 0 {
|
||||
return nil
|
||||
}
|
||||
if a, err := e.db.GetArtifact(ctx, remote.Name, path); err == nil && a != nil {
|
||||
return nil
|
||||
}
|
||||
window := time.Duration(remote.QuarantineDays) * 24 * time.Hour
|
||||
published, err := http.ParseTime(upstream.Get("Last-Modified"))
|
||||
if err != nil {
|
||||
if published, err = e.cache.FirstSeen(ctx, remote.Name, path, window+24*time.Hour); err != nil {
|
||||
slog.Warn("first-seen lookup failed, quarantining", "remote", remote.Name, "path", path, "error", err)
|
||||
published = time.Now()
|
||||
}
|
||||
}
|
||||
if until := published.Add(window); time.Now().Before(until) {
|
||||
return &ProxyError{
|
||||
Status: http.StatusForbidden,
|
||||
Message: fmt.Sprintf("quarantined: published %s, available from %s", published.UTC().Format(time.RFC3339), until.UTC().Format(time.RFC3339)),
|
||||
Published: published.UTC(),
|
||||
Available: until.UTC(),
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// etagTTL keeps a mutable path's ETag well past its freshness key so an expired
|
||||
// path revalidates with a conditional HEAD, while idle paths still age out.
|
||||
func etagTTL(ttl time.Duration) time.Duration {
|
||||
@@ -821,10 +860,19 @@ func fetchBearerToken(ctx context.Context, wwwAuth string, remote models.Remote)
|
||||
type ProxyError struct {
|
||||
Status int
|
||||
Message string
|
||||
// Published and Available are set only on a quarantine refusal.
|
||||
Published time.Time
|
||||
Available time.Time
|
||||
}
|
||||
|
||||
func (e *ProxyError) Error() string { return e.Message }
|
||||
|
||||
// Quarantined reports whether err is a quarantine refusal.
|
||||
func Quarantined(err error) bool {
|
||||
var pe *ProxyError
|
||||
return errors.As(err, &pe) && !pe.Published.IsZero()
|
||||
}
|
||||
|
||||
type UpstreamError struct {
|
||||
Err error
|
||||
}
|
||||
|
||||
@@ -227,7 +227,7 @@ func TestHeadUpstreamUsesRemoteTimeout(t *testing.T) {
|
||||
r := models.Remote{Name: "slow-head", PackageType: models.PackageGeneric, BaseURL: slow.URL, UpstreamResponseHeaderTimeout: 1}
|
||||
|
||||
start := time.Now()
|
||||
_, err := (&Engine{}).headUpstreamOnce(context.Background(), r, "blob.bin", prov(t, models.PackageGeneric))
|
||||
_, err := (&Engine{}).headUpstreamOnce(context.Background(), r, "blob.bin", prov(t, models.PackageGeneric), ClassImmutable)
|
||||
if err == nil || time.Since(start) > 2*time.Second {
|
||||
t.Fatalf("HEAD ignored the remote's response header timeout: err=%v after %s", err, time.Since(start))
|
||||
}
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func must[T any](v T, err error) T {
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func quarantineUpstream(t *testing.T) *httptest.Server {
|
||||
t.Helper()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/fresh.tgz":
|
||||
w.Header().Set("Last-Modified", time.Now().Add(-time.Hour).UTC().Format(http.TimeFormat))
|
||||
case "/old.tgz":
|
||||
w.Header().Set("Last-Modified", time.Now().Add(-30*24*time.Hour).UTC().Format(http.TimeFormat))
|
||||
case "/undated.tgz", "/pkg":
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte("body " + r.URL.Path))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
return srv
|
||||
}
|
||||
|
||||
func quarantineRemote(t *testing.T, name string, enabled bool) models.Remote {
|
||||
r := models.Remote{Name: name, PackageType: models.PackageNPM, RepoType: models.RepoTypeRemote,
|
||||
BaseURL: quarantineUpstream(t).URL, QuarantineEnabled: enabled, QuarantineDays: 3}
|
||||
return seed(t, r)
|
||||
}
|
||||
|
||||
func wantQuarantined(t *testing.T, err error) *ProxyError {
|
||||
t.Helper()
|
||||
var pe *ProxyError
|
||||
if !errors.As(err, &pe) || pe.Status != http.StatusForbidden || !Quarantined(err) {
|
||||
t.Fatalf("want quarantine 403, got %v", err)
|
||||
}
|
||||
return pe
|
||||
}
|
||||
|
||||
func TestQuarantineBlocksFreshArtifact(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
r := quarantineRemote(t, "q-fresh", true)
|
||||
pe := wantQuarantined(t, func() error { _, err := testEngine.Fetch(ctx, r, "fresh.tgz", prov(t, models.PackageNPM)); return err }())
|
||||
if d := time.Until(pe.Available); d < 70*time.Hour || d > 72*time.Hour {
|
||||
t.Errorf("available in %s, want ~71h", d)
|
||||
}
|
||||
if a, _ := testDB.GetArtifact(ctx, r.Name, "fresh.tgz"); a != nil {
|
||||
t.Error("quarantined artifact was cached")
|
||||
}
|
||||
_, err := testEngine.Head(ctx, r, "fresh.tgz", prov(t, models.PackageNPM))
|
||||
_ = wantQuarantined(t, err)
|
||||
}
|
||||
|
||||
func TestQuarantineServesOldArtifact(t *testing.T) {
|
||||
requireStack(t)
|
||||
r := quarantineRemote(t, "q-old", true)
|
||||
res, err := testEngine.Fetch(context.Background(), r, "old.tgz", prov(t, models.PackageNPM))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := readAll(t, res); got != "body /old.tgz" {
|
||||
t.Errorf("body = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuarantineFallsBackToFirstSeen(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
r := quarantineRemote(t, "q-undated", true)
|
||||
p := prov(t, models.PackageNPM)
|
||||
_, err := testEngine.Fetch(ctx, r, "undated.tgz", p)
|
||||
first := wantQuarantined(t, err).Published
|
||||
if time.Since(first) > time.Minute {
|
||||
t.Fatalf("first seen %s, want now", first)
|
||||
}
|
||||
_, err = testEngine.Fetch(ctx, r, "undated.tgz", p)
|
||||
if again := wantQuarantined(t, err).Published; !again.Equal(first) {
|
||||
t.Errorf("first seen moved: %s -> %s", first, again)
|
||||
}
|
||||
|
||||
r.QuarantineDays = 1
|
||||
c := goredis.NewClient(must(goredis.ParseURL(testRedis)))
|
||||
defer func() { _ = c.Close() }()
|
||||
if err := c.Set(ctx, "firstseen:q-undated:undated.tgz", time.Now().Add(-48*time.Hour).Unix(), time.Hour).Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := testEngine.Fetch(ctx, r, "undated.tgz", p); err != nil {
|
||||
t.Errorf("seen 48h ago with 1 day quarantine: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuarantineSkipsMutable(t *testing.T) {
|
||||
requireStack(t)
|
||||
r := quarantineRemote(t, "q-mutable", true)
|
||||
if _, err := testEngine.Fetch(context.Background(), r, "pkg", prov(t, models.PackageNPM)); err != nil {
|
||||
t.Errorf("mutable index blocked: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuarantineDisabled(t *testing.T) {
|
||||
requireStack(t)
|
||||
r := quarantineRemote(t, "q-off", false)
|
||||
if _, err := testEngine.Fetch(context.Background(), r, "fresh.tgz", prov(t, models.PackageNPM)); err != nil {
|
||||
t.Errorf("disabled quarantine blocked: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuarantineKeepsServingCached(t *testing.T) {
|
||||
requireStack(t)
|
||||
ctx := context.Background()
|
||||
r := quarantineRemote(t, "q-cached", false)
|
||||
p := prov(t, models.PackageNPM)
|
||||
if _, err := testEngine.Fetch(ctx, r, "fresh.tgz", p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r.QuarantineEnabled = true
|
||||
expire(t, r.Name, "fresh.tgz")
|
||||
if _, err := testEngine.Fetch(ctx, r, "fresh.tgz", p); err != nil {
|
||||
t.Errorf("artifact cached before quarantine refused on refetch: %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user