diff --git a/.github/workflows/pr-checks.yml b/.github/workflows/pr-checks.yml index d6d9497a..7e8811cf 100644 --- a/.github/workflows/pr-checks.yml +++ b/.github/workflows/pr-checks.yml @@ -92,6 +92,16 @@ jobs: DATABASE_URL: postgres://postgres:postgres@localhost:5432/gitlawb_test run: cargo test --workspace + # The real-node deny harness is behind the `test-harness` feature (so its + # spawn surface never compiles into the production binary), which means + # `--workspace` above skips it. Run it explicitly so the trust-boundary + # regression cases (INV-1/INV-2/INV-8) execute on every PR. Uses the same + # Postgres service; `#[sqlx::test]` provisions an isolated DB per test. + - name: cargo test (real-node deny harness) + env: + DATABASE_URL: postgres://postgres:postgres@localhost:5432/gitlawb_test + run: cargo test -p gitlawb-node --features test-harness --test deny_harness + build-release: name: build --release runs-on: ubuntu-latest diff --git a/crates/gitlawb-node/Cargo.toml b/crates/gitlawb-node/Cargo.toml index 9e42c084..1dc0fb7c 100644 --- a/crates/gitlawb-node/Cargo.toml +++ b/crates/gitlawb-node/Cargo.toml @@ -6,10 +6,28 @@ edition.workspace = true rust-version.workspace = true license.workspace = true +[lib] +name = "gitlawb_node" +path = "src/lib.rs" + [[bin]] name = "gitlawb-node" path = "src/main.rs" +# Exposes the `test_harness` spawn surface (src/test_harness.rs) to the +# real-node deny-harness integration crate. Off by default so the production +# binary never compiles test-only boot code. Enable with +# `cargo test -p gitlawb-node --features test-harness`. +[features] +test-harness = [] + +# Only builds when `test-harness` is enabled; otherwise skipped (not an error), +# so `cargo test --workspace` without the feature stays green. +[[test]] +name = "deny_harness" +path = "tests/deny_harness.rs" +required-features = ["test-harness"] + [dependencies] gitlawb-core = { path = "../gitlawb-core" } ed25519-dalek = { workspace = true } @@ -77,3 +95,9 @@ libp2p-dns = { version = "0.44.0", features = ["tokio"] } [dev-dependencies] mockito = "1" tempfile = "3" +# Used by the deny-harness integration crate (tests/deny_harness.rs): the +# #[sqlx::test] macro for an ephemeral per-test pool, and reqwest as the real +# HTTP client that drives deny paths over the socket. +sqlx = { version = "0.8", features = ["postgres", "runtime-tokio-rustls", "macros", "migrate"] } +reqwest = { workspace = true } +hex = "0.4" diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 5324a4b9..364d81bd 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -252,7 +252,7 @@ impl Db { &self.pool } - #[cfg(test)] + #[cfg(any(test, feature = "test-harness"))] pub fn for_testing(pool: PgPool) -> Self { Self { pool } } @@ -261,7 +261,7 @@ impl Db { /// provisions an empty per-test database, so DB-backed tests must run this /// before seeding. Reuses the production `migrate()` path (the advisory lock /// is harmless on an isolated test DB and migrations are idempotent). - #[cfg(test)] + #[cfg(any(test, feature = "test-harness"))] pub(crate) async fn run_migrations(&self) -> Result<()> { self.migrate().await } diff --git a/crates/gitlawb-node/src/git/repo_store.rs b/crates/gitlawb-node/src/git/repo_store.rs index a5c367e9..73d42fa1 100644 --- a/crates/gitlawb-node/src/git/repo_store.rs +++ b/crates/gitlawb-node/src/git/repo_store.rs @@ -33,7 +33,7 @@ pub struct RepoStore { } impl RepoStore { - #[cfg(test)] + #[cfg(any(test, feature = "test-harness"))] pub fn for_testing(repos_dir: PathBuf, pool: PgPool) -> Self { Self { repos_dir, diff --git a/crates/gitlawb-node/src/lib.rs b/crates/gitlawb-node/src/lib.rs new file mode 100644 index 00000000..a9415f9a --- /dev/null +++ b/crates/gitlawb-node/src/lib.rs @@ -0,0 +1,1653 @@ +// Crate builds as both a library (`gitlawb_node`) and the `gitlawb-node` binary. +// The library exposes the boot surface (build_router, AppState, Config, +// migrations) so the out-of-crate deny-harness integration crate can spawn a +// real node; `src/main.rs` is a thin shim over `run()`. Modules the harness +// reaches are `pub`; the rest stay crate-private. +pub mod api; +mod arweave; +pub mod auth; +mod bootstrap; +mod cert; +pub mod config; +pub mod db; +mod encrypted_pin; +pub mod error; +pub mod git; +mod graphql; +mod icaptcha; +mod ipfs_pin; +mod metrics; +mod operator; +mod p2p; +mod pinata; +mod rate_limit; +pub mod server; +pub mod state; +mod sync; +#[cfg(feature = "test-harness")] +#[doc(hidden)] +pub mod test_harness; +#[cfg(test)] +mod test_support; +mod visibility; +mod webhooks; + +use anyhow::{anyhow, Context, Result}; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::IntoResponse; +use axum::routing::get; +use axum::{Json, Router}; +use clap::Parser; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use tokio::net::TcpListener; +use tokio::sync::watch; +use tracing::{info, warn}; + +use gitlawb_core::http_sig::sign_request; +use gitlawb_core::identity::Keypair; + +pub use config::Config; +use db::Db; +pub use state::AppState; + +#[derive(Clone)] +struct DegradedState { + node_did: String, + db_startup: Arc, +} + +/// Two independent counters with no cross-field invariant — atomics, not a +/// lock, so the retry loop and the degraded handlers never contend. +#[derive(Default)] +struct DbStartupStatus { + attempts: AtomicU64, + next_retry_secs: AtomicU64, +} + +/// Boot and run the node to completion. The `gitlawb-node` binary is a thin +/// `#[tokio::main]` shim over this; keeping the body here (not in `main.rs`) +/// lets the library own the full module tree so integration tests can link it. +pub async fn run() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::from_default_env() + .add_directive("gitlawb_node=debug".parse().unwrap()) + .add_directive("tower_http=info".parse().unwrap()), + ) + .init(); + + let mut config = Config::parse(); + + // Merge the embedded seed list of public network nodes into the runtime + // bootstrap peers. Operators can opt out via GITLAWB_BOOTSTRAP_DISABLE_SEEDS. + bootstrap::merge_seeds(&mut config); + + if !config.public_read { + warn!( + "GITLAWB_PUBLIC_READ=false is reserved; per-repository private-read enforcement is not wired in alpha" + ); + } + + // Load or generate the node's identity keypair + let keypair = load_or_create_keypair(&config)?; + let node_did = keypair.did(); + + // One-time metrics init. Must run before any handler that calls into + // `metrics::record_*` so the registry exists when the first event fires. + // Safe to call even when GITLAWB_METRICS_ADDR is unset — those helpers + // are simply no-ops until something reads from the registry. + metrics::init(env!("CARGO_PKG_VERSION"), &node_did.to_string()); + + info!("╔══════════════════════════════════════════╗"); + info!( + "║ gitlawb node v{} ║", + env!("CARGO_PKG_VERSION") + ); + info!("╚══════════════════════════════════════════╝"); + // Process-wide shutdown signal. One sender lives in AppState (cloned + // into every handler); main() keeps a clone and flips it on SIGINT + // or SIGTERM. Tasks that hold a watch::Receiver get notified at + // their next await point. + let (shutdown_tx, _shutdown_rx_for_main) = watch::channel(false); + spawn_shutdown_signal(shutdown_tx.clone()); + + info!(did = %node_did, "node identity"); + info!(addr = %config.bind_addr(), "binding HTTP listener"); + + // Bind HTTP once, before dependency initialization, and keep this socket + // for the life of the process. The degraded server accepts on a dup of the + // same socket, so the degraded→full handoff never closes the port: while + // the full server initializes, connections queue in the shared backlog + // instead of being refused. + let listener = TcpListener::bind(config.bind_addr()) + .await + .with_context(|| format!("failed to bind to {}", config.bind_addr()))?; + let full_std_listener = listener.into_std()?; + let degraded_listener = TcpListener::from_std( + full_std_listener + .try_clone() + .context("failed to clone HTTP listener for degraded server")?, + )?; + + // Metrics must stay observable during a database outage — the degraded + // window is exactly when dashboards need data — so this listener starts + // before the DB connects. + let metrics_handle = if !config.metrics_addr.is_empty() { + match spawn_metrics_server(&config.metrics_addr, shutdown_tx.subscribe()).await { + Ok(handle) => { + info!(addr = %config.metrics_addr, "metrics endpoint listening"); + Some(handle) + } + Err(e) => { + warn!(err = %e, addr = %config.metrics_addr, "failed to start metrics endpoint — continuing without"); + None + } + } + } else { + info!("metrics endpoint disabled (GITLAWB_METRICS_ADDR not set)"); + None + }; + + let db_startup = Arc::new(DbStartupStatus::default()); + let (db_ready_tx, db_ready_rx) = watch::channel(false); + let mut degraded_handle = tokio::spawn(run_degraded_server( + degraded_listener, + node_did.to_string(), + Arc::clone(&db_startup), + db_ready_rx, + shutdown_tx.subscribe(), + )); + + // Connect to PostgreSQL database. A transient outage or bad secret should + // not crash-loop the process and hammer the database provider; permanent + // misconfiguration surfaces through error-level logs and the /ready check. + let db = tokio::select! { + db = connect_db_with_retry(&config, Arc::clone(&db_startup), shutdown_tx.subscribe()) => { + match db { + Some(db) => db, + None => { + // Shutdown requested while waiting for the database. The + // degraded server only serves one-shot 503s — abort it + // rather than drain, so a slow client can't stall exit. + degraded_handle.abort(); + return Ok(()); + } + } + } + degraded = &mut degraded_handle => { + if *shutdown_tx.borrow() { + return Ok(()); + } + return match degraded { + Ok(Ok(())) => Err(anyhow!("degraded HTTP server stopped before database became ready")), + Ok(Err(err)) => Err(err.context("degraded HTTP server failed")), + Err(err) => Err(anyhow!("degraded HTTP server task failed: {err}")), + }; + } + }; + + // Flip the degraded server into graceful shutdown, but do NOT await the + // drain: one slow in-flight request must not delay the full server, and + // the shared socket means there is no port gap to cover. The drain + // finishes (and logs) in the background. + db_ready_tx.send(true).ok(); + tokio::spawn(async move { + match degraded_handle.await { + Ok(Ok(())) => {} + Ok(Err(err)) => warn!(err = %err, "degraded HTTP server exited with error"), + Err(err) => warn!(err = %err, "degraded HTTP server task failed"), + } + }); + info!(addr = %config.bind_addr(), "database ready; starting full HTTP server"); + + // Prune peer rows that point back at this node (stale self-loop entries) + if let Some(public_url) = config.public_url.as_deref() { + match db.prune_self_peers(public_url).await { + Ok(0) => {} + Ok(n) => info!(removed = n, public_url, "pruned self-loop peer rows"), + Err(e) => warn!(err = %e, "prune_self_peers failed (non-fatal)"), + } + } + + // Prune peer rows with non-public hosts (loopback/private/internal) that + // were injected via the unauthenticated announce route — they poison the + // sync-notify fan-out (SSRF + crowding out real peers). + match db.prune_non_public_peers().await { + Ok(0) => {} + Ok(n) => info!(removed = n, "pruned non-public (poisoned) peer rows"), + Err(e) => warn!(err = %e, "prune_non_public_peers failed (non-fatal)"), + } + + // Ensure repos directory exists + std::fs::create_dir_all(&config.repos_dir).context("failed to create repos directory")?; + + // Start libp2p swarm (if p2p_port > 0) + let p2p_handle = if config.p2p_port > 0 { + let bootstrap_addrs = config + .p2p_bootstrap + .iter() + .filter_map(|s| s.parse().ok()) + .collect(); + let shutdown_rx = shutdown_tx.subscribe(); + match p2p::start( + &node_did.to_string(), + config.p2p_port, + bootstrap_addrs, + Arc::clone(&db), + config.auto_sync, + shutdown_rx, + ) + .await + { + Ok(handle) => { + info!(port = config.p2p_port, peer_id = %handle.local_peer_id, "libp2p swarm started"); + Some(Arc::new(handle)) + } + Err(e) => { + tracing::warn!(err = %e, "failed to start libp2p swarm — continuing without p2p"); + None + } + } + } else { + info!("p2p disabled (p2p_port = 0)"); + None + }; + + // Shared no-redirect HTTP client. See build_http_client for the SSRF rationale. + let http_client = Arc::new(build_http_client()?); + + let (ref_update_tx, _) = tokio::sync::broadcast::channel::(256); + let (task_event_tx, _) = tokio::sync::broadcast::channel::(256); + + let graphql_schema = Arc::new(graphql::build_schema( + Arc::clone(&db), + ref_update_tx.clone(), + task_event_tx.clone(), + )); + + let machine_id = std::env::var("FLY_MACHINE_ID").ok(); + if let Some(ref mid) = machine_id { + info!(" fly machine: {mid}"); + } + + // Initialize Tigris S3 client if bucket is configured + let tigris = if !config.tigris_bucket.is_empty() { + match git::tigris::TigrisClient::new(&config.tigris_bucket).await { + Ok(client) => { + info!(bucket = %config.tigris_bucket, "tigris storage enabled"); + Some(client) + } + Err(e) => { + tracing::warn!(err = %e, "failed to initialize Tigris client — using local-only storage"); + None + } + } + } else { + info!("tigris storage disabled (no bucket configured)"); + None + }; + + let repo_store = + git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, db.pool().clone()); + + // Per-DID limiter for the creation endpoints. Keyed on the authenticated + // DID (attacker-varied), so bound its key set to cap memory. + let rate_limiter = + rate_limit::RateLimiter::new_bounded(10, std::time::Duration::from_secs(3600), 200_000); + + // Per-client-IP flood brake for the creation endpoints. The per-DID limiter + // above is bypassed by a DID farm (one throwaway did:key per repo), which is + // exactly how the recurring spam-repo floods get past both it and the + // iCaptcha gate. Keyed on the resolved client IP so a single-source flood is + // capped regardless of how many identities it mints. Sized well above any + // legitimate per-IP creation rate; GITLAWB_CREATE_RATE_LIMIT overrides, 0 + // disables. Bounded key set — the key is a client-influenced IP. + let create_limit = std::env::var("GITLAWB_CREATE_RATE_LIMIT") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(120); + let create_ip_rate_limiter = rate_limit::RateLimiter::new_bounded( + create_limit, + std::time::Duration::from_secs(3600), + 200_000, + ); + if create_limit == 0 { + tracing::warn!("GITLAWB_CREATE_RATE_LIMIT=0 — per-IP creation rate limiting disabled"); + } + + // Push-path flood brake: max git-receive-pack requests per client IP per + // hour (counts both the info/refs advertisement and the push POST). Sized + // for heavy agent automation while still stopping flood traffic (the June + // 2026 attack pushed several times per second per IP). GITLAWB_PUSH_RATE_LIMIT + // overrides; 0 disables. Bounded key set — the key is a client-influenced IP. + let push_limit = std::env::var("GITLAWB_PUSH_RATE_LIMIT") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(600); + let push_rate_limiter = rate_limit::RateLimiter::new_bounded( + push_limit, + std::time::Duration::from_secs(3600), + 200_000, + ); + if push_limit == 0 { + tracing::warn!("GITLAWB_PUSH_RATE_LIMIT=0 — per-IP push rate limiting disabled"); + } + + // Which forwarded header the edge is trusted to set. Default None (trust + // nothing, key on the socket peer). Fly nodes set GITLAWB_TRUSTED_PROXY=fly; + // a node behind Caddy/NGINX sets it to x-forwarded-for. + let push_limiter_trust = rate_limit::TrustedProxy::from_env_value( + &std::env::var("GITLAWB_TRUSTED_PROXY").unwrap_or_default(), + ); + tracing::info!(trust = ?push_limiter_trust, push_limit, "push rate limiter configured"); + + // Peer-sync flood brakes, keyed on the resolved client IP (per-DID is useless + // here — a did:key farm self-registers). Two buckets so an unsigned notify + // flood can't drain the signed trigger caller's quota (#82). Bounded key sets + // (the key is a client-influenced IP); 0 disables each. + let sync_trigger_rate_limiter = rate_limit::RateLimiter::new_bounded( + config.sync_trigger_rate_limit, + std::time::Duration::from_secs(3600), + 200_000, + ); + let peer_write_rate_limiter = rate_limit::RateLimiter::new_bounded( + config.peer_write_rate_limit, + std::time::Duration::from_secs(3600), + 200_000, + ); + if config.sync_trigger_rate_limit == 0 { + tracing::warn!( + "GITLAWB_SYNC_TRIGGER_RATE_LIMIT=0 — /sync/trigger IP rate limiting disabled" + ); + } + if config.peer_write_rate_limit == 0 { + tracing::warn!("GITLAWB_PEER_WRITE_RATE_LIMIT=0 — peer-write IP rate limiting disabled"); + } + + // Initialize the iCaptcha proof gate (inert unless ICAPTCHA_MODE is set). + icaptcha::init().await; + + let state = AppState { + config: Arc::new(config.clone()), + db, + node_did: node_did.clone(), + node_keypair: Arc::new(keypair), + p2p: p2p_handle, + http_client, + ref_update_tx, + task_event_tx, + graphql_schema, + machine_id, + repo_store, + rate_limiter, + create_ip_rate_limiter, + push_rate_limiter, + push_limiter_trust, + sync_trigger_rate_limiter, + peer_write_rate_limiter, + shutdown_tx: shutdown_tx.clone(), + }; + + // Periodic peer-count poll for the metrics gauge. If p2p is disabled + // we still set the gauge to 0 so dashboards don't show "no data". + { + let p2p_for_metrics = state.p2p.clone(); + let mut shutdown_rx = state.subscribe_shutdown(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(15)); + loop { + tokio::select! { + _ = interval.tick() => { + let count = match &p2p_for_metrics { + Some(h) => h.status().await.map(|s| s.connected_peers).unwrap_or(0), + None => 0, + }; + metrics::set_peers_connected(count as i64); + } + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + return; + } + } + } + } + }); + } + + // Periodic cleanup of expired rate limit entries + consumed-proof ledger + { + let rl = state.rate_limiter.clone(); + let create_ip_rl = state.create_ip_rate_limiter.clone(); + let push_rl = state.push_rate_limiter.clone(); + let sync_trigger_rl = state.sync_trigger_rate_limiter.clone(); + let peer_write_rl = state.peer_write_rate_limiter.clone(); + let db = state.db.clone(); + let mut shutdown_rx = state.subscribe_shutdown(); + tokio::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(300)) => { + rl.cleanup().await; + create_ip_rl.cleanup().await; + push_rl.cleanup().await; + sync_trigger_rl.cleanup().await; + peer_write_rl.cleanup().await; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + if let Err(e) = db.sweep_expired_proofs(now).await { + tracing::warn!(err = %e, "failed to sweep expired iCaptcha proofs"); + } + } + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + break; + } + } + } + } + }); + } + + let router = server::build_router(state.clone()); + // Re-register the socket bound at startup — same fd, so there was never a + // moment with the port closed between the degraded and full servers. + let listener = TcpListener::from_std(full_std_listener) + .context("failed to re-register HTTP listener with the runtime")?; + + info!("✓ node started — did:{}", node_did); + info!(" repos dir: {}", config.repos_dir.display()); + info!( + " database: PostgreSQL ({})", + &config.database_url.split('@').next_back().unwrap_or("?") + ); + + // Publish our DID record to the Kademlia DHT shortly after startup + if let Some(p2p) = &state.p2p { + let did_record = p2p::DidRecord { + did: node_did.to_string(), + http_url: config.public_url.clone().unwrap_or_default(), + peer_id: p2p.local_peer_id.to_string(), + p2p_port: config.p2p_port, + timestamp: chrono::Utc::now().to_rfc3339(), + }; + let p2p_clone = Arc::clone(p2p); + let mut shutdown_rx = state.subscribe_shutdown(); + tokio::spawn(async move { + // Small delay so Kademlia can find peers first + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {} + _ = shutdown_rx.changed() => return, + } + p2p_clone.put_did(did_record).await; + info!("DID record published to Kademlia DHT"); + }); + } + + // Spawn background gossip: announce to bootstrap peers, then ping known peers periodically + { + let gossip_state = state.clone(); + let bootstrap_peers = config.bootstrap_peers.clone(); + let shutdown_rx = state.subscribe_shutdown(); + tokio::spawn(async move { + gossip_task(gossip_state, bootstrap_peers, shutdown_rx).await; + }); + } + + // Start multi-node sync worker if auto_sync is enabled + if config.auto_sync { + sync::start( + Arc::clone(&state.db), + Arc::clone(&state.config), + Arc::clone(&state.node_keypair), + state.subscribe_shutdown(), + ); + info!("auto-sync worker started"); + } + + // On-chain operator setup: verify stake + spawn heartbeat loop + if !state.config.contract_node_staking.is_empty() + && !state.config.operator_private_key.is_empty() + { + match build_operator_client(&state.config, &state.node_did.to_string()) { + Ok(client) => match operator::startup_check(&client).await { + Ok(_) => { + let arc_client = Arc::new(client); + arc_client.spawn_heartbeat_loop(state.subscribe_shutdown()); + } + Err(e) => { + if state.config.operator_strict_mode { + return Err(e.context("strict-mode operator check failed")); + } + tracing::warn!(err = %e, "operator startup check failed — continuing without heartbeat loop"); + } + }, + Err(e) => { + if state.config.operator_strict_mode { + return Err(e.context("strict-mode: failed to build operator client")); + } + tracing::warn!(err = %e, "operator client could not be built — continuing without PoS"); + } + } + } else { + info!("on-chain PoS disabled (GITLAWB_CONTRACT_NODE_STAKING or GITLAWB_OPERATOR_PRIVATE_KEY unset)"); + } + + // axum's `with_graceful_shutdown` begins draining in-flight requests once + // the shutdown watch flips. That drain is otherwise unbounded, so we bound + // it by `grace`: the closure fires `armed_tx` the instant it observes the + // signal, and `drive_serve_with_grace` abandons the drain if it has not + // finished `grace` after that moment. The clock starts at the signal, not at + // server start, so total uptime is never bounded. + let shutdown_signal_for_axum = state.subscribe_shutdown(); + let grace = std::time::Duration::from_secs(config.shutdown_grace_secs); + info!(grace_secs = config.shutdown_grace_secs, "axum server ready"); + + let (armed_tx, armed_rx) = tokio::sync::oneshot::channel::<()>(); + + // `into_make_service_with_connect_info` exposes the socket peer address as + // `ConnectInfo` so the push limiter can key on the real client + // when no trusted proxy header applies (see `rate_limit::client_key`). + let serve = axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(async move { + let mut rx = shutdown_signal_for_axum; + // Wait until the watcher flips to true, then return so axum + // can begin draining. + while !*rx.borrow_and_update() { + if rx.changed().await.is_err() { + // Sender dropped — treat as shutdown. + break; + } + } + // Start the grace clock at the signal, right before axum drains. + let _ = armed_tx.send(()); + }); + + // Race the drain against the grace deadline (see `drive_serve_with_grace`): + // the clock starts when the closure above arms `armed_rx`, i.e. at the + // signal, and the drain is abandoned if it outlasts `grace`. + let (serve_result, grace_expired) = drive_serve_with_grace(serve, armed_rx, grace).await; + + if grace_expired { + warn!( + grace_secs = config.shutdown_grace_secs, + "shutdown grace expired; abandoning in-flight requests" + ); + } + + // Server has stopped accepting new connections and drained in-flight + // requests (or the grace deadline fired). Tear the rest of the system down. + info!("HTTP server stopped, beginning process shutdown"); + if let Some(h) = metrics_handle { + h.abort(); + } + serve_result?; + info!("clean exit"); + Ok(()) +} + +/// Drive the HTTP `serve` future to completion, bounding the post-signal drain +/// by `grace`. `armed` resolves the instant the shutdown signal fires (the serve +/// future's graceful-shutdown closure sends on it), so the grace clock starts at +/// the signal, not at server start — total uptime is never bounded. Returns the +/// serve result plus whether the deadline fired (which abandons in-flight +/// requests so teardown can proceed). +async fn drive_serve_with_grace( + serve: F, + armed: tokio::sync::oneshot::Receiver<()>, + grace: std::time::Duration, +) -> (std::io::Result<()>, bool) +where + F: std::future::IntoFuture>, +{ + let serve = serve.into_future(); + tokio::select! { + result = serve => (result, false), + _ = async { + // Park until the signal arms the clock, then bound the drain. + let _ = armed.await; + tokio::time::sleep(grace).await; + } => (Ok(()), true), + } +} + +fn spawn_shutdown_signal(tx: watch::Sender) { + tokio::spawn(async move { + #[cfg(unix)] + { + use tokio::signal::unix::{signal as unix_signal, SignalKind}; + let mut sigterm = + unix_signal(SignalKind::terminate()).expect("install SIGTERM handler"); + let mut sigint = unix_signal(SignalKind::interrupt()).expect("install SIGINT handler"); + tokio::select! { + _ = sigterm.recv() => info!("SIGTERM received, shutting down"), + _ = sigint.recv() => info!("SIGINT received, shutting down"), + } + } + #[cfg(not(unix))] + { + use tokio::signal; + let _ = signal::ctrl_c().await; + info!("Ctrl-C received, shutting down"); + } + tx.send(true).ok(); + }); +} + +async fn connect_db_with_retry( + config: &Config, + db_startup: Arc, + mut shutdown_rx: watch::Receiver, +) -> Option> { + let initial_retry_secs = config.db_retry_initial_secs; + let max_retry_secs = config.db_retry_max_secs.max(initial_retry_secs); + let acquire_timeout = std::time::Duration::from_secs(config.db_acquire_timeout_secs); + let attempt_timeout = std::time::Duration::from_secs(config.db_connect_timeout_secs); + let mut attempts = 0_u64; + + loop { + if *shutdown_rx.borrow() { + return None; + } + + attempts = attempts.saturating_add(1); + db_startup.attempts.store(attempts, Ordering::Relaxed); + + // Bound the whole attempt, not just the pool connect: migrations + // block on a cross-instance advisory lock, and an unbounded wait + // there would wedge this loop — no retries, no logs, no recovery. + // Timing out and retrying is safe; migrations are idempotent. + let attempt = match tokio::time::timeout( + attempt_timeout, + Db::connect( + &config.database_url, + config.db_max_connections, + acquire_timeout, + ), + ) + .await + { + Ok(result) => result, + Err(_) => Err(anyhow!( + "connect + migrate attempt exceeded {}s (GITLAWB_DB_CONNECT_TIMEOUT_SECS); \ + is another instance holding the migration lock?", + attempt_timeout.as_secs() + )), + }; + + match attempt { + Ok(db) => { + info!(attempts, "database connection established"); + return Some(Arc::new(db)); + } + Err(err) => { + // A bad DATABASE_URL or rejected credentials won't heal on + // their own. Still retry (exiting would crash-loop and hammer + // the provider — and take liveness down with it), but log at + // error level and skip straight to the maximum backoff; the + // /ready health check is what surfaces this to deploys. + let permanent = is_likely_permanent_db_error(&err); + let retry_secs = if permanent { + max_retry_secs + } else { + database_retry_delay_secs(initial_retry_secs, max_retry_secs, attempts) + }; + db_startup + .next_retry_secs + .store(retry_secs, Ordering::Relaxed); + if permanent { + tracing::error!( + attempts, + retry_secs, + err = %err, + "database rejected our configuration (bad DATABASE_URL or credentials?) — retrying, but operator action is likely required" + ); + } else { + warn!( + attempts, + retry_secs, + err = %err, + "database unavailable during startup; retrying" + ); + } + + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(retry_secs)) => {} + changed = shutdown_rx.changed() => { + if changed.is_err() || *shutdown_rx.borrow() { + return None; + } + } + } + } + } + } +} + +/// Errors that indicate misconfiguration rather than a transient outage: a +/// malformed DATABASE_URL, or a server that answered and rejected us — +/// Postgres error class 28xxx (invalid authorization) or 3D000 (database +/// does not exist). Best-effort: an error that anyhow can't downcast back to +/// sqlx just counts as transient. +fn is_likely_permanent_db_error(err: &anyhow::Error) -> bool { + match err.downcast_ref::() { + Some(sqlx::Error::Configuration(_)) => true, + Some(sqlx::Error::Database(db)) => db + .code() + .map(|c| c.starts_with("28") || c.starts_with("3D")) + .unwrap_or(false), + _ => false, + } +} + +fn database_retry_delay_secs(initial_secs: u64, max_secs: u64, attempts: u64) -> u64 { + // The exponent bound only keeps the u32 cast safe — max_secs is the real + // (operator-configurable) cap, and saturating math handles overflow. + let exponent = attempts.saturating_sub(1).min(63) as u32; + initial_secs + .saturating_mul(2_u64.saturating_pow(exponent)) + .min(max_secs) +} + +async fn run_degraded_server( + listener: TcpListener, + node_did: String, + db_startup: Arc, + mut db_ready_rx: watch::Receiver, + mut shutdown_rx: watch::Receiver, +) -> Result<()> { + let addr = listener.local_addr().ok(); + let router = build_degraded_router(node_did, db_startup); + info!(?addr, "degraded HTTP server ready"); + + axum::serve(listener, router) + .with_graceful_shutdown(async move { + // wait_for resolves on predicate-true or sender-drop; either way + // this phase is over. + tokio::select! { + _ = db_ready_rx.wait_for(|ready| *ready) => {} + _ = shutdown_rx.wait_for(|stop| *stop) => {} + } + }) + .await?; + + Ok(()) +} + +fn build_degraded_router(node_did: String, db_startup: Arc) -> Router { + let state = DegradedState { + node_did, + db_startup, + }; + // Everything answers 503 with the same body — including /health and + // /ready, so peer pings (which treat any 2xx /health as alive) and + // uptime monitors correctly see a node that cannot serve traffic. + // `/` additionally carries the node identity for probing peers. + Router::new() + .route("/", get(degraded_node_info)) + .fallback(degraded_unavailable) + .with_state(state) +} + +/// One source of truth for the degraded 503 body, sharing the error +/// vocabulary with error.rs so clients see the same code/message for +/// "database unavailable" regardless of which phase produced it. +fn degraded_body(db_startup: &DbStartupStatus) -> serde_json::Value { + serde_json::json!({ + "status": "degraded", + "database": "initializing", + "error": error::DB_UNAVAILABLE_CODE, + "message": error::DB_UNAVAILABLE_MESSAGE, + "db_attempts": db_startup.attempts.load(Ordering::Relaxed), + "db_next_retry_secs": db_startup.next_retry_secs.load(Ordering::Relaxed), + }) +} + +async fn degraded_node_info(State(state): State) -> impl IntoResponse { + let mut body = degraded_body(&state.db_startup); + if let Some(obj) = body.as_object_mut() { + obj.insert("name".into(), "gitlawb-node".into()); + obj.insert("version".into(), env!("CARGO_PKG_VERSION").into()); + obj.insert("did".into(), state.node_did.clone().into()); + } + (StatusCode::SERVICE_UNAVAILABLE, Json(body)) +} + +async fn degraded_unavailable(State(state): State) -> impl IntoResponse { + ( + StatusCode::SERVICE_UNAVAILABLE, + Json(degraded_body(&state.db_startup)), + ) +} + +/// Spawn a small axum router that exposes only `GET /metrics` on its own +/// listener. Returns the JoinHandle so `main()` can abort it on shutdown. +/// This is deliberately separate from the main router so the metrics port +/// can be firewalled differently from the API port — bind to localhost +/// or a private interface only. +async fn spawn_metrics_server( + addr: &str, + mut shutdown_rx: watch::Receiver, +) -> Result> { + use axum::{response::IntoResponse, routing::get, Router}; + + async fn metrics_handler() -> impl IntoResponse { + match metrics::encode() { + Ok(body) => ( + axum::http::StatusCode::OK, + [( + axum::http::header::CONTENT_TYPE, + "text/plain; version=0.0.4; charset=utf-8", + )], + body, + ), + Err(e) => ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + [( + axum::http::header::CONTENT_TYPE, + "text/plain; charset=utf-8", + )], + format!("metrics encode error: {e}"), + ), + } + } + + let listener = TcpListener::bind(addr) + .await + .with_context(|| format!("failed to bind metrics listener to {addr}"))?; + let app = Router::new().route("/metrics", get(metrics_handler)); + + let handle = tokio::spawn(async move { + if let Err(e) = axum::serve(listener, app) + .with_graceful_shutdown(async move { + while !*shutdown_rx.borrow_and_update() { + if shutdown_rx.changed().await.is_err() { + break; + } + } + }) + .await + { + warn!(err = %e, "metrics server exited with error"); + } + }); + Ok(handle) +} + +fn build_operator_client( + config: &config::Config, + node_did: &str, +) -> Result { + use alloy::primitives::Address; + use std::str::FromStr; + + let contract_address = Address::from_str(&config.contract_node_staking) + .with_context(|| format!("invalid contract address: {}", config.contract_node_staking))?; + + let cfg = operator::OperatorConfig { + rpc_url: config.chain_rpc_url.clone(), + private_key: config.operator_private_key.clone(), + contract_address, + node_did: node_did.to_string(), + heartbeat_interval: std::time::Duration::from_secs(config.heartbeat_interval_hours * 3600), + strict_mode: config.operator_strict_mode, + }; + Ok(operator::OperatorClient::new(cfg)) +} + +/// Announce to bootstrap peers on startup, then periodically ping all known peers. +async fn gossip_task( + state: AppState, + bootstrap_peers: Vec, + mut shutdown_rx: tokio::sync::watch::Receiver, +) { + // If shutdown arrives during the initial delay, exit before announcing. + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {} + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + info!("gossip: shutdown during startup delay, exiting"); + return; + } + } + } + + // Reuse the shared no-redirect client for every gossip outbound call (the + // bootstrap announce POST and the periodic peer /health ping). Peer URLs are + // attacker-influenceable, so a 3xx to a private address must not be followed. + // Do NOT fall back to reqwest::Client::new(): its default follows redirects + // and would reintroduce the SSRF closed here (#93). + let client = state.http_client.clone(); + let my_did = state.node_did.to_string(); + let my_url = state.config.public_url.clone().unwrap_or_default(); + + // Announce ourselves to each bootstrap peer + for peer_url in &bootstrap_peers { + // Cooperative shutdown between peers — a slow peer shouldn't + // block the node exiting. + if *shutdown_rx.borrow() { + info!("gossip: shutdown signalled during peer announce, exiting"); + return; + } + let path = "/api/v1/peers/announce"; + let announce_url = format!("{}{}", peer_url.trim_end_matches('/'), path); + let body = serde_json::json!({ + "did": my_did.clone(), + "http_url": my_url.clone(), + }); + let body_bytes = match serde_json::to_vec(&body) { + Ok(bytes) => bytes, + Err(e) => { + tracing::warn!(err = %e, "failed to serialize peer announce body"); + continue; + } + }; + let signed = sign_request(state.node_keypair.as_ref(), "POST", path, &body_bytes); + // Per-request timeout inside the loop; do not let one hung peer + // block others. The request itself is a normal tokio future so + // it's cancel-safe on shutdown. + match tokio::time::timeout( + std::time::Duration::from_secs(5), + client + .post(&announce_url) + .header("Content-Type", "application/json") + .header("Content-Digest", signed.content_digest) + .header("Signature-Input", signed.signature_input) + .header("Signature", signed.signature) + .body(body_bytes) + .send(), + ) + .await + { + Ok(Ok(resp)) => { + if resp.status().is_success() { + if let Ok(json) = resp.json::().await { + // Add them back to our peer list + if let (Some(their_did), Some(their_url)) = ( + json.get("node_did").and_then(|v| v.as_str()), + json.get("node_url").and_then(|v| v.as_str()), + ) { + if !their_url.is_empty() { + let _ = state.db.upsert_peer(their_did, their_url).await; + tracing::info!(did = %their_did, url = %their_url, "bootstrap peer added"); + } + } + } + } + } + Ok(Err(e)) => { + tracing::warn!(url = %announce_url, err = %e, "failed to announce to bootstrap peer") + } + Err(_) => tracing::warn!(url = %announce_url, "bootstrap peer announce timed out (5s)"), + } + } + + // Periodic ping every 5 minutes — exit on shutdown. + let mut interval = tokio::time::interval(std::time::Duration::from_secs(300)); + loop { + tokio::select! { + _ = interval.tick() => { + let peers = match state.db.list_peers().await { + Ok(p) => p, + Err(_) => continue, + }; + for peer in peers { + let ok = ping_peer_health(&client, &peer.http_url).await; + let _ = state.db.mark_peer_ping(&peer.did, ok).await; + } + } + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + info!("gossip task: shutdown signal received, exiting"); + return; + } + } + } + } +} + +/// Build the shared node HTTP client used for every outbound fan-out (sync +/// trigger, profile/repo fetches, gossip announce + peer pings). +/// +/// No redirects: peer URLs are attacker-influenceable, so a `3xx` to a private +/// address must not be followed (SSRF guard, #78/#93). Do NOT replace with +/// `reqwest::Client::new()` — its default follows redirects. Kept as a named +/// builder so tests bind the redirect guarantee to the real client the node +/// runs, not a hand-rolled equivalent. +fn build_http_client() -> reqwest::Result { + reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .redirect(reqwest::redirect::Policy::none()) + .build() +} + +/// Ping a peer's `/health` endpoint and report whether it answered 2xx. +/// +/// Takes the client by reference so callers supply the shared, no-redirect +/// `state.http_client`. Peer URLs are attacker-influenceable, so a `3xx` to a +/// private address must not be followed. Do NOT call this with a bare +/// `reqwest::Client::new()`: its default follows redirects and would +/// reintroduce the SSRF this guards against (#93). +async fn ping_peer_health(client: &reqwest::Client, http_url: &str) -> bool { + let url = format!("{}/health", http_url.trim_end_matches('/')); + client + .get(&url) + .send() + .await + .map(|r| r.status().is_success()) + .unwrap_or(false) +} + +/// Persist a just-created identity key, removing the file if the write failed +/// (#194, F1). `create_new` makes the inode appear before the PEM is flushed, so a +/// failed `write_all` (ENOSPC/EIO/quota) leaves an empty or partial PEM behind; +/// every later start would then take the `exists()` branch and re-parse that +/// corrupt file forever, exiting `invalid PEM key` instead of regenerating. Remove +/// it on failure so the next start starts clean. +fn write_key_or_cleanup(path: &std::path::Path, write_result: std::io::Result<()>) -> Result<()> { + write_result.map_err(|e| { + // Best-effort: if removal also fails there is nothing more to do, and the + // original write error is the one worth surfacing. + let _ = std::fs::remove_file(path); + anyhow::Error::new(e).context(format!("failed to write key to {}", path.display())) + }) +} + +/// Verify an identity key file is not world/group-readable (#194, F2). A `chmod` +/// that failed or silently no-op'd (read-only mount, ACL mismatch) must not leave +/// a readable private key in use, so this is checked after any tightening attempt +/// and fails closed rather than logging a normal "loaded identity" path. +#[cfg(unix)] +fn ensure_key_mode_0600(path: &std::path::Path) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(path) + .with_context(|| format!("stat identity key {}", path.display()))? + .permissions() + .mode() + & 0o777; + if mode != 0o600 { + anyhow::bail!( + "identity key {} has mode {mode:o}, expected 0600 — refusing to use a \ + world/group-readable private key", + path.display() + ); + } + Ok(()) +} + +/// Outcome of an atomic key publish. +enum KeyPublish { + /// We created the key file; it now holds the complete PEM. + Won, + /// Another concurrent start already created it; ours was discarded. + Lost, +} + +/// Wall-clock budget for reading a key another start may still be publishing. +/// The atomic publish (`publish_key_atomically`) means a *present* key file is +/// always complete, so this normally succeeds on the first read; the deadline +/// only covers cross-host cache lag. It is a wall-clock budget, NOT a fixed +/// retry count, so a slow/stalled filesystem cannot starve a losing start after +/// an arbitrarily short (~100ms) window (#194). +const KEY_RACE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(5); + +/// Load an already-provisioned identity key. On Unix, defensively tighten looser +/// permissions to 0600 (do NOT reject a loose key — that would break existing +/// deployments; just narrow them), then verify the mode is actually 0600. +fn load_existing_key(path: &std::path::Path) -> Result { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if let Ok(meta) = std::fs::metadata(path) { + if meta.permissions().mode() & 0o777 != 0o600 { + // Do NOT silently ignore a failed tighten (#194, F2): surface it + // so a key we cannot secure is not read and used exposed. + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) + .with_context(|| { + format!("could not tighten identity key {} to 0600", path.display()) + })?; + } + } + // Verify the key is actually 0600 before using it — a chmod that + // succeeded-but-no-op'd (some mounts) still leaves it exposed. + ensure_key_mode_0600(path)?; + } + let pem = std::fs::read_to_string(path) + .with_context(|| format!("failed to read key from {}", path.display()))?; + let kp = Keypair::from_pem(&pem).map_err(|e| anyhow::anyhow!("invalid PEM key: {e}"))?; + info!(path = %path.display(), "loaded existing identity"); + Ok(kp) +} + +/// Load a key another concurrent start may still be publishing, polling until it +/// parses or `KEY_RACE_DEADLINE` elapses. An already-provisioned key parses on +/// the first attempt with no sleep. +fn load_racing_key(path: &std::path::Path) -> Result { + let deadline = std::time::Instant::now() + KEY_RACE_DEADLINE; + let mut last_err; + loop { + match load_existing_key(path) { + Ok(kp) => return Ok(kp), + Err(e) => last_err = e, + } + if std::time::Instant::now() >= deadline { + return Err(last_err); + } + std::thread::sleep(std::time::Duration::from_millis(2)); + } +} + +/// Publish `pem` to `final_path` atomically: write the full bytes to a sibling +/// temp file, then `hard_link` the temp into place. `hard_link` is atomic and +/// fails if `final_path` already exists, which gives three guarantees at once: +/// (a) the final path only ever appears with COMPLETE content — a concurrent +/// reader never observes an empty/half-written key, unlike a +/// `create_new`+`write_all` that exposes an empty inode before the PEM is +/// flushed (#194); +/// (b) a lost race never clobbers the winner — exactly one publisher links; +/// (c) a crashed publisher leaves only the temp, never a partial final that +/// would wedge every later start on `invalid PEM key`. +/// `before_link` is a no-op in production; tests use it to widen the +/// post-write / pre-link window deterministically. +fn publish_key_atomically( + final_path: &std::path::Path, + pem: &[u8], + before_link: &dyn Fn(), +) -> Result { + use std::io::Write; + let dir = final_path + .parent() + .filter(|p| !p.as_os_str().is_empty()) + .unwrap_or_else(|| std::path::Path::new(".")); + let stem = final_path + .file_name() + .and_then(|s| s.to_str()) + .unwrap_or("identity.pem"); + // Unique per call (process-global counter) so concurrent publishers never + // collide on the temp name; `create_new` below is the backstop. + static SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let tmp = dir.join(format!(".{stem}.tmp.{}.{seq}", std::process::id())); + + let mut opts = std::fs::OpenOptions::new(); + opts.write(true).create_new(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.mode(0o600); + } + let mut f = opts + .open(&tmp) + .with_context(|| format!("create temp key {}", tmp.display()))?; + // On a failed write, remove the temp so nothing partial is ever linked (#194, F1). + write_key_or_cleanup(&tmp, f.write_all(pem))?; + drop(f); + + before_link(); + + let linked = std::fs::hard_link(&tmp, final_path); + let _ = std::fs::remove_file(&tmp); + match linked { + Ok(()) => Ok(KeyPublish::Won), + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Ok(KeyPublish::Lost), + Err(e) => Err(e) + .with_context(|| format!("link identity key into place at {}", final_path.display())), + } +} + +fn load_or_create_keypair(config: &Config) -> Result { + let key_path = config.resolved_key_path(); + + // Fast path for the common already-provisioned case (still race-safe: a + // concurrently-publishing winner is handled by the retry in load_racing_key). + if key_path.exists() { + return load_racing_key(&key_path); + } + + let kp = Keypair::generate(); + let pem = kp + .to_pem() + .map_err(|e| anyhow::anyhow!("failed to serialize key: {e}"))?; + + if let Some(parent) = key_path.parent() { + std::fs::create_dir_all(parent)?; + } + + // Publish atomically: the final path only ever appears complete, and a lost + // race loads the winner's key rather than overwriting it. + match publish_key_atomically(&key_path, pem.as_bytes(), &|| {})? { + KeyPublish::Won => { + info!(path = %key_path.display(), did = %kp.did(), "generated new node identity"); + Ok(kp) + } + KeyPublish::Lost => load_racing_key(&key_path), + } +} + +#[cfg(test)] +mod gossip_ssrf_tests { + use super::ping_peer_health; + + // Build the client exactly as production does (super::build_http_client) so + // these tests bind the redirect guarantee to the real shared client the + // node runs. A regression that makes build_http_client follow redirects + // fails ping_peer_health_does_not_follow_redirect. + fn production_http_client() -> reqwest::Client { + super::build_http_client().expect("failed to build production http client") + } + + // A peer answering `/health` with a 302 toward an internal address must not + // be followed: the redirect target must never be requested (#93). + #[tokio::test] + async fn ping_peer_health_does_not_follow_redirect() { + let mut server = mockito::Server::new_async().await; + let internal = server + .mock("GET", "/internal-metadata") + .with_status(200) + .expect(0) + .create_async() + .await; + let _health = server + .mock("GET", "/health") + .with_status(302) + .with_header("location", &format!("{}/internal-metadata", server.url())) + .create_async() + .await; + + let ok = ping_peer_health(&production_http_client(), &server.url()).await; + + assert!(!ok, "a 302 must not count as a healthy peer"); + // expect(0) is enforced only at assert time; this fails if the redirect + // was followed to the internal target. + internal.assert_async().await; + } + + #[tokio::test] + async fn ping_peer_health_reports_success_on_200() { + let mut server = mockito::Server::new_async().await; + let _health = server + .mock("GET", "/health") + .with_status(200) + .create_async() + .await; + + let ok = ping_peer_health(&production_http_client(), &server.url()).await; + + assert!(ok, "a 200 /health must count as a healthy peer"); + } + + // A transport error (nothing listening) must map to unhealthy, never a + // spurious healthy — the .unwrap_or(false) arm. + #[tokio::test] + async fn ping_peer_health_reports_unhealthy_on_connection_error() { + let ok = ping_peer_health(&production_http_client(), "http://127.0.0.1:1").await; + assert!(!ok, "a connection error must count as an unhealthy peer"); + } +} + +#[cfg(all(test, unix))] +mod identity_key_tests { + use super::{ + load_or_create_keypair, load_racing_key, publish_key_atomically, Config, KeyPublish, + Keypair, + }; + use clap::Parser; + use std::os::unix::fs::PermissionsExt; + + // #194 (P2): the racing loader must wait out a SLOW winner on a meaningful + // wall-clock deadline, not give up after an arbitrary ~100ms window. Simulate + // a winner that only publishes the key after 250ms (past the old 100ms budget) + // and assert the loader waits it out instead of failing with `invalid PEM key`. + // RED with the old `for _ in 0..50` (2ms) fixed-count loop. + #[test] + fn load_racing_key_waits_out_a_slow_winner() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + let pem = Keypair::generate().to_pem().expect("pem"); + + let writer_path = key_path.clone(); + let writer = std::thread::spawn(move || { + std::thread::sleep(std::time::Duration::from_millis(250)); + publish_key_atomically(&writer_path, pem.as_bytes(), &|| {}).expect("publish"); + }); + + let started = std::time::Instant::now(); + let kp = load_racing_key(&key_path) + .expect("racing load must wait out the slow winner, not fail"); + let waited = started.elapsed(); + writer.join().expect("writer joins"); + + assert!( + waited >= std::time::Duration::from_millis(200), + "loader should have waited for the ~250ms-slow winner, only waited {waited:?}" + ); + assert!( + !format!("{}", kp.did()).is_empty(), + "loaded the published key" + ); + } + + // #194 (P2): the atomic publish preserves the single-winner no-clobber + // guarantee — a second publish of a DIFFERENT key must lose and leave the + // winner's key untouched — and it cleans up its temp files. + #[test] + fn publish_wins_then_second_publish_loses_without_clobbering() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + let pem_a = Keypair::generate().to_pem().expect("pem a"); + let pem_b = Keypair::generate().to_pem().expect("pem b"); + assert_ne!(pem_a.as_str(), pem_b.as_str(), "fixtures must differ"); + + let out = publish_key_atomically(&key_path, pem_a.as_bytes(), &|| {}).expect("publish a"); + assert!(matches!(out, KeyPublish::Won), "first publish wins"); + assert_eq!( + std::fs::read_to_string(&key_path).unwrap().as_str(), + pem_a.as_str(), + "final holds the full PEM" + ); + + let out = publish_key_atomically(&key_path, pem_b.as_bytes(), &|| {}).expect("publish b"); + assert!(matches!(out, KeyPublish::Lost), "second publish loses"); + assert_eq!( + std::fs::read_to_string(&key_path).unwrap().as_str(), + pem_a.as_str(), + "the winner's key must NOT be clobbered by a losing publish" + ); + + let leftovers: Vec<_> = std::fs::read_dir(dir.path()) + .unwrap() + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_string_lossy().contains(".tmp.")) + .collect(); + assert!( + leftovers.is_empty(), + "temp files must be cleaned up, found {leftovers:?}" + ); + } + + // #194 (P2, the core of jatmn's option b): while a winner is between writing + // its temp and linking it into place, a reader watching the FINAL path must + // see it absent or COMPLETE — never empty/partial. RED with a + // create_new(final)+write approach, which exposes an empty final for the whole + // write window. + #[test] + fn publish_never_exposes_a_partial_final() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = std::sync::Arc::new(dir.path().join("identity.pem")); + let pem = Keypair::generate().to_pem().expect("pem"); + + let wp = key_path.clone(); + let writer = std::thread::spawn(move || { + // Hold the post-write / pre-link window open for 200ms. + publish_key_atomically(&wp, pem.as_bytes(), &|| { + std::thread::sleep(std::time::Duration::from_millis(200)); + }) + .expect("publish"); + }); + + let deadline = std::time::Instant::now() + std::time::Duration::from_millis(400); + let mut saw_complete = false; + while std::time::Instant::now() < deadline { + if key_path.exists() { + let body = std::fs::read_to_string(&*key_path).unwrap_or_default(); + assert!( + Keypair::from_pem(&body).is_ok(), + "final key observed in a partial/empty state: {body:?}" + ); + saw_complete = true; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + writer.join().expect("writer joins"); + assert!( + saw_complete, + "reader should have observed the completed key within the window" + ); + } + + // A freshly created identity key must be 0600 immediately, with no + // world-readable disclosure window (the atomic create_new(...).mode(0o600) + // guarantee). This is the RED-then-GREEN anchor for the perms fix: the old + // fs::write + set_permissions sequence left a 0644 window. + #[test] + fn created_key_is_mode_0600() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + let config = Config::parse_from([ + "gitlawb-node", + "--key-path", + key_path.to_str().expect("utf8 path"), + ]); + + let _kp = load_or_create_keypair(&config).expect("create keypair"); + + let mode = std::fs::metadata(&key_path) + .expect("key file exists") + .permissions() + .mode(); + assert_eq!( + mode & 0o777, + 0o600, + "identity key must be 0600, got {:o}", + mode & 0o777 + ); + } + + // A lost create race (file already present) must load the existing key + // rather than overwrite it, and must not error on AlreadyExists. Loading an + // existing loose-permission key tightens it to 0600 defensively. + #[test] + fn existing_key_is_loaded_and_tightened() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + let config = Config::parse_from([ + "gitlawb-node", + "--key-path", + key_path.to_str().expect("utf8 path"), + ]); + + let first = load_or_create_keypair(&config).expect("create keypair"); + // Loosen perms to simulate a legacy/loose on-disk key. + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o644)) + .expect("loosen perms"); + + let second = load_or_create_keypair(&config).expect("load keypair"); + assert_eq!( + first.did(), + second.did(), + "reloading must return the same identity, not a new one" + ); + + let mode = std::fs::metadata(&key_path) + .expect("key file exists") + .permissions() + .mode(); + assert_eq!( + mode & 0o777, + 0o600, + "load path must tighten loose perms to 0600, got {:o}", + mode & 0o777 + ); + } + + // The create-race arm the single-threaded tests skip (they enter via the + // `exists()` fast path). N concurrent starts on ONE fresh path must converge + // on a single identity: the atomic publish links exactly one winner and the + // losers hit `AlreadyExists` and load the winner's key rather than overwriting + // it. RED: replace the atomic publish with a plain `fs::write` and each thread + // returns its own freshly generated key, so the DIDs diverge. + #[test] + fn concurrent_starts_converge_on_one_identity() { + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + let config = std::sync::Arc::new(Config::parse_from([ + "gitlawb-node", + "--key-path", + key_path.to_str().expect("utf8 path"), + ])); + + let n = 8; + let barrier = std::sync::Arc::new(std::sync::Barrier::new(n)); + let handles: Vec<_> = (0..n) + .map(|_| { + let c = config.clone(); + let b = barrier.clone(); + std::thread::spawn(move || { + // Release all threads at once to maximize the race. + b.wait(); + format!("{}", load_or_create_keypair(&c).expect("keypair").did()) + }) + }) + .collect(); + let dids: Vec = handles + .into_iter() + .map(|h| h.join().expect("thread joins")) + .collect(); + + let first = &dids[0]; + assert!( + dids.iter().all(|d| d == first), + "all concurrent starts must converge on one identity, got: {dids:?}" + ); + } + + /// #194 (F1): a failed write of a just-created key file removes it, so a later + /// start regenerates instead of re-parsing an empty/partial PEM forever and + /// wedging on `invalid PEM key`. RED without the `remove_file` in + /// `write_key_or_cleanup`: the partial file survives the error. + #[test] + fn failed_write_removes_the_partial_key_file() { + use super::write_key_or_cleanup; + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + std::fs::write(&key_path, b"partial-pem").expect("seed partial file"); + assert!(key_path.exists()); + + let err = std::io::Error::other("simulated ENOSPC"); + let r = write_key_or_cleanup(&key_path, Err(err)); + assert!(r.is_err(), "the write error must propagate"); + assert!( + !key_path.exists(), + "a failed first write must not leave a partial key file to wedge later starts" + ); + } + + /// A successful write leaves the file in place (the common case). + #[test] + fn successful_write_keeps_the_key_file() { + use super::write_key_or_cleanup; + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + std::fs::write(&key_path, b"good-pem").expect("seed file"); + assert!(write_key_or_cleanup(&key_path, Ok(())).is_ok()); + assert!( + key_path.exists(), + "a successful write leaves the file in place" + ); + } + + /// #194 (F2): a loose (world/group-readable) key that cannot be tightened is + /// rejected fail-closed rather than read and used exposed; a 0600 key is + /// accepted. RED without `ensure_key_mode_0600`: a 0644 key is used silently. + #[test] + fn loose_key_mode_is_rejected_not_used() { + use super::ensure_key_mode_0600; + let dir = tempfile::tempdir().expect("tempdir"); + let key_path = dir.path().join("identity.pem"); + std::fs::write(&key_path, b"key").expect("seed file"); + + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o644)) + .expect("loosen perms"); + let err = ensure_key_mode_0600(&key_path) + .expect_err("a world/group-readable key must be rejected") + .to_string(); + assert!( + err.contains("644"), + "the failure must name the exposed mode: {err}" + ); + + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)) + .expect("tighten perms"); + assert!( + ensure_key_mode_0600(&key_path).is_ok(), + "a 0600 key must be accepted" + ); + } +} + +#[cfg(test)] +mod shutdown_grace_tests { + use super::drive_serve_with_grace; + use std::time::{Duration, Instant}; + + // The must-not case: the drain (serve future) never completes but the signal + // has fired. The grace deadline must fire, report `grace_expired = true`, and + // return promptly so teardown proceeds — not hang forever. RED: a helper that + // just awaits `serve` (ignoring grace) hangs this test past its bound. + #[tokio::test] + async fn hung_drain_is_abandoned_after_grace() { + let (armed_tx, armed_rx) = tokio::sync::oneshot::channel::<()>(); + let _ = armed_tx.send(()); // signal already fired + let serve = std::future::pending::>(); // never drains + let start = Instant::now(); + + let (result, grace_expired) = + drive_serve_with_grace(serve, armed_rx, Duration::from_millis(50)).await; + + assert!(grace_expired, "a drain outlasting grace must be abandoned"); + assert!( + result.is_ok(), + "abandon path returns Ok so teardown proceeds" + ); + assert!( + start.elapsed() < Duration::from_secs(5), + "must not wait indefinitely for the hung drain" + ); + } + + // Normal drain: the serve future completes before grace, so the real serve + // result is propagated and `grace_expired` is false. + #[tokio::test] + async fn completed_drain_keeps_result_and_does_not_expire() { + let (_armed_tx, armed_rx) = tokio::sync::oneshot::channel::<()>(); + let serve = std::future::ready(Ok::<(), std::io::Error>(())); + + let (result, grace_expired) = + drive_serve_with_grace(serve, armed_rx, Duration::from_secs(3600)).await; + + assert!( + !grace_expired, + "a completed drain must not report grace expiry" + ); + assert!(result.is_ok()); + } + + // The grace clock starts at the SIGNAL, not at call time. With the signal + // unsent, a finite drain longer than `grace` still completes normally — total + // uptime is never bounded by grace. RED: drop `armed.await` from the grace + // branch and the deadline fires at 10ms, expiring before the 80ms drain. + #[tokio::test] + async fn grace_clock_starts_at_signal_not_call() { + let (_armed_tx, armed_rx) = tokio::sync::oneshot::channel::<()>(); + // Signal never fires; drain finishes after > grace. + let serve = async { + tokio::time::sleep(Duration::from_millis(80)).await; + Ok::<(), std::io::Error>(()) + }; + + let (result, grace_expired) = + drive_serve_with_grace(serve, armed_rx, Duration::from_millis(10)).await; + + assert!( + !grace_expired, + "grace must not fire while the shutdown signal is unsent" + ); + assert!(result.is_ok()); + } +} diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index aa0483db..efd80ec4 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/crates/gitlawb-node/src/main.rs @@ -1,1086 +1,7 @@ -mod api; -mod arweave; -mod auth; -mod bootstrap; -mod cert; -mod config; -mod db; -mod encrypted_pin; -mod error; -mod git; -mod graphql; -mod icaptcha; -mod ipfs_pin; -mod metrics; -mod operator; -mod p2p; -mod pinata; -mod rate_limit; -mod server; -mod state; -mod sync; -#[cfg(test)] -mod test_support; -mod visibility; -mod webhooks; - -use anyhow::{anyhow, Context, Result}; -use axum::extract::State; -use axum::http::StatusCode; -use axum::response::IntoResponse; -use axum::routing::get; -use axum::{Json, Router}; -use clap::Parser; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; -use tokio::net::TcpListener; -use tokio::sync::watch; -use tracing::{info, warn}; - -use gitlawb_core::http_sig::sign_request; -use gitlawb_core::identity::Keypair; - -use config::Config; -use db::Db; -use state::AppState; - -#[derive(Clone)] -struct DegradedState { - node_did: String, - db_startup: Arc, -} - -/// Two independent counters with no cross-field invariant — atomics, not a -/// lock, so the retry loop and the degraded handlers never contend. -#[derive(Default)] -struct DbStartupStatus { - attempts: AtomicU64, - next_retry_secs: AtomicU64, -} +//! Thin binary entry point. All boot logic lives in the library crate +//! (`src/lib.rs`) so out-of-crate integration tests can link the same code. #[tokio::main] -async fn main() -> Result<()> { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::from_default_env() - .add_directive("gitlawb_node=debug".parse().unwrap()) - .add_directive("tower_http=info".parse().unwrap()), - ) - .init(); - - let mut config = Config::parse(); - - // Merge the embedded seed list of public network nodes into the runtime - // bootstrap peers. Operators can opt out via GITLAWB_BOOTSTRAP_DISABLE_SEEDS. - bootstrap::merge_seeds(&mut config); - - if !config.public_read { - warn!( - "GITLAWB_PUBLIC_READ=false is reserved; per-repository private-read enforcement is not wired in alpha" - ); - } - - // Load or generate the node's identity keypair - let keypair = load_or_create_keypair(&config)?; - let node_did = keypair.did(); - - // One-time metrics init. Must run before any handler that calls into - // `metrics::record_*` so the registry exists when the first event fires. - // Safe to call even when GITLAWB_METRICS_ADDR is unset — those helpers - // are simply no-ops until something reads from the registry. - metrics::init(env!("CARGO_PKG_VERSION"), &node_did.to_string()); - - info!("╔══════════════════════════════════════════╗"); - info!( - "║ gitlawb node v{} ║", - env!("CARGO_PKG_VERSION") - ); - info!("╚══════════════════════════════════════════╝"); - // Process-wide shutdown signal. One sender lives in AppState (cloned - // into every handler); main() keeps a clone and flips it on SIGINT - // or SIGTERM. Tasks that hold a watch::Receiver get notified at - // their next await point. - let (shutdown_tx, _shutdown_rx_for_main) = watch::channel(false); - spawn_shutdown_signal(shutdown_tx.clone()); - - info!(did = %node_did, "node identity"); - info!(addr = %config.bind_addr(), "binding HTTP listener"); - - // Bind HTTP once, before dependency initialization, and keep this socket - // for the life of the process. The degraded server accepts on a dup of the - // same socket, so the degraded→full handoff never closes the port: while - // the full server initializes, connections queue in the shared backlog - // instead of being refused. - let listener = TcpListener::bind(config.bind_addr()) - .await - .with_context(|| format!("failed to bind to {}", config.bind_addr()))?; - let full_std_listener = listener.into_std()?; - let degraded_listener = TcpListener::from_std( - full_std_listener - .try_clone() - .context("failed to clone HTTP listener for degraded server")?, - )?; - - // Metrics must stay observable during a database outage — the degraded - // window is exactly when dashboards need data — so this listener starts - // before the DB connects. - let metrics_handle = if !config.metrics_addr.is_empty() { - match spawn_metrics_server(&config.metrics_addr, shutdown_tx.subscribe()).await { - Ok(handle) => { - info!(addr = %config.metrics_addr, "metrics endpoint listening"); - Some(handle) - } - Err(e) => { - warn!(err = %e, addr = %config.metrics_addr, "failed to start metrics endpoint — continuing without"); - None - } - } - } else { - info!("metrics endpoint disabled (GITLAWB_METRICS_ADDR not set)"); - None - }; - - let db_startup = Arc::new(DbStartupStatus::default()); - let (db_ready_tx, db_ready_rx) = watch::channel(false); - let mut degraded_handle = tokio::spawn(run_degraded_server( - degraded_listener, - node_did.to_string(), - Arc::clone(&db_startup), - db_ready_rx, - shutdown_tx.subscribe(), - )); - - // Connect to PostgreSQL database. A transient outage or bad secret should - // not crash-loop the process and hammer the database provider; permanent - // misconfiguration surfaces through error-level logs and the /ready check. - let db = tokio::select! { - db = connect_db_with_retry(&config, Arc::clone(&db_startup), shutdown_tx.subscribe()) => { - match db { - Some(db) => db, - None => { - // Shutdown requested while waiting for the database. The - // degraded server only serves one-shot 503s — abort it - // rather than drain, so a slow client can't stall exit. - degraded_handle.abort(); - return Ok(()); - } - } - } - degraded = &mut degraded_handle => { - if *shutdown_tx.borrow() { - return Ok(()); - } - return match degraded { - Ok(Ok(())) => Err(anyhow!("degraded HTTP server stopped before database became ready")), - Ok(Err(err)) => Err(err.context("degraded HTTP server failed")), - Err(err) => Err(anyhow!("degraded HTTP server task failed: {err}")), - }; - } - }; - - // Flip the degraded server into graceful shutdown, but do NOT await the - // drain: one slow in-flight request must not delay the full server, and - // the shared socket means there is no port gap to cover. The drain - // finishes (and logs) in the background. - db_ready_tx.send(true).ok(); - tokio::spawn(async move { - match degraded_handle.await { - Ok(Ok(())) => {} - Ok(Err(err)) => warn!(err = %err, "degraded HTTP server exited with error"), - Err(err) => warn!(err = %err, "degraded HTTP server task failed"), - } - }); - info!(addr = %config.bind_addr(), "database ready; starting full HTTP server"); - - // Prune peer rows that point back at this node (stale self-loop entries) - if let Some(public_url) = config.public_url.as_deref() { - match db.prune_self_peers(public_url).await { - Ok(0) => {} - Ok(n) => info!(removed = n, public_url, "pruned self-loop peer rows"), - Err(e) => warn!(err = %e, "prune_self_peers failed (non-fatal)"), - } - } - - // Prune peer rows with non-public hosts (loopback/private/internal) that - // were injected via the unauthenticated announce route — they poison the - // sync-notify fan-out (SSRF + crowding out real peers). - match db.prune_non_public_peers().await { - Ok(0) => {} - Ok(n) => info!(removed = n, "pruned non-public (poisoned) peer rows"), - Err(e) => warn!(err = %e, "prune_non_public_peers failed (non-fatal)"), - } - - // Ensure repos directory exists - std::fs::create_dir_all(&config.repos_dir).context("failed to create repos directory")?; - - // Start libp2p swarm (if p2p_port > 0) - let p2p_handle = if config.p2p_port > 0 { - let bootstrap_addrs = config - .p2p_bootstrap - .iter() - .filter_map(|s| s.parse().ok()) - .collect(); - let shutdown_rx = shutdown_tx.subscribe(); - match p2p::start( - &node_did.to_string(), - config.p2p_port, - bootstrap_addrs, - Arc::clone(&db), - config.auto_sync, - shutdown_rx, - ) - .await - { - Ok(handle) => { - info!(port = config.p2p_port, peer_id = %handle.local_peer_id, "libp2p swarm started"); - Some(Arc::new(handle)) - } - Err(e) => { - tracing::warn!(err = %e, "failed to start libp2p swarm — continuing without p2p"); - None - } - } - } else { - info!("p2p disabled (p2p_port = 0)"); - None - }; - - // Shared no-redirect HTTP client. See build_http_client for the SSRF rationale. - let http_client = Arc::new(build_http_client()?); - - let (ref_update_tx, _) = tokio::sync::broadcast::channel::(256); - let (task_event_tx, _) = tokio::sync::broadcast::channel::(256); - - let graphql_schema = Arc::new(graphql::build_schema( - Arc::clone(&db), - ref_update_tx.clone(), - task_event_tx.clone(), - )); - - let machine_id = std::env::var("FLY_MACHINE_ID").ok(); - if let Some(ref mid) = machine_id { - info!(" fly machine: {mid}"); - } - - // Initialize Tigris S3 client if bucket is configured - let tigris = if !config.tigris_bucket.is_empty() { - match git::tigris::TigrisClient::new(&config.tigris_bucket).await { - Ok(client) => { - info!(bucket = %config.tigris_bucket, "tigris storage enabled"); - Some(client) - } - Err(e) => { - tracing::warn!(err = %e, "failed to initialize Tigris client — using local-only storage"); - None - } - } - } else { - info!("tigris storage disabled (no bucket configured)"); - None - }; - - let repo_store = - git::repo_store::RepoStore::new(config.repos_dir.clone(), tigris, db.pool().clone()); - - // Per-DID limiter for the creation endpoints. Keyed on the authenticated - // DID (attacker-varied), so bound its key set to cap memory. - let rate_limiter = - rate_limit::RateLimiter::new_bounded(10, std::time::Duration::from_secs(3600), 200_000); - - // Per-client-IP flood brake for the creation endpoints. The per-DID limiter - // above is bypassed by a DID farm (one throwaway did:key per repo), which is - // exactly how the recurring spam-repo floods get past both it and the - // iCaptcha gate. Keyed on the resolved client IP so a single-source flood is - // capped regardless of how many identities it mints. Sized well above any - // legitimate per-IP creation rate; GITLAWB_CREATE_RATE_LIMIT overrides, 0 - // disables. Bounded key set — the key is a client-influenced IP. - let create_limit = std::env::var("GITLAWB_CREATE_RATE_LIMIT") - .ok() - .and_then(|v| v.trim().parse::().ok()) - .unwrap_or(120); - let create_ip_rate_limiter = rate_limit::RateLimiter::new_bounded( - create_limit, - std::time::Duration::from_secs(3600), - 200_000, - ); - if create_limit == 0 { - tracing::warn!("GITLAWB_CREATE_RATE_LIMIT=0 — per-IP creation rate limiting disabled"); - } - - // Push-path flood brake: max git-receive-pack requests per client IP per - // hour (counts both the info/refs advertisement and the push POST). Sized - // for heavy agent automation while still stopping flood traffic (the June - // 2026 attack pushed several times per second per IP). GITLAWB_PUSH_RATE_LIMIT - // overrides; 0 disables. Bounded key set — the key is a client-influenced IP. - let push_limit = std::env::var("GITLAWB_PUSH_RATE_LIMIT") - .ok() - .and_then(|v| v.trim().parse::().ok()) - .unwrap_or(600); - let push_rate_limiter = rate_limit::RateLimiter::new_bounded( - push_limit, - std::time::Duration::from_secs(3600), - 200_000, - ); - if push_limit == 0 { - tracing::warn!("GITLAWB_PUSH_RATE_LIMIT=0 — per-IP push rate limiting disabled"); - } - - // Which forwarded header the edge is trusted to set. Default None (trust - // nothing, key on the socket peer). Fly nodes set GITLAWB_TRUSTED_PROXY=fly; - // a node behind Caddy/NGINX sets it to x-forwarded-for. - let push_limiter_trust = rate_limit::TrustedProxy::from_env_value( - &std::env::var("GITLAWB_TRUSTED_PROXY").unwrap_or_default(), - ); - tracing::info!(trust = ?push_limiter_trust, push_limit, "push rate limiter configured"); - - // Peer-sync flood brakes, keyed on the resolved client IP (per-DID is useless - // here — a did:key farm self-registers). Two buckets so an unsigned notify - // flood can't drain the signed trigger caller's quota (#82). Bounded key sets - // (the key is a client-influenced IP); 0 disables each. - let sync_trigger_rate_limiter = rate_limit::RateLimiter::new_bounded( - config.sync_trigger_rate_limit, - std::time::Duration::from_secs(3600), - 200_000, - ); - let peer_write_rate_limiter = rate_limit::RateLimiter::new_bounded( - config.peer_write_rate_limit, - std::time::Duration::from_secs(3600), - 200_000, - ); - if config.sync_trigger_rate_limit == 0 { - tracing::warn!( - "GITLAWB_SYNC_TRIGGER_RATE_LIMIT=0 — /sync/trigger IP rate limiting disabled" - ); - } - if config.peer_write_rate_limit == 0 { - tracing::warn!("GITLAWB_PEER_WRITE_RATE_LIMIT=0 — peer-write IP rate limiting disabled"); - } - - // Initialize the iCaptcha proof gate (inert unless ICAPTCHA_MODE is set). - icaptcha::init().await; - - let state = AppState { - config: Arc::new(config.clone()), - db, - node_did: node_did.clone(), - node_keypair: Arc::new(keypair), - p2p: p2p_handle, - http_client, - ref_update_tx, - task_event_tx, - graphql_schema, - machine_id, - repo_store, - rate_limiter, - create_ip_rate_limiter, - push_rate_limiter, - push_limiter_trust, - sync_trigger_rate_limiter, - peer_write_rate_limiter, - shutdown_tx: shutdown_tx.clone(), - }; - - // Periodic peer-count poll for the metrics gauge. If p2p is disabled - // we still set the gauge to 0 so dashboards don't show "no data". - { - let p2p_for_metrics = state.p2p.clone(); - let mut shutdown_rx = state.subscribe_shutdown(); - tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(15)); - loop { - tokio::select! { - _ = interval.tick() => { - let count = match &p2p_for_metrics { - Some(h) => h.status().await.map(|s| s.connected_peers).unwrap_or(0), - None => 0, - }; - metrics::set_peers_connected(count as i64); - } - _ = shutdown_rx.changed() => { - if *shutdown_rx.borrow() { - return; - } - } - } - } - }); - } - - // Periodic cleanup of expired rate limit entries + consumed-proof ledger - { - let rl = state.rate_limiter.clone(); - let create_ip_rl = state.create_ip_rate_limiter.clone(); - let push_rl = state.push_rate_limiter.clone(); - let sync_trigger_rl = state.sync_trigger_rate_limiter.clone(); - let peer_write_rl = state.peer_write_rate_limiter.clone(); - let db = state.db.clone(); - let mut shutdown_rx = state.subscribe_shutdown(); - tokio::spawn(async move { - loop { - tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_secs(300)) => { - rl.cleanup().await; - create_ip_rl.cleanup().await; - push_rl.cleanup().await; - sync_trigger_rl.cleanup().await; - peer_write_rl.cleanup().await; - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs() as i64) - .unwrap_or(0); - if let Err(e) = db.sweep_expired_proofs(now).await { - tracing::warn!(err = %e, "failed to sweep expired iCaptcha proofs"); - } - } - _ = shutdown_rx.changed() => { - if *shutdown_rx.borrow() { - break; - } - } - } - } - }); - } - - let router = server::build_router(state.clone()); - // Re-register the socket bound at startup — same fd, so there was never a - // moment with the port closed between the degraded and full servers. - let listener = TcpListener::from_std(full_std_listener) - .context("failed to re-register HTTP listener with the runtime")?; - - info!("✓ node started — did:{}", node_did); - info!(" repos dir: {}", config.repos_dir.display()); - info!( - " database: PostgreSQL ({})", - &config.database_url.split('@').next_back().unwrap_or("?") - ); - - // Publish our DID record to the Kademlia DHT shortly after startup - if let Some(p2p) = &state.p2p { - let did_record = p2p::DidRecord { - did: node_did.to_string(), - http_url: config.public_url.clone().unwrap_or_default(), - peer_id: p2p.local_peer_id.to_string(), - p2p_port: config.p2p_port, - timestamp: chrono::Utc::now().to_rfc3339(), - }; - let p2p_clone = Arc::clone(p2p); - let mut shutdown_rx = state.subscribe_shutdown(); - tokio::spawn(async move { - // Small delay so Kademlia can find peers first - tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {} - _ = shutdown_rx.changed() => return, - } - p2p_clone.put_did(did_record).await; - info!("DID record published to Kademlia DHT"); - }); - } - - // Spawn background gossip: announce to bootstrap peers, then ping known peers periodically - { - let gossip_state = state.clone(); - let bootstrap_peers = config.bootstrap_peers.clone(); - let shutdown_rx = state.subscribe_shutdown(); - tokio::spawn(async move { - gossip_task(gossip_state, bootstrap_peers, shutdown_rx).await; - }); - } - - // Start multi-node sync worker if auto_sync is enabled - if config.auto_sync { - sync::start( - Arc::clone(&state.db), - Arc::clone(&state.config), - Arc::clone(&state.node_keypair), - state.subscribe_shutdown(), - ); - info!("auto-sync worker started"); - } - - // On-chain operator setup: verify stake + spawn heartbeat loop - if !state.config.contract_node_staking.is_empty() - && !state.config.operator_private_key.is_empty() - { - match build_operator_client(&state.config, &state.node_did.to_string()) { - Ok(client) => match operator::startup_check(&client).await { - Ok(_) => { - let arc_client = Arc::new(client); - arc_client.spawn_heartbeat_loop(state.subscribe_shutdown()); - } - Err(e) => { - if state.config.operator_strict_mode { - return Err(e.context("strict-mode operator check failed")); - } - tracing::warn!(err = %e, "operator startup check failed — continuing without heartbeat loop"); - } - }, - Err(e) => { - if state.config.operator_strict_mode { - return Err(e.context("strict-mode: failed to build operator client")); - } - tracing::warn!(err = %e, "operator client could not be built — continuing without PoS"); - } - } - } else { - info!("on-chain PoS disabled (GITLAWB_CONTRACT_NODE_STAKING or GITLAWB_OPERATOR_PRIVATE_KEY unset)"); - } - - // axum's `with_graceful_shutdown` waits for in-flight requests to - // complete (up to the configured grace) once the future resolves. - let shutdown_signal_for_axum = state.subscribe_shutdown(); - let grace = std::time::Duration::from_secs(config.shutdown_grace_secs); - info!(grace_secs = config.shutdown_grace_secs, "axum server ready"); - - // `into_make_service_with_connect_info` exposes the socket peer address as - // `ConnectInfo` so the push limiter can key on the real client - // when no trusted proxy header applies (see `rate_limit::client_key`). - let serve_result = axum::serve( - listener, - router.into_make_service_with_connect_info::(), - ) - .with_graceful_shutdown(async move { - let mut rx = shutdown_signal_for_axum; - // Wait until the watcher flips to true, then return so axum - // can begin draining. - while !*rx.borrow_and_update() { - if rx.changed().await.is_err() { - // Sender dropped — treat as shutdown. - break; - } - } - }) - .await; - - // Server has stopped accepting new connections and drained in-flight - // requests. Tear the rest of the system down. - info!("HTTP server stopped, beginning process shutdown"); - if let Some(h) = metrics_handle { - h.abort(); - } - let _ = grace; // recorded for operators in the log above; not enforced - serve_result?; - info!("clean exit"); - Ok(()) -} - -fn spawn_shutdown_signal(tx: watch::Sender) { - tokio::spawn(async move { - #[cfg(unix)] - { - use tokio::signal::unix::{signal as unix_signal, SignalKind}; - let mut sigterm = - unix_signal(SignalKind::terminate()).expect("install SIGTERM handler"); - let mut sigint = unix_signal(SignalKind::interrupt()).expect("install SIGINT handler"); - tokio::select! { - _ = sigterm.recv() => info!("SIGTERM received, shutting down"), - _ = sigint.recv() => info!("SIGINT received, shutting down"), - } - } - #[cfg(not(unix))] - { - use tokio::signal; - let _ = signal::ctrl_c().await; - info!("Ctrl-C received, shutting down"); - } - tx.send(true).ok(); - }); -} - -async fn connect_db_with_retry( - config: &Config, - db_startup: Arc, - mut shutdown_rx: watch::Receiver, -) -> Option> { - let initial_retry_secs = config.db_retry_initial_secs; - let max_retry_secs = config.db_retry_max_secs.max(initial_retry_secs); - let acquire_timeout = std::time::Duration::from_secs(config.db_acquire_timeout_secs); - let attempt_timeout = std::time::Duration::from_secs(config.db_connect_timeout_secs); - let mut attempts = 0_u64; - - loop { - if *shutdown_rx.borrow() { - return None; - } - - attempts = attempts.saturating_add(1); - db_startup.attempts.store(attempts, Ordering::Relaxed); - - // Bound the whole attempt, not just the pool connect: migrations - // block on a cross-instance advisory lock, and an unbounded wait - // there would wedge this loop — no retries, no logs, no recovery. - // Timing out and retrying is safe; migrations are idempotent. - let attempt = match tokio::time::timeout( - attempt_timeout, - Db::connect( - &config.database_url, - config.db_max_connections, - acquire_timeout, - ), - ) - .await - { - Ok(result) => result, - Err(_) => Err(anyhow!( - "connect + migrate attempt exceeded {}s (GITLAWB_DB_CONNECT_TIMEOUT_SECS); \ - is another instance holding the migration lock?", - attempt_timeout.as_secs() - )), - }; - - match attempt { - Ok(db) => { - info!(attempts, "database connection established"); - return Some(Arc::new(db)); - } - Err(err) => { - // A bad DATABASE_URL or rejected credentials won't heal on - // their own. Still retry (exiting would crash-loop and hammer - // the provider — and take liveness down with it), but log at - // error level and skip straight to the maximum backoff; the - // /ready health check is what surfaces this to deploys. - let permanent = is_likely_permanent_db_error(&err); - let retry_secs = if permanent { - max_retry_secs - } else { - database_retry_delay_secs(initial_retry_secs, max_retry_secs, attempts) - }; - db_startup - .next_retry_secs - .store(retry_secs, Ordering::Relaxed); - if permanent { - tracing::error!( - attempts, - retry_secs, - err = %err, - "database rejected our configuration (bad DATABASE_URL or credentials?) — retrying, but operator action is likely required" - ); - } else { - warn!( - attempts, - retry_secs, - err = %err, - "database unavailable during startup; retrying" - ); - } - - tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_secs(retry_secs)) => {} - changed = shutdown_rx.changed() => { - if changed.is_err() || *shutdown_rx.borrow() { - return None; - } - } - } - } - } - } -} - -/// Errors that indicate misconfiguration rather than a transient outage: a -/// malformed DATABASE_URL, or a server that answered and rejected us — -/// Postgres error class 28xxx (invalid authorization) or 3D000 (database -/// does not exist). Best-effort: an error that anyhow can't downcast back to -/// sqlx just counts as transient. -fn is_likely_permanent_db_error(err: &anyhow::Error) -> bool { - match err.downcast_ref::() { - Some(sqlx::Error::Configuration(_)) => true, - Some(sqlx::Error::Database(db)) => db - .code() - .map(|c| c.starts_with("28") || c.starts_with("3D")) - .unwrap_or(false), - _ => false, - } -} - -fn database_retry_delay_secs(initial_secs: u64, max_secs: u64, attempts: u64) -> u64 { - // The exponent bound only keeps the u32 cast safe — max_secs is the real - // (operator-configurable) cap, and saturating math handles overflow. - let exponent = attempts.saturating_sub(1).min(63) as u32; - initial_secs - .saturating_mul(2_u64.saturating_pow(exponent)) - .min(max_secs) -} - -async fn run_degraded_server( - listener: TcpListener, - node_did: String, - db_startup: Arc, - mut db_ready_rx: watch::Receiver, - mut shutdown_rx: watch::Receiver, -) -> Result<()> { - let addr = listener.local_addr().ok(); - let router = build_degraded_router(node_did, db_startup); - info!(?addr, "degraded HTTP server ready"); - - axum::serve(listener, router) - .with_graceful_shutdown(async move { - // wait_for resolves on predicate-true or sender-drop; either way - // this phase is over. - tokio::select! { - _ = db_ready_rx.wait_for(|ready| *ready) => {} - _ = shutdown_rx.wait_for(|stop| *stop) => {} - } - }) - .await?; - - Ok(()) -} - -fn build_degraded_router(node_did: String, db_startup: Arc) -> Router { - let state = DegradedState { - node_did, - db_startup, - }; - // Everything answers 503 with the same body — including /health and - // /ready, so peer pings (which treat any 2xx /health as alive) and - // uptime monitors correctly see a node that cannot serve traffic. - // `/` additionally carries the node identity for probing peers. - Router::new() - .route("/", get(degraded_node_info)) - .fallback(degraded_unavailable) - .with_state(state) -} - -/// One source of truth for the degraded 503 body, sharing the error -/// vocabulary with error.rs so clients see the same code/message for -/// "database unavailable" regardless of which phase produced it. -fn degraded_body(db_startup: &DbStartupStatus) -> serde_json::Value { - serde_json::json!({ - "status": "degraded", - "database": "initializing", - "error": error::DB_UNAVAILABLE_CODE, - "message": error::DB_UNAVAILABLE_MESSAGE, - "db_attempts": db_startup.attempts.load(Ordering::Relaxed), - "db_next_retry_secs": db_startup.next_retry_secs.load(Ordering::Relaxed), - }) -} - -async fn degraded_node_info(State(state): State) -> impl IntoResponse { - let mut body = degraded_body(&state.db_startup); - if let Some(obj) = body.as_object_mut() { - obj.insert("name".into(), "gitlawb-node".into()); - obj.insert("version".into(), env!("CARGO_PKG_VERSION").into()); - obj.insert("did".into(), state.node_did.clone().into()); - } - (StatusCode::SERVICE_UNAVAILABLE, Json(body)) -} - -async fn degraded_unavailable(State(state): State) -> impl IntoResponse { - ( - StatusCode::SERVICE_UNAVAILABLE, - Json(degraded_body(&state.db_startup)), - ) -} - -/// Spawn a small axum router that exposes only `GET /metrics` on its own -/// listener. Returns the JoinHandle so `main()` can abort it on shutdown. -/// This is deliberately separate from the main router so the metrics port -/// can be firewalled differently from the API port — bind to localhost -/// or a private interface only. -async fn spawn_metrics_server( - addr: &str, - mut shutdown_rx: watch::Receiver, -) -> Result> { - use axum::{response::IntoResponse, routing::get, Router}; - - async fn metrics_handler() -> impl IntoResponse { - match metrics::encode() { - Ok(body) => ( - axum::http::StatusCode::OK, - [( - axum::http::header::CONTENT_TYPE, - "text/plain; version=0.0.4; charset=utf-8", - )], - body, - ), - Err(e) => ( - axum::http::StatusCode::INTERNAL_SERVER_ERROR, - [( - axum::http::header::CONTENT_TYPE, - "text/plain; charset=utf-8", - )], - format!("metrics encode error: {e}"), - ), - } - } - - let listener = TcpListener::bind(addr) - .await - .with_context(|| format!("failed to bind metrics listener to {addr}"))?; - let app = Router::new().route("/metrics", get(metrics_handler)); - - let handle = tokio::spawn(async move { - if let Err(e) = axum::serve(listener, app) - .with_graceful_shutdown(async move { - while !*shutdown_rx.borrow_and_update() { - if shutdown_rx.changed().await.is_err() { - break; - } - } - }) - .await - { - warn!(err = %e, "metrics server exited with error"); - } - }); - Ok(handle) -} - -fn build_operator_client( - config: &config::Config, - node_did: &str, -) -> Result { - use alloy::primitives::Address; - use std::str::FromStr; - - let contract_address = Address::from_str(&config.contract_node_staking) - .with_context(|| format!("invalid contract address: {}", config.contract_node_staking))?; - - let cfg = operator::OperatorConfig { - rpc_url: config.chain_rpc_url.clone(), - private_key: config.operator_private_key.clone(), - contract_address, - node_did: node_did.to_string(), - heartbeat_interval: std::time::Duration::from_secs(config.heartbeat_interval_hours * 3600), - strict_mode: config.operator_strict_mode, - }; - Ok(operator::OperatorClient::new(cfg)) -} - -/// Announce to bootstrap peers on startup, then periodically ping all known peers. -async fn gossip_task( - state: AppState, - bootstrap_peers: Vec, - mut shutdown_rx: tokio::sync::watch::Receiver, -) { - // If shutdown arrives during the initial delay, exit before announcing. - tokio::select! { - _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {} - _ = shutdown_rx.changed() => { - if *shutdown_rx.borrow() { - info!("gossip: shutdown during startup delay, exiting"); - return; - } - } - } - - // Reuse the shared no-redirect client for every gossip outbound call (the - // bootstrap announce POST and the periodic peer /health ping). Peer URLs are - // attacker-influenceable, so a 3xx to a private address must not be followed. - // Do NOT fall back to reqwest::Client::new(): its default follows redirects - // and would reintroduce the SSRF closed here (#93). - let client = state.http_client.clone(); - let my_did = state.node_did.to_string(); - let my_url = state.config.public_url.clone().unwrap_or_default(); - - // Announce ourselves to each bootstrap peer - for peer_url in &bootstrap_peers { - // Cooperative shutdown between peers — a slow peer shouldn't - // block the node exiting. - if *shutdown_rx.borrow() { - info!("gossip: shutdown signalled during peer announce, exiting"); - return; - } - let path = "/api/v1/peers/announce"; - let announce_url = format!("{}{}", peer_url.trim_end_matches('/'), path); - let body = serde_json::json!({ - "did": my_did.clone(), - "http_url": my_url.clone(), - }); - let body_bytes = match serde_json::to_vec(&body) { - Ok(bytes) => bytes, - Err(e) => { - tracing::warn!(err = %e, "failed to serialize peer announce body"); - continue; - } - }; - let signed = sign_request(state.node_keypair.as_ref(), "POST", path, &body_bytes); - // Per-request timeout inside the loop; do not let one hung peer - // block others. The request itself is a normal tokio future so - // it's cancel-safe on shutdown. - match tokio::time::timeout( - std::time::Duration::from_secs(5), - client - .post(&announce_url) - .header("Content-Type", "application/json") - .header("Content-Digest", signed.content_digest) - .header("Signature-Input", signed.signature_input) - .header("Signature", signed.signature) - .body(body_bytes) - .send(), - ) - .await - { - Ok(Ok(resp)) => { - if resp.status().is_success() { - if let Ok(json) = resp.json::().await { - // Add them back to our peer list - if let (Some(their_did), Some(their_url)) = ( - json.get("node_did").and_then(|v| v.as_str()), - json.get("node_url").and_then(|v| v.as_str()), - ) { - if !their_url.is_empty() { - let _ = state.db.upsert_peer(their_did, their_url).await; - tracing::info!(did = %their_did, url = %their_url, "bootstrap peer added"); - } - } - } - } - } - Ok(Err(e)) => { - tracing::warn!(url = %announce_url, err = %e, "failed to announce to bootstrap peer") - } - Err(_) => tracing::warn!(url = %announce_url, "bootstrap peer announce timed out (5s)"), - } - } - - // Periodic ping every 5 minutes — exit on shutdown. - let mut interval = tokio::time::interval(std::time::Duration::from_secs(300)); - loop { - tokio::select! { - _ = interval.tick() => { - let peers = match state.db.list_peers().await { - Ok(p) => p, - Err(_) => continue, - }; - for peer in peers { - let ok = ping_peer_health(&client, &peer.http_url).await; - let _ = state.db.mark_peer_ping(&peer.did, ok).await; - } - } - _ = shutdown_rx.changed() => { - if *shutdown_rx.borrow() { - info!("gossip task: shutdown signal received, exiting"); - return; - } - } - } - } -} - -/// Build the shared node HTTP client used for every outbound fan-out (sync -/// trigger, profile/repo fetches, gossip announce + peer pings). -/// -/// No redirects: peer URLs are attacker-influenceable, so a `3xx` to a private -/// address must not be followed (SSRF guard, #78/#93). Do NOT replace with -/// `reqwest::Client::new()` — its default follows redirects. Kept as a named -/// builder so tests bind the redirect guarantee to the real client the node -/// runs, not a hand-rolled equivalent. -fn build_http_client() -> reqwest::Result { - reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(10)) - .redirect(reqwest::redirect::Policy::none()) - .build() -} - -/// Ping a peer's `/health` endpoint and report whether it answered 2xx. -/// -/// Takes the client by reference so callers supply the shared, no-redirect -/// `state.http_client`. Peer URLs are attacker-influenceable, so a `3xx` to a -/// private address must not be followed. Do NOT call this with a bare -/// `reqwest::Client::new()`: its default follows redirects and would -/// reintroduce the SSRF this guards against (#93). -async fn ping_peer_health(client: &reqwest::Client, http_url: &str) -> bool { - let url = format!("{}/health", http_url.trim_end_matches('/')); - client - .get(&url) - .send() - .await - .map(|r| r.status().is_success()) - .unwrap_or(false) -} - -fn load_or_create_keypair(config: &Config) -> Result { - let key_path = config.resolved_key_path(); - - if key_path.exists() { - let pem = std::fs::read_to_string(&key_path) - .with_context(|| format!("failed to read key from {}", key_path.display()))?; - let kp = Keypair::from_pem(&pem).map_err(|e| anyhow::anyhow!("invalid PEM key: {e}"))?; - info!(path = %key_path.display(), "loaded existing identity"); - Ok(kp) - } else { - let kp = Keypair::generate(); - let pem = kp - .to_pem() - .map_err(|e| anyhow::anyhow!("failed to serialize key: {e}"))?; - - if let Some(parent) = key_path.parent() { - std::fs::create_dir_all(parent)?; - } - - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - std::fs::write(&key_path, pem.as_bytes())?; - std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600))?; - } - #[cfg(not(unix))] - std::fs::write(&key_path, pem.as_bytes())?; - - info!(path = %key_path.display(), did = %kp.did(), "generated new node identity"); - Ok(kp) - } -} - -#[cfg(test)] -mod gossip_ssrf_tests { - use super::ping_peer_health; - - // Build the client exactly as production does (super::build_http_client) so - // these tests bind the redirect guarantee to the real shared client the - // node runs. A regression that makes build_http_client follow redirects - // fails ping_peer_health_does_not_follow_redirect. - fn production_http_client() -> reqwest::Client { - super::build_http_client().expect("failed to build production http client") - } - - // A peer answering `/health` with a 302 toward an internal address must not - // be followed: the redirect target must never be requested (#93). - #[tokio::test] - async fn ping_peer_health_does_not_follow_redirect() { - let mut server = mockito::Server::new_async().await; - let internal = server - .mock("GET", "/internal-metadata") - .with_status(200) - .expect(0) - .create_async() - .await; - let _health = server - .mock("GET", "/health") - .with_status(302) - .with_header("location", &format!("{}/internal-metadata", server.url())) - .create_async() - .await; - - let ok = ping_peer_health(&production_http_client(), &server.url()).await; - - assert!(!ok, "a 302 must not count as a healthy peer"); - // expect(0) is enforced only at assert time; this fails if the redirect - // was followed to the internal target. - internal.assert_async().await; - } - - #[tokio::test] - async fn ping_peer_health_reports_success_on_200() { - let mut server = mockito::Server::new_async().await; - let _health = server - .mock("GET", "/health") - .with_status(200) - .create_async() - .await; - - let ok = ping_peer_health(&production_http_client(), &server.url()).await; - - assert!(ok, "a 200 /health must count as a healthy peer"); - } - - // A transport error (nothing listening) must map to unhealthy, never a - // spurious healthy — the .unwrap_or(false) arm. - #[tokio::test] - async fn ping_peer_health_reports_unhealthy_on_connection_error() { - let ok = ping_peer_health(&production_http_client(), "http://127.0.0.1:1").await; - assert!(!ok, "a connection error must count as an unhealthy peer"); - } +async fn main() -> anyhow::Result<()> { + gitlawb_node::run().await } diff --git a/crates/gitlawb-node/src/test_harness.rs b/crates/gitlawb-node/src/test_harness.rs new file mode 100644 index 00000000..7dfe6c81 --- /dev/null +++ b/crates/gitlawb-node/src/test_harness.rs @@ -0,0 +1,266 @@ +//! Test-only spawn surface for the real-node deny harness (see +//! `tests/deny_harness.rs`). Feature-gated behind `test-harness` so it never +//! compiles into the production binary. It exposes a single boot constructor so +//! the out-of-crate integration test can bring up a real node over a bound TCP +//! socket without the integration crate needing access to `graphql`, +//! `rate_limit`, `repo_store`, or the other internals `AppState` is built from. +//! +//! The node is built the same way `test_support::build_state` builds it (p2p +//! disabled, real migrated pool), but bound on `127.0.0.1:0` and served through +//! the real `axum::serve` stack with connect-info, so the middleware order, +//! body limits, and per-IP rate limiters all run exactly as in production. + +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use clap::Parser; +use sqlx::PgPool; +use tokio::net::TcpListener; +use tokio::sync::watch; +use uuid::Uuid; + +use gitlawb_core::identity::Keypair; + +use crate::config::Config; +use crate::db::{Db, RepoRecord, VisibilityMode}; +use crate::rate_limit::{RateLimiter, TrustedProxy}; +use crate::state::AppState; + +/// A running node bound to an ephemeral port. Dropping it signals graceful +/// shutdown and removes the temporary repository directory. +pub struct TestNode { + /// Base URL of the running node, e.g. `http://127.0.0.1:54321`. + pub base_url: String, + /// The node's own DID (for building requests that reference it). + pub node_did: String, + shutdown_tx: watch::Sender, + repos_dir: PathBuf, + /// Seeding handle over the same pool the node serves from. Kept private so + /// the integration crate seeds through the methods below rather than + /// naming `Db`/`RepoRecord` directly. + db: Arc, +} + +impl Drop for TestNode { + fn drop(&mut self) { + // Flip the shared shutdown signal so the serve task exits, then remove + // the temp repos dir. Both are best-effort: a test that already failed + // should not panic again in teardown. + let _ = self.shutdown_tx.send(true); + let _ = std::fs::remove_dir_all(&self.repos_dir); + } +} + +/// Allocate a process-unique temp directory for a spawned node's repositories +/// without pulling in the `tempfile` dev-dependency (which is unavailable to +/// the library crate under `--features test-harness`). +fn unique_repos_dir() -> PathBuf { + static COUNTER: AtomicU64 = AtomicU64::new(0); + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!("gitlawb-deny-harness-{}-{}", std::process::id(), n)) +} + +/// Build an [`AppState`] over the given migrated pool, mirroring +/// `test_support::build_state` but with a real on-disk repository directory so +/// git smart-HTTP routes can serve. P2P is always disabled. +fn build_state(db: Arc, pool: PgPool, repos_dir: PathBuf) -> AppState { + let keypair = Keypair::generate(); + let node_did = keypair.did(); + let (ref_tx, _) = tokio::sync::broadcast::channel(1); + let (task_tx, _) = tokio::sync::broadcast::channel(1); + let schema = Arc::new(crate::graphql::build_schema( + db.clone(), + ref_tx.clone(), + task_tx.clone(), + )); + AppState { + config: Arc::new(Config::parse_from(["gitlawb-node"])), + db, + node_did, + node_keypair: Arc::new(keypair), + p2p: None, + http_client: Arc::new(reqwest::Client::new()), + ref_update_tx: ref_tx, + task_event_tx: task_tx, + graphql_schema: schema, + machine_id: None, + repo_store: crate::git::repo_store::RepoStore::for_testing(repos_dir, pool), + rate_limiter: RateLimiter::new(100, Duration::from_secs(60)), + create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)), + push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)), + push_limiter_trust: TrustedProxy::None, + sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)), + peer_write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)), + shutdown_tx: watch::channel(false).0, + } +} + +/// Spawn a real node bound to `127.0.0.1:0` over the given (already-created, +/// empty) test pool. Runs the schema migrations, builds the router, and serves +/// it on a background task through the production `axum::serve` stack with +/// connect-info so per-IP layers key on the real peer. Returns once the socket +/// is bound and accepting connections. +pub async fn spawn_node(pool: PgPool) -> TestNode { + let db = Arc::new(crate::db::Db::for_testing(pool.clone())); + db.run_migrations() + .await + .expect("test schema migrations should apply"); + + let repos_dir = unique_repos_dir(); + std::fs::create_dir_all(&repos_dir).expect("create temp repos dir"); + + let state = build_state(db.clone(), pool, repos_dir.clone()); + let node_did = state.node_did.to_string(); + let shutdown_tx = state.shutdown_tx.clone(); + let mut shutdown_rx = shutdown_tx.subscribe(); + + let router = crate::server::build_router(state); + + let listener = TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("read bound addr"); + + tokio::spawn(async move { + let _ = axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(async move { + let _ = shutdown_rx.changed().await; + }) + .await; + }); + + TestNode { + base_url: format!("http://{addr}"), + node_did, + shutdown_tx, + repos_dir, + db, + } +} + +impl TestNode { + /// Insert a repo owned by `owner_did` and return its repo id. Mirrors + /// `test_support::seed_repo` (the `disk_path` field is unused by the + /// integration path — `RepoStore` computes the on-disk path from + /// `repos_dir`/owner/name, see [`Self::seed_bare_repo`]). + pub async fn seed_repo(&self, owner_did: &str, name: &str, is_public: bool) -> String { + let now = Utc::now(); + let record = RepoRecord { + id: Uuid::new_v4().to_string(), + name: name.to_string(), + owner_did: owner_did.to_string(), + description: None, + is_public, + default_branch: "main".to_string(), + created_at: now, + updated_at: now, + disk_path: format!("/tmp/{name}"), + forked_from: None, + machine_id: None, + }; + self.db.create_repo(&record).await.expect("seed repo"); + record.id + } + + /// Add a path-scoped visibility rule restricting `path_glob` to + /// `reader_dids` (empty = only the owner). Mode B keeps object SHAs intact + /// and withholds blob content, which is what the read/replication gates + /// enforce. + pub async fn withhold_path( + &self, + repo_id: &str, + path_glob: &str, + reader_dids: &[String], + created_by: &str, + ) { + self.db + .set_visibility_rule( + repo_id, + path_glob, + VisibilityMode::B, + reader_dids, + created_by, + ) + .await + .expect("set visibility rule"); + } + + /// Create a real bare git repository on disk at the exact path + /// `RepoStore::acquire` reads (`//.git`), with + /// one commit on `main` containing `files` (`(path, contents)`). Returns the + /// blob OID of each path so callers can assert a withheld OID never appears + /// in served bytes (U8). Shells out to `git`, mirroring + /// `test_support`'s served-content seam. + /// `object_format` is `"sha1"` for the git smart-HTTP surfaces and + /// `"sha256"` for the content-addressed `/ipfs/{cid}` surface (whose CIDs + /// are the sha2-256 object ids). + pub fn seed_bare_repo( + &self, + owner_did: &str, + name: &str, + files: &[(&str, &str)], + object_format: &str, + ) -> std::collections::HashMap { + let run = |args: &[&str], cwd: &std::path::Path| { + let out = std::process::Command::new("git") + .args(args) + .current_dir(cwd) + .output() + .expect("git runs"); + assert!( + out.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8_lossy(&out.stdout).trim().to_string() + }; + + // Build a source work tree, then bare-clone it into the served path. + let src = self.repos_dir.join(format!("src-{name}")); + std::fs::create_dir_all(&src).expect("create src dir"); + let fmt_arg = format!("--object-format={object_format}"); + run(&["init", "-q", "-b", "main", &fmt_arg], &src); + run(&["config", "user.email", "t@t"], &src); + run(&["config", "user.name", "t"], &src); + for (path, contents) in files { + let full = src.join(path); + if let Some(parent) = full.parent() { + std::fs::create_dir_all(parent).expect("create file parent dir"); + } + std::fs::write(&full, contents).expect("write seed file"); + run(&["add", path], &src); + } + run(&["commit", "-q", "-m", "seed"], &src); + + let mut oids = std::collections::HashMap::new(); + for (path, _) in files { + let oid = run(&["rev-parse", &format!("HEAD:{path}")], &src); + oids.insert((*path).to_string(), oid); + } + // The HEAD commit oid, used as the `want` when driving upload-pack. + oids.insert("HEAD".to_string(), run(&["rev-parse", "HEAD"], &src)); + + let slug = owner_did.replace([':', '/'], "_"); + let bare = self.repos_dir.join(&slug).join(format!("{name}.git")); + std::fs::create_dir_all(bare.parent().unwrap()).expect("create bare parent"); + run( + &[ + "clone", + "--bare", + "-q", + src.to_str().unwrap(), + bare.to_str().unwrap(), + ], + &self.repos_dir, + ); + + oids + } +} diff --git a/crates/gitlawb-node/tests/deny_harness.rs b/crates/gitlawb-node/tests/deny_harness.rs new file mode 100644 index 00000000..42655232 --- /dev/null +++ b/crates/gitlawb-node/tests/deny_harness.rs @@ -0,0 +1,494 @@ +//! Real-node deny harness: boots a real gitlawb-node over a bound TCP socket +//! and drives trust-boundary DENY paths through a real reqwest client, +//! asserting both the refusal status and that no withheld data leaks +//! (INV-1/INV-2/INV-8). Requires `--features test-harness`. +//! +//! Each `#[sqlx::test]` gets an ephemeral per-test database; `spawn_node` runs +//! the schema migrations and serves the real router on `127.0.0.1:0`. + +mod support; + +use std::process::Command; + +use support::assert::assert_denied; +use support::signing::signed_request; + +use gitlawb_core::cid::Cid; +use gitlawb_core::identity::Keypair; +use gitlawb_node::test_harness::spawn_node; + +/// Build the `/ipfs/{cid}` CID for a 64-hex sha2-256 git object id, matching the +/// node's own `Cid::from_sha256_bytes` (the value `get_by_cid` decodes back). +fn cid_for_oid(oid_hex: &str) -> String { + let bytes = hex::decode(oid_hex).expect("hex oid"); + let arr: [u8; 32] = bytes.as_slice().try_into().expect("32-byte sha256 oid"); + Cid::from_sha256_bytes(&arr).to_string() +} + +/// A reqwest client with a bounded timeout so a wedged node route fails the test +/// fast rather than hanging until the CI job timeout. +fn bounded_client() -> reqwest::Client { + reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .expect("client builds") +} + +// ── U2: the signing client produces signatures require_signature accepts ───── + +/// A validly signed receive-pack request clears `require_signature` (it does not +/// get a 401): it proceeds past the signature layer and is denied later for a +/// different reason (the repo does not exist). This proves the signing client +/// is producing signatures the real verifier accepts over the socket. +#[sqlx::test] +async fn signed_receive_pack_clears_signature_layer(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + let kp = Keypair::generate(); + + let resp = signed_request( + &client, + reqwest::Method::POST, + &node.base_url, + "/alice/repo/git-receive-pack", + b"0000".to_vec(), + &kp, + ) + .send() + .await + .expect("request sends"); + + assert_ne!( + resp.status().as_u16(), + 401, + "a valid signature must clear require_signature; got 401" + ); +} + +/// Tampering the body after signing invalidates the content-digest, so the +/// server rejects with 400 `content_digest_mismatch` (distinct from the 401 a +/// missing/invalid signature gets). Proves the signature actually covers the +/// body: the digest is signed and the server re-checks it against the bytes it +/// received. +#[sqlx::test] +async fn tampered_body_after_signing_is_rejected(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + let kp = Keypair::generate(); + + // Sign one body, then replace it before sending. + let mut req = signed_request( + &client, + reqwest::Method::POST, + &node.base_url, + "/alice/repo/git-receive-pack", + b"0000".to_vec(), + &kp, + ) + .build() + .expect("build request"); + *req.body_mut() = Some(reqwest::Body::from(b"tampered".to_vec())); + + let resp = client.execute(req).await.expect("request sends"); + assert_eq!( + resp.status().as_u16(), + 400, + "a body that no longer matches its content-digest must be rejected" + ); +} + +// ── U5(a): INV-8 — an unsigned push is denied and leaks nothing ────────────── + +/// An unauthenticated git-receive-pack (no signature headers) is rejected with +/// 401 before any handler runs, and the denial body carries no repo internals. +#[sqlx::test] +async fn unsigned_receive_pack_is_denied(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + + let url = format!("{}/alice/repo/git-receive-pack", node.base_url); + let resp = client + .post(&url) + .header("content-type", "application/x-git-receive-pack-request") + .body(b"0000".to_vec()) + .send() + .await + .expect("request sends"); + + // No repo was seeded, so there are no OIDs to leak; the assertion still + // enforces the 4xx-and-not-empty-200 INV-8 shape. + assert_denied(resp, 401, &[]).await; +} + +// ── U5(b): INV-8/INV-2 — anonymous /ipfs/{cid} of a withheld blob is denied ── + +/// A public repo with a `/secret/**` withhold rule (readers = one allowed DID). +/// An anonymous content-addressed read of the withheld blob's CID is denied +/// (404) and leaks neither the secret bytes nor its OID; the sibling public +/// blob's CID is served anonymously, proving the withhold is blob-scoped. +#[sqlx::test] +async fn anon_ipfs_read_of_withheld_blob_is_denied(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + + let owner = Keypair::generate(); + let owner_did = owner.did().to_string(); + let reader = Keypair::generate(); + + let repo_id = node.seed_repo(&owner_did, "u5b-repo", true).await; + // sha256 object format: the /ipfs CID is the sha2-256 object id. + let oids = node.seed_bare_repo( + &owner_did, + "u5b-repo", + &[ + ("public/a.txt", "public bytes U5b"), + ("secret/b.txt", "TOPSECRET-U5b"), + ], + "sha256", + ); + node.withhold_path( + &repo_id, + "/secret/**", + &[reader.did().to_string()], + &owner_did, + ) + .await; + + let secret_oid = oids["secret/b.txt"].clone(); + let secret_cid = cid_for_oid(&secret_oid); + let public_cid = cid_for_oid(&oids["public/a.txt"]); + + // Anonymous read of the withheld blob's CID: denied, no leak of content or OID. + let resp = client + .get(format!("{}/ipfs/{secret_cid}", node.base_url)) + .send() + .await + .expect("request sends"); + assert_denied( + resp, + 404, + &["TOPSECRET-U5b", &secret_oid, &secret_oid[..12]], + ) + .await; + + // The sibling public blob's CID is served to anon (withhold is blob-scoped). + let resp = client + .get(format!("{}/ipfs/{public_cid}", node.base_url)) + .send() + .await + .expect("request sends"); + assert_eq!( + resp.status().as_u16(), + 200, + "public blob CID must be served to anon" + ); + assert!( + resp.text().await.unwrap().contains("public bytes U5b"), + "the public blob content is returned" + ); +} + +// ── U6: INV-1 — a validly signed NON-owner mutation is owner-gated (403) ────── + +/// The case the in-crate `oneshot` suite cannot reach over the full stack: a +/// fully valid RFC-9421 signature from a non-owner DID hits a mutation and is +/// rejected by `require_owner` with 403, not merely "not authenticated". The +/// non-owner sends no `x-ucan` header, so `require_ucan_chain` passes through +/// and the request reaches the owner gate. +#[sqlx::test] +async fn wrong_owner_visibility_put_is_forbidden(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + + let owner = Keypair::generate(); + let owner_did = owner.did().to_string(); + let stranger = Keypair::generate(); + + node.seed_repo(&owner_did, "harness-repo", true).await; + + let path = format!("/api/v1/repos/{owner_did}/harness-repo/visibility"); + let body = br#"{"path_glob":"/","reader_dids":[]}"#.to_vec(); + + // Non-owner: valid signature, wrong identity -> 403 from the owner gate. + let resp = signed_request( + &client, + reqwest::Method::PUT, + &node.base_url, + &path, + body.clone(), + &stranger, + ) + .header("content-type", "application/json") + .send() + .await + .expect("request sends"); + assert_denied(resp, 403, &[]).await; + + // Reachability proof: the same request signed by the OWNER is not 403 (it + // reaches the handler). Without this, a 403 produced by an earlier layer or + // a mis-seeded repo would masquerade as a passing INV-1 case. + let resp = signed_request( + &client, + reqwest::Method::PUT, + &node.base_url, + &path, + body, + &owner, + ) + .header("content-type", "application/json") + .send() + .await + .expect("request sends"); + assert!( + resp.status().is_success(), + "owner's signed visibility PUT must reach the handler, got {}", + resp.status() + ); +} + +// ── U7: INV-2 — a read over a withheld path is denied and leaks nothing ─────── + +/// A public repo with a path-scoped withhold rule on `/secret/**`. An anonymous +/// blob read of the withheld path is denied (404 RepoNotFound) and the body +/// carries neither the secret content nor its blob OID; a read of a sibling +/// public path succeeds, proving the gate is path-scoped, not a blanket refusal. +#[sqlx::test] +async fn withheld_path_blob_read_is_denied(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + + let owner = Keypair::generate(); + let owner_did = owner.did().to_string(); + + let repo_id = node.seed_repo(&owner_did, "u7-repo", true).await; + let oids = node.seed_bare_repo( + &owner_did, + "u7-repo", + &[ + ("public.txt", "hello public"), + ("secret/data.txt", "TOPSECRET-CONTENT-U7"), + ], + "sha1", + ); + node.withhold_path(&repo_id, "/secret/**", &[], &owner_did) + .await; + + let secret_oid = oids["secret/data.txt"].clone(); + let short_oid = &secret_oid[..12]; + + // Withheld path: denied, and neither the content nor the OID (full or short) + // may appear in the denial body. + let resp = client + .get(format!( + "{}/api/v1/repos/{owner_did}/u7-repo/blob/secret/data.txt", + node.base_url + )) + .send() + .await + .expect("request sends"); + assert_denied(resp, 404, &["TOPSECRET-CONTENT-U7", &secret_oid, short_oid]).await; + + // Sibling public path: served, proving the withhold is path-scoped. + let resp = client + .get(format!( + "{}/api/v1/repos/{owner_did}/u7-repo/blob/public.txt", + node.base_url + )) + .send() + .await + .expect("request sends"); + assert_eq!( + resp.status().as_u16(), + 200, + "public sibling path must be served" + ); + assert!( + resp.text().await.unwrap().contains("hello public"), + "the public blob content is returned" + ); +} + +// ── U8: INV-2 — an anonymous clone/fetch excludes withheld subtree blobs ────── + +/// The replication/clone surface, the one `oneshot` cannot serve: a public repo +/// with a `/secret/**` withhold rule. A real anonymous `git clone` over +/// git-upload-pack must yield a packfile that omits the withheld blob's object +/// while keeping the sibling public blob. The assertion is packfile-aware: git +/// parsed the served pack, and `cat-file -e` checks object presence in it. +#[sqlx::test] +async fn anon_clone_excludes_withheld_subtree_blobs(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + + let owner = Keypair::generate(); + let owner_did = owner.did().to_string(); + + let repo_id = node.seed_repo(&owner_did, "u8-repo", true).await; + let oids = node.seed_bare_repo( + &owner_did, + "u8-repo", + &[ + ("public/a.txt", "public bytes U8"), + ("secret/b.txt", "TOPSECRET-U8"), + ], + "sha1", + ); + node.withhold_path(&repo_id, "/secret/**", &[], &owner_did) + .await; + + let secret_oid = oids["secret/b.txt"].clone(); + let public_oid = oids["public/a.txt"].clone(); + let head = oids["HEAD"].clone(); + + // Drive git-upload-pack directly (v0 stateless-RPC): a want for HEAD, a flush, + // then done. No side-band capability, so the response is a raw packfile after + // the NAK pkt-line. Vanilla `git clone` negotiates protocol v2 and deadlocks + // against the node's v0 server; the POST is the real replication surface and, + // via a bounded reqwest timeout, cannot wedge the suite. + let pkt = |s: &str| format!("{:04x}{}", s.len() + 4, s).into_bytes(); + let mut req_body: Vec = Vec::new(); + req_body.extend(pkt(&format!("want {head}\n"))); + req_body.extend_from_slice(b"0000"); // flush after wants + req_body.extend(pkt("done\n")); + + let client = bounded_client(); + let resp = client + .post(format!( + "{}/{owner_did}/u8-repo/git-upload-pack", + node.base_url + )) + .header("content-type", "application/x-git-upload-pack-request") + .body(req_body) + .send() + .await + .expect("upload-pack POST sends"); + assert_eq!( + resp.status().as_u16(), + 200, + "anon upload-pack POST must serve a pack" + ); + let bytes = resp.bytes().await.expect("read pack response"); + + // The packfile starts at the "PACK" magic (after the NAK pkt-line). + let pack_start = bytes + .windows(4) + .position(|w| w == b"PACK") + .expect("response must contain a packfile"); + let pack = &bytes[pack_start..]; + + // Index the served pack and list its objects (packfile-aware: a raw byte scan + // could not see an OID inside the zlib-compressed stream). + let dir = std::env::temp_dir().join(format!("gl-u8-pack-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let pack_path = dir.join("deny.pack"); + std::fs::write(&pack_path, pack).unwrap(); + let idx = Command::new("git") + .args(["index-pack", pack_path.to_str().unwrap()]) + .output() + .expect("git index-pack runs"); + assert!( + idx.status.success(), + "the served pack must index cleanly: {}", + String::from_utf8_lossy(&idx.stderr) + ); + let verify = Command::new("git") + .args([ + "verify-pack", + "-v", + pack_path.with_extension("idx").to_str().unwrap(), + ]) + .output() + .expect("git verify-pack runs"); + let listing = String::from_utf8_lossy(&verify.stdout).to_string(); + let _ = std::fs::remove_dir_all(&dir); + + assert!( + !listing.contains(&secret_oid), + "withheld blob {secret_oid} must be absent from the served pack; listing:\n{listing}" + ); + assert!( + listing.contains(&public_oid), + "public blob {public_oid} must be present (withhold is subtree-scoped); listing:\n{listing}" + ); +} + +// ── Additional INV-1 owner-gates over the real stack (fan-out of U6) ────────── + +/// The remaining high-blast-radius owner-gated mutations that previously had +/// only the source-level authz-table guard and no runtime deny test: branch +/// protection, webhook create/delete, and visibility removal. Each rejects a +/// validly-signed non-owner with 403, and the owner reaches the handler (not +/// 403), proving the 403 is the owner gate. All share `did_matches` as the root +/// gate (see the mutation check in the harness commit). +#[sqlx::test] +async fn additional_owner_gates_reject_non_owner(pool: sqlx::PgPool) { + let node = spawn_node(pool).await; + let client = bounded_client(); + let owner = Keypair::generate(); + let owner_did = owner.did().to_string(); + let stranger = Keypair::generate(); + + node.seed_repo(&owner_did, "gated-repo", true).await; + let base = format!("/api/v1/repos/{owner_did}/gated-repo"); + + // (method, path, body, needs-json-content-type) + let cases: Vec<(reqwest::Method, String, Vec, bool)> = vec![ + ( + reqwest::Method::POST, + format!("{base}/branches/main/protect"), + Vec::new(), + false, + ), + ( + reqwest::Method::DELETE, + format!("{base}/branches/main/protect"), + Vec::new(), + false, + ), + ( + reqwest::Method::POST, + format!("{base}/hooks"), + br#"{"url":"https://example.com/hook","events":["*"]}"#.to_vec(), + true, + ), + ( + reqwest::Method::DELETE, + format!("{base}/hooks/deadbeef"), + Vec::new(), + false, + ), + ( + reqwest::Method::DELETE, + format!("{base}/visibility"), + br#"{"path_glob":"/"}"#.to_vec(), + true, + ), + ]; + + let send = |m: reqwest::Method, path: String, body: Vec, json: bool, kp: &Keypair| { + let mut rb = signed_request(&client, m, &node.base_url, &path, body, kp); + if json { + rb = rb.header("content-type", "application/json"); + } + rb.send() + }; + + for (method, path, body, json) in cases { + // Non-owner: valid signature, wrong identity -> 403 owner gate, no leak. + let resp = send(method.clone(), path.clone(), body.clone(), json, &stranger) + .await + .unwrap_or_else(|e| panic!("{method} {path} sends: {e}")); + assert_denied(resp, 403, &[]).await; + + // Owner reaches the handler: NOT 403 (proves the 403 above was the gate, + // not an earlier layer or a 415/404 masquerade). + let resp = send(method.clone(), path.clone(), body, json, &owner) + .await + .unwrap_or_else(|e| panic!("{method} {path} owner sends: {e}")); + assert_ne!( + resp.status().as_u16(), + 403, + "owner must reach {method} {path} (got 403)" + ); + } +} diff --git a/crates/gitlawb-node/tests/support/assert.rs b/crates/gitlawb-node/tests/support/assert.rs new file mode 100644 index 00000000..ccdfdc85 --- /dev/null +++ b/crates/gitlawb-node/tests/support/assert.rs @@ -0,0 +1,117 @@ +//! INV-8 deny assertion. A denial must be an actual refusal (4xx) whose body +//! carries none of the data it is withholding, and must never be a 2xx (an +//! empty-200 rendered as success is the denial-as-success bug this guards). +//! +//! The check is split into a pure `check_denied` (unit-tested below with the +//! self-check scenarios) and an async `assert_denied` wrapper that reads a real +//! `reqwest::Response`. + +/// Pure deny check. Returns `Err(reason)` when the response is not a clean +/// denial. `withheld` are exact tokens (OIDs, paths, private slugs) that must +/// not appear anywhere in the body; register short-OID prefixes and encoded +/// path forms as separate tokens at the call site (a raw substring scan cannot +/// normalize them). +pub fn check_denied( + status: u16, + body: &str, + expected: u16, + withheld: &[&str], +) -> Result<(), String> { + // The harness only ever expects denials; a non-4xx expectation is a + // programming error in the test, not a node behavior under test. + if !(400..500).contains(&expected) { + return Err(format!( + "test bug: assert_denied expected status {expected} is not a 4xx" + )); + } + // INV-8: a denial rendered as success is the headline bug. Catch it + // explicitly so the failure message names it, rather than only failing the + // status-equality check below. + if (200..300).contains(&status) { + return Err(format!( + "denial rendered as success: got {status}, expected denial {expected}. \ + body={body:?}" + )); + } + if status != expected { + return Err(format!( + "expected denial status {expected}, got {status}. body={body:?}" + )); + } + for token in withheld { + if !token.is_empty() && body.contains(token) { + return Err(format!( + "withheld token {token:?} leaked in denial body: {body:?}" + )); + } + } + Ok(()) +} + +/// Drive a real response through the deny check and panic on failure. Reads the +/// full body once. +pub async fn assert_denied(resp: reqwest::Response, expected: u16, withheld: &[&str]) { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + if let Err(reason) = check_denied(status, &body, expected, withheld) { + panic!("{reason}"); + } +} + +#[cfg(test)] +mod tests { + use super::check_denied; + + #[test] + fn clean_403_with_no_leak_passes() { + let r = check_denied( + 403, + r#"{"error":"only the repo owner can do this"}"#, + 403, + &["a1b2c3"], + ); + assert!(r.is_ok(), "{r:?}"); + } + + #[test] + fn empty_200_is_flagged_as_denial_rendered_as_success() { + let r = check_denied(200, "", 403, &[]); + assert!(r.is_err()); + assert!(r.unwrap_err().contains("rendered as success")); + } + + #[test] + fn leaking_403_fails_and_names_the_token() { + let withheld = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef"; + let body = format!("not found near object {withheld}"); + let r = check_denied(403, &body, 403, &[withheld]); + assert!(r.is_err()); + assert!(r.unwrap_err().contains(withheld)); + } + + #[test] + fn wrong_status_fails() { + // A 404 when the owner gate's 403 was expected must not pass: it means + // an earlier layer denied, so the gate under test was never exercised. + let r = check_denied(404, "not found", 403, &[]); + assert!(r.is_err()); + assert!(r.unwrap_err().contains("expected denial status 403")); + } + + #[test] + fn non_4xx_expected_is_a_test_bug() { + // Guards the harness itself: expecting a non-denial status is a test + // authoring error, not a node behavior, and must be rejected. + let r = check_denied(403, "denied", 200, &[]); + assert!(r.is_err()); + assert!(r.unwrap_err().contains("not a 4xx")); + } + + #[test] + fn empty_withheld_token_never_false_positives() { + // An empty token must be skipped, never treated as "contained in every + // body" — otherwise every denial would spuriously fail as a leak. + let r = check_denied(403, "any body", 403, &[""]); + assert!(r.is_ok(), "{r:?}"); + } +} diff --git a/crates/gitlawb-node/tests/support/mod.rs b/crates/gitlawb-node/tests/support/mod.rs new file mode 100644 index 00000000..75828f5f --- /dev/null +++ b/crates/gitlawb-node/tests/support/mod.rs @@ -0,0 +1,6 @@ +//! Shared support for the real-node deny harness: the deny assertion (U4), the +//! RFC-9421 signing client (U2), and (via `gitlawb_node::test_harness`) the +//! node spawner (U3). + +pub mod assert; +pub mod signing; diff --git a/crates/gitlawb-node/tests/support/signing.rs b/crates/gitlawb-node/tests/support/signing.rs new file mode 100644 index 00000000..3e0d543c --- /dev/null +++ b/crates/gitlawb-node/tests/support/signing.rs @@ -0,0 +1,32 @@ +//! Real RFC-9421 request signing for the deny harness (U2). Wraps +//! `gitlawb_core::http_sig::sign_request` (the same entry point the +//! git-remote-gitlawb helper uses in production) so a test can present a valid +//! signature for any identity, reaching the authenticated-but-wrong-owner path +//! that an injected-DID shortcut cannot. + +use gitlawb_core::http_sig::sign_request; +use gitlawb_core::identity::Keypair; + +/// Build a reqwest request carrying a valid RFC-9421 signature for `keypair`. +/// +/// `path` is both appended to `base_url` to form the request URL and signed as +/// the `@path` component, so the value the server reconstructs and the value +/// that was signed are guaranteed identical. `body` is signed via its +/// content-digest and sent as the request body. +pub fn signed_request( + client: &reqwest::Client, + method: reqwest::Method, + base_url: &str, + path: &str, + body: Vec, + keypair: &Keypair, +) -> reqwest::RequestBuilder { + let sig = sign_request(keypair, method.as_str(), path, &body); + let url = format!("{base_url}{path}"); + client + .request(method, url) + .header("content-digest", sig.content_digest) + .header("signature-input", sig.signature_input) + .header("signature", sig.signature) + .body(body) +}