feat(identity): pluggable auth verifier seam for OAuth/OIDC portability

docs/delegated-execution-ledger.playbook.json

@@ -0,0 +1,106 @@
+[
+  {
+    "task_id": "DEL-M0-01-no-plaintext-secrets",
+    "skill": "general:sec",
+    "perk": "secrets",
+    "vars": {
+      "SEARCH_DIR": "/app"
+    },
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M0-02-schemas-conform",
+    "skill": "cws:cws-conform",
+    "perk": "schemas",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M0-03-chain-checkpoint-baseline",
+    "skill": "cws:cws-ledgercheck",
+    "perk": "checkpoint",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M1-01-keystore-conform",
+    "skill": "cws:cws-conform",
+    "perk": "keystore",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M1-02-erasure",
+    "skill": "cws:cws-ledgercheck",
+    "perk": "erasure",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M1-03-modelcheck-prove",
+    "skill": "cws:cws-modelcheck",
+    "perk": "prove",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-01-rt-grant-forged",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-forged",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-02-rt-grant-replay",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-replay",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-03-rt-forged-status",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-forged-status",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-04-rt-ledger-tamper",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-ledger-tamper",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-05-rt-tamper-script",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-tamper-script",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M2-06-rt-net-egress",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-net-egress",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M3-01-py-lint",
+    "skill": "general:py_qc",
+    "perk": "lint",
+    "vars": {
+      "PROJECT_DIR": "/app"
+    },
+    "redeem": false
+  },
+  {
+    "task_id": "DEL-M3-05-sec-audit",
+    "skill": "general:sec",
+    "perk": "audit",
+    "vars": {
+      "PROJECT_DIR": "/app"
+    },
+    "redeem": false
+  }
+]

docs/identity-and-approval.playbook.json

@@ -0,0 +1,114 @@
+[
+  {
+    "task_id": "IDN-M0-01-no-plaintext-secrets",
+    "skill": "general:sec",
+    "perk": "secrets",
+    "vars": { "SEARCH_DIR": "/app" },
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M0-02-keystore-conform",
+    "skill": "cws:cws-conform",
+    "perk": "keystore",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M0-03-chain-checkpoint",
+    "skill": "cws:cws-ledgercheck",
+    "perk": "checkpoint",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-01-rt-grant-forged",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-forged",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-02-rt-grant-replay",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-replay",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-03-rt-grant-expired",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-expired",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-04-rt-grant-wrong-run",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-grant-wrong-run",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-05-rt-no-capability",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-no-capability",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M1-06-rt-cap-mismatch",
+    "skill": "cws:cws-redteam",
+    "perk": "rt-cap-mismatch",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M2-01-sw-grant-forged",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-grant-forged",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M2-02-sw-grant-replay",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-grant-replay",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M2-03-sw-grant-expired",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-grant-expired",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M2-04-rt-ledger-tamper",
+    "skill": "cws:cws-redteam-sw",
+    "perk": "rt-ledger-tamper",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M3-01-erasure",
+    "skill": "cws:cws-ledgercheck",
+    "perk": "erasure",
+    "vars": {},
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M3-02-py-lint",
+    "skill": "general:py_qc",
+    "perk": "lint",
+    "vars": { "PROJECT_DIR": "/app" },
+    "redeem": false
+  },
+  {
+    "task_id": "IDN-M3-03-sec-audit",
+    "skill": "general:sec",
+    "perk": "audit",
+    "vars": { "PROJECT_DIR": "/app" },
+    "redeem": false
+  }
+]

infra/govern/fleetd.py

@@ -298,7 +298,10 @@ def _auth(self):
         reg = self.server.cfg.get("principals") or {}
         if not reg:
             return "local"
-        return principals.authenticate(principals.bearer_of(self.headers.get("Authorization", "")), reg)
+        # Same identity seam as govd (see principals.resolve_principal): default is the bearer-secret match.
+        cfg = getattr(self.server, "cfg", None) or {}
+        return principals.resolve_principal(
+            principals.bearer_of(self.headers.get("Authorization", "")), reg, cfg.get("auth_verifier", ""))
 
     def _rate_ok(self, pid) -> bool:
         reg = self.server.cfg.get("principals") or {}

