This is an automated email from the ASF dual-hosted git repository. smolnar82 pushed a commit to branch knox_idf in repository https://gitbox.apache.org/repos/asf/knox.git
commit ca62c98fcba29de3c8166c094db8945e36b91e1c Author: Sandor Molnar <[email protected]> AuthorDate: Tue Jul 21 15:42:37 2026 +0200 KnoxIDF - Fixed pylint and test issues in Docker-based tests --- .../build/conf/topologies/knoxidf-ldap.xml | 2 +- .github/workflows/tests/common_utils.py | 4 +- .github/workflows/tests/test_knoxidf.py | 184 +++++++++++---------- 3 files changed, 101 insertions(+), 89 deletions(-) diff --git a/.github/workflows/build/conf/topologies/knoxidf-ldap.xml b/.github/workflows/build/conf/topologies/knoxidf-ldap.xml index a82920bdb..d25bb8869 100644 --- a/.github/workflows/build/conf/topologies/knoxidf-ldap.xml +++ b/.github/workflows/build/conf/topologies/knoxidf-ldap.xml @@ -15,7 +15,7 @@ </param> <param> <name>main.ldapRealm.contextFactory.url</name> - <value>ldap://ldap:33389</value> + <value>ldaps://localhost:33390</value> </param> <param> <name>main.ldapRealm.contextFactory.authenticationMechanism</name> diff --git a/.github/workflows/tests/common_utils.py b/.github/workflows/tests/common_utils.py index e80193374..5d11de053 100644 --- a/.github/workflows/tests/common_utils.py +++ b/.github/workflows/tests/common_utils.py @@ -100,6 +100,6 @@ def get_token_claim(token, claim): payload_json = base64.urlsafe_b64decode(payload_b64).decode('utf-8') payload = json.loads(payload_json) return payload.get(claim) - except Exception as e: + except (ValueError, IndexError, json.JSONDecodeError) as e: print(f"Failed to decode token for claim '{claim}': {e}") - return None \ No newline at end of file + return None diff --git a/.github/workflows/tests/test_knoxidf.py b/.github/workflows/tests/test_knoxidf.py index 918cc37d2..cbbbacaf2 100644 --- a/.github/workflows/tests/test_knoxidf.py +++ b/.github/workflows/tests/test_knoxidf.py @@ -13,18 +13,29 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Integration tests for Knox as an OIDC Identity Federation (IDF) provider.""" + import unittest import hashlib import base64 from urllib.parse import urlparse, parse_qs from requests.auth import HTTPBasicAuth -from common_utils import gateway_base_url, knox_get, knox_post, get_token_claim, get_token_id_display_text +from common_utils import ( + gateway_base_url, + knox_get, + knox_post, + get_token_claim, + get_token_id_display_text, +) + class TestKnoxIDF(unittest.TestCase): + """OIDC provider tests covering discovery, client credentials, and auth code flows.""" + def setUp(self): # Get the Knox Gateway URL from environment variables - self.base_url = gateway_base_url() + self.base_url = gateway_base_url() self.knoxidf_ldap_url = f"{self.base_url}gateway/knoxidf-ldap/" self.knoxidf_token_url = f"{self.base_url}gateway/knoxidf-token/" self.username = "guest" @@ -39,7 +50,7 @@ class TestKnoxIDF(unittest.TestCase): response = knox_get(url) self.assertEqual(response.status_code, 200) config = response.json() - + # Construct expected values based on dynamic base_url expected_issuer = f"{self.knoxidf_ldap_url}knoxidf" expected_auth_endpoint = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" @@ -54,31 +65,22 @@ class TestKnoxIDF(unittest.TestCase): self.assertEqual(config.get("jwks_uri"), expected_jwks_uri) self.assertEqual(config.get("response_types_supported"), ["code"]) - self.assertEqual(config.get("grant_types_supported"), ["authorization_code", "refresh_token"]) + self.assertEqual( + config.get("grant_types_supported"), + ["authorization_code", "refresh_token"], + ) self.assertEqual(config.get("id_token_signing_alg_values_supported"), ["RS256"]) - self.assertEqual(config.get("scopes_supported"), ["openid", "email", "profile", "offline_access"]) + self.assertEqual( + config.get("scopes_supported"), + ["openid", "email", "profile", "offline_access"], + ) def test_client_credentials_flow(self): """ Test OIDC Client Credentials Flow. """ # 1. Register client - reg_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/client/register" - print(f"Registering client at: {reg_url}") - data = { - "redirect_uris": "http://localhost/callback";, - "allowed_scopes": "openid,profile,email,offline_access" - } - response = knox_post( - reg_url, - data=data, - auth=HTTPBasicAuth(self.username, self.password), - ) - self.assertEqual(response.status_code, 200) - reg_info = response.json() - print(f"Registration response: {reg_info}") - client_id = reg_info["client_id"] - client_secret = reg_info["client_secret"] + client_id, client_secret = self._register_test_client() # 2. Get token via client_credentials token_url = f"{self.knoxidf_token_url}knoxtoken/api/v1/token" @@ -92,7 +94,7 @@ class TestKnoxIDF(unittest.TestCase): # ClientCredentialsResource uses Basic Auth for client authentication response = knox_post(token_url, data=data, verify=False) if response.status_code != 200: - print(f"Token error response: {response.text}") + print(f"Token error response: {response.text}") self.assertEqual(response.status_code, 200) tokens = response.json() self.assertIn("access_token", tokens) @@ -103,25 +105,9 @@ class TestKnoxIDF(unittest.TestCase): Test OIDC Authorization Code Flow with Refresh Token. """ # 1. Register client - reg_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/client/register" - print(f"Registering client at: {reg_url}") - data = { - "redirect_uris": "http://localhost/callback";, - "allowed_scopes": "openid,profile,email,offline_access" - } - response = knox_post( - reg_url, - data=data, - auth=HTTPBasicAuth(self.username, self.password), - ) - self.assertEqual(response.status_code, 200) - reg_info = response.json() - print(f"Registration response: {reg_info}") - client_id = reg_info["client_id"] - client_secret = reg_info["client_secret"] + client_id, client_secret = self._register_test_client() # 2. Authorize (with Basic Auth for the user 'guest') - auth_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" params = { "response_type": "code", "client_id": client_id, @@ -130,22 +116,7 @@ class TestKnoxIDF(unittest.TestCase): "state": "test_state", "auto_consent": "true" } - print(f"Authorizing at: {auth_url}") - # allow_redirects=False to catch the redirect to redirect_uri - response = knox_get(auth_url, params=params, auth=(self.username, self.password), verify=False, allow_redirects=False) - - # Should be a redirect to the callback URL - self.assertEqual(response.status_code, 303) - location = response.headers.get("Location") - self.assertIsNotNone(location) - self.assertTrue(location.startswith("http://localhost/callback";)) - - parsed_url = urlparse(location) - query_params = parse_qs(parsed_url.query) - self.assertIn("code", query_params) - self.assertIn("state", query_params) - self.assertEqual(query_params["state"][0], "test_state") - code = query_params["code"][0] + code = self._authorize_get_code(params, expect_state="test_state") # 3. Exchange code for tokens token_url = f"{self.knoxidf_token_url}knoxidf/api/v1/token" @@ -159,15 +130,14 @@ class TestKnoxIDF(unittest.TestCase): } response = knox_post(token_url, data=data, verify=False) if response.status_code != 200: - print(f"Code exchange error: {response.text}") + print(f"Code exchange error: {response.text}") self.assertEqual(response.status_code, 200) tokens = response.json() self.assertIn("access_token", tokens) self.assertIn("id_token", tokens) self.assertIn("refresh_token", tokens) - - refresh_token = tokens["refresh_token"] + refresh_token = tokens["refresh_token"] print(f"Refresh token: {refresh_token}") refresh_token_id = get_token_claim(refresh_token, 'knox.id') print(f"Refresh token knox.id: {refresh_token_id}") @@ -185,12 +155,12 @@ class TestKnoxIDF(unittest.TestCase): new_tokens = response.json() self.assertIn("access_token", new_tokens) self.assertIn("refresh_token", new_tokens) - + # Verify rotation: new refresh token should be different self.assertNotEqual(refresh_token, new_tokens["refresh_token"]) # 5. Verify old refresh token is invalidated - print(f"Verifying old refresh token is invalidated...") + print("Verifying old refresh token is invalidated...") # Use same data (with old refresh_token) data_old = { "grant_type": "refresh_token", @@ -198,7 +168,12 @@ class TestKnoxIDF(unittest.TestCase): "client_id": client_id, "client_secret": client_secret } - response = knox_post(token_url, data=data_old, verify=False, headers={"Accept": "application/json"}) + response = knox_post( + token_url, + data=data_old, + verify=False, + headers={"Accept": "application/json"}, + ) self.assertEqual(response.status_code, 401) error_info = response.json() display_id = get_token_id_display_text(refresh_token_id) @@ -214,10 +189,9 @@ class TestKnoxIDF(unittest.TestCase): # 2. PKCE Setup code_verifier = "thisshouldbealongandrandomstringthatissecure" - code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).decode().replace('=', '') + code_challenge = self._s256_challenge(code_verifier) # 3. Authorize - auth_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" params = { "response_type": "code", "client_id": client_id, @@ -228,10 +202,7 @@ class TestKnoxIDF(unittest.TestCase): "code_challenge": code_challenge, "code_challenge_method": "S256" } - response = knox_get(auth_url, params=params, auth=(self.username, self.password), allow_redirects=False) - self.assertEqual(response.status_code, 303) - location = response.headers.get("Location") - code = parse_qs(urlparse(location).query)["code"][0] + code = self._authorize_get_code(params) # 4. Token Exchange token_url = f"{self.knoxidf_token_url}knoxidf/api/v1/token" @@ -260,7 +231,6 @@ class TestKnoxIDF(unittest.TestCase): code_challenge = code_verifier # 3. Authorize - auth_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" params = { "response_type": "code", "client_id": client_id, @@ -271,10 +241,7 @@ class TestKnoxIDF(unittest.TestCase): "code_challenge": code_challenge, "code_challenge_method": "plain" } - response = knox_get(auth_url, params=params, auth=(self.username, self.password), allow_redirects=False) - self.assertEqual(response.status_code, 303) - location = response.headers.get("Location") - code = parse_qs(urlparse(location).query)["code"][0] + code = self._authorize_get_code(params) # 4. Token Exchange token_url = f"{self.knoxidf_token_url}knoxidf/api/v1/token" @@ -297,37 +264,46 @@ class TestKnoxIDF(unittest.TestCase): """ client_id, client_secret = self._register_test_client() code_verifier = "correct-verifier" - code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).decode().replace('=', '') + code_challenge = self._s256_challenge(code_verifier) # Authorize - auth_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" params = { - "response_type": "code", "client_id": client_id, "redirect_uri": "http://localhost/callback";, - "scope": "openid", "state": "pkce_fail", "auto_consent": "true", - "code_challenge": code_challenge, "code_challenge_method": "S256" + "response_type": "code", + "client_id": client_id, + "redirect_uri": "http://localhost/callback";, + "scope": "openid", + "state": "pkce_fail", + "auto_consent": "true", + "code_challenge": code_challenge, + "code_challenge_method": "S256" } - response = knox_get(auth_url, params=params, auth=(self.username, self.password), allow_redirects=False) - code = parse_qs(urlparse(response.headers.get("Location")).query)["code"][0] + code = self._authorize_get_code(params) token_url = f"{self.knoxidf_token_url}knoxidf/api/v1/token" # 1. Invalid verifier data = { - "grant_type": "authorization_code", "code": code, "redirect_uri": "http://localhost/callback";, - "client_id": client_id, "client_secret": client_secret, "code_verifier": "wrong-verifier" + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "http://localhost/callback";, + "client_id": client_id, + "client_secret": client_secret, + "code_verifier": "wrong-verifier" } response = knox_post(token_url, data=data) self.assertEqual(response.status_code, 401) self.assertIn("Invalid code_verifier", response.json()["error_description"]) # Note: the code is revoked after first use, so we need a new one for the next test - response = knox_get(auth_url, params=params, auth=(self.username, self.password), allow_redirects=False) - code = parse_qs(urlparse(response.headers.get("Location")).query)["code"][0] + code = self._authorize_get_code(params) # 2. Missing verifier data = { - "grant_type": "authorization_code", "code": code, "redirect_uri": "http://localhost/callback";, - "client_id": client_id, "client_secret": client_secret + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "http://localhost/callback";, + "client_id": client_id, + "client_secret": client_secret } response = knox_post(token_url, data=data) self.assertEqual(response.status_code, 401) @@ -335,10 +311,46 @@ class TestKnoxIDF(unittest.TestCase): def _register_test_client(self): reg_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/client/register" - data = {"redirect_uris": "http://localhost/callback";, "allowed_scopes": "openid,profile,email,offline_access"} + print(f"Registering client at: {reg_url}") + data = { + "redirect_uris": "http://localhost/callback";, + "allowed_scopes": "openid,profile,email,offline_access" + } response = knox_post(reg_url, data=data, auth=HTTPBasicAuth(self.username, self.password)) + self.assertEqual(response.status_code, 200) reg_info = response.json() + print(f"Registration response: {reg_info}") return reg_info["client_id"], reg_info["client_secret"] + def _authorize_get_code(self, params, expect_state=None): + """Hit the authorize endpoint and return the code from the redirect Location.""" + auth_url = f"{self.knoxidf_ldap_url}knoxidf/api/v1/authorize" + print(f"Authorizing at: {auth_url}") + # allow_redirects=False to catch the redirect to redirect_uri + response = knox_get( + auth_url, + params=params, + auth=(self.username, self.password), + verify=False, + allow_redirects=False, + ) + self.assertEqual(response.status_code, 303) + location = response.headers.get("Location") + self.assertIsNotNone(location) + self.assertTrue(location.startswith("http://localhost/callback";)) + + query_params = parse_qs(urlparse(location).query) + self.assertIn("code", query_params) + if expect_state is not None: + self.assertIn("state", query_params) + self.assertEqual(query_params["state"][0], expect_state) + return query_params["code"][0] + + @staticmethod + def _s256_challenge(code_verifier): + digest = hashlib.sha256(code_verifier.encode()).digest() + return base64.urlsafe_b64encode(digest).decode().replace('=', '') + + if __name__ == '__main__': unittest.main()
