diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 239fefc..2bd39d4 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -1565,7 +1565,9 @@ func (gc *GarbageCollector) fetchUserProfile(ctx context.Context, pdsEndpoint, u client := &http.Client{Timeout: 10 * time.Second} reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", - pdsEndpoint, userDID, atproto.SailorProfileCollection) + pdsEndpoint, + url.QueryEscape(userDID), + url.QueryEscape(atproto.SailorProfileCollection)) req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil) if err != nil { @@ -1609,7 +1611,9 @@ func (gc *GarbageCollector) fetchUserTags(ctx context.Context, pdsEndpoint, user for { reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100", - pdsEndpoint, userDID, atproto.TagCollection) + pdsEndpoint, + url.QueryEscape(userDID), + url.QueryEscape(atproto.TagCollection)) if cursor != "" { reqURL += "&cursor=" + cursor } @@ -1713,7 +1717,9 @@ func (gc *GarbageCollector) fetchUserManifestsFromEndpoint(ctx context.Context, for { reqURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.listRecords?repo=%s&collection=%s&limit=100", - pdsEndpoint, userDID, atproto.ManifestCollection) + pdsEndpoint, + url.QueryEscape(userDID), + url.QueryEscape(atproto.ManifestCollection)) if cursor != "" { reqURL += "&cursor=" + cursor } diff --git a/pkg/hold/gc/predecessor_test.go b/pkg/hold/gc/predecessor_test.go index ef36f35..473f922 100644 --- a/pkg/hold/gc/predecessor_test.go +++ b/pkg/hold/gc/predecessor_test.go @@ -234,3 +234,90 @@ func TestAnalyzeRecordsClearsPredecessorUnresolved(t *testing.T) { "is never re-checked, and GC stops reclaiming its manifests' blobs") } } + +// TestFetchUserRecords_EscapesDIDInRepoParam covers a soft failure that costs +// blobs. A did:web carrying a port spells that port as a literal %3A, so a DID +// interpolated straight into a query string arrives at the PDS decoded back to +// ":" — a different DID, which no repo matches. listRecords answers 200 with an +// empty list, and GC reads that as "this user has no manifests": every blob +// they own drops out of the referenced set and is deleted once past the blob +// grace period. +// +// Nothing about it looks like a failure. There is no error to log and no status +// to notice, which is why it wants a test rather than a code read. +// +// did:plc, which every production user has, contains nothing that needs +// escaping and was never affected. The escaping matched at two of the five +// call sites in this file before this test. +func TestFetchUserRecords_EscapesDIDInRepoParam(t *testing.T) { + // A did:web hold or user on a non-443 port. The %3A is part of the DID. + const userDID = "did:web:127.0.0.1%3A44639:user:alice.test" + const holdDID = "did:web:hold.example.com" + + tests := []struct { + name string + collection string + // record is the value body the fake PDS returns for this collection. + record string + // call exercises one fetch path and reports whether it found the record. + call func(t *testing.T, gc *GarbageCollector, endpoint string) bool + }{ + { + name: "manifests", + collection: atproto.ManifestCollection, + record: `{"$type":"` + atproto.ManifestCollection + `","repository":"myapp", + "digest":"sha256:abc123","holdDid":"` + holdDID + `", + "layers":[{"digest":"sha256:layer1","size":1,"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip"}]}`, + call: func(t *testing.T, gc *GarbageCollector, endpoint string) bool { + manifests, err := gc.fetchUserManifestsFromEndpoint( + context.Background(), userDID, endpoint, holdDID) + if err != nil { + t.Fatalf("fetchUserManifestsFromEndpoint: %v", err) + } + return len(manifests) > 0 + }, + }, + { + name: "tags", + collection: atproto.TagCollection, + record: `{"$type":"` + atproto.TagCollection + `","repository":"myapp", + "tag":"latest","manifestDigest":"sha256:abc123"}`, + call: func(t *testing.T, gc *GarbageCollector, endpoint string) bool { + tags, err := gc.fetchUserTags(context.Background(), endpoint, userDID) + if err != nil { + t.Fatalf("fetchUserTags: %v", err) + } + return len(tags) > 0 + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotRepo string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // r.URL.Query() decodes, which is exactly what a real PDS does + // and exactly where the DID loses its %3A. + gotRepo = r.URL.Query().Get("repo") + w.Header().Set("Content-Type", "application/json") + if gotRepo != userDID { + // What a PDS answers for a repo it does not have: not an + // error, just nothing. + fmt.Fprint(w, `{"records":[]}`) + return + } + fmt.Fprintf(w, `{"records":[{"uri":"at://%s/%s/rkey1","cid":"bafyrei1","value":%s}]}`, + userDID, tt.collection, tt.record) + })) + defer srv.Close() + + gc := &GarbageCollector{logger: newTestLogger()} + + if !tt.call(t, gc, srv.URL) { + t.Errorf("no records found; PDS saw repo=%q, want %q. "+ + "The empty answer is indistinguishable from an empty repo, so GC "+ + "unreferences every blob this user owns", gotRepo, userDID) + } + }) + } +} diff --git a/test/integration/gc_test.go b/test/integration/gc_test.go new file mode 100644 index 0000000..a40412b --- /dev/null +++ b/test/integration/gc_test.go @@ -0,0 +1,108 @@ +//go:build integration + +package integration + +import ( + "fmt" + "testing" + + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/v1/random" + + "atcr.io/internal/testharness" + "atcr.io/pkg/hold" + "atcr.io/pkg/hold/gc" + "atcr.io/pkg/s3" + + _ "github.com/distribution/distribution/v3/registry/auth/token" + _ "github.com/distribution/distribution/v3/registry/storage/driver/inmemory" +) + +// newGC builds a collector over the harness's own hold and bucket. The hold +// server keeps its collector unexported, so the test constructs an equivalent +// one from the same PDS and the same S3 credentials the harness gave the hold. +func newGC(t *testing.T, h *testharness.Harness) *gc.GarbageCollector { + t.Helper() + + storage := hold.StorageConfig{ + AccessKey: "test", + SecretKey: "test", + Region: "us-east-1", + Bucket: "atcr-test", + Endpoint: h.S3URL, + } + s3svc, err := s3.NewS3Service(storage.S3Params()) + if err != nil { + t.Fatalf("build S3 service: %v", err) + } + return gc.NewGarbageCollector(h.Hold.PDS, s3svc, gc.Config{}) +} + +// TestGCPreviewAccountsForAFreshPush walks the whole GC analysis against a real +// S3 listing rather than a mock: push an image, then ask GC what it sees. +// +// The blob grace period is a package constant, so nothing pushed during a test +// can ever age past it and "no orphans" is true no matter how badly the +// reference walk is broken. The load-bearing assertion is therefore the +// referenced set, which grace does not touch: every blob the push put in the +// bucket has to be traced back through the user's manifest to this hold. A +// break anywhere in that chain (manifest fetch from the PDS, hold-ownership +// matching, blob listing) shows up here as referenced < total, which on a hold +// holding exactly one image is the state that would delete live content once +// those blobs aged past seven days. +func TestGCPreviewAccountsForAFreshPush(t *testing.T) { + h := testharness.New(t) + alice := h.AddSailor("alice.test") + + const layers = 3 + img, err := random.Image(1<<20, layers) + if err != nil { + t.Fatalf("build random image: %v", err) + } + + ref, err := name.ParseReference( + fmt.Sprintf("%s/%s/gc-repo:tag", h.AppViewHostPort(), alice.Handle()), + name.Insecure, + ) + if err != nil { + t.Fatalf("parse ref: %v", err) + } + + creds := h.RegistryCreds(alice) + if err := Clients[0].Push(t.Context(), t, ref.String(), img, creds); err != nil { + t.Fatalf("push: %v", err) + } + + preview, err := newGC(t, h).Preview(t.Context()) + if err != nil { + t.Fatalf("GC preview: %v", err) + } + + t.Logf("preview: users=%d manifests=%d blobs=%d referenced=%d orphanedBlobs=%d orphanedRecords=%d", + preview.UsersChecked, preview.ManifestsChecked, preview.TotalBlobs, + preview.ReferencedBlobs, len(preview.OrphanedBlobs), len(preview.OrphanedRecords)) + + if preview.ManifestsChecked < 1 { + t.Fatalf("ManifestsChecked = %d: GC never found the manifest, so every blob below "+ + "is unreferenced for the wrong reason", preview.ManifestsChecked) + } + + // Layers plus the config blob. Fewer means the S3 listing missed content + // the push demonstrably wrote. + if preview.TotalBlobs < layers+1 { + t.Errorf("TotalBlobs = %d, want at least %d (%d layers + config)", + preview.TotalBlobs, layers+1, layers) + } + + if preview.ReferencedBlobs != preview.TotalBlobs { + t.Errorf("ReferencedBlobs = %d of %d: the hold holds one freshly pushed image and "+ + "nothing else, so every blob must trace back to it", + preview.ReferencedBlobs, preview.TotalBlobs) + } + + // Belt and braces. True by grace alone today, but it is the assertion that + // starts carrying weight the moment the grace period becomes configurable. + for _, orphan := range preview.OrphanedBlobs { + t.Errorf("fresh push listed as an orphaned blob: %+v", orphan) + } +}