fix(node): bind wire owner_did to local repo before surfacing as trusted

The peer-supplied owner_did (HTTP notify + gossipsub receive) and the
repo slug arrive over untrusted paths, yet REST and GraphQL projected the
wire value as trusted feed ownership.

- P1: add Db::resolve_ref_update_owner_did so a wire owner_did is echoed
  only when it matches the canonical owner of the actual local repo at that
  slug; otherwise it is dropped (returns null). The canonical DID is
  returned, never the raw wire bytes.
- P3: legacy owner_did = None rows are now attributed via an exact,
  normalized, unique local match (get_repo), not the prefix-tolerant drop
  gate, so a cross-method slug collision cannot synthesize the wrong owner.
- Unify REST (global + per-repo gossip) and GraphQL on the shared resolver
  so the two surfaces cannot drift. Add tests for forged-wire drop and
  exact-legacy attribution.

crates/gitlawb-node/src/api/events.rs

@@ -138,37 +138,29 @@ pub async fn list_ref_updates(
     let caller = auth.as_ref().map(|e| e.0 .0.as_str());
     let updates = collect_visible_ref_updates(&state.db, None, limit, caller).await?;
 
-    // Prefer the stored wire owner_did (full DID, may differ from local
-    // record's trailing-segment match). Fall back to the local record's
-    // owner_did only for backward-compat rows (stored as None) that name a
-    // repo this node hosts, so legacy rows still display an owner.
-    // The deduped set is loaded only when a legacy None row is actually
-    // present, avoiding the scan on every global-feed read.
-    let needs_fallback = updates.iter().any(|u| u.owner_did.is_none());
-    let deduped: Vec<crate::db::RepoRecord> = if needs_fallback {
-        state.db.list_all_repos_deduped().await?
-    } else {
-        Vec::new()
-    };
-    let resolve_local = |slug: &str| -> Option<&str> {
-        for record in &deduped {
-            if crate::visibility::ref_update_row_names_repo(record, slug) {
-                return Some(&record.owner_did);
-            }
-        }
-        None
-    };
+    // Resolve the trusted display owner_did per row. The stored wire value is
+    // untrusted (arrives over gossip / unsigned peer notify), so it is echoed
+    // only when it matches the canonical owner of the local repo the slug
+    // names (#P1); legacy None rows are attributed via an exact unique local
+    // match (#P3). Both surfaces share this resolver so they cannot drift.
+    let owner_dids: Vec<serde_json::Value> = futures::future::join_all(
+        updates
+            .iter()
+            .map(|u| state.db.resolve_ref_update_owner_did(&u.repo, u.owner_did.as_deref())),
+    )
+    .await
+    .into_iter()
+    .map(|r| {
+        r.map(|o| o.map_or(serde_json::Value::Null, |s| serde_json::json!(s)))
+            .unwrap_or(serde_json::Value::Null)
+    })
+    .collect();
 
     let events: Vec<serde_json::Value> = updates
         .iter()
-        .map(|u| {
-            let owner_did = match &u.owner_did {
-                Some(wire) => serde_json::json!(wire),
-                None => match resolve_local(&u.repo) {
-                    Some(local) => serde_json::json!(local),
-                    None => serde_json::Value::Null,
-                },
-            };
+        .enumerate()
+        .map(|(i, u)| {
+            let owner_did = owner_dids[i].clone();
             serde_json::json!({
                 "id":          u.id,
                 "node_did":    u.node_did,
@@ -270,29 +262,44 @@ pub async fn list_repo_events(
     // collect_visible_ref_updates drops any row whose slug matches a local repo the
     // caller cannot read (fail-closed), and propagates DB errors instead of swallowing
     // them.
-    let gossip_events: Vec<serde_json::Value> =
-        collect_visible_ref_updates(&state.db, Some(&repo_id_str), limit, caller)
-            .await?
+    let gossip_updates =
+        collect_visible_ref_updates(&state.db, Some(&repo_id_str), limit, caller).await?;
+    let gossip_owner_dids: Vec<serde_json::Value> = futures::future::join_all(
+        gossip_updates
             .iter()
-            .map(|u| {
-                serde_json::json!({
-                    "type":        "gossipsub",
-                    "id":          u.id,
-                    "repo":        u.repo,
-                    "ref_name":    u.ref_name,
-                    "old_sha":     u.old_sha,
-                    "new_sha":     u.new_sha,
-                    "pusher_did":  u.pusher_did,
-                    "node_did":    u.node_did,
-                    "timestamp":   u.timestamp,
-                    "cert_id":     u.cert_id,
-                    "received_at": u.received_at,
-                    "from_peer":   u.from_peer,
-                    "owner_did":   u.owner_did.as_ref().map(|s| serde_json::json!(s)).unwrap_or(serde_json::json!(record.owner_did)),
-                    "source":      "gossipsub",
-                })
+            .map(|u| state.db.resolve_ref_update_owner_did(&u.repo, u.owner_did.as_deref())),
+    )
+    .await
+    .into_iter()
+    .map(|r| {
+        r.map(|o| o.map_or(serde_json::Value::Null, |s| serde_json::json!(s)))
+            .unwrap_or(serde_json::Value::Null)
+    })
+    .collect();
+
+    let gossip_events: Vec<serde_json::Value> = gossip_updates
+        .iter()
+        .enumerate()
+        .map(|(i, u)| {
+            let owner_did = gossip_owner_dids[i].clone();
+            serde_json::json!({
+                "type":        "gossipsub",
+                "id":          u.id,
+                "repo":        u.repo,
+                "ref_name":    u.ref_name,
+                "old_sha":     u.old_sha,
+                "new_sha":     u.new_sha,
+                "pusher_did":  u.pusher_did,
+                "node_did":    u.node_did,
+                "timestamp":   u.timestamp,
+                "cert_id":     u.cert_id,
+                "received_at": u.received_at,
+                "from_peer":   u.from_peer,
+                "owner_did":   owner_did,
+                "source":      "gossipsub",
             })
-            .collect();
+        })
+        .collect();
 
     // Merge both lists
     let mut all_events: Vec<serde_json::Value> = cert_events;

crates/gitlawb-node/src/db/mod.rs

@@ -2336,6 +2336,56 @@ impl Db {
         Ok(())
     }
 
+    /// Resolve the trusted display `owner_did` for a ref-update row, guarding
+    /// against a forged peer-supplied wire value.
+    ///
+    /// The stored `owner_did` (and the `repo` slug) arrive over gossipsub or the
+    /// unsigned peer-notify HTTP path, so neither is trusted. This method binds
+    /// the wire `owner_did` to the local repo the slug names before it is ever
+    /// surfaced:
+    ///
+    /// * **P1 (untrusted wire value):** a peer-supplied `owner_did` is only
+    ///   echoed when it normalizes equal to the canonical owner of the *actual
+    ///   local repo* at that slug. A caller asserting `did:key:zVictim` on a
+    ///   `zAlice/widget` row is dropped, because `zVictim` does not own the
+    ///   local `zAlice/widget` repo. The canonical DID is returned (not the raw
+    ///   wire bytes), so the projection is always the local source of truth.
+    /// * **P3 (legacy fallback):** a row stored with `owner_did = None` is
+    ///   attributed only via an *exact, normalized, unique* local match —
+    ///   `get_repo` matches the slug's owner key and name exactly (`LIMIT 1`,
+    ///   preferring canonical rows). The loose prefix-tolerant drop gate
+    ///   (`ref_update_row_names_repo`) is never used for attribution, so a
+    ///   cross-method slug collision cannot synthesize the wrong owner. When no
+    ///   unique local repo proves the owner, `None` is returned.
+    ///
+    /// The slug must be `"{owner}/{name}"`; a slug without a `/` cannot be
+    /// attributed and yields `None`.
+    pub async fn resolve_ref_update_owner_did(
+        &self,
+        slug: &str,
+        wire_owner_did: Option<&str>,
+    ) -> Result<Option<String>> {
+        let Some((owner_part, name)) = slug.rsplit_once('/') else {
+            return Ok(None);
+        };
+
+        let repo = self.get_repo(owner_part, name).await?;
+
+        match (wire_owner_did, repo) {
+            // P1: trusted only when the wire DID matches the canonical owner of
+            // the local repo the slug names.
+            (Some(wire), Some(repo))
+                if normalize_owner_key(wire) == normalize_owner_key(&repo.owner_did) =>
+            {
+                Ok(Some(repo.owner_did))
+            }
+            // P3: legacy None row, attributed via an exact unique local match.
+            (None, Some(repo)) => Ok(Some(repo.owner_did)),
+            // Anything else (mismatch, or no proven local owner) is untrusted.
+            _ => Ok(None),
+        }
+    }
+
     /// One page of ref updates (newest first), optionally scoped to one repo.
     /// The `(timestamp DESC, id DESC)` order gives a stable tiebreak so offset
     /// paging does not skip or duplicate rows when timestamps collide. Used by

crates/gitlawb-node/src/graphql/query.rs

@@ -73,44 +73,28 @@ impl QueryRoot {
                 .await
                 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
 
-        // Match the REST feed's owner_did projection: prefer the stored wire
-        // value (full DID), fall back to the local record's owner for
-        // backward-compat rows (stored as None) that name a local repo.
-        // The deduped set is loaded only when a legacy None row actually
-        // needs fallback, avoiding the scan on every query.
-        let needs_fallback = updates.iter().any(|u| u.owner_did.is_none());
-        let deduped: Vec<crate::db::RepoRecord> = if needs_fallback {
-            db.list_all_repos_deduped()
+        // Resolve the trusted display owner_did per row, identical to the REST
+        // feed: the stored wire value is untrusted, so it is echoed only when it
+        // matches the canonical owner of the local repo the slug names (#P1);
+        // legacy None rows are attributed via an exact unique local match (#P3).
+        let mut resolved = Vec::with_capacity(updates.len());
+        for u in updates {
+            let owner_did = db
+                .resolve_ref_update_owner_did(&u.repo, u.owner_did.as_deref())
                 .await
-                .map_err(|e| async_graphql::Error::new(e.to_string()))?
-        } else {
-            Vec::new()
-        };
-        let resolve_local = |slug: &str| -> Option<String> {
-            for record in &deduped {
-                if crate::visibility::ref_update_row_names_repo(record, slug) {
-                    return Some(record.owner_did.clone());
-                }
-            }
-            None
-        };
-
-        Ok(updates
-            .into_iter()
-            .map(|u| {
-                let owner_did = u.owner_did.or_else(|| resolve_local(&u.repo));
-                RefUpdateType {
-                    repo: u.repo,
-                    ref_name: u.ref_name,
-                    old_sha: u.old_sha,
-                    new_sha: u.new_sha,
-                    pusher_did: u.pusher_did,
-                    node_did: u.node_did,
-                    timestamp: u.timestamp,
-                    owner_did,
-                }
-            })
-            .collect())
+                .map_err(|e| async_graphql::Error::new(e.to_string()))?;
+            resolved.push(RefUpdateType {
+                repo: u.repo,
+                ref_name: u.ref_name,
+                old_sha: u.old_sha,
+                new_sha: u.new_sha,
+                pusher_did: u.pusher_did,
+                node_did: u.node_did,
+                timestamp: u.timestamp,
+                owner_did,
+            });
+        }
+        Ok(resolved)
     }
 
     async fn tasks(

crates/gitlawb-node/src/test_support.rs

@@ -3395,6 +3395,15 @@ mod tests {
         let state = test_state(pool).await;
         let owner = "did:key:zEVENTSOWNERAAAAAAAAAAAAAAAAAAAAAAAAA";
 
+        // Seed a local repo the wire owner_did is bound to. The stored wire
+        // owner_did is untrusted; it is only surfaced when it matches the
+        // canonical owner of the local repo the slug names.
+        state
+            .db
+            .create_repo(&seed_repo(owner, "myrepo"))
+            .await
+            .unwrap();
+
         // Insert a gossip event with owner_did set
         state
             .db
@@ -3431,6 +3440,111 @@ mod tests {
         assert_eq!(events[0]["owner_did"], owner);
     }
 
+    // P1: a peer-supplied owner_did that does NOT match the canonical owner of
+    // the local repo the slug names must NOT be surfaced. Here zVictim asserts
+    // ownership of alice's widget repo; the projection must drop it to null
+    // rather than poisoning persisted event ownership.
+    #[sqlx::test]
+    async fn events_drop_forged_peer_owner_did(pool: PgPool) {
+        let state = test_state(pool).await;
+        let alice = "did:key:zALICEOWNERAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let victim = "did:key:zVICTIMOWNERAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+
+        state
+            .db
+            .create_repo(&seed_repo(alice, "widget"))
+            .await
+            .unwrap();
+
+        state
+            .db
+            .insert_ref_update(&crate::db::ReceivedRefUpdate {
+                id: uuid::Uuid::new_v4().to_string(),
+                node_did: "did:key:zNode".into(),
+                pusher_did: "did:key:zPusher".into(),
+                repo: format!("{}/widget", alice.split(':').next_back().unwrap()),
+                owner_did: Some(victim.into()),
+                ref_name: "refs/heads/main".into(),
+                old_sha: "0000000000000000000000000000000000000000".into(),
+                new_sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(),
+                timestamp: "2026-07-02T12:00:00Z".into(),
+                cert_id: None,
+                received_at: "2026-07-02T12:00:01Z".into(),
+                from_peer: "12D3KooWTest".into(),
+            })
+            .await
+            .unwrap();
+
+        let resp = events_router(state)
+            .oneshot(anon_get("/api/v1/events/ref-updates"))
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+        let body = json_body(resp).await;
+        let events = body["events"].as_array().unwrap();
+        assert_eq!(events.len(), 1);
+        // Forged value dropped: owner_did is null, NOT zVictim.
+        assert_eq!(events[0]["owner_did"], serde_json::Value::Null);
+    }
+
+    // P3: a legacy row stored with owner_did = None must be attributed only via
+    // an exact, unique local match — never a loose prefix-tolerant collision.
+    // alice/widget is owned by alice; a stray None row on a different repo whose
+    // owner key shares a segment must not inherit alice's full DID.
+    #[sqlx::test]
+    async fn events_legacy_none_owner_uses_exact_local_match(pool: PgPool) {
+        let state = test_state(pool).await;
+        let alice = "did:key:zALICEOWNER2AAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+        let bob = "did:key:zBOBOWNER2AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
+
+        // Alice owns "widget".
+        state
+            .db
+            .create_repo(&seed_repo(alice, "widget"))
+            .await
+            .unwrap();
+        // Bob owns a distinct "gadget" repo.
+        state
+            .db
+            .create_repo(&seed_repo(bob, "gadget"))
+            .await
+            .unwrap();
+
+        // Legacy None row claiming slug "bob/gadget" (matches bob exactly).
+        state
+            .db
+            .insert_ref_update(&crate::db::ReceivedRefUpdate {
+                id: uuid::Uuid::new_v4().to_string(),
+                node_did: "did:key:zNode".into(),
+                pusher_did: "did:key:zPusher".into(),
+                repo: format!("{}/gadget", bob.split(':').next_back().unwrap()),
+                owner_did: None,
+                ref_name: "refs/heads/main".into(),
+                old_sha: "0000000000000000000000000000000000000000".into(),
+                new_sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into(),
+                timestamp: "2026-07-02T12:00:00Z".into(),
+                cert_id: None,
+                received_at: "2026-07-02T12:00:01Z".into(),
+                from_peer: "12D3KooWTest".into(),
+            })
+            .await
+            .unwrap();
+
+        let resp = events_router(state)
+            .oneshot(anon_get("/api/v1/events/ref-updates"))
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::OK);
+        let body = json_body(resp).await;
+        let events = body["events"].as_array().unwrap();
+        assert_eq!(events.len(), 1);
+        // Attributed to the exact local owner (bob), not alice via collision.
+        assert_eq!(
+            events[0]["owner_did"],
+            serde_json::Value::String(bob.to_string())
+        );
+    }
+
     #[sqlx::test]
     async fn list_all_bounties_past_private_window_finds_public(pool: PgPool) {
         let state = test_state(pool).await;