From 04d13de468ece8b001f1dc63be45da4dd85a1d5c Mon Sep 17 00:00:00 2001 From: juyung Date: Thu, 6 Aug 2026 20:31:30 +0000 Subject: [PATCH] Initial release --- README.md | 110 +++++++++++++++++++++++++++ decrypt.py | 192 +++++++++++++++++++++++++++++++++++++++++++++++ encrypt.py | 136 +++++++++++++++++++++++++++++++++ gen_kyber.py | 17 +++++ requirements.txt | 4 + 5 files changed, 459 insertions(+) create mode 100644 README.md create mode 100644 decrypt.py create mode 100644 encrypt.py create mode 100644 gen_kyber.py create mode 100644 requirements.txt diff --git a/README.md b/README.md new file mode 100644 index 0000000..39ad54e --- /dev/null +++ b/README.md @@ -0,0 +1,110 @@ +# Post-Quantum Hybrid Kyber-1024 + AES-256-CTR File Encryption + +Streaming hybrid file encryption: ML-KEM-1024 (Kyber-1024) handles post-quantum key exchange, AES-256-CTR handles bulk file encryption, and HMAC-SHA256 provides integrity verification. The file signature is `KYB`, and the file extension is ".kyb". + +Simplified flow (Kyber + AES-256 only, HKDF and HMAC omitted for clarity): + + + +## Features + +- Post-quantum secure key exchange via liboqs Kyber-1024 KEM (~256-bit PQ security) +- AES-256-CTR symmetric encryption with a streaming architecture that handles files larger than RAM (64 MB chunks) +- AES + HMAC keys derived from the Kyber shared secret via HKDF-SHA256 (never stored in the file) +- Unforgeable HMAC-SHA256 authenticity tag over the header + ciphertext to detect tampering or corruption +- Self-describing ASCII header (file signature `KYB`, version, algorithm names and lengths), 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)) + +```bash +python3 -m venv venv && . venv/bin/activate +pip install -r requirements.txt +``` + +## Usage + +1. Generate a Kyber-1024 key pair + +```bash +python3 gen_kyber.py +``` + +Produces: + +- `kyber.pub` - public key (1568 bytes) +- `kyber.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 human-readable header line 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 HMAC 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) +-|-|- +`kyber.pub`|Kyber-1024 public key|1568 +`kyber.sec`|Kyber-1024 secret key|3168 + +**Header fields** + +Field | Meaning | Value (v1) +-|-|- +|`KYB`|File signature that identifies a Kyber-encrypted file|`KYB` +`VER`|File-format version; decrypt refuses unsupported versions|`1` +`KEM`|Post-quantum key-encapsulation algorithm|`MLKEM1024` +`CIPHER`|Bulk symmetric cipher|`AES256CTR` +`MAC`|Integrity algorithm|`HMACSHA256` +`KYBERCT`|Kyber ciphertext length in bytes|`1568` +`IV`|AES-CTR nonce / HKDF salt length in bytes|`16` +`TAG`|MAC tag length in bytes|`32` + +**Encrypted file structure** + +`archive.7z.kyb`: + +Section | Size | Description +-|-|- +ASCII header line|82 bytes (v1)|Self-describing format line terminated by a newline, e.g. `KYB\|VER=1\|KEM=MLKEM1024\|CIPHER=AES256CTR\|MAC=HMACSHA256\|KYBERCT=1568\|IV=16\|TAG=32` +Kyber ciphertext|1568 bytes|Encapsulated key (ciphertext from ML-KEM-1024); length read from the header +IV (nonce)|16 bytes|AES-CTR nonce, also the HKDF salt; length read from the header +File ciphertext|same as plaintext|AES-256-CTR encrypted chunks, streamed in 64 MB blocks +HMAC-SHA256 tag|32 bytes|MAC over the header line + Kyber CT + IV + all ciphertext; length read from the header + +## Security notes + +- A fresh 32-byte **Kyber shared secret** is generated **per file** by `encap_secret()` and recovered only with `kyber.sec` via `decap_secret()`. It is the root key material: neither the AES key nor the Kyber private key. +- The AES and HMAC keys are **derived from that Kyber shared secret** with HKDF-SHA256 (domain-separated `info` labels), so no key material is ever stored in the file. +- Because the HMAC key is only derivable by the legitimate recipient (Kyber decapsulation), the tag is **unforgeable**: an attacker without the secret key cannot modify the file undetectably. +- The HMAC tag is computed over the **ASCII header line + Kyber CT + IV + AES ciphertext**, 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 HMAC 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. +- Kyber-1024 is believed to offer security equivalent to ~256-bit classical keys against quantum attackers. +- The secret key file (`kyber.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..6828d37 --- /dev/null +++ b/decrypt.py @@ -0,0 +1,192 @@ +#!/usr/bin/env python3 +""" +Hybrid decrypt a file encrypted with AES-256-CTR + HMAC-SHA256 + Kyber-1024. +Streaming handles files larger than RAM. +Usage: + python3 decrypt.py + +Output: original filename (without .kyb) +Requires: kyber.sec in the same directory. +""" + +import os +import sys +import tempfile +import hmac as hmac_mod +import oqs +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.hmac import HMAC +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +SIGNATURE = "KYB" +SUPPORTED_VER = "1" +INPUT_SUFFIX = ".kyb" +CHUNK_SIZE = 64 * 1024 * 1024 +MAX_HEADER_LEN = 1024 # sane upper bound for the header line + + +def derive_keys(shared_secret: bytes, salt: bytes): + """Derive independent AES and HMAC keys from the Kyber shared secret. + + The IV is used as the HKDF salt, binding the keys to this specific + encryption (defense-in-depth: key/nonce mismatches fail loudly). + """ + aes_key = HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=b"kyber-aes-key" + ).derive(shared_secret) + hmac_key = HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=b"kyber-hmac-key" + ).derive(shared_secret) + return aes_key, hmac_key + + +def parse_header(header_bytes: bytes): + """Parse the ASCII header line into a dict of fields.""" + line = header_bytes.rstrip(b"\n").decode("ascii") + parts = line.split("|") + if parts[0] != SIGNATURE: + raise ValueError("Error: Not a kyber (.kyb) file (bad signature)") + fields = {} + for part in parts[1:]: + if "=" in part: + key, _, value = part.partition("=") + fields[key] = value + return fields + + +def decrypt_file(input_path: str, output_path: str, secret_key: bytes): + with open(input_path, 'rb') as fin: + # 1. Parse the self-describing header line + header_bytes = fin.readline(MAX_HEADER_LEN) + if not header_bytes.endswith(b"\n"): + raise ValueError("Error: File truncated or corrupt header") + fields = parse_header(header_bytes) + + # 2. Version gate: refuse files this tool cannot parse + version = fields.get("VER") + if version != SUPPORTED_VER: + raise ValueError(f"Error: Unsupported format version: {version}") + + # 3. 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") != "AES256CTR": + raise ValueError(f"Error: Unsupported cipher: {fields.get('CIPHER')}") + if fields.get("MAC") != "HMACSHA256": + raise ValueError(f"Error: Unsupported MAC: {fields.get('MAC')}") + + try: + kyber_ct_len = int(fields["KYBERCT"]) + iv_len = int(fields["IV"]) + tag_len = int(fields["TAG"]) + except (KeyError, ValueError): + raise ValueError("Error: Corrupt header (missing or invalid length fields)") + + # Cross-checks: ML-KEM-1024 CT is always 1568 B; HMAC-SHA256 tag always 32 B + if kyber_ct_len != 1568: + raise ValueError("Error: Header inconsistent: MLKEM1024 implies a 1568-byte ciphertext") + if tag_len != 32: + raise ValueError("Error: Header inconsistent: HMACSHA256 implies a 32-byte tag") + + # 4. Read the Kyber ciphertext + IV + kyber_ct = fin.read(kyber_ct_len) + if len(kyber_ct) != kyber_ct_len: + raise ValueError("Error: File truncated: incomplete Kyber ciphertext") + iv = fin.read(iv_len) + if len(iv) != iv_len: + raise ValueError("Error: File truncated: incomplete header") + + # 5. Recover the shared secret, then derive AES + HMAC keys + kem = oqs.KeyEncapsulation("Kyber1024", secret_key) + try: + shared_secret = kem.decap_secret(kyber_ct) + except Exception: + raise ValueError("Error: Failed to decapsulate key - wrong secret key or corrupt header") + finally: + kem.free() + aes_key, hmac_key = derive_keys(shared_secret, iv) + + # 6. Stream-decrypt, authenticating the whole preamble + ciphertext + cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv)) + decryptor = cipher.decryptor() + hmac = HMAC(hmac_key, hashes.SHA256()) + hmac.update(header_bytes) + hmac.update(kyber_ct) + hmac.update(iv) + + # File size minus the MAC tag at the end + fin.seek(0, os.SEEK_END) + total_size = fin.tell() + if total_size < len(header_bytes) + kyber_ct_len + iv_len + tag_len: + raise ValueError("Error: File too small to be a valid .kyb file") + + data_end = total_size - tag_len + body_start = len(header_bytes) + kyber_ct_len + iv_len + fin.seek(body_start) + total_read = data_end - body_start + + # Write to a temp file; only rename into place after the HMAC 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) + hmac.update(chunk) + fout.write(decryptor.update(chunk)) + read += len(chunk) + print(f" Progress: {read}/{total_read} bytes", end='\r') + + # Verify integrity (constant-time comparison) + stored_hmac = fin.read(tag_len) + computed_hmac = hmac.finalize() + if not hmac_mod.compare_digest(stored_hmac, computed_hmac): + print() + raise ValueError("Error: Integrity check failed. File corrupted or tampered") + + fout.write(decryptor.finalize()) + 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 = "kyber.sec" + if not os.path.exists(seckey_path): + print("Error: No Kyber 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..0e308f7 --- /dev/null +++ b/encrypt.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +""" +Hybrid encrypt a file using AES-256-CTR + HMAC-SHA256 + Kyber-1024. +Streaming handles files larger than RAM. +Usage: + python3 encrypt.py + +Output: .kyb +Requires: kyber.pub in the same directory. +""" + +import os +import sys +import oqs +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.hmac import HMAC +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +KYBER_KEM = "Kyber1024" +SIGNATURE = "KYB" +VERSION = "1" +OUT_SUFFIX = ".kyb" +IV_LEN = 16 +TAG_LEN = 32 +CHUNK_SIZE = 64 * 1024 * 1024 # 64 MB + + +def build_header(kyber_ct_len: int, iv_len: int, tag_len: int) -> bytes: + """Build the self-describing ASCII header line (ends with a newline). + + Human-readable and labelled, so a decrypt script (or a person) can learn + every format parameter straight from the file with no hardcoded sizes. + """ + fields = [ + SIGNATURE, + f"VER={VERSION}", + "KEM=MLKEM1024", + "CIPHER=AES256CTR", + "MAC=HMACSHA256", + f"KYBERCT={kyber_ct_len}", + f"IV={iv_len}", + f"TAG={tag_len}", + ] + return "|".join(fields).encode("ascii") + b"\n" + + +def derive_keys(shared_secret: bytes, salt: bytes): + """Derive independent AES and HMAC keys from the Kyber shared secret. + + The IV is used as the HKDF salt, binding the keys to this specific + encryption (defense-in-depth: key/nonce mismatches fail loudly). + """ + aes_key = HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=b"kyber-aes-key" + ).derive(shared_secret) + hmac_key = HKDF( + algorithm=hashes.SHA256(), length=32, salt=salt, + info=b"kyber-hmac-key" + ).derive(shared_secret) + return aes_key, hmac_key + + +def encrypt_file(plaintext_path: str, output_path: str, pub_key: bytes): + iv = os.urandom(IV_LEN) + + # Fresh encapsulation per file -> fresh shared secret -> fresh AES/HMAC keys. + # Never reuse or cache a shared secret across files: the per-file keys are + # exactly what make accidental IV reuse harmless. + kem = oqs.KeyEncapsulation(KYBER_KEM) + kyber_ct, shared_secret = kem.encap_secret(pub_key) + kem.free() + + aes_key, hmac_key = derive_keys(shared_secret, iv) + + header = build_header(len(kyber_ct), IV_LEN, TAG_LEN) + + cipher = Cipher(algorithms.AES(aes_key), modes.CTR(iv)) + encryptor = cipher.encryptor() + hmac = HMAC(hmac_key, hashes.SHA256()) + + total_size = os.path.getsize(plaintext_path) + written = 0 + + with open(output_path, 'wb') as fout: + # Preamble (all authenticated): ASCII header + Kyber CT + IV + preamble = header + kyber_ct + iv + fout.write(preamble) + hmac.update(preamble) + + with open(plaintext_path, 'rb') as fin: + while True: + chunk = fin.read(CHUNK_SIZE) + if not chunk: + break + ct_chunk = encryptor.update(chunk) + hmac.update(ct_chunk) + fout.write(ct_chunk) + written += len(chunk) + print(f" Progress: {written}/{total_size} bytes", end='\r') + + final_ct = encryptor.finalize() + hmac.update(final_ct) # cover every ciphertext byte written (no-op for CTR) + fout.write(final_ct) + fout.write(hmac.finalize()) + + 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 = "kyber.pub" + if not os.path.exists(pubkey_path): + print(f"Error: No Kyber public key found. Run gen_kyber.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_kyber.py b/gen_kyber.py new file mode 100644 index 0000000..55a87cb --- /dev/null +++ b/gen_kyber.py @@ -0,0 +1,17 @@ +import oqs + +# ── Generate Kyber-1024 keypair ── +# ML-KEM-1024 gives ~256-bit post-quantum security +kem = oqs.KeyEncapsulation("Kyber1024") +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("kyber.pub", "wb") as f: + f.write(public_key) +with open("kyber.sec", "wb") as f: + f.write(secret_key) + +print(f"Public key size: {len(public_key)} bytes") # 1568 for Kyber-1024 +print(f"Secret key size: {len(secret_key)} bytes") # 3168 for Kyber-1024 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9c2cab0 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +cffi==2.1.0 +cryptography==49.0.0 +liboqs-python==0.16.0 +pycparser==3.0