633 lines
20 KiB
Python
633 lines
20 KiB
Python
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 "<unavailable-url>"
|
|
|
|
|
|
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()
|