diff --git a/desktop/main.py b/desktop/main.py index 6d30955..70617e4 100644 --- a/desktop/main.py +++ b/desktop/main.py @@ -2,16 +2,22 @@ import logging import os import platform import signal +import socket +import stat import sys +import time import traceback import webbrowser from logging.handlers import RotatingFileHandler from pathlib import Path APP_NAME = "angelhome" +APP_VERSION = "2.1.0" + START_URL = "https://angel.33.backend-3.com" ALLOWED_HOSTS = {"angel.33.backend-3.com"} ALLOW_SUBDOMAINS = False +ALLOWED_PATH_PREFIXES = ["/"] # tighten if desired, e.g. ["/app", "/api/ui"] OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False # Set False so cookies/local storage/session data can persist @@ -25,6 +31,27 @@ LOG_LEVEL = logging.INFO LOG_MAX_BYTES = 2 * 1024 * 1024 LOG_BACKUP_COUNT = 3 +# ---------- Capability toggles ---------- +ALLOW_CLIPBOARD = False +ALLOW_GEOLOCATION = False +ALLOW_NOTIFICATIONS = False +ALLOW_MEDIA_AUDIO_CAPTURE = False +ALLOW_MEDIA_VIDEO_CAPTURE = False +ALLOW_SCREEN_CAPTURE = False +ALLOW_WEBRTC = False + +# ---------- Single instance / IPC ---------- +ENABLE_SINGLE_INSTANCE_IPC = True +IPC_CONNECT_TIMEOUT_MS = 1500 +ACTIVATE_WINDOW_MESSAGE = "ACTIVATE" + +# ---------- Crash loop protection ---------- +MAX_RENDER_CRASHES = 3 +RENDER_CRASH_WINDOW_SECONDS = 60 + +# ---------- Resource request filtering ---------- +BLOCK_UNKNOWN_SCHEMES = True + def is_frozen() -> bool: return bool(getattr(sys, "frozen", False)) @@ -54,17 +81,49 @@ def base_dir() -> Path: return Path(__file__).resolve().parent -def ensure_dir(path: Path) -> Path: +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 -DATA_DIR = ensure_dir(base_dir() / f"{APP_NAME}_data") -LOG_DIR = ensure_dir(DATA_DIR / "logs") -RUNTIME_DIR = ensure_dir(DATA_DIR / "runtime") -CACHE_DIR = ensure_dir(DATA_DIR / "cache") -STORAGE_DIR = ensure_dir(DATA_DIR / "storage") +def set_directory_permissions(path: Path, mode: int) -> None: + """ + Best-effort permission hardening for POSIX. + On Windows this is a no-op because os.chmod is not a full ACL model. + """ + try: + if os.name == "posix": + os.chmod(path, mode) + except Exception: + # logger may not exist yet during early startup + pass + + +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" +IPC_SOCKET_NAME = f"{APP_NAME}_single_instance_socket" + + +def safe_unlink(path: Path) -> None: + try: + if path.exists(): + path.unlink() + except Exception: + pass def configure_runtime_environment() -> None: @@ -80,8 +139,17 @@ def configure_runtime_environment() -> None: 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) os.environ.setdefault("QTWEBENGINE_DISABLE_SANDBOX", "1") - os.environ.setdefault("XDG_RUNTIME_DIR", str(RUNTIME_DIR)) + add_flag("--no-sandbox") add_flag("--disable-extensions") add_flag("--disable-plugins") @@ -101,26 +169,25 @@ def configure_runtime_environment() -> None: 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) configure_runtime_environment() -from PySide6.QtCore import ( - QCoreApplication, - QLockFile, - QStandardPaths, - QTimer, - QUrl, - Slot, -) +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 @@ -168,6 +235,20 @@ def safe_url_string(url) -> str: return "" +def redact_url(url: QUrl | str) -> str: + try: + qurl = url if isinstance(url, QUrl) else QUrl(str(url)) + if not qurl.isValid(): + return "" + redacted = QUrl(qurl) + redacted.setQuery("") + redacted.setFragment("") + redacted.setPassword("") + return redacted.toString() + except Exception: + return "" + + def is_allowed_host(host: str) -> bool: host = (host or "").lower().strip() @@ -183,6 +264,11 @@ def is_allowed_host(host: str) -> bool: return False +def is_allowed_path(path: str) -> bool: + path = path or "/" + return any(path.startswith(prefix) for prefix in ALLOWED_PATH_PREFIXES) + + def is_allowed_url(url: QUrl) -> bool: try: return bool( @@ -190,11 +276,28 @@ def is_allowed_url(url: QUrl) -> bool: and url.isValid() and url.scheme().lower() == "https" and is_allowed_host(url.host()) + and is_allowed_path(url.path()) + and not url.userName() + and not url.password() ) except Exception: return False +def is_allowed_subresource_url(url: QUrl) -> bool: + try: + scheme = url.scheme().lower() + if scheme not in {"https", "wss"}: + return False + if not is_allowed_host(url.host()): + return False + if scheme == "https" and not is_allowed_path(url.path()): + return False + return True + except Exception: + return False + + def validate_configuration() -> None: if not APP_NAME or not isinstance(APP_NAME, str): raise ValueError("APP_NAME must be a non-empty string") @@ -215,19 +318,91 @@ def open_in_system_browser(url: QUrl) -> None: url_str = safe_url_string(url) ok = webbrowser.open(url_str) if ok: - logger.info("Opened external link in system browser: %s", url_str) + logger.info("Opened external link in system browser: %s", redact_url(url)) else: - logger.warning("System browser did not confirm opening URL: %s", url_str) + 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", safe_url_string(url) + "Failed to open URL in system browser: %s", redact_url(url) ) +class RequestInterceptor(QWebEngineUrlRequestInterceptor): + def interceptRequest(self, info: QWebEngineUrlRequestInfo) -> None: + try: + url = info.requestUrl() + scheme = (url.scheme() or "").lower() + host = (url.host() or "").lower() + method = bytes(info.requestMethod()).decode("utf-8", "ignore") + + 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 not is_allowed_host(host): + logger.warning( + "[BLOCKED REQUEST] reason=host scheme=%s host=%s method=%s url=%s resource_type=%s", + scheme, + host, + method, + 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 + else: + logger.warning( + "[BLOCKED REQUEST] reason=unexpected-scheme scheme=%s url=%s", + scheme, + redact_url(url), + ) + 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)) - self.lock.setStaleLockTime(0) + + try: + self.lock.setStaleLockTime(0) + except Exception: + pass def acquire(self) -> bool: try: @@ -243,6 +418,90 @@ class SingleInstance: 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) @@ -283,7 +542,7 @@ class LockedPage(QWebEnginePage): logger.warning( "[BLOCKED NAVIGATION] url=%s is_main_frame=%s nav_type=%s", - safe_url_string(url), + redact_url(url), is_main_frame, nav_type, ) @@ -310,24 +569,66 @@ class LockedPage(QWebEnginePage): @Slot(QUrl, object) def _on_feature_permission_requested(self, security_origin, feature): try: + policy = QWebEnginePage.PermissionPolicy.PermissionDeniedByUser + + feature_name = str(feature) + 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( - "[DENIED PERMISSION] origin=%s feature=%s", - safe_url_string(security_origin), - feature, + "[FEATURE PERMISSION] origin=%s feature=%s allowed=%s", + redact_url(security_origin), + feature_name, + allowed, ) + self.setFeaturePermission( security_origin, feature, - QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, + QWebEnginePage.PermissionPolicy.PermissionGrantedByUser + if allowed + else QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, ) except Exception: - logger.exception("Failed to deny feature permission request") + 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(), safe_url_string(error.url()) + "[TLS ERROR] %s @ %s", error.description(), redact_url(error.url()) ) error.rejectCertificate() except Exception: @@ -337,11 +638,15 @@ class LockedPage(QWebEnginePage): 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.browser.setContextMenuPolicy( + self.browser.contextMenuPolicy() + ) self.setCentralWidget(self.browser) self.profile = self._build_profile() @@ -354,8 +659,8 @@ class MainWindow(QMainWindow): def _build_profile(self) -> QWebEngineProfile: try: - ensure_dir(CACHE_DIR) - ensure_dir(STORAGE_DIR) + ensure_dir(CACHE_DIR, 0o700) + ensure_dir(STORAGE_DIR, 0o700) if OFF_THE_RECORD: profile = QWebEngineProfile(self.browser) @@ -391,6 +696,14 @@ class MainWindow(QMainWindow): 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: @@ -425,6 +738,9 @@ class MainWindow(QMainWindow): ("HyperlinkAuditingEnabled", False), ("PdfViewerEnabled", False), ("PlaybackRequiresUserGesture", True), + ("JavascriptCanAccessClipboard", ALLOW_CLIPBOARD), + ("WebRTCPublicInterfacesOnly", not ALLOW_WEBRTC), + ("ScreenCaptureEnabled", ALLOW_SCREEN_CAPTURE), ): try: attr = getattr(QWebEngineSettings.WebAttribute, attr_name) @@ -434,7 +750,16 @@ class MainWindow(QMainWindow): "WebAttribute unavailable: %s", attr_name, exc_info=True ) - logger.info("Applied WebEngine security settings") + logger.info( + "Applied WebEngine security settings clipboard=%s geolocation=%s notifications=%s audio_capture=%s video_capture=%s screen_capture=%s webrtc=%s", + ALLOW_CLIPBOARD, + ALLOW_GEOLOCATION, + ALLOW_NOTIFICATIONS, + ALLOW_MEDIA_AUDIO_CAPTURE, + ALLOW_MEDIA_VIDEO_CAPTURE, + ALLOW_SCREEN_CAPTURE, + ALLOW_WEBRTC, + ) except Exception: logger.exception("Failed to apply WebEngine settings") raise @@ -460,7 +785,7 @@ class MainWindow(QMainWindow): raise ValueError("Refusing to load disallowed START_URL") self.browser.setUrl(qurl) - logger.info("Loading start URL: %s", START_URL) + logger.info("Loading start URL: %s", redact_url(qurl)) except Exception: logger.exception("Failed to load start URL") QMessageBox.critical( @@ -470,11 +795,33 @@ class MainWindow(QMainWindow): def _deny_download(self, item) -> None: try: - logger.warning("[BLOCKED DOWNLOAD] %s", safe_url_string(item.url())) + 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: + now = time.time() + self.render_crash_times.append(now) + self.render_crash_times = [ + t for t in self.render_crash_times + if now - t <= RENDER_CRASH_WINDOW_SECONDS + ] + return len(self.render_crash_times) + @Slot() def _on_load_started(self): logger.info("Page load started") @@ -485,7 +832,7 @@ class MainWindow(QMainWindow): @Slot(bool) def _on_load_finished(self, ok): - current_url = safe_url_string(self.browser.url()) + current_url = redact_url(self.browser.url()) if ok: logger.info("Page load finished successfully: %s", current_url) else: @@ -501,7 +848,7 @@ class MainWindow(QMainWindow): @Slot(QUrl) def _on_url_changed(self, url): - logger.info("URL changed: %s", safe_url_string(url)) + logger.info("URL changed: %s", redact_url(url)) @Slot(str) def _on_title_changed(self, title): @@ -512,11 +859,30 @@ 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", + "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, @@ -561,7 +927,7 @@ def configure_qt_paths() -> None: try: QCoreApplication.setApplicationName(APP_NAME) QCoreApplication.setOrganizationName(APP_NAME) - QCoreApplication.setApplicationVersion("2.0") + QCoreApplication.setApplicationVersion(APP_VERSION) if is_frozen(): helper_name = ( @@ -580,7 +946,7 @@ def configure_qt_paths() -> None: ] for candidate in candidates: - if candidate.exists(): + if candidate.exists() and candidate.is_file(): os.environ["QTWEBENGINEPROCESS_PATH"] = str(candidate) logger.info("Using QtWebEngineProcess: %s", candidate) break @@ -594,8 +960,27 @@ def configure_qt_paths() -> None: 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 data/log/runtime/cache/storage paths" + ) + except Exception: + logger.exception("Failed to harden runtime paths") + + def main(): instance = None + activation_server = None try: validate_configuration() @@ -607,6 +992,7 @@ def main(): sys.excepthook = excepthook configure_qt_paths() + harden_runtime_paths() QApplication.setApplicationName(APP_NAME) QApplication.setOrganizationName(APP_NAME) @@ -617,11 +1003,15 @@ def main(): instance = SingleInstance(LOCK_FILE) if not instance.acquire(): logger.warning("Another instance is already running") - QMessageBox.warning( - None, - APP_NAME, - "Another instance of this application 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: @@ -635,20 +1025,46 @@ def main(): 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) + window.show() logger.info( - "Application started platform=%s python=%s frozen=%s resource_dir=%s base_dir=%s data_dir=%s", + "Application started platform=%s python=%s frozen=%s resource_dir=%s base_dir=%s data_dir=%s xdg_runtime_dir=%s chromium_flags=%s version=%s", platform.platform(), sys.version.split()[0], is_frozen(), resource_dir(), base_dir(), DATA_DIR, + os.environ.get("XDG_RUNTIME_DIR"), + os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", ""), + APP_VERSION, ) rc = app.exec() + if activation_server: + activation_server.close() + if instance: instance.release() @@ -665,6 +1081,12 @@ def main(): except Exception: print(f"Fatal startup error: {exc}", file=sys.stderr) + try: + if activation_server: + activation_server.close() + except Exception: + pass + try: if instance: instance.release() @@ -675,4 +1097,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file