Least Privilege, Secrets & Securing MCP

# script_id: day_074_least_privilege_mcp_security/scope_matrix
# Least privilege as data: roles -> scopes -> the tools each role may see and call.
# Install: nothing (stdlib, Python 3.11+). Env vars: none. Services: none.
ROLE_SCOPES: dict[str, frozenset[str]] = {
    "investigator": frozenset({"atlas:read", "atlas:ask"}),  # R3 reads, never writes
    "triager": frozenset({"atlas:read", "tracker:label"}),
    "responder": frozenset({"atlas:read", "atlas:ask", "tracker:comment"}),
    "ide": frozenset({"atlas:read", "atlas:ask"}),  # a maintainer's editor over MCP
}
# Compact Day 36 registry: name -> (scope it needs, function-tool schema). Parameters elided.
REGISTRY: dict[str, tuple[str, dict]] = {
    name: (scope, {"type": "function", "name": name, "description": desc,
                   "parameters": {"type": "object", "properties": {}}})
    for name, scope, desc in [
        ("search_docs", "atlas:read", "Search the Orbit docs."),
        ("ask", "atlas:ask", "Answer from the docs, with citations."),
        ("add_label", "tracker:label", "Add a label to an issue."),
        ("post_comment", "tracker:comment", "Post a comment on an issue."),
        ("close_issue", "tracker:close", "Close an issue."),
    ]
}

def scoped_tools(role: str) -> list[dict]:
    """Gate 1: the only tool schemas this role's model ever sees."""
    return [schema for scope, schema in REGISTRY.values() if scope in ROLE_SCOPES[role]]

def require_scope(scope: str, granted: frozenset[str]) -> None:
    """Gate 2, at call time: hiding a tool is not enough, the model can still name it."""
    if scope not in granted:
        raise PermissionError(f"missing scope {scope!r}")

# script_id: day_074_least_privilege_mcp_security/mcp_anti_patterns
# Three MCP security bugs, offline: token passthrough, confused deputy, tool poisoning + rug pull.
# Install: nothing (stdlib, Python 3.11+). Env vars: none. Services: none.
import hashlib
import json

class Tracker:
    """Mock issue-tracker API. strict=True means it checks the token's audience, as it should."""
    def __init__(self, strict: bool) -> None:
        self.strict = strict

    def close(self, issue: int, token: dict) -> str:
        if self.strict and token["aud"] != "tracker":
            return f"401 (token is for aud={token['aud']}, not tracker)"
        if "tracker:close" not in token["scopes"]:
            return "403 (missing tracker:close)"
        return f"closed #{issue}; tracker accepted token for aud={token['aud']}"

# 1. Token passthrough: the MCP server forwards the caller's token downstream.
caller = {"sub": "ana", "aud": "atlas-mcp", "scopes": ["atlas:read", "tracker:close"]}
print("passthrough, lax tracker:   ", Tracker(strict=False).close(7, caller))
print("passthrough, strict tracker:", Tracker(strict=True).close(7, caller))

# script_id: day_074_least_privilege_mcp_security/mcp_anti_patterns
# 2. Confused deputy: the server's own broad credential, used for any caller.
SERVER_TRACKER_TOKEN = {"sub": "atlas-bot", "aud": "tracker", "scopes": ["tracker:close"]}

def close_tool_naive(issue: int, caller: dict, tracker: Tracker) -> str:
    return tracker.close(issue, SERVER_TRACKER_TOKEN)  # acts with ITS power, for anyone

def close_tool_safe(issue: int, caller: dict, tracker: Tracker) -> str:
    if "tracker:close" not in caller["scopes"]:  # the caller's rights come first
        return f"refused: {caller['sub']} lacks tracker:close"
    return tracker.close(issue, SERVER_TRACKER_TOKEN)

reader = {"sub": "mallory", "aud": "atlas-mcp", "scopes": ["atlas:read"]}
print("deputy, naive:", close_tool_naive(9, reader, Tracker(strict=True)))
print("deputy, safe: ", close_tool_safe(9, reader, Tracker(strict=True)))

# script_id: day_074_least_privilege_mcp_security/dev_issuer
# DEV ONLY: a stand-in authorization server that mints RS256 JWTs from local keys.
# Install: uv add pyjwt cryptography   Env: ATLAS_KEY_DIR, ATLAS_MCP_URL, ATLAS_SIGNING_KID (all optional)
import os
import time
import uuid
from pathlib import Path

