Compare commits

..
Author SHA1 Message Date
beardthelion 20f725e25e Merge remote-tracking branch 'origin/main' into feat/commit-status-surface
# Conflicts:
#	crates/gitlawb-node/src/db/mod.rs
2026-11-12 19:18:24 -06:00
beardthelion e8952f517a fix(node): close_pr fallback head fills an empty head only (#304)
A head stored by a push between the handler's read and the UPDATE is no
longer replaced by the fallback.
2026-11-12 18:54:07 -06:00
beardthelion ddc4a2c931 fix(node): refuse re-closing a closed PR, drop 429 Retry-After, pin test git identity
- close_pr refuses a PR that is not open and the UPDATE is conditional on
  status='open', so a repeated close cannot store the branch's current tip
  on a historical pull request.
- Remove Retry-After from 429s. The status-claim caps are per-commit and
  append-only, so a 1s hint invited retries that cannot succeed.
- bare_repo_with_ref sets git author and committer env so the
  heads_still_at_tip tests pass on hosts with no global git identity (CI).
2026-11-12 17:14:01 -06:00
beardthelion 9519078316 test(node): drive the status and push-events routes in the deny sweep
The mounted-route completeness gates require every repo-scoped route to be
a driven registry row: the status read surfaces and the push-events poll
are read-gated (existence-hiding 404 for anon and signed non-readers,
owner reach twin proves the gate) and the claim write is owner-gated on a
public repo. {sha} fills as a fixed 40-hex placeholder.
2026-11-12 12:32:29 -06:00
beardthelion 862293c558 fix(node): add head_commit to test-harness PR fixture 2026-11-12 12:23:02 -06:00
beardthelion 600c358a55 fix(node): write status surfaces from proven landings in the effect executor
The stored-head and catch-up poll writes ran pre-release inside the repo
write lock, so a refused publish left rows claiming a landing the node had
deleted, and a landing reconcile proved only later emitted nothing at all.
Both are durable accounting and belong to the effect executor: the live
path reaches it only after the publish verdict, and the drain replays it
identically once reconcile proves the landing.

The ordering the in-lock placement guaranteed is kept differently: a head
write is conditioned on the branch still sitting at the landing's tip, so
a superseded landing never pins the rollup to a commit the branch moved
past. Event ids derive from (request_id, ordinal) so a recovery replay
collapses on ON CONFLICT instead of double-delivering.

Also: drop the node-global seq from the create response (id already names
the row), pin the claim body's signed format (deny_unknown_fields + v=1),
and carry Retry-After on 429.
2026-11-12 11:48:07 -06:00
beardthelion 48e88972f0 fix(node): harden status surface writers and close-path head freeze
Review-driven fixes on the merged surface:

- close_pr freezes the source head it decided against, resolving the
  recorded push for the branch (then disk) when head_commit is unset;
  a PR closed before its first status read no longer reports
  head_resolved=false forever.
- insert_status_claim_capped bounds its advisory-lock wait with
  SET LOCAL lock_timeout and retries on transient classes, matching
  the push-events writer's contract.
- record_push_events retries only transient failures (the inner bound
  already retries lock waits; a permanent error's second run is a
  guaranteed re-failure) and update_open_pr_heads retries once on
  transient and escalates to error naming the lost branch@sha rows.
- parse_ref_updates lowercases old/new SHAs so pushed values match
  the claim side's lowercase normalization.
- combined_state folds unrecognized stored states to failure, never
  silently to success.
2026-11-12 10:54:37 -06:00
beardthelion 41bbe70853 refactor(node): simplify commit-status surface after main merge
Land the merge on one mechanism per concern instead of two:

- landed_refs now intersects the declared set with the ok_set already
  derived for the outcome commit, instead of re-parsing the report in
  accepted_ref_updates; the helper's unreachable keep-declared arm (a
  trap its own doc contradicted) is gone with the second parse.
- bound_declared_refs is deleted: main's parse_ref_updates already
  refuses past MAX_REF_UPDATES inside the command loop, so the
  post-parse check was dead code whose tests pinned an abandoned
  design; the ordering test now pins the parse itself before git.
- merge_pr resolves the source head with read_ref (exact-name
  show-ref) instead of listing every ref in the repo.
- normalize_sha reuses is_hex_sha; the refs/heads/ mapping reuses
  store::local_branch_ref; test-only db methods and their SQL are
  cfg(test)-gated; duplicated body_json/status_and_bytes test
  helpers are hoisted to test_support.

Merge-artifact repair found by running the tests: scrape guards
track the renamed landed_refs binding and the reclaimed.release /
receive_pack_raw_with_reflog call names; the status migration
upgrade test rolls back to and asserts version 47 (not 24, which
main now owns); a doubled cfg(test), an orphaned doc block, and a
stale dead_code allow on StatusClaim.seq are corrected.
2026-11-12 07:19:08 -06:00
beardthelion e9e5e3d302 Merge remote-tracking branch 'origin/main' into feat/commit-status-surface 2026-11-12 00:40:22 -06:00
Kevin Codex 6b0763ac2a fix(api): make the push-event cursor opaque and bound to its repository
The cursor was the table-global bigserial, handed to clients verbatim and
accepted back as any non-negative integer, then passed straight into a
per-repo seq > $1. Two things followed from that, and both were silent.

A value issued by a DIFFERENT repository — or by this one before a
restore — is an ordinary larger number here, so the query matched
nothing, the surface answered 200 with an empty page and echoed the value
back, and the subscriber sat permanently past history it had never
received with nothing anywhere reading as an error. And because
repo_push_events is written by every repository on the node, the gaps
between one repository's cursors measured how much the others pushed in
between, private ones included.

The cursor is now an opaque token carrying a version, a binding to
(node DID, repository), and the id of the last event served. The binding
is a hash and not a MAC, because there is nothing to forge: the position
inside is a row id resolved against this repository's own rows anyway.
What it buys is that a foreign cursor is refused by shape, with a message
naming which mistake was made, instead of being applied to a repository
it was never issued for. An id that resolves to nothing — a token kept
across a restore that lost the row — is a 400 that names the recovery,
not an empty page the poller reads as being up to date.

The global sequence still orders the walk internally, which is what keeps
the page cheap, but it no longer leaves the node. The position is the row
id, which this surface already publishes in every event, so the token
discloses nothing new.

The two properties an earlier round established are kept, in their new
shape: next_cursor is never null and never moves backwards, and the start
of history is an expressible position rather than an absent one, so a
first poll of a repository with no events still hands back a token the
endpoint accepts back rather than a value a persisting poller would
misread.
2026-08-31 21:56:23 +08:00
Kevin Codex 8652b86d33 fix(api): apply a push's side effects only to the refs git accepted
receive-pack exits zero when the process ran, not when the push applied.
A non-fast-forward, a hook rejection or a failed update comes back as an
ng pkt-line inside a response this handler forwarded without reading, so
every side effect keyed on the refs parsed from the REQUEST fired for
refs that were refused: the pull request on that branch had its stored
head moved to a commit this node does not have, and the catch-up poll
surface handed subscribers a SHA that resolves to nothing.

The report is now parsed and only the refs it names with ok are used.
unpack fail accepts nothing, since the pack never landed. A report that
does not parse at all is inconclusive and keeps every declared ref: a
client that never requested report-status is told nothing about
individual refs, and reading that silence as rejection would permanently
drop the events and the head update for a push git really did accept,
which is the failure the poll surface exists to prevent.

The parser is ported from PR #384, which introduced it for this same
problem, so the two branches converge on one definition of "accepted"
rather than growing two. Whichever lands second should drop its copy. It
diverges by three lines, and they are load-bearing: git double wraps the
report when the client negotiates side-band-64k, which git push over
smart HTTP does. The observed framing from git 2.50.1 is an outer
pkt-line carrying a band byte whose payload is itself a pkt-line stream,
so one pass of strip_sideband leaves "000eunpack ok" as the first line,
the parse fails, and every ref falls back to inconclusive — the parser
answering "cannot tell" for exactly the pushes it exists to classify. A
second pass peels the inner layer and is a no-op on the single-wrapped
shape. The tests carry the real captured byte layout, both wrappings, so
this is pinned rather than assumed.

Both writes also moved inside the repository write lock. That lock orders
the ref updates of two concurrent pushes; released before these ran, the
pair was an unordered race, and B's unconditional UPDATE landing before
A's leaves the stored head — which the rollup prefers over its fallback —
pinned to a commit the branch has moved past. record_push_events has the
same inversion by a different route, since latest_push_sha_for_ref reads
the highest seq and seq order is insert order. There is no per-row
version to make the UPDATE conditional on, so the ordering has to come
from the lock. Holding it across two batched statements on an already
open pool, on a path that just ran a subprocess over the whole pack, is
the cheaper half of that trade. A source guard pins the position, because
this is a property of where the calls sit.

Separately, the 10,000-ref bound was checked only after the parser had
scanned the entire request and kept three heap strings per valid
pkt-line. The git routes raise the body limit to GITLAWB_MAX_PACK_BYTES —
2 GB by default — so a signed caller could force the whole scan and the
whole allocation and only then be told 400: the work the cap exists to
refuse, done in full before the refusal. The parser stops one past the
cap, which is still enough for bound_declared_refs to refuse on, and the
message no longer reports a count that would now be a floor rather than
the request's real one.
2026-08-31 21:56:07 +08:00
Kevin Codex a4e2b9f3f3 fix(status): bind a stored claim to the body its signature covered
A claim is provenance, so the signature stored beside it has to reach the
bytes stored beside it. It did not. The signing string covers the body
only through a content-digest line, and require_signature rebuilds the
covered values from the request: with no Content-Digest header the
covered digest is the empty string on the signing side and the verifying
side alike, the Ed25519 check passes over method, path and nothing else,
and the digest-versus-body comparison was skipped for want of a header.
An owner could sign that once and post any body under it. The row that
produced carried a valid signature, a signing string that verifies under
the owner's key, and a request_body the signing string says nothing
about — history nobody can re-verify, which is the one thing the row is
for.

The write path now recomputes the digest of the bytes it is about to
persist and requires the signing string to carry exactly that line, which
is the write-time form of the check re_verify already ran against a
stored row. Line-exact rather than a substring search: build_signing_string
emits one component per line, and a contains would also be satisfied by
the digest appearing inside another component's value.

This is deliberately not the only line of defence, and deliberately not
the middleware's. PR #306 requires the header in require_signature and
closes the same hole one layer earlier for every signed route; that
belongs there, because no other route's correctness should depend on this
handler. Neither substitutes for the other. Without #306 the empty digest
is still accepted everywhere else; without this, a later relaxation of the
middleware resumes writing unverifiable history in silence. Nothing here
touches require_signature, so the two do not collide.

The route's advertised 8 KiB body bound was also a bound on what gets
stored rather than on what gets buffered: require_signature collects and
hashes the whole body before any handler runs, so by the time the handler
measured it the allocation had happened, and concurrent requests
multiplied exactly the cost the number was written to prevent. A
RequestBodyLimitLayer now carries the same figure at the transport,
outside the auth layers and inside the rate limiters, matching the shape
the git routes already use. The handler keeps its own check: it is the one
that holds for a handler reached by any other route table.

The hand-injected test material had to become coherent with the request
it rides on, because a stand-in built over other bytes now takes every
bare-router test down the refusal branch. It is built from the body it
accompanies and carries a real content-digest line.
2026-08-31 21:55:46 +08:00
beardthelion 61033af261 fix(auth): carry the request body only where a route persists it
The signature middleware handed every signed request a second handle to
the buffered body, and only the status write reads it. The field is
optional now and populated behind a marker the persisting route group
applies.

Be accurate about what that saves, because the first reading of this
finding overstated it and the plan was corrected before implementation.
The push handler takes its body as an extractor, so axum consumes the
request and drops every extension at that point; the extra handle lived
across the remaining middleware chain, not the handler's lifetime. The
whole pack really is pinned for the whole receive-pack, but by the
handler's own parameter, which is pre-existing and untouched here.

Layer order is the failure mode worth guarding. A marker applied inside
the auth layers is never seen by the middleware that reads it, so the
body silently stops being captured while every test stays green. The
mutation that proves this leaves the layer present on the same group and
only moves it inside, and the production-router test still reddens. The
handler also refuses an absent body rather than storing an empty column,
so a lost marker fails loudly at the one place it matters.

Four comments claimed a timestamp or a uuid tiebreak decided ordering
when the code orders on the sequence. Three were named in review; the
fourth turned up next to them, calling the timestamp format load-bearing
for a cursor that no longer reads it. The mirror-dedup comments that
correctly document timestamp ordering were checked and left alone.

The shared scrape helper advanced by one byte to stop an end anchor
matching at position zero, which reports a missing anchor for any region
starting on a multibyte character. It advances by a character now. Not
reachable today since every anchor is ASCII, but five guards share this
helper and one that silently finds nothing is worse than one that fails.
2026-08-08 11:07:10 -05:00
beardthelion 5445dda3f5 fix(db): stop the push cursor skipping a concurrently committed row
The sequence is a bigserial, so Postgres allocates it at INSERT and makes
it visible at COMMIT, and nextval does not roll back. Two overlapping
writes to one repo could therefore commit out of sequence order: one
allocates, another allocates and commits, a poller reads the later row
and advances, and the earlier row then becomes visible behind a cursor
that has already passed it. Never delivered, no error, no gap signal,
which is the one thing the catch-up surface exists to prevent.

This is the second time this branch has traded one ordering hazard for
another. The cursor moved off a wall clock because stamping and
committing disagree; it moved onto a sequence where allocating and
committing disagree.

The insert now runs in an explicit transaction that takes a per-repo
advisory lock, so the lock releases at commit and no writer can allocate
until the previous row is visible. The wait is bounded, because the lock
sits on a pooled connection on the inline push path and an unbounded
wait would turn one busy repo into pool exhaustion for unrelated
requests. Three attempts at a one second timeout leaves headroom under
the pool's own acquire timeout.

That bound needs the caller to stop swallowing failures. It warned and
continued, so any error the lock introduced would silently drop those
events forever while the push had already returned success, reaching the
same outcome by another route. It now retries once and then logs at
error level naming the repo, the refs, and the SHAs that were lost.

The defect is pinned by a test that owns both transaction boundaries
rather than a barrier. A barrier cannot reproduce it: before this change
the write was a single autocommit statement, so allocate and commit were
one round trip from the client and the interleaving window was never
open to a caller of the public API.
2026-08-08 10:07:47 -05:00
beardthelion d3074a6c86 fix(status): make a replayed write idempotent and stop dropping refs
Both defects came from an independent cross-model review, and neither
was visible to the reviewers that share this session's model.

A signed write was protected only by clock skew: no nonce, no
idempotency key, no record of signatures already seen. So a captured
request replayed inside the skew window inserted a new row with a new
sequence number, and because the projection takes the latest claim per
producer and context, a stale success outranked the failure that
superseded it. The append-only design is what made this reversible
rather than merely duplicative: a replay that only duplicated a row
would be harmless, a replay that earns a fresh sequence flips the
answer. An earlier reviewer saw the replay window and concluded
append-only bounded it, which had the mechanism exactly backwards.

An exact replay is now idempotent instead of an error. The row carries a
digest over the signature, its input, and the body, a unique index
enforces it in the database rather than a check-then-insert race, and a
repeat returns the original claim. Erroring would have punished the
legitimate case this design otherwise handles worst: a client whose
response was lost retrying the same signed request and concluding its
report failed when it had succeeded.

The fan-out cap was mine, added while fixing unbounded work, and it
truncated at 256 refs after receive-pack had already accepted them. A
push whose pull request source branch sat past that point left the head
permanently stale with no push event, announced only by a log line git
had already contradicted. The writers chunk now, so nothing accepted is
dropped, and the bound that remains refuses before the accept rather
than after: chunking bounds a statement, not a request, and the work
that scales with ref count sits upstream anyway, one protection query
per ref before the service runs and one certificate and webhook per ref
after, with body limits disabled on the git routes.

Removing the cap also exposed a quadratic branch dedupe it had been
keeping cheap; that is a map now.
2026-08-08 07:21:54 -05:00
beardthelion 612d9d2dbe fix(api): order the poll cursor on a sequence and batch the push writes
The push-event cursor paged on an application-stamped wall clock, so a
row stamped later could commit earlier and a poller past that point
would never see the earlier row. A clock step backwards widens it. It
now pages on a database-assigned sequence, the same decision
status_claims already made, so the two surfaces do not disagree about
what ordering means. The rollup's branch resolve had the same flaw and
is fixed with it.

The receive-pack path issued one round trip per ref for each of two
writers, sequentially, on the user's push, with nothing bounding the ref
count. Both are single statements now and the per-push fan-out is
capped, with a warning when it truncates rather than silent loss. The
head update batches through a VALUES join, with last-write-wins dedup on
duplicate branches: git will not produce them, but a join over
duplicates picks arbitrarily where the loop it replaced was
deterministic.

The cursor was unvalidated. A malformed value produced a wrong page
instead of a refusal, limit zero returned a page with a null cursor, and
a poller persisting that cursor silently restarted from the beginning of
history. It is validated, the limit is clamped, and the returned cursor
never moves backwards. This changes the query parameter names, which is
a breaking wire change on an endpoint that has never shipped; every
consumer was checked and only the route registration reads them.

Three copies of the source-scraping parser became one shared helper.
That refactor exposed a real gap: the empty-region mutation reddened
four of the five guards, but the gossip-containment guard passed on an
empty region because it only asserts a must-not. It now asserts its scan
covered what it claims to cover, so a helper bug cannot make it
vacuously green.

status.rs is split into status/mod.rs and status/tests.rs, production
logic unchanged. Four things read the old path and all were retargeted,
including 22 mutation targets across four specs, each re-confirmed to
still match exactly once.
2026-08-08 07:21:54 -05:00
beardthelion 806f9027c0 fix(status): make the caps hold and the projection mean one identity
Five reviewers independently flagged the capped insert. It counted then
inserted inside one transaction with a docstring promising a concurrent
writer could not slip past, which READ COMMITTED does not deliver, and
every cap test was sequential so deleting the transaction would have
kept them green. A per-repo advisory lock now serializes writers on the
bound they are both testing. The new test runs eight writers through a
barrier at cap minus one: without the lock it left 4, 5, and 5 rows
against a cap of 3 on consecutive runs, so the window was real.

The projection compared producer_did as a raw column while the handler
stored whatever spelling the signer used, so one owner writing as
did:key:X and as X produced two entries for one context and the
superseded claim kept voting. Identity is now canonicalized at write
time, and the read filter is a single equality rather than a set.

That fix had a trap. Applying normalize_owner_key directly, as the
finding suggested, collapses to the bare key, which is not a parseable
DID, and would have quietly undone the re-verifiability the previous
commit established. canonical_did canonicalizes to the full did:key
form instead and is written as a function of normalize_owner_key so the
equivalence with did_matches still holds exactly, including the case
where a naive prefix-add would merge two distinct identities.

With that in place the ported copy of the DID collapse in the status
module is gone rather than relocated, and did_matches itself now
delegates, so the Rust gate, the stored column, and the SQL all trace to
one definition.

The projection also selected the signature columns it never renders,
which the previous commit made worse by adding the request body. It has
its own narrow type now, enforced by a source-read test.

The per-repo cap counted for all time with nothing pruning the table, so
a repo that reached it was permanently closed while 429 told clients to
keep retrying. It is a rolling window now, which still bounds the
fan-out a caller-chosen SHA allows, and makes the refusal true.

Also folded two count queries that scanned the same rows into one.
2026-08-08 07:21:54 -05:00
beardthelion ad71f79e6d fix(status): make a stored claim actually re-verifiable
The row stored the RFC 9421 signing string under a column called
signed_payload. That string covers the body only through the content
digest, so it proved someone signed a request with some digest but not
that this claim row was what they signed. The whole append-only argument
rests on a claim staying verifiable after the request is gone, so the
row now carries the request body as well, in a column that says so, and
a test re-verifies a claim from the row alone: body to digest, digest
into the signing string, signature over that string under the producer's
key. Its negative mutates the stored state and shows the same procedure
refuses.

Migration v24 is amended in place rather than stacked, since it has
never shipped.

The persisted material had no size bound and is caller-influenced, so
each field is now bounded. The write response returned the whole claim,
so the wire carried the stored bytes as a JSON integer array; it now
returns a client-facing type and StatusClaim no longer derives
Serialize, so the leak cannot reappear by accident.

Carrying the body as Bytes rather than a Vec matters: the middleware
runs on every signed request including receive-pack POSTs bounded at
2 GB, so copying would have doubled peak memory on every push to serve
a field only the status write reads.
2026-08-08 07:21:54 -05:00
beardthelion 90b8294c2b fix(api): resolve the rollup head from push events, not pinned CIDs
Two review findings.

The fallback read branch_cids, whose only production writer sits inside
a Pinata pin-success branch. With no Pinata configured that table is
never written, so on a default node the fallback could never resolve and
every open pull request without a stored head answered head_resolved
false forever. It now reads repo_push_events, which the receive-pack
path writes unconditionally for every ref update, with both predicates
and the ordering in SQL against the existing keyset index. Still one
database read per request, still no repository acquire.

This is better, not complete: repo_push_events only carries pushes that
land after this ships, so a pull request whose branch was last pushed
before deployment still will not resolve until it is pushed again. There
is no backfill.

The best-effort persist propagated its error, so a transient database
failure turned a rollup that had already resolved its answer into a 500.
It now catches and logs, matching the sibling writes on the push path.
The test asserts the persist genuinely fails before driving the read, so
it cannot pass vacuously.

Also corrected a doc comment that claimed the resolve branch is never
taken twice for one pull request. Only the persist is once-per-pull-
request; the resolve attempt repeats on every read until it succeeds.
The accompanying test pins the corrected wording rather than a behavior
change, since the old code counted the same way.

The resolve counter is process-global while sqlx test databases are
per-test, so unguarded tests were inflating the counts guarded tests
assert on. The guard now covers every test that triggers a resolve.
2026-08-07 18:16:53 -05:00
beardthelion db3681e48d feat(api): add the catch-up poll surface for missed push webhooks
Webhook delivery is spawned, fire-and-forget, and never retried, so a
checker whose endpoint was unreachable during a push simply never learns
about the commit. Rather than add retry machinery and an unbounded queue
pointed at an owner-supplied URL, delivery reliability becomes a
read-side property: a push records an event, and a subscriber walks
forward from a cursor to find every commit a webhook would have told it
about.

Containment is the load-bearing constraint. The row goes into
repo_push_events, written only by the receive-pack path and read only by
this repo-scoped, gate-checked handler. It must never go into
received_ref_updates, which the unauthenticated global feed also reads;
writing local pushes there would publish private-repo pushes on an
anonymous surface. A test pushes to a private repo and asserts nothing
appears on that feed, and retargeting the write reddens it.

The cursor emits a canonical Z-form timestamp with fixed sub-second
width rather than the default rfc3339. The default emits a +00:00
offset, and + decodes to a space in a query string, so a poller echoing
the cursor back re-read the same page forever. The fixed width also
makes the TEXT column's lexicographic order match time order for the
keyset predicate. Found by seeding a timestamp collision, not by
reading.

The scenario tests drive the recorder directly, so a source-scrape test
pins its call site in the push handler, the same way the stored-head
update is pinned. Removing that call was verified to turn it red.
2026-08-07 16:29:48 -05:00
beardthelion 7be9f08b3d feat(api): add the pull request head rollup
GET /api/v1/repos/{owner}/{repo}/pulls/{number}/status answers for the
pull request's head. Both read surfaces now call one extracted
projection, so "the rollup uses the same projection as the commit read"
holds by construction rather than by two implementations agreeing today.

A pull request row stores branch names, so the head has to come from
somewhere. Preference order is the stored head, then a database-backed
branch lookup for an open pull request that has none. No path in this
module acquires the repository or lists refs from disk: that would
download the whole repository from object storage on a cold node, and
this read group carries no rate limiter. A source-read test enforces
that, and asserts its own scan still covers the handler so it cannot
shrink to vacuously passing.

The read-side persist is self-limiting by SQL, not by handler ordering:
the update is conditioned on the head being absent and the pull request
open, so a push landing mid-request cannot be rolled back to a staler
tip and a closed pull request is never back-filled. Reading the same
pull request twice resolves the branch once.

That property is invisible in the response. A variant that re-resolved
on every call and then still preferred the stored head would return
byte-identical bodies across the whole suite, so the test counts the
work done rather than the results emitted, and injecting exactly that
variant reddens only the counter assertion.

