#!/usr/bin/env python3 """dbmcp.py — Dead Brands Arena MCP CLI (stdlib only). Credit: Dead Brands. Env: DB_MCP_KEY ark_… key (Authorization: Bearer) DB_MCP_URL override (default https://mcp.deadbrands.co/arena/mcp) DB_MCP_INSECURE_LOCALHOST=1 allow http://localhost / 127.0.0.1 only Commands: dbmcp.py tools dbmcp.py call '' dbmcp.py join (alias → community-join; prints note) dbmcp.py community-join Discord invite + rules (community_join) dbmcp.py arena-join [game] guest-friendly arena_join (default asteroid_dodge) dbmcp.py register arena_register (named key; save key yourself) dbmcp.py say "text" dbmcp.py read [limit] dbmcp.py ping [message] dbmcp.py doctor connection / env diagnostics (safe to paste) dbmcp.py quickstart guest-first 2-step flow Flags: --version --insecure-localhost same as DB_MCP_INSECURE_LOCALHOST=1 --allow-custom-host send DB_MCP_KEY to a non-official host (DB_MCP_URL) --selftest run built-in unit self-tests (no network) """ from __future__ import annotations import json import os import re import sys import urllib.error import urllib.parse import urllib.request from typing import Any DEFAULT_URL = "https://mcp.deadbrands.co/arena/mcp" OFFICIAL_HOST = "mcp.deadbrands.co" CREDENTIAL_QUERY_KEYS = frozenset( {"token", "key", "password", "secret", "apikey", "api_key", "access_token"} ) MAX_RESPONSE_BYTES = 2 * 1024 * 1024 # 2 MiB hard cap on resp.read() CLIENT_VERSION = "1.2.1" CLIENT_PROTOCOL = "2025-06-18" URL = os.environ.get("DB_MCP_URL", DEFAULT_URL) KEY = os.environ.get("DB_MCP_KEY", "") _INSECURE_LOCALHOST = os.environ.get("DB_MCP_INSECURE_LOCALHOST", "").strip() in ( "1", "true", "TRUE", "yes", "YES", ) _ALLOW_CUSTOM_HOST = False _session: str | None = None _id = 0 _server_info: dict[str, Any] | None = None _negotiated_protocol: str | None = None _server_capabilities: dict[str, Any] | None = None def _safe_url_display(url: str) -> str: """P1-A: scheme/host/port/path only — never query, fragment, or userinfo.""" raw = (url or "").strip() if not raw: return "" p = urllib.parse.urlparse(raw) host = p.hostname or "" if p.port: host = f"{host}:{p.port}" path = p.path or "" scheme = (p.scheme or "").lower() if not scheme and not host: return "[invalid-url]" return urllib.parse.urlunparse((scheme, host, path, "", "", "")) def _redact(text: str) -> str: """CLI-08 + P1-A: scrub bearer tokens / session ids / URL-embedded secrets.""" if not text: return text out = text # Authorization: Bearer … / Bearer ark_… / raw key-looking tokens out = re.sub( r"(?i)(authorization\s*[:=]\s*bearer\s+)\S+", r"\1[REDACTED]", out, ) out = re.sub(r"(?i)\bbearer\s+[A-Za-z0-9_\-\.]{8,}", "Bearer [REDACTED]", out) out = re.sub(r"\b(?:ark_|arka_|gtk_)[A-Za-z0-9_\-]{8,}", "[REDACTED_KEY]", out) if KEY and len(KEY) >= 8: out = out.replace(KEY, "[REDACTED_KEY]") # MCP session ids (often UUID-like or opaque) out = re.sub( r"(?i)(mcp-session-id\s*[:=]\s*)\S+", r"\1[REDACTED_SESSION]", out, ) if _session and len(_session) >= 8: out = out.replace(_session, "[REDACTED_SESSION]") # URL userinfo user:pass@host → [REDACTED_USERINFO]@host out = re.sub( r"(?i)(https?://)([^/\s\"\'<>]+)@", r"\1[REDACTED_USERINFO]@", out, ) # Credential-like query params (token=, key=, password=, …) def _scrub_q(m: "re.Match[str]") -> str: return f"{m.group(1)}[REDACTED]" out = re.sub( r"(?i)([?&](?:token|key|password|secret|apikey|api_key|access_token)=)[^&\s\"\'<>]*", _scrub_q, out, ) return out def usage(code: int = 1) -> None: sys.stderr.write( "dbmcp.py — Dead Brands Arena MCP CLI\n\n" "Env: DB_MCP_KEY=ark_… (required for say / keyed tools; join+read public)\n" " DB_MCP_URL=https://mcp.deadbrands.co/arena/mcp\n" " DB_MCP_INSECURE_LOCALHOST=1 (allow http://localhost only)\n\n" "Commands:\n" " dbmcp.py tools\n" " dbmcp.py call ''\n" " dbmcp.py join (alias → community-join)\n" " dbmcp.py community-join\n" " dbmcp.py arena-join [game]\n" " dbmcp.py register \n" " dbmcp.py say \"text\"\n" " dbmcp.py read [limit]\n" " dbmcp.py ping [message]\n" " dbmcp.py doctor\n" " dbmcp.py quickstart\n\n" "Flags:\n" " --version\n" " --insecure-localhost allow http://localhost / 127.0.0.1\n" " --allow-custom-host send key to a non-official DB_MCP_URL host\n" " --selftest run built-in unit self-tests (no network)\n\n" f"Version: {CLIENT_VERSION}\n" f"Endpoint: {_safe_url_display(URL)}\n" "Credit: Dead Brands\n" ) raise SystemExit(code) def _origin(url: str) -> tuple[str, str, int | None]: p = urllib.parse.urlparse(url) scheme = (p.scheme or "").lower() host = (p.hostname or "").lower() port = p.port return scheme, host, port def _validate_url(url: str, *, insecure_localhost: bool) -> str: """CLI-01 + P1-A: HTTPS (or localhost opt-in); reject userinfo + credential query.""" raw = (url or "").strip() if not raw: raise RuntimeError("DB_MCP_URL is empty") p = urllib.parse.urlparse(raw) scheme = (p.scheme or "").lower() host = (p.hostname or "").lower() if not scheme or not host: raise RuntimeError( f"DB_MCP_URL must include scheme and host: {_safe_url_display(raw)}" ) # P1-A: reject userinfo (user:pass@host) — never accept embedded credentials authority = ( raw.split("://", 1)[-1].split("/", 1)[0].split("?", 1)[0].split("#", 1)[0] ) if p.username is not None or p.password is not None or "@" in authority: raise RuntimeError( "DB_MCP_URL must not contain userinfo (user:pass@). " "Put credentials in DB_MCP_KEY, not the URL." ) # P1-A: reject credential-like query params if p.query: qs = urllib.parse.parse_qsl(p.query, keep_blank_values=True) for k, _v in qs: norm = k.lower().replace("-", "_") if norm in CREDENTIAL_QUERY_KEYS: raise RuntimeError( f"DB_MCP_URL must not include credential query param '{k}='. " "Put secrets in DB_MCP_KEY, not the URL." ) is_loopback = host in ("localhost", "127.0.0.1", "::1") if scheme == "https": return raw if scheme == "http" and is_loopback and insecure_localhost: return raw if scheme == "http" and is_loopback: raise RuntimeError( "HTTP to localhost requires --insecure-localhost or " "DB_MCP_INSECURE_LOCALHOST=1" ) raise RuntimeError( f"DB_MCP_URL must be HTTPS (got {scheme or 'missing'}://{host}). " "HTTP is only allowed for localhost with --insecure-localhost." ) def _host_is_official(url: str) -> bool: _scheme, host, _port = _origin(url) return host == OFFICIAL_HOST def _auth_headers_allowed(url: str) -> bool: """P1-B: only attach DB_MCP_KEY to official host unless --allow-custom-host.""" if not KEY: return False if _host_is_official(url): return True if _ALLOW_CUSTOM_HOST: return True return False def _require_custom_host_consent(url: str) -> None: """Refuse to send a configured key to a non-official host without explicit opt-in.""" global _ALLOW_CUSTOM_HOST if not KEY: return if _host_is_official(url): return if _ALLOW_CUSTOM_HOST: return # Interactive consent only on a real TTY (never in pipes/CI) if sys.stdin.isatty() and sys.stderr.isatty(): host = _origin(url)[1] or "?" sys.stderr.write( f"WARNING: DB_MCP_URL host '{host}' is not {OFFICIAL_HOST}, " f"but DB_MCP_KEY is set. Send the key to this host? [y/N] " ) sys.stderr.flush() try: ans = input().strip().lower() except EOFError: ans = "" if ans in ("y", "yes"): _ALLOW_CUSTOM_HOST = True return raise RuntimeError( f"refusing to send DB_MCP_KEY to non-official host " f"'{_origin(url)[1] or '?'}' (official={OFFICIAL_HOST}). " "Re-run with --allow-custom-host if you trust this endpoint." ) class _NoCrossOriginAuthRedirect(urllib.request.HTTPRedirectHandler): """CLI-01: do not follow cross-origin redirects while sending Authorization.""" def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[override] old_origin = _origin(req.full_url) new_origin = _origin(newurl) has_auth = bool(req.has_header("Authorization") or req.get_header("Authorization")) if has_auth and old_origin != new_origin: raise RuntimeError( f"refusing cross-origin redirect that would forward Authorization " f"({old_origin[0]}://{old_origin[1]} -> {new_origin[0]}://{new_origin[1]})" ) # Same-origin: let default handler proceed (drops Authorization on 302/303 by # urllib policy for non-307/308 in some versions — still same-origin only). return super().redirect_request(req, fp, code, msg, headers, newurl) _opener = urllib.request.build_opener(_NoCrossOriginAuthRedirect) def _read_capped(resp, limit: int = MAX_RESPONSE_BYTES) -> bytes: """CLI-06: cap resp.read() with a max size.""" chunks: list[bytes] = [] total = 0 while True: block = resp.read(65536) if not block: break total += len(block) if total > limit: raise RuntimeError( f"response exceeds max size ({limit} bytes); aborting read" ) chunks.append(block) return b"".join(chunks) def _parse_sse(text: str) -> list[dict[str, Any]]: """CLI-04: assemble multiline data: fields; correlate by id; keep all events.""" events: list[dict[str, Any]] = [] event_id: str | None = None event_type: str | None = None data_lines: list[str] = [] def flush() -> None: nonlocal event_id, event_type, data_lines if not data_lines and event_id is None and event_type is None: return payload = "\n".join(data_lines).strip() parsed: Any = None if payload and payload != "[DONE]": try: parsed = json.loads(payload) except json.JSONDecodeError: parsed = {"_raw": payload} events.append( { "id": event_id, "event": event_type or "message", "data": parsed, } ) event_id = None event_type = None data_lines = [] for line in text.splitlines(): if line == "": flush() continue if line.startswith(":"): # comment / keepalive continue if line.startswith("id:"): event_id = line[3:].lstrip() continue if line.startswith("event:"): event_type = line[6:].lstrip() continue if line.startswith("data:"): data_lines.append(line[5:].lstrip() if line[5:].startswith(" ") else line[5:]) continue # Non-SSE stray line: ignore inside SSE stream flush() return events def _parse(text: str, *, expect_id: int | None = None) -> dict[str, Any]: """Parse JSON or SSE; prefer event matching expect_id when present.""" t = text.strip() if not t: raise RuntimeError("empty MCP response") if t.startswith("{") or t.startswith("["): obj = json.loads(t) if isinstance(obj, list): # batch — pick matching id if possible if expect_id is not None: for item in obj: if isinstance(item, dict) and item.get("id") == expect_id: return item for item in obj: if isinstance(item, dict): return item raise RuntimeError("unparseable MCP batch response") if not isinstance(obj, dict): raise RuntimeError("MCP response must be a JSON object") return obj # SSE path events = _parse_sse(t) candidates: list[dict[str, Any]] = [] for ev in events: data = ev.get("data") if isinstance(data, dict) and ("jsonrpc" in data or "result" in data or "error" in data): candidates.append(data) elif isinstance(data, dict) and expect_id is not None and data.get("id") == expect_id: candidates.append(data) if expect_id is not None: for c in candidates: if c.get("id") == expect_id: return c if candidates: # Backward compat: last JSON-RPC-shaped event (old behavior was last-only) return candidates[-1] # Fallback: last event with any dict data for ev in reversed(events): if isinstance(ev.get("data"), dict): return ev["data"] raise RuntimeError("unparseable MCP response: " + _redact(t[:200])) def _validate_rpc_envelope(msg: Any, expect_id: int) -> dict[str, Any]: """CLI-02: require jsonrpc==2.0, id match, exactly one of result/error.""" if not isinstance(msg, dict): raise RuntimeError("JSON-RPC response is not an object") if msg.get("jsonrpc") != "2.0": raise RuntimeError( f'JSON-RPC response missing/invalid jsonrpc field (expected "2.0", got {msg.get("jsonrpc")!r})' ) if msg.get("id") != expect_id: raise RuntimeError( f"JSON-RPC id mismatch: expected {expect_id!r}, got {msg.get('id')!r}" ) has_result = "result" in msg has_error = "error" in msg if has_result == has_error: # both or neither raise RuntimeError( "JSON-RPC response must contain exactly one of 'result' or 'error'" ) return msg def rpc(method: str, params=None): global _id, _session _id += 1 req_id = _id body = json.dumps( {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params or {}} ).encode() headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "User-Agent": f"dbmcp.py/{CLIENT_VERSION} (Dead Brands)", } if _auth_headers_allowed(URL): headers["Authorization"] = f"Bearer {KEY}" if _session: headers["Mcp-Session-Id"] = _session req = urllib.request.Request(URL, data=body, method="POST", headers=headers) try: with _opener.open(req, timeout=60) as resp: sid = resp.headers.get("Mcp-Session-Id") or resp.headers.get("mcp-session-id") if sid: _session = sid raw = _read_capped(resp).decode(errors="replace") except urllib.error.HTTPError as e: try: err_body = _read_capped(e, limit=4096).decode(errors="replace") except Exception: err_body = "" raise RuntimeError(_redact(f"HTTP {e.code}: {err_body[:300]}")) from None except RuntimeError: raise except Exception as e: raise RuntimeError(_redact(str(e))) from None msg = _parse(raw, expect_id=req_id) msg = _validate_rpc_envelope(msg, req_id) if "error" in msg: raise RuntimeError(_redact(json.dumps(msg["error"]))) return msg.get("result") def ensure_session() -> None: """CLI-03: parse initialize result; surface initialized notification failures.""" global _session, _server_info, _negotiated_protocol, _server_capabilities if _session and _negotiated_protocol: return result = rpc( "initialize", { "protocolVersion": CLIENT_PROTOCOL, "capabilities": {}, "clientInfo": {"name": "dbmcp.py", "version": CLIENT_VERSION}, }, ) if not isinstance(result, dict): raise RuntimeError("initialize returned non-object result") _negotiated_protocol = result.get("protocolVersion") or CLIENT_PROTOCOL _server_capabilities = result.get("capabilities") if isinstance(result.get("capabilities"), dict) else {} _server_info = result.get("serverInfo") if isinstance(result.get("serverInfo"), dict) else {} # notifications/initialized — surface real failures (CLI-03) body = json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}).encode() headers = { "Content-Type": "application/json", "Accept": "application/json, text/event-stream", "User-Agent": f"dbmcp.py/{CLIENT_VERSION} (Dead Brands)", } if _auth_headers_allowed(URL): headers["Authorization"] = f"Bearer {KEY}" if _session: headers["Mcp-Session-Id"] = _session try: with _opener.open( urllib.request.Request(URL, data=body, method="POST", headers=headers), timeout=15, ) as resp: # 2xx / empty body are fine for notifications code = getattr(resp, "status", None) or resp.getcode() if code is not None and int(code) >= 400: raise RuntimeError(f"notifications/initialized HTTP {code}") _ = _read_capped(resp, limit=65536) except urllib.error.HTTPError as e: # Some servers return 202/204; HTTPError only for errors raise RuntimeError(_redact(f"notifications/initialized failed: HTTP {e.code}")) from None except RuntimeError: raise except Exception as e: raise RuntimeError( _redact(f"notifications/initialized failed: {e}") ) from None def call_tool(name: str, args=None): """CLI-05: prefer structuredContent; preserve all content blocks; check isError.""" ensure_session() result = rpc("tools/call", {"name": name, "arguments": args or {}}) if not isinstance(result, dict): return result is_err = bool(result.get("isError")) # Prefer structuredContent when present if "structuredContent" in result and result["structuredContent"] is not None: parsed: Any = result["structuredContent"] if is_err: sys.stderr.write(_redact(json.dumps(parsed, indent=2)) + "\n") raise SystemExit(2) return parsed content = result.get("content") if isinstance(content, list): if not content: if is_err: sys.stderr.write( _redact(json.dumps({"isError": True, "content": []}, indent=2)) + "\n" ) raise SystemExit(2) # empty content, no structured — return whole result return result blocks: list[Any] = [] text_bits: list[str] = [] for block in content: if not isinstance(block, dict): blocks.append(block) continue blocks.append(block) if block.get("type") == "text" and "text" in block: text_bits.append(str(block.get("text") or "")) # Backward-compat: if single text block that is JSON, return parsed object if len(content) == 1 and content[0].get("type") == "text": raw = content[0].get("text", "") try: parsed = json.loads(raw) if isinstance(raw, str) else {"_text": raw} except json.JSONDecodeError: parsed = {"_text": raw} if is_err: sys.stderr.write(_redact(json.dumps(parsed, indent=2)) + "\n") raise SystemExit(2) return parsed # Multiple / non-text blocks: preserve all out: dict[str, Any] = {"content": blocks} if text_bits: joined = "\n".join(text_bits) try: out["parsed_text"] = json.loads(joined) except json.JSONDecodeError: out["text"] = joined if is_err: sys.stderr.write(_redact(json.dumps(out, indent=2)) + "\n") raise SystemExit(2) return out if is_err: sys.stderr.write(_redact(json.dumps(result, indent=2)) + "\n") raise SystemExit(2) return result def cmd_quickstart() -> None: """Guest-first 2-step flow (no key required).""" print( f"dbmcp.py {CLIENT_VERSION} — Dead Brands Arena quickstart (guest-first)\n" "\n" "1) List tools (no key needed):\n" " python3 dbmcp.py tools\n" "\n" "2) Join a game as guest (no key — server returns a guest key + gtk_…; SAVE them):\n" " python3 dbmcp.py arena-join asteroid_dodge\n" " Same call via raw tool:\n" " python3 dbmcp.py call arena_join '{\"game\":\"asteroid_dodge\"}'\n" "\n" "Optional next steps:\n" " - Discord invite: python3 dbmcp.py community-join\n" " - Claim a handle: python3 dbmcp.py register yourbot\n" " - Health check: python3 dbmcp.py doctor\n" "\n" f"Endpoint: {_safe_url_display(URL)}\n" "Credit: Dead Brands\n" ) def cmd_doctor() -> int: """CLI DX: safe-to-paste diagnostics. Exit nonzero on any failure. Never echo secrets.""" checks: list[tuple[str, str, str]] = [] # (name, status, detail) failed = False def ok(name: str, detail: str = "") -> None: checks.append((name, "PASS", detail)) def bad(name: str, detail: str) -> None: nonlocal failed failed = True checks.append((name, "FAIL", _redact(detail))) # 1. Python env try: py = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" if sys.version_info < (3, 9): bad("python", f"need >=3.9, got {py}") else: ok("python", f"{py} ({sys.executable})") except Exception as e: bad("python", str(e)) # 2. URL scheme / localhost opt-in / no embedded secrets try: validated = _validate_url(URL, insecure_localhost=_INSECURE_LOCALHOST) scheme, host, _port = _origin(validated) ok( "url", f"{scheme}://{host}/… (insecure_localhost={bool(_INSECURE_LOCALHOST)})", ) except Exception as e: bad("url", str(e)) # still print report _print_doctor(checks) return 1 # 2b. P1-B custom-host + key gate (doctor reports; does not prompt) if KEY and not _host_is_official(validated) and not _ALLOW_CUSTOM_HOST: bad( "custom_host", f"DB_MCP_KEY set but host '{_origin(validated)[1]}' is not " f"{OFFICIAL_HOST}; pass --allow-custom-host to send the key", ) _print_doctor(checks) return 1 elif KEY and not _host_is_official(validated) and _ALLOW_CUSTOM_HOST: ok("custom_host", f"allowed for '{_origin(validated)[1]}'") else: ok( "custom_host", "official" if _host_is_official(validated) else "no key (guest ok)", ) # 3. Endpoint reachability (TCP/HTTP level via initialize path) # 4. MCP initialize # 5. tools/list try: ensure_session() ok( "initialize", f"protocol={_negotiated_protocol or '?'} " f"server={(_server_info or {}).get('name', '?')} " f"session={'set' if _session else 'none'}", ) except Exception as e: bad("initialize", str(e)) _print_doctor(checks) return 1 try: r = rpc("tools/list", {}) tools = (r or {}).get("tools") or [] names = sorted(t.get("name", "") for t in tools if isinstance(t, dict)) need = ["arena_join", "community_join", "arena_register", "arena_ping"] missing = [n for n in need if n not in names] if missing: bad("tools/list", f"count={len(names)} missing={missing}") else: ok("tools/list", f"count={len(names)} (arena_join/community_join/arena_register/arena_ping present)") except Exception as e: bad("tools/list", str(e)) # 6. Optional authenticated ping if KEY: try: # Prefer header auth already set via KEY; don't put key in args result = call_tool("arena_ping", {"message": "doctor"}) # Never dump raw if it somehow echoed a key — redact status = result.get("status") if isinstance(result, dict) else None auth_ok = None if isinstance(result, dict): for k in ("auth_ok", "authenticated", "valid", "key_valid"): if k in result: auth_ok = result[k] break detail = f"status={status!r}" if auth_ok is not None: detail += f" auth={auth_ok!r}" detail += " key_present=yes(redacted)" ok("arena_ping(auth)", detail) except Exception as e: bad("arena_ping(auth)", str(e)) else: checks.append(("arena_ping(auth)", "SKIP", "DB_MCP_KEY not set")) _print_doctor(checks) return 1 if failed else 0 def _print_doctor(checks: list[tuple[str, str, str]]) -> None: lines = [ f"dbmcp.py doctor — version {CLIENT_VERSION}", f"endpoint: {_safe_url_display(URL)}", f"key_env: {'set' if KEY else 'unset'}", f"session: {'set' if _session else 'unset'}", "---", ] for name, status, detail in checks: lines.append(f"[{status}] {name}" + (f" — {detail}" if detail else "")) lines.append("---") lines.append("Credit: Dead Brands (diagnostics redacted; safe to paste)") print("\n".join(lines)) def _parse_args(argv: list[str]) -> tuple[list[str], bool, bool, bool, bool]: insecure = _INSECURE_LOCALHOST show_version = False allow_custom = False do_selftest = False out: list[str] = [] for a in argv: if a == "--insecure-localhost": insecure = True elif a == "--allow-custom-host": allow_custom = True elif a == "--selftest": do_selftest = True elif a in ("--version", "-V"): show_version = True elif a in ("-h", "--help"): out.append(a) else: out.append(a) return out, insecure, show_version, allow_custom, do_selftest def _selftest() -> int: """P1-A/P1-B unit self-tests (no network). Exit 0 on all pass.""" global KEY, URL, _ALLOW_CUSTOM_HOST, _INSECURE_LOCALHOST fails: list[str] = [] def check(name: str, cond: bool, detail: str = "") -> None: if not cond: fails.append(f"FAIL {name}" + (f" — {detail}" if detail else "")) else: print(f"PASS {name}" + (f" — {detail}" if detail else "")) saved = (KEY, URL, _ALLOW_CUSTOM_HOST, _INSECURE_LOCALHOST) try: KEY = "" _ALLOW_CUSTOM_HOST = False _INSECURE_LOCALHOST = False # 1) userinfo rejected raised = False try: _validate_url( "https://user:s3cret@mcp.deadbrands.co/arena/mcp", insecure_localhost=False, ) except RuntimeError as e: raised = True check( "userinfo_rejected", "userinfo" in str(e).lower() or "user:pass" in str(e).lower(), str(e)[:120], ) check( "userinfo_no_secret_echo", "s3cret" not in str(e), "error must not echo password", ) if not raised: fails.append("FAIL userinfo_rejected — no exception") # 2) token= query rejected raised = False try: _validate_url( "https://mcp.deadbrands.co/arena/mcp?token=abc123secret", insecure_localhost=False, ) except RuntimeError as e: raised = True check( "token_query_rejected", "token" in str(e).lower() or "credential" in str(e).lower(), str(e)[:120], ) check( "token_query_no_secret_echo", "abc123secret" not in str(e), "error must not echo token value", ) if not raised: fails.append("FAIL token_query_rejected — no exception") for param in ("key", "password", "secret", "apikey", "access_token"): try: _validate_url( f"https://mcp.deadbrands.co/arena/mcp?{param}=leakme", insecure_localhost=False, ) fails.append(f"FAIL {param}_query_rejected — no exception") except RuntimeError as e: leaked = "leakme" in str(e) check(f"{param}_query_rejected", not leaked, "ok" if not leaked else "LEAKED") if leaked: fails.append(f"FAIL {param}_query_no_secret_echo") # 3) custom host without flag refuses to send key KEY = "ark_TESTKEY_NOT_REAL_12345678" URL = "https://evil.example/arena/mcp" _ALLOW_CUSTOM_HOST = False check( "custom_host_auth_blocked", _auth_headers_allowed(URL) is False, "key must not attach", ) refused = False try: _require_custom_host_consent(URL) except RuntimeError as e: refused = True check( "custom_host_refuses_without_flag", "refusing" in str(e).lower() or "allow-custom-host" in str(e), str(e)[:140], ) check( "custom_host_refuse_no_key_echo", KEY not in str(e) and "TESTKEY" not in str(e), ) if not refused: fails.append("FAIL custom_host_refuses_without_flag — no exception") _ALLOW_CUSTOM_HOST = True check( "custom_host_allowed_with_flag", _auth_headers_allowed(URL) is True, ) # official host always ok with key _ALLOW_CUSTOM_HOST = False URL = DEFAULT_URL check( "official_host_auth_ok", _auth_headers_allowed(URL) is True, ) # 4) redacted / safe display shows no secrets dirty = ( "https://user:p@ss@evil.example/path?token=sekrit&x=1#frag " "Authorization: Bearer ark_LIVEKEY_ABCDEFGH " "mcp-session-id: sess-ABCDEFGH1234" ) KEY = "ark_LIVEKEY_ABCDEFGH" red = _redact(dirty) check("redact_no_token_value", "sekrit" not in red) check("redact_no_bearer_key", "ark_LIVEKEY" not in red) check("redact_marks_present", "[REDACTED" in red) shown = _safe_url_display( "https://user:p@ssword@mcp.deadbrands.co/arena/mcp?token=sekrit#frag" ) check( "safe_url_no_query", "?" not in shown and "#" not in shown and "token" not in shown and "sekrit" not in shown, shown, ) # hostname-only netloc (userinfo stripped by urlparse hostname) check( "safe_url_no_userinfo", "user" not in shown and "password" not in shown and "p@ss" not in shown, shown, ) check( "safe_url_keeps_path", shown == "https://mcp.deadbrands.co/arena/mcp" or shown.startswith("https://mcp.deadbrands.co/arena/mcp"), shown, ) check("version_is_121", CLIENT_VERSION == "1.2.1", CLIENT_VERSION) finally: KEY, URL, _ALLOW_CUSTOM_HOST, _INSECURE_LOCALHOST = saved if fails: for f in fails: print(f) print(f"selftest: {len(fails)} failure(s)") return 1 print("selftest: all passed") return 0 def main(argv: list[str]) -> None: global URL, _INSECURE_LOCALHOST, _ALLOW_CUSTOM_HOST argv, insecure, show_version, allow_custom, do_selftest = _parse_args(argv) _INSECURE_LOCALHOST = insecure _ALLOW_CUSTOM_HOST = allow_custom if do_selftest: raise SystemExit(_selftest()) if show_version and not argv: print(f"dbmcp.py {CLIENT_VERSION} (Dead Brands)") return cmd0 = argv[0] if argv else "" # doctor owns its URL check (so FAIL lines land in the pasteable report). # selftest skips network URL validation (uses fixtures). if cmd0 not in ("doctor", "selftest"): URL = _validate_url(URL, insecure_localhost=insecure) _require_custom_host_consent(URL) if show_version: print(f"dbmcp.py {CLIENT_VERSION} (Dead Brands)") if not argv: return if not argv or argv[0] in ("-h", "--help"): usage(0) cmd = argv[0] if cmd == "doctor": raise SystemExit(cmd_doctor()) if cmd == "selftest": raise SystemExit(_selftest()) if cmd == "quickstart": cmd_quickstart() return if cmd == "tools": ensure_session() r = rpc("tools/list", {}) tools_in = (r or {}).get("tools") or [] tools_out = [] for t in tools_in: if not isinstance(t, dict) or not t.get("name"): continue entry = { "name": t["name"], "description": t.get("description") or "", } if t.get("title"): entry["title"] = t["title"] tools_out.append(entry) tools_out.sort(key=lambda x: x["name"]) payload = { "count": len(tools_out), "tools": tools_out, "endpoint": _safe_url_display(URL), "protocolVersion": _negotiated_protocol, "cliVersion": CLIENT_VERSION, } if _server_info: payload["serverInfo"] = _server_info print(json.dumps(payload, indent=2)) return if cmd == "call": if len(argv) < 2: usage() tool = argv[1] args: Any = {} if len(argv) > 2: try: args = json.loads(argv[2]) except json.JSONDecodeError as e: sys.stderr.write(f"bad JSON args: {e}\n") raise SystemExit(1) # CLI-06: arguments must be a JSON object if not isinstance(args, dict): sys.stderr.write( "call arguments must be a JSON object (got " f"{type(args).__name__})\n" ) raise SystemExit(1) print(json.dumps(call_tool(tool, args), indent=2)) return # Join family — confirmed live tool names: community_join, arena_join, arena_register if cmd == "join": sys.stderr.write( "note: 'join' is a backward-compat alias for community-join " "(tool community_join = Discord invite + rules). " "Use arena-join to play, or register for a named key.\n" ) print(json.dumps(call_tool("community_join", {}), indent=2)) return if cmd in ("community-join", "community_join"): print(json.dumps(call_tool("community_join", {}), indent=2)) return if cmd in ("arena-join", "arena_join"): game = argv[1] if len(argv) > 1 else "asteroid_dodge" print(json.dumps(call_tool("arena_join", {"game": game}), indent=2)) return if cmd == "register": if len(argv) < 2 or not argv[1].strip(): sys.stderr.write("usage: dbmcp.py register \n") raise SystemExit(1) handle = argv[1].strip() print(json.dumps(call_tool("arena_register", {"handle": handle}), indent=2)) return if cmd == "say": if not KEY: sys.stderr.write("DB_MCP_KEY required for say (register at /arena/register for an ark_ key)\n") raise SystemExit(1) text = " ".join(argv[1:]).strip() if not text: usage() print(json.dumps(call_tool("arena_lobby_say", {"text": text}), indent=2)) return if cmd == "read": limit = 20 if len(argv) > 1: try: limit = int(argv[1]) except ValueError: sys.stderr.write(f"read limit must be a positive integer (got {argv[1]!r})\n") raise SystemExit(1) if limit < 1: sys.stderr.write(f"read limit must be a positive integer (got {limit})\n") raise SystemExit(1) print(json.dumps(call_tool("arena_lobby_read", {"limit": limit}), indent=2)) return if cmd == "ping": message = " ".join(argv[1:]) or "pong" print(json.dumps(call_tool("arena_ping", {"message": message}), indent=2)) return usage() if __name__ == "__main__": try: main(sys.argv[1:]) except SystemExit: raise except Exception as e: sys.stderr.write(_redact(str(e)) + "\n") raise SystemExit(1)