From c8811c93bbb5423ffe1de79b2faff95b3ddba7f8 Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Wed, 19 Aug 2026 21:40:36 -0400 Subject: [PATCH] Linter undefined or missing import bugs --- core/controllers/tickets/UseTicketController.py | 2 +- core/models/system/SystemProfile.py | 12 ++++++------ .../systemwide/encrypted_proxy/configure_singbox.py | 4 +--- .../systemwide/encrypted_proxy/singbox.py | 2 +- .../systemwide/encrypted_proxy/singbox_runner.py | 13 +++++-------- .../systemwide/encrypted_proxy/vless_config.py | 2 +- .../networking/systemwide/systemwide_wireguard.py | 5 ++++- core/utils/basic_operations/wrap_with.py | 13 +------------ .../basic_operations/write_or_read_from_json.py | 3 +-- 9 files changed, 21 insertions(+), 35 deletions(-) diff --git a/core/controllers/tickets/UseTicketController.py b/core/controllers/tickets/UseTicketController.py index 7e94c63..0e2134d 100644 --- a/core/controllers/tickets/UseTicketController.py +++ b/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 diff --git a/core/models/system/SystemProfile.py b/core/models/system/SystemProfile.py index a90e3f8..f38ccd1 100644 --- a/core/models/system/SystemProfile.py +++ b/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')) - wireguard_configuration_is_attached = not bool(os.waitpid(process.pid, 0)[1] >> 8) + process = subprocess.Popen(('pkexec', 'install', '-D', backup_path, self.get_wireguard_configuration_path(), '-o', 'root', '-m', '744')) + 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.') diff --git a/core/services/networking/systemwide/encrypted_proxy/configure_singbox.py b/core/services/networking/systemwide/encrypted_proxy/configure_singbox.py index 7f20951..ae2f952 100644 --- a/core/services/networking/systemwide/encrypted_proxy/configure_singbox.py +++ b/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 diff --git a/core/services/networking/systemwide/encrypted_proxy/singbox.py b/core/services/networking/systemwide/encrypted_proxy/singbox.py index 84f3185..d6f75a0 100644 --- a/core/services/networking/systemwide/encrypted_proxy/singbox.py +++ b/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)] diff --git a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py b/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py index ce410e5..b4afb54 100644 --- a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py +++ b/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, diff --git a/core/services/networking/systemwide/encrypted_proxy/vless_config.py b/core/services/networking/systemwide/encrypted_proxy/vless_config.py index 1e8577f..9f6ea21 100644 --- a/core/services/networking/systemwide/encrypted_proxy/vless_config.py +++ b/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://", "") diff --git a/core/services/networking/systemwide/systemwide_wireguard.py b/core/services/networking/systemwide/systemwide_wireguard.py index 2bc6e90..3eef708 100644 --- a/core/services/networking/systemwide/systemwide_wireguard.py +++ b/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. diff --git a/core/utils/basic_operations/wrap_with.py b/core/utils/basic_operations/wrap_with.py index d2a6794..1b742c9 100644 --- a/core/utils/basic_operations/wrap_with.py +++ b/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 - diff --git a/core/utils/basic_operations/write_or_read_from_json.py b/core/utils/basic_operations/write_or_read_from_json.py index 47b817c..e5524ff 100644 --- a/core/utils/basic_operations/write_or_read_from_json.py +++ b/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, Optional +from typing import Any import tempfile import shutil