[200~Elaborate HTTPx feature-rich strategy with HTTPx Client Reuse, Tor Management, Tor bootstrap monitor & retry. HTTPx clients can be managed across modules and reused. There's an elaborate HTTPx retry strategy based on status codes. Tor management tries to use the existing Tor default ports, then checks the Tor config, before finally bootstrapping. There is now a ThreadPool futures executor, which even works with Tor, to batch API requests.

change_log.md

@@ -1,6 +1,11 @@
 # Major Change Log: 
 
 
+# HTTPx Client, ThreadPool, & Tor Management  
+### August 2, 2026  
+Elaborate HTTPx feature-rich strategy with HTTPx Client Reuse, Tor Management, Tor bootstrap monitor & retry.  HTTPx clients can be managed across modules and reused.  There's an elaborate HTTPx retry strategy based on status codes.  Tor management tries to use the existing Tor default ports, then checks the Tor config, before finally bootstrapping.  There is now a ThreadPool futures executor, which even works with Tor, to batch API requests.  
+
+
 # Singbox  
 ### July 29, 2026
 Introduced the Singbox Enable Utility & Singbox Sudo Script   

core/Constants.py

@@ -78,3 +78,8 @@ class Constants:
     SINGBOX_TUN_IF:             Final[str] = os.environ.get('SINGBOX_TUN_IF', 'tun0')
     SINGBOX_INTERNAL_SUBNET:    Final[str] = os.environ.get('SINGBOX_INTERNAL_SUBNET', '172.19.0.0/30')
     SINGBOX_INTERNAL_ADDR:      Final[str] = os.environ.get('SINGBOX_INTERNAL_ADDR', '172.19.0.1/30')
+
+    # ── Tor ─────────────────────────────────────────────
+    DEFAULT_TOR_PORT = 9050
+
+    

core/errors/logger.py

@@ -18,7 +18,7 @@ debug_mode = os.getenv("DEBUG", "").lower() in ("1", "true", "yes")
 console_handler = logging.StreamHandler(sys.stdout)
 console_level = logging.DEBUG if debug_mode else logging.WARNING
 console_handler.setLevel(console_level)
-console_formatter = logging.Formatter("%(message)s")
+console_formatter = logging.Formatter("[%(funcName)s] %(message)s")
 console_handler.setFormatter(console_formatter)
 logger.addHandler(console_handler)
 

core/services/networking/api_requests/ApiResponseModel.py

@@ -2,10 +2,29 @@ from enum import Enum
 from dataclasses import dataclass
 from typing import Optional, Any
 
+
+class BackoffStrategy(Enum):
+    """How to handle retries."""
+    NO_RETRY = "no_retry"                           # Permanent error
+    RETRY_IMMEDIATE = "retry_immediate"             # Try again in <1s (502, 503, 504)
+    RETRY_EXPONENTIAL = "retry_exponential"         # Exponential backoff (500, 429)
+    RETRY_WITH_AUTH_REFRESH = "retry_with_auth_refresh"  # 401 after refreshing creds
+    FIX_CLIENT_SIDE_INFO = "fix_client_side_info"    # change payload or 405 get/post
+
 class ErrorType(Enum):
     """Classified error categories."""
     SUCCESS = "success"
+    INVALID_ENDPOINT = "invalid_endpoint"           # 404 - wrong URL
+    INVALID_REQUEST = "invalid_request"             # 400, 405 - bad method/body
+    AUTHENTICATION_ERROR = "authentication_error"   # 401 - need credentials
+    AUTHORIZATION_ERROR = "authorization_error"     # 403 - no permission
+    RATE_LIMITED = "rate_limited"                   # 429
+    SERVER_ERROR = "server_error"                   # 5xx transient
+    NETWORK_ERROR = "network_error"                 # Connection issues
+    TOR_COMMENTED_PORT = "tor_commented_port"
+    TOR_FILE_MISSING = "tor_file_missing"
     DEFAULT_TOR_PORT_DEAD = "default_tor_port_dead"
+    TOR_WORKS_BUT_UNRELIABLE = "tor_works_but_unreliable"
     TOR_NOT_INSTALLED = "tor_not_installed"
     REFUSAL_TO_INSTALL_TOR = "refusal_to_install_tor"
     PORT_OPEN = "port_open"
@@ -16,12 +35,12 @@ class ErrorType(Enum):
     TOR_NOT_WORKING = "tor_not_working"
     TOR_DNS_BLOCKED = "tor_dns_blocked"
     DNS_RESOLUTION = "dns_resolution"
+    DNS_TEMPORARY = "dns_temporary"
+    DNS_PERMANENT = "dns_permanent"
     QUAD9_DNS_RESOLUTION = "quad9_dns_resolution"
     NO_INTERNET = "no_internet"
-    NETWORK_ERROR = "network_error"
     INVALID_INPUT = "invalid_input"
-    RATE_LIMITED = "rate_limited"
-    SERVER_ERROR = "server_error"
+    PERMISSION_ERROR = "permission_error" # duplicate
     DEVELOPER_ERROR = "developer_error"
     UNKNOWN = "unknown"
 
@@ -32,8 +51,9 @@ class ApiResponse:
     error_type: Optional[ErrorType] = None
     data: Optional[Any] = None
     message: Optional[str] = None
-    retry_now: bool = False
-    retry_later: bool = False
+    backoff_strategy: BackoffStrategy = BackoffStrategy.NO_RETRY
+    retry_now: bool = False # legacy
+    retry_later: bool = False # legacy
     tor: bool = False
     ip_address: str = None
     ask_clearweb: bool = None

core/services/networking/api_requests/step3_classify_http.py

@@ -9,7 +9,6 @@ import json
 def classify_http_response(response) -> ApiResponse:
     """
     Classify HTTP response status and extract error info.
-    Replaces evaluate_response().
     """
     import requests
 

core/services/networking/httpx/classify_response.py

@@ -0,0 +1,106 @@
+
+"""
+Purpose:
+    Classify HTTP response status and extract error info.
+
+Returns:
+    ApiResponse with semantic error types and backoff strategies.
+"""
+
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType, BackoffStrategy
+from core.errors.logger import logger
+
+import httpx
+import json
+
+def classify_response(response) -> ApiResponse:
+
+    if not isinstance(response, httpx.Response):
+        error_msg = "Library Error: HTTPx library returned an invalid object response."
+        logger.error(error_msg)
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.NETWORK_ERROR,
+            message=error_msg,
+            backoff_strategy=BackoffStrategy.NO_RETRY
+        )
+
+    logger.info(f"Classifying HTTP response: {response.status_code}")
+
+    # Success case (2xx)
+    if 200 <= response.status_code < 300:
+        try:
+            data = response.json()
+            return ApiResponse(valid=True, data=data)
+        except ValueError:
+            return ApiResponse(valid=True, data=response.text)
+    
+    # Rate limiting (429) - should back off exponentially
+    if response.status_code == 429:
+        logger.error("Rate limited API call!")
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.RATE_LIMITED,
+            message="Rate limit exceeded",
+            backoff_strategy=BackoffStrategy.RETRY_EXPONENTIAL
+        )
+    
+    # Client errors (4xx) - extract error message once
+    if 400 <= response.status_code < 500:
+        try:
+            resp_json = response.json()
+            error_msg = (
+                resp_json.get("message") 
+                or resp_json.get("error") 
+                or resp_json.get("error_code")
+            )
+        except ValueError:
+            error_msg = response.text
+        
+        # Map status codes to semantic error types and retry strategies
+        error_map = {
+            400: (ErrorType.INVALID_REQUEST, BackoffStrategy.FIX_CLIENT_SIDE_INFO),
+            401: (ErrorType.AUTHENTICATION_ERROR, BackoffStrategy.NO_RETRY),
+            403: (ErrorType.AUTHORIZATION_ERROR, BackoffStrategy.NO_RETRY),
+            404: (ErrorType.INVALID_ENDPOINT, BackoffStrategy.NO_RETRY),
+            405: (ErrorType.INVALID_REQUEST, BackoffStrategy.FIX_CLIENT_SIDE_INFO), # most likely to be get vs post.
+        }
+        
+        error_type, backoff_strategy = error_map.get(
+            response.status_code,
+            (ErrorType.INVALID_REQUEST, BackoffStrategy.NO_RETRY)
+        )
+        
+        return ApiResponse(
+            valid=False,
+            error_type=error_type,
+            message=error_msg or f"Client error {response.status_code}",
+            backoff_strategy=backoff_strategy
+        )
+    
+    # Server errors (5xx) - distinguish transient from permanent errors
+    if response.status_code >= 500:
+        # Transient upstream issues - retry immediately
+        if response.status_code in (502, 503, 504):
+            backoff_strategy = BackoffStrategy.RETRY_IMMEDIATE
+        # Transient server errors - retry with exponential backoff
+        elif response.status_code == 500:
+            backoff_strategy = BackoffStrategy.RETRY_EXPONENTIAL
+        # Permanent errors - don't retry
+        else:  # 501, 505, and any others
+            backoff_strategy = BackoffStrategy.NO_RETRY
+        
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.SERVER_ERROR,
+            message=f"Server error {response.status_code}",
+            backoff_strategy=backoff_strategy
+        )
+    
+    # Unexpected status code
+    return ApiResponse(
+        valid=False,
+        error_type=ErrorType.UNKNOWN,
+        message=f"Unexpected status {response.status_code}",
+        backoff_strategy=BackoffStrategy.NO_RETRY
+    )

