Update desktop/main.py

This commit is contained in:
dev 2026-10-14 23:35:27 +00:00
parent 21f1dc81ba
commit d5e044d728

View File

@ -2,16 +2,22 @@ import logging
import os import os
import platform import platform
import signal import signal
import socket
import stat
import sys import sys
import time
import traceback import traceback
import webbrowser import webbrowser
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
from pathlib import Path from pathlib import Path
APP_NAME = "angelhome" APP_NAME = "angelhome"
APP_VERSION = "2.1.0"
START_URL = "https://angel.33.backend-3.com" START_URL = "https://angel.33.backend-3.com"
ALLOWED_HOSTS = {"angel.33.backend-3.com"} ALLOWED_HOSTS = {"angel.33.backend-3.com"}
ALLOW_SUBDOMAINS = False ALLOW_SUBDOMAINS = False
ALLOWED_PATH_PREFIXES = ["/"] # tighten if desired, e.g. ["/app", "/api/ui"]
OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False OPEN_EXTERNAL_LINKS_IN_SYSTEM_BROWSER = False
# Set False so cookies/local storage/session data can persist # 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_MAX_BYTES = 2 * 1024 * 1024
LOG_BACKUP_COUNT = 3 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: def is_frozen() -> bool:
return bool(getattr(sys, "frozen", False)) return bool(getattr(sys, "frozen", False))
@ -54,17 +81,49 @@ def base_dir() -> Path:
return Path(__file__).resolve().parent 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) path.mkdir(parents=True, exist_ok=True)
if mode is not None:
set_directory_permissions(path, mode)
return path return path
DATA_DIR = ensure_dir(base_dir() / f"{APP_NAME}_data") def set_directory_permissions(path: Path, mode: int) -> None:
LOG_DIR = ensure_dir(DATA_DIR / "logs") """
RUNTIME_DIR = ensure_dir(DATA_DIR / "runtime") Best-effort permission hardening for POSIX.
CACHE_DIR = ensure_dir(DATA_DIR / "cache") On Windows this is a no-op because os.chmod is not a full ACL model.
STORAGE_DIR = ensure_dir(DATA_DIR / "storage") """
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" 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: def configure_runtime_environment() -> None:
@ -80,8 +139,17 @@ def configure_runtime_environment() -> None:
os.environ.setdefault("QT_ENABLE_HIGHDPI_SCALING", "1") os.environ.setdefault("QT_ENABLE_HIGHDPI_SCALING", "1")
if system == "linux": 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("QTWEBENGINE_DISABLE_SANDBOX", "1")
os.environ.setdefault("XDG_RUNTIME_DIR", str(RUNTIME_DIR))
add_flag("--no-sandbox") add_flag("--no-sandbox")
add_flag("--disable-extensions") add_flag("--disable-extensions")
add_flag("--disable-plugins") add_flag("--disable-plugins")
@ -101,26 +169,25 @@ def configure_runtime_environment() -> None:
add_flag("--no-default-browser-check") add_flag("--no-default-browser-check")
add_flag("--no-first-run") add_flag("--no-first-run")
if not ALLOW_WEBRTC:
add_flag("--disable-webrtc")
if flags: if flags:
os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags) os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = " ".join(flags)
configure_runtime_environment() configure_runtime_environment()
from PySide6.QtCore import ( from PySide6.QtCore import QCoreApplication, QLockFile, QStandardPaths, QTimer, QUrl, Slot
QCoreApplication, from PySide6.QtNetwork import QLocalServer, QLocalSocket
QLockFile,
QStandardPaths,
QTimer,
QUrl,
Slot,
)
from PySide6.QtWidgets import QApplication, QMainWindow, QMessageBox from PySide6.QtWidgets import QApplication, QMainWindow, QMessageBox
from PySide6.QtWebEngineCore import ( from PySide6.QtWebEngineCore import (
QWebEngineCertificateError, QWebEngineCertificateError,
QWebEnginePage, QWebEnginePage,
QWebEngineProfile, QWebEngineProfile,
QWebEngineSettings, QWebEngineSettings,
QWebEngineUrlRequestInfo,
QWebEngineUrlRequestInterceptor,
) )
from PySide6.QtWebEngineWidgets import QWebEngineView from PySide6.QtWebEngineWidgets import QWebEngineView
@ -168,6 +235,20 @@ def safe_url_string(url) -> str:
return "<unavailable-url>" 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: def is_allowed_host(host: str) -> bool:
host = (host or "").lower().strip() host = (host or "").lower().strip()
@ -183,6 +264,11 @@ def is_allowed_host(host: str) -> bool:
return False 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: def is_allowed_url(url: QUrl) -> bool:
try: try:
return bool( return bool(
@ -190,11 +276,28 @@ def is_allowed_url(url: QUrl) -> bool:
and url.isValid() and url.isValid()
and url.scheme().lower() == "https" and url.scheme().lower() == "https"
and is_allowed_host(url.host()) and is_allowed_host(url.host())
and is_allowed_path(url.path())
and not url.userName()
and not url.password()
) )
except Exception: except Exception:
return False 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: def validate_configuration() -> None:
if not APP_NAME or not isinstance(APP_NAME, str): if not APP_NAME or not isinstance(APP_NAME, str):
raise ValueError("APP_NAME must be a non-empty string") 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) url_str = safe_url_string(url)
ok = webbrowser.open(url_str) ok = webbrowser.open(url_str)
if ok: 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: 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: except Exception:
logger.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: class SingleInstance:
def __init__(self, lock_path: Path): def __init__(self, lock_path: Path):
self.lock = QLockFile(str(lock_path)) self.lock = QLockFile(str(lock_path))
self.lock.setStaleLockTime(0)
try:
self.lock.setStaleLockTime(0)
except Exception:
pass
def acquire(self) -> bool: def acquire(self) -> bool:
try: try:
@ -243,6 +418,90 @@ class SingleInstance:
logger.debug("Failed to release single-instance lock", exc_info=True) 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): class LockedPage(QWebEnginePage):
def __init__(self, profile: QWebEngineProfile, parent=None): def __init__(self, profile: QWebEngineProfile, parent=None):
super().__init__(profile, parent) super().__init__(profile, parent)
@ -283,7 +542,7 @@ class LockedPage(QWebEnginePage):
logger.warning( logger.warning(
"[BLOCKED NAVIGATION] url=%s is_main_frame=%s nav_type=%s", "[BLOCKED NAVIGATION] url=%s is_main_frame=%s nav_type=%s",
safe_url_string(url), redact_url(url),
is_main_frame, is_main_frame,
nav_type, nav_type,
) )
@ -310,24 +569,66 @@ class LockedPage(QWebEnginePage):
@Slot(QUrl, object) @Slot(QUrl, object)
def _on_feature_permission_requested(self, security_origin, feature): def _on_feature_permission_requested(self, security_origin, feature):
try: 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( logger.warning(
"[DENIED PERMISSION] origin=%s feature=%s", "[FEATURE PERMISSION] origin=%s feature=%s allowed=%s",
safe_url_string(security_origin), redact_url(security_origin),
feature, feature_name,
allowed,
) )
self.setFeaturePermission( self.setFeaturePermission(
security_origin, security_origin,
feature, feature,
QWebEnginePage.PermissionPolicy.PermissionDeniedByUser, QWebEnginePage.PermissionPolicy.PermissionGrantedByUser
if allowed
else QWebEnginePage.PermissionPolicy.PermissionDeniedByUser,
) )
except Exception: 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) @Slot(QWebEngineCertificateError)
def _on_certificate_error(self, error): def _on_certificate_error(self, error):
try: try:
logger.error( 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() error.rejectCertificate()
except Exception: except Exception:
@ -337,11 +638,15 @@ class LockedPage(QWebEnginePage):
class MainWindow(QMainWindow): class MainWindow(QMainWindow):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.render_crash_times = []
self.setWindowTitle(WINDOW_TITLE) self.setWindowTitle(WINDOW_TITLE)
self.resize(WINDOW_WIDTH, WINDOW_HEIGHT) self.resize(WINDOW_WIDTH, WINDOW_HEIGHT)
self.browser = QWebEngineView(self) self.browser = QWebEngineView(self)
self.browser.setContextMenuPolicy(
self.browser.contextMenuPolicy()
)
self.setCentralWidget(self.browser) self.setCentralWidget(self.browser)
self.profile = self._build_profile() self.profile = self._build_profile()
@ -354,8 +659,8 @@ class MainWindow(QMainWindow):
def _build_profile(self) -> QWebEngineProfile: def _build_profile(self) -> QWebEngineProfile:
try: try:
ensure_dir(CACHE_DIR) ensure_dir(CACHE_DIR, 0o700)
ensure_dir(STORAGE_DIR) ensure_dir(STORAGE_DIR, 0o700)
if OFF_THE_RECORD: if OFF_THE_RECORD:
profile = QWebEngineProfile(self.browser) profile = QWebEngineProfile(self.browser)
@ -391,6 +696,14 @@ class MainWindow(QMainWindow):
except Exception: except Exception:
logger.debug("setHttpCacheType unavailable", exc_info=True) 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: try:
profile.downloadRequested.connect(self._deny_download) profile.downloadRequested.connect(self._deny_download)
except Exception: except Exception:
@ -425,6 +738,9 @@ class MainWindow(QMainWindow):
("HyperlinkAuditingEnabled", False), ("HyperlinkAuditingEnabled", False),
("PdfViewerEnabled", False), ("PdfViewerEnabled", False),
("PlaybackRequiresUserGesture", True), ("PlaybackRequiresUserGesture", True),
("JavascriptCanAccessClipboard", ALLOW_CLIPBOARD),
("WebRTCPublicInterfacesOnly", not ALLOW_WEBRTC),
("ScreenCaptureEnabled", ALLOW_SCREEN_CAPTURE),
): ):
try: try:
attr = getattr(QWebEngineSettings.WebAttribute, attr_name) attr = getattr(QWebEngineSettings.WebAttribute, attr_name)
@ -434,7 +750,16 @@ class MainWindow(QMainWindow):
"WebAttribute unavailable: %s", attr_name, exc_info=True "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: except Exception:
logger.exception("Failed to apply WebEngine settings") logger.exception("Failed to apply WebEngine settings")
raise raise
@ -460,7 +785,7 @@ class MainWindow(QMainWindow):
raise ValueError("Refusing to load disallowed START_URL") raise ValueError("Refusing to load disallowed START_URL")
self.browser.setUrl(qurl) self.browser.setUrl(qurl)
logger.info("Loading start URL: %s", START_URL) logger.info("Loading start URL: %s", redact_url(qurl))
except Exception: except Exception:
logger.exception("Failed to load start URL") logger.exception("Failed to load start URL")
QMessageBox.critical( QMessageBox.critical(
@ -470,11 +795,33 @@ class MainWindow(QMainWindow):
def _deny_download(self, item) -> None: def _deny_download(self, item) -> None:
try: try:
logger.warning("[BLOCKED DOWNLOAD] %s", safe_url_string(item.url())) logger.warning("[BLOCKED DOWNLOAD] %s", redact_url(item.url()))
item.cancel() item.cancel()
except Exception: except Exception:
logger.exception("Failed while denying download request") 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() @Slot()
def _on_load_started(self): def _on_load_started(self):
logger.info("Page load started") logger.info("Page load started")
@ -485,7 +832,7 @@ class MainWindow(QMainWindow):
@Slot(bool) @Slot(bool)
def _on_load_finished(self, ok): def _on_load_finished(self, ok):
current_url = safe_url_string(self.browser.url()) current_url = redact_url(self.browser.url())
if ok: if ok:
logger.info("Page load finished successfully: %s", current_url) logger.info("Page load finished successfully: %s", current_url)
else: else:
@ -501,7 +848,7 @@ class MainWindow(QMainWindow):
@Slot(QUrl) @Slot(QUrl)
def _on_url_changed(self, url): 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) @Slot(str)
def _on_title_changed(self, title): def _on_title_changed(self, title):
@ -512,11 +859,30 @@ class MainWindow(QMainWindow):
@Slot(object, int) @Slot(object, int)
def _on_render_process_terminated(self, termination_status, exit_code): def _on_render_process_terminated(self, termination_status, exit_code):
crash_count = self._record_render_crash()
logger.error( logger.error(
"Render process terminated: status=%s exit_code=%s", "Render process terminated: status=%s exit_code=%s recent_crashes=%s",
termination_status, termination_status,
exit_code, 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: try:
QMessageBox.critical( QMessageBox.critical(
self, self,
@ -561,7 +927,7 @@ def configure_qt_paths() -> None:
try: try:
QCoreApplication.setApplicationName(APP_NAME) QCoreApplication.setApplicationName(APP_NAME)
QCoreApplication.setOrganizationName(APP_NAME) QCoreApplication.setOrganizationName(APP_NAME)
QCoreApplication.setApplicationVersion("2.0") QCoreApplication.setApplicationVersion(APP_VERSION)
if is_frozen(): if is_frozen():
helper_name = ( helper_name = (
@ -580,7 +946,7 @@ def configure_qt_paths() -> None:
] ]
for candidate in candidates: for candidate in candidates:
if candidate.exists(): if candidate.exists() and candidate.is_file():
os.environ["QTWEBENGINEPROCESS_PATH"] = str(candidate) os.environ["QTWEBENGINEPROCESS_PATH"] = str(candidate)
logger.info("Using QtWebEngineProcess: %s", candidate) logger.info("Using QtWebEngineProcess: %s", candidate)
break break
@ -594,8 +960,27 @@ def configure_qt_paths() -> None:
logger.debug("Failed to configure Qt paths/environment", exc_info=True) 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(): def main():
instance = None instance = None
activation_server = None
try: try:
validate_configuration() validate_configuration()
@ -607,6 +992,7 @@ def main():
sys.excepthook = excepthook sys.excepthook = excepthook
configure_qt_paths() configure_qt_paths()
harden_runtime_paths()
QApplication.setApplicationName(APP_NAME) QApplication.setApplicationName(APP_NAME)
QApplication.setOrganizationName(APP_NAME) QApplication.setOrganizationName(APP_NAME)
@ -617,11 +1003,15 @@ def main():
instance = SingleInstance(LOCK_FILE) instance = SingleInstance(LOCK_FILE)
if not instance.acquire(): if not instance.acquire():
logger.warning("Another instance is already running") logger.warning("Another instance is already running")
QMessageBox.warning( activated = signal_existing_instance(IPC_SOCKET_NAME)
None, if activated:
APP_NAME, logger.info("Signaled existing instance to activate window")
"Another instance of this application is already running.", else:
) QMessageBox.warning(
None,
APP_NAME,
"Another instance of this application is already running.",
)
sys.exit(1) sys.exit(1)
try: try:
@ -635,20 +1025,46 @@ def main():
logger.debug("Failed to query Qt standard paths", exc_info=True) logger.debug("Failed to query Qt standard paths", exc_info=True)
window = MainWindow() 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() window.show()
logger.info( 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(), platform.platform(),
sys.version.split()[0], sys.version.split()[0],
is_frozen(), is_frozen(),
resource_dir(), resource_dir(),
base_dir(), base_dir(),
DATA_DIR, DATA_DIR,
os.environ.get("XDG_RUNTIME_DIR"),
os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", ""),
APP_VERSION,
) )
rc = app.exec() rc = app.exec()
if activation_server:
activation_server.close()
if instance: if instance:
instance.release() instance.release()
@ -665,6 +1081,12 @@ def main():
except Exception: except Exception:
print(f"Fatal startup error: {exc}", file=sys.stderr) print(f"Fatal startup error: {exc}", file=sys.stderr)
try:
if activation_server:
activation_server.close()
except Exception:
pass
try: try:
if instance: if instance:
instance.release() instance.release()
@ -675,4 +1097,4 @@ def main():
if __name__ == "__main__": if __name__ == "__main__":
main() main()