refactor: add strict type annotations to medium-risk modules

Author: NikkeTryHard

launcher/checks.py

@@ -1,13 +1,14 @@
 import logging
 import os
 import sys
+from typing import Dict, List, Optional
 
 from launcher.config import ACTIVE_AUTH_DIR, SAVED_AUTH_DIR
 
 logger = logging.getLogger("CamoufoxLauncher")
 
 
-def ensure_auth_dirs_exist():
+def ensure_auth_dirs_exist() -> None:
     logger.info("正在检查并确保认证文件目录存在...")
     try:
         os.makedirs(ACTIVE_AUTH_DIR, exist_ok=True)
@@ -19,17 +20,19 @@ def ensure_auth_dirs_exist():
         sys.exit(1)
 
 
-def check_dependencies(launch_server, DefaultAddons):
+def check_dependencies(
+    launch_server: Optional[bool], DefaultAddons: Optional[bool]
+) -> None:
     logger.info("--- 步骤 1: 检查依赖项 ---")
-    required_modules = {}
+    required_modules: Dict[str, str] = {}
     if launch_server is not None and DefaultAddons is not None:
         required_modules["camoufox"] = "camoufox (for server and addons)"
     elif launch_server is not None:
         required_modules["camoufox_server"] = "camoufox.server"
         logger.warning(
             "  'camoufox.server' 已导入，但 'camoufox.DefaultAddons' 未导入。排除插件功能可能受限。"
         )
-    missing_py_modules = []
+    missing_py_modules: List[str] = []
     dependencies_ok = True
     if required_modules:
         logger.info("正在检查 Python 模块:")

launcher/config.py

@@ -2,7 +2,7 @@
 import os
 import sys
 from dataclasses import dataclass
-from typing import Optional
+from typing import Dict, Optional
 
 from launcher.utils import get_proxy_from_gsettings
 
@@ -64,7 +64,9 @@ class LauncherConfig:
     internal_camoufox_os: str = "random"
 
 
-def determine_proxy_configuration(internal_camoufox_proxy_arg=None):
+def determine_proxy_configuration(
+    internal_camoufox_proxy_arg: Optional[str] = None,
+) -> Dict[str, Optional[str]]:
     """
     统一的代理配置确定函数
     按优先级顺序：命令行参数 > 环境变量 > 系统设置

launcher/utils.py

@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import threading
+from typing import List, Optional
 
 logger = logging.getLogger("CamoufoxLauncher")
 
@@ -22,8 +23,8 @@ def is_port_in_use(port: int, host: str = "0.0.0.0") -> bool:
             return True
 
 
-def find_pids_on_port(port: int) -> list[int]:
-    pids = []
+def find_pids_on_port(port: int) -> List[int]:
+    pids: List[int] = []
     system_platform = platform.system()
     command = ""
     try:
@@ -166,7 +167,7 @@ def kill_process_interactive(pid: int) -> bool:
 def input_with_timeout(prompt_message: str, timeout_seconds: int = 30) -> str:
     print(prompt_message, end="", flush=True)
     if sys.platform == "win32":
-        user_input_container = [None]
+        user_input_container: List[Optional[str]] = [None]
 
         def get_input_in_thread():
             try:
@@ -190,13 +191,13 @@ def get_input_in_thread():
             return ""
 
 
-def get_proxy_from_gsettings():
+def get_proxy_from_gsettings() -> Optional[str]:
     """
     Retrieves the proxy settings from GSettings on Linux systems.
     Returns a proxy string like "http://host:port" or None.
     """
 
-    def _run_gsettings_command(command_parts: list[str]) -> str | None:
+    def _run_gsettings_command(command_parts: List[str]) -> Optional[str]:
         """Helper function to run gsettings command and return cleaned string output."""
         try:
             process_result = subprocess.run(

stream/cert_manager.py

@@ -1,5 +1,6 @@
 import datetime
 from pathlib import Path
+from typing import Any, Tuple
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
@@ -9,7 +10,7 @@
 
 
 class CertificateManager:
-    def __init__(self, cert_dir="certs"):
+    def __init__(self, cert_dir: str = "certs"):
         self.cert_dir = Path(cert_dir)
         self.cert_dir.mkdir(exist_ok=True)
 
@@ -92,7 +93,7 @@ def _load_ca_cert(self):
         with open(self.ca_cert_path, "rb") as f:
             self.ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend())
 
-    def get_domain_cert(self, domain):
+    def get_domain_cert(self, domain: str) -> Tuple[Any, Any]:
         """Get or generate a certificate for the specified domain"""
         cert_path = self.cert_dir / f"{domain}.crt"
         key_path = self.cert_dir / f"{domain}.key"
@@ -112,7 +113,7 @@ def get_domain_cert(self, domain):
         # Generate new certificate
         return self._generate_domain_cert(domain)
 
-    def _generate_domain_cert(self, domain):
+    def _generate_domain_cert(self, domain: str) -> Tuple[Any, Any]:
         """Generate a certificate for the specified domain signed by the CA"""
         # Generate private key
         private_key = rsa.generate_private_key(
@@ -152,7 +153,7 @@ def _generate_domain_cert(self, domain):
             .add_extension(
                 x509.SubjectAlternativeName([x509.DNSName(domain)]), critical=False
             )
-            .sign(self.ca_key, hashes.SHA256(), default_backend())
+            .sign(self.ca_key, hashes.SHA256(), default_backend())  # type: ignore[arg-type]
         )
 
         # Write certificate to file

stream/interceptors.py

@@ -3,6 +3,7 @@
 import re
 import sys
 import zlib
+from typing import Any, Dict, Tuple, Union
 
 from logging_utils.setup import ColoredFormatter
 
@@ -12,7 +13,7 @@ class HttpInterceptor:
     Class to intercept and process HTTP requests and responses
     """
 
