feat: add new availability features

Author: chadsec1

.gitignore

@@ -1,5 +1,6 @@
 __pycache__/
 account.coldwire
 *.swp
+*.swo
 TODO.md
 TODO.txt

core/constants.py

@@ -6,9 +6,18 @@
 ACCOUNT_FILE_PATH = "account.coldwire"
 
 # Coldwire protocol misc (bytes)
-SMP_TYPE = b"\x00"
-PFS_TYPE = b"\x01"
-MSG_TYPE = b"\x02"
+SMP_TYPES = {
+        "SMP_INIT": b"\x00"
+}
+PFS_TYPES = {
+        "PFS_NEW": b"\x01",
+        "PFS_ACK": b"\x02",
+}
+
+MSG_TYPES = {
+        "MSG_BATCH": b"\x03",
+        "MSG_NEW": b"\x04"
+}
 
 COLDWIRE_DATA_SEP   = b"\0"
 COLDWIRE_LEN_OFFSET = 3

logic/background_worker.py

@@ -7,9 +7,9 @@
     LONGPOLL_MIN,
     LONGPOLL_MAX,
     COLDWIRE_LEN_OFFSET,
-    SMP_TYPE,
-    PFS_TYPE,
-    MSG_TYPE,
+    SMP_TYPES,
+    PFS_TYPES,
+    MSG_TYPES,
     XCHACHA20POLY1305_NONCE_LEN
 )
 from core.crypto import random_number_range
@@ -162,16 +162,16 @@ def background_worker(user_data, user_data_lock, ui_queue, stop_flag):
 
 
             # SMP
-            if bytes([blob_plaintext[0]]) == SMP_TYPE:
-                smp_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext[1:])
+            if bytes([blob_plaintext[0]]) in SMP_TYPES.values():
+                smp_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext)
 
             # PFS
-            elif bytes([blob_plaintext[0]]) == PFS_TYPE:
-                pfs_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext[1:])
+            elif bytes([blob_plaintext[0]]) in PFS_TYPES.values():
+                pfs_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext)
 
             # MSG
-            elif bytes([blob_plaintext[0]]) == MSG_TYPE:
-                messages_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext[1:])
+            elif bytes([blob_plaintext[0]]) in MSG_TYPES.values():
+                messages_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, sender, blob_plaintext)
 
             else:
                 logger.error(

logic/message.py

@@ -28,7 +28,7 @@
     otp_decrypt_with_padding
 )
 from core.constants import ( 
-    MSG_TYPE,
+    MSG_TYPES,
     OTP_MAX_BUCKET,
     OTP_PAD_SIZE,
     ML_KEM_1024_NAME,
@@ -81,7 +81,7 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
     our_new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             our_strand_key, 
-            MSG_TYPE + b"\x00" + our_new_strand_nonce + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob + xchacha_shared_secrets,
+            MSG_TYPES["MSG_BATCH"] + our_new_strand_nonce + otp_batch_signature + kyber_ciphertext_blob + mceliece_ciphertext_blob + xchacha_shared_secrets,
             nonce = our_next_strand_nonce
         )
 
@@ -203,7 +203,7 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
 
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             our_strand_key, 
-            MSG_TYPE + b"\x01" + our_new_strand_nonce + message_encrypted,
+            MSG_TYPES["MSG_NEW"] + our_new_strand_nonce + message_encrypted,
             nonce = our_next_strand_nonce
         )
 
@@ -257,7 +257,7 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
 
 
-    if msgs_plaintext[0] == 0:
+    if bytes([msgs_plaintext[0]]) == MSG_TYPES["MSG_BATCH"]:
         logger.debug("Received a new OTP pads batch from contact (%s).", contact_id)
 
         # /32 because KEM shared_secret is 32 bytes, /64 because sha3_512 output is 64 bytes
@@ -328,7 +328,7 @@ def messages_data_handler(user_data: dict, user_data_lock, user_data_copied: dic
 
 
 
-    elif msgs_plaintext[0] == 1:
+    elif bytes([msgs_plaintext[0]]) == MSG_TYPES["MSG_NEW"]:
         logger.debug("Received a new message from contact (%s).", contact_id)
 
         if len(msgs_plaintext) < OTP_MAX_BUCKET + XCHACHA20POLY1305_NONCE_LEN + 1:

logic/pfs.py

@@ -19,7 +19,7 @@
     random_number_range
 )
 from core.constants import (
-    PFS_TYPE,
+    PFS_TYPES,
     ML_KEM_1024_NAME,
     ML_DSA_87_NAME,
     ML_KEM_1024_PK_LEN,
@@ -60,7 +60,7 @@ def send_pfs_ack(user_data: dict, user_data_lock: threading.Lock, contact_id: st
     our_new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             our_strand_key, 
-            PFS_TYPE + b"\x01" + our_new_strand_nonce,
+            PFS_TYPES["PFS_ACK"] + our_new_strand_nonce,
             nonce = our_next_strand_nonce
         )
 
@@ -148,7 +148,7 @@ def send_new_ephemeral_keys(user_data: dict, user_data_lock: threading.Lock, con
     our_new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             our_strand_key, 
-            PFS_TYPE + b"\x00" + our_new_strand_nonce + publickeys_hashchain_signature + publickeys_hashchain,
+            PFS_TYPES["PFS_NEW"] + our_new_strand_nonce + publickeys_hashchain_signature + publickeys_hashchain,
             nonce = our_next_strand_nonce,
             max_padding = 1024
         )
@@ -234,15 +234,15 @@ def pfs_data_handler(user_data: dict, user_data_lock: threading.Lock, user_data_
         logger.error("Contact (%s) strand key key is missing! Skipping message...", contact_id)
         return 
 
-    if pfs_plaintext[0] == 1:
+    if bytes([pfs_plaintext[0]]) == PFS_TYPES["PFS_ACK"]:
         logger.info("Received acknowlegement of PFS keys from contact %s", contact_id)
         with user_data_lock:
             user_data["contacts"][contact_id]["contact_next_strand_nonce"] = pfs_plaintext[1:]
 
             user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][ML_KEM_1024_NAME]["private_key"] = user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][ML_KEM_1024_NAME]["private_key"]
             user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][ML_KEM_1024_NAME]["public_key"] = user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][ML_KEM_1024_NAME]["public_key"]
 
