diff --git a/.air.hold.toml b/.air.hold.toml index 9c905af..0b8a723 100644 --- a/.air.hold.toml +++ b/.air.hold.toml @@ -6,7 +6,7 @@ pre_cmd = ["go generate ./pkg/hold/..."] cmd = "go build -buildvcs=false -o ./tmp/atcr-hold ./cmd/hold" entrypoint = ["./tmp/atcr-hold", "serve", "--config", "config-hold.example.yaml"] include_ext = ["go", "html", "css", "js"] -exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "pkg/appview", "node_modules"] +exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "node_modules", "scanner", "pkg/appview", "pkg/labeler"] exclude_regex = ["_test\\.go$", "cbor_gen\\.go$", "\\.min\\.js$", "public/css/style\\.css$", "public/icons\\.svg$"] delay = 3000 stop_on_error = true diff --git a/.air.labeler.toml b/.air.labeler.toml index dec1af6..ed460a7 100644 --- a/.air.labeler.toml +++ b/.air.labeler.toml @@ -5,7 +5,7 @@ tmp_dir = "tmp" cmd = "go build -buildvcs=false -o ./tmp/atcr-labeler ./cmd/labeler" entrypoint = ["./tmp/atcr-labeler", "serve", "--config", "config-labeler.example.yaml"] include_ext = ["go", "html", "css", "js"] -exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "pkg/appview", "pkg/hold", "node_modules"] +exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "node_modules", "scanner", "pkg/appview", "pkg/hold"] exclude_regex = ["_test\\.go$", "cbor_gen\\.go$", "\\.min\\.js$", "public/css/style\\.css$", "public/icons\\.svg$"] delay = 3000 stop_on_error = true diff --git a/.air.toml b/.air.toml index c0b0668..af8f69c 100644 --- a/.air.toml +++ b/.air.toml @@ -10,7 +10,7 @@ pre_cmd = ["go generate ./pkg/appview/..."] cmd = "go build -tags billing -buildvcs=false -o ./tmp/atcr-appview ./cmd/appview" entrypoint = ["./tmp/atcr-appview", "serve", "--config", "config-appview.example.yaml"] include_ext = ["go", "html", "css", "js"] -exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "node_modules", "pkg/hold"] +exclude_dir = ["bin", "tmp", "vendor", "deploy", "docs", ".git", "dist", "node_modules", "scanner", "pkg/hold", "pkg/labeler"] exclude_regex = ["_test\\.go$", "cbor_gen\\.go$", "\\.min\\.js$", "public/css/style\\.css$", "public/icons\\.svg$"] delay = 3000 stop_on_error = true diff --git a/config-appview.example.yaml b/config-appview.example.yaml index 115219e..d162586 100644 --- a/config-appview.example.yaml +++ b/config-appview.example.yaml @@ -82,6 +82,10 @@ auth: key_path: /var/lib/atcr/auth/private-key.pem # X.509 certificate matching the JWT signing key. cert_path: /var/lib/atcr/auth/private-key.crt +# Credential helper download settings. +credential_helper: + # Tangled repository URL for credential helper downloads. + tangled_repo: "" # Legal page customization for self-hosted instances. legal: # Organization name for Terms of Service and Privacy Policy. Defaults to server.client_name. @@ -92,6 +96,10 @@ legal: ai: # Anthropic API key for AI Image Advisor. Also reads CLAUDE_API_KEY env var as fallback. api_key: "" +# ATProto labeler for content moderation (DMCA takedowns). +labeler: + # DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables label filtering. + did: "" # Stripe billing integration (requires -tags billing build). billing: # Stripe secret key. Can also be set via STRIPE_SECRET_KEY env var (takes precedence). Billing is enabled automatically when set. diff --git a/config-hold.example.yaml b/config-hold.example.yaml index f22beda..9de70ab 100644 --- a/config-hold.example.yaml +++ b/config-hold.example.yaml @@ -135,3 +135,9 @@ scanner: secret: "" # Minimum interval between re-scans of the same manifest. When set, the hold proactively scans manifests when the scanner is idle. Default: 168h (7 days). Set to 0 to disable. rescan_interval: 168h0m0s +# Labeler subscription settings. When configured, the hold consumes takedown labels from the named labeler and purges affected records on receipt; GC consults the cache to gate blob cleanup. Empty subscribe_url disables. +labeler: + # DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables labeler integration. + did: "" + # Reversibility window for takedowns. Blobs survive this long after a takedown so the action can be reversed. After this window the GC reclaims them. Default: 720h (30 days). + grace_window: 720h0m0s diff --git a/deploy/upcloud/configs/hold.yaml.tmpl b/deploy/upcloud/configs/hold.yaml.tmpl index bffc828..89ac5ed 100644 --- a/deploy/upcloud/configs/hold.yaml.tmpl +++ b/deploy/upcloud/configs/hold.yaml.tmpl @@ -61,4 +61,9 @@ quota: scanner: secret: "{{.ScannerSecret}}" rescan_interval: 168h0m0s +labeler: + # Subscribe to the appview's labeler so takedowns purge records on this + # hold and the GC honors the reversibility window. Empty disables. + did: "did:web:seamark.dev" + grace_window: 720h0m0s diff --git a/docker-compose.yml b/docker-compose.yml index 75f5bfb..49c2387 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,6 +65,13 @@ services: HOLD_REGISTRATION_ALLOW_ALL_CREW: true HOLD_SERVER_TEST_MODE: true HOLD_LOG_LEVEL: debug + # Subscribe to the dev labeler so takedowns purge records on this hold and + # GC honors the reversibility window. Same value the appview uses for + # ATCR_LABELER_DID — accepts a did:web identifier or a raw URL. + HOLD_LABELER_DID: http://172.28.0.4:5002 + # Short grace window for dev so the takedown→GC path is exercisable without + # waiting weeks. Production default is 720h (30 days). + HOLD_LABELER_GRACE_WINDOW: 1h LOG_SHIPPER_BACKEND: victoria LOG_SHIPPER_URL: http://172.28.0.10:9428 # S3 storage config comes from env_file (AWS_*, S3_*) diff --git a/docs/HOLD_XRPC_ENDPOINTS.md b/docs/HOLD_XRPC_ENDPOINTS.md index 77bb8b2..9a70719 100644 --- a/docs/HOLD_XRPC_ENDPOINTS.md +++ b/docs/HOLD_XRPC_ENDPOINTS.md @@ -37,6 +37,7 @@ This document lists all XRPC endpoints implemented in the Hold service (`pkg/hol |----------|--------|-------------| | `/xrpc/com.atproto.repo.deleteRecord` | POST | Delete a record | | `/xrpc/com.atproto.repo.uploadBlob` | POST | Upload ATProto blob | +| `/xrpc/io.atcr.hold.purgeManifest` | POST | Purge layer/scan/image-config records for a manifest (eager delete + takedown). Idempotent. | ### Auth Required (Service Token or DPoP) @@ -82,6 +83,7 @@ All require `blob:write` permission via service token: | `/xrpc/io.atcr.hold.getQuota` | GET | none | Get user quota info | | `/xrpc/io.atcr.hold.getLayersForManifest` | GET | none | Get layer records for a manifest AT-URI | | `/xrpc/io.atcr.hold.image.getConfig` | GET | none | Get OCI image config record for a manifest digest | +| `/xrpc/io.atcr.hold.purgeManifest` | POST | owner/crew admin | Purge layer/scan/image-config records for a single manifest URI. Called by appview on UI delete; called internally on takedown receipt. Does not delete S3 blobs (GC handles those). | | `/xrpc/io.atcr.hold.listTiers` | GET | none | List hold's available tiers with quotas and features (scanOnPush) | | `/xrpc/io.atcr.hold.updateCrewTier` | POST | appview token | Update crew member's tier | diff --git a/lexicons/io/atcr/hold/purgeManifest.json b/lexicons/io/atcr/hold/purgeManifest.json new file mode 100644 index 0000000..28a93fc --- /dev/null +++ b/lexicons/io/atcr/hold/purgeManifest.json @@ -0,0 +1,54 @@ +{ + "lexicon": 1, + "id": "io.atcr.hold.purgeManifest", + "defs": { + "main": { + "type": "procedure", + "description": "Purge layer, scan, and image-config records associated with a manifest. Used by the appview when a user deletes a manifest, and by the hold's own labeler subscriber on takedown receipt. Idempotent: missing records are not errors. Does not delete S3 blobs (GC handles that based on remaining references).", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["manifestUri"], + "properties": { + "manifestUri": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the manifest record, e.g. at://did:plc:xyz/io.atcr.manifest/" + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["success", "layersDeleted", "scanDeleted", "imageConfigDeleted"], + "properties": { + "success": { + "type": "boolean", + "description": "Whether the purge completed successfully" + }, + "layersDeleted": { + "type": "integer", + "description": "Number of layer records deleted" + }, + "scanDeleted": { + "type": "boolean", + "description": "Whether a scan record was deleted" + }, + "imageConfigDeleted": { + "type": "boolean", + "description": "Whether an image config record was deleted" + } + } + } + }, + "errors": [ + { "name": "AuthRequired" }, + { "name": "InvalidRequest" }, + { "name": "PurgeFailed" } + ] + } + } +} diff --git a/pkg/appview/db/labels.go b/pkg/appview/db/labels.go index 60f68cf..37861ea 100644 --- a/pkg/appview/db/labels.go +++ b/pkg/appview/db/labels.go @@ -22,6 +22,8 @@ func (lc *LabelChecker) IsTakenDown(did, repository string) (bool, error) { } // Label represents an ATProto label mirrored from a labeler service. +// Exp is the optional expiration timestamp from the ATProto label spec; +// nil means the label does not expire. type Label struct { ID int64 Src string @@ -29,6 +31,7 @@ type Label struct { Val string Neg bool Cts time.Time + Exp *time.Time SubjectDID string SubjectRepo string Seq int64 @@ -49,7 +52,7 @@ func IsTakenDown(db DBTX, did, repository string) (bool, error) { WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val AND l2.neg = 1 AND l2.id > l1.id ) - AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP) + AND (l1.exp IS NULL OR datetime(l1.exp) > CURRENT_TIMESTAMP) )`, did, repository, ).Scan(&exists) @@ -58,11 +61,15 @@ func IsTakenDown(db DBTX, did, repository string) (bool, error) { // UpsertLabel inserts or updates a label from a labeler subscription. func UpsertLabel(db DBTX, l *Label) error { + var exp any + if l.Exp != nil { + exp = l.Exp.UTC().Format(time.RFC3339) + } _, err := db.Exec( - `INSERT INTO labels (src, uri, val, neg, cts, subject_did, subject_repo, seq) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(src, uri, val, neg) DO UPDATE SET cts = excluded.cts, seq = excluded.seq`, - l.Src, l.URI, l.Val, l.Neg, l.Cts.UTC().Format(time.RFC3339), + `INSERT INTO labels (src, uri, val, neg, cts, exp, subject_did, subject_repo, seq) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(src, uri, val, neg) DO UPDATE SET cts = excluded.cts, exp = excluded.exp, seq = excluded.seq`, + l.Src, l.URI, l.Val, l.Neg, l.Cts.UTC().Format(time.RFC3339), exp, l.SubjectDID, l.SubjectRepo, l.Seq, ) return err diff --git a/pkg/appview/db/labels_test.go b/pkg/appview/db/labels_test.go new file mode 100644 index 0000000..6503f9a --- /dev/null +++ b/pkg/appview/db/labels_test.go @@ -0,0 +1,138 @@ +package db + +import ( + "testing" + "time" +) + +// TestIsTakenDown_ExpRespected verifies that the takedown check honors the +// optional ATProto label expiration: NULL and future-dated exp values count +// the label as active, while past-dated exp values exclude it. +func TestIsTakenDown_ExpRespected(t *testing.T) { + db, err := InitDB("file:TestIsTakenDown_ExpRespected?mode=memory&cache=shared", LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer db.Close() + + now := time.Now().UTC() + past := now.Add(-1 * time.Hour) + future := now.Add(1 * time.Hour) + + cases := []struct { + name string + did string + exp *time.Time + wantHit bool + }{ + {"no_exp_active", "did:plc:noexp", nil, true}, + {"future_exp_active", "did:plc:future", &future, true}, + {"past_exp_inactive", "did:plc:past", &past, false}, + } + + for _, tc := range cases { + label := &Label{ + Src: "did:plc:labeler", + URI: "at://" + tc.did + "/io.atcr.repo.page/foo", + Val: "!takedown", + Neg: false, + Cts: now, + Exp: tc.exp, + SubjectDID: tc.did, + Seq: 1, + } + if err := UpsertLabel(db, label); err != nil { + t.Fatalf("%s: upsert label: %v", tc.name, err) + } + + got, err := IsTakenDown(db, tc.did, "foo") + if err != nil { + t.Fatalf("%s: IsTakenDown: %v", tc.name, err) + } + if got != tc.wantHit { + t.Errorf("%s: IsTakenDown = %v, want %v", tc.name, got, tc.wantHit) + } + } +} + +// TestIsTakenDown_NegationWinsOverExp verifies that a later negation row +// suppresses an earlier non-expired takedown — exp doesn't shield it from +// being reversed by a !takedown neg=1 with a higher id. +func TestIsTakenDown_NegationWinsOverExp(t *testing.T) { + db, err := InitDB("file:TestIsTakenDown_NegationWinsOverExp?mode=memory&cache=shared", LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer db.Close() + + did := "did:plc:reversed" + uri := "at://" + did + "/io.atcr.repo.page/bar" + src := "did:plc:labeler" + now := time.Now().UTC() + future := now.Add(24 * time.Hour) + + if err := UpsertLabel(db, &Label{ + Src: src, URI: uri, Val: "!takedown", Neg: false, + Cts: now, Exp: &future, SubjectDID: did, Seq: 1, + }); err != nil { + t.Fatalf("seed takedown: %v", err) + } + if err := UpsertLabel(db, &Label{ + Src: src, URI: uri, Val: "!takedown", Neg: true, + Cts: now.Add(time.Minute), SubjectDID: did, Seq: 2, + }); err != nil { + t.Fatalf("seed reversal: %v", err) + } + + got, err := IsTakenDown(db, did, "bar") + if err != nil { + t.Fatalf("IsTakenDown: %v", err) + } + if got { + t.Errorf("IsTakenDown = true, want false (reversal should suppress non-expired takedown)") + } +} + +// TestUpsertLabel_ExpUpdatedOnConflict verifies that upserting an existing +// label row updates the exp column (not just cts/seq) — so a labeler that +// extends or removes an expiration is reflected. +func TestUpsertLabel_ExpUpdatedOnConflict(t *testing.T) { + db, err := InitDB("file:TestUpsertLabel_ExpUpdatedOnConflict?mode=memory&cache=shared", LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer db.Close() + + did := "did:plc:updated" + src := "did:plc:labeler" + uri := "at://" + did + "/io.atcr.repo.page/baz" + now := time.Now().UTC() + past := now.Add(-1 * time.Hour) + future := now.Add(1 * time.Hour) + + // Seed with an already-expired label — IsTakenDown should be false. + if err := UpsertLabel(db, &Label{ + Src: src, URI: uri, Val: "!takedown", Neg: false, + Cts: now, Exp: &past, SubjectDID: did, Seq: 1, + }); err != nil { + t.Fatalf("seed: %v", err) + } + if got, _ := IsTakenDown(db, did, "baz"); got { + t.Fatalf("expected expired label to be inactive before update") + } + + // Re-upsert with a future exp — same UNIQUE key, should update exp. + if err := UpsertLabel(db, &Label{ + Src: src, URI: uri, Val: "!takedown", Neg: false, + Cts: now.Add(time.Minute), Exp: &future, SubjectDID: did, Seq: 2, + }); err != nil { + t.Fatalf("re-upsert: %v", err) + } + got, err := IsTakenDown(db, did, "baz") + if err != nil { + t.Fatalf("IsTakenDown after update: %v", err) + } + if !got { + t.Errorf("IsTakenDown = false, want true after extending exp into the future") + } +} diff --git a/pkg/appview/db/migrations/0024_add_exp_to_labels.yaml b/pkg/appview/db/migrations/0024_add_exp_to_labels.yaml new file mode 100644 index 0000000..b5d09e6 --- /dev/null +++ b/pkg/appview/db/migrations/0024_add_exp_to_labels.yaml @@ -0,0 +1,3 @@ +description: Add optional exp (expiration) column to labels for ATProto label spec +query: | + ALTER TABLE labels ADD COLUMN exp TIMESTAMP; diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index 03d821f..778d753 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -37,7 +37,7 @@ func activeTakedownClause(alias string) string { WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val AND l2.neg = 1 AND l2.id > l1.id ) - AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP) + AND (l1.exp IS NULL OR datetime(l1.exp) > CURRENT_TIMESTAMP) )` } diff --git a/pkg/appview/db/schema.sql b/pkg/appview/db/schema.sql index 9b38cbc..09821d3 100644 --- a/pkg/appview/db/schema.sql +++ b/pkg/appview/db/schema.sql @@ -306,6 +306,7 @@ CREATE TABLE IF NOT EXISTS labels ( val TEXT NOT NULL, neg BOOLEAN NOT NULL DEFAULT 0, cts TIMESTAMP NOT NULL, + exp TIMESTAMP, subject_did TEXT NOT NULL, subject_repo TEXT NOT NULL DEFAULT '', seq INTEGER NOT NULL DEFAULT 0, diff --git a/pkg/appview/handlers/hold_purge.go b/pkg/appview/handlers/hold_purge.go new file mode 100644 index 0000000..b118625 --- /dev/null +++ b/pkg/appview/handlers/hold_purge.go @@ -0,0 +1,118 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "time" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/auth" + "atcr.io/pkg/auth/oauth" +) + +// purgeManifestRequest is the JSON body sent to io.atcr.hold.purgeManifest. +type purgeManifestRequest struct { + ManifestURI string `json:"manifestUri"` +} + +// purgeOnHold tells the hold to delete the layer, scan, and image-config +// records associated with a single manifest. This is best-effort: callers +// should treat all errors as "log and continue" because lazy GC on the hold +// will catch up either way (and on third-party holds the user may not even +// have the captain/crew-admin permission needed for the call to succeed). +// +// holdDID identifies which hold owns the manifest's blobs (typically the +// `hold_endpoint` column on the manifests row, or a freshly-resolved value +// from the manifest record). userDID + pdsEndpoint are the OAuth-acting +// user — the service token is minted from their PDS with audience = holdDID. +func purgeOnHold(ctx context.Context, refresher *oauth.Refresher, userDID, pdsEndpoint, holdDID, manifestURI string) { + if holdDID == "" || manifestURI == "" { + return + } + if refresher == nil { + slog.Debug("purgeOnHold: OAuth refresher unavailable; skipping", + "hold_did", holdDID, "manifest", manifestURI) + return + } + + timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + + holdURL, err := atproto.ResolveHoldURL(timeoutCtx, holdDID) + if err != nil { + slog.Warn("purgeOnHold: failed to resolve hold URL", + "hold_did", holdDID, "error", err) + return + } + + serviceToken, err := auth.GetOrFetchServiceToken(timeoutCtx, refresher, userDID, holdDID, pdsEndpoint) + if err != nil { + slog.Warn("purgeOnHold: failed to mint service token", + "hold_did", holdDID, "user_did", userDID, "error", err) + return + } + + body, err := json.Marshal(purgeManifestRequest{ManifestURI: manifestURI}) + if err != nil { + slog.Warn("purgeOnHold: failed to marshal request", + "hold_did", holdDID, "error", err) + return + } + + req, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, + holdURL+atproto.HoldPurgeManifest, bytes.NewReader(body)) + if err != nil { + slog.Warn("purgeOnHold: failed to create request", + "hold_did", holdDID, "error", err) + return + } + req.Header.Set("Authorization", "Bearer "+serviceToken) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + slog.Warn("purgeOnHold: request failed", + "hold_did", holdDID, "manifest", manifestURI, "error", err) + return + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized { + // Sailor pushing to a third-party hold won't have captain/crew-admin + // rights; that's expected. Lazy GC on that hold will reclaim later. + slog.Debug("purgeOnHold: not authorized on hold (lazy GC will handle)", + "hold_did", holdDID, "manifest", manifestURI, "status", resp.StatusCode) + return + } + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + slog.Warn("purgeOnHold: hold returned non-OK status", + "hold_did", holdDID, "manifest", manifestURI, + "status", resp.StatusCode, "body", string(body)) + return + } + + var out struct { + Success bool `json:"success"` + LayersDeleted int `json:"layersDeleted"` + ScanDeleted bool `json:"scanDeleted"` + ImageConfigDeleted bool `json:"imageConfigDeleted"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + slog.Warn("purgeOnHold: failed to parse response", + "hold_did", holdDID, "manifest", manifestURI, "error", err) + return + } + + slog.Info("purgeOnHold: purge succeeded", + "hold_did", holdDID, + "manifest", manifestURI, + "layers_deleted", out.LayersDeleted, + "scan_deleted", out.ScanDeleted, + "image_config_deleted", out.ImageConfigDeleted, + ) +} diff --git a/pkg/appview/handlers/images.go b/pkg/appview/handlers/images.go index 04abc28..d404afa 100644 --- a/pkg/appview/handlers/images.go +++ b/pkg/appview/handlers/images.go @@ -156,6 +156,16 @@ func (h *DeleteManifestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request } } + // Read the appview's cached manifest row before deleting it so we know + // which hold owned the blobs. Best-effort — if not cached, the manifest + // record from the PDS would also have it but we don't pre-fetch the PDS + // record just for this. Without a hold DID we just skip the eager purge + // and fall back to lazy GC on the hold. + var holdDID string + if cached, err := db.GetManifestDetail(h.ReadOnlyDB, user.DID, repo, digest); err == nil && cached != nil { + holdDID = cached.HoldEndpoint + } + // Compute rkey for manifest record (digest without "sha256:" prefix) rkey := strings.TrimPrefix(digest, "sha256:") @@ -176,6 +186,12 @@ func (h *DeleteManifestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request return } + // Tell the hold to drop its layer/scan/image-config records for this + // manifest. Best-effort — failures here only mean the hold's lazy GC + // will clean up later, so we don't reflect the failure to the user. + manifestURI := atproto.BuildManifestURI(user.DID, digest) + purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI) + w.WriteHeader(http.StatusOK) } @@ -233,6 +249,12 @@ func (h *DeleteUntaggedManifestsHandler) ServeHTTP(w http.ResponseWriter, r *htt for _, digest := range digests { rkey := strings.TrimPrefix(digest, "sha256:") + // Snapshot hold ownership before the delete so we can purge after. + var holdDID string + if cached, err := db.GetManifestDetail(h.ReadOnlyDB, user.DID, req.Repo, digest); err == nil && cached != nil { + holdDID = cached.HoldEndpoint + } + if err := pdsClient.DeleteRecord(r.Context(), atproto.ManifestCollection, rkey); err != nil { if handleOAuthError(r.Context(), h.Refresher, user.DID, err) { render.Status(r, http.StatusUnauthorized) @@ -253,6 +275,9 @@ func (h *DeleteUntaggedManifestsHandler) ServeHTTP(w http.ResponseWriter, r *htt continue } + manifestURI := atproto.BuildManifestURI(user.DID, digest) + purgeOnHold(r.Context(), h.Refresher, user.DID, user.PDSEndpoint, holdDID, manifestURI) + deleted++ } diff --git a/pkg/appview/labeler/subscriber.go b/pkg/appview/labeler/subscriber.go index ed383b3..5bdaf18 100644 --- a/pkg/appview/labeler/subscriber.go +++ b/pkg/appview/labeler/subscriber.go @@ -129,12 +129,22 @@ func (s *Subscriber) connect() error { cts, _ := time.Parse(time.RFC3339, le.Cts) did, repo := extractSubjectFromURI(le.Uri) + // Exp is optional in the ATProto label spec — treat unparseable + // values as "no expiration" rather than dropping the label. + var exp *time.Time + if le.Exp != nil { + if t, err := time.Parse(time.RFC3339, *le.Exp); err == nil { + exp = &t + } + } + label := &db.Label{ Src: le.Src, URI: le.Uri, Val: le.Val, Neg: le.Neg != nil && *le.Neg, Cts: cts, + Exp: exp, SubjectDID: did, SubjectRepo: repo, Seq: seq, diff --git a/pkg/atproto/endpoints.go b/pkg/atproto/endpoints.go index 107f6d0..a97e146 100644 --- a/pkg/atproto/endpoints.go +++ b/pkg/atproto/endpoints.go @@ -73,6 +73,14 @@ const ( // Method: DELETE // Response: {"success": true, "crew_deleted": bool, "layers_deleted": int, "stats_deleted": int} HoldDeleteUserData = "/xrpc/io.atcr.hold.deleteUserData" + + // HoldPurgeManifest purges layer, scan, and image-config records for a single + // manifest. Called by the appview on UI manifest delete and by the hold's own + // labeler subscriber on takedown. Idempotent. + // Method: POST + // Request: {"manifestUri": "at://did:.../io.atcr.manifest/"} + // Response: {"success": true, "layersDeleted": N, "scanDeleted": bool, "imageConfigDeleted": bool} + HoldPurgeManifest = "/xrpc/io.atcr.hold.purgeManifest" ) // Hold service crew management endpoints (io.atcr.hold.*) diff --git a/pkg/hold/config.go b/pkg/hold/config.go index e38b416..eb191ec 100644 --- a/pkg/hold/config.go +++ b/pkg/hold/config.go @@ -52,6 +52,7 @@ type Config struct { GC gc.Config `yaml:"gc" comment:"Garbage collection settings."` Quota quota.Config `yaml:"quota" comment:"Storage quota tiers. Empty disables quota enforcement."` Scanner ScannerConfig `yaml:"scanner" comment:"Vulnerability scanner settings. Empty disables scanning."` + Labeler LabelerConfig `yaml:"labeler" comment:"Labeler subscription settings. When configured, the hold consumes takedown labels from the named labeler and purges affected records on receipt; GC consults the cache to gate blob cleanup. Empty subscribe_url disables."` configPath string `yaml:"-"` // internal: path to YAML file for subsystem config loading } @@ -180,6 +181,29 @@ func (s ServerConfig) AppviewURL() string { return URLFromDIDWeb(s.AppviewDID) } +// LabelerConfig defines labeler subscription settings. +// +// When DID is set, the hold opens a websocket to the labeler's +// com.atproto.label.subscribeLabels endpoint and only honors labels whose +// Src matches the same DID. Active takedowns are cached locally with their +// Cts timestamp; on receipt of a !takedown label the hold immediately purges +// layer/scan/image-config records for the labeled manifest (or all manifests +// by the labeled DID for user-level takedowns). Negations drop the cache +// entry. The GC consults the cache when computing referenced sets so blobs +// survive a configurable grace window before being collected, preserving +// reversibility. +type LabelerConfig struct { + // DID or URL of the labeler service. Accepts did:web:... (resolved to + // the corresponding HTTPS host) or a raw http/https URL. Empty disables + // labeler integration. + DID string `yaml:"did" comment:"DID or URL of the ATProto labeler (e.g., did:web:labeler.atcr.io). Empty disables labeler integration."` + + // Grace window for reversibility. Until a takedown is older than this, + // the GC keeps blobs referenced even though their layer records were + // purged. After this window blobs become eligible for collection. + GraceWindow time.Duration `yaml:"grace_window" comment:"Reversibility window for takedowns. Blobs survive this long after a takedown so the action can be reversed. After this window the GC reclaims them. Default: 720h (30 days)."` +} + // ScannerConfig defines vulnerability scanner settings type ScannerConfig struct { // Shared secret for scanner WebSocket authentication. Empty disables scanning. @@ -271,6 +295,10 @@ func setHoldDefaults(v *viper.Viper) { v.SetDefault("scanner.secret", "") v.SetDefault("scanner.rescan_interval", "168h") // 7 days + // Labeler defaults + v.SetDefault("labeler.did", "") + v.SetDefault("labeler.grace_window", "720h") // 30 days + // Log shipper defaults v.SetDefault("log_shipper.batch_size", 100) v.SetDefault("log_shipper.flush_interval", "5s") diff --git a/pkg/hold/gc/gc.go b/pkg/hold/gc/gc.go index 36cace7..89700b5 100644 --- a/pkg/hold/gc/gc.go +++ b/pkg/hold/gc/gc.go @@ -83,6 +83,33 @@ type BackfillConfigsPreview struct { Duration time.Duration `json:"duration"` } +// TakedownGate is the interface the GC uses to consult the labeler cache when +// computing reachability. Defined here (rather than imported) to keep the GC +// package free of any direct dependency on the labeler package. +// +// IsTakenDown returns the takedown's creation timestamp and a boolean +// indicating whether the manifest URI is currently under takedown (either by +// per-manifest label or via a user-level label on its DID). +type TakedownGate interface { + IsTakenDown(manifestURI string) (cts time.Time, ok bool) +} + +// Option configures optional GC behavior. +type Option func(*GarbageCollector) + +// WithTakedownCache wires a takedown gate (typically the hold's labeler cache) +// and a grace window. When set, analyzeRecords protects blobs of taken-down +// manifests from collection until grace expires, and skips reconciliation of +// their layer records so the labeler-driven purge isn't undone. +// +// Passing a nil gate or a non-positive window leaves the GC behaving as before. +func WithTakedownCache(gate TakedownGate, graceWindow time.Duration) Option { + return func(gc *GarbageCollector) { + gc.takedownGate = gate + gc.takedownGrace = graceWindow + } +} + // GarbageCollector handles cleanup of orphaned blobs from storage type GarbageCollector struct { pds *pds.HoldPDS @@ -90,6 +117,11 @@ type GarbageCollector struct { cfg Config logger *slog.Logger + // takedownGate, if non-nil, is consulted in analyzeRecords to gate blob + // reachability and skip reconcile for taken-down manifests. + takedownGate TakedownGate + takedownGrace time.Duration + // stopCh signals the background goroutine to stop stopCh chan struct{} // wg tracks the background goroutine @@ -151,9 +183,10 @@ type analysisResult struct { totalRecords int } -// NewGarbageCollector creates a new GC instance -func NewGarbageCollector(holdPDS *pds.HoldPDS, s3svc *s3.S3Service, cfg Config) *GarbageCollector { - return &GarbageCollector{ +// NewGarbageCollector creates a new GC instance. Optional behavior (such as +// the labeler-aware takedown gate) is configured via Option arguments. +func NewGarbageCollector(holdPDS *pds.HoldPDS, s3svc *s3.S3Service, cfg Config, opts ...Option) *GarbageCollector { + gc := &GarbageCollector{ pds: holdPDS, s3: s3svc, cfg: cfg, @@ -161,6 +194,30 @@ func NewGarbageCollector(holdPDS *pds.HoldPDS, s3svc *s3.S3Service, cfg Config) stopCh: make(chan struct{}), predecessorCache: make(map[string]bool), } + for _, opt := range opts { + opt(gc) + } + return gc +} + +// isManifestTakenDown reports whether the labeler cache (if any) currently +// holds a takedown for this manifest URI, returning the takedown's cts so the +// caller can decide whether the grace window has elapsed. +func (gc *GarbageCollector) isManifestTakenDown(manifestURI string) (time.Time, bool) { + if gc.takedownGate == nil { + return time.Time{}, false + } + return gc.takedownGate.IsTakenDown(manifestURI) +} + +// takedownExpired reports whether a takedown's cts is older than the +// configured grace window. With a non-positive window every takedown is +// considered expired immediately (i.e. blobs are never protected). +func (gc *GarbageCollector) takedownExpired(cts time.Time) bool { + if gc.takedownGrace <= 0 { + return true + } + return time.Since(cts) > gc.takedownGrace } // tryStart attempts to mark GC as running. Returns false if already running. @@ -636,6 +693,31 @@ func (gc *GarbageCollector) analyzeRecords(ctx context.Context) (*analysisResult fetchedUsers[did] = true for _, m := range manifests { result.manifestsChecked++ + + // Labeler-aware reachability: + // - in-grace takedown: blobs stay referenced (so a reversal can + // restore content), but the manifest is NOT added to + // knownManifests so reconcileMissingRecords won't recreate the + // layer records the labeler subscriber just purged. + // - past-grace takedown: skip entirely — digests fall out of + // the referenced set and become eligible for blob GC. + if cts, taken := gc.isManifestTakenDown(m.URI); taken { + if !gc.takedownExpired(cts) { + for _, layer := range m.Record.Layers { + result.referenced[layer.Digest] = true + } + if m.Record.Config != nil && m.Record.Config.Digest != "" { + result.referenced[m.Record.Config.Digest] = true + } + gc.logger.Debug("Manifest under in-grace takedown: blobs protected, reconcile skipped", + "manifest", m.URI, "cts", cts) + } else { + gc.logger.Debug("Manifest takedown past grace: orphaning blobs", + "manifest", m.URI, "cts", cts) + } + continue + } + knownManifests[m.URI] = m // Add all layer digests to referenced set diff --git a/pkg/hold/gc/takedown_gate_test.go b/pkg/hold/gc/takedown_gate_test.go new file mode 100644 index 0000000..3f10c27 --- /dev/null +++ b/pkg/hold/gc/takedown_gate_test.go @@ -0,0 +1,68 @@ +package gc + +import ( + "testing" + "time" +) + +// stubGate satisfies TakedownGate for tests so we can exercise the GC's +// reachability decisions without standing up a full labeler cache or PDS. +type stubGate struct { + uri string + cts time.Time +} + +func (s stubGate) IsTakenDown(uri string) (time.Time, bool) { + if uri == s.uri { + return s.cts, true + } + return time.Time{}, false +} + +func TestIsManifestTakenDown(t *testing.T) { + t.Run("nil gate", func(t *testing.T) { + gc := &GarbageCollector{} + if _, ok := gc.isManifestTakenDown("at://x"); ok { + t.Fatalf("nil gate should report no takedowns") + } + }) + t.Run("matching uri", func(t *testing.T) { + cts := time.Now() + gc := &GarbageCollector{takedownGate: stubGate{uri: "at://x", cts: cts}} + got, ok := gc.isManifestTakenDown("at://x") + if !ok { + t.Fatalf("gate should report takedown for matching URI") + } + if !got.Equal(cts) { + t.Fatalf("cts = %v, want %v", got, cts) + } + }) + t.Run("non-matching uri", func(t *testing.T) { + gc := &GarbageCollector{takedownGate: stubGate{uri: "at://x", cts: time.Now()}} + if _, ok := gc.isManifestTakenDown("at://y"); ok { + t.Fatalf("non-matching URI should not report takedown") + } + }) +} + +func TestTakedownExpired(t *testing.T) { + tests := []struct { + name string + cts time.Time + grace time.Duration + expired bool + }{ + {"in-window", time.Now().Add(-time.Hour), 24 * time.Hour, false}, + {"past-window", time.Now().Add(-48 * time.Hour), 24 * time.Hour, true}, + {"zero grace expires immediately", time.Now(), 0, true}, + {"negative grace expires immediately", time.Now(), -time.Hour, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gc := &GarbageCollector{takedownGrace: tt.grace} + if got := gc.takedownExpired(tt.cts); got != tt.expired { + t.Fatalf("takedownExpired = %v, want %v", got, tt.expired) + } + }) + } +} diff --git a/pkg/hold/labeler/cache.go b/pkg/hold/labeler/cache.go new file mode 100644 index 0000000..0480cba --- /dev/null +++ b/pkg/hold/labeler/cache.go @@ -0,0 +1,177 @@ +// Package labeler provides a labeler subscription client for the hold service. +// +// The hold subscribes to one labeler and mirrors active takedowns into a local +// cache (in-memory + SQLite). On takedown receipt the hold purges layer, scan, +// and image-config records for the affected manifest. The cache is consulted +// by the GC to gate blob deletion: while a takedown is within its grace window, +// the manifest's blob digests stay in the GC's referenced set so reversal can +// still restore them. +package labeler + +import ( + "database/sql" + "fmt" + "strings" + "sync" + "time" +) + +// Cache holds active takedown URIs and their creation timestamps. It is +// thread-safe and persists state to SQLite so the hold can answer takedown +// queries before the labeler subscription has caught up after a restart. +type Cache struct { + mu sync.RWMutex + // manifest URI → cts. Includes both per-record URIs (at://did/coll/rkey) + // and per-DID URIs (at://did) for user-level takedowns. + entries map[string]time.Time + + db *sql.DB +} + +// NewCache opens (or creates) the takedown_cache and labeler_cursor tables on +// the given DB and loads any existing entries into memory. +func NewCache(db *sql.DB) (*Cache, error) { + c := &Cache{ + entries: make(map[string]time.Time), + db: db, + } + + stmts := []string{ + `CREATE TABLE IF NOT EXISTS takedown_cache ( + uri TEXT PRIMARY KEY, + src TEXT NOT NULL, + cts TIMESTAMP NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_takedown_cache_cts ON takedown_cache(cts)`, + `CREATE TABLE IF NOT EXISTS labeler_cursor ( + labeler_did TEXT PRIMARY KEY, + cursor INTEGER NOT NULL + )`, + } + for _, s := range stmts { + if _, err := db.Exec(s); err != nil { + return nil, fmt.Errorf("init labeler schema: %w", err) + } + } + + rows, err := db.Query(`SELECT uri, cts FROM takedown_cache`) + if err != nil { + return nil, fmt.Errorf("load takedown cache: %w", err) + } + defer rows.Close() + for rows.Next() { + var uri string + var cts time.Time + if err := rows.Scan(&uri, &cts); err != nil { + return nil, fmt.Errorf("scan takedown_cache row: %w", err) + } + c.entries[uri] = cts + } + return c, nil +} + +// Set records a positive takedown for uri at cts. Idempotent: re-applying +// updates the timestamp (newer takedowns win). +func (c *Cache) Set(uri, src string, cts time.Time) error { + c.mu.Lock() + c.entries[uri] = cts + c.mu.Unlock() + + _, err := c.db.Exec( + `INSERT INTO takedown_cache (uri, src, cts) VALUES (?, ?, ?) + ON CONFLICT(uri) DO UPDATE SET src = excluded.src, cts = excluded.cts`, + uri, src, cts, + ) + if err != nil { + return fmt.Errorf("persist takedown: %w", err) + } + return nil +} + +// Negate removes a takedown entry. Idempotent. +func (c *Cache) Negate(uri string) error { + c.mu.Lock() + delete(c.entries, uri) + c.mu.Unlock() + + _, err := c.db.Exec(`DELETE FROM takedown_cache WHERE uri = ?`, uri) + if err != nil { + return fmt.Errorf("delete takedown: %w", err) + } + return nil +} + +// IsTakenDown reports whether a manifest URI is taken down, either directly +// (per-manifest takedown) or via a user-level takedown on its DID. The returned +// timestamp is the earliest cts that applies (i.e. the longest-standing +// takedown), which is what the grace check should compare against. +func (c *Cache) IsTakenDown(manifestURI string) (cts time.Time, ok bool) { + c.mu.RLock() + defer c.mu.RUnlock() + + if t, has := c.entries[manifestURI]; has { + cts = t + ok = true + } + + // User-level: at:// with no path, applies to every record by that DID. + did := didFromManifestURI(manifestURI) + if did != "" { + userURI := "at://" + did + if t, has := c.entries[userURI]; has { + if !ok || t.Before(cts) { + cts = t + ok = true + } + } + } + return cts, ok +} + +// IsExpired returns true if cts is older than the grace window. +func IsExpired(cts time.Time, graceWindow time.Duration) bool { + if graceWindow <= 0 { + return true + } + return time.Since(cts) > graceWindow +} + +// GetCursor returns the last persisted cursor for a labeler DID (0 if none). +func (c *Cache) GetCursor(labelerDID string) (int64, error) { + var cursor int64 + err := c.db.QueryRow(`SELECT cursor FROM labeler_cursor WHERE labeler_did = ?`, labelerDID).Scan(&cursor) + if err == sql.ErrNoRows { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("read labeler cursor: %w", err) + } + return cursor, nil +} + +// SetCursor persists the cursor for a labeler DID. +func (c *Cache) SetCursor(labelerDID string, cursor int64) error { + _, err := c.db.Exec( + `INSERT INTO labeler_cursor (labeler_did, cursor) VALUES (?, ?) + ON CONFLICT(labeler_did) DO UPDATE SET cursor = excluded.cursor`, + labelerDID, cursor, + ) + if err != nil { + return fmt.Errorf("persist labeler cursor: %w", err) + } + return nil +} + +// didFromManifestURI extracts the authority (DID) from an at:// URI. +// Returns "" for malformed input. +func didFromManifestURI(uri string) string { + const prefix = "at://" + if !strings.HasPrefix(uri, prefix) { + return "" + } + rest := uri[len(prefix):] + if i := strings.IndexByte(rest, '/'); i >= 0 { + return rest[:i] + } + return rest +} diff --git a/pkg/hold/labeler/cache_test.go b/pkg/hold/labeler/cache_test.go new file mode 100644 index 0000000..be51e3d --- /dev/null +++ b/pkg/hold/labeler/cache_test.go @@ -0,0 +1,191 @@ +package labeler + +import ( + "database/sql" + "testing" + "time" + + _ "github.com/tursodatabase/go-libsql" +) + +func newTestCache(t *testing.T) *Cache { + t.Helper() + db, err := sql.Open("libsql", ":memory:") + if err != nil { + t.Fatalf("open in-memory db: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + c, err := NewCache(db) + if err != nil { + t.Fatalf("NewCache: %v", err) + } + return c +} + +func TestCacheSetAndIsTakenDown(t *testing.T) { + c := newTestCache(t) + uri := "at://did:plc:alice/io.atcr.manifest/abc" + cts := time.Now().UTC().Add(-time.Hour) + + if _, ok := c.IsTakenDown(uri); ok { + t.Fatalf("expected not taken down before Set") + } + if err := c.Set(uri, "did:web:labeler", cts); err != nil { + t.Fatalf("Set: %v", err) + } + + got, ok := c.IsTakenDown(uri) + if !ok { + t.Fatalf("expected taken down after Set") + } + if !got.Equal(cts) { + t.Fatalf("cts = %v, want %v", got, cts) + } +} + +func TestCacheNegateRemovesEntry(t *testing.T) { + c := newTestCache(t) + uri := "at://did:plc:alice/io.atcr.manifest/abc" + if err := c.Set(uri, "did:web:labeler", time.Now()); err != nil { + t.Fatal(err) + } + if err := c.Negate(uri); err != nil { + t.Fatalf("Negate: %v", err) + } + if _, ok := c.IsTakenDown(uri); ok { + t.Fatalf("expected not taken down after Negate") + } +} + +func TestCacheUserLevelTakedownAppliesToAllManifests(t *testing.T) { + c := newTestCache(t) + userURI := "at://did:plc:alice" + cts := time.Now().UTC() + + if err := c.Set(userURI, "did:web:labeler", cts); err != nil { + t.Fatal(err) + } + + manifestURI := "at://did:plc:alice/io.atcr.manifest/anything" + got, ok := c.IsTakenDown(manifestURI) + if !ok { + t.Fatalf("user-level takedown should apply to manifest URI") + } + if !got.Equal(cts) { + t.Fatalf("cts = %v, want %v", got, cts) + } + + otherURI := "at://did:plc:bob/io.atcr.manifest/x" + if _, ok := c.IsTakenDown(otherURI); ok { + t.Fatalf("user-level takedown for alice must not affect bob") + } +} + +func TestCacheChoosesEarliestCtsAcrossSources(t *testing.T) { + c := newTestCache(t) + earlier := time.Now().UTC().Add(-2 * time.Hour) + later := time.Now().UTC() + + manifestURI := "at://did:plc:alice/io.atcr.manifest/x" + userURI := "at://did:plc:alice" + + // Per-manifest later, user-level earlier — IsTakenDown should report the earlier one. + if err := c.Set(manifestURI, "did:web:labeler", later); err != nil { + t.Fatal(err) + } + if err := c.Set(userURI, "did:web:labeler", earlier); err != nil { + t.Fatal(err) + } + + got, ok := c.IsTakenDown(manifestURI) + if !ok { + t.Fatalf("expected taken down") + } + if !got.Equal(earlier) { + t.Fatalf("cts = %v, want earliest (%v)", got, earlier) + } +} + +func TestCachePersistsAcrossInstances(t *testing.T) { + db, err := sql.Open("libsql", ":memory:") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + first, err := NewCache(db) + if err != nil { + t.Fatalf("first NewCache: %v", err) + } + uri := "at://did:plc:alice/io.atcr.manifest/abc" + cts := time.Now().UTC() + if err := first.Set(uri, "did:web:labeler", cts); err != nil { + t.Fatal(err) + } + + // New cache instance, same DB — should warm-load entry. + second, err := NewCache(db) + if err != nil { + t.Fatalf("second NewCache: %v", err) + } + if _, ok := second.IsTakenDown(uri); !ok { + t.Fatalf("second cache should see persisted takedown") + } +} + +func TestCacheCursorRoundTrip(t *testing.T) { + c := newTestCache(t) + const labeler = "did:web:labeler" + + got, err := c.GetCursor(labeler) + if err != nil { + t.Fatal(err) + } + if got != 0 { + t.Fatalf("initial cursor = %d, want 0", got) + } + + if err := c.SetCursor(labeler, 42); err != nil { + t.Fatal(err) + } + got, err = c.GetCursor(labeler) + if err != nil { + t.Fatal(err) + } + if got != 42 { + t.Fatalf("cursor = %d, want 42", got) + } + + if err := c.SetCursor(labeler, 100); err != nil { + t.Fatal(err) + } + got, err = c.GetCursor(labeler) + if err != nil { + t.Fatal(err) + } + if got != 100 { + t.Fatalf("cursor after upsert = %d, want 100", got) + } +} + +func TestIsExpired(t *testing.T) { + tests := []struct { + name string + cts time.Time + grace time.Duration + expired bool + }{ + {"in-window", time.Now().Add(-time.Hour), 24 * time.Hour, false}, + {"past-window", time.Now().Add(-48 * time.Hour), 24 * time.Hour, true}, + {"zero grace expires immediately", time.Now(), 0, true}, + {"negative grace expires immediately", time.Now(), -time.Hour, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsExpired(tt.cts, tt.grace); got != tt.expired { + t.Fatalf("IsExpired = %v, want %v", got, tt.expired) + } + }) + } +} diff --git a/pkg/hold/labeler/subscriber.go b/pkg/hold/labeler/subscriber.go new file mode 100644 index 0000000..36df006 --- /dev/null +++ b/pkg/hold/labeler/subscriber.go @@ -0,0 +1,374 @@ +package labeler + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "net/url" + "strings" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/events" + "github.com/gorilla/websocket" +) + +// TakedownLabelValue is the label value the hold treats as a takedown trigger. +// Mirrors what pkg/labeler/takedown.go emits. +const TakedownLabelValue = "!takedown" + +// Purger is the subset of HoldPDS the subscriber needs to act on takedowns. +// Defined as an interface so tests can substitute a stub without standing up +// a full PDS, and to avoid an import cycle with pkg/hold/pds. +type Purger interface { + PurgeManifestRecords(ctx context.Context, manifestURI string) (PurgeOutcome, error) + PurgeUserManifests(ctx context.Context, userDID string) (PurgeOutcome, error) +} + +// PurgeOutcome mirrors pds.PurgeResult without creating an import cycle. +type PurgeOutcome struct { + LayersDeleted int + ScanDeleted bool + ImageConfigDeleted bool +} + +// Subscriber connects to a labeler's subscribeLabels endpoint, mirrors +// takedowns into the local cache, and triggers record purges on the hold. +type Subscriber struct { + labelerURL string + labelerDID string + cache *Cache + purger Purger + stopCh chan struct{} +} + +// NewSubscriber builds a subscriber for the given labeler. labelerDIDOrURL +// may be either: +// +// - a did:web identifier (e.g. did:web:labeler.atcr.io) → resolved to https://labeler.atcr.io +// - a raw http/https URL (e.g. http://172.28.0.4:5002 for dev) +// +// The websocket URL is derived from the resolved HTTPS endpoint; the +// labeler's DID (used to filter the Src field on incoming labels) is derived +// the same way the appview's labeler subscriber derives it, so a single +// config field suffices. +func NewSubscriber(labelerDIDOrURL string, cache *Cache, purger Purger) *Subscriber { + httpURL := parseLabelerURL(labelerDIDOrURL) + return &Subscriber{ + labelerURL: httpURL, + labelerDID: deriveLabelerDID(labelerDIDOrURL, httpURL), + cache: cache, + purger: purger, + stopCh: make(chan struct{}), + } +} + +// LabelerDID returns the DID derived from the labeler URL. Useful for the +// caller to log the trusted source. +func (s *Subscriber) LabelerDID() string { return s.labelerDID } + +// Start runs the subscription loop in a goroutine. +func (s *Subscriber) Start() { + go s.run() +} + +// Stop signals the subscriber to shut down. Safe to call once. +func (s *Subscriber) Stop() { + close(s.stopCh) +} + +func (s *Subscriber) run() { + backoff := time.Second + for { + select { + case <-s.stopCh: + return + default: + } + + if err := s.connect(); err != nil { + slog.Warn("Hold labeler subscription error, reconnecting", + "labeler", s.labelerURL, + "error", err, + "backoff", backoff, + ) + select { + case <-s.stopCh: + return + case <-time.After(backoff): + } + if backoff < 30*time.Second { + backoff *= 2 + } + } else { + backoff = time.Second + } + } +} + +func (s *Subscriber) connect() error { + cursor, err := s.cache.GetCursor(s.labelerDID) + if err != nil { + return fmt.Errorf("get cursor: %w", err) + } + + wsURL := toWebSocketURL(s.labelerURL) + "/xrpc/com.atproto.label.subscribeLabels" + if cursor > 0 { + wsURL += fmt.Sprintf("?cursor=%d", cursor) + } + + slog.Info("Hold connecting to labeler", "url", wsURL, "cursor", cursor) + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + return fmt.Errorf("websocket dial: %w", err) + } + defer conn.Close() + slog.Info("Hold connected to labeler", "url", s.labelerURL) + + for { + select { + case <-s.stopCh: + return nil + default: + } + + mt, payload, err := conn.ReadMessage() + if err != nil { + return fmt.Errorf("read: %w", err) + } + if mt != websocket.BinaryMessage { + slog.Warn("Hold labeler: ignoring non-binary frame", "type", mt) + continue + } + + seq, labels, err := decodeFrame(payload) + if err != nil { + if errors.Is(err, errInfoFrame) { + continue + } + return fmt.Errorf("decode frame: %w", err) + } + + for _, lbl := range labels { + s.applyLabel(seq, lbl) + } + + if err := s.cache.SetCursor(s.labelerDID, seq); err != nil { + slog.Warn("Hold labeler: failed to persist cursor", "seq", seq, "error", err) + } + } +} + +// applyLabel processes one label. Only takedown labels from a trusted source +// trigger cache mutations and record purges; everything else is ignored. +func (s *Subscriber) applyLabel(seq int64, lbl *comatproto.LabelDefs_Label) { + if lbl == nil { + return + } + if lbl.Val != TakedownLabelValue { + return + } + if !s.trustsSource(lbl.Src) { + slog.Debug("Hold labeler: ignoring untrusted source", + "src", lbl.Src, "uri", lbl.Uri, "seq", seq) + return + } + + cts, _ := time.Parse(time.RFC3339, lbl.Cts) + negated := lbl.Neg != nil && *lbl.Neg + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + if negated { + if err := s.cache.Negate(lbl.Uri); err != nil { + slog.Warn("Hold labeler: failed to drop takedown from cache", + "uri", lbl.Uri, "error", err) + return + } + slog.Info("Hold labeler: takedown reversed", "uri", lbl.Uri, "seq", seq) + return + } + + if err := s.cache.Set(lbl.Uri, lbl.Src, cts); err != nil { + slog.Warn("Hold labeler: failed to record takedown", + "uri", lbl.Uri, "error", err) + return + } + + // User-level vs per-record. The labeler emits per-record labels for every + // individual manifest in a repo-level takedown plus a summary; for a + // user-level takedown only at:// is emitted. We dispatch on shape so + // we don't try to PurgeManifestRecords on a non-manifest URI. + switch shape := classifyURI(lbl.Uri); shape.Kind { + case uriKindManifest: + out, err := s.purger.PurgeManifestRecords(ctx, lbl.Uri) + if err != nil { + slog.Warn("Hold labeler: purge failed", "uri", lbl.Uri, "error", err) + return + } + slog.Info("Hold labeler: purged manifest on takedown", + "uri", lbl.Uri, "layers", out.LayersDeleted, + "scan", out.ScanDeleted, "config", out.ImageConfigDeleted) + case uriKindUser: + out, err := s.purger.PurgeUserManifests(ctx, shape.DID) + if err != nil { + slog.Warn("Hold labeler: user-level purge failed", + "did", shape.DID, "error", err) + return + } + slog.Info("Hold labeler: purged all manifests for user on takedown", + "did", shape.DID, "layers", out.LayersDeleted) + default: + // Repo-level summary URI (at://did/io.atcr.repo/) and other + // non-record subjects: cached for GC's reachability check, but no + // records to purge directly. The per-manifest labels in the same + // stream do the actual record removal. + slog.Debug("Hold labeler: cached non-record takedown", + "uri", lbl.Uri, "kind", string(shape.Kind)) + } +} + +// uriKind classifies the shape of a label subject URI. +type uriKind string + +const ( + uriKindManifest uriKind = "manifest" + uriKindUser uriKind = "user" + uriKindOther uriKind = "other" +) + +type uriShape struct { + Kind uriKind + DID string +} + +// classifyURI inspects an at:// URI and reports whether it points at a single +// manifest record, an entire user, or something else (repo summary etc.). +func classifyURI(uri string) uriShape { + const prefix = "at://" + if !strings.HasPrefix(uri, prefix) { + return uriShape{Kind: uriKindOther} + } + rest := uri[len(prefix):] + parts := strings.SplitN(rest, "/", 3) + if len(parts) == 0 || parts[0] == "" { + return uriShape{Kind: uriKindOther} + } + did := parts[0] + if len(parts) == 1 { + return uriShape{Kind: uriKindUser, DID: did} + } + if len(parts) == 3 && parts[1] == "io.atcr.manifest" && parts[2] != "" { + return uriShape{Kind: uriKindManifest, DID: did} + } + return uriShape{Kind: uriKindOther, DID: did} +} + +// errInfoFrame signals that the frame was an info frame and the caller should +// continue without treating it as a label or an error. +var errInfoFrame = errors.New("hold labeler: info frame") + +// decodeFrame parses a single subscribeLabels binary frame. +// +// ATProto event-stream framing is two concatenated CBOR objects: a {op,t} +// header and a body. We dispatch on op/t; for op=1, t="#labels" we return the +// labels body. #info frames are logged and signal errInfoFrame so the caller +// loops; error frames become Go errors so the run loop reconnects. +func decodeFrame(payload []byte) (int64, []*comatproto.LabelDefs_Label, error) { + r := bytes.NewReader(payload) + var header events.EventHeader + if err := header.UnmarshalCBOR(r); err != nil { + return 0, nil, fmt.Errorf("unmarshal header: %w", err) + } + + switch { + case header.Op == events.EvtKindErrorFrame: + var ef events.ErrorFrame + if err := ef.UnmarshalCBOR(r); err != nil { + return 0, nil, fmt.Errorf("unmarshal error frame: %w", err) + } + return 0, nil, fmt.Errorf("labeler error frame: %s — %s", ef.Error, ef.Message) + case header.Op == events.EvtKindMessage && header.MsgType == "#labels": + var body comatproto.LabelSubscribeLabels_Labels + if err := body.UnmarshalCBOR(r); err != nil { + return 0, nil, fmt.Errorf("unmarshal labels body: %w", err) + } + return body.Seq, body.Labels, nil + case header.Op == events.EvtKindMessage && header.MsgType == "#info": + var info comatproto.LabelSubscribeLabels_Info + if err := info.UnmarshalCBOR(r); err != nil { + return 0, nil, fmt.Errorf("unmarshal info body: %w", err) + } + message := "" + if info.Message != nil { + message = *info.Message + } + slog.Info("Hold labeler: info frame", "name", info.Name, "message", message) + return 0, nil, errInfoFrame + default: + return 0, nil, fmt.Errorf("unexpected frame op=%d t=%q", header.Op, header.MsgType) + } +} + +// trustsSource reports whether labels with the given Src DID should be +// honored. Today this is "matches the configured labeler DID" — a single +// trusted source. If we ever want a list, this becomes a set membership +// check without touching callers. +func (s *Subscriber) trustsSource(src string) bool { + return src == s.labelerDID +} + +// parseLabelerURL accepts either a did:web:... identifier or a raw http/https +// URL and returns the HTTPS (or HTTP for did:web pointing at a hostname with +// %3A-encoded port in test mode) endpoint to talk to. did:web hosts with +// %3A-encoded ports are decoded back to colons. Mirrors the appview's +// ParseLabelerURL so a single config field works in both places. +func parseLabelerURL(labelerDIDOrURL string) string { + if strings.HasPrefix(labelerDIDOrURL, "http://") || strings.HasPrefix(labelerDIDOrURL, "https://") { + return labelerDIDOrURL + } + if strings.HasPrefix(labelerDIDOrURL, "did:web:") { + host := strings.TrimPrefix(labelerDIDOrURL, "did:web:") + host = strings.ReplaceAll(host, "%3A", ":") + return "https://" + host + } + return labelerDIDOrURL +} + +// deriveLabelerDID returns the canonical labeler DID for source filtering. +// When the operator gave us a did:web identifier directly, we use it as-is. +// When they gave us a URL, we derive a did:web from its host (so dev URLs +// like http://172.28.0.4:5002 yield did:web:172.28.0.4%3A5002, matching the +// labeler's own self-served identity). +func deriveLabelerDID(labelerDIDOrURL, httpURL string) string { + if strings.HasPrefix(labelerDIDOrURL, "did:") { + return labelerDIDOrURL + } + u, err := url.Parse(httpURL) + if err != nil { + return labelerDIDOrURL + } + host := u.Hostname() + if port := u.Port(); port != "" { + host += "%3A" + port + } + return "did:web:" + host +} + +// toWebSocketURL converts an HTTP URL to a WebSocket URL. http→ws, https→wss. +func toWebSocketURL(httpURL string) string { + u, err := url.Parse(httpURL) + if err != nil { + return httpURL + } + switch u.Scheme { + case "https": + u.Scheme = "wss" + default: + u.Scheme = "ws" + } + return u.String() +} diff --git a/pkg/hold/labeler/subscriber_test.go b/pkg/hold/labeler/subscriber_test.go new file mode 100644 index 0000000..fd8216d --- /dev/null +++ b/pkg/hold/labeler/subscriber_test.go @@ -0,0 +1,210 @@ +package labeler + +import ( + "context" + "sync" + "testing" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" +) + +// stubPurger captures purge calls so we can assert what the subscriber routed. +type stubPurger struct { + mu sync.Mutex + manifestCalls []string + userLevelCalls []string + purgeManifestErr error + purgeUserError error + manifestOutcome PurgeOutcome + userLevelOutcome PurgeOutcome +} + +func (s *stubPurger) PurgeManifestRecords(_ context.Context, uri string) (PurgeOutcome, error) { + s.mu.Lock() + s.manifestCalls = append(s.manifestCalls, uri) + s.mu.Unlock() + return s.manifestOutcome, s.purgeManifestErr +} + +func (s *stubPurger) PurgeUserManifests(_ context.Context, did string) (PurgeOutcome, error) { + s.mu.Lock() + s.userLevelCalls = append(s.userLevelCalls, did) + s.mu.Unlock() + return s.userLevelOutcome, s.purgeUserError +} + +func (s *stubPurger) snapshot() (manifests, users []string) { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.manifestCalls...), append([]string(nil), s.userLevelCalls...) +} + +func ptrBool(b bool) *bool { return &b } + +func TestApplyLabelManifestTakedownPurges(t *testing.T) { + cache := newTestCache(t) + purger := &stubPurger{} + // Use a did:web identifier directly so we control exactly what Src must + // match — derived URL is https://labeler.example.com. + sub := NewSubscriber("did:web:labeler.example.com", cache, purger) + + uri := "at://did:plc:alice/io.atcr.manifest/abc" + cts := time.Now().UTC().Format(time.RFC3339) + + sub.applyLabel(1, &comatproto.LabelDefs_Label{ + Src: "did:web:labeler.example.com", + Uri: uri, + Val: TakedownLabelValue, + Cts: cts, + }) + + manifests, users := purger.snapshot() + if len(manifests) != 1 || manifests[0] != uri { + t.Fatalf("manifest purges = %v, want [%q]", manifests, uri) + } + if len(users) != 0 { + t.Fatalf("expected no user-level purges, got %v", users) + } + if _, ok := cache.IsTakenDown(uri); !ok { + t.Fatalf("cache should record the takedown") + } +} + +func TestApplyLabelUserLevelTakedownPurgesAllManifests(t *testing.T) { + cache := newTestCache(t) + purger := &stubPurger{} + sub := NewSubscriber("did:web:labeler.example.com", cache, purger) + + uri := "at://did:plc:alice" + sub.applyLabel(1, &comatproto.LabelDefs_Label{ + Src: "did:web:labeler.example.com", + Uri: uri, + Val: TakedownLabelValue, + Cts: time.Now().UTC().Format(time.RFC3339), + }) + + manifests, users := purger.snapshot() + if len(users) != 1 || users[0] != "did:plc:alice" { + t.Fatalf("user-level purges = %v, want [did:plc:alice]", users) + } + if len(manifests) != 0 { + t.Fatalf("expected no per-manifest purges, got %v", manifests) + } + if _, ok := cache.IsTakenDown("at://did:plc:alice/io.atcr.manifest/anything"); !ok { + t.Fatalf("user-level entry should mask any manifest URI for that DID") + } +} + +func TestApplyLabelNegationDropsCacheNoPurge(t *testing.T) { + cache := newTestCache(t) + uri := "at://did:plc:alice/io.atcr.manifest/abc" + if err := cache.Set(uri, "did:web:labeler.example.com", time.Now()); err != nil { + t.Fatal(err) + } + purger := &stubPurger{} + sub := NewSubscriber("did:web:labeler.example.com", cache, purger) + + sub.applyLabel(2, &comatproto.LabelDefs_Label{ + Src: "did:web:labeler.example.com", + Uri: uri, + Val: TakedownLabelValue, + Neg: ptrBool(true), + Cts: time.Now().UTC().Format(time.RFC3339), + }) + + if _, ok := cache.IsTakenDown(uri); ok { + t.Fatalf("negation should drop the takedown from cache") + } + manifests, _ := purger.snapshot() + if len(manifests) != 0 { + t.Fatalf("negation must not trigger purge, got %v", manifests) + } +} + +func TestApplyLabelIgnoresUntrustedSource(t *testing.T) { + cache := newTestCache(t) + purger := &stubPurger{} + // Subscriber is configured for did:web:operator; a label whose Src is a + // different DID must be ignored (today's single-DID trust model). + sub := NewSubscriber("did:web:operator", cache, purger) + + sub.applyLabel(1, &comatproto.LabelDefs_Label{ + Src: "did:web:rogue", + Uri: "at://did:plc:alice/io.atcr.manifest/abc", + Val: TakedownLabelValue, + Cts: time.Now().UTC().Format(time.RFC3339), + }) + + manifests, users := purger.snapshot() + if len(manifests) != 0 || len(users) != 0 { + t.Fatalf("untrusted source must not trigger purge: manifests=%v users=%v", manifests, users) + } +} + +func TestSubscriberDerivesDIDFromURL(t *testing.T) { + tests := []struct { + input string + wantURL string + wantDID string + }{ + {"did:web:labeler.atcr.io", "https://labeler.atcr.io", "did:web:labeler.atcr.io"}, + {"did:web:172.28.0.4%3A5002", "https://172.28.0.4:5002", "did:web:172.28.0.4%3A5002"}, + {"http://172.28.0.4:5002", "http://172.28.0.4:5002", "did:web:172.28.0.4%3A5002"}, + {"https://labeler.atcr.io", "https://labeler.atcr.io", "did:web:labeler.atcr.io"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + sub := NewSubscriber(tt.input, nil, nil) + if sub.labelerURL != tt.wantURL { + t.Errorf("labelerURL = %q, want %q", sub.labelerURL, tt.wantURL) + } + if sub.labelerDID != tt.wantDID { + t.Errorf("labelerDID = %q, want %q", sub.labelerDID, tt.wantDID) + } + }) + } +} + +func TestApplyLabelIgnoresNonTakedownValues(t *testing.T) { + cache := newTestCache(t) + purger := &stubPurger{} + sub := NewSubscriber("did:web:labeler.example.com", cache, purger) + + sub.applyLabel(1, &comatproto.LabelDefs_Label{ + Src: "did:web:labeler.example.com", + Uri: "at://did:plc:alice/io.atcr.manifest/abc", + Val: "spam", + Cts: time.Now().UTC().Format(time.RFC3339), + }) + + manifests, users := purger.snapshot() + if len(manifests) != 0 || len(users) != 0 { + t.Fatalf("non-takedown labels must not trigger purge: manifests=%v users=%v", manifests, users) + } +} + +func TestClassifyURI(t *testing.T) { + tests := []struct { + uri string + kind uriKind + did string + }{ + {"at://did:plc:alice/io.atcr.manifest/abc", uriKindManifest, "did:plc:alice"}, + {"at://did:plc:alice", uriKindUser, "did:plc:alice"}, + {"at://did:plc:alice/io.atcr.repo/myimage", uriKindOther, "did:plc:alice"}, + {"https://example.com", uriKindOther, ""}, + {"", uriKindOther, ""}, + } + for _, tt := range tests { + t.Run(tt.uri, func(t *testing.T) { + got := classifyURI(tt.uri) + if got.Kind != tt.kind { + t.Fatalf("kind = %s, want %s", got.Kind, tt.kind) + } + if got.DID != tt.did { + t.Fatalf("did = %s, want %s", got.DID, tt.did) + } + }) + } +} diff --git a/pkg/hold/pds/purge.go b/pkg/hold/pds/purge.go new file mode 100644 index 0000000..087e4ed --- /dev/null +++ b/pkg/hold/pds/purge.go @@ -0,0 +1,254 @@ +package pds + +import ( + "context" + "fmt" + "log/slog" + "strings" + + "atcr.io/pkg/atproto" + lexutil "github.com/bluesky-social/indigo/lex/util" + "github.com/bluesky-social/indigo/repo" + "github.com/ipfs/go-cid" +) + +// PurgeResult summarizes a PurgeManifestRecords call. +type PurgeResult struct { + LayersDeleted int + ScanDeleted bool + ImageConfigDeleted bool +} + +// PurgeManifestRecords removes layer, scan, and image-config records associated +// with a single manifest AT-URI. The manifest record itself lives in the user's +// PDS and is not touched. S3 blobs are not removed; the GC handles them based +// on remaining references and the labeler grace window. +// +// Idempotent: missing records are not errors. Used by both the +// io.atcr.hold.purgeManifest XRPC handler (called by the appview on UI delete) +// and the hold's labeler subscriber (called on takedown receipt). +func (p *HoldPDS) PurgeManifestRecords(ctx context.Context, manifestURI string) (*PurgeResult, error) { + if manifestURI == "" { + return nil, fmt.Errorf("manifest URI is required") + } + + res := &PurgeResult{} + + // Layer records: TID rkeys, multiple per manifest. Look up the rkeys by + // scanning the layer index and matching the manifest field. + rkeys, err := p.listLayerRkeysForManifest(ctx, manifestURI) + if err != nil { + return res, fmt.Errorf("list layer rkeys: %w", err) + } + for _, rkey := range rkeys { + if err := p.DeleteLayerRecord(ctx, rkey); err != nil { + slog.Warn("Failed to delete layer record during purge", + "rkey", rkey, "manifest", manifestURI, "error", err) + continue + } + res.LayersDeleted++ + } + + // Scan + image-config records share a deterministic rkey scheme based on + // the manifest digest, so we can address them directly. + manifestDigest, err := atproto.ParseManifestURI(manifestURI) + if err != nil { + // Without a parseable digest we can't compute the deterministic rkey. + // Layer records may still have been deleted above; return what we did. + slog.Warn("Cannot parse manifest URI for scan/config purge", + "manifest", manifestURI, "error", err) + return res, nil + } + rkey := atproto.ScanRecordKey(manifestDigest) + res.ScanDeleted = p.tryDeleteRecord(ctx, atproto.ScanCollection, rkey) + res.ImageConfigDeleted = p.tryDeleteRecord(ctx, atproto.ImageConfigCollection, rkey) + + slog.Info("Purged manifest records", + "manifest", manifestURI, + "layers", res.LayersDeleted, + "scan", res.ScanDeleted, + "imageConfig", res.ImageConfigDeleted, + ) + return res, nil +} + +// PurgeUserManifests purges every manifest's records for a given DID. Used by +// user-level takedowns (URI = at://) where the labeler has not enumerated +// individual manifest URIs. +// +// Discovers the set of manifest URIs from the layer index (every layer record +// names its manifest) so we can reuse PurgeManifestRecords for each. +func (p *HoldPDS) PurgeUserManifests(ctx context.Context, userDID string) (*PurgeResult, error) { + if userDID == "" { + return nil, fmt.Errorf("user DID is required") + } + if p.recordsIndex == nil { + return nil, fmt.Errorf("records index not available") + } + + manifestURIs, err := p.collectUserManifestURIs(ctx, userDID) + if err != nil { + return nil, fmt.Errorf("collect manifest URIs: %w", err) + } + + combined := &PurgeResult{} + for uri := range manifestURIs { + r, err := p.PurgeManifestRecords(ctx, uri) + if err != nil { + slog.Warn("Failed to purge manifest in user-level purge", + "manifest", uri, "user", userDID, "error", err) + continue + } + combined.LayersDeleted += r.LayersDeleted + if r.ScanDeleted { + combined.ScanDeleted = true + } + if r.ImageConfigDeleted { + combined.ImageConfigDeleted = true + } + } + + slog.Info("Purged all manifests for user", + "user", userDID, + "manifestCount", len(manifestURIs), + "layersDeleted", combined.LayersDeleted, + ) + return combined, nil +} + +// listLayerRkeysForManifest scans the layer record index, decodes each record +// from the carstore, and returns the rkeys whose `manifest` field matches. +// +// Mirrors ListLayerRecordsForManifest but returns rkeys (which the caller +// needs for deletion) instead of decoded record values. +func (p *HoldPDS) listLayerRkeysForManifest(ctx context.Context, manifestURI string) ([]string, error) { + if p.recordsIndex == nil { + return nil, fmt.Errorf("records index not available") + } + + session, err := p.carstore.ReadOnlySession(p.uid) + if err != nil { + return nil, fmt.Errorf("create session: %w", err) + } + head, err := p.carstore.GetUserRepoHead(ctx, p.uid) + if err != nil { + return nil, fmt.Errorf("get repo head: %w", err) + } + if !head.Defined() { + return nil, nil + } + repoHandle, err := repo.OpenRepo(ctx, session, head) + if err != nil { + return nil, fmt.Errorf("open repo: %w", err) + } + + var rkeys []string + cursor := "" + const batch = 1000 + for { + indexRecords, nextCursor, err := p.recordsIndex.ListRecords(atproto.LayerCollection, batch, cursor, false) + if err != nil { + return nil, fmt.Errorf("list layer records: %w", err) + } + for _, rec := range indexRecords { + path := rec.Collection + "/" + rec.Rkey + _, recBytes, err := repoHandle.GetRecordBytes(ctx, path) + if err != nil { + continue + } + val, err := lexutil.CborDecodeValue(*recBytes) + if err != nil { + continue + } + layer, ok := val.(*atproto.LayerRecord) + if !ok { + continue + } + if layer.Manifest == manifestURI { + rkeys = append(rkeys, rec.Rkey) + } + } + if nextCursor == "" { + break + } + cursor = nextCursor + } + return rkeys, nil +} + +// collectUserManifestURIs walks the user's layer records and returns the set +// of unique manifest AT-URIs they reference. +func (p *HoldPDS) collectUserManifestURIs(ctx context.Context, userDID string) (map[string]struct{}, error) { + uris := make(map[string]struct{}) + + session, err := p.carstore.ReadOnlySession(p.uid) + if err != nil { + return nil, fmt.Errorf("create session: %w", err) + } + head, err := p.carstore.GetUserRepoHead(ctx, p.uid) + if err != nil { + return nil, fmt.Errorf("get repo head: %w", err) + } + if !head.Defined() { + return uris, nil + } + repoHandle, err := repo.OpenRepo(ctx, session, head) + if err != nil { + return nil, fmt.Errorf("open repo: %w", err) + } + + cursor := "" + const batch = 200 + for { + records, nextCursor, err := p.recordsIndex.ListRecordsByDID(atproto.LayerCollection, userDID, batch, cursor) + if err != nil { + return nil, fmt.Errorf("list layer records by DID: %w", err) + } + for _, rec := range records { + path := rec.Collection + "/" + rec.Rkey + _, recBytes, err := repoHandle.GetRecordBytes(ctx, path) + if err != nil { + continue + } + val, err := lexutil.CborDecodeValue(*recBytes) + if err != nil { + continue + } + layer, ok := val.(*atproto.LayerRecord) + if !ok { + continue + } + if layer.Manifest != "" && strings.HasPrefix(layer.Manifest, "at://"+userDID+"/") { + uris[layer.Manifest] = struct{}{} + } + } + if nextCursor == "" { + break + } + cursor = nextCursor + } + return uris, nil +} + +// tryDeleteRecord deletes a record at the given collection/rkey from both the +// repo and the records index. Returns true if a record was actually deleted, +// false if it didn't exist or the delete failed (failures are logged). +func (p *HoldPDS) tryDeleteRecord(ctx context.Context, collection, rkey string) bool { + // Probe via GetRecord — if the record doesn't exist we skip silently. + if _, _, err := p.repomgr.GetRecord(ctx, p.uid, collection, rkey, cid.Undef); err != nil { + return false + } + + if err := p.repomgr.DeleteRecord(ctx, p.uid, collection, rkey); err != nil { + slog.Warn("Failed to delete record from repo", + "collection", collection, "rkey", rkey, "error", err) + return false + } + if p.recordsIndex != nil { + if err := p.recordsIndex.DeleteRecord(collection, rkey); err != nil { + slog.Warn("Failed to delete record from index", + "collection", collection, "rkey", rkey, "error", err) + } + } + return true +} diff --git a/pkg/hold/pds/purge_test.go b/pkg/hold/pds/purge_test.go new file mode 100644 index 0000000..96bc5ed --- /dev/null +++ b/pkg/hold/pds/purge_test.go @@ -0,0 +1,150 @@ +package pds + +import ( + "strings" + "testing" + + "atcr.io/pkg/atproto" +) + +func TestPurgeManifestRecordsRemovesAll(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + ctx := sharedCtx + + const userDID = "did:plc:alice" + const manifestDigest = "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f" + manifestURI := atproto.BuildManifestURI(userDID, manifestDigest) + + // Two layer records for the same manifest plus one for a different one + mustCreateLayer(t, pds, manifestURI, "sha256:layer-a", 1024) + mustCreateLayer(t, pds, manifestURI, "sha256:layer-b", 2048) + otherURI := atproto.BuildManifestURI(userDID, "sha256:0123456789abcdef") + mustCreateLayer(t, pds, otherURI, "sha256:layer-c", 4096) + + // Scan record at the deterministic rkey + scanRkey := atproto.ScanRecordKey(manifestDigest) + scanRec := &atproto.ScanRecord{ + Type: atproto.ScanCollection, + Manifest: manifestURI, + ScannedAt: "2026-05-02T00:00:00Z", + } + if _, _, _, err := pds.repomgr.UpsertRecord(ctx, pds.uid, atproto.ScanCollection, scanRkey, scanRec); err != nil { + t.Fatalf("create scan record: %v", err) + } + + // Image config record at the same deterministic rkey scheme + cfgRec := &atproto.ImageConfigRecord{ + Type: atproto.ImageConfigCollection, + Manifest: manifestURI, + } + if _, _, _, err := pds.repomgr.UpsertRecord(ctx, pds.uid, atproto.ImageConfigCollection, scanRkey, cfgRec); err != nil { + t.Fatalf("create image config record: %v", err) + } + + res, err := pds.PurgeManifestRecords(ctx, manifestURI) + if err != nil { + t.Fatalf("PurgeManifestRecords: %v", err) + } + if res.LayersDeleted != 2 { + t.Errorf("LayersDeleted = %d, want 2", res.LayersDeleted) + } + if !res.ScanDeleted { + t.Errorf("ScanDeleted = false, want true") + } + if !res.ImageConfigDeleted { + t.Errorf("ImageConfigDeleted = false, want true") + } + + // Layer records for OTHER manifest must remain. + rkeys, err := pds.listLayerRkeysForManifest(ctx, otherURI) + if err != nil { + t.Fatalf("list other layers: %v", err) + } + if len(rkeys) != 1 { + t.Fatalf("other manifest layer rkeys = %d, want 1", len(rkeys)) + } +} + +func TestPurgeManifestRecordsIdempotent(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + ctx := sharedCtx + + manifestURI := atproto.BuildManifestURI("did:plc:alice", "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f") + + res, err := pds.PurgeManifestRecords(ctx, manifestURI) + if err != nil { + t.Fatalf("first PurgeManifestRecords: %v", err) + } + if res.LayersDeleted != 0 || res.ScanDeleted || res.ImageConfigDeleted { + t.Fatalf("expected zero-result purge on empty hold, got %+v", res) + } + + // Second call must also succeed (idempotent). + if _, err := pds.PurgeManifestRecords(ctx, manifestURI); err != nil { + t.Fatalf("second PurgeManifestRecords: %v", err) + } +} + +func TestPurgeManifestRecordsRequiresURI(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + if _, err := pds.PurgeManifestRecords(sharedCtx, ""); err == nil || !strings.Contains(err.Error(), "manifest URI is required") { + t.Fatalf("expected manifest URI error, got %v", err) + } +} + +func TestPurgeUserManifestsCollectsAcrossManifests(t *testing.T) { + pds := setupTestPDSWithIndex(t, "did:plc:owner") + ctx := sharedCtx + + const userDID = "did:plc:alice" + manifestA := atproto.BuildManifestURI(userDID, "sha256:e692418e4cbaf90ca69d05a66403747baa33ee08806650b51fab815ad7fc331f") + manifestB := atproto.BuildManifestURI(userDID, "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef") + + mustCreateLayer(t, pds, manifestA, "sha256:layer-a", 100) + mustCreateLayer(t, pds, manifestA, "sha256:layer-b", 200) + mustCreateLayer(t, pds, manifestB, "sha256:layer-c", 300) + + // Different user's layer must survive + manifestOther := atproto.BuildManifestURI("did:plc:bob", "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + mustCreateLayer(t, pds, manifestOther, "sha256:layer-other", 400) + + res, err := pds.PurgeUserManifests(ctx, userDID) + if err != nil { + t.Fatalf("PurgeUserManifests: %v", err) + } + if res.LayersDeleted != 3 { + t.Errorf("LayersDeleted = %d, want 3", res.LayersDeleted) + } + + bobRkeys, err := pds.listLayerRkeysForManifest(ctx, manifestOther) + if err != nil { + t.Fatalf("list bob's layers: %v", err) + } + if len(bobRkeys) != 1 { + t.Fatalf("bob's layer count = %d, want 1 (purge crossed user boundary)", len(bobRkeys)) + } +} + +// mustCreateLayer is a tiny helper for purge tests — keeps the table-driven +// TestPurge body focused on the assertions. +func mustCreateLayer(t *testing.T, pds *HoldPDS, manifestURI, digest string, size int64) { + t.Helper() + rec := atproto.NewLayerRecord(digest, size, "application/vnd.oci.image.layer.v1.tar+gzip", didFromManifest(manifestURI), manifestURI) + if _, _, err := pds.CreateLayerRecord(sharedCtx, rec); err != nil { + t.Fatalf("CreateLayerRecord(%s): %v", digest, err) + } +} + +// didFromManifest pulls the authority out of an at:// URI for layer record +// construction in tests. Production code uses richer helpers. +func didFromManifest(uri string) string { + const prefix = "at://" + if !strings.HasPrefix(uri, prefix) { + return "" + } + rest := uri[len(prefix):] + if i := strings.IndexByte(rest, '/'); i >= 0 { + return rest[:i] + } + return rest +} diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index 233b57b..cd26adb 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -203,6 +203,7 @@ func (h *XRPCHandler) RegisterHandlers(r chi.Router) { r.Use(h.requireOwnerOrCrewAdmin) r.Post(atproto.RepoDeleteRecord, h.HandleDeleteRecord) r.Post(atproto.RepoUploadBlob, h.HandleUploadBlob) + r.Post(atproto.HoldPurgeManifest, h.HandlePurgeManifest) }) // Auth-only endpoints (DPoP auth) @@ -891,6 +892,42 @@ func (h *XRPCHandler) HandleDeleteRecord(w http.ResponseWriter, r *http.Request) }) } +// HandlePurgeManifest deletes layer, scan, and image-config records associated +// with a single manifest AT-URI. Idempotent. Auth: owner or crew admin +// (enforced by middleware). The manifest record itself lives in the user's PDS +// and is not affected. S3 blobs are not removed; the GC handles those based on +// remaining references and the labeler grace window. +func (h *XRPCHandler) HandlePurgeManifest(w http.ResponseWriter, r *http.Request) { + var input struct { + ManifestURI string `json:"manifestUri"` + } + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + http.Error(w, fmt.Sprintf("invalid JSON body: %v", err), http.StatusBadRequest) + return + } + if input.ManifestURI == "" { + http.Error(w, "manifestUri is required", http.StatusBadRequest) + return + } + if !strings.HasPrefix(input.ManifestURI, "at://") { + http.Error(w, "manifestUri must be an at:// URI", http.StatusBadRequest) + return + } + + res, err := h.pds.PurgeManifestRecords(r.Context(), input.ManifestURI) + if err != nil { + http.Error(w, fmt.Sprintf("purge failed: %v", err), http.StatusInternalServerError) + return + } + + render.JSON(w, r, map[string]any{ + "success": true, + "layersDeleted": res.LayersDeleted, + "scanDeleted": res.ScanDeleted, + "imageConfigDeleted": res.ImageConfigDeleted, + }) +} + // HandleSyncGetRecord returns a single record as a CAR file for sync func (h *XRPCHandler) HandleSyncGetRecord(w http.ResponseWriter, r *http.Request) { did := r.URL.Query().Get("did") diff --git a/pkg/hold/server.go b/pkg/hold/server.go index f47183f..ffc62f9 100644 --- a/pkg/hold/server.go +++ b/pkg/hold/server.go @@ -15,6 +15,7 @@ import ( "atcr.io/pkg/hold/admin" holddb "atcr.io/pkg/hold/db" "atcr.io/pkg/hold/gc" + holdlabeler "atcr.io/pkg/hold/labeler" "atcr.io/pkg/hold/oci" "atcr.io/pkg/hold/pds" "atcr.io/pkg/hold/quota" @@ -25,6 +26,36 @@ import ( "github.com/go-chi/chi/v5/middleware" ) +// purgerAdapter bridges *pds.HoldPDS to the holdlabeler.Purger interface, which +// uses its own outcome type to avoid an import cycle with pkg/hold/pds. +type purgerAdapter struct { + pds *pds.HoldPDS +} + +func (a purgerAdapter) PurgeManifestRecords(ctx context.Context, manifestURI string) (holdlabeler.PurgeOutcome, error) { + r, err := a.pds.PurgeManifestRecords(ctx, manifestURI) + if err != nil || r == nil { + return holdlabeler.PurgeOutcome{}, err + } + return holdlabeler.PurgeOutcome{ + LayersDeleted: r.LayersDeleted, + ScanDeleted: r.ScanDeleted, + ImageConfigDeleted: r.ImageConfigDeleted, + }, nil +} + +func (a purgerAdapter) PurgeUserManifests(ctx context.Context, userDID string) (holdlabeler.PurgeOutcome, error) { + r, err := a.pds.PurgeUserManifests(ctx, userDID) + if err != nil || r == nil { + return holdlabeler.PurgeOutcome{}, err + } + return holdlabeler.PurgeOutcome{ + LayersDeleted: r.LayersDeleted, + ScanDeleted: r.ScanDeleted, + ImageConfigDeleted: r.ImageConfigDeleted, + }, nil +} + // HoldServer is the hold service with an exposed router for extensibility. // Consumers can add routes to Router before calling Serve(). type HoldServer struct { @@ -41,12 +72,14 @@ type HoldServer struct { Config *Config // internal fields for shutdown - httpServer *http.Server - broadcaster *pds.EventBroadcaster - scanBroadcaster *pds.ScanBroadcaster - garbageCollector *gc.GarbageCollector - adminUI *admin.AdminUI - holdDB *holddb.HoldDB // shared database connection (nil for :memory:) + httpServer *http.Server + broadcaster *pds.EventBroadcaster + scanBroadcaster *pds.ScanBroadcaster + garbageCollector *gc.GarbageCollector + adminUI *admin.AdminUI + holdDB *holddb.HoldDB // shared database connection (nil for :memory:) + labelerSubscriber *holdlabeler.Subscriber + labelerCache *holdlabeler.Cache } // NewHoldServer initializes PDS, storage, quota, XRPC handlers, and returns @@ -210,8 +243,31 @@ func NewHoldServer(cfg *Config) (*HoldServer, error) { "rescanInterval", rescanInterval) } + // Initialize labeler cache + subscriber if a labeler is configured. + // The cache is created either way so the GC can take a non-nil + // pointer; without a configured DID it just stays empty and exerts + // no effect. + if s.holdDB != nil { + cache, cacheErr := holdlabeler.NewCache(s.holdDB.DB) + if cacheErr != nil { + return nil, fmt.Errorf("failed to initialize labeler cache: %w", cacheErr) + } + s.labelerCache = cache + if cfg.Labeler.DID != "" { + s.labelerSubscriber = holdlabeler.NewSubscriber( + cfg.Labeler.DID, + s.labelerCache, + purgerAdapter{pds: s.PDS}, + ) + slog.Info("Hold labeler subscriber initialized", + "labeler", cfg.Labeler.DID, + "grace_window", cfg.Labeler.GraceWindow) + } + } + // Initialize garbage collector - s.garbageCollector = gc.NewGarbageCollector(s.PDS, s3Service, cfg.GC) + s.garbageCollector = gc.NewGarbageCollector(s.PDS, s3Service, cfg.GC, + gc.WithTakedownCache(s.labelerCache, cfg.Labeler.GraceWindow)) slog.Info("Garbage collector initialized", "enabled", cfg.GC.Enabled) } @@ -333,6 +389,12 @@ func (s *HoldServer) Serve() error { s.garbageCollector.Start(context.Background()) } + // Start labeler subscriber if configured. + if s.labelerSubscriber != nil { + s.labelerSubscriber.Start() + slog.Info("Hold labeler subscriber started", "labeler_did", s.labelerSubscriber.LabelerDID()) + } + // Wait for signal or server error select { case err := <-serverErr: @@ -364,6 +426,12 @@ func (s *HoldServer) shutdown() { slog.Info("Garbage collector stopped") } + // Stop labeler subscriber + if s.labelerSubscriber != nil { + s.labelerSubscriber.Stop() + slog.Info("Labeler subscriber stopped") + } + // Close scan broadcaster database connection if s.scanBroadcaster != nil { if err := s.scanBroadcaster.Close(); err != nil { diff --git a/pkg/labeler/handlers.go b/pkg/labeler/handlers.go index 2b30d52..ee47236 100644 --- a/pkg/labeler/handlers.go +++ b/pkg/labeler/handlers.go @@ -21,7 +21,7 @@ func (s *Server) handleLogin(w http.ResponseWriter, r *http.Request) { } errorMsg := r.URL.Query().Get("error") - w.Header().Set("Content-Type", "text/html") + w.Header().Set("Content-Type", "text/html; charset=utf-8") fmt.Fprintf(w, ` %s Labeler - Login diff --git a/pkg/labeler/takedown.go b/pkg/labeler/takedown.go index 968f264..a91dc97 100644 --- a/pkg/labeler/takedown.go +++ b/pkg/labeler/takedown.go @@ -405,12 +405,25 @@ func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { return } + // Pre-fetch labels per visible takedown so the expand-in-place rows can render + // without a round-trip. N+1 queries are fine here — both lists are paginated to + // 50, this is admin-only, and it keeps the UI JS-light (just a toggle). + labelsByTakedown := make(map[int64][]Label, len(active)+len(reversed)) + for _, t := range append(append([]Takedown{}, active...), reversed...) { + labels, err := GetLabelsByTakedown(s.db, t.ID) + if err != nil { + slog.Warn("Failed to load labels for takedown", "takedown_id", t.ID, "error", err) + continue + } + labelsByTakedown[t.ID] = labels + } + csrf := "" if session := SessionFromContext(r.Context()); session != nil { csrf = session.CSRFToken } - w.Header().Set("Content-Type", "text/html") + w.Header().Set("Content-Type", "text/html; charset=utf-8") fmt.Fprintf(w, ` %s Labeler @@ -427,6 +440,18 @@ nav{display:flex;gap:16px;margin-bottom:24px} form{display:inline} code{background:#f4f4f5;padding:1px 4px;border-radius:3px} .reason{max-width:280px;white-space:pre-wrap} +.toggle{background:none;border:1px solid #d4d4d8;border-radius:4px;padding:2px 8px;font:inherit;cursor:pointer;color:#374151} +.toggle:hover{background:#f4f4f5} +.toggle .caret{display:inline-block;transition:transform .15s ease;margin-right:4px} +.toggle[aria-expanded="true"] .caret{transform:rotate(90deg)} +.detail-row td{background:#fafafa;padding:0} +.detail-row .inner{padding:12px 16px} +.label-list{width:100%%;border-collapse:collapse;font-size:0.88em} +.label-list th,.label-list td{border-bottom:1px solid #eee;padding:4px 8px;background:#fafafa} +.label-list th{background:#f1f1f3} +.tag{display:inline-block;padding:1px 6px;border-radius:3px;font-size:0.85em} +.tag-active{background:#fee2e2;color:#991b1b} +.tag-neg{background:#dcfce7;color:#166534} @@ -442,17 +467,32 @@ code{background:#f4f4f5;padding:1px 4px;border-radius:3px} activeTotal, ) - renderTakedownRows(w, active, csrf, true) + renderTakedownRows(w, active, labelsByTakedown, csrf, true) fmt.Fprintf(w, `