-    def __init__(self, log_dir="logs"):
+    def __init__(self, log_dir: str = "logs"):
         self.log_dir = log_dir
         self.logger = logging.getLogger("http_interceptor")
         self.setup_logging()
@@ -36,7 +37,7 @@ def setup_logging():
         logging.getLogger("websockets").setLevel(logging.ERROR)
 
     @staticmethod
-    def should_intercept(host, path):
+    def should_intercept(host: str, path: str):
         """
         Determine if the request should be intercepted based on host and path
         """
@@ -47,7 +48,9 @@ def should_intercept(host, path):
         # Add more conditions as needed
         return False
 
-    async def process_request(self, request_data, host, path):
+    async def process_request(
+        self, request_data: Union[int, bytes], host: str, path: str
+    ) -> Union[int, bytes]:
         """
         Process the request data before sending to the server
         """
@@ -63,7 +66,13 @@ async def process_request(self, request_data, host, path):
             # Not JSON or not UTF-8, just pass through
             return request_data
 
-    async def process_response(self, response_data, host, path, headers):
+    async def process_response(
+        self,
+        response_data: Union[int, bytes],
+        host: str,
+        path: str,
+        headers: Dict[Any, Any],
+    ) -> Dict[str, Any]:
         """
         Process the response data before sending to the client
         """
@@ -78,7 +87,7 @@ async def process_response(self, response_data, host, path, headers):
         except Exception as e:
             raise e
 
-    def parse_response(self, response_data):
+    def parse_response(self, response_data: bytes) -> Dict[str, Any]:
         pattern = rb'\[\[\[null,.*?]],"model"]'
         matches = []
         for match_obj in re.finditer(pattern, response_data):
@@ -114,7 +123,7 @@ def parse_response(self, response_data):
 
         return resp
 
-    def parse_toolcall_params(self, args):
+    def parse_toolcall_params(self, args: Any) -> Dict[str, Any]:
         try:
             params = args[0]
             func_params = {}
@@ -140,13 +149,13 @@ def parse_toolcall_params(self, args):
             raise e
 
     @staticmethod