import jwt  # PyJWT
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

KEY_DIR = Path(os.environ.get("ATLAS_KEY_DIR", ".atlas_dev_keys"))
ISSUER = os.environ.get("ATLAS_ISSUER", "http://127.0.0.1:9000")  # a name; nothing listens there
PORT = int(os.environ.get("ATLAS_MCP_PORT", "8074"))  # Day 74's port, so Day 40's 8000 stays free
AUDIENCE = os.environ.get("ATLAS_MCP_URL", f"http://127.0.0.1:{PORT}/mcp")  # the service tokens are for
ACTIVE_KID = os.environ.get("ATLAS_SIGNING_KID", "dev-2026-11")

def ensure_key(kid: str) -> bytes:
    """Create the RSA key pair for `kid` once; return the private key (PEM)."""
    priv = KEY_DIR / f"{kid}.pem"
    if not priv.exists():
        KEY_DIR.mkdir(mode=0o700, exist_ok=True)
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        priv.write_bytes(key.private_bytes(serialization.Encoding.PEM,
                         serialization.PrivateFormat.PKCS8, serialization.NoEncryption()))
        priv.chmod(0o600)  # owner-only, like ~/.ssh/id_rsa
        (KEY_DIR / f"{kid}.pub.pem").write_bytes(key.public_key().public_bytes(
            serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo))
    return priv.read_bytes()

# script_id: day_074_least_privilege_mcp_security/dev_issuer
def mint_dev_token(sub: str, scopes: list[str], ttl_s: int = 900, *,
                   audience: str = AUDIENCE, kid: str = ACTIVE_KID) -> str:
    """Sign a short-lived access token. In production an IdP does this, never your app."""
    now = int(time.time())
    claims = {"iss": ISSUER, "sub": sub, "aud": audience, "client_id": f"{sub}-ide",
              "scope": " ".join(scopes), "iat": now, "exp": now + ttl_s, "jti": uuid.uuid4().hex}
    return jwt.encode(claims, ensure_key(kid), algorithm="RS256", headers={"kid": kid})

if __name__ == "__main__":
    token = mint_dev_token("ana", ["atlas:read", "atlas:ask"])
    print(token[:40] + "...")
    print(jwt.get_unverified_header(token))
    claims = jwt.decode(token, options={"verify_signature": False})  # peek only; never trust this
    print({k: claims[k] for k in ("iss", "sub", "aud", "scope")}, "ttl", claims["exp"] - claims["iat"])

eyJhbGciOiJSUzI1NiIsImtpZCI6ImRldi0yMDI2...
{'alg': 'RS256', 'kid': 'dev-2026-11', 'typ': 'JWT'}
{'iss': 'http://127.0.0.1:9000', 'sub': 'ana', 'aud': 'http://127.0.0.1:8074/mcp', 'scope': 'atlas:read atlas:ask'} ttl 900

# script_id: day_074_least_privilege_mcp_security/audit
# Append-only JSON Lines audit log + an @audited decorator. Save as audit.py; the server imports it.
# Install: nothing (stdlib, Python 3.11+). Env vars: none. Services: none.
import functools
import hashlib
import json
import time
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

NEVER_LOG = {"token", "authorization", "api_key", "password", "secret"}

def args_hash(args: dict[str, Any]) -> str:
    """Stable fingerprint of the arguments: proves *what* was called without storing it."""
    return hashlib.sha256(json.dumps(args, sort_keys=True, default=str).encode()).hexdigest()[:16]

class AuditLog:
    """One JSON line per tool call or auth decision."""
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)

    def write(self, **event: Any) -> None:
        clean = {k: ("[REDACTED]" if k.lower() in NEVER_LOG else v) for k, v in event.items()}
        line = {"ts": datetime.now(UTC).isoformat(timespec="seconds"), **clean}
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(line, default=str) + "\n")

# script_id: day_074_least_privilege_mcp_security/audit
def audited(log: AuditLog, principal: Callable[[], dict[str, Any]]) -> Callable:
    """Decorator: who called which tool, an args hash, the outcome and the latency."""
    def wrap(fn: Callable) -> Callable:
        @functools.wraps(fn)  # keeps the signature, so MCP still builds the right schema
        def inner(*args: Any, **kwargs: Any) -> Any:
            start, outcome = time.perf_counter(), "ok"
            try:
                return fn(*args, **kwargs)
            except Exception as e:
                outcome = f"error:{type(e).__name__}"
                raise
            finally:
                log.write(event="tool_call", tool=fn.__name__, **principal(),
                          args_sha=args_hash(kwargs), outcome=outcome,
                          ms=round((time.perf_counter() - start) * 1000, 1))
        return inner
    return wrap

