Compare commits

..
Author SHA1 Message Date
beardthelion 361805b1bd chore: sync Cargo.lock with manifests (0.5.1 + icaptcha-client sha2 dep)
The committed lockfile lagged the manifests: workspace crates were bumped to
0.5.1 and icaptcha-client's Cargo.toml declares sha2, but the lock still
carried 0.5.0 and no sha2 entry. Every build regenerated it, leaving
Cargo.lock perpetually dirty. Regenerated offline (no dependency upgrades) so
the tree stays clean.
2026-08-28 23:16:43 -05:00
beardthelion 526ee87344 fix(node): make the SDL mutation-fence wiring load-bearing; admit deprecated
An adversarial review of the SDL switch found three gaps:

- The fence's completeness source (SDL) and its hidden-mutation cross-check were
  not load-bearing: because the real MutationRoot has no visible=false field,
  reverting `fields` back to introspection and dropping the cross-check left the
  whole suite green (INV-21). Extract the sourcing + cross-check into a generic
  `assert_mutation_fence<Q,M,S>` that the real fence AND a synthetic hidden schema
  both drive. A new #[should_panic] test (fence_rejects_a_hidden_mutation) feeds a
  visible=false mutation that IS registered, so only the cross-check can fire;
  reverting the SDL source or deleting the cross-check now changes/removes that
  panic and turns the test RED (verified).

- The introspection cross-check used bare `fields { name }` (GraphQL default
  includeDeprecated:false), so a `#[graphql(deprecation = "...")]` mutation --
  executable and present in SDL -- would false-red as a hidden visible=false
  mutation. Use `fields(includeDeprecated: true)`; a new test
  (fence_admits_a_deprecated_mutation) pins it, going RED without the flag.

- schema_lists_exactly_the_registered_mutations still sourced from introspection
  and its comment called introspection "the source of truth", contradicting the
  fix. Retargeted at the SDL set with a corrected comment.

Also folds the two introspection navigations into one generic helper
(introspection_mutation_fields) and points the version-safety test at it, so
there is a single introspection query. cargo test -p gitlawb-node --bins green
(497); clippy and fmt clean. Only api/mod.rs changed.
2026-08-28 23:16:38 -05:00
beardthelion 70b01aa320 fix(node): source the GraphQL mutation fence from SDL, not introspection
Resolves jatmn's #219 P1. The every_graphql_mutation_has_its_gate completeness
fence enumerated mutation fields via async-graphql introspection, which filters
out `#[graphql(visible = false)]` fields. Those fields stay executable, so a
future hidden mutation would need no registry row and no marker while both
set-equality checks stayed green -- defeating the fence's guarantee that no live
mutation goes unclassified. (Confirmed by execution: a synthetic visible=false
mutation is absent from introspection, executable, and present in schema.sdl().)

Source completeness from the engine's own SDL instead. A new pure helper
mutation_field_names_from_sdl parses schema.sdl() with async-graphql's parser
AST (async_graphql::parser::parse_schema) and reads the mutation type's fields;
SDL includes visible=false fields and the AST parse is immune to a brace in a
field description. Introspection is kept as a cross-check: any field present in
SDL but hidden from introspection now fails named as a forbidden hidden
mutation. The stale doc comments (which claimed introspection could not hide a
field) are corrected.

A self-checking version-safety test builds a synthetic schema with a
visible=false mutation and asserts the SDL-sourced set includes it while
introspection omits it -- so if a future async-graphql ever filters SDL by
visibility, that test goes RED rather than the gap reopening silently.

Load-bearing verified by execution: stubbing the SDL enumeration to empty turns
four tests RED, including the fence itself; restoring returns green. Only
api/mod.rs changed; the mutation resolvers are untouched. `cargo test -p
gitlawb-node --bins` green (495); clippy and fmt clean.
2026-08-28 23:16:38 -05:00
beardthelion a31fed4a67 docs(node): reconcile GraphQL mutation-auth claims with the shipped state
README's known limitations, the OSS-readiness live-network blockers, and the
maintainer roadmap still listed GraphQL mutation auth as an open item, which
contradicts #87: GraphQL mutations already require a verified signer bound to the
acting DID. Reconcile each to the shipped reality while keeping the genuinely-open
nuance: the GraphQL mutation surface is the agent-task queue, repo writes stay
REST-only, and no owner-gated repo-write mutation is exposed over GraphQL. The
every_graphql_mutation_has_its_gate fence forces any new mutation to be classified
in its table; owner-gate and REST parity stay the INV-1 review check.
2026-08-28 23:16:36 -05:00
beardthelion 626c545553 test(node): add a completeness fence over the GraphQL mutation surface
Gate the GraphQL mutation surface the way REST is gated, as a SEPARATE guard with
its own bucket table (a guard over a heterogeneous set must classify per member,
INV-1). every_graphql_mutation_has_its_gate asserts the mutation set is complete
(every mutation classified) and that each carries its gate marker.

Completeness is derived from async-graphql's own introspection: the field names
come from the engine as structured data, so no source-text or SDL-format shape
(a brace or async fn in a comment or string, a whitespace/generic/skip variant, a
brace in a rustdoc description) can hide a live mutation. A new unregistered
mutation, including a repo-write, fails the build.

The marker check (does each resolver body call its gate) is a best-effort static
backstop, bounded to each resolver's own comment-masked, string-aware body. It is
explicitly not the completeness guarantee and not load-bearingness: a marker inside
a string literal is a documented vacuity residual, and the adversarial sqlx tests in
graphql/mutation.rs are the real proof the gates reject.

Verified by execution: an unregistered mutation reddens (unclassified), a removed
did_matches gate reddens (missing marker), the clean tree is green; no DB, no
feature flag.
2026-08-28 23:16:03 -05:00
32 changed files with 1048 additions and 7175 deletions

No files matched your search

