Compare commits

..
Author SHA1 Message Date
beardthelion 7e9a5be6d4 test(node): complete listing-denial regressions for the new gates 2026-11-07 15:08:26 -06:00
beardthelion 0140c9cd86 fix(node): pass SignatureBodyLimit state to optional_signature layers 2026-11-07 06:57:14 -06:00
beardthelion 5274433503 Merge remote-tracking branch 'origin/main' into bug_fix_2
# Conflicts:
#	crates/gitlawb-node/src/server.rs
2026-11-07 06:50:06 -06:00
Gravirei ac9ea61927 All three findings addressed on bug_fix_2:
P2 — signed_scoped_anchors_full_did_form_matches_stored_slug_through_build_router now seeds
  a second repo under a different owner and asserts irys-other-repo-tx is absent. Negative
  control: flipping the production filter to None made the test go RED with count: 2 showing
  the cross-repo leak.

  P3 — let short = owner_did.split(':').next_back().unwrap() replaced with let short =
  crate::db::normalize_owner_key(&owner_did).to_string(). Reader and writer now share one
  convention.

  P3 — PR #134 description updated: pins bullet now describes the actual pre-merge layering
  (and cites the axum-layer-vs-merge-pitfall memory entry), and round 2's stored-slug
  normalization is added as its own bullet with the regression's coverage.

  5/5 anchor-related tests pass, cargo fmt --check clean.
2026-08-30 17:13:53 +06:00
Gravirei 4242bb85ad fix(node): normalize scoped anchor SQL filter to stored repo slug
After authorize_repo_read succeeds, the SQL filter used the raw ?repo=
string (format!("{owner}/{name}")). Anchor rows are stored with
normalize_owner_key(owner_did)/name (short key), so a caller using
?repo=did🔑.../name passed authz but got anchors: [] — a false
empty page that broke callers that legitimately pass the full DID
form. Build the filter from record.owner_did normalized and
record.name instead, matching the shape the seeder uses.
2026-08-30 11:24:12 +06:00
Gravirei 8dd745e98b fix(node): keep /api/v1/ipfs/pins out of the CID resolver's per-IP flood brake
Follow-up to #134: the previous reordering put `/api/v1/ipfs/pins` and
`/ipfs/{cid}` under the same `rate_limit_by_ip` + `IpRateLimiter` layer
stack, so both routes shared the resolver's flood brake bucket. That
bucket is documented as the once-per-request brake for `GET /ipfs/{cid}`
because the resolver can drive bounded-but-expensive repo walks. The
pin listing is a single `list_pinned_cids()` call and has no reason to
share the bucket. Sharing meant /ipfs/{cid} traffic could exhaust the
bucket and 429 the pins endpoint, and signed pin polling could exhaust
the bucket and 429 legitimate CID reads.

Compose `ipfs_routes` from two sub-routers with independent policies,
then merge:
  * `/ipfs/{cid}` keeps `optional_signature` + `rate_limit_by_ip` + the
    `IpRateLimiter` extension. Same behavior as before #134.
  * `/api/v1/ipfs/pins` gets only `optional_signature`. Auth-required
    (handler denies with 401), but no rate limit — it has nothing to
    flood.

Two new `build_router` regressions exercise the bucket independence
through `TrustedProxy::XForwardedFor`:
  * 5 signed pin requests against a size-2 CID bucket all 200, then a
    `/ipfs/{cid}` from the same IP also 200 (pins did not debit the
    bucket).
  * A `/ipfs/{cid}` exhausts a size-1 CID bucket (200, then 429), then
    a signed pin from the same IP is 200 (the bucket does not gate
    pins).

All 65 tests in `server::tests::` + `test_support::tests::ipfs_*`
still pass, including the existing 429 tests for the CID resolver.
2026-08-29 10:52:55 +06:00
Gravirei f49ed79336 fix(node): gate scoped /api/v1/arweave/anchors reads on repository visibility
Closes the review follow-up on #134: a signed non-reader could pass the
auth-layer check on `?repo=<owner>/<name>` and obtain private-repo anchor
metadata (ref names, old/new SHAs, CIDs, irys tx ids). The auth check
proved only that the signature was valid; it did not bind the request to
the canonical repo-read policy.

* Parse the slug through `validate_repo_slug` (same helper the sync path
  uses) and call `authorize_repo_read` before the SQL query. Missing
  repos, quarantined mirrors, and signed non-readers all collapse to the
  standard 404 (`repo_not_found`).
