diff --git a/.gitignore b/.gitignore index 2024295..1a24f55 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,8 @@ .pacta/ .pacta-*/ .pytest_cache/ +.pytest-tmp/ +.tmp/ artifacts*/ repos/ provider/out/ diff --git a/AGENTS.md b/AGENTS.md index 9697853..9056188 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,3 +14,7 @@ Guidance for future Codex runs in this repository: - Consequence-producing commands must be policy gated. Do not build wallet or trading-agent artifacts from `R3` arithmetic evidence. - Distinguish verifier capability failures from proof failures. Missing `Mathlib`, missing `Aeneas`, or a missing pinned env script means local replay is unavailable; do not treat it as a clean proof. - Third-party proof-checking attestations are allowed only as an explicit trust transformation. They must identify the provider, subject repo/commit, theorem names, observed axioms, and signature status. Untrusted attestations must not authorize builds. +- Transparency receipts must be verified against the exact attestation bytes, Signed Tree Head, log public key, and inclusion proof before they can authorize consequences. +- Keep the Merkle log RFC 9162-style unless a new standard is deliberately adopted and documented. Do not replace it with an ad hoc hash chain. +- Do not pretend ML-DSA exists. If no real ML-DSA backend is available, record the signature slot as unavailable and fail closed for policies that require both Ed25519 and ML-DSA. +- Provider private keys and transparency log state belong under ignored `provider/state/` or `provider/out/` paths. Do not commit local trust state. diff --git a/README.md b/README.md index bac31db..6017531 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ pacta audit --repo ./repos/dalek-ed25519-verified pacta lean-check --repo ./repos/dalek-ed25519-verified pacta report --claims claims.yaml --out report.md pacta score --claims claims.yaml +pacta receipt-verify --attestation provider/out/dalek-ed25519.attestation.yaml --receipt provider/out/dalek-ed25519.receipt.yaml --log-public-key provider/state/local-provider/provider.ed25519.pub pacta agent --config examples/repos.yaml --repo-name dalek-ed25519-verified --offline-fixture --action build-library pacta agent --config examples/repos.yaml --repo-name dalek-ed25519-verified --clone --run-axioms --action build-library --artifact-dir artifacts-live pacta agent --claims claims.yaml --action build-wallet-demo @@ -97,6 +98,66 @@ This changes the trusted base. The agent is no longer trusting local Lean replay The included `examples/dalek-ed25519.attestation.yaml` is an unsigned schema/demo fixture and requires `--allow-unsigned-attestation`. Real provider certificates should be signed and consumed with `--attestation-public-key`. +## Transparency-Logged Attestations + +Standalone signatures prove who signed an attestation, but they do not make the provider accountable for equivocation or silent replacement. The nested provider can also append attestations to a local RFC 9162-style Merkle transparency log and issue inclusion receipts. + +The log uses: + +- `RFC9162_SHA256` Merkle leaf/node hashing with `0x00` leaf and `0x01` node domain separation. +- Signed Tree Heads over canonical JSON tree-head payloads. +- OpenSSL Ed25519 signatures today. +- An explicit `ML-DSA-65` / FIPS 204 signature slot that is `unavailable` unless the host has a real backend. If an agent policy requires both signatures, verification fails closed. + +Example: + +```bash +PYTHONPATH=src:provider/src python -m pacta_provider log-init \ + --log-dir provider/state/transparency-log \ + --provider local-pacta-provider \ + --public-key provider/state/local-provider/provider.ed25519.pub + +PYTHONPATH=src:provider/src python -m pacta_provider log-append \ + --log-dir provider/state/transparency-log \ + --attestation provider/out/dalek-ed25519.attestation.yaml \ + --private-key provider/state/local-provider/provider.ed25519.key \ + --public-key provider/state/local-provider/provider.ed25519.pub \ + --out provider/out/dalek-ed25519.receipt.yaml + +pacta receipt-verify \ + --attestation provider/out/dalek-ed25519.attestation.yaml \ + --receipt provider/out/dalek-ed25519.receipt.yaml \ + --log-public-key provider/state/local-provider/provider.ed25519.pub +``` + +Agents can require the receipt before building anything: + +```bash +pacta agent \ + --config examples/repos.yaml \ + --repo-name dalek-ed25519-verified \ + --repo repos/dalek-ed25519-verified \ + --attestation provider/out/dalek-ed25519.attestation.yaml \ + --trust-attestation-provider local-pacta-provider \ + --attestation-public-key provider/state/local-provider/provider.ed25519.pub \ + --transparency-receipt provider/out/dalek-ed25519.receipt.yaml \ + --transparency-log-public-key provider/state/local-provider/provider.ed25519.pub \ + --require-transparency-receipt \ + --action build-library +``` + +To demand post-quantum log signatures as well: + +```bash +pacta receipt-verify \ + --attestation provider/out/dalek-ed25519.attestation.yaml \ + --receipt provider/out/dalek-ed25519.receipt.yaml \ + --log-public-key provider/state/local-provider/provider.ed25519.pub \ + --require-signatures both +``` + +On a host without ML-DSA support, that command should fail. That is intentional. The system records the missing capability as a deployment blocker instead of treating the Ed25519 signature as quantum-robust. + ## Nested Proof-Check Provider This repository includes a nested provider prototype under `provider/`. It searches read-only under your home/GitClone tree for reusable Lean/Aeneas infrastructure, runs the proof replay, signs the result with OpenSSL Ed25519, and emits an attestation. @@ -121,6 +182,9 @@ pacta agent \ --attestation provider/out/dalek-ed25519.attestation.yaml \ --trust-attestation-provider local-pacta-provider \ --attestation-public-key provider/state/local-provider/provider.ed25519.pub \ + --transparency-receipt provider/out/dalek-ed25519.receipt.yaml \ + --transparency-log-public-key provider/state/local-provider/provider.ed25519.pub \ + --require-transparency-receipt \ --action build-library ``` diff --git a/provider/README.md b/provider/README.md index ccb6fe2..ac9e360 100644 --- a/provider/README.md +++ b/provider/README.md @@ -4,6 +4,8 @@ This nested project is a prototype third-party proof-checking service. It reuses It does not modify anything outside this repository. It may read configured toolchains such as `/Users/oho/GitClone/ClaudeCodeProjects/your-lean-project/aeneas-toolchain/env.sh`. +It can also maintain a local transparency log. The log is an RFC 9162-style Merkle accumulator over signed attestations. It emits Signed Tree Heads with Ed25519 today and records an ML-DSA/FIPS 204 signature slot as `unavailable` unless a real backend is present. Agents that require both signatures must reject such receipts. + ## Commands ```bash @@ -19,6 +21,29 @@ PYTHONPATH=src:provider/src python -m pacta_provider check \ --out provider/out/dalek.attestation.yaml ``` +Transparency log: + +```bash +PYTHONPATH=src:provider/src python -m pacta_provider log-init \ + --log-dir provider/state/transparency-log \ + --provider local-pacta-provider \ + --public-key provider/state/demo-provider/provider.ed25519.pub + +PYTHONPATH=src:provider/src python -m pacta_provider log-append \ + --log-dir provider/state/transparency-log \ + --attestation provider/out/dalek.attestation.yaml \ + --private-key provider/state/demo-provider/provider.ed25519.key \ + --public-key provider/state/demo-provider/provider.ed25519.pub \ + --out provider/out/dalek.receipt.yaml + +PYTHONPATH=src:provider/src python -m pacta_provider log-sth \ + --log-dir provider/state/transparency-log \ + --private-key provider/state/demo-provider/provider.ed25519.key \ + --public-key provider/state/demo-provider/provider.ed25519.pub +``` + The resulting certificate can be consumed by `pacta` with `--attestation`, `--trust-attestation-provider`, and `--attestation-public-key`. -The private key must remain provider-side. Downstream agents only need the public key and a policy decision that the provider name is trusted. +The receipt can be consumed with `--transparency-receipt`, `--transparency-log-public-key`, and `--require-transparency-receipt`. + +The private key must remain provider-side. Downstream agents only need the public key, the inclusion receipt, and a policy decision that the provider name/log key is trusted. diff --git a/provider/src/pacta_provider/cli.py b/provider/src/pacta_provider/cli.py index 8f8215a..2a392e1 100644 --- a/provider/src/pacta_provider/cli.py +++ b/provider/src/pacta_provider/cli.py @@ -10,6 +10,7 @@ from pacta.yamlio import dump_data from .discovery import discover_toolchains from .service import build_attestation +from .transparency_log import TransparencyLog def main(argv: list[str] | None = None) -> int: @@ -48,6 +49,26 @@ def build_parser() -> argparse.ArgumentParser: check.add_argument("--log-dir", default="provider/out/logs") check.add_argument("--out", required=True) check.set_defaults(func=cmd_check) + + log_init = sub.add_parser("log-init", help="Initialize a local RFC9162-style transparency log.") + log_init.add_argument("--log-dir", default="provider/state/transparency-log") + log_init.add_argument("--provider", required=True) + log_init.add_argument("--public-key", required=True) + log_init.set_defaults(func=cmd_log_init) + + log_append = sub.add_parser("log-append", help="Append a signed proof-check attestation and emit an inclusion receipt.") + log_append.add_argument("--log-dir", default="provider/state/transparency-log") + log_append.add_argument("--attestation", required=True) + log_append.add_argument("--private-key", required=True) + log_append.add_argument("--public-key", required=True) + log_append.add_argument("--out", required=True) + log_append.set_defaults(func=cmd_log_append) + + log_sth = sub.add_parser("log-sth", help="Sign and print the latest transparency-log tree head.") + log_sth.add_argument("--log-dir", default="provider/state/transparency-log") + log_sth.add_argument("--private-key", required=True) + log_sth.add_argument("--public-key", required=True) + log_sth.set_defaults(func=cmd_log_sth) return parser @@ -86,3 +107,31 @@ def cmd_check(args: argparse.Namespace) -> int: dump_data(attestation, args.out) print(f"attestation: {args.out}") return 0 + + +def cmd_log_init(args: argparse.Namespace) -> int: + metadata = TransparencyLog(args.log_dir).init(args.provider, args.public_key) + print(json.dumps(metadata, indent=2, sort_keys=True)) + return 0 + + +def cmd_log_append(args: argparse.Namespace) -> int: + receipt = TransparencyLog(args.log_dir).append_attestation( + args.attestation, + private_key_path=args.private_key, + public_key_path=args.public_key, + receipt_out=args.out, + ) + print(f"receipt: {args.out}") + print(f"log_id: {receipt['log_id']}") + print(f"tree_size: {receipt['tree_size']}") + print(f"leaf_hash: {receipt['leaf_hash']}") + print(f"ed25519_sth_signature: {receipt['sth']['signatures']['ed25519']['status']}") + print(f"ml_dsa_sth_signature: {receipt['sth']['signatures']['ml_dsa']['status']}") + return 0 + + +def cmd_log_sth(args: argparse.Namespace) -> int: + sth = TransparencyLog(args.log_dir).latest_sth(args.private_key, args.public_key) + print(json.dumps(sth, indent=2, sort_keys=True)) + return 0 diff --git a/provider/src/pacta_provider/transparency_log.py b/provider/src/pacta_provider/transparency_log.py new file mode 100644 index 0000000..da7d66e --- /dev/null +++ b/provider/src/pacta_provider/transparency_log.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from pacta.signing import canonical_json +from pacta.transparency import ( + HASH_ALGORITHM, + RECEIPT_TYPE, + attestation_leaf, + consistency_proof, + inclusion_proof, + leaf_bytes_for_attestation, + leaf_hash, + make_signed_tree_head, + merkle_root, + proof_to_hex, +) +from pacta.yamlio import dump_data, load_data + + +@dataclass(slots=True) +class LogEntry: + index: int + leaf: dict[str, Any] + leaf_hash: str + + def leaf_bytes(self) -> bytes: + return canonical_json(self.leaf) + + +class TransparencyLog: + def __init__(self, log_dir: str | Path) -> None: + self.log_dir = Path(log_dir) + self.metadata_path = self.log_dir / "metadata.json" + self.entries_path = self.log_dir / "entries.jsonl" + self.sth_path = self.log_dir / "sth.yaml" + + def init(self, provider: str, public_key_path: str | Path) -> dict[str, Any]: + if self.metadata_path.exists() or self.entries_path.exists(): + raise ValueError(f"Transparency log already exists: {self.log_dir}") + self.log_dir.mkdir(parents=True, exist_ok=True) + public_key = Path(public_key_path).read_bytes() + log_id = hashlib.sha256(b"pacta-log-v1\0" + provider.encode("utf-8") + b"\0" + public_key).hexdigest() + metadata = { + "schema_version": 1, + "type": "pacta.transparency.log_metadata.v1", + "provider": provider, + "log_id": log_id, + "hash_algorithm": HASH_ALGORITHM, + "ed25519_public_key_fingerprint_sha256": hashlib.sha256(public_key).hexdigest(), + "created_at": _now(), + "standards": [ + "RFC 9162 Merkle tree hash and inclusion/consistency proof algorithms", + "RFC 8032 Ed25519 signature verification via OpenSSL", + "FIPS 204 ML-DSA signature slot; must be required by policy only when a backend is configured", + ], + } + self.metadata_path.write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8") + self.entries_path.write_text("", encoding="utf-8") + return metadata + + def metadata(self) -> dict[str, Any]: + if not self.metadata_path.exists(): + raise ValueError(f"Transparency log is not initialized: {self.log_dir}") + raw = json.loads(self.metadata_path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError(f"Invalid transparency log metadata: {self.metadata_path}") + return raw + + def entries(self) -> list[LogEntry]: + if not self.entries_path.exists(): + return [] + entries: list[LogEntry] = [] + for line_number, line in enumerate(self.entries_path.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + raw = json.loads(line) + if not isinstance(raw, dict): + raise ValueError(f"Invalid log entry at line {line_number}: {self.entries_path}") + leaf = raw.get("leaf") + if not isinstance(leaf, dict): + raise ValueError(f"Invalid leaf at line {line_number}: {self.entries_path}") + entries.append(LogEntry(index=int(raw["index"]), leaf=leaf, leaf_hash=str(raw["leaf_hash"]))) + return entries + + def latest_sth( + self, + private_key_path: str | Path, + public_key_path: str | Path, + timestamp: str | None = None, + ) -> dict[str, Any]: + metadata = self.metadata() + leaves = [entry.leaf_bytes() for entry in self.entries()] + sth = make_signed_tree_head( + metadata["log_id"], + len(leaves), + merkle_root(leaves).hex(), + timestamp or _now(), + private_key_path, + public_key_path, + ) + dump_data(sth, self.sth_path) + return sth + + def append_attestation( + self, + attestation_path: str | Path, + private_key_path: str | Path, + public_key_path: str | Path, + receipt_out: str | Path | None = None, + ) -> dict[str, Any]: + metadata = self.metadata() + attestation = load_data(attestation_path) + if not isinstance(attestation, dict): + raise ValueError(f"Attestation must be a mapping: {attestation_path}") + entries = self.entries() + previous_size = len(entries) + previous_root = merkle_root([entry.leaf_bytes() for entry in entries]).hex() + leaf = attestation_leaf(attestation) + leaf_bytes = leaf_bytes_for_attestation(attestation) + computed_leaf_hash = leaf_hash(leaf_bytes).hex() + existing = next((entry for entry in entries if entry.leaf_hash == computed_leaf_hash), None) + if existing: + index = existing.index + appended = False + else: + index = len(entries) + record = {"index": index, "leaf_hash": computed_leaf_hash, "leaf": leaf} + with self.entries_path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n") + entries.append(LogEntry(index=index, leaf=leaf, leaf_hash=computed_leaf_hash)) + appended = True + + leaves = [entry.leaf_bytes() for entry in entries] + sth = make_signed_tree_head( + metadata["log_id"], + len(leaves), + merkle_root(leaves).hex(), + _now(), + private_key_path, + public_key_path, + ) + dump_data(sth, self.sth_path) + consistency = [] + if appended and previous_size > 0: + consistency = proof_to_hex(consistency_proof(leaves, previous_size)) + receipt = { + "schema_version": 1, + "type": RECEIPT_TYPE, + "log_id": metadata["log_id"], + "hash_algorithm": HASH_ALGORITHM, + "attestation_path": str(attestation_path), + "attestation_digest_sha256": hashlib.sha256(canonical_json(attestation)).hexdigest(), + "leaf_index": index, + "leaf_hash": computed_leaf_hash, + "tree_size": len(leaves), + "inclusion_proof": proof_to_hex(inclusion_proof(leaves, index)), + "consistency": { + "from_tree_size": previous_size, + "from_root_hash": previous_root, + "proof": consistency, + "status": "not_applicable" if previous_size == 0 or not appended else "included", + }, + "sth": sth, + } + if receipt_out: + Path(receipt_out).parent.mkdir(parents=True, exist_ok=True) + dump_data(receipt, receipt_out) + return receipt + + +def _now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") diff --git a/src/pacta/artifact.py b/src/pacta/artifact.py index e317da6..8b35376 100644 --- a/src/pacta/artifact.py +++ b/src/pacta/artifact.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os import re import shutil import subprocess @@ -68,6 +69,9 @@ def build_proof_gated_capsule( ) cmd = [cargo, "build", "--release", "--manifest-path", str(crate_dir / "Cargo.toml")] + tmpdir = os.environ.get("TMPDIR") + if tmpdir: + Path(tmpdir).mkdir(parents=True, exist_ok=True) try: completed = subprocess.run( cmd, diff --git a/src/pacta/attestation.py b/src/pacta/attestation.py index 8e9e2e1..21408e0 100644 --- a/src/pacta/attestation.py +++ b/src/pacta/attestation.py @@ -6,6 +6,7 @@ from typing import Any from .config import RepoConfig from .signing import verify_attestation_signature +from .transparency import load_receipt, verify_receipt from .yamlio import load_data @@ -34,6 +35,10 @@ def validate_attestation( trusted_provider: str | None = None, public_key_path: str | Path | None = None, allow_unsigned: bool = False, + transparency_receipt_path: str | Path | None = None, + transparency_log_public_key_path: str | Path | None = None, + require_transparency_signatures: str = "ed25519", + require_transparency_receipt: bool = False, ) -> AttestationResult: provider = raw.get("provider") subject = raw.get("subject") or {} @@ -80,6 +85,25 @@ def validate_attestation( elif signature_status != "verified": diagnostics.append(f"Attestation signature status is not acceptable: {signature_status}") + transparency_evidence: dict[str, Any] = {} + if require_transparency_receipt and not transparency_receipt_path: + diagnostics.append("Transparency receipt is required by policy but was not supplied.") + if transparency_receipt_path: + if not transparency_log_public_key_path: + diagnostics.append("Transparency receipt verification requires --transparency-log-public-key.") + else: + receipt = load_receipt(transparency_receipt_path) + receipt_result = verify_receipt( + raw, + receipt, + transparency_log_public_key_path, + require_signatures=require_transparency_signatures, + ) + transparency_evidence = receipt_result.evidence() + transparency_evidence["transparency_receipt_path"] = str(transparency_receipt_path) + if not receipt_result.accepted: + diagnostics.extend(receipt_result.diagnostics) + accepted = not diagnostics evidence = { "evidence_mode": "third_party_attestation", @@ -92,11 +116,14 @@ def validate_attestation( "axiom_log_path": (raw.get("replay") or {}).get("axiom_log_path"), "lean_version": environment.get("lean_version"), "lake_version": environment.get("lake_version"), + **transparency_evidence, } trusted_base = [] if accepted: trusted_base.append(f"Third-party proof-checking attestation provider: {provider}.") trusted_base.append("Provider environment, replay implementation, signing key custody, and log retention.") + if transparency_receipt_path: + trusted_base.append("Transparency log append-only behavior, signed tree head key custody, and monitor/auditor availability.") return AttestationResult( accepted=accepted, provider=str(provider) if provider else None, diff --git a/src/pacta/cli.py b/src/pacta/cli.py index a937706..2d06c90 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -24,6 +24,7 @@ from .profiles import get_profile from .repo import clone_or_fetch, status_for from .report import render_markdown from .risk import score_claim_card +from .transparency import load_receipt, verify_receipt from .yamlio import dump_data, load_data @@ -106,6 +107,10 @@ def build_parser() -> argparse.ArgumentParser: claims.add_argument("--trust-attestation-provider") claims.add_argument("--attestation-public-key") claims.add_argument("--allow-unsigned-attestation", action="store_true") + claims.add_argument("--transparency-receipt") + claims.add_argument("--transparency-log-public-key") + claims.add_argument("--require-transparency-signatures", choices=["ed25519", "both"], default="ed25519") + claims.add_argument("--require-transparency-receipt", action="store_true") claims.set_defaults(func=cmd_claims) report = sub.add_parser("report", help="Generate a human-readable Markdown risk report.") @@ -122,6 +127,13 @@ def build_parser() -> argparse.ArgumentParser: score.add_argument("--claims", required=True) score.set_defaults(func=cmd_score) + receipt_verify = sub.add_parser("receipt-verify", help="Verify a transparency-log inclusion receipt for an attestation.") + receipt_verify.add_argument("--attestation", required=True) + receipt_verify.add_argument("--receipt", required=True) + receipt_verify.add_argument("--log-public-key", required=True) + receipt_verify.add_argument("--require-signatures", choices=["ed25519", "both"], default="ed25519") + receipt_verify.set_defaults(func=cmd_receipt_verify) + agent = sub.add_parser("agent", help="Apply a policy-gated consequence to verification evidence.") agent.add_argument("--claims", help="Existing claim card to act on.") agent.add_argument("--config", help="Repository config used to generate a claim card.") @@ -143,6 +155,10 @@ def build_parser() -> argparse.ArgumentParser: agent.add_argument("--trust-attestation-provider") agent.add_argument("--attestation-public-key") agent.add_argument("--allow-unsigned-attestation", action="store_true") + agent.add_argument("--transparency-receipt") + agent.add_argument("--transparency-log-public-key") + agent.add_argument("--require-transparency-signatures", choices=["ed25519", "both"], default="ed25519") + agent.add_argument("--require-transparency-receipt", action="store_true") agent.set_defaults(func=cmd_agent) return parser @@ -347,6 +363,24 @@ def cmd_score(args: argparse.Namespace) -> int: return 0 +def cmd_receipt_verify(args: argparse.Namespace) -> int: + attestation = load_attestation(args.attestation) + receipt = load_receipt(args.receipt) + result = verify_receipt(attestation, receipt, args.log_public_key, require_signatures=args.require_signatures) + print(f"accepted: {str(result.accepted).lower()}") + print(f"log_id: {result.log_id or 'unknown'}") + print(f"tree_size: {result.tree_size if result.tree_size is not None else 'unknown'}") + print(f"leaf_hash: {result.leaf_hash or 'unknown'}") + print("signatures:") + for name, status in sorted(result.signatures.items()): + print(f" {name}: {status}") + if result.diagnostics: + print("diagnostics:") + for diagnostic in result.diagnostics: + print(f" - {diagnostic}") + return 0 if result.accepted else 1 + + def cmd_agent(args: argparse.Namespace) -> int: card = _card_for_agent(args) decision = run_agent_action( @@ -456,4 +490,8 @@ def _attestation_for_args(args: argparse.Namespace, repo: RepoConfig): trusted_provider=getattr(args, "trust_attestation_provider", None), public_key_path=getattr(args, "attestation_public_key", None), allow_unsigned=bool(getattr(args, "allow_unsigned_attestation", False)), + transparency_receipt_path=getattr(args, "transparency_receipt", None), + transparency_log_public_key_path=getattr(args, "transparency_log_public_key", None), + require_transparency_signatures=str(getattr(args, "require_transparency_signatures", "ed25519")), + require_transparency_receipt=bool(getattr(args, "require_transparency_receipt", False)), ) diff --git a/src/pacta/postquantum.py b/src/pacta/postquantum.py new file mode 100644 index 0000000..cd9797b --- /dev/null +++ b/src/pacta/postquantum.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import importlib.util +import re +import shutil +import subprocess +from dataclasses import dataclass + + +ML_DSA_SCHEME = "ML-DSA-65" +ML_DSA_STANDARD = "FIPS 204" + + +@dataclass(slots=True) +class MlDsaCapability: + available: bool + backend: str | None + reason: str + + def to_signature_slot(self) -> dict[str, str]: + if self.available: + return { + "scheme": ML_DSA_SCHEME, + "standard": ML_DSA_STANDARD, + "status": "not_configured", + "reason": "A backend appears available, but no ML-DSA signing key was configured for this log.", + } + return { + "scheme": ML_DSA_SCHEME, + "standard": ML_DSA_STANDARD, + "status": "unavailable", + "reason": self.reason, + } + + +def detect_ml_dsa() -> MlDsaCapability: + openssl = shutil.which("openssl") + if openssl: + try: + completed = subprocess.run( + [openssl, "list", "-signature-algorithms"], + check=False, + capture_output=True, + text=True, + timeout=15, + ) + except (OSError, subprocess.TimeoutExpired): + completed = None + if completed and completed.returncode == 0: + output = completed.stdout + completed.stderr + if re.search(r"\b(?:ML-?DSA|mldsa|dilithium)\b", output, flags=re.IGNORECASE): + return MlDsaCapability(True, "openssl", "OpenSSL advertises an ML-DSA/Dilithium signature algorithm.") + + for module in ("oqs", "pqcrypto", "dilithium"): + if importlib.util.find_spec(module): + return MlDsaCapability(True, f"python:{module}", f"Python module {module!r} appears importable.") + + return MlDsaCapability( + False, + None, + "No usable ML-DSA/FIPS 204 backend was found. Ed25519 log signatures can be verified, but a policy requiring both signatures must fail closed.", + ) diff --git a/src/pacta/signing.py b/src/pacta/signing.py index 5c6efea..eacea8d 100644 --- a/src/pacta/signing.py +++ b/src/pacta/signing.py @@ -16,7 +16,11 @@ class SignatureError(RuntimeError): def canonical_attestation_payload(attestation: dict[str, Any]) -> bytes: unsigned = {key: value for key, value in attestation.items() if key != "signature"} - return json.dumps(unsigned, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return canonical_json(unsigned) + + +def canonical_json(document: Any) -> bytes: + return json.dumps(document, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def payload_digest(attestation: dict[str, Any]) -> str: @@ -39,28 +43,14 @@ def generate_ed25519_keypair(private_key_path: str | Path, public_key_path: str def sign_attestation(attestation: dict[str, Any], private_key_path: str | Path, public_key_path: str | Path | None = None) -> dict[str, Any]: - openssl = _openssl() payload = canonical_attestation_payload(attestation) - with tempfile.TemporaryDirectory(prefix="pacta-sign-") as tmp: - payload_path = Path(tmp) / "payload.json" - signature_path = Path(tmp) / "payload.sig" - payload_path.write_bytes(payload) - completed = subprocess.run( - [openssl, "pkeyutl", "-sign", "-inkey", str(private_key_path), "-rawin", "-in", str(payload_path), "-out", str(signature_path)], - check=False, - capture_output=True, - text=True, - timeout=30, - ) - if completed.returncode != 0: - raise SignatureError((completed.stderr or completed.stdout or "openssl signing failed").strip()) - signature_bytes = signature_path.read_bytes() + signature_base64 = sign_payload_ed25519(payload, private_key_path) signed = dict(attestation) signed["signature"] = { "scheme": "openssl-ed25519", "status": "signed", "payload_digest_sha256": hashlib.sha256(payload).hexdigest(), - "signature_base64": base64.b64encode(signature_bytes).decode("ascii"), + "signature_base64": signature_base64, } if public_key_path: signed["signature"]["public_key_fingerprint_sha256"] = public_key_fingerprint(public_key_path) @@ -68,7 +58,6 @@ def sign_attestation(attestation: dict[str, Any], private_key_path: str | Path, def verify_attestation_signature(attestation: dict[str, Any], public_key_path: str | Path) -> tuple[bool, str | None]: - openssl = _openssl() signature = attestation.get("signature") or {} if signature.get("scheme") != "openssl-ed25519": return False, f"Unsupported attestation signature scheme: {signature.get('scheme')}" @@ -84,14 +73,38 @@ def verify_attestation_signature(attestation: dict[str, Any], public_key_path: s actual_fingerprint = public_key_fingerprint(public_key_path) if expected_fingerprint != actual_fingerprint: return False, "Attestation public key fingerprint does not match signature metadata." + return verify_payload_ed25519(canonical_attestation_payload(attestation), encoded, public_key_path) + + +def sign_payload_ed25519(payload: bytes, private_key_path: str | Path) -> str: + openssl = _openssl() + with tempfile.TemporaryDirectory(prefix="pacta-sign-") as tmp: + payload_path = Path(tmp) / "payload.bin" + signature_path = Path(tmp) / "payload.sig" + payload_path.write_bytes(payload) + completed = subprocess.run( + [openssl, "pkeyutl", "-sign", "-inkey", str(private_key_path), "-rawin", "-in", str(payload_path), "-out", str(signature_path)], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + if completed.returncode != 0: + raise SignatureError((completed.stderr or completed.stdout or "openssl signing failed").strip()) + signature_bytes = signature_path.read_bytes() + return base64.b64encode(signature_bytes).decode("ascii") + + +def verify_payload_ed25519(payload: bytes, signature_base64: str, public_key_path: str | Path) -> tuple[bool, str | None]: + openssl = _openssl() try: - signature_bytes = base64.b64decode(encoded, validate=True) + signature_bytes = base64.b64decode(signature_base64, validate=True) except ValueError as exc: return False, f"Invalid base64 signature: {exc}" with tempfile.TemporaryDirectory(prefix="pacta-verify-") as tmp: - payload_path = Path(tmp) / "payload.json" + payload_path = Path(tmp) / "payload.bin" signature_path = Path(tmp) / "payload.sig" - payload_path.write_bytes(canonical_attestation_payload(attestation)) + payload_path.write_bytes(payload) signature_path.write_bytes(signature_bytes) completed = subprocess.run( [ diff --git a/src/pacta/transparency.py b/src/pacta/transparency.py new file mode 100644 index 0000000..219607b --- /dev/null +++ b/src/pacta/transparency.py @@ -0,0 +1,340 @@ +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .postquantum import detect_ml_dsa +from .signing import canonical_json, public_key_fingerprint, sign_payload_ed25519, verify_payload_ed25519 +from .yamlio import load_data + +HASH_ALGORITHM = "RFC9162_SHA256" +LEAF_TYPE = "pacta.transparency.attestation_leaf.v1" +RECEIPT_TYPE = "pacta.transparency.receipt.v1" +STH_TYPE = "pacta.transparency.signed_tree_head.v1" + + +@dataclass(slots=True) +class ReceiptVerificationResult: + accepted: bool + diagnostics: list[str] = field(default_factory=list) + log_id: str | None = None + tree_size: int | None = None + root_hash: str | None = None + leaf_hash: str | None = None + signatures: dict[str, str] = field(default_factory=dict) + + def evidence(self) -> dict[str, Any]: + return { + "transparency_receipt_status": "accepted" if self.accepted else "rejected", + "transparency_log_id": self.log_id, + "transparency_tree_size": self.tree_size, + "transparency_root_hash": self.root_hash, + "transparency_leaf_hash": self.leaf_hash, + "transparency_signature_status": self.signatures, + } + + +def leaf_hash(data: bytes) -> bytes: + return hashlib.sha256(b"\x00" + data).digest() + + +def node_hash(left: bytes, right: bytes) -> bytes: + return hashlib.sha256(b"\x01" + left + right).digest() + + +def merkle_root(leaves: list[bytes]) -> bytes: + if not leaves: + return hashlib.sha256(b"").digest() + if len(leaves) == 1: + return leaf_hash(leaves[0]) + split = _largest_power_of_two_less_than(len(leaves)) + return node_hash(merkle_root(leaves[:split]), merkle_root(leaves[split:])) + + +def inclusion_proof(leaves: list[bytes], index: int) -> list[bytes]: + if index < 0 or index >= len(leaves): + raise ValueError(f"Leaf index {index} is outside tree size {len(leaves)}") + return _inclusion_path(leaves, index) + + +def verify_inclusion(leaf: bytes, index: int, tree_size: int, proof: list[bytes], root_hash: bytes) -> bool: + if tree_size <= 0 or index < 0 or index >= tree_size: + return False + try: + calculated, consumed = _calculate_inclusion_root(leaf_hash(leaf), index, tree_size, proof, 0) + except ValueError: + return False + return consumed == len(proof) and calculated == root_hash + + +def consistency_proof(leaves: list[bytes], old_tree_size: int) -> list[bytes]: + if old_tree_size < 0 or old_tree_size > len(leaves): + raise ValueError("old_tree_size must be between 0 and the current tree size") + if old_tree_size in (0, len(leaves)): + return [] + return _consistency_path(old_tree_size, leaves, complete=True) + + +def verify_consistency( + old_tree_size: int, + new_tree_size: int, + old_root_hash: bytes, + new_root_hash: bytes, + proof: list[bytes], +) -> bool: + if old_tree_size < 0 or new_tree_size < old_tree_size: + return False + if old_tree_size == 0: + return True + if old_tree_size == new_tree_size: + return old_root_hash == new_root_hash and not proof + if not proof: + return False + + nodes = list(proof) + if _is_power_of_two(old_tree_size): + nodes.insert(0, old_root_hash) + fn = old_tree_size - 1 + sn = new_tree_size - 1 + while fn & 1: + fn >>= 1 + sn >>= 1 + + old_hash = nodes.pop(0) + new_hash = old_hash + + while nodes: + node = nodes.pop(0) + if sn == 0: + return False + if fn & 1 or fn == sn: + old_hash = node_hash(node, old_hash) + new_hash = node_hash(node, new_hash) + if not (fn & 1): + while fn & 1 == 0 and fn != 0: + fn >>= 1 + sn >>= 1 + else: + new_hash = node_hash(new_hash, node) + fn >>= 1 + sn >>= 1 + + return old_hash == old_root_hash and new_hash == new_root_hash + + +def attestation_leaf(attestation: dict[str, Any]) -> dict[str, Any]: + return { + "schema_version": 1, + "type": LEAF_TYPE, + "attestation": attestation, + } + + +def leaf_bytes_for_attestation(attestation: dict[str, Any]) -> bytes: + return canonical_json(attestation_leaf(attestation)) + + +def leaf_hash_hex_for_attestation(attestation: dict[str, Any]) -> str: + return leaf_hash(leaf_bytes_for_attestation(attestation)).hex() + + +def signed_tree_head_payload(sth: dict[str, Any]) -> bytes: + payload = {key: value for key, value in sth.items() if key != "signatures"} + return canonical_json(payload) + + +def make_signed_tree_head( + log_id: str, + tree_size: int, + root_hash_hex: str, + timestamp: str, + private_key_path: str | Path, + public_key_path: str | Path, +) -> dict[str, Any]: + sth: dict[str, Any] = { + "schema_version": 1, + "type": STH_TYPE, + "log_id": log_id, + "tree_size": tree_size, + "timestamp": timestamp, + "root_hash": root_hash_hex, + "hash_algorithm": HASH_ALGORITHM, + } + payload = signed_tree_head_payload(sth) + sth["signatures"] = { + "ed25519": { + "scheme": "openssl-ed25519", + "status": "signed", + "payload_digest_sha256": hashlib.sha256(payload).hexdigest(), + "signature_base64": sign_payload_ed25519(payload, private_key_path), + "public_key_fingerprint_sha256": public_key_fingerprint(public_key_path), + }, + "ml_dsa": detect_ml_dsa().to_signature_slot(), + } + return sth + + +def verify_signed_tree_head( + sth: dict[str, Any], + public_key_path: str | Path, + require_signatures: str = "ed25519", +) -> tuple[bool, list[str], dict[str, str]]: + diagnostics: list[str] = [] + statuses: dict[str, str] = {} + if sth.get("type") != STH_TYPE: + diagnostics.append(f"Unexpected signed tree head type: {sth.get('type')}") + if sth.get("hash_algorithm") != HASH_ALGORITHM: + diagnostics.append(f"Unexpected hash algorithm: {sth.get('hash_algorithm')}") + + signatures = sth.get("signatures") or {} + ed25519 = signatures.get("ed25519") or {} + if ed25519.get("scheme") != "openssl-ed25519" or ed25519.get("status") != "signed": + diagnostics.append("Signed tree head is missing a usable Ed25519 signature.") + statuses["ed25519"] = str(ed25519.get("status") or "missing") + else: + payload = signed_tree_head_payload(sth) + expected_digest = ed25519.get("payload_digest_sha256") + actual_digest = hashlib.sha256(payload).hexdigest() + if expected_digest and expected_digest != actual_digest: + diagnostics.append("Signed tree head digest does not match signature metadata.") + statuses["ed25519"] = "digest_mismatch" + else: + expected_fingerprint = ed25519.get("public_key_fingerprint_sha256") + if expected_fingerprint and expected_fingerprint != public_key_fingerprint(public_key_path): + diagnostics.append("Signed tree head Ed25519 public-key fingerprint mismatch.") + statuses["ed25519"] = "key_mismatch" + else: + ok, error = verify_payload_ed25519(payload, str(ed25519.get("signature_base64") or ""), public_key_path) + statuses["ed25519"] = "verified" if ok else "failed" + if not ok: + diagnostics.append(f"Signed tree head Ed25519 verification failed: {error}") + + ml_dsa = signatures.get("ml_dsa") or {} + statuses["ml_dsa"] = str(ml_dsa.get("status") or "missing") + if require_signatures == "both" and statuses["ml_dsa"] != "verified": + diagnostics.append("ML-DSA signed tree head signature is required by policy but is not verified.") + elif require_signatures not in {"ed25519", "both"}: + diagnostics.append(f"Unsupported transparency signature policy: {require_signatures}") + return not diagnostics, diagnostics, statuses + + +def load_receipt(path: str | Path) -> dict[str, Any]: + raw = load_data(path) + if not isinstance(raw, dict): + raise ValueError(f"Transparency receipt must be a mapping: {path}") + return raw + + +def verify_receipt( + attestation: dict[str, Any], + receipt: dict[str, Any], + log_public_key_path: str | Path, + require_signatures: str = "ed25519", +) -> ReceiptVerificationResult: + diagnostics: list[str] = [] + if receipt.get("type") != RECEIPT_TYPE: + diagnostics.append(f"Unexpected transparency receipt type: {receipt.get('type')}") + if receipt.get("hash_algorithm") != HASH_ALGORITHM: + diagnostics.append(f"Unexpected transparency receipt hash algorithm: {receipt.get('hash_algorithm')}") + + sth = receipt.get("sth") or {} + sth_ok, sth_diagnostics, statuses = verify_signed_tree_head(sth, log_public_key_path, require_signatures=require_signatures) + diagnostics.extend(sth_diagnostics) + try: + tree_size = int(receipt.get("tree_size")) + leaf_index = int(receipt.get("leaf_index")) + except (TypeError, ValueError): + tree_size = -1 + leaf_index = -1 + diagnostics.append("Transparency receipt has invalid tree_size or leaf_index.") + + if sth.get("tree_size") != tree_size: + diagnostics.append("Transparency receipt tree_size does not match signed tree head.") + if receipt.get("log_id") != sth.get("log_id"): + diagnostics.append("Transparency receipt log_id does not match signed tree head.") + + leaf_bytes = leaf_bytes_for_attestation(attestation) + expected_leaf_hash = leaf_hash(leaf_bytes).hex() + if receipt.get("leaf_hash") != expected_leaf_hash: + diagnostics.append("Transparency receipt leaf hash does not match attestation.") + expected_attestation_digest = hashlib.sha256(canonical_json(attestation)).hexdigest() + if receipt.get("attestation_digest_sha256") and receipt.get("attestation_digest_sha256") != expected_attestation_digest: + diagnostics.append("Transparency receipt attestation digest does not match attestation.") + + try: + root = bytes.fromhex(str(sth.get("root_hash") or "")) + proof = [bytes.fromhex(str(item)) for item in (receipt.get("inclusion_proof") or [])] + except ValueError as exc: + root = b"" + proof = [] + diagnostics.append(f"Transparency proof contains invalid hex: {exc}") + + if root and tree_size >= 0: + if not verify_inclusion(leaf_bytes, leaf_index, tree_size, proof, root): + diagnostics.append("Transparency inclusion proof does not verify against the signed tree head.") + + accepted = not diagnostics and sth_ok + return ReceiptVerificationResult( + accepted=accepted, + diagnostics=diagnostics, + log_id=str(receipt.get("log_id") or sth.get("log_id") or "") or None, + tree_size=tree_size if tree_size >= 0 else None, + root_hash=str(sth.get("root_hash") or "") or None, + leaf_hash=str(receipt.get("leaf_hash") or "") or None, + signatures=statuses, + ) + + +def proof_to_hex(proof: list[bytes]) -> list[str]: + return [item.hex() for item in proof] + + +def _inclusion_path(leaves: list[bytes], index: int) -> list[bytes]: + if len(leaves) == 1: + return [] + split = _largest_power_of_two_less_than(len(leaves)) + if index < split: + return _inclusion_path(leaves[:split], index) + [merkle_root(leaves[split:])] + return _inclusion_path(leaves[split:], index - split) + [merkle_root(leaves[:split])] + + +def _calculate_inclusion_root( + current_hash: bytes, + index: int, + tree_size: int, + proof: list[bytes], + proof_index: int, +) -> tuple[bytes, int]: + if tree_size == 1: + return current_hash, proof_index + split = _largest_power_of_two_less_than(tree_size) + if index < split: + left, consumed = _calculate_inclusion_root(current_hash, index, split, proof, proof_index) + if consumed >= len(proof): + raise ValueError("proof exhausted") + return node_hash(left, proof[consumed]), consumed + 1 + right, consumed = _calculate_inclusion_root(current_hash, index - split, tree_size - split, proof, proof_index) + if consumed >= len(proof): + raise ValueError("proof exhausted") + return node_hash(proof[consumed], right), consumed + 1 + + +def _consistency_path(old_tree_size: int, leaves: list[bytes], complete: bool) -> list[bytes]: + if old_tree_size == len(leaves): + return [] if complete else [merkle_root(leaves)] + split = _largest_power_of_two_less_than(len(leaves)) + if old_tree_size <= split: + return _consistency_path(old_tree_size, leaves[:split], complete) + [merkle_root(leaves[split:])] + return _consistency_path(old_tree_size - split, leaves[split:], complete=False) + [merkle_root(leaves[:split])] + + +def _largest_power_of_two_less_than(value: int) -> int: + if value <= 1: + raise ValueError("value must be greater than one") + return 1 << ((value - 1).bit_length() - 1) + + +def _is_power_of_two(value: int) -> bool: + return value > 0 and value & (value - 1) == 0 diff --git a/tests/test_transparency.py b/tests/test_transparency.py new file mode 100644 index 0000000..63dd61c --- /dev/null +++ b/tests/test_transparency.py @@ -0,0 +1,147 @@ +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "provider" / "src")) + +from pacta.attestation import validate_attestation +from pacta.config import RepoConfig +from pacta.signing import generate_ed25519_keypair, sign_attestation +from pacta.transparency import ( + consistency_proof, + inclusion_proof, + merkle_root, + verify_consistency, + verify_inclusion, + verify_receipt, +) +from pacta_provider.transparency_log import TransparencyLog + + +def _repo() -> RepoConfig: + return RepoConfig( + name="dalek-ed25519-verified", + url="https://github.com/saymrwulf/dalek-ed25519-verified.git", + kind="ed25519", + verification_dir="verification", + verified_backend="serial/u64", + certificates=["CurveFieldProofs.fieldImplementation", "CurveFieldProofs.edwardsImplementation"], + ) + + +def _signed_attestation(tmp_path): + private_key = tmp_path / "provider.key" + public_key = tmp_path / "provider.pub" + generate_ed25519_keypair(private_key, public_key) + attestation = { + "schema_version": 1, + "provider": "local-test-provider", + "subject": { + "component": "dalek-ed25519-verified", + "repo_url": "https://github.com/saymrwulf/dalek-ed25519-verified.git", + "verification_dir": "verification", + }, + "environment": {}, + "replay": {"check_ok": True, "axiom_ok": True}, + "certificates": [ + { + "name": "CurveFieldProofs.fieldImplementation", + "status": "proven", + "axiom_status": "clean", + "observed_axioms": [], + "expected_axioms": [], + }, + { + "name": "CurveFieldProofs.edwardsImplementation", + "status": "proven", + "axiom_status": "clean", + "observed_axioms": [], + "expected_axioms": [], + }, + ], + } + return sign_attestation(attestation, private_key, public_key), private_key, public_key + + +def test_rfc9162_inclusion_and_consistency_proofs_round_trip(): + leaves = [f"leaf-{index}".encode() for index in range(1, 24)] + for tree_size in range(1, len(leaves) + 1): + root = merkle_root(leaves[:tree_size]) + for index in range(tree_size): + proof = inclusion_proof(leaves[:tree_size], index) + assert verify_inclusion(leaves[index], index, tree_size, proof, root) + for old_size in range(tree_size + 1): + proof = consistency_proof(leaves[:tree_size], old_size) + assert verify_consistency(old_size, tree_size, merkle_root(leaves[:old_size]), root, proof) + + +def test_provider_log_receipt_verifies_and_detects_tampering(tmp_path): + attestation, private_key, public_key = _signed_attestation(tmp_path) + attestation_path = tmp_path / "attestation.yaml" + receipt_path = tmp_path / "receipt.yaml" + from pacta.yamlio import dump_data + + dump_data(attestation, attestation_path) + log = TransparencyLog(tmp_path / "log") + log.init("local-test-provider", public_key) + receipt = log.append_attestation(attestation_path, private_key, public_key, receipt_out=receipt_path) + + result = verify_receipt(attestation, receipt, public_key) + assert result.accepted, result.diagnostics + assert result.signatures["ed25519"] == "verified" + + tampered = dict(attestation) + tampered["provider"] = "different-provider" + tampered_result = verify_receipt(tampered, receipt, public_key) + assert not tampered_result.accepted + assert any("leaf hash" in diagnostic for diagnostic in tampered_result.diagnostics) + + +def test_validate_attestation_can_require_transparency_receipt(tmp_path): + attestation, private_key, public_key = _signed_attestation(tmp_path) + from pacta.yamlio import dump_data + + attestation_path = tmp_path / "attestation.yaml" + receipt_path = tmp_path / "receipt.yaml" + dump_data(attestation, attestation_path) + log = TransparencyLog(tmp_path / "log") + log.init("local-test-provider", public_key) + log.append_attestation(attestation_path, private_key, public_key, receipt_out=receipt_path) + + accepted = validate_attestation( + attestation, + _repo(), + trusted_provider="local-test-provider", + public_key_path=public_key, + transparency_receipt_path=receipt_path, + transparency_log_public_key_path=public_key, + require_transparency_receipt=True, + ) + assert accepted.accepted, accepted.diagnostics + assert accepted.evidence["transparency_receipt_status"] == "accepted" + + rejected = validate_attestation( + attestation, + _repo(), + trusted_provider="local-test-provider", + public_key_path=public_key, + require_transparency_receipt=True, + ) + assert not rejected.accepted + assert any("Transparency receipt is required" in item for item in rejected.diagnostics) + + +def test_requiring_both_signatures_fails_without_ml_dsa_backend(tmp_path): + attestation, private_key, public_key = _signed_attestation(tmp_path) + from pacta.yamlio import dump_data + + attestation_path = tmp_path / "attestation.yaml" + receipt_path = tmp_path / "receipt.yaml" + dump_data(attestation, attestation_path) + log = TransparencyLog(tmp_path / "log") + log.init("local-test-provider", public_key) + receipt = log.append_attestation(attestation_path, private_key, public_key, receipt_out=receipt_path) + + result = verify_receipt(attestation, receipt, public_key, require_signatures="both") + assert not result.accepted + assert result.signatures["ed25519"] == "verified" + assert result.signatures["ml_dsa"] != "verified"