# analyze.py: grade and measure each fetched config. Writes results.jsonl with derived fields only: # never a value from an env block, an argument, a URL or a header, never a repo or a path. import json, os, re, subprocess, sys, tomllib, math, shutil from concurrent.futures import ThreadPoolExecutor S = sys.argv[1] BIN = f"{S}/bin/deedglass" EMPTY_HOME = f"{S}/emptyhome" os.makedirs(EMPTY_HOME, exist_ok=True) def strip_jsonc(t): out, i, n, ins = [], 0, len(t), False while i < n: c = t[i] if ins: out.append(c) if c == "\\" and i + 1 < n: out.append(t[i+1]); i += 2; continue if c == '"': ins = False i += 1; continue if c == '"': ins = True; out.append(c); i += 1; continue if t.startswith("//", i): j = t.find("\n", i); i = n if j < 0 else j; continue if t.startswith("/*", i): j = t.find("*/", i + 2); i = n if j < 0 else j + 2; continue out.append(c); i += 1 s = "".join(out) return re.sub(r",(\s*[}\]])", r"\1", s) def parse(kind, b): t = b.decode("utf-8", "replace").lstrip("") if kind == "codex": return tomllib.loads(t) try: return json.loads(t) except Exception: return json.loads(strip_jsonc(t)) def servers_of(kind, d): if not isinstance(d, dict): return {} if kind == "codex": s = d.get("mcp_servers") or {} elif kind == "plain-mcp": s = d.get("mcpServers") or d.get("servers") if not s and isinstance(d.get("mcp"), dict): s = d["mcp"].get("servers") else: s = d.get("mcpServers") or {} return {k: v for k, v in (s or {}).items() if isinstance(v, dict)} if isinstance(s, dict) else {} # ---- server identity and category ------------------------------------------------------------- RUNNERS = {"bash", "sh", "zsh", "wsl", "wsl.exe", "powershell", "powershell.exe", "pwsh", "cmd.exe", "-c", "mise", "env", "sudo", "exec", "npx", "npx.cmd", "bunx", "pnpx", "uvx", "pipx", "dlx", "yarn", "pnpm", "bun", "deno", "node", "python", "python3", "uv", "docker", "podman", "cmd", "/c", "run", "exec", "-m", "x", "tool"} def canon(name, spec): cmd = str(spec.get("command") or "") args = [str(a) for a in (spec.get("args") or []) if isinstance(a, (str, int))] url = spec.get("url") or spec.get("serverUrl") or spec.get("httpUrl") if url: m = re.match(r"https?://([^/:]+)", str(url)) return ("remote:" + m.group(1).lower()) if m else "remote" toks = [os.path.basename(cmd)] + [t for a in args for t in a.split()] args = toks[1:] if toks and toks[0] in ("docker", "podman"): # the image is the first token after run that isn't a flag or a flag's value skip = False for a in args[1:] if args and args[0] == "run" else args: if skip: skip = False; continue if a.startswith("-"): if a in ("-e", "--env", "-v", "--volume", "--name", "--network", "-p", "--env-file", "--mount", "-w", "--user", "-u"): skip = True continue img = a.split("@")[0] img = re.sub(r":[\w.\-]+$", "", img) return "docker:" + img.lower() return "docker" for a in toks: if not a or a.startswith("-") or a.lower() in RUNNERS or a.startswith("/") and a.endswith(("node", "python", "python3")): continue a = a.replace("\\", "/") if "/" in a and not a.startswith("@"): base = a.rstrip("/").split("/")[-1] return "local:" + re.sub(r"\.(js|py|mjs|ts|exe)$", "", base).lower() m = re.match(r"^(@[^/@]+/[^@]+|[^@]+)", a) return (m.group(1) if m else a).lower() return "name:" + str(name).lower() CAT = [ ("database", r"postgres|\bpg\b|pgsql|mysql|mariadb|sqlite|mongo|supabase|redis|neon|prisma|\bsql\b|sql-?server|mssql|clickhouse|snowflake|bigquery|dynamo|firestore|firebase|planetscale|duckdb|oracle|cockroach|cassandra|elasticsearch|opensearch|timescale|turso|libsql|surreal|airtable|database|\bdb\b|-db\b|db-|dbhub|qdrant|pinecone|chroma|weaviate|milvus|couchbase|neo4j|bytebase|toolbox"), ("shell", r"shell|terminal|desktop-commander|commander|\bexec\b|command|bash|\bcli-mcp|iterm|\bssh\b|ssh-|powershell|cmd-mcp|run-?command|code-?runner|interpreter|sandbox-?exec"), ("filesystem", r"filesystem|file-?system|\bfs\b|files\b|server-files"), ("browser", r"playwright|puppeteer|browser|chrome|selenium"), ("cloud", r"aws|azure|gcp|google-cloud|cloudflare|kubernetes|k8s|kubectl|terraform|vercel|netlify|heroku|railway|fly\.io|docker-mcp|pulumi"), ("payments", r"stripe|paypal|razorpay|square|braintree|adyen|coinbase|wallet"), ] def cats(name, cid, spec): hay = (str(name) + " " + cid).lower() out = [c for c, rx in CAT if re.search(rx, hay)] if "shell" in out and re.search(r"filesystem|shadcn|spectrum|commandbox|smithery", hay): out.remove("shell") if "database" in out and re.search(r"mongo-?express|sqlfluff|sql-?formatter", hay): out.remove("database") return out READONLY = re.compile(r"read[-_ ]?only|readonly|--ro\b|access[-_]mode[=:\s]*restricted|read_only|allow[-_]?writes?[=:\s]*false|--no-?write|safe[-_]?mode", re.I) def readonly(spec): blob = json.dumps({k: spec.get(k) for k in ("args", "env", "url", "headers")}) return bool(READONLY.search(blob)) # ---- credentials (counted, never kept) -------------------------------------------------------- TOKEN = [ ("github", r"\b(ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{30,}|\bgithub_pat_[A-Za-z0-9_]{40,}"), ("openai_or_anthropic", r"\bsk-(ant-|proj-)?[A-Za-z0-9_\-]{24,}"), ("aws", r"\b(AKIA|ASIA)[A-Z0-9]{16}\b"), ("google", r"\bAIza[0-9A-Za-z_\-]{35}\b"), ("slack", r"\bxox[abprs]-[A-Za-z0-9\-]{10,}"), ("gitlab", r"\bglpat-[A-Za-z0-9_\-]{20}"), ("stripe", r"\b(sk|rk)_(live|test)_[A-Za-z0-9]{16,}"), ("supabase_or_jwt", r"\beyJ[A-Za-z0-9_\-]{20,}\.eyJ[A-Za-z0-9_\-]{20,}\.[A-Za-z0-9_\-]{10,}"), ("other_vendor", r"\b(xai|gsk|pplx|tvly|fc|lin_api|ntn|secret|pat|key)[-_][A-Za-z0-9]{24,}"), ] PLACE = re.compile(r"your|xxx|\.\.\.|<|>|\$\{|\{\{|changeme|example|placeholder|replace|insert|dummy|sample|todo|redacted|\*\*\*|env:|process\.env|my[-_]|test[-_]?key|here|abc123|123456|password\b|^pass(word)?$|secret$|^token$", re.I) SECRET_NAME = re.compile(r"(KEY|TOKEN|SECRET|PASSWORD|PASSWD|PWD|CREDENTIAL|AUTH|PAT)$|API_?KEY|ACCESS_?KEY|PRIVATE", re.I) DBURL = re.compile(r"\b(postgres(ql)?|mysql|mongodb(\+srv)?|redis|rediss|mssql|sqlserver)://([^:/\s@]+):([^@\s]+)@([^/\s:]+)", re.I) def entropy(s): if not s: return 0 from collections import Counter n = len(s); return -sum(c/n * math.log2(c/n) for c in Counter(s).values()) def cred_types(strings_with_names): found = set() for name, v in strings_with_names: if not isinstance(v, str) or len(v) < 8: continue for t, rx in TOKEN: m = re.search(rx, v) if m and not PLACE.search(m.group(0)) and entropy(m.group(0)) > 3.3: found.add(t) m = DBURL.search(v) if m: pw, host = m.group(5), m.group(6).lower() if not PLACE.search(pw) and pw not in ("postgres", "root", "admin", "user", "pass", "secret") and len(pw) >= 6: found.add("db_url_password_local" if host in ("localhost", "127.0.0.1", "0.0.0.0", "db", "postgres", "mysql", "host.docker.internal") else "db_url_password_remote") if name and SECRET_NAME.search(str(name)) and len(v) >= 20 and " " not in v and not PLACE.search(v) and not v.startswith(("$", "%")) and re.search(r"[0-9]", v) and entropy(v) > 3.6 and not v.startswith(("http", "/", "~", ".")): found.add("other_secret_named_var") return found def strings_of(spec): out = [] for k, v in (spec.get("env") or {}).items() if isinstance(spec.get("env"), dict) else []: out.append((k, v)) for k, v in (spec.get("headers") or {}).items() if isinstance(spec.get("headers"), dict) else []: out.append((k, v if not str(v).lower().startswith("bearer ") else str(v)[7:])) if str(v).lower().startswith("bearer "): out.append(("AUTH_TOKEN", str(v)[7:])) args = [a for a in (spec.get("args") or []) if isinstance(a, str)] for i, a in enumerate(args): if "=" in a and a.startswith("--"): k, _, v = a.partition("="); out.append((k.lstrip("-").upper().replace("-", "_"), v)) elif i > 0 and args[i-1].startswith("--"): out.append((args[i-1].lstrip("-").upper().replace("-", "_"), a)) else: out.append(("", a)) for k in ("url", "serverUrl", "httpUrl"): if spec.get(k): out.append(("", str(spec[k]))) m = re.search(r"[?&](api[_-]?key|token|key|access_token)=([^&]+)", str(spec.get(k) or ""), re.I) if m: out.append(("API_KEY", m.group(2))) return out # ---- permissions ------------------------------------------------------------------------------ BROAD = {"", "npm run", "npm exec", "npx", "pnpm", "pnpm run", "pnpm exec", "pnpm dlx", "yarn", "yarn run", "bun run", "bunx", "uv run", "poetry run", "python", "python3", "node", "bundle exec", "sh", "bash"} def broad_bash(rule): r = str(rule).strip() if r == "Bash": return True m = re.match(r"^Bash\((.*)\)$", r) if not m: return False spec = m.group(1) if spec.endswith(":*"): spec = spec[:-2] before = spec.split("*", 1)[0].strip() return before in BROAD def claude_settings(d): p = d.get("permissions") if isinstance(d.get("permissions"), dict) else {} allow = [str(x) for x in p.get("allow") or [] if isinstance(x, str)] ask = [str(x) for x in p.get("ask") or [] if isinstance(x, str)] deny = [str(x) for x in p.get("deny") or [] if isinstance(x, str)] mode = str(p.get("defaultMode") or "") hooks = d.get("hooks") if isinstance(d.get("hooks"), dict) else {} pre = hooks.get("PreToolUse") or [] pre_any = bool(pre) pre_bash = any((not h.get("matcher")) or re.search(r"Bash|\*|\.\*", str(h.get("matcher"))) for h in pre if isinstance(h, dict)) bypass = mode == "bypassPermissions" or bool(d.get("dangerouslySkipPermissions")) or bool(p.get("dangerouslySkipPermissions")) broad = any(broad_bash(r) for r in allow) bash_deny = any(r.startswith("Bash") for r in deny) return { "has_permissions": bool(allow or ask or deny or mode), "allow_n": len(allow), "ask_n": len(ask), "deny_n": len(deny), "broad_bash": broad, "bypass_mode": bypass, "accept_edits": mode == "acceptEdits", "hooks_any": bool(hooks), "pretool_hook": pre_any, "pretool_bash_hook": pre_bash, "mcp_all_enabled": bool(d.get("enableAllProjectMcpServers")), "shell_unguarded": (broad or bypass) and not pre_bash and not bash_deny, "shell_open": broad or bypass, } def codex_shell(d): blocks = [d] + [v for v in (d.get("profiles") or {}).values() if isinstance(v, dict)] danger = any(b.get("sandbox_mode") == "danger-full-access" for b in blocks) never = any(b.get("approval_policy") == "never" for b in blocks) ro = all(b.get("sandbox_mode") in (None, "read-only") for b in blocks) and any(b.get("sandbox_mode") == "read-only" for b in blocks) return {"codex_danger_full_access": danger, "codex_approval_never": never, "shell_unguarded": danger or (never and not ro), "shell_open": danger or never} def gemini_shell(d): s = json.dumps(d).lower() yolo = '"yolo"' in s or re.search(r'"yolo"\s*:\s*true', s) is not None gen = d.get("general") if isinstance(d.get("general"), dict) else {} tools = d.get("tools") if isinstance(d.get("tools"), dict) else {} auto = bool(d.get("autoAccept") or tools.get("autoAccept")) allowed = [str(x) for x in (d.get("allowedTools") or tools.get("allowed") or []) if isinstance(x, str)] shell_allowed = any(re.match(r"^(run_shell_command|ShellTool)\s*$", a) or a == "run_shell_command()" for a in allowed) sandbox = d.get("sandbox") or tools.get("sandbox") open_ = yolo or shell_allowed return {"gemini_yolo": yolo, "gemini_auto_accept": auto, "shell_unguarded": open_ and not sandbox, "shell_open": open_} # ---- scanner ---------------------------------------------------------------------------------- def place(kind, d, raw, scan): if kind == "root-mcp": rel, body = ".mcp.json", raw elif kind == "cursor": rel, body = ".cursor/mcp.json", raw elif kind == "claude-desktop": rel, body = "claude_desktop_config.json", raw elif kind == "claude-settings": rel, body = ".claude/settings.json", raw elif kind == "plain-mcp": if isinstance(d, dict) and "servers" in d and "mcpServers" not in d: rel, body = ".vscode/mcp.json", raw else: rel, body = ".mcp.json", raw else: srv = servers_of(kind, d) rel, body = ".mcp.json", json.dumps({"mcpServers": srv}, default=str).encode() p = os.path.join(scan, rel); os.makedirs(os.path.dirname(p), exist_ok=True) open(p, "wb").write(body if isinstance(body, bytes) else body.encode()) def run_scan(scan): env = {"HOME": EMPTY_HOME, "PATH": "/usr/bin:/bin", "NO_COLOR": "1", "XDG_CONFIG_HOME": EMPTY_HOME} r = subprocess.run([BIN, "scan", "--dir", scan, "--json", "--no-write"], capture_output=True, text=True, env=env, timeout=60) try: j = json.loads(r.stdout) except Exception: return None routed = {} for c in j.get("configs") or []: if c.get("user"): continue for s in c.get("servers") or []: routed[s.get("name")] = bool(s.get("routed")) return {"grade": j["score"]["grade"], "points": j["score"]["points"], "unchecked": j["score"]["unchecked_write_paths"], "findings": sorted({f["id"] for f in j.get("findings") or []}), "write_tools": j["summary"]["write_capable_tools"], "deductions": sorted({x["id"] for x in j["score"]["deductions"]}), "routed": routed} def one(m): d0 = f"{S}/work/{m['id']}" raw = open(f"{d0}/orig", "rb").read() out = {"id": m["id"], "kind": m["kind"]} try: d = parse(m["kind"], raw) except Exception: out["parse_error"] = True; return out if not isinstance(d, dict): out["parse_error"] = True; return out srv = servers_of(m["kind"], d) scan = f"{d0}/scan"; shutil.rmtree(scan, ignore_errors=True) place(m["kind"], d, raw, scan) sc = run_scan(scan) if sc is None: out["scan_error"] = True; return out routed = sc.pop("routed") out.update(sc) out["servers_n"] = len(srv) rows, creds, risky = [], set(), [] db_write = shell_srv = False for name, spec in srv.items(): cid = canon(name, spec); c = cats(name, cid, spec) ro = readonly(spec) if "database" in c and not ro: db_write = True if "shell" in c and not routed.get(name, False): shell_srv = True creds |= cred_types(strings_of(spec)) rows.append({"id": cid, "cats": c, "ro": ro, "remote": bool(spec.get("url") or spec.get("serverUrl") or spec.get("httpUrl"))}) if m["kind"] in ("claude-settings", "gemini", "codex"): envb = d.get("env") if isinstance(d.get("env"), dict) else {} if m["kind"] == "codex": sep = d.get("shell_environment_policy") if isinstance(d.get("shell_environment_policy"), dict) else {} envb = sep.get("set") if isinstance(sep.get("set"), dict) else {} creds |= cred_types(list(envb.items())) out["servers"] = rows out["cred_types"] = sorted(creds) out["db_write"] = db_write perm = {} if m["kind"] == "claude-settings": perm = claude_settings(d) elif m["kind"] == "codex": perm = codex_shell(d) elif m["kind"] == "gemini": perm = gemini_shell(d) out.update(perm) out["shell_server_unguarded"] = shell_srv out["shell_unguarded"] = bool(perm.get("shell_unguarded")) or shell_srv shutil.rmtree(scan, ignore_errors=True) return out man = json.load(open(f"{S}/work/manifest.json")) with ThreadPoolExecutor(6) as ex: res = list(ex.map(one, man)) with open(f"{S}/results.jsonl", "w") as f: for r in res: f.write(json.dumps(r) + "\n") print("analyzed", len(res), "parse errors", sum(1 for r in res if r.get("parse_error")), "scan errors", sum(1 for r in res if r.get("scan_error")), file=sys.stderr)