core/services/networking/httpx/httpx_client.py

@@ -0,0 +1,91 @@
+
+"""
+Module Purpose:
+    Manage and reuse an HTTPx Client across modules
+"""
+
+from core.errors.logger import logger
+
+import httpx
+from httpx_socks import AsyncProxyTransport
+import httpx_socks
+from httpx_socks import ProxyType
+
+_http_client = None
+
+
+def init_tor_session(port: int = 9050) -> bool:
+    """
+    Purpose:
+        First, Test Tor on the given port.
+        If successfull, then initialize a global http_client.
+
+    Returns:
+        True on success (and creates the client to be accessed later via get_http_session)
+        False on failure.
+    """
+    global _http_client
+    
+    try:
+        transport = httpx_socks.SyncProxyTransport(
+            proxy_type=ProxyType.SOCKS5,
+            proxy_host="127.0.0.1",
+            proxy_port=port,
+        )
+        client = httpx.Client(transport=transport, http2=True, timeout=10)
+        
+        try:
+            response = client.get("https://check.torproject.org/api/ip")
+            data = response.json()
+            is_tor = data.get("IsTor", False)
+            
+            if is_tor:
+                _http_client = client  # Reuse the SAME client that worked
+                logger.info(f"Tor session initialized successfully on port {port}")
+                return True
+            else:
+                client.close()
+                logger.warning("Tor connectivity check failed: IsTor returned False")
+                return False
+                
+        except Exception as e:
+            client.close()
+            logger.error(f"The HTTP call failed: {e}")
+            return False
+            
+    except Exception as e:
+        logger.error(f"Failed to create transport: {e}")
+        return False
+
+
+
+def init_untested_tor(port: int = 9050) -> bool:
+    """
+    This is the same thing as init_tor_session, but without testing if Tor is actually working.
+    """
+    global _http_client
+    
+    transport = httpx_socks.SyncProxyTransport(
+        proxy_type=ProxyType.SOCKS5,
+        proxy_host="127.0.0.1",
+        proxy_port=port,
+    )
+    _http_client = httpx.Client(transport=transport, http2=True, timeout=10)
+    return True
+
+
+
+def get_http_session() -> httpx.Client:
+    """Return the global _http_client or raise RuntimeError."""
+    if _http_client is None:
+        raise RuntimeError("HTTP session not initialized. Call init_tor_session(port) first.")
+    return _http_client
+
+
+def close_http_session():
+    """Close and reset the global _http_client."""
+    global _http_client
+    if _http_client is not None:
+        _http_client.close()
+    _http_client = None
+

core/services/networking/httpx/make_request.py

