fix(microkernel): subprocess default-deny on credentials/paths/network — phase 0B task 0.9(d); ALL FIVE 0B REPAIRS COMPLETE

Contract ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker (headless admission,
approval user:phase-0B-0.9(d) dispatch).

Root cause: plugin-hook subprocesses inherited everything the parent had —
R2 probe replay proved a planted env secret and a host file readable,
sockets openable (I-R2-024 process_probe).

Discharge (finding -> test -> commit):
- finding: I-R2-024 process_probe (research/tui-frontier-r2)
- test: tests/test_microkernel_subprocess_default_deny.py — 11/13 RED on
  unfixed code (probe replay: env_visible=fixture-secret,
  file_visible=synthetic-host-only), 14/14 GREEN after; frozen forever;
  re-verified RED on reverted fix
- fix: scrubbed launch environment + pure-Python audit-hook supervisor
  (plugin_sandbox/sitecustomize.py: open/socket.connect/bind tripwires,
  propagates to grandchildren) — deny files outside the plugin dir, deny
  all networking. Supervised least-privilege per evidence prescription
  (honesty: Python-boundary supervision, not OS isolation).
- no-dead-end: every denial surfaces with the exact unblocking grant
  string (fire(..., grants={paths/env}) / CLI --grants / manifest field);
  three 0.9(c) tests legitimately needing access wired through grants —
  the fix working, not weakened.
- proof: 40/40 microkernel suites; 108 focused scope; repo tooling
  untouched (only fire() children sandboxed); CLI end-to-end (secret +
  HOME invisible by default; named grant delivers exactly that var);
  ruff clean. Scope named; 116 pre-existing failures unchanged.

Phase 0B task 0.9 = COMPLETE (a through e, all landed with frozen fixtures).
This commit is contained in:
Nick committed 2026-10-15 22:47:00 +01:00
1 parent 74a8ffd27d
commit c8926ef872
6 files changed
+798 -6

No files matched your search