-            if user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][ML_KEM_1024_NAME]["private_key"]:
+            if user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_F_NAME]["private_key"]:
                 user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_F_NAME]["private_key"] = user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_F_NAME]["private_key"]
                 user_data["contacts"][contact_id]["ephemeral_keys"]["our_keys"][CLASSIC_MCELIECE_8_F_NAME]["public_key"] = user_data["contacts"][contact_id]["ephemeral_keys"]["staged_keys"][CLASSIC_MCELIECE_8_F_NAME]["public_key"]
 
@@ -255,7 +255,7 @@ def pfs_data_handler(user_data: dict, user_data_lock: threading.Lock, user_data_
         save_account_data(user_data, user_data_lock)
         return
 
-    elif pfs_plaintext[0] == 0:
+    elif bytes([pfs_plaintext[0]]) == PFS_TYPES["PFS_NEW"]:
         pfs_plaintext = pfs_plaintext[1:]
     else:
         logger.error("Skipping unknown PFS of type (%d) from contact (%s)", pfs_plaintext[0], contact_id)

logic/smp.py

@@ -30,7 +30,7 @@
 )
 from base64 import b64encode, b64decode
 from core.constants import (
-        SMP_TYPE,
+        SMP_TYPES,
         SMP_NONCE_LENGTH,
         SMP_PROOF_LENGTH,
         SMP_QUESTION_MAX_LEN,
@@ -75,7 +75,7 @@ def initiate_smp(user_data: dict, user_data_lock: threading.Lock, contact_id: st
                 "recipient": contact_id
             }, 
             headers = session_headers, 
-            blob = SMP_TYPE + kem_public_key, 
+            blob = SMP_TYPES["SMP_INIT"] + kem_public_key, 
             auth_token = auth_token
         )
     except Exception as e:
@@ -136,7 +136,7 @@ def smp_step_2(user_data: dict, user_data_lock, contact_id: str, blob: bytes, ui
         http_request(f"{server_url}/data/send", "POST", metadata = {
                 "recipient": contact_id
             }, 
-            blob = SMP_TYPE + key_ciphertext + ciphertext_nonce + ciphertext_blob, 
+            blob = SMP_TYPES["SMP_INIT"] + key_ciphertext + ciphertext_nonce + ciphertext_blob, 
             headers = session_headers, 
             auth_token = auth_token
         )
@@ -215,7 +215,7 @@ def smp_step_3(user_data: dict, user_data_lock: threading.Lock, contact_id: str,
     our_new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             chacha_key, 
-            SMP_TYPE + our_new_strand_nonce + signing_public_key + our_nonce + our_proof + question.encode("utf-8"),
+            SMP_TYPES["SMP_INIT"] + our_new_strand_nonce + signing_public_key + our_nonce + our_proof + question.encode("utf-8"),
             nonce = our_next_strand_nonce
         )
 
@@ -348,7 +348,7 @@ def smp_step_4_answer_provided(user_data, user_data_lock, contact_id, answer, ui
     our_new_strand_nonce = sha3_512(secrets.token_bytes(XCHACHA20POLY1305_NONCE_LEN))[:XCHACHA20POLY1305_NONCE_LEN]
     _, ciphertext_blob = encrypt_xchacha20poly1305(
             tmp_key, 
-            SMP_TYPE + our_new_strand_nonce + our_proof + our_strand_key + contact_strand_key,
+            SMP_TYPES["SMP_INIT"] + our_new_strand_nonce + our_proof + our_strand_key + contact_strand_key,
             nonce = our_next_strand_nonce
         )
 
@@ -513,7 +513,7 @@ def smp_failure_notify_contact(user_data, user_data_lock, contact_id, ui_queue)
     # it can be any number other than 2, we chose 7 because failure is *technically* the 7th smp step.
     ciphertext_nonce, ciphertext_blob = encrypt_xchacha20poly1305(
             tmp_key, 
-            SMP_TYPE + b"failure", 
+            SMP_TYPES["SMP_INIT"] + b"failure", 
             counter = 7
         )
     try:
@@ -550,12 +550,14 @@ def smp_data_handler(user_data, user_data_lock, user_data_copied, ui_queue, cont
     except Exception:
         smp_step = 2
 
-
 
     if user_data_copied["settings"]["ignore_new_contacts_smp"]:
         logger.info("Skipping SMP request because you have set to ignore new contacts requests.")
         return
 
+    
+    # TEMPORARIY, UNTIL WE REWORK SMP STEP HANDLING
+    message = message[1:]
 
 
     if message == b"failure":