refactor: Remove unused variables and imports

Author: chadsec1

core/crypto.py

@@ -20,7 +20,6 @@
 import secrets
 from core.constants import (
     OTP_PAD_SIZE,
-    OTP_PADDING_LIMIT,
     OTP_PADDING_LENGTH,
     ML_KEM_1024_NAME,
     ML_KEM_1024_SK_LEN,

core/requests.py

@@ -1,7 +1,5 @@
 from urllib import request
-import urllib
 import json
-import time
 
 def http_request(url: str, method: str, auth_token: str = None, payload: dict = None, longpoll: int = -1) -> dict:
     if payload:
@@ -16,15 +14,11 @@ def http_request(url: str, method: str, auth_token: str = None, payload: dict =
     if auth_token:
         req.add_header("Authorization", "Bearer " + auth_token)
 
-    # urllib raises a HTTPError for status code >= 400
-    # try:
+    # 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())
 
-
-    # except urllib.error.HTTPError as e:
-    #    return json.loads(e.read().decode())

logic/authentication.py

@@ -15,23 +15,23 @@ def authenticate_account(user_data: dict) -> dict:
         response = http_request(url + "/authenticate/init", "POST", payload = {"public_key": public_key_encoded, "user_id": user_id })
         if not 'challenge' in response:
             raise ValueError("Server did not give authenticatation challenge! Are you sure this is a Coldwire server ?")
-    except:
+    except Exception:
         if 'proxy_info' in user_data:
             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 ?")
 
     try:
         challenge = b64decode(response["challenge"], validate=True)
-    except:
+    except Exception:
         raise ValueError("Server gave a malformed challenge! Are you sure this is Coldwire server ?")
 
 
     signature = create_signature("Dilithium5", challenge, private_key)
 
     try:
         response = http_request(url + "/authenticate/verify", "POST", payload = {"signature": b64encode(signature).decode(), "challenge": response["challenge"]})
-    except:
+    except Exception:
         raise ValueError("Server gave a malformed response, your account is probably missing from the server")
 
     required_keys = ["status", "user_id", "token"]

logic/background_worker.py

@@ -1,6 +1,4 @@
 from core.requests import http_request
-from logic.storage import save_account_data
-from logic.contacts import save_contact
 from logic.smp import smp_unanswered_questions, smp_data_handler
 from logic.pfs import pfs_data_handler
 from logic.message import messages_data_handler
@@ -9,12 +7,6 @@
     LONGPOLL_MAX
 )
 from core.crypto import random_number_range
-from core.trad_crypto import derive_key_argon2id, sha3_512
-from base64 import b64encode, b64decode
-import hashlib
-import secrets
-import hmac
-import time
 import copy
 import logging
 import json

logic/contacts.py

@@ -2,7 +2,6 @@
 import string
 import json
 import math
-import copy
 
 
 def generate_nickname_id(length: int = 4) -> str:

logic/message.py

@@ -28,7 +28,6 @@
     ML_DSA_87_NAME,  
 )
 from base64 import b64decode, b64encode
-import copy
 import json
 import logging
 
@@ -61,8 +60,8 @@ def generate_and_send_pads(user_data, user_data_lock, contact_id: str, ui_queue)
             "recipient": contact_id
         }
     try:
-        response = http_request(f"{server_url}/messages/send_pads", "POST", payload=payload, auth_token=auth_token)
-    except:
+        http_request(f"{server_url}/messages/send_pads", "POST", payload=payload, auth_token=auth_token)
+    except Exception:
         ui_queue.put({"type": "showerror", "title": "Error", "message": "Failed to send our one-time-pads key batch to the server"})
         return False
 
@@ -87,7 +86,6 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
     Returns:
         bool: True if successful, False otherwise.
     """
-    # We don't deepcopy here as real-time states are required here for maximum future-proofing.
 
     with user_data_lock:
         if contact_id in user_data["tmp"]["ephemeral_key_send_lock"]:
@@ -103,16 +101,14 @@ 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_key"]
-        our_lt_private_key       = user_data["contacts"][contact_id]["lt_sign_keys"]["our_keys"]["private_key"]
-
 
 
     if (not contact_kyber_public_key):
         logger.debug("This shouldn't usually happen, contact kyber keys are not initialized even once yet???")
         ui_queue.put({
                 "type": "showwarning",
                 "title": f"Warning for {contact_id[:32]}",
-                "message": f"Ephemeral keys have not yet initialized, maybe contact is offline. We will notify you when keys are initialized"
+                "message": "Ephemeral keys have not yet initialized, maybe contact is offline. We will notify you when keys are initialized"
             })
 
         send_new_ephemeral_keys(user_data, user_data_lock, contact_id, ui_queue)
@@ -211,7 +207,7 @@ def send_message_processor(user_data, user_data_lock, contact_id: str, message:
     save_account_data(user_data, user_data_lock)
 
     try:
-        response = http_request(f"{server_url}/messages/send_message", "POST", payload = {
+        http_request(f"{server_url}/messages/send_message", "POST", payload = {
                     "message_encrypted": message_encrypted,
                     "recipient": contact_id
                 }, 

logic/pfs.py

@@ -7,16 +7,12 @@
     generate_sign_keys,
     random_number_range
 )
-from core.trad_crypto import derive_key_argon2id, sha3_512
+from core.trad_crypto import sha3_512
 from base64 import b64encode, b64decode
 import secrets
-import hmac
-import time
 import copy
 import json
 import logging
-import oqs
-
 
 logger = logging.getLogger(__name__)
 
@@ -57,8 +53,8 @@ def send_new_ephemeral_keys(user_data, user_data_lock, contact_id, ui_queue) ->
 
 
     try:
-        response = http_request(f"{server_url}/pfs/send_keys", "POST", payload=payload, auth_token=auth_token)
-    except:
+        http_request(f"{server_url}/pfs/send_keys", "POST", payload=payload, auth_token=auth_token)
+    except Exception:
         ui_queue.put({"type": "showerror", "title": "Error", "message": "Failed to send our ephemeral keys to the server"})
         return
 

logic/smp.py

@@ -1,4 +1,6 @@
 """