infra/govern/govd.py

@@ -108,6 +108,10 @@ def load_config(path=None):
     # the monitor (dashboard) token — gates the dashboard. env > config; the default is filled by
     # ensure_monitor_token() once the FINAL mode is known (after any --mode override).
     cfg["monitor_token"] = os.environ.get("GOVD_MONITOR_TOKEN") or cfg.get("monitor_token") or None
+    # Identity SCHEME for Authorization: Bearer. "" / "token_sha" = the built-in bearer-secret match (default,
+    # unchanged). Any other value names a verifier registered via principals.register_verifier — the OAuth/OIDC
+    # seam. An unknown name fails CLOSED (no principal resolves), never falls back to the secret path.
+    cfg["auth_verifier"] = os.environ.get("GOVD_AUTH_VERIFIER") or cfg.get("auth_verifier") or ""
     # P1-T08: the principals registry (id -> token_sha -> quota). A present registry makes Authorization:
     # Bearer mandatory at /govern; absent (local dev) -> auth off, every record carries principal "local".
     pr_path = os.environ.get("GOVD_PRINCIPALS") or cfg.get("principals_path")
@@ -1164,7 +1168,8 @@ def do_POST(self):
             return self._json(503, {"error": "acl_strict requires a configured principals registry"})
         pid = cfg.get("node_name") or "local"            # unauthenticated local-mode runs attribute to the
         if reg:                                          # node's fleet name (GOVD_NODE_NAME), else generic "local"
-            pid = principals.authenticate(principals.bearer_of(self.headers.get("Authorization", "")), reg)
+            pid = principals.resolve_principal(
+                principals.bearer_of(self.headers.get("Authorization", "")), reg, cfg.get("auth_verifier", ""))
             if pid is None:
                 return self._json(401, {"error": "missing/invalid Authorization: Bearer token"})
             spec = reg[pid]

infra/govern/principals.py

@@ -39,6 +39,68 @@ def authenticate(bearer: str, principals: dict):
     return None
 
 
+# ───────────────────────── the auth SEAM (OAuth/OIDC portability) ─────────────────────────
+#
+# `authenticate` answers ONE question — "which principal is this bearer?" — by hashing the token and matching
+# `token_sha`. That is a *bearer-secret* scheme: govd must have seen the secret to know the sha. An external
+# IdP (OAuth/OIDC) answers the same question differently — verify a signature, read a `sub`/`email` claim —
+# and govd never holds a secret at all.
+#
+# Both are the same shape: bearer -> SUBJECT -> principal_id. So the seam is a pluggable *subject resolver*,
+# and everything downstream (`acl_allows`, rate buckets, the provenance `principal`) is untouched, because it
+# already keys off a principal id and never off a token.
+#
+# A principal opts into an external scheme by declaring `subject`, e.g.
+#     {"alice": {"subject": "oidc:sub:1a2b3c", "acl": {...}}}
+# Registering a verifier is the whole integration:
+#     register_verifier("oidc", lambda bearer: verify_jwt_and_return("oidc:sub:" + claims["sub"]))
+#
+# DELIBERATELY NOT DONE HERE: no JWKS fetch, no JWT parsing, no network. A verifier that reaches the network
+# inside the syscall boundary is a new failure+latency mode for every claim, and belongs behind an explicit
+# cache the operator configures. This module only provides the seam and keeps the default behaviour identical.
+
+_VERIFIERS: dict = {}
+
+
+def register_verifier(name: str, fn) -> None:
+    """Register a subject resolver: ``fn(bearer) -> subject_str | None``. It MUST fail closed (return None)
+    on any doubt, and MUST NOT raise — `resolve_principal` treats a raising verifier as a refusal."""
+    _VERIFIERS[str(name)] = fn
+
+
+def subject_of(bearer: str, verifier: str):
+    """The external SUBJECT this bearer asserts, or None. `verifier` names a registered scheme; an unknown
+    scheme fails closed rather than silently falling back to the bearer-secret path."""
+    fn = _VERIFIERS.get(str(verifier or ""))
+    if fn is None or not bearer:
+        return None
+    try:
+        s = fn(bearer)
+    except Exception:
+        return None                                      # a raising verifier is a refusal, never an allow
+    return s or None
+
+
+def resolve_principal(bearer: str, principals: dict, verifier: str = ""):
+    """The principal id for this bearer under `verifier`, else None. THE single identity entry point.
+
+    verifier falsy / "token_sha"  -> the built-in bearer-secret scheme (unchanged default)
+    any registered name           -> subject match against each principal's declared `subject`
+
+    Subject comparison is constant-time and requires a NON-EMPTY declared subject, so a principal that has not
+    opted in can never be matched by an empty/absent claim."""
+    if not verifier or verifier == "token_sha":
+        return authenticate(bearer, principals)
+    subj = subject_of(bearer, verifier)
+    if not subj:
+        return None
+    for pid, spec in principals.items():
+        want = str((spec or {}).get("subject") or "")
+        if want and hmac.compare_digest(want, str(subj)):
+            return pid
+    return None
+
+
 def bearer_of(authorization: str) -> str:
     """Extract the token from an `Authorization: Bearer <token>` header (query tokens are NOT accepted)."""
     if not authorization:

