diff --git a/cmd/tidepool/main.go b/cmd/tidepool/main.go index a0b086c..cc3a0fd 100644 --- a/cmd/tidepool/main.go +++ b/cmd/tidepool/main.go @@ -391,6 +391,7 @@ func run(logger *slog.Logger) error { // store every inbound moderation decision is RECORDED in, and production // should not depend on a type assertion to have one. Moderation: store.NewObjectModeration(database), + Bans: store.NewCommunityBans(database), ServiceActorID: serviceActor.ID, Logger: logger, }) @@ -715,14 +716,18 @@ func startConsumer( // the Create{Page} enqueued atomically with it. Wired whenever the consumer // runs, so postv2 events are admitted rather than skipped at debug. engine, err := accept.NewEngine(accept.Options{ - Repos: repoManager, - Enqueuer: enqueuer, - Actors: minter, - Resolver: resolver, - Communities: store.NewCommunities(database), - Objects: store.NewOutboundObjects(database), - Prefs: store.NewFederationPrefs(database), - Admissions: accept.NewAdmissions(database), + Repos: repoManager, + Enqueuer: enqueuer, + Actors: minter, + Resolver: resolver, + Communities: store.NewCommunities(database), + Objects: store.NewOutboundObjects(database), + Prefs: store.NewFederationPrefs(database), + Admissions: accept.NewAdmissions(database), + // Passed explicitly rather than left to NewEngine's default: the ban gate + // is what keeps a banned author's posts out of a community, and + // production should not depend on a type assertion to have one. + Bans: store.NewCommunityBans(database), APActors: store.NewAPActors(database), MaxPerAuthorPerCommunity: cfg.AdmissionMaxPerAuthorPerCommunity, UserOrigin: cfg.APUserOrigin, diff --git a/internal/accept/admissions.go b/internal/accept/admissions.go index 74640cd..7f24b9a 100644 --- a/internal/accept/admissions.go +++ b/internal/accept/admissions.go @@ -267,6 +267,41 @@ func (a *Admissions) CountAccepted(ctx context.Context, authorDID, communityDID, return n, nil } +// ListAccepted returns the at-uris of the posts an author currently has ACCEPTED +// in one community, oldest first — the input to a ban's removeData purge, and +// the reason idx_admissions_author_community leads with (author_did, +// community_did). +// +// This ledger is the ONLY table that records which community admitted a post, +// which is exactly the scope a ban is entitled to act on. The obvious +// alternatives are both wrong: ap_objects and outbound_objects know what was +// materialized and federated but not by whose decision, so either would purge an +// author's writing in every community over one community's ban. +func (a *Admissions) ListAccepted(ctx context.Context, communityDID, authorDID string) ([]string, error) { + rows, err := a.db.QueryContext(ctx, ` + SELECT post_uri FROM admissions + WHERE author_did = $1 AND community_did = $2 AND status = $3 + ORDER BY created_at`, + authorDID, communityDID, StatusAccepted) + if err != nil { + return nil, fmt.Errorf("accept: list accepted for %s in %s: %w", authorDID, communityDID, err) + } + defer func() { _ = rows.Close() }() + + var postURIs []string + for rows.Next() { + var postURI string + if err := rows.Scan(&postURI); err != nil { + return nil, fmt.Errorf("accept: scan accepted for %s in %s: %w", authorDID, communityDID, err) + } + postURIs = append(postURIs, postURI) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("accept: list accepted for %s in %s: %w", authorDID, communityDID, err) + } + return postURIs, nil +} + // DeleteTx removes the ledger row for a (community, post) on an existing // transaction — the author-delete path, which rides the acceptance-delete commit // so the ledger row and the acceptance record go away together. The post no diff --git a/internal/accept/engine.go b/internal/accept/engine.go index 640a729..99e91d1 100644 --- a/internal/accept/engine.go +++ b/internal/accept/engine.go @@ -81,6 +81,15 @@ const ( // stay distinguishable in the ledger, because they are the two branches of // what an edit against a standing removal is allowed to do. DecisionModeratorRemoved = "moderator-removed" + // DecisionAuthorBanned: the community has BANNED this author (task 17c-3), so + // nothing they write enters it until the ban is lifted or lapses. + // + // It is the same string the removal record's code uses — one decision, one + // vocabulary — and it is taken from there rather than re-typed, because the + // two surfaces answer the same operator question from opposite sides: the + // ledger says why the post was refused, the removal record says why an older + // one went. Two literals is how those drift apart. + DecisionAuthorBanned = materialize.RemovalCodeAuthorBanned ) // ErrModeratorRemovalStands reports that an edit was refused because the @@ -148,6 +157,14 @@ type Options struct { Prefs store.FederationPrefs // Admissions is the decision ledger (migration 021). Admissions *Admissions + // Bans reads whether a community has excluded the author (task 17c-3). + // + // OPTIONAL in the wiring sense only: when it is nil, NewEngine takes the ban + // view of Communities, which the postgres communities store provides. It is + // a separate option rather than methods on store.Communities because half + // the bridge holds that interface to resolve follow state, and none of them + // may reach an exclusion. + Bans store.CommunityBans // APActors reads the author's AP actor row for the delivery-paused admission // check (decision 19). OPTIONAL: nil skips the paused check (the seam is not // wired yet — flagged for the paused-rejection lifecycle). @@ -178,6 +195,7 @@ type Engine struct { objects store.OutboundObjects prefs store.FederationPrefs admissions *Admissions + bans store.CommunityBans apActors store.APActors maxPerCommunity int catalog *lexicon.BaseCatalog @@ -227,6 +245,17 @@ func NewEngine(opts Options) (*Engine, error) { if err != nil { return nil, fmt.Errorf("accept: load lexicon catalog: %w", err) } + // The ban store and the communities store are two repositories over two + // tables; only the admission gate reads the first. The default keeps every + // existing call site working — the postgres communities store IS also that + // repository — without putting exclusions on an interface half the bridge + // holds to resolve follow state. + bans := opts.Bans + if bans == nil { + if fromCommunities, ok := opts.Communities.(store.CommunityBans); ok { + bans = fromCommunities + } + } return &Engine{ repos: opts.Repos, enqueuer: opts.Enqueuer, @@ -236,6 +265,7 @@ func NewEngine(opts Options) (*Engine, error) { objects: opts.Objects, prefs: opts.Prefs, admissions: opts.Admissions, + bans: bans, apActors: opts.APActors, maxPerCommunity: opts.MaxPerAuthorPerCommunity, catalog: catalog, @@ -389,7 +419,26 @@ func (e *Engine) decide(ctx context.Context, did string, commit *consume.CommitE return DecisionCommunityNotFollowed, false, nil } - // 4. Opt-out (decision 11): content pushed outward is exactly what an + // 4. Community ban (task 17c-3): this community has excluded this author, so + // nothing they write enters it. It sits here — after the community gate, + // before the author's own preferences — because it is the community's + // decision about its own space, and admitting the post would sign that + // community's name to content from someone it has excluded. + // + // A nil store is "this deployment records no bans", not "nobody is banned": + // production wires it and NewEngine defaults it off the communities store, + // so the nil is only reachable from a caller that passes neither. + if e.bans != nil { + banned, err := e.bans.Standing(ctx, communityDID, did) + if err != nil { + return "", false, fmt.Errorf("accept: read ban on %s in %s: %w", did, communityDID, err) + } + if banned { + return DecisionAuthorBanned, false, nil + } + } + + // 5. Opt-out (decision 11): content pushed outward is exactly what an // opted-out author refused. federating, err := e.mayFederate(ctx, did) if err != nil { @@ -399,7 +448,7 @@ func (e *Engine) decide(ctx context.Context, did string, commit *consume.CommitE return DecisionOptedOut, false, nil } - // 5. Paused (#account, decision 19): delivery is halted while the identity is + // 6. Paused (#account, decision 19): delivery is halted while the identity is // deactivated/suspended/takendown/throttled, so a new post is not admitted. if e.apActors != nil { actor, err := e.apActors.GetByDID(ctx, did) @@ -415,7 +464,7 @@ func (e *Engine) decide(ctx context.Context, did string, commit *consume.CommitE } } - // 6. Title: required and within Lemmy's cap (postv2 title is OPTIONAL in the + // 7. Title: required and within Lemmy's cap (postv2 title is OPTIONAL in the // lexicon, so this is admission policy, not validation). The cap counts RUNES, // not bytes — Lemmy's limit is on grapheme length, so a multibyte title well // under 200 characters must not be rejected for being over 200 bytes. @@ -427,7 +476,7 @@ func (e *Engine) decide(ctx context.Context, did string, commit *consume.CommitE return DecisionTitleTooLong, false, nil } - // 7. Rate cap: one author must not flood a community Tidepool vouches for. + // 8. Rate cap: one author must not flood a community Tidepool vouches for. // Counts the author's currently-accepted posts in this community, excluding // this post so a repin never counts against itself. 0 means unlimited. if e.maxPerCommunity > 0 { diff --git a/internal/ap/vocab.go b/internal/ap/vocab.go index db87b8a..b531c4c 100644 --- a/internal/ap/vocab.go +++ b/internal/ap/vocab.go @@ -63,6 +63,18 @@ const ( // counter reported it working. TypeLock = "Lock" + // TypeBlock is Lemmy's community ban (activities/block/block_user.rs), + // announced by the community as Announce{Block} and lifted with + // Announce{Undo{Block}}. + // + // Its `object` is the BANNED ACTOR and its `target` is the community the ban + // applies to — neither is a payload. Like Lock it must stay OUT of + // echo.carriesPayload, and here the consequence is sharper: the banned actor + // of a native author IS one of our own personas by definition, so descending + // would classify every inbound ban as our own echo and disable community + // bans entirely while the drop counter reported success. + TypeBlock = "Block" + TypeTombstone = "Tombstone" TypeImage = "Image" TypeLink = "Link" @@ -134,7 +146,18 @@ type Object struct { // of language objects on Group actors; Languages accepts both. Language Languages `json:"language,omitempty"` + // Expires is a Block's ban expiry. It is LOAD-BEARING and not decoration: + // Lemmy sends NO Undo when a temporary ban lapses — the ban simply stops + // applying on their side — so an implementation that drops this column turns + // every timed ban into a permanent one with no activity that can ever clear + // it. + Expires *Time `json:"expires,omitempty"` + // Lemmy extensions. + // RemoveData is Block's purge flag: the moderator also removed that author's + // content in the community they were banned from. A *bool because absent and + // false are the same decision here but only one of them is a statement. + RemoveData *bool `json:"removeData,omitempty"` Sensitive *bool `json:"sensitive,omitempty"` CommentsEnabled *bool `json:"commentsEnabled,omitempty"` PostingRestrictedToMods *bool `json:"postingRestrictedToMods,omitempty"` diff --git a/internal/db/migrations/027_community_bans.sql b/internal/db/migrations/027_community_bans.sql new file mode 100644 index 0000000..316669d --- /dev/null +++ b/internal/db/migrations/027_community_bans.sql @@ -0,0 +1,50 @@ +-- +goose Up +-- Task 17c-3: a community's ban of one author. +-- +-- A ban is an INTERSECTION, and the schema is that intersection: this author, in +-- this community. The two facts the codebase already had are each one dimension +-- short — CancelForActor stops an author everywhere (one community bans them and +-- every other community they write to stops receiving their posts), and +-- CancelForCommunity stops everyone (one user is banned and the community goes +-- dark) — so the primary key is the pair, and every read carries both halves. +CREATE TABLE community_bans ( + community_did TEXT NOT NULL, -- the bridged community that issued the ban + subject_did TEXT NOT NULL, -- the native author it excludes + -- community_ap_id is DENORMALIZED, and it is not tidiness: the two readers + -- hold different handles for the same community. The admission gate has the + -- community DID (it is deciding about a repo it writes into); the delivery + -- queue has only OrderingKey, which IS the AP group id. A join the delivery + -- side cannot make is a scope it cannot apply — and it can never drift, + -- because the DID↔group-id mapping is immutable 1:1 (store.Communities + -- rejects an upsert that moves either). + community_ap_id TEXT NOT NULL, + banned_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- expires_at is MANDATORY to honour, not optional to store. Lemmy's + -- BlockUser carries `expires` for a temporary ban and then sends NOTHING + -- when it lapses — no Undo, no second activity — because it expires locally + -- on their side. Ignore this column and a moderator who chose three days has + -- excluded that author forever, with no message that could ever clear it. + -- NULL means permanent. EVERY read is: + -- expires_at IS NULL OR expires_at > now() + expires_at TIMESTAMPTZ, + reason TEXT NOT NULL DEFAULT '', + -- remove_data records what the moderator asked for, not what we did with it. + -- The content removal happens once, when the ban lands; keeping the flag is + -- what lets an operator answer "was their content purged too?" afterwards — + -- and it is deliberately NOT read by the Undo, because Lemmy models + -- restoration as a SEPARATE restore_data flag and an unban that quietly + -- republished removed posts would reverse a decision nobody reversed. + remove_data BOOLEAN NOT NULL DEFAULT false, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_did, subject_did) +); + +-- No secondary index. Both readers arrive holding BOTH halves of the primary +-- key — the admission gate asks about one author in one community, and the ban +-- write scopes its delivery cancellation by (subject_did, community_ap_id) it +-- already has. Nothing lists bans by community or by author yet; when the admin +-- surface does, the index it wants is (subject_did) for "where is this user +-- banned?", which the PK's leading column cannot serve. + +-- +goose Down +DROP TABLE IF EXISTS community_bans; diff --git a/internal/ingest/consent.go b/internal/ingest/consent.go index 9523e66..9acb220 100644 --- a/internal/ingest/consent.go +++ b/internal/ingest/consent.go @@ -324,6 +324,11 @@ func (h *Handler) handleUndo(ctx context.Context, undo *ap.Object, signer string // nobody can reach: no later activity clears it, because this is the // only one Lemmy will ever send about it. return h.handleLock(ctx, inner, announcer, false) + case ap.TypeBlock: + // The unban, with the Block carried INLINE. It lifts the exclusion and + // nothing else: content removed under removeData stays removed, because + // Lemmy models restoration as a separate restore_data flag. + return h.handleBlock(ctx, inner, announcer, false) case ap.TypeFollow: // A remote undoing a follow of us — the bridge has no followers in // v1 (read-only), nothing to do. diff --git a/internal/ingest/handler.go b/internal/ingest/handler.go index 51ff8c9..5f606c0 100644 --- a/internal/ingest/handler.go +++ b/internal/ingest/handler.go @@ -99,6 +99,10 @@ type HandlerOptions struct { Records RecordGetter Votes VoteAggregator Backfill Backfiller + // Bans is the community-ban store an announced Block is recorded in (task + // 17c-3). Optional in the same wiring sense as Moderation: nil takes the ban + // view of Communities, which the postgres communities store provides. + Bans store.CommunityBans // Moderation is the bridge-owned moderation state announced Locks and // native-comment removals are recorded in (task 17c-2). Optional ONLY in the // wiring sense: when it is nil, NewHandler takes the moderation view of @@ -130,6 +134,8 @@ type Handler struct { votes VoteAggregator backfill Backfiller moderation store.ObjectModeration + bans store.CommunityBans + authorMod AuthorModerator classifier EchoClassifier echoLog *ratelimit.Sampler serviceID string @@ -192,6 +198,21 @@ func NewHandler(opts HandlerOptions) (*Handler, error) { moderation = fromObjects } } + // Same escape, same reason: a ban is a different table from the communities + // row, and the follow-state readers that hold store.Communities must not + // gain the power to exclude anyone. + bans := opts.Bans + if bans == nil { + if fromCommunities, ok := opts.Communities.(store.CommunityBans); ok { + bans = fromCommunities + } + } + // The materializer is the only thing that can act on removeData — it owns + // the one-commit removal AND holds the admissions ledger that says which + // posts this community admitted. It is read off the SAME value the + // dispatcher already drives rather than a second option, because a + // deployment cannot coherently have one and not the other. + authorMod, _ := opts.Materializer.(AuthorModerator) return &Handler{ mat: opts.Materializer, fetcher: opts.Fetcher, @@ -203,6 +224,8 @@ func NewHandler(opts HandlerOptions) (*Handler, error) { votes: opts.Votes, backfill: opts.Backfill, moderation: moderation, + bans: bans, + authorMod: authorMod, classifier: opts.Echo, echoLog: ratelimit.NewSampler(echoDropLogInterval), serviceID: opts.ServiceActorID, @@ -262,6 +285,11 @@ func (h *Handler) Process(ctx context.Context, event *store.InboxEvent) error { return h.handleAccept(ctx, activity, signer) case ap.TypeReject: return h.handleReject(ctx, activity, signer) + case ap.TypeBlock: + // A Block delivered DIRECTLY to the banned user's inbox. Lemmy sends one + // of these alongside every announced ban, and it is the copy we cannot + // authorize: see ignoreDirectBlock. + return h.ignoreDirectBlock(activity, signer) case ap.TypeLike, ap.TypeDislike: // Bare votes (rare; Lemmy normally announces them via the group). // The inbox already bound this top-level activity's actor to the @@ -353,6 +381,11 @@ func (h *Handler) handleAnnounce(ctx context.Context, announce *ap.Object, signe // A community closing one of its own threads. The Undo arrives on the // TypeUndo branch above and lands in the same handler with locked=false. return h.handleLock(ctx, inner, community, true) + case ap.TypeBlock: + // A community banning a native author. This is the AUTHORITATIVE copy — + // the direct one, delivered to the banned user's inbox, is signed by the + // moderator's Person and cannot satisfy decision 18 by construction. + return h.handleBlock(ctx, inner, community, true) default: // Add, Remove, Block, ... — moderation activities the bridge does not // translate yet. Remove in particular is NOT content removal in Lemmy diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index e6a3835..7d0a357 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -264,7 +264,15 @@ func newHarness(t *testing.T) *harness { // so a lock left by one run refuses the next run's comment before the // test that locks it has run — green first, red second, which a single CI // run never sees. - "object_moderation") + "object_moderation", + // Bans (migration 027) are the same trap one turn worse. The moderation + // fixtures' community and author DIDs are package constants, so a + // standing ban refuses the NEXT test's post at admission — and because + // nothing here truncated it, a row survived across `go test` + // invocations, poisoning tests that run BEFORE the ban tests as well as + // after. The failure reads as "my post was not accepted", which names + // neither bans nor the test that left one. + "community_bans") custodian, err := identity.NewCustodian(testKEK) require.NoError(t, err) @@ -679,6 +687,57 @@ func (h *harness) subscribeTechnology() *remoteActor { return group } +// subscribeCommunityURL subscribes to a SECOND community through the real admin +// path — resolve, mint, Follow, Accept — and returns its signing handle. +// +// It takes the AP URL rather than a !name@instance handle because +// resolveCommunity passes URLs straight through: the harness serves ONE +// WebFinger document, so a handle-based second subscribe would have to overwrite +// the first community's and the two would race for the same path. +// +// The Group document advertises the shared inbox the harness captures, exactly +// as the lemmy.world fixture does. Without it the bridge POSTs the Follow to a +// per-community inbox nothing answers, and the subscribe fails as a bad gateway +// — a fixture that looks like a bug in follow delivery. +func (h *harness) subscribeCommunityURL(apGroupID, username string) *remoteActor { + h.t.Helper() + group := h.newRemoteActor(apGroupID, map[string]any{ + "type": "Group", + "id": apGroupID, + "preferredUsername": username, + "inbox": apGroupID + "/inbox", + "endpoints": map[string]any{"sharedInbox": "https://lemmy.world/inbox"}, + "published": "2024-01-01T00:00:00.000000Z", + }) + + rec := h.adminRequest(http.MethodPost, "/admin/communities", + map[string]any{"community": apGroupID}) + require.Equal(h.t, http.StatusAccepted, rec.Code, rec.Body.String()) + + h.mu.Lock() + require.NotEmpty(h.t, h.inboxLog, "subscribe must deliver a Follow") + followRaw := h.inboxLog[len(h.inboxLog)-1] + h.mu.Unlock() + follow, err := ap.ParseObject(followRaw) + require.NoError(h.t, err) + require.Equal(h.t, apGroupID, follow.Object.ID, "the Follow must name THIS community") + + status := h.deliver(group, map[string]any{ + "id": apGroupID + "/activities/accept/follow-1", + "type": "Accept", + "actor": apGroupID, + "object": map[string]any{"id": follow.ID, "type": "Follow", "actor": h.service.ID, "object": apGroupID}, + }) + require.Equal(h.t, http.StatusAccepted, status) + h.drain() + + community, err := h.communities.GetByAPGroupID(context.Background(), apGroupID) + require.NoError(h.t, err) + require.Equal(h.t, store.FollowStateAccepted, community.FollowState) + require.NotEmpty(h.t, community.DID, "a subscribed community holds a minted repo") + return group +} + // firehoseOps flattens all firehose event op paths. func (h *harness) firehoseOps() []string { h.t.Helper() diff --git a/internal/ingest/moderation.go b/internal/ingest/moderation.go index 0e4fef5..190023d 100644 --- a/internal/ingest/moderation.go +++ b/internal/ingest/moderation.go @@ -2,14 +2,256 @@ package ingest import ( "context" + "expvar" "fmt" "tidepool/internal/ap" + "tidepool/internal/echo" "tidepool/internal/errors" "tidepool/internal/materialize" "tidepool/internal/store" ) +// AuthorModerator removes every post one author has ACCEPTED in one community — +// a ban's `removeData: true`. *materialize.Materializer implements it. +// +// It is a SEPARATE interface, obtained by type assertion on the Materializer the +// dispatcher already drives, rather than a method on that interface: the purge +// needs the admissions ledger, which only the materializer holds, and widening +// the interface would oblige every caller that constructs a dispatcher for +// unrelated reasons to implement a moderation transition it never invokes. +type AuthorModerator interface { + // RemoveAuthorPosts returns how many posts were removed. code is the + // removal's own machine-readable reason (author-banned here, never + // moderator-discretion: this content was not judged, its author was). + RemoveAuthorPosts(ctx context.Context, communityDID, authorDID, code, reason string) (int, error) +} + +// The ban counters. Each names a DECIDED non-action, and they are separate +// because the three refusals are different findings an operator has to tell +// apart — a ban we ignored on purpose, a ban aimed at a scope we do not model, +// and a ban for somebody who is not our user — where a single number would read +// as "bans are not arriving". +var ( + // BlockDirectIgnored counts Blocks delivered DIRECTLY to the banned user's + // inbox rather than announced by the community. Lemmy sends both; only the + // announced one can be authorized (see ignoreDirectBlock). + BlockDirectIgnored = expvar.NewInt("tidepool_block_direct_ignored") + // BlockUnscopedTarget counts announced Blocks whose `target` is not a + // community this bridge follows — an instance-wide (Site actor) ban, which + // this scope does not model. + BlockUnscopedTarget = expvar.NewInt("tidepool_block_unscoped_target") + // BlockForeignSubject counts announced Blocks naming somebody who is not one + // of our personas. A Lemmy user banned from a Lemmy community is entirely + // their instance's business; we hold no state that could apply it. + BlockForeignSubject = expvar.NewInt("tidepool_block_foreign_subject") +) + +// ignoreDirectBlock is the DECIDED non-action at the other door. +// +// Lemmy sends a ban twice: announced through the community, and delivered +// directly to the banned user's inbox. They are not redundant copies — they are +// signed by different actors, and only one of them can be authorized. +// BlockUser's actor is the MODERATOR's Person, so on this path decision 18's +// conjunction (the signer IS the community that owns the target) CANNOT pass by +// construction: no implementation turns a person into a group. The announced +// copy is the authoritative one, and we follow every bridged community +// (decision 15), so nothing is lost by refusing this one. +// +// It is COUNTED rather than dropped at debug because a ban that arrives only by +// the path we ignore looks exactly like a ban that never arrived — and the day +// the announced path breaks, this counter is the only thing that tells those +// apart. +// +// The refusal holds even when the activity CLAIMS actor = the Group: the inbox +// binds the VERIFIED signer and never the claim (SEC-1), so a Person-signed +// Block is a Person-signed Block whatever it says about itself. +func (h *Handler) ignoreDirectBlock(block *ap.Object, signer string) error { + BlockDirectIgnored.Add(1) + h.logger.Info("ignoring a directly delivered Block", + "activity", block.ID, "signer", signer, "target", refID(block.Target)) + return skip(block.ID, + "a directly delivered Block is signed by the moderator's Person, which can never be "+ + "the community that owns the ban: the community's own announced copy is the "+ + "authoritative one and is the only path that records it") +} + +// handleBlock applies an announced Block — a community banning a native author — +// and, with banned=false, the Undo that lifts it. +// +// A BAN IS AN INTERSECTION: this author, in this community. Both halves are +// recorded, because both readers need one each — the admission gate holds a +// community DID, the delivery queue holds only an ordering key — and every +// consequence below is scoped by the pair. Cancelling by author alone would +// unpublish them in every community they write to; by community alone would take +// the whole community dark over one user. +// +// AUTHORIZATION is decision 18's conjunction, arriving in its simplest form. For +// content verbs the target's community is read off a mapping; a Block names the +// community DIRECTLY in `target`, so the two conjuncts — the signer IS the +// community, and the ban is FOR that community — collapse into one comparison +// against the verified announcer. A target that names a community we follow but +// is not the announcer is one moderator team excluding somebody from another's +// space; a target that names no community we follow is an instance-wide ban this +// scope does not model. They are different findings and get different reasons. +func (h *Handler) handleBlock(ctx context.Context, block *ap.Object, announcer *store.Community, banned bool) error { + if announcer == nil { + // Unreachable from the dispatch (a bare Block is taken by + // ignoreDirectBlock before it can get here), and refused anyway: without + // a verified community there is nothing to authorize against. + return skip(block.ID, "a Block with no announcing community authorizes nothing") + } + target := refID(block.Target) + if target == "" { + return errors.NewValidationError("block", "block names no target community") + } + if target != announcer.APGroupID { + return h.refuseBlockTarget(ctx, block, target, announcer) + } + + subjectAPID := refID(block.Object) + if subjectAPID == "" { + return errors.NewValidationError("block", "block names no subject actor") + } + // WHOSE ban is this? The subject is an AP actor id, and the only ids we can + // act on are our own personas — the classifier answers that by ENTITY + // EXISTENCE against the serving surface and hands back the DID, which is + // exactly what echo.Identity carries the DID for. Parsing the id's path + // would answer the same question from the shape of a URL a peer chose. + identity, err := h.classifier.Classify(ctx, &ap.Object{ID: subjectAPID}) + if err != nil { + return fmt.Errorf("ingest: identify ban subject %s: %w", subjectAPID, err) + } + if identity.Class != echo.ClassLocalActor || identity.DID == "" { + BlockForeignSubject.Add(1) + return skip(block.ID, + "announced Block names an actor that is not one of our personas: a Lemmy user's ban "+ + "from a Lemmy community is enforced entirely on their instance, and we hold no "+ + "state that could apply it") + } + + if !banned { + return h.liftBan(ctx, block, announcer, identity.DID) + } + return h.applyBan(ctx, block, announcer, identity.DID) +} + +// refuseBlockTarget decides an announced Block whose target is not the +// announcing community, and says WHICH of the two it is. +// +// The distinction cannot be drawn from the URL: Lemmy's Site actor is the +// instance apex, a prefix of every id on that host, so any substring test is +// vacuous. It is drawn from state instead — is this target a community we +// follow? — which is the same question every other announced verb answers. +func (h *Handler) refuseBlockTarget(ctx context.Context, block *ap.Object, target string, announcer *store.Community) error { + _, err := h.communities.GetByAPGroupID(ctx, target) + if errors.IsNotFound(err) { + // An instance-wide ban (target = the Site actor) or a community we do + // not federate. Either way the scope is not one we model: recording it + // against the announcing community would UNDERSTATE it — the author is + // excluded from every community on that instance and we would enforce it + // in one — and dropping it silently leaves them posting into that + // instance collecting 403s until their deliveries poison, with nothing + // naming the cause. + BlockUnscopedTarget.Add(1) + h.logger.Warn("announced Block targets a scope this bridge does not model", + "activity", block.ID, "target", target, "announcer", announcer.APGroupID) + return skip(block.ID, + "announced Block targets "+target+", which is not a community this bridge follows: "+ + "an instance-wide ban is a scope Tidepool does not model, so it is recorded nowhere "+ + "and the author keeps posting into that instance") + } + if err != nil { + return fmt.Errorf("ingest: resolve block target %s: %w", target, err) + } + return skip(block.ID, fmt.Sprintf( + "announced Block targets community %s but was announced by %s: a ban is a community's "+ + "ruling about its OWN space", target, announcer.APGroupID)) +} + +// applyBan records the exclusion and acts on everything it implies. +func (h *Handler) applyBan(ctx context.Context, block *ap.Object, announcer *store.Community, subjectDID string) error { + if h.bans == nil { + // Loud and retryable, never a skip: a ban we cannot store is one that + // stops nothing from the author's next post onward, and marking the + // event processed would leave the community believing we honoured it. + return fmt.Errorf("ingest: no community-ban store is wired, so this ban cannot be recorded") + } + ban := store.CommunityBan{ + CommunityDID: announcer.DID, + SubjectDID: subjectDID, + CommunityAPID: announcer.APGroupID, + RemoveData: block.RemoveData != nil && *block.RemoveData, + } + // The expiry is carried through EXACTLY as sent. Lemmy sends no activity when + // a timed ban lapses — it simply stops applying there — so dropping this + // makes a three-day ban permanent with nothing that could ever clear it. + if block.Expires.OK() { + expires := block.Expires.Time + ban.ExpiresAt = &expires + } + + // The row and the cancellation of already-queued work commit together (see + // store.CommunityBans.Ban): the first stops the author's next post, the + // second stops the ones the queue is holding. + cancelled, err := h.bans.Ban(ctx, ban) + if err != nil { + return fmt.Errorf("ingest: record ban on %s in %s: %w", subjectDID, announcer.APGroupID, err) + } + h.logger.Info("community banned a native author", + "community", announcer.APGroupID, "subject_did", subjectDID, + "expires", ban.ExpiresAt, "remove_data", ban.RemoveData, + "cancelled_deliveries", cancelled, "activity", block.ID) + + if !ban.RemoveData { + return nil + } + if h.authorMod == nil { + return fmt.Errorf( + "ingest: no author-moderation surface is wired, so removeData for %s in %s cannot be applied", + subjectDID, announcer.APGroupID) + } + // Their content in THIS community, from the ledger that knows which posts + // this community admitted. Nothing is enqueued and nothing can be: the + // removal rides ApplyOps, which takes no side effect — and Lemmy has already + // removed this content, so an outbound Delete would be aimed at the + // moderators who just acted. + removed, err := h.authorMod.RemoveAuthorPosts(ctx, announcer.DID, subjectDID, + materialize.RemovalCodeAuthorBanned, block.Summary) + if err != nil { + return fmt.Errorf("ingest: remove %s's posts from %s: %w", subjectDID, announcer.APGroupID, err) + } + h.logger.Info("ban carried removeData; the author's posts in this community were removed", + "community", announcer.APGroupID, "subject_did", subjectDID, "removed", removed) + return nil +} + +// liftBan is Undo{Block}: the exclusion goes, and NOTHING ELSE does. +// +// Content removed under removeData STAYS REMOVED. Lemmy models restoration as a +// separate restore_data flag, so republishing here would reverse a decision +// nobody reversed and push the author's posts back at the community that removed +// them — the same harm as reversing a removal on an author's edit, arriving by +// another door. +func (h *Handler) liftBan(ctx context.Context, block *ap.Object, announcer *store.Community, subjectDID string) error { + if h.bans == nil { + return fmt.Errorf("ingest: no community-ban store is wired, so this ban cannot be lifted") + } + lifted, err := h.bans.Lift(ctx, announcer.DID, subjectDID) + if err != nil { + return fmt.Errorf("ingest: lift ban on %s in %s: %w", subjectDID, announcer.APGroupID, err) + } + if !lifted { + return skip(block.ID, fmt.Sprintf( + "announced Undo{Block} for %s in %s, which held no standing ban: nothing to lift "+ + "(a re-delivered undo, or one for a ban that lapsed on its own)", + subjectDID, announcer.APGroupID)) + } + h.logger.Info("community lifted a native author's ban", + "community", announcer.APGroupID, "subject_did", subjectDID, "activity", block.ID) + return nil +} + // moderationState is the bridge-owned moderation store, or an error naming the // gap. Every moderation path asks for it before deciding anything. // diff --git a/internal/ingest/moderation_ban_test.go b/internal/ingest/moderation_ban_test.go new file mode 100644 index 0000000..b0d1975 --- /dev/null +++ b/internal/ingest/moderation_ban_test.go @@ -0,0 +1,781 @@ +package ingest + +import ( + "context" + "database/sql" + "expvar" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "tidepool/internal/accept" + "tidepool/internal/ap" + "tidepool/internal/errors" + "tidepool/internal/materialize" + "tidepool/internal/testutil" +) + +// TASK 17c-3 — A COMMUNITY BAN IS THREE THINGS AT ONCE. +// +// A ban is not one fact, it is an intersection: THIS author, in THIS community. +// Everything that makes it hard follows from that, and so does every way of +// getting it wrong, because the two obvious implementations are already in the +// codebase and both are one dimension short: +// +// CancelForActor(did) — cancels the author EVERYWHERE. One community +// bans them; every other community they write +// to stops receiving their posts. +// CancelForCommunity(key) — cancels EVERYONE in that community. One user +// is banned; the community goes dark. +// +// A fixture with one community passes the first. A fixture with one actor +// passes the second. Only a world with TWO CO-HOSTED COMMUNITIES and TWO NATIVE +// ACTORS can tell any of them apart — and co-hosted matters twice over, because +// communities on one Lemmy instance SHARE AN INBOX, so nothing about the +// delivery address distinguishes A's traffic from B's. The scope has to come +// from the ordering key, which is the community's own AP id. +// +// This is also why community B is now minted for real: an acceptance is a record +// in B's own repo, and "still accepted in B" cannot be asserted by a community +// that has no repo to accept into. +const ( + mbBlockActivity = "https://lemmy.world/activities/announce/block/mb-ban" + mbUnblockActivity = "https://lemmy.world/activities/announce/undo/mb-ban" + + // The banned author's posts: one to A, one to B, one to A after the ban, + // and one to A after the ban is lifted. + mbPostInARKey = "3lzmbpost00001" + mbPostInBRKey = "3lzmbpost00002" + mbPostAfterBanRKey = "3lzmbpost00003" + mbPostAfterUndoRKey = "3lzmbpost00004" + // The OTHER actor's post to A: the same community, a different author. + mbOtherPostRKey = "3lzmbpost00005" +) + +func mbPostATURI(did, rkey string) string { + return "at://" + did + "/" + materialize.CollectionPostV2 + "/" + rkey +} + +// TestABanIsScopedToOneAuthorInOneCommunity is the OUTER CONTRACT for 17c-3. +// +// GIVEN a native author with accepted posts in community A and in community B, +// and a second author with an accepted post in A, WHEN A announces a Block +// naming the first author, THEN the ban is recorded against A, that author's +// pending deliveries TO A are cancelled while their deliveries to B and the +// other author's deliveries to A are untouched, their next post to A is REJECTED +// at admission as author-banned while their post to B is still accepted — and +// WHEN A announces Undo{Block}, admission to A resumes. +func TestABanIsScopedToOneAuthorInOneCommunity(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + communityBAPID := mtCommunityBAPID + + // --- GIVEN: three accepted posts across two communities and two authors. + // Every one of them is pending delivery — nothing runs the worker here, + // which is what makes "cancelled" a visible state change rather than a + // race with a delivery that already went out. + admitPost(t, world, mtAuthorDID, mbPostInARKey, world.communityADID, "3lzmbrev00001", 1_775_000_001_000_001) + admitPost(t, world, mtAuthorDID, mbPostInBRKey, world.communityBDID, "3lzmbrev00002", 1_775_000_001_000_002) + admitPost(t, world, mtCommenterDID, mbOtherPostRKey, world.communityADID, "3lzmbrev00003", 1_775_000_001_000_003) + + requireEveryDelivery(t, h.db, mtAuthorDID, groupID, "pending", + "precondition: the banned-to-be author has pending work for A") + requireEveryDelivery(t, h.db, mtAuthorDID, communityBAPID, "pending", + "precondition: and pending work for B") + requireEveryDelivery(t, h.db, mtCommenterDID, groupID, "pending", + "precondition: and the OTHER author has pending work for A") + + activitiesBefore := rowCount(t, h.db, "outbound_activities") + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + dropsBefore := dropSnapshot() + + // --- WHEN: community A bans the author. Lemmy's shape: the inner Block is + // the MODERATOR's, targeted at the community, announced by the group. + h.announceBlock(world.groupA, mbBlockActivity, mtAuthorDID, groupID, nil) + + // The echo classifier must not have taken it: a Block's `object` is the + // banned actor, and for a native author that actor is OURS by definition. + // Adding Block to carriesPayload would drop every ban the bridge receives + // and count each one as a suppression. + assert.Equal(t, dropsBefore, dropSnapshot(), + "an announced Block naming our own persona is GENUINE remote traffic: the id it "+ + "names is ours precisely because the community is moderating our user") + + // --- THEN: the cancellation is the INTERSECTION, not either axis alone. + requireEveryDelivery(t, h.db, mtAuthorDID, groupID, "cancelled", + "the banned author's PENDING work for A is cancelled: Lemmy rejects a banned user's "+ + "posts, so every one of these is a delivery that fails, retries and poisons for a "+ + "reason nothing in the queue names") + requireEveryDelivery(t, h.db, mtAuthorDID, communityBAPID, "pending", + "but their work for B is UNTOUCHED: one community's moderators do not decide where "+ + "an author may speak — cancelling by actor alone silently unpublishes them across "+ + "every community they belong to") + requireEveryDelivery(t, h.db, mtCommenterDID, groupID, "pending", + "and the other author's work for A is UNTOUCHED: cancelling by community alone takes "+ + "the whole community dark over one user's ban") + + assert.Equal(t, activitiesBefore, rowCount(t, h.db, "outbound_activities"), + "the ban itself enqueues NOTHING: an inbound moderation action is the community "+ + "telling US what it did, and echoing it back is an activity aimed at the "+ + "moderators who sent it") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), + "and cancels rather than deletes: the row is the evidence, and a delivery that "+ + "vanishes leaves an operator no way to see why the author's post never arrived") + + // --- AND: the ban is recorded, bound to A — the durable half, which is what + // stops the SECOND post rather than the ones already queued. + ban, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + require.True(t, found, + "the ban must be RECORDED: it is the only thing standing between a banned author "+ + "and their next post, and Lemmy will not re-send it — a ban the bridge does not "+ + "hold is one that stops nothing from the second post onward") + assert.Equal(t, groupID, ban.communityAPID, + "with the community's AP id denormalized: the delivery side holds an ordering key, "+ + "not a DID, and a join it cannot make is a scope it cannot apply") + assert.False(t, ban.expires.Valid, + "and no expiry, because this Block named none — a permanent ban") + + // --- AND: admission refuses their next post to A, and only to A. + admitPost(t, world, mtAuthorDID, mbPostAfterBanRKey, world.communityADID, "3lzmbrev00004", 1_775_000_001_000_004) + bannedURI := mbPostATURI(mtAuthorDID, mbPostAfterBanRKey) + status, code := admissionFor(t, h.db, world.communityADID, bannedURI) + assert.Equal(t, accept.StatusRejected, status, + "a banned author's post is REJECTED at admission: accepting it would sign the "+ + "community's name to content from someone that community has excluded") + assert.Equal(t, "author-banned", code, + "with the reason a reader can act on — the same vocabulary the removal lexicon "+ + "already spells, so post.getStatus and the admin surface answer 'why' without a "+ + "second dictionary") + _, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, testDigestRKey(bannedURI)) + assert.True(t, errors.IsNotFound(err), + "and no acceptance is written for it (err=%v)", err) + + // The same author, the same moment, the OTHER community. + admitPost(t, world, mtAuthorDID, mbPostInBRKey+"b", world.communityBDID, "3lzmbrev00005", 1_775_000_001_000_005) + _, _, err = h.manager.GetRecord(ctx, world.communityBDID, materialize.CollectionAcceptance, + testDigestRKey(mbPostATURI(mtAuthorDID, mbPostInBRKey+"b"))) + assert.NoError(t, err, + "while B still accepts them: a ban is a community's decision about its own space, "+ + "and one that follows the author off it is a site ban nobody issued") + + // --- WHEN: the moderators lift it. + h.announceUndoBlock(world.groupA, mbUnblockActivity, mbBlockActivity+"/block", + mtAuthorDID, groupID) + + _, stillBanned := banFor(t, h.db, world.communityADID, mtAuthorDID) + assert.False(t, stillBanned, + "Undo{Block} clears the ban: Lemmy sends this activity exactly once, so a ban that "+ + "survives it can never be lifted by anything") + + admitPost(t, world, mtAuthorDID, mbPostAfterUndoRKey, world.communityADID, "3lzmbrev00006", 1_775_000_001_000_006) + restoredURI := mbPostATURI(mtAuthorDID, mbPostAfterUndoRKey) + status, _ = admissionFor(t, h.db, world.communityADID, restoredURI) + assert.Equal(t, accept.StatusAccepted, status, + "and admission resumes: an unbanned author posting again is the ordinary case, and "+ + "a ban that outlives its Undo is indistinguishable to the author from being "+ + "silently shadowbanned") + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, testDigestRKey(restoredURI)) + assert.NoError(t, err, "with the acceptance to prove it (err=%v)", err) +} + +// admitPost drives one native post into one community through the real +// consumer, engine and enqueuer. It asserts only that the EVENT was handled — +// whether the post was accepted or rejected is what the caller is testing. +func admitPost(t *testing.T, world moderationWorld, did, rkey, communityDID, rev string, timeUS int64) { + t.Helper() + require.NoError(t, world.dispatcher.HandleEvent(context.Background(), + mtPostEventBy(t, did, rkey, communityDID, "create", rev, mtPostCID, timeUS))) +} + +// announceBlock delivers Lemmy's ban shape: Announce{Block} from the community, +// whose INNER Block is attributed to the acting MODERATOR (a Person, never the +// Group — which is why only the ANNOUNCED path can satisfy decision 18), names +// the banned actor as its object, and TARGETS the community the ban applies to. +func (h *harness) announceBlock(group *remoteActor, activityID, subjectDID, target string, extra map[string]any) { + h.t.Helper() + require.Equal(h.t, http.StatusAccepted, h.deliver(group, map[string]any{ + "id": activityID, + "type": "Announce", + "actor": group.id, + "audience": group.id, + "cc": []any{group.id + "/followers"}, + "object": blockActivity(group, activityID+"/block", subjectDID, target, extra), + })) + h.drain() +} + +// announceUndoBlock delivers the unban shape: Announce{Undo{Block}} with the +// Block carried INLINE. +func (h *harness) announceUndoBlock(group *remoteActor, activityID, blockActivityID, subjectDID, target string) { + h.t.Helper() + require.Equal(h.t, http.StatusAccepted, h.deliver(group, map[string]any{ + "id": activityID, + "type": "Announce", + "actor": group.id, + "audience": group.id, + "cc": []any{group.id + "/followers"}, + "object": map[string]any{ + "id": activityID + "/undo", + "type": "Undo", + "actor": modActorID, + "audience": group.id, + "cc": []any{group.id}, + "object": blockActivity(group, blockActivityID, subjectDID, target, nil), + }, + })) + h.drain() +} + +func blockActivity(group *remoteActor, activityID, subjectDID, target string, extra map[string]any) map[string]any { + block := map[string]any{ + "id": activityID, + "type": "Block", + "actor": modActorID, + "object": mtUserOrigin + "/ap/actor/" + subjectDID, + "target": target, + "audience": group.id, + "to": []any{ap.PublicAudience}, + "cc": []any{group.id}, + } + for key, value := range extra { + block[key] = value + } + return block +} + +// communityBan is the bridge's record of one (community, author) ban. +type communityBan struct { + communityAPID string + expires sql.NullTime +} + +func banFor(t *testing.T, db *sql.DB, communityDID, subjectDID string) (communityBan, bool) { + t.Helper() + if !tableExists(t, db, "community_bans") { + // A table that does not exist holds no bans, and saying so is not a + // tolerance: every assertion in this file that REQUIRES a ban fails + // loudly right here, so a table that is missing — or landed under + // another name — is caught by the positive tests rather than hidden by + // the negative ones. What it buys is that each test fails on ITS OWN + // behaviour instead of six tests reporting one missing relation. + return communityBan{}, false + } + var ban communityBan + err := db.QueryRowContext(context.Background(), ` + SELECT community_ap_id, expires_at + FROM community_bans + WHERE community_did = $1 AND subject_did = $2`, + communityDID, subjectDID).Scan(&ban.communityAPID, &ban.expires) + if err == sql.ErrNoRows { + return communityBan{}, false + } + require.NoError(t, err, "read the ban state for %s in %s", subjectDID, communityDID) + return ban, true +} + +// requireEveryDelivery asserts that the actor HAS work on that community's +// ordering key and that every row of it is in the wanted state. +// +// The non-empty check is not ceremony. "Every one of no rows is cancelled" is +// true of a cancellation that DELETED the rows, of a fixture whose posts never +// enqueued, and of a query with a typo in it — three different nothings that all +// read as success. The rows are also the evidence an operator needs afterwards, +// so their continued existence is part of the behaviour, not an artifact of it. +func requireEveryDelivery(t *testing.T, db *sql.DB, actorDID, orderingKey, want, why string) { + t.Helper() + states := deliveryStates(t, db, actorDID, orderingKey) + require.NotEmpty(t, states, + "%s — and there must BE deliveries to say that about: %s has no rows on %s at all", + why, actorDID, orderingKey) + for i, state := range states { + require.Equal(t, want, state, "%s (delivery %d of %d)", why, i+1, len(states)) + } +} + +// tableExists reports whether a table has been created yet. +func tableExists(t *testing.T, db *sql.DB, name string) bool { + t.Helper() + var relation sql.NullString + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT to_regclass($1)`, "public."+name).Scan(&relation)) + return relation.Valid +} + +// resetBanState clears the ban table between runs. +// +// newHarness's truncate list cannot name it until it exists, and it MUST be +// cleared: the fixture's author and community are package-level constants, so a +// ban left by one run would refuse the next run's post before the test that +// issues the ban has run — green first, red second, which a single CI run never +// sees. Fold this into newHarness and delete it once the migration lands. +func resetBanState(t *testing.T, db *sql.DB) { + t.Helper() + if tableExists(t, db, "community_bans") { + testutil.Truncate(t, db, "community_bans") + } +} + +// deliveryStates lists the states of one actor's deliveries on one community's +// ordering key. +// +// The ordering key is what makes a ban expressible at all: co-hosted communities +// SHARE an inbox, so the delivery address says nothing about which community's +// traffic a row carries. A cancellation scoped by inbox would take both. +func deliveryStates(t *testing.T, db *sql.DB, actorDID, orderingKey string) []string { + t.Helper() + rows, err := db.QueryContext(context.Background(), ` + SELECT d.state + FROM outbound_deliveries d + JOIN outbound_activities a ON a.activity_id = d.activity_id + WHERE a.actor_did = $1 AND d.ordering_key = $2 + ORDER BY d.seq`, actorDID, orderingKey) + require.NoError(t, err) + defer func() { require.NoError(t, rows.Close()) }() + + var states []string + for rows.Next() { + var state string + require.NoError(t, rows.Scan(&state)) + states = append(states, state) + } + require.NoError(t, rows.Err()) + return states +} + +// TestADirectBlockIsIgnored pins the DECIDED non-action at the other door. +// +// Lemmy sends a ban twice: announced through the community, and delivered +// DIRECTLY to the banned user's inbox. The two are not redundant copies of one +// decision — they are signed by different actors, and only one of them can be +// authorized. +// +// BlockUser's `actor` is the MODERATOR's Person. On the announced path the HTTP +// signature binds the community Group, so decision 18's conjunction (the signer +// IS the community that owns the target) can hold. On the direct path the signer +// is a Person, so that conjunction CANNOT pass by construction — no amount of +// implementation makes a person into a group. Refusing it is therefore not a gap +// left open, it is the only correct outcome, and it must be a visible decision +// rather than a silent drop: a ban that arrives only by the path we ignore looks +// exactly like a ban that never arrived. +func TestADirectBlockIsIgnored(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + // The moderator as a real signer: their own key, their own actor document. + moderator := h.newRemoteActor(modActorID, person(modActorID, "moderator", nil)) + before := metricValue(mbDirectIgnoredMetric) + + const directBlock = "https://lemmy.world/activities/block/mb-direct" + require.Equal(t, http.StatusAccepted, h.deliver(moderator, + blockActivity(world.groupA, directBlock, mtAuthorDID, groupID, nil))) + h.drain() + + _, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + assert.False(t, found, + "a Person-signed Block records NO ban: the announced copy is the authoritative one, "+ + "and honouring this path would let any account on a Lemmy instance ban a native "+ + "author out of any community co-hosted there") + + assert.Equal(t, before+1, metricValue(mbDirectIgnoredMetric), + "and it is COUNTED: a decided non-action that is invisible reads exactly like a ban "+ + "we never received, so the day the announced path breaks, this counter is the only "+ + "thing that distinguishes 'ignored on purpose' from 'silently lost'") + + event, err := h.events.GetEvent(ctx, directBlock) + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, + "skipped ONCE, not left retrying: nothing about this delivery improves on a second "+ + "attempt, and a wedged ordering key would hold every later activity behind it: %s", + event.Error) + assert.Nil(t, event.FailedAt, "nor poisoned") +} + +// TestADirectBlockClaimingToBeTheGroupIsStillIgnored is SEC-1's fix doing work +// in a new place. +// +// The activity CLAIMS `actor` = the community Group while being SIGNED by a +// moderator's Person. Both live on lemmy.world, so the same-authority tolerance +// at the door lets the delivery in — and before SEC-1 the CLAIM was what got +// bound, which would have made this indistinguishable from the community's own +// announced ban. It is the exact shape that was exploitable three sub-runs ago, +// arriving now at a verb that hands out bans. +func TestADirectBlockClaimingToBeTheGroupIsStillIgnored(t *testing.T) { + h := newHarness(t) + resetBanState(t, h.db) + world := newModerationWorld(t, h) + moderator := h.newRemoteActor(modActorID, person(modActorID, "moderator", nil)) + + const forged = "https://lemmy.world/activities/block/mb-forged" + block := blockActivity(world.groupA, forged, mtAuthorDID, groupID, nil) + block["actor"] = groupID // the claim: "I am the community" + + require.Equal(t, http.StatusAccepted, h.deliver(moderator, block)) + h.drain() + + _, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + assert.False(t, found, + "the VERIFIED SIGNER decides, never the claim: binding the claimed actor would let "+ + "any account on the instance speak as the community — here, to ban a native author "+ + "from a community whose moderators did nothing") + + admitPost(t, world, mtAuthorDID, mbPostAfterBanRKey, world.communityADID, + "3lzmbrev00010", 1_775_000_002_000_001) + status, _ := admissionFor(t, h.db, world.communityADID, mbPostATURI(mtAuthorDID, mbPostAfterBanRKey)) + assert.Equal(t, accept.StatusAccepted, status, + "and the author keeps posting: a forged ban that only half-applied would be a "+ + "shadowban nobody issued and nobody can lift") +} + +// TestALapsedBanDoesNotRefuseAdmission is the half of `expires` that has no +// activity behind it. +// +// Lemmy's BlockUser carries `expires` for a temporary ban, and when that ban +// lapses Lemmy sends NOTHING — no Undo, no second activity, nothing. The ban +// simply stops applying on their side. So an implementation that stores the ban +// and ignores the column turns every 3-day ban into a permanent one, and there +// is no message that will ever clear it: the author is excluded forever by a +// moderator who chose three days. +func TestALapsedBanDoesNotRefuseAdmission(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + h.announceBlock(world.groupA, mbBlockActivity, mtAuthorDID, groupID, + map[string]any{"expires": "2020-01-01T00:00:00Z"}) + + if ban, found := banFor(t, h.db, world.communityADID, mtAuthorDID); found { + require.True(t, ban.expires.Valid, + "a ban recorded from an activity carrying `expires` must carry the expiry with it: "+ + "dropping the column is what makes the lapse unrepresentable") + assert.True(t, ban.expires.Time.Before(timeNow()), + "and it must be the moment the moderator chose, in the past") + } + + admitPost(t, world, mtAuthorDID, mbPostAfterBanRKey, world.communityADID, + "3lzmbrev00011", 1_775_000_003_000_001) + postURI := mbPostATURI(mtAuthorDID, mbPostAfterBanRKey) + status, code := admissionFor(t, h.db, world.communityADID, postURI) + assert.Equal(t, accept.StatusAccepted, status, + "a LAPSED ban must not refuse admission: every read has to be `expires_at IS NULL OR "+ + "expires_at > now()`, because no Undo is coming — the expiry IS the lift") + assert.NotEqual(t, "author-banned", code, "and certainly not for being banned") + + _, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, testDigestRKey(postURI)) + assert.NoError(t, err, "with the acceptance to prove it (err=%v)", err) +} + +// TestAStandingTimedBanRefusesAdmission is the other half: the same activity +// shape, an expiry that has NOT arrived, and a ban that bites. +// +// The pair is the point. An implementation that reads the column as "any expiry +// means expired" passes the lapsed test alone; one that ignores it passes this +// one alone. Only both together say the column is being read. +func TestAStandingTimedBanRefusesAdmission(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + h.announceBlock(world.groupA, mbBlockActivity, mtAuthorDID, groupID, + map[string]any{"expires": "2099-01-01T00:00:00Z"}) + + ban, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + require.True(t, found, "a temporary ban is still a ban and must be recorded") + require.True(t, ban.expires.Valid, "carrying its expiry") + assert.True(t, ban.expires.Time.After(timeNow()), "which has not arrived") + + admitPost(t, world, mtAuthorDID, mbPostAfterBanRKey, world.communityADID, + "3lzmbrev00012", 1_775_000_004_000_001) + postURI := mbPostATURI(mtAuthorDID, mbPostAfterBanRKey) + status, code := admissionFor(t, h.db, world.communityADID, postURI) + assert.Equal(t, accept.StatusRejected, status, + "a ban whose expiry is still ahead applies exactly like a permanent one: a temporary "+ + "ban that admits posts is not a ban at all") + assert.Equal(t, "author-banned", code, "with the same reason") + + _, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, testDigestRKey(postURI)) + assert.True(t, errors.IsNotFound(err), "and no acceptance (err=%v)", err) +} + +// TestACrossCommunityBlockIsRefused is the RELATIONAL case for bans. +// +// Community B announces a Block whose TARGET is community A. B is followed, B +// signs as itself, and B shares an instance with A — nothing about the delivery +// is malformed. It is simply not B's decision: a ban is a community's ruling +// about its OWN space, and one community handing out exclusions from another's +// is the co-hosting failure decision 18 exists to prevent. +func TestACrossCommunityBlockIsRefused(t *testing.T) { + h := newHarness(t) + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + h.announceBlock(world.groupB, "https://lemmy.world/activities/announce/block/mb-cross", + mtAuthorDID, groupID, nil) + + _, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + assert.False(t, found, + "B may not ban an author out of A: one moderator team would otherwise be able to "+ + "exclude anyone from every community co-hosted with theirs") + + _, foundInB := banFor(t, h.db, world.communityBDID, mtAuthorDID) + assert.False(t, foundInB, + "nor may it be recorded against B instead: B announced a ruling about A's space, and "+ + "quietly applying it to B's own invents an exclusion B never issued") + + admitPost(t, world, mtAuthorDID, mbPostAfterBanRKey, world.communityADID, + "3lzmbrev00013", 1_775_000_005_000_001) + status, _ := admissionFor(t, h.db, world.communityADID, mbPostATURI(mtAuthorDID, mbPostAfterBanRKey)) + assert.Equal(t, accept.StatusAccepted, status, "and A keeps admitting the author's posts") +} + +// TestASiteScopedBlockIsSkippedWithItsOwnReason covers the target this scope +// does not model. +// +// Lemmy issues instance-wide bans too, and they arrive as the same verb with +// `target` naming the SITE actor rather than a community. Scope A models +// per-community bans only — so the honest outcome is a refusal that says which +// kind it was. Silently no-op'ing one leaves the user posting into that instance +// and collecting 403s until their deliveries poison, with nothing anywhere +// naming the site ban as the cause. +func TestASiteScopedBlockIsSkippedWithItsOwnReason(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + // Lemmy's Site actor is the instance apex — which is a URL PREFIX of every + // other id on that instance, so "the reason mentions the target" is true of + // any message quoting any lemmy.world url. The distinction has to be drawn + // against another outcome, not against a substring. + const siteActor = "https://lemmy.world/" + const siteBlock = "https://lemmy.world/activities/announce/block/mb-site" + h.announceBlock(world.groupA, siteBlock, mtAuthorDID, siteActor, nil) + + _, found := banFor(t, h.db, world.communityADID, mtAuthorDID) + assert.False(t, found, + "a site ban is not a community ban: recording it against the announcing community "+ + "would understate it — the user is excluded from every community on that instance, "+ + "and we would enforce it in one") + + // The same activity, targeted at the community, is the shape that WORKS. + const communityBlock = "https://lemmy.world/activities/announce/block/mb-site-control" + h.announceBlock(world.groupA, communityBlock, mtCommenterDID, groupID, nil) + + assert.NotEqual(t, + skipReasonFor(t, h, communityBlock, communityBlock), + skipReasonFor(t, h, siteBlock, siteBlock), + "a target this scope does not model must be distinguishable from one it does: today "+ + "both land in the same 'unsupported activity type' default, so an operator whose "+ + "user is collecting 403s across a whole instance reads the same line as someone "+ + "whose ban was applied — and silently no-op'ing the site ban leaves that user "+ + "posting into the instance until their deliveries poison") + + event, err := h.events.GetEvent(ctx, siteBlock) + require.NoError(t, err) + assert.NotNil(t, event.ProcessedAt, "decided once, not retried: %s", event.Error) + assert.Nil(t, event.FailedAt, "nor poisoned") +} + +// mbDirectIgnoredMetric is the counter for direct Blocks. Read through expvar by +// NAME so the test survives the var being renamed or moved — the counter's +// identity is its published name, which is what an operator's dashboard binds +// to, not the Go symbol. +const mbDirectIgnoredMetric = "tidepool_block_direct_ignored" + +func metricValue(name string) int64 { + counter, _ := expvar.Get(name).(*expvar.Int) + if counter == nil { + return 0 + } + return counter.Value() +} + +func timeNow() time.Time { return time.Now() } + +// TestABanWithRemoveDataRemovesTheirPostsInThatCommunityOnly is the destructive +// half, and its scope is the whole design. +// +// `removeData: true` means Lemmy purged that author's content — in THAT +// community. The admissions ledger is the only table that knows which posts were +// ever ACCEPTED there (idx_admissions_author_community), which is why it is the +// input rather than "every post by this author". +// +// The removal code is `author-banned`, not moderator-discretion: they are +// different decisions with different reversals, and a reader that cannot tell +// them apart cannot tell "this post broke a rule" from "its author is no longer +// welcome here". That distinction is what RemovePost's new `code` parameter +// exists for — it hardcodes moderator-discretion today. +func TestABanWithRemoveDataRemovesTheirPostsInThatCommunityOnly(t *testing.T) { + h := newHarness(t) + ctx := context.Background() + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + // The author has accepted posts in BOTH communities. The world already gave + // them one in A; this is the one in B that must survive. + admitPost(t, world, mtAuthorDID, mbPostInBRKey, world.communityBDID, "3lzmbrev00020", 1_775_000_006_000_001) + postInB := mbPostATURI(mtAuthorDID, mbPostInBRKey) + _, _, err := h.manager.GetRecord(ctx, + world.communityBDID, materialize.CollectionAcceptance, testDigestRKey(postInB)) + require.NoError(t, err, "precondition: the author is accepted in B too") + + // And the OTHER author has an accepted post in A, which a ban on someone + // else must not touch. + admitPost(t, world, mtCommenterDID, mbOtherPostRKey, world.communityADID, "3lzmbrev00021", 1_775_000_006_000_002) + otherPost := mbPostATURI(mtCommenterDID, mbOtherPostRKey) + + activitiesBefore := rowCount(t, h.db, "outbound_activities") + deliveriesBefore := rowCount(t, h.db, "outbound_deliveries") + + h.announceBlock(world.groupA, mbBlockActivity, mtAuthorDID, groupID, + map[string]any{"removeData": true}) + + // --- Their post in A is removed, under the ban's own code. + removal, _, err := h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionRemoval, world.digestRKey) + require.NoError(t, err, + "removeData means their content in this community goes: Lemmy has already purged it "+ + "on their side, and leaving it accepted here publishes a community's endorsement "+ + "of content that community has removed") + assert.Equal(t, "author-banned", removal["code"], + "under author-banned, not moderator-discretion: this post was not judged, its AUTHOR "+ + "was — and the two have different reversals, so a reader that cannot tell them "+ + "apart cannot answer why the post went") + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, world.digestRKey) + assert.True(t, errors.IsNotFound(err), "with the acceptance withdrawn (err=%v)", err) + + // --- Their post in B is untouched. + _, _, err = h.manager.GetRecord(ctx, + world.communityBDID, materialize.CollectionAcceptance, testDigestRKey(postInB)) + assert.NoError(t, err, + "their post in B STANDS: removeData is scoped to the banning community's own space, "+ + "and an author banned from one community losing their writing everywhere is a "+ + "site-wide purge issued by a single moderator team (err=%v)", err) + _, _, err = h.manager.GetRecord(ctx, + world.communityBDID, materialize.CollectionRemoval, testDigestRKey(postInB)) + assert.True(t, errors.IsNotFound(err), "and B records no removal of it (err=%v)", err) + + // --- The other author's post in A is untouched. + otherStatus, _ := admissionFor(t, h.db, world.communityADID, otherPost) + assert.Equal(t, accept.StatusAccepted, otherStatus, + "and the OTHER author's post in A stands: the ledger query is (author, community), "+ + "and dropping the author term removes the whole community's backlog") + + // --- Nothing goes out, for two independent reasons. + assert.Equal(t, activitiesBefore, rowCount(t, h.db, "outbound_activities"), + "NOTHING is enqueued. Structurally: removal runs through the materializer, which uses "+ + "ApplyOps and takes no side effect, so it cannot enqueue. Semantically: Lemmy has "+ + "ALREADY removed this content — the ban is us learning what they did, not us "+ + "asking them to do it — so an outbound Delete would be redundant and aimed at the "+ + "moderators who just acted") + assert.Equal(t, deliveriesBefore, rowCount(t, h.db, "outbound_deliveries"), "...and no new delivery") + + // --- And the unban does NOT bring the content back. + h.announceUndoBlock(world.groupA, mbUnblockActivity, mbBlockActivity+"/block", mtAuthorDID, groupID) + + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionRemoval, world.digestRKey) + assert.NoError(t, err, + "the removal SURVIVES the unban: Lemmy models restoration as a separate restore_data "+ + "flag, so an Undo{Block} that quietly republished removed posts would reverse a "+ + "moderator's decision and push the content back at the community that removed it "+ + "— the same harm as reversing a removal on an author's edit, arriving by another "+ + "door (err=%v)", err) + _, _, err = h.manager.GetRecord(ctx, + world.communityADID, materialize.CollectionAcceptance, world.digestRKey) + assert.True(t, errors.IsNotFound(err), + "and no acceptance comes back with it (err=%v)", err) +} + +// TestABanLeavesTerminalDeliveriesAlone pins what cancellation must NOT touch. +// +// Cancellation is for PENDING work. The two terminal states are terminal for +// different reasons and both must survive a ban: +// +// - `delivered` cannot be un-sent. Rewriting it would make the ledger claim we +// withdrew something the peer holds — and that column is not private +// bookkeeping: the vote reseed subtracts live delivered outbound state from +// the API tally, so falsifying it silently corrupts a user-visible score in +// a subsystem nobody would think to look at from here. +// - `poisoned` is an operator surface. A delivery that failed for its own +// reason, swept into "cancelled" by a ban that arrived later, is a fault +// nobody will ever diagnose — and redrive is how someone recovers it. +func TestABanLeavesTerminalDeliveriesAlone(t *testing.T) { + h := newHarness(t) + resetBanState(t, h.db) + world := newModerationWorld(t, h) + + admitPost(t, world, mtAuthorDID, mbPostInARKey, world.communityADID, "3lzmbrev00030", 1_775_000_007_000_001) + admitPost(t, world, mtAuthorDID, mbPostAfterUndoRKey, world.communityADID, "3lzmbrev00031", 1_775_000_007_000_002) + + // Force the two terminal states onto this author's work for A. A worker run + // would be the honest way to reach 'delivered', but the state is what the + // ban's scope is about, and driving it directly is what keeps the fixture + // about the ban rather than about delivery. + terminal := forceDeliveryStates(t, h.db, mtAuthorDID, groupID, "delivered", "poisoned") + require.Len(t, terminal, 2, "precondition: two terminal deliveries to ban across") + + h.announceBlock(world.groupA, mbBlockActivity, mtAuthorDID, groupID, nil) + + assert.Equal(t, "delivered", deliveryState(t, h.db, terminal[0]), + "a DELIVERED row is untouched: it cannot be un-sent, and the reseed reads exactly "+ + "this column to subtract our personas' live votes from the API tally — rewriting "+ + "it moves a number the user sees, from a subsystem this code has never heard of") + assert.Equal(t, "poisoned", deliveryState(t, h.db, terminal[1]), + "and a POISONED row is untouched: sweeping it hides a delivery that failed for its "+ + "own reason behind a ban that arrived afterwards, and redrive is how an operator "+ + "gets it back") +} + +// forceDeliveryStates stamps states onto an actor's pending deliveries for one +// community, in seq order, and returns the activity ids it stamped. +func forceDeliveryStates(t *testing.T, db *sql.DB, actorDID, orderingKey string, states ...string) []string { + t.Helper() + ctx := context.Background() + rows, err := db.QueryContext(ctx, ` + SELECT d.activity_id + FROM outbound_deliveries d + JOIN outbound_activities a ON a.activity_id = d.activity_id + WHERE a.actor_did = $1 AND d.ordering_key = $2 AND d.state = 'pending' + ORDER BY d.seq + LIMIT $3`, actorDID, orderingKey, len(states)) + require.NoError(t, err) + var ids []string + for rows.Next() { + var id string + require.NoError(t, rows.Scan(&id)) + ids = append(ids, id) + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + require.Len(t, ids, len(states), + "the fixture needs %d pending deliveries for %s on %s to stamp", len(states), actorDID, orderingKey) + + for i, id := range ids { + _, err := db.ExecContext(ctx, + `UPDATE outbound_deliveries SET state = $2 WHERE activity_id = $1`, id, states[i]) + require.NoError(t, err) + } + return ids +} + +func deliveryState(t *testing.T, db *sql.DB, activityID string) string { + t.Helper() + var state string + require.NoError(t, db.QueryRowContext(context.Background(), + `SELECT state FROM outbound_deliveries WHERE activity_id = $1`, activityID).Scan(&state)) + return state +} diff --git a/internal/ingest/moderation_lock_test.go b/internal/ingest/moderation_lock_test.go index 6a265c6..29f5e61 100644 --- a/internal/ingest/moderation_lock_test.go +++ b/internal/ingest/moderation_lock_test.go @@ -1168,6 +1168,13 @@ func skipReasonFor(t *testing.T, h *harness, activityID, target string) string { if !ok { return "" } + if target == "" { + // Redacting the empty string inserts the placeholder between every + // character, which turns a readable reason into noise and can make + // a Contains assertion pass or fail for reasons unrelated to the + // code under test. An empty target means redact nothing. + return reason + } return strings.ReplaceAll(reason, target, "") } return "" diff --git a/internal/ingest/moderation_terminal_test.go b/internal/ingest/moderation_terminal_test.go index 1ac05f4..2659d99 100644 --- a/internal/ingest/moderation_terminal_test.go +++ b/internal/ingest/moderation_terminal_test.go @@ -98,23 +98,22 @@ func newModerationWorld(t *testing.T, h *harness) moderationWorld { h.serveLemmyWorldContent() communityADID := testDIDFor(mtCommunityAName, "lemmy.world") - // --- Community B: followed, co-hosted, and able to sign its own deliveries. + // --- Community B: followed, co-hosted, and REAL — minted through the same + // admin subscribe the operator uses, so it has a DID, a signing key and + // a repo of its own. + // + // It was a bare communities row until 17c-3, which was enough while every + // test only needed B to SIGN something and be refused. It is not enough for a + // ban: "cancel that author's deliveries to THAT community" is only + // distinguishable from "cancel them everywhere" if the other community can + // hold content and accept posts, and an acceptance is a record in B's own + // repo — which needs a key we can only get by minting one. + groupB := h.subscribeCommunityURL(mtCommunityBAPID, mtCommunityBName) communityBDID := testDIDFor(mtCommunityBName, "lemmy.world") - _, err := h.communities.UpsertCommunity(ctx, store.Community{ - APGroupID: mtCommunityBAPID, - DID: communityBDID, - PreferredUsername: mtCommunityBName, - Instance: "lemmy.world", - }) + communityB, err := h.communities.GetByAPGroupID(ctx, mtCommunityBAPID) require.NoError(t, err) - require.NoError(t, h.communities.SetFollowState(ctx, mtCommunityBAPID, store.FollowStateAccepted)) - groupB := h.newRemoteActor(mtCommunityBAPID, map[string]any{ - "type": "Group", - "id": mtCommunityBAPID, - "preferredUsername": mtCommunityBName, - "inbox": mtCommunityBAPID + "/inbox", - "published": "2024-01-01T00:00:00.000000Z", - }) + require.Equal(t, communityBDID, communityB.DID, + "precondition: B's minted DID is the one the fixtures name") // --- The native side: a persona service, the real enqueuer, the real // acceptance engine, and the real consumer in front of them. @@ -198,6 +197,16 @@ func mtPostEvent(t *testing.T, operation, rev, cid string, timeUS int64) *consum // per-OBJECT, so telling that apart from per-community needs a second thread in // the SAME community — which needs a second post to root it. func mtPostEventFor(t *testing.T, rkey, operation, rev, cid string, timeUS int64) *consume.JetstreamEvent { + t.Helper() + return mtPostEventBy(t, mtAuthorDID, rkey, testDIDFor(mtCommunityAName, "lemmy.world"), + operation, rev, cid, timeUS) +} + +// mtPostEventBy names the AUTHOR and the COMMUNITY as well. A ban is an +// intersection of the two — this author, in this community — so a fixture that +// can only vary one of them cannot express the difference between a ban and a +// community going dark. +func mtPostEventBy(t *testing.T, did, rkey, communityDID, operation, rev, cid string, timeUS int64) *consume.JetstreamEvent { t.Helper() frame := fmt.Sprintf(`{ "did": %q, "time_us": %d, "kind": "commit", @@ -213,7 +222,7 @@ func mtPostEventFor(t *testing.T, rkey, operation, rev, cid string, timeUS int64 "createdAt": "2026-08-13T10:00:00.000Z" } } -}`, mtAuthorDID, timeUS, rev, operation, rkey, cid, testDIDFor(mtCommunityAName, "lemmy.world")) +}`, did, timeUS, rev, operation, rkey, cid, communityDID) var event consume.JetstreamEvent require.NoError(t, json.Unmarshal([]byte(frame), &event), "the frame must be valid wire JSON") return &event diff --git a/internal/materialize/acceptance.go b/internal/materialize/acceptance.go index 699f8e2..a0efd07 100644 --- a/internal/materialize/acceptance.go +++ b/internal/materialize/acceptance.go @@ -109,6 +109,17 @@ func (m *Materializer) removalStands(ctx context.Context, communityDID, rkey str // post or a comment, and two copies of the literal is exactly how that drifts. const RemovalCodeModeratorDiscretion = "moderator-discretion" +// RemovalCodeAuthorBanned is the removal code for content that went because its +// AUTHOR was banned from the community, not because the post itself was judged. +// +// The distinction is the reader's, and it is not cosmetic: the two decisions +// have different reversals. A moderator-discretion removal is lifted by an +// Undo{Delete} of that post; this one stands even when the ban is lifted, because +// Lemmy models content restoration as a SEPARATE restore_data flag. A reader +// that cannot tell them apart cannot answer why the post went, or what would +// bring it back. +const RemovalCodeAuthorBanned = "author-banned" + // RemovePost records a community's moderator removal of a post: the acceptance // is deleted and a removal written IN ONE COMMIT, at the same digest rkey. // @@ -122,7 +133,79 @@ const RemovalCodeModeratorDiscretion = "moderator-discretion" // deleting the author's record would let one community destroy content for // every other, and tombstoning the mapping would block the post's later edits // and votes from ever materializing again. +// It is the MODERATOR-DISCRETION entry point: an announced Delete carrying a +// summary is a moderator's judgement of the post, and Lemmy gives no +// machine-readable code to narrow it with. A removal that went for another +// reason enters through RemoveAuthorPosts, which names its own. func (m *Materializer) RemovePost(ctx context.Context, mapping *store.APObjectMapping, reason string) error { + return m.removePost(ctx, mapping, RemovalCodeModeratorDiscretion, reason) +} + +// RemoveAuthorPosts removes every post one author currently has ACCEPTED in one +// community — a ban's `removeData: true`. +// +// THE SCOPE IS THE WHOLE DESIGN, and it is why the admissions ledger is the +// input rather than "every post by this author": the ledger is the only table +// that records which community ADMITTED a post. A ban entitles a community to +// act on its own space, so an author banned from one community must not lose +// their writing in the others — that would be a site-wide purge issued by a +// single moderator team. +// +// Each removal is the same one-commit transition RemovePost performs, under the +// ban's own code, so a reader sees why each post went. It CANNOT enqueue: +// removals ride repos.ApplyOps, which takes no side effect — and semantically +// there is nothing to send, because Lemmy has already removed this content on +// their side. An outbound Delete would be aimed at the moderators who just +// acted. +// +// One post's failure does not abandon the rest: the ban has already landed, and +// stopping at the first error would leave an arbitrary prefix removed with no +// record of what remained. The first error is returned after the sweep, so the +// activity retries and the removals — each idempotent — converge. +func (m *Materializer) RemoveAuthorPosts(ctx context.Context, communityDID, authorDID, code, reason string) (int, error) { + if m.ledger == nil { + // Without the ledger there is no way to know WHICH posts this community + // admitted, and "every post by this author" is the wrong answer rather + // than an approximate one. + return 0, fmt.Errorf("materialize: no admissions ledger wired; cannot scope a removeData purge") + } + postURIs, err := m.ledger.ListAccepted(ctx, communityDID, authorDID) + if err != nil { + return 0, fmt.Errorf("materialize: list %s's accepted posts in %s: %w", authorDID, communityDID, err) + } + + var removed int + var firstErr error + for _, postURI := range postURIs { + mapping, err := m.objects.GetByATURI(ctx, postURI) + if errors.IsNotFound(err) { + // Admitted but never mapped: nothing federated under it, so there is + // no acceptance/removal pair to rewrite. + m.logger.Warn("removeData: accepted post has no mapping; skipping", + "community_did", communityDID, "post", postURI) + continue + } + if err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("materialize: load mapping for %s: %w", postURI, err) + } + continue + } + if err := m.removePost(ctx, mapping, code, reason); err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + removed++ + } + m.logger.Info("removed a banned author's posts from a community", + "community_did", communityDID, "author_did", authorDID, + "removed", removed, "accepted", len(postURIs), "code", code) + return removed, firstErr +} + +func (m *Materializer) removePost(ctx context.Context, mapping *store.APObjectMapping, code, reason string) error { communityDID, postURI, rkey, err := m.moderationTarget(ctx, mapping) if err != nil { return err @@ -152,10 +235,15 @@ func (m *Materializer) RemovePost(ctx context.Context, mapping *store.APObjectMa removal := map[string]any{ "$type": CollectionRemoval, "subject": strongRef(postURI, pinned), - // Lemmy sends no machine-readable code, so the open knownValues set's - // catch-all applies. Inventing a narrower code (spam, rule-violation) - // would be the bridge asserting a reason the moderator never gave. - "code": RemovalCodeModeratorDiscretion, + // The code comes from the CALLER, because the removal lexicon's + // knownValues set is open and the two decisions that reach here are + // genuinely different: a post the moderators judged + // (moderator-discretion, the catch-all Lemmy's own activity gives us no + // better answer than) versus content that went because its author was + // banned (author-banned). Inventing anything NARROWER than what the + // caller was told — spam, rule-violation — would still be the bridge + // asserting a reason the moderator never gave. + "code": code, "createdAt": recordDatetime(m.moderationStamp(ctx, communityDID, CollectionRemoval, rkey)), } // Omitted rather than written blank: Lemmy spells "no reason given" as an @@ -177,7 +265,7 @@ func (m *Materializer) RemovePost(ctx context.Context, mapping *store.APObjectMa m.logger.Info("post removed from community by moderator", "community_did", communityDID, "post", postURI, "ap_id", mapping.APID) m.recordModeration(ctx, mapping, func() error { - return m.ledger.RecordRemoval(ctx, communityDID, postURI, mapping.DID, RemovalCodeModeratorDiscretion) + return m.ledger.RecordRemoval(ctx, communityDID, postURI, mapping.DID, code) }) return nil } diff --git a/internal/materialize/materializer.go b/internal/materialize/materializer.go index b49e678..c2ab75f 100644 --- a/internal/materialize/materializer.go +++ b/internal/materialize/materializer.go @@ -157,6 +157,15 @@ type ModerationLedger interface { // outward, which is exactly the case a restore has to pin. "" means the // ledger knows of no decision for this (community, post). LastEvaluatedCID(ctx context.Context, communityDID, postURI string) (string, error) + // ListAccepted returns the at-uris of the posts this author currently has + // ACCEPTED in this community — the input to a ban's removeData purge. + // + // The ledger is the only table that knows this. ap_objects knows what was + // materialized and outbound_objects knows what was federated, but neither + // records which community ADMITTED a post, which is precisely the scope a + // ban is entitled to act on: "their content HERE", never everything they + // ever wrote. + ListAccepted(ctx context.Context, communityDID, authorDID string) ([]string, error) } // Options configures New. Fetcher, Objects, Actors, Communities, Repos, diff --git a/internal/store/communities.go b/internal/store/communities.go index a30fa21..a23915f 100644 --- a/internal/store/communities.go +++ b/internal/store/communities.go @@ -14,11 +14,18 @@ import ( type postgresCommunities struct { db *sql.DB + // The ban repository is EMBEDDED, and only so that a holder of the concrete + // communities store can be type-asserted to store.CommunityBans (see + // ingest.NewHandler and accept.NewEngine, which both already hold one). It + // is deliberately NOT part of the Communities interface: a ban is a + // different table with a different key, and the follow-state readers that + // hold Communities have no business writing exclusions. + postgresCommunityBans } // NewCommunities creates the postgres-backed communities repository. func NewCommunities(db *sql.DB) Communities { - return &postgresCommunities{db: db} + return &postgresCommunities{db: db, postgresCommunityBans: postgresCommunityBans{db: db}} } const communityColumns = ` diff --git a/internal/store/community_bans.go b/internal/store/community_bans.go new file mode 100644 index 0000000..d3f63ce --- /dev/null +++ b/internal/store/community_bans.go @@ -0,0 +1,122 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + + "tidepool/internal/errors" +) + +// postgresCommunityBans is the community_bans repository: one community's ban of +// one author (migration 027). +// +// It is its own repository over its own table, and postgresCommunities EMBEDS it +// so a holder of the concrete communities store can be type-asserted to +// CommunityBans — a wiring convenience that does NOT widen the Communities +// interface (see the embed for why that matters). +type postgresCommunityBans struct { + db *sql.DB +} + +// NewCommunityBans creates the postgres-backed community_bans repository. +func NewCommunityBans(db *sql.DB) CommunityBans { + return &postgresCommunityBans{db: db} +} + +func (r *postgresCommunityBans) Ban(ctx context.Context, ban CommunityBan) (cancelled int64, err error) { + switch { + case ban.CommunityDID == "": + return 0, errors.NewValidationError("community_did", "must not be empty") + case ban.SubjectDID == "": + return 0, errors.NewValidationError("subject_did", "must not be empty") + case ban.CommunityAPID == "": + // The delivery side has no other handle for this community, so a ban + // stored without it is one the queue can never apply. + return 0, errors.NewValidationError("community_ap_id", "must not be empty") + } + + // ONE TRANSACTION, because the two writes are one decision. The row is what + // stops the author's NEXT post; the cancellation is what stops the posts + // already queued. If the row committed and the cancellation did not, the + // queue would keep pushing a banned author's posts at a community that + // rejects them until each one poisons — and nothing would retry the + // cancellation, because Lemmy sends the Block exactly once. + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return 0, fmt.Errorf("ban %q in %q: begin: %w", ban.SubjectDID, ban.CommunityDID, err) + } + defer func() { _ = tx.Rollback() }() + + // banned_at is preserved on conflict: a re-delivered Block is the same ban + // arriving twice, not a new one. Everything the moderator can CHANGE by + // re-issuing (the expiry, the reason, whether content goes) is taken from + // the new activity. + if _, err := tx.ExecContext(ctx, ` + INSERT INTO community_bans ( + community_did, subject_did, community_ap_id, expires_at, reason, remove_data) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (community_did, subject_did) DO UPDATE SET + community_ap_id = EXCLUDED.community_ap_id, + expires_at = EXCLUDED.expires_at, + reason = EXCLUDED.reason, + remove_data = EXCLUDED.remove_data, + updated_at = now()`, + ban.CommunityDID, ban.SubjectDID, ban.CommunityAPID, + ban.ExpiresAt, ban.Reason, ban.RemoveData); err != nil { + return 0, fmt.Errorf("ban %q in %q: %w", ban.SubjectDID, ban.CommunityDID, err) + } + + cancelled, err = cancelPendingForActorInCommunity(ctx, tx, ban.SubjectDID, ban.CommunityAPID) + if err != nil { + return 0, err + } + if err := tx.Commit(); err != nil { + return 0, fmt.Errorf("ban %q in %q: commit: %w", ban.SubjectDID, ban.CommunityDID, err) + } + return cancelled, nil +} + +func (r *postgresCommunityBans) Lift(ctx context.Context, communityDID, subjectDID string) (lifted bool, err error) { + if communityDID == "" || subjectDID == "" { + return false, errors.NewValidationError("ban", "community_did and subject_did must not be empty") + } + // The row is DELETED rather than marked lifted. A ban is current state, not + // a log — the admissions ledger already records what happened to each post — + // and a lifted row that readers had to filter is one more chance to read a + // ban that no longer exists as one that does. + // + // It lifts ONLY the ban. Content removed under removeData stays removed: + // Lemmy models restoration as a separate restore_data flag, so republishing + // here would reverse a decision nobody reversed. + result, err := r.db.ExecContext(ctx, + `DELETE FROM community_bans WHERE community_did = $1 AND subject_did = $2`, + communityDID, subjectDID) + if err != nil { + return false, fmt.Errorf("lift ban on %q in %q: %w", subjectDID, communityDID, err) + } + affected, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("lift ban on %q in %q: rows affected: %w", subjectDID, communityDID, err) + } + return affected > 0, nil +} + +func (r *postgresCommunityBans) Standing(ctx context.Context, communityDID, subjectDID string) (bool, error) { + if communityDID == "" || subjectDID == "" { + return false, errors.NewValidationError("ban", "community_did and subject_did must not be empty") + } + var standing bool + // The expiry test is in the STATEMENT, not in Go, so no reader can forget + // it: a lapsed ban is indistinguishable from no ban, and the only signal + // that it lapsed is the clock — Lemmy sends nothing. + if err := r.db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 FROM community_bans + WHERE community_did = $1 AND subject_did = $2 + AND (expires_at IS NULL OR expires_at > now()))`, + communityDID, subjectDID).Scan(&standing); err != nil { + return false, fmt.Errorf("read ban on %q in %q: %w", subjectDID, communityDID, err) + } + return standing, nil +} diff --git a/internal/store/interfaces.go b/internal/store/interfaces.go index 1fb4642..0953ffa 100644 --- a/internal/store/interfaces.go +++ b/internal/store/interfaces.go @@ -227,6 +227,40 @@ type APActors interface { UpdateProfile(ctx context.Context, did string, profile APActorProfile) error } +// CommunityBans is one community's exclusion of one native author (migration +// 027) — the durable half of an announced Block. +// +// It is its OWN interface, held only by the two places that need it: the +// announced-moderation dispatch that writes bans, and the admission gate that +// reads them. It is not part of Communities, which half the bridge holds to +// resolve follow state and AP group ids. +type CommunityBans interface { + // Ban records the exclusion AND cancels that author's PENDING deliveries to + // that community, in ONE transaction, returning how many were cancelled. + // + // The two are one decision and cannot be separated: the row stops the + // author's NEXT post, the cancellation stops the ones already queued, and a + // row that committed without the cancellation would leave the queue pushing + // a banned author's posts at a community that rejects them until each + // poisons — with nothing to retry, because Lemmy sends the Block once. + // + // Re-banning preserves the original banned_at (a re-delivered Block is the + // same ban twice) while taking the expiry, reason and removeData from the + // new activity, which are the parts a moderator can genuinely re-issue. + Ban(ctx context.Context, ban CommunityBan) (cancelled int64, err error) + + // Lift removes the ban (Undo{Block}), reporting whether one was standing. + // It lifts ONLY the exclusion: content removed under removeData stays + // removed, because Lemmy models restoration as a separate restore_data flag. + Lift(ctx context.Context, communityDID, subjectDID string) (lifted bool, err error) + + // Standing reports whether the author is CURRENTLY banned from the + // community — expiry included, because a lapsed ban must read exactly like + // no ban: Lemmy sends no Undo when a timed ban runs out, so the clock is the + // only thing that ever lifts it. + Standing(ctx context.Context, communityDID, subjectDID string) (bool, error) +} + // Communities tracks the AP groups the bridge subscribes to and their // backfill progress. type Communities interface { diff --git a/internal/store/models.go b/internal/store/models.go index c9fe407..78c3faa 100644 --- a/internal/store/models.go +++ b/internal/store/models.go @@ -175,6 +175,27 @@ type Community struct { FollowAttempts int } +// CommunityBan is one community's exclusion of one native author. +// +// Every field is part of the decision, and two of them are the ones an +// implementation naturally drops: CommunityAPID, without which the delivery +// queue cannot scope a cancellation to this community, and ExpiresAt, without +// which every timed ban becomes permanent. +type CommunityBan struct { + CommunityDID string + SubjectDID string + CommunityAPID string + // ExpiresAt is nil for a permanent ban. Lemmy sends no activity when a + // timed one lapses, so this is the only thing that ever lifts it. + ExpiresAt *time.Time + Reason string + // RemoveData records what the moderator asked for — that the author's + // content in this community go too. It is acted on ONCE, when the ban lands; + // the stored flag is the audit answer to "was their content purged?", never + // an input to the Undo. + RemoveData bool +} + // ServiceKey is one of the bridge's own long-lived keys, keyed by purpose // name. KeyMaterial's encoding is per-row: plaintext PKCS#8 PEM for // "service-actor" (the AP-side RSA signing key — the bridge's own service diff --git a/internal/store/outbound_deliveries.go b/internal/store/outbound_deliveries.go index 0393401..497a3a6 100644 --- a/internal/store/outbound_deliveries.go +++ b/internal/store/outbound_deliveries.go @@ -250,6 +250,44 @@ func (r *postgresOutboundDeliveries) CancelForCommunity(ctx context.Context, ord return r.cancel(ctx, "cancel outbound_deliveries for community", query, orderingKey) } +// cancelPendingForActorInCommunity parks one actor's PENDING deliveries on ONE +// community's ordering key — the INTERSECTION its two neighbours above cannot +// express, and the shape a community ban needs. +// +// Both one-dimensional versions are wrong for a ban, in opposite directions: +// CancelForActor stops the author in every community they write to, and +// CancelForCommunity stops every author in this one. The ordering key is what +// makes the intersection reachable at all — co-hosted communities SHARE an +// inbox, so target_inbox says nothing about which community's traffic a row +// carries. +// +// Terminal rows are untouched, as everywhere else here: `delivered` cannot be +// un-sent (and the vote reseed subtracts exactly that state from the API tally, +// so rewriting it would move a number the user sees), and `poisoned` is an +// operator surface whose redrive is the recovery. +// +// It runs on a caller's transaction: the ban row and this cancellation are one +// decision, and store.CommunityBans.Ban commits them together. +func cancelPendingForActorInCommunity(ctx context.Context, ex execer, actorDID, orderingKey string) (int64, error) { + result, err := ex.ExecContext(ctx, ` + UPDATE outbound_deliveries d + SET state = 'cancelled', claimed_until = NULL, updated_at = now() + FROM outbound_activities a + WHERE d.activity_id = a.activity_id + AND a.actor_did = $1 + AND d.ordering_key = $2 + AND d.state = 'pending'`, actorDID, orderingKey) + if err != nil { + return 0, fmt.Errorf("cancel outbound_deliveries for %q in %q: %w", actorDID, orderingKey, err) + } + affected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("cancel outbound_deliveries for %q in %q: rows affected: %w", + actorDID, orderingKey, err) + } + return affected, nil +} + func (r *postgresOutboundDeliveries) cancel(ctx context.Context, op, query string, arg string) (int64, error) { result, err := r.db.ExecContext(ctx, query, arg) if err != nil {