ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 04:51:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: use external file for shared code

Browse Source
This commit is contained in:
Daniel Barranquero
2026-05-20 17:55:12 +02:00
parent 64f04dc5db
commit 15d3040155
6 changed files with 206 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/okta/services/signon/lib/__init__.py
View File
prowler/providers/okta/services/signon/lib/signon_helpers.py
+92
View File
@@ -0,0 +1,92 @@
"""Shared helpers for the OKTA sign-on STIG checks.
The four `signon_global_session_*` checks share the same plumbing:
they iterate active Global Session Policies in priority order, locate
each policy's Priority 1 active rule, and emit one finding per policy.
This module centralises that plumbing so each check can stay focused
on its STIG-specific predicate.
"""
from typing import Optional
from prowler.lib.check.models import CheckReportOkta
from prowler.providers.okta.services.signon.signon_service import (
GlobalSessionPolicy,
GlobalSessionPolicyRule,
)
def active_policies(
global_session_policies: dict[str, GlobalSessionPolicy],
) -> list[GlobalSessionPolicy]:
"""Return active policies sorted by priority (ascending, name as tiebreaker).
A policy with no `status` is treated as ACTIVE because the Okta SDK
sometimes omits the field on default policies.
"""
return sorted(
[
policy
for policy in global_session_policies.values()
if not policy.status or policy.status.upper() == "ACTIVE"
],
key=lambda policy: (
policy.priority if policy.priority is not None else float("inf"),
policy.name,
),
)
def priority_one_active_rule(
policy: GlobalSessionPolicy,
) -> Optional[GlobalSessionPolicyRule]:
"""Return the policy's Priority 1 active rule, or None.
Okta's evaluator skips inactive rules, so we first filter to active
rules and pick the highest-priority one. If that rule is not at
priority 1 we return None — the policy effectively has no
priority-1 rule for evaluation purposes.
"""
active_rules = sorted(
[
rule
for rule in policy.rules
if not rule.status or rule.status.upper() == "ACTIVE"
],
key=lambda rule: (
rule.priority if rule.priority is not None else float("inf"),
rule.name,
),
)
if not active_rules:
return None
candidate = active_rules[0]
if candidate.priority != 1:
return None
return candidate
def policy_label(policy: GlobalSessionPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Global Session Policy '{policy.name}' (priority {priority}, {kind})"
def no_active_policies_finding(
metadata, org_domain: str, status_extended: str
) -> CheckReportOkta:
"""Build the FAIL finding emitted when no active sign-on policies exist."""
placeholder = GlobalSessionPolicy(
id="signon-policies-missing",
name="(no active sign-on policies)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = status_extended
return report
prowler/providers/okta/services/signon/signon_global_session_cookies_not_persistent/signon_global_session_cookies_not_persistent.py
+29 -76
View File
@@ -1,4 +1,10 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
@@ -20,12 +26,21 @@ class signon_global_session_cookies_not_persistent(Check):
def execute(self) -> list[CheckReportOkta]:
org_domain = signon_client.provider.identity.org_domain
active_policies = _active_policies()
if not active_policies:
return [_no_policies_finding(self.metadata(), org_domain)]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
no_active_policies_finding(
self.metadata(),
org_domain,
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273206 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule that disables persistent global "
"session cookies.",
)
]
findings: list[CheckReportOkta] = []
for policy in active_policies:
for policy in policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
@@ -37,10 +52,10 @@ class signon_global_session_cookies_not_persistent(Check):
def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
label = _policy_label(policy)
priority_one_rule = _priority_one_active_rule(policy)
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if priority_one_rule is None:
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273206 requires "
@@ -48,18 +63,18 @@ def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
"session cookies.",
)
if priority_one_rule.is_default or priority_one_rule.name == "Default Rule":
if rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{priority_one_rule.name}' as its active Priority 1 "
"rule. The STIG requires a non-default Priority 1 rule.",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
use_persistent_cookie = priority_one_rule.use_persistent_cookie
use_persistent_cookie = rule.use_persistent_cookie
if use_persistent_cookie is None:
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"does not assert the 'Okta global session cookies persist across "
"browser sessions' setting.",
)
@@ -67,74 +82,12 @@ def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
if use_persistent_cookie is False:
return (
"PASS",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"disables persistent global session cookies.",
)
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"allows persistent global session cookies, leaving the session active "
"across browser restarts.",
)
def _active_policies() -> list[GlobalSessionPolicy]:
return sorted(
[
policy
for policy in signon_client.global_session_policies.values()
if not policy.status or policy.status.upper() == "ACTIVE"
],
key=lambda policy: (
policy.priority if policy.priority is not None else float("inf"),
policy.name,
),
)
def _priority_one_active_rule(policy: GlobalSessionPolicy):
active_rules = sorted(
[
rule
for rule in policy.rules
if not rule.status or rule.status.upper() == "ACTIVE"
],
key=lambda rule: (
rule.priority if rule.priority is not None else float("inf"),
rule.name,
),
)
if not active_rules:
return None
candidate = active_rules[0]
if candidate.priority != 1:
return None
return candidate
def _policy_label(policy: GlobalSessionPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Global Session Policy '{policy.name}' (priority {priority}, {kind})"
def _no_policies_finding(metadata, org_domain) -> CheckReportOkta:
placeholder = GlobalSessionPolicy(
id="signon-policies-missing",
name="(no active sign-on policies)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = (
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273206 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule that disables persistent global "
"session cookies."
)
return report
prowler/providers/okta/services/signon/signon_global_session_idle_timeout_15min/signon_global_session_idle_timeout_15min.py
+28 -75
View File
@@ -1,4 +1,10 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
@@ -27,12 +33,20 @@ class signon_global_session_idle_timeout_15min(Check):
)
org_domain = signon_client.provider.identity.org_domain
active_policies = _active_policies()
if not active_policies:
return [_no_policies_finding(self.metadata(), org_domain)]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
no_active_policies_finding(
self.metadata(),
org_domain,
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273186 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule with a 15-minute idle timeout.",
)
]
findings: list[CheckReportOkta] = []
for policy in active_policies:
for policy in policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
@@ -44,10 +58,10 @@ class signon_global_session_idle_timeout_15min(Check):
def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str, str]:
label = _policy_label(policy)
priority_one_rule = _priority_one_active_rule(policy)
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if priority_one_rule is None:
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273186 requires "
@@ -55,92 +69,31 @@ def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str,
"minutes.",
)
if priority_one_rule.is_default or priority_one_rule.name == "Default Rule":
if rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{priority_one_rule.name}' as its active Priority 1 "
"rule. The STIG requires a non-default Priority 1 rule.",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
idle_timeout = priority_one_rule.max_session_idle_minutes
idle_timeout = rule.max_session_idle_minutes
if idle_timeout is None:
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"does not define a maximum Okta global session idle time.",
)
if idle_timeout <= threshold:
return (
"PASS",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
f"sets the maximum Okta global session idle time to {idle_timeout} "
f"minutes, meeting the configured threshold of {threshold} minutes.",
)
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
f"sets the maximum Okta global session idle time to {idle_timeout} "
f"minutes, exceeding the configured threshold of {threshold} minutes.",
)
def _active_policies() -> list[GlobalSessionPolicy]:
return sorted(
[
policy
for policy in signon_client.global_session_policies.values()
if not policy.status or policy.status.upper() == "ACTIVE"
],
key=lambda policy: (
policy.priority if policy.priority is not None else float("inf"),
policy.name,
),
)
def _priority_one_active_rule(policy: GlobalSessionPolicy):
active_rules = sorted(
[
rule
for rule in policy.rules
if not rule.status or rule.status.upper() == "ACTIVE"
],
key=lambda rule: (
rule.priority if rule.priority is not None else float("inf"),
rule.name,
),
)
if not active_rules:
return None
candidate = active_rules[0]
if candidate.priority != 1:
return None
return candidate
def _policy_label(policy: GlobalSessionPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Global Session Policy '{policy.name}' (priority {priority}, {kind})"
def _no_policies_finding(metadata, org_domain) -> CheckReportOkta:
placeholder = GlobalSessionPolicy(
id="signon-policies-missing",
name="(no active sign-on policies)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = (
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273186 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule with a 15-minute idle timeout."
)
return report
prowler/providers/okta/services/signon/signon_global_session_lifetime_18h/signon_global_session_lifetime_18h.py
+29 -76
View File
@@ -1,4 +1,10 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
@@ -27,12 +33,20 @@ class signon_global_session_lifetime_18h(Check):
)
org_domain = signon_client.provider.identity.org_domain
active_policies = _active_policies()
if not active_policies:
return [_no_policies_finding(self.metadata(), org_domain)]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
no_active_policies_finding(
self.metadata(),
org_domain,
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273203 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule with an 18-hour session lifetime.",
)
]
findings: list[CheckReportOkta] = []
for policy in active_policies:
for policy in policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
@@ -44,10 +58,10 @@ class signon_global_session_lifetime_18h(Check):
def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str, str]:
label = _policy_label(policy)
priority_one_rule = _priority_one_active_rule(policy)
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if priority_one_rule is None:
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273203 requires "
@@ -55,25 +69,25 @@ def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str,
"minutes.",
)
if priority_one_rule.is_default or priority_one_rule.name == "Default Rule":
if rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{priority_one_rule.name}' as its active Priority 1 "
"rule. The STIG requires a non-default Priority 1 rule.",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
lifetime = priority_one_rule.max_session_lifetime_minutes
lifetime = rule.max_session_lifetime_minutes
if lifetime is None:
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"does not define a maximum Okta global session lifetime.",
)
if lifetime == 0:
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
"disables the maximum Okta global session lifetime by setting it "
"to 0 minutes.",
)
@@ -81,74 +95,13 @@ def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str,
if lifetime <= threshold:
return (
"PASS",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
f"sets the maximum Okta global session lifetime to {lifetime} "
f"minutes, meeting the configured threshold of {threshold} minutes.",
)
return (
"FAIL",
f"Priority 1 non-default rule '{priority_one_rule.name}' in {label} "
f"Priority 1 non-default rule '{rule.name}' in {label} "
f"sets the maximum Okta global session lifetime to {lifetime} minutes, "
f"exceeding the configured threshold of {threshold} minutes.",
)
def _active_policies() -> list[GlobalSessionPolicy]:
return sorted(
[
policy
for policy in signon_client.global_session_policies.values()
if not policy.status or policy.status.upper() == "ACTIVE"
],
key=lambda policy: (
policy.priority if policy.priority is not None else float("inf"),
policy.name,
),
)
def _priority_one_active_rule(policy: GlobalSessionPolicy):
active_rules = sorted(
[
rule
for rule in policy.rules
if not rule.status or rule.status.upper() == "ACTIVE"
],
key=lambda rule: (
rule.priority if rule.priority is not None else float("inf"),
rule.name,
),
)
if not active_rules:
return None
candidate = active_rules[0]
if candidate.priority != 1:
return None
return candidate
def _policy_label(policy: GlobalSessionPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Global Session Policy '{policy.name}' (priority {priority}, {kind})"
def _no_policies_finding(metadata, org_domain) -> CheckReportOkta:
placeholder = GlobalSessionPolicy(
id="signon-policies-missing",
name="(no active sign-on policies)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = (
"No active Okta Global Session Policies were returned by the API. "
"STIG V-273203 requires the policy that governs each user to enforce "
"a Priority 1 non-default rule with an 18-hour session lifetime."
)
return report
prowler/providers/okta/services/signon/signon_global_session_policy_network_zone_enforced/signon_global_session_policy_network_zone_enforced.py
+28 -78
View File
@@ -1,4 +1,10 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
@@ -25,12 +31,20 @@ class signon_global_session_policy_network_zone_enforced(Check):
def execute(self) -> list[CheckReportOkta]:
org_domain = signon_client.provider.identity.org_domain
active_policies = _active_policies()
if not active_policies:
return [_no_policies_finding(self.metadata(), org_domain)]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
no_active_policies_finding(
self.metadata(),
org_domain,
"No active Okta Global Session Policies were returned by the API. "
"STIG V-279691 requires the policy that governs each user to map "
"User's IP to a Network Zone on its Priority 1 active rule.",
)
]
findings: list[CheckReportOkta] = []
for policy in active_policies:
for policy in policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
@@ -42,10 +56,10 @@ class signon_global_session_policy_network_zone_enforced(Check):
def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
label = _policy_label(policy)
priority_one_rule = _priority_one_active_rule(policy)
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if priority_one_rule is None:
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-279691 requires "
@@ -55,84 +69,20 @@ def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
rule_kind = (
"built-in Default Rule"
if priority_one_rule.is_default or priority_one_rule.name == "Default Rule"
if rule.is_default or rule.name == "Default Rule"
else "non-default rule"
)
has_zones = bool(
priority_one_rule.network_zones_include
or priority_one_rule.network_zones_exclude
)
has_zones = bool(rule.network_zones_include or rule.network_zones_exclude)
if has_zones:
return (
"PASS",
f"Priority 1 active {rule_kind} '{priority_one_rule.name}' in "
f"{label} maps User's IP to a Network Zone.",
f"Priority 1 active {rule_kind} '{rule.name}' in {label} maps "
"User's IP to a Network Zone.",
)
return (
"FAIL",
f"Priority 1 active {rule_kind} '{priority_one_rule.name}' in {label} "
"does not map User's IP to a Network Zone. The policy cannot allow "
"or deny access based on the organization's Access Control Policy.",
f"Priority 1 active {rule_kind} '{rule.name}' in {label} does not "
"map User's IP to a Network Zone. The policy cannot allow or deny "
"access based on the organization's Access Control Policy.",
)
def _active_policies() -> list[GlobalSessionPolicy]:
return sorted(
[
policy
for policy in signon_client.global_session_policies.values()
if not policy.status or policy.status.upper() == "ACTIVE"
],
key=lambda policy: (
policy.priority if policy.priority is not None else float("inf"),
policy.name,
),
)
def _priority_one_active_rule(policy: GlobalSessionPolicy):
active_rules = sorted(
[
rule
for rule in policy.rules
if not rule.status or rule.status.upper() == "ACTIVE"
],
key=lambda rule: (
rule.priority if rule.priority is not None else float("inf"),
rule.name,
),
)
if not active_rules:
return None
candidate = active_rules[0]
if candidate.priority != 1:
return None
return candidate
def _policy_label(policy: GlobalSessionPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Global Session Policy '{policy.name}' (priority {priority}, {kind})"
def _no_policies_finding(metadata, org_domain) -> CheckReportOkta:
placeholder = GlobalSessionPolicy(
id="signon-policies-missing",
name="(no active sign-on policies)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = (
"No active Okta Global Session Policies were returned by the API. "
"STIG V-279691 requires the policy that governs each user to map "
"User's IP to a Network Zone on its Priority 1 active rule."
)
return report