if __name__ == "__main__":
    log = AuditLog("audit_demo.jsonl")
    log.write(event="auth", decision="deny", reason="InvalidAudienceError", token="eyJ...")
    print(log.path.read_text().splitlines()[-1])

{"ts": "2026-11-07T04:11:20+00:00", "event": "auth", "decision": "deny", "reason": "InvalidAudienceError", "token": "[REDACTED]"}

# script_id: day_074_least_privilege_mcp_security/secure_mcp_server
# Atlas MCP server, hardened: OAuth 2.1 resource server + per-tool scopes + audit log.
# Install: uv add "mcp[cli]" pyjwt cryptography pydantic   (save audit.py beside this file)
# Env: ORBIT_DATA (default datasets/orbit), ATLAS_KEY_DIR, ATLAS_ISSUER, ATLAS_MCP_PORT, ATLAS_MCP_URL
# Run: python dev_issuer.py once (creates keys), then python secure_mcp_server.py
import os
import re
import sys
from pathlib import Path

import jwt  # PyJWT
from mcp.server.auth.provider import AccessToken, TokenVerifier

from audit import AuditLog, audited

KEY_DIR = Path(os.environ.get("ATLAS_KEY_DIR", ".atlas_dev_keys"))
ISSUER = os.environ.get("ATLAS_ISSUER", "http://127.0.0.1:9000")
PORT = int(os.environ.get("ATLAS_MCP_PORT", "8074"))
RESOURCE = os.environ.get("ATLAS_MCP_URL", f"http://127.0.0.1:{PORT}/mcp")  # our identity = accepted aud
AUDIT = AuditLog(os.environ.get("ATLAS_AUDIT_LOG", "audit.jsonl"))

class JWTTokenVerifier(TokenVerifier):
    """Signature, issuer, audience, expiry -> AccessToken. Anything else -> None (401)."""
    def __init__(self, key_dir: Path, issuer: str, audience: str, leeway_s: int = 30) -> None:
        self.key_dir, self.issuer, self.audience, self.leeway_s = key_dir, issuer, audience, leeway_s
        self.keys = self._load()

    def _load(self) -> dict[str, bytes]:  # kid -> public key; in production, the IdP's JWKS
        return {p.name.removesuffix(".pub.pem"): p.read_bytes() for p in self.key_dir.glob("*.pub.pem")}

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            kid = jwt.get_unverified_header(token).get("kid", "")
            if kid not in self.keys:
                self.keys = self._load()  # a new key appeared (rotation): reload once
            if kid not in self.keys:
                raise jwt.InvalidKeyError(f"unknown kid {kid!r}")  # retired or forged
            claims = jwt.decode(
                token, self.keys[kid], algorithms=["RS256"],  # pinned: never trust "alg"
                audience=self.audience, issuer=self.issuer, leeway=self.leeway_s,
                options={"require": ["exp", "iss", "aud", "sub", "jti"]})
        except jwt.PyJWTError as e:  # bad signature, wrong aud/iss, expired, malformed, unknown kid
            AUDIT.write(event="auth", decision="deny", reason=type(e).__name__)
            return None
        return AccessToken(token=token, client_id=claims.get("client_id", claims["sub"]),
                           scopes=claims.get("scope", "").split(), expires_at=claims["exp"],
                           resource=claims["aud"], subject=claims["sub"], claims=claims)

# script_id: day_074_least_privilege_mcp_security/secure_mcp_server
from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer(
    "atlas", version="1.1.0",
    instructions="Atlas knows Orbit. search_docs finds pages; ask gives a cited answer.",
    token_verifier=JWTTokenVerifier(KEY_DIR, ISSUER, RESOURCE),
    auth=AuthSettings(
        issuer_url=ISSUER,               # advertised in the PRM document: where to get tokens
        resource_server_url=RESOURCE,    # who we are
        required_scopes=["atlas:read"],  # the floor for any call (403 insufficient_scope)
        validate_token_resource=True,    # refuse tokens issued for another service
    ),
)

