diff --git a/Makefile b/Makefile index 3aadf4c..6ecc376 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ # SourceOS Continuum — lifecycle entry points. # Control-plane targets delegate to Makefile.porter (the rehomed Porter control plane). -.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop verify lease sphere push push-webhook rollback edge login provision deploy inference availability +.PHONY: validate onboard dev-up dev-down shim-test test tools-test rollout promotion-gate portal compute mesh-demo grant commons mcp spine run loop verify lease sphere push push-webhook rollback edge login sso provision deploy inference availability validate: ## repo hygiene + CapD validity python3 tools/validate.py @@ -60,6 +60,9 @@ edge: ## edge-worker demo: register the agent-machine into a cloud pool + evolve login: ## login/session demo: authenticate the front door (fail-closed) cd tools && python3 login.py +sso: ## SSO front door demo: OIDC + PKCE + WebAuthn/FIDO2 verification (fail-closed, stdlib-only) + cd tools && python3 sso.py + provision: ## provisioning demo: tenant broker + entitlement tier + BSS meter + /me cd tools && python3 provisioning.py diff --git a/capd/sso-front-door.mesh.capd.json b/capd/sso-front-door.mesh.capd.json new file mode 100644 index 0000000..848f3fa --- /dev/null +++ b/capd/sso-front-door.mesh.capd.json @@ -0,0 +1,31 @@ +{ + "capability_id": "caps.dev.sso-front-door@0.1.0", + "kind": "dev.identity-sso", + "status": "experimental", + "name": "SSO front door — real OIDC + WebAuthn/FIDO2 verification, fail-closed, stdlib-only", + "description": "The 'who is this' layer above the login.py session core: the two real SSO mechanisms the cloud-shell-fog spec names, OIDC and WebAuthn/FIDO2, verified honestly with only the standard library. OIDC ID-token validation is strict and fail-closed (exact iss; aud/azp; exp/nbf/iat; nonce binding; alg 'none' always refused; HS256 verified via hmac). PKCE (S256) binds the authorization code to the client, with state (CSRF) and nonce (replay) checks. WebAuthn assertion verification checks ceremony type, challenge, origin, rpIdHash, user-present/user-verified flags, and signature-counter clone detection. Honest boundary: verifying an ASYMMETRIC signature (RS256/ES256 JWTs, WebAuthn credential keys) needs crypto the stdlib does not ship and must never be hand-rolled, so that one step is a clearly-marked injection point (verify_signature=…) wired in production to a JWKS/cryptography verifier — asymmetric algs are refused, never silently trusted, without it. A verified identity mints a login.py session; the same zero-trust, fail-closed posture the webhook gives machines, now for humans.", + "links": { + "engine": "tools/sso.py", + "session_core": "tools/login.py", + "machine_analogue": "tools/push_webhook.py (HMAC-verified machine door)", + "grant_binding": "tools/mcp_a2a_grant.py", + "spec_witness": "docs/SSO_FRONT_DOOR.md", + "reference_pattern": "OpenID Connect (auth-code + PKCE, ID-token claim validation) + W3C WebAuthn/FIDO2 (assertion verification) — met sovereign+stdlib: real claim/ceremony validation, alg:none refused, asymmetric signature verification as an injected JWKS/crypto step, never hand-rolled" + }, + "composes_with": { + "session": "caps.dev.login-session@0.1.0", + "provisioning": "caps.dev.provisioning@0.1.0", + "control_plane": "caps.infra.paas.continuum-local@0.1.0", + "scales_up_to": "caps.infra.cluster-scaleup.hyperswarm@0.1.0" + }, + "policy": { + "availability": "needs-work", + "fail_closed": true, + "alg_none_refused": true, + "pkce_required": true, + "nonce_replay_bound": true, + "webauthn_clone_detection": true, + "asymmetric_verify_injected_not_hand_rolled": true, + "evidence_emitting": false + } +} diff --git a/docs/PAAS_GAP_REGISTER.md b/docs/PAAS_GAP_REGISTER.md index 7a6b0f7..ca5bd90 100644 --- a/docs/PAAS_GAP_REGISTER.md +++ b/docs/PAAS_GAP_REGISTER.md @@ -24,7 +24,7 @@ diagrams feel seamless is mostly missing — that's what this register makes exp | **Nocalhost-Web** (admin web console) | portal (read-only dev console) | ◑ (no admin console) | | **Nocalhost-API** (REST) | MCP ops surface (agent JSON-RPC) | ◑ (no REST admin API) | | **Nocalhost-Dep** (cluster-side dep, `nocalhost-reserved` ns) | — | ○ **GAP: no cluster-side controller/operator** | -| **Login** / auth | — | ○ **GAP: no login/session/SSO** | +| **Login** / auth | `login.py` (session core) + `sso.py` (real OIDC + WebAuthn/FIDO2 verify) | ◑ (verify core done; asymmetric-sig verifier injected, not yet wired to a live IdP) | | Admin: **Create User** | — | ○ **GAP** (this change closes it) | | Admin: **Configure Cluster** | CapDs (static) | ◑ | | Admin: **Configure Application** | buildpack + workload specs | ◑ | @@ -35,7 +35,7 @@ diagrams feel seamless is mostly missing — that's what this register makes exp |---|---|---| | **Load Balancer** | — | ○ GAP (no ingress LB in continuum) | | DataWorks **UI App Server** | portal | ◑ | -| **BlueMix SSO service** | — | ○ **GAP: no SSO** | +| **BlueMix SSO service** | `sso.py` — OIDC (auth-code + PKCE, ID-token validation) + WebAuthn/FIDO2 | ◑ (fail-closed verify core, stdlib; asymmetric-signature step injected) | | **BlueMix Entitlement service** | admission **tiers** | ◑ (tier data, no entitlement *service*/token) | | **Watson Token Server** (per-tenant) + **/me API** | grants (session tokens) | ◑ (no token server / user-profile `/me`) — this change adds `/me` | | **Service Broker** (add tenant / provision new VM) | `devspace` manifests | ○ **GAP** (this change closes it) | @@ -68,9 +68,17 @@ BlueMix calls it SSO+Entitlement+Broker+Token+BSS. It's the same server. `tools/ (this change) builds the governed core of it — provision/bind/deprovision a tenant (DevSpace + tier + session), meter usage, and `/me` — and the portal exposes `/api/me` + `/api/provision`. -**Still open after this change** (the honest ranked backlog): IDE plugin · running remote terminal · -real login/SSO · cluster-side dep operator · load-balancer + RM-A/B failover HA · ELK/Grafana -dashboards · landing page. Named here so nothing hides. +**Shipped since (front door + deploy).** The developer/human front door is now real and governed: +`login.py` session core + **`sso.py`** (OIDC auth-code+PKCE + WebAuthn/FIDO2, fail-closed, stdlib) → +the git-push-to-deploy ergonomic is complete end-to-end: **`push_webhook.py`** (HMAC-verified trigger) +→ `deploy_flow.py` preview → promotion gate → **`release_ledger.py`** (instant, reproducible, +fail-closed rollback). The machine door (webhook, HMAC) and the human door (SSO, OIDC/WebAuthn) share +one posture: a zero-trust gate, not an open hook. + +**Still open** (the honest ranked backlog): IDE plugin · running remote terminal · **live IdP wiring** +for `sso.py`'s injected asymmetric-signature verifier (JWKS/`cryptography`) · cluster-side dep +operator · load-balancer + RM-A/B failover HA · ELK/Grafana dashboards · landing page. Named here so +nothing hides. ## Cloud Foundry / BlueMix — similar, different, and why ours is best-of-all diff --git a/docs/SSO_FRONT_DOOR.md b/docs/SSO_FRONT_DOOR.md new file mode 100644 index 0000000..032977a --- /dev/null +++ b/docs/SSO_FRONT_DOOR.md @@ -0,0 +1,54 @@ +# SSO front door — real OIDC + WebAuthn/FIDO2, fail-closed, stdlib-only + +`login.py` is the **session core** — it issues and verifies an HMAC-bound, expiring session. But a +session should only be minted once you actually know *who is this*. That decision is SSO, and +`tools/sso.py` does it with the two real mechanisms the cloud-shell-fog spec names — **OIDC** and +**WebAuthn/FIDO2** — using only the standard library. + +It is the human analogue of the push webhook: the webhook is the **machine** door (an HMAC-verified +`git push`); this is the **human** door (an OIDC/WebAuthn-verified login). Same posture — a +zero-trust, fail-closed gate, not an open one. + +## The honest boundary + +Verifying an **asymmetric** signature — RS256/ES256 ID tokens, WebAuthn credential public keys — +needs public-key crypto the Python stdlib does not ship, and hand-rolling RSA/ECDSA is exactly the +thing you must never do. So that one step is a **clearly-marked injection point** +(`verify_signature=…`), wired in production to a JWKS / `cryptography` verifier. Without it, +asymmetric algorithms are **refused, never silently trusted**. + +Everything *else* — and it is the majority of what these protocols get wrong in the field — is real +here and stdlib-doable. That's the point: we build the security-critical validation honestly and +delegate exactly one crypto primitive, rather than fake the whole thing or pull a heavy dep into a +dependency-free stack. + +## What's verified (all fail-closed) + +**OIDC ID-token** (`verify_id_token`) — signature first, then every required claim: +- `alg: none` (or missing) is **always refused** — an unsigned token is never trusted. +- `HS256` verified with `hmac` (stdlib); asymmetric algs require the injected verifier. +- `iss` exact match · `aud` contains us (and `azp` is us when there are multiple audiences) · + `exp`/`nbf`/`iat` with small leeway · `nonce` binds the token to *this* login (replay/CSRF). + +**PKCE + auth-code** (`pkce_pair` / `build_authorization_request` / `verify_callback` / +`build_token_exchange`) — S256 challenge/verifier binding, `state` (CSRF) and `nonce` (replay) checks, +and the token-exchange body that sends the `code_verifier` as proof. All stdlib, no network I/O. + +**WebAuthn assertion** (`verify_assertion`) — per WebAuthn §7.2: +- ceremony `type == webauthn.get` · `challenge` equals the one we issued (constant-time) · + `origin` exact (anti-phishing) · `rpIdHash == sha256(rp_id)` · **UP** (user-present) required, + **UV** enforced when asked · **signature-counter clone detection** (a stored/reported 0 disables it, + else it must strictly increase). The credential signature over + `authenticatorData ‖ sha256(clientDataJSON)` is the injected step. + +A valid verification → `session_from_identity()` mints a `login.py` session. An invalid one mints +nothing. + +## Try it + +```bash +make sso # PKCE ok/tampered · OIDC valid/alg-none-refused/bad-nonce-refused · WebAuthn valid/clone-refused · session +``` + +`sso.py` is a pure verification core (25 unit tests, every rejection path covered). Production wires +`verify_signature` to a JWKS fetcher for the real IdP; the claim/ceremony logic does not change. diff --git a/tools/sso.py b/tools/sso.py new file mode 100644 index 0000000..f39e5e2 --- /dev/null +++ b/tools/sso.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 +"""SSO front door — real OIDC + WebAuthn/FIDO2 verification, fail-closed, stdlib-only. + +`login.py` is the session core (issue/verify an HMAC-bound session). This is the thing that decides +*who is this* before a session is minted: the two real SSO mechanisms the cloud-shell-fog spec names, +OIDC and WebAuthn/FIDO2 — done honestly with only the standard library. + +An honest boundary, stated up front: verifying an **asymmetric** signature (RS256/ES256 JWTs, WebAuthn +credential public keys) needs asymmetric crypto the stdlib does not ship, and hand-rolling RSA/ECDSA +is exactly what you must not do. So the asymmetric signature check is a single, clearly-marked +**injection point** (`verify_signature=...`), wired in production to a JWKS/`cryptography` verifier. +Everything *else* — and it is the majority of what these protocols get wrong in the wild — is real +here and stdlib-doable: + + * **OIDC ID-token**: strict claim validation (iss, aud/azp, exp/iat/nbf, nonce), `alg:none` refused, + HS256 verified with hmac; asymmetric algs require the injected verifier (never silently accepted). + * **PKCE (S256)**: challenge/verifier binding + state (CSRF) + nonce (replay) — all stdlib. + * **WebAuthn assertion**: type, challenge, origin, rpIdHash, user-present/verified flags, and + signature-counter clone detection — all stdlib; the credential signature is the injected step. + +Every check is fail-closed: any failure returns `{"valid": False, "reason": ...}` and no identity is +asserted. Verified identity → `session_from_identity()` mints a `login.py` session. +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import secrets +from datetime import datetime, timezone +from urllib.parse import urlencode + + +# --- base64url + hashing helpers ----------------------------------------------------------------- + +def _b64url_decode(s) -> bytes: + if isinstance(s, str): + s = s.encode("ascii") + return base64.urlsafe_b64decode(s + b"=" * (-len(s) % 4)) + + +def _b64url_encode(b: bytes) -> str: + return base64.urlsafe_b64encode(b).rstrip(b"=").decode("ascii") + + +def _sha256(b: bytes) -> bytes: + return hashlib.sha256(b).digest() + + +def _fail(reason: str) -> dict: + return {"valid": False, "reason": reason} + + +# --- PKCE + authorization-code flow (all stdlib) ------------------------------------------------- + +def pkce_pair(code_verifier: str | None = None) -> dict: + """A PKCE S256 pair. `code_verifier` is a high-entropy secret the client keeps; `code_challenge` + is base64url(sha256(verifier)) sent on the authorize request. Proves the client that started the + flow is the one redeeming the code (defeats code interception).""" + code_verifier = code_verifier or _b64url_encode(secrets.token_bytes(48)) + return {"code_verifier": code_verifier, + "code_challenge": _b64url_encode(_sha256(code_verifier.encode("ascii"))), + "code_challenge_method": "S256"} + + +def verify_pkce(*, code_verifier: str, code_challenge: str, method: str = "S256") -> bool: + """Recompute the challenge from the verifier and constant-time compare (fail-closed).""" + if method == "S256": + expected = _b64url_encode(_sha256(code_verifier.encode("ascii"))) + elif method == "plain": + expected = code_verifier + else: + return False + return hmac.compare_digest(expected, code_challenge) + + +def build_authorization_request(*, authorize_url: str, client_id: str, redirect_uri: str, + scope: str = "openid profile", state: str | None = None, + nonce: str | None = None, code_challenge: str) -> dict: + """Build the /authorize redirect (with state + nonce + PKCE challenge). Keep `state`/`nonce` to + check on the callback.""" + state = state or _b64url_encode(secrets.token_bytes(16)) + nonce = nonce or _b64url_encode(secrets.token_bytes(16)) + params = {"response_type": "code", "client_id": client_id, "redirect_uri": redirect_uri, + "scope": scope, "state": state, "nonce": nonce, + "code_challenge": code_challenge, "code_challenge_method": "S256"} + return {"url": authorize_url + "?" + urlencode(params), "state": state, "nonce": nonce} + + +def verify_callback(*, returned_state: str, expected_state: str, error: str | None = None) -> dict: + """Fail-closed callback check: an IdP-reported error, or a state mismatch (CSRF), aborts.""" + if error: + return _fail(f"authorization error from IdP: {error}") + if not returned_state or not hmac.compare_digest(str(returned_state), str(expected_state)): + return _fail("state mismatch — possible CSRF; authorization rejected") + return {"valid": True} + + +def build_token_exchange(*, token_url: str, code: str, redirect_uri: str, client_id: str, + code_verifier: str, client_secret: str | None = None) -> dict: + """The token-endpoint POST body redeeming the code (sends `code_verifier` — the PKCE proof). No + network I/O here; the caller POSTs `body` to `token_url` over TLS.""" + body = {"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, + "client_id": client_id, "code_verifier": code_verifier} + if client_secret is not None: + body["client_secret"] = client_secret + return {"token_url": token_url, "body": body} + + +# --- OIDC ID-token verification ------------------------------------------------------------------ + +def encode_jwt_hs256(payload: dict, secret: str, *, header: dict | None = None) -> str: + """Mint an HS256 JWT (for symmetric clients + tests). Real IdPs usually sign RS256.""" + h = {"alg": "HS256", "typ": "JWT", **(header or {})} + seg = _b64url_encode(json.dumps(h, separators=(",", ":")).encode()) + "." + \ + _b64url_encode(json.dumps(payload, separators=(",", ":")).encode()) + sig = hmac.new(secret.encode(), seg.encode("ascii"), hashlib.sha256).digest() + return seg + "." + _b64url_encode(sig) + + +def decode_jwt(token: str) -> dict: + """Split a compact JWT into header/payload/signature without verifying. Raises ValueError if + malformed. `signing_input` is the exact bytes the signature covers (never re-serialized).""" + parts = token.split(".") + if len(parts) != 3: + raise ValueError("not a compact JWS (need 3 dot-separated segments)") + header = json.loads(_b64url_decode(parts[0])) + payload = json.loads(_b64url_decode(parts[1])) + if not isinstance(header, dict) or not isinstance(payload, dict): + raise ValueError("JWT header/payload are not JSON objects") + return {"header": header, "payload": payload, + "signing_input": (parts[0] + "." + parts[1]).encode("ascii"), + "signature": _b64url_decode(parts[2])} + + +def verify_id_token(token: str, *, issuer: str, audience: str, nonce: str | None = None, + now: datetime | None = None, hs256_secret: str | None = None, + verify_signature=None, leeway_s: int = 60) -> dict: + """Verify an OIDC ID token, fail-closed. Signature FIRST, then every required claim. + + hs256_secret — verify HS256 tokens with this shared secret (hmac, stdlib). + verify_signature — (signing_input: bytes, signature: bytes, header: dict) -> bool, for asymmetric + algs (RS256/ES256/…). Required for those; without it they are rejected, never + silently trusted. This is the one crypto step delegated out of stdlib. + Returns {valid, sub, claims} or {valid, reason}. + """ + now = now or datetime.now(timezone.utc) + try: + jwt = decode_jwt(token) + except (ValueError, json.JSONDecodeError, base64.binascii.Error) as exc: + return _fail(f"malformed ID token: {exc}") + + alg = jwt["header"].get("alg") + # 1. signature — the load-bearing check. `none` is always refused. + if alg == "none" or not alg: + return _fail("alg 'none' (or missing) is refused — an unsigned ID token is never trusted") + if alg == "HS256": + if not hs256_secret: + return _fail("HS256 token but no shared secret configured") + expected = hmac.new(hs256_secret.encode(), jwt["signing_input"], hashlib.sha256).digest() + if not hmac.compare_digest(expected, jwt["signature"]): + return _fail("HS256 signature invalid") + else: + if verify_signature is None: + return _fail(f"alg {alg} needs an asymmetric verifier (inject verify_signature=…); refused") + if not verify_signature(jwt["signing_input"], jwt["signature"], jwt["header"]): + return _fail(f"{alg} signature invalid") + + c = jwt["payload"] + # 2. issuer — exact match. + if c.get("iss") != issuer: + return _fail(f"iss mismatch (want {issuer!r})") + # 3. audience — client_id must be in aud; with multiple audiences, azp must be us. + aud = c.get("aud") + auds = aud if isinstance(aud, list) else [aud] + if audience not in auds: + return _fail("aud does not include this client") + if len(auds) > 1 and c.get("azp") != audience: + return _fail("multiple audiences but azp is not this client") + # 4. expiry / not-before / issued-at. + ts = now.timestamp() + if "exp" not in c or ts > float(c["exp"]) + leeway_s: + return _fail("token expired") + if "nbf" in c and ts + leeway_s < float(c["nbf"]): + return _fail("token not yet valid (nbf)") + if "iat" not in c: + return _fail("missing iat") + # 5. nonce — binds the token to this login attempt (replay/CSRF). + if nonce is not None and not hmac.compare_digest(str(c.get("nonce", "")), str(nonce)): + return _fail("nonce mismatch — token not bound to this login") + return {"valid": True, "sub": c.get("sub"), "claims": c} + + +# --- WebAuthn / FIDO2 assertion verification ----------------------------------------------------- + +def verify_assertion(*, client_data_json: bytes, authenticator_data: bytes, signature: bytes, + expected_challenge: bytes, expected_origin: str, rp_id: str, + prev_sign_count: int = 0, require_user_verification: bool = False, + verify_signature) -> dict: + """Verify a WebAuthn assertion (navigator.credentials.get), fail-closed, per WebAuthn §7.2. Every + non-crypto check is here; the credential signature is the injected `verify_signature(signed_data, + signature) -> bool` (ES256/RS256/EdDSA over the public key — the one out-of-stdlib step). + + Returns {valid, sign_count} or {valid, reason}. + """ + try: + cd = json.loads(client_data_json.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + return _fail(f"clientDataJSON not valid JSON: {exc}") + if not isinstance(cd, dict): + return _fail("clientDataJSON is not an object") + + # 1. ceremony type. + if cd.get("type") != "webauthn.get": + return _fail("clientData.type is not webauthn.get") + # 2. challenge — must equal the one we issued (defeats replay). clientData carries it base64url. + try: + got_challenge = _b64url_decode(cd.get("challenge", "")) + except base64.binascii.Error: + return _fail("clientData.challenge is not valid base64url") + if not hmac.compare_digest(got_challenge, expected_challenge): + return _fail("challenge mismatch — assertion not for this login") + # 3. origin — must be exactly the RP's origin (defeats cross-origin/phishing). + if cd.get("origin") != expected_origin: + return _fail(f"origin mismatch (want {expected_origin!r})") + + # 4. authenticatorData: rpIdHash + flags + signCount. + if len(authenticator_data) < 37: + return _fail("authenticatorData too short") + if not hmac.compare_digest(authenticator_data[:32], _sha256(rp_id.encode())): + return _fail("rpIdHash mismatch — assertion is for a different Relying Party") + flags = authenticator_data[32] + if not (flags & 0x01): # UP (user present) + return _fail("user-present (UP) flag not set") + if require_user_verification and not (flags & 0x04): # UV (user verified) + return _fail("user-verification required but UV flag not set") + sign_count = int.from_bytes(authenticator_data[33:37], "big") + # 5. signature-counter clone detection: a stored or reported counter of 0 disables the check + # (some authenticators don't implement it); otherwise it MUST strictly increase. + if (sign_count != 0 or prev_sign_count != 0) and sign_count <= prev_sign_count: + return _fail(f"signature counter did not increase ({sign_count} <= {prev_sign_count}) — " + "possible cloned authenticator") + # 6. the credential signature over authenticatorData || sha256(clientDataJSON) — injected crypto. + signed_data = authenticator_data + _sha256(client_data_json) + if not verify_signature(signed_data, signature): + return _fail("assertion signature invalid") + return {"valid": True, "sign_count": sign_count} + + +# --- bridge to the session core ------------------------------------------------------------------ + +def session_from_identity(verify_result: dict, *, key: bytes, tier: str = "pro", + ttl_s: float = 3600.0, now: datetime | None = None, + user: str | None = None) -> dict | None: + """Turn a VALID OIDC/WebAuthn verification into a login.py session. Fail-closed: an invalid + verification never yields a session.""" + if not verify_result.get("valid"): + return None + import login + subject = user or verify_result.get("sub") or (verify_result.get("claims") or {}).get("sub") + if not subject: + return None + return login.issue_session(user=subject, tier=tier, key=key, ttl_s=ttl_s, now=now) + + +if __name__ == "__main__": + import login + + key = b"sso-demo-session-key" + now = datetime.now(timezone.utc) + + # PKCE + p = pkce_pair() + pkce_ok = verify_pkce(code_verifier=p["code_verifier"], code_challenge=p["code_challenge"]) + pkce_bad = verify_pkce(code_verifier="tampered", code_challenge=p["code_challenge"]) + + # OIDC (HS256 for the demo; real IdPs sign RS256 → inject verify_signature) + secret = "oidc-client-secret" + claims = {"iss": "https://idp.sovereign", "aud": "continuum", "sub": "alice", + "nonce": "n-123", "iat": int(now.timestamp()), "exp": int(now.timestamp()) + 300} + tok = encode_jwt_hs256(claims, secret) + oidc_ok = verify_id_token(tok, issuer="https://idp.sovereign", audience="continuum", + nonce="n-123", now=now, hs256_secret=secret) + unsigned = encode_jwt_hs256({**claims}, secret).rsplit(".", 1)[0] + "." # strip sig + none_tok = _b64url_encode(json.dumps({"alg": "none"}).encode()) + "." + \ + _b64url_encode(json.dumps(claims).encode()) + "." + oidc_none = verify_id_token(none_tok, issuer="https://idp.sovereign", audience="continuum", now=now) + oidc_badnonce = verify_id_token(tok, issuer="https://idp.sovereign", audience="continuum", + nonce="wrong", now=now, hs256_secret=secret) + + # WebAuthn (inject the credential-signature step) + rp_id, origin, chal = "continuum.sovereign", "https://continuum.sovereign", secrets.token_bytes(32) + client_data = json.dumps({"type": "webauthn.get", "challenge": _b64url_encode(chal), + "origin": origin}).encode() + auth_data = _sha256(rp_id.encode()) + bytes([0x05]) + (7).to_bytes(4, "big") # UP+UV, counter 7 + wa_ok = verify_assertion(client_data_json=client_data, authenticator_data=auth_data, + signature=b"sig", expected_challenge=chal, expected_origin=origin, + rp_id=rp_id, prev_sign_count=6, verify_signature=lambda d, s: True) + wa_clone = verify_assertion(client_data_json=client_data, authenticator_data=auth_data, + signature=b"sig", expected_challenge=chal, expected_origin=origin, + rp_id=rp_id, prev_sign_count=7, verify_signature=lambda d, s: True) + + sess = session_from_identity(oidc_ok, key=key, tier="pro", now=now) + print(json.dumps({ + "pkce": {"ok": pkce_ok, "tampered_rejected": not pkce_bad}, + "oidc": {"valid": oidc_ok["valid"], "sub": oidc_ok.get("sub"), + "alg_none_refused": not oidc_none["valid"], + "bad_nonce_refused": not oidc_badnonce["valid"]}, + "webauthn": {"valid": wa_ok["valid"], "clone_refused": not wa_clone["valid"], + "clone_reason": wa_clone.get("reason", "")[:40]}, + "session_after_oidc": login.verify_session(sess, key=key, now=now)["valid"], + }, indent=2)) diff --git a/tools/test_sso.py b/tools/test_sso.py new file mode 100644 index 0000000..9187159 --- /dev/null +++ b/tools/test_sso.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +"""Tests for the SSO front door — OIDC + PKCE + WebAuthn, every check fail-closed.""" +import json +from datetime import datetime, timezone + +import login +import sso + +_NOW = datetime(2026, 8, 4, 12, 0, 0, tzinfo=timezone.utc) +_ISS = "https://idp.sovereign" +_AUD = "continuum" +_SECRET = "oidc-client-secret" + + +def _claims(**over): + base = {"iss": _ISS, "aud": _AUD, "sub": "alice", "nonce": "n-1", + "iat": int(_NOW.timestamp()), "exp": int(_NOW.timestamp()) + 300} + base.update(over) + return base + + +def _mint(alg, claims, sig=b"x"): + h = sso._b64url_encode(json.dumps({"alg": alg, "typ": "JWT"}).encode()) + p = sso._b64url_encode(json.dumps(claims).encode()) + return h + "." + p + "." + sso._b64url_encode(sig) + + +# --- PKCE --------------------------------------------------------------------------------------- + +def test_pkce_pair_verifies_and_tampered_verifier_fails(): + p = sso.pkce_pair() + assert p["code_challenge_method"] == "S256" + assert sso.verify_pkce(code_verifier=p["code_verifier"], code_challenge=p["code_challenge"]) + assert not sso.verify_pkce(code_verifier="tampered", code_challenge=p["code_challenge"]) + + +def test_pkce_unknown_method_is_rejected(): + assert sso.verify_pkce(code_verifier="v", code_challenge="v", method="MD5") is False + + +def test_authorization_request_carries_state_nonce_and_challenge(): + p = sso.pkce_pair() + req = sso.build_authorization_request(authorize_url="https://idp/authorize", client_id="c", + redirect_uri="https://app/cb", code_challenge=p["code_challenge"]) + assert "code_challenge_method=S256" in req["url"] and req["state"] and req["nonce"] + assert "response_type=code" in req["url"] + + +def test_callback_state_mismatch_is_csrf_rejected(): + assert sso.verify_callback(returned_state="a", expected_state="a")["valid"] is True + assert sso.verify_callback(returned_state="a", expected_state="b")["valid"] is False + assert sso.verify_callback(returned_state="a", expected_state="a", error="access_denied")["valid"] is False + + +def test_token_exchange_sends_the_pkce_verifier(): + tx = sso.build_token_exchange(token_url="https://idp/token", code="abc", redirect_uri="https://app/cb", + client_id="c", code_verifier="the-verifier") + assert tx["body"]["code_verifier"] == "the-verifier" and tx["body"]["grant_type"] == "authorization_code" + + +# --- OIDC ID token -------------------------------------------------------------------------------- + +def test_valid_hs256_id_token_verifies_with_subject(): + tok = sso.encode_jwt_hs256(_claims(), _SECRET) + r = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, nonce="n-1", now=_NOW, hs256_secret=_SECRET) + assert r["valid"] and r["sub"] == "alice" + + +def test_alg_none_is_always_refused(): + r = sso.verify_id_token(_mint("none", _claims(), sig=b""), issuer=_ISS, audience=_AUD, now=_NOW) + assert r["valid"] is False and "none" in r["reason"] + + +def test_hs256_wrong_secret_is_refused(): + tok = sso.encode_jwt_hs256(_claims(), _SECRET) + r = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, now=_NOW, hs256_secret="wrong") + assert r["valid"] is False and "signature" in r["reason"] + + +def test_asymmetric_alg_without_verifier_is_refused_never_trusted(): + r = sso.verify_id_token(_mint("RS256", _claims()), issuer=_ISS, audience=_AUD, now=_NOW) + assert r["valid"] is False and "asymmetric verifier" in r["reason"] + + +def test_asymmetric_alg_uses_the_injected_verifier(): + tok = _mint("RS256", _claims()) + ok = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, nonce="n-1", now=_NOW, + verify_signature=lambda si, sig, h: True) + bad = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, now=_NOW, + verify_signature=lambda si, sig, h: False) + assert ok["valid"] is True and bad["valid"] is False + + +def test_issuer_and_audience_are_enforced(): + tok = sso.encode_jwt_hs256(_claims(), _SECRET) + assert not sso.verify_id_token(tok, issuer="https://evil", audience=_AUD, now=_NOW, hs256_secret=_SECRET)["valid"] + assert not sso.verify_id_token(tok, issuer=_ISS, audience="other-app", now=_NOW, hs256_secret=_SECRET)["valid"] + + +def test_multiple_audiences_require_azp_to_be_us(): + tok = sso.encode_jwt_hs256(_claims(aud=[_AUD, "other"], azp="other"), _SECRET) + assert not sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, now=_NOW, hs256_secret=_SECRET)["valid"] + ok = sso.encode_jwt_hs256(_claims(aud=[_AUD, "other"], azp=_AUD), _SECRET) + assert sso.verify_id_token(ok, issuer=_ISS, audience=_AUD, nonce="n-1", now=_NOW, hs256_secret=_SECRET)["valid"] + + +def test_expired_token_is_refused(): + tok = sso.encode_jwt_hs256(_claims(exp=int(_NOW.timestamp()) - 3600), _SECRET) + r = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, now=_NOW, hs256_secret=_SECRET) + assert r["valid"] is False and "expired" in r["reason"] + + +def test_missing_iat_is_refused(): + c = _claims() + del c["iat"] + tok = sso.encode_jwt_hs256(c, _SECRET) + assert not sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, now=_NOW, hs256_secret=_SECRET)["valid"] + + +def test_nonce_mismatch_is_refused(): + tok = sso.encode_jwt_hs256(_claims(nonce="n-1"), _SECRET) + r = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, nonce="n-DIFFERENT", now=_NOW, hs256_secret=_SECRET) + assert r["valid"] is False and "nonce" in r["reason"] + + +def test_malformed_token_is_refused(): + assert not sso.verify_id_token("not.a.jwt.at.all", issuer=_ISS, audience=_AUD, now=_NOW)["valid"] + assert not sso.verify_id_token("garbage", issuer=_ISS, audience=_AUD, now=_NOW)["valid"] + + +# --- WebAuthn assertion -------------------------------------------------------------------------- + +_RP = "continuum.sovereign" +_ORIGIN = "https://continuum.sovereign" +_CHAL = b"\x11" * 32 + + +def _assertion(*, challenge_b64=None, origin=_ORIGIN, rp_id=_RP, flags=0x05, count=7, + ctype="webauthn.get"): + cd = {"type": ctype, "challenge": challenge_b64 or sso._b64url_encode(_CHAL), "origin": origin} + client_data = json.dumps(cd).encode() + auth_data = sso._sha256(rp_id.encode()) + bytes([flags]) + count.to_bytes(4, "big") + return client_data, auth_data + + +def test_valid_assertion_passes_with_injected_signature(): + cd, ad = _assertion() + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=6, verify_signature=lambda d, s: True) + assert r["valid"] and r["sign_count"] == 7 + + +def test_wrong_type_challenge_or_origin_are_refused(): + for over in ({"ctype": "webauthn.create"}, + {"challenge_b64": sso._b64url_encode(b"\x22" * 32)}, + {"origin": "https://evil.example"}): + cd, ad = _assertion(**over) + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=0, verify_signature=lambda d, s: True) + assert r["valid"] is False, over + + +def test_wrong_rp_id_is_refused(): + cd, ad = _assertion(rp_id="attacker.example") + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=0, verify_signature=lambda d, s: True) + assert r["valid"] is False and "rpIdHash" in r["reason"] + + +def test_user_present_flag_required(): + cd, ad = _assertion(flags=0x00) # UP not set + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=0, verify_signature=lambda d, s: True) + assert r["valid"] is False and "UP" in r["reason"] + + +def test_user_verification_enforced_when_required(): + cd, ad = _assertion(flags=0x01) # UP set, UV not + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=0, require_user_verification=True, + verify_signature=lambda d, s: True) + assert r["valid"] is False and "UV" in r["reason"] + + +def test_signature_counter_clone_is_refused(): + cd, ad = _assertion(count=5) + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=5, verify_signature=lambda d, s: True) + assert r["valid"] is False and "cloned" in r["reason"] + + +def test_bad_credential_signature_is_refused(): + cd, ad = _assertion() + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=6, verify_signature=lambda d, s: False) + assert r["valid"] is False and "signature" in r["reason"] + + +def test_counter_zero_disables_clone_check(): + # some authenticators never implement the counter; 0/0 must not trip clone detection. + cd, ad = _assertion(count=0) + r = sso.verify_assertion(client_data_json=cd, authenticator_data=ad, signature=b"s", + expected_challenge=_CHAL, expected_origin=_ORIGIN, rp_id=_RP, + prev_sign_count=0, verify_signature=lambda d, s: True) + assert r["valid"] is True + + +# --- bridge to the session core ------------------------------------------------------------------ + +def test_verified_identity_mints_a_session_and_invalid_does_not(): + key = b"k" + tok = sso.encode_jwt_hs256(_claims(), _SECRET) + ok = sso.verify_id_token(tok, issuer=_ISS, audience=_AUD, nonce="n-1", now=_NOW, hs256_secret=_SECRET) + sess = sso.session_from_identity(ok, key=key, tier="pro", now=_NOW) + assert sess is not None and login.verify_session(sess, key=key, now=_NOW)["valid"] + assert sso.session_from_identity({"valid": False}, key=key, now=_NOW) is None + + +if __name__ == "__main__": + import sys + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] + for fn in fns: + fn() + print(f"ok: {len(fns)} sso tests passed") + sys.exit(0) diff --git a/tools/validate.py b/tools/validate.py index 1c56a6b..97e187c 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -29,6 +29,7 @@ "capd/git-push-deploy.mesh.capd.json", "capd/git-push-webhook.mesh.capd.json", "capd/instant-rollback.mesh.capd.json", + "capd/sso-front-door.mesh.capd.json", "capd/provisioning-plane.mesh.capd.json", "tools/promotion_gate.py", "tools/portal_server.py", @@ -54,6 +55,7 @@ "tools/push_webhook.py", "tools/release_ledger.py", "tools/login.py", + "tools/sso.py", "tools/edge_worker.py", ] CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy") @@ -70,6 +72,7 @@ "capd/git-push-deploy.mesh.capd.json": "caps.dev.git-push-deploy", "capd/git-push-webhook.mesh.capd.json": "caps.dev.git-push-webhook", "capd/instant-rollback.mesh.capd.json": "caps.dev.instant-rollback", + "capd/sso-front-door.mesh.capd.json": "caps.dev.sso-front-door", "capd/provisioning-plane.mesh.capd.json": "caps.dev.provisioning", }