ADR-004: Hybrid Post-Quantum Encryption (X25519 + ML-KEM-768)¶

Current Risk: LOW (no CRQCs exist today)
Future Risk: HIGH (15-30 year horizon per NIST estimates)
Attack Vector: "Harvest now, decrypt later"
Impact: Encryption key recovery → confidentiality breach

┌─────────────────────────────────────────────┐
│         HybridKEMProvider                   │
├─────────────────────────────────────────────┤
│  algorithm_name: "x25519+ml-kem-768+aes-256-gcm"│
│                                              │
│  PUBLIC_KEY_SIZE  = 1,216 bytes              │
│    └─ X25519:      32 bytes                  │
│    └─ ML-KEM-768: 1,184 bytes                │
│                                              │
│  PRIVATE_KEY_SIZE = 2,432 bytes              │
│    └─ X25519:      32 bytes                  │
│    └─ ML-KEM-768: 2,400 bytes                │
│                                              │
│  ENCRYPTED_OVERHEAD = 1,148 bytes            │
│    └─ Ephemeral X25519: 32 bytes             │
│    └─ ML-KEM ciphertext: 1,088 bytes         │
│    └─ AES-GCM nonce: 12 bytes                │
│    └─ AES-GCM tag: 16 bytes                  │
└─────────────────────────────────────────────┘

Encrypt(plaintext, recipient_public_key):
1. Generate ephemeral X25519 keypair
2. X25519 DH → classical_secret (32 bytes)
3. ML-KEM encapsulate → (ciphertext, pq_secret) (32 bytes)
4. combined_secret = HKDF(classical_secret || pq_secret, info="AEGIS-HybridKEM-v1")
5. AES-256-GCM encrypt with random 12-byte nonce
6. Return HybridEncryptedBlob

Decrypt(blob, recipient_private_key):
1. X25519 DH with ephemeral key → classical_secret
2. ML-KEM decapsulate → pq_secret
3. combined_secret = HKDF(classical_secret || pq_secret, info="AEGIS-HybridKEM-v1")
4. AES-256-GCM decrypt and verify tag
5. Return plaintext

# src/crypto/__init__.py
def get_hybrid_kem_provider() -> HybridKEMProvider:
    """
    Get the hybrid post-quantum encryption provider.

    Returns HybridKEMProvider for X25519 + ML-KEM-768 + AES-256-GCM encryption.
    """
    if HYBRID_KEM_AVAILABLE and HybridKEMProvider is not None:
        return HybridKEMProvider()
    else:
        raise RuntimeError(
            "HybridKEMProvider not available. "
            "Install liboqs-python>=0.10.0 for post-quantum support."
        )

from crypto import get_hybrid_kem_provider

provider = get_hybrid_kem_provider()

# Generate keypair
keypair = provider.generate_keypair()

# Encrypt sensitive data
plaintext = b"governance private key material"
blob = provider.encrypt(plaintext, keypair.public)

# Decrypt
decrypted = provider.decrypt(blob, keypair.private)
assert decrypted == plaintext

Phase 1 (Core Primitives):
  tests/crypto/test_mlkem.py: 30 tests
  tests/crypto/test_hybrid_kem.py: 40 tests

Phase 2 (Key Store & PII):
  tests/crypto/test_kek_provider.py: 27 tests
  tests/telemetry/test_pii_encryption.py: 31 tests
  tests/workflows/persistence/test_key_store.py: 52 tests

Coverage Expansion (2025-12-29):
  tests/test_override_coverage.py: 99 tests (override.py: 61.27%→95.19%)
  tests/test_persistence_coverage.py: 36 tests (durable.py: 73.49%→100%)
  tests/telemetry/test_coverage.py: 64 tests (decryption.py: 74.10%→94.58%)

Total: 846 tests | 93.60% overall coverage
KeyStoreRepository: 96.52% coverage
All quality gates pass (mypy --strict, ruff, bandit)

from crypto.kek_provider import get_kek_provider, InMemoryKEKProvider

# Auto-detect: uses environment if AEGIS_MASTER_KEK_PUBLIC/PRIVATE set
provider = get_kek_provider("auto")

# For testing
provider = InMemoryKEKProvider()

# Encrypt/decrypt
encrypted = provider.encrypt(b"sensitive data")
decrypted = provider.decrypt(encrypted)

from workflows.persistence import KeyStoreRepository

repo = KeyStoreRepository(session, kek_provider)

# Store governance key (private key encrypted with KEK)
key_id = await repo.store_key(
    role="risk_lead",
    algorithm="bip322-simple",
    private_key=private_key_bytes,
    public_key=public_key_bytes,
    created_by="system"
)

# Retrieve for signing (decrypted)
private_key = await repo.get_private_key("risk_lead", "bip322-simple")

# Audit trail
await repo.record_usage("risk_lead", "bip322-simple", "sign", actor_id, context)
audit_log = await repo.get_audit_trail("risk_lead", "bip322-simple")

from telemetry.encryption import PIIEncryptionEnricher, PIIManifest, PIIPriority
from telemetry.decryption import PIIDecryptor

# Create encryptor
encryptor = PIIEncryptionEnricher(
    public_key=kek_provider.get_public_key(),
    dek_version="dek-v1-20251228",
    min_priority=PIIPriority.HIGH  # Encrypt HIGH and CRITICAL fields
)

# Encrypt PII in telemetry event
event = {
    "event_id": "evt-123",
    "actor_id": "user@acme-corp.test",  # HIGH priority - encrypted
    "decision_actor": "lead@acme-corp.test",  # CRITICAL - encrypted
    "proposal_id": "prop-001"  # Not PII - remains plaintext
}
encrypted_event = encryptor.enrich(event)

# Decrypt
decryptor = PIIDecryptor(
    private_key=kek_provider.get_private_key(),
    verify_integrity=True
)
decrypted_event = decryptor.decrypt_event(encrypted_event)

from workflows.override import OverrideWorkflow
from workflows.persistence import KeyStoreRepository

workflow = OverrideWorkflow(proposal_id="prop-001", reason="Critical fix")
workflow.set_key_store(key_store_repo)

# Sign with stored key (retrieves, signs, records audit)
success, message = await workflow.sign_with_stored_key(
    role="risk_lead",
    actor_id="alice@acme-corp.test"
)