Merge pull request #9 from Freedom-Club-Sec/feat/federation-support

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: " + str(e))
 
     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

@@ -2,6 +2,7 @@
 from logic.smp import smp_unanswered_questions, smp_data_handler
 from logic.pfs import pfs_data_handler, update_ephemeral_keys
 from logic.message import messages_data_handler
+from logic.user import validate_identifier
 from core.constants import (
     LONGPOLL_MIN,
     LONGPOLL_MAX,
@@ -86,12 +87,12 @@ 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 not validate_identifier(message["sender"]):
+                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 sender identifier (%s)...", message["sender"])
                 continue
 
             sender = message["sender"]
@@ -115,10 +116,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/get_user.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

logic/smp.py

@@ -8,3 +8,24 @@ def build_initial_user_data() -> dict:
                 "ignore_new_contacts_smp": False, 
             }
         }
+
+
+def validate_identifier(identifier) -> bool:
+    if identifier.isdigit() and len(identifier) == 16:
+        return True
+
+
+    split = identifier.split("@")
+    if len(split) != 2:
+        return False
+
+    if not split[0].isdigit():
+        return False
+
+    # Max domain length is 253 bytes
+    if len(split[1] > 253):
+        return False
+
+
+    return True
+

logic/user.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()
 

main.py

@@ -1,6 +1,6 @@
 from tkinter import messagebox
 from ui.utils import *
-from logic.get_user import check_if_contact_exists
+from logic.user import validate_identifier 
 from logic.contacts import save_contact
 from logic.storage import save_account_data
 import tkinter as tk
@@ -20,7 +20,7 @@ def __init__(self, master):
         self.entry = tk.Entry(self, font=("Helvetica", 12), bg="gray15", fg="white", insertbackground="white")
         self.entry.pack(pady=5)
         self.entry.focus()
-        enhanced_entry(self.entry, placeholder="I.e. 1234567890123456")
+        enhanced_entry(self.entry, placeholder="I.e. 1234567890123456, [email protected]")
 
         self.status = tk.Label(self, text="", fg="gray", bg="black", font=("Helvetica", 10))
         self.status.pack(pady=(5, 0))
@@ -42,28 +42,25 @@ 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
-
-            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)
-        except ValueError as e:
-            self.status.config(text=e, fg="red")
-            logger.error("Error occured while adding new contact (%s): %s ", contact_id, e)
+        if not validate_identifier(contact_id):
+            logger.debug("Identifier is invalid.")
+            self.status.config(text = "Invalid identifier", fg="red")
             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)
+        
 
         self.master.new_contact(contact_id)
         self.destroy()
-        messagebox.showinfo("Added", "Added the user to your contact list")
+        messagebox.showinfo("Added", f"Added the `{contact_id}` to your contact list.")