diff --git a/desktop/main.py b/desktop/main.py index e6337d1..aea5865 100644 --- a/desktop/main.py +++ b/desktop/main.py @@ -1,24 +1,42 @@ -import logging import os -import platform -import signal -import stat import sys import time +import stat +import signal +import platform import traceback +import logging import webbrowser -from logging.handlers import RotatingFileHandler +import http.client +import threading +import secrets +import hashlib +from collections import deque from pathlib import Path +from logging.handlers import RotatingFileHandler +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlsplit +from typing import Optional APP_NAME = "angelhome" -APP_VERSION = "2.1.1" +APP_VERSION = "1.1.1" + +UPSTREAM_URL = "https://angel.33.backend-3.com" +NETWORK_MODE = "proxy" + +LOCAL_PROXY_HOST = "127.0.0.1" +LOCAL_PROXY_PORT = 0 +LOCAL_ROUTE_CODE = "app" + +CALLS_HTTPS_HOST = "calls.backend-3.com" +CALLS_WSS_HOST = "calls.backend-3.com" + +ALLOWED_PATH_PREFIXES = [ + f"/p/{LOCAL_ROUTE_CODE}", + f"/p/{LOCAL_ROUTE_CODE}/", +] -START_URL = "https://angel.33.backend-3.com" -ALLOWED_HOSTS = {"angel.33.backend-3.com"} -ALLOW_SUBDOMAINS = False -ALLOWED_PATH_PREFIXES = ["/"] OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False - OFF_THE_RECORD = False WINDOW_TITLE = APP_NAME @@ -32,10 +50,10 @@ LOG_BACKUP_COUNT = 3 ALLOW_CLIPBOARD = False ALLOW_GEOLOCATION = False ALLOW_NOTIFICATIONS = False -ALLOW_MEDIA_AUDIO_CAPTURE = False +ALLOW_MEDIA_AUDIO_CAPTURE = True ALLOW_MEDIA_VIDEO_CAPTURE = False ALLOW_SCREEN_CAPTURE = False -ALLOW_WEBRTC = False +ALLOW_WEBRTC = True ENABLE_SINGLE_INSTANCE_IPC = True IPC_CONNECT_TIMEOUT_MS = 1500 @@ -43,40 +61,76 @@ ACTIVATE_WINDOW_MESSAGE = "ACTIVATE" MAX_RENDER_CRASHES = 3 RENDER_CRASH_WINDOW_SECONDS = 60 - -BLOCK_UNKNOWN_SCHEMES = True - -# ---------- Logging / runtime behavior ---------- -# In production/frozen mode, default to no console logging. ENABLE_CONSOLE_LOGGING_WHEN_FROZEN = False - -# If True, redact path too, leaving scheme://host[:port] only. +ENABLE_JS_CONSOLE_LOGGING = False REDACT_URL_PATH = False - -# When frozen on Windows, silence stdout/stderr by default. DETACH_STDIO_WHEN_FROZEN = True +MAX_BODY = 2_000_000 +PROXY_TIMEOUT_SECONDS = 20 +ALLOWED_PROXY_METHODS = {"GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS"} +LOCAL_PROXY_CSRF_HEADER = "X-Angelhome-Proxy-Token" +LOCAL_PROXY_CSRF_TOKEN = secrets.token_urlsafe(32) + +ALLOW_QTWEBENGINE_NO_SANDBOX = False + +RATE_LIMIT_WINDOW_SECONDS = 10.0 +RATE_LIMIT_MAX_REQUESTS = 120 +RATE_LIMIT_MAX_BURST = 40 + +HOP_HEADERS = { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "proxy-connection", +} + +DROP_REQUEST_HEADERS = { + "forwarded", + "x-forwarded-for", + "x-forwarded-host", + "x-forwarded-proto", + "x-real-ip", + "origin", +} + +DROP_RESPONSE_HEADERS = { + "server", + "x-powered-by", + "alt-svc", +} + +_runtime_proxy_port: Optional[int] = None +_runtime_start_url: Optional[str] = None +_proxy_server = None +_proxy_thread = None + def is_frozen() -> bool: return bool(getattr(sys, "frozen", False)) +def is_proxy_mode() -> bool: + return str(NETWORK_MODE).strip().lower() == "proxy" + + +def is_direct_mode() -> bool: + return str(NETWORK_MODE).strip().lower() == "direct" + + def detach_stdio_if_needed() -> None: - """ - Best-effort suppression of terminal output in frozen production builds. - Packaging still matters: - - PyInstaller: use --noconsole / windowed - - cx_Freeze: use Win32GUI base - """ try: if not is_frozen(): return if not DETACH_STDIO_WHEN_FROZEN: return - devnull_r = open(os.devnull, "r", encoding="utf-8", errors="ignore") devnull_w = open(os.devnull, "a", encoding="utf-8", errors="ignore") - try: sys.stdin = devnull_r except Exception: @@ -136,7 +190,16 @@ RUNTIME_DIR = ensure_dir(DATA_DIR / "runtime", 0o700) CACHE_DIR = ensure_dir(DATA_DIR / "cache", 0o700) STORAGE_DIR = ensure_dir(DATA_DIR / "storage", 0o700) LOCK_FILE = RUNTIME_DIR / f"{APP_NAME}.lock" -IPC_SOCKET_NAME = f"{APP_NAME}_single_instance_socket" + + +def build_ipc_socket_name() -> str: + uid = os.getuid() if hasattr(os, "getuid") else "nouid" + seed = f"{APP_NAME}|{base_dir()}|{uid}" + digest = hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16] + return f"{APP_NAME}_single_instance_{digest}" + + +IPC_SOCKET_NAME = build_ipc_socket_name() def configure_runtime_environment() -> None: @@ -148,6 +211,10 @@ def configure_runtime_environment() -> None: if flag not in flags: flags.append(flag) + def remove_flag(flag: str) -> None: + while flag in flags: + flags.remove(flag) + os.environ.setdefault("QT_AUTO_SCREEN_SCALE_FACTOR", "1") os.environ.setdefault("QT_ENABLE_HIGHDPI_SCALING", "1") @@ -161,9 +228,7 @@ def configure_runtime_environment() -> None: pass os.environ["XDG_RUNTIME_DIR"] = str(RUNTIME_DIR) - os.environ.setdefault("QTWEBENGINE_DISABLE_SANDBOX", "1") - add_flag("--no-sandbox") add_flag("--disable-extensions") add_flag("--disable-plugins") add_flag("--disable-background-networking") @@ -173,6 +238,14 @@ def configure_runtime_environment() -> None: add_flag("--no-first-run") add_flag("--password-store=basic") add_flag("--use-mock-keychain") + + if ALLOW_QTWEBENGINE_NO_SANDBOX: + os.environ["QTWEBENGINE_DISABLE_SANDBOX"] = "1" + add_flag("--no-sandbox") + else: + os.environ.pop("QTWEBENGINE_DISABLE_SANDBOX", None) + remove_flag("--no-sandbox") + elif system in {"windows", "darwin"}: add_flag("--disable-extensions") add_flag("--disable-plugins") @@ -187,6 +260,8 @@ def configure_runtime_environment() -> None: if flags: os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags) + else: + os.environ.pop("QTWEBENGINE_CHROMIUM_FLAGS", None) detach_stdio_if_needed() @@ -213,46 +288,64 @@ def safe_url_string(url) -> str: return "" +def current_proxy_port() -> int: + if _runtime_proxy_port is None: + raise RuntimeError("Local proxy port is not initialized") + return _runtime_proxy_port + + +def make_proxy_start_url(port: int) -> str: + return f"http://{LOCAL_PROXY_HOST}:{port}/p/{LOCAL_ROUTE_CODE}/" + + +def current_start_url() -> str: + if _runtime_start_url is None: + raise RuntimeError("Start URL is not initialized") + return _runtime_start_url + + +def initialize_start_url() -> None: + global _runtime_start_url + if is_proxy_mode(): + _runtime_start_url = make_proxy_start_url(current_proxy_port()) + else: + _runtime_start_url = UPSTREAM_URL + + def redact_url(url: QUrl | str, hide_path: bool = REDACT_URL_PATH) -> str: try: qurl = url if isinstance(url, QUrl) else QUrl(str(url)) if not qurl.isValid(): return "" - scheme = (qurl.scheme() or "").lower() host = qurl.host() or "" port = qurl.port() path = qurl.path() or "/" - if not scheme or not host: return "" - if port and port > 0: base = f"{scheme}://{host}:{port}" else: base = f"{scheme}://{host}" - if hide_path: return base - return f"{base}{path}" except Exception: return "" def configure_logging() -> logging.Logger: - logger = logging.getLogger(APP_NAME) - logger.setLevel(LOG_LEVEL) - - if logger.handlers: - return logger + app_logger = logging.getLogger(APP_NAME) + app_logger.setLevel(LOG_LEVEL) + if app_logger.handlers: + return app_logger fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s") if not (is_frozen() and not ENABLE_CONSOLE_LOGGING_WHEN_FROZEN): sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) - logger.addHandler(sh) + app_logger.addHandler(sh) fh = RotatingFileHandler( str(LOG_DIR / f"{APP_NAME}.log"), @@ -261,10 +354,9 @@ def configure_logging() -> logging.Logger: encoding="utf-8", ) fh.setFormatter(fmt) - logger.addHandler(fh) - - logger.propagate = False - return logger + app_logger.addHandler(fh) + app_logger.propagate = False + return app_logger logger = configure_logging() @@ -277,19 +369,10 @@ def is_root() -> bool: return False -def is_allowed_host(host: str) -> bool: - host = (host or "").lower().strip() - - if host in ALLOWED_HOSTS: - return True - - if ALLOW_SUBDOMAINS: - for allowed in ALLOWED_HOSTS: - allowed = (allowed or "").lower().strip() - if host == allowed or host.endswith("." + allowed): - return True - - return False +def upstream_parts(): + parsed = urlsplit(UPSTREAM_URL) + base_path = parsed.path or "/" + return parsed, base_path.rstrip("/") def is_allowed_path(path: str) -> bool: @@ -297,13 +380,14 @@ def is_allowed_path(path: str) -> bool: return any(path.startswith(prefix) for prefix in ALLOWED_PATH_PREFIXES) -def is_allowed_url(url: QUrl) -> bool: +def is_allowed_local_proxy_url(url: QUrl) -> bool: try: + if not (url and url.isValid()): + return False return bool( - url - and url.isValid() - and url.scheme().lower() == "https" - and is_allowed_host(url.host()) + (url.scheme() or "").lower() == "http" + and (url.host() or "").lower() == LOCAL_PROXY_HOST + and url.port() in (-1, current_proxy_port()) and is_allowed_path(url.path()) and not url.userName() and not url.password() @@ -312,19 +396,98 @@ def is_allowed_url(url: QUrl) -> bool: return False +def is_allowed_upstream_url(url: QUrl) -> bool: + try: + if not (url and url.isValid()): + return False + + parsed, base_path = upstream_parts() + expected_scheme = (parsed.scheme or "").lower() + expected_host = (parsed.hostname or "").lower() + expected_port = parsed.port or (443 if expected_scheme == "https" else 80) + + if (url.scheme() or "").lower() != expected_scheme: + return False + if (url.host() or "").lower() != expected_host: + return False + if url.userName() or url.password(): + return False + + actual_port = url.port() if url.port() != -1 else expected_port + if actual_port != expected_port: + return False + + path = url.path() or "/" + if base_path and base_path != "/": + return path == base_path or path.startswith(base_path + "/") + return path.startswith("/") + except Exception: + return False + + +def is_allowed_calls_https_url(url: QUrl) -> bool: + try: + if not (url and url.isValid()): + return False + if (url.scheme() or "").lower() != "https": + return False + if (url.host() or "").lower() != CALLS_HTTPS_HOST: + return False + if url.userName() or url.password(): + return False + path = url.path() or "/" + return ( + path == "/api/call/session" + or path == "/healthz" + or path == "/status" + or path == "/logs" + or path.startswith("/api/call/") + ) + except Exception: + return False + + +def is_allowed_calls_wss_url(url: QUrl) -> bool: + try: + if not (url and url.isValid()): + return False + if (url.scheme() or "").lower() != "wss": + return False + if (url.host() or "").lower() != CALLS_WSS_HOST: + return False + if url.userName() or url.password(): + return False + return (url.path() or "/") == "/ws" + except Exception: + return False + + +def is_allowed_navigation_url(url: QUrl) -> bool: + return is_allowed_local_proxy_url(url) if is_proxy_mode() else is_allowed_upstream_url(url) + + +def is_allowed_request_url(url: QUrl) -> bool: + return ( + is_allowed_navigation_url(url) + or is_allowed_calls_https_url(url) + or is_allowed_calls_wss_url(url) + ) + + def validate_configuration() -> None: if not APP_NAME or not isinstance(APP_NAME, str): raise ValueError("APP_NAME must be a non-empty string") + if NETWORK_MODE not in {"proxy", "direct"}: + raise ValueError("NETWORK_MODE must be 'proxy' or 'direct'") + if not LOCAL_ROUTE_CODE or "/" in LOCAL_ROUTE_CODE: + raise ValueError("LOCAL_ROUTE_CODE must be a simple path segment") - if not ALLOWED_HOSTS or not isinstance(ALLOWED_HOSTS, (set, list, tuple)): - raise ValueError("ALLOWED_HOSTS must be a non-empty set/list/tuple") + upstream = urlsplit(UPSTREAM_URL) + if upstream.scheme not in {"https", "http"} or not upstream.hostname: + raise ValueError(f"Invalid UPSTREAM_URL: {UPSTREAM_URL}") - qurl = QUrl(START_URL) - if not qurl.isValid(): - raise ValueError(f"Invalid START_URL: {START_URL}") - - if not is_allowed_url(qurl): - raise ValueError("START_URL is not allowed by current security policy") + if is_direct_mode() and upstream.scheme != "https": + raise ValueError("Direct mode requires HTTPS upstream for session safety") def open_in_system_browser(url: QUrl) -> None: @@ -339,56 +502,402 @@ def open_in_system_browser(url: QUrl) -> None: logger.exception("Failed to open URL in system browser: %s", redact_url(url)) +def join_paths(base_path: str, rest_path: str) -> str: + base = (base_path or "").rstrip("/") + rest = rest_path or "/" + if not rest.startswith("/"): + rest = "/" + rest + return (base + rest) if base else rest + + +def rewrite_location(location: str): + if not location: + return location, False + + upstream = urlsplit(UPSTREAM_URL) + prefix = f"/p/{LOCAL_ROUTE_CODE}" + + try: + if location.startswith("/"): + return prefix + location, False + + parsed = urlsplit(location) + if parsed.scheme in ("http", "https") and parsed.hostname: + same_origin = ( + parsed.scheme == upstream.scheme + and parsed.hostname == upstream.hostname + and (parsed.port or (443 if parsed.scheme == "https" else 80)) + == (upstream.port or (443 if upstream.scheme == "https" else 80)) + ) + if same_origin: + path = parsed.path or "/" + base = upstream.path.rstrip("/") + if base and path.startswith(base + "/"): + path = path[len(base):] + elif base and path == base: + path = "/" + + new_location = prefix + path + if parsed.query: + new_location += "?" + parsed.query + if parsed.fragment: + new_location += "#" + parsed.fragment + return new_location, False + + return None, True + + return None, True + except Exception: + return None, True + + +def normalize_host_header(value: str) -> str: + return (value or "").strip().lower().rstrip(".") + + +def is_valid_local_host_header(value: str) -> bool: + host = normalize_host_header(value) + port = current_proxy_port() + allowed = { + f"{LOCAL_PROXY_HOST}:{port}", + LOCAL_PROXY_HOST, + f"localhost:{port}", + "localhost", + } + return host in allowed + + +class LocalRateLimiter: + def __init__(self, window_seconds: float, max_requests: int, max_burst: int): + self.window_seconds = float(window_seconds) + self.max_requests = int(max_requests) + self.max_burst = int(max_burst) + self._lock = threading.Lock() + self._events: dict[str, deque[float]] = {} + + def allow(self, key: str) -> bool: + now = time.monotonic() + burst_cutoff = now - 1.0 + window_cutoff = now - self.window_seconds + + with self._lock: + q = self._events.get(key) + if q is None: + q = deque() + self._events[key] = q + + while q and q[0] < window_cutoff: + q.popleft() + + total_recent = len(q) + burst_recent = 0 + for ts in reversed(q): + if ts >= burst_cutoff: + burst_recent += 1 + else: + break + + if total_recent >= self.max_requests or burst_recent >= self.max_burst: + return False + + q.append(now) + + if len(q) == 1: + stale_keys = [k for k, v in self._events.items() if not v or v[-1] < window_cutoff] + for stale_key in stale_keys: + self._events.pop(stale_key, None) + + return True + + +_rate_limiter = LocalRateLimiter( + RATE_LIMIT_WINDOW_SECONDS, + RATE_LIMIT_MAX_REQUESTS, + RATE_LIMIT_MAX_BURST, +) + + +class LocalProxyHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + server_version = APP_NAME + sys_version = "" + + def version_string(self) -> str: + return APP_NAME + + def log_message(self, format: str, *args) -> None: + logger.info("LocalProxy: " + format, *args) + + def _send_security_headers(self) -> None: + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header("X-Frame-Options", "DENY") + self.send_header("Cross-Origin-Opener-Policy", "same-origin") + self.send_header("Cross-Origin-Resource-Policy", "same-origin") + self.send_header("Cross-Origin-Embedder-Policy", "require-corp") + self.send_header( + "Permissions-Policy", + "geolocation=(), microphone=(), camera=(), display-capture=(), notifications=()", + ) + self.send_header("Cache-Control", "no-store") + + def _deny(self, status: int, message: str) -> None: + body = message.encode("utf-8") + self.send_response(status) + self._send_security_headers() + self.send_header("content-type", "text/plain; charset=utf-8") + self.send_header("content-length", str(len(body))) + self.send_header("connection", "close") + self.end_headers() + if self.command.upper() != "HEAD": + self.wfile.write(body) + self.close_connection = True + + def _client_key(self) -> str: + return self.client_address[0] if self.client_address else "unknown" + + def _validate_request(self) -> Optional[tuple[str, str]]: + if self.command.upper() not in ALLOWED_PROXY_METHODS: + return ("method", "method not allowed") + + if not _rate_limiter.allow(self._client_key()): + return ("rate", "too many requests") + + host = self.headers.get("host", "") + if not is_valid_local_host_header(host): + return ("host", "invalid host header") + + token = self.headers.get(LOCAL_PROXY_CSRF_HEADER, "") + if token != LOCAL_PROXY_CSRF_TOKEN: + return ("csrf", "forbidden") + + parts = self.path.split("/", 3) + if len(parts) < 3 or parts[1] != "p" or parts[2] != LOCAL_ROUTE_CODE: + return ("path", "not found") + + return None + + def do_GET(self): + self._handle() + + def do_POST(self): + self._handle() + + def do_PUT(self): + self._handle() + + def do_DELETE(self): + self._handle() + + def do_HEAD(self): + self._handle() + + def do_OPTIONS(self): + self._handle() + + def _handle(self): + validation_error = self._validate_request() + if validation_error: + kind, message = validation_error + if kind == "method": + return self._deny(405, message) + if kind == "rate": + body = b"too many requests" + self.send_response(429) + self._send_security_headers() + self.send_header("Retry-After", "5") + self.send_header("content-type", "text/plain; charset=utf-8") + self.send_header("content-length", str(len(body))) + self.send_header("connection", "close") + self.end_headers() + if self.command.upper() != "HEAD": + self.wfile.write(body) + self.close_connection = True + return + if kind == "host": + return self._deny(400, message) + if kind == "csrf": + return self._deny(403, message) + return self._deny(404, message) + + parts = self.path.split("/", 3) + rest = "/" + (parts[3] if len(parts) >= 4 else "") + upstream = urlsplit(UPSTREAM_URL) + + target = join_paths(upstream.path, rest.split("?", 1)[0]) + if "?" in self.path: + query = self.path.split("?", 1)[1] + target = target + ("?" if "?" not in target else "&") + query + + try: + content_length = int(self.headers.get("content-length") or "0") + except ValueError: + return self._deny(400, "invalid content-length") + + if content_length < 0: + return self._deny(400, "invalid content-length") + if content_length > MAX_BODY: + return self._deny(413, "payload too large") + + body = self.rfile.read(content_length) if content_length else b"" + + connection_cls = ( + http.client.HTTPSConnection + if upstream.scheme == "https" + else http.client.HTTPConnection + ) + + conn = connection_cls( + upstream.hostname, + upstream.port or (443 if upstream.scheme == "https" else 80), + timeout=PROXY_TIMEOUT_SECONDS, + ) + + headers = {} + for key, value in self.headers.items(): + key_lower = key.lower() + if key_lower in HOP_HEADERS: + continue + if key_lower in DROP_REQUEST_HEADERS: + continue + if key_lower == LOCAL_PROXY_CSRF_HEADER.lower(): + continue + headers[key] = value + + headers["Host"] = ( + upstream.hostname + if not upstream.port + else f"{upstream.hostname}:{upstream.port}" + ) + headers["Connection"] = "close" + + try: + conn.request(self.command, target, body=body, headers=headers) + response = conn.getresponse() + data = b"" if self.command.upper() == "HEAD" else response.read() + + blocked_redirect = False + response_headers = [] + + for key, value in response.getheaders(): + key_lower = key.lower() + + if key_lower in HOP_HEADERS: + continue + if key_lower in DROP_RESPONSE_HEADERS: + continue + + if key_lower == "location": + new_location, should_block = rewrite_location(value) + if should_block: + blocked_redirect = True + continue + response_headers.append( + ("Location", new_location if new_location is not None else value) + ) + continue + + response_headers.append((key, value)) + + if blocked_redirect: + return self._deny(403, "blocked external redirect") + + self.send_response(response.status, response.reason) + self._send_security_headers() + + sent_content_length = False + for key, value in response_headers: + if key.lower() == "content-length": + sent_content_length = True + if key.lower() == "connection": + continue + self.send_header(key, value) + + if not sent_content_length: + self.send_header("Content-Length", str(len(data))) + + self.send_header("Connection", "close") + self.end_headers() + + if self.command.upper() != "HEAD": + self.wfile.write(data) + + self.close_connection = True + + except Exception as exc: + logger.warning("Local proxy request failed: %s", exc) + return self._deny(502, "bad gateway") + finally: + conn.close() + + +def start_local_proxy_if_enabled() -> None: + global _proxy_server, _proxy_thread, _runtime_proxy_port + if not is_proxy_mode(): + return + if _proxy_server is not None: + return + + _proxy_server = ThreadingHTTPServer( + (LOCAL_PROXY_HOST, LOCAL_PROXY_PORT), + LocalProxyHandler, + ) + _runtime_proxy_port = int(_proxy_server.server_address[1]) + + _proxy_thread = threading.Thread( + target=_proxy_server.serve_forever, + daemon=True, + name=f"{APP_NAME}-proxy", + ) + _proxy_thread.start() + logger.info("Local proxy started: %s", make_proxy_start_url(_runtime_proxy_port)) + + +def stop_local_proxy_if_running() -> None: + global _proxy_server, _proxy_thread + try: + if _proxy_server is not None: + _proxy_server.shutdown() + _proxy_server.server_close() + except Exception: + logger.debug("Failed to stop local proxy cleanly", exc_info=True) + finally: + _proxy_server = None + _proxy_thread = None + + class RequestInterceptor(QWebEngineUrlRequestInterceptor): def interceptRequest(self, info: QWebEngineUrlRequestInfo) -> None: try: url = info.requestUrl() scheme = (url.scheme() or "").lower() host = (url.host() or "").lower() + port = url.port() method = bytes(info.requestMethod()).decode("utf-8", "ignore") + path = url.path() or "/" - if BLOCK_UNKNOWN_SCHEMES and scheme not in {"https", "wss"}: - logger.warning( - "[BLOCKED REQUEST] reason=scheme scheme=%s method=%s url=%s resource_type=%s", - scheme, - method, - redact_url(url), - info.resourceType(), - ) - info.block(True) - return + if is_proxy_mode() and is_allowed_local_proxy_url(url): + try: + info.setHttpHeader( + LOCAL_PROXY_CSRF_HEADER.encode("ascii"), + LOCAL_PROXY_CSRF_TOKEN.encode("ascii"), + ) + except Exception: + logger.debug("Could not attach proxy CSRF header", exc_info=True) - if not is_allowed_host(host): + if not is_allowed_request_url(url): logger.warning( - "[BLOCKED REQUEST] reason=host scheme=%s host=%s method=%s url=%s resource_type=%s", + "[BLOCKED REQUEST] scheme=%s host=%s port=%s method=%s path=%s url=%s resource_type=%s", scheme, host, + port, method, + path, redact_url(url), info.resourceType(), ) info.block(True) return - - if scheme == "https": - if not is_allowed_path(url.path()): - logger.warning( - "[BLOCKED REQUEST] reason=path method=%s url=%s resource_type=%s", - method, - redact_url(url), - info.resourceType(), - ) - info.block(True) - return - elif scheme == "wss": - if not is_allowed_host(host): - logger.warning( - "[BLOCKED REQUEST] reason=websocket-host method=%s url=%s", - method, - redact_url(url), - ) - info.block(True) - return except Exception: logger.exception("Error while intercepting request") try: @@ -428,7 +937,6 @@ class ActivationServer: def start(self) -> bool: if not ENABLE_SINGLE_INSTANCE_IPC: return True - try: QLocalServer.removeServer(self.server_name) except Exception: @@ -453,7 +961,6 @@ class ActivationServer: self.server.close() except Exception: logger.debug("Failed to close activation server", exc_info=True) - try: QLocalServer.removeServer(self.server_name) except Exception: @@ -466,18 +973,14 @@ class ActivationServer: socket_ = self.server.nextPendingConnection() if socket_ is None: continue - socket_.waitForReadyRead(500) data = bytes(socket_.readAll()).decode("utf-8", "ignore").strip() - logger.info("Received IPC message: %s", data) - if data == ACTIVATE_WINDOW_MESSAGE: try: self.activate_callback() except Exception: logger.exception("Failed to activate existing window") - socket_.disconnectFromServer() except Exception: logger.exception("Failed to handle IPC activation connection") @@ -486,13 +989,11 @@ class ActivationServer: def signal_existing_instance(server_name: str) -> bool: if not ENABLE_SINGLE_INSTANCE_IPC: return False - try: socket_ = QLocalSocket() socket_.connectToServer(server_name) if not socket_.waitForConnected(IPC_CONNECT_TIMEOUT_MS): return False - socket_.write(ACTIVATE_WINDOW_MESSAGE.encode("utf-8")) socket_.flush() socket_.waitForBytesWritten(IPC_CONNECT_TIMEOUT_MS) @@ -506,23 +1007,22 @@ def signal_existing_instance(server_name: str) -> bool: class LockedPage(QWebEnginePage): def __init__(self, profile: QWebEngineProfile, parent=None): super().__init__(profile, parent) - try: self.certificateError.connect(self._on_certificate_error) except Exception: logger.debug("certificateError signal unavailable", exc_info=True) - try: self.featurePermissionRequested.connect(self._on_feature_permission_requested) except Exception: logger.debug("featurePermissionRequested signal unavailable", exc_info=True) - try: self.windowCloseRequested.connect(self._on_window_close_requested) except Exception: logger.debug("windowCloseRequested signal unavailable", exc_info=True) def javaScriptConsoleMessage(self, level, message, line_number, source_id): + if not ENABLE_JS_CONSOLE_LOGGING: + return try: logger.warning( "[JS CONSOLE] level=%s source=%s line=%s message=%s", @@ -536,7 +1036,7 @@ class LockedPage(QWebEnginePage): def acceptNavigationRequest(self, url, nav_type, is_main_frame): try: - if is_allowed_url(url): + if is_allowed_navigation_url(url): return super().acceptNavigationRequest(url, nav_type, is_main_frame) logger.warning( @@ -545,10 +1045,8 @@ class LockedPage(QWebEnginePage): is_main_frame, nav_type, ) - if OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER and url.scheme().lower() in {"http", "https"}: open_in_system_browser(url) - return False except Exception: logger.exception("Error while processing navigation request") @@ -566,7 +1064,6 @@ class LockedPage(QWebEnginePage): def _on_feature_permission_requested(self, security_origin, feature): try: allowed = False - try: feature_enum = QWebEnginePage.Feature except Exception: @@ -588,7 +1085,6 @@ class LockedPage(QWebEnginePage): getattr(feature_enum, "ClipboardReadWrite", None): ALLOW_CLIPBOARD, getattr(feature_enum, "ClipboardRead", None): ALLOW_CLIPBOARD, } - if feature in feature_map: allowed = bool(feature_map[feature]) @@ -634,7 +1130,6 @@ class MainWindow(QMainWindow): def __init__(self): super().__init__() self.render_crash_times = [] - self.setWindowTitle(WINDOW_TITLE) self.resize(WINDOW_WIDTH, WINDOW_HEIGHT) @@ -710,20 +1205,28 @@ class MainWindow(QMainWindow): try: settings = self.browser.settings() settings.setAttribute( - QWebEngineSettings.WebAttribute.JavascriptCanOpenWindows, False + QWebEngineSettings.WebAttribute.JavascriptCanOpenWindows, + False, ) settings.setAttribute( - QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, False + QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, + False, ) settings.setAttribute( - QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, False + QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, + False, ) settings.setAttribute( - QWebEngineSettings.WebAttribute.FullScreenSupportEnabled, False + QWebEngineSettings.WebAttribute.FullScreenSupportEnabled, + False, ) - settings.setAttribute(QWebEngineSettings.WebAttribute.PluginsEnabled, False) settings.setAttribute( - QWebEngineSettings.WebAttribute.ErrorPageEnabled, True + QWebEngineSettings.WebAttribute.PluginsEnabled, + False, + ) + settings.setAttribute( + QWebEngineSettings.WebAttribute.ErrorPageEnabled, + True, ) for attr_name, value in ( @@ -761,10 +1264,9 @@ class MainWindow(QMainWindow): def _load_start_url(self) -> None: try: - qurl = QUrl(START_URL) - if not is_allowed_url(qurl): - raise ValueError("Refusing to load disallowed START_URL") - + qurl = QUrl(current_start_url()) + if not is_allowed_navigation_url(qurl): + raise ValueError("Refusing to load disallowed start URL") self.browser.setUrl(qurl) logger.info("Loading start URL: %s", redact_url(qurl)) except Exception: @@ -792,10 +1294,11 @@ class MainWindow(QMainWindow): logger.exception("Failed to activate main window") def _record_render_crash(self) -> int: - now = time.time() - self.render_crash_times.append(now) + current = time.time() + self.render_crash_times.append(current) self.render_crash_times = [ - t for t in self.render_crash_times if now - t <= RENDER_CRASH_WINDOW_SECONDS + ts for ts in self.render_crash_times + if current - ts <= RENDER_CRASH_WINDOW_SECONDS ] return len(self.render_crash_times) @@ -837,14 +1340,12 @@ class MainWindow(QMainWindow): @Slot(object, int) def _on_render_process_terminated(self, termination_status, exit_code): crash_count = self._record_render_crash() - logger.error( "Render process terminated: status=%s exit_code=%s recent_crashes=%s", termination_status, exit_code, crash_count, ) - if crash_count >= MAX_RENDER_CRASHES: try: QMessageBox.critical( @@ -887,7 +1388,6 @@ def install_signal_handlers(app: QApplication) -> None: signal.signal(signal.SIGINT, lambda *_: app.quit()) except Exception: logger.debug("SIGINT handler unavailable", exc_info=True) - try: signal.signal(signal.SIGTERM, lambda *_: app.quit()) except Exception: @@ -906,7 +1406,6 @@ def configure_qt_paths() -> None: if platform.system().lower() == "windows" else "QtWebEngineProcess" ) - candidates = [ resource_dir() / helper_name, resource_dir() / "PySide6" / "Qt" / "libexec" / helper_name, @@ -915,7 +1414,6 @@ def configure_qt_paths() -> None: base_dir() / "PySide6" / "Qt" / "libexec" / helper_name, base_dir() / "Qt" / "libexec" / helper_name, ] - for candidate in candidates: if candidate.exists() and candidate.is_file(): os.environ["QTWEBENGINEPROCESS_PATH"] = str(candidate) @@ -948,7 +1446,6 @@ def harden_runtime_paths() -> None: def main(): instance = None activation_server = None - try: validate_configuration() @@ -963,6 +1460,9 @@ def main(): QApplication.setApplicationName(APP_NAME) QApplication.setOrganizationName(APP_NAME) + start_local_proxy_if_enabled() + initialize_start_url() + app = QApplication(sys.argv) install_signal_handlers(app) @@ -992,7 +1492,10 @@ def main(): window = MainWindow() - activation_server = ActivationServer(IPC_SOCKET_NAME, window.activate_and_raise) + activation_server = ActivationServer( + IPC_SOCKET_NAME, + window.activate_and_raise, + ) activation_server.start() try: @@ -1007,50 +1510,58 @@ def main(): except Exception: logger.debug("Could not connect lock cleanup", exc_info=True) + try: + app.aboutToQuit.connect(stop_local_proxy_if_running) + except Exception: + logger.debug("Could not connect proxy cleanup", exc_info=True) + window.show() logger.info( - "Application started platform=%s python=%s frozen=%s version=%s resource_dir=%s base_dir=%s data_dir=%s", + "Application started platform=%s python=%s frozen=%s version=%s network_mode=%s resource_dir=%s base_dir=%s data_dir=%s start_url=%s sandbox_disabled=%s", platform.platform(), sys.version.split()[0], is_frozen(), APP_VERSION, + NETWORK_MODE, resource_dir(), base_dir(), DATA_DIR, + current_start_url(), + os.environ.get("QTWEBENGINE_DISABLE_SANDBOX") == "1", ) rc = app.exec() if activation_server: activation_server.close() - if instance: instance.release() + stop_local_proxy_if_running() sys.exit(rc) except Exception as exc: logger.critical("Fatal startup error: %s", exc, exc_info=True) - try: if QApplication.instance() is not None: QMessageBox.critical(None, APP_NAME, f"Fatal startup error:\n\n{exc}") except Exception: pass - try: if activation_server: activation_server.close() except Exception: pass - try: if instance: instance.release() except Exception: pass - + try: + stop_local_proxy_if_running() + except Exception: + pass sys.exit(1)