An unresolvable head is not a fifth wire state. The state field stays
inside the four values and the condition rides in head_resolved beside
the pull request's own state, so the two cases differ on that boolean
alone and both are tested.
2026-08-07 15:46:00 -05:00
beardthelion 6663fed064 feat(api): serve the combined commit status behind the visibility gate
GET /api/v1/repos/{owner}/{repo}/commits/{sha}/status returns the
projection over the claim history, computed per read rather than
materialized. A materialized index gated only at write time is how this
codebase previously kept serving a repo's slug, owner, branches, and
SHAs after it was made private, so the regression test writes a claim
while public, tightens visibility, and asserts an anonymous read finds
no trace of it.

Three outcomes stay distinct, which is the point of the unit. A caller
who cannot read the repo gets the repo's own not-found, compared byte
for byte against a missing repo so the endpoint is not an existence
oracle. A commit nobody has reported on gets 200 with the explicit
pending zero-count body, asserted exactly so no client can render it as
green. A failed lookup propagates as a 500 carrying the db_error code,
never as an empty success; nothing on this path uses ok-or-default.

Latest-per-context is decided by the database-assigned sequence, not by
the timestamp and not by the uuid. The ordering test seeds the winning
row with the lexically smaller id and a far older timestamp, so it fails
under either wrong key rather than passing by luck.

The projection filters to claims authorized by the current owner, so an
ownership transfer drops the prior owner's claims from the answer while
the history keeps them. That filter is a set-membership test inside the
query, which is the constraint the delegated-capability follow-on
inherits. It expands the owner DID into its equivalent forms, and a test
runs that expansion against did_matches over a ten-case matrix so the
two cannot drift apart silently.
2026-08-07 14:53:53 -05:00
beardthelion ac31868205 feat(api): add the owner-gated commit status write path
POST /api/v1/repos/{owner}/{repo}/statuses/{sha} appends a status claim.
Authorization runs read-visibility first, then owner, and the order is
the security property: authorize_repo_read denies a quarantined repo
before the visibility gate and answers with the repo's own not-found, so
a caller who cannot read the repo cannot learn it exists. Loading the
repo and comparing the owner would answer 403 there and turn the
endpoint into an existence oracle. That case is tested directly, since
it is what separates a correct implementation from a plausible one.

Three caps, all evaluated in the insert transaction and all refusing
with 429: per producer and context, distinct contexts per commit, and
total rows per repo. The per-tuple cap alone bounds nothing, because the
context string and the SHA are both caller-chosen and the SHA is never
existence-checked. The route group also carries the per-DID and per-IP
limiters that the creation routes use; the plain write group carries
neither and this endpoint appends on every call.

Wider than the endpoint: require_signature now attaches the verified
RFC 9421 material as a request extension, so a handler can persist what
the signature actually covered. Rebuilding the canonical string in the
handler would store a string nobody verified, and none of it survives
the request. The write fails closed with a 500 when that material is
absent, rather than storing an empty payload, because a claim nobody can
re-verify is not history the later attestation substrate can adopt.

Nine injected-defect mutations confirm the gates are load-bearing rather
than decorative, including that removing the route group's merge leaves
a group that compiles, lints, and does not exist at runtime.
2026-08-07 14:16:38 -05:00
beardthelion af17831429 feat(pulls): track the source branch head on open pull requests
The rollup needs an honest target commit, and a pull request row stores
only branch names. Resolving the branch at read time is racy (the SHA
moves between the resolve and the rollup) and gives a merged or
deleted-branch pull request no answer at all, since the merge commit is
computed and never persisted.

The receive-pack path now sets head_commit for open pull requests whose
source branch matches an updated ref, one statement keyed on repo,
branch, and open status. Excluding closed and merged rows from the WHERE
clause is what freezes the value. The merge path stamps the source head
it actually consumed, which covers a push landing between the handler
reading the row and committing the merge.

A ref arriving on the push path is a full refs/heads/ name while
source_branch stores the bare branch, so the normalisation is tested
directly; a silently non-matching WHERE clause would make this a no-op
that still reads green. Branch deletions are skipped rather than stored,
since 40 zeros is a target no commit resolves to.

The scenario tests drive the update helper directly, because a real
receive-pack POST needs a pack file. That proves the mechanism and says
nothing about the sink, so a source-scrape test pins the call site in
the push handler. Removing that call was verified to turn it red.
2026-08-07 13:17:32 -05:00
beardthelion 04263dbd49 feat(db): add migration v24 for status claims, PR head, and push events
Schema for the commit-status surface, all three objects in one versioned
entry rather than three: an append-only status_claims table, a nullable
head_commit column on pull_requests, and repo_push_events for the
catch-up poll surface.

Version 24 rather than 18 because origin/pr-173 and
origin/fix/issue-135-ipfs-cid-tree-gate both claim through 23, and the
migration runner keys only on version without comparing the recorded
name, so a collision is a silent full skip rather than an error.

status_claims orders on a database-assigned bigserial seq, not the uuid
id. A uuid v4 is random, so using it as the timestamp tiebreak would let
a retried pending claim beat a success roughly half the time when both
land in the same rfc3339 tick.

Each claim stores the producer's RFC 9421 signature headers and the
canonical signed bytes. Append-only preserves rows, but only stored
signature material preserves provenance, and a claim nobody can verify
after the request is gone is not history a signed-attestation substrate
can adopt later.

The table is repo_push_events, not push_events: the latter already
exists in v1 for agent trust scoring, so a second CREATE TABLE under
IF NOT EXISTS is a silent no-op whose follow-on index then fails.

The upgrade-path test is the load-bearing one. A sqlx::test provisions
an empty database and runs the whole array, so the fresh-chain tests
stay green even when DDL is appended to an already-applied entry.
2026-08-07 12:07:35 -05:00
30 changed files with 9217 additions and 5806 deletions

No files matched your search

