add nested proof check provider

This commit is contained in:
saymrwulf 2026-07-03 13:03:58 +02:00
parent 2282bb43c7
commit 0522cdfdca
18 changed files with 671 additions and 14 deletions

2
.gitignore vendored
View file

@ -4,6 +4,8 @@
.pytest_cache/ .pytest_cache/
artifacts*/ artifacts*/
repos/ repos/
provider/out/
provider/state/
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
*.egg-info/ *.egg-info/

View file

@ -89,12 +89,42 @@ pacta agent --config examples/repos.yaml \
--repo-name dalek-ed25519-verified \ --repo-name dalek-ed25519-verified \
--attestation examples/dalek-ed25519.attestation.yaml \ --attestation examples/dalek-ed25519.attestation.yaml \
--trust-attestation-provider example-proof-checker.invalid \ --trust-attestation-provider example-proof-checker.invalid \
--allow-unsigned-attestation \
--action build-library --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`. 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 ## Truth Boundary

24
provider/README.md Normal file
View file

@ -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.

16
provider/pyproject.toml Normal file
View file

@ -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"]

View file

@ -0,0 +1,5 @@
"""PACTA proof-checking attestation provider."""
__all__ = ["__version__"]
__version__ = "0.1.0"

View file

@ -0,0 +1,5 @@
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -5,6 +5,7 @@ from pathlib import Path
from typing import Any from typing import Any
from .config import RepoConfig from .config import RepoConfig
from .signing import verify_attestation_signature
from .yamlio import load_data from .yamlio import load_data
@ -31,6 +32,8 @@ def validate_attestation(
repo: RepoConfig, repo: RepoConfig,
path: str | Path | None = None, path: str | Path | None = None,
trusted_provider: str | None = None, trusted_provider: str | None = None,
public_key_path: str | Path | None = None,
allow_unsigned: bool = False,
) -> AttestationResult: ) -> AttestationResult:
provider = raw.get("provider") provider = raw.get("provider")
subject = raw.get("subject") or {} subject = raw.get("subject") or {}
@ -61,8 +64,21 @@ def validate_attestation(
signature = raw.get("signature") or {} signature = raw.get("signature") or {}
environment = raw.get("environment") or {} environment = raw.get("environment") or {}
signature_status = signature.get("status", "not_checked") signature_status = signature.get("status", "not_checked")
if signature_status not in {"verified", "not_implemented"}: if public_key_path:
diagnostics.append(f"Attestation signature status is not acceptable for this prototype: {signature_status}") 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 accepted = not diagnostics
evidence = { evidence = {
@ -72,6 +88,8 @@ def validate_attestation(
"attestation_signature_status": signature_status, "attestation_signature_status": signature_status,
"attestation_log_url": raw.get("log_url") or signature.get("log_url"), "attestation_log_url": raw.get("log_url") or signature.get("log_url"),
"attestation_issued_at": raw.get("issued_at"), "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"), "lean_version": environment.get("lean_version"),
"lake_version": environment.get("lake_version"), "lake_version": environment.get("lake_version"),
} }

View file

@ -45,6 +45,7 @@ def build_claim_card(
tools = detect_tools() tools = detect_tools()
certs = _certificate_claims(repo, axiom_audit, offline_fixture, attestation) certs = _certificate_claims(repo, axiom_audit, offline_fixture, attestation)
scanned_files = layout.relative_files() if layout else [] scanned_files = layout.relative_files() if layout else []
attestation_evidence = attestation.evidence if attestation else {}
card: dict[str, Any] = { card: dict[str, Any] = {
"component": repo.name, "component": repo.name,
"repo_url": repo.url, "repo_url": repo.url,
@ -59,13 +60,13 @@ def build_claim_card(
"exclusions": profile.exclusions, "exclusions": profile.exclusions,
"trusted_base": [*profile.trusted_base, *(attestation.trusted_base if attestation and attestation.accepted else [])], "trusted_base": [*profile.trusted_base, *(attestation.trusted_base if attestation and attestation.accepted else [])],
"evidence": { "evidence": {
"lean_version": tools.lean_version, "lean_version": attestation_evidence.get("lean_version") or tools.lean_version,
"lake_version": tools.lake_version, "lake_version": attestation_evidence.get("lake_version") or tools.lake_version,
"check_log_path": lean_check.log_path if lean_check else None, "check_log_path": attestation_evidence.get("check_log_path") or (lean_check.log_path if lean_check else None),
"axiom_log_path": axiom_audit.log_path if axiom_audit 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), "replay_blockers": _replay_blockers(lean_check, axiom_audit, attestation),
"scanned_files": scanned_files, "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": { "risk": {
"level": "R0", "level": "R0",

View file

@ -104,6 +104,8 @@ def build_parser() -> argparse.ArgumentParser:
claims.add_argument("--lean-project-dir") claims.add_argument("--lean-project-dir")
claims.add_argument("--attestation") claims.add_argument("--attestation")
claims.add_argument("--trust-attestation-provider") 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) claims.set_defaults(func=cmd_claims)
report = sub.add_parser("report", help="Generate a human-readable Markdown risk report.") 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("--lean-project-dir")
agent.add_argument("--attestation") agent.add_argument("--attestation")
agent.add_argument("--trust-attestation-provider") 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) agent.set_defaults(func=cmd_agent)
return parser return parser
@ -445,4 +449,11 @@ def _attestation_for_args(args: argparse.Namespace, repo: RepoConfig):
if not attestation_path: if not attestation_path:
return None return None
raw = load_attestation(attestation_path) 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)),
)

View file

@ -98,8 +98,11 @@ def build_lean_invocation(
tools: LeanTools, tools: LeanTools,
use_lake_env: bool = False, use_lake_env: bool = False,
output_path: str | Path | None = None, output_path: str | Path | None = None,
root_path: str | Path | None = None,
) -> list[str]: ) -> list[str]:
args = ["lean"] args = ["lean"]
if root_path is not None:
args.append(f"--root={root_path}")
if output_path is not None: if output_path is not None:
args.extend(["-o", str(output_path)]) args.extend(["-o", str(output_path)])
args.append(str(file_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"env_script: {env_script or ''}\n")
log.write(f"lean_project_dir: {project_dir or ''}\n\n") log.write(f"lean_project_dir: {project_dir or ''}\n\n")
for path in files: 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") log.write("$ " + " ".join(cmd) + "\n")
try: try:
completed = subprocess.run( completed = subprocess.run(
@ -326,6 +335,15 @@ def _log_file(log_dir: str | Path | None, name: str) -> Path:
return directory / name 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]: def _base_env(base_env: dict[str, str] | None, env_script: str | Path | None) -> dict[str, str]:
env = dict(base_env or os.environ) env = dict(base_env or os.environ)
if not env_script: 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" command = f"source {shlex.quote(str(path))}; env"
try: try:
completed = subprocess.run( completed = subprocess.run(
["/bin/zsh", "-lc", command], ["/bin/bash", "-lc", command],
check=False, check=False,
capture_output=True, capture_output=True,
text=True, text=True,

124
src/pacta/signing.py Normal file
View file

@ -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

View file

@ -1,6 +1,7 @@
from pacta.attestation import load_attestation, validate_attestation from pacta.attestation import load_attestation, validate_attestation
from pacta.claims import build_claim_card from pacta.claims import build_claim_card
from pacta.config import RepoConfig from pacta.config import RepoConfig
from pacta.signing import generate_ed25519_keypair, sign_attestation
def _repo(): def _repo():
@ -15,7 +16,13 @@ def _repo():
def test_trusted_attestation_can_drive_r3_claim(tmp_path): def test_trusted_attestation_can_drive_r3_claim(tmp_path):
raw = load_attestation("examples/dalek-ed25519.attestation.yaml") 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) card = build_claim_card(_repo(), tmp_path, attestation=result)
assert result.accepted assert result.accepted
assert card["risk"]["level"] == "R3" 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) card = build_claim_card(_repo(), tmp_path, attestation=result)
assert not result.accepted assert not result.accepted
assert card["risk"]["level"] == "R0" 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)

View file

@ -26,8 +26,8 @@ def test_parse_axiom_output_no_axioms_wording():
def test_mac_safe_lean_command_is_argument_list(): def test_mac_safe_lean_command_is_argument_list():
tools = LeanTools(lean="/usr/local/bin/lean", lake=None) tools = LeanTools(lean="/usr/local/bin/lean", lake=None)
cmd = build_lean_invocation(Path("Proofs/A.lean"), tools, output_path=Path("Proofs/A.olean")) cmd = build_lean_invocation(Path("Proofs/A.lean"), tools, output_path=Path("Proofs/A.olean"), root_path=Path("."))
assert cmd == ["/usr/local/bin/lean", "-o", "Proofs/A.olean", "Proofs/A.lean"] assert cmd == ["/usr/local/bin/lean", "--root=.", "-o", "Proofs/A.olean", "Proofs/A.lean"]
assert "timeout" not in cmd assert "timeout" not in cmd
assert "taskset" not in cmd assert "taskset" not in cmd
assert "free" not in cmd assert "free" not in cmd

50
tests/test_provider.py Normal file
View file

@ -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"

20
tests/test_signing.py Normal file
View file

@ -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