mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
chore: add decode to return better status extended when no permission
This commit is contained in:
@@ -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):
|
||||
|
||||
@@ -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}",
|
||||
|
||||
@@ -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
|
||||
|
||||
+9
@@ -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",
|
||||
|
||||
+7
@@ -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 [
|
||||
|
||||
+7
@@ -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 [
|
||||
|
||||
+7
@@ -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 [
|
||||
|
||||
+7
@@ -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 [
|
||||
|
||||
@@ -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...")
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
):
|
||||
|
||||
+24
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
+14
@@ -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",
|
||||
|
||||
+14
@@ -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(
|
||||
[
|
||||
|
||||
+14
@@ -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(
|
||||
[
|
||||
|
||||
+14
@@ -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
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user