Validate deadman arming and make sysfs writes safe and reversible

dau_utils/deadman.py

@@ -111,15 +111,20 @@ def _run(command: Sequence[str], *, check: bool) -> subprocess.CompletedProcess:
         raise DeadmanError(f"{shlex.join(command)} could not run: {error}") from error
 
 
+def _queried_active(name: str) -> bool:
+    """What systemctl says about ``name``; raises DeadmanError when it cannot say."""
+    result = _run(("systemctl", "is-active", name), check=False)
+    return result.stdout.strip() not in _INACTIVE_STATES
+
+
 def _is_active(name: str) -> bool:
-    """True only if systemctl positively reports ``name`` not running. A failed
+    """True unless systemctl positively reports ``name`` not running. A failed
     query (D-Bus down, sudo denied, a hang) is treated as active -- we cannot
     claim a unit is stopped unless systemctl confirms it."""
     try:
-        result = _run(("systemctl", "is-active", name), check=False)
+        return _queried_active(name)
     except DeadmanError:
         return True
-    return result.stdout.strip() not in _INACTIVE_STATES
 
 
 def is_armed(*, unit: str = DEFAULT_UNIT) -> bool:
@@ -138,7 +143,13 @@ def arm(timeout_s: int = DEFAULT_TIMEOUT_S, *, unit: str = DEFAULT_UNIT) -> None
         raise DeadmanError(f"{unit} is already armed; disarm it before arming again")
     _run(disarm_commands(unit=unit)[1], check=False)
     _run(command, check=True)
-    if not is_armed(unit=unit):
+    # the confirmation must be a positive answer: a query that fails here is
+    # not "probably armed", it is a host whose protection cannot be shown
+    try:
+        confirmed = _queried_active(f"{unit}.timer") or _queried_active(f"{unit}.service")
+    except DeadmanError as error:
+        raise DeadmanError(f"systemd-run returned but the timer could not be confirmed ({error}); the host is NOT protected") from error
+    if not confirmed:
         raise DeadmanError(f"systemd-run returned but {unit}.timer is not active; the host is NOT protected")
 
 

dau_utils/pci_runtime_pm.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import argparse
+import os
 import re
 import shlex
 import subprocess
@@ -51,7 +52,16 @@ def __post_init__(self) -> None:
             raise RuntimePmError(f"runtime PM write value must be one of {sorted(_VALUES)}, got {self.value!r}")
 
 
+def check_pattern(pattern: str) -> str:
+    """A discovery pattern must select something; an empty one matches every line."""
+    if not pattern.strip():
+        raise RuntimePmError("a discovery pattern must not be empty")
+    return pattern
+
+
 def discover_pci_devices(lspci_output: str, *, patterns: Sequence[str] = ()) -> tuple[str, ...]:
+    for pattern in patterns:
+        check_pattern(pattern)
     matches: list[str] = []
     for line in lspci_output.splitlines():
         if any(pattern in line for pattern in patterns):
@@ -80,14 +90,42 @@ def missing_targets(writes: Sequence[RuntimePmWrite]) -> tuple[RuntimePmWrite, .
     return tuple(write for write in writes if not write.path.exists())
 
 
+def _write_attribute(path: Path, value: str) -> None:
+    """Write one sysfs attribute without following a symlink at the attribute
+    itself. The device entry under the sysfs root is a symlink by design, but
+    the attribute files are regular; one that is a link points somewhere this
+    tool must not write."""
+    if path.parent.name == "power" and path.parent.is_symlink():
+        raise RuntimePmError(f"{path.parent} is a symlink; refusing to write through it")
+    fd = os.open(path, os.O_WRONLY | os.O_NOFOLLOW | os.O_TRUNC)
+    try:
+        os.write(fd, f"{value}\n".encode())
+    finally:
+        os.close(fd)
+
+
 def apply_runtime_pm_writes(writes: Sequence[RuntimePmWrite]) -> tuple[RuntimePmWrite, ...]:
-    """Apply every write, or none: a missing attribute refuses the whole plan
-    so a device is never left half held or half released."""
+    """Apply every write, or none. A missing attribute refuses the plan before
+    anything is written; a write that fails part-way restores the attributes
+    already changed, so a device is never left half held or half released."""
     missing = missing_targets(writes)
     if missing:
         raise RuntimePmError("missing sysfs attributes: " + ", ".join(str(write.path) for write in missing))
-    for write in writes:
-        write.path.write_text(f"{write.value}\n")
+    previous: list[tuple[RuntimePmWrite, str]] = []
+    try:
+        for write in writes:
+            before = write.path.read_text()
+            _write_attribute(write.path, write.value)
+            previous.append((write, before))
+    except OSError as error:
+        restored = []
+        for done, before in reversed(previous):
+            try:
+                _write_attribute(done.path, before.strip())
+                restored.append(str(done.path))
+            except OSError:
+                pass
+        raise RuntimePmError(f"writing {write.path} failed ({error}); restored {len(restored)} of {len(previous)} earlier writes") from error
     return tuple(writes)
 
 
@@ -108,6 +146,8 @@ def main(argv: Sequence[str] | None = None) -> int:
     args = parser.parse_args(argv)
     if args.lspci_output is not None and not args.pattern:
         parser.error("--lspci-output only makes sense with --pattern")
+    if any(not pattern.strip() for pattern in args.pattern):
+        parser.error("--pattern must not be empty; an empty pattern matches every device")
 
     try:
         if args.device:
@@ -135,7 +175,12 @@ def main(argv: Sequence[str] | None = None) -> int:
             print(f"missing {write.path}", file=sys.stderr)
         print(f"applied 0 of {len(writes)} runtime PM writes: the plan is refused whole", file=sys.stderr)
         return 1
-    for write in apply_runtime_pm_writes(writes):
+    try:
+        applied = apply_runtime_pm_writes(writes)
+    except RuntimePmError as error:
+        print(f"runtime PM: {error}", file=sys.stderr)
+        return 1
+    for write in applied:
         print(f"wrote {write.path} {write.value}")
     return 0
 

dau_utils/tests/test_deadman.py

@@ -226,3 +226,22 @@ def test_module_entrypoint_runs_cli_for_uninstalled_checkout(capsys, monkeypatch
         assert exc.code == 0
 
     assert "--on-active=42" in capsys.readouterr().out
+
+
+def test_arm_does_not_report_protection_it_could_not_confirm(monkeypatch) -> None:
+    """A query that fails AFTER systemd-run is not 'probably armed': the host
+    may or may not reboot, and arm must say it could not tell."""
+    calls: list[tuple[str, ...]] = []
+
+    def run(command, **kwargs):
+        command = tuple(command)
+        calls.append(command)
+        if command[:2] == ("systemctl", "is-active"):
+            if any(c[:3] == ("sudo", "-n", "systemd-run") for c in calls):
+                raise subprocess.TimeoutExpired(command, 15.0)  # systemd stopped answering after the run
+            return subprocess.CompletedProcess(command, 0, stdout="inactive\n", stderr="")
+        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(deadman.subprocess, "run", run)
+    with pytest.raises(DeadmanError, match="could not be confirmed"):
+        arm(120)

dau_utils/tests/test_pci_runtime_pm.py

@@ -191,3 +191,46 @@ def test_module_entrypoint_runs_cli_for_uninstalled_checkout(capsys, monkeypatch
         "write /sys/bus/pci/devices/0000:04:00.0/power/control on",
         "write /sys/bus/pci/devices/0000:04:00.0/d3cold_allowed 0",
     ]
+
+
+def test_an_empty_pattern_is_refused() -> None:
+    """An empty shell variable would otherwise match every enumerated device."""
+    with pytest.raises(SystemExit):
+        main(["hold", "--pattern", "", "--lspci-output", LSPCI_OUTPUT, "--dry-run"])
+    with pytest.raises(RuntimePmError, match="must not be empty"):
+        discover_pci_devices(LSPCI_OUTPUT, patterns=(" ",))
+
+
+def test_an_attribute_that_is_a_symlink_is_not_written_through(tmp_path, capsys) -> None:
+    device_root = tmp_path / "0000:04:00.0"
+    (device_root / "power").mkdir(parents=True)
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.write_text("keep\n")
+    (device_root / "power" / "control").symlink_to(elsewhere)
+    (device_root / "d3cold_allowed").write_text("1\n")
+
+    assert main(["hold", "--device", "0000:04:00.0", "--sysfs-root", str(tmp_path)]) == 1
+    assert elsewhere.read_text() == "keep\n"
+    assert (device_root / "d3cold_allowed").read_text() == "1\n"  # nothing half applied
+    assert "runtime PM:" in capsys.readouterr().err
+
+
+def test_a_failed_write_restores_the_attributes_already_changed(tmp_path, monkeypatch, capsys) -> None:
+    device_root = tmp_path / "0000:04:00.0"
+    (device_root / "power").mkdir(parents=True)
+    control = device_root / "power" / "control"
+    d3cold = device_root / "d3cold_allowed"
+    control.write_text("auto\n")
+    d3cold.write_text("1\n")
+    real_write = pci_runtime_pm._write_attribute
+
+    def failing(path, value):
+        if path.name == "d3cold_allowed" and value == "0":
+            raise OSError("device went away")
+        real_write(path, value)
+
+    monkeypatch.setattr(pci_runtime_pm, "_write_attribute", failing)
+    assert main(["hold", "--device", "0000:04:00.0", "--sysfs-root", str(tmp_path)]) == 1
+    assert control.read_text() == "auto\n"  # the first write was rolled back
+    assert d3cold.read_text() == "1\n"
+    assert "restored 1 of 1" in capsys.readouterr().err