Update protocol

Author: chadsec1

core/constants.py

@@ -39,6 +39,11 @@
 XCHACHA20POLY1305_SIZE_LEN       = 3
 XCHACHA20POLY1305_MAX_RANODM_PAD = OTP_PAD_SIZE
 
+CHACHA20POLY1305_NONCE_LEN      = 12
+CHACHA20POLY1305_SIZE_LEN       = 2
+CHACHA20POLY1305_MAX_RANODM_PAD = OTP_PAD_SIZE
+
+
 
 
 SMP_NONCE_LENGTH      = 64
@@ -62,12 +67,12 @@
 ML_DSA_87_SIGN_LEN = 4627
 
 
-CLASSIC_MCELIECE_8_F_NAME      = "Classic-McEliece-8192128f"
-CLASSIC_MCELIECE_8_F_SK_LEN    = 14120
-CLASSIC_MCELIECE_8_F_PK_LEN    = 1357824 
-CLASSIC_MCELIECE_8_F_CT_LEN    = 208 
+CLASSIC_MCELIECE_8_NAME      = "Classic-McEliece-8192128"
+CLASSIC_MCELIECE_8_SK_LEN    = 14120
+CLASSIC_MCELIECE_8_PK_LEN    = 1357824 
+CLASSIC_MCELIECE_8_CT_LEN    = 208 
 
-CLASSIC_MCELIECE_8_F_ROTATE_AT = 3   # Default OTP batches needed to be sent for a key rotation to occur
+CLASSIC_MCELIECE_8_ROTATE_AT = 3   # Default OTP batches needed to be sent for a key rotation to occur
 
 
 
@@ -82,16 +87,16 @@
         "PK_LEN"  : ML_DSA_87_PK_LEN,
         "SIGN_LEN": ML_DSA_87_SIGN_LEN
     },
-    CLASSIC_MCELIECE_8_F_NAME: {
-        "SK_LEN": CLASSIC_MCELIECE_8_F_SK_LEN,
-        "PK_LEN": CLASSIC_MCELIECE_8_F_PK_LEN,
-        "CT_LEN": CLASSIC_MCELIECE_8_F_CT_LEN
+    CLASSIC_MCELIECE_8_NAME: {
+        "SK_LEN": CLASSIC_MCELIECE_8_SK_LEN,
+        "PK_LEN": CLASSIC_MCELIECE_8_PK_LEN,
+        "CT_LEN": CLASSIC_MCELIECE_8_CT_LEN
     },
 }
 
 # hash parameters
 ARGON2_MEMORY      = 1 * 1024**3   # GB
-ARGON2_ITERS       = 25
+ARGON2_ITERS       = 3
 ARGON2_OUTPUT_LEN  = 64           # bytes
 ARGON2_SALT_LEN    = 16           # bytes (Must be always 16 for interoperability with implementations using libsodium.)
 ARGON2_LANES       = 4

core/crypto.py

@@ -4,7 +4,7 @@
 Post-quantum cryptographic operations for Coldwire.
 
 Implements:
-- Key generation (ML-KEM-1024,  ML-DSA-87, Classic-McEliece-8192128f)
+- Key generation (ML-KEM-1024,  ML-DSA-87, Classic-McEliece-8192128)
 - Signature creation and verification
 - One-Time Pad (OTP) encryption with padding
 - Retrieving shared secrets from KEM chunks

core/trad_crypto.py

@@ -14,6 +14,11 @@
     XCHACHA20POLY1305_NONCE_LEN,
     XCHACHA20POLY1305_SIZE_LEN,
     XCHACHA20POLY1305_MAX_RANODM_PAD,
+
+    CHACHA20POLY1305_NONCE_LEN,
+    CHACHA20POLY1305_SIZE_LEN,
+    CHACHA20POLY1305_MAX_RANODM_PAD,
+    
     ARGON2_ITERS,
     ARGON2_MEMORY,
     ARGON2_LANES,
