test okta delegation

Author: rcholic

.env.example

@@ -0,0 +1,14 @@
+# AgentIdentity Okta compatibility checks (examples/tests)
+
+# Core Okta app settings used by OBO compatibility demo/test.
+OKTA_ISSUER=https://<your-okta-domain>/oauth2/default
+OKTA_CLIENT_ID=<your-okta-client-id>
+OKTA_CLIENT_SECRET=<your-okta-client-secret>
+OKTA_AUDIENCE=api://predicate-authority
+OKTA_SCOPE=authority:check
+
+# Enable live compatibility check test (disabled by default).
+OKTA_OBO_COMPAT_CHECK_ENABLED=0
+
+# Set to 1/true only if your Okta tenant supports token exchange/OBO.
+OKTA_SUPPORTS_TOKEN_EXCHANGE=0

docs/authorityd-operations.md

@@ -191,6 +191,45 @@ Checkpoints:
 - validation error includes a reason category (e.g. issuer mismatch),
 - error text does not include raw token string or sensitive claim values.
 
+4) Okta token exchange/OBO compatibility (tenant capability-gated):
+
+```bash
+# If tenant supports token exchange:
+export OKTA_OBO_COMPAT_CHECK_ENABLED=1
+export OKTA_SUPPORTS_TOKEN_EXCHANGE=true
+python3 -m pytest tests/test_okta_obo_compatibility.py -k "live_check_when_enabled"
+
+# If tenant does NOT support token exchange:
+export OKTA_OBO_COMPAT_CHECK_ENABLED=1
+export OKTA_SUPPORTS_TOKEN_EXCHANGE=false
+python3 -m pytest tests/test_okta_obo_compatibility.py -k "live_check_when_enabled"
+```
+
+Checkpoints:
+
+- `client_credentials_ok` must pass in both modes,
+- when `OKTA_SUPPORTS_TOKEN_EXCHANGE=true`, token exchange must succeed,
+- when `OKTA_SUPPORTS_TOKEN_EXCHANGE=false`, token exchange path is explicitly gated as tenant-disabled (no false failure).
+
+### Example demo script: Okta delegation compatibility
+
+Run example from repo root:
+
+```bash
+python3 examples/delegation/okta_obo_compat_demo.py \
+  --issuer "$OKTA_ISSUER" \
+  --client-id "$OKTA_CLIENT_ID" \
+  --client-secret "$OKTA_CLIENT_SECRET" \
+  --audience "$OKTA_AUDIENCE" \
+  --scope "${OKTA_SCOPE:-authority:check}" \
+  --supports-token-exchange
+```
+
+Notes:
+
+- omit `--supports-token-exchange` for tenants that do not support OBO/token exchange,
+- script reports whether delegation path should use IdP token exchange or authority mandate delegation.
+
 ### Secret storage policy (Okta credentials)
 
 - never commit Okta client secrets/API tokens/private keys to repo files,

docs/predicate-authority-user-manual.md

@@ -219,6 +219,65 @@ predicate-authorityd \
 
 ---
 
+## Okta delegation compatibility check (capability-gated)
+
+Use this when you want to verify whether your Okta tenant can do IdP token
+exchange/OBO for delegation, or if you should use authority mandate delegation
+as the fallback path.
+
+### 1) Set environment variables
+
+```bash
+cp .env.example .env
+
+export OKTA_ISSUER="https://<org>.okta.com/oauth2/default"
+export OKTA_CLIENT_ID="<okta-client-id>"
+export OKTA_CLIENT_SECRET="<okta-client-secret>"
+export OKTA_AUDIENCE="api://predicate-authority"
+export OKTA_SCOPE="authority:check"
+```
+
+### 2) Run compatibility test (live check is opt-in)
+
+```bash
+# Tenant supports token exchange/OBO
+export OKTA_OBO_COMPAT_CHECK_ENABLED=1
+export OKTA_SUPPORTS_TOKEN_EXCHANGE=true
+python -m pytest tests/test_okta_obo_compatibility.py -k "live_check_when_enabled"
+
+# Tenant does NOT support token exchange/OBO
+export OKTA_OBO_COMPAT_CHECK_ENABLED=1
+export OKTA_SUPPORTS_TOKEN_EXCHANGE=false
+python -m pytest tests/test_okta_obo_compatibility.py -k "live_check_when_enabled"
+```
+
+Expected behavior:
+
+- `client_credentials` path succeeds in both modes.
+- if `OKTA_SUPPORTS_TOKEN_EXCHANGE=true`, token exchange should succeed.
+- if `OKTA_SUPPORTS_TOKEN_EXCHANGE=false`, test is explicitly gated and does not
+  fail as a false negative.
+
+### 3) Run demo script in `examples/`
+
+```bash
+python examples/delegation/okta_obo_compat_demo.py \
+  --issuer "$OKTA_ISSUER" \
+  --client-id "$OKTA_CLIENT_ID" \
+  --client-secret "$OKTA_CLIENT_SECRET" \
+  --audience "$OKTA_AUDIENCE" \
+  --scope "${OKTA_SCOPE:-authority:check}" \
+  --supports-token-exchange
+```
+
+If your tenant does not support token exchange, omit
+`--supports-token-exchange`. The script reports which delegation path to use:
+
+- `idp_token_exchange` (when supported), or
+- `authority_mandate_delegation` (fallback).
+
+---
+
 ## Local identity registry + flush queue
 
 Enable ephemeral task identity registry and local ledger queue:

