"""Offline PKI bootstrap for one explicitly paired OMNI Windows node. The optional ``cryptography`` package is imported only after the operator runs this module. Nothing is downloaded and existing files are never overwritten. """ from __future__ import annotations import argparse from datetime import datetime, timedelta, timezone import hashlib import ipaddress import json import os from pathlib import Path import re from typing import Any def short_authentication_string(node_fingerprint: str, client_fingerprint: str) -> str: digest = hashlib.sha256( bytes.fromhex(node_fingerprint) + bytes.fromhex(client_fingerprint) ).hexdigest()[:16].upper() return "-".join(digest[index : index + 4] for index in range(0, 16, 4)) def generate_pki( output_dir: Path, *, node_id: str, conductor_id: str, lan_ip: str, ) -> dict[str, Any]: node_id = _validate_id(node_id, "node_id") conductor_id = _validate_id(conductor_id, "conductor_id") address = _validate_lan_ip(lan_ip) root = Path(output_dir).expanduser().resolve() if root.exists() and any(root.iterdir()): raise FileExistsError("output directory must be new or empty; existing PKI is never overwritten") root.mkdir(parents=True, exist_ok=True) try: from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID except ImportError as exc: raise RuntimeError("offline PKI generation requires the optional cryptography package") from exc now = datetime.now(timezone.utc) not_before = now - timedelta(minutes=5) not_after = now + timedelta(days=3650) ca_key = ec.generate_private_key(ec.SECP384R1()) ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "OMNI Local Pairing CA")]) ca_cert = ( x509.CertificateBuilder() .subject_name(ca_name) .issuer_name(ca_name) .public_key(ca_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(not_before) .not_valid_after(not_after) .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, content_commitment=False, key_encipherment=False, data_encipherment=False, key_agreement=False, key_cert_sign=True, crl_sign=True, encipher_only=False, decipher_only=False, ), critical=True, ) .sign(ca_key, hashes.SHA384()) ) node_key = ec.generate_private_key(ec.SECP384R1()) node_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, node_id)]) node_cert = ( x509.CertificateBuilder() .subject_name(node_name) .issuer_name(ca_cert.subject) .public_key(node_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(not_before) .not_valid_after(not_after) .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) .add_extension( x509.SubjectAlternativeName( [x509.IPAddress(address), x509.IPAddress(ipaddress.ip_address("127.0.0.1"))] ), critical=False, ) .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, content_commitment=False, key_encipherment=False, data_encipherment=False, key_agreement=True, key_cert_sign=False, crl_sign=False, encipher_only=False, decipher_only=False, ), critical=True, ) .sign(ca_key, hashes.SHA384()) ) client_key = ec.generate_private_key(ec.SECP384R1()) client_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, conductor_id)]) client_cert = ( x509.CertificateBuilder() .subject_name(client_name) .issuer_name(ca_cert.subject) .public_key(client_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(not_before) .not_valid_after(not_after) .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) .add_extension(x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), critical=True) .add_extension( x509.KeyUsage( digital_signature=True, content_commitment=False, key_encipherment=False, data_encipherment=False, key_agreement=True, key_cert_sign=False, crl_sign=False, encipher_only=False, decipher_only=False, ), critical=True, ) .sign(ca_key, hashes.SHA384()) ) private_format = serialization.PrivateFormat.PKCS8 private_encoding = serialization.Encoding.PEM no_encryption = serialization.NoEncryption() certificate_encoding = serialization.Encoding.PEM files = { "ca.pem": ca_cert.public_bytes(certificate_encoding), "ca-key.pem": ca_key.private_bytes(private_encoding, private_format, no_encryption), "node.pem": node_cert.public_bytes(certificate_encoding), "node-key.pem": node_key.private_bytes(private_encoding, private_format, no_encryption), "conductor.pem": client_cert.public_bytes(certificate_encoding), "conductor-key.pem": client_key.private_bytes( private_encoding, private_format, no_encryption, ), "paired-nodes.json": b'{"version":1,"nodes":[]}\n', } for name, content in files.items(): _write_exclusive(root / name, content, private=name.endswith("-key.pem")) node_fingerprint = node_cert.fingerprint(hashes.SHA256()).hex() client_fingerprint = client_cert.fingerprint(hashes.SHA256()).hex() summary = { "node_id": node_id, "conductor_id": conductor_id, "lan_ip": str(address), "node_certificate_fingerprint_sha256": node_fingerprint, "conductor_certificate_fingerprint_sha256": client_fingerprint, "sas": short_authentication_string(node_fingerprint, client_fingerprint), "node_files": ["ca.pem", "node.pem", "node-key.pem", "paired-nodes.json"], "conductor_files": ["ca.pem", "conductor.pem", "conductor-key.pem"], "private_keys_unencrypted": True, "network_activity": False, } _write_exclusive( root / "pairing-info.json", (json.dumps(summary, ensure_ascii=False, indent=2) + "\n").encode("utf-8"), private=False, ) return summary def _write_exclusive(path: Path, content: bytes, *, private: bool) -> None: descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600 if private else 0o644) try: os.write(descriptor, content) os.fsync(descriptor) finally: os.close(descriptor) def _validate_id(value: str, label: str) -> str: if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}", value): raise ValueError(f"{label} is invalid") return value def _validate_lan_ip(value: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address: address = ipaddress.ip_address(value) if ( not address.is_private or address.is_loopback or address.is_unspecified or address.is_link_local or address.is_multicast or address.is_reserved ): raise ValueError("lan_ip must be a concrete private non-loopback address") return address def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Generate an offline local OMNI pairing PKI") parser.add_argument("--output", required=True, type=Path) parser.add_argument("--node-id", required=True) parser.add_argument("--conductor-id", required=True) parser.add_argument("--lan-ip", required=True) return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) try: summary = generate_pki( args.output, node_id=args.node_id, conductor_id=args.conductor_id, lan_ip=args.lan_ip, ) except (FileExistsError, RuntimeError, ValueError) as exc: _parser().error(str(exc)) return 2 print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())