"""Tests for turnstone.core.oidc — OIDC authentication support.""" from __future__ import annotations import asyncio import base64 import dataclasses import hashlib import types import urllib.parse from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch import httpx import jwt as pyjwt import pytest from tests.conftest import make_oidc_test_config as _make_config from turnstone.core.oidc import ( OIDCError, OIDCKeyNotFoundError, _ensure_default_role, _sanitize_log_text, apply_role_mapping, build_authorize_url, discover_oidc, exchange_code, generate_pkce_verifier, initialize_oidc_state, load_oidc_config, provision_oidc_user, validate_discovered_endpoint, validate_id_token, validate_issuer_url, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _mock_storage(**overrides): """Build a MagicMock with sensible storage defaults.""" s = MagicMock() s.get_oidc_identity.return_value = overrides.get("identity") s.get_user.return_value = overrides.get("user") s.get_user_by_username.return_value = overrides.get("user_by_username") s.get_role.return_value = overrides.get("role") s.find_existing_usernames.return_value = overrides.get("existing_usernames", set()) s.replace_oidc_roles.return_value = overrides.get("replace_oidc_roles", (set(), set())) return s def _mock_async_client(mock_get): """Build a patched httpx.AsyncClient context manager for async tests.""" class _AsyncCtx: async def __aenter__(self): return self async def __aexit__(self, *args): pass async def get(self, url): return await mock_get(url) return _AsyncCtx() # --------------------------------------------------------------------------- # Config Loading # --------------------------------------------------------------------------- class TestLoadOIDCConfig: def test_load_oidc_config_from_env(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_SCOPES", "openid") monkeypatch.setenv("TURNSTONE_OIDC_PROVIDER_NAME", "Okta") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.enabled is True assert cfg.issuer == "https://auth.example.com" assert cfg.client_id == "cid" assert cfg.client_secret == "csecret" assert cfg.scopes == "openid" assert cfg.provider_name == "Okta" def test_load_oidc_config_allow_private_network_env(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.internal.example") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_ALLOW_PRIVATE_NETWORK", "true") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.allow_private_network is True def test_load_oidc_config_allow_private_network_toml(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.internal.example") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_ALLOW_PRIVATE_NETWORK", raising=False) with patch( "turnstone.core.config.load_config", return_value={"allow_private_network": True}, ): cfg = load_oidc_config() assert cfg.allow_private_network is True def test_load_oidc_config_allow_private_network_default_off(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_ALLOW_PRIVATE_NETWORK", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.allow_private_network is False def test_load_oidc_config_capture_user_credential_env(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_CAPTURE_USER_CREDENTIAL", "true") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.capture_user_credential is True def test_load_oidc_config_capture_user_credential_toml(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_CAPTURE_USER_CREDENTIAL", raising=False) with patch( "turnstone.core.config.load_config", return_value={"capture_user_credential": True}, ): cfg = load_oidc_config() assert cfg.capture_user_credential is True def test_load_oidc_config_capture_default_off(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_CAPTURE_USER_CREDENTIAL", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.capture_user_credential is False def test_capture_appends_offline_access_to_scopes(self, monkeypatch): """Enabling capture requests offline_access without operator scope edits.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_SCOPES", raising=False) with patch( "turnstone.core.config.load_config", return_value={"capture_user_credential": True}, ): cfg = load_oidc_config() assert cfg.scopes == "openid email profile offline_access" def test_capture_scope_append_is_idempotent(self, monkeypatch): """An operator who already lists offline_access doesn't get it twice.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_SCOPES", "openid offline_access email") with patch( "turnstone.core.config.load_config", return_value={"capture_user_credential": True}, ): cfg = load_oidc_config() assert cfg.scopes == "openid offline_access email" def test_no_capture_leaves_scopes_untouched(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_SCOPES", raising=False) monkeypatch.delenv("TURNSTONE_OIDC_CAPTURE_USER_CREDENTIAL", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.scopes == "openid email profile" def test_load_oidc_config_disabled_when_missing(self, monkeypatch): monkeypatch.delenv("TURNSTONE_OIDC_ISSUER", raising=False) monkeypatch.delenv("TURNSTONE_OIDC_CLIENT_ID", raising=False) monkeypatch.delenv("TURNSTONE_OIDC_CLIENT_SECRET", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.enabled is False def test_load_oidc_config_partial_env(self, monkeypatch): """Only issuer set, no client_id -> enabled=False.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.delenv("TURNSTONE_OIDC_CLIENT_ID", raising=False) monkeypatch.delenv("TURNSTONE_OIDC_CLIENT_SECRET", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.enabled is False assert cfg.issuer == "https://auth.example.com" assert cfg.client_id == "" def test_load_oidc_config_role_map_parsing(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_ROLE_CLAIM", "roles") monkeypatch.setenv("TURNSTONE_OIDC_ROLE_MAP", "admin:builtin-admin,eng:builtin-operator") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.role_claim == "roles" assert cfg.role_map == {"admin": "builtin-admin", "eng": "builtin-operator"} def test_load_oidc_config_password_enabled_false(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_PASSWORD_ENABLED", "false") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.enabled is True assert cfg.password_enabled is False def test_load_oidc_config_password_enabled_true(self, monkeypatch): monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_PASSWORD_ENABLED", "true") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.password_enabled is True def test_load_oidc_config_role_map_empty_entries(self, monkeypatch): """Role map with empty/whitespace entries should be silently skipped.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_ROLE_MAP", "admin:builtin-admin, , :, foo:") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.role_map == {"admin": "builtin-admin"} def test_load_oidc_config_defaults(self, monkeypatch): """Defaults for scopes and provider_name when not set.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_SCOPES", raising=False) monkeypatch.delenv("TURNSTONE_OIDC_PROVIDER_NAME", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.scopes == "openid email profile" assert cfg.provider_name == "SSO" def test_load_oidc_config_redirect_base_from_env(self, monkeypatch): """TURNSTONE_OIDC_REDIRECT_BASE populates redirect_base.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://app.example.com") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "https://app.example.com" def test_load_oidc_config_redirect_base_strips_trailing_slash(self, monkeypatch): """Trailing slashes are stripped from redirect_base.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://app.example.com/") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "https://app.example.com" def test_load_oidc_config_redirect_base_default_empty(self, monkeypatch): """redirect_base defaults to empty string when not set.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_REDIRECT_BASE", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_rejects_path(self, monkeypatch): """redirect_base with a path component is rejected (falls back to empty).""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://app.example.com/subpath") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_rejects_no_scheme(self, monkeypatch): """redirect_base without a scheme is rejected.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "app.example.com") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_rejects_userinfo(self, monkeypatch): """redirect_base with userinfo (user:pass@host) is rejected.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://user:pass@app.example.com") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_rejects_invalid_port(self, monkeypatch): """redirect_base with non-numeric port is rejected.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://app.example.com:abc") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_rejects_missing_hostname(self, monkeypatch): """redirect_base without a hostname is rejected.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "https://") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "" def test_load_oidc_config_redirect_base_allows_http(self, monkeypatch): """http:// redirect_base is allowed (with warning) for local dev.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv("TURNSTONE_OIDC_REDIRECT_BASE", "http://localhost:8000") with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.redirect_base == "http://localhost:8000" def test_load_oidc_config_parses_trusted_endpoint_hosts(self, monkeypatch): """TURNSTONE_OIDC_TRUSTED_ENDPOINT_HOSTS is split, lowercased, and trimmed.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.setenv( "TURNSTONE_OIDC_TRUSTED_ENDPOINT_HOSTS", "Foo.Example.com, BAR.example.com ,, ", ) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.trusted_endpoint_hosts == ("foo.example.com", "bar.example.com") def test_load_oidc_config_trusted_endpoint_hosts_default_empty(self, monkeypatch): """trusted_endpoint_hosts defaults to () when env var is absent.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_TRUSTED_ENDPOINT_HOSTS", raising=False) with patch("turnstone.core.config.load_config", return_value={}): cfg = load_oidc_config() assert cfg.trusted_endpoint_hosts == () def test_load_oidc_config_trusted_endpoint_hosts_from_toml_list(self, monkeypatch): """config.toml may provide trusted_endpoint_hosts as a list.""" monkeypatch.setenv("TURNSTONE_OIDC_ISSUER", "https://auth.example.com") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_ID", "cid") monkeypatch.setenv("TURNSTONE_OIDC_CLIENT_SECRET", "csecret") monkeypatch.delenv("TURNSTONE_OIDC_TRUSTED_ENDPOINT_HOSTS", raising=False) with patch( "turnstone.core.config.load_config", return_value={"trusted_endpoint_hosts": ["FOO.example.com", "bar.example.com"]}, ): cfg = load_oidc_config() assert cfg.trusted_endpoint_hosts == ("foo.example.com", "bar.example.com") # --------------------------------------------------------------------------- # SSRF Validation # --------------------------------------------------------------------------- class TestValidateIssuerURL: """Tests for ``validate_issuer_url`` SSRF protection.""" def test_valid_https_url(self): """Public HTTPS issuer URL passes validation.""" # Should not raise -- mock DNS to return a public IP. with patch( "socket.getaddrinfo", return_value=[ (2, 1, 6, "", ("93.184.216.34", 0)), ], ): validate_issuer_url("https://idp.example.com") def test_private_address_hint_mentions_opt_in(self): """The rejection message points the operator at allow_private_network.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]), pytest.raises(OIDCError, match="allow_private_network"), ): validate_issuer_url("https://auth.internal.example") def test_allow_private_accepts_private_issuer(self): """The opt-in accepts an issuer resolving to RFC 1918 space.""" with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]): validate_issuer_url("https://auth.internal.example", allow_private=True) def test_allow_private_still_rejects_link_local(self): """Link-local (cloud metadata) is refused even with the opt-in.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("169.254.169.254", 0))]), pytest.raises(OIDCError, match="refused even with private"), ): validate_issuer_url("https://md.internal.example", allow_private=True) def test_rejects_http_non_localhost(self): """HTTP is rejected for non-localhost hosts.""" with pytest.raises(OIDCError, match="must use HTTPS"): validate_issuer_url("http://idp.example.com") def test_allows_http_localhost(self): """HTTP is allowed for localhost (development).""" with patch( "socket.getaddrinfo", return_value=[ (2, 1, 6, "", ("127.0.0.1", 0)), ], ): validate_issuer_url("http://localhost:8080") def test_allows_http_localhost_subdomain(self): """HTTP is allowed for *.localhost subdomains.""" with patch( "socket.getaddrinfo", return_value=[ (2, 1, 6, "", ("127.0.0.1", 0)), ], ): validate_issuer_url("http://keycloak.localhost:8080") def test_rejects_embedded_credentials(self): """URLs with userinfo (user:pass@host) are rejected.""" with pytest.raises(OIDCError, match="embedded credentials"): validate_issuer_url("https://admin:secret@idp.example.com") def test_rejects_username_only(self): """URLs with just a username are rejected.""" with pytest.raises(OIDCError, match="embedded credentials"): validate_issuer_url("https://admin@idp.example.com") def test_rejects_no_hostname(self): """URLs without a hostname are rejected.""" with pytest.raises(OIDCError, match="no hostname"): validate_issuer_url("https://") def test_rejects_private_10_range(self): """Hostnames resolving to 10.x.x.x are rejected.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.1", 0))]), pytest.raises(OIDCError, match="non-public address.*10.0.0.1"), ): validate_issuer_url("https://internal.corp.example.com") def test_rejects_private_172_range(self): """Hostnames resolving to 172.16-31.x.x are rejected.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("172.16.0.1", 0))]), pytest.raises(OIDCError, match="non-public address.*172.16.0.1"), ): validate_issuer_url("https://internal.corp.example.com") def test_rejects_private_192_168_range(self): """Hostnames resolving to 192.168.x.x are rejected.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("192.168.1.1", 0))]), pytest.raises(OIDCError, match="non-public address.*192.168.1.1"), ): validate_issuer_url("https://internal.corp.example.com") def test_rejects_loopback_127(self): """Hostnames resolving to 127.x.x.x are rejected (non-localhost host).""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("127.0.0.1", 0))]), pytest.raises(OIDCError, match="non-public address.*127.0.0.1"), ): validate_issuer_url("https://evil.example.com") def test_rejects_ipv6_loopback(self): """Hostnames resolving to ::1 are rejected (non-localhost host).""" with ( patch("socket.getaddrinfo", return_value=[(10, 1, 6, "", ("::1", 0, 0, 0))]), pytest.raises(OIDCError, match="non-public address.*::1"), ): validate_issuer_url("https://evil.example.com") def test_rejects_ipv6_private(self): """Hostnames resolving to fc00::/7 are rejected.""" with ( patch("socket.getaddrinfo", return_value=[(10, 1, 6, "", ("fd00::1", 0, 0, 0))]), pytest.raises(OIDCError, match="non-public address.*fd00::1"), ): validate_issuer_url("https://evil.example.com") def test_rejects_link_local(self): """Hostnames resolving to link-local addresses are rejected. Refused as link-local rather than as merely non-public, and WITHOUT the allow_private_network hint: that opt-in never admits link-local, so offering it would send the operator to a dead end. """ with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("169.254.169.254", 0))]), pytest.raises(OIDCError, match="link-local.*169.254.169.254") as exc_info, ): validate_issuer_url("https://metadata.internal") assert "allow_private_network" not in str(exc_info.value) def test_rejects_unresolvable_hostname(self): """DNS resolution failure is rejected.""" import socket as _socket with ( patch("socket.getaddrinfo", side_effect=_socket.gaierror("not found")), pytest.raises(OIDCError, match="cannot be resolved"), ): validate_issuer_url("https://nonexistent.invalid") def test_rejects_mixed_addresses(self): """If any resolved address is private, the URL is rejected.""" with ( patch( "socket.getaddrinfo", return_value=[ (2, 1, 6, "", ("93.184.216.34", 0)), (2, 1, 6, "", ("10.0.0.1", 0)), ], ), pytest.raises(OIDCError, match="non-public address.*10.0.0.1"), ): validate_issuer_url("https://dual-homed.example.com") def test_discover_rejects_ssrf(self): """discover_oidc returns enabled=False when issuer URL fails SSRF check.""" config = _make_config( issuer="http://10.0.0.1:8080", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) async def _run(): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) # --------------------------------------------------------------------------- # Discovered Endpoint Validation # --------------------------------------------------------------------------- class TestValidateDiscoveredEndpoint: """Tests for ``validate_discovered_endpoint`` SSRF + same-origin protection.""" _PUBLIC_ADDR = [(2, 1, 6, "", ("93.184.216.34", 0))] _PRIVATE_ADDR = [(2, 1, 6, "", ("169.254.169.254", 0))] _LOOPBACK_ADDR = [(2, 1, 6, "", ("127.0.0.1", 0))] @staticmethod def _issuer(url: str = "https://idp.example.com") -> urllib.parse.ParseResult: return urllib.parse.urlparse(url) def test_valid_same_origin(self): """Same scheme/host/port as issuer passes.""" with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR): validate_discovered_endpoint( "https://idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_private_endpoint_hint_mentions_opt_in(self): """A discovered endpoint resolving private carries the opt-in hint just like the issuer does — the remediation is the same knob.""" with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]), pytest.raises(OIDCError, match="allow_private_network"), ): validate_discovered_endpoint( "https://idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_http_when_issuer_is_https(self): """http:// discovered endpoint rejected when issuer is https://.""" with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), pytest.raises(OIDCError, match="must use HTTPS"), ): validate_discovered_endpoint( "http://idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_allows_http_when_issuer_is_localhost(self): """http:// discovered endpoint allowed in dev when issuer is http://localhost.""" with patch("socket.getaddrinfo", return_value=self._LOOPBACK_ADDR): validate_discovered_endpoint( "http://localhost:8080/token", self._issuer("http://localhost:8080"), allow_http=True, trusted_endpoint_hosts=frozenset(), ) def test_rejects_link_local_ip(self): """Endpoint resolving to a link-local IP is rejected, with no dead-end hint.""" with ( patch("socket.getaddrinfo", return_value=self._PRIVATE_ADDR), pytest.raises(OIDCError, match="link-local.*169.254.169.254") as exc_info, ): validate_discovered_endpoint( "https://idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) assert "allow_private_network" not in str(exc_info.value) def test_rejects_private_ip(self): """Endpoint resolving to a genuinely private IP is rejected — and IS hinted. ``_PRIVATE_ADDR`` is 169.254.169.254, which is link-local rather than private, so this case covers what the name promises: an RFC 1918 address the operator CAN reach by enabling the opt-in, and is told so. """ with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.7", 0))]), pytest.raises(OIDCError, match="non-public address.*10.0.0.7") as exc_info, ): validate_discovered_endpoint( "https://idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) assert "allow_private_network" in str(exc_info.value) def test_rejects_embedded_credentials(self): """Endpoint with userinfo (user:pass@host) rejected.""" with pytest.raises(OIDCError, match="embedded credentials"): validate_discovered_endpoint( "https://user:pass@idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_different_host(self): """Endpoint on a different host than the issuer is rejected.""" with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), pytest.raises(OIDCError, match="not trusted"), ): validate_discovered_endpoint( "https://attacker.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_subdomain(self): """Sibling-subdomain endpoint is rejected (strict equality).""" with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), pytest.raises(OIDCError, match="not trusted"), ): validate_discovered_endpoint( "https://login.idp.example.com/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_different_port(self): """Endpoint on a different port than the issuer is rejected.""" with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), pytest.raises(OIDCError, match="port.*does not match issuer"), ): validate_discovered_endpoint( "https://idp.example.com:9443/token", self._issuer(), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_different_scheme(self): """Endpoint scheme must match the issuer's scheme.""" with ( patch("socket.getaddrinfo", return_value=self._LOOPBACK_ADDR), pytest.raises(OIDCError, match="scheme.*does not match issuer"), ): validate_discovered_endpoint( "https://localhost:8080/token", self._issuer("http://localhost:8080"), allow_http=True, trusted_endpoint_hosts=frozenset(), ) def test_accepts_google_known_endpoints(self): """Issuer accounts.google.com accepts the four well-known multi-origin hosts.""" google_endpoints = ( "https://accounts.google.com/o/oauth2/v2/auth", "https://oauth2.googleapis.com/token", "https://www.googleapis.com/oauth2/v3/certs", "https://openidconnect.googleapis.com/v1/userinfo", ) with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR): for endpoint in google_endpoints: validate_discovered_endpoint( endpoint, self._issuer("https://accounts.google.com"), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_rejects_unknown_host_for_known_issuer(self): """Even with a known issuer, foreign endpoints outside the allow-map are rejected.""" with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), pytest.raises(OIDCError, match="not trusted"), ): validate_discovered_endpoint( "https://attacker.com/token", self._issuer("https://accounts.google.com"), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_accepts_operator_trusted_endpoint_host(self): """Operator-supplied trusted_endpoint_hosts permits cross-host endpoints.""" with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR): validate_discovered_endpoint( "https://idp-token.example.net/token", self._issuer("https://idp.example.com"), allow_http=False, trusted_endpoint_hosts=frozenset({"idp-token.example.net"}), ) def test_accepts_explicit_default_port_match(self): """Issuer omits :443; endpoint includes :443 explicitly -> still same origin.""" with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR): validate_discovered_endpoint( "https://idp.example.com:443/token", self._issuer("https://idp.example.com"), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_accepts_implicit_default_port_match_reverse(self): """Issuer includes :443; endpoint omits the port -> still same origin.""" with patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR): validate_discovered_endpoint( "https://idp.example.com/token", self._issuer("https://idp.example.com:443"), allow_http=False, trusted_endpoint_hosts=frozenset(), ) def test_discover_rejects_endpoint_on_other_host(self): """discover_oidc returns enabled=False when token_endpoint targets a foreign host.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://attacker.com/token", "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_rejects_foreign_userinfo(self): """A foreign userinfo_endpoint is rejected even when other endpoints are same-origin. The required-endpoint loop runs first, so this test pins the userinfo branch's reject path and proves it executes. """ config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://idp.example.com/token", "userinfo_endpoint": "https://attacker.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_accepts_google_multi_origin(self): """discover_oidc accepts Google's legitimate multi-origin discovery doc.""" config = _make_config( issuer="https://accounts.google.com", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://accounts.google.com/o/oauth2/v2/auth", "token_endpoint": "https://oauth2.googleapis.com/token", "userinfo_endpoint": "https://openidconnect.googleapis.com/v1/userinfo", "jwks_uri": "https://www.googleapis.com/oauth2/v3/certs", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is True assert result.token_endpoint == "https://oauth2.googleapis.com/token" assert result.jwks_uri == "https://www.googleapis.com/oauth2/v3/certs" asyncio.run(_run()) def test_discover_accepts_entra_userinfo_on_graph(self): """discover_oidc accepts Entra's cross-host userinfo on graph.microsoft.com via the built-in allow-list, so Azure AD OIDC works out of the box with no trusted_endpoint_hosts override.""" tenant = "11111111-1111-1111-1111-111111111111" config = _make_config( issuer=f"https://login.microsoftonline.com/{tenant}/v2.0", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/authorize", "token_endpoint": f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token", "userinfo_endpoint": "https://graph.microsoft.com/oidc/userinfo", "jwks_uri": f"https://login.microsoftonline.com/{tenant}/discovery/v2.0/keys", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is True assert result.userinfo_endpoint == "https://graph.microsoft.com/oidc/userinfo" asyncio.run(_run()) def test_discover_rejects_http_endpoint(self): """discover_oidc returns enabled=False when an endpoint is http:// in prod.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "http://idp.example.com/token", "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_rejects_endpoint_resolving_to_private_ip(self): """discover_oidc returns enabled=False when an endpoint host resolves privately. DNS may legitimately rotate between the issuer check and the per-endpoint re-resolution; defence-in-depth requires we re-validate each discovered URL. """ config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://idp.example.com/token", "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() # Issuer validation passes (public), then DNS rotates so each discovered # endpoint resolves to a link-local address. results = iter( [ self._PUBLIC_ADDR, self._PRIVATE_ADDR, self._PRIVATE_ADDR, self._PRIVATE_ADDR, self._PRIVATE_ADDR, ] ) def _resolve(*_args, **_kwargs): return next(results) async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", side_effect=_resolve), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_accepts_localhost_http_flow(self): """Localhost issuer with http:// endpoints is accepted in dev mode.""" config = _make_config( issuer="http://localhost:8080", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "http://localhost:8080/authorize", "token_endpoint": "http://localhost:8080/token", "userinfo_endpoint": "http://localhost:8080/userinfo", "jwks_uri": "http://localhost:8080/jwks", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._LOOPBACK_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is True assert result.token_endpoint == "http://localhost:8080/token" asyncio.run(_run()) def test_discover_accepts_empty_userinfo(self): """Empty userinfo_endpoint is allowed (skipped) when other endpoints are valid.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://idp.example.com/token", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is True assert result.userinfo_endpoint == "" asyncio.run(_run()) # --------------------------------------------------------------------------- # Redirect URI Builder # --------------------------------------------------------------------------- class TestBuildOIDCRedirectURI: """Tests for ``_build_oidc_redirect_uri`` in auth.py.""" def test_build_redirect_uri_uses_redirect_base_only(self): """The redirect URI is built solely from ``redirect_base``.""" from turnstone.core.auth import _build_oidc_redirect_uri config = _make_config(redirect_base="https://example.com") result = _build_oidc_redirect_uri(config) assert result == "https://example.com/v1/api/auth/oidc/callback" # --------------------------------------------------------------------------- # PKCE # --------------------------------------------------------------------------- class TestPKCE: def test_generate_pkce_verifier_shape(self): verifier = generate_pkce_verifier() # Verifier should be URL-safe base64 assert isinstance(verifier, str) assert len(verifier) > 40 # 48 bytes -> ~64 chars def test_pkce_verifier_uniqueness(self): """Each call should produce a unique verifier.""" v1 = generate_pkce_verifier() v2 = generate_pkce_verifier() assert v1 != v2 # --------------------------------------------------------------------------- # Authorization URL # --------------------------------------------------------------------------- class TestBuildAuthorizeURL: def test_build_authorize_url_contains_required_params(self): config = _make_config() verifier = generate_pkce_verifier() url = build_authorize_url( config=config, redirect_uri="https://app.example.com/callback", state="test-state", nonce="test-nonce", code_verifier=verifier, ) assert url.startswith("https://idp.example.com/authorize?") assert "response_type=code" in url assert "client_id=my-client" in url assert "redirect_uri=" in url assert "scope=openid" in url assert "state=test-state" in url assert "nonce=test-nonce" in url assert "code_challenge=" in url assert "code_challenge_method=S256" in url def test_build_authorize_url_pkce(self): """code_challenge in URL should be correct S256 of the verifier.""" config = _make_config() verifier = generate_pkce_verifier() url = build_authorize_url( config=config, redirect_uri="https://app.example.com/callback", state="s", nonce="n", code_verifier=verifier, ) # Extract code_challenge from URL parsed = urllib.parse.urlparse(url) params = urllib.parse.parse_qs(parsed.query) actual_challenge = params["code_challenge"][0] # Compute expected challenge digest = hashlib.sha256(verifier.encode("ascii")).digest() expected = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") assert actual_challenge == expected def test_build_authorize_url_redirect_uri_encoded(self): config = _make_config() verifier = generate_pkce_verifier() redirect = "https://app.example.com/callback?extra=1" url = build_authorize_url( config=config, redirect_uri=redirect, state="s", nonce="n", code_verifier=verifier, ) # The redirect_uri should be URL-encoded parsed = urllib.parse.urlparse(url) params = urllib.parse.parse_qs(parsed.query) assert params["redirect_uri"][0] == redirect # --------------------------------------------------------------------------- # ID Token Validation # --------------------------------------------------------------------------- class TestValidateIDToken: _FAKE_JWKS = {"keys": [{"kid": "key1", "kty": "RSA", "n": "abc", "e": "AQAB"}]} def test_validate_id_token_nonce_mismatch(self): """Nonce mismatch should raise OIDCError.""" config = _make_config() mock_pyjwk = MagicMock() mock_pyjwk.return_value.key = "fake-key" with ( patch("jwt.get_unverified_header", return_value={"kid": "key1", "alg": "RS256"}), patch("jwt.PyJWK", mock_pyjwk), patch("jwt.decode", return_value={"sub": "user1", "nonce": "wrong-nonce"}), pytest.raises(OIDCError, match="nonce mismatch"), ): validate_id_token( raw_token="fake.jwt.token", jwks_data=self._FAKE_JWKS, config=config, nonce="expected-nonce", ) def test_validate_id_token_success(self): """Successful validation returns decoded claims.""" config = _make_config() mock_pyjwk = MagicMock() mock_pyjwk.return_value.key = "fake-key" expected_claims = { "sub": "user1", "email": "user@example.com", "nonce": "test-nonce", } with ( patch("jwt.get_unverified_header", return_value={"kid": "key1", "alg": "RS256"}), patch("jwt.PyJWK", mock_pyjwk), patch("jwt.decode", return_value=expected_claims) as mock_decode, ): claims = validate_id_token( raw_token="fake.jwt.token", jwks_data=self._FAKE_JWKS, config=config, nonce="test-nonce", ) assert claims == expected_claims mock_decode.assert_called_once_with( "fake.jwt.token", "fake-key", algorithms=[ "RS256", "RS384", "RS512", "ES256", "ES384", "ES512", "PS256", "PS384", "PS512", ], audience="my-client", issuer="https://idp.example.com", ) def test_validate_id_token_kid_not_found(self): """Unknown kid raises OIDCError with descriptive message.""" config = _make_config() jwks_data = {"keys": [{"kid": "other-key", "kty": "RSA"}]} with ( patch("jwt.get_unverified_header", return_value={"kid": "unknown", "alg": "RS256"}), pytest.raises(OIDCError, match="not found in JWKS"), ): validate_id_token( raw_token="bad.token", jwks_data=jwks_data, config=config, nonce="n", ) def test_validate_id_token_raises_keynotfound_for_unknown_kid(self): """Unknown kid raises the OIDCKeyNotFoundError subclass, not generic OIDCError.""" config = _make_config() jwks_data = {"keys": [{"kid": "other-key", "kty": "RSA"}]} with ( patch("jwt.get_unverified_header", return_value={"kid": "unknown", "alg": "RS256"}), pytest.raises(OIDCKeyNotFoundError) as exc_info, ): validate_id_token( raw_token="bad.token", jwks_data=jwks_data, config=config, nonce="n", ) assert isinstance(exc_info.value, OIDCError) assert "not found in JWKS" in str(exc_info.value) def test_validate_id_token_invalid_jwt(self): """Invalid JWT raises OIDCError.""" config = _make_config() mock_pyjwk = MagicMock() mock_pyjwk.return_value.key = "fake-key" with ( patch("jwt.get_unverified_header", return_value={"kid": "key1", "alg": "RS256"}), patch("jwt.PyJWK", mock_pyjwk := MagicMock(return_value=MagicMock(key="fake-key"))), patch("jwt.decode", side_effect=pyjwt.InvalidTokenError("expired")), pytest.raises(OIDCError, match="ID token validation failed"), ): validate_id_token( raw_token="expired.token", jwks_data=self._FAKE_JWKS, config=config, nonce="n", ) def test_validate_id_token_invalid_header(self): """Malformed token header raises OIDCError.""" config = _make_config() with ( patch("jwt.get_unverified_header", side_effect=pyjwt.DecodeError("bad header")), pytest.raises(OIDCError, match="Invalid ID token header"), ): validate_id_token( raw_token="garbage", jwks_data=self._FAKE_JWKS, config=config, nonce="n", ) def test_validate_id_token_retry_after_kid_rotation(self): """Sign with a fresh key, miss old JWKS, succeed against rotated JWKS. Drives the kid-not-found -> JWKS rotation -> retry path at the ``validate_id_token`` unit level. Uses real RSA signing so the decoded claims come back through pyjwt rather than from a mock. """ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from jwt.algorithms import RSAAlgorithm key = rsa.generate_private_key(public_exponent=65537, key_size=2048) priv_pem = key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) new_jwk = RSAAlgorithm.to_jwk(key.public_key(), as_dict=True) new_jwk["kid"] = "new-key" new_jwk["alg"] = "RS256" config = _make_config(issuer="https://idp.example.com") token = pyjwt.encode( { "sub": "user1", "aud": config.client_id, "iss": config.issuer, "nonce": "test-nonce", }, priv_pem, algorithm="RS256", headers={"kid": "new-key"}, ) # Stale JWKS with an unrelated old key -> kid lookup fails. old_jwks = { "keys": [{"kid": "old-key", "kty": "RSA", "n": "abc", "e": "AQAB"}], } with pytest.raises(OIDCKeyNotFoundError): validate_id_token(token, old_jwks, config, "test-nonce") # Rotated JWKS includes the new key -> validation succeeds. new_jwks = {"keys": [new_jwk]} claims = validate_id_token(token, new_jwks, config, "test-nonce") assert claims["sub"] == "user1" assert claims["nonce"] == "test-nonce" # --------------------------------------------------------------------------- # Token Exchange # --------------------------------------------------------------------------- def _mock_async_post_client(mock_post): """Build a patched httpx.AsyncClient context manager that supports POST.""" class _AsyncCtx: async def __aenter__(self): return self async def __aexit__(self, *args): pass async def post(self, url, data=None): return await mock_post(url, data) return _AsyncCtx() class TestExchangeCode: def test_exchange_code_rejects_non_dict_body(self): """A 200 response whose JSON body isn't a JSON object must be rejected.""" config = _make_config() mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = [1, 2, 3] async def _post(_url, _data): return mock_response async def _run(): client = _mock_async_post_client(_post) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="non-dict body"), ): await exchange_code(config, "code", "https://app.example.com/cb", "verifier") asyncio.run(_run()) def test_exchange_code_error_body_is_sanitized(self): """A 4xx response body containing CR/LF must not appear raw in the error message.""" config = _make_config() mock_response = MagicMock() mock_response.status_code = 400 mock_response.text = "evil\nlog injection\rline" async def _post(_url, _data): return mock_response async def _run(): client = _mock_async_post_client(_post) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError) as exc_info, ): await exchange_code(config, "code", "https://app.example.com/cb", "verifier") msg = str(exc_info.value) assert "\n" not in msg assert "\r" not in msg assert "evil" in msg assert "log injection" in msg asyncio.run(_run()) def test_exchange_code_4xx_status_in_error(self): """A 4xx response surfaces the status code in the OIDCError message.""" config = _make_config() mock_response = MagicMock() mock_response.status_code = 400 mock_response.text = "invalid_grant" async def _post(_url, _data): return mock_response async def _run(): client = _mock_async_post_client(_post) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="returned 400"), ): await exchange_code(config, "code", "https://app.example.com/cb", "verifier") asyncio.run(_run()) def test_exchange_code_5xx_status_in_error(self): """A 5xx response surfaces the status code in the OIDCError message.""" config = _make_config() mock_response = MagicMock() mock_response.status_code = 500 mock_response.text = "internal error" async def _post(_url, _data): return mock_response async def _run(): client = _mock_async_post_client(_post) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="returned 500"), ): await exchange_code(config, "code", "https://app.example.com/cb", "verifier") asyncio.run(_run()) def test_exchange_code_network_error_wraps_to_oidc_error(self): """``httpx.RequestError`` from the transport surfaces as ``OIDCError``.""" config = _make_config() async def _post(_url, _data): raise httpx.ConnectError("connection refused") async def _run(): client = _mock_async_post_client(_post) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="Token exchange request failed"), ): await exchange_code(config, "code", "https://app.example.com/cb", "verifier") asyncio.run(_run()) # --------------------------------------------------------------------------- # JWKS Fetch # --------------------------------------------------------------------------- def _mock_async_get_client(mock_get): """Build a patched httpx.AsyncClient context manager that supports GET-with-timeout.""" class _AsyncCtx: async def __aenter__(self): return self async def __aexit__(self, *args): pass async def get(self, url, timeout=None): return await mock_get(url) return _AsyncCtx() class TestFetchJWKS: """Coverage for ``fetch_jwks`` failure modes — malformed responses must surface as ``OIDCError`` rather than ``AttributeError``/``KeyError``. """ def test_fetch_jwks_non_200_raises(self): """Non-2xx response (raise_for_status fires) -> OIDCError.""" from turnstone.core.oidc import fetch_jwks mock_response = MagicMock() mock_response.raise_for_status.side_effect = httpx.HTTPStatusError( "404", request=MagicMock(), response=mock_response ) async def _get(_url): return mock_response async def _run(): client = _mock_async_get_client(_get) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="JWKS fetch failed"), ): await fetch_jwks("https://idp.example.com/jwks") asyncio.run(_run()) def test_fetch_jwks_non_dict_body_raises(self): """Body decoded as a list rather than a JSON object -> OIDCError.""" from turnstone.core.oidc import fetch_jwks mock_response = MagicMock() mock_response.json.return_value = [1, 2, 3] mock_response.raise_for_status = MagicMock() async def _get(_url): return mock_response async def _run(): client = _mock_async_get_client(_get) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="not a JSON object"), ): await fetch_jwks("https://idp.example.com/jwks") asyncio.run(_run()) def test_fetch_jwks_dict_missing_keys_raises(self): """Body is a dict but lacks the ``keys`` array -> OIDCError.""" from turnstone.core.oidc import fetch_jwks mock_response = MagicMock() mock_response.json.return_value = {"not_keys": []} mock_response.raise_for_status = MagicMock() async def _get(_url): return mock_response async def _run(): client = _mock_async_get_client(_get) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="missing 'keys' array"), ): await fetch_jwks("https://idp.example.com/jwks") asyncio.run(_run()) def test_fetch_jwks_keys_not_a_list_raises(self): """``keys`` present but not a list -> OIDCError (not iterable type).""" from turnstone.core.oidc import fetch_jwks mock_response = MagicMock() mock_response.json.return_value = {"keys": "definitely-not-a-list"} mock_response.raise_for_status = MagicMock() async def _get(_url): return mock_response async def _run(): client = _mock_async_get_client(_get) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="missing 'keys' array"), ): await fetch_jwks("https://idp.example.com/jwks") asyncio.run(_run()) def test_fetch_jwks_network_error_raises(self): """``httpx.RequestError`` from the transport -> OIDCError.""" from turnstone.core.oidc import fetch_jwks async def _get(_url): raise httpx.ConnectError("connection refused") async def _run(): client = _mock_async_get_client(_get) with ( patch("httpx.AsyncClient", return_value=client), pytest.raises(OIDCError, match="JWKS fetch failed"), ): await fetch_jwks("https://idp.example.com/jwks") asyncio.run(_run()) class TestSanitizeLogText: def test_sanitize_log_text_escapes_control_chars(self): """CR/LF, NUL, and tab characters must be escaped, not preserved.""" out = _sanitize_log_text("a\r\nb\tc\x00d", 500) assert "\r" not in out assert "\n" not in out assert "\t" not in out assert "\x00" not in out assert "a" in out and "b" in out and "c" in out and "d" in out def test_sanitize_log_text_truncates_after_escape(self): """The limit caps the rendered length, including the escape sequences.""" out = _sanitize_log_text("\n" * 100, 10) assert len(out) == 10 # --------------------------------------------------------------------------- # User Provisioning # --------------------------------------------------------------------------- class TestProvisionOIDCUser: def test_provision_oidc_user_existing(self): """Existing identity -> returns existing user, updates last_login.""" config = _make_config() existing_user = { "user_id": "u1", "username": "alice", "display_name": "Alice", "password_hash": "!oidc", } existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-123", "user_id": "u1", "email": "alice@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", } storage = _mock_storage(identity=existing_identity, user=existing_user) claims = {"sub": "sub-123", "email": "alice@example.com", "name": "Alice"} user = provision_oidc_user(storage, config, claims) assert user["user_id"] == "u1" assert user["username"] == "alice" storage.update_oidc_identity_login.assert_called_once() # Should not create a new user storage.create_oidc_user.assert_not_called() def test_provision_oidc_user_new(self): """No identity -> creates user + identity atomically.""" config = _make_config() storage = _mock_storage() # After create_oidc_user, get_user should return the new user new_user = { "user_id": "u-new", "username": "bob", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = {"sub": "sub-456", "preferred_username": "bob", "email": "bob@example.com"} with patch("turnstone.core.oidc.uuid") as mock_uuid: mock_uuid.uuid4.return_value = MagicMock(hex="u-new-hex-00000000000000000000") user = provision_oidc_user(storage, config, claims) assert user["username"] == "bob" storage.create_oidc_user.assert_called_once() # Positional args: user_id, username, display_name, password_hash, issuer, subject, email call_args = storage.create_oidc_user.call_args assert call_args[0][1] == "bob" assert call_args[0][3] == "!oidc" assert call_args[0][4] == "https://idp.example.com" assert call_args[0][5] == "sub-456" assert call_args[0][6] == "bob@example.com" def test_provision_oidc_user_username_dedup(self): """First username taken -> appends suffix.""" config = _make_config() # Bulk lookup reports "bob" already taken; "bob2" is free. storage = _mock_storage(existing_usernames={"bob"}) new_user = { "user_id": "u-new", "username": "bob2", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = {"sub": "sub-789", "preferred_username": "bob", "email": "bob@example.com"} user = provision_oidc_user(storage, config, claims) assert user["username"] == "bob2" # create_oidc_user should have been called with "bob2" as username call_args = storage.create_oidc_user.call_args assert call_args[0][1] == "bob2" # Single bulk query rather than per-candidate get_user_by_username storage.find_existing_usernames.assert_called_once() storage.get_user_by_username.assert_not_called() def test_provision_oidc_user_email_prefix(self): """No preferred_username -> uses email prefix.""" config = _make_config() storage = _mock_storage() new_user = { "user_id": "u-new", "username": "charlie", "display_name": "charlie@example.com", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = {"sub": "sub-abc", "email": "charlie@example.com"} provision_oidc_user(storage, config, claims) call_args = storage.create_oidc_user.call_args assert call_args[0][1] == "charlie" def test_provision_oidc_user_missing_user_raises(self): """Identity references missing user -> raises OIDCError.""" config = _make_config() existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-orphan", "user_id": "u-gone", "email": "gone@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", } storage = _mock_storage(identity=existing_identity, user=None) claims = {"sub": "sub-orphan", "email": "gone@example.com"} with pytest.raises(OIDCError, match="missing user"): provision_oidc_user(storage, config, claims) def test_provision_oidc_user_fallback_username(self): """No preferred_username and no email -> falls back to 'user'.""" config = _make_config() storage = _mock_storage() new_user = { "user_id": "u-new", "username": "user", "display_name": "", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = {"sub": "sub-noemail"} provision_oidc_user(storage, config, claims) call_args = storage.create_oidc_user.call_args assert call_args[0][1] == "user" def test_provision_oidc_user_username_conflict_raises(self): """Username TOCTOU -> create_oidc_user raises StorageConflictError; provision_oidc_user wraps it as OIDCError without leaving role rows. """ from turnstone.core.storage import StorageConflictError config = _make_config() storage = _mock_storage() storage.create_oidc_user.side_effect = StorageConflictError("username already taken: bob") claims = {"sub": "sub-race", "preferred_username": "bob", "email": "bob@example.com"} with pytest.raises(OIDCError, match="username already taken"): provision_oidc_user(storage, config, claims) storage.assign_role.assert_not_called() def test_provision_oidc_user_identity_conflict_raises(self): """Concurrent (issuer, subject) creates -> StorageConflictError -> OIDCError.""" from turnstone.core.storage import StorageConflictError config = _make_config() storage = _mock_storage() storage.create_oidc_user.side_effect = StorageConflictError( "OIDC identity already linked: (https://idp.example.com, sub-race)" ) claims = {"sub": "sub-race", "preferred_username": "bob", "email": "bob@example.com"} with pytest.raises(OIDCError, match="OIDC identity already linked"): provision_oidc_user(storage, config, claims) storage.assign_role.assert_not_called() def test_provision_oidc_user_null_oid_tid_collapse_to_empty(self): """A present-but-null oid/tid claim must store "" — never the string "None". `claims.get("oid", "")` returns None (not the "" default) when the key is present with a JSON null, and str(None) == "None" would slip past both the server_default and the truthy backfill guard, storing a bogus non-empty sentinel that collides across every null-emitting user. New-user path. """ config = _make_config() storage = _mock_storage() storage.get_user.return_value = { "user_id": "u-new", "username": "bob", "display_name": "Bob", "password_hash": "!oidc", } claims = {"sub": "sub-null", "preferred_username": "bob", "oid": None, "tid": None} with patch("turnstone.core.oidc.uuid") as mock_uuid: mock_uuid.uuid4.return_value = MagicMock(hex="u-new-hex-00000000000000000000") provision_oidc_user(storage, config, claims) kwargs = storage.create_oidc_user.call_args.kwargs assert kwargs["oid"] == "" assert kwargs["tid"] == "" def test_provision_oidc_user_null_oid_tid_not_backfilled_existing(self): """Existing-identity path: null oid/tid claims must not backfill "None". The truthy guard in update_oidc_identity_login only protects against ""; a "None" produced by str(None) is truthy and would be written, clobbering a real value captured on an earlier login. """ config = _make_config() existing_user = { "user_id": "u1", "username": "alice", "display_name": "Alice", "password_hash": "!oidc", } existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-123", "user_id": "u1", "email": "alice@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", "oid": "obj-real", "tid": "ten-real", } storage = _mock_storage(identity=existing_identity, user=existing_user) claims = {"sub": "sub-123", "email": "alice@example.com", "oid": None, "tid": None} provision_oidc_user(storage, config, claims) kwargs = storage.update_oidc_identity_login.call_args.kwargs assert kwargs["oid"] == "" assert kwargs["tid"] == "" def test_existing_identity_self_heals_zero_roles(self): """Existing identity user with zero roles -> safety-net assigns builtin-viewer. Models the bug-1 strand: a prior login committed user + identity but ``apply_role_mapping`` raised before reaching the safety-net. On the next login the existing-identity branch must self-heal. """ config = _make_config() existing_user = { "user_id": "u-stranded", "username": "alice", "display_name": "Alice", "password_hash": "!oidc", } existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-stranded", "user_id": "u-stranded", "email": "alice@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", } storage = _mock_storage( identity=existing_identity, user=existing_user, role={"role_id": "builtin-viewer", "name": "Viewer"}, ) storage.list_user_roles.return_value = [] claims = {"sub": "sub-stranded", "email": "alice@example.com", "name": "Alice"} user = provision_oidc_user(storage, config, claims) assert user["user_id"] == "u-stranded" storage.list_user_roles.assert_called_once_with("u-stranded") storage.assign_role.assert_called_once_with("u-stranded", "builtin-viewer", "oidc-default") def test_existing_identity_does_not_re_assign_when_user_has_roles(self): """User already has at least one role -> safety-net no-ops.""" config = _make_config() existing_user = { "user_id": "u1", "username": "alice", "display_name": "Alice", "password_hash": "!oidc", } existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-123", "user_id": "u1", "email": "alice@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", } storage = _mock_storage( identity=existing_identity, user=existing_user, role={"role_id": "builtin-viewer", "name": "Viewer"}, ) storage.list_user_roles.return_value = [{"role_id": "builtin-operator"}] claims = {"sub": "sub-123", "email": "alice@example.com"} provision_oidc_user(storage, config, claims) storage.list_user_roles.assert_called_once_with("u1") storage.assign_role.assert_not_called() def test_existing_identity_with_claim_mapped_roles_skips_default(self): """Claim-driven mapping populated roles -> hint short-circuits the helper.""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) existing_user = { "user_id": "u1", "username": "alice", "display_name": "Alice", "password_hash": "!oidc", } existing_identity = { "issuer": "https://idp.example.com", "subject": "sub-123", "user_id": "u1", "email": "alice@example.com", "created": "2024-01-01T00:00:00", "last_login": "2024-01-01T00:00:00", } storage = _mock_storage( identity=existing_identity, user=existing_user, role={"role_id": "builtin-admin", "name": "Admin"}, ) claims = {"sub": "sub-123", "groups": "admin"} provision_oidc_user(storage, config, claims) storage.list_user_roles.assert_not_called() storage.assign_role.assert_not_called() def test_new_user_safety_net_still_fires(self): """Fresh user with no IdP-mapped roles -> builtin-viewer fallback applied.""" config = _make_config() storage = _mock_storage(role={"role_id": "builtin-viewer", "name": "Viewer"}) storage.list_user_roles.return_value = [] new_user = { "user_id": "u-new", "username": "bob", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = {"sub": "sub-new", "preferred_username": "bob", "email": "bob@example.com"} provision_oidc_user(storage, config, claims) storage.assign_role.assert_called_once() called_args = storage.assign_role.call_args[0] assert called_args[1] == "builtin-viewer" assert called_args[2] == "oidc-default" def test_new_user_safety_net_skipped_when_apply_role_mapping_assigns_role(self): """Claim-driven mapping populates roles -> safety-net hint short-circuits.""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage(role={"role_id": "builtin-admin", "name": "Admin"}) new_user = { "user_id": "u-new", "username": "bob", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user claims = { "sub": "sub-new", "preferred_username": "bob", "email": "bob@example.com", "groups": "admin", } provision_oidc_user(storage, config, claims) storage.list_user_roles.assert_not_called() storage.assign_role.assert_not_called() def test_ensure_default_role_noop_when_builtin_viewer_missing(self): """builtin-viewer absent from role table -> helper does nothing.""" storage = _mock_storage(role=None) _ensure_default_role(storage, "u1") storage.get_role.assert_called_once_with("builtin-viewer") storage.list_user_roles.assert_not_called() storage.assign_role.assert_not_called() # --------------------------------------------------------------------------- # Username derivation — UUID-retry fallback tiers # --------------------------------------------------------------------------- class TestDeriveUsername: """Tier-3 (UUID-retry) and tier-4 (give-up) coverage for ``_derive_username``. Tier 1 (sanitised candidate) and tier 2 (numeric suffix dedup) are already exercised through ``TestProvisionOIDCUser``; these tests drive the post-batch-6 fallback path that runs after every numeric suffix collides. """ def _claims(self) -> dict[str, str]: return { "sub": "sub-x", "preferred_username": "bob", "email": "bob@example.com", } def _all_numeric_candidates_taken(self) -> set[str]: """The full set of tier-1 + tier-2 candidates ``_derive_username`` enumerates.""" return {"bob", *(f"bob{n}" for n in range(2, 11))} def test_derive_username_falls_into_uuid_retry_when_all_suffixes_taken(self): """All 10 numeric candidates taken -> tier 3 returns a UUID-suffixed username.""" config = _make_config() storage = _mock_storage(existing_usernames=self._all_numeric_candidates_taken()) # First UUID candidate is free. storage.get_user_by_username.return_value = None new_user = { "user_id": "u-new", "username": "ignored-by-test", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user provision_oidc_user(storage, config, self._claims()) # The username actually persisted is the tier-3 UUID candidate. called_with = storage.create_oidc_user.call_args[0][1] assert called_with.startswith("bob") # bob + 32-hex UUID hex == 35 chars total. suffix = called_with[len("bob") :] assert len(suffix) == 32 assert all(c in "0123456789abcdef" for c in suffix) # Exactly one tier-3 lookup happened (no retry collision). assert storage.get_user_by_username.call_count == 1 def test_derive_username_uuid_retry_succeeds_on_second_attempt(self): """First UUID candidate collides; second attempt is free.""" config = _make_config() storage = _mock_storage(existing_usernames=self._all_numeric_candidates_taken()) # First lookup: collision (existing user); second: free. storage.get_user_by_username.side_effect = [ {"user_id": "other", "username": "bob-already-used"}, None, ] new_user = { "user_id": "u-new", "username": "ignored-by-test", "display_name": "Bob", "password_hash": "!oidc", } storage.get_user.return_value = new_user provision_oidc_user(storage, config, self._claims()) assert storage.get_user_by_username.call_count == 2 # The username persisted is the second UUID candidate. persisted = storage.create_oidc_user.call_args[0][1] assert persisted.startswith("bob") assert len(persisted) == len("bob") + 32 def test_derive_username_uuid_retry_exhausted_raises(self): """All 3 UUID-retry attempts collide -> OIDCError raised.""" config = _make_config() storage = _mock_storage(existing_usernames=self._all_numeric_candidates_taken()) # Every UUID candidate hits an existing user. storage.get_user_by_username.return_value = {"user_id": "other", "username": "x"} with pytest.raises(OIDCError, match="Failed to generate unique username"): provision_oidc_user(storage, config, self._claims()) # Tier 3 is bounded to 3 attempts. assert storage.get_user_by_username.call_count == 3 storage.create_oidc_user.assert_not_called() # --------------------------------------------------------------------------- # Role Mapping # --------------------------------------------------------------------------- class TestApplyRoleMapping: def test_apply_role_mapping_basic(self): """Maps claim value to role.""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage(role={"role_id": "builtin-admin", "name": "Admin"}) claims = {"sub": "u1", "groups": "admin"} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", {"builtin-admin"}) storage.assign_role.assert_not_called() storage.unassign_role.assert_not_called() storage.list_user_roles.assert_not_called() def test_apply_role_mapping_list_claim(self): """Claim is a list of strings -> single replace call.""" config = _make_config( role_claim="roles", role_map={"admin": "builtin-admin", "editor": "builtin-operator"}, ) storage = _mock_storage() # get_role returns non-None for both roles storage.get_role.return_value = {"role_id": "some-role"} claims = {"sub": "u1", "roles": ["admin", "editor"]} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with( "u1", {"builtin-admin", "builtin-operator"} ) def test_apply_role_mapping_no_config(self): """No role_claim configured -> no-op.""" config = _make_config(role_claim="", role_map={}) storage = _mock_storage() claims = {"sub": "u1", "roles": "admin"} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_not_called() def test_apply_role_mapping_unknown_role(self): """Claim maps to nonexistent role -> empty replace set.""" config = _make_config( role_claim="groups", role_map={"admin": "nonexistent-role"}, ) storage = _mock_storage(role=None) # role doesn't exist claims = {"sub": "u1", "groups": "admin"} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", set()) def test_apply_role_mapping_no_matching_claim_value(self): """Claim value not in role_map -> empty replace set.""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage() claims = {"sub": "u1", "groups": "viewer"} # "viewer" not in role_map apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", set()) def test_apply_role_mapping_claim_missing(self): """Claim key not present in claims -> empty replace set.""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage() claims = {"sub": "u1"} # no "groups" key apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", set()) def test_apply_role_mapping_no_role_map(self): """role_claim set but role_map empty -> no-op (early return).""" config = _make_config(role_claim="groups", role_map={}) storage = _mock_storage() claims = {"sub": "u1", "groups": "admin"} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_not_called() def test_apply_role_mapping_revokes_stale_oidc_roles(self, caplog): """Roles previously assigned by OIDC but no longer in claims are revoked. Storage owns the diff via ``replace_oidc_roles``; ``apply_role_mapping`` only logs the returned ``(added, removed)`` sets. Manual roles are invisible to ``apply_role_mapping`` post-perf-3 — protection now lives in the storage layer's ``WHERE assigned_by = 'oidc'`` filter. """ config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin", "eng": "builtin-operator"}, ) storage = _mock_storage( replace_oidc_roles=({"builtin-operator"}, {"builtin-admin"}), ) storage.get_role.return_value = {"role_id": "some-role"} # IdP now only says "eng", not "admin" claims = {"sub": "u1", "groups": ["eng"]} with caplog.at_level("INFO", logger="turnstone.core.oidc"): apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", {"builtin-operator"}) assert any( "Revoked role" in record.getMessage() and "builtin-admin" in record.getMessage() for record in caplog.records ) def test_apply_role_mapping_revokes_all_oidc_roles_when_claim_absent(self): """Empty desired set propagates to storage (diff happens server-side).""" config = _make_config( role_claim="groups", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage( replace_oidc_roles=(set(), {"builtin-admin", "builtin-operator"}), ) storage.get_role.return_value = {"role_id": "some-role"} claims = {"sub": "u1"} # no "groups" key apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", set()) def test_apply_role_mapping_int_claim(self): """Numeric claim values stringify before role_map lookup (no crash).""" config = _make_config( role_claim="roles", role_map={"42": "builtin-viewer"}, ) storage = _mock_storage(role={"role_id": "builtin-viewer", "name": "Viewer"}) claims = {"sub": "u1", "roles": 42} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", {"builtin-viewer"}) def test_apply_role_mapping_dict_claim(self): """Dict claim values stringify but won't reliably hit role_map -> empty set.""" config = _make_config( role_claim="roles", role_map={"admin": "builtin-admin"}, ) storage = _mock_storage() # Dict reprs are unstable across runtime versions, so callers cannot # reasonably configure role_map keys to match. The contract is just # "don't crash". claims = {"sub": "u1", "roles": {"role": "admin"}} apply_role_mapping(storage, "u1", claims, config) storage.replace_oidc_roles.assert_called_once_with("u1", set()) # --------------------------------------------------------------------------- # Discovery (async) # --------------------------------------------------------------------------- class TestDiscoverOIDC: # Mock DNS result for a public IP — reused across discovery tests. _PUBLIC_ADDR = [(2, 1, 6, "", ("93.184.216.34", 0))] def test_discover_oidc_success(self): """Mock httpx response, verify endpoints populated.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://idp.example.com/token", "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.authorization_endpoint == "https://idp.example.com/authorize" assert result.token_endpoint == "https://idp.example.com/token" assert result.userinfo_endpoint == "https://idp.example.com/userinfo" assert result.jwks_uri == "https://idp.example.com/.well-known/jwks.json" assert result.enabled is True asyncio.run(_run()) def test_discover_oidc_private_issuer_rejected_by_default(self): """Without the opt-in, a private-resolving issuer disables OIDC.""" config = _make_config( issuer="https://auth.internal.example", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) async def _run(): with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_oidc_private_issuer_with_opt_in(self): """allow_private_network=True lets a private-resolving IdP discover.""" config = _make_config( issuer="https://auth.internal.example", allow_private_network=True, authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) discovery_doc = { "authorization_endpoint": "https://auth.internal.example/authorize", "token_endpoint": "https://auth.internal.example/token", "userinfo_endpoint": "https://auth.internal.example/userinfo", "jwks_uri": "https://auth.internal.example/jwks", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is True assert result.token_endpoint == "https://auth.internal.example/token" asyncio.run(_run()) def test_discover_oidc_failure(self): """Mock httpx error -> enabled=False returned.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) async def _failing_get(url): raise httpx.ConnectError("connection refused") async def _run(): client = _mock_async_client(_failing_get) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_oidc_no_issuer(self): """Empty issuer -> enabled=False.""" config = _make_config(issuer="") async def _run(): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_oidc_missing_required_endpoints(self): """Discovery doc missing authorization_endpoint -> enabled=False.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) # Document missing authorization_endpoint discovery_doc = { "token_endpoint": "https://idp.example.com/token", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): result = await discover_oidc(config) assert result.enabled is False asyncio.run(_run()) def test_discover_oidc_handles_non_dict_response(self): """IdP returning a list/null body must not raise AttributeError.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) for payload in (["not", "a", "dict"], None, "string-body", 42): mock_response = MagicMock() mock_response.json.return_value = payload mock_response.raise_for_status = MagicMock() async def _run(resp=mock_response): client = _mock_async_client(lambda url: _async_return(resp)) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): return await discover_oidc(config) result = asyncio.run(_run()) assert result.enabled is False # --------------------------------------------------------------------------- # Runtime re-discovery (boot-time transient failure self-heal) # --------------------------------------------------------------------------- class TestRuntimeRediscovery: """A node that boots during a transient IdP outage keeps enabled=False forever without a runtime retry — OIDC login stays dark and every oauth_obo mint on the node fails "transient" until an operator restarts it. ``maybe_rediscover_oidc`` heals that, cooldown-gated; config-caused discovery failures (bad issuer, SSRF rejection) are NOT retried.""" _PUBLIC_ADDR = [(2, 1, 6, "", ("93.184.216.34", 0))] def test_fetch_failure_marks_config_retryable(self): """The transient branch (IdP unreachable) sets discovery_retryable.""" config = _make_config( authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="" ) async def _raise_connect(url): raise httpx.ConnectError("boom") async def _run(): client = _mock_async_client(_raise_connect) with ( patch("socket.getaddrinfo", return_value=self._PUBLIC_ADDR), patch("httpx.AsyncClient", return_value=client), ): return await discover_oidc(config) result = asyncio.run(_run()) assert result.enabled is False assert result.discovery_retryable is True def test_config_rejection_is_not_retryable(self): """An SSRF-rejected issuer is a config problem — retrying is pointless.""" config = _make_config( issuer="https://auth.internal.example", authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) async def _run(): with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]): return await discover_oidc(config) result = asyncio.run(_run()) assert result.enabled is False assert result.discovery_retryable is False def _disabled_retryable_state(self) -> SimpleNamespace: cfg = _make_config( enabled=False, authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) cfg = dataclasses.replace(cfg, discovery_retryable=True) return SimpleNamespace(oidc_config=cfg) def test_rediscover_swaps_enabled_config_on_success(self): """Drives the REAL discover_oidc through a mocked HTTP discovery GET (NOT a mock of discover_oidc itself): discover_oidc preserves the input config's ``enabled`` on success and only clears it on failure, so a probe started from the disabled boot config must first force enabled=True or the recovered config never installs. An earlier version of this test mocked discover_oidc to return enabled=True and so masked exactly that dead-code bug.""" from turnstone.core.oidc import maybe_rediscover_oidc state = self._disabled_retryable_state() # issuer=https://idp.example.com discovery_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://idp.example.com/token", "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = discovery_doc mock_response.raise_for_status = MagicMock() async def _run(): client = _mock_async_client(lambda url: _async_return(mock_response)) with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]), patch("httpx.AsyncClient", return_value=client), ): await maybe_rediscover_oidc(state) asyncio.run(_run()) # The real discover_oidc succeeded and the recovered config was installed. assert state.oidc_config.enabled is True assert state.oidc_config.token_endpoint == "https://idp.example.com/token" # The healed config no longer advertises a retryable failure. assert state.oidc_config.discovery_retryable is False def test_rediscover_latches_terminal_on_config_error_and_stops_probing(self): """Review finding: probing with enabled forced True carries the retryable boot flag into discover_oidc, whose config-error branches must latch discovery_retryable=False (terminal) — and maybe_rediscover must INSTALL that terminal config — or a config-invalid IdP (endpoint failing SSRF/same-origin) re-probes every cooldown window forever. Drives the real discover_oidc: the discovered token_endpoint is on a foreign host, so validation rejects it as a config error.""" from turnstone.core.oidc import maybe_rediscover_oidc state = self._disabled_retryable_state() # issuer=https://idp.example.com bad_doc = { "authorization_endpoint": "https://idp.example.com/authorize", "token_endpoint": "https://attacker.example/token", # foreign host "userinfo_endpoint": "https://idp.example.com/userinfo", "jwks_uri": "https://idp.example.com/.well-known/jwks.json", } mock_response = MagicMock() mock_response.json.return_value = bad_doc mock_response.raise_for_status = MagicMock() probes = {"n": 0} async def _get(url): probes["n"] += 1 return mock_response async def _run(): client = _mock_async_client(_get) with ( patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]), patch("httpx.AsyncClient", return_value=client), ): await maybe_rediscover_oidc(state) # Same window: cooldown already gates a second probe. await maybe_rediscover_oidc(state) # Force the cooldown open — but the config is now terminal, so # the retryable guard should short-circuit before any probe. state.oidc_rediscover_last = None await maybe_rediscover_oidc(state) asyncio.run(_run()) # Still disabled, but LATCHED terminal (not retryable) — one probe only. assert state.oidc_config.enabled is False assert state.oidc_config.discovery_retryable is False assert probes["n"] == 1 def test_rediscover_cooldown_gates_repeat_probes(self): from turnstone.core.oidc import maybe_rediscover_oidc state = self._disabled_retryable_state() still_down = state.oidc_config # discover keeps returning disabled with patch( "turnstone.core.oidc.discover_oidc", new=AsyncMock(return_value=still_down) ) as disc: asyncio.run(maybe_rediscover_oidc(state)) asyncio.run(maybe_rediscover_oidc(state)) asyncio.run(maybe_rediscover_oidc(state)) # One IdP probe per cooldown window, however many callers ask. disc.assert_awaited_once() assert state.oidc_config.enabled is False def test_rediscover_noop_when_not_retryable_or_enabled(self): from turnstone.core.oidc import maybe_rediscover_oidc # Operator-disabled (retryable False): never probes. state = SimpleNamespace( oidc_config=_make_config( enabled=False, authorization_endpoint="", token_endpoint="", userinfo_endpoint="", jwks_uri="", ) ) with patch("turnstone.core.oidc.discover_oidc", new=AsyncMock()) as disc: asyncio.run(maybe_rediscover_oidc(state)) disc.assert_not_awaited() # Already enabled: never probes. state2 = SimpleNamespace(oidc_config=_make_config(enabled=True)) with patch("turnstone.core.oidc.discover_oidc", new=AsyncMock()) as disc2: asyncio.run(maybe_rediscover_oidc(state2)) disc2.assert_not_awaited() # --------------------------------------------------------------------------- # Lifespan integration: initialize_oidc_state # --------------------------------------------------------------------------- class TestInitializeOIDCState: def test_initialize_skips_when_disabled(self): """Disabled config: jwks_data set to None, oidc_config unchanged.""" cfg = _make_config(enabled=False) state = types.SimpleNamespace(oidc_config=cfg, jwks_data="stale") asyncio.run(initialize_oidc_state(state)) assert state.oidc_config is cfg assert state.jwks_data is None def test_initialize_disables_on_discovery_exception(self): """Unexpected exception in discovery -> enabled flipped to False.""" cfg = _make_config( authorization_endpoint="", token_endpoint="", jwks_uri="", ) state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) async def _boom(_cfg): raise RuntimeError("boom") with patch("turnstone.core.oidc.discover_oidc", side_effect=_boom): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config.enabled is False assert state.jwks_data is None def test_initialize_disables_on_discovery_returning_disabled(self): """Discovery returns enabled=False (e.g. SSRF reject) -> propagate.""" cfg = _make_config( authorization_endpoint="", token_endpoint="", jwks_uri="", ) state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) disabled_cfg = dataclasses.replace(cfg, enabled=False) async def _disabled(_cfg, *, client=None): return disabled_cfg with patch("turnstone.core.oidc.discover_oidc", side_effect=_disabled): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config is disabled_cfg assert state.oidc_config.enabled is False assert state.jwks_data is None def test_initialize_disables_when_redirect_base_unset(self, caplog): """Discovery succeeds but redirect_base is empty -> disable + log error.""" cfg = _make_config(redirect_base="") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) async def _ok(c, *, client=None): return c async def _jwks_unexpected(_uri, *, client=None): raise AssertionError("fetch_jwks must not be called when redirect_base is empty") with ( patch("turnstone.core.oidc.discover_oidc", side_effect=_ok), patch("turnstone.core.oidc.fetch_jwks", side_effect=_jwks_unexpected), caplog.at_level("ERROR", logger="turnstone.core.oidc"), ): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config.enabled is False assert state.jwks_data is None assert any("TURNSTONE_OIDC_REDIRECT_BASE" in record.message for record in caplog.records) def test_initialize_keeps_enabled_but_no_jwks_on_jwks_failure(self): """JWKS fetch failure preserves enabled=True for lazy retry.""" cfg = _make_config(redirect_base="https://app.example.com") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) async def _ok(c, *, client=None): return c async def _jwks_boom(_uri, *, client=None): raise OIDCError("jwks down") with ( patch("turnstone.core.oidc.discover_oidc", side_effect=_ok), patch("turnstone.core.oidc.fetch_jwks", side_effect=_jwks_boom), ): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config.enabled is True assert state.oidc_config is cfg assert state.jwks_data is None def test_initialize_success(self): """Both discovery and JWKS prefetch succeed.""" cfg = _make_config(redirect_base="https://app.example.com") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) jwks = {"keys": [{"kid": "k1", "kty": "RSA"}]} async def _ok(c, *, client=None): return c async def _jwks(_uri, *, client=None): return jwks with ( patch("turnstone.core.oidc.discover_oidc", side_effect=_ok), patch("turnstone.core.oidc.fetch_jwks", side_effect=_jwks), ): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config is cfg assert state.oidc_config.enabled is True assert state.jwks_data == jwks assert state.oidc_http_client is not None async def _async_return(value): """Helper: return a value from an async function.""" return value class TestCloseOIDCState: def test_close_when_never_initialised(self): """close_oidc_state on bare state must not raise.""" from turnstone.core.oidc import close_oidc_state state = types.SimpleNamespace() asyncio.run(close_oidc_state(state)) def test_close_releases_long_lived_client(self): """The long-lived client installed by initialize_oidc_state is aclosed.""" from turnstone.core.oidc import close_oidc_state client = MagicMock() async def _aclose(): client.aclose_called = True client.aclose = _aclose state = types.SimpleNamespace(oidc_http_client=client) asyncio.run(close_oidc_state(state)) assert getattr(client, "aclose_called", False) is True assert state.oidc_http_client is None class TestLongLivedHTTPClientPassthrough: def test_initialize_passes_long_lived_client_to_jwks_only(self): """Discovery uses a transient client; JWKS uses the long-lived one. Splitting the two avoids leaking the long-lived client when a disable check (discovery exception, discovery-returned-disabled, missing redirect_base) returns before JWKS prefetch — see the post-condition contract on initialize_oidc_state. """ cfg = _make_config(redirect_base="https://app.example.com") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) seen_clients: list[Any] = [] async def _discover(c, *, client=None): seen_clients.append(("discover", client)) return c async def _jwks(_uri, *, client=None): seen_clients.append(("jwks", client)) return {"keys": [{"kid": "k1"}]} with ( patch("turnstone.core.oidc.discover_oidc", side_effect=_discover), patch("turnstone.core.oidc.fetch_jwks", side_effect=_jwks), ): asyncio.run(initialize_oidc_state(state)) assert state.oidc_http_client is not None kinds = {kind for kind, _ in seen_clients} assert {"discover", "jwks"} <= kinds jwks_clients = [c for kind, c in seen_clients if kind == "jwks"] assert all(c is state.oidc_http_client for c in jwks_clients) # Discovery client is a separate transient (already closed). discover_clients = [c for kind, c in seen_clients if kind == "discover"] assert all(c is not state.oidc_http_client for c in discover_clients) def test_initialize_does_not_leak_client_on_discovery_failure(self): """Discovery exception path must close the transient and leave http_client None.""" cfg = _make_config(redirect_base="https://app.example.com") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) async def _boom(_cfg, *, client=None): raise RuntimeError("boom") with patch("turnstone.core.oidc.discover_oidc", side_effect=_boom): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config.enabled is False assert state.oidc_http_client is None assert state.jwks_data is None def test_initialize_does_not_leak_client_on_missing_redirect_base(self): """Missing redirect_base path returns before creating the long-lived client.""" cfg = _make_config(redirect_base="") state = types.SimpleNamespace(oidc_config=cfg, jwks_data=None) async def _discover(c, *, client=None): return c with patch("turnstone.core.oidc.discover_oidc", side_effect=_discover): asyncio.run(initialize_oidc_state(state)) assert state.oidc_config.enabled is False assert state.oidc_http_client is None assert state.jwks_data is None