def caller() -> dict[str, str]:
    """Who is calling, for the audit log. Never the token itself."""
    tok = get_access_token()
    if tok is None:
        return {"sub": "anonymous"}
    return {"sub": tok.subject or "?", "client_id": tok.client_id, "jti": (tok.claims or {}).get("jti", "?")}

def require_scope(scope: str) -> None:
    """Per-tool check. ToolError (not PermissionError) so the client sees the message."""
    tok = get_access_token()
    if tok is None or scope not in tok.scopes:
        AUDIT.write(event="authz", decision="deny", need=scope, **caller())
        raise ToolError(f"This tool needs scope '{scope}'. Request a token that includes it.")

# script_id: day_074_least_privilege_mcp_security/secure_mcp_server
from typing import Annotated, Literal

from pydantic import BaseModel, Field

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
WORD = re.compile(r"[a-z0-9]+")

def parse_page(path: Path) -> dict[str, str]:
    """Front matter + body, Day 40's parser in compact form."""
    _, head, body = path.read_text(encoding="utf-8").split("---\n", 2)
    meta = {k.strip(): v.split(" #")[0].strip().strip('"')
            for k, _, v in (line.partition(":") for line in head.splitlines())}
    return meta | {"body": body}

PAGES = {p.stem: parse_page(p) for p in sorted((DATA_DIR / "docs").glob("*.md"))}

class DocHit(BaseModel):
    doc_id: str
    title: str
    section: str
    snippet: str

class Citation(BaseModel):
    doc_id: str
    quote: str

class Answer(BaseModel):
    answer: str
    citations: list[Citation]
    degraded: bool = False  # True = no model call was made

def keyword_search(query: str, k: int, applies_to: str | None = None) -> list[DocHit]:
    """Stand-in for Day 40's scorer: query-word counts, title words count 5x."""
    terms, scored = WORD.findall(query.lower()), []
    for doc_id, p in PAGES.items():
        if applies_to and p.get("applies_to") not in (applies_to, "all"):
            continue
        words = WORD.findall(p["body"].lower()) + 5 * WORD.findall(p.get("title", "").lower())
        if score := sum(words.count(t) for t in terms):
            scored.append((score, doc_id))
    return [DocHit(doc_id=d, title=PAGES[d].get("title", d), section=PAGES[d].get("section", ""),
                   snippet=" ".join(PAGES[d]["body"].split())[:200]) for _, d in sorted(scored, reverse=True)[:k]]

# script_id: day_074_least_privilege_mcp_security/secure_mcp_server
@mcp.tool()
@audited(AUDIT, principal=caller)
def search_docs(
    query: Annotated[str, Field(min_length=2, max_length=200)],
    k: Annotated[int, Field(ge=1, le=10)] = 5,
    applies_to: Literal["2.x", "3.x", "all"] | None = None,
) -> list[DocHit]:
    """Find Orbit documentation pages by keyword, best match first. Needs atlas:read."""
    return keyword_search(query, k, applies_to)  # the server-wide floor already checked atlas:read

@mcp.tool()
@audited(AUDIT, principal=caller)
def ask(question: Annotated[str, Field(min_length=3, max_length=500)]) -> Answer:
    """Answer a question about Orbit with citations. Needs atlas:ask (it spends model tokens)."""
    require_scope("atlas:ask")
    hits = keyword_search(question, 3)
    # Offline stand-in. For real answers, paste Day 40's answer_question(question, hits) here.
    return Answer(answer="Closest passages (no model call made).", degraded=True,
                  citations=[Citation(doc_id=h.doc_id, quote=h.snippet) for h in hits])

if __name__ == "__main__":
    if not list(KEY_DIR.glob("*.pub.pem")):
        sys.exit(f"No public keys in {KEY_DIR}/. Run `python dev_issuer.py` first.")
    print(f"Atlas MCP server {mcp.version} on {RESOURCE} (OAuth resource server)", file=sys.stderr)
    mcp.run(transport="streamable-http", host="127.0.0.1", port=PORT)

$ python dev_issuer.py && python secure_mcp_server.py   # your process details will differ
Atlas MCP server 1.1.0 on http://127.0.0.1:8074/mcp (OAuth resource server)
INFO:     Uvicorn running on http://127.0.0.1:8074 (Press CTRL+C to quit)

# script_id: day_074_least_privilege_mcp_security/client_check
# Starts secure_mcp_server.py, then calls it as four different callers.
# Install: uv add "mcp[cli]" pyjwt cryptography   (dev_issuer.py, audit.py, secure_mcp_server.py beside it)
# Env: ORBIT_DATA (default datasets/orbit). Services: none (it starts the server itself).
import asyncio
import subprocess
import sys
import time

