feat: v0.12.0 — Post-quantum crypto + government hardening

CHANGELOG.md

@@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.12.0] - 2026-04-06
+
+### Added
+- **Post-quantum cryptography (delivered)**:
+  - `MLKEMKeyManager` — ML-KEM-768 key encapsulation (FIPS 203)
+  - `MLDSASigner` — ML-DSA-65 digital signatures (FIPS 204)
+  - `HybridEncryptor` — ML-KEM-768 + AES-256-GCM hybrid encryption
+  - `[pq]` installation extra: `pip install qp-vault[pq]`
+- **Input bounds**: `top_k` capped at 1000, `threshold` range 0-1, query max 10K chars
+- **Batch limits**: max 100 items per `/batch` request
+- **Plugin hash verification**: `manifest.json` with SHA3-256 hashes in plugins_dir
+- **Tenant-locked vault**: `Vault(path, tenant_id="x")` enforces single-tenant scope
+
+### Security
+- SearchRequest Pydantic validators prevent unbounded parameter attacks
+- Plugin files verified against manifest before execution
+
 ## [0.11.0] - 2026-04-06
 
 ### Added

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "qp-vault"
-version = "0.11.0"
+version = "0.12.0"
 description = "Governed knowledge store for autonomous organizations. Trust tiers, cryptographic audit trails, content-addressed storage, air-gap native."
 readme = "README.md"
 license = "Apache-2.0"
@@ -68,6 +68,9 @@ encryption = [
     "cryptography>=42",
     "pynacl>=1.6.2",
 ]
+pq = [
+    "liboqs-python>=0.14.1",
+]
 integrity = [
     "numpy>=2.0",
 ]
@@ -87,7 +90,7 @@ dev = [
     "ruff>=0.9",
 ]
 all = [
-    "qp-vault[sqlite,postgres,docling,local,openai,capsule,encryption,integrity,fastapi,cli]",
+    "qp-vault[sqlite,postgres,docling,local,openai,capsule,encryption,pq,integrity,fastapi,cli]",
 ]
 
 [project.scripts]

src/qp_vault/__init__.py

@@ -26,7 +26,7 @@
 Docs: https://github.com/quantumpipes/vault
 """
 
-__version__ = "0.11.0"
+__version__ = "0.12.0"
 __author__ = "Quantum Pipes Technologies, LLC"
 __license__ = "Apache-2.0"
 

src/qp_vault/encryption/__init__.py

@@ -3,10 +3,26 @@
 
 """Encryption at rest for qp-vault.
 