@@ -0,0 +1,151 @@
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType, BackoffStrategy
+from core.errors.logger import logger
+from core.services.networking.httpx.classify_response import classify_response
+
+from typing import Optional
+import httpx
+import asyncio
+import json
+import time
+import socket
+
+def make_request(
+    method: str,
+    url: str,
+    client: httpx.Client,
+    payload: Optional[dict] = None,
+) -> ApiResponse:
+
+    if method == "post" and not payload:
+        return ApiResponse(valid=False, error_type=ErrorType.INVALID_INPUT, message="Can't have a POST request without a payload")
+
+    initial_result = _make_request(
+        method=method,
+        url=url,
+        client=client,
+        payload=payload
+    )
+
+    if initial_result.valid:
+        return initial_result
+
+    logger.error(f"{method.upper()} request failed: {initial_result.error_type}")
+
+    # Immediate Retry
+    if initial_result.backoff_strategy == BackoffStrategy.RETRY_IMMEDIATE:
+        time.sleep(3)
+        second_result = _make_request(
+            method=method,
+            url=url,
+            client=client,
+            payload=payload
+        )
+        if second_result.valid:
+            return second_result
+        
+        if second_result.backoff_strategy == BackoffStrategy.RETRY_IMMEDIATE:
+            logger.error("We are avoiding retrying immediately twice.")
+            second_result.backoff_strategy = BackoffStrategy.RETRY_EXPONENTIAL
+            return second_result
+
+    # Delayed Retry
+    elif initial_result.backoff_strategy == BackoffStrategy.RETRY_EXPONENTIAL:
+        time.sleep(10)
+        second_result = _make_request(
+            method=method,
+            url=url,
+            client=client,
+            payload=payload
+        )
+        return second_result
+
+    # Invalid:
+    elif initial_result.backoff_strategy == BackoffStrategy.FIX_CLIENT_SIDE_INFO and initial_result.error_type == ErrorType.INVALID_REQUEST:
+        logger.error("This was classified Invalid.  This might be a mistake of doing GET when it's POST")
+        return switch_get_and_post(method=method, url=url, client=client, payload=payload)
+
+    return initial_result
+
+def _make_request(
+    method: str,
+    url: str,
+    client: httpx.Client,
+    payload: Optional[dict] = None,
+) -> ApiResponse:
+    logger.debug(f"Executing {method.upper()} to {url}")
+    try:
+        if method.lower() == "get":
+            response = client.get(url)
+        else:
+            response = client.post(url, json=payload)
+
+        return classify_response(response)
+
+    except httpx.TimeoutException as e:
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.NETWORK_ERROR,
+            message=f"Request timeout: {str(e)}",
+            backoff_strategy=BackoffStrategy.RETRY_EXPONENTIAL
+        )
+
+
+    except httpx.ConnectError as e:
+        if isinstance(e.__cause__, socket.gaierror):
+            gaierror = e.__cause__
+            if gaierror.errno in (-3, -11):  # EAI_AGAIN
+                # Transient DNS failure
+                return ApiResponse(valid=False, error_type=ErrorType.DNS_TEMPORARY, backoff_strategy=BackoffStrategy.RETRY_EXPONENTIAL)
+            else:
+                # Permanent DNS failure (bad domain)
+                return ApiResponse(valid=False, error_type=ErrorType.DNS_PERMANENT, backoff_strategy=BackoffStrategy.NO_RETRY)
+        else:
+            # Non-DNS connection issue,
+            error_type = ErrorType.CONNECT_ERROR
+            return ApiResponse(valid=False, error_type=error_type, backoff_strategy=BackoffStrategy.RETRY_EXPONENTIAL)
+
+
+    except httpx.ProxyError as e:
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.TOR_NOT_WORKING,
+            message=f"Proxy error: {str(e)}",
+            backoff_strategy=BackoffStrategy.NO_RETRY
+        )
+
+    except httpx.SSLError as e:
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.NETWORK_ERROR,
+            message=f"SSL/certificate error: {str(e)}",
+            backoff_strategy=BackoffStrategy.NO_RETRY  # Permanent cert issue
+        )
+
+    except httpx.RequestError as e:
+        # Catches any other httpx request errors not covered above
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.NETWORK_ERROR,
+            message=f"Request error: {str(e)}",
+            backoff_strategy=BackoffStrategy.RETRY_EXPONENTIAL
+        )
+
+
+def switch_get_and_post(method: str, url: str, client: httpx.Client, payload: dict) -> ApiResponse:
+    if method == "get":
+        new_method = "post"
+    else:
+        new_method = "get"
+    second_result = _make_request(
+        method=new_method,
+        url=url,
+        client=client,
+        payload=payload
+    )
+    if second_result.valid:
+        logger.info("Our strategy worked! This is a mistake of GET/POST")
+        return second_result
+    else:
+        # nevermind,
+        logger.info("Our strategy of switching GET/POST did NOT work.")
+        return initial_result

core/services/networking/httpx/parallel_threading.py

@@ -0,0 +1,52 @@
+from core.services.networking.httpx.make_request import make_request
+from core.services.networking.httpx.httpx_client import init_tor_session, get_http_session, init_untested_tor
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType, BackoffStrategy
+from core.errors.logger import logger
+
+import time
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime
+import httpx
+import httpx_socks # pip install httpx-socks
+from httpx_socks import ProxyType
+
+def if_a_new_client_is_needed():
+    transport = httpx_socks.SyncProxyTransport(
+        proxy_type=ProxyType.SOCKS5,
+        proxy_host="127.0.0.1",
+        proxy_port=9050,
+    )
+    client = httpx.Client(transport=transport, http2=True, timeout=10)
+    return client
+
+
+ENDPOINTS = {
+    "locations": "https://api.hydraveil.net/api/v1/locations",
+    "operators": "https://api.hydraveil.net/api/v1/operators",
+    "client": "https://api.hydraveil.net/api/v1/platforms/linux-x86_64/appimage/client-versions",
+    "sub_plans": "https://api.hydraveil.net/api/v1/subscription-plans",
+    "applications": "https://api.hydraveil.net/api/v1/platforms/linux-x86_64/applications",
+}
+
+
+def parallel_thread(client: httpx.Client, endpoints: dict = ENDPOINTS) -> dict:
+    start_total = time.time()
+
+    with ThreadPoolExecutor(max_workers=5) as executor:
+        futures = {
+            key: executor.submit(
+                make_request,
+                method="get",
+                url=url,
+                client=client
+            )
+            for key, url in endpoints.items()
+        }
+
+        # Wait for all jobs to finish and collect results (BLOCKING)
+        results = {key: future.result() for key, future in futures.items()}
+
+    total_parallel = time.time() - start_total
+    logger.info(f"\nTotal time for all API calls: {total_parallel:.2f}s")
+    return results
+

core/services/networking/tor_tools/bootstrap.py

