commit ccc0b2fed2d861a47bdb14fc9cb406fc4cf06c06 Author: juyung Date: Sat Aug 8 17:27:56 2026 +0000 Initial release diff --git a/README.md b/README.md new file mode 100644 index 0000000..09fcbb5 --- /dev/null +++ b/README.md @@ -0,0 +1,111 @@ +# Post-Quantum Hybrid ML-KEM-1024 + AES-256-GCM File Encryption + +Streaming hybrid file encryption: ML-KEM-1024 handles post-quantum key exchange, and AES-256-GCM handles bulk file encryption with built-in authenticity. The file signature is `KYB`, and the file extension is ".kyb". + +Simplified flow (ML-KEM-1024 + AES-256 only, HKDF omitted for clarity): + + + +## Features + +- Post-quantum secure key exchange via liboqs ML-KEM-1024 KEM (~256-bit PQ security) +- AES-256-GCM authenticated encryption with a streaming architecture that handles files larger than RAM (64 MB chunks) +- AES key derived from the ML-KEM shared secret via HKDF-SHA256 (never stored in the file) +- Unforgeable 128-bit GCM tag over the header + ciphertext to detect tampering or corruption +- Self-describing CBOR header (RFC 8949; file signature `KYB`, version, algorithm and length fields), so each file identifies its own format and unsupported future versions are rejected cleanly + +## Requirements + +- `oqs` ([liboqs-python](https://github.com/open-quantum-safe/liboqs-python)) +- `cryptography` ([pyca/cryptography](https://github.com/pyca/cryptography)) +- `cbor2` ([cbor2](https://github.com/agronholm/cbor2)) + +```bash +python3 -m venv venv && . venv/bin/activate +pip install -r requirements.txt +``` + +## Usage + +1. Generate an ML-KEM-1024 key pair + +```bash +python3 gen_keys.py +``` + +Produces: + +- `mlkem1024.pub` - public key (1568 bytes) +- `mlkem1024.sec` - secret key (3168 bytes) + +2. Encrypt a file + +```bash +python3 encrypt.py archive.7z +``` + +Output: `archive.7z.kyb` + +The `.kyb` file is self-contained: a compact CBOR header at the start describes the exact format, algorithms, and lengths. See [File formats](#file-formats) for the layout. + +3. Decrypt a file + +```bash +python3 decrypt.py archive.7z.kyb +``` + +Output: `archive.7z` + +The script automatically verifies the GCM tag. If the file has been tampered with, you'll see: + +``` +Error: Integrity check failed. File corrupted or tampered +``` + +and the temp file is discarded. No plaintext ever appears at the destination path until the tag has been verified. + +## File formats + +**Key files (raw binary)** + +File|Content|Size (bytes) +-|-|- +`mlkem1024.pub`|ML-KEM-1024 public key|1568 +`mlkem1024.sec`|ML-KEM-1024 secret key|3168 + +**CBOR header** + +Layout: 3-byte signature `KYB` + 1-byte version, then a CBOR map (RFC 8949) of field name → value pairs. CBOR is self-delimiting, so the reader knows exactly where the map ends; unknown keys are simply ignored, keeping the format forward-compatible. The `KDF` and `INFO` fields record the key-derivation parameters, making the file fully self-describing: with the secret key alone, a from-scratch decryptor can derive the AES key from the header. + +Key | Value (v1) +-|- +`KEM`|`MLKEM1024` +`CIPHER`|`AES256GCM` +`MLKEMCT`|`1568` +`IV`|`12` +`TAG`|`16` +`KDF`|`HKDF-SHA256` +`INFO`|`mlkem-aes-key` + +**Encrypted file structure** + +`archive.7z.kyb`: + +Section | Size | Description +-|-|- +CBOR header|91 bytes (v1)|Signature `KYB` (3 B) + version (1 B) + CBOR map (87 B) +ML-KEM ciphertext|1568 bytes|Encapsulated key (ciphertext from ML-KEM-1024); length read from the header +IV (nonce)|12 bytes|GCM nonce, also the HKDF salt; length read from the header +File ciphertext|same as plaintext|AES-256-GCM encrypted chunks, streamed in 64 MB blocks +GCM auth tag|16 bytes|Tag over the header + ML-KEM CT + IV + all ciphertext; length read from the header + +## Security notes + +- A fresh 32-byte **ML-KEM shared secret** is generated **per file** by `encap_secret()` and recovered only with `mlkem1024.sec` via `decap_secret()`. It is the root key material: neither the AES key nor the ML-KEM private key. +- The AES key is **derived from that ML-KEM shared secret** with HKDF-SHA256 (domain-separated `info` label, both recorded in the header), so no key material is ever stored in the file. +- Because the AES key is only derivable by the legitimate recipient (ML-KEM decapsulation), the GCM tag is **unforgeable**: an attacker without the secret key cannot modify the file undetectably. +- The GCM tag is computed over the **CBOR header + ML-KEM CT + IV + AES ciphertext** (via additional authenticated data), so any tampering with the encrypted data, the nonce, or the format header is detected before decryption. +- Decrypted plaintext is written to a temporary file and atomically renamed into place **only after** the GCM tag verifies. +- The header carries a format **version**; decrypt checks it up front, so a file written by a newer (or unknown) format is refused with a clear error instead of being mis-parsed. +- ML-KEM-1024 is believed to offer security equivalent to ~256-bit classical keys against quantum attackers. +- The secret key file (`mlkem1024.sec`) is the single point of failure. Losing it means losing every encrypted file. Treat it like a private SSH key: keep it offline, back it up, and restrict access (e.g. `chmod 600`). For stronger protection, consider encrypting the key file with a passphrase or splitting it into pieces (XOR-splitting or Shamir secret sharing) across separate locations. diff --git a/decrypt.py b/decrypt.py new file mode 100644 index 0000000..fb42155 --- /dev/null +++ b/decrypt.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +""" +Hybrid decrypt a file encrypted with AES-256-GCM + ML-KEM-1024. +Streaming handles files larger than RAM. +Usage: + python3 decrypt.py + +Output: original filename (without .kyb) +Requires: mlkem1024.sec in the same directory. +""" + +import os +import sys +import tempfile +from io import BytesIO +import cbor2 +import oqs +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +SIGNATURE = b"KYB" +SUPPORTED_VER = 1 +INPUT_SUFFIX = ".kyb" +CHUNK_SIZE = 64 * 1024 * 1024 +MAX_HEADER_LEN = 1024 # sane upper bound for the CBOR header map +HEADER_PREFIX_LEN = 4 # signature (3) + version (1) +KDF_NAME = "HKDF-SHA256" # KDF advertised in the header +KDF_INFO = "mlkem-aes-key" # HKDF domain-separation label + + +def derive_key(shared_secret: bytes, salt: bytes) -> bytes: + """Derive the single AES-256-GCM key from the ML-KEM shared secret. + + The nonce is used as the HKDF salt, binding the key to this specific + encryption (defense-in-depth: key/nonce mismatches fail loudly). + """ + return HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=KDF_INFO.encode("ascii") + ).derive(shared_secret) + + +def decrypt_file(input_path: str, output_path: str, secret_key: bytes): + with open(input_path, 'rb') as fin: + # 1. Read the fixed prefix: signature (3) + version (1) + prefix = fin.read(HEADER_PREFIX_LEN) + if len(prefix) != HEADER_PREFIX_LEN: + raise ValueError("Error: File truncated or corrupt header") + if prefix[:3] != SIGNATURE: + raise ValueError("Error: Not a .kyb file (bad signature)") + + # 2. Version gate: refuse files this tool cannot parse + if prefix[3] != SUPPORTED_VER: + raise ValueError(f"Error: Unsupported format version: {prefix[3]}") + + # 3. Read the CBOR header map (bounded read, then locate its end) + chunk = fin.read(MAX_HEADER_LEN) + if not chunk: + raise ValueError("Error: File truncated or corrupt header") + decoder = cbor2.CBORDecoder(BytesIO(chunk)) + try: + fields = decoder.decode() + cbor_len = decoder.fp.tell() + except EOFError: + raise ValueError("Error: File truncated or corrupt header") + except Exception: + raise ValueError("Error: Corrupt header (invalid CBOR)") + if not isinstance(fields, dict): + raise ValueError("Error: Corrupt header (CBOR map expected)") + header_bytes = prefix + chunk[:cbor_len] + # Position exactly after the CBOR map so the ML-KEM CT reads correctly + fin.seek(HEADER_PREFIX_LEN + cbor_len) + + # 4. Validate the advertised algorithms and derive the lengths + kem_name = fields.get("KEM") + if kem_name not in ("MLKEM1024",): + raise ValueError(f"Error: Unsupported KEM: {kem_name}") + if fields.get("CIPHER") != "AES256GCM": + raise ValueError(f"Error: Unsupported cipher: {fields.get('CIPHER')}") + if fields.get("KDF") != KDF_NAME: + raise ValueError(f"Error: Unsupported KDF: {fields.get('KDF')}") + if fields.get("INFO") != KDF_INFO: + raise ValueError(f"Error: Unsupported KDF info: {fields.get('INFO')}") + + try: + mlkem_ct_len = fields["MLKEMCT"] + iv_len = fields["IV"] + tag_len = fields["TAG"] + except (KeyError, TypeError): + raise ValueError("Error: Corrupt header (missing or invalid length fields)") + + # Cross-checks: ML-KEM-1024 CT is always 1568 B; GCM nonce 12 B; tag 16 B + if mlkem_ct_len != 1568: + raise ValueError("Error: Header inconsistent: MLKEM1024 implies a 1568-byte ciphertext") + if iv_len != 12: + raise ValueError("Error: Header inconsistent: AES256GCM implies a 12-byte nonce") + if tag_len != 16: + raise ValueError("Error: Header inconsistent: AES256GCM implies a 16-byte tag") + + # 5. Read the ML-KEM ciphertext + IV + mlkem_ct = fin.read(mlkem_ct_len) + if len(mlkem_ct) != mlkem_ct_len: + raise ValueError("Error: File truncated: incomplete ML-KEM ciphertext") + iv = fin.read(iv_len) + if len(iv) != iv_len: + raise ValueError("Error: File truncated: incomplete header") + + # 6. Recover the shared secret, then derive the AES key + kem = oqs.KeyEncapsulation("ML-KEM-1024", secret_key) + try: + shared_secret = kem.decap_secret(mlkem_ct) + except Exception: + raise ValueError("Error: Failed to decapsulate key - wrong secret key or corrupt header") + finally: + kem.free() + aes_key = derive_key(shared_secret, iv) + + # File size minus the GCM tag at the end + fin.seek(0, os.SEEK_END) + total_size = fin.tell() + if total_size < len(header_bytes) + mlkem_ct_len + iv_len + tag_len: + raise ValueError("Error: File too small to be a valid .kyb file") + + # The trailing GCM tag must be supplied up front (pyca/cryptography) + fin.seek(total_size - tag_len) + gcm_tag = fin.read(tag_len) + + data_end = total_size - tag_len + body_start = len(header_bytes) + mlkem_ct_len + iv_len + fin.seek(body_start) + total_read = data_end - body_start + + # 7. Stream-decrypt, authenticating the whole preamble + ciphertext + cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv, gcm_tag)) + decryptor = cipher.decryptor() + decryptor.authenticate_additional_data(header_bytes + mlkem_ct + iv) + + # Write to a temp file; only rename into place after the tag verifies, + # so corrupted plaintext never appears at the destination path. + fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(output_path) or ".", + prefix=".dec-", suffix=".tmp") + try: + with os.fdopen(fd, 'wb') as fout: + read = 0 + while fin.tell() < data_end: + chunk_size = min(CHUNK_SIZE, data_end - fin.tell()) + chunk = fin.read(chunk_size) + fout.write(decryptor.update(chunk)) + read += len(chunk) + print(f" Progress: {read}/{total_read} bytes", end='\r') + + # Verify integrity: finalize() checks the GCM tag + try: + decryptor.finalize() + except InvalidTag: + print() + raise ValueError("Error: Integrity check failed. File corrupted or tampered") + + fout.flush() + os.fsync(fout.fileno()) + + # Atomic move: only a fully verified file lands at output_path + os.replace(temp_path, output_path) + except BaseException: + if os.path.exists(temp_path): + os.remove(temp_path) + raise + + print(f"\nSuccess: {input_path} → {output_path} ({total_read} bytes)") + + +if __name__ == "__main__": + if len(sys.argv) != 2: + print(__doc__) + sys.exit(1) + + infile = sys.argv[1] + + if not os.path.exists(infile): + print(f"Error: File not found: {infile}") + sys.exit(1) + + if not infile.endswith(INPUT_SUFFIX): + print(f"Warning: File doesn't end with {INPUT_SUFFIX}. Decrypting anyway") + + seckey_path = "mlkem1024.sec" + if not os.path.exists(seckey_path): + print("Error: No ML-KEM-1024 secret key found.") + sys.exit(1) + + with open(seckey_path, 'rb') as f: + secret_key = f.read() + + outfile = infile[:-len(INPUT_SUFFIX)] if infile.endswith(INPUT_SUFFIX) else infile + ".dec" + decrypt_file(infile, outfile, secret_key) + diff --git a/encrypt.py b/encrypt.py new file mode 100644 index 0000000..a71514e --- /dev/null +++ b/encrypt.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +""" +Hybrid encrypt a file using AES-256-GCM + ML-KEM-1024. +Streaming handles files larger than RAM. +Usage: + python3 encrypt.py + +Output: .kyb +Requires: mlkem1024.pub in the same directory. +""" + +import os +import sys +import cbor2 +import oqs +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +MLKEM_KEM = "ML-KEM-1024" +SIGNATURE = b"KYB" # 3-byte file signature +VERSION = 1 # 1-byte format version +KDF_NAME = "HKDF-SHA256" # KDF advertised in the header +KDF_INFO = "mlkem-aes-key" # HKDF domain-separation label +OUT_SUFFIX = ".kyb" +IV_LEN = 12 # GCM standard nonce length (96 bits) +TAG_LEN = 16 # full GCM auth tag (128 bits) +CHUNK_SIZE = 64 * 1024 * 1024 # 64 MB + + +def build_header(mlkem_ct_len: int, iv_len: int, tag_len: int) -> bytes: + """Build the CBOR header: signature + version + CBOR map of fields. + + CBOR (RFC 8949) is compact, self-delimiting, and standardized in IETF + crypto formats (e.g. COSE), so the header stays forward-compatible. + """ + header_map = { + "KEM": "MLKEM1024", + "CIPHER": "AES256GCM", + "MLKEMCT": mlkem_ct_len, + "IV": iv_len, + "TAG": tag_len, + "KDF": "HKDF-SHA256", + "INFO": "mlkem-aes-key" + } + return SIGNATURE + bytes([VERSION]) + cbor2.dumps(header_map, canonical=True) + + +def derive_key(shared_secret: bytes, salt: bytes) -> bytes: + """Derive the single AES-256-GCM key from the ML-KEM shared secret. + + The nonce is used as the HKDF salt, binding the key to this specific + encryption (defense-in-depth: key/nonce mismatches fail loudly). + """ + return HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=KDF_INFO.encode("ascii") + ).derive(shared_secret) + + +def encrypt_file(plaintext_path: str, output_path: str, pub_key: bytes): + iv = os.urandom(IV_LEN) + + # Each file gets a brand-new AES key: + # fresh shared secret (encap_secret) -> fresh derived key (derive_key). + # Since every file has its own unique key, even a repeated nonce between + # files would be safe. Never cache or reuse a shared secret across files - + # doing so would break that guarantee. + kem = oqs.KeyEncapsulation(MLKEM_KEM) + mlkem_ct, shared_secret = kem.encap_secret(pub_key) + kem.free() + + aes_key = derive_key(shared_secret, iv) + + header = build_header(len(mlkem_ct), IV_LEN, TAG_LEN) + + cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv)) + encryptor = cipher.encryptor() + # Preamble (header + ML-KEM CT + nonce) is authenticated as GCM AAD + preamble = header + mlkem_ct + iv + encryptor.authenticate_additional_data(preamble) + + total_size = os.path.getsize(plaintext_path) + written = 0 + + try: + with open(output_path, 'wb') as fout: + fout.write(preamble) + + with open(plaintext_path, 'rb') as fin: + while True: + chunk = fin.read(CHUNK_SIZE) + if not chunk: + break + ct_chunk = encryptor.update(chunk) + fout.write(ct_chunk) + written += len(chunk) + print(f" Progress: {written}/{total_size} bytes", end='\r') + + # finalize() computes the tag over AAD + all ciphertext + encryptor.finalize() + fout.write(encryptor.tag) + except BaseException: + # Never leave a partial .kyb behind on failure + if os.path.exists(output_path): + os.remove(output_path) + raise + + print(f"\nSuccess: {plaintext_path} → {output_path} ({total_size} bytes)") + + +if __name__ == "__main__": + if len(sys.argv) != 2: + print(__doc__) + sys.exit(1) + + infile = sys.argv[1] + + if not os.path.exists(infile): + print(f"Error: File not found: {infile}") + sys.exit(1) + + pubkey_path = "mlkem1024.pub" + if not os.path.exists(pubkey_path): + print(f"Error: No ML-KEM-1024 public key found. Run gen_keys.py first.") + sys.exit(1) + + with open(pubkey_path, 'rb') as f: + pub_key = f.read() + + outfile = infile + OUT_SUFFIX + if os.path.exists(outfile): + print(f"Error: Output already exists: {outfile}") + sys.exit(1) + + encrypt_file(infile, outfile, pub_key) + diff --git a/gen_keys.py b/gen_keys.py new file mode 100644 index 0000000..efe53fa --- /dev/null +++ b/gen_keys.py @@ -0,0 +1,17 @@ +import oqs + +# ── Generate ML-KEM-1024 keypair ── +# ML-KEM-1024 gives ~256-bit post-quantum security +kem = oqs.KeyEncapsulation("ML-KEM-1024") +public_key = kem.generate_keypair() # bytes +secret_key = kem.export_secret_key() # bytes +kem.free() + +# ── Save to disk (PEM-like, but just raw bytes is fine) ── +with open("mlkem1024.pub", "wb") as f: + f.write(public_key) +with open("mlkem1024.sec", "wb") as f: + f.write(secret_key) + +print(f"Public key size: {len(public_key)} bytes") # 1568 for ML-KEM-1024 +print(f"Secret key size: {len(secret_key)} bytes") # 3168 for ML-KEM-1024 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a050457 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +cbor2==5.6.2 +cffi==2.1.0 +cryptography==49.0.0 +liboqs-python==0.16.0 +pycparser==3.0