refactor: code cleanup

Author: chadsec1

core/constants.py

@@ -36,12 +36,12 @@
 OTP_MAX_MESSAGE_LEN = OTP_PAD_SIZE - OTP_SIZE_LENGTH
 
 XCHACHA20POLY1305_NONCE_LEN      = 24
-XCHACHA20POLY1305_SIZE_LEN       = 3
+XCHACHA20POLY1305_SIZE_LEN       = 4
 XCHACHA20POLY1305_MAX_RANODM_PAD = OTP_PAD_SIZE
 
 CHACHA20POLY1305_NONCE_LEN      = 12
-CHACHA20POLY1305_SIZE_LEN       = 2
-CHACHA20POLY1305_MAX_RANODM_PAD = OTP_PAD_SIZE
+CHACHA20POLY1305_SIZE_LEN       = 4
+CHACHA20POLY1305_MAX_RANODM_PAD = 64 # DEBUG
 
 
 
@@ -72,7 +72,6 @@
 CLASSIC_MCELIECE_8_PK_LEN    = 1357824 
 CLASSIC_MCELIECE_8_CT_LEN    = 208 
 
-CLASSIC_MCELIECE_8_ROTATE_AT = 3   # Default OTP batches needed to be sent for a key rotation to occur
 
 
 
@@ -95,7 +94,11 @@
 }
 
 # hash parameters
-ARGON2_MEMORY      = 1 * 1024**3   # GB
+
+ARGON2_MEMORY      =  4 * 1024**3   # GB
+
+# ARGON2_MEMORY      =  1 * 1024**2   # Debug
+
 ARGON2_ITERS       = 3
 ARGON2_OUTPUT_LEN  = 64           # bytes
 ARGON2_SALT_LEN    = 16           # bytes (Must be always 16 for interoperability with implementations using libsodium.)

core/trad_crypto.py

@@ -143,7 +143,7 @@ def decrypt_xchacha20poly1305(key: bytes, nonce: bytes, ciphertext: bytes) -> by
 
 
 