+164 -3
View File
@@ -30,6 +30,7 @@ import copy
import hashlib
import json
import os
import re
import signal
import subprocess
import sys
@@ -51,7 +52,8 @@ _live_processes: dict = {}
# Closed manifest schema: anything outside this set is refused (I-R2-023
# lifecycle_threat_matrix install phase — "register accepts arbitrary
# manifest fields"). _file/_module are discovery's internal keys.
# manifest fields"). _file/_module are discovery's internal keys;
# "grants" is the 0.9(d) explicit-grant declaration.
_ALLOWED_MANIFEST_FIELDS = {
"name",
"version",
@@ -63,10 +65,24 @@ _ALLOWED_MANIFEST_FIELDS = {
"priority",
"auto_activate",
"triggers",
"grants",
"_file",
"_module",
}
# Supervised least-privilege subprocess defaults (I-R2-024 process_probe,
# phase 0B 0.9(d)): a fired hook inherits NOTHING from this process — no
# environment, no implicit paths, no network. Access arrives only via the
# explicit owner grant route (fire(..., grants=...) or manifest "grants"),
# and every denial surfaces in the hook result with the exact grant that
# unblocks it (no-dead-end law: never silently enforced).
SANDBOX_BOOT_DIR = os.path.join(SCRIPTS_DIR, "plugin_sandbox")
_SANDBOX_ENV_PASSTHROUGH = ("PATH", "LANG", "LC_ALL", "LC_CTYPE", "TZ", "TMPDIR")
_SANDBOX_DENIAL_RE = re.compile(
r"rune-sandbox: denied kind=(\S+) target=(.*?); unblock: (.+)"
)
_GRANT_KEYS = ("env", "paths", "network")
def now_iso():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
@@ -77,6 +93,92 @@ def _digest_file(fpath):
return hashlib.sha256(f.read()).hexdigest()
def _validate_grants(grants):
"""Typed validation of the explicit-grant surface (closed schema):
{"env": [names], "paths": [roots], "network": bool}."""
if grants in (None, {}):
return None
if not isinstance(grants, dict):
return "'grants' must be a dict: {'env': [...], 'paths': [...], 'network': bool}"
unknown = [k for k in grants if k not in _GRANT_KEYS]
if unknown:
return f"'grants' has unknown key(s): {', '.join(sorted(unknown))}"
for key in ("env", "paths"):
value = grants.get(key, [])
if not isinstance(value, list) or not all(
isinstance(x, str) and x.strip() for x in value
):
return f"'grants.{key}' must be a list of non-empty strings"
if "network" in grants and not isinstance(grants["network"], bool):
return "'grants.network' must be a boolean"
return None
def _merge_grants(manifest_grants, call_grants):
"""Owner override route: per-call grants union the plugin's declared
manifest grants. The default stays deny; grants only widen."""
merged = {"env": [], "paths": [], "network": False}
for source in (manifest_grants or {}, call_grants or {}):
merged["env"] = sorted(set(merged["env"]) | set(source.get("env", []) or []))
merged["paths"] = sorted(set(merged["paths"]) | set(source.get("paths", []) or []))
merged["network"] = merged["network"] or bool(source.get("network", False))
return merged
def _sandbox_env(fpath, grants):
"""Build the hook subprocess environment: nothing inherited.
Non-sensitive launch basics only (PATH/locale/temp), the sandbox
bootstrap on PYTHONPATH, and exactly the granted env vars, granted
path roots and network flag. Kernel-controlled names (PYTHONPATH,
RUNE_SANDBOX_*) can be widened by a grant but never replaced — a
grant must not be able to switch supervision off."""
plugin_dir = os.path.dirname(os.path.abspath(fpath))
roots = sorted(
{
os.path.realpath(p)
for p in [plugin_dir, *(grants.get("paths", []) or [])]
if p
}
)
env = {
"PYTHONPATH": SANDBOX_BOOT_DIR,
"PYTHONIOENCODING": "utf-8",
"RUNE_SANDBOX_ACTIVE": "1",
"RUNE_SANDBOX_ROOTS": os.pathsep.join(roots),
"RUNE_SANDBOX_WRITE_ROOTS": os.pathsep.join(roots),
"RUNE_SANDBOX_NETWORK": "1" if grants.get("network") else "0",
"RUNE_SANDBOX_HINT": (
"rune plugin sandbox: default-deny env/paths/network; owner grant "
"route: microkernel.fire(..., grants={'env': [...], 'paths': [...], "
"'network': True}) or manifest 'grants'"
),
}
for name in _SANDBOX_ENV_PASSTHROUGH:
value = os.environ.get(name)
if value is not None:
env[name] = value
for name in grants.get("env", []) or []:
if name == "PYTHONPATH" or name.startswith("RUNE_SANDBOX_"):
continue # kernel-controlled: widened below, never replaced
if name in os.environ:
env[name] = os.environ[name]
if "PYTHONPATH" in (grants.get("env", []) or []) and os.environ.get("PYTHONPATH"):
env["PYTHONPATH"] = SANDBOX_BOOT_DIR + os.pathsep + os.environ["PYTHONPATH"]
return env
def _parse_denial(stderr):
"""Extract the last rune-sandbox denial from a hook's stderr, so the
owner sees exactly what was denied and the one-call grant that
unblocks it (no-dead-end law)."""
matches = _SANDBOX_DENIAL_RE.findall(stderr or "")
if not matches:
return None
kind, target, grant = matches[-1]
return {"kind": kind, "target": target, "grant": grant}
def _validate_manifest(manifest):
"""Schema-validate a manifest; return an error string or None."""
if not isinstance(manifest, dict):
@@ -128,6 +230,9 @@ def _validate_manifest(manifest):
)
):
return "'triggers' must map names to booleans or handler names"
grants_error = _validate_grants(manifest.get("grants"))
if grants_error:
return grants_error
for field in ("_file", "_module"):
if field in manifest and not isinstance(manifest[field], str):
return f"'{field}' must be a string"
@@ -213,10 +318,21 @@ def register(manifest):
previous = registry["plugins"].get(name)
removed_capabilities = []
added_grants = {}
if previous is not None:
old_caps = previous.get("manifest", {}).get("capabilities", []) or []
removed_capabilities = sorted(set(old_caps) - set(manifest.get("capabilities", [])))
_deindex_plugin(registry, name)
old_grants = previous.get("manifest", {}).get("grants", {}) or {}
new_grants = manifest.get("grants", {}) or {}
for key in ("env", "paths"):
delta = sorted(
set(new_grants.get(key, []) or []) - set(old_grants.get(key, []) or [])
)
if delta:
added_grants[key] = delta
if new_grants.get("network") and not old_grants.get("network"):
added_grants["network"] = True
file_path = manifest.get("_file", "")
digest = _digest_file(file_path) if file_path and os.path.exists(file_path) else None
@@ -257,6 +373,9 @@ def register(manifest):
"name": name,
"capabilities": manifest.get("capabilities", []),
"removed_capabilities": removed_capabilities,
# 0.9(d) declared-need transparency: a re-registration that
# EXPANDS what the subprocess may touch is surfaced, not silent.
"added_grants": added_grants,
}
return {"status": "registered", "name": name, "capabilities": manifest.get("capabilities", [])}
@@ -344,7 +463,7 @@ def resolve(capability):
# ---------------------------------------------------------------------------
def fire(event, data=None, timeout=30):
def fire(event, data=None, timeout=30, grants=None):
"""Fire all hooks for an event.
Each hook runs as a subprocess — plugin code never executes in the
@@ -352,9 +471,21 @@ def fire(event, data=None, timeout=30):
Results are collected and returned.
Token efficiency: the kernel only sees the result, not the plugin code.
I-R2-024 / 0.9(d) default-deny: each subprocess starts with NOTHING
inherited — scrubbed environment, no paths outside its own plugin
dir, no network (supervised via the plugin_sandbox audit bootstrap).
The no-dead-end owner route is `grants`:
{"env": [names], "paths": [roots], "network": bool} — unioned with
the plugin's declared manifest "grants". Denials surface per-result
as status "denied" with the exact grant that unblocks them.
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
grants_error = _validate_grants(grants)
if grants_error:
raise ValueError(grants_error)
registry = _load_registry()
listeners = registry["hooks"].get(event, [])
@@ -372,6 +503,11 @@ def fire(event, data=None, timeout=30):
as its own session/process group so the WHOLE tree is killable, and
a timeout kills and verifies the entire group — not just the direct
child.
I-R2-024 / 0.9(d): the subprocess launches with the scrubbed
sandbox environment (default-deny env/paths/network, manifest
grants unioned with the per-call owner grants), and a sandbox
denial surfaces as status "denied" with the exact grant.
"""
plugin = registry["plugins"].get(name, {})
if not plugin.get("active", False):
@@ -389,6 +525,13 @@ def fire(event, data=None, timeout=30):
"error": "plugin file changed since registration (digest mismatch)",
}
effective = _merge_grants(plugin.get("manifest", {}).get("grants"), grants)
sandbox_summary = {
"env": effective["env"],
"paths": effective["paths"],
"network": effective["network"],
}
hook_start = datetime.now()
proc = None
try:
@@ -398,6 +541,7 @@ def fire(event, data=None, timeout=30):
stderr=subprocess.PIPE,
text=True,
start_new_session=True,
env=_sandbox_env(fpath, effective),
)
_live_processes.setdefault(name, {})[proc.pid] = now_iso()
try:
@@ -416,9 +560,20 @@ def fire(event, data=None, timeout=30):
"name": name,
"status": "ok",
"result": hook_result,
"sandbox": sandbox_summary,
"duration_ms": round(duration, 1),
}
else:
denied = _parse_denial(stderr)
if denied is not None:
return {
"name": name,
"status": "denied",
"denied": denied,
"sandbox": sandbox_summary,
"error": stderr.strip()[:200],
"duration_ms": round(duration, 1),
}
return {
"name": name,
"status": "error",
@@ -705,6 +860,11 @@ def main():
parser.add_argument("--capability", default=None, help="Capability to resolve")
parser.add_argument("--event", default=None, help="Event to fire")
parser.add_argument("--data", default=None, help="JSON data for hook")
parser.add_argument(
"--grants",
default=None,
help="Owner grant override JSON: {'env': [...], 'paths': [...], 'network': bool}",
)
parser.add_argument("--json", action="store_true", help="JSON output")
args = parser.parse_args()
@@ -740,7 +900,8 @@ def main():
print("Error: --event required")
sys.exit(1)
data = json.loads(args.data) if args.data else None
result = fire(args.event, data=data)
grants = json.loads(args.grants) if args.grants else None
result = fire(args.event, data=data, grants=grants)
print(
json.dumps(result, indent=2)
if args.json
+149
View File
@@ -0,0 +1,149 @@
"""plugin_sandbox/sitecustomize.py — supervised least-privilege bootstrap
for microkernel plugin subprocesses (I-R2-024 process_probe, phase 0B
repair 0.9(d)).
The kernel launches every plugin hook with this directory on PYTHONPATH,
so `site` imports this module at interpreter startup inside the
subprocess. It installs an audit hook that enforces the same
default-deny the scrubbed launch environment already applies to
credentials: PATH access outside the granted roots and NETWORK sockets
are denied unless the kernel granted them.
No-dead-end law: every denial raises PermissionError carrying the exact
grant that unblocks it ("grants={'paths': [...]}"/"grants={'network':
True}"); microkernel.fire() surfaces that message in the hook result so
the owner can grant it in one call — denials are never silent.
Honesty note (I-R2-024 decision): this is SUPERVISION at the Python
runtime boundary — audit events cover io.open/os.open/socket — not
OS-level isolation. It hardens the one supervised process tier; it does
not resurrect the killed sandbox-ladder claim.
"""
from __future__ import annotations
import os
import sys
_DENY_PREFIX = "rune-sandbox: denied"
def _real(path: str) -> str | None:
try:
return os.path.realpath(os.path.abspath(path))
except Exception: # noqa: BLE001 — best-effort normalization; denial polarity handles the rest
return None
def _roots(raw: str) -> list[str]:
out = []
for part in (raw or "").split(os.pathsep):
part = part.strip()
if part:
root = _real(part)
if root:
out.append(root)
return out
def _under(path: str, roots) -> bool:
return any(path == root or path.startswith(root + os.sep) for root in roots)
def install() -> None:
if os.environ.get("RUNE_SANDBOX_ACTIVE") != "1":
# Not a kernel-supervised process — this module must be inert
# anywhere else it might get imported.
return
import sysconfig
read_roots = set(_roots(os.environ.get("RUNE_SANDBOX_ROOTS", "")))
write_roots = set(_roots(os.environ.get("RUNE_SANDBOX_WRITE_ROOTS", "")))
network = os.environ.get("RUNE_SANDBOX_NETWORK") == "1"
# The interpreter's own installation stays readable, or nothing
# (including this supervision) could start.
for prefix in (
sys.prefix,
getattr(sys, "base_prefix", sys.prefix),
sys.exec_prefix or sys.prefix,
getattr(sys, "base_exec_prefix", sys.prefix),
os.path.dirname(os.path.abspath(sys.executable or "python")),
):
if prefix:
root = _real(prefix)
if root:
read_roots.add(root)
try:
paths = sysconfig.get_paths()
except Exception: # noqa: BLE001 — degraded install: stdlib roots still allowed above
paths = {}
for key in ("stdlib", "platstdlib", "purelib", "platlib"):
p = paths.get(key)
if p:
root = _real(p)
if root:
read_roots.add(root)
# No global temp-dir allowance: /tmp holds other processes' secrets
# and scratch state, so a plugin that needs temp storage takes the
# explicit grant route (RUNE_SANDBOX_ROOTS carries granted roots).
# Device sinks are baseline (not secrets): subprocess.DEVNULL opens
# /dev/null for write on every standard spawn.
devnull = _real(os.devnull)
if devnull:
read_roots.add(devnull)
write_roots.add(devnull)
write_bits = (
getattr(os, "O_WRONLY", 1)
| getattr(os, "O_RDWR", 2)
| getattr(os, "O_APPEND", 8)
| getattr(os, "O_CREAT", 64)
| getattr(os, "O_TRUNC", 512)
)
def _is_write(mode, flags) -> bool:
if isinstance(mode, str) and any(c in mode for c in "wax+"):
return True
return isinstance(flags, int) and bool(flags & write_bits)
def _deny(kind: str, target, grant: str):
raise PermissionError(
f"{_DENY_PREFIX} kind={kind} target={target}; unblock: {grant}"
)
def _audit(event, args):
try:
if event == "open":
path = args[0] if args else None
if not isinstance(path, (str, bytes, os.PathLike)):
return
if isinstance(path, bytes):
path = os.fsdecode(path)
elif isinstance(path, os.PathLike):
path = os.fspath(path)
real = _real(path)
if real is None:
return
if _is_write(
args[1] if len(args) > 1 else None,
args[2] if len(args) > 2 else None,
):
if not _under(real, write_roots):
_deny("path-write", path, f"grants={{'paths': [{path!r}]}}")
elif not _under(real, read_roots):
_deny("path-read", path, f"grants={{'paths': [{path!r}]}}")
elif event in ("socket.connect", "socket.bind", "socket.getaddrinfo"):
if not network:
_deny("network", event, "grants={'network': True}")
except PermissionError:
raise
except Exception: # noqa: BLE001 — a supervision bug must never
# silently open the sandbox; fail closed with a visible marker
_deny("sandbox-internal", event, "owner action required: report sandbox bug (no grant unblocks this)")
sys.addaudithook(_audit)
install()
@@ -0,0 +1,138 @@
{
"checkpoints": [
{
"at": "2026-10-15T21:25:24Z",
"evidence": "predicted-frontier",
"note": "{\"level0_direct\": [], \"level1_obligations\": [], \"contracts\": [], \"level2_verification\": []}",
"phase": "draft"
},
{
"at": "2026-10-15T21:25:39Z",
"evidence": "user:phase-0B-0.9(d) dispatch",
"note": "",
"phase": "approved"
},
{
"at": "2026-10-15T21:25:55Z",
"evidence": "builder:repair-worker",
"note": "",
"phase": "building"
}
],
"evidence_refs": [
"contract-created:ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker"
],
"ext": {
"activation": {
"activated": true,
"builder": "repair-worker",
"evidence": [
"change scoped to an existing project/target",
"target project: rune"
],
"lifecycle": "proportional",
"mode": "EXISTING_PROJECT_CHANGE",
"note": "attached to existing project authority; Need/completeness/patterns/contract flow governs proportionally",
"project_authority": "rune_internal",
"target_project": "rune"
},
"affected_paths": [
"src/rune/microkernel.py",
"src/rune/plugin_sandbox/sitecustomize.py",
"tests/test_microkernel_lifecycle_hardening.py",
"tests/test_microkernel_subprocess_default_deny.py"
],
"affected_regions": [],
"affected_ui_surfaces": [],
"expected_runtime_effects": [],
"predicted_frontier": {
"change_kind": "implementation_change",
"contextual_not_obligations": [],
"contracts": [],
"level0_direct": [],
"level1_obligations": [],
"level2_verification": [],
"max_level": 2,
"traversal_scope": "spine-structural-only"
},
"scope_provenance": {
"caller_added": [],
"generated_from_affected_surface": [
"src/rune/microkernel.py",
"src/rune/plugin_sandbox/sitecustomize.py",
"tests/test_microkernel_lifecycle_hardening.py",
"tests/test_microkernel_subprocess_default_deny.py"
]
},
"task_ref": "",
"unresolved_surfaces": [
{
"kind": "source_path",
"reason": "no authority records this path as an entity (path remains file-scope only)",
"surface": "src/rune/microkernel.py"
},
{
"kind": "source_path",
"reason": "no authority records this path as an entity (path remains file-scope only)",
"surface": "src/rune/plugin_sandbox/sitecustomize.py"
},
{
"kind": "source_path",
"reason": "no authority records this path as an entity (path remains file-scope only)",
"surface": "tests/test_microkernel_lifecycle_hardening.py"
},
{
"kind": "source_path",
"reason": "no authority records this path as an entity (path remains file-scope only)",
"surface": "tests/test_microkernel_subprocess_default_deny.py"
}
],
"workflow_config_surfaces": []
},
"id": "contract:ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker",
"identities": {
"acceptance": [
"bounded repair verified: fix subprocess default-deny on credentials/paths/network with no-dead-end resolution in src/rune/microkernel.py (I-R2-02"
],
"affected_entities": [],
"expected_consumers": [],
"expected_new_entities": [
"test:microkernel_lifecycle_hardening",
"test:microkernel_subprocess_default_deny"
],
"gap_evidence": "change scoped to an existing project/target; target project: rune",
"migration_obligations": [],
"need": "fix subprocess default-deny on credentials/paths/network with no-dead-end resolution in src/rune/microkernel.py (I-R2-024 process_probe: synthetic env secret and host file readable today)",
"next_action": "complete implementation, then verify",
"non_goals": [],
"origin": "worker",
"regression_obligations": [
"fixture stays RED against the unfixed kernel (revert src/rune/microkernel.py + remove src/rune/plugin_sandbox)"
],
"requirement": "headless admission (classified: EXISTING_PROJECT_CHANGE)",
"research_refs": [
"research/tui-frontier-r2"
],
"resolution": "",
"rollback_ref": "git revert",
"scope": [
"src/rune/microkernel.py",
"src/rune/plugin_sandbox/sitecustomize.py",
"tests/test_microkernel_lifecycle_hardening.py",
"tests/test_microkernel_subprocess_default_deny.py"
],
"toolchain": {},
"verification_obligations": [
"RUNE_CA_CORE=python RUNE_CM_CORE=python PYTHONPATH=src python -m pytest tests/test_microkernel_lifecycle_hardening.py tests/test_microkernel_fire_datetime.py -q",
"RUNE_CA_CORE=python RUNE_CM_CORE=python PYTHONPATH=src python -m pytest tests/test_microkernel_subprocess_default_deny.py -q",
"pytest tests/test_microkernel_lifecycle_hardening.py",
"pytest tests/test_microkernel_subprocess_default_deny.py"
]
},
"kind": "construction_contract",
"maturity": "DECLARED",
"observed_at": "2026-10-15T21:25:55Z",
"phase": "building",
"status": "open",
"tags": []
}
@@ -6835,3 +6835,5 @@
{"builder": "repair-worker", "detail": "ADMIT-2026-10-15T20:28:00Z-15eb44-repair-worker \u2192 building", "kind": "builder_session", "mode": "rune_internal", "observed_at": "2026-10-15T20:28:33Z", "operation": "transition"}
{"builder": "repair-worker", "detail": "ADMIT-2026-10-15T21:04:13Z-7ea89c-repair-worker \u2192 approved", "kind": "builder_session", "mode": "rune_internal", "observed_at": "2026-10-15T21:04:31Z", "operation": "transition"}
{"builder": "repair-worker", "detail": "ADMIT-2026-10-15T21:04:13Z-7ea89c-repair-worker \u2192 building", "kind": "builder_session", "mode": "rune_internal", "observed_at": "2026-10-15T21:04:46Z", "operation": "transition"}
{"builder": "repair-worker", "detail": "ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker \u2192 approved", "kind": "builder_session", "mode": "rune_internal", "observed_at": "2026-10-15T21:25:39Z", "operation": "transition"}
{"builder": "repair-worker", "detail": "ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker \u2192 building", "kind": "builder_session", "mode": "rune_internal", "observed_at": "2026-10-15T21:25:55Z", "operation": "transition"}
+13 -3
View File
@@ -320,7 +320,9 @@ class TestVerifiedProcessTreeTermination:
plugin.write_text(SPAWN_PLUGIN.format(pidfile=str(pidfile)), encoding="utf-8")
kernel.register({**BASE_MANIFEST, "_file": str(plugin)})
fired = kernel.fire("probe", {}, timeout=10)
# 0.9(d): subprocess paths are default-deny — the pidfile scratch
# dir reaches the plugin tree only via the explicit grant.
fired = kernel.fire("probe", {}, timeout=10, grants={"paths": [str(tmp_path)]})
assert fired["results"][0]["status"] == "ok", fired
child_pid = _child_pid(pidfile)
try:
@@ -337,7 +339,7 @@ class TestVerifiedProcessTreeTermination:
plugin = tmp_path / "plugin.py"
plugin.write_text(SPAWN_PLUGIN.format(pidfile=str(pidfile)), encoding="utf-8")
kernel.register({**BASE_MANIFEST, "_file": str(plugin)})
kernel.fire("probe", {}, timeout=10)
kernel.fire("probe", {}, timeout=10, grants={"paths": [str(tmp_path)]})
child_pid = _child_pid(pidfile)
try:
kernel.unregister("probe")
@@ -362,7 +364,15 @@ class TestVerifiedProcessTreeTermination:
monkeypatch.setenv("RUNE_TEST_SPIN", "1")
try:
result = kernel.fire("probe", {}, timeout=1.0)
# 0.9(d): env and paths are default-deny in fired subprocesses
# — the spin flag and pidfile scratch dir reach the plugin tree
# only via the explicit grants.
result = kernel.fire(
"probe",
{},
timeout=1.0,
grants={"env": ["RUNE_TEST_SPIN"], "paths": [str(tmp_path)]},
)
child_pid = _child_pid(pidfile)
statuses = [r["status"] for r in result["results"]]
assert "timeout" in statuses, result
@@ -0,0 +1,332 @@
"""Phase 0B repair 0.9(d) — subprocess default-deny on
credentials/paths/network (I-R2-024 process_probe).
Frozen regression fixture replaying the R2 probe
(research/tui-frontier-r2/receipts/targeted_extension_probe.py): a plugin
subprocess fired by microkernel.fire() could read a planted synthetic env
secret (RUNE_SYNTHETIC_SECRET) and a host-only file, and could bind a
socket — no default-deny on credentials, paths, or network
(targeted-extension-placement-evidence.json I-R2-024.process_probe:
credential_default_deny=false, path_default_deny=false,
network_default_deny=false).
Contract: ADMIT-2026-10-15T21:25:21Z-6c3df3-repair-worker.
Owner no-dead-end law: every deny pairs with a fast owner-resolution
route — per-call fire(..., grants=...) or declared manifest "grants" —
and every denial is SURFACED in the fire result with the exact grant
that unblocks it, never silently enforced.
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from rune import microkernel as mk # noqa: E402
@pytest.fixture()
def kernel(tmp_path, monkeypatch):
"""Isolated registry DB + clean process table, restored after."""
monkeypatch.setattr(mk, "REGISTRY_DB", str(tmp_path / "registry.json"))
table = getattr(mk, "_live_processes", None)
saved_table = dict(table) if table is not None else None
if table is not None:
table.clear()
yield mk
if table is not None:
table.clear()
if saved_table is not None:
table.update(saved_table)
@pytest.fixture()
def planted_secret(monkeypatch):
"""The R2 probe's synthetic environment secret."""
monkeypatch.setenv("RUNE_SYNTHETIC_SECRET", "fixture-secret")
@pytest.fixture()
def plugin_layout(tmp_path):
"""Plugin in its own dir + a host-only file OUTSIDE any granted root
(the probe's layout: plugin reads what it was never given)."""
plugin_dir = tmp_path / "plugindir"
plugin_dir.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
host_only = outside / "host-only.txt"
host_only.write_text("synthetic-host-only", encoding="utf-8")
return {"plugin_dir": plugin_dir, "host_only": host_only}
# The probe's plugin, verbatim in spirit: report what the environment
# and the host filesystem leaked to the subprocess.
ENV_LEAK_PLUGIN = (
"import json, os\n"
"print(json.dumps({'env_visible': os.environ.get('RUNE_SYNTHETIC_SECRET')}))\n"
)
FILE_LEAK_PLUGIN = """import json, pathlib, sys
target = sys.argv[1] if len(sys.argv) > 1 else None
try:
content = pathlib.Path(target).read_text()
print(json.dumps({'file_visible': content}))
except PermissionError as exc:
print(json.dumps({'file_visible': None, 'denied': str(exc)}))
"""
NETWORK_PLUGIN = """import json, socket
try:
s = socket.socket()
s.bind(("127.0.0.1", 0))
s.close()
print(json.dumps({'bound': True}))
except PermissionError as exc:
print(json.dumps({'bound': False, 'denied': str(exc)}))
"""
def _register(kernel, layout, plugin_text, *, name="probe", extra=None):
plugin = layout["plugin_dir"] / "plugin.py"
plugin.write_text(plugin_text, encoding="utf-8")
manifest = {
"name": name,
"version": "1",
"capabilities": ["cap.d"],
"hooks": {"probe": True},
"_file": str(plugin),
}
manifest.update(extra or {})
result = kernel.register(manifest)
assert "error" not in result, result
return plugin
def _hook_payload(kernel):
fired = kernel.fire("probe", {}, timeout=30)
assert fired["hooks_fired"] == 1, fired
return fired["results"][0]
# ---------------------------------------------------------------------------
# THE R2 PROBE REPLAY — red against the unfixed kernel
# ---------------------------------------------------------------------------
class TestR2ProbeReplay:
def test_synthetic_env_secret_not_inherited(self, kernel, planted_secret, plugin_layout):
"""I-R2-024: the subprocess must not inherit the environment —
the planted RUNE_SYNTHETIC_SECRET is invisible by default."""
_register(kernel, plugin_layout, ENV_LEAK_PLUGIN)
result = _hook_payload(kernel)
assert result["status"] == "ok", result
leaked = result["result"]["env_visible"]
assert leaked is None, f"synthetic env secret leaked to subprocess: {leaked!r}"
def test_host_only_file_not_readable(self, kernel, plugin_layout):
"""I-R2-024: a host file outside the plugin's granted roots must
not be readable from the subprocess (the probe baked the target
path into the plugin source; so does this replay)."""
host_only = plugin_layout["host_only"]
_register(
kernel, plugin_layout, FILE_LEAK_PLUGIN.replace("sys.argv[1]", repr(str(host_only)))
)
result = _hook_payload(kernel)
assert result["status"] == "ok", result
payload = result["result"]
assert payload["file_visible"] is None, "host-only file readable by subprocess"
assert "rune-sandbox" in payload.get("denied", ""), payload
def test_network_denied_by_default(self, kernel, plugin_layout):
"""I-R2-024: no socket bind/connect without an explicit network
grant."""
_register(kernel, plugin_layout, NETWORK_PLUGIN)
result = _hook_payload(kernel)
assert result["status"] == "ok", result
payload = result["result"]
assert payload["bound"] is False, "subprocess bound a socket with no grant"
assert "rune-sandbox" in payload.get("denied", ""), payload
# ---------------------------------------------------------------------------
# No-dead-end law — every deny pairs with a fast owner grant route
# ---------------------------------------------------------------------------
class TestGrantRoute:
def test_env_grant_unblocks_legitimate_need(self, kernel, planted_secret, plugin_layout):
"""A plugin that legitimately needs one variable receives it via
the explicit grant — the default stays deny for everything else."""
_register(kernel, plugin_layout, ENV_LEAK_PLUGIN)
fired = kernel.fire(
"probe", {}, timeout=30, grants={"env": ["RUNE_SYNTHETIC_SECRET"]}
)
result = fired["results"][0]
assert result["status"] == "ok", result
assert result["result"]["env_visible"] == "fixture-secret"
def test_path_grant_unblocks_legitimate_need(self, kernel, plugin_layout):
host_only = plugin_layout["host_only"]
_register(
kernel, plugin_layout, FILE_LEAK_PLUGIN.replace("sys.argv[1]", repr(str(host_only)))
)
fired = kernel.fire("probe", {}, timeout=30, grants={"paths": [str(host_only)]})
result = fired["results"][0]
assert result["status"] == "ok", result
assert result["result"]["file_visible"] == "synthetic-host-only"
def test_network_grant_unblocks_legitimate_need(self, kernel, plugin_layout):
_register(kernel, plugin_layout, NETWORK_PLUGIN)
fired = kernel.fire("probe", {}, timeout=30, grants={"network": True})
result = fired["results"][0]
assert result["status"] == "ok", result
assert result["result"]["bound"] is True
def test_manifest_declares_persistent_grants(self, kernel, plugin_layout):
"""The plugin's own declared need (recorded at registration) is
honored without per-call grants."""
host_only = plugin_layout["host_only"]
_register(
kernel,
plugin_layout,
FILE_LEAK_PLUGIN.replace("sys.argv[1]", repr(str(host_only))),
extra={"grants": {"paths": [str(host_only)]}},
)
result = _hook_payload(kernel)
assert result["status"] == "ok", result
assert result["result"]["file_visible"] == "synthetic-host-only"
def test_plugin_dir_stays_usable_without_grants(self, kernel, plugin_layout):
"""Default-deny must not dead-end normal plugin work: the
plugin's own directory remains readable/writable with NO grant."""
plugin = (
"import json, pathlib\n"
f"here = pathlib.Path({str(plugin_layout['plugin_dir'])!r}) / 'state.txt'\n"
"here.write_text('ok')\n"
"print(json.dumps({'state': here.read_text()}))\n"
)
_register(kernel, plugin_layout, plugin)
result = _hook_payload(kernel)
assert result["status"] == "ok", result
assert result["result"] == {"state": "ok"}
def test_scratch_dir_via_grant_route(self, kernel, plugin_layout, tmp_path):
"""A scratch dir outside the plugin dir is denied by default and
arrives via the explicit path grant — surfaced, one call."""
scratch = tmp_path / "scratch.txt"
plugin = (
"import json, pathlib\n"
f"pathlib.Path({str(scratch)!r}).write_text('ok')\n"
"print(json.dumps({'tmp': 'written'}))\n"
)
_register(kernel, plugin_layout, plugin)
denied = kernel.fire("probe", {}, timeout=30)["results"][0]
assert denied["status"] == "denied", denied
assert denied["denied"]["kind"] == "path-write"
granted = kernel.fire(
"probe", {}, timeout=30, grants={"paths": [str(tmp_path)]}
)["results"][0]
assert granted["status"] == "ok", granted
assert scratch.read_text() == "ok"
# ---------------------------------------------------------------------------
# Denial surfacing — never silent
# ---------------------------------------------------------------------------
class TestDenialSurfacing:
def test_denial_surfaces_with_exact_grant_hint(
self, kernel, plugin_layout, monkeypatch
):
"""An unhandled denial crashes the hook and the fire result
carries a machine-readable denial naming the exact grant."""
host_only = plugin_layout["host_only"]
plugin = (
"import pathlib\n"
f"pathlib.Path({str(host_only)!r}).read_text()\n"
)
_register(kernel, plugin_layout, plugin)
fired = kernel.fire("probe", {}, timeout=30)
result = fired["results"][0]
assert result["status"] == "denied", result
denied = result["denied"]
assert denied["kind"] == "path-read"
assert str(host_only) in denied["target"]
assert "grants=" in denied["grant"], denied
assert str(host_only) in denied["grant"], denied
def test_successful_result_carries_effective_grants(
self, kernel, planted_secret, plugin_layout
):
"""Transparency: what the subprocess actually received is part of
the result envelope, not a silent property."""
_register(kernel, plugin_layout, ENV_LEAK_PLUGIN)
fired = kernel.fire(
"probe", {}, timeout=30, grants={"env": ["RUNE_SYNTHETIC_SECRET"]}
)
result = fired["results"][0]
sandbox = result["sandbox"]
assert sandbox["env"] == ["RUNE_SYNTHETIC_SECRET"]
assert sandbox["network"] is False
# ---------------------------------------------------------------------------
# Closed manifest schema — grants are typed, not arbitrary
# ---------------------------------------------------------------------------
class TestManifestGrantSchema:
def test_register_rejects_malformed_grants(self, kernel, plugin_layout):
plugin = plugin_layout["plugin_dir"] / "plugin.py"
bad = kernel.register(
{
"name": "bad-grants",
"version": "1",
"capabilities": [],
"hooks": {"probe": True},
"_file": str(plugin),
"grants": {"env": "not-a-list"},
}
)
assert "error" in bad, bad
def test_register_rejects_unknown_grant_key(self, kernel, plugin_layout):
plugin = plugin_layout["plugin_dir"] / "plugin.py"
bad = kernel.register(
{
"name": "bad-grants",
"version": "1",
"capabilities": [],
"hooks": {"probe": True},
"_file": str(plugin),
"grants": {"subprocess": True},
}
)
assert "error" in bad, bad
def test_update_surfaces_added_grants(self, kernel, plugin_layout):
"""Re-registration that EXPANDS grants is surfaced (declared-
need transparency), mirroring removed_capabilities."""
host_only = plugin_layout["host_only"]
plugin = plugin_layout["plugin_dir"] / "plugin.py"
plugin.write_text(ENV_LEAK_PLUGIN, encoding="utf-8")
base = {
"name": "probe",
"version": "1",
"capabilities": ["cap.d"],
"hooks": {"probe": True},
"_file": str(plugin),
}
kernel.register(base)
updated = kernel.register(
{**base, "version": "2", "grants": {"paths": [str(host_only)]}}
)
assert updated.get("status") == "updated", updated
assert updated.get("added_grants") == {"paths": [str(host_only)]}, updated