Linter undefined or missing import bugs

core/controllers/tickets/UseTicketController.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Union
 
 if TYPE_CHECKING:
     from core.observers.TicketObserver import TicketObserver

core/models/system/SystemProfile.py

@@ -121,17 +121,17 @@ class SystemProfile(BaseProfile):
         with open(backup_path, 'w') as configuration_file:
             configuration_file.write(config_data)
 
-        wireguard_configuration_is_attached = False
+        configuration_is_attached = False
         failed_attempt_count = 0
 
-        while not wireguard_configuration_is_attached and failed_attempt_count < 3:
+        while not configuration_is_attached and failed_attempt_count < 3:
 
-            process = subprocess.Popen(('pkexec', 'install', '-D', wireguard_configuration_file_backup_path, self.get_wireguard_configuration_path(), '-o', 'root', '-m', '744'))
+            process = subprocess.Popen(('pkexec', 'install', '-D', backup_path, self.get_wireguard_configuration_path(), '-o', 'root', '-m', '744'))
-            wireguard_configuration_is_attached = not bool(os.waitpid(process.pid, 0)[1] >> 8)
+            configuration_is_attached = not bool(os.waitpid(process.pid, 0)[1] >> 8)
 
-            if not wireguard_configuration_is_attached:
+            if not configuration_is_attached:
                 failed_attempt_count += 1
 
-        if not wireguard_configuration_is_attached:
+        if not configuration_is_attached:
             raise ProfileModificationError('The WireGuard configuration could not be attached.')
 

core/services/networking/systemwide/encrypted_proxy/configure_singbox.py

@@ -18,10 +18,8 @@ from core.models.manage import pydantic_manager
 # errors & observers
 from core.Constants import Constants
 from core.errors.logger import logger
-from core.Errors import MissingSubscriptionError
 from core.observers.ConnectionObserver import ConnectionObserver
-from core.Errors import ProfileModificationError
+from core.Errors import ProfileModificationError, CommandNotFoundError, MissingSubscriptionError
-
 
 # generic
 from pydantic import ValidationError

core/services/networking/systemwide/encrypted_proxy/singbox.py

@@ -25,7 +25,7 @@ def start(profile_id: int) -> Result:
     if not made_output_file.valid:
         logger.error("There was an issue with making the blank file for singbox's output.")
         if not write_string_to_text_file(content_to_write="hello world", file_path=Constants.SINGBOX_OUTPUT):
-            return Result(valid=False, error_type=MISSING_FILE, message=f"Could not create the file to output the singbox content at {Constants.SINGBOX_OUTPUT}")
+            return Result(valid=False, error_type=ResultError.MISSING_FILE, message=f"Could not create the file to output the singbox content at {Constants.SINGBOX_OUTPUT}")
 
     # start singbox:
     command = ["sudo", SINGBOX_WRAPPER, "arm", str(profile_id)]

core/services/networking/systemwide/encrypted_proxy/singbox_runner.py

@@ -79,7 +79,7 @@ def launch_singbox_binary(profile_id: int) -> Result:
             return Result(valid=True, error_type=ResultError.PROCESS_MISMATCH, data=double_check.data)
         else:
             error_msg = f"While Singbox might have literally allowed the binary to begin, it's killing the process on id {process_id}"
-            logger.error(f"[{function_name}] {error_msg}")
+            logger.error(error_msg)
             return Result(valid=False, error_type=ResultError.PROCESS_GOT_KILLED, message=error_msg, data=process_id)
 
 
@@ -102,13 +102,13 @@ def _attempt_start_with_retry(profile_id: int, quantity_of_attempts: int = 2) ->
     try:
         double_check = pid_tools.get_pid_by_app_name(exact_app_name="sing-box")
     except Exception as e:
-        return Result(valid=False, error_type=PROCESS_WONT_START, message=str(e))
+        return Result(valid=False, error_type=ResultError.PROCESS_WONT_START, message=str(e))
 
     if double_check.valid:
         return double_check
     else:
         # I'd like to clarify the error_type here, instead of just returning the pid check or an individual attempt.
