diff --git a/sage/gateway/being_gate_client.py b/sage/gateway/being_gate_client.py index 416d09582..bf85b3b82 100644 --- a/sage/gateway/being_gate_client.py +++ b/sage/gateway/being_gate_client.py @@ -1713,6 +1713,10 @@ def run_grant_mounts(ctx: Optional[dict] = None) -> tuple: ctx = ctx or {} home = os.path.realpath(ctx.get("memory_root") or "/nonexistent-home") found, skipped = {}, [] # real path -> (kind, "rw"|"ro") + # hestia #1386 (part C): the nearest grant decides, so a child grant stricter than the mount + # above it must be mounted after it, not folded into it. Off unless the gate decides that way. + nearest = (ctx or {}).get("reach_rule") == "nearest" + unreadable = [] for item in ctx.get("granted_reach") or (): try: root, recursive = item[0], item[1] @@ -1724,6 +1728,8 @@ def run_grant_mounts(ctx: Optional[dict] = None) -> tuple: if "r" not in modes: skipped.append((os.path.abspath(os.path.expanduser(root.strip())), f"your grant there is '{modes}', without r (read)")) + if nearest and recursive: + unreadable.append(os.path.realpath(os.path.abspath(os.path.expanduser(root.strip())))) continue how = "rw" if ("w" in modes and "d" in modes) else "ro" lexical = os.path.abspath(os.path.expanduser(root.strip())) @@ -1753,9 +1759,19 @@ def run_grant_mounts(ctx: Optional[dict] = None) -> tuple: # De-nesting by the union rule: a mount inside a directory mount adds nothing unless it # is WIDER there (rw inside ro). Kept ones bind after their parent (sorted), so they win. dirs = [(p, h) for p, (k, h) in found.items() if k == "dir"] - mounts = sorted((p, k, h) for p, (k, h) in found.items() - if not any(p != d and p.startswith(d + os.sep) and (dh == "rw" or h == "ro") - for d, dh in dirs)) + if nearest: + # Keep a child unless it grants exactly what the mount above it already grants. + mounts = sorted((p, k, h) for p, (k, h) in found.items() + if not any(p != d and p.startswith(d + os.sep) and dh == h for d, dh in dirs)) + # A recursive grant without r beneath a mounted directory: the parent would show it, the + # gate would not. Masked with an empty read-only directory, like the home. + mounts += sorted((u, "mask", "ro") for u in unreadable + if os.path.isdir(u) and any(u.startswith(d + os.sep) for d, _ in dirs)) + mounts.sort() + else: + mounts = sorted((p, k, h) for p, (k, h) in found.items() + if not any(p != d and p.startswith(d + os.sep) and (dh == "rw" or h == "ro") + for d, dh in dirs)) mask_home = any(k == "dir" and home.startswith(p + os.sep) for p, k, _h in mounts) return mounts, sorted(set(skipped)), mask_home @@ -1824,8 +1840,10 @@ def run_command(args: dict, ctx: Optional[dict] = None) -> str: # symlink inside a granted tree that points into the home resolves inside the sandbox, to # the mask. mounts, _skipped, mask_home = run_grant_mounts(ctx) - grant_binds = "".join(f" {'--bind' if h == 'rw' else '--ro-bind'} {shlex.quote(p)} {shlex.quote(p)}" - for p, _k, h in mounts) + grant_binds = "".join( + f" --tmpfs {shlex.quote(p)} --remount-ro {shlex.quote(p)}" if k == "mask" else + f" {'--bind' if h == 'rw' else '--ro-bind'} {shlex.quote(p)} {shlex.quote(p)}" + for p, k, h in mounts) if mask_home: _qh = shlex.quote(_home) grant_binds += f" --tmpfs {_qh} --remount-ro {_qh}" @@ -3261,6 +3279,46 @@ def _granted_roots(core, policy, workspace: str, granted_paths=()) -> tuple: return () + +def _floor_read_reach(core, policy, workspace: str, own) -> tuple: + """``((root, recursive, "r"), ...)`` for the society's READ-ONLY floor this member's gate is + served (`society_floor_read` -> policy.read_scope, hestia #1381), so the dispatcher's own + confinement admits the reads the gate admits there. + + Measured 2026-10-25 (cbp-being): dp put shared-context/forum on the floor with modes rw, which + the daemon serves read-only. The gate admitted a read there; this list carried only the + member's own grants, so the dispatcher would have refused the same read "outside your reach". + + THE FLOOR APPLIES ONLY WHERE NO OWN GRANT REACHES (dp, 2026-10-23/24; hestia #1386): a floor + root at or beneath one of the member's own grant roots is left out, so the member's grant + decides there. Only with a mode-aware core (the third element says "r"); an older consumer + reads a pair as all modes, so nothing is added for it.""" + import os + if getattr(core, "scope_roots_with_modes", None) is None: + return () + entries = list(getattr(policy, "read_scope", ()) or ()) + if not entries: + return () + ws = os.path.realpath(os.path.expanduser(workspace)).replace("\\", "/").rstrip("/") + roots = [(str(r), bool(rec)) for r, rec in core._scope_roots_with_reach(entries, workspace)] + roots += [(os.path.realpath(os.path.join(ws, e)).replace("\\", "/").rstrip("/"), True) + for e in entries if isinstance(e, str) and e and not e.startswith("path:") + and "/" not in e and "\\" not in e and e not in (".", "..", "*")] + covered = [(str(it[0]).rstrip("/"), bool(it[1])) for it in own] + out = [] + for root, rec in roots: + # Suppressed only where an own grant reaches EVERYTHING the floor entry reaches (nomad's + # counterexample, Nova's HOLD on #533): an own RECURSIVE grant at or above the floor's root + # covers its whole subtree; an own EXACT grant covers only its one path, so it suppresses + # an exact floor entry on that same path and never a recursive one (whose children it does + # not reach). + if any((orec and (root == r or root.startswith(r + "/"))) + or (not orec and not rec and root == r) + for r, orec in covered): + continue + out.append((root, rec, "r")) + return tuple(out) + def _granted_reach(core, policy, workspace: str, granted_paths=()) -> tuple: """``((root, recursive), ...)`` for every path grant, via the core's own reach resolver (hestia_gate_core._scope_roots_with_reach, since #1002: exact unless spelled `/**`), @@ -3271,6 +3329,15 @@ def _granted_reach(core, policy, workspace: str, granted_paths=()) -> tuple: if policy is None: return () try: + # A CORE THAT ENFORCES MODES (hestia #1345 part B) says which grants it admits and in + # which modes, narrowed ones included, as (root, recursive, modes). Taken whole, so the + # dispatcher confines by exactly the modes the gate judged by. + with_modes = getattr(core, "scope_roots_with_modes", None) + if with_modes is not None: + scopes = list(getattr(policy, "scope", ()) or ()) + own = (tuple((str(r), bool(rec), str(md)) for r, rec, md in with_modes(policy, workspace)) + + tuple((r, True) for r in _repo_name_roots(core, scopes, workspace, granted_paths))) + return own + _floor_read_reach(core, policy, workspace, own) reach = getattr(core, "_scope_roots_with_reach", None) if reach is not None: scopes = list(getattr(policy, "scope", ()) or ()) @@ -3314,6 +3381,11 @@ class GatewayVerdict: # said the invariant was being checked; nothing was. Found while writing patch_apply, # which had copied the same pattern faithfully enough to inherit the same hole. command: Optional[str] = None + # WHICH RULE THE GATE DECIDED BY, so the dispatcher confines by the same one (hestia #1386): + # "nearest" when the core exports nearest_grant_modes (part C: the deepest grant covering a path + # decides it), "union" otherwise (part B: any covering grant admits). Read once, here, from the + # core this verdict was judged with; the dispatcher never guesses it. + reach_rule: str = "union" @property def blocks(self) -> bool: @@ -3534,7 +3606,14 @@ def _compose_ctx(self, effector: Optional[str] = None) -> dict: "member": getattr(self, "member_id", None), # where `camera` gets a frame (instance.json camera_source; "cortex" or absent) "camera_source": getattr(self, "camera_source", None), - **({"granted_reach": self._policy_grants()[1]} if effector == "run" else {})} + **({"granted_reach": self._policy_grants()[1], "reach_rule": self._reach_rule()} + if effector == "run" else {})} + + def _reach_rule(self) -> str: + """"nearest" when this seat's gate core decides by the nearest explicit grant (hestia #1386, + part C: it exports nearest_grant_modes), else "union" (part B). See GatewayVerdict.""" + return "nearest" if callable(getattr(getattr(self, "_core", None), "nearest_grant_modes", + None)) else "union" # -- gate one intent (intent -> verdict), fail-closed -------------------- def _policy_grants(self) -> tuple: @@ -3650,7 +3729,8 @@ def gate(self, intent: BeingIntent) -> GatewayVerdict: granted, granted_reach = self._policy_grants() if dec != "deny" else ((), ()) return GatewayVerdict(dec, rule, getattr(d, "reason", "") or ("ok" if dec != "deny" else ""), innate=False, stage="single-gate", command=judged_command, - granted=granted, granted_reach=granted_reach) + granted=granted, granted_reach=granted_reach, + reach_rule=self._reach_rule()) except Exception as e: # a gate that raises is a refused act, never an ungoverned one return GatewayVerdict("deny", "gate.raised", innate=True, stage="single-gate", reason=f"{type(e).__name__}: {e}") @@ -3730,7 +3810,8 @@ def gate(self, intent: BeingIntent) -> GatewayVerdict: # than recomposed by the caller, which is the whole point of the field. return GatewayVerdict(v.decision, v.rule, v.reason or "ok", v.innate, stage="local-law", command=judged_command, - granted=granted, granted_reach=granted_reach) + granted=granted, granted_reach=granted_reach, + reach_rule=self._reach_rule()) # -- the F1a seam: gate, then dispatch, then consume the result ---------- def _resolve_relative_to_grants(self, intent: BeingIntent): diff --git a/sage/gateway/hestia_dispatch.py b/sage/gateway/hestia_dispatch.py index 40e59e486..2ea460693 100644 --- a/sage/gateway/hestia_dispatch.py +++ b/sage/gateway/hestia_dispatch.py @@ -3008,7 +3008,7 @@ def _do_reach_search(self, intent: BeingIntent) -> ResultEnvelope: reach = _granted_reach_of(getattr(self, "_verdict", None)) inside = target is not None and any( target == os.path.realpath(r) or (rec and target.startswith(os.path.realpath(r) + os.sep)) - for r, rec, *_ in reach) + for r, rec, *md in reach if _mode_has(md, "r")) if not inside: return ResultEnvelope(ok=False, error=( f"search refused: {target or 'that path'} is outside the reach you hold. " @@ -3284,7 +3284,9 @@ def _do_run(self, intent: BeingIntent) -> ResultEnvelope: "workspace": getattr(self, "workspace", None), # the grants this verdict carries: run_grant_mounts composes the same mounts the # law judged from the gate's copy of them, or the line differs and nothing runs - "granted_reach": _granted_reach_of(getattr(self, "_verdict", None))} + "granted_reach": _granted_reach_of(getattr(self, "_verdict", None)), + # the rule the gate decided by (hestia #1386), so both sides compose one line + "reach_rule": getattr(getattr(self, "_verdict", None), "reach_rule", "union")} try: cmd = run_command(intent.args, ctx) names = run_names(intent.args, ctx) @@ -3302,8 +3304,8 @@ def _do_run(self, intent: BeingIntent) -> ResultEnvelope: f"record is written, and a run must not outlive its beat. Ask again next beat; " f"the file is unchanged and nothing of yours was touched.")) root = os.path.realpath(self.memory_root) - granted = [(os.path.realpath(r), rec) for r, rec, *_ in - _granted_reach_of(getattr(self, "_verdict", None))] + granted = [(os.path.realpath(r), rec) for r, rec, *md in + _granted_reach_of(getattr(self, "_verdict", None)) if _mode_has(md, "r")] def _allowed(full: str, is_script: bool) -> bool: if full == root or full.startswith(root + os.sep): @@ -4744,6 +4746,12 @@ def _release_stage_lock(lock_path: str, fd: int) -> None: os.close(fd) +def _mode_has(rest, mode: str) -> bool: + """Whether a reach item's optional third element (its modes, hestia #1345 part B) includes + `mode`. An item without one is a full-mode grant, as every grant was before modes.""" + return mode in (str(rest[0]) if rest else "rwed") + + def _is_reach_pointer(raw: str) -> bool: """`hestia://reach` or `hestia://reach/` (SAGE #530).""" s = (raw or "").strip() diff --git a/sage/gateway/reference_f1a.py b/sage/gateway/reference_f1a.py index 1c8694b3e..20a4a85b3 100644 --- a/sage/gateway/reference_f1a.py +++ b/sage/gateway/reference_f1a.py @@ -887,6 +887,21 @@ def __call__(self, intent: BeingIntent, verdict: GatewayVerdict) -> ResultEnvelo reach = tuple(getattr(verdict, "granted_reach", ()) or ()) self._extra_reach = (tuple((Path(r).resolve(), bool(rec)) for r, rec, *_ in reach) if reach else tuple((r, True) for r in self._extra_roots)) + # AND IN WHICH MODES (hestia #1345 part B): a grant narrowed below rwed carries them as a + # third element. A root without one is full-mode, as every grant was before modes. + # Keyed on (root, recursive), and two grants on the same key UNION their modes, as the + # gate does ("the law's reach is their union", run_grant_mounts): keyed on the root alone, + # a full `/f/**` and an `r` grant on `/f` collapsed to whichever came last, and a write the + # gate admitted was refused here (nomad review of #533, point 1). A grant without modes + # is full-mode. + self._extra_modes = {} + for r, rec, *md in reach: + k = (Path(r).resolve(), bool(rec)) + got = self._extra_modes.get(k, "") + (str(md[0]) if md else "rwed") + self._extra_modes[k] = "".join(c for c in "rwed" if c in got) + # And BY WHICH RULE (hestia #1386): "nearest" when the gate decided by the nearest + # explicit grant (part C), so this confinement decides the same way. + self._reach_rule = getattr(verdict, "reach_rule", "union") handler = getattr(self, f"_do_{intent.effector}", None) if handler is None: # a consequential network act the reference won't run — real F1a's job @@ -969,10 +984,36 @@ def _safe_path(self, raw: str, writing: bool = False) -> Path: f"notes/{p.name} is what was said TO you, and it stays as it was said. Your " "reply belongs in your journal, in a conversation with `say`, or in an " "appeal, all of which are read") - reach =((self.memory_root, True),) + tuple( - getattr(self, "_extra_reach", None) - or ((r, True) for r in (getattr(self, "_extra_roots", ()) or ()))) + held = tuple(getattr(self, "_extra_reach", None) + or ((r, True) for r in (getattr(self, "_extra_roots", ()) or ()))) + # The dispatcher confines by the same modes the gate judged by: reading needs r, writing + # w. A grant without the mode is not reach for this act (the home is always all four). + need = "w" if writing else "r" + modes = getattr(self, "_extra_modes", None) or {} + if (getattr(self, "_reach_rule", "union") == "nearest" + and not (p == self.memory_root or self.memory_root in p.parents)): + near = self._nearest_modes(p, held, modes) + if near is not None: + if need in near: + return p + raise ValueError( + f"'{p}' is under a grant of yours narrowed to {near}, the nearest grant covering " + f"it, and {'writing' if writing else 'reading'} needs {need}. A broader grant " + f"above it does not apply here: the nearest explicit grant decides (hestia " + f"#1386). Nothing was {'changed' if writing else 'read'}. memory_read " + f"hestia://reach/{p} shows what you hold there") + reach = ((self.memory_root, True),) + tuple( + (r, rec) for r, rec in held if need in modes.get((r, rec), "rwed")) if not any(p == r or (recursive and r in p.parents) for r, recursive in reach): + narrowed = [modes[(r, rec)] for r, rec in held + if (r, rec) in modes and need not in modes[(r, rec)] + and (p == r or (rec and r in p.parents))] + if narrowed: + raise ValueError( + f"'{p}' is under a grant of yours narrowed to {narrowed[0]}, and " + f"{'writing' if writing else 'reading'} needs {need}. Nothing was " + f"{'changed' if writing else 'read'}. memory_read hestia://reach/{p} shows " + f"what you hold there") # A NEAR MISS OF ITS OWN HOUSE IS NOT A TRESPASS. dp, 2026-09-25, fleet directive: # "each being has a home ... i want each to feel at home there ... addressing # unnecessary frictions." @@ -1009,12 +1050,36 @@ def _safe_path(self, raw: str, writing: bool = False) -> Path: def _reach_pairs(self) -> tuple: """((root, recursive), ...) beyond the home for this verdict. `_extra_reach` when the - verdict carried reach; otherwise the bare roots, read by prefix as _safe_path reads them.""" - return tuple(getattr(self, "_extra_reach", None) or ( + verdict carried reach; otherwise the bare roots, read by prefix as _safe_path reads them. + Only grants that include r (#1345 part B): every caller names a place to READ.""" + modes = getattr(self, "_extra_modes", None) or {} + return tuple((r, rec) for r, rec in (getattr(self, "_extra_reach", None) or ( (r, True) for r in (getattr(self, "_extra_roots", ()) or ()))) + if "r" in modes.get((r, rec), "rwed")) + + @staticmethod + def _nearest_modes(p: Path, held, modes) -> Optional[str]: + """The modes of the nearest held grant covering `p` (grants on one node union), or None. + The gate's rule (hestia #1386), over this verdict's (root, recursive) nodes.""" + best, got = -1, set() + for r, rec in held: + if p == r or (rec and r in p.parents): + depth = len(r.parts) + m = modes.get((r, rec), "rwed") + if depth > best: + best, got = depth, set(m) + elif depth == best: + got |= set(m) + return None if best < 0 else "".join(c for c in "rwed" if c in got) def _within_reach(self, p: Path) -> bool: p = p.resolve() + if getattr(self, "_reach_rule", "union") == "nearest" and not ( + p == self.memory_root or self.memory_root in p.parents): + held = tuple((Path(r).resolve(), rec) for r, rec in (getattr(self, "_extra_reach", None) or ( + (r, True) for r in (getattr(self, "_extra_roots", ()) or ())))) + near = self._nearest_modes(p, held, getattr(self, "_extra_modes", None) or {}) + return near is not None and "r" in near pairs = ((self.memory_root, True),) + tuple((Path(r).resolve(), rec) for r, rec in self._reach_pairs()) return any(p == r or (rec and r in p.parents) for r, rec in pairs) diff --git a/sage/gateway/tests/test_grant_modes_dispatch.py b/sage/gateway/tests/test_grant_modes_dispatch.py new file mode 100644 index 000000000..d934911a5 --- /dev/null +++ b/sage/gateway/tests/test_grant_modes_dispatch.py @@ -0,0 +1,271 @@ +"""hestia #1345 part B, SAGE's side: the dispatcher confines by the modes the gate judged by. + +A core that enforces modes exports scope_roots_with_modes(); _granted_reach passes the modes +through as a third element, and the dispatcher's own confinement requires r to read and w to +write. A run mounts every grant that holds r (`run_grant_mounts`: `--ro-bind` unless the grant +holds both w and d); the live single gate judges a run by its script path, never by that line +(hestia #1388 README, point 3).""" +import os +import sys +import tempfile +from types import SimpleNamespace + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))) +from sage.gateway.being_gate_client import BeingIntent, GatewayVerdict, _granted_reach # noqa: E402 +from sage.gateway.hestia_dispatch import _mode_has # noqa: E402 +from sage.gateway.reference_f1a import ReferenceF1aDispatcher # noqa: E402 + + +def _world(modes): + ws = os.path.realpath(tempfile.mkdtemp(prefix="modes-ws-")) + home = os.path.join(ws, "home") + forum = os.path.join(ws, "forum") + os.makedirs(home) + os.makedirs(forum) + open(os.path.join(forum, "thread.md"), "w").write("hello\n") + v = GatewayVerdict("allow", granted=(forum,), granted_reach=((forum, True, modes),)) + return ReferenceF1aDispatcher(memory_root=home), forum, v + + +def test_a_read_only_grant_reads_and_refuses_a_write_naming_the_mode(): + d, forum, v = _world("r") + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "thread.md")}), v) + assert r.ok and "hello" in (r.result or ""), r + r = d(BeingIntent("memory_write", {"path": os.path.join(forum, "new.md"), "content": "x"}), v) + assert not r.ok and "narrowed to r" in (r.error or "") and "writing needs w" in r.error, r + assert not os.path.exists(os.path.join(forum, "new.md")), "nothing may be written" + + +def test_a_write_only_grant_appends_and_refuses_a_read(): + d, forum, v = _world("w") + r = d(BeingIntent("memory_write", {"path": os.path.join(forum, "drop.md"), "content": "x"}), v) + assert r.ok, r + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "thread.md")}), v) + assert not r.ok or "hello" not in (r.result or ""), r + assert "narrowed to w" in (r.error or r.result or ""), r + + +def test_a_full_mode_or_modeless_grant_is_unchanged(): + for modes in ("rwed", None): + ws = os.path.realpath(tempfile.mkdtemp(prefix="modes-ws-")) + home, forum = os.path.join(ws, "home"), os.path.join(ws, "forum") + os.makedirs(home) + os.makedirs(forum) + item = (forum, True, modes) if modes else (forum, True) + v = GatewayVerdict("allow", granted=(forum,), granted_reach=(item,)) + d = ReferenceF1aDispatcher(memory_root=home) + r = d(BeingIntent("memory_write", {"path": os.path.join(forum, "n.md"), "content": "x"}), v) + assert r.ok, (modes, r) + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "n.md")}), v) + assert r.ok and "x" in r.result, (modes, r) + + +def test_the_home_is_always_all_four_modes(): + d, forum, v = _world("r") + r = d(BeingIntent("memory_write", {"path": "notes/x.md", "content": "y"}), v) + assert r.ok, r + + +def test_granted_reach_takes_the_cores_modes_when_it_has_them(): + pol = SimpleNamespace(scope=("path:/w/a/**",)) + core = SimpleNamespace(scope_roots_with_modes=lambda p, ws: (("/w/a", True, "rwed"), ("/w/f", True, "r")), + _scope_roots_with_reach=lambda s, ws: (("/w/a", True),)) + got = _granted_reach(core, pol, "/w") + assert ("/w/f", True, "r") in got and ("/w/a", True, "rwed") in got, got + # an older core without modes: today's two-element reading + core_old = SimpleNamespace(_scope_roots_with_reach=lambda s, ws: (("/w/a", True),)) + assert _granted_reach(core_old, pol, "/w") == (("/w/a", True),) + + +def test_mode_has_reads_a_missing_third_element_as_full(): + assert _mode_has([], "r") and _mode_has(["r"], "r") and not _mode_has(["w"], "r") + + +def test_two_grants_on_one_root_union_their_modes_as_the_gate_does(): + """nomad review of #533, point 1: keyed on the root alone, a full recursive grant and an `r` + grant on the same root collapsed to whichever came last, so a write the gate (union) admitted + was refused here. Keyed on (root, recursive) with modes unioned, the two agree.""" + ws = os.path.realpath(tempfile.mkdtemp(prefix="modes-ws-")) + home, forum = os.path.join(ws, "home"), os.path.join(ws, "forum") + os.makedirs(home) + os.makedirs(forum) + for reach in (((forum, True, "rwed"), (forum, True, "r")), + ((forum, True, "r"), (forum, True, "rwed")), + ((forum, True, "r"), (forum, True, "w"))): + v = GatewayVerdict("allow", granted=(forum,), granted_reach=reach) + d = ReferenceF1aDispatcher(memory_root=home) + r = d(BeingIntent("memory_write", {"path": os.path.join(forum, "u.md"), "content": "x"}), v) + assert r.ok, (reach, r) + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "u.md")}), v) + assert r.ok and "x" in r.result, (reach, r) + # An exact r grant on the root and a recursive w grant on it are different keys. + v = GatewayVerdict("allow", granted=(forum,), granted_reach=((forum, False, "r"), (forum, True, "w"))) + d = ReferenceF1aDispatcher(memory_root=home) + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "u.md")}), v) + assert not r.ok or "x" not in (r.result or ""), r + + +# -- hestia #1386 (part C): the nearest explicit grant decides, exactly when the gate does ------- + +def _nested(rule, child_modes="r", parent_modes="rwed"): + ws = os.path.realpath(tempfile.mkdtemp(prefix="near-ws-")) + home, proj = os.path.join(ws, "home"), os.path.join(ws, "proj") + os.makedirs(home) + os.makedirs(os.path.join(proj, "sub")) + os.makedirs(os.path.join(proj, "other")) + open(os.path.join(proj, "sub", "t.md"), "w").write("child\n") + reach = ((proj, True, parent_modes), (os.path.join(proj, "sub"), True, child_modes)) + v = GatewayVerdict("allow", granted=(proj, os.path.join(proj, "sub")), granted_reach=reach, + reach_rule=rule) + return ReferenceF1aDispatcher(memory_root=home), proj, v + + +def test_under_the_union_rule_a_nested_narrowed_grant_is_part_b_parity(): + """Part B's gate admits a write under an `r` child of a full parent; so must this.""" + d, proj, v = _nested("union") + r = d(BeingIntent("memory_write", {"path": os.path.join(proj, "sub", "n.md"), "content": "x"}), v) + assert r.ok, r + + +def test_under_the_nearest_rule_the_nearer_narrowed_grant_decides(): + d, proj, v = _nested("nearest") + r = d(BeingIntent("memory_write", {"path": os.path.join(proj, "sub", "n.md"), "content": "x"}), v) + assert not r.ok and "nearest grant covering" in (r.error or "") and "#1386" in r.error, r + assert not os.path.exists(os.path.join(proj, "sub", "n.md")) + r = d(BeingIntent("memory_read", {"path": os.path.join(proj, "sub", "t.md")}), v) + assert r.ok and "child" in (r.result or ""), r + r = d(BeingIntent("memory_write", {"path": os.path.join(proj, "other", "n.md"), "content": "x"}), v) + assert r.ok, r + + +def test_under_the_nearest_rule_a_write_only_child_cannot_be_read_through_its_parent(): + d, proj, v = _nested("nearest", child_modes="w", parent_modes="r") + r = d(BeingIntent("memory_read", {"path": os.path.join(proj, "sub", "t.md")}), v) + assert not r.ok or "child" not in (r.result or ""), r + assert "narrowed to w" in (r.error or r.result or ""), r + + +def test_the_rule_comes_from_the_core_the_verdict_was_judged_with(): + from sage.gateway.being_gate_client import BeingGateClient + c = BeingGateClient.__new__(BeingGateClient) + c._core = SimpleNamespace() + assert c._reach_rule() == "union" + c._core = SimpleNamespace(nearest_grant_modes=lambda *a, **k: None) + assert c._reach_rule() == "nearest" + + +def test_run_mounts_keep_a_stricter_child_and_mask_an_unreadable_one_under_the_nearest_rule(): + from sage.gateway.being_gate_client import run_grant_mounts + ws = os.path.realpath(tempfile.mkdtemp(prefix="near-mnt-")) + home, proj = os.path.join(ws, "home"), os.path.join(ws, "proj") + for dd in (home, os.path.join(proj, "ro"), os.path.join(proj, "hidden")): + os.makedirs(dd) + reach = ((proj, True, "rwed"), (os.path.join(proj, "ro"), True, "r"), + (os.path.join(proj, "hidden"), True, "w")) + union, _s, _m = run_grant_mounts({"memory_root": home, "granted_reach": reach}) + assert union == [(proj, "dir", "rw")], union + near, _s, _m = run_grant_mounts({"memory_root": home, "granted_reach": reach, + "reach_rule": "nearest"}) + assert (proj, "dir", "rw") in near and (os.path.join(proj, "ro"), "dir", "ro") in near, near + assert (os.path.join(proj, "hidden"), "mask", "ro") in near, near + assert near.index((proj, "dir", "rw")) < near.index((os.path.join(proj, "ro"), "dir", "ro")), \ + "a child's bind must come after its parent's, or the parent's bind hides it" + + +# -- the read-only floor reaches the dispatcher (dp, 2026-10-25: forum on the floor, modes rw) ---- + +def _core_with_modes(): + def _roots(entries, ws): + out = [] + for e in entries: + if isinstance(e, str) and e.startswith("path:"): + raw = e[len("path:"):] + rec = raw.endswith("/**") + out.append((raw[:-3] if rec else raw, rec)) + return tuple(out) + + def _with_modes(policy, ws): + return tuple((r, rec, "rwed") for r, rec in _roots(policy.scope, ws)) + return SimpleNamespace(scope_roots_with_modes=_with_modes, _scope_roots_with_reach=_roots, + _scope_parts=lambda s, w: ((), ())) + + +def test_a_read_only_floor_entry_is_reach_for_reading_and_never_for_writing(): + ws = os.path.realpath(tempfile.mkdtemp(prefix="floor-ws-")) + home, forum = os.path.join(ws, "home"), os.path.join(ws, "shared-context", "forum") + os.makedirs(home) + os.makedirs(forum) + open(os.path.join(forum, "note.md"), "w").write("from sprout\n") + policy = SimpleNamespace(scope=(), read_scope=(f"path:{forum}/**",)) + reach = _granted_reach(_core_with_modes(), policy, ws) + assert (forum, True, "r") in reach, reach + d = ReferenceF1aDispatcher(memory_root=home) + v = GatewayVerdict("allow", granted=tuple(r for r, *_ in reach), granted_reach=reach) + r = d(BeingIntent("memory_read", {"path": os.path.join(forum, "note.md")}), v) + assert r.ok and "from sprout" in (r.result or ""), r + r = d(BeingIntent("memory_write", {"path": os.path.join(forum, "x.md"), "content": "x"}), v) + assert not r.ok and not os.path.exists(os.path.join(forum, "x.md")), r + + +def test_the_floor_steps_aside_where_an_own_grant_reaches(): + ws = os.path.realpath(tempfile.mkdtemp(prefix="floor-ws-")) + sc = os.path.join(ws, "shared-context") + forum = os.path.join(sc, "forum") + os.makedirs(forum) + policy = SimpleNamespace(scope=(f"path:{sc}/**",), read_scope=(f"path:{forum}/**",)) + reach = _granted_reach(_core_with_modes(), policy, ws) + assert (sc, True, "rwed") in reach and not [x for x in reach if x[0] == forum], reach + + +def test_an_older_core_adds_no_floor_it_could_only_express_as_all_modes(): + ws = os.path.realpath(tempfile.mkdtemp(prefix="floor-ws-")) + forum = os.path.join(ws, "forum") + core = SimpleNamespace(_scope_roots_with_reach=lambda e, w: (), _scope_parts=lambda s, w: ((), ())) + policy = SimpleNamespace(scope=(), read_scope=(f"path:{forum}/**",)) + assert _granted_reach(core, policy, ws) == () + + +def _floor_world(): + ws = os.path.realpath(tempfile.mkdtemp(prefix="floor-ws-")) + home, forum = os.path.join(ws, "home"), os.path.join(ws, "shared-context", "forum") + os.makedirs(home) + os.makedirs(forum) + open(os.path.join(forum, "note.md"), "w").write("from sprout\n") + return ws, home, forum + + +def _dispatch_read_and_write(home, forum, reach): + d = ReferenceF1aDispatcher(memory_root=home) + v = GatewayVerdict("allow", granted=tuple(r for r, *_ in reach), granted_reach=reach) + read = d(BeingIntent("memory_read", {"path": os.path.join(forum, "note.md")}), v) + write = d(BeingIntent("memory_write", {"path": os.path.join(forum, "x.md"), "content": "x"}), v) + return read, write + + +def test_an_exact_own_grant_never_suppresses_a_recursive_floor_beneath_it(): + """nomad's counterexample (Nova HOLD on #533): an EXACT own grant on /f does not reach /f/child, + so it must not drop a RECURSIVE read floor on /f/** -- the gate admits that read.""" + ws, home, forum = _floor_world() + policy = SimpleNamespace(scope=(f"path:{forum}",), read_scope=(f"path:{forum}/**",)) + reach = _granted_reach(_core_with_modes(), policy, ws) + assert (forum, True, "r") in reach, reach + read, write = _dispatch_read_and_write(home, forum, reach) + assert read.ok and "from sprout" in (read.result or ""), read + assert not write.ok and not os.path.exists(os.path.join(forum, "x.md")), write + + +def test_a_recursive_own_grant_covers_the_floor_and_decides_there(): + ws, home, forum = _floor_world() + sc = os.path.dirname(forum) + policy = SimpleNamespace(scope=(f"path:{sc}/**",), read_scope=(f"path:{forum}/**",)) + reach = _granted_reach(_core_with_modes(), policy, ws) + assert not [x for x in reach if x[0] == forum], f"the own grant covers the floor's subtree: {reach}" + read, write = _dispatch_read_and_write(home, forum, reach) + assert read.ok and write.ok, (read, write) + + +def test_an_exact_own_grant_suppresses_only_an_exact_floor_on_the_same_path(): + ws, home, forum = _floor_world() + policy = SimpleNamespace(scope=(f"path:{forum}",), read_scope=(f"path:{forum}",)) + reach = _granted_reach(_core_with_modes(), policy, ws) + assert not [x for x in reach if x[0] == forum and len(x) > 2 and x[2] == "r"], reach