feat(validate): authenticate the probe on auth-protected apps + 🔑 auth_failed benchmark outcome

The test-apply couldn't validate a band-aid on an auth-protected endpoint: the probe ran
unauthenticated, hit 401, and every protected finding was mislabeled auth_required. Now the
validation probe can authenticate, so an exploit is demonstrated (real 200 baseline → 403 block)
instead of a bare 401.

A — auto token capture + injection: probe_from_spec captures a bearer token from a login/setup
JSON response (access_token/token/jwt/…, nested one level) and injects Authorization: Bearer on
the exploit/legit requests. Cookie apps already carry the session via the shared jar; token apps
(VAmPI/crAPI) now work too.

B — operator-supplied login: VPCOPILOT_PROBE_USER/PASS (or a bearer VPCOPILOT_PROBE_TOKEN, +
optional LOGIN_PATH/USER_FIELD/PASS_FIELD) log in first over the shared session and FAIL LOUD
(honest auth_failed) if no session is established — so a wrong credential can't masquerade as
"needs auth". apply._probe_auth_from_env() is the single chokepoint every _run_validation reads,
reaching apply_* + refiner + the console (which load_dotenv's before each action) with no param
threading. CLI: apply --probe-user/--probe-pass/--probe-login-path/--probe-token. Keys documented
in .env.example and surfaced (masked) in the console Setup panel.

auth_failed is a distinct benchmark outcome from auth_required (creds supplied but login failed vs
no creds at all) — counted separately, excluded from failures. Rendered as 🔑 login-failed in the
⑥ Benchmark full matrix, outcome bar (plum) + legend, per-finding outcome column, and the live ③
Mitigate result, each pointing at the VPCOPILOT_PROBE_* knob to fix it.

Tests: probe token capture/injection, cookie login, loud auth-failure, guessed-login supersede,
env builder, and the auth_failed benchmark classification. Full non-live suite green (130), ruff
clean, console JS node --check clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Daniel Henley

.env.example

@@ -22,6 +22,13 @@ GITHUB_TOKEN=        # repo-scoped PAT for opening PRs (or use `gh auth login`)
 # VPCOPILOT_PROTECTED_LBS=nimbus-www   # comma-separated LBs that refuse mutation without an override
 # VPCOPILOT_REFINE_ATTEMPTS=3          # max self-heal attempts when a policy doesn't block the exploit
 # VPCOPILOT_REQUIRE_PROBE=1            # fail-closed: don't validate a finding that has no derived probe
+# --- Validation auth (auth-protected targets: let the probe log in so it can demonstrate the exploit) ---
+# VPCOPILOT_PROBE_USER=                # username the validation probe logs in with (cookie or token app)
+# VPCOPILOT_PROBE_PASS=                # its password
+# VPCOPILOT_PROBE_LOGIN_PATH=/api/login  # login endpoint (default /api/login)
+# VPCOPILOT_PROBE_TOKEN=               # OR a bearer token to inject directly (instead of user/pass)
+# VPCOPILOT_PROBE_USER_FIELD=username  # login JSON field names, if the app isn't username/password
+# VPCOPILOT_PROBE_PASS_FIELD=password
 # Console action defaults (so the UI isn't pinned to one app):
 # VPCOPILOT_DEFAULT_LB=your-lb
 # VPCOPILOT_DEFAULT_URL=https://your-app.example.com

src/vpcopilot/apply.py

@@ -178,19 +178,47 @@ def _load_probe(out_dir: str, finding_id) -> dict | None:
     return None
 
 
+def _probe_auth_from_env() -> dict | None:
+    """Operator-supplied validation auth (Layer B), read once from the environment so it reaches
+    every apply_* / refiner validation through the single `_run_validation` chokepoint — no param
+    threaded through each signature, and the console picks it up because it load_dotenv's before
+    each action. Returns None when nothing is set (validation then runs unauthenticated, as before).
+
+      VPCOPILOT_PROBE_TOKEN       a bearer token injected as `Authorization: Bearer <token>`
+      VPCOPILOT_PROBE_USER/PASS   credentials the probe logs in with first (cookie or token)
+      VPCOPILOT_PROBE_LOGIN_PATH  login endpoint (default /api/login)
+      VPCOPILOT_PROBE_USER_FIELD / _PASS_FIELD   JSON field names, if the app isn't username/password
+    """
+    import os
+    token = os.environ.get("VPCOPILOT_PROBE_TOKEN")
+    user, pw = os.environ.get("VPCOPILOT_PROBE_USER"), os.environ.get("VPCOPILOT_PROBE_PASS")
+    if not (token or (user and pw)):
+        return None
+    auth: dict = {"login_path": os.environ.get("VPCOPILOT_PROBE_LOGIN_PATH", "/api/login")}
+    if token:
+        auth["token"] = token
+    if user and pw:
+        auth.update(username=user, password=pw,
+                    user_field=os.environ.get("VPCOPILOT_PROBE_USER_FIELD", "username"),
+                    pass_field=os.environ.get("VPCOPILOT_PROBE_PASS_FIELD", "password"))
+    return auth
+
+
 def _run_validation(target_url: str, finding_id, out_dir: str, fallback, log, *,
-                    require_probe: bool = False) -> dict:
+                    require_probe: bool = False, auth: dict | None = None) -> dict:
     """Normalized {exploit_status, exploit_blocked, legit_ok}. Prefers the finding's derived probe
     (works on any app). B5 — fail closed: when no finding-probe exists, the Nimbus-specific fallback
     is only meaningful against the Nimbus demo, so we (a) log a loud warning and tag the result
     `fallback`, and (b) if require_probe (param or VPCOPILOT_REQUIRE_PROBE) is set, refuse to fall
     back at all and return a non-passing `no_probe` result — never silently 'validate' a real app
-    with a probe that hits the wrong endpoints."""
+    with a probe that hits the wrong endpoints. `auth` (else VPCOPILOT_PROBE_*) authenticates the
+    probe so an auth-protected endpoint validates for real instead of returning a bare 401."""
     import os
     from .probe import probe_from_spec
+    auth = auth or _probe_auth_from_env()
     spec = _load_probe(out_dir, finding_id)
     if spec:
-        return probe_from_spec(target_url, spec, log=log)
+        return probe_from_spec(target_url, spec, log=log, auth=auth)
     require_probe = require_probe or os.environ.get("VPCOPILOT_REQUIRE_PROBE", "").lower() in ("1", "true", "yes")
     if require_probe:
         log("  ⚠ no finding-derived probe and require_probe set — cannot validate this target; "
@@ -215,8 +243,9 @@ def _log_baseline(before: dict, log: Callable) -> None:
         log("  ⚠ baseline exploit → 404: the finding's endpoint likely does not exist on this target "
             "— the band-aid can't be validated (check the endpoint). Applying anyway.")
     elif st == 401:
-        log("  ⚠ baseline exploit → 401: the endpoint requires authentication, so the unauthenticated "
-            "probe can't demonstrate the exploit — the band-aid can't be validated this way. Applying anyway.")
+        log("  ⚠ baseline exploit → 401: the endpoint requires authentication. Supply validation "
+            "credentials so the probe authenticates — VPCOPILOT_PROBE_USER/PASS (or --probe-user/"
+            "--probe-pass), or a bearer VPCOPILOT_PROBE_TOKEN — then re-run. Applying anyway.")
 
 
 def apply_service_policy(lb: str, policy_name: str, target_url: str, *,

src/vpcopilot/bench_model.py

@@ -71,6 +71,7 @@ def build(out_dir: str, model_tag: str, target: str = "", config_path: str | Non
             "passed": bool(a.get("passed")), "attempts": a.get("attempts") or 1,
             "unfixable": bool(a.get("unfixable")),
             "before_status": (ba.get("before") or {}).get("exploit_status"),
+            "before_auth_failed": bool((ba.get("before") or {}).get("auth_failed")),
             "after_status": (ba.get("after") or {}).get("exploit_status"),
             "reason": a.get("reason"),
         })
@@ -81,10 +82,12 @@ def build(out_dir: str, model_tag: str, target: str = "", config_path: str | Non
     from .controls import CONTROLS
 
     def _outcome(p):
+        if p.get("before_auth_failed"):    # operator supplied creds but the login didn't establish a
+            return "auth_failed"            # session (wrong credential/path) — distinct from auth_required
         if p.get("before_status") == 404:  # baseline exploit 404'd — the finding's endpoint doesn't
             return "endpoint_missing"       # exist on the target; the band-aid can't be validated
-        if p.get("before_status") == 401:  # exploit needs auth — the unauthenticated probe can't
-            return "auth_required"          # demonstrate it, so the band-aid can't be validated
+        if p.get("before_status") == 401:  # exploit needs auth and none was supplied — the probe
+            return "auth_required"          # can't demonstrate it, so the band-aid can't be validated
         kind = CONTROLS[p["control"]].validation if p["control"] in CONTROLS else "live"
         if kind in ("config", "behavioral"):  # config-level defense-in-depth (waf / data_guard /
             return "applied"                   # rate_limit / bot / malicious_user): enabled = applied;
@@ -102,6 +105,7 @@ def _outcome(p):
     applied = sum(1 for p in per if p["outcome"] == "applied")
     missing = sum(1 for p in per if p["outcome"] == "endpoint_missing")
     needs_auth = sum(1 for p in per if p["outcome"] == "auth_required")
+    auth_failed = sum(1 for p in per if p["outcome"] == "auth_failed")
     healed = sum(1 for p in per if p["passed"] and (p["attempts"] or 1) > 1)
     atts = [p["attempts"] for p in per if p["attempts"]]
     v = metrics.get("verify") or {}
@@ -127,7 +131,8 @@ def _outcome(p):
             "blocked": blocked,               # real single-request exploit block (→403)
             "applied_behavioral": applied,     # passed at config level; not single-request testable
             "endpoint_missing": missing,       # baseline exploit 404'd — finding's endpoint doesn't exist
-            "auth_required": needs_auth,       # baseline exploit 401'd — needs auth; probe can't validate
+            "auth_required": needs_auth,       # baseline exploit 401'd — no creds supplied; probe can't validate
+            "auth_failed": auth_failed,        # creds WERE supplied but the login failed — fix the credential
             "block_rate": round(blocked / attempted, 2) if attempted else None,
             "pass_rate": round(passed / attempted, 2) if attempted else None,
             "self_healed": healed,
@@ -165,7 +170,7 @@ def to_markdown(b: dict) -> str:
               f"- attempted **{pq['attempted']}** · **blocked** (real single-request exploit→403) "
               f"**{pq['blocked']}** ({br}) · applied-but-behavioral {pq['applied_behavioral']} · "
               f"failed {pq['failed']} · endpoint-missing {pq.get('endpoint_missing', 0)} · "
-              f"needs-auth {pq.get('auth_required', 0)} · "
+              f"needs-auth {pq.get('auth_required', 0)} · login-failed {pq.get('auth_failed', 0)} · "
               f"self-healed {pq['self_healed']} · avg attempts {pq['avg_attempts'] or '—'}",
               "",
               "> _blocked_ = a fired exploit was stopped at the edge (per-request positive security). "
@@ -178,7 +183,8 @@ def to_markdown(b: dict) -> str:
                   "|---|---|---|---|---|---|---|"]
         icon = {"blocked": "✅ blocked", "applied": "🟡 applied (behavioral)",
                 "not_blocked": "❌ not blocked", "unfixable": "⚠️ unfixable",
-                "endpoint_missing": "🚫 endpoint missing", "auth_required": "🔒 needs auth"}
+                "endpoint_missing": "🚫 endpoint missing", "auth_required": "🔒 needs auth",
+                "auth_failed": "🔑 login failed"}
         for p in pq["per_finding"]:
             ba = (f"{p['before_status']}→{p['after_status']}" if p["before_status"] is not None else "—")
             lines.append(f"| {p['finding_id']} | {p['severity'] or '—'} | {p['vuln_class'] or '—'} | "

src/vpcopilot/cli.py

@@ -127,9 +127,20 @@ def apply(
     refine: bool = typer.Option(True, "--refine/--no-refine", help="refine the policy until it actually blocks the exploit (default on)"),
     refine_attempts: int = typer.Option(None, "--refine-attempts", help="max refine attempts (default $VPCOPILOT_REFINE_ATTEMPTS or 3)"),
     allow_protected_lb: bool = typer.Option(False, "--allow-protected-lb", help="permit mutating a protected LB"),
+    probe_user: str = typer.Option(None, "--probe-user", help="validation login username for an auth-protected app (or set VPCOPILOT_PROBE_USER)"),
+    probe_pass: str = typer.Option(None, "--probe-pass", help="validation login password (or VPCOPILOT_PROBE_PASS)"),
+    probe_login_path: str = typer.Option(None, "--probe-login-path", help="login endpoint path, default /api/login (or VPCOPILOT_PROBE_LOGIN_PATH)"),
+    probe_token: str = typer.Option(None, "--probe-token", help="bearer token for validation instead of user/pass (or VPCOPILOT_PROBE_TOKEN)"),
     out: str = typer.Option("out", help="output directory"),
 ):
     """Gated apply: (create from scan) -> snapshot -> self-test -> attach -> validate -> refine/rollback."""
+    import os
+    # Auth for the validation probe (auth-protected targets). Set env so it reaches every
+    # _run_validation call inside apply/refine via _probe_auth_from_env — the single chokepoint.
+    for flag, key in ((probe_user, "VPCOPILOT_PROBE_USER"), (probe_pass, "VPCOPILOT_PROBE_PASS"),
+                      (probe_login_path, "VPCOPILOT_PROBE_LOGIN_PATH"), (probe_token, "VPCOPILOT_PROBE_TOKEN")):
+        if flag:
+            os.environ[key] = flag
     logf = lambda m: rprint(f"[dim]{m}[/dim]")  # noqa: E731
     kw = dict(dry_run=dry_run, keep=keep, allow_protected=allow_protected_lb, probe=probe, out_dir=out, log=logf)
     if from_scan and refine and not dry_run and not create_only:

src/vpcopilot/console/app.py

@@ -57,10 +57,14 @@ def _active_tag() -> str:
     except Exception:  # noqa: BLE001
         return "default"
 
-SECRET_KEYS = {"ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "XC_API_TOKEN", "GITHUB_TOKEN"}
+SECRET_KEYS = {"ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "XC_API_TOKEN", "GITHUB_TOKEN",
+               "VPCOPILOT_PROBE_PASS", "VPCOPILOT_PROBE_TOKEN"}
 MANAGED_KEYS = [
     "ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "OLLAMA_API_BASE",
     "XC_API_URL", "XC_API_TOKEN", "XC_NAMESPACE", "GITHUB_TOKEN",
+    # Validation auth for an auth-protected target — the probe logs in so it can demonstrate the
+    # exploit (loaded before every apply via load_dotenv, so a change here takes effect next run).
+    "VPCOPILOT_PROBE_USER", "VPCOPILOT_PROBE_PASS", "VPCOPILOT_PROBE_LOGIN_PATH", "VPCOPILOT_PROBE_TOKEN",
 ]
 
 app = FastAPI(title="virtual-patch-copilot console")