feat: add strandlock protocol for messages

Author: chadsec1

core/crypto.py

@@ -62,7 +62,7 @@ def verify_signature(algorithm: str, message: bytes, signature: bytes, public_ke
     with oqs.Signature(algorithm) as verifier:
         return verifier.verify(message, signature[:ALGOS_BUFFER_LIMITS[algorithm]["SIGN_LEN"]], public_key[:ALGOS_BUFFER_LIMITS[algorithm]["PK_LEN"]])
 
-def generate_sign_keys(algorithm: str = ML_DSA_87_NAME) Tuple[bytes, bytes]:
+def generate_sign_keys(algorithm: str = ML_DSA_87_NAME) -> Tuple[bytes, bytes]:
     """
     Generates a new post-quantum signature keypair.
 
@@ -77,7 +77,7 @@ def generate_sign_keys(algorithm: str = ML_DSA_87_NAME) Tuple[bytes, bytes]:
         private_key = signer.export_secret_key()
         return private_key, public_key
 
-def otp_encrypt_with_padding(plaintext: bytes, key: bytes) -> bytes:
+def otp_encrypt_with_padding(plaintext: bytes, key: bytes) -> Tuple[bytes, bytes]:
     """
     Encrypts plaintext using a one-time pad with random or bucket padding.
 
@@ -98,7 +98,7 @@ def otp_encrypt_with_padding(plaintext: bytes, key: bytes) -> bytes:
     if len(plaintext) < OTP_MAX_BUCKET:
         pad_len = OTP_MAX_BUCKET - len(plaintext)
     else:
-        pad_len = secrets.randbelow(OTP_MAX_RANDOM_PAD)
+        pad_len = secrets.randbelow(OTP_MAX_RANDOM_PAD + 1)
 
     padding = secrets.token_bytes(pad_len)
 
@@ -107,7 +107,7 @@ def otp_encrypt_with_padding(plaintext: bytes, key: bytes) -> bytes:
     padded_plaintext = plaintext_length_bytes + plaintext + padding
 
     if len(padded_plaintext) > len(key):
-        raise ValueError("Plaintext is larger than key!")
+        raise ValueError("Padded plaintext is larger than key!")
 
     return one_time_pad(padded_plaintext, key)
 
@@ -122,7 +122,7 @@ def otp_decrypt_with_padding(ciphertext: bytes, key: bytes) -> bytes:
     Returns:
         Original plaintext bytes without padding.
     """
-    plaintext_with_padding = one_time_pad(ciphertext, key)
+    plaintext_with_padding, _ = one_time_pad(ciphertext, key)
 
     plaintext_length = int.from_bytes(plaintext_with_padding[:OTP_SIZE_LENGTH], "big")
 
@@ -147,7 +147,9 @@ def one_time_pad(plaintext: bytes, key: bytes) -> bytes:
     for index, plain_byte in enumerate(plaintext):
         key_byte = key[index]
         otpd_plaintext += bytes([plain_byte ^ key_byte])
-    return otpd_plaintext
+
+    key = key[len(otpd_plaintext):]
+    return otpd_plaintext, key
 
 def generate_kem_keys(algorithm: str) -> Tuple[bytes, bytes]:
     """

logic/message.py

@@ -13,7 +13,11 @@
 from core.requests import http_request
 from logic.storage import save_account_data
 from logic.pfs import send_new_ephemeral_keys
-from core.trad_crypto import sha3_512
+from core.trad_crypto import (
+        sha3_512,
+        encrypt_xchacha20poly1305,
+        decrypt_xchacha20poly1305
+)
 from core.crypto import (
     generate_shared_secrets,
     decrypt_shared_secrets,
@@ -24,18 +28,21 @@
     otp_decrypt_with_padding
 )
 from core.constants import ( 
-    KEYS_HASH_CHAIN_LEN,
+    MESSAGE_HASH_CHAIN_LEN,
+    OTP_MAX_BUCKET,
     OTP_PAD_SIZE,
     OTP_SIZE_LENGTH,
     ML_KEM_1024_NAME,
     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
+    CLASSIC_MCELIECE_8_F_CT_LEN,
+    XCHACHA20POLY1305_NONCE_LEN
+
 )
 from base64 import b64decode, b64encode
-import json
+import secrets
 import logging
 
 logger = logging.getLogger(__name__)
@@ -65,9 +72,10 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
 
     otp_batch_signature = create_signature(ML_DSA_87_NAME, kyber_ciphertext_blob + mceliece_ciphertext_blob, our_lt_private_key)
 
+    hash_chain_seed = secrets.token_bytes(MESSAGE_HASH_CHAIN_LEN)
     ciphertext_nonce, ciphertext_blob = encrypt_xchacha20poly1305(
             our_strand_key, 
-            b"\x00" + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob
+            b"\x00" + hash_chain_seed + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob
         )
 
 
@@ -80,12 +88,14 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
         ui_queue.put({"type": "showerror", "title": "Error", "message": "Failed to send our one-time-pads key batch to the server"})
         return False
 
-    pads = one_time_pad(kyber_shared_secrets, mceliece_shared_secrets)
+    pads, _ = one_time_pad(kyber_shared_secrets, mceliece_shared_secrets)
 
     # We update & save only at the end, so if request fails, we do not desync our state.
     with user_data_lock:
-        user_data["contacts"][contact_id]["our_pads"]["pads"]       = pads[64:]
-        user_data["contacts"][contact_id]["our_pads"]["hash_chain"] = pads[:64]
+        user_data["contacts"][contact_id]["our_strand_key"]         = pads[:32]
+        user_data["contacts"][contact_id]["our_pads"]["pads"]       = pads[32:]
+
+        user_data["contacts"][contact_id]["our_pads"]["hash_chain"] = hash_chain_seed
 
     save_account_data(user_data, user_data_lock)
 
@@ -111,9 +121,10 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
         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]
 
