183 lines
7.3 KiB
Python
183 lines
7.3 KiB
Python
"""Launch fingerprint, env scrub, binary lookup and the `claude -p` argv (B1, B2, B7, B8)."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
|
|
MODEL_RE = re.compile(r"^claude-[a-z0-9-]+(\[1m\])?$")
|
|
EFFORTS = ("low", "medium", "high", "xhigh", "max")
|
|
PERMISSION_MODE = "dontAsk"
|
|
DEFAULT_TOOLS = ["Bash", "Read", "Edit", "Write", "Glob", "Grep", "Skill"]
|
|
DEFAULT_DISALLOWED = ["Bash(git push:*)", "Bash(git rebase:*)", "Bash(git reset --hard:*)",
|
|
"Bash(claude:*)"]
|
|
|
|
# Env pinned into every launch; part of the fingerprint.
|
|
ENV_PINNED_LAUNCH = {"CLAUDE_CODE_PROMPT_CACHE_TTL": "1h", "DISABLE_AUTOUPDATER": "1"}
|
|
|
|
# Host env that must not reach a child `claude -p`.
|
|
SCRUB_DROP = frozenset({
|
|
# Desktop host vars
|
|
"CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_CODE_HOST_SESSION_ID",
|
|
"CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_DESKTOP_APP_VERSION",
|
|
"CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_TERMINAL_MCP_TOOLS",
|
|
"CLAUDE_CODE_REPORT_FINDINGS", "CLAUDE_CODE_SESSION_ID", "CLAUDE_AGENT_SDK_VERSION",
|
|
"CLAUDE_CODE_ENABLE_ASK_USER_QUESTION_TOOL", "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING",
|
|
# fingerprint and auth path
|
|
"CLAUDE_EFFORT", "ANTHROPIC_MODEL", "ANTHROPIC_SMALL_FAST_MODEL", "CLAUDE_CODE_SUBAGENT_MODEL",
|
|
"ENABLE_PROMPT_CACHING_1H", "FORCE_PROMPT_CACHING_5M", "CLAUDE_CODE_SUBAGENT_PROMPT_CACHE_TTL",
|
|
"ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", "CLAUDE_CODE_USE_BEDROCK", "CLAUDE_CODE_USE_VERTEX",
|
|
# uv
|
|
"VIRTUAL_ENV",
|
|
})
|
|
SCRUB_DROP_PATTERNS = (re.compile(r"^ANTHROPIC_DEFAULT_.+_MODEL$"), re.compile(r"^UV_"))
|
|
SCRUB_KEEP = frozenset({
|
|
"CLAUDECODE", "CLAUDE_CODE_CHILD_SESSION", "AI_AGENT", "CLAUDE_PID", "CLAUDE_CODE_GIT_BASH_PATH",
|
|
"PATH", "HOME", "USERPROFILE",
|
|
"ANTHROPIC_BASE_URL", # kept: gateways need it
|
|
})
|
|
SCRUB_PINNED = {**ENV_PINNED_LAUNCH, "SPIRE_GREMLIN_CHILD": "1"}
|
|
SCRUB_PINNED_BINARY = "CLAUDE_CODE_EXECPATH" # set to the stored absolute binary path
|
|
|
|
|
|
def canonical_json(obj) -> str:
|
|
return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def _sha256_bytes(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
def _sha256_file(path: str) -> str:
|
|
with open(path, "rb") as f:
|
|
return _sha256_bytes(f.read())
|
|
|
|
|
|
def scrub_class(name: str) -> str | None:
|
|
"""'drop', 'pinned', 'keep' or None (on no list; kept)."""
|
|
n = name.upper()
|
|
if n in SCRUB_DROP or any(p.match(n) for p in SCRUB_DROP_PATTERNS):
|
|
return "drop"
|
|
if n in SCRUB_PINNED or n == SCRUB_PINNED_BINARY:
|
|
return "pinned"
|
|
if n in SCRUB_KEEP:
|
|
return "keep"
|
|
return None
|
|
|
|
|
|
def _venv_prefixes(environ) -> list[str]:
|
|
out = []
|
|
venv = environ.get("VIRTUAL_ENV")
|
|
if venv:
|
|
out.append(venv)
|
|
if os.path.normcase(os.path.abspath(sys.prefix)) != os.path.normcase(os.path.abspath(sys.base_prefix)):
|
|
out.append(sys.prefix)
|
|
return [os.path.normcase(os.path.normpath(os.path.abspath(x))) for x in out]
|
|
|
|
|
|
def _under(path: str, prefixes) -> bool:
|
|
try:
|
|
p = os.path.normcase(os.path.normpath(os.path.abspath(path)))
|
|
except ValueError:
|
|
return False
|
|
return any(p == px or p.startswith(px.rstrip("\\/") + os.sep) for px in prefixes)
|
|
|
|
|
|
def scrubbed_env(binary: str, environ=None) -> dict:
|
|
"""The child env: host env minus the drop list, venv PATH entries removed, pinned vars on
|
|
top, `CLAUDE_CODE_EXECPATH` = the stored binary path."""
|
|
if environ is None:
|
|
environ = os.environ
|
|
prefixes = _venv_prefixes(environ)
|
|
env = {}
|
|
for k, v in environ.items():
|
|
if scrub_class(k) in ("drop", "pinned"):
|
|
continue
|
|
if k.upper() == "PATH":
|
|
v = os.pathsep.join(e for e in v.split(os.pathsep) if e and not _under(e, prefixes))
|
|
env[k] = v
|
|
env.update(SCRUB_PINNED)
|
|
env[SCRUB_PINNED_BINARY] = binary
|
|
return env
|
|
|
|
|
|
def resolve_binary(explicit: str | None = None) -> str | None:
|
|
"""Absolute path of `claude`: `explicit`, else $CLAUDE_CODE_EXECPATH, else PATH."""
|
|
path = explicit or os.environ.get("CLAUDE_CODE_EXECPATH") or shutil.which("claude")
|
|
return os.path.abspath(path) if path else None
|
|
|
|
|
|
def binary_version(path: str) -> str | None:
|
|
"""First token of `<path> --version`, or None on any failure."""
|
|
try:
|
|
r = subprocess.run([path, "--version"], stdin=subprocess.DEVNULL, capture_output=True,
|
|
timeout=60)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
toks = r.stdout.decode("utf-8", "replace").split()
|
|
return toks[0] if r.returncode == 0 and toks else None
|
|
|
|
|
|
def _argv(fp: dict) -> list[str]:
|
|
"""The fingerprinted command line; its order must never change (cache hit)."""
|
|
a = [fp["binary"], "-p", "--output-format", "stream-json", "--verbose",
|
|
"--model", fp["model"], "--effort", fp["effort"],
|
|
"--permission-mode", fp["permission_mode"], "--tools", ",".join(fp["tools"])]
|
|
if fp.get("allowed_tools"):
|
|
a += ["--allowedTools", ",".join(fp["allowed_tools"])]
|
|
if fp.get("disallowed_tools"):
|
|
a += ["--disallowedTools", ",".join(fp["disallowed_tools"])]
|
|
a += ["--strict-mcp-config", "--mcp-config", fp["mcp_config"],
|
|
"--append-system-prompt-file", fp["append_prompt"]["path"]]
|
|
if fp.get("agent"):
|
|
a += ["--agent", fp["agent"]]
|
|
return a
|
|
|
|
|
|
def build_fingerprint(*, binary, model, effort, cwd, agent, preamble_path, mcp_config) -> dict:
|
|
"""Validate the launch fields; return {"fingerprint", "argv", "fp_hash"}. Writes nothing."""
|
|
if not MODEL_RE.match(model or ""):
|
|
raise ValueError("model must be a full id (aliases are refused): %r" % model)
|
|
if effort not in EFFORTS:
|
|
raise ValueError("unknown effort %r (one of %s)" % (effort, ", ".join(EFFORTS)))
|
|
if not os.path.isdir(cwd):
|
|
raise ValueError("cwd is not a directory: %s" % cwd)
|
|
if not os.path.isfile(preamble_path):
|
|
raise ValueError("preamble not found: %s" % preamble_path)
|
|
if not binary or not os.path.isfile(binary):
|
|
raise ValueError("claude binary not found: %s" % binary)
|
|
preamble_path = os.path.abspath(preamble_path)
|
|
fp = {
|
|
"binary": os.path.abspath(binary),
|
|
"binary_version": binary_version(binary),
|
|
"model": model,
|
|
"effort": effort,
|
|
"permission_mode": PERMISSION_MODE,
|
|
"tools": list(DEFAULT_TOOLS),
|
|
# under dontAsk every tool not allowed is denied
|
|
"allowed_tools": list(DEFAULT_TOOLS),
|
|
"disallowed_tools": list(DEFAULT_DISALLOWED),
|
|
"agent": agent or None,
|
|
"cwd": os.path.abspath(cwd),
|
|
"append_prompt": {"path": preamble_path, "sha256": _sha256_file(preamble_path)},
|
|
"mcp_config": os.path.abspath(mcp_config),
|
|
"env_pinned": dict(ENV_PINNED_LAUNCH),
|
|
}
|
|
return {"fingerprint": fp, "argv": _argv(fp),
|
|
"fp_hash": _sha256_bytes(canonical_json(fp).encode("utf-8"))}
|
|
|
|
|
|
def train_argv(run: dict, sid: str) -> list[str]:
|
|
return list(run["argv"]) + ["--session-id", sid]
|
|
|
|
|
|
def fork_argv(run: dict, base_sid: str, sid: str) -> list[str]:
|
|
return list(run["argv"]) + ["--resume", base_sid, "--fork-session", "--session-id", sid]
|
|
|
|
|
|
def resume_argv(run: dict, base_sid: str) -> list[str]:
|
|
"""No `--session-id`: claude refuses it on a resume without a fork."""
|
|
return list(run["argv"]) + ["--resume", base_sid]
|