mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-09 21:14:22 +00:00
feat(okta): 4 new signon service checks (#11224)
This commit is contained in:
Daniel Barranquero
@@ -16,13 +16,14 @@ def set_mocked_okta_provider(
|
||||
session = OktaSession(
|
||||
org_domain=OKTA_ORG_DOMAIN,
|
||||
client_id=OKTA_CLIENT_ID,
|
||||
scopes=["okta.policies.read"],
|
||||
scopes=["okta.policies.read", "okta.brands.read"],
|
||||
private_key=OKTA_PRIVATE_KEY,
|
||||
)
|
||||
if identity is None:
|
||||
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
|
||||
):
|
||||
|
||||
+257
@@ -0,0 +1,257 @@
|
||||
from unittest import mock
|
||||
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
from tests.providers.okta.services.signon.signon_fixtures import (
|
||||
DOD_BANNER_HTML_SNIPPET,
|
||||
build_signon_client,
|
||||
sign_in_page,
|
||||
)
|
||||
|
||||
CHECK_PATH = (
|
||||
"prowler.providers.okta.services.signon."
|
||||
"signon_dod_warning_banner_configured."
|
||||
"signon_dod_warning_banner_configured.signon_client"
|
||||
)
|
||||
|
||||
|
||||
class Test_signon_dod_warning_banner_configured:
|
||||
def test_manual_when_no_brands_detected(self):
|
||||
signon_client = build_signon_client(sign_in_pages={})
|
||||
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 "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",
|
||||
brand_name="Primary",
|
||||
is_customized=True,
|
||||
page_content=f"<html><body>{DOD_BANNER_HTML_SNIPPET}</body></html>",
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 == "PASS"
|
||||
assert "DOD Notice and Consent Banner detected" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
assert "customized sign-in page" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_customized_page_missing_banner(self):
|
||||
page = sign_in_page(
|
||||
brand_id="brand-1",
|
||||
brand_name="Primary",
|
||||
is_customized=True,
|
||||
page_content="<html><body><h1>Welcome to ACME</h1></body></html>",
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 == "FAIL"
|
||||
assert "does not contain" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_default_page_contains_banner(self):
|
||||
page = sign_in_page(
|
||||
brand_id="brand-1",
|
||||
brand_name="Primary",
|
||||
is_customized=False,
|
||||
page_content=f"<html><body>{DOD_BANNER_HTML_SNIPPET}</body></html>",
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 == "PASS"
|
||||
assert "default sign-in page" in findings[0].status_extended
|
||||
|
||||
def test_manual_when_page_content_missing(self):
|
||||
page = sign_in_page(
|
||||
brand_id="brand-1",
|
||||
brand_name="Primary",
|
||||
is_customized=False,
|
||||
page_content=None,
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 "could not be retrieved from the Okta API" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_manual_when_fetch_error(self):
|
||||
page = sign_in_page(
|
||||
brand_id="brand-1",
|
||||
brand_name="Primary",
|
||||
is_customized=False,
|
||||
fetch_error="403 Forbidden: invalid_scope",
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 "Could not retrieve" in findings[0].status_extended
|
||||
assert "403" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_only_partial_banner_markers_are_present(self):
|
||||
page = sign_in_page(
|
||||
brand_id="brand-1",
|
||||
brand_name="Primary",
|
||||
is_customized=True,
|
||||
page_content=(
|
||||
"<html><body>This U.S. Government portal is for authorized use "
|
||||
"only.</body></html>"
|
||||
),
|
||||
)
|
||||
signon_client = build_signon_client(sign_in_pages={"brand-1": page})
|
||||
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 == "FAIL"
|
||||
assert "does not contain" in findings[0].status_extended
|
||||
|
||||
def test_emits_one_finding_per_brand(self):
|
||||
compliant = sign_in_page(
|
||||
brand_id="brand-prod",
|
||||
brand_name="Prod",
|
||||
is_customized=True,
|
||||
page_content=f"<html>{DOD_BANNER_HTML_SNIPPET}</html>",
|
||||
)
|
||||
missing = sign_in_page(
|
||||
brand_id="brand-sandbox",
|
||||
brand_name="Sandbox",
|
||||
is_customized=True,
|
||||
page_content="<html><body>No banner here</body></html>",
|
||||
)
|
||||
no_custom = sign_in_page(
|
||||
brand_id="brand-legacy",
|
||||
brand_name="Legacy",
|
||||
is_customized=False,
|
||||
page_content=f"<html>{DOD_BANNER_HTML_SNIPPET}</html>",
|
||||
)
|
||||
signon_client = build_signon_client(
|
||||
sign_in_pages={
|
||||
"brand-prod": compliant,
|
||||
"brand-sandbox": missing,
|
||||
"brand-legacy": no_custom,
|
||||
}
|
||||
)
|
||||
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) == 3
|
||||
by_brand = {f.resource_id: f.status for f in findings}
|
||||
assert by_brand == {
|
||||
"brand-prod": "PASS",
|
||||
"brand-sandbox": "FAIL",
|
||||
"brand-legacy": "PASS",
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Shared helpers for `signon` service check tests.
|
||||
|
||||
The original idle-timeout check test file defined these helpers locally;
|
||||
they were extracted here so the four checks added on top of the same
|
||||
service (`signon_global_session_lifetime_18h`,
|
||||
`signon_global_session_cookies_not_persistent`,
|
||||
`signon_global_session_policy_network_zone_enforced`,
|
||||
`signon_dod_warning_banner_configured`) can reuse them without copy-paste.
|
||||
"""
|
||||
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicy,
|
||||
GlobalSessionPolicyRule,
|
||||
SignInPage,
|
||||
)
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def default_policy(rules):
|
||||
return GlobalSessionPolicy(
|
||||
id="pol-default",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="ACTIVE",
|
||||
is_default=True,
|
||||
rules=rules,
|
||||
)
|
||||
|
||||
|
||||
def custom_policy(rules, name: str = "Admins Policy"):
|
||||
return GlobalSessionPolicy(
|
||||
id="pol-custom",
|
||||
name=name,
|
||||
priority=1,
|
||||
status="ACTIVE",
|
||||
is_default=False,
|
||||
rules=rules,
|
||||
)
|
||||
|
||||
|
||||
def default_rule(
|
||||
idle_min: int = 480,
|
||||
lifetime_min: int = None,
|
||||
use_persistent_cookie: bool = None,
|
||||
priority: int = 2,
|
||||
status: str = "ACTIVE",
|
||||
):
|
||||
return GlobalSessionPolicyRule(
|
||||
id="rule-default",
|
||||
name="Default Rule",
|
||||
priority=priority,
|
||||
status=status,
|
||||
is_default=True,
|
||||
max_session_idle_minutes=idle_min,
|
||||
max_session_lifetime_minutes=lifetime_min,
|
||||
use_persistent_cookie=use_persistent_cookie,
|
||||
)
|
||||
|
||||
|
||||
def non_default_rule(
|
||||
name: str,
|
||||
*,
|
||||
idle_min: int = None,
|
||||
lifetime_min: int = None,
|
||||
use_persistent_cookie: bool = None,
|
||||
network_zones_include: list = None,
|
||||
network_zones_exclude: list = None,
|
||||
priority: int = 1,
|
||||
status: str = "ACTIVE",
|
||||
):
|
||||
return GlobalSessionPolicyRule(
|
||||
id=f"rule-{name.lower().replace(' ', '-')}",
|
||||
name=name,
|
||||
priority=priority,
|
||||
status=status,
|
||||
is_default=False,
|
||||
max_session_idle_minutes=idle_min,
|
||||
max_session_lifetime_minutes=lifetime_min,
|
||||
use_persistent_cookie=use_persistent_cookie,
|
||||
network_zones_include=network_zones_include or [],
|
||||
network_zones_exclude=network_zones_exclude or [],
|
||||
)
|
||||
|
||||
|
||||
def sign_in_page(
|
||||
brand_id: str = "brand-1",
|
||||
brand_name: str = "Default Brand",
|
||||
is_customized: bool = True,
|
||||
page_content: str = None,
|
||||
fetch_error: str = None,
|
||||
):
|
||||
return SignInPage(
|
||||
brand_id=brand_id,
|
||||
brand_name=brand_name,
|
||||
is_customized=is_customized,
|
||||
page_content=page_content,
|
||||
fetch_error=fetch_error,
|
||||
)
|
||||
|
||||
|
||||
# Condensed DTM-08-060 banner that covers all four marker groups the check
|
||||
# requires (see BANNER_MARKER_GROUPS in the check module). Lets PASS tests
|
||||
# avoid pasting the full ~1300-char banner verbatim.
|
||||
DOD_BANNER_HTML_SNIPPET = (
|
||||
"<div>You are accessing a U.S. Government (USG) Information System "
|
||||
"(IS) that is provided for USG-authorized use only. "
|
||||
"Communications using, or data stored on, this IS may be intercepted, "
|
||||
"searched, monitored, and recorded.</div>"
|
||||
)
|
||||
+189
@@ -0,0 +1,189 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicy,
|
||||
GlobalSessionPolicyRule,
|
||||
)
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
from tests.providers.okta.services.signon.signon_fixtures import (
|
||||
build_signon_client,
|
||||
custom_policy,
|
||||
default_policy,
|
||||
default_rule,
|
||||
non_default_rule,
|
||||
)
|
||||
|
||||
CHECK_PATH = (
|
||||
"prowler.providers.okta.services.signon."
|
||||
"signon_global_session_cookies_not_persistent."
|
||||
"signon_global_session_cookies_not_persistent.signon_client"
|
||||
)
|
||||
|
||||
|
||||
def _run_check(signon_client):
|
||||
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_global_session_cookies_not_persistent.signon_global_session_cookies_not_persistent import (
|
||||
signon_global_session_cookies_not_persistent,
|
||||
)
|
||||
|
||||
return signon_global_session_cookies_not_persistent().execute()
|
||||
|
||||
|
||||
class Test_signon_global_session_cookies_not_persistent:
|
||||
def test_no_policies(self):
|
||||
findings = _run_check(build_signon_client({}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_priority_one_rule_disables_persistent_cookies(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Non-persistent cookies",
|
||||
use_persistent_cookie=False,
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "disables persistent global session cookies" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
assert "priority 99, default" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_priority_one_rule_uses_persistent_cookies(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Persistent cookies enabled",
|
||||
use_persistent_cookie=True,
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "allows persistent global session cookies" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_fail_when_priority_one_rule_does_not_assert_setting(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
GlobalSessionPolicyRule(
|
||||
id="rule-no-session",
|
||||
name="No Session Block",
|
||||
priority=1,
|
||||
status="ACTIVE",
|
||||
is_default=False,
|
||||
use_persistent_cookie=None,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "does not assert" in findings[0].status_extended
|
||||
|
||||
def test_emits_one_finding_per_policy(self):
|
||||
admins_policy = custom_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Sticky admin",
|
||||
use_persistent_cookie=True,
|
||||
priority=1,
|
||||
)
|
||||
],
|
||||
name="Admins Policy",
|
||||
)
|
||||
strict_default = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Non-persistent",
|
||||
use_persistent_cookie=False,
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-custom": admins_policy, "pol-default": strict_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 2
|
||||
by_name = {f.resource_name: f for f in findings}
|
||||
assert by_name["Admins Policy"].status == "FAIL"
|
||||
assert "priority 1, custom" in by_name["Admins Policy"].status_extended
|
||||
assert by_name["Default Policy"].status == "PASS"
|
||||
|
||||
def test_inactive_policy_is_skipped(self):
|
||||
inactive = GlobalSessionPolicy(
|
||||
id="pol-inactive",
|
||||
name="Disabled Policy",
|
||||
priority=1,
|
||||
status="INACTIVE",
|
||||
is_default=False,
|
||||
rules=[non_default_rule("Sticky", use_persistent_cookie=True, priority=1)],
|
||||
)
|
||||
active_default = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Non-persistent",
|
||||
use_persistent_cookie=False,
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-inactive": inactive, "pol-default": active_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 1
|
||||
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",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="INACTIVE",
|
||||
is_default=True,
|
||||
rules=[
|
||||
non_default_rule("Compliant", use_persistent_cookie=False, priority=1)
|
||||
],
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": only_inactive}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
+138
-232
@@ -5,6 +5,13 @@ from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicyRule,
|
||||
)
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
from tests.providers.okta.services.signon.signon_fixtures import (
|
||||
build_signon_client,
|
||||
custom_policy,
|
||||
default_policy,
|
||||
default_rule,
|
||||
non_default_rule,
|
||||
)
|
||||
|
||||
CHECK_PATH = (
|
||||
"prowler.providers.okta.services.signon."
|
||||
@@ -13,128 +20,53 @@ CHECK_PATH = (
|
||||
)
|
||||
|
||||
|
||||
def _build_signon_client(policies, audit_config: dict = None):
|
||||
client = mock.MagicMock()
|
||||
client.global_session_policies = policies
|
||||
client.provider = set_mocked_okta_provider()
|
||||
client.audit_config = audit_config or {}
|
||||
return client
|
||||
def _run_check(signon_client):
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
|
||||
def _default_policy(rules):
|
||||
return GlobalSessionPolicy(
|
||||
id="pol-default",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="ACTIVE",
|
||||
is_default=True,
|
||||
rules=rules,
|
||||
)
|
||||
|
||||
|
||||
def _custom_policy(rules):
|
||||
return GlobalSessionPolicy(
|
||||
id="pol-custom",
|
||||
name="Admins Policy",
|
||||
priority=1,
|
||||
status="ACTIVE",
|
||||
is_default=False,
|
||||
rules=rules,
|
||||
)
|
||||
|
||||
|
||||
def _default_rule(idle_min=480, priority=2, status="ACTIVE"):
|
||||
return GlobalSessionPolicyRule(
|
||||
id="rule-default",
|
||||
name="Default Rule",
|
||||
priority=priority,
|
||||
status=status,
|
||||
is_default=True,
|
||||
max_session_idle_minutes=idle_min,
|
||||
)
|
||||
|
||||
|
||||
def _non_default_rule(name, idle_min, priority=1, status="ACTIVE"):
|
||||
return GlobalSessionPolicyRule(
|
||||
id=f"rule-{name.lower().replace(' ', '-')}",
|
||||
name=name,
|
||||
priority=priority,
|
||||
status=status,
|
||||
is_default=False,
|
||||
max_session_idle_minutes=idle_min,
|
||||
)
|
||||
return signon_global_session_idle_timeout_15min().execute()
|
||||
|
||||
|
||||
class Test_signon_global_session_idle_timeout_15min:
|
||||
def test_no_policies(self):
|
||||
signon_client = _build_signon_client({})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "was not found" in findings[0].status_extended
|
||||
findings = _run_check(build_signon_client({}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_priority_one_non_default_rule_is_compliant(self):
|
||||
policy = _default_policy(
|
||||
policy = default_policy(
|
||||
[
|
||||
_non_default_rule("Strict 15min", 15, priority=1),
|
||||
_default_rule(priority=2),
|
||||
non_default_rule("Strict 15min", idle_min=15, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "Strict 15min" in findings[0].status_extended
|
||||
assert "Priority 1 non-default rule" in findings[0].status_extended
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "Strict 15min" in findings[0].status_extended
|
||||
assert "Default Policy" in findings[0].status_extended
|
||||
assert "priority 99, default" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_only_default_rule(self):
|
||||
policy = _default_policy([_default_rule(priority=1)])
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "uses 'Default Rule' as its active Priority 1 rule" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
policy = default_policy([default_rule(priority=1)])
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "uses 'Default Rule' as its active Priority 1 rule" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_fail_when_priority_one_non_default_rule_has_null_idle(self):
|
||||
# Rules without a session block leave max_session_idle_minutes as
|
||||
# None. The check must treat those as non-compliant — they cannot
|
||||
# enforce any timeout.
|
||||
policy = _default_policy(
|
||||
policy = default_policy(
|
||||
[
|
||||
GlobalSessionPolicyRule(
|
||||
id="rule-no-session",
|
||||
@@ -144,161 +76,135 @@ class Test_signon_global_session_idle_timeout_15min:
|
||||
is_default=False,
|
||||
max_session_idle_minutes=None,
|
||||
),
|
||||
_default_rule(priority=2),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No Session Block" in findings[0].status_extended
|
||||
assert "does not define" in findings[0].status_extended
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No Session Block" in findings[0].status_extended
|
||||
assert "does not define" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_priority_one_non_default_rule_exceeds_threshold(self):
|
||||
policy = _default_policy(
|
||||
policy = default_policy(
|
||||
[
|
||||
_non_default_rule("Loose 60min", 60, priority=1),
|
||||
_default_rule(priority=2),
|
||||
non_default_rule("Loose 60min", idle_min=60, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "Loose 60min" in findings[0].status_extended
|
||||
assert "exceeding the configured threshold" in findings[0].status_extended
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "Loose 60min" in findings[0].status_extended
|
||||
assert "exceeding the configured threshold" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_compliant_non_default_rule_is_not_priority_one(self):
|
||||
policy = _default_policy(
|
||||
policy = default_policy(
|
||||
[
|
||||
_default_rule(priority=1),
|
||||
_non_default_rule("Strict 15min", 15, priority=2),
|
||||
default_rule(priority=1),
|
||||
non_default_rule("Strict 15min", idle_min=15, priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "uses 'Default Rule' as its active Priority 1 rule" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "uses 'Default Rule' as its active Priority 1 rule" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_ignores_other_custom_policies(self):
|
||||
default_policy = _default_policy(
|
||||
def test_emits_one_finding_per_policy(self):
|
||||
# Custom policy at priority 1 with a permissive rule + Default Policy
|
||||
# with a strict rule -> two findings, ordered by policy priority.
|
||||
admins_policy = custom_policy(
|
||||
[
|
||||
_non_default_rule("Strict 15min", 15, priority=1),
|
||||
_default_rule(priority=2),
|
||||
non_default_rule("Admin Loose", idle_min=120, priority=1),
|
||||
default_rule(priority=2),
|
||||
],
|
||||
name="Admins Policy",
|
||||
)
|
||||
strict_default = default_policy(
|
||||
[
|
||||
non_default_rule("Strict 15min", idle_min=15, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
custom_policy = _custom_policy(
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-custom": admins_policy, "pol-default": strict_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 2
|
||||
by_name = {f.resource_name: f for f in findings}
|
||||
assert by_name["Admins Policy"].status == "FAIL"
|
||||
assert "priority 1, custom" in by_name["Admins Policy"].status_extended
|
||||
assert by_name["Default Policy"].status == "PASS"
|
||||
assert "priority 99, default" in by_name["Default Policy"].status_extended
|
||||
|
||||
def test_inactive_policy_is_skipped(self):
|
||||
inactive = GlobalSessionPolicy(
|
||||
id="pol-inactive",
|
||||
name="Disabled Policy",
|
||||
priority=1,
|
||||
status="INACTIVE",
|
||||
is_default=False,
|
||||
rules=[non_default_rule("Loose 120min", idle_min=120, priority=1)],
|
||||
)
|
||||
active_default = default_policy(
|
||||
[
|
||||
_non_default_rule("Loose Admin Rule", 60, priority=1),
|
||||
_default_rule(priority=2),
|
||||
non_default_rule("Strict 15min", idle_min=15, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client(
|
||||
{"pol-custom": custom_policy, "pol-default": default_policy}
|
||||
)
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-inactive": inactive, "pol-default": active_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].resource_name == "Default Policy"
|
||||
assert findings[0].status == "PASS"
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert findings[0].resource_name == "Default Policy"
|
||||
|
||||
def test_fail_when_default_policy_is_inactive(self):
|
||||
policy = GlobalSessionPolicy(
|
||||
def test_fail_when_all_policies_inactive(self):
|
||||
only_inactive = GlobalSessionPolicy(
|
||||
id="pol-default",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="INACTIVE",
|
||||
is_default=True,
|
||||
rules=[_non_default_rule("Strict 15min", 15, priority=1)],
|
||||
rules=[non_default_rule("Strict 15min", idle_min=15, priority=1)],
|
||||
)
|
||||
signon_client = _build_signon_client({"pol-default": policy})
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": only_inactive}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "status 'INACTIVE'" 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):
|
||||
# 30-minute rule fails the STIG default of 15, but passes a relaxed
|
||||
# threshold of 60 minutes set in audit_config.
|
||||
policy = _default_policy(
|
||||
policy = default_policy(
|
||||
[
|
||||
_non_default_rule("Relaxed 30min", 30, priority=1),
|
||||
_default_rule(priority=2),
|
||||
non_default_rule("Relaxed 30min", idle_min=30, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
signon_client = _build_signon_client(
|
||||
{"pol-default": policy},
|
||||
audit_config={"okta_max_session_idle_minutes": 60},
|
||||
)
|
||||
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_global_session_idle_timeout_15min.signon_global_session_idle_timeout_15min import (
|
||||
signon_global_session_idle_timeout_15min,
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-default": policy},
|
||||
audit_config={"okta_max_session_idle_minutes": 60},
|
||||
)
|
||||
|
||||
findings = signon_global_session_idle_timeout_15min().execute()
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "threshold of 60 minutes" in findings[0].status_extended
|
||||
)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "threshold of 60 minutes" in findings[0].status_extended
|
||||
|
||||
+212
@@ -0,0 +1,212 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicy,
|
||||
GlobalSessionPolicyRule,
|
||||
)
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
from tests.providers.okta.services.signon.signon_fixtures import (
|
||||
build_signon_client,
|
||||
custom_policy,
|
||||
default_policy,
|
||||
default_rule,
|
||||
non_default_rule,
|
||||
)
|
||||
|
||||
CHECK_PATH = (
|
||||
"prowler.providers.okta.services.signon."
|
||||
"signon_global_session_lifetime_18h."
|
||||
"signon_global_session_lifetime_18h.signon_client"
|
||||
)
|
||||
|
||||
|
||||
def _run_check(signon_client):
|
||||
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_global_session_lifetime_18h.signon_global_session_lifetime_18h import (
|
||||
signon_global_session_lifetime_18h,
|
||||
)
|
||||
|
||||
return signon_global_session_lifetime_18h().execute()
|
||||
|
||||
|
||||
class Test_signon_global_session_lifetime_18h:
|
||||
def test_no_policies(self):
|
||||
findings = _run_check(build_signon_client({}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_priority_one_non_default_rule_is_compliant(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("18h rule", lifetime_min=1080, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "18h rule" in findings[0].status_extended
|
||||
assert "1080 minutes" in findings[0].status_extended
|
||||
assert "priority 99, default" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_lifetime_exceeds_threshold(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("Loose 24h rule", lifetime_min=1440, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "1440 minutes" in findings[0].status_extended
|
||||
assert "exceeding the configured threshold" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_priority_one_rule_has_no_lifetime(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
GlobalSessionPolicyRule(
|
||||
id="rule-no-session",
|
||||
name="No Session Block",
|
||||
priority=1,
|
||||
status="ACTIVE",
|
||||
is_default=False,
|
||||
max_session_lifetime_minutes=None,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "does not define" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_lifetime_is_disabled_with_zero(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("Unlimited Lifetime", lifetime_min=0, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "0 minutes" in findings[0].status_extended
|
||||
assert "disables the maximum Okta global session lifetime" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_fail_when_default_rule_is_priority_one(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
default_rule(priority=1, lifetime_min=1080),
|
||||
non_default_rule("Compliant", lifetime_min=1080, priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "uses 'Default Rule' as its active Priority 1 rule" in (
|
||||
findings[0].status_extended
|
||||
)
|
||||
|
||||
def test_emits_one_finding_per_policy(self):
|
||||
admins_policy = custom_policy(
|
||||
[
|
||||
non_default_rule("Admin Long Lived", lifetime_min=2880, priority=1),
|
||||
default_rule(priority=2),
|
||||
],
|
||||
name="Admins Policy",
|
||||
)
|
||||
strict_default = default_policy(
|
||||
[
|
||||
non_default_rule("18h rule", lifetime_min=1080, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-custom": admins_policy, "pol-default": strict_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 2
|
||||
by_name = {f.resource_name: f for f in findings}
|
||||
assert by_name["Admins Policy"].status == "FAIL"
|
||||
assert "priority 1, custom" in by_name["Admins Policy"].status_extended
|
||||
assert by_name["Default Policy"].status == "PASS"
|
||||
|
||||
def test_inactive_policy_is_skipped(self):
|
||||
inactive = GlobalSessionPolicy(
|
||||
id="pol-inactive",
|
||||
name="Disabled Policy",
|
||||
priority=1,
|
||||
status="INACTIVE",
|
||||
is_default=False,
|
||||
rules=[non_default_rule("Loose", lifetime_min=2880, priority=1)],
|
||||
)
|
||||
active_default = default_policy(
|
||||
[
|
||||
non_default_rule("18h rule", lifetime_min=1080, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-inactive": inactive, "pol-default": active_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].resource_name == "Default Policy"
|
||||
assert findings[0].status == "PASS"
|
||||
|
||||
def test_fail_when_all_policies_inactive(self):
|
||||
only_inactive = GlobalSessionPolicy(
|
||||
id="pol-default",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="INACTIVE",
|
||||
is_default=True,
|
||||
rules=[non_default_rule("18h rule", lifetime_min=1080, priority=1)],
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": only_inactive}))
|
||||
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
|
||||
|
||||
def test_threshold_overridden_via_audit_config(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("Relaxed 24h", lifetime_min=1440, priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-default": policy},
|
||||
audit_config={"okta_max_session_lifetime_minutes": 1440},
|
||||
)
|
||||
)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "threshold of 1440 minutes" in findings[0].status_extended
|
||||
+222
@@ -0,0 +1,222 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicy,
|
||||
GlobalSessionPolicyRule,
|
||||
)
|
||||
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
|
||||
from tests.providers.okta.services.signon.signon_fixtures import (
|
||||
build_signon_client,
|
||||
custom_policy,
|
||||
default_policy,
|
||||
default_rule,
|
||||
non_default_rule,
|
||||
)
|
||||
|
||||
CHECK_PATH = (
|
||||
"prowler.providers.okta.services.signon."
|
||||
"signon_global_session_policy_network_zone_enforced."
|
||||
"signon_global_session_policy_network_zone_enforced.signon_client"
|
||||
)
|
||||
|
||||
|
||||
def _run_check(signon_client):
|
||||
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_global_session_policy_network_zone_enforced.signon_global_session_policy_network_zone_enforced import (
|
||||
signon_global_session_policy_network_zone_enforced,
|
||||
)
|
||||
|
||||
return signon_global_session_policy_network_zone_enforced().execute()
|
||||
|
||||
|
||||
class Test_signon_global_session_policy_network_zone_enforced:
|
||||
def test_no_policies(self):
|
||||
findings = _run_check(build_signon_client({}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No active Okta Global Session Policies" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_priority_one_non_default_rule_includes_zone(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Allow-from-VPN",
|
||||
network_zones_include=["zone-corp"],
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "Allow-from-VPN" in findings[0].status_extended
|
||||
assert "non-default rule" in findings[0].status_extended
|
||||
assert "priority 99, default" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_priority_one_non_default_rule_excludes_zone(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Block-blacklist",
|
||||
network_zones_exclude=["zone-blocked"],
|
||||
priority=1,
|
||||
),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "Block-blacklist" in findings[0].status_extended
|
||||
|
||||
def test_pass_when_only_default_rule_has_zones(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
GlobalSessionPolicyRule(
|
||||
id="rule-default-zoned",
|
||||
name="Default Rule",
|
||||
priority=1,
|
||||
status="ACTIVE",
|
||||
is_default=True,
|
||||
network_zones_include=["zone-corp"],
|
||||
),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "PASS"
|
||||
assert "built-in Default Rule" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_priority_one_rule_has_no_zones(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("Plain non-default", priority=1),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "Plain non-default" in findings[0].status_extended
|
||||
assert "does not map" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_only_lower_priority_rule_has_zones(self):
|
||||
policy = default_policy(
|
||||
[
|
||||
non_default_rule("No-zones top", priority=1),
|
||||
non_default_rule(
|
||||
"Zoned-but-low",
|
||||
network_zones_include=["zone-corp"],
|
||||
priority=2,
|
||||
),
|
||||
default_rule(priority=3),
|
||||
]
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "No-zones top" in findings[0].status_extended
|
||||
|
||||
def test_fail_when_only_default_rule_has_no_zones(self):
|
||||
policy = default_policy([default_rule(priority=1)])
|
||||
findings = _run_check(build_signon_client({"pol-default": policy}))
|
||||
assert len(findings) == 1
|
||||
assert findings[0].status == "FAIL"
|
||||
assert "built-in Default Rule" in findings[0].status_extended
|
||||
|
||||
def test_emits_one_finding_per_policy(self):
|
||||
admins_policy = custom_policy(
|
||||
[
|
||||
non_default_rule("No-zones admin", priority=1),
|
||||
default_rule(priority=2),
|
||||
],
|
||||
name="Admins Policy",
|
||||
)
|
||||
zoned_default = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Allow-corp",
|
||||
network_zones_include=["zone-corp"],
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-custom": admins_policy, "pol-default": zoned_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 2
|
||||
by_name = {f.resource_name: f for f in findings}
|
||||
assert by_name["Admins Policy"].status == "FAIL"
|
||||
assert "priority 1, custom" in by_name["Admins Policy"].status_extended
|
||||
assert by_name["Default Policy"].status == "PASS"
|
||||
|
||||
def test_inactive_policy_is_skipped(self):
|
||||
inactive = GlobalSessionPolicy(
|
||||
id="pol-inactive",
|
||||
name="Disabled Policy",
|
||||
priority=1,
|
||||
status="INACTIVE",
|
||||
is_default=False,
|
||||
rules=[non_default_rule("No-zones", priority=1)],
|
||||
)
|
||||
active_default = default_policy(
|
||||
[
|
||||
non_default_rule(
|
||||
"Allow-corp",
|
||||
network_zones_include=["zone-corp"],
|
||||
priority=1,
|
||||
),
|
||||
default_rule(priority=2),
|
||||
]
|
||||
)
|
||||
findings = _run_check(
|
||||
build_signon_client(
|
||||
{"pol-inactive": inactive, "pol-default": active_default}
|
||||
)
|
||||
)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].resource_name == "Default Policy"
|
||||
assert findings[0].status == "PASS"
|
||||
|
||||
def test_fail_when_all_policies_inactive(self):
|
||||
only_inactive = GlobalSessionPolicy(
|
||||
id="pol-default",
|
||||
name="Default Policy",
|
||||
priority=99,
|
||||
status="INACTIVE",
|
||||
is_default=True,
|
||||
rules=[
|
||||
non_default_rule(
|
||||
"Allow-corp",
|
||||
network_zones_include=["zone-corp"],
|
||||
priority=1,
|
||||
)
|
||||
],
|
||||
)
|
||||
findings = _run_check(build_signon_client({"pol-default": only_inactive}))
|
||||
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,12 +1,18 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.okta.models import OktaIdentityInfo
|
||||
from prowler.providers.okta.services.signon.signon_service import (
|
||||
GlobalSessionPolicy,
|
||||
GlobalSessionPolicyRule,
|
||||
SignInPage,
|
||||
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(
|
||||
@@ -48,12 +54,29 @@ def _fake_rule(
|
||||
return r
|
||||
|
||||
|
||||
def _fake_brand(brand_id: str, name: str):
|
||||
b = mock.MagicMock()
|
||||
b.id = brand_id
|
||||
b.name = name
|
||||
return b
|
||||
|
||||
|
||||
def _fake_sign_in_page(page_content: str):
|
||||
p = mock.MagicMock()
|
||||
p.page_content = page_content
|
||||
return p
|
||||
|
||||
|
||||
def _resp(headers: dict = None):
|
||||
r = mock.MagicMock()
|
||||
r.headers = headers or {}
|
||||
return r
|
||||
|
||||
|
||||
async def _empty_brands(*_a, **_k):
|
||||
return ([], _resp({}), None)
|
||||
|
||||
|
||||
class Test_next_after_cursor:
|
||||
def test_no_resp_returns_none(self):
|
||||
assert _next_after_cursor(None) is None
|
||||
@@ -97,6 +120,7 @@ class Test_Signon_service:
|
||||
mocked = mock.MagicMock()
|
||||
mocked.list_policies = fake_list_policies
|
||||
mocked.list_policy_rules = fake_list_rules
|
||||
mocked.list_brands = _empty_brands
|
||||
mocked_client_cls.return_value = mocked
|
||||
|
||||
service = Signon(provider)
|
||||
@@ -140,6 +164,7 @@ class Test_Signon_service:
|
||||
mocked = mock.MagicMock()
|
||||
mocked.list_policies = fake_list_policies
|
||||
mocked.list_policy_rules = fake_list_rules
|
||||
mocked.list_brands = _empty_brands
|
||||
mocked_client_cls.return_value = mocked
|
||||
service = Signon(provider)
|
||||
|
||||
@@ -157,7 +182,251 @@ class Test_Signon_service:
|
||||
) as mocked_client_cls:
|
||||
mocked = mock.MagicMock()
|
||||
mocked.list_policies = failing
|
||||
mocked.list_brands = _empty_brands
|
||||
mocked_client_cls.return_value = mocked
|
||||
service = Signon(provider)
|
||||
|
||||
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."""
|
||||
|
||||
def _build_with_brands(
|
||||
self,
|
||||
provider,
|
||||
brands_response,
|
||||
sign_in_page_responses: dict,
|
||||
default_sign_in_page_responses: dict | None = None,
|
||||
):
|
||||
async def fake_list_policies(*_a, **_k):
|
||||
return ([], _resp({}), None)
|
||||
|
||||
async def fake_list_brands(*_a, **_k):
|
||||
return brands_response
|
||||
|
||||
async def fake_get_sign_in_page(brand_id, *_a, **_k):
|
||||
return sign_in_page_responses[brand_id]
|
||||
|
||||
async def fake_get_default_sign_in_page(brand_id, *_a, **_k):
|
||||
return default_sign_in_page_responses[brand_id]
|
||||
|
||||
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.get_customized_sign_in_page = fake_get_sign_in_page
|
||||
mocked.get_default_sign_in_page = fake_get_default_sign_in_page
|
||||
mocked_client_cls.return_value = mocked
|
||||
return Signon(provider)
|
||||
|
||||
def test_fetches_brand_with_customized_page(self):
|
||||
provider = set_mocked_okta_provider()
|
||||
brand = _fake_brand("brand-1", "Primary")
|
||||
page = _fake_sign_in_page("<html>banner here</html>")
|
||||
service = self._build_with_brands(
|
||||
provider,
|
||||
brands_response=([brand], _resp({}), None),
|
||||
sign_in_page_responses={"brand-1": (page, _resp({}), None)},
|
||||
)
|
||||
|
||||
assert "brand-1" in service.sign_in_pages
|
||||
result = service.sign_in_pages["brand-1"]
|
||||
assert isinstance(result, SignInPage)
|
||||
assert result.is_customized is True
|
||||
assert result.page_content == "<html>banner here</html>"
|
||||
assert result.fetch_error is None
|
||||
|
||||
def test_404_falls_back_to_default_sign_in_page(self):
|
||||
provider = set_mocked_okta_provider()
|
||||
brand = _fake_brand("brand-1", "Primary")
|
||||
default_page = _fake_sign_in_page("<html>default banner here</html>")
|
||||
service = self._build_with_brands(
|
||||
provider,
|
||||
brands_response=([brand], _resp({}), None),
|
||||
sign_in_page_responses={
|
||||
"brand-1": (None, _resp({}), Exception("404 Not Found"))
|
||||
},
|
||||
default_sign_in_page_responses={"brand-1": (default_page, _resp({}), None)},
|
||||
)
|
||||
|
||||
assert service.sign_in_pages["brand-1"].is_customized is False
|
||||
assert service.sign_in_pages["brand-1"].fetch_error is None
|
||||
assert (
|
||||
service.sign_in_pages["brand-1"].page_content
|
||||
== "<html>default banner here</html>"
|
||||
)
|
||||
|
||||
def test_default_sign_in_page_error_captured_when_customized_page_missing(self):
|
||||
provider = set_mocked_okta_provider()
|
||||
brand = _fake_brand("brand-1", "Primary")
|
||||
service = self._build_with_brands(
|
||||
provider,
|
||||
brands_response=([brand], _resp({}), None),
|
||||
sign_in_page_responses={
|
||||
"brand-1": (None, _resp({}), Exception("404 Not Found"))
|
||||
},
|
||||
default_sign_in_page_responses={
|
||||
"brand-1": (None, _resp({}), Exception("403 Forbidden"))
|
||||
},
|
||||
)
|
||||
|
||||
result = service.sign_in_pages["brand-1"]
|
||||
assert result.is_customized is False
|
||||
assert "403" in result.fetch_error
|
||||
|
||||
def test_403_captured_into_fetch_error(self):
|
||||
provider = set_mocked_okta_provider()
|
||||
brand = _fake_brand("brand-1", "Primary")
|
||||
service = self._build_with_brands(
|
||||
provider,
|
||||
brands_response=([brand], _resp({}), None),
|
||||
sign_in_page_responses={
|
||||
"brand-1": (None, _resp({}), Exception("403 Forbidden: invalid_scope"))
|
||||
},
|
||||
default_sign_in_page_responses={},
|
||||
)
|
||||
|
||||
result = service.sign_in_pages["brand-1"]
|
||||
assert result.is_customized is False
|
||||
assert "403" in result.fetch_error
|
||||
|
||||
def test_returns_empty_on_brands_api_error(self):
|
||||
provider = set_mocked_okta_provider()
|
||||
|
||||
async def fake_list_policies(*_a, **_k):
|
||||
return ([], _resp({}), None)
|
||||
|
||||
async def failing_brands(*_a, **_k):
|
||||
return ([], _resp({}), Exception("Brands API unavailable"))
|
||||
|
||||
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 = failing_brands
|
||||
mocked_client_cls.return_value = mocked
|
||||
service = Signon(provider)
|
||||
|
||||
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")
|
||||
brand_b = _fake_brand("brand-b", "Brand B")
|
||||
page_a = _fake_sign_in_page("<html>A</html>")
|
||||
|
||||
service = self._build_with_brands(
|
||||
provider,
|
||||
brands_response=([brand_a, brand_b], _resp({}), None),
|
||||
sign_in_page_responses={
|
||||
"brand-a": (page_a, _resp({}), None),
|
||||
"brand-b": (None, _resp({}), Exception("404 not found")),
|
||||
},
|
||||
default_sign_in_page_responses={
|
||||
"brand-b": (
|
||||
_fake_sign_in_page("<html>default B</html>"),
|
||||
_resp({}),
|
||||
None,
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
assert set(service.sign_in_pages.keys()) == {"brand-a", "brand-b"}
|
||||
assert service.sign_in_pages["brand-a"].page_content == "<html>A</html>"
|
||||
assert service.sign_in_pages["brand-b"].is_customized is False
|
||||
assert service.sign_in_pages["brand-b"].page_content == "<html>default B</html>"
|
||||
|
||||
Reference in New Issue
Block a user