Initial release

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):
+
+<a href="https://notes.juyung.com/gallery/6nIr0qID0vri0DR4Jy7HCbS6/hJv-1N1hIqFxzXIv85KqR9dq"  target="_blank"><img src='https://notes.juyung.com/uploads/small2x/f9/20/0805d761b05dbbf2046a68268b58.png' style="height: 720px;" /></a>
+
+## 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.
+

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 <filename.kyb>
+
+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)
+    

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 <filename>
+
+Output: <filename>.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)
+    

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

requirements.txt

@@ -0,0 +1,4 @@
+cffi==2.1.0
+cryptography==49.0.0
+liboqs-python==0.16.0
+pycparser==3.0