diff --git a/prowler/providers/okta/models.py b/prowler/providers/okta/models.py index 0f9c14a87b..b3f5c1f1cc 100644 --- a/prowler/providers/okta/models.py +++ b/prowler/providers/okta/models.py @@ -32,6 +32,12 @@ class OktaSession(BaseModel): class OktaIdentityInfo(BaseModel): org_domain: str client_id: str + # Scopes actually granted in the access token (`scp` claim). Used by + # services to distinguish "no data" from "no permission" so checks can + # surface the missing scope rather than a misleading FAIL. Empty when + # decoding the token was not possible — callers must treat empty as + # "unknown" and fall back to attempting the API call. + granted_scopes: list[str] = [] class OktaOutputOptions(ProviderOutputOptions): diff --git a/prowler/providers/okta/okta_provider.py b/prowler/providers/okta/okta_provider.py index f863585bfd..a6bff16d80 100644 --- a/prowler/providers/okta/okta_provider.py +++ b/prowler/providers/okta/okta_provider.py @@ -1,4 +1,6 @@ import asyncio +import base64 +import json import os import re from os import environ @@ -285,14 +287,32 @@ class OktaProvider(Provider): org URL plus the service-app client ID. We still hit the cheapest scope-covered endpoint (`list_policies` with limit=1) to fail loud when credentials, scopes, or the granted admin role are wrong. + + After the probe succeeds, the access token's `scp` claim is + decoded and exposed on the identity. Services compare it against + their required scope so checks can emit "missing scope X" rather + than a misleading "no resources returned" finding. """ async def _probe(): client = OktaSDKClient(session.to_sdk_config()) - return await client.list_policies(type="OKTA_SIGN_ON", limit="1") + result = await client.list_policies(type="OKTA_SIGN_ON", limit="1") + access_token = None + # The OAuth helper caches the token on `_access_token` after + # the first authenticated call. Reach through `_request_executor` + # — a documented internal but a moving target across SDK + # versions, so any failure here degrades silently to empty + # granted_scopes (services then fall back to attempting calls). + try: + oauth = getattr(client._request_executor, "_oauth", None) + if oauth is not None: + access_token = getattr(oauth, "_access_token", None) + except Exception: + access_token = None + return result, access_token try: - result = asyncio.run(_probe()) + result, access_token = asyncio.run(_probe()) # SDK returns (items, resp, err) on the normal path and (items, err) # only on early request-creation errors. The error is always last. err = result[-1] @@ -305,6 +325,11 @@ class OktaProvider(Provider): "forbidden", "not authorized", "permission", + # Okta emits HTTP 400 `consent_required` when none of the + # requested scopes are consented on the service app — + # semantically a permission gap, not a credential one. + "consent_required", + "not allowed", ) if any(signal in err_text for signal in permission_signals): raise OktaInsufficientPermissionsError( @@ -321,6 +346,7 @@ class OktaProvider(Provider): return OktaIdentityInfo( org_domain=session.org_domain, client_id=session.client_id, + granted_scopes=OktaProvider._decode_token_scopes(access_token), ) except (OktaInvalidCredentialsError, OktaInsufficientPermissionsError): raise @@ -330,6 +356,40 @@ class OktaProvider(Provider): ) raise OktaSetUpIdentityError(original_exception=error) + @staticmethod + def _decode_token_scopes(access_token: Optional[str]) -> list[str]: + """Return the `scp` claim from a JWT access token, or `[]` on failure. + + No signature verification: the token came from Okta over TLS via + the SDK's OAuth handshake, so the only thing we extract is the + scope claim. Any decode error returns an empty list — callers + must treat empty as "unknown" rather than "no scopes granted". + """ + if not access_token: + return [] + try: + parts = access_token.split(".") + if len(parts) < 2: + return [] + payload_b64 = parts[1] + # Base64url pad to a multiple of 4 — JWT segments are + # unpadded per RFC 7515. + padding = "=" * (-len(payload_b64) % 4) + payload_bytes = base64.urlsafe_b64decode(payload_b64 + padding) + payload = json.loads(payload_bytes) + scp = payload.get("scp") + if isinstance(scp, list): + return [str(s) for s in scp if s] + if isinstance(scp, str): + return [s for s in scp.split(" ") if s] + return [] + except Exception as error: + logger.warning( + f"Could not decode Okta access token scopes: " + f"{error.__class__.__name__}: {error}" + ) + return [] + def print_credentials(self): report_lines = [ f"Okta Domain: {Fore.YELLOW}{self.identity.org_domain}{Style.RESET_ALL}", diff --git a/prowler/providers/okta/services/signon/lib/signon_helpers.py b/prowler/providers/okta/services/signon/lib/signon_helpers.py index bb45c17852..7c84fe5733 100644 --- a/prowler/providers/okta/services/signon/lib/signon_helpers.py +++ b/prowler/providers/okta/services/signon/lib/signon_helpers.py @@ -13,6 +13,7 @@ from prowler.lib.check.models import CheckReportOkta from prowler.providers.okta.services.signon.signon_service import ( GlobalSessionPolicy, GlobalSessionPolicyRule, + SignInPage, ) @@ -90,3 +91,56 @@ def no_active_policies_finding( report.status = "FAIL" report.status_extended = status_extended return report + + +_SCOPE_ADVICE = ( + "Grant it on the service app's Okta API Scopes tab in the Okta Admin " + "Console, then re-run the check." +) + + +def missing_policy_scope_finding( + metadata, org_domain: str, scope: str +) -> CheckReportOkta: + """Build the MANUAL finding for a sign-on policy check when the API scope is not granted.""" + placeholder = GlobalSessionPolicy( + id="signon-policies-scope-missing", + name="(scope not granted)", + priority=1, + status="MISSING", + is_default=False, + rules=[], + ) + report = CheckReportOkta( + metadata=metadata, resource=placeholder, org_domain=org_domain + ) + report.status = "MANUAL" + report.status_extended = ( + f"Could not retrieve Global Session Policies: the Okta service app " + f"is missing the required `{scope}` API scope. {_SCOPE_ADVICE}" + ) + return report + + +def missing_brand_scope_finding( + metadata, org_domain: str, scope: str +) -> CheckReportOkta: + """Build the MANUAL finding for a brand/sign-in-page check when the API scope is not granted.""" + placeholder = SignInPage( + brand_id="signon-brands-scope-missing", + brand_name="(scope not granted)", + is_customized=False, + ) + report = CheckReportOkta( + metadata=metadata, + resource=placeholder, + org_domain=org_domain, + resource_name=placeholder.brand_name, + resource_id=placeholder.brand_id, + ) + report.status = "MANUAL" + report.status_extended = ( + f"Could not retrieve Okta brand sign-in pages: the Okta service app " + f"is missing the required `{scope}` API scope. {_SCOPE_ADVICE}" + ) + return report diff --git a/prowler/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured.py b/prowler/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured.py index 4e85532cc0..347f04f245 100644 --- a/prowler/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured.py +++ b/prowler/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured.py @@ -1,4 +1,7 @@ from prowler.lib.check.models import Check, CheckReportOkta +from prowler.providers.okta.services.signon.lib.signon_helpers import ( + missing_brand_scope_finding, +) from prowler.providers.okta.services.signon.signon_client import signon_client from prowler.providers.okta.services.signon.signon_service import SignInPage @@ -43,6 +46,12 @@ class signon_dod_warning_banner_configured(Check): org_domain = signon_client.provider.identity.org_domain findings: list[CheckReportOkta] = [] + missing_scope = signon_client.missing_scope.get("sign_in_pages") + if missing_scope: + return [ + missing_brand_scope_finding(self.metadata(), org_domain, missing_scope) + ] + if not signon_client.sign_in_pages: placeholder = SignInPage( brand_id="no-brands-detected", diff --git a/prowler/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent.py b/prowler/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent.py index 81be40872c..c5ffd849b5 100644 --- a/prowler/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent.py +++ b/prowler/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent.py @@ -1,6 +1,7 @@ from prowler.lib.check.models import Check, CheckReportOkta from prowler.providers.okta.services.signon.lib.signon_helpers import ( active_policies, + missing_policy_scope_finding, no_active_policies_finding, policy_label, priority_one_active_rule, @@ -26,6 +27,12 @@ class signon_global_session_cookies_not_persistent(Check): def execute(self) -> list[CheckReportOkta]: org_domain = signon_client.provider.identity.org_domain + missing_scope = signon_client.missing_scope.get("global_session_policies") + if missing_scope: + return [ + missing_policy_scope_finding(self.metadata(), org_domain, missing_scope) + ] + policies = active_policies(signon_client.global_session_policies) if not policies: return [ diff --git a/prowler/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min.py b/prowler/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min.py index 9981f360e0..3f20a22adb 100644 --- a/prowler/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min.py +++ b/prowler/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min.py @@ -1,6 +1,7 @@ from prowler.lib.check.models import Check, CheckReportOkta from prowler.providers.okta.services.signon.lib.signon_helpers import ( active_policies, + missing_policy_scope_finding, no_active_policies_finding, policy_label, priority_one_active_rule, @@ -33,6 +34,12 @@ class signon_global_session_idle_timeout_15min(Check): ) org_domain = signon_client.provider.identity.org_domain + missing_scope = signon_client.missing_scope.get("global_session_policies") + if missing_scope: + return [ + missing_policy_scope_finding(self.metadata(), org_domain, missing_scope) + ] + policies = active_policies(signon_client.global_session_policies) if not policies: return [ diff --git a/prowler/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h.py b/prowler/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h.py index 9a973f6834..25003eef90 100644 --- a/prowler/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h.py +++ b/prowler/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h.py @@ -1,6 +1,7 @@ from prowler.lib.check.models import Check, CheckReportOkta from prowler.providers.okta.services.signon.lib.signon_helpers import ( active_policies, + missing_policy_scope_finding, no_active_policies_finding, policy_label, priority_one_active_rule, @@ -33,6 +34,12 @@ class signon_global_session_lifetime_18h(Check): ) org_domain = signon_client.provider.identity.org_domain + missing_scope = signon_client.missing_scope.get("global_session_policies") + if missing_scope: + return [ + missing_policy_scope_finding(self.metadata(), org_domain, missing_scope) + ] + policies = active_policies(signon_client.global_session_policies) if not policies: return [ diff --git a/prowler/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced.py b/prowler/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced.py index 29e1d49cbc..7a7e51706c 100644 --- a/prowler/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced.py +++ b/prowler/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced.py @@ -1,6 +1,7 @@ from prowler.lib.check.models import Check, CheckReportOkta from prowler.providers.okta.services.signon.lib.signon_helpers import ( active_policies, + missing_policy_scope_finding, no_active_policies_finding, policy_label, priority_one_active_rule, @@ -31,6 +32,12 @@ class signon_global_session_policy_network_zone_enforced(Check): def execute(self) -> list[CheckReportOkta]: org_domain = signon_client.provider.identity.org_domain + missing_scope = signon_client.missing_scope.get("global_session_policies") + if missing_scope: + return [ + missing_policy_scope_finding(self.metadata(), org_domain, missing_scope) + ] + policies = active_policies(signon_client.global_session_policies) if not policies: return [ diff --git a/prowler/providers/okta/services/signon/signon_service.py b/prowler/providers/okta/services/signon/signon_service.py index d47ad41fc1..7c32363501 100644 --- a/prowler/providers/okta/services/signon/signon_service.py +++ b/prowler/providers/okta/services/signon/signon_service.py @@ -29,6 +29,12 @@ def _next_after_cursor(resp) -> Optional[str]: return None +REQUIRED_SCOPES: dict[str, str] = { + "global_session_policies": "okta.policies.read", + "sign_in_pages": "okta.brands.read", +} + + class Signon(OktaService): """Fetches OKTA_SIGN_ON policies, rules, and brand sign-in pages. @@ -41,14 +47,34 @@ class Signon(OktaService): customized page, the service falls back to the default sign-in page exposed by the Okta Management API and tracks it with `is_customized=False`. + + Before each fetch the service compares its required OAuth scope + (see `REQUIRED_SCOPES`) against the access token's granted scopes + (`provider.identity.granted_scopes`). When a scope is known to be + missing, the fetch is skipped and the resource is recorded in + `self.missing_scope` so checks can report the missing scope explicitly + instead of emitting a misleading "no resources returned" finding. + When granted_scopes is empty (token decode unavailable), the service + treats permissions as unknown and attempts the fetch — preserving + the prior behavior. """ def __init__(self, provider): super().__init__(__class__.__name__, provider) + granted = set(getattr(provider.identity, "granted_scopes", None) or []) + self.missing_scope: dict[str, Optional[str]] = { + resource: (scope if granted and scope not in granted else None) + for resource, scope in REQUIRED_SCOPES.items() + } + self.global_session_policies: dict[str, GlobalSessionPolicy] = ( - self._list_global_session_policies() + {} + if self.missing_scope["global_session_policies"] + else self._list_global_session_policies() + ) + self.sign_in_pages: dict[str, SignInPage] = ( + {} if self.missing_scope["sign_in_pages"] else self._list_sign_in_pages() ) - self.sign_in_pages: dict[str, SignInPage] = self._list_sign_in_pages() def _list_global_session_policies(self) -> dict: logger.info("Signon - Listing OKTA_SIGN_ON policies and rules...") diff --git a/tests/providers/okta/okta_fixtures.py b/tests/providers/okta/okta_fixtures.py index 58eed16862..2b7ca6927c 100644 --- a/tests/providers/okta/okta_fixtures.py +++ b/tests/providers/okta/okta_fixtures.py @@ -23,6 +23,7 @@ def set_mocked_okta_provider( identity = OktaIdentityInfo( org_domain=OKTA_ORG_DOMAIN, client_id=OKTA_CLIENT_ID, + granted_scopes=["okta.policies.read", "okta.brands.read"], ) provider = MagicMock() diff --git a/tests/providers/okta/okta_provider_test.py b/tests/providers/okta/okta_provider_test.py index da325d84ca..5f2757edae 100644 --- a/tests/providers/okta/okta_provider_test.py +++ b/tests/providers/okta/okta_provider_test.py @@ -1,3 +1,5 @@ +import base64 +import json from unittest import mock import pytest @@ -20,6 +22,54 @@ from tests.providers.okta.okta_fixtures import ( ) +def _make_jwt(payload: dict) -> str: + """Build an unsigned JWT carrying the given payload dict. + + The signature segment is irrelevant — `_decode_token_scopes` reads + the payload without verification. + """ + + def _b64u(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + header = _b64u(json.dumps({"alg": "none"}).encode()) + body = _b64u(json.dumps(payload).encode()) + return f"{header}.{body}.sig" + + +class Test_OktaProvider_decode_token_scopes: + def test_returns_scopes_from_list_scp_claim(self): + token = _make_jwt({"scp": ["okta.policies.read", "okta.brands.read"]}) + assert OktaProvider._decode_token_scopes(token) == [ + "okta.policies.read", + "okta.brands.read", + ] + + def test_returns_scopes_from_space_separated_scp_string(self): + token = _make_jwt({"scp": "okta.policies.read okta.brands.read"}) + assert OktaProvider._decode_token_scopes(token) == [ + "okta.policies.read", + "okta.brands.read", + ] + + def test_returns_empty_list_when_token_is_none(self): + assert OktaProvider._decode_token_scopes(None) == [] + + def test_returns_empty_list_when_token_is_empty_string(self): + assert OktaProvider._decode_token_scopes("") == [] + + def test_returns_empty_list_when_scp_claim_missing(self): + token = _make_jwt({"sub": "client-id"}) + assert OktaProvider._decode_token_scopes(token) == [] + + def test_returns_empty_list_when_token_is_malformed(self): + assert OktaProvider._decode_token_scopes("not.a.jwt-with-bad-base64!!") == [] + + def test_returns_empty_list_when_payload_is_not_json(self): + bad = base64.urlsafe_b64encode(b"not json").rstrip(b"=").decode() + assert OktaProvider._decode_token_scopes(f"hdr.{bad}.sig") == [] + + @pytest.fixture def _clear_okta_env(monkeypatch): for var in ( @@ -272,6 +322,49 @@ class Test_OktaProvider_setup_identity: assert identity.org_domain == OKTA_ORG_DOMAIN assert identity.client_id == OKTA_CLIENT_ID + def test_populates_granted_scopes_from_access_token_scp_claim( + self, _clear_okta_env, tmp_path + ): + session = self._session(tmp_path) + + async def fake_list_policies(*_a, **_k): + return ([], mock.MagicMock(headers={}), None) + + with mock.patch( + "prowler.providers.okta.okta_provider.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = fake_list_policies + mocked._request_executor._oauth._access_token = _make_jwt( + {"scp": ["okta.policies.read", "okta.brands.read"]} + ) + mocked_client_cls.return_value = mocked + identity = OktaProvider.setup_identity(session) + + assert identity.granted_scopes == [ + "okta.policies.read", + "okta.brands.read", + ] + + def test_granted_scopes_empty_when_token_unavailable( + self, _clear_okta_env, tmp_path + ): + session = self._session(tmp_path) + + async def fake_list_policies(*_a, **_k): + return ([], mock.MagicMock(headers={}), None) + + with mock.patch( + "prowler.providers.okta.okta_provider.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = fake_list_policies + mocked._request_executor._oauth._access_token = None + mocked_client_cls.return_value = mocked + identity = OktaProvider.setup_identity(session) + + assert identity.granted_scopes == [] + def test_raises_invalid_credentials_when_probe_returns_error( self, _clear_okta_env, tmp_path ): @@ -323,6 +416,35 @@ class Test_OktaProvider_setup_identity: with pytest.raises(OktaInsufficientPermissionsError): OktaProvider.setup_identity(session) + def test_raises_insufficient_permissions_on_consent_required( + self, _clear_okta_env, tmp_path + ): + # When zero requested scopes are consented on the service app, Okta + # rejects the token request with HTTP 400 `consent_required` rather + # than `invalid_scope` — must still be classified as a permission + # gap so the user is pointed at the Okta API Scopes tab, not at + # credential troubleshooting. + session = self._session(tmp_path) + + async def failing_list_policies(*_a, **_k): + return ( + [], + None, + Exception( + "Okta HTTP 400 consent_required You are not allowed any " + "of the requested scopes." + ), + ) + + with mock.patch( + "prowler.providers.okta.okta_provider.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = failing_list_policies + mocked_client_cls.return_value = mocked + with pytest.raises(OktaInsufficientPermissionsError): + OktaProvider.setup_identity(session) + def test_wraps_unexpected_errors_in_setup_identity_error( self, _clear_okta_env, tmp_path ): diff --git a/tests/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured_test.py b/tests/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured_test.py index 619142f390..96406a5fd1 100644 --- a/tests/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured_test.py +++ b/tests/providers/okta/services/signon/signon_dod_warning_banner_configured/signon_dod_warning_banner_configured_test.py @@ -33,6 +33,30 @@ class Test_signon_dod_warning_banner_configured: assert findings[0].status == "MANUAL" assert "No Okta brands were retrieved" in findings[0].status_extended + def test_missing_brand_scope_returns_manual_finding_naming_the_scope(self): + signon_client = build_signon_client( + missing_scope={ + "global_session_policies": None, + "sign_in_pages": "okta.brands.read", + } + ) + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=set_mocked_okta_provider(), + ), + mock.patch(CHECK_PATH, new=signon_client), + ): + from prowler.providers.okta.services.signon.signon_dod_warning_banner_configured.signon_dod_warning_banner_configured import ( + signon_dod_warning_banner_configured, + ) + + findings = signon_dod_warning_banner_configured().execute() + assert len(findings) == 1 + assert findings[0].status == "MANUAL" + assert "okta.brands.read" in findings[0].status_extended + assert "missing the required" in findings[0].status_extended + def test_pass_when_customized_page_contains_banner(self): page = sign_in_page( brand_id="brand-1", diff --git a/tests/providers/okta/services/signon/signon_fixtures.py b/tests/providers/okta/services/signon/signon_fixtures.py index 99859fad81..1399a19293 100644 --- a/tests/providers/okta/services/signon/signon_fixtures.py +++ b/tests/providers/okta/services/signon/signon_fixtures.py @@ -22,12 +22,18 @@ def build_signon_client( policies: dict = None, audit_config: dict = None, sign_in_pages: dict = None, + missing_scope: dict = None, ): client = mock.MagicMock() client.global_session_policies = policies or {} client.provider = set_mocked_okta_provider() client.audit_config = audit_config or {} client.sign_in_pages = sign_in_pages or {} + # Default to "all scopes granted" so existing tests keep working. + client.missing_scope = missing_scope or { + "global_session_policies": None, + "sign_in_pages": None, + } return client diff --git a/tests/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent_test.py b/tests/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent_test.py index a293d99ee2..cd10d84be8 100644 --- a/tests/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent_test.py +++ b/tests/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent_test.py @@ -158,6 +158,20 @@ class Test_signon_global_session_cookies_not_persistent: assert findings[0].resource_name == "Default Policy" assert findings[0].status == "PASS" + def test_missing_scope_returns_manual_finding_naming_the_scope(self): + findings = _run_check( + build_signon_client( + missing_scope={ + "global_session_policies": "okta.policies.read", + "sign_in_pages": None, + } + ) + ) + assert len(findings) == 1 + assert findings[0].status == "MANUAL" + assert "okta.policies.read" in findings[0].status_extended + assert "missing the required" in findings[0].status_extended + def test_fail_when_all_policies_inactive(self): only_inactive = GlobalSessionPolicy( id="pol-default", diff --git a/tests/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min_test.py b/tests/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min_test.py index 770a195c40..177a4b1c8f 100644 --- a/tests/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min_test.py +++ b/tests/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min_test.py @@ -178,6 +178,20 @@ class Test_signon_global_session_idle_timeout_15min: assert findings[0].status == "FAIL" assert "No active Okta Global Session Policies" in findings[0].status_extended + def test_missing_scope_returns_manual_finding_naming_the_scope(self): + findings = _run_check( + build_signon_client( + missing_scope={ + "global_session_policies": "okta.policies.read", + "sign_in_pages": None, + } + ) + ) + assert len(findings) == 1 + assert findings[0].status == "MANUAL" + assert "okta.policies.read" in findings[0].status_extended + assert "missing the required" in findings[0].status_extended + def test_threshold_overridden_via_audit_config(self): policy = default_policy( [ diff --git a/tests/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h_test.py b/tests/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h_test.py index a43828f429..1a22c20573 100644 --- a/tests/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h_test.py +++ b/tests/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h_test.py @@ -180,6 +180,20 @@ class Test_signon_global_session_lifetime_18h: assert findings[0].status == "FAIL" assert "No active Okta Global Session Policies" in findings[0].status_extended + def test_missing_scope_returns_manual_finding_naming_the_scope(self): + findings = _run_check( + build_signon_client( + missing_scope={ + "global_session_policies": "okta.policies.read", + "sign_in_pages": None, + } + ) + ) + assert len(findings) == 1 + assert findings[0].status == "MANUAL" + assert "okta.policies.read" in findings[0].status_extended + assert "missing the required" in findings[0].status_extended + def test_threshold_overridden_via_audit_config(self): policy = default_policy( [ diff --git a/tests/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced_test.py b/tests/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced_test.py index e5e0c229f4..1d3067a257 100644 --- a/tests/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced_test.py +++ b/tests/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced_test.py @@ -206,3 +206,17 @@ class Test_signon_global_session_policy_network_zone_enforced: assert len(findings) == 1 assert findings[0].status == "FAIL" assert "No active Okta Global Session Policies" in findings[0].status_extended + + def test_missing_scope_returns_manual_finding_naming_the_scope(self): + findings = _run_check( + build_signon_client( + missing_scope={ + "global_session_policies": "okta.policies.read", + "sign_in_pages": None, + } + ) + ) + assert len(findings) == 1 + assert findings[0].status == "MANUAL" + assert "okta.policies.read" in findings[0].status_extended + assert "missing the required" in findings[0].status_extended diff --git a/tests/providers/okta/services/signon/signon_service_test.py b/tests/providers/okta/services/signon/signon_service_test.py index 3e2d4ddaa8..c408bc1221 100644 --- a/tests/providers/okta/services/signon/signon_service_test.py +++ b/tests/providers/okta/services/signon/signon_service_test.py @@ -1,5 +1,6 @@ from unittest import mock +from prowler.providers.okta.models import OktaIdentityInfo from prowler.providers.okta.services.signon.signon_service import ( GlobalSessionPolicy, GlobalSessionPolicyRule, @@ -7,7 +8,11 @@ from prowler.providers.okta.services.signon.signon_service import ( Signon, _next_after_cursor, ) -from tests.providers.okta.okta_fixtures import set_mocked_okta_provider +from tests.providers.okta.okta_fixtures import ( + OKTA_CLIENT_ID, + OKTA_ORG_DOMAIN, + set_mocked_okta_provider, +) def _fake_policy( @@ -183,6 +188,65 @@ class Test_Signon_service: assert service.global_session_policies == {} + def test_skips_policy_fetch_when_scope_missing(self): + identity = OktaIdentityInfo( + org_domain=OKTA_ORG_DOMAIN, + client_id=OKTA_CLIENT_ID, + granted_scopes=["okta.brands.read"], # policies scope missing + ) + provider = set_mocked_okta_provider(identity=identity) + + list_policies_called = False + + async def fake_list_policies(*_a, **_k): + nonlocal list_policies_called + list_policies_called = True + return ([], _resp({}), None) + + with mock.patch( + "prowler.providers.okta.lib.service.service.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = fake_list_policies + mocked.list_brands = _empty_brands + mocked_client_cls.return_value = mocked + service = Signon(provider) + + assert list_policies_called is False + assert service.global_session_policies == {} + assert service.missing_scope["global_session_policies"] == "okta.policies.read" + assert service.missing_scope["sign_in_pages"] is None + + def test_unknown_granted_scopes_falls_back_to_attempting_fetch(self): + # When the JWT couldn't be decoded, granted_scopes is empty and the + # service must still attempt the fetch — preserves prior behavior. + identity = OktaIdentityInfo( + org_domain=OKTA_ORG_DOMAIN, + client_id=OKTA_CLIENT_ID, + granted_scopes=[], + ) + provider = set_mocked_okta_provider(identity=identity) + + list_policies_called = False + + async def fake_list_policies(*_a, **_k): + nonlocal list_policies_called + list_policies_called = True + return ([], _resp({}), None) + + with mock.patch( + "prowler.providers.okta.lib.service.service.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = fake_list_policies + mocked.list_brands = _empty_brands + mocked_client_cls.return_value = mocked + service = Signon(provider) + + assert list_policies_called is True + assert service.missing_scope["global_session_policies"] is None + assert service.missing_scope["sign_in_pages"] is None + class Test_Signon_service_brands: """Brand sign-in page fetching for the DOD banner check.""" @@ -308,6 +372,38 @@ class Test_Signon_service_brands: assert service.sign_in_pages == {} + def test_skips_brand_fetch_when_scope_missing(self): + identity = OktaIdentityInfo( + org_domain=OKTA_ORG_DOMAIN, + client_id=OKTA_CLIENT_ID, + granted_scopes=["okta.policies.read"], # brands scope missing + ) + provider = set_mocked_okta_provider(identity=identity) + + async def fake_list_policies(*_a, **_k): + return ([], _resp({}), None) + + list_brands_called = False + + async def fake_list_brands(*_a, **_k): + nonlocal list_brands_called + list_brands_called = True + return ([], _resp({}), None) + + with mock.patch( + "prowler.providers.okta.lib.service.service.OktaSDKClient" + ) as mocked_client_cls: + mocked = mock.MagicMock() + mocked.list_policies = fake_list_policies + mocked.list_brands = fake_list_brands + mocked_client_cls.return_value = mocked + service = Signon(provider) + + assert list_brands_called is False + assert service.sign_in_pages == {} + assert service.missing_scope["sign_in_pages"] == "okta.brands.read" + assert service.missing_scope["global_session_policies"] is None + def test_handles_multiple_brands(self): provider = set_mocked_okta_provider() brand_a = _fake_brand("brand-a", "Brand A")