Reversed (%d)

`, reversedTotal) - renderTakedownRows(w, reversed, csrf, false) + renderTakedownRows(w, reversed, labelsByTakedown, csrf, false) - fmt.Fprint(w, ``) + // Tiny inline toggle: flips [hidden] on the sibling detail row and the + // aria-expanded attribute on the button (which the .caret CSS rotates). + fmt.Fprint(w, ` +`) } // renderTakedownRows writes either an active table (with a Reverse button) or a -// reversed-history table (with a reversed-at column instead). -func renderTakedownRows(w http.ResponseWriter, ts []Takedown, csrf string, withReverse bool) { +// reversed-history table (with a reversed-at column instead). Each main row is +// followed by a hidden detail row that the inline JS toggles to show the labels +// linked to that takedown. +func renderTakedownRows(w http.ResponseWriter, ts []Takedown, labelsByID map[int64][]Label, csrf string, withReverse bool) { if len(ts) == 0 { if withReverse { fmt.Fprint(w, `

No active takedowns.

`) @@ -461,6 +501,7 @@ func renderTakedownRows(w http.ResponseWriter, ts []Takedown, csrf string, withR } return } + const totalCols = 6 fmt.Fprint(w, ``) if withReverse { fmt.Fprint(w, ``) @@ -503,25 +544,68 @@ func renderTakedownRows(w http.ResponseWriter, ts []Takedown, csrf string, withR lastCol = rev + by } + detailID := fmt.Sprintf("td-%d-detail", t.ID) fmt.Fprintf(w, ` - + -`, + +`, template.HTMLEscapeString(t.Input), subject, reason, - t.LabelCount, + detailID, detailID, t.LabelCount, t.CreatedAt.Format("2006-01-02 15:04"), lastCol, + detailID, totalCols, renderLabelList(labelsByID[t.ID]), ) } fmt.Fprint(w, `
InputSubjectReasonLabelsCreatedAction
%s %s %s%d %s %s
`) } +// renderLabelList returns an HTML fragment listing every label linked to a takedown, +// marking each as active (neg=0 with no later neg=1 row) or negated. Pure string +// build-up so it can be embedded inside a via fmt.Fprintf. +func renderLabelList(labels []Label) string { + if len(labels) == 0 { + return `No labels recorded for this takedown.` + } + + // Compute which positive labels have been overridden by a later negation row + // (same URI). Used to badge the "active" vs "negated" state correctly even + // when the takedown row itself is still marked active. + negatedURIs := make(map[string]bool, len(labels)) + for _, l := range labels { + if l.Neg { + negatedURIs[l.URI] = true + } + } + + var b strings.Builder + b.WriteString(``) + for _, l := range labels { + var tag string + switch { + case l.Neg: + tag = `negation` + case negatedURIs[l.URI]: + tag = `negated` + default: + tag = `active` + } + fmt.Fprintf(&b, ``, + tag, + template.HTMLEscapeString(l.URI), + l.Cts.Format("2006-01-02 15:04"), + ) + } + b.WriteString(`
StateURICreated
%s%s%s
`) + return b.String() +} + func (s *Server) handleTakedownForm(w http.ResponseWriter, r *http.Request) { msg := r.URL.Query().Get("msg") errorMsg := r.URL.Query().Get("error") @@ -530,7 +614,7 @@ func (s *Server) handleTakedownForm(w http.ResponseWriter, r *http.Request) { csrf = session.CSRFToken } - w.Header().Set("Content-Type", "text/html") + w.Header().Set("Content-Type", "text/html; charset=utf-8") fmt.Fprintf(w, ` %s Labeler - New Takedown