diff --git a/desktop/main.py b/desktop/main.py new file mode 100644 index 0000000..23e9b7d --- /dev/null +++ b/desktop/main.py @@ -0,0 +1,632 @@ +import logging +import os +import platform +import signal +import sys +import traceback +import webbrowser +from logging.handlers import RotatingFileHandler +from pathlib import Path + +APP_NAME = "angelhome" +START_URL = "https://angel.33.backend-3.com" +ALLOWED_HOSTS = {"angel.33.backend-3.com"} +ALLOW_SUBDOMAINS = False +OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False + +# Set False so cookies/local storage/session data can persist +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 + + +def is_frozen() -> bool: + return bool(getattr(sys, "frozen", False)) + + +def resource_dir() -> Path: + """ + Directory where bundled resources live. + In PyInstaller one-file mode, this may point to _MEIPASS. + In one-folder mode, it is usually the executable folder. + """ + 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: + """ + Writable base directory for persistent app data. + Always use executable folder when frozen. + Never write into _MEIPASS. + """ + if is_frozen(): + return Path(sys.executable).resolve().parent + return Path(__file__).resolve().parent + + +def ensure_dir(path: Path) -> Path: + path.mkdir(parents=True, exist_ok=True) + 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") +LOCK_FILE = RUNTIME_DIR / f"{APP_NAME}.lock" + + +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) + + os.environ.setdefault("QT_AUTO_SCREEN_SCALE_FACTOR", "1") + os.environ.setdefault("QT_ENABLE_HIGHDPI_SCALING", "1") + + if system == "linux": + 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") + 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") + 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 flags: + os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags) + + +configure_runtime_environment() + +from PySide6.QtCore import QCoreApplication, QLockFile, QStandardPaths, QTimer, QUrl, Slot +from PySide6.QtWidgets import QApplication, QMainWindow, QMessageBox +from PySide6.QtWebEngineCore import ( + QWebEngineCertificateError, + QWebEnginePage, + QWebEngineProfile, + QWebEngineSettings, +) +from PySide6.QtWebEngineWidgets import QWebEngineView + + +def configure_logging() -> logging.Logger: + logger = logging.getLogger(APP_NAME) + logger.setLevel(LOG_LEVEL) + + if logger.handlers: + return logger + + fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s") + + sh = logging.StreamHandler(sys.stdout) + sh.setFormatter(fmt) + 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) + logger.addHandler(fh) + + logger.propagate = False + return logger + + +logger = configure_logging() + + +def is_root() -> bool: + try: + return hasattr(os, "geteuid") and os.geteuid() == 0 + except Exception: + return False + + +def safe_url_string(url) -> str: + try: + return url.toString() if isinstance(url, QUrl) else str(url) + except Exception: + return "" + + +def is_allowed_host(host: str) -> bool: + host = (host or "").lower().strip() + + if host in ALLOWED_HOSTS: + return True + + if ALLOW_SUBDOMAINS: + for allowed in ALLOWED_HOSTS: + allowed = (allowed or "").lower().strip() + if host == allowed or host.endswith("." + allowed): + return True + + return False + + +def is_allowed_url(url: QUrl) -> bool: + try: + return bool( + url + and url.isValid() + and url.scheme().lower() == "https" + and is_allowed_host(url.host()) + ) + 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") + + if not ALLOWED_HOSTS or not isinstance(ALLOWED_HOSTS, (set, list, tuple)): + raise ValueError("ALLOWED_HOSTS must be a non-empty set/list/tuple") + + qurl = QUrl(START_URL) + if not qurl.isValid(): + raise ValueError(f"Invalid START_URL: {START_URL}") + + if not is_allowed_url(qurl): + raise ValueError("START_URL is not allowed by current security policy") + + +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", url_str) + else: + logger.warning("System browser did not confirm opening URL: %s", url_str) + except Exception: + logger.exception("Failed to open URL in system browser: %s", safe_url_string(url)) + + +class SingleInstance: + def __init__(self, lock_path: Path): + self.lock = QLockFile(str(lock_path)) + self.lock.setStaleLockTime(0) + + 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 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): + 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_url(url): + return super().acceptNavigationRequest(url, nav_type, is_main_frame) + + logger.warning( + "[BLOCKED NAVIGATION] url=%s is_main_frame=%s nav_type=%s", + safe_url_string(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: + logger.warning( + "[DENIED PERMISSION] origin=%s feature=%s", + safe_url_string(security_origin), + feature, + ) + self.setFeaturePermission( + security_origin, + feature, + QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, + ) + except Exception: + logger.exception("Failed to deny feature permission request") + + @Slot(QWebEngineCertificateError) + def _on_certificate_error(self, error): + try: + logger.error("[TLS ERROR] %s @ %s", error.description(), safe_url_string(error.url())) + error.rejectCertificate() + except Exception: + logger.exception("Failed to reject certificate") + + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + + 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) + ensure_dir(STORAGE_DIR) + + 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", + CACHE_DIR, + 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: + 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), + ): + 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(START_URL) + if not is_allowed_url(qurl): + raise ValueError("Refusing to load disallowed START_URL") + + self.browser.setUrl(qurl) + logger.info("Loading start URL: %s", START_URL) + 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", safe_url_string(item.url())) + item.cancel() + except Exception: + logger.exception("Failed while denying download request") + + @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 = safe_url_string(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", safe_url_string(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): + logger.error( + "Render process terminated: status=%s exit_code=%s", + termination_status, + exit_code, + ) + 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("2.0") + + 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(): + 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 main(): + instance = None + + try: + validate_configuration() + + if is_root(): + logger.error("Refusing to run as root") + print("Refusing to run as root.") + sys.exit(1) + + sys.excepthook = excepthook + configure_qt_paths() + + QApplication.setApplicationName(APP_NAME) + QApplication.setOrganizationName(APP_NAME) + + app = QApplication(sys.argv) + install_signal_handlers(app) + + 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.") + 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() + window.show() + + logger.info( + "Application started platform=%s python=%s frozen=%s resource_dir=%s base_dir=%s data_dir=%s", + platform.platform(), + sys.version.split()[0], + is_frozen(), + resource_dir(), + base_dir(), + DATA_DIR, + ) + + rc = app.exec() + + if instance: + instance.release() + + 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}") + else: + print(f"Fatal startup error: {exc}", file=sys.stderr) + except Exception: + print(f"Fatal startup error: {exc}", file=sys.stderr) + + try: + if instance: + instance.release() + except Exception: + pass + + sys.exit(1) + + +if __name__ == "__main__": + main()