web_app.py

#!/usr/bin/env python3
"""Local browser UI for OpenBox. Independent open-source software not affiliated with LaunchBox or Unbroken Software, LLC."""

import email.utils
from datetime import datetime, timezone
import json
import mimetypes
import os
import queue as queue_module
import secrets
import signal
import socket
import socketserver
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

from api_errors import ApiError, BadRequest, RangeParseError, RouteNotFound
from env_config import bootstrap_env
from openbox_logging import configure_logging
from openbox import DATA, STATE_STORE, load_state, purge_demo_games, update_state
from plugins import emit_plugin_event
from pkg.parity.parity_backup import AUTO_BACKUP_ITEMS, AUTO_BACKUP_KEEP, auto_backup_due, create_backup, restore_backup
from pkg.parity.parity_deeplinks import handle_cli
from pkg.parity.parity_emulator_defs import merge_profiles_from_definitions
from pkg.parity.parity_gameyfin import GameyfinError
from pkg.parity.parity_gamescope import OPENBOX_STEAM_GAME_ID, is_gamescope_guest, mark_process_windows, open_ui
from pkg.platform_compat import IS_LINUX, ensure_macos_tool_path, process_alive, terminate_process_tree
from routes import dispatch_get, dispatch_post
from routes.registry import route
from state_store import (
    StateCorruptError,
    StateFutureVersionError,
    StateMigrationError,
    StateSchemaTooOldError,
    secure_text_write,
)
from api_errors import problem_payload
from stock_themes import ensure_stock_themes
from webapp_state import (
    JOB_MANAGER,
    LOGGER,
    PROCESS_LOCK,
    ROOT,
    RUNNING,
    STARTUP_ERRORS,
    TOKEN,
    WATCH_STOP,
    approved_backup_file,
    auto_import_worker,
    control_game_session,
    reattach_session,
    reconcile_sessions_on_startup,
    run_configured_commands,
    shutdown_webhooks,
    SSE_MAX_EVENT_BYTES,
    SSE_QUEUE_SIZE,
    SSE_WRITE_TIMEOUT,
    register_event_subscriber,
    unregister_event_subscriber,
)
from pkg.state.imports import start_media_queue_worker
from pkg.state.machines import SESSION_ACTIVE, SESSION_STOPPING
from handlers.data import DataHandlers
from handlers.discovery import DiscoveryHandlers
from handlers.emulators import EmulatorsHandlers
from handlers.export import ExportHandlers
from handlers.extensions import ExtensionsHandlers
from handlers.faugus import FaugusHandlers
from handlers.health import HealthHandlers
from handlers.imports import ImportsHandlers
from handlers.insights import InsightsHandlers
from handlers.jobs import JobsHandlers
from handlers.launch import LaunchHandlers
from handlers.picker import PickerHandlers
from handlers.constellation import ConstellationHandlers
from handlers.party import PartyHandlers
from handlers.setup import SetupHandlers
from handlers.library import LibraryHandlers
from handlers.media import MediaHandlers
from handlers.metadata import MetadataHandlers
from handlers.sessions import SessionHandlers
from handlers.screenscraper import ScreenScraperHandlers
from handlers.settings import SettingsHandlers
from handlers.timemachine import TimeMachineHandlers
from handlers.wine import WineHandlers

import re as _re

def _sanitize_error_message(error):
    """Return a client-safe error string without absolute filesystem paths.

    Full details are logged server-side; the client receives the prefix
    before the path to avoid leaking home-directory layout.
    """
    raw = str(error)
    if not raw:
        return raw
    # Absolute paths may be POSIX or Windows-shaped; strip either before the
    # message can reach a client.
    path_pattern = r"(?:~)?/[^\s\"']*|[A-Za-z]:\\[^\s\"']*|\\\\[^\s\"']+"
    if not _re.search(path_pattern, raw):
        return raw
    match = _re.match(r"^([^:]*?):\s+(.*)$", raw)
    if match and _re.search(path_pattern, match.group(2)):
        prefix = match.group(1).strip()
        # Preserve a trailing period for consistency
        if not prefix.endswith("."):
            prefix += "."
        return prefix
    # Fallback: replace any absolute path token with [path]
    sanitized = _re.sub(path_pattern, "[path]", raw)
    sanitized = _re.sub(r"\[path\](?:\s*\[path\])+", "[path]", sanitized)
    # Collapse " : [path]" to "."
    sanitized = _re.sub(r"\s*:\s*\[path\].*", ".", sanitized)
    sanitized = sanitized.strip()
    if sanitized:
        return sanitized
    return raw


# Host header values the API accepts. The server binds loopback only; rejecting
# any other Host closes DNS-rebinding, where a remote page resolves to
# 127.0.0.1 and carries the attacker's hostname in the Host header.
LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"})

# Rate limiting for authentication failures (per client IP, loopback only).
# 10 failures in 60s triggers 429 with Retry-After; success resets the window.
_AUTH_FAILURES: dict[str, list[float]] = {}
_AUTH_FAILURES_LOCK = threading.Lock()
_AUTH_MAX_FAILURES = 10
_AUTH_WINDOW_SECONDS = 60.0

# SSE subscriber-cap rejection: when register_event_subscriber() reports the
# cap is full, answer 503 immediately instead of opening a heartbeat-only
# stream that would never deliver events. Retry-After nudges the client to
# retry once another tab closes its stream.
SSE_BUSY_RETRY_AFTER = 5

