diff --git a/.env.example b/.env.example index b70d1117..4dff5a78 100644 --- a/.env.example +++ b/.env.example @@ -88,8 +88,12 @@ GITLAWB_P2P_PORT=7546 # GITLAWB_BOOTSTRAP_DISABLE_SEEDS=true for isolated dev/test networks. GITLAWB_BOOTSTRAP_PEERS= GITLAWB_BOOTSTRAP_DISABLE_SEEDS=false -# Require signatures on peer announce/sync write routes. Keep false during -# rolling upgrades so existing live nodes can still communicate. +# Require a peer write to prove its DID on every transport: RFC 9421 signatures +# on the announce/sync routes, and an Ed25519 payload signature on gossip +# ref-update events. Keep false during rolling upgrades so existing live nodes +# can still communicate. Upgrade every gossip-publishing peer to a build that +# signs events before setting this true, or their updates are dropped on +# arrival with no error visible on their side. GITLAWB_REQUIRE_SIGNED_PEER_WRITES=false # Require the authenticated pusher to be the repo owner on git-receive-pack. diff --git a/README.md b/README.md index 643992c2..0708e855 100644 --- a/README.md +++ b/README.md @@ -322,6 +322,12 @@ GITLAWB_REQUIRE_SIGNED_PEER_WRITES=true `POST /api/v1/sync/trigger` is not part of the staged rollout: it always requires a signature in both config modes and returns 401 without one, because each call drives an O(peers) outbound fan-out. +The flag is not HTTP-only. It also gates inbound gossip ref-update events on the libp2p mesh, which carry an Ed25519 payload signature rather than RFC 9421 headers because there are no HTTP headers to sign. Set to `true`, an unsigned gossip event is dropped. Set to `false`, an unsigned event is accepted with a warning during the rolling-upgrade window, and the write budget it consumes is charged to the forwarding peer rather than to the DID it claims, since an unsigned event's claimed DID is asserted and not proven. A present-but-invalid signature is refused in both modes; that is forgery, not an un-upgraded peer. + +Because the flag now spans both transports, the rollout order matters in one direction: **upgrade every gossip-publishing peer to a build that signs events before you set this to `true`.** Enabling it while an old publisher is still live drops that publisher's ref-updates on arrival, and the publisher sees no error, because gossip has no response to carry one. Upgrading the HTTP peers alone is not sufficient. + +There is a second precondition on gossip ingest that the flag does not control. A publisher's `node_did` must already exist in the receiving node's peers table, or the event is dropped as an unknown peer DID. This holds in both modes, signed and unsigned alike: a valid signature proves key possession, not membership. Rows reach that table over HTTP, either from `GITLAWB_BOOTSTRAP_PEERS` when this node contacts a bootstrap peer and records the DID it reports, or from a prior `POST /api/v1/peers/announce`. So a peer that only ever joined the libp2p mesh, with no HTTP announce and no bootstrap contact in either direction, will have its ref-updates dropped even with a good signature and the flag off. If gossip is silently not landing from a peer you can see on the mesh, check that its DID is in `GET /api/v1/peers` on the receiving side first. + --- ## Configuration @@ -345,7 +351,7 @@ Important node settings: | `GITLAWB_BOOTSTRAP_PEERS` | Comma-separated HTTP peer URLs. | | `GITLAWB_P2P_BOOTSTRAP` | Comma-separated libp2p multiaddrs. | | `GITLAWB_BOOTSTRAP_DISABLE_SEEDS` | Disable embedded seed peers for isolated dev/test networks. | -| `GITLAWB_REQUIRE_SIGNED_PEER_WRITES` | Require signed peer announce/sync writes. | +| `GITLAWB_REQUIRE_SIGNED_PEER_WRITES` | Require peer writes to prove their DID on every transport: RFC 9421 signatures on the announce/sync routes, and an Ed25519 payload signature on gossip ref-update events. Upgrade every gossip publisher before enabling, or their updates are dropped with no error on their side. | | `GITLAWB_AUTO_SYNC` | Enable automatic sync from known peers. | | `GITLAWB_MAX_PACK_BYTES` | Max git pack body size for smart-HTTP routes. | | `GITLAWB_GIT_SERVICE_TIMEOUT_SECS` | Max seconds a served git upload-pack, receive-pack, or `info/refs` advertisement may run before it is aborted (504). Default 600. Also bounds the withheld-blob classification walk (on both the upload-pack serve and receive-pack replication paths) and the push-side pin-candidate discovery (`rev-list` / `cat-file`), each reaped via process-group teardown at the deadline. On the path-scoped upload-pack path the classification walk and the pack serve share ONE deadline, so this value bounds their combined duration rather than granting each stage a full budget: a walk that consumes it leaves the serve nothing and the clone gets a 504. Serving large path-scoped repos may therefore need a higher value than they did when each stage was budgeted separately. Accepted range is 1 to 3153600000 (100 years), since the node derives deadlines from this value and a larger one cannot be represented. | diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b09cb6da..2a0d3d56 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -2461,6 +2461,11 @@ async fn post_receive_replication_tail( if announce { if let Some(p2p) = &p2p_handle { p2p.publish_ref_update(crate::p2p::RefUpdateEvent { + // Named, not literal: the ingest gate compares `v` + // against this same constant, so a bump has to move + // both ends together rather than leaving the emitter + // on a version the gate no longer accepts. + v: crate::p2p::CURRENT_REF_UPDATE_VERSION, node_did: node_did_str.clone(), pusher_did: pusher_did_clone.clone(), repo: repo_slug.clone(), @@ -2471,6 +2476,15 @@ async fn post_receive_replication_tail( timestamp: chrono::Utc::now().to_rfc3339(), cert_id: None, cid: cid.map(|s| s.to_string()), + // Left unsigned here on purpose. The swarm loop + // signs the event with the node keypair via + // `p2p::signed_publish_bytes` immediately before it + // publishes, and skips the publish outright if + // signing fails, so the wire always carries a + // signature even though this construction site does + // not. Setting one here would be signed over a + // payload the publisher has not finished building. + sig: None, }) .await; } diff --git a/crates/gitlawb-node/src/config.rs b/crates/gitlawb-node/src/config.rs index fefa063e..f2f2cc99 100644 --- a/crates/gitlawb-node/src/config.rs +++ b/crates/gitlawb-node/src/config.rs @@ -67,7 +67,9 @@ pub struct Config { #[arg(long, env = "GITLAWB_BOOTSTRAP_PEERS", value_delimiter = ',')] pub bootstrap_peers: Vec, - /// Require RFC 9421 signatures on peer announce/sync write routes. + /// Require a peer write to prove its DID on any transport: RFC 9421 + /// signatures on the peer announce/sync write routes, and a payload + /// signature on gossip ref-update events. /// Keep false during rolling upgrades so existing live nodes can still gossip. #[arg( long, diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 50c3bdda..7d6467e0 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -2280,15 +2280,7 @@ impl Db { // The read-then-write race is benign: if a row appears in between, // this write becomes an UPDATE and the UnprovenRepoint guard below // still refuses any http_url change. - PeerWriteAuthority::Unproven - if !sqlx::query_scalar::<_, bool>( - "SELECT EXISTS(SELECT 1 FROM peers WHERE did = $1)", - ) - .bind(did) - .fetch_one(&self.pool) - .await - .unwrap_or(false) => - { + PeerWriteAuthority::Unproven if !self.peer_exists(did).await.unwrap_or(false) => { match did.parse::() { // Only reachable from the bootstrap announce-back in main.rs, // which passes the contacted peer's raw JSON string; the @@ -2418,6 +2410,21 @@ impl Db { Ok(()) } + /// Whether a DID has a row in `peers`, by keyed lookup. + /// + /// For callers that only need the membership answer. `list_peers` fetches + /// and materializes every row, so using it as a membership test on a hot + /// path (the gossip ingest gate) makes the cost of one event grow with the + /// size of the table. + pub async fn peer_exists(&self, did: &str) -> Result { + Ok( + sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM peers WHERE did = $1)") + .bind(did) + .fetch_one(&self.pool) + .await?, + ) + } + pub async fn list_peers(&self) -> Result> { let rows = sqlx::query( "SELECT did, http_url, last_seen, last_ping_ok, announced_at @@ -6700,6 +6707,41 @@ mod peer_authority_tests { db } + /// `peer_exists` answers the membership question the gossip ingest gate + /// asks on every event, so both answers are pinned here: a registered DID + /// is true, and an unregistered one is false rather than an error or a + /// prefix match on a registered DID. + #[sqlx::test] + async fn peer_exists_answers_both_ways(pool: sqlx::PgPool) { + let db = db(pool).await; + let did = VICTIM_DID; + + assert!( + !db.peer_exists(did).await.unwrap(), + "an empty peers table must answer false, not error" + ); + + db.upsert_peer( + did, + "https://peer.example.com", + PeerWriteAuthority::Proven(did), + ) + .await + .unwrap(); + + assert!(db.peer_exists(did).await.unwrap()); + assert!( + !db.peer_exists(OTHER_DID).await.unwrap(), + "a different DID must not match" + ); + assert!( + !db.peer_exists(&VICTIM_DID[..VICTIM_DID.len() - 1]) + .await + .unwrap(), + "the lookup is an equality test, not a prefix test" + ); + } + /// The whole row, read back through `list_peers` rather than raw SQL, so a /// case that claims "unchanged" is comparing every column a consumer sees. async fn row(db: &Db, did: &str) -> Option<(String, String, Option, bool, String)> { @@ -7116,6 +7158,7 @@ mod peer_authority_tests { /// | `a_legacy_row_can_still_refresh_its_liveness` (db/mod.rs) | test-only. Seeds a PRE-GATE row by raw SQL on purpose: `upsert_peer` cannot create one, since the gate it is testing refuses exactly that DID. The fixture models what a deployed table already holds | /// | `gossip_ping_round_requires_two_failures_before_persisting_unreachable` (main.rs) | test-only. Seeds a peer row by raw SQL so the gossip ping round can probe readiness hysteresis without going through `upsert_peer` | /// | `manual_ping_uses_readiness_without_mutating_federation_gate` (api/peers.rs) | test-only. Seeds a peer row by raw SQL so the manual ping route can assert readiness probing without mutating federation gate state | +/// | `seed_peer` (p2p/mod.rs) | test-only. Seeds a peer row by raw SQL so the gossip ingest tests can drive the known-peer gate directly, including the did:web case `upsert_peer` would refuse | /// /// And the `upsert_peer` CALL-SITE authority table, which the ledger above /// structurally cannot hold, because the bootstrap site issues no SQL of its own @@ -7213,6 +7256,7 @@ mod peers_table_writer_guard { ("prune_non_public_peers", 1), ("prune_self_peers", 1), ("seed_local_peer", 1), + ("seed_peer", 1), ("upsert_peer", 2), ]; diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index 10aaca5b..0f853953 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/crates/gitlawb-node/src/main.rs @@ -89,7 +89,7 @@ async fn main() -> Result<()> { } // Load or generate the node's identity keypair - let keypair = load_or_create_keypair(&config)?; + let keypair = Arc::new(load_or_create_keypair(&config)?); let node_did = keypair.did(); // One-time metrics init. Must run before any handler that calls into @@ -236,6 +236,8 @@ async fn main() -> Result<()> { Arc::clone(&db), config.auto_sync, shutdown_rx, + Arc::clone(&keypair), + config.require_signed_peer_writes, ) .await { @@ -371,7 +373,7 @@ async fn main() -> Result<()> { config: Arc::new(config.clone()), db, node_did: node_did.clone(), - node_keypair: Arc::new(keypair), + node_keypair: keypair, p2p: p2p_handle, http_client, ref_update_tx, diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index c95ef1d1..653f338c 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -11,6 +11,8 @@ //! `gitlawb_sync_queue_processed_total{status}` //! * are webhooks reaching their endpoints? — //! `gitlawb_webhook_deliveries_total{result}` +//! * is inbound gossip being admitted or shed, and for which reason? +//! `gitlawb_gossip_ingest_events_total{outcome}` //! * how big are the packs we're sending and receiving? — //! `gitlawb_pack_size_bytes` //! * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards @@ -49,6 +51,7 @@ static AUTH_SUCCESSES: OnceLock = OnceLock::new(); static AUTH_FAILURES: OnceLock = OnceLock::new(); static SYNC_PROCESSED: OnceLock = OnceLock::new(); static WEBHOOK_DELIVERIES: OnceLock = OnceLock::new(); +static GOSSIP_INGEST: OnceLock = OnceLock::new(); static PACK_SIZE: OnceLock = OnceLock::new(); static PEERS_CONNECTED: OnceLock = OnceLock::new(); @@ -168,6 +171,21 @@ fn init_inner(version: &str, node_did: &str) { .set(webhook_deliveries) .expect("set WEBHOOK_DELIVERIES once"); + let gossip_ingest = IntCounterVec::new( + Opts::new( + "gitlawb_gossip_ingest_events_total", + "Total inbound gossip ref-update events and what the ingest path decided about each", + ), + &["outcome"], + ) + .expect("gitlawb_gossip_ingest_events_total definition"); + registry + .register(Box::new(gossip_ingest.clone())) + .expect("register gitlawb_gossip_ingest_events_total"); + GOSSIP_INGEST + .set(gossip_ingest) + .expect("set GOSSIP_INGEST once"); + let pack_size = Histogram::with_opts( HistogramOpts::new( "gitlawb_pack_size_bytes", @@ -270,6 +288,33 @@ pub fn record_webhook_delivery(result: &str) { } } +/// Record what the gossip ingest path decided about one inbound ref-update. +/// `outcome` ∈ {accepted, unsigned_admitted, write_failed, rejected, +/// source_rate_limited, author_rate_limited, unsigned_source_rate_limited}. +/// +/// `accepted` is reserved for signature-verified events. An unsigned event that +/// survives the rolling-upgrade window is `unsigned_admitted`, so the +/// authenticated-admission rate is not silently padded by legacy compatibility +/// traffic; an operator can tell whether the fleet still relies on the +/// compatibility allowance. +/// +/// The three shed reasons stay separate labels rather than one `rate_limited` +/// because they answer different operator questions: `source_rate_limited` is a +/// forwarding peer over the pre-parse brake, `unsigned_source_rate_limited` is +/// that same forwarder relaying unproven traffic, and `author_rate_limited` +/// names a principal a signature actually proved. Collapsing them would make a +/// flood of unauthenticated garbage indistinguishable from one registered peer +/// pushing too hard, which is exactly the distinction an alert needs. +/// +/// The label is the variant, never the free-form reason string that +/// `Rejected`/`WriteFailed` carry: those are attacker-influenced and would blow +/// up the label cardinality of a process-wide registry. +pub fn record_gossip_ingest(outcome: &str) { + if let Some(c) = GOSSIP_INGEST.get() { + c.with_label_values(&[outcome]).inc(); + } +} + /// Record a pack body size observation (bytes). pub fn observe_pack_size(bytes: f64) { if let Some(h) = PACK_SIZE.get() { @@ -365,6 +410,7 @@ mod tests { record_auth_failure("test/route", "test_reason"); record_sync_processed("done"); record_webhook_delivery("ok"); + record_gossip_ingest("accepted"); observe_pack_size(1024.0); set_peers_connected(0); } diff --git a/crates/gitlawb-node/src/p2p/mod.rs b/crates/gitlawb-node/src/p2p/mod.rs index 80e28a4a..9fec1e18 100644 --- a/crates/gitlawb-node/src/p2p/mod.rs +++ b/crates/gitlawb-node/src/p2p/mod.rs @@ -25,14 +25,260 @@ use tokio::sync::{mpsc, oneshot}; use tracing::{debug, info, warn}; use uuid::Uuid; -use crate::db::{Db, ReceivedRefUpdate}; +use gitlawb_core::identity::Keypair; + +use crate::db::{Db, PeerWriteDenied, ReceivedRefUpdate}; /// Topic for ref-update notifications published after every push. pub const REF_UPDATES_TOPIC: &str = "gitlawb/ref-updates/v1"; +/// Pre-parse budget, keyed on the mesh peer that HANDED us the message: 2000 +/// events per 60 seconds. +/// +/// This one is not an authorization control and must not be sized like one. +/// `propagation_source` is free to mint and identifies a forwarder, not an +/// author, so it is weak in both directions: an attacker rotates it, and a +/// flood relayed through an honest neighbour debits that neighbour. All it buys +/// is a bound on raw CPU before anything is parsed or verified, so it is sized +/// to be well clear of any legitimate burst. Gossipsub re-shares mesh-wide, so +/// one edge carries the traffic of every author routed through it; 2000 per +/// minute is roughly 33 parse-plus-Ed25519-verify per second, a fraction of a +/// core, while still bounding what a single edge can cost. +const GOSSIP_SOURCE_MAX_EVENTS: usize = 2000; +/// Post-auth budget, keyed on the authenticated `node_did`: 500 events per 60 +/// seconds. +/// +/// This is the tight bound, and it is where the tightness belongs, because it +/// is charged ONLY when a signature has proven who is asking. An unsigned +/// event's `node_did` is a claim anyone on the mesh can make, so charging this +/// bucket on one would let an attacker deny a DID it names; unsigned traffic is +/// bounded on the forwarder instead, below. It bounds the two durable writes +/// per event, charged to the principal that authored them. `api::repos` +/// publishes one event per updated ref, so the number has to admit a whole +/// large push: 500 covers a tag-heavy push, an initial import, or a mirror +/// backfill of a few hundred refs arriving in one window. +const GOSSIP_AUTHOR_MAX_EVENTS: usize = 500; +/// Budget for UNSIGNED events, keyed on `propagation_source`: 1500 events per +/// 60 seconds. +/// +/// An unsigned event's `node_did` is asserted, not proven, so it cannot be +/// charged to an author without handing an attacker a way to deny a chosen +/// victim. The forwarder is the only identity available, and this is the bound +/// that keeps an unsigned flood from buying an unlimited number of `peer_exists` +/// round trips and durable writes. +/// +/// Sized deliberately, and NOT at the author cap. 500 is what ONE author's +/// large push needs: `a_sixty_one_ref_push_from_one_known_peer_is_accepted_whole` +/// exists because a 60-per-source bound broke a 61-ref push. A forwarder +/// aggregates many unsigned authors, so sizing a forwarder-keyed bucket AT the +/// per-author cap re-imposes exactly the mesh-edge denial the pre-parse brake's +/// doc comment above warns against. 1500 is three times the largest legitimate +/// single-author burst, which covers the aggregation this network can actually +/// produce (one global topic, and a peers table in the low hundreds), while +/// staying strictly below the 2000 pre-parse brake so it binds first. +/// +/// It is defeated by `PeerId` rotation, like the pre-parse brake it sits under. +/// That is inherent to keying on a free identity and is why the unsigned path +/// is a rolling-upgrade allowance rather than a permanent one. +const GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS: usize = 1500; +/// Two forwarder-keyed bounds now exist, and both must stay looser than the +/// per-author budget: sizing either at or below it puts the tight bound back on +/// the mesh edge, which is the shape being fixed here. The unsigned bound must +/// in turn stay under the pre-parse brake, or the brake it nests inside never +/// binds. Enforced at compile time rather than in a test, because it is a +/// relation between constants and a test can only catch it after someone runs +/// it. +const _: () = assert!( + GOSSIP_SOURCE_MAX_EVENTS > GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS + && GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS > GOSSIP_AUTHOR_MAX_EVENTS, + "both forwarder-keyed bounds must stay looser than the per-author bound, \ + and the unsigned bound must stay under the pre-parse brake" +); +const GOSSIP_INGEST_WINDOW: Duration = Duration::from_secs(60); +/// Ceiling on tracked source peers, matching the bound the HTTP brakes use in +/// `main.rs`. Keeps a source-rotation flood from growing the limiter's own map. +const GOSSIP_INGEST_MAX_SOURCES: usize = 200_000; +/// Ceiling on tracked author DIDs. Reaching this map costs an attacker a +/// registered peer row per key, but registration is open through the announce +/// path, so the bound is not left to that. +const GOSSIP_INGEST_MAX_AUTHORS: usize = 200_000; +/// How often the swarm loop evicts expired keys from the ingest limiters. +/// +/// Matches the 300s the HTTP-side sweeper in `main.rs` runs on, so both halves +/// of the node reclaim limiter keys on the same cadence rather than each having +/// its own tuning. Any interval comfortably above [`GOSSIP_INGEST_WINDOW`] +/// works: a key whose window has not elapsed is retained by `cleanup` anyway, +/// so sweeping more often would cost lock traffic and reclaim nothing extra. +const GOSSIP_INGEST_SWEEP_INTERVAL: Duration = Duration::from_secs(300); + +/// The gossip ingest budgets, built the same way for the swarm loop and for the +/// tests so a test can never assert against a budget production does not run. +/// +/// They are deliberately separate limiters rather than one: they key on +/// different identities (a forwarder before parsing, a forwarder again for +/// unproven traffic, an author after authentication) and sit at different +/// points in the path. `unsigned` is its own limiter rather than a second +/// `check` against `source`, which would spend the same budget twice and bound +/// nothing tighter than `source` already does. +pub(crate) struct IngestLimiters { + /// Keyed on `propagation_source`, checked before the parse. + source: crate::rate_limit::RateLimiter, + /// Keyed on `propagation_source`, checked in the unsigned branch before the + /// `peer_exists` round trip, because an unsigned event names no principal + /// that could be charged instead. + unsigned: crate::rate_limit::RateLimiter, + /// Keyed on the `node_did` a signature PROVED, checked before the writes. + author: crate::rate_limit::RateLimiter, +} + +impl IngestLimiters { + pub(crate) fn new() -> Self { + Self::with_window(GOSSIP_INGEST_WINDOW) + } + + /// The single place the three limiters are built. `new` is the only + /// production caller and passes the real window; the parameter exists so a + /// test can exercise expiry without sleeping a minute, and it deliberately + /// leaves the caps and key ceilings alone so a test still asserts against + /// the budgets production runs. + fn with_window(window: Duration) -> Self { + Self { + source: crate::rate_limit::RateLimiter::new_bounded( + GOSSIP_SOURCE_MAX_EVENTS, + window, + GOSSIP_INGEST_MAX_SOURCES, + ), + unsigned: crate::rate_limit::RateLimiter::new_bounded( + GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS, + window, + GOSSIP_INGEST_MAX_SOURCES, + ), + author: crate::rate_limit::RateLimiter::new_bounded( + GOSSIP_AUTHOR_MAX_EVENTS, + window, + GOSSIP_INGEST_MAX_AUTHORS, + ), + } + } + + /// Every limiter in this struct, by DESTRUCTURING `Self` rather than by a + /// hand-written list. A fourth limiter added later fails to compile here + /// (both the pattern and the array length), which is what keeps [`cleanup`] + /// from silently skipping it. + /// + /// This is the same completeness idea `sweep_rate_limiters` in `main.rs` + /// documents for the `AppState` limiters, except that one is driven off a + /// hand-written list and a missed field there costs only a review; here the + /// compiler refuses. + /// + /// [`cleanup`]: IngestLimiters::cleanup + fn each(&self) -> [&crate::rate_limit::RateLimiter; 3] { + let Self { + source, + unsigned, + author, + } = self; + [source, unsigned, author] + } + + /// Evict expired entries from every ingest limiter. + /// + /// These limiters are locals of the swarm task, not fields of `AppState`, + /// so the periodic `sweep_rate_limiters` in `main.rs` cannot reach them and + /// the swarm loop has to sweep its own. Without this a key stays resident + /// from the first event a peer forwards until the map hits its 200k ceiling + /// and the inline capacity sweep fires, so a node that has merely SEEN a lot + /// of forwarders over its uptime pays for all of them at once. + async fn cleanup(&self) { + for limiter in self.each() { + limiter.cleanup().await; + } + } +} + +#[cfg(test)] +impl IngestLimiters { + /// Every limiter in this struct, paired with the cap and key ceiling it is + /// documented to carry. + /// + /// Derived by DESTRUCTURING `Self` rather than listed by hand, so a fourth + /// limiter added later fails to compile here instead of quietly escaping + /// the wiring tests that assert each one is built as documented. + fn all(&self) -> Vec<(&'static str, &crate::rate_limit::RateLimiter, usize, usize)> { + let Self { + source, + unsigned, + author, + } = self; + vec![ + ( + "source", + source, + GOSSIP_SOURCE_MAX_EVENTS, + GOSSIP_INGEST_MAX_SOURCES, + ), + ( + "unsigned", + unsigned, + GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS, + GOSSIP_INGEST_MAX_SOURCES, + ), + ( + "author", + author, + GOSSIP_AUTHOR_MAX_EVENTS, + GOSSIP_INGEST_MAX_AUTHORS, + ), + ] + } +} + +/// The event format version this build emits and understands. +/// +/// 0 is the versionless form: the field set this struct shipped with, before +/// `v` existed. It is not a placeholder for "unset", it is a real version whose +/// wire encoding happens to omit the key. +pub(crate) const CURRENT_REF_UPDATE_VERSION: u32 = 0; + +/// True for the version whose key is omitted from the wire form. Free function +/// because `skip_serializing_if` takes a path, not a closure. +fn is_zero(v: &u32) -> bool { + *v == 0 +} + /// A ref-update event published to Gossipsub when a push lands. +/// +/// The signing bytes are this struct serialized with `sig` set to None (see +/// [`signing_bytes`]), so the FIELD SET IS A WIRE FORMAT, not a struct that can +/// be extended. Any field added here changes the signing bytes for every event +/// that carries it: a node that does not know the new field re-serializes +/// without it, computes different bytes, and rejects the event as a bad +/// signature. That failure names forgery while describing a version skew, which +/// is the worst direction for it to fail in. +/// +/// So `v` carries the format version INSIDE the signed bytes, and a field +/// addition means bumping it and keeping a verification path for every version +/// still in the wild, not just editing this struct. A version alongside the +/// signature rather than under it would be attacker-mutable and prove nothing. +/// `ingest_ref_update` refuses anything above [`CURRENT_REF_UPDATE_VERSION`] in +/// its own words, so a newer publisher's events fail as an unsupported version +/// rather than as a signature mismatch; that guard is what makes the loud +/// failure real, and it cannot be retrofitted into receivers already deployed. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct RefUpdateEvent { + /// Format version of the signed field set, 0 being the versionless form + /// this struct shipped with. + /// + /// Declared FIRST so a later version's signing bytes lead with it, and + /// skipped when zero so a v0 event's wire bytes and signing bytes stay + /// byte-identical to what the pre-version build emits: no `"v"` key + /// appears, `GOLDEN_SIGNING_BYTES` is unchanged, and every signature + /// already in flight still verifies. `#[serde(default)]` is the other half: + /// an event from a peer that predates the field parses as 0, and + /// re-serializing reproduces its exact original bytes, which IS the v0 + /// verification path. + #[serde(default, skip_serializing_if = "is_zero")] + pub v: u32, /// gitlawb DID of the node publishing the event pub node_did: String, /// DID of the agent who pushed @@ -56,6 +302,481 @@ pub struct RefUpdateEvent { pub cert_id: Option, /// IPFS CID of the latest commit object (set after pinning completes) pub cid: Option, + /// Ed25519 signature (base64url, no padding) by the key behind `node_did`, + /// over the signing bytes defined by `signing_bytes` (this struct serialized + /// with `sig` set to None). Optional for backward compat with older peers + /// that don't include it; enforcement is the operator flag's job. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sig: Option, +} + +/// The bytes a `RefUpdateEvent` signature is computed over: the event +/// serialized with `sig` set to None. +/// +/// This is the ONLY producer of signing input on either side. Emit and verify +/// both call it, so the two cannot drift. `skip_serializing_if` on `sig` keeps +/// the output byte-identical to the legacy wire form (no `"sig": null`), and +/// serde's derive serializes in declaration order, so both sides re-serializing +/// the same struct definition agree regardless of the order fields arrived in. +fn signing_bytes(event: &RefUpdateEvent) -> serde_json::Result> { + let mut unsigned = event.clone(); + unsigned.sig = None; + serde_json::to_vec(&unsigned) +} + +/// Sign an event in place: sets `sig` to the base64url signature by `keypair` +/// over [`signing_bytes`]. +fn sign_ref_update(keypair: &Keypair, event: &mut RefUpdateEvent) -> serde_json::Result<()> { + let bytes = signing_bytes(event)?; + event.sig = Some(keypair.sign_b64(&bytes)); + Ok(()) +} + +/// The bytes the node publishes for one outbound ref-update: the event signed +/// by the node keypair, then serialized. +/// +/// The swarm loop and the round-trip test share this one function, so the bytes +/// a test verifies are the bytes the mesh actually receives. +fn signed_publish_bytes(keypair: &Keypair, event: &RefUpdateEvent) -> serde_json::Result> { + let mut event = event.clone(); + sign_ref_update(keypair, &mut event)?; + serde_json::to_vec(&event) +} + +/// Exactly the pair the swarm loop hands `gossipsub.publish` for one outbound +/// ref-update: the topic, and the signed bytes. +/// +/// Extracted from the publish arm so a test can hold what the loop publishes. +/// The arm itself sits inside a `select!` that no test drives, so before this +/// existed the only thing standing between a regression and the mesh was that +/// the arm happened to call `signed_publish_bytes`; an arm rewritten to +/// `serde_json::to_vec(&event)` would have published unsigned bytes with the +/// whole suite green. Now that regression has to be made HERE to stay silent. +/// +/// Be exact about what this does and does not close. It closes +/// sign-before-publish and the topic the bytes go out on. It does NOT observe +/// the `select!` arm dispatching to it, and it does not observe `require_signed` +/// arriving from `main.rs`; both need a live swarm. Those remain uncovered +/// seams, named rather than implied away. +fn ref_update_publish_args( + keypair: &Keypair, + event: &RefUpdateEvent, +) -> serde_json::Result<(gossipsub::IdentTopic, Vec)> { + let bytes = signed_publish_bytes(keypair, event)?; + Ok((gossipsub::IdentTopic::new(REF_UPDATES_TOPIC), bytes)) +} + +/// Resolve the public key behind a claimed `node_did`, refusing anything that +/// is not a resolvable `did:key`. +/// +/// The did-method and resolution refusals answer in the SAME words as the +/// peers-table gate in db/mod.rs, so the two surfaces that judge the same input +/// do not drift into separate vocabularies. The sentences are built from +/// `PeerWriteDenied` itself rather than retyped, so they cannot. +fn resolve_node_did(node_did: &str) -> Result { + let unresolvable = |reason: String| { + PeerWriteDenied::UnresolvableDid { + did: node_did.to_string(), + reason, + } + .to_string() + }; + + let did = node_did + .parse::() + .map_err(|e| unresolvable(e.to_string()))?; + if !did.is_did_key() { + return Err(PeerWriteDenied::UnsupportedDidMethod { + did: node_did.to_string(), + } + .to_string()); + } + did.to_verifying_key() + .map_err(|e| unresolvable(e.to_string())) +} + +/// Verify that `event.sig` is an Ed25519 signature over [`signing_bytes`] by +/// the key behind `event.node_did`. +/// +/// The signature is bound to the claimed identity structurally: the key comes +/// from `node_did` and nowhere else, so a valid signature by some other key +/// never passes. +fn verify_ref_update(event: &RefUpdateEvent) -> Result<(), String> { + use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; + + let verifying_key = resolve_node_did(&event.node_did)?; + + let sig_b64 = event + .sig + .as_deref() + .ok_or_else(|| "event carries no signature".to_string())?; + + let sig_bytes: [u8; 64] = URL_SAFE_NO_PAD + .decode(sig_b64) + .map_err(|_| "signature is not valid base64url".to_string())? + .try_into() + .map_err(|_| "signature is not 64 bytes".to_string())?; + + let bytes = signing_bytes(event).map_err(|e| e.to_string())?; + gitlawb_core::identity::verify(&verifying_key, &bytes, &sig_bytes) + .map_err(|_| "signature does not verify against node_did".to_string()) +} + +/// What the ingest path decided about one inbound gossip message. +#[derive(Debug)] +pub(crate) enum IngestOutcome { + /// The event was authenticated AND every write it implies landed. + Accepted, + /// The event was admitted WITHOUT authentication, through the rolling-upgrade + /// window, and every write it implies landed. + /// + /// Deliberately a distinct outcome from `Accepted`: a valid signature proves + /// who the sender is, an unsigned event proves nothing, and counting the two + /// under one label would make the fleet's authenticated-admission rate + /// indistinguishable from its reliance on the legacy compatibility path. + UnsignedAdmitted, + /// The event passed every guard, but a durable write failed. The decision + /// was still "admit it", so this is not a refusal; it exists because + /// returning an admission outcome (`Accepted` or `UnsignedAdmitted`) for an + /// event whose row never landed would make the outcome an observability + /// lie. + WriteFailed(String), + /// The event was dropped. Nothing is stored, so the reason exists only to + /// be logged and counted. + Rejected(String), + /// The forwarding peer is over the pre-parse ingest budget. Dropped without + /// being parsed or verified, which is the whole point of that brake. + SourceRateLimited, + /// The authenticated author is over its own write budget. Carries the DID + /// so the drop names a principal and not just a mesh edge. + AuthorRateLimited(String), + /// A forwarder is over the budget for UNSIGNED events relayed down its + /// edge. Carries the `propagation_source`. + /// + /// Deliberately not folded into `AuthorRateLimited`: that variant asserts a + /// proven principal, and an unsigned event's `node_did` is a claim. Naming + /// a forwarder as an author would be the same class of observability lie + /// that `WriteFailed` exists to avoid. + UnsignedSourceRateLimited(String), +} + +impl IngestOutcome { + /// The `outcome` label this decision is counted under in + /// `gitlawb_gossip_ingest_events_total`. + /// + /// A method with an exhaustive match rather than a label written at each + /// call site: a seventh variant added later fails to compile here instead of + /// landing in production as an outcome no dashboard can see. `/metrics` is + /// the only externally observable surface this daemon has, so an uncounted + /// outcome is an invisible one. + /// + /// Returns `&'static str` on purpose. The reason strings `Rejected` and + /// `WriteFailed` carry are shaped by the sender and would make the label set + /// unbounded in a process-wide registry, so they stay in the log line and + /// only the variant reaches the counter. + fn metric_label(&self) -> &'static str { + match self { + IngestOutcome::Accepted => "accepted", + IngestOutcome::UnsignedAdmitted => "unsigned_admitted", + IngestOutcome::WriteFailed(_) => "write_failed", + IngestOutcome::Rejected(_) => "rejected", + IngestOutcome::SourceRateLimited => "source_rate_limited", + IngestOutcome::AuthorRateLimited(_) => "author_rate_limited", + IngestOutcome::UnsignedSourceRateLimited(_) => "unsigned_source_rate_limited", + } + } +} + +#[cfg(test)] +thread_local! { + /// Test-only tally of the `peer_exists` round trips this path makes. + /// + /// It exists so a test can prove a guard runs BEFORE the database is + /// touched rather than merely before a debit. Asserting an outcome cannot + /// tell those two placements apart, and the difference is the whole point + /// of hoisting the slug check: a malformed event has to cost nothing, not + /// merely be charged to nobody. + /// + /// Thread-local rather than a global counter because `#[sqlx::test]` drives + /// each test on its own current-thread runtime, so the count is naturally + /// per-test and cannot be polluted by a test running in parallel. + static PEER_EXISTS_CALLS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +/// Handle one inbound gossip ref-update: authenticate it, then write it. +/// +/// The swarm loop and the tests share this one path, so a guard cannot hold in +/// one and not the other. The guards are the same trio the HTTP twin +/// (`api::peers::notify_sync`) applies, carried by a payload signature because +/// gossip has no HTTP signature to key off: the sender proves the `node_did` it +/// claims, that DID is a known peer, and the repo slug is well formed. Every +/// refusal drops the event; nothing is stored (KTD-4), so an unauthenticated +/// sender cannot grow a table. +pub(crate) async fn ingest_ref_update( + db: &Db, + limiters: &IngestLimiters, + require_signed: bool, + auto_sync: bool, + data: &[u8], + propagation_source: &PeerId, +) -> IngestOutcome { + // FIRST, ahead of the parse and ahead of signature verification. Verifying + // a signature is the expensive step on this path, so a brake placed after it + // would let an unauthenticated flood buy exactly the CPU the brake exists to + // protect. Same ordering rationale as the HTTP sync-trigger brake in + // server.rs, which is layered outermost so it runs before auth. + // + // It is kept generous on purpose. The key is a forwarder, so a tight bound + // here denies an honest neighbour on someone else's flood. The tighter + // bounds live below: unsigned traffic on a second forwarder-keyed bucket, + // and verified traffic on the author the signature proved. + if !limiters.source.check(&propagation_source.to_string()).await { + return IngestOutcome::SourceRateLimited; + } + + let event = match serde_json::from_slice::(data) { + Ok(event) => event, + Err(e) => return IngestOutcome::Rejected(format!("malformed ref-update event: {e}")), + }; + + // Version gate, immediately after the parse and ahead of every gate that + // reads a field, the signature match included. + // + // A version this build does not know means the meaning of every field below + // is unknown, so no gate down there is entitled to judge the event, and + // admitting it would write rows whose semantics this node cannot state. + // + // The reason it must run ahead of the SIGNATURE specifically is + // observability, and it is the whole point of carrying a version at all. + // The signing bytes cover the version, so a newer publisher's correctly + // signed event reaches a v0 receiver as bytes that receiver cannot + // reproduce: without this gate it lands as "signature does not verify + // against node_did", which is an accusation of forgery levelled at an + // honest peer, and it reads that way in the logs and counters an operator + // would use to notice a mesh partition forming. Same class of lie + // `UnsignedSourceRateLimited` was split out to avoid. + // + // It goes in now, with the version, because it cannot be added later: the + // receivers that need it are the ones already deployed by the time a v1 + // publisher exists. + if event.v > CURRENT_REF_UPDATE_VERSION { + return IngestOutcome::Rejected(format!( + "unsupported ref-update event version {}; this build understands version {}", + event.v, CURRENT_REF_UPDATE_VERSION + )); + } + + // did-method gate first, and in BOTH enforcement modes: a non-did:key peer + // is unauthenticatable by design, and running this before the flag branch + // is what keeps the answer independent of flag state. + if let Err(reason) = resolve_node_did(&event.node_did) { + return IngestOutcome::Rejected(reason); + } + + // #272: the slug reaches a `PathBuf::join` in the sync worker, so it is + // rejected here, before the ref-update row and the queue row. + // + // It sits this high on purpose, above the signature verify and above the + // `peer_exists` round trip, not merely above the budgets. It depends on + // nothing but the parsed struct, so hoisting it costs nothing and removes + // the work entirely: a structurally invalid event now buys no Ed25519 + // verify and no database query. Placing it below either of those would buy + // the fairness property (a malformed event charges nobody) while leaving + // the cost in place, charged only to the pre-parse brake that `PeerId` + // rotation defeats. + if let Err(e) = crate::git::repo_store::validate_repo_slug(&event.repo) { + return IngestOutcome::Rejected(format!("invalid repo field: {e}")); + } + + // Whether `node_did` was PROVEN by a signature on this event, as opposed to + // merely asserted. Only a proven author may be charged the author budget + // below, and there is exactly one debit site reading this, so the two arms + // cannot drift apart. + let mut verified = false; + // Whether this event arrived with no signature at all and was let through + // by the rolling-upgrade window. Carried down the same way `verified` is, + // and for the same reason: the warning it drives belongs below the gates + // that can still drop the event, not in the arm that merely reached them. + // + // Deliberately a second flag rather than `!verified`. Those two coincide + // only because the present-but-invalid case returns early above, which is a + // property of the arms as they stand today, not something this variable's + // meaning should depend on. + let mut unsigned = false; + match event.sig { + // A signature that is present must verify. A present-but-invalid one is + // forgery, never a peer that has not upgraded yet, so the flag does not + // enter into it. + Some(_) => { + if let Err(reason) = verify_ref_update(&event) { + return IngestOutcome::Rejected(reason); + } + verified = true; + } + None if require_signed => { + return IngestOutcome::Rejected("unsigned ref-update event".to_string()); + } + // Rolling-upgrade window, same posture and same pointer at the flag as + // the HTTP twin's unsigned-notify warning. + None => { + unsigned = true; + // Unsigned traffic is bounded here, on the forwarder, because it is + // the only identity this event establishes. Charged BEFORE + // `peer_exists`, which is a Postgres round trip per event: a brake + // sitting below it would not bound what an unsigned flood costs the + // node, leaving that cost to the deliberately loose pre-parse brake + // alone. + // + // No new key-farming axis: `propagation_source` is already a key in + // the pre-parse map, under the same ceiling. + // + // Only this arm. Charging it on the verified arm too would let a + // spent unsigned edge shed a peer's genuine signed pushes, which is + // the mesh-edge denial the source brake's doc comment above exists + // to avoid. + let source_key = propagation_source.to_string(); + if !limiters.unsigned.check(&source_key).await { + return IngestOutcome::UnsignedSourceRateLimited(source_key); + } + } + } + + // Authentication is not authorization: a freshly minted did:key signs its + // own events perfectly well, so the signature alone says nothing about who + // this peer is to us. Be precise about what this check buys, because it is + // NOT a closed membership boundary: `upsert_peer` accepts an + // `PeerWriteAuthority::Unproven` announce for an unseen did:key, so an + // attacker can self-register a fresh DID through the announce path and then + // pass this gate. What it does buy is that an unregistered DID cannot write + // at all, and that combined with the signature check above, an existing + // peer cannot be impersonated: claiming a registered DID now requires the + // key behind it. + // + // Shed an author that is ALREADY over budget before paying for the lookup. + // The brake belongs in front of the work it bounds: an over-budget author is + // going to be refused whatever the peer table says, so the round trip is work + // done for a request that was never going to be admitted, and a signed author + // can drive it at the pre-parse source rate. + // + // Deliberately a READ, not the charge. `check` inserts a window for a key it + // has never seen, so probing with it here would let a flood of self-minted + // signed DIDs occupy the bounded author map before `peer_exists` had a chance + // to refuse them, turning a brake into a memory-fill surface. `is_over_budget` + // allocates nothing and an unseen DID reads as within budget, so the only + // behaviour that changes is WHEN an already-over-budget author is refused. + // The charge itself stays below the lookup, unmoved. + if verified && limiters.author.is_over_budget(&event.node_did).await { + return IngestOutcome::AuthorRateLimited(event.node_did.clone()); + } + + // Keyed lookup, not `list_peers`: this runs on every event that survives + // the parse, and scanning the whole table per event makes ingest cost grow + // with the peer count. + #[cfg(test)] + PEER_EXISTS_CALLS.with(|calls| calls.set(calls.get() + 1)); + match db.peer_exists(&event.node_did).await { + Ok(true) => {} + Ok(false) => { + return IngestOutcome::Rejected(format!("unknown peer DID: {}", event.node_did)); + } + Err(e) => return IngestOutcome::Rejected(format!("peer lookup failed: {e}")), + } + + // The tight budget, and the only one keyed on an identity the sender had to + // prove. It is charged ONLY when the signature verified, which is what + // `verified` carries down from the match above. + // + // Charging it on an unsigned event would make it a victim-selection + // mechanism rather than a fairness control. `peer_exists` proves the DID is + // registered; it does not prove this sender holds it, and in the + // rolling-upgrade window anyone on the open mesh can name a registered DID. + // An attacker would then spend a chosen victim's budget from an unrelated + // `PeerId` and the victim's own signed pushes would come back + // `AuthorRateLimited`. A rate-limit key derived from an unproven claim is + // worse than no key at all, because the denial it produces is targetable. + // Unsigned traffic is bounded on the forwarder instead, in the `None` arm + // above. + // + // What this does NOT buy: an aggregate write bound. A verified signature + // proves key possession, not a scarce principal, and the announce path + // registers fresh did:keys, so an attacker who rotates keys still gets a + // fresh budget each time. The aggregate per-edge bound remains the + // pre-parse source brake. What it buys is that a NAMED victim's budget is + // no longer spendable by anyone else. + if verified && !limiters.author.check(&event.node_did).await { + return IngestOutcome::AuthorRateLimited(event.node_did.clone()); + } + + // Below every gate that can still drop the event, so "accepted" is a report + // of an outcome rather than a prediction of one. + // + // It used to sit up in the `None` arm, which put it above the unsigned + // forwarder budget and above `peer_exists`. An event shed by either of + // those left a log that said the node accepted it and a database that had + // never heard of it, which is the same class of lie as an accusation of + // forgery for a version skew: it misdirects whoever is reading the logs to + // find out why gossip is not landing. Single site on purpose. A warning + // emitted in the arm AND again down here would double-count in anything + // grepping for it. + if unsigned { + warn!( + did = %event.node_did, + "accepted unsigned gossip ref-update; set GITLAWB_REQUIRE_SIGNED_PEER_WRITES=true after all peers upgrade" + ); + } + + info!( + from = %propagation_source, + repo = %event.repo, + ref_name = %event.ref_name, + new_sha = %event.new_sha, + "ref-update received via gossipsub" + ); + + let update = ReceivedRefUpdate { + id: Uuid::new_v4().to_string(), + node_did: event.node_did.clone(), + pusher_did: event.pusher_did.clone(), + repo: event.repo.clone(), + owner_did: event.owner_did.clone(), + ref_name: event.ref_name.clone(), + old_sha: event.old_sha.clone(), + new_sha: event.new_sha.clone(), + timestamp: event.timestamp.clone(), + cert_id: event.cert_id.clone(), + received_at: Utc::now().to_rfc3339(), + // The peer that FORWARDED this message through the mesh, not the + // author. The authenticated author is `node_did` beside it. + from_peer: propagation_source.to_string(), + }; + // Both writes are still attempted independently: a failed row must not cost + // the queue entry, and a failed queue entry must not undo the row. Only the + // OUTCOME changes, so `Accepted` keeps meaning "authenticated AND stored". + let mut write_error: Option = None; + if let Err(e) = db.insert_ref_update(&update).await { + warn!(err = %e, "failed to store received ref-update"); + write_error = Some(format!("failed to store received ref-update: {e}")); + } + if auto_sync { + if let Err(e) = db + .enqueue_sync( + &event.repo, + &event.node_did, + &event.ref_name, + &event.new_sha, + event.cid.as_deref(), + ) + .await + { + warn!(err = %e, "failed to enqueue sync for received ref-update"); + write_error.get_or_insert(format!("failed to enqueue sync: {e}")); + } + } + match write_error { + Some(reason) => IngestOutcome::WriteFailed(reason), + None if unsigned => IngestOutcome::UnsignedAdmitted, + None => IngestOutcome::Accepted, + } } /// A DID record stored in the Kademlia DHT — maps a gitlawb DID to a node. @@ -167,6 +888,9 @@ struct GitlawbBehaviour { /// Start the libp2p swarm. Returns a handle for sending commands and the /// listening multiaddrs. Runs the event loop as a background tokio task /// that exits cleanly when `shutdown_rx` flips to `true`. +// Wide, but each argument is a distinct piece of node configuration and there +// is exactly one call site; bundling them would buy nothing. +#[allow(clippy::too_many_arguments)] pub async fn start( node_did: &str, listen_port: u16, @@ -174,6 +898,8 @@ pub async fn start( db: Arc, auto_sync: bool, shutdown_rx: tokio::sync::watch::Receiver, + keypair: Arc, + require_signed: bool, ) -> Result { // Derive a stable libp2p Ed25519 key from a seed based on the node DID. // In production you'd load/persist this key alongside the identity PEM. @@ -196,6 +922,10 @@ pub async fn start( info!(peer_id = %local_peer_id, "libp2p identity"); + // Per-source ingest brake, held across the whole swarm loop so budgets + // accumulate per forwarding peer. + let ingest_limiters = IngestLimiters::new(); + let (cmd_tx, mut cmd_rx) = mpsc::channel::(64); let handle = P2pHandle { @@ -279,8 +1009,21 @@ pub async fn start( // Start the event loop as a background task tokio::spawn(async move { let mut shutdown_rx = shutdown_rx; + // The ingest limiters are owned by this task, so this loop is the only + // thing that can sweep them. Delay on a missed tick because a sweep the + // loop was too busy to run has no value in being run twice back to back. + let mut ingest_sweep = tokio::time::interval(GOSSIP_INGEST_SWEEP_INTERVAL); + ingest_sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); loop { tokio::select! { + // Reclaim ingest-limiter keys whose window has elapsed. Kept in + // the select! rather than a second task so the limiters stay + // task-local and need no Arc or lock shared with the rest of the + // node. `tokio::time::interval` fires its first tick + // immediately, which sweeps an empty map and is a no-op. + _ = ingest_sweep.tick() => { + ingest_limiters.cleanup().await; + } // Graceful shutdown: exit the swarm loop when the // process-wide signal flips. This drops the Swarm // which closes all libp2p connections cleanly. @@ -299,38 +1042,58 @@ pub async fn start( SwarmEvent::Behaviour(GitlawbBehaviourEvent::Gossipsub( gossipsub::Event::Message { propagation_source, message, .. } )) => { - if let Ok(event) = serde_json::from_slice::(&message.data) { - info!( + let outcome = ingest_ref_update( + &db, + &ingest_limiters, + require_signed, + auto_sync, + &message.data, + &propagation_source, + ).await; + // Counted before the log match, once, for every + // variant including `Accepted`. The logs describe an + // individual drop; the counter is what an alert can + // read, and without an accepted count the shed + // reasons have no denominator to be a rate of. + crate::metrics::record_gossip_ingest(outcome.metric_label()); + match outcome { + IngestOutcome::Accepted => {} + IngestOutcome::UnsignedAdmitted => {} + IngestOutcome::WriteFailed(reason) => warn!( from = %propagation_source, - repo = %event.repo, - ref_name = %event.ref_name, - new_sha = %event.new_sha, - "ref-update received via gossipsub" - ); - let update = ReceivedRefUpdate { - id: Uuid::new_v4().to_string(), - node_did: event.node_did.clone(), - pusher_did: event.pusher_did.clone(), - repo: event.repo.clone(), - owner_did: event.owner_did.clone(), - ref_name: event.ref_name.clone(), - old_sha: event.old_sha.clone(), - new_sha: event.new_sha.clone(), - timestamp: event.timestamp.clone(), - cert_id: event.cert_id.clone(), - received_at: Utc::now().to_rfc3339(), - from_peer: propagation_source.to_string(), - }; - let _ = db.insert_ref_update(&update).await; - if auto_sync { - let _ = db.enqueue_sync( - &event.repo, - &event.node_did, - &event.ref_name, - &event.new_sha, - event.cid.as_deref(), - ).await; - } + reason = %reason, + "admitted gossip ref-update but a write failed" + ), + IngestOutcome::Rejected(reason) => warn!( + from = %propagation_source, + reason = %reason, + "dropped gossip ref-update" + ), + // Both arms are warn, not debug: a dropped + // ref-update is a ref this node will not + // federate and the publisher gets no + // back-pressure signal, so the budget and the + // window are named here to make a silent + // federation miss a diagnosable one. + IngestOutcome::SourceRateLimited => warn!( + from = %propagation_source, + limit = GOSSIP_SOURCE_MAX_EVENTS, + window_secs = GOSSIP_INGEST_WINDOW.as_secs(), + "dropped gossip ref-update: forwarding peer over the pre-parse ingest budget" + ), + IngestOutcome::AuthorRateLimited(did) => warn!( + from = %propagation_source, + did = %did, + limit = GOSSIP_AUTHOR_MAX_EVENTS, + window_secs = GOSSIP_INGEST_WINDOW.as_secs(), + "dropped gossip ref-update: authenticated peer over its write budget" + ), + IngestOutcome::UnsignedSourceRateLimited(source) => warn!( + from = %source, + limit = GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS, + window_secs = GOSSIP_INGEST_WINDOW.as_secs(), + "dropped gossip ref-update: forwarding peer over its unsigned-event budget" + ), } } // ── Kademlia results ────────────────────────── @@ -383,12 +1146,16 @@ pub async fn start( Some(cmd) = cmd_rx.recv() => { match cmd { P2pCommand::PublishRefUpdate(event) => { - if let Ok(bytes) = serde_json::to_vec(&event) { - let topic = gossipsub::IdentTopic::new(REF_UPDATES_TOPIC); - match swarm.behaviour_mut().gossipsub.publish(topic, bytes) { - Ok(id) => info!(msg_id = %id, repo = %event.repo, "published ref-update"), - Err(e) => warn!(err = %e, "failed to publish ref-update"), + match ref_update_publish_args(&keypair, &event) { + Ok((topic, bytes)) => { + match swarm.behaviour_mut().gossipsub.publish(topic, bytes) { + Ok(id) => info!(msg_id = %id, repo = %event.repo, "published ref-update"), + Err(e) => warn!(err = %e, "failed to publish ref-update"), + } } + // Skip the publish rather than emit something a + // verifying peer would drop anyway. + Err(e) => warn!(err = %e, "failed to sign ref-update; not publishing"), } } P2pCommand::AddKnownPeer { peer_id, addr } => { @@ -446,6 +1213,7 @@ mod tests { #[test] fn ref_update_event_round_trip_with_owner_did() { let event = RefUpdateEvent { + v: 0, node_did: "did:key:zNode".into(), pusher_did: "did:key:zPusher".into(), repo: "zOwner/myrepo".into(), @@ -456,6 +1224,7 @@ mod tests { timestamp: "2026-07-02T12:00:00Z".into(), cert_id: None, cid: None, + sig: None, }; let json = serde_json::to_value(&event).unwrap(); // owner_did must be present in the serialized output @@ -501,4 +1270,1956 @@ mod tests { let deserialized: RefUpdateEvent = serde_json::from_value(with_null).unwrap(); assert_eq!(deserialized.owner_did, None); } + + /// A fully populated event used by the wire-format tests. Every optional + /// field is Some so the serialized form exercises the widest field set. + fn populated_event() -> RefUpdateEvent { + RefUpdateEvent { + v: 0, + node_did: "did:key:zNode".into(), + pusher_did: "did:key:zPusher".into(), + repo: "zOwner/myrepo".into(), + owner_did: Some("did:key:zOwner".into()), + ref_name: "refs/heads/main".into(), + old_sha: "0000000000000000000000000000000000000000".into(), + new_sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(), + timestamp: "2026-07-02T12:00:00Z".into(), + cert_id: Some("cert-1".into()), + cid: Some("bafycid".into()), + sig: None, + } + } + + /// The load-bearing backward-compatibility test (R12). An un-upgraded peer + /// runs `from_slice::` against the PRE-CHANGE field set, so + /// this replicates that struct verbatim and proves bytes produced by the + /// new code still parse into it. If this fails, upgraded nodes' events are + /// silently dropped by every node that has not upgraded yet. + #[test] + fn signed_event_still_parses_under_the_pre_change_field_set() { + #[derive(serde::Deserialize)] + #[allow(dead_code)] + struct LegacyRefUpdateEvent { + node_did: String, + pusher_did: String, + repo: String, + #[serde(default)] + owner_did: Option, + ref_name: String, + old_sha: String, + new_sha: String, + timestamp: String, + cert_id: Option, + cid: Option, + } + + /// The same field set with unknown keys REFUSED. The permissive struct + /// above is serde's default, which drops keys it does not know, so it + /// is structurally blind to an ADDED field and would stay green against + /// the `"sig": null` regression `unsigned_event_serializes_with_no_sig_key` + /// warns about. This one sees the addition, which is what lets the two + /// assertions below state the intent: `sig` is a deliberate new key, so + /// an unsigned event is byte-compatible with the old wire form and a + /// signed one is not. + #[derive(Debug, serde::Deserialize)] + #[allow(dead_code)] + #[serde(deny_unknown_fields)] + struct StrictLegacyRefUpdateEvent { + node_did: String, + pusher_did: String, + repo: String, + #[serde(default)] + owner_did: Option, + ref_name: String, + old_sha: String, + new_sha: String, + timestamp: String, + cert_id: Option, + cid: Option, + } + + let mut event = populated_event(); + event.sig = Some("c2lnbmF0dXJl".into()); + let bytes = serde_json::to_vec(&event).unwrap(); + + let legacy: LegacyRefUpdateEvent = serde_json::from_slice(&bytes) + .expect("new-code bytes must deserialize under the old field set"); + assert_eq!(legacy.repo, "zOwner/myrepo"); + assert_eq!(legacy.owner_did, Some("did:key:zOwner".into())); + + // An UNSIGNED event carries no `sig` key at all, so it is byte-identical + // in shape to the pre-change wire form and parses even under the strict + // reader. This is what `skip_serializing_if` buys; drop it and a + // `"sig": null` key appears here and this goes red. + let unsigned_bytes = serde_json::to_vec(&populated_event()).unwrap(); + let strict: StrictLegacyRefUpdateEvent = serde_json::from_slice(&unsigned_bytes) + .expect("an unsigned event must carry no field the pre-change reader did not know"); + assert_eq!(strict.repo, "zOwner/myrepo"); + + // A SIGNED event does carry the new key, and that is intentional, not a + // compatibility bug: the permissive reader above is what makes it + // harmless. Pinning the refusal here documents `sig` as the one added + // field, so a SECOND addition cannot slip in unnoticed. + serde_json::from_slice::(&bytes) + .expect_err("a signed event must be visibly carrying the added `sig` key"); + } + + #[test] + fn legacy_json_without_sig_parses_with_sig_none() { + let old_json = serde_json::json!({ + "node_did": "did:key:zNode", + "pusher_did": "did:key:zPusher", + "repo": "zOwner/myrepo", + "ref_name": "refs/heads/main", + "old_sha": "0000000000000000000000000000000000000000", + "new_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "timestamp": "2026-07-02T12:00:00Z", + "cert_id": null, + "cid": null + }); + let deserialized: RefUpdateEvent = serde_json::from_value(old_json).unwrap(); + assert_eq!(deserialized.sig, None); + + let with_null = serde_json::json!({ + "node_did": "did:key:zNode", + "pusher_did": "did:key:zPusher", + "repo": "zOwner/myrepo", + "ref_name": "refs/heads/main", + "old_sha": "0000000000000000000000000000000000000000", + "new_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "timestamp": "2026-07-02T12:00:00Z", + "cert_id": null, + "cid": null, + "sig": null + }); + let deserialized: RefUpdateEvent = serde_json::from_value(with_null).unwrap(); + assert_eq!(deserialized.sig, None); + } + + /// `skip_serializing_if` is not cosmetic: the signing bytes are the event + /// with `sig` set to None, so a `"sig": null` key would change them and + /// break byte-identity with the legacy wire form. + #[test] + fn unsigned_event_serializes_with_no_sig_key() { + let json = serde_json::to_string(&populated_event()).unwrap(); + assert!( + !json.contains("\"sig\""), + "an unsigned event must carry no sig key at all, got: {json}" + ); + } + + /// Golden signing input, pinned byte for byte. + /// + /// If this fails, the wire signing input changed. That is not a constant to + /// re-pin: every already-signed event in flight, and every signature made by + /// a previously deployed build, becomes unverifiable against the new build, + /// so the change needs a rollout plan (ship the reader everywhere before + /// anything emits the new form). A field REORDER or rename produces exactly + /// this failure, and the emit-to-ingest round trip is structurally blind to + /// it because both sides re-serialize the same new declaration order. + const GOLDEN_SIGNING_BYTES: &str = concat!( + r#"{"node_did":"did:key:zNode","pusher_did":"did:key:zPusher","#, + r#""repo":"zOwner/myrepo","owner_did":"did:key:zOwner","#, + r#""ref_name":"refs/heads/main","#, + r#""old_sha":"0000000000000000000000000000000000000000","#, + r#""new_sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","#, + r#""timestamp":"2026-07-02T12:00:00Z","cert_id":"cert-1","cid":"bafycid"}"#, + ); + + #[test] + fn signing_bytes_match_the_golden_constant() { + let bytes = signing_bytes(&populated_event()).unwrap(); + assert_eq!( + String::from_utf8(bytes).unwrap(), + GOLDEN_SIGNING_BYTES, + "the wire signing input changed; see the comment on GOLDEN_SIGNING_BYTES" + ); + } + + /// The same golden discipline, applied to the optional shape production + /// actually emits. + /// + /// `GOLDEN_SIGNING_BYTES` above pins an event with `owner_did`, `cert_id`, + /// and `cid` all populated, and no real publish looks like that. The sole + /// production publish site, `api::repos::post_receive_replication_tail`, + /// always passes `cert_id: None`, and `cid` is None on every push whose + /// pinning has not finished. So the encoding of a null-valued optional, the + /// one carried by essentially every live event, was pinned nowhere. + /// + /// What this catches that the all-`Some` constant structurally cannot: + /// adding `skip_serializing_if = "Option::is_none"` to any of those three + /// fields omits the key rather than writing `null`, which changes the + /// signing input for every event in flight while leaving the all-`Some` + /// golden byte-identical. That is not hypothetical; injecting exactly that + /// attribute on `cert_id` left the whole suite green, both goldens passing, + /// with the production signing input silently changed. + /// + /// Frozen for the same reason as the constant above: a failure here is a + /// wire-format change that needs a rollout plan, not a constant to re-pin. + const GOLDEN_SIGNING_BYTES_ALL_NONE: &str = concat!( + r#"{"node_did":"did:key:zNode","pusher_did":"did:key:zPusher","#, + r#""repo":"zOwner/myrepo","owner_did":null,"#, + r#""ref_name":"refs/heads/main","#, + r#""old_sha":"0000000000000000000000000000000000000000","#, + r#""new_sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","#, + r#""timestamp":"2026-07-02T12:00:00Z","cert_id":null,"cid":null}"#, + ); + + /// The all-`None` optional shape: what the production publish site emits. + fn all_none_optionals_event() -> RefUpdateEvent { + RefUpdateEvent { + owner_did: None, + cert_id: None, + cid: None, + ..populated_event() + } + } + + #[test] + fn signing_bytes_of_the_all_none_shape_match_the_golden_constant() { + let bytes = signing_bytes(&all_none_optionals_event()).unwrap(); + assert_eq!( + String::from_utf8(bytes).unwrap(), + GOLDEN_SIGNING_BYTES_ALL_NONE, + "the wire signing input for null-valued optionals changed; see the comment on GOLDEN_SIGNING_BYTES_ALL_NONE" + ); + } + + /// The signature must be excluded from its own input, so a signed event and + /// its unsigned original produce identical signing bytes. + #[test] + fn signing_bytes_ignore_the_sig_field() { + let mut signed = populated_event(); + signed.sig = Some("c2lnbmF0dXJl".into()); + assert_eq!( + signing_bytes(&signed).unwrap(), + signing_bytes(&populated_event()).unwrap() + ); + } + + /// A complete wire artifact signed by a build that had no version field. + /// + /// Captured at commit e3dc6f07, from a tree where `RefUpdateEvent` carried + /// no `v` field at all, and confirmed to pass `verify_ref_update` there. + /// The capture ORDER is the whole value. A v0 event re-serializes with no + /// `"v"` key under `skip_serializing_if`, so an artifact captured after the + /// field was added is byte-identical to this one, and nothing in a test run + /// would distinguish the two. The only thing that makes this artifact + /// evidence rather than decoration is that the code under test did not + /// exist when it was produced. + /// + /// Never regenerate it. A signature the test just produced is + /// self-consistent by construction: it proves the current encoder agrees + /// with the current verifier, which stays true of an encoding that has + /// silently stopped accepting every event already in flight. That is the + /// only failure this constant can see, and regenerating it is precisely + /// how you blind it. Same discipline as `GOLDEN_SIGNING_BYTES`, for the + /// same reason: freeze it forever. `LEGACY_SIGNED_EVENT_V0_SHA256` below is + /// what makes "never regenerate it" a check rather than a request. + /// + /// Non-degenerate on purpose: `owner_did`, `cert_id`, and `cid` are all + /// populated, so the interaction between the optional fields' encoding and + /// the version field's is pinned rather than left unexercised by an + /// artifact that happened to carry none of them. + const LEGACY_SIGNED_EVENT_V0: &str = r#"{"node_did":"did:key:z6MkiAJwX3dtfEY6KGeDDgxXB6ZZWCAxTSHDtJEyUVynqYtq","pusher_did":"did:key:zPusher","repo":"zOwner/myrepo","owner_did":"did:key:zOwner","ref_name":"refs/heads/main","old_sha":"0000000000000000000000000000000000000000","new_sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","timestamp":"2026-07-02T12:00:00Z","cert_id":"cert-1","cid":"bafycid","sig":"-lH5aObROlqoTFjnjSXjbDgCVscLfVaKb1Y1gJL1tVsiBZlZnLKi55QgSo0ALTNtI_DyKo0ColzJMxL7w7ZODQ"}"#; + + /// SHA-256 of `LEGACY_SIGNED_EVENT_V0`, hex, lowercase. This is what makes + /// the capture claim above checkable instead of merely attested. + /// + /// The obvious guard does not work, which is why this one exists. Asserting + /// the artifact carries no `"v"` key proves nothing about its provenance: a + /// v0 event re-serializes with no version key under `skip_serializing_if`, + /// so an artifact regenerated from CURRENT code carries no `"v"` key either + /// and that assertion passes against precisely the regeneration it was meant + /// to refuse. A digest has no such blind spot. Any edit to those bytes, + /// regeneration included, moves it. + /// + /// Frozen alongside the artifact. If it fails, the constant was edited: + /// restore the original from commit e3dc6f07 rather than re-pinning the + /// digest, since re-pinning is exactly the act of blinding the test that the + /// "never regenerate it" paragraph above warns against. + const LEGACY_SIGNED_EVENT_V0_SHA256: &str = + "2482e053c8ab1841d784f523f1ef5e3d0bd5f9d563565af8fca8dd34a1e264fc"; + + /// The compatibility test the rest of this module cannot substitute for: + /// the only one here that verifies an artifact it did not itself sign. + /// + /// Every other signature test in this file signs and verifies in one + /// breath, which is self-consistent under any field set. Change the signed + /// field set on both sides and they all stay green forever while every + /// event a deployed peer already emitted becomes unverifiable. Driving a + /// pre-change artifact through the post-change verifier is the one + /// observation that can tell those two worlds apart. + #[test] + fn an_event_signed_before_the_version_field_existed_still_verifies() { + // What this checks, exactly: that the constant still holds the bytes + // captured at e3dc6f07, byte for byte. It does not, and cannot, observe + // that those bytes predate the version field; that is established by the + // capture commit and by review, not by anything a test run can see. What + // the digest does buy is that a later edit which regenerates the artifact + // from current code fails HERE, loudly, instead of quietly decaying this + // test into the fresh-artifact round trip it exists to not be. + use sha2::{Digest, Sha256}; + let digest = hex::encode(Sha256::digest(LEGACY_SIGNED_EVENT_V0.as_bytes())); + assert_eq!( + digest, LEGACY_SIGNED_EVENT_V0_SHA256, + "the frozen legacy artifact was edited; restore it from commit e3dc6f07 rather than \ + re-pinning this digest" + ); + + // Exactly the bytes a peer would receive, straight off the wire. + let event: RefUpdateEvent = serde_json::from_slice(LEGACY_SIGNED_EVENT_V0.as_bytes()) + .expect("an event from before the version field must still parse"); + verify_ref_update(&event).expect( + "an event signed before the version field existed must still verify; the signed \ + field set is a wire format and this build has changed it", + ); + } + + /// Build a populated event whose `node_did` is the given keypair's DID. + fn event_for(keypair: &Keypair) -> RefUpdateEvent { + let mut event = populated_event(); + event.node_did = keypair.did().to_string(); + event + } + + #[test] + fn sign_then_verify_round_trips() { + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + assert!(event.sig.is_some(), "signing must populate sig"); + + // Through the wire, since that is how a peer receives it. + let bytes = serde_json::to_vec(&event).unwrap(); + let received: RefUpdateEvent = serde_json::from_slice(&bytes).unwrap(); + verify_ref_update(&received).expect("a correctly signed event must verify"); + assert_eq!(received.repo, "zOwner/myrepo"); + assert_eq!(received.node_did, keypair.did().to_string()); + } + + /// A cryptographically valid signature that does not bind the claimed + /// identity. This is the RUSTSEC-2022-0009 shape: libp2p-core accepted a + /// valid signature without checking it derived the claimed peer id, so the + /// signature proved someone signed, not that the claimed party did. Here + /// keypair A signs an event claiming keypair B's DID; the bytes carry a + /// real signature, and verification must still refuse it because it does + /// not verify against the key behind `node_did`. + #[test] + fn a_signature_that_does_not_bind_the_claimed_did_is_rejected() { + let signer = Keypair::generate(); + let claimed = Keypair::generate(); + let mut event = event_for(&claimed); + // Signed by `signer` over bytes that name `claimed` as node_did. + sign_ref_update(&signer, &mut event).unwrap(); + assert!(event.sig.is_some()); + + verify_ref_update(&event) + .expect_err("a signature by a key other than node_did's must be refused"); + } + + #[test] + fn tampering_with_a_signed_field_fails_verification() { + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + verify_ref_update(&event).expect("baseline must verify before tampering"); + + for tamper in [ + |e: &mut RefUpdateEvent| e.repo = "attacker/evil".into(), + |e: &mut RefUpdateEvent| e.ref_name = "refs/heads/attacker".into(), + |e: &mut RefUpdateEvent| e.new_sha = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into(), + |e: &mut RefUpdateEvent| e.old_sha = "cccccccccccccccccccccccccccccccccccccccc".into(), + |e: &mut RefUpdateEvent| e.pusher_did = "did:key:zAttacker".into(), + |e: &mut RefUpdateEvent| e.owner_did = Some("did:key:zAttacker".into()), + |e: &mut RefUpdateEvent| e.timestamp = "2030-01-01T00:00:00Z".into(), + |e: &mut RefUpdateEvent| e.cert_id = Some("cert-2".into()), + |e: &mut RefUpdateEvent| e.cid = Some("bafyother".into()), + ] { + let mut tampered = event.clone(); + tamper(&mut tampered); + verify_ref_update(&tampered) + .expect_err("mutating any signed field must fail verification"); + } + } + + /// The version is inside the signed region, in both directions, and the + /// second direction is the one skip-when-zero creates. + /// + /// A version key beside the signature rather than under it is + /// attacker-mutable and proves nothing, so downgrading a v1 event to v0 + /// has to be as detectable as upgrading a v0 event to v1. The two + /// directions are not symmetric here: flipping 0 to 1 ADDS a key to the + /// signing bytes, while flipping 1 to 0 REMOVES one, and only the second + /// exercises the `skip_serializing_if` arm. Testing one direction would + /// leave an encoding that emits the key unconditionally, or one that never + /// emits it, indistinguishable from the correct one. + #[test] + fn the_version_is_covered_by_the_signature_in_both_directions() { + let keypair = Keypair::generate(); + + // Signed at v0, where the key is absent from the signing bytes. + // Raising it on the received copy makes the key appear. + let mut at_zero = event_for(&keypair); + sign_ref_update(&keypair, &mut at_zero).unwrap(); + verify_ref_update(&at_zero).expect("baseline must verify before tampering"); + let mut raised = at_zero.clone(); + raised.v = 1; + verify_ref_update(&raised) + .expect_err("raising the version on a signed event must fail verification"); + + // Signed at v1, where the key IS in the signing bytes. Lowering it to + // zero on the received copy makes the key vanish, which is the arm a + // one-directional test never reaches. + let mut at_one = event_for(&keypair); + at_one.v = 1; + sign_ref_update(&keypair, &mut at_one).unwrap(); + verify_ref_update(&at_one).expect("a v1 event must verify against its own signature"); + let mut lowered = at_one.clone(); + lowered.v = 0; + verify_ref_update(&lowered) + .expect_err("lowering the version on a signed event must fail verification"); + } + + /// An event from a peer that predates the field parses as v0, and + /// re-serializing it reproduces the versionless bytes. That round trip IS + /// the v0 verification path: there is no separate legacy code path to keep + /// working, only the property that the default and the skip agree. + #[test] + fn json_without_a_version_key_parses_as_v0_and_signs_without_one() { + let old_json = serde_json::json!({ + "node_did": "did:key:zNode", + "pusher_did": "did:key:zPusher", + "repo": "zOwner/myrepo", + "ref_name": "refs/heads/main", + "old_sha": "0000000000000000000000000000000000000000", + "new_sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "timestamp": "2026-07-02T12:00:00Z", + "cert_id": null, + "cid": null + }); + let deserialized: RefUpdateEvent = serde_json::from_value(old_json).unwrap(); + assert_eq!( + deserialized.v, 0, + "an event with no version key is the versionless form, which is v0" + ); + + let bytes = String::from_utf8(signing_bytes(&deserialized).unwrap()).unwrap(); + assert!( + !bytes.contains("\"v\""), + "a v0 event's signing bytes must carry no version key, or every peer \ + running the pre-version build computes different bytes; got: {bytes}" + ); + } + + #[test] + fn an_event_with_no_signature_is_rejected() { + let keypair = Keypair::generate(); + let event = event_for(&keypair); + assert_eq!(event.sig, None); + verify_ref_update(&event).expect_err("an unsigned event must not verify"); + } + + /// Two surfaces judging the same input answer with the same sentence. The + /// literals here are copied from `PeerWriteDenied` in db/mod.rs on purpose: + /// if that wording changes, this goes red rather than letting the gossip + /// surface drift into its own vocabulary for the same refusal. + #[test] + fn a_non_did_key_node_did_is_rejected_with_the_shared_sentence() { + let keypair = Keypair::generate(); + let mut event = populated_event(); + event.node_did = "did:web:example.com".into(); + sign_ref_update(&keypair, &mut event).unwrap(); + + let err = verify_ref_update(&event).expect_err("did:web must never authenticate"); + assert_eq!( + err, + "methodNotSupported: only did:key peers can be registered without a proof of control: did:web:example.com" + ); + } + + // ── Ingest-path tests ───────────────────────────────────────────────── + // + // Every rejection case asserts BOTH sinks are empty: `received_ref_updates` + // and `sync_queue`. They are two separate writes, so a guard that stops one + // and not the other is exactly the bug this path is being fixed for, and a + // row-count-only assertion would pass against that half-fix. + + use sqlx::PgPool; + + async fn ingest_db(pool: &PgPool) -> Db { + let db = Db::for_testing(pool.clone()); + db.run_migrations() + .await + .expect("test schema migrations should apply"); + db + } + + async fn seed_peer(pool: &PgPool, did: &str) { + sqlx::query( + "INSERT INTO peers (did, http_url, last_seen, last_ping_ok, announced_at) + VALUES ($1, $2, $3, FALSE, $3)", + ) + .bind(did) + .bind("https://peer.example.com") + .bind(Utc::now().to_rfc3339()) + .execute(pool) + .await + .expect("seed peer"); + } + + async fn count(pool: &PgPool, table: &str) -> i64 { + sqlx::query_scalar::<_, i64>(&format!("SELECT COUNT(*) FROM {table}")) + .fetch_one(pool) + .await + .expect("count rows") + } + + /// Both sinks, asserted separately. `context` names the case so a failure + /// says which mode and which guard let the write through. + async fn assert_nothing_written(pool: &PgPool, context: &str) { + assert_eq!( + count(pool, "received_ref_updates").await, + 0, + "{context}: a rejected event must write no received_ref_updates row" + ); + assert_eq!( + count(pool, "sync_queue").await, + 0, + "{context}: a rejected event must enqueue no sync_queue row" + ); + } + + fn bytes_of(event: &RefUpdateEvent) -> Vec { + serde_json::to_vec(event).expect("serialize event") + } + + /// Zero the `peer_exists` tally, then read it back. The pair exists so a + /// test can assert the DATABASE WAS NEVER TOUCHED, which no outcome value + /// can express: a guard that runs above the debit and a guard that runs + /// above the round trip return the identical `Rejected`. + fn reset_peer_exists_calls() { + PEER_EXISTS_CALLS.with(|calls| calls.set(0)); + } + + fn peer_exists_calls() -> usize { + PEER_EXISTS_CALLS.with(|calls| calls.get()) + } + + /// Ingest one event against a limiter with no history, for the cases that + /// are about a guard other than the rate brake. The rate-limit tests below + /// hold one limiter across calls instead, since that is the state they + /// assert on. + async fn ingest_with_fresh_limiter( + db: &Db, + require_signed: bool, + auto_sync: bool, + data: &[u8], + propagation_source: &PeerId, + ) -> IngestOutcome { + ingest_ref_update( + db, + &IngestLimiters::new(), + require_signed, + auto_sync, + data, + propagation_source, + ) + .await + } + + fn rejection_reason(outcome: IngestOutcome, context: &str) -> String { + match outcome { + IngestOutcome::Rejected(reason) => reason, + IngestOutcome::Accepted => panic!("{context}: the event must be rejected"), + IngestOutcome::UnsignedAdmitted => { + panic!("{context}: the event must be rejected, not admitted unsigned") + } + IngestOutcome::WriteFailed(reason) => { + panic!("{context}: the event must be rejected by a guard, not admitted and then failed to write: {reason}") + } + IngestOutcome::SourceRateLimited + | IngestOutcome::AuthorRateLimited(_) + | IngestOutcome::UnsignedSourceRateLimited(_) => { + panic!("{context}: the event must be rejected by a guard, not by a rate brake") + } + } + } + + /// Drive one prepared event through ingest in BOTH flag modes and assert it + /// is refused by a guard, with no row and no queue entry either way. + /// + /// Six tests wrote this loop out by hand. Sharing it is not only less + /// repetition: it means "in both modes" has ONE definition, so a case that + /// only ever exercised `require_signed=true` cannot creep in unnoticed under + /// a name that promises both. + /// + /// `label` names the witness ("tampered event"), and the mode is appended, + /// so a failure still says which of the two directions broke. + /// + /// `expected_reason` is `Some((reason, why))` for the cases where the + /// SENTENCE is the thing under test, carrying its own explanation of why + /// that wording is load-bearing, since a shared assertion message could not + /// say anything specific enough to be useful. `None` where the test only + /// claims a guard refused it, and the particular guard is pinned by the + /// witness rather than by the string. + async fn assert_rejected_in_both_modes( + db: &Db, + pool: &PgPool, + data: &[u8], + label: &str, + expected_reason: Option<(&str, &str)>, + ) { + for require_signed in [true, false] { + let context = format!("{label}, require_signed={require_signed}"); + let outcome = + ingest_with_fresh_limiter(db, require_signed, true, data, &PeerId::random()).await; + assert_nothing_written(pool, &context).await; + let reason = rejection_reason(outcome, &context); + if let Some((expected, why)) = expected_reason { + assert_eq!(reason, expected, "{context}: {why}"); + } + } + } + + /// R1, the core must-not: enforcement on, an unsigned event that merely + /// CLAIMS a known peer's DID writes nothing. Anyone on the open mesh can + /// send these, so this is the whole point of the unit. + #[sqlx::test] + async fn flag_on_unsigned_event_claiming_a_known_peer_writes_nothing(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let event = event_for(&keypair); + seed_peer(&pool, &event.node_did).await; + + let outcome = + ingest_with_fresh_limiter(&db, true, true, &bytes_of(&event), &PeerId::random()).await; + + assert_nothing_written(&pool, "unsigned event with enforcement on").await; + rejection_reason(outcome, "unsigned event with enforcement on"); + } + + /// An event from a build newer than this one is refused AS a version + /// problem, in its own words. + /// + /// The witness is valid in every other respect on purpose: correctly + /// signed by a key that resolves from its own `node_did`, seeded as a known + /// peer, well-formed slug. So the version is the only rule that can reject + /// it, and without the guard it is not rejected at all, it is ACCEPTED, and + /// this node writes rows whose field semantics it does not know. + /// + /// The assertion is on the SENTENCE, not merely on the refusal, and that is + /// the point of the test rather than a detail of it. An honest v1 + /// publisher's events would otherwise land as a signature mismatch, + /// indistinguishable in logs and counters from forgery, and an operator + /// watching a partition form would be reading an accusation instead of a + /// version skew. Asserting only that the event was rejected cannot tell the + /// two apart. + #[sqlx::test] + async fn an_unknown_version_is_refused_as_an_unknown_version(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + event.v = CURRENT_REF_UPDATE_VERSION + 1; + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + // The witness must be rejectable only by the version rule, so confirm + // the signature path would have admitted it. + verify_ref_update(&event).expect("the witness must be correctly signed"); + + assert_rejected_in_both_modes( + &db, + &pool, + &bytes_of(&event), + "v1 event", + Some(( + "unsupported ref-update event version 1; this build understands version 0", + "an unknown version must be named as one, never reported as a signature failure", + )), + ) + .await; + } + + /// R2: a cryptographically valid signature that does not bind the claimed + /// identity (the RUSTSEC-2022-0009 shape). Rejected in BOTH modes, because + /// a present-but-wrong signature is forgery, never a legacy peer. + #[sqlx::test] + async fn a_signature_by_another_key_is_rejected_in_both_modes(pool: PgPool) { + let db = ingest_db(&pool).await; + let signer = Keypair::generate(); + let claimed = Keypair::generate(); + let mut event = event_for(&claimed); + sign_ref_update(&signer, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + assert_rejected_in_both_modes(&db, &pool, &bytes_of(&event), "foreign-key signature", None) + .await; + } + + /// R2: a signed event whose payload was edited after signing. + #[sqlx::test] + async fn a_tampered_signed_event_is_rejected_in_both_modes(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + event.new_sha = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into(); + + assert_rejected_in_both_modes(&db, &pool, &bytes_of(&event), "tampered event", None).await; + } + + /// R11: a non-did:key `node_did` cannot be authenticated by design, and the + /// refusal answers in the SAME sentence as the peers-table gate. + /// + /// What this test guards is the REFUSAL WORDING, not the did-method + /// gate itself. The event here is signed, so with that gate deleted + /// `verify_ref_update` resolves `node_did` itself and returns the identical + /// sentence; the assertion below cannot tell the two apart. The test that + /// isolates the gate is + /// `an_unsigned_non_did_key_event_from_a_known_peer_is_rejected_by_the_did_method_gate`. + #[sqlx::test] + async fn a_non_did_key_node_did_is_rejected_in_both_modes(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = populated_event(); + event.node_did = "did:web:example.com".into(); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + assert_rejected_in_both_modes( + &db, + &pool, + &bytes_of(&event), + "did:web node_did", + Some(( + "methodNotSupported: only did:key peers can be registered without a proof of control: did:web:example.com", + "the gossip surface must reuse the peers-table refusal sentence", + )), + ) + .await; + } + + /// The load-bearing test for the did-method gate, and the ONLY + /// combination that isolates it. + /// + /// Three inputs, each chosen to take one of the other guards out of the + /// picture. The event is UNSIGNED, so `verify_ref_update` is never called + /// and cannot resolve `node_did` on the gate's behalf. `require_signed` is + /// FALSE, so the unsigned branch admits it rather than refusing it for a + /// missing signature. The did:web DID is seeded into `peers`, so the + /// known-peer gate admits it too. Everything downstream (the repo slug) is + /// valid. With the gate present this is refused with the shared sentence; + /// delete the gate and this exact event is accepted and written. + #[sqlx::test] + async fn an_unsigned_non_did_key_event_from_a_known_peer_is_rejected_by_the_did_method_gate( + pool: PgPool, + ) { + let db = ingest_db(&pool).await; + let mut event = populated_event(); + event.node_did = "did:web:example.com".into(); + assert_eq!( + event.sig, None, + "the gate must be what decides, not the sig" + ); + seed_peer(&pool, &event.node_did).await; + + let context = "unsigned did:web event from a seeded peer, require_signed=false"; + let outcome = + ingest_with_fresh_limiter(&db, false, true, &bytes_of(&event), &PeerId::random()).await; + + assert_nothing_written(&pool, context).await; + assert_eq!( + rejection_reason(outcome, context), + "methodNotSupported: only did:key peers can be registered without a proof of control: did:web:example.com", + "{context}: only the did-method gate can refuse this, so this is what goes red if it is removed" + ); + } + + /// R3: authentication is not authorization. A correctly signed event from a + /// DID nobody registered is still refused, mirroring the HTTP twin's + /// unconditional known-peer gate. + #[sqlx::test] + async fn a_signed_event_from_an_unknown_peer_is_rejected_in_both_modes(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + // Deliberately NOT seeded into the peers table. + + assert_rejected_in_both_modes(&db, &pool, &bytes_of(&event), "unknown peer DID", None) + .await; + } + + /// R4: the #272 slug guard, on this transport too. The slug reaches a + /// `PathBuf::join` in the sync worker, so it is rejected before the row and + /// before the queue entry. + #[sqlx::test] + async fn a_traversal_slug_is_rejected_before_any_write(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + event.repo = "../../x".into(); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + assert_rejected_in_both_modes(&db, &pool, &bytes_of(&event), "traversal slug", None).await; + } + + /// R6: the acceptance path, which is what keeps federation alive. A guard + /// that rejects everything would pass every test above and fail here. + #[sqlx::test] + async fn flag_on_signed_known_peer_event_is_accepted_end_to_end(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + let source = PeerId::random(); + + let outcome = ingest_with_fresh_limiter(&db, true, true, &bytes_of(&event), &source).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "a correctly signed event from a known peer must be accepted, got {outcome:?}" + ); + + let row: (String, String, String, String, String) = sqlx::query_as( + "SELECT node_did, pusher_did, repo, ref_name, from_peer FROM received_ref_updates", + ) + .fetch_one(&pool) + .await + .expect("exactly one ref-update row"); + assert_eq!(row.0, event.node_did); + assert_eq!(row.1, "did:key:zPusher"); + assert_eq!(row.2, "zOwner/myrepo"); + assert_eq!(row.3, "refs/heads/main"); + // R9: from_peer records the FORWARDER, not the author. + assert_eq!(row.4, source.to_string()); + + assert_eq!( + count(&pool, "sync_queue").await, + 1, + "auto_sync on must enqueue the accepted event" + ); + } + + /// The auto_sync=false half: the row lands, the queue stays empty. Without + /// it, an ingest that enqueued unconditionally would go unnoticed. + #[sqlx::test] + async fn accepted_event_does_not_enqueue_when_auto_sync_is_off(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + let outcome = + ingest_with_fresh_limiter(&db, true, false, &bytes_of(&event), &PeerId::random()).await; + assert!(matches!(outcome, IngestOutcome::Accepted)); + + assert_eq!(count(&pool, "received_ref_updates").await, 1); + assert_eq!( + count(&pool, "sync_queue").await, + 0, + "auto_sync off must not enqueue" + ); + } + + /// R7, the rolling-upgrade window: with enforcement off, an unsigned event + /// from a KNOWN peer is still admitted. Turning the flag on is the + /// operator's step, not a code change, so this path has to keep working + /// until they take it. + /// + /// It is `UnsignedAdmitted`, not `Accepted`: the event wrote rows without + /// authenticating its sender, and the two must not be observably the same. + /// + /// The ingest path also emits a `warn!` pointing at the flag on this + /// branch. Nothing here asserts that, so the log line is uncovered: + /// deleting it leaves this test green. Say so rather than implying the + /// wording is pinned. + #[sqlx::test] + async fn flag_off_unsigned_known_peer_event_is_accepted(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let event = event_for(&keypair); + assert_eq!(event.sig, None); + seed_peer(&pool, &event.node_did).await; + + let outcome = + ingest_with_fresh_limiter(&db, false, true, &bytes_of(&event), &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "an unsigned known-peer event must be admitted (distinct from Accepted) through the rolling-upgrade window, got {outcome:?}" + ); + assert_eq!(count(&pool, "received_ref_updates").await, 1); + assert_eq!(count(&pool, "sync_queue").await, 1); + } + + // ── Emit side ───────────────────────────────────────────────────────── + + /// The round trip that matters: bytes built by the emit path are fed to the + /// real ingest with enforcement ON, and must be accepted. + /// + /// Nothing else proves emit and verify agree on the signing input by + /// execution. The golden test pins the input's shape, and the helper tests + /// sign and verify through `sign_ref_update` directly, but only this one + /// exercises what the node actually puts on the wire. If it fails once the + /// fleet turns enforcement on, every node drops every other node's events. + #[sqlx::test] + async fn emitted_bytes_verify_through_ingest_with_enforcement_on(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + // Straight off the publish path: the caller hands over an unsigned + // event, exactly as `publish_ref_update` does. + let event = event_for(&keypair); + assert_eq!(event.sig, None, "the emit path is what adds the signature"); + seed_peer(&pool, &event.node_did).await; + + let bytes = signed_publish_bytes(&keypair, &event).expect("emit path must produce bytes"); + + let published: RefUpdateEvent = + serde_json::from_slice(&bytes).expect("published bytes must parse"); + assert!( + published.sig.is_some(), + "an emitted event must carry a signature" + ); + + let outcome = ingest_with_fresh_limiter(&db, true, true, &bytes, &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "bytes from the emit path must survive ingest with enforcement on, got {outcome:?}" + ); + assert_eq!(count(&pool, "received_ref_updates").await, 1); + } + + /// The publish arm's own output, driven end to end: whatever + /// `ref_update_publish_args` hands gossipsub must be signed, must go out on + /// the wire topic peers subscribe to, and must survive ingest with + /// enforcement ON. + /// + /// This is deliberately not a second copy of the round trip above. That test + /// calls `signed_publish_bytes`, one layer below the loop; this one calls + /// the function the `select!` arm calls, so an arm that stopped signing + /// would have to be rewritten rather than merely reordered to keep the suite + /// green. The topic is asserted against the literal string, not against + /// `REF_UPDATES_TOPIC`, because comparing the constant to itself proves + /// nothing: the topic name is a wire format shared with every deployed peer, + /// and renaming it silently partitions the mesh into two meshes that each + /// look healthy. + /// + /// What is still NOT observed, and is not implied to be: the `select!` arm + /// dispatching a `P2pCommand::PublishRefUpdate` into this function, and + /// `require_signed` reaching `ingest_ref_update` from `main.rs` the right + /// way round. Both live inside `p2p::start`, which needs a live swarm to + /// drive, so an inverted flag threaded from the config would still leave + /// this green. Uncovered seam, named. + #[sqlx::test] + async fn the_publish_arms_output_is_signed_and_survives_enforced_ingest(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + // Unsigned on the way in, exactly as `publish_ref_update` hands it over. + let event = event_for(&keypair); + assert_eq!( + event.sig, None, + "the publish path is what adds the signature" + ); + seed_peer(&pool, &event.node_did).await; + + let (topic, bytes) = + ref_update_publish_args(&keypair, &event).expect("the publish arm must produce args"); + + assert_eq!( + topic.to_string(), + "gitlawb/ref-updates/v1", + "the publish topic is a wire format; renaming it partitions the mesh" + ); + + let published: RefUpdateEvent = + serde_json::from_slice(&bytes).expect("published bytes must parse"); + assert!( + published.sig.is_some(), + "the bytes the loop hands gossipsub must carry a signature" + ); + + let outcome = ingest_with_fresh_limiter(&db, true, true, &bytes, &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "the publish arm's bytes must survive ingest with enforcement on, got {outcome:?}" + ); + assert_eq!(count(&pool, "received_ref_updates").await, 1); + } + + // ── Durable-write failure ───────────────────────────────────────────── + // + // `WriteFailed` is unreachable from every other test here: a rejection stops + // above both writes, and an acceptance has both succeed. So the variant that + // exists to stop `Accepted` from meaning "authenticated but never stored" + // was never once observed, and `rejection_reason` panics rather than + // distinguishes if it turns up. + // + // The failure is made REAL by dropping the target table on the live test + // database, so the error comes back from Postgres on the actual write rather + // than from a stub standing in for one. Both directions are driven, because + // the property is that the two writes are attempted INDEPENDENTLY: a test + // that only broke the first could not tell that from "the first failure + // aborts the rest", and each case therefore asserts what landed in the sink + // that was left intact. + + /// The ref-update row fails, the queue entry still lands. + #[sqlx::test] + async fn a_failed_ref_update_insert_reports_write_failed_and_still_enqueues(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + sqlx::query("DROP TABLE received_ref_updates") + .execute(&pool) + .await + .expect("drop the ref-update sink so its write genuinely fails"); + + let outcome = + ingest_with_fresh_limiter(&db, true, true, &bytes_of(&event), &PeerId::random()).await; + + match outcome { + IngestOutcome::WriteFailed(reason) => assert!( + reason.contains("failed to store received ref-update"), + "the outcome must name the write that failed, got: {reason}" + ), + other => panic!( + "an event whose durable write failed must not be reported as accepted, got {other:?}" + ), + } + + assert_eq!( + count(&pool, "sync_queue").await, + 1, + "the queue entry is a separate write and must not be lost to the row's failure" + ); + } + + /// The mirror: the queue entry fails, the ref-update row still lands. + #[sqlx::test] + async fn a_failed_enqueue_reports_write_failed_and_still_stores_the_row(pool: PgPool) { + let db = ingest_db(&pool).await; + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + sqlx::query("DROP TABLE sync_queue") + .execute(&pool) + .await + .expect("drop the queue sink so its write genuinely fails"); + + let outcome = + ingest_with_fresh_limiter(&db, true, true, &bytes_of(&event), &PeerId::random()).await; + + match outcome { + IngestOutcome::WriteFailed(reason) => assert!( + reason.contains("failed to enqueue sync"), + "the outcome must name the write that failed, got: {reason}" + ), + other => panic!( + "an event whose enqueue failed must not be reported as accepted, got {other:?}" + ), + } + + assert_eq!( + count(&pool, "received_ref_updates").await, + 1, + "the ref-update row is a separate write and must not be lost to the enqueue failure" + ); + } + + // ── Ingest rate limits ──────────────────────────────────────────────── + + /// The documented numbers, and then the check that matters: each cap + /// actually reaches the limiter production builds. Asserting the constants + /// alone leaves `IngestLimiters::new` free to pass the wrong one, and the + /// three caps are close enough in shape that a copy-paste swap reads fine. + #[test] + fn the_ingest_budgets_are_wired_as_documented() { + assert_eq!(GOSSIP_SOURCE_MAX_EVENTS, 2000); + assert_eq!(GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS, 1500); + assert_eq!(GOSSIP_AUTHOR_MAX_EVENTS, 500); + assert_eq!(GOSSIP_INGEST_WINDOW, Duration::from_secs(60)); + assert_eq!(GOSSIP_INGEST_MAX_SOURCES, 200_000); + assert_eq!(GOSSIP_INGEST_MAX_AUTHORS, 200_000); + + let limiters = IngestLimiters::new(); + for (name, limiter, cap, _) in limiters.all() { + assert_eq!( + limiter.max_requests(), + cap, + "the {name} limiter must be built with the cap it is documented to carry" + ); + } + } + + /// The key ceilings have to reach the limiters production builds, not just + /// exist as constants. Asserting the constants alone leaves + /// `IngestLimiters::new` free to call the unbounded-ish `new`, which + /// silently swaps in `DEFAULT_MAX_KEYS` and passes every other test here. + /// + /// Driving 200_000 distinct keys to observe the cap by behavior would cost + /// more than it proves, so this reads the wired values through the + /// test-only accessor instead. What it does NOT cover is the eviction + /// behavior at the cap; that is `rate_limit`'s own test's job. + #[test] + fn every_ingest_limiter_carries_its_key_ceiling() { + let limiters = IngestLimiters::new(); + for (name, limiter, _, ceiling) in limiters.all() { + assert_eq!( + limiter.max_keys(), + ceiling, + "the {name} limiter must be built bounded by its documented key ceiling" + ); + } + } + + /// The ingest limiters live as locals of the swarm task, so the periodic + /// `sweep_rate_limiters` in `main.rs` cannot see them and its completeness + /// test cannot cover them. This is that pair's counterpart: it proves the + /// swarm loop's own sweep actually reclaims a key from EVERY ingest limiter, + /// not just the first one someone remembered. + /// + /// Both loops walk `all()`, which destructures `Self`, and `cleanup` walks + /// `each()`, which does the same. So a fourth limiter added later cannot + /// slip past this the way `/ipfs` slipped past the `AppState` sweeper: it + /// fails to compile in three places before it can fail silently in one. + /// + /// The short window comes from `with_window` rather than a limiter built by + /// hand, so the thing under test is the real struct with its real caps. + #[tokio::test] + async fn the_swarm_sweep_evicts_expired_keys_from_every_ingest_limiter() { + let window = Duration::from_millis(30); + let limiters = IngestLimiters::with_window(window); + + for (name, limiter, _, _) in limiters.all() { + assert!( + limiter.check("forwarding-peer").await, + "the {name} limiter must admit the first event, or this test proves nothing" + ); + assert_eq!(limiter.tracked_keys().await, 1); + } + + tokio::time::sleep(window * 3).await; + limiters.cleanup().await; + + for (name, limiter, _, _) in limiters.all() { + assert_eq!( + limiter.tracked_keys().await, + 0, + "the {name} ingest limiter was not swept" + ); + } + } + + /// Every outcome the ingest path can return has to reach the counter under + /// its own label, because `/metrics` is the only externally observable + /// surface this daemon exposes and an unlabelled outcome is an invisible + /// one. The exhaustive match in `metric_label` is what makes a seventh + /// variant a compile error; this pins the labels themselves, since a + /// copy-paste that gave two variants the same string would still compile and + /// would silently merge a shed reason into another on the dashboard. + #[test] + fn every_ingest_outcome_carries_a_distinct_metric_label() { + let labels = [ + IngestOutcome::Accepted.metric_label(), + IngestOutcome::UnsignedAdmitted.metric_label(), + IngestOutcome::WriteFailed("db down".into()).metric_label(), + IngestOutcome::Rejected("malformed".into()).metric_label(), + IngestOutcome::SourceRateLimited.metric_label(), + IngestOutcome::AuthorRateLimited("did:key:a".into()).metric_label(), + IngestOutcome::UnsignedSourceRateLimited("peer".into()).metric_label(), + ]; + assert_eq!( + labels, + [ + "accepted", + "unsigned_admitted", + "write_failed", + "rejected", + "source_rate_limited", + "author_rate_limited", + "unsigned_source_rate_limited", + ] + ); + + let unique: std::collections::HashSet<&str> = labels.iter().copied().collect(); + assert_eq!( + unique.len(), + labels.len(), + "two outcomes share a label, which would merge them in the counter" + ); + } + + /// The ordering test, and the reason the pre-parse brake exists at all. + /// Signature verification is the expensive step, so a limiter sitting after + /// it lets an unauthenticated flood buy exactly the CPU the brake was meant + /// to protect. + /// + /// This discriminates the ordering by execution rather than by inspection. + /// The flood is garbage that neither parses nor verifies. With the check + /// first, those messages spend the source's budget and the next event, a + /// perfectly valid signed known-peer one, comes back `SourceRateLimited`. + /// Move the check below the parse or below verification and the garbage + /// never reaches the limiter, the budget is untouched, and that last event + /// is accepted: this assertion is what goes red. + #[sqlx::test] + async fn rate_limit_runs_before_parse_and_signature_verification(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let source = PeerId::random(); + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + for i in 0..GOSSIP_SOURCE_MAX_EVENTS { + let outcome = + ingest_ref_update(&db, &limiters, true, true, b"not json at all", &source).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "flood message {i} is inside the budget, so it is admitted and then dropped as malformed, got {outcome:?}" + ); + } + + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &source).await; + assert!( + matches!(outcome, IngestOutcome::SourceRateLimited), + "the event past the budget from one source inside the window must be rate limited; \ + an unverifiable flood has to spend the budget, which only happens if the \ + check precedes the parse and the signature work. Got {outcome:?}" + ); + assert_nothing_written(&pool, "source over its ingest budget").await; + } + + /// The pre-parse budget is per source peer, not one global bucket. Without + /// this, one noisy or hostile mesh source would silence the whole fleet. + #[sqlx::test] + async fn rate_limit_is_per_source_not_global(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let throttled = PeerId::random(); + let other = PeerId::random(); + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + for _ in 0..GOSSIP_SOURCE_MAX_EVENTS { + ingest_ref_update(&db, &limiters, true, true, b"not json at all", &throttled).await; + } + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &throttled).await; + assert!( + matches!(outcome, IngestOutcome::SourceRateLimited), + "the first source must be over budget, got {outcome:?}" + ); + + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &other).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "a second peer keeps its own budget while the first is throttled, got {outcome:?}" + ); + assert_eq!( + count(&pool, "received_ref_updates").await, + 1, + "the second peer's event is the only one that should have been written" + ); + } + + /// FINDING 2, the victim-denial case, and the reason the tight bound moved + /// off `propagation_source`. Junk relayed through an honest neighbour is + /// charged to that neighbour's key, because it is the only identity + /// available before parsing. What must NOT follow is that the neighbour + /// stops being a usable path for real traffic: a correctly signed event + /// from a known author arriving down the same edge is still accepted. + /// + /// The flood here runs to one below the source ceiling, which is far past + /// the old 60-per-source bound, so under that bound this event is the one + /// that came back rate limited. + /// + /// What this canNOT express: the junk never reaches the author limiter at + /// all (it does not parse), and whether the neighbour's own budget is spent + /// on OTHER receivers is a property of the live mesh, which needs the swarm + /// loop and is out of scope here. + #[sqlx::test] + async fn a_junk_flood_down_one_edge_does_not_deny_a_valid_author_on_that_edge(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let neighbour = PeerId::random(); + let keypair = Keypair::generate(); + let mut event = event_for(&keypair); + sign_ref_update(&keypair, &mut event).unwrap(); + seed_peer(&pool, &event.node_did).await; + + for _ in 0..GOSSIP_SOURCE_MAX_EVENTS - 1 { + ingest_ref_update(&db, &limiters, true, true, b"not json at all", &neighbour).await; + } + + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &neighbour).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "an honest author must still get through an edge that carried a junk flood, got {outcome:?}" + ); + assert_eq!(count(&pool, "received_ref_updates").await, 1); + assert_eq!(count(&pool, "sync_queue").await, 1); + } + + /// The per-author budget, end to end: a full budget of signed events from + /// one known author is accepted, the next one is refused and writes NOTHING + /// to either sink, and a DIFFERENT known author on the SAME mesh edge is + /// unaffected. + /// + /// The last assertion is the other half of FINDING 2. With the tight bound + /// keyed on `propagation_source`, one author exhausting the budget took the + /// edge down for every author sharing it; keyed on the authenticated DID, + /// the cost lands on the principal that incurred it. + #[sqlx::test] + async fn the_author_budget_bounds_one_author_without_touching_another(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let source = PeerId::random(); + let noisy = Keypair::generate(); + let quiet = Keypair::generate(); + seed_peer(&pool, &noisy.did().to_string()).await; + seed_peer(&pool, &quiet.did().to_string()).await; + + let mut event = event_for(&noisy); + sign_ref_update(&noisy, &mut event).unwrap(); + for i in 0..GOSSIP_AUTHOR_MAX_EVENTS { + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &source).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "event {i} is inside the author budget and must be accepted, got {outcome:?}" + ); + } + let accepted = GOSSIP_AUTHOR_MAX_EVENTS as i64; + assert_eq!(count(&pool, "received_ref_updates").await, accepted); + assert_eq!(count(&pool, "sync_queue").await, accepted); + + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &source).await; + match &outcome { + IngestOutcome::AuthorRateLimited(did) => assert_eq!( + did, &event.node_did, + "the refusal must name the author it was charged to" + ), + other => panic!("the over-budget author must be refused, got {other:?}"), + } + // Both sinks, separately: an over-budget refusal is a refusal, so + // neither the row nor the queue entry may move. + assert_eq!( + count(&pool, "received_ref_updates").await, + accepted, + "an over-budget refusal must write no received_ref_updates row" + ); + assert_eq!( + count(&pool, "sync_queue").await, + accepted, + "an over-budget refusal must enqueue no sync_queue row" + ); + + let mut other_event = event_for(&quiet); + other_event.repo = "zOwner/otherrepo".into(); + sign_ref_update(&quiet, &mut other_event).unwrap(); + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&other_event), &source).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "a second author sharing the mesh edge keeps its own budget, got {outcome:?}" + ); + assert_eq!(count(&pool, "received_ref_updates").await, accepted + 1); + } + + /// The victim must-not, and the whole reason the author budget is charged + /// only to a proven author. With enforcement off, an unsigned event's + /// `node_did` is asserted, not proven: anyone on the open mesh can name a + /// registered DID. If that claim debits the author bucket, an attacker + /// spends a NAMED victim's budget from an unrelated `PeerId` and the + /// victim's own genuine signed pushes come back `AuthorRateLimited`. The + /// attacker picks the target, which is what makes this a P1 rather than a + /// fairness wart. + /// + /// The flood runs to exactly the author cap, so under the pre-fix shape the + /// signed event below is the first one past it. + #[sqlx::test] + async fn an_unsigned_flood_naming_a_victim_does_not_spend_the_victim_budget(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let attacker_edge = PeerId::random(); + let victim = Keypair::generate(); + seed_peer(&pool, &victim.did().to_string()).await; + + // Unsigned, claiming the victim's DID, relayed from an edge the victim + // has nothing to do with. + let claim = bytes_of(&event_for(&victim)); + for i in 0..GOSSIP_AUTHOR_MAX_EVENTS { + let outcome = + ingest_ref_update(&db, &limiters, false, true, &claim, &attacker_edge).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "unsigned event {i} is inside every budget and is admitted in the rolling-upgrade window, got {outcome:?}" + ); + } + + let mut genuine = event_for(&victim); + genuine.ref_name = "refs/heads/genuine".into(); + sign_ref_update(&victim, &mut genuine).unwrap(); + let outcome = ingest_ref_update( + &db, + &limiters, + false, + true, + &bytes_of(&genuine), + &PeerId::random(), + ) + .await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "the victim's own signed push must survive an unsigned flood that merely claimed its DID, got {outcome:?}" + ); + // The sink moved, not just the outcome class: `Accepted` alone would + // still hold if the writes had been skipped. + assert_eq!( + count(&pool, "received_ref_updates").await, + GOSSIP_AUTHOR_MAX_EVENTS as i64 + 1, + "the victim's signed event must land its own row" + ); + assert_eq!( + count(&pool, "sync_queue").await, + GOSSIP_AUTHOR_MAX_EVENTS as i64 + 1, + "the victim's signed event must land its own queue entry" + ); + } + + /// The brake must sit in front of the work it bounds, not behind it. + /// + /// An author already over its budget is going to be shed no matter what the + /// peer lookup says, so paying a Postgres round trip first is work done for a + /// request that was never going to be admitted. A signed author can drive that + /// at the pre-parse source rate, and with peer-id rotation past it. + /// + /// The counter is what makes this a real assertion rather than a restatement + /// of the code: it observes the lookup NOT happening, which reading the + /// ordering cannot do. + #[sqlx::test] + async fn an_over_budget_author_is_shed_without_a_peer_lookup(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let author = Keypair::generate(); + seed_peer(&pool, &author.did().to_string()).await; + + // A distinct signed event per iteration: identical bytes are a replay and + // would be dropped by that guard instead of reaching the budget. + let signed_nth = |n: usize| { + let mut e = event_for(&author); + e.ref_name = format!("refs/heads/b{n}"); + sign_ref_update(&author, &mut e).unwrap(); + e + }; + for i in 0..GOSSIP_AUTHOR_MAX_EVENTS { + let outcome = ingest_ref_update( + &db, + &limiters, + true, + true, + &bytes_of(&signed_nth(i)), + &PeerId::random(), + ) + .await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "event {i} is inside the budget, got {outcome:?}" + ); + } + + reset_peer_exists_calls(); + let outcome = ingest_ref_update( + &db, + &limiters, + true, + true, + &bytes_of(&signed_nth(GOSSIP_AUTHOR_MAX_EVENTS)), + &PeerId::random(), + ) + .await; + assert!( + matches!(outcome, IngestOutcome::AuthorRateLimited(_)), + "the over-budget author must be shed, got {outcome:?}" + ); + assert_eq!( + peer_exists_calls(), + 0, + "an over-budget author must be shed BEFORE the peer lookup; the brake \ + belongs in front of the work it bounds" + ); + } + + /// The must-not that keeps the early shed from becoming a key-farming axis. + /// + /// The early check is a READ. An author the limiter has never seen must not + /// gain a map entry from it, or a flood of self-minted signed DIDs could fill + /// the bounded author map without ever registering a peer, and legitimate new + /// authors would then be shed once it was full. + #[sqlx::test] + async fn the_early_author_check_does_not_track_an_unseen_did(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let stranger = Keypair::generate(); + // Deliberately NOT seeded: an unregistered DID must be refused by the peer + // lookup, and must leave no trace in the author limiter on the way. + let mut e = event_for(&stranger); + sign_ref_update(&stranger, &mut e).unwrap(); + + let before = limiters.author.tracked_keys().await; + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&e), &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "an unregistered DID is refused by the peer lookup, got {outcome:?}" + ); + assert_eq!( + limiters.author.tracked_keys().await, + before, + "the early check must not allocate a map entry for a DID it has never \ + seen, or the shed becomes its own memory-fill surface" + ); + } + + /// Exhaust one edge's unsigned budget and hand back the source that was + /// spent, so the callers below can assert what happens next. + async fn spend_unsigned_budget(db: &Db, limiters: &IngestLimiters, claim: &[u8]) -> PeerId { + let source = PeerId::random(); + for i in 0..GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS { + let outcome = ingest_ref_update(db, limiters, false, true, claim, &source).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "unsigned event {i} is inside the unsigned budget and must be admitted, got {outcome:?}" + ); + } + source + } + + /// Taking the author debit off unsigned traffic must not leave that traffic + /// unbounded. It is charged to the forwarder instead, which is the only + /// identity an unsigned event actually establishes. + /// + /// The last assertion is the half that keeps the new bucket honest: a spent + /// unsigned budget must not gate VERIFIED traffic down the same edge. That + /// is what fails if the charge is put in both signature arms rather than + /// only the unsigned one. + #[sqlx::test] + async fn the_unsigned_path_is_bounded_on_the_forwarder(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let author = Keypair::generate(); + seed_peer(&pool, &author.did().to_string()).await; + + let claim = bytes_of(&event_for(&author)); + let source = spend_unsigned_budget(&db, &limiters, &claim).await; + let spent = GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS as i64; + assert_eq!( + count(&pool, "received_ref_updates").await, + spent, + "every admitted unsigned event must land its row" + ); + assert_eq!( + count(&pool, "sync_queue").await, + spent, + "every admitted unsigned event must land its queue entry" + ); + + let outcome = ingest_ref_update(&db, &limiters, false, true, &claim, &source).await; + match &outcome { + IngestOutcome::UnsignedSourceRateLimited(named) => assert_eq!( + named, + &source.to_string(), + "the refusal must name the forwarder it was charged to" + ), + other => panic!("the event past the unsigned budget must be shed, got {other:?}"), + } + assert_eq!( + count(&pool, "received_ref_updates").await, + spent, + "a shed unsigned event must write no received_ref_updates row" + ); + assert_eq!( + count(&pool, "sync_queue").await, + spent, + "a shed unsigned event must enqueue no sync_queue row" + ); + + let mut signed = event_for(&author); + signed.ref_name = "refs/heads/signed".into(); + sign_ref_update(&author, &mut signed).unwrap(); + let outcome = + ingest_ref_update(&db, &limiters, false, true, &bytes_of(&signed), &source).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "the unsigned budget must not gate verified traffic down the same edge, got {outcome:?}" + ); + assert_eq!( + count(&pool, "received_ref_updates").await, + spent + 1, + "the signed event must land its own row" + ); + } + + /// The unsigned bucket keys on the FORWARDER, not on the DID the event + /// claims. Keying it on `node_did` would pass the victim must-not above + /// while rebuilding the same victim-selection hole one layer down: an + /// attacker would again spend a named victim's budget with events nobody + /// proved they authored. + #[sqlx::test] + async fn the_unsigned_budget_keys_on_the_forwarder_not_the_claimed_did(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let author = Keypair::generate(); + seed_peer(&pool, &author.did().to_string()).await; + + let claim = bytes_of(&event_for(&author)); + let spent_source = spend_unsigned_budget(&db, &limiters, &claim).await; + assert!(matches!( + ingest_ref_update(&db, &limiters, false, true, &claim, &spent_source).await, + IngestOutcome::UnsignedSourceRateLimited(_) + )); + + // Same bytes, same claimed DID, a different forwarder. + let fresh_source = PeerId::random(); + let outcome = ingest_ref_update(&db, &limiters, false, true, &claim, &fresh_source).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "the claimed DID must not carry a spent budget between forwarders, got {outcome:?}" + ); + assert_eq!( + count(&pool, "received_ref_updates").await, + GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS as i64 + 1, + "the event from the fresh forwarder must land its own row" + ); + } + + /// The honest-relay cost of keying on the forwarder, made explicit rather + /// than discovered later: once an edge has spent its unsigned budget, a + /// perfectly legitimate unsigned event from a DIFFERENT author arriving + /// down that edge is shed too. + /// + /// This is the tradeoff the 1500 cap is sized against, and it is why the + /// cap is three times the largest legitimate single-author burst rather + /// than equal to it. + #[sqlx::test] + async fn a_spent_unsigned_edge_sheds_a_second_unsigned_author_too(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let noisy = Keypair::generate(); + let bystander = Keypair::generate(); + seed_peer(&pool, &noisy.did().to_string()).await; + seed_peer(&pool, &bystander.did().to_string()).await; + + let claim = bytes_of(&event_for(&noisy)); + let source = spend_unsigned_budget(&db, &limiters, &claim).await; + + let mut other = event_for(&bystander); + other.repo = "zOwner/otherrepo".into(); + let outcome = + ingest_ref_update(&db, &limiters, false, true, &bytes_of(&other), &source).await; + match &outcome { + IngestOutcome::UnsignedSourceRateLimited(named) => assert_eq!( + named, + &source.to_string(), + "the shed must name the forwarder, not the bystanding author" + ), + other => panic!( + "a second unsigned author down a spent edge is shed by design, got {other:?}" + ), + } + assert_eq!( + count(&pool, "received_ref_updates").await, + GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS as i64, + "the shed event must write no row" + ); + } + + /// A structurally invalid event charges nobody AND costs nothing. + /// + /// Both halves are asserted, because they are different properties and only + /// the second is what the hoist actually buys. Charging nobody would hold + /// with the slug check merely above the debits; costing nothing needs it + /// above the `peer_exists` round trip and above the signature verify as + /// well. The `peer_exists` tally is what tells those two placements apart, + /// since both return the same `Rejected`. + /// + /// Run on both sides, because the two budgets are charged in different + /// branches: a malformed signed event must leave the author budget intact, + /// and a malformed unsigned event must leave the forwarder's unsigned + /// budget intact. + #[sqlx::test] + async fn a_malformed_slug_charges_nobody_and_costs_nothing(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let author = Keypair::generate(); + seed_peer(&pool, &author.did().to_string()).await; + + // ── Signed side: a full author budget of invalid events. ────────── + let mut bad = event_for(&author); + bad.repo = "../../x".into(); + sign_ref_update(&author, &mut bad).unwrap(); + let bad_signed = bytes_of(&bad); + let signed_edge = PeerId::random(); + reset_peer_exists_calls(); + for i in 0..GOSSIP_AUTHOR_MAX_EVENTS { + let outcome = + ingest_ref_update(&db, &limiters, false, true, &bad_signed, &signed_edge).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "malformed-slug event {i} must be rejected, got {outcome:?}" + ); + } + assert_nothing_written(&pool, "signed malformed-slug flood").await; + assert_eq!( + peer_exists_calls(), + 0, + "a malformed event must cost no peer lookup: the slug check has to run above the \ + round trip, not merely above the debit" + ); + + let mut good = event_for(&author); + sign_ref_update(&author, &mut good).unwrap(); + let outcome = + ingest_ref_update(&db, &limiters, false, true, &bytes_of(&good), &signed_edge).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "the author budget must be untouched by its own malformed events, got {outcome:?}" + ); + assert_eq!( + count(&pool, "received_ref_updates").await, + 1, + "the valid signed event must land its row" + ); + assert_eq!(count(&pool, "sync_queue").await, 1); + + // ── Unsigned side: a full unsigned budget of invalid events. ────── + let mut bad_unsigned = event_for(&author); + bad_unsigned.repo = "../../x".into(); + let bad_unsigned = bytes_of(&bad_unsigned); + let unsigned_edge = PeerId::random(); + reset_peer_exists_calls(); + for i in 0..GOSSIP_UNSIGNED_SOURCE_MAX_EVENTS { + let outcome = + ingest_ref_update(&db, &limiters, false, true, &bad_unsigned, &unsigned_edge).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "unsigned malformed-slug event {i} must be rejected, got {outcome:?}" + ); + } + assert_eq!( + peer_exists_calls(), + 0, + "a malformed unsigned event must cost no peer lookup either" + ); + + let valid_unsigned = bytes_of(&event_for(&author)); + let outcome = + ingest_ref_update(&db, &limiters, false, true, &valid_unsigned, &unsigned_edge).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "the forwarder's unsigned budget must be untouched by malformed events, got {outcome:?}" + ); + assert_eq!( + count(&pool, "received_ref_updates").await, + 2, + "the valid unsigned event must land its row" + ); + } + + /// FINDING 1: one push, many refs. `api::repos` publishes ONE gossip event + /// per updated ref, so a tag-heavy push, an initial import, or a mirror + /// backfill arrives as a burst of N events down a single mesh edge inside + /// one window. The HTTP twin batches the same push into a single + /// `/sync/notify`, so the brake is the only thing that makes the two + /// transports disagree about whether the push federated. + /// + /// 61 distinct refs is the smallest burst that exceeded the original + /// 60-per-source bound, and the tail was dropped with no back-pressure + /// signal to the publisher: a silent federation miss. Both budgets have to + /// clear it, and every event has to reach both sinks. + #[sqlx::test] + async fn a_sixty_one_ref_push_from_one_known_peer_is_accepted_whole(pool: PgPool) { + let db = ingest_db(&pool).await; + let limiters = IngestLimiters::new(); + let source = PeerId::random(); + let keypair = Keypair::generate(); + seed_peer(&pool, &keypair.did().to_string()).await; + + const REFS: usize = 61; + for i in 0..REFS { + let mut event = event_for(&keypair); + event.ref_name = format!("refs/tags/v{i}"); + sign_ref_update(&keypair, &mut event).unwrap(); + let outcome = + ingest_ref_update(&db, &limiters, true, true, &bytes_of(&event), &source).await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "ref {i} of a {REFS}-ref push must be accepted, got {outcome:?}" + ); + } + + assert_eq!( + count(&pool, "received_ref_updates").await, + REFS as i64, + "every ref in the push must land a received_ref_updates row" + ); + assert_eq!( + count(&pool, "sync_queue").await, + REFS as i64, + "every ref in the push must be enqueued for sync" + ); + } + + #[test] + fn an_unresolvable_did_key_is_rejected_with_the_shared_sentence() { + let keypair = Keypair::generate(); + let mut event = populated_event(); + // A did:key whose method id is not a decodable ed25519 multibase key. + event.node_did = "did:key:zNotARealKey".into(); + sign_ref_update(&keypair, &mut event).unwrap(); + + let err = verify_ref_update(&event).expect_err("an unresolvable did:key must be refused"); + assert!( + err.starts_with("cannot resolve DID 'did:key:zNotARealKey': "), + "expected the shared unresolvable-DID sentence, got: {err}" + ); + assert!( + !err.ends_with(": "), + "the sentence must carry the underlying reason, got: {err}" + ); + } + + #[derive(Clone, Default)] + struct CapturedWarnings(std::sync::Arc>>); + impl CapturedWarnings { + fn text(&self) -> String { + String::from_utf8_lossy(&self.0.lock().unwrap()).to_string() + } + fn saw_warn(&self) -> bool { + self.text().contains("accepted unsigned gossip ref-update") + } + } + impl std::io::Write for CapturedWarnings { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedWarnings { + type Writer = CapturedWarnings; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + fn capture_warnings() -> (CapturedWarnings, tracing::subscriber::DefaultGuard) { + let logs = CapturedWarnings::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(logs.clone()) + .with_max_level(tracing::Level::WARN) + .with_ansi(false) + .finish(); + let guard = tracing::subscriber::set_default(subscriber); + (logs, guard) + } + + #[sqlx::test] + async fn the_unsigned_warn_fires_only_on_admission(pool: PgPool) { + let db = ingest_db(&pool).await; + let author = Keypair::generate(); + let event = event_for(&author); + let claim = bytes_of(&event); + + // A: unsigned, peer NOT registered -> refused by peer_exists, no warn. + { + let (logs, _g) = capture_warnings(); + let outcome = + ingest_with_fresh_limiter(&db, false, true, &claim, &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "A outcome {outcome:?}" + ); + assert!(!logs.saw_warn(), "A must not warn, got: {}", logs.text()); + } + + seed_peer(&pool, &event.node_did).await; + + // B: unsigned, registered peer -> admitted, warn fires exactly once. + { + let (logs, _g) = capture_warnings(); + let outcome = + ingest_with_fresh_limiter(&db, false, true, &claim, &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedAdmitted), + "B outcome {outcome:?}" + ); + assert!(logs.saw_warn(), "B must warn, got: {}", logs.text()); + assert_eq!( + logs.text() + .matches("accepted unsigned gossip ref-update") + .count(), + 1, + "B must warn once, got: {}", + logs.text() + ); + } + + // C: unsigned budget spent -> shed on the forwarder, no warn. + { + let limiters = IngestLimiters::new(); + let source = spend_unsigned_budget(&db, &limiters, &claim).await; + let (logs, _g) = capture_warnings(); + let outcome = ingest_ref_update(&db, &limiters, false, true, &claim, &source).await; + assert!( + matches!(outcome, IngestOutcome::UnsignedSourceRateLimited(_)), + "C outcome {outcome:?}" + ); + assert!(!logs.saw_warn(), "C must not warn, got: {}", logs.text()); + } + + // D: unsigned with enforcement ON -> refused, no warn. + { + let (logs, _g) = capture_warnings(); + let outcome = + ingest_with_fresh_limiter(&db, true, true, &claim, &PeerId::random()).await; + assert!( + matches!(outcome, IngestOutcome::Rejected(_)), + "D outcome {outcome:?}" + ); + assert!(!logs.saw_warn(), "D must not warn, got: {}", logs.text()); + } + + // E: SIGNED and admitted -> the unsigned warn must not fire. + { + let signer = Keypair::generate(); + let mut signed = event_for(&signer); + sign_ref_update(&signer, &mut signed).unwrap(); + seed_peer(&pool, &signed.node_did).await; + let (logs, _g) = capture_warnings(); + let outcome = + ingest_with_fresh_limiter(&db, true, true, &bytes_of(&signed), &PeerId::random()) + .await; + assert!( + matches!(outcome, IngestOutcome::Accepted), + "E outcome {outcome:?}" + ); + assert!(!logs.saw_warn(), "E must not warn, got: {}", logs.text()); + } + } } diff --git a/crates/gitlawb-node/src/rate_limit.rs b/crates/gitlawb-node/src/rate_limit.rs index d203f69f..8c0f766b 100644 --- a/crates/gitlawb-node/src/rate_limit.rs +++ b/crates/gitlawb-node/src/rate_limit.rs @@ -121,6 +121,41 @@ impl RateLimiter { true } + /// Is this key ALREADY over its budget, answered without touching the map? + /// + /// Exists so a caller can shed an over-budget key before doing expensive work + /// on its behalf, which is the brake-placement rule: the brake belongs in + /// front of the work it bounds, not behind it. + /// + /// Read-only is the load-bearing part, not an optimisation. `check` inserts a + /// window for a key it has never seen, so using it as an early probe would let + /// a flood of unseen keys occupy the bounded map before any other gate had a + /// chance to refuse them. This allocates nothing and inserts nothing: an + /// untracked key is reported as within budget, and the real charge still + /// happens at the `check` call site further down. + /// + /// Expired timestamps are counted out rather than pruned, since pruning would + /// need the write lock this is deliberately avoiding. The count is therefore + /// exact for the decision it drives. + pub(crate) async fn is_over_budget(&self, key: &str) -> bool { + if self.max_requests == 0 { + return false; + } + let now = Instant::now(); + let state = self.state.lock().await; + match state.get(key) { + None => false, + Some(window) => { + let live = window + .timestamps + .iter() + .filter(|t| now.duration_since(**t) < self.window) + .count(); + live >= self.max_requests + } + } + } + /// Number of keys currently tracked. Tests use it to observe what a sweep /// reclaimed; there is no production reader. #[cfg(test)] @@ -128,6 +163,23 @@ impl RateLimiter { self.state.lock().await.len() } + /// The configured key ceiling. Tests use it to assert that a limiter a + /// production factory builds is actually bounded, which flooding the real + /// ceiling in distinct keys would be far too slow to show. + #[cfg(test)] + pub fn max_keys(&self) -> usize { + self.max_keys + } + + /// The configured per-key request cap. Tests use it to assert that a + /// limiter a production factory builds carries the cap it is documented + /// with; re-asserting the constant on its own cannot tell whether the + /// factory passed that constant or a different one. + #[cfg(test)] + pub fn max_requests(&self) -> usize { + self.max_requests + } + pub async fn cleanup(&self) { let now = Instant::now(); let mut state = self.state.lock().await;