Update desktop/main.py
This commit is contained in:
parent
d5e044d728
commit
0cbd50f97c
278
desktop/main.py
278
desktop/main.py
@ -2,7 +2,6 @@ import logging
|
||||
import os
|
||||
import platform
|
||||
import signal
|
||||
import socket
|
||||
import stat
|
||||
import sys
|
||||
import time
|
||||
@ -12,15 +11,14 @@ from logging.handlers import RotatingFileHandler
|
||||
from pathlib import Path
|
||||
|
||||
APP_NAME = "angelhome"
|
||||
APP_VERSION = "2.1.0"
|
||||
APP_VERSION = "2.1.1"
|
||||
|
||||
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"]
|
||||
ALLOWED_PATH_PREFIXES = ["/"]
|
||||
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
|
||||
@ -31,7 +29,6 @@ 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
|
||||
@ -40,29 +37,63 @@ 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
|
||||
|
||||
# ---------- Logging / runtime behavior ----------
|
||||
# In production/frozen mode, default to no console logging.
|
||||
ENABLE_CONSOLE_LOGGING_WHEN_FROZEN = False
|
||||
|
||||
# If True, redact path too, leaving scheme://host[:port] only.
|
||||
REDACT_URL_PATH = False
|
||||
|
||||
# When frozen on Windows, silence stdout/stderr by default.
|
||||
DETACH_STDIO_WHEN_FROZEN = True
|
||||
|
||||
|
||||
def is_frozen() -> bool:
|
||||
return bool(getattr(sys, "frozen", False))
|
||||
|
||||
|
||||
def detach_stdio_if_needed() -> None:
|
||||
"""
|
||||
Best-effort suppression of terminal output in frozen production builds.
|
||||
Packaging still matters:
|
||||
- PyInstaller: use --noconsole / windowed
|
||||
- cx_Freeze: use Win32GUI base
|
||||
"""
|
||||
try:
|
||||
if not is_frozen():
|
||||
return
|
||||
if not DETACH_STDIO_WHEN_FROZEN:
|
||||
return
|
||||
|
||||
devnull_r = open(os.devnull, "r", encoding="utf-8", errors="ignore")
|
||||
devnull_w = open(os.devnull, "a", encoding="utf-8", errors="ignore")
|
||||
|
||||
try:
|
||||
sys.stdin = devnull_r
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
sys.stdout = devnull_w
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
sys.stderr = devnull_w
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
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():
|
||||
@ -71,16 +102,19 @@ def resource_dir() -> Path:
|
||||
|
||||
|
||||
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 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:
|
||||
@ -88,19 +122,6 @@ def ensure_dir(path: Path, mode: int | None = None) -> Path:
|
||||
return path
|
||||
|
||||
|
||||
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:
|
||||
@ -118,14 +139,6 @@ 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:
|
||||
system = platform.system().lower()
|
||||
existing_flags = os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", "").strip()
|
||||
@ -176,6 +189,7 @@ def configure_runtime_environment() -> None:
|
||||
os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags)
|
||||
|
||||
|
||||
detach_stdio_if_needed()
|
||||
configure_runtime_environment()
|
||||
|
||||
from PySide6.QtCore import QCoreApplication, QLockFile, QStandardPaths, QTimer, QUrl, Slot
|
||||
@ -192,6 +206,40 @@ from PySide6.QtWebEngineCore import (
|
||||
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 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:
|
||||
logger = logging.getLogger(APP_NAME)
|
||||
logger.setLevel(LOG_LEVEL)
|
||||
@ -201,9 +249,10 @@ def configure_logging() -> logging.Logger:
|
||||
|
||||
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s")
|
||||
|
||||
sh = logging.StreamHandler(sys.stdout)
|
||||
sh.setFormatter(fmt)
|
||||
logger.addHandler(sh)
|
||||
if not (is_frozen() and not ENABLE_CONSOLE_LOGGING_WHEN_FROZEN):
|
||||
sh = logging.StreamHandler(sys.stdout)
|
||||
sh.setFormatter(fmt)
|
||||
logger.addHandler(sh)
|
||||
|
||||
fh = RotatingFileHandler(
|
||||
str(LOG_DIR / f"{APP_NAME}.log"),
|
||||
@ -228,27 +277,6 @@ def is_root() -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def safe_url_string(url) -> str:
|
||||
try:
|
||||
return url.toString() if isinstance(url, QUrl) else str(url)
|
||||
except Exception:
|
||||
return "<unavailable-url>"
|
||||
|
||||
|
||||
def redact_url(url: QUrl | str) -> str:
|
||||
try:
|
||||
qurl = url if isinstance(url, QUrl) else QUrl(str(url))
|
||||
if not qurl.isValid():
|
||||
return "<invalid-url>"
|
||||
redacted = QUrl(qurl)
|
||||
redacted.setQuery("")
|
||||
redacted.setFragment("")
|
||||
redacted.setPassword("")
|
||||
return redacted.toString()
|
||||
except Exception:
|
||||
return "<unavailable-url>"
|
||||
|
||||
|
||||
def is_allowed_host(host: str) -> bool:
|
||||
host = (host or "").lower().strip()
|
||||
|
||||
@ -284,20 +312,6 @@ def is_allowed_url(url: QUrl) -> bool:
|
||||
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")
|
||||
@ -320,13 +334,9 @@ def open_in_system_browser(url: QUrl) -> None:
|
||||
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)
|
||||
)
|
||||
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)
|
||||
)
|
||||
logger.exception("Failed to open URL in system browser: %s", redact_url(url))
|
||||
|
||||
|
||||
class RequestInterceptor(QWebEngineUrlRequestInterceptor):
|
||||
@ -379,14 +389,6 @@ class RequestInterceptor(QWebEngineUrlRequestInterceptor):
|
||||
)
|
||||
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:
|
||||
@ -398,7 +400,6 @@ class RequestInterceptor(QWebEngineUrlRequestInterceptor):
|
||||
class SingleInstance:
|
||||
def __init__(self, lock_path: Path):
|
||||
self.lock = QLockFile(str(lock_path))
|
||||
|
||||
try:
|
||||
self.lock.setStaleLockTime(0)
|
||||
except Exception:
|
||||
@ -512,9 +513,7 @@ class LockedPage(QWebEnginePage):
|
||||
logger.debug("certificateError signal unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
self.featurePermissionRequested.connect(
|
||||
self._on_feature_permission_requested
|
||||
)
|
||||
self.featurePermissionRequested.connect(self._on_feature_permission_requested)
|
||||
except Exception:
|
||||
logger.debug("featurePermissionRequested signal unavailable", exc_info=True)
|
||||
|
||||
@ -547,10 +546,7 @@ class LockedPage(QWebEnginePage):
|
||||
nav_type,
|
||||
)
|
||||
|
||||
if OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER and url.scheme().lower() in {
|
||||
"http",
|
||||
"https",
|
||||
}:
|
||||
if OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER and url.scheme().lower() in {"http", "https"}:
|
||||
open_in_system_browser(url)
|
||||
|
||||
return False
|
||||
@ -569,9 +565,6 @@ 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:
|
||||
@ -601,8 +594,8 @@ class LockedPage(QWebEnginePage):
|
||||
|
||||
logger.warning(
|
||||
"[FEATURE PERMISSION] origin=%s feature=%s allowed=%s",
|
||||
redact_url(security_origin),
|
||||
feature_name,
|
||||
redact_url(security_origin, hide_path=True),
|
||||
str(feature),
|
||||
allowed,
|
||||
)
|
||||
|
||||
@ -628,7 +621,9 @@ class LockedPage(QWebEnginePage):
|
||||
def _on_certificate_error(self, error):
|
||||
try:
|
||||
logger.error(
|
||||
"[TLS ERROR] %s @ %s", error.description(), redact_url(error.url())
|
||||
"[TLS ERROR] %s @ %s",
|
||||
error.description(),
|
||||
redact_url(error.url()),
|
||||
)
|
||||
error.rejectCertificate()
|
||||
except Exception:
|
||||
@ -644,9 +639,6 @@ class MainWindow(QMainWindow):
|
||||
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()
|
||||
@ -671,8 +663,8 @@ class MainWindow(QMainWindow):
|
||||
profile.setPersistentStoragePath(str(STORAGE_DIR))
|
||||
logger.info(
|
||||
"Using persistent profile cache=%s storage=%s",
|
||||
CACHE_DIR,
|
||||
STORAGE_DIR,
|
||||
str(CACHE_DIR),
|
||||
str(STORAGE_DIR),
|
||||
)
|
||||
|
||||
try:
|
||||
@ -746,20 +738,9 @@ class MainWindow(QMainWindow):
|
||||
attr = getattr(QWebEngineSettings.WebAttribute, attr_name)
|
||||
settings.setAttribute(attr, value)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"WebAttribute unavailable: %s", attr_name, exc_info=True
|
||||
)
|
||||
logger.debug("WebAttribute unavailable: %s", attr_name, exc_info=True)
|
||||
|
||||
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,
|
||||
)
|
||||
logger.info("Applied WebEngine security settings")
|
||||
except Exception:
|
||||
logger.exception("Failed to apply WebEngine settings")
|
||||
raise
|
||||
@ -788,9 +769,7 @@ class MainWindow(QMainWindow):
|
||||
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."
|
||||
)
|
||||
QMessageBox.critical(self, APP_NAME, "Failed to load the application start URL.")
|
||||
raise
|
||||
|
||||
def _deny_download(self, item) -> None:
|
||||
@ -806,7 +785,6 @@ class MainWindow(QMainWindow):
|
||||
self.showNormal()
|
||||
else:
|
||||
self.show()
|
||||
|
||||
self.raise_()
|
||||
self.activateWindow()
|
||||
logger.info("Activated existing main window")
|
||||
@ -817,8 +795,7 @@ class MainWindow(QMainWindow):
|
||||
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
|
||||
t for t in self.render_crash_times if now - t <= RENDER_CRASH_WINDOW_SECONDS
|
||||
]
|
||||
return len(self.render_crash_times)
|
||||
|
||||
@ -873,11 +850,9 @@ class MainWindow(QMainWindow):
|
||||
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."
|
||||
),
|
||||
"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")
|
||||
@ -895,18 +870,14 @@ class MainWindow(QMainWindow):
|
||||
try:
|
||||
self.browser.reload()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Failed to reload browser after render process termination"
|
||||
)
|
||||
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}"
|
||||
)
|
||||
QMessageBox.critical(None, APP_NAME, f"An unexpected error occurred:\n\n{exc_value}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@ -951,9 +922,7 @@ def configure_qt_paths() -> None:
|
||||
logger.info("Using QtWebEngineProcess: %s", candidate)
|
||||
break
|
||||
else:
|
||||
logger.warning(
|
||||
"QtWebEngineProcess helper not found in expected locations"
|
||||
)
|
||||
logger.warning("QtWebEngineProcess helper not found in expected locations")
|
||||
else:
|
||||
os.environ.pop("QTWEBENGINEPROCESS_PATH", None)
|
||||
except Exception:
|
||||
@ -971,9 +940,7 @@ def harden_runtime_paths() -> None:
|
||||
(STORAGE_DIR, 0o700),
|
||||
]
|
||||
)
|
||||
logger.info(
|
||||
"Hardened directory permissions for data/log/runtime/cache/storage paths"
|
||||
)
|
||||
logger.info("Hardened directory permissions for app data paths")
|
||||
except Exception:
|
||||
logger.exception("Failed to harden runtime paths")
|
||||
|
||||
@ -987,7 +954,6 @@ def main():
|
||||
|
||||
if is_root():
|
||||
logger.error("Refusing to run as root")
|
||||
print("Refusing to run as root.")
|
||||
sys.exit(1)
|
||||
|
||||
sys.excepthook = excepthook
|
||||
@ -1026,9 +992,7 @@ def main():
|
||||
|
||||
window = MainWindow()
|
||||
|
||||
activation_server = ActivationServer(
|
||||
IPC_SOCKET_NAME, window.activate_and_raise
|
||||
)
|
||||
activation_server = ActivationServer(IPC_SOCKET_NAME, window.activate_and_raise)
|
||||
activation_server.start()
|
||||
|
||||
try:
|
||||
@ -1039,25 +1003,21 @@ def main():
|
||||
logger.debug("Could not connect activation server cleanup", exc_info=True)
|
||||
|
||||
try:
|
||||
app.aboutToQuit.connect(
|
||||
lambda: instance.release() if instance else None
|
||||
)
|
||||
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 xdg_runtime_dir=%s chromium_flags=%s version=%s",
|
||||
"Application started platform=%s python=%s frozen=%s version=%s resource_dir=%s base_dir=%s data_dir=%s",
|
||||
platform.platform(),
|
||||
sys.version.split()[0],
|
||||
is_frozen(),
|
||||
APP_VERSION,
|
||||
resource_dir(),
|
||||
base_dir(),
|
||||
DATA_DIR,
|
||||
os.environ.get("XDG_RUNTIME_DIR"),
|
||||
os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", ""),
|
||||
APP_VERSION,
|
||||
)
|
||||
|
||||
rc = app.exec()
|
||||
@ -1076,10 +1036,8 @@ def main():
|
||||
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)
|
||||
pass
|
||||
|
||||
try:
|
||||
if activation_server:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user