-def encrypt_chacha20poly1305(key: bytes, plaintext: bytes, nonce: bytes = None, max_padding: int = XCHACHA20POLY1305_MAX_RANODM_PAD) -> tuple[bytes, bytes]:
+def encrypt_chacha20poly1305(key: bytes, plaintext: bytes, nonce: bytes = None, max_padding: int = CHACHA20POLY1305_MAX_RANODM_PAD) -> tuple[bytes, bytes]:
     """
     Encrypt plaintext using ChaCha20Poly1305.
 
@@ -153,7 +153,7 @@ def encrypt_chacha20poly1305(key: bytes, plaintext: bytes, nonce: bytes = None,
         key: A 32-byte ChaCha20Poly1305 key.
         plaintext: Data to encrypt.
         nonce: An (optional) nonce to be used.
-        max_padding: an (optional) maximum padding limit number to message. Cannot be larger than what `XCHACHA20POLY1305_MAX_RANODM_PAD` could store. Set to 0 for no padding.
+        max_padding: an (optional) maximum padding limit number to message. Cannot be larger than what `CHACHA20POLY1305_MAX_RANODM_PAD` could store. Set to 0 for no padding.
     Returns:
         A tuple (nonce, ciphertext) where:
         - nonce: The randomly generated nonce or the same given nonce.

logic/contacts.py

@@ -5,8 +5,7 @@
 
 from core.constants import (
     ML_KEM_1024_NAME,
-    CLASSIC_MCELIECE_8_NAME,
-    CLASSIC_MCELIECE_8_ROTATE_AT
+    CLASSIC_MCELIECE_8_NAME
 )
 
 def generate_nickname_id(length: int = 4) -> str:
@@ -75,9 +74,7 @@ def save_contact(user_data: dict, user_data_lock, contact_id: str) -> None:
                     "our_keys": {
                         CLASSIC_MCELIECE_8_NAME: {
                             "public_key": None,
-                            "private_key": None,
-                            "rotation_counter": 0,
-                            "rotate_at": CLASSIC_MCELIECE_8_ROTATE_AT,
+                            "private_key": None
                         },
                         ML_KEM_1024_NAME: {
                             "public_key": None,

logic/message.py

@@ -319,21 +319,18 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
             user_data["contacts"][contact_id]["contact_next_strand_key"] = contact_next_strand_key
 
-            user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotation_counter"] += 1
-            
             staged_kyber_private_key = bool(user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][ML_KEM_1024_NAME]["private_key"])
+            staged_mceliece_private_key = bool(user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"])
 
-            rotation_counter = user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotation_counter"]
 
 
-        logger.debug("Incremented McEliece's rotation_counter by 1 (now is %d) for contact (%s)", rotation_counter, contact_id)
 
         logger.info("Saved contact (%s) new batch of One-Time-Pads, new strand key, and new hash chain seed", contact_id)
         save_account_data(user_data, user_data_lock)
 
         # Why ???????
         # Nvm, I know why, PFS.
-        if not staged_kyber_private_key:
+        if (not staged_kyber_private_key) and (not staged_mceliece_private_key):
             logger.info("Rotating our ephemeral keys") 
             send_new_ephemeral_keys(user_data, user_data_lock, contact_id, ui_queue)
             save_account_data(user_data, user_data_lock)

logic/pfs.py

@@ -2,7 +2,7 @@
     Handles Perfect Forward Secrecy (PFS) ephemeral keys exchange and rotation for contacts.
 
     Handles:
-    - Generates and rotates ephemeral (one-time use) ML-KEM-1024 keys and (medium-term) Classic McEliece keys.
+    - Generates and rotates ephemeral (one-time use) ML-KEM-1024 and Classic McEliece keys.
     - Uses per-contact hash chains to prevent replay attacks, and verifies authenticity using ML-DSA-87.
     - Sends and receives signed ephemeral keys using long-term signing keys.
     - Updates local account storage with new key material after successful exchange.
@@ -24,7 +24,6 @@
     CHACHA20POLY1305_NONCE_LEN,
     CLASSIC_MCELIECE_8_NAME,
     CLASSIC_MCELIECE_8_PK_LEN,
-    CLASSIC_MCELIECE_8_ROTATE_AT,
     KEYS_HASH_CHAIN_LEN
 )
 from core.trad_crypto import (
@@ -86,8 +85,7 @@ def send_new_ephemeral_keys(user_data: dict, user_data_lock: threading.Lock, con
     Generate, encrypt, and send fresh ephemeral keys to a contact.
 
     - Maintains a per-contact hash chain for signing key material.
-    - Generates new Kyber1024 keys every call.
-    - Optionally rotates McEliece keys if rotation threshold is reached.
+    - Generates new Kyber1024 and McEliece keys every call.
     - Signs all key material with the long-term signing key.
     - Sends to the server using an authenticated HTTP request.
     - If successful, stores new keys in `user_data["tmp"]` for later update.
@@ -112,8 +110,6 @@ def send_new_ephemeral_keys(user_data: dict, user_data_lock: threading.Lock, con
     session_headers  = user_data_copied["tmp"]["session_headers"]
 
 
-    rotation_counter = user_data_copied["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotation_counter"] 
-    rotate_at        = user_data_copied["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotate_at"]
 
     lt_sign_private_key = user_data_copied["contacts"][contact_id]["lt_sign_keys"]["our_keys"]["private_key"]
 
@@ -131,14 +127,11 @@ def send_new_ephemeral_keys(user_data: dict, user_data_lock: threading.Lock, con
 
     # Generate new ML-KEM-1024 keys for us
     kyber_private_key, kyber_public_key = generate_kem_keys(ML_KEM_1024_NAME)
-    publickeys_hashchain = our_hash_chain + kyber_public_key
 
-    rotate_mceliece = False
-    if (rotate_at == rotation_counter) or (user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"] is None):
-        # Generate Classic McEliece 8192128f keys
-        mceliece_private_key, mceliece_public_key = generate_kem_keys(CLASSIC_MCELIECE_8_NAME)
-        publickeys_hashchain += mceliece_public_key
-        rotate_mceliece = True
+    # Generate Classic McEliece 8192128 keys
+    mceliece_private_key, mceliece_public_key = generate_kem_keys(CLASSIC_MCELIECE_8_NAME)
+    
+    publickeys_hashchain = our_hash_chain + kyber_public_key + mceliece_public_key    
 
     # Sign them with our per-contact long-term private key
     publickeys_hashchain_signature = create_signature(ML_DSA_87_NAME, publickeys_hashchain, lt_sign_private_key)
@@ -174,13 +167,8 @@ def send_new_ephemeral_keys(user_data: dict, user_data_lock: threading.Lock, con
         user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][ML_KEM_1024_NAME]["public_key"] = kyber_public_key
 
 
-        if rotate_mceliece:
-            user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"] = mceliece_private_key
-            user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_NAME]["public_key"] = mceliece_public_key
-
-
-            user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotation_counter"] = 0
-            user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["rotate_at"]        = CLASSIC_MCELIECE_8_ROTATE_AT
+        user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"] = mceliece_private_key
+        user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_NAME]["public_key"] = mceliece_public_key
 
 
 
@@ -260,9 +248,7 @@ def pfs_data_handler(user_data: dict, user_data_lock: threading.Lock, user_data_
 
 
     if (
-        (len(pfs_plaintext) < ML_KEM_1024_PK_LEN + ML_DSA_87_SIGN_LEN + KEYS_HASH_CHAIN_LEN) 
-        or 
-        len(pfs_plaintext) > ML_KEM_1024_PK_LEN + ML_DSA_87_SIGN_LEN + CLASSIC_MCELIECE_8_PK_LEN + KEYS_HASH_CHAIN_LEN
+        len(pfs_plaintext) != ML_KEM_1024_PK_LEN + ML_DSA_87_SIGN_LEN + CLASSIC_MCELIECE_8_PK_LEN + KEYS_HASH_CHAIN_LEN
     ):
         logger.error("Contact (%s) gave us a PFS request with malformed strand plaintext length (%d)", contact_id, len(pfs_plaintext))
         return
@@ -295,17 +281,12 @@ def pfs_data_handler(user_data: dict, user_data_lock: threading.Lock, user_data_
             logger.error("Contact keys hash chain does not match our computed hash chain! Skipping this PFS message...")
             return
 
-    contact_kyber_public_key = contact_publickeys_hashchain[KEYS_HASH_CHAIN_LEN: ML_KEM_1024_PK_LEN + KEYS_HASH_CHAIN_LEN]
-
-    if len(contact_publickeys_hashchain) == ML_KEM_1024_PK_LEN + CLASSIC_MCELIECE_8_PK_LEN + KEYS_HASH_CHAIN_LEN:
-        logger.info("contact (%s) has rotated their Kyber and McEliece keys", contact_id)
+    
+    logger.info("contact (%s) has rotated their Kyber and McEliece keys", contact_id)
 
-        contact_mceliece_public_key = contact_publickeys_hashchain[ML_KEM_1024_PK_LEN + KEYS_HASH_CHAIN_LEN:]
-        with user_data_lock:
-            user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_NAME] = contact_mceliece_public_key
+    contact_kyber_public_key = contact_publickeys_hashchain[KEYS_HASH_CHAIN_LEN: ML_KEM_1024_PK_LEN + KEYS_HASH_CHAIN_LEN]
 
-    elif len(contact_publickeys_hashchain) == ML_KEM_1024_PK_LEN + KEYS_HASH_CHAIN_LEN:
-        logger.info("contact (%s) has rotated their Kyber keys", contact_id)
+    contact_mceliece_public_key = contact_publickeys_hashchain[ML_KEM_1024_PK_LEN + KEYS_HASH_CHAIN_LEN:]
 
 
     logger.info("We are acknowledging contact's new PFS keys")
@@ -315,7 +296,8 @@ def pfs_data_handler(user_data: dict, user_data_lock: threading.Lock, user_data_
     with user_data_lock:
         user_data["contacts"][contact_id]["lt_sign_keys"]["contact_hash_chain"] = contact_hash_chain
         user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][ML_KEM_1024_NAME] = contact_kyber_public_key
-
+        user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_NAME] = contact_mceliece_public_key
+        
         our_kyber_private_key = user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][ML_KEM_1024_NAME]["private_key"]
         our_mceliece_private_key = user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"]