examples/delegation/okta_obo_compat_demo.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+
+def _ensure_repo_root_on_syspath() -> None:
+    repo_root = Path(__file__).resolve().parents[2]
+    root = str(repo_root)
+    if root not in sys.path:
+        sys.path.insert(0, root)
+
+
+def run(
+    issuer: str,
+    client_id: str,
+    client_secret: str,
+    audience: str,
+    scope: str,
+    supports_token_exchange: bool,
+    timeout_s: float,
+) -> dict[str, object]:
+    _ensure_repo_root_on_syspath()
+    from predicate_authority import (  # pylint: disable=import-error
+        OktaCompatibilityConfig,
+        OktaTenantCapabilities,
+        run_okta_obo_compatibility_check,
+    )
+
+    result = run_okta_obo_compatibility_check(
+        config=OktaCompatibilityConfig(
+            issuer=issuer,
+            client_id=client_id,
+            client_secret=client_secret,
+            audience=audience,
+            scope=scope,
+        ),
+        capabilities=OktaTenantCapabilities(supports_token_exchange=supports_token_exchange),
+        timeout_s=timeout_s,
+    )
+    result["delegation_path"] = (
+        "idp_token_exchange"
+        if bool(result.get("token_exchange_ok", False))
+        else "authority_mandate_delegation"
+    )
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Okta OBO compatibility demo for delegation flow.")
+    parser.add_argument("--issuer", default=os.getenv("OKTA_ISSUER"))
+    parser.add_argument("--client-id", default=os.getenv("OKTA_CLIENT_ID"))
+    parser.add_argument("--client-secret", default=os.getenv("OKTA_CLIENT_SECRET"))
+    parser.add_argument("--audience", default=os.getenv("OKTA_AUDIENCE"))
+    parser.add_argument("--scope", default=os.getenv("OKTA_SCOPE", "authority:check"))
+    parser.add_argument(
+        "--supports-token-exchange",
+        action="store_true",
+        help="Set if this Okta tenant is expected to support token exchange/OBO.",
+    )
+    parser.add_argument("--timeout-s", type=float, default=5.0)
+    args = parser.parse_args()
+
+    missing = [
+        name
+        for name, value in (
+            ("issuer", args.issuer),
+            ("client_id", args.client_id),
+            ("client_secret", args.client_secret),
+            ("audience", args.audience),
+        )
+        if value is None or str(value).strip() == ""
+    ]
+    if missing:
+        raise SystemExit(f"Missing required arguments/env vars: {', '.join(missing)}")
+
+    payload = run(
+        issuer=str(args.issuer),
+        client_id=str(args.client_id),
+        client_secret=str(args.client_secret),
+        audience=str(args.audience),
+        scope=str(args.scope),
+        supports_token_exchange=bool(args.supports_token_exchange),
+        timeout_s=float(args.timeout_s),
+    )
+    print(json.dumps(payload, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

predicate_authority/__init__.py

@@ -33,6 +33,13 @@
     TaskIdentityRecord,
 )
 from predicate_authority.mandate import LocalMandateSigner
+from predicate_authority.okta_compat import (
+    OktaCompatibilityConfig,
+    OktaCompatibilityError,
+    OktaTenantCapabilities,
+    parse_bool,
+    run_okta_obo_compatibility_check,
+)
 from predicate_authority.policy import PolicyEngine, PolicyMatchResult
 from predicate_authority.policy_source import PolicyFileSource, PolicyReloadResult
 from predicate_authority.proof import InMemoryProofLedger
@@ -79,6 +86,9 @@
     "OktaIdentityBridge",
     "OktaTokenClaims",
     "OpenTelemetryTraceEmitter",
+    "OktaCompatibilityConfig",
+    "OktaCompatibilityError",
+    "OktaTenantCapabilities",
     "PolicyEngine",
     "PolicyFileSource",
     "PolicyMatchResult",
@@ -94,4 +104,6 @@
     "TaskIdentityRecord",
     "TokenValidationError",
     "UsageCreditRecord",
+    "parse_bool",
+    "run_okta_obo_compatibility_check",
 ]