From 2932bdf46c32c2008a08b50ecf41fdfd56a19efd Mon Sep 17 00:00:00 2001 From: Dan Date: Sat, 1 Aug 2026 22:32:27 +0100 Subject: [PATCH] Make execution lifecycle recovery PID-safe (#56) Record PID creation times and deadlines, make restart recovery identity-safe, and use UUID batch IDs. --- CHANGELOG.md | 1 + plugins/violin_guard/execution.py | 100 +++++++++++++++--- plugins/violin_guard/state.py | 2 +- tests/guard/state/test_batch_integrity.py | 12 +++ .../guard/state/test_executor_and_adapters.py | 11 ++ 5 files changed, 113 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35175ae..d3e6def 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## 3.0.0 +- Made background execution restart-safe by recording PID creation times and deadlines, refusing to signal reused PIDs, recovering matching processes, and marking missing processes as lost; new pending batches now use collision-resistant UUIDs. - Made burst execution atomic at admission: every command is preflighted before launch, required sync credit is reserved under one lock, and unused reservations are returned after partial batches. - Fixed target extraction for dotted identifiers and direct `/dev/tcp`/`/dev/udp` redirections, and prevented network-capable local-looking commands from bypassing execution accounting. - Made interrupted skill preparation recoverable with expiring reservations and stale-owner protection; batch review now remains tied to the delivered execution receipt. diff --git a/plugins/violin_guard/execution.py b/plugins/violin_guard/execution.py index fe24de7..532ef3a 100644 --- a/plugins/violin_guard/execution.py +++ b/plugins/violin_guard/execution.py @@ -145,6 +145,49 @@ def _terminate_process(proc: subprocess.Popen) -> None: _terminate_pid(proc.pid) +def _process_create_time(proc: psutil.Process) -> float | None: + with contextlib.suppress(psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + return float(proc.create_time()) + return None + + +def _matching_process(record: dict[str, Any]) -> psutil.Process | None: + """Return the tracked process only when PID and creation time both match.""" + pid = record.get("pid") + expected = record.get("pid_create_time") + if not isinstance(pid, int) or pid <= 0 or not isinstance(expected, int | float): + return None + try: + proc = psutil.Process(pid) + actual = proc.create_time() + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + return None + return proc if abs(float(actual) - float(expected)) <= 1.0 else None + + +def _terminate_tracked_process(proc: psutil.Process) -> None: + """Terminate a process object already verified against its manifest identity.""" + with contextlib.suppress(psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + children = proc.children(recursive=True) + procs = children + [proc] + for child in procs: + with contextlib.suppress(psutil.NoSuchProcess, psutil.AccessDenied): + child.terminate() + _, alive = psutil.wait_procs(procs, timeout=2) + for child in alive: + with contextlib.suppress(psutil.NoSuchProcess, psutil.AccessDenied): + child.kill() + + +def _deadline_expired(record: dict[str, Any]) -> bool: + deadline = record.get("deadline_at") + if not isinstance(deadline, str): + return False + with contextlib.suppress(ValueError): + return datetime.now(UTC) >= datetime.fromisoformat(deadline.replace("Z", "+00:00")) + return False + + def _preview(path: Path) -> str: with path.open("rb") as handle: return handle.read(PREVIEW_BYTES).decode("utf-8", errors="replace") @@ -401,7 +444,17 @@ def execute( ) proc = subprocess.Popen(process_argv, **popen_kwargs) - record.update(status="running", pid=proc.pid) + created = _process_create_time(psutil.Process(proc.pid)) + if created is None: + _terminate_process(proc) + raise RuntimeError("could not record process creation time") + deadline_at = datetime.fromtimestamp(datetime.now(UTC).timestamp() + timeout, UTC) + record.update( + status="running", + pid=proc.pid, + pid_create_time=created, + deadline_at=deadline_at.isoformat().replace("+00:00", "Z"), + ) state.atomic_json(manifest_path, record) if background: @@ -523,23 +576,38 @@ def status(eng_dir: str, execution_id: str) -> dict[str, Any]: if not record: raise ValueError("execution not found") if record.get("background") and record.get("status") == "running": - pid = record.get("pid") - if isinstance(pid, int) and pid > 0 and not _pid_is_running(pid): + proc = _matching_process(record) + if proc is None: + # A live monitor can be finalizing a normally exited process at + # the same moment status observes that its PID has disappeared. + # Give that atomic writer a short opportunity before classifying + # an untracked process as lost (important after application restart). + time.sleep(0.1) + with state.lock_file(manifest_path): + refreshed = state.read_json(manifest_path) + if refreshed.get("status") != "running": + return refreshed record = _finalize_background( engagement=engagement, manifest_path=manifest_path, command=record["command"], phase=record["phase"], exit_code=-1, - status_name="completed", + status_name="lost", + ) + elif _deadline_expired(record): + _terminate_tracked_process(proc) + record = _finalize_background( + engagement=engagement, + manifest_path=manifest_path, + command=record["command"], + phase=record["phase"], + exit_code=-1, + status_name="timed_out", ) return record -def _pid_is_running(pid: int) -> bool: - return psutil.pid_exists(pid) - - def cancel(eng_dir: str, execution_id: str) -> dict[str, Any]: engagement = _resolve_engagement(eng_dir) record = status(str(engagement), execution_id) @@ -547,13 +615,21 @@ def cancel(eng_dir: str, execution_id: str) -> dict[str, Any]: if record.get("status") not in {"starting", "running"}: return {**record, "cancel_requested": False, "message": "execution is not running"} - pid = record.get("pid") - if not isinstance(pid, int) or pid <= 0: - raise ValueError("running execution has no valid tracked PID") + proc = _matching_process(record) + if proc is None: + manifest_path = engagement / record["evidence_paths"]["manifest"] + return _finalize_background( + engagement=engagement, + manifest_path=manifest_path, + command=record["command"], + phase=record["phase"], + exit_code=-1, + status_name="lost", + ) record["cancel_requested"] = True record["cancel_requested_at"] = _utc_now() state.atomic_json(manifest_path, record) - _terminate_pid(pid) + _terminate_tracked_process(proc) return {**record, "message": "cancellation requested for tracked process group"} diff --git a/plugins/violin_guard/state.py b/plugins/violin_guard/state.py index 95c4ba2..b84d940 100644 --- a/plugins/violin_guard/state.py +++ b/plugins/violin_guard/state.py @@ -282,7 +282,7 @@ def mark_pending_sync( if not task_id: raise ValueError("pending execution requires a captured active PTT task") data["pending"] = { - "batch_id": old.get("batch_id") or datetime.now(UTC).strftime("%Y%m%d%H%M%S"), + "batch_id": old.get("batch_id") or str(uuid.uuid4()), "commands": commands, "phase": command_phase, "created_at": old.get("created_at") diff --git a/tests/guard/state/test_batch_integrity.py b/tests/guard/state/test_batch_integrity.py index e393a76..2aae92c 100644 --- a/tests/guard/state/test_batch_integrity.py +++ b/tests/guard/state/test_batch_integrity.py @@ -57,6 +57,18 @@ def test_appending_work_invalidates_an_earlier_review(tmp_path: Path) -> None: assert state.get_pending_sync(eng)["ptt_reviewed"] is False +def test_new_pending_batches_use_unique_uuid_ids(tmp_path: Path) -> None: + eng = _engagement(tmp_path) + state.mark_pending_sync(eng, "nmap -p 80 10.10.10.10", "RECON", "PT-010") + first = state.get_pending_sync(eng)["batch_id"] + state.clear_pending_sync(eng) + state.mark_pending_sync(eng, "nmap -p 443 10.10.10.10", "RECON", "PT-010") + second = state.get_pending_sync(eng)["batch_id"] + assert first != second + assert len(first) == 36 + assert len(second) == 36 + + def _completed_batch_with_active_replacement(eng: Path, replacement: str = "PT-011") -> str: command = "nmap -p 80 10.10.10.10" history.append_history(eng, command, "RECON", 0, "evidence/executions/test.json") diff --git a/tests/guard/state/test_executor_and_adapters.py b/tests/guard/state/test_executor_and_adapters.py index be4f09e..83d3060 100644 --- a/tests/guard/state/test_executor_and_adapters.py +++ b/tests/guard/state/test_executor_and_adapters.py @@ -1,3 +1,4 @@ +import os import sys import time from pathlib import Path @@ -59,6 +60,8 @@ def test_background_execution_is_tracked_until_completion(tmp_path): assert receipt["status"] == "running" assert isinstance(receipt["pid"], int) + assert isinstance(receipt["pid_create_time"], float) + assert receipt["deadline_at"].endswith("Z") deadline = time.monotonic() + 5 current = receipt while current["status"] == "running" and time.monotonic() < deadline: @@ -91,6 +94,14 @@ def test_background_execution_can_be_cancelled_by_execution_id(tmp_path): assert current["status"] == "cancelled" +def test_process_identity_rejects_reused_pid(): + proc = execution.psutil.Process(os.getpid()) + record = {"pid": proc.pid, "pid_create_time": proc.create_time()} + assert execution._matching_process(record) is not None + record["pid_create_time"] += 10 + assert execution._matching_process(record) is None + + def test_executor_rejects_cwd_escape(tmp_path): eng = _engagement(tmp_path) with pytest.raises(ValueError, match="inside the engagement"):