* The unscoped global listing stays auth-only (#121 narrowing; the wider
  permissionless-identity enumeration is the #136 class, out of scope).
* Clamp the limit to `[0, 200]` so `?limit=-1` no longer reaches Postgres
  as `LIMIT -1`.

Five new `build_router` regressions cover the full production contract
through the real `optional_signature` middleware: scoped unsigned 401,
scoped owner 200, scoped non-reader 404 with no anchor metadata leaked
in the body, scoped missing-repo 404 indistinguishable from non-reader,
and `?limit=-1` clamped to an empty 200.
2026-08-29 03:32:01 +06:00
Gravirei 752e117f07 fix(node): format arweave routes for consistency in build_router function 2026-08-28 16:09:52 +06:00
Gravirei f70bb0008e fix(node): enforce authentication on /api/v1/ipfs/pins and /api/v1/arweave/anchors routes 2026-08-28 16:04:51 +06:00
Gravirei 063a5e689f fix(node): gate /ipfs/pins and /arweave/anchors behind authentication (#134)
Add auth rejection to list_pins and list_anchors handlers. The pin index
spans the entire node and would expose metadata for every object ever
pushed; anonymous callers must not see it.

- Require AuthenticatedDid extension in both handlers, return 401 when absent
- Add server.rs regression test for anonymous rejection through build_router
- Fix closed-pool tests to pass authenticated requests (test 503 path)
2026-08-26 16:22:29 +06:00
18 changed files with 136 additions and 3555 deletions

No files matched your search

@@ -137,11 +137,6 @@ enum ShimMode {
struct Shim {
base_url: String,
posts: Arc<AtomicUsize>,
/// Advertisement (GET /info/refs) count. One `git fetch` performs exactly
/// one advertisement, so tests assert this is EXACTLY one: any fetch-level
/// retry that comes back re-advertises and goes red on the spot, which the
/// lower-bound POST assertions alone cannot see (#275 round 6).
gets: Arc<AtomicUsize>,
stop: Arc<AtomicBool>,
handle: Option<std::thread::JoinHandle<()>>,
}
@@ -171,17 +166,15 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim {
let addr = listener.local_addr().unwrap();
let base_url = format!("http://{addr}");
let posts = Arc::new(AtomicUsize::new(0));
let gets = Arc::new(AtomicUsize::new(0));
let stop = Arc::new(AtomicBool::new(false));
let posts_t = posts.clone();
let gets_t = gets.clone();
let stop_t = stop.clone();
let handle = std::thread::spawn(move || {
while !stop_t.load(Ordering::SeqCst) {
match listener.accept() {
Ok((stream, _)) => {
handle_conn(stream, &repo, mode, &posts_t, &gets_t);
handle_conn(stream, &repo, mode, &posts_t);
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(5));
@@ -194,7 +187,6 @@ fn start_shim(repo: PathBuf, mode: ShimMode) -> Shim {
Shim {
base_url,
posts,
gets,
stop,
handle: Some(handle),
}
@@ -220,13 +212,7 @@ fn normalize_accepted_stream(stream: &TcpStream) {
stream.set_read_timeout(Some(Duration::from_secs(30))).ok();
}
fn handle_conn(
stream: TcpStream,
repo: &Path,
mode: ShimMode,
posts: &AtomicUsize,
gets: &AtomicUsize,
) {
fn handle_conn(stream: TcpStream, repo: &Path, mode: ShimMode, posts: &AtomicUsize) {
normalize_accepted_stream(&stream);
let mut reader = BufReader::new(stream);
@@ -277,7 +263,6 @@ fn handle_conn(
receive_pack(repo, false, &body),
)
} else if method == "GET" && target.contains("/info/refs") {
gets.fetch_add(1, Ordering::SeqCst);
// v0 advertisement, wrapped exactly as the node's info_refs does.
let adv = upload_pack(repo, true, b"");
let mut wrapped = pkt(b"# service=git-upload-pack\n");
@@ -1022,7 +1007,6 @@ fn shim_answers_a_connection_that_arrives_non_blocking() {
let server = std::thread::spawn(move || {
let posts = AtomicUsize::new(0);
let gets = AtomicUsize::new(0);
let stream = loop {
match listener.accept() {
Ok((s, _)) => break s,
@@ -1037,7 +1021,7 @@ fn shim_answers_a_connection_that_arrives_non_blocking() {
// the assertions below run on every platform.
stream.set_nonblocking(true).unwrap();
accepted_tx.send(()).unwrap();
handle_conn(stream, &repo, ShimMode::Normal, &posts, &gets);
handle_conn(stream, &repo, ShimMode::Normal, &posts);
});
let mut client = TcpStream::connect(addr).unwrap();
@@ -1101,17 +1085,6 @@ fn real_git_multi_round_fetch_completes() {
posts >= 2,
"fixture did not force multi-round negotiation (observed {posts} POST(s)); the bridging path was not exercised"
);
// Exactly one advertisement: the POST bound above is a floor, so a
// reintroduced fetch-level retry could satisfy it by running a second full
// fetch instead of multi-round negotiation. Re-advertising is what a
// second fetch cannot avoid, so this is the committed guard on the
// retry's absence.
let gets = shim.gets.load(Ordering::SeqCst);
assert_eq!(
gets, 1,
"expected exactly one advertisement GET per fetch; {gets} means a \
fetch-level retry is back"
);
// The fetched tip is present and the clone's object graph is intact.
let server_head = String::from_utf8(git(&server, &["rev-parse", "HEAD"])).unwrap();
@@ -1204,17 +1177,6 @@ fn real_git_withheld_shaped_first_post() {
"helper hung on a withheld-shaped response (should forward-and-terminate, not deadlock). stderr:\n{stderr}"
);
// Exactly one advertisement, pass or fail: the POST assertions below are
// floors, so a reintroduced fetch-level retry could mask the withheld
// shape behind a second full fetch. A retry cannot avoid re-advertising,
// so this is the committed guard on its absence.
let gets = shim.gets.load(Ordering::SeqCst);
assert_eq!(
gets, 1,
"expected exactly one advertisement GET per fetch; {gets} means a \
fetch-level retry is back"
);
if out.status.success() {
// Real git accepted the mid-negotiation pack: the withheld multi-round
// path works end to end with no extra handling.
+4 -157
View File
@@ -5,28 +5,6 @@
//!
//! The schema is frozen at v1. All fields are mandatory for forward compatibility.
//! Nodes that receive a certificate with an unknown version MUST reject it.
//!
//! ## Replay
//!
//! A certificate proves *who* asserted a ref update and *what* they asserted.
//! It does not prove the assertion is fresh. [`RefUpdateBody::seq`] and
//! [`RefUpdateBody::nonce`] are inside the signed bytes (see
//! [`RefUpdateBody::to_signing_bytes`]), so a verifier can trust the values,
//! but deciding whether a nonce is being reused or a `seq` is going backwards
//! needs state.
//!
//! This crate is deliberately stateless — it is embedded in both nodes and
//! clients — so replay is **not** enforced here and cannot be. Ownership of
//! the check sits with the verifier that accepts and persists certificates,
//! which must:
//!
//! - reject a `nonce` it has already accepted for the same `(repo, ref_name)`;
//! - reject a `seq` that does not advance past the last accepted `seq` for
//! that `(repo, ref_name)`.
//!
//! Until such a verifier exists, replay of ref-update certificates is not
//! enforced anywhere in this workspace. This contract exists so the first one
//! inherits the requirement instead of re-deriving it.
use std::collections::HashSet;
@@ -63,19 +41,11 @@ pub struct RefUpdateBody {
pub from: String,
/// The new commit hash (SHA-256 hex).
pub to: String,
/// Monotonically increasing sequence number, scoped to `(repo, ref_name)`.
///
/// Signed, but not enforced by this crate: core holds no state. The
/// verifier that persists certificates must reject a `seq` that does not
/// advance for the same repo and ref. See the module docs on replay.
/// Monotonically increasing sequence number. Prevents replay.
pub seq: u64,
/// RFC 3339 timestamp of this update.
pub timestamp: DateTime<Utc>,
/// Random nonce, unique per certificate.
///
/// Signed, but not enforced by this crate: core holds no state. The
/// verifier that persists certificates must reject a nonce it has already
/// accepted for the same repo and ref. See the module docs on replay.
/// Random nonce for deduplication.
pub nonce: String,
}
@@ -170,16 +140,7 @@ impl RefUpdateCert {
///
/// Counts distinct signer DIDs, not signature entries: a repeated
/// signature from the same maintainer counts once.
///
/// `threshold` must be at least 1. Zero is rejected instead of being
/// treated as "no signatures required": a caller that derives the
/// threshold from configuration and gets 0 would otherwise accept any
/// certificate, whether its signatures verify or not.
pub fn satisfies_threshold(&self, maintainers: &[Did], threshold: usize) -> Result<bool> {
if threshold == 0 {
return Err(Error::RefCert("threshold must be at least 1".to_string()));
}
let valid = self.verify_all()?;
let distinct_signers: HashSet<&Did> =
valid.iter().filter(|d| maintainers.contains(d)).collect();
@@ -417,7 +378,7 @@ mod tests {
}
#[test]
fn threshold_zero_is_rejected() {
fn threshold_zero_is_always_satisfied() {
let kp = Keypair::generate();
let cert = RefUpdateCert::new(
kp.did(),
@@ -428,120 +389,6 @@ mod tests {
&kp,
)
.unwrap();
// Zero must never read as "no signatures required", even with an empty
// maintainer set.
let err = cert.satisfies_threshold(&[], 0).unwrap_err();
assert!(
err.to_string().contains("threshold must be at least 1"),
"zero must be refused as an invalid threshold"
);
}
/// The threshold is checked before any signature work, so a zero is
/// reported as a threshold error even when the certificate's signatures
/// are also invalid. Otherwise a broken certificate would let a caller
/// mistake the failure for a legitimate verification result.
#[test]
fn threshold_zero_is_rejected_before_signature_verification() {
let kp = Keypair::generate();
let other = Keypair::generate();
let mut cert = RefUpdateCert::new(
kp.did(),
"refs/heads/main".to_string(),
dummy_hash('0'),
dummy_hash('a'),
1,
&kp,
)
.unwrap();
cert.signatures[0].sig = other.sign_b64(b"unrelated bytes");
let err = cert.satisfies_threshold(&[kp.did()], 0).unwrap_err();
assert!(
err.to_string().contains("threshold must be at least 1"),
"the invalid threshold must win over the invalid signature"
);
}
/// `seq` and `nonce` are the verifier's replay inputs, so they have to be
/// inside the signed body; tapping either one must invalidate the
/// certificate rather than leave it verifying with different values.
#[test]
fn replay_fields_are_covered_by_the_signature() {
let kp = Keypair::generate();
let cert = RefUpdateCert::new(
kp.did(),
"refs/heads/main".to_string(),
dummy_hash('0'),
dummy_hash('a'),
7,
&kp,
)
.unwrap();
cert.verify_all().expect("baseline must verify");
let mut bumped = cert.clone();
bumped.body.seq = 8;
assert!(
matches!(bumped.verify_all(), Err(Error::SignatureInvalid)),
"changing seq must invalidate the signature"
);
let mut renamed = cert.clone();
renamed.body.nonce = "different-nonce".to_string();
assert!(
matches!(renamed.verify_all(), Err(Error::SignatureInvalid)),
"changing nonce must invalidate the signature"
);
}
/// A signature minted by a real key must not verify when the signer is
/// spelled in another multibase alphabet. Otherwise the same identity
/// would satisfy a maintainer allowlist under several DID strings.
///
/// Base64url is deliberate: its 47-character method-id fits under the
/// pre-existing length bound, so this test reaches the canonical-prefix
/// rule and fails on the pre-fix code instead of being refused for length.
#[test]
fn verify_all_rejects_a_non_canonical_did_spelling() {
let kp = Keypair::generate();
let mut cert = RefUpdateCert::new(
kp.did(),
"refs/heads/main".to_string(),
dummy_hash('0'),
dummy_hash('a'),
1,
&kp,
)
.unwrap();
let mut prefixed = vec![0xed, 0x01];
prefixed.extend_from_slice(&kp.verifying_key().to_bytes());
cert.signatures[0].signer = format!(
"did:key:{}",
multibase::encode(multibase::Base::Base64Url, &prefixed)
)
.parse()
.expect("method is still `key`");
let err = cert
.verify_all()
.expect_err("a non-canonical did:key spelling must not resolve");
assert!(
err.to_string().contains("base58btc"),
"rejection must come from the canonical-prefix rule, not the length guard"
);
assert!(
!matches!(
cert.satisfies_threshold(&[cert.signatures[0].signer.clone()], 1),
Ok(true)
),
"the alternate spelling must not satisfy a threshold"
);
// The canonical spelling of the same key is unaffected.
cert.signatures[0].signer = kp.did();
assert!(cert.satisfies_threshold(&[kp.did()], 1).unwrap());
assert!(cert.satisfies_threshold(&[], 0).unwrap());
}
}
+3 -158
View File
@@ -20,19 +20,6 @@ use crate::{Error, Result};
/// Value: 0xed01 encoded as a varint.
const ED25519_MULTICODEC: &[u8] = &[0xed, 0x01];
/// The only multibase prefix a `did:key` method-specific id may carry.
///
/// The DID:key spec encodes Ed25519 keys as base58btc, which is multibase
/// `z`. `multibase::decode` happily accepts every other alphabet, and each
/// spelling of the same key bytes would produce a distinct DID string while
/// gates, allowlists, and database rows compare DIDs as strings — one
/// identity would read as several. Rejecting the alternatives here keeps the
/// mapping from `did:key` to key single-valued for every caller of
/// [`Did::to_verifying_key`] (HTTP signatures, UCAN, ref certs, peer
/// announces). `gitlawb-attest` enforces the same rule for attestation
/// signers.
const DID_KEY_MULTIBASE_PREFIX: char = 'z';
/// A Decentralized Identifier.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
@@ -84,14 +71,8 @@ impl Did {
/// Resolve the Ed25519 verifying key from a `did:key`.
///
/// Only the canonical base58btc (`z`) spelling is accepted. Other
/// multibase encodings decode to the same key but spell a different DID
/// string, so accepting them would let one identity appear under several
/// identifiers to string-comparing gates and allowlists.
///
/// Returns `Err` if the DID is not a `did:key`, uses a non-canonical
/// multibase prefix, the key bytes are invalid, or the key is a
/// small-order (weak) point.
/// Returns `Err` if the DID is not a `did:key`, the key bytes are malformed,
/// or the key is a small-order (weak) point.
pub fn to_verifying_key(&self) -> Result<VerifyingKey> {
if !self.is_did_key() {
return Err(Error::InvalidDid(format!(
@@ -113,27 +94,9 @@ impl Did {
));
}
// Cheap prefix gate before the decode, then the base check on what
// multibase actually parsed. The first keeps a clearly non-canonical id
// off the decoder. The second is belt-and-braces — multibase maps a
// leading `z` only to `Base58Btc` — but it states the requirement
// directly instead of relying on that mapping, and `gitlawb-attest`
// makes the same pair of checks.
if !self.method_id().starts_with(DID_KEY_MULTIBASE_PREFIX) {
return Err(Error::InvalidDid(
"did:key must use base58btc (z-prefix)".to_string(),
));
}
let (base, bytes) =
let (_, bytes) =
multibase::decode(self.method_id()).map_err(|e| Error::InvalidDid(e.to_string()))?;
if base != Base::Base58Btc {
return Err(Error::InvalidDid(
"did:key must use base58btc (z-prefix)".to_string(),
));
}
if !bytes.starts_with(ED25519_MULTICODEC) {
return Err(Error::InvalidDid(
"not an ed25519 multicodec key".to_string(),
@@ -426,124 +389,6 @@ mod tests {
did.to_verifying_key().expect("a real did:key must resolve");
}
/// `multibase::decode` accepts every alphabet, so the same Ed25519 key can
/// be spelled as several `did:key` strings. Only base58btc is canonical;
/// the rest must be refused so one identity cannot appear as several to
/// gates and allowlists that compare DID strings.
#[test]
fn to_verifying_key_rejects_non_base58btc_multibase() {
use ed25519_dalek::SigningKey;
let signing = SigningKey::from_bytes(&[9u8; 32]);
let mut prefixed = ED25519_MULTICODEC.to_vec();
prefixed.extend_from_slice(&signing.verifying_key().to_bytes());
let canonical = Did::from_verifying_key(&signing.verifying_key());
assert_eq!(
canonical.to_verifying_key().unwrap().to_bytes(),
signing.verifying_key().to_bytes(),
"base58btc is the accepted spelling"
);
// These spellings fit inside the method-id bound, so they reach the
// canonical-prefix check and must be refused there.
for base in [Base::Base64Url, Base::Base32Lower] {
let alternative = format!("did:key:{}", multibase::encode(base, &prefixed));
let did: Did = alternative.parse().expect("method is still `key`");
assert_ne!(
did.as_str(),
canonical.as_str(),
"the same key must not compare equal under another spelling"
);
let err = did
.to_verifying_key()
.expect_err("non-canonical multibase must be refused");
assert!(
matches!(err, Error::InvalidDid(_)),
"must be an InvalidDid, got {err:?}"
);
assert!(
err.to_string().contains("base58btc"),
"the error must name the canonical encoding: {err}"
);
}
// A base16 spelling of the same key is longer than the method-id
// bound, so it is refused by the length guard before the prefix check.
// Either way it must not resolve.
for base in [Base::Base16Lower, Base::Base16Upper] {
let alternative = format!("did:key:{}", multibase::encode(base, &prefixed));
let did: Did = alternative.parse().expect("method is still `key`");
assert!(
matches!(did.to_verifying_key(), Err(Error::InvalidDid(_))),
"a base16 spelling of a real key must not resolve"
);
}
}
/// base58flickr (`Z`) is the non-bitcoin sibling alphabet. Its spelling of
/// a real key fits under the length bound, so it must be refused for the
/// prefix, not merely for its size.
#[test]
fn to_verifying_key_rejects_a_base58flickr_spelling() {
use ed25519_dalek::SigningKey;
let signing = SigningKey::from_bytes(&[11u8; 32]);
let mut prefixed = ED25519_MULTICODEC.to_vec();
prefixed.extend_from_slice(&signing.verifying_key().to_bytes());
let did: Did = format!(
"did:key:{}",
multibase::encode(Base::Base58Flickr, &prefixed)
)
.parse()
.expect("method is still `key`");
let err = did
.to_verifying_key()
.expect_err("base58flickr must be refused");
assert!(
err.to_string().contains("base58btc"),
"the error must name the canonical encoding: {err}"
);
}
/// An empty method-specific id must be refused by the prefix rule without
/// panicking on a missing first character.
#[test]
fn to_verifying_key_rejects_an_empty_method_id() {
let did: Did = "did:key:".parse().expect("method is still `key`");
let err = did
.to_verifying_key()
.expect_err("an empty method id must be refused");
assert!(
err.to_string().contains("base58btc"),
"the error must name the canonical encoding: {err}"
);
}
/// A canonical base58btc spelling is not enough on its own: the bytes must
/// also carry the ed25519-pub multicodec. This pins the check that stops a
/// key of another type from resolving as an Ed25519 key.
#[test]
fn to_verifying_key_rejects_base58btc_with_a_wrong_multicodec() {
use ed25519_dalek::SigningKey;
let signing = SigningKey::from_bytes(&[12u8; 32]);
let mut buf = vec![0x12, 0x00]; // sha2-256, not ed25519-pub
buf.extend_from_slice(&signing.verifying_key().to_bytes());
let did: Did = format!("did:key:{}", multibase::encode(Base::Base58Btc, &buf))
.parse()
.expect("method is still `key`");
let err = did
.to_verifying_key()
.expect_err("a non-ed25519 multicodec must be refused");
assert!(
err.to_string().contains("ed25519"),
"the error must name the expected multicodec: {err}"
);
}
/// A `did:key` can encode a small-order (weak) Ed25519 point. Such a key
/// satisfies the verification equation for any message and, converted to
/// Montgomery form for X25519, yields the all-zero shared secret, so an
-20
View File
@@ -566,26 +566,6 @@ mod tests {
assert!(sig.check_created().is_ok());
}
#[test]
fn extreme_created_timestamps_reject_without_wrapping() {
let kp = Keypair::generate();
let did = kp.did();
let parse = |created: i64| {
let sig_input =
format!(r#"sig1=("@method");keyid="{did}";alg="ed25519";created={created}"#);
HttpSignature::parse(&sig_input, "sig1=:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA:").unwrap()
};
// i64::MIN used to panic in debug (subtraction overflow) and wrap to a
// passing skew in release; now - 2^63 wrapped to i64::MIN, whose abs()
// wrapped back to itself and read as fresh. Both must reject.
for created in [i64::MIN, i64::MAX, Utc::now().timestamp() - i64::MAX] {
assert!(
parse(created).check_created().is_err(),
"created={created} must not pass the freshness gate"
);
}
}
#[test]
fn parse_error_missing_sig1_prefix() {
let err = HttpSignature::parse(
+4 -18
View File
@@ -80,20 +80,15 @@ pub(crate) fn did_matches(a: &str, b: &str) -> bool {
!ka.contains(':') && !kb.contains(':') && ka == kb
}
/// Client-facing denial shared by every repo-owner gate copy. `require_repo_owner`
/// renders it as `AppError::Forbidden`; a handler that emits its own
/// `(StatusCode, Json)` envelope instead of `AppError` (e.g.
/// `api::tasks::create_task`) must use this constant too, so two copies of one
/// gate cannot drift in observable output.
pub(crate) const REPO_OWNER_ONLY_MESSAGE: &str = "only the repo owner can perform this action";
/// 403 unless `caller` is the repo owner. Uses [`did_matches`] so the owner check
/// and the author check (close policy) share one normalization.
pub(crate) fn require_repo_owner(record: &RepoRecord, caller: &str) -> Result<()> {
if did_matches(caller, &record.owner_did) {
Ok(())
} else {
Err(AppError::Forbidden(REPO_OWNER_ONLY_MESSAGE.into()))
Err(AppError::Forbidden(
"only the repo owner can perform this action".into(),
))
}
}
@@ -247,11 +242,6 @@ mod authz_guard {
(events, "list_repo_events", "authorize_repo_read("),
// Bucket C — signer-self: the acting DID is matched/bound to auth.0
(tasks, "create_task", "did_matches("),
// #496: create_task is signer-self AND owner-when-repo-scoped — a
// repo_id naming a hosted repo admits tasks only from its owner.
// Both halves are pinned: the did_matches row guards the signer
// binding, this one the repo-ownership gate.
(tasks, "create_task", "require_repo_owner("),
(tasks, "claim_task", "did_matches("),
(tasks, "complete_task", "did_matches("),
(tasks, "fail_task", "did_matches("),
@@ -351,11 +341,7 @@ mod authz_guard {
(
"create_task",
"createTask",
&[
"require_signer(ctx)?",
"did_matches(",
"require_repo_owner(",
],
&["require_signer(ctx)?", "did_matches("],
),
(
"claim_task",
+46 -763
View File
@@ -3,8 +3,6 @@ use axum::http::StatusCode;
use axum::response::Response;
use axum::Json;
use bytes::Bytes;
use futures::{stream, StreamExt};
use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
@@ -24,18 +22,6 @@ use crate::webhooks;
/// The git all-zeros object id — the create/delete sentinel in a ref update.
const ZERO_SHA: &str = "0000000000000000000000000000000000000000";
const FEDERATED_PEER_CONCURRENCY: usize = 4;
const FEDERATED_PEER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const FEDERATED_AGGREGATE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10);
pub(crate) const MAX_FEDERATED_PEERS: usize = 200;
const MAX_FEDERATED_PEER_BYTES: usize = 512 * 1024;
const MAX_FEDERATED_PEER_REPOS: usize = 200;
const MAX_FEDERATED_REPOS: usize = 1_000;
const MAX_FEDERATED_REPO_JSON_BYTES: usize = 2 * 1024 * 1024;
// The aggregate byte budget counts serialized repo values and separators. Keep
// a fixed reserve for the response object's keys, counters, and closing bytes.
const FEDERATED_RESPONSE_OVERHEAD_BYTES: usize = 256;
/// The set of blob and withheld-subtree tree OIDs withheld from **anonymous**
/// replication for a repo, or `None` when the repo must not replicate at all
/// (private / mode A / undetermined, fail closed). This is the anonymous
@@ -3222,206 +3208,11 @@ pub async fn list_refs(
))
}
struct FederatedPeerRepos {
node_url: String,
node_did: String,
repos: Vec<serde_json::Value>,
truncated: bool,
}
struct FederatedPeerPage {
repos: Vec<serde_json::Value>,
truncated: bool,
}
struct BoundedFederatedPeerRows(Vec<serde_json::Value>, bool);
impl<'de> serde::Deserialize<'de> for BoundedFederatedPeerRows {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct RowsVisitor;
impl<'de> serde::de::Visitor<'de> for RowsVisitor {
type Value = BoundedFederatedPeerRows;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a bounded array of repository objects")
}
fn visit_seq<A>(self, mut rows: A) -> std::result::Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
let mut repos = Vec::with_capacity(MAX_FEDERATED_PEER_REPOS);
while repos.len() < MAX_FEDERATED_PEER_REPOS {
let Some(repo) = rows.next_element::<serde_json::Value>()? else {
return Ok(BoundedFederatedPeerRows(repos, false));
};
if !repo.is_object() {
return Err(<A::Error as serde::de::Error>::custom(
"peer repository row is not an object",
));
}
repos.push(repo);
}
let mut truncated = false;
while rows.next_element::<serde::de::IgnoredAny>()?.is_some() {
truncated = true;
}
Ok(BoundedFederatedPeerRows(repos, truncated))
}
}
deserializer.deserialize_seq(RowsVisitor)
}
}
struct FederatedRepoBudget {
repos: Vec<serde_json::Value>,
json_bytes: usize,
max_repos: usize,
max_json_bytes: usize,
truncated: bool,
}
impl FederatedRepoBudget {
fn new(capacity: usize, max_repos: usize, max_json_bytes: usize) -> Self {
Self {
repos: Vec::with_capacity(capacity.min(max_repos)),
json_bytes: FEDERATED_RESPONSE_OVERHEAD_BYTES,
max_repos,
max_json_bytes,
truncated: false,
}
}
fn push(&mut self, repo: serde_json::Value) -> bool {
if self.is_saturated() {
self.truncated = true;
return false;
}
let Ok(repo_bytes) = serde_json::to_vec(&repo) else {
return true;
};
let charged = repo_bytes.len().saturating_add(1);
if self.json_bytes.saturating_add(charged) > self.max_json_bytes {
self.truncated = true;
return false;
}
self.json_bytes += charged;
self.repos.push(repo);
true
}
fn is_saturated(&self) -> bool {
self.repos.len() >= self.max_repos || self.json_bytes >= self.max_json_bytes
}
}
async fn fetch_federated_peer_repos(
client: &reqwest::Client,
url: &str,
) -> Option<FederatedPeerPage> {
let mut response = client.get(url).send().await.ok()?;
if !response.status().is_success()
|| response
.content_length()
.is_some_and(|len| len > MAX_FEDERATED_PEER_BYTES as u64)
{
return None;
}
let total = response
.headers()
.get("x-total-count")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.trim().parse::<u64>().ok());
let capacity = response
.content_length()
.and_then(|len| usize::try_from(len).ok())
.unwrap_or(0)
.min(MAX_FEDERATED_PEER_BYTES);
let mut body = Vec::with_capacity(capacity);
while let Some(chunk) = response.chunk().await.ok()? {
if body.len().saturating_add(chunk.len()) > MAX_FEDERATED_PEER_BYTES {
return None;
}
body.extend_from_slice(&chunk);
}
let BoundedFederatedPeerRows(repos, overflow) = serde_json::from_slice(&body).ok()?;
let truncated = overflow || total.is_some_and(|total| total > repos.len() as u64);
Some(FederatedPeerPage { repos, truncated })
}
fn enrich_federated_repo(
mut repo: serde_json::Value,
node_url: &str,
node_did: &str,
) -> Option<serde_json::Value> {
let object = repo.as_object_mut()?;
object.insert(
"node_url".to_string(),
serde_json::Value::String(node_url.to_string()),
);
object.insert(
"node_did".to_string(),
serde_json::Value::String(node_did.to_string()),
);
object.insert("local".to_string(), serde_json::Value::Bool(false));
Some(repo)
}
async fn collect_federated_fetches<I, F>(fetches: I, aggregate: &mut FederatedRepoBudget) -> usize
where
I: IntoIterator<Item = F>,
F: Future<Output = Option<FederatedPeerRepos>>,
{
let mut responses = stream::iter(fetches).buffer_unordered(FEDERATED_PEER_CONCURRENCY);
let mut nodes_queried = 0;
let deadline = tokio::time::Instant::now() + FEDERATED_AGGREGATE_DEADLINE;
loop {
let response = match tokio::time::timeout_at(deadline, responses.next()).await {
Ok(Some(response)) => response,
Ok(None) => break,
Err(_) => {
aggregate.truncated = true;
tracing::warn!("federated aggregation deadline exceeded");
break;
}
};
let Some(response) = response else {
aggregate.truncated = true;
tracing::warn!("federated peer fetch failed or exceeded its response limits");
continue;
};
nodes_queried += 1;
aggregate.truncated |= response.truncated;
for repo in response.repos {
let Some(repo) = enrich_federated_repo(repo, &response.node_url, &response.node_did)
else {
continue;
};
if !aggregate.push(repo) || aggregate.is_saturated() {
aggregate.truncated = true;
return nodes_queried;
}
}
}
nodes_queried
}
/// GET /api/v1/repos/federated
///
/// Query all known peers for their public repos and return a merged view of
/// the network. Each repo includes a `node_url` and `node_did` indicating
/// which node hosts it. Peer work is bounded by `FEDERATED_AGGREGATE_DEADLINE`
/// and `MAX_FEDERATED_PEERS`; omitted
/// results set `truncated`. The route admits 12 requests per minute per client IP.
/// which node hosts it. Results from unreachable peers are silently omitted.
pub async fn list_federated_repos(
State(state): State<AppState>,
auth: Option<Extension<AuthenticatedDid>>,
@@ -3449,76 +3240,66 @@ pub async fn list_federated_repos(
.unwrap_or_else(|| "http://127.0.0.1:7545".to_string());
let local_node_did = state.node_did.to_string();
let mut aggregate = FederatedRepoBudget::new(
local_repos.len(),
MAX_FEDERATED_REPOS,
MAX_FEDERATED_REPO_JSON_BYTES,
);
let mut all_repos: Vec<serde_json::Value> = Vec::with_capacity(local_repos.len());
for (r, count) in &local_repos {
let Ok(mut repo) = serde_json::to_value(to_response(r, &state, *count)) else {
continue;
};
repo["node_url"] = serde_json::Value::String(local_node_url.clone());
repo["node_did"] = serde_json::Value::String(local_node_did.clone());
repo["local"] = serde_json::Value::Bool(true);
if !aggregate.push(repo) || aggregate.is_saturated() {
aggregate.truncated = true;
break;
}
let mut v = serde_json::to_value(to_response(r, &state, *count)).unwrap_or_default();
v["node_url"] = serde_json::Value::String(local_node_url.clone());
v["node_did"] = serde_json::Value::String(local_node_did.clone());
v["local"] = serde_json::Value::Bool(true);
all_repos.push(v);
}
// Keep peer work bounded even when the table contains many reachable rows.
// Dropping the buffered stream at the aggregate ceiling cancels its in-flight
// request futures and leaves the remaining iterator entries unpolled.
// One extra row detects omitted peers without materializing the full table.
let mut peers = state
.db
.list_federation_peers(MAX_FEDERATED_PEERS as i64 + 1)
.await?;
aggregate.truncated |= peers.len() > MAX_FEDERATED_PEERS;
peers.truncate(MAX_FEDERATED_PEERS);
let fetches: Vec<_> = peers
// Query peers in parallel
let peers = state.db.list_peers().await.unwrap_or_default();
let client = &state.http_client;
let fetch_tasks: Vec<_> = peers
.into_iter()
.filter(|p| p.last_ping_ok && !p.http_url.is_empty())
.map(|peer| {
let client = Arc::clone(&state.http_client);
let url = format!(
"{}/api/v1/repos?limit={MAX_FEDERATED_PEER_REPOS}&offset=0",
peer.http_url.trim_end_matches('/')
);
let client = Arc::clone(client);
let url = format!("{}/api/v1/repos", peer.http_url.trim_end_matches('/'));
let peer_did = peer.did.clone();
let peer_url = peer.http_url.clone();
async move {
tokio::time::timeout(
FEDERATED_PEER_TIMEOUT,
fetch_federated_peer_repos(&client, &url),
tokio::spawn(async move {
let result = tokio::time::timeout(
std::time::Duration::from_secs(5),
client.get(&url).send(),
)
.await
.ok()
.flatten()
.map(|page| FederatedPeerRepos {
node_url: peer_url,
node_did: peer_did,
repos: page.repos,
truncated: page.truncated,
})
}
.await;
match result {
Ok(Ok(resp)) if resp.status().is_success() => {
if let Ok(repos) = resp.json::<Vec<serde_json::Value>>().await {
let enriched: Vec<serde_json::Value> = repos
.into_iter()
.map(|mut r| {
r["node_url"] = serde_json::Value::String(peer_url.clone());
r["node_did"] = serde_json::Value::String(peer_did.clone());
r["local"] = serde_json::Value::Bool(false);
r
})
.collect();
return enriched;
}
}
_ => {}
}
vec![]
})
})
.collect();
let peer_nodes_queried = if aggregate.is_saturated() {
0
} else {
collect_federated_fetches(fetches, &mut aggregate).await
};
for task in fetch_tasks {
if let Ok(repos) = task.await {
all_repos.extend(repos);
}
}
let count = aggregate.repos.len();
let truncated = aggregate.truncated;
let count = all_repos.len();
Ok(Json(serde_json::json!({
"repos": aggregate.repos,
"repos": all_repos,
"count": count,
"nodes_queried": 1 + peer_nodes_queried,
"truncated": truncated,
"nodes_queried": 1, // local + peers that responded
})))
}
@@ -4504,46 +4285,6 @@ mod tests {
}
}
#[sqlx::test]
async fn federated_peer_query_is_bounded(pool: sqlx::PgPool) {
let state = crate::test_support::test_state(pool.clone()).await;
sqlx::query("INSERT INTO peers (did, http_url, last_ping_ok, announced_at) SELECT 'peer-' || n, 'https://example.com', TRUE, '2026-01-01T00:00:00Z' FROM generate_series(1, $1) n")
.bind(MAX_FEDERATED_PEERS as i64 + 5)
.execute(&pool).await.unwrap();
let peers = state.db.list_federation_peers(i64::MAX).await.unwrap();
assert_eq!(peers.len(), MAX_FEDERATED_PEERS + 1);
assert_eq!(state.db.list_federation_peers(2).await.unwrap().len(), 2);
assert!(state.db.list_federation_peers(0).await.unwrap().is_empty());
}
#[sqlx::test]
async fn federated_route_brakes_repeated_anonymous_requests(pool: sqlx::PgPool) {
use tower::ServiceExt;
let mut state = crate::test_support::test_state(pool).await;
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state.federated_rate_limiter =
crate::rate_limit::RateLimiter::new_bounded(1, std::time::Duration::from_secs(60), 10);
let router = crate::server::build_router(state);
for n in 0..2 {
let mut request = axum::http::Request::builder()
.uri("/api/v1/repos/federated")
.body(axum::body::Body::empty())
.unwrap();
request.extensions_mut().insert(axum::extract::ConnectInfo(
"203.0.113.42:5000".parse::<std::net::SocketAddr>().unwrap(),
));
let response = router.clone().oneshot(request).await.unwrap();
assert_eq!(
response.status(),
if n == 0 {
StatusCode::OK
} else {
StatusCode::TOO_MANY_REQUESTS
}
);
}
}
#[test]
fn fork_clone_guard_removes_its_own_mirror_on_drop() {
let root = tempfile::TempDir::new().unwrap();
@@ -4591,464 +4332,6 @@ mod tests {
);
}
#[tokio::test]
async fn federated_route_uses_the_state_owned_bucket_before_database_access() {
use tower::ServiceExt;
let mut state = crate::test_support::test_state_lazy();
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state.federated_rate_limiter =
crate::rate_limit::RateLimiter::new_bounded(1, std::time::Duration::from_secs(60), 10);
assert!(state.federated_rate_limiter.check("203.0.113.42").await);
// Both routers must use the same exhausted bucket from AppState.
for router in [
crate::server::build_router(state.clone()),
crate::server::build_router(state),
] {
let mut request = axum::http::Request::builder()
.uri("/api/v1/repos/federated")
.body(axum::body::Body::empty())
.unwrap();
request.extensions_mut().insert(axum::extract::ConnectInfo(
"203.0.113.42:5000".parse::<std::net::SocketAddr>().unwrap(),
));
let response =
tokio::time::timeout(std::time::Duration::from_secs(1), router.oneshot(request))
.await
.expect("the exhausted bucket must reject before accessing the lazy database")
.unwrap();
assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS);
}
}
#[tokio::test]
async fn federated_aggregate_deadline_returns_partial_results() {
let mut budget = FederatedRepoBudget::new(0, 1000, MAX_FEDERATED_REPO_JSON_BYTES);
let fetches =
(0..8).map(|_| async { std::future::pending::<Option<FederatedPeerRepos>>().await });
let nodes = tokio::time::timeout(
FEDERATED_AGGREGATE_DEADLINE + std::time::Duration::from_secs(2),
collect_federated_fetches(fetches, &mut budget),
)
.await
.expect("the aggregate must stop before the caller deadline");
assert_eq!(nodes, 0);
assert!(budget.truncated);
}
#[tokio::test]
async fn federated_peer_response_enforces_byte_and_row_ceilings() {
let mut row_server = mockito::Server::new_async().await;
let row_body =
serde_json::to_string(&vec![serde_json::json!({}); MAX_FEDERATED_PEER_REPOS + 1])
.unwrap();
let row_mock = row_server
.mock("GET", "/api/v1/repos")
.with_status(200)
.with_header("content-type", "application/json")
.with_body(row_body)
.expect(1)
.create_async()
.await;
let client = reqwest::Client::new();
let page =
fetch_federated_peer_repos(&client, &format!("{}/api/v1/repos", row_server.url()))
.await
.unwrap();
assert_eq!(page.repos.len(), MAX_FEDERATED_PEER_REPOS);
assert!(page.truncated);
row_mock.assert_async().await;
let byte_body = serde_json::to_string(&vec![serde_json::json!({
"description": "x".repeat(MAX_FEDERATED_PEER_BYTES)
})])
.unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let byte_server = tokio::spawn(async move {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let (mut socket, _) = listener.accept().await.unwrap();
let mut request = [0; 1_024];
let _ = socket.read(&mut request).await.unwrap();
socket
.write_all(
format!(
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
transfer-encoding: chunked\r\nconnection: close\r\n\r\n{:x}\r\n",
byte_body.len()
)
.as_bytes(),
)
.await
.unwrap();
let _ = socket.write_all(byte_body.as_bytes()).await;
let _ = socket.write_all(b"\r\n0\r\n\r\n").await;
});
assert!(
fetch_federated_peer_repos(&client, &format!("http://{address}/api/v1/repos"))
.await
.is_none(),
"a chunked peer body above the byte ceiling must be omitted without Content-Length"
);
byte_server.await.unwrap();
}
#[tokio::test]
async fn federated_peer_total_count_prevents_false_complete_page() {
let mut server = mockito::Server::new_async().await;
let body = serde_json::to_string(
&(0..MAX_FEDERATED_PEER_REPOS)
.map(|id| serde_json::json!({ "id": id }))
.collect::<Vec<_>>(),
)
.unwrap();
let page_mock = server
.mock("GET", "/api/v1/repos")
.with_status(200)
.with_header("content-type", "application/json")
.with_header("x-total-count", &(MAX_FEDERATED_PEER_REPOS + 1).to_string())
.with_body(body)
.expect(1)
.create_async()
.await;
let page = fetch_federated_peer_repos(
&reqwest::Client::new(),
&format!("{}/api/v1/repos", server.url()),
)
.await
.expect("the capped peer page is otherwise valid");
page_mock.assert_async().await;
assert_eq!(page.repos.len(), MAX_FEDERATED_PEER_REPOS);
assert!(page.truncated, "the peer total proves another row exists");
let mut aggregate = FederatedRepoBudget::new(300, 300, MAX_FEDERATED_REPO_JSON_BYTES);
let nodes = collect_federated_fetches(
[std::future::ready(Some(FederatedPeerRepos {
node_url: server.url(),
node_did: "did:key:peer".to_string(),
repos: page.repos,
truncated: page.truncated,
}))],
&mut aggregate,
)
.await;
assert_eq!(nodes, 1);
assert_eq!(aggregate.repos.len(), MAX_FEDERATED_PEER_REPOS);
assert!(
aggregate.truncated,
"the aggregate must not report the paged peer as complete"
);
let mut malformed_server = mockito::Server::new_async().await;
let malformed_mock = malformed_server
.mock("GET", "/api/v1/repos")
.with_status(200)
.with_header("content-type", "application/json")
.with_header("x-total-count", "not-a-count")
.with_body("[{}]")
.expect(1)
.create_async()
.await;
let malformed = fetch_federated_peer_repos(
&reqwest::Client::new(),
&format!("{}/api/v1/repos", malformed_server.url()),
)
.await
.expect("a malformed optional total must not discard a bounded page");
malformed_mock.assert_async().await;
assert!(!malformed.truncated);
assert_eq!(malformed.repos.len(), 1);
}
#[test]
fn federated_aggregate_stays_inside_row_and_serialized_byte_budgets() {
let mut rows = FederatedRepoBudget::new(3, 2, 4_096);
assert!(rows.push(serde_json::json!({ "name": "one" })));
assert!(rows.push(serde_json::json!({ "name": "two" })));
assert!(rows.is_saturated());
assert!(!rows.push(serde_json::json!({ "name": "three" })));
assert_eq!(rows.repos.len(), 2);
assert!(rows.truncated);
let max_bytes = 600;
let mut bytes = FederatedRepoBudget::new(8, 8, max_bytes);
while bytes.push(serde_json::json!({ "description": "x".repeat(100) })) {}
let count = bytes.repos.len();
let truncated = bytes.truncated;
let response = serde_json::json!({
"repos": bytes.repos,
"count": count,
"nodes_queried": 1,
"truncated": truncated,
});
assert!(serde_json::to_vec(&response).unwrap().len() <= max_bytes);
assert!(response["truncated"].as_bool().unwrap());
}
struct ActiveFetch {
active: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}
impl Drop for ActiveFetch {
fn drop(&mut self) {
self.active
.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
}
}
#[tokio::test]
async fn federated_collection_bounds_concurrency_and_cancels_at_aggregate_limit() {
use std::sync::atomic::{AtomicUsize, Ordering};
let active = Arc::new(AtomicUsize::new(0));
let max_active = Arc::new(AtomicUsize::new(0));
let started = Arc::new(AtomicUsize::new(0));
let mut senders = Vec::new();
let mut fetches = Vec::new();
for i in 0..10 {
let (tx, rx) = tokio::sync::oneshot::channel();
senders.push(Some(tx));
let active = Arc::clone(&active);
let max_active = Arc::clone(&max_active);
let started = Arc::clone(&started);
fetches.push(async move {
started.fetch_add(1, Ordering::SeqCst);
let current = active.fetch_add(1, Ordering::SeqCst) + 1;
max_active.fetch_max(current, Ordering::SeqCst);
let _active = ActiveFetch { active };
rx.await.ok()?;
Some(FederatedPeerRepos {
node_url: format!("https://peer-{i}.example"),
node_did: format!("did:key:peer-{i}"),
repos: vec![serde_json::json!({ "id": i })],
truncated: false,
})
});
}
let task = tokio::spawn(async move {
let mut aggregate = FederatedRepoBudget::new(1, 1, usize::MAX);
let nodes = collect_federated_fetches(fetches, &mut aggregate).await;
(aggregate, nodes)
});
tokio::time::timeout(std::time::Duration::from_secs(1), async {
while started.load(Ordering::SeqCst) < FEDERATED_PEER_CONCURRENCY {
tokio::task::yield_now().await;
}
})
.await
.expect("the bounded fetch window should fill");
assert_eq!(started.load(Ordering::SeqCst), FEDERATED_PEER_CONCURRENCY);
assert_eq!(active.load(Ordering::SeqCst), FEDERATED_PEER_CONCURRENCY);
assert_eq!(
max_active.load(Ordering::SeqCst),
FEDERATED_PEER_CONCURRENCY
);
senders[0].take().unwrap().send(()).unwrap();
let (aggregate, nodes) = tokio::time::timeout(std::time::Duration::from_secs(1), task)
.await
.expect("the aggregate ceiling should stop collection")
.unwrap();
assert_eq!(nodes, 1);
assert_eq!(aggregate.repos.len(), 1);
assert!(aggregate.truncated);
assert_eq!(started.load(Ordering::SeqCst), FEDERATED_PEER_CONCURRENCY);
assert_eq!(
active.load(Ordering::SeqCst),
0,
"in-flight fetches must be dropped"
);
}
async fn json_body(resp: axum::response::Response) -> serde_json::Value {
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.expect("body bytes");
serde_json::from_slice(&bytes).expect("json body")
}
fn federated_request() -> axum::http::Request<axum::body::Body> {
let mut request = axum::http::Request::builder()
.uri("/api/v1/repos/federated")
.body(axum::body::Body::empty())
.unwrap();
request.extensions_mut().insert(axum::extract::ConnectInfo(
"203.0.113.42:5000".parse::<std::net::SocketAddr>().unwrap(),
));
request
}
/// Pins the federated response contract end to end: the envelope fields
/// (`count`, `nodes_queried`, `truncated`) and the per-repo annotations
/// (`node_url`, `node_did`, `local`) are asserted on an actual HTTP 200
/// body driven through `build_router`, not on a self-constructed value.
/// The seeded repo is deliberately owned by a DID other than the node:
/// the envelope must carry the node's DID, so a regression that reports
/// the owner DID instead would fail the `node_did` assertion.
#[sqlx::test]
async fn federated_route_envelope_annotates_repos_on_real_200_body(pool: sqlx::PgPool) {
use tower::ServiceExt;
let state = crate::test_support::test_state(pool).await;
let node_did = state.node_did.to_string();
state
.db
.create_repo(&repo_owned_by("did:key:z6MkEnvelopeOwnerNotTheNode"))
.await
.expect("seed local repo");
let router = crate::server::build_router(state);
let response = router.oneshot(federated_request()).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["count"].as_u64(), Some(1));
assert_eq!(
body["nodes_queried"].as_u64(),
Some(1),
"no peers seeded, only the local node was queried"
);
assert_eq!(body["truncated"].as_bool(), Some(false));
let repo = &body["repos"][0];
assert_eq!(repo["name"].as_str(), Some("demo"));
assert_eq!(repo["local"].as_bool(), Some(true));
assert_eq!(repo["node_did"].as_str(), Some(node_did.as_str()));
assert!(
repo["node_url"].as_str().is_some_and(|u| !u.is_empty()),
"local repos must carry the node's URL, got {repo}"
);
}
/// `truncated` must be observable on the wire when the peer table holds
/// more rows than `MAX_FEDERATED_PEERS`. Every seeded peer points at the
/// same reachable stub returning `[]`, so no fetch can fail and no peer
/// page can report truncation: the count bound is the only reason the
/// body says `truncated`. The `nodes_queried` assertion pins the fetch
/// side: every peer inside the bound is queried, and dropping the count
/// bound would report `truncated: false`.
#[sqlx::test]
async fn federated_route_reports_truncated_when_peer_table_overflows(pool: sqlx::PgPool) {
use tower::ServiceExt;
let mut peer_server = mockito::Server::new_async().await;
let peer_mock = peer_server
.mock("GET", "/api/v1/repos")
.match_query(mockito::Matcher::Exact(format!(
"limit={MAX_FEDERATED_PEER_REPOS}&offset=0"
)))
.with_status(200)
.with_header("content-type", "application/json")
.with_body("[]")
.expect(MAX_FEDERATED_PEERS)
.create_async()
.await;
let state = crate::test_support::test_state(pool.clone()).await;
sqlx::query("INSERT INTO peers (did, http_url, last_ping_ok, announced_at) SELECT 'peer-' || n, $1, TRUE, '2026-01-01T00:00:00Z' FROM generate_series(1, $2) n")
.bind(peer_server.url())
.bind(MAX_FEDERATED_PEERS as i64 + 2)
.execute(&pool)
.await
.unwrap();
let router = crate::server::build_router(state);
let response = router.oneshot(federated_request()).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(
body["truncated"].as_bool(),
Some(true),
"more peers than the bound must surface as truncated"
);
assert_eq!(
body["nodes_queried"].as_u64(),
Some(1 + MAX_FEDERATED_PEERS as u64),
"the stub is reachable, so every peer inside the bound is queried"
);
peer_mock.assert_async().await;
}
/// A single unreachable peer must still flip `truncated` on the wire via
/// the failed-fetch path, even though the peer table stays under the
/// bound. Guards the `aggregate.truncated = true` on fetch failure.
#[sqlx::test]
async fn federated_route_reports_truncated_when_peer_fetch_fails(pool: sqlx::PgPool) {
use tower::ServiceExt;
let state = crate::test_support::test_state(pool.clone()).await;
sqlx::query("INSERT INTO peers (did, http_url, last_ping_ok, announced_at) VALUES ('did:key:deadpeer', 'http://127.0.0.1:1/', TRUE, '2026-01-01T00:00:00Z')")
.execute(&pool)
.await
.unwrap();
let router = crate::server::build_router(state);
let response = router.oneshot(federated_request()).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(
body["truncated"].as_bool(),
Some(true),
"a failed peer fetch must surface as truncated"
);
assert_eq!(body["nodes_queried"].as_u64(), Some(1));
}
/// Peer repos must carry `node_url`, `node_did` and `local: false` on the
/// real response body. Guards `enrich_federated_repo`: dropping an
/// annotation there must turn this red.
#[sqlx::test]
async fn federated_route_annotates_peer_repos_on_real_200_body(pool: sqlx::PgPool) {
use tower::ServiceExt;
let mut peer_server = mockito::Server::new_async().await;
let peer_mock = peer_server
.mock("GET", "/api/v1/repos")
// Pin the outbound peer query parameters: production must request
// the bounded page. Dropping the parameters stops matching this
// mock, so the peer fetch fails and the body reports
// `truncated: true` instead of `false` below.
.match_query(mockito::Matcher::Exact(format!(
"limit={MAX_FEDERATED_PEER_REPOS}&offset=0"
)))
.with_status(200)
.with_header("content-type", "application/json")
.with_body(r#"[{"id": "peer-repo-1", "name": "from-peer"}]"#)
.expect(1)
.create_async()
.await;
let state = crate::test_support::test_state(pool.clone()).await;
sqlx::query("INSERT INTO peers (did, http_url, last_ping_ok, announced_at) VALUES ('did:key:peer1', $1, TRUE, '2026-01-01T00:00:00Z')")
.bind(peer_server.url())
.execute(&pool)
.await
.unwrap();
let router = crate::server::build_router(state);
let response = router.oneshot(federated_request()).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = json_body(response).await;
assert_eq!(body["truncated"].as_bool(), Some(false));
assert_eq!(body["nodes_queried"].as_u64(), Some(2));
let repo = body["repos"]
.as_array()
.unwrap()
.iter()
.find(|r| r["id"].as_str() == Some("peer-repo-1"))
.expect("peer repo present in federated body");
assert_eq!(repo["node_url"].as_str(), Some(peer_server.url().as_str()));
assert_eq!(repo["node_did"].as_str(), Some("did:key:peer1"));
assert_eq!(repo["local"].as_bool(), Some(false));
peer_mock.assert_async().await;
}
#[test]
fn upload_pack_request_finalizes_only_with_done_pktline() {
let want = "0032want 1111111111111111111111111111111111111111\n";
+11 -123
View File
@@ -30,33 +30,6 @@ fn forbidden(msg: &str) -> (StatusCode, Json<Value>) {
)
}
/// This module's error shape with an opaque body: the real error is logged
/// server-side and never serialized (#250), matching the envelope
/// `AppError::Db` renders. Also mirrors its status split: pool-class failures
/// are a retryable 503 `db_unavailable`, everything else the opaque 500
/// `db_error`, so a transient outage is not reported as permanent.
fn db_error(e: anyhow::Error) -> (StatusCode, Json<Value>) {
tracing::error!(error = %format!("{e:#}"), "task database error");
if let Some(sql) = e.chain().find_map(|c| c.downcast_ref::<sqlx::Error>()) {
if crate::error::db_unavailable(sql) {
return (
StatusCode::SERVICE_UNAVAILABLE,
Json(json!({
"error": crate::error::DB_UNAVAILABLE_CODE,
"message": crate::error::DB_UNAVAILABLE_MESSAGE,
})),
);
}
}
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({
"error": "db_error",
"message": crate::error::DB_ERROR_MESSAGE,
})),
)
}
// ── Request / response types ──────────────────────────────────────────────────
#[derive(Deserialize)]
@@ -128,37 +101,6 @@ pub async fn create_task(
if !crate::api::did_matches(&auth.0, &body.delegator_did) {
return Err(forbidden("delegator_did must be the authenticated signer"));
}
// #496: a caller-supplied repo_id must name a repo the caller owns. Without
// this, any signed caller can plant a task — payload and UCAN included —
// under a foreign repo id, where the named assignee can claim it. Resolve
// the id against hosted repos: a hosted, non-quarantined repo admits tasks
// only from its owner (403 otherwise). An id naming no hosted repo
// (unknown, or quarantined — which is hidden as if it did not exist, so
// 403ing it would confirm a real id) is stored verbatim as an opaque
// label: this is a write gate and does not bind an unresolved id to a
// repo. Task reads are caller-independent on this base (the list endpoints
// are not repo-scoped), before and after this change. The 403-vs-201 split
// between a hosted repo id and an unknown one is itself a one-bit oracle —
// a stranger holding a private repo's id learns it is hosted here — but
// denying foreign repos requires drawing exactly that line, and repo ids
// already surface via the task list on this base. Repo-less tasks are
// unaffected.
if let Some(repo_id) = body.repo_id.as_deref() {
let record = state.db.get_repo_by_id(repo_id).await.map_err(db_error)?;
if let Some(record) = record {
let quarantined = state
.db
.is_repo_quarantined(&record.id)
.await
.map_err(db_error)?;
if !quarantined && crate::api::require_repo_owner(&record, &auth.0).is_err() {
// Same denial text the GraphQL copy relays from
// `require_repo_owner`, so the two copies of this gate do not
// disagree in observable output.
return Err(forbidden(crate::api::REPO_OWNER_ONLY_MESSAGE));
}
}
}
let now = Utc::now().to_rfc3339();
let task = AgentTask {
id: Uuid::new_v4().to_string(),
@@ -175,7 +117,12 @@ pub async fn create_task(
updated_at: now,
deadline: body.deadline,
};
let task = state.db.create_task(&task).await.map_err(db_error)?;
state.db.create_task(&task).await.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(json!({ "error": e.to_string() })),
)
})?;
Ok((StatusCode::CREATED, Json(task_to_json(&task))))
}
@@ -228,28 +175,16 @@ pub async fn claim_task(
return Err(forbidden("assignee_did must be the authenticated signer"));
}
let task = state.db.claim_task(&id, &auth.0).await.map_err(|e| {
if e.downcast_ref::<crate::db::TaskReservedForOtherAssignee>()
.is_some()
{
return forbidden("task not claimable: reserved for another assignee");
}
(
StatusCode::CONFLICT,
Json(json!({ "error": e.to_string() })),
)
})?;
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.filter(|s| !s.is_empty())
.map(str::to_string)
.unwrap_or_else(|| auth.0.clone());
let _ = state.task_event_tx.send(TaskEventBroadcast {
task_id: id,
old_status: "pending".to_string(),
new_status: "claimed".to_string(),
by_did,
by_did: auth.0,
at: Utc::now().to_rfc3339(),
});
Ok(Json(task_to_json(&task)))
@@ -284,14 +219,14 @@ pub async fn complete_task(
})?;
if !crate::api::did_matches(
&auth.0,
crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
existing.assignee_did.as_deref().unwrap_or_default(),
) {
return Err(forbidden("only the task assignee can complete it"));
}
let by_did = auth.0;
let task = state
.db
.finish_task(&id, "completed", body.result.as_deref(), &by_did)
.finish_task(&id, "completed", body.result.as_deref())
.await
.map_err(|e| {
(
@@ -299,13 +234,6 @@ pub async fn complete_task(
Json(json!({ "error": e.to_string() })),
)
})?;
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.filter(|s| !s.is_empty())
.map(str::to_string)
.unwrap_or(by_did);
let _ = state.task_event_tx.send(TaskEventBroadcast {
task_id: id,
old_status: "claimed".to_string(),
@@ -344,7 +272,7 @@ pub async fn fail_task(
})?;
if !crate::api::did_matches(
&auth.0,
crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
existing.assignee_did.as_deref().unwrap_or_default(),
) {
return Err(forbidden("only the task assignee can fail it"));
}
@@ -352,7 +280,7 @@ pub async fn fail_task(
let reason = body.reason.unwrap_or_default();
let task = state
.db
.finish_task(&id, "failed", Some(&reason), &by_did)
.finish_task(&id, "failed", Some(&reason))
.await
.map_err(|e| {
(
@@ -360,13 +288,6 @@ pub async fn fail_task(
Json(json!({ "error": e.to_string() })),
)
})?;
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.filter(|s| !s.is_empty())
.map(str::to_string)
.unwrap_or(by_did);
let _ = state.task_event_tx.send(TaskEventBroadcast {
task_id: id,
old_status: "claimed".to_string(),
@@ -376,36 +297,3 @@ pub async fn fail_task(
});
Ok(Json(task_to_json(&task)))
}
#[cfg(test)]
mod tests {
use super::*;
/// The gate's DB helper must mirror `AppError::Db`: a pool-level failure is
/// the retryable 503 `db_unavailable`, a query fault the opaque 500.
#[test]
fn db_error_matches_app_error_status_split() {
let unavailable = db_error(anyhow::Error::from(sqlx::Error::PoolTimedOut));
assert_eq!(unavailable.0, StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(unavailable.1 .0["error"], crate::error::DB_UNAVAILABLE_CODE);
assert_eq!(
unavailable.1 .0["message"],
crate::error::DB_UNAVAILABLE_MESSAGE
);
let query = db_error(anyhow::Error::from(sqlx::Error::Protocol(
"column \"machine_id\" does not exist".into(),
)));
assert_eq!(query.0, StatusCode::INTERNAL_SERVER_ERROR);
assert_eq!(query.1 .0["error"], "db_error");
assert_eq!(query.1 .0["message"], crate::error::DB_ERROR_MESSAGE);
assert!(!query.1 .0.to_string().contains("machine_id"));
}
/// `.context()` layers must not hide the sqlx error from the status split.
#[test]
fn db_error_sees_sqlx_through_context() {
let wrapped = anyhow::Error::from(sqlx::Error::PoolClosed).context("repo lookup failed");
assert_eq!(db_error(wrapped).0, StatusCode::SERVICE_UNAVAILABLE);
}
}
-1
View File
@@ -668,7 +668,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)),
federated_rate_limiter: RateLimiter::new_bounded(12, Duration::from_secs(60), 10_000),
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)),
+17 -138
View File
@@ -3027,26 +3027,6 @@ impl Db {
)
}
pub async fn list_federation_peers(&self, limit: i64) -> Result<Vec<PeerRecord>> {
let rows = sqlx::query(
"SELECT did, http_url, last_seen, last_ping_ok, announced_at
FROM peers WHERE last_ping_ok = TRUE AND http_url <> '' ORDER BY last_seen DESC NULLS LAST, did LIMIT $1",
)
.bind(limit.clamp(0, crate::api::repos::MAX_FEDERATED_PEERS as i64 + 1))
.fetch_all(&self.pool)
.await?;
Ok(rows
.into_iter()
.map(|r| PeerRecord {
did: r.get("did"),
http_url: r.get("http_url"),
last_seen: r.get("last_seen"),
last_ping_ok: r.get::<bool, _>("last_ping_ok"),
announced_at: r.get("announced_at"),
})
.collect())
}
pub async fn list_peers(&self) -> Result<Vec<PeerRecord>> {
let rows = sqlx::query(
"SELECT did, http_url, last_seen, last_ping_ok, announced_at
@@ -4000,54 +3980,28 @@ impl Db {
// ── Agent Tasks ───────────────────────────────────────────────────────────────
/// Permanent authorization denial: the task is reserved for a different assignee.
/// Handlers downcast this to return 403 (vs 409 for a lost claim race).
#[derive(Debug, thiserror::Error)]
#[error("task not claimable: reserved for another assignee")]
pub struct TaskReservedForOtherAssignee;
/// ASCII whitespace treated as blank for assignee slots — must stay in sync with
/// `BTRIM(assignee_did, E' \t\n\r')` in `claim_task`'s SQL open-slot check.
const ASSIGNEE_BLANK: &[char] = &[' ', '\t', '\n', '\r'];
fn assignee_slot_blank(s: &str) -> bool {
s.trim_matches(ASSIGNEE_BLANK).is_empty()
}
/// Trim assignee DIDs for auth matching (same ASCII blank set as SQL/open checks).
pub fn trim_assignee_did(s: &str) -> &str {
s.trim_matches(ASSIGNEE_BLANK)
}
impl Db {
/// Persist a task. Whitespace-only `assignee_did` is stored as `NULL` (open).
/// Returns the normalized row shape so REST/GraphQL create responses match GET.
pub async fn create_task(&self, task: &AgentTask) -> Result<AgentTask> {
let mut stored = task.clone();
stored.assignee_did = stored
.assignee_did
.take()
.filter(|s| !assignee_slot_blank(s));
pub async fn create_task(&self, task: &AgentTask) -> Result<()> {
sqlx::query(
"INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)",
)
.bind(&stored.id)
.bind(&stored.repo_id)
.bind(&stored.kind)
.bind(&stored.status)
.bind(&stored.delegator_did)
.bind(&stored.assignee_did)
.bind(&stored.capability)
.bind(&stored.ucan_token)
.bind(&stored.payload)
.bind(&stored.result)
.bind(&stored.created_at)
.bind(&stored.updated_at)
.bind(&stored.deadline)
.bind(&task.id)
.bind(&task.repo_id)
.bind(&task.kind)
.bind(&task.status)
.bind(&task.delegator_did)
.bind(&task.assignee_did)
.bind(&task.capability)
.bind(&task.ucan_token)
.bind(&task.payload)
.bind(&task.result)
.bind(&task.created_at)
.bind(&task.updated_at)
.bind(&task.deadline)
.execute(&self.pool)
.await?;
Ok(stored)
Ok(())
}
pub async fn get_task(&self, id: &str) -> Result<Option<AgentTask>> {
@@ -4104,102 +4058,38 @@ impl Db {
Ok(rows.into_iter().map(row_to_task).collect())
}
/// Claim a pending task for `assignee_did`.
///
/// If the task was created with a non-blank pre-set `assignee_did`, only that
/// agent (DID-normalized via [`crate::api::did_matches`]) may claim it. Open
/// tasks (`NULL` / blank) remain first-claimer-wins. A reserved claim keeps
/// the **exact** stored DID form via `COALESCE($4, $2)` (no BTRIM rewrite) so
/// exact-match list filters still work. The UPDATE also re-checks the
/// assignee slot as defense-in-depth against a future writer racing the
/// pre-check.
pub async fn claim_task(&self, id: &str, assignee_did: &str) -> Result<AgentTask> {
let now = Utc::now().to_rfc3339();
// Denial path only needs status + assignee — do not load payload/UCAN
// for a permissionless caller who will be rejected.
let row = sqlx::query("SELECT status, assignee_did FROM agent_tasks WHERE id = $1")
.bind(id)
.fetch_optional(&self.pool)
.await?
.ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))?;
let status: String = row.get("status");
if status != "pending" {
return Err(anyhow::anyhow!(
"task not claimable: not found or already claimed"
));
}
let stored: Option<String> = row.get("assignee_did");
// Blank reservations are treated as open. Keep the exact stored string
// for the UPDATE equality check and SET (do not BTRIM the reserved form).
// Blank definition matches SQL `BTRIM(..., E' \t\n\r')` (not space-only).
let reserved_exact = stored.as_deref().filter(|r| !assignee_slot_blank(r));
if let Some(reserved) = reserved_exact {
if !crate::api::did_matches(assignee_did, trim_assignee_did(reserved)) {
return Err(TaskReservedForOtherAssignee.into());
}
}
// Keep the exact reserved form ($4); only fill assignee on open claims ($2).
// Treat blank stored values as open for the SQL slot check.
let row = sqlx::query(
"UPDATE agent_tasks SET status='claimed',
assignee_did = COALESCE($4, $2),
updated_at=$3
"UPDATE agent_tasks SET status='claimed', assignee_did=$2, updated_at=$3
WHERE id=$1 AND status='pending'
AND (
($4::text IS NULL AND (assignee_did IS NULL OR BTRIM(assignee_did, E' \t\n\r') = ''))
OR assignee_did = $4
)
RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
)
.bind(id)
.bind(assignee_did)
.bind(&now)
.bind(reserved_exact)
.fetch_optional(&self.pool)
.await?;
row.map(row_to_task)
.ok_or_else(|| anyhow::anyhow!("task not claimable: not found or already claimed"))
}
/// Transition a claimed task to `new_status` (`completed` / `failed`).
///
/// `actor_did` must be the task's assignee (DID-normalized). The UPDATE
/// binds the exact stored `assignee_did` so a concurrent reassignment /
/// re-claim cannot finish under a check-then-act race in the handler.
pub async fn finish_task(
&self,
id: &str,
new_status: &str,
result: Option<&str>,
actor_did: &str,
) -> Result<AgentTask> {
let now = Utc::now().to_rfc3339();
let existing = self
.get_task(id)
.await?
.ok_or_else(|| anyhow::anyhow!("task not found or not in claimed state"))?;
if existing.status != "claimed" {
return Err(anyhow::anyhow!("task not found or not in claimed state"));
}
let Some(ref assigned) = existing.assignee_did else {
return Err(anyhow::anyhow!("task not found or not in claimed state"));
};
// Trim matches claim_task so tab-padded stored values can finish.
if !crate::api::did_matches(actor_did, trim_assignee_did(assigned)) {
return Err(anyhow::anyhow!(
"task not finishable: only the assignee may finish it"
));
}
let row = sqlx::query(
"UPDATE agent_tasks SET status=$2, result=$3, updated_at=$4
WHERE id=$1 AND status='claimed' AND assignee_did=$5
WHERE id=$1 AND status='claimed'
RETURNING id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline",
)
.bind(id)
.bind(new_status)
.bind(result)
.bind(&now)
.bind(assigned.as_str())
.fetch_optional(&self.pool)
.await?;
row.map(row_to_task)
@@ -9393,10 +9283,6 @@ mod peer_authority_tests {
/// | `gossip_ping_round_requires_two_failures_before_persisting_unreachable` (main.rs) | test-only fixture seed. Raw SQL because the test drives the readiness HYSTERESIS, which needs a row already at `last_ping_ok = TRUE` before the round runs; it never exercises the announce gate |
/// | `manual_ping_uses_readiness_without_mutating_federation_gate` (api/peers.rs) | test-only fixture seed, same shape and same reason: the row under test must pre-exist so the assertion is about what the ping does NOT rewrite |
/// | `seed_peer` (p2p/mod.rs) | test-only fixture seed. Raw SQL so the gossip ingest tests can drive the known-peer gate directly, including the did:web case `upsert_peer` would refuse |
/// | `federated_peer_query_is_bounded` (api/repos.rs) | test-only fixture seed for the bounded federation query; inserts reachable rows without exercising peer admission |
/// | `federated_route_annotates_peer_repos_on_real_200_body` (api/repos.rs) | test-only fixture seed; inserts the mock peer row the bounded federation query reads, so the response-envelope assertions run against a real HTTP 200 body |
/// | `federated_route_reports_truncated_when_peer_fetch_fails` (api/repos.rs) | test-only fixture seed; seeds one unreachable peer so the fetch-failure path is the witness under test |
/// | `federated_route_reports_truncated_when_peer_table_overflows` (api/repos.rs) | test-only fixture seed; overflows the peer table past `MAX_FEDERATED_PEERS` so the count bound is the only witness for `truncated` |
///
/// And the `upsert_peer` CALL-SITE authority table, which the ledger above
/// structurally cannot hold, because the bootstrap site issues no SQL of its own
@@ -9482,13 +9368,6 @@ mod peers_table_writer_guard {
/// listed function that no longer has one.
const LEDGER: &[(&str, usize)] = &[
("a_legacy_row_can_still_refresh_its_liveness", 1),
("federated_peer_query_is_bounded", 1),
("federated_route_annotates_peer_repos_on_real_200_body", 1),
("federated_route_reports_truncated_when_peer_fetch_fails", 1),
(
"federated_route_reports_truncated_when_peer_table_overflows",
1,
),
(
"gossip_ping_round_requires_two_failures_before_persisting_unreachable",
1,
+1 -1
View File
@@ -101,7 +101,7 @@ pub const DB_ERROR_MESSAGE: &str = "a database error occurred";
/// Connection-level sqlx failures that mean the database is unreachable right
/// now (retryable, 503), as opposed to server-reported query errors.
pub(crate) fn db_unavailable(e: &sqlx::Error) -> bool {
fn db_unavailable(e: &sqlx::Error) -> bool {
matches!(
e,
sqlx::Error::PoolTimedOut
+7 -415
View File
@@ -36,33 +36,6 @@ impl MutationRoot {
}
let delegator_did = caller.to_string();
let db = ctx.data_unchecked::<Arc<Db>>();
// #496: same write gate as the REST handler — a repo_id naming a
// hosted, non-quarantined repo admits tasks only from its owner, so a
// stranger cannot file under a foreign repo id. An id naming no hosted
// repo (unknown, or quarantined — hidden as if it did not exist) is
// stored verbatim as an opaque label; the gate does not bind an
// unresolved id to a repo. Task reads are caller-independent on this
// base (the tasks resolver is not repo-scoped), before and after this
// change. As in tasks.rs, the success-vs-denial split between a hosted
// repo id and an unknown one is itself a one-bit oracle — a stranger
// holding a private repo's id learns it is hosted here — but denying
// foreign repos requires drawing exactly that line.
if let Some(repo_id) = input.repo_id.as_deref() {
let record = db
.get_repo_by_id(repo_id)
.await
.map_err(crate::graphql::graphql_db_err)?;
if let Some(record) = record {
let quarantined = db
.is_repo_quarantined(&record.id)
.await
.map_err(crate::graphql::graphql_db_err)?;
if !quarantined {
crate::api::require_repo_owner(&record, caller)
.map_err(crate::graphql::graphql_app_err)?;
}
}
}
let now = Utc::now().to_rfc3339();
let task = AgentTask {
id: Uuid::new_v4().to_string(),
@@ -79,14 +52,10 @@ impl MutationRoot {
updated_at: now,
deadline: input.deadline,
};
// Return the stored row, not the pre-insert input: create_task
// normalizes the assignee (whitespace-only becomes NULL/open), and a
// response that shows the raw input lies about the reservation.
let stored = db
.create_task(&task)
db.create_task(&task)
.await
.map_err(crate::graphql::graphql_db_err)?;
Ok(AgentTaskType::from(stored))
Ok(AgentTaskType::from(task))
}
async fn claim_task(
@@ -108,19 +77,11 @@ impl MutationRoot {
.claim_task(&id, &assignee_did)
.await
.map_err(crate::graphql::graphql_db_err)?;
// Report the stored assignee form like REST, so subscribers see the
// same identity whichever API surface ran the transition.
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.unwrap_or(assignee_did.as_str())
.to_string();
let _ = tx.send(TaskEventBroadcast {
task_id: id,
old_status: "pending".to_string(),
new_status: "claimed".to_string(),
by_did,
by_did: assignee_did,
at: Utc::now().to_rfc3339(),
});
Ok(AgentTaskType::from(task))
@@ -149,24 +110,15 @@ impl MutationRoot {
.await
.map_err(crate::graphql::graphql_db_err)?
.ok_or_else(|| async_graphql::Error::new("task not found"))?;
if !crate::api::did_matches(
caller,
crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
) {
if !crate::api::did_matches(caller, existing.assignee_did.as_deref().unwrap_or_default()) {
return Err(async_graphql::Error::new(
"only the task assignee can complete it",
));
}
let task = db
.finish_task(&id, "completed", input.result.as_deref(), caller)
.finish_task(&id, "completed", input.result.as_deref())
.await
.map_err(crate::graphql::graphql_db_err)?;
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.unwrap_or(by_did.as_str())
.to_string();
let _ = tx.send(TaskEventBroadcast {
task_id: id,
old_status: "claimed".to_string(),
@@ -199,25 +151,16 @@ impl MutationRoot {
.await
.map_err(crate::graphql::graphql_db_err)?
.ok_or_else(|| async_graphql::Error::new("task not found"))?;
if !crate::api::did_matches(
caller,
crate::db::trim_assignee_did(existing.assignee_did.as_deref().unwrap_or_default()),
) {
if !crate::api::did_matches(caller, existing.assignee_did.as_deref().unwrap_or_default()) {
return Err(async_graphql::Error::new(
"only the task assignee can fail it",
));
}
let reason = input.reason.unwrap_or_default();
let task = db
.finish_task(&id, "failed", Some(&reason), caller)
.finish_task(&id, "failed", Some(&reason))
.await
.map_err(crate::graphql::graphql_db_err)?;
let by_did = task
.assignee_did
.as_deref()
.map(crate::db::trim_assignee_did)
.unwrap_or(by_did.as_str())
.to_string();
let _ = tx.send(TaskEventBroadcast {
task_id: id,
old_status: "claimed".to_string(),
@@ -243,22 +186,6 @@ mod tests {
.join("; ")
}
/// The `id` scalar returned by a successful `createTask` mutation.
fn created_id(resp: &Response) -> String {
assert!(resp.errors.is_empty(), "graphql errors: {:?}", resp.errors);
let async_graphql::Value::Object(obj) = &resp.data else {
panic!("data not an object: {:?}", resp.data);
};
let async_graphql::Value::Object(task) = obj.get("createTask").expect("createTask key")
else {
panic!("createTask not an object");
};
let async_graphql::Value::String(id) = task.get("id").expect("id key") else {
panic!("id not a string");
};
id.clone()
}
/// N2: every GraphQL mutation requires a verified signer and binds the acting
/// DID to it. Unsigned is rejected; a signer other than the claimed actor is
/// rejected; the matching signer passes the auth gate.
@@ -426,339 +353,4 @@ mod tests {
errors(&resp)
);
}
/// #496 (GraphQL): createTask applies the same repo-ownership gate as the
/// REST handler — a stranger naming a hosted repo is rejected without
/// persisting anything, while the owner files. The quarantined and
/// unknown-id arms are pinned too, including that each accepted write
/// stores the supplied id verbatim, so this copy of the gate cannot regress
/// silently.
#[sqlx::test]
async fn create_task_rejects_foreign_repo_id(pool: PgPool) {
let state = crate::test_support::test_state(pool).await;
let owner = "did:key:zGQLTASKOWNERAAAAAAAAAAAAAAAAAAAAAAAAAA";
let stranger = "did:key:zGQLTASKSTRANGERBBBBBBBBBBBBBBBBBBBBBB";
let now = chrono::Utc::now();
let repo = crate::db::RepoRecord {
id: uuid::Uuid::new_v4().to_string(),
name: "gql-task-gate-repo".to_string(),
owner_did: owner.to_string(),
description: None,
is_public: true,
default_branch: "main".to_string(),
created_at: now,
updated_at: now,
disk_path: "/tmp/gql-task-gate-repo".to_string(),
forked_from: None,
machine_id: None,
};
state.db.create_repo(&repo).await.expect("seed repo");
let schema = state.graphql_schema.as_ref();
let q = |actor: &str| {
format!(
r#"mutation {{ createTask(delegatorDid: "{actor}", input: {{ kind: "build", capability: "repo:write", repoId: "{}" }}) {{ id }} }}"#,
repo.id
)
};
// Stranger signs as themselves (so the signer binding passes) but
// names the victim's repo → rejected by the ownership gate, leaking
// nothing about the repo.
let resp = schema
.execute(Request::new(q(stranger)).data(AuthenticatedDid(stranger.into())))
.await;
let errs = errors(&resp);
assert!(
errs.contains(crate::api::REPO_OWNER_ONLY_MESSAGE),
"a non-owner must not file under a foreign repo id (shared denial text): {errs}"
);
// The denial must not persist: no task row may carry the victim's
// repo id afterwards.
let stored = state
.db
.list_tasks(None, None, 50)
.await
.expect("list tasks");
assert!(
stored
.iter()
.all(|t| t.repo_id.as_deref() != Some(repo.id.as_str())),
"a denied write must not leave a task under the foreign repo id"
);
// The owner files under their own repo id.
let resp = schema
.execute(Request::new(q(owner)).data(AuthenticatedDid(owner.into())))
.await;
assert!(
errors(&resp).is_empty(),
"the owner should file against their repo: {}",
errors(&resp)
);
// Unknown repo ids stay accepted as opaque labels.
let q_unknown = format!(
r#"mutation {{ createTask(delegatorDid: "{stranger}", input: {{ kind: "build", capability: "repo:write", repoId: "no-such-repo-id" }}) {{ id }} }}"#
);
let resp = schema
.execute(Request::new(q_unknown).data(AuthenticatedDid(stranger.into())))
.await;
assert!(
errors(&resp).is_empty(),
"an id naming no hosted repo stays an opaque label: {}",
errors(&resp)
);
// Read back the row: the unresolved id must be stored verbatim, not
// dropped or rewritten to a repo it never named.
let stored = state
.db
.get_task(&created_id(&resp))
.await
.expect("get task")
.expect("task must be stored");
assert_eq!(
stored.repo_id.as_deref(),
Some("no-such-repo-id"),
"an unresolved id must be stored verbatim as a label"
);
// A quarantined repo id is treated as unknown (accepted, never a 403
// that would confirm the id names a real repo). This pins the
// `!quarantined` arm on this copy of the gate.
state
.db
.set_repo_quarantine(&repo.id, true)
.await
.expect("quarantine");
let resp = schema
.execute(Request::new(q(stranger)).data(AuthenticatedDid(stranger.into())))
.await;
assert!(
errors(&resp).is_empty(),
"a quarantined repo id must not 403 (existence oracle): {}",
errors(&resp)
);
// Read back the row: the quarantined id is stored verbatim as a label
// too, so the accepted write is not silently dropped.
let stored = state
.db
.get_task(&created_id(&resp))
.await
.expect("get task")
.expect("task must be stored");
assert_eq!(
stored.repo_id.as_deref(),
Some(repo.id.as_str()),
"a quarantined id must be stored verbatim as a label"
);
}
/// #496 (GraphQL): a repo-lookup fault in createTask's gate must stay
/// opaque. Dropping a column `get_repo_by_id` selects faults that lookup;
/// the resolver must return the fixed DB message and leak no schema text
/// (#250), as `create_task_db_error_message_is_opaque` pins for the
/// agent_tasks fault.
#[sqlx::test]
async fn create_task_gate_repo_lookup_fault_is_opaque(pool: PgPool) {
let state = crate::test_support::test_state(pool.clone()).await;
let owner = "did:key:zGQLTASKFAULTOWNERAAAAAAAAAAAAAAAAAAAAAAAA";
let now = chrono::Utc::now();
let repo = crate::db::RepoRecord {
id: uuid::Uuid::new_v4().to_string(),
name: "gql-task-gate-fault-repo".to_string(),
owner_did: owner.to_string(),
description: None,
is_public: true,
default_branch: "main".to_string(),
created_at: now,
updated_at: now,
disk_path: "/tmp/gql-task-gate-fault-repo".to_string(),
forked_from: None,
machine_id: None,
};
state.db.create_repo(&repo).await.expect("seed repo");
sqlx::query("ALTER TABLE repos DROP COLUMN machine_id")
.execute(&pool)
.await
.expect("drop machine_id");
let q = format!(
r#"mutation {{ createTask(delegatorDid: "{owner}", input: {{ kind: "build", capability: "repo:write", repoId: "{}" }}) {{ id }} }}"#,
repo.id
);
let resp = state
.graphql_schema
.execute(Request::new(&q).data(AuthenticatedDid(owner.into())))
.await;
let errs = errors(&resp);
assert!(
errs.contains(crate::graphql::GRAPHQL_DB_ERROR_MESSAGE),
"gate sqlx fault must be opaque: {errs}"
);
assert!(
!errs.contains("machine_id") && !errs.contains("column"),
"schema text leaked through the gate: {errs}"
);
}
/// Pre-assigned GraphQL claimTask must reject a stranger (no UCAN leak)
/// and admit the reserved assignee.
#[sqlx::test]
async fn claim_task_honors_preassigned_assignee(pool: PgPool) {
let state = crate::test_support::test_state(pool).await;
let reserved = "did:key:zGQLRESERVEDAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
let thief = "did:key:zGQLTHIEFBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let now = chrono::Utc::now().to_rfc3339();
let task = crate::db::AgentTask {
id: "task-gql-reserved".into(),
repo_id: None,
kind: "build".into(),
status: "pending".into(),
delegator_did: "did:key:zGQLDELEGATORCCCCCCCCCCCCCCCCCCCCCCCCCC".into(),
assignee_did: Some(reserved.into()),
capability: "repo:write".into(),
ucan_token: Some("gql-ucan-secret".into()),
payload: None,
result: None,
created_at: now.clone(),
updated_at: now,
deadline: None,
};
state.db.create_task(&task).await.expect("seed");
let schema = state.graphql_schema.as_ref();
let q = |actor: &str| {
format!(
r#"mutation {{ claimTask(id: "task-gql-reserved", assigneeDid: "{actor}") {{ id status ucanToken }} }}"#
)
};
let resp = schema
.execute(Request::new(q(thief)).data(AuthenticatedDid(thief.into())))
.await;
let errs = errors(&resp);
assert!(
errs.contains("reserved"),
"stranger claim must surface reserved deny distinctly: {errs}"
);
assert!(
!errs.contains("gql-ucan-secret"),
"UCAN must not leak in GraphQL errors: {errs}"
);
if let Ok(data) = resp.data.into_json() {
let s = data.to_string();
assert!(
!s.contains("gql-ucan-secret"),
"UCAN must not leak in GraphQL data on failed claim: {s}"
);
}
// Parity with REST: failed steal must leave the row untouched.
let still = state
.db
.get_task("task-gql-reserved")
.await
.expect("get")
.expect("task exists");
assert_eq!(
still.status, "pending",
"status must stay pending after steal"
);
assert_eq!(
still.assignee_did.as_deref(),
Some(reserved),
"assignee_did must stay reserved after steal"
);
let resp = schema
.execute(Request::new(q(reserved)).data(AuthenticatedDid(reserved.into())))
.await;
assert!(
errors(&resp).is_empty(),
"reserved assignee must claim: {}",
errors(&resp)
);
let data = resp.data.into_json().expect("json data");
assert_eq!(data["claimTask"]["status"], "claimed");
assert_eq!(data["claimTask"]["ucanToken"], "gql-ucan-secret");
}
/// create_task normalizes a whitespace-only assignee to NULL (open task),
/// so the create response must show the stored row, not echo the input: a
/// client trusting the echoed "reservation" would believe a claimable task
/// is reserved. Before the fix the response was built from the pre-insert
/// struct and returned the padded input verbatim.
#[sqlx::test]
async fn create_task_reports_stored_assignee_not_input(pool: PgPool) {
let state = crate::test_support::test_state(pool).await;
let caller = "did:key:zGQLCREATORAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
let schema = state.graphql_schema.as_ref();
let q = format!(
r#"mutation {{ createTask(delegatorDid: "{caller}", input: {{ kind: "build", capability: "repo:write", assigneeDid: " " }}) {{ id assigneeDid }} }}"#
);
let resp = schema
.execute(Request::new(q).data(AuthenticatedDid(caller.into())))
.await;
let data = resp.data.into_json().expect("json data");
assert!(
data["createTask"]["assigneeDid"].is_null(),
"whitespace-only assigneeDid must come back null (stored shape): {data}"
);
}
/// A task seeded with a whitespace-padded assignee must still be
/// completable by that assignee over GraphQL: REST and Db::finish_task
/// trim the stored value before comparing, and the GraphQL gate must do
/// the same or padded reservations finish only through REST.
#[sqlx::test]
async fn graphql_finish_accepts_padded_stored_assignee(pool: PgPool) {
let state = crate::test_support::test_state(pool).await;
let assignee = "did:key:zGQLPADDEDAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
let now = chrono::Utc::now().to_rfc3339();
let task = crate::db::AgentTask {
id: "task-gql-padded".into(),
repo_id: None,
kind: "build".into(),
status: "pending".into(),
delegator_did: "did:key:zGQLDELEGATORDDDDDDDDDDDDDDDDDDDDDDDDDD".into(),
assignee_did: Some(format!("\t{assignee}\n")),
capability: "repo:write".into(),
ucan_token: None,
payload: None,
result: None,
created_at: now.clone(),
updated_at: now,
deadline: None,
};
state.db.create_task(&task).await.expect("seed");
let schema = state.graphql_schema.as_ref();
let claim = format!(
r#"mutation {{ claimTask(id: "task-gql-padded", assigneeDid: "{assignee}") {{ id status }} }}"#
);
let resp = schema
.execute(Request::new(claim).data(AuthenticatedDid(assignee.into())))
.await;
assert!(
errors(&resp).is_empty(),
"reserved assignee must claim a padded reservation: {}",
errors(&resp)
);
let finish = format!(
r#"mutation {{ completeTask(id: "task-gql-padded", byDid: "{assignee}", input: {{}}) {{ id status }} }}"#
);
let resp = schema
.execute(Request::new(finish).data(AuthenticatedDid(assignee.into())))
.await;
let errs = errors(&resp);
assert!(
errs.is_empty(),
"assignee must complete a padded reservation: {errs}"
);
let data = resp.data.into_json().expect("json data");
assert_eq!(data["completeTask"]["status"], "completed");
}
}
-5
View File
@@ -429,8 +429,6 @@ async fn main() -> Result<()> {
std::time::Duration::from_secs(3600),
200_000,
);
let federated_rate_limiter =
rate_limit::RateLimiter::new_bounded(12, std::time::Duration::from_secs(60), 10_000);
if config.sync_trigger_rate_limit == 0 {
tracing::warn!(
"GITLAWB_SYNC_TRIGGER_RATE_LIMIT=0 — /sync/trigger IP rate limiting disabled"
@@ -469,7 +467,6 @@ async fn main() -> Result<()> {
push_limiter_trust,
sync_trigger_rate_limiter,
peer_write_rate_limiter,
federated_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(
@@ -1174,7 +1171,6 @@ mod rate_limiter_sweep_tests {
state.close_issue_rate_limiter = RateLimiter::new(10, window);
state.sync_trigger_rate_limiter = RateLimiter::new(10, window);
state.peer_write_rate_limiter = RateLimiter::new(10, window);
state.federated_rate_limiter = RateLimiter::new(10, window);
state.ipfs_rate_limiter = RateLimiter::new(10, window);
let limiters = |s: &crate::state::AppState| {
@@ -1185,7 +1181,6 @@ mod rate_limiter_sweep_tests {
s.close_issue_rate_limiter.clone(),
s.sync_trigger_rate_limiter.clone(),
s.peer_write_rate_limiter.clone(),
s.federated_rate_limiter.clone(),
s.ipfs_rate_limiter.clone(),
]
};
+3 -692
View File
@@ -7,11 +7,11 @@
//! The node's PeerId comes from an Ed25519 keypair loaded from a persistent
//! key file, so the PeerId is stable across restarts.
use std::collections::{hash_map::DefaultHasher, HashMap, HashSet, VecDeque};
use std::collections::{hash_map::DefaultHasher, HashMap};
use std::hash::{Hash, Hasher};
use std::path::{Component, Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::time::Duration;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
@@ -315,17 +315,6 @@ fn is_zero(v: &u32) -> bool {
*v == 0
}
// Identify is an untrusted address source. Keep enough entries for normal
// multi-homed peers while bounding both one peer and the whole routing table.
const IDENTIFY_ADDRESS_LIMIT: usize = 8;
// Bound canonicalization and sorting too, including rejected/duplicate entries.
const IDENTIFY_REPORT_ADDRESS_LIMIT: usize = 64;
const IDENTIFY_NEW_ADDRESS_LIMIT: usize = 8;
const IDENTIFY_GLOBAL_ADDRESS_LIMIT: usize = 1024;
const IDENTIFY_ADDRESS_WINDOW: Duration = Duration::from_secs(60);
const IDENTIFY_ADDRESS_TTL: Duration = Duration::from_secs(30 * 60);
const IDENTIFY_ADDRESS_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
/// A ref-update event published to Gossipsub when a push lands.
///
/// The signing bytes are this struct serialized with `sig` set to None (see
@@ -1642,200 +1631,6 @@ fn did_to_kad_key(did: &str) -> kad::RecordKey {
kad::RecordKey::new(&format!("/gitlawb/did/{did}").as_bytes())
}
/// Returns true when `address` was explicitly configured for `peer` through
/// AddKnownPeer. Explicit addresses are retained in Kademlia even when an
/// overlapping Identify lease expires or is evicted.
fn is_explicit_address(
explicit_addresses: &HashMap<PeerId, HashSet<Multiaddr>>,
peer: &PeerId,
address: &Multiaddr,
) -> bool {
explicit_addresses
.get(peer)
.is_some_and(|addresses| addresses.contains(address))
}
#[derive(Debug)]
struct IdentifyAddress {
address: Multiaddr,
expires_at: Instant,
}
#[derive(Debug)]
struct IdentifyPeerAddresses {
addresses: VecDeque<IdentifyAddress>,
window_started: Instant,
new_addresses: usize,
}
#[derive(Debug, Default, PartialEq, Eq)]
struct IdentifyAddressChanges {
added: Vec<Multiaddr>,
refreshed: Vec<Multiaddr>,
removed: Vec<(PeerId, Multiaddr)>,
}
#[derive(Debug, Default)]
struct IdentifyAddressBook {
peers: HashMap<PeerId, IdentifyPeerAddresses>,
address_count: usize,
}
impl IdentifyAddressBook {
fn update(
&mut self,
peer_id: PeerId,
now: Instant,
addresses: &[Multiaddr],
) -> IdentifyAddressChanges {
let mut changes = IdentifyAddressChanges::default();
if self.peers.contains_key(&peer_id) {
changes.removed.extend(self.expire_peer(peer_id, now));
}
let reported = addresses
.iter()
.take(IDENTIFY_REPORT_ADDRESS_LIMIT)
.filter_map(|address| address.clone().with_p2p(peer_id).ok())
.collect::<HashSet<_>>();
if reported.is_empty() && !self.peers.contains_key(&peer_id) {
return changes;
}
let state = self
.peers
.entry(peer_id)
.or_insert_with(|| IdentifyPeerAddresses {
addresses: VecDeque::new(),
window_started: now,
new_addresses: 0,
});
if now.saturating_duration_since(state.window_started) >= IDENTIFY_ADDRESS_WINDOW {
state.window_started = now;
state.new_addresses = 0;
}
// Refresh every retained address in the bounded report before admission.
// Refreshed addresses are reported so the Identify handler can re-admit
// them to Kademlia. Kademlia may have dropped a retained address
// after a failed dial or never kept it for a full k-bucket, and a
// TTL refresh alone would never re-add it.
for existing in &mut state.addresses {
if reported.contains(&existing.address) {
existing.expires_at = now + IDENTIFY_ADDRESS_TTL;
changes.refreshed.push(existing.address.clone());
}
}
let mut new_addresses: Vec<Multiaddr> = reported
.iter()
.filter(|address| {
!state
.addresses
.iter()
.any(|existing| &existing.address == *address)
})
.cloned()
.collect();
// HashSet iteration is deliberately unordered. Sort new candidates so
// an oversized Identify report has deterministic admission behavior.
new_addresses.sort();
for address in new_addresses {
if state.new_addresses >= IDENTIFY_NEW_ADDRESS_LIMIT {
break;
}
let evicted = if state.addresses.len() < IDENTIFY_ADDRESS_LIMIT {
None
} else {
state
.addresses
.iter()
.position(|existing| !reported.contains(&existing.address))
.and_then(|index| state.addresses.remove(index))
};
if let Some(evicted) = evicted {
self.address_count -= 1;
changes.removed.push((peer_id, evicted.address));
} else if state.addresses.len() >= IDENTIFY_ADDRESS_LIMIT {
// Every retained address was refreshed in this report. Keep
// that stable subset and drop surplus new candidates.
break;
}
if self.address_count >= IDENTIFY_GLOBAL_ADDRESS_LIMIT {
// Reject the admission instead of evicting another peer's
// entry: cross-peer eviction lets a flood of new identities
// push honest peers' addresses out of the book and, when the
// evicted address is their last one, out of the Kademlia
// routing table. Per-peer eviction above still applies to the
// reporting peer's own addresses.
break;
}
state.new_addresses += 1;
state.addresses.push_back(IdentifyAddress {
address: address.clone(),
expires_at: now + IDENTIFY_ADDRESS_TTL,
});
self.address_count += 1;
changes.added.push(address);
}
// Do not keep a row when every candidate was rejected: a zero-address
// row is invisible to address_count and would otherwise linger until
// the expiry tick. This matches expire_peer, which drops empty rows.
if state.addresses.is_empty() {
self.peers.remove(&peer_id);
}
changes
}
fn expire(&mut self, now: Instant) -> Vec<(PeerId, Multiaddr)> {
let mut removed = Vec::new();
let peers = self.peers.keys().copied().collect::<Vec<_>>();
for peer_id in peers {
removed.extend(self.expire_peer(peer_id, now));
}
removed
}
fn expire_peer(&mut self, peer_id: PeerId, now: Instant) -> Vec<(PeerId, Multiaddr)> {
let expired = self
.peers
.get_mut(&peer_id)
.map(|state| Self::expire_state(state, now))
.unwrap_or_default();
self.address_count -= expired.len();
if self
.peers
.get(&peer_id)
.is_some_and(|state| state.addresses.is_empty())
{
self.peers.remove(&peer_id);
}
expired
.into_iter()
.map(|address| (peer_id, address))
.collect()
}
fn expire_state(state: &mut IdentifyPeerAddresses, now: Instant) -> Vec<Multiaddr> {
let mut expired = Vec::new();
let mut retained = VecDeque::with_capacity(state.addresses.len());
while let Some(address) = state.addresses.pop_front() {
if address.expires_at <= now {
expired.push(address.address);
} else {
retained.push_back(address);
}
}
state.addresses = retained;
expired
}
}
/// Combined libp2p behaviour.
#[derive(NetworkBehaviour)]
#[behaviour(prelude = "libp2p_swarm::derive_prelude")]
@@ -4165,14 +3960,6 @@ pub async fn start(
// Track in-flight GetRecord queries → reply channels
let mut pending_get_did: HashMap<kad::QueryId, oneshot::Sender<Option<DidRecord>>> =
HashMap::new();
// Keep Identify-derived entries across disconnects so a bootstrap address
// learned through Identify remains usable for redial. The TTL cleanup below
// removes entries that are no longer refreshed without touching explicit
// AddKnownPeer addresses.
let mut identify_addresses = IdentifyAddressBook::default();
let mut explicit_addresses: HashMap<PeerId, HashSet<Multiaddr>> = HashMap::new();
let mut identify_cleanup = tokio::time::interval(IDENTIFY_ADDRESS_CLEANUP_INTERVAL);
identify_cleanup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
// Start the event loop as a background task
tokio::spawn(async move {
@@ -4192,13 +3979,6 @@ pub async fn start(
_ = ingest_sweep.tick() => {
run_ingest_sweep(&ingest_limiters, &replay_guard, ingest_now()).await;
}
_ = identify_cleanup.tick() => {
for (peer_id, address) in identify_addresses.expire(Instant::now()) {
if !is_explicit_address(&explicit_addresses, &peer_id, &address) {
swarm.behaviour_mut().kademlia.remove_address(&peer_id, &address);
}
}
}
// Graceful shutdown: exit the swarm loop when the
// process-wide signal flips. This drops the Swarm
// which closes all libp2p connections cleanly.
@@ -4333,21 +4113,7 @@ pub async fn start(
identify::Event::Received { peer_id, info, .. }
)) => {
debug!(peer = %peer_id, "identify received");
let changes = identify_addresses.update(
peer_id,
Instant::now(),
&info.listen_addrs,
);
for (removed_peer, addr) in changes.removed {
if !is_explicit_address(&explicit_addresses, &removed_peer, &addr)
{
swarm
.behaviour_mut()
.kademlia
.remove_address(&removed_peer, &addr);
}
}
for addr in changes.refreshed.into_iter().chain(changes.added) {
for addr in info.listen_addrs {
swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
}
}
@@ -4384,14 +4150,6 @@ pub async fn start(
}
}
P2pCommand::AddKnownPeer { peer_id, addr } => {
let Ok(addr) = addr.clone().with_p2p(peer_id) else {
warn!(%peer_id, %addr, "dropping known-peer address with a foreign peer suffix");
continue;
};
explicit_addresses
.entry(peer_id)
.or_default()
.insert(addr.clone());
swarm.behaviour_mut().kademlia.add_address(&peer_id, addr);
}
P2pCommand::Dial(addr) => {
@@ -4443,453 +4201,6 @@ pub async fn start(
mod tests {
use super::*;
fn identify_address(peer_id: PeerId, octet: u8, port: u16) -> Multiaddr {
format!("/ip4/192.0.2.{octet}/udp/{port}/quic-v1")
.parse::<Multiaddr>()
.unwrap()
.with_p2p(peer_id)
.unwrap()
}
fn identify_address_without_peer(octet: u8, port: u16) -> Multiaddr {
format!("/ip4/192.0.2.{octet}/udp/{port}/quic-v1")
.parse::<Multiaddr>()
.unwrap()
}
#[test]
fn identify_empty_reports_do_not_retain_peer_entries() {
let now = Instant::now();
let mut book = IdentifyAddressBook::default();
let foreign = identify_address(PeerId::random(), 1, 10_000);
for _ in 0..IDENTIFY_GLOBAL_ADDRESS_LIMIT + 1 {
let peer = PeerId::random();
assert_eq!(
book.update(peer, now, &[]),
IdentifyAddressChanges::default()
);
assert_eq!(
book.update(peer, now, std::slice::from_ref(&foreign)),
IdentifyAddressChanges::default()
);
}
assert!(book.peers.is_empty());
assert_eq!(book.address_count, 0);
}
#[test]
fn identify_fully_rejected_peer_leaves_no_row_at_full_budget() {
let now = Instant::now();
let mut book = IdentifyAddressBook::default();
// Fill the global budget with one-address peers.
for index in 0..IDENTIFY_GLOBAL_ADDRESS_LIMIT {
let peer = PeerId::random();
let address = identify_address(peer, 1, 10_000 + index as u16);
book.update(peer, now, std::slice::from_ref(&address));
}
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
// A fresh peer reporting at saturation has every candidate rejected
// by the global-budget check. The row created by or_insert_with must
// be dropped: a zero-address row is invisible to address_count and
// would otherwise linger until the expiry tick.
let fresh = PeerId::random();
let fresh_address = identify_address(fresh, 2, 30_000);
let changes = book.update(fresh, now, std::slice::from_ref(&fresh_address));
assert!(changes.added.is_empty());
assert!(!book.peers.contains_key(&fresh));
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
}
#[test]
fn identify_update_reports_expired_addresses_as_removed() {
let peer = PeerId::random();
let now = Instant::now();
let address = identify_address(peer, 1, 10_000);
let mut book = IdentifyAddressBook::default();
book.update(peer, now, std::slice::from_ref(&address));
// An update past the TTL expires the stale address through
// expire_peer and reports it in changes.removed so the Identify
// handler can drop it from Kademlia.
let changes = book.update(
peer,
now + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1),
std::slice::from_ref(&address),
);
assert_eq!(changes.removed, vec![(peer, address.clone())]);
assert_eq!(changes.added, vec![address]);
}
#[test]
fn identify_expire_drops_rows_emptied_by_expiry() {
let peer = PeerId::random();
let now = Instant::now();
let address = identify_address(peer, 1, 10_000);
let mut book = IdentifyAddressBook::default();
book.update(peer, now, std::slice::from_ref(&address));
assert!(book.peers.contains_key(&peer));
// Once every address in the row has expired, expire() must drop the
// row itself, matching the empty-row drop in update().
let removed = book.expire(now + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1));
assert_eq!(removed, vec![(peer, address)]);
assert!(!book.peers.contains_key(&peer));
assert_eq!(book.address_count, 0);
}
#[test]
fn identify_report_processing_stops_at_the_input_limit() {
let peer = PeerId::random();
let now = Instant::now();
let foreign = identify_address(PeerId::random(), 1, 10_000);
let accepted = identify_address(peer, 2, 10_001);
let ignored = identify_address(peer, 3, 10_002);
let mut report = vec![foreign; IDENTIFY_REPORT_ADDRESS_LIMIT - 1];
report.push(accepted.clone());
report.push(ignored);
let mut book = IdentifyAddressBook::default();
let changes = book.update(peer, now, &report);
assert_eq!(changes.added, vec![accepted]);
assert!(changes.removed.is_empty());
assert_eq!(book.address_count, 1);
}
#[test]
fn identify_addresses_grow_without_eviction_below_the_peer_cap() {
let peer = PeerId::random();
let now = Instant::now();
let first = identify_address(peer, 1, 10_000);
let second = identify_address(peer, 2, 10_001);
let mut book = IdentifyAddressBook::default();
book.update(peer, now, std::slice::from_ref(&first));
let changes = book.update(
peer,
now + IDENTIFY_ADDRESS_WINDOW,
std::slice::from_ref(&second),
);
assert_eq!(changes.added, vec![second]);
assert!(changes.removed.is_empty());
assert_eq!(book.peers[&peer].addresses.len(), 2);
assert_eq!(book.address_count, 2);
}
#[test]
fn identify_global_budget_rejects_new_admissions() {
let now = Instant::now();
let mut book = IdentifyAddressBook::default();
let peers = (0..IDENTIFY_GLOBAL_ADDRESS_LIMIT)
.map(|index| {
let peer = PeerId::random();
let address = identify_address(peer, 1, 10_000 + index as u16);
book.update(peer, now, std::slice::from_ref(&address));
(peer, address)
})
.collect::<Vec<_>>();
let (peer, refreshed) = &peers[0];
let new_address = identify_address(*peer, 2, 30_000);
let changes = book.update(
*peer,
now + IDENTIFY_ADDRESS_WINDOW,
&[new_address.clone(), refreshed.clone()],
);
// The budget is full: the new address is rejected rather than
// evicting another peer's entry, and the refreshed address keeps its
// slot and is reported for re-admission to Kademlia.
assert!(changes.added.is_empty());
assert!(changes.removed.is_empty());
assert_eq!(changes.refreshed, vec![(*refreshed).clone()]);
assert_eq!(book.peers.len(), IDENTIFY_GLOBAL_ADDRESS_LIMIT);
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
assert_eq!(
book.peers[peer]
.addresses
.iter()
.map(|a| &a.address)
.collect::<Vec<_>>(),
vec![refreshed]
);
}
#[test]
fn identify_per_peer_eviction_applies_at_full_global_budget() {
let now = Instant::now();
let mut book = IdentifyAddressBook::default();
// One filler peer goes in before the victim so the evicted address
// is not also the globally-oldest insertion. If saturation evicted
// the globally-oldest entry instead of the victim's own oldest,
// the sentinel's address would be the one removed.
let sentinel = PeerId::random();
let sentinel_address = identify_address(sentinel, 200, 19_999);
book.update(sentinel, now, std::slice::from_ref(&sentinel_address));
// The victim peer holds a full per-peer row while the rest of the
// global budget is filled by one-address peers.
let victim = PeerId::random();
let victim_addresses: Vec<Multiaddr> = (0..IDENTIFY_ADDRESS_LIMIT)
.map(|index| identify_address(victim, index as u8 + 1, 10_000 + index as u16))
.collect();
book.update(victim, now, &victim_addresses);
for index in 0..IDENTIFY_GLOBAL_ADDRESS_LIMIT - IDENTIFY_ADDRESS_LIMIT - 1 {
let peer = PeerId::random();
let address = identify_address(peer, 200, 20_000 + index as u16);
book.update(peer, now, &[address]);
}
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
// The victim reports a ninth address after the rate window rolls over.
// Per-peer eviction must still apply at global saturation: the oldest
// non-reported address is evicted to make room, and the global count
// is unchanged. This ordering only holds because eviction runs before
// the global-budget check; hoisting that check above the eviction
// block makes this test fail.
let ninth = identify_address(victim, 100, 30_000);
let changes = book.update(
victim,
now + IDENTIFY_ADDRESS_WINDOW,
std::slice::from_ref(&ninth),
);
assert_eq!(changes.added, vec![ninth]);
assert_eq!(changes.removed, vec![(victim, victim_addresses[0].clone())]);
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
assert_eq!(book.peers[&victim].addresses.len(), IDENTIFY_ADDRESS_LIMIT);
// The sentinel was inserted first; per-peer eviction must leave it
// alone. A global-oldest eviction policy would have removed it.
assert_eq!(
book.peers[&sentinel]
.addresses
.iter()
.map(|a| &a.address)
.collect::<Vec<_>>(),
vec![&sentinel_address]
);
}
#[test]
fn explicit_address_helper_detects_retained_addresses() {
let peer = PeerId::random();
let other = PeerId::random();
let address = identify_address(peer, 1, 10_000);
let mut explicit: HashMap<PeerId, HashSet<Multiaddr>> = HashMap::new();
// Unknown peer: nothing is explicit.
assert!(!is_explicit_address(&explicit, &peer, &address));
explicit.entry(peer).or_default().insert(address.clone());
assert!(is_explicit_address(&explicit, &peer, &address));
// The same address under a different peer is not retained.
assert!(!is_explicit_address(&explicit, &other, &address));
// A different address for the same peer is not retained.
assert!(!is_explicit_address(
&explicit,
&peer,
&identify_address(peer, 2, 10_001)
));
}
#[test]
fn identify_addresses_are_capped_with_fifo_eviction() {
let peer_id = PeerId::random();
let now = Instant::now();
let initial = (0..IDENTIFY_ADDRESS_LIMIT)
.map(|index| identify_address(peer_id, index as u8 + 1, 10_000 + index as u16))
.collect::<Vec<_>>();
let replacement = identify_address(peer_id, 200, 20_000);
let mut book = IdentifyAddressBook::default();
let first = book.update(peer_id, now, &initial);
assert_eq!(first.added, initial);
assert!(first.removed.is_empty());
let second = book.update(
peer_id,
now + IDENTIFY_ADDRESS_WINDOW,
std::slice::from_ref(&replacement),
);
assert_eq!(second.added, vec![replacement]);
assert_eq!(second.removed, vec![(peer_id, initial[0].clone())]);
}
#[test]
fn identify_oversized_reports_keep_the_same_refreshed_subset() {
let peer_id = PeerId::random();
let now = Instant::now();
let mut report = (0..IDENTIFY_ADDRESS_LIMIT + 2)
.map(|index| identify_address(peer_id, index as u8 + 1, 10_000 + index as u16))
.collect::<Vec<_>>();
let mut book = IdentifyAddressBook::default();
let first = book.update(peer_id, now, &report);
assert_eq!(first.added.len(), IDENTIFY_ADDRESS_LIMIT);
assert_eq!(first.removed.len(), 0);
let retained = first.added.clone();
report.reverse();
let second = book.update(peer_id, now + IDENTIFY_ADDRESS_WINDOW, &report);
assert!(second.added.is_empty());
assert!(second.removed.is_empty());
let current = book
.peers
.get(&peer_id)
.unwrap()
.addresses
.iter()
.map(|entry| entry.address.clone())
.collect::<Vec<_>>();
assert_eq!(current, retained);
}
#[test]
fn identify_addresses_rate_limit_new_entries_across_updates() {
let peer_id = PeerId::random();
let now = Instant::now();
let initial = (0..IDENTIFY_NEW_ADDRESS_LIMIT)
.map(|index| identify_address(peer_id, index as u8 + 1, 10_000 + index as u16))
.collect::<Vec<_>>();
let replacement = identify_address(peer_id, 200, 20_000);
let mut book = IdentifyAddressBook::default();
assert_eq!(book.update(peer_id, now, &initial).added, initial);
assert!(book
.update(peer_id, now, std::slice::from_ref(&replacement))
.added
.is_empty());
assert_eq!(
book.update(
peer_id,
now + IDENTIFY_ADDRESS_WINDOW,
std::slice::from_ref(&replacement),
)
.added,
vec![replacement]
);
}
#[test]
fn identify_addresses_have_a_global_budget_across_peers() {
let now = Instant::now();
let mut book = IdentifyAddressBook::default();
let mut first = None;
for index in 0..IDENTIFY_GLOBAL_ADDRESS_LIMIT {
let peer_id = PeerId::random();
let address = identify_address(peer_id, (index % 254) as u8 + 1, 10_000 + index as u16);
if first.is_none() {
first = Some((peer_id, address.clone()));
}
assert!(book.update(peer_id, now, &[address]).removed.is_empty());
}
let (peer_id, first_address) = first.unwrap();
let address = identify_address(peer_id, 250, 30_000);
let changes = book.update(
peer_id,
now + IDENTIFY_ADDRESS_WINDOW,
std::slice::from_ref(&address),
);
// The global budget is full, so the admission is rejected and the
// peer's existing entry is left alone.
assert!(changes.added.is_empty());
assert!(changes.removed.is_empty());
assert_eq!(
book.peers[&peer_id]
.addresses
.iter()
.map(|a| &a.address)
.collect::<Vec<_>>(),
vec![&first_address]
);
assert_eq!(book.address_count, IDENTIFY_GLOBAL_ADDRESS_LIMIT);
}
#[test]
fn identify_addresses_refresh_and_expire() {
let peer_id = PeerId::random();
let now = Instant::now();
let address = identify_address(peer_id, 1, 10_000);
let mut book = IdentifyAddressBook::default();
assert_eq!(
book.update(peer_id, now, std::slice::from_ref(&address))
.added,
vec![address.clone()]
);
let refreshed = now + IDENTIFY_ADDRESS_TTL - Duration::from_secs(1);
let refresh = book.update(peer_id, refreshed, std::slice::from_ref(&address));
assert!(refresh.added.is_empty());
assert!(refresh.removed.is_empty());
// Refreshed addresses are reported so the Identify handler re-adds
// them to Kademlia even though they were not newly admitted.
assert_eq!(refresh.refreshed, vec![address.clone()]);
assert!(book
.expire(now + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1))
.is_empty());
assert_eq!(
book.expire(refreshed + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1)),
vec![(peer_id, address)]
);
}
#[test]
fn identify_refresh_reports_only_addresses_in_the_report() {
let peer_id = PeerId::random();
let now = Instant::now();
let reported = identify_address(peer_id, 1, 10_000);
let unreported = identify_address(peer_id, 2, 20_000);
let mut book = IdentifyAddressBook::default();
// Admit both addresses up front so the peer holds two.
assert_eq!(
book.update(peer_id, now, &[reported.clone(), unreported.clone()])
.added
.len(),
2
);
// A later report carrying only one of the two addresses must refresh
// just that one: pins the negative direction of the refresh
// predicate, so an address the peer stopped advertising keeps its
// original TTL instead of being refreshed back onto the Kademlia
// re-admit path on every report.
let report_at = now + IDENTIFY_ADDRESS_TTL - Duration::from_secs(1);
let changes = book.update(peer_id, report_at, std::slice::from_ref(&reported));
assert!(changes.added.is_empty());
assert!(changes.removed.is_empty());
assert_eq!(changes.refreshed, vec![reported.clone()]);
// The unreported address still expires at its original TTL, while
// the refreshed address survives until its refreshed TTL.
assert_eq!(
book.expire(now + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1)),
vec![(peer_id, unreported)]
);
assert!(book.expire(report_at + Duration::from_secs(1)).is_empty());
assert_eq!(
book.expire(report_at + IDENTIFY_ADDRESS_TTL + Duration::from_secs(1)),
vec![(peer_id, reported)]
);
}
#[test]
fn identify_addresses_reject_foreign_peer_suffixes() {
let peer_id = PeerId::random();
let other_peer_id = PeerId::random();
let address = identify_address(other_peer_id, 1, 10_000);
let mut book = IdentifyAddressBook::default();
assert!(book
.update(peer_id, Instant::now(), &[address])
.added
.is_empty());
}
#[test]
fn identify_addresses_canonicalize_the_peer_suffix_once() {
let peer_id = PeerId::random();
let raw = identify_address_without_peer(1, 10_000);
let canonical = identify_address(peer_id, 1, 10_000);
let mut book = IdentifyAddressBook::default();
assert_eq!(
book.update(peer_id, Instant::now(), &[raw]).added,
vec![canonical]
);
}
#[test]
fn dir_mode_replace_authority_is_the_write_bits() {
assert!(!dir_mode_allows_untrusted_replace(0o755));
+1 -9
View File
@@ -412,15 +412,7 @@ pub fn build_router(state: AppState) -> Router {
// ── Read routes — open for public repos ───────────────────────────────
let read_routes = Router::new()
.route("/api/v1/repos", get(repos::list_repos))
.route(
"/api/v1/repos/federated",
get(repos::list_federated_repos)
.route_layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.route_layer(axum::Extension(rate_limit::IpRateLimiter {
limiter: state.federated_rate_limiter.clone(),
trust: state.push_limiter_trust,
})),
)
.route("/api/v1/repos/federated", get(repos::list_federated_repos))
.route("/api/v1/repos/{owner}/{repo}", get(repos::get_repo))
.route(
"/api/v1/repos/{owner}/{repo}/commits",
-4
View File
@@ -190,9 +190,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 federation budget: 12 requests per minute, with at most
/// 10,000 tracked source keys. Shared by router instances and swept below.
pub federated_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:
@@ -354,7 +351,6 @@ impl AppState {
self.ipfs_work_rate_limiter.cleanup().await;
self.sync_trigger_rate_limiter.cleanup().await;
self.peer_write_rate_limiter.cleanup().await;
self.federated_rate_limiter.cleanup().await;
}
/// Trigger graceful shutdown. Idempotent — calling more than once
-785
View File
@@ -116,7 +116,6 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
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)),
federated_rate_limiter: RateLimiter::new_bounded(12, Duration::from_secs(60), 10_000),
shutdown_tx: tokio::sync::watch::channel(false).0,
// Generous — no test drives the handler-level shed (git_permit is unit-tested).
git_read_semaphore: Arc::new(tokio::sync::Semaphore::new(64)),
@@ -1037,345 +1036,6 @@ mod tests {
);
}
/// #496: create_task must reject a caller-supplied repo_id that names a
/// hosted repo the caller does not own. A stranger naming the victim's repo
/// gets 403, leaking nothing about the repo and persisting nothing; the
/// owner files (201). Repo-less tasks stay allowed; an id naming no hosted
/// repo (unknown, or quarantined, hidden as if it did not exist) is stored
/// verbatim as an opaque label, and the 403-vs-201 split between a hosted
/// id and an unknown one is itself a one-bit oracle. This is a write gate:
/// task reads are caller-independent and not repo-scoped on this base.
#[sqlx::test]
async fn create_task_rejects_foreign_repo_id(pool: PgPool) {
let owner = "did:key:zTASKREPOOWNERAAAAAAAAAAAAAAAAAAAAAAAAAA";
let stranger = "did:key:zTASKREPOSTRANGERBBBBBBBBBBBBBBBBBBBBBB";
let state = test_state(pool).await;
let repo = seed_repo(owner, "task-gate-repo");
state.db.create_repo(&repo).await.expect("seed repo");
let router = || {
Router::new()
.route(
"/api/v1/tasks",
axum::routing::post(crate::api::tasks::create_task),
)
.with_state(state.clone())
};
let post_as = |signer: &str, body: String| {
router().oneshot(signed_request_as(
signer,
Method::POST,
"/api/v1/tasks",
Body::from(body),
))
};
let body_with = |signer: &str, repo_id: Option<&str>| {
let repo_field = repo_id
.map(|id| format!(r#","repo_id":"{id}""#))
.unwrap_or_default();
format!(
r#"{{"kind":"build","capability":"repo:write","delegator_did":"{signer}"{repo_field}}}"#
)
};
let created_id = |bytes: &[u8]| -> String {
let v: serde_json::Value = serde_json::from_slice(bytes).expect("task json");
v["id"].as_str().expect("task id").to_string()
};
// Stranger filing under the victim's repo id → exact 403.
let resp = post_as(stranger, body_with(stranger, Some(&repo.id)))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::FORBIDDEN,
"a non-owner must not file tasks against a foreign repo id"
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&bytes);
assert!(
text.contains(r#""error":"forbidden""#),
"denial must keep the forbidden envelope: {text}"
);
assert!(
text.contains(crate::api::REPO_OWNER_ONLY_MESSAGE),
"denial must carry the shared owner-denial text (same as GraphQL): {text}"
);
// The denial must not persist: no task row may carry the victim's
// repo id afterwards — a 403 that still writes is the injection.
let stored = state
.db
.list_tasks(None, None, 50)
.await
.expect("list tasks");
assert!(
stored
.iter()
.all(|t| t.repo_id.as_deref() != Some(repo.id.as_str())),
"a denied write must not leave a task under the foreign repo id"
);
// Owner filing under their own repo id → 201.
let resp = post_as(owner, body_with(owner, Some(&repo.id)))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::CREATED,
"the owner must be able to file tasks against their repo"
);
// Repo-less task → 201 (unaffected).
let resp = post_as(stranger, body_with(stranger, None)).await.unwrap();
assert_eq!(
resp.status(),
StatusCode::CREATED,
"repo-less task creation must keep working"
);
// Unknown repo id → 201 as an opaque label.
let resp = post_as(stranger, body_with(stranger, Some("no-such-repo-id")))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::CREATED,
"an id naming no hosted repo stays an opaque label"
);
// Read back the row: the unresolved id must be stored verbatim, not
// dropped or rewritten to a repo it never named.
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let task = state
.db
.get_task(&created_id(&bytes))
.await
.expect("get task")
.expect("task must be stored");
assert_eq!(
task.repo_id.as_deref(),
Some("no-such-repo-id"),
"an unresolved id must be stored verbatim as a label"
);
// Quarantined repo id → treated as unknown (201), never a 403 that
// would confirm the id names a real repo.
state
.db
.set_repo_quarantine(&repo.id, true)
.await
.expect("quarantine");
let resp = post_as(stranger, body_with(stranger, Some(&repo.id)))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::CREATED,
"a quarantined repo id must not 403 (existence oracle)"
);
// Read back the row: the quarantined id is stored verbatim as a label
// too, so the accepted write is not silently dropped.
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let task = state
.db
.get_task(&created_id(&bytes))
.await
.expect("get task")
.expect("task must be stored");
assert_eq!(
task.repo_id.as_deref(),
Some(repo.id.as_str()),
"a quarantined id must be stored verbatim as a label"
);
}
/// #496 (REST): a repo-lookup fault in the create_task gate must render the
/// module's opaque `db_error` envelope, not the raw sqlx text. Dropping a
/// column `get_repo_by_id` selects faults that first lookup; the 500 body
/// stays fixed and names no column (#250/#226).
#[sqlx::test]
async fn create_task_gate_repo_lookup_fault_is_opaque(pool: PgPool) {
let owner = "did:key:zTASKFAULTOWNERAAAAAAAAAAAAAAAAAAAAAAAAA";
let state = test_state(pool.clone()).await;
let repo = seed_repo(owner, "task-gate-fault-repo");
state.db.create_repo(&repo).await.expect("seed repo");
let router = || {
Router::new()
.route(
"/api/v1/tasks",
axum::routing::post(crate::api::tasks::create_task),
)
.with_state(state.clone())
};
let body = format!(
r#"{{"kind":"build","capability":"repo:write","delegator_did":"{owner}","repo_id":"{}"}}"#,
repo.id
);
// Drop a column the lookup's SELECT names so get_repo_by_id faults.
sqlx::query("ALTER TABLE repos DROP COLUMN machine_id")
.execute(&pool)
.await
.expect("drop machine_id");
let resp = router()
.oneshot(signed_request_as(
owner,
Method::POST,
"/api/v1/tasks",
Body::from(body),
))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::INTERNAL_SERVER_ERROR,
"a faulted repo lookup must be a 500"
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&bytes);
assert!(
text.contains(r#""error":"db_error""#),
"gate DB fault must keep the opaque envelope: {text}"
);
assert!(
text.contains(crate::error::DB_ERROR_MESSAGE),
"gate DB fault must carry the fixed message: {text}"
);
assert!(
!text.contains("machine_id") && !text.contains("column"),
"schema text leaked through the gate: {text}"
);
}
/// #496 (REST): the gate's second lookup (`is_repo_quarantined`) runs through
/// the same opaque helper. Dropping the column only it selects leaves
/// `get_repo_by_id` succeeding, so this pins that second call site too.
#[sqlx::test]
async fn create_task_gate_quarantine_lookup_fault_is_opaque(pool: PgPool) {
let owner = "did:key:zTASKFAULT2OWNERAAAAAAAAAAAAAAAAAAAAAAAA";
let state = test_state(pool.clone()).await;
let repo = seed_repo(owner, "task-gate-fault2-repo");
state.db.create_repo(&repo).await.expect("seed repo");
let router = || {
Router::new()
.route(
"/api/v1/tasks",
axum::routing::post(crate::api::tasks::create_task),
)
.with_state(state.clone())
};
let body = format!(
r#"{{"kind":"build","capability":"repo:write","delegator_did":"{owner}","repo_id":"{}"}}"#,
repo.id
);
// `quarantined` is not in get_repo_by_id's SELECT, so that lookup still
// succeeds and the fault lands on is_repo_quarantined.
sqlx::query("ALTER TABLE repos DROP COLUMN quarantined")
.execute(&pool)
.await
.expect("drop quarantined");
let resp = router()
.oneshot(signed_request_as(
owner,
Method::POST,
"/api/v1/tasks",
Body::from(body),
))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::INTERNAL_SERVER_ERROR,
"a faulted quarantine lookup must be a 500"
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&bytes);
assert!(
text.contains(r#""error":"db_error""#),
"quarantine DB fault must keep the opaque envelope: {text}"
);
assert!(
text.contains(crate::error::DB_ERROR_MESSAGE),
"quarantine DB fault must carry the fixed message: {text}"
);
assert!(
!text.contains("quarantined") && !text.contains("column"),
"schema text leaked through the gate: {text}"
);
}
/// #496 (REST): the create_task INSERT goes through the same opaque helper,
/// so a write fault must not serialize sqlx text either. A repo-less task
/// reaches the INSERT without a gate lookup; dropping a column the INSERT
/// names faults it.
#[sqlx::test]
async fn create_task_insert_fault_is_opaque(pool: PgPool) {
let signer = "did:key:zTASKINSERTFAULTAAAAAAAAAAAAAAAAAAAAAAAA";
let state = test_state(pool.clone()).await;
let router = || {
Router::new()
.route(
"/api/v1/tasks",
axum::routing::post(crate::api::tasks::create_task),
)
.with_state(state.clone())
};
let body =
format!(r#"{{"kind":"build","capability":"repo:write","delegator_did":"{signer}"}}"#);
sqlx::query("ALTER TABLE agent_tasks DROP COLUMN status")
.execute(&pool)
.await
.expect("drop status");
let resp = router()
.oneshot(signed_request_as(
signer,
Method::POST,
"/api/v1/tasks",
Body::from(body),
))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::INTERNAL_SERVER_ERROR,
"a faulted INSERT must be a 500"
);
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&bytes);
assert!(
text.contains(r#""error":"db_error""#),
"INSERT fault must keep the opaque envelope: {text}"
);
assert!(
text.contains(crate::error::DB_ERROR_MESSAGE),
"INSERT fault must carry the fixed message: {text}"
);
assert!(
!text.contains("status") && !text.contains("column"),
"schema text leaked through the INSERT: {text}"
);
}
/// N3: get_tree gates on the REQUESTED subtree, not the repo root. A caller
/// denied a withheld subtree is rejected there (404) but passes the gate on a
/// non-withheld path (so the rejection is path-scoped, not repo-wide).
@@ -1526,451 +1186,6 @@ mod tests {
);
}
/// Pre-assigned tasks must not be stealable: a stranger claiming a pending
/// task reserved for another agent must fail, and must not receive the
/// UCAN. The reserved assignee still claims successfully.
#[sqlx::test]
async fn claim_task_honors_preassigned_assignee(pool: PgPool) {
let delegator = "did:key:zCLAIMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
let reserved = "did:key:zCLAIMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let thief = "did:key:zCLAIMTHIEFCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
let state = test_state(pool).await;
let mut task = seed_task("task-reserved", delegator);
task.assignee_did = Some(reserved.to_string());
task.ucan_token = Some("ucan-secret-for-reserved-only".into());
state.db.create_task(&task).await.expect("seed");
let router = || {
Router::new()
.route(
"/api/v1/tasks/{id}/claim",
axum::routing::post(crate::api::tasks::claim_task),
)
.with_state(state.clone())
};
let uri = "/api/v1/tasks/task-reserved/claim";
// Thief signs as themselves and asks to claim — must fail, UCAN stays put.
let resp = router()
.oneshot(signed_request_as(
thief,
Method::POST,
uri,
Body::from(format!(r#"{{"assignee_did":"{thief}"}}"#)),
))
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::FORBIDDEN,
"stranger must not steal a pre-assigned task, got {}",
resp.status()
);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&body);
assert!(
!text.contains("ucan-secret-for-reserved-only"),
"UCAN must not leak on a failed steal: {text}"
);
let still = state.db.get_task("task-reserved").await.unwrap().unwrap();
assert_eq!(still.status, "pending");
assert_eq!(still.assignee_did.as_deref(), Some(reserved));
// Reserved assignee claims successfully and receives the UCAN.
let resp = router()
.oneshot(signed_request_as(
reserved,
Method::POST,
uri,
Body::from(format!(r#"{{"assignee_did":"{reserved}"}}"#)),
))
.await
.unwrap();
assert!(
resp.status().is_success(),
"reserved assignee must claim, got {}",
resp.status()
);
let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let text = String::from_utf8_lossy(&body);
assert!(
text.contains("ucan-secret-for-reserved-only"),
"reserved assignee must receive UCAN: {text}"
);
assert!(text.contains("\"status\":\"claimed\"") || text.contains("claimed"));
}
/// Open (unassigned) pending tasks remain first-claimer-wins after the
/// pre-assignment gate.
#[sqlx::test]
async fn claim_task_open_still_first_claimer_wins(pool: PgPool) {
let delegator = "did:key:zOPENDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
let claimer = "did:key:zOPENCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let state = test_state(pool).await;
state
.db
.create_task(&seed_task("task-open", delegator))
.await
.expect("seed");
let router = Router::new()
.route(
"/api/v1/tasks/{id}/claim",
axum::routing::post(crate::api::tasks::claim_task),
)
.with_state(state.clone());
let resp = router
.oneshot(signed_request_as(
claimer,
Method::POST,
"/api/v1/tasks/task-open/claim",
Body::from(format!(r#"{{"assignee_did":"{claimer}"}}"#)),
))
.await
.unwrap();
assert!(
resp.status().is_success(),
"open task must still be claimable, got {}",
resp.status()
);
let got = state.db.get_task("task-open").await.unwrap().unwrap();
assert_eq!(got.status, "claimed");
assert_eq!(got.assignee_did.as_deref(), Some(claimer));
}
/// ASCII-blank assignees (`ASSIGNEE_BLANK`: space, tab, LF, CR) must be open
/// at the claim SQL blank branch. Seeds via raw INSERT so create_task
/// normalization is not in the path.
#[sqlx::test]
async fn claim_task_ascii_blank_reservations_are_open(pool: PgPool) {
let delegator = "did:key:zBLANKDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAA";
let claimer = "did:key:zBLANKCLAIMERBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let state = test_state(pool.clone()).await;
let now = chrono::Utc::now().to_rfc3339();
for (id, blank) in [
("task-blank-empty", ""),
("task-blank-tab", "\t"),
("task-blank-space", " "),
("task-blank-lf", "\n"),
("task-blank-cr", "\r"),
] {
sqlx::query(
"INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
)
.bind(id)
.bind(delegator)
.bind(blank)
.bind(&now)
.execute(&pool)
.await
.unwrap_or_else(|e| panic!("seed {id}: {e}"));
let claimed = state
.db
.claim_task(id, claimer)
.await
.unwrap_or_else(|e| panic!("{id} must be open: {e}"));
assert_eq!(claimed.status, "claimed", "{id}");
assert_eq!(claimed.assignee_did.as_deref(), Some(claimer), "{id}");
}
}
/// create_task must null whitespace-only assignees and return that shape
/// (fail-on-remove for the blank filter).
#[sqlx::test]
async fn create_task_normalizes_whitespace_assignee(pool: PgPool) {
let delegator = "did:key:zCREATENORMDELEGAAAAAAAAAAAAAAAAAAAAAAAA";
let state = test_state(pool).await;
let mut task = seed_task("task-create-norm", delegator);
task.assignee_did = Some(" \t\n ".into());
let stored = state.db.create_task(&task).await.expect("create");
assert!(
stored.assignee_did.is_none(),
"create_task must return normalized NULL assignee, got {:?}",
stored.assignee_did
);
let got = state
.db
.get_task("task-create-norm")
.await
.unwrap()
.unwrap();
assert!(got.assignee_did.is_none(), "row must store NULL assignee");
}
/// Claiming a reserved task must keep the stored DID form so exact-match
/// list filters still find it. Also covers bare→full claim + list by bare
/// (claimer full DID must not be required for list equality).
#[sqlx::test]
async fn claim_task_keeps_stored_assignee_did_form(pool: PgPool) {
let delegator = "did:key:zFORMDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAA";
let bare = "zFORMRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let full = format!("did:key:{bare}");
let state = test_state(pool).await;
let mut task = seed_task("task-form", delegator);
task.assignee_did = Some(bare.to_string());
state.db.create_task(&task).await.expect("seed");
let claimed = state
.db
.claim_task("task-form", &full)
.await
.expect("claim with full DID");
assert_eq!(
claimed.assignee_did.as_deref(),
Some(bare),
"reserved claim must keep the stored bare-key form"
);
let listed_bare = state
.db
.list_tasks(Some("claimed"), Some(bare), 10)
.await
.unwrap();
assert!(
listed_bare.iter().any(|t| t.id == "task-form"),
"delegator filtering by bare key must still find the task"
);
let listed_full = state
.db
.list_tasks(Some("claimed"), Some(&full), 10)
.await
.unwrap();
assert!(
!listed_full.iter().any(|t| t.id == "task-form"),
"exact list by claimer's full DID must miss the bare stored form"
);
}
/// Padded non-blank reservation must survive claim unchanged (exact form).
#[sqlx::test]
async fn claim_task_keeps_padded_reservation_exact(pool: PgPool) {
let delegator = "did:key:zPADDELEGATORAAAAAAAAAAAAAAAAAAAAAAAAAAA";
let key = "zPADRESERVEDBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
let padded = format!(" {key}\t");
let full = format!("did:key:{key}");
let state = test_state(pool.clone()).await;
let now = chrono::Utc::now().to_rfc3339();
sqlx::query(
"INSERT INTO agent_tasks (id, repo_id, kind, status, delegator_did, assignee_did, capability, ucan_token, payload, result, created_at, updated_at, deadline)
VALUES ($1,NULL,'build','pending',$2,$3,'repo:write',NULL,NULL,NULL,$4,$4,NULL)",
)
.bind("task-padded")
.bind(delegator)
.bind(&padded)
.bind(&now)
.execute(&pool)
.await
.expect("seed padded");
let claimed = state
.db
.claim_task("task-padded", &full)
.await
.expect("claim padded reservation");
assert_eq!(
claimed.assignee_did.as_deref(),
Some(padded.as_str()),
"claim must not BTRIM a non-blank reservation"
);
let listed = state
.db
.list_tasks(Some("claimed"), Some(&padded), 10)
.await
.unwrap();
assert!(
listed.iter().any(|t| t.id == "task-padded"),
"exact list by original padded form must still find the task"
);
let finished = state
.db
.finish_task("task-padded", "completed", None, &full)
.await
.expect("padded assignee must finish after trim match");
assert_eq!(finished.status, "completed");
}
/// Direct coverage for the DB-layer finish_task assignee Rust gate (handlers
/// 403 before calling in, so HTTP tests alone leave this unproven).
#[sqlx::test]
async fn finish_task_rejects_non_assignee_at_db(pool: PgPool) {
let delegator = "did:key:zFINISHDELEGATORAAAAAAAAAAAAAAAAAAAAAAAA";
let assignee = "did:key:zFINISHASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
let stranger = "did:key:zFINISHSTRANGERCCCCCCCCCCCCCCCCCCCCCCCCC";
let state = test_state(pool).await;
state
.db
.create_task(&seed_task("task-finish-gate", delegator))
.await
.expect("seed");
state
.db
.claim_task("task-finish-gate", assignee)
.await
.expect("claim");
let err = state
.db
.finish_task("task-finish-gate", "completed", None, stranger)
.await
.expect_err("stranger must not finish");
assert!(
err.to_string().contains("assignee") || err.to_string().contains("finishable"),
"expected assignee denial, got {err}"
);
let still = state
.db
.get_task("task-finish-gate")
.await
.unwrap()
.unwrap();
assert_eq!(
still.status, "claimed",
"row must stay claimed after denial"
);
let done = state
.db
.finish_task("task-finish-gate", "completed", None, assignee)
.await
.expect("assignee finishes");
assert_eq!(done.status, "completed");
}
/// Fail-on-remove for `finish_task`'s `AND assignee_did=$5`: a concurrent writer
/// that changes the stored assignee between the production read and UPDATE must
/// leave the task claimed (production `Db::finish_task` path).
#[sqlx::test]
async fn finish_task_sql_assignee_predicate_is_load_bearing(pool: PgPool) {
let delegator = "did:key:zFINISHSQLDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
let assignee = "did:key:zFINISHSQLASSIGNBBBBBBBBBBBBBBBBBBBBBBBB";
let other = "did:key:zFINISHSQLOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
let state = test_state(pool.clone()).await;
let id = "task-finish-sql";
state
.db
.create_task(&seed_task(id, delegator))
.await
.expect("seed");
state.db.claim_task(id, assignee).await.expect("claim");
let pool_bg = pool.clone();
let id_bg = id.to_string();
let other_bg = other.to_string();
let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
let hog = tokio::spawn(async move {
let mut tx = pool_bg.begin().await.expect("begin");
sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
.bind(&id_bg)
.fetch_one(&mut *tx)
.await
.expect("lock row");
locked_tx.send(()).ok();
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
sqlx::query(
"UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'claimed'",
)
.bind(&id_bg)
.bind(&other_bg)
.execute(&mut *tx)
.await
.expect("race assignee");
tx.commit().await.expect("commit");
});
locked_rx.await.expect("row lock held before finish");
let err = state
.db
.finish_task(id, "completed", None, assignee)
.await
.expect_err("assignee slot race must fail production finish_task");
hog.await.expect("hog task");
assert!(
err.to_string().contains("not in claimed state"),
"expected finish denial, got {err}"
);
let still = state.db.get_task(id).await.unwrap().unwrap();
assert_eq!(still.status, "claimed");
assert_eq!(still.assignee_did.as_deref(), Some(other));
}
/// Fail-on-remove for the claim UPDATE assignee-slot re-check: a concurrent
/// writer that changes `assignee_did` after the pre-read must block claim.
#[sqlx::test]
async fn claim_task_update_assignee_slot_recheck_is_load_bearing(pool: PgPool) {
let delegator = "did:key:zCLAIMSLOTDELEGAAAAAAAAAAAAAAAAAAAAAAAAA";
let reserved = "did:key:zCLAIMSLOTASSIGNEEBBBBBBBBBBBBBBBBBBBBBBBBB";
let other = "did:key:zCLAIMSLOTOTHERCCCCCCCCCCCCCCCCCCCCCCCCC";
let state = test_state(pool.clone()).await;
let id = "task-claim-slot";
let now = chrono::Utc::now().to_rfc3339();
sqlx::query(
"INSERT INTO agent_tasks
(id, repo_id, kind, status, delegator_did, assignee_did, capability,
ucan_token, payload, result, created_at, updated_at, deadline)
VALUES ($1, NULL, 'build', 'pending', $2, $3, 'repo:write',
NULL, NULL, NULL, $4, $4, NULL)",
)
.bind(id)
.bind(delegator)
.bind(reserved)
.bind(&now)
.execute(&pool)
.await
.expect("seed reserved");
let pool_bg = pool.clone();
let id_bg = id.to_string();
let other_bg = other.to_string();
let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
let hog = tokio::spawn(async move {
let mut tx = pool_bg.begin().await.expect("begin");
sqlx::query("SELECT assignee_did FROM agent_tasks WHERE id = $1 FOR UPDATE")
.bind(&id_bg)
.fetch_one(&mut *tx)
.await
.expect("lock row");
locked_tx.send(()).ok();
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
sqlx::query(
"UPDATE agent_tasks SET assignee_did = $2 WHERE id = $1 AND status = 'pending'",
)
.bind(&id_bg)
.bind(&other_bg)
.execute(&mut *tx)
.await
.expect("race assignee");
tx.commit().await.expect("commit");
});
locked_rx.await.expect("row lock held before claim");
let err = state
.db
.claim_task(id, reserved)
.await
.expect_err("assignee slot race must fail production claim_task");
hog.await.expect("hog task");
assert!(
err.to_string().contains("not found or already claimed"),
"expected SQL re-check denial, got {err}"
);
assert!(
err.downcast_ref::<crate::db::TaskReservedForOtherAssignee>()
.is_none(),
"must prove the UPDATE predicate, not the Rust pre-check: {err}"
);
let still = state.db.get_task(id).await.unwrap().unwrap();
assert_eq!(still.status, "pending");
assert_eq!(still.assignee_did.as_deref(), Some(other));
}
/// Adversarial-review GATE-2 (create_pr): opening a PR requires read access.
/// A non-reader is denied on a private repo before any PR is created; the
/// owner is allowed.
+36 -177
View File
@@ -3,7 +3,7 @@
//! Checks:
//! 1. Identity — ~/.gitlawb/identity.pem exists, DID parseable
//! 2. Registration — ~/.gitlawb/ucan.json exists (registered with a node)
//! 3. GITLAWB_NODE — env var points at a public node (unset is advisory)
//! 3. GITLAWB_NODE — env var is set to a non-localhost URL
//! 4. Node — GITLAWB_NODE is reachable (default: https://node.twigpine.com)
//! 5. git helper — git-remote-gitlawb is in PATH
//! 6. git — git is installed
@@ -80,6 +80,7 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
});
let mut checks = Vec::new();
let mut all_ok = true;
// ── 1. Identity ───────────────────────────────────────────────────────
let pem_path = dir.join("identity.pem");
@@ -140,16 +141,32 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
}
// ── 3. GITLAWB_NODE env var ───────────────────────────────────────────
checks.push(gitlawb_node_env_check(
std::env::var("GITLAWB_NODE").ok().as_deref(),
));
match std::env::var("GITLAWB_NODE") {
// A loopback host is a legitimate setup (self-hosted node, dev
// harness) — the connectivity check below fails loudly if it is not
// actually reachable, so don't red-flag the configuration itself.
Ok(v) if is_loopback_url(&v) => {
checks.push(Check::pass(
"GITLAWB_NODE",
format!(
"{v} (local node — intentional for self-hosting/dev; unset to target the public network)"
),
));
}
Ok(v) if !v.is_empty() => {
checks.push(Check::pass("GITLAWB_NODE", v.to_string()));
}
_ => {
checks.push(Check::fail(
"GITLAWB_NODE",
"not set — git-remote-gitlawb will fall back to http://127.0.0.1:7545",
"export GITLAWB_NODE=https://node.twigpine.com",
));
}
}
// ── 4. Node connectivity ──────────────────────────────────────────────
// GITLAWB_NODE="" binds args.node to "" through clap, and check 3 groups
// empty with unset. The probe must apply the same normalization or it
// would dial "" and Fail while the env row only Warns.
let node_url = effective_node_url(&args.node);
let client = NodeClient::new(node_url, None);
let client = NodeClient::new(&args.node, None);
match client.get("/").await {
Ok(resp) if resp.status().is_success() => {
let info = resp.json::<serde_json::Value>().await.unwrap_or_default();
@@ -158,7 +175,7 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
let short_did = did.chars().take(30).collect::<String>();
checks.push(Check::pass(
"node",
format!("{} — v{version} ({short_did}…)", node_url),
format!("{} — v{version} ({short_did}…)", args.node),
));
// Capability drift: a node newer than this CLI may require features
// (RFC 9421 signing, iCaptcha) the CLI doesn't speak.
@@ -177,14 +194,14 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
Ok(resp) => {
checks.push(Check::fail(
"node",
format!("{} returned HTTP {}", node_url, resp.status()),
format!("{} returned HTTP {}", args.node, resp.status()),
"check GITLAWB_NODE env var or try gl register --node <url>",
));
}
Err(e) => {
checks.push(Check::fail(
"node",
format!("{} unreachable: {e}", node_url),
format!("{} unreachable: {e}", args.node),
"check your internet connection or set GITLAWB_NODE",
));
}
@@ -268,21 +285,23 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
CheckState::Fail => "✗",
};
println!(" {icon} {:<24} {}", check.label, check.detail);
if matches!(check.state, CheckState::Fail) {
all_ok = false;
}
}
println!();
let has_failures_now = has_failures(&checks);
let has_issues = checks
.iter()
.any(|c| matches!(c.state, CheckState::Fail | CheckState::Warn));
if !has_issues {
println!("Everything looks good. Run `gl quickstart` to create your first repo.");
} else {
if has_failures_now {
println!("Some checks failed. Suggested fixes:");
} else {
if all_ok {
println!("Setup looks good with some warnings:");
} else {
println!("Some checks failed. Suggested fixes:");
}
for check in &checks {
if matches!(check.state, CheckState::Fail | CheckState::Warn) {
@@ -292,75 +311,14 @@ pub async fn run(args: DoctorArgs) -> Result<()> {
}
}
println!();
if has_failures_now {
if !all_ok {
println!("Run `gl quickstart` for a guided setup.");
}
}
// #357: a non-zero exit is the only signal scripting and CI have. Warn-only
// rows stay exit 0 (iCaptcha offline, version drift, shell-alias shadowing
// are all non-fatal); Fail-class rows (identity, registration, node
// reachable, git-remote-gitlawb present, git present) flip to exit 1.
// Direct `process::exit` rather than returning Err: the prose above is the
// user-facing summary, and an anyhow Error frame would just duplicate it.
if has_failures(&checks) {
std::process::exit(1);
}
Ok(())
}
/// Classify the `GITLAWB_NODE` env var for `gl doctor`. Extracted so unit
/// tests can pin the Warn tier directly (#357): an unset env is advisory, not
/// a failure, because `--node` defaults to `https://node.twigpine.com` and the
/// CLI works without it. Flipping this back to `Check::fail` must break a
/// test, not just change prose.
fn gitlawb_node_env_check(env_val: Option<&str>) -> Check {
match env_val {
// A loopback host is a legitimate setup (self-hosted node, dev
// harness) — the connectivity check below fails loudly if it is not
// actually reachable, so don't red-flag the configuration itself.
Some(v) if is_loopback_url(v) => Check::pass(
"GITLAWB_NODE",
format!(
"{v} (local node — intentional for self-hosting/dev; unset to target the public network)"
),
),
Some(v) if !v.is_empty() => Check::pass("GITLAWB_NODE", v.to_string()),
// `--node` defaults to `https://node.twigpine.com` and the CLI works
// fine without the variable, so an unset env is advisory, not a
// failure. Warn-class keeps the exit code 0 for a stock working
// install; this used to be Fail, which the obvious "return non-zero
// on any failure" change would have flipped to exit 1 on every
// default-config install (#357).
_ => Check::warn(
"GITLAWB_NODE",
"not set — git-remote-gitlawb will fall back to http://127.0.0.1:7545",
format!("export GITLAWB_NODE={PUBLIC_NODE}"),
),
}
}
/// The URL the connectivity probe should dial. `GITLAWB_NODE` set but empty
/// (or whitespace-only) binds `args.node` to `""` through clap rather than
/// to the `default_value`, and check 3 already groups empty with unset, so
/// the probe applies the same normalization: an empty effective URL means
/// the public node, same as unset.
fn effective_node_url(node: &str) -> &str {
if node.trim().is_empty() {
PUBLIC_NODE
} else {
node
}
}
/// True when at least one check is Fail-class. Warn-only runs (iCaptcha offline,
/// version drift, shell-alias shadowing, GITLAWB_NODE unset) keep exit 0;
/// this is the gating predicate for #357's exit-status fix.
fn has_failures(checks: &[Check]) -> bool {
checks.iter().any(|c| matches!(c.state, CheckState::Fail))
}
/// Check if a binary name exists anywhere on PATH.
/// True when the rc file contains a real `unalias` command naming `gl` —
/// not a comment, and not a longer word like `unalias global`. Ordering
@@ -686,103 +644,4 @@ mod tests {
assert!(matches!(check.state, CheckState::Ok));
assert!(check.detail.contains("GitHub API returned HTTP 403"));
}
/// #357: an all-Ok run does not flip the exit code to non-zero — a healthy
/// install must keep returning 0 so scripts like `gl doctor && gl push`
/// keep working.
#[test]
fn exit_predicate_is_false_for_all_ok() {
let checks = vec![
Check::pass("identity", "ok"),
Check::pass("registration", "ok"),
Check::pass("git", "ok"),
];
assert!(!has_failures(&checks));
}
/// #357: Warn-class rows (iCaptcha offline, version drift, shell-alias
/// shadowing, GITLAWB_NODE unset on a default-config install) are NOT
/// exit-status failures — they are still printed to the user but the
/// process exits 0. This is the rule that the previous "any
/// Fail-or-Warn row → exit 1" change would have broken for stock
/// installs that simply have not set the env var.
#[test]
fn exit_predicate_is_false_for_warn_only() {
let checks = vec![
Check::pass("identity", "ok"),
Check::warn("GITLAWB_NODE", "unset", "export GITLAWB_NODE=..."),
Check::warn("iCaptcha", "offline", "check connectivity"),
Check::warn("version", "drift", "upgrade"),
Check::warn("shell alias", "omz git plugin", "unalias gl"),
];
assert!(!has_failures(&checks));
}
/// #357: a single Fail-class row trips the exit code. The set of Fail
/// rows is whatever `run` writes through `Check::fail`: identity missing
/// or unparseable, registration missing/malformed, node unreachable,
/// git-remote-gitlawb absent, git absent.
#[test]
fn exit_predicate_is_true_when_a_fail_row_is_present() {
let checks = vec![
Check::pass("registration", "ok"),
Check::fail("identity", "missing", "gl identity new"),
Check::pass("git", "ok"),
];
assert!(has_failures(&checks));
let mixed_warns = vec![
Check::pass("identity", "ok"),
Check::warn("GITLAWB_NODE", "unset", "export"),
Check::fail("node", "unreachable", "check network"),
Check::warn("iCaptcha", "offline", "check"),
];
assert!(has_failures(&mixed_warns));
}
/// #357: `GITLAWB_NODE` unset must be Warn, not Fail. This pins the
/// re-tier so flipping it back to `Check::fail` breaks the suite — the
/// previous `has_failures`-only tests built their own `Check::warn` rows
/// and could not see the re-tier at all.
#[test]
fn gitlawb_node_env_unset_is_warn() {
let check = gitlawb_node_env_check(None);
assert!(matches!(check.state, CheckState::Warn));
assert!(check.detail.contains("not set"));
let empty = gitlawb_node_env_check(Some(""));
assert!(matches!(empty.state, CheckState::Warn));
}
/// `GITLAWB_NODE=""` binds `args.node` to `""` through clap, and check 3
/// groups empty with unset. The probe must dial the same URL unset
/// would — removing the normalization makes check 4 probe `""` and Fail
/// while the env row only Warns, so an empty env install exits 1.
#[test]
fn effective_node_url_treats_empty_and_blank_as_unset() {
assert_eq!(effective_node_url(""), PUBLIC_NODE);
assert_eq!(effective_node_url(" "), PUBLIC_NODE);
assert_eq!(effective_node_url("\t\n"), PUBLIC_NODE);
assert_eq!(
effective_node_url("http://127.0.0.1:7545"),
"http://127.0.0.1:7545"
);
}
#[test]
fn gitlawb_node_env_set_is_pass() {
let check = gitlawb_node_env_check(Some("https://node.twigpine.com"));
assert!(matches!(check.state, CheckState::Ok));
assert!(check.detail.contains("https://node.twigpine.com"));
}
#[test]
fn gitlawb_node_env_loopback_is_pass() {
for url in ["http://127.0.0.1:7545", "http://localhost:7545"] {
let check = gitlawb_node_env_check(Some(url));
assert!(
matches!(check.state, CheckState::Ok),
"loopback {url} must be Ok, not Warn/Fail"
);
}
}
}
-48
View File
@@ -1,48 +0,0 @@
//! #357: pin the `gl doctor` exit-code wiring, not just the predicate.
//!
//! The unit tests for `has_failures` never touch `std::process::exit(1)` or
//! the `GITLAWB_NODE` Warn tier. Deleting the exit gate outright kept
//! `cargo test -p gl doctor::` green, so this binary probe is the only thing
//! that breaks when the wiring is gutted.
use std::process::Command;
#[test]
fn doctor_exits_nonzero_when_fail_rows_present() {
let dir = tempfile::tempdir().expect("temp dir");
// Use an unroutable node so the "node unreachable" check is Fail-class.
// The temp dir has no identity.pem / ucan.json, so identity + registration
// are also Fail — at least one Fail must flip the process to exit 1.
let bin = env!("CARGO_BIN_EXE_gl");
let output = Command::new(bin)
.args([
"doctor",
"--dir",
dir.path().to_str().unwrap(),
"--node",
"http://127.0.0.1:1",
])
// Inherit PATH but neutralize GITLAWB_NODE so the test is deterministic
// across machines that have it set (otherwise it would be Pass).
.env_remove("GITLAWB_NODE")
// iCaptcha probes a real URL by default; keep it unreachable too so
// it stays Warn and does not affect the Fail gating.
.env("GITLAWB_ICAPTCHA_URL", "http://127.0.0.1:1")
.output()
.expect("run gl doctor");
// The command should have exited 1 because at least one Fail row exists.
// If the `std::process::exit(1)` gate is removed, this becomes 0 and the
// test fails — that is the regression it pins.
assert!(
!output.status.success(),
"gl doctor must exit non-zero when Fail rows are present; stdout: {}\nstderr: {}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert_eq!(
output.status.code(),
Some(1),
"exit code must be 1, not some other non-zero"
);
}