20_multi_tenant.py

"""20: Multi-tenant isolation. Two tenants in one process: separate cwd, separate config dir, no shared settings or memory."""

import asyncio
import tempfile
from pathlib import Path

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher, ResultMessage, query


def confine_to(root: Path):
    """PreToolUse hook: deny any file tool whose path resolves outside the tenant's working directory."""
    async def hook(input_data, tool_use_id, context):
        target = input_data["tool_input"].get("file_path") or input_data["tool_input"].get("path")
        if target and not Path(target).resolve().is_relative_to(root.resolve()):
            print(f"  [hook] DENY {input_data['tool_name']}({target})")
            return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny",
                                           "permissionDecisionReason": f"{target} is outside the tenant workspace"}}
        return {}
    return hook


def make_tenant(name: str, secret: str):
    root = Path(tempfile.mkdtemp(prefix=f"ws-{name}-"))
    work, config = root / "work", root / "config"
    work.mkdir(); config.mkdir()
    (work / "CLAUDE.md").write_text(f"# {name} conventions\nThe internal project codename is {secret}.\n")
    (work / "notes.txt").write_text(f"{name} private notes\n")
    return work, config


async def ask(work, config, prompt, sources, **extra):
    options = ClaudeAgentOptions(
        cwd=str(work),
        setting_sources=sources,                                      # [] = no host/project settings, no CLAUDE.md injection
        env={"CLAUDE_CONFIG_DIR": str(config),                        # per-tenant config dir: transcripts, state
             "CLAUDE_CODE_DISABLE_AUTO_MEMORY": "1"},                 # auto memory would load regardless of setting_sources
        max_turns=3, **extra,
    )
    async for m in query(prompt=prompt, options=options):
        if isinstance(m, ResultMessage):
            return (m.result or "").strip(), m.session_id, len(m.permission_denials)


async def main():
    a_work, a_config = make_tenant("acme", "BLUEBIRD")
    b_work, b_config = make_tenant("globex", "REDWOOD")
    q = "Without using any tools: what is the internal project codename? Reply with the codename only, or 'unknown'."

    print("=== 1. settings isolation: setting_sources=[] means CLAUDE.md is NOT injected ===")
    a, a_sid, _ = await ask(a_work, a_config, q, sources=[], tools=[])
    b, b_sid, _ = await ask(b_work, b_config, q, sources=[], tools=[])
    print(f"acme   -> {a!r}\nglobex -> {b!r}")

    print("\n=== 2. the same tenants with project settings loaded on purpose (setting_sources=['project']) ===")
    a2, _, _ = await ask(a_work, a_config, q, sources=["project"], tools=[])
    b2, _, _ = await ask(b_work, b_config, q, sources=["project"], tools=[])
    print(f"acme   -> {a2!r}\nglobex -> {b2!r}")

    print("\n=== 3. file isolation: does cwd alone bound what tools can reach? ===")
    probe = f"Read the file {a_work}/notes.txt and quote its contents in one line. If the read fails, say exactly why."
    c, _, denials = await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"])
    print(f"globex, cwd only          -> {c}   (denials={denials})")

    print("\n=== 3b. file isolation done right: a PreToolUse hook (or the OS sandbox) confines paths to the tenant root ===")
    d, _, denials = await ask(b_work, b_config, probe, sources=[], allowed_tools=["Read"],
                              hooks={"PreToolUse": [HookMatcher(matcher="Read|Edit|Write|Glob|Grep", hooks=[confine_to(b_work)])]})
    print(f"globex, with confine hook -> {d}   (denials={denials})")

    print("\n=== 4. transcript isolation: each tenant's sessions live under its own CLAUDE_CONFIG_DIR ===")
    for name, cfg in (("acme", a_config), ("globex", b_config)):
        print(f"{name}: {len(list(cfg.glob('projects/*/*.jsonl')))} transcript(s) under {cfg}")
    print(f"tenant B's config dir contains tenant A's session: {any(a_sid in str(p) for p in b_config.rglob('*.jsonl'))}")


asyncio.run(main())