From 06c41bccba02296f7ecb6a973a2789bd008744d3 Mon Sep 17 00:00:00 2001 From: yellowcooln <12516003+yellowcooln@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:04:34 -0400 Subject: [PATCH] fix(plugins): harden provisioning, runtime, IPC and native upgrades --- docs/plugins.md | 68 ++++ manage.sh | 172 +++++++--- openhop-plugin-manager.service | 2 + pyproject.toml | 1 + repeater/plugins/catalogue.py | 100 +++++- repeater/plugins/github_releases.py | 40 ++- repeater/plugins/ipc.py | 170 ++++++++-- repeater/plugins/manager.py | 314 +++++++++++------- repeater/plugins/manifest.py | 49 ++- .../plugins/openhop-plugin-manager.service | 36 ++ repeater/plugins/runtime.py | 219 ++++++++---- repeater/plugins/storage.py | 52 ++- repeater/web/api_endpoints.py | 21 +- repeater/web/http_server.py | 21 +- repeater/web/openapi.yaml | 57 ++++ repeater/web/plugin_endpoints.py | 16 +- repeater/web/update_endpoints.py | 134 +++++--- tests/test_api_endpoints_core_coverage.py | 2 + tests/test_api_endpoints_setup.py | 4 +- tests/test_native_updater_security.py | 189 +++++++++++ tests/test_plugin_api_endpoints.py | 48 +++ tests/test_plugin_download_boundaries.py | 173 ++++++++++ tests/test_plugin_ipc.py | 9 +- tests/test_plugin_ipc_hardening.py | 166 +++++++++ tests/test_plugin_manager_lifecycle.py | 12 + tests/test_plugin_runtime_hardening.py | 309 +++++++++++++++++ tests/test_plugin_static_security.py | 60 ++++ tests/test_plugin_timeout_openapi.py | 22 ++ tests/test_setup_authorization_hardening.py | 191 +++++++++++ tests/test_update_endpoints_unit.py | 7 +- 30 files changed, 2294 insertions(+), 370 deletions(-) create mode 100644 repeater/plugins/openhop-plugin-manager.service create mode 100644 tests/test_native_updater_security.py create mode 100644 tests/test_plugin_download_boundaries.py create mode 100644 tests/test_plugin_ipc_hardening.py create mode 100644 tests/test_plugin_runtime_hardening.py create mode 100644 tests/test_plugin_static_security.py create mode 100644 tests/test_plugin_timeout_openapi.py create mode 100644 tests/test_setup_authorization_hardening.py diff --git a/docs/plugins.md b/docs/plugins.md index 9abb05c..09a5461 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -4,6 +4,26 @@ The plugin system installs and supervises **external** applications around the Repeater. Plugins are not imported into the Repeater process. They talk to existing interfaces (REST, WebSocket/SSE, Companion frame server, static web). +## Trust and access + +Plugins are trusted applications, **not sandboxed code**. A separate process and +virtualenv keep Python dependencies separate; they do not create a filesystem, +network, or credential boundary. Install only plugins and dependencies you trust. +Native plugins currently run under the `repeater` service account; container +plugins run under the configured container account. Never run a plugin manager as +root merely to work around installation permissions. + +The current Repeater API tokens are administrator-equivalent. A token that can +read status can also administer plugins, install uploaded wheels, and read plugin +settings. Do not hand these tokens to untrusted or telemetry-only consumers. +Scoped integration credentials would require a separate authorization feature. + +Schema-2 catalogue checksums approve the plugin wheel itself. Python dependencies +resolved by pip are not covered by that checksum and can change on installation +or environment rebuild. The curated catalogue is not a claim that the complete +transitive dependency set is hash-locked. Deployments requiring reproducibility +should use a controlled, pinned dependency source. + ## Layout Default root (under Repeater storage): @@ -70,6 +90,12 @@ UI-only application plugin: A plugin may declare both `runtime` and `ui`. +The UI entry must be inside a dedicated asset subtree such as `ui/index.html`. +A release-root entry (`index.html`), dot/parent path segments, and reserved +runtime/storage subtrees (`venv`, `data`, `logs`, `releases`, `current`) are +rejected. Existing plugins using these invalid layouts must be repackaged; +serving the whole release directory is intentionally no longer supported. + ## Plugin manager service Native installs: @@ -92,6 +118,48 @@ both services and their plugin processes, and restarts the manager if it exits unexpectedly while Repeater remains running. Set `OPENHOP_PLUGIN_MANAGER=0` to run Repeater without the manager. +### Native upgrade prerequisite + +Existing native installations must run the updated `manage.sh upgrade` once as +an administrator to replace the privileged OTA helper. Installing only a new +Python wheel cannot safely replace an existing root-owned helper. Subsequent +OTA upgrades through the refreshed helper install the manager unit from the +installed package; no source checkout is required on the host. + +The helper runs from a trusted directory with a clean environment and isolated +Python module lookup. It rejects writable/non-root-owned execution trees instead +of silently granting existing code root ownership. If this check fails, an +administrator must restore the application environment from a trusted package. +Do not bypass the check or make the venv service-writable. + +## Operation safety and limits + +The manager serves bounded concurrent IPC requests, so a slow installation does +not monopolize the socket. Mutations of the same plugin are serialized; +conflicting requests return a busy response rather than interleaving state. +Local wheel inputs are copied to manager-owned staging before installation. + +Installation has a longer IPC budget than ordinary status requests. If transport +is lost after dispatch, HTTP 504 with `outcome: unknown` means the operation may +still finish: **it does not mean cancellation or rollback**. Inspect plugin +status before retrying. An uploaded source is retained when completion is +ambiguous so an accepted install cannot lose its input. + +Venv creation and pip execution are time-limited, installer output is bounded, +and plugin stdout/stderr are drained into bounded logs while the plugin runs. +Stop/disable terminate the supervised process group, including ordinary child +workers. This is process supervision, not containment of malicious applications +that deliberately detach into new sessions or write other files. + +Failed Python-version rebuilds restore the previous venv and remain retryable; +the new marker is accepted only after the new environment and entrypoint exist. + +Archive extraction limits expanded member size and total size and rejects +symlink-backed destinations. Catalogue JSON and wheel response bodies have size +limits and elapsed-time budgets; configured catalogue URLs must use HTTPS. +DNS resolution and urllib connection/header setup are not guaranteed to obey a +strict whole-request wall-clock deadline. + ## Install a local wheel Via API (authenticated): diff --git a/manage.sh b/manage.sh index ca0b2f7..cd52944 100755 --- a/manage.sh +++ b/manage.sh @@ -767,20 +767,40 @@ EOF echo ">>> Installing OTA upgrade wrapper..." cat > /usr/local/bin/pymc-do-upgrade <<'UPGRADEEOF' -#!/bin/bash -# pymc-do-upgrade: invoked by the repeater service user via sudo for OTA upgrades. -# Usage: sudo /usr/local/bin/pymc-do-upgrade [channel] [pretend-version] +#!/bin/bash -p +# Ignore caller shell startup hooks, then run with a clean environment. +exec /usr/bin/env -i PATH=/usr/sbin:/usr/bin:/sbin:/bin HOME=/root LANG=C.UTF-8 PIP_CONFIG_FILE=/dev/null /bin/bash --noprofile --norc -s -- "$@" <<'CLEANENV' set -Eeuo pipefail +umask 022 +cd / +[[ "$EUID" -eq 0 ]] || { echo "Root is required" >&2; exit 1; } +[[ "$#" -le 2 ]] || { echo "Too many arguments" >&2; exit 1; } +# Never repair untrusted executable trees by chowning attacker-owned code. +require_root_owned() { + local path="$1" unsafe + unsafe=$(/usr/bin/find -L "$path" -maxdepth 0 \( ! -uid 0 -o -perm /022 \) -print) || exit 1 + [[ -z "$unsafe" ]] || { echo "Unsafe root execution path: $path" >&2; exit 1; } +} +for path in / /opt /opt/openhop_repeater /usr /usr/bin /usr/bin/python3; do + require_root_owned "$path" +done +if [[ -e /opt/openhop_repeater/venv ]]; then + unsafe=$(/usr/bin/find -L /opt/openhop_repeater/venv \( ! -uid 0 -o -perm /022 \) -print -quit) || exit 1 + [[ -z "$unsafe" ]] || { echo "Unsafe venv; administrator repair required" >&2; exit 1; } +fi CHANNEL="${1:-main}" PRETEND_VERSION="${2:-}" VENV_DIR="/opt/openhop_repeater/venv" -VENV_PIP="$VENV_DIR/bin/pip" VENV_PYTHON="$VENV_DIR/bin/python" # Validate: only allow safe git ref characters if ! [[ "$CHANNEL" =~ ^[a-zA-Z0-9._/-]{1,80}$ ]]; then echo "Invalid channel name: $CHANNEL" >&2 exit 1 fi +if [[ -n "$PRETEND_VERSION" ]] && ! [[ "$PRETEND_VERSION" =~ ^[a-zA-Z0-9][a-zA-Z0-9.!+_-]{0,127}$ ]]; then + echo "Invalid version" >&2 + exit 1 +fi # If caller supplied a version string, tell setuptools_scm to use it (sudo # strips env vars so it is passed as a positional argument instead). Scoped to # this distribution: the unscoped variable applies to every setuptools_scm @@ -790,8 +810,8 @@ fi # ---- Migration: ensure venv exists (handles upgrades from system-pip era) ---- if [ ! -x "$VENV_PYTHON" ]; then echo "[pymc-do-upgrade] Creating venv at $VENV_DIR ..." - python3 -m venv --system-site-packages "$VENV_DIR" - "$VENV_PYTHON" -m pip install --upgrade pip setuptools wheel + /usr/bin/python3 -I -m venv --system-site-packages "$VENV_DIR" + "$VENV_PYTHON" -I -m pip install --upgrade pip setuptools wheel fi # ---- Migration: clean up legacy service unit issues ---- SVC_UNIT=/etc/systemd/system/openhop-repeater.service @@ -822,10 +842,10 @@ fi [ -d /opt/pymc_repeater/repeater ] && rm -rf /opt/pymc_repeater/repeater [ -d /opt/pymc_repeater/pymc-repeater ] && rm -rf /opt/pymc_repeater/pymc-repeater # ---- Remove old system-level packages to avoid confusion ---- -python3 -m pip uninstall -y openhop_repeater 2>/dev/null || true -python3 -m pip uninstall -y openhop_core 2>/dev/null || true -python3 -m pip uninstall -y pymc_repeater 2>/dev/null || true -python3 -m pip uninstall -y pymc_core 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y openhop_repeater 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y openhop_core 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y pymc_repeater 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y pymc_core 2>/dev/null || true # ---- Try R2 wheels first for faster OTA upgrades ---- R2_BASE_URL="https://wheel.pymc.dev/pymc_build_deps" MACHINE_ARCH=$(uname -m) @@ -836,16 +856,28 @@ case "$MACHINE_ARCH" in *) ARCH_TAG=""; PLATFORM_TAG="" ;; esac if [ -n "$ARCH_TAG" ]; then - PY_TAG=$("$VENV_PYTHON" -c 'import sys; v=f"cp{sys.version_info.major}{sys.version_info.minor}"; print(f"{v}-{v}")' 2>/dev/null || echo "cp311-cp311") + PY_TAG=$("$VENV_PYTHON" -I -c 'import sys; v=f"cp{sys.version_info.major}{sys.version_info.minor}"; print(f"{v}-{v}")' 2>/dev/null || echo "cp311-cp311") WHEEL_BASE="${R2_BASE_URL}/${ARCH_TAG}/${PLATFORM_TAG}/${PY_TAG}" echo "[pymc-do-upgrade] Trying dependencies from R2 wheels..." - "$VENV_PYTHON" -m pip install --find-links "${WHEEL_BASE}/index.html" --no-cache-dir "pycryptodome>=3.23.0" "PyNaCl>=1.5.0" cffi "pyyaml>=6.0.0" 2>/dev/null || true + "$VENV_PYTHON" -I -m pip install --find-links "${WHEEL_BASE}/index.html" --no-cache-dir "pycryptodome>=3.23.0" "PyNaCl>=1.5.0" cffi "pyyaml>=6.0.0" 2>/dev/null || true fi # ---- Install openhop_repeater from git ---- -if "$VENV_PYTHON" -m pip install \ +if "$VENV_PYTHON" -I -m pip install \ --upgrade \ --no-cache-dir \ "openhop_repeater[hardware] @ git+https://github.com/openhop-dev/openhop_repeater.git@${CHANNEL}"; then + # Use the just-installed distribution, never a caller checkout or cwd. + PLUGIN_UNIT=$("$VENV_PYTHON" -I -c 'from importlib.metadata import distribution; print(distribution("openhop_repeater").locate_file("repeater/plugins/openhop-plugin-manager.service"))') + if [[ -f "$PLUGIN_UNIT" ]]; then + require_root_owned "$PLUGIN_UNIT" + install -o root -g root -m 0644 "$PLUGIN_UNIT" /etc/systemd/system/openhop-plugin-manager.service + systemctl daemon-reload + systemctl enable openhop-plugin-manager + # ExecCondition skips startup successfully when plugins.enabled is false. + systemctl restart openhop-plugin-manager || echo "[pymc-do-upgrade] WARNING: inspect openhop-plugin-manager service status" >&2 + else + echo "[pymc-do-upgrade] WARNING: selected package does not ship a plugin-manager unit; older channels may not support plugins. Re-run manage.sh upgrade from a plugin-capable release if unexpected." >&2 + fi # Keep web/OTA updates aligned with manage.sh install/upgrade defaults. RADIO_BASE_URL="https://raw.githubusercontent.com/openhop-dev/openhop_repeater/${CHANNEL}" RADIO_STORAGE_DIR="/var/lib/openhop_repeater" @@ -855,7 +887,9 @@ if "$VENV_PYTHON" -m pip install \ else exit 1 fi +CLEANENV UPGRADEEOF + chown root:root /usr/local/bin/pymc-do-upgrade chmod 0755 /usr/local/bin/pymc-do-upgrade echo "75"; echo "# Starting service..." @@ -1283,20 +1317,40 @@ EOF chmod 0440 /etc/sudoers.d/openhop-repeater cat > /usr/local/bin/pymc-do-upgrade <<'UPGRADEEOF' -#!/bin/bash -# pymc-do-upgrade: invoked by the repeater service user via sudo for OTA upgrades. -# Usage: sudo /usr/local/bin/pymc-do-upgrade [channel] [pretend-version] +#!/bin/bash -p +# Ignore caller shell startup hooks, then run with a clean environment. +exec /usr/bin/env -i PATH=/usr/sbin:/usr/bin:/sbin:/bin HOME=/root LANG=C.UTF-8 PIP_CONFIG_FILE=/dev/null /bin/bash --noprofile --norc -s -- "$@" <<'CLEANENV' set -Eeuo pipefail +umask 022 +cd / +[[ "$EUID" -eq 0 ]] || { echo "Root is required" >&2; exit 1; } +[[ "$#" -le 2 ]] || { echo "Too many arguments" >&2; exit 1; } +# Never repair untrusted executable trees by chowning attacker-owned code. +require_root_owned() { + local path="$1" unsafe + unsafe=$(/usr/bin/find -L "$path" -maxdepth 0 \( ! -uid 0 -o -perm /022 \) -print) || exit 1 + [[ -z "$unsafe" ]] || { echo "Unsafe root execution path: $path" >&2; exit 1; } +} +for path in / /opt /opt/openhop_repeater /usr /usr/bin /usr/bin/python3; do + require_root_owned "$path" +done +if [[ -e /opt/openhop_repeater/venv ]]; then + unsafe=$(/usr/bin/find -L /opt/openhop_repeater/venv \( ! -uid 0 -o -perm /022 \) -print -quit) || exit 1 + [[ -z "$unsafe" ]] || { echo "Unsafe venv; administrator repair required" >&2; exit 1; } +fi CHANNEL="${1:-main}" PRETEND_VERSION="${2:-}" VENV_DIR="/opt/openhop_repeater/venv" -VENV_PIP="$VENV_DIR/bin/pip" VENV_PYTHON="$VENV_DIR/bin/python" # Validate: only allow safe git ref characters if ! [[ "$CHANNEL" =~ ^[a-zA-Z0-9._/-]{1,80}$ ]]; then echo "Invalid channel name: $CHANNEL" >&2 exit 1 fi +if [[ -n "$PRETEND_VERSION" ]] && ! [[ "$PRETEND_VERSION" =~ ^[a-zA-Z0-9][a-zA-Z0-9.!+_-]{0,127}$ ]]; then + echo "Invalid version" >&2 + exit 1 +fi # If caller supplied a version string, tell setuptools_scm to use it (sudo # strips env vars so it is passed as a positional argument instead). Scoped to # this distribution: the unscoped variable applies to every setuptools_scm @@ -1306,8 +1360,8 @@ fi # ---- Migration: ensure venv exists (handles upgrades from system-pip era) ---- if [ ! -x "$VENV_PYTHON" ]; then echo "[pymc-do-upgrade] Creating venv at $VENV_DIR ..." - python3 -m venv --system-site-packages "$VENV_DIR" - "$VENV_PYTHON" -m pip install --upgrade pip setuptools wheel + /usr/bin/python3 -I -m venv --system-site-packages "$VENV_DIR" + "$VENV_PYTHON" -I -m pip install --upgrade pip setuptools wheel fi # ---- Migration: clean up legacy service unit issues ---- SVC_UNIT=/etc/systemd/system/openhop-repeater.service @@ -1338,40 +1392,54 @@ fi [ -d /opt/pymc_repeater/repeater ] && rm -rf /opt/pymc_repeater/repeater [ -d /opt/pymc_repeater/pymc-repeater ] && rm -rf /opt/pymc_repeater/pymc-repeater # ---- Remove old system-level packages to avoid confusion ---- -python3 -m pip uninstall -y openhop_repeater 2>/dev/null || true -python3 -m pip uninstall -y openhop_core 2>/dev/null || true -python3 -m pip uninstall -y pymc_repeater 2>/dev/null || true -python3 -m pip uninstall -y pymc_core 2>/dev/null || true - # ---- Try R2 wheels first for faster OTA upgrades ---- - R2_BASE_URL="https://wheel.pymc.dev/pymc_build_deps" - MACHINE_ARCH=$(uname -m) - case "$MACHINE_ARCH" in - aarch64) ARCH_TAG="arm64"; PLATFORM_TAG="aarch64" ;; - armv7l|armv7) ARCH_TAG="armv7"; PLATFORM_TAG="armv7l" ;; - x86_64) ARCH_TAG="x86_64"; PLATFORM_TAG="x86_64" ;; - *) ARCH_TAG=""; PLATFORM_TAG="" ;; - esac - if [ -n "$ARCH_TAG" ]; then - PY_TAG=$("$VENV_PYTHON" -c 'import sys; v=f"cp{sys.version_info.major}{sys.version_info.minor}"; print(f"{v}-{v}")' 2>/dev/null || echo "cp311-cp311") - WHEEL_BASE="${R2_BASE_URL}/${ARCH_TAG}/${PLATFORM_TAG}/${PY_TAG}" - echo "[pymc-do-upgrade] Trying dependencies from R2 wheels..." - "$VENV_PYTHON" -m pip install --find-links "${WHEEL_BASE}/index.html" --no-cache-dir "pycryptodome>=3.23.0" "PyNaCl>=1.5.0" cffi "pyyaml>=6.0.0" 2>/dev/null || true - fi - # ---- Install openhop_repeater from git ---- - if "$VENV_PYTHON" -m pip install \ - --upgrade \ - --no-cache-dir \ - "openhop_repeater[hardware] @ git+https://github.com/openhop-dev/openhop_repeater.git@${CHANNEL}"; then - # Keep web/OTA updates aligned with manage.sh install/upgrade defaults. - RADIO_BASE_URL="https://raw.githubusercontent.com/openhop-dev/openhop_repeater/${CHANNEL}" - RADIO_STORAGE_DIR="/var/lib/openhop_repeater" - mkdir -p "$RADIO_STORAGE_DIR" - wget -qO "$RADIO_STORAGE_DIR/radio-settings.json" "${RADIO_BASE_URL}/radio-settings.json" 2>/dev/null || true - wget -qO "$RADIO_STORAGE_DIR/radio-presets.json" "${RADIO_BASE_URL}/radio-presets.json" 2>/dev/null || true - else - exit 1 - fi +/usr/bin/python3 -I -m pip uninstall -y openhop_repeater 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y openhop_core 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y pymc_repeater 2>/dev/null || true +/usr/bin/python3 -I -m pip uninstall -y pymc_core 2>/dev/null || true +# ---- Try R2 wheels first for faster OTA upgrades ---- +R2_BASE_URL="https://wheel.pymc.dev/pymc_build_deps" +MACHINE_ARCH=$(uname -m) +case "$MACHINE_ARCH" in + aarch64) ARCH_TAG="arm64"; PLATFORM_TAG="aarch64" ;; + armv7l|armv7) ARCH_TAG="armv7"; PLATFORM_TAG="armv7l" ;; + x86_64) ARCH_TAG="x86_64"; PLATFORM_TAG="x86_64" ;; + *) ARCH_TAG=""; PLATFORM_TAG="" ;; +esac +if [ -n "$ARCH_TAG" ]; then + PY_TAG=$("$VENV_PYTHON" -I -c 'import sys; v=f"cp{sys.version_info.major}{sys.version_info.minor}"; print(f"{v}-{v}")' 2>/dev/null || echo "cp311-cp311") + WHEEL_BASE="${R2_BASE_URL}/${ARCH_TAG}/${PLATFORM_TAG}/${PY_TAG}" + echo "[pymc-do-upgrade] Trying dependencies from R2 wheels..." + "$VENV_PYTHON" -I -m pip install --find-links "${WHEEL_BASE}/index.html" --no-cache-dir "pycryptodome>=3.23.0" "PyNaCl>=1.5.0" cffi "pyyaml>=6.0.0" 2>/dev/null || true +fi +# ---- Install openhop_repeater from git ---- +if "$VENV_PYTHON" -I -m pip install \ + --upgrade \ + --no-cache-dir \ + "openhop_repeater[hardware] @ git+https://github.com/openhop-dev/openhop_repeater.git@${CHANNEL}"; then + # Use the just-installed distribution, never a caller checkout or cwd. + PLUGIN_UNIT=$("$VENV_PYTHON" -I -c 'from importlib.metadata import distribution; print(distribution("openhop_repeater").locate_file("repeater/plugins/openhop-plugin-manager.service"))') + if [[ -f "$PLUGIN_UNIT" ]]; then + require_root_owned "$PLUGIN_UNIT" + install -o root -g root -m 0644 "$PLUGIN_UNIT" /etc/systemd/system/openhop-plugin-manager.service + systemctl daemon-reload + systemctl enable openhop-plugin-manager + # ExecCondition skips startup successfully when plugins.enabled is false. + systemctl restart openhop-plugin-manager || echo "[pymc-do-upgrade] WARNING: inspect openhop-plugin-manager service status" >&2 + else + echo "[pymc-do-upgrade] WARNING: selected package does not ship a plugin-manager unit; older channels may not support plugins. Re-run manage.sh upgrade from a plugin-capable release if unexpected." >&2 + fi + # Keep web/OTA updates aligned with manage.sh install/upgrade defaults. + RADIO_BASE_URL="https://raw.githubusercontent.com/openhop-dev/openhop_repeater/${CHANNEL}" + RADIO_STORAGE_DIR="/var/lib/openhop_repeater" + mkdir -p "$RADIO_STORAGE_DIR" + wget -qO "$RADIO_STORAGE_DIR/radio-settings.json" "${RADIO_BASE_URL}/radio-settings.json" 2>/dev/null || true + wget -qO "$RADIO_STORAGE_DIR/radio-presets.json" "${RADIO_BASE_URL}/radio-presets.json" 2>/dev/null || true +else + exit 1 +fi +CLEANENV UPGRADEEOF + chown root:root /usr/local/bin/pymc-do-upgrade chmod 0755 /usr/local/bin/pymc-do-upgrade echo " ✓ Permissions updated" diff --git a/openhop-plugin-manager.service b/openhop-plugin-manager.service index b7f7046..8549e6b 100644 --- a/openhop-plugin-manager.service +++ b/openhop-plugin-manager.service @@ -16,6 +16,8 @@ User=repeater Group=repeater WorkingDirectory=/var/lib/openhop_repeater +# Disabled plugins are an intentional opt-out, not a failed service startup. +ExecCondition=/opt/openhop_repeater/venv/bin/python -I -c "import sys,yaml; c=yaml.safe_load(open(sys.argv[1])) or {}; p=c.get('plugins'); sys.exit(1 if isinstance(p, dict) and p.get('enabled') is False else 0)" /etc/openhop_repeater/config.yaml ExecStart=/opt/openhop_repeater/venv/bin/openhop-plugin-manager --config /etc/openhop_repeater/config.yaml Restart=on-failure diff --git a/pyproject.toml b/pyproject.toml index cd0211f..a8dfab4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,6 +75,7 @@ repeater = [ "web/*.yaml", "web/*.html", "presets/*.yaml", + "plugins/*.service", ] [tool.ruff] diff --git a/repeater/plugins/catalogue.py b/repeater/plugins/catalogue.py index 3eb71de..35fb63e 100644 --- a/repeater/plugins/catalogue.py +++ b/repeater/plugins/catalogue.py @@ -7,8 +7,10 @@ import json import logging import os import re +import socket import ssl import tempfile +import threading import time import urllib.error import urllib.request @@ -26,6 +28,66 @@ USER_AGENT = "openhop-repeater-plugin-manager/1.0" GITHUB_RELEASE_ORIGIN = "https://github.com" GITHUB_RELEASE_ASSET_HOST = "release-assets.githubusercontent.com" MAX_WHEEL_BYTES = 100 * 1024 * 1024 +MAX_JSON_BYTES = 4 * 1024 * 1024 +JSON_DOWNLOAD_TIMEOUT_SECONDS = 30.0 +WHEEL_DOWNLOAD_TIMEOUT_SECONDS = 120.0 + + +def _bounded_chunks(response, *, max_bytes: int, deadline: float): + """Stream a bounded body, sharing the budget started before opening the URL. + + HTTPResponse.read(n) can wait for n bytes despite a trickling peer. read1 + returns after one buffered/socket read; resetting the socket timeout bounds + an idle final read by the *remaining* budget. The shutdown watchdog also + interrupts read1's internal chunk-header/trailer readline loops. + + urllib's DNS, connect, TLS and response-header/redirect processing happen + before the response is exposed here: their timeout is not a strict total + deadline. Injected non-socket openers must themselves provide bounded reads. + """ + + def remaining(): + budget = deadline - time.monotonic() + if budget <= 0: + raise TimeoutError("download deadline exceeded") + return budget + + raw = getattr(getattr(response, "fp", None), "raw", None) + sock = getattr(raw, "_sock", None) + watchdog = None + if isinstance(sock, socket.socket): + + def interrupt(): + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass # Response may already have closed its socket at EOF. + + watchdog = threading.Timer(remaining(), interrupt) + watchdog.daemon = True + watchdog.start() + read = getattr(response, "read1", None) or response.read + total = 0 + try: + while True: + budget = remaining() + if isinstance(sock, socket.socket) and sock.fileno() >= 0: + sock.settimeout(budget) + chunk = read(min(64 * 1024, max_bytes - total + 1)) + remaining() + if not chunk: + if getattr(response, "length", None): + raise ValueError("download ended before Content-Length was received") + return + total += len(chunk) + if total > max_bytes: + raise ValueError(f"download exceeds {max_bytes} bytes") + yield chunk + finally: + if watchdog is not None: + watchdog.cancel() + watchdog.join() + _REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]*$") @@ -269,7 +331,12 @@ def parse_catalogue(data: Any) -> Catalogue: class CatalogueClient: - """Fetch, cache, and download from the curated plugin catalogue.""" + """Fetch, cache, and download from the curated plugin catalogue. + + Configured catalogue URLs must use HTTPS; HTTP and other schemes are + rejected with code 400, not silently upgraded. Custom HTTPS origins, + ports, paths and query strings remain supported. + """ def __init__( self, @@ -280,6 +347,10 @@ class CatalogueClient: user_agent: str = USER_AGENT, ): self.url = str(url or DEFAULT_CATALOGUE_URL).strip() or DEFAULT_CATALOGUE_URL + # Reject insecure configuration; never silently rewrite HTTP to HTTPS. + parsed_url = urlsplit(self.url) + if parsed_url.scheme != "https" or not parsed_url.hostname: + raise CatalogueError("Plugin catalogue URL must use HTTPS", code=400) self.cache_ttl = cache_ttl self._opener = opener or self._default_open self.user_agent = user_agent @@ -314,9 +385,14 @@ class CatalogueClient: }, method="GET", ) + deadline = time.monotonic() + JSON_DOWNLOAD_TIMEOUT_SECONDS try: - with self._opener(request, timeout=30.0) as resp: - body = resp.read() + with self._opener(request, timeout=JSON_DOWNLOAD_TIMEOUT_SECONDS) as resp: + body = bytearray() + for chunk in _bounded_chunks(resp, max_bytes=MAX_JSON_BYTES, deadline=deadline): + body.extend(chunk) + except (ValueError, TimeoutError) as exc: + raise CatalogueError(f"Plugin catalogue download failed: {exc}", code=502) from exc except urllib.error.HTTPError as exc: raise CatalogueError( f"Plugin catalogue is currently unavailable (HTTP {exc.code})", @@ -378,23 +454,23 @@ class CatalogueClient: try: with os.fdopen(fd, "wb") as output: try: - with self._opener(request, timeout=120.0) as response: + deadline = time.monotonic() + WHEEL_DOWNLOAD_TIMEOUT_SECONDS + with self._opener(request, timeout=WHEEL_DOWNLOAD_TIMEOUT_SECONDS) as response: final_url = getattr(response, "geturl", lambda: plugin.wheel_url)() _validate_download_response_url( plugin.wheel_url, final_url or plugin.wheel_url, ) - while True: - chunk = response.read(1024 * 1024) - if not chunk: - break + for chunk in _bounded_chunks( + response, max_bytes=MAX_WHEEL_BYTES, deadline=deadline + ): total += len(chunk) - if total > MAX_WHEEL_BYTES: - raise CatalogueError( - "approved plugin wheel exceeds 100 MiB", code=502 - ) digest.update(chunk) output.write(chunk) + except (ValueError, TimeoutError) as exc: + raise CatalogueError( + f"approved plugin wheel download failed: {exc}", code=502 + ) from exc except CatalogueError: raise except urllib.error.HTTPError as exc: diff --git a/repeater/plugins/github_releases.py b/repeater/plugins/github_releases.py index 0deef90..e03b948 100644 --- a/repeater/plugins/github_releases.py +++ b/repeater/plugins/github_releases.py @@ -14,6 +14,14 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional +from .catalogue import ( + JSON_DOWNLOAD_TIMEOUT_SECONDS, + MAX_JSON_BYTES, + MAX_WHEEL_BYTES, + WHEEL_DOWNLOAD_TIMEOUT_SECONDS, + _bounded_chunks, +) + logger = logging.getLogger("PluginGitHubReleases") GITHUB_API_BASE = "https://api.github.com" @@ -224,9 +232,23 @@ class GitHubReleaseClient: }, method="GET", ) + deadline = time.monotonic() + WHEEL_DOWNLOAD_TIMEOUT_SECONDS + temporary = None try: - with self._opener(request, timeout=120.0) as resp: - data = resp.read() + with tempfile.NamedTemporaryFile( + prefix=".wheel-", dir=dest_dir, delete=False + ) as output: + temporary = Path(output.name) + total = 0 + with self._opener(request, timeout=WHEEL_DOWNLOAD_TIMEOUT_SECONDS) as resp: + for chunk in _bounded_chunks( + resp, max_bytes=MAX_WHEEL_BYTES, deadline=deadline + ): + output.write(chunk) + total += len(chunk) + if not total: + raise GitHubReleasesError("downloaded wheel is empty", code=502) + temporary.replace(dest) except GitHubReleasesError: raise except urllib.error.HTTPError as exc: @@ -235,10 +257,9 @@ class GitHubReleaseClient: raise GitHubReleasesError(f"failed to download wheel: {exc.reason}", code=502) from exc except Exception as exc: raise GitHubReleasesError(f"failed to download wheel: {exc}", code=502) from exc - - if not data: - raise GitHubReleasesError("downloaded wheel is empty", code=502) - dest.write_bytes(data) + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) return dest def download_latest_wheel( @@ -268,9 +289,12 @@ class GitHubReleaseClient: }, method="GET", ) + deadline = time.monotonic() + JSON_DOWNLOAD_TIMEOUT_SECONDS try: - with self._opener(request, timeout=30.0) as resp: - body = resp.read() + with self._opener(request, timeout=JSON_DOWNLOAD_TIMEOUT_SECONDS) as resp: + body = bytearray() + for chunk in _bounded_chunks(resp, max_bytes=MAX_JSON_BYTES, deadline=deadline): + body.extend(chunk) except GitHubReleasesError: raise except urllib.error.HTTPError as exc: diff --git a/repeater/plugins/ipc.py b/repeater/plugins/ipc.py index c919e26..2b2c26b 100644 --- a/repeater/plugins/ipc.py +++ b/repeater/plugins/ipc.py @@ -7,6 +7,7 @@ import logging import os import socket import threading +import time from pathlib import Path from typing import Any, Callable, Optional @@ -15,6 +16,10 @@ from .manager import PluginManager, PluginManagerError logger = logging.getLogger("PluginIPC") DEFAULT_TIMEOUT = 30.0 +OPERATION_TIMEOUT = 900.0 # completion budget, not a subprocess cancellation deadline +MAX_CONNECTIONS = 16 +SLOW_OPS = frozenset({"install", "catalogue_install", "update", "catalogue", "check_update"}) +INSTALL_OPS = frozenset({"install", "catalogue_install", "update"}) MAX_LINE_BYTES = 8 * 1024 * 1024 # install paths are short; keep headroom @@ -24,27 +29,54 @@ class PluginIPCError(Exception): self.code = code +class PluginIPCOutcomeUnknown(PluginIPCError): + """Transport stopped waiting; the manager may still complete the operation.""" + + def __init__(self, *, upload_safe: bool = False): + super().__init__( + "Plugin manager response was lost or timed out; outcome is unknown. " + "The operation may still complete. Check plugin status before retrying.", + 504, + ) + self.outcome = "unknown" + self.upload_safe = upload_safe + + class PluginManagerUnavailable(PluginIPCError): def __init__(self, message: str = "plugin manager unavailable"): super().__init__(message, code=503) -def _read_line(conn: socket.socket, max_bytes: int = MAX_LINE_BYTES) -> bytes: - buf = bytearray() +def _read_line( + conn: socket.socket, + max_bytes: int = MAX_LINE_BYTES, + *, + buffer: Optional[bytearray] = None, + deadline: Optional[float] = None, +) -> bytes: + buf = buffer if buffer is not None else bytearray() + if deadline is None: + deadline = time.monotonic() + (conn.gettimeout() or DEFAULT_TIMEOUT) while True: - chunk = conn.recv(4096) - if not chunk: - break - buf.extend(chunk) - if b"\n" in chunk: - break + end = buf.find(b"\n") + if end >= 0: + if end > max_bytes: + raise PluginIPCError("IPC message too large", 413) + line = bytes(buf[:end]) + del buf[: end + 1] + return line if len(buf) > max_bytes: raise PluginIPCError("IPC message too large", 413) - if not buf: - return b"" - # Take first line only - line, _, _rest = bytes(buf).partition(b"\n") - return line + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("IPC response deadline exceeded") + conn.settimeout(remaining) + chunk = conn.recv(min(4096, max_bytes + 1 - len(buf))) + if not chunk: + if buf: + raise PluginIPCError("incomplete IPC message", 502) + return b"" + buf.extend(chunk) def _send_msg(conn: socket.socket, payload: dict[str, Any]) -> None: @@ -61,6 +93,10 @@ class PluginIPCServer: self._sock: Optional[socket.socket] = None self._thread: Optional[threading.Thread] = None self._stop = threading.Event() + self._slots = threading.BoundedSemaphore(MAX_CONNECTIONS) + self._slow_slot = threading.BoundedSemaphore(1) + self._workers: set[threading.Thread] = set() + self._workers_lock = threading.Lock() self._handlers: dict[str, Callable[[dict[str, Any]], Any]] = { "list": self._op_list, "status": self._op_status, @@ -119,6 +155,11 @@ class PluginIPCServer: if thread and thread.is_alive(): thread.join(timeout=3) self._thread = None + deadline = time.monotonic() + 3 + with self._workers_lock: + workers = list(self._workers) + for worker in workers: + worker.join(timeout=max(0, deadline - time.monotonic())) if self.socket_path.exists(): try: self.socket_path.unlink() @@ -139,15 +180,33 @@ class PluginIPCServer: if self._stop.is_set(): break continue - try: - self._handle_connection(conn) - except Exception as exc: - logger.debug("IPC connection error: %s", exc) - finally: + if not self._slots.acquire(blocking=False): + # No unbounded executor queue; a saturated transport rejects immediately. try: - conn.close() + conn.settimeout(0.1) + _send_msg( + conn, {"ok": False, "error": "plugin IPC busy; retry later", "code": 503} + ) except OSError: pass + finally: + conn.close() + continue + worker = threading.Thread(target=self._connection_worker, args=(conn,), daemon=True) + with self._workers_lock: + self._workers.add(worker) + worker.start() + + def _connection_worker(self, conn: socket.socket) -> None: + try: + self._handle_connection(conn) + except Exception as exc: + logger.debug("IPC connection error: %s", exc) + finally: + conn.close() + with self._workers_lock: + self._workers.discard(threading.current_thread()) + self._slots.release() def _handle_connection(self, conn: socket.socket) -> None: conn.settimeout(DEFAULT_TIMEOUT) @@ -163,8 +222,26 @@ class PluginIPCServer: op = request.get("op") if not isinstance(op, str) or op not in self._handlers: raise PluginIPCError(f"unknown op: {op!r}", 400) - result = self._handlers[op](request) - _send_msg(conn, {"id": req_id, "ok": True, "result": result}) + slow = op in SLOW_OPS + if slow and not self._slow_slot.acquire(blocking=False): + raise PluginIPCError("plugin download/install busy; retry later", 503) + try: + if op == "install": + wheel_path = request.get("wheel_path") + if not isinstance(wheel_path, str) or not wheel_path.strip(): + raise PluginIPCError("wheel_path is required", 400) + with self.manager.stage_upload(wheel_path.strip()) as staged: + if request.get("completion_protocol") == 1: + _send_msg(conn, {"id": req_id, "ok": True, "processing": True}) + result = self.manager.install(staged) + else: + if op in INSTALL_OPS and request.get("completion_protocol") == 1: + _send_msg(conn, {"id": req_id, "ok": True, "processing": True}) + result = self._handlers[op](request) + _send_msg(conn, {"id": req_id, "ok": True, "result": result}) + finally: + if slow: + self._slow_slot.release() except PluginManagerError as exc: _send_msg( conn, @@ -175,6 +252,9 @@ class PluginIPCServer: conn, {"id": req_id, "ok": False, "error": str(exc), "code": int(exc.code)}, ) + except OSError: + # A disconnected waiter is not an operation failure or cancellation. + raise except Exception as exc: logger.exception("IPC handler failure") _send_msg( @@ -287,6 +367,7 @@ class PluginIPCClient: def __init__(self, socket_path: Path | str, timeout: float = DEFAULT_TIMEOUT): self.socket_path = Path(socket_path) self.timeout = timeout + self.operation_timeout = OPERATION_TIMEOUT self._next_id = 1 self._lock = threading.Lock() @@ -302,29 +383,48 @@ class PluginIPCClient: self._next_id += 1 request = {"id": req_id, "op": op, **params} + if op in INSTALL_OPS: + request["completion_protocol"] = 1 try: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(str(self.socket_path)) except OSError as exc: + sock.close() raise PluginManagerUnavailable(f"plugin manager unavailable: {exc}") from exc + upload_safe = False try: _send_msg(sock, request) - raw = _read_line(sock) - if not raw: - raise PluginManagerUnavailable("empty response from plugin manager") - try: - response = json.loads(raw.decode("utf-8")) - except json.JSONDecodeError as exc: - raise PluginIPCError(f"invalid IPC response: {exc}", 502) from exc - if not isinstance(response, dict): - raise PluginIPCError("invalid IPC response type", 502) - if not response.get("ok"): - code = int(response.get("code") or 500) - err = str(response.get("error") or "plugin manager error") - raise PluginIPCError(err, code) - return response.get("result") + buffer = bytearray() + deadline = time.monotonic() + self.timeout + completion_deadline = time.monotonic() + self.operation_timeout + while True: + try: + raw = _read_line(sock, buffer=buffer, deadline=deadline) + except PluginIPCError as exc: + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc + if not raw: + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) + try: + response = json.loads(raw.decode("utf-8")) + except (ValueError, UnicodeError) as exc: + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc + if not isinstance(response, dict): + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) + if not response.get("ok"): + code = int(response.get("code") or 500) + err = str(response.get("error") or "plugin manager error") + raise PluginIPCError(err, code) + if response.get("id") != req_id: + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) + if response.get("processing") is True: + upload_safe = op == "install" + deadline = completion_deadline + continue + return response.get("result") + except OSError as exc: + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc finally: try: sock.close() diff --git a/repeater/plugins/manager.py b/repeater/plugins/manager.py index 016945f..6bf529e 100644 --- a/repeater/plugins/manager.py +++ b/repeater/plugins/manager.py @@ -6,10 +6,18 @@ import json import logging import shutil import threading +import os +import stat +import tempfile +import time +from contextlib import contextmanager +from functools import wraps + + from pathlib import Path from typing import Any, Optional -from .catalogue import DEFAULT_CATALOGUE_URL, CatalogueClient, CatalogueError +from .catalogue import DEFAULT_CATALOGUE_URL, MAX_WHEEL_BYTES, CatalogueClient, CatalogueError from .github_releases import ( GitHubReleaseClient, GitHubReleasesError, @@ -23,6 +31,17 @@ from .storage import PluginStorage logger = logging.getLogger("PluginManager") +def _serialized_plugin(fn): + """Keep a complete lifecycle transaction exclusive, without queuing callers.""" + + @wraps(fn) + def wrapped(self, plugin_id, *args, **kwargs): + with self._operation(plugin_id): + return fn(self, plugin_id, *args, **kwargs) + + return wrapped + + class PluginManagerError(Exception): """Base error for plugin manager operations.""" @@ -46,9 +65,60 @@ class PluginManager: self.storage = storage self.runtime = runtime or PluginRuntime(storage) self._lock = threading.RLock() + self._operations: dict[str, tuple[int, int]] = {} self.catalogue = catalogue_client or CatalogueClient(catalogue_url or DEFAULT_CATALOGUE_URL) self.github = github_client or GitHubReleaseClient() + @contextmanager + def _operation(self, plugin_id: str): + owner = threading.get_ident() + with self._lock: + active = self._operations.get(plugin_id) + if active and active[0] != owner: + raise PluginManagerError(f"plugin operation already in progress: {plugin_id}", 409) + self._operations[plugin_id] = (owner, active[1] + 1 if active else 1) + try: + yield + finally: + with self._lock: + depth = self._operations[plugin_id][1] + if depth == 1: + del self._operations[plugin_id] + else: + self._operations[plugin_id] = (owner, depth - 1) + + @contextmanager + def stage_upload(self, wheel_path: str): + """Own a bounded regular-file copy before IPC acknowledges processing. + + Read through one opened descriptor, so unlinking an HTTP upload cannot + invalidate an in-progress copy. Never consume the caller's path later. + """ + root = self.storage.root / ".ipc-staging" + root.mkdir(parents=True, exist_ok=True) + limit = MAX_WHEEL_BYTES + deadline = time.monotonic() + 30 + with tempfile.TemporaryDirectory(prefix="install-", dir=root) as directory: + source = Path(wheel_path) + staged = Path(directory) / source.name + fd = os.open(source, os.O_RDONLY | os.O_NONBLOCK) + with os.fdopen(fd, "rb") as incoming: + metadata = os.fstat(incoming.fileno()) + if not stat.S_ISREG(metadata.st_mode): + raise PluginManagerError("wheel must be a regular file", 400) + if metadata.st_size > limit: + raise PluginManagerError("wheel exceeds 100 MiB limit", 413) + total = 0 + with staged.open("wb") as outgoing: + while chunk := incoming.read(1024 * 1024): + total += len(chunk) + if total > limit: + raise PluginManagerError("wheel exceeds 100 MiB limit", 413) + if time.monotonic() > deadline: + raise PluginManagerError("wheel staging deadline exceeded", 504) + outgoing.write(chunk) + yield staged + def start(self) -> None: """Load state and start enabled service plugins.""" self.storage.ensure_root() @@ -97,7 +167,7 @@ class PluginManager: manifest = load_manifest_from_wheel(wheel_path) except ManifestError as exc: raise PluginManagerError(str(exc), 400) from exc - with self._lock: + with self._operation(manifest.id): try: info = self.runtime.install_wheel(wheel_path, manifest) except Exception as exc: @@ -106,78 +176,78 @@ class PluginManager: # Return full status return self.status(info["id"]) + @_serialized_plugin def enable(self, plugin_id: str) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - state = dict(state) - state["enabled"] = True - self.storage.write_state(plugin_id, state) - manifest = self.storage.load_current_manifest(plugin_id) - if manifest and manifest.runtime is not None: - try: - self.runtime.start(plugin_id) - except Exception as exc: - raise PluginManagerError(f"enable start failed: {exc}", 500) from exc - else: - self.runtime._runtime_state[plugin_id] = PluginState.STOPPED - return self.status(plugin_id) - - def disable(self, plugin_id: str) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - try: - self.runtime.stop(plugin_id, mark_disabled=True) - except Exception as exc: - logger.warning("disable stop error for %s: %s", plugin_id, exc) - state = dict(state) - state["enabled"] = False - self.storage.write_state(plugin_id, state) - self.runtime._runtime_state[plugin_id] = PluginState.DISABLED - return self.status(plugin_id) - - def start_plugin(self, plugin_id: str) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - if not state.get("enabled", False): - raise PluginManagerError( - f"plugin is disabled; enable it before starting: {plugin_id}", 409 - ) + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + state = dict(state) + state["enabled"] = True + self.storage.write_state(plugin_id, state) + manifest = self.storage.load_current_manifest(plugin_id) + if manifest and manifest.runtime is not None: try: self.runtime.start(plugin_id) except Exception as exc: - raise PluginManagerError(f"start failed: {exc}", 500) from exc + raise PluginManagerError(f"enable start failed: {exc}", 500) from exc + else: + self.runtime._runtime_state[plugin_id] = PluginState.STOPPED return self.status(plugin_id) + @_serialized_plugin + def disable(self, plugin_id: str) -> dict[str, Any]: + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + try: + self.runtime.stop(plugin_id, mark_disabled=True) + except Exception as exc: + logger.warning("disable stop error for %s: %s", plugin_id, exc) + state = dict(state) + state["enabled"] = False + self.storage.write_state(plugin_id, state) + self.runtime._runtime_state[plugin_id] = PluginState.DISABLED + return self.status(plugin_id) + + @_serialized_plugin + def start_plugin(self, plugin_id: str) -> dict[str, Any]: + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + if not state.get("enabled", False): + raise PluginManagerError( + f"plugin is disabled; enable it before starting: {plugin_id}", 409 + ) + try: + self.runtime.start(plugin_id) + except Exception as exc: + raise PluginManagerError(f"start failed: {exc}", 500) from exc + return self.status(plugin_id) + + @_serialized_plugin def stop_plugin(self, plugin_id: str) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - try: - self.runtime.stop(plugin_id) - except Exception as exc: - raise PluginManagerError(f"stop failed: {exc}", 500) from exc + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + try: + self.runtime.stop(plugin_id) + except Exception as exc: + raise PluginManagerError(f"stop failed: {exc}", 500) from exc return self.status(plugin_id) + @_serialized_plugin def restart_plugin(self, plugin_id: str) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - if not state.get("enabled", False): - raise PluginManagerError( - f"plugin is disabled; enable it before restarting: {plugin_id}", 409 - ) - try: - self.runtime.restart(plugin_id) - except Exception as exc: - raise PluginManagerError(f"restart failed: {exc}", 500) from exc + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + if not state.get("enabled", False): + raise PluginManagerError( + f"plugin is disabled; enable it before restarting: {plugin_id}", 409 + ) + try: + self.runtime.restart(plugin_id) + except Exception as exc: + raise PluginManagerError(f"restart failed: {exc}", 500) from exc return self.status(plugin_id) def logs(self, plugin_id: str, tail: int = 200) -> dict[str, Any]: @@ -224,6 +294,7 @@ class PluginManager: "config": config, } + @_serialized_plugin def set_config( self, plugin_id: str, @@ -245,23 +316,22 @@ class PluginManager: if len(encoded.encode("utf-8")) > 256 * 1024: raise PluginManagerError("config exceeds 256 KiB limit", 413) - with self._lock: - try: - self.storage.write_config(plugin_id, config) - except ValueError as exc: - raise PluginManagerError(str(exc), 400) from exc + try: + self.storage.write_config(plugin_id, config) + except ValueError as exc: + raise PluginManagerError(str(exc), 400) from exc - restarted = False - if restart and state.get("enabled", False): - manifest = self.storage.load_current_manifest(plugin_id) - if manifest and manifest.runtime is not None: - try: - self.runtime.restart(plugin_id) - restarted = True - except Exception as exc: - raise PluginManagerError( - f"config saved but restart failed: {exc}", 500 - ) from exc + restarted = False + if restart and state.get("enabled", False): + manifest = self.storage.load_current_manifest(plugin_id) + if manifest and manifest.runtime is not None: + try: + self.runtime.restart(plugin_id) + restarted = True + except Exception as exc: + raise PluginManagerError( + f"config saved but restart failed: {exc}", 500 + ) from exc result = self.get_config(plugin_id) result["restarted"] = restarted @@ -286,19 +356,19 @@ class PluginManager: "runtime": runtime, } + @_serialized_plugin def uninstall(self, plugin_id: str, *, delete_data: bool = False) -> dict[str, Any]: - with self._lock: - state = self.storage.read_state(plugin_id) - if state is None: - raise PluginManagerError(f"plugin not found: {plugin_id}", 404) - try: - self.runtime.stop(plugin_id, mark_disabled=True) - except Exception as exc: - logger.warning("uninstall stop error for %s: %s", plugin_id, exc) - self.storage.remove_release_code(plugin_id, keep_data=not delete_data) - self.runtime._runtime_state.pop(plugin_id, None) - self.runtime._crash_times.pop(plugin_id, None) - self.runtime._exit_codes.pop(plugin_id, None) + state = self.storage.read_state(plugin_id) + if state is None: + raise PluginManagerError(f"plugin not found: {plugin_id}", 404) + try: + self.runtime.stop(plugin_id, mark_disabled=True) + except Exception as exc: + logger.warning("uninstall stop error for %s: %s", plugin_id, exc) + self.storage.remove_release_code(plugin_id, keep_data=not delete_data) + self.runtime._runtime_state.pop(plugin_id, None) + self.runtime._crash_times.pop(plugin_id, None) + self.runtime._exit_codes.pop(plugin_id, None) return { "id": plugin_id, "uninstalled": True, @@ -359,6 +429,7 @@ class PluginManager: plugins_out.append(row) return {"schema": catalogue.schema, "plugins": plugins_out} + @_serialized_plugin def install_from_catalogue( self, plugin_id: str, @@ -414,19 +485,18 @@ class PluginManager: 400, ) - with self._lock: - try: - info = self.runtime.install_wheel(wheel_path, manifest) - except Exception as exc: - raise PluginManagerError(f"install failed: {exc}", 500) from exc - # Record catalogue provenance (repository locked at install time) - st = self.storage.read_state(info["id"]) or {} - st = dict(st) - st["version"] = info["version"] - st["source"] = "catalogue" - st["repository"] = entry.repository - # Keep enabled flag from install_wheel; enable next - self.storage.write_state(info["id"], st) + try: + info = self.runtime.install_wheel(wheel_path, manifest) + except Exception as exc: + raise PluginManagerError(f"install failed: {exc}", 500) from exc + # Record catalogue provenance (repository locked at install time) + st = self.storage.read_state(info["id"]) or {} + st = dict(st) + st["version"] = info["version"] + st["source"] = "catalogue" + st["repository"] = entry.repository + # Keep enabled flag from install_wheel; enable next + self.storage.write_state(info["id"], st) # Catalogue installs are enabled by default return self.enable(info["id"]) @@ -497,6 +567,7 @@ class PluginManager: "releaseTag": latest.tag if latest else None, } + @_serialized_plugin def update_plugin( self, plugin_id: str, @@ -601,23 +672,22 @@ class PluginManager: 400, ) - with self._lock: - # Stop running process before swapping release - try: - self.runtime.stop(plugin_id) - except Exception as exc: - logger.debug("update stop %s: %s", plugin_id, exc) - try: - info = self.runtime.install_wheel(wheel_path, manifest) - except Exception as exc: - raise PluginManagerError(f"update install failed: {exc}", 500) from exc - st = self.storage.read_state(info["id"]) or {} - st = dict(st) - st["version"] = info["version"] - st["source"] = st.get("source") or "catalogue" - st["repository"] = repository # never silently retarget - st["enabled"] = was_enabled - self.storage.write_state(info["id"], st) + # Stop running process before swapping release + try: + self.runtime.stop(plugin_id) + except Exception as exc: + logger.debug("update stop %s: %s", plugin_id, exc) + try: + info = self.runtime.install_wheel(wheel_path, manifest) + except Exception as exc: + raise PluginManagerError(f"update install failed: {exc}", 500) from exc + st = self.storage.read_state(info["id"]) or {} + st = dict(st) + st["version"] = info["version"] + st["source"] = st.get("source") or "catalogue" + st["repository"] = repository # never silently retarget + st["enabled"] = was_enabled + self.storage.write_state(info["id"], st) if was_enabled: return self.enable(plugin_id) diff --git a/repeater/plugins/manifest.py b/repeater/plugins/manifest.py index 0070c62..3e4111d 100644 --- a/repeater/plugins/manifest.py +++ b/repeater/plugins/manifest.py @@ -18,6 +18,49 @@ SUPPORTED_SCHEMA = 1 SUPPORTED_RUNTIME_TYPES = frozenset({"python"}) SUPPORTED_UI_TYPES = frozenset({"application"}) MANIFEST_FILENAME = "openhop-plugin.json" +ZIP_METADATA_MAX_BYTES = 1024 * 1024 +ZIP_MEMBER_MAX_BYTES = 16 * 1024 * 1024 +ZIP_TOTAL_MAX_BYTES = 256 * 1024 * 1024 +ZIP_MAX_MEMBERS = 4096 + + +def validate_archive_limits(zf) -> None: + """Bound expanded wheel size before reading or handing it to pip.""" + members = zf.infolist() + if len(members) > ZIP_MAX_MEMBERS: + raise ManifestError("wheel exceeds archive member count limit") + if sum(member.file_size for member in members) > ZIP_TOTAL_MAX_BYTES: + raise ManifestError("wheel exceeds expanded archive size limit") + if any(member.file_size > ZIP_MEMBER_MAX_BYTES for member in members): + raise ManifestError("wheel member exceeds expanded size limit") + + +def read_archive_member(zf, member, *, metadata: bool = False) -> bytes: + """Check advertised and actual expanded lengths; never use an unbounded read.""" + limit = ZIP_METADATA_MAX_BYTES if metadata else ZIP_MEMBER_MAX_BYTES + info = zf.getinfo(member) if isinstance(member, str) else member + if info.file_size > limit: + raise ManifestError("wheel member exceeds expanded size limit") + with zf.open(info) as source: + data = source.read(limit + 1) + if len(data) > limit: + raise ManifestError("wheel member exceeds expanded size limit") + return data + + +def ui_subtree(entry: str) -> str: + """Return a dedicated public directory, never the release or runtime root.""" + _reject_path_traversal(entry, "ui.entry") + parts = entry.split("/") + if ( + len(parts) < 2 + or any(p in {"", ".", ".."} for p in parts) + or parts[0].startswith(".") + or ":" in entry + or parts[0].lower() in {"venv", "data", "logs", "releases", "current"} + ): + raise ManifestError("ui.entry must be inside a dedicated, non-reserved UI subtree") + return parts[0] class ManifestError(ValueError): @@ -139,8 +182,7 @@ def _validate_ui(raw: Any) -> UISpec: raise ManifestError("ui.entry is required") entry = entry.strip().replace("\\", "/") _reject_path_traversal(entry, "ui.entry") - if not entry: - raise ManifestError("ui.entry is required") + ui_subtree(entry) return UISpec(type=str(utype), entry=entry) @@ -234,6 +276,7 @@ def load_manifest_from_wheel(wheel_path: Path | str) -> PluginManifest: try: with zipfile.ZipFile(wheel_path) as zf: + validate_archive_limits(zf) candidates = [ name for name in zf.namelist() @@ -244,7 +287,7 @@ def load_manifest_from_wheel(wheel_path: Path | str) -> PluginManifest: # Prefer share/openhop/plugins/... layout, else first match preferred = [c for c in candidates if "share/openhop/plugins/" in c] chosen = preferred[0] if preferred else candidates[0] - raw = zf.read(chosen) + raw = read_archive_member(zf, chosen, metadata=True) except zipfile.BadZipFile as exc: raise ManifestError(f"invalid wheel archive: {exc}") from exc diff --git a/repeater/plugins/openhop-plugin-manager.service b/repeater/plugins/openhop-plugin-manager.service new file mode 100644 index 0000000..8549e6b --- /dev/null +++ b/repeater/plugins/openhop-plugin-manager.service @@ -0,0 +1,36 @@ +# Systemd service file template for openHop Plugin Manager. +# Install as /etc/systemd/system/openhop-plugin-manager.service +# +# Supervises installed plugins as child processes. Does not replace +# openhop-repeater.service. Repeater continues to work if this unit is stopped. + +[Unit] +Description=openHop Plugin Manager +After=network-online.target openhop-repeater.service +Wants=network-online.target +# Soft dependency: manager can start without Repeater, and vice versa + +[Service] +Type=simple +User=repeater +Group=repeater +WorkingDirectory=/var/lib/openhop_repeater + +# Disabled plugins are an intentional opt-out, not a failed service startup. +ExecCondition=/opt/openhop_repeater/venv/bin/python -I -c "import sys,yaml; c=yaml.safe_load(open(sys.argv[1])) or {}; p=c.get('plugins'); sys.exit(1 if isinstance(p, dict) and p.get('enabled') is False else 0)" /etc/openhop_repeater/config.yaml +ExecStart=/opt/openhop_repeater/venv/bin/openhop-plugin-manager --config /etc/openhop_repeater/config.yaml + +Restart=on-failure +RestartSec=5 +TimeoutStopSec=15 + +MemoryHigh=256M + +StandardOutput=journal +StandardError=journal +SyslogIdentifier=openhop-plugin-manager + +ReadWritePaths=/var/log/openhop_repeater /var/lib/openhop_repeater /etc/openhop_repeater + +[Install] +WantedBy=multi-user.target diff --git a/repeater/plugins/runtime.py b/repeater/plugins/runtime.py index d566842..6c33b51 100644 --- a/repeater/plugins/runtime.py +++ b/repeater/plugins/runtime.py @@ -10,18 +10,23 @@ import subprocess # nosec B404 - argument arrays only, never shell=True import sys import threading import time +import uuid from collections import deque from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Callable, Optional -from .manifest import PluginManifest -from .storage import PluginPaths, PluginStorage +from .manifest import PluginManifest, read_archive_member, ui_subtree, validate_archive_limits +from .storage import PluginPaths, PluginStorage, write_release_asset logger = logging.getLogger("PluginRuntime") STOP_TIMEOUT_SECONDS = 5.0 +VENV_TIMEOUT_SECONDS = 120.0 +PIP_TIMEOUT_SECONDS = 300.0 +PROCESS_KILL_TIMEOUT_SECONDS = 2.0 +INSTALL_OUTPUT_MAX_BYTES = 64 * 1024 CRASH_WINDOW_SECONDS = 60.0 CRASH_MAX_EXITS = 5 LOG_MAX_BYTES = 5 * 1024 * 1024 # rotate simply by truncating when oversized @@ -42,6 +47,8 @@ class ProcessHandle: process: subprocess.Popen log_fp: object started_at: float = field(default_factory=time.monotonic) + process_group: Optional[int] = None + log_thread: Optional[threading.Thread] = None class PluginRuntime: @@ -64,7 +71,7 @@ class PluginRuntime: self.crash_window = crash_window self.crash_max_exits = crash_max_exits self._popen = popen_factory or subprocess.Popen - self._run = run_factory or subprocess.run + self._run = run_factory or self._run_install_command self._lock = threading.RLock() self._handles: dict[str, ProcessHandle] = {} self._runtime_state: dict[str, PluginState] = {} @@ -184,9 +191,53 @@ class PluginRuntime: persisted_version[0], persisted_version[1], ) - shutil.rmtree(venv_path) - self._create_venv(venv_path) - self._pip_install(venv_path, wheels[0]) + # Build at the final absolute path: moving a freshly built venv would + # invalidate pip-generated shebangs. Keep the old environment for rollback. + backup = venv_path.with_name(f".venv-backup-{uuid.uuid4().hex}") + venv_path.rename(backup) + try: + self._create_venv(venv_path) + self._pip_install(venv_path, wheels[0]) + self.resolve_entrypoint(paths, version, manifest.runtime.entrypoint) + except BaseException: + if venv_path.exists(): + shutil.rmtree(venv_path) + backup.rename(venv_path) + raise + else: + shutil.rmtree(backup) + + def _run_install_command(self, cmd, *, timeout, **kwargs): + """Drain output into bounded memory and kill the entire installer group.""" + # Callers construct Python/venv/pip argv; installing trusted wheels is intentional. + proc = subprocess.Popen( # nosec B603 + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=(os.name == "posix"), + ) + handle = ProcessHandle( + "installer", proc, None, process_group=proc.pid if os.name == "posix" else None + ) + output = bytearray() + + def drain(): + with proc.stdout as source: + while chunk := source.read1(64 * 1024): + output.extend(chunk) + del output[:-INSTALL_OUTPUT_MAX_BYTES] + + reader = threading.Thread(target=drain, name="plugin-install-output", daemon=True) + reader.start() + try: + proc.wait(timeout=timeout) + finally: + self._signal_handle(handle, signal.SIGKILL) + proc.wait(timeout=PROCESS_KILL_TIMEOUT_SECONDS) + reader.join(timeout=PROCESS_KILL_TIMEOUT_SECONDS) + return subprocess.CompletedProcess( + cmd, proc.returncode, output.decode("utf-8", errors="replace"), "" + ) def _create_venv(self, venv_path: Path) -> None: if venv_path.exists(): @@ -197,6 +248,7 @@ class PluginRuntime: venv_path.parent.mkdir(parents=True, exist_ok=True) result = self._run( [self.python_executable, "-m", "venv", str(venv_path)], + timeout=VENV_TIMEOUT_SECONDS, check=False, capture_output=True, text=True, @@ -214,6 +266,7 @@ class PluginRuntime: try: result = self._run( [str(py), "-m", "pip", "install", "--upgrade", str(install_path)], + timeout=PIP_TIMEOUT_SECONDS, check=False, capture_output=True, text=True, @@ -251,9 +304,12 @@ class PluginRuntime: version = None try: with zipfile.ZipFile(wheel_path) as zf: + validate_archive_limits(zf) for member in zf.namelist(): if member.endswith(".dist-info/METADATA") and member.count("/") == 1: - text = zf.read(member).decode("utf-8", errors="replace") + text = read_archive_member(zf, member, metadata=True).decode( + "utf-8", errors="replace" + ) for line in text.splitlines(): if line.startswith("Name:"): name = line.split(":", 1)[1].strip() @@ -319,31 +375,18 @@ class PluginRuntime: import zipfile entry = manifest.ui.entry.replace("\\", "/") - # Copy any archive members under the entry's top-level directory (e.g. ui/) - top = entry.split("/", 1)[0] - prefixes = (f"{top}/",) + top = ui_subtree(entry) try: with zipfile.ZipFile(wheel_path) as zf: - for name in zf.namelist(): - # Also copy bare openhop-plugin.json already handled - rel = None - for prefix in prefixes: - # Match ".../ui/..." or "ui/..." - idx = name.find(prefix) - if idx >= 0: - rel = name[idx:] - break - if rel is None: + validate_archive_limits(zf) + for member in zf.infolist(): + parts = member.filename.split("/") + if member.is_dir() or top not in parts[:-1]: continue - if rel.endswith("/"): - continue - # Reject traversal in archive member names - if ".." in Path(rel).parts: - continue - dest = release_dir / rel - dest.parent.mkdir(parents=True, exist_ok=True) - with zf.open(name) as src, open(dest, "wb") as out: - out.write(src.read()) + if any(part in {"", ".", ".."} or "\\" in part for part in parts): + raise ValueError("unsafe UI archive path") + rel = "/".join(parts[parts.index(top) :]) + write_release_asset(release_dir, rel, read_archive_member(zf, member)) except zipfile.BadZipFile as exc: raise RuntimeError(f"invalid wheel while extracting UI: {exc}") from exc @@ -362,6 +405,7 @@ class PluginRuntime: try: with zipfile.ZipFile(wheel_path) as zf: + validate_archive_limits(zf) candidates = [ name for name in zf.namelist() @@ -373,9 +417,11 @@ class PluginRuntime: chosen = preferred[0] if preferred else candidates[0] if ".." in Path(chosen).parts: return - dest = release_dir / "config.default.json" - with zf.open(chosen) as src, open(dest, "wb") as out: - out.write(src.read()) + write_release_asset( + release_dir, + "config.default.json", + read_archive_member(zf, chosen, metadata=True), + ) except zipfile.BadZipFile: return @@ -500,7 +546,7 @@ class PluginRuntime: self._runtime_state[plugin_id] = PluginState.STARTING paths.logs_dir.mkdir(parents=True, exist_ok=True) self._rotate_log_if_needed(paths.log_file) - log_fp = open(paths.log_file, "a", encoding="utf-8", buffering=1) + log_fp = open(paths.log_file, "ab", buffering=0) env = os.environ.copy() # Never expose a stale manager-only credential left behind by an @@ -512,11 +558,12 @@ class PluginRuntime: try: proc = self._popen( [str(exe)], - stdout=log_fp, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, cwd=str(paths.data_dir), shell=False, + start_new_session=(os.name == "posix"), ) except Exception: log_fp.close() @@ -524,13 +571,41 @@ class PluginRuntime: raise self._handles[plugin_id] = ProcessHandle( - plugin_id=plugin_id, process=proc, log_fp=log_fp + plugin_id=plugin_id, + process=proc, + log_fp=log_fp, + process_group=proc.pid if os.name == "posix" else None, ) + handle = self._handles[plugin_id] + handle.log_thread = threading.Thread( + target=self._drain_plugin_log, + args=(handle,), + name=f"plugin-log-{plugin_id}", + daemon=True, + ) + handle.log_thread.start() self._runtime_state[plugin_id] = PluginState.RUNNING self._exit_codes[plugin_id] = None logger.info("Started plugin %s pid=%s", plugin_id, proc.pid) return PluginState.RUNNING + @staticmethod + def _drain_plugin_log(handle: ProcessHandle) -> None: + """Continuously cap captured stdout/stderr, including newline-free output.""" + source = getattr(handle.process, "stdout", None) + if source is None: + return + try: + with source: + while chunk := source.read1(64 * 1024): + chunk = chunk[-LOG_MAX_BYTES:] + if handle.log_fp.tell() + len(chunk) > LOG_MAX_BYTES: + handle.log_fp.seek(0) + handle.log_fp.truncate() + handle.log_fp.write(chunk) + except (OSError, ValueError): + logger.debug("Plugin log stream closed for %s", handle.plugin_id) + def stop(self, plugin_id: str, *, mark_disabled: bool = False) -> PluginState: with self._lock: handle = self._handles.get(plugin_id) @@ -543,37 +618,8 @@ class PluginRuntime: ) return self._runtime_state[plugin_id] - proc = handle.process - if proc.poll() is not None: - self._cleanup_handle(plugin_id) - self._runtime_state[plugin_id] = ( - PluginState.DISABLED if mark_disabled else PluginState.STOPPED - ) - return self._runtime_state[plugin_id] - - try: - proc.send_signal(signal.SIGTERM) - except ProcessLookupError: - pass - except Exception as exc: - logger.debug("SIGTERM failed for %s: %s", plugin_id, exc) - - # Wait outside lock - deadline = time.monotonic() + self.stop_timeout - while time.monotonic() < deadline: - if proc.poll() is not None: - break - time.sleep(0.05) - - if proc.poll() is None: - try: - proc.kill() - except ProcessLookupError: - pass - try: - proc.wait(timeout=2) - except Exception as exc: - logger.debug("Wait after killing %s failed: %s", plugin_id, exc) + # Keep the group identity captured at spawn even if the leader exited. + self._terminate_handle(handle) with self._lock: self._cleanup_handle(plugin_id) @@ -583,6 +629,43 @@ class PluginRuntime: logger.info("Stopped plugin %s", plugin_id) return self._runtime_state[plugin_id] + @staticmethod + def _signal_handle(handle: ProcessHandle, sig: int) -> None: + group = handle.process_group + if group is not None: + # Never signal the manager's own group or a non-positive group ID. + if group <= 1 or group == os.getpgrp(): + raise RuntimeError("refusing to signal unsafe plugin process group") + try: + os.killpg(group, sig) + except ProcessLookupError: + pass + elif handle.process.poll() is None: + if sig == signal.SIGKILL: + handle.process.kill() + else: + handle.process.send_signal(sig) + + def _terminate_handle(self, handle: ProcessHandle) -> None: + self._signal_handle(handle, signal.SIGTERM) + deadline = time.monotonic() + self.stop_timeout + while time.monotonic() < deadline: + handle.process.poll() # reap the leader, but do not equate it to the group + if handle.process_group is None: + if handle.process.poll() is not None: + break + else: + try: + os.killpg(handle.process_group, 0) + except ProcessLookupError: + break + time.sleep(0.05) + self._signal_handle(handle, signal.SIGKILL) + try: + handle.process.wait(timeout=PROCESS_KILL_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + logger.warning("Plugin %s did not reap after SIGKILL", handle.plugin_id) + def restart(self, plugin_id: str) -> PluginState: self.stop(plugin_id) return self.start(plugin_id) @@ -591,11 +674,15 @@ class PluginRuntime: handle = self._handles.pop(plugin_id, None) if handle is None: return + # Crash supervision also owns descendants left behind by an exited leader. + self._signal_handle(handle, signal.SIGKILL) try: if handle.process.poll() is not None: self._exit_codes[plugin_id] = handle.process.returncode except Exception as exc: logger.debug("Could not record exit status for %s: %s", plugin_id, exc) + if handle.log_thread is not None: + handle.log_thread.join(timeout=PROCESS_KILL_TIMEOUT_SECONDS) try: handle.log_fp.close() except Exception as exc: diff --git a/repeater/plugins/storage.py b/repeater/plugins/storage.py index 0cbec22..56c9fe2 100644 --- a/repeater/plugins/storage.py +++ b/repeater/plugins/storage.py @@ -6,7 +6,9 @@ import json import logging import os import shutil +import stat import tempfile +import uuid from dataclasses import dataclass from pathlib import Path from typing import Any, Optional @@ -18,6 +20,7 @@ logger = logging.getLogger("PluginStorage") STATE_FILENAME = "state.json" DEFAULT_PLUGINS_SUBDIR = "plugins" DEFAULT_SOCKET_NAME = "plugin-manager.sock" +LOG_TAIL_MAX_BYTES = 256 * 1024 @dataclass(frozen=True) @@ -82,6 +85,49 @@ def safe_join(root: Path, *parts: str) -> Path: return candidate +def write_release_asset(root: Path, relative: str, data: bytes) -> None: + """Atomically write an asset without following directory or leaf symlinks. + + Walk from the filesystem root with pinned directory descriptors, so a + concurrent symlink substitution cannot redirect extraction. Replacing the + leaf also avoids overwriting another file through an existing hard link. + """ + parts = relative.split("/") + if any(p in {"", ".", ".."} or "\\" in p for p in parts): + raise ValueError("unsafe asset path") + absolute = root.absolute() + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW + fd = os.open(absolute.anchor, flags) + temporary = None + try: + for component in (*absolute.parts[1:], *parts[:-1]): + try: + os.mkdir(component, dir_fd=fd) + except FileExistsError: + pass + child = os.open(component, flags, dir_fd=fd) + os.close(fd) + fd = child + try: + mode = os.stat(parts[-1], dir_fd=fd, follow_symlinks=False).st_mode + if not stat.S_ISREG(mode): + raise ValueError("asset destination must be a regular file, not a symlink") + except FileNotFoundError: + pass + temporary = f".asset-{uuid.uuid4().hex}" + out_fd = os.open( + temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o644, dir_fd=fd + ) + with os.fdopen(out_fd, "wb") as out: + out.write(data) + os.replace(temporary, parts[-1], src_dir_fd=fd, dst_dir_fd=fd) + temporary = None + finally: + if temporary is not None: + os.unlink(temporary, dir_fd=fd) + os.close(fd) + + class PluginStorage: """Manages plugin directories and manager-owned state.json files.""" @@ -276,8 +322,10 @@ class PluginStorage: return [] lines = max(1, min(int(lines), 5000)) try: - # Simple tail: read whole file if small; otherwise scan from end - data = log_path.read_text(encoding="utf-8", errors="replace") + with log_path.open("rb") as source: + source.seek(0, os.SEEK_END) + source.seek(max(0, source.tell() - LOG_TAIL_MAX_BYTES)) + data = source.read(LOG_TAIL_MAX_BYTES).decode("utf-8", errors="replace") except OSError: return [] all_lines = data.splitlines() diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index c8f4dc5..db24e3b 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -673,7 +673,10 @@ class APIEndpoints: has_default_name = node_name in ["mesh-repeater-01", ""] admin_password = config.get("repeater", {}).get("security", {}).get("admin_password", "") - has_default_password = admin_password in ["admin123", ""] + has_default_password = admin_password is None or ( + isinstance(admin_password, str) + and (not admin_password.strip() or admin_password == "admin123") # nosec B105 + ) radio_type_raw = config.get("radio_type") radio_type = "" if radio_type_raw is None else str(radio_type_raw).lower().strip() @@ -684,7 +687,12 @@ class APIEndpoints: "default_password": has_default_password, "radio_not_configured": radio_not_configured, } - return has_default_name or has_default_password or radio_not_configured, reasons + # Name and radio settings are diagnostics, not authorization signals: + # radio-less installations and default names are valid after provisioning. + # Legacy configs have no marker, so a custom password also closes setup. + # An explicit false marker must never override existing credentials. + setup_completed = config.get("setup_completed") is True + return has_default_password and not setup_completed, reasons def _default_policy_document(self) -> dict: return { @@ -1738,8 +1746,10 @@ class APIEndpoints: ) # Explicit setup writes canonical names only. config_yaml = normalize_modem_config(config_yaml) + config_yaml["setup_completed"] = True with open(self._config_path, "w") as f: yaml.dump(config_yaml, f, default_flow_style=False, sort_keys=False) + self.config["setup_completed"] = True logger.info( f"Setup wizard completed: node_name={node_name}, hardware={hardware_key}, freq={freq_mhz}MHz" @@ -8321,6 +8331,13 @@ class APIEndpoints: if not updated_sections: return self._error("No valid configuration sections found in import") + # Keep authenticated restores closed even when restoring defaults. + # Partial first-run restores must still allow the existing wizard; + # finish provisioning once credentials have actually been supplied. + # setup_completed is deliberately not an importable section. + if request_user or not self._setup_status_from_config(self.config)[0]: + self.config["setup_completed"] = True + # Persist and live-reload self.config_manager.update_and_save( updates={}, # Already applied above diff --git a/repeater/web/http_server.py b/repeater/web/http_server.py index 565d718..93c8e6e 100644 --- a/repeater/web/http_server.py +++ b/repeater/web/http_server.py @@ -294,6 +294,7 @@ class StatsApp: def _serve_plugin_ui(self, plugin_id: str, relative_parts: tuple[str, ...]): """Serve static assets for an enabled application UI plugin.""" + from repeater.plugins.manifest import ui_subtree from repeater.plugins.storage import PluginStorage, resolve_plugins_root, safe_join try: @@ -317,19 +318,15 @@ class StatsApp: raise cherrypy.NotFound() entry = manifest.ui.entry.replace("\\", "/") - entry_parts = tuple(p for p in entry.split("/") if p) - if not entry_parts: - raise cherrypy.NotFound() + ui_subtree(entry) + entry_parts = tuple(entry.split("/")) - # Document root is the directory containing the entry file so a - # relative Vite base (./assets/...) works at /plugins/{id}/. - # Example: entry ui/index.html → doc_root = release/ui - if len(entry_parts) > 1: - doc_root = safe_join(release_root, *entry_parts[:-1]) - entry_name = entry_parts[-1] - else: - doc_root = release_root - entry_name = entry_parts[0] + # Check the resolved subtree too: an old release may contain a UI + # symlink pointing at a reserved directory or at the release root. + doc_root = safe_join(release_root, *entry_parts[:-1]) + public_parts = doc_root.relative_to(release_root.resolve()).parts + entry_name = entry_parts[-1] + ui_subtree("/".join((*public_parts, entry_name))) if not relative_parts: serve_parts = (entry_name,) diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index 1d42f2a..cba364a 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -4581,6 +4581,8 @@ paths: - BearerAuth: [] - ApiKeyAuth: [] responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Plugin list content: @@ -4614,6 +4616,8 @@ paths: type: boolean description: Force refresh of the catalogue cache responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Catalogue entries '502': @@ -4642,6 +4646,8 @@ paths: type: string description: Optional assertion of the currently approved version responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Installed plugin status '400': @@ -4665,6 +4671,8 @@ paths: schema: type: boolean responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Update check result '404': @@ -4687,6 +4695,8 @@ paths: version: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Updated plugin status '400': @@ -4706,6 +4716,8 @@ paths: schema: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Config object '404': @@ -4738,6 +4750,8 @@ paths: type: boolean default: false responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Saved '400': @@ -4760,6 +4774,8 @@ paths: schema: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Runtime object '404': @@ -4796,6 +4812,8 @@ paths: type: string description: Absolute path to a .whl on the Repeater host responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Installed '400': @@ -4822,6 +4840,8 @@ paths: type: string example: openhop.nomad responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Enabled '404': @@ -4847,6 +4867,8 @@ paths: id: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Disabled '404': @@ -4872,6 +4894,8 @@ paths: id: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Started '409': @@ -4897,6 +4921,8 @@ paths: id: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Stopped '503': @@ -4920,6 +4946,8 @@ paths: id: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Restarted '503': @@ -4945,6 +4973,8 @@ paths: type: integer default: 200 responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Log lines content: @@ -4985,6 +5015,8 @@ paths: type: boolean default: false responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Uninstalled '503': @@ -5004,6 +5036,8 @@ paths: schema: type: string responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Status content: @@ -5038,6 +5072,8 @@ paths: type: boolean default: false responses: + '504': + $ref: '#/components/responses/PluginOutcomeUnknown' '200': description: Uninstalled '503': @@ -5045,6 +5081,27 @@ paths: components: + responses: + PluginOutcomeUnknown: + description: | + The manager response was lost or its completion deadline expired. + This is not cancellation: the operation may still complete. Check plugin + status before retrying. A busy manager instead returns 503; conflicting + lifecycle operations on the same plugin return 409. + content: + application/json: + schema: + type: object + required: [success, error, outcome] + properties: + success: + type: boolean + enum: [false] + error: + type: string + outcome: + type: string + enum: [unknown] schemas: PluginStatus: diff --git a/repeater/web/plugin_endpoints.py b/repeater/web/plugin_endpoints.py index 36d585e..90613be 100644 --- a/repeater/web/plugin_endpoints.py +++ b/repeater/web/plugin_endpoints.py @@ -12,7 +12,12 @@ from typing import Any, Optional import cherrypy from repeater.config import resolve_storage_dir -from repeater.plugins.ipc import PluginIPCClient, PluginIPCError, PluginManagerUnavailable +from repeater.plugins.ipc import ( + PluginIPCClient, + PluginIPCError, + PluginIPCOutcomeUnknown, + PluginManagerUnavailable, +) from repeater.plugins.storage import resolve_plugin_socket_path, resolve_plugins_root logger = logging.getLogger("HTTPServer") @@ -52,6 +57,8 @@ class PluginAPIEndpoints: def _handle_ipc(self, fn): try: return self._ok(fn()) + except PluginIPCOutcomeUnknown as exc: + return {**self._err(str(exc), 504), "outcome": "unknown"} except PluginManagerUnavailable as exc: return self._err(str(exc), 503) except PluginIPCError as exc: @@ -209,6 +216,13 @@ class PluginAPIEndpoints: try: return {"plugin": client.install(str(wheel_path.resolve()))} + except PluginIPCOutcomeUnknown as exc: + # A legacy manager can still consume this path after disconnect. + # Only the new processing acknowledgement proves it owns a copy. + if not exc.upload_safe: + cleanup = False + logger.warning("Retaining unacknowledged plugin upload after lost response") + raise finally: if cleanup and wheel_path is not None: # Remove staging dir (preferred) or lone temp file diff --git a/repeater/web/update_endpoints.py b/repeater/web/update_endpoints.py index 5a4cafe..9cc031a 100644 --- a/repeater/web/update_endpoints.py +++ b/repeater/web/update_endpoints.py @@ -915,50 +915,59 @@ def _disable_legacy_services() -> None: subprocess.run([_SYSTEMCTL_BIN, "daemon-reload"], check=False) # nosec B603 -def _ensure_plugin_manager_service() -> None: - """Install and enable the plugin-manager service unit when present.""" - service_unit = "/etc/systemd/system/openhop-plugin-manager.service" - candidates = [ - "/root/openhop-repeater/openhop-plugin-manager.service", - "/opt/openhop_repeater/openhop-plugin-manager.service", - "/opt/openhop_repeater/openhop-repeater/openhop-plugin-manager.service", - "/root/scripts/openhop-plugin-manager.service", - "/opt/openhop_repeater/openhop-repeater/openhop-plugin-manager.service", - ] +def _native_upgrade_environment() -> dict: + """Do not forward Python, pip, git or shell hooks into privileged tools.""" + return { + "PATH": "/usr/sbin:/usr/bin:/sbin:/bin", + "HOME": "/root", + "LANG": "C.UTF-8", + "PIP_CONFIG_FILE": "/dev/null", + } - source = next((path for path in candidates if os.path.isfile(path)), None) - if source and not os.path.isfile(service_unit): - _state.append_line(f"[pyMC updater] Installing plugin-manager service from {source}") - try: - os.makedirs(os.path.dirname(service_unit), exist_ok=True) - shutil.copy2(source, service_unit) - except OSError as exc: - logger.warning(f"[Update] Could not install plugin-manager unit: {exc}") - return - if not os.path.isfile(service_unit): - return - - _state.append_line("[pyMC updater] Ensuring openhop-plugin-manager service is enabled") - for cmd in ( - [_SYSTEMCTL_BIN, "daemon-reload"], - [_SYSTEMCTL_BIN, "enable", "openhop-plugin-manager"], - ): - subprocess.run(cmd, check=False, capture_output=True, text=True) # nosec B603 - - restart = subprocess.run( - [_SYSTEMCTL_BIN, "restart", "openhop-plugin-manager"], - check=False, - capture_output=True, - text=True, - ) # nosec B603 - if restart.returncode != 0: - subprocess.run( - [_SYSTEMCTL_BIN, "start", "openhop-plugin-manager"], +def _ensure_plugin_manager_service() -> bool: + """Root-only fallback; the sudo wrapper provisions normal native OTA.""" + if os.geteuid() != 0 or is_buildroot() or is_container(): + return True + # Resolve with a fresh isolated interpreter: this HTTP process still has + # the old package imported, and source checkouts need not exist on OTA hosts. + code = """ +from importlib.metadata import distribution +from pathlib import Path +import subprocess +import sys +unit = Path(distribution('openhop_repeater').locate_file( + 'repeater/plugins/openhop-plugin-manager.service')) +if not unit.is_file(): + print('Selected package has no plugin-manager unit; older channels may not support plugins. ' + 'Re-run manage.sh upgrade from a plugin-capable release if unexpected.', file=sys.stderr) + sys.exit(0) +unit = unit.resolve(strict=True) +for path in (unit, *unit.parents): + info = path.stat() + if info.st_uid != 0 or info.st_mode & 0o022: + raise PermissionError('Untrusted packaged service path') +subprocess.run(['/usr/bin/install', '-o', 'root', '-g', 'root', '-m', '0644', + str(unit), '/etc/systemd/system/openhop-plugin-manager.service'], check=True) +for action in ('daemon-reload', 'enable', 'restart'): + command = ['/bin/systemctl', action] + if action != 'daemon-reload': + command.append('openhop-plugin-manager') + subprocess.run(command, check=True) +""" + try: + result = subprocess.run( + ["/opt/openhop_repeater/venv/bin/python", "-I", "-c", code], + cwd="/", + env=_native_upgrade_environment(), check=False, capture_output=True, text=True, ) # nosec B603 + return result.returncode == 0 + except OSError as exc: + logger.warning(f"[Update] Could not provision plugin-manager unit: {exc}") + return False def _do_install() -> None: @@ -975,6 +984,7 @@ def _do_install() -> None: text=True, bufsize=1, env=env, + cwd="/", ) # nosec B603 if proc.stdout is not None: for line in proc.stdout: @@ -997,10 +1007,9 @@ def _do_install() -> None: # did not match its code — and when that value was stale (a failed check, or # a channel switch) pip saw the requirement as already satisfied and skipped # the install, pinning the node to a version it could no longer leave. - env = _os.environ.copy() + env = _os.environ.copy() if is_buildroot() else _native_upgrade_environment() _VENV_DIR = "/opt/openhop_repeater/venv" - _VENV_PIP = os.path.join(_VENV_DIR, "bin", "pip") _VENV_PYTHON = os.path.join(_VENV_DIR, "bin", "python") _state.append_line(f"[pyMC updater] Installing from channel '{channel}'…") @@ -1030,14 +1039,32 @@ def _do_install() -> None: # Ensure venv exists (migration from system-pip era) if not os.path.isfile(_VENV_PYTHON): _state.append_line("[pyMC updater] Creating venv (first-time migration)…") - _run(["/usr/bin/python3", "-m", "venv", "--system-site-packages", _VENV_DIR], env=env) - _run([_VENV_PIP, "install", "--upgrade", "pip", "setuptools", "wheel"], env=env) + _run( + ["/usr/bin/python3", "-I", "-m", "venv", "--system-site-packages", _VENV_DIR], + env=env, + ) + _run( + [ + _VENV_PYTHON, + "-I", + "-m", + "pip", + "install", + "--upgrade", + "pip", + "setuptools", + "wheel", + ], + env=env, + ) # Clean up system-level packages to avoid shadowing - _run(["/usr/bin/python3", "-m", "pip", "uninstall", "-y", "openhop_repeater"], env=env) - _run(["/usr/bin/python3", "-m", "pip", "uninstall", "-y", "openhop_core"], env=env) - _run(["/usr/bin/python3", "-m", "pip", "uninstall", "-y", "pymc_repeater"], env=env) - _run(["/usr/bin/python3", "-m", "pip", "uninstall", "-y", "pymc_core"], env=env) + _run( + ["/usr/bin/python3", "-I", "-m", "pip", "uninstall", "-y", "openhop_repeater"], env=env + ) + _run(["/usr/bin/python3", "-I", "-m", "pip", "uninstall", "-y", "openhop_core"], env=env) + _run(["/usr/bin/python3", "-I", "-m", "pip", "uninstall", "-y", "pymc_repeater"], env=env) + _run(["/usr/bin/python3", "-I", "-m", "pip", "uninstall", "-y", "pymc_core"], env=env) # Remove stale source trees that could shadow the venv package _cleanup_stale_source_trees() @@ -1046,7 +1073,10 @@ def _do_install() -> None: _state.append_line("[pyMC updater] Running as root – venv pip install") _state.append_line(f"[pyMC updater] Target: {install_spec}") cmd = [ - _VENV_PIP, + _VENV_PYTHON, + "-I", + "-m", + "pip", "install", "--upgrade", "--no-cache-dir", @@ -1054,6 +1084,11 @@ def _do_install() -> None: ] elif _os.path.isfile(_UPGRADE_WRAPPER): _state.append_line(f"[pyMC updater] Using sudo wrapper: {_UPGRADE_WRAPPER}") + _state.append_line( + "[pyMC updater] Native OTA security/service provisioning requires the current " + "root-owned helper. Existing installations must run manage.sh upgrade once; " + "a wheel update cannot replace the privileged helper." + ) # The wrapper handles venv creation/migration internally # Second argument (pretend-version) deliberately omitted — see above. cmd = [_SUDO_BIN, _UPGRADE_WRAPPER, channel] @@ -1069,7 +1104,10 @@ def _do_install() -> None: if success: _cleanup_stale_dist_info() - _ensure_plugin_manager_service() + if not _ensure_plugin_manager_service(): + _state.append_line( + "[warning] Package updated, but optional plugin-manager service provisioning failed; inspect its service status or re-run manage.sh upgrade." + ) _state.append_line("[pyMC updater] Restarting service in 3 seconds…") time.sleep(3) restart_ok = False diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index 656a0bd..5080903 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/tests/test_api_endpoints_core_coverage.py @@ -762,6 +762,8 @@ def test_config_import_rejects_missing_config_object(cherrypy_ctx): def test_config_import_updates_sections_and_preserves_redacted(cherrypy_ctx): request, _ = cherrypy_ctx request.method = "POST" + # A custom administrator password closes anonymous first-run import. + request.user = {"username": "admin", "auth_type": "jwt"} api = _make_api( { diff --git a/tests/test_api_endpoints_setup.py b/tests/test_api_endpoints_setup.py index 28a7264..494c332 100644 --- a/tests/test_api_endpoints_setup.py +++ b/tests/test_api_endpoints_setup.py @@ -7,7 +7,7 @@ def _make_api(config): return api -def test_needs_setup_triggers_when_radio_type_missing(): +def test_needs_setup_stays_closed_when_radio_type_missing(): api = _make_api( { "repeater": { @@ -19,7 +19,7 @@ def test_needs_setup_triggers_when_radio_type_missing(): result = api.needs_setup() - assert result["needs_setup"] is True + assert result["needs_setup"] is False assert result["reasons"]["radio_not_configured"] is True assert result["reasons"]["default_name"] is False assert result["reasons"]["default_password"] is False diff --git a/tests/test_native_updater_security.py b/tests/test_native_updater_security.py new file mode 100644 index 0000000..3ec6195 --- /dev/null +++ b/tests/test_native_updater_security.py @@ -0,0 +1,189 @@ +"""No updater execution: inspect templates and run only benign Python lookup probes.""" + +import os +import re +import shlex +import shutil +import subprocess +import sys +from pathlib import Path +from zipfile import ZipFile + +import pytest + +from repeater.web import update_endpoints as ue + +ROOT = Path(__file__).resolve().parents[1] + + +def wrappers(): + return re.findall( + r"<<'UPGRADEEOF'\n(.*?)\nUPGRADEEOF", (ROOT / "manage.sh").read_text(), re.DOTALL + ) + + +@pytest.mark.parametrize("wrapper", wrappers(), ids=["install", "upgrade"]) +def test_wrapper_isolates_every_python_invocation(wrapper, tmp_path): + commands = [ + line.strip() + for line in wrapper.splitlines() + if re.match(r'(python3|/usr/bin/python3|"\$VENV_PYTHON") ', line.strip()) + ] + assert commands + for line in commands: + args = shlex.split(line.rstrip("\\")) + assert args[1] == "-I", line + # Reproduce module lookup only, never run pip or the updater. + (tmp_path / "pip.py").write_text("raise AssertionError('must not import caller module')\n") + env = dict(os.environ, PYTHONPATH=str(tmp_path), PYTHONHOME=str(tmp_path)) + probe = subprocess.run( + [ + sys.executable, + "-I", + "-c", + "import importlib.util; spec = importlib.util.find_spec('pip'); " + "print(spec.origin if spec else 'pip-not-installed')", + ], + cwd=tmp_path, + env=env, + capture_output=True, + text=True, + check=True, + ) + assert str(tmp_path) not in probe.stdout + + +@pytest.mark.parametrize("wrapper", wrappers(), ids=["install", "upgrade"]) +def test_wrapper_cleans_environment_and_requires_trusted_paths(wrapper): + assert wrapper.startswith("#!/bin/bash -p\n") + assert "/usr/bin/env -i" in wrapper + assert "cd /" in wrapper + assert "require_root_owned" in wrapper + assert "-perm /022" in wrapper + assert '[[ "$#" -le 2 ]]' in wrapper + + +@pytest.mark.parametrize("wrapper", wrappers(), ids=["install", "upgrade"]) +def test_wrapper_provisions_packaged_service_after_success(wrapper): + install = wrapper.index("# ---- Install openhop_repeater from git ----") + provision = wrapper.index("plugins/openhop-plugin-manager.service", install) + assert provision > install + assert "install -o root -g root -m 0644" in wrapper[install:] + assert "systemctl enable openhop-plugin-manager" in wrapper[install:] + assert "systemctl restart openhop-plugin-manager" in wrapper[install:] + assert 'if [[ -f "$PLUGIN_UNIT" ]]; then' in wrapper[provision:] + assert "does not ship a plugin-manager unit" in wrapper[provision:] + + +def test_built_wheel_contains_service_without_checkout(tmp_path): + # Build locally with installed build tools only: no pip, network or install. + project = tmp_path / "project" + project.mkdir() + for name in ("pyproject.toml", "setup.py", "README.md", "LICENSE"): + shutil.copy2(ROOT / name, project / name) + shutil.copytree( + ROOT / "repeater", project / "repeater", ignore=shutil.ignore_patterns("__pycache__") + ) + result = subprocess.run( + [ + sys.executable, + "-c", + "from setuptools.build_meta import build_wheel; build_wheel('dist')", + ], + cwd=project, + env=dict(os.environ, SETUPTOOLS_SCM_PRETEND_VERSION_FOR_OPENHOP_REPEATER="1.2.3"), + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + wheel = next((project / "dist").glob("*.whl")) + with ZipFile(wheel) as archive: + assert ( + archive.read("repeater/plugins/openhop-plugin-manager.service") + == (ROOT / "openhop-plugin-manager.service").read_bytes() + ) + + +def test_package_delivers_identical_service_template(): + resource = ROOT / "repeater/plugins/openhop-plugin-manager.service" + assert resource.is_file() + assert resource.read_bytes() == (ROOT / "openhop-plugin-manager.service").read_bytes() + assert '"plugins/*.service"' in (ROOT / "pyproject.toml").read_text() + + +def test_unprivileged_helper_never_provisions_services(monkeypatch): + monkeypatch.setattr(ue.os, "geteuid", lambda: 1000) + monkeypatch.setattr(ue.os.path, "isfile", lambda path: True) + + def forbidden(*args, **kwargs): + pytest.fail("Unprivileged HTTP helper must not install or manage system units") + + monkeypatch.setattr(ue.shutil, "copy2", forbidden) + monkeypatch.setattr(ue.subprocess, "run", forbidden) + ue._ensure_plugin_manager_service() + + +def test_root_helper_reads_service_from_new_distribution(monkeypatch): + monkeypatch.setattr(ue.os, "geteuid", lambda: 0) + monkeypatch.setattr(ue, "is_container", lambda: False) + monkeypatch.setattr(ue, "is_buildroot", lambda: False) + calls = [] + + def run(cmd, **kwargs): + calls.append((cmd, kwargs)) + return type("Result", (), {"returncode": 0})() + + monkeypatch.setattr(ue.subprocess, "run", run) + assert ue._ensure_plugin_manager_service() is True + cmd, kwargs = calls[0] + assert cmd[:3] == ["/opt/openhop_repeater/venv/bin/python", "-I", "-c"] + assert "plugins/openhop-plugin-manager.service" in cmd[3] + assert kwargs["cwd"] == "/" + assert kwargs["env"]["HOME"] == "/root" + + +def test_root_helper_reports_service_provision_failure(monkeypatch): + monkeypatch.setattr(ue.os, "geteuid", lambda: 0) + monkeypatch.setattr(ue, "is_container", lambda: False) + monkeypatch.setattr(ue, "is_buildroot", lambda: False) + monkeypatch.setattr( + ue.subprocess, "run", lambda *a, **kw: type("Result", (), {"returncode": 1})() + ) + assert ue._ensure_plugin_manager_service() is False + + +def test_wrapper_templates_are_identical(): + assert len(wrappers()) == 2 + assert wrappers()[0] == wrappers()[1] + + +@pytest.mark.parametrize("enabled, expected", [(False, 1), (True, 0), (None, 0)]) +def test_service_condition_respects_plugin_opt_out(tmp_path, enabled, expected): + unit = (ROOT / "openhop-plugin-manager.service").read_text() + condition = next( + (line for line in unit.splitlines() if line.startswith("ExecCondition=")), None + ) + assert condition is not None, "disabled plugins must not become a failed service" + args = shlex.split(condition.split("=", 1)[1]) + config = tmp_path / "synthetic.yaml" + config.write_text( + "plugins:\n enabled: " + {False: "false", True: "true", None: "null"}[enabled] + ) + # Only run the condition against synthetic data, not systemd or the manager. + result = subprocess.run( + [sys.executable, *args[1:-1], str(config)], capture_output=True, check=False + ) + assert result.returncode == expected, result.stderr + + +def test_trusted_path_guard_accepts_system_python(): + wrapper = wrappers()[0] + guard = wrapper[wrapper.index("require_root_owned() {") : wrapper.index("\nfor path in")] + # Exercise just the read-only path guard, never the wrapper body. + result = subprocess.run( + ["/bin/bash", "-c", guard + "\nrequire_root_owned /usr/bin/python3"], + capture_output=True, + check=False, + ) + assert result.returncode == 0, result.stderr diff --git a/tests/test_plugin_api_endpoints.py b/tests/test_plugin_api_endpoints.py index b86a789..03cb4b9 100644 --- a/tests/test_plugin_api_endpoints.py +++ b/tests/test_plugin_api_endpoints.py @@ -78,6 +78,54 @@ def test_list_and_status_via_client_mock(tmp_path, cherrypy_request): assert runtime["runtime"]["schema"] == 1 +@pytest.mark.parametrize("upload_safe", [False, True]) +def test_upload_timeout_reports_unknown_and_preserves_unacknowledged_file( + tmp_path, cherrypy_request, upload_safe +): + from io import BytesIO + from pathlib import Path + from types import SimpleNamespace + from repeater.plugins.ipc import PluginIPCOutcomeUnknown + + api = PluginAPIEndpoints({"storage": {"storage_dir": str(tmp_path)}}) + uploaded = [] + + def install(path): + uploaded.append(Path(path)) + raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) + + client = SimpleNamespace(install=install) + cherrypy.request.method = "POST" + wheel = SimpleNamespace(filename="demo-1-py3-none-any.whl", file=BytesIO(b"wheel")) + with patch.object(api, "_client_or_raise", return_value=client): + result = api.install(wheel=wheel) + assert cherrypy.response.status == 504 + assert result["outcome"] == "unknown" + assert "may still complete" in result["error"] + assert uploaded[0].exists() is (not upload_safe) + + +def test_upload_completed_install_preserves_sync_contract(tmp_path, cherrypy_request): + from io import BytesIO + from pathlib import Path + from types import SimpleNamespace + + api = PluginAPIEndpoints({"storage": {"storage_dir": str(tmp_path)}}) + uploaded = [] + + def install(path): + uploaded.append(Path(path)) + assert uploaded[0].read_bytes() == b"wheel" + return {"id": "demo", "version": "1"} + + cherrypy.request.method = "POST" + wheel = SimpleNamespace(filename="demo-1-py3-none-any.whl", file=BytesIO(b"wheel")) + with patch.object(api, "_client_or_raise", return_value=SimpleNamespace(install=install)): + result = api.install(wheel=wheel) + assert result == {"success": True, "plugin": {"id": "demo", "version": "1"}} + assert not uploaded[0].exists() + + def test_ui_serving_index_assets_spa_and_disabled(tmp_path): plugins_root = tmp_path / "plugins" storage = PluginStorage(plugins_root) diff --git a/tests/test_plugin_download_boundaries.py b/tests/test_plugin_download_boundaries.py new file mode 100644 index 0000000..a6fd307 --- /dev/null +++ b/tests/test_plugin_download_boundaries.py @@ -0,0 +1,173 @@ +"""Offline resource-boundary regressions for both plugin download paths.""" + +import hashlib +import http.client +import io +import socket +import threading +import time +from contextlib import contextmanager + +import pytest + +from repeater.plugins import catalogue, github_releases + +WHEEL = b"test wheel" +URL = "https://github.com/org/repo/releases/download/v1.0/pkg-1.0-py3-none-any.whl" + + +def operation(kind, opener, tmp_path): + if kind.startswith("catalogue"): + client = catalogue.CatalogueClient(opener=opener) + if kind.endswith("json"): + return client.fetch() + plugin = catalogue.CataloguePlugin( + "test", + "Test", + "", + "org/repo", + version="1.0", + wheel_url=URL, + sha256=hashlib.sha256(WHEEL).hexdigest(), + ) + return client.download_wheel(plugin, tmp_path) + client = github_releases.GitHubReleaseClient(opener=opener) + if kind.endswith("json"): + return client.list_releases("org/repo") + release = client._parse_release( + { + "tag_name": "v1.0", + "assets": [{"name": "pkg-1.0-py3-none-any.whl", "browser_download_url": URL}], + } + ) + return client.download_wheel(release, tmp_path) + + +KINDS = ["catalogue-json", "releases-json", "catalogue-wheel", "releases-wheel"] + + +@pytest.mark.parametrize("kind", KINDS) +def test_truncated_content_length_is_rejected(kind, tmp_path): + payload = ( + b'{"schema": 1, "plugins": []}' + if kind == "catalogue-json" + else b"[]" + if kind == "releases-json" + else WHEEL + ) + reader, writer = socket.socketpair() + try: + writer.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n" + payload) + writer.shutdown(socket.SHUT_WR) + response = http.client.HTTPResponse(reader) + response.begin() + + def opener(request, timeout): + response.url = request.full_url + return response + + error = ( + catalogue.CatalogueError + if kind.startswith("catalogue") + else github_releases.GitHubReleasesError + ) + with pytest.raises(error): + operation(kind, opener, tmp_path) + assert response.closed + assert list(tmp_path.iterdir()) == [] + finally: + reader.close() + writer.close() + + +@pytest.mark.parametrize( + "url", + [ + "http://example.test/catalogue.json", + "file:///tmp/catalogue.json", + "ftp://example.test/catalogue.json", + ], +) +def test_initial_catalogue_url_requires_https(url): + with pytest.raises(catalogue.CatalogueError, match="HTTPS"): + catalogue.CatalogueClient(url=url) + + +def test_custom_https_catalogue_url_is_preserved(): + url = "https://custom.example.test:8443/custom.json?channel=beta" + assert catalogue.CatalogueClient(url=url).url == url + + +@pytest.mark.parametrize("kind", KINDS) +def test_response_size_is_bounded_without_content_length(kind, monkeypatch, tmp_path): + module = catalogue if kind.startswith("catalogue") else github_releases + cap = "MAX_JSON_BYTES" if kind.endswith("json") else "MAX_WHEEL_BYTES" + monkeypatch.setattr(module, cap, 128, raising=False) + response = io.BytesIO(b" " * 129 if kind.endswith("json") else WHEEL * 20) + error = catalogue.CatalogueError if module is catalogue else github_releases.GitHubReleasesError + with pytest.raises(error, match="exceeds"): + operation(kind, lambda *a, **kw: response, tmp_path) + assert response.closed + assert list(tmp_path.iterdir()) == [] + + +@contextmanager +def slow_response(mode): + """Real stdlib HTTPResponse over local sockets, never the public network.""" + reader, writer = socket.socketpair() + stop = threading.Event() + + def serve(): + try: + if mode == "chunk-header": + writer.sendall(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n1;") + else: + writer.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n") + for _ in range(30): + if stop.wait(0.02): + return + if mode != "idle": + writer.sendall(b" ") + except OSError: + pass + finally: + writer.close() + + thread = threading.Thread(target=serve, daemon=True) + thread.start() + response = http.client.HTTPResponse(reader) + response.begin() + + def opener(request, timeout): + response.url = request.full_url + reader.settimeout(timeout) + return response + + try: + yield opener, response + finally: + stop.set() + reader.close() + thread.join(timeout=2) + response.close() + + +@pytest.mark.parametrize("kind", KINDS) +@pytest.mark.parametrize("mode", ["trickle", "idle", "chunk-header"]) +def test_total_body_deadline_interrupts_real_http_response(kind, mode, monkeypatch, tmp_path): + module = catalogue if kind.startswith("catalogue") else github_releases + setting = ( + "JSON_DOWNLOAD_TIMEOUT_SECONDS" + if kind.endswith("json") + else "WHEEL_DOWNLOAD_TIMEOUT_SECONDS" + ) + monkeypatch.setattr(module, setting, 0.15, raising=False) + error = catalogue.CatalogueError if module is catalogue else github_releases.GitHubReleasesError + with slow_response(mode) as (opener, response): + started = time.monotonic() + with pytest.raises(error): + operation(kind, opener, tmp_path) + elapsed = time.monotonic() - started + assert elapsed < 0.45, f"deadline did not interrupt {mode}: {elapsed:.3f}s" + assert response.closed + assert list(tmp_path.iterdir()) == [] diff --git a/tests/test_plugin_ipc.py b/tests/test_plugin_ipc.py index 177af80..00d9313 100644 --- a/tests/test_plugin_ipc.py +++ b/tests/test_plugin_ipc.py @@ -21,7 +21,7 @@ from repeater.plugins.storage import PluginStorage class FakeProc: - def __init__(self, pid=1): + def __init__(self, pid=4242): self.pid = pid self._code = None @@ -74,7 +74,12 @@ def _manager(tmp_path: Path) -> tuple[PluginManager, Path]: return PluginManager(storage, runtime), sock_path -def test_ipc_list_status_start_stop(tmp_path: Path): +def test_ipc_list_status_start_stop(tmp_path: Path, monkeypatch): + def absent_fake_group(*_args): + raise ProcessLookupError + + # Lifecycle behavior uses fake Popen objects here; never signal host PIDs. + monkeypatch.setattr("os.killpg", absent_fake_group) manager, sock_path = _manager(tmp_path) wheel = _wheel(tmp_path / "demo-0.1.0-py3-none-any.whl") manager.install(wheel) diff --git a/tests/test_plugin_ipc_hardening.py b/tests/test_plugin_ipc_hardening.py new file mode 100644 index 0000000..daa3175 --- /dev/null +++ b/tests/test_plugin_ipc_hardening.py @@ -0,0 +1,166 @@ +"""Synthetic socket/lifecycle regressions; no subprocesses or network services.""" + +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from repeater.plugins.ipc import PluginIPCClient, PluginIPCError, PluginIPCServer +from repeater.plugins.manager import PluginManager, PluginManagerError +from repeater.plugins.storage import PluginStorage + + +class SlowManager(PluginManager): + def __init__(self, root): + super().__init__(PluginStorage(root)) + self.entered = threading.Event() + self.release = threading.Event() + self.consumed = None + + def install(self, wheel_path): + self.entered.set() + assert self.release.wait(3) + self.consumed = Path(wheel_path).read_bytes() + return {"id": "demo", "version": "1"} + + def status(self, plugin_id): + return {"id": plugin_id, "state": "STOPPED"} + + +@pytest.fixture +def service(tmp_path): + manager = SlowManager(tmp_path / "plugins") + server = PluginIPCServer(tmp_path / "s", manager) + server.start() + try: + yield manager, server + finally: + manager.release.set() + server.stop() + + +def test_install_does_not_block_status(service, tmp_path): + manager, server = service + wheel = tmp_path / "demo.whl" + wheel.write_bytes(b"wheel") + with ThreadPoolExecutor(1) as pool: + future = pool.submit(PluginIPCClient(server.socket_path).install, str(wheel)) + try: + assert manager.entered.wait(1) + assert PluginIPCClient(server.socket_path, timeout=0.2).status("other")["id"] == "other" + finally: + manager.release.set() + assert future.result()["id"] == "demo" + + +def test_install_has_separate_completion_budget(service, tmp_path): + manager, server = service + wheel = tmp_path / "demo.whl" + wheel.write_bytes(b"wheel") + client = PluginIPCClient(server.socket_path, timeout=0.05) + client.operation_timeout = 1 + timer = threading.Timer(0.15, manager.release.set) + timer.start() + try: + assert client.install(str(wheel))["id"] == "demo" + finally: + timer.join() + + +def test_timeout_is_unknown_and_manager_owns_upload(service, tmp_path): + manager, server = service + wheel = tmp_path / "demo.whl" + wheel.write_bytes(b"wheel") + client = PluginIPCClient(server.socket_path, timeout=0.1) + client.operation_timeout = 0.1 + with pytest.raises(PluginIPCError) as error: + client.install(str(wheel)) + assert error.value.code == 504 + assert error.value.outcome == "unknown" + assert error.value.upload_safe is True + wheel.unlink() + manager.release.set() + # A completion barrier is used rather than sleeping before checking consumption. + server.stop() + assert manager.consumed == b"wheel" + assert not list((manager.storage.root / ".ipc-staging").glob("*")) + + +def test_slow_capacity_rejects_without_blocking_control(service, tmp_path): + manager, server = service + wheel = tmp_path / "demo.whl" + wheel.write_bytes(b"wheel") + # Capacity is deliberately one slow operation, with no waiting operation queue. + with ThreadPoolExecutor(1) as pool: + future = pool.submit(PluginIPCClient(server.socket_path).install, str(wheel)) + try: + assert manager.entered.wait(1) + with pytest.raises(PluginIPCError) as error: + PluginIPCClient(server.socket_path, timeout=0.2).install(str(wheel)) + assert error.value.code == 503 + assert PluginIPCClient(server.socket_path, timeout=0.2).status("other")["id"] == "other" + finally: + manager.release.set() + future.result() + + +def test_truncated_completion_is_unknown(tmp_path): + import socket + from repeater.plugins.ipc import _read_line + + path = tmp_path / "legacy" + listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + listener.bind(str(path)) + listener.listen(1) + + def legacy(): + with listener.accept()[0] as conn: + _read_line(conn) + conn.sendall(b'{"ok":true') + + worker = threading.Thread(target=legacy) + worker.start() + try: + with pytest.raises(PluginIPCError) as error: + PluginIPCClient(path).install("unused.whl") + assert error.value.code == 504 + assert error.value.outcome == "unknown" + assert error.value.upload_safe is False + finally: + worker.join(1) + listener.close() + + +def test_manager_install_does_not_hold_global_control_lock(tmp_path, monkeypatch): + entered, release = threading.Event(), threading.Event() + storage = PluginStorage(tmp_path) + storage.write_state("other", {"enabled": True, "version": "1"}) + + def install(path, manifest): + entered.set() + assert release.wait(3) + return {"id": "demo", "version": "1"} + + runtime = SimpleNamespace( + install_wheel=install, + stop=lambda *a, **k: None, + status_dict=lambda pid: {"id": pid}, + _runtime_state={}, + ) + manager = PluginManager(storage, runtime) + monkeypatch.setattr( + "repeater.plugins.manager.load_manifest_from_wheel", lambda p: SimpleNamespace(id="demo") + ) + with ThreadPoolExecutor(2) as pool: + installing = pool.submit(manager.install, tmp_path / "demo.whl") + try: + assert entered.wait(1) + assert pool.submit(manager.disable, "other").result(timeout=0.2)["id"] == "other" + with pytest.raises(PluginManagerError) as error: + manager.disable("demo") + assert error.value.code == 409 + finally: + release.set() + installing.result() diff --git a/tests/test_plugin_manager_lifecycle.py b/tests/test_plugin_manager_lifecycle.py index 3ce2f69..7cd6d28 100644 --- a/tests/test_plugin_manager_lifecycle.py +++ b/tests/test_plugin_manager_lifecycle.py @@ -2,7 +2,9 @@ from __future__ import annotations +import io import json +import os import subprocess import sys import zipfile @@ -17,11 +19,21 @@ from repeater.plugins.runtime import PluginRuntime, PluginState from repeater.plugins.storage import PluginStorage +@pytest.fixture(autouse=True) +def isolate_fake_process_groups(monkeypatch): + # Fake PIDs must never signal real host processes. + def absent_group(*args): + raise ProcessLookupError + + monkeypatch.setattr(os, "killpg", absent_group) + + class FakeProc: def __init__(self, pid: int = 4242): self.pid = pid self._code = None self.signals = [] + self.stdout = io.BytesIO(b"hello from plugin\n") def poll(self): return self._code diff --git a/tests/test_plugin_runtime_hardening.py b/tests/test_plugin_runtime_hardening.py new file mode 100644 index 0000000..18378f8 --- /dev/null +++ b/tests/test_plugin_runtime_hardening.py @@ -0,0 +1,309 @@ +"""Regression tests for untrusted archives and plugin process lifetime.""" + +import json +import os +import signal +import subprocess +import sys +import time +import zipfile +from pathlib import Path + +import pytest + +from repeater.plugins import manifest as manifest_module +from repeater.plugins import runtime as runtime_module +from repeater.plugins import storage as storage_module +from repeater.plugins.manifest import ManifestError, load_manifest_from_wheel, parse_manifest +from repeater.plugins.runtime import PluginRuntime +from repeater.plugins.storage import PluginStorage + + +def manifest_data(entry="ui/index.html"): + return { + "schema": 1, + "id": "openhop.demo", + "name": "Demo", + "version": "1.0.0", + "ui": {"type": "application", "entry": entry}, + } + + +@pytest.mark.parametrize( + "entry", + [ + "index.html", + "venv/index.html", + "venv/bin/python", + ".venv/index.html", + "./index.html", + "ui/./index.html", + ], +) +def test_ui_requires_dedicated_non_reserved_subtree(entry): + with pytest.raises(ManifestError): + parse_manifest(manifest_data(entry)) + + +@pytest.mark.parametrize("kind", ["directory", "leaf", "config"]) +def test_extract_rejects_existing_symlinks(tmp_path, kind): + release = tmp_path / "release" + release.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + victim = outside / "index.html" + victim.write_text("untouched") + wheel = tmp_path / "demo.whl" + with zipfile.ZipFile(wheel, "w") as zf: + zf.writestr("ui/index.html", "changed") + zf.writestr("config.default.json", "changed") + if kind == "directory": + (release / "ui").symlink_to(outside, target_is_directory=True) + elif kind == "leaf": + (release / "ui").mkdir() + (release / "ui/index.html").symlink_to(victim) + else: + (release / "config.default.json").symlink_to(victim) + runtime = PluginRuntime(PluginStorage(tmp_path / "plugins")) + with pytest.raises((ValueError, OSError)): + if kind == "config": + runtime._extract_config_default_file(wheel, release) + else: + runtime._extract_ui_assets(wheel, release, parse_manifest(manifest_data())) + assert victim.read_text() == "untouched" + + +def test_ui_matches_path_components_not_substrings(tmp_path): + wheel = tmp_path / "demo.whl" + release = tmp_path / "release" + release.mkdir() + with zipfile.ZipFile(wheel, "w") as zf: + zf.writestr("pkg/ui/index.html", "good") + zf.writestr("notui/secret.txt", "wrong") + PluginRuntime(PluginStorage(tmp_path))._extract_ui_assets( + wheel, release, parse_manifest(manifest_data()) + ) + assert (release / "ui/index.html").read_text() == "good" + assert not (release / "ui/secret.txt").exists() + + +@pytest.mark.parametrize("kind", ["manifest", "config", "metadata", "ui", "total", "count"]) +def test_expanded_archive_limits(tmp_path, monkeypatch, kind): + monkeypatch.setattr(manifest_module, "ZIP_METADATA_MAX_BYTES", 128, raising=False) + monkeypatch.setattr(manifest_module, "ZIP_MEMBER_MAX_BYTES", 128, raising=False) + monkeypatch.setattr(manifest_module, "ZIP_TOTAL_MAX_BYTES", 200, raising=False) + monkeypatch.setattr(manifest_module, "ZIP_MAX_MEMBERS", 3, raising=False) + wheel = tmp_path / "upload.whl" + members = { + "manifest": {"openhop-plugin.json": json.dumps(manifest_data()) + " " * 256}, + "config": {"config.default.json": " " * 256}, + "metadata": {"demo.dist-info/METADATA": " " * 256}, + "ui": {"ui/index.html": "x" * 256}, + "total": {"ui/a": "x" * 110, "ui/b": "x" * 110}, + "count": {f"ui/{i}": "x" for i in range(4)}, + }[kind] + with zipfile.ZipFile(wheel, "w", compression=zipfile.ZIP_DEFLATED) as zf: + for name, content in members.items(): + zf.writestr(name, content) + release = tmp_path / "release" + release.mkdir() + runtime = PluginRuntime(PluginStorage(tmp_path / "plugins")) + with pytest.raises(ValueError, match="limit|exceeds"): + if kind == "manifest": + load_manifest_from_wheel(wheel) + elif kind == "config": + runtime._extract_config_default_file(wheel, release) + elif kind == "metadata": + runtime._ensure_pip_wheel_filename(wheel) + else: + runtime._extract_ui_assets(wheel, release, parse_manifest(manifest_data())) + + +def prepared_runtime(tmp_path, **kwargs): + storage = PluginStorage(tmp_path / "plugins") + paths = storage.ensure_plugin_layout("openhop.demo", "1.0.0") + data = manifest_data() + data["runtime"] = {"type": "python", "entrypoint": "demo-cli"} + manifest = parse_manifest(data) + storage.write_manifest(manifest.id, manifest.version, manifest) + storage.write_state(manifest.id, {"version": manifest.version, "enabled": True}) + storage.set_current(manifest.id, manifest.version) + return PluginRuntime(storage, **kwargs), paths + + +def test_failed_rebuild_restores_old_environment_and_retries(tmp_path): + calls = [] + fail = True + + def run(cmd, **kwargs): + calls.append(cmd) + if "venv" in cmd: + target = Path(cmd[-1]) + (target / "bin").mkdir(parents=True) + (target / "bin/python").symlink_to(sys.executable) + (target / "pyvenv.cfg").write_text( + f"version = {sys.version_info.major}.{sys.version_info.minor}.0\n" + ) + else: + if fail: + raise RuntimeError("pip failed") + script = Path(cmd[0]).parent / "demo-cli" + script.write_text(f"#!{cmd[0]}\nprint('working')\n") + script.chmod(0o755) + return subprocess.CompletedProcess(cmd, 0, "", "") + + runtime, paths = prepared_runtime(tmp_path, run_factory=run) + venv = paths.venv_dir("1.0.0") + (venv / "bin").mkdir(parents=True) + (venv / "bin/python").write_text("old") + (venv / "pyvenv.cfg").write_text("version = 2.7.0\n") + (paths.release_dir("1.0.0") / "demo-1.0.0-py3-none-any.whl").touch() + with pytest.raises(RuntimeError, match="pip failed"): + runtime.ensure_venv_compatible("openhop.demo") + assert (venv / "bin/python").read_text() == "old" + assert (venv / "pyvenv.cfg").read_text() == "version = 2.7.0\n" + fail = False + runtime.ensure_venv_compatible("openhop.demo") + entry = runtime.resolve_entrypoint(paths, "1.0.0", "demo-cli") + assert subprocess.check_output([str(entry)], text=True, timeout=5).strip() == "working" + assert len([cmd for cmd in calls if "pip" in cmd]) == 2 + + +def test_install_subprocesses_have_deadlines(tmp_path): + calls = [] + + def run(cmd, **kwargs): + calls.append(kwargs) + return subprocess.CompletedProcess(cmd, 0, "", "") + + runtime = PluginRuntime(PluginStorage(tmp_path), run_factory=run) + runtime._create_venv(tmp_path / "venv") + (tmp_path / "venv/bin").mkdir(parents=True) + (tmp_path / "venv/bin/python").touch() + runtime._pip_install(tmp_path / "venv", tmp_path / "demo-1.0.0-py3-none-any.whl") + assert all(0 < call.get("timeout", 0) <= 300 for call in calls) + + +def test_log_tail_reads_only_bounded_suffix(tmp_path, monkeypatch): + monkeypatch.setattr(storage_module, "LOG_TAIL_MAX_BYTES", 128, raising=False) + storage = PluginStorage(tmp_path / "plugins") + paths = storage.ensure_plugin_layout("openhop.demo", "1.0.0") + paths.log_file.write_text("a" * 1024 + "\nlast line\n") + result = storage.tail_log("openhop.demo") + assert result[-1] == "last line" + assert len("\n".join(result).encode()) <= 128 + + +def test_running_plugin_log_is_continuously_bounded(tmp_path, monkeypatch): + monkeypatch.setattr(runtime_module, "LOG_MAX_BYTES", 1024) + runtime, paths = prepared_runtime(tmp_path) + bindir = paths.venv_dir("1.0.0") / "bin" + bindir.mkdir(parents=True) + entry = bindir / "demo-cli" + marker = tmp_path / "ready" + entry.write_text( + f"#!{sys.executable}\nimport os,time\nfrom pathlib import Path\nos.write(1, b'x' * 32768)\nPath({str(marker)!r}).touch()\ntime.sleep(60)\n" + ) + entry.chmod(0o755) + try: + runtime.start("openhop.demo") + deadline = time.monotonic() + 5 + while not marker.exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert marker.exists() + time.sleep(0.1) + assert paths.log_file.stat().st_size <= 1024 + finally: + runtime.stop("openhop.demo") + + +def test_install_output_is_bounded(tmp_path, monkeypatch): + monkeypatch.setattr(runtime_module, "INSTALL_OUTPUT_MAX_BYTES", 128, raising=False) + runtime = PluginRuntime(PluginStorage(tmp_path)) + result = runtime._run( + [sys.executable, "-c", "print('x' * 8192)"], + capture_output=True, + text=True, + check=False, + timeout=5, + ) + assert len(result.stdout.encode()) <= 128 + + +@pytest.mark.skipif(sys.platform != "linux", reason="Linux process-group regression") +def test_install_timeout_kills_child_processes(tmp_path): + marker = tmp_path / "child.pid" + code = ( + "import subprocess,time; from pathlib import Path; " + f"p=subprocess.Popen([{sys.executable!r}, '-c', 'import time; time.sleep(60)']); " + f"Path({str(marker)!r}).write_text(str(p.pid)); time.sleep(60)" + ) + runtime = PluginRuntime(PluginStorage(tmp_path)) + pid = None + try: + with pytest.raises(subprocess.TimeoutExpired): + runtime._run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + check=False, + timeout=0.3, + ) + assert marker.exists() + pid = int(marker.read_text()) + deadline = time.monotonic() + 2 + while alive(pid) and time.monotonic() < deadline: + time.sleep(0.01) + assert not alive(pid) + finally: + if marker.exists(): + pid = int(marker.read_text()) + if pid and alive(pid): + os.kill(pid, signal.SIGKILL) + + +def alive(pid): + try: + return Path(f"/proc/{pid}/stat").read_text().split()[2] != "Z" + except (FileNotFoundError, ProcessLookupError): + return False + + +@pytest.mark.skipif(sys.platform != "linux", reason="Linux process-group regression") +@pytest.mark.parametrize("parent_exits", [False, True]) +def test_stop_kills_term_ignoring_descendant_even_after_parent_exit(tmp_path, parent_exits): + runtime, paths = prepared_runtime(tmp_path, stop_timeout=0.15) + venv = paths.venv_dir("1.0.0") + (venv / "bin").mkdir(parents=True) + entry = venv / "bin/demo-cli" + marker = tmp_path / "child.pid" + child = "import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)" + entry.write_text( + f"#!{sys.executable}\nimport subprocess,time\nfrom pathlib import Path\np = subprocess.Popen([{sys.executable!r}, '-c', {child!r}])\nPath({str(marker)!r}).write_text(str(p.pid))\ntime.sleep({0.1 if parent_exits else 60})\n" + ) + entry.chmod(0o755) + pid = None + proc = None + try: + runtime.start("openhop.demo") + proc = runtime._handles["openhop.demo"].process + deadline = time.monotonic() + 5 + while not marker.exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert marker.exists() + pid = int(marker.read_text()) + time.sleep(0.15) + if parent_exits: + proc.wait(timeout=5) + runtime.stop("openhop.demo") + deadline = time.monotonic() + 2 + while alive(pid) and time.monotonic() < deadline: + time.sleep(0.01) + assert not alive(pid) + finally: + if pid and alive(pid): + os.kill(pid, signal.SIGKILL) + if proc and proc.poll() is None: + proc.kill() + proc.wait(timeout=5) + runtime.stop("openhop.demo") diff --git a/tests/test_plugin_static_security.py b/tests/test_plugin_static_security.py new file mode 100644 index 0000000..6b9ef55 --- /dev/null +++ b/tests/test_plugin_static_security.py @@ -0,0 +1,60 @@ +"""Public plugin serving must never expose manager/runtime release files.""" + +from types import SimpleNamespace + +import cherrypy +import pytest + +from repeater.plugins.storage import PluginStorage +from repeater.web.http_server import StatsApp + + +@pytest.fixture +def plugin_ui(tmp_path, monkeypatch): + storage = PluginStorage(tmp_path / "plugins") + paths = storage.ensure_plugin_layout("audit.ui", "1.0.0") + release = paths.release_dir("1.0.0") + storage.set_current("audit.ui", "1.0.0") + storage.write_state("audit.ui", {"version": "1.0.0", "enabled": True}) + (release / "config.default.json").write_text("PRIVATE-DEFAULT-FIXTURE") + (release / "venv").mkdir() + (release / "venv" / "pyvenv.cfg").write_text("PRIVATE-RUNTIME-FIXTURE") + manifest = SimpleNamespace(ui=SimpleNamespace(entry="ui/index.html")) + # Exercise serving's own boundary independently of manifest validation. + monkeypatch.setattr(PluginStorage, "load_current_manifest", lambda *_: manifest) + monkeypatch.setattr(cherrypy, "response", SimpleNamespace(headers={})) + app = object.__new__(StatsApp) + app.config = {"plugins": {"root": str(storage.root)}} + return app, release, manifest + + +@pytest.mark.parametrize("entry", ["index.html", "./index.html", "venv/index.html"]) +def test_public_plugin_rejects_release_or_runtime_document_root(plugin_ui, entry): + app, _, manifest = plugin_ui + manifest.ui.entry = entry + requested = "pyvenv.cfg" if entry.startswith("venv/") else "config.default.json" + with pytest.raises(cherrypy.HTTPError) as exc: + app._serve_plugin_ui("audit.ui", (requested,)) + assert exc.value.status == 404 + + +def test_public_plugin_rejects_ui_symlink_to_runtime(plugin_ui): + app, release, _ = plugin_ui + (release / "ui").symlink_to("venv", target_is_directory=True) + with pytest.raises(cherrypy.HTTPError) as exc: + app._serve_plugin_ui("audit.ui", ("pyvenv.cfg",)) + assert exc.value.status == 404 + + +def test_public_plugin_keeps_assets_and_spa_fallback(plugin_ui): + app, release, _ = plugin_ui + ui = release / "ui" + ui.mkdir() + (ui / "index.html").write_text("PUBLIC-INDEX") + (ui / "app.js").write_text("PUBLIC-SCRIPT") + assert app._serve_plugin_ui("audit.ui", ("app.js",)) == b"PUBLIC-SCRIPT" + assert app._serve_plugin_ui("audit.ui", ("client-route",)) == b"PUBLIC-INDEX" + assert app._serve_plugin_ui("audit.ui", ("config.default.json",)) == b"PUBLIC-INDEX" + with pytest.raises(cherrypy.HTTPError) as exc: + app._serve_plugin_ui("audit.ui", ("..", "config.default.json")) + assert exc.value.status == 400 diff --git a/tests/test_plugin_timeout_openapi.py b/tests/test_plugin_timeout_openapi.py new file mode 100644 index 0000000..fbf9bf0 --- /dev/null +++ b/tests/test_plugin_timeout_openapi.py @@ -0,0 +1,22 @@ +"""Document ambiguous plugin IPC completion rather than claiming cancellation.""" + +from pathlib import Path + +import yaml + + +def test_plugin_operations_document_unknown_completion(): + schema = yaml.safe_load((Path(__file__).parents[1] / "repeater/web/openapi.yaml").read_text()) + response = schema["components"].get("responses", {}).get("PluginOutcomeUnknown") + assert response is not None + body = response["content"]["application/json"]["schema"] + assert body["properties"]["outcome"]["enum"] == ["unknown"] + for path, item in schema["paths"].items(): + if path != "/plugins" and not path.startswith("/plugins/"): + continue + for method, operation in item.items(): + if method not in {"get", "post", "delete", "put", "patch", "head"}: + continue + assert operation["responses"]["504"] == { + "$ref": "#/components/responses/PluginOutcomeUnknown" + }, (path, method) diff --git a/tests/test_setup_authorization_hardening.py b/tests/test_setup_authorization_hardening.py new file mode 100644 index 0000000..7eb9263 --- /dev/null +++ b/tests/test_setup_authorization_hardening.py @@ -0,0 +1,191 @@ +"""Provisioning must not reopen when optional node/radio settings change.""" + +import copy +import json +import threading +from http.client import HTTPConnection +from types import SimpleNamespace +from unittest.mock import Mock +from wsgiref.simple_server import WSGIRequestHandler, make_server + +import cherrypy +import pytest +import yaml + +from repeater.config_manager import ConfigManager +from repeater.web.api_endpoints import APIEndpoints + + +def make_config(password="custom-test-password", node_name="mesh-repeater-01", radio=None): + return { + "repeater": {"node_name": node_name, "security": {"admin_password": password}}, + "radio_type": radio, + } + + +@pytest.fixture +def api_env(tmp_path, monkeypatch): + config = make_config() + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump(config)) + api = APIEndpoints(config=config, config_path=str(path)) + api.config_manager = ConfigManager(str(path), config) + monkeypatch.setattr(api.config_manager, "live_update_daemon", Mock(return_value=True)) + monkeypatch.setattr( + cherrypy, + "request", + SimpleNamespace(method="POST", json={}, user=None, headers={}, params={}), + ) + monkeypatch.setattr(cherrypy, "response", SimpleNamespace(status=200, headers={})) + monkeypatch.setattr(threading.Thread, "start", Mock()) + return api, path + + +@pytest.mark.parametrize( + "radio", [None, "", "none", "null", "disabled", "off", "no_radio", "modem_usb"] +) +@pytest.mark.parametrize("node_name", ["", "mesh-repeater-01", "configured-node"]) +def test_custom_password_closes_setup_independent_of_optional_settings(radio, node_name): + api = APIEndpoints.__new__(APIEndpoints) + needs_setup, reasons = api._setup_status_from_config( + make_config(node_name=node_name, radio=radio) + ) + assert needs_setup is False + assert reasons["default_password"] is False + + +@pytest.mark.parametrize("password", [None, "", "admin123", " ", "\t\n"]) +def test_legacy_fresh_config_remains_usable(password): + api = APIEndpoints.__new__(APIEndpoints) + assert api._setup_status_from_config(make_config(password=password))[0] is True + + +@pytest.mark.parametrize("endpoint", ["setup_wizard", "config_import"]) +def test_anonymous_mutation_denied_from_persisted_custom_password(api_env, endpoint): + api, path = api_env + before = path.read_bytes() + # Persisted credentials take precedence over stale startup defaults. + api.config["repeater"]["security"]["admin_password"] = "admin123" + cherrypy.request.json = {"config": make_config(password="replacement-password")} + result = getattr(api, endpoint)() + assert cherrypy.response.status == 403 + assert result["success"] is False + assert path.read_bytes() == before + api.config_manager.live_update_daemon.assert_not_called() + + +def test_completed_wizard_persists_marker_even_with_legacy_default_password(api_env): + api, path = api_env + api.config.update(make_config(password="admin123")) + path.write_text(yaml.safe_dump(api.config)) + cherrypy.request.json = { + "node_name": "mesh-repeater-01", + "hardware_key": "kiss", + "admin_password": "admin123", + "radio_preset": {"frequency": 869.618, "bandwidth": 62.5}, + } + assert api.setup_wizard()["success"] is True + persisted = yaml.safe_load(path.read_text()) + assert persisted["setup_completed"] is True + assert api.config["setup_completed"] is True + persisted["radio_type"] = None + assert api._setup_status_from_config(persisted)[0] is False + assert api.needs_setup()["needs_setup"] is False + assert api.setup_wizard()["success"] is False + assert cherrypy.response.status == 403 + + +def test_explicit_false_marker_cannot_override_custom_password(): + api = APIEndpoints.__new__(APIEndpoints) + config = make_config() + config["setup_completed"] = False + assert api._setup_status_from_config(config)[0] is False + + +def test_authenticated_restore_of_defaults_does_not_reopen_provisioning(api_env): + api, path = api_env + cherrypy.request.user = {"username": "admin", "auth_type": "jwt"} + cherrypy.request.json = { + "config": dict(make_config(password="admin123"), setup_completed=False) + } + assert api.config_import()["success"] is True + persisted = yaml.safe_load(path.read_text()) + assert persisted["setup_completed"] is True + assert api._setup_status_from_config(persisted)[0] is False + + +@pytest.mark.parametrize("password", [False, 12345678, [], {}, " admin123 "]) +def test_malformed_or_nonempty_custom_credentials_do_not_open_setup(password): + api = APIEndpoints.__new__(APIEndpoints) + assert api._setup_status_from_config(make_config(password=password))[0] is False + + +@pytest.mark.parametrize("password", [None, "admin123"]) +def test_fresh_partial_restore_still_allows_wizard(api_env, password): + api, path = api_env + api.config.update(make_config(password=password)) + path.write_text(yaml.safe_dump(api.config)) + cherrypy.request.json = {"config": {"radio_type": None}} + assert api.config_import()["success"] is True + assert api.needs_setup()["needs_setup"] is True + + +def test_fresh_restore_custom_password_closes_followup_restore(api_env): + api, path = api_env + api.config.update(make_config(password="admin123")) + path.write_text(yaml.safe_dump(api.config)) + cherrypy.request.json = {"config": make_config()} + assert api.config_import()["success"] is True + assert api.needs_setup()["needs_setup"] is False + assert api.config_import()["success"] is False + assert cherrypy.response.status == 403 + + +class QuietHandler(WSGIRequestHandler): + def log_message(self, format, *args): + pass + + +@pytest.mark.parametrize("endpoint", ["setup_wizard", "config_import"]) +def test_anonymous_loopback_dispatch_rejects_provisioned_null_radio( + tmp_path, endpoint, monkeypatch +): + # Older direct-handler tests leave synthetic request/response objects behind. + # A real WSGI dispatch must use CherryPy's thread-local request context. + monkeypatch.setattr(cherrypy, "request", cherrypy._ThreadLocalProxy("request")) + monkeypatch.setattr(cherrypy, "response", cherrypy._ThreadLocalProxy("response")) + config = make_config() + path = tmp_path / "config.yaml" + path.write_text(yaml.safe_dump(config)) + api = APIEndpoints(config=copy.deepcopy(config), config_path=str(path)) + api.config_manager = Mock() + api.config_manager.save_to_file.return_value = True + # Same public provisioning routes: authorization must happen in the handlers. + app = cherrypy.Application( + SimpleNamespace(api=api), + config={ + "/api/setup_wizard": {"tools.require_auth.on": False}, + "/api/config_import": {"tools.require_auth.on": False}, + }, + ) + with make_server("127.0.0.1", 0, app, handler_class=QuietHandler) as server: + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + connection = HTTPConnection("127.0.0.1", server.server_port, timeout=5) + try: + connection.request( + "POST", + f"/api/{endpoint}", + json.dumps({"config": make_config()}), + {"Content-Type": "application/json"}, + ) + response = connection.getresponse() + body = json.loads(response.read()) + assert response.status == 403 + assert body["success"] is False + assert yaml.safe_load(path.read_text()) == config + api.config_manager.update_and_save.assert_not_called() + finally: + connection.close() + server.shutdown() + thread.join(timeout=5) diff --git a/tests/test_update_endpoints_unit.py b/tests/test_update_endpoints_unit.py index 625e3ef..e803331 100644 --- a/tests/test_update_endpoints_unit.py +++ b/tests/test_update_endpoints_unit.py @@ -440,7 +440,7 @@ def test_do_install_wrapper_success_then_restart_failure(isolated_state, monkeyp assert "restart failed" in (st.error_message or "") -def test_do_install_restarts_plugin_manager_service_when_present(isolated_state, monkeypatch): +def test_do_install_leaves_plugin_manager_provisioning_to_wrapper(isolated_state, monkeypatch): st = isolated_state st.channel = "main" st.latest_version = "4.0.0" @@ -482,8 +482,7 @@ def test_do_install_restarts_plugin_manager_service_when_present(isolated_state, ue._do_install() - assert any(cmd[:3] == ["/bin/systemctl", "enable", "openhop-plugin-manager"] for cmd in calls) - assert any(cmd[:3] == ["/bin/systemctl", "restart", "openhop-plugin-manager"] for cmd in calls) + assert calls == [] assert st.state == "complete" @@ -604,6 +603,8 @@ def test_do_install_root_does_not_force_a_predicted_version(isolated_state, monk assert "SETUPTOOLS_SCM_PRETEND_VERSION" not in call["env"] install = [c for c in calls if any("git+https://github.com" in str(x) for x in c["cmd"])] assert install, "expected a pip install from git" + assert install[-1]["cmd"][:4] == ["/opt/openhop_repeater/venv/bin/python", "-I", "-m", "pip"] + assert install[-1]["env"] == ue._native_upgrade_environment() assert any("@dev" in str(x) for x in install[-1]["cmd"])