Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion codec_consent.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@
"""
import os

__all__ = ["gate_enabled", "is_destructive_skill", "chat_consent_ok", "mcp_refuse_message"]
__all__ = ["gate_enabled", "is_destructive_skill", "chat_consent_ok",
"mcp_refuse_message", "mcp_allowed"]

# Known high-power built-ins that are destructive but NOT in _HTTP_BLOCKED.
# (terminal / python_exec / process_manager / pm2_control / ax_control are
Expand All @@ -35,6 +36,17 @@
"skill_forge", # writes a skill to disk (no review gate)
})

# Destructive skills that are nonetheless SAFE to expose over MCP because they
# self-guard. `file_write` is write-only (no read/delete/list), blocks the
# system tree + credential files + all of ~/.codec, restricts writes to $HOME
# or /tmp, caps size at 500 KB, and audits every write to ~/.codec/file_write.log.
# It was purpose-built for remote (claude.ai) callers — the blanket MCP refusal
# contradicted its own design and broke the Claude→CODEC→save-a-file flow.
# Allowing it here is the "scoped-safe write" the operator wants, WITHOUT opening
# delete/move (file_ops), shells (terminal/python_exec), messaging
# (imessage_send) or browser control (pilot), which all remain refused over MCP.
_MCP_SELF_GUARDED = frozenset({"file_write"})


def gate_enabled() -> bool:
"""Consent gate on by default; CONSENT_GATE_ENABLED=false disables it."""
Expand Down Expand Up @@ -94,6 +106,21 @@ def chat_consent_ok(tool_name, query, *, registry=None) -> bool:
return False


def mcp_allowed(tool_name, registry=None) -> bool:
"""May this skill run over MCP (remote claude.ai caller)?

