fix: update shared constants to reflect new NIST sizes

Author: chadsec1

core/constants.py

@@ -19,7 +19,11 @@
 SMP_NONCE_LENGTH      = 64
 SMP_PROOF_LENGTH      = 64
 SMP_ANSWER_OUTPUT_LEN = 64
-SMP_QUESTION_MAX_LEN  = 500
+SMP_QUESTION_MAX_LEN  = 512
+
+KEYS_HASH_CHAIN_LEN    = 64
+MESSAGE_HASH_CHAIN_LEN = 64
+
 
 # NIST-specified key sizes (bytes) and metadata
 ML_KEM_1024_NAME   = "ML-KEM-1024"
@@ -29,18 +33,17 @@
 
 
 ML_DSA_87_NAME     = "ML-DSA-87"  
-ML_DSA_87_SK_LEN   = 4864
+ML_DSA_87_SK_LEN   = 4896
 ML_DSA_87_PK_LEN   = 2592
-ML_DSA_87_SIGN_LEN = 4595
-
+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_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_F_ROTATE_AT = 3 # Default OTP batches needed to be sent for a key rotation to occur
+CLASSIC_MCELIECE_8_F_ROTATE_AT = 3   # Default OTP batches needed to be sent for a key rotation to occur
 
 
 

core/requests.py

@@ -1,5 +1,9 @@
-from urllib import request
+from urllib import request, error
+import urllib
 import json
+import logging
+
+logger = logging.getLogger(__name__)
 
 _ORIGINAL_SOCKET = None
 
@@ -66,10 +70,15 @@ def http_request(url: str, method: str, auth_token: str = None, payload: dict =
         req.add_header("Authorization", "Bearer " + auth_token)
 
     # NOTE: urllib raises a HTTPError for status code >= 400
-    if longpoll == -1:
-        with request.urlopen(req) as response:
-            return json.loads(response.read().decode())
-    else:
-        with request.urlopen(req, timeout=longpoll) as response:
-            return json.loads(response.read().decode())
 
+    try:
+        if longpoll == -1:
+            with request.urlopen(req) as response:
+                return json.loads(response.read().decode())
+        else:
+            with request.urlopen(req, timeout=longpoll) as response:
+                return json.loads(response.read().decode())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode()
+        logger.error("We received error from server: %s", body)
+        raise Exception(body)

logic/pfs.py

@@ -11,7 +11,8 @@
     ML_KEM_1024_NAME,
     ML_DSA_87_NAME,
     CLASSIC_MCELIECE_8_F_NAME,
-    CLASSIC_MCELIECE_8_F_ROTATE_AT
+    CLASSIC_MCELIECE_8_F_ROTATE_AT,
+    KEYS_HASH_CHAIN_LEN
 )
 from core.trad_crypto import sha3_512
 from base64 import b64encode, b64decode
@@ -39,7 +40,7 @@ def send_new_ephemeral_keys(user_data, user_data_lock, contact_id, ui_queue) ->
     if not user_data_copied["contacts"][contact_id]["lt_sign_keys"]["our_hash_chain"]:
         with user_data_lock:
             # Set up the hash chain's initial seed
-            user_data["contacts"][contact_id]["lt_sign_keys"]["our_hash_chain"] = secrets.token_bytes(64)
+            user_data["contacts"][contact_id]["lt_sign_keys"]["our_hash_chain"] = secrets.token_bytes(KEYS_HASH_CHAIN_LEN)
 
             our_hash_chain = user_data["contacts"][contact_id]["lt_sign_keys"]["our_hash_chain"] 
     else:
@@ -157,7 +158,7 @@ def pfs_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, mess
         logger.error("contact (%s) sent message of unknown pfs_type (%s)", contact_id, message["pfs_type"])
         return
 
-    contact_hash_chain = contact_publickeys_hashchain[:64]
+    contact_hash_chain = contact_publickeys_hashchain[:KEYS_HASH_CHAIN_LEN]
 
     # If we do not have a hashchain for the contact, we don't need to compute the chain, just save.
     if not user_data_copied["contacts"][contact_id]["lt_sign_keys"]["contact_hash_chain"]:
@@ -172,11 +173,11 @@ def pfs_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, mess
             logger.error("Contact hash chain does not match our computed hash chain, we are skipping this PFS message...")
             return
 
-    contact_kyber_public_key = contact_publickeys_hashchain[64: ALGOS_BUFFER_LIMITS[ML_KEM_1024_NAME]["PK_LEN"] + 64]
+    contact_kyber_public_key = contact_publickeys_hashchain[KEYS_HASH_CHAIN_LEN: ALGOS_BUFFER_LIMITS[ML_KEM_1024_NAME]["PK_LEN"] + KEYS_HASH_CHAIN_LEN]
     if message["pfs_type"] == "full":
         logger.info("contact (%s) has rotated their Kyber and McEliece keys", contact_id)
 
-        contact_mceliece_public_key = contact_publickeys_hashchain[ALGOS_BUFFER_LIMITS[ML_KEM_1024_NAME]["PK_LEN"] + 64:]
+        contact_mceliece_public_key = contact_publickeys_hashchain[ALGOS_BUFFER_LIMITS[ML_KEM_1024_NAME]["PK_LEN"] + KEYS_HASH_CHAIN_LEN:]
         with user_data_lock:
             user_data["contacts"][contact_id]["ephemeral_keys"]["contact_public_keys"][CLASSIC_MCELIECE_8_F_NAME] = contact_mceliece_public_key
 

logic/smp.py

@@ -245,7 +245,7 @@ def smp_step_4(user_data, user_data_lock, contact_id, message, ui_queue) -> None
 
     # Derieve a high-entropy secret key from the low-entropy answer 
     argon2id_salt = sha3_512(our_nonce + contact_nonce)
-    answer_secret, _ = derive_key_argon2id(answer.encode(), salt=argon2id_salt, output_length=64)
+    answer_secret, _ = derive_key_argon2id(answer.encode(), salt=argon2id_salt, output_length=SMP_ANSWER_OUTPUT_LEN)
 
     # Compute the proof
     our_message = our_nonce + contact_nonce + our_key_fingerprint