import httpx2
from mcp import Client
from mcp.client.streamable_http import streamable_http_client

from dev_issuer import AUDIENCE, mint_dev_token

PRM = AUDIENCE.replace("/mcp", "/.well-known/oauth-protected-resource/mcp")  # RFC 9728 path

def wait_for_server(proc: subprocess.Popen, timeout_s: float = 20) -> None:
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline and proc.poll() is None:
        try:
            if httpx2.get(PRM).status_code == 200:
                return
        except httpx2.TransportError:
            pass
        time.sleep(0.3)
    raise RuntimeError("server didn't start (port busy?); run secure_mcp_server.py to see why")

async def call_as(token: str, tool: str, args: dict) -> str:
    http = httpx2.AsyncClient(headers={"Authorization": f"Bearer {token}"},
                              timeout=httpx2.Timeout(30, read=300))
    async with Client(streamable_http_client(AUDIENCE, http_client=http)) as client:
        result = await client.call_tool(tool, args)
    if result.is_error:
        return f"denied: {result.content[0].text}"
    data = result.structured_content or {}
    return f"ok, cites {[c['doc_id'] for c in data['citations']]}" if "citations" in data else f"ok: {data}"

# script_id: day_074_least_privilege_mcp_security/client_check
async def main() -> None:
    print("metadata:", httpx2.get(PRM).json())
    r = httpx2.post(AUDIENCE, json={})  # 1. no token
    print("no token ->", r.status_code, r.headers.get("www-authenticate"))
    other = mint_dev_token("ana", ["atlas:read"], audience="https://tracker.example/api")
    r = httpx2.post(AUDIENCE, json={}, headers={"Authorization": f"Bearer {other}"})
    print("token for another service ->", r.status_code)  # 2. the passthrough defence
    reader = mint_dev_token("ana", ["atlas:read"])
    print("read-only token, ask ->", await call_as(reader, "ask", {"question": "lease expiry"}))
    asker = mint_dev_token("ana", ["atlas:read", "atlas:ask"])
    print("read+ask token, ask ->", await call_as(asker, "ask", {"question": "lease expiry"}))