# Security headers shared by every response (including the SSE stream) so the
# policy can't drift between code paths.
CSP_DEFAULT = "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline' https://fonts.bunny.net; font-src 'self' https://fonts.bunny.net; script-src 'self'; object-src 'none'; base-uri 'none'; frame-ancestors 'none'"
# Document responses are framed by the in-app reader iframe, so they relax
# only the framing directives; every other policy stays identical.
CSP_FRAMEABLE = CSP_DEFAULT.replace("frame-ancestors 'none'", "frame-ancestors 'self'")



class OpenBoxServer(ThreadingHTTPServer):
    """The loopback server.

    The default accept backlog of five connections is too small for a page
    that fires several requests at once: macOS resets the connection when
    the queue overflows instead of queueing it.
    """

    request_queue_size = 128

    def server_bind(self):
        # HTTPServer.server_bind() reverse-resolves the bind address with
        # socket.getfqdn(). On macOS that goes through mDNSResponder and can
        # stall startup past the native host's boot timeout -- the server then
        # never publishes its port. Nothing here reads server_name, so bind
        # directly and keep the loopback address.
        socketserver.TCPServer.server_bind(self)
        self.server_name = "127.0.0.1"
        self.server_port = self.server_address[1]


class Handler(LibraryHandlers, ImportsHandlers, MediaHandlers, MetadataHandlers, SessionHandlers, SettingsHandlers, TimeMachineHandlers, ExtensionsHandlers, HealthHandlers, JobsHandlers, EmulatorsHandlers, ExportHandlers, LaunchHandlers, SetupHandlers, DataHandlers, DiscoveryHandlers, WineHandlers, FaugusHandlers, InsightsHandlers, ScreenScraperHandlers, PickerHandlers, ConstellationHandlers, PartyHandlers, BaseHTTPRequestHandler):
    server_version = "OpenBox/1"
    protocol_version = "HTTP/1.1"
    MAX_BODY = 65536
    REQUEST_TIMEOUT = 30

    def setup(self):
        super().setup()
        self.connection.settimeout(self.REQUEST_TIMEOUT)
        try:
            self.connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            self.connection.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024 * 1024)
        except (OSError, AttributeError):
            pass

    def log_message(self, *_):
        pass

    def send_response(self, code, message=None):
        LOGGER.debug("HTTP %s %s -> %s", getattr(self, "command", "?"), urlparse(getattr(self, "path", "")).path, code)
        super().send_response(code, message)
    def headers_common(self, content_type, cache_control="no-store", frameable=False):
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", cache_control)
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Content-Security-Policy", CSP_FRAMEABLE if frameable else CSP_DEFAULT)
        self.send_header("X-Frame-Options", "SAMEORIGIN" if frameable else "DENY")


    def send_bytes(self, status, data, content_type, cache_control="no-store", etag=None, last_modified=None, extra_headers=None):
        if etag and self.headers.get("If-None-Match", "").strip() == etag:
            self.send_response(304)
            self.headers_common(content_type, cache_control=cache_control)
            self.send_header("ETag", etag)
            self.end_headers()
            return
        self.send_response(status)
        self.headers_common(content_type, cache_control=cache_control)
        if etag:
            self.send_header("ETag", etag)
        if last_modified:
            self.send_header("Last-Modified", last_modified)
        for name, value in {**getattr(self, "route_headers", {}), **(extra_headers or {})}.items():
            self.send_header(name, value)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)
        self.wfile.flush()

    def _cache_headers(self, path, stat_result):
        etag = f'"{stat_result.st_mtime_ns:x}-{stat_result.st_size:x}"'
        last_modified = email.utils.formatdate(stat_result.st_mtime, usegmt=True)
        return etag, last_modified

    def send_file(self, status, path, content_type=None, extra_headers=None, frameable=False, cache_control=None):
        path = Path(path)
        stat_result = path.stat()
        size = stat_result.st_size
        etag, last_modified = self._cache_headers(path, stat_result)
        request_cache_control = cache_control or "private, max-age=31536000, immutable"
        headers = getattr(self, "headers", None) or {}
        conditional = headers.get("If-None-Match", "")
        if etag in {item.strip() for item in conditional.split(",")}:
            self.send_response(304)
            self.headers_common(content_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream", cache_control=request_cache_control, frameable=frameable)
            self.send_header("ETag", etag)
            self.send_header("Last-Modified", last_modified)
            for name, value in (extra_headers or {}).items():
                self.send_header(name, value)
            self.end_headers()
            return
        if not conditional:
            if_modified_since = headers.get("If-Modified-Since", "")
            if if_modified_since:
                try:
                    since = email.utils.parsedate_to_datetime(if_modified_since)
                    if stat_result.st_mtime < since.timestamp() + 1:
                        self.send_response(304)
                        self.headers_common(content_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream", cache_control=request_cache_control, frameable=frameable)
                        self.send_header("ETag", etag)
                        self.send_header("Last-Modified", last_modified)
                        for name, value in (extra_headers or {}).items():
                            self.send_header(name, value)
                        self.end_headers()
                        return
                except (TypeError, ValueError):
                    pass
        start, end = 0, size - 1
        response_status = status
        range_header = headers.get("Range", "")
        if range_header.startswith("bytes="):
            spec = range_header[6:].split(",", 1)[0].strip()
            try:
                if "-" not in spec:
                    raise RangeParseError("Invalid byte range.")
                left, right = spec.split("-", 1)
                if left:
                    start = int(left)
                    end = int(right) if right else end
                elif right:
                    length = int(right)
                    start = max(0, size - length)
            except ValueError as error:
                raise RangeParseError("Invalid byte range.") from error
            if start < 0 or start >= size or end < start:
                self.send_response(416)
                self.headers_common(content_type or "application/octet-stream", cache_control=request_cache_control, frameable=frameable)
                self.send_header("Content-Range", f"bytes */{size}")
                self.send_header("Content-Length", "0")
                for name, value in (extra_headers or {}).items():
                    self.send_header(name, value)
                self.end_headers()
                return
            end = min(end, size - 1)
            response_status = 206
        length = max(0, end - start + 1)
        self.send_response(response_status)
        self.headers_common(content_type or mimetypes.guess_type(path.name)[0] or "application/octet-stream", cache_control=request_cache_control, frameable=frameable)
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("ETag", etag)
        self.send_header("Last-Modified", last_modified)
        self.send_header("Content-Length", str(length))
        if response_status == 206:
            self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
        for name, value in (extra_headers or {}).items():
            self.send_header(name, value)
        try:
            self.end_headers()
            self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError):
            return
        # Use os.sendfile for zero-copy transfer on full-file requests (Linux).
        # Note: settimeout() puts the raw fd in non-blocking mode, so sendfile
        # can hit EAGAIN (BlockingIOError) once the kernel send buffer fills.
        # Its return value is the exact byte count per call, and a call that
        # raises transfers nothing, so the buffered path can safely resume at
        # start + sendfile_sent.
        remaining = length
        if response_status == 200 and hasattr(os, "sendfile") and hasattr(self, "connection"):
            sent = 0
            try:
                file_fd = os.open(str(path), os.O_RDONLY)
                try:
                    offset = start
                    while sent < length:
                        transferred = os.sendfile(self.connection.fileno(), file_fd, offset, min(1024 * 1024, length - sent))
                        if transferred == 0:
                            break
                        offset += transferred
                        sent += transferred
                finally:
                    os.close(file_fd)
            except (BrokenPipeError, ConnectionResetError):
                return
            except (OSError, AttributeError):
                pass
            start += sent
            remaining -= sent
        if remaining:
            try:
                with path.open("rb") as source:
                    source.seek(start)
                    while remaining:
                        chunk = source.read(min(1024 * 1024, remaining))
                        if not chunk:
                            break
                        self.wfile.write(chunk)
                        remaining -= len(chunk)
                self.wfile.flush()
            except (BrokenPipeError, ConnectionResetError, OSError):
                pass

    def send_json(self, status, payload, extra_headers=None):
        if status >= 400 and isinstance(payload, dict):
            # Errors are RFC 9457-shaped problem documents; `error` stays for one major cycle.
            body = problem_payload(status, payload, request_id=str(payload.get("request_id") or ""), instance=self.path)
            self.send_bytes(status, json.dumps(body).encode(), "application/problem+json; charset=utf-8", extra_headers=extra_headers)
            return
        data = json.dumps(payload).encode()
        self.send_bytes(status, data, "application/json; charset=utf-8", extra_headers=extra_headers)

    def _auth_client_ip(self):
        try:
            return self.client_address[0] if getattr(self, "client_address", None) else "127.0.0.1"
        except Exception:
            return "127.0.0.1"

    def _is_auth_rate_limited(self):
        now = time.monotonic()
        ip = self._auth_client_ip()
        with _AUTH_FAILURES_LOCK:
            attempts = _AUTH_FAILURES.get(ip, [])
            attempts = [t for t in attempts if now - t < _AUTH_WINDOW_SECONDS]
            _AUTH_FAILURES[ip] = attempts
            if len(attempts) >= _AUTH_MAX_FAILURES:
                oldest = min(attempts) if attempts else now
                return int(_AUTH_WINDOW_SECONDS - (now - oldest)) + 1
            return 0

    def _record_auth_failure(self):
        now = time.monotonic()
        ip = self._auth_client_ip()
        with _AUTH_FAILURES_LOCK:
            attempts = _AUTH_FAILURES.get(ip, [])
            attempts = [t for t in attempts if now - t < _AUTH_WINDOW_SECONDS]
            attempts.append(now)
            _AUTH_FAILURES[ip] = attempts
            if len(attempts) >= _AUTH_MAX_FAILURES:
                oldest = min(attempts)
                return int(_AUTH_WINDOW_SECONDS - (now - oldest)) + 1
            return 0

    def _clear_auth_failures(self):
        ip = self._auth_client_ip()
        with _AUTH_FAILURES_LOCK:
            _AUTH_FAILURES.pop(ip, None)

    def authorized(self):
        query_token = parse_qs(urlparse(self.path).query).get("token", [""])[0]
        provided = self.headers.get("X-OpenBox-Token", "") or query_token
        # Compare bytes: compare_digest raises TypeError on non-ASCII str,
        # which escaped as a 400 and skipped the auth-failure rate limiter.
        ok = secrets.compare_digest(provided.encode("utf-8", "replace"), TOKEN.encode("utf-8"))
        if ok:
            self._clear_auth_failures()
            return True
        # Wrong token: still check rate limit so handle_unauthorized can
        # decide between 403 and 429, but do not block correct tokens.
        return False

    def handle_unauthorized(self):
        retry = self._record_auth_failure()
        if retry:
            self.send_json(
                429,
                {"error": "Too many authentication failures. Try again later.", "code": "RATE_LIMITED", "retry_after": retry},
                extra_headers={"Retry-After": str(retry)},
            )
        else:
            # Re-check if this failure pushed us over the limit
            retry = self._is_auth_rate_limited()
            if retry:
                self.send_json(
                    429,
                    {"error": "Too many authentication failures. Try again later.", "code": "RATE_LIMITED", "retry_after": retry},
                    extra_headers={"Retry-After": str(retry)},
                )
            else:
                self.send_json(403, {"error": "Unauthorized"})

    def body(self):
        raw_length = self.headers.get("Content-Length", "0")
        try:
            length = int(raw_length)
        except (TypeError, ValueError) as error:
            raise ValueError("Content-Length must be a valid number.") from error
        if length < 0:
            raise ValueError("Content-Length must not be negative.")
        if length > self.MAX_BODY:
            raise ValueError("Request is too large.")
        raw = self.rfile.read(length)
        if len(raw) != length:
            raise ValueError("Request body was truncated.")
        return json.loads(raw or b"{}")

    def _do_GET(self):
        parsed = urlparse(self.path)
        try:
            dispatch_get(self, parsed)
        except ApiError:
            raise
        except (OSError, ValueError, TypeError, AttributeError, KeyError, IndexError, json.JSONDecodeError, GameyfinError, FileNotFoundError, RuntimeError, subprocess.SubprocessError) as error:
            LOGGER.warning("Request %s failed: %s", parsed.path, error)
            raise BadRequest(_sanitize_error_message(error)) from None

    @route("GET", ["/", "/index.html"], public=True)
    def _api_get_index(self, parsed):
        if parsed.path in ("/", "/index.html"):
            html = (ROOT / "index.html").read_bytes().decode("utf-8")
            # The launcher opens the page with ?token= in the URL; app.js reads
            # it from location.search. Static assets are public paths and need
            # no token, so the shell itself carries no secrets.
            self.send_bytes(200, html.encode("utf-8"), "text/html; charset=utf-8")
            return
    @route("GET", [
        "/static/app.js",
        "/static/util.js",
        "/static/state.js",
        "/static/library.js",
        "/static/settings.js",
        "/static/imports.js",
        "/static/metadata.js",
        "/static/media.js",
        "/static/reader.js",
        "/static/sessions.js",
        "/static/recap.js",
        "/static/bigbox.js",
        "/static/storefront.js",
        "/static/dialogs.js",
        "/static/setup.js",
        "/static/activity.js",
        "/static/insights.js",
        "/static/mood.js",
        "/static/picker.js",
        "/static/constellation.js",
        "/static/wrapped.js",
        "/static/timeline.js",
        "/static/mastery.js",
        "/static/party.js",
        "/static/timemachine.js",
        "/static/defs.js",
        "/static/moments.js",
        "/static/palette.js",
        "/static/arcaderoom.js",
        "/static/household.js",
        "/static/clips.js",
        "/static/whatsnew.js",
        "/static/worker.search.js",
        "/static/i18n.js",
        "/static/details.js",
        "/static/emulators.js",
        "/static/app.css",
        "/static/logo.png",
    ], public=True)
    def _api_get_static(self, parsed):
        # Static UI assets (app.js/app.css plus the ES-module chunks) live
        # next to index.html. JS/CSS change between releases, so serve them
        # no-cache: the mtime+size ETag still answers 304 for unchanged files,
        # but clients revalidate instead of running stale modules for a year.
        name = Path(parsed.path).name
        if not (name.endswith((".js", ".svg")) or name in {"app.css", "logo.png"}):
            raise RouteNotFound("Not found")
        if name == "logo.png":
            asset = ROOT / "assets" / "openbox-logo.png"
            if asset.is_file():
                self.send_file(200, asset, "image/png")
                return
            raise RouteNotFound("Not found")
        asset = ROOT / "static" / name
        if not asset.is_file():
            raise RouteNotFound("Not found")
        if name.endswith(".js"):
            content_type = "text/javascript; charset=utf-8"
        elif name.endswith(".svg"):
            content_type = "image/svg+xml"
        else:
            content_type = "text/css; charset=utf-8"
        self.send_file(200, asset, content_type, cache_control="no-cache")
        return

    @route("GET", "/assets/placeholder-cover.svg", public=True)
    def _api_get_placeholder_cover(self, parsed):
        asset = ROOT / "assets" / "placeholder-cover.svg"
        if not asset.is_file():
            raise RouteNotFound("Not found")
        self.send_file(200, asset, "image/svg+xml", cache_control="public, max-age=31536000, immutable")
        return

    @route("GET", ["/locales/en.json", "/locales/es.json", "/locales/de.json", "/locales/fr.json", "/locales/pt.json"], public=True)
    def _api_get_locale(self, parsed):
        """Serve locale JSON files for the i18n system (1.7.2)."""
        name = Path(parsed.path).name
        asset = ROOT / "locales" / name
        if not asset.is_file():
            raise RouteNotFound("Not found")
        self.send_file(200, asset, "application/json; charset=utf-8")
        return

    @route("GET", ["/favicon.ico", "/favicon.svg"], public=True)
    def _api_get_favicon(self, parsed):
        if parsed.path in ("/favicon.svg", "/favicon.ico"):
            # Browsers request an icon on every initial load; serve the
            # repo icon instead of a 404 console error.
            icon = ROOT / "openbox.svg"
            if icon.is_file():
                self.send_bytes(200, icon.read_bytes(), "image/svg+xml")
                return
        # Without this the handler wrote no response and the client hung
        # until REQUEST_TIMEOUT.
        raise RouteNotFound("Not found")

    @route("GET", "/api/events")
    def _api_get_api_events(self, parsed):
        subscriber_queue = queue_module.Queue(maxsize=SSE_QUEUE_SIZE)
        if not register_event_subscriber(subscriber_queue):
            # Subscriber cap reached: fail fast with 503 instead of opening
            # a stream that would only ever send heartbeats and no events.
            self.send_json(
                503,
                {"error": "Server is busy. Close an unused tab and try again.", "code": "SSE_BUSY", "retry_after": SSE_BUSY_RETRY_AFTER},
                extra_headers={"Retry-After": str(SSE_BUSY_RETRY_AFTER)},
            )
            return
        try:
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Connection", "close")
            self.send_header("X-Accel-Buffering", "no")
            self.send_header("X-Content-Type-Options", "nosniff")
            self.send_header("Referrer-Policy", "no-referrer")
            self.send_header("Content-Security-Policy", CSP_DEFAULT)
            self.send_header("X-Frame-Options", "DENY")
            self.end_headers()
            self.connection.settimeout(SSE_WRITE_TIMEOUT)
            while True:
                try:
                    item = subscriber_queue.get(timeout=15)
                except queue_module.Empty:
                    # Keep the connection alive with a heartbeat comment.
                    self.wfile.write(b": ping\n\n")
                    self.wfile.flush()
                    continue
                if item is None:
                    break
                kind, data = item
                event = f"event: {kind}\ndata: {data}\n\n".encode()
                if len(event) > SSE_MAX_EVENT_BYTES:
                    continue
                self.wfile.write(event)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass
        finally:
            unregister_event_subscriber(subscriber_queue)


    def _do_POST(self):
        try:
            payload = self.body()
            if not isinstance(payload, dict):
                raise ValueError("Request body must be a JSON object.")
            route = urlparse(self.path).path
            dispatch_post(self, route, payload)
        except ApiError:
            raise
        except (OSError, ValueError, TypeError, AttributeError, KeyError, IndexError, json.JSONDecodeError, GameyfinError, FileNotFoundError, RuntimeError, subprocess.SubprocessError) as error:
            LOGGER.warning("Request %s failed: %s", urlparse(self.path).path, error)
            raise BadRequest(_sanitize_error_message(error)) from None

    def _loopback_host(self):
        headers = getattr(self, "headers", None)
        host = headers.get("Host", "") if headers is not None else ""
        if not host:
            LOGGER.warning("Rejecting request with missing or empty Host header")
            return False
        # Normalize Host: strip port, handle IPv6 brackets, lower-case.
        # "LOCALHOST:8080" -> "localhost", "[::1]:8080" -> "::1", "::1" -> "::1"
        host = host.strip()
        # Extract host part without port; IPv6 in brackets
        if host.startswith("["):
            end = host.find("]")
            if end != -1:
                hostname = host[1:end]
            else:
                hostname = host.strip("[]")
        else:
            # Split on last colon only if the part after colon is numeric port
            # and there is only one colon (avoid breaking bare IPv6 ::1 without brackets)
            if host.count(":") == 1:
                hostname = host.rsplit(":", 1)[0]
            elif "::" in host and host.count(":") > 1:
                # Bare IPv6 like ::1 without brackets and without port (rare)
                hostname = host
            else:
                # Normal host or host:port with multiple colons (already handled bracket case)
                # Fallback: if last segment after colon is digits and host looks like host:port
                try:
                    last = host.rsplit(":", 1)[1]
                    if last.isdigit():
                        hostname = host.rsplit(":", 1)[0]
                    else:
                        hostname = host
                except IndexError:
                    hostname = host
                hostname = hostname.strip("[]")
        hostname = hostname.lower().strip()
        return hostname in LOOPBACK_HOSTS

    def _handle_request(self, method):
        path = urlparse(self.path).path
        request_id = secrets.token_hex(4)
        LOGGER.debug("HTTP %s %s started [%s]", method, path, request_id)
        if not self._loopback_host():
            self.send_json(403, {"error": "Request host is not allowed.", "code": "BAD_HOST", "request_id": request_id})
            return
        try:
            getattr(self, f"_{method}")()
        except ApiError as error:
            LOGGER.info("HTTP %s %s [%s] %s: %s", method, path, request_id, error.code, error.message)
            self.send_json(error.status, error.to_payload(request_id))
        except StateFutureVersionError as error:
            LOGGER.error("OpenBox library format is newer than this build: %s", error)
            self.send_json(409, {"error": str(error), "code": "STATE_FUTURE_VERSION", "request_id": request_id})
        except StateSchemaTooOldError as error:
            LOGGER.error("OpenBox library format is older than this build: %s", error)
            self.send_json(409, {"error": str(error), "code": "STATE_SCHEMA_TOO_OLD", "request_id": request_id})
        except StateCorruptError as error:
            LOGGER.error("OpenBox state is unavailable: %s", error)
            self.send_json(503, {"error": "OpenBox library data needs recovery before this operation can continue.", "code": "STATE_UNAVAILABLE", "request_id": request_id})
        except Exception:
            LOGGER.exception("Unhandled HTTP %s %s [%s]", method, path, request_id)
            self.send_json(500, {"error": "Unexpected server error. Copy the diagnostic log from Settings and include it in your report.", "code": "INTERNAL_ERROR", "request_id": request_id})

    def do_GET(self):
        self._handle_request("do_GET")

    def do_POST(self):
        self._handle_request("do_POST")


def _auto_backup_tick():
    """Create the weekly automatic backup when due (1.12.0). Best-effort."""
    try:
        settings = load_state().get("settings", {})
        if not auto_backup_due(settings):
            return
        try:
            keep = max(0, int(settings.get("backup_auto_keep", AUTO_BACKUP_KEEP)))
        except (TypeError, ValueError):
            keep = AUTO_BACKUP_KEEP
        archive = create_backup(DATA.parent, load_state(), AUTO_BACKUP_ITEMS, keep=keep, running_map=RUNNING)
        stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")

        def _record(current):
            current.setdefault("settings", {})["last_auto_backup"] = stamp

        update_state(_record)
        LOGGER.info("Automatic backup saved to %s", archive.name)
    except Exception:
        LOGGER.exception("Automatic backup failed")


def _auto_backup_worker():
    """Hourly tick; WATCH_STOP wakes it for shutdown."""
    while not WATCH_STOP.wait(3600):
        _auto_backup_tick()


def bigbox_launch_url(url):
    """Append the Big Box deeplink to the opened URL when --bigbox is passed.

    Manual sys.argv scan, mirroring the --game-mode handling in main().
    """
    if "--bigbox" in sys.argv:
        return f"{url}&deeplink=bigbox"
    return url


def _queue_scheduled_scans():
    """Queue the library-health rescan and the Launch Audit that rides on its schedule.

    The Launch Audit uses the same health_rescan setting, so the grid's readiness
    badges are refreshed by the same daily or weekly run, with no setting of its own.
    """
    from handlers.launch import AUDIT_JOB_NAME, run_launch_audit
    from handlers.library_health import SCAN_JOB_NAME, run_health_scan
    JOB_MANAGER.submit(SCAN_JOB_NAME, run_health_scan, replace=True)
    # Never replace a check the user started: a deep audit must not be cancelled by the schedule.
    audit_state = (JOB_MANAGER.snapshots().get(AUDIT_JOB_NAME) or {}).get("state")
    if audit_state not in ("queued", "running"):
        JOB_MANAGER.submit(AUDIT_JOB_NAME, lambda ctx: run_launch_audit(ctx, deep=False), replace=True)


def _health_rescan_tick():
    """Queue the scheduled library-health rescan (and Launch Audit) when due (Flagship 8, H6). Best-effort."""
    try:
        from pkg.parity.parity_library_health import rescan_due
        if RUNNING:
            return  # never contend with an active game session
        settings = load_state().get("settings", {})
        if not rescan_due(settings):
            return
        _queue_scheduled_scans()
        LOGGER.info("Scheduled library health rescan and launch audit queued")
    except Exception:
        LOGGER.exception("Scheduled health rescan failed")


def _health_rescan_worker():
    """Hourly tick; WATCH_STOP wakes it for shutdown."""
    while not WATCH_STOP.wait(3600):
        _health_rescan_tick()


def reap_sessions_for_shutdown():
    """Mark persisted sessions stopping, wait 2s, force-kill survivors, remove rows."""
    try:
        _state = load_state()
        _active = [s for s in _state.get("active_sessions", []) if s.get("status") == SESSION_ACTIVE]
        if not _active:
            return

        def _mark_stopping(state):
            for sess in state.get("active_sessions", []):
                if sess.get("status") == SESSION_ACTIVE:
                    sess["status"] = SESSION_STOPPING

        try:
            update_state(_mark_stopping)
        except Exception:
            LOGGER.exception("Failed to mark sessions stopping")
        # Give processes the shutdown window (2s) to exit gracefully, then
        # force-kill the ones still alive.
        import time as _time

        deadline = _time.monotonic() + 2.0
        while _time.monotonic() < deadline:
            if not any(sess.get("pid") and process_alive(sess["pid"]) for sess in _active):
                break
            _time.sleep(0.2)
        for sess in _active:
            pid = sess.get("pid")
            if not pid or not process_alive(pid):
                continue
            terminate_process_tree(pid, force=True, pgid=sess.get("pgid"))

        def _remove_stopping(state):
            state["active_sessions"] = [
                s for s in state.get("active_sessions", []) if s.get("status") != SESSION_STOPPING
            ]

        try:
            update_state(_remove_stopping)
        except Exception:
            LOGGER.exception("Failed to remove stopping sessions on shutdown")
    except Exception:
        LOGGER.exception("Graceful shutdown session cleanup failed")


def main():
    if sys.version_info < (3, 11):
        running = ".".join(str(part) for part in sys.version_info[:3])
        print(
            f"OpenBox requires Python 3.11 or newer. This interpreter is Python {running}.",
            file=sys.stderr,
        )
        raise SystemExit(1)
    from pkg.state import library_cli

    if library_cli.is_library_command(sys.argv[1:]):
        raise SystemExit(library_cli.main(sys.argv[1:]))
    bootstrap_env(DATA.parent)
    configure_logging(DATA.parent)
    LOGGER.info("OpenBox web UI starting")
    if os.environ.get("OPENBOX_ENABLE_V1_API"):
        LOGGER.warning(
            "OPENBOX_ENABLE_V1_API is no longer supported: the /api/v1 surface was removed in 2.1. Use /api/v2."
        )
    # An app started from Finder inherits a minimal PATH, so tools the user
    # installed with Homebrew would be invisible to the launcher and to the
    # emulators it probes.
    os.environ["PATH"] = ensure_macos_tool_path(os.environ.get("PATH", ""))
    args = sys.argv[1:]
    if "--game-mode" in args and not IS_LINUX:
        # Game Mode is a gamescope feature and gamescope is Linux only.
        print("Game Mode needs gamescope and is only available on Linux.", file=sys.stderr)
        raise SystemExit(2)
    if "--backup" in args:
        items = []
        keep = 0
        if "--items" in args:
            items = args[args.index("--items") + 1].split(",")
        if "--keep" in args:
            keep = int(args[args.index("--keep") + 1])
        state = load_state()
        archive = create_backup(DATA.parent, state, items or ["library", "settings"], keep=keep, running_map=RUNNING)
        print(archive)
        return
    if "--restore-backup" in args:
        archive = approved_backup_file(args[args.index("--restore-backup") + 1])
        restored = restore_backup(archive, DATA.parent, running_map=RUNNING)["items"]
        print(",".join(restored))
        return
    cli_code = handle_cli(args, DATA.parent)
    if cli_code is not None:
        raise SystemExit(cli_code)
    ensure_stock_themes(DATA.parent / "themes", ROOT)
    for relative in ("plugins/data", "plugins/credentials", "plugin-stats", "emulator-configs", "cache/steam"):
        runtime_dir = DATA.parent / relative
        runtime_dir.mkdir(parents=True, exist_ok=True)
        try:
            runtime_dir.chmod(0o700)
        except OSError:
            pass
    def bootstrap_state(state):
        purge_demo_games(state)
        profiles = state.setdefault("profiles", {})
        profiles.update(merge_profiles_from_definitions(profiles))
    try:
        update_state(bootstrap_state)
    except (StateFutureVersionError, StateSchemaTooOldError, StateMigrationError) as error:
        # Never recover automatically from these: the primary is a newer format or a
        # failed upgrade, and the copy in library.json.bak may be older.
        STARTUP_ERRORS.append(str(error))
    except StateCorruptError as error:
        STARTUP_ERRORS.append(f"Library data could not be read: {error}")
        try:
            STATE_STORE.recover()
            update_state(bootstrap_state)
            STARTUP_ERRORS.append("Library data was restored from the last-known-good backup.")
        except (OSError, StateCorruptError) as recovery_error:
            STARTUP_ERRORS.append(f"Automatic recovery failed: {recovery_error}")
    # The native windows paint the page background before it loads: record the saved theme's.
    try:
        from handlers.extensions import write_native_background
        write_native_background(load_state().get("settings", {}).get("theme", ""))
    except Exception:  # best effort: without the file the hosts keep their built-in colour
        LOGGER.warning("Could not record the native window background", exc_info=True)
    # Reconcile persisted active_sessions from previous run (Days 0-14, Task 3).
    # Must run once before watchers/jobs start, inside a transaction so .bak/snapshots stay consistent.
    try:
        _reconcile_state = load_state()
        reattached, abandoned = reconcile_sessions_on_startup(_reconcile_state)
        if reattached or abandoned:
            def _persist_reconciled(state):
                state["active_sessions"] = reattached + abandoned
            update_state(_persist_reconciled)
            if reattached:
                LOGGER.info("Reattached %d active session(s) after restart", len(reattached))
            if abandoned:
                LOGGER.info("Marked %d session(s) as abandoned (PID mismatch or gone)", len(abandoned))
            # Re-register verified sessions so the running API and controls work
            # after a server restart, then let the normal watcher finish them.
            for _sess in reattached:
                try:
                    if not reattach_session(_sess, _reconcile_state):
                        LOGGER.warning("Could not reattach persisted session %s", _sess.get("launch_id", ""))
                except Exception:
                    LOGGER.exception("Failed to reattach persisted session %s", _sess.get("launch_id", ""))
    except Exception:
        LOGGER.exception("Session reconciliation failed; continuing with empty active_sessions")
    WATCH_STOP.clear()
    JOB_MANAGER.submit("auto-import", auto_import_worker)
    start_media_queue_worker()
    threading.Thread(target=_auto_backup_worker, name="auto-backup", daemon=True).start()
    threading.Thread(target=_health_rescan_worker, name="health-rescan", daemon=True).start()
    # "On startup" rescan: run once at boot unless a game session is active.
    try:
        from pkg.parity.parity_library_health import normalize_rescan_setting
        if normalize_rescan_setting(load_state().get("settings", {}).get("health_rescan")) == "on_startup" and not RUNNING:
            _queue_scheduled_scans()
    except Exception:
        LOGGER.exception("On-startup health rescan failed")
    server = OpenBoxServer(("127.0.0.1", 0), Handler)
    try:
        run_configured_commands("startup_commands")
    except (StateFutureVersionError, StateSchemaTooOldError, StateMigrationError, StateCorruptError) as error:
        LOGGER.error("Startup commands skipped because the library cannot be read: %s", error)
        if not STARTUP_ERRORS:
            STARTUP_ERRORS.append(str(error))
    # Lifecycle event (F1e): the app finished starting. Best-effort.
    emit_plugin_event(DATA.parent / "plugins", "app_startup", {})
    port = server.server_address[1]
    secure_text_write(DATA.parent / "server.port", str(port))
    secure_text_write(DATA.parent / "server.token", TOKEN)
    try:
        _settings = load_state().get("settings", {})
    except (StateFutureVersionError, StateSchemaTooOldError, StateMigrationError, StateCorruptError) as error:
        LOGGER.error("Settings were not loaded because the library cannot be read: %s", error)
        _settings = {}
        if not STARTUP_ERRORS:
            STARTUP_ERRORS.append(str(error))
    secure_text_write(
        DATA.parent / "native-host-flags",
        f"{int(bool(_settings.get('tray_enabled')))} {int(bool(_settings.get('minimize_to_tray')))}\n",
    )
    try:
        from pkg.parity.parity_tray_recent import write_recent_games
        write_recent_games(DATA.parent, load_state())
    except (OSError, ValueError, StateFutureVersionError, StateSchemaTooOldError, StateMigrationError, StateCorruptError):
        LOGGER.exception("Could not publish recent games for the tray")
    url = f"http://127.0.0.1:{port}/?token={TOKEN}"
    # Log without the token; the token-bearing URL is only printed to the
    # native host via the token file and to a browser when explicitly opened.
    safe_url = f"http://127.0.0.1:{port}/"
    print(f"http://127.0.0.1:{port}/", flush=True)
    LOGGER.info("OpenBox web UI URL: %s", safe_url)
    force_game_mode = "--game-mode" in sys.argv
    url = bigbox_launch_url(url)
    guest = is_gamescope_guest(force=force_game_mode)
    # Desktop sessions open the UI in a chrome-less app window by default;
    # flags override.
    if "--app-window" in sys.argv:
        native_window = True
    elif "--no-app-window" in sys.argv:
        native_window = False
    else:
        native_window = not guest
    width = None
    height = None
    for i, arg in enumerate(sys.argv):
        if arg in ("--fullscreen-width", "--width") and i + 1 < len(sys.argv):
            try:
                width = int(sys.argv[i + 1])
            except ValueError:
                pass
        elif arg in ("--fullscreen-height", "--height") and i + 1 < len(sys.argv):
            try:
                height = int(sys.argv[i + 1])
            except ValueError:
                pass
        elif arg.startswith("--resolution=") or (arg == "--resolution" and i + 1 < len(sys.argv)):
            raw = arg.split("=", 1)[1] if "=" in arg else sys.argv[i + 1]
            if "x" in raw.lower():
                try:
                    w_str, h_str = raw.lower().split("x", 1)
                    width, height = int(w_str), int(h_str)
                except ValueError:
                    pass
    if "--no-browser" not in sys.argv:
        opened = open_ui(url, guest=guest, force_game_mode=force_game_mode, native_window=native_window, width=width, height=height)
        browser_pid = opened.get("pid")
        if opened.get("mode") == "kiosk" and guest and browser_pid:
            browser_name = Path(str(opened.get("browser") or "")).name
            class_hint = "google-chrome" if "chrome" in browser_name.casefold() else browser_name
            threading.Thread(
                target=mark_process_windows,
                kwargs={
                    "pid": browser_pid,
                    "app_id": OPENBOX_STEAM_GAME_ID,
                    "window_class": class_hint,
                },
                daemon=True,
            ).start()
    def stop():
        """Graceful shutdown: stop sessions, then stop accepting requests."""
        # Days 0-14, Task 3: mark sessions stopping, graceful stop, force-kill pgid if needed, persist before remove.
        with PROCESS_LOCK:
            launch_ids = list(RUNNING.keys())
        for launch_id in launch_ids:
            try:
                control_game_session(launch_id, "stop")
            except ValueError:
                pass
        # Mark live sessions stopping, give them the shutdown window, then remove the rows.
        reap_sessions_for_shutdown()
        # Lifecycle event (F1e): the app is shutting down. Best-effort and
        # bounded by the per-plugin timeout; never blocks teardown.
        emit_plugin_event(DATA.parent / "plugins", "app_shutdown", {})
        shutdown_webhooks(wait_seconds=2.0)

    def request_shutdown(_signum, _frame):
        # Serve_forever is running in this thread; raise so the finally block
        # runs the graceful teardown. The host sends SIGTERM after POSTing
        # /api/shutdown, so this is the backstop, not the primary path.
        raise KeyboardInterrupt

    previous_sigterm = signal.signal(signal.SIGTERM, request_shutdown)
    previous_sigint = signal.signal(signal.SIGINT, request_shutdown)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        stop()
        WATCH_STOP.set()
        JOB_MANAGER.cancel("auto-import")
        JOB_MANAGER.shutdown(wait=False, cancel_futures=True)
        server.server_close()
        (DATA.parent / "server.token").unlink(missing_ok=True)
        (DATA.parent / "server.port").unlink(missing_ok=True)
        run_configured_commands("shutdown_commands")
        signal.signal(signal.SIGTERM, previous_sigterm)
        signal.signal(signal.SIGINT, previous_sigint)


if __name__ == "__main__":
    main()