@@ -0,0 +1,214 @@
+from core.services.networking.tor_tools.monitor_bootstrap import monitor_bootstrap_progress
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+from core.services.networking.tor_tools import ports
+from core.Constants import Constants
+from essentials.observers.ConnectionObserver import ConnectionObserver
+from core.errors.logger import logger
+
+import asyncio
+import subprocess
+import socket
+import re
+import time
+from typing import Optional
+from enum import Enum
+import atexit
+import threading
+
+_tor_process: Optional[subprocess.Popen] = None
+
+
+GIVE_UP_WITH_NO_DIAGNOSIS = 2 # attempts
+
+ERROR_PATTERNS = {
+    "data_dir_conflict": "another Tor process is running with the same data directory",
+    "permission_denied": "permission denied",
+    "port_in_use": r"(address already in use|port.*in use|cannot assign requested address)",
+    "not_installed": r"(no such file|command not found|tor: not found)",
+}
+
+class BootstrapStrategy(Enum):
+    """Explicit retry strategies"""
+    FRESH_PORT = "fresh_port"
+    NEW_FOLDER_AND_PORT = "new_folder_and_port"
+    REINSTALL_AND_RETRY = "reinstall_and_retry"
+    GIVE_UP = "give_up"
+
+
+def bootstrap(
+    port: int, 
+    observer: ConnectionObserver,
+    use_new_folder: bool = False,
+    max_attempts: int = 3,
+) -> ApiResponse:
+    """
+    Attempt to bootstrap Tor with retry logic and diagnostic strategies.
+    
+    Uses a multi-attempt loop with failure diagnosis to determine the best
+    recovery action: retry with fresh port, reinstall Tor, use new folder, etc.
+    
+    Args:
+        port: Initial port for Tor connection
+        observer: ConnectionObserver for monitoring bootstrap progress
+        use_new_folder: Whether to use a fresh data directory
+        max_attempts: Maximum retry attempts before giving up
+    
+    Returns:
+        ApiResponse: Success or failure with corresponding error type
+    """
+    logger.info("Starting bootstrap..")
+
+    # ===== Setup Variables =====
+    attempt = 0
+    current_port = port
+    
+    # ===== Retry Loop =====
+    while attempt < max_attempts:
+        # --- Execute Attempt ---
+        attempt += 1
+        logger.info(f"Bootstrap attempt {attempt}/{max_attempts}")
+        
+        if attempt <= GIVE_UP_WITH_NO_DIAGNOSIS:
+            logger.info("Last try with no diagnosis..")
+
+        result = _attempt_single_bootstrap(current_port, observer, use_new_folder)
+        
+        # --- Diagnose & Act ---
+        if result.valid:
+            return result
+        
+        diagnosis = diagnose_bootstrap_failure(result)
+        strategy = assign_an_action(diagnosis, attempt)
+        
+        # Execute the recommended strategy or give up
+        if strategy == BootstrapStrategy.GIVE_UP:
+            return result
+        elif strategy == BootstrapStrategy.FRESH_PORT:
+            current_port = ports.get_random_available_port()
+        elif strategy == BootstrapStrategy.NEW_FOLDER_AND_PORT:
+            current_port = ports.get_random_available_port()
+            use_new_folder = True
+        elif strategy == BootstrapStrategy.REINSTALL_AND_RETRY:
+            if not install_tor().valid:
+                return ApiResponse(valid=False, error_type=ErrorType.INSTALL_FAILED)
+            use_new_folder = True
+    
+    # ===== Exhausted All Attempts =====
+    return ApiResponse(valid=False, error_type=ErrorType.CANT_BOOTSTRAP)
+
+
+def diagnose_bootstrap_failure(result: ApiResponse) -> str | None:
+
+    if result.error_type == ErrorType.TOR_NOT_INSTALLED:
+        return "not_installed" # this never made it to output due to except block
+
+    if not result.data:
+        return None
+
+    raw_output = result.data
+
+    for error_key, pattern in ERROR_PATTERNS.items():
+        if re.search(pattern, raw_output, re.IGNORECASE):
+            detected_error = error_key
+            logger.error(f"Detected error: {error_key}")
+            return detected_error
+    return None
+
+
+def assign_an_action(diagnosis: Optional[str], attempt: int) -> BootstrapStrategy:
+    if diagnosis == "data_dir_conflict":
+        return BootstrapStrategy.NEW_FOLDER_AND_PORT
+    if diagnosis == "port_in_use":
+        return BootstrapStrategy.FRESH_PORT
+    if diagnosis == "not_installed":
+        return BootstrapStrategy.REINSTALL_AND_RETRY
+    
+    # No diagnosis case
+    if attempt <= GIVE_UP_WITH_NO_DIAGNOSIS:
+        logger.info(f"Attempt {attempt}: trying new folder/port despite no diagnosis")
+        return BootstrapStrategy.NEW_FOLDER_AND_PORT
+    
+    logger.info(f"Attempt {attempt}: giving up (no diagnosis)")
+    return BootstrapStrategy.GIVE_UP
+
+
+def _attempt_single_bootstrap(
+    port: int, 
+    observer: ConnectionObserver, 
+    use_new_folder: bool
+) -> ApiResponse:
+    """
+    Start Tor, monitor it, verify it works.
+    """
+    global _tor_process
+    
+    logger.info(f"Bootstrapping Tor on port {port}...")
+    
+    if port is None:
+        return ApiResponse(valid=False, error_type=ErrorType.INVALID_INPUT)
+    
+    command = _build_tor_command(port, use_new_folder)
+
+    try:
+        _tor_process = subprocess.Popen(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=2**16
+        )
+        
+        # Register cleanup hook once
+        atexit.register(_cleanup_on_exit)
+
+        logger.info("Ran the command")
+
+        # Call monitor (will be threaded internally)
+        success, error_output = monitor_bootstrap_progress(
+            _tor_process, observer, timeout_seconds=60
+        )
+
+        logger.error(f"Round completed! Success? {success}")
+
+        if success:
+            return ApiResponse(valid=True, port=port)
+        else:
+            return ApiResponse(
+                valid=False, 
+                data=error_output, 
+                error_type=ErrorType.CANT_BOOTSTRAP
+            )
+            
+    except FileNotFoundError:
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_NOT_INSTALLED)
+    except Exception as e:
+        error_msg = f"Bootstrap error: {str(e)}"
+        logger.error(error_msg)
+        logger.info(error_msg)
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_NOT_WORKING, data=str(e))
+
+
+def _cleanup_on_exit():
+    """Automatically called when app exits."""
+    global _tor_process
+    
+    if _tor_process and _tor_process.poll() is None:  # Still running
+        logger.info("Terminating Tor process...")
+        _tor_process.terminate()
+        try:
+            _tor_process.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            logger.warning("Force killing Tor...")
+            _tor_process.kill()
+
+
+def _build_tor_command(port: int, use_new_folder: bool) -> list[str]:
+    """Extract command building to reduce clutter."""
+    command = ["tor", "--SocksPort", str(port)]
+    
+    if use_new_folder:
+        command.extend(["--DataDirectory", f"/tmp/tor-{port}"])
+    
+    return command
+
+

core/services/networking/tor_tools/evaluate_tor.py