@@ -92,7 +97,7 @@ def encrypt_xchacha20poly1305(key: bytes, plaintext: bytes, nonce: bytes = None,
         raise ValueError(f"Max_padding is less than 0! ({max_padding})")
 
     if max_padding > 2 ** (XCHACHA20POLY1305_SIZE_LEN * 8) - 1:
-        raise ValueError(f"Max_padding is more than ``XCHACHA20POLY1305_SIZE_LEN`! ({max_padding})")
+        raise ValueError(f"Max_padding is more than `XCHACHA20POLY1305_SIZE_LEN`! ({max_padding})")
 
     padding = secrets.token_bytes(secrets.randbelow(max_padding + 1))
     padding_length_bytes = len(padding).to_bytes(XCHACHA20POLY1305_SIZE_LEN, "big")
@@ -131,3 +136,72 @@ 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]:
+    """
+    Encrypt plaintext using ChaCha20Poly1305.
+
+    A random nonce is generated for each encryption unless you specify one.
+
+    Args:
+        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.
+    Returns:
+        A tuple (nonce, ciphertext) where:
+        - nonce: The randomly generated nonce or the same given nonce.
+        - ciphertext: The encrypted data including the authentication tag.
+    """
+    if nonce is None:
+        nonce = sha3_512(secrets.token_bytes(CHACHA20POLY1305_NONCE_LEN))[:CHACHA20POLY1305_NONCE_LEN]
+
+    if max_padding < 0:
+        raise ValueError(f"Max_padding is less than 0! ({max_padding})")
+
+    if max_padding > 2 ** (CHACHA20POLY1305_SIZE_LEN * 8) - 1:
+        raise ValueError(f"Max_padding is more than `CHACHA20POLY1305_SIZE_LEN`! ({max_padding})")
+
+    padding = secrets.token_bytes(secrets.randbelow(max_padding + 1))
+    padding_length_bytes = len(padding).to_bytes(CHACHA20POLY1305_SIZE_LEN, "big")
+
+    padded_plaintext = padding_length_bytes + plaintext + padding
+
+
+    ciphertext = bindings.crypto_aead_chacha20poly1305_ietf_encrypt(padded_plaintext, None, nonce, key) 
+
+    return nonce, ciphertext
+
+
+def decrypt_chacha20poly1305(key: bytes, nonce: bytes, ciphertext: bytes) -> bytes:
+    """
+    Decrypt ciphertext using ChaCha20Poly1305.
+
+    Raises an exception if authentication fails.
+
+    Args:
+        key: The 32-byte ChaCha20Poly1305 key used for encryption.
+        nonce: The nonce used during encryption.
+        ciphertext: The encrypted data including the authentication tag.
+
+    Returns:
+        The decrypted plaintext bytes with no padding.
+    """
+
+    padded_plaintext = bindings.crypto_aead_chacha20poly1305_ietf_decrypt(ciphertext, None, nonce, key)
+
+    padding_length = int.from_bytes(padded_plaintext[:CHACHA20POLY1305_SIZE_LEN], "big")
+
+    if padding_length < 0:
+        raise ValueError(f"Negative padding length ({padding_length}), ciphertext likely corrupted, or key is invalid!")
+
+    return padded_plaintext[CHACHA20POLY1305_SIZE_LEN : -padding_length]
+
+
+

logic/background_worker.py

@@ -13,13 +13,13 @@
     SMP_TYPES,
     PFS_TYPES,
     MSG_TYPES,
-    XCHACHA20POLY1305_NONCE_LEN
+    CHACHA20POLY1305_NONCE_LEN
 )
 from core.crypto import (
         random_number_range
 )
 from core.trad_crypto import (
-        decrypt_xchacha20poly1305
+        decrypt_chacha20poly1305
 )
 from base64 import b64decode, urlsafe_b64encode
 import copy
@@ -174,11 +174,11 @@ def background_worker(user_data, user_data_lock, ui_queue, stop_flag):
 
                 if contact_next_strand_key and contact_next_strand_nonce:
                     try:
