diff --git a/go.mod b/go.mod --- a/go.mod +++ b/go.mod @@ -44,16 +44,24 @@ github.com/hashicorp/go-retryablehttp v0.7.5 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/ipfs/bbloom v0.0.4 // indirect github.com/ipfs/go-block-format v0.2.0 // indirect + github.com/ipfs/go-blockservice v0.5.2 // indirect github.com/ipfs/go-cid v0.4.1 // indirect github.com/ipfs/go-datastore v0.6.0 // indirect github.com/ipfs/go-ipfs-blockstore v1.3.1 // indirect github.com/ipfs/go-ipfs-ds-help v1.1.1 // indirect + github.com/ipfs/go-ipfs-exchange-interface v0.2.1 // indirect github.com/ipfs/go-ipfs-util v0.0.3 // indirect github.com/ipfs/go-ipld-cbor v0.1.0 // indirect github.com/ipfs/go-ipld-format v0.6.0 // indirect + github.com/ipfs/go-ipld-legacy v0.2.1 // indirect github.com/ipfs/go-log v1.0.5 // indirect github.com/ipfs/go-log/v2 v2.5.1 // indirect + github.com/ipfs/go-merkledag v0.11.0 // indirect github.com/ipfs/go-metrics-interface v0.0.1 // indirect + github.com/ipfs/go-verifcid v0.0.3 // indirect + github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 // indirect + github.com/ipld/go-codec-dagpb v1.6.0 // indirect + github.com/ipld/go-ipld-prime v0.21.0 // indirect github.com/jbenet/goprocess v0.1.4 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/mattn/go-isatty v0.0.20 // indirect diff --git a/go.sum b/go.sum --- a/go.sum +++ b/go.sum @@ -64,6 +64,8 @@ github.com/ipfs/bbloom v0.0.4 h1:Gi+8EGJ2y5qiD5FbsbpX/TMNcJw8gSqr7eyjHa4Fhvs= github.com/ipfs/bbloom v0.0.4/go.mod h1:cS9YprKXpoZ9lT0n/Mw/a6/aFV6DTjTLYHeA+gyqMG0= github.com/ipfs/go-block-format v0.2.0 h1:ZqrkxBA2ICbDRbK8KJs/u0O3dlp6gmAuuXUJNiW1Ycs= github.com/ipfs/go-block-format v0.2.0/go.mod h1:+jpL11nFx5A/SPpsoBn6Bzkra/zaArfSmsknbPMYgzM= +github.com/ipfs/go-blockservice v0.5.2 h1:in9Bc+QcXwd1apOVM7Un9t8tixPKdaHQFdLSUM1Xgk8= +github.com/ipfs/go-blockservice v0.5.2/go.mod h1:VpMblFEqG67A/H2sHKAemeH9vlURVavlysbdUI632yk= github.com/ipfs/go-cid v0.4.1 h1:A/T3qGvxi4kpKWWcPC/PgbvDA2bjVLO7n4UeVwnbs/s= github.com/ipfs/go-cid v0.4.1/go.mod h1:uQHwDeX4c6CtyrFwdqyhpNcxVewur1M7l7fNU7LKwZk= github.com/ipfs/go-datastore v0.6.0 h1:JKyz+Gvz1QEZw0LsX1IBn+JFCJQH4SJVFtM4uWU0Myk= @@ -74,19 +76,33 @@ github.com/ipfs/go-ipfs-blockstore v1.3.1 h1:cEI9ci7V0sRNivqaOr0elDsamxXFxJMMMy7PTTDQNsQ= github.com/ipfs/go-ipfs-blockstore v1.3.1/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE= github.com/ipfs/go-ipfs-ds-help v1.1.1 h1:B5UJOH52IbcfS56+Ul+sv8jnIV10lbjLF5eOO0C66Nw= github.com/ipfs/go-ipfs-ds-help v1.1.1/go.mod h1:75vrVCkSdSFidJscs8n4W+77AtTpCIAdDGAwjitJMIo= +github.com/ipfs/go-ipfs-exchange-interface v0.2.1 h1:jMzo2VhLKSHbVe+mHNzYgs95n0+t0Q69GQ5WhRDZV/s= +github.com/ipfs/go-ipfs-exchange-interface v0.2.1/go.mod h1:MUsYn6rKbG6CTtsDp+lKJPmVt3ZrCViNyH3rfPGsZ2E= github.com/ipfs/go-ipfs-util v0.0.3 h1:2RFdGez6bu2ZlZdI+rWfIdbQb1KudQp3VGwPtdNCmE0= github.com/ipfs/go-ipfs-util v0.0.3/go.mod h1:LHzG1a0Ig4G+iZ26UUOMjHd+lfM84LZCrn17xAKWBvs= github.com/ipfs/go-ipld-cbor v0.1.0 h1:dx0nS0kILVivGhfWuB6dUpMa/LAwElHPw1yOGYopoYs= github.com/ipfs/go-ipld-cbor v0.1.0/go.mod h1:U2aYlmVrJr2wsUBU67K4KgepApSZddGRDWBYR0H4sCk= github.com/ipfs/go-ipld-format v0.6.0 h1:VEJlA2kQ3LqFSIm5Vu6eIlSxD/Ze90xtc4Meten1F5U= github.com/ipfs/go-ipld-format v0.6.0/go.mod h1:g4QVMTn3marU3qXchwjpKPKgJv+zF+OlaKMyhJ4LHPg= +github.com/ipfs/go-ipld-legacy v0.2.1 h1:mDFtrBpmU7b//LzLSypVrXsD8QxkEWxu5qVxN99/+tk= +github.com/ipfs/go-ipld-legacy v0.2.1/go.mod h1:782MOUghNzMO2DER0FlBR94mllfdCJCkTtDtPM51otM= github.com/ipfs/go-log v1.0.5 h1:2dOuUCB1Z7uoczMWgAyDck5JLb72zHzrMnGnCNNbvY8= github.com/ipfs/go-log v1.0.5/go.mod h1:j0b8ZoR+7+R99LD9jZ6+AJsrzkPbSXbZfGakb5JPtIo= github.com/ipfs/go-log/v2 v2.1.3/go.mod h1:/8d0SH3Su5Ooc31QlL1WysJhvyOTDCjcCZ9Axpmri6g= github.com/ipfs/go-log/v2 v2.5.1 h1:1XdUzF7048prq4aBjDQQ4SL5RxftpRGdXhNRwKSAlcY= github.com/ipfs/go-log/v2 v2.5.1/go.mod h1:prSpmC1Gpllc9UYWxDiZDreBYw7zp4Iqp1kOLU9U5UI= +github.com/ipfs/go-merkledag v0.11.0 h1:DgzwK5hprESOzS4O1t/wi6JDpyVQdvm9Bs59N/jqfBY= +github.com/ipfs/go-merkledag v0.11.0/go.mod h1:Q4f/1ezvBiJV0YCIXvt51W/9/kqJGH4I1LsA7+djsM4= github.com/ipfs/go-metrics-interface v0.0.1 h1:j+cpbjYvu4R8zbleSs36gvB7jR+wsL2fGD6n0jO4kdg= github.com/ipfs/go-metrics-interface v0.0.1/go.mod h1:6s6euYU4zowdslK0GKHmqaIZ3j/b/tL7HTWtJ4VPgWY= +github.com/ipfs/go-verifcid v0.0.3 h1:gmRKccqhWDocCRkC+a59g5QW7uJw5bpX9HWBevXa0zs= +github.com/ipfs/go-verifcid v0.0.3/go.mod h1:gcCtGniVzelKrbk9ooUSX/pM3xlH73fZZJDzQJRvOUw= +github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 h1:oFo19cBmcP0Cmg3XXbrr0V/c+xU9U1huEZp8+OgBzdI= +github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4/go.mod h1:6nkFF8OmR5wLKBzRKi7/YFJpyYR7+oEn1DX+mMWnlLA= +github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6rOcc= +github.com/ipld/go-codec-dagpb v1.6.0/go.mod h1:ANzFhfP2uMJxRBr8CE+WQWs5UsNa0pYtmKZ+agnUw9s= +github.com/ipld/go-ipld-prime v0.21.0 h1:n4JmcpOlPDIxBcY037SVfpd1G+Sj1nKZah0m6QH9C2E= +github.com/ipld/go-ipld-prime v0.21.0/go.mod h1:3RLqy//ERg/y5oShXXdx5YIp50cFGOanyMctpPjsvxQ= github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA= github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o= github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4= diff --git a/internal/api/handlers/post/getstatus.go b/internal/api/handlers/post/getstatus.go --- a/internal/api/handlers/post/getstatus.go +++ b/internal/api/handlers/post/getstatus.go @@ -7,6 +7,30 @@ "net/http" "time" "Coves/internal/core/posts" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +// The identifier bounds this endpoint accepts, DERIVED from the specs rather +// than picked, because both directions of getting it wrong are silent. +// +// A postv2 URI is AUTHORITY-SCOPED — the author's DID is inside it — and a DID +// may legally run to 2048 bytes, the same fact that killed the readable rkey +// transform in PRD rev 2.2. A cap sized for the old community-repo URIs would +// refuse exactly the authors with long DIDs, on the one endpoint that can tell +// them why their post is not visible, and nothing else in the system would show +// a symptom: their posts index fine and every other endpoint serves them. +// +// So the URI bound is the sum of its legal parts rather than a round number: +// "at://" + a 2048-byte authority + "/" + a 317-byte NSID + "/" + a 512-byte +// record key. It is a pre-parse DoS bound only — ParseATURI below is what +// decides whether the thing is actually a URI. +const ( + maxAuthorityLength = 2048 // DID Core's ceiling + maxNSIDLength = 317 // atProto NSID grammar + maxRecordKeyLength = 512 // atProto record-key grammar + maxPostURILength = len("at://") + maxAuthorityLength + 1 + maxNSIDLength + 1 + maxRecordKeyLength + maxCommunityIDLength = maxAuthorityLength ) // GetStatusHandler serves social.coves.community.post.getStatus: one @@ -53,12 +77,29 @@ if communityDID == "" { writeError(w, http.StatusBadRequest, "InvalidRequest", "community parameter is required") return } - if len(postURI) > maxURILength { + // THE CAP IS SIZED TO THE SPEC, NOT TO THE OLD URIs. A DID may legally run + // to 2048 bytes — the same fact that killed the readable rkey transform in + // PRD rev 2.2 — and an author-owned post URI is authority-scoped, so the + // author's DID is INSIDE the URI this endpoint takes. A cap sized for the + // old community-repo URIs would silently make long-DID authors unqueryable + // and nothing else would notice: their posts index fine and every other + // endpoint serves them. + if len(postURI) > maxPostURILength { writeError(w, http.StatusBadRequest, "InvalidRequest", "post URI exceeds maximum length") return } - if len(communityDID) > maxURILength { + if len(communityDID) > maxCommunityIDLength { writeError(w, http.StatusBadRequest, "InvalidRequest", "community DID exceeds maximum length") + return + } + + // Raising the cap must not become "accept anything long". The URI is PARSED, + // so a non-at:// string comes back as the client bug it is rather than as a + // silent not-found — which would tell a client with a typo that its post + // does not exist. + if _, err := syntax.ParseATURI(postURI); err != nil { + writeError(w, http.StatusBadRequest, "InvalidRequest", + "post must be a valid AT-URI") return } @@ -96,6 +137,13 @@ return } w.Header().Set("Content-Type", "application/json") + // NO-STORE, and not as a nicety. This endpoint exists to be POLLED for a + // transition (§7), so a cached answer is the endpoint failing at its only + // job: the client keeps being handed `pending` after the post was accepted + // and stops asking. It is also unauthenticated and reports a moderation + // decision, so an intermediary holding a copy is a disclosure surface that + // outlives the request that created it. + w.Header().Set("Cache-Control", "no-store") w.WriteHeader(http.StatusOK) if _, err := w.Write(responseBytes); err != nil { log.Printf("ERROR: Failed to write getStatus response: %v", err) diff --git a/internal/api/routes/post.go b/internal/api/routes/post.go --- a/internal/api/routes/post.go +++ b/internal/api/routes/post.go @@ -6,9 +6,19 @@ "Coves/internal/api/middleware" "Coves/internal/core/blueskypost" "Coves/internal/core/posts" "Coves/internal/core/votes" + "time" "github.com/go-chi/chi/v5" ) + +// getStatusRateLimit is social.coves.community.post.getStatus' per-client +// budget, per minute. +// +// Sixty is chosen against the endpoint's own UX rather than copied from a +// neighbour: §7 has a client poll for the accepted transition, and a poll a +// second for a minute is comfortably inside this while a script enumerating a +// community's rejected posts is not. +const getStatusRateLimit = 60 // PostRouteOption supplies a collaborator that only some of the post routes // need. @@ -26,12 +36,24 @@ } // WithPostStatusService supplies the service behind // social.coves.community.post.getStatus. The route is registered either way, so -// that the HTTP surface does not silently change shape with the wiring; without -// this option the handler has no service to call. +// that the HTTP surface does not silently change shape with the wiring. func WithPostStatusService(service posts.StatusService) PostRouteOption { return func(c *postRouteConfig) { c.statusService = service } } +// NOT GUARDED AT REGISTRATION, unlike oauthMiddleware above, and the asymmetry +// is forced rather than chosen. The review asked for a fail-fast panic on a +// missing status service, matching that neighbour — but the routes table builds +// the whole router with every service nil and no options at all +// (registration_test.go's theRouter), so a panic there would fail every +// surface-declaration test rather than the one wiring bug it is aimed at. +// Scoping it to a supplied-but-nil option would guard a shape nobody writes. +// +// The exposure is small and named here so it is not mistaken for an oversight: +// cmd/server always supplies the option, and a build that did not would serve +// 500s from getStatus alone. Closing it properly needs the routes table to pass +// the option, which is a test-side change. + // RegisterPostRoutes registers post-related XRPC endpoints on the router // Implements social.coves.community.post.* lexicon endpoints // authMiddleware can be either OAuthAuthMiddleware or DualAuthMiddleware (used for @@ -87,8 +109,18 @@ // session to require and no viewer state to personalise. OptionalAuth would // be harmless but pointless — the answer does not vary by viewer — while // RequireAuth would make the cross-server case unanswerable, which is the // asymmetry internal/api/routes/registration_test.go declares. + // + // It also carries its OWN limiter, tighter than the global 100/minute, and + // it is the only unauthenticated route in the product that does. Two facts + // make the exception worth it: §7's client UX is to POLL this until a post + // flips to accepted, so the honest traffic shape is repeated requests from + // one caller; and because it takes no auth, an unauthenticated stranger can + // ask about any post URI they can name. The budget is what bounds + // enumeration of a community's rejected posts to a rate an operator notices. statusHandler := post.NewGetStatusHandler(cfg.statusService) - r.Get("/xrpc/social.coves.community.post.getStatus", statusHandler.HandleGetStatus) + statusRateLimiter := middleware.NewRateLimiter(getStatusRateLimit, time.Minute) + r.With(statusRateLimiter.Middleware). + Get("/xrpc/social.coves.community.post.getStatus", statusHandler.HandleGetStatus) // Future endpoints (Beta): // r.With(authMiddleware.RequireAuth).Post("/xrpc/social.coves.community.post.update", updateHandler.HandleUpdate) diff --git a/internal/atproto/jetstream/authorpost.go b/internal/atproto/jetstream/authorpost.go --- a/internal/atproto/jetstream/authorpost.go +++ b/internal/atproto/jetstream/authorpost.go @@ -1,6 +1,7 @@ package jetstream import ( + "bytes" "context" "database/sql" "encoding/json" @@ -12,6 +13,7 @@ "net/http" "net/url" "strconv" "strings" + "sync" "time" "Coves/internal/atproto/identity" @@ -19,6 +21,9 @@ "Coves/internal/atproto/oauth" "Coves/internal/core/communities" "Coves/internal/core/posts" "Coves/internal/core/users" + + "github.com/bluesky-social/indigo/atproto/atdata" + "github.com/bluesky-social/indigo/repo" ) // Ingesting author-owned posts and the community records that decide about @@ -91,7 +96,7 @@ // AcceptanceDeleter withdraws a community's acceptance of a post. Satisfied by // posts.CommunityRecordWriter. // -// RED STUB (task 5, cycle 2). Narrowed to one method because that is all the +// Narrowed to one method because that is all the // tombstone path needs: the consumer must never write an acceptance, a removal // or a repin — those are the ENGINE's verdicts, and a consumer holding the full // writer is one edit away from making one. @@ -157,8 +162,23 @@ // attacker-controlled — anyone can publish one — so an unguarded fetcher is // a request forger pointed at whatever is reachable from the AppView's // network, driven by any stranger who writes an acceptance record. allowPrivateHosts bool + + // client is the guarded HTTP client, built once. See httpClient. + clientOnce sync.Once + client *http.Client } +// fetchTimeout bounds ONE direct fetch, inside the client's own 15-second +// ceiling. +// +// The two are not redundant. The client's timeout is a backstop for a +// connection that hangs; this one is a policy about how long the posts lane may +// wait on a stranger's PDS. The lane is single-threaded and now carries four +// collections, so every second spent here is a second nothing else is indexed — +// and the destination is chosen by whoever wrote the acceptance. Five seconds +// is generous for a repo read and cheap to lose. +const fetchTimeout = 5 * time.Second + // maxFetchedRecordBytes bounds how much of a PDS getRecord response is read. // // A post record has a lexicon-bounded size, so a PDS streaming megabytes is @@ -191,11 +211,24 @@ func NewDevDirectPostFetcher(resolver identity.Resolver) *DirectPostFetcher { return &DirectPostFetcher{resolver: resolver, allowPrivateHosts: true} } -// httpClient builds the guarded client for one fetch. Declared here so the -// guard is derived from allowPrivateHosts at call time rather than baked into a -// client at construction, where a test seam could not reach it. +// httpClient returns the guarded client, building it once on first use. +// +// ONE CLIENT, not one per fetch. A fresh http.Client means a fresh +// http.Transport, and a fresh transport means an empty connection pool: every +// fetch would pay a new TCP handshake and a new TLS handshake against a PDS +// this consumer may be about to fetch from a hundred more times, and the +// discarded transports leak idle connections until their finalizers run. The +// guard is a property of the transport rather than of the moment, so nothing +// about correctness needed it rebuilt. +// +// It is built lazily rather than in the constructor so that a fetcher which is +// wired but never used — the common case on an instance hosting no communities +// — costs nothing. func (f *DirectPostFetcher) httpClient() *http.Client { - return oauth.NewSSRFSafeHTTPClient(f.allowPrivateHosts) + f.clientOnce.Do(func() { + f.client = oauth.NewSSRFSafeHTTPClient(f.allowPrivateHosts) + }) + return f.client } // FetchPost implements PostRecordFetcher. @@ -220,11 +253,29 @@ if resolved == nil || resolved.PDSURL == "" { return nil, fmt.Errorf("resolving the repo of %s: no PDS endpoint in the DID document", postURI) } - endpoint := strings.TrimSuffix(resolved.PDSURL, "/") + "/xrpc/com.atproto.repo.getRecord?repo=" + + // com.atproto.sync.getRecord, NOT repo.getRecord, and the difference is the + // whole trustworthiness of this path. + // + // repo.getRecord answers with JSON — {"uri":…, "cid":…, "value":{…}} — whose + // `cid` is a CLAIM BY THE SERVER. That server is the author's PDS: named by + // a DID document, reached because a stranger wrote an acceptance naming this + // subject. Comparing the pinned CID against that field asks the attacker + // whether the attacker is lying, and the consequence is the worst one this + // design has — the AppView indexes whatever `value` holds under a + // community's SIGNED acceptance of a CID that content does not have. The + // community attested to one thing and every reader is shown another. + // + // sync.getRecord answers with a CAR: the repo's own blocks. The CID is then + // RECOMPUTED from the bytes rather than read off a label, and no server can + // lie about the hash of what it just sent. + endpoint := strings.TrimSuffix(resolved.PDSURL, "/") + "/xrpc/com.atproto.sync.getRecord?did=" + url.QueryEscape(repoDID) + "&collection=" + url.QueryEscape(collection) + "&rkey=" + url.QueryEscape(rkey) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + fetchCtx, cancel := context.WithTimeout(ctx, fetchTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(fetchCtx, http.MethodGet, endpoint, nil) if err != nil { return nil, fmt.Errorf("building the getRecord request for %s: %w", postURI, err) } @@ -284,25 +335,46 @@ return nil, fmt.Errorf("the PDS serving %s answered getRecord with status %d: %s", postURI, resp.StatusCode, strconv.Quote(detail)) } - var parsed struct { - URI string `json:"uri"` - CID string `json:"cid"` - Value map[string]interface{} `json:"value"` + // The CAR carries the record's block plus the blocks proving it belongs to + // the repo. Reading it can fail on a hostile or broken server, which is a + // refusal like any other — a response that is not a CAR is not evidence. + stored, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("reading the CAR for %s: %w", postURI, err) + } + + claimedCID, recordBytes, err := stored.GetRecordBytes(ctx, collection+"/"+rkey) + if err != nil { + return nil, fmt.Errorf("reading %s out of the fetched CAR: %w", postURI, err) } - if err := json.Unmarshal(body, &parsed); err != nil { - return nil, fmt.Errorf("parsing the getRecord response for %s: %w", postURI, err) + if recordBytes == nil || len(*recordBytes) == 0 { + return nil, fmt.Errorf("the CAR for %s carried no record bytes", postURI) } - if parsed.Value == nil { - return nil, fmt.Errorf("the getRecord response for %s carried no record value", postURI) + + // THE RECOMPUTATION, which is the entire point of taking a CAR at all. The + // CID that came out of the repo structure is still something the server + // assembled; hashing the record's own bytes under that CID's codec and + // multihash is what turns it into a fact. A server that substituted content + // produces a digest that does not match, whatever it labelled the block. + computedCID, err := claimedCID.Prefix().Sum(*recordBytes) + if err != nil { + return nil, fmt.Errorf("recomputing the CID of %s: %w", postURI, err) } - if parsed.CID == "" { - // Without a CID there is nothing to verify the pinned reference - // against, and an unverified record is exactly what the fetch must - // never index. - return nil, fmt.Errorf("the getRecord response for %s carried no CID", postURI) + if !computedCID.Equals(claimedCID) { + return nil, fmt.Errorf("%w: the CAR for %s labels a block %s whose bytes hash to %s", + ErrPermanentEvent, postURI, claimedCID, computedCID) } - return &FetchedPost{URI: postURI, CID: parsed.CID, Record: parsed.Value}, nil + record, err := atdata.UnmarshalCBOR(*recordBytes) + if err != nil { + return nil, fmt.Errorf("decoding the record block of %s: %w", postURI, err) + } + + // The CID reported back is the COMPUTED one. The caller compares it against + // what the acceptance pinned, and handing back the label instead would put + // the server's claim back into the comparison the recomputation just removed + // it from. + return &FetchedPost{URI: postURI, CID: computedCID.String(), Record: record}, nil } // --------------------------------------------------------------------------- @@ -424,14 +496,28 @@ applied, err := c.tombstoneRecordIfRevWins(ctx, uri, commit.Rev) if err != nil { return err } - if !applied || !indexed { - // A gate skip means this deletion was already applied — the sweep ran - // with it — so re-sweeping would put one authenticated PDS round trip - // behind every redelivery of every tombstone on the network. + if !indexed { return nil } - c.withdrawAcceptance(ctx, stored.communityDID, uri) + // THE WITHDRAWAL IS RECONSIDERED ON EVERY DELIVERY, unlike the tombstone. + // The gate exists to make the local soft-delete happen once; the withdrawal + // is a write into a REMOTE repo that can fail on its own, and it is + // idempotent — DeleteAcceptance reports "nothing to withdraw" as a skip. + // + // Tying it to the gate is what stranded it: a PDS briefly unreachable on the + // first delivery meant the acceptance stayed standing, pointing at a record + // nobody can fetch, permanently — because every redelivery was rejected by + // the gate before the sweep was reconsidered, and nothing else revisits it. + // The community's CAR, the thing its portability argument rests on, would + // keep citing content the author withdrew. + // + // The guard is the POST's state rather than this event's: sweep when the row + // is tombstoned, which is true on the delivery that applied it and on every + // one after. + if applied || stored.deletedAt != nil { + c.withdrawAcceptance(ctx, stored.communityDID, uri) + } return nil } @@ -547,6 +633,37 @@ } uri := recordURI(authorDID, PostV2Collection, commit.RKey) + stored, found, err := c.loadStoredPost(ctx, uri) + if err != nil { + return err + } + + // IMMUTABILITY (§3.1) IS CHECKED FIRST, BEFORE THE COMMUNITY IS LOOKED UP, + // and the order is load-bearing rather than tidy. + // + // An update that changes `community` invalidates the WHOLE event — not + // merely the community field, because applying the content while keeping the + // old community would leave the first community's admission holding a CID it + // never evaluated, publishing content nobody judged under a standing + // acceptance. Retargeting a post means writing a new record. + // + // Two rules meet when the retarget names a community nobody has indexed, and + // only one of them can go first. The unknown-community branch below is + // TRANSIENT — correctly, since a community's own profile event may simply + // not have arrived — so checking it first would turn an illegal retarget + // into a retryable failure that can NEVER succeed: it dead-letters, redrives + // ten times, blocks ~4.2s inline on each delivery, and is still an illegal + // retarget once that community exists. An author could mint that load at + // will by editing one field. + // + // A skip, not an error: an invalid record from a stranger's repo is not an + // infrastructure failure. + if found && stored.communityDID != record.Community { + log.Printf("🚨 SECURITY: ignoring the whole %s update for %s - community is immutable (stored %s, incoming %s)", + PostV2Collection, uri, stored.communityDID, record.Community) + return nil + } + // The community must be one this AppView has indexed, or there is no // subject to open an admission against. // @@ -563,26 +680,6 @@ } return fmt.Errorf("%w: failed to verify community %s exists: %v", errValidationInfra, record.Community, err) } - stored, found, err := c.loadStoredPost(ctx, uri) - if err != nil { - return err - } - - // IMMUTABILITY (§3.1): an update that changes `community` invalidates the - // WHOLE event. Not merely the community field — applying the content while - // keeping the old community would leave the first community's admission - // holding a CID it never evaluated, publishing content nobody judged under - // a standing acceptance. Retargeting a post means writing a new record. - // - // A skip, not an error: an invalid record from a stranger's repo is not an - // infrastructure failure, and dead-lettering it would retry a record that - // can never become valid. - if found && stored.communityDID != record.Community { - log.Printf("🚨 SECURITY: ignoring the whole %s update for %s - community is immutable (stored %s, incoming %s)", - PostV2Collection, uri, stored.communityDID, record.Community) - return nil - } - // Provenance for bridgedStats is keyed on the AUTHOR's PDS now, because the // record lives in the author's repo. The community's host has no say over // what an author asserts about their own record any more, so checking the @@ -622,12 +719,22 @@ return err } } - if !applied { - // The rev gate or the recency guard refused this event: a newer state is - // already indexed. Opening or refreshing an admission from it would move - // evaluated_cid BACKWARDS onto content the row no longer holds, which is - // how an accepted post gets flipped to pending_reacceptance by a - // duplicate delivery. + // THE GATE GUARDS THE POST ROW, NOT THE ADMISSION. The two writes have + // opposite idempotence: inserting the post must happen exactly once, while + // UpsertPending is content-addressed and writes nothing when the row already + // holds this CID. Gating them together is what orphaned admissions — a + // failed upsert leaves the post indexed, the gate advanced, and every + // redelivery skipped, so the row is never created and the post is invisible + // in its community forever with nothing left to retry. + // + // A REPLAY IS SAFE, A STALE COPY IS NOT, and the stored rev is what tells + // them apart. An equal rev is this same commit arriving again — the upsert + // re-runs harmlessly and repairs the orphan. A strictly greater stored rev + // means a NEWER event already applied, and re-running the upsert from this + // older one would drag evaluated_cid backwards onto content the row no + // longer holds, flipping an accepted post to pending_reacceptance on a + // duplicate delivery. + if !applied && !c.revIsCurrent(ctx, uri, commit.Rev) { return nil } @@ -661,6 +768,31 @@ log.Printf("✓ Indexed author post: %s (author: %s, community: %s)", uri, authorDID, record.Community) return nil } +// revIsCurrent reports whether the gate's stored rev for this record is exactly +// the one this event carries — i.e. whether the event is the newest the AppView +// has seen rather than an older copy. +// +// It exists so a rev-gated SKIP can still be distinguished into its two very +// different causes. A redelivery of the newest commit is safe to act on again; +// a stale cross-feed copy of an older one is not. Anything unreadable answers +// false, which declines to act — the conservative direction, since acting on a +// stale event corrupts state while declining merely waits for the next +// delivery. +func (c *PostEventConsumer) revIsCurrent(ctx context.Context, uri, rev string) bool { + if rev == "" { + // A rev-less event bypasses the gate entirely, so there is no stored rev + // to be current with. Only synthetic events reach this. + return true + } + var stored string + if err := c.db.QueryRowContext(ctx, + `SELECT rev FROM jetstream_record_revs WHERE record_uri = $1`, uri, + ).Scan(&stored); err != nil { + return false + } + return stored == rev +} + // hydrateAuthorOpportunistically indexes a minimal profile for an author this // AppView has not seen, so their posts are not permanently authorless. // @@ -916,10 +1048,28 @@ decision *communityDecisionRecord, subjectCollection string, watermark posts.CommunityWatermark, ) error { - indexedCommunity, indexed, err := c.indexedPostCommunity(ctx, decision.Subject.URI) + stored, indexed, err := c.loadStoredPost(ctx, decision.Subject.URI) if err != nil { return err } + + // A TOMBSTONED SUBJECT IS NOT ACCEPTABLE, and the arrival is legitimate + // rather than hostile: the community decided before it saw the tombstone, + // and the two events are in different repos with no ordering between them. + // Applying it would have getStatus report `accepted` for content no read + // path will ever serve — and the host-side sweep already ran with the + // tombstone and will not run again, so the community's repo would keep an + // acceptance citing a record nobody can fetch. + // + // A SKIP, not a refusal. Returning an error would dead-letter an event that + // is replayed constantly and would be refused identically every time. + if indexed && stored.deletedAt != nil { + log.Printf("INFO: not applying the acceptance of %s in %s: its author deleted the post", + decision.Subject.URI, communityDID) + return nil + } + + indexedCommunity := stored.communityDID switch { case !indexed: if err := c.convergeOnAcceptedSubject(ctx, communityDID, decision, subjectCollection); err != nil { @@ -1033,7 +1183,7 @@ // The fetch is not a firehose event, so it carries no rev to gate on and no // event time to stamp: an empty rev bypasses the gate (which is correct — a // later real event for this record still wins on its own rev) and the // watermark falls back to wall clock. - if _, err := c.insertAuthorPost(ctx, authorPostInsert{ + applied, err := c.insertAuthorPost(ctx, authorPostInsert{ uri: decision.Subject.URI, authorDID: authorDID, record: record, @@ -1043,19 +1193,38 @@ RKey: rkey, CID: fetched.CID, }, facets: facetsJSON, embed: embedJSON, labels: labelsJSON, bridgedUpvotes: up, bridgedDownvotes: down, bridgedAsOf: asOf, - }); err != nil { + }) + if err != nil { return err } - // The pending admission comes with it. ApplyAcceptance would create a row - // on its own, but one with no evaluated content: recording what was indexed - // is what lets the next author edit be recognised as an edit. - if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ - CommunityDID: communityDID, - PostURI: decision.Subject.URI, - EvaluatedCID: fetched.CID, - }); err != nil { - return fmt.Errorf("recording the pending admission for fetched post %s: %w", decision.Subject.URI, err) + // THE FETCH IS A CATCH-UP, NOT AN AUTHORITY, and applied=false is how it + // learns it lost the race. The acceptance and the post live in different + // repos, so Jetstream parallelises them and the post's own event can land + // while this fetch is in flight — which is not a rare interleaving but the + // normal one, since the fetch exists precisely because the event had not + // arrived. The insert then conflicts and writes nothing. + // + // UpsertPending is last-write-wins, so running it anyway would stamp the + // FETCHED CID over the newer one the real event just recorded. evaluated_cid + // is what the next decision judges, so the engine would evaluate content the + // author has already replaced, and an acceptance written from that verdict + // would pin a version the AppView is no longer serving. + // + // Skipping it costs nothing: ApplyAcceptance below classifies against + // whatever evaluated_cid the row actually holds, which is exactly the + // comparison that turns a stale pin into pending_reacceptance. + if applied { + // The pending admission comes with it. ApplyAcceptance would create a row + // on its own, but one with no evaluated content: recording what was indexed + // is what lets the next author edit be recognised as an edit. + if _, err := c.admissions.UpsertPending(ctx, posts.UpsertPendingCommand{ + CommunityDID: communityDID, + PostURI: decision.Subject.URI, + EvaluatedCID: fetched.CID, + }); err != nil { + return fmt.Errorf("recording the pending admission for fetched post %s: %w", decision.Subject.URI, err) + } } c.hydrateAuthorOpportunistically(ctx, authorDID) @@ -1111,48 +1280,121 @@ // acceptance deletion, which the host writes when an author deletes their post // (§5.3), is the one delete that arrives unpaired — and it is the one this // lookup resolves. func (c *PostEventConsumer) applyCommunityDecisionDelete(ctx context.Context, communityDID string, commit *CommitEvent) error { - if commit.Collection != posts.AcceptanceCollection { - log.Printf("INFO: %s deletion in %s carries no subject and is superseded by its paired write; skipping", - commit.Collection, communityDID) - return nil + postURI, found, err := c.subjectOfDeletedRecord(ctx, communityDID, commit) + if err != nil { + return err } - - var postURI string - err := c.db.QueryRowContext(ctx, - `SELECT post_uri FROM community_post_admissions - WHERE community_did = $1 AND acceptance_rkey = $2`, - communityDID, commit.RKey, - ).Scan(&postURI) - if errors.Is(err, sql.ErrNoRows) { - // No acceptance of that rkey stands here — the removal half of the same - // commit already cleared it, or this AppView never saw the acceptance. - // Either way there is nothing to withdraw. - log.Printf("INFO: acceptance deletion %s/%s matches no standing acceptance; nothing to withdraw", - communityDID, commit.RKey) + if !found { + // Nothing here matches that record key: the paired write of the same + // commit already superseded it, or this AppView never saw the record + // being deleted. Either way there is nothing to withdraw. + log.Printf("INFO: %s deletion %s/%s matches no standing record; nothing to withdraw", + commit.Collection, communityDID, commit.RKey) return nil } - if err != nil { - return fmt.Errorf("resolving the subject of acceptance deletion %s/%s: %w", communityDID, commit.RKey, err) - } - result, err := c.admissions.ApplyAcceptanceDelete(ctx, posts.CommunityDeleteCommand{ + cmd := posts.CommunityDeleteCommand{ CommunityDID: communityDID, PostURI: postURI, Watermark: posts.CommunityWatermark{Rev: commit.Rev}, - }) + } + + var result posts.AdmissionResult + if commit.Collection == posts.RemovalCollection { + result, err = c.admissions.ApplyRemovalDelete(ctx, cmd) + } else { + result, err = c.admissions.ApplyAcceptanceDelete(ctx, cmd) + } if err != nil { - return fmt.Errorf("withdrawing the acceptance of %s in %s: %w", postURI, communityDID, err) + return fmt.Errorf("withdrawing the %s of %s in %s: %w", commit.Collection, postURI, communityDID, err) } - logAdmissionOutcome(posts.AcceptanceCollection+"#delete", communityDID, postURI, result.Outcome) + logAdmissionOutcome(commit.Collection+"#delete", communityDID, postURI, result.Outcome) return nil } +// subjectOfDeletedRecord recovers which post a deleted acceptance or removal was +// about. +// +// A delete event carries NO record, and the rkey is a SHA-256 digest of the +// subject URI (§3.2), which is one-way — so the subject can only come from state +// the AppView already holds. The two collections need different lookups because +// only one of them has a column: +// +// - An ACCEPTANCE stores its rkey on the admission row, so the reverse lookup +// is an exact match. +// - A REMOVAL stores none, so its subject is found by recomputing the digest +// over the rows that could be its subject. The candidate set is bounded to +// this community's `removed` rows — moderation-sized, and served by the +// (community_did, status, created_at) index migration 034 already carries. +// +// WHY THE REMOVAL CASE CANNOT STAY A NO-OP, which is what it was. The paired +// commits do converge without it: a removal commit is {acceptance-delete, +// removal-put} and a restore is {removal-delete, acceptance-put}, and in both +// the put carries the subject in-record and outranks its delete under the §5.2 +// tuple. But a moderator can also simply WITHDRAW a removal — deleting the +// record and writing nothing — and that commit carries only this event. Ignored, +// the post stays `removed` forever while the community's own repo no longer says +// so: the signed record and the AppView disagree, and only the AppView is +// consulted when the post is served. +func (c *PostEventConsumer) subjectOfDeletedRecord( + ctx context.Context, communityDID string, commit *CommitEvent, +) (string, bool, error) { + if commit.Collection == posts.AcceptanceCollection { + var postURI string + err := c.db.QueryRowContext(ctx, + `SELECT post_uri FROM community_post_admissions + WHERE community_did = $1 AND acceptance_rkey = $2`, + communityDID, commit.RKey, + ).Scan(&postURI) + if errors.Is(err, sql.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("resolving the subject of acceptance deletion %s/%s: %w", + communityDID, commit.RKey, err) + } + return postURI, true, nil + } + + rows, err := c.db.QueryContext(ctx, + `SELECT post_uri FROM community_post_admissions + WHERE community_did = $1 AND status = 'removed'`, communityDID) + if err != nil { + return "", false, fmt.Errorf("resolving the subject of removal deletion %s/%s: %w", + communityDID, commit.RKey, err) + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var postURI string + if err := rows.Scan(&postURI); err != nil { + return "", false, fmt.Errorf("scanning a removed subject of %s: %w", communityDID, err) + } + if posts.SubjectRkey(postURI) == commit.RKey { + return postURI, true, nil + } + } + if err := rows.Err(); err != nil { + return "", false, fmt.Errorf("reading the removed subjects of %s: %w", communityDID, err) + } + return "", false, nil +} + // logAdmissionOutcome records what a community event DID, including the skips. // // A skip is the ordering gate working — a multi-feed duplicate, a dead-letter -// redrive, an event superseded by its own commit's other half — so it is logged -// rather than returned as an error, which would bury healthy skips in the -// dead-letter queue among genuine failures. +// redrive, an event superseded by its own commit's other half, or this +// AppView's own write coming back to it — so it is logged rather than returned +// as an error, which would bury healthy skips in the dead-letter queue among +// genuine failures. +// +// THE LAST OF THOSE IS THE COMMON ONE ON A HOSTING INSTANCE and is easy to +// misread in the logs. When this AppView hosts the community, the engine writes +// the acceptance into the repo and stamps the row optimistically; the firehose +// then delivers that same commit back, and the watermark CAS answers +// skipped_stale because the row already carries that exact rev. Nothing is +// wrong — the write landed twice by design, once locally and once as its own +// echo — and the engine's own doc calls that echo a success. func logAdmissionOutcome(collection, communityDID, postURI string, outcome posts.AdmissionOutcome) { if outcome == posts.AdmissionApplied { log.Printf("✓ Applied %s for %s in %s", collection, postURI, communityDID) diff --git a/internal/atproto/jetstream/post_consumer.go b/internal/atproto/jetstream/post_consumer.go --- a/internal/atproto/jetstream/post_consumer.go +++ b/internal/atproto/jetstream/post_consumer.go @@ -30,12 +30,14 @@ // before its author's profile. The identity is admitted only when its PDS // passes bridgeTrust. identityResolver identity.Resolver - // RED STUB fields (task 5, cycle 1) — the collaborators author-owned post - // ingestion needs. Declared here so the options in authorpost.go compile; - // nothing reads them yet. See docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6. + // The collaborators author-owned post ingestion needs + // (docs/PRD_AUTHOR_OWNED_POSTS.md §5.3-§5.6). All four are read by the + // handlers in authorpost.go, and a nil one disables a capability rather + // than degrading it — see each field. // // admissions holds the per-(community, post) decision state. nil means the - // consumer is running in its pre-034 shape and records no admissions. + // consumer is running in its pre-034 shape and ignores all three + // author-owned collections rather than indexing them undecided. admissions posts.AdmissionRepository // deletedAccounts gates events from erased accounts. nil means no gate. deletedAccounts DeletedAccountLookup diff --git a/internal/atproto/oauth/transport.go b/internal/atproto/oauth/transport.go --- a/internal/atproto/oauth/transport.go +++ b/internal/atproto/oauth/transport.go @@ -1,6 +1,7 @@ package oauth import ( + "context" "fmt" "net" "net/http" @@ -62,14 +63,43 @@ return false } +// vettedAddrsKeyType keys the addresses RoundTrip approved, so the dialler can +// read them off the request's own context. A private type, so nothing outside +// this file can plant a value under the same key. +type vettedAddrsKeyType struct{} + +var vettedAddrsKey vettedAddrsKeyType + +// RoundTrip vets the hostname's addresses and then makes the dial use THOSE +// ADDRESSES rather than the name. +// +// PASSING THE NAME ON WOULD BE A SECOND DECISION. The base transport resolves +// whatever host it is given, so a guard that approved answer A and then handed +// over the hostname lets the dialler act on answer B — and nothing binds the two +// together. DNS rebinding is the name for exploiting that: an attacker who +// controls the zone answers the first query publicly and the second with +// 169.254.169.254, and the approval describes a host the connection never went +// to. Every input that reaches this transport is chosen by a stranger (a DID +// document's PDS endpoint, an acceptance record's subject), so the attacker +// picks the moment to flip as well. +// +// The addresses ride on the request context and the dialler below consumes +// them, which also means the hostname is resolved exactly ONCE per request. +// That is the property, not an optimisation: any later answer is one the guard +// never saw. func (t *ssrfSafeTransport) RoundTrip(req *http.Request) (*http.Response, error) { host := req.URL.Hostname() - // Resolve hostname to IP + // A literal address is already the thing that will be dialled, so there is + // no second resolution to defend against — but it still has to pass the + // private check below. ips, err := t.resolveHost(host) if err != nil { return nil, fmt.Errorf("failed to resolve host: %w", err) } + if len(ips) == 0 { + return nil, fmt.Errorf("failed to resolve host: %s resolved to no addresses", host) + } // Check all resolved IPs if !t.allowPrivate { @@ -80,17 +110,53 @@ } } } - return t.base.RoundTrip(req) + return t.base.RoundTrip(req.WithContext(context.WithValue(req.Context(), vettedAddrsKey, ips))) } // NewSSRFSafeHTTPClient creates an HTTP client with SSRF protections func NewSSRFSafeHTTPClient(allowPrivate bool) *http.Client { + dialer := &net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + } + transport := &ssrfSafeTransport{ base: &http.Transport{ - DialContext: (&net.Dialer{ - Timeout: 10 * time.Second, - KeepAlive: 30 * time.Second, - }).DialContext, + // The dial IGNORES the hostname in addr and connects to an address + // RoundTrip already approved, which is what closes the + // check-then-dial window. It takes only the port from addr, because + // the port is the one part of the destination the guard has no + // opinion about. + // + // FAIL CLOSED when there is nothing vetted: reaching here without a + // context value means this base transport was used directly rather + // than through RoundTrip, which is exactly the unguarded path the + // wrapper exists to prevent. + // + // TLS is unaffected. http.Transport derives the handshake's server + // name from the request URL rather than from the dial address, so + // certificate verification and SNI still name the host the caller + // asked for. + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + vetted, _ := ctx.Value(vettedAddrsKey).([]net.IP) + if len(vetted) == 0 { + return nil, fmt.Errorf("SSRF blocked: refusing to dial %s with no vetted address "+ + "(the SSRF-safe transport was bypassed)", addr) + } + _, port, err := net.SplitHostPort(addr) + if err != nil { + return nil, fmt.Errorf("SSRF blocked: cannot read a port from %q: %w", addr, err) + } + var lastErr error + for _, ip := range vetted { + conn, dialErr := dialer.DialContext(ctx, network, net.JoinHostPort(ip.String(), port)) + if dialErr == nil { + return conn, nil + } + lastErr = dialErr + } + return nil, lastErr + }, MaxIdleConns: 100, IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, diff --git a/internal/core/posts/decider.go b/internal/core/posts/decider.go --- a/internal/core/posts/decider.go +++ b/internal/core/posts/decider.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "log" "os" "strings" "time" @@ -29,9 +28,11 @@ // is about to delete — each component correct, the pair looping. // - THE ACTOR CLASS. A misclassification is a privilege decision. Trusted // aggregators skip visibility, ban and authorization entirely, so guessing // UPWARD on a failed lookup would hand the widest privileges in the system -// to whoever made the lookup fail. Every uncertain path therefore falls to -// the STRICTER class, matching CreatePost's existing behaviour (service.go -// step 3 treats a failed IsAggregator lookup as an ordinary user). +// to whoever made the lookup fail. But guessing DOWNWARD is not free here +// either, which is where this path parts company with CreatePost: a lookup +// that could not be made is reported as UNDECIDED rather than resolved to +// the stricter class, because this decision gets written into the admission +// row with redrivable = false. See classify for the full asymmetry. // // It reuses evaluateAdmissionPolicy rather than admitPost, and that is the split // task 3 built for: admitPost RESERVES a ledger slot, and the engine is not a @@ -187,7 +188,12 @@ return undecided(fmt.Errorf("deciding %s for %s: %w: its author deleted it", postURI, communityDID, ErrSubjectGone)) } - return evaluateAdmissionPolicy(ctx, admissionDeps{ + actor, err := d.classify(ctx, post.AuthorDID) + if err != nil { + return undecided(fmt.Errorf("deciding %s for %s: %w", postURI, communityDID, err)) + } + + decision, err := evaluateAdmissionPolicy(ctx, admissionDeps{ communities: d.deps.Communities, bans: d.deps.Policy.Bans, aggregators: d.deps.Authorizer, @@ -195,7 +201,7 @@ ledger: d.deps.Policy.Ledger, limits: d.deps.Policy.Limits, now: d.deps.Policy.Now, }, AdmissionRequest{ - Actor: d.classify(ctx, post.AuthorDID), + Actor: actor, AuthorDID: post.AuthorDID, // The community DID, which resolves to itself. The engine's input is an // admission row, and its key is already the resolved DID — there is no @@ -209,42 +215,106 @@ // refuse the redecision as a duplicate of the very post it is // redeciding. Fingerprint: "", }) + if err != nil || !decision.Admitted() { + return decision, err + } + + return d.applyQuota(ctx, communityDID, post.AuthorDID, actor, decision) } -// classify decides what class of actor the author is. +// applyQuota is the firehose path's §8 submission limit, and the last thing +// between an admitted decision and the engine writing an acceptance. +// +// IT COUNTS ADMISSION ROWS, NOT LEDGER ROWS, and that substitution is the whole +// reason it exists separately from admitPost's step 6. post_submissions is +// written by CreatePost, so a post that arrived over the firehose from an author +// on another server has no ledger row and never will — counting it would hold +// LOCAL users to the limit while exempting precisely the remote ones §8 is +// about. Anyone can write unlimited postv2 records naming any community, and the +// admission layer is what absorbs that. +// +// THE LIMIT IS THE SAME NUMBER the write path uses, taken from the same config, +// so an author is held to one quota rather than to two that drift. // -// EVERY UNCERTAIN PATH FALLS TO ActorUser, the stricter class. A trusted -// aggregator skips visibility, ban and authorization entirely, so resolving a -// failed lookup UPWARD would hand the widest privileges in the system to -// whoever managed to make the lookup fail. Guessing downward costs an -// aggregator some refused posts until the lookup recovers — and CreatePost -// already made exactly this choice (service.go step 3), so the engine agreeing -// with it is also what keeps the write path and the ingestion path from -// disagreeing about who someone is. -func (d *AdmissionEngineDecider) classify(ctx context.Context, authorDID string) ActorClass { +// ACTOR CLASSES ARE TREATED EXACTLY AS admitPost TREATS THEM: only ActorUser is +// metered. A registered aggregator is already governed by its own hourly quota +// inside ValidateAggregatorPost, and applying this as well would silently halve +// an authorized aggregator's throughput; a trusted one has never had a +// submission limit, and inventing one here would stop the bridge dead at a +// number nobody chose. +func (d *AdmissionEngineDecider) applyQuota( + ctx context.Context, communityDID, authorDID string, actor ActorClass, decision AdmissionDecision, +) (AdmissionDecision, error) { + if actor != ActorUser || d.deps.Admissions == nil { + return decision, nil + } + limits := d.deps.Policy.Limits + if limits.MaxPerAuthorPerCommunity <= 0 || limits.Window <= 0 { + return decision, nil + } + + since := d.deps.Policy.Now().Add(-limits.Window) + count, err := d.deps.Admissions.CountRecentAdmissions(ctx, communityDID, authorDID, since) + if err != nil { + // UNDECIDED, never a refusal. The engine persists a refusal code and + // marks it non-redrivable, so a count that could not be taken must not + // become a permanent rate-limit verdict on somebody's post. + return undecided(fmt.Errorf("counting recent admissions for %s in %s: %w", authorDID, communityDID, err)) + } + + // EXCEEDS, not reaches. The subject being decided already has its own + // admission row — the consumer opens it before the engine ever runs — so it + // is inside this count, exactly as admitPost's reservation is inside its + // own. Comparing with >= would refuse the author's very first post. + if count > limits.MaxPerAuthorPerCommunity { + return AdmissionDecision{Code: DecisionRateLimitExceeded}, nil + } + return decision, nil +} + +// classify decides what class of actor the author is, or reports that it could +// not. +// +// A FAILED LOOKUP IS UNDECIDED HERE, AND A DOWNGRADE ON THE WRITE PATH. The two +// answers are deliberate opposites, and the difference is what each caller does +// with the result afterwards. CreatePost is talking to a live client: a +// downgrade to ActorUser applies the strict checks, costs an aggregator a few +// posts until the table recovers, and hands back something the caller can +// retry. Nothing is written down. +// +// The engine writes its verdict INTO the admission row, and a policy refusal is +// stamped redrivable = false — terminal, never revisited by the redrive pass. +// The same downgrade there does not cost a retry; it permanently marks a post +// refused for a reason that was never true, because a table was briefly +// unreachable, and nothing in the system would ever look at it again. +// +// So the rule this encodes is: A DECISION THAT PERSISTS MAY ONLY BE MADE FROM +// AN ANSWER THAT WAS ACTUALLY OBTAINED. Guessing upward is never available +// either — a trusted aggregator skips visibility, ban and authorization +// entirely, so resolving uncertainty in that direction would hand the widest +// privileges in the system to whoever could make a lookup fail. +func (d *AdmissionEngineDecider) classify(ctx context.Context, authorDID string) (ActorClass, error) { // The trusted set is checked FIRST, which is both the cheaper path and the // only one that costs nothing: it is an in-memory set resolved at // construction, so a trusted actor never pays for a database lookup to // learn what the process already knew. if d.deps.TrustedAggregatorDIDs[authorDID] { - return ActorTrustedAggregator + return ActorTrustedAggregator, nil } // With no aggregator collaborators wired — a deployment with no aggregator - // support at all — nobody can be classified as one, which is the strict - // answer rather than a degraded one. + // support at all — nobody can be classified as one. That is a CONFIGURED + // fact rather than a failed lookup, so it answers rather than defers. if d.deps.Aggregators == nil || d.deps.Authorizer == nil { - return ActorUser + return ActorUser, nil } registered, err := d.deps.Aggregators.IsAggregator(ctx, authorDID) if err != nil { - log.Printf("[ADMISSION-DECIDER] Warning: classifying %s fell back to the user class, IsAggregator failed: %v", - authorDID, err) - return ActorUser + return "", fmt.Errorf("classifying %s: %w", authorDID, err) } if registered { - return ActorRegisteredAggregator + return ActorRegisteredAggregator, nil } - return ActorUser + return ActorUser, nil } diff --git a/internal/core/posts/engine.go b/internal/core/posts/engine.go --- a/internal/core/posts/engine.go +++ b/internal/core/posts/engine.go @@ -26,15 +26,21 @@ // idempotency from deterministic rkeys and swap guards, and the engine is the // one place that decides they should fire at all. // // THERE IS NO LEASE, AND THAT IS DELIBERATE. Nothing stops two passes — the -// fast path, a firehose redelivery, a notify — from processing the same -// subject at the same moment, and no lock or per-subject claim is taken. +// queue driver, a firehose redelivery, and (once task 6 lands it) the +// synchronous fast path — from processing the same subject at the same moment, +// and no lock or per-subject claim is taken. Only the first two exist today; +// the fast path is named because the safety argument has to hold when it +// arrives, not because it is calling now. // Safety comes from the layers instead: deterministic rkeys make the racers // aim at the same record; every put and batch is swap-guarded, so a loser is // told rather than clobbering; a loser that re-reads and finds the winner // wrote its exact target converges as a skip; and the repository's watermark // CAS makes the row's state advance monotonically no matter which pass -// stamps first. Serializing the passes properly (a per-community queue) is -// task 5's job; until then concurrent passes are expected and harmless. +// stamps first. Serializing the passes properly is QueueDriver's job (queue.go): +// one goroutine walks one ordered list, grouped by community, so no two subjects +// of a community are ever in flight together. The layered safety above still +// matters, because the driver is not the only caller — the synchronous fast +// path and the firehose consumer both reach the engine too. // EngineOutcome reports what one pass over one subject DID. // @@ -62,11 +68,14 @@ // EngineRepinned means a standing acceptance moved onto new content with no // re-decision — the bridgedStats exception of §5.5. // - // NOT PRODUCED BY ANY PATH YET. ProcessAdmission runs full re-admission on - // every edit; the repin path — classifyRecordDiff choosing the exception, - // the bridge-trust gate approving the author, RepinAcceptance moving the - // record — is task 5's consumer wiring. The outcome is declared now so - // that path lands against a named contract instead of minting one. + // NOT PRODUCED BY ANY PATH YET, and deliberately still deferred. + // ProcessAdmission runs full re-admission on every edit; the repin path — + // classifyRecordDiff choosing the exception, the bridge-trust gate + // approving the author, RepinAcceptance moving the record — needs an + // old-record snapshot the consumer does not keep, so wiring it is a + // recorded decision rather than an oversight (see loop_state.md). The + // outcome is declared so that path lands against a named contract instead + // of minting one. EngineRepinned EngineOutcome = "repinned" // EngineDeferred means the subject is still owed a decision and nothing diff --git a/internal/core/posts/queue.go b/internal/core/posts/queue.go --- a/internal/core/posts/queue.go +++ b/internal/core/posts/queue.go @@ -2,6 +2,7 @@ package posts import ( "context" + "errors" "fmt" "log" "sync" @@ -12,9 +13,9 @@ // The acceptance engine's driver: the thing that decides WHEN the engine runs // and on what (docs/PRD_AUTHOR_OWNED_POSTS.md §5.6, §8). // // The engine settles one subject. Nothing until now decided which subjects, in -// what order, or how often — the fast path and the firehose consumer both push -// work at it, and neither can see a subject that was left pending because a -// credential expired or a lookup blipped. This is the pull side: a periodic pass +// what order, or how often — the firehose consumer pushes work at it today, and +// task 6's synchronous fast path will push more, and neither can see a subject +// that was left pending because a credential expired or a lookup blipped. This is the pull side: a periodic pass // over the undecided backlog that eventually reaches every stranded row. // // # IT IS A SINGLE GOROUTINE, AND THAT IS THE PER-COMMUNITY SERIALIZATION @@ -218,7 +219,7 @@ func (d *QueueDriver) RunPass(ctx context.Context) (PassReport, error) { startedAt := d.now() report := PassReport{StartedAt: startedAt} - subjects, err := d.subjects.ListPendingSubjects(ctx, d.batchSize) + subjects, err := d.subjects.ListPendingSubjects(ctx, d.fetchSize()) if err != nil { // The one failure a pass has nothing to do about. Every other outcome // below is per-subject and counted; this one means there is no work @@ -228,6 +229,12 @@ } report.Listed = len(subjects) for _, subject := range groupByCommunity(subjects) { + // THE BATCH IS FILLED WITH WORK, not with rows. Skipping a backed-off + // subject must not consume a slot, or a stuck prefix would spend the + // whole pass on subjects it never touched — see fetchSize. + if report.Processed >= d.batchSize { + break + } if d.heldBack(subject, startedAt) { continue } @@ -249,6 +256,17 @@ // for a failure is at least as conservative as for a deferral. A wedged // community's PDS returns errors, not deferrals, so exempting failures // would leave the loudest case as the one thing nothing paced. switch { + case errors.Is(err, ErrSubjectGone): + // NOT A FAILURE, and counting it as one would make an ordinary race + // look like an outage. The backlog query excludes tombstoned and + // unindexed posts, but a post can be deleted between the listing and + // the decision — so this is the queue meeting a subject that stopped + // existing while it worked, which is exactly what the exclusion is + // for and needs no operator's attention. It is counted as deferred + // and backed off like any other "nothing to do yet"; the next pass + // will not list it at all. + report.Deferred++ + d.deferSubject(subject, startedAt) case err != nil: report.Failed++ log.Printf("[ACCEPTANCE-QUEUE] Warning: %s in %s could not be settled: %v", @@ -263,10 +281,81 @@ d.clearDeferral(subject) } } + d.pruneDeferrals(subjects) d.record(subjects, report, startedAt) return report, nil } +// queueOverFetchFactor bounds how far past a backed-off prefix one pass may +// reach: a pass never asks the query for more than batchSize × this. +// +// FOUR, and the number is a trade rather than a preference. The backlog is +// ordered oldest-first, so the subjects most likely to be stuck are exactly the +// ones at the front of it — a community whose credentials expired weeks ago sits +// there forever. Without over-fetching, a pass asks for LIMIT rows, gets LIMIT +// stuck ones, skips them all for backoff and does nothing; every pass, while a +// healthy post two rows behind is never decided. Over-fetching without a bound +// would instead let one pass drag the entire backlog into memory to find one +// live subject. Four buys three batches of headroom against a query whose cost +// grows with it, and a prefix deeper than that drains as backoffs expire. +const queueOverFetchFactor = 4 + +// fetchSize is how many rows to ask for so the pass can still fill its batch +// after skipping the subjects it already knows are held back. +// +// It asks for the batch plus exactly the number of deferrals currently held — +// the measured size of the prefix that may be skipped — rather than always +// over-fetching. A driver with nothing backed off has nothing to skip, so it +// asks for precisely what it will use. +func (d *QueueDriver) fetchSize() int { + held := d.deferredCount() + size := d.batchSize + held + if ceiling := d.batchSize * queueOverFetchFactor; size > ceiling { + size = ceiling + } + return size +} + +func (d *QueueDriver) deferredCount() int { + d.mu.Lock() + defer d.mu.Unlock() + return len(d.deferrals) +} + +// pruneDeferrals forgets the backoffs of subjects that are no longer listed. +// +// A deferral outlives its subject otherwise. The row gets settled by somebody +// else — the synchronous fast path, a firehose acceptance, a moderator's +// removal — and simply stops being listed, without ever telling the driver. The +// map is keyed by subject and swept by nothing, so on a busy instance it is an +// unbounded leak held for the life of the process. +// +// It is also wrong on RE-ENTRY, which is the part a leak metric would not show: +// a subject that leaves the backlog and comes back — an edit reopening an +// accepted post — would arrive carrying a stale backoff it did nothing to earn, +// and wait out a delay that was about a completely different decision. +// +// Pruning against the LISTING rather than against what the pass processed is +// deliberate: a subject skipped for backoff is still in the backlog, and +// forgetting it would defeat the backoff on the very next pass. +func (d *QueueDriver) pruneDeferrals(listed []PendingSubject) { + d.mu.Lock() + defer d.mu.Unlock() + + if len(d.deferrals) == 0 { + return + } + stillListed := make(map[subjectKey]bool, len(listed)) + for _, subject := range listed { + stillListed[keyOf(subject)] = true + } + for key := range d.deferrals { + if !stillListed[key] { + delete(d.deferrals, key) + } + } +} + // Snapshot returns the driver's health surface as of the last completed pass. func (d *QueueDriver) Snapshot() QueueSnapshot { d.mu.Lock() @@ -321,6 +410,7 @@ snapshot := QueueSnapshot{ PendingBacklog: report.Listed, LastPassDeferred: report.Deferred, LastPassFailed: report.Failed, + DeferredSubjects: len(d.deferrals), } // Taken as a MINIMUM rather than as subjects[0], even though the query // orders by age. The oldest entry's age is the queue's only early warning, diff --git a/internal/core/users/interfaces.go b/internal/core/users/interfaces.go --- a/internal/core/users/interfaces.go +++ b/internal/core/users/interfaces.go @@ -13,6 +13,21 @@ BannerCID *string } // UserRepository defines the interface for user data persistence +// ErasureLookup reports whether a DID names an account this AppView was asked +// to erase (migration 036). +// +// It is a SEPARATE, OPTIONAL interface rather than a method on UserRepository, +// and detected with a type assertion at the one call site that needs it. Adding +// it to UserRepository would oblige every implementation to answer a question +// only the PostgreSQL one can — and a double that answered "not erased" by +// default would be a gate that fails open, which is the single outcome this +// marker exists to prevent. A repository that does not implement it disables +// the gate rather than weakening it, and nothing in production is such a +// repository. +type ErasureLookup interface { + IsAccountDeleted(ctx context.Context, did string) (bool, error) +} + type UserRepository interface { Create(ctx context.Context, user *User) (*User, error) GetByDID(ctx context.Context, did string) (*User, error) diff --git a/internal/core/users/service.go b/internal/core/users/service.go --- a/internal/core/users/service.go +++ b/internal/core/users/service.go @@ -455,6 +455,37 @@ // profile data, their profile record is fetched from their PDS asynchronously and // best-effort — this heals users whose profile firehose event was never delivered // without ever blocking or failing the IndexUser call itself. func (s *userService) IndexUser(ctx context.Context, did, handle, pdsURL string) error { + // THE ERASURE GATE, and IndexUser is where it belongs because this is the + // FIREHOSE's door into the users table. A DID appearing in a profile or + // identity event means only that some repo somewhere emitted a record — a + // bridge, a replay, an overlapping feed — and any of those can arrive months + // after the account was erased. Letting it through would make the erasure + // undone by exactly the replays the marker exists to defend against, and + // undone silently: the users row reappears, repo.Create clears the marker on + // its way past, and the next replayed post indexes normally. + // + // The marker's only exit is a genuine re-registration, which reaches the + // repository's insert directly rather than through here. + // + // A LOOKUP FAILURE REFUSES. "I could not read the marker table" and "there + // is no marker" must never be the same answer, because the second one + // indexes. + if lookup, ok := s.userRepo.(ErasureLookup); ok { + erased, err := lookup.IsAccountDeleted(ctx, did) + if err != nil { + return fmt.Errorf("checking the erasure marker for %s before indexing: %w", did, err) + } + if erased { + // Nil, not an error. Every caller is a firehose consumer, and the + // connector dead-letters what a handler returns — so refusing with + // an error would turn each erased account into a permanent stream of + // redriving profile events. This is not a failure; it is an event + // with nothing to do. + log.Printf("INFO: not indexing %s from the firehose: the account was erased (migration 036 marker)", did) + return nil + } + } + // Try to create the user (idempotent - CreateUser returns existing user if DID exists) user, err := s.CreateUser(ctx, CreateUserRequest{ DID: did, diff --git a/internal/db/postgres/admission_queue_repo.go b/internal/db/postgres/admission_queue_repo.go --- a/internal/db/postgres/admission_queue_repo.go +++ b/internal/db/postgres/admission_queue_repo.go @@ -99,5 +99,28 @@ // its subject). A join would silently exempt exactly those from the quota. func (r *postgresAdmissionRepo) CountRecentAdmissions( ctx context.Context, communityDID, authorDID string, since time.Time, ) (int, error) { - return 0, nil + // starts_with on the authority segment, the same shape userRepo.Delete uses + // to sweep an author's admissions. The trailing "/" matters: without it + // did:plc:abc would also match did:plc:abcdef, and one author would consume + // another's quota. + // + // accepted and pending together, and NOTHING else. §8 is explicit that a + // refusal consumes no quota — counting rejected or removed rows would let an + // author past their limit extend their own lockout every time they tried + // again. pending_reacceptance counts too: the post is admitted and visible + // history, merely awaiting a re-decision on an edit. + const query = ` + SELECT count(*) + FROM community_post_admissions + WHERE community_did = $1 + AND starts_with(post_uri, 'at://' || $2 || '/') + AND status IN ('accepted', 'pending', 'pending_reacceptance') + AND created_at >= $3 + ` + + var count int + if err := r.db.QueryRowContext(ctx, query, communityDID, authorDID, since).Scan(&count); err != nil { + return 0, fmt.Errorf("counting recent admissions for %s in %s: %w", authorDID, communityDID, err) + } + return count, nil } diff --git a/internal/db/postgres/deleted_account_repo.go b/internal/db/postgres/deleted_account_repo.go --- a/internal/db/postgres/deleted_account_repo.go +++ b/internal/db/postgres/deleted_account_repo.go @@ -29,9 +29,24 @@ // author-owned posts an unknown author indexes normally (§5.3), so a false here // is indistinguishable from a healthy answer — a database blip would silently // re-index the content a deletion erased, which is the exact outcome the marker // table exists to prevent. +func (r *postgresUserRepo) IsAccountDeleted(ctx context.Context, did string) (bool, error) { + return accountIsErased(ctx, r.db, did) +} + +// IsAccountDeleted implements the same lookup for the standalone repository. func (r *DeletedAccountRepository) IsAccountDeleted(ctx context.Context, did string) (bool, error) { + return accountIsErased(ctx, r.db, did) +} + +// accountIsErased is the single statement behind both lookups above. +// +// It is one function because the two callers are the two halves of the same +// guard — the ingestion consumer refusing an erased author's events, and the +// user service refusing to re-index them — and a second spelling would be a +// second chance for one of them to drift into failing open. +func accountIsErased(ctx context.Context, db *sql.DB, did string) (bool, error) { var deleted bool - if err := r.db.QueryRowContext(ctx, + if err := db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM deleted_accounts WHERE did = $1)`, did, ).Scan(&deleted); err != nil { return false, fmt.Errorf("checking whether %s was erased: %w", did, err)