@@ -0,0 +1,101 @@
+from core.services.networking.httpx.httpx_client import init_tor_session, get_http_session, init_untested_tor
+from core.services.networking.tor_tools import tor_files
+from core.services.networking.tor_tools import ports
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+from core.models.Result import Result, ResultError
+from core.utils.run_commands import run_generic_command
+from core.Constants import Constants
+from essentials.observers.ConnectionObserver import ConnectionObserver
+from core.errors.logger import logger
+
+import httpx
+from httpx_socks import AsyncProxyTransport
+import asyncio
+import subprocess
+import socket
+from typing import Optional
+
+def evaluate_pre_existing_tor(observer: ConnectionObserver) -> ApiResponse:
+    """
+    Rank:
+        Module's Lead Orchestrator
+    
+    Purpose:
+        Evaluate if the pre-existing Tor Setup is working.  And if so, on what port.
+
+    Scope of Duty:
+        1) Is it installed?
+        2) Is the default port running ANYTHING?
+        3) Does the default port work?
+        4) Check the config, does that port work?
+    """
+    # # Step 1) Is Tor Installed?
+    if not is_installed('tor'):
+        return ApiResponse(valid=False, error_type=ResultError.TOR_NOT_INSTALLED)
+
+    # Step 2) Is Default Tor port Listening?
+    default_port_listening = ports.is_port_in_use(Constants.DEFAULT_TOR_PORT)
+    logger.info(f"Is ANYTHING listening on the Default Tor port? {default_port_listening}")
+
+    # Step 3) Let's try the default if so,
+    if default_port_listening:
+        # if this works, it setup a client for us to reuse
+        working_on_default = init_tor_session(port=Constants.DEFAULT_TOR_PORT)
+
+        if working_on_default:
+            return ApiResponse(valid=True, port=Constants.DEFAULT_TOR_PORT)
+
+    # Get info from Tor's torrc config:
+    config = tor_files.diagnose_config()
+
+    # Valid config testing,
+    if not config.valid:
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_INSTALLED_BUT_DEAD, port=config.port)
+    
+    return test_config_port(config)
+
+
+def test_config_port(config: ApiResponse) -> ApiResponse:
+    if not config.port:
+        return config
+
+    logger.info(f"We are testing the Torrc config port of {config.port}")
+
+    # Is it Default back again?
+    if config.port == Constants.DEFAULT_TOR_PORT:
+        logger.info(f"We already tried the config's port.")
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_INSTALLED_BUT_DEAD, port=config.port)
+
+    # Is that new port active?
+    listening = ports.is_port_in_use(config.port)
+    if not listening:
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_INSTALLED_BUT_DEAD, port=config.port)
+
+    # okay test it,
+    config_test = init_tor_session(port=config.port)
+
+    if config_test.valid:
+        return ApiResponse(valid=True, error_type=ErrorType.TOR_ON_DIFFERENT_PORT, port=config.port)
+    else:
+        return ApiResponse(valid=False, error_type=ErrorType.TOR_INSTALLED_BUT_DEAD, port=config.port)
+
+
+def is_installed(app_name: str) -> bool:
+    """
+    Rank:
+        Neutral Utility
+
+    Purpose:
+        Checks if an app is installed
+    """
+    try:
+        command = [app_name, "--version"]
+        human_readable_goal = f"Checking if {app_name} is installed"
+        result = run_generic_command(command, human_readable_goal, timeout=5)
+        return result.valid
+    except FileNotFoundError:
+        return False
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Error: exited with status {e.returncode}")
+        logger.error(f"stderr: {e.stderr.decode()}")
+        return False

core/services/networking/tor_tools/install_tor.py

@@ -0,0 +1,86 @@
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+from essentials.observers.ConnectionObserver import ConnectionObserver
+from core.errors.logger import logger
+
+import subprocess
+from typing import Optional
+
+
+def get_distro_package_manager() -> Optional[str]:
+    """
+    Detect Linux distribution and return appropriate package manager.
+    Returns: 'apt', 'dnf', 'pacman', or None if undetected.
+    """
+    try:
+        with open('/etc/os-release', 'r') as f:
+            content = f.read().lower()
+            
+            if 'debian' in content or 'ubuntu' in content or 'mint' in content:
+                return 'apt'
+            elif 'fedora' in content or 'rhel' in content or 'centos' in content:
+                return 'dnf'
+            elif 'arch' in content or 'manjaro' in content:
+                return 'pacman'
+    except FileNotFoundError:
+        pass
+    
+    return None
+
+
+async def install_tor() -> ApiResponse:
+    """
+    Prompt user to install Tor with pkexec, detecting distro for correct package manager.
+    """
+    pm = get_distro_package_manager()
+    
+    if pm is None:
+        logger.error("Could not detect Linux distribution")
+        return ApiResponse(
+            valid=False, 
+            error_type=ErrorType.TOR_NOT_INSTALLED,
+            data="Could not detect Linux distribution. Please install Tor manually."
+        )
+    
+    commands = {
+        'apt': ['pkexec', 'apt', 'install', '-y', 'tor'],
+        'dnf': ['pkexec', 'dnf', 'install', '-y', 'tor'],
+        'pacman': ['pkexec', 'pacman', '-S', '--noconfirm', 'tor'],
+    }
+    
+    cmd = commands[pm]
+    observer.notify("custom_message", f"Attempting to install Tor via {pm}...")
+    
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=120,  # Installation timeout
+        )
+        
+        if result.returncode == 0:
+            logger.info(f"Tor installed successfully via {pm}")
+            return ApiResponse(valid=True)
+        else:
+            error_msg = result.stderr.strip() or result.stdout.strip()
+            logger.error(f"Tor installation failed ({pm}): {error_msg}")
+            return ApiResponse(
+                valid=False,
+                error_type=ErrorType.TOR_NOT_INSTALLED,
+                data=error_msg
+            )
+    
+    except subprocess.TimeoutExpired:
+        logger.error("Tor installation timed out")
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.TOR_NOT_INSTALLED,
+            data="Installation timed out after 120 seconds"
+        )
+    except Exception as e:
+        logger.error(f"Tor installation failed: {e}")
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.TOR_NOT_INSTALLED,
+            data=str(e)
+        )

core/services/networking/tor_tools/monitor_bootstrap.py

