From 7a13a3fba6aaf6731d06403249cc4ae6537aa307 Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 1 Oct 2026 22:11:28 -0700 Subject: [PATCH] feat(moderation): optional Cloudflare CDN purge after removal Removal and the consumer's media reconcile can now purge a Cloudflare-fronted image hostname by URL, best-effort, once, after commit. With nothing configured, behaviour is unchanged apart from one Info line at boot. - Config: CLOUDFLARE_CACHE_PURGE_ZONE_ID and CLOUDFLARE_CACHE_PURGE_API_TOKEN. Both empty means purge is off; exactly one set is a boot error; both set requires the image proxy enabled with absolute http(s) base/CDN URLs. The token never appears in errors, logs or formatted config. - imageproxy purger: one URL per preset per purge base for each (owner DID, CID) pair, byte-identical to HydrateImageURL output; POST to /zones/{zone}/purge_cache in batches of at most 100 URLs with a pair's URLs kept together; failures carry a code (HTTP status, api_unsuccessful, transport) and never the token or response body. - Moderation: removeContent and the media reconcile purge every owner-scoped block they committed after commit; a purge failure is logged and never fails or rolls back the removal. Every blocked blob gets an owner-scoped block for the subject's owner, illegal-content included. - Wiring, prod env example and compose entries; testkit fake Cloudflare purge endpoint; T0/T1 contract tests; CSAM PRD notes updated. Co-Authored-By: Claude Opus 5.5 (1M context) --- .env.prod.example | 11 + cmd/server/cdn_purge_wiring.go | 69 ++++ cmd/server/cdn_purge_wiring_test.go | 159 ++++++++ cmd/server/wiring.go | 27 +- docker-compose.prod.yml | 4 + docs/PRD_CSAM_SCANNING.md | 18 +- .../handlers/imageproxy/cache_headers_test.go | 36 ++ internal/api/handlers/imageproxy/handler.go | 17 +- .../moderation_media_integration_test.go | 83 ++++- .../moderation_post_media_integration_test.go | 73 ++++ .../post_moderation_consumer_test.go | 66 +++- internal/config/cdn_purge_config_test.go | 94 +++++ internal/config/config.go | 58 ++- internal/config/testing.go | 1 + internal/core/imageproxy/cdn_purge.go | 161 ++++++++ internal/core/imageproxy/cdn_purge_test.go | 313 ++++++++++++++++ internal/core/moderation/cdn_purge.go | 42 +++ internal/core/moderation/cdn_purge_test.go | 344 ++++++++++++++++++ internal/core/moderation/media.go | 10 +- internal/core/moderation/remove.go | 1 + internal/core/moderation/remove_rules_test.go | 8 + internal/core/moderation/types.go | 1 + tests/testkit/cloudflare_purge.go | 77 ++++ 23 files changed, 1641 insertions(+), 32 deletions(-) create mode 100644 cmd/server/cdn_purge_wiring.go create mode 100644 cmd/server/cdn_purge_wiring_test.go create mode 100644 internal/config/cdn_purge_config_test.go create mode 100644 internal/core/imageproxy/cdn_purge.go create mode 100644 internal/core/imageproxy/cdn_purge_test.go create mode 100644 internal/core/moderation/cdn_purge.go create mode 100644 internal/core/moderation/cdn_purge_test.go create mode 100644 tests/testkit/cloudflare_purge.go diff --git a/.env.prod.example b/.env.prod.example index 546af69..5a69add 100644 --- a/.env.prod.example +++ b/.env.prod.example @@ -309,6 +309,17 @@ IMAGE_PROXY_CACHE_MAX_GB=10 # Not needed when img.coves.social is itself CDN-fronted (the current setup). # IMAGE_PROXY_CDN_URL= +# Optional Cloudflare edge purge on moderation removal. Leave both empty to +# turn it off (self-hosters without Cloudflare need neither); setting only one +# refuses to start. When set, each removal purges the edge URLs of every image +# preset for the removed blobs under IMAGE_PROXY_BASE_URL and, when it differs, +# IMAGE_PROXY_CDN_URL. Without it, edge copies expire within the one-day +# Cache-Control max-age. +# The token needs only Zone -> Cache Purge -> Purge on the coves.social zone. +# It is separate from Caddy's CLOUDFLARE_API_TOKEN above (DNS-01); do not reuse it. +# CLOUDFLARE_CACHE_PURGE_ZONE_ID= +# CLOUDFLARE_CACHE_PURGE_API_TOKEN= + # Deliberate opt-out for self-hosted deployments that are NOT fronting media # with a scanning CDN and want blobs served straight from the PDS. Only has an # effect with IMAGE_PROXY_ENABLED=false; it exists so that path is a choice diff --git a/cmd/server/cdn_purge_wiring.go b/cmd/server/cdn_purge_wiring.go new file mode 100644 index 0000000..22a9ae9 --- /dev/null +++ b/cmd/server/cdn_purge_wiring.go @@ -0,0 +1,69 @@ +package main + +import ( + "fmt" + "log/slog" + "time" + + "Coves/internal/config" + "Coves/internal/core/imageproxy" + "Coves/internal/core/moderation" +) + +// cloudflareAPIBase is the Cloudflare API root production purges against. +const cloudflareAPIBase = "https://api.cloudflare.com/client/v4" // coves:allow-public-host: fixed Cloudflare cache purge endpoint + +// moderationDependencies are the storage, media and resolution seams the +// moderation service and media reconciler are built from. +type moderationDependencies struct { + subjectReader moderation.SubjectReader + store moderation.Store + mediaBinder moderation.TransactionBinder + mediaPurger moderation.MediaPurger + communityResolver moderation.CommunityResolver + handleResolver moderation.HandleResolver +} + +// buildModeration builds the moderation service and the media reconciler, and +// gives both the CDN purger when Cloudflare purge is configured: the service +// purges after a removal commits, the reconciler after a consumer reconcile +// commits. apiBase is cloudflareAPIBase in production. +func buildModeration(cfg *config.Config, apiBase string, dependencies moderationDependencies) (moderation.Service, *moderation.MediaReconciler, error) { + cdnPurger, err := buildCDNPurger(cfg.Media.CDNPurge, apiBase) + if err != nil { + return nil, nil, fmt.Errorf("creating CDN purger: %w", err) + } + moderationConfig := moderation.Config{ + InstanceDID: cfg.Instance.DID, + IdempotencyRetention: cfg.Moderation.IdempotencyRetention, + MaxLiveIdempotencyKeys: cfg.Moderation.MaxLiveIdempotencyKeys, + Purger: dependencies.mediaPurger, + CursorSecret: cfg.CursorSecret, + CommunityResolver: dependencies.communityResolver, + HandleResolver: dependencies.handleResolver, + } + var reconcilerOptions []moderation.MediaReconcilerOption + if cdnPurger != nil { + moderationConfig.CDNPurger = cdnPurger + reconcilerOptions = append(reconcilerOptions, moderation.WithCDNPurger(cdnPurger)) + } + service := moderation.NewService(dependencies.subjectReader, dependencies.store, moderationConfig) + reconciler := moderation.NewMediaReconciler( + dependencies.mediaBinder, cfg.Instance.DID, dependencies.mediaPurger, reconcilerOptions...) + return service, reconciler, nil +} + +// buildCDNPurger returns nil when no Cloudflare edge invalidation is configured. +func buildCDNPurger(purge config.CDNPurgeConfig, apiBase string) (moderation.CDNPurger, error) { + if !purge.Enabled() { + slog.Info("CDN purge is off; shared caches can keep an image for at most a day") + return nil, nil + } + return imageproxy.NewCloudflarePurger(imageproxy.CloudflarePurgerConfig{ + APIBase: apiBase, + ZoneID: purge.ZoneID, + APIToken: purge.APIToken, + BaseURLs: purge.BaseURLs, + Timeout: 10 * time.Second, + }) +} diff --git a/cmd/server/cdn_purge_wiring_test.go b/cmd/server/cdn_purge_wiring_test.go new file mode 100644 index 0000000..478611f --- /dev/null +++ b/cmd/server/cdn_purge_wiring_test.go @@ -0,0 +1,159 @@ +package main + +import ( + "context" + "log/slog" + "sync" + "testing" + "time" + + "Coves/internal/config" + "Coves/internal/core/blobs" + "Coves/internal/core/imageproxy" + "Coves/internal/core/moderation" + "Coves/tests/testkit" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type cdnBootLogHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (handler *cdnBootLogHandler) Enabled(context.Context, slog.Level) bool { return true } +func (handler *cdnBootLogHandler) Handle(_ context.Context, record slog.Record) error { + handler.mu.Lock() + defer handler.mu.Unlock() + handler.records = append(handler.records, record.Clone()) + return nil +} +func (handler *cdnBootLogHandler) WithAttrs([]slog.Attr) slog.Handler { return handler } +func (handler *cdnBootLogHandler) WithGroup(string) slog.Handler { return handler } +func (handler *cdnBootLogHandler) snapshot() []slog.Record { + handler.mu.Lock() + defer handler.mu.Unlock() + return append([]slog.Record(nil), handler.records...) +} + +func TestBuildCDNPurgerDisabledAndConfigured(t *testing.T) { + previous := slog.Default() + handler := &cdnBootLogHandler{} + slog.SetDefault(slog.New(handler)) + t.Cleanup(func() { slog.SetDefault(previous) }) + + purger, err := buildCDNPurger(config.CDNPurgeConfig{}, cloudflareAPIBase) + require.NoError(t, err) + require.True(t, purger == nil, "disabled purger must be an untyped nil interface") + records := handler.snapshot() + require.Len(t, records, 1) + assert.Equal(t, slog.LevelInfo, records[0].Level) + + purger, err = buildCDNPurger(config.CDNPurgeConfig{ZoneID: "zone-abc", APIToken: "tok", BaseURLs: []string{"https://img.example.test"}}, cloudflareAPIBase) + require.NoError(t, err) + assert.NotNil(t, purger) +} + +// cdnWiringStore runs a removal against one indexed comment with no prior +// moderation state. Transaction methods a removal never calls are left to the +// embedded nil interface, so reaching one panics instead of passing silently. +type cdnWiringStore struct { + moderation.Store + transaction *cdnWiringTransaction +} + +func (store *cdnWiringStore) InTransaction(ctx context.Context, fn func(context.Context, moderation.Transaction) error) error { + return fn(ctx, store.transaction) +} + +type cdnWiringTransaction struct { + moderation.Transaction + comment moderation.IndexedComment +} + +func (*cdnWiringTransaction) LockActor(context.Context, string) error { return nil } +func (*cdnWiringTransaction) LiveIdempotencyRecord(context.Context, string, string, string, time.Time) (*moderation.IdempotencyRecord, error) { + return nil, nil +} +func (*cdnWiringTransaction) CountLiveIdempotencyKeys(context.Context, string, time.Time) (int, error) { + return 0, nil +} +func (*cdnWiringTransaction) SaveIdempotencyRecord(context.Context, moderation.IdempotencyRecord) error { + return nil +} +func (*cdnWiringTransaction) LockSubject(context.Context, string) (int64, error) { return 0, nil } +func (transaction *cdnWiringTransaction) ReadIndexedComment(context.Context, string) (*moderation.IndexedComment, error) { + comment := transaction.comment + return &comment, nil +} +func (*cdnWiringTransaction) ActiveRemoval(context.Context, string, string) (*moderation.Action, error) { + return nil, nil +} +func (*cdnWiringTransaction) ActiveLabels(context.Context, string, string) ([]moderation.Action, error) { + return nil, nil +} +func (*cdnWiringTransaction) InsertAction(_ context.Context, action moderation.Action) (*moderation.Action, error) { + action.ID = "cdn-wiring-action" + return &action, nil +} +func (*cdnWiringTransaction) SetRemovalDecision(context.Context, string, string, string, bool) error { + return nil +} +func (*cdnWiringTransaction) SetSubjectVersion(context.Context, string, int64) error { return nil } +func (*cdnWiringTransaction) InsertMediaBlocks(context.Context, []moderation.MediaBlock) error { + return nil +} + +// TestBuildModerationPurgesCloudflareOnRemovalAndReconcile pins the production +// wiring: with CDN purge configured, both the removal path (moderation service) +// and the consumer reconcile path (media reconciler) must reach Cloudflare. +func TestBuildModerationPurgesCloudflareOnRemovalAndReconcile(t *testing.T) { + const ( + ownerDID = "did:plc:cdnwiringauthor" + subjectURI = "at://did:plc:cdnwiringauthor/social.coves.community.comment/3kcdnwiring" + removedImage = "bafyreicdnwiringremoved" + reconciledBlob = "bafyreicdnwiringreconciled" + imageBaseURL = "https://img.cdn-wiring.test" + ) + endpoint := testkit.NewCloudflarePurgeEndpoint(t) + cfg := &config.Config{ + Instance: config.InstanceConfig{DID: "did:web:cdn-wiring.test"}, + Moderation: config.ModerationConfig{IdempotencyRetention: time.Hour, MaxLiveIdempotencyKeys: 10}, + Media: config.MediaConfig{ + ImageProxy: imageproxy.Config{BaseURL: imageBaseURL}, + CDNPurge: config.CDNPurgeConfig{ZoneID: "zone-cdn-wiring", APIToken: "token-cdn-wiring", BaseURLs: []string{imageBaseURL}}, + }, + } + store := &cdnWiringStore{transaction: &cdnWiringTransaction{comment: moderation.IndexedComment{ + URI: subjectURI, CID: "bafyreicdnwiringsubject", OwnerDID: ownerDID, ImageCIDs: []string{removedImage}, + }}} + + service, reconciler, err := buildModeration(cfg, endpoint.URL(), moderationDependencies{store: store}) + require.NoError(t, err) + + presets := imageproxy.ListPresets() + require.NotEmpty(t, presets) + assertPurged := func(t *testing.T, requests []testkit.CloudflarePurgeRequest, cid string) { + t.Helper() + require.Len(t, requests, 1) + assert.Equal(t, "/zones/zone-cdn-wiring/purge_cache", requests[0].Path) + assert.Equal(t, "Bearer token-cdn-wiring", requests[0].Authorization) + require.NoError(t, requests[0].DecodeError) + assert.Contains(t, requests[0].Files, blobs.HydrateImageProxyURL(imageBaseURL, presets[0].Name, ownerDID, cid)) + } + + result, err := service.RemoveContent(t.Context(), "did:plc:cdnwiringadmin", moderation.RemoveContentRequest{ + Subject: moderation.StrongRef{URI: subjectURI, CID: "bafyreicdnwiringsubject"}, + ExpectedVersion: moderation.InitialVersion, IdempotencyKey: "cdn-wiring-removal", + Reason: "social.coves.moderation.defs#reasonSpam", + }) + require.NoError(t, err) + require.Equal(t, moderation.OutcomeApplied, result.Outcome) + assertPurged(t, endpoint.Requests(), removedImage) + + reconciler.Purge([]moderation.MediaBlock{{OwnerDID: ownerDID, BlobCID: reconciledBlob, ActionID: "cdn-wiring-action"}}) + requests := endpoint.Requests() + require.Len(t, requests, 2, "the media reconciler must purge Cloudflare after a reconcile commit") + assertPurged(t, requests[1:], reconciledBlob) +} diff --git a/cmd/server/wiring.go b/cmd/server/wiring.go index 9e50b63..d5a0b4f 100644 --- a/cmd/server/wiring.go +++ b/cmd/server/wiring.go @@ -211,21 +211,18 @@ func buildApplication( if err != nil { return nil, err } - app.moderationService = moderation.NewService( - moderation.NewRepositorySubjectReader(app.postRepo, app.commentRepo), - postgresRepo.NewModerationRepository(app.db), - moderation.Config{ - InstanceDID: app.cfg.Instance.DID, - IdempotencyRetention: app.cfg.Moderation.IdempotencyRetention, - MaxLiveIdempotencyKeys: app.cfg.Moderation.MaxLiveIdempotencyKeys, - Purger: purger, - CursorSecret: app.cfg.CursorSecret, - CommunityResolver: app.communityService, - HandleResolver: app.identityResolver, - }, - ) - app.mediaReconciler = moderation.NewMediaReconciler( - postgresRepo.NewModerationRepository(app.db), app.cfg.Instance.DID, purger) + moderationStore := postgresRepo.NewModerationRepository(app.db) + app.moderationService, app.mediaReconciler, err = buildModeration(app.cfg, cloudflareAPIBase, moderationDependencies{ + subjectReader: moderation.NewRepositorySubjectReader(app.postRepo, app.commentRepo), + store: moderationStore, + mediaBinder: moderationStore, + mediaPurger: purger, + communityResolver: app.communityService, + handleResolver: app.identityResolver, + }) + if err != nil { + return nil, err + } app.buildJetstreamInfrastructure() if err = app.buildBridgedVotePoller(); err != nil { return nil, err diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 2382a7b..ee412d1 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -155,6 +155,10 @@ services: # a variable only present in .env is invisible to the container, so setting # it there without this line silently has no effect. IMAGE_PROXY_CDN_URL: ${IMAGE_PROXY_CDN_URL:-} + # Optional Cloudflare edge purge on moderation removal. Both must be set + # together; this token is separate from Caddy's DNS-01 token. + CLOUDFLARE_CACHE_PURGE_ZONE_ID: ${CLOUDFLARE_CACHE_PURGE_ZONE_ID:-} + CLOUDFLARE_CACHE_PURGE_API_TOKEN: ${CLOUDFLARE_CACHE_PURGE_API_TOKEN:-} IMAGE_PROXY_CACHE_PATH: /var/cache/coves/images IMAGE_PROXY_CACHE_MAX_GB: ${IMAGE_PROXY_CACHE_MAX_GB:-10} IMAGE_PROXY_FETCH_TIMEOUT_SECONDS: ${IMAGE_PROXY_FETCH_TIMEOUT_SECONDS:-30} diff --git a/docs/PRD_CSAM_SCANNING.md b/docs/PRD_CSAM_SCANNING.md index a445272..3c04c98 100644 --- a/docs/PRD_CSAM_SCANNING.md +++ b/docs/PRD_CSAM_SCANNING.md @@ -1,7 +1,7 @@ # PRD: CSAM Scanning via Cloudflare + Media Choke Point **Status:** Workstreams 1 and 2 implemented (AppView + Caddy). Workstream 3 (Cloudflare zone config) is manual dashboard/DNS work: as of 2026-09-28 Cloudflare proxies `img.coves.social`, and no Cloudflare API token with the Cache Purge permission exists yet. The other WS3 steps (cache rule, CSAM Scanning Tool, SSL mode) are not confirmed here. Workstream 4 (takedown runbook) not started. -**Last updated:** 2026-09-29 +**Last updated:** 2026-10-01 ## Problem @@ -68,7 +68,7 @@ Remaining at deploy time: **the bind-mount trap** — Caddyfile changes require On the `coves.social` zone (we already own it — DNS-01 tokens exist): 1. **DNS**: `img.coves.social` A/AAAA → OVH origin IP, **Proxied** (orange cloud). All other records stay DNS-only (grey) — especially anything under `tdpl.io` and `coves.me`. -2. **Cache**: add a Cache Rule for `img.coves.social/*`: *Eligible for cache*, with Edge TTL set to respect origin `Cache-Control` headers, and set the zone's Browser Cache TTL to *Respect Existing Headers*. Blobs are content-addressed (CID in URL), but caching is no longer immutable: a moderation removal must stop serving an image, so the origin advertises `public, max-age=86400` and an overriding edge or browser TTL would outlive that bound. Optionally enable Tiered Cache. +2. **Cache**: add a Cache Rule for `img.coves.social/*`: *Eligible for cache*, with Edge TTL set to respect origin `Cache-Control` headers, and set the zone's Browser Cache TTL to *Respect Existing Headers*. Keep Cloudflare's default cache key; do not build a custom key from headers, cookies or query parameters, because URL purge only matches the default key. Blobs are content-addressed (CID in URL), but caching is no longer immutable: a moderation removal must stop serving an image, so the origin advertises `public, max-age=86400` and an overriding edge or browser TTL would outlive that bound. The image proxy serves any request with a query string as `no-store`, so query-string variants are never shared-cacheable and the bare-URL purge covers every cacheable edge entry. Optionally enable Tiered Cache. 3. **Enable CSAM Scanning Tool**: Dashboard → Caching → Configuration → CSAM Scanning Tool → Configure. Provide a monitored role address (e.g. `abuse@coves.social`, forwarded to admins) and verify it. Agree to the service-specific terms. 4. **SSL mode**: Full (strict) for the zone (origin has valid certs via Caddy). 5. Do **not** enable Cloudflare features that interfere with API semantics on other hostnames — only `img` is proxied, so blast radius is zero. @@ -89,7 +89,7 @@ Build an admin takedown flow (CLI or admin endpoint), input = DID + CID: - Delete the record + blob from the owning PDS (native: PDS admin API; bridged: tidepool bridge admin path). - Remove/tombstone the post in the AppView index. - Purge the imageproxy disk cache for **all presets** of that DID+CID (add a purge-by-blob admin method to `imageproxy.DiskCache` — cache keys are preset-scoped). - - Purge the Cloudflare edge cache by URL for each preset variant (single-file purge is available on free plans). Cloudflare's own block covers the exact matched URL; we purge the sibling preset URLs. + - Purge the Cloudflare edge cache by URL for each preset variant (single-file purge is available on free plans). Cloudflare's own block covers the exact matched URL; we purge the sibling preset URLs. Moderation removal already purges the imageproxy disk cache and, when configured, the edge through `imageproxy.CloudflarePurger`. Set both `CLOUDFLARE_CACHE_PURGE_ZONE_ID` and `CLOUDFLARE_CACHE_PURGE_API_TOKEN`; the token needs Zone → Cache Purge → Purge on the `coves.social` zone only, separate from Caddy's `CLOUDFLARE_API_TOKEN`. The DID+CID takedown tooling can reuse this purger. 4. **Report**: file our own NCMEC CyberTipline report (Cloudflare's third-party report does not replace the provider's own obligation). 5. **Act on the source**: ban the native account, or for bridged content: report to the origin Lemmy instance's admins and, on repeat, drop the instance at the bridge (instance blocklist) — this is where "rely on Lemmy moderation" plugs in. 6. **Log** the entire action (who, what, when) to an audit table. Never log or store the image content outside the preservation store. @@ -104,7 +104,10 @@ Phase 1 can be a documented manual runbook using existing tools (psql, PDS admin | Direct PDS `getBlob` remains publicly fetchable | Required by atproto sync (relays, other AppViews) | API no longer emits these URLs; optionally rate-limit `getBlob` at Caddy for non-relay UAs | | Only known-hash CSAM is detected | Fuzzy hash lists can't catch novel content | Community reporting (`internal/core/adminreports/`) + moderator review remain the backstop | | Video blobs unscanned | Image proxy is stills-only | Track as separate workstream | -| Browser caches, and any shared cache nobody purges, keep a removed image for up to one day | A moderation removal purges only the proxy's own disk cache; a copy already served with `Cache-Control: public, max-age=86400` stays valid wherever it was stored | The one-day `max-age` bounds the exposure without any configuration, including for self-hosters with no CDN; the optional CDN purge on removal, once built, will shorten it for an edge that is configured for it | +| Browser caches, and any shared cache nobody purges, keep a removed image for up to one day | A moderation removal purges the proxy's disk cache and optionally Cloudflare's edge; a copy already served with `Cache-Control: public, max-age=86400` stays valid in browser caches and any shared cache not purged | The one-day `max-age` bounds the exposure without any configuration, including for self-hosters with no CDN | +| Other owners' edge copies of an `illegal-content` CID are not purged | The edge purge names only the owner of each owner-scoped block, not the other owners covered by an ownerless block | The one-day cache lifetime bounds the gap; a planned follow-up names the other owners | +| A failed edge purge is not retried | The removal logs the failure but does not queue a retry | The one-day cache lifetime bounds the gap; a planned follow-up adds durable purge targets and retries | +| An in-flight response can reach the edge after its purge | A response authorized before the removal commits can still be written after the edge purge | The one-day cache lifetime bounds the gap; a planned follow-up adds a final purge after `HTTP_WRITE_TIMEOUT` | | Responses cached under the pre-deploy `public, max-age=31536000, immutable` header | Copies stored before the one-day header shipped keep their original lifetime | Browsers keep them for up to a year and nothing server-side can reach them; the edge keeps them until a one-time Purge Everything of the `coves.social` zone after deploy (Rollout order step 4; WS3 proxies only `img.coves.social`, so this drops only cached images) | | Bridge PDS stores blobs regardless of scanning | Blobs land before any serve-time scan | Phase 2 ingest scanning; instance allow/blocklist at the bridge is the coarse control | | `record.embed` still carries blob references | Post and comment responses include the verbatim atproto record, whose embed is unprojected by design (the lexicon calls it verbatim). A client *could* build a `getBlob` URL from it | Neither client reads `record.embed` today. Coves image URLs use the proxy, with the explicit foreign Bluesky CDN exception below; this does not mean no blob reference reaches a client. The same record bytes are public on the PDS regardless. Revisit if a client starts reading it | @@ -118,9 +121,10 @@ WS1 and WS2 are both in the tree, so they ship together. The ordering constraint 2. **Deploy Caddy** — `docker compose up -d --force-recreate caddy` (bind-mount trap). Verify `curl -sD- -o /dev/null https://img.coves.social/img/avatar/plain//` returns 200 with `Cache-Control: public, max-age=86400`, and that `https://img.coves.social/` 404s. (Use `-sD- -o /dev/null`, not `-I`: the image route is registered GET-only and chi does not map HEAD to it, so `-I` returns 405 and shows none of the cache headers.) 3. **Deploy the AppView** with `IMAGE_PROXY_BASE_URL=https://img.coves.social`. Startup now fails loudly on a misconfigured proxy rather than silently falling back. Verify feeds render, then watch proxy cache hit rate and origin bandwidth. 4. **Purge the edge once** — after the first AppView deploy that sends `Cache-Control: public, max-age=86400`, run Purge Everything on the `coves.social` zone in the Cloudflare dashboard. Edge copies stored under the old `public, max-age=31536000, immutable` header otherwise keep serving for up to a year, including images removed since. Browser copies cannot be purged (WS5). -5. **Client follow-ups** — ship the two `coves-frontend` type/`extractEmbedUrl` changes noted in WS1. -6. WS4 runbook written and dry-run before announcing Lemmy federation more broadly. -7. Phase 2 (ingest-time hash matching) scheduled after federation traffic is real. +5. **Verify edge invalidation (outside hermetic CI)** — warm an image through the Cloudflare edge, remove the content, then confirm the edge serves 404 for that image URL. +6. **Client follow-ups** — ship the two `coves-frontend` type/`extractEmbedUrl` changes noted in WS1. +7. WS4 runbook written and dry-run before announcing Lemmy federation more broadly. +8. Phase 2 (ingest-time hash matching) scheduled after federation traffic is real. ## Open questions diff --git a/internal/api/handlers/imageproxy/cache_headers_test.go b/internal/api/handlers/imageproxy/cache_headers_test.go index 09a1131..485ebe8 100644 --- a/internal/api/handlers/imageproxy/cache_headers_test.go +++ b/internal/api/handlers/imageproxy/cache_headers_test.go @@ -53,6 +53,42 @@ func TestHandler_RoutedImageCacheHeaders(t *testing.T) { assert.Equal(t, served.Header().Get("ETag"), conditional.Header().Get("ETag")) }) + // A purge names only the bare URL, so a query-string variant must never be + // shared-cacheable: it is served, unchanged, with no-store. + t.Run("query-string variant is served as no-store on 200 and 304", func(t *testing.T) { + service := &mockService{getImageFunc: func(context.Context, string, string, string, string) ([]byte, error) { + return []byte("image bytes"), nil + }} + router := route(service) + requestPath := func(target, ifNoneMatch string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, target, nil) + if ifNoneMatch != "" { + req.Header.Set("If-None-Match", ifNoneMatch) + } + response := httptest.NewRecorder() + router.ServeHTTP(response, req) + return response + } + + bare := requestPath(path, "") + require.Equal(t, http.StatusOK, bare.Code) + assert.Equal(t, cacheControl, bare.Header().Get("Cache-Control")) + + variant := requestPath(path+"?x=1", "") + require.Equal(t, http.StatusOK, variant.Code) + assert.Equal(t, "image bytes", variant.Body.String()) + assert.Equal(t, etag, variant.Header().Get("ETag")) + assert.Equal(t, "no-store", variant.Header().Get("Cache-Control")) + + conditionalVariant := requestPath(path+"?x=1", etag) + require.Equal(t, http.StatusNotModified, conditionalVariant.Code) + assert.Equal(t, "no-store", conditionalVariant.Header().Get("Cache-Control")) + + conditionalBare := requestPath(path, etag) + require.Equal(t, http.StatusNotModified, conditionalBare.Code) + assert.Equal(t, cacheControl, conditionalBare.Header().Get("Cache-Control")) + }) + for _, test := range []struct { name string serveFirst bool diff --git a/internal/api/handlers/imageproxy/handler.go b/internal/api/handlers/imageproxy/handler.go index 5f21fc7..0e7f836 100644 --- a/internal/api/handlers/imageproxy/handler.go +++ b/internal/api/handlers/imageproxy/handler.go @@ -24,11 +24,22 @@ import ( // the pressure being shed, so the two must never drift apart. var processorBusyRetryAfterSeconds = strconv.Itoa(int(imageproxy.DefaultProcessQueueWait / time.Second)) -// successCacheControl is the cache policy of a served image and of its 304. +// successCacheControl is the cache policy of a served image and of its 304 +// when the request has no query string. // One day bounds how long a browser or an unpurged shared cache keeps an image // after a moderation removal, with no configuration and no CDN required. const successCacheControl = "public, max-age=86400" +// successCacheControlFor returns no-store for a request with a query string. +// A CDN purge names only the bare image URL, while a shared cache keys on the +// query string, so a cached query-string variant would survive the purge. +func successCacheControlFor(r *http.Request) string { + if r.URL.RawQuery != "" { + return "no-store" + } + return successCacheControl +} + // Service defines the interface for the image proxy service. // This interface is implemented by the imageproxy package's service layer. type Service interface { @@ -115,7 +126,7 @@ func (h *Handler) HandleImage(w http.ResponseWriter, r *http.Request) { writeErrorResponse(w, http.StatusNotFound, "blob not found") return } - w.Header().Set("Cache-Control", successCacheControl) + w.Header().Set("Cache-Control", successCacheControlFor(r)) w.Header().Set("ETag", etag) w.WriteHeader(http.StatusNotModified) return @@ -155,7 +166,7 @@ func (h *Handler) HandleImage(w http.ResponseWriter, r *http.Request) { // Set response headers w.Header().Set("Content-Type", "image/jpeg") - w.Header().Set("Cache-Control", successCacheControl) + w.Header().Set("Cache-Control", successCacheControlFor(r)) w.Header().Set("ETag", etag) // Write image data diff --git a/internal/api/routes/moderation_media_integration_test.go b/internal/api/routes/moderation_media_integration_test.go index 634cce8..4453717 100644 --- a/internal/api/routes/moderation_media_integration_test.go +++ b/internal/api/routes/moderation_media_integration_test.go @@ -21,6 +21,7 @@ import ( imagehandler "Coves/internal/api/handlers/imageproxy" "Coves/internal/api/routes" "Coves/internal/atproto/identity" + "Coves/internal/core/blobs" "Coves/internal/core/imageproxy" "Coves/internal/core/moderation" "Coves/internal/db/postgres" @@ -156,7 +157,7 @@ type moderationMediaHarness struct { proxyService *imageproxy.ImageProxyService } -func newModerationMediaHarness(t *testing.T, blockFetch bool) (*moderationMediaHarness, *waitingMediaFetcher) { +func newModerationMediaHarness(t *testing.T, blockFetch bool, cdnPurgers ...moderation.CDNPurger) (*moderationMediaHarness, *waitingMediaFetcher) { t.Helper() db := testkit.DB(t) ownerName := testkit.UniqueIDWithPrefix(t, "mediaowner") @@ -195,9 +196,13 @@ func newModerationMediaHarness(t *testing.T, blockFetch bool) (*moderationMediaH store := postgres.NewModerationRepository(db) proxyService, err := imageproxy.NewService(cache, processor, fetcher, store, imageproxy.DefaultConfig()) require.NoError(t, err) + config := moderation.Config{InstanceDID: fixtures.InstanceDID(), IdempotencyRetention: 24 * time.Hour, MaxLiveIdempotencyKeys: 1000, Purger: proxyService} + if len(cdnPurgers) > 0 { + config.CDNPurger = cdnPurgers[0] + } service := moderation.NewService( moderation.NewRepositorySubjectReader(postgres.NewPostRepository(db), postgres.NewCommentRepository(db)), - store, moderation.Config{InstanceDID: fixtures.InstanceDID(), IdempotencyRetention: 24 * time.Hour, MaxLiveIdempotencyKeys: 1000, Purger: proxyService}, + store, config, ) router := chi.NewRouter() routes.RegisterImageProxyRoutes(router, imagehandler.NewHandler(proxyService, mediaPDSResolver{url: pdsServer.URL})) @@ -207,6 +212,80 @@ func newModerationMediaHarness(t *testing.T, blockFetch bool) (*moderationMediaH moderation: service, postURI: postURI, postCID: post.CID, ownerA: ownerA, ownerB: ownerB}, waiting } +func newMediaCloudflarePurger(t *testing.T, endpoint *testkit.CloudflarePurgeEndpoint, token string) *imageproxy.CloudflarePurger { + t.Helper() + purger, err := imageproxy.NewCloudflarePurger(imageproxy.CloudflarePurgerConfig{ + APIBase: endpoint.URL(), ZoneID: "zone-abc", APIToken: token, + BaseURLs: []string{"https://img.example.test"}, Timeout: 5 * time.Second, + }) + require.NoError(t, err) + require.NotNil(t, purger) + return purger +} + +func TestModerationCommentCDNPurgeAfterPostgresCommit(t *testing.T) { + const token = "route-cdn-token-SENTINEL" + const base = "https://img.example.test" + const spam = "social.coves.moderation.defs#reasonSpam" + for _, scenario := range []string{"removal", "stale version", "restore"} { + t.Run(scenario, func(t *testing.T) { + endpoint := testkit.NewCloudflarePurgeEndpoint(t) + purger := newMediaCloudflarePurger(t, endpoint, token) + h, _ := newModerationMediaHarness(t, false, purger) + imageCID := mediaImageCID("cdn comment " + scenario) + subject := h.comment(t, h.ownerA, imageCID) + observed := make(chan struct { + blocked bool + err error + }, 1) + endpoint.SetHook(func([]string) { + blocked, err := postgres.NewModerationRepository(h.db).IsBlocked(t.Context(), h.ownerA, imageCID) + // Non-blocking: an unexpected second request must not wedge the handler. + select { + case observed <- struct { + blocked bool + err error + }{blocked, err}: + default: + } + }) + if scenario == "stale version" { + result, err := h.moderation.RemoveContent(t.Context(), fixtures.DID(testkit.UniqueIDWithPrefix(t, "mediaadmin")), moderation.RemoveContentRequest{ + Subject: subject, ExpectedVersion: "v7", IdempotencyKey: "remove-" + testkit.UniqueID(t), Reason: spam, + }) + require.ErrorIs(t, err, moderation.ErrStateConflict) + assert.Nil(t, result) + assert.Empty(t, endpoint.Requests()) + return + } + removed := h.remove(t, subject, spam) + requests := endpoint.Requests() + require.Len(t, requests, 1) + request := requests[0] + assert.NoError(t, request.DecodeError) + assert.Equal(t, http.MethodPost, request.Method) + assert.Equal(t, "/zones/zone-abc/purge_cache", request.Path) + assert.Equal(t, "Bearer "+token, request.Authorization) + var expected []string + for _, preset := range []string{"avatar", "avatar_small", "banner", "content_preview", "content_full", "embed_thumbnail"} { + expected = append(expected, blobs.HydrateImageProxyURL(base, preset, h.ownerA, imageCID)) + } + assert.ElementsMatch(t, expected, request.Files) + select { + case committed := <-observed: + require.NoError(t, committed.err) + assert.True(t, committed.blocked, "the endpoint must see the block on a separate connection") + default: + t.Fatal("Cloudflare request hook did not check the committed block") + } + if scenario == "restore" { + h.restore(t, subject, removed) + assert.Len(t, endpoint.Requests(), 1, "restore must not make another purge request") + } + }) + } +} + func (h *moderationMediaHarness) comment(t *testing.T, owner string, imageCIDs ...string) moderation.StrongRef { t.Helper() images := make([]any, 0, len(imageCIDs)) diff --git a/internal/api/routes/moderation_post_media_integration_test.go b/internal/api/routes/moderation_post_media_integration_test.go index 77d0749..0213426 100644 --- a/internal/api/routes/moderation_post_media_integration_test.go +++ b/internal/api/routes/moderation_post_media_integration_test.go @@ -3,9 +3,15 @@ package routes_test import ( + "context" "encoding/json" + "fmt" + "log/slog" "net/http" "os" + "slices" + "strings" + "sync" "testing" "time" @@ -79,6 +85,73 @@ func (h *moderationMediaHarness) indexedImagePost(t *testing.T, collection, imag return subject, blobOwnerDID } +type postMediaLogCapture struct { + mu sync.Mutex + records []slog.Record +} + +func (*postMediaLogCapture) Enabled(context.Context, slog.Level) bool { return true } +func (capture *postMediaLogCapture) Handle(_ context.Context, record slog.Record) error { + capture.mu.Lock() + defer capture.mu.Unlock() + capture.records = append(capture.records, record.Clone()) + return nil +} +func (capture *postMediaLogCapture) WithAttrs([]slog.Attr) slog.Handler { return capture } +func (capture *postMediaLogCapture) WithGroup(string) slog.Handler { return capture } +func (capture *postMediaLogCapture) snapshot() []slog.Record { + capture.mu.Lock() + defer capture.mu.Unlock() + return append([]slog.Record(nil), capture.records...) +} + +// Not parallel: this test replaces slog.Default while the removal runs. +func TestModerationPostCDNFailureDoesNotUndoLocalRemoval(t *testing.T) { + const token = "post-cdn-token-SENTINEL" + const bodySentinel = "cf-response-body-SENTINEL" + endpoint := testkit.NewCloudflarePurgeEndpoint(t) + endpoint.SetResponse(http.StatusInternalServerError, bodySentinel) + cdn := newMediaCloudflarePurger(t, endpoint, token) + h, _ := newModerationMediaHarness(t, false, cdn) + imageCID := mediaImageCID("post cdn failure") + subject, authorDID := h.indexedImagePost(t, moderation.PostV2Collection, imageCID) + require.Equal(t, h.ownerA, authorDID) + require.Equal(t, http.StatusOK, h.request(t, postMediaPreset, authorDID, imageCID)) + _, err := os.Stat(h.cachePath(postMediaPreset, authorDID, imageCID)) + require.NoError(t, err, "the post image must be cached on disk before removal") + + logs := &postMediaLogCapture{} + previousLogger := slog.Default() + slog.SetDefault(slog.New(logs)) + t.Cleanup(func() { slog.SetDefault(previousLogger) }) + removed := h.remove(t, subject, postMediaSpamReason) + require.Equal(t, moderation.OutcomeApplied, removed.Outcome) + assert.Len(t, endpoint.Requests(), 1, "the failing purge must have been attempted") + h.assertNoCachedBlob(t, imageCID, authorDID) + require.Equal(t, http.StatusNotFound, h.request(t, postMediaPreset, authorDID, imageCID)) + + var foundError bool + for _, record := range logs.snapshot() { + var values []string + var text strings.Builder + text.WriteString(record.Message) + record.Attrs(func(attr slog.Attr) bool { + values = append(values, attr.Value.String()) + fmt.Fprint(&text, " ", attr) + return true + }) + assert.NotContains(t, text.String(), token, "no log record may expose the Cloudflare token") + assert.NotContains(t, text.String(), bodySentinel, "no log record may expose the Cloudflare response") + if record.Level == slog.LevelError && + slices.Contains(values, authorDID) && + slices.Contains(values, imageCID) && + slices.Contains(values, "http_500") { + foundError = true + } + } + assert.True(t, foundError, "an Error record must carry the author DID, CID and http_500") +} + func TestModerationPostV2MediaRemovalColdAndWarm(t *testing.T) { for _, warm := range []bool{false, true} { name := "cold" diff --git a/internal/atproto/jetstream/post_moderation_consumer_test.go b/internal/atproto/jetstream/post_moderation_consumer_test.go index 9e642da..652d818 100644 --- a/internal/atproto/jetstream/post_moderation_consumer_test.go +++ b/internal/atproto/jetstream/post_moderation_consumer_test.go @@ -9,7 +9,9 @@ import ( "testing" "time" + "Coves/internal/core/blobs" "Coves/internal/core/communityFeeds" + "Coves/internal/core/imageproxy" "Coves/internal/core/moderation" "Coves/internal/core/posts" "Coves/internal/crypto/credentialcipher/credentialciphertest" @@ -84,17 +86,21 @@ type postModerationConsumerFixture struct { create *JetstreamEvent } -func newPostModerationConsumerFixture(t *testing.T) postModerationConsumerFixture { +func newPostModerationConsumerFixture(t *testing.T, cdnPurgers ...moderation.CDNPurger) postModerationConsumerFixture { t.Helper() db := testkit.DB(t) f := newPV2Fixture(t, db) postRepository := postgres.NewPostRepository(db) moderationRepository := postgres.NewModerationRepository(db) purger := &postModerationPurger{db: db} + var options []moderation.MediaReconcilerOption + if len(cdnPurgers) > 0 { + options = append(options, moderation.WithCDNPurger(cdnPurgers[0])) + } f.consumer = NewPostEventConsumer( postRepository, postgres.NewCommunityRepository(db, credentialciphertest.Fixed()), f.users, db, WithAdmissions(f.admissions), WithDeletedAccounts(postgres.NewDeletedAccountRepository(db)), - WithPostMediaReconciler(moderation.NewMediaReconciler(moderationRepository, fixtures.InstanceDID(), purger)), + WithPostMediaReconciler(moderation.NewMediaReconciler(moderationRepository, fixtures.InstanceDID(), purger, options...)), ) moderator := moderation.NewService( moderation.NewRepositorySubjectReader(postRepository, postgres.NewCommentRepository(db)), @@ -130,6 +136,62 @@ func newPostModerationConsumerFixture(t *testing.T) postModerationConsumerFixtur } } +func TestModerationPostConsumerCDNPurgeAfterEditCommit(t *testing.T) { + const token = "consumer-cdn-token-SENTINEL" + const base = "https://img.example.test" + endpoint := testkit.NewCloudflarePurgeEndpoint(t) + cdn, err := imageproxy.NewCloudflarePurger(imageproxy.CloudflarePurgerConfig{ + APIBase: endpoint.URL(), ZoneID: "zone-abc", APIToken: token, + BaseURLs: []string{base}, Timeout: 5 * time.Second, + }) + require.NoError(t, err) + require.NotNil(t, cdn) + f := newPostModerationConsumerFixture(t, cdn) + removed := f.remove(t, "social.coves.moderation.defs#reasonSpam") + assert.Empty(t, endpoint.Requests(), "the moderator has no CDN purger") + observed := make(chan struct { + blocked bool + err error + }, 1) + endpoint.SetHook(func([]string) { + blocked, err := postgres.NewModerationRepository(f.db).IsBlocked(t.Context(), pv2Author, postModerationCIDTwo) + // Non-blocking: an unexpected second request must not wedge the handler. + select { + case observed <- struct { + blocked bool + err error + }{blocked, err}: + default: + } + }) + update := pv2Event(pv2Author, "update", f.rkey, f.revs[1], postModerationCIDOne, + f.createdAt+1_000_000, postModerationRecord(postModerationCIDOne, postModerationCIDTwo)) + require.NoError(t, f.consumer.HandleEvent(t.Context(), update)) + assert.Equal(t, 1, countRows(t, f.db, ` + SELECT count(*) FROM moderation_media_blocks + WHERE action_id = $1 AND owner_did = $2 AND blob_cid = $3 AND active + `, removed.Action.ID, pv2Author, postModerationCIDTwo)) + requests := endpoint.Requests() + require.Len(t, requests, 1) + request := requests[0] + assert.NoError(t, request.DecodeError) + assert.Equal(t, "POST", request.Method) + assert.Equal(t, "/zones/zone-abc/purge_cache", request.Path) + assert.Equal(t, "Bearer "+token, request.Authorization) + var expected []string + for _, preset := range []string{"avatar", "avatar_small", "banner", "content_preview", "content_full", "embed_thumbnail"} { + expected = append(expected, blobs.HydrateImageProxyURL(base, preset, pv2Author, postModerationCIDTwo)) + } + assert.ElementsMatch(t, expected, request.Files) + select { + case committed := <-observed: + require.NoError(t, committed.err) + assert.True(t, committed.blocked, "the endpoint must see the edited image block on a separate connection") + default: + t.Fatal("Cloudflare request hook did not check the committed edited image block") + } +} + func (f postModerationConsumerFixture) remove(t *testing.T, reason string) *moderation.MutationResult { t.Helper() removed, err := f.moderator.RemoveContent(t.Context(), fixtures.DID(testkit.UniqueIDWithPrefix(t, "postmediaadmin")), moderation.RemoveContentRequest{ diff --git a/internal/config/cdn_purge_config_test.go b/internal/config/cdn_purge_config_test.go new file mode 100644 index 0000000..f63687b --- /dev/null +++ b/internal/config/cdn_purge_config_test.go @@ -0,0 +1,94 @@ +package config + +import ( + "encoding/json" + "fmt" + "os" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const cdnPurgeTestToken = "cf-purge-token-SENTINEL-7f3a" + +func TestLoadCDNPurge(t *testing.T) { + for _, mode := range []string{"dev", "production"} { + for _, test := range []struct { + name, zone, token, base, cdn, proxy, missing string + baseURLs []string + devOnly bool + }{ + {name: "unset", base: "https://img.example.test"}, + {name: "zone only", zone: "zone-abc", base: "https://img.example.test", missing: "CLOUDFLARE_CACHE_PURGE_API_TOKEN"}, + {name: "token only", token: cdnPurgeTestToken, base: "https://img.example.test", missing: "CLOUDFLARE_CACHE_PURGE_ZONE_ID"}, + {name: "proxy disabled", zone: "zone-abc", token: cdnPurgeTestToken, base: "https://img.example.test", proxy: "false", missing: "IMAGE_PROXY_ENABLED"}, + {name: "relative base", zone: "zone-abc", token: cdnPurgeTestToken, base: "/media", cdn: "https://cdn.example.test", missing: "IMAGE_PROXY_BASE_URL"}, + {name: "empty base", zone: "zone-abc", token: cdnPurgeTestToken, base: "", cdn: "https://cdn.example.test", missing: "IMAGE_PROXY_BASE_URL"}, + {name: "relative CDN", zone: "zone-abc", token: cdnPurgeTestToken, base: "https://img.example.test", cdn: "/media", missing: "IMAGE_PROXY_CDN_URL", devOnly: true}, + {name: "base only", zone: "zone-abc", token: cdnPurgeTestToken, base: "https://img.example.test", baseURLs: []string{"https://img.example.test"}}, + {name: "distinct CDN", zone: "zone-abc", token: cdnPurgeTestToken, base: "https://img.example.test", cdn: "https://cdn.example.test", baseURLs: []string{"https://img.example.test", "https://cdn.example.test"}}, + {name: "same CDN", zone: "zone-abc", token: cdnPurgeTestToken, base: "https://img.example.test", cdn: "https://img.example.test", baseURLs: []string{"https://img.example.test"}}, + {name: "unset empty base", base: "", devOnly: true}, + {name: "unset proxy disabled", base: "https://img.example.test", proxy: "false"}, + } { + if test.devOnly && mode == "production" { + continue + } + t.Run(mode+"/"+test.name, func(t *testing.T) { + clearEnv(t) + if mode == "production" { + prodEnv(t) + t.Setenv("ALLOW_UNPROXIED_MEDIA", "true") + } else { + t.Setenv("IS_DEV_ENV", "true") + } + t.Setenv("IMAGE_PROXY_ENABLED", test.proxy) + if test.proxy == "" { + t.Setenv("IMAGE_PROXY_ENABLED", "true") + } + t.Setenv("IMAGE_PROXY_BASE_URL", test.base) + t.Setenv("IMAGE_PROXY_CDN_URL", test.cdn) + t.Setenv("CLOUDFLARE_CACHE_PURGE_ZONE_ID", "") + t.Setenv("CLOUDFLARE_CACHE_PURGE_API_TOKEN", "") + // Control: the otherwise identical environment must load without purge. + if test.missing != "" { + _, err := Load() + require.NoError(t, err, "control without purge") + } + t.Setenv("CLOUDFLARE_CACHE_PURGE_ZONE_ID", test.zone) + t.Setenv("CLOUDFLARE_CACHE_PURGE_API_TOKEN", test.token) + cfg, err := Load() + if test.missing != "" { + require.Error(t, err) + assert.ErrorContains(t, err, test.missing) + assert.NotContains(t, err.Error(), cdnPurgeTestToken) + return + } + require.NoError(t, err) + assert.Equal(t, test.zone != "", cfg.Media.CDNPurge.Enabled()) + if test.zone != "" { + assert.Equal(t, "zone-abc", cfg.Media.CDNPurge.ZoneID) + assert.Equal(t, cdnPurgeTestToken, cfg.Media.CDNPurge.APIToken) + assert.Equal(t, test.baseURLs, cfg.Media.CDNPurge.BaseURLs) + } + assert.NotContains(t, fmt.Sprintf("%v", cfg), cdnPurgeTestToken) + assert.NotContains(t, fmt.Sprintf("%+v", cfg), cdnPurgeTestToken) + assert.NotContains(t, fmt.Sprintf("%#v", cfg), cdnPurgeTestToken) + encoded, err := json.Marshal(cfg) + require.NoError(t, err) + assert.NotContains(t, string(encoded), cdnPurgeTestToken) + }) + } + } +} + +func TestClearEnvForTestClearsCDNPurgeSettings(t *testing.T) { + for _, name := range []string{"CLOUDFLARE_CACHE_PURGE_ZONE_ID", "CLOUDFLARE_CACHE_PURGE_API_TOKEN"} { + t.Setenv(name, "set") + } + ClearEnvForTest(t) + for _, name := range []string{"CLOUDFLARE_CACHE_PURGE_ZONE_ID", "CLOUDFLARE_CACHE_PURGE_API_TOKEN"} { + assert.Empty(t, os.Getenv(name), "%s must be in loadedEnvVars", name) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index a4b65d6..89a370b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -405,6 +405,9 @@ type MediaConfig struct { // the production invariant below before anything starts. ImageProxy imageproxy.Config + // CDNPurge configures optional Cloudflare edge invalidation for blocked media. + CDNPurge CDNPurgeConfig + // AllowUnproxiedMedia acknowledges, for a production deployment, that // media will be served straight from PDS blob endpoints. // @@ -581,9 +584,19 @@ func Load() (*Config, error) { imageProxy.Enabled = imageProxyEnabled cfg.Media = MediaConfig{ - ImageProxy: imageProxy, + ImageProxy: imageProxy, + CDNPurge: CDNPurgeConfig{ + ZoneID: lookup("CLOUDFLARE_CACHE_PURGE_ZONE_ID"), + APIToken: lookup("CLOUDFLARE_CACHE_PURGE_API_TOKEN"), + }, AllowUnproxiedMedia: allowUnproxiedMedia, } + if cfg.Media.CDNPurge.Enabled() { + cfg.Media.CDNPurge.BaseURLs = []string{imageProxy.BaseURL} + if imageProxy.CDNURL != "" && imageProxy.CDNURL != imageProxy.BaseURL { + cfg.Media.CDNPurge.BaseURLs = append(cfg.Media.CDNPurge.BaseURLs, imageProxy.CDNURL) + } + } cfg.CursorSecret = stringVar("CURSOR_SECRET", devCursorSecret) @@ -1051,6 +1064,24 @@ func (c *Config) loadModeration() error { // instead of one restart per mistake. func (c *Config) Validate() error { var problems []string + zoneID := c.Media.CDNPurge.ZoneID + apiToken := c.Media.CDNPurge.APIToken + switch { + case zoneID != "" && apiToken == "": + problems = append(problems, "CLOUDFLARE_CACHE_PURGE_API_TOKEN is required when CLOUDFLARE_CACHE_PURGE_ZONE_ID is set") + case zoneID == "" && apiToken != "": + problems = append(problems, "CLOUDFLARE_CACHE_PURGE_ZONE_ID is required when CLOUDFLARE_CACHE_PURGE_API_TOKEN is set") + case zoneID != "" && apiToken != "": + if !c.Media.ImageProxy.Enabled { + problems = append(problems, "IMAGE_PROXY_ENABLED must be true when CDN purge is configured") + } + if !absoluteHTTPURL(c.Media.ImageProxy.BaseURL) { + problems = append(problems, "IMAGE_PROXY_BASE_URL must be an absolute http(s) URL when CDN purge is configured") + } + if cdn := c.Media.ImageProxy.CDNURL; cdn != "" && !absoluteHTTPURL(cdn) { + problems = append(problems, "IMAGE_PROXY_CDN_URL must be an absolute http(s) URL when CDN purge is configured") + } + } for _, did := range c.Moderation.Admins { if _, err := syntax.ParseDID(did); err != nil { problems = append(problems, "MODERATION_ADMINS must contain only valid DIDs") @@ -1378,3 +1409,28 @@ func (c *Config) mediaProblems() []string { return problems } + +// absoluteHTTPURL accepts only complete HTTP(S) origins, in both dev and production. +func absoluteHTTPURL(raw string) bool { + parsed, err := url.Parse(raw) + return err == nil && (parsed.Scheme == "http" || parsed.Scheme == "https") && parsed.Hostname() != "" +} + +// CDNPurgeConfig holds optional Cloudflare zone credentials and image origins. +type CDNPurgeConfig struct { + ZoneID string + APIToken string `json:"-"` + BaseURLs []string +} + +// Enabled reports whether both required Cloudflare credentials are present. +func (c CDNPurgeConfig) Enabled() bool { return c.ZoneID != "" && c.APIToken != "" } + +// String redacts the credential when a Config is formatted with %v or %+v. +// GoString covers %#v, and the json:"-" tag on APIToken covers JSON encoding. +func (c CDNPurgeConfig) String() string { + return fmt.Sprintf("{ZoneID:%q APIToken:[REDACTED] BaseURLs:%v}", c.ZoneID, c.BaseURLs) +} + +// GoString redacts the credential when a Config is formatted with %#v. +func (c CDNPurgeConfig) GoString() string { return c.String() } diff --git a/internal/config/testing.go b/internal/config/testing.go index 5979023..b2babf8 100644 --- a/internal/config/testing.go +++ b/internal/config/testing.go @@ -31,6 +31,7 @@ var loadedEnvVars = []string{ // variable added there and not here leaks between tests silently — keep the // two in sync, or move the parsing here. "IMAGE_PROXY_ENABLED", "IMAGE_PROXY_BASE_URL", "IMAGE_PROXY_CDN_URL", + "CLOUDFLARE_CACHE_PURGE_ZONE_ID", "CLOUDFLARE_CACHE_PURGE_API_TOKEN", "IMAGE_PROXY_CACHE_PATH", "IMAGE_PROXY_CACHE_MAX_GB", "IMAGE_PROXY_CACHE_TTL_DAYS", "IMAGE_PROXY_CLEANUP_INTERVAL_MINUTES", "IMAGE_PROXY_FETCH_TIMEOUT_SECONDS", "IMAGE_PROXY_MAX_SOURCE_SIZE_MB", "IMAGE_PROXY_MAX_SOURCE_MEGAPIXELS", diff --git a/internal/core/imageproxy/cdn_purge.go b/internal/core/imageproxy/cdn_purge.go new file mode 100644 index 0000000..3a06488 --- /dev/null +++ b/internal/core/imageproxy/cdn_purge.go @@ -0,0 +1,161 @@ +package imageproxy + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "sort" + "strings" + "time" + + "Coves/internal/core/blobs" +) + +// CloudflarePurgerConfig identifies the zone, image origins and request deadline. +type CloudflarePurgerConfig struct { + APIBase string + ZoneID string + APIToken string + BaseURLs []string + Timeout time.Duration +} + +// CloudflarePurger invalidates all registered image presets for each blocked pair. +type CloudflarePurger struct { + endpoint string + token string + bases []string + presets []string + client *http.Client +} + +// CDNPurgeFailure identifies a pair Cloudflare did not acknowledge. +type CDNPurgeFailure struct { + Blob BlockedBlob + Code string +} + +// CDNPurgeResult separates acknowledged and failed owner-scoped blocks. +type CDNPurgeResult struct { + Acknowledged []BlockedBlob + Failed []CDNPurgeFailure +} + +// NewCloudflarePurger constructs a bounded client for a fixed Cloudflare endpoint. +func NewCloudflarePurger(config CloudflarePurgerConfig) (*CloudflarePurger, error) { + if config.APIBase == "" { + return nil, fmt.Errorf("APIBase is required") + } + parsed, err := url.Parse(config.APIBase) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" { + return nil, fmt.Errorf("APIBase must be an absolute http(s) URL") + } + if config.ZoneID == "" { + return nil, fmt.Errorf("ZoneID is required") + } + if config.APIToken == "" { + return nil, fmt.Errorf("APIToken is required") + } + if len(config.BaseURLs) == 0 { + return nil, fmt.Errorf("BaseURLs is required") + } + for _, base := range config.BaseURLs { + parsed, err := url.Parse(base) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" { + return nil, fmt.Errorf("BaseURLs must contain only absolute http(s) URLs") + } + } + if config.Timeout <= 0 { + return nil, fmt.Errorf("Timeout must be greater than zero") + } + presets := ListPresets() + if len(presets)*len(config.BaseURLs) > 100 || len(presets) == 0 { + return nil, fmt.Errorf("one pair must fit within 100 purge URLs") + } + names := make([]string, 0, len(presets)) + for _, preset := range presets { + names = append(names, preset.Name) + } + sort.Strings(names) + // coves:allow-bare-client: fixed Cloudflare API, never an untrusted media URL + client := &http.Client{Timeout: config.Timeout, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + return &CloudflarePurger{ + endpoint: strings.TrimRight(config.APIBase, "/") + "/zones/" + url.PathEscape(config.ZoneID) + "/purge_cache", + token: config.APIToken, + bases: append([]string(nil), config.BaseURLs...), + presets: names, + client: client, + }, nil +} + +// PurgeBlobs sends whole owner/CID pairs in batches of no more than 100 URLs. +func (p *CloudflarePurger) PurgeBlobs(ctx context.Context, blocked []BlockedBlob) CDNPurgeResult { + var result CDNPurgeResult + if len(blocked) == 0 { + return result + } + pairSize := len(p.bases) * len(p.presets) + pairsPerRequest := 100 / pairSize + for start := 0; start < len(blocked); start += pairsPerRequest { + end := min(start+pairsPerRequest, len(blocked)) + batch := blocked[start:end] + files := make([]string, 0, len(batch)*pairSize) + for _, blob := range batch { + for _, base := range p.bases { + for _, preset := range p.presets { + files = append(files, blobs.HydrateImageProxyURL(base, preset, blob.OwnerDID, blob.CID)) + } + } + } + code := p.purgeBatch(ctx, files) + if code == "" { + result.Acknowledged = append(result.Acknowledged, batch...) + } else { + for _, blob := range batch { + result.Failed = append(result.Failed, CDNPurgeFailure{Blob: blob, Code: code}) + } + } + } + return result +} + +// purgeBatch returns only a stable failure code; neither credentials nor remote +// response content may be reflected into moderation logs. +func (p *CloudflarePurger) purgeBatch(ctx context.Context, files []string) string { + data, err := json.Marshal(struct { + Files []string `json:"files"` + }{Files: files}) + if err != nil { + return "transport" + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, p.endpoint, bytes.NewReader(data)) + if err != nil { + return "transport" + } + request.Header.Set("Authorization", "Bearer "+p.token) + request.Header.Set("Content-Type", "application/json") + response, err := p.client.Do(request) + if err != nil { + return "transport" + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Sprintf("http_%d", response.StatusCode) + } + var acknowledgement struct { + Success bool `json:"success"` + } + decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20)) + if err := decoder.Decode(&acknowledgement); err != nil || !acknowledgement.Success { + return "api_unsuccessful" + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + return "api_unsuccessful" + } + return "" +} diff --git a/internal/core/imageproxy/cdn_purge_test.go b/internal/core/imageproxy/cdn_purge_test.go new file mode 100644 index 0000000..38f9264 --- /dev/null +++ b/internal/core/imageproxy/cdn_purge_test.go @@ -0,0 +1,313 @@ +package imageproxy + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "sync" + "testing" + "time" + + "Coves/internal/core/blobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + cdnTestToken = "cf-purge-token-SENTINEL-7f3a" + cdnTestBody = "BODY-SENTINEL-cf-purge-token-SENTINEL-7f3a" +) + +var cdnTestBlob = BlockedBlob{OwnerDID: moderationTestOwner, CID: moderationTestCID} + +var cdnTestPresets = []string{"avatar", "avatar_small", "banner", "content_preview", "content_full", "embed_thumbnail"} + +type cdnRequest struct { + method, path, authorization string + files []string + decodeError error +} + +type cdnEndpoint struct { + mu sync.Mutex + requests []cdnRequest + respond func([]string) (int, string) + wait func(*http.Request) +} + +func (endpoint *cdnEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) { + var body struct { + Files []string `json:"files"` + } + decodeError := json.NewDecoder(r.Body).Decode(&body) + request := cdnRequest{r.Method, r.URL.Path, r.Header.Get("Authorization"), body.Files, decodeError} + endpoint.mu.Lock() + endpoint.requests = append(endpoint.requests, request) + endpoint.mu.Unlock() + if endpoint.wait != nil { + endpoint.wait(r) + } + status, response := http.StatusOK, `{"success":true}` + if endpoint.respond != nil { + status, response = endpoint.respond(body.Files) + } + w.WriteHeader(status) + _, _ = w.Write([]byte(response)) +} + +func (endpoint *cdnEndpoint) snapshot() []cdnRequest { + endpoint.mu.Lock() + defer endpoint.mu.Unlock() + return append([]cdnRequest(nil), endpoint.requests...) +} + +func newCDNTestPurger(t *testing.T, server *httptest.Server, bases []string, timeout time.Duration) *CloudflarePurger { + t.Helper() + purger, err := NewCloudflarePurger(CloudflarePurgerConfig{APIBase: server.URL, ZoneID: "zone-abc", APIToken: cdnTestToken, BaseURLs: bases, Timeout: timeout}) + require.NoError(t, err) + require.NotNil(t, purger) + return purger +} + +func cdnExpectedURLs(blob BlockedBlob, bases []string) []string { + var urls []string + for _, base := range bases { + for _, preset := range cdnTestPresets { + urls = append(urls, blobs.HydrateImageURL(blobs.ImageURLConfig{ProxyEnabled: true, ProxyBaseURL: base}, "", blob.OwnerDID, blob.CID, preset)) + } + } + return urls +} + +func TestCloudflarePurgerOneBlobURLsAndRequest(t *testing.T) { + endpoint := &cdnEndpoint{} + server := httptest.NewServer(endpoint) + defer server.Close() + result := newCDNTestPurger(t, server, []string{"https://img.example.test"}, 5*time.Second).PurgeBlobs(t.Context(), []BlockedBlob{cdnTestBlob}) + requests := endpoint.snapshot() + require.Len(t, requests, 1) + assert.NoError(t, requests[0].decodeError) + assert.Equal(t, "POST", requests[0].method) + assert.Equal(t, "/zones/zone-abc/purge_cache", requests[0].path) + assert.Equal(t, "Bearer "+cdnTestToken, requests[0].authorization) + assert.ElementsMatch(t, []string{ + "https://img.example.test/img/avatar/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + "https://img.example.test/img/avatar_small/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + "https://img.example.test/img/banner/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + "https://img.example.test/img/content_preview/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + "https://img.example.test/img/content_full/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + "https://img.example.test/img/embed_thumbnail/plain/did:plc:moderationimageowner/bafyreihgdyzzpkkzq2izfnhcmm77ycuacvkuziwbnqxfxtqsz7tmxwhnshi", + }, requests[0].files) + assert.ElementsMatch(t, cdnExpectedURLs(cdnTestBlob, []string{"https://img.example.test"}), requests[0].files) + assert.Equal(t, []BlockedBlob{cdnTestBlob}, result.Acknowledged) + assert.Empty(t, result.Failed) +} + +func TestCloudflarePurgerTwoBasesOneRequest(t *testing.T) { + endpoint := &cdnEndpoint{} + server := httptest.NewServer(endpoint) + defer server.Close() + result := newCDNTestPurger(t, server, []string{"https://img.example.test", "https://cdn.example.test"}, 5*time.Second).PurgeBlobs(t.Context(), []BlockedBlob{cdnTestBlob}) + requests := endpoint.snapshot() + require.Len(t, requests, 1) + assert.NoError(t, requests[0].decodeError) + assert.Len(t, requests[0].files, 12) + assert.ElementsMatch(t, cdnExpectedURLs(cdnTestBlob, []string{"https://img.example.test", "https://cdn.example.test"}), requests[0].files) + for _, preset := range cdnTestPresets { + assert.Contains(t, requests[0].files, blobs.HydrateImageURL(blobs.ImageURLConfig{ProxyEnabled: true, ProxyBaseURL: "https://img.example.test", CDNURL: "https://cdn.example.test"}, "", cdnTestBlob.OwnerDID, cdnTestBlob.CID, preset)) + } + assert.Equal(t, []BlockedBlob{cdnTestBlob}, result.Acknowledged) + assert.Empty(t, result.Failed) +} + +func cdnTwentyBlobs() []BlockedBlob { + var pairs []BlockedBlob + for index := range 20 { + pairs = append(pairs, BlockedBlob{OwnerDID: fmt.Sprintf("did:plc:cdnowner%02d", index), CID: moderationTestCID}) + } + return pairs +} + +func TestCloudflarePurgerBatchesWholePairsAndEmptyInput(t *testing.T) { + for _, test := range []struct { + name string + bases []string + }{ + {"one base", []string{"https://img.example.test"}}, + {"two bases", []string{"https://img.example.test", "https://cdn.example.test"}}, + } { + t.Run(test.name, func(t *testing.T) { + endpoint := &cdnEndpoint{} + server := httptest.NewServer(endpoint) + defer server.Close() + purger := newCDNTestPurger(t, server, test.bases, 5*time.Second) + empty := purger.PurgeBlobs(t.Context(), nil) + assert.Empty(t, endpoint.snapshot()) + assert.Empty(t, empty.Acknowledged) + assert.Empty(t, empty.Failed) + pairs := cdnTwentyBlobs() + result := purger.PurgeBlobs(t.Context(), pairs) + requests := endpoint.snapshot() + require.Greater(t, len(requests), 1) + var allFiles []string + for _, request := range requests { + assert.NoError(t, request.decodeError) + assert.LessOrEqual(t, len(request.files), 100) + allFiles = append(allFiles, request.files...) + } + var expected []string + for _, pair := range pairs { + urls := cdnExpectedURLs(pair, test.bases) + expected = append(expected, urls...) + containing := 0 + for _, request := range requests { + if slices.Contains(request.files, urls[0]) { + containing++ + for _, url := range urls { + assert.Contains(t, request.files, url) + } + } + } + assert.Equal(t, 1, containing, "pair %v", pair) + } + assert.ElementsMatch(t, expected, allFiles) + assert.ElementsMatch(t, pairs, result.Acknowledged) + assert.Empty(t, result.Failed) + }) + } +} + +func TestCloudflarePurgerFailureOnlyMarksItsBatch(t *testing.T) { + bases := []string{"https://img.example.test", "https://cdn.example.test"} + pairs := cdnTwentyBlobs() + firstURL := cdnExpectedURLs(pairs[0], bases)[0] + endpoint := &cdnEndpoint{respond: func(files []string) (int, string) { + if slices.Contains(files, firstURL) { + return 500, `{"success":false}` + } + return 200, `{"success":true}` + }} + server := httptest.NewServer(endpoint) + defer server.Close() + result := newCDNTestPurger(t, server, bases, 5*time.Second).PurgeBlobs(t.Context(), pairs) + requests := endpoint.snapshot() + require.Greater(t, len(requests), 1) + var failed, acknowledged []BlockedBlob + failedIndex := -1 + for index, request := range requests { + if slices.Contains(request.files, firstURL) { + failedIndex = index + } + } + require.GreaterOrEqual(t, failedIndex, 0) + assert.Less(t, failedIndex, len(requests)-1, "failure must precede a later successful batch") + for _, pair := range pairs { + inFailedRequest := slices.Contains(requests[failedIndex].files, cdnExpectedURLs(pair, bases)[0]) + if inFailedRequest { + failed = append(failed, pair) + } else { + acknowledged = append(acknowledged, pair) + } + } + assert.ElementsMatch(t, acknowledged, result.Acknowledged) + assert.Len(t, result.Failed, len(failed)) + var failedBlobs []BlockedBlob + for _, failure := range result.Failed { + assert.Equal(t, "http_500", failure.Code) + failedBlobs = append(failedBlobs, failure.Blob) + assert.NotContains(t, result.Acknowledged, failure.Blob) + } + assert.ElementsMatch(t, failed, failedBlobs) +} + +func TestCloudflarePurgerResponseFailuresAndTimeout(t *testing.T) { + for _, test := range []struct { + name string + status int + body, code string + closed, timeout bool + }{ + {"unsuccessful", 200, `{"success":false,"detail":"` + cdnTestBody + `"}`, "api_unsuccessful", false, false}, + {"invalid json", 200, `not json ` + cdnTestBody, "api_unsuccessful", false, false}, + {"empty object", 200, `{}`, "api_unsuccessful", false, false}, + {"missing success", 200, `{"detail":"` + cdnTestBody + `"}`, "api_unsuccessful", false, false}, + {"accepted", 202, `{"success":true,"detail":"` + cdnTestBody + `"}`, "", false, false}, + {"rate limited", 429, cdnTestBody, "http_429", false, false}, + {"server error", 500, cdnTestBody, "http_500", false, false}, + {"closed server", 200, cdnTestBody, "transport", true, false}, + {"timeout", 200, cdnTestBody, "transport", false, true}, + } { + t.Run(test.name, func(t *testing.T) { + release := make(chan struct{}) + endpoint := &cdnEndpoint{respond: func([]string) (int, string) { return test.status, test.body }} + if test.timeout { + endpoint.wait = func(r *http.Request) { + select { + case <-r.Context().Done(): + case <-release: + } + } + } + server := httptest.NewServer(endpoint) + if test.closed { + server.Close() + } else { + t.Cleanup(server.Close) + } + if test.timeout { + t.Cleanup(func() { close(release) }) + } + timeout := 5 * time.Second + if test.timeout { + timeout = 50 * time.Millisecond + } + purger := newCDNTestPurger(t, server, []string{"https://img.example.test"}, timeout) + var result CDNPurgeResult + if test.timeout { + done := make(chan CDNPurgeResult, 1) + go func() { done <- purger.PurgeBlobs(t.Context(), []BlockedBlob{cdnTestBlob}) }() + select { + case result = <-done: + case <-time.After(10 * time.Second): + t.Fatal("purge did not respect timeout") + } + } else { + result = purger.PurgeBlobs(t.Context(), []BlockedBlob{cdnTestBlob}) + } + if test.code == "" { + assert.Equal(t, []BlockedBlob{cdnTestBlob}, result.Acknowledged) + assert.Empty(t, result.Failed) + } else { + assert.Empty(t, result.Acknowledged) + assert.Equal(t, []CDNPurgeFailure{{Blob: cdnTestBlob, Code: test.code}}, result.Failed) + } + assert.NotContains(t, fmt.Sprintf("%+v", result), cdnTestToken) + assert.NotContains(t, fmt.Sprintf("%+v", result), cdnTestBody) + }) + } +} + +func TestNewCloudflarePurgerRequiresConfiguration(t *testing.T) { + for _, test := range []struct { + name string + change func(*CloudflarePurgerConfig) + }{ + {"APIBase", func(c *CloudflarePurgerConfig) { c.APIBase = "" }}, + {"ZoneID", func(c *CloudflarePurgerConfig) { c.ZoneID = "" }}, + {"APIToken", func(c *CloudflarePurgerConfig) { c.APIToken = "" }}, + {"BaseURLs", func(c *CloudflarePurgerConfig) { c.BaseURLs = nil }}, + {"Timeout", func(c *CloudflarePurgerConfig) { c.Timeout = 0 }}, + } { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + defer server.Close() + config := CloudflarePurgerConfig{APIBase: server.URL, ZoneID: "zone-abc", APIToken: cdnTestToken, BaseURLs: []string{"https://img.example.test"}, Timeout: 5 * time.Second} + test.change(&config) + _, err := NewCloudflarePurger(config) + require.Error(t, err) + assert.NotContains(t, err.Error(), cdnTestToken) + }) + } +} diff --git a/internal/core/moderation/cdn_purge.go b/internal/core/moderation/cdn_purge.go new file mode 100644 index 0000000..7835e1c --- /dev/null +++ b/internal/core/moderation/cdn_purge.go @@ -0,0 +1,42 @@ +package moderation + +import ( + "context" + "log/slog" + + "Coves/internal/core/imageproxy" +) + +// CDNPurger invalidates CDN URLs for owner-scoped media blocks. +type CDNPurger interface { + PurgeBlobs(ctx context.Context, blobs []imageproxy.BlockedBlob) imageproxy.CDNPurgeResult +} + +// MediaReconcilerOption configures optional post-commit media actions. +type MediaReconcilerOption func(*MediaReconciler) + +// WithCDNPurger enables CDN invalidation after media reconciliation commits. +func WithCDNPurger(purger CDNPurger) MediaReconcilerOption { + return func(reconciler *MediaReconciler) { reconciler.cdnPurger = purger } +} + +// purgeCDNMediaBlocks invalidates only the URLs belonging to owner-scoped +// blocks. CDN failures are best effort; the committed block already prevents +// serving the blob at the origin. +func purgeCDNMediaBlocks(ctx context.Context, purger CDNPurger, blocks []MediaBlock) { + if purger == nil { + return + } + var owned []imageproxy.BlockedBlob + for _, block := range blocks { + if block.OwnerDID != "" { + owned = append(owned, imageproxy.BlockedBlob{OwnerDID: block.OwnerDID, CID: block.BlobCID}) + } + } + if len(owned) == 0 { + return + } + for _, failure := range purger.PurgeBlobs(ctx, owned).Failed { + slog.Error("moderation CDN purge failed", "did", failure.Blob.OwnerDID, "cid", failure.Blob.CID, "code", failure.Code) + } +} diff --git a/internal/core/moderation/cdn_purge_test.go b/internal/core/moderation/cdn_purge_test.go new file mode 100644 index 0000000..b8adf69 --- /dev/null +++ b/internal/core/moderation/cdn_purge_test.go @@ -0,0 +1,344 @@ +package moderation_test + +import ( + "context" + "errors" + "log/slog" + "sync" + "testing" + + "Coves/internal/core/imageproxy" + "Coves/internal/core/moderation" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type recordedCDNPurge struct { + blobs []imageproxy.BlockedBlob + contextLive bool + removalCommitted bool +} + +type recordingCDNPurger struct { + mu sync.Mutex + store *inMemoryModerationStore + result imageproxy.CDNPurgeResult + calls []recordedCDNPurge +} + +func (purger *recordingCDNPurger) PurgeBlobs(ctx context.Context, blobs []imageproxy.BlockedBlob) imageproxy.CDNPurgeResult { + call := recordedCDNPurge{blobs: append([]imageproxy.BlockedBlob(nil), blobs...), contextLive: ctx.Err() == nil} + if purger.store != nil && purger.store.mu.TryLock() { + call.removalCommitted = purger.store.state.activeRemovals[inMemoryModerationDecisionKey{removeRulesInstanceDID, removeRulesURI}] != "" + for _, blob := range blobs { + found := false + for block, active := range purger.store.state.mediaBlocks { + if active && block.OwnerDID == blob.OwnerDID && block.BlobCID == blob.CID { + found = true + } + } + call.removalCommitted = call.removalCommitted && found + } + purger.store.mu.Unlock() + } + purger.mu.Lock() + defer purger.mu.Unlock() + purger.calls = append(purger.calls, call) + return purger.result +} + +func (purger *recordingCDNPurger) snapshot() []recordedCDNPurge { + purger.mu.Lock() + defer purger.mu.Unlock() + return append([]recordedCDNPurge(nil), purger.calls...) +} + +type afterBlocksFailureStore struct { + *inMemoryModerationStore + err error +} + +func (store *afterBlocksFailureStore) InTransaction(ctx context.Context, fn func(context.Context, moderation.Transaction) error) error { + return store.inMemoryModerationStore.InTransaction(ctx, func(ctx context.Context, tx moderation.Transaction) error { + return fn(ctx, &afterBlocksFailureTransaction{Transaction: tx, err: store.err}) + }) +} + +type afterBlocksFailureTransaction struct { + moderation.Transaction + err error + inserted bool +} + +func (tx *afterBlocksFailureTransaction) InsertMediaBlocks(ctx context.Context, blocks []moderation.MediaBlock) error { + if err := tx.Transaction.InsertMediaBlocks(ctx, blocks); err != nil { + return err + } + tx.inserted = true + return nil +} + +func (tx *afterBlocksFailureTransaction) SaveIdempotencyRecord(ctx context.Context, record moderation.IdempotencyRecord) error { + if tx.inserted { + return tx.err + } + return tx.Transaction.SaveIdempotencyRecord(ctx, record) +} + +type cancelAfterCommitStore struct { + *inMemoryModerationStore + cancel context.CancelFunc +} + +func (store *cancelAfterCommitStore) InTransaction(ctx context.Context, fn func(context.Context, moderation.Transaction) error) error { + err := store.inMemoryModerationStore.InTransaction(ctx, fn) + if err == nil { + store.cancel() + } + return err +} + +type recordingCDNLogHandler struct { + mu sync.Mutex + records []slog.Record +} + +func (*recordingCDNLogHandler) Enabled(context.Context, slog.Level) bool { return true } +func (handler *recordingCDNLogHandler) Handle(_ context.Context, record slog.Record) error { + handler.mu.Lock() + defer handler.mu.Unlock() + handler.records = append(handler.records, record.Clone()) + return nil +} +func (handler *recordingCDNLogHandler) WithAttrs([]slog.Attr) slog.Handler { return handler } +func (handler *recordingCDNLogHandler) WithGroup(string) slog.Handler { return handler } +func (handler *recordingCDNLogHandler) snapshot() []slog.Record { + handler.mu.Lock() + defer handler.mu.Unlock() + return append([]slog.Record(nil), handler.records...) +} + +func TestRemoveContentCDNPurgeAfterCommit(t *testing.T) { + for _, test := range []struct { + name string + images []string + reason string + want []imageproxy.BlockedBlob + rollback bool + cancel bool + failures bool + }{ + { + name: "spam three images", reason: removeRulesSpam, + images: []string{"bafyreifirstimage", "bafyreisecondimage", "bafyreithirdimage"}, + want: []imageproxy.BlockedBlob{ + {OwnerDID: "did:plc:commentauthor", CID: "bafyreifirstimage"}, + {OwnerDID: "did:plc:commentauthor", CID: "bafyreisecondimage"}, + {OwnerDID: "did:plc:commentauthor", CID: "bafyreithirdimage"}, + }, + }, + { + name: "illegal content excludes ownerless blocks", reason: removeRulesIllegal, + images: []string{"bafyreifirstimage", "bafyreisecondimage"}, + want: []imageproxy.BlockedBlob{ + {OwnerDID: "did:plc:commentauthor", CID: "bafyreifirstimage"}, + {OwnerDID: "did:plc:commentauthor", CID: "bafyreisecondimage"}, + }, + }, + {name: "no images", reason: removeRulesSpam}, + { + name: "transaction rolls back after blocks", reason: removeRulesSpam, + images: []string{"bafyreifirstimage"}, rollback: true, + }, + { + name: "caller cancelled after commit", reason: removeRulesSpam, + images: []string{"bafyreifirstimage"}, cancel: true, + want: []imageproxy.BlockedBlob{{OwnerDID: "did:plc:commentauthor", CID: "bafyreifirstimage"}}, + }, + { + name: "two failed pairs leave mutation and local purge intact", reason: removeRulesSpam, + images: []string{"bafyreifirstimage", "bafyreisecondimage", "bafyreithirdimage"}, failures: true, + want: []imageproxy.BlockedBlob{ + {OwnerDID: "did:plc:commentauthor", CID: "bafyreifirstimage"}, + {OwnerDID: "did:plc:commentauthor", CID: "bafyreisecondimage"}, + {OwnerDID: "did:plc:commentauthor", CID: "bafyreithirdimage"}, + }, + }, + } { + t.Run(test.name, func(t *testing.T) { + scenario := newRemoveRulesScenario() + comment := scenario.store.state.indexedComments[removeRulesURI] + comment.ImageCIDs = test.images + scenario.store.state.indexedComments[removeRulesURI] = comment + scenario.request.Reason = test.reason + cdn := &recordingCDNPurger{store: scenario.store, result: imageproxy.CDNPurgeResult{Acknowledged: test.want}} + var store moderation.Store = scenario.store + var ctx context.Context = t.Context() + if test.rollback { + store = &afterBlocksFailureStore{inMemoryModerationStore: scenario.store, err: errCDNRollback} + } + if test.cancel { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(t.Context()) + t.Cleanup(cancel) + store = &cancelAfterCommitStore{inMemoryModerationStore: scenario.store, cancel: cancel} + } + if test.failures { + cdn.result = imageproxy.CDNPurgeResult{ + Acknowledged: []imageproxy.BlockedBlob{{OwnerDID: "did:plc:commentauthor", CID: "bafyreithirdimage"}}, + Failed: []imageproxy.CDNPurgeFailure{ + {Blob: imageproxy.BlockedBlob{OwnerDID: "did:plc:commentauthor", CID: "bafyreifirstimage"}, Code: "http_500"}, + {Blob: imageproxy.BlockedBlob{OwnerDID: "did:plc:commentauthor", CID: "bafyreisecondimage"}, Code: "transport"}, + }, + } + logs := &recordingCDNLogHandler{} + previous := slog.Default() + slog.SetDefault(slog.New(logs)) + t.Cleanup(func() { slog.SetDefault(previous) }) + scenario.withCDNPurger(store, cdn) + result, err := scenario.service.RemoveContent(ctx, removeRulesAdminDID, scenario.request) + require.NoError(t, err) + baseline := newRemoveRulesScenario() + baselineComment := baseline.store.state.indexedComments[removeRulesURI] + baselineComment.ImageCIDs = test.images + baseline.store.state.indexedComments[removeRulesURI] = baselineComment + baselineCDN := &recordingCDNPurger{store: baseline.store, result: imageproxy.CDNPurgeResult{Acknowledged: test.want}} + baseline.withCDNPurger(baseline.store, baselineCDN) + baselineResult, baselineErr := baseline.service.RemoveContent(t.Context(), removeRulesAdminDID, baseline.request) + require.NoError(t, baselineErr) + require.Equal(t, moderation.OutcomeApplied, baselineResult.Outcome) + require.Equal(t, moderation.OutcomeApplied, result.Outcome) + assert.Equal(t, baselineResult, result) + assert.ElementsMatch(t, []removeRulesOwnerPurge{ + {ownerDID: "did:plc:commentauthor", blobCID: "bafyreifirstimage"}, + {ownerDID: "did:plc:commentauthor", blobCID: "bafyreisecondimage"}, + {ownerDID: "did:plc:commentauthor", blobCID: "bafyreithirdimage"}, + }, scenario.purger.ownerPurges) + assertCDNPurgeCalls(t, cdn.snapshot(), test.want) + assertCDNFailureLogs(t, logs.snapshot()) + return + } + scenario.withCDNPurger(store, cdn) + result, err := scenario.service.RemoveContent(ctx, removeRulesAdminDID, scenario.request) + if test.rollback { + require.ErrorIs(t, err, errCDNRollback) + assert.Nil(t, result) + assert.Empty(t, scenario.store.state.mediaBlocks) + assert.Empty(t, scenario.store.state.activeRemovals) + assert.Empty(t, scenario.purger.ownerPurges) + assert.Empty(t, cdn.snapshot()) + return + } + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, moderation.OutcomeApplied, result.Outcome) + if test.cancel { + assert.ErrorIs(t, ctx.Err(), context.Canceled) + } + if len(test.want) == 0 { + assert.Empty(t, cdn.snapshot()) + return + } + assertCDNPurgeCalls(t, cdn.snapshot(), test.want) + }) + } +} + +var errCDNRollback = errors.New("injected after media insertion") + +func assertCDNPurgeCalls(t *testing.T, calls []recordedCDNPurge, want []imageproxy.BlockedBlob) { + t.Helper() + require.Len(t, calls, 1) + assert.ElementsMatch(t, want, calls[0].blobs) + assert.True(t, calls[0].removalCommitted, "removal and blocks must be visible outside the transaction") + assert.True(t, calls[0].contextLive, "purge must have a live context even after caller cancellation") +} + +func assertCDNFailureLogs(t *testing.T, records []slog.Record) { + t.Helper() + var errorsOnly []slog.Record + for _, record := range records { + if record.Level == slog.LevelError { + errorsOnly = append(errorsOnly, record) + } + } + require.Len(t, errorsOnly, 2) + want := []struct{ cid, code string }{{"bafyreifirstimage", "http_500"}, {"bafyreisecondimage", "transport"}} + seen := make(map[string]int) + for _, record := range errorsOnly { + values := make([]string, 0) + record.Attrs(func(attr slog.Attr) bool { + values = append(values, attr.Value.String()) + return true + }) + assert.Contains(t, values, "did:plc:commentauthor") + assert.NotContains(t, values, "bafyreithirdimage") + matched := false + for _, pair := range want { + if containsCDNLogValue(values, pair.cid) { + matched = true + seen[pair.cid]++ + assert.Contains(t, values, pair.code) + } + } + assert.True(t, matched, "unexpected CDN failure record attributes: %v", values) + } + assert.Equal(t, map[string]int{"bafyreifirstimage": 1, "bafyreisecondimage": 1}, seen) +} + +func containsCDNLogValue(values []string, wanted string) bool { + for _, value := range values { + if value == wanted { + return true + } + } + return false +} + +func TestRestoreContentDoesNotPurgeCDN(t *testing.T) { + scenario := newRemoveRulesScenario() + cdn := &recordingCDNPurger{store: scenario.store} + scenario.withCDNPurger(scenario.store, cdn) + removed, err := scenario.service.RemoveContent(t.Context(), removeRulesAdminDID, scenario.request) + require.NoError(t, err) + require.NotNil(t, removed) + require.NotNil(t, removed.Action) + before := cdn.snapshot() + require.Len(t, before, 1, "removal must exercise the configured purger") + restored, err := scenario.service.RestoreContent(t.Context(), restoreRulesAdminDID, moderation.RestoreContentRequest{ + ActionID: removed.Action.ID, ReviewedSubject: &moderation.StrongRef{URI: removeRulesURI, CID: removeRulesCID}, + ExpectedVersion: removed.State.Version, IdempotencyKey: "restore-cdn", Reason: removeRulesSpam, + }) + require.NoError(t, err) + require.NotNil(t, restored) + assert.Equal(t, moderation.OutcomeApplied, restored.Outcome) + assert.Equal(t, before, cdn.snapshot()) +} + +func TestMediaReconcilerPurgesOwnerScopedBlobsFromCDN(t *testing.T) { + local := &removeRulesPurger{} + cdn := &recordingCDNPurger{} + reconciler := moderation.NewMediaReconciler(postRulesMediaBinder{&postRulesMediaTransaction{}}, removeRulesInstanceDID, local, moderation.WithCDNPurger(cdn)) + reconciler.Purge(nil) + reconciler.Purge([]moderation.MediaBlock{{BlobCID: "bafyreiownerlessonly"}}) + assert.Empty(t, cdn.snapshot()) + assert.Equal(t, []string{"bafyreiownerlessonly"}, local.blobPurges) + reconciler.Purge([]moderation.MediaBlock{ + {OwnerDID: "did:plc:postauthor", BlobCID: "bafyreifirstimage"}, + {BlobCID: "bafyreithirdimage"}, + {OwnerDID: "did:plc:postauthor", BlobCID: "bafyreisecondimage"}, + }) + calls := cdn.snapshot() + require.Len(t, calls, 1) + assert.ElementsMatch(t, []imageproxy.BlockedBlob{ + {OwnerDID: "did:plc:postauthor", CID: "bafyreifirstimage"}, + {OwnerDID: "did:plc:postauthor", CID: "bafyreisecondimage"}, + }, calls[0].blobs) + assert.True(t, calls[0].contextLive) + assert.ElementsMatch(t, []removeRulesOwnerPurge{ + {ownerDID: "did:plc:postauthor", blobCID: "bafyreifirstimage"}, + {ownerDID: "did:plc:postauthor", blobCID: "bafyreisecondimage"}, + }, local.ownerPurges) + assert.Equal(t, []string{"bafyreiownerlessonly", "bafyreithirdimage"}, local.blobPurges) +} diff --git a/internal/core/moderation/media.go b/internal/core/moderation/media.go index d683282..6a3880e 100644 --- a/internal/core/moderation/media.go +++ b/internal/core/moderation/media.go @@ -31,11 +31,16 @@ type MediaReconciler struct { binder TransactionBinder instanceDID string purger MediaPurger + cdnPurger CDNPurger } // NewMediaReconciler builds a MediaReconciler. -func NewMediaReconciler(binder TransactionBinder, instanceDID string, purger MediaPurger) *MediaReconciler { - return &MediaReconciler{binder: binder, instanceDID: instanceDID, purger: purger} +func NewMediaReconciler(binder TransactionBinder, instanceDID string, purger MediaPurger, options ...MediaReconcilerOption) *MediaReconciler { + reconciler := &MediaReconciler{binder: binder, instanceDID: instanceDID, purger: purger} + for _, option := range options { + option(reconciler) + } + return reconciler } // ReconcileTx blocks images newly present on a subject with an active @@ -83,6 +88,7 @@ func (r *MediaReconciler) ReconcileIncomingTx(ctx context.Context, tx *sql.Tx, s // Purge removes cached bytes of newly blocked blobs after commit. func (r *MediaReconciler) Purge(blocks []MediaBlock) { purgeMediaBlocks(r.purger, blocks) + purgeCDNMediaBlocks(context.Background(), r.cdnPurger, blocks) } func purgeMediaBlocks(purger MediaPurger, blocks []MediaBlock) { diff --git a/internal/core/moderation/remove.go b/internal/core/moderation/remove.go index ca0655e..98577e7 100644 --- a/internal/core/moderation/remove.go +++ b/internal/core/moderation/remove.go @@ -168,5 +168,6 @@ func (s *service) removeContent(ctx context.Context, actorDID string, request Re return nil, fmt.Errorf("%w: %w", ErrModerationUnavailable, err) } purgeMediaBlocks(s.config.Purger, newlyBlocked) + purgeCDNMediaBlocks(context.WithoutCancel(ctx), s.config.CDNPurger, newlyBlocked) return result, nil } diff --git a/internal/core/moderation/remove_rules_test.go b/internal/core/moderation/remove_rules_test.go index ad9e03c..a03a6fa 100644 --- a/internal/core/moderation/remove_rules_test.go +++ b/internal/core/moderation/remove_rules_test.go @@ -82,6 +82,14 @@ func newRemoveRulesScenario() removeRulesScenario { } } +func (scenario *removeRulesScenario) withCDNPurger(store moderation.Store, cdn moderation.CDNPurger) { + scenario.service = moderation.NewService(scenario.reader, store, moderation.Config{ + InstanceDID: removeRulesInstanceDID, IdempotencyRetention: 24 * time.Hour, + MaxLiveIdempotencyKeys: 1000, Now: func() time.Time { return scenario.store.now }, + Purger: scenario.purger, CDNPurger: cdn, + }) +} + func assertRemoveRulesNoWrites(t *testing.T, scenario removeRulesScenario) { t.Helper() assert.Empty(t, scenario.store.writeCalls) diff --git a/internal/core/moderation/types.go b/internal/core/moderation/types.go index 5182c23..808cbea 100644 --- a/internal/core/moderation/types.go +++ b/internal/core/moderation/types.go @@ -151,6 +151,7 @@ type Config struct { IdempotencyRetention time.Duration MaxLiveIdempotencyKeys int Purger MediaPurger + CDNPurger CDNPurger Now func() time.Time CursorSecret string CommunityResolver CommunityResolver diff --git a/tests/testkit/cloudflare_purge.go b/tests/testkit/cloudflare_purge.go new file mode 100644 index 0000000..4501f6b --- /dev/null +++ b/tests/testkit/cloudflare_purge.go @@ -0,0 +1,77 @@ +package testkit + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" +) + +type CloudflarePurgeRequest struct { + Method, Path, Authorization string + Files []string + DecodeError error +} + +type CloudflarePurgeEndpoint struct { + server *httptest.Server + mu sync.Mutex + requests []CloudflarePurgeRequest + status int + body string + hook func([]string) +} + +func NewCloudflarePurgeEndpoint(t *testing.T) *CloudflarePurgeEndpoint { + t.Helper() + endpoint := &CloudflarePurgeEndpoint{status: http.StatusOK, body: `{"success":true}`} + endpoint.server = httptest.NewServer(http.HandlerFunc(endpoint.serve)) + t.Cleanup(endpoint.server.Close) + return endpoint +} + +func (endpoint *CloudflarePurgeEndpoint) URL() string { return endpoint.server.URL } + +func (endpoint *CloudflarePurgeEndpoint) SetResponse(status int, body string) { + endpoint.mu.Lock() + defer endpoint.mu.Unlock() + endpoint.status, endpoint.body = status, body +} + +func (endpoint *CloudflarePurgeEndpoint) SetHook(hook func([]string)) { + endpoint.mu.Lock() + defer endpoint.mu.Unlock() + endpoint.hook = hook +} + +func (endpoint *CloudflarePurgeEndpoint) Requests() []CloudflarePurgeRequest { + endpoint.mu.Lock() + defer endpoint.mu.Unlock() + requests := make([]CloudflarePurgeRequest, len(endpoint.requests)) + for i, request := range endpoint.requests { + requests[i] = request + requests[i].Files = append([]string(nil), request.Files...) + } + return requests +} + +func (endpoint *CloudflarePurgeEndpoint) serve(w http.ResponseWriter, r *http.Request) { + var payload struct { + Files []string `json:"files"` + } + decodeError := json.NewDecoder(r.Body).Decode(&payload) + files := append([]string(nil), payload.Files...) + endpoint.mu.Lock() + endpoint.requests = append(endpoint.requests, CloudflarePurgeRequest{ + Method: r.Method, Path: r.URL.Path, Authorization: r.Header.Get("Authorization"), + Files: files, DecodeError: decodeError, + }) + status, body, hook := endpoint.status, endpoint.body, endpoint.hook + endpoint.mu.Unlock() + if hook != nil { + hook(append([]string(nil), files...)) + } + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) +} -- 2.54.0