-18
View File
@@ -103,13 +103,6 @@ GITLAWB_REQUIRE_SIGNED_PEER_WRITES=false
# is on. See docs/RUN-A-NODE.md.
GITLAWB_ENFORCE_OWNER_PUSH=true
# Require a `nonce` parameter on signatures that authorize a write. Without one
# the replay ledger keys on the signing-string hash, which rejects a repeated
# byte-identical mutation inside the same second. Keep false until every client
# signs a nonce; with this and GITLAWB_REQUIRE_SIGNED_PEER_WRITES both true, a
# peer still on a pre-nonce binary can no longer write to this node.
GITLAWB_REQUIRE_SIGNATURE_NONCE=false
# Comma-separated libp2p multiaddrs.
# Example: /ip4/1.2.3.4/udp/7546/quic-v1/p2p/12D3KooW...
GITLAWB_P2P_BOOTSTRAP=
@@ -288,17 +281,6 @@ GITLAWB_IPFS_RATE_LIMIT=600
# the client IP. 0 disables. Default 120.
GITLAWB_CREATE_RATE_LIMIT=120
# ── Signed-write rate limiting (per client IP, uses GITLAWB_TRUSTED_PROXY) ────
# Max requests per client IP per hour on the write routes that require an HTTP
# Signature (tasks, PR merge/close/review/comment, webhooks, branch protection,
# stars, replicas, labels, visibility, agent deregistration, bounties, profile,
# issue close/comment). Every signed attempt spends a consumed_signatures row
# before the handler checks ownership or existence, and any freshly minted
# did:key can sign, so this brake is what bounds the durable writes an
# unregistered caller can force. Separate bucket from the creation, push and
# peer-sync limits. 0 disables. Default 600.
GITLAWB_SIGNED_WRITE_RATE_LIMIT=600
# ── Peer-sync rate limiting (per client IP, uses GITLAWB_TRUSTED_PROXY below) ─
# /api/v1/peers/announce and /api/v1/sync/notify accept unsigned requests from
# known peers and run at higher frequency, so a generous bucket. Separate from
Generated
+12 -94
View File
@@ -770,23 +770,6 @@ dependencies = [
"zeroize",
]
[[package]]
name = "ark-ff"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7a806ac6c8307b929df4645776290a50ee2aac754ad09d8bdf73391309e43af"
dependencies = [
"ark-ff-asm 0.6.0",
"ark-ff-macros 0.6.0",
"ark-serialize 0.6.0",
"ark-std 0.6.0",
"digest 0.10.7",
"educe",
"num-bigint",
"num-traits",
"zeroize",
]
[[package]]
name = "ark-ff-asm"
version = "0.3.0"
@@ -817,16 +800,6 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "ark-ff-asm"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1479009684adc073dff49a1025d3a7065b317a9ead25aaaca38cdc70058ba8a2"
dependencies = [
"quote",
"syn 2.0.117",
]
[[package]]
name = "ark-ff-macros"
version = "0.3.0"
@@ -865,19 +838,6 @@ dependencies = [
"syn 2.0.117",
]
[[package]]
name = "ark-ff-macros"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a0691ed21ef00ef89c1e9bda832eba493dda3ec2f8d892fb25b705f73f06bb8"
dependencies = [
"num-bigint",
"num-traits",
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "ark-serialize"
version = "0.3.0"
@@ -911,30 +871,6 @@ dependencies = [
"num-bigint",
]
[[package]]
name = "ark-serialize"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a74dd304fd536fb95d0a328e72be759209cc496a9da094c5bc56e5fea4f9e86b"
dependencies = [
"ark-serialize-derive",
"ark-std 0.6.0",
"digest 0.10.7",
"num-bigint",
"serde_with",
]
[[package]]
name = "ark-serialize-derive"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4f153690697a2b91e5e1251ff98411ee5371500a111a0fd317a70e588eb300f9"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "ark-std"
version = "0.3.0"
@@ -965,16 +901,6 @@ dependencies = [
"rand 0.8.6",
]
[[package]]
name = "ark-std"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "367c9c827ed431bff6868b7aa926e05b16eb46603cc8b6e768e4a5553fa1d155"
dependencies = [
"num-traits",
"rand 0.8.6",
]
[[package]]
name = "arrayref"
version = "0.3.9"
@@ -3374,22 +3300,19 @@ dependencies = [
[[package]]
name = "git-remote-gitlawb"
version = "0.7.1"
version = "0.5.1"
dependencies = [
"anyhow",
"gitlawb-core",
"libc",
"mockito",
"reqwest",
"tempfile",
"tracing",
"tracing-subscriber",
"windows-sys 0.61.2",
]
[[package]]
name = "gitlawb-attest"
version = "0.7.1"
version = "0.5.1"
dependencies = [
"base64",
"ed25519-dalek",
@@ -3406,7 +3329,7 @@ dependencies = [
[[package]]
name = "gitlawb-core"
version = "0.7.1"
version = "0.5.1"
dependencies = [
"anyhow",
"base64",
@@ -3427,14 +3350,13 @@ dependencies = [
"sha2",
"thiserror 2.0.18",
"tokio",
"url",
"uuid",
"zeroize",
]
[[package]]
name = "gitlawb-node"
version = "0.7.1"
version = "0.5.1"
dependencies = [
"alloy",
"anyhow",
@@ -3467,7 +3389,6 @@ dependencies = [
"libp2p-swarm",
"mockito",
"multiaddr",
"multibase",
"prometheus",
"reqwest",
"serde",
@@ -3491,11 +3412,10 @@ dependencies = [
[[package]]
name = "gl"
version = "0.7.1"
version = "0.5.1"
dependencies = [
"alloy",
"anyhow",
"base64",
"chrono",
"clap",
"dirs",
@@ -3544,9 +3464,9 @@ dependencies = [
[[package]]
name = "h2"
version = "0.4.18"
version = "0.4.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "839c0e8a181239723652be9062bb56ca5bf5f64011f73b623f6f4fc59086a228"
checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
dependencies = [
"atomic-waker",
"bytes",
@@ -3865,7 +3785,7 @@ dependencies = [
"libc",
"percent-encoding",
"pin-project-lite",
"socket2 0.5.10",
"socket2 0.6.3",
"tokio",
"tower-service",
"tracing",
@@ -3901,7 +3821,6 @@ version = "0.4.0"
dependencies = [
"anyhow",
"gitlawb-core",
"mockito",
"reqwest",
"serde",
"serde_json",
@@ -5433,7 +5352,7 @@ dependencies = [
"quinn-udp",
"rustc-hash",
"rustls",
"socket2 0.5.10",
"socket2 0.6.3",
"thiserror 2.0.18",
"tokio",
"tracing",
@@ -5470,7 +5389,7 @@ dependencies = [
"cfg_aliases",
"libc",
"once_cell",
"socket2 0.5.10",
"socket2 0.6.3",
"tracing",
"windows-sys 0.52.0",
]
@@ -5820,15 +5739,14 @@ dependencies = [
[[package]]
name = "ruint"
version = "1.20.0"
version = "1.17.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f5e99bff0393163bb25029a6af25d3d8d202ba5b5438a74d1bd8789f5c822970"
checksum = "c141e807189ad38a07276942c6623032d3753c8859c146104ac2e4d68865945a"
dependencies = [
"alloy-rlp",
"ark-ff 0.3.0",
"ark-ff 0.4.2",
"ark-ff 0.5.0",
"ark-ff 0.6.0",
"bytes",
"fastrlp 0.3.1",
"fastrlp 0.4.0",
+1 -12
View File
@@ -71,7 +71,7 @@ Known limitations:
- Agent lifecycle revocation is not enforced by HTTP Signature authorization; do not rely on removing or revoking an agent record to block a compromised signer.
- Read visibility is not a blanket data-classification boundary: task, IPFS-pin, and Arweave-anchor listings are not repository-gated; withheld path names can be visible to a root reader; and later visibility changes cannot retract content already announced or externally anchored.
- Peer writes are signed by upgraded nodes, but strict signed-peer enforcement is opt-in during rolling upgrades.
- Current GraphQL mutations require an authenticated signer, but there is no mutation-specific guardrail that prevents a future mutation from omitting that check.
- GraphQL mutations require a signed identity (a verified signer bound to the acting DID) but are limited to the agent-task queue; repo writes are not exposed over GraphQL and remain REST-only.
- Pull-request review comments do not yet have threaded line-level anchors, and merges do not enforce approval requirements.
See:
@@ -369,14 +369,6 @@ GITLAWB_REQUIRE_SIGNED_PEER_WRITES=true
`POST /api/v1/sync/trigger` is not part of the staged rollout: it always requires a signature in both config modes and returns 401 without one, because each call drives an O(peers) outbound fan-out.
A verified signature is spent once on the write routes that require one: repo, issue, PR, task, bounty, profile, webhook, label, star, replica, visibility and agent mutations, plus git push and `POST /api/v1/sync/trigger`. A captured request to any of those cannot be replayed. The two peer-write routes are the exception. `POST /api/v1/peers/announce` and `POST /api/v1/sync/notify` reach the ledger only when `GITLAWB_REQUIRE_SIGNED_PEER_WRITES=true`; under the default (`false`) they verify a signature when the headers are present but never spend it, so a captured signed peer write is still replayable there until the fleet upgrades and the flag is turned on. The ledger keys on the RFC 9421 `nonce` parameter when the client signs one and falls back to the signing-string hash when it does not. The fallback is correct but coarser: a client that repeats a byte-identical mutation within the same second is rejected as a replay and has to re-sign. `GITLAWB_REQUIRE_SIGNATURE_NONCE` closes that fallback once every client emits a nonce:
```bash
GITLAWB_REQUIRE_SIGNATURE_NONCE=true
```
With it on, a nonce-less signature on a write route is rejected with 400 and `X-Gitlawb-Error: signature_nonce_required`. Signed reads are never affected. Flipping it on has a federation precondition: three of the node's own outbound signed calls (peer announce, replica registration, sync-notify) are peer traffic, so a node running with both this and `GITLAWB_REQUIRE_SIGNED_PEER_WRITES` set to true will reject peer writes from any node still on a pre-nonce binary. Upgrade the fleet before enabling it.
---
## Configuration
@@ -401,9 +393,6 @@ Important node settings:
| `GITLAWB_P2P_BOOTSTRAP` | Comma-separated libp2p multiaddrs. |
| `GITLAWB_BOOTSTRAP_DISABLE_SEEDS` | Disable embedded seed peers for isolated dev/test networks. |
| `GITLAWB_REQUIRE_SIGNED_PEER_WRITES` | Require signed peer announce/sync writes. Defaults to `false` during the staged rollout below. |
| `GITLAWB_REQUIRE_SIGNATURE_NONCE` | Require a signed `nonce` on write routes. Default false; see the staged rollout note above. |
| `GITLAWB_SIGNED_WRITE_RATE_LIMIT` | Per-client-IP requests per hour on the signed write routes (tasks, PR merge/close/review/comment, webhooks, branch protection, stars, replicas, labels, visibility, agent deregistration, bounties, profile, issue close/comment). Each signed attempt spends a ledger row before the handler checks anything, so the brake bounds what an unregistered caller can write. Own bucket; `0` disables. Default 600. |
<!-- verbatim -->
| `GITLAWB_ENFORCE_OWNER_PUSH` | Require the authenticated pusher to be the repo owner on `git-receive-pack`. **Defaults to `true`.** A `did:key` signature is authentication, not authorization — anyone can mint a key and sign — so with this off every signed caller may push to every repository, private ones included. Delegated and CI keys count as non-owners: a UCAN `git/push` capability is verified but not yet honored for authorization, so they cannot push while this is on. Set `false` only for a rolling upgrade; see [`docs/RUN-A-NODE.md`](docs/RUN-A-NODE.md). |
| `GITLAWB_AUTO_SYNC` | Enable automatic sync from known peers. |
| `GITLAWB_MAX_PACK_BYTES` | Max git pack body size for smart-HTTP routes. |
+13 -370
View File
@@ -8,16 +8,11 @@
//!
//! RFC 9421 headers produced by `sign_request`:
//! Content-Digest: sha-256=:base64hash:
//! Signature-Input: sig1=("@method" "@path" "content-digest");keyid="did:key:z6Mk...";alg="ed25519";created=<unix>;nonce="<32 hex chars>"
//! Signature-Input: sig1=("@method" "@path" "content-digest");keyid="did:key:z6Mk...";alg="ed25519";created=<unix>
//! Signature: sig1=:base64signature:
//!
//! The `nonce` is a per-request 128-bit value. It sits in the parameter tail,
//! which is what `@signature-params` is built from, so it is covered by the
//! signature without appearing in the covered-component list.
use base64::{engine::general_purpose::STANDARD, Engine};
use chrono::Utc;
use rand::{rngs::OsRng, RngCore};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
@@ -28,28 +23,15 @@ use crate::{Error, Result};
/// The component identifiers covered by every gitlawb signature.
pub const COVERED_COMPONENTS: &[&str] = &["@method", "@path", "content-digest"];
/// How old a signature may be and still verify.
///
/// Do not shrink this without measuring large-pack pushes: the verifier
/// buffers the entire request body before checking `created`, so a multi-GB
/// pack spends its upload time inside this budget.
pub const MAX_SIGNATURE_AGE_SECS: i64 = 300;
/// How far into the future a signature's `created` may sit. Absorbs ordinary
/// clock skew without letting a signer pre-date its way to a wider window.
pub const MAX_FUTURE_SKEW_SECS: i64 = 30;
/// The three headers produced by RFC 9421 signing.
#[derive(Debug, Clone)]
pub struct SignedHeaders {
/// `Content-Digest: sha-256=:base64:`
pub content_digest: String,
/// `Signature-Input: sig1=(...);keyid="...";alg="ed25519";created=<unix>;nonce="..."`
/// `Signature-Input: sig1=(...);keyid="...";alg="ed25519";created=<unix>`
pub signature_input: String,
/// `Signature: sig1=:base64:`
pub signature: String,
/// The per-request nonce carried in `Signature-Input`.
pub nonce: String,
}
/// A parsed RFC 9421 signature (from Signature-Input + Signature headers).
@@ -60,8 +42,6 @@ pub struct HttpSignature {
pub created: i64,
pub components: Vec<String>,
pub signature_bytes: Vec<u8>,
/// The `nonce` param, absent on signatures from a pre-nonce signer.
pub nonce: Option<String>,
}
impl HttpSignature {
@@ -84,15 +64,6 @@ impl HttpSignature {
.find(')')
.ok_or_else(|| Error::HttpSignature("missing ')' in Signature-Input".into()))?;
// Both parentheses are located independently, so a header carrying them
// out of order (`sig1=)(`) would build a reversed range and panic. This
// runs on unauthenticated header text, so it has to be a rejection.
if close < open {
return Err(Error::HttpSignature(
"'(' must precede ')' in Signature-Input".into(),
));
}
let components_str = &rest[open + 1..close];
let params_str = &rest[close + 1..]; // starts with ';'
@@ -122,9 +93,6 @@ impl HttpSignature {
.parse()
.map_err(|_| Error::HttpSignature("invalid created timestamp".into()))?;
// Optional: absent on signatures from a signer that predates the nonce.
let nonce = params.get("nonce").map(|v| v.trim_matches('"').to_string());
// Signature: sig1=:base64bytes:
let sig_b64 = sig_header
.trim()
@@ -142,39 +110,18 @@ impl HttpSignature {
created,
components,
signature_bytes,
nonce,
})
}
/// Reject a `created` timestamp outside the acceptance window.
///
/// The window is deliberately asymmetric. A signature may be up to
/// [`MAX_SIGNATURE_AGE_SECS`] old, because the verifier buffers the whole
/// request body before this check runs and the git routes accept packs up
/// to gigabytes, so upload time is charged against that budget. It may be
/// only [`MAX_FUTURE_SKEW_SECS`] in the future, which is enough to absorb
/// ordinary clock skew.
///
/// This used to compare `(now - created).abs()` against a single bound,
/// which let a signer pre-date `created` and hold a signature valid for
/// twice the intended span (#253).
/// Reject if the `created` timestamp is more than 5 minutes from now.
pub fn check_created(&self) -> Result<()> {
let now = Utc::now().timestamp();
if self.created > now + MAX_FUTURE_SKEW_SECS {
let skew = (now - self.created).abs();
if skew > 300 {
return Err(Error::HttpSignature(format!(
"signature is dated {}s in the future (max {MAX_FUTURE_SKEW_SECS}s)",
self.created - now
"clock skew too large: {skew}s (max 300s)"
)));
}
let age = now - self.created;
if age > MAX_SIGNATURE_AGE_SECS {
return Err(Error::HttpSignature(format!(
"signature is {age}s old (max {MAX_SIGNATURE_AGE_SECS}s)"
)));
}
Ok(())
}
@@ -211,36 +158,6 @@ pub fn build_signing_string(
Ok(lines.join("\n"))
}
/// Render `COVERED_COMPONENTS` as the parenthesised component list that goes
/// inside `Signature-Input`, e.g. `"@method" "@path" "content-digest"`.
///
/// Both the wire header and [`build_signing_string`]'s input are built from
/// this one source. Writing the list out by hand in the header while passing
/// the const to the signing-string builder lets a client sign over one set
/// while advertising another, which no verifier can reconcile.
fn covered_components_list() -> String {
COVERED_COMPONENTS
.iter()
.map(|c| format!("\"{c}\""))
.collect::<Vec<_>>()
.join(" ")
}
/// The request-derived value for every component this signer knows how to
/// cover. Kept alongside [`covered_components_list`] so the guard test can
/// check the two agree without signing anything.
fn request_values_for(
method: &str,
path_and_query: &str,
content_digest: &str,
) -> HashMap<String, String> {
let mut values = HashMap::new();
values.insert("@method".to_string(), method.to_uppercase());
values.insert("@path".to_string(), path_and_query.to_string());
values.insert("content-digest".to_string(), content_digest.to_string());
values
}
/// Sign an HTTP request per RFC 9421 and return the three headers to inject.
pub fn sign_request(
keypair: &Keypair,
@@ -252,32 +169,22 @@ pub fn sign_request(
let content_digest = compute_content_digest(body);
let did = keypair.did();
// Full Signature-Input header value. The advertised component list is
// derived from COVERED_COMPONENTS rather than written out, so the wire
// header and the list we actually sign over cannot drift apart.
let advertised = covered_components_list();
// 128 bits from the OS CSPRNG, appended to the parameter tail. The tail is
// sliced into `sig_params_value` below and so lands in the
// `@signature-params` line, which puts the nonce under the signature
// (RFC 9421 §2.3) without touching COVERED_COMPONENTS. Verifiers that
// predate the nonce rebuild `@signature-params` from the received header
// text, so they keep verifying an unknown parameter unchanged.
let nonce = generate_nonce();
// Full Signature-Input header value
let signature_input = format!(
r#"sig1=({advertised});keyid="{did}";alg="ed25519";created={created};nonce="{nonce}""#
r#"sig1=("@method" "@path" "content-digest");keyid="{did}";alg="ed25519";created={created}"#
);
// The @signature-params component value is the part after "sig1="
let sig_params_value = &signature_input["sig1=".len()..];
let request_values = request_values_for(method, path_and_query, &content_digest);
let mut request_values = HashMap::new();
request_values.insert("@method".to_string(), method.to_uppercase());
request_values.insert("@path".to_string(), path_and_query.to_string());
request_values.insert("content-digest".to_string(), content_digest.clone());
// Guarded by `sign_request_supplies_every_covered_component`: that test
// fails cleanly if COVERED_COMPONENTS gains an entry `request_values_for`
// cannot supply, which is the only way this could fail.
let signing_string =
build_signing_string(COVERED_COMPONENTS, sig_params_value, &request_values)
.expect("request_values_for covers COVERED_COMPONENTS (see guard test)");
.expect("required components always present when building");
let sig_bytes = keypair.sign(signing_string.as_bytes());
let sig_b64 = STANDARD.encode(sig_bytes.to_bytes());
@@ -286,17 +193,9 @@ pub fn sign_request(
content_digest,
signature_input,
signature: format!("sig1=:{sig_b64}:"),
nonce,
}
}
/// Draw a fresh 128-bit nonce from the OS CSPRNG, hex-encoded.
fn generate_nonce() -> String {
let mut bytes = [0u8; 16];
OsRng.fill_bytes(&mut bytes);
hex::encode(bytes)
}
/// Compute RFC 9421 Content-Digest value: `sha-256=:base64(sha256(body)):`
pub fn compute_content_digest(body: &[u8]) -> String {
let mut hasher = Sha256::new();
@@ -424,95 +323,6 @@ mod tests {
assert!(d.len() > 12);
}
/// U1 guard: the component list on the wire must be exactly what we sign
/// over. Load-bearing by mutation, not by red-then-green: adding a fourth
/// entry to COVERED_COMPONENTS turns this red.
#[test]
fn signature_input_advertises_exactly_covered_components() {
let kp = Keypair::generate();
let headers = sign_request(&kp, "POST", "/api/test", b"body");
let parsed = HttpSignature::parse(&headers.signature_input, &headers.signature)
.expect("emitted Signature-Input must parse");
assert_eq!(
parsed.components, COVERED_COMPONENTS,
"the advertised component list drifted from COVERED_COMPONENTS"
);
}
/// U1 guard: every covered component must have a request-derived value.
/// This is the check that fails *cleanly* when COVERED_COMPONENTS grows,
/// instead of letting sign_request panic inside build_signing_string.
#[test]
fn sign_request_supplies_every_covered_component() {
let values = request_values_for("POST", "/api/test", "sha-256=:abc:");
let missing: Vec<&str> = COVERED_COMPONENTS
.iter()
.copied()
.filter(|c| !values.contains_key(*c))
.collect();
assert!(
missing.is_empty(),
"COVERED_COMPONENTS entries with no value in request_values_for: {missing:?}. \
sign_request would panic on every call; add the value or drop the component."
);
}
/// U3 boundary: the forward bound is real and tight.
#[test]
fn future_dated_created_outside_skew_is_rejected() {
let kp = Keypair::generate();
let mut headers = sign_request(&kp, "GET", "/api/v1/agents", b"");
let future = Utc::now().timestamp() + MAX_FUTURE_SKEW_SECS + 1;
headers.signature_input = headers
.signature_input
.split(";created=")
.next()
.map(|head| format!("{head};created={future}"))
.expect("signature_input carries created");
let sig = HttpSignature::parse(&headers.signature_input, &headers.signature).unwrap();
assert!(
sig.check_created().is_err(),
"created beyond the forward skew allowance must be rejected"
);
}
/// U3 boundary: just inside the forward allowance still verifies, so the
/// bound absorbs real clock skew rather than being effectively zero.
#[test]
fn future_dated_created_within_skew_is_accepted() {
let kp = Keypair::generate();
let mut headers = sign_request(&kp, "GET", "/api/v1/agents", b"");
let future = Utc::now().timestamp() + MAX_FUTURE_SKEW_SECS - 1;
headers.signature_input = headers
.signature_input
.split(";created=")
.next()
.map(|head| format!("{head};created={future}"))
.expect("signature_input carries created");
let sig = HttpSignature::parse(&headers.signature_input, &headers.signature).unwrap();
assert!(sig.check_created().is_ok(), "within skew must be accepted");
}
/// U3 regression: the backward budget must not be narrowed. A large pack
/// spends this long uploading before check_created ever runs.
#[test]
fn backward_budget_still_accepts_a_nearly_stale_signature() {
let kp = Keypair::generate();
let mut headers = sign_request(&kp, "POST", "/api/test", b"body");
let old = Utc::now().timestamp() - (MAX_SIGNATURE_AGE_SECS - 20);
headers.signature_input = headers
.signature_input
.split(";created=")
.next()
.map(|head| format!("{head};created={old}"))
.expect("signature_input carries created");
let sig = HttpSignature::parse(&headers.signature_input, &headers.signature).unwrap();
assert!(
sig.check_created().is_ok(),
"a signature well inside the age budget must still verify"
);
}
#[test]
fn clock_skew_rejection() {
let kp = Keypair::generate();
@@ -574,173 +384,6 @@ mod tests {
assert_ne!(d1, d2);
}
/// U2: the nonce is per-call, so two signatures over an identical request
/// differ and so do the `Signature-Input` values they were computed over.
#[test]
fn two_signatures_over_an_identical_request_differ() {
let kp = Keypair::generate();
let a = sign_request(&kp, "POST", "/api/register", b"same body");
let b = sign_request(&kp, "POST", "/api/register", b"same body");
assert_ne!(
a.nonce, b.nonce,
"each sign_request call must draw a fresh nonce"
);
assert_ne!(
a.signature_input, b.signature_input,
"the nonce must reach the wire header, not just the struct"
);
assert_ne!(
a.signature, b.signature,
"a covered nonce must change the signature bytes"
);
}
/// U2: `parse` round-trips the nonce, and a nonce-less header still parses.
#[test]
fn parse_exposes_nonce_and_tolerates_its_absence() {
let kp = Keypair::generate();
let headers = sign_request(&kp, "POST", "/api/register", b"body");
let parsed = HttpSignature::parse(&headers.signature_input, &headers.signature).unwrap();
assert_eq!(
parsed.nonce.as_deref(),
Some(headers.nonce.as_str()),
"the parsed nonce must be the one that was signed, unquoted"
);
let did = kp.did();
let nonceless = format!(
r#"sig1=("@method" "@path" "content-digest");keyid="{did}";alg="ed25519";created=1000"#
);
let parsed = HttpSignature::parse(&nonceless, &headers.signature).unwrap();
assert_eq!(
parsed.nonce, None,
"a pre-U2 signer's header must parse with no nonce"
);
}
/// U2: 128 bits of entropy, hex-encoded.
#[test]
fn nonce_is_128_bits() {
let kp = Keypair::generate();
let headers = sign_request(&kp, "GET", "/api/v1/agents", b"");
let raw = hex::decode(&headers.nonce).expect("nonce must be hex");
assert_eq!(raw.len(), 16, "nonce must carry exactly 128 bits");
}
/// U2 backward compatibility, the load-bearing one: a nonce-bearing
/// signature still verifies through the same reconstruct-and-verify flow
/// the node runs, with no verifier change.
#[test]
fn nonce_bearing_signature_verifies_through_the_unchanged_flow() {
use crate::identity::verify;
let kp = Keypair::generate();
let body = b"{\"did\":\"did:key:z6Mk\"}";
let headers = sign_request(&kp, "POST", "/api/register", body);
assert!(
headers.signature_input.contains(";nonce=\""),
"this test is only meaningful once the header carries a nonce"
);
let sig = HttpSignature::parse(&headers.signature_input, &headers.signature).unwrap();
assert_eq!(
sig.components, COVERED_COMPONENTS,
"the nonce must not disturb the covered-component list"
);
// Exactly what the node does: rebuild @signature-params from the
// received header text and verify over it.
let sig_params_value = headers.signature_input.strip_prefix("sig1=").unwrap();
let mut request_values = HashMap::new();
request_values.insert("@method".to_string(), "POST".to_string());
request_values.insert("@path".to_string(), "/api/register".to_string());
request_values.insert("content-digest".to_string(), headers.content_digest.clone());
let components_ref: Vec<&str> = sig.components.iter().map(String::as_str).collect();
let signing_string =
build_signing_string(&components_ref, sig_params_value, &request_values).unwrap();
let vk = sig.key_id.to_verifying_key().unwrap();
let sig_bytes: [u8; 64] = sig.signature_bytes.clone().try_into().unwrap();
assert!(
verify(&vk, signing_string.as_bytes(), &sig_bytes).is_ok(),
"a nonce-bearing signature must verify unchanged"
);
// And the nonce is genuinely covered: strip it and verification fails.
let without_nonce = sig_params_value
.split(";nonce=")
.next()
.unwrap()
.to_string();
let tampered =
build_signing_string(&components_ref, &without_nonce, &request_values).unwrap();
assert!(
verify(&vk, tampered.as_bytes(), &sig_bytes).is_err(),
"the nonce must be inside @signature-params, not decoration"
);
}
/// A `)` before the `(` must be rejected, not sliced backwards. The two
/// parentheses are found independently, so an out-of-order pair used to
/// build a reversed range and panic inside `parse` — reachable with two
/// headers and no credentials at all.
#[test]
fn reversed_parentheses_are_rejected_not_panicked() {
let err = HttpSignature::parse("sig1=)(", "sig1=:AAAA:")
.expect_err("reversed parentheses must return an error");
let msg = err.to_string();
assert!(
msg.contains("'(' must precede ')'"),
"expected the paren-ordering error, got: {msg}"
);
}
/// `parse` runs on attacker-controlled header text before any
/// authentication, so no input may reach a panic. Every entry here must
/// come back as `Err`; the test failing by panic is the case it exists for.
#[test]
fn malformed_signature_input_always_errors_never_panics() {
let did = Keypair::generate().did();
let long = format!("sig1=({}", "(".repeat(4096));
let cases: Vec<(&str, &str)> = vec![
("", "sig1=:AAAA:"),
("sig1=", "sig1=:AAAA:"),
("sig1=()", "sig1=:AAAA:"),
("sig1=)(", "sig1=:AAAA:"),
("sig1=(", "sig1=:AAAA:"),
("sig1=)", "sig1=:AAAA:"),
("sig1=))((", "sig1=:AAAA:"),
("sig1=;", "sig1=:AAAA:"),
(";", "sig1=:AAAA:"),
("sig1=)(", ""),
("sig1=)(", "sig1=:not base64!:"),
("sig1=(\"@method\");created=notanumber", "sig1=:AAAA:"),
("sig1=(\"@method\");created=", "sig1=:AAAA:"),
("sig1=(é)é;created=1000", "sig1=:AAAA:"),
("sig1=)é(", "sig1=:AAAA:"),
("sig1=é)(é", "sig1=:AAAA:"),
(&long, "sig1=:AAAA:"),
];
for (input, header) in cases {
let out = HttpSignature::parse(input, header);
assert!(
out.is_err(),
"malformed Signature-Input {input:?} parsed successfully"
);
}
// An unbalanced quote inside an otherwise well-formed header is
// tolerated: `trim_matches('"')` strips whatever quotes are there.
// The invariant that matters is that it returns rather than panics, so
// assert it parses and leaves the quote-stripped component behind.
let unbalanced = format!(r#"sig1=("@method);keyid="{did}";alg="ed25519";created=1000"#);
let parsed = HttpSignature::parse(&unbalanced, "sig1=:AAAA:")
.expect("an unbalanced quote is tolerated, not a parse failure");
assert_eq!(parsed.components, vec!["@method".to_string()]);
}
#[test]
fn method_uppercased_in_signing_string() {
let kp = Keypair::generate();
-1
View File
@@ -5,7 +5,6 @@ pub mod encrypt;
pub mod error;
pub mod http_sig;
pub mod identity;
pub mod node_denial;
// `url` is the one dependency here that drags a tail (idna, then the icu
// crates), and gitlawb-core is allowlisted to stay embeddable. Every client that
// needs this predicate already parses URLs, so they opt in and nothing else
-229
View File
@@ -1,229 +0,0 @@
//! The denial codes the node puts in `X-Gitlawb-Error`, in one place.
//!
//! When the node refuses a write it answers with an `X-Gitlawb-Error` header
//! and a matching `error` field in the JSON body. The `gl` client keys off that
//! header to turn a denial into a hard error instead of handing the caller a
//! response that pretty-prints like a success.
//!
//! # Why this is not named for the signature ledger
//!
//! It was, and only the ledger's `consume_signature` middleware built these
//! responses. Then the per-client flood brake in `rate_limit` started answering
//! 429 on the same five route groups the ledger's `signature_ledger_full` 429
//! already covered, and the only thing telling the two apart was the *absence*
//! of a header — the same signal we already accept a proxy may strip. Two
//! unrelated refusals were indistinguishable on the wire.
//!
//! The fix is to give the brake a code, which means this type's subject is the
//! wire vocabulary rather than one middleware. Putting [`RateLimited`] here
//! instead of in a sibling enum is deliberate: the exhaustive match in `gl` is
//! the whole reason the type exists, and a parallel type would need its own
//! parallel match, which is exactly the duplication that let
//! `signature_nonce_too_short` drift in the first place. One enum, one match,
//! one place to add a code.
//!
//! Both halves used to carry their own list of string literals, agreeing only
//! by having been typed the same way twice in two crates. They drifted exactly
//! as you would expect: `signature_nonce_too_short` was added to the node after
//! the client's list was written, so for a while the client returned `Ok(400)`
//! for that denial and the write looked like it might have happened.
//!
//! This enum is the single source of truth. The node builds its responses from
//! it, so the wire strings and statuses come from one place, and `gl` matches
//! it with no wildcard arm, so adding a variant here is a compile error in the
//! client until the client handles it.
//!
//! # Deliberately not `#[non_exhaustive]`
//!
//! Marking this `#[non_exhaustive]` would force every downstream crate to add a
//! wildcard arm, which is precisely the silent fallthrough this type exists to
//! prevent. The compile-time guarantee is worth more here than the freedom to
//! add a variant without touching the client: the client and the node ship from
//! this same repo and are versioned together.
//!
//! The guarantee is over the codes *this repo's node emits*, not over arbitrary
//! input. A client talks to nodes it does not control, so [`from_code`] returns
//! `Option` and an unrecognised string stays unrecognised at runtime.
//!
//! [`from_code`]: NodeDenial::from_code
//! [`RateLimited`]: NodeDenial::RateLimited
/// A refusal the node names in `X-Gitlawb-Error`.
///
/// The `as_str` value is the wire contract: it appears verbatim in the
/// `X-Gitlawb-Error` response header and in the body's `error` field, and
/// scripts match on it. Never change one.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum NodeDenial {
/// 400 — the node requires a `nonce` in `Signature-Input` and the request
/// carried none. A pre-nonce client; the caller must upgrade.
NonceRequired,
/// 400 — the request carried a `nonce`, but too short to be unique. The
/// client already emits the parameter, it just does not fill it properly.
NonceTooShort,
/// 409 — the node already admitted a request bearing this signature.
Replayed,
/// 429 — too many unexpired signatures for this identity. A rate condition,
/// not a permanent refusal.
LedgerFull,
/// 500 — the ledger was reached without a verified identity, meaning the
/// node's own auth layer order is wrong. The one code that is not the
/// caller's fault.
IdentityMissing,
/// 503 — the ledger backend is down and the node is failing closed.
LedgerUnavailable,
/// 429 — the node's per-client flood brake refused the request before any
/// handler ran. Not a ledger outcome: it is keyed on the caller's resolved
/// network address, not on an identity or a signature, so it clears when
/// the address's window ages out and no amount of re-signing helps. Shares
/// its status with [`LedgerFull`](Self::LedgerFull), which is precisely why
/// it needs a code of its own.
RateLimited,
}
impl NodeDenial {
/// Every variant. Kept in the same order as the declaration so a reader can
/// check it by eye; the `all_is_exhaustive` test below asserts it
/// is complete.
pub const ALL: [Self; 7] = [
Self::NonceRequired,
Self::NonceTooShort,
Self::Replayed,
Self::LedgerFull,
Self::IdentityMissing,
Self::LedgerUnavailable,
Self::RateLimited,
];
/// The wire code, as it appears in `X-Gitlawb-Error` and in the body.
pub const fn as_str(self) -> &'static str {
match self {
Self::NonceRequired => "signature_nonce_required",
Self::NonceTooShort => "signature_nonce_too_short",
Self::Replayed => "signature_replayed",
Self::LedgerFull => "signature_ledger_full",
Self::IdentityMissing => "signature_identity_missing",
Self::LedgerUnavailable => "signature_ledger_unavailable",
Self::RateLimited => "rate_limited",
}
}
/// The HTTP status the node pairs with this code.
///
/// A `u16` rather than an `http::StatusCode` because this crate carries no
/// HTTP dependency; the node converts once, and its own test proves every
/// value here is a valid status.
pub const fn status(self) -> u16 {
match self {
Self::NonceRequired | Self::NonceTooShort => 400,
Self::Replayed => 409,
Self::LedgerFull | Self::RateLimited => 429,
Self::IdentityMissing => 500,
Self::LedgerUnavailable => 503,
}
}
/// Parse a code off the wire.
///
/// Returns `None` for anything this build does not know, because the string
/// comes from a node the caller does not control. An unknown code is not an
/// error to report; it is a denial this client cannot describe, and the
/// caller decides what to do with the raw response.
pub fn from_code(code: &str) -> Option<Self> {
Self::ALL.into_iter().find(|d| d.as_str() == code)
}
}
impl std::fmt::Display for NodeDenial {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Guards [`NodeDenial::ALL`] against a variant added above it and
/// forgotten here: the match is exhaustive, so a new variant stops this
/// compiling, and the assertion catches the case where someone adds the arm
/// but not the `ALL` entry.
#[test]
fn all_is_exhaustive() {
fn tag(d: NodeDenial) -> u8 {
match d {
NodeDenial::NonceRequired => 0,
NodeDenial::NonceTooShort => 1,
NodeDenial::Replayed => 2,
NodeDenial::LedgerFull => 3,
NodeDenial::IdentityMissing => 4,
NodeDenial::LedgerUnavailable => 5,
NodeDenial::RateLimited => 6,
}
}
let mut tags: Vec<u8> = NodeDenial::ALL.iter().copied().map(tag).collect();
tags.sort_unstable();
assert_eq!(
tags,
(0..NodeDenial::ALL.len() as u8).collect::<Vec<_>>(),
"NodeDenial::ALL must list every variant exactly once",
);
}
/// The wire contract, pinned literally. These strings and statuses are what
/// deployed nodes emit and what deployed clients and scripts match on, so a
/// change here is a protocol break, not a rename.
#[test]
fn wire_codes_and_statuses_are_pinned() {
let pinned: [(NodeDenial, &str, u16); 7] = [
(NodeDenial::NonceRequired, "signature_nonce_required", 400),
(NodeDenial::NonceTooShort, "signature_nonce_too_short", 400),
(NodeDenial::Replayed, "signature_replayed", 409),
(NodeDenial::LedgerFull, "signature_ledger_full", 429),
(
NodeDenial::IdentityMissing,
"signature_identity_missing",
500,
),
(
NodeDenial::LedgerUnavailable,
"signature_ledger_unavailable",
503,
),
(NodeDenial::RateLimited, "rate_limited", 429),
];
assert_eq!(pinned.len(), NodeDenial::ALL.len());
for (denial, code, status) in pinned {
assert_eq!(denial.as_str(), code);
assert_eq!(denial.status(), status);
}
}
#[test]
fn from_code_round_trips_every_variant() {
for denial in NodeDenial::ALL {
assert_eq!(NodeDenial::from_code(denial.as_str()), Some(denial));
}
}
#[test]
fn from_code_rejects_what_this_build_does_not_know() {
for unknown in [
"",
"signature_",
"signature_nonce",
"signature_replayed_",
" signature_replayed",
"SIGNATURE_REPLAYED",
"repo_exists",
"human_detected",
"signature_seventh_code_from_a_newer_node",
] {
assert_eq!(
NodeDenial::from_code(unknown),
None,
"must not recognise {unknown:?}",
);
}
}
}
-1
View File
@@ -75,7 +75,6 @@ alloy = { version = "1", default-features = false, features = [
libp2p-dns = { version = "0.44.0", features = ["tokio"] }
[dev-dependencies]
multibase = { workspace = true }
mockito = "1"
tempfile = "3"
# `test-util` for `#[tokio::test(start_paused = true)]`: lets the deadline tests assert
+548
View File
@@ -279,6 +279,554 @@ mod authz_guard {
}
}
/// Completeness fence over the GraphQL mutation surface: the mirror of
/// `every_in_scope_mutation_has_its_gate` for `graphql/mutation.rs`. Kept a
/// SEPARATE guard with its own table (never folded into the REST rows above):
/// a guard over a heterogeneous set must classify per member, and these
/// same-named task rows belong to the GraphQL surface, not `api/tasks.rs`.
///
/// COMPLETENESS comes from the engine's own SDL (`schema.sdl()`), parsed as a
/// structured AST (#219), NOT introspection and NOT a text parse of source.
/// This matters: async-graphql OMITS `#[graphql(visible = false)]` fields from
/// introspection while leaving them executable, so an introspection-sourced fence
/// could be defeated by a hidden mutation. SDL includes those fields, so no
/// visibility flag (and, via the AST parse, no brace-in-description quirk) can hide
/// a live mutation. The SDL field set must equal the registered set: a new field with
/// no row fails (a repo-write cannot slip in unlisted); a registered name absent from
/// the schema fails (renamed or removed). The set is also cross-checked against
/// introspection, so any field present in SDL but hidden from introspection fails
/// named as a forbidden hidden mutation.
///
/// The MARKER check is the only source-scraped part, and it is not the
/// completeness guarantee. Three honest limits, so a green check is not misread
/// as full authorization coverage:
/// 1. A present marker is not load-bearingness (the gate may not redden on a
/// hostile probe: the #203 present-but-vacuous case). The adversarial tests
/// in `graphql/mutation.rs` are the load-bearing layer.
/// 2. A present marker is not gate-correctness (it may be the wrong bucket for
/// the operation's resource: a repo-write registered as signer-self would
/// pass, so INV-1 review remains the defense).
/// 3. The marker check scrapes the comment-masked, brace-bounded resolver body;
/// a `{`/`}` or a `did_matches(` inside a STRING literal is not masked, so it
/// could mis-slice that body or satisfy the marker vacuously (the same
/// vacuous-marker class as limit 1). Completeness is unaffected: it comes
/// from the SDL AST, not from this scrape.
#[test]
fn every_graphql_mutation_has_its_gate() {
// (Rust snake name, schema field in camelCase, expected gate marker). The
// four are Bucket C signer-self (`did_matches(`); a repo-write would be
// Bucket A (`require_repo_owner(`). Same-named REST rows live in the guard
// above against `api/tasks.rs`.
let registered: &[(&str, &str, &str)] = &[
("create_task", "createTask", "did_matches("),
("claim_task", "claimTask", "did_matches("),
("complete_task", "completeTask", "did_matches("),
("fail_task", "failTask", "did_matches("),
];
// Completeness + hidden-mutation cross-check against the REAL schema (#219),
// delegated to the generic `assert_mutation_fence` so the SDL sourcing and the
// cross-check are ALSO exercised by a synthetic hidden schema in
// `fence_rejects_a_hidden_mutation` (INV-21). Because the real MutationRoot has no
// hidden field today, that synthetic test — not a hypothetical production canary —
// is what turns reverting the SDL source or deleting the cross-check RED.
{
use crate::graphql::mutation::MutationRoot;
use crate::graphql::query::QueryRoot;
use crate::graphql::subscription::SubscriptionRoot;
let schema =
async_graphql::Schema::build(QueryRoot, MutationRoot, SubscriptionRoot).finish();
let registered_camel: Vec<&str> = registered.iter().map(|(_, c, _)| *c).collect();
assert_mutation_fence(&schema, &registered_camel);
}
// Marker present (source scrape; the vacuity limit, backstopped by the
// adversarial tests): each resolver body carries its bucket marker.
let masked = mask_comments(include_str!("../graphql/mutation.rs"));
for (snake, _, marker) in registered {
let body = resolver_body(&masked, snake);
assert!(
body.contains(marker),
"GraphQL mutation `{snake}` is missing its gate marker `{marker}`: gate removed"
);
}
}
/// Version-safety self-check (#219, INV-21): the fence's completeness guarantee rests
/// on `schema.sdl()` INCLUDING `visible = false` fields while introspection omits
/// them. That is observed async-graphql behavior, not a documented contract, so pin
/// it here: a synthetic schema with a hidden mutation must have that field in its
/// SDL-derived set and NOT in its introspection-derived set. If a future
/// async-graphql upgrade ever filters SDL by visibility, the first assertion flips
/// RED — the assumption is load-bearing, not trusted.
#[test]
fn sdl_source_sees_hidden_mutations_that_introspection_omits() {
use async_graphql::{EmptySubscription, Object, Schema};
struct Q;
#[Object]
impl Q {
async fn ping(&self) -> i32 {
0
}
}
struct M;
#[Object]
impl M {
async fn visible_mut(&self) -> i32 {
1
}
#[graphql(visible = false)]
async fn hidden_mut(&self) -> i32 {
2
}
}
let schema = Schema::build(Q, M, EmptySubscription).finish();
// SDL source (the fix): includes the hidden field.
let sdl_fields = mutation_field_names_from_sdl(&schema.sdl());
assert!(
sdl_fields.iter().any(|f| f == "hiddenMut"),
"SDL-sourced set MUST include the visible=false mutation; if this is RED, \
async-graphql started filtering SDL by visibility and the fence's \
completeness guarantee is broken: {sdl_fields:?}"
);
// Introspection (the fence's own helper): OMITS the hidden field. This is the
// exact gap the fix closes — an introspection-sourced fence never sees `hiddenMut`.
let introspected = introspection_mutation_fields(&schema);
assert!(
!introspected.iter().any(|f| f == "hiddenMut"),
"introspection is expected to OMIT the visible=false mutation (that is the \
gap the SDL source closes); got {introspected:?}"
);
// The exact diff the fence computes: hidden ∈ (sdl \ introspection).
let hidden: Vec<&String> = sdl_fields
.iter()
.filter(|f| !introspected.iter().any(|i| i == *f))
.collect();
assert_eq!(
hidden,
vec!["hiddenMut"],
"the SDL-vs-introspection diff must surface exactly the hidden mutation"
);
}
/// INV-21 wiring guard (#219): drives the generic `assert_mutation_fence` — the SAME
/// helper the real fence calls — against a synthetic schema whose mutation is
/// `visible = false` and IS registered, so completeness passes and ONLY the
/// hidden-from-introspection cross-check can fail. This is what makes the fence's SDL
/// sourcing load-bearing without a production canary: reverting the source to
/// introspection (the hidden field then absent from the SDL set) or deleting the
/// cross-check removes this panic (or changes its message), turning the
/// `should_panic` test RED. The real MutationRoot has no hidden field, so nothing on
/// the real schema could catch that regression.
#[test]
#[should_panic(expected = "hidden from introspection")]
fn fence_rejects_a_hidden_mutation() {
use async_graphql::{EmptySubscription, Object, Schema};
struct Q;
#[Object]
impl Q {
async fn ping(&self) -> i32 {
0
}
}
struct M;
#[Object]
impl M {
async fn visible_mut(&self) -> i32 {
1
}
#[graphql(visible = false)]
async fn hidden_mut(&self) -> i32 {
2
}
}
let schema = Schema::build(Q, M, EmptySubscription).finish();
// Both fields registered, so completeness passes and only the cross-check fires.
assert_mutation_fence(&schema, &["visibleMut", "hiddenMut"]);
}
/// Deprecated-mutation guard (#219): a `#[graphql(deprecation = "...")]` mutation is
/// executable and present in SDL, but GraphQL introspection omits it by default. The
/// fence's introspection helper uses `includeDeprecated: true`, so a deprecated (but
/// visible) mutation is NOT mistaken for a hidden one. With it registered, the fence
/// must NOT panic. Remove `includeDeprecated: true` and this goes RED — the deprecated
/// field would look hidden and false-red as a forbidden visible=false mutation.
#[test]
fn fence_admits_a_deprecated_mutation() {
use async_graphql::{EmptySubscription, Object, Schema};
struct Q;
#[Object]
impl Q {
async fn ping(&self) -> i32 {
0
}
}
struct M;
#[Object]
impl M {
async fn visible_mut(&self) -> i32 {
1
}
#[graphql(deprecation = "use visibleMut")]
async fn deprecated_mut(&self) -> i32 {
2
}
}
let schema = Schema::build(Q, M, EmptySubscription).finish();
// Must not panic: the deprecated field is visible (present in introspection with
// includeDeprecated) and registered, so it is neither hidden nor unclassified.
assert_mutation_fence(&schema, &["visibleMut", "deprecatedMut"]);
}
/// Completeness + hidden-mutation cross-check for a BUILT schema (#219). Generic over
/// the schema type so the real fence AND the synthetic `fence_rejects_a_hidden_mutation`
/// / `fence_admits_a_deprecated_mutation` tests drive the SAME sourcing and cross-check.
/// Completeness is sourced from SDL (which includes `visible = false` fields); the
/// cross-check fails any SDL field hidden from introspection. This genericity is the
/// INV-21 fix: because the real MutationRoot has no hidden field, only a synthetic
/// hidden schema through THIS helper makes reverting the SDL source or removing the
/// cross-check RED, rather than leaving the class latent behind a production canary.
fn assert_mutation_fence<Q, M, S>(
schema: &async_graphql::Schema<Q, M, S>,
registered_camel: &[&str],
) where
Q: async_graphql::ObjectType + 'static,
M: async_graphql::ObjectType + 'static,
S: async_graphql::SubscriptionType + 'static,
{
let sdl_fields = mutation_field_names_from_sdl(&schema.sdl());
let introspected = introspection_mutation_fields(schema);
// No SDL field may be hidden from introspection (a visible=false mutation).
for f in &sdl_fields {
assert!(
introspected.iter().any(|i| i == f),
"GraphQL mutation `{f}` is in the schema but hidden from introspection \
(a `#[graphql(visible = false)]` field): hidden mutations are forbidden \
by this fence — make it visible so its classification is enforceable, or \
remove it"
);
}
// Completeness both directions against the registered (camelCase) set.
for f in &sdl_fields {
assert!(
registered_camel.contains(&f.as_str()),
"unclassified GraphQL mutation `{f}`: add it to \
every_graphql_mutation_has_its_gate with a deliberate gate \
(`require_repo_owner(` for a repo-write, `did_matches(` for a task op)"
);
}
for c in registered_camel {
assert!(
sdl_fields.iter().any(|f| f == c),
"registered GraphQL mutation `{c}` is not in the schema \
(renamed or removed); update the table"
);
}
}
/// The mutation field names via the engine's introspection, run against a passed
/// schema. STRUCTURED (names arrive as data). Uses `fields(includeDeprecated: true)`:
/// GraphQL's default omits `@deprecated` fields, but a `#[graphql(deprecation = "...")]`
/// mutation is still executable and present in SDL, so without this the cross-check
/// above would false-red a deprecated mutation as a hidden (visible=false) one. Only a
/// real visibility filter should trip that arm; deprecation is a lifecycle marker.
fn introspection_mutation_fields<Q, M, S>(
schema: &async_graphql::Schema<Q, M, S>,
) -> Vec<String>
where
Q: async_graphql::ObjectType + 'static,
M: async_graphql::ObjectType + 'static,
S: async_graphql::SubscriptionType + 'static,
{
use async_graphql::Value;
let resp = tokio::runtime::Runtime::new()
.expect("tokio runtime")
.block_on(schema.execute(
"{ __schema { mutationType { fields(includeDeprecated: true) { name } } } }",
));
assert!(
resp.errors.is_empty(),
"schema introspection failed: {:?}",
resp.errors
);
let mut names = Vec::new();
if let Value::Object(root) = &resp.data {
if let Some(Value::Object(sc)) = root.get("__schema") {
if let Some(Value::Object(mt)) = sc.get("mutationType") {
if let Some(Value::List(field_list)) = mt.get("fields") {
for field in field_list {
if let Value::Object(fo) = field {
if let Some(Value::String(n)) = fo.get("name") {
names.push(n.clone());
}
}
}
}
}
}
}
names
}
/// The mutation field names sourced from the built schema's SDL — the completeness
/// source (#219). Builds the same DB-free schema as the introspection helper and runs
/// its `.sdl()` (sync, no runtime needed) through [`mutation_field_names_from_sdl`].
/// SDL includes `#[graphql(visible = false)]` fields, so this source cannot be
/// defeated by hiding a mutation from introspection.
fn graphql_mutation_fields_from_sdl() -> Vec<String> {
use crate::graphql::mutation::MutationRoot;
use crate::graphql::query::QueryRoot;
use crate::graphql::subscription::SubscriptionRoot;
let schema =
async_graphql::Schema::build(QueryRoot, MutationRoot, SubscriptionRoot).finish();
mutation_field_names_from_sdl(&schema.sdl())
}
/// The mutation type's field names extracted from an async-graphql SDL string via
/// the engine's OWN parser AST (`async_graphql::parser::parse_schema`), NOT
/// introspection. This is the completeness source (#219): unlike introspection,
/// which async-graphql filters by `visible` (a `#[graphql(visible = false)]`
/// mutation is executable but ABSENT from introspection), `schema.sdl()` includes
/// every registered field, so a hidden mutation still appears here and must carry a
/// classified row. Parsing the AST (not a brace/line text scan) means a `}` inside a
/// field description cannot mis-slice the type block. camelCase, as async-graphql
/// emits. An empty result on a navigation miss is safe: the set-equality in
/// `every_graphql_mutation_has_its_gate` turns "no fields" into a loud failure for
/// every registered name, never a vacuous pass.
fn mutation_field_names_from_sdl(sdl: &str) -> Vec<String> {
use async_graphql::parser::parse_schema;
use async_graphql::parser::types::{TypeKind, TypeSystemDefinition};
let doc = parse_schema(sdl).expect("async-graphql SDL must parse");
// The mutation root type name: from the `schema { mutation: X }` block if
// present, else the GraphQL default "Mutation".
let mutation_type = doc
.definitions
.iter()
.find_map(|def| match def {
TypeSystemDefinition::Schema(s) => s
.node
.mutation
.as_ref()
.map(|m| m.node.as_str().to_string()),
_ => None,
})
.unwrap_or_else(|| "Mutation".to_string());
// The fields of that object type.
doc.definitions
.iter()
.find_map(|def| match def {
TypeSystemDefinition::Type(t) if t.node.name.node.as_str() == mutation_type => {
match &t.node.kind {
TypeKind::Object(obj) => Some(
obj.fields
.iter()
.map(|f| f.node.name.node.as_str().to_string())
.collect::<Vec<_>>(),
),
_ => None,
}
}
_ => None,
})
.unwrap_or_default()
}
#[test]
fn sdl_extractor_includes_a_hidden_visible_false_field() {
// A field that would be hidden from introspection must still be listed from SDL.
let sdl = "schema { query: QueryRoot mutation: MutationRoot }\n\
type QueryRoot { ping: Int! }\n\
type MutationRoot { createTask: Int! hiddenMut: Int! }\n";
let fields = mutation_field_names_from_sdl(sdl);
assert!(
fields.iter().any(|f| f == "createTask"),
"visible field must be listed: {fields:?}"
);
assert!(
fields.iter().any(|f| f == "hiddenMut"),
"a visible=false field (present in SDL, hidden from introspection) MUST be \
listed by the completeness source: {fields:?}"
);
}
#[test]
fn sdl_extractor_is_not_fooled_by_a_brace_in_a_field_description() {
// A `}` inside a field description must not truncate the type block (the AST
// parse is immune to the text-parse quirk the old design feared).
let sdl = "schema { query: QueryRoot mutation: MutationRoot }\n\
type QueryRoot { ping: Int! }\n\
type MutationRoot {\n\
\"\"\"a description with a } brace and { another\"\"\"\n\
createTask: Int!\n\
claimTask: Int!\n\
}\n";
let fields = mutation_field_names_from_sdl(sdl);
assert!(
fields.iter().any(|f| f == "createTask") && fields.iter().any(|f| f == "claimTask"),
"both fields must be extracted despite a brace in a description: {fields:?}"
);
}
#[test]
fn sdl_extractor_returns_empty_for_a_query_only_schema() {
let sdl = "schema { query: QueryRoot }\ntype QueryRoot { ping: Int! }\n";
assert!(
mutation_field_names_from_sdl(sdl).is_empty(),
"a schema with no mutation root yields no mutation fields (no panic)"
);
}
/// The comment-masked body of `async fn <snake>` in `graphql/mutation.rs`, bounded
/// to the resolver's OWN braces (its `{ .. }`), for the marker check. Brace-matching
/// on the comment-masked source means a `}` in a trailing helper cannot extend the
/// body and a comment marker does not count. A `{`/`}` inside a STRING literal is
/// not masked, a documented marker-check-only residual (limit 3); completeness does
/// not use this.
fn resolver_body(masked_src: &str, snake: &str) -> String {
let needle = format!("async fn {snake}(");
let start = masked_src.find(&needle).unwrap_or_else(|| {
panic!("resolver `async fn {snake}(` not found in graphql/mutation.rs")
});
let open = start
+ masked_src[start..]
.find('{')
.expect("resolver body opening brace");
let mut depth = 0i32;
let mut end = masked_src.len();
for (i, c) in masked_src[open..].char_indices() {
match c {
'{' => depth += 1,
'}' => {
depth -= 1;
if depth == 0 {
end = open + i + 1;
break;
}
}
_ => {}
}
}
masked_src[open..end].to_string()
}
/// `src` with `//` line comments and `/* */` block comments (nested) blanked to
/// spaces, WITHOUT opening a comment inside a `"..."` string (so a `"https://"` does
/// not erase the rest of its line). String CONTENTS are left intact, so a brace or
/// a gate marker inside a string is not masked; for the marker check that is a
/// documented vacuity residual (limit 3), and completeness does not use this at
/// all. Char literals and raw/byte strings are not tracked (absent in resolver
/// bodies); a `'"'` char or a raw string is the remaining narrow residual.
fn mask_comments(src: &str) -> String {
let chars: Vec<char> = src.chars().collect();
let mut out = String::with_capacity(src.len());
let mut i = 0;
let mut block_depth = 0usize;
let mut in_string = false;
while i < chars.len() {
let c = chars[i];
let next = chars.get(i + 1).copied();
if block_depth > 0 {
if c == '/' && next == Some('*') {
block_depth += 1;
out.push_str(" ");
i += 2;
} else if c == '*' && next == Some('/') {
block_depth -= 1;
out.push_str(" ");
i += 2;
} else {
out.push(if c == '\n' { '\n' } else { ' ' });
i += 1;
}
} else if in_string {
out.push(c);
if c == '\\' {
if let Some(n) = next {
out.push(n);
i += 2;
} else {
i += 1;
}
} else {
if c == '"' {
in_string = false;
}
i += 1;
}
} else if c == '"' {
in_string = true;
out.push('"');
i += 1;
} else if c == '/' && next == Some('/') {
while i < chars.len() && chars[i] != '\n' {
out.push(' ');
i += 1;
}
} else if c == '/' && next == Some('*') {
block_depth = 1;
out.push_str(" ");
i += 2;
} else {
out.push(c);
i += 1;
}
}
out
}
/// The SDL field set (the completeness source of truth, #219) is exactly the
/// registered mutations. If async-graphql adds or renames a field, this reddens
/// and forces the `every_graphql_mutation_has_its_gate` table to be updated. Uses
/// SDL, not introspection: introspection omits `visible = false` fields, so an
/// introspection-based pin here would go green while a hidden mutation existed.
#[test]
fn schema_lists_exactly_the_registered_mutations() {
let mut fields = graphql_mutation_fields_from_sdl();
fields.sort();
assert_eq!(
fields,
vec!["claimTask", "completeTask", "createTask", "failTask"],
"the SDL mutation field set is the authoritative discovered set"
);
}
/// The marker check is bounded to a resolver's OWN body and is comment- and
/// string-aware: a `//` inside a `"https://"` string does not erase a real gate on
/// that line (no false-red), and a trailing module helper's `did_matches(` does not
/// rescue a resolver whose own gate was removed (no false-green).
#[test]
fn resolver_body_is_bounded_and_comment_and_string_aware() {
let src = "\
impl MutationRoot {
async fn ok(&self) -> Result<u32> {
let _ = (\"see https://x/y\", did_matches(a, b)); // URL string then gate, one line
Ok(0)
}
async fn last(&self) -> Result<u32> {
// did_matches( only in a comment
Ok(0)
}
}
fn helper() { let _ = did_matches(z, z); }
#[cfg(test)]
mod tests { async fn t() { let _ = did_matches(x, y); } }
";
let masked = mask_comments(src);
assert!(
resolver_body(&masked, "ok").contains("did_matches("),
"a `//` inside a string must not erase the real gate on that line"
);
assert!(
!resolver_body(&masked, "last").contains("did_matches("),
"the marker check must be bounded to the resolver's own body (a trailing \
helper's marker must not rescue a resolver whose gate was removed)"
);
}
/// Proves the comment-stripping that GUARD-1 added: a marker that appears only
/// in a full-line comment (the real `replicas.rs` false-pass shape) must NOT
/// satisfy a row.
-34
View File
@@ -514,32 +514,9 @@ pub async fn ping_peer(
})))
}
/// The `x-gitlawb-error` code a peer put on a rejection, if it sent one.
///
/// Kept verbatim: it is what separates a ledger rejection from a replay
/// rejection on the same-ish status, and it is what an operator greps for. A
/// peer behind a proxy that strips unknown `X-` headers returns `None`, which
/// is why every caller logs the status too rather than relying on this alone.
pub(crate) fn peer_error_code(headers: &reqwest::header::HeaderMap) -> Option<&str> {
headers.get("x-gitlawb-error")?.to_str().ok()
}
#[cfg(test)]
mod tests {
use super::is_public_http_url;
use super::peer_error_code;
#[test]
fn peer_error_code_reads_the_header_when_present() {
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("x-gitlawb-error", "signature_ledger_full".parse().unwrap());
assert_eq!(peer_error_code(&headers), Some("signature_ledger_full"));
}
#[test]
fn peer_error_code_is_none_without_the_header() {
assert_eq!(peer_error_code(&reqwest::header::HeaderMap::new()), None);
}
#[test]
fn accepts_public_https_and_http() {
@@ -717,7 +694,6 @@ mod tests {
.layer(Extension(IpRateLimiter {
limiter: RateLimiter::new(60, Duration::from_secs(3600)),
trust: TrustedProxy::None,
skip_reads: false,
}))
.with_state(state);
let mut req =
@@ -766,7 +742,6 @@ mod tests {
.layer(Extension(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(3600)),
trust: TrustedProxy::None,
skip_reads: false,
}))
.with_state(state);
let peer = "203.0.113.5:5000".parse::<SocketAddr>().unwrap();
@@ -1452,15 +1427,6 @@ mod tests {
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
// The brake shares 429 with the ledger's `signature_ledger_full` on
// this route group, so the code is what tells a client which one it
// hit. Asserted end-to-end here, through the real router.
assert_eq!(
resp.headers()
.get("x-gitlawb-error")
.and_then(|v| v.to_str().ok()),
Some("rate_limited"),
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
+24 -787
View File
@@ -8,7 +8,6 @@ use std::sync::Arc;
use crate::auth::{caller_authorized_to_push, AuthenticatedDid};
use crate::db::RepoRecord;
use chrono::{DateTime, Utc};
use gitlawb_core::node_denial::NodeDenial;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
@@ -1910,114 +1909,10 @@ fn fork_withheld_blocks(
/// and as the signing path, so they can never drift apart.
const SYNC_NOTIFY_PATH: &str = "/api/v1/sync/notify";
/// How hard the sender retries a peer that says "not now".
///
/// Passed in rather than read from a constant so the tests can pin the attempt
/// counts without sleeping for real seconds (tokio's `start_paused` needs the
/// `test-util` feature, which this crate does not carry).
struct NotifyRetryPolicy {
/// Backoff before each retry. The length IS the retry count, so a ref gets
/// `backoff.len() + 1` attempts.
backoff: &'static [std::time::Duration],
/// Total time one peer's whole fan-out may spend sleeping between
/// retries, shared across every ref in the push.
///
/// A per-ref bound alone is not enough: a `git push --mirror` of 600
/// branches fans out 600 requests, so a per-ref budget multiplies by the
/// ref count and would keep the background task alive against a wedged
/// peer for hours. With a shared budget the whole fan-out adds at most
/// this much wall clock, however wide the push.
///
/// The tradeoff, stated plainly because it is not obvious from the code:
/// the budget is spent first-come, so on a peer rejecting every ref the
/// first ~10 failing refs consume all of it and every later ref gets a
/// single attempt with no retry at all. That is deliberate — retries are
/// worth most on a peer that is briefly rejecting and worth nothing on one
/// that is rejecting everything, and the alternative (dividing the budget
/// by the ref count) would make each share too small to ride out anything.
/// Refs that do not land are reported, not silently dropped.
budget: std::time::Duration,
}
/// Backoff before each retry of a retryable `/sync/notify` rejection.
const NOTIFY_RETRY_BACKOFF: [std::time::Duration; 2] = [
std::time::Duration::from_secs(1),
std::time::Duration::from_secs(5),
];
/// Production retry policy for the peer sync fan-out. 60s is the total sleep
/// budget shared across every peer in one push's fan-out, not per peer, so an
/// attacker-fillable peer set cannot multiply retry wall clock.
const NOTIFY_RETRY_POLICY: NotifyRetryPolicy = NotifyRetryPolicy {
backoff: &NOTIFY_RETRY_BACKOFF,
budget: std::time::Duration::from_secs(60),
};
/// What a peer's response status says the sender should do next.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NotifyDisposition {
Delivered,
/// The peer did not apply the update and said "not now". Re-sending is
/// safe because `sign_request` mints a fresh nonce per attempt, so the
/// retry carries a signature the peer's ledger has never seen.
Retryable,
/// Anything else, including 409 `signature_replayed`: the peer already
/// admitted this signed request, so re-sending risks a duplicate apply.
Fatal,
}
/// Classify a peer's `/sync/notify` response.
///
/// The gate is the peer's `X-Gitlawb-Error` code, not the status. Only two
/// codes mean "the write did not happen and another attempt can clear it":
/// `signature_ledger_full` (the per-identity spent-signature cap, which drains
/// as entries expire) and `signature_ledger_unavailable`. Everything else is
/// `Fatal`, including a 429 or 503 carrying any other code and one carrying no
/// code at all.
///
/// Status alone is not enough because the peer's per-client flood brake answers
/// 429 on this route too, and on a peer running the default config (where
/// `require_signed_peer_writes` is off) `/sync/notify` never reaches the ledger
/// at all — so every 429 the sender can see is the brake. Retrying that spends
/// the whole budget on the one class it can never recover from (the brake's
/// window is an hour, the budget is a minute) while pushing the sender further
/// into the peer's peer-write bucket, which `/peers/announce` shares.
///
/// The status still has to agree with the code, so a stray `signature_ledger_full`
/// header on a 409 cannot turn "the peer already applied this" into a retry.
/// The result of one `/sync/notify` attempt, with the reason kept for the
/// end-of-fan-out summary.
#[must_use]
enum NotifyAttempt {
Delivered,
Retryable(String),
Fatal(String),
}
fn classify_notify_status(status: reqwest::StatusCode, code: Option<&str>) -> NotifyDisposition {
if status.is_success() {
return NotifyDisposition::Delivered;
}
// `from_code` rather than string literals: this is the second consumer of
// the same wire vocabulary as `gl`, with the same drift exposure, and the
// enum is what makes a new code a compile error rather than a silent miss.
match code.and_then(NodeDenial::from_code) {
Some(denial @ (NodeDenial::LedgerFull | NodeDenial::LedgerUnavailable))
if status.as_u16() == denial.status() =>
{
NotifyDisposition::Retryable
}
_ => NotifyDisposition::Fatal,
}
}
/// Send one signed `/sync/notify` request for a single ref update.
///
/// The receiver is single-ref, so a multi-ref push fans out one request per
/// ref — each signed over its own body — carrying that ref's real `old_sha`.
///
/// One attempt only. Whether a failure is worth another attempt is the
/// caller's call, because the retry budget is shared across the whole fan-out.
#[allow(clippy::too_many_arguments)]
async fn notify_peer_of_ref(
http_client: &reqwest::Client,
@@ -2031,7 +1926,7 @@ async fn notify_peer_of_ref(
node_did: &str,
pusher_did: &str,
owner_did: &str,
) -> NotifyAttempt {
) {
let body = serde_json::json!({
"repo": repo_slug,
"ref_name": ref_name,
@@ -2046,7 +1941,7 @@ async fn notify_peer_of_ref(
Ok(bytes) => bytes,
Err(e) => {
tracing::warn!(peer = %peer_did, ref_name = %ref_name, err = %e, "failed to serialize peer sync notify");
return NotifyAttempt::Fatal(format!("body not serializable: {e}"));
return;
}
};
let signed =
@@ -2061,69 +1956,23 @@ async fn notify_peer_of_ref(
.send()
.await
{
Ok(r) if r.status().is_success() => {
tracing::info!(peer = %peer_did, repo = %repo_slug, ref_name = %ref_name, "notified peer to sync")
}
Ok(r) => {
let status = r.status();
let peer_code = crate::api::peers::peer_error_code(r.headers());
let code = peer_code.unwrap_or("none").to_string();
match classify_notify_status(status, peer_code) {
NotifyDisposition::Delivered => {
tracing::info!(peer = %peer_did, repo = %repo_slug, ref_name = %ref_name, "notified peer to sync");
NotifyAttempt::Delivered
}
NotifyDisposition::Retryable => {
tracing::warn!(peer = %peer_did, ref_name = %ref_name, status = %status, error_code = %code, "peer sync notify returned error");
NotifyAttempt::Retryable(format!("{status} {code}"))
}
NotifyDisposition::Fatal => {
tracing::warn!(peer = %peer_did, ref_name = %ref_name, status = %status, error_code = %code, "peer sync notify returned error");
NotifyAttempt::Fatal(format!("{status} {code}"))
}
}
tracing::warn!(peer = %peer_did, ref_name = %ref_name, status = %r.status(), "peer sync notify returned error")
}
// A transport error is never retried: the request may well have
// reached the peer and been applied, with only the response lost, and
// re-sending a mutation on that guess is exactly what the spent
// signature ledger exists to stop.
Err(e) => {
tracing::warn!(peer = %peer_did, ref_name = %ref_name, err = %e, "failed to notify peer");
NotifyAttempt::Fatal(format!("transport error: {e}"))
tracing::warn!(peer = %peer_did, ref_name = %ref_name, err = %e, "failed to notify peer")
}
}
}
/// What one peer's fan-out achieved. The refs that never landed are named
/// rather than merely counted, so the summary log and the tests are reading the
/// same fact: a ref the peer rejected past the bound is reported unfederated,
/// not quietly dropped.
#[derive(Debug, Default, PartialEq, Eq)]
struct FanOutSummary {
total: usize,
unfederated: Vec<String>,
/// Why the last failing ref failed, as `"<status> <code>"` or a transport
/// message. Empty when nothing failed.
last_reason: String,
}
/// Notify a single peer of every ref in a push — one request per ref.
///
/// Looping here (rather than sending one flattened request) is what keeps a
/// multi-ref push from collapsing to its first ref; each ref carries its real
/// `old_sha`.
///
/// Because every request in the fan-out is signed by the same node keypair, a
/// peer that requires signed peer writes charges them all to one identity, so a
/// ref the peer is briefly rejecting with `signature_ledger_full` is worth
/// another attempt: `sign_request` draws a fresh nonce per attempt, so the
/// retry carries a signature the peer has never seen rather than a replay.
///
/// What the retry does NOT solve is a wide push against a peer whose ledger cap
/// is genuinely exceeded (`git push --mirror` of hundreds of branches against a
/// few-hundred-row cap with minutes of retention). There the rejection clears
/// only when the window ages out, which is longer than the whole fan-out's
/// budget, so those refs do not federate however many times they are retried.
/// They are counted and reported in one summary line — a per-ref warning inside
/// a 600-ref loop is not something an operator will ever correlate back to the
/// push — and returned to the caller so a test can assert on them.
#[allow(clippy::too_many_arguments)]
async fn notify_peer_of_refs(
http_client: &reqwest::Client,
@@ -2135,71 +1984,22 @@ async fn notify_peer_of_refs(
node_did: &str,
pusher_did: &str,
owner_did: &str,
retry: &NotifyRetryPolicy,
shared_sleep_budget: &mut std::time::Duration,
) -> FanOutSummary {
let mut failed: Vec<&str> = Vec::new();
let mut last_reason = String::new();
) {
for (ref_name, old_sha, new_sha) in ref_updates {
let mut attempt = 0usize;
loop {
let outcome = notify_peer_of_ref(
http_client,
node_keypair,
peer_did,
notify_url,
repo_slug,
ref_name,
old_sha,
new_sha,
node_did,
pusher_did,
owner_did,
)
.await;
let reason = match outcome {
NotifyAttempt::Delivered => break,
NotifyAttempt::Fatal(reason) => reason,
NotifyAttempt::Retryable(reason) => {
// Two bounds, both required. `backoff` caps the attempts
// for this ref; `shared_sleep_budget` caps the sleeping for
// the whole fan-out across every peer and ref. Either one
// running out ends this ref here.
match retry.backoff.get(attempt) {
Some(&wait) if wait <= *shared_sleep_budget => {
*shared_sleep_budget -= wait;
attempt += 1;
tokio::time::sleep(wait).await;
continue;
}
_ => reason,
}
}
};
failed.push(ref_name);
last_reason = reason;
break;
}
}
if !failed.is_empty() {
tracing::warn!(
peer = %peer_did,
repo = %repo_slug,
failed = failed.len(),
total = ref_updates.len(),
first_failed_ref = %failed[0],
last_reason = %last_reason,
"refs failed to federate to peer; they will not be retried again"
);
}
FanOutSummary {
total: ref_updates.len(),
unfederated: failed.into_iter().map(str::to_string).collect(),
last_reason,
notify_peer_of_ref(
http_client,
node_keypair,
peer_did,
notify_url,
repo_slug,
ref_name,
old_sha,
new_sha,
node_did,
pusher_did,
owner_did,
)
.await;
}
}
@@ -3046,7 +2846,6 @@ async fn post_receive_replication_tail(
// anchoring above; this is the lowest-priority best-effort step.
if announce {
if let Ok(peers) = db_for_peers.list_peers().await {
let mut shared_retry_sleep = NOTIFY_RETRY_POLICY.budget;
for peer in peers {
if peer.http_url.is_empty() {
continue;
@@ -3068,8 +2867,6 @@ async fn post_receive_replication_tail(
&node_did_str,
&pusher_did_clone,
&record.owner_did,
&NOTIFY_RETRY_POLICY,
&mut shared_retry_sleep,
)
.await;
}
@@ -3546,36 +3343,6 @@ mod tests {
const OWNER_SHORT: &str = "z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH";
const STRANGER_DID: &str = "did:key:z6Mkffonly5tranger0000000000000000000000000000000";
#[allow(clippy::too_many_arguments)]
async fn notify_peer_with_budget(
http_client: &reqwest::Client,
node_keypair: &Keypair,
peer_did: &str,
notify_url: &str,
repo_slug: &str,
ref_updates: &[(String, String, String)],
node_did: &str,
pusher_did: &str,
owner_did: &str,
retry: &NotifyRetryPolicy,
) -> FanOutSummary {
let mut sleep_budget = retry.budget;
notify_peer_of_refs(
http_client,
node_keypair,
peer_did,
notify_url,
repo_slug,
ref_updates,
node_did,
pusher_did,
owner_did,
retry,
&mut sleep_budget,
)
.await
}
#[test]
fn upload_pack_request_finalizes_only_with_done_pktline() {
let want = "0032want 1111111111111111111111111111111111111111\n";
@@ -4810,7 +4577,7 @@ mod tests {
(ref_b.to_string(), old_b.to_string(), new_b.to_string()),
];
notify_peer_with_budget(
notify_peer_of_refs(
&http_client,
&keypair,
"did:key:zPeer",
@@ -4820,7 +4587,6 @@ mod tests {
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&NOTIFY_RETRY_POLICY,
)
.await;
@@ -4859,7 +4625,7 @@ mod tests {
new_sha.to_string(),
)];
notify_peer_with_budget(
notify_peer_of_refs(
&http_client,
&keypair,
"did:key:zPeer",
@@ -4869,541 +4635,12 @@ mod tests {
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&NOTIFY_RETRY_POLICY,
)
.await;
_mock.assert_async().await;
}
// ── retry of a retryable peer rejection ──────────────────────────────
//
// A peer running signed peer writes charges every /sync/notify against a
// per-identity spent-signature ledger. A wide push fans out one signed
// request per ref under the same node identity, so the tail of a large
// push hits the cap and gets 429 signature_ledger_full, a retryable rate
// condition. Without a sending-side retry those refs never federate.
//
// The tests drive a millisecond-scale policy so they pin the attempt
// counts without sleeping for real seconds; the production numbers are
// pinned separately by notify_retry_policy_bound_is_what_we_think.
const FAST_BACKOFF: [std::time::Duration; 2] = [
std::time::Duration::from_millis(5),
std::time::Duration::from_millis(10),
];
/// Same shape as the production policy (two retries), budgeted for exactly
/// two fully retried refs.
const FAST_POLICY: NotifyRetryPolicy = NotifyRetryPolicy {
backoff: &FAST_BACKOFF,
budget: std::time::Duration::from_millis(30),
};
fn refs(n: usize) -> Vec<(String, String, String)> {
(0..n)
.map(|i| {
(
format!("refs/heads/b{i}"),
format!("{i:040x}"),
format!("{:040x}", i + 100),
)
})
.collect()
}
// The production bound is a real operational number, so state it once here
// rather than leave it implied by constants nobody reads.
#[test]
fn notify_retry_policy_bound_is_what_we_think() {
assert_eq!(
NOTIFY_RETRY_BACKOFF.len() + 1,
3,
"one send plus two retries per ref"
);
assert_eq!(NOTIFY_RETRY_POLICY.backoff, &NOTIFY_RETRY_BACKOFF);
let per_ref: u64 = NOTIFY_RETRY_BACKOFF.iter().map(|d| d.as_secs()).sum();
assert_eq!(per_ref, 6, "1s + 5s of backoff per retried ref");
assert_eq!(
NOTIFY_RETRY_POLICY.budget.as_secs(),
60,
"the whole fan-out across every peer adds at most one minute of retry sleep"
);
}
// Retry sleep budget is shared across peers in one push fan-out, not minted
// fresh per announced peer.
#[tokio::test]
async fn notify_fanout_retry_budget_is_shared_across_peers() {
let mut server_a = mockito::Server::new_async().await;
let mut server_b = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let mock_a = server_a
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(429)
.with_header("x-gitlawb-error", "signature_ledger_full")
.expect(6)
.create_async()
.await;
let mock_b = server_b
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(429)
.with_header("x-gitlawb-error", "signature_ledger_full")
.expect(1)
.create_async()
.await;
let mut shared_sleep = FAST_POLICY.budget;
notify_peer_of_refs(
&http_client,
&keypair,
"did:key:zPeerA",
&format!("{}{SYNC_NOTIFY_PATH}", server_a.url()),
"owner/repo",
&refs(2),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
&mut shared_sleep,
)
.await;
notify_peer_of_refs(
&http_client,
&keypair,
"did:key:zPeerB",
&format!("{}{SYNC_NOTIFY_PATH}", server_b.url()),
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
&mut shared_sleep,
)
.await;
mock_a.assert_async().await;
mock_b.assert_async().await;
}
// 429 twice then 200: a peer whose ledger rejection clears within the
// budget does federate, and the attempt count is pinned so the bound
// cannot silently grow.
//
// This proves the TRANSIENT case only. It is not evidence for the wide
// push that motivated the retry: there the cap is exceeded for as long as
// the retention window, which outlasts the whole fan-out's budget, and
// `notify_peer_reports_refs_a_permanently_429_peer_never_took` is what
// pins that outcome.
#[tokio::test]
async fn notify_peer_retries_a_transient_ledger_rejection_and_federates_on_retry() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let rejected = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(429)
.with_header("x-gitlawb-error", "signature_ledger_full")
.expect(2)
.create_async()
.await;
// Created second, so it only serves once the 429 mock has taken its
// two hits; it also catches any attempt beyond the third.
let accepted = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(200)
.expect(1)
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
rejected.assert_async().await;
accepted.assert_async().await;
}
// The wide-push shape the retry cannot fix: a peer whose ledger stays at
// its cap answers 429 for the whole fan-out. Two things must hold. It stops
// at the bound rather than spinning (an unbounded retry never terminates,
// and an off-by-one shows up in the count), and the ref it never delivered
// comes back named as unfederated rather than being dropped on the floor.
#[tokio::test]
async fn notify_peer_reports_refs_a_permanently_429_peer_never_took() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let mock = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(429)
.with_header("x-gitlawb-error", "signature_ledger_full")
.expect(FAST_BACKOFF.len() + 1)
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
let summary = notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
assert_eq!(summary.total, 1);
assert_eq!(
summary.unfederated,
vec!["refs/heads/b0".to_string()],
"a ref the peer never took must be reported unfederated, not dropped",
);
assert!(
summary.last_reason.contains("signature_ledger_full"),
"the reason must name the peer's code, got: {}",
summary.last_reason,
);
}
// The retry budget is shared across the whole per-peer fan-out, so a
// wedged peer cannot multiply the bound by the ref count: a 600-ref mirror
// push must not cost 600 times the per-ref retry budget.
#[tokio::test]
async fn notify_peer_retry_budget_is_shared_across_the_fan_out() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
// Each fully retried ref spends 5ms + 10ms of the 30ms budget, so the
// first two refs get three attempts each and the third, with the
// budget gone, gets a single attempt.
let per_ref: u128 = FAST_BACKOFF.iter().map(|d| d.as_millis()).sum();
let budgeted = (FAST_POLICY.budget.as_millis() / per_ref) as usize;
assert_eq!(budgeted, 2, "test constants drifted");
let mock = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(429)
.with_header("x-gitlawb-error", "signature_ledger_full")
.expect(budgeted * (FAST_BACKOFF.len() + 1) + 1)
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(budgeted + 1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
}
// A peer's per-IP flood brake also answers 429, and the sender can never
// clear it inside the retry budget (the brake's window is an hour). Only
// the ledger's own 429 is worth another attempt, so a 429 carrying any
// other code — or no code at all — is a single attempt.
#[tokio::test]
async fn notify_peer_does_not_retry_a_429_with_an_unrelated_code() {
for headers in [
vec![("x-gitlawb-error", "rate_limited")],
vec![("x-gitlawb-error", "some_code_from_a_newer_node")],
vec![],
] {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let mut m = server.mock("POST", SYNC_NOTIFY_PATH).with_status(429);
for (k, v) in &headers {
m = m.with_header(*k, v);
}
let mock = m.expect(1).create_async().await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
}
}
// Same rule on the other retryable status: only the ledger's own 503 is
// worth another attempt.
#[tokio::test]
async fn notify_peer_does_not_retry_a_503_with_an_unrelated_code() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let mock = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(503)
.expect(1)
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
}
// 409 signature_replayed means the peer ALREADY admitted this signed
// request. Re-sending risks a duplicate apply, which is what the ledger
// exists to prevent, so it must stay a single attempt.
#[tokio::test]
async fn notify_peer_does_not_retry_a_409_replay_rejection() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let mock = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(409)
.with_header("x-gitlawb-error", "signature_replayed")
.expect(1)
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
}
// A transport error may mean the peer applied the update and only the
// response was lost, so it is never retried. There is no mock to count
// here, so the assertion is wall clock against a backoff long enough that
// a single retry could not hide inside it.
#[tokio::test]
async fn notify_peer_does_not_retry_a_transport_error() {
const SLOW_BACKOFF: [std::time::Duration; 2] = [
std::time::Duration::from_secs(5),
std::time::Duration::from_secs(5),
];
const SLOW_POLICY: NotifyRetryPolicy = NotifyRetryPolicy {
backoff: &SLOW_BACKOFF,
budget: std::time::Duration::from_secs(60),
};
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
// Port 1 is reserved and nothing listens there: connect refused.
let notify_url = format!("http://127.0.0.1:1{SYNC_NOTIFY_PATH}");
let started = std::time::Instant::now();
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&refs(1),
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&SLOW_POLICY,
)
.await;
assert!(
started.elapsed() < std::time::Duration::from_secs(2),
"a transport error slept on the backoff, so it was retried: {:?}",
started.elapsed(),
);
}
// The happy path must not gain a single extra request: N refs, N POSTs.
#[tokio::test]
async fn notify_peer_sends_exactly_one_request_per_ref_on_success() {
let mut server = mockito::Server::new_async().await;
let keypair = Keypair::generate();
let http_client = reqwest::Client::new();
let ref_updates = refs(3);
let mock = server
.mock("POST", SYNC_NOTIFY_PATH)
.with_status(200)
.expect(ref_updates.len())
.create_async()
.await;
let notify_url = format!("{}{SYNC_NOTIFY_PATH}", server.url());
notify_peer_with_budget(
&http_client,
&keypair,
"did:key:zPeer",
&notify_url,
"owner/repo",
&ref_updates,
"did:key:zNode",
"did:key:zPusher",
"did:key:zOwner",
&FAST_POLICY,
)
.await;
mock.assert_async().await;
}
#[test]
fn classify_notify_status_separates_retryable_from_fatal() {
use reqwest::StatusCode;
assert_eq!(
classify_notify_status(StatusCode::OK, None),
NotifyDisposition::Delivered
);
assert_eq!(
classify_notify_status(StatusCode::TOO_MANY_REQUESTS, Some("signature_ledger_full")),
NotifyDisposition::Retryable,
"the ledger's own 429 is the retryable rate condition"
);
assert_eq!(
classify_notify_status(
StatusCode::SERVICE_UNAVAILABLE,
Some("signature_ledger_unavailable")
),
NotifyDisposition::Retryable,
"503 signature_ledger_unavailable did not apply the write"
);
assert_eq!(
classify_notify_status(StatusCode::CONFLICT, Some("signature_replayed")),
NotifyDisposition::Fatal,
"409 signature_replayed means the peer already applied it"
);
assert_eq!(
classify_notify_status(StatusCode::FORBIDDEN, None),
NotifyDisposition::Fatal
);
assert_eq!(
classify_notify_status(StatusCode::BAD_REQUEST, None),
NotifyDisposition::Fatal
);
}
// The gate is the code, not the status. Every one of these is a 429 or a
// 503 — retryable under the old status-only rule — and every one must be
// fatal.
#[test]
fn classify_notify_status_gates_on_the_code_not_the_status() {
use reqwest::StatusCode;
for (status, code) in [
// The peer's per-IP flood brake: an hour-long window the sender
// cannot clear inside a 60s budget.
(StatusCode::TOO_MANY_REQUESTS, Some("rate_limited")),
(StatusCode::SERVICE_UNAVAILABLE, Some("rate_limited")),
// No code at all — a bare status, or a proxy that stripped the
// header. Absence is not evidence of a ledger rejection.
(StatusCode::TOO_MANY_REQUESTS, None),
(StatusCode::SERVICE_UNAVAILABLE, None),
// A code from some other node, or a newer one this build predates.
(StatusCode::TOO_MANY_REQUESTS, Some("")),
(StatusCode::TOO_MANY_REQUESTS, Some("upstream_overloaded")),
(
StatusCode::SERVICE_UNAVAILABLE,
Some("signature_ledger_full"), // right code, wrong status
),
] {
assert_eq!(
classify_notify_status(status, code),
NotifyDisposition::Fatal,
"{status} with code {code:?} must not be retried",
);
}
}
// A 409 carries "the peer already admitted this signed request", so it is
// fatal no matter what code rides along — including a header claiming the
// ledger is full, which a hostile or confused peer could send to talk the
// sender into re-applying a mutation.
#[test]
fn classify_notify_status_never_retries_a_409() {
use reqwest::StatusCode;
for code in [
None,
Some("signature_replayed"),
Some("signature_ledger_full"),
Some("signature_ledger_unavailable"),
Some("rate_limited"),
] {
assert_eq!(
classify_notify_status(StatusCode::CONFLICT, code),
NotifyDisposition::Fatal,
"409 with code {code:?} must never be retried",
);
}
}
#[tokio::test]
async fn to_response_generates_correct_clone_url_slug() {
let state = crate::test_support::test_state_lazy();
+2 -541
View File
@@ -1,16 +1,14 @@
use axum::body::Body;
use axum::extract::{Request, State};
use axum::http::{Method, StatusCode};
use axum::http::StatusCode;
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::Json;
use http_body_util::BodyExt;
use serde_json::json;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use gitlawb_core::did::Did;
use gitlawb_core::node_denial::NodeDenial;
use gitlawb_core::ucan::Ucan;
use crate::state::AppState;
@@ -19,39 +17,6 @@ use crate::state::AppState;
#[derive(Clone, Debug)]
pub struct AuthenticatedDid(pub String);
/// The canonical identity of a verified HTTP signature, published into request
/// extensions by [`require_signature`] so a spent-signature ledger can be keyed
/// without re-parsing headers.
///
/// `signing_string_hash` is a hex SHA-256 of the *reconstructed* signing string,
/// never of the raw `Signature` header. `HttpSignature::parse` trims, and the
/// header is not a covered component, so whitespace variants are distinct header
/// bytes that reconstruct to one signing string; hashing the reconstruction is
/// what collapses those variants to a single ledger key. The signing string also
/// embeds `keyid` and `created` through its `@signature-params` line, so the
/// hash cannot collide across DIDs.
#[derive(Clone, Debug)]
pub struct SignatureIdentity {
/// The signing DID, as it appeared in the `keyid` parameter. Human-readable
/// and good for logs; NOT an identity key — see [`Self::key_fingerprint`].
pub keyid: String,
/// Hex of the 32 resolved Ed25519 public key bytes: the canonical identity.
///
/// `Did` stores the wire string verbatim while `to_verifying_key` resolves
/// it through `multibase::decode`, which accepts any multibase prefix. One
/// keypair therefore has many valid `did:key` spellings (`z6Mk…` base58btc,
/// `f…` base16, `m…` base64url, and more) that are all distinct strings and
/// all resolve to the same key. Anything counting per identity by string
/// equality hands that keypair a fresh budget per spelling, so per-identity
/// accounting keys on this instead.
pub key_fingerprint: String,
/// The `nonce` parameter, absent on signatures from a pre-nonce signer.
/// Present but short is not the same as unique: see [`unique_nonce`].
pub nonce: Option<String>,
/// Hex SHA-256 of the reconstructed signing string: exactly 64 characters.
pub signing_string_hash: String,
}
/// Whether `caller` is authorized to push to `record`.
///
/// Phase 1 (`GITLAWB_ENFORCE_OWNER_PUSH`): owner-only, via the canonical
@@ -66,7 +31,6 @@ pub fn caller_authorized_to_push(record: &crate::db::RepoRecord, caller: &str) -
use gitlawb_core::http_sig::{
build_signing_string, compute_content_digest, HttpSignature, COVERED_COMPONENTS,
MAX_FUTURE_SKEW_SECS, MAX_SIGNATURE_AGE_SECS,
};
use gitlawb_core::identity::verify;
@@ -74,7 +38,7 @@ use gitlawb_core::identity::verify;
///
/// Every write request must carry:
/// Content-Digest: sha-256=:base64hash:
/// Signature-Input: sig1=("@method" "@path" "content-digest");keyid="did:key:...";alg="ed25519";created=<unix>;nonce="..."
/// Signature-Input: sig1=("@method" "@path" "content-digest");keyid="did:key:...";alg="ed25519";created=<unix>
/// Signature: sig1=:base64signature:
///
/// The middleware:
@@ -278,15 +242,6 @@ pub async fn require_signature(request: Request, next: Next) -> Response {
request
.extensions_mut()
.insert(AuthenticatedDid(sig.key_id.to_string()));
// Publish the canonical ledger key while the verified signing string is still
// in hand. Hashing `signing_string` (the reconstruction that was just verified
// against) rather than any raw header is deliberate: see `SignatureIdentity`.
request.extensions_mut().insert(SignatureIdentity {
keyid: sig.key_id.to_string(),
key_fingerprint: hex::encode(verifying_key.to_bytes()),
nonce: sig.nonce.clone(),
signing_string_hash: hex::encode(Sha256::digest(signing_string.as_bytes())),
});
next.run(request).await
}
@@ -411,251 +366,6 @@ pub async fn require_ucan_chain(
next.run(request).await
}
/// How long a spent signature stays in the ledger, in seconds.
///
/// Retention must cover the whole window in which the signature would still be
/// accepted, or a key is evicted while its signature still passes the time
/// check. `check_created` accepts a request when `created` falls in
/// `[now - 300, now + 30]`, so for one fixed signature the acceptable *arrival*
/// window is `[created - 30, created + 300]`: 330 seconds wide. Expiry is
/// computed from arrival rather than from `created`, which can only over-retain
/// (an early arrival is charged the full 330s from when it landed), never
/// under-retain.
///
/// 330 + [`CROSS_INSTANCE_SKEW_MARGIN_SECS`] = 390. The margin is not slack:
/// single-node a flush 330 is exactly right (at `now = created + 300` the sweep
/// predicate `expires_at < now` is false so the row survives, and at
/// `created + 301` the signature is already too old), but the sweep runs on
/// whichever instance's timer fires, using ITS clock. With instance B running
/// `d` seconds ahead of A, a flush TTL has B delete the row at true time
/// `created + 300 - d` while A still accepts the replay until `created + 300`,
/// a replay window of exactly `d`. 60s covers the disagreement an NTP-synced
/// fleet can actually reach.
///
/// Derived rather than written out so it cannot drift from the window it has to
/// cover: widening either skew bound in `gitlawb-core` carries the TTL with it.
const SIGNATURE_LEDGER_TTL_SECS: i64 =
MAX_SIGNATURE_AGE_SECS + MAX_FUTURE_SKEW_SECS + CROSS_INSTANCE_SKEW_MARGIN_SECS;
/// Clock disagreement between two instances that the ledger TTL must absorb.
const CROSS_INSTANCE_SKEW_MARGIN_SECS: i64 = 60;
/// The shortest nonce this node will treat as a unique ledger key.
///
/// `HttpSignature::parse` maps the raw parameter, so `nonce=""` arrives as
/// `Some("")` rather than `None`. Length is the only property a verifier can
/// check (entropy is not observable), so it is the floor: 16 characters is 64
/// bits even on the weakest plausible alphabet, hex at 4 bits per character. At
/// the enforced ceiling of `MAX_LIVE_SIGNATURES_PER_KEYID` (512) live rows per
/// identity, the birthday probability of an accidental collision within one
/// identity is about `512^2 / 2^65`, roughly 1e-14.
///
/// Set below the 32 hex characters (128 bits) our own `sign_request` emits so
/// the floor costs no upgrade churn, yet still admits a third-party client
/// signing a 22-character base64 UUID or a 16-character hex draw.
const MIN_NONCE_CHARS: usize = 16;
/// The nonce, but only when it is long enough to stand in as a unique key.
///
/// Without this filter an empty nonce is `Some("")`, which both satisfies the
/// staged `require a nonce` flag (whose stated purpose is that every client
/// signs one) and puts [`ledger_key`] on its `(key, nonce)` arm with a value
/// that is CONSTANT per identity: every mutation from that keypair would
/// collapse onto one ledger key, so the first one in a retention window would
/// succeed and every later one would be refused as a replay.
fn unique_nonce(identity: &SignatureIdentity) -> Option<&str> {
identity
.nonce
.as_deref()
.filter(|nonce| nonce.chars().count() >= MIN_NONCE_CHARS)
}
/// The ledger key for a verified signature: always a 64-character hex SHA-256,
/// which is what the `consumed_signatures` CHECK constraint requires.
///
/// Two disjoint schemes, kept apart by a domain tag so a nonce key can never
/// collide with a signing-string key:
/// * with a nonce of usable width, `(key_fingerprint, nonce)` — short,
/// fixed-width, and it lets two legitimately identical requests be told
/// apart;
/// * otherwise the signing-string hash, which is canonical by construction
/// and collapses whitespace variants of the `Signature` header onto one key.
///
/// A too-short nonce takes the second arm rather than the first. That is the
/// safe direction: the hash arm is unique by construction, so a client with a
/// weak nonce loses only the ability to repeat byte-identical requests inside
/// one second, whereas trusting the nonce would collapse its whole traffic onto
/// one key. The fingerprint rather than `keyid` keeps the nonce arm on the
/// resolved key, so two spellings of one DID cannot reuse a nonce.
fn ledger_key(identity: &SignatureIdentity) -> String {
let mut hasher = Sha256::new();
match unique_nonce(identity) {
Some(nonce) => {
hasher.update(b"gitlawb/sig-nonce\x00");
hasher.update(identity.key_fingerprint.as_bytes());
hasher.update(b"\x00");
hasher.update(nonce.as_bytes());
}
None => {
hasher.update(b"gitlawb/sig-string\x00");
hasher.update(identity.signing_string_hash.as_bytes());
}
}
hex::encode(hasher.finalize())
}
/// Refuse a signed write with one of the shared ledger denial codes.
///
/// The code and status both come from [`NodeDenial`], which `gl` matches
/// exhaustively, so the two halves cannot drift: adding a variant there is a
/// compile error in the client until it handles it.
fn ledger_rejection(denial: NodeDenial, message: &str) -> Response {
let code = denial.as_str();
(
// Infallible: every `NodeDenial::status` is a real HTTP status,
// which `denial_statuses_are_valid_http_codes` proves over every
// variant.
StatusCode::from_u16(denial.status()).expect("NodeDenial status is a valid HTTP code"),
[("X-Gitlawb-Error", code)],
Json(json!({ "error": code, "message": message })),
)
.into_response()
}
/// Axum middleware that spends a verified HTTP signature exactly once.
///
/// Layer it so it runs *after* [`require_signature`] (which publishes the
/// [`SignatureIdentity`] read here) and after [`require_ucan_chain`], but before
/// the handler. Consuming last means only a request that cleared every auth
/// check spends its signature, so a valid signature paired with a rejected UCAN
/// can be retried with the same bytes; consuming before the handler is what
/// closes the concurrent-replay race.
pub async fn consume_signature(
State(state): State<AppState>,
request: Request,
next: Next,
) -> Response {
// Skip reads. `write_routes` chains `PUT/DELETE/GET` on one path
// (`/api/v1/repos/{owner}/{repo}/visibility`, `server.rs`), so
// `list_visibility` is served by a router inside `add_auth_layers` and `gl`
// drives it through `get_signed`. Ledgering there would put a database write
// on the signed read path, which R7 forbids. A method check is the general
// form of that exclusion and needs no router split: replaying a read has no
// side effect to spend.
if matches!(*request.method(), Method::GET | Method::HEAD) {
crate::metrics::record_signature_ledger("skipped_read");
return next.run(request).await;
}
let identity = match request.extensions().get::<SignatureIdentity>() {
Some(identity) => identity.clone(),
None => {
crate::metrics::record_signature_ledger("identity_missing");
// Fail closed. If this passed the request through, a wrong layer
// order would silently delete the replay defense while every test
// that exercises the correct stack kept passing.
tracing::error!(
path = %request.uri().path(),
"signature ledger reached without a verified SignatureIdentity — check the layer order",
);
return ledger_rejection(
NodeDenial::IdentityMissing,
"the request reached the signature ledger without a verified identity",
);
}
};
// Staged rollout of R6: once every client signs a nonce, an operator closes
// the signing-string fallback here. This runs after the method skip above,
// so it never reaches a signed read, and before the ledger is charged, so a
// refused request spends nothing.
if state.config.require_signature_nonce && unique_nonce(&identity).is_none() {
// A present-but-short nonce gets its own code. It is a different
// client bug from a pre-nonce signer (the client already emits the
// parameter, it just does not fill it), and the two need different
// instructions.
return match identity.nonce {
None => {
crate::metrics::record_signature_ledger("nonce_required");
tracing::warn!(did = %identity.keyid, "rejected a nonce-less signature: a nonce is required");
ledger_rejection(
NodeDenial::NonceRequired,
"this node requires a `nonce` parameter in Signature-Input — upgrade your client",
)
}
Some(nonce) => {
crate::metrics::record_signature_ledger("nonce_too_short");
tracing::warn!(
did = %identity.keyid,
len = nonce.chars().count(),
"rejected a signature whose nonce is too short to be unique",
);
ledger_rejection(
NodeDenial::NonceTooShort,
&format!(
"the `nonce` parameter in Signature-Input must be at least \
{MIN_NONCE_CHARS} characters drawn from a CSPRNG — \
`gl` signs 32 hex characters",
),
)
}
};
}
let key = ledger_key(&identity);
let now = chrono::Utc::now().timestamp();
// Charge the cap against the RESOLVED key, never the wire DID: see
// `SignatureIdentity::key_fingerprint`. This does not weaken single-use,
// which never depended on the identity column — a replay has to reproduce
// the signed bytes, and `keyid` sits inside `@signature-params`, so
// re-spelling the DID changes the signing string and needs a fresh
// signature. It fixes the per-identity cap only.
match state
.db
.consume_signature(
&key,
&identity.key_fingerprint,
now,
now + SIGNATURE_LEDGER_TTL_SECS,
)
.await
{
Ok(crate::db::ConsumeSignature::Inserted) => {
crate::metrics::record_signature_ledger("admitted");
next.run(request).await
}
Ok(crate::db::ConsumeSignature::Replayed) => {
crate::metrics::record_signature_ledger("replayed");
tracing::warn!(did = %identity.keyid, "rejected a replayed HTTP signature");
ledger_rejection(
NodeDenial::Replayed,
"this signature has already been used — sign a fresh request",
)
}
Ok(crate::db::ConsumeSignature::IdentityLedgerFull) => {
// A rate condition, not a permanent rejection: the caller's live
// rows drain as they expire, so this is retryable.
crate::metrics::record_signature_ledger("identity_ledger_full");
tracing::warn!(did = %identity.keyid, "signature ledger full for this identity");
ledger_rejection(
NodeDenial::LedgerFull,
"too many unexpired signatures for this identity — retry shortly",
)
}
Err(e) => {
// Fail closed (KTD5). An outage is exactly when a holder of a
// captured signature would try, and the mutation handlers all need
// the same database anyway.
crate::metrics::record_signature_ledger("unavailable");
tracing::error!(did = %identity.keyid, err = %e, "signature ledger unavailable");
ledger_rejection(
NodeDenial::LedgerUnavailable,
"the signature ledger is unavailable — retry shortly",
)
}
}
}
fn human_detected(message: &str) -> impl IntoResponse {
(
StatusCode::UNAUTHORIZED,
@@ -699,39 +409,6 @@ mod tests {
.unwrap()
}
/// `ledger_rejection` converts `NodeDenial::status` with an `expect`.
/// This is what makes that infallible: every variant is a real HTTP status,
/// and it is the status the node has always paired with that code.
#[test]
fn denial_statuses_are_valid_http_codes() {
for denial in NodeDenial::ALL {
let status = StatusCode::from_u16(denial.status())
.unwrap_or_else(|e| panic!("{denial} has an invalid status: {e}"));
assert_eq!(status.as_u16(), denial.status());
}
}
/// The exact bytes `consume_signature` puts on the wire for each denial:
/// the `X-Gitlawb-Error` header, the body's `error` field, and the status
/// must all agree, because `gl` keys on the header and `gl init` keys on the
/// body field.
#[tokio::test]
async fn ledger_rejection_emits_the_code_in_both_the_header_and_the_body() {
for denial in NodeDenial::ALL {
let resp = ledger_rejection(denial, "why it was refused");
assert_eq!(resp.status().as_u16(), denial.status(), "{denial}");
assert_eq!(
resp.headers().get("X-Gitlawb-Error").unwrap(),
denial.as_str(),
"{denial}",
);
let body = resp.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"], denial.as_str(), "{denial}");
assert_eq!(json["message"], "why it was refused", "{denial}");
}
}
#[test]
fn validate_ucan_chain_valid() {
let node = Keypair::generate();
@@ -853,7 +530,6 @@ mod tests {
push_limiter_trust: crate::rate_limit::TrustedProxy::None,
sync_trigger_rate_limiter: RateLimiter::new(60, Duration::from_secs(3600)),
peer_write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
signed_write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
shutdown_tx: tokio::sync::watch::channel(false).0,
git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(64)),
git_write_semaphore: Arc::new(tokio::sync::Semaphore::new(64)),
@@ -873,221 +549,6 @@ mod tests {
}
}
/// The ledger TTL must keep a spent signature past the last instant any
/// instance would still accept it, INCLUDING one whose clock disagrees.
/// The sweep runs on whichever instance's timer fires, using its own clock,
/// so a TTL flush against the acceptance window lets an instance running
/// ahead delete a row that a lagging instance would still honour.
#[test]
fn ledger_ttl_keeps_a_margin_over_the_acceptance_window() {
// For one fixed signature the arrival window is
// `[created - MAX_FUTURE_SKEW_SECS, created + MAX_SIGNATURE_AGE_SECS]`.
let arrival_window = MAX_SIGNATURE_AGE_SECS + MAX_FUTURE_SKEW_SECS;
assert_eq!(arrival_window, 330, "the window this TTL is derived from");
// The floor is written out rather than read from
// `CROSS_INSTANCE_SKEW_MARGIN_SECS`, so zeroing that constant is caught
// here instead of quietly satisfying the comparison against itself.
let margin = SIGNATURE_LEDGER_TTL_SECS - arrival_window;
assert!(
margin >= 60,
"TTL {SIGNATURE_LEDGER_TTL_SECS}s leaves {margin}s over a {arrival_window}s \
arrival window, under the 60s skew budget: an instance running ahead \
would sweep a row still accepted elsewhere"
);
}
// ── `gitlawb_signature_ledger_total{outcome}` ────────────────────────────
//
// The ledger fails CLOSED, so a fault local to it (statement timeout, lock
// contention) turns every REST mutation into a 503 while the handlers' own
// queries stay healthy. These assert the counter is really charged on each
// terminal outcome, driven through the layer and read back through the same
// gather/encode path `/metrics` serves.
//
// Each assertion is a strict increase rather than an exact delta: the
// counter is process-wide, `metrics::init` is a `OnceLock`, and the ledger
// tests in `test_support` run in the same binary, so a concurrent test can
// add to any outcome. Pollution can only ever add, never subtract, so a
// strict increase never flakes; to attribute a delta exactly (e.g. when
// mutating out a `record_*` call to confirm RED), run these serially:
// `cargo test --bin gitlawb-node ledger_metrics -- --test-threads=1`.
fn ledger_metric(outcome: &str) -> u64 {
let needle = format!("gitlawb_signature_ledger_total{{outcome=\"{outcome}\"}} ");
crate::metrics::encode()
.expect("encode should succeed after init")
.lines()
.find_map(|line| line.strip_prefix(needle.as_str()))
.map(|value| value.trim().parse().expect("counter value is an integer"))
.unwrap_or(0)
}
/// One route, mounted behind `consume_signature` alone. The layer reads the
/// `SignatureIdentity` extension, so tests inject it directly rather than
/// producing real RFC-9421 signatures.
fn ledger_app(state: crate::state::AppState) -> Router {
Router::new()
.route(
"/probe",
axum::routing::post(|| async { StatusCode::OK }).get(|| async { StatusCode::OK }),
)
.layer(middleware::from_fn_with_state(state, consume_signature))
}
fn ledger_identity(seed: &str, nonce: Option<&str>) -> SignatureIdentity {
SignatureIdentity {
keyid: format!("did:key:zLedgerMetrics{seed}"),
key_fingerprint: hex::encode(Sha256::digest(format!("key/{seed}").as_bytes())),
nonce: nonce.map(str::to_owned),
signing_string_hash: hex::encode(Sha256::digest(format!("sig/{seed}").as_bytes())),
}
}
fn ledger_request(identity: Option<SignatureIdentity>) -> Request {
let builder = Request::builder().method(Method::POST).uri("/probe");
let builder = match identity {
Some(identity) => builder.extension(identity),
None => builder,
};
builder.body(Body::empty()).expect("request builder")
}
async fn ledger_status(state: crate::state::AppState, request: Request) -> StatusCode {
ledger_app(state)
.oneshot(request)
.await
.expect("router response")
.status()
}
#[tokio::test]
async fn ledger_metrics_count_a_skipped_read() {
crate::metrics::init("0.0.0-test", "did:key:test");
let before = ledger_metric("skipped_read");
let request = Request::builder()
.method(Method::GET)
.uri("/probe")
.body(Body::empty())
.expect("request builder");
let status = ledger_status(make_test_state(Keypair::generate().did()), request).await;
assert_eq!(status, StatusCode::OK, "a read must still pass through");
assert!(
ledger_metric("skipped_read") > before,
"the read skip must be counted, or the outcomes do not sum to the layer's traffic"
);
}
#[tokio::test]
async fn ledger_metrics_count_a_missing_identity() {
crate::metrics::init("0.0.0-test", "did:key:test");
let before = ledger_metric("identity_missing");
let status = ledger_status(
make_test_state(Keypair::generate().did()),
ledger_request(None),
)
.await;
assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
assert!(ledger_metric("identity_missing") > before);
}
#[tokio::test]
async fn ledger_metrics_count_the_nonce_rejections() {
use clap::Parser;
crate::metrics::init("0.0.0-test", "did:key:test");
let mut state = make_test_state(Keypair::generate().did());
state.config = Arc::new(crate::config::Config::parse_from([
"gitlawb-node",
"--require-signature-nonce",
]));
let before_required = ledger_metric("nonce_required");
let status = ledger_status(
state.clone(),
ledger_request(Some(ledger_identity("no-nonce", None))),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert!(ledger_metric("nonce_required") > before_required);
let before_short = ledger_metric("nonce_too_short");
let status = ledger_status(
state,
ledger_request(Some(ledger_identity("short-nonce", Some("abc")))),
)
.await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert!(ledger_metric("nonce_too_short") > before_short);
}
/// The fail-closed 503 is the outcome an operator most needs a numerator
/// for: it is returned by the ledger while the handlers are healthy. The
/// state here carries a lazy pool pointed at a database that does not
/// exist, so `consume_signature` returns `Err`.
#[tokio::test]
async fn ledger_metrics_count_a_ledger_error() {
crate::metrics::init("0.0.0-test", "did:key:test");
let before = ledger_metric("unavailable");
let status = ledger_status(
make_test_state(Keypair::generate().did()),
ledger_request(Some(ledger_identity("db-down", None))),
)
.await;
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
assert!(ledger_metric("unavailable") > before);
}
#[sqlx::test]
async fn ledger_metrics_count_admitted_and_replayed(pool: sqlx::PgPool) {
crate::metrics::init("0.0.0-test", "did:key:test");
let state = crate::test_support::test_state(pool).await;
let identity = ledger_identity("spend-once", None);
let before_admitted = ledger_metric("admitted");
let status = ledger_status(state.clone(), ledger_request(Some(identity.clone()))).await;
assert_eq!(status, StatusCode::OK, "a fresh signature is admitted");
assert!(ledger_metric("admitted") > before_admitted);
let before_replayed = ledger_metric("replayed");
let status = ledger_status(state, ledger_request(Some(identity))).await;
assert_eq!(status, StatusCode::CONFLICT, "the same signature replays");
assert!(ledger_metric("replayed") > before_replayed);
}
#[sqlx::test]
async fn ledger_metrics_count_a_full_identity_ledger(pool: sqlx::PgPool) {
crate::metrics::init("0.0.0-test", "did:key:test");
let state = crate::test_support::test_state(pool.clone()).await;
let identity = ledger_identity("capped", None);
// Seeded through SQL rather than 512 requests. The cap is charged
// against the resolved key, which is the `keyid` column's value.
sqlx::query(
"INSERT INTO consumed_signatures (sig_hash, keyid, expires_at)
SELECT md5(i::text) || md5((i + 1)::text), $1, $2
FROM generate_series(1, $3) AS i",
)
.bind(&identity.key_fingerprint)
.bind(9_000_000_000i64)
.bind(crate::db::MAX_LIVE_SIGNATURES_PER_KEYID)
.execute(&pool)
.await
.expect("seed a full ledger for this identity");
let before = ledger_metric("identity_ledger_full");
let status = ledger_status(state, ledger_request(Some(identity))).await;
assert_eq!(status, StatusCode::TOO_MANY_REQUESTS);
assert!(ledger_metric("identity_ledger_full") > before);
}
#[tokio::test]
async fn require_ucan_chain_no_header_passes_through() {
let state = make_test_state(Keypair::generate().did());
-90
View File
@@ -76,14 +76,6 @@ pub struct Config {
)]
pub require_signed_peer_writes: bool,
/// Require an RFC 9421 `nonce` parameter on signatures that authorize a
/// mutation. Keep false until every client signs one: nodes fall back to
/// keying the replay ledger on the signing-string hash without it, which
/// rejects a byte-identical mutation repeated inside the same second.
/// Enforced on write routes only, so signed reads are never affected.
#[arg(long, env = "GITLAWB_REQUIRE_SIGNATURE_NONCE", default_value_t = false)]
pub require_signature_nonce: bool,
/// Require the authenticated pusher to be the repo owner on `git-receive-pack`.
/// Authentication (a valid did:key signature) is not authorization on its own:
/// any party can mint a did:key and sign as it, so with this off every signed
@@ -205,19 +197,6 @@ pub struct Config {
#[arg(long, env = "GITLAWB_PEER_WRITE_RATE_LIMIT", default_value_t = 600)]
pub peer_write_rate_limit: usize,
/// Per-client-IP rate limit for the signed write routes (tasks, pull-request
/// merge/close/review/comment, webhooks, branch protection, stars, replicas,
/// labels, visibility, agent deregistration, bounties, profile, issue
/// close/comment), in requests per hour. Every signed attempt on these routes
/// commits a `consumed_signatures` row before the handler runs any
/// authorization or existence check, and a signature proves nothing about
/// registration (any freshly minted `did:key` self-signs), so without a brake
/// an unregistered caller can force a durable write per request. Its own
/// bucket, so a flood here cannot drain the creation, push, or peer-sync
/// quotas. `0` disables. Default: 600.
#[arg(long, env = "GITLAWB_SIGNED_WRITE_RATE_LIMIT", default_value_t = 600)]
pub signed_write_rate_limit: usize,
/// Optional address to bind a Prometheus `/metrics` exposition endpoint on.
/// Example: `127.0.0.1:9091`. Leave empty (default) to disable.
/// Bind to localhost or a private interface — the metrics endpoint is
@@ -792,75 +771,6 @@ mod tests {
);
}
#[test]
fn require_signature_nonce_ships_off() {
assert!(
!Config::parse_from(["gitlawb-node"]).require_signature_nonce,
"the staged flag must default to false"
);
assert!(
Config::parse_from(["gitlawb-node", "--require-signature-nonce"])
.require_signature_nonce
);
}
/// Child half of [`nonce_flag_env_values_never_silently_enforce`]. It runs in
/// a subprocess with `GITLAWB_REQUIRE_SIGNATURE_NONCE` set, because setting
/// that variable in-process would race every other test that parses a
/// `Config` and would take the whole binary down if the value were invalid.
#[test]
#[ignore = "driven as a subprocess by nonce_flag_env_values_never_silently_enforce"]
fn nonce_flag_env_probe() {
match Config::try_parse_from(["gitlawb-node"]) {
Ok(config) => println!("PROBE:ok:{}", config.require_signature_nonce),
Err(_) => println!("PROBE:err"),
}
}
/// The flag is read as a boolean, not as "is the variable present". An
/// operator writing `GITLAWB_REQUIRE_SIGNATURE_NONCE=0` (or leaving it
/// empty) must never end up enforcing.
#[test]
fn nonce_flag_env_values_never_silently_enforce() {
fn probe(value: &str) -> String {
let out = std::process::Command::new(
std::env::current_exe().expect("path to the test binary"),
)
.args([
"--exact",
"config::tests::nonce_flag_env_probe",
"--ignored",
"--nocapture",
])
.env("GITLAWB_REQUIRE_SIGNATURE_NONCE", value)
.output()
.expect("run the env probe");
let stdout = String::from_utf8_lossy(&out.stdout);
stdout
.lines()
.find(|l| l.starts_with("PROBE:"))
.unwrap_or("PROBE:missing")
.to_string()
}
// The probe can observe enforcement, so a "never true" assertion below
// is not vacuous.
assert_eq!(probe("true"), "PROBE:ok:true");
for value in ["false", "0", ""] {
let observed = probe(value);
assert_ne!(
observed, "PROBE:ok:true",
"GITLAWB_REQUIRE_SIGNATURE_NONCE={value:?} must not enforce, got {observed}"
);
}
assert_eq!(
probe("false"),
"PROBE:ok:false",
"the documented disable value must parse to false, not error"
);
}
/// #174 (RED-before/GREEN-after): the upper bound is what keeps every duration
/// derived from this knob in range — the lease steal bound's `* 2 + 60` on the write
/// path, and the `Instant::now() + Duration::from_secs(..)` deadlines in
+13 -880
View File
@@ -3,41 +3,9 @@ use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sqlx::{postgres::PgPoolOptions, PgPool, Row};
use std::time::Duration;
use tracing::{info, warn};
use tracing::info;
use uuid::Uuid;
/// A migration version this build defines is already recorded in the database
/// under a different name. Distinct type, not a bare `anyhow!`, because the
/// startup retry loop has to tell this apart from a transient outage: it never
/// heals on its own, so it must be reported as a schema conflict rather than
/// "the database is coming up". See `is_likely_permanent_db_error`.
#[derive(Debug, thiserror::Error)]
#[error(
"migration version {version} is already applied in this database as {recorded_name:?}, but \
this build defines v{version} as {build_name:?}. Either two migrations claimed the same \
version, in which case renumber this build's migration above every applied version instead \
of reusing v{version}; or {recorded_name:?} is leftover bookkeeping from abandoned numbering \
on a build that no longer exists, in which case drop the stale row with \
`DELETE FROM schema_migrations WHERE version = {version}` after confirming its schema \
objects match what v{version} ({build_name:?}) creates."
)]
pub struct MigrationVersionCollision {
pub version: i64,
pub recorded_name: String,
pub build_name: &'static str,
}
/// A version recorded in `schema_migrations` that this build's catalogue does
/// not define at all: a row written by a since-renumbered build, say, or by an
/// older binary during a rollback. Never fatal on its own (see the note in
/// `run_pending_migrations`), but reported so it is visible before a future
/// migration claims that number and turns it into a boot failure.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecordedOrphanMigration {
pub version: i64,
pub name: String,
}
// ── Public data types ─────────────────────────────────────────────────────────
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -405,54 +373,22 @@ impl Db {
.execute(&mut *lock_conn)
.await;
for orphan in result.as_ref().map(Vec::as_slice).unwrap_or_default() {
warn!(
version = orphan.version,
name = %orphan.name,
"schema_migrations records a version this build does not define; it is \
leftover bookkeeping (a renumbered or rolled-back build). Harmless now, but a \
future migration claiming this version will abort startup. Confirm its schema \
objects are already covered and then run \
`DELETE FROM schema_migrations WHERE version = <version>`"
);
}
result.map(|_| ())
result
}
/// Apply every migration whose version isn't yet recorded, in order.
/// Must be called while holding the migration advisory lock.
///
/// Returns the recorded versions this build's catalogue does not define.
/// The caller reports them; they are deliberately NOT fatal. A legitimate
/// rollback produces them transiently (an older binary re-runs an
/// idempotent migration under its old number and records it), and aborting
/// startup would turn a supported roll-forward into an outage. The reason
/// to surface them anyway is that such a row is exactly what trips the
/// collision check later, once some other branch claims that version.
async fn run_pending_migrations(&self) -> Result<Vec<RecordedOrphanMigration>> {
async fn run_pending_migrations(&self) -> Result<()> {
for m in MIGRATIONS {
// Compare the recorded *name*, not just the version. Two branches
// developed in parallel can each claim the same version number;
// with a version-only check the one that merges second is silently
// skipped, leaving the schema it needs missing while
// `schema_migrations` still reads healthy. Fail loudly at startup
// instead, so the collision is fixed by renumbering.
let recorded: Option<String> =
sqlx::query_scalar("SELECT name FROM schema_migrations WHERE version = $1")
.bind(m.version)
.fetch_optional(&self.pool)
.await?;
let already: bool = sqlx::query(
"SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied",
)
.bind(m.version)
.fetch_one(&self.pool)
.await?
.get::<bool, _>("applied");
if let Some(recorded) = recorded {
if recorded != m.name {
return Err(MigrationVersionCollision {
version: m.version,
recorded_name: recorded,
build_name: m.name,
}
.into());
}
if already {
continue;
}
@@ -498,20 +434,7 @@ impl Db {
);
}
// The loop above only ever looks at versions this build defines, so a
// recorded version the catalogue no longer mentions is invisible to it.
// That is precisely the row a renumber leaves behind. Find it by asking
// the database what it has, rather than by asking about what we know.
let rows: Vec<(i64, String)> =
sqlx::query_as("SELECT version, name FROM schema_migrations ORDER BY version ASC")
.fetch_all(&self.pool)
.await
.context("listing recorded migrations")?;
Ok(rows
.into_iter()
.filter(|(version, _)| !MIGRATIONS.iter().any(|m| m.version == *version))
.map(|(version, name)| RecordedOrphanMigration { version, name })
.collect())
Ok(())
}
/// Returns `(version, name, applied_at)` for every applied migration,
@@ -975,41 +898,6 @@ const MIGRATIONS: &[Migration] = &[
"ALTER TABLE received_ref_updates ADD COLUMN IF NOT EXISTS owner_did TEXT",
],
},
Migration {
// v12..v15 are claimed by the in-flight IPFS CID branch
// (fix/issue-135-ipfs-cid-tree-gate). That branch is behind main and
// still numbers its four migrations 11..14, so once it rebases past
// main's v11 (ref_update_owner_did) they shift up to 12..15. Numbering
// above that range keeps whichever branch merges second from having
// its migration skipped as "already applied".
version: 16,
name: "consumed_signatures",
stmts: &[
// Single-use ledger for HTTP message signatures. `sig_hash` is a
// fixed-width hex SHA-256 digest of the canonical signature key,
// never the raw nonce or signing string: the value is
// attacker-influenced, so storing a digest bounds bytes per row and
// keeps unbounded input out of the table. The CHECK enforces the
// fixed width at the schema level so a caller cannot store
// something else by accident.
// `keyid` backs the per-identity live-row cap
// (`MAX_LIVE_SIGNATURES_PER_KEYID`). Despite the column name it
// holds the hex of the RESOLVED public key, not the wire DID: the
// cap compares by string equality and one keypair has many valid
// multibase spellings. `expires_at` is the
// unix-seconds instant after which the signature can no longer be
// accepted, used by the sweep.
r#"CREATE TABLE IF NOT EXISTS consumed_signatures (
sig_hash TEXT NOT NULL PRIMARY KEY
CHECK (char_length(sig_hash) = 64),
keyid TEXT NOT NULL,
expires_at BIGINT NOT NULL
)"#,
"CREATE INDEX IF NOT EXISTS idx_consumed_signatures_expires ON consumed_signatures(expires_at)",
// The cap counts live rows for one keyid, so it needs both columns.
"CREATE INDEX IF NOT EXISTS idx_consumed_signatures_keyid_expires ON consumed_signatures(keyid, expires_at)",
],
},
// Reservation: v17 is deliberately not main's current_max + 1. The runner keys the
// applied set on the integer alone, so a version another in-flight branch also
// claims is skipped in full on whichever side merges second: no error, no warning,
@@ -1237,59 +1125,12 @@ const MIGRATIONS: &[Migration] = &[
},
];
// ── HTTP-signature replay ledger ──────────────────────────────────────────────
/// Outcome of charging one HTTP message signature against the replay ledger.
/// The three cases are deliberately distinct so a caller can answer each with
/// its own status code: a replay is the client's fault and is permanent for
/// that signature, whereas a full identity ledger is a rate condition the same
/// client can retry once its live rows expire.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[must_use]
pub enum ConsumeSignature {
/// The signature had not been seen; it is now spent.
Inserted,
/// The signature was already spent. Reject as a replay.
Replayed,
/// This identity already holds `MAX_LIVE_SIGNATURES_PER_KEYID` live rows.
/// The signature was NOT recorded.
IdentityLedgerFull,
}
/// Cap on live (unexpired) ledger rows a single `keyid` may hold.
///
/// The ledger is an attacker-influenced key map on routes that carry no rate
/// limiter (tasks, bounties, issue and PR comments, hooks, profile), and
/// identities are permissionless, so a flood of *distinct* signatures from one
/// identity would otherwise allocate rows without bound. Hashing the key bounds
/// bytes per row; only this bounds row count.
///
/// 512 over the 390s retention window is ~1.3 sustained requests per second
/// from one identity, an order of magnitude above any real client (a push plus
/// its API calls is tens of requests).
///
/// The cap counts LIVE rows (`expires_at >= now`), and the sweep runs on an
/// independent 300s tick (`main.rs`), so expired-but-unswept rows sit in the
/// table uncounted. On-disk peak is therefore about double the cap, not equal
/// to it: at the sustained ceiling an identity retires 512/390 rows per second,
/// so up to `512 * 300 / 390` ≈ 394 expired rows accumulate between sweeps, for
/// a peak near 906 and a hard bound of 1024 (one sweep period is shorter than
/// the TTL, so at most one cap's worth can be pending deletion).
///
/// Measured against Postgres at 1M rows, laid out the way the cap produces them
/// (64-hex `sig_hash`, 64-hex identity, 512 rows per identity): heap 166 MB, PK
/// index 119 MB, expires index 21 MB, identity+expires index 101 MB, total 407
/// MB — about 430 bytes per row including indexes. One identity's worst case is
/// thus ~1024 rows, roughly 440 KB.
pub const MAX_LIVE_SIGNATURES_PER_KEYID: i64 = 512;
/// Advisory-lock class for serializing per-identity ledger cap checks.
const LEDGER_CAP_LOCK_CLASS: i32 = 0x676c;
/// Max distinct source repos recorded per pinned object (F1, #173 jatmn round 8).
/// Bounds both the resolver's per-OID source loop and the `pin_repo_sources` growth,
/// so an adversary re-pushing one object from many repos cannot make resolution
/// O(repos) (R2, INV-10).
pub const MAX_PIN_SOURCES: i64 = 16;
// ── Repos ─────────────────────────────────────────────────────────────────────
pub(crate) fn normalize_owner_key(did: &str) -> &str {
@@ -1811,91 +1652,6 @@ impl Db {
Ok(result.rows_affected() > 0)
}
/// Atomically consume an HTTP message signature. `sig_hash` is the
/// fixed-width hex SHA-256 digest of the canonical signature key,
/// `identity` the signing identity the per-identity cap is charged against,
/// `now` the arrival instant in unix seconds, and `expires_at` the instant
/// after which the signature can no longer be accepted (so the row can be
/// swept).
///
/// `identity` must be a canonical form of the resolved public key, never
/// the wire DID: the cap compares by string equality, and one keypair has
/// many valid `did:key` spellings. The column keeps its historical `keyid`
/// name (renaming it would need a new migration for no behavioral gain).
///
/// A `sig_hash` of any width other than 64 violates the table's CHECK
/// constraint and comes back as a query `Err`, which `consume_signature`
/// (`auth/mod.rs`) reports as 503 `signature_ledger_unavailable` — so a
/// caller that ever passes one would send an operator chasing a phantom
/// database outage. Unreachable today: both `ledger_key` arms return a hex
/// SHA-256.
///
/// One statement does the whole decision. The `INSERT ... ON CONFLICT DO
/// NOTHING` is what makes the check atomic: two concurrent replays of the
/// same signature cannot both win, because the primary key arbitrates. A
/// separate SELECT first would reintroduce exactly that race.
pub async fn consume_signature(
&self,
sig_hash: &str,
identity: &str,
now: i64,
expires_at: i64,
) -> Result<ConsumeSignature> {
let mut tx = self.pool.begin().await?;
sqlx::query("SELECT pg_advisory_xact_lock($1, hashtext($2::text))")
.bind(LEDGER_CAP_LOCK_CLASS)
.bind(identity)
.execute(&mut *tx)
.await?;
let (inserted, live, already_spent): (i64, i64, bool) = sqlx::query_as(
r#"WITH live AS (
SELECT COUNT(*) AS n FROM consumed_signatures
WHERE keyid = $2 AND expires_at >= $4
), ins AS (
INSERT INTO consumed_signatures (sig_hash, keyid, expires_at)
SELECT $1, $2, $3 FROM live WHERE live.n < $5
ON CONFLICT (sig_hash) DO NOTHING
RETURNING 1
)
SELECT (SELECT COUNT(*) FROM ins) AS inserted,
(SELECT n FROM live) AS live_count,
EXISTS(
SELECT 1 FROM consumed_signatures WHERE sig_hash = $1
) AS already_spent"#,
)
.bind(sig_hash)
.bind(identity)
.bind(expires_at)
.bind(now)
.bind(MAX_LIVE_SIGNATURES_PER_KEYID)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
if inserted > 0 {
Ok(ConsumeSignature::Inserted)
} else if already_spent {
Ok(ConsumeSignature::Replayed)
} else if live >= MAX_LIVE_SIGNATURES_PER_KEYID {
Ok(ConsumeSignature::IdentityLedgerFull)
} else {
Ok(ConsumeSignature::Replayed)
}
}
/// Delete ledger rows for signatures that can no longer be accepted.
/// Returns rows removed. This is what keeps the table proportional to the
/// retention window rather than to total requests seen.
pub async fn sweep_expired_signatures(&self, now: i64) -> Result<u64> {
let result = sqlx::query("DELETE FROM consumed_signatures WHERE expires_at < $1")
.bind(now)
.execute(&self.pool)
.await?;
Ok(result.rows_affected())
}
/// Delete consumed-proof rows whose proof has expired. Returns rows removed.
pub async fn sweep_expired_proofs(&self, now: i64) -> Result<u64> {
let result = sqlx::query("DELETE FROM icaptcha_consumed_proofs WHERE expires_at < $1")
@@ -5340,120 +5096,6 @@ mod migration_tests {
}
}
#[cfg(test)]
mod migration_guard_tests {
use super::{MigrationVersionCollision, MIGRATIONS};
async fn record(pool: &sqlx::PgPool, version: i64, name: &str) {
sqlx::query(
"INSERT INTO schema_migrations (version, name, applied_at) VALUES ($1, $2, $3)",
)
.bind(version)
.bind(name)
.bind("2026-07-01T00:00:00Z")
.execute(pool)
.await
.unwrap();
}
/// A recorded version this build's catalogue no longer defines is
/// reported, and does not by itself stop startup.
#[sqlx::test]
async fn a_recorded_version_the_catalogue_does_not_define_is_reported(pool: sqlx::PgPool) {
let db = super::Db::for_testing(pool);
db.migrate().await.unwrap();
record(&db.pool, 9_001, "ghost_migration").await;
let orphans = db
.run_pending_migrations()
.await
.expect("an unknown recorded version must not fail startup");
assert!(
orphans
.iter()
.any(|o| o.version == 9_001 && o.name == "ghost_migration"),
"expected the orphan row to be reported, got {orphans:?}"
);
// And startup as a whole still succeeds.
db.migrate().await.expect("migrate must still succeed");
}
/// The concrete rollback sequence: this build records v16, an older binary
/// whose catalogue still numbers the same migration v12 records
/// (12, 'consumed_signatures') as a no-op, then we roll forward. The stale
/// row must surface as an orphan, not bail.
#[sqlx::test]
async fn a_rolled_back_binarys_stale_row_surfaces_without_bailing(pool: sqlx::PgPool) {
let db = super::Db::for_testing(pool);
db.migrate().await.unwrap();
// The rolled-back binary finds no v12 row, re-runs
// CREATE TABLE IF NOT EXISTS consumed_signatures as a no-op, records v12.
record(&db.pool, 12, "consumed_signatures").await;
let orphans = db
.run_pending_migrations()
.await
.expect("a rollback's leftover bookkeeping row must not abort startup");
assert!(
orphans
.iter()
.any(|o| o.version == 12 && o.name == "consumed_signatures"),
"expected the stale (12, consumed_signatures) row to be reported, got {orphans:?}"
);
assert!(
!MIGRATIONS.iter().any(|m| m.version == 12),
"this test assumes the catalogue does not define v12"
);
}
/// A version the catalogue DOES define, recorded under a different name,
/// is a genuine collision and still aborts startup.
#[sqlx::test]
async fn a_genuine_name_collision_still_bails(pool: sqlx::PgPool) {
let db = super::Db::for_testing(pool);
db.migrate().await.unwrap();
sqlx::query("UPDATE schema_migrations SET name = $1 WHERE version = 16")
.bind("pinned_cids_repo_provenance")
.execute(&db.pool)
.await
.unwrap();
let err = db
.run_pending_migrations()
.await
.expect_err("a name collision on a defined version must abort startup");
let collision = err
.downcast_ref::<MigrationVersionCollision>()
.expect("the collision must be a distinguishable MigrationVersionCollision");
assert_eq!(collision.version, 16);
assert_eq!(collision.recorded_name, "pinned_cids_repo_provenance");
let msg = format!("{err:#}");
assert!(msg.contains("16"), "message must name the version: {msg}");
assert!(
msg.contains("pinned_cids_repo_provenance"),
"message must name the recorded name: {msg}"
);
assert!(
msg.contains("consumed_signatures"),
"message must name this build's name: {msg}"
);
// Both remedies must be offered, so the operator is not sent to the
// wrong one.
assert!(
msg.contains("renumber"),
"message must offer the renumber remedy: {msg}"
);
assert!(
msg.contains("DELETE FROM schema_migrations"),
"message must offer the stale-row remedy: {msg}"
);
}
}
#[cfg(test)]
mod agent_discovery_tests {
use super::{filter_discoverable, AgentRow};
@@ -6507,515 +6149,6 @@ mod icaptcha_ledger_tests {
}
}
/// Exercises the HTTP-signature replay ledger (`consumed_signatures`): its
/// single-use accessor, the sweep that bounds it, the per-identity cap, and the
/// v16 upgrade path from a database sitting at the version below it, plus the
/// runner's guard against a version applied under a different migration name.
#[cfg(test)]
mod signature_ledger_tests {
use super::{ConsumeSignature, Db, MIGRATIONS};
use sqlx::PgPool;
async fn db(pool: PgPool) -> Db {
let db = Db::for_testing(pool);
db.run_migrations().await.unwrap();
db
}
/// The ledger key is a fixed-width hex SHA-256 digest, never a raw nonce
/// (KTD6). Build one deterministically from a seed so tests stay readable.
fn key(seed: &str) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(seed.as_bytes());
hex::encode(h.finalize())
}
/// First sighting of a signature is recorded (allowed); the identical key
/// again is a replay (rejected); a distinct key is independently allowed.
#[sqlx::test]
async fn consume_signature_is_single_use(pool: PgPool) {
let db = db(pool).await;
let now = 1_000i64;
let exp = now + 330;
let kid = "did:key:zAlice";
assert_eq!(
db.consume_signature(&key("a"), kid, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted,
"first sighting is recorded and allowed"
);
assert_eq!(
db.consume_signature(&key("a"), kid, now, exp)
.await
.unwrap(),
ConsumeSignature::Replayed,
"the identical signature is a replay and must be rejected"
);
assert_eq!(
db.consume_signature(&key("b"), kid, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted,
"a different signature is independent and allowed"
);
}
/// The sweep deletes only rows whose `expires_at` is strictly before the
/// cutoff and returns the deleted count. A swept signature is usable again
/// (its acceptance window has passed, so the time check now rejects it
/// anyway), while an unswept one keeps rejecting replays.
#[sqlx::test]
async fn sweep_expired_signatures_removes_only_expired(pool: PgPool) {
let db = db(pool).await;
let kid = "did:key:zAlice";
let (old_a, old_b, fresh) = (key("old-a"), key("old-b"), key("fresh"));
for (k, exp) in [(&old_a, 100i64), (&old_b, 199), (&fresh, 500)] {
assert_eq!(
db.consume_signature(k, kid, 0, exp).await.unwrap(),
ConsumeSignature::Inserted
);
}
let deleted = db.sweep_expired_signatures(200).await.unwrap();
assert_eq!(
deleted, 2,
"only the two rows with expires_at < 200 are swept"
);
assert_eq!(
db.consume_signature(&old_a, kid, 200, 530).await.unwrap(),
ConsumeSignature::Inserted,
"a swept signature is free again"
);
assert_eq!(
db.consume_signature(&fresh, kid, 200, 530).await.unwrap(),
ConsumeSignature::Replayed,
"an unexpired spent signature survives the sweep and still blocks replays"
);
}
/// KTD6, proved by execution rather than by pointing at the TTL: a single
/// identity flooding *distinct* signatures fills its cap and is then
/// rejected with an outcome distinct from a replay, and its row count stops
/// growing. A second identity is untouched by the first one's cap.
#[sqlx::test]
async fn per_identity_cap_bounds_row_count(pool: PgPool) {
let db = db(pool).await;
let now = 1_000i64;
let exp = now + 330; // every row stays live for the whole run
let flooder = "did:key:zFlooder";
let cap = super::MAX_LIVE_SIGNATURES_PER_KEYID;
// Fill exactly to the cap with distinct keys.
for i in 0..cap {
assert_eq!(
db.consume_signature(&key(&format!("flood-{i}")), flooder, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted,
"distinct signature {i} is below the cap and must be recorded"
);
}
// Every further distinct signature is refused, with an outcome the
// caller can tell apart from a replay.
for i in cap..cap + 25 {
assert_eq!(
db.consume_signature(&key(&format!("flood-{i}")), flooder, now, exp)
.await
.unwrap(),
ConsumeSignature::IdentityLedgerFull,
"signature {i} is over the cap and must be refused distinctly"
);
}
// Storage stopped growing: the flood allocated cap rows, not cap + 25.
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM consumed_signatures WHERE keyid = $1"
)
.bind(flooder)
.fetch_one(&db.pool)
.await
.unwrap(),
cap,
"the flood must not allocate a row per request seen"
);
// A second identity is unaffected by the first one's cap, and its own
// replays still read as replays.
let bystander = "did:key:zBystander";
assert_eq!(
db.consume_signature(&key("bystander-1"), bystander, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted,
"the cap is per identity, not global"
);
assert_eq!(
db.consume_signature(&key("bystander-1"), bystander, now, exp)
.await
.unwrap(),
ConsumeSignature::Replayed
);
// The cap counts *live* rows only: once the flooder's rows age out and
// are swept, it can sign again.
// Everything inserted above shares `exp`, so this clears the flooder's
// `cap` rows plus the bystander's single row.
assert_eq!(
db.sweep_expired_signatures(exp + 1).await.unwrap(),
cap as u64 + 1
);
assert_eq!(
db.consume_signature(&key("flood-after-sweep"), flooder, exp + 1, exp + 331)
.await
.unwrap(),
ConsumeSignature::Inserted,
"the cap must lift once the identity's rows expire"
);
}
/// A replay at the cap must read as `Replayed`, not `IdentityLedgerFull`.
#[sqlx::test]
async fn replay_at_identity_cap_returns_replayed(pool: PgPool) {
let db = db(pool).await;
let kid = "did:key:zFlooder";
let cap = super::MAX_LIVE_SIGNATURES_PER_KEYID;
let now = 3_000i64;
let exp = now + 330;
for i in 0..cap - 1 {
assert_eq!(
db.consume_signature(&key(&format!("cap-{i}")), kid, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted
);
}
let spent = key("spent-once");
assert_eq!(
db.consume_signature(&spent, kid, now, exp).await.unwrap(),
ConsumeSignature::Inserted
);
assert_eq!(
db.consume_signature(&spent, kid, now, exp).await.unwrap(),
ConsumeSignature::Replayed,
"a replay must not be reported as the cap being full"
);
}
/// Concurrent distinct signatures cannot push an identity above the cap.
#[sqlx::test]
async fn concurrent_consume_signature_respects_per_identity_cap(pool: PgPool) {
use std::sync::Arc;
let db = Arc::new(db(pool).await);
let kid = "did:key:zRace";
let cap = super::MAX_LIVE_SIGNATURES_PER_KEYID;
let now = 4_000i64;
let exp = now + 330;
for i in 0..cap - 5 {
assert_eq!(
db.consume_signature(&key(&format!("seed-{i}")), kid, now, exp)
.await
.unwrap(),
ConsumeSignature::Inserted
);
}
let mut handles = Vec::new();
for i in 0..20 {
let db = Arc::clone(&db);
handles.push(tokio::spawn(async move {
db.consume_signature(&key(&format!("race-{i}")), kid, now, exp)
.await
}));
}
for handle in handles {
drop(handle.await.unwrap());
}
let live: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM consumed_signatures WHERE keyid = $1 AND expires_at >= $2",
)
.bind(kid)
.bind(now)
.fetch_one(&db.pool)
.await
.unwrap();
assert!(
live <= cap,
"concurrent inserts must not exceed the per-identity cap, saw {live}"
);
}
/// Upgrade path: an existing node sitting at the highest migration below
/// v16 must gain the ledger table and its sweep index when it applies v16,
/// and the table must round-trip. A fresh-database test cannot cover this —
/// it never exercises the "everything but v16" -> v16 step in isolation.
#[sqlx::test]
async fn migration_v16_creates_signature_ledger(pool: PgPool) {
let db = Db::for_testing(pool);
// Build the whole schema, then tear the v16 objects back down and
// re-seed `schema_migrations` with every migration below v16 to
// simulate a node that has run everything else but not yet v16.
db.run_migrations().await.unwrap();
sqlx::query("DROP TABLE IF EXISTS consumed_signatures")
.execute(&db.pool)
.await
.unwrap();
sqlx::query("DELETE FROM schema_migrations")
.execute(&db.pool)
.await
.unwrap();
for m in MIGRATIONS.iter().take_while(|m| m.version < 16) {
sqlx::query(
"INSERT INTO schema_migrations (version, name, applied_at)
VALUES ($1, $2, $3)",
)
.bind(m.version)
.bind(m.name)
.bind("2026-07-01T00:00:00Z")
.execute(&db.pool)
.await
.unwrap();
}
// Derive the baseline from the catalogue rather than hardcoding it, so
// the test keeps meaning "upgrade from whatever ships below v16" as
// other branches land migrations in the 12..16 range.
let baseline = MIGRATIONS
.iter()
.map(|m| m.version)
.filter(|v| *v < 16)
.max()
.expect("catalogue must have migrations below v16");
assert_eq!(
sqlx::query_scalar::<_, i64>("SELECT MAX(version) FROM schema_migrations")
.fetch_one(&db.pool)
.await
.unwrap(),
baseline,
"baseline must be the highest migration below v16"
);
// ── Apply the pending v16 migration ───────────────────────────────
db.run_migrations().await.unwrap();
// (a) The table exists.
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM information_schema.tables
WHERE table_name = 'consumed_signatures'"
)
.fetch_one(&db.pool)
.await
.unwrap(),
1,
"v16 must create consumed_signatures"
);
// (b) The sweep index exists.
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM pg_indexes
WHERE tablename = 'consumed_signatures'
AND indexname = 'idx_consumed_signatures_expires'"
)
.fetch_one(&db.pool)
.await
.unwrap(),
1,
"the sweep needs an index on expires_at"
);
// (b2) The per-keyid cap index exists too.
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM pg_indexes
WHERE tablename = 'consumed_signatures'
AND indexname = 'idx_consumed_signatures_keyid_expires'"
)
.fetch_one(&db.pool)
.await
.unwrap(),
1,
"the per-keyid live-row cap needs an index on (keyid, expires_at)"
);
// (c) It round-trips: insert, read back, and the digest survives whole.
let k = key("upgrade-path");
sqlx::query(
"INSERT INTO consumed_signatures (sig_hash, keyid, expires_at) VALUES ($1, $2, $3)",
)
.bind(&k)
.bind("did:key:zUpgrade")
.bind(9_000_000_000i64)
.execute(&db.pool)
.await
.unwrap();
let (got_key, got_keyid, got_exp): (String, String, i64) = sqlx::query_as(
"SELECT sig_hash, keyid, expires_at FROM consumed_signatures WHERE sig_hash = $1",
)
.bind(&k)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(got_key, k);
assert_eq!(got_keyid, "did:key:zUpgrade");
assert_eq!(got_exp, 9_000_000_000i64);
}
/// Re-running migrations over an already-migrated database is a no-op: v16
/// applies cleanly a second time, records exactly one `schema_migrations`
/// row, and does not disturb rows already in the ledger.
#[sqlx::test]
async fn migrations_are_idempotent(pool: PgPool) {
let db = db(pool).await;
let k = key("survives-remigration");
assert_eq!(
db.consume_signature(&k, "did:key:zAlice", 1_000, 1_330)
.await
.unwrap(),
ConsumeSignature::Inserted
);
// Third and fourth passes over the same database.
db.run_migrations().await.unwrap();
db.run_migrations().await.unwrap();
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM schema_migrations WHERE version = 16"
)
.fetch_one(&db.pool)
.await
.unwrap(),
1,
"v16 must be recorded exactly once"
);
assert_eq!(
db.consume_signature(&k, "did:key:zAlice", 1_000, 1_330)
.await
.unwrap(),
ConsumeSignature::Replayed,
"re-running migrations must not drop or recreate the ledger"
);
// The skip-by-version check is only half of it. Force v16's statements
// to actually execute a second time (a node that lost its
// `schema_migrations` row, or a re-applied deploy) and they must still
// succeed against the objects they already created.
sqlx::query("DELETE FROM schema_migrations WHERE version = 16")
.execute(&db.pool)
.await
.unwrap();
db.run_migrations()
.await
.expect("v16's own statements must be safe to re-execute");
assert_eq!(
db.consume_signature(&k, "did:key:zAlice", 1_000, 1_330)
.await
.unwrap(),
ConsumeSignature::Replayed,
"re-executing v16 must not wipe the ledger"
);
}
/// A version recorded under a *different* name than the catalogue defines
/// means two branches independently claimed the same version number. The
/// old applied-check only looked at `version`, so the loser's migration was
/// silently skipped and the schema quietly went missing. The runner must
/// refuse to start and name the version, the expected name, and the name
/// the database actually recorded.
#[sqlx::test]
async fn migration_version_applied_under_other_name_is_rejected(pool: PgPool) {
let db = Db::for_testing(pool);
db.run_migrations().await.unwrap();
// Rewrite v16's recorded name as if a colliding branch had applied its
// own v16 first. The ledger objects stay in place; only the bookkeeping
// disagrees, which is exactly the undetectable case.
sqlx::query("UPDATE schema_migrations SET name = $1 WHERE version = 16")
.bind("something_else")
.execute(&db.pool)
.await
.unwrap();
let err = db
.run_migrations()
.await
.expect_err("a name mismatch on an applied version must abort startup");
let msg = format!("{err:#}");
assert!(
msg.contains("16"),
"error must name the colliding version, got: {msg}"
);
assert!(
msg.contains("consumed_signatures"),
"error must name the migration this build expects, got: {msg}"
);
assert!(
msg.contains("something_else"),
"error must name what the database recorded, got: {msg}"
);
}
/// The guard must not be a blanket failure: a database whose recorded names
/// all match the catalogue migrates cleanly, and a version that is not yet
/// applied still applies normally even while earlier versions are present.
#[sqlx::test]
async fn matching_names_and_pending_versions_still_migrate(pool: PgPool) {
let db = Db::for_testing(pool);
db.run_migrations().await.unwrap();
// (a) Every recorded name matches -> a second run is a clean no-op.
db.run_migrations()
.await
.expect("matching names must not trip the guard");
for m in MIGRATIONS {
let recorded: String =
sqlx::query_scalar("SELECT name FROM schema_migrations WHERE version = $1")
.bind(m.version)
.fetch_one(&db.pool)
.await
.unwrap();
assert_eq!(recorded, m.name, "v{} name must round-trip", m.version);
}
// (b) A pending version still applies: drop v16's bookkeeping (and its
// table) so it is genuinely unapplied while every lower version remains recorded
// with matching names.
sqlx::query("DROP TABLE IF EXISTS consumed_signatures")
.execute(&db.pool)
.await
.unwrap();
sqlx::query("DELETE FROM schema_migrations WHERE version = 16")
.execute(&db.pool)
.await
.unwrap();
db.run_migrations()
.await
.expect("an unapplied version must still apply");
assert_eq!(
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM information_schema.tables
WHERE table_name = 'consumed_signatures'"
)
.fetch_one(&db.pool)
.await
.unwrap(),
1,
"the pending migration must have run"
);
}
}
/// Exercises the iCaptcha propagation quarantine: the `quarantined` flag on
/// repos and its interaction with `upsert_mirror_repo` and the listing surfaces.
#[cfg(test)]
+3 -25
View File
@@ -165,13 +165,9 @@ impl IntoResponse for AppError {
AppError::UnresolvableDid(msg) => {
(StatusCode::BAD_REQUEST, "unresolvable_did", msg.clone())
}
// Same wire code as the flood brake's `too_many_requests`, taken
// from the shared enum rather than retyped so the two cannot drift.
AppError::TooManyRequests(msg) => (
StatusCode::TOO_MANY_REQUESTS,
gitlawb_core::node_denial::NodeDenial::RateLimited.as_str(),
msg.clone(),
),
AppError::TooManyRequests(msg) => {
(StatusCode::TOO_MANY_REQUESTS, "rate_limited", msg.clone())
}
AppError::Incomplete(msg) => {
(StatusCode::UNPROCESSABLE_ENTITY, "incomplete", msg.clone())
}
@@ -242,13 +238,6 @@ impl IntoResponse for AppError {
}
let mut resp = (status, Json(body)).into_response();
if matches!(self, AppError::TooManyRequests(_)) {
use axum::http::HeaderValue;
let code = gitlawb_core::node_denial::NodeDenial::RateLimited.as_str();
if let Ok(v) = HeaderValue::from_str(code) {
resp.headers_mut().insert("X-Gitlawb-Error", v);
}
}
// Both retryable 503s advertise when to retry: Overloaded (capacity shed) and
// SearchIncomplete (a bounded CID search cut short by a cap — retry may complete
// it). They ride the shared tail above for body/status, so the header is attached
@@ -285,17 +274,6 @@ mod tests {
);
}
#[test]
fn too_many_requests_carries_rate_limited_header() {
use axum::http::HeaderValue;
let resp = AppError::TooManyRequests("slow down".into()).into_response();
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(
resp.headers().get("X-Gitlawb-Error"),
Some(&HeaderValue::from_static("rate_limited"))
);
}
#[test]
fn overloaded_maps_to_503_with_retry_after() {
let resp = AppError::Overloaded("x".into()).into_response();
+18 -370
View File
@@ -32,7 +32,7 @@ use axum::routing::get;
use axum::{Json, Router};
use clap::Parser;
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::sync::watch;
@@ -51,35 +51,14 @@ struct DegradedState {
db_startup: Arc<DbStartupStatus>,
}
/// Independent fields with no cross-field invariant — atomics, not a lock, so
/// the retry loop and the degraded handlers never contend.
/// 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,
/// Set once the retry loop sees a migration version collision. Latched:
/// the condition is a code-level numbering bug that cannot heal between
/// attempts, and it changes what the readiness payload must say.
schema_conflict: AtomicBool,
}
impl DbStartupStatus {
fn mark_schema_conflict(&self) {
self.schema_conflict.store(true, Ordering::Relaxed);
}
fn schema_conflict(&self) -> bool {
self.schema_conflict.load(Ordering::Relaxed)
}
}
/// Distinct from `error::DB_UNAVAILABLE_CODE`: the database is reachable and
/// healthy, the schema it holds disagrees with this build. Retrying will never
/// clear it, so the readiness payload must not read as "still initializing".
const DB_SCHEMA_CONFLICT_CODE: &str = "db_schema_conflict";
const DB_SCHEMA_CONFLICT_MESSAGE: &str =
"database schema conflicts with this build; operator action is required";
/// Hard ceiling on the advisory-lock pool's `max_connections`.
///
/// `max_concurrent_git_pushes` is validated all the way up to 1_048_576, and the lock
@@ -417,22 +396,6 @@ async fn main() -> Result<()> {
std::time::Duration::from_secs(3600),
200_000,
);
// Flood brake for the signed write routes. Every signed attempt there spends
// a `consumed_signatures` row before the handler checks anything, and any
// freshly minted did:key signs, so the brake is what bounds the durable
// writes an unregistered caller can force. Its own bucket so it cannot drain
// the creation, push, or peer-sync quotas. Bounded key set (a
// client-influenced IP); 0 disables.
let signed_write_rate_limiter = rate_limit::RateLimiter::new_bounded(
config.signed_write_rate_limit,
std::time::Duration::from_secs(3600),
200_000,
);
if config.signed_write_rate_limit == 0 {
tracing::warn!(
"GITLAWB_SIGNED_WRITE_RATE_LIMIT=0 — signed-write IP rate limiting disabled"
);
}
if config.sync_trigger_rate_limit == 0 {
tracing::warn!(
"GITLAWB_SYNC_TRIGGER_RATE_LIMIT=0 — /sync/trigger IP rate limiting disabled"
@@ -479,7 +442,6 @@ async fn main() -> Result<()> {
push_limiter_trust,
sync_trigger_rate_limiter,
peer_write_rate_limiter,
signed_write_rate_limiter,
shutdown_tx: shutdown_tx.clone(),
git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_git_ops)),
git_write_semaphore: Arc::new(tokio::sync::Semaphore::new(
@@ -592,8 +554,7 @@ async fn main() -> Result<()> {
});
}
// Periodic cleanup of expired rate limit entries, the consumed-proof
// ledger, and the HTTP-signature replay ledger
// Periodic cleanup of expired rate limit entries + consumed-proof ledger
{
let cleanup_state = state.clone();
let db = state.db.clone();
@@ -602,6 +563,8 @@ async fn main() -> Result<()> {
loop {
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(300)) => {
// Sweep every per-IP/DID limiter (incl. the ipfs walk brake)
// so bounded maps shed stale keys instead of sitting at cap.
cleanup_state.sweep_rate_limiters().await;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@@ -610,9 +573,6 @@ async fn main() -> Result<()> {
if let Err(e) = db.sweep_expired_proofs(now).await {
tracing::warn!(err = %e, "failed to sweep expired iCaptcha proofs");
}
if let Err(e) = db.sweep_expired_signatures(now).await {
tracing::warn!(err = %e, "failed to sweep expired signature ledger rows");
}
}
_ = shutdown_rx.changed() => {
if *shutdown_rx.borrow() {
@@ -861,14 +821,7 @@ async fn connect_db_with_retry(
// 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.
// One call decides the backoff, the log level and the readiness
// signal, and applies the latch. Everything below reads from
// its result, so the branch cannot drift away from the latch.
let fault = record_db_startup_failure(&err, &db_startup);
let DbStartupFault {
permanent,
schema_conflict,
} = fault;
let permanent = is_likely_permanent_db_error(&err);
let retry_secs = if permanent {
max_retry_secs
} else {
@@ -877,14 +830,7 @@ async fn connect_db_with_retry(
db_startup
.next_retry_secs
.store(retry_secs, Ordering::Relaxed);
if schema_conflict {
tracing::error!(
attempts,
retry_secs,
err = %err,
"database schema conflicts with this build's migration catalogue; this will not heal on retry; operator action is required"
);
} else if permanent {
if permanent {
tracing::error!(
attempts,
retry_secs,
@@ -913,52 +859,12 @@ async fn connect_db_with_retry(
}
}
/// What a failed startup attempt means for the retry loop and for what the
/// node tells the outside world.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct DbStartupFault {
/// Jump straight to the maximum backoff and log at error level.
permanent: bool,
/// The schema disagrees with this build. Latched into `DbStartupStatus`,
/// so the degraded 503 body stops saying "initializing".
schema_conflict: bool,
}
/// Classify a failed startup attempt **and** record its consequence on
/// `db_startup`. The two are one function on purpose: a classification that
/// never reaches the readiness payload leaves /ready reporting "database
/// initializing" forever for a fault that cannot heal, and separate
/// classify-then-latch steps can silently drift apart. `connect_db_with_retry`
/// derives its backoff and log level from the returned value, so the call site
/// cannot keep the decision while dropping the latch.
fn record_db_startup_failure(err: &anyhow::Error, db_startup: &DbStartupStatus) -> DbStartupFault {
let schema_conflict = err
.downcast_ref::<db::MigrationVersionCollision>()
.is_some();
if schema_conflict {
db_startup.mark_schema_conflict();
}
DbStartupFault {
permanent: is_likely_permanent_db_error(err),
schema_conflict,
}
}
/// Errors that indicate misconfiguration rather than a transient outage: a
/// malformed DATABASE_URL, a server that answered and rejected us — Postgres
/// error class 28xxx (invalid authorization) or 3D000 (database does not
/// exist), or a migration version collision. Best-effort: an error that anyhow
/// can't downcast back to a known type just counts as transient.
/// 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 {
// A version collision is a code-level numbering bug. It is the loudest
// thing the migration guard can report, so it must never be filed under
// "database unavailable, retrying".
if err
.downcast_ref::<db::MigrationVersionCollision>()
.is_some()
{
return true;
}
match err.downcast_ref::<sqlx::Error>() {
Some(sqlx::Error::Configuration(_)) => true,
Some(sqlx::Error::Database(db)) => db
@@ -1022,27 +928,11 @@ fn build_degraded_router(node_did: String, db_startup: Arc<DbStartupStatus>) ->
/// 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 {
// A schema conflict is not a database that is still coming up. Saying
// "initializing" for it hides a permanent, operator-actionable fault behind
// a status that reads as "wait longer".
let (state, code, message) = if db_startup.schema_conflict() {
(
"schema_conflict",
DB_SCHEMA_CONFLICT_CODE,
DB_SCHEMA_CONFLICT_MESSAGE,
)
} else {
(
"initializing",
error::DB_UNAVAILABLE_CODE,
error::DB_UNAVAILABLE_MESSAGE,
)
};
serde_json::json!({
"status": "degraded",
"database": state,
"error": code,
"message": message,
"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),
})
@@ -1140,40 +1030,6 @@ fn build_operator_client(
Ok(operator::OperatorClient::new(cfg))
}
/// A bootstrap peer's answer to our signed announce, reduced to what the
/// caller has to act on.
#[derive(Debug, PartialEq, Eq)]
enum AnnounceOutcome {
Accepted,
/// Any non-2xx. Carries the peer's `x-gitlawb-error` code when it sent
/// one, which is what makes a spent-signature-ledger rejection (429
/// `signature_ledger_full`, 409 `signature_replayed`) diagnosable rather
/// than just "some 4xx".
Rejected {
status: u16,
error_code: Option<String>,
},
}
/// Classify a bootstrap announce response.
///
/// Split out from `gossip_task` so the rejection path is unit-testable: the
/// whole defect was that a non-2xx fell off the end of an `if` with no `else`,
/// leaving a node whose announces are being refused looking exactly like one
/// succeeding.
fn classify_announce_response(
status: reqwest::StatusCode,
headers: &reqwest::header::HeaderMap,
) -> AnnounceOutcome {
if status.is_success() {
return AnnounceOutcome::Accepted;
}
AnnounceOutcome::Rejected {
status: status.as_u16(),
error_code: api::peers::peer_error_code(headers).map(str::to_string),
}
}
/// Announce to bootstrap peers on startup, then periodically ping all known peers.
async fn gossip_task(
state: AppState,
@@ -1238,8 +1094,8 @@ async fn gossip_task(
)
.await
{
Ok(Ok(resp)) => match classify_announce_response(resp.status(), resp.headers()) {
AnnounceOutcome::Accepted => {
Ok(Ok(resp)) => {
if resp.status().is_success() {
if let Ok(json) = resp.json::<serde_json::Value>().await {
// Add them back to our peer list
if let (Some(their_did), Some(their_url)) = (
@@ -1278,19 +1134,7 @@ async fn gossip_task(
}
}
}
// Without this arm a refused announce produced no log line at
// all, so a node whose peers reject it looked identical to one
// succeeding. Same shape as the two sibling signed calls
// (sync.rs replica registration, repos.rs peer sync notify).
AnnounceOutcome::Rejected { status, error_code } => {
warn!(
url = %announce_url,
status = status,
error_code = error_code.as_deref().unwrap_or("none"),
"bootstrap peer announce rejected"
);
}
},
}
Ok(Err(e)) => {
tracing::warn!(url = %announce_url, err = %e, "failed to announce to bootstrap peer")
}
@@ -1985,199 +1829,3 @@ mod gossip_ssrf_tests {
assert!(!ok, "a connection error must count as an unready peer");
}
}
#[cfg(test)]
mod gossip_announce_tests {
use super::{classify_announce_response, AnnounceOutcome};
use reqwest::header::HeaderMap;
use reqwest::StatusCode;
fn headers(code: Option<&str>) -> HeaderMap {
let mut h = HeaderMap::new();
if let Some(c) = code {
h.insert("x-gitlawb-error", c.parse().unwrap());
}
h
}
#[test]
fn a_2xx_announce_is_accepted() {
assert_eq!(
classify_announce_response(StatusCode::OK, &headers(None)),
AnnounceOutcome::Accepted
);
}
// The defect: every non-2xx fell off the end of an `if` with no `else`, so
// a node whose announces were being refused logged nothing at all. Each of
// these must come back Rejected, carrying the status and, when the peer
// sent one, the x-gitlawb-error code that names the reason.
#[test]
fn a_ledger_rejection_is_reported_with_its_error_code() {
assert_eq!(
classify_announce_response(
StatusCode::TOO_MANY_REQUESTS,
&headers(Some("signature_ledger_full"))
),
AnnounceOutcome::Rejected {
status: 429,
error_code: Some("signature_ledger_full".to_string()),
}
);
assert_eq!(
classify_announce_response(StatusCode::CONFLICT, &headers(Some("signature_replayed"))),
AnnounceOutcome::Rejected {
status: 409,
error_code: Some("signature_replayed".to_string()),
}
);
assert_eq!(
classify_announce_response(
StatusCode::SERVICE_UNAVAILABLE,
&headers(Some("signature_ledger_unavailable"))
),
AnnounceOutcome::Rejected {
status: 503,
error_code: Some("signature_ledger_unavailable".to_string()),
}
);
}
// A proxy that strips unknown X- headers, or a peer that never sets one,
// must still produce a rejection with its status rather than silence.
#[test]
fn a_rejection_without_an_error_code_is_still_reported() {
assert_eq!(
classify_announce_response(StatusCode::FORBIDDEN, &headers(None)),
AnnounceOutcome::Rejected {
status: 403,
error_code: None,
}
);
assert_eq!(
classify_announce_response(StatusCode::INTERNAL_SERVER_ERROR, &headers(None)),
AnnounceOutcome::Rejected {
status: 500,
error_code: None,
}
);
}
}
#[cfg(test)]
mod db_startup_signal_tests {
use super::{
degraded_body, is_likely_permanent_db_error, record_db_startup_failure, DbStartupStatus,
};
use crate::db::MigrationVersionCollision;
fn collision_error() -> anyhow::Error {
anyhow::Error::new(MigrationVersionCollision {
version: 16,
recorded_name: "pinned_cids_repo_provenance".to_string(),
build_name: "consumed_signatures",
})
// Wrapped the way the real call path wraps it, so the classifier has
// to look through the context chain.
.context("connecting to postgres")
}
// The wire between the classifier and the readiness payload. Classifying a
// collision correctly is useless if nothing latches it: the backoff and the
// log level would be right while /ready still reported "initializing"
// forever, which is the defect. This drives the same function
// `connect_db_with_retry` calls on every failed attempt, so the latch is
// what is under test, not just its two endpoints.
#[test]
fn a_collision_latches_the_readiness_signal() {
let status = DbStartupStatus::default();
assert!(!status.schema_conflict(), "precondition: not yet latched");
let fault = record_db_startup_failure(&collision_error(), &status);
assert!(fault.schema_conflict, "the fault must name the conflict");
assert!(fault.permanent, "and classify it as permanent");
assert!(
status.schema_conflict(),
"the collision must latch the startup status, or the readiness \
payload keeps claiming the database is initializing"
);
assert_eq!(
degraded_body(&status)["database"],
"schema_conflict",
"the latch must reach the payload the degraded router serves"
);
}
#[test]
fn an_ordinary_outage_does_not_latch_a_schema_conflict() {
let status = DbStartupStatus::default();
let fault = record_db_startup_failure(
&anyhow::Error::new(sqlx::Error::PoolTimedOut).context("connecting to postgres"),
&status,
);
assert!(!fault.schema_conflict);
assert!(!fault.permanent);
assert!(
!status.schema_conflict(),
"a transient outage must not latch the conflict signal"
);
assert_eq!(degraded_body(&status)["database"], "initializing");
}
// Defect 1: a schema collision never heals on its own, so it must classify
// as permanent and take the error-level, operator-action branch, not the
// "database unavailable during startup; retrying" warn branch.
#[test]
fn a_migration_collision_is_permanent() {
assert!(
is_likely_permanent_db_error(&collision_error()),
"a migration version collision must classify as permanent"
);
}
#[test]
fn an_ordinary_connection_error_is_still_transient() {
let err = anyhow::Error::new(sqlx::Error::PoolTimedOut).context("connecting to postgres");
assert!(
!is_likely_permanent_db_error(&err),
"a pool timeout is a transient outage, not a permanent misconfiguration"
);
let io = anyhow::Error::new(sqlx::Error::Io(std::io::Error::new(
std::io::ErrorKind::ConnectionRefused,
"connection refused",
)))
.context("connecting to postgres");
assert!(
!is_likely_permanent_db_error(&io),
"a refused TCP connection is a transient outage"
);
}
// Defect 1, readiness half: the payload must not claim the database is
// merely initializing when the real cause is a schema conflict.
#[test]
fn the_degraded_body_distinguishes_a_schema_conflict_from_an_ordinary_wait() {
let waiting = DbStartupStatus::default();
let body = degraded_body(&waiting);
assert_eq!(body["database"], "initializing");
assert_eq!(body["error"], crate::error::DB_UNAVAILABLE_CODE);
let conflicted = DbStartupStatus::default();
conflicted.mark_schema_conflict();
let body = degraded_body(&conflicted);
assert_eq!(
body["database"], "schema_conflict",
"a schema conflict must not read as 'initializing'"
);
assert_eq!(body["error"], super::DB_SCHEMA_CONFLICT_CODE);
assert_ne!(
body["error"],
crate::error::DB_UNAVAILABLE_CODE,
"the conflict must carry its own code"
);
assert_eq!(body["status"], "degraded");
}
}
-49
View File
@@ -7,8 +7,6 @@
//! * is fetch traffic flowing? — `gitlawb_fetches_total`
//! * are signature checks passing or failing? —
//! `gitlawb_auth_successes_total` / `gitlawb_auth_failures_total`
//! * is the spent-signature replay ledger admitting or refusing writes? —
//! `gitlawb_signature_ledger_total{outcome}`
//! * is the sync worker making progress? —
//! `gitlawb_sync_queue_processed_total{status}`
//! * are webhooks reaching their endpoints? —
@@ -49,7 +47,6 @@ static PUSHES: OnceLock<IntCounterVec> = OnceLock::new();
static FETCHES: OnceLock<IntCounterVec> = OnceLock::new();
static AUTH_SUCCESSES: OnceLock<IntCounterVec> = OnceLock::new();
static AUTH_FAILURES: OnceLock<IntCounterVec> = OnceLock::new();
static SIGNATURE_LEDGER: OnceLock<IntCounterVec> = OnceLock::new();
static SYNC_PROCESSED: OnceLock<IntCounterVec> = OnceLock::new();
static WEBHOOK_DELIVERIES: OnceLock<IntCounterVec> = OnceLock::new();
static PACK_SIZE: OnceLock<Histogram> = OnceLock::new();
@@ -60,17 +57,6 @@ static PEERS_CONNECTED: OnceLock<IntGauge> = OnceLock::new();
/// more than once is a silent no-op. MUST be called from `main()` after
/// the node DID is known.
pub fn init(version: &str, node_did: &str) {
// `Once` rather than a bare `REGISTRY.get().is_some()` check: the registry
// is published last, so that check leaves a window in which a second
// concurrent caller walks into the `set(..).expect(..)` calls below and
// panics on an already-set `OnceLock`. `main` calls this once, but tests
// call it from several threads. `call_once` makes the later callers block
// and then return, which is what the `expect`s already assume.
static INIT: std::sync::Once = std::sync::Once::new();
INIT.call_once(|| init_once(version, node_did));
}
fn init_once(version: &str, node_did: &str) {
// Guard with Once so two concurrent callers cannot both pass an unsynchronized
// `REGISTRY.get()` check and then race on `OnceLock::set(...).expect(...)`,
// panicking the loser (#192 F4). Once runs the body exactly once, so the
@@ -152,21 +138,6 @@ fn init_inner(version: &str, node_did: &str) {
.set(auth_failures)
.expect("set AUTH_FAILURES once");
let signature_ledger = IntCounterVec::new(
Opts::new(
"gitlawb_signature_ledger_total",
"Total requests seen by the spent-signature replay ledger, by outcome",
),
&["outcome"],
)
.expect("gitlawb_signature_ledger_total definition");
registry
.register(Box::new(signature_ledger.clone()))
.expect("register gitlawb_signature_ledger_total");
SIGNATURE_LEDGER
.set(signature_ledger)
.expect("set SIGNATURE_LEDGER once");
let sync_processed = IntCounterVec::new(
Opts::new(
"gitlawb_sync_queue_processed_total",
@@ -280,25 +251,6 @@ pub fn record_auth_failure(route: &str, reason: &str) {
}
}
/// Record one terminal outcome of the spent-signature replay ledger layer.
///
/// Every request that enters the layer lands on exactly one outcome, so the
/// series sum is the layer's request count and any one outcome is a rate over
/// it. `outcome` is a fixed set of literals chosen at the call sites in
/// `auth::consume_signature` — never a DID, keyid, or path, which would be
/// attacker-controlled and unbounded.
///
/// The sum is the LAYER's request count, not the routes' traffic. The per-client
/// flood brake (`rate_limit`) rejects upstream of this layer, so its 429s appear
/// in no series here even though they are served on the same routes. An operator
/// counting refusals has to read `rate_limited` responses separately; this
/// counter answers "what is the ledger doing", not "how many 429s went out".
pub fn record_signature_ledger(outcome: &str) {
if let Some(c) = SIGNATURE_LEDGER.get() {
c.with_label_values(&[outcome]).inc();
}
}
/// Record one sync_queue item outcome. `status` ∈ {done, failed, skipped,
/// rejected, deferred}. `rejected` is terminal and caused by a guard (a
/// malformed slug, or a path resolving outside repos_dir); `deferred` leaves
@@ -411,7 +363,6 @@ mod tests {
record_fetch("test/repo");
record_auth_success("test/route");
record_auth_failure("test/route", "test_reason");
record_signature_ledger("test_outcome");
record_sync_processed("done");
record_webhook_delivery("ok");
observe_pack_size(1024.0);
+16 -147
View File
@@ -4,12 +4,9 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use axum::extract::{ConnectInfo, Request};
use axum::http::{HeaderMap, Method, StatusCode};
use axum::http::{HeaderMap, StatusCode};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::Json;
use gitlawb_core::node_denial::NodeDenial;
use serde_json::json;
use tokio::sync::Mutex;
use crate::auth::AuthenticatedDid;
@@ -279,10 +276,12 @@ pub async fn rate_limit_by_did(request: Request, next: Next) -> Response {
if let (Some(limiter), Some(did)) = (limiter, did) {
if !limiter.check(&did).await {
// Same response as the per-IP brake: a client cannot act on the
// difference between the two buckets, and both need the code that
// tells them apart from the ledger's 429.
return too_many_requests();
return (
StatusCode::TOO_MANY_REQUESTS,
[("retry-after", "60")],
"rate limit exceeded — try again later",
)
.into_response();
}
}
@@ -361,18 +360,6 @@ pub fn client_key(
pub struct IpRateLimiter {
pub limiter: RateLimiter,
pub trust: TrustedProxy,
/// Exempt GET/HEAD from this bucket. The signed-write brake classifies
/// whole routers, but `write_routes` chains PUT/DELETE/GET on one path
/// (`/api/v1/repos/{owner}/{repo}/visibility`, `server.rs`), and `gl
/// visibility list` deliberately signs that owner-only GET — so without
/// this the read spends the write quota and a normal listing 429s after a
/// few writes from the same IP (one NAT egress makes the coupling vivid).
/// The method check is the same general form `consume_signature` uses for
/// its ledger skip: a read has no write side effect for the brake to
/// protect. Leave this `false` on any limiter whose job IS braking reads —
/// the `/ipfs/{cid}` brake exists precisely to cap GET-driven history
/// walks, so a blanket skip inside `rate_limit_by_ip` would delete it.
pub skip_reads: bool,
}
/// Infallible extractor for the socket peer address from `ConnectInfo`. Yields
@@ -397,24 +384,14 @@ impl<S: Send + Sync> axum::extract::FromRequestParts<S> for PeerAddr {
}
}
/// The shared 429 response for the flood brakes. Route-agnostic: this serves
/// the push path AND the peer-sync routes, so the message stays generic (the
/// offending path is recorded in the warn log at the call site).
///
/// Carries `X-Gitlawb-Error: rate_limited` and the same JSON shape the ledger's
/// `ledger_rejection` uses, because the brake shares 429 with the ledger's
/// `signature_ledger_full` on every route group the brake covers. Without a
/// code the only discriminator between two unrelated conditions is the
/// *absence* of a header, which is exactly what a proxy stripping unknown `X-`
/// headers produces. A client cannot act on a difference it cannot see: `gl`
/// would print "invalid JSON response" for a rate limit, and a peer's sync
/// sender would retry a brake it can never clear inside its budget.
/// The shared 429 response for the per-IP flood brakes. Route-agnostic: this
/// middleware now serves the push path AND the peer-sync routes, so the message
/// stays generic (the offending path is recorded in the warn log below).
pub fn too_many_requests() -> Response {
let code = NodeDenial::RateLimited.as_str();
(
StatusCode::TOO_MANY_REQUESTS,
[("retry-after", "60"), ("X-Gitlawb-Error", code)],
Json(json!({ "error": code, "message": "rate limit exceeded — try again later" })),
[("retry-after", "60")],
"rate limit exceeded — try again later",
)
.into_response()
}
@@ -430,16 +407,10 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response {
.map(|c| c.0);
if let Some(limiter) = limiter {
// Opt-in read exemption (see `IpRateLimiter::skip_reads`): only a
// limiter that declares itself write-only skips GET/HEAD, so the /ipfs
// read brake and every other read-facing bucket keep braking reads.
let is_read = matches!(*request.method(), Method::GET | Method::HEAD);
if !(limiter.skip_reads && is_read) {
if let Some(key) = client_key(request.headers(), peer, limiter.trust) {
if !limiter.limiter.check(&key).await {
tracing::warn!(key = %key, path = %request.uri().path(), "per-IP rate limit exceeded");
return too_many_requests();
}
if let Some(key) = client_key(request.headers(), peer, limiter.trust) {
if !limiter.limiter.check(&key).await {
tracing::warn!(key = %key, path = %request.uri().path(), "per-IP rate limit exceeded");
return too_many_requests();
}
}
}
@@ -451,41 +422,6 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response {
mod tests {
use super::*;
/// The brake shares 429 with the spent-signature ledger's
/// `signature_ledger_full`, so it has to be told apart by something other
/// than the status. It carries the same shape the ledger uses: an
/// `X-Gitlawb-Error` header and a JSON body whose `error` matches it.
#[tokio::test]
async fn brake_429_carries_a_machine_readable_code_and_json_body() {
let resp = too_many_requests();
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(
resp.headers()
.get("retry-after")
.and_then(|v| v.to_str().ok()),
Some("60"),
);
assert_eq!(
resp.headers()
.get("x-gitlawb-error")
.and_then(|v| v.to_str().ok()),
Some("rate_limited"),
"a brake 429 must be distinguishable from a ledger 429 by its code",
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.expect("brake body");
let body: serde_json::Value =
serde_json::from_slice(&bytes).expect("brake body must be JSON, not text/plain");
assert_eq!(body["error"], "rate_limited");
assert!(
body["message"]
.as_str()
.is_some_and(|m| m.contains("rate limit exceeded")),
"body: {body}",
);
}
#[test]
fn per_caller_concurrency_caps_one_caller_and_frees_on_drop() {
let lim = PerCallerConcurrency::new(2, 100);
@@ -775,7 +711,6 @@ mod tests {
let router = ip_limited_router(IpRateLimiter {
limiter: RateLimiter::new(2, Duration::from_secs(60)),
trust: TrustedProxy::None,
skip_reads: false,
});
let peer: SocketAddr = "203.0.113.7:5000".parse().unwrap();
// Two requests inside the budget, the third over it (shared Arc state).
@@ -792,7 +727,6 @@ mod tests {
let router = ip_limited_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::None,
skip_reads: false,
});
let a: SocketAddr = "203.0.113.1:1".parse().unwrap();
let b: SocketAddr = "203.0.113.2:1".parse().unwrap();
@@ -801,70 +735,6 @@ mod tests {
assert_eq!(post_from(&router, a).await, StatusCode::TOO_MANY_REQUESTS);
}
/// `skip_reads` mechanics, isolated from the production router: on a
/// limiter that declares itself write-only, GET and HEAD never touch the
/// bucket — they pass an already-exhausted one AND never debit it — while
/// a mutation from the same peer stays braked. On a limiter that does not
/// opt in (the /ipfs read brake), GET is braked exactly as before. The
/// production wiring (the signed-write brake on `write_routes`, whose one
/// GET is the signed visibility listing) is proven end-to-end by
/// `signed_write_brake_rejects_writes_but_serves_visibility_get` in
/// `test_support`; this pins the middleware seam it rides on.
#[tokio::test]
async fn skip_reads_exempts_get_and_head_but_still_brakes_writes() {
use tower::ServiceExt;
let router = |skip_reads: bool| {
axum::Router::new()
.route(
"/thing",
axum::routing::get(|| async { StatusCode::OK })
.head(|| async { StatusCode::OK })
.put(|| async { StatusCode::OK }),
)
.layer(axum::middleware::from_fn(rate_limit_by_ip))
.layer(axum::Extension(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::None,
skip_reads,
}))
};
let peer: SocketAddr = "203.0.113.9:5000".parse().unwrap();
let send = |router: &axum::Router, method: axum::http::Method| {
let mut req = axum::http::Request::builder()
.method(method)
.uri("/thing")
.body(axum::body::Body::empty())
.unwrap();
req.extensions_mut().insert(ConnectInfo(peer));
let router = router.clone();
async move { router.oneshot(req).await.unwrap().status() }
};
// Opted-in limiter: the PUT spends the single slot...
let r = router(true);
assert_eq!(send(&r, axum::http::Method::PUT).await, StatusCode::OK);
// ...reads from the same peer still pass the exhausted bucket...
assert_eq!(send(&r, axum::http::Method::GET).await, StatusCode::OK);
assert_eq!(send(&r, axum::http::Method::HEAD).await, StatusCode::OK);
// ...and did not debit it behind the scenes some other way — the next
// mutation is still the one that gets braked.
assert_eq!(
send(&r, axum::http::Method::PUT).await,
StatusCode::TOO_MANY_REQUESTS,
"the write brake must keep braking writes while exempting reads"
);
// Non-opted-in limiter: GET is braked like any other method, so the
// exemption cannot silently spread to read-braking buckets like /ipfs.
let r = router(false);
assert_eq!(send(&r, axum::http::Method::GET).await, StatusCode::OK);
assert_eq!(
send(&r, axum::http::Method::GET).await,
StatusCode::TOO_MANY_REQUESTS,
"a limiter that did not opt in must keep braking reads"
);
}
// ── creation-route layering: IP brake runs BEFORE auth ──────────────
/// A minimal stand-in for the creation route group: an inner "auth" layer
@@ -910,7 +780,6 @@ mod tests {
let router = ip_limited_over_auth_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::None,
skip_reads: false,
});
let peer: SocketAddr = "203.0.113.42:7000".parse().unwrap();
+11 -69
View File
@@ -42,33 +42,12 @@ async fn graphql_playground() -> impl IntoResponse {
))
}
/// Applies the standard auth middleware stack to a router. Axum runs the LAST
/// `.layer` first, so the three below run in this order for every matched
/// request: `require_signature` (verifies the RFC 9421 signature, sets
/// `AuthenticatedDid` and `SignatureIdentity`), then `require_ucan_chain`
/// (reads the DID), then `consume_signature` (spends the signature).
///
/// The ledger runs last of the three on purpose. If it ran before
/// `require_ucan_chain`, a request with a good signature but a rejected UCAN
/// would burn its key and the client could not retry those exact bytes.
/// Consuming before the handler, rather than inside it, is what closes the
/// concurrent-replay race: two copies of the same signature race the ledger, not
/// the handler.
///
/// It is not the last check overall, so some rejections do still cost a ledger
/// row. Any layer added inside the router passed in runs AFTER the ledger
/// (`creation_routes` puts `rate_limit_by_did` there), and every check the
/// handler itself makes (the iCaptcha 403, ownership, existence) runs later
/// still. A request refused by the per-DID throttle, by iCaptcha, or by the
/// handler has already spent its signature and must re-sign to retry. Only a
/// per-IP brake layered outside this stack rejects a request before the ledger
/// is charged.
/// Applies the standard auth middleware pair to a router: HTTP Signature verification
/// followed by UCAN chain validation. The two layers run in this order for every
/// matched request: `require_signature` first (sets `AuthenticatedDid`), then
/// `require_ucan_chain` (reads it).
fn add_auth_layers(router: Router<AppState>, state: AppState) -> Router<AppState> {
router
.layer(middleware::from_fn_with_state(
state.clone(),
auth::consume_signature,
))
.layer(middleware::from_fn_with_state(
state,
auth::require_ucan_chain,
@@ -87,28 +66,6 @@ pub fn build_router(state: AppState) -> Router {
.layer(middleware::from_fn(auth::optional_signature))
.route_service("/graphql/ws", GraphQLSubscription::new(schema));
// ── Per-IP flood brake for the signed write routes ────────────────────
// Every signed attempt on a route wrapped in `add_auth_layers` commits a
// `consumed_signatures` row before the handler runs any authorization or
// existence check, and `require_signature` resolves the key from the did:key
// itself, so any freshly minted keypair can force that durable write on a
// repo it does not own or that does not exist. The five groups below carried
// no brake at all; this one caps them on the resolved client IP. Its own
// bucket, so a flood here cannot drain the creation, push, or peer-sync
// quotas. Each group layers it OUTSIDE `add_auth_layers` (axum runs the last
// `.layer` first) so an over-limit request is rejected before signature
// verification burns CPU and before the ledger is charged.
// `skip_reads`: `write_routes` chains PUT/DELETE/GET on the visibility
// path, and `gl visibility list` signs that owner-only GET. A write-only
// brake must not let reads spend the write quota (or a listing starts
// 429ing behind a shared NAT after a handful of writes), mirroring the
// method skip `consume_signature` applies to the ledger for the same GET.
let signed_write_ip_limiter = rate_limit::IpRateLimiter {
limiter: state.signed_write_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: true,
};
// ── Task routes (write — require HTTP Signature) ───────────────────────
let task_write_routes = add_auth_layers(
Router::new()
@@ -117,9 +74,7 @@ pub fn build_router(state: AppState) -> Router {
.route("/api/v1/tasks/{id}/complete", post(tasks::complete_task))
.route("/api/v1/tasks/{id}/fail", post(tasks::fail_task)),
state.clone(),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(signed_write_ip_limiter.clone()));
);
// ── Task routes (read — open) ──────────────────────────────────────────
let task_read_routes = Router::new()
@@ -138,7 +93,6 @@ pub fn build_router(state: AppState) -> Router {
let create_ip_limiter = rate_limit::IpRateLimiter {
limiter: state.create_ip_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: false,
};
let creation_routes = add_auth_layers(
Router::new()
@@ -157,7 +111,7 @@ pub fn build_router(state: AppState) -> Router {
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(create_ip_limiter));
// ── Write routes — require HTTP Signature, plus the signed-write IP brake ─
// ── Write routes — require HTTP Signature (no rate limit) ─────────────
let write_routes = add_auth_layers(
Router::new()
.route(
@@ -227,9 +181,7 @@ pub fn build_router(state: AppState) -> Router {
axum::routing::delete(agents::deregister_agent),
),
state.clone(),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(signed_write_ip_limiter.clone()));
);
// Body limit is raised to GITLAWB_MAX_PACK_BYTES (default 2 GB) for git
// routes only — all other API routes keep axum's default 2 MB cap.
@@ -243,7 +195,6 @@ pub fn build_router(state: AppState) -> Router {
let push_limiter = rate_limit::IpRateLimiter {
limiter: state.push_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: false,
};
let git_write_routes = add_auth_layers(
Router::new()
@@ -271,7 +222,6 @@ pub fn build_router(state: AppState) -> Router {
let ipfs_limiter = rate_limit::IpRateLimiter {
limiter: state.ipfs_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: false,
};
let ipfs_routes = Router::new()
.route("/ipfs/{cid}", get(ipfs::get_by_cid))
@@ -308,9 +258,7 @@ pub fn build_router(state: AppState) -> Router {
post(bounties::dispute_bounty),
),
state.clone(),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(signed_write_ip_limiter.clone()));
);
// ── Bounty routes (read — open) ──────────────────────────────────────
let bounty_read_routes = Router::new()
@@ -331,11 +279,9 @@ pub fn build_router(state: AppState) -> Router {
let profile_write_routes = add_auth_layers(
Router::new().route("/api/v1/profile", axum::routing::put(profiles::set_profile)),
state.clone(),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(signed_write_ip_limiter.clone()));
);
// ── Issue routes (write — require HTTP Signature) ────────────────────
// ── Issue routes (write — require HTTP Signature, no rate limit) ─────
let issue_write_routes = add_auth_layers(
Router::new()
.route(
@@ -347,9 +293,7 @@ pub fn build_router(state: AppState) -> Router {
post(issues::create_issue_comment),
),
state.clone(),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(signed_write_ip_limiter));
);
// ── Peer discovery routes ─────────────────────────────────────────────
// Peer writes accept signatures when present and can require them after a
@@ -372,7 +316,6 @@ pub fn build_router(state: AppState) -> Router {
.layer(axum::Extension(rate_limit::IpRateLimiter {
limiter: state.sync_trigger_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: false,
}));
// announce + notify keep their rolling-upgrade signature behavior (unsigned
@@ -393,7 +336,6 @@ pub fn build_router(state: AppState) -> Router {
.layer(axum::Extension(rate_limit::IpRateLimiter {
limiter: state.peer_write_rate_limiter.clone(),
trust: state.push_limiter_trust,
skip_reads: false,
}));
// ── Read routes — open for public repos ───────────────────────────────
-10
View File
@@ -186,16 +186,6 @@ pub struct AppState {
/// sink as trigger and accepts unsigned requests from known peers, so it is
/// braked too; each peer's distinct IP gets its own bucket.
pub peer_write_rate_limiter: RateLimiter,
/// Per-client-IP limiter for the signed write routes (tasks, pull-request
/// merge/close/review/comment, webhooks, branch protection, stars, replicas,
/// labels, visibility, agent deregistration, bounties, profile, issue
/// close/comment). Every signed attempt on those routes commits a
/// `consumed_signatures` row before the handler checks ownership or
/// existence, and `require_signature` resolves the key from the did:key
/// itself, so any freshly minted keypair can force that durable write. Its
/// own bucket so a flood here cannot drain the creation, push, or peer-sync
/// quotas. Keyed by `push_limiter_trust`.
pub signed_write_rate_limiter: RateLimiter,
/// Process-wide graceful-shutdown signal. Sending `true` causes every
/// task that holds a `watch::Receiver` to exit at its next await point.
/// Used by:
File diff suppressed because it is too large. Load diff
+12 -620
View File
@@ -3,32 +3,18 @@
//! Writes are signed with RFC 9421 HTTP Signatures. When the node gates a write
//! behind iCaptcha (HTTP 403 `icaptcha_proof_required`, advertised via the
//! `x-icaptcha-url` / `x-icaptcha-level` headers), the client transparently
//! solves the challenge and re-signs the write with the `x-icaptcha-proof`
//! header attached (see `crates/icaptcha-client`).
//! solves the challenge and retries the same signed request with the
//! `x-icaptcha-proof` header — see `crates/icaptcha-client`.
use anyhow::{Context, Result};
use gitlawb_core::http_sig::sign_request;
use gitlawb_core::identity::Keypair;
use gitlawb_core::node_denial::NodeDenial;
use icaptcha_client::IcaptchaCfg;
/// Max times we'll fetch a fresh proof and retry a 403-iCaptcha response
/// (absorbs proof expiry / first-seen replay).
const MAX_ICAPTCHA_RETRIES: usize = 2;
/// Max bytes buffered from a denial body before we give up on parsing it. Node
/// error bodies are a few hundred bytes of JSON; anything past this is not one,
/// and reading it unbounded is exactly the allocation a hostile node would aim
/// for.
pub(crate) const DENIAL_BODY_CAP: usize = 64 * 1024;
/// Max bytes read from a git smart-HTTP ref advertisement. Real ads are small;
/// this bounds hostile 200 responses on the MCP `git_refs` path.
pub(crate) const GIT_REFS_BODY_CAP: usize = 256 * 1024;
/// Max characters of a node-supplied message we will echo to the terminal.
const NODE_MSG_CHARS: usize = 200;
/// Total request timeout: from the start of connecting through the end of the
/// response body, so it bounds a slow download and not just a slow handshake.
const TOTAL_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
@@ -172,12 +158,9 @@ impl NodeClient {
}
/// Sign + send a write. On a 403 iCaptcha challenge (detected via the
/// `x-icaptcha-*` headers) attach the proof and send the write again, up to
/// [`MAX_ICAPTCHA_RETRIES`]. Each attempt goes through `send_once`, which
/// signs afresh, so the retry is a new signature over the same bytes, not a
/// resend of the original one. Emits an actionable hint on a 401 "not an
/// agent" (the old-CLI / unregistered failure mode), and converts every
/// denial the node names in `x-gitlawb-error` into an error.
/// `x-icaptcha-*` headers) solve it and retry the same signed request with
/// the proof header, up to [`MAX_ICAPTCHA_RETRIES`]. Emits an actionable
/// hint on a 401 "not an agent" (the old-CLI / unregistered failure mode).
async fn send_signed(
&self,
method: &str,
@@ -211,10 +194,6 @@ impl NodeClient {
continue;
}
}
if let Some(rejection) = signature_rejection(&resp) {
return Err(rejection.into_error(method, path, resp).await);
}
return Ok(resp);
}
}
@@ -270,145 +249,6 @@ impl NodeClient {
}
}
/// A write the node refused with a code this build recognises.
struct SignatureRejection {
/// Which denial it was. Its `as_str` is the `x-gitlawb-error` code, printed
/// verbatim so scripts can match on it.
denial: NodeDenial,
/// What the user should actually do next.
hint: &'static str,
}
impl SignatureRejection {
/// Consume the response to fold the node's own message into the error.
/// Called only after [`signature_rejection`] has decided from the status and
/// header, since reading the body takes the response by value.
///
/// The message is attacker controlled (any node the user points `gl` at
/// wrote it), so the read is capped and the string is stripped of control
/// characters and bidi overrides before it can reach a terminal: without
/// that, the text whose whole job is to say "your write was REFUSED" can
/// clear the screen and print a fake success line.
async fn into_error(self, method: &str, path: &str, resp: reqwest::Response) -> anyhow::Error {
let status = resp.status();
let capped = read_body_capped(resp, DENIAL_BODY_CAP).await;
let node_msg = serde_json::from_str::<serde_json::Value>(&capped.text)
.ok()
.and_then(|b| b["message"].as_str().map(sanitize_node_msg))
.filter(|m| !m.is_empty());
let Self { denial, hint } = self;
let code = denial.as_str();
match node_msg {
Some(m) => anyhow::anyhow!("{method} {path} rejected ({status} {code}): {hint} ({m})"),
None => anyhow::anyhow!("{method} {path} rejected ({status} {code}): {hint}"),
}
}
}
/// Recognise a node rejection from the status and the
/// `x-gitlawb-error` header, before anything reads the body.
///
/// Known limit: the node also puts the same code in the body's `error` field,
/// and this does not look at it, so a proxy or CDN that strips unknown `X-`
/// headers switches the detection off. Reading the body to recover the second
/// signal is destructive, and handing the caller back a response it can still
/// parse needs `http::Response::builder` (the only constructor
/// `reqwest::Response::from` takes), which is a dependency this crate does not
/// carry. The commands that must not mistake a denial for a success do not rely
/// on this alone: `gl init` compares the structured error code from the body,
/// and the `gl task` writes check the status before parsing.
///
/// None of these is ever retried. A `signature_replayed` 409 means the node
/// already admitted a request carrying this signature, so re-sending the same
/// bytes risks applying the mutation twice, which is the duplicate write the
/// ledger exists to prevent. (The iCaptcha 403 retry above does not have that
/// problem: `verify_request` runs inside the handler, so the ledger has already
/// been charged by the time the challenge is returned, but `send_once` signs
/// every attempt afresh and each signature carries its own nonce, so the retry
/// arrives under a ledger key the node has never seen.) The 400, 429, 500 and
/// 503 cases did not apply the write; 429 and 503 are retryable, but doing it
/// automatically would only hammer a node that is already saying "not now", so
/// they surface to the user instead.
///
/// The set of codes is [`NodeDenial`], shared with the node so both halves
/// spell them the same way. The `match` below has no wildcard arm on purpose:
/// adding a denial to the node stops this crate compiling until someone writes
/// the caller's instructions for it. That is the whole point of the type, since
/// the previous arrangement (a list of literals in each crate) silently missed
/// `signature_nonce_too_short` for several commits and returned `Ok(400)` for it.
fn signature_rejection(resp: &reqwest::Response) -> Option<SignatureRejection> {
let code = resp.headers().get("x-gitlawb-error")?.to_str().ok()?;
// A code this build does not know: not our denial to describe. The header
// came from a node the user chose, which may be newer, older, or hostile,
// so the response goes back to the caller untouched.
let denial = NodeDenial::from_code(code)?;
// The status has to agree with the code, so a stray header on a 200 cannot
// turn a success into a refusal.
if resp.status().as_u16() != denial.status() {
return None;
}
let hint = match denial {
NodeDenial::NonceRequired => {
"this node requires a nonce in the request signature and the request did not carry \
one. The write did not happen; upgrade `gl` and run the command again"
}
NodeDenial::NonceTooShort => {
"this node rejected the nonce in the request signature as too short to be unique. \
The write did not happen; upgrade `gl` and run the command again"
}
NodeDenial::Replayed => {
"the node already admitted a request with this signature and will not take it twice. \
Do not resend: check whether the change took effect before running the command again"
}
NodeDenial::LedgerFull => {
"this identity has too many unexpired signatures on the node, so it is refusing more \
signed writes for now. The write did not happen; wait a moment and run the command \
again"
}
NodeDenial::IdentityMissing => {
"the node reached its signature ledger without a verified identity and refused the \
write. The write did not happen; this is a fault on the node, so report it to the \
operator"
}
NodeDenial::LedgerUnavailable => {
"the node's signature ledger is unavailable, so it is refusing signed writes. \
The write did not happen; retry later or ask the node operator to check the node"
}
NodeDenial::RateLimited => {
"the node is throttling requests from your network address, so it refused this one \
before running it. The write did not happen; this is not about your identity or \
your signature, so re-running immediately will not help — wait and try again"
}
};
Some(SignatureRejection { denial, hint })
}
/// Read a node reply, turning a non-2xx into an error instead of handing the
/// caller an error body that pretty-prints like a success. The message is read
/// under a cap and sanitized, same as a signature denial.
pub(crate) async fn json_or_denial<T: serde::de::DeserializeOwned>(
what: &str,
resp: reqwest::Response,
) -> Result<T> {
let status = resp.status();
if !status.is_success() {
let capped = read_body_capped(resp, DENIAL_BODY_CAP).await;
let msg = serde_json::from_str::<serde_json::Value>(&capped.text)
.ok()
.and_then(|v| {
v.get("message")
.or_else(|| v.get("error"))
.and_then(|m| m.as_str())
.map(str::to_string)
})
.unwrap_or(capped.text);
anyhow::bail!("{what} failed ({status}): {}", sanitize_node_msg(&msg));
}
resp.json::<T>()
.await
.with_context(|| format!("invalid JSON response from {what}"))
}
/// Run the (blocking) iCaptcha solve loop off the async runtime.
async fn obtain_proof(cfg: IcaptchaCfg) -> Result<String> {
tokio::task::spawn_blocking(move || icaptcha_client::obtain_proof(&cfg, None))
@@ -429,41 +269,21 @@ pub(crate) struct CappedBody {
pub(crate) read_failed: bool,
}
/// The byte twin of [`CappedBody`]: the same completion flags, but the payload is
/// left as the exact bytes that arrived. Byte-oriented protocol consumers (the git
/// smart-HTTP ref advertisement on the MCP `git_refs` path) must read through this
/// one: git ref names are byte strings that need not be UTF-8, so a lossy decode
/// followed by `as_bytes()` is not a display nicety but corruption — each invalid
/// byte becomes the 3-byte U+FFFD, changing both the ref name and the pkt-line
/// length accounting, and a valid repository stops validating.
pub(crate) struct CappedBytes {
/// The bytes read, exactly as they arrived.
pub(crate) bytes: Vec<u8>,
/// The cap cut the body short.
pub(crate) truncated: bool,
/// A chunk read FAILED part-way through, so the body is not merely short, it is
/// unfinished and the node may have had more to say.
pub(crate) read_failed: bool,
}
/// Read at most `cap` bytes of a response body, raw. Bounds the allocation from a
/// hostile or broken node returning a huge body — the display is capped
/// Read at most `cap` bytes of a response body. Bounds the allocation from a
/// hostile or broken node returning a huge error body — the display is capped
/// separately, but the read itself must not be unbounded (INV-6, read half).
///
/// `truncated` reports whether the cap cut the body short. A caller that CLASSIFIES
/// on the body needs it: a cut body fails JSON parse, and a parse failure is
/// indistinguishable from a node that sent no code at all, so without this flag an
/// oversized body silently picks a different arm. A protocol consumer needs it even
/// more: a cut pkt-line stream can still be a syntactically valid *prefix*, so
/// without the flag a bounded read is indistinguishable from a clean EOF and the
/// prefix parses as the complete answer.
/// oversized body silently picks a different arm.
///
/// `read_failed` reports the other way a body can end early. A mid-body read error
/// used to leave through the same exit as a clean end of stream, so a 500 whose body
/// died in transit surfaced as an empty message and the caller was told
/// `node returned 500: ` with nothing after the colon. That is a report of what the
/// node said, and the node never got to say it.
pub(crate) async fn read_body_capped_bytes(mut resp: reqwest::Response, cap: usize) -> CappedBytes {
pub(crate) async fn read_body_capped(mut resp: reqwest::Response, cap: usize) -> CappedBody {
let mut buf: Vec<u8> = Vec::new();
let mut truncated = false;
let mut read_failed = false;
@@ -489,27 +309,13 @@ pub(crate) async fn read_body_capped_bytes(mut resp: reqwest::Response, cap: usi
if buf.len() >= cap {
truncated = true;
}
CappedBytes {
bytes: buf,
CappedBody {
text: String::from_utf8_lossy(&buf).into_owned(),
truncated,
read_failed,
}
}
/// Text view of [`read_body_capped_bytes`], for callers that go on to display or
/// JSON-parse node-supplied *messages* (denial bodies). The decode is lossy on
/// purpose there — a denial is prose for a terminal, and a hostile byte becomes a
/// visible U+FFFD instead of raw output. Anything that parses a byte protocol out
/// of the body must take the bytes variant instead; see [`CappedBytes`].
pub(crate) async fn read_body_capped(resp: reqwest::Response, cap: usize) -> CappedBody {
let raw = read_body_capped_bytes(resp, cap).await;
CappedBody {
text: String::from_utf8_lossy(&raw.bytes).into_owned(),
truncated: raw.truncated,
read_failed: raw.read_failed,
}
}
/// Strip terminal-dangerous characters from (and cap the length of) a
/// node-supplied error string before surfacing it. The node a caller talks to
/// could be hostile and embed escape sequences in its error body; those must not
@@ -519,7 +325,7 @@ pub(crate) async fn read_body_capped(resp: reqwest::Response, cap: usize) -> Cap
pub(crate) fn sanitize_node_msg(s: &str) -> String {
s.chars()
.filter(|c| !c.is_control() && !gitlawb_core::sanitize::is_bidi_format(*c))
.take(NODE_MSG_CHARS)
.take(200)
.collect()
}
@@ -757,349 +563,6 @@ mod tests {
m.assert();
}
// ── send_signed signature-ledger rejections ─────────────────────────
/// Mock one node reply and send a signed POST through `send_signed`,
/// asserting the node was called exactly once (i.e. nothing retried).
async fn send_signed_once(
status: usize,
headers: &[(&str, &str)],
body: &str,
) -> (Result<reqwest::Response>, mockito::ServerGuard) {
let mut server = Server::new_async().await;
let mut m = server
.mock("POST", "/api/v1/repos")
.with_status(status)
.with_header("content-type", "application/json");
for (k, v) in headers {
m = m.with_header(*k, v);
}
let m = m.with_body(body).expect(1).create_async().await;
let client = NodeClient::new(server.url(), Some(test_keypair()));
let resp = client.send_signed("POST", "/api/v1/repos", b"{}").await;
m.assert_async().await;
(resp, server)
}
#[tokio::test]
async fn send_signed_errors_on_replayed_signature_and_never_retries() {
let (resp, _server) = send_signed_once(
409,
&[("x-gitlawb-error", "signature_replayed")],
r#"{"error":"signature_replayed","message":"this signature was already used"}"#,
)
.await;
let err = resp
.expect_err("a replayed signature must surface as an error, not Ok(409)")
.to_string();
assert!(err.contains("signature_replayed"), "got: {err}");
assert!(
err.contains("already"),
"message must tell the user the node already applied it, got: {err}"
);
assert!(
err.contains("this signature was already used"),
"must surface the node's message, got: {err}"
);
}
#[tokio::test]
async fn send_signed_errors_on_ledger_full_and_never_retries() {
let (resp, _server) = send_signed_once(
429,
&[("x-gitlawb-error", "signature_ledger_full")],
r#"{"error":"signature_ledger_full","message":"ledger at capacity"}"#,
)
.await;
let err = resp
.expect_err("a full ledger must surface as an error")
.to_string();
assert!(err.contains("signature_ledger_full"), "got: {err}");
assert!(
!err.contains("already"),
"must not claim the write was applied, got: {err}"
);
}
#[tokio::test]
async fn send_signed_errors_on_ledger_unavailable_and_never_retries() {
let (resp, _server) = send_signed_once(
503,
&[("x-gitlawb-error", "signature_ledger_unavailable")],
r#"{"error":"signature_ledger_unavailable","message":"ledger down"}"#,
)
.await;
let err = resp
.expect_err("an unavailable ledger must surface as an error")
.to_string();
assert!(err.contains("signature_ledger_unavailable"), "got: {err}");
}
#[tokio::test]
async fn send_signed_errors_on_nonce_required_and_never_retries() {
let (resp, _server) = send_signed_once(
400,
&[("x-gitlawb-error", "signature_nonce_required")],
r#"{"error":"signature_nonce_required","message":"this node requires a nonce"}"#,
)
.await;
let err = resp
.expect_err("a nonce-less signature must surface as an error, not Ok(400)")
.to_string();
assert!(err.contains("signature_nonce_required"), "got: {err}");
assert!(
!err.contains("already"),
"must not claim the write was applied, got: {err}"
);
}
#[tokio::test]
async fn send_signed_errors_on_identity_missing_and_never_retries() {
let (resp, _server) = send_signed_once(
500,
&[("x-gitlawb-error", "signature_identity_missing")],
r#"{"error":"signature_identity_missing","message":"no verified identity"}"#,
)
.await;
let err = resp
.expect_err("a missing signature identity must surface as an error, not Ok(500)")
.to_string();
assert!(err.contains("signature_identity_missing"), "got: {err}");
assert!(
!err.contains("already"),
"must not claim the write was applied, got: {err}"
);
}
#[tokio::test]
async fn send_signed_errors_on_nonce_too_short_and_never_retries() {
let (resp, _server) = send_signed_once(
400,
&[("x-gitlawb-error", "signature_nonce_too_short")],
r#"{"error":"signature_nonce_too_short","message":"the nonce must be at least 16 characters"}"#,
)
.await;
let err = resp
.expect_err("a too-short nonce must surface as an error, not Ok(400)")
.to_string();
assert!(err.contains("signature_nonce_too_short"), "got: {err}");
assert!(
!err.contains("already"),
"must not claim the write was applied, got: {err}"
);
assert!(
err.contains("the nonce must be at least 16 characters"),
"must surface the node's message, got: {err}"
);
}
/// Two unrelated conditions now share 429: the node's per-client flood
/// brake and the spent-signature ledger's per-identity cap. They must reach
/// the user as different errors, and neither may be retried.
#[tokio::test]
async fn send_signed_tells_a_rate_limit_brake_apart_from_a_full_ledger() {
let (brake, _s1) = send_signed_once(
429,
&[("x-gitlawb-error", "rate_limited")],
r#"{"error":"rate_limited","message":"rate limit exceeded — try again later"}"#,
)
.await;
let brake = brake
.expect_err("a rate-limit brake must surface as an error, not Ok(429)")
.to_string();
let (ledger, _s2) = send_signed_once(
429,
&[("x-gitlawb-error", "signature_ledger_full")],
r#"{"error":"signature_ledger_full","message":"ledger at capacity"}"#,
)
.await;
let ledger = ledger
.expect_err("a full ledger must surface as an error")
.to_string();
assert!(brake.contains("rate_limited"), "got: {brake}");
assert!(
!brake.contains("signature_ledger_full"),
"a brake must not be described as a ledger denial: {brake}"
);
assert!(
brake.contains("network address"),
"the brake hint must say it is keyed on the client address: {brake}"
);
assert!(ledger.contains("signature_ledger_full"), "got: {ledger}");
assert!(
!ledger.contains("rate_limited"),
"a ledger denial must not be described as a brake: {ledger}"
);
assert!(
ledger.contains("unexpired signatures"),
"the ledger hint must say it is keyed on the identity: {ledger}"
);
assert_ne!(brake, ledger);
}
/// Every signature denial the node can return, as (status, code). Derived
/// from the shared enum rather than retyped, so it cannot drift from what
/// the node emits.
fn all_denials() -> Vec<(usize, &'static str)> {
NodeDenial::ALL
.iter()
.map(|d| (d.status() as usize, d.as_str()))
.collect()
}
#[tokio::test]
async fn send_signed_errors_on_every_denial_carrying_the_header() {
for (status, code) in all_denials() {
let body = format!(r#"{{"error":"{code}","message":"node says no"}}"#);
let (resp, _server) =
send_signed_once(status, &[("x-gitlawb-error", code)], &body).await;
let err = match resp {
Ok(r) => panic!("{code} returned Ok({}) instead of an error", r.status()),
Err(e) => e.to_string(),
};
assert!(err.contains(code), "wrong code surfaced for {code}: {err}");
assert!(
err.contains("node says no"),
"node message dropped for {code}: {err}"
);
}
}
#[tokio::test]
async fn send_signed_does_not_see_a_denial_carried_only_in_the_body() {
// Documents a known limit rather than a wanted behaviour: detection is
// header-keyed, so a proxy stripping `x-gitlawb-error` leaves the denial
// to the caller (see `signature_rejection`). The callers that must not
// read one as a success handle it themselves: `gl init` compares
// `error` from the body, `gl task` checks the status before parsing.
for (status, code) in all_denials() {
let body = format!(r#"{{"error":"{code}","message":"node says no"}}"#);
let (resp, _server) = send_signed_once(status, &[], &body).await;
let resp = resp.unwrap_or_else(|e| panic!("{code} without the header: {e}"));
assert_eq!(resp.status().as_u16(), status as u16);
// The body is untouched, which is what lets the caller decide.
let payload: serde_json::Value = resp.json().await.unwrap();
assert_eq!(payload["error"], code);
}
}
#[tokio::test]
async fn send_signed_sanitizes_and_caps_a_hostile_denial_message() {
// The message that exists to say "your write was REFUSED" must not be
// able to clear the screen and print a fake success line.
let hostile = format!(
"\u{1b}[2J\u{1b}[31mSUCCESS: task created\u{202e}{}",
"A".repeat(5000)
);
let body = serde_json::json!({
"error": "signature_replayed",
"message": hostile,
})
.to_string();
let (resp, _server) =
send_signed_once(409, &[("x-gitlawb-error", "signature_replayed")], &body).await;
let err = resp.expect_err("a replay must error").to_string();
assert!(
!err.contains('\u{1b}'),
"ESC leaked to the terminal: {err:?}"
);
assert!(
!err.contains('\u{202e}'),
"RLO bidi override leaked: {err:?}"
);
assert!(err.contains("signature_replayed"), "got: {err}");
assert!(
err.chars().count() < 600,
"denial message not bounded: {} chars",
err.chars().count()
);
}
#[tokio::test]
async fn send_signed_returns_repo_exists_409_unchanged() {
// The pre-existing 409. Callers (`gl init`, `gl repo create`) inspect the
// status themselves, so it must keep returning Ok(resp).
let (resp, _server) = send_signed_once(
409,
&[("x-node-marker", "kept")],
r#"{"error":"repo_exists","message":"already exists"}"#,
)
.await;
let resp = resp.expect("a non-replay 409 must still return Ok");
assert_eq!(resp.status(), 409);
// Status, headers and body must all survive the denial check: `gl init`
// reads the code out of the body to decide "already exists, continue".
assert_eq!(resp.headers().get("x-node-marker").unwrap(), "kept");
let payload: serde_json::Value = resp.json().await.expect("body must survive intact");
assert_eq!(payload["error"], "repo_exists");
assert_eq!(payload["message"], "already exists");
}
#[tokio::test]
async fn send_signed_returns_409_with_other_error_header_unchanged() {
// Only the three signature_* codes are converted; any other
// x-gitlawb-error on a 409 is left to the caller.
let (resp, _server) = send_signed_once(
409,
&[("x-gitlawb-error", "repo_exists")],
r#"{"error":"repo_exists"}"#,
)
.await;
let resp = resp.expect("an unrelated 409 error code must still return Ok");
assert_eq!(resp.status(), 409);
}
#[tokio::test]
async fn send_signed_returns_401_invalid_signature_unchanged() {
let (resp, _server) = send_signed_once(
401,
&[("x-gitlawb-error", "invalid_signature")],
r#"{"error":"invalid_signature"}"#,
)
.await;
let resp = resp.expect("a 401 must still return Ok, distinct from a replay error");
assert_eq!(resp.status(), 401);
}
#[tokio::test]
async fn send_signed_returns_401_unsigned_request_unchanged() {
// An unsigned write is rejected with human_detected; the client prints a
// hint and returns the response, and must not be mistaken for a replay.
let mut server = Server::new_async().await;
let m = server
.mock("POST", "/api/v1/repos")
.with_status(401)
.with_header("content-type", "application/json")
.with_header("x-gitlawb-error", "human_detected")
.with_body(r#"{"error":"human_detected"}"#)
.expect(1)
.create_async()
.await;
let client = NodeClient::new(server.url(), None);
let resp = client
.send_signed("POST", "/api/v1/repos", b"{}")
.await
.expect("an unsigned-request rejection must still return Ok");
assert_eq!(resp.status(), 401);
m.assert_async().await;
}
#[tokio::test]
async fn send_signed_does_not_convert_a_409_on_another_status() {
// The header alone must not trigger the conversion: the status has to
// match too, so a 200 carrying a stray header stays a success.
let (resp, _server) = send_signed_once(
200,
&[("x-gitlawb-error", "signature_replayed")],
r#"{"ok":true}"#,
)
.await;
let resp = resp.expect("a 200 must stay Ok regardless of headers");
assert_eq!(resp.status(), 200);
}
// ── send_signed iCaptcha retry (full integration) ────────────────────
/// Set GITLAWB_ICAPTCHA_URL and GITLAWB_ICAPTCHA_INSECURE so the iCaptcha
@@ -1220,77 +683,6 @@ mod tests {
ic.answer.assert();
}
#[tokio::test]
async fn send_signed_signs_the_icaptcha_retry_afresh() {
// What makes the 403 retry safe is that it is a NEW signature, not a
// resend: `verify_request` runs inside the handler, so the ledger was
// already charged for the first attempt. Each `send_once` signs again
// and `sign_request` draws a fresh nonce, so the retry lands on a ledger
// key the node has not seen. Pin that: the two attempts must not carry
// the same `Signature-Input`.
let mut node = Server::new_async().await;
let mut icaptcha = Server::new_async().await;
let ic = MockIcaptcha::new(&mut icaptcha, 1).await;
let seen: std::sync::Arc<Mutex<Vec<String>>> = Default::default();
let record = {
let seen = seen.clone();
move |req: &mockito::Request| {
let v = req
.header("signature-input")
.first()
.map(|h| String::from_utf8_lossy(h.as_bytes()).into_owned())
.unwrap_or_default();
seen.lock().unwrap().push(v);
true
}
};
let n1 = node
.mock("POST", "/api/register")
.match_header("x-icaptcha-proof", mockito::Matcher::Missing)
.match_request(record.clone())
.with_status(403)
.with_header("content-type", "application/json")
.with_header("x-icaptcha-url", &ic.url)
.with_header("x-icaptcha-level", "3")
.with_body(r#"{"error":"icaptcha_proof_required"}"#)
.expect(1)
.create_async()
.await;
let n2 = node
.mock("POST", "/api/register")
.match_header("x-icaptcha-proof", "mock.proof")
.match_request(record)
.with_status(201)
.with_header("content-type", "application/json")
.with_body(r#"{"status":"created"}"#)
.expect(1)
.create_async()
.await;
let client = NodeClient::new(node.url(), Some(test_keypair()));
let resp = client
.send_signed("POST", "/api/register", b"{}")
.await
.unwrap();
assert_eq!(resp.status(), 201);
n1.assert();
n2.assert();
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2, "expected two attempts, got {seen:?}");
assert!(
!seen[0].is_empty() && !seen[1].is_empty(),
"both attempts must be signed: {seen:?}"
);
assert_ne!(
seen[0], seen[1],
"the retry reused the first signature, so it would land on the same \
ledger key the node already spent"
);
}
#[tokio::test]
async fn send_signed_returns_403_after_icaptcha_retries_exhausted() {
let mut node = Server::new_async().await;
+55 -173
View File
@@ -9,7 +9,7 @@ use clap::Args;
use serde_json::{json, Value};
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -33,14 +33,6 @@ pub struct InitArgs {
pub async fn run(args: InitArgs) -> Result<()> {
let cwd = std::env::current_dir().context("cannot determine current directory")?;
run_in(&cwd, args).await
}
/// The body of [`run`], with the working directory passed in. Split out so the
/// whole flow (including what it prints on a node denial) is testable without
/// mutating the process-wide current directory.
async fn run_in(cwd: &std::path::Path, args: InitArgs) -> Result<()> {
let cwd = cwd.to_path_buf();
// 1. Ensure git repo exists
let git_dir = cwd.join(".git");
@@ -95,12 +87,16 @@ async fn run_in(cwd: &std::path::Path, args: InitArgs) -> Result<()> {
.post("/api/register", &body)
.await
.context("failed to connect to node")?;
// No tolerated failure here: `register_agent` upserts, so re-registering a
// known DID is a 201, not a conflict (node `api/register.rs`, and the ON
// CONFLICT clause in `db::register_agent`). An older check let any message
// containing "already" through, which included the replay denial "this
// signature has already been used".
let payload: Value = json_or_denial("registration", resp).await?;
let status = resp.status();
let payload: Value = resp.json().await.context("invalid JSON from register")?;
if !status.is_success() {
let msg = payload["message"].as_str().unwrap_or("unknown error");
// "already registered" is fine
if !msg.contains("already") {
anyhow::bail!("registration failed ({status}): {msg}");
}
}
// Save UCAN if returned
if let Some(ucan) = payload.get("ucan").and_then(|v| v.as_str()) {
@@ -143,22 +139,15 @@ async fn run_in(cwd: &std::path::Path, args: InitArgs) -> Result<()> {
.await
.context("failed to create repo")?;
let repo_status = resp.status();
let repo_result: Value = resp.json().await.context("invalid JSON from create repo")?;
if !repo_status.is_success() {
let capped = crate::http::read_body_capped(resp, crate::http::DENIAL_BODY_CAP).await;
let repo_result: Value = serde_json::from_str(&capped.text).unwrap_or(Value::Null);
if !repo_already_exists(&repo_result) {
let msg = repo_result["message"]
.as_str()
.or_else(|| repo_result["error"].as_str())
.unwrap_or("unknown error");
anyhow::bail!(
"create repo failed ({repo_status}): {}",
crate::http::sanitize_node_msg(msg)
);
let msg = repo_result["message"].as_str().unwrap_or("unknown error");
if !msg.contains("exists") && !msg.contains("already") {
anyhow::bail!("create repo failed ({repo_status}): {msg}");
}
println!(" Repository already exists — continuing.");
} else {
let _repo_result: Value = resp.json().await.context("invalid JSON from create repo")?;
println!(" Repository created.");
}
@@ -231,14 +220,6 @@ async fn run_in(cwd: &std::path::Path, args: InitArgs) -> Result<()> {
Ok(())
}
/// Is this failed create-repo reply the benign "you already own that repo" case?
/// `AppError::RepoExists` is the only thing the node renders as `repo_exists`
/// (node `error.rs`), and it is the only non-success reply `gl init` may treat
/// as "keep going".
fn repo_already_exists(payload: &Value) -> bool {
payload["error"].as_str() == Some("repo_exists")
}
fn generate_identity(dir: Option<&std::path::Path>) -> Result<gitlawb_core::identity::Keypair> {
let base = if let Some(d) = dir {
d.to_path_buf()
@@ -351,155 +332,56 @@ mod tests {
assert_eq!(resp["name"], "test-repo");
}
/// A work dir with a git repo in it, plus an identity dir, wired to `node`.
fn init_args(node: String, id_dir: &TempDir) -> (InitArgs, TempDir) {
let work = TempDir::new().unwrap();
std::process::Command::new("git")
.args(["init", "-b", "main"])
.current_dir(work.path())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.unwrap();
let args = InitArgs {
name: Some("test-repo".to_string()),
node,
dir: Some(id_dir.path().to_path_buf()),
description: None,
};
(args, work)
}
fn has_gitlawb_remote(work: &TempDir) -> bool {
std::process::Command::new("git")
.args(["remote", "get-url", "gitlawb"])
.current_dir(work.path())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
async fn mock_register(server: &mut mockito::ServerGuard) -> mockito::Mock {
server
.mock("POST", "/api/register")
.with_status(201)
.with_header("content-type", "application/json")
.with_body(r#"{"status":"accepted","message":"welcome","ucan":"test.token"}"#)
.create_async()
.await
}
#[tokio::test]
async fn init_fails_on_replay_409_without_the_error_header() {
// The node refused the create (the signature was already spent) and a
// proxy stripped `x-gitlawb-error`, so the client layer cannot see the
// denial and hands the 409 straight back. `gl init` used to read
// "already" out of the prose, print "Repository already exists —
// continuing", add the remote, print "Ready! Push with", and exit 0 for
// a repo that does not exist. Comparing the structured error code is
// what makes it fail instead, before any of that.
let id_dir = TempDir::new().unwrap();
write_identity(&id_dir);
let mut server = mockito::Server::new_async().await;
let _reg = mock_register(&mut server).await;
let _repo = server
.mock("POST", "/api/v1/repos")
.with_status(409)
.with_header("content-type", "application/json")
.with_body(
r#"{"error":"signature_replayed","message":"this signature has already been used - sign a fresh request"}"#,
)
.create_async()
.await;
async fn test_init_handles_already_registered() {
let dir = TempDir::new().unwrap();
let kp = write_identity(&dir);
let (args, work) = init_args(server.url(), &id_dir);
let err = run_in(work.path(), args)
.await
.expect_err("a replayed signature must fail `gl init`, not report success");
let err = format!("{err:#}");
assert!(err.contains("create repo failed"), "got: {err}");
assert!(err.contains("409"), "status not surfaced: {err}");
// The remote is added several lines after the point where the old code
// decided "already exists, continuing", so its absence proves we bailed
// before printing "Ready! Push with".
assert!(
!has_gitlawb_remote(&work),
"init continued past the denial and configured the remote"
);
}
#[tokio::test]
async fn init_still_continues_on_a_real_repo_exists_409() {
// The regression guard for the fix above: a genuine repo_exists conflict
// must behave exactly as before (continue, add the remote, succeed).
let id_dir = TempDir::new().unwrap();
write_identity(&id_dir);
let mut server = mockito::Server::new_async().await;
let _reg = mock_register(&mut server).await;
let _repo = server
.mock("POST", "/api/v1/repos")
.with_status(409)
.with_header("content-type", "application/json")
.with_body(
r#"{"error":"repo_exists","message":"repository 'test-repo' already exists"}"#,
)
.create_async()
.await;
let (args, work) = init_args(server.url(), &id_dir);
run_in(work.path(), args)
.await
.expect("an existing repo must still be a successful `gl init`");
assert!(
has_gitlawb_remote(&work),
"init must still configure the remote for an existing repo"
);
}
#[tokio::test]
async fn init_fails_when_registration_is_refused() {
// Registration is an upsert on the node, so any non-success is a real
// failure. The old `contains("already")` tolerance also swallowed the
// replay denial here.
let id_dir = TempDir::new().unwrap();
write_identity(&id_dir);
let mut server = mockito::Server::new_async().await;
let _reg = server
.mock("POST", "/api/register")
.with_status(409)
.with_header("content-type", "application/json")
.with_body(
r#"{"error":"signature_replayed","message":"this signature has already been used"}"#,
)
.with_body(r#"{"message":"already registered"}"#)
.create_async()
.await;
let (args, work) = init_args(server.url(), &id_dir);
let err = run_in(work.path(), args)
.await
.expect_err("a refused registration must fail `gl init`");
let err = format!("{err:#}");
assert!(err.contains("registration failed"), "got: {err}");
assert!(err.contains("409"), "status not surfaced: {err}");
assert!(!has_gitlawb_remote(&work));
let client = NodeClient::new(server.url(), Some(kp.clone()));
let body = serde_json::to_vec(&json!({
"did": kp.did().to_string(),
"capabilities": ["git:push"],
}))
.unwrap();
let resp = client.post("/api/register", &body).await.unwrap();
let status = resp.status();
let payload: Value = resp.json().await.unwrap();
let msg = payload["message"].as_str().unwrap_or("");
// Should not bail because message contains "already"
assert!(!status.is_success());
assert!(msg.contains("already"));
}
#[test]
fn repo_already_exists_matches_the_code_not_the_prose() {
assert!(repo_already_exists(
&json!({"error":"repo_exists","message":"repository 'x' already exists"})
));
// The replay denial's message contains "already" and "exists" is one
// word away; only the code separates them.
assert!(!repo_already_exists(&json!({
"error": "signature_replayed",
"message": "this signature has already been used - sign a fresh request"
})));
assert!(!repo_already_exists(
&json!({"message":"repository already exists"})
));
assert!(!repo_already_exists(&json!({})));
#[tokio::test]
async fn test_init_handles_repo_exists() {
let dir = TempDir::new().unwrap();
let kp = write_identity(&dir);
let mut server = mockito::Server::new_async().await;
let _repo = server
.mock("POST", "/api/v1/repos")
.with_status(409)
.with_header("content-type", "application/json")
.with_body(r#"{"message":"repository already exists"}"#)
.create_async()
.await;
let client = NodeClient::new(server.url(), Some(kp.clone()));
let body = serde_json::to_vec(&json!({"name": "existing", "is_public": true})).unwrap();
let resp = client.post("/api/v1/repos", &body).await.unwrap();
let status = resp.status();
let result: Value = resp.json().await.unwrap();
let msg = result["message"].as_str().unwrap_or("");
assert!(!status.is_success());
assert!(msg.contains("exists") || msg.contains("already"));
}
}
+29 -6
View File
@@ -7,7 +7,7 @@ use serde_json::{json, Value};
use std::path::PathBuf;
use uuid::Uuid;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
fn signed_client(node: &str, dir: Option<&std::path::Path>) -> NodeClient {
@@ -198,7 +198,13 @@ async fn cmd_create(
.post(&path, &request_body)
.await
.context("failed to connect to node")?;
let result: Value = json_or_denial("create issue", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("create issue failed ({status}): {msg}");
}
let id = result["id"].as_str().unwrap_or("?");
println!("✓ Created issue #{id}");
@@ -258,7 +264,13 @@ async fn cmd_show(repo: String, id: String, node: String, dir: Option<PathBuf>)
.get_authed(&path)
.await
.context("failed to connect to node")?;
let issue: Value = json_or_denial("show", resp).await?;
let status = resp.status();
let issue: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = issue["message"].as_str().unwrap_or("issue not found");
anyhow::bail!("show failed ({status}): {msg}");
}
let title = issue["title"].as_str().unwrap_or("(no title)");
let status = issue["status"].as_str().unwrap_or("?");
@@ -289,8 +301,13 @@ async fn cmd_close(repo: String, id: String, node: String, dir: Option<PathBuf>)
.post(&format!("{path}/close"), &body)
.await
.context("failed to connect to node")?;
// Parsed only to reject a denial; the success body carries nothing to print.
json_or_denial::<Value>("close issue", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("close issue failed ({status}): {msg}");
}
println!("✗ Closed issue {id}");
Ok(())
@@ -315,7 +332,13 @@ async fn cmd_issue_comment(
)
.await
.context("failed to connect to node")?;
json_or_denial::<Value>("comment", resp).await?;
let code = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !code.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("comment failed ({code}): {msg}");
}
println!("· Comment posted on issue {id}");
Ok(())
+153 -774
View File
File diff suppressed because it is too large. Load diff
+29 -5
View File
@@ -5,7 +5,7 @@ use clap::{Args, Subcommand};
use serde_json::Value;
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -233,7 +233,13 @@ async fn cmd_create(
.post(&format!("/api/v1/repos/{owner}/{repo}/pulls"), &payload)
.await
.context("failed to connect to node")?;
let pr: Value = json_or_denial("create PR", resp).await?;
let status = resp.status();
let pr: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = pr["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("create PR failed ({status}): {msg}");
}
let number = pr["number"].as_i64().unwrap_or(0);
println!("✓ Opened PR #{number}: {title}");
@@ -409,7 +415,13 @@ async fn cmd_merge(repo: String, number: u64, node: String, dir: Option<PathBuf>
)
.await
.context("failed to connect to node")?;
let result: Value = json_or_denial("merge", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("merge failed ({status}): {msg}");
}
let sha = result["merge_sha"].as_str().unwrap_or("?");
println!("✓ Merged PR #{number}");
@@ -441,7 +453,13 @@ async fn cmd_review(
)
.await
.context("failed to connect to node")?;
json_or_denial::<Value>("review", resp).await?;
let code = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !code.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("review failed ({code}): {msg}");
}
let icon = match status.as_str() {
"approved" => "✓",
@@ -471,7 +489,13 @@ async fn cmd_comment(
)
.await
.context("failed to connect to node")?;
json_or_denial::<Value>("comment", resp).await?;
let code = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !code.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("comment failed ({code}): {msg}");
}
println!("· Comment posted on PR #{number}");
Ok(())
+21 -67
View File
@@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize};
use serde_json::json;
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -248,10 +248,16 @@ async fn cmd_set(
.await
.context("failed to update profile")?;
// Status before body. A denial need not be JSON at all — the per-client
// rate-limit brake used to answer text/plain — and parsing first turned a
// 429 into "invalid JSON response", which names the wrong problem.
let resp_body: serde_json::Value = json_or_denial("profile update", resp).await?;
let status = resp.status();
let resp_body: serde_json::Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = resp_body
.get("message")
.and_then(|v| v.as_str())
.unwrap_or("unknown error");
anyhow::bail!("profile update failed ({status}): {msg}");
}
println!();
println!("✓ Profile updated");
@@ -392,8 +398,16 @@ async fn cmd_import(path: PathBuf, pin: bool, node: String, dir: Option<PathBuf>
.await
.context("failed to import profile")?;
// Parsed only to reject a denial; the success body carries nothing to print.
json_or_denial::<serde_json::Value>("profile import", resp).await?;
let status = resp.status();
let resp_body: serde_json::Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = resp_body
.get("message")
.and_then(|v| v.as_str())
.unwrap_or("unknown error");
anyhow::bail!("profile import failed ({status}): {msg}");
}
println!("✓ Profile imported successfully");
Ok(())
@@ -442,66 +456,6 @@ fn did_short(did: &str) -> &str {
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
fn write_identity(dir: &TempDir) {
let kp = gitlawb_core::identity::Keypair::generate();
std::fs::write(
dir.path().join("identity.pem"),
kp.to_pem().unwrap().as_bytes(),
)
.unwrap();
}
async fn set_against(status: usize, headers: &[(&str, &str)], body: &str) -> String {
let dir = TempDir::new().unwrap();
write_identity(&dir);
let mut server = mockito::Server::new_async().await;
let mut m = server.mock("PUT", "/api/v1/profile").with_status(status);
for (k, v) in headers {
m = m.with_header(*k, v);
}
let _m = m.with_body(body).create_async().await;
let err = cmd_set(
Some("Axiom".into()),
None,
None,
None,
None,
None,
None,
None,
false,
server.url(),
Some(dir.path().to_path_buf()),
)
.await
.expect_err("a 429 must not be reported as success");
format!("{err:#}")
}
/// A per-IP brake 429 is newly reachable on this route. Parsing the body
/// before checking the status turned it into "invalid JSON response",
/// which tells the user nothing about the real condition.
#[tokio::test]
async fn profile_set_surfaces_a_rate_limit_brake_not_invalid_json() {
// Header present: the client recognises the code up front.
let err = set_against(
429,
&[("x-gitlawb-error", "rate_limited")],
r#"{"error":"rate_limited","message":"rate limit exceeded — try again later"}"#,
)
.await;
assert!(!err.contains("invalid JSON"), "got: {err}");
assert!(err.contains("rate_limited"), "got: {err}");
// Header stripped by a proxy, body not JSON: the status still has to
// decide, so this must read as a 429, not as a parse failure.
let err = set_against(429, &[], "rate limit exceeded — try again later").await;
assert!(!err.contains("invalid JSON"), "got: {err}");
assert!(err.contains("429"), "got: {err}");
assert!(err.contains("rate limit exceeded"), "got: {err}");
}
#[test]
fn test_did_short_extracts_suffix() {
+11 -2
View File
@@ -8,7 +8,7 @@ use clap::Args;
use serde_json::{json, Value};
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -55,7 +55,16 @@ pub async fn run(args: RegisterArgs) -> Result<()> {
.await
.context("failed to connect to node")?;
let payload: Value = json_or_denial("registration", resp).await?;
let status = resp.status();
let payload: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = payload
.get("message")
.and_then(|v| v.as_str())
.unwrap_or("unknown error");
anyhow::bail!("registration failed ({status}): {msg}");
}
// Save bootstrap UCAN
let ucan = payload.get("ucan").and_then(|v| v.as_str()).unwrap_or("");
+29 -50
View File
@@ -5,7 +5,7 @@ use clap::{Args, Subcommand};
use serde_json::{json, Value};
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -251,7 +251,13 @@ async fn cmd_create(
.post("/api/v1/repos", &body)
.await
.context("failed to connect to node")?;
let payload: Value = json_or_denial("create", resp).await?;
let status = resp.status();
let payload: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = payload["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("create failed ({status}): {msg}");
}
let clone_url = payload["clone_url"].as_str().unwrap_or("");
let gitlawb_url = format!("gitlawb://{owner_did}/{name}");
@@ -573,9 +579,13 @@ async fn cmd_fork(
.post(&format!("/api/v1/repos/{owner}/{repo_name}/fork"), &body)
.await
.context("failed to connect to node")?;
// Status before body: a rate-limit brake or any other non-JSON denial must
// read as the denial, not as "invalid JSON response".
let result: Value = json_or_denial("fork", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON response")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("fork failed ({status}): {msg}");
}
let fork_name = result["name"].as_str().unwrap_or(&repo_name);
let owner_did = result["owner_did"].as_str().unwrap_or("?");
@@ -613,7 +623,13 @@ async fn cmd_label_add(
.post(&format!("/api/v1/repos/{owner}/{name}/labels"), &body)
.await
.context("failed to connect to node")?;
let result: Value = json_or_denial("add label", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("add label failed ({status}): {msg}");
}
let added = result["added"].as_bool().unwrap_or(true);
if added {
@@ -638,8 +654,13 @@ async fn cmd_label_remove(
.delete(&format!("/api/v1/repos/{owner}/{name}/labels/{label}"), &[])
.await
.context("failed to connect to node")?;
// Parsed only to reject a denial; the success body carries nothing to print.
json_or_denial::<Value>("remove label", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("remove label failed ({status}): {msg}");
}
println!("- Label removed: {label} from {owner}/{name}");
Ok(())
@@ -799,48 +820,6 @@ mod tests {
assert!(!owner.contains(':'));
}
async fn fork_against(status: usize, headers: &[(&str, &str)], body: &str) -> String {
let dir = TempDir::new().unwrap();
write_identity(&dir);
let mut server = mockito::Server::new_async().await;
let mut m = server
.mock("POST", "/api/v1/repos/alice/myrepo/fork")
.with_status(status);
for (k, v) in headers {
m = m.with_header(*k, v);
}
let _m = m.with_body(body).create_async().await;
let err = cmd_fork(
"alice/myrepo".to_string(),
None,
server.url(),
Some(dir.path().to_path_buf()),
)
.await
.expect_err("a 429 must not be reported as success");
format!("{err:#}")
}
/// Same shape as `gl profile set`: the fork route gained a per-IP brake, so
/// a 429 with a non-JSON body has to read as a rate limit rather than as a
/// parse failure.
#[tokio::test]
async fn fork_surfaces_a_rate_limit_brake_not_invalid_json() {
let err = fork_against(
429,
&[("x-gitlawb-error", "rate_limited")],
r#"{"error":"rate_limited","message":"rate limit exceeded — try again later"}"#,
)
.await;
assert!(!err.contains("invalid JSON"), "got: {err}");
assert!(err.contains("rate_limited"), "got: {err}");
let err = fork_against(429, &[], "rate limit exceeded — try again later").await;
assert!(!err.contains("invalid JSON"), "got: {err}");
assert!(err.contains("429"), "got: {err}");
assert!(err.contains("rate limit exceeded"), "got: {err}");
}
#[tokio::test]
async fn test_cmd_create_success() {
let dir = TempDir::new().unwrap();
+24 -144
View File
@@ -5,7 +5,7 @@ use clap::{Args, Subcommand};
use serde_json::{json, Value};
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -166,14 +166,13 @@ async fn cmd_create(
"delegator_did": delegator_did,
}))?;
let resp = client
let resp: Value = client
.post("/api/v1/tasks", &body)
.await
.context("failed to create task")?;
// Check the status BEFORE parsing. Without this a denial (a 400
// `signature_nonce_required`, a 500, a 404) pretty-prints the node's error
// JSON and exits 0, so a script keying on `$?` treats the task as created.
let resp: Value = json_or_denial("create task", resp).await?;
.context("failed to create task")?
.json()
.await
.context("invalid JSON response")?;
print_json(&resp);
Ok(())
}
@@ -222,11 +221,13 @@ async fn cmd_claim(id: String, node: String, dir: Option<PathBuf>) -> Result<()>
let client = NodeClient::new(&node, Some(keypair));
let body = serde_json::to_vec(&json!({ "assignee_did": assignee_did }))?;
let resp = client
let resp: Value = client
.post(&format!("/api/v1/tasks/{}/claim", id), &body)
.await
.context("failed to claim task")?;
let resp: Value = json_or_denial("claim task", resp).await?;
.context("failed to claim task")?
.json()
.await
.context("invalid JSON response")?;
print_json(&resp);
Ok(())
}
@@ -242,11 +243,13 @@ async fn cmd_complete(
let client = NodeClient::new(&node, Some(keypair));
let body = serde_json::to_vec(&json!({ "result": result, "by_did": by_did }))?;
let resp = client
let resp: Value = client
.post(&format!("/api/v1/tasks/{}/complete", id), &body)
.await
.context("failed to complete task")?;
let resp: Value = json_or_denial("complete task", resp).await?;
.context("failed to complete task")?
.json()
.await
.context("invalid JSON response")?;
print_json(&resp);
Ok(())
}
@@ -262,11 +265,13 @@ async fn cmd_fail(
let client = NodeClient::new(&node, Some(keypair));
let body = serde_json::to_vec(&json!({ "reason": reason, "by_did": by_did }))?;
let resp = client
let resp: Value = client
.post(&format!("/api/v1/tasks/{}/fail", id), &body)
.await
.context("failed to fail task")?;
let resp: Value = json_or_denial("fail task", resp).await?;
.context("failed to fail task")?
.json()
.await
.context("invalid JSON response")?;
print_json(&resp);
Ok(())
}
@@ -334,129 +339,6 @@ mod tests {
assert!(err.to_string().contains("no identity found"));
}
/// Write an identity into a fresh dir so the signed task commands run.
fn identity_dir() -> tempfile::TempDir {
let dir = tempfile::TempDir::new().unwrap();
let kp = gitlawb_core::identity::Keypair::generate();
std::fs::write(
dir.path().join("identity.pem"),
kp.to_pem().unwrap().as_bytes(),
)
.unwrap();
dir
}
/// Every denial a task write can come back with, as (status, body).
/// The last entry is a plain node failure with no signature code: the task
/// commands must fail on that too, not just on the ledger denials.
fn denials() -> Vec<(usize, String)> {
gitlawb_core::node_denial::NodeDenial::ALL
.iter()
.map(|d| {
(
d.status() as usize,
format!(
r#"{{"error":"{}","message":"node refused the write"}}"#,
d.as_str()
),
)
})
.chain(std::iter::once((
500,
r#"{"error":"internal_error","message":"boom"}"#.to_string(),
)))
.collect()
}
#[tokio::test]
async fn task_writes_fail_on_every_denial() {
for (status, body) in denials() {
let dir = identity_dir();
let mut server = mockito::Server::new_async().await;
let _m = server
.mock("POST", mockito::Matcher::Any)
.with_status(status)
.with_header("content-type", "application/json")
.with_body(&body)
.expect_at_least(1)
.create_async()
.await;
let path = Some(dir.path().to_path_buf());
let e = cmd_create(
"deploy".into(),
"agent:task".into(),
None,
None,
None,
None,
None,
server.url(),
path.clone(),
)
.await;
assert!(
e.is_err(),
"task create returned Ok for {status} {body}: a script keying on $? \
would treat the task as created"
);
for (name, res) in [
(
"claim",
cmd_claim("t1".into(), server.url(), path.clone()).await,
),
(
"complete",
cmd_complete("t1".into(), None, server.url(), path.clone()).await,
),
(
"fail",
cmd_fail("t1".into(), None, server.url(), path.clone()).await,
),
] {
assert!(res.is_err(), "task {name} returned Ok for {status} {body}");
}
}
}
#[tokio::test]
async fn task_create_denial_message_is_sanitized_and_bounded() {
let dir = identity_dir();
let mut server = mockito::Server::new_async().await;
let hostile = format!("\u{1b}[2Jfake success\u{202e}{}", "A".repeat(5000));
let body = serde_json::json!({"error": "internal_error", "message": hostile}).to_string();
let _m = server
.mock("POST", "/api/v1/tasks")
.with_status(500)
.with_header("content-type", "application/json")
.with_body(&body)
.create_async()
.await;
let err = cmd_create(
"deploy".into(),
"agent:task".into(),
None,
None,
None,
None,
None,
server.url(),
Some(dir.path().to_path_buf()),
)
.await
.unwrap_err()
.to_string();
assert!(!err.contains('\u{1b}'), "ESC leaked: {err:?}");
assert!(!err.contains('\u{202e}'), "RLO leaked: {err:?}");
assert!(
err.chars().count() < 600,
"not bounded: {} chars",
err.chars().count()
);
}
#[tokio::test]
async fn test_create_task_server_error() {
let mut server = mockito::Server::new_async().await;
@@ -476,7 +358,8 @@ mod tests {
.create_async()
.await;
let err = cmd_create(
// Should still succeed (prints JSON, doesn't check status code)
cmd_create(
"deploy".to_string(),
"agent:task".to_string(),
None,
@@ -488,10 +371,7 @@ mod tests {
Some(dir.path().to_path_buf()),
)
.await
.unwrap_err()
.to_string();
assert!(err.contains("500"), "status not surfaced: {err}");
assert!(err.contains("internal error"), "message dropped: {err}");
.unwrap();
}
// ── list ─────────────────────────────────────────────────────────
+22 -5
View File
@@ -5,7 +5,7 @@ use clap::{Args, Subcommand};
use serde_json::Value;
use std::path::PathBuf;
use crate::http::{json_or_denial, NodeClient};
use crate::http::NodeClient;
use crate::identity::load_keypair_from_dir;
#[derive(Args)]
@@ -102,7 +102,13 @@ async fn cmd_create(
.post(&format!("/api/v1/repos/{owner}/{name}/hooks"), &payload)
.await
.context("failed to connect to node")?;
let hook: Value = json_or_denial("create webhook", resp).await?;
let status = resp.status();
let hook: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = hook["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("create webhook failed ({status}): {msg}");
}
let id = hook["id"].as_str().unwrap_or("?");
let hook_events = hook["events"]
@@ -142,7 +148,13 @@ async fn cmd_list(repo: String, node: String, dir: Option<PathBuf>) -> Result<()
let resp = client
.get_signed(&format!("/api/v1/repos/{owner}/{name}/hooks"))
.await?;
let body: Value = json_or_denial("list webhooks", resp).await?;
let status = resp.status();
let body: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = body["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("list webhooks failed ({status}): {msg}");
}
let hooks = body
.get("webhooks")
@@ -190,8 +202,13 @@ async fn cmd_delete(repo: String, id: String, node: String, dir: Option<PathBuf>
)
.await
.context("failed to connect to node")?;
// Parsed only to reject a denial; the success body carries nothing to print.
json_or_denial::<Value>("delete webhook", resp).await?;
let status = resp.status();
let result: Value = resp.json().await.context("invalid JSON")?;
if !status.is_success() {
let msg = result["message"].as_str().unwrap_or("unknown error");
anyhow::bail!("delete webhook failed ({status}): {msg}");
}
println!("✓ Webhook {id} deleted");
Ok(())
+1 -1
View File
@@ -41,7 +41,7 @@ Owner focus: protocol/auth.
- Close default-open repo write authorization and wire trusted UCAN delegation into repository permissions.
- Close the remaining visibility gaps: restrict or document task, pin, and anchor metadata routes, and publish an irreversible-publication policy.
- Add UCAN revocation or blocklisting, with an emergency compromised-key runbook.
- Add mutation-aware GraphQL auth before GraphQL becomes a public write API surface.
- GraphQL mutations require a verified signer (shipped, #87). Keep the surface limited to the agent-task queue; do not expose an owner-gated repo-write mutation over GraphQL without its owner-gate and REST parity. The `every_graphql_mutation_has_its_gate` fence forces any new mutation to be classified in its table (discovered from the schema, so none can slip in unlisted); the owner-gate and REST parity themselves stay the INV-1 review check, not something the fence proves.
- Harden peer registration and outbound peer calls against SSRF and peer-list poisoning.
- After all live peers upgrade, enable signed peer writes by default.
+1 -1
View File
@@ -92,7 +92,7 @@ Fixed or staged in this pass:
Live-network blockers to prioritize:
- GraphQL POST is still open for compatibility; GraphQL mutations should get mutation-aware auth before it becomes a public write API surface.
- GraphQL mutations now require a verified signer bound to the acting DID (#87), and the surface stays limited to the agent-task queue, so there is no public repo-write GraphQL API to gate. Queries remain open by design. (Resolved; kept here as posture, not a live blocker.)
- Push authorization is still not capability-complete. A valid DID signature is authentication, not authorization. Owner checks are now enforced on every branch, protected or not (`GITLAWB_ENFORCE_OWNER_PUSH`, on by default); what remains is that a UCAN `git/push` capability is not yet honored, so a delegated or CI key cannot push.
- UCAN chain validation is incomplete and UCAN revocation/blocklisting is not implemented as an operator feature.
- Private repository reads are not enforced. `is_public` and `GITLAWB_PUBLIC_READ` exist, but per-repository private-read behavior is not wired.