@@ -0,0 +1,227 @@
+"""
+Purpose:
+    Monitor Tor bootstrap via re with timeouts, in a 2 phase system.
+
+does NOT:
+    Evaluate the output beyond bootstrap or not.
+
+Phase 1)
+    Require the bootstrap pattern to be seen within a given number of seconds or timeout immediately.
+    That time is NO_PROGRESS_TIMEOUT
+
+Phase 2)
+    Require a 1% or more bootstrap change in TIMEOUT_WINDOW. This is a continous check, every time the TIMEOUT_WINDOW passes.
+
+Return Scheme:
+    Tuple of (success: bool, raw_stderr_output: Optional[str])
+
+Return conditions:
+    - If success: (True, None)
+    - If timeout before 1%: (False, raw_output_string)
+    - If 100%: (True, None)
+
+Called by:
+    bootstrap
+"""
+
+
+from essentials.observers.ConnectionObserver import ConnectionObserver
+from core.errors.logger import logger
+
+# generic
+from dataclasses import dataclass, field
+import asyncio
+import subprocess
+import re
+import time
+from typing import Optional
+import queue
+import threading
+
+
+NO_PROGRESS_TIMEOUT = 15 # seconds
+TIMEOUT_WINDOW = 30 # seconds
+
+
+@dataclass
+class BootstrapState:
+    first_pattern_time: Optional[float] = None
+    last_progress_value: Optional[int] = None
+    last_progress_time: Optional[float] = None
+    stderr_buffer: list[str] = field(default_factory=list)
+
+
+def monitor_bootstrap_progress(
+    process: subprocess.Popen,
+    observer: ConnectionObserver,
+    timeout_seconds: int = 60
+) -> tuple[bool, Optional[str]]:
+    """
+    Monitor Tor bootstrap progress with a two-phase timeout strategy.
+    
+    Two-phase timeouts:
+    1. NO_PROGRESS_TIMEOUT: Fail if no bootstrap output appears within this window.
+    2. timeout_seconds: Fail if bootstrap doesn't reach 100% within this total time.
+    
+    Returns:
+        (True, None) if bootstrap completes successfully.
+        (False, stderr_output) if bootstrap fails or times out.
+    """
+    
+    # ============================================================================
+    # SETUP: Pattern matching, state tracking, timing
+    # ============================================================================
+    progress_pattern = re.compile(r'\[notice\]\s+Bootstrapped\s+(\d+)%')
+    state = BootstrapState()
+    start_time = time.time()
+    output_queue = queue.Queue()
+    
+    # ============================================================================
+    # STREAM READER: Helper function to read from stdout/stderr non-blockingly
+    # ============================================================================
+    def read_stream(stream, name):
+        """
+        Read lines from a stream and enqueue them for processing.
+        
+        Declared as a closure here (not at module level) to:
+        - Capture output_queue without global state or extra parameters
+        - Keep this reader function tightly scoped to monitor_bootstrap_progress
+        - Avoid polluting the module namespace with a helper used only here
+        
+        Args:
+            stream: The stdout or stderr handle from the Popen process.
+            name: Label for this stream ('stdout' or 'stderr') for debugging.
+        """
+        try:
+            for line in iter(stream.readline, ''):
+                if line:
+                    output_queue.put((name, line.strip()))
+        except Exception as e:
+            output_queue.put(('error', str(e)))
+    
+    # ============================================================================
+    # START READER THREADS: Non-blocking concurrent stream reading
+    # ============================================================================
+    # Daemon threads so they don't block app exit; the main loop feeds the queue.
+    stdout_thread = threading.Thread(
+        target=read_stream, 
+        args=(process.stdout, 'stdout'), 
+        daemon=True
+    )
+    stderr_thread = threading.Thread(
+        target=read_stream, 
+        args=(process.stderr, 'stderr'), 
+        daemon=True
+    )
+    stdout_thread.start()
+    stderr_thread.start()
+    
+    # ============================================================================
+    # MAIN MONITORING LOOP: Poll queue, evaluate progress, enforce timeouts
+    # ============================================================================
+    while True:
+        elapsed = time.time() - start_time
+        
+        # Phase 2: Overall timeout — bootstrap must complete within timeout_seconds
+        if elapsed > timeout_seconds:
+            logger.info(f"Bootstrap timeout after {timeout_seconds}s")
+            return False, "\n".join(state.stderr_buffer)
+        
+        # Phase 1: No-progress timeout — must see output within NO_PROGRESS_TIMEOUT
+        if state.first_pattern_time is None and elapsed > NO_PROGRESS_TIMEOUT:
+            logger.error(f"No bootstrap output detected within {NO_PROGRESS_TIMEOUT}s")
+            return False, "\n".join(state.stderr_buffer)
+        
+        try:
+            # Non-blocking queue read (timeout=1.0s to recheck timeouts periodically)
+            name, line = output_queue.get(timeout=1.0)
+            
+            # Handle stream read errors
+            if name == 'error':
+                logger.error(f"Stream read error: {line}")
+                return False, f"Error reading output: {line}"
+            
+            # Buffer all output (useful for debugging/error reporting)
+            logger.info(f"OUTPUT: {line}")
+            state.stderr_buffer.append(line)
+            
+            # ================================================================
+            # PROGRESS EVALUATION: Extract progress percentage and evaluate
+            # ================================================================
+            match = progress_pattern.search(line)
+            if match:
+                progress_pct = int(match.group(1))
+                result = _evaluate_progress(progress_pct, state, observer)
+                
+                # None = keep going; tuple = exit condition reached
+                if result is None:
+                    continue
+                
+                done, stderr_output = result
+                if done:
+                    logger.info("Bootstrap completed successfully")
+                    return True, None
+                else:
+                    logger.error("Bootstrap failed")
+                    return False, stderr_output
+        
+        except queue.Empty:
+            # Queue timeout (no output in 1s) — loop continues and checks timeouts above
+            # This prevents the loop from blocking indefinitely if Tor is silent
+            pass
+        
+        except Exception as e:
+            logger.error(f"Error monitoring bootstrap: {e}", exc_info=True)
+            return False, f"Error reading output: {str(e)}"
+
+
+def _evaluate_progress(
+    progress: int,
+    state: BootstrapState,
+    observer: ConnectionObserver
+) -> Optional[tuple[bool, Optional[str]]]:
+    """
+    Purpose:
+        Evaluate the progress value.
+    
+    
+    Returns:
+        None if we should keep going, or,
+        (success, stderr) tuple if we should exit.
+    """
+    # Record first time we see any pattern
+    if state.first_pattern_time is None:
+        state.first_pattern_time = time.time()
+        logger.error(f"Tor Bootstrap {progress}%")
+        return None
+
+    # Success?
+    if progress == 100:  # note: this 100% check has to be before the progress increase, or it will reset before hitting it,
+        logger.info("Tor Bootstrap 100%. Returning..")
+        return True, None
+
+    # Progress increased?
+    if state.last_progress_value is not None and progress > state.last_progress_value:
+        state.last_progress_value = progress
+        state.last_progress_time = time.time()
+        print(f"Tor Bootstrap {progress}%")
+        return None
+    
+    # First percentage?
+    if state.last_progress_value is None:
+        state.last_progress_value = progress
+        state.last_progress_time = time.time()
+        logger.error(f"Tor Bootstrap {progress}%")
+        return None
+    
+    # Stalled?
+    if progress == state.last_progress_value:
+        assert state.last_progress_time is not None
+        stall_duration = time.time() - state.last_progress_time
+        if stall_duration > TIMEOUT_WINDOW:
+            logger.error(f"Bootstrap stalled at {progress}% for {stall_duration:.1f}s")
+            return False, "\n".join(state.stderr_buffer)
+        return None
+        
+    return None
+

