mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 13:32:50 +00:00
359 lines
14 KiB
Python
359 lines
14 KiB
Python
"""Regression tests for untrusted archives and plugin process lifetime."""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize("stage", ["old_renamed", "new_config"])
|
|
def test_fresh_runtime_recovers_interrupted_rebuild_snapshot(tmp_path, stage):
|
|
snapshot = tmp_path / "snapshot"
|
|
calls = []
|
|
capturing = True
|
|
|
|
def run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
if "venv" in cmd:
|
|
target = Path(cmd[-1])
|
|
if capturing and stage == "old_renamed":
|
|
shutil.copytree(tmp_path / "plugins", snapshot, symlinks=True)
|
|
raise RuntimeError("snapshot captured")
|
|
(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"
|
|
)
|
|
if capturing:
|
|
shutil.copytree(tmp_path / "plugins", snapshot, symlinks=True)
|
|
raise RuntimeError("snapshot captured")
|
|
else:
|
|
entry = Path(cmd[0]).parent / "demo-cli"
|
|
entry.write_text(f"#!{cmd[0]}\nprint('recovered')\n")
|
|
entry.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="snapshot captured"):
|
|
runtime.ensure_venv_compatible("openhop.demo")
|
|
|
|
capturing = False
|
|
calls.clear()
|
|
fresh = PluginRuntime(PluginStorage(snapshot), run_factory=run)
|
|
fresh.ensure_venv_compatible("openhop.demo")
|
|
assert len([cmd for cmd in calls if "pip" in cmd]) == 1
|
|
entry = fresh.resolve_entrypoint(fresh.storage.paths_for("openhop.demo"), "1.0.0", "demo-cli")
|
|
assert subprocess.check_output([str(entry)], text=True, timeout=5).strip() == "recovered"
|
|
calls.clear()
|
|
fresh.ensure_venv_compatible("openhop.demo")
|
|
assert calls == []
|
|
|
|
|
|
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")
|