-        our_pads         = user_data["contacts"][contact_id]["our_pads"]["pads"]
+        our_pads = user_data["contacts"][contact_id]["our_pads"]["pads"]
 
-     
+
+
     if contact_kyber_public_key is None or contact_mceliece_public_key is None:
         logger.debug("This shouldn't happen, contact ephemeral keys are not initialized even once yet???")
         ui_queue.put({
@@ -141,53 +152,54 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
         our_hash_chain  = user_data["contacts"][contact_id]["our_pads"]["hash_chain"]
 
 
-    message_encoded = message.encode("utf-8")
-    next_hash_chain = sha3_512(our_hash_chain + message_encoded)
-    message_encoded = next_hash_chain + message_encoded
-
-    message_otp_padding_length = max(0, OTP_PADDING_LIMIT - OTP_PADDING_LENGTH - len(message_encoded))
-
-    if (len(message_encoded) + OTP_PADDING_LENGTH + message_otp_padding_length) > len(our_pads):
-        logger.info("Your message size (%d)  is larger than our pads size (%s), therefore we are generating new pads for you", len(message_encoded) + OTP_PADDING_LENGTH + message_otp_padding_length, len(our_pads))
-        
-        if not generate_and_send_pads(user_data, user_data_lock, contact_id, ui_queue):
-            return False
-
-        with user_data_lock:
-            our_pads        = user_data["contacts"][contact_id]["our_pads"]["pads"]
-            our_hash_chain  = user_data["contacts"][contact_id]["our_pads"]["hash_chain"]
-        
-        # We remove old hashchain from message and calculate new next hash in the chain
-        message_encoded = message_encoded[64:]
-        next_hash_chain = sha3_512(our_hash_chain + message_encoded)
-        message_encoded = next_hash_chain + message_encoded
-
 
-    message_otp_pad = our_pads[:len(message_encoded) + OTP_PADDING_LENGTH + message_otp_padding_length]
+    while True:
+        message_encoded = message.encode("utf-8")
+        try:
+            # We one-time-pad encrypt the message with padding
+            #
+            # NOTE: The padding only protects short-messages which are easy to infer what is said based purely on message length 
+            # With messages larger than padding_limit, we assume the message entropy give enough security to make an adversary assumption
+            # of message context (almost) useless. 
+            #
+            message_encrypted, new_pads = otp_encrypt_with_padding(message_encoded, our_pads)
+            logger.debug("Our old pad size is %d and new size after the message is %d", len(our_pads), len(new_pads))
+            break
+        except ValueError as e:
+            logger.debug("Failed to encrypt message to contact (%s) with error: %s", contact_id, str(e))
+            logger.info("Your message size (%d) when padded, is larger than our pads size (%s), therefore we are generating new pads for you", len(message), len(our_pads))
+            
+            if not generate_and_send_pads(user_data, user_data_lock, contact_id, ui_queue):
+                return False
 
-    logger.debug("Our pad size is %d and new size after the message is %d", len(our_pads), len(our_pads) - len(message_otp_pad))
+            with user_data_lock:
+                our_pads        = user_data["contacts"][contact_id]["our_pads"]["pads"]
+                our_hash_chain  = user_data["contacts"][contact_id]["our_pads"]["hash_chain"]
+            
 
-    # We one-time-pad encrypt the message with padding
-    #
-    # NOTE: The padding only protects short-messages which are easy to infer what is said based purely on message length 
-    # With messages larger than padding_limit, we assume the message entropy give enough security to make an adversary assumption
-    # of message context (almost) useless. 
-    #
-    message_encrypted = otp_encrypt_with_padding(message_encoded, message_otp_pad, padding_limit = message_otp_padding_length)
-    message_encrypted = b64encode(message_encrypted).decode()
 
     # Unlike in other functions, we truncate pads here and compute the next hash chain regardless of request being successful or not
     # because a malicious server could make our requests fail to force us to re-use the same pad for our next message 
     # which would break all of our security
+
+    next_hash_chain = sha3_512(our_hash_chain + message_encrypted)
+    
     with user_data_lock:
-        user_data["contacts"][contact_id]["our_pads"]["pads"]       = user_data["contacts"][contact_id]["our_pads"]["pads"][len(message_encoded) + OTP_PADDING_LENGTH + message_otp_padding_length:]
+        user_data["contacts"][contact_id]["our_pads"]["pads"]       = user_data["contacts"][contact_id]["our_pads"]["pads"][len(message_encrypted):]
         user_data["contacts"][contact_id]["our_pads"]["hash_chain"] = next_hash_chain
 
+        our_strand_key = user_data["contacts"][contact_id]["our_strand_key"]
+
     save_account_data(user_data, user_data_lock)
+
+    ciphertext_nonce, ciphertext_blob = encrypt_xchacha20poly1305(
+            our_strand_key, 
+            b"\x01" + next_hash_chain + message_encrypted
+        )
 
     try:
-        http_request(f"{server_url}/messages/send_message", "POST", payload = {
-                    "message_encrypted": message_encrypted,
+        http_request(f"{server_url}/messages/send", "POST", payload = {
+                    "ciphertext_blob": b64encode(ciphertext_nonce + ciphertext_blob).decode(),
                     "recipient": contact_id
                 }, 
                 auth_token=auth_token
@@ -243,17 +255,18 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
         logger.error("Failed to decrypt `ciphertext_blob` from contact (%s) with error: %s", contact_id, str(e))
         return
 
-    # b"\x00" + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob
 
     if msgs_plaintext[0] == 0:
         logger.debug("Received a new OTP pads batch from contact (%s).", contact_id)
 
-        if len(msgs_plaintext) != ( (ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_F_CT_LEN) * (OTP_PAD_SIZE // 32)) + ML_DSA_87_SIGN_LEN + 1:
-            logger.error("Contact (%s) gave us a message request with malformed strand plaintext length (%d)", contact_id, len(msgss_plaintext))
+        if len(msgs_plaintext) != ( (ML_KEM_1024_CT_LEN + CLASSIC_MCELIECE_8_F_CT_LEN) * (OTP_PAD_SIZE // 32)) + ML_DSA_87_SIGN_LEN + MESSAGE_HASH_CHAIN_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[:ML_DSA_87_SIGN_LEN]
-        otp_hashchain_ciphertext = msgs_plaintext[ML_DSA_87_SIGN_LEN:]
+        otp_hashchain_signature  = msgs_plaintext[1 + MESSAGE_HASH_CHAIN_LEN : MESSAGE_HASH_CHAIN_LEN + ML_DSA_87_SIGN_LEN + 1]
+        otp_hashchain_ciphertext = msgs_plaintext[ML_DSA_87_SIGN_LEN + MESSAGE_HASH_CHAIN_LEN + 1:]
+
+        contact_hash_chain = msgs_plaintext[1 : MESSAGE_HASH_CHAIN_LEN + 1]
 
         try:
             valid_signature = verify_signature(ML_DSA_87_NAME, otp_hashchain_ciphertext, otp_hashchain_signature, contact_public_key)
@@ -280,10 +293,10 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
             logger.error("Failed to decrypt McEliece's shared_secrets 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_kyber_pads, contact_mceliece_pads)
         contact_strand_key = contact_pads[:32]
-        contact_hash_chain = contact_pads[32:32 + KEY_HASH_CHAIN_LEN]
-        contact_pads = contact_pads[32 + KEY_HASH_CHAIN_LEN:]
+        contact_pads = contact_pads[32:]
+
 
         with user_data_lock:
             user_data["contacts"][contact_id]["contact_pads"]["pads"]       = contact_pads
@@ -297,7 +310,7 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
             rotation_counter = user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_F_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)
@@ -311,13 +324,30 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
 
 
-    elif message["msg_type"] == "new_message":
-        message_encrypted = b64decode(message["message_encrypted"], validate=True)
+    elif msgs_plaintext[0] == 1:
+        logger.debug("Received a new message from contact (%s).", contact_id)
+
+        if len(msgs_plaintext) < OTP_MAX_BUCKET + MESSAGE_HASH_CHAIN_LEN + 1:
+            logger.error("Contact (%s) gave us a message request with malformed strand plaintext length (%d)", contact_id, len(msgs_plaintext))
+            return
+
+
+        hash_chain        = msgs_plaintext[1:MESSAGE_HASH_CHAIN_LEN + 1]
+        message_encrypted = msgs_plaintext[MESSAGE_HASH_CHAIN_LEN + 1:]
 
+        
         with user_data_lock:
             contact_pads       = user_data["contacts"][contact_id]["contact_pads"]["pads"]
             contact_hash_chain = user_data["contacts"][contact_id]["contact_pads"]["hash_chain"]
 
+        
+        next_hash_chain = sha3_512(contact_hash_chain + message_encrypted)
+       
+        if next_hash_chain != hash_chain:
+            logger.warning("Message hash chain did not match, this could be a possible replay attack, or a failed tampering attempt. Skipping this message...")
+            return
+
+
         if (not contact_pads) or (len(message_encrypted) > len(contact_pads)):
             # TODO: Maybe reset our local pads as well?
             # I feel like we should do something more when we hit this case, but I am not sure.
@@ -328,15 +358,6 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
         # immediately truncate the pads
         contact_pads = contact_pads[len(message_encrypted):] 
 
-        hash_chain        = message_decrypted[:64] 
-        message_decrypted = message_decrypted[64:]
-
-        next_hash_chain = sha3_512(contact_hash_chain + message_decrypted)
-
-        if next_hash_chain != hash_chain:
-            logger.warning("Message hash chain did not match, this could be a possible replay attack, or a failed tampering attempt. Skipping this message...")
-            return
-
 
         # and save the new pads and the hash chain
         with user_data_lock:
@@ -358,3 +379,6 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
             "contact_id": contact_id,
             "message": message_decoded
         })
+
+    else:
+        logger.error("Received unknown message type (%d)", msgs_plaintext[0])