Compare commits

...
Author SHA1 Message Date
Hugo P.Brito c623441882 feat(sdk): add Okta STIG service checks
- Add Network Zone, API Token, and Authenticator checks

- Handle missing Okta scopes with MANUAL findings

- Add service and check tests with parametrized password policy coverage
2026-06-04 16:09:32 +02:00
Daniel Barranquero 91b2829c6f chore: modify stale comment 2026-05-21 12:11:33 +02:00
Daniel Barranquero b5ead15ca4 chore: add decode to return better status extended when no permission 2026-05-21 12:10:09 +02:00
Daniel Barranquero b676a1b251 chore: add markdown to the metadata 2026-05-21 10:29:33 +02:00
Daniel Barranquero 15d3040155 chore: use external file for shared code 2026-05-20 17:55:12 +02:00
Daniel Barranquero 64f04dc5db Merge branch 'master' into PROWLER-1511-okta-stig-signon-service-global-session-policy-checks-sdk 2026-05-20 17:36:31 +02:00
Daniel Barranquero 0abf73b4ae feat: add changelog 2026-05-19 17:25:07 +02:00
Daniel Barranquero 97278ac418 Merge branch 'master' into PROWLER-1511-okta-stig-signon-service-global-session-policy-checks-sdk 2026-05-19 17:23:23 +02:00
Daniel Barranquero b3b395c4f7 chore: update banner check metadata 2026-05-19 17:11:01 +02:00
Daniel Barranquero 7856e7d6cd feat(okta): scan every policy 2026-05-19 13:54:05 +02:00
Daniel Barranquero 1ce6458c6e chore: revision 2026-05-19 13:15:27 +02:00
Daniel Barranquero 427e8e1600 feat(sdk): add signon_dod_warning_banner_configured Okta check
- Enforce DISA STIG V-273192 / OKTA-APP-000200 on Okta brand sign-in pages
- Inspect customized sign-in page HTML for DTM-08-060 banner markers per brand
- Extend Signon service to fetch brands and customized sign-in pages
- Add okta.brands.read to default scopes, CLI args, provider, docs and fixtures
- Return MANUAL when a brand has no customization or the API rejects access
2026-05-19 12:58:24 +02:00
Daniel Barranquero 327a601a67 feat(sdk): add signon_global_session_policy_network_zone_enforced Okta check
- Enforce DISA STIG V-279691 / OKTA-APP-003242 on the Default Global Session Policy
- Require at least one active non-default rule mapping User's IP to a Network Zone
- Allow include or exclude zones to satisfy the Access Control Policy alignment
2026-05-19 12:58:09 +02:00
Daniel Barranquero 6187585919 feat(sdk): add signon_global_session_cookies_not_persistent Okta check
- Enforce DISA STIG V-273206 / OKTA-APP-001710 on the Default Global Session Policy
- Require Priority 1 non-default rule that disables persistent session cookies
- Treat unset use_persistent_cookie as non-compliant
2026-05-19 12:58:02 +02:00
Daniel Barranquero 42b19038e0 feat(sdk): add signon_global_session_lifetime_18h Okta check
- Enforce DISA STIG V-273203 / OKTA-APP-001665 on the Default Global Session Policy
- Require Priority 1 non-default rule with max session lifetime <= 18 hours
- Threshold overridable via okta_max_session_lifetime_minutes audit config key
2026-05-19 12:57:55 +02:00
Daniel Barranquero c73231ec95 refactor(sdk): extract shared signon test fixtures
- Move policy/rule helpers into signon_fixtures.py for reuse across checks
- Add SignInPage helper and DOD banner HTML snippet for upcoming banner check
- Switch idle-timeout test to import from the shared module
2026-05-19 12:57:47 +02:00
107 changed files with 5853 additions and 362 deletions
@@ -30,15 +30,17 @@ If a different authentication method is needed (SSWS API token, OAuth with user
### Required OAuth Scopes
For the initial check (`signon_global_session_idle_timeout_15min`) only one scope is required:
The bundled signon checks require the following read-only scopes:
- `okta.policies.read`
- `okta.brands.read`
Additional scopes will be needed as more services and checks are added, this are the current ones needed:
Additional scopes will be needed as more services and checks are added. These are the current ones needed:
| Scope | Used by |
|---|---|
| `okta.policies.read` | Sign-on / password / authentication policies |
| `okta.brands.read` | Sign-in page customizations (DOD Notice and Consent Banner check) |
### Required Admin Role
@@ -96,7 +98,7 @@ Okta displays the private key **only once**. If you close the modal without copy
### 5. Grant the required OAuth scopes
On the app, open the **Okta API Scopes** tab and click **Grant** on every scope Prowler needs. For the initial release, granting only `okta.policies.read` is sufficient.
On the app, open the **Okta API Scopes** tab and click **Grant** on every scope Prowler needs. The bundled signon checks require `okta.policies.read` and `okta.brands.read`.
![Okta — grant OAuth scopes](/user-guide/providers/okta/images/grant-permissions.png)
@@ -130,8 +132,8 @@ export OKTA_PRIVATE_KEY_FILE="/secure/path/to/prowler-okta.pem"
# or
export OKTA_PRIVATE_KEY="$(cat /secure/path/to/prowler-okta.pem)"
# Optional — defaults to "okta.policies.read"
export OKTA_SCOPES="okta.policies.read"
# Optional — defaults to "okta.policies.read,okta.brands.read"
export OKTA_SCOPES="okta.policies.read,okta.brands.read"
uv run python prowler-cli.py okta
```
@@ -172,7 +174,7 @@ Prowler validates credentials at startup by listing one sign-on policy. This err
Raised when the credential probe succeeds at the OAuth layer but the request is rejected because the service app lacks the required scope or admin role:
- **`invalid_scope`** — the `okta.policies.read` scope is not granted on the service app. Grant it from **Okta API Scopes**.
- **`invalid_scope`** — one of the requested scopes (`okta.policies.read` or `okta.brands.read`) is not granted on the service app. Grant the missing scope from **Okta API Scopes**.
- **`Forbidden` / `not authorized`** — the **Read-Only Administrator** role is not assigned to the service app. Assign it from **Admin roles**.
### `invalid_dpop_proof`
@@ -12,7 +12,7 @@ Set up authentication for Okta with the [Okta Authentication](/user-guide/provid
- An Okta organization. The UI examples below use **Identity Engine** terminology such as **Global Session Policy**; Classic Engine exposes the equivalent sign-on policy concepts under older names.
- A **Super Administrator** account on that organization for the one-time service-app setup.
- An **API Services** app integration in the Okta Admin Console with the `okta.policies.read` scope granted and the **Read-Only Administrator** role assigned.
- An **API Services** app integration in the Okta Admin Console with the `okta.policies.read` and `okta.brands.read` scopes granted and the **Read-Only Administrator** role assigned.
- Python 3.10+ and Prowler 5.27.0 or later installed locally.
<CardGroup cols={2}>
@@ -44,8 +44,8 @@ Follow the [Okta Authentication](/user-guide/providers/okta/authentication) guid
export OKTA_ORG_DOMAIN="acme.okta.com"
export OKTA_CLIENT_ID="0oa1234567890abcdef"
export OKTA_PRIVATE_KEY_FILE="/secure/path/to/prowler-okta.pem"
# Optional — defaults to "okta.policies.read"
export OKTA_SCOPES="okta.policies.read"
# Optional — defaults to "okta.policies.read,okta.brands.read"
export OKTA_SCOPES="okta.policies.read,okta.brands.read"
```
The private key file may contain either a PEM-encoded RSA key or a JWK JSON document.
@@ -113,20 +113,21 @@ This is stricter than simply finding the same timeout value somewhere else in th
### Default Scopes
Prowler requests a fixed set of OAuth scopes on every token exchange. The default is a single scope that covers the bundled initial check:
Prowler requests a fixed set of OAuth scopes on every token exchange. The defaults cover the bundled signon checks:
- `okta.policies.read`
- `okta.brands.read`
The service app must have that scope granted in the **Okta API Scopes** tab. When the granted set is narrower than the requested set, the token request fails with an `invalid_scope` error and the scan stops at provider initialization.
The service app must have these scopes granted in the **Okta API Scopes** tab. When the granted set is narrower than the requested set, the token request fails with an `invalid_scope` error and the scan stops at provider initialization.
When additional checks are enabled — or when running against a service app that exposes a different scope set — override the default with `OKTA_SCOPES` (comma-separated string for the env var) or `--okta-scopes` (space-separated list for the CLI):
```bash
# Environment variable — comma-separated
export OKTA_SCOPES="okta.policies.read,okta.apps.read,okta.users.read"
export OKTA_SCOPES="okta.policies.read,okta.brands.read,okta.apps.read,okta.users.read"
# CLI flag — space-separated
prowler okta --okta-scopes okta.policies.read okta.apps.read okta.users.read
prowler okta --okta-scopes okta.policies.read okta.brands.read okta.apps.read okta.users.read
```
For the full catalog of OAuth scopes exposed by the Okta Management API, refer to the [Okta OAuth 2.0 scopes documentation](https://developer.okta.com/docs/api/oauth2/).
+1
View File
@@ -11,6 +11,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Google Workspace Groups service with 3 new checks [(#11186)](https://github.com/prowler-cloud/prowler/pull/11186)
- `ses_identity_dkim_enabled` check for AWS provider [(#10923)](https://github.com/prowler-cloud/prowler/pull/10923)
- `sagemaker_models_registry_in_use` check for AWS provider, verifying that at least one SageMaker Model Package Group has an approved model package to enforce ML governance workflows [(#11196)](https://github.com/prowler-cloud/prowler/pull/11196)
- `signon_dod_warning_banner_configured`, `signon_global_session_lifetime_18h`, `signon_global_session_cookies_not_persistent` and `signon_global_session_policy_network_zone_enforced` checks for Okta provider [(#11224)](https://github.com/prowler-cloud/prowler/pull/11224)
### 🔄 Changed
@@ -35,8 +35,8 @@ def init_parser(self):
nargs="+",
help=(
"OAuth scopes to request, space-separated "
"(e.g. okta.policies.read okta.users.read). Defaults to the "
"read scopes required by the bundled checks."
"(e.g. okta.policies.read okta.brands.read okta.users.read). "
"Defaults to the read scopes required by the bundled checks."
),
default=None,
metavar="OKTA_SCOPES",
@@ -0,0 +1,38 @@
from pydantic import BaseModel
from prowler.lib.check.models import CheckReportOkta
class MissingScopeResource(BaseModel):
"""Synthetic resource used when a check cannot evaluate an Okta API."""
id: str
name: str
def missing_scope_finding(
*,
metadata,
org_domain: str,
resource_id: str,
resource_name: str,
missing_scopes: list[str],
action: str,
) -> CheckReportOkta:
"""Build a MANUAL finding for checks blocked by missing OAuth scopes."""
resource = MissingScopeResource(id=resource_id, name=resource_name)
report = CheckReportOkta(
metadata=metadata,
resource=resource,
org_domain=org_domain,
resource_id=resource.id,
resource_name=resource.name,
)
report.status = "MANUAL"
report.status_extended = (
f"Prowler could not {action} because the Okta service app is missing "
f"required OAuth scope(s): {', '.join(missing_scopes)}. Grant the "
"scope(s) to the service app and rerun the check, or review the "
"configuration manually in the Okta Admin Console."
)
return report
@@ -32,3 +32,8 @@ class OktaService:
def _run(coro):
"""Run an okta-sdk-python coroutine from synchronous code."""
return asyncio.run(coro)
def _missing_scopes(self, required_scopes: list[str]) -> list[str]:
"""Return required OAuth scopes not granted to the Okta service app."""
granted_scopes = set(getattr(self.provider.session, "scopes", []) or [])
return [scope for scope in required_scopes if scope not in granted_scopes]
+6
View File
@@ -32,6 +32,12 @@ class OktaSession(BaseModel):
class OktaIdentityInfo(BaseModel):
org_domain: str
client_id: str
# Scopes actually granted in the access token (`scp` claim). Used by
# services to distinguish "no data" from "no permission" so checks can
# surface the missing scope rather than a misleading FAIL. Empty when
# decoding the token was not possible — callers must treat empty as
# "unknown" and fall back to attempting the API call.
granted_scopes: list[str] = []
class OktaOutputOptions(ProviderOutputOptions):
+70 -3
View File
@@ -1,4 +1,6 @@
import asyncio
import base64
import json
import os
import re
from os import environ
@@ -30,7 +32,14 @@ from prowler.providers.okta.exceptions.exceptions import (
from prowler.providers.okta.lib.mutelist.mutelist import OktaMutelist
from prowler.providers.okta.models import OktaIdentityInfo, OktaSession
DEFAULT_SCOPES = ["okta.policies.read"]
DEFAULT_SCOPES = [
"okta.policies.read",
"okta.brands.read",
"okta.networkZones.read",
"okta.apiTokens.read",
"okta.roles.read",
"okta.authenticators.read",
]
# Accept only Okta-managed domains. Custom (vanity) domains are rejected on
# purpose — they're a recurring source of typos and silent misconfig and
# Prowler's audience overwhelmingly uses Okta-managed hosts. The TLDs below
@@ -285,14 +294,32 @@ class OktaProvider(Provider):
org URL plus the service-app client ID. We still hit the cheapest
scope-covered endpoint (`list_policies` with limit=1) to fail loud
when credentials, scopes, or the granted admin role are wrong.
After the probe succeeds, the access token's `scp` claim is
decoded and exposed on the identity. Services compare it against
their required scope so checks can emit "missing scope X" rather
than a misleading "no resources returned" finding.
"""
async def _probe():
client = OktaSDKClient(session.to_sdk_config())
return await client.list_policies(type="OKTA_SIGN_ON", limit="1")
result = await client.list_policies(type="OKTA_SIGN_ON", limit="1")
access_token = None
# The OAuth helper caches the token on `_access_token` after
# the first authenticated call. Reach through `_request_executor`
# — a documented internal but a moving target across SDK
# versions, so any failure here degrades silently to empty
# granted_scopes (services then fall back to attempting calls).
try:
oauth = getattr(client._request_executor, "_oauth", None)
if oauth is not None:
access_token = getattr(oauth, "_access_token", None)
except Exception:
access_token = None
return result, access_token
try:
result = asyncio.run(_probe())
result, access_token = asyncio.run(_probe())
# SDK returns (items, resp, err) on the normal path and (items, err)
# only on early request-creation errors. The error is always last.
err = result[-1]
@@ -305,6 +332,11 @@ class OktaProvider(Provider):
"forbidden",
"not authorized",
"permission",
# Okta emits HTTP 400 `consent_required` when none of the
# requested scopes are consented on the service app —
# semantically a permission gap, not a credential one.
"consent_required",
"not allowed",
)
if any(signal in err_text for signal in permission_signals):
raise OktaInsufficientPermissionsError(
@@ -321,6 +353,7 @@ class OktaProvider(Provider):
return OktaIdentityInfo(
org_domain=session.org_domain,
client_id=session.client_id,
granted_scopes=OktaProvider._decode_token_scopes(access_token),
)
except (OktaInvalidCredentialsError, OktaInsufficientPermissionsError):
raise
@@ -330,6 +363,40 @@ class OktaProvider(Provider):
)
raise OktaSetUpIdentityError(original_exception=error)
@staticmethod
def _decode_token_scopes(access_token: Optional[str]) -> list[str]:
"""Return the `scp` claim from a JWT access token, or `[]` on failure.
No signature verification: the token came from Okta over TLS via
the SDK's OAuth handshake, so the only thing we extract is the
scope claim. Any decode error returns an empty list — callers
must treat empty as "unknown" rather than "no scopes granted".
"""
if not access_token:
return []
try:
parts = access_token.split(".")
if len(parts) < 2:
return []
payload_b64 = parts[1]
# Base64url pad to a multiple of 4 — JWT segments are
# unpadded per RFC 7515.
padding = "=" * (-len(payload_b64) % 4)
payload_bytes = base64.urlsafe_b64decode(payload_b64 + padding)
payload = json.loads(payload_bytes)
scp = payload.get("scp")
if isinstance(scp, list):
return [str(s) for s in scp if s]
if isinstance(scp, str):
return [s for s in scp.split(" ") if s]
return []
except Exception as error:
logger.warning(
f"Could not decode Okta access token scopes: "
f"{error.__class__.__name__}: {error}"
)
return []
def print_credentials(self):
report_lines = [
f"Okta Domain: {Fore.YELLOW}{self.identity.org_domain}{Style.RESET_ALL}",
@@ -0,0 +1,4 @@
from prowler.providers.common.provider import Provider
from prowler.providers.okta.services.apitoken.api_token_service import ApiToken
api_token_client = ApiToken(Provider.get_global_provider())
@@ -0,0 +1,201 @@
from typing import Optional
from urllib.parse import parse_qs, urlparse
from pydantic import BaseModel, Field
from prowler.lib.logger import logger
from prowler.providers.okta.lib.service.service import OktaService
API_TOKENS_READ_SCOPE = "okta.apiTokens.read"
NETWORK_ZONES_READ_SCOPE = "okta.networkZones.read"
ROLES_READ_SCOPE = "okta.roles.read"
def _next_after_cursor(resp) -> Optional[str]:
"""Extract the Okta pagination cursor from a Link header."""
if resp is None:
return None
headers = getattr(resp, "headers", None) or {}
link = headers.get("link") or headers.get("Link") or ""
if not link:
return None
for part in link.split(","):
if 'rel="next"' not in part:
continue
url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
if cursor:
return cursor
return None
def _normalise_sdk_result(result) -> tuple[list, object, object]:
"""Return `(items, response, error)` for Okta SDK list call variants."""
if isinstance(result, tuple):
err = result[-1]
items = result[0] or []
resp = result[1] if len(result) >= 3 else None
return list(items), resp, err
return list(result or []), None, None
def _value(value) -> str:
"""Return plain string values from Okta SDK enums and raw strings."""
if value is None:
return ""
enum_value = getattr(value, "value", None)
if enum_value is not None:
return str(enum_value)
return str(value)
class ApiToken(OktaService):
"""Fetches Okta API token metadata, token owners' roles, and zones."""
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
self.missing_scopes: list[str] = self._missing_scopes(
[API_TOKENS_READ_SCOPE, NETWORK_ZONES_READ_SCOPE, ROLES_READ_SCOPE]
)
self.known_network_zone_ids: set[str] = self._list_known_network_zone_ids()
self.api_tokens: dict[str, OktaApiToken] = self._list_api_tokens()
def _list_api_tokens(self) -> dict[str, "OktaApiToken"]:
"""List active API token metadata and owner roles."""
if API_TOKENS_READ_SCOPE in self.missing_scopes:
logger.warning(
"ApiToken - Skipping API Tokens API call because required "
f"scope is missing: {API_TOKENS_READ_SCOPE}"
)
return {}
logger.info("ApiToken - Listing Okta API tokens...")
try:
return self._run(self._fetch_api_tokens())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return {}
async def _fetch_api_tokens(self) -> dict[str, "OktaApiToken"]:
result: dict[str, OktaApiToken] = {}
items, _resp, err = _normalise_sdk_result(await self.client.list_api_tokens())
if err is not None:
logger.error(f"Error listing API tokens: {err}")
return result
for token in items:
token_id = _value(getattr(token, "id", None))
user_id = _value(getattr(token, "user_id", None))
roles = await self._fetch_user_role_types(user_id) if user_id else []
network = getattr(token, "network", None)
token_obj = OktaApiToken(
id=token_id,
name=_value(getattr(token, "name", None)) or token_id,
client_name=_value(getattr(token, "client_name", None)),
user_id=user_id,
network_connection=_value(getattr(network, "connection", None)),
network_includes=list(getattr(network, "include", None) or []),
network_excludes=list(getattr(network, "exclude", None) or []),
owner_roles=roles,
)
result[token_obj.id] = token_obj
return result
async def _fetch_user_role_types(self, user_id: str) -> list[str]:
"""Return normalized admin role types assigned to the token owner."""
if ROLES_READ_SCOPE in self.missing_scopes:
logger.warning(
"ApiToken - Skipping assigned role lookup for token owner "
f"{user_id} because required scope is missing: {ROLES_READ_SCOPE}"
)
return []
items, _resp, err = _normalise_sdk_result(
await self.client.list_assigned_roles_for_user(user_id)
)
if err is not None:
logger.error(f"Error listing roles for token owner {user_id}: {err}")
return []
roles = []
for role in items:
role_type = _value(getattr(role, "type", None))
role_label = _value(getattr(role, "label", None))
roles.append(role_type or role_label)
return [role for role in roles if role]
def _list_known_network_zone_ids(self) -> set[str]:
"""List known Network Zone ids and names for token condition validation."""
if API_TOKENS_READ_SCOPE in self.missing_scopes:
logger.warning(
"ApiToken - Skipping Network Zones API call because API token "
f"listing is unavailable without {API_TOKENS_READ_SCOPE}."
)
return set()
if NETWORK_ZONES_READ_SCOPE in self.missing_scopes:
logger.warning(
"ApiToken - Skipping Network Zones API call because required "
f"scope is missing: {NETWORK_ZONES_READ_SCOPE}"
)
return set()
logger.info("ApiToken - Listing Network Zones for token restrictions...")
try:
return self._run(self._fetch_known_network_zone_ids())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return set()
async def _fetch_known_network_zone_ids(self) -> set[str]:
identifiers: set[str] = set()
items, err = await self._fetch_all_network_zones()
if err is not None:
logger.error(f"Error listing Network Zones for API token checks: {err}")
return identifiers
for zone in items:
zone_id = _value(getattr(zone, "id", None))
zone_name = _value(getattr(zone, "name", None))
if zone_id:
identifiers.add(zone_id)
if zone_name:
identifiers.add(zone_name)
return identifiers
async def _fetch_all_network_zones(self) -> tuple[list, object]:
"""Drain all Network Zone pages for API token reference validation."""
all_items = []
result = await self.client.list_network_zones(after=None, limit=200)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return [], err
all_items.extend(items)
while True:
cursor = _next_after_cursor(resp)
if not cursor:
break
result = await self.client.list_network_zones(after=cursor, limit=200)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return all_items, err
all_items.extend(items)
return all_items, None
class OktaApiToken(BaseModel):
"""Normalized Okta API token metadata used by checks."""
id: str
name: str
client_name: str = ""
user_id: str = ""
network_connection: str = ""
network_includes: list[str] = Field(default_factory=list)
network_excludes: list[str] = Field(default_factory=list)
owner_roles: list[str] = Field(default_factory=list)
class ApiTokenSummary(BaseModel):
"""Synthetic resource for org-level API token findings."""
id: str = "okta-api-tokens"
name: str = "Okta API Tokens"
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "apitoken_not_super_admin",
"CheckTitle": "Okta API tokens are not owned by Super Admin users",
"CheckType": [],
"ServiceName": "apitoken",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether visible Okta API tokens are owned by users without the Super Admin role, because API tokens inherit the administrative permissions of their owner.",
"Risk": "When an API token is owned by a Super Admin, exposure of that token may grant broad organization administration capability to an attacker.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/guides/api-security/",
"https://developer.okta.com/docs/reference/api/roles/"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Create a dedicated service account, assign only required admin roles, rotate the API token, and revoke Super Admin-owned tokens.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review API token owners, move tokens to dedicated service accounts with least-privilege admin roles, rotate affected tokens, and revoke Super Admin-owned tokens.",
"Url": "https://hub.prowler.com/check/apitoken_not_super_admin"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,60 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.apitoken.api_token_client import api_token_client
from prowler.providers.okta.services.apitoken.api_token_service import (
API_TOKENS_READ_SCOPE,
ROLES_READ_SCOPE,
)
from prowler.providers.okta.services.apitoken.lib.api_token_helpers import (
owner_has_super_admin,
)
class apitoken_not_super_admin(Check):
"""Ensure Okta API tokens are not owned by Super Admin users."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate every active API token owner's assigned admin roles."""
org_domain = api_token_client.provider.identity.org_domain
missing_scopes = [
scope
for scope in (API_TOKENS_READ_SCOPE, ROLES_READ_SCOPE)
if scope in api_token_client.missing_scopes
]
if missing_scopes:
return [
missing_scope_finding(
metadata=self.metadata(),
org_domain=org_domain,
resource_id="okta-api-tokens",
resource_name="Okta API Tokens",
missing_scopes=missing_scopes,
action="evaluate API token owner admin roles",
)
]
findings: list[CheckReportOkta] = []
for token in api_token_client.api_tokens.values():
report = CheckReportOkta(
metadata=self.metadata(), resource=token, org_domain=org_domain
)
if owner_has_super_admin(token):
report.status = "FAIL"
report.status_extended = (
f"API token '{token.name}' is owned by user '{token.user_id}' "
"with the Super Admin role. Use a dedicated service account "
"with least-privilege admin roles instead."
)
else:
roles = (
", ".join(token.owner_roles)
if token.owner_roles
else "no admin roles returned"
)
report.status = "PASS"
report.status_extended = (
f"API token '{token.name}' owner '{token.user_id}' is not "
f"assigned Super Admin ({roles})."
)
findings.append(report)
return findings
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "apitoken_restricted_to_network_zone",
"CheckTitle": "Okta API tokens are restricted to known Network Zones",
"CheckType": [],
"ServiceName": "apitoken",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether visible Okta API tokens are constrained to known Network Zones rather than being usable from Any IP.",
"Risk": "When an API token can be used from Any IP or references unknown Network Zones, an exposed token may be replayed from attacker-controlled infrastructure.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/reference/api/api-tokens/",
"https://help.okta.com/oie/en-us/content/topics/security/api.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > API > Tokens: edit each token Security section and select specific Network Zones instead of Any IP.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review each Okta API token and restrict it to one or more known IP-based Network Zones instead of Any IP.",
"Url": "https://hub.prowler.com/check/apitoken_restricted_to_network_zone"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,45 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.apitoken.api_token_client import api_token_client
from prowler.providers.okta.services.apitoken.api_token_service import (
API_TOKENS_READ_SCOPE,
NETWORK_ZONES_READ_SCOPE,
)
from prowler.providers.okta.services.apitoken.lib.api_token_helpers import (
network_zone_restriction_status,
)
class apitoken_restricted_to_network_zone(Check):
"""Ensure Okta API tokens are restricted to known Network Zones."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate every active API token's network condition."""
org_domain = api_token_client.provider.identity.org_domain
missing_scopes = [
scope
for scope in (API_TOKENS_READ_SCOPE, NETWORK_ZONES_READ_SCOPE)
if scope in api_token_client.missing_scopes
]
if missing_scopes:
return [
missing_scope_finding(
metadata=self.metadata(),
org_domain=org_domain,
resource_id="okta-api-tokens",
resource_name="Okta API Tokens",
missing_scopes=missing_scopes,
action="evaluate API token Network Zone restrictions",
)
]
findings: list[CheckReportOkta] = []
for token in api_token_client.api_tokens.values():
report = CheckReportOkta(
metadata=self.metadata(), resource=token, org_domain=org_domain
)
report.status, report.status_extended = network_zone_restriction_status(
token, api_token_client.known_network_zone_ids
)
findings.append(report)
return findings
@@ -0,0 +1,49 @@
from prowler.providers.okta.services.apitoken.api_token_service import OktaApiToken
ANYWHERE_CONNECTIONS = {"", "ANYWHERE", "ANY_IP"}
def network_zone_restriction_status(
token: OktaApiToken, known_network_zone_ids: set[str]
) -> tuple[str, str]:
"""Evaluate whether an API token is restricted to known Network Zones."""
connection = token.network_connection.upper()
if connection in ANYWHERE_CONNECTIONS:
return (
"FAIL",
f"API token '{token.name}' can be used from any IP address. "
"Restrict the token to one or more known Okta Network Zones.",
)
referenced_zones = token.network_includes + token.network_excludes
if not referenced_zones:
return (
"FAIL",
f"API token '{token.name}' is not open to Any IP, but it does not "
"reference a specific Okta Network Zone.",
)
unknown_zones = [
zone for zone in referenced_zones if zone not in known_network_zone_ids
]
if unknown_zones:
return (
"FAIL",
f"API token '{token.name}' references unknown Network Zone(s): "
f"{', '.join(unknown_zones)}.",
)
return (
"PASS",
f"API token '{token.name}' is restricted to known Okta Network Zone(s): "
f"{', '.join(referenced_zones)}.",
)
def owner_has_super_admin(token: OktaApiToken) -> bool:
"""Return True when any token owner role is Super Admin."""
for role in token.owner_roles:
normalized = role.strip().replace(" ", "_").upper()
if normalized in {"SUPER_ADMIN", "SUPER_ADMINISTRATOR"}:
return True
return False
@@ -0,0 +1,6 @@
from prowler.providers.common.provider import Provider
from prowler.providers.okta.services.authenticator.authenticator_service import (
Authenticator,
)
authenticator_client = Authenticator(Provider.get_global_provider())
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_okta_verify_fips_compliant",
"CheckTitle": "Okta Verify requires FIPS-compliant devices",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether the Okta Verify authenticator requires FIPS-compliant devices for enrollment where FIPS-compliant authenticator usage is required.",
"Risk": "When Okta Verify enrollment allows non-FIPS devices, users may authenticate with devices that do not meet the expected cryptographic assurance level.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/en-us/content/topics/mobile/ov-admin-config.htm",
"https://support.okta.com/help/s/article/How-to-Enable-FIPS-Encryption-on-Okta-Verify"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authenticators > Okta Verify: set FIPS Compliance to users enrolling in Okta Verify can use FIPS compliant devices only.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review the Okta Verify authenticator settings and enable FIPS compliance so enrollment is limited to FIPS-compliant devices.",
"Url": "https://hub.prowler.com/check/authenticator_okta_verify_fips_compliant"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,60 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.authenticator_service import (
AUTHENTICATORS_READ_SCOPE,
)
from prowler.providers.okta.services.authenticator.lib.authenticator_helpers import (
find_authenticator_by_key,
missing_authenticator_resource,
)
class authenticator_okta_verify_fips_compliant(Check):
"""Ensure Okta Verify restricts enrollment to FIPS-compliant devices."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate Okta Verify FIPS compliance settings."""
org_domain = authenticator_client.provider.identity.org_domain
if AUTHENTICATORS_READ_SCOPE in authenticator_client.missing_scopes:
return [
missing_scope_finding(
metadata=self.metadata(),
org_domain=org_domain,
resource_id="okta-authenticators",
resource_name="Okta Authenticators",
missing_scopes=[AUTHENTICATORS_READ_SCOPE],
action="evaluate Okta Verify FIPS compliance settings",
)
]
authenticator = find_authenticator_by_key(
authenticator_client.authenticators, "okta_verify"
)
resource = authenticator or missing_authenticator_resource(
"okta_verify", "Okta Verify authenticator"
)
report = CheckReportOkta(
metadata=self.metadata(), resource=resource, org_domain=org_domain
)
if not authenticator or authenticator.status.upper() != "ACTIVE":
report.status = "FAIL"
report.status_extended = (
"Okta Verify authenticator is not active or missing."
)
elif authenticator.fips.upper() == "REQUIRED":
report.status = "PASS"
report.status_extended = (
"Okta Verify authenticator requires FIPS-compliant devices "
"for enrollment."
)
else:
report.status = "FAIL"
report.status_extended = (
"Okta Verify authenticator is active but does not require "
f"FIPS-compliant devices for enrollment (current value: "
f"{authenticator.fips or 'unset'})."
)
return [report]
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_common_password_check",
"CheckTitle": "Okta password policies check passwords against common-password dictionaries",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces common-password dictionary checks.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce common-password dictionary checks.",
"Url": "https://hub.prowler.com/check/authenticator_password_common_password_check"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_common_password_check(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="common_password_check",
requirement="common-password dictionary checks",
compliant=lambda value: value is True,
actual_label="common password check enabled",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_complexity_lowercase",
"CheckTitle": "Okta password policies require lowercase characters",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces at least one lowercase character.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce at least one lowercase character.",
"Url": "https://hub.prowler.com/check/authenticator_password_complexity_lowercase"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_complexity_lowercase(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_lower_case",
requirement="at least one lowercase character",
compliant=lambda value: value is not None and value >= 1,
actual_label="minimum lowercase characters",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_complexity_number",
"CheckTitle": "Okta password policies require numeric characters",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces at least one numeric character.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce at least one numeric character.",
"Url": "https://hub.prowler.com/check/authenticator_password_complexity_number"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_complexity_number(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_number",
requirement="at least one numeric character",
compliant=lambda value: value is not None and value >= 1,
actual_label="minimum numeric characters",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_complexity_symbol",
"CheckTitle": "Okta password policies require symbol characters",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces at least one symbol character.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce at least one symbol character.",
"Url": "https://hub.prowler.com/check/authenticator_password_complexity_symbol"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_complexity_symbol(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_symbol",
requirement="at least one symbol character",
compliant=lambda value: value is not None and value >= 1,
actual_label="minimum symbol characters",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_complexity_uppercase",
"CheckTitle": "Okta password policies require uppercase characters",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces at least one uppercase character.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce at least one uppercase character.",
"Url": "https://hub.prowler.com/check/authenticator_password_complexity_uppercase"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_complexity_uppercase(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_upper_case",
requirement="at least one uppercase character",
compliant=lambda value: value is not None and value >= 1,
actual_label="minimum uppercase characters",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_history_5",
"CheckTitle": "Okta password policies remember at least 5 previous passwords",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces password history of at least 5 previous passwords.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce password history of at least 5 previous passwords.",
"Url": "https://hub.prowler.com/check/authenticator_password_history_5"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_history_5(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="history_count",
requirement="password history of at least 5 previous passwords",
compliant=lambda value: value is not None and value >= 5,
actual_label="password history count",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_lockout_threshold_3",
"CheckTitle": "Okta password policies lock accounts after 3 or fewer failed attempts",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces password lockout after 3 or fewer failed attempts.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce password lockout after 3 or fewer failed attempts.",
"Url": "https://hub.prowler.com/check/authenticator_password_lockout_threshold_3"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_lockout_threshold_3(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="max_attempts",
requirement="password lockout after 3 or fewer failed attempts",
compliant=lambda value: value is not None and value <= 3,
actual_label="maximum failed attempts",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_maximum_age_60d",
"CheckTitle": "Okta password policies enforce a maximum password age of 60 days",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces maximum password age of 60 days or less.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce maximum password age of 60 days or less.",
"Url": "https://hub.prowler.com/check/authenticator_password_maximum_age_60d"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_maximum_age_60d(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="max_age_days",
requirement="maximum password age of 60 days or less",
compliant=lambda value: value is not None and 0 < value <= 60,
actual_label="maximum age days",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_minimum_age_24h",
"CheckTitle": "Okta password policies enforce a 24-hour minimum password age",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces minimum password age of at least 24 hours.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce minimum password age of at least 24 hours.",
"Url": "https://hub.prowler.com/check/authenticator_password_minimum_age_24h"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_minimum_age_24h(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_age_minutes",
requirement="minimum password age of at least 24 hours",
compliant=lambda value: value is not None and value >= 1440,
actual_label="minimum age minutes",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "authenticator_password_minimum_length_15",
"CheckTitle": "Okta password policies require at least 15 characters",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether every visible active Okta Password Policy enforces minimum password length of at least 15 characters.",
"Risk": "When active password policies are weak or inconsistent, users governed by less strict policies may face increased credential guessing, password reuse, and account takeover risk.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/policy",
"https://help.okta.com/en-us/content/topics/security/policies/configure-password-policies.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authentication > Password: edit each active password policy and adjust password settings to the STIG requirement.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review every active Okta Password Policy and configure password settings to enforce minimum password length of at least 15 characters.",
"Url": "https://hub.prowler.com/check/authenticator_password_minimum_length_15"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,24 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.lib.password_policy_helpers import (
execute_password_policy_check,
)
class authenticator_password_minimum_length_15(Check):
"""Ensure Okta Password Policies enforce the required STIG setting."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate all active Okta Password Policies."""
return execute_password_policy_check(
metadata=self.metadata(),
org_domain=authenticator_client.provider.identity.org_domain,
password_policies=authenticator_client.password_policies,
missing_scopes=authenticator_client.missing_scopes,
field_name="min_length",
requirement="minimum password length of at least 15 characters",
compliant=lambda value: value is not None and value >= 15,
actual_label="minimum length",
)
@@ -0,0 +1,246 @@
from typing import Optional
from urllib.parse import parse_qs, urlparse
from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.providers.okta.lib.service.service import OktaService
AUTHENTICATORS_READ_SCOPE = "okta.authenticators.read"
POLICIES_READ_SCOPE = "okta.policies.read"
def _next_after_cursor(resp) -> Optional[str]:
"""Extract the Okta pagination cursor from a Link header."""
if resp is None:
return None
headers = getattr(resp, "headers", None) or {}
link = headers.get("link") or headers.get("Link") or ""
if not link:
return None
for part in link.split(","):
if 'rel="next"' not in part:
continue
url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
if cursor:
return cursor
return None
def _normalise_sdk_result(result) -> tuple[list, object, object]:
"""Return `(items, response, error)` for Okta SDK list call variants."""
if isinstance(result, tuple):
err = result[-1]
items = result[0] or []
resp = result[1] if len(result) >= 3 else None
return list(items), resp, err
return list(result or []), None, None
def _value(value) -> str:
"""Return plain string values from Okta SDK enums and raw strings."""
if value is None:
return ""
enum_value = getattr(value, "value", None)
if enum_value is not None:
return str(enum_value)
return str(value)
def _int_or_none(value) -> Optional[int]:
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _bool_or_none(value) -> Optional[bool]:
if value is None:
return None
if isinstance(value, bool):
return value
return bool(value)
class Authenticator(OktaService):
"""Fetches Okta Password Policies and Authenticators for STIG checks."""
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
self.missing_scopes: list[str] = self._missing_scopes(
[POLICIES_READ_SCOPE, AUTHENTICATORS_READ_SCOPE]
)
self.password_policies: dict[str, PasswordPolicy] = (
self._list_password_policies()
)
self.authenticators: dict[str, OktaAuthenticator] = self._list_authenticators()
def _list_password_policies(self) -> dict[str, "PasswordPolicy"]:
"""List PASSWORD policies with normalized password settings."""
if POLICIES_READ_SCOPE in self.missing_scopes:
logger.warning(
"Authenticator - Skipping Policies API call because required "
f"scope is missing: {POLICIES_READ_SCOPE}"
)
return {}
logger.info("Authenticator - Listing Okta PASSWORD policies...")
try:
return self._run(self._fetch_password_policies())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return {}
async def _fetch_password_policies(self) -> dict[str, "PasswordPolicy"]:
result: dict[str, PasswordPolicy] = {}
items, err = await self._paginate(
lambda after: self.client.list_policies(type="PASSWORD", after=after)
)
if err is not None:
logger.error(f"Error listing PASSWORD policies: {err}")
return result
for policy in items:
policy_obj = self._build_password_policy(policy)
result[policy_obj.id] = policy_obj
return result
def _list_authenticators(self) -> dict[str, "OktaAuthenticator"]:
"""List org authenticators with normalized settings."""
if AUTHENTICATORS_READ_SCOPE in self.missing_scopes:
logger.warning(
"Authenticator - Skipping Authenticators API call because required "
f"scope is missing: {AUTHENTICATORS_READ_SCOPE}"
)
return {}
logger.info("Authenticator - Listing Okta authenticators...")
try:
return self._run(self._fetch_authenticators())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return {}
async def _fetch_authenticators(self) -> dict[str, "OktaAuthenticator"]:
result: dict[str, OktaAuthenticator] = {}
items, _resp, err = _normalise_sdk_result(
await self.client.list_authenticators()
)
if err is not None:
logger.error(f"Error listing authenticators: {err}")
return result
for authenticator in items:
auth_obj = self._build_authenticator(authenticator)
result[auth_obj.id] = auth_obj
return result
@staticmethod
async def _paginate(fetch):
"""Drain all pages of an SDK list call using Okta Link headers."""
all_items = []
result = await fetch(None)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return [], err
all_items.extend(items)
while True:
cursor = _next_after_cursor(resp)
if not cursor:
break
result = await fetch(cursor)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return all_items, err
all_items.extend(items)
return all_items, None
@staticmethod
def _build_password_policy(policy) -> "PasswordPolicy":
settings = getattr(policy, "settings", None)
password_settings = getattr(settings, "password", None) if settings else None
lockout = (
getattr(password_settings, "lockout", None) if password_settings else None
)
complexity = (
getattr(password_settings, "complexity", None)
if password_settings
else None
)
dictionary = getattr(complexity, "dictionary", None) if complexity else None
age = getattr(password_settings, "age", None) if password_settings else None
policy_id = _value(getattr(policy, "id", None))
return PasswordPolicy(
id=policy_id,
name=_value(getattr(policy, "name", None)) or policy_id,
status=_value(getattr(policy, "status", None)),
priority=_int_or_none(getattr(policy, "priority", None)),
is_default=bool(getattr(policy, "system", False)),
max_attempts=_int_or_none(getattr(lockout, "max_attempts", None)),
min_length=_int_or_none(getattr(complexity, "min_length", None)),
min_upper_case=_int_or_none(getattr(complexity, "min_upper_case", None)),
min_lower_case=_int_or_none(getattr(complexity, "min_lower_case", None)),
min_number=_int_or_none(getattr(complexity, "min_number", None)),
min_symbol=_int_or_none(getattr(complexity, "min_symbol", None)),
min_age_minutes=_int_or_none(getattr(age, "min_age_minutes", None)),
max_age_days=_int_or_none(getattr(age, "max_age_days", None)),
history_count=_int_or_none(getattr(age, "history_count", None)),
common_password_check=_bool_or_none(getattr(dictionary, "common", None)),
)
@staticmethod
def _build_authenticator(authenticator) -> "OktaAuthenticator":
settings = getattr(authenticator, "settings", None)
compliance = getattr(settings, "compliance", None) if settings else None
auth_id = _value(getattr(authenticator, "id", None))
return OktaAuthenticator(
id=auth_id,
key=_value(getattr(authenticator, "key", None)),
name=_value(getattr(authenticator, "name", None)) or auth_id,
status=_value(getattr(authenticator, "status", None)),
type=_value(getattr(authenticator, "type", None)),
fips=_value(getattr(compliance, "fips", None)),
)
class PasswordPolicy(BaseModel):
"""Normalized Okta Password Policy settings used by checks."""
id: str
name: str
status: str = ""
priority: Optional[int] = None
is_default: bool = False
max_attempts: Optional[int] = None
min_length: Optional[int] = None
min_upper_case: Optional[int] = None
min_lower_case: Optional[int] = None
min_number: Optional[int] = None
min_symbol: Optional[int] = None
min_age_minutes: Optional[int] = None
max_age_days: Optional[int] = None
history_count: Optional[int] = None
common_password_check: Optional[bool] = None
class OktaAuthenticator(BaseModel):
"""Normalized Okta Authenticator settings used by checks."""
id: str
key: str
name: str
status: str = ""
type: str = ""
fips: str = ""
class AuthenticatorSummary(BaseModel):
"""Synthetic resource for org-level authenticator findings."""
id: str
name: str
@@ -0,0 +1,36 @@
{
"Provider": "okta",
"CheckID": "authenticator_smart_card_active",
"CheckTitle": "Okta Smart Card authenticator is active",
"CheckType": [],
"ServiceName": "authenticator",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "IAM",
"Description": "Prowler evaluates whether the Smart Card IdP authenticator is configured and active for certificate-based authentication scenarios required by policy.",
"Risk": "When Smart Card authentication is unavailable, users may be unable to satisfy certificate-based authentication requirements where they are mandated.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/authenticator"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Authenticators: add or activate the Smart Card IdP authenticator.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review Okta Authenticators and configure or activate the Smart Card IdP authenticator where certificate-based authentication is required.",
"Url": "https://hub.prowler.com/check/authenticator_smart_card_active"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,56 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.authenticator.authenticator_client import (
authenticator_client,
)
from prowler.providers.okta.services.authenticator.authenticator_service import (
AUTHENTICATORS_READ_SCOPE,
)
from prowler.providers.okta.services.authenticator.lib.authenticator_helpers import (
find_authenticator_by_key,
missing_authenticator_resource,
)
class authenticator_smart_card_active(Check):
"""Ensure the Smart Card IdP authenticator is active."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate the Smart Card IdP authenticator status."""
org_domain = authenticator_client.provider.identity.org_domain
if AUTHENTICATORS_READ_SCOPE in authenticator_client.missing_scopes:
return [
missing_scope_finding(
metadata=self.metadata(),
org_domain=org_domain,
resource_id="okta-authenticators",
resource_name="Okta Authenticators",
missing_scopes=[AUTHENTICATORS_READ_SCOPE],
action="evaluate Smart Card IdP authenticator status",
)
]
authenticator = find_authenticator_by_key(
authenticator_client.authenticators, "smart_card_idp"
)
resource = authenticator or missing_authenticator_resource(
"smart_card_idp", "Smart Card IdP authenticator"
)
report = CheckReportOkta(
metadata=self.metadata(), resource=resource, org_domain=org_domain
)
if authenticator and authenticator.status.upper() == "ACTIVE":
report.status = "PASS"
report.status_extended = "Smart Card IdP authenticator is ACTIVE."
elif authenticator:
report.status = "FAIL"
report.status_extended = (
f"Smart Card IdP authenticator is not active; current status is "
f"{authenticator.status}."
)
else:
report.status = "FAIL"
report.status_extended = (
"Smart Card IdP authenticator is not active or missing."
)
return [report]
@@ -0,0 +1,19 @@
from prowler.providers.okta.services.authenticator.authenticator_service import (
AuthenticatorSummary,
OktaAuthenticator,
)
def find_authenticator_by_key(
authenticators: dict[str, OktaAuthenticator], key: str
) -> OktaAuthenticator | None:
"""Return the first authenticator with the requested key."""
for authenticator in authenticators.values():
if authenticator.key == key:
return authenticator
return None
def missing_authenticator_resource(key: str, name: str) -> AuthenticatorSummary:
"""Build a synthetic resource for a missing authenticator."""
return AuthenticatorSummary(id=f"{key}-missing", name=name)
@@ -0,0 +1,101 @@
from collections.abc import Callable
from prowler.lib.check.models import CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.authenticator.authenticator_service import (
POLICIES_READ_SCOPE,
PasswordPolicy,
)
def active_password_policies(
password_policies: dict[str, PasswordPolicy],
) -> list[PasswordPolicy]:
"""Return active password policies sorted by priority."""
return sorted(
[
policy
for policy in password_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 password_policy_label(policy: PasswordPolicy) -> str:
kind = "default" if policy.is_default else "custom"
priority = policy.priority if policy.priority is not None else "unset"
return f"Password Policy '{policy.name}' (priority {priority}, {kind})"
def no_active_password_policies_finding(
metadata, org_domain: str, requirement: str
) -> CheckReportOkta:
"""Build the FAIL finding emitted when no active password policies exist."""
placeholder = PasswordPolicy(
id="password-policies-missing",
name="(no active password policies)",
status="MISSING",
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "FAIL"
report.status_extended = (
"No active Okta Password Policies were returned by the API. "
f"The organization must enforce: {requirement}."
)
return report
def execute_password_policy_check(
*,
metadata,
org_domain: str,
password_policies: dict[str, PasswordPolicy],
missing_scopes: list[str],
field_name: str,
requirement: str,
compliant: Callable[[object], bool],
actual_label: str,
) -> list[CheckReportOkta]:
"""Evaluate a scalar password-policy setting across all active policies."""
if POLICIES_READ_SCOPE in missing_scopes:
return [
missing_scope_finding(
metadata=metadata,
org_domain=org_domain,
resource_id="okta-password-policies",
resource_name="Okta Password Policies",
missing_scopes=[POLICIES_READ_SCOPE],
action="evaluate active Password Policy settings",
)
]
policies = active_password_policies(password_policies)
if not policies:
return [no_active_password_policies_finding(metadata, org_domain, requirement)]
findings: list[CheckReportOkta] = []
for policy in policies:
actual = getattr(policy, field_name)
report = CheckReportOkta(
metadata=metadata, resource=policy, org_domain=org_domain
)
if compliant(actual):
report.status = "PASS"
report.status_extended = (
f"{password_policy_label(policy)} enforces {requirement} "
f"({actual_label}: {actual})."
)
else:
report.status = "FAIL"
report.status_extended = (
f"{password_policy_label(policy)} does not enforce {requirement} "
f"({actual_label}: {actual})."
)
findings.append(report)
return findings
@@ -0,0 +1,53 @@
from prowler.providers.okta.services.network.network_zone_service import OktaNetworkZone
ANONYMIZER_CATEGORY_MARKERS = (
"ANONYM",
"PROXY",
"TOR",
"VPN",
)
def active_blocklist_zones(
network_zones: dict[str, OktaNetworkZone],
) -> list[OktaNetworkZone]:
"""Return active Network Zones configured for blocklist usage."""
return sorted(
[
zone
for zone in network_zones.values()
if zone.status.upper() == "ACTIVE" and zone.usage.upper() == "BLOCKLIST"
],
key=lambda zone: (zone.name, zone.id),
)
def is_ip_blocklist_with_entries(zone: OktaNetworkZone) -> bool:
"""Return True when an IP blocklist zone contains gateway/proxy entries."""
return zone.type.upper() == "IP" and bool(zone.gateways or zone.proxies)
def is_enhanced_dynamic_anonymizer_blocklist(zone: OktaNetworkZone) -> bool:
"""Return True for active Enhanced Dynamic blocklists covering anonymizers."""
if zone.type.upper() != "DYNAMIC_V2":
return False
if zone.system and zone.name == "DefaultEnhancedDynamicZone":
return True
categories = [category.upper() for category in zone.ip_service_categories]
return any(
marker in category
for category in categories
for marker in ANONYMIZER_CATEGORY_MARKERS
)
def compliant_anonymized_proxy_blocklist(
network_zones: dict[str, OktaNetworkZone],
) -> tuple[OktaNetworkZone | None, str]:
"""Find the Network Zone that satisfies anonymized-proxy blocklisting."""
for zone in active_blocklist_zones(network_zones):
if is_ip_blocklist_with_entries(zone):
return zone, "active IP blocklist with gateway or proxy IP entries"
if is_enhanced_dynamic_anonymizer_blocklist(zone):
return zone, "active Enhanced Dynamic Zone blocklist for anonymizers"
return None, ""
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "network_zone_block_anonymized_proxies",
"CheckTitle": "Okta blocks anonymized proxy access with active Network Zone blocklists",
"CheckType": [],
"ServiceName": "network",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "network",
"Description": "Prowler evaluates whether Okta has an active Network Zone blocklist that can block anonymized proxy sources, either through gateway/proxy IP entries or an Enhanced Dynamic Zone anonymizer category.",
"Risk": "When anonymized proxy sources are not covered by active Okta Network Zone blocklists, attackers may hide their source network while attempting credential attacks or session establishment from untrusted infrastructure.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://developer.okta.com/docs/api/openapi/okta-management/management/tags/networkzone",
"https://help.okta.com/en-us/content/topics/security/network/about-enhanced-dynamic-zones.htm"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "Security > Networks: configure BlockedIpZone gateway/proxy IP entries or activate DefaultEnhancedDynamicZone / Enhanced Dynamic Zone blocklisting for anonymizers.",
"Terraform": ""
},
"Recommendation": {
"Text": "Review Okta Network Zones and configure an active IP blocklist or Enhanced Dynamic Zone blocklist that covers anonymizers before authentication.",
"Url": "https://hub.prowler.com/check/network_zone_block_anonymized_proxies"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,52 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.lib.service.scope import missing_scope_finding
from prowler.providers.okta.services.network.lib.network_zone_helpers import (
compliant_anonymized_proxy_blocklist,
)
from prowler.providers.okta.services.network.network_zone_client import (
network_zone_client,
)
from prowler.providers.okta.services.network.network_zone_service import (
NetworkZoneSummary,
)
class network_zone_block_anonymized_proxies(Check):
"""Ensure Okta actively blocks anonymized proxy sources before auth."""
def execute(self) -> list[CheckReportOkta]:
"""Evaluate whether an active blocklist covers anonymized proxies."""
org_domain = network_zone_client.provider.identity.org_domain
if network_zone_client.missing_scopes:
return [
missing_scope_finding(
metadata=self.metadata(),
org_domain=org_domain,
resource_id="okta-network-zones",
resource_name="Okta Network Zones",
missing_scopes=network_zone_client.missing_scopes,
action="evaluate Network Zone anonymized proxy blocklists",
)
]
matching_zone, reason = compliant_anonymized_proxy_blocklist(
network_zone_client.network_zones
)
resource = matching_zone or NetworkZoneSummary()
report = CheckReportOkta(
metadata=self.metadata(), resource=resource, org_domain=org_domain
)
if matching_zone:
report.status = "PASS"
report.status_extended = (
f"Okta Network Zone '{matching_zone.name}' is an {reason}."
)
else:
report.status = "FAIL"
report.status_extended = (
"No active Okta Network Zone blocklist was found that blocks "
"anonymized proxies. Existing zones do not actively block gateway "
"or proxy IPs, nor an Enhanced Dynamic Zone anonymizer category."
)
return [report]
@@ -0,0 +1,4 @@
from prowler.providers.common.provider import Provider
from prowler.providers.okta.services.network.network_zone_service import NetworkZone
network_zone_client = NetworkZone(Provider.get_global_provider())
@@ -0,0 +1,152 @@
from typing import Optional
from urllib.parse import parse_qs, urlparse
from pydantic import BaseModel, Field
from prowler.lib.logger import logger
from prowler.providers.okta.lib.service.service import OktaService
NETWORK_ZONES_READ_SCOPE = "okta.networkZones.read"
def _next_after_cursor(resp) -> Optional[str]:
"""Extract the Okta pagination cursor from a Link header."""
if resp is None:
return None
headers = getattr(resp, "headers", None) or {}
link = headers.get("link") or headers.get("Link") or ""
if not link:
return None
for part in link.split(","):
if 'rel="next"' not in part:
continue
url_segment = part.split(";", 1)[0].strip().lstrip("<").rstrip(">")
cursor = parse_qs(urlparse(url_segment).query).get("after", [None])[0]
if cursor:
return cursor
return None
def _normalise_sdk_result(result) -> tuple[list, object, object]:
"""Return `(items, response, error)` for Okta SDK list call variants."""
if isinstance(result, tuple):
err = result[-1]
items = result[0] or []
resp = result[1] if len(result) >= 3 else None
return list(items), resp, err
return list(result or []), None, None
def _value(value) -> str:
"""Return plain string values from Okta SDK enums and raw strings."""
if value is None:
return ""
enum_value = getattr(value, "value", None)
if enum_value is not None:
return str(enum_value)
return str(value)
class NetworkZone(OktaService):
"""Fetches Okta Network Zones for STIG network-zone checks."""
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
self.missing_scopes: list[str] = self._missing_scopes(
[NETWORK_ZONES_READ_SCOPE]
)
self.network_zones: dict[str, OktaNetworkZone] = self._list_network_zones()
def _list_network_zones(self) -> dict[str, "OktaNetworkZone"]:
"""List all Network Zones visible to the configured Okta service app."""
if self.missing_scopes:
logger.warning(
"NetworkZone - Skipping Network Zones API call because required "
f"scope(s) are missing: {', '.join(self.missing_scopes)}"
)
return {}
logger.info("NetworkZone - Listing Okta Network Zones...")
try:
return self._run(self._fetch_all())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return {}
async def _fetch_all(self) -> dict[str, "OktaNetworkZone"]:
result: dict[str, OktaNetworkZone] = {}
all_zones, err = await self._paginate(
lambda after: self.client.list_network_zones(after=after, limit=200)
)
if err is not None:
logger.error(f"Error listing Network Zones: {err}")
return result
for zone in all_zones:
zone_obj = self._build_zone(zone)
result[zone_obj.id] = zone_obj
return result
@staticmethod
async def _paginate(fetch):
"""Drain all pages of an SDK list call using Okta Link headers."""
all_items = []
result = await fetch(None)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return [], err
all_items.extend(items)
while True:
cursor = _next_after_cursor(resp)
if not cursor:
break
result = await fetch(cursor)
items, resp, err = _normalise_sdk_result(result)
if err is not None:
return all_items, err
all_items.extend(items)
return all_items, None
@staticmethod
def _build_zone(zone) -> "OktaNetworkZone":
zone_id = _value(getattr(zone, "id", None))
return OktaNetworkZone(
id=zone_id,
name=_value(getattr(zone, "name", None)) or zone_id,
status=_value(getattr(zone, "status", None)),
type=_value(getattr(zone, "type", None)),
usage=_value(getattr(zone, "usage", None)),
system=bool(getattr(zone, "system", False)),
gateways=list(getattr(zone, "gateways", None) or []),
proxies=list(getattr(zone, "proxies", None) or []),
asns=list(getattr(zone, "asns", None) or []),
locations=list(getattr(zone, "locations", None) or []),
ip_service_categories=[
_value(category)
for category in (getattr(zone, "ip_service_categories", None) or [])
],
)
class OktaNetworkZone(BaseModel):
"""Normalized Okta Network Zone attributes used by checks."""
id: str
name: str
status: str = ""
type: str = ""
usage: str = ""
system: bool = False
gateways: list[str] = Field(default_factory=list)
proxies: list[str] = Field(default_factory=list)
asns: list[str] = Field(default_factory=list)
locations: list[str] = Field(default_factory=list)
ip_service_categories: list[str] = Field(default_factory=list)
class NetworkZoneSummary(BaseModel):
"""Synthetic resource for org-level Network Zone findings."""
id: str = "okta-network-zones"
name: str = "Okta Network Zones"
@@ -0,0 +1,146 @@
"""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,
SignInPage,
)
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
_SCOPE_ADVICE = (
"Grant it on the service app's Okta API Scopes tab in the Okta Admin "
"Console, then re-run the check."
)
def missing_policy_scope_finding(
metadata, org_domain: str, scope: str
) -> CheckReportOkta:
"""Build the MANUAL finding for a sign-on policy check when the API scope is not granted."""
placeholder = GlobalSessionPolicy(
id="signon-policies-scope-missing",
name="(scope not granted)",
priority=1,
status="MISSING",
is_default=False,
rules=[],
)
report = CheckReportOkta(
metadata=metadata, resource=placeholder, org_domain=org_domain
)
report.status = "MANUAL"
report.status_extended = (
f"Could not retrieve Global Session Policies: the Okta service app "
f"is missing the required `{scope}` API scope. {_SCOPE_ADVICE}"
)
return report
def missing_brand_scope_finding(
metadata, org_domain: str, scope: str
) -> CheckReportOkta:
"""Build the MANUAL finding for a brand/sign-in-page check when the API scope is not granted."""
placeholder = SignInPage(
brand_id="signon-brands-scope-missing",
brand_name="(scope not granted)",
is_customized=False,
)
report = CheckReportOkta(
metadata=metadata,
resource=placeholder,
org_domain=org_domain,
resource_name=placeholder.brand_name,
resource_id=placeholder.brand_id,
)
report.status = "MANUAL"
report.status_extended = (
f"Could not retrieve Okta brand sign-in pages: the Okta service app "
f"is missing the required `{scope}` API scope. {_SCOPE_ADVICE}"
)
return report
@@ -0,0 +1,38 @@
{
"Provider": "okta",
"CheckID": "signon_dod_warning_banner_configured",
"CheckTitle": "Okta sign-in page displays the Standard Mandatory DOD Notice and Consent Banner",
"CheckType": [],
"ServiceName": "signon",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "informational",
"ResourceType": "NotDefined",
"ResourceGroup": "governance",
"Description": "Each Okta brand's sign-in page must present the **Standard Mandatory DOD Notice and Consent Banner** (`DTM-08-060`) before granting access.\n\nThe check inspects the sign-in page HTML returned by the Okta Management API, using the *customized* page when present and otherwise falling back to the *default* sign-in page.\n\nAligns with **DISA STIG V-273192 / OKTA-APP-000200**.",
"Risk": "Without the **DOD Notice and Consent Banner**, users are not informed that the system is a U.S. Government information system subject to monitoring.\n\n- **Legal basis** for incident response and prosecution is weakened\n- **Alignment** with federal laws, Executive Orders, directives, and standards is broken\n- **Implied consent** to monitoring cannot be asserted on connection",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/oie/en-us/content/topics/settings/settings-customization.htm",
"https://developer.okta.com/docs/api/openapi/okta-management/management/tag/CustomPages/",
"https://developer.okta.com/docs/api/openapi/okta-management/management/tag/Brands/"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Follow the supplemental *Okta DOD Warning Banner Configuration Guide* shipped with the **DISA Okta IDaaS STIG** package.\n2. Sign in to the **Okta Admin Console** as a *Super Admin*.\n3. Navigate to **Customizations** > **Brands** and select the brand.\n4. Edit the **Sign-in page** customization.\n5. Insert the full `DTM-08-060` Standard Mandatory DOD Notice and Consent Banner text into the page content.\n6. Publish the customization and verify the banner is presented before sign-in.",
"Terraform": ""
},
"Recommendation": {
"Text": "Customize the Okta sign-in page for **each brand** to display the **Standard Mandatory DOD Notice and Consent Banner** (`DTM-08-060`) before users authenticate.\n\nApplies only to Okta tenants under **U.S. Department of Defense** scope; non-DOD organizations can mute this check.",
"Url": "https://hub.prowler.com/check/signon_dod_warning_banner_configured"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Applicable only to Okta tenants under U.S. Department of Defense scope (DISA Okta IDaaS STIG, control V-273192 / OKTA-APP-000200). For non-DOD organizations this check is not applicable and can be muted."
}
@@ -0,0 +1,129 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
missing_brand_scope_finding,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import SignInPage
# Distinctive marker groups drawn from the DTM-08-060 Standard Mandatory
# DOD Notice and Consent Banner. The HTML can vary across brands, so the
# check looks for the banner's core ideas rather than requiring an exact
# string match.
BANNER_MARKER_GROUPS = (
("u.s. government", "us government"),
("information system", "information systems"),
("authorized use only", "authorized use"),
(
"subject to monitoring",
"may be intercepted",
"searched, monitored, and recorded",
"consent to monitoring",
),
)
def _matched_banner_groups(content_lower: str) -> list[str]:
matched_markers: list[str] = []
for marker_group in BANNER_MARKER_GROUPS:
for marker in marker_group:
if marker in content_lower:
matched_markers.append(marker)
break
return matched_markers
class signon_dod_warning_banner_configured(Check):
"""STIG V-273192 / OKTA-APP-000200.
Okta must display the Standard Mandatory DOD Notice and Consent
Banner (DTM-08-060) before granting access to the application. The
check inspects each brand's sign-in page HTML returned by the Okta
Management API, using the customized page when present and otherwise
falling back to the default sign-in page.
"""
def execute(self) -> list[CheckReportOkta]:
org_domain = signon_client.provider.identity.org_domain
findings: list[CheckReportOkta] = []
missing_scope = signon_client.missing_scope.get("sign_in_pages")
if missing_scope:
return [
missing_brand_scope_finding(self.metadata(), org_domain, missing_scope)
]
if not signon_client.sign_in_pages:
placeholder = SignInPage(
brand_id="no-brands-detected",
brand_name="(no brands detected)",
is_customized=False,
)
report = CheckReportOkta(
metadata=self.metadata(),
resource=placeholder,
org_domain=org_domain,
resource_name=placeholder.brand_name,
resource_id=placeholder.brand_id,
)
report.status = "MANUAL"
report.status_extended = (
"No Okta brands were retrieved from the Brands API. Verify "
"the sign-in page for the organization displays the DOD "
"Notice and Consent Banner (DTM-08-060) in the Admin Console."
)
findings.append(report)
return findings
for page in signon_client.sign_in_pages.values():
report = CheckReportOkta(
metadata=self.metadata(),
resource=page,
org_domain=org_domain,
resource_name=page.brand_name or page.brand_id,
resource_id=page.brand_id,
)
if page.fetch_error:
report.status = "MANUAL"
report.status_extended = (
f"Could not retrieve the sign-in page for "
f"brand '{page.brand_name or page.brand_id}' ({page.fetch_error}). "
"Inspect the brand manually to confirm the "
"DOD Notice and Consent Banner (DTM-08-060) is displayed."
)
findings.append(report)
continue
if not page.page_content:
report.status = "MANUAL"
report.status_extended = (
f"Sign-in page content for brand "
f"'{page.brand_name or page.brand_id}' could not be "
"retrieved from the Okta API. Verify the DOD Notice and "
"Consent Banner (DTM-08-060) manually in the Admin Console."
)
findings.append(report)
continue
page_type = "customized" if page.is_customized else "default"
content_lower = page.page_content.lower()
matches = _matched_banner_groups(content_lower)
if len(matches) == len(BANNER_MARKER_GROUPS):
report.status = "PASS"
report.status_extended = (
f"DOD Notice and Consent Banner detected on the {page_type} "
f"sign-in page for brand '{page.brand_name or page.brand_id}' "
f"({len(matches)} of {len(BANNER_MARKER_GROUPS)} required "
"marker groups matched)."
)
else:
report.status = "FAIL"
report.status_extended = (
f"{page_type.title()} sign-in page for brand "
f"'{page.brand_name or page.brand_id}' does not contain "
"the DOD Notice and Consent Banner (DTM-08-060)."
)
findings.append(report)
return findings
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "signon_global_session_cookies_not_persistent",
"CheckTitle": "Default Global Session Policy has a Priority 1 non-default rule disabling persistent global session cookies",
"CheckType": [],
"ServiceName": "signon",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "governance",
"Description": "Every active Okta **Global Session Policy** needs a **Priority 1** rule that is **not** the built-in `Default Rule`, setting *Okta global session cookies persist across browser sessions* to `Disabled`.\n\nOkta evaluates policies by group assignment, so a permissive custom policy can govern users. Aligns with **DISA STIG V-273206 / OKTA-APP-001710**.",
"Risk": "Persistent global session cookies keep an authenticated Okta session alive across browser restarts.\n\n- **Surviving sessions** outlive the browsing context the user expected\n- **Cached authorization decisions** remain in effect after the browser closes\n- **Forgotten or shared devices** continue to hold authenticated access until cookies expire on their own",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/oie/en-us/content/topics/identity-engine/policies/about-okta-sign-on-policies.htm",
"https://developer.okta.com/docs/api/openapi/okta-management/management/tag/Policy/"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Security** > **Global Session Policy**.\n3. Open the **Default Policy** (and repeat for every other active policy).\n4. Add or edit a non-default rule.\n5. Move that rule to **Priority 1** so it is evaluated before the built-in `Default Rule`.\n6. Set *Okta global session cookies persist across browser sessions* to `Disabled`.\n7. Save the rule.",
"Terraform": "```hcl\nresource \"okta_policy_rule_signon\" \"<example_resource_name>\" {\n policy_id = okta_policy_signon.default.id\n name = \"<example_resource_name>\"\n status = \"ACTIVE\"\n priority = 1 # Critical: rule must sit at Priority 1 before the Default Rule\n session_persistent = false # Critical: disable persistent global session cookies\n}\n```"
},
"Recommendation": {
"Text": "Configure each active **Global Session Policy** so a non-default rule at **Priority 1**:\n- Sets *Okta global session cookies persist across browser sessions* to `Disabled`\n- Is enabled (`ACTIVE`) and evaluated before the built-in `Default Rule`\n\nReview group assignments to confirm the rule actually governs the intended users.",
"Url": "https://hub.prowler.com/check/signon_global_session_cookies_not_persistent"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,100 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
missing_policy_scope_finding,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
class signon_global_session_cookies_not_persistent(Check):
"""STIG V-273206 / OKTA-APP-001710.
Every active Global Session Policy must have an active Priority 1
rule that is not the built-in Default Rule, and that rule must
disable persistent global session cookies so the session does not
survive across browser restarts.
Okta evaluates sign-on policies in priority order based on group
assignments, so a permissive custom policy can govern a user's
session even when the Default Policy is strict. The check emits one
finding per active policy to surface that risk.
"""
def execute(self) -> list[CheckReportOkta]:
org_domain = signon_client.provider.identity.org_domain
missing_scope = signon_client.missing_scope.get("global_session_policies")
if missing_scope:
return [
missing_policy_scope_finding(self.metadata(), org_domain, missing_scope)
]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
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 policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
status, status_extended = _evaluate_policy(policy)
report.status = status
report.status_extended = status_extended
findings.append(report)
return findings
def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273206 requires "
"a non-default Priority 1 rule that disables persistent global "
"session cookies.",
)
if rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
use_persistent_cookie = rule.use_persistent_cookie
if use_persistent_cookie is None:
return (
"FAIL",
f"Priority 1 non-default rule '{rule.name}' in {label} "
"does not assert the 'Okta global session cookies persist across "
"browser sessions' setting.",
)
if use_persistent_cookie is False:
return (
"PASS",
f"Priority 1 non-default rule '{rule.name}' in {label} "
"disables persistent global session cookies.",
)
return (
"FAIL",
f"Priority 1 non-default rule '{rule.name}' in {label} "
"allows persistent global session cookies, leaving the session active "
"across browser restarts.",
)
@@ -9,8 +9,8 @@
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "governance",
"Description": "The **Default Global Session Policy** must have a **Priority 1** rule that is **not** the built-in `Default Rule`, and that rule must set **Maximum Okta global session idle time** to 15 minutes or less. The threshold defaults to 15 minutes and is overridable via the `okta_max_session_idle_minutes` key in the audit config.",
"Risk": "Without a 15-minute idle timeout, an unattended workstation leaves an authenticated Okta session open indefinitely, allowing an attacker physical or remote access to take over the user's identity and pivot into every downstream application that trusts Okta SSO.",
"Description": "Every active Okta **Global Session Policy** needs a **Priority 1** rule that is **not** the built-in `Default Rule`, setting *Maximum Okta global session idle time* to `15` minutes or less.\n\nOkta evaluates policies by group assignment, so a permissive custom policy can govern users. Threshold override: `okta_max_session_idle_minutes`. Aligns with **DISA STIG V-273186**.",
"Risk": "Without a `15`-minute idle timeout, an unattended workstation leaves an authenticated Okta session open indefinitely.\n\n- **Session takeover** of the user's identity by anyone with physical or remote access\n- **Lateral movement** into every downstream application that trusts Okta SSO\n- **Bypassed reauthentication** even after the user has stepped away",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/oie/en-us/content/topics/identity-engine/policies/about-okta-sign-on-policies.htm",
@@ -20,11 +20,11 @@
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Sign in to the Okta Admin Console as a Super Admin\n2. Go to Security > Global Session Policy\n3. Open the Default Policy\n4. Add or edit a non-default rule\n5. Move that rule to Priority 1 so it is evaluated before the built-in Default Rule\n6. Set 'Maximum Okta global session idle time' to 15 minutes or less\n7. Save the rule",
"Terraform": "resource \"okta_policy_rule_signon\" \"prowler_idle_timeout_15min\" {\n policy_id = okta_policy_signon.default.id\n name = \"Prowler-enforced idle timeout\"\n status = \"ACTIVE\"\n session_idle = 15\n session_persistent = false\n}\n"
"Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Security** > **Global Session Policy**.\n3. Open the **Default Policy** (and repeat for every other active policy).\n4. Add or edit a non-default rule.\n5. Move that rule to **Priority 1** so it is evaluated before the built-in `Default Rule`.\n6. Set *Maximum Okta global session idle time* to `15` minutes or less.\n7. Save the rule.",
"Terraform": "```hcl\nresource \"okta_policy_rule_signon\" \"<example_resource_name>\" {\n policy_id = okta_policy_signon.default.id\n name = \"<example_resource_name>\"\n status = \"ACTIVE\"\n priority = 1 # Critical: rule must sit at Priority 1 before the Default Rule\n session_idle = 15 # Critical: enforce idle timeout at 15 minutes or less\n session_persistent = false # Critical: avoid persistent global session cookies\n}\n```"
},
"Recommendation": {
"Text": "Configure the Default Global Session Policy so its Priority 1 non-default rule sets the Maximum Okta global session idle time to 15 minutes or less.",
"Text": "Configure each active **Global Session Policy** so a non-default rule at **Priority 1**:\n- Sets *Maximum Okta global session idle time* to `15` minutes or less\n- Is enabled (`ACTIVE`) and evaluated before the built-in `Default Rule`\n\nReview group assignments to confirm the rule actually governs the intended users.",
"Url": "https://hub.prowler.com/check/signon_global_session_idle_timeout_15min"
}
},
@@ -1,4 +1,11 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
missing_policy_scope_finding,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
@@ -8,11 +15,16 @@ DEFAULT_THRESHOLD_MINUTES = 15
class signon_global_session_idle_timeout_15min(Check):
"""STIG V-273186 / OKTA-APP-000020.
The DISA STIG requires the Okta Default Policy to have an active
Priority 1 rule that is not the built-in Default Rule, and that
rule must set the maximum Okta global session idle time to the
configured threshold or lower (defaults to 15 minutes per STIG;
override via `okta_max_session_idle_minutes` in the audit config).
Every active Global Session Policy must have an active Priority 1
rule that is not the built-in Default Rule, and that rule must set
the maximum Okta global session idle time to the configured
threshold or lower (defaults to 15 minutes per STIG; override via
`okta_max_session_idle_minutes` in the audit config).
Okta evaluates sign-on policies in priority order based on group
assignments, so a permissive custom policy can govern a user's
session even when the Default Policy is strict. The check emits one
finding per active policy to surface that risk.
"""
def execute(self) -> list[CheckReportOkta]:
@@ -21,106 +33,74 @@ class signon_global_session_idle_timeout_15min(Check):
"okta_max_session_idle_minutes", DEFAULT_THRESHOLD_MINUTES
)
org_domain = signon_client.provider.identity.org_domain
policy = self._get_default_policy()
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
missing_scope = signon_client.missing_scope.get("global_session_policies")
if missing_scope:
return [
missing_policy_scope_finding(self.metadata(), org_domain, missing_scope)
]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
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 policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
status, status_extended = _evaluate_policy(policy, threshold)
report.status = status
report.status_extended = status_extended
findings.append(report)
return findings
def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str, str]:
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273186 requires "
f"a non-default Priority 1 rule with idle timeout <= {threshold} "
"minutes.",
)
if policy.id == "default-policy-missing":
report.status = "FAIL"
report.status_extended = (
"Default Global Session Policy was not found. STIG V-273186 "
"requires the Default Policy to contain an active Priority 1 "
f"non-default rule with idle timeout <= {threshold} minutes."
)
return [report]
if policy.status and policy.status.upper() != "ACTIVE":
report.status = "FAIL"
report.status_extended = (
f"Default Global Session Policy '{policy.name}' is in "
f"status '{policy.status}'. STIG V-273186 requires an active "
"Default Policy with an active Priority 1 non-default rule."
)
return [report]
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 rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
if not active_rules:
report.status = "FAIL"
report.status_extended = (
f"Default Global Session Policy '{policy.name}' has no active "
"rules. STIG V-273186 requires an active Priority 1 non-default "
f"rule with idle timeout <= {threshold} minutes."
)
return [report]
priority_one_rule = active_rules[0]
if priority_one_rule.priority != 1:
report.status = "FAIL"
report.status_extended = (
f"Default Global Session Policy '{policy.name}' has no active "
f"Priority 1 rule. The first active rule is '{priority_one_rule.name}' "
f"at priority {priority_one_rule.priority}."
)
return [report]
if priority_one_rule.is_default or priority_one_rule.name == "Default Rule":
report.status = "FAIL"
report.status_extended = (
f"Default Global Session Policy '{policy.name}' uses "
f"'{priority_one_rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule."
)
return [report]
idle_timeout = priority_one_rule.max_session_idle_minutes
if idle_timeout is None:
report.status = "FAIL"
report.status_extended = (
f"Priority 1 non-default rule '{priority_one_rule.name}' in "
f"Default Global Session Policy '{policy.name}' does not define "
"a maximum Okta global session idle time."
)
return [report]
if idle_timeout <= threshold:
report.status = "PASS"
report.status_extended = (
f"Priority 1 non-default rule '{priority_one_rule.name}' in "
f"Default Global Session Policy '{policy.name}' sets the "
f"maximum Okta global session idle time to {idle_timeout} "
f"minutes, meeting the configured threshold of {threshold} minutes."
)
else:
report.status = "FAIL"
report.status_extended = (
f"Priority 1 non-default rule '{priority_one_rule.name}' in "
f"Default Global Session Policy '{policy.name}' sets the "
f"maximum Okta global session idle time to {idle_timeout} "
f"minutes, exceeding the configured threshold of {threshold} minutes."
)
return [report]
@staticmethod
def _get_default_policy() -> GlobalSessionPolicy:
for policy in signon_client.global_session_policies.values():
if policy.is_default or policy.name == "Default Policy":
return policy
return GlobalSessionPolicy(
id="default-policy-missing",
name="Default Policy",
priority=1,
status="MISSING",
is_default=True,
rules=[],
idle_timeout = rule.max_session_idle_minutes
if idle_timeout is None:
return (
"FAIL",
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 '{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 '{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.",
)
@@ -0,0 +1,37 @@
{
"Provider": "okta",
"CheckID": "signon_global_session_lifetime_18h",
"CheckTitle": "Default Global Session Policy has a Priority 1 non-default rule limiting session lifetime to 18 hours",
"CheckType": [],
"ServiceName": "signon",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "governance",
"Description": "Every active Okta **Global Session Policy** needs a **Priority 1** rule that is **not** the built-in `Default Rule`, setting *Maximum Okta global session lifetime* to `18` hours or less.\n\nOkta evaluates policies by group assignment, so a permissive custom policy can govern users. Threshold override: `okta_max_session_lifetime_minutes` (minutes). Aligns with **DISA STIG V-273203**.",
"Risk": "Without an enforced session lifetime, an authenticated Okta session can be reused indefinitely without reauthentication.\n\n- **Stolen session material** continues to grant access long after sign-in\n- **Authorization changes** (role revocation, group removal) take effect only on the next reauth\n- **Token replay** against downstream apps stays viable for the session window",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/oie/en-us/content/topics/identity-engine/policies/about-okta-sign-on-policies.htm",
"https://developer.okta.com/docs/api/openapi/okta-management/management/tag/Policy/"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Security** > **Global Session Policy**.\n3. Open the **Default Policy** (and repeat for every other active policy).\n4. Add or edit a non-default rule.\n5. Move that rule to **Priority 1** so it is evaluated before the built-in `Default Rule`.\n6. Set *Maximum Okta global session lifetime* to `18` hours or less. Do **not** set it to `0`, which disables the limit.\n7. Save the rule.",
"Terraform": "```hcl\nresource \"okta_policy_rule_signon\" \"<example_resource_name>\" {\n policy_id = okta_policy_signon.default.id\n name = \"<example_resource_name>\"\n status = \"ACTIVE\"\n priority = 1 # Critical: rule must sit at Priority 1 before the Default Rule\n session_lifetime = 1080 # Critical: 18 hours in minutes; do not use 0 (disables the limit)\n session_persistent = false # Critical: avoid persistent global session cookies\n}\n```"
},
"Recommendation": {
"Text": "Configure each active **Global Session Policy** so a non-default rule at **Priority 1**:\n- Sets *Maximum Okta global session lifetime* to `18` hours or less (`1080` minutes)\n- Never sets the lifetime to `0`, which disables the limit\n- Is enabled (`ACTIVE`) and evaluated before the built-in `Default Rule`\n\nReview group assignments to confirm the rule actually governs the intended users.",
"Url": "https://hub.prowler.com/check/signon_global_session_lifetime_18h"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,114 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
missing_policy_scope_finding,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
DEFAULT_THRESHOLD_MINUTES = 18 * 60
class signon_global_session_lifetime_18h(Check):
"""STIG V-273203 / OKTA-APP-001665.
Every active Global Session Policy must have an active Priority 1
rule that is not the built-in Default Rule, and that rule must set
the maximum Okta global session lifetime to the configured threshold
or lower (defaults to 18 hours per STIG; override via
`okta_max_session_lifetime_minutes` in the audit config).
Okta evaluates sign-on policies in priority order based on group
assignments, so a permissive custom policy can govern a user's
session even when the Default Policy is strict. The check emits one
finding per active policy to surface that risk.
"""
def execute(self) -> list[CheckReportOkta]:
audit_config = signon_client.audit_config or {}
threshold = audit_config.get(
"okta_max_session_lifetime_minutes", DEFAULT_THRESHOLD_MINUTES
)
org_domain = signon_client.provider.identity.org_domain
missing_scope = signon_client.missing_scope.get("global_session_policies")
if missing_scope:
return [
missing_policy_scope_finding(self.metadata(), org_domain, missing_scope)
]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
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 policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
status, status_extended = _evaluate_policy(policy, threshold)
report.status = status
report.status_extended = status_extended
findings.append(report)
return findings
def _evaluate_policy(policy: GlobalSessionPolicy, threshold: int) -> tuple[str, str]:
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-273203 requires "
f"a non-default Priority 1 rule with session lifetime <= {threshold} "
"minutes.",
)
if rule.is_default or rule.name == "Default Rule":
return (
"FAIL",
f"{label} uses '{rule.name}' as its active Priority 1 rule. "
"The STIG requires a non-default Priority 1 rule.",
)
lifetime = rule.max_session_lifetime_minutes
if lifetime is None:
return (
"FAIL",
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 '{rule.name}' in {label} "
"disables the maximum Okta global session lifetime by setting it "
"to 0 minutes.",
)
if lifetime <= threshold:
return (
"PASS",
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 '{rule.name}' in {label} "
f"sets the maximum Okta global session lifetime to {lifetime} minutes, "
f"exceeding the configured threshold of {threshold} minutes.",
)
@@ -0,0 +1,38 @@
{
"Provider": "okta",
"CheckID": "signon_global_session_policy_network_zone_enforced",
"CheckTitle": "Default Global Session Policy applies a Network Zone condition aligned with the Access Control Policy",
"CheckType": [],
"ServiceName": "signon",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "NotDefined",
"ResourceGroup": "governance",
"Description": "Every active Okta **Global Session Policy** must apply the *IF User's IP is* condition, mapped to a **Network Zone**, on its **Priority 1** active rule.\n\nUnlike the idle / lifetime / cookie STIGs, this control does **not** exclude the built-in `Default Rule`. Okta evaluates policies by group assignment. Aligns with **DISA STIG V-279691**.",
"Risk": "When the Global Session Policy does not restrict access by **Network Zone**, every authenticated entity establishes a session regardless of source IP.\n\n- **Stolen credentials** reach the Okta dashboard from any internet-routable address\n- **Out-of-band sessions** bypass the organization's Access Control Policy\n- **Network anomalies** cannot become deny decisions at sign-on",
"RelatedUrl": "",
"AdditionalURLs": [
"https://help.okta.com/oie/en-us/content/topics/security/network/network-zones.htm",
"https://help.okta.com/oie/en-us/content/topics/identity-engine/policies/about-okta-sign-on-policies.htm",
"https://developer.okta.com/docs/api/openapi/okta-management/management/tag/Policy/"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "1. Sign in to the **Okta Admin Console** as a *Super Admin*.\n2. Navigate to **Security** > **Networks** and define the **Network Zones** (allow / deny) that match the organization's Access Control Policy.\n3. Navigate to **Security** > **Global Session Policy**.\n4. Open the **Default Policy** (and repeat for every other active policy).\n5. Edit the rule that sits at **Priority 1**, or add a new one and move it to **Priority 1**.\n6. Under *Conditions*, set *IF User's IP is* to `In zone` (allow) or `Not in zone` (deny) and select the **Network Zone**.\n7. Save the rule.",
"Terraform": "```hcl\nresource \"okta_policy_rule_signon\" \"<example_resource_name>\" {\n policy_id = okta_policy_signon.default.id\n name = \"<example_resource_name>\"\n status = \"ACTIVE\"\n priority = 1 # Critical: rule must sit at Priority 1\n network_connection = \"ZONE\" # Critical: bind the rule to a Network Zone\n network_includes = [okta_network_zone.allowed.id] # Critical: zones that reflect the Access Control Policy\n}\n```"
},
"Recommendation": {
"Text": "Configure the **Priority 1** active rule in each Global Session Policy so it:\n- Maps the *IF User's IP is* condition to a **Network Zone** aligned with the organization's Access Control Policy\n- Uses `In zone` for allow-list zones and `Not in zone` for deny-list zones\n- Is enabled (`ACTIVE`) and evaluated before the built-in `Default Rule`, or *is* the `Default Rule` itself\n\nReview group assignments to confirm the rule actually governs the intended users.",
"Url": "https://hub.prowler.com/check/signon_global_session_policy_network_zone_enforced"
}
},
"Categories": [
"identity-access"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,95 @@
from prowler.lib.check.models import Check, CheckReportOkta
from prowler.providers.okta.services.signon.lib.signon_helpers import (
active_policies,
missing_policy_scope_finding,
no_active_policies_finding,
policy_label,
priority_one_active_rule,
)
from prowler.providers.okta.services.signon.signon_client import signon_client
from prowler.providers.okta.services.signon.signon_service import GlobalSessionPolicy
class signon_global_session_policy_network_zone_enforced(Check):
"""STIG V-279691 / OKTA-APP-003242.
Every active Global Session Policy must apply an "IF User's IP is"
condition mapped to a Network Zone on its Priority 1 active rule so
access can be allowed or denied per the organization's Access
Control Policy.
Unlike the idle / lifetime / persistent-cookie STIGs, V-279691 does
not exclude the built-in Default Rule, so a zone condition on the
Default Rule is still effective when no non-default rule sits at
Priority 1.
The check emits one finding per active policy because Okta evaluates
sign-on policies in priority order based on group assignments, and a
permissive custom policy can govern a user's session even when the
Default Policy is strict.
"""
def execute(self) -> list[CheckReportOkta]:
org_domain = signon_client.provider.identity.org_domain
missing_scope = signon_client.missing_scope.get("global_session_policies")
if missing_scope:
return [
missing_policy_scope_finding(self.metadata(), org_domain, missing_scope)
]
policies = active_policies(signon_client.global_session_policies)
if not policies:
return [
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 policies:
report = CheckReportOkta(
metadata=self.metadata(), resource=policy, org_domain=org_domain
)
status, status_extended = _evaluate_policy(policy)
report.status = status
report.status_extended = status_extended
findings.append(report)
return findings
def _evaluate_policy(policy: GlobalSessionPolicy) -> tuple[str, str]:
label = policy_label(policy)
rule = priority_one_active_rule(policy)
if rule is None:
return (
"FAIL",
f"{label} has no Priority 1 active rule. STIG V-279691 requires "
"the policy to apply an IP-based Network Zone condition on its "
"Priority 1 active rule.",
)
rule_kind = (
"built-in Default Rule"
if rule.is_default or rule.name == "Default Rule"
else "non-default rule"
)
has_zones = bool(rule.network_zones_include or rule.network_zones_exclude)
if has_zones:
return (
"PASS",
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} '{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.",
)
@@ -29,18 +29,51 @@ def _next_after_cursor(resp) -> Optional[str]:
return None
REQUIRED_SCOPES: dict[str, str] = {
"global_session_policies": "okta.policies.read",
"sign_in_pages": "okta.brands.read",
}
class Signon(OktaService):
"""Fetches OKTA_SIGN_ON policies and their rules.
"""Fetches OKTA_SIGN_ON policies, rules, and brand sign-in pages.
Populates `self.global_session_policies` keyed by policy id. Each
policy carries its rules; downstream checks read directly from this
structure.
Also populates `self.sign_in_pages` keyed by brand id with sign-in page
HTML used by the DOD warning-banner check. When a brand has no
customized page, the service falls back to the default sign-in page
exposed by the Okta Management API and tracks it with
`is_customized=False`.
Before each fetch the service compares its required OAuth scope
(see `REQUIRED_SCOPES`) against the access token's granted scopes
(`provider.identity.granted_scopes`). When a scope is known to be
missing, the fetch is skipped and the resource is recorded in
`self.missing_scope` so checks can report the missing scope explicitly
instead of emitting a misleading "no resources returned" finding.
When granted_scopes is empty (token decode unavailable), the service
treats permissions as unknown and attempts the fetch — preserving
the prior behavior.
"""
def __init__(self, provider):
super().__init__(__class__.__name__, provider)
granted = set(getattr(provider.identity, "granted_scopes", None) or [])
self.missing_scope: dict[str, Optional[str]] = {
resource: (scope if granted and scope not in granted else None)
for resource, scope in REQUIRED_SCOPES.items()
}
self.global_session_policies: dict[str, GlobalSessionPolicy] = (
self._list_global_session_policies()
{}
if self.missing_scope["global_session_policies"]
else self._list_global_session_policies()
)
self.sign_in_pages: dict[str, SignInPage] = (
{} if self.missing_scope["sign_in_pages"] else self._list_sign_in_pages()
)
def _list_global_session_policies(self) -> dict:
@@ -125,6 +158,74 @@ class Signon(OktaService):
)
return rules_out
def _list_sign_in_pages(self) -> dict:
logger.info("Signon - Listing brand sign-in pages...")
try:
return self._run(self._fetch_brands_and_pages())
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return {}
async def _fetch_brands_and_pages(self) -> dict:
result: dict[str, SignInPage] = {}
all_brands, err = await self._paginate(
lambda after: self.client.list_brands(after=after)
)
if err is not None:
logger.error(f"Error listing brands: {err}")
return result
for brand in all_brands:
brand_id = getattr(brand, "id", "") or ""
brand_name = getattr(brand, "name", "") or ""
result[brand_id] = await self._fetch_sign_in_page(brand_id, brand_name)
return result
async def _fetch_sign_in_page(self, brand_id: str, brand_name: str) -> "SignInPage":
page_result = await self.client.get_customized_sign_in_page(brand_id)
page_err = page_result[-1]
page_data = page_result[0]
if page_err is None:
return SignInPage(
brand_id=brand_id,
brand_name=brand_name,
is_customized=True,
page_content=getattr(page_data, "page_content", None),
)
if not self._is_missing_customized_page_error(page_err):
return SignInPage(
brand_id=brand_id,
brand_name=brand_name,
is_customized=False,
fetch_error=str(page_err),
)
default_page_result = await self.client.get_default_sign_in_page(brand_id)
default_page_err = default_page_result[-1]
default_page_data = default_page_result[0]
if default_page_err is not None:
return SignInPage(
brand_id=brand_id,
brand_name=brand_name,
is_customized=False,
fetch_error=str(default_page_err),
)
return SignInPage(
brand_id=brand_id,
brand_name=brand_name,
is_customized=False,
page_content=getattr(default_page_data, "page_content", None),
)
@staticmethod
def _is_missing_customized_page_error(error) -> bool:
err_text = str(error).lower()
return "404" in err_text or "not found" in err_text or "e0000007" in err_text
@staticmethod
async def _paginate(fetch):
"""Drain all pages of an SDK list call.
@@ -176,3 +277,11 @@ class GlobalSessionPolicy(BaseModel):
status: str = ""
is_default: bool = False
rules: list[GlobalSessionPolicyRule] = []
class SignInPage(BaseModel):
brand_id: str
brand_name: str = ""
is_customized: bool = False
page_content: Optional[str] = None
fetch_error: Optional[str] = None
+14 -1
View File
@@ -11,18 +11,31 @@ def set_mocked_okta_provider(
session: OktaSession = None,
identity: OktaIdentityInfo = None,
audit_config: dict = None,
scopes: list[str] = None,
):
if session is None:
session = OktaSession(
org_domain=OKTA_ORG_DOMAIN,
client_id=OKTA_CLIENT_ID,
scopes=["okta.policies.read"],
scopes=(
scopes
if scopes is not None
else [
"okta.policies.read",
"okta.brands.read",
"okta.networkZones.read",
"okta.apiTokens.read",
"okta.roles.read",
"okta.authenticators.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()
+122
View File
@@ -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
):
@@ -0,0 +1,41 @@
from unittest import mock
from prowler.providers.okta.services.apitoken.api_token_service import OktaApiToken
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def build_api_token_client(
tokens: dict = None,
known_network_zone_ids: set[str] = None,
missing_scopes: list[str] = None,
):
client = mock.MagicMock()
client.api_tokens = tokens or {}
client.known_network_zone_ids = known_network_zone_ids or {"nzo-corp"}
client.missing_scopes = missing_scopes or []
client.provider = set_mocked_okta_provider()
return client
def api_token(
token_id: str = "00Tabcdefg1234567890",
name: str = "CI token",
*,
user_id: str = "00uabcdefg1234567890",
network_connection: str = "ZONE",
network_includes: list[str] = None,
network_excludes: list[str] = None,
owner_roles: list[str] = None,
):
return OktaApiToken(
id=token_id,
name=name,
client_name="Okta API",
user_id=user_id,
network_connection=network_connection,
network_includes=(
network_includes if network_includes is not None else ["nzo-corp"]
),
network_excludes=network_excludes or [],
owner_roles=owner_roles or ["READ_ONLY_ADMIN"],
)
@@ -0,0 +1,248 @@
from types import SimpleNamespace
from unittest import mock
import pytest
from prowler.providers.okta.okta_provider import DEFAULT_SCOPES
from prowler.providers.okta.services.apitoken.api_token_service import (
API_TOKENS_READ_SCOPE,
NETWORK_ZONES_READ_SCOPE,
ROLES_READ_SCOPE,
ApiToken,
)
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def _resp(headers: dict = None):
return SimpleNamespace(headers=headers or {})
def _sdk_token(
token_id: str = "00Tabcdefg1234567890",
name: str = "CI token",
*,
user_id: str = "00uabcdefg1234567890",
connection: str = "ZONE",
include: list[str] = None,
exclude: list[str] = None,
):
return SimpleNamespace(
id=token_id,
name=name,
client_name="Okta API",
user_id=user_id,
network=SimpleNamespace(
connection=connection,
include=include if include is not None else ["nzo-corp"],
exclude=exclude or [],
),
)
def _sdk_role(role_type: str):
return SimpleNamespace(type=role_type, label=role_type.replace("_", " ").title())
def _sdk_zone(zone_id: str, name: str):
return SimpleNamespace(id=zone_id, name=name)
class Test_ApiToken_service:
def test_fetches_tokens_roles_and_known_network_zones(self):
provider = set_mocked_okta_provider()
token = _sdk_token()
async def fake_list_api_tokens():
return ([token], _resp({}), None)
async def fake_list_assigned_roles_for_user(user_id):
assert user_id == token.user_id
return ([_sdk_role("READ_ONLY_ADMIN")], _resp({}), None)
async def fake_list_network_zones(after=None, limit=None):
assert after is None
assert limit == 200
return ([_sdk_zone("nzo-corp", "Corporate")], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_api_tokens = fake_list_api_tokens
mocked.list_assigned_roles_for_user = fake_list_assigned_roles_for_user
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = ApiToken(provider)
assert set(service.api_tokens.keys()) == {token.id}
assert service.api_tokens[token.id].network_connection == "ZONE"
assert service.api_tokens[token.id].owner_roles == ["READ_ONLY_ADMIN"]
assert service.known_network_zone_ids == {"nzo-corp", "Corporate"}
def test_role_fetch_error_keeps_token_with_empty_roles(self):
provider = set_mocked_okta_provider()
token = _sdk_token()
async def fake_list_api_tokens():
return ([token], _resp({}), None)
async def fake_roles_error(user_id):
assert user_id == token.user_id
return ([], _resp({}), Exception("forbidden"))
async def fake_list_network_zones(after=None, limit=None):
assert after is None
assert limit == 200
return ([], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_api_tokens = fake_list_api_tokens
mocked.list_assigned_roles_for_user = fake_roles_error
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = ApiToken(provider)
assert service.api_tokens[token.id].owner_roles == []
def test_paginates_known_network_zones_for_token_validation(self):
provider = set_mocked_okta_provider()
token = _sdk_token(include=["nzo-page-2"])
next_link = '<https://acme.okta.com/api/v1/zones?after=cursor-2>; rel="next"'
async def fake_list_api_tokens():
return ([token], _resp({}), None)
async def fake_list_assigned_roles_for_user(user_id):
assert user_id == token.user_id
return ([_sdk_role("READ_ONLY_ADMIN")], _resp({}), None)
async def fake_list_network_zones(after=None, limit=None):
assert limit == 200
if after is None:
return (
[_sdk_zone("nzo-page-1", "First")],
_resp({"link": next_link}),
None,
)
return ([_sdk_zone("nzo-page-2", "Second")], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_api_tokens = fake_list_api_tokens
mocked.list_assigned_roles_for_user = fake_list_assigned_roles_for_user
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = ApiToken(provider)
assert service.known_network_zone_ids == {
"nzo-page-1",
"First",
"nzo-page-2",
"Second",
}
def test_returns_empty_on_token_api_error(self):
provider = set_mocked_okta_provider()
async def failing():
return ([], _resp({}), Exception("forbidden"))
async def fake_list_network_zones(after=None, limit=None):
assert after is None
assert limit == 200
return ([], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_api_tokens = failing
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = ApiToken(provider)
assert service.api_tokens == {}
@pytest.mark.parametrize(
"missing_scope, expected_calls, expected_tokens, expected_zones, expected_roles",
[
(
API_TOKENS_READ_SCOPE,
[],
set(),
set(),
None,
),
(
NETWORK_ZONES_READ_SCOPE,
["list_api_tokens", "list_assigned_roles_for_user"],
{"00Tabcdefg1234567890"},
set(),
["READ_ONLY_ADMIN"],
),
(
ROLES_READ_SCOPE,
["list_network_zones", "list_api_tokens"],
{"00Tabcdefg1234567890"},
{"nzo-corp", "Corporate"},
[],
),
],
)
def test_missing_scope_skips_corresponding_sdk_call(
self,
missing_scope,
expected_calls,
expected_tokens,
expected_zones,
expected_roles,
):
provider = set_mocked_okta_provider(
scopes=[scope for scope in DEFAULT_SCOPES if scope != missing_scope]
)
token = _sdk_token()
calls = []
async def fake_list_api_tokens():
if missing_scope == API_TOKENS_READ_SCOPE:
raise AssertionError("list_api_tokens must not be called")
calls.append("list_api_tokens")
return ([token], _resp({}), None)
async def fake_list_assigned_roles_for_user(user_id):
if missing_scope == ROLES_READ_SCOPE:
raise AssertionError("list_assigned_roles_for_user must not be called")
assert user_id == token.user_id
calls.append("list_assigned_roles_for_user")
return ([_sdk_role("READ_ONLY_ADMIN")], _resp({}), None)
async def fake_list_network_zones(after=None, limit=None):
if missing_scope in {API_TOKENS_READ_SCOPE, NETWORK_ZONES_READ_SCOPE}:
raise AssertionError("list_network_zones must not be called")
assert after is None
assert limit == 200
calls.append("list_network_zones")
return ([_sdk_zone("nzo-corp", "Corporate")], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_api_tokens = fake_list_api_tokens
mocked.list_assigned_roles_for_user = fake_list_assigned_roles_for_user
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = ApiToken(provider)
assert service.missing_scopes == [missing_scope]
assert set(service.api_tokens.keys()) == expected_tokens
assert service.known_network_zone_ids == expected_zones
if expected_roles is not None:
assert service.api_tokens[token.id].owner_roles == expected_roles
assert calls == expected_calls
@@ -0,0 +1,60 @@
from unittest import mock
import pytest
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.api_token.api_token_fixtures import (
api_token,
build_api_token_client,
)
CHECK_PATH = (
"prowler.providers.okta.services.apitoken."
"apitoken_not_super_admin.apitoken_not_super_admin.api_token_client"
)
def _run_check(api_token_client):
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(CHECK_PATH, new=api_token_client),
):
from prowler.providers.okta.services.apitoken.apitoken_not_super_admin.apitoken_not_super_admin import (
apitoken_not_super_admin,
)
return apitoken_not_super_admin().execute()
class Test_apitoken_not_super_admin:
@pytest.mark.parametrize(
"missing_scope", ["okta.apiTokens.read", "okta.roles.read"]
)
def test_missing_required_scope_returns_manual(self, missing_scope):
findings = _run_check(
build_api_token_client({}, missing_scopes=[missing_scope])
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert missing_scope in findings[0].status_extended
def test_no_tokens_returns_no_findings(self):
findings = _run_check(build_api_token_client({}))
assert findings == []
def test_token_owner_without_super_admin_passes(self):
token = api_token(owner_roles=["READ_ONLY_ADMIN"])
findings = _run_check(build_api_token_client({token.id: token}))
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == token.id
def test_token_owner_with_super_admin_fails(self):
token = api_token(owner_roles=["SUPER_ADMIN"])
findings = _run_check(build_api_token_client({token.id: token}))
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "Super Admin" in findings[0].status_extended
@@ -0,0 +1,75 @@
from unittest import mock
import pytest
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.api_token.api_token_fixtures import (
api_token,
build_api_token_client,
)
CHECK_PATH = (
"prowler.providers.okta.services.apitoken."
"apitoken_restricted_to_network_zone.apitoken_restricted_to_network_zone.api_token_client"
)
def _run_check(api_token_client):
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(CHECK_PATH, new=api_token_client),
):
from prowler.providers.okta.services.apitoken.apitoken_restricted_to_network_zone.apitoken_restricted_to_network_zone import (
apitoken_restricted_to_network_zone,
)
return apitoken_restricted_to_network_zone().execute()
class Test_apitoken_restricted_to_network_zone:
@pytest.mark.parametrize(
"missing_scope", ["okta.apiTokens.read", "okta.networkZones.read"]
)
def test_missing_required_scope_returns_manual(self, missing_scope):
findings = _run_check(
build_api_token_client({}, missing_scopes=[missing_scope])
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert missing_scope in findings[0].status_extended
def test_no_tokens_returns_no_findings(self):
findings = _run_check(build_api_token_client({}))
assert findings == []
def test_token_restricted_to_known_network_zone_passes(self):
token = api_token(network_connection="ZONE", network_includes=["nzo-corp"])
findings = _run_check(
build_api_token_client(
{token.id: token}, known_network_zone_ids={"nzo-corp"}
)
)
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == token.id
def test_token_open_to_anywhere_fails(self):
token = api_token(network_connection="ANYWHERE", network_includes=[])
findings = _run_check(build_api_token_client({token.id: token}))
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "from any IP" in findings[0].status_extended
def test_token_restricted_to_unknown_zone_fails(self):
token = api_token(network_connection="ZONE", network_includes=["nzo-missing"])
findings = _run_check(
build_api_token_client(
{token.id: token}, known_network_zone_ids={"nzo-corp"}
)
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "unknown Network Zone" in findings[0].status_extended
@@ -0,0 +1,73 @@
from unittest import mock
from prowler.providers.okta.services.authenticator.authenticator_service import (
OktaAuthenticator,
PasswordPolicy,
)
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def build_authenticator_client(
password_policies: dict = None,
authenticators: dict = None,
missing_scopes: list[str] = None,
):
client = mock.MagicMock()
client.password_policies = password_policies or {}
client.authenticators = authenticators or {}
client.missing_scopes = missing_scopes or []
client.provider = set_mocked_okta_provider()
return client
def password_policy(
policy_id: str = "pol-password",
name: str = "Default Password Policy",
*,
status: str = "ACTIVE",
priority: int = 1,
max_attempts: int = 3,
min_length: int = 15,
min_upper_case: int = 1,
min_lower_case: int = 1,
min_number: int = 1,
min_symbol: int = 1,
min_age_minutes: int = 1440,
max_age_days: int = 60,
history_count: int = 5,
common_password_check: bool = True,
):
return PasswordPolicy(
id=policy_id,
name=name,
status=status,
priority=priority,
max_attempts=max_attempts,
min_length=min_length,
min_upper_case=min_upper_case,
min_lower_case=min_lower_case,
min_number=min_number,
min_symbol=min_symbol,
min_age_minutes=min_age_minutes,
max_age_days=max_age_days,
history_count=history_count,
common_password_check=common_password_check,
)
def authenticator(
auth_id: str = "aut-okta-verify",
key: str = "okta_verify",
name: str = "Okta Verify",
*,
status: str = "ACTIVE",
fips: str = "REQUIRED",
):
return OktaAuthenticator(
id=auth_id,
key=key,
name=name,
status=status,
type="app",
fips=fips,
)
@@ -0,0 +1,64 @@
from unittest import mock
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.authenticator.authenticator_fixtures import (
authenticator,
build_authenticator_client,
)
CHECK_PATH = (
"prowler.providers.okta.services.authenticator."
"authenticator_okta_verify_fips_compliant."
"authenticator_okta_verify_fips_compliant.authenticator_client"
)
def _run_check(authenticator_client):
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(CHECK_PATH, new=authenticator_client),
):
from prowler.providers.okta.services.authenticator.authenticator_okta_verify_fips_compliant.authenticator_okta_verify_fips_compliant import (
authenticator_okta_verify_fips_compliant,
)
return authenticator_okta_verify_fips_compliant().execute()
class Test_authenticator_okta_verify_fips_compliant:
def test_missing_authenticators_scope_returns_manual(self):
findings = _run_check(
build_authenticator_client(
authenticators={}, missing_scopes=["okta.authenticators.read"]
)
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert "okta.authenticators.read" in findings[0].status_extended
def test_okta_verify_fips_required_passes(self):
okta_verify = authenticator(key="okta_verify", fips="REQUIRED")
findings = _run_check(
build_authenticator_client(authenticators={okta_verify.id: okta_verify})
)
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == okta_verify.id
def test_okta_verify_without_fips_required_fails(self):
okta_verify = authenticator(key="okta_verify", fips="OPTIONAL")
findings = _run_check(
build_authenticator_client(authenticators={okta_verify.id: okta_verify})
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "FIPS" in findings[0].status_extended
def test_missing_okta_verify_fails(self):
findings = _run_check(build_authenticator_client(authenticators={}))
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "Okta Verify authenticator is not active" in findings[0].status_extended
@@ -0,0 +1,212 @@
from unittest import mock
import pytest
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.authenticator.authenticator_fixtures import (
build_authenticator_client,
password_policy,
)
PASSWORD_POLICY_CHECK_CASES = [
(
"authenticator_password_common_password_check",
"common_password_check",
True,
False,
"common-password dictionary checks",
),
(
"authenticator_password_complexity_lowercase",
"min_lower_case",
1,
0,
"at least one lowercase character",
),
(
"authenticator_password_complexity_number",
"min_number",
1,
0,
"at least one numeric character",
),
(
"authenticator_password_complexity_symbol",
"min_symbol",
1,
0,
"at least one symbol character",
),
(
"authenticator_password_complexity_uppercase",
"min_upper_case",
1,
0,
"at least one uppercase character",
),
(
"authenticator_password_history_5",
"history_count",
5,
4,
"password history of at least 5 previous passwords",
),
(
"authenticator_password_lockout_threshold_3",
"max_attempts",
3,
4,
"password lockout after 3 or fewer failed attempts",
),
(
"authenticator_password_maximum_age_60d",
"max_age_days",
60,
61,
"maximum password age of 60 days or less",
),
(
"authenticator_password_minimum_age_24h",
"min_age_minutes",
1440,
1439,
"minimum password age of at least 24 hours",
),
(
"authenticator_password_minimum_length_15",
"min_length",
15,
14,
"minimum password length of at least 15 characters",
),
]
def _run_password_policy_check(check_name: str, authenticator_client):
check_path = (
f"prowler.providers.okta.services.authenticator.{check_name}."
f"{check_name}.authenticator_client"
)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(check_path, new=authenticator_client),
):
module = __import__(
f"prowler.providers.okta.services.authenticator.{check_name}.{check_name}",
fromlist=[check_name],
)
return getattr(module, check_name)().execute()
class Test_authenticator_password_policy_checks:
@pytest.mark.parametrize(
"check_name, field_name, compliant_value, non_compliant_value, expected_phrase",
PASSWORD_POLICY_CHECK_CASES,
)
def test_missing_policies_scope_returns_manual(
self,
check_name,
field_name,
compliant_value,
non_compliant_value,
expected_phrase,
):
findings = _run_password_policy_check(
check_name,
build_authenticator_client(
password_policies={}, missing_scopes=["okta.policies.read"]
),
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert "okta.policies.read" in findings[0].status_extended
@pytest.mark.parametrize(
"check_name, field_name, compliant_value, non_compliant_value, expected_phrase",
PASSWORD_POLICY_CHECK_CASES,
)
def test_no_active_password_policies_fails(
self,
check_name,
field_name,
compliant_value,
non_compliant_value,
expected_phrase,
):
findings = _run_password_policy_check(
check_name, build_authenticator_client(password_policies={})
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "No active Okta Password Policies" in findings[0].status_extended
assert expected_phrase in findings[0].status_extended
@pytest.mark.parametrize(
"check_name, field_name, compliant_value, non_compliant_value, expected_phrase",
PASSWORD_POLICY_CHECK_CASES,
)
def test_compliant_password_policy_passes(
self,
check_name,
field_name,
compliant_value,
non_compliant_value,
expected_phrase,
):
policy = password_policy(**{field_name: compliant_value})
findings = _run_password_policy_check(
check_name,
build_authenticator_client(password_policies={policy.id: policy}),
)
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == policy.id
assert expected_phrase in findings[0].status_extended
@pytest.mark.parametrize(
"check_name, field_name, compliant_value, non_compliant_value, expected_phrase",
PASSWORD_POLICY_CHECK_CASES,
)
def test_non_compliant_password_policy_fails(
self,
check_name,
field_name,
compliant_value,
non_compliant_value,
expected_phrase,
):
policy = password_policy(**{field_name: non_compliant_value})
findings = _run_password_policy_check(
check_name,
build_authenticator_client(password_policies={policy.id: policy}),
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert findings[0].resource_id == policy.id
assert expected_phrase in findings[0].status_extended
def test_multiple_active_password_policies_emit_one_finding_each(self):
check_name = "authenticator_password_minimum_length_15"
compliant = password_policy(policy_id="pol-good", name="Strict", min_length=15)
weak = password_policy(
policy_id="pol-weak", name="Weak", min_length=8, priority=2
)
findings = _run_password_policy_check(
check_name,
build_authenticator_client(
password_policies={compliant.id: compliant, weak.id: weak}
),
)
assert len(findings) == 2
by_name = {finding.resource_name: finding for finding in findings}
assert by_name["Strict"].status == "PASS"
assert by_name["Weak"].status == "FAIL"
@@ -0,0 +1,203 @@
from types import SimpleNamespace
from unittest import mock
import pytest
from prowler.providers.okta.okta_provider import DEFAULT_SCOPES
from prowler.providers.okta.services.authenticator.authenticator_service import (
AUTHENTICATORS_READ_SCOPE,
POLICIES_READ_SCOPE,
Authenticator,
OktaAuthenticator,
PasswordPolicy,
)
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def _resp(headers: dict = None):
return SimpleNamespace(headers=headers or {})
def _sdk_password_policy(policy_id: str = "pol-password", name: str = "Default"):
return SimpleNamespace(
id=policy_id,
name=name,
priority=1,
status="ACTIVE",
system=True,
settings=SimpleNamespace(
password=SimpleNamespace(
lockout=SimpleNamespace(max_attempts=3),
complexity=SimpleNamespace(
min_length=15,
min_upper_case=1,
min_lower_case=1,
min_number=1,
min_symbol=1,
dictionary=SimpleNamespace(common=True),
),
age=SimpleNamespace(
min_age_minutes=1440,
max_age_days=60,
history_count=5,
),
)
),
)
def _sdk_authenticator(
auth_id: str = "aut-okta-verify",
key: str = "okta_verify",
status: str = "ACTIVE",
fips: str = "REQUIRED",
):
return SimpleNamespace(
id=auth_id,
key=key,
name="Okta Verify" if key == "okta_verify" else "Smart Card IdP",
status=status,
type="app",
settings=SimpleNamespace(compliance=SimpleNamespace(fips=fips)),
)
class Test_Authenticator_service:
def test_fetches_password_policies_and_authenticators(self):
provider = set_mocked_okta_provider()
policy = _sdk_password_policy()
okta_verify = _sdk_authenticator()
async def fake_list_policies(type, after=None):
assert type == "PASSWORD"
assert after is None
return ([policy], _resp({}), None)
async def fake_list_authenticators():
return ([okta_verify], _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_authenticators = fake_list_authenticators
mocked_client_cls.return_value = mocked
service = Authenticator(provider)
assert isinstance(service.password_policies[policy.id], PasswordPolicy)
assert service.password_policies[policy.id].min_length == 15
assert isinstance(service.authenticators[okta_verify.id], OktaAuthenticator)
assert service.authenticators[okta_verify.id].fips == "REQUIRED"
def test_returns_empty_collections_on_api_errors(self):
provider = set_mocked_okta_provider()
async def failing_policies(type, after=None):
assert type == "PASSWORD"
assert after is None
return ([], _resp({}), Exception("forbidden"))
async def failing_authenticators():
return ([], _resp({}), Exception("forbidden"))
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_policies = failing_policies
mocked.list_authenticators = failing_authenticators
mocked_client_cls.return_value = mocked
service = Authenticator(provider)
assert service.password_policies == {}
assert service.authenticators == {}
def test_paginates_password_policies(self):
provider = set_mocked_okta_provider()
page_1 = _sdk_password_policy("pol-1", "First")
page_2 = _sdk_password_policy("pol-2", "Second")
next_link = '<https://acme.okta.com/api/v1/policies?after=cursor-2>; rel="next"'
calls = []
async def fake_list_policies(type, after=None):
assert type == "PASSWORD"
calls.append(after)
if after is None:
return ([page_1], _resp({"link": next_link}), None)
return ([page_2], _resp({}), None)
async def fake_list_authenticators():
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_authenticators = fake_list_authenticators
mocked_client_cls.return_value = mocked
service = Authenticator(provider)
assert calls == [None, "cursor-2"]
assert set(service.password_policies.keys()) == {"pol-1", "pol-2"}
@pytest.mark.parametrize(
"missing_scope, expected_calls, expected_policies, expected_authenticators",
[
(
POLICIES_READ_SCOPE,
["list_authenticators"],
set(),
{"aut-okta-verify"},
),
(
AUTHENTICATORS_READ_SCOPE,
["list_policies"],
{"pol-password"},
set(),
),
],
)
def test_missing_scope_skips_corresponding_sdk_call(
self,
missing_scope,
expected_calls,
expected_policies,
expected_authenticators,
):
provider = set_mocked_okta_provider(
scopes=[scope for scope in DEFAULT_SCOPES if scope != missing_scope]
)
policy = _sdk_password_policy()
okta_verify = _sdk_authenticator()
calls = []
async def fake_list_policies(type, after=None):
if missing_scope == POLICIES_READ_SCOPE:
raise AssertionError("list_policies must not be called")
assert type == "PASSWORD"
assert after is None
calls.append("list_policies")
return ([policy], _resp({}), None)
async def fake_list_authenticators():
if missing_scope == AUTHENTICATORS_READ_SCOPE:
raise AssertionError("list_authenticators must not be called")
calls.append("list_authenticators")
return ([okta_verify], _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_authenticators = fake_list_authenticators
mocked_client_cls.return_value = mocked
service = Authenticator(provider)
assert service.missing_scopes == [missing_scope]
assert set(service.password_policies.keys()) == expected_policies
assert set(service.authenticators.keys()) == expected_authenticators
assert calls == expected_calls
@@ -0,0 +1,73 @@
from unittest import mock
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.authenticator.authenticator_fixtures import (
authenticator,
build_authenticator_client,
)
CHECK_PATH = (
"prowler.providers.okta.services.authenticator."
"authenticator_smart_card_active.authenticator_smart_card_active.authenticator_client"
)
def _run_check(authenticator_client):
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(CHECK_PATH, new=authenticator_client),
):
from prowler.providers.okta.services.authenticator.authenticator_smart_card_active.authenticator_smart_card_active import (
authenticator_smart_card_active,
)
return authenticator_smart_card_active().execute()
class Test_authenticator_smart_card_active:
def test_missing_authenticators_scope_returns_manual(self):
findings = _run_check(
build_authenticator_client(
authenticators={}, missing_scopes=["okta.authenticators.read"]
)
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert "okta.authenticators.read" in findings[0].status_extended
def test_smart_card_active_passes(self):
smart_card = authenticator(
auth_id="aut-smart-card",
key="smart_card_idp",
name="Smart Card IdP",
status="ACTIVE",
)
findings = _run_check(
build_authenticator_client(authenticators={smart_card.id: smart_card})
)
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == smart_card.id
def test_missing_smart_card_fails(self):
findings = _run_check(build_authenticator_client(authenticators={}))
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "not active" in findings[0].status_extended
def test_inactive_smart_card_fails(self):
smart_card = authenticator(
auth_id="aut-smart-card",
key="smart_card_idp",
name="Smart Card IdP",
status="INACTIVE",
)
findings = _run_check(
build_authenticator_client(authenticators={smart_card.id: smart_card})
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "INACTIVE" in findings[0].status_extended
@@ -0,0 +1,87 @@
from unittest import mock
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
from tests.providers.okta.services.network_zone.network_zone_fixtures import (
build_network_zone_client,
network_zone,
)
CHECK_PATH = (
"prowler.providers.okta.services.network."
"network_zone_block_anonymized_proxies."
"network_zone_block_anonymized_proxies.network_zone_client"
)
def _run_check(network_zone_client):
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_okta_provider(),
),
mock.patch(CHECK_PATH, new=network_zone_client),
):
from prowler.providers.okta.services.network.network_zone_block_anonymized_proxies.network_zone_block_anonymized_proxies import (
network_zone_block_anonymized_proxies,
)
return network_zone_block_anonymized_proxies().execute()
class Test_network_zone_block_anonymized_proxies:
def test_missing_network_zone_scope_returns_manual(self):
findings = _run_check(
build_network_zone_client({}, missing_scopes=["okta.networkZones.read"])
)
assert len(findings) == 1
assert findings[0].status == "MANUAL"
assert "okta.networkZones.read" in findings[0].status_extended
def test_no_zones_fails(self):
findings = _run_check(build_network_zone_client({}))
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "No active Okta Network Zone blocklist" in findings[0].status_extended
def test_pass_with_active_ip_blocklist_gateway(self):
zone = network_zone(gateways=["198.51.100.10/32"])
findings = _run_check(build_network_zone_client({zone.id: zone}))
assert len(findings) == 1
assert findings[0].status == "PASS"
assert findings[0].resource_id == zone.id
assert "gateway" in findings[0].status_extended
def test_pass_with_active_enhanced_dynamic_anonymizer_blocklist(self):
zone = network_zone(
zone_id="nzo-enhanced",
name="DefaultEnhancedDynamicZone",
zone_type="DYNAMIC_V2",
system=True,
ip_service_categories=["ANONYMIZER"],
)
findings = _run_check(build_network_zone_client({zone.id: zone}))
assert len(findings) == 1
assert findings[0].status == "PASS"
assert "Enhanced Dynamic" in findings[0].status_extended
def test_existing_zones_without_anonymized_proxy_blocklist_fail(self):
policy_zone = network_zone(
zone_id="nzo-policy",
name="Corporate Policy Zone",
usage="POLICY",
gateways=["10.0.0.0/8"],
)
inactive_blocklist = network_zone(
zone_id="nzo-inactive",
name="Inactive Blocklist",
status="INACTIVE",
gateways=["203.0.113.0/24"],
)
findings = _run_check(
build_network_zone_client(
{policy_zone.id: policy_zone, inactive_blocklist.id: inactive_blocklist}
)
)
assert len(findings) == 1
assert findings[0].status == "FAIL"
assert "do not actively block" in findings[0].status_extended
@@ -0,0 +1,37 @@
from unittest import mock
from prowler.providers.okta.services.network.network_zone_service import OktaNetworkZone
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def build_network_zone_client(zones: dict = None, missing_scopes: list[str] = None):
client = mock.MagicMock()
client.network_zones = zones or {}
client.missing_scopes = missing_scopes or []
client.provider = set_mocked_okta_provider()
return client
def network_zone(
zone_id: str = "nzo-1",
name: str = "BlockedIpZone",
*,
status: str = "ACTIVE",
zone_type: str = "IP",
usage: str = "BLOCKLIST",
system: bool = False,
gateways: list[str] = None,
proxies: list[str] = None,
ip_service_categories: list[str] = None,
):
return OktaNetworkZone(
id=zone_id,
name=name,
status=status,
type=zone_type,
usage=usage,
system=system,
gateways=gateways or [],
proxies=proxies or [],
ip_service_categories=ip_service_categories or [],
)
@@ -0,0 +1,153 @@
from types import SimpleNamespace
from unittest import mock
import pytest
from prowler.providers.okta.services.network.network_zone_service import (
NETWORK_ZONES_READ_SCOPE,
NetworkZone,
OktaNetworkZone,
_next_after_cursor,
)
from tests.providers.okta.okta_fixtures import set_mocked_okta_provider
def _resp(headers: dict = None):
return SimpleNamespace(headers=headers or {})
def _sdk_zone(
zone_id: str,
name: str,
*,
status: str = "ACTIVE",
zone_type: str = "IP",
usage: str = "BLOCKLIST",
system: bool = False,
gateways: list[str] = None,
proxies: list[str] = None,
ip_service_categories: list[str] = None,
):
return SimpleNamespace(
id=zone_id,
name=name,
status=status,
type=zone_type,
usage=usage,
system=system,
gateways=gateways or [],
proxies=proxies or [],
ip_service_categories=ip_service_categories or [],
)
class Test_network_zone_pagination:
def test_no_link_header_returns_none(self):
assert _next_after_cursor(_resp({})) is None
def test_extracts_next_after_cursor(self):
link = (
'<https://acme.okta.com/api/v1/zones?limit=20>; rel="self", '
'<https://acme.okta.com/api/v1/zones?after=next-page>; rel="next"'
)
assert _next_after_cursor(_resp({"Link": link})) == "next-page"
class Test_NetworkZone_service:
def test_fetches_ip_and_enhanced_dynamic_zones(self):
provider = set_mocked_okta_provider()
ip_zone = _sdk_zone(
"nzo-ip",
"Blocked IPs",
gateways=["203.0.113.10/32"],
)
enhanced_zone = _sdk_zone(
"nzo-enhanced",
"DefaultEnhancedDynamicZone",
zone_type="DYNAMIC_V2",
system=True,
ip_service_categories=["ANONYMIZER"],
)
async def fake_list_network_zones(after=None, limit=None):
assert after is None
assert limit == 200
return ([ip_zone, enhanced_zone], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = NetworkZone(provider)
assert set(service.network_zones.keys()) == {"nzo-ip", "nzo-enhanced"}
assert isinstance(service.network_zones["nzo-ip"], OktaNetworkZone)
assert service.network_zones["nzo-ip"].gateways == ["203.0.113.10/32"]
assert service.network_zones["nzo-enhanced"].type == "DYNAMIC_V2"
assert service.network_zones["nzo-enhanced"].ip_service_categories == [
"ANONYMIZER"
]
def test_paginates_network_zones(self):
provider = set_mocked_okta_provider()
page_1 = _sdk_zone("nzo-1", "First")
page_2 = _sdk_zone("nzo-2", "Second")
next_link = '<https://acme.okta.com/api/v1/zones?after=cursor-2>; rel="next"'
calls = []
async def fake_list_network_zones(after=None, limit=None):
assert limit == 200
calls.append(after)
if after is None:
return ([page_1], _resp({"link": next_link}), None)
return ([page_2], _resp({}), None)
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_network_zones = fake_list_network_zones
mocked_client_cls.return_value = mocked
service = NetworkZone(provider)
assert calls == [None, "cursor-2"]
assert set(service.network_zones.keys()) == {"nzo-1", "nzo-2"}
def test_returns_empty_on_api_error(self):
provider = set_mocked_okta_provider()
async def failing(after=None, limit=None):
assert after is None
assert limit == 200
return ([], _resp({}), Exception("forbidden"))
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_network_zones = failing
mocked_client_cls.return_value = mocked
service = NetworkZone(provider)
assert service.network_zones == {}
@pytest.mark.parametrize("missing_scope", [NETWORK_ZONES_READ_SCOPE])
def test_missing_scope_skips_network_zones_sdk_call(self, missing_scope):
provider = set_mocked_okta_provider(scopes=[])
async def forbidden_list_network_zones(after=None, limit=None):
raise AssertionError("list_network_zones must not be called")
with mock.patch(
"prowler.providers.okta.lib.service.service.OktaSDKClient"
) as mocked_client_cls:
mocked = mock.MagicMock()
mocked.list_network_zones = forbidden_list_network_zones
mocked_client_cls.return_value = mocked
service = NetworkZone(provider)
assert service.missing_scopes == [missing_scope]
assert service.network_zones == {}

Some files were not shown because too many files have changed in this diff Show More