diff --git a/Cargo.lock b/Cargo.lock index d12daa43..24beb9ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3300,7 +3300,7 @@ dependencies = [ [[package]] name = "git-remote-gitlawb" -version = "0.5.0" +version = "0.5.1" dependencies = [ "anyhow", "gitlawb-core", @@ -3312,7 +3312,7 @@ dependencies = [ [[package]] name = "gitlawb-attest" -version = "0.5.0" +version = "0.5.1" dependencies = [ "base64", "ed25519-dalek", @@ -3329,7 +3329,7 @@ dependencies = [ [[package]] name = "gitlawb-core" -version = "0.5.0" +version = "0.5.1" dependencies = [ "anyhow", "base64", @@ -3356,7 +3356,7 @@ dependencies = [ [[package]] name = "gitlawb-node" -version = "0.5.0" +version = "0.5.1" dependencies = [ "alloy", "anyhow", @@ -3412,7 +3412,7 @@ dependencies = [ [[package]] name = "gl" -version = "0.5.0" +version = "0.5.1" dependencies = [ "alloy", "anyhow", @@ -3824,6 +3824,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "thiserror 2.0.18", "tracing", ] diff --git a/crates/gitlawb-node/src/api/events.rs b/crates/gitlawb-node/src/api/events.rs index 99cc2eeb..af790fc4 100644 --- a/crates/gitlawb-node/src/api/events.rs +++ b/crates/gitlawb-node/src/api/events.rs @@ -138,9 +138,37 @@ 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 = 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 + }; + let events: Vec = 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, + }, + }; serde_json::json!({ "id": u.id, "node_did": u.node_did, @@ -153,6 +181,7 @@ pub async fn list_ref_updates( "cert_id": u.cert_id, "received_at": u.received_at, "from_peer": u.from_peer, + "owner_did": owner_did, }) }) .collect(); @@ -225,6 +254,7 @@ pub async fn list_repo_events( "pusher_did": c.pusher_did, "node_did": c.node_did, "timestamp": c.issued_at, + "owner_did": record.owner_did, "source": "local", }) }) @@ -258,6 +288,7 @@ pub async fn list_repo_events( "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", }) }) @@ -326,6 +357,7 @@ mod ref_updates_feed_tests { cert_id: None, received_at: Utc::now().to_rfc3339(), from_peer: "peer1".into(), + owner_did: None, } } @@ -1105,7 +1137,14 @@ mod ref_updates_feed_tests { .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); - assert_eq!(count(&body_json(resp).await), 2); + let body = body_json(resp).await; + assert_eq!(count(&body), 2); + for event in body["events"].as_array().unwrap() { + assert_eq!( + event["owner_did"], OWNER, + "each event must carry the local owner_did" + ); + } } // Anon reads a quarantined mirror → 404 (withheld without disclosing existence diff --git a/crates/gitlawb-node/src/api/peers.rs b/crates/gitlawb-node/src/api/peers.rs index a125c313..5c535f0c 100644 --- a/crates/gitlawb-node/src/api/peers.rs +++ b/crates/gitlawb-node/src/api/peers.rs @@ -347,6 +347,10 @@ pub struct NotifyRequest { pub timestamp: Option, #[serde(default)] pub cert_id: Option, + /// Full owner DID — added in #144 for DID-aware feed gating. + /// Optional for backward compat with older senders. + #[serde(default)] + pub owner_did: Option, } pub async fn notify_sync( @@ -391,6 +395,7 @@ pub async fn notify_sync( node_did: req.node_did.clone(), pusher_did: req.pusher_did.clone().unwrap_or_default(), repo: req.repo.clone(), + owner_did: req.owner_did.clone(), ref_name: req.ref_name.clone(), old_sha: req.old_sha.clone().unwrap_or_default(), new_sha: req.new_sha.clone(), diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b9cbc352..b38b177b 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -772,6 +772,7 @@ async fn notify_peer_of_ref( new_sha: &str, node_did: &str, pusher_did: &str, + owner_did: &str, ) { let body = serde_json::json!({ "repo": repo_slug, @@ -781,6 +782,7 @@ async fn notify_peer_of_ref( "pusher_did": pusher_did, "old_sha": old_sha, "timestamp": chrono::Utc::now().to_rfc3339(), + "owner_did": owner_did, }); let body_bytes = match serde_json::to_vec(&body) { Ok(bytes) => bytes, @@ -828,6 +830,7 @@ async fn notify_peer_of_refs( ref_updates: &[(String, String, String)], node_did: &str, pusher_did: &str, + owner_did: &str, ) { for (ref_name, old_sha, new_sha) in ref_updates { notify_peer_of_ref( @@ -841,6 +844,7 @@ async fn notify_peer_of_refs( new_sha, node_did, pusher_did, + owner_did, ) .await; } @@ -1261,6 +1265,7 @@ pub async fn git_receive_pack( node_did: node_did_str.clone(), pusher_did: pusher_did_clone.clone(), repo: repo_slug.clone(), + owner_did: Some(record.owner_did.clone()), ref_name: ref_name.clone(), old_sha: old_sha.clone(), new_sha: new_sha.clone(), @@ -1292,6 +1297,7 @@ pub async fn git_receive_pack( pusher_did: pusher_did_clone.clone(), node_did: node_did_str.clone(), timestamp: now_ts.clone(), + owner_did: record.owner_did.clone(), }); } } @@ -1362,6 +1368,7 @@ pub async fn git_receive_pack( &ref_updates_clone, &node_did_str, &pusher_did_clone, + &record.owner_did, ) .await; } @@ -2402,6 +2409,9 @@ mod tests { mockito::Matcher::PartialJsonString(format!(r#"{{"ref_name":"{ref_a}"}}"#)), mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{old_a}"}}"#)), mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_a}"}}"#)), + mockito::Matcher::PartialJsonString( + r#"{"owner_did":"did:key:zOwner"}"#.to_string(), + ), ])) .with_status(200) .expect(1) @@ -2413,6 +2423,9 @@ mod tests { mockito::Matcher::PartialJsonString(format!(r#"{{"ref_name":"{ref_b}"}}"#)), mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{old_b}"}}"#)), mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_b}"}}"#)), + mockito::Matcher::PartialJsonString( + r#"{"owner_did":"did:key:zOwner"}"#.to_string(), + ), ])) .with_status(200) .expect(1) @@ -2434,6 +2447,7 @@ mod tests { &ref_updates, "did:key:zNode", "did:key:zPusher", + "did:key:zOwner", ) .await; @@ -2456,6 +2470,9 @@ mod tests { .match_body(mockito::Matcher::AllOf(vec![ mockito::Matcher::PartialJsonString(format!(r#"{{"old_sha":"{zero}"}}"#)), mockito::Matcher::PartialJsonString(format!(r#"{{"new_sha":"{new_sha}"}}"#)), + mockito::Matcher::PartialJsonString( + r#"{"owner_did":"did:key:zOwner"}"#.to_string(), + ), ])) .with_status(200) .expect(1) @@ -2478,6 +2495,7 @@ mod tests { &ref_updates, "did:key:zNode", "did:key:zPusher", + "did:key:zOwner", ) .await; diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 5324a4b9..aa9bcf0d 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -175,6 +175,9 @@ pub struct ReceivedRefUpdate { pub cert_id: Option, pub received_at: String, pub from_peer: String, + /// Full owner DID — populated by new peers; None for events from older + /// peers that predate the wire-format change (#144). + pub owner_did: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -872,6 +875,14 @@ const MIGRATIONS: &[Migration] = &[ "CREATE UNIQUE INDEX IF NOT EXISTS idx_ref_certs_repo_ref ON ref_certificates(repo_id, ref_name)", ], }, + Migration { + version: 11, + name: "ref_update_owner_did", + stmts: &[ + // Index deferred — the feed gate (#144) does not read owner_did yet. + "ALTER TABLE received_ref_updates ADD COLUMN IF NOT EXISTS owner_did TEXT", + ], + }, ]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -2304,8 +2315,8 @@ impl Db { sqlx::query( "INSERT INTO received_ref_updates (id, node_did, pusher_did, repo, ref_name, old_sha, new_sha, timestamp, - cert_id, received_at, from_peer) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) + cert_id, received_at, from_peer, owner_did) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT(id) DO NOTHING", ) .bind(&update.id) @@ -2319,6 +2330,7 @@ impl Db { .bind(&update.cert_id) .bind(&update.received_at) .bind(&update.from_peer) + .bind(&update.owner_did) .execute(&self.pool) .await?; Ok(()) @@ -2348,7 +2360,7 @@ impl Db { after: Option<(&str, &str)>, ) -> Result> { const COLS: &str = "id, node_did, pusher_did, repo, ref_name, old_sha, new_sha, \ - timestamp, cert_id, received_at, from_peer"; + timestamp, cert_id, received_at, from_peer, owner_did"; // Positional params in bind order: repo?, after_ts?, after_id?, limit. let mut conds: Vec = Vec::new(); @@ -2686,6 +2698,7 @@ fn row_to_ref_update(r: sqlx::postgres::PgRow) -> ReceivedRefUpdate { cert_id: r.get("cert_id"), received_at: r.get("received_at"), from_peer: r.get("from_peer"), + owner_did: r.get("owner_did"), } } @@ -3406,6 +3419,115 @@ mod migration_tests { // it, you must also update the backfill. assert_eq!(MIGRATIONS[0].name, MIGRATION_V1_NAME); } + + /// Simulate an existing node at v9 with populated received_ref_updates, + /// then apply the v11 migration and verify (a) owner_did IS NULL on + /// existing rows, (b) the column exists and is nullable TEXT, and + /// (c) idempotent re-run does not error. + #[sqlx::test] + async fn migration_v11_creates_owner_did_column(pool: sqlx::PgPool) { + let db = super::Db::for_testing(pool); + + // Create all tables by running the full migration chain from scratch, + // then drop the owner_did column to simulate a pre-v10 schema. + db.migrate().await.unwrap(); + sqlx::query("ALTER TABLE received_ref_updates DROP COLUMN owner_did") + .execute(&db.pool) + .await + .unwrap(); + + // Truncate schema_migrations and re-seed at v9 — simulate an existing + // node that has run v1..v9 but not yet v10. + sqlx::query("DELETE FROM schema_migrations") + .execute(&db.pool) + .await + .unwrap(); + for m in MIGRATIONS.iter().take_while(|m| m.version < 10) { + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) + VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind("2026-07-01T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + } + + // ── Simulate an existing node with rows recorded before v11 ──────── + // The owner_did column does not exist yet, so we INSERT without it. + let row_id = uuid::Uuid::new_v4().to_string(); + sqlx::query( + "INSERT INTO received_ref_updates + (id, node_did, pusher_did, repo, ref_name, old_sha, new_sha, + timestamp, cert_id, received_at, from_peer) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)", + ) + .bind(&row_id) + .bind("did:key:zNode") + .bind("did:key:zPusher") + .bind("z6MkOwner/myrepo") + .bind("refs/heads/main") + .bind("0000000000000000000000000000000000000000") + .bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + .bind("2026-07-01T12:00:00Z") + .bind::>(None) + .bind("2026-07-01T12:00:01Z") + .bind("12D3KooWPeer") + .execute(&db.pool) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM received_ref_updates") + .fetch_one(&db.pool) + .await + .unwrap(), + 1, + "pre-migration row must exist" + ); + + // ── Apply pending migrations (v10 ref_cert_unique_per_ref, v11 owner_did) ── + db.migrate().await.unwrap(); + + // ── Assertions ──────────────────────────────────────────────────── + + // (a) Existing row has owner_did IS NULL (not overwritten). + let owner: Option = + sqlx::query_scalar("SELECT owner_did FROM received_ref_updates WHERE id = $1") + .bind(&row_id) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(owner, None, "existing row's owner_did must be NULL"); + + // (b) Column exists and is nullable TEXT. + let col: (String, String, String) = sqlx::query_as( + "SELECT column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_name = 'received_ref_updates' AND column_name = 'owner_did'", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(col.0, "owner_did"); + assert_eq!(col.1, "text"); + assert_eq!(col.2, "YES", "owner_did must be nullable"); + + // (c) Version 11 is recorded as applied. + let v11_count: (i64,) = + sqlx::query_as("SELECT COUNT(*) FROM schema_migrations WHERE version = 11") + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!( + v11_count.0, 1, + "migration v11 must be recorded in schema_migrations" + ); + + // (d) Re-run: idempotent — ADD COLUMN IF NOT EXISTS must not error. + db.migrate().await.unwrap(); + } } #[cfg(test)] @@ -4454,6 +4576,7 @@ mod ref_update_keyset_paging_tests { cert_id: None, received_at: ts.into(), from_peer: "peer".into(), + owner_did: None, } } @@ -4562,6 +4685,7 @@ mod ref_update_keyset_repo_filtered_tests { cert_id: None, received_at: ts.into(), from_peer: "peer".into(), + owner_did: None, } } @@ -4668,6 +4792,7 @@ mod ref_update_keyset_same_timestamp_tests { cert_id: None, received_at: ts.into(), from_peer: "peer".into(), + owner_did: None, } } @@ -5263,3 +5388,170 @@ mod ref_certificate_tests { ); } } +#[cfg(test)] +mod ref_update_db_tests { + use super::{Db, ReceivedRefUpdate}; + use sqlx::PgPool; + + async fn db(pool: PgPool) -> Db { + let db = Db::for_testing(pool); + db.run_migrations().await.unwrap(); + db + } + + fn update( + id: &str, + repo: &str, + owner_did: Option<&str>, + ref_name: &str, + sha: &str, + ) -> ReceivedRefUpdate { + ReceivedRefUpdate { + id: id.to_string(), + node_did: "did:key:zNode".into(), + pusher_did: "did:key:zPusher".into(), + repo: repo.to_string(), + owner_did: owner_did.map(|s| s.to_string()), + ref_name: ref_name.to_string(), + old_sha: "0000000000000000000000000000000000000000".into(), + new_sha: sha.to_string(), + timestamp: "2026-07-02T12:00:00Z".into(), + cert_id: None, + received_at: "2026-07-02T12:00:01Z".into(), + from_peer: "12D3KooWTest".into(), + } + } + + #[sqlx::test] + async fn insert_and_list_with_owner_did(pool: PgPool) { + let db = db(pool).await; + db.insert_ref_update(&update( + "u1", + "zOwner/myrepo", + Some("did:key:zOwner"), + "refs/heads/main", + "aaaa", + )) + .await + .unwrap(); + + let all = db.list_ref_updates_keyset(None, 100, None).await.unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].owner_did.as_deref(), Some("did:key:zOwner")); + assert_eq!(all[0].repo, "zOwner/myrepo"); + } + + #[sqlx::test] + async fn insert_and_list_without_owner_did(pool: PgPool) { + let db = db(pool).await; + db.insert_ref_update(&update( + "u2", + "zOwner/myrepo", + None, + "refs/heads/main", + "bbbb", + )) + .await + .unwrap(); + + let all = db.list_ref_updates_keyset(None, 100, None).await.unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].owner_did, None); + } + + #[sqlx::test] + async fn list_repo_ref_updates_filters_by_repo(pool: PgPool) { + let db = db(pool).await; + db.insert_ref_update(&update( + "u3", + "alice/repo1", + Some("did:key:zAlice"), + "refs/heads/main", + "cccc", + )) + .await + .unwrap(); + db.insert_ref_update(&update( + "u4", + "bob/repo2", + Some("did:key:zBob"), + "refs/heads/feat", + "dddd", + )) + .await + .unwrap(); + + let alice_events = db + .list_ref_updates_keyset(Some("alice/repo1"), 100, None) + .await + .unwrap(); + assert_eq!(alice_events.len(), 1); + assert_eq!(alice_events[0].id, "u3"); + assert_eq!(alice_events[0].owner_did.as_deref(), Some("did:key:zAlice")); + + let bob_events = db + .list_ref_updates_keyset(Some("bob/repo2"), 100, None) + .await + .unwrap(); + assert_eq!(bob_events.len(), 1); + assert_eq!(bob_events[0].id, "u4"); + assert_eq!(bob_events[0].owner_did.as_deref(), Some("did:key:zBob")); + + let empty = db + .list_ref_updates_keyset(Some("other/repo"), 100, None) + .await + .unwrap(); + assert!(empty.is_empty()); + } + + #[sqlx::test] + async fn list_ref_updates_filtered_by_repo(pool: PgPool) { + let db = db(pool).await; + db.insert_ref_update(&update( + "u5", + "ownerA/proj", + Some("did:key:zA"), + "refs/heads/main", + "eeee", + )) + .await + .unwrap(); + db.insert_ref_update(&update( + "u6", + "ownerB/proj", + Some("did:web:host:zB"), + "refs/heads/main", + "ffff", + )) + .await + .unwrap(); + + let filtered = db + .list_ref_updates_keyset(Some("ownerA/proj"), 100, None) + .await + .unwrap(); + assert_eq!(filtered.len(), 1); + assert_eq!(filtered[0].id, "u5"); + + let all = db.list_ref_updates_keyset(None, 100, None).await.unwrap(); + assert_eq!(all.len(), 2); + } + + #[sqlx::test] + async fn insert_update_idempotent_on_conflict(pool: PgPool) { + let db = db(pool).await; + let u = update( + "u7", + "repo/x", + Some("did:key:zX"), + "refs/heads/main", + "gggg", + ); + db.insert_ref_update(&u).await.unwrap(); + db.insert_ref_update(&u).await.unwrap(); + + let all = db.list_ref_updates_keyset(None, 100, None).await.unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].new_sha, "gggg"); + } +} diff --git a/crates/gitlawb-node/src/graphql/query.rs b/crates/gitlawb-node/src/graphql/query.rs index 55148e02..cf779e83 100644 --- a/crates/gitlawb-node/src/graphql/query.rs +++ b/crates/gitlawb-node/src/graphql/query.rs @@ -73,16 +73,42 @@ 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 = if needs_fallback { + db.list_all_repos_deduped() + .await + .map_err(|e| async_graphql::Error::new(e.to_string()))? + } else { + Vec::new() + }; + let resolve_local = |slug: &str| -> Option { + 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| 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, + .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()) } @@ -162,6 +188,7 @@ mod tests { cert_id: None, received_at: Utc::now().to_rfc3339(), from_peer: "peer1".into(), + owner_did: None, } } @@ -245,10 +272,10 @@ mod tests { .await .unwrap(); let schema = schema(db); - let q = r#"{ refUpdates { repo refName } }"#; + let q = r#"{ refUpdates { repo refName ownerDid } }"#; let resp = anon(&schema, q).await; assert_eq!(count(&resp), 1); - // The one row returned must be the public repo's. + // The one row returned must be the public repo's with owner_did echoed. let async_graphql::Value::Object(obj) = &resp.data else { unreachable!() }; @@ -262,6 +289,11 @@ mod tests { row.get("repo").unwrap(), &async_graphql::Value::from("z6MkOwner/openrepo") ); + assert_eq!( + row.get("ownerDid").unwrap(), + &async_graphql::Value::from(OWNER), + "ownerDid must fall back to the local record's owner for legacy rows" + ); } // Scenario 4 — alias fail-closed: private repo's row stored full-DID form. diff --git a/crates/gitlawb-node/src/graphql/subscription.rs b/crates/gitlawb-node/src/graphql/subscription.rs index 6c639bc6..7248cbf4 100644 --- a/crates/gitlawb-node/src/graphql/subscription.rs +++ b/crates/gitlawb-node/src/graphql/subscription.rs @@ -39,6 +39,7 @@ impl SubscriptionRoot { pusher_did: ev.pusher_did, node_did: ev.node_did, timestamp: ev.timestamp, + owner_did: Some(ev.owner_did), }), _ => None, } diff --git a/crates/gitlawb-node/src/graphql/types.rs b/crates/gitlawb-node/src/graphql/types.rs index 918701fd..4264a581 100644 --- a/crates/gitlawb-node/src/graphql/types.rs +++ b/crates/gitlawb-node/src/graphql/types.rs @@ -57,6 +57,7 @@ pub struct RefUpdateType { pub pusher_did: String, pub node_did: String, pub timestamp: String, + pub owner_did: Option, } #[derive(SimpleObject, Clone)] diff --git a/crates/gitlawb-node/src/p2p/mod.rs b/crates/gitlawb-node/src/p2p/mod.rs index 5a6992b1..80e28a4a 100644 --- a/crates/gitlawb-node/src/p2p/mod.rs +++ b/crates/gitlawb-node/src/p2p/mod.rs @@ -39,6 +39,11 @@ pub struct RefUpdateEvent { pub pusher_did: String, /// Repository identifier (owner/name) pub repo: String, + /// Full owner DID — added in #144 for display and storage; not yet + /// wired into the feed gate matcher. Optional for backward compat with + /// older peers that don't include it. + #[serde(default)] + pub owner_did: Option, /// Git ref that changed (e.g., "refs/heads/main") pub ref_name: String, /// SHA before the push (all-zeros for new ref) @@ -307,6 +312,7 @@ pub async fn start( 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(), @@ -432,3 +438,67 @@ pub async fn start( Ok(handle) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ref_update_event_round_trip_with_owner_did() { + let event = RefUpdateEvent { + 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: None, + cid: None, + }; + let json = serde_json::to_value(&event).unwrap(); + // owner_did must be present in the serialized output + assert_eq!(json["owner_did"], "did:key:zOwner"); + assert_eq!(json["repo"], "zOwner/myrepo"); + + let deserialized: RefUpdateEvent = serde_json::from_value(json).unwrap(); + assert_eq!(deserialized.owner_did, Some("did:key:zOwner".into())); + } + + #[test] + fn ref_update_event_backward_compat_no_owner_did() { + 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.owner_did, None); + assert_eq!(deserialized.repo, "zOwner/myrepo"); + } + + #[test] + fn ref_update_event_backward_compat_null_owner_did() { + let with_null = serde_json::json!({ + "node_did": "did:key:zNode", + "pusher_did": "did:key:zPusher", + "repo": "zOwner/myrepo", + "owner_did": null, + "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(with_null).unwrap(); + assert_eq!(deserialized.owner_did, None); + } +} diff --git a/crates/gitlawb-node/src/state.rs b/crates/gitlawb-node/src/state.rs index 6a84b3c2..d3e53f3a 100644 --- a/crates/gitlawb-node/src/state.rs +++ b/crates/gitlawb-node/src/state.rs @@ -17,6 +17,7 @@ pub struct RefUpdateBroadcast { pub pusher_did: String, pub node_did: String, pub timestamp: String, + pub owner_did: String, } #[derive(Clone, Debug)] diff --git a/crates/gitlawb-node/src/test_support.rs b/crates/gitlawb-node/src/test_support.rs index 4fe52f64..175879c1 100644 --- a/crates/gitlawb-node/src/test_support.rs +++ b/crates/gitlawb-node/src/test_support.rs @@ -3379,6 +3379,58 @@ mod tests { ); } + // ── Ref-update events (issue #144: owner_did wire format) ───────────────── + + fn events_router(state: AppState) -> Router { + Router::new() + .route( + "/api/v1/events/ref-updates", + axum::routing::get(crate::api::events::list_ref_updates), + ) + .with_state(state) + } + + #[sqlx::test] + async fn events_returns_inserted_ref_updates(pool: PgPool) { + let state = test_state(pool).await; + let owner = "did:key:zEVENTSOWNERAAAAAAAAAAAAAAAAAAAAAAAAA"; + + // Insert a gossip event with owner_did set + 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!("{}/myrepo", owner.split(':').next_back().unwrap()), + owner_did: Some(owner.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); + assert_eq!( + events[0]["repo"], + format!("{}/myrepo", owner.split(':').next_back().unwrap()) + ); + assert_eq!(events[0]["owner_did"], owner); + } + #[sqlx::test] async fn list_all_bounties_past_private_window_finds_public(pool: PgPool) { let state = test_state(pool).await; @@ -3549,6 +3601,42 @@ mod tests { assert!(resp.status().is_success()); } + #[sqlx::test] + async fn events_limit_respects_limit_param(pool: PgPool) { + let state = test_state(pool).await; + let owner = "did:key:zEVENTLIMITAAAAAAAAAAAAAAAAAAAAAAAA"; + + for i in 0..5 { + 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!("{}/r{i}", owner.split(':').next_back().unwrap()), + owner_did: Some(owner.into()), + ref_name: "refs/heads/main".into(), + old_sha: "0000000000000000000000000000000000000000".into(), + new_sha: format!("{i:040x}"), + timestamp: format!("2026-07-02T12:00:{i:02}Z"), + cert_id: None, + received_at: format!("2026-07-02T12:00:{i:02}Z"), + from_peer: "12D3KooWTest".into(), + }) + .await + .unwrap(); + } + + let resp = events_router(state) + .oneshot(anon_get("/api/v1/events/ref-updates?limit=2")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + assert_eq!(body["count"].as_i64(), Some(2)); + assert_eq!(body["events"].as_array().unwrap().len(), 2); + } + #[sqlx::test] async fn claim_bounty_gate_denies_non_reader_on_private(pool: PgPool) { let state = test_state(pool).await; diff --git a/crates/icaptcha-client/Cargo.toml b/crates/icaptcha-client/Cargo.toml index c55055a1..0a3a9215 100644 --- a/crates/icaptcha-client/Cargo.toml +++ b/crates/icaptcha-client/Cargo.toml @@ -15,3 +15,4 @@ serde_json = { workspace = true } anyhow = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } +sha2 = { workspace = true } diff --git a/crates/icaptcha-client/src/lib.rs b/crates/icaptcha-client/src/lib.rs index c41e368e..e32de31a 100644 --- a/crates/icaptcha-client/src/lib.rs +++ b/crates/icaptcha-client/src/lib.rs @@ -19,6 +19,7 @@ use anyhow::{anyhow, bail, Context, Result}; use serde::Deserialize; use serde_json::json; +pub mod pow; pub mod solvers; /// Default iCaptcha service base URL (used when the node doesn't advertise one). @@ -81,18 +82,42 @@ fn sanitize_excerpt(s: &str) -> String { out } -/// Whether `u` parses as an `https` URL. +/// Whether `u` parses as a trusted URL. +/// +/// Production trusts only `https`. When `GITLAWB_ICAPTCHA_INSECURE` is set (a +/// runtime escape hatch for integration tests) the function also trusts +/// `http://127.0.0.1` and `http://localhost` so the full iCaptcha retry path +/// can be exercised against a local mockito server. fn is_https(u: &str) -> bool { - reqwest::Url::parse(u) - .map(|p| p.scheme() == "https") - .unwrap_or(false) + let Ok(parsed) = reqwest::Url::parse(u) else { + return false; + }; + if parsed.scheme() == "https" { + return true; + } + if std::env::var_os("GITLAWB_ICAPTCHA_INSECURE").is_some() + && parsed.scheme() == "http" + && matches!(parsed.host_str(), Some("127.0.0.1" | "localhost")) + { + return true; + } + false } -/// Lowercased host of a URL, if it parses and has one. -fn host_of(u: &str) -> Option { - reqwest::Url::parse(u) - .ok() - .and_then(|p| p.host_str().map(|h| h.to_ascii_lowercase())) +/// Lowercased origin string (`scheme://host[:port]`) of a URL. +/// +/// Includes the effective port so that `http://localhost:3000` and +/// `http://localhost:9000` are treated as distinct origins — a hostile node +/// cannot redirect the iCaptcha solve to a different loopback port. +fn origin_key(u: &str) -> Option { + let parsed = reqwest::Url::parse(u).ok()?; + let scheme = parsed.scheme(); + let host = parsed.host_str()?.to_ascii_lowercase(); + let port = parsed + .port_or_known_default() + .map(|p| format!(":{p}")) + .unwrap_or_default(); + Some(format!("{scheme}://{host}{port}")) } /// Decide which iCaptcha origin to actually talk to, and whether it is trusted @@ -109,12 +134,13 @@ fn host_of(u: &str) -> Option { /// exfiltrated to an origin the operator did not choose. fn resolve_solver_url(advertised: Option<&str>, operator: Option<&str>) -> (String, bool) { let operator = operator.filter(|u| is_https(u)); - let operator_host = operator.and_then(host_of); - let default_host = host_of(DEFAULT_URL); + let operator_origin = operator.as_ref().and_then(|u| origin_key(u)); + let default_origin = origin_key(DEFAULT_URL); - let allowed = |host: &Option| -> bool { - host.as_ref() - .map(|h| Some(h) == default_host.as_ref() || Some(h) == operator_host.as_ref()) + let allowed = |origin: &Option| -> bool { + origin + .as_ref() + .map(|o| Some(o) == default_origin.as_ref() || Some(o) == operator_origin.as_ref()) .unwrap_or(false) }; @@ -125,7 +151,7 @@ fn resolve_solver_url(advertised: Option<&str>, operator: Option<&str>) -> (Stri }; let chosen = match advertised { - Some(a) if is_https(a) && allowed(&host_of(a)) => a.to_string(), + Some(a) if is_https(a) && allowed(&origin_key(a)) => a.to_string(), Some(a) => { tracing::warn!( advertised = %a, @@ -137,7 +163,7 @@ fn resolve_solver_url(advertised: Option<&str>, operator: Option<&str>) -> (Stri }; // Key goes only to the operator's own configured origin. - let key_trusted = operator_host.is_some() && host_of(&chosen) == operator_host; + let key_trusted = operator_origin.is_some() && origin_key(&chosen) == operator_origin; (chosen, key_trusted) } @@ -201,6 +227,10 @@ pub struct Challenge { pub difficulty: u32, pub prompt: String, pub token: String, + /// Proof-of-work to solve and echo back as `powNonce` on the answer. Present + /// only when the service has PoW enabled; absent/None means no PoW required. + #[serde(default)] + pub pow: Option, } #[derive(Debug, Deserialize)] @@ -237,7 +267,28 @@ pub fn obtain_proof(cfg: &IcaptchaCfg, solver: Option<&Solver>) -> Result Some(pow::solve(p).ok_or_else(|| { + anyhow!( + "cannot solve iCaptcha proof-of-work (algorithm '{}', difficulty {})", + p.algorithm, + p.difficulty + ) + })?), + None => None, + }; + + match submit_answer( + &client, + cfg, + &challenge.token, + &answer, + pow_nonce.as_deref(), + )? { AnswerResult::Passed { proof } => return Ok(proof), AnswerResult::Continue { challenge: next } => challenge = next, AnswerResult::Failed { reason } => { @@ -273,11 +324,14 @@ fn submit_answer( cfg: &IcaptchaCfg, token: &str, answer: &str, + pow_nonce: Option<&str>, ) -> Result { let url = format!("{}/v1/answer", cfg.url.trim_end_matches('/')); - let mut req = client - .post(&url) - .json(&json!({ "token": token, "answer": answer })); + let mut body = json!({ "token": token, "answer": answer }); + if let Some(nonce) = pow_nonce { + body["powNonce"] = json!(nonce); + } + let mut req = client.post(&url).json(&body); if let Some(key) = &cfg.api_key { req = req.bearer_auth(key); } @@ -320,6 +374,37 @@ fn interactive_prompt(challenge: &Challenge) -> Option { #[cfg(test)] mod tests { use super::*; + use std::ffi::OsString; + use std::sync::{Mutex, MutexGuard}; + + /// Serializes tests that touch the process-global + /// `GITLAWB_ICAPTCHA_INSECURE` env var so they never race. + static ICAPTCHA_ENV_LOCK: Mutex<()> = Mutex::new(()); + + /// Set `GITLAWB_ICAPTCHA_INSECURE` for the test lifetime, restoring any + /// prior value on drop. + struct InsecureEnv { + _lock: MutexGuard<'static, ()>, + prev: Option, + } + + impl InsecureEnv { + fn new() -> Self { + let lock = ICAPTCHA_ENV_LOCK.lock().unwrap(); + let prev = std::env::var_os("GITLAWB_ICAPTCHA_INSECURE"); + std::env::set_var("GITLAWB_ICAPTCHA_INSECURE", "1"); + InsecureEnv { _lock: lock, prev } + } + } + + impl Drop for InsecureEnv { + fn drop(&mut self) { + match self.prev.take() { + Some(v) => std::env::set_var("GITLAWB_ICAPTCHA_INSECURE", v), + None => std::env::remove_var("GITLAWB_ICAPTCHA_INSECURE"), + } + } + } // ── P1: hostile error bodies must not reach the terminal raw ────────── @@ -403,4 +488,20 @@ mod tests { assert_eq!(url, DEFAULT_URL); assert!(!key_trusted); } + + #[test] + fn rejects_insecure_advertised_url_on_different_port() { + // With GITLAWB_ICAPTCHA_INSECURE=1, two loopback URLs on different + // ports must be treated as distinct origins so a hostile node cannot + // redirect the bearer key to a different listener on localhost. + let _env = InsecureEnv::new(); + + let op = "http://localhost:3000"; + let (url, key_trusted) = resolve_solver_url(Some("http://localhost:9000"), Some(op)); + assert_eq!( + url, op, + "must fall back to operator origin, not use different port" + ); + assert!(key_trusted, "key stays trusted with operator origin"); + } } diff --git a/crates/icaptcha-client/src/pow.rs b/crates/icaptcha-client/src/pow.rs new file mode 100644 index 00000000..a3eb5b40 --- /dev/null +++ b/crates/icaptcha-client/src/pow.rs @@ -0,0 +1,130 @@ +//! Proof-of-work solver for the iCaptcha answer step. +//! +//! iCaptcha requires a small proof-of-work with each answer so that minting a +//! proof costs CPU — a per-proof cost a distributed abuser cannot dodge by +//! rotating IPs. The work is bound to the per-challenge id: find a `nonce` such +//! that `sha256("{challenge}:{nonce}")` has at least `difficulty` leading zero +//! bits. This mirrors `icaptcha/src/pow.ts`. +//! +//! At the service default (~20 bits) this is well under a second for a single +//! solve; we still cap the search so a hostile/misconfigured difficulty can +//! never hang the client. + +use sha2::{Digest, Sha256}; + +/// PoW parameters advertised by the service in a challenge (mirrors the +/// service's `PowChallenge`). +#[derive(Debug, Clone, serde::Deserialize)] +pub struct PowChallenge { + /// Only `sha256-leading-zero-bits` is understood; other algorithms are + /// rejected by [`solve`] rather than silently mis-solved. + pub algorithm: String, + /// The string to hash against (the per-challenge id). + pub challenge: String, + /// Required leading zero bits. + pub difficulty: u32, +} + +/// Hard cap on nonce search iterations. 2^26 ≈ 67M keeps a ~20-bit target (~1M +/// expected hashes) comfortably solvable while bounding worst-case work if the +/// service ever advertises an unexpectedly high difficulty. +const MAX_ITERS: u64 = 1 << 26; + +const ALGORITHM: &str = "sha256-leading-zero-bits"; + +/// Leading zero bits of a byte slice. +fn leading_zero_bits(bytes: &[u8]) -> u32 { + let mut bits = 0; + for &b in bytes { + if b == 0 { + bits += 8; + } else { + bits += b.leading_zeros(); + break; + } + } + bits +} + +/// sha256(`{challenge}:{nonce}`). +fn pow_hash(challenge: &str, nonce: &str) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(challenge.as_bytes()); + hasher.update(b":"); + hasher.update(nonce.as_bytes()); + hasher.finalize().into() +} + +/// Solve the PoW, returning a nonce whose hash meets the difficulty. Returns +/// `None` for an unknown algorithm or if no nonce is found within the cap (a +/// difficulty far above what the service uses in practice). +pub fn solve(pow: &PowChallenge) -> Option { + if pow.algorithm != ALGORITHM { + tracing::warn!(algorithm = %pow.algorithm, "unknown iCaptcha PoW algorithm; cannot solve"); + return None; + } + if pow.difficulty == 0 { + return Some("0".to_string()); + } + for i in 0..MAX_ITERS { + let nonce = format!("{i:x}"); + if leading_zero_bits(&pow_hash(&pow.challenge, &nonce)) >= pow.difficulty { + return Some(nonce); + } + } + tracing::warn!( + difficulty = pow.difficulty, + max_iters = MAX_ITERS, + "iCaptcha PoW not solved within iteration cap" + ); + None +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pow(challenge: &str, difficulty: u32) -> PowChallenge { + PowChallenge { + algorithm: ALGORITHM.to_string(), + challenge: challenge.to_string(), + difficulty, + } + } + + #[test] + fn leading_zero_bits_counts_across_bytes() { + assert_eq!(leading_zero_bits(&[0xff]), 0); + assert_eq!(leading_zero_bits(&[0x7f]), 1); + assert_eq!(leading_zero_bits(&[0x01]), 7); + assert_eq!(leading_zero_bits(&[0x00, 0xff]), 8); + assert_eq!(leading_zero_bits(&[0x00, 0x01]), 15); + } + + #[test] + fn solves_and_the_solution_verifies() { + let p = pow("challenge-abc", 12); + let nonce = solve(&p).expect("should solve a 12-bit target"); + assert!(leading_zero_bits(&pow_hash(&p.challenge, &nonce)) >= 12); + } + + #[test] + fn solution_is_bound_to_the_challenge() { + // A nonce solving one challenge should (overwhelmingly) not solve another + // at the same difficulty — the work is challenge-specific. + let nonce = solve(&pow("challenge-A", 12)).unwrap(); + assert!(leading_zero_bits(&pow_hash("challenge-B", &nonce)) < 12); + } + + #[test] + fn rejects_unknown_algorithm() { + let mut p = pow("c", 8); + p.algorithm = "scrypt".to_string(); + assert!(solve(&p).is_none()); + } + + #[test] + fn zero_difficulty_is_trivially_solved() { + assert_eq!(solve(&pow("c", 0)).as_deref(), Some("0")); + } +}