-        return Result(valid=False, error_type=PROCESS_WONT_START, message=error_msg)
+        return Result(valid=False, error_type=ResultError.PROCESS_WONT_START, message=error_msg)
 
 
 def end_singbox() -> Result:
@@ -190,14 +190,11 @@ def start_singbox(
         6) Turn on DNS
         7) Update the State
     """
-
-    function_name = "START_SINGBOX"
-
     # ============= INPUT VALIDATION =============
     requirements = [profile_id, server_ip]
     for each_requirement in requirements:
         if each_requirement is None:
-            return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"Invalid inputs into {function_name}")
+            return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"Invalid inputs into singbox runner")
     logger.info("All requirements met.")
 
     # ========== KILL IT IF ALREADY UP ==========
@@ -262,7 +259,7 @@ def start_singbox(
         return Result(valid=False, error_type=ResultError.LEAK_ISSUE, message=error_msg)
 
     # ============= FIREWALL =============
-    logger.info(f"[{function_name}] Attempting to enable the Firewall for {Constants.SINGBOX_TUN_IF} and {Constants.SINGBOX_INTERNAL_SUBNET}...")
+    logger.info(f"Attempting to enable the Firewall for {Constants.SINGBOX_TUN_IF} and {Constants.SINGBOX_INTERNAL_SUBNET}...")
     firewall_result = enable_firewall_w_retry(  # this function is "generic" as it's protocol neutral
         interface_name=Constants.SINGBOX_TUN_IF,
         server_ip=server_ip,

core/services/networking/systemwide/encrypted_proxy/vless_config.py

@@ -1,5 +1,5 @@
 from core.Constants import Constants
-
+from urllib.parse import unquote
 
 def parse_vless_link(link: str) -> dict:
     link = link.replace("vless://", "")

core/services/networking/systemwide/systemwide_wireguard.py

@@ -12,6 +12,7 @@ from core.services.networking.systemwide.wireguard.wg_firewall_dns import set_dn
 from core.errors.logger import logger
 from core.errors.exceptions import FirewallError
 from core.utils.run_commands import run_generic_command
+from core.observers.TicketObserver import TicketObserver
 
 from core.Constants import Constants
 from core.Errors import ConnectionTerminationError, CommandNotFoundError
@@ -245,7 +246,9 @@ def _establish_connection_with_retry(
 def establish_system_connection(
     profile: SystemProfile,
     ignore: tuple[type[Exception]] = (),
-    connection_observer: Optional[ConnectionObserver] = None):
+    connection_observer: Optional[ConnectionObserver] = None,
+    ticket_observer: Optional[TicketObserver] = None # Passed in only for assassin's link to tickets in ConnectionController.establish_session_connection
+    ):
     """
     Purpose:
         Calls the private function with the full flow, so it can be retried on failure.

core/utils/basic_operations/wrap_with.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+# from dataclasses import dataclass
 from typing import TypeVar, Callable
 import functools
 
@@ -27,14 +27,3 @@ def wrap_with(wrapper_fn: Callable[[T], U]) -> Callable:
         return wrapper
     return decorator
 
-
-# def wrap_with(wrapper_fn: Callable[[Result], Result]) -> Callable:
-#     def decorator(func: Callable[..., Result]) -> Callable[..., Result]:
-#         @functools.wraps(func)
-#         def wrapper(*args, **kwargs) -> Result:
-#             result = func(*args, **kwargs)
-#             processed_result = wrapper_fn(result)  # ← Any function plugged in
-#             return processed_result
-#         return wrapper
-#     return decorator
-

core/utils/basic_operations/write_or_read_from_json.py

@@ -2,8 +2,7 @@ from core.errors.logger import logger
 
 import json
 import os
-from pathlib import Path
+from typing import Any
-from typing import Any, Optional
 import tempfile
 import shutil