From 4fa6dfd0b9a3b931b45ccb3a4724750d40814778 Mon Sep 17 00:00:00 2001 From: Revant Nandgaonkar Date: Fri, 7 Nov 2025 08:59:47 +0530 Subject: [PATCH 1/3] tests: add tests for auth.py --- .pre-commit-config.yaml | 2 + castlecraft/auth.py | 80 +++--- castlecraft/hooks.py | 6 +- castlecraft/services/oauth2.py | 4 +- pyproject.toml | 9 +- release.py | 6 +- tests/__init__.py | 0 tests/test_auth.py | 488 +++++++++++++++++++++++++++++++++ 8 files changed, 539 insertions(+), 56 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/test_auth.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 404288c..d72ed6d 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,6 +24,8 @@ repos: rev: 7.1.1 hooks: - id: flake8 + args: + - '--max-line-length=120' additional_dependencies: - 'flake8-print==4.0.1' diff --git a/castlecraft/auth.py b/castlecraft/auth.py index 401e78c..04d5fe1 100644 --- a/castlecraft/auth.py +++ b/castlecraft/auth.py @@ -19,9 +19,7 @@ def validate(): if not idp: return - authorization_header = frappe.get_request_header( # noqa: E501 - "Authorization", "" - ).split(" ") + authorization_header = frappe.get_request_header("Authorization", "").split(" ") if len(authorization_header) == 2: token = authorization_header[1] @@ -48,11 +46,7 @@ def validate_bearer_with_introspection(token, idp): if cached_token: token_json = json.loads(cached_token) exp = token_json.get("exp") - email = frappe.get_value( - "User", - token_json.get(idp.email_key), - "email", - ) + email = token_json.get(idp.email_key) if exp: exp = datetime.datetime.fromtimestamp( int( @@ -91,52 +85,50 @@ def validate_bearer_with_introspection(token, idp): ) token_response = r.json() exp = token_response.get("exp") - email = frappe.get_value( - "User", - token_response.get(idp.email_key), - "email", - ) + email_from_token = token_response.get(idp.email_key) + if exp: exp = datetime.datetime.fromtimestamp( - int( - token_response.get("exp"), - ) + int(token_response.get("exp")), ) else: exp = now + datetime.timedelta( 0, int(token_response.get("expires_in")) or 0 ) - if now < exp: - email = frappe.get_value( - "User", - token_response.get(idp.email_key), - "email", - ) - if email and token_response.get(idp.email_key): + # Token is valid if it's active and not expired. + # It may or may not have an email at this stage. + if now < exp and token_response.get("active"): + email = None + + # If we have an email, check if the user exists. + user_exists = frappe.db.exists("User", email_from_token) + + if user_exists: + email = email_from_token + cache_bearer_token(token, token_response, exp, now) + cache_user_from_sub( + token_response.get("sub"), + json.dumps({"email": email, "token": token}), + ) + is_valid = True + # If user doesn't exist (or no email in token), + # check if we can create one. + elif idp.create_user: + # User does not exist, but we can create them. + user_data = token_response + if idp.fetch_user_info: + if idp.profile_endpoint: + user_data = request_user_info(token, idp) + + user = create_and_save_user(user_data, idp) + email = user.email cache_bearer_token(token, token_response, exp, now) cache_user_from_sub( token_response.get("sub"), json.dumps({"email": email, "token": token}), ) is_valid = True - - if idp.create_user and not frappe.db.exists("User", email): - user_data = token_response - - if idp.fetch_user_info: - if not idp.profile_endpoint: - return - user_data = request_user_info(token, idp) - - user = create_and_save_user(user_data, idp) - email = user.email - cache_bearer_token(token, token_response, exp, now) - cache_user_from_sub( - token_response.get("sub"), - json.dumps({"email": email, "token": token}), - ) - is_valid = True if is_valid: frappe.set_user(email) @@ -292,9 +284,7 @@ def get_padded_b64str(b64string): def get_b64_decoded_json(b64str): - return json.loads( - base64.b64decode(get_padded_b64str(b64str)).decode("utf-8") - ) # noqa: E501 + return json.loads(base64.b64decode(get_padded_b64str(b64str)).decode("utf-8")) def validate_signature(token, idp=None): @@ -306,9 +296,7 @@ def validate_signature(token, idp=None): public_keys = {} for jwk in keys: kid = jwk["kid"] - public_keys[kid] = jwt.algorithms.RSAAlgorithm.from_jwk( - json.dumps(jwk) - ) # noqa: E501 + public_keys[kid] = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk)) kid = jwt.get_unverified_header(token)["kid"] key = public_keys[kid] diff --git a/castlecraft/hooks.py b/castlecraft/hooks.py index ef531d7..46016ca 100644 --- a/castlecraft/hooks.py +++ b/castlecraft/hooks.py @@ -13,13 +13,13 @@ auth_hooks = ["castlecraft.auth.validate"] has_permission = { - "CFE User Claim": "castlecraft.castlecraft.doctype.cfe_user_claim.cfe_user_claim.has_permission", # noqa: E501 + "CFE User Claim": "castlecraft.castlecraft.doctype.cfe_user_claim.cfe_user_claim.has_permission", } override_whitelisted_methods = { - "frappe.integrations.oauth2.openid_profile": "castlecraft.services.oauth2.openid_profile", # noqa: E501 + "frappe.integrations.oauth2.openid_profile": "castlecraft.services.oauth2.openid_profile", } permission_query_conditions = { - "CFE User Claim": "castlecraft.castlecraft.doctype.cfe_user_claim.cfe_user_claim.get_permission_query_conditions", # noqa: E501 + "CFE User Claim": "castlecraft.castlecraft.doctype.cfe_user_claim.cfe_user_claim.get_permission_query_conditions", } diff --git a/castlecraft/services/oauth2.py b/castlecraft/services/oauth2.py index 6721ce8..8f47bf2 100644 --- a/castlecraft/services/oauth2.py +++ b/castlecraft/services/oauth2.py @@ -75,9 +75,7 @@ def sync_claims(): """ - authorization_header = frappe.get_request_header( # noqa: E501 - "Authorization", "" - ).split(" ") + authorization_header = frappe.get_request_header("Authorization", "").split(" ") if len(authorization_header) == 2: token = authorization_header[1] diff --git a/pyproject.toml b/pyproject.toml index 921cf6d..20f0fc9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,5 +16,12 @@ build-backend = "flit_core.buildapi" [project.optional-dependencies] ci = [ - "pre-commit==3.0.3", + "pre-commit>=3.7.0", + "pytest==8.4.2", +] + +[tool.pytest.ini_options] +filterwarnings = [ + "ignore:datetime.datetime.utcfromtimestamp\\(\\) is deprecated:DeprecationWarning:pytz", + "ignore:datetime.datetime.utcfromtimestamp\\(\\) is deprecated:DeprecationWarning:dateutil", ] diff --git a/release.py b/release.py index 8a9c2e7..4ee39a1 100644 --- a/release.py +++ b/release.py @@ -63,11 +63,11 @@ def main(): def increment_semver(version: str, incr_type: SemVerType): current_version = Version(version) if incr_type == "major": - return f"{current_version.major + 1}.0.0" # noqa: E501 + return f"{current_version.major + 1}.0.0" elif incr_type == "minor": - return f"{current_version.major}.{current_version.minor + 1}.0" # noqa: E501 + return f"{current_version.major}.{current_version.minor + 1}.0" elif incr_type == "micro": - return f"{current_version.major}.{current_version.minor}.{current_version.micro + 1}" # noqa: E501 + return f"{current_version.major}.{current_version.minor}.{current_version.micro + 1}" def get_args_parser(): diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..d9d628c --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,488 @@ +import datetime +import json +import time +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth + + +class MockDoc(dict): + """A mock object that simulates a Frappe DocType using a dictionary.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.__dict__ = self + + def get_password(self, key): + return self.get(key) + + +class TestAuth(unittest.TestCase): + def setUp(self): + # This will hold mock objects for patching + self.mocks = {} + + # Common test data + self.test_user_email = "test@example.com" + self.access_token = "valid-token-123" + + # Mock IDP configuration for Introspection + self.mock_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + idp_name="test-idp", + enabled=1, + authorization_type="Introspection", + email_key="email", + introspection_endpoint="https://idp.example.com/introspect", + token_key="token", + auth_header_enabled=1, + client_id="test-client", + client_secret="test-secret", + create_user=False, + fetch_user_info=False, + profile_endpoint="https://idp.example.com/userinfo", + user_roles=[], + ) + ) + + # Mock IDP configuration for JWT Verification + self.mock_jwt_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + idp_name="jwt-idp", + enabled=1, + authorization_type="JWT Verification", + email_key="email", + create_user=False, + jwks_endpoint="https://idp.example.com/.well-known/jwks.json", + allowed_audience=[MockDoc(dict(aud="test-audience"))], + ) + ) + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_valid_uncached_token_existing_user( + self, mock_requests, mock_frappe, mock_get_idp + ): + """ + Test `validate_bearer_with_introspection` with a valid, + uncached token for an existing user. + """ + # --- Arrange --- + + # 1. Mock the main `validate` function's dependencies + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + + # 2. Mock introspection-specific dependencies + # Simulate cache miss + mock_frappe.cache().get_value.return_value = None + # Simulate existing user using the correct method + mock_frappe.db.exists.return_value = True + + # 3. Mock the external HTTP request to the introspection endpoint + mock_response = MagicMock() + mock_response.status_code = 200 + introspection_payload = { + "active": True, + "email": self.test_user_email, + "sub": "user-subject-123", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + + # Assert that an HTTP POST request was + # made to the introspection endpoint + mock_requests.post.assert_called_once_with( + self.mock_idp.introspection_endpoint, + data={"token": self.access_token}, + # HTTPBasicAuth is tricky to assert directly + auth=unittest.mock.ANY, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + # Assert that the user session was set in Frappe + mock_frappe.set_user.assert_called_once_with(self.test_user_email) + + # Assert that the token was cached after successful validation + mock_frappe.cache().set_value.assert_called() + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_valid_cached_token( + self, mock_requests, mock_frappe, mock_get_idp + ): + """ + Test introspection with a valid, + cached token to ensure it bypasses HTTP requests. + """ + # --- Arrange --- + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + + # Simulate cache hit with a valid, non-expired token payload + cached_payload = { + "active": True, + "email": self.test_user_email, + "sub": "user-subject-123", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_frappe.cache().get_value.return_value = json.dumps(cached_payload) + # This test should also rely on db.exists for consistency + mock_frappe.db.exists.return_value = True + + # --- Act --- + auth.validate() + + # --- Assert --- + # Assert that NO external HTTP request was made + mock_requests.post.assert_not_called() + + # Assert that the user session was set + mock_frappe.set_user.assert_called_once_with(self.test_user_email) + + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_new_user_creation( + self, mock_requests, mock_frappe, mock_get_idp, mock_create_user + ): + """ + Test introspection with a valid token for a + non-existent user when `create_user` is enabled. + """ + # --- Arrange --- + self.mock_idp.create_user = True + new_user_email = "new.user@example.com" + + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + + # Simulate cache miss and non-existent user + mock_frappe.cache().get_value.return_value = None + mock_frappe.db.exists.return_value = False + + # Mock introspection response + mock_response = MagicMock() + introspection_payload = { + "active": True, + "email": new_user_email, + "sub": "new-user-sub-456", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # Mock the user creation function to return a mock user + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + mock_create_user.assert_called_once_with(introspection_payload, self.mock_idp) + mock_frappe.set_user.assert_called_once_with(new_user_email) + + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_with_fetch_user_info( + self, mock_requests, mock_frappe, mock_get_idp, mock_create_user + ): + """Test introspection with `fetch_user_info` enabled for a new user.""" + # --- Arrange --- + self.mock_idp.create_user = True + self.mock_idp.fetch_user_info = True + new_user_email = "another.new.user@example.com" + + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + mock_frappe.cache().get_value.return_value = None + mock_frappe.db.exists.return_value = False + + # Mock introspection response (POST) + introspection_response = MagicMock() + introspection_response.json.return_value = { + "active": True, + "sub": "another-sub-789", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + + # Mock userinfo response (GET) + userinfo_response = MagicMock() + userinfo_payload = { + "email": new_user_email, + "given_name": "Another", + "family_name": "User", + } + userinfo_response.json.return_value = userinfo_payload + + # `requests` will be called twice: once for introspect (POST), once for userinfo (GET) + mock_requests.post.return_value = introspection_response + mock_requests.get.return_value = userinfo_response + + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once_with( + self.mock_idp.introspection_endpoint, + data=unittest.mock.ANY, + auth=unittest.mock.ANY, + headers=unittest.mock.ANY, + ) + mock_requests.get.assert_called_once_with( + self.mock_idp.profile_endpoint, + headers={"Authorization": f"Bearer {self.access_token}"}, + ) + # User creation should be called with the data from the userinfo endpoint + mock_create_user.assert_called_once_with(userinfo_payload, self.mock_idp) + mock_frappe.set_user.assert_called_once_with(new_user_email) + + @patch("castlecraft.auth.validate_signature") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + def test_jwt_verification_existing_user( + self, mock_frappe, mock_get_idp, mock_validate_signature + ): + """Test JWT verification with a valid token for an existing user.""" + # --- Arrange --- + jwt_token = "a.b.c" + mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + + # Simulate cache miss and existing user + mock_frappe.cache().get_value.return_value = None + mock_frappe.get_value.return_value = self.test_user_email + + # Mock the b64decode and signature validation + with patch("castlecraft.auth.get_b64_decoded_json") as mock_b64_decode: + decoded_payload = { + "email": self.test_user_email, + "exp": time.time() + 3600, + "aud": "test-audience", + "sub": "jwt-user-sub", + } + mock_b64_decode.return_value = decoded_payload + mock_validate_signature.return_value = decoded_payload + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_validate_signature.assert_called_once_with(jwt_token, self.mock_jwt_idp) + mock_frappe.set_user.assert_called_once_with(self.test_user_email) + mock_frappe.cache().set_value.assert_called() + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_inactive_token( + self, mock_requests, mock_frappe, mock_get_idp + ): + """Test introspection with a token that is inactive.""" + # --- Arrange --- + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + mock_frappe.cache().get_value.return_value = None + + mock_response = MagicMock() + introspection_payload = {"active": False} + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + mock_frappe.set_user.assert_not_called() + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + @patch("castlecraft.auth.requests") + def test_introspection_new_user_creation_disabled( + self, mock_requests, mock_frappe, mock_get_idp + ): + """Test auth fails for a new user when `create_user` is disabled.""" + # --- Arrange --- + self.mock_idp.create_user = False + mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + mock_frappe.cache().get_value.return_value = None + mock_frappe.db.exists.return_value = False # User does not exist + + mock_response = MagicMock() + introspection_payload = { + "active": True, + "email": "new.user@example.com", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + mock_frappe.set_user.assert_not_called() + + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.validate_signature") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + def test_jwt_verification_new_user_creation( + self, mock_frappe, mock_get_idp, mock_validate_signature, mock_create_user + ): + """Test JWT verification with user creation for a new user.""" + # --- Arrange --- + self.mock_jwt_idp.create_user = True + new_user_email = "new.jwt.user@example.com" + jwt_token = "a.b.c" + + mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + mock_frappe.cache().get_value.return_value = None + mock_frappe.get_value.return_value = None # User does not exist + + decoded_payload = { + "email": new_user_email, + "exp": time.time() + 3600, + "aud": "test-audience", + } + + with patch("castlecraft.auth.get_b64_decoded_json") as mock_b64_decode: + mock_b64_decode.return_value = decoded_payload + mock_validate_signature.return_value = decoded_payload + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_create_user.assert_called_once_with(decoded_payload, self.mock_jwt_idp) + mock_frappe.set_user.assert_called_once_with(new_user_email) + + @patch("castlecraft.auth.validate_signature") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.frappe") + def test_jwt_verification_invalid_signature( + self, mock_frappe, mock_get_idp, mock_validate_signature + ): + """Test JWT verification fails with an invalid signature.""" + # --- Arrange --- + jwt_token = "a.b.c" + mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + mock_frappe.cache().get_value.return_value = None + + # Simulate signature validation failure + mock_validate_signature.side_effect = Exception("Invalid signature") + + with patch("castlecraft.auth.get_b64_decoded_json"): + # --- Act --- + auth.validate() + + # --- Assert --- + mock_frappe.set_user.assert_not_called() + + +class TestAuthHelpers(unittest.TestCase): + @patch("castlecraft.auth.frappe") + def test_get_idp_with_name(self, mock_frappe): + """Test get_idp fetches a named IDP.""" + # --- Arrange --- + idp_name = "specific-idp" + + # --- Act --- + auth.get_idp(idp_name) + + # --- Assert --- + mock_frappe.get_cached_doc.assert_called_once_with( + "CFE Identity Provider", idp_name + ) + mock_frappe.get_last_doc.assert_not_called() + + @patch("castlecraft.auth.frappe") + def test_get_idp_default(self, mock_frappe): + """Test get_idp fetches the default (last) IDP when no name is given.""" + # --- Arrange --- + # --- Act --- + auth.get_idp() + + # --- Assert --- + mock_frappe.get_last_doc.assert_called_once_with( + "CFE Identity Provider", filters={"enabled": 1} + ) + mock_frappe.get_cached_doc.assert_not_called() + + @patch("castlecraft.auth.frappe") + def test_create_and_save_user(self, mock_frappe): + """Test create_and_save_user correctly maps claims and roles.""" + # --- Arrange --- + idp_config = MockDoc( + dict( + email_key="email_address", + first_name_key="given_name", + full_name_key="name", + user_roles=[ + MockDoc(dict(role="Blogger")), + MockDoc(dict(role="Website Manager")), + ], + user_fields=[MockDoc(dict(claim="custom_claim"))], + ) + ) + token_payload = { + "email_address": "test.user@example.com", + "given_name": "Test", + "name": "Test User", + "custom_claim": "custom_value", + } + + mock_user_doc = MagicMock() + mock_claims_doc = MagicMock() + mock_frappe.new_doc.return_value = mock_user_doc + mock_frappe.get_doc.return_value = mock_claims_doc + # Simulate that the roles exist in the DB + mock_frappe.db.get_value.return_value = True + + # --- Act --- + created_user = auth.create_and_save_user(token_payload, idp_config) + + # --- Assert --- + self.assertEqual(created_user, mock_user_doc) + self.assertEqual(mock_user_doc.email, "test.user@example.com") + self.assertEqual(mock_user_doc.first_name, "Test") + self.assertEqual(mock_user_doc.full_name, "Test User") + + # Check that roles were appended + self.assertEqual(mock_user_doc.append.call_count, 2) + mock_user_doc.append.assert_any_call("roles", {"role": "Blogger"}) + mock_user_doc.append.assert_any_call("roles", {"role": "Website Manager"}) + + # Check that custom claims were appended + mock_claims_doc.append.assert_called_once_with( + "claims", {"claim": "custom_claim", "value": "custom_value"} + ) + mock_user_doc.save.assert_called_once() + mock_claims_doc.save.assert_called_once() + mock_frappe.db.commit.assert_called_once() From ac9d3728393f2c6c33a97dcde95f2bda1a2e66fb Mon Sep 17 00:00:00 2001 From: Revant Nandgaonkar Date: Fri, 7 Nov 2025 22:15:00 +0530 Subject: [PATCH 2/3] feat: Refactor auth.py for robust JWT validation, caching, and testability This commit introduces significant improvements to the authentication module (`auth.py`) and its testing infrastructure. Key changes include: - **Enhanced JWT Verification**: Implemented proper caching for validated JWT payloads, ensuring performance for repeated token usage. Corrected `jwt.decode` input handling and added support for configurable `audience_claim_key` in `validate_signature`. - **Refined Authentication Flows**: Streamlined user data processing in both `validate_bearer_with_introspection` and `validate_bearer_with_jwt_verification`, especially when `fetch_user_info` is enabled. - **Improved Testability**: The monolithic `tests/test_auth.py` has been split into granular, focused test files (`test_auth_helpers.py`, `test_auth_introspection.py`, `test_jwt_existing_user.py`, `test_jwt_invalid.py`, `test_jwt_new_user.py`, `test_jwt_userinfo.py`). - **Robust Mocking**: Introduced `tests/conftest.py` with `MockDoc` and an `autouse` `mock_frappe` fixture to provide consistent and isolated testing environments. - **DocType Updates**: Added `audience_claim_key` to the `CFE Identity Provider` DocType to support flexible audience validation. These changes address previous logical flaws, enhance the reliability and testability of the authentication system, and lay the groundwork for future feature development. --- TODO.md | 96 ++++ castlecraft/auth.py | 149 +++--- .../cfe_identity_provider.json | 19 +- pyproject.toml | 1 + tests/conftest.py | 28 + tests/test_auth.py | 488 ------------------ tests/test_auth_helpers.py | 84 +++ tests/test_auth_introspection.py | 238 +++++++++ tests/test_jwt_existing_user.py | 66 +++ tests/test_jwt_invalid.py | 100 ++++ tests/test_jwt_new_user.py | 81 +++ tests/test_jwt_userinfo.py | 133 +++++ 12 files changed, 931 insertions(+), 552 deletions(-) create mode 100644 TODO.md create mode 100644 tests/conftest.py delete mode 100644 tests/test_auth.py create mode 100644 tests/test_auth_helpers.py create mode 100644 tests/test_auth_introspection.py create mode 100644 tests/test_jwt_existing_user.py create mode 100644 tests/test_jwt_invalid.py create mode 100644 tests/test_jwt_new_user.py create mode 100644 tests/test_jwt_userinfo.py diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..cb550fb --- /dev/null +++ b/TODO.md @@ -0,0 +1,96 @@ +# Project TODO: Enhance `auth.py` with Testing and New Features + +This document outlines the steps to improve the `castlecraft/auth.py` module by adding comprehensive test coverage and implementing a new hybrid authentication flow using Test-Driven Development (TDD). + +## Phase 1: Establish Test Coverage for Existing `auth.py` + +The goal of this phase is to ensure the current authentication logic is fully tested and stable before introducing changes. We will use `pytest` and `unittest.mock` to isolate the code from the Frappe framework and external services. + +- [x] **Setup Test Environment** (Complete) + - [x] Create a `tests` directory within the `castlecraft` app if it doesn't exist. + - [x] Create a test file, e.g., `tests/test_auth.py`. + - [x] Ensure `pytest` is installed and configured in the development environment. + +- [x] **Write Tests for `validate_bearer_with_introspection()`** (Complete) + - [x] **Mock Dependencies:** + - [x] Mock `frappe` calls: `frappe.cache()`, `frappe.get_value`, `frappe.db.exists`, `frappe.set_user`, `frappe.get_conf`, `frappe.log_error`, `frappe.new_doc`, `frappe.get_doc`. + - [x] Mock `requests.post` to simulate calls to the introspection endpoint. + - [x] **Test Scenarios:** + - [x] Test with a valid, uncached token that successfully introspects. + - [x] Test with a valid, cached token to ensure it bypasses the HTTP request. + - [x] Test with an invalid or expired token. + - [x] Test with a valid token for a user that does not exist in Frappe (`create_user` is enabled). + - [x] Verify `create_and_save_user` is called. + - [x] Verify user is created with correct roles and claims. + - [x] Test with `fetch_user_info` enabled, ensuring `request_user_info` is called. + - [x] Test with a valid token but `create_user` is disabled for a non-existent user (should fail authentication). + - [x] Test failure when the introspection endpoint is not configured. + +- [x] **Write Tests for `validate_bearer_with_jwt_verification()`** (Complete) + - [x] **Mock Dependencies:** + - [x] Mock `frappe` calls (similar to introspection tests). + - [x] Mock `requests.get` to simulate fetching keys from the JWKS endpoint. + - [x] Mock `jwt.decode` and related functions if needed, or provide mock keys for real decoding. + - [x] **Test Scenarios:** + - [x] Test with a valid, signed JWT for an existing user. + - [x] Test with a JWT that has an invalid signature or incorrect `aud` (audience). + - [x] Test with an expired JWT. + - [x] Test with a valid JWT for a user that does not exist (`create_user` is enabled). + - [x] Test with a valid, cached JWT to ensure it bypasses full validation. + +- [x] **Write Tests for Helper Functions** (Complete) + - [x] Test `get_idp()` to ensure it correctly fetches the named IDP or the default one. + - [x] Test `create_and_save_user()` to verify it correctly maps claims to the `User` and `CFE User Claim` doctypes. + +## Phase 2: TDD for Enhanced JWT Verification Flow + +This phase focuses on enhancing the existing "JWT Verification" flow by: + +1. Using the `fetch_user_info` checkbox to optionally fetch user details from a `profile_endpoint` after validating the JWT. +2. Making the JWT audience claim (`aud`) configurable to support providers like AWS Cognito which may use a different claim (e.g., `client_id`). + +- [x] **Step 1: Write the Failing Tests** + - [x] In `tests/test_auth.py`, add new tests for the enhanced JWT flow. + - [x] **Test Scenario 1: JWT with User Info Fetching** + - [x] Configure a mock IDP with `authorization_type` as `JWT Verification` and `fetch_user_info` enabled. + - [x] Create a valid JWT that might not contain an email claim. + - [x] Mock the JWKS endpoint for signature validation. + - [x] Mock the `profile_endpoint` to return a JSON payload with user details (including email). + - [x] Call `auth.validate()` and assert that `validate_signature` is called, followed by `request_user_info`. + - [x] Assert that `frappe.set_user` is called with the email from the userinfo response. + - [x] Add a similar test for new user creation, asserting that `create_and_save_user` is called with the userinfo payload. + - [x] **Test Scenario 2: Configurable Audience Claim** + - [x] Configure a mock IDP with a custom `audience_claim_key` (e.g., `client_id`). + - [x] Create a JWT that uses `client_id` for audience instead of `aud`. + - [x] Assert that `validate_signature` correctly validates the token. + - [x] **Run the tests and confirm they fail** as the implementation is not yet updated. + +- [x] **Step 2: Implement the Feature** + - [x] In `auth.py`, modify `validate_bearer_with_jwt_verification(token, idp)`: + - [x] After `validate_signature` succeeds, check if `idp.fetch_user_info` is true and `idp.profile_endpoint` is set. + - [x] If so, call `request_user_info(token, idp)` and use its response as the primary source for `user_data`. Otherwise, use the JWT payload. + - [x] Base the user lookup and creation logic on this `user_data`. + - [x] In `auth.py`, modify `validate_signature(token, idp)`: + - [x] Read a new `audience_claim_key` field from the IDP, defaulting to `"aud"`. + - [x] When calling `jwt.decode`, adjust the parameters to handle audience validation based on this configurable key. + - [x] Add the `audience_claim_key` field to the `CFE Identity Provider` DocType. + +- [x] **Step 3: Make the Tests Pass** + - [x] Run the test suite again. + - [x] Debug and refine the implementation until the new test case passes. + +- [ ] **Step 4: Refactor and Add More Tests** + - [x] Refactor the code for clarity and to remove duplication. Consider if `validate_bearer_with_introspection` and the new function can share logic for user lookup, creation, and caching. + - [ ] Add edge-case tests: + - [x] An invalid JWT should fail before calling the userinfo endpoint. + - [ ] The flow should fail gracefully if the userinfo endpoint returns an error. + +## Phase 3: Finalization and Documentation + +- [ ] **Review and Merge** + - [ ] Ensure all new and existing tests are passing. + - [ ] Create a Pull/Merge Request with the changes. + - [ ] Conduct a final code review. + +- [ ] **Update Documentation** + - [ ] Update `README.md` or other relevant documentation to explain the new `JWT with Userinfo` authorization type and its configuration. diff --git a/castlecraft/auth.py b/castlecraft/auth.py index 04d5fe1..920ba10 100644 --- a/castlecraft/auth.py +++ b/castlecraft/auth.py @@ -104,23 +104,23 @@ def validate_bearer_with_introspection(token, idp): # If we have an email, check if the user exists. user_exists = frappe.db.exists("User", email_from_token) + # Determine the final user data + user_data = token_response + if idp.fetch_user_info and idp.profile_endpoint: + user_data = request_user_info(token, idp) + if user_exists: email = email_from_token cache_bearer_token(token, token_response, exp, now) cache_user_from_sub( - token_response.get("sub"), + user_data.get("sub"), json.dumps({"email": email, "token": token}), ) is_valid = True # If user doesn't exist (or no email in token), # check if we can create one. elif idp.create_user: - # User does not exist, but we can create them. - user_data = token_response - if idp.fetch_user_info: - if idp.profile_endpoint: - user_data = request_user_info(token, idp) - + # User does not exist, create them using the final user_data user = create_and_save_user(user_data, idp) email = user.email cache_bearer_token(token, token_response, exp, now) @@ -143,64 +143,79 @@ def validate_bearer_with_introspection(token, idp): def validate_bearer_with_jwt_verification(token, idp): - is_valid = False try: form_dict = frappe.local.form_dict + is_valid = False + final_email = None + payload = None + + # 1. Check for a cached, validated payload using the token itself as the key. + cached_payload_str = frappe.cache().get_value(f"cc_jwt_payload|{token}") now = datetime.datetime.now() - b64_jwt_header, b64_jwt_body, b64_jwt_signature = token.split(".") - body = get_b64_decoded_json(b64_jwt_body) - email = frappe.get_value( - "User", - body.get(idp.email_key, "email"), - "email", - ) - cached_token = get_cached_jwt(email) - - if cached_token and cached_token == token: - ( - cached_b64_jwt_header, - cached_b64_jwt_body, - cached_b64_jwt_signature, - ) = cached_token.split(".") - body = get_b64_decoded_json(cached_b64_jwt_body) - exp = datetime.datetime.fromtimestamp(int(body.get("exp"))) - is_valid = True if now < exp else False - - if not is_valid: - delete_cached_jwt(email) + + if cached_payload_str: + cached_payload = json.loads(cached_payload_str) + exp = cached_payload.get("exp") + if exp and now < datetime.datetime.fromtimestamp(int(exp)): + # Cache hit and token is not expired. + payload = cached_payload + else: + # Token is expired, remove from cache. + frappe.cache().delete_key(f"cc_jwt_payload|{token}") + + # For JWT flows, always validate the signature first. + if not payload: + # 2. Cache miss or expired, perform full validation. payload = validate_signature(token, idp) - if email: - cache_jwt( - email, - token, - datetime.datetime.fromtimestamp(int(body.get("exp"))), - now, - ) - cache_user_from_sub( - payload.get("sub"), - json.dumps({"email": email, "token": token}), - ) - is_valid = True + # 3. If fetch_user_info is enabled, call the userinfo endpoint to get the final user data. + if idp.fetch_user_info and idp.profile_endpoint: + user_data = request_user_info(token, idp) + else: + user_data = payload + + email_from_payload = user_data.get(idp.email_key) + if not email_from_payload: + # If we can't get an email, we cannot proceed. + return + + user_email = frappe.get_value("User", email_from_payload, "email") + + if user_email: + # User exists, log them in. + final_email = user_email + is_valid = True + elif idp.create_user: + # User does not exist, but we are allowed to create them. + user = create_and_save_user(user_data, idp) + final_email = user.email + is_valid = True + else: + # User does not exist, and we are not allowed to create them. + is_valid = False + final_email = None + + if is_valid and final_email: + frappe.set_user(final_email) + frappe.local.form_dict = form_dict - elif idp.create_user and body.get(idp.email_key, "email"): - user = create_and_save_user(body, idp) - email = user.email - cache_jwt( - email, - token, - datetime.datetime.fromtimestamp(int(payload.get("exp"))), - now, + # 4. Cache the newly validated payload and other user details. + if payload.get("exp"): + # Cache the payload against the token for the "fast path". + frappe.cache().set_value( + f"cc_jwt_payload|{token}", + json.dumps(payload), + expires_in_sec=datetime.datetime.fromtimestamp( + int(payload.get("exp")) + ) + - now, ) + + if payload.get("sub"): cache_user_from_sub( payload.get("sub"), - json.dumps({"email": email, "token": token}), + json.dumps({"email": final_email, "token": token}), ) - is_valid = True - - if is_valid: - frappe.set_user(email) - frappe.local.form_dict = form_dict except Exception: if frappe.get_conf().get("castlecraft_enable_log"): @@ -290,6 +305,7 @@ def get_b64_decoded_json(b64str): def validate_signature(token, idp=None): idp = idp or get_idp() allowed_audience = [audience.aud for audience in idp.allowed_audience] + audience_claim_key = idp.audience_claim_key or "aud" r = requests.get(idp.jwks_endpoint) jwks_keys = r.json() keys = jwks_keys.get("keys") @@ -298,16 +314,31 @@ def validate_signature(token, idp=None): kid = jwk["kid"] public_keys[kid] = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk)) - kid = jwt.get_unverified_header(token)["kid"] + kid = jwt.get_unverified_header(token.encode("utf-8"))["kid"] key = public_keys[kid] - return jwt.decode( - get_padded_b64str(token), + payload = jwt.decode( + token, # The full JWT token string should be passed directly to jwt.decode key=key, algorithms=["RS256"], - audience=allowed_audience, + audience=allowed_audience if audience_claim_key == "aud" else None, ) + if audience_claim_key != "aud": + claim_value = payload.get(audience_claim_key) + if not claim_value: + raise jwt.MissingRequiredClaimError(audience_claim_key) + + if isinstance(claim_value, str): + claim_value = [claim_value] + + if not any(c in allowed_audience for c in claim_value): + raise jwt.InvalidAudienceError( + f"Invalid audience. Expected one of {allowed_audience}" + ) + + return payload + def get_cached_bearer_token(token: str): return frappe.cache().get_value(f"cc_bearer|{token}") diff --git a/castlecraft/castlecraft/doctype/cfe_identity_provider/cfe_identity_provider.json b/castlecraft/castlecraft/doctype/cfe_identity_provider/cfe_identity_provider.json index 5886c6f..0465301 100644 --- a/castlecraft/castlecraft/doctype/cfe_identity_provider/cfe_identity_provider.json +++ b/castlecraft/castlecraft/doctype/cfe_identity_provider/cfe_identity_provider.json @@ -17,18 +17,19 @@ "cb_idp_1", "create_user", "user_roles", + "fetch_user_info", + "profile_endpoint", "introspection_config_section", "auth_header_enabled", - "fetch_user_info", + "introspection_endpoint", "cb_ic_0", "client_id", "client_secret", - "token_key", "cb_ic_1", - "introspection_endpoint", - "profile_endpoint", + "token_key", "jwt_verification_config_section", "jwks_endpoint", + "audience_claim_key", "allowed_audience", "user_field_map_section", "user_fields" @@ -120,6 +121,7 @@ "mandatory_depends_on": "eval:doc.authorization_type==='Introspection'" }, { + "depends_on": "eval:doc.fetch_user_info", "fieldname": "profile_endpoint", "fieldtype": "Small Text", "label": "Profile Endpoint", @@ -127,6 +129,7 @@ }, { "collapsible": 1, + "default": "aud", "depends_on": "eval:doc.authorization_type==='JWT Verification'", "fieldname": "jwt_verification_config_section", "fieldtype": "Section Break", @@ -192,11 +195,16 @@ "fieldtype": "Table MultiSelect", "label": "User Roles", "options": "CFE IDP User Role" + }, + { + "fieldname": "audience_claim_key", + "fieldtype": "Data", + "label": "Audience Claim Key" } ], "index_web_pages_for_search": 1, "links": [], - "modified": "2024-07-13 19:42:38.695792", + "modified": "2025-11-07 20:35:31.551416", "modified_by": "Administrator", "module": "Castlecraft", "name": "CFE Identity Provider", @@ -228,6 +236,7 @@ "write": 1 } ], + "row_format": "Dynamic", "sort_field": "modified", "sort_order": "DESC", "states": [] diff --git a/pyproject.toml b/pyproject.toml index 20f0fc9..3ff02a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,7 @@ build-backend = "flit_core.buildapi" ci = [ "pre-commit>=3.7.0", "pytest==8.4.2", + "pytest-mock>=3.14.0", ] [tool.pytest.ini_options] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..660914a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,28 @@ +# /home/revant/Projects/cfe-bench/development/frappe-bench/apps/castlecraft/tests/conftest.py +from unittest.mock import MagicMock + +import pytest + + +class MockDoc(dict): + """A mock object that simulates a Frappe DocType using a dictionary.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.__dict__ = self + + def get_password(self, key): + return self.get(key) + + +@pytest.fixture(autouse=True) +def mock_frappe(mocker): + """An autouse fixture that mocks castlecraft.auth.frappe for all tests.""" + # Patch the frappe module where it's used + mock = mocker.patch("castlecraft.auth.frappe") + # Pre-configure the cache on this mock to ensure test isolation + cache_mock = MagicMock() + cache_mock.get_value.return_value = None + mock.cache.return_value = cache_mock + # The mock is now active for the duration of the test. + # We don't need to return it unless a test specifically needs to inspect it. diff --git a/tests/test_auth.py b/tests/test_auth.py deleted file mode 100644 index d9d628c..0000000 --- a/tests/test_auth.py +++ /dev/null @@ -1,488 +0,0 @@ -import datetime -import json -import time -import unittest -from unittest.mock import MagicMock, patch - -from castlecraft import auth - - -class MockDoc(dict): - """A mock object that simulates a Frappe DocType using a dictionary.""" - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.__dict__ = self - - def get_password(self, key): - return self.get(key) - - -class TestAuth(unittest.TestCase): - def setUp(self): - # This will hold mock objects for patching - self.mocks = {} - - # Common test data - self.test_user_email = "test@example.com" - self.access_token = "valid-token-123" - - # Mock IDP configuration for Introspection - self.mock_idp = MockDoc( - dict( - doctype="CFE Identity Provider", - idp_name="test-idp", - enabled=1, - authorization_type="Introspection", - email_key="email", - introspection_endpoint="https://idp.example.com/introspect", - token_key="token", - auth_header_enabled=1, - client_id="test-client", - client_secret="test-secret", - create_user=False, - fetch_user_info=False, - profile_endpoint="https://idp.example.com/userinfo", - user_roles=[], - ) - ) - - # Mock IDP configuration for JWT Verification - self.mock_jwt_idp = MockDoc( - dict( - doctype="CFE Identity Provider", - idp_name="jwt-idp", - enabled=1, - authorization_type="JWT Verification", - email_key="email", - create_user=False, - jwks_endpoint="https://idp.example.com/.well-known/jwks.json", - allowed_audience=[MockDoc(dict(aud="test-audience"))], - ) - ) - - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_valid_uncached_token_existing_user( - self, mock_requests, mock_frappe, mock_get_idp - ): - """ - Test `validate_bearer_with_introspection` with a valid, - uncached token for an existing user. - """ - # --- Arrange --- - - # 1. Mock the main `validate` function's dependencies - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - - # 2. Mock introspection-specific dependencies - # Simulate cache miss - mock_frappe.cache().get_value.return_value = None - # Simulate existing user using the correct method - mock_frappe.db.exists.return_value = True - - # 3. Mock the external HTTP request to the introspection endpoint - mock_response = MagicMock() - mock_response.status_code = 200 - introspection_payload = { - "active": True, - "email": self.test_user_email, - "sub": "user-subject-123", - "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), - } - mock_response.json.return_value = introspection_payload - mock_requests.post.return_value = mock_response - - # --- Act --- - auth.validate() - - # --- Assert --- - - # Assert that an HTTP POST request was - # made to the introspection endpoint - mock_requests.post.assert_called_once_with( - self.mock_idp.introspection_endpoint, - data={"token": self.access_token}, - # HTTPBasicAuth is tricky to assert directly - auth=unittest.mock.ANY, - headers={"Content-Type": "application/x-www-form-urlencoded"}, - ) - - # Assert that the user session was set in Frappe - mock_frappe.set_user.assert_called_once_with(self.test_user_email) - - # Assert that the token was cached after successful validation - mock_frappe.cache().set_value.assert_called() - - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_valid_cached_token( - self, mock_requests, mock_frappe, mock_get_idp - ): - """ - Test introspection with a valid, - cached token to ensure it bypasses HTTP requests. - """ - # --- Arrange --- - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - - # Simulate cache hit with a valid, non-expired token payload - cached_payload = { - "active": True, - "email": self.test_user_email, - "sub": "user-subject-123", - "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), - } - mock_frappe.cache().get_value.return_value = json.dumps(cached_payload) - # This test should also rely on db.exists for consistency - mock_frappe.db.exists.return_value = True - - # --- Act --- - auth.validate() - - # --- Assert --- - # Assert that NO external HTTP request was made - mock_requests.post.assert_not_called() - - # Assert that the user session was set - mock_frappe.set_user.assert_called_once_with(self.test_user_email) - - @patch("castlecraft.auth.create_and_save_user") - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_new_user_creation( - self, mock_requests, mock_frappe, mock_get_idp, mock_create_user - ): - """ - Test introspection with a valid token for a - non-existent user when `create_user` is enabled. - """ - # --- Arrange --- - self.mock_idp.create_user = True - new_user_email = "new.user@example.com" - - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - - # Simulate cache miss and non-existent user - mock_frappe.cache().get_value.return_value = None - mock_frappe.db.exists.return_value = False - - # Mock introspection response - mock_response = MagicMock() - introspection_payload = { - "active": True, - "email": new_user_email, - "sub": "new-user-sub-456", - "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), - } - mock_response.json.return_value = introspection_payload - mock_requests.post.return_value = mock_response - - # Mock the user creation function to return a mock user - mock_user = MockDoc(dict(email=new_user_email)) - mock_create_user.return_value = mock_user - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_requests.post.assert_called_once() - mock_create_user.assert_called_once_with(introspection_payload, self.mock_idp) - mock_frappe.set_user.assert_called_once_with(new_user_email) - - @patch("castlecraft.auth.create_and_save_user") - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_with_fetch_user_info( - self, mock_requests, mock_frappe, mock_get_idp, mock_create_user - ): - """Test introspection with `fetch_user_info` enabled for a new user.""" - # --- Arrange --- - self.mock_idp.create_user = True - self.mock_idp.fetch_user_info = True - new_user_email = "another.new.user@example.com" - - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - mock_frappe.cache().get_value.return_value = None - mock_frappe.db.exists.return_value = False - - # Mock introspection response (POST) - introspection_response = MagicMock() - introspection_response.json.return_value = { - "active": True, - "sub": "another-sub-789", - "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), - } - - # Mock userinfo response (GET) - userinfo_response = MagicMock() - userinfo_payload = { - "email": new_user_email, - "given_name": "Another", - "family_name": "User", - } - userinfo_response.json.return_value = userinfo_payload - - # `requests` will be called twice: once for introspect (POST), once for userinfo (GET) - mock_requests.post.return_value = introspection_response - mock_requests.get.return_value = userinfo_response - - mock_user = MockDoc(dict(email=new_user_email)) - mock_create_user.return_value = mock_user - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_requests.post.assert_called_once_with( - self.mock_idp.introspection_endpoint, - data=unittest.mock.ANY, - auth=unittest.mock.ANY, - headers=unittest.mock.ANY, - ) - mock_requests.get.assert_called_once_with( - self.mock_idp.profile_endpoint, - headers={"Authorization": f"Bearer {self.access_token}"}, - ) - # User creation should be called with the data from the userinfo endpoint - mock_create_user.assert_called_once_with(userinfo_payload, self.mock_idp) - mock_frappe.set_user.assert_called_once_with(new_user_email) - - @patch("castlecraft.auth.validate_signature") - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - def test_jwt_verification_existing_user( - self, mock_frappe, mock_get_idp, mock_validate_signature - ): - """Test JWT verification with a valid token for an existing user.""" - # --- Arrange --- - jwt_token = "a.b.c" - mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" - mock_get_idp.return_value = self.mock_jwt_idp - - # Simulate cache miss and existing user - mock_frappe.cache().get_value.return_value = None - mock_frappe.get_value.return_value = self.test_user_email - - # Mock the b64decode and signature validation - with patch("castlecraft.auth.get_b64_decoded_json") as mock_b64_decode: - decoded_payload = { - "email": self.test_user_email, - "exp": time.time() + 3600, - "aud": "test-audience", - "sub": "jwt-user-sub", - } - mock_b64_decode.return_value = decoded_payload - mock_validate_signature.return_value = decoded_payload - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_validate_signature.assert_called_once_with(jwt_token, self.mock_jwt_idp) - mock_frappe.set_user.assert_called_once_with(self.test_user_email) - mock_frappe.cache().set_value.assert_called() - - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_inactive_token( - self, mock_requests, mock_frappe, mock_get_idp - ): - """Test introspection with a token that is inactive.""" - # --- Arrange --- - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - mock_frappe.cache().get_value.return_value = None - - mock_response = MagicMock() - introspection_payload = {"active": False} - mock_response.json.return_value = introspection_payload - mock_requests.post.return_value = mock_response - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_requests.post.assert_called_once() - mock_frappe.set_user.assert_not_called() - - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - @patch("castlecraft.auth.requests") - def test_introspection_new_user_creation_disabled( - self, mock_requests, mock_frappe, mock_get_idp - ): - """Test auth fails for a new user when `create_user` is disabled.""" - # --- Arrange --- - self.mock_idp.create_user = False - mock_frappe.get_request_header.return_value = f"Bearer {self.access_token}" - mock_get_idp.return_value = self.mock_idp - mock_frappe.cache().get_value.return_value = None - mock_frappe.db.exists.return_value = False # User does not exist - - mock_response = MagicMock() - introspection_payload = { - "active": True, - "email": "new.user@example.com", - "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), - } - mock_response.json.return_value = introspection_payload - mock_requests.post.return_value = mock_response - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_requests.post.assert_called_once() - mock_frappe.set_user.assert_not_called() - - @patch("castlecraft.auth.create_and_save_user") - @patch("castlecraft.auth.validate_signature") - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - def test_jwt_verification_new_user_creation( - self, mock_frappe, mock_get_idp, mock_validate_signature, mock_create_user - ): - """Test JWT verification with user creation for a new user.""" - # --- Arrange --- - self.mock_jwt_idp.create_user = True - new_user_email = "new.jwt.user@example.com" - jwt_token = "a.b.c" - - mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" - mock_get_idp.return_value = self.mock_jwt_idp - mock_frappe.cache().get_value.return_value = None - mock_frappe.get_value.return_value = None # User does not exist - - decoded_payload = { - "email": new_user_email, - "exp": time.time() + 3600, - "aud": "test-audience", - } - - with patch("castlecraft.auth.get_b64_decoded_json") as mock_b64_decode: - mock_b64_decode.return_value = decoded_payload - mock_validate_signature.return_value = decoded_payload - mock_user = MockDoc(dict(email=new_user_email)) - mock_create_user.return_value = mock_user - - # --- Act --- - auth.validate() - - # --- Assert --- - mock_create_user.assert_called_once_with(decoded_payload, self.mock_jwt_idp) - mock_frappe.set_user.assert_called_once_with(new_user_email) - - @patch("castlecraft.auth.validate_signature") - @patch("castlecraft.auth.get_idp") - @patch("castlecraft.auth.frappe") - def test_jwt_verification_invalid_signature( - self, mock_frappe, mock_get_idp, mock_validate_signature - ): - """Test JWT verification fails with an invalid signature.""" - # --- Arrange --- - jwt_token = "a.b.c" - mock_frappe.get_request_header.return_value = f"Bearer {jwt_token}" - mock_get_idp.return_value = self.mock_jwt_idp - mock_frappe.cache().get_value.return_value = None - - # Simulate signature validation failure - mock_validate_signature.side_effect = Exception("Invalid signature") - - with patch("castlecraft.auth.get_b64_decoded_json"): - # --- Act --- - auth.validate() - - # --- Assert --- - mock_frappe.set_user.assert_not_called() - - -class TestAuthHelpers(unittest.TestCase): - @patch("castlecraft.auth.frappe") - def test_get_idp_with_name(self, mock_frappe): - """Test get_idp fetches a named IDP.""" - # --- Arrange --- - idp_name = "specific-idp" - - # --- Act --- - auth.get_idp(idp_name) - - # --- Assert --- - mock_frappe.get_cached_doc.assert_called_once_with( - "CFE Identity Provider", idp_name - ) - mock_frappe.get_last_doc.assert_not_called() - - @patch("castlecraft.auth.frappe") - def test_get_idp_default(self, mock_frappe): - """Test get_idp fetches the default (last) IDP when no name is given.""" - # --- Arrange --- - # --- Act --- - auth.get_idp() - - # --- Assert --- - mock_frappe.get_last_doc.assert_called_once_with( - "CFE Identity Provider", filters={"enabled": 1} - ) - mock_frappe.get_cached_doc.assert_not_called() - - @patch("castlecraft.auth.frappe") - def test_create_and_save_user(self, mock_frappe): - """Test create_and_save_user correctly maps claims and roles.""" - # --- Arrange --- - idp_config = MockDoc( - dict( - email_key="email_address", - first_name_key="given_name", - full_name_key="name", - user_roles=[ - MockDoc(dict(role="Blogger")), - MockDoc(dict(role="Website Manager")), - ], - user_fields=[MockDoc(dict(claim="custom_claim"))], - ) - ) - token_payload = { - "email_address": "test.user@example.com", - "given_name": "Test", - "name": "Test User", - "custom_claim": "custom_value", - } - - mock_user_doc = MagicMock() - mock_claims_doc = MagicMock() - mock_frappe.new_doc.return_value = mock_user_doc - mock_frappe.get_doc.return_value = mock_claims_doc - # Simulate that the roles exist in the DB - mock_frappe.db.get_value.return_value = True - - # --- Act --- - created_user = auth.create_and_save_user(token_payload, idp_config) - - # --- Assert --- - self.assertEqual(created_user, mock_user_doc) - self.assertEqual(mock_user_doc.email, "test.user@example.com") - self.assertEqual(mock_user_doc.first_name, "Test") - self.assertEqual(mock_user_doc.full_name, "Test User") - - # Check that roles were appended - self.assertEqual(mock_user_doc.append.call_count, 2) - mock_user_doc.append.assert_any_call("roles", {"role": "Blogger"}) - mock_user_doc.append.assert_any_call("roles", {"role": "Website Manager"}) - - # Check that custom claims were appended - mock_claims_doc.append.assert_called_once_with( - "claims", {"claim": "custom_claim", "value": "custom_value"} - ) - mock_user_doc.save.assert_called_once() - mock_claims_doc.save.assert_called_once() - mock_frappe.db.commit.assert_called_once() diff --git a/tests/test_auth_helpers.py b/tests/test_auth_helpers.py new file mode 100644 index 0000000..86db97b --- /dev/null +++ b/tests/test_auth_helpers.py @@ -0,0 +1,84 @@ +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc + + +class TestAuthHelpers(unittest.TestCase): + @patch("castlecraft.auth.frappe") + def test_get_idp_with_name(self, mock_frappe): + """Test get_idp fetches a named IDP.""" + # --- Arrange --- + idp_name = "specific-idp" + + # --- Act --- + auth.get_idp(idp_name) + + # --- Assert --- + mock_frappe.get_cached_doc.assert_called_once_with( + "CFE Identity Provider", idp_name + ) + mock_frappe.get_last_doc.assert_not_called() + + @patch("castlecraft.auth.frappe") + def test_get_idp_default(self, mock_frappe): + """Test get_idp fetches the default (last) IDP when no name is given.""" + # --- Arrange --- + # --- Act --- + auth.get_idp() + + # --- Assert --- + mock_frappe.get_last_doc.assert_called_once_with( + "CFE Identity Provider", filters={"enabled": 1} + ) + mock_frappe.get_cached_doc.assert_not_called() + + @patch("castlecraft.auth.frappe") + def test_create_and_save_user(self, mock_frappe): + """Test create_and_save_user correctly maps claims and roles.""" + # --- Arrange --- + idp_config = MockDoc( + dict( + email_key="email_address", + first_name_key="given_name", + full_name_key="name", + user_roles=[ + MockDoc(dict(role="Blogger")), + MockDoc(dict(role="Website Manager")), + ], + user_fields=[MockDoc(dict(claim="custom_claim"))], + ) + ) + token_payload = { + "email_address": "test.user@example.com", + "given_name": "Test", + "name": "Test User", + "custom_claim": "custom_value", + } + + mock_user_doc = MagicMock() + mock_claims_doc = MagicMock() + mock_frappe.new_doc.return_value = mock_user_doc + mock_frappe.get_doc.return_value = mock_claims_doc + mock_frappe.db.get_value.return_value = True + + # --- Act --- + created_user = auth.create_and_save_user(token_payload, idp_config) + + # --- Assert --- + self.assertEqual(created_user, mock_user_doc) + self.assertEqual(mock_user_doc.email, "test.user@example.com") + self.assertEqual(mock_user_doc.first_name, "Test") + self.assertEqual(mock_user_doc.full_name, "Test User") + + self.assertEqual(mock_user_doc.append.call_count, 2) + mock_user_doc.append.assert_any_call("roles", {"role": "Blogger"}) + mock_user_doc.append.assert_any_call("roles", {"role": "Website Manager"}) + + mock_claims_doc.append.assert_called_once_with( + "claims", {"claim": "custom_claim", "value": "custom_value"} + ) + mock_user_doc.save.assert_called_once() + mock_claims_doc.save.assert_called_once() + mock_frappe.db.commit.assert_called_once() diff --git a/tests/test_auth_introspection.py b/tests/test_auth_introspection.py new file mode 100644 index 0000000..30d2d3b --- /dev/null +++ b/tests/test_auth_introspection.py @@ -0,0 +1,238 @@ +import datetime +import json +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc # Import MockDoc from conftest + + +class TestAuthIntrospection(unittest.TestCase): + def setUp(self): + self.test_user_email = "test@example.com" + self.access_token = "valid-token-123" + + self.mock_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + idp_name="test-idp", + enabled=1, + authorization_type="Introspection", + email_key="email", + introspection_endpoint="https://idp.example.com/introspect", + token_key="token", + auth_header_enabled=1, + client_id="test-client", + client_secret="test-secret", + create_user=False, + fetch_user_info=False, + profile_endpoint="https://idp.example.com/userinfo", + user_roles=[], + ) + ) + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_valid_uncached_token_existing_user( + self, mock_requests, mock_get_idp + ): + """ + Test `validate_bearer_with_introspection` with a valid, + uncached token for an existing user. + """ + # --- Arrange --- + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + auth.frappe.db.exists.return_value = True + + mock_response = MagicMock() + mock_response.status_code = 200 + introspection_payload = { + "active": True, + "email": self.test_user_email, + "sub": "user-subject-123", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once_with( + self.mock_idp.introspection_endpoint, + data={"token": self.access_token}, + auth=unittest.mock.ANY, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + auth.frappe.set_user.assert_called_once_with(self.test_user_email) + auth.frappe.cache().set_value.assert_called() + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_valid_cached_token(self, mock_requests, mock_get_idp): + """ + Test introspection with a valid, + cached token to ensure it bypasses HTTP requests. + """ + # --- Arrange --- + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + + cached_payload = { + "active": True, + "email": self.test_user_email, + "sub": "user-subject-123", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + auth.frappe.cache().get_value.return_value = json.dumps(cached_payload) + auth.frappe.db.exists.return_value = True + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_not_called() + auth.frappe.set_user.assert_called_once_with(self.test_user_email) + + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_new_user_creation( + self, mock_requests, mock_get_idp, mock_create_user + ): + """ + Test introspection with a valid token for a + non-existent user when `create_user` is enabled. + """ + # --- Arrange --- + self.mock_idp.create_user = True + new_user_email = "new.user@example.com" + + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + auth.frappe.db.exists.return_value = False + + mock_response = MagicMock() + introspection_payload = { + "active": True, + "email": new_user_email, + "sub": "new-user-sub-456", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + mock_create_user.assert_called_once_with(introspection_payload, self.mock_idp) + auth.frappe.set_user.assert_called_once_with(new_user_email) + + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_with_fetch_user_info( + self, mock_requests, mock_get_idp, mock_create_user + ): + """Test introspection with `fetch_user_info` enabled for a new user.""" + # --- Arrange --- + self.mock_idp.create_user = True + self.mock_idp.fetch_user_info = True + new_user_email = "another.new.user@example.com" + + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + auth.frappe.db.exists.return_value = False + + introspection_response = MagicMock() + introspection_response.json.return_value = { + "active": True, + "sub": "another-sub-789", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + + userinfo_response = MagicMock() + userinfo_payload = { + "email": new_user_email, + "given_name": "Another", + "family_name": "User", + } + userinfo_response.json.return_value = userinfo_payload + + mock_requests.post.return_value = introspection_response + mock_requests.get.return_value = userinfo_response + + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once_with( + self.mock_idp.introspection_endpoint, + data=unittest.mock.ANY, + auth=unittest.mock.ANY, + headers=unittest.mock.ANY, + ) + mock_requests.get.assert_called_once_with( + self.mock_idp.profile_endpoint, + headers={"Authorization": f"Bearer {self.access_token}"}, + ) + mock_create_user.assert_called_once_with(userinfo_payload, self.mock_idp) + auth.frappe.set_user.assert_called_once_with(new_user_email) + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_inactive_token(self, mock_requests, mock_get_idp): + """Test introspection with a token that is inactive.""" + # --- Arrange --- + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + + mock_response = MagicMock() + introspection_payload = {"active": False} + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + auth.frappe.set_user.assert_not_called() + + @patch("castlecraft.auth.get_idp") + @patch("castlecraft.auth.requests") + def test_introspection_new_user_creation_disabled( + self, mock_requests, mock_get_idp + ): + """Test auth fails for a new user when `create_user` is disabled.""" + # --- Arrange --- + self.mock_idp.create_user = False + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_idp + auth.frappe.db.exists.return_value = False + + mock_response = MagicMock() + introspection_payload = { + "active": True, + "email": "new.user@example.com", + "exp": (datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp(), + } + mock_response.json.return_value = introspection_payload + mock_requests.post.return_value = mock_response + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_requests.post.assert_called_once() + auth.frappe.set_user.assert_not_called() diff --git a/tests/test_jwt_existing_user.py b/tests/test_jwt_existing_user.py new file mode 100644 index 0000000..d394f98 --- /dev/null +++ b/tests/test_jwt_existing_user.py @@ -0,0 +1,66 @@ +import time +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc + + +class TestJWTExistingUser(unittest.TestCase): + def setUp(self): + self.test_user_email = "test@example.com" + self.mock_jwt_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + authorization_type="JWT Verification", + email_key="email", + jwks_endpoint="https://idp.example.com/.well-known/jwks.json", + fetch_user_info=False, # Ensure this is a JWT-only flow + audience_claim_key="aud", + allowed_audience=[MockDoc(dict(aud="test-audience"))], + ) + ) + + @patch("castlecraft.auth.requests") + @patch( + "castlecraft.auth.frappe.get_value" + ) # Mock frappe.get_value for pre-validation + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.get_b64_decoded_json") + @patch("castlecraft.auth.get_idp") + def test_jwt_verification_existing_user( + self, mock_get_idp, mock_b64_decode, mock_jwt, mock_get_value, mock_requests + ): + """Test JWT verification with a valid token for an existing user.""" + # --- Arrange --- Use a valid-looking JWT structure for the token + jwt_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InRlc3Qta2lkIn0.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiZW1haWwiOiJ0ZXN0QGV4YW1wbGUuY29tIiwiaWF0IjoxNTE2MjM5MDIyLCJhdWQiOiJ0ZXN0LWF1ZGllbmNlIn0.signature_part" # noqa: E501 + auth.frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + + # Reset side_effect for this specific test to ensure isolation + auth.frappe.get_value.reset_mock() + mock_get_value.return_value = self.test_user_email + + decoded_payload = { + "email": self.test_user_email, + "exp": time.time() + 3600, + "aud": "test-audience", + "sub": "jwt-user-sub", + } + mock_b64_decode.return_value = decoded_payload + + # Mock the internal dependencies of validate_signature + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.return_value = decoded_payload + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_jwt.decode.assert_called_once() + auth.frappe.set_user.assert_called_once_with(self.test_user_email) + auth.frappe.cache().set_value.assert_called() diff --git a/tests/test_jwt_invalid.py b/tests/test_jwt_invalid.py new file mode 100644 index 0000000..f79c9ea --- /dev/null +++ b/tests/test_jwt_invalid.py @@ -0,0 +1,100 @@ +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc + + +class TestJWTInvalid(unittest.TestCase): + def setUp(self): + self.access_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InRlc3Qta2lkIn0.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiZW1haWwiOiJ0ZXN0QGV4YW1wbGUuY29tIiwiaWF0IjoxNTE2MjM5MDIyLCJhdWQiOiJ0ZXN0LWF1ZGllbmNlIn0.another_invalid_signature_part" # noqa: E501 + self.mock_jwt_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + authorization_type="JWT Verification", + jwks_endpoint="https://idp.example.com/.well-known/jwks.json", + fetch_user_info=False, # Set to False to force JWT decoding path + profile_endpoint="https://idp.example.com/userinfo", + audience_claim_key="aud", + allowed_audience=[MockDoc(dict(aud="test-audience"))], + ) + ) + + @patch( + "castlecraft.auth.frappe.get_value", return_value=None + ) # Ensure no user is found in pre-validation + @patch("castlecraft.auth.requests") + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.get_idp") + def test_jwt_verification_invalid_signature( + self, mock_get_idp, mock_jwt, mock_requests, mock_get_value + ): + """Test JWT verification fails with an invalid signature.""" + # --- Arrange --- Use a valid-looking JWT structure for the token + jwt_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InRlc3Qta2lkIn0.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiZW1haWwiOiJ0ZXN0QGV4YW1wbGUuY29tIiwiaWF0IjoxNTE2MjM5MDIyLCJhdWQiOiJ0ZXN0LWF1ZGllbmNlIn0.another_invalid_signature_part" # noqa: E501 + auth.frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + + # Mock the internal calls of validate_signature to simulate failure + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.side_effect = Exception("Invalid signature") + + with patch("castlecraft.auth.get_b64_decoded_json"): + # --- Act --- + auth.validate() + + # --- Assert --- + auth.frappe.set_user.assert_not_called() + + @patch( + "castlecraft.auth.frappe.get_value", return_value=None + ) # Ensure no user is found in pre-validation + @patch("castlecraft.auth.requests") + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.request_user_info") + @patch("castlecraft.auth.get_idp") + def test_jwt_with_userinfo_invalid_signature( + self, + mock_get_idp, + mock_request_user_info, + mock_jwt, + mock_requests, + mock_get_value, + ): + """Test flow fails if JWT signature is invalid before userinfo call.""" + # --- Arrange --- + # For this test, we need to simulate an IdP that *does* fetch user info. + # We must explicitly reconstruct the MockDoc to ensure all attributes are present. + mock_jwt_idp_with_userinfo = MockDoc( + dict( + doctype=self.mock_jwt_idp.doctype, + authorization_type=self.mock_jwt_idp.authorization_type, + jwks_endpoint=self.mock_jwt_idp.jwks_endpoint, + fetch_user_info=True, + profile_endpoint=self.mock_jwt_idp.profile_endpoint, + audience_claim_key=self.mock_jwt_idp.audience_claim_key, + allowed_audience=self.mock_jwt_idp.allowed_audience, + ) + ) + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = mock_jwt_idp_with_userinfo + + # Mock the internal calls of validate_signature to simulate failure + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.side_effect = Exception("Invalid signature") + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_jwt.decode.assert_called_once() + mock_request_user_info.assert_not_called() + auth.frappe.set_user.assert_not_called() diff --git a/tests/test_jwt_new_user.py b/tests/test_jwt_new_user.py new file mode 100644 index 0000000..1845501 --- /dev/null +++ b/tests/test_jwt_new_user.py @@ -0,0 +1,81 @@ +# /home/revant/Projects/cfe-bench/development/frappe-bench/apps/castlecraft/tests/test_jwt_new_user.py +import time +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc + + +class TestJWTNewUser(unittest.TestCase): + def setUp(self): + self.mock_jwt_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + authorization_type="JWT Verification", + email_key="email", + create_user=True, + fetch_user_info=False, # Explicitly define this to prevent AttributeError + jwks_endpoint="https://idp.example.com/.well-known/jwks.json", + audience_claim_key="aud", + allowed_audience=[MockDoc(dict(aud="test-audience"))], + first_name_key="given_name", + full_name_key="name", + user_roles=[], + user_fields=[], + ) + ) + + @patch("castlecraft.auth.requests") + @patch( + "castlecraft.auth.frappe.get_value" + ) # Mock frappe.get_value for pre-validation + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.get_b64_decoded_json") + @patch("castlecraft.auth.get_idp") + def test_jwt_verification_new_user_creation( + self, + mock_get_idp, + mock_b64_decode, + mock_create_user, + mock_jwt, + mock_get_value, + mock_requests, + ): + """Test JWT verification with user creation for a new user.""" + # --- Arrange --- + new_user_email = "new.jwt.user@example.com" + jwt_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InRlc3Qta2lkIn0.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiZW1haWwiOiJuZXcuand0LnVzZXJAZXhhbXBsZS5jb20iLCJpYXQiOjE1MTYyMzkwMjIsImF1ZCI6InRlc3QtYXVkaWVuY2UifQ.signature_part" # noqa: E501 + + auth.frappe.get_request_header.return_value = f"Bearer {jwt_token}" + mock_get_idp.return_value = self.mock_jwt_idp + # User is not found after successful JWT decoding + mock_get_value.return_value = None + + decoded_payload = { + "email": new_user_email, + "exp": time.time() + 3600, + "aud": "test-audience", + "sub": "new-user-sub-456", # Ensure sub is present for cache_user_from_sub + } + mock_b64_decode.return_value = decoded_payload + + # Mock the internal dependencies of validate_signature + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.return_value = decoded_payload + + mock_user = MockDoc(dict(email=new_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_jwt.decode.assert_called_once() + mock_create_user.assert_called_once_with(decoded_payload, self.mock_jwt_idp) + auth.frappe.set_user.assert_called_once_with(new_user_email) diff --git a/tests/test_jwt_userinfo.py b/tests/test_jwt_userinfo.py new file mode 100644 index 0000000..58e88ef --- /dev/null +++ b/tests/test_jwt_userinfo.py @@ -0,0 +1,133 @@ +import time +import unittest +from unittest.mock import MagicMock, patch + +from castlecraft import auth +from tests.conftest import MockDoc + + +class TestJWTUserInfo(unittest.TestCase): + def setUp(self): + self.test_user_email = "test@example.com" + self.access_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6InRlc3Qta2lkIn0.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiZW1haWwiOiJ0ZXN0QGV4YW1wbGUuY29tIiwiaWF0IjoxNTE2MjM5MDIyLCJhdWQiOiJ0ZXN0LWF1ZGllbmNlIn0.signature_part" # noqa: E501 + self.mock_jwt_idp = MockDoc( + dict( + doctype="CFE Identity Provider", + authorization_type="JWT Verification", + email_key="email", + create_user=True, + fetch_user_info=True, + profile_endpoint="https://idp.example.com/userinfo", + jwks_endpoint="https://idp.example.com/.well-known/jwks.json", + audience_claim_key="aud", + allowed_audience=[MockDoc(dict(aud="test-audience"))], + ) + ) + + @patch( + "castlecraft.auth.frappe.get_value" + ) # Mock frappe.get_value for pre-validation + @patch("castlecraft.auth.requests") + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.request_user_info") + @patch("castlecraft.auth.get_b64_decoded_json") + @patch("castlecraft.auth.get_idp") + def test_jwt_with_userinfo_existing_user( + self, + mock_get_idp, + mock_b64_decode, + mock_request_user_info, + mock_jwt, + mock_requests, + mock_get_value, + ): + """Test JWT with userinfo for an existing user.""" + # --- Arrange --- + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_jwt_idp + # After userinfo is fetched, the user is found + mock_get_value.return_value = self.test_user_email + + jwt_payload = { + "sub": "user-sub-123", + "exp": time.time() + 3600, + "aud": "test-audience", + "email": self.test_user_email, # Ensure email is in payload for frappe.get_value + } + # Mock the internal dependencies of validate_signature + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.return_value = jwt_payload + userinfo_payload = {"email": self.test_user_email, "name": "Test User"} + mock_b64_decode.return_value = jwt_payload + mock_request_user_info.return_value = userinfo_payload + + # --- Act --- + auth.validate() + + # --- Assert --- + # JWT is always validated first + mock_jwt.decode.assert_called_once() + mock_request_user_info.assert_called_once_with( + self.access_token, self.mock_jwt_idp + ) + auth.frappe.set_user.assert_called_once_with(self.test_user_email) + + @patch("castlecraft.auth.requests") + @patch( + "castlecraft.auth.frappe.get_value" + ) # Mock frappe.get_value for pre-validation + @patch("castlecraft.auth.jwt") + @patch("castlecraft.auth.create_and_save_user") + @patch("castlecraft.auth.request_user_info") + @patch("castlecraft.auth.get_b64_decoded_json") + @patch("castlecraft.auth.get_idp") + def test_jwt_with_userinfo_new_user_creation( + self, + mock_get_idp, + mock_b64_decode, # Corrected order + mock_request_user_info, + mock_create_user, + mock_jwt, + mock_get_value, + mock_requests, + ): + """Test JWT with userinfo creates a new user.""" + # --- Arrange --- + auth.frappe.get_request_header.return_value = f"Bearer {self.access_token}" + mock_get_idp.return_value = self.mock_jwt_idp + # After userinfo is fetched, user is not found, triggering creation + mock_get_value.return_value = None + + jwt_payload = { + "sub": "new-user-sub-456", + "exp": time.time() + 3600, + "aud": "test-audience", + "email": self.test_user_email, # Ensure email is in payload for frappe.get_value + } + # Mock the internal dependencies of validate_signature + mock_requests.get.return_value.json.return_value = { + "keys": [{"kid": "test-kid"}] + } + mock_jwt.get_unverified_header.return_value = {"kid": "test-kid"} + mock_jwt.algorithms.RSAAlgorithm.from_jwk.return_value = MagicMock() + mock_jwt.decode.return_value = jwt_payload + userinfo_payload = {"email": self.test_user_email, "given_name": "New"} + mock_b64_decode.return_value = jwt_payload + mock_request_user_info.return_value = userinfo_payload + + mock_user = MockDoc(dict(email=self.test_user_email)) + mock_create_user.return_value = mock_user + + # --- Act --- + auth.validate() + + # --- Assert --- + mock_jwt.decode.assert_called_once() + mock_request_user_info.assert_called_once() + # When creating a user from userinfo, only the userinfo payload is passed + mock_create_user.assert_called_once_with(userinfo_payload, self.mock_jwt_idp) + auth.frappe.set_user.assert_called_once_with(self.test_user_email) From 83d2a24495e4ae06e0ea4fe2809a6c8524cd823f Mon Sep 17 00:00:00 2001 From: Revant Nandgaonkar Date: Sat, 8 Nov 2025 08:42:56 +0530 Subject: [PATCH 3/3] ci: add tests --- .github/workflows/test.yml | 37 ++++++++++++++++++++++++++++++++++ castlecraft/services/oauth2.py | 11 ++++------ 2 files changed, 41 insertions(+), 7 deletions(-) create mode 100644 .github/workflows/test.yml diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..beafe0c --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,37 @@ +name: Test + +on: + push: + branches: + - main + pull_request: + branches: + - main + +jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v3 + + - name: Setup Python + uses: actions/setup-python@v4 + with: + python-version: '3.11' + + - name: Cache pip dependencies + uses: actions/cache@v3 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-${{ hashFiles('**/pyproject.toml') }} + restore-keys: | + ${{ runner.os }}-pip- + + - name: Install Dependencies + run: | + pip install -e git+https://github.com/frappe/frappe.git@version-15#egg=frappe + pip install -e ."[ci]" + + - name: Test + run: pytest -v tests diff --git a/castlecraft/services/oauth2.py b/castlecraft/services/oauth2.py index 8f47bf2..a82a82e 100644 --- a/castlecraft/services/oauth2.py +++ b/castlecraft/services/oauth2.py @@ -4,13 +4,12 @@ from frappe.exceptions import DoesNotExistError from frappe.oauth import get_userinfo -from castlecraft.auth import get_enabled_idp, get_userinfo_from_idp +from castlecraft.auth import get_idp, get_userinfo_from_idp from castlecraft.utils.format import respond_error from castlecraft.auth import ( # isort: skip - delete_cached_bearer_token, - delete_cached_jwt, get_cached_user_from_sub, + delete_cached_bearer_token, validate_signature, ) @@ -47,12 +46,10 @@ def back_channel_logout(logout_token=None): verified_token = validate_signature(logout_token) if verified_token.get("sub"): payload = get_cached_user_from_sub(verified_token.get("sub")) - email = payload.get("email") token = payload.get("token") - if email: - delete_cached_jwt(email) if token: delete_cached_bearer_token(token) + frappe.cache().delete_key(f"cc_jwt_payload|{token}") except Exception: frappe.log_error(traceback.format_exc(), error_string) respond_error(error_string, 400) @@ -79,7 +76,7 @@ def sync_claims(): if len(authorization_header) == 2: token = authorization_header[1] - idp = get_enabled_idp() + idp = get_idp() idp_user_claims = get_userinfo_from_idp(token, idp) idp_claim_map = [field.claim for field in idp.user_fields] claims = []