diff --git a/services/flask-backend/app/oidc.py b/services/flask-backend/app/oidc.py index 752e46f3..6f154ca9 100644 --- a/services/flask-backend/app/oidc.py +++ b/services/flask-backend/app/oidc.py @@ -6,15 +6,27 @@ - Initiate OIDC authorize redirect - Handle callback and user linking - CRUD for provider management (admin) + +Security hardening (v2): +- ID-token signature validation (OC1) +- Email-verified gating on auto-provisioning (OC2) +- PKCE + nonce anti-replay (OC1, PKCE) +- Token cookies instead of URL params (OH4) +- Link identity only via authenticated OIDC flow (OH5) """ from __future__ import annotations +import base64 +import hashlib import logging +import secrets +import time import uuid from datetime import datetime +from typing import Any -from quart import Blueprint, current_app, g, jsonify, redirect, request, session +from quart import Blueprint, Response, current_app, g, jsonify, redirect, request, session from .async_db import run_sync from .auth import auth_required, create_access_token, create_refresh_token @@ -26,6 +38,10 @@ oidc_bp = Blueprint("oidc", __name__) +# JWKS cache: {provider_id: {"fetched_at": timestamp, "keys": [...]}} +_jwks_cache: dict[int, dict[str, Any]] = {} +_JWKS_TTL_SECONDS = 3600 # 1 hour + def _define_oidc_tables(db) -> None: """Define OIDC tables for runtime use.""" @@ -38,6 +54,7 @@ def _define_oidc_tables(db) -> None: db.Field("client_id", "string", length=255), db.Field("client_secret_encrypted", "text"), db.Field("discovery_url", "text"), + db.Field("issuer", "string", length=512), # OAuth 2.0 iss claim for validation db.Field("enabled", "boolean", default=True), db.Field("tenant_id", "integer"), db.Field("created_at", "datetime"), @@ -51,6 +68,7 @@ def _define_oidc_tables(db) -> None: db.Field("provider_id", "reference oidc_providers"), db.Field("external_sub", "string", length=255), db.Field("external_email", "string", length=255), + db.Field("email_verified", "boolean", default=False), # Validated by provider db.Field("linked_at", "datetime"), migrate=False, ) @@ -77,6 +95,7 @@ def _init_oidc_tables(db) -> None: client_id VARCHAR(255) NOT NULL, client_secret_encrypted TEXT, discovery_url TEXT, + issuer VARCHAR(512), enabled BOOLEAN DEFAULT TRUE, tenant_id INTEGER REFERENCES tenants(id), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP @@ -89,6 +108,7 @@ def _init_oidc_tables(db) -> None: provider_id INTEGER NOT NULL REFERENCES oidc_providers(id) ON DELETE CASCADE, external_sub VARCHAR(255) NOT NULL, external_email VARCHAR(255), + email_verified BOOLEAN DEFAULT FALSE, linked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(provider_id, external_sub) ) @@ -105,6 +125,7 @@ def _init_oidc_tables(db) -> None: client_id VARCHAR(255) NOT NULL, client_secret_encrypted TEXT, discovery_url TEXT, + issuer VARCHAR(512), enabled BOOLEAN DEFAULT TRUE, tenant_id INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP @@ -117,6 +138,7 @@ def _init_oidc_tables(db) -> None: provider_id INTEGER NOT NULL REFERENCES oidc_providers(id), external_sub VARCHAR(255) NOT NULL, external_email VARCHAR(255), + email_verified BOOLEAN DEFAULT FALSE, linked_at DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE(provider_id, external_sub) ) @@ -166,6 +188,166 @@ def _decrypt_secret(ciphertext: str) -> str: return ciphertext +# --- PKCE & Nonce Helpers --- + + +def _generate_pkce_pair() -> tuple[str, str]: + """ + Generate PKCE code_verifier and code_challenge (S256). + + Returns: (code_verifier, code_challenge) + """ + code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("utf-8").rstrip("=") + challenge_bytes = hashlib.sha256(code_verifier.encode()).digest() + code_challenge = base64.urlsafe_b64encode(challenge_bytes).decode("utf-8").rstrip("=") + return code_verifier, code_challenge + + +def _generate_nonce() -> str: + """Generate a cryptographic nonce for replay protection.""" + return base64.urlsafe_b64encode(secrets.token_bytes(16)).decode("utf-8").rstrip("=") + + +# --- JWKS & ID Token Validation --- + + +async def _fetch_jwks(provider_id: int, jwks_uri: str) -> dict[str, Any]: + """ + Fetch and cache JWKS from provider. + + Returns: {"keys": [...]} + """ + global _jwks_cache + + # Check cache + if provider_id in _jwks_cache: + cached = _jwks_cache[provider_id] + if time.time() - cached["fetched_at"] < _JWKS_TTL_SECONDS: + return cached["keys"] + + # Fetch fresh + import httpx + + try: + async with httpx.AsyncClient() as http: + resp = await http.get(jwks_uri, timeout=10) + data = resp.json() + _jwks_cache[provider_id] = { + "fetched_at": time.time(), + "keys": data.get("keys", []), + } + return data.get("keys", []) + except Exception as e: + logger.error(f"JWKS fetch failed for provider {provider_id}: {e}") + # Return cached on error if available + if provider_id in _jwks_cache: + return _jwks_cache[provider_id]["keys"] + return [] + + +def _validate_id_token( + id_token: str, + jwks_keys: list[dict[str, Any]], + client_id: str, + issuer: str, + nonce: str, +) -> dict[str, Any]: + """ + Validate and decode ID token using JWKS. + + Hardened against algorithm confusion attacks (CVE-style): only asymmetric + algorithms (RS*, ES*, PS*) are accepted; HMAC (HS*) and 'none' are rejected. + + Returns: decoded claims dict + Raises: ValueError on validation failure + """ + from authlib.jose import JsonWebSignature, JsonWebKey + + # Allowed asymmetric signature algorithms only (reject HMAC and 'none') + ALLOWED_ALGORITHMS: list[str] = [ + "RS256", + "RS384", + "RS512", + "ES256", + "ES384", + "ES512", + "PS256", + "PS384", + "PS512", + ] + + # Decode header to find kid and validate alg early + try: + import json + + header_b64 = id_token.split(".")[0] + header_b64 += "=" * (4 - len(header_b64) % 4) + header = json.loads(base64.urlsafe_b64decode(header_b64)) + except Exception as e: + raise ValueError(f"Invalid ID token header: {e}") + + kid = header.get("kid") + if not kid: + raise ValueError("ID token missing 'kid' in header") + + # Defensively validate algorithm in header (belt-and-suspenders) + token_alg: str | None = header.get("alg") + if not token_alg: + raise ValueError("ID token missing 'alg' in header") + if token_alg not in ALLOWED_ALGORITHMS: + raise ValueError( + f"ID token algorithm '{token_alg}' not allowed. " + f"Only asymmetric algorithms {ALLOWED_ALGORITHMS} are accepted." + ) + + # Find matching JWKS key + jwks_key_data = None + for key_data in jwks_keys: + if key_data.get("kid") == kid: + jwks_key_data = key_data + break + + if not jwks_key_data: + raise ValueError(f"No JWKS key found for kid {kid}") + + # Ensure JWK is not a symmetric (oct) key — must be asymmetric (RSA/EC) + jwks_kty: str | None = jwks_key_data.get("kty") + if jwks_kty == "oct": + raise ValueError("ID token JWKS key is symmetric (oct) — only asymmetric keys allowed") + if jwks_kty not in ("RSA", "EC", "OKP"): + raise ValueError(f"ID token JWKS key type '{jwks_kty}' not supported") + + # Verify signature and decode with algorithm allowlist + try: + jws = JsonWebSignature(algorithms=ALLOWED_ALGORITHMS) + jwk = JsonWebKey.import_key(jwks_key_data) + jws_obj = jws.deserialize_compact(id_token, jwk) + claims_dict = json.loads(jws_obj.payload) + except Exception as e: + raise ValueError(f"ID token signature verification failed: {e}") + + # Validate claims + now = time.time() + + if claims_dict.get("exp", 0) < now: + raise ValueError("ID token expired") + + if claims_dict.get("iss") != issuer: + raise ValueError(f"ID token issuer mismatch: {claims_dict.get('iss')} != {issuer}") + + token_aud = claims_dict.get("aud") + if isinstance(token_aud, list): + if client_id not in token_aud: + raise ValueError(f"ID token aud mismatch: {token_aud} does not contain {client_id}") + elif token_aud != client_id: + raise ValueError(f"ID token aud mismatch: {token_aud} != {client_id}") + + if claims_dict.get("nonce") != nonce: + raise ValueError("ID token nonce mismatch or missing") + + return claims_dict + + # --- Public Endpoints --- @@ -204,6 +386,7 @@ async def authorize(provider_slug: str): Initiate OIDC authorization flow. Redirects user to the external identity provider. + Includes PKCE (OC1) and nonce (OC1) for security. """ db = get_db() _define_oidc_tables(db) @@ -212,10 +395,15 @@ async def authorize(provider_slug: str): if not provider or not provider.enabled: return jsonify({"error": "Provider not found or disabled"}), 404 - # Generate state for CSRF protection + # Generate state, PKCE, nonce for CSRF + replay protection state = str(uuid.uuid4()) + code_verifier, code_challenge = _generate_pkce_pair() + nonce = _generate_nonce() + session["oidc_state"] = state session["oidc_provider_id"] = provider.id + session["oidc_code_verifier"] = code_verifier + session["oidc_nonce"] = nonce try: from authlib.integrations.httpx_client import AsyncOAuth2Client @@ -234,6 +422,12 @@ async def authorize(provider_slug: str): async with httpx.AsyncClient() as http: metadata = (await http.get(discovery_url)).json() auth_endpoint = metadata.get("authorization_endpoint", "") + # Store issuer from discovery for later validation + issuer = metadata.get("issuer", "") + jwks_uri = metadata.get("jwks_uri", "") + if issuer and not provider.issuer: + # Optionally cache issuer from discovery (advisory only) + pass else: return jsonify({"error": "No discovery URL configured"}), 400 @@ -247,6 +441,9 @@ async def authorize(provider_slug: str): redirect_uri=callback_url, state=state, scope="openid email profile", + nonce=nonce, + code_challenge=code_challenge, + code_challenge_method="S256", ) return redirect(uri) @@ -263,7 +460,10 @@ async def callback(provider_slug: str): """ Handle OIDC callback from external provider. - Links or creates user, then issues JWT tokens. + Security hardening (OC1, OC2, OH4): + - Validates ID token signature, iss, aud, exp, nonce (OC1) + - Checks email_verified on auto-provisioning (OC2) + - Returns tokens in httpOnly cookies, not URL params (OH4) """ db = get_db() _define_oidc_tables(db) @@ -276,6 +476,7 @@ async def callback(provider_slug: str): state = request.args.get("state", "") expected_state = session.pop("oidc_state", None) if not state or state != expected_state: + logger.warning(f"State mismatch for {provider_slug}") return jsonify({"error": "Invalid state parameter"}), 400 code = request.args.get("code", "") @@ -283,8 +484,16 @@ async def callback(provider_slug: str): error = request.args.get("error", "unknown") return jsonify({"error": f"Authorization failed: {error}"}), 400 + # Retrieve PKCE verifier and nonce from session + code_verifier = session.pop("oidc_code_verifier", None) + nonce = session.pop("oidc_nonce", None) + if not code_verifier or not nonce: + logger.warning(f"Missing PKCE/nonce for {provider_slug}") + return jsonify({"error": "Invalid session state"}), 400 + try: from authlib.integrations.httpx_client import AsyncOAuth2Client + import httpx client_secret = _decrypt_secret(provider.client_secret_encrypted or "") @@ -298,39 +507,95 @@ async def callback(provider_slug: str): client_secret=client_secret, ) - # Discover token endpoint - import httpx - + # Discover endpoints and issuer async with httpx.AsyncClient() as http: metadata = (await http.get(provider.discovery_url)).json() token_endpoint = metadata.get("token_endpoint", "") - userinfo_endpoint = metadata.get("userinfo_endpoint", "") + jwks_uri = metadata.get("jwks_uri", "") + issuer = metadata.get("issuer", "") + + if not issuer: + logger.error(f"No issuer in discovery for {provider_slug}") + return jsonify({"error": "Provider configuration incomplete"}), 500 - # Exchange code for tokens + # Exchange code for tokens, requesting id_token token_response = await client.fetch_token( token_endpoint, code=code, redirect_uri=callback_url, + code_verifier=code_verifier, # PKCE verification ) - # Get user info - async with httpx.AsyncClient() as http: - userinfo = ( - await http.get( - userinfo_endpoint, - headers={ - "Authorization": f"Bearer {token_response['access_token']}" - }, - ) - ).json() + id_token = token_response.get("id_token") + if not id_token: + logger.warning(f"No id_token in response for {provider_slug}") + return jsonify({"error": "Provider did not return id_token"}), 400 + + # Fetch JWKS and validate ID token signature + claims + jwks_keys = await _fetch_jwks(provider.id, jwks_uri) + if not jwks_keys: + logger.error(f"Failed to fetch JWKS for provider {provider_slug}") + return jsonify({"error": "JWKS unavailable"}), 500 + + try: + id_token_claims = _validate_id_token( + id_token, + jwks_keys, + provider.client_id, + issuer, + nonce, + ) + except ValueError as e: + logger.warning(f"ID token validation failed: {e}") + return jsonify({"error": "ID token validation failed"}), 401 - external_sub = userinfo.get("sub", "") - external_email = userinfo.get("email", "") - external_name = userinfo.get("name", "") + # Extract validated claims + external_sub = id_token_claims.get("sub", "") + external_email = id_token_claims.get("email", "") + external_name = id_token_claims.get("name", "") + email_verified = id_token_claims.get("email_verified", False) if not external_sub: - return jsonify({"error": "No subject in ID token"}), 400 + logger.warning(f"Missing sub claim in ID token for {provider_slug}") + return jsonify({"error": "Invalid ID token"}), 400 + + # Check for link_mode (authenticated user linking) + link_mode = session.pop("oidc_link_mode", False) + link_user_id = session.pop("oidc_link_user_id", None) + + if link_mode and link_user_id: + # User is linking an external identity to their existing account (OH5) + from .models import get_user_by_id + + user = get_user_by_id(link_user_id) + if not user: + logger.warning(f"Link mode: user {link_user_id} not found") + return jsonify({"error": "User not found"}), 404 + + if not email_verified: + logger.warning(f"Link mode: email not verified for {external_email}") + return jsonify({ + "error": "Email not verified by provider. Cannot link." + }), 403 + + # Create link + db.user_oidc_links.insert( + user_id=user["id"], + provider_id=provider.id, + external_sub=external_sub, + external_email=external_email, + email_verified=email_verified, + ) + db.commit() + + # Redirect to account settings or profile page (not through cookies, user already auth'd) + frontend_url = current_app.config.get("FRONTEND_URL", "") + if not frontend_url: + frontend_url = f"{request.scheme}://{request.host}" + return redirect(f"{frontend_url}/account?linked=ok") + + # Standard login flow (not link_mode) # Check for existing link link = ( db( @@ -343,22 +608,22 @@ async def callback(provider_slug: str): if link: # Existing linked user — login - user = await run_sync( - lambda uid: __import__( - "app.models", fromlist=["get_user_by_id"] - ).get_user_by_id(uid), - link.user_id, - ) - if not user: - from .models import get_user_by_id + from .models import get_user_by_id - user = get_user_by_id(link.user_id) + user = get_user_by_id(link.user_id) else: - # Try to find user by email and auto-link - user = get_user_by_email(external_email) if external_email else None - - if not user and external_email: - # Create new user + # No existing link. Check if auto-linking is enabled. + # Default: require email_verified + config flag + auto_link_enabled = current_app.config.get("OIDC_AUTO_LINK_VERIFIED_EMAIL", False) + + if auto_link_enabled and email_verified and external_email: + # Auto-link existing user by verified email + user = get_user_by_email(external_email) + else: + user = None + + if not user and external_email and email_verified: + # Auto-provision new user (only if email_verified) from .auth import hash_password from .models import create_user @@ -393,8 +658,15 @@ async def callback(provider_slug: str): scope_level="tenant", scope_id=default_tenant.id, ) + elif not user and not email_verified: + # Reject unverified email auto-provisioning + logger.warning(f"Rejecting unverified email {external_email} for {provider_slug}") + return jsonify({ + "error": "Email not verified by provider. Please verify and try again." + }), 403 if not user: + logger.warning(f"No user found/created for {external_email}") return jsonify({"error": "Could not create or find user"}), 400 # Create link @@ -403,60 +675,74 @@ async def callback(provider_slug: str): provider_id=provider.id, external_sub=external_sub, external_email=external_email, + email_verified=email_verified, ) db.commit() if not user.get("is_active"): + logger.warning(f"Login attempt for deactivated user {user.get('id')}") return jsonify({"error": "Account deactivated"}), 401 # Issue JWT tokens access_token = create_access_token(user["id"], user.get("role", "viewer")) refresh_token, _ = create_refresh_token(user["id"]) - # Redirect to frontend with tokens + # Redirect to frontend without tokens in URL (OH4) frontend_url = current_app.config.get("FRONTEND_URL", "") if not frontend_url: frontend_url = f"{request.scheme}://{request.host}" - redirect_url = ( - f"{frontend_url}/login" - f"?access_token={access_token}" - f"&refresh_token={refresh_token}" - f"&token_type=Bearer" + redirect_url = f"{frontend_url}/login?sso=ok" + + # Build response with httpOnly cookie tokens + resp = redirect(redirect_url) + resp.set_cookie( + "access_token", + access_token, + httponly=True, + secure=True, + samesite="Lax", + max_age=3600, + path="/", + ) + resp.set_cookie( + "refresh_token", + refresh_token, + httponly=True, + secure=True, + samesite="Lax", + max_age=2592000, # 30 days + path="/api/v1/auth/refresh", ) - return redirect(redirect_url) + return resp except ImportError: + logger.error("authlib not installed") return jsonify({"error": "authlib not installed"}), 500 except Exception as e: - logger.error(f"OIDC callback error: {e}") + logger.error(f"OIDC callback error for {provider_slug}: {e}", exc_info=True) return jsonify({"error": "SSO callback failed"}), 500 # --- Authenticated Endpoints --- -@oidc_bp.route("/oidc/link", methods=["POST"]) +@oidc_bp.route("/oidc/link/", methods=["GET"]) @auth_required -async def link_identity(): +async def initiate_link_identity(provider_slug: str): """ - Link an external identity to the current user. + Initiate OIDC linking for an authenticated user (OH5). - Body: { "provider_id": 1, "external_sub": "...", "external_email": "..." } + Redirects to provider auth flow with link_mode=true in state. + On callback, if user is authenticated, the external identity is linked. """ db = get_db() _define_oidc_tables(db) - data = await request.get_json() - if not data: - return jsonify({"error": "Request body required"}), 400 - - provider_id = data.get("provider_id") - external_sub = data.get("external_sub") - - if not provider_id or not external_sub: - return jsonify({"error": "provider_id and external_sub required"}), 400 + provider = db(db.oidc_providers.slug == provider_slug).select().first() + if not provider or not provider.enabled: + return jsonify({"error": "Provider not found or disabled"}), 404 user_id = g.current_user["id"] @@ -464,7 +750,7 @@ async def link_identity(): existing = ( db( (db.user_oidc_links.user_id == user_id) - & (db.user_oidc_links.provider_id == provider_id) + & (db.user_oidc_links.provider_id == provider.id) ) .select() .first() @@ -472,15 +758,59 @@ async def link_identity(): if existing: return jsonify({"error": "Already linked to this provider"}), 409 - db.user_oidc_links.insert( - user_id=user_id, - provider_id=provider_id, - external_sub=external_sub, - external_email=data.get("external_email", ""), - ) - db.commit() + # Generate state, PKCE, nonce with link_mode flag + state = str(uuid.uuid4()) + code_verifier, code_challenge = _generate_pkce_pair() + nonce = _generate_nonce() - return jsonify({"message": "Identity linked"}), 201 + session["oidc_state"] = state + session["oidc_provider_id"] = provider.id + session["oidc_code_verifier"] = code_verifier + session["oidc_nonce"] = nonce + session["oidc_link_mode"] = True # Flag to link on callback + session["oidc_link_user_id"] = user_id + + try: + from authlib.integrations.httpx_client import AsyncOAuth2Client + import httpx + + client_secret = _decrypt_secret(provider.client_secret_encrypted or "") + client = AsyncOAuth2Client( + client_id=provider.client_id, + client_secret=client_secret, + ) + + # Discover OIDC endpoints + discovery_url = provider.discovery_url + if discovery_url: + async with httpx.AsyncClient() as http: + metadata = (await http.get(discovery_url)).json() + auth_endpoint = metadata.get("authorization_endpoint", "") + else: + return jsonify({"error": "No discovery URL configured"}), 400 + + callback_url = ( + f"{request.scheme}://{request.host}" + f"/api/v1/auth/oidc/{provider_slug}/callback" + ) + + uri, _ = client.create_authorization_url( + auth_endpoint, + redirect_uri=callback_url, + state=state, + scope="openid email profile", + nonce=nonce, + code_challenge=code_challenge, + code_challenge_method="S256", + ) + + return redirect(uri) + + except ImportError: + return jsonify({"error": "authlib not installed"}), 500 + except Exception as e: + logger.error(f"OIDC link initiate error: {e}") + return jsonify({"error": "Failed to initiate linking"}), 500 @oidc_bp.route("/oidc/link/", methods=["DELETE"]) @@ -590,7 +920,7 @@ async def update_provider(provider_id: int): return jsonify({"error": "Request body required"}), 400 update_fields = {} - for field in ["name", "provider_type", "client_id", "discovery_url", "enabled"]: + for field in ["name", "provider_type", "client_id", "discovery_url", "issuer", "enabled"]: if field in data: update_fields[field] = data[field] diff --git a/services/flask-backend/tests/test_oidc_security.py b/services/flask-backend/tests/test_oidc_security.py new file mode 100644 index 00000000..7a10e138 --- /dev/null +++ b/services/flask-backend/tests/test_oidc_security.py @@ -0,0 +1,427 @@ +""" +OIDC security regression tests. + +Tests for algorithm confusion attack (CVE-style): +- Forged HS256 tokens signed with public key are rejected +- Valid RS256 tokens are accepted +- Tokens with 'alg: none' are rejected +- Symmetric (oct) JWKS keys are rejected +""" + +from __future__ import annotations + +import base64 +import json +import time +from datetime import datetime, timedelta, timezone + +import pytest + +# regression: algorithm confusion vulnerability in _validate_id_token + + +def _encode_b64url(data: bytes) -> str: + """Encode bytes to URL-safe base64 without padding.""" + return base64.urlsafe_b64encode(data).decode("utf-8").rstrip("=") + + +def _create_test_keypair() -> tuple[str, str]: + """ + Generate RSA 2048 keypair for testing. + + Returns: (private_pem, public_pem) + """ + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import rsa + + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("utf-8") + + public_pem = private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ).decode("utf-8") + + return private_pem, public_pem + + +def _build_jwks_from_public_key(public_pem: str, kid: str = "test-kid") -> dict: + """ + Build a JWKS structure from a public RSA key. + + Returns: {"keys": [...]} + """ + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import rsa + + public_key = serialization.load_pem_public_key( + public_pem.encode(), backend=default_backend() + ) + + if not isinstance(public_key, rsa.RSAPublicKey): + raise ValueError("Expected RSA public key") + + numbers = public_key.public_numbers() + + # Build JWK manually in JWKS format + jwk_keys = [ + { + "kty": "RSA", + "use": "sig", + "kid": kid, + "n": _encode_b64url(numbers.n.to_bytes(256, "big")), + "e": _encode_b64url(numbers.e.to_bytes(3, "big")), + "alg": "RS256", + } + ] + + return {"keys": jwk_keys} + + +def _sign_jwt_rs256(payload: dict, private_pem: str) -> str: + """ + Sign a JWT payload with RS256 (asymmetric). + + Returns: JWT token string + """ + import jwt + + header = {"alg": "RS256", "typ": "JWT", "kid": "test-kid"} + + return jwt.encode(payload, private_pem, algorithm="RS256", headers=header) + + +def _sign_jwt_hs256_with_public_key(payload: dict, public_key_str: str) -> str: + """ + Forge a JWT token with HS256 algorithm, signed using the public key as HMAC secret. + + This is the attack: use the public key (from JWKS) as an HMAC secret. + Since PyJWT now prevents this, we manually construct the JWT. + + Returns: Forged JWT token string + """ + import hashlib + import hmac + + header = {"alg": "HS256", "typ": "JWT", "kid": "test-kid"} + + header_b64 = _encode_b64url(json.dumps(header).encode()) + payload_b64 = _encode_b64url(json.dumps(payload).encode()) + + # Create signature using public key as HMAC secret (the attack) + signing_input = f"{header_b64}.{payload_b64}".encode() + signature = hmac.new( + public_key_str.encode(), signing_input, hashlib.sha256 + ).digest() + signature_b64 = _encode_b64url(signature) + + return f"{header_b64}.{payload_b64}.{signature_b64}" + + +def _create_none_token(payload: dict) -> str: + """ + Create a JWT token with 'alg: none' (unverified signature). + + Returns: Forged token + """ + header = {"alg": "none", "typ": "JWT", "kid": "test-kid"} + + header_b64 = _encode_b64url(json.dumps(header).encode()) + payload_b64 = _encode_b64url(json.dumps(payload).encode()) + # No signature for 'alg: none' + return f"{header_b64}.{payload_b64}." + + +@pytest.mark.asyncio +async def test_valid_rs256_token_accepted(app) -> None: + """Valid RS256 token is accepted.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "user-123", + "email": "test@example.com", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + "email_verified": True, + } + + token = _sign_jwt_rs256(payload, private_pem) + + # This should NOT raise + result = _validate_id_token( + token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + assert result["sub"] == "user-123" + assert result["email"] == "test@example.com" + + +@pytest.mark.asyncio +async def test_forged_hs256_token_rejected(app) -> None: + """Forged HS256 token signed with public key is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "attacker-admin", + "email": "attacker@example.com", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + "email_verified": True, + "role": "admin", # Attacker tries to claim admin role + } + + # Create forged token using public key as HMAC secret + forged_token = _sign_jwt_hs256_with_public_key(payload, public_pem) + + # This MUST raise ValueError due to algorithm not allowed + with pytest.raises(ValueError, match="algorithm.*not allowed|HS256"): + _validate_id_token( + forged_token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_none_algorithm_rejected(app) -> None: + """Token with 'alg: none' is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "attacker", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + } + + # Create unverified 'none' token + none_token = _create_none_token(payload) + + # This MUST raise ValueError + with pytest.raises(ValueError, match="algorithm.*not allowed|none"): + _validate_id_token( + none_token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_symmetric_oct_key_rejected(app) -> None: + """JWKS with symmetric (oct) key is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "user-123", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + } + + token = _sign_jwt_rs256(payload, private_pem) + + # JWKS with symmetric (oct) key instead of RSA + malicious_jwks_keys = [ + { + "kty": "oct", + "kid": "test-kid", + "k": _encode_b64url(b"this-is-a-secret"), # Symmetric key + "alg": "HS256", + } + ] + + # This MUST raise ValueError because kty is 'oct' (symmetric) + with pytest.raises(ValueError, match="symmetric|oct"): + _validate_id_token( + token, + malicious_jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_expired_token_rejected(app) -> None: + """Expired token is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now - timedelta(hours=1)).timestamp()) # Already expired + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "user-123", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int((now - timedelta(hours=2)).timestamp()), + } + + token = _sign_jwt_rs256(payload, private_pem) + + with pytest.raises(ValueError, match="expired"): + _validate_id_token( + token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_missing_kid_rejected(app) -> None: + """Token without 'kid' header is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "user-123", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + } + + # Manually create token without kid + import jwt + + header = {"alg": "RS256", "typ": "JWT"} # No 'kid' + token = jwt.encode(payload, private_pem, algorithm="RS256", headers=header) + + with pytest.raises(ValueError, match="missing.*kid"): + _validate_id_token( + token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_issuer_mismatch_rejected(app) -> None: + """Token with mismatched issuer is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://attacker.com", # Wrong issuer + "aud": "test-client-id", + "sub": "user-123", + "nonce": "test-nonce-123", + "exp": exp, + "iat": int(now.timestamp()), + } + + token = _sign_jwt_rs256(payload, private_pem) + + with pytest.raises(ValueError, match="issuer mismatch"): + _validate_id_token( + token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", # Expected issuer + nonce="test-nonce-123", + ) + + +@pytest.mark.asyncio +async def test_nonce_mismatch_rejected(app) -> None: + """Token with mismatched nonce is rejected.""" + from app.oidc import _validate_id_token + + private_pem, public_pem = _create_test_keypair() + jwks_data = _build_jwks_from_public_key(public_pem, kid="test-kid") + jwks_keys = jwks_data["keys"] + + now = datetime.now(timezone.utc) + exp = int((now + timedelta(hours=1)).timestamp()) + + payload = { + "iss": "https://example.com", + "aud": "test-client-id", + "sub": "user-123", + "nonce": "wrong-nonce", # Wrong nonce + "exp": exp, + "iat": int(now.timestamp()), + } + + token = _sign_jwt_rs256(payload, private_pem) + + with pytest.raises(ValueError, match="nonce mismatch"): + _validate_id_token( + token, + jwks_keys, + client_id="test-client-id", + issuer="https://example.com", + nonce="test-nonce-123", # Expected nonce + )