From 957cc3de82e169e33964b26ffd4c06a929aa73ac Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 10 Jul 2026 10:36:32 +0600 Subject: [PATCH 1/8] =?UTF-8?q?fix:=20correct=20v10=20migration=20expand/c?= =?UTF-8?q?ontract=20=E2=80=94=20keep=20old=20PK,=20add=20UNIQUE=20(repo,?= =?UTF-8?q?=20sha256=5Fhex)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rather than dropping the old sha256_hex PK and replacing it with a UNIQUE index on sha256_hex (which conflicts with the per-repo schema), this expand phase: • Keeps the old PK for pre-v10 ON CONFLICT(sha256_hex) compatibility. • Adds UNIQUE (repo, sha256_hex) for post-v10 ON CONFLICT(repo, sha256_hex). • Defers the PK switch to a future contract migration. The migration test now exercises both phases: 1. Verifies that duplicate SHA across repos is rejected in phase 1. 2. Applies the contract commands and verifies it succeeds in phase 2. --- crates/gitlawb-node/src/db/mod.rs | 356 ++++++++++++++++++++++++++++++ 1 file changed, 356 insertions(+) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 5324a4b..407310e 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -853,6 +853,24 @@ const MIGRATIONS: &[Migration] = &[ "ALTER TABLE repos ADD COLUMN IF NOT EXISTS quarantined BOOLEAN NOT NULL DEFAULT FALSE", ], }, + // Expand/contract migration for blue/green deploy compatibility. + // + // Phase 1 (this migration — expand): + // • Add repo/owner_did, backfill, NOT NULL. + // • Add UNIQUE (repo, sha256_hex) so post-v10 code can use + // ON CONFLICT(repo, sha256_hex). + // • Keep the old sha256_hex primary key — pre-v10 binaries that issue + // INSERT … ON CONFLICT(sha256_hex) still have their conflict target. + // + // During phase 1 the old PK on sha256_hex means the same Git object + // cannot appear under two repos yet. Pre-v10 writers keep working, + // post-v10 writers use the new compound conflict path. + // + // Phase 2 (future migration — contract, once all old writers are drained): + // • ALTER TABLE pinned_cids DROP CONSTRAINT pinned_cids_pkey; + // • DROP INDEX IF EXISTS pinned_cids_repo_sha_hex_key; + // • ALTER TABLE pinned_cids ADD PRIMARY KEY (repo, sha256_hex); + // • Same SHA can now appear in different repos. Migration { version: 10, name: "ref_cert_unique_per_ref", @@ -872,6 +890,69 @@ const MIGRATIONS: &[Migration] = &[ "CREATE UNIQUE INDEX IF NOT EXISTS idx_ref_certs_repo_ref ON ref_certificates(repo_id, ref_name)", ], }, + // Expand/contract migration for blue/green deploy compatibility. + // + // Phase 1 (this migration — expand): + // • Add repo/owner_did, backfill, NOT NULL. + // • Add UNIQUE (repo, sha256_hex) so post-v10 code can use + // ON CONFLICT(repo, sha256_hex). + // • Keep the old sha256_hex primary key — pre-v10 binaries that issue + // INSERT … ON CONFLICT(sha256_hex) still have their conflict target. + // + // During phase 1 the old PK on sha256_hex means the same Git object + // cannot appear under two repos yet. record_pinned_cid still uses + // ON CONFLICT(sha256_hex) so both pre-v10 and post-v10 writers target + // the kept PK; the compound conflict path will replace it in phase 2. + // + // Phase 2 (future migration — contract, once all old writers are drained): + // • ALTER TABLE pinned_cids DROP CONSTRAINT pinned_cids_pkey; + // • DROP INDEX IF EXISTS pinned_cids_repo_sha_hex_key; + // • ALTER TABLE pinned_cids ADD PRIMARY KEY (repo, sha256_hex); + // • Same SHA can now appear in different repos. + Migration { + version: 11, + name: "pinned_cids_repo_owner", + stmts: &[ + "ALTER TABLE pinned_cids ADD COLUMN IF NOT EXISTS repo TEXT", + "ALTER TABLE pinned_cids ADD COLUMN IF NOT EXISTS owner_did TEXT", + // Backfill repo/owner only when the cid maps to exactly one + // unambiguous (repo, owner_did) pair under did:key-aware owner key + // normalization. This avoids nondeterministic assignments. + r#"UPDATE pinned_cids p + SET repo = m.repo, + owner_did = m.owner_did + FROM ( + SELECT + bc.cid, + MIN(bc.repo) AS repo, + MIN(r.owner_did) AS owner_did + FROM branch_cids bc + JOIN repos r + ON r.name = split_part(bc.repo, '/', 2) + AND (CASE WHEN r.owner_did LIKE 'did:key:%' AND position(':' in substr(r.owner_did, 9)) = 0 THEN substr(r.owner_did, 9) ELSE r.owner_did END) + = split_part(bc.repo, '/', 1) + GROUP BY bc.cid + HAVING COUNT(DISTINCT (bc.repo || '|' || r.owner_did)) = 1 + ) m + WHERE p.cid = m.cid"#, + // Fallback for remaining rows + "UPDATE pinned_cids SET repo = '' WHERE repo IS NULL", + "UPDATE pinned_cids SET owner_did = '' WHERE owner_did IS NULL", + "ALTER TABLE pinned_cids ALTER COLUMN repo SET NOT NULL", + "ALTER TABLE pinned_cids ALTER COLUMN owner_did SET NOT NULL", + // New unique constraint for post-v11 ON CONFLICT(repo, sha256_hex) + "CREATE UNIQUE INDEX IF NOT EXISTS pinned_cids_repo_sha_hex_key ON pinned_cids (repo, sha256_hex)", + // Old PK on sha256_hex kept intact for pre-v11 ON CONFLICT(sha256_hex) + "CREATE INDEX IF NOT EXISTS idx_pinned_cids_repo_owner ON pinned_cids (repo, owner_did)", + ], + }, + Migration { + version: 12, + name: "arweave_anchors_repo_owner_index", + stmts: &[ + "CREATE INDEX IF NOT EXISTS idx_arweave_anchors_repo_owner_anchored ON arweave_anchors (repo, owner_did, anchored_at DESC)", + ], + }, ]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -3406,6 +3487,281 @@ mod migration_tests { // it, you must also update the backfill. assert_eq!(MIGRATIONS[0].name, MIGRATION_V1_NAME); } + #[sqlx::test] + async fn test_migration_v11_upgrade_path(pool: PgPool) { + let db = Db::for_testing(pool); + + // Run migrations up to version 9 + async fn run_migrations_up_to(db: &Db, version: i64) { + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT NOT NULL PRIMARY KEY, + name TEXT NOT NULL, + applied_at TEXT NOT NULL + )"#, + ) + .execute(&db.pool) + .await + .unwrap(); + + for m in super::MIGRATIONS { + if m.version > version { + break; + } + let already: bool = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied", + ) + .bind(m.version) + .fetch_one(&db.pool) + .await + .unwrap() + .get::("applied"); + + if already { + continue; + } + + let mut tx = db.pool.begin().await.unwrap(); + for stmt in m.stmts { + sqlx::query(stmt).execute(&mut *tx).await.unwrap(); + } + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind(chrono::Utc::now().to_rfc3339()) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + } + } + + run_migrations_up_to(&db, 9).await; + + // Seed a repo, branch_cids, and pinned_cids under v9 schema + sqlx::query( + "INSERT INTO repos (id, name, owner_did, description, is_public, default_branch, created_at, updated_at, disk_path) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)" + ) + .bind("repo-123") + .bind("myrepo") + .bind("did:key:z6Mkwowner") + .bind("desc") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/repo-123") + .execute(&db.pool) + .await + .unwrap(); + + sqlx::query( + "INSERT INTO branch_cids (repo, ref_name, sha, cid, node_did, updated_at) + VALUES ($1, $2, $3, $4, $5, $6)", + ) + .bind("z6Mkwowner/myrepo") + .bind("refs/heads/main") + .bind("old-sha") + .bind("old-cid") + .bind("node-did") + .bind("2026-07-03T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) + VALUES ($1, $2, $3)", + ) + .bind("old-sha") + .bind("old-cid") + .bind("2026-07-03T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + + // Run remaining migrations (v10 = ref_cert_dedup, v11 = pinned_cids) + db.run_migrations().await.unwrap(); + + // Verify backfilling of repo and owner_did columns + let row = sqlx::query( + "SELECT sha256_hex, cid, repo, owner_did FROM pinned_cids WHERE sha256_hex = 'old-sha'", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + + assert_eq!(row.get::("repo"), "z6Mkwowner/myrepo"); + assert_eq!(row.get::("owner_did"), "did:key:z6Mkwowner"); + + // Phase 1 (expand): the old PK on sha256_hex still rejects duplicate + // SHA across repos — pre-v10 ON CONFLICT(sha256_hex) keeps working. + let res = sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind("old-sha") + .bind("old-cid") + .bind("2026-07-03T00:00:00Z") + .bind("other-repo") + .bind("other-owner") + .execute(&db.pool) + .await; + + assert!( + res.is_err(), + "Phase 1: old PK on sha256_hex must reject duplicate SHA across repos" + ); + + // Phase 2 (contract): drop the old PK and UNIQUE, promote to compound PK. + // Once all pre-v10 writers are drained this step makes the migration + // complete — same SHA can appear in different repos. + sqlx::query("ALTER TABLE pinned_cids DROP CONSTRAINT pinned_cids_pkey") + .execute(&db.pool) + .await + .unwrap(); + sqlx::query("DROP INDEX IF EXISTS pinned_cids_repo_sha_hex_key") + .execute(&db.pool) + .await + .unwrap(); + sqlx::query("ALTER TABLE pinned_cids ADD PRIMARY KEY (repo, sha256_hex)") + .execute(&db.pool) + .await + .unwrap(); + + // Now the same SHA works in a different repo (compound PK allows it). + let res = sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind("old-sha") + .bind("old-cid") + .bind("2026-07-03T00:00:00Z") + .bind("other-repo") + .bind("other-owner") + .execute(&db.pool) + .await; + + assert!( + res.is_ok(), + "Phase 2: compound PK must allow same SHA in different repos" + ); + } + + /// A pinned CID whose SHA is not a current branch_cids ref tip falls back to + /// repo = '' after migration v11. This tests that the backfill does not + /// silently orphan such pins by leaving repo NULL/unqueryable; the empty + /// string is at least queryable by list_pinned_cids_for_repos callers. + #[sqlx::test] + async fn test_migration_v11_orphan_non_tip_pin(pool: PgPool) { + let db = Db::for_testing(pool); + + // Run migrations up to version 9 + async fn run_migrations_up_to(db: &Db, version: i64) { + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT NOT NULL PRIMARY KEY, + name TEXT NOT NULL, + applied_at TEXT NOT NULL + )"#, + ) + .execute(&db.pool) + .await + .unwrap(); + + for m in super::MIGRATIONS { + if m.version > version { + break; + } + let already: bool = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied", + ) + .bind(m.version) + .fetch_one(&db.pool) + .await + .unwrap() + .get::("applied"); + + if already { + continue; + } + + let mut tx = db.pool.begin().await.unwrap(); + for stmt in m.stmts { + sqlx::query(stmt).execute(&mut *tx).await.unwrap(); + } + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind(chrono::Utc::now().to_rfc3339()) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + } + } + + run_migrations_up_to(&db, 9).await; + + // Seed a repo and a pinned_cid, but no matching branch_cids entry. + sqlx::query( + "INSERT INTO repos (id, name, owner_did, description, is_public, default_branch, created_at, updated_at, disk_path) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)" + ) + .bind("repo-orphan") + .bind("orphan-repo") + .bind("did:key:z6Mkworphan") + .bind("desc") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/orphan") + .execute(&db.pool) + .await + .unwrap(); + + // This CID is a pinned object that is NOT a current ref tip — + // no matching row in branch_cids exists. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) + VALUES ($1, $2, $3)", + ) + .bind("orphan-sha") + .bind("orphan-cid") + .bind("2026-07-03T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + + // Run remaining migrations (v10 = ref_cert_dedup, v11 = pinned_cids) + db.run_migrations().await.unwrap(); + + // The orphan pin should have fallen back to repo = '' because + // branch_cids had no matching cid to backfill from. + let row = sqlx::query( + "SELECT sha256_hex, repo, owner_did FROM pinned_cids WHERE sha256_hex = 'orphan-sha'", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + + assert_eq!( + row.get::("repo"), + "", + "non-tip pin must fall back to empty repo" + ); + assert_eq!( + row.get::("owner_did"), + "", + "non-tip pin must fall back to empty owner_did" + ); + } } #[cfg(test)] From 5b34cebb5a4c5e35c8db0d8bae7530f37d828f13 Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 10 Jul 2026 12:22:50 +0600 Subject: [PATCH 2/8] =?UTF-8?q?fix:=20address=205=20findings=20=E2=80=94?= =?UTF-8?q?=20migration=20per-repo=20writes,=20walk=20budget=20skip,=20MAX?= =?UTF-8?q?=5FPROBES=20discard,=20response=20body=20bound,=20node=20status?= =?UTF-8?q?=20400?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [P1] Defer new per-repo conflict target to phase 2 record_pinned_cid and record_pinata_cid now target ON CONFLICT(sha256_hex) (the kept old PK) so both pre-v10 and post-v10 writers share the same conflict target during the expand phase. [P1] Walk budget boundary pin no longer skipped Push Some(false) for the unprocessed boundary pin and set walk_limit_idx to i+1 so the strict-< cursor never skips it permanently. [P2] Structural pins past MAX_PROBES not silently discarded Track the earliest unprobed structural candidate (probe_limit) and stop phase 3 before it, with page_truncated set for continuation. [P2] Bound response body before JSON deserialization Both CLI consumers (ipfs_cmd::list_pins_paginated, node::fetch_pins) read bytes first, check a 64 MiB cap, then deserialize. [P2] Node status panel recovers from 400 expired truncated_cursor Adds last_next_cursor tracking and restores from it on 400 + had_truncated, preserving already-accumulated counts instead of returning Unavailable. --- crates/gitlawb-node/src/api/ipfs.rs | 352 +++++++++++++++++++++++++++- crates/gitlawb-node/src/db/mod.rs | 116 ++++++++- crates/gl/src/ipfs_cmd.rs | 102 ++++++++ crates/gl/src/node.rs | 106 ++++++++- 4 files changed, 658 insertions(+), 18 deletions(-) diff --git a/crates/gitlawb-node/src/api/ipfs.rs b/crates/gitlawb-node/src/api/ipfs.rs index f3de757..d8fc5a9 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -223,7 +223,357 @@ pub async fn list_pins(State(state): State) -> Result Option<(String, String, String)> { + let bytes = URL_SAFE_NO_PAD.decode(s.as_bytes()).ok()?; + let decoded = String::from_utf8(bytes).ok()?; + let parts: Vec<&str> = decoded.splitn(3, '|').collect(); + if parts.len() == 3 { + Some(( + parts[0].to_string(), + parts[1].to_string(), + parts[2].to_string(), + )) + } else { + None + } + }; + let encode_cursor = |pa: &str, r: &str, sha: &str| -> String { + URL_SAFE_NO_PAD.encode(format!("{pa}|{r}|{sha}")) + }; + + let initial_cursor = match query.cursor.as_ref() { + Some(c) => match decode_cursor(c) { + Some(cursor) => Some(cursor), + None => { + return Err(AppError::BadRequest( + "invalid cursor: expected base64-encoded pinned_at|repo|sha256_hex".into(), + )) + } + }, + None => None, + }; + + // Truncated resume cursor: XChaCha20Poly1305 AEAD token. Decrypts to the + // same (pinned_at, repo, sha256_hex) cursor on the server side but the + // caller cannot decode hidden-row metadata from the wire format. If the + // token is present but undecodable we return an explicit error so the + // client does not silently restart at page 1. + let truncated_resume = match query.truncated_cursor.as_ref() { + Some(t) => { + let seed = state.node_keypair.to_seed(); + match decode_opaque_cursor(&seed, t) { + Some(c) => Some(c), + None => { + return Err(AppError::BadRequest( + "invalid or expired truncated_cursor".into(), + )) + } + } + } + None => None, + }; + + // Build a lookup of slug -> (repo, rules) once. + let mut repos_by_slug = HashMap::new(); + for r in repos { + let short = crate::db::normalize_owner_key(&r.owner_did); + let slug = format!("{}/{}", short, r.name); + let rules = rules_by_repo.get(&r.id).cloned().unwrap_or_default(); + repos_by_slug.insert(slug, (r, rules)); + } + + // Use keyset pagination to fetch batches and post-filter path-scoped + // hidden pins so the caller still receives up to `max_visible` visible + // entries even when newer pins are hidden under /secret/** rules. + // Keyset cursor avoids duplicate/skip rows when new pins land between + // batches (unlike LIMIT/OFFSET) and removes the cost of deep OFFSET + // re-scanning. + // + // The loop is bounded by MAX_BATCHES to prevent a single request from + // scanning an unbounded number of hidden rows. Path-scoped git walks + // are independently bounded by MAX_WALKS as a secondary safeguard. + // + // next_cursor is derived from the last *accepted* (visible) pin, never + // from the last scanned row, to avoid leaking withheld-blob metadata + // or skipping rows the caller was never shown. + const BATCH_SIZE: i64 = 200; + const MAX_BATCHES: usize = 10; + const MAX_WALKS: usize = 50; + const MAX_PROBES: usize = 200; + let mut batch_count = 0usize; + let mut pins = Vec::new(); + let mut db_cursor: Option<(String, String, String)> = truncated_resume.or(initial_cursor); + let mut response_cursor: Option<(String, String, String)> = None; + let mut allowed_blobs_by_repo: HashMap, HashSet, PathBuf)> = + HashMap::new(); + let mut page_truncated = false; + // Per-repo cache of sha256_hex → is_structural (true for commit/tree/tag). + let mut structural_cache: HashMap> = HashMap::new(); + let mut probe_count = 0usize; + + 'fetch: loop { + if batch_count >= MAX_BATCHES { + break; + } + batch_count += 1; + + let batch = if query_repos.is_empty() { + Vec::new() + } else { + state + .db + .list_pinned_cids_for_repos( + &query_repos, + &query_owner_dids, + BATCH_SIZE, + db_cursor + .as_ref() + .map(|(pa, r, sha)| (pa.as_str(), r.as_str(), sha.as_str())), + ) + .await + .map_err(AppError::Internal)? + }; + + if batch.is_empty() { + break; + } + + // ── Phase 1 — collect structural candidates per repo ────────────── + // Track per-pin outcome: None = structural candidate (needs type check + // before final decision), Some(false) = hidden, Some(true) = visible. + let mut pin_outcome: Vec> = Vec::with_capacity(batch.len()); + let mut structural_candidates: HashMap> = HashMap::new(); + let mut walk_limit_idx = batch.len(); + + for (i, pin) in batch.iter().enumerate() { + if pin.repo.is_empty() { + db_cursor = Some(( + pin.pinned_at.clone(), + pin.repo.clone(), + pin.sha256_hex.clone(), + )); + pin_outcome.push(None); + continue; + } + let Some((repo, rules)) = repos_by_slug.get(&pin.repo) else { + // Unknown slug — advance cursor past it, no visibility check. + db_cursor = Some(( + pin.pinned_at.clone(), + pin.repo.clone(), + pin.sha256_hex.clone(), + )); + pin_outcome.push(None); + continue; + }; + + if !has_path_scoped_rule(rules) { + // No path-scoped rules — every pin from this repo is visible. + pin_outcome.push(Some(true)); + continue; + } + + // Path-scoped repo — ensure walk result is cached. + if !allowed_blobs_by_repo.contains_key(&repo.id) { + if allowed_blobs_by_repo.len() >= MAX_WALKS { + // Walk budget exhausted. Mark this pin denied (we cannot + // determine visibility without the walk) and advance past + // it so the next request never revisits the boundary pin + // with a strict-< cursor predicate that would skip it. + pin_outcome.push(Some(false)); + page_truncated = true; + walk_limit_idx = i + 1; + break; + } + let (allowed, all_blobs, repo_path) = + match state.repo_store.acquire(&repo.owner_did, &repo.name).await { + Ok(rp) => { + let rp_clone = rp.clone(); + let r_clone = rules.clone(); + let is_public = repo.is_public; + let owner = repo.owner_did.clone(); + let caller_for_walk = caller_owned.clone(); + + // Wall-clock timeout bounds the request's wait; + // note: tokio::time::timeout does NOT cancel the + // spawn_blocking task, so the blocking thread may + // continue running until completion. + let walk_fut = tokio::task::spawn_blocking(move || { + allowed_blob_set_for_caller( + &rp_clone, + &r_clone, + is_public, + &owner, + caller_for_walk.as_deref(), + ) + }); + match tokio::time::timeout(std::time::Duration::from_secs(60), walk_fut) + .await + { + Ok(Ok(Ok((a, b)))) => (a, b, rp), + _ => (HashSet::new(), HashSet::new(), PathBuf::new()), + } + } + Err(_) => (HashSet::new(), HashSet::new(), PathBuf::new()), + }; + allowed_blobs_by_repo.insert(repo.id.clone(), (allowed, all_blobs, repo_path)); + } + + let (allowed, all_blobs, repo_path) = allowed_blobs_by_repo.get(&repo.id).unwrap(); + if allowed.contains(&pin.sha256_hex) { + pin_outcome.push(Some(true)); + } else if !all_blobs.contains(&pin.sha256_hex) && !repo_path.as_os_str().is_empty() { + // Candidate for structural-object check. + pin_outcome.push(None); // deferred — decided after phase 2 + structural_candidates + .entry(repo.id.clone()) + .or_default() + .push((i, pin.sha256_hex.clone())); + } else { + pin_outcome.push(Some(false)); + } + } + + // Track the first unprobed structural candidate so phase 3 stops + // before it and the next request picks up where we left off. + let mut probe_limit = std::usize::MAX; + + // ── Phase 2 — batch-check structural candidates per repo ────────── + for (repo_id, candidates) in &structural_candidates { + if probe_count >= MAX_PROBES { + // Record the earliest unprobed index and stop. + for &(idx, _) in candidates { + if idx < probe_limit { + probe_limit = idx; + } + } + break; + } + let rp = allowed_blobs_by_repo + .get(repo_id) + .map(|(_, _, p)| p.clone()) + .unwrap_or_default(); + if rp.as_os_str().is_empty() { + continue; + } + // Filter to SHAs not already cached. + let repo_cache = structural_cache.entry(repo_id.clone()).or_default(); + let to_check: Vec = candidates + .iter() + .filter(|(_, sha)| !repo_cache.contains_key(sha)) + .map(|(_, sha)| sha.clone()) + .collect(); + if to_check.is_empty() { + continue; + } + let remaining = MAX_PROBES.saturating_sub(probe_count); + let to_check: Vec = to_check.into_iter().take(remaining).collect(); + probe_count += to_check.len(); + + let rp_for_block = rp.clone(); + let probe_fut = + tokio::task::spawn_blocking(move || batch_object_types(&rp_for_block, &to_check)); + let results = + match tokio::time::timeout(std::time::Duration::from_secs(30), probe_fut).await { + Ok(Ok(Ok(map))) => map, + _ => HashMap::new(), + }; + for (sha, obj_type) in results { + repo_cache.insert(sha, obj_type.is_some_and(|t| t != "blob")); + } + } + + // ── Phase 3 — emit visible pins ─────────────────────────────────── + for (i, pin) in batch.into_iter().enumerate() { + if i >= walk_limit_idx.min(probe_limit) { + // Past the MAX_WALKS wall or past an unprobed structural + // candidate — remaining pins belong to repos that never got + // walked or probed; handled by next request. + if i < probe_limit && page_truncated == false { + page_truncated = true; + } + break; + } + + let Some((repo, rules)) = repos_by_slug.get(&pin.repo) else { + // Already advanced past in phase 1 — just maintain cursor. + db_cursor = Some(( + pin.pinned_at.clone(), + pin.repo.clone(), + pin.sha256_hex.clone(), + )); + continue; + }; + + if !has_path_scoped_rule(rules) { + let pa = pin.pinned_at.clone(); + let r = pin.repo.clone(); + let sha = pin.sha256_hex.clone(); + response_cursor = Some((pa.clone(), r.clone(), sha.clone())); + pins.push(pin); + db_cursor = Some((pa, r, sha)); + } else { + let pa = pin.pinned_at.clone(); + let r = pin.repo.clone(); + let sha = pin.sha256_hex.clone(); + + let visible = match pin_outcome[i] { + Some(v) => v, + None => { + // Structural candidate — consult cache. + structural_cache + .get(&repo.id) + .and_then(|c| c.get(&pin.sha256_hex)) + .copied() + .unwrap_or(false) + } + }; + if visible { + response_cursor = Some((pa.clone(), r.clone(), sha.clone())); + pins.push(pin); + } + db_cursor = Some((pa, r, sha)); + } + + if pins.len() >= max_visible as usize { + break 'fetch; + } + } + } + let page_filled = pins.len() >= max_visible as usize; + if !page_truncated { + page_truncated = batch_count >= MAX_BATCHES; + } + pins.truncate(max_visible as usize); + + let mut body = serde_json::json!({ "pins": pins, "count": pins.len(), }))) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 407310e..226d627 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -863,8 +863,9 @@ const MIGRATIONS: &[Migration] = &[ // INSERT … ON CONFLICT(sha256_hex) still have their conflict target. // // During phase 1 the old PK on sha256_hex means the same Git object - // cannot appear under two repos yet. Pre-v10 writers keep working, - // post-v10 writers use the new compound conflict path. + // cannot appear under two repos yet. record_pinned_cid still uses + // ON CONFLICT(sha256_hex) so both pre-v10 and post-v10 writers target + // the kept PK; the compound conflict path will replace it in phase 2. // // Phase 2 (future migration — contract, once all old writers are drained): // • ALTER TABLE pinned_cids DROP CONSTRAINT pinned_cids_pkey; @@ -2247,10 +2248,28 @@ impl Db { Ok(row.get::("cnt") > 0) } - pub async fn record_pinned_cid(&self, sha256_hex: &str, cid: &str) -> Result<()> { + pub async fn get_pinned_cid(&self, sha256_hex: &str) -> Result> { + let row = sqlx::query("SELECT cid FROM pinned_cids WHERE sha256_hex = $1 LIMIT 1") + .bind(sha256_hex) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| r.get("cid"))) + } + + pub async fn record_pinned_cid( + &self, + sha256_hex: &str, + cid: &str, + repo: &str, + owner_did: &str, + ) -> Result<()> { + // Phase 1 (expand): targets the kept sha256_hex PK so pre-v10 and + // post-v10 writers share the same conflict target. Phase 2 (contract) + // will switch to ON CONFLICT(repo, sha256_hex) after the old PK is + // dropped and (repo, sha256_hex) becomes the new primary key. sqlx::query( - "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) - VALUES ($1, $2, $3) + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT(sha256_hex) DO NOTHING", ) .bind(sha256_hex) @@ -2337,6 +2356,93 @@ impl Db { ) .fetch_all(&self.pool) .await?; + Ok(row.get::("cnt") > 0) + } + + pub async fn get_pinata_cid(&self, sha256_hex: &str) -> Result> { + let row = sqlx::query("SELECT pinata_cid FROM pinned_cids WHERE sha256_hex = $1 AND pinata_cid IS NOT NULL LIMIT 1") + .bind(sha256_hex) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|r| r.get("pinata_cid"))) + } + + /// Record the Pinata CID for a git object. + /// Phase 1 targets the kept sha256_hex PK (same as record_pinned_cid); + /// phase 2 will switch to ON CONFLICT(repo, sha256_hex). + pub async fn record_pinata_cid( + &self, + sha256_hex: &str, + pinata_cid: &str, + repo: &str, + owner_did: &str, + ) -> Result<()> { + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, pinata_cid, repo, owner_did) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT(sha256_hex) DO UPDATE SET pinata_cid = EXCLUDED.pinata_cid", + ) + .bind(sha256_hex) + .bind(pinata_cid) // fallback local cid if row is new + .bind(Utc::now().to_rfc3339()) + .bind(pinata_cid) + .bind(repo) + .bind(owner_did) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Bounded global pin query: returns pins for any of the given (repo, owner_did) + /// pairs, ordered by pinned_at DESC, capped at `limit`. + /// Fetch pinned CIDs matching the readable (repo, owner_did) pairs, + /// ordered by (pinned_at DESC, repo DESC, sha256_hex DESC) for stable + /// keyset pagination. When `cursor` is `Some((pinned_at, repo, + /// sha256_hex))`, only rows strictly before that cursor are returned (no + /// duplicates across batches). + pub async fn list_pinned_cids_for_repos( + &self, + repos: &[String], + owner_dids: &[String], + limit: i64, + cursor: Option<(&str, &str, &str)>, + ) -> Result> { + let rows = if let Some((pa, r, sha)) = cursor { + sqlx::query( + "SELECT sha256_hex, cid, pinned_at, pinata_cid, repo, owner_did + FROM pinned_cids + WHERE (repo, owner_did) IN ( + SELECT * FROM UNNEST($1::text[], $2::text[]) + ) + AND (pinned_at, repo, sha256_hex) < ($3::text, $4::text, $5::text) + ORDER BY pinned_at DESC, repo DESC, sha256_hex DESC + LIMIT $6", + ) + .bind(repos) + .bind(owner_dids) + .bind(pa) + .bind(r) + .bind(sha) + .bind(limit) + .fetch_all(&self.pool) + .await? + } else { + sqlx::query( + "SELECT sha256_hex, cid, pinned_at, pinata_cid, repo, owner_did + FROM pinned_cids + WHERE (repo, owner_did) IN ( + SELECT * FROM UNNEST($1::text[], $2::text[]) + ) + ORDER BY pinned_at DESC, repo DESC, sha256_hex DESC + LIMIT $3", + ) + .bind(repos) + .bind(owner_dids) + .bind(limit) + .fetch_all(&self.pool) + .await? + }; + Ok(rows .into_iter() .map(|r| PinnedCidRecord { diff --git a/crates/gl/src/ipfs_cmd.rs b/crates/gl/src/ipfs_cmd.rs index d4cb64f..6ca16c9 100644 --- a/crates/gl/src/ipfs_cmd.rs +++ b/crates/gl/src/ipfs_cmd.rs @@ -86,6 +86,108 @@ async fn cmd_list(node: String, dir: Option) -> Result<()> { Ok(()) } +/// Paginate through the full pin listing, collecting all pins and handling +/// the expired-truncated_cursor retry (P2). The last_next_cursor restore +/// may cause a duplicate page (self-limiting via the cycle guard). +async fn list_pins_paginated(client: &NodeClient) -> Result<(Vec, bool)> { + let mut all_pins = Vec::new(); + let mut cursor: Option = None; + let mut truncated_cursor: Option = None; + // Persist the last next_cursor across the truncated leg so that an + // expired truncated_cursor (400) can resume from where we left off + // rather than restarting at page 1 (P2). Note: this may re-fetch + // the page before the truncated one (self-limiting via cycle guard). + let mut last_next_cursor: Option = None; + // Advancement guard: track every cursor value seen to detect cycles. + let mut seen_cursors: std::collections::HashSet = std::collections::HashSet::new(); + let mut incomplete = false; + let mut pages = 0u32; + // Bounds: at most 10 000 pages, 1 000 000 rows total, or 64 MiB per + // response body limits unbounded loops and prevents a single oversized + // page from exhausting memory before the row cap is checked. + const MAX_PAGES: u32 = 10_000; + const MAX_ROWS: usize = 1_000_000; + const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024; + + loop { + pages += 1; + if pages > MAX_PAGES { + incomplete = true; + break; + } + + let mut path = "/api/v1/ipfs/pins".to_string(); + let mut params = Vec::new(); + let mut had_truncated = false; + if let Some(c) = cursor.take() { + params.push(format!("cursor={}", urlencoding::encode(&c))); + } + if let Some(tc) = truncated_cursor.take() { + had_truncated = true; + params.push(format!("truncated_cursor={}", urlencoding::encode(&tc))); + } + if !params.is_empty() { + path.push('?'); + path.push_str(¶ms.join("&")); + } + + let resp = client.get_signed(&path).await?; + if !resp.status().is_success() { + let status = resp.status(); + if status == 400 && had_truncated { + // Server rejected the truncated_cursor (expired or wrong key). + // Restore from last_next_cursor so we don't restart at page 1. + cursor = last_next_cursor.clone(); + // incomplete is left unchanged: the initial false propagates + // when we recover to a natural end; otherwise MAX_PAGES, + // MAX_ROWS, or the cycle guard will set it. + continue; + } + let body = resp.text().await.unwrap_or_default(); + anyhow::bail!("node returned {status} for pins listing: {body}"); + } + let body = resp.bytes().await.context("failed to read pins response")?; + if body.len() > MAX_RESPONSE_BYTES { + anyhow::bail!("pins response body ({len} bytes) exceeds {MAX_RESPONSE_BYTES} limit", len = body.len()); + } + let resp: Value = serde_json::from_slice(&body) + .with_context(|| format!("failed to parse pins response ({len} bytes)", len = body.len()))?; + + let pins = resp["pins"].as_array().cloned().unwrap_or_default(); + + if all_pins.len() + pins.len() > MAX_ROWS { + incomplete = true; + break; + } + + let next = resp["next_cursor"].as_str().map(String::from); + let new_trunc = resp["truncated_cursor"].as_str().map(String::from); + + // Detect cursor cycling: keys on the exact (next_cursor, truncated) + // pair, so a node that returns a fresh pair every page never trips it. + // MAX_PAGES provides the ultimate bound (10 K round-trips per listing). + let cycle_key = + next.as_deref().unwrap_or("").to_string() + "|" + new_trunc.as_deref().unwrap_or(""); + if !cycle_key.is_empty() && !seen_cursors.insert(cycle_key) { + incomplete = true; + break; + } + + all_pins.extend(pins); + + if next.is_none() && new_trunc.is_none() { + break; + } + if let Some(ref n) = next { + last_next_cursor = Some(n.clone()); + } + cursor = next; + truncated_cursor = new_trunc; + } + + Ok((all_pins, incomplete)) +} + async fn cmd_get(cid: String, node: String) -> Result<()> { let client = NodeClient::new(&node, None); let path = format!("/ipfs/{cid}"); diff --git a/crates/gl/src/node.rs b/crates/gl/src/node.rs index 367ba57..a822f2e 100644 --- a/crates/gl/src/node.rs +++ b/crates/gl/src/node.rs @@ -241,19 +241,101 @@ async fn fetch_pins(node: &str, auth: PinsAuth) -> PinsPanel { PinsAuth::DirUnusable(dir) => return PinsPanel::IdentityError(dir), }; let client = NodeClient::new(node, Some(kp)); - let resp = match client.get_signed("/api/v1/ipfs/pins").await { - Ok(r) => r, - Err(_) => return PinsPanel::Unavailable, - }; - if !resp.status().is_success() { - return PinsPanel::Unavailable; + let mut total: u64 = 0; + let mut cursor: Option = None; + let mut truncated_cursor: Option = None; + // Persist the last next_cursor across the truncated leg so that an + // expired truncated_cursor (400) can resume from where we left off + // rather than restarting at page 1 (P2). + let mut last_next_cursor: Option = None; + let mut seen_cursors: std::collections::HashSet = std::collections::HashSet::new(); + let mut incomplete = false; + let mut pages = 0u32; + const MAX_PAGES: u32 = 10_000; + const MAX_ROWS: u64 = 1_000_000; + const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024; + + loop { + pages += 1; + if pages > MAX_PAGES { + incomplete = true; + break; + } + + let mut path = "/api/v1/ipfs/pins".to_string(); + let mut params = Vec::new(); + let mut had_truncated = false; + if let Some(c) = cursor.take() { + params.push(format!("cursor={}", urlencoding::encode(&c))); + } + if let Some(tc) = truncated_cursor.take() { + had_truncated = true; + params.push(format!("truncated_cursor={}", urlencoding::encode(&tc))); + } + if !params.is_empty() { + path.push('?'); + path.push_str(¶ms.join("&")); + } + + let resp = match client.get_signed(&path).await { + Ok(r) => r, + Err(_) => { + incomplete = true; + break; + } + }; + if !resp.status().is_success() { + if resp.status().as_u16() == 400 && had_truncated { + // Server rejected the truncated_cursor (expired or wrong key). + // Restore from last_next_cursor so we don't restart at page 1 + // and preserve the already-accumulated count. + cursor = last_next_cursor.clone(); + incomplete = true; + continue; + } + incomplete = true; + break; + } + let body = match resp.bytes().await { + Ok(b) => b, + Err(_) => return PinsPanel::Unavailable, + }; + if body.len() > MAX_RESPONSE_BYTES { + return PinsPanel::Unavailable; + } + let Ok(body) = serde_json::from_slice::(&body) else { + return PinsPanel::Unavailable; + }; + + let page_pins = body["pins"].as_array().map(|a| a.len() as u64).unwrap_or(0); + + if total + page_pins > MAX_ROWS { + incomplete = true; + break; + } + total += page_pins; + + let next = body["next_cursor"].as_str().map(String::from); + let new_trunc = body["truncated_cursor"].as_str().map(String::from); + + // Detect cursor cycling + let cycle_key = + next.as_deref().unwrap_or("").to_string() + "|" + new_trunc.as_deref().unwrap_or(""); + if !cycle_key.is_empty() && !seen_cursors.insert(cycle_key) { + incomplete = true; + break; + } + + if next.is_none() && new_trunc.is_none() { + break; + } + if let Some(ref n) = next { + last_next_cursor = Some(n.clone()); + } + cursor = next; + truncated_cursor = new_trunc; } - let Ok(body) = resp.json::().await else { - return PinsPanel::Unavailable; - }; - let count = body["count"] - .as_u64() - .unwrap_or_else(|| body["pins"].as_array().map(|a| a.len() as u64).unwrap_or(0)); + let count = total; if count == 0 { PinsPanel::Empty } else { From 6e61c73b0ef61bc8b685f325ec10cb9bbb6ad7da Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 10 Jul 2026 13:03:58 +0600 Subject: [PATCH 3/8] =?UTF-8?q?fix:=20fmt=20and=20probe=5Flimit=20rework?= =?UTF-8?q?=20=E2=80=94=20treat=20unprobed=20candidates=20conservatively?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Format ipfs_cmd.rs per cargo fmt (split long format! lines). - Replace break-on-unprobed-candidate with conservative hidden outcome + page_truncated flag, so already-visible pins later in the batch are still surfaced and the over-fetch test passes. --- crates/gitlawb-node/src/api/ipfs.rs | 36 ++++++++++++----------------- crates/gl/src/ipfs_cmd.rs | 13 ++++++++--- 2 files changed, 25 insertions(+), 24 deletions(-) diff --git a/crates/gitlawb-node/src/api/ipfs.rs b/crates/gitlawb-node/src/api/ipfs.rs index d8fc5a9..b5031ce 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -461,19 +461,9 @@ pub async fn list_pins(State(state): State) -> Result= MAX_PROBES { - // Record the earliest unprobed index and stop. - for &(idx, _) in candidates { - if idx < probe_limit { - probe_limit = idx; - } - } break; } let rp = allowed_blobs_by_repo @@ -512,13 +502,9 @@ pub async fn list_pins(State(state): State) -> Result= walk_limit_idx.min(probe_limit) { - // Past the MAX_WALKS wall or past an unprobed structural - // candidate — remaining pins belong to repos that never got - // walked or probed; handled by next request. - if i < probe_limit && page_truncated == false { - page_truncated = true; - } + if i >= walk_limit_idx { + // Past the MAX_WALKS wall — remaining pins belong to repos + // that never got walked; handled by next request. break; } @@ -547,12 +533,20 @@ pub async fn list_pins(State(state): State) -> Result v, None => { - // Structural candidate — consult cache. - structural_cache + // Structural candidate — consult cache. If the + // candidate was never probed (MAX_PROBES exhausted) + // the cache won't have it; treat it conservatively + // as non-structural (hidden) and flag truncation + // so the caller can continue with a follow-up + // request. + let cached = structural_cache .get(&repo.id) .and_then(|c| c.get(&pin.sha256_hex)) - .copied() - .unwrap_or(false) + .copied(); + if cached.is_none() { + page_truncated = true; + } + cached.unwrap_or(false) } }; if visible { diff --git a/crates/gl/src/ipfs_cmd.rs b/crates/gl/src/ipfs_cmd.rs index 6ca16c9..29d358b 100644 --- a/crates/gl/src/ipfs_cmd.rs +++ b/crates/gl/src/ipfs_cmd.rs @@ -148,10 +148,17 @@ async fn list_pins_paginated(client: &NodeClient) -> Result<(Vec, bool)> } let body = resp.bytes().await.context("failed to read pins response")?; if body.len() > MAX_RESPONSE_BYTES { - anyhow::bail!("pins response body ({len} bytes) exceeds {MAX_RESPONSE_BYTES} limit", len = body.len()); + anyhow::bail!( + "pins response body ({len} bytes) exceeds {MAX_RESPONSE_BYTES} limit", + len = body.len() + ); } - let resp: Value = serde_json::from_slice(&body) - .with_context(|| format!("failed to parse pins response ({len} bytes)", len = body.len()))?; + let resp: Value = serde_json::from_slice(&body).with_context(|| { + format!( + "failed to parse pins response ({len} bytes)", + len = body.len() + ) + })?; let pins = resp["pins"].as_array().cloned().unwrap_or_default(); From dee90b00ecfcccd6c66116a34756a30e67fd8e40 Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 10 Jul 2026 13:20:14 +0600 Subject: [PATCH 4/8] fix: restore Unavailable for transport/rejection errors in node status panel Only the 400 + had_truncated path should recover gracefully; transport errors and non-success rejection responses still return Unavailable. --- crates/gl/src/node.rs | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/crates/gl/src/node.rs b/crates/gl/src/node.rs index a822f2e..444f4c1 100644 --- a/crates/gl/src/node.rs +++ b/crates/gl/src/node.rs @@ -279,10 +279,7 @@ async fn fetch_pins(node: &str, auth: PinsAuth) -> PinsPanel { let resp = match client.get_signed(&path).await { Ok(r) => r, - Err(_) => { - incomplete = true; - break; - } + Err(_) => return PinsPanel::Unavailable, }; if !resp.status().is_success() { if resp.status().as_u16() == 400 && had_truncated { @@ -293,8 +290,7 @@ async fn fetch_pins(node: &str, auth: PinsAuth) -> PinsPanel { incomplete = true; continue; } - incomplete = true; - break; + return PinsPanel::Unavailable; } let body = match resp.bytes().await { Ok(b) => b, From b7ba1d0b239f2e2e14007fe819cdc7f0d0c68d92 Mon Sep 17 00:00:00 2001 From: Gravirei Date: Sun, 12 Jul 2026 12:37:30 +0600 Subject: [PATCH 5/8] =?UTF-8?q?fix:=20address=20review=20findings=20?= =?UTF-8?q?=E2=80=94=20migration,=20pin=20listing,=20client=20hardening?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 fixes: - Record_pinned_cid: DO UPDATE repo/owner_did instead of DO NOTHING so shared SHAs appear in scoped listings for each pinning repo - Record_pinata_cid: same DO UPDATE coverage + wrapper for callers without repo context P2 fixes: - MAX_WALKS: stop before the first unprocessed pin (walk_limit_idx = i) instead of marking it hidden and advancing past it - MAX_PROBES: track probe_limit and stop Phase 3 before the first unprobed structural candidate; uses walk_limit_idx.min(probe_limit) - Rebuild list_pins with proper auth, pagination, and missing helpers (derive_cursor_key, create_opaque_cursor, decode_opaque_cursor, batch_object_types) - Reject large responses via Content-Length before buffering - Only retry on 400 when server explicitly says 'expired truncated_cursor' --- Cargo.lock | 14 +- crates/gitlawb-node/Cargo.toml | 3 + crates/gitlawb-node/src/api/ipfs.rs | 1092 ++++++++++++++++++++++++++- crates/gitlawb-node/src/db/mod.rs | 96 ++- crates/gl/src/ipfs_cmd.rs | 27 +- crates/gl/src/node.rs | 27 +- 6 files changed, 1167 insertions(+), 92 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d12daa4..93880cb 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", @@ -3368,6 +3368,7 @@ dependencies = [ "axum", "base64", "bytes", + "chacha20poly1305", "chrono", "cid", "clap", @@ -3376,6 +3377,7 @@ dependencies = [ "futures", "gitlawb-core", "hex", + "hkdf", "hmac", "http-body-util", "libc", @@ -3390,6 +3392,7 @@ dependencies = [ "mockito", "multiaddr", "prometheus", + "rand 0.8.6", "reqwest", "serde", "serde_json", @@ -3412,7 +3415,7 @@ dependencies = [ [[package]] name = "gl" -version = "0.5.0" +version = "0.5.1" dependencies = [ "alloy", "anyhow", @@ -3824,6 +3827,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "thiserror 2.0.18", "tracing", ] diff --git a/crates/gitlawb-node/Cargo.toml b/crates/gitlawb-node/Cargo.toml index 9e42c08..66f8ec7 100644 --- a/crates/gitlawb-node/Cargo.toml +++ b/crates/gitlawb-node/Cargo.toml @@ -52,6 +52,9 @@ libp2p-kad = "0.48" libp2p-quic = { version = "0.13", features = ["tokio"] } libp2p-swarm = { version = "0.47", features = ["macros", "tokio"] } multiaddr = "0.18" +chacha20poly1305 = "0.10" +hkdf = "0.12" +rand = { workspace = true } futures = "0.3" aws-sdk-s3 = { version = "1", default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] } aws-config = { version = "1", features = ["behavior-version-latest"] } diff --git a/crates/gitlawb-node/src/api/ipfs.rs b/crates/gitlawb-node/src/api/ipfs.rs index b5031ce..f5d800b 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -15,14 +15,25 @@ //! see `get_by_cid`). use axum::{ - extract::{Path, State}, + extract::{Path, Query, State}, http::{HeaderMap, HeaderName, HeaderValue, StatusCode}, response::{IntoResponse, Response}, Extension, Json, }; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::Engine; +use chacha20poly1305::{aead::{Aead, AeadCore, KeyInit}, XChaCha20Poly1305, XNonce}; +use rand::rngs::OsRng; use cid::CidGeneric; +use hkdf::Hkdf; +use serde::Deserialize; +use sha2::Sha256; use std::collections::{HashMap, HashSet}; +use std::io::Write; +use std::path::PathBuf; +use std::process::{Command, Stdio}; use std::str::FromStr; +use std::time::{SystemTime, UNIX_EPOCH}; use crate::auth::AuthenticatedDid; use crate::error::{AppError, Result}; @@ -211,15 +222,204 @@ pub async fn get_by_cid( ))) } +/// Query parameters for `GET /api/v1/ipfs/pins`. +#[derive(Debug, Deserialize, Clone)] +pub struct ListPinsQuery { + #[serde(default = "default_limit")] + pub limit: i64, + pub cursor: Option, + pub truncated_cursor: Option, +} + +fn default_limit() -> i64 { + 50 +} + +/// Derive a dedicated 32-byte cursor cipher key from the node's Ed25519 seed +/// using HKDF with a domain-separated info string. This decouples cursor +/// confidentiality from the write-signing identity and avoids feeding the raw +/// seed into an unrelated primitive. +fn derive_cursor_key(seed: &[u8; 32]) -> [u8; 32] { + let hk = Hkdf::::new(None, seed.as_slice()); + let mut okm = [0u8; 32]; + hk.expand(b"gitlawb-ipfs-cursor-v1", &mut okm) + .expect("32 bytes is a valid HKDF output length"); + okm +} + +/// Create an opaque, self-contained truncated cursor token using +/// XChaCha20Poly1305 AEAD. +/// +/// Format: `base64_url_no_pad(nonce_24 || ciphertext)` where `ciphertext` = +/// XChaCha20Poly1305-encrypt(expiry_be_8 || cursor_string) with the 16-byte +/// AEAD tag appended by the encryptor. The caller cannot decode hidden-row +/// metadata without the server's Ed25519 seed. Tokens are durable (survive +/// restart, cross-node routing, retries) and expire after 600 seconds. +fn create_opaque_cursor(seed: &[u8; 32], cursor: &str) -> String { + let cursor_key = derive_cursor_key(seed); + let cipher = XChaCha20Poly1305::new_from_slice(&cursor_key) + .expect("32-byte key is valid for XChaCha20Poly1305"); + + let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng); + + let expiry = (SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + + 600) + .to_be_bytes(); + + let mut plaintext = Vec::with_capacity(8 + cursor.len()); + plaintext.extend_from_slice(&expiry); + plaintext.extend_from_slice(cursor.as_bytes()); + + let ciphertext = cipher + .encrypt(&nonce, plaintext.as_ref()) + .expect("AEAD encrypt should never fail"); + + let mut token = Vec::with_capacity(24 + ciphertext.len()); + token.extend_from_slice(nonce.as_ref()); + token.extend_from_slice(&ciphertext); + + URL_SAFE_NO_PAD.encode(&token) +} + +/// Decode and verify an opaque truncated cursor token. +/// Returns the original cursor string if valid and not expired. +fn decode_opaque_cursor(seed: &[u8; 32], token: &str) -> Option<(String, String, String)> { + let cursor_key = derive_cursor_key(seed); + let cipher = XChaCha20Poly1305::new_from_slice(&cursor_key) + .expect("32-byte key is valid for XChaCha20Poly1305"); + + let data = URL_SAFE_NO_PAD.decode(token.as_bytes()).ok()?; + if data.len() < 24 + 1 { + return None; + } + + let (nonce_bytes, ciphertext) = data.split_at(24); + let nonce = XNonce::from_slice(nonce_bytes); + + let plaintext = cipher.decrypt(nonce, ciphertext).ok()?; + if plaintext.len() < 8 { + return None; + } + + let expiry = u64::from_be_bytes(plaintext[..8].try_into().ok()?); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + if now >= expiry { + return None; + } + + let cursor = std::str::from_utf8(&plaintext[8..]).ok()?; + + let parts: Vec<&str> = cursor.splitn(3, '|').collect(); + if parts.len() == 3 { + Some(( + parts[0].to_string(), + parts[1].to_string(), + parts[2].to_string(), + )) + } else { + None + } +} + +/// Batch-check git object types for many SHAs in a single repo, using one +/// `git cat-file --batch-check` subprocess instead of N individual `cat-file -t` +/// calls. Returns a map from SHA → `Some("blob"|"commit"|"tree"|"tag")` or +/// `None` (missing/dangling). +/// +/// Must be called from a blocking context (e.g. `tokio::task::spawn_blocking`) +/// since it spawns a child process and reads its output synchronously. +fn batch_object_types( + repo_path: &std::path::Path, + shas: &[String], +) -> Result>> { + use anyhow::Context; + + let mut child = Command::new("git") + .args(["cat-file", "--batch-check"]) + .current_dir(repo_path) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .context("failed to spawn git cat-file --batch-check")?; + + { + let stdin = child.stdin.as_mut().context("stdin not captured")?; + for sha in shas { + writeln!(stdin, "{sha}").context("failed to write sha to cat-file stdin")?; + } + } + + let output = child + .wait_with_output() + .context("git cat-file --batch-check failed")?; + let stdout = String::from_utf8_lossy(&output.stdout); + + let mut results = HashMap::with_capacity(shas.len()); + for line in stdout.lines() { + let parts: Vec<&str> = line.splitn(3, ' ').collect(); + if parts.len() < 2 { + continue; + } + let sha = parts[0].to_string(); + match parts[1] { + "missing" => { + results.insert(sha, None); + } + obj_type => { + results.insert(sha, Some(obj_type.to_string())); + } + } + } + Ok(results) +} + /// GET /api/v1/ipfs/pins /// /// Returns all CIDs that have been pinned to the local IPFS node from git /// objects received via push. Each entry includes the git SHA-256 hex, the /// CIDv1 string, and the timestamp when it was pinned. -pub async fn list_pins(State(state): State) -> Result> { - let pins = state +/// +/// Requires authentication: the global pin index would otherwise disclose +/// metadata for every object ever pushed here (#121). +/// +/// The global listing filters each pinned object on current repo visibility +/// to prevent metadata disclosure when repos are made private after push (#136). +/// Only pins from repos the caller can currently read are returned. +pub async fn list_pins( + State(state): State, + Query(query): Query, + auth: Option>, +) -> Result> { + let caller = auth.as_ref().map(|e| e.0 .0.as_str()); + + // Reject anonymous callers: the pin index spans the entire node and would + // expose metadata for every object ever pushed here (#121). + if caller.is_none() { + return Err(AppError::Unauthorized( + "authentication required for pin listing".into(), + )); + } + let caller_owned = caller.map(|c| c.to_string()); + + // Build the set of readable repo slugs and owner DIDs from the deduped repo view + // (mirror rows already collapsed, quarantined excluded), then query + // pins bounded in SQL. + let repos = state .db - .list_pinned_cids() + .list_all_repos_deduped() + .await + .map_err(AppError::Internal)?; + let repo_ids: Vec = repos.iter().map(|r| r.id.clone()).collect(); + let rules_by_repo = state + .db + .list_visibility_rules_for_repos(&repo_ids) .await .map_err(AppError::Internal)?; @@ -332,12 +532,13 @@ pub async fn list_pins(State(state): State) -> Result = truncated_resume.or(initial_cursor); let mut response_cursor: Option<(String, String, String)> = None; - let mut allowed_blobs_by_repo: HashMap, HashSet, PathBuf)> = + let mut allowed_blobs_by_repo: HashMap, PathBuf)> = HashMap::new(); let mut page_truncated = false; // Per-repo cache of sha256_hex → is_structural (true for commit/tree/tag). let mut structural_cache: HashMap> = HashMap::new(); let mut probe_count = 0usize; + let mut probe_limit = std::usize::MAX; 'fetch: loop { if batch_count >= MAX_BATCHES { @@ -403,16 +604,14 @@ pub async fn list_pins(State(state): State) -> Result= MAX_WALKS { - // Walk budget exhausted. Mark this pin denied (we cannot - // determine visibility without the walk) and advance past - // it so the next request never revisits the boundary pin - // with a strict-< cursor predicate that would skip it. - pin_outcome.push(Some(false)); + // Walk budget exhausted. Stop before this pin and leave + // db_cursor at the last processed pin so the next request + // picks up here and retries the walk. page_truncated = true; - walk_limit_idx = i + 1; + walk_limit_idx = i; break; } - let (allowed, all_blobs, repo_path) = + let (allowed, repo_path) = match state.repo_store.acquire(&repo.owner_did, &repo.name).await { Ok(rp) => { let rp_clone = rp.clone(); @@ -437,20 +636,22 @@ pub async fn list_pins(State(state): State) -> Result (a, b, rp), - _ => (HashSet::new(), HashSet::new(), PathBuf::new()), + Ok(Ok(Ok(allowed))) => (allowed, rp), + _ => (HashSet::new(), PathBuf::new()), } } - Err(_) => (HashSet::new(), HashSet::new(), PathBuf::new()), + Err(_) => (HashSet::new(), PathBuf::new()), }; - allowed_blobs_by_repo.insert(repo.id.clone(), (allowed, all_blobs, repo_path)); + allowed_blobs_by_repo.insert(repo.id.clone(), (allowed, repo_path)); } - let (allowed, all_blobs, repo_path) = allowed_blobs_by_repo.get(&repo.id).unwrap(); + let (allowed, repo_path) = allowed_blobs_by_repo.get(&repo.id).unwrap(); if allowed.contains(&pin.sha256_hex) { pin_outcome.push(Some(true)); - } else if !all_blobs.contains(&pin.sha256_hex) && !repo_path.as_os_str().is_empty() { - // Candidate for structural-object check. + } else if !repo_path.as_os_str().is_empty() { + // Not in the allowed set — could be a withheld blob or a + // structural object (commit/tree/tag). Mark as structural + // candidate; Phase 2 will probe the type. pin_outcome.push(None); // deferred — decided after phase 2 structural_candidates .entry(repo.id.clone()) @@ -464,11 +665,19 @@ pub async fn list_pins(State(state): State) -> Result= MAX_PROBES { + // Probe budget exhausted. Track the first unprobed structural + // candidate index so Phase 3 stops before it and the next + // request picks up from the preceding safe cursor. + for &(idx, _) in candidates { + if idx < probe_limit { + probe_limit = idx; + } + } break; } let rp = allowed_blobs_by_repo .get(repo_id) - .map(|(_, _, p)| p.clone()) + .map(|(_, p)| p.clone()) .unwrap_or_default(); if rp.as_os_str().is_empty() { continue; @@ -502,9 +711,13 @@ pub async fn list_pins(State(state): State) -> Result= walk_limit_idx { - // Past the MAX_WALKS wall — remaining pins belong to repos - // that never got walked; handled by next request. + if i >= walk_limit_idx.min(probe_limit) { + // Past the MAX_WALKS wall or an unprobed structural candidate. + // Remaining pins are handled by the next request; db_cursor + // stays at the last processed pin so no row is skipped. + if i < probe_limit && !page_truncated { + page_truncated = true; + } break; } @@ -533,20 +746,15 @@ pub async fn list_pins(State(state): State) -> Result v, None => { - // Structural candidate — consult cache. If the + // Structural candidate — consult cache. If the // candidate was never probed (MAX_PROBES exhausted) - // the cache won't have it; treat it conservatively - // as non-structural (hidden) and flag truncation - // so the caller can continue with a follow-up - // request. - let cached = structural_cache + // this shouldn't be reached (probe_limit stops Phase 3 + // before unprobed rows), but handle it defensively. + structural_cache .get(&repo.id) .and_then(|c| c.get(&pin.sha256_hex)) - .copied(); - if cached.is_none() { - page_truncated = true; - } - cached.unwrap_or(false) + .copied() + .unwrap_or(false) } }; if visible { @@ -570,5 +778,819 @@ pub async fn list_pins(State(state): State) -> Result