mirror of
https://github.com/saymrwulf/proof-aware-crypto-tooling-agent.git
synced 2026-09-04 20:03:40 +00:00
add transparency log trust provider
This commit is contained in:
parent
0522cdfdca
commit
0461d2f997
13 changed files with 975 additions and 22 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -2,6 +2,8 @@
|
|||
.pacta/
|
||||
.pacta-*/
|
||||
.pytest_cache/
|
||||
.pytest-tmp/
|
||||
.tmp/
|
||||
artifacts*/
|
||||
repos/
|
||||
provider/out/
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
64
README.md
64
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
|
||||
```
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
178
provider/src/pacta_provider/transparency_log.py
Normal file
178
provider/src/pacta_provider/transparency_log.py
Normal file
|
|
@ -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")
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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)),
|
||||
)
|
||||
|
|
|
|||
62
src/pacta/postquantum.py
Normal file
62
src/pacta/postquantum.py
Normal file
|
|
@ -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.",
|
||||
)
|
||||
|
|
@ -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(
|
||||
[
|
||||
|
|
|
|||
340
src/pacta/transparency.py
Normal file
340
src/pacta/transparency.py
Normal file
|
|
@ -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
|
||||
147
tests/test_transparency.py
Normal file
147
tests/test_transparency.py
Normal file
|
|
@ -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"
|
||||
Loading…
Reference in a new issue