krow_backend/scripts/verify-deploy.py

#!/usr/bin/env python3
"""
Verify a deployed Krow API, endpoint by endpoint.

    KROW_EMAIL=you@example.com KROW_PASSWORD=... \
      python3 scripts/verify-deploy.py https://mcp.krowforce.com

    make verify-deploy BASE=https://mcp.krowforce.com

Credentials come from the environment, never from an argument, so they do not
land in shell history or in a process list.

READ-ONLY by default. The two write paths (hire, assignment) are exercised only
with --write, because they change tenant data and a smoke test that mutates the
thing it is checking is not a smoke test.

Why this exists: this API runs its auth middleware BEFORE routing, so an
unauthenticated probe answers 401 for every path — including paths that do not
exist. `curl` against a deployed host therefore cannot tell a missing endpoint
from a guarded one, and the only honest check is an authenticated one.

Exit code is non-zero if any check fails.
"""
import json, os, sys, time, urllib.request, urllib.parse, urllib.error, http.cookiejar

BASE = (sys.argv[1] if len(sys.argv) > 1 and not sys.argv[1].startswith("-")
        else os.environ.get("KROW_BASE_URL", "http://127.0.0.1:8080")).rstrip("/")
WRITE = "--write" in sys.argv

class BrowserLikePolicy(http.cookiejar.DefaultCookiePolicy):
    """Accept Secure cookies over http://localhost, as every browser does.

    Browsers treat localhost as a potentially-trustworthy origin, so a Secure
    cookie set through a dev-server proxy is stored and sent. Python's default
    policy refuses it, which makes a perfectly working frontend look like a
    broken session: login returns 200 and the very next request is 401.

    Only localhost. Anywhere else, a Secure cookie over plaintext is refused as
    it should be.
    """

    @staticmethod
    def _trustworthy(request):
        host = urllib.parse.urlparse(request.get_full_url()).hostname or ""
        return host in ("localhost", "127.0.0.1", "::1", "[::1]")

    def set_ok_secure(self, cookie, request):
        return self._trustworthy(request) or super().set_ok_secure(cookie, request)

    def return_ok_secure(self, cookie, request):
        return self._trustworthy(request) or super().return_ok_secure(cookie, request)


jar = http.cookiejar.CookieJar(policy=BrowserLikePolicy())
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
results = []

