112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify signed Python profile membership and emit a bound verification receipt."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
|
|
class VerificationError(ValueError):
|
|
pass
|
|
|
|
|
|
def canonical(value: dict) -> bytes:
|
|
return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode()
|
|
|
|
|
|
def sha256(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def verify(manifest: dict, envelope: dict, trust: dict, public_key: Path,
|
|
environment: str, verified_at: str, verifier_build_sha256: str) -> dict:
|
|
body = canonical(manifest)
|
|
manifest_digest = sha256(body)
|
|
if manifest.get("namespace") != "offline-workstation/python-runtime-profiles":
|
|
raise VerificationError("NAMESPACE_INVALID")
|
|
if envelope.get("schema") != "com.cezen.nexus-one.signature-envelope.v1" \
|
|
or envelope.get("algorithm") != "ed25519":
|
|
raise VerificationError("ENVELOPE_INVALID")
|
|
if envelope.get("manifest_digest") != f"sha256:{manifest_digest}":
|
|
raise VerificationError("MANIFEST_DIGEST_MISMATCH")
|
|
if envelope.get("policy_version") != trust.get("policy_version"):
|
|
raise VerificationError("POLICY_VERSION_MISMATCH")
|
|
keys = [item for item in trust.get("keys", []) if item.get("key_id") == envelope.get("key_id")]
|
|
if len(keys) != 1:
|
|
raise VerificationError("KEY_UNKNOWN_OR_AMBIGUOUS")
|
|
key = keys[0]
|
|
required = {"algorithm": "ed25519", "purpose": "profile-membership", "status": "active",
|
|
"environment": environment}
|
|
if any(key.get(name) != value for name, value in required.items()):
|
|
raise VerificationError("KEY_POLICY_REJECTED")
|
|
der = subprocess.run(
|
|
["openssl", "pkey", "-pubin", "-in", str(public_key), "-outform", "DER"],
|
|
check=True, capture_output=True,
|
|
).stdout
|
|
fingerprint = sha256(der)
|
|
if fingerprint != key.get("public_key_sha256"):
|
|
raise VerificationError("KEY_FINGERPRINT_MISMATCH")
|
|
try:
|
|
signature = base64.b64decode(envelope.get("signature", ""), validate=True)
|
|
except ValueError as error:
|
|
raise VerificationError("SIGNATURE_ENCODING_INVALID") from error
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
body_path = Path(directory) / "manifest.canonical.json"
|
|
signature_path = Path(directory) / "signature.bin"
|
|
body_path.write_bytes(body)
|
|
signature_path.write_bytes(signature)
|
|
result = subprocess.run([
|
|
"openssl", "pkeyutl", "-verify", "-pubin", "-inkey", str(public_key),
|
|
"-rawin", "-in", str(body_path), "-sigfile", str(signature_path),
|
|
], capture_output=True)
|
|
if result.returncode != 0:
|
|
raise VerificationError("SIGNATURE_INVALID")
|
|
return {
|
|
"schema": "com.cezen.nexus-one.profile-membership-verification-receipt.v1",
|
|
"environment": environment,
|
|
"result": "verified",
|
|
"manifest_sha256": manifest_digest,
|
|
"envelope_sha256": sha256(canonical(envelope)),
|
|
"key_id": envelope["key_id"],
|
|
"key_fingerprint_sha256": fingerprint,
|
|
"trust_snapshot_sha256": sha256(canonical(trust)),
|
|
"policy_version": trust["policy_version"],
|
|
"namespace": manifest["namespace"],
|
|
"profiles": sorted(manifest["profiles"]),
|
|
"verified_at": verified_at,
|
|
"verifier_build_sha256": verifier_build_sha256,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--manifest", type=Path, required=True)
|
|
parser.add_argument("--envelope", type=Path, required=True)
|
|
parser.add_argument("--trust-snapshot", type=Path, required=True)
|
|
parser.add_argument("--public-key", type=Path, required=True)
|
|
parser.add_argument("--environment", choices=("production", "test"), required=True)
|
|
parser.add_argument("--verified-at", required=True)
|
|
args = parser.parse_args()
|
|
script_digest = sha256(Path(__file__).read_bytes())
|
|
try:
|
|
receipt = verify(
|
|
json.loads(args.manifest.read_text()), json.loads(args.envelope.read_text()),
|
|
json.loads(args.trust_snapshot.read_text()), args.public_key,
|
|
args.environment, args.verified_at, script_digest,
|
|
)
|
|
except (VerificationError, subprocess.CalledProcessError, json.JSONDecodeError) as error:
|
|
print(json.dumps({"result": "rejected", "code": str(error)}, sort_keys=True))
|
|
return 1
|
|
print(json.dumps(receipt, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|