diff --git a/stelloauth/rootfs/usr/local/bin/addon-supervisor b/stelloauth/rootfs/usr/local/bin/addon-supervisor new file mode 100755 index 0000000..b0c1782 --- /dev/null +++ b/stelloauth/rootfs/usr/local/bin/addon-supervisor @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import dataclasses +import json +import logging +import os +import re +import shutil +import signal +import subprocess +import sys +import time +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Callable, NamedTuple + + +OPTIONS_PATH = Path("/data/options.json") +PROFILE_PATH = Path("/tmp/cloakserve") +CLOAK_ROOT = "http://127.0.0.1:9222/" +CLOAK_VERSION = "http://127.0.0.1:9222/json/version?fingerprint=addon-readiness" +CLOAK_CLOSE = "http://127.0.0.1:9222/fingerprint/addon-readiness/close" +STELLOAUTH_ROOT = "http://127.0.0.1:8080/" +DURATION = re.compile(r"^[1-9][0-9]*(?:ms|s|m|h)$") +CLOAK_COMMAND = [ + "/usr/local/bin/cloakserve", + "--headless=true", + "--idle-timeout=30", + "--data-dir=/tmp/cloakserve", +] +STELLOAUTH_COMMAND = ["/usr/local/bin/stelloauth"] + + +class ConfigError(RuntimeError): + pass + + +@dataclasses.dataclass(frozen=True) +class Options: + queue_timeout: str + rate_limit_count: int + rate_limit_duration: str + + +class HttpResponse(NamedTuple): + status: int + body: bytes + + +class ReadinessError(RuntimeError): + pass + + +def load_options(path: Path) -> Options: + try: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict) or set(value) != { + "queue_timeout", + "rate_limit_count", + "rate_limit_duration", + }: + raise ValueError + + queue_timeout = value["queue_timeout"] + rate_limit_count = value["rate_limit_count"] + rate_limit_duration = value["rate_limit_duration"] + if not isinstance(queue_timeout, str) or DURATION.fullmatch(queue_timeout) is None: + raise ValueError + if ( + isinstance(rate_limit_count, bool) + or not isinstance(rate_limit_count, int) + or not 1 <= rate_limit_count <= 20 + ): + raise ValueError + if ( + not isinstance(rate_limit_duration, str) + or DURATION.fullmatch(rate_limit_duration) is None + ): + raise ValueError + return Options(queue_timeout, rate_limit_count, rate_limit_duration) + except (OSError, UnicodeError, json.JSONDecodeError, KeyError, TypeError, ValueError): + raise ConfigError("Invalid add-on configuration") from None + + +def build_environment(options: Options) -> dict[str, str]: + environment = os.environ.copy() + environment.update( + { + "CLOAK_CDP_URL": "http://127.0.0.1:9222", + "CLOAK_MAX_SESSIONS": "1", + "CLOAK_QUEUE_TIMEOUT": options.queue_timeout, + "RATE_LIMIT_COUNT": str(options.rate_limit_count), + "RATE_LIMIT_DURATION": options.rate_limit_duration, + "HTTP_ADDRESS": "0.0.0.0", + "PORT": "8080", + "METRICS_ADDRESS": "127.0.0.1", + "METRICS_PORT": "9090", + } + ) + return environment + + +def http_request(method: str, url: str, timeout: float) -> HttpResponse: + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + data = b"" if method == "POST" else None + request = urllib.request.Request(url, data=data, method=method) + with opener.open(request, timeout=timeout) as response: + return HttpResponse(response.status, response.read()) + + +def probe_cloak( + request: Callable[[str, str, float], HttpResponse] = http_request, +) -> None: + if request("GET", CLOAK_ROOT, 2.0).status != 200: + raise ReadinessError("CloakBrowser root probe failed") + + version = request("GET", CLOAK_VERSION, 2.0) + if version.status != 200: + raise ReadinessError("CloakBrowser CDP probe failed") + try: + document = json.loads(version.body) + if not isinstance(document, dict): + raise ValueError + websocket_url = document["webSocketDebuggerUrl"] + if ( + not isinstance(websocket_url, str) + or not websocket_url.startswith("ws://127.0.0.1:") + ): + raise ValueError + parsed = urllib.parse.urlsplit(websocket_url) + if ( + parsed.scheme != "ws" + or parsed.hostname != "127.0.0.1" + or parsed.username is not None + or parsed.password is not None + or parsed.port is None + ): + raise ValueError + except (json.JSONDecodeError, KeyError, TypeError, UnicodeError, ValueError): + raise ReadinessError("CloakBrowser CDP response invalid") from None + + if request("POST", CLOAK_CLOSE, 2.0).status != 200: + raise ReadinessError("CloakBrowser readiness profile close failed") + + +def probe_stelloauth( + request: Callable[[str, str, float], HttpResponse] = http_request, +) -> None: + if request("GET", STELLOAUTH_ROOT, 2.0).status != 200: + raise ReadinessError("Stelloauth root probe failed") + + +def wait_until_ready( + name: str, + probe: Callable[[], None], + timeout: float, + stopping: Callable[[], bool], + monotonic: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, +) -> None: + deadline = monotonic() + timeout + delay = 0.25 + while monotonic() < deadline and not stopping(): + try: + probe() + return + except (OSError, ValueError, ReadinessError): + sleep(delay) + delay = min(delay * 2, 2.0) + raise ReadinessError(f"{name} did not become ready") + + +class ProcessManager: + def __init__( + self, + environment: dict[str, str], + *, + popen: Callable[..., subprocess.Popen] = subprocess.Popen, + killpg: Callable[[int, int], None] = os.killpg, + cloak_probe: Callable[[], None] = probe_cloak, + stelloauth_probe: Callable[[], None] = probe_stelloauth, + monotonic: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + profile_path: Path = PROFILE_PATH, + ) -> None: + self.environment = environment + self._popen = popen + self._killpg = killpg + self._cloak_probe = cloak_probe + self._stelloauth_probe = stelloauth_probe + self._monotonic = monotonic + self._sleep = sleep + self._profile_path = profile_path + self._children: list[tuple[str, subprocess.Popen]] = [] + self._stopping = False + self._term_sent = False + self._shutdown_deadline: float | None = None + + def _handle_signal(self, _signum: int, _frame: object) -> None: + if self._stopping: + return + logging.info("Shutdown requested") + self._stopping = True + self._begin_shutdown() + + def _clean_profiles(self) -> None: + logging.info("Cleaning CloakBrowser profiles") + if self._profile_path.is_symlink(): + self._profile_path.unlink() + elif self._profile_path.exists(): + shutil.rmtree(self._profile_path) + self._profile_path.mkdir(parents=True, mode=0o700) + self._profile_path.chmod(0o700) + + def _spawn(self, name: str, command: list[str]) -> subprocess.Popen: + process = self._popen( + command, + env=self.environment, + start_new_session=True, + ) + self._children.append((name, process)) + if self._stopping: + try: + self._killpg(process.pid, signal.SIGTERM) + except OSError: + pass + return process + + def _first_exited_child(self) -> tuple[str, int] | None: + for name, process in self._children: + returncode = process.poll() + if returncode is not None: + return name, returncode + return None + + def _startup_stopping(self) -> bool: + return self._stopping or self._first_exited_child() is not None + + def _signal_running(self, sent_signal: int) -> None: + for _name, process in self._children: + if process.poll() is not None: + continue + try: + self._killpg(process.pid, sent_signal) + except OSError: + pass + + def _begin_shutdown(self) -> None: + if self._shutdown_deadline is None: + self._shutdown_deadline = self._monotonic() + 10.0 + if not self._term_sent: + self._signal_running(signal.SIGTERM) + self._term_sent = True + + def _reap_all(self) -> None: + for _name, process in self._children: + try: + process.wait(timeout=None) + except OSError: + logging.error("Child process reap failed") + + def _shutdown(self) -> None: + if not self._children: + return + logging.info("Stopping child processes") + self._begin_shutdown() + assert self._shutdown_deadline is not None + while ( + any(process.poll() is None for _name, process in self._children) + and self._monotonic() < self._shutdown_deadline + ): + remaining = self._shutdown_deadline - self._monotonic() + self._sleep(min(0.25, max(0.0, remaining))) + + if any(process.poll() is None for _name, process in self._children): + logging.info("Forcing child processes to stop") + self._signal_running(signal.SIGKILL) + self._reap_all() + logging.info("Child processes stopped") + + @staticmethod + def _failure_status(returncode: int) -> int: + return returncode if returncode != 0 else 1 + + def _startup_failure(self, service: str) -> int: + exited = self._first_exited_child() + if exited is not None: + name, returncode = exited + logging.error(f"{name} exited unexpectedly") + status = self._failure_status(returncode) + else: + logging.error(f"{service} readiness failed") + status = 1 + self._shutdown() + return status + + def _run(self) -> int: + try: + self._clean_profiles() + if self._stopping: + self._shutdown() + return 0 + + logging.info("Starting CloakBrowser") + self._spawn("CloakBrowser", CLOAK_COMMAND) + wait_until_ready( + "CloakBrowser", + self._cloak_probe, + 60.0, + self._startup_stopping, + self._monotonic, + self._sleep, + ) + if self._stopping: + self._shutdown() + return 0 + if self._first_exited_child() is not None: + return self._startup_failure("CloakBrowser") + logging.info("CloakBrowser ready") + + logging.info("Starting Stelloauth") + self._spawn("Stelloauth", STELLOAUTH_COMMAND) + wait_until_ready( + "Stelloauth", + self._stelloauth_probe, + 30.0, + self._startup_stopping, + self._monotonic, + self._sleep, + ) + if self._stopping: + self._shutdown() + return 0 + if self._first_exited_child() is not None: + return self._startup_failure("Stelloauth") + logging.info("Stelloauth listening on 0.0.0.0:8080") + except ReadinessError: + if self._stopping: + self._shutdown() + return 0 + service = "Stelloauth" if len(self._children) > 1 else "CloakBrowser" + return self._startup_failure(service) + except OSError: + logging.error("Process startup failed") + self._shutdown() + return 1 + + while not self._stopping: + exited = self._first_exited_child() + if exited is not None: + name, returncode = exited + logging.error(f"{name} exited unexpectedly") + status = self._failure_status(returncode) + self._shutdown() + return status + self._sleep(0.25) + + self._shutdown() + return 0 + + def run(self) -> int: + previous_handlers = { + signal.SIGTERM: signal.signal(signal.SIGTERM, self._handle_signal), + signal.SIGINT: signal.signal(signal.SIGINT, self._handle_signal), + } + try: + return self._run() + finally: + for handled_signal, previous_handler in previous_handlers.items(): + signal.signal(handled_signal, previous_handler) + + +def main() -> int: + logging.basicConfig(level=logging.INFO, format="%(message)s") + try: + options = load_options(OPTIONS_PATH) + except ConfigError: + logging.error("Invalid add-on configuration") + return 2 + return ProcessManager(build_environment(options)).run() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py new file mode 100644 index 0000000..3ed7699 --- /dev/null +++ b/tests/test_supervisor.py @@ -0,0 +1,688 @@ +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +import logging +import signal +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +ROOT = Path(__file__).parents[1] +SUPERVISOR_PATH = ROOT / "stelloauth/rootfs/usr/local/bin/addon-supervisor" + + +def load_supervisor(): + loader = importlib.machinery.SourceFileLoader("addon_supervisor", str(SUPERVISOR_PATH)) + spec = importlib.util.spec_from_loader(loader.name, loader) + assert spec is not None + module = importlib.util.module_from_spec(spec) + sys.modules[loader.name] = module + loader.exec_module(module) + return module + + +@pytest.fixture +def supervisor(): + return load_supervisor() + + +def test_options_load_valid_defaults(supervisor, tmp_path: Path) -> None: + path = tmp_path / "options.json" + path.write_text( + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": 5, + "rate_limit_duration": "1h", + } + ), + encoding="utf-8", + ) + + assert supervisor.load_options(path) == supervisor.Options("60s", 5, "1h") + + +def test_environment_maps_options_and_preserves_parent(supervisor, monkeypatch) -> None: + monkeypatch.setenv("PARENT_SENTINEL", "preserved") + + environment = supervisor.build_environment(supervisor.Options("60s", 5, "1h")) + + assert environment["PARENT_SENTINEL"] == "preserved" + assert {key: environment[key] for key in ( + "CLOAK_CDP_URL", + "CLOAK_MAX_SESSIONS", + "CLOAK_QUEUE_TIMEOUT", + "RATE_LIMIT_COUNT", + "RATE_LIMIT_DURATION", + "HTTP_ADDRESS", + "PORT", + "METRICS_ADDRESS", + "METRICS_PORT", + )} == { + "CLOAK_CDP_URL": "http://127.0.0.1:9222", + "CLOAK_MAX_SESSIONS": "1", + "CLOAK_QUEUE_TIMEOUT": "60s", + "RATE_LIMIT_COUNT": "5", + "RATE_LIMIT_DURATION": "1h", + "HTTP_ADDRESS": "0.0.0.0", + "PORT": "8080", + "METRICS_ADDRESS": "127.0.0.1", + "METRICS_PORT": "9090", + } + + +@pytest.mark.parametrize( + "content", + [ + "{}", + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": 5, + "rate_limit_duration": "1h", + "unknown-SENTINEL": "secret-SENTINEL", + } + ), + json.dumps({"queue_timeout": "60s", "rate_limit_count": 5}), + json.dumps( + { + "queue_timeout": "0s-SENTINEL", + "rate_limit_count": 5, + "rate_limit_duration": "1h", + } + ), + json.dumps( + { + "queue_timeout": "60-SENTINEL", + "rate_limit_count": 5, + "rate_limit_duration": "1h", + } + ), + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": True, + "rate_limit_duration": "1h", + } + ), + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": 0, + "rate_limit_duration": "1h", + } + ), + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": 21, + "rate_limit_duration": "1h", + } + ), + json.dumps( + { + "queue_timeout": "60s", + "rate_limit_count": 5, + "rate_limit_duration": "1-SENTINEL", + } + ), + '{"queue_timeout":"malformed-SENTINEL"', + ], +) +def test_invalid_options_raise_fixed_non_secret_error( + supervisor, tmp_path: Path, caplog, content: str +) -> None: + path = tmp_path / "options.json" + path.write_text(content, encoding="utf-8") + + with caplog.at_level(logging.INFO), pytest.raises( + supervisor.ConfigError, match="^Invalid add-on configuration$" + ): + supervisor.load_options(path) + + assert "SENTINEL" not in caplog.text + assert "SENTINEL" not in str(sys.exc_info()) + + +def test_unreadable_options_raise_fixed_non_secret_error( + supervisor, tmp_path: Path, caplog +) -> None: + missing = tmp_path / "missing-SENTINEL.json" + + with caplog.at_level(logging.INFO), pytest.raises( + supervisor.ConfigError, match="^Invalid add-on configuration$" + ): + supervisor.load_options(missing) + + assert "SENTINEL" not in caplog.text + + +def test_invalid_options_main_logs_only_fixed_error( + supervisor, tmp_path: Path, caplog, monkeypatch +) -> None: + path = tmp_path / "options.json" + path.write_text('{"credential":"secret-SENTINEL"}', encoding="utf-8") + monkeypatch.setattr(supervisor, "OPTIONS_PATH", path) + + with caplog.at_level(logging.INFO): + assert supervisor.main() == 2 + + assert caplog.messages == ["Invalid add-on configuration"] + assert "SENTINEL" not in caplog.text + + +def test_probe_cloak_uses_real_cdp_websocket_and_closes_profile(supervisor) -> None: + calls: list[tuple[str, str, float]] = [] + + def request(method: str, url: str, timeout: float): + calls.append((method, url, timeout)) + if url == supervisor.CLOAK_VERSION: + return supervisor.HttpResponse( + 200, + json.dumps( + { + "webSocketDebuggerUrl": ( + "ws://127.0.0.1:9222/devtools/browser/readiness" + ) + } + ).encode(), + ) + return supervisor.HttpResponse(200, b"ok") + + supervisor.probe_cloak(request) + + assert [(method, url) for method, url, _ in calls] == [ + ("GET", supervisor.CLOAK_ROOT), + ("GET", supervisor.CLOAK_VERSION), + ("POST", supervisor.CLOAK_CLOSE), + ] + assert all(timeout == 2.0 for _, _, timeout in calls) + + +@pytest.mark.parametrize( + "root_status,version_status,version_body,close_status", + [ + (200, 200, b"{}", 200), + (200, 200, b'{"webSocketDebuggerUrl":""}', 200), + (200, 200, b"not-json-SENTINEL", 200), + ( + 200, + 200, + b'{"webSocketDebuggerUrl":"ws://192.0.2.1:9222/devtools/browser/x"}', + 200, + ), + ( + 200, + 200, + b'{"webSocketDebuggerUrl":"WS://127.0.0.1:9222/devtools/browser/x"}', + 200, + ), + ( + 503, + 200, + b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}', + 200, + ), + ( + 200, + 503, + b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}', + 200, + ), + ( + 200, + 200, + b'{"webSocketDebuggerUrl":"ws://127.0.0.1:9222/devtools/browser/x"}', + 503, + ), + ], +) +def test_probe_cloak_rejects_invalid_readiness( + supervisor, root_status, version_status, version_body, close_status +) -> None: + responses = { + supervisor.CLOAK_ROOT: supervisor.HttpResponse(root_status, b"root"), + supervisor.CLOAK_VERSION: supervisor.HttpResponse(version_status, version_body), + supervisor.CLOAK_CLOSE: supervisor.HttpResponse(close_status, b"close"), + } + + with pytest.raises(supervisor.ReadinessError): + supervisor.probe_cloak(lambda _method, url, _timeout: responses[url]) + + +def test_probe_stelloauth_requires_http_200(supervisor) -> None: + calls = [] + + def request(method: str, url: str, timeout: float): + calls.append((method, url, timeout)) + return supervisor.HttpResponse(200, b"SENTINEL-body-is-ignored") + + supervisor.probe_stelloauth(request) + assert calls == [("GET", supervisor.STELLOAUTH_ROOT, 2.0)] + + with pytest.raises(supervisor.ReadinessError): + supervisor.probe_stelloauth( + lambda _method, _url, _timeout: supervisor.HttpResponse(503, b"") + ) + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + self.sleeps: list[float] = [] + + def monotonic(self) -> float: + return self.now + + def sleep(self, delay: float) -> None: + self.sleeps.append(delay) + self.now += delay + + +def test_readiness_backoff_starts_at_quarter_second_and_caps_at_two(supervisor) -> None: + clock = FakeClock() + attempts = 0 + + def probe() -> None: + nonlocal attempts + attempts += 1 + if attempts <= 5: + raise OSError("transient-SENTINEL") + + supervisor.wait_until_ready( + "Service", probe, 60.0, lambda: False, clock.monotonic, clock.sleep + ) + + assert clock.sleeps == [0.25, 0.5, 1.0, 2.0, 2.0] + + +@pytest.mark.parametrize("name,timeout", [("CloakBrowser", 60.0), ("Stelloauth", 30.0)]) +def test_readiness_expires_at_service_timeout(supervisor, name: str, timeout: float) -> None: + clock = FakeClock() + + with pytest.raises( + supervisor.ReadinessError, match=f"^{name} did not become ready$" + ): + supervisor.wait_until_ready( + name, + lambda: (_ for _ in ()).throw(ValueError("transient-SENTINEL")), + timeout, + lambda: False, + clock.monotonic, + clock.sleep, + ) + + assert timeout <= clock.now <= timeout + 2.0 + assert max(clock.sleeps) == 2.0 + + +def test_readiness_stop_flag_aborts_immediately(supervisor) -> None: + clock = FakeClock() + called = False + + def probe() -> None: + nonlocal called + called = True + + with pytest.raises( + supervisor.ReadinessError, match="^CloakBrowser did not become ready$" + ): + supervisor.wait_until_ready( + "CloakBrowser", probe, 60.0, lambda: True, clock.monotonic, clock.sleep + ) + + assert called is False + assert clock.sleeps == [] + + +def test_probe_http_request_disables_proxies(supervisor, monkeypatch) -> None: + observed = {} + + class Response: + status = 200 + + def read(self) -> bytes: + return b"response" + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + class Opener: + def open(self, request, timeout): + observed["request"] = request + observed["timeout"] = timeout + return Response() + + def build_opener(handler): + observed["handler"] = handler + return Opener() + + monkeypatch.setattr(supervisor.urllib.request, "build_opener", build_opener) + + assert supervisor.http_request("POST", "http://127.0.0.1/", 3.0) == ( + 200, + b"response", + ) + assert observed["handler"].proxies == {} + assert observed["request"].get_method() == "POST" + assert observed["timeout"] == 3.0 + + +class FakeProcess: + def __init__(self, pid: int, *, ignores_term: bool = False) -> None: + self.pid = pid + self.returncode: int | None = None + self.ignores_term = ignores_term + self.wait_calls: list[float | None] = [] + + def poll(self) -> int | None: + return self.returncode + + def wait(self, timeout: float | None) -> int: + self.wait_calls.append(timeout) + if self.returncode is None: + self.returncode = -9 + return self.returncode + + +def manager_harness( + supervisor, + tmp_path: Path, + *, + cloak_probe=None, + stelloauth_probe=None, + ignores_term: tuple[bool, bool] = (False, False), +): + clock = FakeClock() + events: list[object] = [] + processes = [ + FakeProcess(1001, ignores_term=ignores_term[0]), + FakeProcess(1002, ignores_term=ignores_term[1]), + ] + spawned: list[FakeProcess] = [] + + def popen(command, *, env, start_new_session): + process = processes[len(spawned)] + spawned.append(process) + events.append(("start", list(command), env, start_new_session)) + return process + + def killpg(pid: int, sent_signal: int) -> None: + events.append(("signal", pid, sent_signal)) + process = next(item for item in processes if item.pid == pid) + if sent_signal == signal.SIGKILL or not process.ignores_term: + process.returncode = -sent_signal + + def default_cloak_probe() -> None: + events.append("probe cloak") + + def default_stelloauth_probe() -> None: + events.append("probe stelloauth") + + profile_path = tmp_path / "cloakserve" + environment = {"SENSITIVE_SENTINEL": "credential-code-cookie-token-SENTINEL"} + manager = supervisor.ProcessManager( + environment, + popen=popen, + killpg=killpg, + cloak_probe=cloak_probe or default_cloak_probe, + stelloauth_probe=stelloauth_probe or default_stelloauth_probe, + monotonic=clock.monotonic, + sleep=clock.sleep, + profile_path=profile_path, + ) + return SimpleNamespace( + manager=manager, + clock=clock, + events=events, + processes=processes, + spawned=spawned, + profile_path=profile_path, + environment=environment, + ) + + +def test_lifecycle_startup_order_and_profiles(supervisor, tmp_path: Path, monkeypatch) -> None: + harness = manager_harness(supervisor, tmp_path) + harness.profile_path.mkdir() + stale = harness.profile_path / "stale-profile" + stale.write_text("stale", encoding="utf-8") + + original_cloak = harness.manager._cloak_probe + original_stelloauth = harness.manager._stelloauth_probe + + def cloak_probe() -> None: + assert not stale.exists() + assert harness.profile_path.stat().st_mode & 0o777 == 0o700 + original_cloak() + + def stelloauth_probe() -> None: + original_stelloauth() + harness.manager._handle_signal(signal.SIGTERM, None) + + harness.manager._cloak_probe = cloak_probe + harness.manager._stelloauth_probe = stelloauth_probe + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 0 + + starts_and_probes = [event for event in harness.events if event == "probe cloak" or event == "probe stelloauth" or (isinstance(event, tuple) and event[0] == "start")] + assert [(event[0], event[1]) if isinstance(event, tuple) else event for event in starts_and_probes] == [ + ("start", supervisor.CLOAK_COMMAND), + "probe cloak", + ("start", supervisor.STELLOAUTH_COMMAND), + "probe stelloauth", + ] + for event in starts_and_probes: + if isinstance(event, tuple): + assert event[2] is harness.environment + assert event[3] is True + + +def test_cloak_readiness_failure_never_starts_stelloauth( + supervisor, tmp_path: Path, monkeypatch +) -> None: + def failing_probe() -> None: + raise supervisor.ReadinessError("secret-SENTINEL") + + harness = manager_harness(supervisor, tmp_path, cloak_probe=failing_probe) + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 1 + assert len(harness.spawned) == 1 + assert harness.clock.now >= 60.0 + assert harness.processes[0].wait_calls == [None] + + +def test_stelloauth_readiness_failure_stops_both_children( + supervisor, tmp_path: Path, monkeypatch +) -> None: + def failing_probe() -> None: + raise OSError("credential-SENTINEL") + + harness = manager_harness(supervisor, tmp_path, stelloauth_probe=failing_probe) + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 1 + assert len(harness.spawned) == 2 + assert {(event[1], event[2]) for event in harness.events if event[0] == "signal"} == { + (1001, signal.SIGTERM), + (1002, signal.SIGTERM), + } + assert [process.wait_calls for process in harness.processes] == [[None], [None]] + + +@pytest.mark.parametrize( + "child_index,child_status,expected_status", + [(0, 0, 1), (0, 7, 7), (1, 0, 1), (1, 9, 9)], +) +def test_child_exit_stops_sibling_and_returns_failure( + supervisor, + tmp_path: Path, + monkeypatch, + child_index: int, + child_status: int, + expected_status: int, +) -> None: + harness = manager_harness(supervisor, tmp_path) + slept = False + + def sleep(delay: float) -> None: + nonlocal slept + harness.clock.sleep(delay) + if not slept: + slept = True + harness.processes[child_index].returncode = child_status + + harness.manager._sleep = sleep + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == expected_status + sibling = harness.processes[1 - child_index] + assert ("signal", sibling.pid, signal.SIGTERM) in harness.events + assert [process.wait_calls for process in harness.processes] == [[None], [None]] + + +@pytest.mark.parametrize("incoming_signal", [signal.SIGTERM, signal.SIGINT]) +def test_signal_shutdown_forwards_sigterm_and_returns_zero( + supervisor, tmp_path: Path, monkeypatch, incoming_signal: int +) -> None: + harness = manager_harness(supervisor, tmp_path) + triggered = False + + def sleep(delay: float) -> None: + nonlocal triggered + if not triggered: + triggered = True + harness.manager._handle_signal(incoming_signal, None) + harness.clock.sleep(delay) + + harness.manager._sleep = sleep + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 0 + assert {(event[1], event[2]) for event in harness.events if event[0] == "signal"} == { + (1001, signal.SIGTERM), + (1002, signal.SIGTERM), + } + assert [process.wait_calls for process in harness.processes] == [[None], [None]] + + +def test_signal_during_cloak_readiness_never_starts_stelloauth( + supervisor, tmp_path: Path, monkeypatch +) -> None: + harness = manager_harness(supervisor, tmp_path) + + def cloak_probe() -> None: + harness.manager._handle_signal(signal.SIGTERM, None) + + harness.manager._cloak_probe = cloak_probe + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 0 + assert len(harness.spawned) == 1 + assert ("signal", 1001, signal.SIGTERM) in harness.events + assert harness.processes[0].wait_calls == [None] + + +def test_child_exit_during_cloak_readiness_never_starts_stelloauth( + supervisor, tmp_path: Path, monkeypatch +) -> None: + harness = manager_harness(supervisor, tmp_path) + + def cloak_probe() -> None: + harness.processes[0].returncode = 7 + + harness.manager._cloak_probe = cloak_probe + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 7 + assert len(harness.spawned) == 1 + assert harness.processes[0].wait_calls == [None] + + +def test_signal_while_starting_child_still_terminates_new_process_group( + supervisor, tmp_path: Path, monkeypatch +) -> None: + harness = manager_harness(supervisor, tmp_path) + original_popen = harness.manager._popen + starts = 0 + + def popen(command, *, env, start_new_session): + nonlocal starts + starts += 1 + if starts == 2: + harness.manager._handle_signal(signal.SIGTERM, None) + return original_popen(command, env=env, start_new_session=start_new_session) + + harness.manager._popen = popen + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 0 + assert ("signal", 1002, signal.SIGTERM) in harness.events + assert [process.wait_calls for process in harness.processes] == [[None], [None]] + + +def test_shutdown_uses_one_ten_second_deadline_then_sigkills_remaining_groups( + supervisor, tmp_path: Path, monkeypatch +) -> None: + harness = manager_harness(supervisor, tmp_path, ignores_term=(True, True)) + triggered = False + + def sleep(delay: float) -> None: + nonlocal triggered + if not triggered: + triggered = True + harness.manager._handle_signal(signal.SIGTERM, None) + harness.clock.sleep(delay) + + harness.manager._sleep = sleep + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + assert harness.manager.run() == 0 + assert harness.clock.now == pytest.approx(10.0) + assert [ + (event[1], event[2]) for event in harness.events if event[0] == "signal" + ] == [ + (1001, signal.SIGTERM), + (1002, signal.SIGTERM), + (1001, signal.SIGKILL), + (1002, signal.SIGKILL), + ] + assert [process.wait_calls for process in harness.processes] == [[None], [None]] + + +def test_lifecycle_logs_are_fixed_and_contain_no_sensitive_values( + supervisor, tmp_path: Path, monkeypatch, caplog +) -> None: + harness = manager_harness(supervisor, tmp_path) + + def stelloauth_probe() -> None: + harness.manager._handle_signal(signal.SIGTERM, None) + + harness.manager._stelloauth_probe = stelloauth_probe + monkeypatch.setattr(supervisor.signal, "signal", lambda *_args: None) + + with caplog.at_level(logging.INFO): + assert harness.manager.run() == 0 + + assert caplog.messages == [ + "Cleaning CloakBrowser profiles", + "Starting CloakBrowser", + "CloakBrowser ready", + "Starting Stelloauth", + "Shutdown requested", + "Stopping child processes", + "Child processes stopped", + ] + lowered = caplog.text.lower() + for sensitive in ("sentinel", "credential", "code", "cookie", "token"): + assert sensitive not in lowered