# Census Preflight — GitHub Action (CENSUS-AUDIT/1)
# Not a malware scan, pentest, or permission review.
# PASS is not a sandbox. Review code and permissions before you connect.
#
# Copy this file into .github/workflows/census-preflight.yml
# Source of truth: https://mcpcensus.com/agent-setup/census-preflight.yml
# Contract:        https://mcpcensus.com/agent-setup/audit
#
# What it does: finds every MCP client config in the repo (.mcp.json,
# .cursor/mcp.json, .vscode/mcp.json, claude_desktop_config.json, mcp.json,
# .codex/config.toml) and POSTs each one to https://api.mcpcensus.com/v1/audit.
# The Census resolves each entry (url → hosted server, npx → npm package,
# uvx → PyPI package) and returns one PASS / REVIEW / BLOCK / UNKNOWN line per
# entry. Any BLOCK fails the job. REVIEW and UNKNOWN warn (set STRICT=1 to fail).
#
# CENSUS_API_KEY is optional. Keyless: one config of ≤25 entries per IP per UTC
# day (the same config again is free). With a free account key: 1,000 credits a
# month, audit = 1 credit per config per day. Prices: https://mcpcensus.com/pricing
# Only url/command/args are read; env and headers are stripped before posting.

name: Census Preflight

on:
  workflow_dispatch:
    inputs:
      policy_id:
        description: Built-in policy id
        required: false
        default: builtin:baseline
        type: string
      strict:
        description: Fail on REVIEW / UNKNOWN too
        required: false
        default: "0"
        type: string
  pull_request:
    paths:
      - "**/mcp.json"
      - "**/.mcp.json"
      - "**/.cursor/mcp.json"
      - "**/.vscode/mcp.json"
      - "**/claude_desktop_config.json"
      - "**/.codex/config.toml"
  schedule:
    # nightly re-audit: liveness, tool lists and registry status change daily
    - cron: "17 3 * * *"

jobs:
  audit:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Census audit of MCP configs
        env:
          CENSUS_API_KEY: ${{ secrets.CENSUS_API_KEY }}
          POLICY_ID: ${{ github.event.inputs.policy_id || 'builtin:baseline' }}
          STRICT: ${{ github.event.inputs.strict || '0' }}
        run: |
          set -euo pipefail
          python3 - <<'PY'
          import json, os, sys, urllib.error, urllib.request
          from pathlib import Path

          API = "https://api.mcpcensus.com/v1/audit"
          policy = os.environ.get("POLICY_ID") or "builtin:baseline"
          strict = (os.environ.get("STRICT") or "0").strip() in ("1", "true", "yes")
          key = (os.environ.get("CENSUS_API_KEY") or "").strip()

          NAMES = {".mcp.json", "mcp.json", "claude_desktop_config.json"}
          configs = []
          for p in Path(".").rglob("*"):
              if not p.is_file() or "node_modules" in p.parts or ".git" in p.parts:
                  continue
              if p.name in NAMES or (p.name == "config.toml" and p.parent.name == ".codex"):
                  configs.append(p)
          configs.sort()

          if not configs:
              print("No MCP client config files found (.mcp.json, .cursor/mcp.json, .vscode/mcp.json, claude_desktop_config.json, .codex/config.toml).")
              print("Not a malware scan. Nothing to evaluate.")
              sys.exit(0)

          def strip_secrets(obj):
              # Only url / command / args / type are needed. Drop env, headers, tokens.
              if isinstance(obj, dict):
                  return {k: strip_secrets(v) for k, v in obj.items() if k not in ("env", "headers", "http_headers", "bearer_token_env_var", "auth", "oauth", "inputs")}
              if isinstance(obj, list):
                  return [strip_secrets(x) for x in obj]
              return obj

          total_block = total_review = total_unknown = failed = 0
          print(f"Auditing {len(configs)} config file(s) under {policy}")
          print("Not a malware scan. PASS is not a sandbox.")
          for cfg in configs:
              raw = cfg.read_text(encoding="utf-8", errors="replace")
              if cfg.suffix == ".toml":
                  body, ctype = raw.encode(), "text/plain"
              else:
                  try:
                      doc = strip_secrets(json.loads(raw))
                  except Exception as e:
                      print(f"::warning::{cfg}: not valid JSON ({e}); skipped")
                      continue
                  body, ctype = json.dumps({"config": doc, "policy_id": policy, "strict": strict}).encode(), "application/json"
              headers = {"content-type": ctype, "accept": "application/json",
                         "user-agent": "census-preflight-action/2 (+https://mcpcensus.com/agent-setup/census-preflight.yml)"}
              if key:
                  headers["x-api-key"] = key
              req = urllib.request.Request(API, data=body, headers=headers, method="POST")
              try:
                  with urllib.request.urlopen(req, timeout=60) as r:
                      data = json.loads(r.read().decode())
              except urllib.error.HTTPError as e:
                  payload = e.read().decode()[:800]
                  print(f"::error::{cfg} HTTP {e.code}: {payload}")
                  if e.code == 402:
                      print("Empty wallet. Buy credits at https://mcpcensus.com/pricing")
                  if e.code == 429:
                      print("Keyless allowance used for today. Add CENSUS_API_KEY (free at https://mcpcensus.com/account).")
                  failed += 1
                  continue
              except Exception as e:
                  print(f"::error::{cfg} request failed: {e}")
                  failed += 1
                  continue

              print(f"\n{cfg}  (config_digest {data.get('config_digest','?')[:23]}…)")
              for e in data.get("entries", []):
                  d = e.get("decision", "UNKNOWN")
                  name = e.get("server_name") or "-"
                  reasons = ",".join(r.get("code", "") for r in e.get("reasons", []))
                  line = f"  {e.get('alias')} → {d}  {name}  [{reasons}]"
                  if d == "BLOCK":
                      print(f"::error::{cfg}: {e.get('alias')} BLOCK ({name}) — do not connect. Public-evidence policy miss, not a malware verdict.")
                      total_block += 1
                  elif d == "REVIEW":
                      print(f"::warning::{cfg}: {e.get('alias')} REVIEW ({name}) — ask a human. PASS is not a sandbox either.")
                      total_review += 1
                  elif d == "UNKNOWN":
                      print(f"::warning::{cfg}: {e.get('alias')} UNKNOWN — the Census holds no row for this entry ({reasons}).")
                      total_unknown += 1
                  print(line)
              s = data.get("summary", {})
              print(f"  summary: pass={s.get('pass',0)} review={s.get('review',0)} block={s.get('block',0)} unknown={s.get('unknown',0)} exit_code={data.get('exit_code')}")

          print(f"\nblock={total_block} review={total_review} unknown={total_unknown} failed_requests={failed}")
          if total_block or failed:
              sys.exit(1)
          if strict and (total_review or total_unknown):
              sys.exit(2)
          PY