feat: federation support & SMP adjustments

Author: chadsec1

core/constants.py

@@ -10,6 +10,7 @@
 PFS_TYPE = b"\x01"
 MSG_TYPE = b"\x02"
 
+COLDWIRE_DATA_SEP   = b"\0"
 COLDWIRE_LEN_OFFSET = 3
 
 # network defaults (seconds & bytes)

core/crypto.py

@@ -14,7 +14,6 @@
 
 import oqs
 import secrets
-from typing import Tuple
 from core.constants import (
     OTP_PAD_SIZE,
     OTP_MAX_RANDOM_PAD,
@@ -62,7 +61,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 +76,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) -> Tuple[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.
 
@@ -151,7 +150,7 @@ def one_time_pad(plaintext: bytes, key: bytes) -> bytes:
     key = key[len(otpd_plaintext):]
     return otpd_plaintext, key
 
-def generate_kem_keys(algorithm: str) -> Tuple[bytes, bytes]:
+def generate_kem_keys(algorithm: str) -> tuple[bytes, bytes]:
     """
     Generates a KEM keypair.
 
@@ -166,7 +165,7 @@ def generate_kem_keys(algorithm: str) -> Tuple[bytes, bytes]:
         private_key = kem.export_secret_key()
         return private_key, public_key
 
-def encap_shared_secret(public_key: bytes, algorithm: str) -> Tuple[bytes, bytes]:
+def encap_shared_secret(public_key: bytes, algorithm: str) -> tuple[bytes, bytes]:
     """
     Derive a KEM shared secret from a public key.
 
@@ -226,7 +225,7 @@ def decrypt_shared_secrets(ciphertext_blob: bytes, private_key: bytes, algorithm
 
     return shared_secrets #[:otp_pad_size]
 
-def generate_shared_secrets(public_key: bytes, algorithm: str = None, size: int = OTP_PAD_SIZE) -> Tuple[bytes, bytes]:
+def generate_shared_secrets(public_key: bytes, algorithm: str = None, size: int = OTP_PAD_SIZE) -> tuple[bytes, bytes]:
     """
     Generates many shared secrets via `algorithm` encapsulation in chunks.
 

core/requests.py

@@ -99,7 +99,7 @@ def http_request(url: str, method: str, auth_token: str = None, metadata: dict =
             body = b""
 
             if metadata is not None:
-                body += encode_field("metadata", json.dumps({"metadata": metadata}), boundary, CRLF)
+                body += encode_field("metadata", json.dumps(metadata), boundary, CRLF)
 
             if blob is not None:
                 body += encode_file("blob", "blob.bin", blob, boundary, CRLF)

logic/authentication.py

@@ -44,8 +44,11 @@ def authenticate_account(user_data: dict) -> dict:
             raise ValueError("Could not connect to server! Are you sure your proxy settings are valid ?")
         else:
             raise ValueError("Could not connect to server! Are you sure the URL is valid ?")
-    
-    response = json.loads(response.decode())
+   
+    try:
+        response = json.loads(response.decode())
+    except Exception as e:
+        raise ValueError("Error while parsing server JSON response: ")
 
     if not 'challenge' in response:
         raise ValueError("Server did not give authenticatation challenge! Are you sure this is a Coldwire server ?")

logic/background_worker.py

@@ -86,13 +86,19 @@ def background_worker(user_data, user_data_lock, ui_queue, stop_flag):
             logger.debug("Received data: %s", str(message)[:3000])
 
             # Sanity check universal message fields
-            if (not "sender" in message) or (not message["sender"].isdigit()) or (len(message["sender"]) != 16):
-                logger.error("Impossible condition, either you have discovered a bug in Coldwire, or the server is attempting to denial-of-service you. Skipping data message with no (or malformed) sender...")
-
-                if "sender" in message:
-                    logger.debug("Impossible condition's sender is: %s", message["sender"])
+            if (not "sender" in message):
+                logger.error("Impossible condition, either you have discovered a bug in Coldwire, or the server is attempting to denial-of-service you. Skipping data message with no sender...")
+                continue
 
+            if message["sender"].isdigit() and len(message["sender"]) != 16:
+                logger.error("Impossible condition, either you have discovered a bug in Coldwire, or the server is attempting to denial-of-service you. Skipping data message with malformed same-server sender (%s)...", message["sender"])
                 continue
+        
+            if (not message["sender"].isdigit()): 
+                split = message["sender"].split("@")
+                if (len(split) != 2) or (not split[0].isdigit()):
+                    logger.error("Impossible condition, either you have discovered a bug in Coldwire, or the server is attempting to denial-of-service you. Skipping data message with malformed federated-server sender (%s)...", message["sender"])
+                    continue
 
             sender = message["sender"]
             blob   = message["blob"]
@@ -115,10 +121,10 @@ def background_worker(user_data, user_data_lock, ui_queue, stop_flag):
                         try:
                             blob_plaintext = decrypt_xchacha20poly1305(chacha_key, blob[:XCHACHA20POLY1305_NONCE_LEN], blob[XCHACHA20POLY1305_NONCE_LEN:])
                         except Exception as e:
-                            logger.debug("Failed to decrypt blob from contact (%s) probably due to invalid nonce: %s", sender, str(e))
                             if contact_next_strand_nonce is None:
                                 raise Exception("Unable to decrypt apparent SMP request due to missing contact strand nonce.")
 
+                            logger.debug("Failed to decrypt blob from contact (%s) probably due to invalid nonce: %s, we will try decrypting using strand nonce", sender, str(e))
                             blob_plaintext = decrypt_xchacha20poly1305(chacha_key, contact_next_strand_nonce, blob)
 
                     except Exception as e:

logic/smp.py

@@ -244,6 +244,10 @@ def smp_step_3(user_data: dict, user_data_lock: threading.Lock, contact_id: str,
 
 
 def smp_step_4_request_answer(user_data, user_data_lock, contact_id, smp_plaintext, ui_queue) -> None:
+    with user_data_lock:
+        our_nonce = b64decode(user_data["contacts"][contact_id]["lt_sign_key_smp"]["our_nonce"])
+
+
     contact_new_strand_nonce = smp_plaintext[:XCHACHA20POLY1305_NONCE_LEN]
 
     contact_signing_public_key = smp_plaintext[XCHACHA20POLY1305_NONCE_LEN : ML_DSA_87_PK_LEN + XCHACHA20POLY1305_NONCE_LEN]
@@ -254,6 +258,12 @@ def smp_step_4_request_answer(user_data, user_data_lock, contact_id, smp_plainte
 
     question      = smp_plaintext[SMP_NONCE_LENGTH + XCHACHA20POLY1305_NONCE_LEN + SMP_PROOF_LENGTH + ML_DSA_87_PK_LEN:].decode("utf-8")
 
+    if our_nonce == contact_nonce:
+        logger.warning("SMP Verification failed at step 4: Contact nonce is the same as our nonce!")
+        smp_failure_notify_contact(user_data, user_data_lock, contact_id, ui_queue)
+        return
+
+
 
     with user_data_lock: 
         user_data["contacts"][contact_id]["lt_sign_key_smp"]["question"] = question

main.py

@@ -25,7 +25,7 @@ def setup_logging(debug: bool) -> None:
     logger.addHandler(handler)
 
 def parse_args():
-    parser = argparse.ArgumentParser(description="Coldwire - Post-Quantum secure messenger")
+    parser = argparse.ArgumentParser(description="Coldwire - Ultra-Paranoid, Post-Quantum Secure Messenger")
     parser.add_argument("--debug", action="store_true", help="Enable debug logging")
     return parser.parse_args()
 

ui/add_contact_prompt.py

@@ -42,18 +42,19 @@ def __init__(self, master):
 
     def add_contact(self):
         contact_id = self.entry.get().strip()
-        if not (contact_id.isdigit() and len(contact_id) == 16):
+        """if not (contact_id.isdigit() and len(contact_id) == 16):
             self.status.config(text="Invalid User ID", fg="red")
             return
+        """
 
         if contact_id == self.master.user_data["user_id"]:
             self.status.config(text="You cannot add yourself", fg="red")
             return
 
         try:
-            if not check_if_contact_exists(self.master.user_data, self.master.user_data_lock, contact_id):
-                logger.error("[BUG] This should never execute, because the server should return a 40X error code and that should cause an exception..")
-                return
+            # if not check_if_contact_exists(self.master.user_data, self.master.user_data_lock, contact_id):
+            #    logger.error("[BUG] This should never execute, because the server should return a 40X error code and that should cause an exception..")
+            #    return
 
             save_contact(self.master.user_data, self.master.user_data_lock, contact_id)
             save_account_data(self.master.user_data, self.master.user_data_lock)

ui/connect_window.py

@@ -1,4 +1,6 @@
-from ui.utils import *
+from ui.utils import (
+        enhanced_entry
+)
 from ui.password_window import PasswordWindow
 from logic.storage import save_account_data
 from logic.authentication import authenticate_account

ui/password_window.py

@@ -1,6 +1,8 @@
 import tkinter as tk
 from tkinter import messagebox
-from ui.utils import *
+from ui.utils import (
+        enhanced_entry
+)
 
 class PasswordWindow(tk.Toplevel):
     def __init__(self, master, callback):