fix(kvb-org-rasci): open the register and its lock without following links

The council root is writable by every agent, and operator.py runs
rasci.database() outside the sandbox. A symlinked rasci.json made it
read any JSON file the operator can read and write it back into the
council as a regular file; a symlinked rasci.lock made it create files
at the link target; a FIFO hung every register call.

load() now reads through nofollow.read_text and database() locks and
writes through nofollow.locked and nofollow.replace. A register that is
no object of task objects, or nested past the recursion limit, exits 2
instead of a traceback. nofollow.open_below also refuses a file with
more than one link, so a hard link to a file outside the council is
refused like a symlink.

🎯 Fix: 96 %
💥 Break: 3 %
💸 Cost: 99 %
⏱️ Time: 95 %
🛡️ Security: 100 %
🏆 Score: 99
🚦 Gate: ✅ 90/5, cost and time <= 100, security 100 %

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-07 16:38:28 +00:00
1 parent 13b360227b
commit 8e180be256
4 files changed
+98 -20

No files matched your search

+11
View File
@@ -100,6 +100,17 @@ class TestNofollow(unittest.TestCase):
write(link)
self.assertFalse(dangling.exists())
def test_a_hard_linked_name_is_refused_and_its_target_kept(self):
followers = {name: write for name, write in self.writes(self.root).items() if name not in ("replace", "create")}
for name, write in followers.items():
with self.subTest(write=name):
link = self.root / name
os.link(self.victim, link)
with self.assertRaises(OSError):
write(link)
self.assertEqual(self.victim.read_text(encoding="utf-8"), "keep")
link.unlink()
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")
+53 -1
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import datetime
import json
import os
import subprocess
import sys
import tempfile
@@ -35,7 +36,7 @@ class TestRasci(unittest.TestCase):
json.dumps({"sessionId": sid, "name": name}), encoding="utf-8"
)
def _run(self, *args: str, script: Path = RASCI, sid: str = "s1") -> subprocess.CompletedProcess:
def _run(self, *args: str, script: Path = RASCI, sid: str = "s1", timeout: float | None = None) -> subprocess.CompletedProcess:
env = {
"PATH": "/usr/bin:/bin",
"HOME": str(self.home),
@@ -49,6 +50,7 @@ class TestRasci(unittest.TestCase):
env=env,
cwd=self.tmp.name,
check=False,
timeout=timeout,
)
def _out(self, *args: str) -> list[str]:
@@ -333,6 +335,56 @@ class TestRasci(unittest.TestCase):
with self.subTest(args=args):
self.assertEqual(self._run(*args).returncode, 2)
def test_a_symlinked_register_is_refused_and_never_copied_into_the_council(self):
outside = Path(self.tmp.name) / "outside.json"
outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8")
self.root.mkdir()
(self.root / "rasci.json").symlink_to(outside)
for args in (("show", "--all"), ("set", "work/a", "--title", "A")):
with self.subTest(args=args):
result = self._run(*args, timeout=60)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertNotIn("outside", result.stdout)
self.assertTrue((self.root / "rasci.json").is_symlink())
copies = [p.name for p in self.root.iterdir() if p.is_file() and not p.is_symlink()
and "marker-outside" in p.read_text(encoding="utf-8")]
self.assertEqual(copies, [])
def test_a_hard_linked_register_is_refused(self):
outside = Path(self.tmp.name) / "outside.json"
outside.write_text(json.dumps({"outside": {"title": "marker-outside"}}), encoding="utf-8")
self.root.mkdir()
os.link(outside, self.root / "rasci.json")
result = self._run("show", "--all", timeout=60)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertNotIn("outside", result.stdout)
def test_a_symlinked_lock_is_refused_and_creates_nothing_at_its_target(self):
aimed = Path(self.tmp.name) / "elsewhere" / "aimed"
aimed.parent.mkdir()
self.root.mkdir()
(self.root / "rasci.lock").symlink_to(aimed)
result = self._run("set", "work/a", "--title", "A", timeout=60)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertFalse(aimed.exists())
self.assertFalse((self.root / "rasci.json").exists())
def test_a_fifo_register_fails_instead_of_hanging(self):
self.root.mkdir()
os.mkfifo(self.root / "rasci.json")
for args in (("show",), ("set", "work/a")):
with self.subTest(args=args):
self.assertEqual(self._run(*args, timeout=20).returncode, 2)
def test_a_register_that_is_no_object_of_tasks_exits_2(self):
self.root.mkdir()
for text in ("[1, 2]", json.dumps({"work/a": 3}), "[" * 100000):
with self.subTest(text=text[:12]):
(self.root / "rasci.json").write_text(text, encoding="utf-8")
result = self._run("show", "--all", timeout=60)
self.assertEqual(result.returncode, 2, result.stderr)
self.assertIn("rasci:", result.stderr)
class TestOrgSkillsUseTheRegister(unittest.TestCase):
def test_every_org_skill_names_the_register(self):