glo/desktop/main.py

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 "<unavailable-url>"


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 "<invalid-url>"
        scheme = (qurl.scheme() or "").lower()
        host = qurl.host() or ""
        port = qurl.port()
        path = qurl.path() or "/"
        if not scheme or not host:
            return "<redacted-url>"
        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 "<unavailable-url>"


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()