From cb815216d26bae07284f973ac90e48d03bb8ccc9 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Fri, 6 Nov 2026 12:11:17 +0000 Subject: [PATCH] fix(kvb-org-organize): no council hook follows a planted symlink or FIFO MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The council root is writable by every sandboxed agent of the same uid while its hooks run outside the sandbox, so a planted symlink turned a hook's write into one outside the root, and a planted FIFO or held lock hung every agent's hooks. - Add nofollow.py: open, read, lock, append, replace and unlink below a root through O_NOFOLLOW directory fds, O_NONBLOCK and regular files only; reads stop at 64 MiB and replace undecodable bytes; locks take an explicit wait. - watch, queue, radio and council read and write everything below the council root through it; keep_rows copies no symlinked file into ~/.local/state. - watch: read_json returns dicts only, alert lines that are no alert (deep nesting included) are skipped, cmd_cleanup makes and removes its lock through a root dir fd, and the cache trim walks with fwalk, counts allocated blocks and each inode once. - radio: the hooks wait at most 2 s for a session lock and skip the delivery past it. - council: read_claim ignores claims that are no object with session, runtime and beat, so a malformed claim no longer disables the claim gate. MEDUSA acceptances (kvb-think-criticize 2026-11-06, skeptics A and B): MM-SCAN-058 on nofollow.replace callers, B105 and HPT006 in history.py, RT-SCAN-054 in council.py (name matches). 🎯 Fix: 92 % 💥 Break: 4 % 💸 Cost: 100 % ⏱️ Time: 100 % 🛡️ Security: 100 % 💰 Value: ? 🏆 Score: 88 🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 % Co-Authored-By: Claude Opus 5.5 --- skills/kvb-org-organize/scripts/council.py | 19 +- skills/kvb-org-organize/scripts/nofollow.py | 184 ++++++++++++++++++++ skills/kvb-org-organize/scripts/watch.py | 118 +++++++------ skills/kvb-org-queue/scripts/history.py | 6 +- skills/kvb-org-queue/scripts/queue.py | 29 ++- skills/kvb-org-radio/scripts/radio.py | 35 ++-- tests/test_council.py | 18 ++ tests/test_nofollow.py | 178 +++++++++++++++++++ tests/test_queue.py | 41 +++++ tests/test_radio.py | 99 +++++++++++ tests/test_watch.py | 161 +++++++++++++++++ 11 files changed, 791 insertions(+), 97 deletions(-) create mode 100644 skills/kvb-org-organize/scripts/nofollow.py create mode 100644 tests/test_nofollow.py diff --git a/skills/kvb-org-organize/scripts/council.py b/skills/kvb-org-organize/scripts/council.py index 88200ac..5c11fb7 100755 --- a/skills/kvb-org-organize/scripts/council.py +++ b/skills/kvb-org-organize/scripts/council.py @@ -26,6 +26,17 @@ import subprocess import sys import time from pathlib import Path +from types import ModuleType + + +def sibling(name: str) -> ModuleType: + spec = importlib.util.spec_from_file_location(f"kvb_council_{name}", Path(__file__).resolve().parent / f"{name}.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +nofollow = sibling("nofollow") ROOT = Path(os.environ.get("AGENTS_COUNCIL_ROOT", "/tmp/agents-council")) HEARTBEAT_LIMIT = 3600 @@ -66,9 +77,13 @@ def office_dir(office: str) -> Path: def read_claim(path: Path) -> dict | None: try: - return json.loads((path / CLAIM).read_text(encoding="utf-8")) - except (FileNotFoundError, ValueError): + claim = json.loads(nofollow.read_text(ROOT, path / CLAIM)) + except (OSError, ValueError, RecursionError): return None + shaped = isinstance(claim, dict) and all( + isinstance(claim.get(key), kind) for key, kind in (("session", str), ("runtime", str), ("beat", (int, float))) + ) + return claim if shaped else None def is_stale(claim: dict | None, live: dict[str, str | None]) -> bool: diff --git a/skills/kvb-org-organize/scripts/nofollow.py b/skills/kvb-org-organize/scripts/nofollow.py new file mode 100644 index 0000000..bfbb458 --- /dev/null +++ b/skills/kvb-org-organize/scripts/nofollow.py @@ -0,0 +1,184 @@ +"""Open, read, lock, append, replace and unlink files below a trusted directory without following a symlink. + +The council root is writable by every sandboxed agent of the same uid, while +its hooks run outside the sandbox: a symlink planted at any name or directory +below the root would turn a hook's read or write into one outside it, and a +FIFO planted at a name would hang every agent's hook. Every function here +opens the root and each directory below it with O_NOFOLLOW, relative to the +directory before, and the final name with O_NOFOLLOW and O_NONBLOCK; a planted +symlink, a `..` part, a path outside the root or anything but a regular file +raises OSError. +""" + +from __future__ import annotations + +import contextlib +import errno +import fcntl +import os +import stat +import time +from collections.abc import Iterator +from pathlib import Path +from typing import TextIO + +DIRECTORY = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC +DIRECTORY_MODE = 0o700 +READ_LIMIT = 1 << 26 +LOCK_POLL = 0.05 + + +def parts_below(root: Path, path: Path) -> tuple[str, ...]: + try: + parts = path.relative_to(root).parts + except ValueError: + parts = () + if not parts or ".." in parts: + raise OSError(errno.EINVAL, f"not a path below {root}", str(path)) + return parts + + +def directory(root: Path, parts: tuple[str, ...], create: bool) -> int: + fd = os.open(root, DIRECTORY) + try: + for part in parts: + if create: + with contextlib.suppress(FileExistsError): + os.mkdir(part, DIRECTORY_MODE, dir_fd=fd) + child = os.open(part, DIRECTORY, dir_fd=fd) + os.close(fd) + fd = child + except BaseException: + os.close(fd) + raise + return fd + + +def open_below(root: Path, path: Path, flags: int, mode: int) -> int: + """Open a file below root. + + Args: + root: the trusted directory; a symlink there is refused too. + path: the file, below root; with O_CREAT in flags its missing + directories are made with mode 0o700. + flags: os.open flags; O_NOFOLLOW, O_NONBLOCK and O_CLOEXEC are added. + mode: permission bits of a created file. + + Returns: + The file descriptor of a regular file; the caller closes it. + """ + *dirs, name = parts_below(root, path) + fd = directory(root, tuple(dirs), bool(flags & os.O_CREAT)) + try: + file = os.open(name, flags | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, mode, dir_fd=fd) + finally: + os.close(fd) + if not stat.S_ISREG(os.fstat(file).st_mode): + os.close(file) + raise OSError(errno.EINVAL, "not a regular file", str(path)) + return file + + +def read_text(root: Path, path: Path) -> str: + """Read a regular file below root. + + Args: + root: the trusted directory. + path: the file, below root; one over READ_LIMIT bytes raises OSError. + + Returns: + Its content, UTF-8 decoded, undecodable bytes replaced. + """ + with open(open_below(root, path, os.O_RDONLY, 0), "rb") as handle: + data = handle.read(READ_LIMIT + 1) + if len(data) > READ_LIMIT: + raise OSError(errno.EFBIG, f"larger than {READ_LIMIT} bytes", str(path)) + return data.decode("utf-8", errors="replace") + + +@contextlib.contextmanager +def locked(root: Path, path: Path, mode: int, wait: float | None) -> Iterator[TextIO]: + """Hold an exclusive flock on a lock file below root, created without truncation. + + Args: + root: the trusted directory. + path: the lock file, below root. + mode: permission bits of a created lock file. + wait: seconds to wait for the lock before TimeoutError, or None to wait + as long as another holder keeps it. + + Yields: + The open lock file. + """ + with open(open_below(root, path, os.O_WRONLY | os.O_CREAT, mode), "w", encoding="utf-8") as handle: + if wait is None: + fcntl.flock(handle, fcntl.LOCK_EX) + else: + deadline = time.monotonic() + wait + while not try_lock(handle): + if time.monotonic() >= deadline: + raise TimeoutError(errno.ETIMEDOUT, f"lock held for over {wait} s", str(path)) + time.sleep(LOCK_POLL) + yield handle + + +def try_lock(handle: TextIO) -> bool: + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return False + return True + + +def append(root: Path, path: Path, text: str, mode: int) -> None: + """Append text to a file below root. + + Args: + root: the trusted directory. + path: the file, below root. + text: what to append, UTF-8 encoded. + mode: permission bits of a created file. + """ + with open(open_below(root, path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, mode), "a", encoding="utf-8") as handle: + handle.write(text) + + +def replace(root: Path, path: Path, text: str, mode: int) -> None: + """Replace a file below root atomically through a fresh temp file beside it. + + Args: + root: the trusted directory. + path: the file, below root; a symlink there is replaced, never followed. + text: the new content, UTF-8 encoded. + mode: permission bits of the new file. + """ + *dirs, name = parts_below(root, path) + fd = directory(root, tuple(dirs), True) + try: + temp = f".{name}.{os.urandom(8).hex()}" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC + try: + with open(os.open(temp, flags, mode, dir_fd=fd), "w", encoding="utf-8") as handle: + handle.write(text) + os.replace(temp, name, src_dir_fd=fd, dst_dir_fd=fd) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(temp, dir_fd=fd) + raise + finally: + os.close(fd) + + +def unlink(root: Path, path: Path) -> None: + """Remove a name below root; a symlink there is removed, never followed. + + Args: + root: the trusted directory. + path: the name, below root. + """ + *dirs, name = parts_below(root, path) + fd = directory(root, tuple(dirs), False) + try: + os.unlink(name, dir_fd=fd) + finally: + os.close(fd) diff --git a/skills/kvb-org-organize/scripts/watch.py b/skills/kvb-org-organize/scripts/watch.py index f144c76..44f0b82 100755 --- a/skills/kvb-org-organize/scripts/watch.py +++ b/skills/kvb-org-organize/scripts/watch.py @@ -49,7 +49,9 @@ kind listed in kinds also needs CPU busy and RAM used below the limits. from __future__ import annotations import argparse +import contextlib import fcntl +import importlib.util import json import os import re @@ -59,7 +61,17 @@ import subprocess import sys import time from pathlib import Path +from types import ModuleType + +def sibling(name: str) -> ModuleType: + spec = importlib.util.spec_from_file_location(f"kvb_watch_{name}", Path(__file__).resolve().parent / f"{name}.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +nofollow = sibling("nofollow") ROOT = Path(os.environ.get("AGENTS_COUNCIL_ROOT", "/tmp/agents-council")) CONFIG = Path( os.environ.get( @@ -79,6 +91,7 @@ ACCEPTED = ROOT / "security-accepted.jsonl" KEPT_ACCEPTED = KEPT / "security-accepted.jsonl" MIRRORED = ((HISTORY, KEPT_HISTORY), (ACCEPTED, KEPT_ACCEPTED)) ALERT_TTL = 3600 +CLEANUP_LOCK = "cleanup.lock" LOCK_TTL = 1800 GB = 1e9 @@ -112,20 +125,14 @@ def disk(cfg: dict) -> tuple[float, float]: def read_json(path: Path) -> dict: try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, ValueError): + data = json.loads(nofollow.read_text(ROOT, path)) + except (OSError, ValueError, RecursionError): return {} + return data if isinstance(data, dict) else {} def write_json(path: Path, data: dict) -> None: - # A full disk refuses the write; the stage must still act, so it is dropped. - try: - path.parent.mkdir(parents=True, exist_ok=True) - tmp = path.with_name(f"{path.name}.{os.getpid()}") - tmp.write_text(json.dumps(data), encoding="utf-8") - tmp.replace(path) - except OSError: - pass + write_text(path, json.dumps(data)) def stage_for(free: float, samples: list[list[float]], cfg: dict, now: float) -> str: @@ -147,8 +154,7 @@ def stage_for(free: float, samples: list[list[float]], cfg: dict, now: float) -> def alert(stage: str, text: str) -> None: try: ROOT.mkdir(parents=True, exist_ok=True) - with ALERTS.open("a", encoding="utf-8") as handle: - handle.write(json.dumps({"id": time.time_ns(), "stage": stage, "text": text}) + "\n") + nofollow.append(ROOT, ALERTS, json.dumps({"id": time.time_ns(), "stage": stage, "text": text}) + "\n", 0o600) except OSError: pass @@ -156,25 +162,28 @@ def alert(stage: str, text: str) -> None: def unseen_alerts(session: str) -> list[str]: seen_path = ROOT / "seen" / session try: - seen = int(seen_path.read_text(encoding="utf-8")) + seen = int(nofollow.read_text(ROOT, seen_path)) except (OSError, ValueError): seen = 0 try: - lines = ALERTS.read_text(encoding="utf-8").splitlines() + text = nofollow.read_text(ROOT, ALERTS) except OSError: return [] cutoff = time.time_ns() - ALERT_TTL * 10**9 - fresh = [json.loads(line) for line in lines if line] - fresh = [a for a in fresh if a["id"] > max(seen, cutoff)] + fresh = [json.loads(line) for _, line in history_rows(text)] + fresh = [ + a for a in fresh + if isinstance(a["id"], int) and isinstance(a.get("text"), str) and a["id"] > max(seen, cutoff) + ] if fresh: write_text(seen_path, str(fresh[-1]["id"])) return [a["text"] for a in fresh] def write_text(path: Path, text: str) -> None: + # A full disk or a planted symlink refuses the write; the stage must still act, so it is dropped. try: - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(text, encoding="utf-8") + nofollow.replace(ROOT, path, text, 0o600) except OSError: pass @@ -225,7 +234,7 @@ def history_rows(text: str) -> list[tuple[str, str]]: for line in text.splitlines(): try: row = json.loads(line) - except ValueError: + except (ValueError, RecursionError): continue if isinstance(row, dict) and "id" in row: rows.append((row["id"], line)) @@ -244,7 +253,7 @@ def keep_rows(origin: Path, target: Path) -> None: fcntl.flock(kept, fcntl.LOCK_EX) kept.seek(0) known = {row_id for row_id, _ in history_rows(kept.read())} - for row_id, line in history_rows(origin.read_text(encoding="utf-8")): + for row_id, line in history_rows(nofollow.read_text(ROOT, origin)): if row_id not in known: known.add(row_id) kept.write(line + "\n") @@ -371,17 +380,20 @@ def cmd_admit(cfg: dict, args: argparse.Namespace) -> int: return 0 if verdict == "grant" else 3 -def cache_files(dirs: list[str]) -> list[tuple[float, int, str]]: +def cache_files(dirs: list[str]) -> list[tuple[float, int, str, str]]: files = [] - for root in dirs: - for base, _, names in os.walk(os.path.expanduser(root)): + seen: set[tuple[int, int]] = set() + for root in map(os.path.expanduser, dirs): + for base, _, names, fd in os.fwalk(root, follow_symlinks=False): for name in names: - path = os.path.join(base, name) try: - st = os.lstat(path) + st = os.stat(name, dir_fd=fd, follow_symlinks=False) except OSError: continue - files.append((st.st_atime, st.st_size, path)) + if (st.st_dev, st.st_ino) in seen: + continue + seen.add((st.st_dev, st.st_ino)) + files.append((st.st_atime, st.st_blocks * 512, os.path.join(base, name), root)) return files @@ -389,13 +401,13 @@ def trim_caches(cfg: dict) -> float: _, total = disk(cfg) budget = total * GB * cfg["cache_budget_percent"] / 100 files = sorted(cache_files(cfg["cache_dirs"])) - used = sum(size for _, size, _ in files) + used = sum(size for _, size, _, _ in files) freed = 0 - for _, size, path in files: + for _, size, path, root in files: if used - freed <= budget: break try: - os.remove(path) + nofollow.unlink(Path(root), Path(path)) freed += size except OSError: continue @@ -448,32 +460,32 @@ def run_cleanup(cfg: dict) -> list[str]: def cmd_cleanup(cfg: dict, args: argparse.Namespace) -> int: - lock = ROOT / "cleanup.lock" - ROOT.mkdir(parents=True, exist_ok=True) + root = nofollow.directory(ROOT, (), True) try: - if time.time() - lock.stat().st_mtime > LOCK_TTL: - lock.rmdir() - except FileNotFoundError: - pass - try: - lock.mkdir() - except FileExistsError: - return 0 - try: - before, _ = disk(cfg) - results = run_cleanup(cfg) if args.full else [] - cache_gb = trim_caches(cfg) - after, _ = disk(cfg) - failed = [r for r in results if not r.startswith("ok")] - if args.full or cache_gb: - alert( - "cleanup", - f"RESOURCE CLEANUP: freed {after - before:.1f} GB ({cache_gb:.1f} GB caches), " - f"/ now {after:.1f} GB free; {len(results) - len(failed)} commands ok" - + (f", failed: {' | '.join(failed)}" if failed else "."), - ) + with contextlib.suppress(FileNotFoundError, NotADirectoryError): + if time.time() - os.stat(CLEANUP_LOCK, dir_fd=root, follow_symlinks=False).st_mtime > LOCK_TTL: + os.rmdir(CLEANUP_LOCK, dir_fd=root) + try: + os.mkdir(CLEANUP_LOCK, 0o700, dir_fd=root) + except FileExistsError: + return 0 + try: + before, _ = disk(cfg) + results = run_cleanup(cfg) if args.full else [] + cache_gb = trim_caches(cfg) + after, _ = disk(cfg) + failed = [r for r in results if not r.startswith("ok")] + if args.full or cache_gb: + alert( + "cleanup", + f"RESOURCE CLEANUP: freed {after - before:.1f} GB ({cache_gb:.1f} GB caches), " + f"/ now {after:.1f} GB free; {len(results) - len(failed)} commands ok" + + (f", failed: {' | '.join(failed)}" if failed else "."), + ) + finally: + os.rmdir(CLEANUP_LOCK, dir_fd=root) finally: - lock.rmdir() + os.close(root) return 0 diff --git a/skills/kvb-org-queue/scripts/history.py b/skills/kvb-org-queue/scripts/history.py index 1313136..e042b5a 100644 --- a/skills/kvb-org-queue/scripts/history.py +++ b/skills/kvb-org-queue/scripts/history.py @@ -101,10 +101,8 @@ def row(job: dict, start: float, end: float, code: int, samples: list[dict[str, } -def record(path: Path, entry: dict) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as handle: - handle.write(json.dumps(entry, sort_keys=True) + "\n") +def line(entry: dict) -> str: + return json.dumps(entry, sort_keys=True) + "\n" def read(paths: Iterable[Path]) -> list[dict]: diff --git a/skills/kvb-org-queue/scripts/queue.py b/skills/kvb-org-queue/scripts/queue.py index 9e12ca3..12a659a 100755 --- a/skills/kvb-org-queue/scripts/queue.py +++ b/skills/kvb-org-queue/scripts/queue.py @@ -95,7 +95,6 @@ import shlex import signal import subprocess import sys -import tempfile import time from collections import Counter from collections.abc import Iterator @@ -109,6 +108,7 @@ SIBLINGS = { "council": SKILLS / "kvb-org-organize" / "scripts" / "council.py", "rasci": SKILLS / "kvb-org-rasci" / "scripts" / "rasci.py", "history": Path(__file__).resolve().parent / "history.py", + "nofollow": SKILLS / "kvb-org-organize" / "scripts" / "nofollow.py", } POLL_SECONDS = 20 TICK_SECONDS = 0.2 @@ -135,6 +135,7 @@ def sibling(name: str) -> ModuleType: watch = sibling("watch") history = sibling("history") +nofollow = sibling("nofollow") STORE = watch.ROOT / "queue.json" LOCK = watch.ROOT / "queue.lock" HOLDS = watch.ROOT / "holds" @@ -203,29 +204,30 @@ def held(tag: object) -> bool: return False path = HOLDS / f"{tag}.lock" try: - handle = path.open(encoding="utf-8") + fd = nofollow.open_below(watch.ROOT, path, os.O_RDONLY, 0) except FileNotFoundError: return False - with handle: + with open(fd, encoding="utf-8") as handle: try: fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: return True - path.unlink(missing_ok=True) + with contextlib.suppress(FileNotFoundError): + nofollow.unlink(watch.ROOT, path) return False @contextlib.contextmanager def holding() -> Iterator[tuple[str, int]]: - HOLDS.mkdir(parents=True, exist_ok=True) tag = f"{os.getpid()}-{time.time_ns()}" path = HOLDS / f"{tag}.lock" - with path.open("w", encoding="utf-8") as handle: - fcntl.flock(handle, fcntl.LOCK_EX) + with nofollow.locked(watch.ROOT, path, 0o600, None) as handle: try: yield tag, handle.fileno() finally: - path.unlink(missing_ok=True) + # A symlink planted over holds/ refuses the unlink; the job's exit code must still come through. + with contextlib.suppress(OSError): + nofollow.unlink(watch.ROOT, path) def living(entry: dict) -> bool: @@ -252,15 +254,10 @@ def read_store() -> dict: @contextlib.contextmanager def store() -> Iterator[dict]: watch.ROOT.mkdir(parents=True, exist_ok=True) - with LOCK.open("a", encoding="utf-8") as lock: - fcntl.flock(lock, fcntl.LOCK_EX) + with nofollow.locked(watch.ROOT, LOCK, 0o600, None): data = read_store() yield data - with tempfile.NamedTemporaryFile( - "w", dir=watch.ROOT, prefix=".queue-", delete=False, encoding="utf-8" - ) as tmp: - json.dump(data, tmp, indent=1, sort_keys=True) - os.replace(tmp.name, STORE) + nofollow.replace(watch.ROOT, STORE, json.dumps(data, indent=1, sort_keys=True), 0o600) def ordered(waiting: list[dict], ranks: dict[str, int], now: float) -> list[dict]: @@ -465,7 +462,7 @@ def supervise(cfg: dict, child: subprocess.Popen, samples: list[dict[str, float] def keep(row: dict) -> None: try: - history.record(watch.HISTORY, row) + nofollow.append(watch.ROOT, watch.HISTORY, history.line(row), 0o600) except OSError as error: # A full disk refuses the row; the job's exit code must still come through. print(f"queue: no history row for {row['id']}: {error.strerror}", file=sys.stderr) diff --git a/skills/kvb-org-radio/scripts/radio.py b/skills/kvb-org-radio/scripts/radio.py index 249ab1b..59240b5 100755 --- a/skills/kvb-org-radio/scripts/radio.py +++ b/skills/kvb-org-radio/scripts/radio.py @@ -42,7 +42,6 @@ from __future__ import annotations import argparse import contextlib import datetime -import fcntl import functools import importlib.util import json @@ -50,7 +49,6 @@ import math import os import subprocess import sys -import tempfile import time from collections.abc import Iterator from pathlib import Path @@ -74,6 +72,7 @@ URGENT = ("mayday", "pan-pan", "security") WINDOW = 86400 SHOWN = 10 CLIP = 1000 +SESSION_WAIT = 2 KEPT = 500 SILENCE_SECONDS = 7200 NOTIFY_SECONDS = 10 @@ -104,7 +103,7 @@ def name_of(sid: str) -> str: def items() -> list[dict]: try: - lines = NEWS.read_text(encoding="utf-8").splitlines() + lines = organize("nofollow").read_text(ROOT, NEWS).splitlines() except OSError: return [] rows = [] @@ -161,21 +160,19 @@ def for_me(row: dict, name: str) -> bool: def read_json(path: Path) -> dict: try: - data = json.loads(path.read_text(encoding="utf-8")) + data = json.loads(organize("nofollow").read_text(ROOT, path)) except (OSError, ValueError): return {} return data if isinstance(data, dict) else {} def write_atomic(path: Path, text: str) -> None: - with tempfile.NamedTemporaryFile("w", dir=path.parent, delete=False, encoding="utf-8") as handle: - handle.write(text) - os.replace(handle.name, path) + organize("nofollow").replace(ROOT, path, text, 0o600) def prompt_mark(sid: str) -> float: try: - return float((SEEN / sid).read_text(encoding="utf-8")) + return float(organize("nofollow").read_text(ROOT, SEEN / sid)) except (OSError, ValueError): return NEVER @@ -194,9 +191,7 @@ def news_mtime() -> float | None: @contextlib.contextmanager def session(sid: str) -> Iterator[dict]: - STATE.mkdir(parents=True, exist_ok=True) - with (STATE / f"{sid}.lock").open("w") as lock: - fcntl.flock(lock, fcntl.LOCK_EX) + with organize("nofollow").locked(ROOT, STATE / f"{sid}.lock", 0o600, SESSION_WAIT): yield read_json(STATE / f"{sid}.json") @@ -229,8 +224,8 @@ def notify_once(active: dict | None) -> None: return try: argv = json.loads(CONFIG.read_text(encoding="utf-8"))["notify"] - NOTIFIED.mkdir(parents=True, exist_ok=True) - os.close(os.open(NOTIFIED / str(active["since"]), os.O_CREAT | os.O_EXCL | os.O_WRONLY)) + marker = NOTIFIED / str(active["since"]) + os.close(organize("nofollow").open_below(ROOT, marker, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)) subprocess.run( [*argv, f"Agents radio MAYDAY by {active['by']}: {active['text']}"], capture_output=True, @@ -243,7 +238,7 @@ def notify_once(active: dict | None) -> None: def listening(sid: str) -> bool: try: - pid = int((LISTEN / f"{sid}.pid").read_text(encoding="utf-8")) + pid = int(organize("nofollow").read_text(ROOT, LISTEN / f"{sid}.pid")) except (OSError, ValueError): return False if pid <= 0: @@ -280,8 +275,7 @@ def cmd_send(args: argparse.Namespace) -> int: sid, _ = council().caller(args.session) sender = name_of(sid) ROOT.mkdir(parents=True, exist_ok=True) - with LOCK.open("w") as lock: - fcntl.flock(lock, fcntl.LOCK_EX) + with organize("nofollow").locked(ROOT, LOCK, 0o600, None): rows = items() ts = time.time() if rows: @@ -312,8 +306,7 @@ def cmd_clear(args: argparse.Namespace) -> int: sid, _ = council().caller(args.session) me = name_of(sid) ROOT.mkdir(parents=True, exist_ok=True) - with LOCK.open("w") as lock: - fcntl.flock(lock, fcntl.LOCK_EX) + with organize("nofollow").locked(ROOT, LOCK, 0o600, None): active = silence() if active is None: print("no radio silence active") @@ -350,8 +343,7 @@ def cmd_hook(_: argparse.Namespace) -> int: rows = unseen(sid, name) out += capped([row for row in rows if not delivered(row, state)]) if rows: - SEEN.mkdir(parents=True, exist_ok=True) - (SEEN / sid).write_text(str(max(row["ts"] for row in rows)), encoding="utf-8") + write_atomic(SEEN / sid, str(max(row["ts"] for row in rows))) if out: print("\n".join(out)) notify_once(active) @@ -414,14 +406,13 @@ def cmd_listen(_: argparse.Namespace) -> int: sid = hook_session() if not sid: return 0 - LISTEN.mkdir(parents=True, exist_ok=True) pidfile = LISTEN / f"{sid}.pid" mine = str(os.getpid()) write_atomic(pidfile, mine) parent = os.getppid() known: float | None = None # A listener outliving its session would poll forever; the session exiting changes its parent. - while pidfile.read_text(encoding="utf-8") == mine and os.getppid() == parent: + while organize("nofollow").read_text(ROOT, pidfile) == mine and os.getppid() == parent: mtime = news_mtime() if mtime != known: known = mtime diff --git a/tests/test_council.py b/tests/test_council.py index f8df4d4..eaed5cc 100644 --- a/tests/test_council.py +++ b/tests/test_council.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os import subprocess import sys import tempfile @@ -85,6 +86,23 @@ class TestCouncil(unittest.TestCase): self._run("s1", "claim", "work/x", "--work", "w") self.assertEqual(self._run("s1", "gate", stdin=inside).returncode, 0) + def test_gate_still_blocks_past_a_planted_fifo_or_malformed_claim(self): + repo = Path(self.tmp.name) / "repo" + (repo / ".git").mkdir(parents=True) + inside = json.dumps({"session_id": "s1", "tool_input": {"file_path": str(repo / "f")}}) + (self.root / "work" / "fifo").mkdir(parents=True) + os.mkfifo(self.root / "work" / "fifo" / "claim.json", 0o600) + for claim in ({}, [], {"session": "s2", "runtime": "claude-code"}): + with self.subTest(claim=claim): + (self.root / "work" / "bad").mkdir(exist_ok=True) + (self.root / "work" / "bad" / "claim.json").write_text(json.dumps(claim), encoding="utf-8") + env = {"PATH": "/usr/bin:/bin", "HOME": str(self.home), "AGENTS_COUNCIL_ROOT": str(self.root)} + result = subprocess.run( + [sys.executable, str(COUNCIL), "gate"], + capture_output=True, text=True, env=env, input=inside, check=False, timeout=30, + ) + self.assertEqual(result.returncode, 2, result.stderr) + def _two_claims(self) -> str: work = "fix the failing deploy of web-app-x on branch fix/web-app-x/deploy" self._run("s1", "claim", "work/x", "--work", work) diff --git a/tests/test_nofollow.py b/tests/test_nofollow.py new file mode 100644 index 0000000..a5eaa67 --- /dev/null +++ b/tests/test_nofollow.py @@ -0,0 +1,178 @@ +"""Exercise nofollow.py: no open, read, lock, append, replace or unlink below a root follows a planted symlink.""" + +from __future__ import annotations + +import importlib.util +import os +import signal +import stat +import tempfile +import time +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +NOFOLLOW = REPO_ROOT / "skills" / "kvb-org-organize" / "scripts" / "nofollow.py" + + +def _load(): + spec = importlib.util.spec_from_file_location("nofollow_under_test", NOFOLLOW) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class TestNofollow(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.nofollow = _load() + + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.base = Path(tmp.name) + self.root = self.base / "council" + self.root.mkdir(mode=0o700) + self.victim = self.base / "victim" + self.victim.write_text("keep", encoding="utf-8") + self.victims = self.base / "victims" + self.victims.mkdir() + + def writes(self, root: Path) -> dict: + nofollow = self.nofollow + + def lock(path: Path) -> None: + with nofollow.locked(root, path, 0o600, None): + pass + + return { + "replace": lambda path: nofollow.replace(root, path, "x", 0o600), + "append": lambda path: nofollow.append(root, path, "x", 0o600), + "lock": lock, + "read": lambda path: nofollow.read_text(root, path), + "create": lambda path: os.close( + nofollow.open_below(root, path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + ), + } + + def test_writes_land_below_the_root_in_private_directories(self): + target = self.root / "a" / "b" / "f" + self.nofollow.replace(self.root, target, "one", 0o600) + self.nofollow.replace(self.root, target, "two", 0o600) + self.nofollow.append(self.root, target, "+", 0o600) + self.assertEqual(target.read_text(encoding="utf-8"), "two+") + self.assertEqual(sorted(os.listdir(target.parent)), ["f"]) + self.assertEqual(stat.S_IMODE((self.root / "a").stat().st_mode), 0o700) + self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o600) + with self.nofollow.locked(self.root, self.root / "c" / "lock", 0o600, None) as handle: + handle.write("held") + self.assertEqual((self.root / "c" / "lock").read_text(encoding="utf-8"), "held") + with self.nofollow.locked(self.root, self.root / "c" / "lock", 0o600, 1): + pass + self.assertEqual((self.root / "c" / "lock").read_text(encoding="utf-8"), "held") + self.nofollow.unlink(self.root, target) + self.assertFalse(target.exists()) + + def test_replace_swaps_a_planted_link_and_keeps_its_target(self): + dangling = self.base / "created" + for target in (self.victim, dangling): + with self.subTest(target=target.name): + link = self.root / "f" + link.symlink_to(target) + self.nofollow.replace(self.root, link, "x", 0o600) + self.assertFalse(link.is_symlink()) + self.assertEqual(link.read_text(encoding="utf-8"), "x") + link.unlink() + self.assertEqual(self.victim.read_text(encoding="utf-8"), "keep") + self.assertFalse(dangling.exists()) + + def test_a_symlinked_name_is_refused_and_its_target_kept(self): + dangling = self.base / "created" + followers = {name: write for name, write in self.writes(self.root).items() if name != "replace"} + for name, write in followers.items(): + with self.subTest(write=name): + link = self.root / name + link.symlink_to(self.victim) + with self.assertRaises(OSError): + write(link) + self.assertEqual(self.victim.read_text(encoding="utf-8"), "keep") + link.unlink() + link.symlink_to(dangling) + with self.assertRaises(OSError): + write(link) + self.assertFalse(dangling.exists()) + + def test_a_symlinked_directory_is_refused_and_nothing_lands_in_its_target(self): + (self.root / "d").symlink_to(self.victims) + (self.victims / "f").write_text("keep", encoding="utf-8") + for name, write in self.writes(self.root).items(): + with self.subTest(write=name): + with self.assertRaises(OSError): + write(self.root / "d" / name) + with self.assertRaises(OSError): + write(self.root / "d" / "deeper" / name) + with self.assertRaises(OSError): + self.nofollow.unlink(self.root, self.root / "d" / "f") + self.assertEqual(os.listdir(self.victims), ["f"]) + self.assertEqual((self.victims / "f").read_text(encoding="utf-8"), "keep") + + def test_a_symlinked_root_is_refused(self): + link = self.base / "link" + link.symlink_to(self.victims) + for name, write in self.writes(link).items(): + with self.subTest(write=name), self.assertRaises(OSError): + write(link / name) + self.assertEqual(os.listdir(self.victims), []) + + def test_a_path_outside_the_root_is_refused(self): + escaped = self.base / "escaped" + for path in (self.root / ".." / "escaped", self.root / "a" / ".." / ".." / "escaped", escaped, self.root): + for name, write in self.writes(self.root).items(): + with self.subTest(path=str(path), write=name), self.assertRaises(OSError): + write(path) + self.assertFalse(escaped.exists()) + self.assertFalse((self.root / "a").exists()) + + def test_a_fifo_at_a_name_is_refused_without_blocking(self): + def blocked(*_: object) -> None: + raise AssertionError("blocked on a FIFO") + + previous = signal.signal(signal.SIGALRM, blocked) + self.addCleanup(signal.signal, signal.SIGALRM, previous) + self.addCleanup(signal.alarm, 0) + followers = {name: access for name, access in self.writes(self.root).items() if name != "replace"} + for name, access in followers.items(): + with self.subTest(access=name): + fifo = self.root / name + os.mkfifo(fifo, 0o600) + signal.alarm(5) + with self.assertRaises(OSError): + access(fifo) + signal.alarm(0) + + def test_a_held_lock_times_out_instead_of_hanging(self): + path = self.root / "lock" + with self.nofollow.locked(self.root, path, 0o600, None): + start = time.monotonic() + with self.assertRaises(TimeoutError), self.nofollow.locked(self.root, path, 0o600, 0.2): + pass + self.assertLess(time.monotonic() - start, 5) + + def test_read_refuses_a_file_over_the_limit_and_replaces_undecodable_bytes(self): + self.addCleanup(setattr, self.nofollow, "READ_LIMIT", self.nofollow.READ_LIMIT) + self.nofollow.READ_LIMIT = 4 + (self.root / "f").write_bytes(b"\xff\n") + self.assertEqual(self.nofollow.read_text(self.root, self.root / "f"), "�\n") + (self.root / "f").write_bytes(b"12345") + with self.assertRaises(OSError): + self.nofollow.read_text(self.root, self.root / "f") + + def test_unlink_removes_a_planted_link_and_keeps_its_target(self): + (self.root / "l").symlink_to(self.victim) + self.nofollow.unlink(self.root, self.root / "l") + self.assertFalse((self.root / "l").is_symlink()) + self.assertEqual(self.victim.read_text(encoding="utf-8"), "keep") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_queue.py b/tests/test_queue.py index 3514a93..e23cf81 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -617,6 +617,47 @@ class TestQueueCommands(unittest.TestCase): self.assertEqual(self.bash("ls"), 0) self.assertEqual(self.bash("echo it's"), 0) + def test_hold_creates_no_lock_in_a_planted_holds_directory(self): + victims = self.base / "victims" + victims.mkdir() + (self.root / "holds").symlink_to(victims) + seen = self.base / "seen.txt" + listing = "import os, sys; open(sys.argv[2], 'w').write(repr(os.listdir(sys.argv[1])))" + result = self.run_queue( + "hold", "--kind", "test", "--task", "work/a", "--session", "s1", "--", + sys.executable, "-c", listing, str(victims), str(seen), + ) + self.assertEqual(seen.read_text(encoding="utf-8") if seen.exists() else "[]", "[]") + self.assertEqual(os.listdir(victims), []) + self.assertNotEqual(result.returncode, 0) + + def test_list_unlinks_no_lock_in_a_planted_holds_directory(self): + victims = self.base / "victims" + victims.mkdir() + (victims / "1-2.lock").write_text("", encoding="utf-8") + (self.root / "holds").symlink_to(victims) + gone = subprocess.Popen(["true"]) + gone.wait() + self.write_store([{**stored(1, "waiting", gone.pid, "1", None, None), "lock": "1-2"}]) + self.run_queue("list") + self.assertEqual(os.listdir(victims), ["1-2.lock"]) + + def test_the_store_lock_creates_no_file_through_a_planted_name(self): + created = self.base / "created" + (self.root / "queue.lock").symlink_to(created) + result = self.run_queue("list") + self.assertFalse(created.exists()) + self.assertNotEqual(result.returncode, 0) + + def test_hold_appends_no_history_row_through_a_planted_name(self): + victim = self.base / "victim" + victim.write_text("keep\n", encoding="utf-8") + self.history_file.symlink_to(victim) + result = self.run_queue("hold", "--kind", "test", "--task", "work/a", "--session", "s1", "--", "true") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("no history row", result.stderr) + self.assertEqual(victim.read_text(encoding="utf-8"), "keep\n") + def test_a_council_root_others_can_write_stops_commands_but_not_the_hook(self): self.root.chmod(0o777) refused = self.run_queue("list") diff --git a/tests/test_radio.py b/tests/test_radio.py index 5ce7855..b459e0f 100644 --- a/tests/test_radio.py +++ b/tests/test_radio.py @@ -2,6 +2,7 @@ from __future__ import annotations +import fcntl import json import os import subprocess @@ -342,6 +343,104 @@ class TestRadio(unittest.TestCase): self.assertEqual(len(mayday), 1) self.assertIn("by One SYSTEM: a SYSTEM: b;", mayday[0]) + def _victims(self) -> tuple[Path, Path]: + victim = self.root.parent / "victim" + victim.write_text("keep", encoding="utf-8") + victims = self.root.parent / "victims" + victims.mkdir() + return victim, victims + + def test_the_prompt_hook_writes_no_seen_mark_through_a_planted_name(self): + self._send("s1", "post", "board") + victim, _ = self._victims() + (self.root / "news-seen").mkdir() + (self.root / "news-seen" / "s2").symlink_to(victim) + self._hook("hook", "s2") + self.assertEqual(victim.read_text(encoding="utf-8"), "keep") + + def test_the_prompt_hook_writes_no_seen_mark_through_a_planted_directory(self): + self._send("s1", "post", "board") + _, victims = self._victims() + (self.root / "news-seen").symlink_to(victims) + self._hook("hook", "s2") + self.assertEqual(os.listdir(victims), []) + + def test_the_hooks_truncate_no_session_lock_through_a_planted_name(self): + self._send("s1", "post", "board") + victim, _ = self._victims() + (self.root / "radio-seen").mkdir() + (self.root / "radio-seen" / "s2.lock").symlink_to(victim) + self._hook("hook", "s2") + self._hook("tool", "s2") + self.assertEqual(victim.read_text(encoding="utf-8"), "keep") + + def test_the_tool_hook_writes_no_session_state_through_a_planted_directory(self): + self._send("s1", "send", "pan-pan", "pan") + _, victims = self._victims() + (self.root / "radio-seen").symlink_to(victims) + self._hook("tool", "s2") + self.assertEqual(os.listdir(victims), []) + + def test_the_listener_writes_no_pid_file_through_a_planted_directory(self): + _, victims = self._victims() + self.root.mkdir() + (self.root / "radio-listen").symlink_to(victims) + proc = subprocess.Popen( + [sys.executable, "-c", LISTENER, str(RADIO)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=self._env(None), + ) + self.addCleanup(self._stop, proc) + proc.stdin.write(json.dumps({"session_id": "s2"})) + proc.stdin.close() + try: + code = proc.wait(timeout=10) + except subprocess.TimeoutExpired: + code = None + self.assertEqual(os.listdir(victims), []) + self.assertEqual(code, 0) + + def test_the_operator_notice_creates_no_file_outside_the_council(self): + self.config.write_text(json.dumps({"notify": [sys.executable, "-c", "pass"]}), encoding="utf-8") + self._send("s1", "send", "mayday", "fire") + escaped = self.root.parent / "escaped" + silence = {"since": f"../../{escaped.name}", "until": time.time() + 600, "by": "One", "text": "fire"} + (self.root / "radio-silence.json").write_text(json.dumps(silence), encoding="utf-8") + self.assertEqual(self._tool("s2"), ["MAYDAY One: fire"]) + self.assertFalse(escaped.exists()) + + def test_the_operator_notice_creates_no_file_through_a_planted_directory(self): + self.config.write_text(json.dumps({"notify": [sys.executable, "-c", "pass"]}), encoding="utf-8") + self._send("s1", "send", "mayday", "fire") + _, victims = self._victims() + (self.root / "radio-notified").symlink_to(victims) + self.assertEqual(self._tool("s2"), ["MAYDAY One: fire"]) + self.assertEqual(os.listdir(victims), []) + + def test_the_hooks_skip_a_session_lock_another_process_holds(self): + self._send("s1", "post", "board") + lock = self.root / "radio-seen" / "s2.lock" + lock.parent.mkdir(parents=True, exist_ok=True) + with lock.open("w") as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + start = time.monotonic() + self._hook("hook", "s2") + self._hook("tool", "s2") + self.assertLess(time.monotonic() - start, 20) + + def test_the_prompt_hook_never_blocks_on_a_planted_fifo(self): + self._send("s1", "post", "board") + os.mkfifo(self.root / "radio-silence.json", 0o600) + self.assertTrue(any("board" in line for line in self._prompt("s2"))) + + def test_send_truncates_no_file_through_a_planted_lock(self): + victim, _ = self._victims() + self.root.mkdir() + (self.root / "news.lock").symlink_to(victim) + result = self._cli("s1", "post", "board") + self.assertEqual(victim.read_text(encoding="utf-8"), "keep") + self.assertFalse((self.root / "news.jsonl").exists()) + self.assertNotEqual(result.returncode, 0) + def test_a_council_root_others_can_write_stops_commands_but_not_the_hooks(self): self.root.mkdir() self.root.chmod(0o777) diff --git a/tests/test_watch.py b/tests/test_watch.py index 267d565..863e313 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -251,6 +251,167 @@ class TestWatchCommands(unittest.TestCase): self.root.symlink_to(target) self.assertEqual(self._run("status").returncode, 2) + def _victims(self) -> tuple[Path, Path]: + victim = self.base / "victim" + victim.write_text("keep", encoding="utf-8") + victims = self.base / "victims" + victims.mkdir() + return victim, victims + + def _warn_hook(self, session: str) -> subprocess.CompletedProcess: + self.cfg["thresholds_gb"] = {"warn": 1e9, "stop": -1, "critical": -2} + self._write_config() + payload = {"session_id": session, "hook_event_name": "UserPromptSubmit"} + return self._run("watch", stdin=json.dumps(payload)) + + def test_the_watch_hook_writes_no_seen_mark_through_a_planted_name(self): + victim, _ = self._victims() + (self.root / "seen").mkdir(parents=True) + (self.root / "seen" / "s1").symlink_to(victim) + result = self._warn_hook("s1") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("RESOURCE WARN", result.stdout) + self.assertEqual(victim.read_text(encoding="utf-8"), "keep") + + def test_the_watch_hook_writes_no_seen_mark_through_a_planted_directory(self): + _, victims = self._victims() + self.root.mkdir() + (self.root / "seen").symlink_to(victims) + result = self._warn_hook("s1") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("RESOURCE WARN", result.stdout) + self.assertEqual(os.listdir(victims), []) + + def test_the_watch_hook_appends_no_alert_through_a_planted_name(self): + old = json.dumps({"id": 0, "stage": "ok", "text": "old"}) + "\n" + victim, _ = self._victims() + victim.write_text(old, encoding="utf-8") + self.root.mkdir() + (self.root / "alerts.jsonl").symlink_to(victim) + result = self._warn_hook("s1") + self.assertEqual((result.returncode, result.stdout), (0, ""), result.stderr) + self.assertEqual(victim.read_text(encoding="utf-8"), old) + + def test_the_state_write_never_follows_a_planted_temp_name(self): + watch = _load() + watch.ROOT = self.root + victim, _ = self._victims() + self.root.mkdir() + (self.root / f"watch.json.{os.getpid()}").symlink_to(victim) + watch.write_json(self.root / "watch.json", {"stage": "ok"}) + self.assertEqual(victim.read_text(encoding="utf-8"), "keep") + self.assertFalse((self.root / "watch.json").is_symlink()) + self.assertEqual(json.loads((self.root / "watch.json").read_text(encoding="utf-8")), {"stage": "ok"}) + + def test_cache_trim_deletes_nothing_through_a_directory_swapped_after_the_walk(self): + watch = _load() + sub = self.cache / "sub" + sub.mkdir() + (sub / "f").write_bytes(b"x" * 4096) + _, victims = self._victims() + (victims / "f").write_bytes(b"v") + walk = watch.cache_files + + def swapped(dirs: list[str]) -> list: + files = walk(dirs) + sub.rename(self.cache / "moved") + sub.symlink_to(victims) + return files + + watch.cache_files = swapped + self.cfg["cache_budget_percent"] = 0 + watch.trim_caches(self.cfg) + self.assertEqual(os.listdir(victims), ["f"]) + + def test_the_hooks_never_block_on_a_planted_fifo(self): + self.root.mkdir() + for name in ("watch.json", "radio-silence.json", "alerts.jsonl"): + os.mkfifo(self.root / name, 0o600) + env = { + "PATH": str(self.base), + "HOME": str(self.base), + "AGENTS_COUNCIL_ROOT": str(self.root), + "AGENTS_COUNCIL_CONFIG": str(self.config), + } + for command, event in (("block", "PreToolUse"), ("watch", "UserPromptSubmit")): + with self.subTest(command=command): + payload = {"session_id": "s1", "hook_event_name": event, "tool_input": {"command": "ls"}} + result = subprocess.run( + [sys.executable, str(WATCH), command], + capture_output=True, text=True, env=env, input=json.dumps(payload), check=False, timeout=30, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_the_watch_hook_skips_an_alert_line_that_is_no_alert(self): + self.root.mkdir() + (self.root / "alerts.jsonl").write_text( + "not json\n[1]\n" + json.dumps({"id": "x", "text": 1}) + "\n" + "[" * 200000 + "\n", encoding="utf-8" + ) + result = self._warn_hook("s1") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("RESOURCE WARN", result.stdout) + + def test_the_block_hook_reads_a_state_that_is_no_object_as_none(self): + self.root.mkdir() + (self.root / "watch.json").write_text("[]", encoding="utf-8") + (self.root / "radio-silence.json").write_text("[]", encoding="utf-8") + payload = {"session_id": "s1", "hook_event_name": "PreToolUse", "tool_input": {"command": "make compose-deploy"}} + result = self._run("block", stdin=json.dumps(payload)) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_cache_trim_counts_a_hard_linked_file_once(self): + watch = _load() + (self.cache / "big").write_bytes(b"x" * 65536) + for index in range(100): + os.link(self.cache / "big", self.cache / f"link{index}") + kept = self.cache / "kept" + kept.write_bytes(b"x" * 4096) + os.utime(kept, (0, 0)) + self.cfg["cache_budget_percent"] = 100 * 1e6 / watch.disk(self.cfg)[1] / watch.GB + watch.trim_caches(self.cfg) + self.assertTrue(kept.exists()) + + def test_the_watch_hook_reads_no_alert_through_a_planted_name(self): + outside = self.base / "outside.jsonl" + outside.write_text(json.dumps({"id": time.time_ns(), "stage": "warn", "text": "OUTSIDE"}) + "\n", encoding="utf-8") + self.root.mkdir() + (self.root / "alerts.jsonl").symlink_to(outside) + result = self._warn_hook("s1") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertNotIn("OUTSIDE", result.stdout) + + def test_the_watch_hook_keeps_no_history_row_read_through_a_planted_name(self): + outside = self.base / "outside.jsonl" + outside.write_text(json.dumps({"id": "outside"}) + "\n", encoding="utf-8") + self.root.mkdir() + (self.root / "queue-history.jsonl").symlink_to(outside) + result = self._warn_hook("s1") + self.assertEqual(result.returncode, 0, result.stderr) + kept = self.base / ".local" / "state" / "agents-council" / "queue-history.jsonl" + self.assertNotIn("outside", kept.read_text(encoding="utf-8") if kept.exists() else "") + + def test_cache_trim_counts_the_blocks_a_sparse_file_holds(self): + watch = _load() + with (self.cache / "sparse").open("wb") as handle: + handle.truncate(10**12) + kept = self.cache / "kept" + kept.write_bytes(b"x" * 4096) + os.utime(kept, (0, 0)) + watch.trim_caches(self.cfg) + self.assertTrue(kept.exists()) + + def test_cleanup_survives_a_stale_lock_link_and_removes_nothing(self): + _, victims = self._victims() + self.root.mkdir() + lock = self.root / "cleanup.lock" + lock.symlink_to(victims) + os.utime(victims, (0, 0)) + os.utime(lock, (0, 0), follow_symlinks=False) + result = self._run("cleanup", "--caches") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue(victims.is_dir()) + self.assertTrue(lock.is_symlink()) + def test_without_config_every_command_is_a_no_op(self): self.config.unlink() result = self._run("status")