core/services/networking/tor_tools/ports.py

@@ -0,0 +1,221 @@
+from core.errors.logger import logger
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+from core.observers.ConnectionObserver import ConnectionObserver
+from core.observers.BaseObserver import BaseObserver
+from core.Constants import Constants
+
+from typing import Optional
+import subprocess
+import asyncio
+import subprocess
+import socket
+import httpx
+from httpx_socks import AsyncProxyTransport
+import json
+import re
+
+
+def is_port_in_use(port: int) -> bool:
+    """
+    Rank:
+        Neutral Utility
+
+    Purpose:
+        Quick socket check to see if port is already listening.
+    """
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.settimeout(0.5)
+            result = sock.connect_ex(('127.0.0.1', port))
+            return result == 0  # 0 means connection succeeded, port is in use
+    except Exception:
+        return False
+
+
+def get_random_available_port() -> int:
+    """
+    Rank:
+        Neutral Utility
+
+    Purpose:
+        Get a random available port.
+
+    In this project, Called by:
+        establish_tor_connection (other module)
+    """
+    socket_instance = socket.socket()
+    socket_instance.bind(("", 0))
+    port_number = socket_instance.getsockname()[1]
+    socket_instance.close()
+    return port_number
+
+
+
+# Not using this, due to sudo permission issues. But shelving it for later,
+def find_what_tor_listens_on() -> Optional[int]:
+    """
+    Rank:
+        Cross-Module Coordinator
+
+    Purpose:
+        Find what port the Tor process is listening on.
+
+    Method:
+        First tries ss
+        Then tries pidof
+
+    Called by:
+        evaluate_tor_install (other module)
+    """
+    logger.info("Evaluating Tor's current listening port..")
+
+    port = get_port_via_ss('tor')
+    if port and port is not None:
+        return port
+
+    port = get_port_via_pidof('tor')
+    return port
+
+
+def get_port_via_ss(app_name: str) -> Optional[int]:
+    """
+    Rank:
+        Neutral Utility
+    
+    Purpose:
+        Find what port an app is running on via ss.
+
+    Called by:
+        get_tor_listening_port
+    """
+    try:
+        result = subprocess.run(
+            ["ss", "-tlnp"],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        logger.info(f"ss raw output: {result.stdout.split('\n')}")
+        for line in result.stdout.split('\n'):
+            if app_name in line.lower():
+                # Parse any format: *:9050, 127.0.0.1:9050, 0.0.0.0:9050
+                match = re.search(r':(\d+)\s', line)
+                if match:
+                    return int(match.group(1))
+    except (subprocess.CalledProcessError, FileNotFoundError) as e:
+        logger.error(f"ss failed to find the port: {str(e)}")
+    logger.info("ss is returning None..")
+    return None
+
+
+def get_port_via_pidof(app_name: str) -> Optional[int]:
+    """
+    Rank:
+        Neutral Utility
+    
+    Purpose:
+        Find what port an app is running on via pidof.
+
+    Called by:
+        get_tor_listening_port
+    """
+    try:
+        # Get PID(s) of tor process
+        pidof = subprocess.run(
+            ["pidof", app_name],
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        pids = pidof.stdout.strip().split()
+        
+        # Check each PID for listening sockets
+        for pid in pids:
+            result = subprocess.run(
+                ["lsof", "-i", "-n", "-P", "-p", pid],
+                capture_output=True,
+                text=True
+            )
+            for line in result.stdout.split('\n'):
+                if 'LISTEN' in line:
+                    # Parse lines like: tor 1234 user 5u IPv4 12345 LISTEN *:9050
+                    match = re.search(r'\*:(\d+)', line)
+                    if match:
+                        return int(match.group(1))
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        pass
+    return None
+
+
+async def kill_via_fuser(port: int) -> bool:
+    """
+    Rank:
+        Neutral Utility
+    
+    Purpose:
+        Find and kill whatever is on a port via fuser.
+
+    Called by:
+        kill_9050
+    """
+    try:
+        result = subprocess.run(
+            ["fuser", "-k", f"{port}/tcp"],
+            capture_output=True,
+            timeout=5
+        )
+        await asyncio.sleep(0.5)  # Give it a moment
+        return True
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return False
+    except Exception as e:
+        logger.error("error", f"Failed to free port 9050: {e}")
+        return False
+
+
+def kill_via_lsof(port: int) -> bool:
+    """
+    Rank:
+        Neutral Utility
+    
+    Purpose:
+        Find and kill whatever is on a port via lsof.
+
+    Called by:
+        kill_9050
+    """
+    try:
+        result = subprocess.run(
+            ["lsof", f"-ti:{port}"],
+            capture_output=True,
+            text=True
+        )
+        pids = result.stdout.strip().split('\n')
+        for pid in pids:
+            os.kill(int(pid), signal.SIGKILL)
+        return True
+    except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
+        return False
+    except Exception as e:
+        logger.error(f"Failed to free port {port} with lsof: {e}")
+        return False
+
+
+async def kill_9050() -> bool:
+    """
+    Rank:
+        Cross-Module Coordinator
+
+    Purpose:
+        Kills whatever is on port 9050
+
+    Method:
+        First tries fuser
+        Then tries lsof
+    """
+    fuser_killed_it = await kill_via_fuser(Constants.DEFAULT_TOR_PORT)
+    if fuser_killed_it:
+        return True
+
+    return kill_via_lsof(Constants.DEFAULT_TOR_PORT)
+

core/services/networking/tor_tools/tor_files.py

@@ -0,0 +1,163 @@
+
+"""
+Manage Tor Files.
+
+Scope:
+    Tor Configs
+    Lock files
+"""
+
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+from core.errors.logger import logger
+
+# generic
+from pathlib import Path
+import re
+from enum import Enum
+from dataclasses import dataclass
+from typing import Optional, Any
+import subprocess
+
+
+def remove_lock_file() -> bool:
+    lock_file = Path.home() / ".tor" / "lock"
+    if lock_file.exists():
+        logger.info("Found stale lock file. Removing...")
+        try:
+            lock_file.unlink()
+            return True
+        except Exception as e:
+            logger.error("We could NOT remove the lock file in the default folder.")
+            return False
+    else:
+        return False
+
+
+
+def find_torrc() -> Optional[Path]:
+    """
+    Purpose:
+        Find torrc configuration file.
+
+    Method:
+        Checks /etc/tor/torrc first, then searches system-wide.
+
+    Returns:
+        Path object if found,
+        None otherwise.
+    """
+    # Check default location
+    default_path = Path("/etc/tor/torrc")
+    if default_path.exists():
+        return default_path
+    
+    # Fall back to system-wide search
+    try:
+        result = subprocess.run(
+            ["find", "/", "-name", "torrc", "-type", "f"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            stderr=subprocess.DEVNULL
+        )
+        torrc_files = [f for f in result.stdout.strip().split('\n') if f]
+        return Path(torrc_files[0]) if torrc_files else None
+    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
+        return None
+
+
+
+def diagnose_config() -> ApiResponse:
+    """Diagnose Tor configuration and find the SOCKS port."""
+    
+    torrc_path = Path("/etc/tor/torrc")
+    
+    # Step 1: Check if torrc exists
+    torrc_path = find_torrc()
+    
+    if not torrc_path:
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.TOR_FILE_MISSING,
+            message="Tor configuration file not found"
+        )
+    
+    # Step 2: Read the file
+    try:
+        with open(torrc_path, 'r') as f:
+            content = f.read()
+    except PermissionError:
+        return ApiResponse(
+            valid=False,
+            error_type=ErrorType.PERMISSION_ERROR,
+            message="Permission denied reading /etc/tor/torrc"
+        )
+    
+    # Step 3: Find SocksPort lines
+    socks_lines = [line.strip() for line in content.split('\n') if 'SocksPort' in line]
+    
+    if not socks_lines:
+        # No config found — using default
+        return ApiResponse(
+            valid=True,
+            error_type=ErrorType.TOR_COMMENTED_PORT,
+            message="No SocksPort configuration found - using default port 9050",
+            port=None
+        )
+    
+    # Check for uncommented SocksPort
+    for line in socks_lines:
+        if line.startswith('#'):
+            continue  # This line is commented
+        
+        # Found uncommented SocksPort — extract port
+        # Handles: "SocksPort 9050" or "SocksPort 127.0.0.1:9050"
+        match = re.search(r'SocksPort\s+(?:[\d.]+:)?(\d+)', line)
+        if match:
+            port = int(match.group(1))
+            return ApiResponse(
+                valid=True,
+                data=line,
+                message=f"Tor SOCKS configured to port {port}",
+                port=port
+            )
+    
+    # All SocksPort lines are commented
+    return ApiResponse(
+        valid=True,
+        error_type=ErrorType.TOR_COMMENTED_PORT,
+        message="All SocksPort configurations are commented - using default port 9050",
+        port=9050
+    )
+
+
+
+
+
+# I cut these, but might re-add them later:
+
+# def do_we_bootstrap_to_a_new_folder(result: ApiResponse) -> bool:
+#     if failed_to_remove_tor_lock_file():
+#         return True
+#     zombie_tor = result.error_type == ErrorType.TOR_INSTALLED_BUT_DEAD
+#     default_folder = default_tor_folder_exists()
+#     return sum([zombie_tor, default_folder]) >= 2
+
+# def does_default_tor_folder_exist():
+#     default_folder = Path.home() / ".tor"
+#     return Path.exists(default_folder)
+
+
+# def failed_to_remove_tor_lock_file() -> bool:
+#     lock_file = Path.home() / ".tor" / "lock"
+#     if lock_file.exists():
+#         logger.info("Found stale lock file. Removing...")
+#         try:
+#             lock_file.unlink()
+#             return False
+#         except Exception as e:
+#             logger.error("We could NOT remove the lock file in the default folder.")
+#             return True
+#     else:
+#         return False
+

core/services/networking/tor_tools/tor_orchestrator.py

@@ -0,0 +1,85 @@
+
+from core.services.networking.httpx.httpx_client import init_tor_session, get_http_session, init_untested_tor
+from core.services.networking.tor_tools.evaluate_tor import evaluate_pre_existing_tor
+from core.services.networking.tor_tools.install_tor import install_tor
+from core.services.networking.tor_tools import tor_files
+from core.services.networking.tor_tools import ports
+from core.services.networking.tor_tools.bootstrap import bootstrap
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+
+from core.Constants import Constants
+from essentials.observers.ConnectionObserver import ConnectionObserver
+from core.errors.logger import logger
+
+import asyncio
+import httpx
+from httpx_socks import AsyncProxyTransport
+# pip install httpx-socks
+
+
+def establish_tor_connection(observer: ConnectionObserver) -> ApiResponse:
+    """
+    Rank:
+        King Orchestrator
+    
+    Purpose:
+        Get a working port for a Tor connection, via any available means.
+    """
+    # Step 1) Find out if the defaults or any others work.
+    result = evaluate_pre_existing_tor(observer)
+
+    if result.valid:
+        return result
+
+    logger.info(f"Pre-existing Tor did not work, here's what category we have so far: {result.error_type}")
+
+    # Step 2) Install Tor if needed.
+    if result.error_type == ErrorType.TOR_NOT_INSTALLED:
+        installed = install_tor()
+        if not installed:
+            return ApiResponse(valid=False, error_type=ErrorType.REFUSAL_TO_INSTALL_TOR)
+
+    # Step 3) Remove the lock
+    removal_result = tor_files.remove_lock_file()
+
+    # Step 4) Get a new port:
+    port = ports.get_random_available_port()
+
+    # BOOTSTRAP TIME!
+    logger.info(f"""
+Ladies and Gentlemen, this is your captain speaking,
+
+We are strapped in for bootstrapping, please fasten your seats & tables to the upright position.
+
+Pre-flight Checks Perfomed:
+
+1) Tor is installed!
+2) Default port doesn't work of {Constants.DEFAULT_TOR_PORT}
+3) Config port checked of {result.port}
+4) Did we remove the lock file? {removal_result}
+5) We got a new port of {port}
+
+Have a good flight...
+    """)
+    bootstrap_results = bootstrap(
+        port=port,
+        observer=observer,
+        use_new_folder=False
+    )
+
+    if bootstrap_results.valid:
+        working_on_default = init_tor_session(port=bootstrap_results.port)
+        if working_on_default:
+            print(f"worked!! got a tor session on {port}")
+            return ApiResponse(valid=True, port=bootstrap_results.port)
+        else:
+            return ApiResponse(valid=False, error_type=ErrorType.UNKNOWN)
+
+
+# If I need this later:
+# async def create_async_tor_client(port: int) -> httpx.AsyncClient:
+#     """Create an AsyncClient ready for API requests."""
+#     transport = AsyncProxyTransport.from_url(f"socks5://127.0.0.1:{port}")
+#     return httpx.AsyncClient(transport=transport, http2=True, timeout=30)
+
+# potential: pip install httpx-socks[asyncio] if the async at bottom is used.