if __name__ == "__main__":
    mint_dev_token("setup", [])  # make sure a key pair exists before the server starts
    server = subprocess.Popen([sys.executable, "secure_mcp_server.py"],
                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        wait_for_server(server)
        asyncio.run(main())
    finally:
        server.terminate()

metadata: {'resource': 'http://127.0.0.1:8074/mcp', 'authorization_servers': ['http://127.0.0.1:9000'], 'scopes_supported': ['atlas:read'], 'bearer_methods_supported': ['header']}
no token -> 401 Bearer error="invalid_token", error_description="Authentication required", resource_metadata="http://127.0.0.1:8074/.well-known/oauth-protected-resource/mcp"
token for another service -> 401
read-only token, ask -> denied: Error executing tool ask: This tool needs scope 'atlas:ask'. Request a token that includes it.
read+ask token, ask -> ok, cites ['leases-and-heartbeats', 'troubleshooting-duplicates', 'environment-variables-2x']

{"ts": "2026-11-07T04:22:41+00:00", "event": "auth", "decision": "deny", "reason": "InvalidAudienceError"}
{"ts": "2026-11-07T04:22:41+00:00", "event": "authz", "decision": "deny", "need": "atlas:ask", "sub": "ana", "client_id": "ana-ide", "jti": "b1675bbb6cd74cdeaf087b43548861f6"}
{"ts": "2026-11-07T04:22:41+00:00", "event": "tool_call", "tool": "ask", "sub": "ana", "client_id": "ana-ide", "jti": "b1675bbb6cd74cdeaf087b43548861f6", "args_sha": "8f1c43c79e989cab", "outcome": "error:ToolError", "ms": 0.2}
{"ts": "2026-11-07T04:22:41+00:00", "event": "tool_call", "tool": "ask", "sub": "ana", "client_id": "ana-ide", "jti": "1ba18fb213e34ab2933b3390f6ef990f", "args_sha": "8f1c43c79e989cab", "outcome": "ok", "ms": 5.8}

# script_id: day_074_least_privilege_mcp_security/scoped_r3_tools
# One credential per role + a scope-checked, audited ToolExecutor for the tracker tools.
# Install: nothing beyond the stdlib (save audit.py beside this file).
# Env: ORBIT_DATA (default datasets/orbit); ATLAS_TRACKER_TOKEN_<ROLE> (fake defaults for the demo).
import json
import os
from dataclasses import dataclass
from pathlib import Path

from audit import AuditLog, args_hash

DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
ROLE_SCOPES = {  # compact copy of the matrix above
    "investigator": frozenset({"atlas:read", "atlas:ask"}),
    "triager": frozenset({"atlas:read", "tracker:label"}),
    "responder": frozenset({"atlas:read", "atlas:ask", "tracker:comment"}),
}
TOOL_SCOPE = {"add_label": "tracker:label", "post_comment": "tracker:comment",
              "close_issue": "tracker:close"}

class Secret:
    """Wraps a credential so print(), f-strings, repr() and logs show stars, never the value."""
    def __init__(self, value: str) -> None:
        self._value = value

    def reveal(self) -> str:  # the one greppable place the raw value leaves
        return self._value

    def __repr__(self) -> str:
        return "Secret('****')"
    __str__ = __repr__

@dataclass(frozen=True)
class Credential:
    role: str
    scopes: frozenset[str]
    token: Secret

def load_credential(role: str) -> Credential:
    """One env var per role: a leaked triager token can label, nothing more."""
    var = f"ATLAS_TRACKER_TOKEN_{role.upper()}"
    return Credential(role, ROLE_SCOPES[role], Secret(os.environ.get(var, f"dev-fake-{role}-token")))

# script_id: day_074_least_privilege_mcp_security/scoped_r3_tools
class MockTracker:
    """Compact Day 37 stand-in: records writes instead of calling a real tracker API."""
    def __init__(self) -> None:
        self.writes: list[str] = []

    def call(self, tool: str, args: dict, token: Secret) -> str:
        assert token.reveal(), "a real client sends Authorization: Bearer <token> here"
        self.writes.append(f"{tool} {args}")
        return "ok"

class ToolExecutor:
    def __init__(self, cred: Credential, tracker: MockTracker, audit: AuditLog) -> None:
        self.cred, self.tracker, self.audit = cred, tracker, audit

    def run(self, tool: str, args: dict) -> dict:
        need = TOOL_SCOPE.get(tool, "unknown-tool")
        allowed = need in self.cred.scopes
        self.audit.write(event="tool_call", role=self.cred.role, tool=tool, need=need,
                         decision="allow" if allowed else "deny", args_sha=args_hash(args))
        if not allowed:  # an error the model can read, not an exception that kills the run
            return {"ok": False, "error": f"{self.cred.role} may not call {tool} (needs {need})"}
        return {"ok": True, "result": self.tracker.call(tool, args, self.cred.token)}

def assert_no_secrets(text: str, *creds: Credential) -> str:
    """Run on every prompt, sandbox input and log line before it leaves the process."""
    for c in creds:
        if c.token.reveal() in text:
            raise ValueError(f"{c.role} credential found in outbound text")
    return text

# script_id: day_074_least_privilege_mcp_security/scoped_r3_tools
issues = [json.loads(line) for line in (DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8")]
attack = next(i for i in issues if "ignore your previous instructions" in i["body"])  # find by text
cred = load_credential("triager")
prompt = assert_no_secrets(f"Triage this issue:\n{attack['title']}\n{attack['body']}", cred)

proposed = [("add_label", {"issue": attack["id"], "label": "severity:S1"}),
            ("close_issue", {"issue": attack["id"]})]
log, tracker = AuditLog("tool_audit.jsonl"), MockTracker()
executor = ToolExecutor(cred, tracker, log)
print(f"issue #{attack['id']}, credential for {cred.role}: {cred.token}")
for tool, args in proposed:
    print(f"{tool} -> {executor.run(tool, args)}")
print("tracker writes:", tracker.writes)
print("last audit line:", log.path.read_text().splitlines()[-1])

issue #20, credential for triager: Secret('****')
add_label -> {'ok': True, 'result': 'ok'}
close_issue -> {'ok': False, 'error': 'triager may not call close_issue (needs tracker:close)'}
tracker writes: ["add_label {'issue': 20, 'label': 'severity:S1'}"]
last audit line: {"ts": "2026-11-07T04:22:41+00:00", "event": "tool_call", "role": "triager", "tool": "close_issue", "need": "tracker:close", "decision": "deny", "args_sha": "335d5145aff3b819"}