+    logic/smp.py
+    ----------
     The socialist millionaire problem
     A variant of Yao's millionaire problem
 
@@ -34,16 +36,13 @@
 from logic.storage import save_account_data
 from logic.contacts import save_contact
 from logic.pfs import send_new_ephemeral_keys
-from core.crypto import random_number_range, generate_sign_keys
+from core.crypto import generate_sign_keys
 from core.trad_crypto import derive_key_argon2id, sha3_512
 from base64 import b64encode, b64decode
 import hashlib
 import secrets
 import hmac
-import time
-import copy
 import logging
-import json
 
 
 logger = logging.getLogger(__name__)
@@ -71,7 +70,7 @@ def initiate_smp(user_data: dict, user_data_lock, contact_id: str, question: str
             "recipient": contact_id
 
         }, auth_token=auth_token)
-    except:
+    except Exception:
         raise ValueError("Could not connect to server")
 
     if (not ("status" in response)) or response["status"] != "success":
@@ -135,7 +134,7 @@ def smp_step_2_answer_provided(user_data, user_data_lock, contact_id, answer, ui
             "recipient": contact_id
 
         }, auth_token=auth_token)
-    except:
+    except Exception:
         logger.error("Failed to send proof request to server, either you are offline or the server is down")
         smp_failure_notify_contact(user_data, user_data_lock, contact_id, ui_queue)
         return
@@ -217,11 +216,11 @@ def smp_step_3(user_data, user_data_lock, contact_id, message, ui_queue) -> None
     logger.debug("Message to contact: %s", our_message)
 
     try:
-        response = http_request(f"{server_url}/smp/step_3", "POST", payload = {
+        http_request(f"{server_url}/smp/step_3", "POST", payload = {
             "proof": our_message,
             "recipient": contact_id
         }, auth_token=auth_token)
-    except:
+    except Exception:
         logger.error("Failed to send proof request to server, either you are offline or the server is down")
         smp_failure_notify_contact(user_data, user_data_lock, contact_id, ui_queue)
         return
@@ -232,16 +231,11 @@ def smp_step_3(user_data, user_data_lock, contact_id, message, ui_queue) -> None
 
 def smp_step_4(user_data, user_data_lock, contact_id, message, ui_queue) -> None:
     with user_data_lock:
-        server_url = user_data["server_url"]
-        auth_token = user_data["token"]
-
         answer = normalize_answer(user_data["contacts"][contact_id]["lt_sign_key_smp"]["answer"])
 
         our_public_key     = user_data["contacts"][contact_id]["lt_sign_keys"]["our_keys"]["public_key"]
         our_nonce          = b64decode(user_data["contacts"][contact_id]["lt_sign_key_smp"]["our_nonce"], validate=True)
         contact_nonce      = b64decode(user_data["contacts"][contact_id]["lt_sign_key_smp"]["contact_nonce"], validate=True)
-
-        contact_public_key = user_data["contacts"][contact_id]["lt_sign_keys"]["contact_public_key"]
 
     our_key_fingerprint = sha3_512(our_public_key)
 
@@ -323,8 +317,8 @@ def smp_failure_notify_contact(user_data, user_data_lock, contact_id, ui_queue)
     smp_failure(user_data, user_data_lock, contact_id, ui_queue)
 
     try:
-        response = http_request(f"{server_url}/smp/failure", "POST", payload = {"recipient": contact_id}, auth_token=auth_token)
-    except:
+        http_request(f"{server_url}/smp/failure", "POST", payload = {"recipient": contact_id}, auth_token=auth_token)
+    except Exception:
         logger.error("Failed to send SMP failure to server, either you are offline or the server is down")
         pass
 

ui/chat_window.py

@@ -1,5 +1,4 @@
 import tkinter as tk
-from tkinter import messagebox
 from logic.message import send_message_processor
 
 class ChatWindow(tk.Toplevel):

ui/connect_window.py

@@ -3,7 +3,6 @@
 from logic.storage import save_account_data
 from logic.authentication import authenticate_account
 from core.crypto import generate_sign_keys
-from base64 import b64encode
 from urllib.parse import urlparse
 import tkinter as tk
 import logging
@@ -200,7 +199,7 @@ def on_connect(self, event=None):
 
         try:
             self.server_url_fixed = self.clean_server_url(self.server_url_fixed)
-        except:
+        except Exception:
             self.status_label.config(text="Server address is invalid!")
             return