Loading tests/fixtures/realm-test.json 0 → 100644 +63 −0 Changes for tests/fixtures/realm-test.json: 63 added lines, 0 removed lines. Original line number Diff line number Diff line { "realm": "oop-test", "enabled": true, "sslRequired": "none", "registrationAllowed": true, "accessTokenLifespan": 60, "roles": { "realm": [ { "name": "app-provider", "description": "Developers" }, { "name": "operator", "description": "People who run the platform" }, { "name": "platform-admin", "description": "User administration" } ] }, "clients": [ { "clientId": "oop-portal", "name": "OOP Portal", "enabled": true, "protocol": "openid-connect", "publicClient": false, "clientAuthenticatorType": "client-secret", "secret": "test-client-secret", "standardFlowEnabled": true, "implicitFlowEnabled": false, "directAccessGrantsEnabled": false, "serviceAccountsEnabled": false, "redirectUris": ["http://localhost:8000/*"], "webOrigins": ["http://localhost:8000"], "attributes": { "pkce.code.challenge.method": "S256", "post.logout.redirect.uris": "http://localhost:8000/*" }, "protocolMappers": [ { "name": "audience-self", "protocol": "openid-connect", "protocolMapper": "oidc-audience-mapper", "consentRequired": false, "config": { "included.client.audience": "oop-portal", "access.token.claim": "true", "introspection.token.claim": "true", "id.token.claim": "false" } } ] } ], "users": [ { "username": "test-user", "enabled": true, "email": "test-user@example.com", "emailVerified": true, "firstName": "Test", "lastName": "User", "credentials": [ { "type": "password", "value": "test-password", "temporary": false } ], "realmRoles": ["default-roles-oop-test", "app-provider", "platform-admin"] } ] } tests/integration/conftest.py +48 −0 Changes for tests/integration/conftest.py: 48 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -6,8 +6,11 @@ transaction that is rolled back afterwards, so tests cannot affect each other. """ import asyncio import time from collections.abc import AsyncIterator, Generator from pathlib import Path import httpx import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine Loading Loading @@ -74,3 +77,48 @@ async def db_session(postgres_url: str) -> AsyncIterator[AsyncSession]: await transaction.rollback() finally: await engine.dispose() KEYCLOAK_IMAGE = "quay.io/keycloak/keycloak:26.7.3" TEST_REALM = "oop-test" REALM_FILE = Path(__file__).parent.parent / "fixtures" / "realm-test.json" def _wait_for_realm(issuer: str, attempts: int = 120) -> None: """Keycloak answers on the port before the realm is imported; wait for the realm.""" for remaining in range(attempts, 0, -1): try: if ( httpx.get(f"{issuer}/.well-known/openid-configuration", timeout=2).status_code == 200 ): return except httpx.HTTPError: pass if remaining == 1: raise TimeoutError(f"realm never appeared at {issuer}") time.sleep(1) @pytest.fixture(scope="session") def keycloak_issuer() -> Generator[str, None, None]: """ A real Keycloak with a small realm of our own, so the flow can be driven end to end. The realm mirrors the deployed one where it matters: a confidential client, PKCE required, self-registration on, and the audience mapper our verification depends on. """ container = ( DockerContainer(KEYCLOAK_IMAGE) .with_command("start-dev --import-realm") .with_env("KC_BOOTSTRAP_ADMIN_USERNAME", "admin") .with_env("KC_BOOTSTRAP_ADMIN_PASSWORD", "admin") .with_volume_mapping(str(REALM_FILE), "/opt/keycloak/data/import/realm.json", "ro") .with_exposed_ports(8080) ) with container: host = container.get_container_host_ip() port = container.get_exposed_port(8080) issuer = f"http://{host}:{port}/realms/{TEST_REALM}" _wait_for_realm(issuer) yield issuer tests/integration/test_keycloak_flow.py 0 → 100644 +211 −0 Changes for tests/integration/test_keycloak_flow.py: 211 added lines, 0 removed lines. Original line number Diff line number Diff line """ The login flow against a real Keycloak. The unit tests fake the network, so they prove what we send but not that Keycloak accepts it. These drive the whole flow against Keycloak 26.7.3 in a container: the login page, the code, the exchange with PKCE and the client secret, the tokens, a refresh, and logout. No browser is involved; the test submits the login form itself. """ import html import re from collections.abc import Iterator from urllib.parse import parse_qs, urlparse import httpx import pytest from portal_service import config as config_module from portal_service.adapters.keycloak.oidc_client import KeycloakOIDCClient from portal_service.api.auth.verification import _decode_id_token, _decode_token from portal_service.application.pkce import code_challenge, new_code_verifier, new_nonce, new_state from portal_service.config import OIDCSettings from portal_service.domain.errors import IdentityProviderUnavailableError, InvalidGrantError pytestmark = pytest.mark.integration CLIENT_ID = "oop-portal" CLIENT_SECRET = "test-client-secret" REDIRECT_URI = "http://localhost:8000/api/auth/callback" USERNAME = "test-user" PASSWORD = "test-password" _FORM_ACTION = re.compile(r'action="([^"]+)"') @pytest.fixture(scope="module") def settings(keycloak_issuer: str) -> OIDCSettings: return OIDCSettings( issuer=keycloak_issuer, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, redirect_uri=REDIRECT_URI, ) @pytest.fixture def client(settings: OIDCSettings) -> KeycloakOIDCClient: return KeycloakOIDCClient(settings) @pytest.fixture def portal_settings_env(settings: OIDCSettings, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: """Point the app's own settings at the container, so verification uses its keys.""" monkeypatch.setenv("OIDC__ISSUER", settings.issuer) monkeypatch.setenv("OIDC__CLIENT_ID", CLIENT_ID) monkeypatch.setenv("OIDC__CLIENT_SECRET", CLIENT_SECRET) monkeypatch.setenv("OIDC__REDIRECT_URI", REDIRECT_URI) config_module.get_settings.cache_clear() yield config_module.get_settings.cache_clear() def _cookies_from(response: httpx.Response) -> str: """ Keycloak marks its session cookies Secure and SameSite=None. A browser on HTTPS sends them back; an HTTP client over plain http will not, so the login POST would arrive without a session ("cookie_not_found"). The test carries them by hand. """ jar: dict[str, str] = {} for hop in [*response.history, response]: for header in hop.headers.get_list("set-cookie"): name, _, rest = header.partition("=") jar[name.strip()] = rest.split(";", 1)[0] return "; ".join(f"{name}={value}" for name, value in jar.items()) def _log_in(authorization_url: str) -> str: """Do what a browser would: load the login page, submit it, read the code.""" with httpx.Client(follow_redirects=True, timeout=15) as browser: page = browser.get(authorization_url) page.raise_for_status() match = _FORM_ACTION.search(page.text) assert match, "Keycloak did not return a login form" submitted = browser.post( html.unescape(match.group(1)), data={"username": USERNAME, "password": PASSWORD}, headers={"Cookie": _cookies_from(page)}, follow_redirects=False, ) assert submitted.status_code in (302, 303), submitted.text[:400] query = parse_qs(urlparse(submitted.headers["location"]).query) assert "code" in query, f"no code in {submitted.headers['location']}" return query["code"][0] async def test_a_user_can_log_in_and_we_can_read_the_tokens( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() nonce = new_nonce() url = client.authorization_url( state=new_state(), nonce=nonce, code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.expires_in > 0 assert issued.tokens.refresh_token is not None assert issued.tokens.id_token is not None # Our own verification, against Keycloak's real signing keys. id_claims = _decode_id_token(issued.tokens.id_token, expected_nonce=nonce) assert id_claims.preferred_username == USERNAME access_claims = _decode_token(issued.tokens.access_token) assert "app-provider" in access_claims.roles assert "platform-admin" in access_claims.roles async def test_a_code_cannot_be_used_twice( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) code = _log_in(url) await client.exchange_code(code=code, code_verifier=verifier) # Keycloak's own invalid_grant, not one we invented in a mock. with pytest.raises(InvalidGrantError): await client.exchange_code(code=code, code_verifier=verifier) async def test_the_wrong_pkce_verifier_is_refused( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(new_code_verifier()) ) code = _log_in(url) with pytest.raises(InvalidGrantError): await client.exchange_code(code=code, code_verifier=new_code_verifier()) async def test_a_wrong_client_secret_is_not_the_users_fault( settings: OIDCSettings, portal_settings_env: None ) -> None: good = KeycloakOIDCClient(settings) verifier = new_code_verifier() url = good.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) code = _log_in(url) wrong = KeycloakOIDCClient(settings.model_copy(update={"client_secret": "wrong"})) with pytest.raises(IdentityProviderUnavailableError): await wrong.exchange_code(code=code, code_verifier=verifier) async def test_a_session_can_be_refreshed( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.tokens.refresh_token is not None refreshed = await client.refresh(issued.tokens.refresh_token) assert _decode_token(refreshed.tokens.access_token).preferred_username == USERNAME async def test_logging_out_ends_the_keycloak_session( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.tokens.id_token is not None and issued.tokens.refresh_token is not None logout_url = client.logout_url( id_token=issued.tokens.id_token, post_logout_redirect_uri="http://localhost:8000/" ) with httpx.Client(follow_redirects=False, timeout=15) as browser: # Not followed: Keycloak redirects to the portal, which is not running here. ended = browser.get(logout_url) assert ended.status_code in (200, 302, 303) # The session is gone at Keycloak, so its refresh token is worthless. with pytest.raises(InvalidGrantError): await client.refresh(issued.tokens.refresh_token) def test_the_signup_page_is_available(client: KeycloakOIDCClient) -> None: url = client.registration_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(new_code_verifier()) ) with httpx.Client(follow_redirects=True, timeout=15) as browser: page = browser.get(url) assert page.status_code == 200 assert "password" in page.text.lower() Loading
tests/fixtures/realm-test.json 0 → 100644 +63 −0 Changes for tests/fixtures/realm-test.json: 63 added lines, 0 removed lines. Original line number Diff line number Diff line { "realm": "oop-test", "enabled": true, "sslRequired": "none", "registrationAllowed": true, "accessTokenLifespan": 60, "roles": { "realm": [ { "name": "app-provider", "description": "Developers" }, { "name": "operator", "description": "People who run the platform" }, { "name": "platform-admin", "description": "User administration" } ] }, "clients": [ { "clientId": "oop-portal", "name": "OOP Portal", "enabled": true, "protocol": "openid-connect", "publicClient": false, "clientAuthenticatorType": "client-secret", "secret": "test-client-secret", "standardFlowEnabled": true, "implicitFlowEnabled": false, "directAccessGrantsEnabled": false, "serviceAccountsEnabled": false, "redirectUris": ["http://localhost:8000/*"], "webOrigins": ["http://localhost:8000"], "attributes": { "pkce.code.challenge.method": "S256", "post.logout.redirect.uris": "http://localhost:8000/*" }, "protocolMappers": [ { "name": "audience-self", "protocol": "openid-connect", "protocolMapper": "oidc-audience-mapper", "consentRequired": false, "config": { "included.client.audience": "oop-portal", "access.token.claim": "true", "introspection.token.claim": "true", "id.token.claim": "false" } } ] } ], "users": [ { "username": "test-user", "enabled": true, "email": "test-user@example.com", "emailVerified": true, "firstName": "Test", "lastName": "User", "credentials": [ { "type": "password", "value": "test-password", "temporary": false } ], "realmRoles": ["default-roles-oop-test", "app-provider", "platform-admin"] } ] }
tests/integration/conftest.py +48 −0 Changes for tests/integration/conftest.py: 48 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -6,8 +6,11 @@ transaction that is rolled back afterwards, so tests cannot affect each other. """ import asyncio import time from collections.abc import AsyncIterator, Generator from pathlib import Path import httpx import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine Loading Loading @@ -74,3 +77,48 @@ async def db_session(postgres_url: str) -> AsyncIterator[AsyncSession]: await transaction.rollback() finally: await engine.dispose() KEYCLOAK_IMAGE = "quay.io/keycloak/keycloak:26.7.3" TEST_REALM = "oop-test" REALM_FILE = Path(__file__).parent.parent / "fixtures" / "realm-test.json" def _wait_for_realm(issuer: str, attempts: int = 120) -> None: """Keycloak answers on the port before the realm is imported; wait for the realm.""" for remaining in range(attempts, 0, -1): try: if ( httpx.get(f"{issuer}/.well-known/openid-configuration", timeout=2).status_code == 200 ): return except httpx.HTTPError: pass if remaining == 1: raise TimeoutError(f"realm never appeared at {issuer}") time.sleep(1) @pytest.fixture(scope="session") def keycloak_issuer() -> Generator[str, None, None]: """ A real Keycloak with a small realm of our own, so the flow can be driven end to end. The realm mirrors the deployed one where it matters: a confidential client, PKCE required, self-registration on, and the audience mapper our verification depends on. """ container = ( DockerContainer(KEYCLOAK_IMAGE) .with_command("start-dev --import-realm") .with_env("KC_BOOTSTRAP_ADMIN_USERNAME", "admin") .with_env("KC_BOOTSTRAP_ADMIN_PASSWORD", "admin") .with_volume_mapping(str(REALM_FILE), "/opt/keycloak/data/import/realm.json", "ro") .with_exposed_ports(8080) ) with container: host = container.get_container_host_ip() port = container.get_exposed_port(8080) issuer = f"http://{host}:{port}/realms/{TEST_REALM}" _wait_for_realm(issuer) yield issuer
tests/integration/test_keycloak_flow.py 0 → 100644 +211 −0 Changes for tests/integration/test_keycloak_flow.py: 211 added lines, 0 removed lines. Original line number Diff line number Diff line """ The login flow against a real Keycloak. The unit tests fake the network, so they prove what we send but not that Keycloak accepts it. These drive the whole flow against Keycloak 26.7.3 in a container: the login page, the code, the exchange with PKCE and the client secret, the tokens, a refresh, and logout. No browser is involved; the test submits the login form itself. """ import html import re from collections.abc import Iterator from urllib.parse import parse_qs, urlparse import httpx import pytest from portal_service import config as config_module from portal_service.adapters.keycloak.oidc_client import KeycloakOIDCClient from portal_service.api.auth.verification import _decode_id_token, _decode_token from portal_service.application.pkce import code_challenge, new_code_verifier, new_nonce, new_state from portal_service.config import OIDCSettings from portal_service.domain.errors import IdentityProviderUnavailableError, InvalidGrantError pytestmark = pytest.mark.integration CLIENT_ID = "oop-portal" CLIENT_SECRET = "test-client-secret" REDIRECT_URI = "http://localhost:8000/api/auth/callback" USERNAME = "test-user" PASSWORD = "test-password" _FORM_ACTION = re.compile(r'action="([^"]+)"') @pytest.fixture(scope="module") def settings(keycloak_issuer: str) -> OIDCSettings: return OIDCSettings( issuer=keycloak_issuer, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, redirect_uri=REDIRECT_URI, ) @pytest.fixture def client(settings: OIDCSettings) -> KeycloakOIDCClient: return KeycloakOIDCClient(settings) @pytest.fixture def portal_settings_env(settings: OIDCSettings, monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: """Point the app's own settings at the container, so verification uses its keys.""" monkeypatch.setenv("OIDC__ISSUER", settings.issuer) monkeypatch.setenv("OIDC__CLIENT_ID", CLIENT_ID) monkeypatch.setenv("OIDC__CLIENT_SECRET", CLIENT_SECRET) monkeypatch.setenv("OIDC__REDIRECT_URI", REDIRECT_URI) config_module.get_settings.cache_clear() yield config_module.get_settings.cache_clear() def _cookies_from(response: httpx.Response) -> str: """ Keycloak marks its session cookies Secure and SameSite=None. A browser on HTTPS sends them back; an HTTP client over plain http will not, so the login POST would arrive without a session ("cookie_not_found"). The test carries them by hand. """ jar: dict[str, str] = {} for hop in [*response.history, response]: for header in hop.headers.get_list("set-cookie"): name, _, rest = header.partition("=") jar[name.strip()] = rest.split(";", 1)[0] return "; ".join(f"{name}={value}" for name, value in jar.items()) def _log_in(authorization_url: str) -> str: """Do what a browser would: load the login page, submit it, read the code.""" with httpx.Client(follow_redirects=True, timeout=15) as browser: page = browser.get(authorization_url) page.raise_for_status() match = _FORM_ACTION.search(page.text) assert match, "Keycloak did not return a login form" submitted = browser.post( html.unescape(match.group(1)), data={"username": USERNAME, "password": PASSWORD}, headers={"Cookie": _cookies_from(page)}, follow_redirects=False, ) assert submitted.status_code in (302, 303), submitted.text[:400] query = parse_qs(urlparse(submitted.headers["location"]).query) assert "code" in query, f"no code in {submitted.headers['location']}" return query["code"][0] async def test_a_user_can_log_in_and_we_can_read_the_tokens( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() nonce = new_nonce() url = client.authorization_url( state=new_state(), nonce=nonce, code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.expires_in > 0 assert issued.tokens.refresh_token is not None assert issued.tokens.id_token is not None # Our own verification, against Keycloak's real signing keys. id_claims = _decode_id_token(issued.tokens.id_token, expected_nonce=nonce) assert id_claims.preferred_username == USERNAME access_claims = _decode_token(issued.tokens.access_token) assert "app-provider" in access_claims.roles assert "platform-admin" in access_claims.roles async def test_a_code_cannot_be_used_twice( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) code = _log_in(url) await client.exchange_code(code=code, code_verifier=verifier) # Keycloak's own invalid_grant, not one we invented in a mock. with pytest.raises(InvalidGrantError): await client.exchange_code(code=code, code_verifier=verifier) async def test_the_wrong_pkce_verifier_is_refused( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(new_code_verifier()) ) code = _log_in(url) with pytest.raises(InvalidGrantError): await client.exchange_code(code=code, code_verifier=new_code_verifier()) async def test_a_wrong_client_secret_is_not_the_users_fault( settings: OIDCSettings, portal_settings_env: None ) -> None: good = KeycloakOIDCClient(settings) verifier = new_code_verifier() url = good.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) code = _log_in(url) wrong = KeycloakOIDCClient(settings.model_copy(update={"client_secret": "wrong"})) with pytest.raises(IdentityProviderUnavailableError): await wrong.exchange_code(code=code, code_verifier=verifier) async def test_a_session_can_be_refreshed( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.tokens.refresh_token is not None refreshed = await client.refresh(issued.tokens.refresh_token) assert _decode_token(refreshed.tokens.access_token).preferred_username == USERNAME async def test_logging_out_ends_the_keycloak_session( client: KeycloakOIDCClient, portal_settings_env: None ) -> None: verifier = new_code_verifier() url = client.authorization_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(verifier) ) issued = await client.exchange_code(code=_log_in(url), code_verifier=verifier) assert issued.tokens.id_token is not None and issued.tokens.refresh_token is not None logout_url = client.logout_url( id_token=issued.tokens.id_token, post_logout_redirect_uri="http://localhost:8000/" ) with httpx.Client(follow_redirects=False, timeout=15) as browser: # Not followed: Keycloak redirects to the portal, which is not running here. ended = browser.get(logout_url) assert ended.status_code in (200, 302, 303) # The session is gone at Keycloak, so its refresh token is worthless. with pytest.raises(InvalidGrantError): await client.refresh(issued.tokens.refresh_token) def test_the_signup_page_is_available(client: KeycloakOIDCClient) -> None: url = client.registration_url( state=new_state(), nonce=new_nonce(), code_challenge=code_challenge(new_code_verifier()) ) with httpx.Client(follow_redirects=True, timeout=15) as browser: page = browser.get(url) assert page.status_code == 200 assert "password" in page.text.lower()
