From 6b092c74f06a56d35ed27e28d1af38c8a251d73a Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Fri, 12 Jun 2026 16:23:59 +0200 Subject: [PATCH 1/3] refactor: oidc polciy test along with new tests. Signed-off-by: Alexander Cristurean --- testsuite/oidc/keycloak/objects.py | 41 ++++- testsuite/oidc/test_client.py | 172 ------------------ .../extensions/oidc_policy/conftest.py | 34 ++-- .../test_oidc_policy_confidential_client.py | 154 ++++++---------- .../test_oidc_policy_jwt_validation.py | 62 +++++++ .../test_oidc_policy_public_client.py | 120 ++++++------ .../oidc_policy/test_oidc_policy_redirect.py | 116 ++++++++++++ .../test_oidc_policy_service_client.py | 136 +++++--------- 8 files changed, 386 insertions(+), 449 deletions(-) delete mode 100644 testsuite/oidc/test_client.py create mode 100644 testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py create mode 100644 testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py diff --git a/testsuite/oidc/keycloak/objects.py b/testsuite/oidc/keycloak/objects.py index f3d0de6eb..84cb1a0f4 100644 --- a/testsuite/oidc/keycloak/objects.py +++ b/testsuite/oidc/keycloak/objects.py @@ -1,11 +1,50 @@ """Object wrappers for Keycloak resources""" +from dataclasses import dataclass, field from functools import cached_property -from typing import List +from typing import List, Literal from keycloak import KeycloakOpenID, KeycloakAdmin +@dataclass +class ClientConfig: # pylint: disable=too-many-instance-attributes + """Configuration for creating OIDC test clients""" + + client_id: str + client_type: Literal["confidential", "public", "service"] + redirect_uris: list[str] + web_origins: list[str] + root_url: str + public_client: bool = False + standard_flow_enabled: bool = True + service_accounts_enabled: bool = False + direct_access_grants_enabled: bool = True + default_client_scopes: list[str] = field(default_factory=lambda: ["openid", "profile", "email"]) + optional_client_scopes: list[str] = field(default_factory=lambda: ["offline_access", "microprofile-jwt"]) + + def to_keycloak_payload(self): + """Convert to Keycloak client creation payload""" + return { + "name": self.client_id, + "clientId": self.client_id, + "publicClient": self.public_client, + "standardFlowEnabled": self.standard_flow_enabled, + "serviceAccountsEnabled": self.service_accounts_enabled, + "protocol": "openid-connect", + "redirectUris": self.redirect_uris, + "webOrigins": self.web_origins, + "directAccessGrantsEnabled": self.direct_access_grants_enabled, + "rootUrl": self.root_url, + "defaultClientScopes": self.default_client_scopes, + "optionalClientScopes": self.optional_client_scopes, + "attributes": { + "backchannel.logout.session.required": "true", + "use.refresh.tokens": "true", + }, + } + + class Realm: """Helper class for Keycloak realm manipulation""" diff --git a/testsuite/oidc/test_client.py b/testsuite/oidc/test_client.py deleted file mode 100644 index 62d05bcc9..000000000 --- a/testsuite/oidc/test_client.py +++ /dev/null @@ -1,172 +0,0 @@ -"""OIDC test client wrapper for easier testing""" - -from dataclasses import dataclass, field -from typing import Literal -from keycloak import KeycloakOpenID - -from testsuite.httpx.auth import HttpxOidcClientAuth -from testsuite.oidc import Token -from testsuite.kuadrant.extensions.oidc_policy import Provider - - -@dataclass -class ClientConfig: # pylint: disable=too-many-instance-attributes - """Configuration for creating OIDC test clients""" - - client_id: str - client_type: Literal["confidential", "public", "service"] - redirect_uris: list[str] - web_origins: list[str] - root_url: str - public_client: bool = False - standard_flow_enabled: bool = True - service_accounts_enabled: bool = False - direct_access_grants_enabled: bool = True - default_client_scopes: list[str] = field(default_factory=lambda: ["openid", "profile", "email"]) - optional_client_scopes: list[str] = field(default_factory=lambda: ["offline_access", "microprofile-jwt"]) - - def to_keycloak_payload(self): - """Convert to Keycloak client creation payload""" - return { - "name": self.client_id, - "clientId": self.client_id, - "publicClient": self.public_client, - "standardFlowEnabled": self.standard_flow_enabled, - "serviceAccountsEnabled": self.service_accounts_enabled, - "protocol": "openid-connect", - "redirectUris": self.redirect_uris, - "webOrigins": self.web_origins, - "directAccessGrantsEnabled": self.direct_access_grants_enabled, - "rootUrl": self.root_url, - "defaultClientScopes": self.default_client_scopes, - "optionalClientScopes": self.optional_client_scopes, - "attributes": { - "backchannel.logout.session.required": "true", - "use.refresh.tokens": "true", - }, - } - - -class OIDCTestClient: - """Wrapper for OIDC client with testing utilities""" - - def __init__(self, keycloak_oidc_client: KeycloakOpenID): - self.oidc_client = keycloak_oidc_client - - def get_token(self, username: str, password: str) -> Token: - """Get access token for user credentials""" - token_data = self.oidc_client.token(username, password) - return Token( - token_data["access_token"], - self._create_refresh_func(), - token_data.get("refresh_token", ""), - ) - - def get_service_account_token(self) -> Token: - """Get service account token (for confidential clients)""" - token_data = self.oidc_client.token(grant_type="client_credentials") - return Token( - token_data["access_token"], - self._create_refresh_func(), - token_data.get("refresh_token", ""), - ) - - def get_auth(self, username: str, password: str, location: str = "authorization") -> HttpxOidcClientAuth: - """Get HttpxOidcClientAuth for testing""" - token = self.get_token(username, password) - return HttpxOidcClientAuth(token, location) - - def create_provider_config(self, oidc_provider) -> Provider: - """Create Provider configuration for OIDC policy""" - return Provider( - issuerURL=oidc_provider.well_known["issuer"], - clientID=self.oidc_client.client_id, - authorizationEndpoint=oidc_provider.well_known["authorization_endpoint"], - tokenEndpoint=oidc_provider.well_known["token_endpoint"], - ) - - def _create_refresh_func(self): - """Create refresh token function""" - - def refresh_token_func(refresh_token: str) -> Token: - new_token_data = self.oidc_client.refresh_token(refresh_token) - return Token( - new_token_data["access_token"], - refresh_token_func, - new_token_data.get("refresh_token", ""), - ) - - return refresh_token_func - - @classmethod - def create_confidential_client(cls, keycloak, hostname: str, client_id: str = "my-confidential-client"): - """Factory method for confidential client""" - config = ClientConfig( - client_id=client_id, - client_type="confidential", - public_client=False, - redirect_uris=[f"http://{hostname}/*"], - web_origins=[f"http://{hostname}"], - root_url=f"http://{hostname}", - service_accounts_enabled=True, - ) - - keycloak_client = keycloak.realm.create_client(**config.to_keycloak_payload()) - - oidc_client = KeycloakOpenID( - server_url=keycloak.server_url, - client_id=keycloak_client.auth_id, - realm_name=keycloak.realm_name, - client_secret_key=keycloak_client.secret, - ) - - return cls(oidc_client) - - @classmethod - def create_public_client(cls, keycloak, hostname: str, client_id: str = "my-public-client"): - """Factory method for public client""" - config = ClientConfig( - client_id=client_id, - client_type="public", - public_client=True, - redirect_uris=[f"http://{hostname}/*"], - web_origins=[f"http://{hostname}"], - root_url=f"http://{hostname}", - service_accounts_enabled=False, - ) - - keycloak_client = keycloak.realm.create_client(**config.to_keycloak_payload()) - - oidc_client = KeycloakOpenID( - server_url=keycloak.server_url, - client_id=keycloak_client.auth_id, - realm_name=keycloak.realm_name, - ) - - return cls(oidc_client) - - @classmethod - def create_service_client(cls, keycloak, hostname: str, client_id: str = "my-service-client"): - """Factory method for service account client""" - config = ClientConfig( - client_id=client_id, - client_type="service", - public_client=False, - standard_flow_enabled=False, - service_accounts_enabled=True, - redirect_uris=[f"http://{hostname}/*"], - web_origins=[f"http://{hostname}"], - root_url=f"http://{hostname}", - direct_access_grants_enabled=False, - ) - - keycloak_client = keycloak.realm.create_client(**config.to_keycloak_payload()) - - oidc_client = KeycloakOpenID( - server_url=keycloak.server_url, - client_id=keycloak_client.auth_id, - realm_name=keycloak.realm_name, - client_secret_key=keycloak_client.secret, - ) - - return cls(oidc_client) diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py index 0a72db00f..c8ccaf63b 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py @@ -1,16 +1,12 @@ -"""Shared pytest fixtures for OIDC policy testing. - -This module provides shared fixtures for testing OIDC (OpenID Connect) policy functionality, -including gateway setup and policy management. Client-specific fixtures are now located -in their respective test files. -""" +"""Shared pytest fixtures for OIDC policy testing.""" from contextlib import contextmanager + import pytest from testsuite.gateway import Gateway, GatewayListener from testsuite.gateway.gateway_api.gateway import KuadrantGateway -from testsuite.kuadrant.extensions.oidc_policy import OIDCPolicy +from testsuite.kuadrant.extensions.oidc_policy import OIDCPolicy, Provider @pytest.fixture(scope="module") @@ -25,10 +21,9 @@ def gateway(request, domain_name, base_domain, cluster, blame, label) -> Gateway return gw -# JWT Cookie Helper fixture @contextmanager def set_jwt_cookie(client, token_value: str): - """Context manager for setting JWT cookies with automatic cleanup""" + """Context manager for setting JWT cookies with automatic cleanup.""" client.cookies.set("jwt", token_value) try: yield @@ -36,24 +31,21 @@ def set_jwt_cookie(client, token_value: str): client.cookies.clear() -# OIDC Policy fixtures - these are shared and will be overridden in individual test files @pytest.fixture(scope="module") -def oidc_policy_provider_config(oidc_provider, test_client): +def oidc_policy_provider_config(oidc_provider, keycloak_client): """Create Provider configuration for the OIDC policy.""" - return test_client.create_provider_config(oidc_provider) + return Provider( + issuerURL=oidc_provider.well_known["issuer"], + clientID=keycloak_client.client_id, + authorizationEndpoint=oidc_provider.well_known["authorization_endpoint"], + tokenEndpoint=oidc_provider.well_known["token_endpoint"], + ) @pytest.fixture(scope="module") def oidc_policy(cluster, blame, oidc_policy_provider_config, gateway): - """Create OIDC policy instance for testing. - - Note: This fixture depends on 'provider' which should be defined in each test file - with the appropriate client-specific configuration. - """ - oidc_policy = OIDCPolicy.create_instance( - cluster, blame("oidc-policy"), gateway, provider=oidc_policy_provider_config - ) - return oidc_policy + """Create OIDC policy instance targeting the gateway.""" + return OIDCPolicy.create_instance(cluster, blame("oidc-policy"), gateway, provider=oidc_policy_provider_config) @pytest.fixture(scope="module", autouse=True) diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py index 0210d4691..91a11f813 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py @@ -1,128 +1,86 @@ -"""Tests for OIDC policy functionality with Confidential Client. - -This module tests OIDC authentication flows using a confidential client with -Authorization Code Flow. Confidential clients can securely store client secrets -and are typically used in server-side web applications. - -Key discovery: OIDC policy supports JWT token authentication via 'jwt' cookie, -allowing programmatic testing without full OAuth2 redirect flows. -""" +"""Tests for OIDCPolicy with a confidential client (Authorization Code Flow).""" from urllib.parse import quote -import jwt as jwt_lib +import jwt as jwt_lib import pytest -from testsuite.oidc.test_client import OIDCTestClient +from keycloak import KeycloakOpenID + +from testsuite.oidc import Token +from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only, pytest.mark.extensions] @pytest.fixture(scope="module") -def test_client(keycloak, hostname): - """Create confidential OIDC test client""" - client = OIDCTestClient.create_confidential_client(keycloak, hostname.hostname) - return client +def keycloak_client(keycloak, hostname, blame): + """Create confidential OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("confidential"), + client_type="confidential", + public_client=False, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + service_accounts_enabled=True, + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + client_secret_key=kc_client.secret, + ) @pytest.fixture(scope="module") -def auth(test_client, keycloak): - """Get authentication object for confidential client.""" - return test_client.get_auth(keycloak.test_username, keycloak.test_password) +def auth(keycloak_client, keycloak): + """Get a Token for the test user.""" + + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + data = keycloak_client.token(keycloak.test_username, keycloak.test_password) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) -def test_confidential_client_authorization_flow(client, auth, test_client, gateway): - """Test confidential client authorization code flow with secrets and enhanced features.""" - # Test unauthenticated request redirects + +def test_unauthenticated_redirect(client, keycloak_client, gateway): + """Unauthenticated request redirects to OIDC provider with correct params.""" response = client.get("/") - assert response.status_code == 302, "Unauthenticated request should redirect" - assert "Location" in response.headers, "Redirect must include Location header" + assert response.status_code == 302 + assert "Location" in response.headers location = response.headers["Location"] - - # Validate OAuth2 redirect parameters - assert "response_type=code" in location, "Should use Authorization Code Flow" - assert "scope=openid" in location, "Should request OpenID scope" - assert test_client.oidc_client.client_id in location, "Should include correct client ID" + assert "response_type=code" in location + assert "scope=openid" in location + assert keycloak_client.client_id in location expected_redirect_uri = f"redirect_uri=http%3A%2F%2F{quote(gateway.model.spec.listeners[0].hostname, safe=':.')}" - assert expected_redirect_uri in location, "Should have correct redirect URI" - - # Test JWT cookie authentication with valid token - with set_jwt_cookie(client, auth.token.access_token): - response = client.get("/") - assert response.status_code == 200 + assert expected_redirect_uri in location - token = jwt_lib.decode(auth.token.access_token, options={"verify_signature": False}) - # Basic validations - assert "openid" in token["scope"], "Token should include OpenID scope" - assert token["typ"] == "Bearer", "Should be a Bearer token" - assert ( - token["azp"] == test_client.oidc_client.client_id - ), f"Token should be issued for {test_client.oidc_client.client_id}" - - # Confidential client specific token validations - assert ( - token["azp"] == test_client.oidc_client.client_id - ), "Token should be issued for correct confidential client" - assert "preferred_username" in token, "Confidential client tokens should contain user identity" - assert "email" in token, "Confidential client tokens should contain user email" - assert token["preferred_username"] == "testuser", "Should be test user" - assert token["email"] == "testuser@anything.invalid", "Token should contain user email" - - # Verify token contains expected scopes and enhanced claims - assert "openid" in token["scope"], "Token should include OpenID scope" - - # May also have additional scopes like profile, email (configured in client setup) - expected_claims = ["preferred_username", "email"] - for claim in expected_claims: - assert claim in token, f"Token should contain {claim} claim" - - -def test_confidential_client_token(client, test_client, auth): - """Test advanced confidential client features including refresh tokens and logout support.""" - with set_jwt_cookie(client, auth.token.access_token): +def test_jwt_cookie_authentication(client, auth): + """Valid JWT cookie grants access.""" + with set_jwt_cookie(client, auth.access_token): response = client.get("/") assert response.status_code == 200 - # Test refresh token support - assert hasattr(auth.token, "refresh_token"), "Should have refresh token" - assert auth.token.refresh_token is not None, "Refresh token should not be None" - assert hasattr(auth.token, "refresh_function"), "Should have refresh function" - assert auth.token.refresh_function is not None, "Refresh function should not be None" - - token = jwt_lib.decode(auth.token.access_token, options={"verify_signature": False}) - # Basic validations - assert "openid" in token["scope"], "Token should include OpenID scope" - assert token["typ"] == "Bearer", "Should be a Bearer token" - assert ( - token["azp"] == test_client.oidc_client.client_id - ), f"Token should be issued for {test_client.oidc_client.client_id}" +def test_token_claims(auth, keycloak_client, keycloak): + """Confidential client token contains user identity claims.""" + token = jwt_lib.decode(auth.access_token, options={"verify_signature": False}) - # Verify token properties that support logout scenarios (backchannel logout) - assert "azp" in token, "Should have authorized party for logout tracking" - assert "iat" in token, "Should have issued at time" - assert "exp" in token, "Should have expiration time" + assert token["typ"] == "Bearer" + assert "openid" in token["scope"] + assert token["azp"] == keycloak_client.client_id + assert token["preferred_username"].lower() == keycloak.test_username.lower() + assert "email" in token -def test_confidential_client_malformed_jwt(client): - """Test that malformed JWT is rejected and redirects to auth.""" - with set_jwt_cookie(client, "not.a.jwt"): - response = client.get("/") - assert response.status_code == 302, "Malformed JWT should redirect to authentication" - - -def test_confidential_client_tampered_jwt_signature(client, auth): - """Test that JWT with tampered signature is rejected.""" - # Tamper with the signature part of a valid JWT - parts = auth.token.access_token.split(".") - if len(parts) != 3: - pytest.fail("Invalid JWT token format") - - tampered_token = f"{parts[0]}.{parts[1]}.tampered_signature" - with set_jwt_cookie(client, tampered_token): - response = client.get("/") - assert response.status_code == 302, "JWT with tampered signature should redirect to authentication" +def test_refresh_token_present(auth): + """Confidential client token includes a refresh token.""" + assert auth.refresh_token is not None + assert auth.refresh_function is not None diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py new file mode 100644 index 000000000..0b0b058b8 --- /dev/null +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py @@ -0,0 +1,62 @@ +"""Tests for OIDCPolicy JWT validation (client-type independent).""" + +import pytest + +from keycloak import KeycloakOpenID + +from testsuite.oidc import Token +from testsuite.oidc.keycloak.objects import ClientConfig +from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie + +pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only, pytest.mark.extensions] + + +@pytest.fixture(scope="module") +def keycloak_client(keycloak, hostname, blame): + """Create confidential OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("jwt-validation"), + client_type="confidential", + public_client=False, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + service_accounts_enabled=True, + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + client_secret_key=kc_client.secret, + ) + + +@pytest.fixture(scope="module") +def auth(keycloak_client, keycloak): + """Get a Token for the test user.""" + + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + data = keycloak_client.token(keycloak.test_username, keycloak.test_password) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + +def test_malformed_jwt(client): + """Malformed JWT is rejected and redirects to auth.""" + with set_jwt_cookie(client, "not.a.jwt"): + response = client.get("/") + assert response.status_code == 302 + + +def test_tampered_jwt_signature(client, auth): + """JWT with tampered signature is rejected.""" + parts = auth.access_token.split(".") + assert len(parts) == 3, "Invalid JWT token format" + + tampered_token = f"{parts[0]}.{parts[1]}.tampered_signature" + with set_jwt_cookie(client, tampered_token): + response = client.get("/") + assert response.status_code == 302 diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py index 2975351b9..d4d574cb9 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py @@ -1,97 +1,79 @@ -"""Tests for OIDC policy functionality with Public Client. - -This module tests OIDC authentication flows using a public client with PKCE -(Authorization Code Flow). Public clients are typically used in SPAs and mobile apps -where client secrets cannot be securely stored. - -Key discovery: OIDC policy supports JWT token authentication via 'jwt' cookie, -allowing programmatic testing without full OAuth2 redirect flows. -""" +"""Tests for OIDCPolicy with a public client (Authorization Code Flow + PKCE).""" from urllib.parse import quote import jwt as jwt_lib import pytest -from testsuite.oidc.test_client import OIDCTestClient +from keycloak import KeycloakOpenID + +from testsuite.oidc import Token +from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only, pytest.mark.extensions] @pytest.fixture(scope="module") -def test_client(keycloak, hostname): - """Create public OIDC test client""" - client = OIDCTestClient.create_public_client(keycloak, hostname.hostname) - return client +def keycloak_client(keycloak, hostname, blame): + """Create public OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("public"), + client_type="public", + public_client=True, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + ) @pytest.fixture(scope="module") -def auth(test_client, keycloak): - """Get authentication object for public client.""" - return test_client.get_auth(keycloak.test_username, keycloak.test_password) +def auth(keycloak_client, keycloak): + """Get a Token for the test user.""" + + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + data = keycloak_client.token(keycloak.test_username, keycloak.test_password) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) -def test_public_client_authentication_flow(client, auth, test_client, gateway): - """Test complete public client authentication flow with PKCE.""" - # Test unauthenticated request redirects to OAuth2 + +def test_unauthenticated_redirect(client, keycloak_client, gateway): + """Unauthenticated request redirects to OIDC provider with PKCE params.""" response = client.get("/") - assert response.status_code == 302, "Unauthenticated request should redirect" - assert "Location" in response.headers, "Redirect must include Location header" + assert response.status_code == 302 + assert "Location" in response.headers location = response.headers["Location"] - - # Validate OAuth2 redirect parameters - assert "response_type=code" in location, "Should use Authorization Code Flow" - assert "scope=openid" in location, "Should request OpenID scope" - assert test_client.oidc_client.client_id in location, "Should include correct client ID" + assert "response_type=code" in location + assert "scope=openid" in location + assert keycloak_client.client_id in location + assert "code_challenge" in location or "response_type=code" in location expected_redirect_uri = f"redirect_uri=http%3A%2F%2F{quote(gateway.model.spec.listeners[0].hostname, safe=':.')}" - assert expected_redirect_uri in location, "Should have correct redirect URI" + assert expected_redirect_uri in location - # Public clients should use PKCE - assert ( - "code_challenge" in location or "response_type=code" in location - ), "Should use Authorization Code Flow with PKCE" - # Test JWT cookie authentication with valid token - with set_jwt_cookie(client, auth.token.access_token): +def test_jwt_cookie_authentication(client, auth): + """Valid JWT cookie grants access.""" + with set_jwt_cookie(client, auth.access_token): response = client.get("/") - assert response.status_code == 200, "Valid JWT cookie should allow access" - - token = jwt_lib.decode(auth.token.access_token, options={"verify_signature": False}) + assert response.status_code == 200 - # Basic validations - assert "openid" in token["scope"], "Token should include OpenID scope" - assert token["typ"] == "Bearer", "Should be a Bearer token" - assert ( - token["azp"] == test_client.oidc_client.client_id - ), f"Token should be issued for {test_client.oidc_client.client_id}" - # Public client specific token validations - assert token["azp"] == test_client.oidc_client.client_id, "Token should be issued for correct public client" - assert "preferred_username" in token, "Public client tokens should contain user identity" - assert "email" in token, "Public client tokens should contain user email" - assert token["preferred_username"] == "testuser", "Should be test user" - assert token["email"] == "testuser@anything.invalid", "Token should contain user email" - assert "openid" in token["scope"], "Token should include OpenID scope" +def test_token_claims(auth, keycloak_client, keycloak): + """Public client token contains user identity claims.""" + token = jwt_lib.decode(auth.access_token, options={"verify_signature": False}) - -def test_public_client_malformed_jwt(client): - """Test that malformed JWT is rejected and redirects to auth.""" - with set_jwt_cookie(client, "not.a.jwt"): - response = client.get("/") - assert response.status_code == 302, "Malformed JWT should redirect to authentication" - - -def test_public_client_tampered_jwt_signature(client, auth): - """Test that JWT with tampered signature is rejected.""" - # Tamper with the signature part of a valid JWT - parts = auth.token.access_token.split(".") - if len(parts) != 3: - pytest.fail("Invalid JWT token format") - - tampered_token = f"{parts[0]}.{parts[1]}.tampered_signature" - with set_jwt_cookie(client, tampered_token): - response = client.get("/") - assert response.status_code == 302, "JWT with tampered signature should redirect to authentication" + assert token["typ"] == "Bearer" + assert "openid" in token["scope"] + assert token["azp"] == keycloak_client.client_id + assert token["preferred_username"].lower() == keycloak.test_username.lower() + assert "email" in token diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py new file mode 100644 index 000000000..72f993067 --- /dev/null +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py @@ -0,0 +1,116 @@ +"""Tests for OIDCPolicy redirect handling bug fixes (kuadrant-operator#2017). + +Validates fixes for: +- Bug 1: Port dropped from auto-constructed redirect URI when listener uses non-standard port +- Bug 2: OPA cookie parser breaks on '=' in values (indirectly tested via query string test) +- Bug 3: Target cookie drops query string after OIDC auth redirect +""" + +from urllib.parse import unquote + +import pytest + +from keycloak import KeycloakOpenID + +from testsuite.gateway import GatewayListener +from testsuite.gateway.exposers import StaticLocalHostname +from testsuite.gateway.gateway_api.gateway import KuadrantGateway +from testsuite.oidc import Token +from testsuite.oidc.keycloak.objects import ClientConfig +from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie + +pytestmark = [ + pytest.mark.authorino, + pytest.mark.kuadrant_only, + pytest.mark.extensions, + pytest.mark.issue("https://github.com/Kuadrant/kuadrant-operator/issues/2017"), +] + +CUSTOM_PORT = 8001 + + +@pytest.fixture(scope="module") +def gateway(request, domain_name, base_domain, cluster, blame, label): + """Gateway with a non-standard listener port to test port inclusion in redirect URI.""" + fqdn = f"{domain_name}-kuadrant.{base_domain}" + gw = KuadrantGateway.create_instance(cluster, blame("gw"), {"app": label}) + gw.add_listener(GatewayListener(hostname=fqdn, port=CUSTOM_PORT)) + request.addfinalizer(gw.delete) + gw.commit() + gw.wait_for_ready() + return gw + + +@pytest.fixture(scope="module") +def hostname(gateway, domain_name, exposer): + """Hostname that connects on the non-standard gateway port.""" + fqdn = f"{domain_name}-kuadrant.{exposer.base_domain}" + ip = gateway.refresh().model.status.addresses[0].value + + return StaticLocalHostname(fqdn, lambda: f"{ip}:{CUSTOM_PORT}") + + +@pytest.fixture(scope="module") +def keycloak_client(keycloak, hostname, blame): + """Create confidential OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("redirect"), + client_type="confidential", + public_client=False, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + service_accounts_enabled=True, + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + client_secret_key=kc_client.secret, + ) + + +@pytest.fixture(scope="module") +def auth(keycloak_client, keycloak): + """Get a Token for the test user.""" + + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + data = keycloak_client.token(keycloak.test_username, keycloak.test_password) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + +def test_redirect_uri_includes_listener_port(client, gateway): + """Auto-constructed redirect URI must include the non-standard listener port.""" + response = client.get("/") + assert response.status_code == 302 + + location = unquote(response.headers["Location"]) + gw_hostname = gateway.model.spec.listeners[0].hostname + expected = f"redirect_uri=http://{gw_hostname}:{CUSTOM_PORT}/auth/callback" + assert expected in location, f"redirect_uri should include port {CUSTOM_PORT}, got: {location}" + + +def test_query_string_preserved_in_target_cookie(client): + """Target cookie must include query string parameters from the original request.""" + response = client.get("/get?foo=bar&baz=qux") + assert response.status_code == 302 + + set_cookie_headers = response.headers.get_list("set-cookie") + target_cookies = [c for c in set_cookie_headers if c.startswith("target=")] + assert target_cookies, "Response should set a 'target' cookie" + + target_value = target_cookies[0].split(";")[0] + assert ( + target_value == "target=/get?foo=bar&baz=qux" + ), f"Target cookie should include query string, got: {target_value}" + + +def test_authenticated_request_with_query_params(client, auth): + """Authenticated requests with query parameters should succeed.""" + with set_jwt_cookie(client, auth.access_token): + response = client.get("/get?foo=bar") + assert response.status_code == 200 diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py index 75a0e0166..6e09b30c0 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py @@ -1,110 +1,70 @@ -"""Tests for OIDC policy functionality with Service Client. +"""Tests for OIDCPolicy with a service client (Client Credentials Flow).""" -This module tests OIDC authentication flows using a service client for machine-to-machine -authentication (Client Credentials Flow). Service clients are used for backend services -that need to authenticate without user interaction. - -Key discovery: OIDC policy supports JWT token authentication via 'jwt' cookie, -allowing programmatic testing without full OAuth2 redirect flows. -""" - -import pytest import jwt as jwt_lib +import pytest + +from keycloak import KeycloakOpenID -from testsuite.httpx.auth import HttpxOidcClientAuth -from testsuite.oidc.test_client import OIDCTestClient +from testsuite.oidc import Token +from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie pytestmark = [pytest.mark.authorino, pytest.mark.kuadrant_only, pytest.mark.extensions] @pytest.fixture(scope="module") -def test_client(keycloak, hostname): - """Create service OIDC test client""" - client = OIDCTestClient.create_service_client(keycloak, hostname.hostname) - return client +def keycloak_client(keycloak, hostname, blame): + """Create service OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("service"), + client_type="service", + public_client=False, + standard_flow_enabled=False, + service_accounts_enabled=True, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + direct_access_grants_enabled=False, + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + client_secret_key=kc_client.secret, + ) @pytest.fixture(scope="module") -def auth(test_client): - """Get authentication object for service client using service account token.""" - token = test_client.get_service_account_token() - return HttpxOidcClientAuth(token, "authorization") +def auth(keycloak_client): + """Get a Token using client credentials grant.""" + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) -def test_service_client_machine_to_machine_flow(client, auth, test_client): - """Test service client machine-to-machine authentication flow.""" - # Test unauthenticated request - response = client.get("/") - assert response.status_code == 302, "Service client should redirect when no credentials found" - assert "x-ext-auth-reason" in response.headers, "Should provide auth failure reason" - - # Verify no sensitive information in redirect - location = response.headers.get("Location", "") - assert "client_credentials" not in location, "Grant type should not appear in redirects" - - # Service clients shouldn't have redirect URIs configured - verify this works - # The redirect should not contain redirect_uri parameter for service clients - # Note: The actual redirect behavior may vary based on OIDC policy configuration - - # Test JWT cookie authentication with valid token - with set_jwt_cookie(client, auth.token.access_token): - response = client.get("/") - assert response.status_code == 200, "Valid service client JWT should allow access" + data = keycloak_client.token(grant_type="client_credentials") + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - token = jwt_lib.decode(auth.token.access_token, options={"verify_signature": False}) - # Basic validations - assert "openid" in token["scope"], "Token should include OpenID scope" - assert token["typ"] == "Bearer", "Should be a Bearer token" - assert ( - token["azp"] == test_client.oidc_client.client_id - ), f"Token should be issued for {test_client.oidc_client.client_id}" - - # Service client specific validations - assert "azp" in token, "Service client tokens should have authorized party (azp)" - assert token["azp"] == test_client.oidc_client.client_id, "Token should be issued for correct service client" - - # Service clients use client_credentials flow, so no user context - assert ( - "preferred_username" not in token or token.get("preferred_username") != "testuser" - ), "Service client token should not contain user identity" - - # Should have service account context instead - assert "clientId" in token or "azp" in token, "Should have client context" - - # Service client tokens should have specific characteristics - assert "openid" in token["scope"], "Should include OpenID scope" - assert token["typ"] == "Bearer", "Should be Bearer token" - - # Should NOT have user-specific claims (machine-to-machine) - user_claims = ["preferred_username", "email", "given_name", "family_name"] - for claim in user_claims: - if claim in token: - assert token[claim] != "testuser", f"Service client should not have user claim: {claim}" - - # Verify no sensitive information in response headers - with set_jwt_cookie(client, auth.token.access_token): - response = client.get("/") - for header_name, header_value in response.headers.items(): - assert "client_secret" not in str(header_value).lower(), f"Client secret found in {header_name}" +def test_unauthenticated_redirect(client): + """Unauthenticated request redirects to OIDC provider.""" + response = client.get("/") + assert response.status_code == 302 -def test_service_client_malformed_jwt(client): - """Test that malformed JWT is rejected and redirects to auth.""" - with set_jwt_cookie(client, "not.a.jwt"): +def test_jwt_cookie_authentication(client, auth): + """Valid service account JWT cookie grants access.""" + with set_jwt_cookie(client, auth.access_token): response = client.get("/") - assert response.status_code == 302, "Malformed JWT should redirect to authentication" + assert response.status_code == 200 -def test_service_client_tampered_jwt_signature(client, auth): - """Test that JWT with tampered signature is rejected.""" - # Tamper with the signature part of a valid JWT - parts = auth.token.access_token.split(".") - if len(parts) != 3: - pytest.fail("Invalid JWT token format") +def test_token_claims(auth, keycloak_client): + """Service client token has client context but no user identity.""" + token = jwt_lib.decode(auth.access_token, options={"verify_signature": False}) - tampered_token = f"{parts[0]}.{parts[1]}.tampered_signature" - with set_jwt_cookie(client, tampered_token): - response = client.get("/") - assert response.status_code == 302, "JWT with tampered signature should redirect to authentication" + assert token["typ"] == "Bearer" + assert "openid" in token["scope"] + assert token["azp"] == keycloak_client.client_id + assert token.get("preferred_username") != "testuser" From 0f363b2c1e8b451368912ff0501903fcde93f4eb Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Mon, 15 Jun 2026 15:13:27 +0200 Subject: [PATCH 2/3] fix: standard hostname name for exposers. Signed-off-by: Alexander Cristurean --- testsuite/gateway/exposers.py | 2 +- .../tests/singlecluster/extensions/oidc_policy/conftest.py | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/testsuite/gateway/exposers.py b/testsuite/gateway/exposers.py index 9f33688c2..6a57f98f6 100644 --- a/testsuite/gateway/exposers.py +++ b/testsuite/gateway/exposers.py @@ -69,7 +69,7 @@ class LoadBalancerServiceExposer(Exposer): """Exposer using Load Balancer service for Gateway""" def expose_hostname(self, name, gateway: Gateway) -> Hostname: - hostname = f"{name}.{self.base_domain}" + hostname = f"{name}-{self.cluster.project}.{self.base_domain}" return StaticLocalHostname( hostname, gateway.external_ip, lambda: gateway.get_tls_cert(hostname), force_https=self.passthrough ) diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py index c8ccaf63b..5169dffae 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py @@ -8,11 +8,10 @@ from testsuite.gateway.gateway_api.gateway import KuadrantGateway from testsuite.kuadrant.extensions.oidc_policy import OIDCPolicy, Provider - @pytest.fixture(scope="module") def gateway(request, domain_name, base_domain, cluster, blame, label) -> Gateway: """Create and configure the test Gateway.""" - fqdn = f"{domain_name}-kuadrant.{base_domain}" + fqdn = f"{domain_name}-{cluster.project}.{base_domain}" gw = KuadrantGateway.create_instance(cluster, blame("gw"), {"app": label}) gw.add_listener(GatewayListener(hostname=fqdn)) request.addfinalizer(gw.delete) From 36f2d119d83bc29049af97b76f817e54e890492d Mon Sep 17 00:00:00 2001 From: Alexander Cristurean Date: Wed, 1 Jul 2026 17:34:26 +0200 Subject: [PATCH 3/3] fix: cosmetic changes. Signed-off-by: Alexander Cristurean --- testsuite/oidc/keycloak/objects.py | 3 +- .../extensions/oidc_policy/conftest.py | 16 +++- .../test_oidc_policy_confidential_client.py | 14 ---- .../test_oidc_policy_custom_redirect.py | 84 +++++++++++++++++++ .../test_oidc_policy_jwt_validation.py | 14 ---- .../test_oidc_policy_public_client.py | 15 ---- .../oidc_policy/test_oidc_policy_redirect.py | 20 +---- .../test_oidc_policy_service_client.py | 5 +- 8 files changed, 105 insertions(+), 66 deletions(-) create mode 100644 testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_custom_redirect.py diff --git a/testsuite/oidc/keycloak/objects.py b/testsuite/oidc/keycloak/objects.py index 84cb1a0f4..8af3bfeca 100644 --- a/testsuite/oidc/keycloak/objects.py +++ b/testsuite/oidc/keycloak/objects.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field from functools import cached_property -from typing import List, Literal +from typing import List from keycloak import KeycloakOpenID, KeycloakAdmin @@ -12,7 +12,6 @@ class ClientConfig: # pylint: disable=too-many-instance-attributes """Configuration for creating OIDC test clients""" client_id: str - client_type: Literal["confidential", "public", "service"] redirect_uris: list[str] web_origins: list[str] root_url: str diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py index 5169dffae..28911a464 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/conftest.py @@ -1,4 +1,4 @@ -"""Shared pytest fixtures for OIDC policy testing.""" +"""Shared pytest fixtures for OIDC policy testing. Client-specific fixtures are in each test file.""" from contextlib import contextmanager @@ -7,6 +7,8 @@ from testsuite.gateway import Gateway, GatewayListener from testsuite.gateway.gateway_api.gateway import KuadrantGateway from testsuite.kuadrant.extensions.oidc_policy import OIDCPolicy, Provider +from testsuite.oidc import Token + @pytest.fixture(scope="module") def gateway(request, domain_name, base_domain, cluster, blame, label) -> Gateway: @@ -47,6 +49,18 @@ def oidc_policy(cluster, blame, oidc_policy_provider_config, gateway): return OIDCPolicy.create_instance(cluster, blame("oidc-policy"), gateway, provider=oidc_policy_provider_config) +@pytest.fixture(scope="module") +def auth(keycloak_client, keycloak): + """Get a Token for the test user via password grant.""" + + def _refresh(refresh_token): + data = keycloak_client.refresh_token(refresh_token) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + data = keycloak_client.token(keycloak.test_username, keycloak.test_password) + return Token(data["access_token"], _refresh, data.get("refresh_token", "")) + + @pytest.fixture(scope="module", autouse=True) def commit(request, oidc_policy): """Commit and wait for OIDC policy to be ready.""" diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py index 91a11f813..6d1cae1bc 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_confidential_client.py @@ -7,7 +7,6 @@ from keycloak import KeycloakOpenID -from testsuite.oidc import Token from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie @@ -19,7 +18,6 @@ def keycloak_client(keycloak, hostname, blame): """Create confidential OIDC client on Keycloak.""" config = ClientConfig( client_id=blame("confidential"), - client_type="confidential", public_client=False, redirect_uris=[f"http://{hostname.hostname}/*"], web_origins=[f"http://{hostname.hostname}"], @@ -35,18 +33,6 @@ def keycloak_client(keycloak, hostname, blame): ) -@pytest.fixture(scope="module") -def auth(keycloak_client, keycloak): - """Get a Token for the test user.""" - - def _refresh(refresh_token): - data = keycloak_client.refresh_token(refresh_token) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - data = keycloak_client.token(keycloak.test_username, keycloak.test_password) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - def test_unauthenticated_redirect(client, keycloak_client, gateway): """Unauthenticated request redirects to OIDC provider with correct params.""" response = client.get("/") diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_custom_redirect.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_custom_redirect.py new file mode 100644 index 000000000..7ff789319 --- /dev/null +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_custom_redirect.py @@ -0,0 +1,84 @@ +"""Tests for OIDCPolicy with custom redirectURI (kuadrant-operator#2032). + +Validates that when provider.redirectURI is set, the OIDC redirect uses the custom +callback URL instead of the auto-constructed one from the gateway listener. +""" + +from urllib.parse import unquote + +import pytest + +from keycloak import KeycloakOpenID + +from testsuite.kuadrant.extensions.oidc_policy import OIDCPolicy, Provider +from testsuite.oidc.keycloak.objects import ClientConfig + +pytestmark = [ + pytest.mark.authorino, + pytest.mark.kuadrant_only, + pytest.mark.extensions, + pytest.mark.issue("https://github.com/Kuadrant/kuadrant-operator/issues/2017"), +] + +CUSTOM_CALLBACK_PATH = "/custom/callback" + + +@pytest.fixture(scope="module") +def keycloak_client(keycloak, hostname, blame): + """Create confidential OIDC client on Keycloak.""" + config = ClientConfig( + client_id=blame("custom-redir"), + public_client=False, + redirect_uris=[f"http://{hostname.hostname}/*"], + web_origins=[f"http://{hostname.hostname}"], + root_url=f"http://{hostname.hostname}", + service_accounts_enabled=True, + ) + kc_client = keycloak.realm.create_client(**config.to_keycloak_payload()) + return KeycloakOpenID( + server_url=keycloak.server_url, + client_id=kc_client.auth_id, + realm_name=keycloak.realm_name, + client_secret_key=kc_client.secret, + ) + + +@pytest.fixture(scope="module") +def custom_redirect_uri(gateway): + """Custom redirect URI pointing to a non-default callback path.""" + gw_hostname = gateway.model.spec.listeners[0].hostname + return f"http://{gw_hostname}{CUSTOM_CALLBACK_PATH}" + + +@pytest.fixture(scope="module") +def oidc_policy(cluster, blame, oidc_provider, keycloak_client, gateway, custom_redirect_uri): + """OIDCPolicy with custom redirectURI set.""" + provider = Provider( + issuerURL=oidc_provider.well_known["issuer"], + clientID=keycloak_client.client_id, + authorizationEndpoint=oidc_provider.well_known["authorization_endpoint"], + tokenEndpoint=oidc_provider.well_known["token_endpoint"], + redirectURI=custom_redirect_uri, + ) + return OIDCPolicy.create_instance(cluster, blame("oidc-policy"), gateway, provider=provider) + + +def test_redirect_uses_custom_redirect_uri(client, custom_redirect_uri): + """Initial redirect must use the custom redirectURI in the authorize URL.""" + response = client.get("/") + assert response.status_code == 302 + + location = unquote(response.headers["Location"]) + assert ( + f"redirect_uri={custom_redirect_uri}" in location + ), f"Expected custom redirect_uri={custom_redirect_uri} in location, got: {location}" + + +def test_custom_callback_path_in_redirect_uri(client): + """The redirect_uri parameter must contain the custom callback path, not the default /auth/callback.""" + response = client.get("/") + assert response.status_code == 302 + + location = unquote(response.headers["Location"]) + assert CUSTOM_CALLBACK_PATH in location, f"Custom callback path not found in: {location}" + assert "/auth/callback" not in location or CUSTOM_CALLBACK_PATH in location diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py index 0b0b058b8..4551c52fb 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_jwt_validation.py @@ -4,7 +4,6 @@ from keycloak import KeycloakOpenID -from testsuite.oidc import Token from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie @@ -16,7 +15,6 @@ def keycloak_client(keycloak, hostname, blame): """Create confidential OIDC client on Keycloak.""" config = ClientConfig( client_id=blame("jwt-validation"), - client_type="confidential", public_client=False, redirect_uris=[f"http://{hostname.hostname}/*"], web_origins=[f"http://{hostname.hostname}"], @@ -32,18 +30,6 @@ def keycloak_client(keycloak, hostname, blame): ) -@pytest.fixture(scope="module") -def auth(keycloak_client, keycloak): - """Get a Token for the test user.""" - - def _refresh(refresh_token): - data = keycloak_client.refresh_token(refresh_token) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - data = keycloak_client.token(keycloak.test_username, keycloak.test_password) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - def test_malformed_jwt(client): """Malformed JWT is rejected and redirects to auth.""" with set_jwt_cookie(client, "not.a.jwt"): diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py index d4d574cb9..c0da83b49 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_public_client.py @@ -7,7 +7,6 @@ from keycloak import KeycloakOpenID -from testsuite.oidc import Token from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie @@ -19,7 +18,6 @@ def keycloak_client(keycloak, hostname, blame): """Create public OIDC client on Keycloak.""" config = ClientConfig( client_id=blame("public"), - client_type="public", public_client=True, redirect_uris=[f"http://{hostname.hostname}/*"], web_origins=[f"http://{hostname.hostname}"], @@ -33,18 +31,6 @@ def keycloak_client(keycloak, hostname, blame): ) -@pytest.fixture(scope="module") -def auth(keycloak_client, keycloak): - """Get a Token for the test user.""" - - def _refresh(refresh_token): - data = keycloak_client.refresh_token(refresh_token) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - data = keycloak_client.token(keycloak.test_username, keycloak.test_password) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - def test_unauthenticated_redirect(client, keycloak_client, gateway): """Unauthenticated request redirects to OIDC provider with PKCE params.""" response = client.get("/") @@ -55,7 +41,6 @@ def test_unauthenticated_redirect(client, keycloak_client, gateway): assert "response_type=code" in location assert "scope=openid" in location assert keycloak_client.client_id in location - assert "code_challenge" in location or "response_type=code" in location expected_redirect_uri = f"redirect_uri=http%3A%2F%2F{quote(gateway.model.spec.listeners[0].hostname, safe=':.')}" assert expected_redirect_uri in location diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py index 72f993067..e40a66e3f 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_redirect.py @@ -15,7 +15,6 @@ from testsuite.gateway import GatewayListener from testsuite.gateway.exposers import StaticLocalHostname from testsuite.gateway.gateway_api.gateway import KuadrantGateway -from testsuite.oidc import Token from testsuite.oidc.keycloak.objects import ClientConfig from testsuite.tests.singlecluster.extensions.oidc_policy.conftest import set_jwt_cookie @@ -32,7 +31,7 @@ @pytest.fixture(scope="module") def gateway(request, domain_name, base_domain, cluster, blame, label): """Gateway with a non-standard listener port to test port inclusion in redirect URI.""" - fqdn = f"{domain_name}-kuadrant.{base_domain}" + fqdn = f"{domain_name}-{cluster.project}.{base_domain}" gw = KuadrantGateway.create_instance(cluster, blame("gw"), {"app": label}) gw.add_listener(GatewayListener(hostname=fqdn, port=CUSTOM_PORT)) request.addfinalizer(gw.delete) @@ -42,9 +41,9 @@ def gateway(request, domain_name, base_domain, cluster, blame, label): @pytest.fixture(scope="module") -def hostname(gateway, domain_name, exposer): +def hostname(gateway, domain_name, cluster, exposer): """Hostname that connects on the non-standard gateway port.""" - fqdn = f"{domain_name}-kuadrant.{exposer.base_domain}" + fqdn = f"{domain_name}-{cluster.project}.{exposer.base_domain}" ip = gateway.refresh().model.status.addresses[0].value return StaticLocalHostname(fqdn, lambda: f"{ip}:{CUSTOM_PORT}") @@ -55,7 +54,6 @@ def keycloak_client(keycloak, hostname, blame): """Create confidential OIDC client on Keycloak.""" config = ClientConfig( client_id=blame("redirect"), - client_type="confidential", public_client=False, redirect_uris=[f"http://{hostname.hostname}/*"], web_origins=[f"http://{hostname.hostname}"], @@ -71,18 +69,6 @@ def keycloak_client(keycloak, hostname, blame): ) -@pytest.fixture(scope="module") -def auth(keycloak_client, keycloak): - """Get a Token for the test user.""" - - def _refresh(refresh_token): - data = keycloak_client.refresh_token(refresh_token) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - data = keycloak_client.token(keycloak.test_username, keycloak.test_password) - return Token(data["access_token"], _refresh, data.get("refresh_token", "")) - - def test_redirect_uri_includes_listener_port(client, gateway): """Auto-constructed redirect URI must include the non-standard listener port.""" response = client.get("/") diff --git a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py index 6e09b30c0..6ed746f5f 100644 --- a/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py +++ b/testsuite/tests/singlecluster/extensions/oidc_policy/test_oidc_policy_service_client.py @@ -17,7 +17,6 @@ def keycloak_client(keycloak, hostname, blame): """Create service OIDC client on Keycloak.""" config = ClientConfig( client_id=blame("service"), - client_type="service", public_client=False, standard_flow_enabled=False, service_accounts_enabled=True, @@ -60,11 +59,11 @@ def test_jwt_cookie_authentication(client, auth): assert response.status_code == 200 -def test_token_claims(auth, keycloak_client): +def test_token_claims(auth, keycloak_client, keycloak): """Service client token has client context but no user identity.""" token = jwt_lib.decode(auth.access_token, options={"verify_signature": False}) assert token["typ"] == "Bearer" assert "openid" in token["scope"] assert token["azp"] == keycloak_client.client_id - assert token.get("preferred_username") != "testuser" + assert token.get("preferred_username", "").lower() != keycloak.test_username.lower()