-Provides AES-256-GCM symmetric encryption for chunk content.
-Requires: pip install qp-vault[encryption]
+Classical: AES-256-GCM (FIPS 197)
+Post-quantum: ML-KEM-768 key encapsulation (FIPS 203) + ML-DSA-65 signatures (FIPS 204)
+Hybrid: ML-KEM-768 + AES-256-GCM (quantum-resistant data encryption)
+
+Install:
+    pip install qp-vault[encryption]     # AES-256-GCM only
+    pip install qp-vault[pq]             # + ML-KEM-768 + ML-DSA-65
+    pip install qp-vault[encryption,pq]  # Full hybrid encryption
 """
 
 from qp_vault.encryption.aes_gcm import AESGCMEncryptor
 
 __all__ = ["AESGCMEncryptor"]
+
+# Conditional PQ exports (available when liboqs-python installed)
+try:
+    from qp_vault.encryption.hybrid import HybridEncryptor
+    from qp_vault.encryption.ml_dsa import MLDSASigner
+    from qp_vault.encryption.ml_kem import MLKEMKeyManager
+
+    __all__ += ["MLKEMKeyManager", "MLDSASigner", "HybridEncryptor"]  # type: ignore[assignment]
+except ImportError:
+    pass

src/qp_vault/encryption/hybrid.py

@@ -0,0 +1,94 @@
+# Copyright 2026 Quantum Pipes Technologies, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""Hybrid encryption: ML-KEM-768 key exchange + AES-256-GCM data encryption.
+
+Combines post-quantum key encapsulation (FIPS 203) with classical
+symmetric encryption (FIPS 197) for defense-in-depth:
+
+1. ML-KEM-768 encapsulates a shared secret (32 bytes)
+2. Shared secret is used as AES-256-GCM key
+3. Data is encrypted with AES-256-GCM
+
+Format: kem_ciphertext_len (4 bytes) || kem_ciphertext || aes_nonce (12) || aes_ciphertext || aes_tag (16)
+
+Requires: pip install qp-vault[pq,encryption]
+"""
+
+from __future__ import annotations
+
+import struct
+
+from qp_vault.encryption.aes_gcm import AESGCMEncryptor
+from qp_vault.encryption.ml_kem import MLKEMKeyManager
+
+
+class HybridEncryptor:
+    """ML-KEM-768 + AES-256-GCM hybrid encryption.
+
+    Provides quantum-resistant data encryption by wrapping AES keys
+    with ML-KEM-768 key encapsulation.
+
+    Usage:
+        enc = HybridEncryptor()
+        pub, sec = enc.generate_keypair()
+        ciphertext = enc.encrypt(b"secret data", pub)
+        plaintext = enc.decrypt(ciphertext, sec)
+    """
+
+    def __init__(self) -> None:
+        self._kem = MLKEMKeyManager()
+
+    def generate_keypair(self) -> tuple[bytes, bytes]:
+        """Generate an ML-KEM-768 keypair for hybrid encryption.
+
+        Returns:
+            (public_key, secret_key) — store secret_key securely.
+        """
+        return self._kem.generate_keypair()
+
+    def encrypt(self, plaintext: bytes, public_key: bytes) -> bytes:
+        """Encrypt data with hybrid ML-KEM-768 + AES-256-GCM.
+
+        Args:
+            plaintext: Data to encrypt.
+            public_key: ML-KEM-768 public key.
+
+        Returns:
+            Hybrid ciphertext: kem_ct_len(4) || kem_ct || aes_encrypted
+        """
+        # Step 1: ML-KEM-768 key encapsulation -> shared secret (32 bytes)
+        kem_ciphertext, shared_secret = self._kem.encapsulate(public_key)
+
+        # Step 2: AES-256-GCM encrypt with the shared secret as key
+        aes = AESGCMEncryptor(key=shared_secret[:32])
+        aes_encrypted = aes.encrypt(plaintext)
+
+        # Step 3: Pack: kem_ct_len || kem_ct || aes_encrypted
+        return struct.pack(">I", len(kem_ciphertext)) + kem_ciphertext + aes_encrypted
+
+    def decrypt(self, data: bytes, secret_key: bytes) -> bytes:
+        """Decrypt hybrid ML-KEM-768 + AES-256-GCM ciphertext.
+
+        Args:
+            data: Hybrid ciphertext from encrypt().
+            secret_key: ML-KEM-768 secret key.
+
+        Returns:
+            Decrypted plaintext.
+        """
+        # Step 1: Unpack kem_ct_len
+        if len(data) < 4:
+            raise ValueError("Hybrid ciphertext too short")
+        kem_ct_len = struct.unpack(">I", data[:4])[0]
+
+        # Step 2: Extract KEM ciphertext and AES ciphertext
+        kem_ciphertext = data[4 : 4 + kem_ct_len]
+        aes_encrypted = data[4 + kem_ct_len :]
+
+        # Step 3: ML-KEM-768 decapsulation -> shared secret
+        shared_secret = self._kem.decapsulate(kem_ciphertext, secret_key)
+
+        # Step 4: AES-256-GCM decrypt
+        aes = AESGCMEncryptor(key=shared_secret[:32])
+        return aes.decrypt(aes_encrypted)

src/qp_vault/encryption/ml_dsa.py