tests/test_principals.py

@@ -42,3 +42,52 @@ def test_load_principals_absent_is_empty(tmp_path):
 
 def test_selftest_ok():
     assert P.principals_selftest()["ok"] is True
+
+
+# ───────────────────────── the auth SEAM (OAuth/OIDC portability) ─────────────────────────
+# Every branch here is an IDENTITY decision, so each asserts a fail-CLOSED outcome explicitly.
+# The default path must stay byte-identical in behaviour: an existing deployment names no verifier.
+
+def _seam_reg():
+    return {"alice": {"subject": "oidc:sub:1a2b3c"},
+            "bot": {"token_sha": P.token_sha("s3cret")},
+            "nosub": {"subject": "", "token_sha": P.token_sha("other")}}
+
+
+def test_seam_default_is_unchanged_bearer_secret():
+    reg = _seam_reg()
+    assert P.resolve_principal("s3cret", reg) == "bot"          # no verifier named
+    assert P.resolve_principal("s3cret", reg, "") == "bot"
+    assert P.resolve_principal("s3cret", reg, "token_sha") == "bot"
+    assert P.resolve_principal("wrong", reg) is None
+
+
+def test_seam_unknown_verifier_fails_closed():
+    # An unknown scheme must NOT silently fall back to the secret path — that would let a typo in
+    # auth_verifier quietly re-enable bearer secrets on a deployment that meant to disable them.
+    assert P.resolve_principal("s3cret", _seam_reg(), "not-registered") is None
+
+
+def test_seam_registered_verifier_maps_subject_to_principal():
+    P.register_verifier("t_oidc", lambda b: "oidc:sub:1a2b3c" if b == "good" else None)
+    reg = _seam_reg()
+    assert P.resolve_principal("good", reg, "t_oidc") == "alice"
+    assert P.resolve_principal("bad", reg, "t_oidc") is None
+    # a bearer SECRET must not authenticate under an external scheme
+    assert P.resolve_principal("s3cret", reg, "t_oidc") is None
+
+
+def test_seam_raising_verifier_is_a_refusal():
+    def boom(_b):
+        raise RuntimeError("jwks unreachable")
+    P.register_verifier("t_boom", boom)
+    assert P.resolve_principal("anything", _seam_reg(), "t_boom") is None
+
+
+def test_seam_empty_subject_never_matches():
+    # A principal that has not opted in (no/blank `subject`) must be unreachable via the external scheme,
+    # even when the verifier returns an empty claim.
+    P.register_verifier("t_empty", lambda _b: "")
+    assert P.resolve_principal("x", _seam_reg(), "t_empty") is None
+    P.register_verifier("t_blank", lambda _b: None)
+    assert P.resolve_principal("x", _seam_reg(), "t_blank") is None