-    def _decompress_zlib_stream(compressed_stream):
+    def _decompress_zlib_stream(compressed_stream: Union[bytearray, bytes]) -> bytes:
         decompressor = zlib.decompressobj(wbits=zlib.MAX_WBITS | 32)  # zlib header
         decompressed = decompressor.decompress(compressed_stream)
         return decompressed
 
     @staticmethod
-    def _decode_chunked(response_body: bytes) -> tuple[bytes, bool]:
+    def _decode_chunked(response_body: bytes) -> Tuple[bytes, bool]:
         chunked_data = bytearray()
         while True:
             # print(' '.join(format(x, '02x') for x in response_body))
@@ -165,7 +174,7 @@ def _decode_chunked(response_body: bytes) -> tuple[bytes, bool]:
             if length == 0:
                 length_crlf_idx = response_body.find(b"0\r\n\r\n")
                 if length_crlf_idx != -1:
-                    return chunked_data, True
+                    return bytes(chunked_data), True
 
             if length + 2 > len(response_body):
                 break
@@ -177,4 +186,4 @@ def _decode_chunked(response_body: bytes) -> tuple[bytes, bool]:
                 break
 
             response_body = response_body[length_crlf_idx + 2 + length + 2 :]
-        return chunked_data, False
+        return bytes(chunked_data), False

stream/main.py

@@ -1,15 +1,15 @@
 import argparse
 import asyncio
 import logging
-import multiprocessing
 import sys
 from pathlib import Path
+from typing import Any, Optional
 
 from logging_utils import GridFormatter, set_source
 from stream.proxy_server import ProxyServer
 
 
-def parse_args():
+def parse_args() -> argparse.Namespace:
     """Parse command line arguments"""
     parser = argparse.ArgumentParser(
         description="HTTPS Proxy Server with SSL Inspection"
@@ -34,7 +34,7 @@ def parse_args():
     return parser.parse_args()
 
 
-async def main():
+async def main() -> None:
     """Main entry point"""
     args = parse_args()
 
@@ -83,7 +83,9 @@ async def main():
         sys.exit(1)
 
 
-async def builtin(queue: multiprocessing.Queue = None, port=None, proxy=None):
+async def builtin(
+    queue: Optional[Any] = None, port: Optional[int] = None, proxy: Optional[str] = None
+) -> None:
     # Set up logging with GridFormatter for consistent output
     set_source("PROXY")
 

stream/proxy_connector.py

@@ -2,6 +2,8 @@
 import ssl as ssl_module
 import urllib.parse
 
+from typing import Any, Optional, Tuple
+
 from aiohttp import TCPConnector
 from python_socks.async_.asyncio import Proxy
 
@@ -11,7 +13,7 @@ class ProxyConnector:
     Class to handle connections through different types of proxies
     """
 
-    def __init__(self, proxy_url=None):
+    def __init__(self, proxy_url: Optional[str] = None):
         self.proxy_url = proxy_url
         self.connector = None
 
@@ -33,15 +35,18 @@ def _setup_connector(self):
         else:
             raise ValueError(f"Unsupported proxy type: {proxy_type}")
 
-    async def create_connection(self, host, port, ssl=None):
+    async def create_connection(
+        self, host: str, port: int, ssl: Optional[Any] = None
+    ) -> Tuple[Any, Any]:
         """Create a connection to the target host through the proxy"""
         if not self.connector:
             # Direct connection without proxy
             reader, writer = await asyncio.open_connection(host, port, ssl=ssl)
             return reader, writer
 
         # SOCKS proxy connection
-        proxy = Proxy.from_url(self.proxy_url)
+        assert self.proxy_url is not None  # Type guard for pyright
+        proxy = Proxy.from_url(self.proxy_url)  # type: ignore[misc]
         sock = await proxy.connect(dest_host=host, dest_port=port)
         if ssl is None:
             reader, writer = await asyncio.open_connection(