#!/usr/bin/env python3 """Create and open customer-controlled Clavenar secure-exchange envelopes. Requires Python 3.11+ and cryptography 42+. The tool performs no network I/O. """ from __future__ import annotations import argparse import base64 from datetime import datetime, timedelta, timezone import hashlib import io import json import os from pathlib import Path, PurePosixPath import secrets import sys import tarfile from typing import Any from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.x25519 import ( X25519PrivateKey, X25519PublicKey, ) from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives import hashes CONTRACT = "clavenar.customer-secure-exchange/v1" RECIPIENT_CONTRACT = "clavenar.customer-secure-exchange-recipient/v1" PRIVATE_CONTRACT = "clavenar.customer-secure-exchange-private-key/v1" MAX_FILES = 128 MAX_FILE_BYTES = 8 * 1024 * 1024 MAX_PLAINTEXT_BYTES = 16 * 1024 * 1024 MAX_LIFETIME_HOURS = 168 class ExchangeError(ValueError): """An envelope or local exchange input violates the governed boundary.""" def canonical_json(value: Any) -> bytes: return json.dumps( value, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def b64encode(value: bytes) -> str: return base64.b64encode(value).decode("ascii") def b64decode(value: Any, label: str, *, length: int | None = None) -> bytes: if not isinstance(value, str): raise ExchangeError(f"{label} must be base64 text") try: decoded = base64.b64decode(value, validate=True) except (ValueError, base64.binascii.Error) as error: raise ExchangeError(f"{label} is not canonical base64") from error if b64encode(decoded) != value: raise ExchangeError(f"{label} is not canonical base64") if length is not None and len(decoded) != length: raise ExchangeError(f"{label} must decode to {length} bytes") return decoded def sha256(value: bytes) -> str: return f"sha256:{hashlib.sha256(value).hexdigest()}" def utc_now() -> datetime: return datetime.now(timezone.utc).replace(microsecond=0) def format_time(value: datetime) -> str: return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def parse_time(value: Any, label: str) -> datetime: if not isinstance(value, str): raise ExchangeError(f"{label} must be an RFC 3339 UTC timestamp") try: parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") except ValueError as error: raise ExchangeError( f"{label} must be an RFC 3339 second-precision UTC timestamp" ) from error return parsed.replace(tzinfo=timezone.utc) def public_bytes(key: X25519PublicKey) -> bytes: return key.public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) def private_bytes(key: X25519PrivateKey) -> bytes: return key.private_bytes( encoding=serialization.Encoding.Raw, format=serialization.PrivateFormat.Raw, encryption_algorithm=serialization.NoEncryption(), ) def key_id(raw_public: bytes) -> str: return sha256(raw_public) def write_private_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(descriptor, "wb") as stream: stream.write(json.dumps(value, indent=2, sort_keys=True).encode("utf-8")) stream.write(b"\n") except Exception: path.unlink(missing_ok=True) raise def write_public_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("x", encoding="utf-8") as stream: json.dump(value, stream, indent=2, sort_keys=True) stream.write("\n") def generate_keypair(private_path: Path, public_path: Path) -> str: if private_path.resolve() == public_path.resolve(): raise ExchangeError("private and public key paths must differ") private = X25519PrivateKey.generate() raw_private = private_bytes(private) raw_public = public_bytes(private.public_key()) identifier = key_id(raw_public) write_private_json( private_path, { "algorithm": "X25519", "contract": PRIVATE_CONTRACT, "keyId": identifier, "privateKey": b64encode(raw_private), "schemaVersion": 1, }, ) write_public_json( public_path, { "algorithm": "X25519", "contract": RECIPIENT_CONTRACT, "keyId": identifier, "publicKey": b64encode(raw_public), "schemaVersion": 1, }, ) return identifier def read_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise ExchangeError(f"cannot read {label}: {path}") from error if not isinstance(value, dict): raise ExchangeError(f"{label} must be one JSON object") return value def load_recipient(path: Path) -> tuple[str, X25519PublicKey]: value = read_json(path, "recipient") if ( value.get("contract") != RECIPIENT_CONTRACT or value.get("schemaVersion") != 1 or value.get("algorithm") != "X25519" ): raise ExchangeError(f"invalid recipient contract: {path}") raw = b64decode(value.get("publicKey"), "recipient public key", length=32) identifier = key_id(raw) if value.get("keyId") != identifier: raise ExchangeError(f"recipient key commitment mismatch: {path}") return identifier, X25519PublicKey.from_public_bytes(raw) def load_private(path: Path) -> tuple[str, X25519PrivateKey]: value = read_json(path, "private key") if ( value.get("contract") != PRIVATE_CONTRACT or value.get("schemaVersion") != 1 or value.get("algorithm") != "X25519" ): raise ExchangeError("invalid private-key contract") raw = b64decode(value.get("privateKey"), "private key", length=32) private = X25519PrivateKey.from_private_bytes(raw) identifier = key_id(public_bytes(private.public_key())) if value.get("keyId") != identifier: raise ExchangeError("private-key commitment mismatch") return identifier, private def collect_files(root: Path) -> list[tuple[PurePosixPath, bytes]]: if not root.is_dir() or root.is_symlink(): raise ExchangeError("input must be one real directory") collected: list[tuple[PurePosixPath, bytes]] = [] total = 0 for candidate in sorted(root.rglob("*")): relative = PurePosixPath(candidate.relative_to(root).as_posix()) if candidate.is_symlink(): raise ExchangeError(f"symlinks are forbidden: {relative}") if candidate.is_dir(): continue if not candidate.is_file() or relative.is_absolute() or ".." in relative.parts: raise ExchangeError(f"unsafe input path: {relative}") payload = candidate.read_bytes() if len(payload) > MAX_FILE_BYTES: raise ExchangeError(f"file exceeds {MAX_FILE_BYTES} bytes: {relative}") total += len(payload) if total > MAX_PLAINTEXT_BYTES: raise ExchangeError( f"plaintext exceeds {MAX_PLAINTEXT_BYTES} aggregate bytes" ) collected.append((relative, payload)) if len(collected) > MAX_FILES: raise ExchangeError(f"bundle exceeds {MAX_FILES} files") if not collected: raise ExchangeError("input directory must contain at least one file") return collected def build_archive( files: list[tuple[PurePosixPath, bytes]], ) -> tuple[bytes, dict[str, Any]]: manifest_files = [ {"path": str(path), "sha256": sha256(payload), "size": len(payload)} for path, payload in files ] buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w") as archive: for path, payload in files: info = tarfile.TarInfo(str(path)) info.size = len(payload) info.mode = 0o600 info.mtime = 0 info.uid = 0 info.gid = 0 info.uname = "" info.gname = "" archive.addfile(info, io.BytesIO(payload)) manifest = { "fileCount": len(files), "files": manifest_files, "plaintextBytes": sum(len(payload) for _, payload in files), } return buffer.getvalue(), manifest def derive_wrap_key( shared_secret: bytes, bundle_id: str, role: str, recipient_key_id: str, ) -> bytes: return HKDF( algorithm=hashes.SHA256(), length=32, salt=hashlib.sha256(bundle_id.encode("ascii")).digest(), info=( f"{CONTRACT}:{role}:{recipient_key_id}" ).encode("ascii"), ).derive(shared_secret) def wrap_data_key( data_key: bytes, public_key: X25519PublicKey, *, bundle_id: str, created_at: str, expires_at: str, role: str, recipient_key_id: str, ) -> dict[str, Any]: ephemeral = X25519PrivateKey.generate() ephemeral_public = public_bytes(ephemeral.public_key()) shared = ephemeral.exchange(public_key) wrap_key = derive_wrap_key(shared, bundle_id, role, recipient_key_id) nonce = secrets.token_bytes(12) aad = canonical_json( { "bundleId": bundle_id, "contract": CONTRACT, "createdAt": created_at, "expiresAt": expires_at, "keyId": recipient_key_id, "role": role, } ) return { "algorithm": "X25519-HKDF-SHA256+A256GCM", "ephemeralPublicKey": b64encode(ephemeral_public), "keyId": recipient_key_id, "nonce": b64encode(nonce), "role": role, "wrappedKey": b64encode(AESGCM(wrap_key).encrypt(nonce, data_key, aad)), } def envelope_header(envelope: dict[str, Any]) -> dict[str, Any]: return { "bundleId": envelope["bundleId"], "contract": envelope["contract"], "createdAt": envelope["createdAt"], "expiresAt": envelope["expiresAt"], "manifest": envelope["manifest"], "recipients": envelope["recipients"], "schemaVersion": envelope["schemaVersion"], } def create_envelope( input_root: Path, customer_recipient: Path, clavenar_recipient: Path, output_path: Path, lifetime_hours: int, ) -> dict[str, Any]: if lifetime_hours < 1 or lifetime_hours > MAX_LIFETIME_HOURS: raise ExchangeError( f"lifetime must be between 1 and {MAX_LIFETIME_HOURS} hours" ) customer_id, customer_key = load_recipient(customer_recipient) clavenar_id, clavenar_key = load_recipient(clavenar_recipient) if customer_id == clavenar_id: raise ExchangeError("customer and Clavenar recipients must be distinct") files = collect_files(input_root) plaintext, manifest = build_archive(files) created = utc_now() created_at = format_time(created) expires_at = format_time(created + timedelta(hours=lifetime_hours)) bundle_id = f"cx-{secrets.token_hex(16)}" data_key = secrets.token_bytes(32) recipients = [ wrap_data_key( data_key, customer_key, bundle_id=bundle_id, created_at=created_at, expires_at=expires_at, role="customer", recipient_key_id=customer_id, ), wrap_data_key( data_key, clavenar_key, bundle_id=bundle_id, created_at=created_at, expires_at=expires_at, role="clavenar", recipient_key_id=clavenar_id, ), ] envelope: dict[str, Any] = { "bundleId": bundle_id, "contract": CONTRACT, "createdAt": created_at, "expiresAt": expires_at, "manifest": manifest, "recipients": recipients, "schemaVersion": 1, } nonce = secrets.token_bytes(12) ciphertext = AESGCM(data_key).encrypt( nonce, plaintext, canonical_json(envelope_header(envelope)), ) envelope["payload"] = { "algorithm": "A256GCM", "ciphertext": b64encode(ciphertext), "nonce": b64encode(nonce), "sha256": sha256(ciphertext), } output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("x", encoding="utf-8") as stream: json.dump(envelope, stream, indent=2, sort_keys=True) stream.write("\n") return envelope def validate_envelope( value: dict[str, Any], *, now: datetime | None = None, allow_expired: bool = False, ) -> dict[str, Any]: if ( value.get("contract") != CONTRACT or value.get("schemaVersion") != 1 or not isinstance(value.get("bundleId"), str) or not value["bundleId"].startswith("cx-") ): raise ExchangeError("invalid envelope identity") created = parse_time(value.get("createdAt"), "createdAt") expires = parse_time(value.get("expiresAt"), "expiresAt") lifetime = expires - created if lifetime <= timedelta(0) or lifetime > timedelta(hours=MAX_LIFETIME_HOURS): raise ExchangeError("envelope lifetime is invalid") if not allow_expired and expires <= (now or utc_now()): raise ExchangeError("envelope is expired") manifest = value.get("manifest") files = manifest.get("files") if isinstance(manifest, dict) else None if ( not isinstance(files, list) or manifest.get("fileCount") != len(files) or not 1 <= len(files) <= MAX_FILES or not isinstance(manifest.get("plaintextBytes"), int) or not 0 <= manifest["plaintextBytes"] <= MAX_PLAINTEXT_BYTES ): raise ExchangeError("invalid bounded manifest") seen_paths: set[str] = set() manifest_bytes = 0 for item in files: path = PurePosixPath(item.get("path", "")) if ( not str(path) or path.is_absolute() or ".." in path.parts or str(path) in seen_paths or not isinstance(item.get("size"), int) or not 0 <= item["size"] <= MAX_FILE_BYTES or not isinstance(item.get("sha256"), str) or not item["sha256"].startswith("sha256:") or len(item["sha256"]) != 71 ): raise ExchangeError("manifest contains an unsafe file entry") seen_paths.add(str(path)) manifest_bytes += item["size"] if manifest_bytes != manifest["plaintextBytes"]: raise ExchangeError("manifest byte total is inconsistent") recipients = value.get("recipients") if ( not isinstance(recipients, list) or [entry.get("role") for entry in recipients] != ["customer", "clavenar"] or len({entry.get("keyId") for entry in recipients}) != 2 ): raise ExchangeError("envelope requires distinct customer and Clavenar recipients") for recipient in recipients: if ( recipient.get("algorithm") != "X25519-HKDF-SHA256+A256GCM" or not isinstance(recipient.get("keyId"), str) or not recipient["keyId"].startswith("sha256:") or len(recipient["keyId"]) != 71 ): raise ExchangeError("invalid recipient entry") b64decode( recipient.get("ephemeralPublicKey"), "ephemeral public key", length=32, ) b64decode(recipient.get("nonce"), "recipient nonce", length=12) b64decode(recipient.get("wrappedKey"), "wrapped key", length=48) payload = value.get("payload") if not isinstance(payload, dict) or payload.get("algorithm") != "A256GCM": raise ExchangeError("invalid encrypted payload") nonce = b64decode(payload.get("nonce"), "payload nonce", length=12) ciphertext = b64decode(payload.get("ciphertext"), "payload ciphertext") if len(ciphertext) < 16 or payload.get("sha256") != sha256(ciphertext): raise ExchangeError("encrypted payload commitment mismatch") return { "bundleId": value["bundleId"], "createdAt": format_time(created), "expiresAt": format_time(expires), "fileCount": len(files), "plaintextBytes": manifest["plaintextBytes"], "recipients": [entry["role"] for entry in recipients], "payloadBytes": len(ciphertext), } def unwrap_data_key( envelope: dict[str, Any], recipient: dict[str, Any], private: X25519PrivateKey, ) -> bytes: ephemeral_raw = b64decode( recipient["ephemeralPublicKey"], "ephemeral public key", length=32, ) shared = private.exchange(X25519PublicKey.from_public_bytes(ephemeral_raw)) wrap_key = derive_wrap_key( shared, envelope["bundleId"], recipient["role"], recipient["keyId"], ) aad = canonical_json( { "bundleId": envelope["bundleId"], "contract": CONTRACT, "createdAt": envelope["createdAt"], "expiresAt": envelope["expiresAt"], "keyId": recipient["keyId"], "role": recipient["role"], } ) try: return AESGCM(wrap_key).decrypt( b64decode(recipient["nonce"], "recipient nonce", length=12), b64decode(recipient["wrappedKey"], "wrapped key", length=48), aad, ) except Exception as error: raise ExchangeError("recipient key cannot authenticate the envelope") from error def verify_archive(plaintext: bytes, manifest: dict[str, Any]) -> list[tuple[str, bytes]]: expected = {item["path"]: item for item in manifest["files"]} actual: list[tuple[str, bytes]] = [] try: with tarfile.open(fileobj=io.BytesIO(plaintext), mode="r:") as archive: for member in archive.getmembers(): if ( not member.isfile() or member.issym() or member.islnk() or PurePosixPath(member.name).is_absolute() or ".." in PurePosixPath(member.name).parts or member.name not in expected ): raise ExchangeError("decrypted archive contains an unsafe entry") stream = archive.extractfile(member) if stream is None: raise ExchangeError("decrypted archive file is unreadable") payload = stream.read(MAX_FILE_BYTES + 1) expectation = expected[member.name] if ( len(payload) != expectation["size"] or sha256(payload) != expectation["sha256"] ): raise ExchangeError("decrypted file does not match the manifest") actual.append((member.name, payload)) except (tarfile.TarError, OSError) as error: raise ExchangeError("decrypted payload is not a safe archive") from error if [name for name, _ in actual] != list(expected): raise ExchangeError("decrypted archive inventory does not match the manifest") return actual def open_envelope( envelope_path: Path, private_path: Path, output_root: Path, ) -> dict[str, Any]: envelope = read_json(envelope_path, "envelope") summary = validate_envelope(envelope) identifier, private = load_private(private_path) matching = [ recipient for recipient in envelope["recipients"] if recipient["keyId"] == identifier ] if len(matching) != 1: raise ExchangeError("private key is not an envelope recipient") data_key = unwrap_data_key(envelope, matching[0], private) payload = envelope["payload"] try: plaintext = AESGCM(data_key).decrypt( b64decode(payload["nonce"], "payload nonce", length=12), b64decode(payload["ciphertext"], "payload ciphertext"), canonical_json(envelope_header(envelope)), ) except Exception as error: raise ExchangeError("encrypted payload authentication failed") from error files = verify_archive(plaintext, envelope["manifest"]) output_root.mkdir(parents=True, exist_ok=False, mode=0o700) for name, content in files: destination = output_root.joinpath(*PurePosixPath(name).parts) destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700) descriptor = os.open( destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, ) with os.fdopen(descriptor, "wb") as stream: stream.write(content) summary["recipientRole"] = matching[0]["role"] summary["output"] = str(output_root) return summary def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Local-only Clavenar customer secure exchange", ) commands = parser.add_subparsers(dest="command", required=True) keygen = commands.add_parser("keygen", help="create an X25519 keypair") keygen.add_argument("--private", required=True, type=Path) keygen.add_argument("--public", required=True, type=Path) pack = commands.add_parser("pack", help="encrypt one bounded directory") pack.add_argument("--input", required=True, type=Path) pack.add_argument("--customer-recipient", required=True, type=Path) pack.add_argument("--clavenar-recipient", required=True, type=Path) pack.add_argument("--output", required=True, type=Path) pack.add_argument("--lifetime-hours", type=int, default=72) inspect = commands.add_parser("inspect", help="validate public envelope metadata") inspect.add_argument("--input", required=True, type=Path) unpack = commands.add_parser("unpack", help="authenticate and decrypt an envelope") unpack.add_argument("--input", required=True, type=Path) unpack.add_argument("--private", required=True, type=Path) unpack.add_argument("--output", required=True, type=Path) return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) if args.command == "keygen": result = { "keyId": generate_keypair(args.private, args.public), "private": str(args.private), "public": str(args.public), "status": "created", } elif args.command == "pack": envelope = create_envelope( args.input, args.customer_recipient, args.clavenar_recipient, args.output, args.lifetime_hours, ) result = validate_envelope(envelope) result["output"] = str(args.output) result["status"] = "encrypted" elif args.command == "inspect": result = validate_envelope(read_json(args.input, "envelope")) result["status"] = "valid" else: result = open_envelope(args.input, args.private, args.output) result["status"] = "decrypted" print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except ExchangeError as error: print(f"secure-exchange: {error}", file=sys.stderr) raise SystemExit(2)