solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import base64
7import hashlib
8import json
9import subprocess
10from collections.abc import Mapping, Sequence
11from datetime import UTC, datetime
12from pathlib import Path
13from typing import Any
14
15import pytest
16
17import scripts.advisory_mirror_audit as audit
18from scripts.transparency_signing import FakeTransparencySigner
19
20NOW = datetime(2026, 7, 24, 12, 0, tzinfo=UTC)
21RECEIPT_UTC = "2026-07-24T11:30:00Z"
22TEST_KEY_ID = "A1B2C3D4E5F60708"
23DERIVED_NAME = "rustsec-advisory-db.git-testderived"
24
25
26def _run_git(repo: Path, argv: Sequence[str]) -> str:
27 result = subprocess.run(
28 ["git", *argv],
29 cwd=repo,
30 capture_output=True,
31 text=True,
32 check=False,
33 )
34 assert result.returncode == 0, result.stderr
35 return result.stdout.strip()
36
37
38def _audit_root(tmp_path: Path) -> Path:
39 root = tmp_path / "root"
40 (root / "core").mkdir(parents=True)
41 (root / "core" / "deny.toml").write_text(
42 '[licenses]\nallow = ["MIT"]\n', encoding="utf-8"
43 )
44 (root / "core" / "Cargo.lock").write_text("fixture lock\n", encoding="utf-8")
45 (root / "core" / "Cargo.toml").write_text(
46 "[workspace]\nmembers = []\n",
47 encoding="utf-8",
48 )
49 (root / "target" / "release-evidence").mkdir(parents=True)
50 (root / "target" / "release-evidence" / "preserve.txt").write_text(
51 "release evidence\n",
52 encoding="utf-8",
53 )
54 (root / "dist" / "release-candidate").mkdir(parents=True)
55 (root / "dist" / "release-candidate" / "preserve.txt").write_text(
56 "release candidate\n",
57 encoding="utf-8",
58 )
59 return root
60
61
62def _advisory_repo(
63 tmp_path: Path, *, advisory_count: int = 1
64) -> tuple[Path, str, Path]:
65 repo = tmp_path / f"repo-{advisory_count}"
66 repo.mkdir()
67 _run_git(repo, ["init", "-b", "main"])
68 _run_git(repo, ["config", "user.name", "Audit Test"])
69 _run_git(repo, ["config", "user.email", "audit-test@example.invalid"])
70 if advisory_count:
71 for index in range(advisory_count):
72 path = repo / "crates" / f"probe{index}" / f"RUSTSEC-2026-{index:04d}.md"
73 path.parent.mkdir(parents=True)
74 path.write_text(
75 "```toml\n"
76 "[advisory]\n"
77 f'id = "RUSTSEC-2026-{index:04d}"\n'
78 f'package = "probe{index}"\n'
79 'date = "2026-01-01"\n'
80 'url = "https://example.invalid/RUSTSEC-2026-0001"\n'
81 'categories = ["unmaintained"]\n'
82 "keywords = []\n\n"
83 "[versions]\n"
84 "patched = []\n"
85 "```\n",
86 encoding="utf-8",
87 )
88 else:
89 (repo / "README.md").write_text("empty advisory db\n", encoding="utf-8")
90 _run_git(repo, ["add", "."])
91 _run_git(repo, ["commit", "-m", "fixture advisory db"])
92 commit = _run_git(repo, ["rev-parse", "HEAD"])
93 bundle = tmp_path / f"advisory-{advisory_count}.bundle"
94 _run_git(repo, ["bundle", "create", str(bundle), "HEAD", "refs/heads/main"])
95 return repo, commit, bundle
96
97
98def _pubkey_bytes(key_id: str = TEST_KEY_ID) -> bytes:
99 raw_id = bytes.fromhex(key_id)[::-1]
100 blob = b"Ed" + raw_id + (b"\x11" * 32)
101 return (
102 f"untrusted comment: minisign public key {key_id}\n".encode("ascii")
103 + base64.b64encode(blob)
104 + b"\n"
105 )
106
107
108def _write_pubkey(path: Path, *, key_id: str = TEST_KEY_ID) -> str:
109 raw = _pubkey_bytes(key_id)
110 path.write_bytes(raw)
111 return hashlib.sha256(raw).hexdigest()
112
113
114def _receipt_bytes(
115 commit: str, utc: str = RECEIPT_UTC, *, max_age: int = 86400
116) -> bytes:
117 return (
118 f'{{"max_age":{max_age},"synced_commit":"{commit}","utc":"{utc}"}}\n'
119 ).encode("utf-8")
120
121
122def _trusted_comment(commit: str, utc: str = RECEIPT_UTC) -> str:
123 return (
124 f"{audit.TRUSTED_COMMENT_SCHEME} synced_commit={commit} utc={utc} max_age=86400"
125 )
126
127
128def _write_packet(
129 tmp_path: Path,
130 *,
131 commit: str,
132 utc: str = RECEIPT_UTC,
133 key_id: str = TEST_KEY_ID,
134 signer: FakeTransparencySigner | None = None,
135 trusted_comment: str | None = None,
136) -> tuple[Path, Path, Path, str, FakeTransparencySigner]:
137 pubkey = tmp_path / "pub.key"
138 pubkey_sha = _write_pubkey(pubkey, key_id=key_id)
139 receipt = tmp_path / "freshness.json"
140 receipt.write_bytes(_receipt_bytes(commit, utc))
141 signature = receipt.parent / f"{receipt.name}.minisig"
142 fake = signer or FakeTransparencySigner()
143 fake.sign_file(
144 receipt,
145 signature,
146 trusted_comment=trusted_comment or _trusted_comment(commit, utc),
147 )
148 return receipt, signature, pubkey, pubkey_sha, fake
149
150
151class NoCallRunner:
152 events: list[list[str]]
153
154 def __init__(self) -> None:
155 self.events = []
156
157 def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]:
158 self.events.append(list(argv))
159 raise AssertionError(f"unexpected command: {argv}")
160
161
162class HybridRunner:
163 def __init__(
164 self,
165 *,
166 derived_name: str = DERIVED_NAME,
167 discovery_exit: int = 1,
168 final_exit: int = 0,
169 final_stderr_extra: str = "",
170 final_scanned_path: Path | None = None,
171 version: str = audit.CARGO_DENY_VERSION,
172 ) -> None:
173 self.derived_name = derived_name
174 self.discovery_exit = discovery_exit
175 self.final_exit = final_exit
176 self.final_stderr_extra = final_stderr_extra
177 self.final_scanned_path = final_scanned_path
178 self.version = version
179 self.events: list[list[str]] = []
180 self.cargo_envs: list[Mapping[str, str]] = []
181 self.cargo_cwds: list[Path | None] = []
182 self.config_bytes: bytes | None = None
183 self.check_count = 0
184
185 def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]:
186 command = list(argv)
187 self.events.append(command)
188 if command[0] == "git":
189 return subprocess.run(command, **kwargs)
190 if command[0] == "cargo-deny":
191 if command == ["cargo-deny", "--version"]:
192 return subprocess.CompletedProcess(
193 command, 0, f"cargo-deny {self.version}\n", ""
194 )
195 if command[-2:] == ["check", "advisories"]:
196 self.check_count += 1
197 self.cargo_envs.append(kwargs.get("env", {}))
198 self.cargo_cwds.append(kwargs.get("cwd"))
199 config_path = Path(command[command.index("--config") + 1])
200 self.config_bytes = config_path.read_bytes()
201 db_parent = _config_db_path(config_path)
202 scanned = self.final_scanned_path if self.check_count == 2 else None
203 if scanned is None:
204 scanned = db_parent / self.derived_name
205 stderr = (
206 f"2026-07-24 [DEBUG] Opening advisory database at '{scanned}'\n"
207 )
208 if self.check_count == 1:
209 return subprocess.CompletedProcess(
210 command, self.discovery_exit, "", stderr
211 )
212 stderr += self.final_stderr_extra
213 return subprocess.CompletedProcess(command, self.final_exit, "", stderr)
214 raise AssertionError(f"unexpected cargo-deny command: {command}")
215 raise AssertionError(f"unexpected command: {command}")
216
217
218class FakeGitRunner:
219 def __init__(
220 self,
221 *,
222 heads_commit: str,
223 clone_commit: str | None = None,
224 fail: str | None = None,
225 ) -> None:
226 self.heads_commit = heads_commit
227 self.clone_commit = clone_commit or heads_commit
228 self.fail = fail
229 self.events: list[list[str]] = []
230
231 def __call__(self, argv, **kwargs) -> subprocess.CompletedProcess[str]:
232 command = list(argv)
233 self.events.append(command)
234 if self.fail and self.fail in " ".join(command):
235 return subprocess.CompletedProcess(command, 1, "", "failed")
236 if command == ["cargo-deny", "--version"]:
237 return subprocess.CompletedProcess(command, 0, "cargo-deny 0.20.2\n", "")
238 if command[:3] == ["git", "bundle", "verify"]:
239 return subprocess.CompletedProcess(command, 0, "", "")
240 if command[:3] == ["git", "bundle", "list-heads"]:
241 return subprocess.CompletedProcess(
242 command,
243 0,
244 f"{self.heads_commit} HEAD\n{self.heads_commit} refs/heads/main\n",
245 "",
246 )
247 if command[:2] == ["git", "clone"]:
248 Path(command[3]).mkdir(parents=True)
249 return subprocess.CompletedProcess(command, 0, "", "")
250 if command[:3] == ["git", "-C", command[2]] and command[-2:] == [
251 "rev-parse",
252 "HEAD",
253 ]:
254 return subprocess.CompletedProcess(command, 0, self.clone_commit + "\n", "")
255 if command[-2:] == ["check", "advisories"]:
256 config_path = Path(command[command.index("--config") + 1])
257 db_parent = _config_db_path(config_path)
258 return subprocess.CompletedProcess(
259 command,
260 1,
261 "",
262 f"Opening advisory database at '{db_parent / DERIVED_NAME}'\n",
263 )
264 raise AssertionError(f"unexpected command: {command}")
265
266
267def _config_db_path(config_path: Path) -> Path:
268 import tomllib
269
270 parsed = tomllib.loads(config_path.read_text(encoding="utf-8"))
271 return Path(parsed["advisories"]["db-path"])
272
273
274def _invoke_green(
275 tmp_path: Path,
276 *,
277 runner: HybridRunner | None = None,
278 advisory_count: int = 1,
279 utc: str = RECEIPT_UTC,
280) -> tuple[bytes, HybridRunner, Path, Path, Path, Path]:
281 root = _audit_root(tmp_path)
282 _repo, commit, bundle = _advisory_repo(tmp_path, advisory_count=advisory_count)
283 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
284 tmp_path,
285 commit=commit,
286 utc=utc,
287 )
288 active_runner = runner or HybridRunner()
289 output = audit.audit_advisory_mirror(
290 root,
291 bundle=bundle,
292 receipt=receipt,
293 pubkey=pubkey,
294 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
295 runner=active_runner,
296 verifier=fake,
297 clock=lambda: NOW,
298 pinned_key_id=TEST_KEY_ID,
299 pinned_pubkey_sha256=pubkey_sha,
300 )
301 return output, active_runner, root, bundle, receipt, pubkey
302
303
304def _inventory(root: Path) -> dict[str, tuple[str, int]]:
305 result: dict[str, tuple[str, int]] = {}
306 if not root.exists():
307 return result
308 for path in sorted(item for item in root.rglob("*") if item.is_file()):
309 rel = path.relative_to(root).as_posix()
310 raw = path.read_bytes()
311 result[rel] = (hashlib.sha256(raw).hexdigest(), len(raw))
312 return result
313
314
315def test_green_packet_emits_exact_success_json_and_uses_bound_snapshot(
316 tmp_path: Path,
317) -> None:
318 output, runner, root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path)
319
320 payload = json.loads(output)
321 assert list(payload) == [
322 "product",
323 "advisory_cohort",
324 "synced_commit",
325 "receipt_utc",
326 "max_age",
327 "checked_at",
328 "cargo_lock_sha256",
329 "cargo_deny_version",
330 "verdict",
331 ]
332 assert payload["product"] == "solstone-journal"
333 assert payload["advisory_cohort"] == audit.ADVISORY_COHORT_ID
334 assert payload["receipt_utc"] == RECEIPT_UTC
335 assert payload["max_age"] == 86400
336 assert payload["checked_at"] == "2026-07-24T12:00:00Z"
337 assert payload["cargo_deny_version"] == "0.20.2"
338 assert (
339 payload["cargo_lock_sha256"]
340 == hashlib.sha256((root / "core" / "Cargo.lock").read_bytes()).hexdigest()
341 )
342 expected = json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n"
343 assert output == expected
344 assert output.count(b"\n") == 1
345 assert runner.check_count == 2
346 assert all(env.get("CARGO_NET_OFFLINE") == "true" for env in runner.cargo_envs)
347 assert runner.config_bytes is not None
348 config_text = runner.config_bytes.decode("utf-8")
349 assert "git-fetch-with-cli" not in config_text
350 assert "maximum-db-staleness" not in config_text
351
352
353def test_green_packet_with_real_git_bundle_materialization(tmp_path: Path) -> None:
354 output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path)
355
356 assert json.loads(output)["verdict"] == "pass"
357 assert any(command[:3] == ["git", "bundle", "verify"] for command in runner.events)
358 assert any(
359 command[:3] == ["git", "bundle", "list-heads"] for command in runner.events
360 )
361 assert any(command[:2] == ["git", "clone"] for command in runner.events)
362 assert runner.check_count == 2
363
364
365@pytest.mark.parametrize("missing", ["bundle", "receipt", "pubkey", "locator"])
366def test_required_inputs_fail_before_git_or_cargo(tmp_path: Path, missing: str) -> None:
367 root = _audit_root(tmp_path)
368 _repo, commit, bundle = _advisory_repo(tmp_path)
369 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
370 tmp_path, commit=commit
371 )
372 if missing == "bundle":
373 bundle = tmp_path / "missing.bundle"
374 elif missing == "receipt":
375 receipt = tmp_path / "missing.json"
376 elif missing == "pubkey":
377 pubkey = tmp_path / "missing.pub"
378 locator = (
379 ""
380 if missing == "locator"
381 else "ssh://mirror.example.invalid/rustsec-advisory-db.git"
382 )
383 runner = NoCallRunner()
384
385 with pytest.raises(audit.ReleasePolicyError):
386 audit.audit_advisory_mirror(
387 root,
388 bundle=bundle,
389 receipt=receipt,
390 pubkey=pubkey,
391 locator=locator,
392 runner=runner,
393 verifier=fake,
394 clock=lambda: NOW,
395 pinned_key_id=TEST_KEY_ID,
396 pinned_pubkey_sha256=pubkey_sha,
397 )
398 assert runner.events == []
399
400
401@pytest.mark.parametrize("kind", ["missing", "symlink", "directory"])
402def test_adjacent_signature_is_required_and_regular_before_git_or_cargo(
403 tmp_path: Path,
404 kind: str,
405) -> None:
406 root = _audit_root(tmp_path)
407 _repo, commit, bundle = _advisory_repo(tmp_path)
408 receipt, signature, pubkey, pubkey_sha, fake = _write_packet(
409 tmp_path, commit=commit
410 )
411 signature.unlink()
412 if kind == "symlink":
413 target = tmp_path / "sig-target"
414 target.write_text("signature", encoding="utf-8")
415 signature.symlink_to(target)
416 elif kind == "directory":
417 signature.mkdir()
418 runner = NoCallRunner()
419
420 with pytest.raises(audit.ReleasePolicyError):
421 audit.audit_advisory_mirror(
422 root,
423 bundle=bundle,
424 receipt=receipt,
425 pubkey=pubkey,
426 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
427 runner=runner,
428 verifier=fake,
429 clock=lambda: NOW,
430 pinned_key_id=TEST_KEY_ID,
431 pinned_pubkey_sha256=pubkey_sha,
432 )
433 assert runner.events == []
434
435
436@pytest.mark.parametrize("target_name", ["bundle", "receipt", "pubkey"])
437@pytest.mark.parametrize("kind", ["missing", "symlink", "directory"])
438def test_unsafe_input_paths_and_symlinks_fail_before_git_or_cargo(
439 tmp_path: Path,
440 target_name: str,
441 kind: str,
442) -> None:
443 root = _audit_root(tmp_path)
444 _repo, commit, bundle = _advisory_repo(tmp_path)
445 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
446 tmp_path, commit=commit
447 )
448 paths = {"bundle": bundle, "receipt": receipt, "pubkey": pubkey}
449 target = paths[target_name]
450 if target.exists() or target.is_symlink():
451 target.unlink()
452 if kind == "symlink":
453 link_target = tmp_path / f"{target_name}-target"
454 link_target.write_text("target", encoding="utf-8")
455 target.symlink_to(link_target)
456 elif kind == "directory":
457 target.mkdir()
458 runner = NoCallRunner()
459
460 with pytest.raises(audit.ReleasePolicyError):
461 audit.audit_advisory_mirror(
462 root,
463 bundle=paths["bundle"],
464 receipt=paths["receipt"],
465 pubkey=paths["pubkey"],
466 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
467 runner=runner,
468 verifier=fake,
469 clock=lambda: NOW,
470 pinned_key_id=TEST_KEY_ID,
471 pinned_pubkey_sha256=pubkey_sha,
472 )
473 assert runner.events == []
474
475
476@pytest.mark.parametrize(
477 ("locator", "accepted"),
478 [
479 ("https://github.com/rustsec/advisory-db", False),
480 ("https://github.com/rustsec/advisory-db.git", False),
481 ("https://github.com/rustsec/advisory-db/", False),
482 ("https://github.com/rustsec/advisory-db.git/", False),
483 ("github.com:rustsec/advisory-db.git", False),
484 ("github.com:rustsec/advisory-db.git/", False),
485 ("ssh://github.com/rustsec/advisory-db.git", False),
486 ("ssh://github.com:22/rustsec/advisory-db.git", False),
487 ("git://github.com/rustsec/advisory-db.git", False),
488 ("http://github.com/rustsec/advisory-db", False),
489 ("git+ssh://github.com/rustsec/advisory-db.git", False),
490 ("https://raw.github.com/rustsec/advisory-db", False),
491 ("https://mirror.github.com/rustsec/advisory-db", False),
492 ("https://foo.github.com/rustsec/advisory-db.git", False),
493 ("https://github.com/rustsec/advisory-db?x=1", False),
494 ("https://github.com/rustsec/advisory-db#frag", False),
495 ("github.com/rustsec/advisory-db.git", False),
496 ("", False),
497 (" ", False),
498 ("https://mirror.example.invalid/rustsec/advisory-db", True),
499 ("https://mirror.example.invalid/rustsec/advisory-db.git", False),
500 ("https://mirror.example.invalid/rustsec/rustsec-advisory-db.git", True),
501 ("https://mirror.example.invalid/rustsec/rustsec-advisory-db", False),
502 ("https://mirror.example.invalid/rustsec/advisory-db/", False),
503 ("https://mirror.example.invalid/rustsec/rustsec-advisory-db.git/", False),
504 ("https://mirror.example.invalid/rustsec/advisory-db?x=1", False),
505 ("https://mirror.example.invalid/rustsec/advisory-db#frag", False),
506 ("ssh://git@mirror.example.invalid/rustsec/advisory-db", True),
507 ("git@mirror.example.invalid:rustsec/advisory-db", True),
508 ("git@mirror.example.invalid:rustsec/rustsec-advisory-db.git", True),
509 ("git@mirror.example.invalid:rustsec/advisory-db.git", False),
510 ("https://mirror.example.invalid/rustsec/advisory-db\n", False),
511 ("https://mirror.example.invalid/rustsec/advisory-db\t", False),
512 ("https://mirror.example.invalid/rustsec/advisory-db\x00", False),
513 ("https://github.com.evil/rustsec/advisory-db", True),
514 ],
515)
516def test_validate_locator_q3_oracle(locator: str, accepted: bool) -> None:
517 assert (audit.validate_locator(locator) == []) is accepted
518
519
520@pytest.mark.parametrize(
521 "raw",
522 [
523 b'{"synced_commit":"{commit}","max_age":86400,"utc":"2026-07-24T11:30:00Z"}\n',
524 b'{ "max_age": 86400, "synced_commit": "{commit}", "utc": "2026-07-24T11:30:00Z" }\n',
525 b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z"}',
526 b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z"}\n\n',
527 b'{"max_age":86400,"synced_commit":"{commit}","utc":"2026-07-24T11:30:00Z","x":1}\n',
528 ],
529)
530def test_receipt_body_requires_canonical_bytes(tmp_path: Path, raw: bytes) -> None:
531 _repo, commit, _bundle = _advisory_repo(tmp_path)
532 receipt = tmp_path / "freshness.json"
533 receipt.write_bytes(raw.replace(b"{commit}", commit.encode("ascii")))
534
535 with pytest.raises(audit.ReleasePolicyError):
536 audit._read_receipt_authority(receipt)
537
538
539@pytest.mark.parametrize(
540 "payload",
541 [
542 {"max_age": 1, "synced_commit": "a" * 40, "utc": RECEIPT_UTC},
543 {"max_age": True, "synced_commit": "a" * 40, "utc": RECEIPT_UTC},
544 {"max_age": "86400", "synced_commit": "a" * 40, "utc": RECEIPT_UTC},
545 {"max_age": 86400, "synced_commit": "a" * 64, "utc": RECEIPT_UTC},
546 {"max_age": 86400, "synced_commit": "A" * 40, "utc": RECEIPT_UTC},
547 {"max_age": 86400, "synced_commit": "a" * 40, "utc": "2026-99-99T00:00:00Z"},
548 {
549 "max_age": 86400,
550 "synced_commit": "a" * 40,
551 "utc": "2026-07-24T11:30:00+00:00",
552 },
553 ],
554)
555def test_receipt_fields_are_strict(tmp_path: Path, payload: dict[str, Any]) -> None:
556 receipt = tmp_path / "freshness.json"
557 receipt.write_text(
558 json.dumps(payload, separators=(",", ":")) + "\n", encoding="utf-8"
559 )
560
561 with pytest.raises(audit.ReleasePolicyError):
562 audit._read_receipt_authority(receipt)
563
564
565def test_trusted_comment_mismatch_fails(tmp_path: Path) -> None:
566 root = _audit_root(tmp_path)
567 _repo, commit, bundle = _advisory_repo(tmp_path)
568 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
569 tmp_path,
570 commit=commit,
571 trusted_comment="wrong",
572 )
573
574 with pytest.raises(audit.ReleasePolicyError) as exc:
575 audit.audit_advisory_mirror(
576 root,
577 bundle=bundle,
578 receipt=receipt,
579 pubkey=pubkey,
580 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
581 runner=NoCallRunner(),
582 verifier=fake,
583 clock=lambda: NOW,
584 pinned_key_id=TEST_KEY_ID,
585 pinned_pubkey_sha256=pubkey_sha,
586 )
587 assert "trusted comment mismatch" in exc.value.failures[0].error
588
589
590def test_pubkey_sha256_mismatch_fails_before_minisign(tmp_path: Path) -> None:
591 _repo, commit, _bundle = _advisory_repo(tmp_path)
592 _receipt, _signature, pubkey, _pubkey_sha, _fake = _write_packet(
593 tmp_path, commit=commit
594 )
595
596 with pytest.raises(audit.ReleasePolicyError):
597 audit._validate_pubkey_binding(
598 pubkey,
599 pinned_key_id=TEST_KEY_ID,
600 pinned_pubkey_sha256="0" * 64,
601 )
602
603
604def test_pubkey_key_id_mismatch_fails_before_minisign(tmp_path: Path) -> None:
605 pubkey = tmp_path / "pub.key"
606 pubkey_sha = _write_pubkey(pubkey)
607
608 with pytest.raises(audit.ReleasePolicyError):
609 audit._validate_pubkey_binding(
610 pubkey,
611 pinned_key_id="0000000000000000",
612 pinned_pubkey_sha256=pubkey_sha,
613 )
614
615
616@pytest.mark.parametrize(
617 "raw",
618 [
619 b"untrusted comment\nnot base64\n",
620 b"untrusted comment\nRWQ=\n",
621 b"untrusted comment\n" + base64.b64encode(b"XX" + b"\x00" * 40) + b"\n",
622 b"only one line\n",
623 ],
624)
625def test_pubkey_blob_shape_is_strict(tmp_path: Path, raw: bytes) -> None:
626 pubkey = tmp_path / "pub.key"
627 pubkey.write_bytes(raw)
628 pubkey_sha = hashlib.sha256(raw).hexdigest()
629
630 with pytest.raises(audit.ReleasePolicyError):
631 audit._validate_pubkey_binding(
632 pubkey,
633 pinned_key_id=TEST_KEY_ID,
634 pinned_pubkey_sha256=pubkey_sha,
635 )
636
637
638def test_signature_mutation_fails(tmp_path: Path) -> None:
639 root = _audit_root(tmp_path)
640 _repo, commit, bundle = _advisory_repo(tmp_path)
641 receipt, signature, pubkey, pubkey_sha, fake = _write_packet(
642 tmp_path, commit=commit
643 )
644 signature.write_text(
645 signature.read_text(encoding="utf-8").replace("=", "A", 1), encoding="utf-8"
646 )
647
648 with pytest.raises(audit.ReleasePolicyError):
649 audit.audit_advisory_mirror(
650 root,
651 bundle=bundle,
652 receipt=receipt,
653 pubkey=pubkey,
654 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
655 runner=NoCallRunner(),
656 verifier=fake,
657 clock=lambda: NOW,
658 pinned_key_id=TEST_KEY_ID,
659 pinned_pubkey_sha256=pubkey_sha,
660 )
661
662
663@pytest.mark.parametrize(
664 "utc",
665 ["2026-07-24T12:06:00Z", "2026-07-23T11:59:59Z"],
666)
667def test_receipt_future_and_stale_times_fail(tmp_path: Path, utc: str) -> None:
668 _repo, commit, _bundle = _advisory_repo(tmp_path)
669 receipt = tmp_path / "freshness.json"
670 receipt.write_bytes(_receipt_bytes(commit, utc))
671 authority = audit._read_receipt_authority(receipt)
672
673 with pytest.raises(audit.ReleasePolicyError):
674 audit._validate_receipt_freshness(authority, clock=lambda: NOW)
675
676
677def test_minisign_preflight_uses_product_binary_check(
678 tmp_path: Path,
679 monkeypatch: pytest.MonkeyPatch,
680) -> None:
681 calls: list[tuple[str, object]] = []
682
683 def check(minisign: str) -> str:
684 calls.append(("check", minisign))
685 return "minisign 0.12"
686
687 class Local:
688 def __init__(
689 self, *, secret_key: Path, public_key: Path, minisign: str
690 ) -> None:
691 calls.append(("init", (secret_key, public_key, minisign)))
692
693 def verify_file(
694 self,
695 message_path: Path,
696 signature_path: Path,
697 *,
698 expected_trusted_comment: str,
699 ) -> None:
700 calls.append(
701 ("verify", (message_path, signature_path, expected_trusted_comment))
702 )
703
704 monkeypatch.setattr(audit, "check_minisign_binary", check)
705 monkeypatch.setattr(audit, "LocalMinisignSigner", Local)
706 verifier = audit.PublicKeyMinisignVerifier(
707 tmp_path / "pub.key", minisign="minisign-test"
708 )
709 verifier.check()
710 verifier.verify_file(
711 tmp_path / "freshness.json",
712 tmp_path / "freshness.json.minisig",
713 expected_trusted_comment="comment",
714 )
715
716 assert calls[0] == ("check", "minisign-test")
717 assert calls[1][0] == "init"
718 assert calls[2][0] == "verify"
719
720
721def test_bundle_verify_failure_stops_before_clone_and_cargo(tmp_path: Path) -> None:
722 root = _audit_root(tmp_path)
723 _repo, commit, bundle = _advisory_repo(tmp_path)
724 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
725 tmp_path, commit=commit
726 )
727 runner = FakeGitRunner(heads_commit=commit, fail="bundle verify")
728
729 with pytest.raises(audit.ReleasePolicyError):
730 audit.audit_advisory_mirror(
731 root,
732 bundle=bundle,
733 receipt=receipt,
734 pubkey=pubkey,
735 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
736 runner=runner,
737 verifier=fake,
738 clock=lambda: NOW,
739 pinned_key_id=TEST_KEY_ID,
740 pinned_pubkey_sha256=pubkey_sha,
741 )
742 assert not any(command[:2] == ["git", "clone"] for command in runner.events)
743 assert not any(command[-2:] == ["check", "advisories"] for command in runner.events)
744
745
746@pytest.mark.parametrize(
747 "stdout",
748 [
749 "a" * 40 + " HEAD\n",
750 "a" * 40 + " refs/heads/main\n",
751 "a" * 40 + " HEAD\n" + "a" * 40 + " refs/heads/main\n" + "a" * 40 + " refs/x\n",
752 "b" * 40 + " HEAD\n" + "a" * 40 + " refs/heads/main\n",
753 "malformed\n" + "a" * 40 + " refs/heads/main\n",
754 ],
755)
756def test_bundle_heads_must_be_exact_head_and_main(stdout: str) -> None:
757 with pytest.raises(audit.ReleasePolicyError):
758 audit._parse_bundle_heads(stdout, synced_commit="a" * 40)
759
760
761def test_clone_head_must_match_receipt_commit(tmp_path: Path) -> None:
762 root = _audit_root(tmp_path)
763 _repo, commit, bundle = _advisory_repo(tmp_path)
764 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
765 tmp_path, commit=commit
766 )
767 runner = FakeGitRunner(heads_commit=commit, clone_commit="b" * 40)
768
769 with pytest.raises(audit.ReleasePolicyError):
770 audit.audit_advisory_mirror(
771 root,
772 bundle=bundle,
773 receipt=receipt,
774 pubkey=pubkey,
775 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
776 runner=runner,
777 verifier=fake,
778 clock=lambda: NOW,
779 pinned_key_id=TEST_KEY_ID,
780 pinned_pubkey_sha256=pubkey_sha,
781 )
782
783
784def test_zero_advisory_clone_fails_before_cargo(tmp_path: Path) -> None:
785 root = _audit_root(tmp_path)
786 _repo, commit, bundle = _advisory_repo(tmp_path, advisory_count=0)
787 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
788 tmp_path, commit=commit
789 )
790 runner = HybridRunner()
791
792 with pytest.raises(audit.ReleasePolicyError):
793 audit.audit_advisory_mirror(
794 root,
795 bundle=bundle,
796 receipt=receipt,
797 pubkey=pubkey,
798 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
799 runner=runner,
800 verifier=fake,
801 clock=lambda: NOW,
802 pinned_key_id=TEST_KEY_ID,
803 pinned_pubkey_sha256=pubkey_sha,
804 )
805 assert runner.check_count == 0
806
807
808def test_discovery_run_nonzero_is_expected_when_debug_line_present(
809 tmp_path: Path,
810) -> None:
811 output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green(
812 tmp_path,
813 runner=HybridRunner(discovery_exit=23),
814 )
815
816 assert json.loads(output)["verdict"] == "pass"
817 assert runner.check_count == 2
818
819
820@pytest.mark.parametrize("scanned", [Path("/outside/db"), Path("nested/child")])
821def test_discovery_path_must_be_direct_child_of_temp_parent(
822 tmp_path: Path,
823 scanned: Path,
824) -> None:
825 parent = tmp_path / "parent"
826 parent.mkdir()
827 if not scanned.is_absolute():
828 scanned = parent / scanned
829
830 with pytest.raises(audit.ReleasePolicyError):
831 audit._assert_direct_child(scanned, parent)
832
833
834def test_discovered_path_must_not_preexist(tmp_path: Path) -> None:
835 root = _audit_root(tmp_path)
836 _repo, commit, bundle = _advisory_repo(tmp_path)
837 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
838 tmp_path, commit=commit
839 )
840 temp_root = tmp_path / "temp"
841 preexisting = temp_root / "db-root" / DERIVED_NAME
842 preexisting.mkdir(parents=True)
843
844 with pytest.raises(audit.ReleasePolicyError):
845 audit.audit_advisory_mirror(
846 root,
847 bundle=bundle,
848 receipt=receipt,
849 pubkey=pubkey,
850 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
851 runner=HybridRunner(),
852 verifier=fake,
853 clock=lambda: NOW,
854 temp_path_factory=lambda _label: temp_root,
855 cleanup_rmdir=lambda path: None,
856 pinned_key_id=TEST_KEY_ID,
857 pinned_pubkey_sha256=pubkey_sha,
858 )
859
860
861def test_alternate_or_ambient_database_substitution_is_rejected(tmp_path: Path) -> None:
862 runner = HybridRunner(final_scanned_path=tmp_path / "other-db")
863
864 with pytest.raises(audit.ReleasePolicyError) as exc:
865 _invoke_green(tmp_path, runner=runner)
866 assert exc.value.failures[0].actual == "redacted"
867
868
869def test_runner_never_sees_remote_git_or_cargo_fetch_operations(tmp_path: Path) -> None:
870 _output, runner, _root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path)
871 flattened = [" ".join(command) for command in runner.events]
872
873 assert not any("fetch db" in item for item in flattened)
874 assert not any(item.startswith("git fetch") for item in flattened)
875 assert not any(item.startswith("git pull") for item in flattened)
876 assert not any(item.startswith("git ls-remote") for item in flattened)
877 assert not any("github.com" in item for item in flattened)
878
879
880def test_final_cargo_deny_failure_is_redacted_and_no_success(tmp_path: Path) -> None:
881 runner = HybridRunner(
882 final_exit=1,
883 final_stderr_extra=(
884 "/private/path TOKEN=abc ghp_abcdefghijklmnopqrst localhost "
885 "ssh://mirror.example.invalid/rustsec-advisory-db.git"
886 ),
887 )
888
889 with pytest.raises(audit.ReleasePolicyError) as exc:
890 _invoke_green(tmp_path, runner=runner)
891 text = "\n".join(failure.actual for failure in exc.value.failures)
892 assert "mirror.example.invalid" not in text
893 assert "/private/path" not in text
894 assert "TOKEN=" not in text
895 assert "ghp_" not in text
896
897
898def test_child_output_redaction_masks_locator_temp_path_and_token_canaries() -> None:
899 redacted = audit._redact_child_output(
900 "secret /tmp/private TOKEN=value ghp_abcdefghijklmnopqrst host.local",
901 secrets={"/tmp/private"},
902 )
903
904 assert "/tmp/private" not in redacted
905 assert "TOKEN=" not in redacted
906 assert "ghp_" not in redacted
907 assert audit.validate_public_evidence_text("child-output", redacted) == []
908
909
910def test_cleanup_failure_suppresses_success_and_combines_errors(tmp_path: Path) -> None:
911 def fail_cleanup(_path: Path) -> None:
912 raise OSError("cleanup failed")
913
914 root = _audit_root(tmp_path / "cleanup")
915 _repo, commit, bundle = _advisory_repo(tmp_path / "cleanup")
916 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
917 tmp_path / "cleanup", commit=commit
918 )
919 with pytest.raises(audit.ReleasePolicyError) as cleanup_exc:
920 audit.audit_advisory_mirror(
921 root,
922 bundle=bundle,
923 receipt=receipt,
924 pubkey=pubkey,
925 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
926 runner=HybridRunner(),
927 verifier=fake,
928 clock=lambda: NOW,
929 cleanup_rmdir=fail_cleanup,
930 pinned_key_id=TEST_KEY_ID,
931 pinned_pubkey_sha256=pubkey_sha,
932 )
933 assert any(
934 "cleanup failed" in failure.error for failure in cleanup_exc.value.failures
935 )
936
937
938def test_cleanup_failure_combines_with_primary_error(tmp_path: Path) -> None:
939 def fail_cleanup(_path: Path) -> None:
940 raise OSError("cleanup failed")
941
942 root = _audit_root(tmp_path)
943 _repo, commit, bundle = _advisory_repo(tmp_path)
944 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
945 tmp_path, commit=commit
946 )
947
948 with pytest.raises(audit.ReleasePolicyError) as exc:
949 audit.audit_advisory_mirror(
950 root,
951 bundle=bundle,
952 receipt=receipt,
953 pubkey=pubkey,
954 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
955 runner=HybridRunner(final_exit=1),
956 verifier=fake,
957 clock=lambda: NOW,
958 cleanup_rmdir=fail_cleanup,
959 pinned_key_id=TEST_KEY_ID,
960 pinned_pubkey_sha256=pubkey_sha,
961 )
962 errors = [failure.error for failure in exc.value.failures]
963 assert "advisory mirror cargo-deny final check failed" in errors
964 assert any("cleanup failed" in error for error in errors)
965
966
967def test_exact_success_schema_and_witness_binding(tmp_path: Path) -> None:
968 output, _runner, root, _bundle, _receipt, _pubkey = _invoke_green(tmp_path)
969 payload = json.loads(output)
970
971 assert payload == {
972 "product": "solstone-journal",
973 "advisory_cohort": audit.ADVISORY_COHORT_ID,
974 "synced_commit": payload["synced_commit"],
975 "receipt_utc": RECEIPT_UTC,
976 "max_age": 86400,
977 "checked_at": "2026-07-24T12:00:00Z",
978 "cargo_lock_sha256": hashlib.sha256(
979 (root / "core" / "Cargo.lock").read_bytes()
980 ).hexdigest(),
981 "cargo_deny_version": "0.20.2",
982 "verdict": "pass",
983 }
984 serialized = json.dumps(payload, separators=(",", ":")).encode("utf-8") + b"\n"
985 assert output == serialized
986 assert b"mirror.example.invalid" not in output
987
988
989def test_success_inventory_is_non_destructive(
990 tmp_path: Path,
991 monkeypatch: pytest.MonkeyPatch,
992) -> None:
993 root = _audit_root(tmp_path)
994 packet_root = tmp_path / "packet"
995 packet_root.mkdir()
996 cargo_home = tmp_path / "cargo-home"
997 cargo_home.mkdir()
998 (cargo_home / "preserve.txt").write_text("ambient cargo\n", encoding="utf-8")
999 monkeypatch.setenv("CARGO_HOME", str(cargo_home))
1000 temp_root = tmp_path / "audit-temp"
1001 _repo, commit, bundle = _advisory_repo(packet_root)
1002 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
1003 packet_root, commit=commit
1004 )
1005 before_root = _inventory(root)
1006 before_packet = _inventory(packet_root)
1007 before_cargo = _inventory(cargo_home)
1008
1009 audit.audit_advisory_mirror(
1010 root,
1011 bundle=bundle,
1012 receipt=receipt,
1013 pubkey=pubkey,
1014 locator="ssh://mirror.example.invalid/rustsec-advisory-db.git",
1015 runner=HybridRunner(),
1016 verifier=fake,
1017 clock=lambda: NOW,
1018 temp_path_factory=lambda _label: temp_root,
1019 pinned_key_id=TEST_KEY_ID,
1020 pinned_pubkey_sha256=pubkey_sha,
1021 )
1022
1023 assert _inventory(root) == before_root
1024 assert _inventory(packet_root) == before_packet
1025 assert _inventory(cargo_home) == before_cargo
1026 assert not temp_root.exists()
1027
1028
1029@pytest.mark.parametrize(
1030 "stage",
1031 ["input", "locator", "pubkey", "signature", "time", "bundle", "cargo", "cleanup"],
1032)
1033def test_failure_inventory_is_non_destructive(
1034 tmp_path: Path,
1035 monkeypatch: pytest.MonkeyPatch,
1036 stage: str,
1037) -> None:
1038 root = _audit_root(tmp_path)
1039 packet_root = tmp_path / "packet"
1040 packet_root.mkdir()
1041 cargo_home = tmp_path / "cargo-home"
1042 cargo_home.mkdir()
1043 (cargo_home / "preserve.txt").write_text("ambient cargo\n", encoding="utf-8")
1044 monkeypatch.setenv("CARGO_HOME", str(cargo_home))
1045 temp_root = tmp_path / "audit-temp"
1046 _repo, commit, bundle = _advisory_repo(packet_root)
1047 receipt, _signature, pubkey, pubkey_sha, fake = _write_packet(
1048 packet_root, commit=commit
1049 )
1050 locator = "ssh://mirror.example.invalid/rustsec-advisory-db.git"
1051 runner: object = HybridRunner(final_exit=1)
1052 pinned_pubkey_sha = pubkey_sha
1053 cleanup_rmdir = audit._remove_tree
1054 if stage == "input":
1055 bundle = packet_root / "missing.bundle"
1056 runner = NoCallRunner()
1057 elif stage == "locator":
1058 locator = ""
1059 runner = NoCallRunner()
1060 elif stage == "pubkey":
1061 pinned_pubkey_sha = "0" * 64
1062 runner = NoCallRunner()
1063 elif stage == "signature":
1064 (receipt.parent / f"{receipt.name}.minisig").write_text(
1065 "bad signature\n",
1066 encoding="utf-8",
1067 )
1068 runner = NoCallRunner()
1069 elif stage == "time":
1070 utc = "2026-07-24T12:06:00Z"
1071 receipt.write_bytes(_receipt_bytes(commit, utc))
1072 fake.sign_file(
1073 receipt,
1074 receipt.parent / f"{receipt.name}.minisig",
1075 trusted_comment=_trusted_comment(commit, utc),
1076 )
1077 runner = NoCallRunner()
1078 elif stage == "bundle":
1079 runner = FakeGitRunner(heads_commit=commit, fail="bundle verify")
1080 elif stage == "cleanup":
1081
1082 def fail_cleanup(_path: Path) -> None:
1083 raise OSError("cleanup failed")
1084
1085 runner = HybridRunner()
1086 cleanup_rmdir = fail_cleanup
1087 before_root = _inventory(root)
1088 before_packet = _inventory(packet_root)
1089 before_cargo = _inventory(cargo_home)
1090
1091 with pytest.raises(audit.ReleasePolicyError):
1092 audit.audit_advisory_mirror(
1093 root,
1094 bundle=bundle,
1095 receipt=receipt,
1096 pubkey=pubkey,
1097 locator=locator,
1098 runner=runner,
1099 verifier=fake,
1100 clock=lambda: NOW,
1101 temp_path_factory=lambda _label: temp_root,
1102 cleanup_rmdir=cleanup_rmdir,
1103 pinned_key_id=TEST_KEY_ID,
1104 pinned_pubkey_sha256=pinned_pubkey_sha,
1105 )
1106
1107 assert _inventory(root) == before_root
1108 assert _inventory(packet_root) == before_packet
1109 assert _inventory(cargo_home) == before_cargo
1110 if stage == "cleanup":
1111 assert temp_root.exists()
1112 else:
1113 assert not temp_root.exists()
1114
1115
1116def test_audit_config_bytes_omits_fetch_head_and_staleness_fields(
1117 tmp_path: Path,
1118) -> None:
1119 cfg = audit.audit_config_bytes(
1120 b'[licenses]\nallow = ["MIT"]\n',
1121 db_root=tmp_path / "db-root",
1122 db_urls=("ssh://mirror.example.invalid/rustsec-advisory-db.git",),
1123 ).decode("utf-8")
1124
1125 assert "git-fetch-with-cli" not in cfg
1126 assert "maximum-db-staleness" not in cfg
1127 assert "[advisories]" in cfg
1128 with pytest.raises(audit.ReleasePolicyError):
1129 audit.audit_config_bytes(
1130 b'[advisories]\ndb-path = "x"\n',
1131 db_root=tmp_path / "db-root",
1132 db_urls=("ssh://mirror.example.invalid/rustsec-advisory-db.git",),
1133 )
1134
1135
1136def test_main_prints_success_bytes_only_on_success(
1137 monkeypatch: pytest.MonkeyPatch,
1138 capsys: pytest.CaptureFixture[str],
1139) -> None:
1140 monkeypatch.setattr(
1141 audit, "audit_advisory_mirror", lambda *args, **kwargs: b'{"ok":1}\n'
1142 )
1143
1144 result = audit.main(
1145 [
1146 "--bundle",
1147 "bundle",
1148 "--receipt",
1149 "receipt",
1150 "--pubkey",
1151 "pubkey",
1152 "--locator",
1153 "ssh://mirror.example.invalid/rustsec-advisory-db.git",
1154 ]
1155 )
1156
1157 captured = capsys.readouterr()
1158 assert result == 0
1159 assert captured.out == '{"ok":1}\n'
1160 assert captured.err == ""
1161
1162
1163@pytest.mark.parametrize("empty_name", ["bundle", "receipt", "pubkey", "locator"])
1164def test_main_empty_inputs_fail_before_audit_orchestration(
1165 monkeypatch: pytest.MonkeyPatch,
1166 capsys: pytest.CaptureFixture[str],
1167 empty_name: str,
1168) -> None:
1169 calls = []
1170
1171 def unexpected(*args, **kwargs):
1172 calls.append((args, kwargs))
1173 raise AssertionError("audit should not run")
1174
1175 monkeypatch.setattr(audit, "audit_advisory_mirror", unexpected)
1176 values = {
1177 "bundle": "bundle",
1178 "receipt": "receipt",
1179 "pubkey": "pubkey",
1180 "locator": "ssh://mirror.example.invalid/rustsec-advisory-db.git",
1181 }
1182 values[empty_name] = ""
1183
1184 result = audit.main(
1185 [
1186 "--bundle",
1187 values["bundle"],
1188 "--receipt",
1189 values["receipt"],
1190 "--pubkey",
1191 values["pubkey"],
1192 "--locator",
1193 values["locator"],
1194 ]
1195 )
1196
1197 captured = capsys.readouterr()
1198 assert result == 1
1199 assert calls == []
1200 assert captured.out == ""
1201 assert f"input {empty_name} is empty" in captured.err
1202
1203
1204def test_main_prints_redacted_failures_only_on_error(
1205 monkeypatch: pytest.MonkeyPatch,
1206 capsys: pytest.CaptureFixture[str],
1207) -> None:
1208 def fail(*args, **kwargs):
1209 raise audit.ReleasePolicyError(
1210 [
1211 audit._failure_record(
1212 "failed",
1213 expected="public",
1214 actual="/private/path TOKEN=value",
1215 )
1216 ]
1217 )
1218
1219 monkeypatch.setattr(audit, "audit_advisory_mirror", fail)
1220 result = audit.main(
1221 [
1222 "--bundle",
1223 "bundle",
1224 "--receipt",
1225 "receipt",
1226 "--pubkey",
1227 "pubkey",
1228 "--locator",
1229 "ssh://mirror.example.invalid/rustsec-advisory-db.git",
1230 ]
1231 )
1232
1233 captured = capsys.readouterr()
1234 assert result == 1
1235 assert captured.out == ""
1236 assert "/private/path" not in captured.err
1237 assert "TOKEN=" not in captured.err