diff --git a/docs/deployment/authentication/okta-auth.mdx b/docs/deployment/authentication/okta-auth.mdx index 5b05d0cf47..112fdbaf50 100644 --- a/docs/deployment/authentication/okta-auth.mdx +++ b/docs/deployment/authentication/okta-auth.mdx @@ -26,6 +26,9 @@ Keep supports Okta as an authentication provider, enabling: | `OKTA_AUDIENCE` | Expected audience claim in the token. Falls back to `OKTA_CLIENT_ID` if not set | No | | `OKTA_JWKS_URL` | Explicit JWKS URL. If not set, derived from `OKTA_ISSUER` | No | | `OKTA_API_TOKEN` | Okta API token for management operations | No | +| `OKTA_ADMIN_GROUPS` | Comma-separated Okta group names that map to the `admin` role | No | +| `OKTA_NOC_GROUPS` | Comma-separated Okta group names that map to the `noc` role | No | +| `OKTA_WEBHOOK_GROUPS` | Comma-separated Okta group names that map to the `webhook` role | No | ### Frontend Environment Variables @@ -58,16 +61,38 @@ Keep supports Okta as an authentication provider, enabling: ### Role Mapping -Keep extracts the user role from the JWT token. The role is determined in the following order: +Keep resolves the user role from the JWT token using the following priority order: 1. `keep_role` claim in the token 2. `role` claim in the token -3. First entry in the `groups` claim -4. Falls back to `user` role +3. Okta group → Keep role mapping (via `OKTA_*_GROUPS` environment variables) +4. First entry in the `groups` claim (used as-is) +5. Falls back to `noc` role -To configure role mapping, add a custom claim to your Okta authorization server: +#### Option A — Explicit claim (recommended for simple setups) + +Add a custom claim to your Okta authorization server: + +1. Navigate to **Security** > **API** > **Authorization Servers** +2. Select your authorization server (e.g., `default`) +3. Go to the **Claims** tab +4. Add a claim named `keep_role` with the value `admin`, `noc`, or `webhook` + +#### Option B — Group-based mapping (recommended for team setups) + +Map Okta groups to Keep roles using environment variables. When multiple groups match, +the highest-privilege role wins (`admin` > `noc` > `webhook`). + +```bash +OKTA_ADMIN_GROUPS=keep-admins,platform-team +OKTA_NOC_GROUPS=keep-noc,ops-team +OKTA_WEBHOOK_GROUPS=keep-webhooks +``` + +You must configure Okta to include the `groups` claim in the JWT: 1. Navigate to **Security** > **API** > **Authorization Servers** 2. Select your authorization server (e.g., `default`) 3. Go to the **Claims** tab -4. Add a claim named `keep_role` or `groups` that maps to the user's Keep role +4. Add a claim named `groups` of type **Groups** with a filter matching the relevant groups +5. Set **Include in** to `Access Token` diff --git a/keep-ui/auth.config.ts b/keep-ui/auth.config.ts index 11aecf037a..0e1fe3b853 100644 --- a/keep-ui/auth.config.ts +++ b/keep-ui/auth.config.ts @@ -106,9 +106,14 @@ async function refreshAccessToken(token: any) { ); } + // For Okta, prefer id_token so the backend can read app-level claims (groups). + const newAccessToken = + authType === AuthType.OKTA + ? (refreshedTokens.id_token ?? refreshedTokens.access_token) + : refreshedTokens.access_token; return { ...token, - accessToken: refreshedTokens.access_token, + accessToken: newAccessToken, accessTokenExpires: Date.now() + (refreshedTokens.expires_in || 3600) * 1000, refreshToken: refreshedTokens.refresh_token ?? token.refreshToken, }; @@ -258,7 +263,7 @@ const baseProviderConfigs = { clientId: process.env.OKTA_CLIENT_ID!, clientSecret: process.env.OKTA_CLIENT_SECRET!, issuer: process.env.OKTA_ISSUER!, - authorization: { params: { scope: "openid email profile" } }, + authorization: { params: { scope: "openid email profile groups" } }, }), ], [AuthType.ONELOGIN]: [ @@ -362,10 +367,36 @@ export const config = { role = (profile as any).keep_role; accessToken = account.access_token; } else if (authType === AuthType.OKTA) { - // Extract tenant and role from Okta token tenantId = (profile as any).keep_tenant_id || "keep"; - role = (profile as any).keep_role || "user"; - accessToken = account.access_token; + // Use the ID token so the backend can see app-level claims (e.g. groups). + // App-level Okta group claims are embedded in the ID token only, not the + // access token or userinfo response. + accessToken = account.id_token ?? account.access_token; + // Explicit claim takes priority + role = (profile as any).keep_role || (profile as any).role; + // If no explicit claim, resolve from groups via OKTA_*_GROUPS mappings + if (!role) { + const groups: string[] = (profile as any).groups || []; + const groupMappings: Record = {}; + for (const [envVar, targetRole] of [ + ["OKTA_ADMIN_GROUPS", "admin"], + ["OKTA_NOC_GROUPS", "noc"], + ["OKTA_WEBHOOK_GROUPS", "webhook"], + ] as const) { + const val = runtimeEnv(envVar) || ""; + for (const g of val.split(",").map((s) => s.trim()).filter(Boolean)) { + groupMappings[g] = targetRole; + } + } + for (const priority of ["admin", "noc", "webhook"] as const) { + const match = groups.find((g) => groupMappings[g] === priority); + if (match) { + role = priority; + break; + } + } + } + role = role || "noc"; } else if (authType === AuthType.ONELOGIN) { // Extract tenant and role from OneLogin token - use ID token for user data tenantId = (profile as any).keep_tenant_id || "keep"; diff --git a/keep/identitymanager/identity_managers/okta/okta_authverifier.py b/keep/identitymanager/identity_managers/okta/okta_authverifier.py index b82710f853..a4ed14cad4 100644 --- a/keep/identitymanager/identity_managers/okta/okta_authverifier.py +++ b/keep/identitymanager/identity_managers/okta/okta_authverifier.py @@ -2,15 +2,25 @@ import os import jwt +import requests from fastapi import Depends, HTTPException +from keep.api.core.config import config +from keep.api.core.db import ( + create_user, + update_user_last_sign_in, + update_user_role, + user_exists, +) +from keep.api.core.dependencies import SINGLE_TENANT_UUID from keep.identitymanager.authenticatedentity import AuthenticatedEntity from keep.identitymanager.authverifierbase import AuthVerifierBase, oauth2_scheme logger = logging.getLogger(__name__) -# Define constant locally instead of importing it -DEFAULT_ROLE_NAME = "noc" # Default role name for user access +DEFAULT_ROLE_NAME = "noc" +ROLE_PRIORITY = ["admin", "noc", "webhook"] + class OktaAuthVerifier(AuthVerifierBase): """Handles authentication and authorization for Okta""" @@ -21,24 +31,65 @@ def __init__(self, scopes: list[str] = []) -> None: self.okta_audience = os.environ.get("OKTA_AUDIENCE") self.okta_client_id = os.environ.get("OKTA_CLIENT_ID") self.jwks_url = os.environ.get("OKTA_JWKS_URL") - + # If no explicit JWKS URL is provided, we need an issuer to construct it if not self.jwks_url and not self.okta_issuer: raise Exception("Missing both OKTA_JWKS_URL and OKTA_ISSUER environment variables") - + # Remove trailing slash if present on issuer if self.okta_issuer and self.okta_issuer.endswith("/"): self.okta_issuer = self.okta_issuer[:-1] - + # Initialize JWKS client - prefer direct JWKS URL if available if not self.jwks_url: self.jwks_url = f"{self.okta_issuer}/.well-known/jwks.json" - - # At this point, self.jwks_url is guaranteed to be a string + assert self.jwks_url is not None self.jwks_client = jwt.PyJWKClient(self.jwks_url) logger.info(f"Initialized JWKS client with URL: {self.jwks_url}") + # Userinfo endpoint for fetching group claims not present in the access token + self.userinfo_url = os.environ.get( + "OKTA_USERINFO_URL", + f"{self.okta_issuer}/v1/userinfo" if self.okta_issuer else None, + ) + + # Build group → Keep role mapping from environment variables. + # OKTA_ADMIN_GROUPS, OKTA_NOC_GROUPS, OKTA_WEBHOOK_GROUPS accept + # comma-separated Okta group names. + self.group_mappings: dict[str, str] = {} + for env_var, target_role in [ + ("OKTA_ADMIN_GROUPS", "admin"), + ("OKTA_NOC_GROUPS", "noc"), + ("OKTA_WEBHOOK_GROUPS", "webhook"), + ]: + groups_str = config(env_var, default="") + for group in [g.strip() for g in groups_str.split(",") if g.strip()]: + self.group_mappings[group] = target_role + if self.group_mappings: + logger.info(f"Okta group mappings loaded: {self.group_mappings}") + + self.auto_create_user = config("OKTA_AUTO_CREATE_USER", default=True, cast=bool) + + def _get_userinfo(self, token: str) -> dict: + """Call Okta's userinfo endpoint to retrieve claims not present in the access token (e.g. groups).""" + if not self.userinfo_url: + return {} + try: + resp = requests.get( + self.userinfo_url, + headers={"Authorization": f"Bearer {token}"}, + timeout=5, + ) + if resp.status_code == 200: + data = resp.json() + logger.debug(f"Userinfo response keys: {list(data.keys())}") + return data + logger.warning(f"Userinfo endpoint returned {resp.status_code}") + except Exception: + logger.exception("Failed to call userinfo endpoint") + return {} + def _verify_bearer_token(self, token: str = Depends(oauth2_scheme)) -> AuthenticatedEntity: if not token: raise HTTPException(status_code=401, detail="No token provided") @@ -47,43 +98,97 @@ def _verify_bearer_token(self, token: str = Depends(oauth2_scheme)) -> Authentic # Get the signing key directly from the JWT signing_key = self.jwks_client.get_signing_key_from_jwt(token).key + # Build accepted audiences: always include client_id (ID token aud) + # and optionally the configured OKTA_AUDIENCE (access token aud). + audiences = [self.okta_client_id] if self.okta_client_id else [] + if self.okta_audience and self.okta_audience != self.okta_client_id: + audiences.append(self.okta_audience) # Decode and verify the token payload = jwt.decode( token, key=signing_key, algorithms=["RS256"], - audience=self.okta_audience or self.okta_client_id, + audience=audiences or None, issuer=self.okta_issuer, options={"verify_exp": True} ) - # Extract user info from token with simplified role handling - tenant_id = payload.get("keep_tenant_id", "keep") # Default to 'keep' if not specified - email = payload.get("email") or payload.get("sub") or payload.get("preferred_username") - - # Look for role in standard locations with a default of "user" - groups = payload.get("groups", []) - role_name = ( - payload.get("keep_role") or - payload.get("role") or - (groups[0] if groups else None) or - DEFAULT_ROLE_NAME # Use constant for consistency + # Only call userinfo when the groups claim is absent from the token. + # When the frontend sends the Okta ID token (which carries app-level + # group claims directly), calling userinfo with it would return 401 + # because the userinfo endpoint expects an access token, not an ID token. + if "groups" not in payload: + userinfo = self._get_userinfo(token) + else: + userinfo = {} + + tenant_id = payload.get("keep_tenant_id", "keep") + email = ( + payload.get("email") or userinfo.get("email") + or payload.get("sub") or payload.get("preferred_username") ) - + name = ( + userinfo.get("name") or userinfo.get("displayName") + or payload.get("name") or payload.get("displayName") + or email + ) + groups = userinfo.get("groups", []) or payload.get("groups", []) + + logger.debug(f"Token claims — email={email}, groups={groups}, group_mappings={self.group_mappings}") + + # Explicit claim overrides always take priority + role_name = payload.get("keep_role") or payload.get("role") + + # If no explicit claim, try to resolve role from Okta groups via + # the configured OKTA_*_GROUPS mappings (highest privilege wins). + if not role_name and self.group_mappings and groups: + for priority_role in ROLE_PRIORITY: + for group in groups: + if self.group_mappings.get(group) == priority_role: + role_name = priority_role + logger.info(f"Resolved role '{role_name}' from Okta group '{group}'") + break + if role_name: + break + + # Final fallback: use first group as-is only when no mappings are + # configured (raw-group mode), otherwise fall back to the default role. + if not role_name: + if not self.group_mappings and groups: + role_name = groups[0] + else: + role_name = DEFAULT_ROLE_NAME + + logger.debug(f"Resolved role='{role_name}' for {email}") + org_id = payload.get("org_id") org_realm = payload.get("org_realm") - + if not email: raise HTTPException(status_code=401, detail="No email in token") - - logger.info(f"Successfully verified token for user with email: {email}") + + tenant_id_for_db = SINGLE_TENANT_UUID + exists = user_exists(tenant_id=tenant_id_for_db, username=email) + if self.auto_create_user and not exists: + logger.info(f"Auto-provisioning Okta user: {email} with role={role_name}") + try: + create_user(tenant_id=tenant_id_for_db, username=email, password="", role=role_name) + except Exception: + logger.exception(f"Failed to auto-create user {email}") + elif exists: + try: + update_user_last_sign_in(tenant_id=tenant_id_for_db, username=email) + update_user_role(tenant_id=tenant_id_for_db, username=email, role=role_name) + except Exception: + logger.exception(f"Failed to update user {email}") return AuthenticatedEntity( tenant_id=tenant_id, email=email, role=role_name, org_id=org_id, org_realm=org_realm, - token=token + token=token, + name=name, ) except jwt.exceptions.InvalidKeyError as e: diff --git a/keep/identitymanager/identity_managers/okta/okta_identitymanager.py b/keep/identitymanager/identity_managers/okta/okta_identitymanager.py index 31a9f70725..3cf9a5e9cf 100644 --- a/keep/identitymanager/identity_managers/okta/okta_identitymanager.py +++ b/keep/identitymanager/identity_managers/okta/okta_identitymanager.py @@ -1,6 +1,6 @@ import os - +from keep.api.core.db import get_users as get_users_from_db from keep.api.models.user import Group, Role, User from keep.contextmanager.contextmanager import ContextManager from keep.identitymanager.authenticatedentity import AuthenticatedEntity @@ -67,9 +67,18 @@ def get_sso_wizard_url(self, authenticated_entity: AuthenticatedEntity) -> str: return f"{self.okta_issuer}/sso/{tenant_id}" def get_users(self) -> list[User]: - """Get all users from Okta - disabled""" - self.logger.info("get_users called but management functions are disabled") - return [] + """Get users from Keep's local DB (Okta user management is not used)""" + db_users = get_users_from_db() + return [ + User( + email=user.username, + name=user.username, + role=user.role, + last_login=str(user.last_sign_in) if user.last_sign_in else None, + created_at=str(user.created_at), + ) + for user in db_users + ] def create_user(self, user_email: str, user_name: str, password: str, role: str, groups: list[str] = []) -> dict: """Create a new user in Okta - disabled""" diff --git a/tests/test_okta_auth.py b/tests/test_okta_auth.py new file mode 100644 index 0000000000..ff973d2215 --- /dev/null +++ b/tests/test_okta_auth.py @@ -0,0 +1,361 @@ +"""Tests for Okta auth verifier: group-to-role mapping, userinfo enrichment, and user auto-provisioning.""" +import importlib +import sys +from unittest.mock import MagicMock, patch + +import pytest + +MOCK_TOKEN = "mock.okta.token" + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +OKTA_ENV = { + "OKTA_ISSUER": "https://okta.example.com", + "OKTA_AUDIENCE": "api://default", + "OKTA_CLIENT_ID": "client123", + "OKTA_CLIENT_SECRET": "secret", + "OKTA_DOMAIN": "okta.example.com", +} + + +def _make_verifier(extra_env: dict | None = None, monkeypatch=None): + """Instantiate OktaAuthVerifier with mocked JWKS client.""" + env = {**OKTA_ENV, **(extra_env or {})} + for k, v in env.items(): + monkeypatch.setenv(k, v) + + # Remove cached modules so env vars are picked up fresh + for mod in list(sys.modules.keys()): + if "okta" in mod.lower() or "keep.api.core.config" in mod: + sys.modules.pop(mod, None) + + from keep.identitymanager.identity_managers.okta.okta_authverifier import OktaAuthVerifier + + verifier = OktaAuthVerifier.__new__(OktaAuthVerifier) + verifier.scopes = [] + + mock_jwks = MagicMock() + mock_signing_key = MagicMock() + mock_signing_key.key = "mock_key" + mock_jwks.get_signing_key_from_jwt.return_value = mock_signing_key + + with patch("jwt.PyJWKClient", return_value=mock_jwks): + verifier.__init__() + + return verifier + + +# --------------------------------------------------------------------------- +# _get_userinfo +# --------------------------------------------------------------------------- + + +def test_get_userinfo_returns_claims(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + verifier.userinfo_url = "https://okta.example.com/v1/userinfo" + + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = { + "email": "user@example.com", + "name": "John Doe", + "groups": ["Keep_Admin", "Everyone"], + } + + with patch("requests.get", return_value=mock_resp) as mock_get: + result = verifier._get_userinfo(MOCK_TOKEN) + + mock_get.assert_called_once_with( + "https://okta.example.com/v1/userinfo", + headers={"Authorization": f"Bearer {MOCK_TOKEN}"}, + timeout=5, + ) + assert result["name"] == "John Doe" + assert result["groups"] == ["Keep_Admin", "Everyone"] + + +def test_get_userinfo_returns_empty_on_error(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + verifier.userinfo_url = "https://okta.example.com/v1/userinfo" + + mock_resp = MagicMock() + mock_resp.status_code = 401 + + with patch("requests.get", return_value=mock_resp): + result = verifier._get_userinfo(MOCK_TOKEN) + + assert result == {} + + +def test_get_userinfo_returns_empty_when_no_url(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + verifier.userinfo_url = None + + with patch("requests.get") as mock_get: + result = verifier._get_userinfo(MOCK_TOKEN) + + mock_get.assert_not_called() + assert result == {} + + +def test_get_userinfo_returns_empty_on_exception(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + verifier.userinfo_url = "https://okta.example.com/v1/userinfo" + + with patch("requests.get", side_effect=ConnectionError("timeout")): + result = verifier._get_userinfo(MOCK_TOKEN) + + assert result == {} + + +# --------------------------------------------------------------------------- +# Group mappings loading +# --------------------------------------------------------------------------- + + +def test_group_mappings_loaded_from_env(monkeypatch): + verifier = _make_verifier( + extra_env={ + "OKTA_ADMIN_GROUPS": "Keep_Admin,Ops_Admin", + "OKTA_NOC_GROUPS": "Keep_User", + "OKTA_WEBHOOK_GROUPS": "Keep_Webhook", + }, + monkeypatch=monkeypatch, + ) + assert verifier.group_mappings["Keep_Admin"] == "admin" + assert verifier.group_mappings["Ops_Admin"] == "admin" + assert verifier.group_mappings["Keep_User"] == "noc" + assert verifier.group_mappings["Keep_Webhook"] == "webhook" + + +def test_group_mappings_empty_when_not_configured(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + assert verifier.group_mappings == {} + + +# --------------------------------------------------------------------------- +# Role resolution from groups +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "groups, expected_role", + [ + (["Keep_Admin", "Keep_User"], "admin"), # admin wins over noc + (["Keep_User"], "noc"), + (["Keep_Webhook"], "webhook"), + (["Keep_User", "Keep_Webhook"], "noc"), # noc wins over webhook + ([], "noc"), # default role + (["Unknown_Group"], "noc"), # unmapped group → default + ], +) +def test_role_resolution_priority(monkeypatch, groups, expected_role): + verifier = _make_verifier( + extra_env={ + "OKTA_ADMIN_GROUPS": "Keep_Admin", + "OKTA_NOC_GROUPS": "Keep_User", + "OKTA_WEBHOOK_GROUPS": "Keep_Webhook", + }, + monkeypatch=monkeypatch, + ) + + jwt_payload = { + "sub": "user@example.com", + "email": "user@example.com", + } + userinfo = {"email": "user@example.com", "name": "Test User", "groups": groups} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=True), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_last_sign_in"), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_role"), + ): + mock_key.return_value.key = "mock_key" + entity = verifier._verify_bearer_token(MOCK_TOKEN) + + assert entity.role == expected_role + + +def test_explicit_keep_role_claim_overrides_groups(monkeypatch): + """keep_role in JWT always takes priority over group mapping.""" + verifier = _make_verifier( + extra_env={"OKTA_ADMIN_GROUPS": "Keep_Admin", "OKTA_NOC_GROUPS": "Keep_User"}, + monkeypatch=monkeypatch, + ) + + jwt_payload = { + "sub": "user@example.com", + "email": "user@example.com", + "keep_role": "webhook", # explicit override + } + userinfo = {"email": "user@example.com", "groups": ["Keep_Admin"]} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=True), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_last_sign_in"), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_role"), + ): + mock_key.return_value.key = "mock_key" + entity = verifier._verify_bearer_token(MOCK_TOKEN) + + assert entity.role == "webhook" + + +# --------------------------------------------------------------------------- +# User auto-provisioning +# --------------------------------------------------------------------------- + + +def test_user_created_on_first_login(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + + jwt_payload = {"sub": "newuser@example.com", "email": "newuser@example.com"} + userinfo = {"email": "newuser@example.com", "name": "New User", "groups": []} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=False) as mock_exists, + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.create_user") as mock_create, + ): + mock_key.return_value.key = "mock_key" + verifier._verify_bearer_token(MOCK_TOKEN) + + mock_exists.assert_called_once() + mock_create.assert_called_once_with( + tenant_id="keep", + username="newuser@example.com", + password="", + role="noc", + ) + + +def test_user_updated_on_subsequent_login(monkeypatch): + verifier = _make_verifier( + extra_env={"OKTA_ADMIN_GROUPS": "Keep_Admin"}, + monkeypatch=monkeypatch, + ) + + jwt_payload = {"sub": "existing@example.com", "email": "existing@example.com"} + userinfo = {"email": "existing@example.com", "name": "Existing User", "groups": ["Keep_Admin"]} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=True), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.create_user") as mock_create, + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_last_sign_in") as mock_last_sign_in, + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_role") as mock_update_role, + ): + mock_key.return_value.key = "mock_key" + verifier._verify_bearer_token(MOCK_TOKEN) + + mock_create.assert_not_called() + mock_last_sign_in.assert_called_once() + mock_update_role.assert_called_once_with(tenant_id="keep", username="existing@example.com", role="admin") + + +def test_auto_create_disabled(monkeypatch): + verifier = _make_verifier( + extra_env={"OKTA_AUTO_CREATE_USER": "false"}, + monkeypatch=monkeypatch, + ) + + jwt_payload = {"sub": "newuser@example.com", "email": "newuser@example.com"} + userinfo = {"email": "newuser@example.com", "groups": []} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=False), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.create_user") as mock_create, + ): + mock_key.return_value.key = "mock_key" + verifier._verify_bearer_token(MOCK_TOKEN) + + mock_create.assert_not_called() + + +# --------------------------------------------------------------------------- +# Name / email extraction +# --------------------------------------------------------------------------- + + +def test_name_comes_from_userinfo(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + + jwt_payload = {"sub": "user@example.com", "email": "user@example.com"} + userinfo = {"email": "user@example.com", "name": "Jane Doe", "groups": []} + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=True), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_last_sign_in"), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_role"), + ): + mock_key.return_value.key = "mock_key" + entity = verifier._verify_bearer_token(MOCK_TOKEN) + + assert entity.name == "Jane Doe" + + +def test_name_falls_back_to_email_when_absent(monkeypatch): + verifier = _make_verifier(monkeypatch=monkeypatch) + + jwt_payload = {"sub": "user@example.com", "email": "user@example.com"} + userinfo = {"email": "user@example.com", "groups": []} # no name + + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", return_value=jwt_payload), + patch.object(verifier, "_get_userinfo", return_value=userinfo), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.user_exists", return_value=True), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_last_sign_in"), + patch("keep.identitymanager.identity_managers.okta.okta_authverifier.update_user_role"), + ): + mock_key.return_value.key = "mock_key" + entity = verifier._verify_bearer_token(MOCK_TOKEN) + + assert entity.name == "user@example.com" + + +# --------------------------------------------------------------------------- +# Token errors +# --------------------------------------------------------------------------- + + +def test_missing_token_raises_401(monkeypatch): + from fastapi import HTTPException + + verifier = _make_verifier(monkeypatch=monkeypatch) + with pytest.raises(HTTPException) as exc_info: + verifier._verify_bearer_token("") + assert exc_info.value.status_code == 401 + + +def test_expired_token_raises_401(monkeypatch): + import jwt as pyjwt + from fastapi import HTTPException + + verifier = _make_verifier(monkeypatch=monkeypatch) + with ( + patch.object(verifier.jwks_client, "get_signing_key_from_jwt") as mock_key, + patch("jwt.decode", side_effect=pyjwt.ExpiredSignatureError), + ): + mock_key.return_value.key = "mock_key" + with pytest.raises(HTTPException) as exc_info: + verifier._verify_bearer_token(MOCK_TOKEN) + assert exc_info.value.status_code == 401