@@ -0,0 +1,80 @@
+# Copyright 2026 Quantum Pipes Technologies, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""ML-DSA-65 post-quantum signatures for provenance attestation.
+
+Signs and verifies data using ML-DSA-65 (FIPS 204), providing
+quantum-resistant digital signatures for provenance records,
+Merkle proofs, and audit attestations.
+
+Requires: pip install qp-vault[pq]
+"""
+
+from __future__ import annotations
+
+try:
+    import oqs
+    HAS_OQS = True
+except ImportError:
+    HAS_OQS = False
+
+
+class MLDSASigner:
+    """ML-DSA-65 digital signature manager (FIPS 204).
+
+    Generates keypairs, signs data, and verifies signatures.
+    Used for provenance attestation and audit record signing.
+
+    Usage:
+        signer = MLDSASigner()
+        pub, sec = signer.generate_keypair()
+        signature = signer.sign(b"data", sec)
+        assert signer.verify(b"data", signature, pub)
+    """
+
+    ALGORITHM = "ML-DSA-65"
+
+    def __init__(self) -> None:
+        if not HAS_OQS:
+            raise ImportError(
+                "liboqs-python is required for ML-DSA-65. "
+                "Install with: pip install qp-vault[pq]"
+            )
+
+    def generate_keypair(self) -> tuple[bytes, bytes]:
+        """Generate an ML-DSA-65 keypair.
+
+        Returns:
+            (public_key, secret_key) as bytes.
+        """
+        sig = oqs.Signature(self.ALGORITHM)
+        public_key = sig.generate_keypair()
+        secret_key = sig.export_secret_key()
+        return public_key, secret_key
+
+    def sign(self, message: bytes, secret_key: bytes) -> bytes:
+        """Sign a message with ML-DSA-65.
+
+        Args:
+            message: The data to sign.
+            secret_key: ML-DSA-65 secret key.
+
+        Returns:
+            The signature bytes.
+        """
+        sig = oqs.Signature(self.ALGORITHM, secret_key=secret_key)
+        return sig.sign(message)
+
+    def verify(self, message: bytes, signature: bytes, public_key: bytes) -> bool:
+        """Verify an ML-DSA-65 signature.
+
+        Args:
+            message: The original signed data.
+            signature: The signature to verify.
+            public_key: ML-DSA-65 public key.
+
+        Returns:
+            True if signature is valid.
+        """
+        sig = oqs.Signature(self.ALGORITHM)
+        return sig.verify(message, signature, public_key)

src/qp_vault/encryption/ml_kem.py

@@ -0,0 +1,81 @@
+# Copyright 2026 Quantum Pipes Technologies, LLC
+# SPDX-License-Identifier: Apache-2.0
+
+"""ML-KEM-768 key encapsulation for post-quantum key exchange.
+
+Wraps AES-256-GCM data encryption keys (DEK) with ML-KEM-768 (FIPS 203).
+The encapsulated key can only be decapsulated by the holder of the
+ML-KEM-768 secret key, providing quantum-resistant key protection.
+
+Requires: pip install qp-vault[pq]
+"""
+
+from __future__ import annotations
+
+try:
+    import oqs
+    HAS_OQS = True
+except ImportError:
+    HAS_OQS = False
+
+
+class MLKEMKeyManager:
+    """ML-KEM-768 key encapsulation manager (FIPS 203).
+
+    Generates keypairs, encapsulates shared secrets, and decapsulates
+    them. Used to wrap AES-256-GCM keys for post-quantum protection.
+
+    Usage:
+        km = MLKEMKeyManager()
+        pub, sec = km.generate_keypair()
+        ciphertext, shared_secret = km.encapsulate(pub)
+        recovered = km.decapsulate(ciphertext, sec)
+        assert shared_secret == recovered
+    """
+
+    ALGORITHM = "ML-KEM-768"
+
+    def __init__(self) -> None:
+        if not HAS_OQS:
+            raise ImportError(
+                "liboqs-python is required for ML-KEM-768. "
+                "Install with: pip install qp-vault[pq]"
+            )
+
+    def generate_keypair(self) -> tuple[bytes, bytes]:
+        """Generate an ML-KEM-768 keypair.
+
+        Returns:
+            (public_key, secret_key) as bytes.
+        """
+        kem = oqs.KeyEncapsulation(self.ALGORITHM)
+        public_key = kem.generate_keypair()
+        secret_key = kem.export_secret_key()
+        return public_key, secret_key
+
+    def encapsulate(self, public_key: bytes) -> tuple[bytes, bytes]:
+        """Encapsulate a shared secret using a public key.
+
+        Args:
+            public_key: ML-KEM-768 public key.
+
+        Returns:
+            (ciphertext, shared_secret) — ciphertext is sent to key holder,
+            shared_secret is used as AES-256-GCM key.
+        """
+        kem = oqs.KeyEncapsulation(self.ALGORITHM)
+        ciphertext, shared_secret = kem.encap_secret(public_key)
+        return ciphertext, shared_secret
+
+    def decapsulate(self, ciphertext: bytes, secret_key: bytes) -> bytes:
+        """Decapsulate a shared secret using the secret key.
+
+        Args:
+            ciphertext: The encapsulated ciphertext from encapsulate().
+            secret_key: ML-KEM-768 secret key.
+
+        Returns:
+            The shared secret (same as returned by encapsulate).
+        """
+        kem = oqs.KeyEncapsulation(self.ALGORITHM, secret_key=secret_key)
+        return kem.decap_secret(ciphertext)

src/qp_vault/integrations/fastapi_routes.py

@@ -48,9 +48,9 @@ class AddResourceRequest(BaseModel):
         lifecycle: str = "active"
 
     class SearchRequest(BaseModel):
-        query: str
-        top_k: int = 10
-        threshold: float = 0.0
+        query: str = Field(..., max_length=10000)
+        top_k: int = Field(10, ge=1, le=1000)
+        threshold: float = Field(0.0, ge=0.0, le=1.0)
         trust_min: str | None = None
         layer: str | None = None
         collection: str | None = None
@@ -293,6 +293,8 @@ async def search_faceted(req: SearchRequest) -> dict[str, Any]:
     @router.post("/batch")
     async def add_batch(req: dict[str, Any]) -> dict[str, Any]:
         sources = req.get("sources", [])
+        if len(sources) > 100:
+            raise HTTPException(status_code=400, detail="Batch limited to 100 items")
         trust = req.get("trust", "working")
         tenant_id = req.get("tenant_id")
         resources = await vault.add_batch(