diff --git a/README.md b/README.md index 96c3f0d..edaf519 100644 --- a/README.md +++ b/README.md @@ -301,6 +301,8 @@ curl -X DELETE localhost:8091/admin/communities \ curl localhost:8091/admin/metrics \ -H "Authorization: Bearer dev-admin-token" # expvar counters +# (includes tidepool_lexicon_validation_failures — non-zero in production, +# where validation failures log-and-write, means investigate) # re-emit a repo's records onto the firehose as delete+create commit pairs # (identical values, so at-uris and CIDs are unchanged). For the relay @@ -311,8 +313,22 @@ curl localhost:8091/admin/metrics \ curl -X POST localhost:8091/admin/reemit \ -H "Authorization: Bearer dev-admin-token" \ -d '{"did":"did:plc:aaa..."}' -# (includes tidepool_lexicon_validation_failures — non-zero in production, -# where validation failures log-and-write, means investigate) + +# origin-verified cleanup for deletes the bridge missed: each ap_id is +# re-fetched from its origin (redirects off that origin refuse the fetch) +# and deleted ONLY if the origin serves a tombstone (HTTP 410 or an AP +# Tombstone body). Live objects, unknown ids, actor profiles, and 404s are +# reported, never touched — a 404 also covers the 401/403 a secure-mode +# instance serves for objects it will not show us, so it means "not +# visible", not "deleted". Max 200 ap_ids per request (chunk beyond that; +# 400 otherwise). Idempotent. +curl -X POST localhost:8091/admin/objects/sweep-deleted \ + -H "Authorization: Bearer dev-admin-token" \ + -d '{"ap_ids":["https://lemmy.world/post/123"]}' +# 200 with {"requested":N,"swept":M,"deleted":…,"failed":…, +# "truncated":bool,"result":[{"ap_id":…,"outcome":…}]}. +# swept < requested ("truncated":true) means the client hung up mid-batch — +# the remaining ids were never checked; re-run them. ``` ### Declarative follow list (`FOLLOW_LIST_PATH`) diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index be5a424..f930a10 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -405,6 +405,7 @@ func run(logger *slog.Logger) error { Service: serviceActor, Backfill: backfill, Repos: repoManager, + Sweeper: handler, Logger: logger, }) if err != nil { diff --git a/internal/ingest/follow.go b/internal/ingest/follow.go index 86355b9..069c2e4 100644 --- a/internal/ingest/follow.go +++ b/internal/ingest/follow.go @@ -28,7 +28,9 @@ type FollowClient interface { SendActivity(ctx context.Context, inboxURL string, activity any) error } -// AdminOptions configures NewAdmin. Everything except Logger is required. +// AdminOptions configures NewAdmin. Token, Client, Materializer, +// Communities, and Service are required; Backfill, Repos, Sweeper, and +// Logger are optional (each missing dependency's endpoint answers 501). type AdminOptions struct { // Token is the bearer token protecting /admin (config.AdminToken). Token string @@ -45,17 +47,24 @@ type AdminOptions struct { Backfill Backfiller // Repos serves POST /admin/reemit (optional; the endpoint answers 501 // when nil). See reemit.go for what re-emission is for. - Repos RepoReemitter - Logger *slog.Logger + Repos RepoReemitter + // Sweeper serves POST /admin/objects/sweep-deleted (optional; the + // endpoint answers 501 when nil). See sweep.go. + Sweeper DeleteSweeper + Logger *slog.Logger } -// Admin is the operator API driving the community subscription lifecycle: +// Admin is the operator API driving the community subscription lifecycle, +// plus the maintenance endpoints for bridged objects, repos, and counters: // // POST /admin/communities {"community":"!tech@lemmy.world"} // DELETE /admin/communities {"community":"!tech@lemmy.world"} // GET /admin/communities // POST /admin/communities/backfill {"community":"!tech@lemmy.world"} // POST /admin/communities/reconcile (follow list configured only) +// POST /admin/reemit {"did":"did:plc:..."} (or {} for all) +// POST /admin/objects/sweep-deleted {"ap_ids":["https://..."]} +// GET /admin/metrics (tidepool's own expvar counters) // // All endpoints require "Authorization: Bearer $ADMIN_TOKEN". type Admin struct { @@ -66,6 +75,7 @@ type Admin struct { service *ap.ServiceActor backfill Backfiller repos RepoReemitter + sweeper DeleteSweeper logger *slog.Logger // reconciler serves POST /admin/communities/reconcile; nil (the // endpoint answers 501) unless a follow list is configured. Set once @@ -107,6 +117,7 @@ func NewAdmin(opts AdminOptions) (*Admin, error) { service: opts.Service, backfill: opts.Backfill, repos: opts.Repos, + sweeper: opts.Sweeper, logger: logger, }, nil } @@ -126,6 +137,7 @@ func (a *Admin) Routes(r chi.Router) { r.Post("/communities/backfill", a.handleBackfill) r.Post("/communities/reconcile", a.handleReconcile) r.Post("/reemit", a.handleReemit) + r.Post("/objects/sweep-deleted", a.handleSweepDeleted) r.Method(http.MethodGet, "/metrics", http.HandlerFunc(scopedMetrics)) }) } diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 59eed46..74e7c0c 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -323,6 +323,7 @@ func newHarness(t *testing.T) *harness { Communities: communities, Service: h.service, Backfill: h.backfills, + Sweeper: h.handler, }) require.NoError(t, err) diff --git a/internal/ingest/sweep.go b/internal/ingest/sweep.go new file mode 100644 index 0000000..c187fb0 --- /dev/null +++ b/internal/ingest/sweep.go @@ -0,0 +1,217 @@ +// Origin-verified delete sweep: the cleanup path for deletes the bridge +// missed (a dropped activity, an outage, the pre-fix cross-authority +// authorization bug). Unlike handleDelete there is no delivered activity to +// authorize, so the origin's own word IS the authorization: an object is +// swept only when its origin serves a tombstone (HTTP 410 or an ActivityPub +// Tombstone body — how Lemmy serves deleted content), fetched with the +// redirect authority pinned to the object's own origin so an open redirect +// cannot hand that decision to someone else. A plain 404 is NOT treated as +// deleted: ap.Client also folds 401/403 into it, so 404 means "the origin +// will not show it to us", not "it is gone". A transient outage (5xx, a +// network failure) is not evidence either — it reports OutcomeError and the +// mapping is left alone. FOLLOWUPS.md:39 records the same 410-not-404 rule +// at the bridge's other trust-the-origin site, the inbox's actor +// self-delete confirmation. + +package ingest + +import ( + "context" + "fmt" + "net/http" + "time" + + "tidepool/internal/errors" + "tidepool/internal/materialize" +) + +// SweepOutcome is the verdict for one swept ap_id: exactly one per id, and +// only OutcomeDeleted mutates anything. +type SweepOutcome string + +// Sweep outcomes. +// +// OutcomeActorSkipped covers actor profiles: the Delete(Actor) scrub is +// terminal and deliberate, never swept. OutcomeUnavailable covers the +// ambiguous 404 — ap.Client maps 401/403 there too (secure-mode instances +// hide objects from unauthorized fetchers), so it means "not visible to us", +// which is not grounds for deleting a bridged record. +const ( + OutcomeDeleted SweepOutcome = "deleted" // origin served a tombstone; delete applied + OutcomeAlreadyDeleted SweepOutcome = "already-deleted" // mapping was already soft-deleted + OutcomeUnknown SweepOutcome = "unknown" // no mapping for this ap_id + OutcomeStillLive SweepOutcome = "still-live" // origin still serves the object + OutcomeUnavailable SweepOutcome = "unavailable" // origin 404/401/403s — ambiguous, not applied + OutcomeActorSkipped SweepOutcome = "actor-skipped" // target is an actor profile + OutcomeError SweepOutcome = "error" // lookup, fetch, or delete failed +) + +const ( + // maxSweepBatch caps the ap_ids one request may sweep. Every id costs a + // serial signed fetch of a remote origin, so a large batch holds one + // request open for minutes; chunking keeps each pass short enough to + // complete, observe, and re-run. + maxSweepBatch = 200 + // sweepMutationTimeout bounds the detached tombstone+delete pair for one + // id, so a stuck DB cannot pin a goroutine after the request is gone. + sweepMutationTimeout = 30 * time.Second +) + +// mutated reports whether the outcome changed durable state; only the +// applied delete does. +func (o SweepOutcome) mutated() bool { return o == OutcomeDeleted } + +// failed reports whether the id could not be verified at all (as opposed to +// being verified and deliberately left alone). +func (o SweepOutcome) failed() bool { return o == OutcomeError } + +// DeleteSweeper is the slice of the ingest Handler the admin sweep endpoint +// drives — the seam that keeps the admin API from depending on the whole +// dispatcher, like RepoReemitter for re-emission. +type DeleteSweeper interface { + SweepDeleted(ctx context.Context, apID string) DeleteSweepResult +} + +// DeleteSweepResult reports one swept ap_id. (Unrelated to reconcile.go's +// SweepResult, which reports one follow-list reconciliation pass.) +type DeleteSweepResult struct { + APID string `json:"ap_id"` + Outcome SweepOutcome `json:"outcome"` + Error string `json:"error,omitempty"` +} + +// SweepDeleted looks one ap_id's mapping up and, only when that mapping is +// live and is not an actor profile, re-verifies the object against its +// origin and applies the delete iff the origin serves a tombstone. Unknown, +// already-deleted, and actor ids are reported without any fetch. Idempotent; +// every non-tombstone answer is a no-op with a diagnostic outcome. +func (h *Handler) SweepDeleted(ctx context.Context, apID string) DeleteSweepResult { + res := DeleteSweepResult{APID: apID} + fail := func(err error) DeleteSweepResult { + // Logged here, not just returned: an operator scripting a batch reads + // the response, but the failure must also be greppable in the bridge's + // own log next to the deletes it did apply. + h.logger.Error("sweep: ap_id failed", "ap_id", apID, "error", err) + res.Outcome = OutcomeError + res.Error = err.Error() + return res + } + + mapping, err := h.objects.GetByAPID(ctx, apID) + if errors.IsNotFound(err) { + res.Outcome = OutcomeUnknown + return res + } + if err != nil { + return fail(fmt.Errorf("look up mapping: %w", err)) + } + if mapping.Collection == materialize.CollectionActorProfile || + mapping.Collection == materialize.CollectionCommunityProfile { + res.Outcome = OutcomeActorSkipped + return res + } + if mapping.IsDeleted() { + res.Outcome = OutcomeAlreadyDeleted + return res + } + + // Pinned to the object's own authority: this fetch's answer is what + // authorizes the delete, so a redirect off the origin (an open-redirect + // bug, a compromised origin) must fail it rather than let an attacker + // host serve the 410. + _, err = h.fetcher.FetchObjectSameAuthority(ctx, apID) + switch { + case err == nil: + res.Outcome = OutcomeStillLive + case errors.IsTombstoned(err): + // Detached from the request: a client disconnect between the two + // mutations would otherwise leave the marker recorded and the record + // still live — a half-applied delete nothing retries. (The queue + // detaches its bookkeeping for the same reason; see queue.go.) Marker + // first, as in handleDelete, so a crash between them cannot let a + // re-delivered Create resurrect the object. + mutCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), sweepMutationTimeout) + defer cancel() + // Global scope (""): the 410 came from the object's OWN origin, so + // this marker carries origin authority, not any community's. + if err := h.tombstones.Record(mutCtx, apID, ""); err != nil { + return fail(fmt.Errorf("record tombstone: %w", err)) + } + // Content-only entry: the actor cases were already decided above + // (OutcomeActorSkipped), so the materializer must not re-derive the + // branch from a fresh bridged_actors read and turn a swept post into an + // actor scrub. + if err := h.mat.HandleDeleteRecord(mutCtx, apID); err != nil { + return fail(fmt.Errorf("apply delete: %w", err)) + } + h.logger.Info("sweep: applied origin-verified delete", "ap_id", apID) + res.Outcome = OutcomeDeleted + case errors.IsNotFound(err): + res.Outcome = OutcomeUnavailable + default: + return fail(fmt.Errorf("fetch origin: %w", err)) + } + return res +} + +// handleSweepDeleted serves POST /admin/objects/sweep-deleted: +// {"ap_ids":["https://...", ...]}, at most maxSweepBatch per request. Each +// id is re-verified against its origin and deleted only if the origin serves +// a tombstone; the response reports one outcome per id plus how many ids +// were requested and whether the pass was truncated (the client hung up +// mid-batch) — a short result list otherwise looks like a complete pass. +// Always 200 once the batch starts, including partial success: per-id +// outcomes, not the status, carry the verdict. Safe to re-run: applied +// deletes report already-deleted on the next pass. +func (a *Admin) handleSweepDeleted(w http.ResponseWriter, r *http.Request) { + if a.sweeper == nil { + http.Error(w, "delete sweep not configured", http.StatusNotImplemented) + return + } + var req struct { + APIDs []string `json:"ap_ids"` + } + if err := decodeJSONBody(r, &req); err != nil || len(req.APIDs) == 0 { + http.Error(w, `body must be {"ap_ids":["https://..."]}`, http.StatusBadRequest) + return + } + if len(req.APIDs) > maxSweepBatch { + http.Error(w, fmt.Sprintf("at most %d ap_ids per request; chunk the list", maxSweepBatch), + http.StatusBadRequest) + return + } + + ctx := r.Context() + results := make([]DeleteSweepResult, 0, len(req.APIDs)) + deleted, failed := 0, 0 + for _, apID := range req.APIDs { + if ctx.Err() != nil { + break + } + res := a.sweeper.SweepDeleted(ctx, apID) + switch { + case res.Outcome.mutated(): + deleted++ + case res.Outcome.failed(): + failed++ + } + results = append(results, res) + } + + truncated := len(results) < len(req.APIDs) + summary := []any{"requested", len(req.APIDs), "swept", len(results), + "deleted", deleted, "failed", failed, "truncated", truncated} + if failed > 0 || truncated { + a.logger.Warn("delete sweep finished incomplete", summary...) + } else { + a.logger.Info("delete sweep complete", summary...) + } + writeJSON(w, http.StatusOK, map[string]any{ + "requested": len(req.APIDs), + "swept": len(results), + "deleted": deleted, + "failed": failed, + "truncated": truncated, + "result": results, + }) +} diff --git a/internal/ingest/sweep_test.go b/internal/ingest/sweep_test.go new file mode 100644 index 0000000..7195907 --- /dev/null +++ b/internal/ingest/sweep_test.go @@ -0,0 +1,198 @@ +package ingest + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// sweepResponse mirrors handleSweepDeleted's JSON body. +type sweepResponse struct { + Requested int `json:"requested"` + Swept int `json:"swept"` + Deleted int `json:"deleted"` + Failed int `json:"failed"` + Truncated bool `json:"truncated"` + Result []DeleteSweepResult `json:"result"` +} + +func (h *harness) sweep(apIDs ...string) sweepResponse { + h.t.Helper() + rec := h.adminRequest(http.MethodPost, "/admin/objects/sweep-deleted", + map[string]any{"ap_ids": apIDs}) + require.Equal(h.t, http.StatusOK, rec.Code) + var out sweepResponse + require.NoError(h.t, json.Unmarshal(rec.Body.Bytes(), &out)) + require.Len(h.t, out.Result, len(apIDs)) + // A completed pass must account for every id it was handed; only a + // mid-batch client disconnect may report fewer. + require.Equal(h.t, len(apIDs), out.Requested) + require.Equal(h.t, len(apIDs), out.Swept) + require.False(h.t, out.Truncated) + return out +} + +// assertUntouched: the mapping is still live and no create-after-delete +// marker was recorded — what every non-tombstone sweep outcome must leave. +func (h *harness) assertUntouched(apID string) { + h.t.Helper() + ctx := context.Background() + mapping, err := h.objects.GetByAPID(ctx, apID) + require.NoError(h.t, err) + assert.False(h.t, mapping.IsDeleted(), "the mapping must still be live") + tombstoned, err := h.tombstones.ExistsFor(ctx, apID, "") + require.NoError(h.t, err) + assert.False(h.t, tombstoned, "no tombstone marker may be recorded") +} + +// TestSweepDeletedAppliesOriginTombstone: the missed-delete cleanup path. A +// live object is a no-op; once the origin serves a Tombstone the sweep +// applies the delete (record gone, mapping soft-deleted, marker recorded); +// re-running is idempotent. +func TestSweepDeletedAppliesOriginTombstone(t *testing.T) { + h := newHarness(t) + group := h.subscribeTechnology() + h.serveLemmyWorldContent() + ctx := context.Background() + + require.Equal(t, http.StatusAccepted, + h.deliver(group, loadFixture(t, "announce_create_page_lemmy_world.json"))) + h.drain() + mapping, err := h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + require.False(t, mapping.IsDeleted()) + + // Origin still serves the object: nothing may change. + out := h.sweep(pageID) + assert.Equal(t, OutcomeStillLive, out.Result[0].Outcome) + assert.Zero(t, out.Deleted) + mapping, err = h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + assert.False(t, mapping.IsDeleted(), "a live object must never be swept") + + // The origin now serves Lemmy's deleted-object shape. + h.serveObject(urlPath(t, pageID), map[string]any{"id": pageID, "type": "Tombstone"}) + out = h.sweep(pageID) + assert.Equal(t, OutcomeDeleted, out.Result[0].Outcome) + assert.Equal(t, 1, out.Deleted) + + mapping, err = h.objects.GetByAPID(ctx, pageID) + require.NoError(t, err) + assert.True(t, mapping.IsDeleted(), "the mapping must be soft-deleted") + _, _, err = h.manager.GetRecord(ctx, mapping.DID, mapping.Collection, mapping.RKey) + assert.Error(t, err, "the record must be deleted from the repo") + tombstoned, err := h.tombstones.ExistsFor(ctx, pageID, "") + require.NoError(t, err) + assert.True(t, tombstoned, "the create-after-delete marker must be recorded") + + out = h.sweep(pageID) + assert.Equal(t, OutcomeAlreadyDeleted, out.Result[0].Outcome) + assert.Zero(t, out.Deleted) +} + +// TestSweepDeletedLeavesUnavailableObjectAlone is THE safety rule: a 404 is +// not a delete. ap.Client maps 401/403 to the same not-found error, so a +// secure-mode instance hiding an object we may not fetch is indistinguishable +// from a missing one — neither may cost the user their bridged record. +func TestSweepDeletedLeavesUnavailableObjectAlone(t *testing.T) { + h := newHarness(t) + group := h.subscribeTechnology() + h.serveLemmyWorldContent() + + // Nothing is registered for this post's path, so the fixture mux 404s it. + postID := h.bridgePost(group, "unavailable-404") + + out := h.sweep(postID) + assert.Equal(t, OutcomeUnavailable, out.Result[0].Outcome) + assert.Zero(t, out.Deleted) + assert.Zero(t, out.Failed) + h.assertUntouched(postID) +} + +// TestSweepDeletedReportsFetchError: an origin that answers 5xx has told us +// nothing, so the id is reported failed and left alone — an outage must +// never read as a delete. +func TestSweepDeletedReportsFetchError(t *testing.T) { + h := newHarness(t) + group := h.subscribeTechnology() + h.serveLemmyWorldContent() + + const slug = "origin-outage" + h.mux.HandleFunc("GET /post/"+slug, func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + }) + postID := h.bridgePost(group, slug) + + out := h.sweep(postID) + assert.Equal(t, OutcomeError, out.Result[0].Outcome) + assert.NotEmpty(t, out.Result[0].Error, "a failed id must carry its reason") + assert.Equal(t, 1, out.Failed) + assert.Zero(t, out.Deleted) + h.assertUntouched(postID) +} + +// TestSweepDeletedRejectsCrossAuthorityRedirect: the sweep's fetch IS its +// authorization to delete, so a redirect off the object's own origin (an +// open-redirect bug, a compromised origin) to a host serving a tombstone +// must fail the fetch instead of deleting the record. +func TestSweepDeletedRejectsCrossAuthorityRedirect(t *testing.T) { + h := newHarness(t) + group := h.subscribeTechnology() + h.serveLemmyWorldContent() + + const slug = "redirect-hijack" + h.mux.HandleFunc("GET /post/"+slug, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Location", "https://evil.example/post/attacker-tombstone") + w.WriteHeader(http.StatusFound) + }) + // The attacker host happily serves the tombstone the sweep is looking for. + h.mux.HandleFunc("GET /post/attacker-tombstone", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusGone) + }) + postID := h.bridgePost(group, slug) + + out := h.sweep(postID) + assert.Equal(t, OutcomeError, out.Result[0].Outcome) + assert.Contains(t, out.Result[0].Error, "authority", + "the refusal must name the authority hop, not some later failure") + assert.Zero(t, out.Deleted) + h.assertUntouched(postID) +} + +// TestSweepDeletedRefusals: ids the sweep must never touch — actor profiles +// (the Delete(Actor) scrub is deliberate, never swept) and unmapped ids — +// plus the request-shape refusals (empty list, oversized batch). +func TestSweepDeletedRefusals(t *testing.T) { + h := newHarness(t) + group := h.subscribeTechnology() + h.serveLemmyWorldContent() + + require.Equal(t, http.StatusAccepted, + h.deliver(group, loadFixture(t, "announce_create_page_lemmy_world.json"))) + h.drain() + + out := h.sweep(personID, "https://lemmy.world/post/999999") + assert.Equal(t, OutcomeActorSkipped, out.Result[0].Outcome) + assert.Equal(t, OutcomeUnknown, out.Result[1].Outcome) + assert.Zero(t, out.Deleted) + assert.Equal(t, 2, out.Requested) + assert.Equal(t, 2, out.Swept) + assert.False(t, out.Truncated, "a batch that ran to completion is not truncated") + + rec := h.adminRequest(http.MethodPost, "/admin/objects/sweep-deleted", map[string]any{}) + assert.Equal(t, http.StatusBadRequest, rec.Code, "an empty id list is a request error") + + oversized := make([]string, maxSweepBatch+1) + for i := range oversized { + oversized[i] = fmt.Sprintf("https://lemmy.world/post/%d", i) + } + rec = h.adminRequest(http.MethodPost, "/admin/objects/sweep-deleted", + map[string]any{"ap_ids": oversized}) + assert.Equal(t, http.StatusBadRequest, rec.Code, "the batch cap is a request error") + assert.Contains(t, rec.Body.String(), "chunk", "the refusal must tell the operator what to do") +}