github-devloop implementation for #791

scripts/generate_testing_package_release.py

@@ -226,7 +226,7 @@ def signing_seed(seed_file: Path | None) -> bytes:
     if seed_file is not None and environment_value is not None:
         raise ValueError(f"use either --seed-file or {SEED_ENVIRONMENT_VARIABLE}, not both")
     if seed_file is not None:
-        value = seed_file.read_text(encoding="ascii").strip()
+        value = seed_file.read_text(encoding="ascii")
     elif environment_value is not None:
         value = environment_value
     else:

scripts/testing_package_release_test.py

@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 import base64
+import copy
 import hashlib
+import json
 import os
 import shutil
 import subprocess
@@ -106,6 +108,183 @@ def assert_bundle_uses_pinned_git_tree() -> None:
         assert bundled_bytes != substituted_bytes
 
 
+STAGES = ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified", "manifest-verified", "bundle-verified", "materialized", "executed")
+ARTIFACTS = {
+    "release": ROOT / "package-release/testing-package-release.v1.json",
+    "envelope": ROOT / "package-release/testing-package-release.v1.dsse.json",
+    "bundle": ROOT / "package-release/testing-package-bundle.v1.json",
+    "manifest": ROOT / "package-release/testing-package-manifest.v1.json",
+    "schema-catalog": ROOT / "schema-release/testing-schema-catalog.v1.json",
+    "schema-release": ROOT / "schema-release/testing-package-schema-release.v1.json",
+}
+
+def canonical(value: object, *, lf: bool = True) -> bytes:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + (b"\n" if lf else b"")
+
+def stages(path: Path) -> tuple[str, ...]:
+    return tuple(path.read_text().splitlines()) if path.exists() else ()
+
+def assert_rejection(result: subprocess.CompletedProcess[str], stage_log: Path, message: str, allowed: tuple[str, ...]) -> None:
+    assert message in result.stderr, result.stderr
+    observed = stages(stage_log)
+    assert observed == ("release-digest-matched", *allowed), (message, observed)
+    assert "materialized" not in observed and "executed" not in observed
+
+def signed_case(root: Path, mutate, *, target: str, message: str, allowed: tuple[str, ...]) -> None:
+    paths = {}
+    for name, source in ARTIFACTS.items():
+        target_path = root / source.name
+        target_path.write_bytes(source.read_bytes())
+        paths[name] = target_path
+    authorization = root / "authorization.json"
+    release = json.loads(paths["release"].read_bytes())
+    for binding, artifact in (("bundle", "bundle"), ("manifest", "manifest"), ("schema_catalog", "schema-catalog"), ("schema_release", "schema-release")):
+        release[binding]["path"] = paths[artifact].relative_to(ROOT).as_posix()
+    value = json.loads(paths[target].read_bytes())
+    mutate(value)
+    if target == "manifest":
+        value["manifest_digest"] = hashlib.sha256(canonical({key: item for key, item in value.items() if key != "manifest_digest"}, lf=False)).hexdigest()
+    paths[target].write_bytes(canonical(value, lf=target != "manifest"))
+    if target == "manifest":
+        release["manifest"].update(size_bytes=paths[target].stat().st_size, sha256=hashlib.sha256(paths[target].read_bytes()).hexdigest(), manifest_digest=value.get("manifest_digest", "0" * 64))
+    elif target == "bundle":
+        release["bundle"].update(size_bytes=paths[target].stat().st_size, sha256=hashlib.sha256(paths[target].read_bytes()).hexdigest())
+    if target != "release":
+        paths["release"].write_bytes(canonical(release))
+    seed = hashlib.sha256((target + message).encode()).digest()
+    envelope, authorization_bytes = generator.signed_artifacts(paths["release"].read_bytes(), seed)
+    paths["envelope"].write_bytes(envelope); authorization.write_bytes(authorization_bytes)
+    pin = hashlib.sha256(authorization_bytes).hexdigest(); log = root / "stages.log"
+    release_sha256 = hashlib.sha256(paths["release"].read_bytes()).hexdigest()
+    result = run_verifier(pin, expected_release_sha256=release_sha256,
+                          authorization=authorization, paths=paths, stage_log=log, success=False)
+    assert_rejection(result, log, message, allowed)
+
+def assert_rejection_matrix() -> None:
+    legitimate_pin = hashlib.sha256(AUTHORIZATION.read_bytes()).hexdigest()
+    expected = ["--expected-release-sha256", RELEASE_SHA256]
+    cli = [
+        ([*expected, "--release", str(ARTIFACTS["release"])], "--trusted-authorization-sha256 is required exactly once"),
+        ([*expected, "--trusted-authorization-sha256", legitimate_pin.upper()], "exactly 64 lowercase hexadecimal"),
+        ([*expected, "--trusted-authorization-sha256", legitimate_pin[:-1]], "exactly 64 lowercase hexadecimal"),
+        ([*expected, "--trusted-authorization-sha256", "g" * 64], "exactly 64 lowercase hexadecimal"),
+        ([*expected, "--trusted-authorization-sha256", legitimate_pin, "--trusted-authorization-sha256", legitimate_pin], "arguments must be unique"),
+        ([*expected, "--trusted-authorization-sha256", "--release", str(ARTIFACTS["release"])], "arguments must be unique"),
+        ([*expected, "--trusted-authorization-sha256", legitimate_pin, "--release", str(ARTIFACTS["release"]), "--release", str(ARTIFACTS["release"])], "arguments must be unique"),
+        ([*expected, "--trusted-authorization-sha256", legitimate_pin, "--unknown", "value"], "unknown argument"),
+    ]
+    for arguments, message in cli:
+        result = subprocess.run(["node", str(VERIFIER), *arguments], cwd=ROOT, text=True, capture_output=True)
+        assert result.returncode != 0 and message in result.stderr
+    with tempfile.TemporaryDirectory(prefix=".testing-package-release-matrix-", dir=ROOT) as directory:
+        root = Path(directory)
+        authorization_cases = [
+            (lambda value: value.update(unexpected=True), "authorization record fields"),
+            (lambda value: value.update(algorithm="rsa"), "authorization record profile is unsupported"),
+            (lambda value: value.update(schema="wrong"), "authorization record profile is unsupported"),
+            (lambda value: value.update(keyid="wrong"), "authorization record profile is unsupported"),
+            (lambda value: value["authorization"].update(subject="wrong"), "authorization record profile is unsupported"),
+            (lambda value: value["authorization"].update(unexpected=True), "authorization scope fields"),
+            (lambda value: value.update(publicKey=value["publicKey"].rstrip("=")), "canonical standard base64"),
+            (lambda value: value.update(publicKey="AA=="), "decode to exactly 32 bytes"),
+        ]
+        for index, (mutate, message) in enumerate(authorization_cases):
+            authorization = json.loads(AUTHORIZATION.read_bytes()); mutate(authorization)
+            auth_path = root / f"authorization-{index}.json"; auth_path.write_bytes(canonical(authorization)); log = root / f"authorization-{index}.log"
+            result = run_verifier(hashlib.sha256(auth_path.read_bytes()).hexdigest(), authorization=auth_path, stage_log=log, success=False)
+            assert_rejection(result, log, message, ("trust-pin-matched",))
+        noncanonical = root / "authorization-noncanonical.json"; noncanonical.write_bytes(AUTHORIZATION.read_bytes()[:-1])
+        log = root / "authorization-noncanonical.log"
+        result = run_verifier(hashlib.sha256(noncanonical.read_bytes()).hexdigest(), authorization=noncanonical, stage_log=log, success=False)
+        assert_rejection(result, log, "authorization record bytes are not canonical", ("trust-pin-matched",))
+        seed = hashlib.sha256(b"envelope-matrix").digest()
+        envelope_bytes, authorization_bytes = generator.signed_artifacts(ARTIFACTS["release"].read_bytes(), seed)
+        envelope_authorization = root / "envelope-authorization.json"; envelope_authorization.write_bytes(authorization_bytes)
+        envelope_pin = hashlib.sha256(authorization_bytes).hexdigest()
+        envelope_cases = [
+            (lambda value: value.update(unexpected=True), "DSSE envelope fields"),
+            (lambda value: value.update(payloadType="wrong"), "DSSE envelope profile is unsupported"),
+            (lambda value: value.update(signatures=[]), "DSSE envelope profile is unsupported"),
+            (lambda value: value.update(signatures=value["signatures"] * 2), "DSSE envelope profile is unsupported"),
+            (lambda value: value["signatures"][0].update(unexpected=True), "DSSE signature fields"),
+            (lambda value: value["signatures"][0].update(keyid="wrong"), "DSSE keyid is unsupported"),
+            (lambda value: value["signatures"][0].update(sig="AA=="), "decode to exactly 64 bytes"),
+            (lambda value: value["signatures"][0].update(sig=("A" if value["signatures"][0]["sig"][0] != "A" else "B") + value["signatures"][0]["sig"][1:]), "Ed25519 DSSE verification failed"),
+            (lambda value: value.update(payload=value["payload"] + " "), "canonical standard base64"),
+        ]
+        for index, (mutate, message) in enumerate(envelope_cases):
+            envelope = json.loads(envelope_bytes); mutate(envelope)
+            envelope_path = root / f"envelope-{index}.json"; envelope_path.write_bytes(canonical(envelope)); log = root / f"envelope-{index}.log"
+            result = run_verifier(envelope_pin, authorization=envelope_authorization, paths={"envelope": envelope_path}, stage_log=log, success=False)
+            assert_rejection(result, log, message, ("trust-pin-matched", "public-key-imported"))
+        envelope_path = root / "envelope-noncanonical.json"; envelope_path.write_bytes(envelope_bytes[:-1]); log = root / "envelope-noncanonical.log"
+        result = run_verifier(envelope_pin, authorization=envelope_authorization, paths={"envelope": envelope_path}, stage_log=log, success=False)
+        assert_rejection(result, log, "DSSE envelope bytes are not canonical", ("trust-pin-matched", "public-key-imported"))
+        cases = [
+            ("release", lambda value: value.update(schema="wrong"), "release profile is unsupported", ("trust-pin-matched", "public-key-imported", "dsse-verified")),
+            ("release", lambda value: value["mappings"].append(copy.deepcopy(value["mappings"][0])), "exactly one mapping", ("trust-pin-matched", "public-key-imported", "dsse-verified")),
+            ("manifest", lambda value: value["runtime_requirements"].update(lua="5.3.0"), "manifest runtime requirements are unsupported", ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified")),
+            ("manifest", lambda value: value["entrypoints"].append(copy.deepcopy(value["entrypoints"][0])), "manifest must expose exactly testing-runner.run", ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified")),
+            ("bundle", lambda value: value["files"].append({**value["files"][-1], "path": "libraries/unexpected.lua"}), "bundle files do not match the release allowlist", ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified", "manifest-verified")),
+            ("bundle", lambda value: value["files"].reverse(), "bundle paths must be unique and sorted", ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified", "manifest-verified")),
+            ("bundle", lambda value: value["files"][0].update(path="../escape.lua"), "bundle file path is unsafe", ("trust-pin-matched", "public-key-imported", "dsse-verified", "release-verified", "manifest-verified")),
+        ]
+        for index, case in enumerate(cases):
+            case_root = root / str(index); case_root.mkdir(); signed_case(case_root, case[1], target=case[0], message=case[2], allowed=case[3])
+        publication = root / "publication"; publication.mkdir()
+        paths = {name: publication / source.name for name, source in ARTIFACTS.items()}
+        for name, source in ARTIFACTS.items(): paths[name].write_bytes(source.read_bytes())
+        release = json.loads(paths["release"].read_bytes())
+        for binding, artifact in (("bundle", "bundle"), ("manifest", "manifest"), ("schema_catalog", "schema-catalog"), ("schema_release", "schema-release")):
+            release[binding]["path"] = paths[artifact].relative_to(ROOT).as_posix()
+        paths["release"].write_bytes(canonical(release))
+        envelope, authorization_bytes = generator.signed_artifacts(paths["release"].read_bytes(), hashlib.sha256(b"publication").digest())
+        paths["envelope"].write_bytes(envelope); auth_path = publication / "authorization.json"; auth_path.write_bytes(authorization_bytes)
+        paths["schema-catalog"].write_bytes(paths["schema-catalog"].read_bytes()[:-1] + b" ")
+        log = root / "publication.log"
+        release_sha256 = hashlib.sha256(paths["release"].read_bytes()).hexdigest()
+        result = run_verifier(hashlib.sha256(authorization_bytes).hexdigest(),
+                              expected_release_sha256=release_sha256,
+                              authorization=auth_path, paths=paths, stage_log=log, success=False)
+        assert_rejection(result, log, "schema publication binding mismatch", ("trust-pin-matched", "public-key-imported", "dsse-verified"))
+
+
+def assert_generator_rejections() -> None:
+    variable = generator.SEED_ENVIRONMENT_VARIABLE
+    original = os.environ.get(variable)
+    seed = base64.b64encode(hashlib.sha256(b"generator-matrix").digest()).decode()
+    try:
+        os.environ.pop(variable, None)
+        try:
+            generator.signing_seed(None)
+        except ValueError as error:
+            assert "signing seed is required" in str(error)
+        else:
+            raise AssertionError("missing signing seed was accepted")
+        with tempfile.TemporaryDirectory(prefix="testing-package-release-seed-") as directory:
+            seed_path = Path(directory) / "seed"; seed_path.write_text(seed)
+            os.environ[variable] = seed
+            try:
+                generator.signing_seed(seed_path)
+            except ValueError as error:
+                assert "use either --seed-file" in str(error)
+            else:
+                raise AssertionError("duplicate signing seed sources were accepted")
+            os.environ.pop(variable)
+            seed_path.write_text(seed + "\n")
+            try:
+                generator.signing_seed(seed_path)
+            except ValueError as error:
+                assert seed not in str(error) and "canonical standard base64" in str(error)
+            else:
+                raise AssertionError("noncanonical signing seed was accepted")
+    finally:
+        if original is None:
+            os.environ.pop(variable, None)
+        else:
+            os.environ[variable] = original
+
+
 def main() -> int:
     assert hashlib.sha256(RELEASE.read_bytes()).hexdigest() == RELEASE_SHA256
     assert hashlib.sha256(AUTHORIZATION.read_bytes()).hexdigest() == AUTHORIZATION_SHA256
@@ -117,6 +296,7 @@ def main() -> int:
         SOURCE_COMMIT,
     ], cwd=ROOT, check=True)
     assert_bundle_uses_pinned_git_tree()
+    assert_generator_rejections()
     schema_paths = tuple(sorted((ROOT / "schemas").glob("*.schema.json")))
     registry = offline_registry(schema_paths)
     _, validator = validator_for_schema_file(
@@ -134,6 +314,8 @@ def main() -> int:
         assert "testing-package-release: VERIFIED AND EXECUTED" in result.stdout
         assert_stages(stage_log, SUCCESS_STAGES)
 
+    assert_rejection_matrix()
+
     with tempfile.TemporaryDirectory(prefix="testing-package-release-policy-") as directory:
         root = Path(directory)
         stage_log = root / "stages.log"