2be53ecd7b
utf-8-sig exists to tolerate BOMs that Windows tooling adds to files users edit. /proc and /sys files are generated by the Linux kernel, never BOM'd and absent on Windows, so -sig there only muddies the read/write policy. Switch every literal /proc/ and /sys/ read to utf-8 and teach the footgun read rule that string literals starting with /proc/ or /sys/ are exempt (user-edited files keep utf-8-sig).
264 lines
11 KiB
Python
264 lines
11 KiB
Python
"""Bot Desktop package install: sudo hand-off and single-flight invariants."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from tools.bot_desktop import install, runtime
|
|
|
|
_REAL_INSTALL_COMMAND = runtime.install_command # captured before the fixture pins a sudo line
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_host(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: False)
|
|
monkeypatch.setattr(runtime, "is_supported_host", lambda: True)
|
|
monkeypatch.setattr(runtime, "install_command", lambda: "sudo apt-get install -y tigervnc-standalone-server")
|
|
yield
|
|
install._running.clear()
|
|
|
|
|
|
def test_empty_password_cancels_without_spawning(monkeypatch):
|
|
monkeypatch.setattr(install.subprocess, "Popen", lambda *a, **k: pytest.fail("package manager spawned"))
|
|
lines: list[str] = []
|
|
code = install.install_packages(ask_password=lambda: "", on_line=lines.append)
|
|
assert code == -1
|
|
assert any("cancelled" in line for line in lines)
|
|
|
|
|
|
def test_second_install_for_same_profile_is_refused(monkeypatch):
|
|
gate = threading.Event()
|
|
entered = threading.Event()
|
|
|
|
def slow_run(cmd, *, ask_password, on_line, timeout_seconds):
|
|
entered.set()
|
|
gate.wait(5)
|
|
return 0
|
|
|
|
monkeypatch.setattr(install, "_run", slow_run)
|
|
worker = threading.Thread(target=install.install_packages, kwargs={"ask_password": lambda: "pw", "on_line": lambda _l: None})
|
|
worker.start()
|
|
assert entered.wait(5)
|
|
with pytest.raises(install.InstallBusy):
|
|
install.install_packages(ask_password=lambda: "pw", on_line=lambda _l: None)
|
|
gate.set()
|
|
worker.join(5)
|
|
install.assert_not_running() # slot released once the run finishes
|
|
|
|
|
|
def test_claim_is_atomic_and_refuses_a_second_claim():
|
|
"""The gateway claims BEFORE spawning its worker; a second Install click must fail at claim time, not
|
|
pass a read-only check and race the worker for the slot."""
|
|
key = install.claim()
|
|
with pytest.raises(install.InstallBusy):
|
|
install.claim()
|
|
with pytest.raises(install.InstallBusy):
|
|
install.install_packages(ask_password=lambda: "pw", on_line=lambda _l: None)
|
|
install.release(key)
|
|
install.claim() # free again
|
|
install.release(key)
|
|
|
|
|
|
@pytest.mark.platforms("linux")
|
|
def test_timeout_kills_the_package_managers_whole_process_group(monkeypatch):
|
|
"""sudo forks the package manager into the same (new) session; killing sudo alone leaves apt/dnf
|
|
holding the dpkg lock as root. The timeout must take the group."""
|
|
import subprocess
|
|
import time
|
|
|
|
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: True)
|
|
# stand-in for `sudo apt-get ...`: a parent that spawns a child and waits, both in the new session
|
|
fake = ["sudo"]
|
|
real_popen = subprocess.Popen
|
|
|
|
def popen(argv, **kw):
|
|
if argv[:1] != fake:
|
|
return real_popen(argv, **kw)
|
|
return real_popen(["bash", "-c", "sleep 30 >/dev/null 2>&1 & echo child $!; wait"], **kw)
|
|
|
|
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
|
lines: list[str] = []
|
|
code = install._run("sudo apt-get install -y x", ask_password=lambda: "", on_line=lines.append, timeout_seconds=0.5)
|
|
assert code != 0
|
|
child = next(int(line.split()[1]) for line in lines if line.startswith("child "))
|
|
from pathlib import Path
|
|
|
|
def gone() -> bool: # /proc-based: a reparented orphan sits outside our subtree, where os.kill(pid, 0) is guarded
|
|
try:
|
|
return "Z" in (Path(f"/proc/{child}/stat").read_text(encoding="utf-8").rsplit(")", 1)[1].split() or ["Z"])[0]
|
|
except OSError:
|
|
return True
|
|
|
|
for _ in range(50): # the child must die with the group, not linger reparented to init
|
|
if gone():
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
subprocess.run(["kill", "-9", str(child)], check=False)
|
|
pytest.fail("grandchild survived the install timeout")
|
|
|
|
|
|
def test_passwordless_sudo_runs_the_install_without_asking_for_a_password(monkeypatch):
|
|
"""A host with NOPASSWD sudo must not pop the masked password card — the card blocks the install
|
|
for up to five minutes on an answer nobody needs to give — and the package command still runs
|
|
and streams its output."""
|
|
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: True)
|
|
spawned: list[list[str]] = []
|
|
|
|
class _Proc:
|
|
"""Stands in for the sudo child: the drain reads the pipe by fd, so stdout is a real pipe with
|
|
the output already written and the write end closed (EOF)."""
|
|
pid = 4242
|
|
returncode = 0
|
|
stdin = __import__("io").StringIO()
|
|
|
|
def __init__(self, argv, **kw):
|
|
spawned.append(argv)
|
|
r, w = os.pipe()
|
|
os.write(w, b"Reading package lists...\nDone\n")
|
|
os.close(w)
|
|
self.stdout = os.fdopen(r, "rb")
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
def poll(self):
|
|
return 0
|
|
|
|
monkeypatch.setattr(install.subprocess, "Popen", _Proc)
|
|
monkeypatch.setattr(install.os, "killpg", lambda *a: None)
|
|
lines: list[str] = []
|
|
code = install.install_packages(ask_password=lambda: pytest.fail("sudo password asked despite NOPASSWD"),
|
|
on_line=lines.append)
|
|
assert code == 0
|
|
assert spawned and spawned[0][:1] == ["sudo"] and "-S" not in spawned[0]
|
|
assert "Done" in lines
|
|
|
|
|
|
@pytest.mark.platforms("linux")
|
|
def test_timeout_returns_and_frees_the_slot_even_when_a_descendant_survives(monkeypatch):
|
|
"""From an unprivileged Hermes, killpg reaches the sudo leader but not a root-owned apt child; that
|
|
child keeps the pipe's write end open, so draining stdout never sees EOF and the profile slot stays
|
|
taken forever. The timeout must end the drain and release the slot regardless of what survived.
|
|
Stand-in for the unkillable root child: a grandchild in its own session holding our stdout."""
|
|
import subprocess
|
|
import time
|
|
|
|
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: True)
|
|
monkeypatch.setattr(install, "_TERM_GRACE_SECONDS", 0.2, raising=False)
|
|
real_popen = subprocess.Popen
|
|
|
|
def popen(argv, **kw):
|
|
if argv[:1] != ["sudo"]:
|
|
return real_popen(argv, **kw)
|
|
return real_popen(["sh", "-c", "setsid sleep 30 & echo child $!; wait"], **kw)
|
|
|
|
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
|
lines: list[str] = []
|
|
started = time.monotonic()
|
|
worker = threading.Thread(target=lambda: lines.append(
|
|
f"code {install.install_packages(ask_password=lambda: '', on_line=lines.append, timeout_seconds=0.5)}"))
|
|
worker.start()
|
|
worker.join(3.0)
|
|
survivor = next((int(line.split()[1]) for line in lines if line.startswith("child ")), None)
|
|
if survivor:
|
|
subprocess.run(["kill", "-9", str(survivor)], check=False)
|
|
assert not worker.is_alive(), f"install_packages hung {time.monotonic() - started:.1f}s on a surviving descendant"
|
|
assert any(line.startswith("code ") and line != "code 0" for line in lines), lines
|
|
assert any("timed out" in line for line in lines), lines
|
|
install.release(install.claim()) # slot is free again
|
|
|
|
|
|
def test_root_installs_without_sudo_and_without_asking(monkeypatch):
|
|
"""The official Docker image runs Hermes as uid 0 with no sudo binary: the package manager must be
|
|
run directly, and the password card must never be raised for a user who already is root."""
|
|
import subprocess
|
|
|
|
monkeypatch.setattr(install.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr("shutil.which", lambda name, *a, **k: None)
|
|
monkeypatch.setattr(runtime, "package_manager", lambda: "apt")
|
|
monkeypatch.setattr(runtime, "install_command", _REAL_INSTALL_COMMAND)
|
|
spawned: list[list[str]] = []
|
|
real_popen = subprocess.Popen
|
|
|
|
def popen(argv, **kw):
|
|
spawned.append(list(argv))
|
|
return real_popen(["true"], **kw)
|
|
|
|
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
|
code = install.install_packages(ask_password=lambda: pytest.fail("root was asked for a sudo password"),
|
|
on_line=lambda _l: None)
|
|
assert code == 0
|
|
assert spawned and spawned[0][0] == "apt-get", spawned
|
|
|
|
|
|
def test_no_sudo_binary_returns_the_host_command_instead_of_a_password_card(monkeypatch):
|
|
"""Unprivileged with no sudo on the host (minimal containers): a password card would be a dead end,
|
|
the user needs the exact command to run on the host instead."""
|
|
monkeypatch.setattr(install.os, "geteuid", lambda: 1000)
|
|
monkeypatch.setattr("shutil.which", lambda name, *a, **k: None)
|
|
monkeypatch.setattr(install.subprocess, "Popen", lambda *a, **k: pytest.fail("package manager spawned"))
|
|
lines: list[str] = []
|
|
code = install.install_packages(ask_password=lambda: pytest.fail("password card raised without sudo"),
|
|
on_line=lines.append)
|
|
assert code == install.NO_SUDO
|
|
assert any("apt-get install" in line for line in lines), lines
|
|
|
|
|
|
def test_timeout_finishes_the_group_when_only_the_leader_dies_on_term(monkeypatch):
|
|
"""TERM ends the leader shell at once, but a descendant in the same group that ignores TERM used to be
|
|
left alive holding the dpkg lock: the kill helper returned as soon as the leader was reaped. Cleanup is
|
|
done only when the whole process group is gone."""
|
|
import subprocess
|
|
import time
|
|
|
|
monkeypatch.setattr(install, "_sudo_nopasswd", lambda: True)
|
|
monkeypatch.setattr(install, "_TERM_GRACE_SECONDS", 0.3, raising=False)
|
|
real_popen = subprocess.Popen
|
|
|
|
def popen(argv, **kw):
|
|
if argv[:1] != ["sudo"]:
|
|
return real_popen(argv, **kw)
|
|
return real_popen(["sh", "-c", "sh -c 'trap \"\" TERM; echo child $$; sleep 30' & wait"], **kw)
|
|
|
|
monkeypatch.setattr(install.subprocess, "Popen", popen)
|
|
lines: list[str] = []
|
|
install.install_packages(ask_password=lambda: "", on_line=lines.append, timeout_seconds=0.5)
|
|
child = next(int(line.split()[1]) for line in lines if line.startswith("child "))
|
|
deadline = time.monotonic() + 2.0
|
|
while time.monotonic() < deadline and _alive(child):
|
|
time.sleep(0.05)
|
|
try:
|
|
assert not _alive(child), "the TERM-ignoring descendant survived the timeout cleanup"
|
|
finally:
|
|
subprocess.run(["kill", "-9", str(child)], check=False)
|
|
|
|
|
|
def _alive(pid: int) -> bool:
|
|
import psutil
|
|
|
|
return psutil.pid_exists(pid)
|
|
|
|
|
|
def test_install_slot_is_held_across_processes(tmp_path):
|
|
"""The gateway (Install card) and the CLI (`screen install`) are separate processes on one host: the
|
|
single-flight must be a file lock in the profile's state dir, not only the in-process set."""
|
|
import subprocess
|
|
import sys
|
|
|
|
key = install.claim()
|
|
try:
|
|
probe = subprocess.run([sys.executable, "-c",
|
|
"import fcntl,sys\nfh=open(sys.argv[1],'a+',encoding='utf-8')\n"
|
|
"try:\n fcntl.flock(fh.fileno(), fcntl.LOCK_EX|fcntl.LOCK_NB); print('free')\n"
|
|
"except OSError:\n print('held')", str(runtime.state_dir() / "install.lock")],
|
|
capture_output=True, text=True)
|
|
assert probe.stdout.strip() == "held", probe
|
|
finally:
|
|
install.release(key)
|
|
install.release(install.claim()) # freed: claimable again
|