-18
View File
@@ -365,24 +365,6 @@ GITLAWB_RATE_LIMIT_MAX=10
# limiter. Default 3600.
GITLAWB_RATE_LIMIT_WINDOW_SECS=3600
# ── Write rate limiting (non-creation authenticated writes) ───────────────
# Max requests per client IP per hour on the shared write bucket. It is debited
# by: issue and PR comments, issue close, labels, stars, merges, protect and
# unprotect, replicas, webhook create/delete, visibility (including the GET on
# that group), tasks, bounties, profile, and every HTTP request to /graphql.
# The /graphql brake wraps the whole route, so queries and the playground GET
# consume this bucket too, not only mutations (GraphQL WebSocket subscriptions
# are excluded). Issue comment and issue close debit it inside the handler,
# after the read gate, rather than in router middleware. Issue close also draws
# on its own GITLAWB_CLOSE_ISSUE_RATE_LIMIT bucket, and a close refused by
# either cap spends nothing from the other. Separate from the creation and push
# brakes. Uses GITLAWB_TRUSTED_PROXY to resolve the client IP.
# NOTE: this is a per-IP aggregate across ALL those write actions, so behind a
# shared NAT/egress IP (or with GITLAWB_TRUSTED_PROXY unset) many users collapse
# onto one bucket — raise this for automation-heavy or multi-user single-IP
# deployments. 0 disables. Default 600.
GITLAWB_WRITE_RATE_LIMIT=600
# ── Peer-sync rate limiting (per client IP, uses GITLAWB_TRUSTED_PROXY below) ─
# /api/v1/peers/announce and /api/v1/sync/notify accept unsigned requests from
# known peers and run at higher frequency, so a generous bucket. Separate from
Generated
+6 -6
View File
@@ -2208,7 +2208,7 @@ version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.48.0",
]
[[package]]
@@ -2701,7 +2701,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7ab67060fc6b8ef687992d439ca0fa36e7ed17e9a0b16b25b601e8757df720de"
dependencies = [
"data-encoding",
"syn 2.0.117",
"syn 1.0.109",
]
[[package]]
@@ -3540,7 +3540,6 @@ dependencies = [
"async-compression",
"async-graphql",
"async-graphql-axum",
"async-trait",
"aws-config",
"aws-sdk-s3",
"axum",
@@ -4984,10 +4983,11 @@ dependencies = [
[[package]]
name = "multihash"
version = "0.19.5"
version = "0.19.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "577c63b00ad74d57e8c9aa870b5fccebf2fd64a308a5aee9f1bb88e4aea19447"
checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d"
dependencies = [
"core2",
"unsigned-varint",
]
@@ -7959,7 +7959,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.61.2",
"windows-sys 0.48.0",
]
[[package]]
-1
View File
@@ -30,7 +30,6 @@ required-features = ["test-harness"]
[dependencies]
gitlawb-core = { path = "../gitlawb-core" }
async-trait = "0.1"
ed25519-dalek = { workspace = true }
base64 = { workspace = true }
tokio = { workspace = true }
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
+2 -505
View File
@@ -164,8 +164,6 @@ pub async fn create_issue_comment(
State(state): State<AppState>,
Extension(auth): Extension<AuthenticatedDid>,
Path((owner, repo, issue_id)): Path<(String, String, String)>,
headers: axum::http::HeaderMap,
crate::rate_limit::PeerAddr(peer): crate::rate_limit::PeerAddr,
Json(req): Json<CreateIssueCommentRequest>,
) -> Result<(StatusCode, Json<IssueComment>)> {
if req.body.trim().is_empty() {
@@ -178,17 +176,6 @@ pub async fn create_issue_comment(
let (record, _rules) =
crate::api::authorize_repo_read(&state, &owner, &repo, Some(auth.0.as_str()), "/").await?;
// Shared per-IP write brake, debited AFTER the read gate so a 429 cannot
// distinguish a hidden repo from a missing one (INV-12 — the same
// placement close_issue uses for its brake). The route group carries no
// outer brake for exactly this reason.
if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
if let Some(retry_after) = state.write_rate_limiter.check_retry(&key).await {
tracing::warn!(repo = %repo, key = %key, "create_issue_comment rate limited");
return Err(AppError::RateLimited(retry_after));
}
}
let disk_path = state
.repo_store
.acquire(&record.owner_did, &record.name)
@@ -285,22 +272,7 @@ pub async fn close_issue(
// close requests for arbitrary issue ids. Applied after the read gate so a
// denied reader still sees 404, not 429.
if let Some(key) = crate::rate_limit::client_key(&headers, peer, state.push_limiter_trust) {
// Both buckets debit here, after the read gate (INV-12): the shared
// per-IP write brake — the route group deliberately carries no outer
// brake — and this route's own close budget.
//
// A close refused by either cap spends nothing from the other bucket.
// The write bucket is debited first, and refunded when the close cap
// then refuses, so the guarantee holds even when a concurrent close
// takes the last close slot between the two checks. A write-bucket
// refusal advertises its window's Retry-After; the close-cap refusal
// keeps the plain 429 it has always returned.
if let Some(retry_after) = state.write_rate_limiter.check_retry(&key).await {
tracing::warn!(repo = %repo, key = %key, "close_issue rate limited");
return Err(AppError::RateLimited(retry_after));
}
if !state.close_issue_rate_limiter.check(&key).await {
state.write_rate_limiter.refund(&key).await;
tracing::warn!(repo = %repo, key = %key, "close_issue rate limited");
return Err(AppError::TooManyRequests(
"rate limit exceeded — try again later".into(),
@@ -462,47 +434,11 @@ pub async fn close_issue(
}
}
// Snapshot the issue's pre-close JSON so a failed durable upload can roll
// the close back on local disk, mirroring create_issue. Best-effort: if the
// snapshot read fails we proceed without a rollback closure (the close still
// applies and a failed upload still surfaces as a 5xx below).
//
// Resolve first so the compensating rewrite below lands on the SAME ref the
// close mutates. The path id may be an 8-char prefix, and rewriting
// refs/gitlawb/issues/{prefix} with the pre-close content would leave the
// canonical {full_id} ref closed while the client saw a 5xx.
let full_issue_id = git_issues::resolve_issue_id(&disk_path, &issue_id)
.ok()
.flatten();
let prior_json = full_issue_id
.as_deref()
.and_then(|id| git_issues::get_issue(&disk_path, id).ok().flatten());
let close_result = git_issues::close_issue(&disk_path, &issue_id);
// Always release the advisory lock — even on error; upload to Tigris only on success.
// Same short-circuit as create_issue, and before the 200 body below. With a
// snapshot in hand, a failed upload rewrites the ref back to the pre-close
// content instead of leaving the local repo closed while the client saw 5xx.
let release_result = match (close_result.is_ok(), prior_json, full_issue_id) {
(true, Some(json), Some(resolved_id)) => {
let git_bin = state.git_bin.clone();
let deadline = std::time::Instant::now()
+ std::time::Duration::from_secs(state.config.git_service_timeout_secs);
guard
.release_compensating(true, move |path| {
// Same bounded-write strategy create_issue's compensator
// uses: the rollback rewrites the ref with the pre-close
// content through the deadline-aware runner, so a hung git
// cannot stall the guard past git_service_timeout.
git_issues::create_issue_bounded(&git_bin, path, &resolved_id, &json, deadline)
})
.await
.into_result()
}
_ => guard.release(close_result.is_ok()).await.into_result(),
};
release_result?;
// Same short-circuit as create_issue, and before the 200 body below.
guard.release(close_result.is_ok()).await.into_result()?;
let updated = close_result
.map_err(|e| AppError::Git(e.to_string()))?
@@ -1297,198 +1233,6 @@ mod lock_pool_shed_tests {
);
}
/// A close refused by its own cap must not spend the shared write budget:
/// the 429 completed no close, so stars/comments/GraphQL keep their slot.
#[sqlx::test]
async fn close_issue_refused_by_close_cap_does_not_debit_write_bucket(pool: PgPool) {
use std::net::SocketAddr;
use std::time::Duration;
let owner = "did:key:zCLOSEDUALOWNERAAAAAAAAAAAAAAAAAAAAAAA";
let mut state = crate::test_support::test_state(pool).await;
state.close_issue_rate_limiter =
crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
state.write_rate_limiter = crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state
.db
.create_repo(&seed_repo(owner, "dual-close"))
.await
.expect("seed repo");
let peer: SocketAddr = "203.0.113.77:7000".parse().unwrap();
let key = peer.ip().to_string();
assert!(
state.close_issue_rate_limiter.check(&key).await,
"exhaust only the close bucket"
);
let res = close_issue(
State(state.clone()),
Extension(AuthenticatedDid(owner.to_string())),
Path((owner.to_string(), "dual-close".to_string(), "1".to_string())),
axum::http::HeaderMap::new(),
crate::rate_limit::PeerAddr(Some(peer)),
)
.await;
assert!(
matches!(res, Err(AppError::TooManyRequests(_))),
"close cap exhausted must 429: {:?}",
res
);
assert!(
state.write_rate_limiter.check(&key).await,
"a close refused by the close cap must leave the write bucket untouched"
);
}
/// A close refused by the shared write cap must not spend the close budget.
#[sqlx::test]
async fn close_issue_refused_by_write_cap_does_not_debit_close_bucket(pool: PgPool) {
use std::net::SocketAddr;
use std::time::Duration;
let owner = "did:key:zCLOSEDUALOWNERBBBBBBBBBBBBBBBBBBBBBBB";
let mut state = crate::test_support::test_state(pool).await;
state.close_issue_rate_limiter =
crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
state.write_rate_limiter = crate::rate_limit::RateLimiter::new(1, Duration::from_secs(60));
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state
.db
.create_repo(&seed_repo(owner, "dual-write"))
.await
.expect("seed repo");
let peer: SocketAddr = "203.0.113.78:7000".parse().unwrap();
let key = peer.ip().to_string();
assert!(
state.write_rate_limiter.check(&key).await,
"exhaust only the write bucket"
);
let res = close_issue(
State(state.clone()),
Extension(AuthenticatedDid(owner.to_string())),
Path((owner.to_string(), "dual-write".to_string(), "1".to_string())),
axum::http::HeaderMap::new(),
crate::rate_limit::PeerAddr(Some(peer)),
)
.await;
let resp = res
.expect_err("write cap exhausted must refuse")
.into_response();
assert_eq!(
resp.status(),
StatusCode::TOO_MANY_REQUESTS,
"write cap exhausted must 429"
);
assert!(
state.close_issue_rate_limiter.check(&key).await,
"a close refused by the write cap must leave the close bucket untouched"
);
}
/// 429 from the shared write bucket must advertise a window-derived
/// Retry-After, as the outer-middleware 429s on the other write routes do.
fn assert_429_with_window_retry_after(err: AppError, what: &str) {
let resp = err.into_response();
assert_eq!(
resp.status(),
StatusCode::TOO_MANY_REQUESTS,
"{what}: an exhausted write bucket must 429"
);
let secs: u64 = resp
.headers()
.get("retry-after")
.unwrap_or_else(|| panic!("{what}: the write-bucket 429 must carry retry-after"))
.to_str()
.unwrap()
.parse()
.expect("retry-after must be integer seconds");
assert!(
(1..=3600).contains(&secs),
"{what}: retry-after {secs}s must fall inside the 3600s window, not a constant"
);
}
#[sqlx::test]
async fn close_issue_write_cap_429_carries_retry_after(pool: PgPool) {
use std::net::SocketAddr;
use std::time::Duration;
let owner = "did:key:zCLOSERETRYOWNERAAAAAAAAAAAAAAAAAAAAAA";
let mut state = crate::test_support::test_state(pool).await;
state.write_rate_limiter =
crate::rate_limit::RateLimiter::new(1, Duration::from_secs(3600));
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state
.db
.create_repo(&seed_repo(owner, "retry-close"))
.await
.expect("seed repo");
let peer: SocketAddr = "203.0.113.61:7000".parse().unwrap();
assert!(state.write_rate_limiter.check(&peer.ip().to_string()).await);
let res = close_issue(
State(state),
Extension(AuthenticatedDid(owner.to_string())),
Path((
owner.to_string(),
"retry-close".to_string(),
"1".to_string(),
)),
axum::http::HeaderMap::new(),
crate::rate_limit::PeerAddr(Some(peer)),
)
.await;
assert_429_with_window_retry_after(
res.expect_err("write cap exhausted must refuse"),
"close_issue",
);
}
#[sqlx::test]
async fn create_issue_comment_write_cap_429_carries_retry_after(pool: PgPool) {
use std::net::SocketAddr;
use std::time::Duration;
let owner = "did:key:zCOMMENTRETRYOWNERAAAAAAAAAAAAAAAAAAAA";
let mut state = crate::test_support::test_state(pool).await;
state.write_rate_limiter =
crate::rate_limit::RateLimiter::new(1, Duration::from_secs(3600));
state.push_limiter_trust = crate::rate_limit::TrustedProxy::None;
state
.db
.create_repo(&seed_repo(owner, "retry-comment"))
.await
.expect("seed repo");
let peer: SocketAddr = "203.0.113.62:7000".parse().unwrap();
assert!(state.write_rate_limiter.check(&peer.ip().to_string()).await);
let res = create_issue_comment(
State(state),
Extension(AuthenticatedDid(owner.to_string())),
Path((
owner.to_string(),
"retry-comment".to_string(),
"1".to_string(),
)),
axum::http::HeaderMap::new(),
crate::rate_limit::PeerAddr(Some(peer)),
Json(CreateIssueCommentRequest {
body: "x".to_string(),
}),
)
.await;
assert_429_with_window_retry_after(
res.expect_err("write cap exhausted must refuse"),
"create_issue_comment",
);
}
/// close_issue and receive-pack must not share one per-IP bucket.
#[sqlx::test]
async fn close_issue_rate_limit_does_not_drain_push_bucket(pool: PgPool) {
@@ -1576,251 +1320,4 @@ mod lock_pool_shed_tests {
server.abort();
}
/// The compensating rewrite after a refused publish must land on the ref the
/// close actually mutated. A close addressed by an 8-char prefix resolves to
/// the full id and closes `refs/gitlawb/issues/{full_id}`; a compensator
/// that rewrites the path id verbatim mints `refs/gitlawb/issues/{prefix}`
/// with the pre-close content and leaves the canonical ref closed while the
/// client saw a 5xx.
///
/// Reaching the compensator observably takes one contrivance: a definite
/// refusal invalidates (deletes) the live tree BEFORE the closure runs, so
/// the ref it writes exists only when that deletion cannot complete. The
/// fixture makes the tree read-only except the two directories the close
/// and its rollback write (`refs/gitlawb/issues/` and its reflog), and
/// pre-stores the closed blob so the mutation's `hash-object -w` needs no
/// write either. Without the contrivance the compensator runs against a
/// path that is already gone and there is nothing to assert on.
#[sqlx::test]
#[cfg(unix)]
async fn close_via_prefix_restores_the_full_ref_when_publish_refuses(pool: PgPool) {
use axum::response::IntoResponse;
use std::os::unix::fs::PermissionsExt;
// Nothing is stored under the key (404 on HEAD: no under-lock download,
// a create-only publish fence) and the PUT is refused outright (a 4xx
// classifies as provably never dispatched), so release reaches the
// definite-refusal arm that invalidates the tree and runs the
// compensator.
let app = axum::Router::new().route(
"/{*key}",
axum::routing::any(|method: axum::http::Method| async move {
if method == axum::http::Method::HEAD {
axum::http::StatusCode::NOT_FOUND.into_response()
} else {
axum::http::StatusCode::FORBIDDEN.into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let owner = "did:key:zISSUECLOSEROLLBACKBBBBBBBBBBBBBBB";
let repo_name = "rollback-close-prefix";
let mut state = crate::test_support::test_state(pool.clone()).await;
let disk = tempfile::TempDir::new().unwrap();
state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
disk.path().to_path_buf(),
crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
);
state
.db
.create_repo(&seed_repo(owner, repo_name))
.await
.expect("seed repo");
// Seed the live tree at the path the store resolves, with the issue
// under its full id so the route can be addressed by the 8-char prefix.
let live = state
.repo_store
.acquire(owner, repo_name)
.await
.expect("resolve live path");
crate::git::store::init_bare(&live).expect("init bare repo");
let full_id = "abc12345-0000-0000-0000-000000000000";
let prefix = "abc12345";
let open_json = serde_json::to_string(&IssueRecord {
id: full_id.to_string(),
title: "seeded".to_string(),
body: Some(String::new()),
author: Some(owner.to_string()),
created_at: chrono::Utc::now().to_rfc3339(),
status: "open".to_string(),
signed_payload: None,
})
.expect("serialize seeded issue");
git_issues::create_issue(&live, full_id, &open_json).expect("seed issue blob");
// Pre-store the blob `git_issues::close_issue` will write: the same
// parse/set-status/serialize sequence, so the bytes are identical and
// the mutation's `hash-object -w` is a no-op against read-only objects/.
let mut closed_value: serde_json::Value =
serde_json::from_str(&open_json).expect("parse seeded issue");
closed_value["status"] = serde_json::Value::String("closed".into());
let closed_json = serde_json::to_string(&closed_value).expect("closed issue json");
{
use std::io::Write;
let mut child = std::process::Command::new("git")
.args(["hash-object", "--stdin", "-w"])
.current_dir(&live)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::null())
.spawn()
.expect("spawn hash-object");
child
.stdin
.take()
.expect("stdin")
.write_all(closed_json.as_bytes())
.expect("write blob");
assert!(
child.wait().expect("hash-object").success(),
"pre-storing the closed blob must succeed"
);
}
// Stand the tree through the refusal's remove_dir_all: read-only
// everywhere, then write restored only where the close mutation and its
// rollback actually land (the ref and its reflog).
fn make_readonly(path: &std::path::Path) -> std::io::Result<()> {
let mut p = path.metadata()?.permissions();
p.set_readonly(true);
std::fs::set_permissions(path, p)?;
if path.is_dir() {
for entry in std::fs::read_dir(path)? {
make_readonly(&entry?.path())?;
}
}
Ok(())
}
make_readonly(&live).expect("make the live tree unremovable");
for dir in [
live.join("refs/gitlawb/issues"),
live.join("logs/refs/gitlawb/issues"),
] {
std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755))
.expect("restore write where the rollback lands");
}
let shed = close_issue(
State(state.clone()),
Extension(AuthenticatedDid(owner.to_string())),
Path((owner.to_string(), repo_name.to_string(), prefix.to_string())),
axum::http::HeaderMap::new(),
crate::rate_limit::PeerAddr(None),
)
.await;
let err = shed.expect_err("a refused publish must not report a closed issue");
assert!(
err.into_response().status().is_server_error(),
"a definite publish refusal must surface as a 5xx"
);
let restored = git_issues::get_issue(&live, full_id).expect("read the full ref");
assert_eq!(
restored.as_deref(),
Some(open_json.as_str()),
"the compensator must restore the pre-close content on the FULL ref, \
not on refs/gitlawb/issues/{prefix}"
);
assert_eq!(
git_issues::resolve_issue_id(&live, prefix).expect("resolve the prefix"),
Some(full_id.to_string()),
"no stray refs/gitlawb/issues/{{prefix}} may be minted beside the \
canonical ref: the prefix must still resolve to the full id"
);
let issues = git_issues::list_issues(&live).expect("list issues");
assert_eq!(
issues.len(),
1,
"exactly one issue ref may survive the rollback: {issues:?}"
);
// Restore writability so the TempDir can clean itself up.
let _ = std::process::Command::new("chmod")
.args(["-R", "u+w"])
.arg(&live)
.output();
server.abort();
}
/// INV-12 ordering: both issue write routes debit the shared write bucket
/// AFTER authorize_repo_read, so an exhausted bucket cannot answer 429 to a
/// caller the gate must 404 — that status difference would leak the repo's
/// existence. The route group carries no outer brake for this reason.
#[sqlx::test]
async fn an_exhausted_write_bucket_cannot_preempt_the_read_gate(pool: PgPool) {
let mut state = crate::test_support::test_state(pool.clone()).await;
state.write_rate_limiter =
crate::rate_limit::RateLimiter::new(1, std::time::Duration::from_secs(3600));
state
.db
.create_repo(&crate::db::RepoRecord {
id: "rid-write-brake-order".to_string(),
name: "priv".to_string(),
owner_did: "did:key:z6MkWriteBrakeOwner".to_string(),
description: None,
is_public: false,
default_branch: "main".to_string(),
created_at: chrono::Utc::now(),
updated_at: chrono::Utc::now(),
disk_path: "/tmp/write-brake-order".to_string(),
forked_from: None,
machine_id: None,
})
.await
.unwrap();
let headers = axum::http::HeaderMap::new();
let peer = || crate::rate_limit::PeerAddr(Some("203.0.113.77:5000".parse().unwrap()));
let key = crate::rate_limit::client_key(&headers, peer().0, state.push_limiter_trust)
.expect("a peer addr yields a brake key");
assert!(state.write_rate_limiter.check(&key).await);
assert!(
!state.write_rate_limiter.check(&key).await,
"the shared write bucket is empty for this caller"
);
let stranger = || crate::auth::AuthenticatedDid("did:key:z6MkStranger".to_string());
let path = || {
(
"did:key:z6MkWriteBrakeOwner".to_string(),
"priv".to_string(),
"1".to_string(),
)
};
let res = close_issue(
axum::extract::State(state.clone()),
axum::Extension(stranger()),
axum::extract::Path(path()),
headers.clone(),
peer(),
)
.await;
assert!(
matches!(res, Err(AppError::RepoNotFound(_))),
"close on an unreadable repo must reach the read gate, not the empty write bucket: {res:?}"
);
let res = create_issue_comment(
axum::extract::State(state),
axum::Extension(stranger()),
axum::extract::Path(path()),
headers,
peer(),
axum::Json(CreateIssueCommentRequest {
body: "x".to_string(),
}),
)
.await;
assert!(
matches!(res, Err(AppError::RepoNotFound(_))),
"comment on an unreadable repo must reach the read gate, not the empty write bucket: {res:?}"
);
}
}
+29
View File
@@ -22,6 +22,7 @@ pub mod replicas;
pub mod repos;
pub mod resolve;
pub mod stars;
pub mod status;
pub mod task_cursor;
pub mod tasks;
pub mod visibility;
@@ -68,6 +69,12 @@ pub(crate) async fn authorize_repo_read(
/// representation only within `did:key`; never let a bare id match across methods —
/// `did:web` / `did:gitlawb` share the base58 space with `did:key`, so a
/// trailing-segment compare would treat `did:key:X` and `did:gitlawb:X` as equal.
///
/// The collapse itself lives in exactly one place, [`crate::db::normalize_owner_key`],
/// which is also the function the stored `owner_did` / `authorizing_did` columns
/// are normalized through and which `OWNER_KEY_CASE_SQL` mirrors byte for byte.
/// Two identities match when they normalize to the same key; a second copy of the
/// rule here is how the Rust gate and the SQL filters would drift apart.
pub(crate) fn did_matches(a: &str, b: &str) -> bool {
// One implementation, shared with `gl` and `git-remote-gitlawb` through core:
// every boundary that compares a DID applies the same collapse rule.
@@ -205,6 +212,7 @@ mod authz_guard {
let events = include_str!("events.rs");
let tasks = include_str!("tasks.rs");
let stars = include_str!("stars.rs");
let status = include_str!("status/mod.rs");
let protect = include_str!("protect.rs");
let visibility = include_str!("visibility.rs");
let profiles = include_str!("profiles.rs");
@@ -230,6 +238,13 @@ mod authz_guard {
// (existence hiding) and require_repo_owner guards the owner half.
(webhooks, "list_webhooks", "authorize_repo_read("),
(webhooks, "list_webhooks", "require_repo_owner("),
// Same two-half shape as list_webhooks: authorize_repo_read runs
// first (a quarantined or unreadable repo gets the missing-repo
// not-found, never a 403 that would confirm existence), then
// require_repo_owner 403s a non-owner of a readable repo. Both halves
// are pinned, because dropping either changes what a stranger learns.
(status, "create_status", "authorize_repo_read("),
(status, "create_status", "require_repo_owner("),
(labels, "add_label", "authorize_repo_read("),
(labels, "add_label", "require_repo_owner("),
(labels, "remove_label", "authorize_repo_read("),
@@ -261,6 +276,19 @@ mod authz_guard {
(protect, "list_protected_branches", "authorize_repo_read("),
(labels, "list_labels", "authorize_repo_read("),
(events, "list_repo_events", "authorize_repo_read("),
// The catch-up poll surface gates before any push-event row is read,
// so a stranger cannot use it to learn that a private repo exists or
// is being pushed to.
(events, "list_repo_push_events", "authorize_repo_read("),
// The status read gates before any claim data is loaded, so its deny
// is the repo's own not-found rather than an existence oracle over
// which commits were reported on.
(status, "commit_status", "authorize_repo_read("),
// The rollup gates on the same helper before the pull request row is
// loaded, so a caller who cannot read the repo cannot learn which
// pull request numbers exist on it — and the fallback's branch
// resolve and head persist sit behind that same gate.
(status, "pull_request_status", "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
@@ -1171,6 +1199,7 @@ mod tests { async fn t() { let _ = did_matches(x, y); } }
(include_str!("replicas.rs"), "replicas.rs"),
(include_str!("repos.rs"), "repos.rs"),
(include_str!("stars.rs"), "stars.rs"),
(include_str!("status/mod.rs"), "status/mod.rs"),
(include_str!("visibility.rs"), "visibility.rs"),
(include_str!("webhooks.rs"), "webhooks.rs"),
];
+169 -131
View File
@@ -76,6 +76,7 @@ pub async fn create_pr(
status: "open".to_string(),
merged_by_did: None,
merged_at: None,
head_commit: None,
created_at: now.clone(),
updated_at: now,
};
@@ -226,17 +227,15 @@ pub async fn merge_pr(
let disk_path = guard.path().to_path_buf();
let merger_did = auth.0;
// Snapshot the target ref's pre-merge OID so a refused durable upload can
// rewind the local merge, mirroring the close_issue / create_issue
// compensating pattern. A read failure or an absent ref (a first-commit
// merge onto an unborn branch) yields no rollback closure: the merge still
// applies and a refused publish still surfaces as 5xx.
let target_ref = format!("refs/heads/{}", pr.target_branch);
let git_timeout = std::time::Duration::from_secs(state.config.git_service_timeout_secs);
let prior_tip = store::read_ref_bounded(&state.git_bin, &disk_path, &target_ref, git_timeout)
.await
.ok()
.flatten();
// The source head this merge is about to consume, read under the write lock.
// The PR row above was loaded before the lock, so a push landing in between
// already moved the branch and the row's own `head_commit` is stale; the
// value frozen on the merged PR has to be what was merged. `None` when the
// ref cannot be resolved, which leaves whatever the push path last stored.
let merged_source_head =
store::read_ref(&disk_path, &store::local_branch_ref(&pr.source_branch))
.ok()
.flatten();
let merge_result = store::merge_branch(
&disk_path,
@@ -250,31 +249,14 @@ pub async fn merge_pr(
// Short-circuit on a refused publish before the PR is marked merged and
// before the webhook fires. Both are irreversible announcements of a merge
// commit that only exists on this node's disk.
let release_result = match (merge_result.is_ok(), prior_tip) {
(true, Some(tip)) => {
let git_bin = state.git_bin.clone();
let deadline = std::time::Instant::now() + git_timeout;
guard
.release_compensating(true, move |path| {
crate::git::visibility_pack::run_bounded_git(
&git_bin,
&["update-ref", &target_ref, &tip],
path,
b"",
deadline,
)
.map(|_| ())
})
.await
.into_result()
}
_ => guard.release(merge_result.is_ok()).await.into_result(),
};
release_result?;
guard.release(merge_result.is_ok()).await.into_result()?;
let merge_sha = merge_result.map_err(|e| AppError::Git(e.to_string()))?;
state.db.merge_pr(&pr.id, &merger_did).await?;
state
.db
.merge_pr(&pr.id, &merger_did, merged_source_head.as_deref())
.await?;
let _ = state.db.touch_repo(&record.id).await;
webhooks::fire_event(
@@ -324,7 +306,36 @@ pub async fn close_pr(
));
}
state.db.close_pr(&pr.id).await?;
// Same refusal merge_pr gives, before the head is resolved: a repeated close
// must not store the branch's current tip on a historical pull request.
if pr.status != "open" {
return Err(AppError::BadRequest(format!("PR is already {}", pr.status)));
}
// Freeze the head this close decided against, the same snapshot merge_pr
// takes on its path: a closed pull request's rollup serves only the stored
// head, so a PR closed before its first status read would report
// head_resolved=false forever. The recorded push for the branch is the
// canonical answer; the disk read covers a push whose event write failed.
// An unresolvable head keeps whatever the push path last stored.
let branch_ref = store::local_branch_ref(&pr.source_branch);
let closed_head = if pr.head_commit.is_some() {
None
} else {
state
.db
.latest_push_sha_for_ref(&record.id, &branch_ref)
.await
.ok()
.flatten()
.or_else(|| {
store::read_ref(std::path::Path::new(&record.disk_path), &branch_ref)
.ok()
.flatten()
})
.map(|sha| sha.to_ascii_lowercase())
};
state.db.close_pr(&pr.id, closed_head.as_deref()).await?;
webhooks::fire_event(
std::sync::Arc::clone(&state.db),
@@ -471,131 +482,157 @@ pub async fn list_comments(
}
#[cfg(test)]
mod merge_rollback_tests {
mod tests {
use super::*;
use axum::response::IntoResponse;
use sqlx::PgPool;
use axum::body::Body;
use axum::http::{Method, StatusCode};
use axum::routing::post;
use axum::Router;
use std::process::Command;
use tower::ServiceExt;
fn git(dir: &std::path::Path, args: &[&str]) -> String {
let out = std::process::Command::new("git")
.args(["-C", &*dir.to_string_lossy()])
.args([
"-c",
"user.name=merge-rollback-test",
"-c",
"[email protected]",
])
const OWNER_DID: &str = "did:key:z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH";
const STALE_SHA: &str = "9999999999999999999999999999999999999999";
fn git(dir: &std::path::Path, args: &[&str]) {
let out = Command::new("git")
.args(args)
.current_dir(dir)
.env("GIT_AUTHOR_NAME", "t")
.env("GIT_AUTHOR_EMAIL", "[email protected]")
.env("GIT_COMMITTER_NAME", "t")
.env("GIT_COMMITTER_EMAIL", "[email protected]")
.output()
.unwrap();
.expect("git runs");
assert!(
out.status.success(),
"git {args:?}: {}",
"git {args:?} failed: {}",
String::from_utf8_lossy(&out.stderr)
);
String::from_utf8_lossy(&out.stdout).trim().to_string()
}
/// Build a bare repo whose `main` and `feature` refs diverge: one root
/// commit on main, one commit on feature, so merge_branch creates a real
/// merge commit that a refused publish must rewind.
fn seeded_merge_repo(dir: &std::path::Path) -> (String, String) {
crate::git::store::init_bare(dir).expect("bare repo");
let empty_tree = git(dir, &["mktree"]);
let root = git(dir, &["commit-tree", &empty_tree, "-m", "root"]);
git(dir, &["update-ref", "refs/heads/main", &root]);
let tip2 = git(
dir,
&["commit-tree", &empty_tree, "-m", "feature", "-p", &root],
/// A bare repo with `main` and a `feature` branch one commit ahead.
/// Returns the bare path and the real `feature` tip.
fn bare_repo_with_feature(
root: &std::path::Path,
owner_slug: &str,
) -> (std::path::PathBuf, String) {
let work = root.join("work");
std::fs::create_dir_all(&work).unwrap();
git(&work, &["init", "-q", "-b", "main", "."]);
std::fs::write(work.join("a.txt"), "one").unwrap();
git(&work, &["add", "."]);
git(&work, &["commit", "-qm", "one"]);
git(&work, &["checkout", "-qb", "feature"]);
std::fs::write(work.join("b.txt"), "two").unwrap();
git(&work, &["add", "."]);
git(&work, &["commit", "-qm", "two"]);
git(&work, &["checkout", "-q", "main"]);
let bare_dir = root.join(owner_slug);
std::fs::create_dir_all(&bare_dir).unwrap();
let bare = bare_dir.join("demo.git");
git(
root,
&[
"clone",
"-q",
"--bare",
work.to_str().unwrap(),
bare.to_str().unwrap(),
],
);
git(dir, &["update-ref", "refs/heads/feature", &tip2]);
(root, tip2)
let tip = Command::new("git")
.args(["rev-parse", "refs/heads/feature"])
.current_dir(&bare)
.output()
.unwrap();
let tip = String::from_utf8_lossy(&tip.stdout).trim().to_string();
assert_eq!(tip.len(), 40, "feature tip must resolve");
(bare, tip)
}
/// On a definite publish refusal, the local merge commit must be rewound:
/// the client got a 5xx, so the repo cannot show a merge the mirror never
/// learned about (same contract create_issue's rollback enforces).
/// Merging freezes the stored head at the source commit the merge actually
/// consumed. The PR row is loaded before the write lock is taken, so a push
/// that lands in between leaves the row's own `head_commit` stale; seeding a
/// value that matches nothing on disk stands in for that race, and the
/// merged PR must still come out pointing at the real `feature` tip.
#[sqlx::test]
async fn merge_pr_rewinds_local_merge_when_publish_refuses(pool: PgPool) {
// Endpoint that answers HEAD with an etag but refuses every write.
let app = axum::Router::new().route(
"/{*key}",
axum::routing::any(|method: axum::http::Method| async move {
if method == axum::http::Method::HEAD {
let mut resp = axum::http::StatusCode::OK.into_response();
resp.headers_mut()
.insert("etag", axum::http::HeaderValue::from_static("\"gen-1\""));
resp
} else {
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let endpoint = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
async fn merge_stamps_the_source_head_it_merged_over_a_racing_value(pool: sqlx::PgPool) {
let tmp = tempfile::tempdir().unwrap();
let owner_slug = OWNER_DID.replace([':', '/'], "_");
let (bare, feature_tip) = bare_repo_with_feature(tmp.path(), &owner_slug);
let owner = "did:key:zMERGEROLLBACKAAAAAAAAAAAAAAAAAAAAAAAA";
let repo_name = "rollback-merge";
let mut state = crate::test_support::test_state(pool.clone()).await;
let disk = tempfile::TempDir::new().unwrap();
let repo_path = disk.path().join("repo.git");
let (main_tip, _feature_tip) = seeded_merge_repo(&repo_path);
state.repo_store =
crate::git::repo_store::RepoStore::for_testing(tmp.path().to_path_buf(), pool);
let mut seeded = crate::db::RepoRecord {
id: Uuid::new_v4().to_string(),
name: repo_name.to_string(),
owner_did: owner.to_string(),
let now = Utc::now();
let repo = crate::db::RepoRecord {
id: uuid::Uuid::new_v4().to_string(),
name: "demo".into(),
owner_did: OWNER_DID.into(),
description: None,
is_public: true,
default_branch: "main".to_string(),
created_at: Utc::now(),
updated_at: Utc::now(),
disk_path: String::new(),
default_branch: "main".into(),
created_at: now,
updated_at: now,
disk_path: bare.to_string_lossy().into_owned(),
forked_from: None,
machine_id: None,
};
seeded.disk_path = repo_path.to_string_lossy().to_string();
let repo_id = seeded.id.clone();
state.db.create_repo(&seeded).await.expect("seed repo");
let now = Utc::now().to_rfc3339();
state.db.create_repo(&repo).await.unwrap();
let pr = PullRequest {
id: Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
title: "add b".into(),
body: None,
author_did: OWNER_DID.into(),
source_branch: "feature".into(),
target_branch: "main".into(),
status: "open".into(),
merged_by_did: None,
merged_at: None,
head_commit: None,
created_at: now.to_rfc3339(),
updated_at: now.to_rfc3339(),
};
state.db.create_pr(&pr).await.unwrap();
// The stale value the racing push scenario leaves on the row.
state
.db
.create_pr(&PullRequest {
id: Uuid::new_v4().to_string(),
repo_id,
number: 1,
title: "rb".to_string(),
body: None,
author_did: owner.to_string(),
source_branch: "feature".to_string(),
target_branch: "main".to_string(),
status: "open".to_string(),
merged_by_did: None,
merged_at: None,
created_at: now.clone(),
updated_at: now,
})
.set_open_pr_heads(&repo.id, "feature", STALE_SHA)
.await
.expect("seed PR");
state.repo_store = crate::git::repo_store::RepoStore::for_testing_with_tigris(
disk.path().to_path_buf(),
crate::git::repo_store::build_lock_pool(&pool, 2, std::time::Duration::from_secs(1)),
crate::git::tigris::TigrisClient::for_testing_with_endpoint("test-bucket", &endpoint),
);
.unwrap();
let res = merge_pr(
State(state),
Extension(AuthenticatedDid(owner.to_string())),
Path((owner.to_string(), repo_name.to_string(), 1)),
)
.await;
assert!(res.is_err(), "a refused publish must surface an error");
let db = state.db.clone();
let app = Router::new()
.route(
"/api/v1/repos/{owner}/{repo}/pulls/{number}/merge",
post(merge_pr),
)
.with_state(state);
let tip_after = git(&repo_path, &["rev-parse", "refs/heads/main"]);
let response = app
.oneshot(crate::test_support::signed_request_as(
OWNER_DID,
Method::POST,
"/api/v1/repos/z6MkpTHR8VNsBxYAAWHut2Geadd9jSwuBV8xRoAnwWsdvktH/demo/pulls/1/merge",
Body::empty(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let merged = db.get_pr(&repo.id, 1).await.unwrap().unwrap();
assert_eq!(merged.status, "merged");
assert_eq!(
tip_after, main_tip,
"a refused publish must rewind the local merge to the pre-merge tip"
merged.head_commit,
Some(feature_tip),
"the merge must stamp the source head it consumed, not the racing value"
);
}
}
@@ -643,6 +680,7 @@ mod lock_pool_shed_tests {
state
.db
.create_pr(&PullRequest {
head_commit: None,
id: Uuid::new_v4().to_string(),
repo_id: repo_id.clone(),
number: 1,
File diff suppressed because it is too large. Load diff
+665
View File
@@ -0,0 +1,665 @@
//! Commit status claims: the write path and the combined read.
//!
//! POST /api/v1/repos/:owner/:repo/statuses/:sha — append one claim (owner only)
//! GET /api/v1/repos/:owner/:repo/commits/:sha/status — the combined projection
//!
//! Claims are append-only: a producer reporting twice for the same context
//! leaves both rows and the visible status is a projection over the history.
//! Reporting twice means two REQUESTS, though. An exact repeat of one already
//! accepted is answered with 200 and the row it wrote, not a second row: the
//! projection elects the highest `seq`, so a replayed claim would not duplicate
//! a verdict but overturn the one that superseded it.
use axum::extract::{Extension, Path, State};
use axum::http::StatusCode;
use axum::Json;
use chrono::Utc;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::auth::AuthenticatedDid;
use crate::db::StatusClaim;
use crate::error::{AppError, Result};
use crate::state::AppState;
/// Claims for one (repo, commit, producer, context). Generous for a real
/// producer; a producer past it is misbehaving.
const MAX_CLAIMS_PER_TUPLE: i64 = 100;
/// Distinct contexts for one (repo, commit). The context string is caller-chosen
/// and free-form, so the tuple cap alone bounds nothing.
const MAX_CONTEXTS_PER_COMMIT: i64 = 50;
/// Claim rows for one repo within the trailing window the db layer counts over.
/// The commit SHA is caller-chosen and never existence-checked, so a writer at
/// the two caps above can still fan out over fresh 40-hex strings without this
/// one. It bounds the RATE of that fan-out rather than a lifetime total, because
/// nothing prunes the table: a lifetime bound would close the surface on a repo
/// permanently while answering with a status every client retries.
const MAX_CLAIMS_PER_REPO_PER_WINDOW: i64 = 10_000;
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CreateStatusRequest {
/// Payload format version. The signature covers these bytes, so the
/// field set is the signed format: `deny_unknown_fields` keeps a future
/// producer's claim from silently reading as this one, and `v` names the
/// version explicitly so a later revision has somewhere to live. Absent
/// means 1.
pub v: Option<u32>,
pub state: String,
pub context: String,
pub target_url: Option<String>,
pub description: Option<String>,
}
/// The wire state set (KTD-1): GitHub's four commit-status states, so absence
/// stays distinguishable without a fifth value no client understands.
const CLAIM_STATES: [&str; 4] = ["error", "failure", "pending", "success"];
/// Bounds on the signature material this write path persists verbatim. Every one
/// of the four is caller-influenced: the headers come off the request, the
/// signing string grows with the component list the caller's Signature-Input
/// chose, and the body is the caller's. Without a bound the claim log's row size
/// is set by whoever writes to it, so these are explicit and generous rather than
/// implicit and absent. A conforming `sign_request` produces roughly 100, 200 and
/// 400 bytes for the first three.
const MAX_SIGNATURE_CHARS: usize = 512;
const MAX_SIGNATURE_INPUT_CHARS: usize = 1024;
const MAX_SIGNING_STRING_CHARS: usize = 4096;
/// The body is a `CreateStatusRequest`, whose own fields are already capped at
/// roughly 3.3 KB in total by the three limits below; this leaves room for JSON
/// framing and nothing more.
///
/// Enforced twice, at two different moments, because the handler's check alone
/// is not the bound it advertises: `require_signature` buffers the whole body
/// before any handler runs, so by the time [`create_status`] measures it the
/// allocation has already happened. `build_router` applies the same number as a
/// `RequestBodyLimitLayer` at the transport, outside the auth layers, which is
/// what makes it a bound on what is ALLOCATED and not only on what is STORED.
/// The handler keeps its own check regardless — it is the one that holds for a
/// handler reached by any other route table, and it is what the material size
/// tests drive.
pub(crate) const MAX_REQUEST_BODY_BYTES: usize = 8192;
const MAX_CONTEXT_CHARS: usize = 255;
const MAX_TARGET_URL_CHARS: usize = 2048;
const MAX_DESCRIPTION_CHARS: usize = 1024;
/// The 201 body for an accepted claim: what the client needs to identify the row
/// it just wrote, and nothing else.
///
/// Deliberately not the stored [`StatusClaim`]. That struct carries the signature
/// material, and serializing it echoed the signature, the signing string and the
/// whole request body back on every write — the body as a JSON array of integers.
/// Write-time provenance belongs in the row, not on the wire, which is the same
/// line [`StatusEntry`] draws on the read side.
#[derive(Serialize)]
pub struct CreatedStatus {
pub id: String,
pub repo_id: String,
pub commit_sha: String,
pub state: String,
pub context: String,
pub target_url: Option<String>,
pub description: Option<String>,
pub producer_did: String,
pub created_at: String,
}
impl CreatedStatus {
fn from_claim(claim: StatusClaim) -> Self {
Self {
id: claim.id,
repo_id: claim.repo_id,
commit_sha: claim.commit_sha,
state: claim.state,
context: claim.context,
target_url: claim.target_url,
description: claim.description,
producer_did: claim.producer_did,
created_at: claim.created_at,
}
}
}
/// POST /api/v1/repos/:owner/:repo/statuses/:sha
pub async fn create_status(
State(state): State<AppState>,
Extension(auth): Extension<AuthenticatedDid>,
Path((owner, name, sha)): Path<(String, String, String)>,
material: Option<Extension<crate::auth::SignatureMaterial>>,
Json(req): Json<CreateStatusRequest>,
) -> Result<(StatusCode, Json<CreatedStatus>)> {
// Read-visibility first, then owner, and the order is the security property.
// authorize_repo_read denies a quarantined repo before the visibility gate and
// answers with the repo's own not-found, byte-identical to a missing repo, so
// a caller who cannot read the repo cannot learn it exists. Loading the repo
// and comparing the owner would answer 403 there, turning this endpoint into
// an existence oracle. require_repo_owner then 403s a non-owner of a repo the
// caller can read, where existence is not secret.
let (record, _rules) =
crate::api::authorize_repo_read(&state, &owner, &name, Some(&auth.0), "/").await?;
crate::api::require_repo_owner(&record, &auth.0)?;
let commit_sha = normalize_sha(&sha)?;
if req.v.is_some_and(|v| v != 1) {
return Err(AppError::BadRequest(
"unsupported status payload version".into(),
));
}
let claim_state = validate_state(&req.state)?;
let context = validate_context(&req.context)?;
let target_url = validate_target_url(req.target_url.as_deref())?;
let description = validate_description(req.description.as_deref())?;
// KTD-5: the signature material is captured here because the request carries
// the only copy, and `signing_string` is the exact byte sequence the
// middleware verified. Its absence means this handler was reached without
// `require_signature` — a server misconfiguration (a route group that lost
// its auth layer, a handler mounted somewhere new), not anything the client
// did, so it is a 500 and no row is written. Storing an empty payload instead
// would leave every claim from that moment on unverifiable with nothing going
// red, which is exactly the absence-renders-as-success shape.
let Some(Extension(material)) = material else {
return Err(AppError::Internal(anyhow::anyhow!(
"status claim write reached without verified signature material — \
the route is not behind require_signature"
)));
};
// The body is carried only behind the `PersistsSignedBody` marker, which
// this route's group applies outside the auth layers. Its absence means the
// marker did not reach the middleware, because the layer was dropped or
// reordered. That is the same class of misconfiguration as the branch above
// and gets the same answer. Storing an empty body instead would write claims
// that carry a signature over a digest of bytes nobody kept, which is
// exactly the case `stored_claim_re_verifies_and_a_tampered_row_does_not`
// exists to make impossible.
let Some(body) = material.body.clone() else {
return Err(AppError::Internal(anyhow::anyhow!(
"status claim write reached with signature material carrying no request body: \
the route lost its persist-body marker or applied it inside the auth layers"
)));
};
// Bounded before anything is written. The material is verified, which says
// the caller holds the key, not that what they signed is a reasonable size.
bound("signature", material.signature.len(), MAX_SIGNATURE_CHARS)?;
bound(
"signature_input",
material.signature_input.len(),
MAX_SIGNATURE_INPUT_CHARS,
)?;
bound(
"signing_string",
material.signing_string.len(),
MAX_SIGNING_STRING_CHARS,
)?;
bound("request body", body.len(), MAX_REQUEST_BODY_BYTES)?;
// After the bounds, so the scan below walks a signing string of known size.
require_body_bound_signature(&material, &body)?;
let digest = request_digest(&material, &body);
let (signature, signature_input, signing_string, request_body) = (
material.signature,
material.signature_input,
material.signing_string,
body.to_vec(),
);
let claim = StatusClaim {
id: Uuid::new_v4().to_string(),
// Ignored on insert: the database assigns the ordering key (KTD-3).
seq: 0,
repo_id: record.id.clone(),
commit_sha,
state: claim_state,
context,
target_url,
description,
// Both are the owner identity today (KTD-5); they split when delegated
// capabilities land.
producer_did: auth.0.clone(),
authorizing_did: auth.0,
signature,
signature_input,
signing_string,
request_body,
request_digest: digest,
created_at: Utc::now().to_rfc3339(),
};
let caps = crate::db::ClaimCaps {
per_tuple: MAX_CLAIMS_PER_TUPLE,
contexts_per_commit: MAX_CONTEXTS_PER_COMMIT,
per_repo_window: MAX_CLAIMS_PER_REPO_PER_WINDOW,
};
match state.db.insert_status_claim_capped(&claim, &caps).await? {
crate::db::ClaimInsert::Inserted(_) => {
Ok((StatusCode::CREATED, Json(CreatedStatus::from_claim(claim))))
}
// Already recorded, so 200 and the original row rather than 201 and a
// second one. The claim the caller gets back is the stored one, id
// included, which is what makes a retry safe to treat as success.
crate::db::ClaimInsert::AlreadyRecorded(existing) => {
Ok((StatusCode::OK, Json(CreatedStatus::from_claim(*existing))))
}
crate::db::ClaimInsert::CapExceeded(which) => Err(AppError::TooManyRequests(format!(
"claim limit reached for {which}"
))),
}
}
/// Refuse a claim whose signature does not cover the body about to be stored.
///
/// A claim IS provenance, so the signature stored beside it has to bind the
/// bytes stored beside it. It does not do that directly: `signing_string`
/// reaches the body only through a `content-digest` line. And a caller can sign
/// a content-digest the request then never carries. `require_signature` rebuilds
/// the covered values from the request, so an ABSENT `Content-Digest` header
/// makes the covered digest the empty string on the signing side and the
/// verifying side alike; the Ed25519 check passes over method, path and an empty
/// digest, and any body at all rides along. The row that write produces asserts
/// a provenance that cannot be re-verified: the signing string covers a digest
/// of bytes that are not the ones in `request_body`.
///
/// The predicate here is the write-time form of the one `re_verify` runs against
/// a stored row in the tests — recompute the digest of the bytes about to be
/// persisted and require the signing string to carry exactly that line — so what
/// this route accepts and what a third party can later re-verify are the same
/// question, asked at both ends.
///
/// This is a route property, checked where the claim becomes history, and it is
/// deliberately not the only line of defence: requiring the header in
/// `require_signature` closes the same hole one layer earlier and for every
/// signed route, and that change belongs there rather than here because no other
/// route's correctness should depend on the status handler. The two agree and
/// neither substitutes for the other — without the middleware's requirement the
/// empty digest is still accepted everywhere else, and without this one a later
/// relaxation of the middleware would silently resume writing unverifiable
/// history.
///
/// A 400, not a 500: the material is exactly what the client signed, and what
/// they signed does not cover what they sent.
fn require_body_bound_signature(
material: &crate::auth::SignatureMaterial,
body: &[u8],
) -> Result<()> {
let covered = format!(
"\"content-digest\": {}",
gitlawb_core::http_sig::compute_content_digest(body)
);
// Line-exact rather than a substring search: `build_signing_string` emits one
// component per line, and a `contains` would also be satisfied by the digest
// appearing INSIDE some other covered component's value, which proves
// nothing about what the signature covered.
if material.signing_string.lines().any(|line| line == covered) {
return Ok(());
}
Err(AppError::BadRequest(
"the signature does not cover this request body: send a Content-Digest header \
computed over the body and name \"content-digest\" among the signature's covered \
components"
.into(),
))
}
/// The identity of one signed write, as a hex sha-256.
///
/// All four inputs together, because a replay is the same bytes arriving twice:
/// the signature, the input that names what it covers, the canonical string it
/// was verified over, and the body that string covers through a content-digest.
/// The signature alone would nearly do (it covers the other three transitively),
/// but hashing what is actually stored means the digest describes the row rather
/// than a claim about it.
///
/// Each field is length-prefixed so no two different requests can concatenate to
/// the same bytes: without it, a signature ending in some prefix of the next
/// field would collide with the shorter signature that absorbed it.
///
/// This is not a nonce, and deliberately so. A nonce table needs pruning and a
/// pruning window is a second replay window; the claim row IS the record of what
/// was accepted, so uniqueness on it is self-maintaining.
fn request_digest(material: &crate::auth::SignatureMaterial, body: &[u8]) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
for part in [
material.signature.as_bytes(),
material.signature_input.as_bytes(),
material.signing_string.as_bytes(),
body,
] {
h.update((part.len() as u64).to_be_bytes());
h.update(part);
}
format!("{:x}", h.finalize())
}
/// One reported context in the combined response. Signature material is
/// deliberately absent: it is write-time provenance, not read-surface data.
#[derive(Serialize)]
pub struct StatusEntry {
pub state: String,
pub context: String,
pub target_url: Option<String>,
pub description: Option<String>,
pub producer_did: String,
pub created_at: String,
}
/// The combined commit status. `state` never leaves the four-value set (KTD-1),
/// so absence is carried by `total_count` 0 with the pending state rather than a
/// fifth value, and `reported_only` (R19) says out loud that the state covers the
/// contexts that reported, not every check a caller expected.
#[derive(Serialize)]
pub struct CombinedStatus {
pub state: String,
pub sha: String,
pub total_count: usize,
pub statuses: Vec<StatusEntry>,
pub reported_only: bool,
}
/// GET /api/v1/repos/:owner/:repo/commits/:sha/status
///
/// The auth extension is optional and MUST be last in the extractor list: the
/// route group carries `optional_signature`, so an unsigned caller reaches here
/// with no `AuthenticatedDid` at all and a public repo still answers.
pub async fn commit_status(
State(state): State<AppState>,
Path((owner, name, sha)): Path<(String, String, String)>,
auth: Option<Extension<AuthenticatedDid>>,
) -> Result<Json<CombinedStatus>> {
let caller = auth.as_ref().map(|Extension(a)| a.0.as_str());
// The gate runs first and on the requested path, before any claim data is
// touched. Its deny is the repo's own not-found, byte-identical to a missing
// repo, so the status surface cannot answer "does this repo exist".
let (record, _rules) =
crate::api::authorize_repo_read(&state, &owner, &name, caller, "/").await?;
let commit_sha = normalize_sha(&sha)?;
let statuses = project_claims(&state, &record, &commit_sha).await?;
Ok(Json(CombinedStatus {
state: combined_state(&statuses).to_string(),
sha: commit_sha,
total_count: statuses.len(),
statuses,
reported_only: true,
}))
}
/// The projection, in one place. Both read surfaces call it, so R3's "the rollup
/// is derived by the same projection as the commit read" holds by construction
/// rather than by two implementations happening to agree.
///
/// KTD-5: current authorization, not write-time authorization. An ownership
/// transfer drops the prior owner's claims from the projection while the
/// append-only history keeps them.
async fn project_claims(
state: &AppState,
record: &crate::db::RepoRecord,
commit_sha: &str,
) -> Result<Vec<StatusEntry>> {
let claims = state
.db
.latest_status_claims(&record.id, commit_sha, &record.owner_did)
.await?;
Ok(claims
.into_iter()
.map(|c| StatusEntry {
state: c.state,
context: c.context,
target_url: c.target_url,
description: c.description,
producer_did: c.producer_did,
created_at: c.created_at,
})
.collect())
}
/// The pull request head rollup (R11). `state` stays inside the four wire values
/// whatever happened to the head (KTD-1): an unresolvable head is carried by
/// `head_resolved`, alongside the pull request's own state, so a client can tell
/// "the head could not be resolved" from "the head resolved and nothing reported"
/// without a fifth state value no client understands.
#[derive(Serialize)]
pub struct PullRequestStatus {
pub number: i64,
/// The pull request's own state: open, closed, or merged.
pub pull_request_state: String,
pub head_resolved: bool,
/// The target commit, absent exactly when `head_resolved` is false.
pub sha: Option<String>,
pub state: String,
pub total_count: usize,
pub statuses: Vec<StatusEntry>,
pub reported_only: bool,
}
/// GET /api/v1/repos/:owner/:repo/pulls/:number/status
///
/// Same optional-auth shape as the commit read, and the auth extension MUST stay
/// last in the extractor list.
pub async fn pull_request_status(
State(state): State<AppState>,
Path((owner, name, number)): Path<(String, String, i64)>,
auth: Option<Extension<AuthenticatedDid>>,
) -> Result<Json<PullRequestStatus>> {
let caller = auth.as_ref().map(|Extension(a)| a.0.as_str());
// The gate runs first, on the requested path, before the pull request row is
// loaded. Its deny is the repo's own not-found, byte-identical to a missing
// repo. Once it passes, a missing pull request number is the plain not-found:
// existence is no longer secret.
let (record, _rules) =
crate::api::authorize_repo_read(&state, &owner, &name, caller, "/").await?;
let pr = state
.db
.get_pr(&record.id, number)
.await?
.ok_or_else(|| AppError::NotFound(format!("PR #{number} not found")))?;
let head = rollup_head(&state, &record, &pr).await?;
let statuses = match &head {
Some(sha) => project_claims(&state, &record, sha).await?,
None => Vec::new(),
};
Ok(Json(PullRequestStatus {
number: pr.number,
pull_request_state: pr.status,
head_resolved: head.is_some(),
sha: head,
state: combined_state(&statuses).to_string(),
total_count: statuses.len(),
statuses,
reported_only: true,
}))
}
/// Branch-head lookups performed by the rollup fallback, counted so a test can
/// assert WORK DONE rather than only the answer returned. A fallback that
/// re-resolved on every read and then discarded the answer would be invisible in
/// the response and visible here.
#[cfg(test)]
pub(crate) static BRANCH_RESOLVES: std::sync::atomic::AtomicUsize =
std::sync::atomic::AtomicUsize::new(0);
/// The rollup's target commit (KTD-4).
///
/// The stored head wins whenever it is set. It is frozen at close or merge, so a
/// closed or merged pull request keeps the commit the decision was made against,
/// and only an OPEN pull request with no stored head falls back to the branch.
///
/// The fallback is one database read: the latest push this node recorded for the
/// pull request's source branch, from `repo_push_events`. It deliberately does
/// NOT go through `git::store` ref listing, which needs an acquired repository
/// path — on a cold node that downloads the whole repository from object storage,
/// and this read group carries no rate limiter, so an anonymous caller could
/// drive repeated downloads with a URL.
///
/// `repo_push_events` is the source rather than `branch_cids` because
/// `record_push_events` writes it for every ref update unconditionally, whereas
/// the sole writer of `branch_cids` sits behind a pin CID. A node with no object
/// pinning configured never writes that table, so a resolve keyed on it could
/// never answer there and every open pull request without a stored head reported
/// `head_resolved: false` forever. The residual limit is that only pushes taken
/// after this shipped have rows, so a branch last pushed before then still does
/// not resolve.
///
/// The resolved head is persisted, which makes an unauthenticated GET write. That
/// is only acceptable because the write is self-limiting: it fires exactly when
/// `head_commit` is absent and it fills it, and the fill is conditioned on that
/// same absence in SQL, so the PERSIST happens at most once per pull request.
/// The resolve attempt itself is not once-only — a pull request whose branch does
/// not resolve stores nothing, so there is nothing to short-circuit on and the
/// lookup runs again on every read until it succeeds.
async fn rollup_head(
state: &AppState,
record: &crate::db::RepoRecord,
pr: &crate::db::PullRequest,
) -> Result<Option<String>> {
if let Some(stored) = pr.head_commit.clone() {
return Ok(Some(stored));
}
if pr.status != "open" {
return Ok(None);
}
// The push path records the full ref while a pull request stores the bare
// branch name, the same mismatch `crate::api::repos::branch_from_ref`
// reconciles on the write side; this is that mapping run backwards. The empty
// branch is excluded for the same reason it is there: `refs/heads/` is not a
// branch and nothing can have written it.
if pr.source_branch.is_empty() {
return Ok(None);
}
let ref_name = crate::git::store::local_branch_ref(&pr.source_branch);
#[cfg(test)]
BRANCH_RESOLVES.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let resolved = state
.db
.latest_push_sha_for_ref(&record.id, &ref_name)
.await?
// A recorded head that is not a commit SHA is no target at all, and
// storing it would leave the rollup pointing at nothing forever.
.and_then(|sha| normalize_sha(&sha).ok());
let Some(sha) = resolved else {
return Ok(None);
};
// Best effort, and deliberately not `?`. The head is already resolved and the
// response does not depend on this write landing, so a transient database
// failure here must not turn a served read into a 500; the next read resolves
// again and re-attempts the fill. Same catch-and-log the sibling writes on the
// push path use (`update_open_pr_heads`, `record_push_events`).
if let Err(e) = state.db.set_pr_head_if_absent(&pr.id, &sha).await {
tracing::warn!(
err = %e,
pr_id = %pr.id,
"failed to persist a resolved pull request head; serving the resolved head anyway"
);
}
Ok(Some(sha))
}
/// KTD-1's fold: any error or failure yields failure, else any pending yields
/// pending, else success. An empty set is pending, never success — total absence
/// of a verdict is its own state and must not read as a pass (R10).
fn combined_state(statuses: &[StatusEntry]) -> &'static str {
if statuses.is_empty() {
return "pending";
}
if statuses
.iter()
.any(|s| s.state == "error" || s.state == "failure")
{
"failure"
} else if statuses.iter().any(|s| s.state == "pending") {
"pending"
} else if statuses.iter().all(|s| s.state == "success") {
"success"
} else {
// A stored state outside the vocabulary is not a pass: an
// unrecognized verdict folds toward failure, never silently to success.
"failure"
}
}
/// 40 hex characters, lowercased. The SHA is never existence-checked (KTD-7):
/// a claim describes a commit object, and the object may not have arrived yet.
fn normalize_sha(sha: &str) -> Result<String> {
if crate::api::repos::is_hex_sha(sha) {
Ok(sha.to_ascii_lowercase())
} else {
Err(AppError::BadRequest(
"commit sha must be exactly 40 hexadecimal characters".into(),
))
}
}
/// One size bound on persisted signature material, measured in bytes.
fn bound(what: &str, len: usize, max: usize) -> Result<()> {
if len > max {
return Err(AppError::BadRequest(format!(
"{what} must be at most {max} bytes"
)));
}
Ok(())
}
fn validate_state(state: &str) -> Result<String> {
if CLAIM_STATES.contains(&state) {
Ok(state.to_string())
} else {
Err(AppError::BadRequest(format!(
"state must be one of {}",
CLAIM_STATES.join(", ")
)))
}
}
/// The context is the projection key, so it is trimmed, bounded, and required to
/// be free of control characters.
fn validate_context(context: &str) -> Result<String> {
let trimmed = context.trim();
if trimmed.is_empty() || trimmed.chars().count() > MAX_CONTEXT_CHARS {
return Err(AppError::BadRequest(format!(
"context must be 1 to {MAX_CONTEXT_CHARS} characters"
)));
}
if trimmed.chars().any(char::is_control) {
return Err(AppError::BadRequest(
"context must not contain control characters".into(),
));
}
Ok(trimmed.to_string())
}
fn validate_target_url(url: Option<&str>) -> Result<Option<String>> {
let Some(url) = url else { return Ok(None) };
if url.chars().count() > MAX_TARGET_URL_CHARS {
return Err(AppError::BadRequest(format!(
"target_url must be at most {MAX_TARGET_URL_CHARS} characters"
)));
}
let lower = url.to_ascii_lowercase();
if lower.starts_with("http://") || lower.starts_with("https://") {
Ok(Some(url.to_string()))
} else {
Err(AppError::BadRequest(
"target_url must be an http or https URL".into(),
))
}
}
fn validate_description(description: Option<&str>) -> Result<Option<String>> {
match description {
Some(d) if d.chars().count() > MAX_DESCRIPTION_CHARS => Err(AppError::BadRequest(format!(
"description must be at most {MAX_DESCRIPTION_CHARS} characters"
))),
other => Ok(other.map(str::to_string)),
}
}
#[cfg(test)]
mod tests;
File diff suppressed because it is too large. Load diff
+168 -8
View File
@@ -17,6 +17,53 @@ use crate::state::AppState;
#[derive(Clone, Debug)]
pub struct AuthenticatedDid(pub String);
/// The RFC 9421 material this request was actually verified against, injected
/// alongside [`AuthenticatedDid`] by `require_signature`.
///
/// A handler that records a signed claim as history has to store it: the header
/// values and the canonical signing string are gone once the request ends, and a
/// stored claim nobody can re-verify is not history. `signing_string` is the exact
/// byte sequence the Ed25519 verification succeeded over, not a rebuild — a
/// handler reconstructing it from headers would be verifying a different string
/// than the middleware did.
#[derive(Clone, Debug)]
pub struct SignatureMaterial {
/// The `Signature` header value.
pub signature: String,
/// The `Signature-Input` header value.
pub signature_input: String,
/// The canonical bytes the signature covered.
pub signing_string: String,
/// The buffered request body, on the routes that persist it. The signing
/// string covers it only through the content-digest, so without these bytes
/// a stored claim can be shown to carry *a* valid signature over *some*
/// digest and nothing more. Carried here because the body is consumed
/// downstream by the extractor and cannot be recovered afterwards.
///
/// `Option`, and populated only behind [`PersistsSignedBody`], because a
/// route that never stores the body has no use for a second handle to it.
/// The unlike-sized field is this one: the three above are header-derived
/// and bounded by the server's header limit, while a receive-pack POST
/// carries up to GITLAWB_MAX_PACK_BYTES (2 GB by default). `Bytes`, not
/// `Vec<u8>`, so the handle the marked routes do take is a refcount bump
/// over the buffer the middleware already holds rather than a copy.
pub body: Option<axum::body::Bytes>,
}
/// Marks a route group whose handler persists the signed request body.
///
/// Applied as an extension layer OUTSIDE the auth layers, so it is on the
/// request before [`require_signature`] decides whether to carry the body. A
/// marker applied inside them is never seen, and the failure is silent: the
/// handler still compiles, tests that inject material by hand still pass, and
/// production stores nothing. The status write handler refuses an absent body
/// for that reason.
///
/// A marker rather than a path check because `require_signature` must not learn
/// which URL persists claims; that knowledge belongs at the route table.
#[derive(Clone, Copy, Debug)]
pub struct PersistsSignedBody;
/// A UCAN that passed full chain validation, with the root issuer the chain
/// rests on. Inserted into request extensions by [`require_ucan_chain`] when
/// `X-Ucan` is present; absent when the header is.
@@ -375,10 +422,28 @@ async fn verify_signature(limit: usize, request: Request, next: Next) -> Respons
tracing::info!(did = %sig.key_id, "✓ authenticated request");
// The body travels only where it is stored. On every other signed route the
// middleware's handle would keep the buffer alive for the rest of the layer
// chain, until the handler's own extractor consumes the request.
let body = parts
.extensions
.get::<PersistsSignedBody>()
.map(|_| body_bytes.clone());
let material = SignatureMaterial {
signature: sig_header,
signature_input: sig_input,
signing_string,
body,
};
let mut request = Request::from_parts(parts, Body::from(body_bytes));
request
.extensions_mut()
.insert(AuthenticatedDid(sig.key_id.to_string()));
// Carry the verified material forward for handlers that persist a signed
// claim; none of it can be recovered once the request is gone.
request.extensions_mut().insert(material);
next.run(request).await
}
@@ -738,12 +803,6 @@ mod tests {
task_tx.clone(),
task_cursor_key.clone(),
));
let ws_schema = Arc::new(graphql::build_ws_schema(
db.clone(),
ref_tx.clone(),
task_tx.clone(),
task_cursor_key.clone(),
));
crate::state::AppState {
config: Arc::new(Config::parse_from(["gitlawb-node"])),
db,
@@ -754,13 +813,11 @@ mod tests {
ref_update_tx: ref_tx,
task_event_tx: task_tx,
graphql_schema: schema,
graphql_ws_schema: ws_schema,
task_cursor_key,
machine_id: None,
repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
write_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
@@ -798,6 +855,109 @@ mod tests {
}
}
/// Reports what the signature middleware put in `SignatureMaterial.body`,
/// so a test can tell "captured" from "deliberately not captured" apart
/// from "the middleware never ran".
async fn report_body(material: Option<axum::Extension<SignatureMaterial>>) -> String {
match material {
Some(axum::Extension(m)) => match m.body {
Some(bytes) => format!("captured:{}", bytes.len()),
None => "absent".to_string(),
},
None => "no-material".to_string(),
}
}
fn signed_post(path: &str, body: &[u8]) -> Request {
let kp = Keypair::generate();
let signed = gitlawb_core::http_sig::sign_request(&kp, "POST", path, body);
Request::builder()
.method("POST")
.uri(path)
.header("content-digest", signed.content_digest)
.header("signature-input", signed.signature_input)
.header("signature", signed.signature)
.body(axum::body::Body::from(body.to_vec()))
.unwrap()
}
/// The body handle travels only on a route that asked for it.
///
/// Both arms drive a genuinely signed request through `require_signature`;
/// the only difference is the [`PersistsSignedBody`] marker layer. A
/// middleware that captured unconditionally would make both arms read
/// `captured`, and one that never read the marker would make both read
/// `absent`, so the pair pins the decision rather than the mechanism.
#[tokio::test]
async fn the_body_is_carried_only_when_the_route_marks_itself_as_persisting_it() {
let body = br#"{"state":"success"}"#;
// The marker layer is added AFTER the middleware layer, which is what
// makes it the outer of the two and therefore the one that runs first.
// This is the same relative position `build_router` uses on the status
// write group, where the marker sits outside `add_auth_layers`.
let marked = Router::new()
.route("/", axum::routing::post(report_body))
.layer(middleware::from_fn_with_state(
crate::auth::SignatureBodyLimit(65536),
require_signature,
))
.layer(axum::Extension(PersistsSignedBody));
let resp = marked.oneshot(signed_post("/", body)).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
assert_eq!(
String::from_utf8(seen.to_vec()).unwrap(),
format!("captured:{}", body.len()),
"a route carrying the persist marker must receive the buffered body"
);
let unmarked = Router::new()
.route("/", axum::routing::post(report_body))
.layer(middleware::from_fn_with_state(
crate::auth::SignatureBodyLimit(65536),
require_signature,
));
let resp = unmarked.oneshot(signed_post("/", body)).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
assert_eq!(
String::from_utf8(seen.to_vec()).unwrap(),
"absent",
"a signed route that never persists the body must not be handed a handle to it"
);
}
/// The marker is read from the request the middleware was given, and the
/// request it hands downstream still carries it. Both halves matter: the
/// first is the capture decision, the second is that rebuilding the request
/// from its parts does not drop extensions a later layer may want.
#[tokio::test]
async fn the_marker_survives_the_request_rebuild() {
async fn report_marker(marker: Option<axum::Extension<PersistsSignedBody>>) -> String {
match marker {
Some(_) => "marked".to_string(),
None => "unmarked".to_string(),
}
}
let app = Router::new()
.route("/", axum::routing::post(report_marker))
.layer(middleware::from_fn_with_state(
crate::auth::SignatureBodyLimit(65536),
require_signature,
))
.layer(axum::Extension(PersistsSignedBody));
let resp = app.oneshot(signed_post("/", b"{}")).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let seen = axum::body::to_bytes(resp.into_body(), 2048).await.unwrap();
assert_eq!(
String::from_utf8(seen.to_vec()).unwrap(),
"marked",
"the marker must reach the handler, so the middleware saw it too"
);
}
#[tokio::test]
async fn require_ucan_chain_no_header_passes_through() {
let state = make_test_state(Keypair::generate().did());
+6 -33
View File
@@ -1,4 +1,4 @@
use clap::{Parser, Subcommand};
use clap::Parser;
use std::path::{Path, PathBuf};
/// Upper bound on `git_service_timeout_secs`, `ipfs_request_budget_secs`, and
@@ -28,45 +28,18 @@ use std::path::{Path, PathBuf};
/// panicked are rejected.
pub const GIT_SERVICE_TIMEOUT_SECS_MAX: u64 = 100 * 365 * 24 * 60 * 60;
/// Optional admin subcommands. When none is given, the binary runs the node
/// daemon as before — the default (no-subcommand) startup path is unchanged.
#[derive(Subcommand, Debug, Clone)]
pub enum Command {
/// Dry-run (default) or, with --execute, delete empty spam-burst repos owned
/// by the known burst DID. Never touches the hard-excluded DIDs, and verifies
/// each repo is empty (zero git refs) per repo before selecting it.
PurgeSpam {
/// Actually delete the candidates. Omit for a dry-run that prints the
/// candidate list and deletes nothing.
#[arg(long, default_value_t = false)]
execute: bool,
},
}
#[derive(Parser, Debug, Clone)]
#[command(name = "gitlawb-node", about = "Twigpine node daemon", version)]
pub struct Config {
/// Admin subcommand to run instead of the node daemon. Absent = run the node.
#[command(subcommand)]
pub command: Option<Command>,
/// Directory where bare git repositories are stored. `global` so it can
/// follow a subcommand (e.g. `gitlawb-node purge-spam --repos-dir …`).
#[arg(
long,
env = "GITLAWB_REPOS_DIR",
default_value = "./data/repos",
global = true
)]
/// Directory where bare git repositories are stored
#[arg(long, env = "GITLAWB_REPOS_DIR", default_value = "./data/repos")]
pub repos_dir: PathBuf,
/// PostgreSQL connection URL (Supabase or any Postgres instance). `global` so
/// admin subcommands accept it in either position.
/// PostgreSQL connection URL (Supabase or any Postgres instance)
#[arg(
long,
env = "DATABASE_URL",
default_value = "postgresql://localhost/gitlawb",
global = true
default_value = "postgresql://localhost/gitlawb"
)]
pub database_url: String,
@@ -275,7 +248,7 @@ pub struct Config {
/// Tigris (S3-compatible) bucket for repo storage.
/// Leave empty to disable Tigris and use local-only storage.
#[arg(long, env = "GITLAWB_TIGRIS_BUCKET", default_value = "", global = true)]
#[arg(long, env = "GITLAWB_TIGRIS_BUCKET", default_value = "")]
pub tigris_bucket: String,
/// Maximum pack body size for git-receive-pack and git-upload-pack, in bytes.
File diff suppressed because it is too large. Load diff
+32
View File
@@ -1285,6 +1285,38 @@ async fn run_effect_bundle(
}
}
// 7a. Commit-status surfaces: the stored pull-request heads and the
// catch-up poll feed are durable accounting like everything else
// in this bundle, so they ride the same proven-landing authority —
// derived from the accepted transition rows, never from a live
// parse, and a landing reconcile proves only later still emits
// them here on the drain pass. Head writes check each branch's
// current tip before moving it: this executor runs after the repo
// write lock is released, so a newer push may already have moved
// the branch and only the still-current landing may win the head.
let landed: Vec<crate::api::repos::RefUpdate> = accepted_children
.iter()
.map(|c| crate::api::repos::RefUpdate {
old_sha: c.old_sha.clone(),
new_sha: c.new_sha.clone(),
ref_name: c.ref_name.clone(),
})
.collect();
let heads =
crate::api::repos::heads_still_at_tip(std::path::Path::new(&repo.disk_path), &landed)?;
if let Err(e) = crate::api::repos::update_open_pr_heads(&state.db, &req.repo_id, &heads).await {
return Ok(EffectsOutcome::Retry {
last_error: format!("stored pr heads: {e}"),
});
}
if let Err(e) =
crate::api::repos::record_push_events(&state.db, &req.id, accepted_children).await
{
return Ok(EffectsOutcome::Retry {
last_error: format!("push events: {e}"),
});
}
// 8. Per-ref certs and anchor jobs. Each accepted child gets one
// of each. Failures are accumulated; the first one is
// returned as the Retry reason. Anchor identity is by occurrence
+10 -21
View File
@@ -50,13 +50,6 @@ pub enum AppError {
#[error("too many requests: {0}")]
TooManyRequests(String),
/// A 429 from a per-IP limiter that knows when its window frees a slot, so
/// the response advertises that delay as `Retry-After` the way the router
/// middleware's 429s do. `TooManyRequests` stays for limiters that have no
/// such delay to give.
#[error("too many requests: retry in {}s", .0.as_secs())]
RateLimited(std::time::Duration),
#[error("incomplete: {0}")]
Incomplete(String),
@@ -212,11 +205,6 @@ impl IntoResponse for AppError {
AppError::TooManyRequests(msg) => {
(StatusCode::TOO_MANY_REQUESTS, "rate_limited", msg.clone())
}
AppError::RateLimited(_) => (
StatusCode::TOO_MANY_REQUESTS,
"rate_limited",
crate::rate_limit::RATE_LIMIT_MESSAGE.to_string(),
),
AppError::Incomplete(msg) => {
(StatusCode::UNPROCESSABLE_ENTITY, "incomplete", msg.clone())
}
@@ -233,9 +221,6 @@ impl IntoResponse for AppError {
// 504, distinct from the 500 git_error and from the read-gate's 404 /
// the auth 401, so the client can tell a deadline from a failure.
AppError::Timeout(msg) => (StatusCode::GATEWAY_TIMEOUT, "git_timeout", msg.clone()),
// 503, retryable: a transient resource contention (e.g. the repo lock
// pool is pinned by a write burst or a purge holds the key), distinct
// from the 500 git_error so the client knows to back off and retry.
// 503 with a FIXED body: the caller should retry, and must not be told
// which repo is contended or for how long.
AppError::RepoBusy => (
@@ -330,12 +315,6 @@ impl IntoResponse for AppError {
axum::http::HeaderValue::from_static("1"),
);
}
if let AppError::RateLimited(retry_after) = &self {
resp.headers_mut().insert(
axum::http::header::RETRY_AFTER,
axum::http::HeaderValue::from(retry_after.as_secs().max(1)),
);
}
resp
}
}
@@ -379,6 +358,16 @@ mod tests {
);
}
/// A 429 does not advertise a retry interval: the status-claim caps behind
/// it are per-commit and append-only, so a client honoring a 1s hint would
/// retry a request that can never succeed.
#[test]
fn too_many_requests_carries_no_retry_after() {
let resp = AppError::TooManyRequests("x".into()).into_response();
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
assert!(resp.headers().get("retry-after").is_none());
}
/// #226: raw sqlx/DB detail must never appear in the Internal 500 body.
#[tokio::test]
async fn internal_error_body_is_opaque() {
-31
View File
@@ -61,37 +61,6 @@ pub fn create_issue(repo_path: &Path, issue_id: &str, json: &str) -> Result<()>
Ok(())
}
/// Deadline-bounded variant of [`create_issue`] for compensation paths: the
/// same hash-object + update-ref write, but every git child runs through the
/// shared bounded runner so a hung git cannot stall the write guard past
/// `git_service_timeout`.
pub fn create_issue_bounded(
git_bin: &str,
repo_path: &Path,
issue_id: &str,
json: &str,
deadline: std::time::Instant,
) -> Result<()> {
let out = crate::git::visibility_pack::run_bounded_git(
git_bin,
&["hash-object", "--stdin", "-w"],
repo_path,
json.as_bytes(),
deadline,
)?;
let hash = String::from_utf8(out).context("git hash-object output is not valid UTF-8")?;
let hash = hash.trim();
let ref_name = format!("refs/gitlawb/issues/{issue_id}");
crate::git::visibility_pack::run_bounded_git(
git_bin,
&["update-ref", &ref_name, hash],
repo_path,
b"",
deadline,
)?;
Ok(())
}
/// Remove a single issue ref while the write guard still holds the advisory lock,
/// after a definite publish refusal. Best-effort: a failed delete leaves a
/// local-only orphan, which is still better than telling the client to retry into a
+28 -243
View File
@@ -3,8 +3,7 @@
//! Every handler that needs access to a git repo on disk goes through `RepoStore`:
//!
//! - `acquire()` — ensures the repo is on local disk (downloads from Tigris on cache miss).
//! - `upload_under_guard()`: uploads the updated repo to Tigris after a write, under a
//! per-repo advisory lock the caller already holds (the fork tail's durability step).
//! - `release_after_write()` — uploads the updated repo to Tigris after a write operation.
//! - `init()` — creates a new bare repo locally and uploads to Tigris.
//!
//! When Tigris is disabled (bucket empty), this is a simple passthrough to local disk.
@@ -51,7 +50,7 @@ impl ConfirmedLiveTree {
}
}
/// Centralized repo storage: local disk cache + optional object-store backend.
/// Centralized repo storage: local disk cache + optional Tigris backend.
#[derive(Clone)]
pub struct RepoStore {
repos_dir: PathBuf,
@@ -802,22 +801,7 @@ impl RepoStore {
}
/// Initialize a new bare repo on local disk and upload to Tigris.
/// Test-only convenience over `init_handle`; every
/// production caller keeps the upload inside its lock.
#[cfg(test)]
pub async fn init(&self, owner_did: &str, repo_name: &str) -> Result<PathBuf> {
let (path, upload) = self.init_handle(owner_did, repo_name).await?;
drop(upload);
Ok(path)
}
/// [`init`](Self::init) plus the background archive upload's JoinHandle, so
/// callers holding a lock can keep the upload inside the lock's lifetime.
pub async fn init_handle(
&self,
owner_did: &str,
repo_name: &str,
) -> Result<(PathBuf, Option<tokio::task::JoinHandle<Result<()>>>)> {
let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?;
store::init_bare(&local_path).context("initializing bare repo")?;
@@ -828,13 +812,12 @@ impl RepoStore {
clear_quarantine(&local_path, repo_name);
// Upload to Tigris in background
let mut upload_handle = None;
if let Some(ref tigris) = self.tigris {
let tigris = tigris.clone();
let owner_slug = owner_slug.clone();
let repo_name = repo_name.to_string();
let path = local_path.clone();
upload_handle = Some(tokio::spawn(async move {
tokio::spawn(async move {
// Create-only, and load-bearing: this uploads a freshly
// initialized EMPTY repo, so a user who pushes immediately
// after creating one would have their archive replaced by this
@@ -845,26 +828,20 @@ impl RepoStore {
.upload(&owner_slug, &repo_name, &path, UploadPrecondition::IfAbsent)
.await
{
Ok(_) => Ok(()),
// Distinct from the error arm: the fence refusing means
// someone else already published this key, so the empty
// bytes dropped — but for a create caller the archive then
// belongs to another writer, which IS a creation failure.
Ok(_) => {}
// Distinct from the warn arm: the fence refusing is the
// design working, not a storage failure.
Err(UploadError::PreconditionLost { status }) => {
info!(repo = %repo_name, status, "dropped the empty-repo upload: another writer already published this repo");
Err(anyhow::anyhow!(
"empty-repo upload fenced: key already published (status {status})"
))
}
Err(e) => {
warn!(repo = %repo_name, err = %e, "failed to upload new repo to tigris");
Err(anyhow::anyhow!("empty-repo upload failed: {e}"))
}
}
}));
});
}
Ok((local_path.into_path_buf(), upload_handle))
Ok(local_path.into_path_buf())
}
/// Upload a repo to Tigris after a write operation (push, merge, fork, etc.).
@@ -916,102 +893,6 @@ impl RepoStore {
Ok(())
}
/// Whether an object store is configured. Used by purge-spam to decide
/// whether a repo with no local copy can be a remote-unverified candidate
/// (its emptiness verified under the lock after a refresh) versus fail-closed.
pub fn has_object_store(&self) -> bool {
self.tigris.is_some()
}
/// Try to take the per-repo advisory lock WITHOUT the write guard's
/// download/upload pipeline: a bare lock for callers (purge-spam, the
/// create/fork namespace serialization) that must serialize against the
/// write path on the same key but are not themselves publishing a tree.
///
/// Returns `Some(guard)` if the lock was free, `None` if another writer holds
/// it (so the caller can skip rather than block). The guard holds a dedicated
/// pooled connection for its whole lifetime, so the lock lives on that one
/// connection and `release()` unlocks it on the SAME connection — a plain
/// pool query could unlock on a different connection and silently fail.
pub async fn try_lock_repo(
&self,
owner_did: &str,
repo_name: &str,
) -> Result<Option<RepoLockGuard>> {
self.try_lock_repo_on(&self.lock_pool, owner_did, repo_name)
.await
}
/// [`try_lock_repo`](Self::try_lock_repo) against a caller-chosen pool. The
/// Postgres advisory lock is DATABASE-global, so the lock still mutually
/// excludes holders of the same key from any other pool — only the connection
/// the guard pins comes from `pool`. Backs the `lock_pool`-based delegators.
async fn try_lock_repo_on(
&self,
pool: &PgPool,
owner_did: &str,
repo_name: &str,
) -> Result<Option<RepoLockGuard>> {
let (owner_slug, _local) = self.local_path(owner_did, repo_name)?;
let lock_key = advisory_lock_key(&owner_slug, repo_name);
let conn = pool.acquire().await.context("acquiring lock connection")?;
// Route through LockProbe so a cancellation racing the query cannot
// return a connection holding an acquired session lock to the pool:
// the probe's Drop closes the session unless take_conn hands it out.
let mut probe = LockProbe::new(conn);
if !probe
.try_lock(lock_key)
.await
.context("try advisory lock")?
{
return Ok(None);
}
let conn = probe
.take_conn()
.context("lock probe lost its connection after a confirmed acquire")?;
Ok(Some(RepoLockGuard {
conn: Some(conn),
lock_key,
repo_name: repo_name.to_string(),
}))
}
/// Bounded-wait lock-only acquire, without any object-store I/O. Retries
/// `try_lock_repo` with short backoff. Used by `create_repo` and `fork_repo`
/// to serialize their existence-check -> write -> row-insert spans against a
/// concurrent same-key purge (which holds this same lock across delete-row +
/// remove-dir). A key held past the bounded wait surfaces as `Ok(None)`,
/// which the create/fork handlers map to a retryable 503.
pub async fn lock_repo_blocking(
&self,
owner_did: &str,
repo_name: &str,
) -> Result<Option<RepoLockGuard>> {
self.lock_repo_blocking_on(&self.lock_pool, owner_did, repo_name)
.await
}
/// [`lock_repo_blocking`](Self::lock_repo_blocking) against a caller-chosen
/// pool. The advisory lock is DB-global, so the guard mutually excludes the same
/// key across pools; only the pinned connection comes from `pool`. Backs the
/// `lock_pool`-based [`lock_repo_blocking`](Self::lock_repo_blocking) delegator.
pub async fn lock_repo_blocking_on(
&self,
pool: &PgPool,
owner_did: &str,
repo_name: &str,
) -> Result<Option<RepoLockGuard>> {
for attempt in 0..30 {
if let Some(g) = self.try_lock_repo_on(pool, owner_did, repo_name).await? {
return Ok(Some(g));
}
if attempt < 29 {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
}
}
Ok(None)
}
/// Compute the local disk path and owner slug for a repo.
///
/// Three-layer defence against path traversal:
@@ -1122,12 +1003,6 @@ pub(crate) fn claim_fork_disk_path(disk_path: &Path, attempt: &PublishAttemptId)
let Some(path) = fork_attempt_path(disk_path) else {
return;
};
// The stamp may be claimed BEFORE the mirror clone starts, when the owner
// directory does not exist yet; create it so the claim is not silently lost
// (a missing stamp makes every later ownership check refuse to clean).
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
if let Err(e) = std::fs::write(&path, attempt.as_str()) {
warn!(
path = %disk_path.display(),
@@ -1673,61 +1548,6 @@ fn invalidate_local_write_cache(local_path: &Path, repo_name: &str, reason: &str
}
}
impl RepoStore {
/// Refresh the local copy from the authoritative object-store archive so an
/// emptiness recheck reflects remote state (used by purge-spam on a Tigris
/// deployment, where the admin node's local disk can be stale). Downloads
/// the archive over the local path when it exists; a no-op when no object
/// store is configured (single-machine). Errors propagate so the caller can
/// fail closed rather than delete on a stale-local view.
/// Returns the archive generation the refresh verified (its ETag), so the
/// caller can fence a later delete on exactly the view it checked.
pub async fn refresh_from_archive(
&self,
owner_did: &str,
repo_name: &str,
) -> Result<Option<String>> {
let Some(ref store) = self.tigris else {
return Ok(None);
};
let (owner_slug, local_path) = self.local_path(owner_did, repo_name)?;
let Some(generation) = store.head_generation(&owner_slug, repo_name).await? else {
return Ok(None);
};
store
.download(&owner_slug, repo_name, &local_path, None)
.await?;
Ok(generation.etag)
}
/// Delete the object-store archive for a repo. A no-op when no object store
/// is configured. Used by purge-spam so a deleted repo's archive cannot be
/// downloaded into a later repo created with the same owner/name. The
/// delete is fenced on the generation the store sees, so a cross-node
/// publish racing the purge makes it fail rather than erasing the
/// successor's archive.
/// `expected` is the generation the caller verified empty (an ETag from
/// [`refresh_from_archive`]); a moved generation means a PUT landed after
/// the check and the archive must be kept, so `Ok(false)` is a skip.
pub async fn delete_archive(
&self,
owner_did: &str,
repo_name: &str,
expected: Option<&str>,
) -> Result<bool> {
let Some(ref store) = self.tigris else {
return Ok(true);
};
let (owner_slug, _local_path) = self.local_path(owner_did, repo_name)?;
store.delete_current(&owner_slug, repo_name, expected).await
}
/// The bound applied to transfers that run while a repo lock is held.
pub(crate) fn transfer_bound(&self) -> std::time::Duration {
self.lock_held_transfer_timeout
}
}
/// Strict allowlist validator for `owner_did` and `repo_name`.
///
/// Rejects any character that isn't explicitly safe, plus length and
@@ -1884,7 +1704,7 @@ fn validate_owner_did(owner_did: &str) -> Result<()> {
Ok(())
}
pub(crate) fn validate_repo_name(repo_name: &str) -> Result<()> {
fn validate_repo_name(repo_name: &str) -> Result<()> {
if repo_name.is_empty() {
anyhow::bail!("repo_name is empty");
}
@@ -2935,41 +2755,6 @@ pub fn build_lock_pool(source: &PgPool, max_connections: u32, acquire_timeout: D
.connect_lazy_with((*source.connect_options()).clone())
}
/// Lock-only guard from [`RepoStore::try_lock_repo`]. Holds the Postgres advisory
/// lock (and the dedicated connection it lives on) until `release()`. No Tigris
/// I/O — unlike [`RepoWriteGuard`], releasing does NOT upload anything.
pub struct RepoLockGuard {
conn: Option<sqlx::pool::PoolConnection<sqlx::Postgres>>,
lock_key: i64,
repo_name: String,
}
impl RepoLockGuard {
/// Release the advisory lock on the same connection that took it, then return
/// the connection to the pool.
pub async fn release(mut self) {
if let Some(mut conn) = self.conn.take() {
let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
.bind(self.lock_key)
.execute(&mut *conn)
.await;
}
}
}
impl Drop for RepoLockGuard {
fn drop(&mut self) {
// Same guarantee as RepoWriteGuard: a lock guard dropped without
// release() closes its connection so Postgres frees the advisory lock,
// rather than returning a lock-holding connection to the pool.
if let Some(mut conn) = self.conn.take() {
warn!(repo = %self.repo_name,
"RepoLockGuard dropped without release() — closing its connection to free the advisory lock");
conn.close_on_drop();
}
}
}
/// Compute a stable i64 hash for a Postgres advisory lock key.
///
/// Uses SHA-256 (not `DefaultHasher`) so the same `(owner_slug, repo_name)`
@@ -2995,7 +2780,7 @@ pub(crate) fn advisory_lock_key(owner_slug: &str, repo_name: &str) -> i64 {
i64::from_le_bytes(digest[..8].try_into().expect("sha256 output is >= 8 bytes"))
}
#[cfg(test)]
pub(crate) mod tests {
mod tests {
use super::*;
use crate::git::tigris::ATTEMPT_METADATA_KEY;
use std::collections::HashMap;
@@ -5132,7 +4917,7 @@ pub(crate) mod tests {
}
/// Seed the minimum bare-repo shape so the upload has something to archive.
pub(crate) fn seed_bare_repo(path: &Path) {
fn seed_bare_repo(path: &Path) {
std::fs::create_dir_all(path.join("objects/info")).unwrap();
std::fs::create_dir_all(path.join("refs/heads")).unwrap();
std::fs::write(path.join("HEAD"), "ref: refs/heads/main\n").unwrap();
@@ -5295,7 +5080,7 @@ pub(crate) mod tests {
/// models the real S3 arm we care about (body fully transmitted, commit
/// decided later) with no timing in it.
#[derive(Clone, Debug)]
pub(crate) struct CapturedPut {
struct CapturedPut {
key: String,
body: Vec<u8>,
if_match: Option<String>,
@@ -5305,7 +5090,7 @@ pub(crate) mod tests {
/// One stored object.
#[derive(Clone, Debug)]
pub(crate) struct MockObject {
struct MockObject {
body: Vec<u8>,
etag: String,
/// The `x-amz-meta-gitlawb-attempt` it was written with. This is what
@@ -5318,14 +5103,14 @@ pub(crate) mod tests {
/// `status` is `None` while a PUT is parked: it arrived and was logged, but
/// no precondition has been evaluated for it yet.
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct PutAttempt {
struct PutAttempt {
if_match: Option<String>,
if_none_match: Option<String>,
status: Option<u16>,
}
#[derive(Default)]
pub(crate) struct MockState {
struct MockState {
/// Keyed by request path. A single-slot store was enough while every test
/// drove one repo; fork creation touches the SOURCE key and the FORK key
/// in one request, and collapsing them would make an assertion about one
@@ -5360,7 +5145,7 @@ pub(crate) mod tests {
///
/// The fence tests downstream are only worth anything if a precondition can
/// actually fail here, so this helper carries its own semantics tests below.
pub(crate) struct S3Mock {
struct S3Mock {
endpoint: String,
state: Arc<std::sync::Mutex<MockState>>,
gate: Arc<tokio::sync::Notify>,
@@ -5430,7 +5215,7 @@ pub(crate) mod tests {
}
impl S3Mock {
pub(crate) async fn start() -> Self {
async fn start() -> Self {
use axum::response::IntoResponse;
let state = Arc::new(std::sync::Mutex::new(MockState::default()));
@@ -5639,7 +5424,7 @@ pub(crate) mod tests {
}
}
pub(crate) fn endpoint(&self) -> &str {
fn endpoint(&self) -> &str {
&self.endpoint
}
@@ -5660,7 +5445,7 @@ pub(crate) mod tests {
}
/// The stored object under a specific repo key.
pub(crate) fn object_for(&self, owner_slug: &str, repo_name: &str) -> Option<MockObject> {
fn object_for(&self, owner_slug: &str, repo_name: &str) -> Option<MockObject> {
let key = format!("test-bucket/repos/v1/{owner_slug}/{repo_name}.tar.zst");
self.state.lock().unwrap().objects.get(&key).cloned()
}
@@ -5674,7 +5459,7 @@ pub(crate) mod tests {
/// to pick up a precondition and then PUTs on it is therefore fencing
/// on a generation that is already gone, which is the only way to drive
/// two consecutive lost preconditions deterministically.
pub(crate) fn roll_generation_after_next_heads(&self, n: u32) {
fn roll_generation_after_next_heads(&self, n: u32) {
self.state.lock().unwrap().roll_after_heads = n;
}
@@ -5693,7 +5478,7 @@ pub(crate) mod tests {
/// Deliver the next PUT in full and then fail it WITHOUT committing. The
/// client's knowledge is identical to the arm above — that is the whole
/// point — so only asking the store can tell the two apart.
pub(crate) fn fail_next_put_after_delivery(&self) {
fn fail_next_put_after_delivery(&self) {
self.state.lock().unwrap().lose_next_response = Some(false);
}
@@ -5704,7 +5489,7 @@ pub(crate) mod tests {
/// Fail the next HEAD for ONE repo key with a 500, so a caller's
/// reconciliation cannot be answered either way.
pub(crate) fn fail_next_head_for(&self, owner_slug: &str, repo_name: &str) {
fn fail_next_head_for(&self, owner_slug: &str, repo_name: &str) {
let key = format!("test-bucket/repos/v1/{owner_slug}/{repo_name}.tar.zst");
*self
.state
@@ -5716,7 +5501,7 @@ pub(crate) mod tests {
}
/// How many DELETEs the mock actually carried out.
pub(crate) fn deletes(&self) -> u32 {
fn deletes(&self) -> u32 {
self.state.lock().unwrap().deletes
}
@@ -5746,7 +5531,7 @@ pub(crate) mod tests {
.0
}
pub(crate) fn shutdown(&self) {
fn shutdown(&self) {
self.server.abort();
}
}
@@ -6477,25 +6262,25 @@ pub(crate) mod tests {
/// A client aimed at the same key the store under test publishes to, so a
/// test can seed the archive or land an interfering publish of its own.
pub(crate) fn mock_tigris(mock: &S3Mock) -> TigrisClient {
fn mock_tigris(mock: &S3Mock) -> TigrisClient {
TigrisClient::for_testing_with_endpoint("test-bucket", mock.endpoint())
}
/// The slug `local_path` derives from a DID, needed because a test seeds
/// and reads the archive key directly.
pub(crate) fn owner_slug_of(owner_did: &str) -> String {
fn owner_slug_of(owner_did: &str) -> String {
owner_did.replace([':', '/'], "_")
}
/// A bare-repo-shaped directory carrying `marker`, so a test can tell whose
/// tree is stored without comparing compressed bytes.
pub(crate) fn marked_repo(path: &Path, marker: &str) {
fn marked_repo(path: &Path, marker: &str) {
seed_bare_repo(path);
std::fs::write(path.join("MARKER"), marker).unwrap();
}
/// The marker inside whatever archive is currently stored under the key.
pub(crate) async fn stored_marker(mock: &S3Mock, owner_slug: &str, repo_name: &str) -> String {
async fn stored_marker(mock: &S3Mock, owner_slug: &str, repo_name: &str) -> String {
let out = TempDir::new().unwrap();
let validated = super::validated_repo_disk_path(out.path(), "did:key:stored", "stored")
.expect("test repo path must validate");
+12 -10
View File
@@ -683,12 +683,14 @@ pub fn list_refs(repo_path: &Path) -> Result<Vec<(String, String)>> {
/// a 40-char SHA-1 hex string. The reconcile compares two hex
/// strings.
///
/// Production callers use [`read_ref_bounded`]; this unbounded form is kept
/// for test harnesses that stage markers without a timeout budget.
#[allow(dead_code)]
/// `show-ref --verify` matches the exact ref name — unlike `rev-parse`,
/// which peels expressions, so it is the right primitive when the name is
/// caller-adjacent. Most production callers use [`read_ref_bounded`]; this
/// unbounded form also serves test harnesses that stage markers without a
/// timeout budget.
pub fn read_ref(repo_path: &Path, ref_name: &str) -> Result<Option<String>> {
// allow-unbounded-git: #[allow(dead_code)] harness form; production
// callers use read_ref_bounded.
// allow-unbounded-git: exact-name resolution; callers that need a timeout
// budget use read_ref_bounded.
let output = Command::new("git")
.args(["show-ref", "--verify", "--hash", ref_name])
.current_dir(repo_path)
@@ -1577,7 +1579,7 @@ fn guard_refs(target_branch: &str, source_branch: &str) -> Result<()> {
/// Stored short branch names are passed to git as explicit local-branch refs so a
/// syntax-valid name cannot be reinterpreted as a revision shorthand at the sink.
fn local_branch_ref(short_name: &str) -> String {
pub(crate) fn local_branch_ref(short_name: &str) -> String {
format!("refs/heads/{short_name}")
}
@@ -1817,10 +1819,6 @@ pub fn repo_disk_path(repos_dir: &Path, owner_did: &str, repo_name: &str) -> std
#[cfg(test)]
mod tests {
use super::{branch_diff_names, merge_branch};
use std::path::Path;
use std::process::Command;
use super::{validate_git_ref, validate_git_ref_with_git, GitRefValidationError};
#[test]
@@ -1887,6 +1885,10 @@ mod tests {
);
}
use super::{branch_diff_names, merge_branch};
use std::path::Path;
use std::process::Command;
/// #26 split 1/4: a bare repo this node creates must KEEP REFLOGS, because
/// the durable-outbox reconcile has no other way to prove that a stranded
/// transition actually landed. `core.logAllRefUpdates` defaults to false for
-43
View File
@@ -522,49 +522,6 @@ impl TigrisClient {
}
}
/// Delete whatever archive currently exists, fenced on the generation read
/// at call time: a concurrent publish between the HEAD and the DELETE makes
/// the store refuse rather than erasing the successor's object. Used by
/// purge-spam, which deletes a repo it did not publish (so
/// [`delete_if_attempt_matches`](Self::delete_if_attempt_matches) cannot
/// apply) while still needing the same moved-generation protection.
/// Returns `true` when an archive was deleted.
/// Delete only if the object's generation still matches `expected` — the
/// generation the caller checked for emptiness. A PUT landing between that
/// check and this delete makes the generation move; deleting the newer
/// (possibly non-empty) archive is exactly the outcome the fence exists to
/// prevent, so a moved generation refuses rather than deletes.
pub async fn delete_current(
&self,
owner_slug: &str,
repo_name: &str,
expected: Option<&str>,
) -> Result<bool> {
let Some(generation) = self.head_generation(owner_slug, repo_name).await? else {
return Ok(false);
};
if generation.etag.as_deref() != expected {
return Ok(false);
}
let key = Self::repo_key(owner_slug, repo_name);
let mut req = self.s3.delete_object().bucket(&self.bucket).key(&key);
if let Some(etag) = generation.etag.as_deref() {
req = req.if_match(etag);
}
match req.send().await {
Ok(_) => Ok(true),
Err(e) => {
if e.raw_response()
.map(|raw| raw.status().as_u16())
.is_some_and(|s| s == 412 || s == 409)
{
anyhow::bail!("archive generation moved between head and delete; refusing");
}
Err(anyhow::anyhow!("tigris DELETE {key}: {e}"))
}
}
}
/// Download a repo archive from Tigris and extract to local disk.
pub async fn download(
&self,
-29
View File
@@ -14,11 +14,6 @@ use subscription::SubscriptionRoot;
pub type GitlawbSchema = Schema<QueryRoot, MutationRoot, SubscriptionRoot>;
/// No mutations: the /graphql/ws transport keeps queries and subscriptions but
/// has no mutation root, so the socket can never be used to run a write and
/// sidestep the write-brake bucket the HTTP route pays.
pub type GitlawbWsSchema = Schema<QueryRoot, async_graphql::EmptyMutation, SubscriptionRoot>;
// Keep ordinary schema discovery and application queries usable while
// bounding validation work and the number of resolver selections per request.
const GRAPHQL_MAX_COMPLEXITY: usize = 400;
@@ -176,30 +171,6 @@ pub fn build_schema(
.finish()
}
/// Build the subscriptions-only schema served over /graphql/ws. Carries the
/// same broadcast senders the subscription resolvers consume. Mutations are
/// disabled — the schema's mutation root is `EmptyMutation` — but queries still
/// run via `QueryRoot` (they are reads and already ride the TaskReadBrake), so
/// "subscriptions-only" refers to the write surface, not the operation set.
pub fn build_ws_schema(
db: Arc<Db>,
ref_update_tx: tokio::sync::broadcast::Sender<RefUpdateBroadcast>,
task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
task_cursor_key: crate::api::task_cursor::TaskCursorKey,
) -> GitlawbWsSchema {
apply_query_limits(Schema::build(
QueryRoot,
async_graphql::EmptyMutation,
SubscriptionRoot,
))
.data(db)
.data(ref_update_tx)
.data(task_event_tx)
.data(task_cursor_key)
.extension(TaskReadBrakeExtension)
.finish()
}
#[cfg(test)]
mod tests {
use super::*;
-29
View File
@@ -187,11 +187,6 @@ where
F: std::future::Future<Output = Result<T>>,
{
let left = deadline.saturating_duration_since(Instant::now());
// `timeout` polls the future before it checks the timer, so a zero budget
// would still start the work. A spent deadline must not.
if left.is_zero() {
return Err(BoundedDbError::Elapsed);
}
match tokio::time::timeout(left, fut).await {
Ok(Ok(v)) => Ok(v),
Ok(Err(e)) => Err(BoundedDbError::Db(e)),
@@ -3438,30 +3433,6 @@ mod tests {
);
}
/// A spent deadline must not start the work at all. `tokio::time::timeout`
/// polls the inner future before it checks the timer, so a future that is
/// ready on its first poll would otherwise slip through a zero budget and
/// return a value. Deterministic: the future has no await point.
#[tokio::test]
async fn db_bounded_spent_deadline_does_not_poll_a_ready_future() {
let spent = Instant::now() - Duration::from_secs(5);
let polled = std::sync::atomic::AtomicBool::new(false);
let out: Result<u8, BoundedDbError> = db_bounded(spent, async {
polled.store(true, std::sync::atomic::Ordering::SeqCst);
Ok(7u8)
})
.await;
assert!(
matches!(out, Err(BoundedDbError::Elapsed)),
"a spent deadline must yield the timeout arm even for a ready future: {out:?}"
);
assert!(
!polled.load(std::sync::atomic::Ordering::SeqCst),
"a spent deadline must not start the work"
);
}
/// U5 (#173): the elapsed arm of the discovery record leaves neither row behind.
///
/// What this proves and what it does NOT: it shows the wrapper composition
+28 -176
View File
@@ -3,7 +3,6 @@
// migrations) so the out-of-crate deny-harness integration crate can spawn a
// real node; `src/main.rs` is a thin shim over `run()`. Modules the harness
// reaches are `pub`; the rest stay crate-private.
mod admin;
mod ans104;
pub mod api;
mod arweave;
@@ -85,12 +84,6 @@ pub async fn run() -> Result<()> {
let mut config = Config::parse();
// Admin subcommands run out-of-band and exit, never starting the node. The
// no-subcommand path below is the unchanged daemon startup.
if let Some(command) = config.command.clone() {
return run_admin_command(command, &config).await;
}
// Merge the embedded seed list of public network nodes into the runtime
// bootstrap peers. Operators can opt out via GITLAWB_BOOTSTRAP_DISABLE_SEEDS.
bootstrap::merge_seeds(&mut config);
@@ -334,12 +327,6 @@ pub async fn run() -> Result<()> {
task_event_tx.clone(),
task_cursor_key.clone(),
));
let graphql_ws_schema = Arc::new(graphql::build_ws_schema(
Arc::clone(&db),
ref_update_tx.clone(),
task_event_tx.clone(),
task_cursor_key.clone(),
));
let machine_id = std::env::var("FLY_MACHINE_ID").ok();
if let Some(ref mid) = machine_id {
@@ -395,26 +382,36 @@ pub async fn run() -> Result<()> {
// capped regardless of how many identities it mints. Sized well above any
// legitimate per-IP creation rate; GITLAWB_CREATE_RATE_LIMIT overrides, 0
// disables. Bounded key set — the key is a client-influenced IP.
let create_ip_rate_limiter = build_ip_limiter("GITLAWB_CREATE_RATE_LIMIT", 120, "creation");
// Per-IP brake for the authenticated non-creation write surface (issue/PR
// comments, issue close, labels, stars, merges, protect, replicas, webhook
// create/delete, visibility, tasks, bounties, profile, every HTTP /graphql
// request). Issue comment/close debit it in the handler, after the read
// gate; the rest via the router-level brake. Its own bucket, separate from the
// creation brake, so a write flood cannot drain the creation budget and vice
// versa (same rationale as the sync_trigger / peer_write split). Sized above
// any legitimate per-IP write rate — real agent automation makes many small
// writes per hour. GITLAWB_WRITE_RATE_LIMIT overrides; 0 disables. Bounded
// key set — the key is a client-influenced IP.
let write_rate_limiter = build_ip_limiter("GITLAWB_WRITE_RATE_LIMIT", 600, "write");
let create_limit = std::env::var("GITLAWB_CREATE_RATE_LIMIT")
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.unwrap_or(120);
let create_ip_rate_limiter = rate_limit::RateLimiter::new_bounded(
create_limit,
std::time::Duration::from_secs(3600),
200_000,
);
if create_limit == 0 {
tracing::warn!("GITLAWB_CREATE_RATE_LIMIT=0 — per-IP creation rate limiting disabled");
}
// Push-path flood brake: max git-receive-pack requests per client IP per
// hour (counts both the info/refs advertisement and the push POST). Sized
// for heavy agent automation while still stopping flood traffic (the June
// 2026 attack pushed several times per second per IP). GITLAWB_PUSH_RATE_LIMIT
// overrides; 0 disables. Bounded key set — the key is a client-influenced IP.
let push_rate_limiter = build_ip_limiter("GITLAWB_PUSH_RATE_LIMIT", 600, "push");
let push_limit = std::env::var("GITLAWB_PUSH_RATE_LIMIT")
.ok()
.and_then(|v| v.trim().parse::<usize>().ok())
.unwrap_or(600);
let push_rate_limiter = rate_limit::RateLimiter::new_bounded(
push_limit,
std::time::Duration::from_secs(3600),
200_000,
);
if push_limit == 0 {
tracing::warn!("GITLAWB_PUSH_RATE_LIMIT=0 — per-IP push rate limiting disabled");
}
// close_issue drives a full archive download for non-owner author pre-checks.
// Keep it on its own bucket so it cannot drain the receive-pack limit.
@@ -439,7 +436,7 @@ pub async fn run() -> Result<()> {
let push_limiter_trust = rate_limit::TrustedProxy::from_env_value(
&std::env::var("GITLAWB_TRUSTED_PROXY").unwrap_or_default(),
);
tracing::info!(trust = ?push_limiter_trust, "push rate limiter trust configured");
tracing::info!(trust = ?push_limiter_trust, push_limit, "push rate limiter configured");
// Peer-sync flood brakes, keyed on the resolved client IP (per-DID is useless
// here — a did:key farm self-registers). Two buckets so an unsigned notify
@@ -480,12 +477,10 @@ pub async fn run() -> Result<()> {
ref_update_tx,
task_event_tx,
graphql_schema,
graphql_ws_schema,
machine_id,
repo_store,
rate_limiter,
create_ip_rate_limiter,
write_rate_limiter,
push_rate_limiter,
close_issue_rate_limiter,
ipfs_max_history_walks: crate::api::ipfs::MAX_HISTORY_WALKS_PER_REQUEST,
@@ -635,17 +630,14 @@ pub async fn run() -> Result<()> {
// Periodic cleanup of expired rate limit entries + consumed-proof ledger
{
// Clone the whole state so the sweep uses AppState::cleanup_rate_limiters
// (the single source of truth that reaps EVERY limiter, including
// write_rate_limiter) rather than a hand-maintained list that can omit one.
let state = state.clone();
let sweep_state = state.clone();
let db = state.db.clone();
let mut shutdown_rx = state.subscribe_shutdown();
tokio::spawn(async move {
loop {
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(300)) => {
state.cleanup_rate_limiters().await;
sweep_state.sweep_rate_limiters().await;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
@@ -892,110 +884,6 @@ pub async fn run() -> Result<()> {
Ok(())
}
/// Pure resolver for a usize rate-limit env value. Returns `(resolved, invalid)`:
/// an absent or empty value resolves to `default` and is NOT invalid; a non-empty
/// value that fails to parse resolves to `default` and IS invalid (so the caller
/// warns rather than silently disabling the intended, usually stricter, brake).
fn resolve_rate_limit(raw: Option<&str>, default: usize) -> (usize, bool) {
match raw.map(str::trim) {
None | Some("") => (default, false),
Some(t) => match t.parse::<usize>() {
Ok(n) => (n, false),
Err(_) => (default, true),
},
}
}
/// Read a rate-limit env var, warning on a present-but-unparseable value. A typo
/// like `GITLAWB_WRITE_RATE_LIMIT=5O` must not silently revert to the default and
/// leave the operator believing a stricter cap is in force.
fn rate_limit_from_env(var: &str, default: usize) -> usize {
let raw = std::env::var(var).ok();
let (value, invalid) = resolve_rate_limit(raw.as_deref(), default);
if invalid {
tracing::warn!(
var,
value = raw.as_deref().unwrap_or(""),
default,
"invalid rate-limit value — using default; the intended brake is NOT applied"
);
}
value
}
/// Build a bounded per-client-IP rate limiter from an env var: resolve the limit
/// (honoring the present-but-unparseable warning in `rate_limit_from_env`), build
/// the limiter with the shared 1-hour window and 200k-key cap, and warn when the
/// limit is 0 (disabled). Collapses the identical create/write/push setup.
fn build_ip_limiter(var: &str, default: usize, label: &str) -> rate_limit::RateLimiter {
let limit = rate_limit_from_env(var, default);
let limiter =
rate_limit::RateLimiter::new_bounded(limit, std::time::Duration::from_secs(3600), 200_000);
if limit == 0 {
tracing::warn!("{var}=0 — per-IP {label} rate limiting disabled");
} else {
tracing::info!(%var, limit, "per-IP {label} rate limiter configured");
}
limiter
}
/// Dispatch an admin subcommand. Connects the database directly (no server, no
/// p2p, no metrics) and runs the requested tool to completion, then exits.
async fn run_admin_command(command: config::Command, config: &Config) -> Result<()> {
match command {
config::Command::PurgeSpam { execute } => {
let acquire_timeout = std::time::Duration::from_secs(config.db_acquire_timeout_secs);
let db = Db::connect(
&config.database_url,
config.db_max_connections,
acquire_timeout,
)
.await
.context("connecting to database for purge-spam")?;
// Build the object store so purge is authoritative against remote
// state: the emptiness recheck consults the archive (not just
// possibly-stale local disk) and a successful purge deletes the
// archive too. When no bucket is configured this stays None and purge
// is local-only, exactly as before.
let object_store: Option<git::tigris::TigrisClient> = if config.tigris_bucket.is_empty()
{
None
} else {
match git::tigris::TigrisClient::new(&config.tigris_bucket).await {
Ok(client) => Some(client),
Err(e) => {
// Fail closed: without the configured store we cannot
// do the authoritative recheck, and a local-only purge
// on a Tigris deployment is the stale-view delete this
// guards against. Refuse to run rather than fall back.
anyhow::bail!(
"purge-spam: GITLAWB_TIGRIS_BUCKET is set but the object-store client failed to initialize ({e}); refusing to run a local-only purge on a Tigris deployment"
);
}
}
};
// Give the purge-spam store a dedicated advisory-lock pool, separate
// from `db`'s app pool. The guard then holds a lock-pool connection
// while `delete_repo_by_id` runs on the app pool, so a purge at
// GITLAWB_DB_MAX_CONNECTIONS=1 does not self-deadlock (KTD3).
let lock_pool = Db::lock_pool(
&config.database_url,
config.db_lock_pool_max_connections,
acquire_timeout,
)?;
let repo_store = git::repo_store::RepoStore::new(
config.repos_dir.clone(),
object_store,
lock_pool,
std::time::Duration::from_secs(config.lock_held_transfer_timeout_secs),
);
admin::run_purge_spam(&db, &repo_store, &config.repos_dir, execute)
.await
.map(|_| ())
}
}
}
/// U4 (#173): spawn the periodic legacy provider-CID repair sweep. Releases before this
/// version stored the PROVIDER CID (Kubo dag-pb / Pinata CIDv0) in `pinned_cids.cid`,
/// and this version's `/ipfs/{cid}` resolver withholds any row whose stored key is not
@@ -1597,7 +1485,6 @@ mod rate_limiter_sweep_tests {
let mut state = crate::test_support::test_state_lazy();
state.rate_limiter = RateLimiter::new(10, window);
state.create_ip_rate_limiter = RateLimiter::new(10, window);
state.write_rate_limiter = RateLimiter::new(10, window);
state.push_rate_limiter = RateLimiter::new(10, window);
state.close_issue_rate_limiter = RateLimiter::new(10, window);
state.sync_trigger_rate_limiter = RateLimiter::new(10, window);
@@ -1611,7 +1498,6 @@ mod rate_limiter_sweep_tests {
[
s.rate_limiter.clone(),
s.create_ip_rate_limiter.clone(),
s.write_rate_limiter.clone(),
s.push_rate_limiter.clone(),
s.close_issue_rate_limiter.clone(),
s.sync_trigger_rate_limiter.clone(),
@@ -1628,7 +1514,7 @@ mod rate_limiter_sweep_tests {
}
tokio::time::sleep(window * 3).await;
state.cleanup_rate_limiters().await;
state.sweep_rate_limiters().await;
for (i, l) in limiters(&state).into_iter().enumerate() {
assert_eq!(l.tracked_keys().await, 0, "limiter {i} was not swept");
@@ -1839,40 +1725,6 @@ async fn gossip_ping_round(
}
}
#[cfg(test)]
mod rate_limit_env_tests {
use super::resolve_rate_limit;
#[test]
fn absent_uses_default_without_warning() {
assert_eq!(resolve_rate_limit(None, 600), (600, false));
}
#[test]
fn empty_or_whitespace_uses_default_without_warning() {
assert_eq!(resolve_rate_limit(Some(""), 600), (600, false));
assert_eq!(resolve_rate_limit(Some(" "), 600), (600, false));
}
#[test]
fn valid_value_parses() {
assert_eq!(resolve_rate_limit(Some("50"), 600), (50, false));
assert_eq!(resolve_rate_limit(Some(" 50 "), 600), (50, false));
// 0 is a VALID value (disables the brake); the ==0 warning is handled
// separately at the call site. It must NOT be flagged invalid here.
assert_eq!(resolve_rate_limit(Some("0"), 600), (0, false));
}
#[test]
fn present_but_unparseable_defaults_and_flags_invalid() {
// The L8 case: a fat-fingered stricter cap ("5O", letter O) must default
// AND be flagged so the caller warns, not silently disable the brake.
assert_eq!(resolve_rate_limit(Some("5O"), 600), (600, true));
assert_eq!(resolve_rate_limit(Some("abc"), 600), (600, true));
assert_eq!(resolve_rate_limit(Some("-1"), 600), (600, true));
}
}
#[cfg(test)]
mod gossip_ssrf_tests {
use super::{
+21 -309
View File
@@ -70,21 +70,18 @@ impl RateLimiter {
self.window.min(Duration::from_secs(1))
}
/// Returns `true` if the request is allowed. Thin wrapper over
/// [`check_retry`](Self::check_retry) for callers that only need the
/// allow/deny decision, not the rejection's Retry-After delay.
pub(crate) async fn check(&self, key: &str) -> bool {
self.check_retry(key).await.is_none()
/// The configured window, exposed so a 429 can carry an honest
/// `retry-after`: a hardcoded value lies when the operator retunes the
/// window, and a client that honors the header would retry into another
/// 429 until the real window elapses.
pub fn window(&self) -> Duration {
self.window
}
/// Record a request against `key`. Returns `None` when it is allowed, or
/// `Some(retry_after)` when it is rejected — the delay a client should wait
/// before retrying, so a 429 can advertise a value consistent with the
/// actual window instead of a constant.
pub(crate) async fn check_retry(&self, key: &str) -> Option<Duration> {
pub(crate) async fn check(&self, key: &str) -> bool {
// max_requests == 0 means the limiter is disabled, not "block all".
if self.max_requests == 0 {
return None;
return true;
}
let now = Instant::now();
let mut state = self.state.lock().await;
@@ -96,25 +93,10 @@ impl RateLimiter {
.timestamps
.retain(|t| now.duration_since(*t) < self.window);
if window.timestamps.len() >= self.max_requests {
// Insertion order is preserved, so after the retain the first
// element is the oldest live request. It ages out (freeing a
// slot) one window after it landed, so that remaining time is
// the delay to advertise. Clamp to >= 1 so a 429 never tells a
// client to retry immediately.
let retry = window
.timestamps
.first()
.map(|oldest| {
self.window
.as_secs()
.saturating_sub(now.duration_since(*oldest).as_secs())
})
.unwrap_or(0)
.max(1);
return Some(Duration::from_secs(retry));
return false;
}
window.timestamps.push(now);
return None;
return true;
}
// New key. Enforce the cap BEFORE inserting so a flood of distinct keys
@@ -135,11 +117,7 @@ impl RateLimiter {
}
drop(last_sweep);
if state.len() >= self.max_keys {
// The key is untracked (never inserted), so there is no oldest
// entry to derive a delay from. Advertise the full window as a
// conservative bound: capacity frees as other keys expire, at
// the latest one window from now.
return Some(self.window.max(Duration::from_secs(1)));
return false;
}
}
state.insert(
@@ -148,23 +126,7 @@ impl RateLimiter {
timestamps: vec![now],
},
);
None
}
/// Give back one slot a prior [`check`](Self::check) took for `key`, for a
/// caller that debits two buckets and must undo the first when the second
/// refuses. Drops the newest timestamp: the count is what the cap reads, so
/// that is equivalent to dropping the caller's own even if another request
/// landed in between. An unknown key is a no-op, and a key left with no
/// timestamps is removed so a refund never holds a map entry.
pub(crate) async fn refund(&self, key: &str) {
let mut state = self.state.lock().await;
if let Some(window) = state.get_mut(key) {
window.timestamps.pop();
if window.timestamps.is_empty() {
state.remove(key);
}
}
true
}
/// Non-consuming check: is this key ALREADY at its limit for the current window?
@@ -216,19 +178,6 @@ impl RateLimiter {
});
}
/// Test-only: insert a fully-expired entry (no live timestamps) that the next
/// `cleanup()` must reclaim. Models a key whose window has aged out, without
/// depending on wall-clock sleeps or a short window. Used cross-module by the
/// AppState cleanup guard test (H2).
#[cfg(test)]
pub(crate) async fn insert_reclaimable_for_test(&self, key: &str) {
self.state
.lock()
.await
.insert(key.to_string(), Window { timestamps: vec![] });
}
/// Test-only: number of distinct keys currently tracked.
/// Number of distinct keys currently tracked. Test-only introspection so a
/// cross-module test can assert that a sweep actually evicted expired entries
/// and observe what it reclaimed. There is no production reader.
@@ -236,23 +185,6 @@ impl RateLimiter {
pub(crate) async fn tracked_keys(&self) -> usize {
self.state.lock().await.len()
}
/// Test-only: seed a key's bucket with entries of the given ages (how long
/// ago each request landed), so a test can drive the Retry-After computation
/// off a controlled oldest-entry age without wall-clock sleeps. Ages that
/// would underflow the monotonic clock collapse to `now`.
#[cfg(test)]
pub(crate) async fn seed_aged_entry_for_test(&self, key: &str, ages: &[Duration]) {
let now = Instant::now();
let timestamps = ages
.iter()
.map(|a| now.checked_sub(*a).unwrap_or(now))
.collect();
self.state
.lock()
.await
.insert(key.to_string(), Window { timestamps });
}
}
/// Per-source concurrency cap derived from the write-pool size: one resolved client
@@ -368,8 +300,8 @@ pub async fn rate_limit_by_did(request: Request, next: Next) -> Response {
.map(|a| a.0.clone());
if let (Some(limiter), Some(did)) = (limiter, did) {
if let Some(retry_after) = limiter.check_retry(&did).await {
return too_many_requests(retry_after);
if !limiter.check(&did).await {
return too_many_requests(limiter.window().as_secs());
}
}
@@ -472,21 +404,18 @@ impl<S: Send + Sync> axum::extract::FromRequestParts<S> for PeerAddr {
}
}
/// The shared 429 response for the rate-limit middlewares (per-DID and per-IP
/// alike). Route-agnostic: the message stays generic (the offending path, when
/// logged, is recorded at the call site).
/// The message every per-IP brake answers with, shared so the GraphQL surface
/// (which cannot return a 429 status inside a 200 GraphQL envelope) says the
/// same thing as the REST routes.
pub const RATE_LIMIT_MESSAGE: &str = "rate limit exceeded — try again later";
/// The shared 429 response for the rate-limit middlewares (per-DID and per-IP
/// alike). Route-agnostic: the message stays generic (the offending path, when
/// logged, is recorded at the call site).
pub fn too_many_requests(retry_after: Duration) -> Response {
/// The shared 429 response for the per-IP flood brakes. Route-agnostic: this
/// middleware now serves the push path AND the peer-sync routes, so the message
/// stays generic (the offending path is recorded in the warn log below).
pub fn too_many_requests(retry_after_secs: u64) -> Response {
(
StatusCode::TOO_MANY_REQUESTS,
[("retry-after", retry_after.as_secs().to_string())],
[("retry-after", retry_after_secs.to_string())],
RATE_LIMIT_MESSAGE,
)
.into_response()
@@ -553,9 +482,9 @@ pub async fn rate_limit_by_ip(request: Request, next: Next) -> Response {
if let Some(limiter) = limiter {
if let Some(key) = client_key(request.headers(), peer, limiter.trust) {
if let Some(retry_after) = limiter.limiter.check_retry(&key).await {
if !limiter.limiter.check(&key).await {
tracing::warn!(key = %key, path = %request.uri().path(), "per-IP rate limit exceeded");
return too_many_requests(retry_after);
return too_many_requests(limiter.limiter.window().as_secs());
}
}
}
@@ -941,221 +870,4 @@ mod tests {
"an exhausted IP must be braked with 429 before auth runs, not leak to 401"
);
}
// ── Adversarial TrustedProxy verification through the middleware (U5) ──
// These complement the client_key unit tests above by driving hostile
// headers through rate_limit_by_ip on a real router: the property under test
// is that the *bucket key* cannot be rotated by a spoofer.
async fn post_with(
router: &axum::Router,
peer: SocketAddr,
hdrs: &[(&str, &str)],
) -> StatusCode {
use tower::ServiceExt;
let mut b = axum::http::Request::builder()
.method(axum::http::Method::POST)
.uri("/api/v1/repos");
for (k, v) in hdrs {
b = b.header(*k, *v);
}
let mut req = b.body(axum::body::Body::empty()).unwrap();
req.extensions_mut().insert(ConnectInfo(peer));
router.clone().oneshot(req).await.unwrap().status()
}
#[tokio::test]
async fn xff_spoofed_leftmost_hop_cannot_rotate_bucket() {
// XForwardedFor mode, budget 1. A spoofer varies the client-controlled
// leftmost hop every request but the trusted proxy always appends the
// same rightmost hop. All requests must key to the rightmost hop, so the
// second is braked despite a fresh leftmost value.
let router = ip_limited_over_auth_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::XForwardedFor,
});
let p1: SocketAddr = "10.0.0.1:1".parse().unwrap();
let p2: SocketAddr = "10.0.0.2:1".parse().unwrap();
// First request passes the brake, reaches (failing) auth.
assert_eq!(
post_with(&router, p1, &[("x-forwarded-for", "9.9.9.1, 203.0.113.5")]).await,
StatusCode::UNAUTHORIZED
);
// Different leftmost + different socket peer, SAME rightmost trusted hop:
// braked, because the key is the rightmost hop, not the spoofed value.
assert_eq!(
post_with(&router, p2, &[("x-forwarded-for", "9.9.9.2, 203.0.113.5")]).await,
StatusCode::TOO_MANY_REQUESTS,
"a spoofer varying the leftmost XFF hop must not rotate its bucket key"
);
}
#[tokio::test]
async fn xff_distinct_real_clients_get_distinct_buckets() {
// Distinct trusted (rightmost) hops are distinct clients: neither throttles
// the other, so a busy legitimate client cannot starve a different one.
let router = ip_limited_over_auth_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::XForwardedFor,
});
let peer: SocketAddr = "10.0.0.9:1".parse().unwrap();
assert_eq!(
post_with(
&router,
peer,
&[("x-forwarded-for", "1.1.1.1, 203.0.113.5")]
)
.await,
StatusCode::UNAUTHORIZED
);
assert_eq!(
post_with(
&router,
peer,
&[("x-forwarded-for", "1.1.1.1, 203.0.113.6")]
)
.await,
StatusCode::UNAUTHORIZED,
"a different trusted client hop must get its own bucket"
);
}
#[tokio::test]
async fn absent_trusted_header_collapses_to_socket_peer() {
// The documented fallback: in a trusted-proxy mode, a request with no
// forwarded header keys on the socket peer. Behind a real edge every
// request shares the proxy's socket, so they collapse onto one bucket —
// asserted here by name so any change to this behavior is deliberate.
let router = ip_limited_over_auth_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(60)),
trust: TrustedProxy::XForwardedFor,
});
let proxy: SocketAddr = "172.16.0.1:1".parse().unwrap();
assert_eq!(
post_with(&router, proxy, &[]).await,
StatusCode::UNAUTHORIZED
);
assert_eq!(
post_with(&router, proxy, &[]).await,
StatusCode::TOO_MANY_REQUESTS,
"with no trusted header, requests fall back to the socket peer key (collapse)"
);
}
// ── Retry-After derived from the window, not a constant 60 (U5) ───────
// A 429 must advertise the delay until the oldest live request ages out of
// the limiter's window, so the value tracks the real window instead of the
// hardcoded 60 that contradicted every limiter's 3600s window.
async fn resp_from(router: &axum::Router, peer: SocketAddr) -> Response {
use tower::ServiceExt;
let mut req = axum::http::Request::builder()
.method(axum::http::Method::POST)
.uri("/o/r/git-receive-pack")
.body(axum::body::Body::empty())
.unwrap();
req.extensions_mut().insert(ConnectInfo(peer));
router.clone().oneshot(req).await.unwrap()
}
fn retry_after_secs(resp: &Response) -> u64 {
resp.headers()
.get("retry-after")
.expect("429 must carry a retry-after header")
.to_str()
.unwrap()
.parse()
.expect("retry-after must be an integer number of seconds")
}
#[tokio::test]
async fn retry_after_reflects_full_window_for_freshly_filled_bucket() {
// A just-filled bucket's oldest entry is ~now, so the advertised delay
// must be close to the whole window (100s), NOT the old constant 60.
let router = ip_limited_router(IpRateLimiter {
limiter: RateLimiter::new(1, Duration::from_secs(100)),
trust: TrustedProxy::None,
});
let peer: SocketAddr = "203.0.113.50:6000".parse().unwrap();
assert_eq!(resp_from(&router, peer).await.status(), StatusCode::OK);
let resp = resp_from(&router, peer).await;
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
let retry = retry_after_secs(&resp);
assert!(
(95..=100).contains(&retry),
"freshly-filled bucket must advertise ~window (95..=100s), got {retry} (constant-60 bug if 60)"
);
}
#[tokio::test]
async fn retry_after_shrinks_as_oldest_entry_ages() {
// Seed the sole slot with an entry already 95s into a 100s window: the
// next request is rejected and must advertise only the ~5s remaining,
// not 60.
let limiter = RateLimiter::new(1, Duration::from_secs(100));
let peer: SocketAddr = "203.0.113.51:6000".parse().unwrap();
limiter
.seed_aged_entry_for_test(&peer.ip().to_string(), &[Duration::from_secs(95)])
.await;
let router = ip_limited_router(IpRateLimiter {
limiter,
trust: TrustedProxy::None,
});
let resp = resp_from(&router, peer).await;
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
let retry = retry_after_secs(&resp);
assert!(
(1..=10).contains(&retry),
"an aged bucket must advertise the small remaining delay (1..=10s), got {retry}"
);
}
#[tokio::test]
async fn retry_after_clamps_to_at_least_one_second() {
// Oldest entry is 99s into a 100s window: the raw remainder rounds toward
// 1s and must never be advertised as 0 (a 0/blank retry lets a rejected
// client hammer immediately).
let limiter = RateLimiter::new(1, Duration::from_secs(100));
let peer: SocketAddr = "203.0.113.52:6000".parse().unwrap();
limiter
.seed_aged_entry_for_test(&peer.ip().to_string(), &[Duration::from_secs(99)])
.await;
let router = ip_limited_router(IpRateLimiter {
limiter,
trust: TrustedProxy::None,
});
let resp = resp_from(&router, peer).await;
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
let retry = retry_after_secs(&resp);
assert!(retry >= 1, "retry-after must clamp to >= 1, got {retry}");
assert!(
retry <= 2,
"with 99s of a 100s window elapsed, the remaining delay is ~1s, got {retry}"
);
}
/// `refund` gives back exactly one slot a prior `check` took, so a caller
/// that debits two buckets can undo the first when the second refuses.
#[tokio::test]
async fn refund_returns_one_slot_and_is_a_no_op_for_an_unknown_key() {
let limiter = RateLimiter::new(2, Duration::from_secs(60));
assert!(limiter.check("k").await);
assert!(limiter.check("k").await);
assert!(!limiter.check("k").await, "bucket is full");
limiter.refund("k").await;
assert!(limiter.check("k").await, "one slot came back");
assert!(!limiter.check("k").await, "and only one");
// An unknown key must neither panic nor create an entry.
limiter.refund("never-seen").await;
assert!(limiter.state.lock().await.get("never-seen").is_none());
// Refunding the only slot drops the key, so a refunded key does not
// hold a map entry (the limiter's key cap bounds its own state).
let single = RateLimiter::new(1, Duration::from_secs(60));
assert!(single.check("solo").await);
single.refund("solo").await;
assert!(single.state.lock().await.get("solo").is_none());
}
}
+78 -64
View File
@@ -15,7 +15,7 @@ use tracing::Level;
use crate::api::{
agents, arweave, bounties, certs, changelog, events, ipfs, issues, labels, peers, profiles,
protect, pulls, register, replicas, repos, resolve, stars, tasks, visibility, webhooks,
protect, pulls, register, replicas, repos, resolve, stars, status, tasks, visibility, webhooks,
};
use crate::auth;
use crate::rate_limit;
@@ -67,7 +67,7 @@ async fn graphql_ws_handler(
key: rate_limit::client_key(&headers, peer, state.push_limiter_trust),
request_count: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
});
let schema = state.graphql_ws_schema.as_ref().clone();
let schema = state.graphql_schema.as_ref().clone();
upgrade
.protocols(ALL_WEBSOCKET_PROTOCOLS)
.on_upgrade(move |stream| {
@@ -118,57 +118,18 @@ fn add_auth_layers(
))
}
/// Attach the shared per-IP write brake to an authenticated write-route group,
/// outermost so a flood is rejected before signature verification. Every
/// authenticated non-creation write group MUST call this or it silently ships
/// unbraked with no compile error. Deliberately a separate call, NOT folded into
/// `add_auth_layers`: the creation routes use a different bucket
/// (`create_ip_rate_limiter`) and the graphql group is not built through
/// `add_auth_layers`, so a helper keyed on auth would attach the wrong limiter.
trait WriteBraked {
fn write_braked(self, limiter: &rate_limit::IpRateLimiter) -> Self;
}
impl WriteBraked for Router<AppState> {
fn write_braked(self, limiter: &rate_limit::IpRateLimiter) -> Self {
self.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(limiter.clone()))
}
}
pub fn build_router(state: AppState) -> Router {
// ── GraphQL routes ─────────────────────────────────────────────────────
// Per-IP write brake, shared across every authenticated non-creation write
// group (see `AppState::write_rate_limiter`). Built once here so each group
// clones the same bucket; attached outermost on each group so a flood is
// rejected before auth runs. Separate bucket from the creation brake.
let write_ip_limiter = rate_limit::IpRateLimiter {
limiter: state.write_rate_limiter.clone(),
trust: state.push_limiter_trust,
};
let graphql_http = Router::new()
let graphql_routes = Router::new()
.route("/graphql", get(graphql_playground).post(graphql_handler))
// Attach the verified DID to /graphql when a signature is present. The
// layer covers only routes added before it, so /graphql/ws (mounted on
// its own router below, read-only subscriptions) gets no brake. The
// per-IP write brake covers /graphql (KTD-5: an HTTP-layer brake cannot
// tell a query POST from a mutation POST, so it counts every /graphql
// request — acceptable, both cost work); /graphql/ws stays unbraked.
.layer(middleware::from_fn_with_state(
auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
auth::optional_signature,
))
.write_braked(&write_ip_limiter);
let graphql_ws = Router::new()
.route("/graphql/ws", get(graphql_ws_handler))
// Attach the verified DID to /graphql and /graphql/ws when a signature is present.
.layer(middleware::from_fn_with_state(
auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
auth::optional_signature,
));
let graphql_routes = graphql_http.merge(graphql_ws);
// ── Task routes (write — require HTTP Signature + per-IP write brake) ──
// ── Task routes (write — require HTTP Signature) ───────────────────────
let task_write_routes = add_auth_layers(
Router::new()
.route("/api/v1/tasks", post(tasks::create_task))
@@ -177,8 +138,7 @@ pub fn build_router(state: AppState) -> Router {
.route("/api/v1/tasks/{id}/fail", post(tasks::fail_task)),
state.clone(),
SIGNED_BODY_LIMIT,
)
.write_braked(&write_ip_limiter);
);
// ── Task routes (read — open, but scoped) ──────────────────────────────
// `optional_signature` attaches the verified DID when a signature is present
@@ -237,13 +197,7 @@ pub fn build_router(state: AppState) -> Router {
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(create_ip_limiter));
// ── Write routes — require HTTP Signature + per-IP write brake ────────
// Same per-IP flood defense as the creation routes, on its own `write`
// bucket (see `AppState::write_rate_limiter`). Per-DID is deliberately not
// paired (a DID farm never trips it; busy legitimate agents would false-
// positive). The bundled visibility GET is a read that also draws from the
// write bucket — harmless, since the bucket is sized far above any legit
// per-IP read rate.
// ── Write routes — require HTTP Signature (no rate limit) ─────────────
let write_routes = add_auth_layers(
Router::new()
.route(
@@ -314,8 +268,69 @@ pub fn build_router(state: AppState) -> Router {
),
state.clone(),
SIGNED_BODY_LIMIT,
);
// ── Status claim writes — signature, plus the per-DID throttle AND the
// per-IP flood brake `creation_routes` carries. Not in `write_routes`, which
// has neither: every call here appends a row, so an unthrottled writer grows
// the claim log on demand and the three per-repo caps would be the only bound.
let status_write_routes = add_auth_layers(
Router::new()
.route(
"/api/v1/repos/{owner}/{repo}/statuses/{sha}",
post(status::create_status),
)
.layer(middleware::from_fn(rate_limit::rate_limit_by_did))
.layer(axum::Extension(state.rate_limiter.clone())),
state.clone(),
SIGNED_BODY_LIMIT,
)
.write_braked(&write_ip_limiter);
// The route's advertised body bound, applied at the transport. It sits
// OUTSIDE the auth layers because `require_signature` collects and hashes
// the whole body before any handler runs: the handler's own check measures
// an allocation that has already happened, so without this layer the 8 KiB
// figure bounded what gets stored and not what gets buffered, and
// concurrent requests multiplied exactly the cost it was written to
// prevent. It sits INSIDE the rate limiters so an oversized request is
// still counted by the per-IP brake before it is refused.
//
// A declared Content-Length over the limit is answered 413 without reading
// a byte; a body that declares nothing (or understates) is truncated at the
// limit and fails where it is read, which `require_signature` reports as an
// unreadable body.
.layer(RequestBodyLimitLayer::new(status::MAX_REQUEST_BODY_BYTES))
// The one route that persists the signed request body, so the one route
// whose signature material carries it. Applied OUTSIDE `add_auth_layers`
// (outermost = runs first) because `require_signature` reads this marker to
// decide whether to carry the body; a layer added inside the auth pair runs
// after the middleware and is never seen. `create_status` refuses an absent
// body rather than storing an empty column, so getting this order wrong
// fails loudly instead of quietly writing unverifiable claims.
.layer(axum::Extension(auth::PersistsSignedBody))
.layer(middleware::from_fn(rate_limit::rate_limit_by_ip))
.layer(axum::Extension(rate_limit::IpRateLimiter {
limiter: state.create_ip_rate_limiter.clone(),
trust: state.push_limiter_trust,
}));
// ── Status claim reads — open, behind `optional_signature` so the handler
// sees the caller when one signs and still answers a public repo for an
// anonymous reader. Without the layer the handler's optional auth extension
// is None for EVERY caller, including the owner, and every private repo
// reads as not-found.
let status_read_routes = Router::new()
.route(
"/api/v1/repos/{owner}/{repo}/commits/{sha}/status",
get(status::commit_status),
)
.route(
"/api/v1/repos/{owner}/{repo}/pulls/{number}/status",
get(status::pull_request_status),
)
.layer(middleware::from_fn_with_state(
auth::SignatureBodyLimit(SIGNED_BODY_LIMIT),
auth::optional_signature,
));
// Body limit is raised to GITLAWB_MAX_PACK_BYTES (default 2 GB) for git
// routes only — all other API routes keep axum's default 2 MB cap.
@@ -428,8 +443,7 @@ pub fn build_router(state: AppState) -> Router {
),
state.clone(),
SIGNED_BODY_LIMIT,
)
.write_braked(&write_ip_limiter);
);
// ── Bounty routes (read — open) ──────────────────────────────────────
let bounty_read_routes = Router::new()
@@ -454,15 +468,9 @@ pub fn build_router(state: AppState) -> Router {
Router::new().route("/api/v1/profile", axum::routing::put(profiles::set_profile)),
state.clone(),
SIGNED_BODY_LIMIT,
)
.write_braked(&write_ip_limiter);
);
// ── Issue routes (write — require HTTP Signature; NO outer write brake) ──
// Both handlers are read-gated before they mutate, so an outer per-IP
// brake would answer 429 before authorize_repo_read could return the
// opaque 404 a hidden repo owes the caller — the brake's status would
// distinguish a real repo from a missing one (INV-12). The write bucket
// is debited inside each handler, right after the read gate.
// ── Issue routes (write — require HTTP Signature, no rate limit) ─────
let issue_write_routes = add_auth_layers(
Router::new()
.route(
@@ -578,6 +586,10 @@ pub fn build_router(state: AppState) -> Router {
"/api/v1/repos/{owner}/{repo}/events",
get(events::list_repo_events),
)
.route(
"/api/v1/repos/{owner}/{repo}/push-events",
get(events::list_repo_push_events),
)
.route("/api/v1/agents", get(agents::list_agents))
.route("/api/v1/agents/{did}", get(agents::show_agent))
.route("/api/v1/agents/{did}/trust", get(agents::get_trust))
@@ -683,6 +695,8 @@ pub fn build_router(state: AppState) -> Router {
.merge(profile_write_routes)
.merge(creation_routes)
.merge(write_routes)
.merge(status_write_routes)
.merge(status_read_routes)
.merge(git_write_routes)
.merge(git_read_routes)
.merge(issue_write_routes)
+18 -65
View File
@@ -72,7 +72,6 @@ pub struct AppState {
pub task_event_tx: tokio::sync::broadcast::Sender<TaskEventBroadcast>,
/// GraphQL schema (queries + mutations + subscriptions)
pub graphql_schema: Arc<crate::graphql::GitlawbSchema>,
pub graphql_ws_schema: Arc<crate::graphql::GitlawbWsSchema>,
/// MAC key for task-list continuation tokens, derived from the node
/// keypair seed. Held here (and handed to the GraphQL schema) so REST and
/// GraphQL mint and accept the same tokens.
@@ -93,21 +92,6 @@ pub struct AppState {
/// resolved client IP is what actually stops a single-source flood (same
/// rationale as `push_rate_limiter`). Keyed by `push_limiter_trust`.
pub create_ip_rate_limiter: RateLimiter,
/// Per-client-IP rate limiter for the authenticated write surface that is
/// not repo/agent creation: issue/PR comments, labels, stars, merges,
/// protect/unprotect, replicas, webhook create/delete, visibility (its GET
/// shares the group), tasks, bounties, profile, and every HTTP `/graphql`
/// request (queries and the playground GET as well as mutations; the
/// WebSocket route is excluded). Issue comment and issue close debit it in
/// the handler after the read gate, not via router middleware; close also
/// draws on `close_issue_rate_limiter` and spends neither bucket when
/// either refuses. Separate bucket from `create_ip_rate_limiter`
/// (its own budget, per the `sync_trigger`/`peer_write` precedent) so a
/// write flood cannot drain the creation budget and vice versa. Per-DID is
/// deliberately NOT paired here: a DID farm never trips it, and busy
/// legitimate agents would false-positive. `GITLAWB_WRITE_RATE_LIMIT`
/// overrides the default, 0 disables. Keyed by `push_limiter_trust`.
pub write_rate_limiter: RateLimiter,
/// Per-client-IP rate limiter for git-receive-pack. Per-DID limits cannot
/// brake a push flood from a DID farm (one throwaway DID per repo), so the
/// push path throttles on the resolved client IP instead.
@@ -388,6 +372,24 @@ impl AppState {
self.shutdown_tx.subscribe()
}
/// Sweep expired entries from every per-IP/DID rate limiter. Driven by the
/// periodic cleanup task so a bounded limiter's key map sheds stale entries
/// instead of sitting near its cap until an inline capacity sweep reclaims
/// them. Every limiter on the state is swept here; adding a new limiter means
/// adding it to this list.
pub(crate) async fn sweep_rate_limiters(&self) {
self.rate_limiter.cleanup().await;
self.create_ip_rate_limiter.cleanup().await;
self.push_rate_limiter.cleanup().await;
self.close_issue_rate_limiter.cleanup().await;
self.ipfs_rate_limiter.cleanup().await;
self.ipfs_work_rate_limiter.cleanup().await;
self.sync_trigger_rate_limiter.cleanup().await;
self.peer_write_rate_limiter.cleanup().await;
self.task_read_rate_limiter.cleanup().await;
self.federated_rate_limiter.cleanup().await;
}
/// Trigger graceful shutdown. Idempotent — calling more than once
/// has no effect. Returns `true` if this call was the one that
/// flipped the signal.
@@ -409,26 +411,6 @@ impl AppState {
*self.shutdown_tx.borrow()
}
/// Reclaim expired entries from EVERY rate limiter. Single source of truth for
/// the periodic cleanup loop (main.rs), co-located with the limiter fields so a
/// newly-added limiter is cleaned here rather than silently omitted from a
/// hand-maintained list in main() (H2: `write_rate_limiter` was the omitted
/// one). Every limiter field above MUST be swept here; the guard test
/// `cleanup_rate_limiters_reaps_the_write_limiter` fails if it is not.
pub(crate) async fn cleanup_rate_limiters(&self) {
self.rate_limiter.cleanup().await;
self.create_ip_rate_limiter.cleanup().await;
self.write_rate_limiter.cleanup().await;
self.push_rate_limiter.cleanup().await;
self.close_issue_rate_limiter.cleanup().await;
self.ipfs_rate_limiter.cleanup().await;
self.ipfs_work_rate_limiter.cleanup().await;
self.sync_trigger_rate_limiter.cleanup().await;
self.peer_write_rate_limiter.cleanup().await;
self.task_read_rate_limiter.cleanup().await;
self.federated_rate_limiter.cleanup().await;
}
/// Legacy-probe budget wired from the `GITLAWB_IPFS_MAX_LEGACY_PROBES` operator
/// knob. The knob seeds `ipfs_max_legacy_probes` at construction so it controls the
/// per-request legacy (NULL-provenance) probe fan-out it advertises. It deliberately
@@ -1291,35 +1273,6 @@ mod repo_identity_key_tests {
}
}
#[cfg(test)]
mod tests {
/// H2 guard: the write-surface limiter must be reaped by the shared cleanup
/// routine. Seeds a reclaimable entry into `write_rate_limiter` and asserts
/// `cleanup_rate_limiters` removes it. Goes RED if `write_rate_limiter` is
/// dropped from `cleanup_rate_limiters` — the exact H2 omission, now guarded.
#[tokio::test]
async fn cleanup_rate_limiters_reaps_the_write_limiter() {
let state = crate::test_support::test_state_lazy();
state
.write_rate_limiter
.insert_reclaimable_for_test("some-ip")
.await;
assert_eq!(
state.write_rate_limiter.tracked_keys().await,
1,
"precondition: the reclaimable entry is tracked"
);
state.cleanup_rate_limiters().await;
assert_eq!(
state.write_rate_limiter.tracked_keys().await,
0,
"write_rate_limiter must be reaped by cleanup_rate_limiters (H2)"
);
}
}
#[cfg(test)]
mod repo_write_lease_tests {
use super::RepoWriteLeases;
+1 -8
View File
@@ -170,12 +170,6 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool, repos_dir: PathBuf) -> AppS
task_tx.clone(),
task_cursor_key.clone(),
));
let ws_schema = Arc::new(crate::graphql::build_ws_schema(
db.clone(),
ref_tx.clone(),
task_tx.clone(),
task_cursor_key.clone(),
));
AppState {
config: Arc::new(Config::parse_from(["gitlawb-node"])),
db,
@@ -186,13 +180,11 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool, repos_dir: PathBuf) -> AppS
ref_update_tx: ref_tx,
task_event_tx: task_tx,
graphql_schema: schema,
graphql_ws_schema: ws_schema,
task_cursor_key,
machine_id: None,
repo_store: crate::git::repo_store::RepoStore::for_testing(repos_dir, pool),
rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
write_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
close_issue_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
federated_rate_limiter: RateLimiter::new(12, Duration::from_secs(60)),
@@ -398,6 +390,7 @@ impl TestNode {
status: "open".to_string(),
merged_by_did: None,
merged_at: None,
head_commit: None,
created_at: now.clone(),
updated_at: now,
};
+231 -18
View File
@@ -95,13 +95,6 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
task_tx.clone(),
task_cursor_key.clone(),
));
let ws_schema = Arc::new(graphql::build_ws_schema(
db.clone(),
ref_tx.clone(),
task_tx.clone(),
task_cursor_key.clone(),
));
AppState {
config: Arc::new(Config::parse_from(["gitlawb-node"])),
db,
@@ -112,13 +105,11 @@ fn build_state(db: Arc<crate::db::Db>, pool: PgPool) -> AppState {
ref_update_tx: ref_tx,
task_event_tx: task_tx,
graphql_schema: schema,
graphql_ws_schema: ws_schema,
task_cursor_key,
machine_id: None,
repo_store: crate::git::repo_store::RepoStore::for_testing(PathBuf::from("/tmp"), pool),
rate_limiter: RateLimiter::new(100, Duration::from_secs(60)),
create_ip_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
write_rate_limiter: RateLimiter::new(1000, Duration::from_secs(3600)),
push_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
close_issue_rate_limiter: RateLimiter::new(120, Duration::from_secs(3600)),
ipfs_rate_limiter: RateLimiter::new(600, Duration::from_secs(3600)),
@@ -184,6 +175,153 @@ pub(crate) fn signed_request_as(did: &str, method: Method, uri: &str, body: Body
.expect("request builder")
}
/// Slice a region out of a module's own source (obtained with `include_str!`)
/// and drop full-line `//` comments from it.
///
/// Several guards in `api/` assert that a piece of PRODUCTION code does or does
/// not contain something: that the receive-pack handler really calls the
/// push-event recorder, that the status module never acquires the repository.
/// Each guard carried its own copy of this parse, and three copies of a parser
/// is three places for the slicing to drift apart while every guard still reads
/// green.
///
/// `start` and `end` are literal anchors. The region runs from the first
/// occurrence of `start` (inclusive; the top of the file when `None`) to the
/// next occurrence of `end` (exclusive; end of file when `None`).
///
/// Returns `None` when an anchor is missing, rather than silently falling back
/// to a larger or empty region. Each caller supplies its own message for that
/// case and keeps its own assertion that the region really covers what it claims
/// to, which is what stops a bug in this one helper from turning every guard
/// that uses it into a vacuous pass.
///
/// Comment lines are stripped so that a doc comment naming the very call a guard
/// searches for cannot stand in for the call itself.
pub(crate) fn scrape_source_region(
src: &str,
start: Option<&str>,
end: Option<&str>,
) -> Option<String> {
let from = match start {
Some(anchor) => src.find(anchor)?,
None => 0,
};
let rest = &src[from..];
let to = match end {
// Search past the first CHARACTER, so an `end` anchor that also matches
// at the very start of the region cannot collapse it to nothing. A fixed
// one-byte skip does the same job on ASCII and lands mid-character on a
// multibyte first character, where the slice fails and the helper answers
// `None`, which is indistinguishable, to every caller, from a missing
// anchor.
Some(anchor) => {
let skip = rest.chars().next().map_or(0, char::len_utf8);
rest.get(skip..)?.find(anchor)? + skip
}
None => rest.len(),
};
Some(
rest[..to]
.lines()
.filter(|l| !l.trim_start().starts_with("//"))
.collect::<Vec<_>>()
.join("\n"),
)
}
#[cfg(test)]
mod scrape_source_region_tests {
use super::scrape_source_region;
const SRC: &str = "prelude line\nfn target() {\n // calls_the_thing()\n calls_the_thing();\n}\nfn later() {\n calls_the_thing();\n}\n";
#[test]
fn stops_at_the_end_anchor_so_a_later_function_cannot_satisfy_a_guard() {
let body = scrape_source_region(SRC, Some("fn target()"), Some("\n}")).unwrap();
assert!(body.contains("calls_the_thing();"));
assert!(
!body.contains("fn later()"),
"the region must stop at the first column-zero brace, got: {body}"
);
}
#[test]
fn strips_full_line_comments_so_a_doc_comment_cannot_stand_in_for_code() {
let commented = "fn target() {\n // calls_the_thing();\n}\n";
let body = scrape_source_region(commented, Some("fn target()"), Some("\n}")).unwrap();
assert!(
!body.contains("calls_the_thing"),
"a commented-out call must not survive the scrape, got: {body}"
);
}
#[test]
fn no_start_anchor_reads_from_the_top_of_the_file() {
let body = scrape_source_region(SRC, None, Some("fn later()")).unwrap();
assert!(body.starts_with("prelude line"));
assert!(!body.contains("fn later()"));
}
#[test]
fn no_end_anchor_reads_to_the_end_of_the_file() {
let body = scrape_source_region(SRC, Some("fn later()"), None).unwrap();
assert!(body.trim_end().ends_with('}'));
}
/// A missing anchor is None, never a silently wider or empty region: that is
/// what lets each caller keep its own "the thing I meant to scan is gone"
/// message instead of asserting against whatever the fallback produced.
#[test]
fn a_missing_anchor_is_none_rather_than_a_fallback_region() {
assert!(scrape_source_region(SRC, Some("fn absent()"), Some("\n}")).is_none());
assert!(scrape_source_region(SRC, Some("fn target()"), Some("~absent~")).is_none());
}
/// A region whose first character is multibyte is scraped, not reported
/// missing.
///
/// The end-anchor search skips the region's first CHARACTER. Skipping a
/// fixed byte instead lands mid-character here, the slice fails, and the
/// helper answers `None`, which every caller reads as "the anchor is gone"
/// and reports as a missing region. A guard that says "not found"
/// for input that is present is worse than one that fails loudly, and the
/// five callers sharing this helper all inherit it.
#[test]
fn a_multibyte_first_character_does_not_read_as_a_missing_anchor() {
let src = "é prelude\nfn target() {\n body();\n}\nfn later() {}\n";
let body = scrape_source_region(src, None, Some("\n}"))
.expect("a region starting on a multibyte character must be scraped");
assert!(
body.starts_with("é prelude") && body.contains("body();"),
"the region must cover the multibyte start through the end anchor, got: {body}"
);
assert!(
!body.contains("fn later()"),
"the region must still stop at the end anchor, got: {body}"
);
// The same through a multibyte START anchor, so the skip is measured
// from the region rather than from the top of the file.
let body = scrape_source_region(src, Some("é"), Some("\n}"))
.expect("a multibyte start anchor must be scraped");
assert!(body.starts_with('é'), "got: {body}");
}
/// The end anchor is searched past the region's first character, so a
/// region whose own start also matches the end anchor is not collapsed to
/// nothing.
#[test]
fn an_end_anchor_matching_at_the_start_does_not_collapse_the_region() {
let src = "\n}\nfn target() {\n body();\n}\n";
let body = scrape_source_region(src, None, Some("\n}")).unwrap();
assert!(
body.contains("fn target()") && body.contains("body();"),
"got: {body}"
);
}
}
/// A local endpoint whose TCP accept succeeds instantly but that never writes an
/// HTTP response, so any request against it stalls deterministically until the
/// caller's own timeout. (A non-routable address hangs only if the network
@@ -191,7 +329,6 @@ pub(crate) fn signed_request_as(did: &str, method: Method, uri: &str, body: Body
/// test pass for the wrong reason.) The accepted sockets are parked in the
/// spawned task, which dies with the test's runtime, so the peer never sees a
/// close mid-test.
#[allow(dead_code)]
pub(crate) async fn silent_http_endpoint() -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let endpoint = format!("http://{}", listener.local_addr().unwrap());
@@ -204,6 +341,71 @@ pub(crate) async fn silent_http_endpoint() -> String {
endpoint
}
/// The response body decoded as JSON.
#[allow(dead_code)] // shared response helper for api test mods
pub(crate) async fn body_json(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")
}
/// Status plus the full response body, for the byte-identical deny comparison.
#[allow(dead_code)] // shared response helper for api test mods
pub(crate) async fn status_and_bytes(
resp: axum::response::Response,
) -> (axum::http::StatusCode, Vec<u8>) {
let status = resp.status();
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap()
.to_vec();
(status, bytes)
}
/// A `PendingRefTransition` test row in `applied` state: the fields the
/// commit-status effect writers read (`ref_name`, `new_sha`, `ordinal`,
/// `repo_id`) are the parameters, the rest is inert filler matching the
/// shape `insert_pending_ref_transitions` writes. `applied_at` stays None
/// so the recorded event falls back to the per-push timestamp the writer
/// computes once for the whole batch — the shape a real push produces.
#[allow(dead_code)] // commit-status surface tests across api modules
pub(crate) fn pending_child(
request_id: &str,
repo_id: &str,
ref_name: &str,
old_sha: &str,
new_sha: &str,
ordinal: i32,
) -> crate::db::PendingRefTransition {
crate::db::PendingRefTransition {
id: crate::db::deterministic_id(&[
"pending_ref_transition",
request_id,
repo_id,
ref_name,
old_sha,
new_sha,
]),
request_id: request_id.to_string(),
repo_id: repo_id.to_string(),
ref_name: ref_name.to_string(),
old_sha: old_sha.to_string(),
new_sha: new_sha.to_string(),
pusher_did: "did:key:z6pusher".to_string(),
node_did: "did:key:z6node".to_string(),
signature_header: String::new(),
signature_input: String::new(),
content_digest: String::new(),
state: crate::db::pending_state::APPLIED.to_string(),
created_at: "2026-01-01T00:00:00Z".to_string(),
applied_at: None,
cancelled_at: None,
ordinal,
git_target_kind: Some("update".to_string()),
}
}
/// Body of the `for-each-ref` arm of a fake-git fixture in the
/// COLUMN SHAPE `blob_paths` phase 2 parses. The caller is
/// responsible for wrapping this in a full `case "$1" in ... esac`
@@ -2282,6 +2484,7 @@ mod tests {
// Insert the poisoned row DIRECTLY, as a pre-fix row or an ungated writer would.
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -2640,6 +2843,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -2758,6 +2962,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -2882,6 +3087,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -2998,6 +3204,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3182,6 +3389,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3308,6 +3516,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3437,7 +3646,7 @@ mod tests {
run(&["commit", "-q", "-m", "seed"], &src);
// The tag that shadows the missing target branch: "release" resolves via
// DWIM but refs/heads/release never exists.
run(&["tag", "-m", "release", "release"], &src);
run(&["tag", "release"], &src);
run(&["checkout", "-q", "-b", "feature"], &src);
std::fs::write(src.join("feature.txt"), b"feature-only").unwrap();
run(&["add", "feature.txt"], &src);
@@ -3476,6 +3685,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3597,7 +3807,7 @@ mod tests {
// moves past it. Revision DWIM prefers refs/tags/ over refs/heads/ for
// an ambiguous short name, so this fixture is the strongest shadowing
// shape: a naive resolution would merge onto the stale tag commit.
run(&["tag", "-m", "release", "release"], &src);
run(&["tag", "release"], &src);
run(&["checkout", "-q", "-b", "release"], &src);
std::fs::write(src.join("release.txt"), b"release-only").unwrap();
run(&["add", "release.txt"], &src);
@@ -3640,6 +3850,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3830,6 +4041,7 @@ mod tests {
state.db.create_repo(&repo).await.expect("seed repo");
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -3982,6 +4194,7 @@ mod tests {
let _ = std::fs::remove_file(&pwn);
let pr = crate::db::PullRequest {
head_commit: None,
id: uuid::Uuid::new_v4().to_string(),
repo_id: repo.id.clone(),
number: 1,
@@ -17963,12 +18176,12 @@ mod tests {
}
/// The periodic cleanup task must sweep the ipfs walk limiter, not only its
/// five siblings. Drives `AppState::cleanup_rate_limiters` — the exact method the
/// five siblings. Drives `AppState::sweep_rate_limiters` — the exact method the
/// 300s loop calls — and asserts the ipfs limiter's expired entry is evicted.
/// Dropping `ipfs_rate_limiter.cleanup()` from that method leaves the entry in
/// place (`tracked_keys` stays 1): the RED proof that the sweep covers it.
#[sqlx::test]
async fn cleanup_rate_limiters_includes_ipfs_limiter(pool: PgPool) {
async fn sweep_rate_limiters_includes_ipfs_limiter(pool: PgPool) {
let mut state = test_state(pool).await;
// Short window so a single recorded hit is already expired at sweep time.
state.ipfs_rate_limiter = crate::rate_limit::RateLimiter::new(5, Duration::from_millis(50));
@@ -17985,7 +18198,7 @@ mod tests {
// Expire the entry (still mapped — cleanup hasn't run), then sweep.
tokio::time::sleep(Duration::from_millis(60)).await;
state.cleanup_rate_limiters().await;
state.sweep_rate_limiters().await;
assert_eq!(
state.ipfs_rate_limiter.tracked_keys().await,
@@ -18288,12 +18501,12 @@ mod tests {
/// U5 (R6): the periodic cleanup task sweeps the NEW work-budget limiter too, not
/// only the route limiter and its siblings. Mirrors
/// `cleanup_rate_limiters_includes_ipfs_limiter`: drive `cleanup_rate_limiters` and
/// `sweep_rate_limiters_includes_ipfs_limiter`: drive `sweep_rate_limiters` and
/// assert the work limiter's expired entry is evicted. Dropping the
/// `ipfs_work_rate_limiter.cleanup()` call from that method leaves the entry in place
/// (`tracked_keys` stays 1): the RED proof the sweep covers it.
#[sqlx::test]
async fn cleanup_rate_limiters_includes_ipfs_work_limiter(pool: PgPool) {
async fn sweep_rate_limiters_includes_ipfs_work_limiter(pool: PgPool) {
let mut state = test_state(pool).await;
state.ipfs_work_rate_limiter =
crate::rate_limit::RateLimiter::new(5, Duration::from_millis(50));
@@ -18309,7 +18522,7 @@ mod tests {
);
tokio::time::sleep(Duration::from_millis(60)).await;
state.cleanup_rate_limiters().await;
state.sweep_rate_limiters().await;
assert_eq!(
state.ipfs_work_rate_limiter.tracked_keys().await,
@@ -915,6 +915,7 @@ fn fill(path: &str, fixture: &Fixture, repo: &str, id_source: IdSource) -> Strin
};
path.replace("{owner}", &fixture.owner_did)
.replace("{repo}", repo)
.replace("{sha}", "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
.replace("{number}", "1")
.replace("{id}", id)
.replace("{label}", "bug")
@@ -234,6 +234,23 @@ pub fn deny_bearing_routes() -> &'static [Row] {
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
// create_status: signed non-owner on a public repo passes the read
// gate and hits `require_repo_owner` — a plain owner-gate 403. The
// body is a minimal valid claim (the request is body-bound signed, so
// the probe's RFC-9421 content-digest coverage is load-bearing: a
// signer that did not cover the body would 400 before the gate).
Row {
method: "POST",
path: "/api/v1/repos/{owner}/{repo}/statuses/{sha}",
gate: GateClass::OwnerGate,
handler: "status::create_status",
body: Some(r#"{"state":"pending","context":"ci/build"}"#),
needs: NO_ENTITY,
reach: Reach::None,
principals: NO_PRINCIPAL,
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
// #195 (F1): close_pr / close_issue are owner-OR-author gates (pulls.rs:276,
// issues.rs:255). A plain OwnerGate tests only the owner arm, so reverting
// the author arm is invisible. MultiPrincipalGate drives BOTH arms.
@@ -784,6 +801,50 @@ pub fn deny_bearing_routes() -> &'static [Row] {
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
// pull_request_status: the seeded PRIVATE PR #1 is the reach target, so
// the owner twin gets a 200 rollup, not the absent-entity 404 the gate's
// own existence-hiding would make indistinguishable.
Row {
method: "GET",
path: "/api/v1/repos/{owner}/{repo}/pulls/{number}/status",
gate: GateClass::ReadGate,
handler: "status::pull_request_status",
body: None,
needs: &["priv_pr"],
reach: Reach::ReaderReads,
principals: NO_PRINCIPAL,
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
// commit_status / list_repo_push_events: repo-scoped optional-auth reads
// whose deny is the repo's own existence-hiding 404. Neither needs a
// seeded entity — a status rollup over zero claims and an empty
// push-event page are both valid 2xx responses, so the owner twin
// proving the gate is the repo lookup, not the data.
Row {
method: "GET",
path: "/api/v1/repos/{owner}/{repo}/commits/{sha}/status",
gate: GateClass::ReadGate,
handler: "status::commit_status",
body: None,
needs: NO_ENTITY,
reach: Reach::ReaderReads,
principals: NO_PRINCIPAL,
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
Row {
method: "GET",
path: "/api/v1/repos/{owner}/{repo}/push-events",
gate: GateClass::ReadGate,
handler: "events::list_repo_push_events",
body: None,
needs: NO_ENTITY,
reach: Reach::ReaderReads,
principals: NO_PRINCIPAL,
id_source: IdSource::Fixed,
anon_read: AnonRead::Deny404,
},
// get_cert: id-keyed by a real ref-certificate issued by an owner push to
// the private repo. No cert-content marker is seeded: get_cert read-gates
// the repo on "/" and 404s a non-reader BEFORE fetching the cert, so cert