import os import sys import time import stat import signal import platform import traceback import logging import webbrowser 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 = "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}/", ] OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False OFF_THE_RECORD = False WINDOW_TITLE = APP_NAME WINDOW_WIDTH = 1200 WINDOW_HEIGHT = 800 LOG_LEVEL = logging.INFO LOG_MAX_BYTES = 2 * 1024 * 1024 LOG_BACKUP_COUNT = 3 ALLOW_CLIPBOARD = False ALLOW_GEOLOCATION = False ALLOW_NOTIFICATIONS = False ALLOW_MEDIA_AUDIO_CAPTURE = True ALLOW_MEDIA_VIDEO_CAPTURE = False ALLOW_SCREEN_CAPTURE = False ALLOW_WEBRTC = True ENABLE_SINGLE_INSTANCE_IPC = True IPC_CONNECT_TIMEOUT_MS = 1500 ACTIVATE_WINDOW_MESSAGE = "ACTIVATE" MAX_RENDER_CRASHES = 3 RENDER_CRASH_WINDOW_SECONDS = 60 ENABLE_CONSOLE_LOGGING_WHEN_FROZEN = False ENABLE_JS_CONSOLE_LOGGING = False REDACT_URL_PATH = False 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: 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: pass try: sys.stdout = devnull_w except Exception: pass try: sys.stderr = devnull_w except Exception: pass except Exception: pass def resource_dir() -> Path: if is_frozen() and hasattr(sys, "_MEIPASS"): return Path(sys._MEIPASS) if is_frozen(): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parent def base_dir() -> Path: if is_frozen(): return Path(sys.executable).resolve().parent return Path(__file__).resolve().parent def set_directory_permissions(path: Path, mode: int) -> None: try: if os.name == "posix": os.chmod(path, mode) except Exception: pass def ensure_dir(path: Path, mode: int | None = None) -> Path: path.mkdir(parents=True, exist_ok=True) if mode is not None: set_directory_permissions(path, mode) return path def harden_directory_permissions(paths: list[tuple[Path, int]]) -> None: for path, mode in paths: try: set_directory_permissions(path, mode) except Exception: pass DATA_DIR = ensure_dir(base_dir() / f"{APP_NAME}_data", 0o700) LOG_DIR = ensure_dir(DATA_DIR / "logs", 0o700) 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" 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: system = platform.system().lower() existing_flags = os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", "").strip() flags = existing_flags.split() if existing_flags else [] def add_flag(flag: str) -> 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") if system == "linux": ensure_dir(RUNTIME_DIR, 0o700) try: current_mode = stat.S_IMODE(os.stat(RUNTIME_DIR).st_mode) if current_mode != 0o700: os.chmod(RUNTIME_DIR, 0o700) except Exception: pass os.environ["XDG_RUNTIME_DIR"] = str(RUNTIME_DIR) add_flag("--disable-extensions") add_flag("--disable-plugins") add_flag("--disable-background-networking") add_flag("--disable-component-update") add_flag("--disable-sync") add_flag("--no-default-browser-check") 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") add_flag("--disable-background-networking") add_flag("--disable-component-update") add_flag("--disable-sync") add_flag("--no-default-browser-check") add_flag("--no-first-run") if not ALLOW_WEBRTC: add_flag("--disable-webrtc") if flags: os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags) else: os.environ.pop("QTWEBENGINE_CHROMIUM_FLAGS", None) detach_stdio_if_needed() configure_runtime_environment() from PySide6.QtCore import QCoreApplication, QLockFile, QStandardPaths, QTimer, QUrl, Slot from PySide6.QtNetwork import QLocalServer, QLocalSocket from PySide6.QtWidgets import QApplication, QMainWindow, QMessageBox from PySide6.QtWebEngineCore import ( QWebEngineCertificateError, QWebEnginePage, QWebEngineProfile, QWebEngineSettings, QWebEngineUrlRequestInfo, QWebEngineUrlRequestInterceptor, ) from PySide6.QtWebEngineWidgets import QWebEngineView def safe_url_string(url) -> str: try: return url.toString() if isinstance(url, QUrl) else str(url) except Exception: 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: 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) app_logger.addHandler(sh) fh = RotatingFileHandler( str(LOG_DIR / f"{APP_NAME}.log"), maxBytes=LOG_MAX_BYTES, backupCount=LOG_BACKUP_COUNT, encoding="utf-8", ) fh.setFormatter(fmt) app_logger.addHandler(fh) app_logger.propagate = False return app_logger logger = configure_logging() def is_root() -> bool: try: return hasattr(os, "geteuid") and os.geteuid() == 0 except Exception: 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: path = path or "/" return any(path.startswith(prefix) for prefix in ALLOWED_PATH_PREFIXES) def is_allowed_local_proxy_url(url: QUrl) -> bool: try: if not (url and url.isValid()): return False return bool( (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() ) except Exception: 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") upstream = urlsplit(UPSTREAM_URL) if upstream.scheme not in {"https", "http"} or not upstream.hostname: raise ValueError(f"Invalid UPSTREAM_URL: {UPSTREAM_URL}") 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: try: url_str = safe_url_string(url) ok = webbrowser.open(url_str) if ok: logger.info("Opened external link in system browser: %s", redact_url(url)) else: logger.warning("System browser did not confirm opening URL: %s", redact_url(url)) except Exception: 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 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_request_url(url): logger.warning( "[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 except Exception: logger.exception("Error while intercepting request") try: info.block(True) except Exception: pass class SingleInstance: def __init__(self, lock_path: Path): self.lock = QLockFile(str(lock_path)) try: self.lock.setStaleLockTime(0) except Exception: pass def acquire(self) -> bool: try: return self.lock.tryLock(0) except Exception: logger.exception("Failed to acquire single-instance lock") return False def release(self) -> None: try: self.lock.unlock() except Exception: logger.debug("Failed to release single-instance lock", exc_info=True) class ActivationServer: def __init__(self, server_name: str, activate_callback): self.server_name = server_name self.activate_callback = activate_callback self.server = None def start(self) -> bool: if not ENABLE_SINGLE_INSTANCE_IPC: return True try: QLocalServer.removeServer(self.server_name) except Exception: pass self.server = QLocalServer() self.server.newConnection.connect(self._handle_connection) if not self.server.listen(self.server_name): logger.warning( "Failed to start local activation server: %s", self.server.errorString() if hasattr(self.server, "errorString") else "unknown", ) return False logger.info("Activation server listening: %s", self.server_name) return True def close(self) -> None: try: if self.server is not None: self.server.close() except Exception: logger.debug("Failed to close activation server", exc_info=True) try: QLocalServer.removeServer(self.server_name) except Exception: logger.debug("Failed to remove activation server name", exc_info=True) @Slot() def _handle_connection(self) -> None: try: while self.server.hasPendingConnections(): 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") 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) socket_.disconnectFromServer() return True except Exception: logger.debug("Failed to signal existing instance", exc_info=True) return False 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", level, source_id, line_number, message, ) except Exception: logger.exception("Failed to log JavaScript console message") def acceptNavigationRequest(self, url, nav_type, is_main_frame): try: if is_allowed_navigation_url(url): return super().acceptNavigationRequest(url, nav_type, is_main_frame) logger.warning( "[BLOCKED NAVIGATION] url=%s is_main_frame=%s nav_type=%s", redact_url(url), 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") return False def createWindow(self, _type): logger.warning("[BLOCKED POPUP] createWindow request denied") return None @Slot() def _on_window_close_requested(self): logger.warning("[BLOCKED WINDOW CLOSE] windowCloseRequested denied") @Slot(QUrl, object) def _on_feature_permission_requested(self, security_origin, feature): try: allowed = False try: feature_enum = QWebEnginePage.Feature except Exception: feature_enum = None if feature_enum is not None: feature_map = { getattr(feature_enum, "Geolocation", None): ALLOW_GEOLOCATION, getattr(feature_enum, "Notifications", None): ALLOW_NOTIFICATIONS, getattr(feature_enum, "MediaAudioCapture", None): ALLOW_MEDIA_AUDIO_CAPTURE, getattr(feature_enum, "MediaVideoCapture", None): ALLOW_MEDIA_VIDEO_CAPTURE, getattr(feature_enum, "MediaAudioVideoCapture", None): ( ALLOW_MEDIA_AUDIO_CAPTURE and ALLOW_MEDIA_VIDEO_CAPTURE ), getattr(feature_enum, "DesktopVideoCapture", None): ALLOW_SCREEN_CAPTURE, getattr(feature_enum, "DesktopAudioVideoCapture", None): ( ALLOW_SCREEN_CAPTURE and ALLOW_MEDIA_AUDIO_CAPTURE ), getattr(feature_enum, "ClipboardReadWrite", None): ALLOW_CLIPBOARD, getattr(feature_enum, "ClipboardRead", None): ALLOW_CLIPBOARD, } if feature in feature_map: allowed = bool(feature_map[feature]) logger.warning( "[FEATURE PERMISSION] origin=%s feature=%s allowed=%s", redact_url(security_origin, hide_path=True), str(feature), allowed, ) self.setFeaturePermission( security_origin, feature, QWebEnginePage.PermissionPolicy.PermissionGrantedByUser if allowed else QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, ) except Exception: logger.exception("Failed to process feature permission request") try: self.setFeaturePermission( security_origin, feature, QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, ) except Exception: pass @Slot(QWebEngineCertificateError) def _on_certificate_error(self, error): try: logger.error( "[TLS ERROR] %s @ %s", error.description(), redact_url(error.url()), ) error.rejectCertificate() except Exception: logger.exception("Failed to reject certificate") class MainWindow(QMainWindow): def __init__(self): super().__init__() self.render_crash_times = [] self.setWindowTitle(WINDOW_TITLE) self.resize(WINDOW_WIDTH, WINDOW_HEIGHT) self.browser = QWebEngineView(self) self.setCentralWidget(self.browser) self.profile = self._build_profile() self.page = LockedPage(self.profile, self.browser) self.browser.setPage(self.page) self._apply_webengine_settings() self._connect_browser_signals() self._load_start_url() def _build_profile(self) -> QWebEngineProfile: try: ensure_dir(CACHE_DIR, 0o700) ensure_dir(STORAGE_DIR, 0o700) if OFF_THE_RECORD: profile = QWebEngineProfile(self.browser) logger.info("Using off-the-record web profile") else: profile = QWebEngineProfile(APP_NAME, self.browser) profile.setCachePath(str(CACHE_DIR)) profile.setPersistentStoragePath(str(STORAGE_DIR)) logger.info( "Using persistent profile cache=%s storage=%s", str(CACHE_DIR), str(STORAGE_DIR), ) try: if OFF_THE_RECORD: profile.setPersistentCookiesPolicy( QWebEngineProfile.PersistentCookiesPolicy.NoPersistentCookies ) else: profile.setPersistentCookiesPolicy( QWebEngineProfile.PersistentCookiesPolicy.AllowPersistentCookies ) except Exception: logger.debug("setPersistentCookiesPolicy unavailable", exc_info=True) try: profile.setHttpCacheType( QWebEngineProfile.HttpCacheType.MemoryHttpCache if OFF_THE_RECORD else QWebEngineProfile.HttpCacheType.DiskHttpCache ) except Exception: logger.debug("setHttpCacheType unavailable", exc_info=True) try: self.request_interceptor = RequestInterceptor() profile.setUrlRequestInterceptor(self.request_interceptor) logger.info("Installed QWebEngineUrlRequestInterceptor") except Exception: logger.exception("Failed to install request interceptor") raise try: profile.downloadRequested.connect(self._deny_download) except Exception: logger.debug("downloadRequested signal unavailable", exc_info=True) return profile except Exception as exc: logger.exception("Failed to build QWebEngineProfile") raise RuntimeError("Could not initialize browser profile") from exc def _apply_webengine_settings(self) -> None: try: settings = self.browser.settings() settings.setAttribute( QWebEngineSettings.WebAttribute.JavascriptCanOpenWindows, False, ) settings.setAttribute( QWebEngineSettings.WebAttribute.LocalContentCanAccessFileUrls, False, ) settings.setAttribute( QWebEngineSettings.WebAttribute.LocalContentCanAccessRemoteUrls, False, ) settings.setAttribute( QWebEngineSettings.WebAttribute.FullScreenSupportEnabled, False, ) settings.setAttribute( QWebEngineSettings.WebAttribute.PluginsEnabled, False, ) settings.setAttribute( QWebEngineSettings.WebAttribute.ErrorPageEnabled, True, ) for attr_name, value in ( ("HyperlinkAuditingEnabled", False), ("PdfViewerEnabled", False), ("PlaybackRequiresUserGesture", True), ("JavascriptCanAccessClipboard", ALLOW_CLIPBOARD), ("WebRTCPublicInterfacesOnly", not ALLOW_WEBRTC), ("ScreenCaptureEnabled", ALLOW_SCREEN_CAPTURE), ): try: attr = getattr(QWebEngineSettings.WebAttribute, attr_name) settings.setAttribute(attr, value) except Exception: logger.debug("WebAttribute unavailable: %s", attr_name, exc_info=True) logger.info("Applied WebEngine security settings") except Exception: logger.exception("Failed to apply WebEngine settings") raise def _connect_browser_signals(self) -> None: for signal_name, handler in ( ("loadStarted", self._on_load_started), ("loadProgress", self._on_load_progress), ("loadFinished", self._on_load_finished), ("urlChanged", self._on_url_changed), ("renderProcessTerminated", self._on_render_process_terminated), ("titleChanged", self._on_title_changed), ): try: getattr(self.browser, signal_name).connect(handler) except Exception: logger.debug("Could not connect signal: %s", signal_name, exc_info=True) def _load_start_url(self) -> None: try: 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: logger.exception("Failed to load start URL") QMessageBox.critical(self, APP_NAME, "Failed to load the application start URL.") raise def _deny_download(self, item) -> None: try: logger.warning("[BLOCKED DOWNLOAD] %s", redact_url(item.url())) item.cancel() except Exception: logger.exception("Failed while denying download request") def activate_and_raise(self) -> None: try: if self.isMinimized(): self.showNormal() else: self.show() self.raise_() self.activateWindow() logger.info("Activated existing main window") except Exception: logger.exception("Failed to activate main window") def _record_render_crash(self) -> int: current = time.time() self.render_crash_times.append(current) self.render_crash_times = [ ts for ts in self.render_crash_times if current - ts <= RENDER_CRASH_WINDOW_SECONDS ] return len(self.render_crash_times) @Slot() def _on_load_started(self): logger.info("Page load started") @Slot(int) def _on_load_progress(self, progress): logger.debug("Page load progress: %s%%", progress) @Slot(bool) def _on_load_finished(self, ok): current_url = redact_url(self.browser.url()) if ok: logger.info("Page load finished successfully: %s", current_url) else: logger.error("Page load failed: %s", current_url) QTimer.singleShot( 0, lambda: QMessageBox.warning( self, APP_NAME, "The page failed to load. Please verify network and server availability.", ), ) @Slot(QUrl) def _on_url_changed(self, url): logger.info("URL changed: %s", redact_url(url)) @Slot(str) def _on_title_changed(self, title): try: self.setWindowTitle(f"{WINDOW_TITLE} - {title}" if title else WINDOW_TITLE) except Exception: logger.exception("Failed to update window title") @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( self, APP_NAME, "The embedded browser process has crashed repeatedly.\n\n" "Automatic reload has been stopped to avoid a crash loop.\n" "Please restart the application or contact support.", ) except Exception: logger.exception("Failed showing crash-loop dialog") return try: QMessageBox.critical( self, APP_NAME, "The embedded browser process terminated unexpectedly. The page will be reloaded.", ) except Exception: logger.exception("Failed showing render-process-terminated dialog") try: self.browser.reload() except Exception: logger.exception("Failed to reload browser after render process termination") def excepthook(exc_type, exc_value, exc_tb): tb_text = "".join(traceback.format_exception(exc_type, exc_value, exc_tb)) logger.critical("Unhandled exception:\n%s", tb_text) try: QMessageBox.critical(None, APP_NAME, f"An unexpected error occurred:\n\n{exc_value}") except Exception: pass def install_signal_handlers(app: QApplication) -> None: try: 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: logger.debug("SIGTERM handler unavailable", exc_info=True) def configure_qt_paths() -> None: try: QCoreApplication.setApplicationName(APP_NAME) QCoreApplication.setOrganizationName(APP_NAME) QCoreApplication.setApplicationVersion(APP_VERSION) if is_frozen(): helper_name = ( "QtWebEngineProcess.exe" if platform.system().lower() == "windows" else "QtWebEngineProcess" ) candidates = [ resource_dir() / helper_name, resource_dir() / "PySide6" / "Qt" / "libexec" / helper_name, resource_dir() / "Qt" / "libexec" / helper_name, base_dir() / helper_name, 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) logger.info("Using QtWebEngineProcess: %s", candidate) break else: logger.warning("QtWebEngineProcess helper not found in expected locations") else: os.environ.pop("QTWEBENGINEPROCESS_PATH", None) except Exception: logger.debug("Failed to configure Qt paths/environment", exc_info=True) def harden_runtime_paths() -> None: try: harden_directory_permissions( [ (DATA_DIR, 0o700), (LOG_DIR, 0o700), (RUNTIME_DIR, 0o700), (CACHE_DIR, 0o700), (STORAGE_DIR, 0o700), ] ) logger.info("Hardened directory permissions for app data paths") except Exception: logger.exception("Failed to harden runtime paths") def main(): instance = None activation_server = None try: validate_configuration() if is_root(): logger.error("Refusing to run as root") sys.exit(1) sys.excepthook = excepthook configure_qt_paths() harden_runtime_paths() QApplication.setApplicationName(APP_NAME) QApplication.setOrganizationName(APP_NAME) start_local_proxy_if_enabled() initialize_start_url() app = QApplication(sys.argv) install_signal_handlers(app) instance = SingleInstance(LOCK_FILE) if not instance.acquire(): logger.warning("Another instance is already running") activated = signal_existing_instance(IPC_SOCKET_NAME) if activated: logger.info("Signaled existing instance to activate window") else: QMessageBox.warning( None, APP_NAME, "Another instance of this application is already running.", ) sys.exit(1) try: logger.info( "Qt AppDataLocation(s): %s", QStandardPaths.standardLocations( QStandardPaths.StandardLocation.AppDataLocation ), ) except Exception: logger.debug("Failed to query Qt standard paths", exc_info=True) window = MainWindow() activation_server = ActivationServer( IPC_SOCKET_NAME, window.activate_and_raise, ) activation_server.start() try: app.aboutToQuit.connect( lambda: activation_server.close() if activation_server else None ) except Exception: logger.debug("Could not connect activation server cleanup", exc_info=True) try: app.aboutToQuit.connect(lambda: instance.release() if instance else None) 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 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) if __name__ == "__main__": main()