From 4bbaf690ef6c0f07331ee0d8dd223fa17c118726 Mon Sep 17 00:00:00 2001 From: Michael Heller Date: Sun, 2 Aug 2026 22:37:00 -0400 Subject: [PATCH] feat(agentd): enforce the consent-plane terminal-surface envelope (enforce-sweep) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the enforce-sweep (after goose voice + netwatch) to turtle-agentd, the terminal-surface tool-execution host. An AUTONOMOUS AGENT acting on the terminal surface may read/edit/test (discover/implement/verify) but may NOT egress (push/publish/network) or operate (deploy/infra) — the consent-plane surface envelope (deny_purposes=[egress,operate]) now holds at runtime. - consent_plane_check(command, actor_id): denies egress/operate commands for agent actors; humans keep full authority (envelope governs agents only). _is_agent_actor is fail-closed — only an explicit `human:` id gets authority, every other non-empty id is governed. - classification is local (regex over network/publish/deploy/infra commands) so containment holds even when the consent engine is unreachable. - wired into policy_evaluate: a consent-plane deny short-circuits before Policy Fabric (defence in depth). - 5 tests: agent egress denied, agent read/edit/test allowed, human unrestricted, policy_evaluate short-circuits, fail-closed actor detection. Daemon smoke green. --- assets/sourceos/bin/turtle-agentd | 74 +++++++++++++++++++++ assets/sourceos/tests/test_turtle_agentd.py | 52 +++++++++++++++ 2 files changed, 126 insertions(+) diff --git a/assets/sourceos/bin/turtle-agentd b/assets/sourceos/bin/turtle-agentd index 114429f0ea9..2b8d62af5e6 100755 --- a/assets/sourceos/bin/turtle-agentd +++ b/assets/sourceos/bin/turtle-agentd @@ -700,6 +700,70 @@ def _local_policy_check(command: str) -> tuple[str, str, str | None]: return "allow", "No local policy rules matched.", None +# --------------------------------------------------------------------------- +# Consent-plane terminal-surface envelope (enforce-sweep) +# --------------------------------------------------------------------------- +# The consent-plane surface catalog gives the `terminal` surface +# purposes=[discover, implement, verify] and deny_purposes=[egress, operate]. +# That envelope contains an AUTONOMOUS AGENT: a prompt-injected / runaway agent +# acting on the terminal surface may read, edit, and test, but may NOT egress +# (push/publish/network-send) or operate (deploy/infra). It does NOT restrict the +# human operator, who holds full authority — so this check applies only to agent +# actors. The classification is local so containment holds even when the consent +# engine is unreachable (fail-closed for the agent); a reachable engine adds the +# full role x space x consent check on top. + +_EGRESS_OPERATE_PATTERNS: list[re.Pattern[str]] = [ + re.compile(p, re.IGNORECASE) for p in ( + r"\bgit\s+push\b", r"\bgit\s+(fetch|pull|clone)\b", + r"\bgh\s+(pr\s+merge|release\s+create|repo\s+create|api\b.*-X\s*(POST|PUT|PATCH|DELETE))", + r"\b(curl|wget)\b", r"\bscp\b", r"\brsync\b.*::|\brsync\b.*@", r"\bssh\b\s+\S+@", + r"\bkubectl\s+(apply|create|delete|edit|patch|scale|rollout|drain|cordon)", + r"\bdocker\s+(push|run|-)", r"\bhelm\s+(install|upgrade|uninstall)", + r"\bterraform\s+(apply|destroy)", r"\bsystemctl\s+(start|stop|restart|enable|disable)", + r"\b(aws|gcloud|az)\s+\S+\s+(create|delete|update|deploy|put|set)", + r"\bnpm\s+publish\b", r"\bpip\s+(upload|install)\b", r"\bcargo\s+publish\b", + r"\bnc\b\s+\S+\s+\d+", r"\b(dd|mkfs|fdisk)\b", + ) +] + + +def _is_agent_actor(actor_id: str | None) -> bool: + """The consent-plane envelope governs autonomous agents, not the human. + Only an explicit 'human:' actor keeps full authority; every other non-empty + id (agent:*, urn:srcos:agent:*, bot:*, svc:*, or an unrecognized id) is + governed — fail-closed, so an unknown actor cannot slip past as human. + (turtle-agentd defaults actor_id to 'human:local-user' at its call sites, so + an empty id here means the daemon's own default human path.)""" + a = (actor_id or "").lower() + if not a or a.startswith("human:"): + return False + return True + + +def _terminal_purpose(command: str) -> str | None: + """Return 'egress-or-operate' when the command leaves the surface (network / + publish / deploy / infra mutation); None for read/edit/test (discover/ + implement/verify), which the terminal surface allows.""" + for pat in _EGRESS_OPERATE_PATTERNS: + if pat.search(command): + return "egress-or-operate" + return None + + +def consent_plane_check(command: str, actor_id: str | None) -> tuple[str, str]: + """Apply the terminal-surface envelope to an agent actor's command. + Returns (decision, reason). Deny wins over Policy Fabric (defence in depth).""" + if not _is_agent_actor(actor_id): + return "allow", "human actor — consent-plane envelope governs agents only" + if _terminal_purpose(command): + return ("deny", + "[consent-plane] terminal surface denies egress/operate for an agent actor " + "(role x surface x space containment); a network/publish/deploy action is not " + "admissible on the terminal surface. Escalate to a human operator.") + return "allow", "[consent-plane] discover/implement/verify admissible on terminal surface" + + # --------------------------------------------------------------------------- # Policy Fabric wire (Track D) # --------------------------------------------------------------------------- @@ -730,6 +794,16 @@ def policy_evaluate( "source": "local-policy", } + # Consent-plane terminal-surface envelope: an agent actor may not egress/operate + # on the terminal surface. Deny short-circuits (defence in depth, holds offline). + cp_decision, cp_reason = consent_plane_check(command, actor_id) + if cp_decision == "deny": + return { + "decision": {"outcome": "deny", "reason": cp_reason, "matched_rule": "consent-plane:terminal"}, + "decision_id": None, + "source": "consent-plane", + } + if risk_level is None: # host is high-risk, isolated domains are lower risk_level = "high" if execution_domain == "host" else "medium" diff --git a/assets/sourceos/tests/test_turtle_agentd.py b/assets/sourceos/tests/test_turtle_agentd.py index 67683deb4b0..89f2c0527ab 100644 --- a/assets/sourceos/tests/test_turtle_agentd.py +++ b/assets/sourceos/tests/test_turtle_agentd.py @@ -80,3 +80,55 @@ def main() -> int: if __name__ == "__main__": raise SystemExit(main()) + + +# --------------------------------------------------------------------- consent-plane +def _agentd_module(): + """Import turtle-agentd (extensionless) to unit-test the consent-plane envelope.""" + from importlib.machinery import SourceFileLoader + import importlib.util + loader = SourceFileLoader("turtle_agentd_mod", str(AGENTD)) + spec = importlib.util.spec_from_loader("turtle_agentd_mod", loader) + mod = importlib.util.module_from_spec(spec) + loader.exec_module(mod) + return mod + + +_ad = _agentd_module() + + +def test_agent_actor_detection(): + assert _ad._is_agent_actor("urn:srcos:agent:turtle-copilot") + assert _ad._is_agent_actor("agent:autonomous") + assert _ad._is_agent_actor("mystery:actor") # unknown -> governed (fail-closed) + assert not _ad._is_agent_actor("human:local-user") # only explicit human has authority + assert not _ad._is_agent_actor("") # empty == daemon's default human path + + +def test_agent_egress_denied_on_terminal_surface(): + for cmd in ("git push origin main", "kubectl apply -f x.yaml", "curl https://evil.example", + "gh pr merge 12", "docker push repo/img", "scp f user@host:/tmp"): + d, reason = _ad.consent_plane_check(cmd, "agent:autonomous") + assert d == "deny", cmd + assert "consent-plane" in reason + + +def test_agent_read_edit_test_allowed_on_terminal(): + for cmd in ("ls -la", "cat README.md", "grep -r foo .", "git status", "git diff", + "pytest -q", "cargo test"): + d, _ = _ad.consent_plane_check(cmd, "agent:autonomous") + assert d == "allow", cmd + + +def test_human_keeps_full_authority(): + # the envelope governs agents only — a human may push/deploy + d, _ = _ad.consent_plane_check("git push origin main", "human:local-user") + assert d == "allow" + + +def test_policy_evaluate_denies_agent_egress(): + # deny short-circuits before Policy Fabric, offline + r = _ad.policy_evaluate("terminal.execute_command", "git push origin main", + execution_domain="host", actor_id="agent:autonomous") + assert r["decision"]["outcome"] == "deny" + assert r["source"] == "consent-plane"