-                        blob_plaintext = decrypt_xchacha20poly1305(contact_next_strand_key, contact_next_strand_nonce, blob)
+                        blob_plaintext = decrypt_chacha20poly1305(contact_next_strand_key, contact_next_strand_nonce, blob)
 
                         contact_next_strand_key   = blob_plaintext[:32]
-                        contact_next_strand_nonce = blob_plaintext[32:32 + XCHACHA20POLY1305_NONCE_LEN]
-                        blob_plaintext = blob_plaintext[32 + XCHACHA20POLY1305_NONCE_LEN:]
+                        contact_next_strand_nonce = blob_plaintext[32:32 + CHACHA20POLY1305_NONCE_LEN]
+                        blob_plaintext = blob_plaintext[32 + CHACHA20POLY1305_NONCE_LEN:]
 
                         with user_data_lock:
                             user_data["contacts"][sender]["contact_next_strand_key"]   = contact_next_strand_key

logic/contacts.py

@@ -5,8 +5,8 @@
 
 from core.constants import (
     ML_KEM_1024_NAME,
-    CLASSIC_MCELIECE_8_F_NAME,
-    CLASSIC_MCELIECE_8_F_ROTATE_AT
+    CLASSIC_MCELIECE_8_NAME,
+    CLASSIC_MCELIECE_8_ROTATE_AT
 )
 
 def generate_nickname_id(length: int = 4) -> str:
