From 0522cdfdca011d5126eb7862e8f6ec4ab0c09665 Mon Sep 17 00:00:00 2001 From: saymrwulf Date: Fri, 3 Jul 2026 13:03:58 +0200 Subject: [PATCH] add nested proof check provider --- .gitignore | 2 + README.md | 32 ++++- provider/README.md | 24 ++++ provider/pyproject.toml | 16 +++ provider/src/pacta_provider/__init__.py | 5 + provider/src/pacta_provider/__main__.py | 5 + provider/src/pacta_provider/cli.py | 88 ++++++++++++++ provider/src/pacta_provider/discovery.py | 82 +++++++++++++ provider/src/pacta_provider/service.py | 141 +++++++++++++++++++++++ src/pacta/attestation.py | 22 +++- src/pacta/claims.py | 11 +- src/pacta/cli.py | 13 ++- src/pacta/lean.py | 22 +++- src/pacta/signing.py | 124 ++++++++++++++++++++ tests/test_attestation.py | 24 +++- tests/test_lean.py | 4 +- tests/test_provider.py | 50 ++++++++ tests/test_signing.py | 20 ++++ 18 files changed, 671 insertions(+), 14 deletions(-) create mode 100644 provider/README.md create mode 100644 provider/pyproject.toml create mode 100644 provider/src/pacta_provider/__init__.py create mode 100644 provider/src/pacta_provider/__main__.py create mode 100644 provider/src/pacta_provider/cli.py create mode 100644 provider/src/pacta_provider/discovery.py create mode 100644 provider/src/pacta_provider/service.py create mode 100644 src/pacta/signing.py create mode 100644 tests/test_provider.py create mode 100644 tests/test_signing.py diff --git a/.gitignore b/.gitignore index 7e8f782..2024295 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,8 @@ .pytest_cache/ artifacts*/ repos/ +provider/out/ +provider/state/ __pycache__/ *.py[cod] *.egg-info/ diff --git a/README.md b/README.md index 915c8dc..bac31db 100644 --- a/README.md +++ b/README.md @@ -89,12 +89,42 @@ pacta agent --config examples/repos.yaml \ --repo-name dalek-ed25519-verified \ --attestation examples/dalek-ed25519.attestation.yaml \ --trust-attestation-provider example-proof-checker.invalid \ + --allow-unsigned-attestation \ --action build-library ``` This changes the trusted base. The agent is no longer trusting local Lean replay; it is trusting the proof-checking service, its environment, signing key custody, and log retention. Without an explicitly trusted provider, attestation evidence scores `R0`. -The included `examples/dalek-ed25519.attestation.yaml` is a schema/demo fixture. A production service should publish verifiable signatures, transparency-log entries, and reproducible logs; the prototype records signature metadata but does not yet implement cryptographic signature verification. +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`. + +## 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. + +```bash +PYTHONPATH=src:provider/src python -m pacta_provider discover --root ~/GitClone +PYTHONPATH=src:provider/src python -m pacta_provider init-key --key-dir provider/state/local-provider +PYTHONPATH=src:provider/src python -m pacta_provider check \ + --config examples/repos.yaml \ + --repo-name dalek-ed25519-verified \ + --repo repos/dalek-ed25519-verified \ + --provider local-pacta-provider \ + --private-key provider/state/local-provider/provider.ed25519.key \ + --public-key provider/state/local-provider/provider.ed25519.pub \ + --env-script /path/to/aeneas-toolchain/env.sh \ + --lean-project-dir '$AENEAS_HOME/backends/lean' \ + --out provider/out/dalek-ed25519.attestation.yaml + +pacta agent \ + --config examples/repos.yaml \ + --repo-name 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 \ + --action build-library +``` + +This is the intended trust transformation: local agents can avoid constructing the full verifier environment, but they must explicitly trust the provider identity and verification key. ## Truth Boundary diff --git a/provider/README.md b/provider/README.md new file mode 100644 index 0000000..ccb6fe2 --- /dev/null +++ b/provider/README.md @@ -0,0 +1,24 @@ +# PACTA Proof Check Provider + +This nested project is a prototype third-party proof-checking service. It reuses host Lean/Aeneas infrastructure, runs portable PACTA replay/audit checks, and emits signed attestation certificates. + +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`. + +## Commands + +```bash +PYTHONPATH=src:provider/src python -m pacta_provider discover +PYTHONPATH=src:provider/src python -m pacta_provider init-key --key-dir provider/state/demo-provider +PYTHONPATH=src:provider/src python -m pacta_provider check \ + --config examples/repos.yaml \ + --repo-name dalek-ed25519-verified \ + --repo repos/dalek-ed25519-verified \ + --provider local-pacta-provider \ + --private-key provider/state/demo-provider/provider.ed25519.key \ + --public-key provider/state/demo-provider/provider.ed25519.pub \ + --out provider/out/dalek.attestation.yaml +``` + +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. diff --git a/provider/pyproject.toml b/provider/pyproject.toml new file mode 100644 index 0000000..dbe9988 --- /dev/null +++ b/provider/pyproject.toml @@ -0,0 +1,16 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "pacta-proof-check-provider" +version = "0.1.0" +description = "Nested proof-checking attestation provider for PACTA." +requires-python = ">=3.11" +dependencies = [] + +[project.scripts] +pacta-provider = "pacta_provider.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/provider/src/pacta_provider/__init__.py b/provider/src/pacta_provider/__init__.py new file mode 100644 index 0000000..54ca0fc --- /dev/null +++ b/provider/src/pacta_provider/__init__.py @@ -0,0 +1,5 @@ +"""PACTA proof-checking attestation provider.""" + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/provider/src/pacta_provider/__main__.py b/provider/src/pacta_provider/__main__.py new file mode 100644 index 0000000..a049ad7 --- /dev/null +++ b/provider/src/pacta_provider/__main__.py @@ -0,0 +1,5 @@ +from .cli import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/provider/src/pacta_provider/cli.py b/provider/src/pacta_provider/cli.py new file mode 100644 index 0000000..8f8215a --- /dev/null +++ b/provider/src/pacta_provider/cli.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from pacta.config import load_config +from pacta.signing import generate_ed25519_keypair +from pacta.yamlio import dump_data + +from .discovery import discover_toolchains +from .service import build_attestation + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + return args.func(args) + except Exception as exc: + print(f"error: {exc}") + return 2 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="pacta-provider", description="PACTA proof-checking attestation provider.") + sub = parser.add_subparsers(dest="command", required=True) + + discover = sub.add_parser("discover", help="Find reusable local Lean/Aeneas toolchains.") + discover.add_argument("--root", action="append") + discover.add_argument("--max-depth", type=int, default=6) + discover.set_defaults(func=cmd_discover) + + init_key = sub.add_parser("init-key", help="Create an Ed25519 provider signing keypair.") + init_key.add_argument("--key-dir", default="provider/state/local-provider") + init_key.set_defaults(func=cmd_init_key) + + check = sub.add_parser("check", help="Run proof checks and emit a signed attestation.") + check.add_argument("--config", required=True) + check.add_argument("--repo-name", required=True) + check.add_argument("--repo", required=True) + check.add_argument("--provider", required=True) + check.add_argument("--private-key", required=True) + check.add_argument("--public-key", required=True) + check.add_argument("--env-script") + check.add_argument("--lean-project-dir") + check.add_argument("--timeout", type=int, default=120) + check.add_argument("--log-dir", default="provider/out/logs") + check.add_argument("--out", required=True) + check.set_defaults(func=cmd_check) + return parser + + +def cmd_discover(args: argparse.Namespace) -> int: + candidates = discover_toolchains(args.root, max_depth=args.max_depth) + print(json.dumps([candidate.to_dict() for candidate in candidates], indent=2)) + return 0 if candidates else 1 + + +def cmd_init_key(args: argparse.Namespace) -> int: + key_dir = Path(args.key_dir) + private_key = key_dir / "provider.ed25519.key" + public_key = key_dir / "provider.ed25519.pub" + if private_key.exists() or public_key.exists(): + raise ValueError(f"Refusing to overwrite existing key files in {key_dir}") + generate_ed25519_keypair(private_key, public_key) + print(f"private_key: {private_key}") + print(f"public_key: {public_key}") + return 0 + + +def cmd_check(args: argparse.Namespace) -> int: + config = load_config(args.config) + repo = config.repo_named(args.repo_name) + attestation = build_attestation( + repo, + args.repo, + provider=args.provider, + private_key=args.private_key, + public_key=args.public_key, + env_script=args.env_script, + lean_project_dir=args.lean_project_dir, + timeout=args.timeout, + log_dir=args.log_dir, + ) + dump_data(attestation, args.out) + print(f"attestation: {args.out}") + return 0 diff --git a/provider/src/pacta_provider/discovery.py b/provider/src/pacta_provider/discovery.py new file mode 100644 index 0000000..6fc18d9 --- /dev/null +++ b/provider/src/pacta_provider/discovery.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +from pacta.lean import build_lean_env, detect_tools, resolve_lean_project_dir + + +SKIP_DIRS = {".git", ".venv", "__pycache__", "node_modules", "target", ".lake", ".pytest_cache"} + + +@dataclass(slots=True) +class ToolchainCandidate: + env_script: Path + lean_project_dir: Path | None + lean: str | None + lake: str | None + aeneas_home: str | None + + def to_dict(self) -> dict[str, str | None]: + return { + "env_script": str(self.env_script), + "lean_project_dir": str(self.lean_project_dir) if self.lean_project_dir else None, + "lean": self.lean, + "lake": self.lake, + "aeneas_home": self.aeneas_home, + } + + +def discover_toolchains(roots: Iterable[str | Path] | None = None, max_depth: int = 6) -> list[ToolchainCandidate]: + home = Path.home() + search_roots = [home / "GitClone", home] if roots is None else [Path(root).expanduser() for root in roots] + scripts: list[Path] = [] + for root in search_roots: + if root.exists(): + scripts.extend(_find_env_scripts(root, max_depth=max_depth)) + candidates: list[ToolchainCandidate] = [] + seen: set[Path] = set() + for script in scripts: + if script in seen: + continue + seen.add(script) + env = build_lean_env("verification", env_script=script) + tools = detect_tools(env) + project = resolve_lean_project_dir("$AENEAS_HOME/backends/lean", env) + if project is None: + project = _nearby_lean_project(script) + candidates.append( + ToolchainCandidate( + env_script=script, + lean_project_dir=project, + lean=tools.lean, + lake=tools.lake, + aeneas_home=env.get("AENEAS_HOME"), + ) + ) + return candidates + + +def _find_env_scripts(root: Path, max_depth: int) -> list[Path]: + root = root.resolve() + found: list[Path] = [] + for current, dirs, files in os.walk(root): + current_path = Path(current) + rel_depth = len(current_path.relative_to(root).parts) + if rel_depth >= max_depth: + dirs[:] = [] + else: + dirs[:] = [d for d in dirs if d not in SKIP_DIRS] + if "env.sh" in files and ("aeneas" in str(current_path).lower() or "lean" in str(current_path).lower()): + found.append(current_path / "env.sh") + return sorted(found) + + +def _nearby_lean_project(script: Path) -> Path | None: + for parent in [script.parent, *script.parents]: + for candidate in (parent / "aeneas" / "backends" / "lean", parent / "backends" / "lean"): + if (candidate / "lakefile.lean").exists() or (candidate / "lakefile.toml").exists(): + return candidate + return None diff --git a/provider/src/pacta_provider/service.py b/provider/src/pacta_provider/service.py new file mode 100644 index 0000000..ea7ca90 --- /dev/null +++ b/provider/src/pacta_provider/service.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +import shutil +import subprocess +from typing import Any + +from pacta.attestation import validate_attestation +from pacta.config import RepoConfig +from pacta.lean import LeanCheckResult, build_lean_env, detect_tools, lean_check_files, resolve_lean_project_dir, run_axiom_audit +from pacta.manifest import discover_layout +from pacta.profiles import get_profile +from pacta.repo import git_commit +from pacta.signing import sign_attestation + + +def build_attestation( + repo: RepoConfig, + repo_path: str | Path, + provider: str, + private_key: str | Path, + public_key: str | Path, + env_script: str | Path | None = None, + lean_project_dir: str | Path | None = None, + timeout: int = 120, + log_dir: str | Path = "provider/out/logs", +) -> dict[str, Any]: + path = Path(repo_path) + profile = get_profile(repo.kind, repo) + layout = discover_layout(path, repo.verification_dir) + check = lean_check_files( + layout.compile_order, + layout.verification_dir, + timeout=timeout, + log_dir=log_dir, + env_script=env_script or repo.env_script, + lean_project_dir=lean_project_dir or repo.lean_project_dir, + ) + axiom = None + if check.attempted and check.ok: + axiom = run_axiom_audit( + path / repo.verification_dir, + profile.axiom_imports, + repo.certificates or profile.default_certificates, + profile.expected_axioms, + timeout=timeout, + log_dir=log_dir, + env_script=env_script or repo.env_script, + lean_project_dir=lean_project_dir or repo.lean_project_dir, + ) + certs = [ + { + "name": cert.name, + "status": cert.status, + "axiom_status": cert.axiom_status, + "observed_axioms": cert.observed_axioms, + "expected_axioms": cert.expected_axioms, + "diagnostics": cert.diagnostics, + } + for cert in axiom.certificates + ] + else: + certs = _failed_certificates(repo, profile.expected_axioms, check) + provider_env = build_lean_env(path / repo.verification_dir, env_script=env_script or repo.env_script) + project_dir = resolve_lean_project_dir(lean_project_dir or repo.lean_project_dir, provider_env) + tools = detect_tools(provider_env) + lean_version = _project_version(["lake", "env", "lean", "--version"], project_dir, provider_env) or tools.lean_version + lake_version = _project_version(["lake", "--version"], project_dir, provider_env) or tools.lake_version + unsigned = { + "schema_version": 1, + "provider": provider, + "issued_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), + "subject": { + "component": repo.name, + "repo_url": repo.url, + "repo_commit": git_commit(path), + "verification_dir": repo.verification_dir, + "kind": repo.kind, + "verified_backend": repo.verified_backend, + }, + "environment": { + "lean_version": lean_version, + "lake_version": lake_version, + "env_script": str(env_script or repo.env_script or ""), + "lean_project_dir": str(project_dir or lean_project_dir or repo.lean_project_dir or ""), + }, + "replay": { + "check_attempted": check.attempted, + "check_ok": check.ok, + "check_log_path": check.log_path, + "checked_files": len(check.checked_files), + "failed_files": check.failed_files, + "diagnostics": check.diagnostics, + "axiom_attempted": axiom.attempted if axiom else False, + "axiom_ok": axiom.ok if axiom else False, + "axiom_log_path": axiom.log_path if axiom else None, + "axiom_diagnostics": axiom.diagnostics if axiom else [], + }, + "certificates": certs, + } + signed = sign_attestation(unsigned, private_key, public_key) + # Self-check before emitting a certificate. + validation = validate_attestation(signed, repo, trusted_provider=provider, public_key_path=public_key) + if not validation.accepted: + signed.setdefault("provider_warnings", []).extend(validation.diagnostics) + return signed + + +def _failed_certificates(repo: RepoConfig, expected_axioms: list[str], check: LeanCheckResult) -> list[dict[str, Any]]: + return [ + { + "name": name, + "status": "failed" if check.attempted else "unknown", + "axiom_status": "not_checked", + "observed_axioms": [], + "expected_axioms": expected_axioms, + } + for name in repo.certificates + ] + + +def _project_version(cmd: list[str], cwd: Path | None, env: dict[str, str]) -> str | None: + executable = shutil.which(cmd[0], path=env.get("PATH")) + if not executable: + return None + try: + completed = subprocess.run( + [executable, *cmd[1:]], + check=False, + capture_output=True, + text=True, + timeout=20, + cwd=str(cwd) if cwd else None, + env=env, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if completed.returncode != 0: + return None + return (completed.stdout or completed.stderr).strip() or None diff --git a/src/pacta/attestation.py b/src/pacta/attestation.py index 7f1a6a3..8e9e2e1 100644 --- a/src/pacta/attestation.py +++ b/src/pacta/attestation.py @@ -5,6 +5,7 @@ from pathlib import Path from typing import Any from .config import RepoConfig +from .signing import verify_attestation_signature from .yamlio import load_data @@ -31,6 +32,8 @@ def validate_attestation( repo: RepoConfig, path: str | Path | None = None, trusted_provider: str | None = None, + public_key_path: str | Path | None = None, + allow_unsigned: bool = False, ) -> AttestationResult: provider = raw.get("provider") subject = raw.get("subject") or {} @@ -61,8 +64,21 @@ def validate_attestation( signature = raw.get("signature") or {} environment = raw.get("environment") or {} signature_status = signature.get("status", "not_checked") - if signature_status not in {"verified", "not_implemented"}: - diagnostics.append(f"Attestation signature status is not acceptable for this prototype: {signature_status}") + if public_key_path: + ok, error = verify_attestation_signature(raw, public_key_path) + if ok: + signature_status = "verified" + else: + diagnostics.append(f"Attestation signature verification failed: {error}") + elif signature_status == "signed": + diagnostics.append("Signed attestation requires --attestation-public-key.") + elif signature_status == "not_implemented": + if allow_unsigned: + signature_status = "not_implemented" + else: + diagnostics.append("Unsigned attestation requires --allow-unsigned-attestation.") + elif signature_status != "verified": + diagnostics.append(f"Attestation signature status is not acceptable: {signature_status}") accepted = not diagnostics evidence = { @@ -72,6 +88,8 @@ def validate_attestation( "attestation_signature_status": signature_status, "attestation_log_url": raw.get("log_url") or signature.get("log_url"), "attestation_issued_at": raw.get("issued_at"), + "check_log_path": (raw.get("replay") or {}).get("check_log_path"), + "axiom_log_path": (raw.get("replay") or {}).get("axiom_log_path"), "lean_version": environment.get("lean_version"), "lake_version": environment.get("lake_version"), } diff --git a/src/pacta/claims.py b/src/pacta/claims.py index 799aae7..ec45d43 100644 --- a/src/pacta/claims.py +++ b/src/pacta/claims.py @@ -45,6 +45,7 @@ def build_claim_card( tools = detect_tools() certs = _certificate_claims(repo, axiom_audit, offline_fixture, attestation) scanned_files = layout.relative_files() if layout else [] + attestation_evidence = attestation.evidence if attestation else {} card: dict[str, Any] = { "component": repo.name, "repo_url": repo.url, @@ -59,13 +60,13 @@ def build_claim_card( "exclusions": profile.exclusions, "trusted_base": [*profile.trusted_base, *(attestation.trusted_base if attestation and attestation.accepted else [])], "evidence": { - "lean_version": tools.lean_version, - "lake_version": tools.lake_version, - "check_log_path": lean_check.log_path if lean_check else None, - "axiom_log_path": axiom_audit.log_path if axiom_audit else None, + "lean_version": attestation_evidence.get("lean_version") or tools.lean_version, + "lake_version": attestation_evidence.get("lake_version") or tools.lake_version, + "check_log_path": attestation_evidence.get("check_log_path") or (lean_check.log_path if lean_check else None), + "axiom_log_path": attestation_evidence.get("axiom_log_path") or (axiom_audit.log_path if axiom_audit else None), "replay_blockers": _replay_blockers(lean_check, axiom_audit, attestation), "scanned_files": scanned_files, - **(attestation.evidence if attestation else {"evidence_mode": "local_or_fixture"}), + **(attestation_evidence if attestation else {"evidence_mode": "local_or_fixture"}), }, "risk": { "level": "R0", diff --git a/src/pacta/cli.py b/src/pacta/cli.py index c7d9429..a937706 100644 --- a/src/pacta/cli.py +++ b/src/pacta/cli.py @@ -104,6 +104,8 @@ def build_parser() -> argparse.ArgumentParser: claims.add_argument("--lean-project-dir") claims.add_argument("--attestation") claims.add_argument("--trust-attestation-provider") + claims.add_argument("--attestation-public-key") + claims.add_argument("--allow-unsigned-attestation", action="store_true") claims.set_defaults(func=cmd_claims) report = sub.add_parser("report", help="Generate a human-readable Markdown risk report.") @@ -139,6 +141,8 @@ def build_parser() -> argparse.ArgumentParser: agent.add_argument("--lean-project-dir") agent.add_argument("--attestation") agent.add_argument("--trust-attestation-provider") + agent.add_argument("--attestation-public-key") + agent.add_argument("--allow-unsigned-attestation", action="store_true") agent.set_defaults(func=cmd_agent) return parser @@ -445,4 +449,11 @@ def _attestation_for_args(args: argparse.Namespace, repo: RepoConfig): if not attestation_path: return None raw = load_attestation(attestation_path) - return validate_attestation(raw, repo, path=attestation_path, trusted_provider=getattr(args, "trust_attestation_provider", None)) + return validate_attestation( + raw, + repo, + path=attestation_path, + 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)), + ) diff --git a/src/pacta/lean.py b/src/pacta/lean.py index 89e1fe6..b47f1b6 100644 --- a/src/pacta/lean.py +++ b/src/pacta/lean.py @@ -98,8 +98,11 @@ def build_lean_invocation( tools: LeanTools, use_lake_env: bool = False, output_path: str | Path | None = None, + root_path: str | Path | None = None, ) -> list[str]: args = ["lean"] + if root_path is not None: + args.append(f"--root={root_path}") if output_path is not None: args.extend(["-o", str(output_path)]) args.append(str(file_path)) @@ -163,7 +166,13 @@ def lean_check_files( log.write(f"env_script: {env_script or ''}\n") log.write(f"lean_project_dir: {project_dir or ''}\n\n") for path in files: - cmd = build_lean_invocation(path, tools, use_lake_env=use_lake_env, output_path=path.with_suffix(".olean")) + cmd = build_lean_invocation( + path, + tools, + use_lake_env=use_lake_env, + output_path=path.with_suffix(".olean"), + root_path=_lean_root_for_file(path, verification), + ) log.write("$ " + " ".join(cmd) + "\n") try: completed = subprocess.run( @@ -326,6 +335,15 @@ def _log_file(log_dir: str | Path | None, name: str) -> Path: return directory / name +def _lean_root_for_file(path: Path, verification_dir: Path) -> Path: + gen_dir = verification_dir / "gen" + try: + path.relative_to(gen_dir) + return gen_dir + except ValueError: + return verification_dir + + def _base_env(base_env: dict[str, str] | None, env_script: str | Path | None) -> dict[str, str]: env = dict(base_env or os.environ) if not env_script: @@ -336,7 +354,7 @@ def _base_env(base_env: dict[str, str] | None, env_script: str | Path | None) -> command = f"source {shlex.quote(str(path))}; env" try: completed = subprocess.run( - ["/bin/zsh", "-lc", command], + ["/bin/bash", "-lc", command], check=False, capture_output=True, text=True, diff --git a/src/pacta/signing.py b/src/pacta/signing.py new file mode 100644 index 0000000..5c6efea --- /dev/null +++ b/src/pacta/signing.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +import shutil +import subprocess +import tempfile +from pathlib import Path +from typing import Any + + +class SignatureError(RuntimeError): + pass + + +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") + + +def payload_digest(attestation: dict[str, Any]) -> str: + return hashlib.sha256(canonical_attestation_payload(attestation)).hexdigest() + + +def public_key_fingerprint(public_key_path: str | Path) -> str: + data = Path(public_key_path).read_bytes() + return hashlib.sha256(data).hexdigest() + + +def generate_ed25519_keypair(private_key_path: str | Path, public_key_path: str | Path) -> None: + openssl = _openssl() + private_path = Path(private_key_path) + public_path = Path(public_key_path) + private_path.parent.mkdir(parents=True, exist_ok=True) + public_path.parent.mkdir(parents=True, exist_ok=True) + subprocess.run([openssl, "genpkey", "-algorithm", "ed25519", "-out", str(private_path)], check=True, timeout=30) + subprocess.run([openssl, "pkey", "-in", str(private_path), "-pubout", "-out", str(public_path)], check=True, timeout=30) + + +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() + 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"), + } + if public_key_path: + signed["signature"]["public_key_fingerprint_sha256"] = public_key_fingerprint(public_key_path) + return signed + + +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')}" + encoded = signature.get("signature_base64") + if not encoded: + return False, "Attestation signature is missing signature_base64." + expected_digest = signature.get("payload_digest_sha256") + actual_digest = payload_digest(attestation) + if expected_digest and expected_digest != actual_digest: + return False, "Attestation payload digest does not match signature metadata." + expected_fingerprint = signature.get("public_key_fingerprint_sha256") + if expected_fingerprint: + actual_fingerprint = public_key_fingerprint(public_key_path) + if expected_fingerprint != actual_fingerprint: + return False, "Attestation public key fingerprint does not match signature metadata." + try: + signature_bytes = base64.b64decode(encoded, 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" + signature_path = Path(tmp) / "payload.sig" + payload_path.write_bytes(canonical_attestation_payload(attestation)) + signature_path.write_bytes(signature_bytes) + completed = subprocess.run( + [ + openssl, + "pkeyutl", + "-verify", + "-pubin", + "-inkey", + str(public_key_path), + "-rawin", + "-in", + str(payload_path), + "-sigfile", + str(signature_path), + ], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + if completed.returncode != 0: + return False, (completed.stderr or completed.stdout or "openssl verification failed").strip() + return True, None + + +def _openssl() -> str: + path = shutil.which("openssl") + if not path: + raise SignatureError("openssl is not available on PATH") + return path diff --git a/tests/test_attestation.py b/tests/test_attestation.py index b5fa36d..ea52836 100644 --- a/tests/test_attestation.py +++ b/tests/test_attestation.py @@ -1,6 +1,7 @@ from pacta.attestation import load_attestation, validate_attestation from pacta.claims import build_claim_card from pacta.config import RepoConfig +from pacta.signing import generate_ed25519_keypair, sign_attestation def _repo(): @@ -15,7 +16,13 @@ def _repo(): def test_trusted_attestation_can_drive_r3_claim(tmp_path): raw = load_attestation("examples/dalek-ed25519.attestation.yaml") - result = validate_attestation(raw, _repo(), path="examples/dalek-ed25519.attestation.yaml", trusted_provider="example-proof-checker.invalid") + result = validate_attestation( + raw, + _repo(), + path="examples/dalek-ed25519.attestation.yaml", + trusted_provider="example-proof-checker.invalid", + allow_unsigned=True, + ) card = build_claim_card(_repo(), tmp_path, attestation=result) assert result.accepted assert card["risk"]["level"] == "R3" @@ -30,3 +37,18 @@ def test_untrusted_attestation_scores_r0(tmp_path): card = build_claim_card(_repo(), tmp_path, attestation=result) assert not result.accepted assert card["risk"]["level"] == "R0" + + +def test_signed_attestation_requires_public_key(tmp_path): + private_key = tmp_path / "provider.key" + public_key = tmp_path / "provider.pub" + generate_ed25519_keypair(private_key, public_key) + raw = load_attestation("examples/dalek-ed25519.attestation.yaml") + raw["provider"] = "signed-test-provider" + raw["signature"] = {} + signed = sign_attestation(raw, private_key, public_key) + result = validate_attestation(signed, _repo(), trusted_provider="signed-test-provider") + card = build_claim_card(_repo(), tmp_path, attestation=result) + assert not result.accepted + assert card["risk"]["level"] == "R0" + assert any("attestation-public-key" in item for item in result.diagnostics) diff --git a/tests/test_lean.py b/tests/test_lean.py index 4827084..c075fd9 100644 --- a/tests/test_lean.py +++ b/tests/test_lean.py @@ -26,8 +26,8 @@ def test_parse_axiom_output_no_axioms_wording(): def test_mac_safe_lean_command_is_argument_list(): tools = LeanTools(lean="/usr/local/bin/lean", lake=None) - cmd = build_lean_invocation(Path("Proofs/A.lean"), tools, output_path=Path("Proofs/A.olean")) - assert cmd == ["/usr/local/bin/lean", "-o", "Proofs/A.olean", "Proofs/A.lean"] + cmd = build_lean_invocation(Path("Proofs/A.lean"), tools, output_path=Path("Proofs/A.olean"), root_path=Path(".")) + assert cmd == ["/usr/local/bin/lean", "--root=.", "-o", "Proofs/A.olean", "Proofs/A.lean"] assert "timeout" not in cmd assert "taskset" not in cmd assert "free" not in cmd diff --git a/tests/test_provider.py b/tests/test_provider.py new file mode 100644 index 0000000..eb87fd1 --- /dev/null +++ b/tests/test_provider.py @@ -0,0 +1,50 @@ +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "provider" / "src")) + +from pacta.config import RepoConfig +from pacta.signing import generate_ed25519_keypair +from pacta_provider.discovery import discover_toolchains +from pacta_provider.service import build_attestation + + +def test_provider_discovery_finds_env_script(tmp_path): + root = tmp_path / "toolchains" / "aeneas-toolchain" + lean = root / "aeneas" / "backends" / "lean" + lean.mkdir(parents=True) + (lean / "lakefile.lean").write_text("", encoding="utf-8") + (root / "env.sh").write_text( + 'SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"\n' + 'export AENEAS_HOME="$SCRIPT_DIR/aeneas"\n', + encoding="utf-8", + ) + candidates = discover_toolchains([tmp_path], max_depth=5) + assert candidates + assert candidates[0].lean_project_dir == lean + + +def test_provider_builds_signed_attestation_for_fixture(tmp_path): + private_key = tmp_path / "provider.key" + public_key = tmp_path / "provider.pub" + generate_ed25519_keypair(private_key, public_key) + repo = 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"], + expected_axioms=[], + ) + attestation = build_attestation( + repo, + Path("tests/fixtures/mini-ed25519-verified"), + provider="local-test-provider", + private_key=private_key, + public_key=public_key, + timeout=30, + log_dir=tmp_path / "logs", + ) + assert attestation["signature"]["status"] == "signed" + assert attestation["certificates"][0]["status"] == "proven" diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..d72a746 --- /dev/null +++ b/tests/test_signing.py @@ -0,0 +1,20 @@ +from pacta.signing import generate_ed25519_keypair, sign_attestation, verify_attestation_signature + + +def test_openssl_ed25519_attestation_signature_round_trip(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": "mini"}, + "certificates": [], + } + signed = sign_attestation(attestation, private_key, public_key) + ok, error = verify_attestation_signature(signed, public_key) + assert ok, error + signed["subject"]["component"] = "tampered" + ok, error = verify_attestation_signature(signed, public_key) + assert not ok + assert error