Non-destructive skills always may. A destructive skill may ONLY if it
self-guards (see _MCP_SELF_GUARDED) — currently just file_write, which
enforces its own path safety + audit. Everything else destructive is refused.
Never raises; fail-closed (any error → not allowed)."""
try:
if not is_destructive_skill(tool_name, registry=registry):
return True
return tool_name in _MCP_SELF_GUARDED
except Exception:
return False


def mcp_refuse_message(tool_name) -> str:
return (
f"Skill '{tool_name}' is a destructive/high-power operation and is not "
Expand Down
2 changes: 1 addition & 1 deletion codec_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ def tool_fn(task: str, context: str = "") -> str:
# imessage_send, pilot, …) + any skill declaring SKILL_DESTRUCTIVE.
try:
import codec_consent
if codec_consent.gate_enabled() and codec_consent.is_destructive_skill(rkey, registry=registry):
if codec_consent.gate_enabled() and not codec_consent.mcp_allowed(rkey, registry=registry):
_audit(sname, event="denied",
task_len=tlen, context_len=clen,
duration_ms=(time.time()-t0)*1000,
Expand Down
2 changes: 1 addition & 1 deletion skills/.manifest.json
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@
"fact_extract.py": "a43ed03b8c51704415f4135de46a44e097f757305af3921dd389b8dfd0540906",
"file_ops.py": "890f581c891a2c89dbd51177cf7b8152dba224a590a7f526737e9003f9046dce",
"file_search.py": "422274fb2386a1e3606a9f5539b3be6a381ebb4c0d968957b14150299a25eb34",
"file_write.py": "0c7a91354464c7aceb3af2b6ee8ba903093ba203b9f810f1de86edb5f1eaf7a6",
"file_write.py": "7bfffb3b55580e8b5abc1fbdef43402233068267ad1d28fdc9bafd53d84430d2",
"google_calendar.py": "7ef8c9a7fd02a5c2b52c7ad09094bda0c15851360345c5cb110cfd032ef9a562",
"google_docs.py": "75980457cb9304e970e9dcaaa12e2acca51559344d2f022896260a6a884f8318",
"google_drive.py": "af70de60bb98e2ce2aa6ac4522349b56b2932ed831b182453a22e5f96edd246c",
Expand Down
23 changes: 22 additions & 1 deletion skills/file_write.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,17 @@
"/var", "/dev", "/Volumes",
]

# Sensitive directories UNDER $HOME. The filename-pattern check below only sees
# the basename, so it misses a write INTO these dirs (e.g. ~/.ssh/authorized_keys
# — an SSH-key injection — has basename "authorized_keys", which matches nothing).
# Block the whole subtree of each. Especially important now file_write is exposed
# to remote (claude.ai) callers over MCP. (~/.config is intentionally NOT here —
# too broad; legit apps write there.)
_BLOCKED_HOME_SUBDIRS = [
".ssh", ".aws", ".gnupg", ".kube", ".gcloud", ".docker", ".azure",
".config/gcloud", ".config/gh", ".password-store",
]

# Security-sensitive CODEC directories. Per audit finding D-4, the
# file_write skill must NEVER write into these — they govern skill loading,
# plugin lifecycle hooks, agent permission grants, audit-log integrity,
Expand Down Expand Up @@ -94,6 +105,13 @@ def _build_blocked_roots() -> list[str]:
except Exception:
roots.append(p)
roots.extend(_codec_blocked_roots())
# Sensitive dirs under $HOME (block the whole subtree).
home = os.path.expanduser("~")
for sub in _BLOCKED_HOME_SUBDIRS:
try:
roots.append(os.path.realpath(os.path.join(home, sub)))
except Exception:
roots.append(os.path.join(home, sub))
return roots


Expand All @@ -108,7 +126,10 @@ def _build_blocked_roots() -> list[str]:
".ssh", ".gnupg", ".env", "credentials", "secrets", "secret",
".aws", ".gcloud", ".kube", "id_rsa", "id_ed25519", "id_dsa",
".netrc", ".npmrc", ".pypirc", "keychain", "password", "token",
"api_key", "apikey", "private_key",
"api_key", "apikey", "private_key", "authorized_keys",
# Shell-init files: writing one = arbitrary code execution on next shell.
".zshrc", ".zshenv", ".zprofile", ".zlogin", ".bashrc", ".bash_profile",
".bash_login", ".profile", "crontab",
]
# Block extensions that could be executable shells / trust-sensitive.
_BLOCKED_EXTS = [".pem", ".key", ".p12", ".pfx", ".keystore"]
Expand Down
22 changes: 22 additions & 0 deletions tests/test_consent.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,3 +89,25 @@ def _boom(*a, **k):

monkeypatch.setattr(codec_ask_user, "ask", _boom)
assert codec_consent.chat_consent_ok("file_ops", "x", registry=_FakeReg(set())) is False


# ── mcp_allowed: file_write self-guards, so it's allowed over MCP (2026-07) ────

def test_mcp_allows_file_write_but_refuses_other_destructive():
"""Claude→CODEC→save-a-file (beat #18) was blanket-refused. file_write
self-guards (path safety + audit), so it's now allowed over MCP; delete/
shell/messaging/browser stay refused."""
reg = _FakeReg(set())
assert codec_consent.mcp_allowed("file_write", registry=reg) is True
for s in ("file_ops", "imessage_send", "pilot", "skill_forge", "terminal", "python_exec"):
assert codec_consent.mcp_allowed(s, registry=reg) is False, s


def test_mcp_allows_nondestructive_skills():
reg = _FakeReg(set())
for s in ("weather", "calculator", "web_search"):
assert codec_consent.mcp_allowed(s, registry=reg) is True, s


def test_mcp_allowed_fails_closed_on_bad_input():
assert codec_consent.mcp_allowed(None) in (False, True) # never raises
45 changes: 45 additions & 0 deletions tests/test_file_write.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
import sys
from pathlib import Path

import pytest

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO))
sys.path.insert(0, str(REPO / "skills"))
Expand Down Expand Up @@ -257,3 +259,46 @@ def fake_log_event(event_type, *args, **kwargs):
assert "skills" in extra["target_path"], (
f"target_path should reference the attempted sensitive path: {extra}"
)


# ── sensitive-dir hardening (2026-07): file_write is now exposed to remote MCP
# callers, so a write INTO ~/.ssh (basename doesn't match the .ssh pattern) must
# be blocked at the directory level, not just by filename. ──
import importlib as _il
import file_write as _fw
_il.reload(_fw)


@pytest.mark.parametrize("path", [
"~/.ssh/authorized_keys", # SSH key injection
"~/.ssh/config",
"~/.ssh/random",
"~/.aws/credentials",
"~/.gnupg/anything",
"~/.kube/config",
"~/.gcloud/creds",
"~/.config/gcloud/token",
"~/.zshrc", # code exec on next shell
"~/.bashrc",
"~/.bash_profile",
"~/.zshenv",
])
def test_sensitive_targets_refused(path):
out = _fw.run(f"save to {path} content: pwned")
assert "refused" in out.lower(), f"{path} was NOT blocked: {out}"


@pytest.mark.parametrize("path", [
"~/Downloads/note.txt",
"~/Desktop/scratch.md",
"~/Documents/plan.md",
])
def test_safe_targets_still_allowed(tmp_path, path):
import os
out = _fw.run(f"save to {path} content: ok-{os.getpid()}")
p = os.path.expanduser(path)
try:
assert "saved" in out.lower() and os.path.exists(p), out
finally:
if os.path.exists(p):
os.remove(p)