@@ -69,23 +69,23 @@ def save_contact(user_data: dict, user_data_lock, contact_id: str) -> None:
                 },
                 "ephemeral_keys": {
                     "contact_public_keys": {
-                        CLASSIC_MCELIECE_8_F_NAME: None,
+                        CLASSIC_MCELIECE_8_NAME: None,
                         ML_KEM_1024_NAME: None
                     },
                     "our_keys": {
-                        CLASSIC_MCELIECE_8_F_NAME: {
+                        CLASSIC_MCELIECE_8_NAME: {
                             "public_key": None,
                             "private_key": None,
                             "rotation_counter": 0,
-                            "rotate_at": CLASSIC_MCELIECE_8_F_ROTATE_AT,
+                            "rotate_at": CLASSIC_MCELIECE_8_ROTATE_AT,
                         },
                         ML_KEM_1024_NAME: {
                             "public_key": None,
                             "private_key": None,
                         },
                     },
                     "staged_keys": {
-                        CLASSIC_MCELIECE_8_F_NAME: {
+                        CLASSIC_MCELIECE_8_NAME: {
                             "public_key": None,
                             "private_key": None,
                         },

logic/message.py

@@ -13,8 +13,7 @@
 from logic.pfs import send_new_ephemeral_keys
 from core.trad_crypto import (
         sha3_512,
-        encrypt_xchacha20poly1305,
-        decrypt_xchacha20poly1305
+        encrypt_chacha20poly1305
 )
 from core.crypto import (
     generate_shared_secrets,
@@ -33,9 +32,9 @@
     ML_KEM_1024_CT_LEN,
     ML_DSA_87_NAME,  
     ML_DSA_87_SIGN_LEN,
-    CLASSIC_MCELIECE_8_F_NAME,
-    CLASSIC_MCELIECE_8_F_CT_LEN,
-    XCHACHA20POLY1305_NONCE_LEN
+    CLASSIC_MCELIECE_8_NAME,
+    CLASSIC_MCELIECE_8_CT_LEN,
+    CHACHA20POLY1305_NONCE_LEN
 
 )
 from base64 import b64encode
@@ -57,7 +56,7 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
         auth_token = user_data["token"]
 
         contact_kyber_public_key    = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][ML_KEM_1024_NAME]
-        contact_mceliece_public_key = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_F_NAME]
+        contact_mceliece_public_key = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_NAME]
         our_lt_private_key          = user_data["contacts"][contact_id]["lt_sign_keys"]["our_keys"]["private_key"]
 
         our_next_strand_key = user_data["contacts"][contact_id]["our_next_strand_key"]
@@ -67,22 +66,22 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
         session_headers  = user_data["tmp"]["session_headers"]
 
     kyber_ciphertext_blob   , kyber_shared_secrets    = generate_shared_secrets(contact_kyber_public_key, ML_KEM_1024_NAME)
-    mceliece_ciphertext_blob, mceliece_shared_secrets = generate_shared_secrets(contact_mceliece_public_key, CLASSIC_MCELIECE_8_F_NAME)
+    mceliece_ciphertext_blob, mceliece_shared_secrets = generate_shared_secrets(contact_mceliece_public_key, CLASSIC_MCELIECE_8_NAME)
 
-    xchacha_shared_secrets = b''
-    while len(xchacha_shared_secrets) < OTP_PAD_SIZE:
-        xchacha_shared_secrets += sha3_512(secrets.token_bytes(64))
+    chacha_shared_secrets = b''
+    while len(chacha_shared_secrets) < OTP_PAD_SIZE:
+        chacha_shared_secrets += sha3_512(secrets.token_bytes(64))
 
 
     otp_batch_signature = create_signature(ML_DSA_87_NAME, kyber_ciphertext_blob + mceliece_ciphertext_blob, our_lt_private_key)
 
     # Here, the strandkey is actually just added to make messages structure uniform and easier to process in implementations
     # once contact receives this, he will save this new random key, then, process the batch, and save the new key derived from the batch.
 
-    new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
-    _, ciphertext_blob = encrypt_xchacha20poly1305(
+    new_strand_nonce = sha3_512(secrets.token_bytes(CHACHA20POLY1305_NONCE_LEN))[:CHACHA20POLY1305_NONCE_LEN]
+    _, ciphertext_blob = encrypt_chacha20poly1305(
             our_next_strand_key, 
-            sha3_512(secrets.token_bytes(32))[:32] + new_strand_nonce + MSG_TYPES["MSG_BATCH"] + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob + xchacha_shared_secrets,
+            sha3_512(secrets.token_bytes(32))[:32] + new_strand_nonce + MSG_TYPES["MSG_BATCH"] + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob + chacha_shared_secrets,
             nonce = our_next_strand_nonce
         )
 
@@ -101,7 +100,7 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
 
     # XOR shared secrets together for hybrid encryption
     pads, _ = one_time_pad(kyber_shared_secrets, mceliece_shared_secrets)
-    pads, _ = one_time_pad(pads, xchacha_shared_secrets)
+    pads, _ = one_time_pad(pads, chacha_shared_secrets)
 
     # Derive key from pad + truncate it.
     new_strand_key = pads[:32]
@@ -138,7 +137,7 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
         session_headers  = user_data["tmp"]["session_headers"]
 
         contact_kyber_public_key    = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][ML_KEM_1024_NAME]
-        contact_mceliece_public_key = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_F_NAME]
+        contact_mceliece_public_key = user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_NAME]
 
         our_pads = user_data["contacts"][contact_id]["our_pads"]
 
@@ -197,7 +196,7 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
     # which would break all of our security
 
     new_strand_key   = sha3_512(secrets.token_bytes(32))[:32]
-    new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
+    new_strand_nonce = sha3_512(secrets.token_bytes(CHACHA20POLY1305_NONCE_LEN))[:CHACHA20POLY1305_NONCE_LEN]
 
     with user_data_lock:
         user_data["contacts"][contact_id]["our_pads"] = user_data["contacts"][contact_id]["our_pads"][len(message_encrypted):]
@@ -210,7 +209,7 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
 
     save_account_data(user_data, user_data_lock)
 