def call(method, path, body=None, accept="application/json", timeout=45):
    """Returns (status, text, headers). Never raises for an HTTP status."""
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(BASE + path, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", accept)
    try:
        with opener.open(req, timeout=timeout) as r:
            return r.status, r.read().decode("utf-8", "replace"), dict(r.headers)
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace"), dict(e.headers)
    except Exception as e:
        return 0, f"{type(e).__name__}: {e}", {}

def check(name, ok, detail=""):
    results.append((name, bool(ok), detail))
    print(f"[{'  ok  ' if ok else ' FAIL '}] {name}" + (f" — {detail}" if detail else ""))
    return ok

def group(title):
    print(f"\n── {title} " + "─" * max(0, 66 - len(title)))

def as_json(text):
    try:
        return json.loads(text)
    except Exception:
        return None

# resource -> the operations it declares (internal/domain/resources_gen.go).
# Anything not declared is deliberately unregistered: "the database having a
# table is never a reason for an endpoint to exist" (api.go). `badges` declares
# nothing at all, so every badges path is correctly a 404.
RESOURCE_OPS = {
    "job-postings":     ["List", "Get", "Create", "Update"],
    "job-applications": ["List", "Create", "Update", "Delete"],
    "ai-interviews":    ["List", "Create"],
    "staff":            ["List", "Create", "Update"],
    "worker-profiles":  ["List", "Create", "Update"],
    "employee-roles":   ["List", "Get", "Create", "Update"],
    "courses":          ["List", "Get", "Create", "Update"],
    "learning-paths":   ["List"],
    "role-categories":  ["List", "Create"],
    "certifications":   ["List", "Create", "Delete"],
    "user-activity":    ["List", "Create"],
    "evidence":         ["List", "Create", "Update"],
    "assignments":      ["List", "Create"],
    "shift-records":    ["List"],
    "badges":           [],
}
RESOURCES = [r for r, ops in RESOURCE_OPS.items() if "List" in ops]

print(f"Verifying {BASE}   ({'read/write' if WRITE else 'read-only'})")

# ── 1. Reachable, and guarded ────────────────────────────────────────────────
group("Reachable and guarded")
s, t, _ = call("GET", "/health")
health = as_json(t) or {}
check("/health answers 200", s == 200, f"{s} {health.get('status', t[:40])}")
check("/health reports a healthy database", health.get("status") == "ok",
      f"status={health.get('status')} (degraded = schema unmigrated or dirty)")
s, _, _ = call("GET", "/api/v1/job-postings")
check("a protected endpoint refuses an anonymous caller", s in (401, 403), f"{s}")

# ── 2. Sign in ───────────────────────────────────────────────────────────────
group("Authentication")
email, password = os.environ.get("KROW_EMAIL"), os.environ.get("KROW_PASSWORD")
if not (email and password):
    print("\nKROW_EMAIL / KROW_PASSWORD are not set — cannot check anything behind auth.")
    print("Everything below needs a session. Set them and re-run.")
    sys.exit(2)

s, t, _ = call("POST", "/api/v1/auth/login", {"email": email, "password": password})
if not check("sign-in succeeds", s == 200, str(s)):
    print("\nNo session — stopping. Every remaining check needs one.")
    sys.exit(1)
check("a session cookie was set", len(jar) > 0, f"{len(jar)} cookie(s)")

s, t, _ = call("GET", "/api/v1/me")
me = (as_json(t) or {}).get("data") or {}
check("GET /api/v1/me returns the signed-in user", s == 200 and bool(me), f"{s}")
check("...and it is the account that signed in",
      str(me.get("email", "")).lower() == email.lower(), me.get("email", "?"))
role = me.get("role", "?")
print(f"       signed in as {me.get('email','?')} (role: {role})")

s, _, _ = call("GET", "/api/v1/me/preferences")
check("GET /api/v1/me/preferences", s == 200, f"{s}")

# Which build is actually serving. Without this, "did my deploy land?" has no
# answer and a redeploy that silently rolled back looks identical to one that
# worked. Set KROW_EXPECT_VERSION to make a stale deployment a failure.
s, t, _ = call("GET", "/api/v1/version")
build = (as_json(t) or {}).get("data") or {}
running = build.get("version", "")
check("GET /api/v1/version reports the running build", s == 200 and bool(running),
      f"{s}, version={running or 'none'}, env={build.get('env')}, "
      f"endpoints={build.get('endpoints')}")
if running == "unknown":
    check("...and the build was actually stamped", False,
          "reports \"unknown\" — built without -X main.version, so it cannot be traced")
expected = os.environ.get("KROW_EXPECT_VERSION")
if expected:
    check("...and it is the build you expected", running == expected,
          f"running {running!r}, expected {expected!r}")

# ── 3. Every resource collection ─────────────────────────────────────────────
group(f"Resource endpoints ({len(RESOURCES)} collections)")
first_ids = {}
for r in RESOURCES:
    s, t, _ = call("GET", f"/api/v1/{r}")
    body = as_json(t) or {}
    recs = body.get("data")
    ok = s == 200 and isinstance(recs, list) and isinstance(body.get("meta"), dict)
    total = (body.get("meta") or {}).get("total")
    check(f"GET /api/v1/{r}", ok, f"{s}" + (f", {len(recs)} records, meta.total={total}" if ok else f" {t[:90]}"))
    if ok and recs:
        first_ids[r] = recs[0].get("id")

group("Reading one record by id (only where the resource declares Get)")
for r, rid in first_ids.items():
    s, t, _ = call("GET", f"/api/v1/{r}/{rid}")
    if "Get" in RESOURCE_OPS[r]:
        check(f"GET /api/v1/{r}/{{id}}", s == 200, f"{s}")
    else:
        check(f"GET /api/v1/{r}/{{id}} is refused — it declares no Get",
              s in (404, 405), f"{s}")

group("A resource that declares nothing exposes nothing")
for r, ops in RESOURCE_OPS.items():
    if ops:
        continue
    s, _, _ = call("GET", f"/api/v1/{r}")
    check(f"GET /api/v1/{r} → 404", s == 404, f"{s}")

group("An id that does not exist is 404, not 500")
s, _, _ = call("GET", "/api/v1/job-postings/00000000-0000-0000-0000-000000000000")
check("unknown id → 404", s == 404, f"{s}")
s, _, _ = call("GET", "/api/v1/job-postings/not-a-uuid")
check("malformed id → 4xx, never 5xx", 400 <= s < 500, f"{s}")

# ── 4. The agent layer ───────────────────────────────────────────────────────
group("Agent and skill registry")
s, t, _ = call("GET", "/api/v1/agent-definitions")
body = as_json(t) or {}
agents = body.get("data") if isinstance(body, dict) else body
agents = agents if isinstance(agents, list) else []
check("GET /api/v1/agent-definitions", s == 200 and isinstance(agents, list), f"{s}, {len(agents)} agents")
if agents:
    print("       " + ", ".join(sorted(str(a.get("definition_id") or a.get("id")) for a in agents)))
    # Two different keys, deliberately. The registry endpoint is a CRUD resource
    # keyed by uuid (repo.GetAgent: WHERE id = $1::uuid); the run endpoint is
    # addressed by the stable definition_id a spec author writes. Passing the
    # definition_id to the registry endpoint is a 404, which is correct.
    row_uuid = agents[0].get("id")
    s, _, _ = call("GET", f"/api/v1/agent-definitions/{row_uuid}")
    check("GET /api/v1/agent-definitions/{uuid}", s == 200, f"{s}")
    s, _, _ = call("GET", f"/api/v1/agent-definitions/{agents[0].get('definition_id')}")
    check("...and the definition_id is not a uuid, so it is refused there", s == 404, f"{s}")

s, t, _ = call("GET", "/api/v1/skill-definitions")
body = as_json(t) or {}
skills = body.get("data") if isinstance(body, dict) else body
skills = skills if isinstance(skills, list) else []
check("GET /api/v1/skill-definitions", s == 200, f"{s}, {len(skills)} skills")

s, t, _ = call("GET", "/api/v1/owliver/suggestions?page=control-center")
check("GET /api/v1/owliver/suggestions?page=...", s == 200, f"{s}")
s, _, _ = call("GET", "/api/v1/owliver/suggestions")
check("...and it requires a page rather than guessing one", s == 400, f"{s}")
s, _, _ = call("GET", "/api/v1/owliver/suggestions?page=not-a-real-page")
check("...and rejects a page that does not exist", s == 400, f"{s}")

# ── 5. An actual agent run ───────────────────────────────────────────────────
group("Running an agent (this calls the model — it costs tokens)")
run_id = None
if not agents:
    check("an agent run completes", False, "no agents are published on this deployment")
else:
    aid = agents[0].get("definition_id") or agents[0].get("id")
    t0 = time.time()
    s, t, _ = call("POST", f"/api/v1/agents/{aid}/runs",
                   {"input": "What can you help me with? Answer in one sentence."})
    body = as_json(t) or {}
    took = time.time() - t0
    ok = s == 200 and body.get("termination") is not None
    check(f"POST /api/v1/agents/{aid}/runs", ok,
          f"{s}, termination={body.get('termination')}, {took:.1f}s" if ok else f"{s} {t[:160]}")
    if ok:
        run_id = body.get("runId") or body.get("run_id")
        check("...the run terminated cleanly",
              body.get("termination") in ("Completed", "ConfirmationPending"),
              str(body.get("termination")))
        check("...and it produced an answer",
              bool(body.get("output") or body.get("message") or body.get("confirmations")),
              (body.get("output") or body.get("message") or "")[:70] or "confirmation proposed")
        usage = body.get("usage") or {}
        check("...with token accounting attached",
              (usage.get("inputTokens", 0) or 0) > 0,
              f"in={usage.get('inputTokens')} out={usage.get('outputTokens')}")

    if run_id:
        s, t, _ = call("GET", f"/api/v1/runs/{run_id}")
        rb = as_json(t) or {}
        check("GET /api/v1/runs/{id} returns the trajectory", s == 200, f"{s}")
        entries = rb.get("entries")
        check("...with the trajectory persisted",
              isinstance(entries, list) and len(entries) > 0,
              f"{len(entries) if isinstance(entries, list) else 0} entries, "
              f"termination={rb.get('termination')}, model={rb.get('model') or '?'}")
        check("...and the run pins the agent version it started with",
              isinstance(rb.get("agentVersion"), int) and rb["agentVersion"] > 0,
              f"v{rb.get('agentVersion')}")

    # Streaming is the path the chat panel actually uses.
    s, t, h = call("POST", f"/api/v1/agents/{aid}/runs",
                   {"input": "Say hello in five words."}, accept="text/event-stream")
    ctype = (h.get("Content-Type") or h.get("content-type") or "")
    check("the same endpoint streams on Accept: text/event-stream",
          s == 200 and "event-stream" in ctype, f"{s}, content-type={ctype or 'none'}")
    check("...and the stream carries more than one event",
          t.count("data:") > 1, f"{t.count('data:')} data frames")

# ── 6. Writes (only with --write) ────────────────────────────────────────────
group("Write paths")
if not WRITE:
    print("       skipped — re-run with --write to exercise hire and assignment")
else:
    s, t, _ = call("POST", "/api/v1/job-postings/x/assignments", {})
    check("POST assignments rejects a bad request rather than 500", 400 <= s < 500, f"{s}")
    s, t, _ = call("POST", "/api/v1/job-applications/x/hire", {})
    check("POST hire rejects a bad request rather than 500", 400 <= s < 500, f"{s}")

# ── 7. Sign out ──────────────────────────────────────────────────────────────
group("Sign out")
s, _, _ = call("POST", "/api/v1/auth/logout")
check("POST /api/v1/auth/logout", s in (200, 204), f"{s}")
s, _, _ = call("GET", "/api/v1/me")
check("the session is dead afterwards", s in (401, 403), f"{s}")

# ── Summary ──────────────────────────────────────────────────────────────────
failed = [r for r in results if not r[1]]
print(f"\n{len(results) - len(failed)}/{len(results)} checks passed")
if failed:
    print("\nFailed:")
    for name, _, detail in failed:
        print(f"  - {name}{f' ({detail})' if detail else ''}")
    sys.exit(1)