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 f3de757..5797396 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -15,14 +15,30 @@ //! 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 cid::CidGeneric; +use hkdf::Hkdf; +use rand::rngs::OsRng; +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::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; use crate::auth::AuthenticatedDid; use crate::error::{AppError, Result}; @@ -144,12 +160,14 @@ pub async fn get_by_cid( let caller_for_walk = caller_owned.clone(); // Full-history walk shells out to git — keep it off the async runtime. let walk = tokio::task::spawn_blocking(move || { + let uncancelled = AtomicBool::new(false); allowed_blob_set_for_caller( &rp, &r, is_public, &owner, caller_for_walk.as_deref(), + &uncancelled, ) }) .await; @@ -211,20 +229,1407 @@ 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], + cancelled: &AtomicBool, +) -> 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")?; + } + } + + // Drop stdin so the child sees EOF on its input pipe. + drop(child.stdin.take()); + + // Poll for completion, checking the cancellation flag between iterations. + let output = loop { + if cancelled.load(Ordering::Relaxed) { + let _ = child.kill(); + let _ = child.wait(); + return Ok(HashMap::new()); + } + match child.try_wait() { + Ok(Some(_status)) => { + break child.wait_with_output().context( + "git cat-file --batch-check failed", + )?; + } + Ok(None) => { + std::thread::sleep(std::time::Duration::from_millis(50)); + continue; + } + Err(e) => { + return Err(AppError::Git(format!( + "git cat-file --batch-check wait failed: {e}", + ))); + } + } + }; + 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)?; + + // Build parallel vectors of readable (slug, owner_did) pairs to query in SQL. + // This avoids filter-before-limit leaks or loss of pages. + let mut query_repos = Vec::new(); + let mut query_owner_dids = Vec::new(); + + for r in &repos { + let rules = rules_by_repo.get(&r.id).map(Vec::as_slice).unwrap_or(&[]); + if visibility_check(rules, r.is_public, &r.owner_did, caller, "/") == Decision::Deny { + continue; + } + let short = crate::db::normalize_owner_key(&r.owner_did); + let slug = format!("{}/{}", short, r.name); + query_repos.push(slug); + query_owner_dids.push(r.owner_did.clone()); + } + + let max_visible = query.limit.clamp(0, 200); + + if max_visible == 0 { + return Ok(Json(serde_json::json!({ + "pins": [], + "count": 0, + }))); + } + + // Decode the optional keyset cursor from base64. + // Internal format: "pinned_at|repo|sha256_hex" (3-tuple) for normal + // pagination, or just "pinned_at" (1-tuple) for the truncated resume. + let decode_cursor = |s: &str| -> 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, 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 = usize::MAX; + + '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. 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; + break; + } + let (allowed, 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(); + let cancelled = Arc::new(AtomicBool::new(false)); + let cancelled_clone = Arc::clone(&cancelled); + + 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(), + &cancelled_clone, + ) + }); + match tokio::time::timeout(std::time::Duration::from_secs(60), walk_fut) + .await + { + Ok(Ok(Ok(allowed))) => (allowed, rp), + _ => { + cancelled.store(true, Ordering::Relaxed); + (HashSet::new(), PathBuf::new()) + } + } + } + Err(_) => (HashSet::new(), PathBuf::new()), + }; + allowed_blobs_by_repo.insert(repo.id.clone(), (allowed, repo_path)); + } + + 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 !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()) + .or_default() + .push((i, pin.sha256_hex.clone())); + } else { + pin_outcome.push(Some(false)); + } + } + + // ── Phase 2 — batch-check structural candidates per repo ────────── + for (repo_id, candidates) in &structural_candidates { + if probe_count >= 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()) + .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 cancelled_probe = Arc::new(AtomicBool::new(false)); + let cancelled_probe_clone = Arc::clone(&cancelled_probe); + let probe_fut = tokio::task::spawn_blocking(move || { + batch_object_types(&rp_for_block, &to_check, &cancelled_probe_clone) + }); + let results = + match tokio::time::timeout(std::time::Duration::from_secs(30), probe_fut).await { + Ok(Ok(Ok(map))) => map, + _ => { + cancelled_probe.store(true, Ordering::Relaxed); + 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 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; + } + + 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(); - Ok(Json(serde_json::json!({ + let visible = match pin_outcome[i] { + Some(v) => v, + None => { + // Structural candidate — consult cache. If the + // candidate was never probed (MAX_PROBES exhausted) + // 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() + .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(), - }))) + }); + + if page_filled { + // Page is full — provide a cursor from the last visible row so the + // caller can paginate. next_cursor is emitted regardless of whether + // MAX_BATCHES was also hit in this iteration. + if let Some((ref pa, ref r, ref sha)) = response_cursor { + body["next_cursor"] = serde_json::json!(encode_cursor(pa, r, sha)); + } + } else if page_truncated { + // Scan bound hit before filling the page. + body["truncated"] = serde_json::json!(true); + if let Some((ref pa, ref r, ref sha)) = response_cursor { + body["next_cursor"] = serde_json::json!(encode_cursor(pa, r, sha)); + } else if let Some((ref pa, ref r, ref sha)) = db_cursor { + let cursor_str = format!("{pa}|{r}|{sha}"); + let seed = state.node_keypair.to_seed(); + let token = create_opaque_cursor(&seed, &cursor_str); + body["truncated_cursor"] = serde_json::json!(token); + } + } + + Ok(Json(body)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::auth::AuthenticatedDid; + use crate::test_support::test_state; + use axum::extract::{Extension, Query, State}; + use sqlx::PgPool; + + #[sqlx::test] + async fn test_ipfs_cursor_guard(pool: PgPool) { + let app_state = test_state(pool.clone()).await; + + // Seed a path-scoped repo + 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-ipfs-test") + .bind("ipfstest") + .bind("did:key:z6Mkwowner") + .bind("desc") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/ipfstest") + .execute(app_state.db.pool()) + .await + .unwrap(); + + sqlx::query( + "INSERT INTO visibility_rules (id, repo_id, path_glob, mode, reader_dids, created_by, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7)", + ) + .bind("rule-1") + .bind("repo-ipfs-test") + .bind("/secret/**") + .bind("deny") + .bind("") + .bind("did:key:z6Mkwowner") + .bind("2026-07-03T00:00:00Z") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Seed another repo with NO path-scoped rules for visible pagination. + 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-ipfs-vis") + .bind("ipfsvis") + .bind("did:key:z6Mkwowner") + .bind("desc") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/ipfsvis") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Insert 1 visible pin, then a 2005-pin hidden stretch, then 1 visible pin. + // The hidden pins go in `ipfstest` (which has a deny rule and no physical repo so allowed_blobs is empty). + // The visible pins go in `ipfsvis` (which has no rules, so they are always visible). + // Note: three separate execute calls — sqlx prepared statements do not + // support multiple semicolon-delimited statements in a single query(). + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('vis-1-sha', 'vis-1-cid', '2026-07-03T10:00:00Z', 'z6Mkwowner/ipfsvis', 'did:key:z6Mkwowner')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + SELECT 'hid-sha-' || i, 'hid-cid-' || i, '2026-07-03T09:00:00Z', 'z6Mkwowner/ipfstest', 'did:key:z6Mkwowner' + FROM generate_series(1, 2005) as i", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('vis-2-sha', 'vis-2-cid', '2026-07-03T08:00:00Z', 'z6Mkwowner/ipfsvis', 'did:key:z6Mkwowner')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Visible pagination case asserting the cursor equals the last returned row + let auth = Extension(AuthenticatedDid("did:key:z6Mkcaller".to_string())); + let mut q = ListPinsQuery { + limit: 1, + cursor: None, + truncated_cursor: None, + }; + + let res1 = list_pins( + State(app_state.clone()), + Query(q.clone()), + Some(auth.clone()), + ) + .await + .unwrap() + .0; + let pins1 = res1["pins"].as_array().unwrap(); + assert_eq!(pins1.len(), 1); + assert_eq!(pins1[0]["sha256_hex"], "vis-1-sha"); + + let cursor1 = res1["next_cursor"].as_str().unwrap().to_string(); + // Decode to ensure it equals the last returned row + let bytes = URL_SAFE_NO_PAD.decode(cursor1.as_bytes()).unwrap(); + let decoded = String::from_utf8(bytes).unwrap(); + assert!(decoded.contains("vis-1-sha")); + + // Case 2: Follow the cursor. The next 2005 rows are hidden. + // It will hit MAX_BATCHES (2000 rows) and return a truncated_cursor + // whose XChaCha20Poly1305-encrypted payload conceals the hidden SHA. + q.cursor = Some(cursor1); + let res2 = list_pins( + State(app_state.clone()), + Query(q.clone()), + Some(auth.clone()), + ) + .await + .unwrap() + .0; + assert!(res2.get("pins").unwrap().as_array().unwrap().is_empty()); + assert_eq!(res2.get("truncated").unwrap().as_bool(), Some(true)); + assert!(res2.get("next_cursor").is_none()); + let truncated_cursor = res2["truncated_cursor"] + .as_str() + .expect("truncated_cursor should be present") + .to_string(); + + // Case 3: Resume with truncated_cursor. It should skip past the hidden + // batch and reach the older visible pin (vis-2-sha at 08:00:00Z). + q.cursor = None; + q.truncated_cursor = Some(truncated_cursor); + let res3 = list_pins( + State(app_state.clone()), + Query(q.clone()), + Some(auth.clone()), + ) + .await + .unwrap() + .0; + let pins3 = res3["pins"].as_array().unwrap(); + assert!( + !pins3.is_empty(), + "must surface vis-2-sha behind hidden window" + ); + assert_eq!(pins3[0]["sha256_hex"], "vis-2-sha"); + } + + #[test] + fn test_truncated_cursor_does_not_leak_hidden_sha() { + // The token is AEAD-encrypted with XChaCha20Poly1305: the hidden + // sha256_hex must NOT be recoverable by a caller who knows the + // pinned_at and repo prefix. Unlike a stream-cipher XOR construction + // (where known plaintext at offset i reveals keystream[i] via + // keystream[i] = ciphertext[i] XOR plaintext[i]), the AEAD ciphertext + // is ChaCha20 encryption with a per-nonce block counter applied to + // 16-byte blocks, then authenticated by Poly1305 — so XOR at a single + // offset does not yield a reusable keystream byte and the tag prevents + // any chosen-ciphertext oracle. + // + // This test demonstrates the unrecoverability property by attempting a + // known-plaintext attack against the ciphertext suffix. + let seed = [0xab; 32]; // arbitrary test seed + let pinned_at = "2026-07-03T09:00:00Z"; + let repo = "z6Mkwowner/ipfstest"; + let hidden_sha = "ab".repeat(32); // 64-char hex — well-known hidden SHA + + let cursor = format!("{pinned_at}|{repo}|{hidden_sha}"); + let token = create_opaque_cursor(&seed, &cursor); + + // Decode the raw token bytes — these are (nonce_24 || ciphertext). + let raw = URL_SAFE_NO_PAD.decode(token.as_bytes()).unwrap(); + let (_nonce, ciphertext) = raw.split_at(24); + + // Known plaintext: the first 19 chars of pinned_at "2026-07-03T09:00:00Z" + // plus "|z6Mkwowner/ipfstest|" = 38 bytes we know at the start. + // In the XOR-from-stream-cipher world, XOR of known plaintext with the + // ciphertext yields the keystream for those positions. If the keystream + // were reused at the sha suffix (modulo 32), XOR of known suffix with + // the recovered keystream would yield the hidden sha. + let known_prefix = format!("{pinned_at}|{repo}|"); + let known_bytes = known_prefix.as_bytes(); + + let attempted_keystream: Vec = known_bytes + .iter() + .zip(ciphertext.iter()) + .map(|(p, c)| p ^ c) + .collect(); + + // Use the "recovered keystream" at the same positions in the suffix + // (which would be valid only with a repeating XOR keystream). The + // suffix is the last 64 bytes of the ciphertext (hidden_sha length). + if ciphertext.len() >= known_bytes.len() + 64 { + let suffix_start = ciphertext.len() - 64; + let attempted_sha: String = ciphertext[suffix_start..] + .iter() + .zip(attempted_keystream.iter().cycle()) + .map(|(c, k)| (c ^ k) as char) + .collect(); + + // With a real AEAD the "recovered" suffix is garbage, not the sha. + assert_ne!( + attempted_sha, hidden_sha, + "XOR-based known-plaintext attack on AEAD must NOT recover the hidden sha" + ); + } + + // Substring check: the token bytes must not contain the sha256_hex in + // the clear. + let raw_str = std::str::from_utf8(&raw).unwrap_or(""); + assert!( + !raw_str.contains(&hidden_sha), + "truncated_cursor token MUST NOT contain hidden sha256_hex in the clear" + ); + + // Positive round-trip: correct seed decodes the full cursor. + let decoded = decode_opaque_cursor(&seed, &token).unwrap(); + assert_eq!(decoded.0, pinned_at); + assert_eq!(decoded.1, repo); + assert_eq!(decoded.2, hidden_sha); + + // Wrong key must not decode. + let wrong_seed = [0xcd; 32]; + assert!(decode_opaque_cursor(&wrong_seed, &token).is_none()); + } + + #[sqlx::test] + async fn test_max_walks_plaintext_not_in_response_cursor(pool: PgPool) { + let app_state = test_state(pool.clone()).await; + + // ── Create one visible repo (no path-scoped rules) ──────────────── + 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-walks-vis") + .bind("walksvis") + .bind("did:key:z6Mkwowner") + .bind("visible") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/walksvis") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // ── Seed > MAX_WALKS (50) path-scoped repos with hidden pins ───── + let num_wall_repos = 55usize; + for i in 0..num_wall_repos { + let repo_id = format!("repo-wall-{i}"); + let repo_name = format!("wall{i}"); + let disk_path = format!("/srv/{repo_name}"); + 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_id) + .bind(&repo_name) + .bind("did:key:z6Mkwowner") + .bind("desc") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind(&disk_path) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Add a /secret/** deny rule so the repo is path-scoped. + sqlx::query( + "INSERT INTO visibility_rules (id, repo_id, path_glob, mode, reader_dids, created_by, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7)", + ) + .bind(format!("rule-wall-{i}")) + .bind(&repo_id) + .bind("/secret/**") + .bind("deny") + .bind("") + .bind("did:key:z6Mkwowner") + .bind("2026-07-03T00:00:00Z") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // One hidden pin per wall repo. + let sha = format!("wallsha{i:04}"); + let slug = format!("z6Mkwowner/{repo_name}"); + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(&sha) + .bind(format!("cid-wall-{i}")) + .bind("2026-07-03T09:00:00Z") + .bind(&slug) + .bind("did:key:z6Mkwowner") + .execute(app_state.db.pool()) + .await + .unwrap(); + } + + // ── One visible pin (newest timestamp so it appears first) ──────── + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('vis-walks-sha', 'vis-walks-cid', '2026-07-03T10:00:00Z', 'z6Mkwowner/walksvis', 'did:key:z6Mkwowner')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + + let auth = Extension(AuthenticatedDid("did:key:z6Mkstranger".to_string())); + let res = list_pins( + State(app_state.clone()), + Query(ListPinsQuery { + limit: 50, + cursor: None, + truncated_cursor: None, + }), + Some(auth), + ) + .await + .unwrap() + .0; + + // The visible pin (newest) must be returned. + let pins = res["pins"].as_array().unwrap(); + assert_eq!(pins.len(), 1, "must return the visible pin"); + assert_eq!(pins[0]["sha256_hex"], "vis-walks-sha"); + + // The page is truncated (not filled) because MAX_WALKS was hit. + assert_eq!(res.get("truncated").and_then(|v| v.as_bool()), Some(true)); + // next_cursor IS present — it points to the VISIBLE pin shown to the + // caller (no leak). When response_cursor holds a visible pin the + // plaintext cursor is safe; the P1 leak only happened when skip_pos + // (an un-walked hidden pin) was put in response_cursor. + let nc = res["next_cursor"] + .as_str() + .expect("next_cursor must be present for visible pin pagination"); + let bytes = URL_SAFE_NO_PAD.decode(nc.as_bytes()).unwrap(); + let decoded = String::from_utf8(bytes).unwrap(); + assert!( + decoded.contains("vis-walks-sha"), + "next_cursor must reference the visible pin, not a hidden SHA: {decoded}" + ); + // No truncated_cursor — next_cursor handles pagination. + assert!( + res.get("truncated_cursor").is_none(), + "truncated_cursor must NOT be present when next_cursor suffices" + ); + + // ── Second request: skip past the visible pin into the hidden wall ── + // The response must use the AEAD token (no plaintext next_cursor) + // because no visible pin is in the returned batch. + let auth = Extension(AuthenticatedDid("did:key:z6Mkstranger".to_string())); + let res2 = list_pins( + State(app_state.clone()), + Query(ListPinsQuery { + limit: 50, + cursor: Some(nc.to_string()), + truncated_cursor: None, + }), + Some(auth), + ) + .await + .unwrap() + .0; + + let pins2 = res2["pins"].as_array().unwrap(); + assert!(pins2.is_empty(), "second page has no visible pins"); + assert_eq!(res2.get("truncated").and_then(|v| v.as_bool()), Some(true)); + // next_cursor must NOT be present — no visible pin in this batch. + assert!( + res2.get("next_cursor").is_none(), + "next_cursor must not be present when no visible pin is returned" + ); + // truncated_cursor MUST be present and AEAD-encrypted. + let token = res2["truncated_cursor"] + .as_str() + .expect("truncated_cursor must be present for hidden-only page"); + for i in 0..num_wall_repos { + let sha = format!("wallsha{i:04}"); + assert!( + !token.contains(&sha), + "truncated_cursor must not contain hidden sha256_hex in the clear: {sha}" + ); + } + } + + #[sqlx::test] + async fn test_structural_pin_included_withheld_blob_excluded(pool: PgPool) { + let app_state = test_state(pool.clone()).await; + + // ── Create a real on-disk bare repo with objects ────────────────── + let owner_did = "did:key:z6Mkwowner"; + let repo_name = "structest"; + let owner_slug = owner_did.replace([':', '/'], "_"); + let repo_path = std::path::PathBuf::from("/tmp") + .join(&owner_slug) + .join(format!("{repo_name}.git")); + + // Remove leftovers from a prior failed run, then init a bare repo. + let _ = std::fs::remove_dir_all(&repo_path); + crate::git::store::init_bare(&repo_path).unwrap(); + + // Create a blob: echo -n "secret content" | git hash-object -w --stdin + let mut blob_child = Command::new("git") + .args([ + "-C", + repo_path.to_str().unwrap(), + "hash-object", + "-w", + "--stdin", + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + blob_child + .stdin + .as_mut() + .unwrap() + .write_all(b"secret content") + .unwrap(); + // Drop stdin to close it so hash-object can finish. + drop(blob_child.stdin.take()); + let blob_output = blob_child.wait_with_output().unwrap(); + assert!( + blob_output.status.success(), + "git hash-object failed: {}", + String::from_utf8_lossy(&blob_output.stderr) + ); + let blob_sha = String::from_utf8_lossy(&blob_output.stdout) + .trim() + .to_string(); + assert!(!blob_sha.is_empty(), "blob sha must not be empty"); + + // Create a sub-tree for "secret/" containing the blob at "file.txt" + let sub_tree_input = format!("100644 blob {blob_sha}\tfile.txt"); + let mut sub_tree_child = Command::new("git") + .args(["-C", repo_path.to_str().unwrap(), "mktree"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + sub_tree_child + .stdin + .as_mut() + .unwrap() + .write_all(sub_tree_input.as_bytes()) + .unwrap(); + drop(sub_tree_child.stdin.take()); + let sub_tree_output = sub_tree_child.wait_with_output().unwrap(); + assert!( + sub_tree_output.status.success(), + "git mktree for secret/ failed: {}", + String::from_utf8_lossy(&sub_tree_output.stderr) + ); + let sub_tree_sha = String::from_utf8_lossy(&sub_tree_output.stdout) + .trim() + .to_string(); + assert!(!sub_tree_sha.is_empty(), "sub-tree sha must not be empty"); + + // Create the root tree containing the secret/ sub-tree at path "secret" + let root_tree_input = format!("040000 tree {sub_tree_sha}\tsecret"); + let mut root_tree_child = Command::new("git") + .args(["-C", repo_path.to_str().unwrap(), "mktree"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + root_tree_child + .stdin + .as_mut() + .unwrap() + .write_all(root_tree_input.as_bytes()) + .unwrap(); + drop(root_tree_child.stdin.take()); + let root_tree_output = root_tree_child.wait_with_output().unwrap(); + assert!( + root_tree_output.status.success(), + "git mktree for root tree failed: {}", + String::from_utf8_lossy(&root_tree_output.stderr) + ); + let tree_sha = String::from_utf8_lossy(&root_tree_output.stdout) + .trim() + .to_string(); + assert!(!tree_sha.is_empty(), "root tree sha must not be empty"); + + // Create a commit pointing to the tree + let commit_output = Command::new("git") + .args([ + "-C", + repo_path.to_str().unwrap(), + "commit-tree", + &tree_sha, + "-m", + "initial", + ]) + .env("GIT_AUTHOR_NAME", "test") + .env("GIT_AUTHOR_EMAIL", "test@test.com") + .env("GIT_COMMITTER_NAME", "test") + .env("GIT_COMMITTER_EMAIL", "test@test.com") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .unwrap(); + assert!( + commit_output.status.success(), + "git commit-tree failed: {}", + String::from_utf8_lossy(&commit_output.stderr) + ); + let commit_sha = String::from_utf8_lossy(&commit_output.stdout) + .trim() + .to_string(); + assert!(!commit_sha.is_empty(), "commit sha must not be empty"); + + // Update HEAD so the blob walk can reach the blob. + // In a bare repo HEAD is a symref to refs/heads/main, so we update the ref. + let update_output = Command::new("git") + .args([ + "-C", + repo_path.to_str().unwrap(), + "update-ref", + "refs/heads/main", + &commit_sha, + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .unwrap(); + assert!( + update_output.status.success(), + "git update-ref failed: {}", + String::from_utf8_lossy(&update_output.stderr) + ); + + // ── Seed the DB ─────────────────────────────────────────────────── + // Slug must match what list_pins computes from normalize_owner_key: + // normalize_owner_key("did:key:z6Mkwowner") = "z6Mkwowner" + // slug = "z6Mkwowner/structest" + let repo_slug = format!("z6Mkwowner/{repo_name}"); + 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-structest") + .bind(repo_name) + .bind(owner_did) + .bind("structural test repo") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind(repo_path.to_str().unwrap()) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Add a /secret/** deny rule so the blob is withheld from strangers. + sqlx::query( + "INSERT INTO visibility_rules (id, repo_id, path_glob, mode, reader_dids, created_by, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7)" + ) + .bind("rule-structest") + .bind("repo-structest") + .bind("/secret/**") + .bind("deny") + .bind("") + .bind("did:key:z6Mkwowner") + .bind("2026-07-03T00:00:00Z") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Pin the blob (must be withheld under /secret/**). + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(&blob_sha) + .bind("blob-cid") + .bind("2026-07-03T12:00:00Z") + .bind(&repo_slug) + .bind(owner_did) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Pin the tree (structural — must be visible). + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(&tree_sha) + .bind("tree-cid") + .bind("2026-07-03T11:00:00Z") + .bind(&repo_slug) + .bind(owner_did) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Pin the commit (structural — must be visible). + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(&commit_sha) + .bind("commit-cid") + .bind("2026-07-03T10:00:00Z") + .bind(&repo_slug) + .bind(owner_did) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // ── Call list_pins as a stranger ────────────────────────────────── + let stranger = Extension(AuthenticatedDid("did:key:z6Mkstranger".to_string())); + let res = list_pins( + State(app_state.clone()), + Query(ListPinsQuery { + limit: 50, + cursor: None, + truncated_cursor: None, + }), + Some(stranger), + ) + .await + .unwrap() + .0; + + let pins = res["pins"].as_array().unwrap(); + let sha_hexes: Vec<&str> = pins + .iter() + .filter_map(|p| p["sha256_hex"].as_str()) + .collect(); + + // The withheld blob at /secret/** must NOT appear. + assert!( + !sha_hexes.contains(&blob_sha.as_str()), + "withheld blob pin under /secret/** must NOT appear for stranger" + ); + // The tree and commit are structural objects not in the blob set — + // they MUST appear (KTD3). + assert!( + sha_hexes.contains(&tree_sha.as_str()), + "structural tree pin must appear for stranger" + ); + assert!( + sha_hexes.contains(&commit_sha.as_str()), + "structural commit pin must appear for stranger" + ); + + // Clean up the on-disk repo. + let _ = std::fs::remove_dir_all(&repo_path); + } + + #[sqlx::test] + async fn test_stranger_denied_private_repo_pins(pool: PgPool) { + let app_state = test_state(pool.clone()).await; + + // Seed a fully private repo (is_public = false). + 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-private") + .bind("privaterepo") + .bind("did:key:z6Mkwowner") + .bind("private repo") + .bind(false) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/privaterepo") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Insert a pin owned by the owner. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('priv-sha-1', 'priv-cid-1', '2026-07-03T12:00:00Z', 'z6Mkwowner/privaterepo', 'did:key:z6Mkwowner')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // A stranger (not the owner, not a listed reader) must see no pins. + let stranger_auth = Extension(AuthenticatedDid("did:key:z6Mkstranger".to_string())); + let res = list_pins( + State(app_state.clone()), + Query(ListPinsQuery { + limit: 50, + cursor: None, + truncated_cursor: None, + }), + Some(stranger_auth), + ) + .await + .unwrap() + .0; + assert_eq!( + res["pins"].as_array().unwrap().len(), + 0, + "stranger must not see pins from a private repo" + ); + assert_eq!(res["count"].as_u64().unwrap(), 0); + } + + #[sqlx::test] + async fn test_orphan_empty_repo_pins_excluded(pool: PgPool) { + let app_state = test_state(pool.clone()).await; + + // Seed a public repo (so the caller has some readable repo context). + 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-public") + .bind("pubrepo") + .bind("did:key:z6Mkwowner") + .bind("public repo") + .bind(true) + .bind("main") + .bind("2026-07-03T00:00:00Z") + .bind("2026-07-03T00:00:00Z") + .bind("/srv/pubrepo") + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Insert a legit pin for the public repo. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('legit-sha', 'legit-cid', '2026-07-03T12:00:00Z', 'z6Mkwowner/pubrepo', 'did:key:z6Mkwowner')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Insert a legacy orphan pin with repo = '' (empty string) and owner_did = ''. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ('orphan-sha', 'orphan-cid', '2026-07-03T11:00:00Z', '', '')", + ) + .execute(app_state.db.pool()) + .await + .unwrap(); + + // Signed caller must see the legit pin but NOT the orphan. + let auth = Extension(AuthenticatedDid("did:key:z6Mkcaller".to_string())); + let res = list_pins( + State(app_state.clone()), + Query(ListPinsQuery { + limit: 50, + cursor: None, + truncated_cursor: None, + }), + Some(auth), + ) + .await + .unwrap() + .0; + let pins = res["pins"].as_array().unwrap(); + let sha_hexes: Vec<&str> = pins + .iter() + .filter_map(|p| p["sha256_hex"].as_str()) + .collect(); + assert!(sha_hexes.contains(&"legit-sha"), "legit pin must appear"); + assert!( + !sha_hexes.contains(&"orphan-sha"), + "orphan pin with repo='' must NOT appear" + ); + } } diff --git a/crates/gitlawb-node/src/api/repos.rs b/crates/gitlawb-node/src/api/repos.rs index b9cbc35..6bccc42 100644 --- a/crates/gitlawb-node/src/api/repos.rs +++ b/crates/gitlawb-node/src/api/repos.rs @@ -3,6 +3,7 @@ use axum::http::StatusCode; use axum::response::Response; use axum::Json; use bytes::Bytes; +use std::sync::atomic::AtomicBool; use std::sync::Arc; use crate::auth::{caller_authorized_to_push, AuthenticatedDid}; @@ -65,8 +66,9 @@ async fn replication_withheld_set( Some(rules) => { let owner_did = owner_did.to_string(); tokio::task::spawn_blocking(move || { + let uncancelled = AtomicBool::new(false); crate::git::visibility_pack::withheld_blob_oids( - &disk_path, &rules, is_public, &owner_did, None, + &disk_path, &rules, is_public, &owner_did, None, &uncancelled, ) }) .await @@ -108,8 +110,9 @@ async fn fail_closed_full_scan_objects( candidates: Vec, ) -> Vec { tokio::task::spawn_blocking(move || -> anyhow::Result> { + let uncancelled = AtomicBool::new(false); let allowed = crate::git::visibility_pack::replicable_blob_set( - &disk_path, &rules, is_public, &owner_did, + &disk_path, &rules, is_public, &owner_did, &uncancelled, )?; let all_blobs = crate::git::push_delta::all_blob_oids(&disk_path)?; Ok(crate::git::visibility_pack::replicable_objects_fail_closed( @@ -659,12 +662,14 @@ pub async fn git_upload_pack( let caller_owned = caller.map(str::to_string); let is_public = record.is_public; tokio::task::spawn_blocking(move || { + let uncancelled = AtomicBool::new(false); visibility_pack::withheld_blob_oids( &path, &rules, is_public, &owner_did, caller_owned.as_deref(), + &uncancelled, ) }) .await @@ -1110,18 +1115,22 @@ pub async fn git_receive_pack( let rules_for_enc = rules_opt.clone(); let repo_id = record.id.clone(); let owner_did = record.owner_did.clone(); + let owner_short = crate::db::normalize_owner_key(&record.owner_did).to_string(); let is_public = record.is_public; let irys_url = state.config.irys_url.clone(); let http_client = std::sync::Arc::clone(&state.http_client); let node_did_str = state.node_did.to_string(); let node_seed = state.node_keypair.to_seed(); let repo_name = record.name.clone(); + let repo_slug = format!("{owner_short}/{repo_name}"); tokio::spawn(async move { let pinned = crate::ipfs_pin::pin_new_objects( &ipfs_api, &repo_path_clone, object_list_ipfs, &db_clone, + &repo_slug, + &owner_did, ) .await; if !pinned.is_empty() { @@ -1141,8 +1150,9 @@ pub async fn git_receive_pack( let p = repo_path_clone.clone(); let owner = owner_did.clone(); let recip = tokio::task::spawn_blocking(move || { + let uncancelled = AtomicBool::new(false); crate::git::visibility_pack::withheld_blob_recipients( - &p, &rules, is_public, &owner, + &p, &rules, is_public, &owner, &uncancelled, ) }) .await; @@ -1209,6 +1219,7 @@ pub async fn git_receive_pack( crate::db::normalize_owner_key(&record.owner_did), record.name ); + let owner_did_for_pinata = record.owner_did.clone(); let ref_updates_clone = ref_updates .iter() .map(|u| (u.ref_name.clone(), u.old_sha.clone(), u.new_sha.clone())) @@ -1232,6 +1243,8 @@ pub async fn git_receive_pack( &repo_path_clone, object_list_pinata, &db_clone, + &repo_slug, + &owner_did_for_pinata, ) .await } else { diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 5324a4b..68d5d40 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -160,6 +160,8 @@ pub struct PinnedCidRecord { pub cid: String, pub pinned_at: String, pub pinata_cid: Option, + pub repo: String, + pub owner_did: String, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -853,6 +855,25 @@ 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. 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: 10, name: "ref_cert_unique_per_ref", @@ -872,6 +893,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 ───────────────────────────────────────────────────────────────────── @@ -2166,15 +2250,51 @@ impl Db { Ok(row.get::("cnt") > 0) } - pub async fn record_pinned_cid(&self, sha256_hex: &str, cid: &str) -> Result<()> { + #[allow(dead_code)] + 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"))) + } + + /// Record a pinned CID with explicit repo/owner_did association. + /// 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. + pub async fn record_pinned_cid_full( + &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. + // + // During phase 1 the old PK on sha256_hex means a shared Git object + // (same SHA used in multiple repos) can live in only one physical row. + // DO UPDATE ensures each repo's writer can still record its (repo, + // owner_did) association so the scoped listing query finds it. The + // DO UPDATE with COALESCE so an empty-context write (caller without + // repo/owner_did) never overwrites a populated association from the + // v11 migration backfill or a prior repo-context write. sqlx::query( - "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) - VALUES ($1, $2, $3) - ON CONFLICT(sha256_hex) DO NOTHING", + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, repo, owner_did) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT(sha256_hex) DO UPDATE SET + repo = COALESCE(NULLIF(EXCLUDED.repo, ''), pinned_cids.repo), + owner_did = COALESCE(NULLIF(EXCLUDED.owner_did, ''), pinned_cids.owner_did)", ) .bind(sha256_hex) .bind(cid) .bind(Utc::now().to_rfc3339()) + .bind(repo) + .bind(owner_did) .execute(&self.pool) .await?; Ok(()) @@ -2250,9 +2370,10 @@ impl Db { Ok(row.map(|r| r.get("recipients_tag"))) } + #[allow(dead_code)] pub async fn list_pinned_cids(&self) -> Result> { let rows = sqlx::query( - "SELECT sha256_hex, cid, pinned_at, pinata_cid FROM pinned_cids ORDER BY pinned_at DESC", + "SELECT sha256_hex, cid, pinned_at, pinata_cid, repo, owner_did FROM pinned_cids ORDER BY pinned_at DESC", ) .fetch_all(&self.pool) .await?; @@ -2263,6 +2384,80 @@ impl Db { cid: r.get("cid"), pinned_at: r.get("pinned_at"), pinata_cid: r.get("pinata_cid"), + repo: r.get("repo"), + owner_did: r.get("owner_did"), + }) + .collect()) + } + + #[allow(dead_code)] + 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"))) + } + + /// 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 { + sha256_hex: r.get("sha256_hex"), + cid: r.get("cid"), + pinned_at: r.get("pinned_at"), + pinata_cid: r.get("pinata_cid"), + repo: r.get("repo"), + owner_did: r.get("owner_did"), }) .collect()) } @@ -2279,18 +2474,27 @@ impl Db { } /// Record the Pinata CID for a git object. - /// Inserts the row if it doesn't exist (objects pinned directly to Pinata - /// without a prior local IPFS pin get cid = pinata_cid). - pub async fn record_pinata_cid(&self, sha256_hex: &str, pinata_cid: &str) -> Result<()> { + /// Record the Pinata CID with explicit repo/owner_did association. + pub async fn record_pinata_cid_full( + &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) - VALUES ($1, $2, $3, $4) - ON CONFLICT(sha256_hex) DO UPDATE SET pinata_cid = EXCLUDED.pinata_cid", + "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, + repo = COALESCE(NULLIF(EXCLUDED.repo, ''), pinned_cids.repo), + owner_did = COALESCE(NULLIF(EXCLUDED.owner_did, ''), pinned_cids.owner_did)", ) .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(()) @@ -3328,7 +3532,8 @@ impl Db { #[cfg(test)] mod migration_tests { - use super::{MIGRATIONS, MIGRATION_V1_NAME}; + use super::{Db, MIGRATIONS, MIGRATION_V1_NAME}; + use sqlx::{PgPool, Row}; #[test] fn migrations_are_non_empty() { @@ -3406,6 +3611,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)] diff --git a/crates/gitlawb-node/src/git/visibility_pack.rs b/crates/gitlawb-node/src/git/visibility_pack.rs index cb70e39..fa318bd 100644 --- a/crates/gitlawb-node/src/git/visibility_pack.rs +++ b/crates/gitlawb-node/src/git/visibility_pack.rs @@ -9,6 +9,7 @@ use crate::visibility::{visibility_check, Decision}; use anyhow::{Context, Result}; use std::collections::{BTreeSet, HashMap, HashSet}; use std::path::Path; +use std::sync::atomic::{AtomicBool, Ordering}; /// Fail closed unless every ref ultimately resolves to a commit (a ref pointing /// directly at a blob or tree, or an annotated tag — even a nested one — of such @@ -147,7 +148,7 @@ fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { /// Fails closed: if commit enumeration or any tree walk fails, returns an error so /// the caller aborts the serve/pin rather than producing a partial (under-withheld) /// set. -fn blob_paths(repo_path: &Path) -> Result> { +fn blob_paths(repo_path: &Path, cancelled: &AtomicBool) -> Result> { assert_all_refs_are_commits(repo_path)?; // Enumerate every reachable commit, not just ref tips. `--all` walks all refs; @@ -173,6 +174,9 @@ fn blob_paths(repo_path: &Path) -> Result> { let commits_stdout = String::from_utf8_lossy(&commits.stdout); let mut out: HashSet<(String, String)> = HashSet::new(); for commit in commits_stdout.lines() { + if cancelled.load(Ordering::Relaxed) { + break; + } let commit = commit.trim(); if commit.is_empty() { continue; @@ -235,8 +239,9 @@ pub fn withheld_blob_oids( is_public: bool, owner_did: &str, caller: Option<&str>, + cancelled: &AtomicBool, ) -> Result> { - let pairs = blob_paths(repo_path)?; + let pairs = blob_paths(repo_path, cancelled)?; Ok(withheld_from_pairs( &pairs, rules, is_public, owner_did, caller, )) @@ -309,8 +314,9 @@ pub fn replicable_blob_set( rules: &[VisibilityRule], is_public: bool, owner_did: &str, + cancelled: &AtomicBool, ) -> Result> { - allowed_blob_set_for_caller(repo_path, rules, is_public, owner_did, None) + allowed_blob_set_for_caller(repo_path, rules, is_public, owner_did, None, cancelled) } /// Reachable blob OIDs that visibility ALLOWS `caller` at some path. The @@ -331,8 +337,9 @@ pub fn allowed_blob_set_for_caller( is_public: bool, owner_did: &str, caller: Option<&str>, + cancelled: &AtomicBool, ) -> Result> { - let pairs = blob_paths(repo_path)?; + let pairs = blob_paths(repo_path, cancelled)?; let mut allowed = HashSet::new(); for (oid, path) in &pairs { if visibility_check(rules, is_public, owner_did, caller, path) == Decision::Allow { @@ -370,9 +377,10 @@ pub fn withheld_blob_recipients( rules: &[VisibilityRule], is_public: bool, owner_did: &str, + cancelled: &AtomicBool, ) -> Result>> { // One history walk feeds both the withheld set and the recipient mapping. - let pairs = blob_paths(repo_path)?; + let pairs = blob_paths(repo_path, cancelled)?; let withheld = withheld_from_pairs(&pairs, rules, is_public, owner_did, None); if withheld.is_empty() { return Ok(HashMap::new()); @@ -473,7 +481,7 @@ mod tests { let (_td, bare, secret_oid, public_oid) = fixture(); let rules = [rule("/secret/**", &[])]; // caller = None models the public / any peer: what must not replicate. - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "secret blob must be withheld" @@ -491,7 +499,7 @@ mod tests { let (_td, bare, secret, public) = fixture(); let rules = [rule("/secret/**", &["did:key:zFriend"])]; let withheld = - withheld_blob_oids(&bare, &rules, true, OWNER, Some("did:key:zStranger")).unwrap(); + withheld_blob_oids(&bare, &rules, true, OWNER, Some("did:key:zStranger"), &AtomicBool::new(false)).unwrap(); assert!(withheld.contains(&secret), "secret blob must be withheld"); assert!( !withheld.contains(&public), @@ -503,7 +511,7 @@ mod tests { fn owner_withholds_nothing() { let (_td, bare, secret, public) = fixture(); let rules = [rule("/secret/**", &["did:key:zFriend"])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, Some(OWNER)).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, Some(OWNER), &AtomicBool::new(false)).unwrap(); assert!(withheld.is_empty(), "owner sees everything"); let _ = (secret, public); } @@ -513,14 +521,14 @@ mod tests { let (_td, bare, _secret, _public) = fixture(); let rules = [rule("/secret/**", &["did:key:zFriend"])]; let withheld = - withheld_blob_oids(&bare, &rules, true, OWNER, Some("did:key:zFriend")).unwrap(); + withheld_blob_oids(&bare, &rules, true, OWNER, Some("did:key:zFriend"), &AtomicBool::new(false)).unwrap(); assert!(withheld.is_empty(), "listed reader sees the subtree"); } #[test] fn no_subtree_rules_withholds_nothing() { let (_td, bare, _secret, _public) = fixture(); - let withheld = withheld_blob_oids(&bare, &[], true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &[], true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.is_empty(), "public repo, no rules, nothing withheld" @@ -579,7 +587,7 @@ mod tests { ]; for (rules, caller) in cases { assert!(!has_path_scoped_rule(&rules)); - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, caller).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, caller, &AtomicBool::new(false)).unwrap(); assert!( withheld.is_empty(), "no path-scoped rule must withhold nothing for a gate-passing caller (caller={caller:?})" @@ -603,7 +611,7 @@ mod tests { // the walk is empty and the served set is complete. let root_only = vec![rule("/", &["did:key:zReader"])]; assert!(!has_path_scoped_rule(&root_only)); - let withheld_a = withheld_blob_oids(&bare, &root_only, true, OWNER, reader).unwrap(); + let withheld_a = withheld_blob_oids(&bare, &root_only, true, OWNER, reader, &AtomicBool::new(false)).unwrap(); assert!( withheld_a.is_empty(), "root-only rules withhold nothing for a gate-passing reader; the skip is safe" @@ -622,7 +630,7 @@ mod tests { rule("/secret/**", &["did:key:zOther"]), ]; assert!(has_path_scoped_rule(&scoped)); - let withheld_b = withheld_blob_oids(&bare, &scoped, true, OWNER, reader).unwrap(); + let withheld_b = withheld_blob_oids(&bare, &scoped, true, OWNER, reader, &AtomicBool::new(false)).unwrap(); let served_b = replicable_objects(all, &withheld_b); assert!( !served_b.contains(&secret), @@ -636,7 +644,7 @@ mod tests { // Branch C — same path-scoped rules, but the caller is the owner. The // owner bypasses every rule, so the walk withholds nothing and the full // pack (secret included) is served even though a path-scoped rule exists. - let withheld_c = withheld_blob_oids(&bare, &scoped, true, OWNER, Some(OWNER)).unwrap(); + let withheld_c = withheld_blob_oids(&bare, &scoped, true, OWNER, Some(OWNER), &AtomicBool::new(false)).unwrap(); assert!( withheld_c.is_empty(), "the owner bypasses path-scoped rules and is served everything" @@ -742,7 +750,7 @@ mod tests { ); let rules: Vec = vec![]; - let allowed = replicable_blob_set(&work, &rules, true, OWNER).unwrap(); + let allowed = replicable_blob_set(&work, &rules, true, OWNER, &AtomicBool::new(false)).unwrap(); assert!( !allowed.contains(&dangling_oid), "dangling blob is unreachable, so never in the allowed set" @@ -823,7 +831,7 @@ mod tests { // Every gate-relevant caller: anonymous, listed reader, owner. None of // them can put the dangling blob in the allowed set — it has no path. for caller in [None, Some(reader), Some(OWNER)] { - let allowed = allowed_blob_set_for_caller(&work, &rules, true, OWNER, caller).unwrap(); + let allowed = allowed_blob_set_for_caller(&work, &rules, true, OWNER, caller, &AtomicBool::new(false)).unwrap(); assert!( !allowed.contains(&dangling_oid), "dangling blob must be absent from allowed-set (caller={caller:?})" @@ -842,7 +850,7 @@ mod tests { let (_td, repo, secret_oid, public_oid) = fixture(); let reader = "did:key:zReader"; let rules = vec![rule("/secret/**", &[reader])]; - let map = withheld_blob_recipients(&repo, &rules, true, OWNER).unwrap(); + let map = withheld_blob_recipients(&repo, &rules, true, OWNER, &AtomicBool::new(false)).unwrap(); let recips = map.get(&secret_oid).expect("secret blob has recipients"); assert!(recips.contains(OWNER)); @@ -895,7 +903,7 @@ mod tests { run(&["update-ref", "-d", &head_ref]); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "blob reachable only via refs/custom/* must still be withheld" @@ -940,7 +948,7 @@ mod tests { run(&["update-ref", "-d", &head_ref]); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "blob reachable only via detached HEAD must still be withheld" @@ -1007,7 +1015,7 @@ mod tests { ); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "secret blob deleted at the tip but reachable in history must be withheld" @@ -1064,7 +1072,7 @@ mod tests { ); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "secret blob at a non-ASCII path must be withheld" @@ -1141,7 +1149,7 @@ mod tests { ); let rules = [rule(nfc_rule, &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "NFC-authored deny rule must withhold the secret blob under the NFD-named directory" @@ -1217,7 +1225,7 @@ mod tests { ); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "secret blob at a path with TAB/newline must be withheld" @@ -1304,7 +1312,7 @@ mod tests { ); let rules = [rule("/s\u{fffd}cret/**", &[])]; - let result = withheld_blob_oids(&bare, &rules, true, OWNER, None); + let result = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)); assert!( result.is_err(), "a non-UTF-8 path must fail closed (Err), not be lossy-decoded and leaked" @@ -1319,7 +1327,7 @@ mod tests { // ref and under-withholding. std::fs::write(bare.join("refs/heads/blobref"), format!("{secret}\n")).unwrap(); let rules = [rule("/secret/**", &[])]; - let result = withheld_blob_oids(&bare, &rules, true, OWNER, None); + let result = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)); assert!( result.is_err(), "a ref that cannot be traversed must fail closed (Err)" @@ -1350,7 +1358,7 @@ mod tests { run(&["tag", "-a", "-m", "outer", "v2", "v1"]); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( withheld.contains(&secret_oid), "secret blob must still be withheld with annotated and nested tags present" @@ -1378,7 +1386,7 @@ mod tests { run(&["tag", "-a", "-m", "blobtag", "blobtag", &secret]); let rules = [rule("/secret/**", &[])]; - let result = withheld_blob_oids(&bare, &rules, true, OWNER, None); + let result = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)); assert!( result.is_err(), "an annotated tag of a blob must fail closed (Err)" @@ -1397,7 +1405,7 @@ mod tests { ) .unwrap(); let rules = [rule("/secret/**", &[])]; - let result = withheld_blob_oids(&bare, &rules, true, OWNER, None); + let result = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)); assert!( result.is_err(), "a ref pointing at a missing object must fail closed (Err)" @@ -1425,7 +1433,7 @@ mod tests { let (tx, rx) = std::sync::mpsc::channel(); std::thread::spawn(move || { let rules = [rule("/secret/**", &[])]; - let is_err = withheld_blob_oids(&bare, &rules, true, OWNER, None).is_err(); + let is_err = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).is_err(); let _ = tx.send(is_err); }); match rx.recv_timeout(std::time::Duration::from_secs(10)) { @@ -1487,7 +1495,7 @@ mod tests { ); let rules = [rule("/secret/**", &[])]; - let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None).unwrap(); + let withheld = withheld_blob_oids(&bare, &rules, true, OWNER, None, &AtomicBool::new(false)).unwrap(); assert!( !withheld.contains(&shared_oid), "a blob also reachable via an allowed path must not be withheld" diff --git a/crates/gitlawb-node/src/ipfs_pin.rs b/crates/gitlawb-node/src/ipfs_pin.rs index 3b34619..30af2e4 100644 --- a/crates/gitlawb-node/src/ipfs_pin.rs +++ b/crates/gitlawb-node/src/ipfs_pin.rs @@ -93,12 +93,18 @@ pub async fn cat(ipfs_api: &str, cid: &str) -> Result> { /// object's bytes. The twin in `pinata.rs` mirrors this shape — change both in /// lockstep. /// +/// `repo_slug` and `owner_did` are recorded alongside each pin so the scoped +/// listing query (`list_pinned_cids_for_repos`) can find them. Pass empty +/// strings if not available (the scoped listing will omit such pins). +/// /// Returns a list of `(sha256_hex, cid)` pairs for objects pinned this call. pub async fn pin_new_objects( ipfs_api: &str, repo_path: &std::path::Path, object_list: Vec, db: &crate::db::Db, + repo_slug: &str, + owner_did: &str, ) -> Vec<(String, String)> { if ipfs_api.is_empty() { return vec![]; @@ -130,7 +136,7 @@ pub async fn pin_new_objects( // Pin to IPFS match pin_git_object(ipfs_api, &sha, &data).await { Ok(cid) if !cid.is_empty() => { - if let Err(e) = db.record_pinned_cid(&sha, &cid).await { + if let Err(e) = db.record_pinned_cid_full(&sha, &cid, repo_slug, owner_did).await { tracing::warn!(sha = %sha, err = %e, "failed to record pinned CID in DB"); } pinned.push((sha, cid)); diff --git a/crates/gitlawb-node/src/pinata.rs b/crates/gitlawb-node/src/pinata.rs index 6c9c0bf..824d5b8 100644 --- a/crates/gitlawb-node/src/pinata.rs +++ b/crates/gitlawb-node/src/pinata.rs @@ -76,6 +76,9 @@ pub async fn pin_object( /// this shape — change both in lockstep. Objects already recorded with a /// `pinata_cid` are skipped. Returns `(sha_hex, cid)` pairs for each newly /// pinned object. +/// +/// `repo_slug` and `owner_did` are recorded alongside each pin so the scoped +/// listing query can find them. Pass empty strings if not available. pub async fn pin_new_objects( client: &reqwest::Client, upload_url: &str, @@ -83,6 +86,8 @@ pub async fn pin_new_objects( repo_path: &std::path::Path, object_list: Vec, db: &crate::db::Db, + repo_slug: &str, + owner_did: &str, ) -> Vec<(String, String)> { if jwt.is_empty() { return vec![]; @@ -111,7 +116,7 @@ pub async fn pin_new_objects( match pin_object(client, upload_url, jwt, &sha, &data).await { Ok(cid) if !cid.is_empty() => { - if let Err(e) = db.record_pinata_cid(&sha, &cid).await { + if let Err(e) = db.record_pinata_cid_full(&sha, &cid, repo_slug, owner_did).await { tracing::warn!(sha = %sha, err = %e, "failed to record pinata_cid in DB"); } pinned.push((sha, cid)); diff --git a/crates/gitlawb-node/src/server.rs b/crates/gitlawb-node/src/server.rs index f4c0d3e..b811fd7 100644 --- a/crates/gitlawb-node/src/server.rs +++ b/crates/gitlawb-node/src/server.rs @@ -210,14 +210,14 @@ pub fn build_router(state: AppState) -> Router { .layer(axum::Extension(push_limiter)); // ── IPFS content-addressed retrieval and pin listing ────────────────── - // `/ipfs/{cid}` carries `optional_signature` so `get_by_cid` sees the caller - // identity and can apply per-repo visibility (#110); anonymous callers stay - // anonymous and still read genuinely public content. `/api/v1/ipfs/pins` - // stays unsigned — gating the pin index is tracked separately (#121). + // Both `/ipfs/{cid}` and `/api/v1/ipfs/pins` carry `optional_signature` + // so handlers see the caller identity and can apply per-repo visibility. + // Anonymous callers stay anonymous for public content; the pin listing + // requires authentication (handled in the handler itself). let ipfs_routes = Router::new() .route("/ipfs/{cid}", get(ipfs::get_by_cid)) - .layer(middleware::from_fn(auth::optional_signature)) - .merge(Router::new().route("/api/v1/ipfs/pins", get(ipfs::list_pins))); + .merge(Router::new().route("/api/v1/ipfs/pins", get(ipfs::list_pins))) + .layer(middleware::from_fn(auth::optional_signature)); // ── Arweave permanent anchors ────────────────────────────────────────── let arweave_routes = Router::new().route("/api/v1/arweave/anchors", get(arweave::list_anchors)); diff --git a/crates/gl/src/ipfs_cmd.rs b/crates/gl/src/ipfs_cmd.rs index d4cb64f..3729749 100644 --- a/crates/gl/src/ipfs_cmd.rs +++ b/crates/gl/src/ipfs_cmd.rs @@ -7,6 +7,7 @@ use std::path::PathBuf; use anyhow::{Context, Result}; use clap::{Args, Subcommand}; +use reqwest::header::CONTENT_LENGTH; use serde_json::Value; use crate::http::NodeClient; @@ -86,6 +87,132 @@ 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). +#[expect(dead_code)] +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?; + + // Check Content-Length before buffering the body. + if let Some(cl) = resp + .headers() + .get(CONTENT_LENGTH) + .and_then(|v| v.to_str().ok().and_then(|s| s.parse::().ok())) + { + if cl > MAX_RESPONSE_BYTES { + anyhow::bail!( + "pins response Content-Length ({cl} bytes) exceeds {MAX_RESPONSE_BYTES} limit" + ); + } + } + + if !resp.status().is_success() { + let status = resp.status(); + if status == 400 && had_truncated { + let body = resp.text().await.unwrap_or_default(); + // Only treat a 400 as expired-cursor when the server explicitly + // says so. Any other 400 — malformed cursor, protocol change, + // node bug — is surfaced as an error. + if body.contains("invalid or expired truncated_cursor") { + cursor = last_next_cursor.clone(); + continue; + } + anyhow::bail!("node returned 400 for pins listing: {body}"); + } + 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..a54b736 100644 --- a/crates/gl/src/node.rs +++ b/crates/gl/src/node.rs @@ -3,6 +3,7 @@ use anyhow::Result; use clap::{Args, Subcommand}; use gitlawb_core::identity::Keypair; +use reqwest::header::CONTENT_LENGTH; use serde_json::Value; use std::path::PathBuf; @@ -241,19 +242,107 @@ 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 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 { + 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(_) => return PinsPanel::Unavailable, + }; + + // Check Content-Length before buffering the body. + if let Some(cl) = resp + .headers() + .get(CONTENT_LENGTH) + .and_then(|v| v.to_str().ok().and_then(|s| s.parse::().ok())) + { + if cl > MAX_RESPONSE_BYTES { + return PinsPanel::Unavailable; + } + } + + if !resp.status().is_success() { + if resp.status().as_u16() == 400 && had_truncated { + let body = match resp.text().await { + Ok(b) => b, + Err(_) => return PinsPanel::Unavailable, + }; + if body.contains("invalid or expired truncated_cursor") { + cursor = last_next_cursor.clone(); + continue; + } + } + return PinsPanel::Unavailable; + } + 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 { + 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) { + 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 {