diff --git a/src/quorumgit/cli.py b/src/quorumgit/cli.py index 4749d04..1a1b5e3 100644 --- a/src/quorumgit/cli.py +++ b/src/quorumgit/cli.py @@ -10,7 +10,18 @@ import json import sys -from . import __version__, audit, config, gate, handoff, registry, store, trees, work +from . import ( + __version__, + audit, + config, + gate, + handoff, + receive, + registry, + store, + trees, + work, +) def _agent(args, cfg) -> str: @@ -159,8 +170,6 @@ def cmd_claim(args, cfg) -> int: takeover_operation = None takeover_approval = None if args.takeover: - # Bind the approval to a stable incumbent. claim_task() and every - # release take this same task lock, closing the stale-holder window. work.lock_task(conn, args.task_id) holder = work.active_claim_for_task(conn, args.task_id) if holder and not holder["expired"]: @@ -184,7 +193,7 @@ def cmd_claim(args, cfg) -> int: f"{json.dumps(takeover_operation, sort_keys=True)}", file=sys.stderr, ) - conn.commit() # keep the conflict/audit trail + conn.commit() return 1 try: claim_id, classification, _ = work.claim_task( @@ -198,15 +207,9 @@ def cmd_claim(args, cfg) -> int: takeover_approved=takeover_operation is not None, ) except work.ClaimRefused as exc: - # A refused claim is non-destructive: the holder was not released - # and the approval was not consumed. Commit only the conflict - # event and audit trail. conn.commit() print(f"[quorumgit] REFUSED: {exc}", file=sys.stderr) return 1 - # The takeover succeeded: consume the approval in the same - # transaction, so replacement claim + holder release + consumption - # commit together or not at all. if takeover_operation is not None: try: assert takeover_approval is not None @@ -391,15 +394,20 @@ def cmd_approve_hash(args, cfg) -> int: def cmd_hook_install(args, cfg) -> int: with store.connect(cfg) as conn: - path = gate.install_hook(conn, args.repo) + path = receive.install_hooks(conn, args.repo) conn.commit() - print(f"pre-receive hook installed: {path}") + print(f"receive hooks installed (pre-receive: {path})") return 0 def cmd_hook_pre_receive(args, cfg) -> int: with store.connect(cfg) as conn: - return gate.run_pre_receive(conn, args.repo, sys.stdin) + return receive.run_pre_receive(conn, args.repo, sys.stdin) + + +def cmd_hook_post_receive(args, cfg) -> int: + with store.connect(cfg) as conn: + return receive.run_post_receive(conn, args.repo, sys.stdin) # -------------------------------------------------------------------- audit @@ -541,6 +549,8 @@ def add(name, fn, configure=None, parent=None): lambda sp: sp.add_argument("--repo", required=True), parent=hook) add("pre-receive", cmd_hook_pre_receive, lambda sp: sp.add_argument("--repo", required=True), parent=hook) + add("post-receive", cmd_hook_post_receive, + lambda sp: sp.add_argument("--repo", required=True), parent=hook) add("audit", cmd_audit, lambda sp: ( sp.add_argument("--entity"), diff --git a/src/quorumgit/migrations/003_git_mutations.sql b/src/quorumgit/migrations/003_git_mutations.sql new file mode 100644 index 0000000..f882f7a --- /dev/null +++ b/src/quorumgit/migrations/003_git_mutations.sql @@ -0,0 +1,111 @@ +-- Reserve exact incoming Git mutations between pre-receive validation and +-- post-receive completion. Reservations freeze governance for the affected ref +-- while Git is in flight and bind any protected-operation approval instance to +-- that exact mutation until Git either completes or the reservation expires. + +CREATE TABLE git_mutations ( + id INTEGER PRIMARY KEY, + repository_id INTEGER NOT NULL REFERENCES repositories(id), + refname TEXT NOT NULL CHECK (refname <> ''), + oldrev TEXT NOT NULL CHECK (oldrev <> ''), + newrev TEXT NOT NULL CHECK (newrev <> ''), + mutation_hash TEXT NOT NULL CHECK ( + length(mutation_hash) = 71 + AND substr(mutation_hash, 1, 7) = 'sha256:' + AND substr(mutation_hash, 8) NOT GLOB '*[^0-9a-f]*' + ), + agent_id INTEGER NOT NULL REFERENCES agents(id), + approval_id INTEGER REFERENCES approvals(id), + status TEXT NOT NULL DEFAULT 'reserved' + CHECK (status IN ('reserved', 'completed', 'expired')), + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + expires_at INTEGER NOT NULL, + completed_at INTEGER, + CHECK (status = 'completed' OR completed_at IS NULL), + CHECK (status <> 'completed' OR completed_at IS NOT NULL) +); + +-- quorumgit-statement +CREATE UNIQUE INDEX git_mutations_one_reserved_ref + ON git_mutations(repository_id, refname) + WHERE status = 'reserved'; + +-- quorumgit-statement +CREATE UNIQUE INDEX git_mutations_one_reserved_exact + ON git_mutations(mutation_hash) + WHERE status = 'reserved'; + +-- quorumgit-statement +CREATE UNIQUE INDEX git_mutations_one_reserved_approval + ON git_mutations(approval_id) + WHERE status = 'reserved' AND approval_id IS NOT NULL; + +-- quorumgit-statement +CREATE INDEX git_mutations_by_repository_status + ON git_mutations(repository_id, status, expires_at); + +-- quorumgit-statement +CREATE TRIGGER claims_blocked_by_git_mutation_insert +BEFORE INSERT ON claims +WHEN EXISTS ( + SELECT 1 + FROM tasks t + JOIN git_mutations m ON m.repository_id = t.repository_id + WHERE t.id = NEW.task_id + AND m.refname = 'refs/heads/' || NEW.branch + AND m.status = 'reserved' + AND m.expires_at >= unixepoch() +) +BEGIN + SELECT RAISE(ABORT, 'branch has an in-flight Git mutation'); +END; + +-- quorumgit-statement +CREATE TRIGGER claims_blocked_by_git_mutation_update +BEFORE UPDATE OF lease_expires_at, released_at, release_reason ON claims +WHEN EXISTS ( + SELECT 1 + FROM tasks t + JOIN git_mutations m ON m.repository_id = t.repository_id + WHERE t.id = OLD.task_id + AND m.refname = 'refs/heads/' || OLD.branch + AND m.status = 'reserved' + AND m.expires_at >= unixepoch() +) +BEGIN + SELECT RAISE(ABORT, 'branch has an in-flight Git mutation'); +END; + +-- quorumgit-statement +CREATE TRIGGER handoffs_blocked_by_git_mutation_insert +BEFORE INSERT ON handoffs +WHEN EXISTS ( + SELECT 1 + FROM claims c + JOIN tasks t ON t.id = c.task_id + JOIN git_mutations m ON m.repository_id = t.repository_id + WHERE c.id = NEW.from_claim_id + AND m.refname = 'refs/heads/' || c.branch + AND m.status = 'reserved' + AND m.expires_at >= unixepoch() +) +BEGIN + SELECT RAISE(ABORT, 'branch has an in-flight Git mutation'); +END; + +-- quorumgit-statement +CREATE TRIGGER handoffs_blocked_by_git_mutation_update +BEFORE UPDATE OF status, to_agent_id, resolved_at ON handoffs +WHEN EXISTS ( + SELECT 1 + FROM claims c + JOIN tasks t ON t.id = c.task_id + JOIN git_mutations m ON m.repository_id = t.repository_id + WHERE c.id = OLD.from_claim_id + AND m.refname = 'refs/heads/' || c.branch + AND m.status = 'reserved' + AND m.expires_at >= unixepoch() +) +BEGIN + SELECT RAISE(ABORT, 'branch has an in-flight Git mutation'); +END; diff --git a/src/quorumgit/mutations.py b/src/quorumgit/mutations.py new file mode 100644 index 0000000..54ab9d5 --- /dev/null +++ b/src/quorumgit/mutations.py @@ -0,0 +1,296 @@ +"""Durable reservations for Git ref mutations in flight. + +`pre-receive` validates governance and reserves the exact ref transition before +Git performs it. Governance changes that could invalidate that decision are +blocked while the reservation is live. `post-receive` completes the same +reservation only after Git reports the ref update as successful. +""" + +from __future__ import annotations + +from typing import Any + +from . import audit +from .canonical import stable_hash +from .store import Connection, begin_immediate + +DEFAULT_RESERVATION_SECONDS = 300 + + +class MutationError(RuntimeError): + pass + + +def mutation_payload( + repository: str, + refname: str, + oldrev: str, + newrev: str, +) -> dict[str, str]: + return { + "type": "git_ref_mutation", + "repository": repository, + "refname": refname, + "oldrev": oldrev, + "newrev": newrev, + } + + +def mutation_hash( + repository: str, + refname: str, + oldrev: str, + newrev: str, +) -> str: + return stable_hash(mutation_payload(repository, refname, oldrev, newrev)) + + +def _row_dict(row) -> dict[str, Any]: + return { + "id": row[0], + "repository_id": row[1], + "refname": row[2], + "oldrev": row[3], + "newrev": row[4], + "mutation_hash": row[5], + "agent_id": row[6], + "approval_id": row[7], + "status": row[8], + "created_at": row[9], + "expires_at": row[10], + "completed_at": row[11], + } + + +def expire_stale(conn: Connection, repository_id: int | None = None) -> int: + """Expire abandoned reservations and audit each transition. + + A failed receive can exit after pre-receive without a completion callback. + Reservations are therefore leases rather than permanent locks. Expiry does + not consume an approval and does not claim that Git changed the ref. + """ + begin_immediate(conn) + params: tuple[Any, ...] + if repository_id is None: + rows = conn.execute( + """ + SELECT id FROM git_mutations + WHERE status = 'reserved' AND expires_at < unixepoch() + ORDER BY id + """ + ).fetchall() + params = () + where = "status = 'reserved' AND expires_at < unixepoch()" + else: + rows = conn.execute( + """ + SELECT id FROM git_mutations + WHERE repository_id = ? AND status = 'reserved' + AND expires_at < unixepoch() + ORDER BY id + """, + (repository_id,), + ).fetchall() + params = (repository_id,) + where = ( + "repository_id = ? AND status = 'reserved' " + "AND expires_at < unixepoch()" + ) + if not rows: + return 0 + conn.execute( + f"UPDATE git_mutations SET status = 'expired' WHERE {where}", # noqa: S608 -- fixed clauses only + params, + ) + for row in rows: + audit.record( + conn, + "git_mutation.expired", + "git_mutation", + row[0], + detail={"reason": "reservation_timeout"}, + ) + return len(rows) + + +def active_for_ref( + conn: Connection, + repository_id: int, + refname: str, +) -> dict[str, Any] | None: + expire_stale(conn, repository_id) + row = conn.execute( + """ + SELECT id, repository_id, refname, oldrev, newrev, mutation_hash, + agent_id, approval_id, status, created_at, expires_at, completed_at + FROM git_mutations + WHERE repository_id = ? AND refname = ? AND status = 'reserved' + LIMIT 1 + """, + (repository_id, refname), + ).fetchone() + return _row_dict(row) if row else None + + +def assert_ref_available( + conn: Connection, + repository_id: int, + refname: str, + *, + allow_mutation_id: int | None = None, +) -> None: + current = active_for_ref(conn, repository_id, refname) + if current is None or current["id"] == allow_mutation_id: + return + raise MutationError( + f"Git ref {refname!r} has in-flight mutation {current['id']} " + f"({current['oldrev']} -> {current['newrev']}); retry after it completes " + "or its reservation expires." + ) + + +def assert_branch_available( + conn: Connection, + repository_id: int, + branch: str, +) -> None: + assert_ref_available(conn, repository_id, f"refs/heads/{branch}") + + +def reserve( + conn: Connection, + *, + repository_id: int, + repository: str, + refname: str, + oldrev: str, + newrev: str, + agent_id: int, + agent: str, + approval_id: int | None = None, + reservation_seconds: int = DEFAULT_RESERVATION_SECONDS, +) -> dict[str, Any]: + """Reserve one exact ref transition. + + Retries of the identical mutation by the same agent and approval instance + renew the existing lease; a different in-flight mutation on the same ref is + rejected. The approval remains merely approved until Git actually succeeds. + """ + if reservation_seconds <= 0: + raise MutationError("Git mutation reservation duration must be positive.") + begin_immediate(conn) + expire_stale(conn, repository_id) + digest = mutation_hash(repository, refname, oldrev, newrev) + current = active_for_ref(conn, repository_id, refname) + if current is not None: + if ( + current["mutation_hash"] == digest + and current["agent_id"] == agent_id + and current["approval_id"] == approval_id + ): + conn.execute( + "UPDATE git_mutations SET expires_at = unixepoch() + ? WHERE id = ?", + (reservation_seconds, current["id"]), + ) + current["expires_at"] = None + audit.record( + conn, + "git_mutation.renewed", + "git_mutation", + current["id"], + agent=agent, + detail={"mutation_hash": digest, "refname": refname}, + ) + return current + raise MutationError( + f"Git ref {refname!r} already has in-flight mutation {current['id']}." + ) + + row = conn.execute( + """ + INSERT INTO git_mutations ( + repository_id, refname, oldrev, newrev, mutation_hash, + agent_id, approval_id, expires_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, unixepoch() + ?) + RETURNING id, repository_id, refname, oldrev, newrev, mutation_hash, + agent_id, approval_id, status, created_at, expires_at, completed_at + """, + ( + repository_id, + refname, + oldrev, + newrev, + digest, + agent_id, + approval_id, + reservation_seconds, + ), + ).fetchone() + assert row is not None + result = _row_dict(row) + audit.record( + conn, + "git_mutation.reserved", + "git_mutation", + result["id"], + agent=agent, + detail={ + "repository": repository, + "refname": refname, + "oldrev": oldrev, + "newrev": newrev, + "mutation_hash": digest, + "approval_id": approval_id, + }, + ) + return result + + +def reserved_exact( + conn: Connection, + *, + repository_id: int, + repository: str, + refname: str, + oldrev: str, + newrev: str, + agent_id: int, +) -> dict[str, Any] | None: + expire_stale(conn, repository_id) + digest = mutation_hash(repository, refname, oldrev, newrev) + row = conn.execute( + """ + SELECT id, repository_id, refname, oldrev, newrev, mutation_hash, + agent_id, approval_id, status, created_at, expires_at, completed_at + FROM git_mutations + WHERE repository_id = ? AND mutation_hash = ? AND agent_id = ? + AND status = 'reserved' + LIMIT 1 + """, + (repository_id, digest, agent_id), + ).fetchone() + return _row_dict(row) if row else None + + +def complete(conn: Connection, mutation_id: int, *, agent: str) -> None: + begin_immediate(conn) + cur = conn.execute( + """ + UPDATE git_mutations + SET status = 'completed', completed_at = unixepoch() + WHERE id = ? AND status = 'reserved' + """, + (mutation_id,), + ) + if cur.rowcount != 1: + raise MutationError( + f"Git mutation {mutation_id} is not an active reservation." + ) + audit.record( + conn, + "git_mutation.completed", + "git_mutation", + mutation_id, + agent=agent, + ) diff --git a/src/quorumgit/receive.py b/src/quorumgit/receive.py new file mode 100644 index 0000000..c0dedc0 --- /dev/null +++ b/src/quorumgit/receive.py @@ -0,0 +1,338 @@ +"""Receive-hook protocol for exact Git mutation reservation and completion. + +Pre-receive validates current governance and durably reserves each exact ref +transition without consuming its approval. Post-receive runs only after Git has +updated refs, revalidates the resulting ref, consumes any bound approval, and +marks the reservation completed in the same database transaction. +""" + +from __future__ import annotations + +import os +import shlex +import subprocess +import sys +from pathlib import Path +from typing import Iterable + +from . import audit, gate, mutations +from .registry import RegistryError, get_agent +from .store import Connection, begin_immediate +from .work import live_claim_for_branch, open_handoff_for_branch + +POST_HOOK_MARKER = "# quorumgit-managed-post-receive v1" + + +class ReceiveError(gate.GateError): + pass + + +def _registered_pusher(conn: Connection) -> dict: + pusher = os.environ.get("QUORUMGIT_AGENT") or None + if pusher is None: + raise ReceiveError( + "Pusher identity required; set QUORUMGIT_AGENT to a registered agent." + ) + try: + return get_agent(conn, pusher) + except RegistryError as exc: + raise ReceiveError(f"Pusher identity is not registered: {pusher}") from exc + + +def _git_dir() -> str: + git_dir = os.environ.get("GIT_DIR") + if not git_dir: + raise ReceiveError("GIT_DIR is not set.") + return git_dir + + +def _policy_for_update( + conn: Connection, + repository: str, + git_dir: str, + pusher: str, + oldrev: str, + newrev: str, + refname: str, +) -> tuple[dict, dict | None]: + repo = gate._verify_repository_binding(conn, repository, git_dir) + branch = refname.removeprefix("refs/heads/") + + if refname.startswith("refs/heads/"): + pending = open_handoff_for_branch(conn, repo["id"], branch) + if pending: + raise ReceiveError( + f"Branch {branch!r} is frozen pending handoff " + f"{pending['id']}; accept the handoff before pushing." + ) + claim = live_claim_for_branch(conn, repo["id"], branch) + if claim and claim["agent"] != pusher: + raise ReceiveError( + f"Branch {branch!r} is claimed by {claim['agent']} " + f"(claim {claim['id']}); pusher is {pusher}." + ) + + protected = refname in repo["protected_refs"] + deletion = gate._is_zero(newrev) + forced = ( + not deletion + and not gate._is_zero(oldrev) + and not gate._is_fast_forward(git_dir, oldrev, newrev) + ) + approval = None + if protected or deletion or forced: + operation = { + "type": "protected_ref_update" + if protected + else ("ref_delete" if deletion else "force_update"), + "repository": repository, + "refname": refname, + "oldrev": oldrev, + "newrev": newrev, + } + approval = gate.approved_instance(conn, operation) + if approval is None: + raise ReceiveError( + f"{operation['type']} on {refname} requires an approval " + f"bound to this exact update (hash {gate.operation_hash(operation)})." + ) + return repo, approval + + +def run_pre_receive( + conn: Connection, + repository: str, + stdin_lines: Iterable[str], +) -> int: + """Validate and reserve each incoming ref mutation. Fail closed.""" + try: + begin_immediate(conn) + git_dir = _git_dir() + pusher = _registered_pusher(conn) + saw_update = False + for line in stdin_lines: + parts = line.split() + if not parts: + continue + if len(parts) != 3: + raise ReceiveError(f"Malformed pre-receive input: {line!r}") + saw_update = True + oldrev, newrev, refname = parts + repo, approval = _policy_for_update( + conn, + repository, + git_dir, + pusher["name"], + oldrev, + newrev, + refname, + ) + mutation = mutations.reserve( + conn, + repository_id=repo["id"], + repository=repository, + refname=refname, + oldrev=oldrev, + newrev=newrev, + agent_id=pusher["id"], + agent=pusher["name"], + approval_id=approval["id"] if approval else None, + ) + audit.record( + conn, + "gate.update_reserved", + "repository", + repo["id"], + agent=pusher["name"], + detail={ + "mutation_id": mutation["id"], + "refname": refname, + "oldrev": oldrev, + "newrev": newrev, + "approval_id": approval["id"] if approval else None, + }, + ) + if not saw_update: + raise ReceiveError("No ref updates supplied on stdin.") + except Exception as exc: + conn.rollback() + print(f"[quorumgit] REJECTED: {exc}", file=sys.stderr) + return 1 + conn.commit() + print("[quorumgit] reserved.") + return 0 + + +def _verify_ref_result(git_dir: str, refname: str, newrev: str) -> None: + if gate._is_zero(newrev): + result = subprocess.run( + ["git", "--git-dir", git_dir, "show-ref", "--verify", "--quiet", refname], + capture_output=True, + ) + if result.returncode == 1: + return + if result.returncode == 0: + raise ReceiveError(f"Git reports deleted ref {refname} still exists.") + raise ReceiveError(f"Unable to verify deletion of {refname}.") + + result = subprocess.run( + ["git", "--git-dir", git_dir, "rev-parse", "--verify", refname], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise ReceiveError(f"Unable to verify completed Git ref {refname}.") + actual = result.stdout.strip() + if actual != newrev: + raise ReceiveError( + f"Git ref {refname} is {actual}, expected completed mutation {newrev}." + ) + + +def run_post_receive( + conn: Connection, + repository: str, + stdin_lines: Iterable[str], +) -> int: + """Finalize reservations after Git has successfully updated its refs.""" + try: + begin_immediate(conn) + git_dir = _git_dir() + pusher = _registered_pusher(conn) + repo = gate._verify_repository_binding(conn, repository, git_dir) + saw_update = False + for line in stdin_lines: + parts = line.split() + if not parts: + continue + if len(parts) != 3: + raise ReceiveError(f"Malformed post-receive input: {line!r}") + saw_update = True + oldrev, newrev, refname = parts + _verify_ref_result(git_dir, refname, newrev) + mutation = mutations.reserved_exact( + conn, + repository_id=repo["id"], + repository=repository, + refname=refname, + oldrev=oldrev, + newrev=newrev, + agent_id=pusher["id"], + ) + if mutation is None: + raise ReceiveError( + f"No live QuorumGit reservation matches completed ref {refname}." + ) + if mutation["approval_id"] is not None: + approval = gate.get_approval_by_id(conn, mutation["approval_id"]) + gate.consume_approval( + conn, + approval["id"], + approval["operation"], + agent=pusher["name"], + ) + mutations.complete(conn, mutation["id"], agent=pusher["name"]) + audit.record( + conn, + "gate.update_completed", + "repository", + repo["id"], + agent=pusher["name"], + detail={ + "mutation_id": mutation["id"], + "refname": refname, + "oldrev": oldrev, + "newrev": newrev, + "approval_id": mutation["approval_id"], + }, + ) + if not saw_update: + raise ReceiveError("No completed ref updates supplied on stdin.") + except Exception as exc: + conn.rollback() + print(f"[quorumgit] COMPLETION ERROR: {exc}", file=sys.stderr) + return 1 + conn.commit() + print("[quorumgit] completed.") + return 0 + + +def _effective_hook(repository_path: str | Path, name: str) -> Path: + repo_path = Path(repository_path).resolve() + result = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "--git-path", f"hooks/{name}"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + detail = (result.stderr or result.stdout).strip() + suffix = f": {detail}" if detail else "" + raise ReceiveError(f"Unable to resolve Git {name} hook path{suffix}") + raw = result.stdout.strip() + if not raw: + raise ReceiveError(f"Git returned no {name} hook path.") + hook = Path(raw) + if not hook.is_absolute(): + hook = repo_path / hook + return hook.resolve() + + +def _post_hook_script(repository: str) -> str: + return ( + "#!/bin/sh\n" + f"{POST_HOOK_MARKER}\n" + f"exec {shlex.quote(sys.executable)} -m quorumgit hook post-receive " + f"--repo {shlex.quote(repository)}\n" + ) + + +def _preflight_post_hook(path: Path, expected: str) -> str: + if not path.exists(): + return "installed" + try: + existing = path.read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + raise ReceiveError(f"Cannot inspect existing post-receive hook {path}: {exc}") from exc + if existing == expected: + return "verified" + if POST_HOOK_MARKER in existing.splitlines()[:3]: + raise ReceiveError( + f"Existing QuorumGit-managed post-receive hook differs at {path}; " + "refusing to overwrite it." + ) + raise ReceiveError( + f"Existing post-receive hook at {path} is not owned by QuorumGit; " + "refusing to overwrite or silently chain it." + ) + + +def install_hooks(conn: Connection, repository: str) -> Path: + """Install the pre-receive gate plus its completion hook safely.""" + begin_immediate(conn) + repo = gate.get_repository(conn, repository) + post_path = _effective_hook(repo["path"], "post-receive") + expected_post = _post_hook_script(repository) + post_action = _preflight_post_hook(post_path, expected_post) + + # `gate.install_hook` performs the existing pre-receive ownership, + # core.hooksPath, repository-identity, and atomic-write checks. Preflight + # the post hook first so an unrelated post hook cannot leave a partial new + # installation behind. + pre_path = gate.install_hook(conn, repository) + if post_action == "installed": + gate._write_hook_atomically(post_path, expected_post) + else: + post_path.chmod(0o755) + + if _effective_hook(repo["path"], "post-receive") != post_path: + raise ReceiveError("Installed post-receive hook is not Git's effective hook.") + if post_path.read_text(encoding="utf-8") != expected_post: + raise ReceiveError(f"Installed post-receive hook failed verification: {post_path}") + audit.record( + conn, + f"hook.post_receive_{post_action}", + "repository", + repo["id"], + detail={"path": str(post_path)}, + ) + return pre_path diff --git a/tests/test_authorization_binding.py b/tests/test_authorization_binding.py index 4212f72..b064d04 100644 --- a/tests/test_authorization_binding.py +++ b/tests/test_authorization_binding.py @@ -35,6 +35,13 @@ def _apply_001(cfg: Config): return conn +def _migrate_after_001(cfg: Config) -> list[str]: + """Apply every packaged successor while keeping the identity migration first.""" + applied = store.migrate(cfg) + assert applied and applied[0] == "002_approval_identities.sql" + return applied + + def _register_001_agent(conn, name: str) -> int: row = conn.execute( "INSERT INTO agents (name) VALUES (?) RETURNING id", (name,) @@ -185,13 +192,16 @@ def test_contract_rejects_001_only_store_until_all_migrations_apply(tmp_path): conn.commit() conn.close() - message = "Missing required migrations: ['002_approval_identities.sql']" - with pytest.raises(store.ContractViolation, match=re.escape(message)): - store.verify_contract(cfg) - with pytest.raises(store.ContractViolation, match=re.escape(message)): - store.connect(cfg) + for opener in (store.verify_contract, store.connect): + with pytest.raises(store.ContractViolation) as caught: + opener(cfg) + message = str(caught.value) + assert "002_approval_identities.sql" in message + assert "003_git_mutations.sql" in message + assert "Run `quorumgit init`" in message - assert store.migrate(cfg) == ["002_approval_identities.sql"] + applied = _migrate_after_001(cfg) + assert "003_git_mutations.sql" in applied store.verify_contract(cfg) upgraded = store.connect(cfg) upgraded.close() @@ -208,7 +218,8 @@ def test_identity_migration_preserves_and_backfills_history(tmp_path): conn.commit() conn.close() - assert store.migrate(cfg) == ["002_approval_identities.sql"] + applied = _migrate_after_001(cfg) + assert "003_git_mutations.sql" in applied migrated = store.connect(cfg) try: requester = migrated.execute( @@ -248,7 +259,8 @@ def test_migration_does_not_backfill_requester_registered_after_request(tmp_path conn.commit() conn.close() - assert store.migrate(cfg) == ["002_approval_identities.sql"] + applied = _migrate_after_001(cfg) + assert "003_git_mutations.sql" in applied migrated = store.connect(cfg) try: row = migrated.execute( @@ -280,7 +292,8 @@ def test_migration_does_not_backfill_voter_registered_after_vote(tmp_path): conn.commit() conn.close() - assert store.migrate(cfg) == ["002_approval_identities.sql"] + applied = _migrate_after_001(cfg) + assert "003_git_mutations.sql" in applied migrated = store.connect(cfg) try: approval = migrated.execute( diff --git a/tests/test_git_mutation_protocol.py b/tests/test_git_mutation_protocol.py new file mode 100644 index 0000000..64dba20 --- /dev/null +++ b/tests/test_git_mutation_protocol.py @@ -0,0 +1,253 @@ +"""Regression coverage for the receive reservation/completion protocol.""" + +from __future__ import annotations + +import subprocess +import uuid +from pathlib import Path + +import pytest + +from quorumgit import gate, receive, registry, store, work +from quorumgit.config import Config +from tests.conftest import make_git_repo +from tests.test_gate import _commit, _push + + +def _setup(committed_conn, tmp_path, cfg): + seed = make_git_repo(tmp_path / "seed") + hub = tmp_path / "hub.git" + subprocess.run( + ["git", "clone", "--bare", str(seed), str(hub)], + check=True, + capture_output=True, + ) + repo_name = f"mutation-{uuid.uuid4().hex[:8]}" + owner = f"owner-{uuid.uuid4().hex[:8]}" + operator = f"operator-{uuid.uuid4().hex[:8]}" + registry.add_repository( + committed_conn, + repo_name, + hub, + protected_refs=["refs/heads/main"], + ) + registry.add_agent(committed_conn, owner) + registry.add_agent(committed_conn, operator) + receive.install_hooks(committed_conn, repo_name) + committed_conn.commit() + + clone = tmp_path / "clone" + subprocess.run( + ["git", "clone", str(hub), str(clone)], + check=True, + capture_output=True, + ) + return repo_name, hub, clone, owner, operator + + +def _approve_main_update(conn, repo_name, hub, clone, operator): + oldrev = subprocess.run( + ["git", "--git-dir", str(hub), "rev-parse", "refs/heads/main"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + newrev = subprocess.run( + ["git", "-C", str(clone), "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + operation = { + "type": "protected_ref_update", + "repository": repo_name, + "refname": "refs/heads/main", + "oldrev": oldrev, + "newrev": newrev, + } + approval = gate.request_approval(conn, operation, requested_by=operator) + gate.vote(conn, approval["id"], operator, True) + conn.commit() + return approval, operation + + +def test_successful_push_completes_reservation_and_consumes_approval( + committed_conn, tmp_path, cfg +): + repo_name, hub, clone, owner, operator = _setup(committed_conn, tmp_path, cfg) + _commit(clone, "reserved-success.txt") + approval, operation = _approve_main_update( + committed_conn, repo_name, hub, clone, operator + ) + + result = _push(clone, owner, "main", cfg=cfg) + assert result.returncode == 0, result.stderr + + completed = committed_conn.execute( + """ + SELECT status, approval_id, completed_at + FROM git_mutations + WHERE repository_id = (SELECT id FROM repositories WHERE name = ?) + AND refname = 'refs/heads/main' + ORDER BY id DESC LIMIT 1 + """, + (repo_name,), + ).fetchone() + assert completed is not None + assert completed[0] == "completed" + assert completed[1] == approval["id"] + assert completed[2] is not None + assert gate.get_approval_by_id(committed_conn, approval["id"])["status"] == ( + "consumed" + ) + assert subprocess.run( + ["git", "--git-dir", str(hub), "rev-parse", "refs/heads/main"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() == operation["newrev"] + + +def test_rejected_update_keeps_approval_and_retry_reuses_reservation( + committed_conn, tmp_path, cfg +): + repo_name, hub, clone, owner, operator = _setup(committed_conn, tmp_path, cfg) + _commit(clone, "reserved-retry.txt") + approval, operation = _approve_main_update( + committed_conn, repo_name, hub, clone, operator + ) + + update_hook = hub / "hooks" / "update" + update_hook.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8") + update_hook.chmod(0o755) + rejected = _push(clone, owner, "main", cfg=cfg) + assert rejected.returncode != 0 + assert gate.get_approval_by_id(committed_conn, approval["id"])["status"] == ( + "approved" + ) + reservation = committed_conn.execute( + """ + SELECT id, status, approval_id FROM git_mutations + WHERE repository_id = (SELECT id FROM repositories WHERE name = ?) + AND refname = 'refs/heads/main' AND status = 'reserved' + """, + (repo_name,), + ).fetchone() + assert reservation is not None and reservation[2] == approval["id"] + assert subprocess.run( + ["git", "--git-dir", str(hub), "rev-parse", "refs/heads/main"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() == operation["oldrev"] + + update_hook.unlink() + retried = _push(clone, owner, "main", cfg=cfg) + assert retried.returncode == 0, retried.stderr + rows = committed_conn.execute( + """ + SELECT id, status FROM git_mutations + WHERE repository_id = (SELECT id FROM repositories WHERE name = ?) + AND refname = 'refs/heads/main' + ORDER BY id + """, + (repo_name,), + ).fetchall() + assert rows == [(reservation[0], "completed")] + assert gate.get_approval_by_id(committed_conn, approval["id"])["status"] == ( + "consumed" + ) + + +def test_inflight_branch_freezes_claim_transition(tmp_path, monkeypatch): + """The reservation freezes ownership in an isolated authority store.""" + local_cfg = Config(data_dir=tmp_path / "mutation-data", agent=None) + store.migrate(local_cfg) + store.verify_contract(local_cfg) + conn = store.connect(local_cfg) + try: + repo_name, hub, clone, owner, _operator = _setup(conn, tmp_path, local_cfg) + task = work.create_task(conn, repo_name, "reserved branch") + claim_id, _, _ = work.claim_task( + conn, + task, + owner, + branch="feat/reserved", + scope_globs=["src/**"], + ) + conn.commit() + + _commit(clone, "src/reserved.py", branch="feat/reserved") + newrev = subprocess.run( + ["git", "-C", str(clone), "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + zero = "0" * 40 + line = f"{zero} {newrev} refs/heads/feat/reserved\n" + monkeypatch.setenv("GIT_DIR", str(hub)) + monkeypatch.setenv("QUORUMGIT_AGENT", owner) + assert receive.run_pre_receive(conn, repo_name, [line]) == 0 + + blocked = store.connect(local_cfg) + try: + with pytest.raises(ValueError, match="in-flight Git mutation"): + work.release_claim(blocked, claim_id, owner) + blocked.rollback() + finally: + blocked.close() + + # A real receive has already placed incoming objects in Git's object + # quarantine before pre-receive runs. This direct hook test mirrors that + # condition by fetching the committed object into the bare hub before + # updating its ref manually. + subprocess.run( + ["git", "--git-dir", str(hub), "fetch", str(clone), newrev], + check=True, + capture_output=True, + ) + subprocess.run( + [ + "git", + "--git-dir", + str(hub), + "update-ref", + "refs/heads/feat/reserved", + newrev, + ], + check=True, + capture_output=True, + ) + assert receive.run_post_receive(conn, repo_name, [line]) == 0 + + after = store.connect(local_cfg) + try: + work.release_claim(after, claim_id, owner) + after.commit() + finally: + after.close() + finally: + conn.close() + store.destroy(local_cfg) + + +def test_install_refuses_unrelated_post_hook_without_changing_it(conn, tmp_path): + seed = make_git_repo(tmp_path / "post-seed") + hub = tmp_path / "post-hub.git" + subprocess.run( + ["git", "clone", "--bare", str(seed), str(hub)], + check=True, + capture_output=True, + ) + repo_name = f"post-{uuid.uuid4().hex[:8]}" + registry.add_repository(conn, repo_name, hub) + post = hub / "hooks" / "post-receive" + original = "#!/bin/sh\necho existing-post\n" + post.write_text(original, encoding="utf-8") + post.chmod(0o755) + + with pytest.raises(receive.ReceiveError, match="not owned by QuorumGit"): + receive.install_hooks(conn, repo_name) + assert post.read_text(encoding="utf-8") == original + assert not (hub / "hooks" / "pre-receive").exists()