-    _, ciphertext_blob = encrypt_xchacha20poly1305(
+    _, ciphertext_blob = encrypt_chacha20poly1305(
             our_next_strand_key, 
             new_strand_key + new_strand_nonce + MSG_TYPES["MSG_NEW"] + message_encrypted,
             nonce = our_next_strand_nonce
@@ -275,14 +274,14 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
         # /32 because KEM shared_secret is 32 bytes, /64 because sha3_512 output is 64 bytes
 
-        if len(msgs_plaintext) != ( (ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_F_CT_LEN) * (OTP_PAD_SIZE // 32)) + (64 * (OTP_PAD_SIZE // 64)) + ML_DSA_87_SIGN_LEN + 1:
+        if len(msgs_plaintext) != ( (ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_CT_LEN) * (OTP_PAD_SIZE // 32)) + (64 * (OTP_PAD_SIZE // 64)) + ML_DSA_87_SIGN_LEN + 1:
             logger.error("Contact (%s) gave us a otp batch message request with malformed strand plaintext length (%d)", contact_id, len(msgs_plaintext))
             return
 
         otp_hashchain_signature  = msgs_plaintext[1: ML_DSA_87_SIGN_LEN + 1]
-        otp_hashchain_ciphertext = msgs_plaintext[ML_DSA_87_SIGN_LEN + 1: ML_DSA_87_SIGN_LEN + 1 + ((ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_F_CT_LEN) * (OTP_PAD_SIZE // 32))]
+        otp_hashchain_ciphertext = msgs_plaintext[ML_DSA_87_SIGN_LEN + 1: ML_DSA_87_SIGN_LEN + 1 + ((ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_CT_LEN) * (OTP_PAD_SIZE // 32))]
 
-        xchacha_pads = msgs_plaintext[ML_DSA_87_SIGN_LEN + 1 + ((ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_F_CT_LEN) * (OTP_PAD_SIZE // 32)):]
+        chacha_pads = msgs_plaintext[ML_DSA_87_SIGN_LEN + 1 + ((ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_CT_LEN) * (OTP_PAD_SIZE // 32)):]
 
         try:
             valid_signature = verify_signature(ML_DSA_87_NAME, otp_hashchain_ciphertext, otp_hashchain_signature, contact_public_key)
@@ -294,7 +293,7 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
             return
 
         our_kyber_key = user_data_copied["contacts"][contact_id]["ephemeral_keys"]["our_keys"][ML_KEM_1024_NAME]["private_key"]
-        our_mceliece_key = user_data_copied["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_F_NAME]["private_key"]
+        our_mceliece_key = user_data_copied["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_NAME]["private_key"]
 
         try:
             contact_kyber_pads = decrypt_shared_secrets(otp_hashchain_ciphertext[:ML_KEM_1024_CT_LEN * (OTP_PAD_SIZE // 32)], our_kyber_key, ML_KEM_1024_NAME)
@@ -303,13 +302,13 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
             return
 
         try:
-            contact_mceliece_pads = decrypt_shared_secrets(otp_hashchain_ciphertext[ML_KEM_1024_CT_LEN * (OTP_PAD_SIZE // 32):], our_mceliece_key, CLASSIC_MCELIECE_8_F_NAME)
+            contact_mceliece_pads = decrypt_shared_secrets(otp_hashchain_ciphertext[ML_KEM_1024_CT_LEN * (OTP_PAD_SIZE // 32):], our_mceliece_key, CLASSIC_MCELIECE_8_NAME)
         except Exception as e:
             logger.error("Failed to decrypt Classic-McEliece8192128's ciphertext from contact (%s), received error: %s", contact_id, str(e))
             return
 
         contact_pads, _ = one_time_pad(contact_kyber_pads, contact_mceliece_pads)
-        contact_pads, _ = one_time_pad(contact_pads, xchacha_pads)
+        contact_pads, _ = one_time_pad(contact_pads, chacha_pads)
 
         contact_next_strand_key = contact_pads[:32]
         contact_pads = contact_pads[32:]
@@ -320,11 +319,11 @@ 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_F_NAME]["rotation_counter"] += 1
+            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"])
 
-            rotation_counter = user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_F_NAME]["rotation_counter"]
+            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)