From 5cc358d09ae3c131e4f5ea03062ae60f14ede9c6 Mon Sep 17 00:00:00 2001 From: samuel-sirven-bib Date: Mon, 6 Jul 2026 10:08:08 +0200 Subject: [PATCH 1/2] refactor(auth): extract OIDC callback steps into helper functions --- timesketch/views/auth.py | 60 +++++++++++++++++++++++++++++----------- 1 file changed, 44 insertions(+), 16 deletions(-) diff --git a/timesketch/views/auth.py b/timesketch/views/auth.py index c6b081c068..a7aef2b99b 100644 --- a/timesketch/views/auth.py +++ b/timesketch/views/auth.py @@ -442,18 +442,13 @@ def validate_api_token(): return abort(HTTP_STATUS_CODE_BAD_REQUEST, "User is not authenticated.") -@auth_views.route("/login/google_openid_connect/", methods=["GET"]) -def google_openid_connect(): - """Handler for the Google OpenID Connect callback. - - Reference: - https://developers.google.com/identity/protocols/OpenIDConnect +def _get_oidc_csrf_validated_code(): + """Validate the OIDC callback request and return the authorization code. Returns: - Redirect response. + The `code` query parameter from the OIDC callback request. """ error = request.args.get("error", None) - if error: current_app.logger.error(f"OAuth2 flow error: {error}") return abort(HTTP_STATUS_CODE_BAD_REQUEST, f"OAuth2 flow error: {error!s}") @@ -470,6 +465,18 @@ def google_openid_connect(): if client_csrf_token != server_csrf_token: return abort(HTTP_STATUS_CODE_BAD_REQUEST, "Invalid CSRF token") + return code + + +def _fetch_validated_oidc_jwt(code): + """Fetch and validate the OIDC JWT for an authorization code. + + Args: + code: Authorization code returned by the OIDC provider. + + Returns: + The decoded JWT as a dict. + """ try: encoded_jwt = get_encoded_jwt_over_https(code) except JwtFetchError as e: @@ -490,7 +497,9 @@ def google_openid_connect(): # Fetch the public key and try to validate the JWT. try: - public_key = get_public_key_for_jwt(encoded_jwt, discovery_document["jwks_uri"]) + public_key = get_public_key_for_jwt( + encoded_jwt, discovery_document["jwks_uri"] + ) decoded_jwt = decode_jwt(encoded_jwt, public_key, algorithm, expected_audience) validate_jwt(decoded_jwt, expected_issuer, expected_domain) except (JwtValidationError, JwtKeyError) as e: @@ -500,15 +509,34 @@ def google_openid_connect(): f"Unable to validate request, with error: {e!s}", ) - validated_email = decoded_jwt.get("email") + return decoded_jwt + + +def _check_oidc_user_allowed(validated_email): + """Abort if GOOGLE_OIDC_ALLOWED_USERS is set and the user isn't in it.""" allowed_users = current_app.config.get("GOOGLE_OIDC_ALLOWED_USERS") + if allowed_users and validated_email not in allowed_users: + return abort( + HTTP_STATUS_CODE_UNAUTHORIZED, "Unauthorized request, user not allowed" + ) + return None - # Check if the authenticating user is allowed. - if allowed_users: - if validated_email not in allowed_users: - return abort( - HTTP_STATUS_CODE_UNAUTHORIZED, "Unauthorized request, user not allowed" - ) + +@auth_views.route("/login/google_openid_connect/", methods=["GET"]) +def google_openid_connect(): + """Handler for the Google OpenID Connect callback. + + Reference: + https://developers.google.com/identity/protocols/OpenIDConnect + + Returns: + Redirect response. + """ + code = _get_oidc_csrf_validated_code() + decoded_jwt = _fetch_validated_oidc_jwt(code) + + validated_email = decoded_jwt.get("email") + _check_oidc_user_allowed(validated_email) user = User.get_or_create(username=validated_email, name=validated_email) login_user(user) From e9ef6f974339790510b9c2546f9d22a47a446fb2 Mon Sep 17 00:00:00 2001 From: samuel-sirven-bib Date: Mon, 6 Jul 2026 10:09:32 +0200 Subject: [PATCH 2/2] feat(auth): add OIDC group/role based access control and sync --- data/timesketch.conf | 31 +++++ timesketch/lib/google_auth.py | 80 +++++++++++- timesketch/lib/google_auth_test.py | 84 +++++++++++++ timesketch/views/auth.py | 135 ++++++++++++++++++++- timesketch/views/auth_test.py | 188 +++++++++++++++++++++++++++++ 5 files changed, 512 insertions(+), 6 deletions(-) diff --git a/data/timesketch.conf b/data/timesketch.conf index 20aba88730..9327d52faa 100644 --- a/data/timesketch.conf +++ b/data/timesketch.conf @@ -159,6 +159,37 @@ GOOGLE_OIDC_HOSTED_DOMAIN = None # Additional Google GSuite domains allowed API access. GOOGLE_OIDC_API_ALLOWED_DOMAINS = [] +# Name of the claim in the token containing groups/roles (e.g. "groups", "roles"). +# Leave as None to disable all group/role based features. +GOOGLE_OIDC_GROUPS_CLAIM = None + +# Separator to split the groups claim if returned as a delimited string. +GOOGLE_OIDC_GROUPS_SEPARATOR = None + +# OAuth2 scopes to request. Defaults to openid email profile. +# Add additional scopes as needed for your identity provider. +GOOGLE_OIDC_SCOPES = ["openid", "email", "profile"] + +# Regex to extract group names from raw values. Use the first capture group if +# present, otherwise the whole match. +GOOGLE_OIDC_GROUPS_REGEX = None + +# List of groups/roles allowed to log in. Leave empty to allow all authenticated users. +GOOGLE_OIDC_ALLOWED_GROUPS = [] + +# A group /role that should be granted admin privileges. +# Users in this group will be created as Timesketch administrators. +GOOGLE_OIDC_ADMIN_GROUP = "" + +# Auto-create Timesketch groups and add user as member for each group in +# token. Also grants admin privileges to members of GOOGLE_OIDC_ADMIN_GROUP. +GOOGLE_OIDC_GROUPS_SYNC_ENABLED = False + +# Remove user from Timesketch groups no longer present in the token. Also +# revokes admin privileges if the user is no longer a member of the admin group in +# GOOGLE_OIDC_ADMIN_GROUP. +GOOGLE_OIDC_GROUPS_REMOVE_STALE = False + # If populated only these users (email addresses) will be able to login to # this server. This can be used when access should be limited to a specific # set of users. diff --git a/timesketch/lib/google_auth.py b/timesketch/lib/google_auth.py index 589a12b4c7..8717a835ec 100644 --- a/timesketch/lib/google_auth.py +++ b/timesketch/lib/google_auth.py @@ -20,6 +20,8 @@ import json import hashlib import os +import re +from typing import Optional # six.moves is a dynamically-created namespace that doesn't actually # exist and therefore pylint can't statically analyze it. @@ -127,7 +129,9 @@ def get_oauth2_authorize_url(hosted_domain: str = ""): redirect_uri = url_for( "user_views.google_openid_connect", _scheme="https", _external=True ) - scopes = ("openid", "email", "profile") + scopes = current_app.config.get( + "GOOGLE_OIDC_SCOPES", ["openid", "email", "profile"] + ) # Add the generated CSRF token to the client session for later validation. session[CSRF_KEY] = csrf_token @@ -280,6 +284,80 @@ def validate_jwt(decoded_jwt: str, expected_issuer: str, expected_domain: str = raise JwtValidationError(f"Missing domain: {e}") from e +def _extract_group_name(group_name: str, regex_pattern: Optional[str]): + """Apply an optional regex to a single raw group name. + + Args: + group_name: Raw, already-stripped group name. + regex_pattern: Optional regex to extract the group name. The first + capture group is used if present, otherwise the whole match. + + Returns: + The extracted group name, or None if it doesn't match/is empty. + """ + if not regex_pattern: + return group_name + + match = re.search(regex_pattern, group_name) + if not match: + return None + + extracted = ( + match.group(1) + if (match.groups() and match.group(1) is not None) + else match.group(0) + ) + return extracted.strip() or None + + +def _split_raw_groups(raw_value, separator: Optional[str]): + """Normalize a raw claim value into a list of raw group values.""" + if isinstance(raw_value, str): + return raw_value.split(separator) if separator else [raw_value] + if isinstance(raw_value, (list, tuple, set)): + return list(raw_value) + return [raw_value] + + +def get_groups_from_jwt( + decoded_jwt: dict, + claim_name: Optional[str] = None, + separator: Optional[str] = None, + regex_pattern: Optional[str] = None, +): + """Extract and normalize group/role names from a decoded token claim. + + Args: + decoded_jwt: Decoded token claims dict. + claim_name: Claim name holding groups/roles (e.g. "groups", "roles"). + None returns empty list. + separator: Optional separator for string claims (e.g. semicolon). + regex_pattern: Optional regex to extract group names. First capture + group is used if present. + + Returns: + List of extracted group/role names; empty list if claim missing/empty. + """ + if not claim_name: + return [] + + raw_value = decoded_jwt.get(claim_name) + if not raw_value: + return [] + + groups = [] + for raw_group in _split_raw_groups(raw_value, separator): + group_name = str(raw_group).strip() + if not group_name: + continue + + group_name = _extract_group_name(group_name, regex_pattern) + if group_name: + groups.append(group_name) + + return groups + + def get_public_key_for_jwt(encoded_jwt: str, url: str): """Get public key for JWT in order to verify the signature. diff --git a/timesketch/lib/google_auth_test.py b/timesketch/lib/google_auth_test.py index e31def7c78..33e232b143 100644 --- a/timesketch/lib/google_auth_test.py +++ b/timesketch/lib/google_auth_test.py @@ -23,6 +23,7 @@ from timesketch.lib.google_auth import decode_jwt from timesketch.lib.google_auth import validate_jwt from timesketch.lib.google_auth import get_public_key_for_jwt +from timesketch.lib.google_auth import get_groups_from_jwt from timesketch.lib.google_auth import JwtValidationError from timesketch.lib.google_auth import JwtKeyError @@ -431,3 +432,86 @@ def test_valid_oidc_jwt(self): self.assertIsInstance(test_decoded_jwt, dict) self.assertEqual(test_decoded_jwt.get("email"), "test@example.com") + + +class TestGetGroupsFromJwt(BaseTest): + """Tests for the google_auth.get_groups_from_jwt function.""" + + def test_get_groups_from_jwt(self): + """Test extracting groups from a decoded JWT for various configs.""" + cases = [ + ( + "no claim name configured returns an empty list", + {"groups": ["Admins"]}, + None, + None, + None, + [], + ), + ( + "missing claim in the token returns an empty list", + {"email": "test@example.com"}, + "groups", + None, + None, + [], + ), + ( + "empty claim value returns an empty list", + {"groups": []}, + "groups", + None, + None, + [], + ), + ( + "list claim is stripped and empty entries are dropped", + {"groups": ["Admins", " Users ", ""]}, + "groups", + None, + None, + ["Admins", "Users"], + ), + ( + "string claim is split using the configured separator", + {"roles": "a;b; c ;"}, + "roles", + ";", + None, + ["a", "b", "c"], + ), + ( + "regex extracts the group name from each raw value", + {"groups": "CN=Admins,OU=X;CN=Users,OU=Y"}, + "groups", + ";", + r"CN=([^,]+)", + ["Admins", "Users"], + ), + ( + "values that don't match the regex are dropped", + {"groups": "no-match-here"}, + "groups", + None, + r"CN=([^,]+)", + [], + ), + ] + for ( + description, + decoded_jwt, + claim_name, + separator, + regex_pattern, + expected, + ) in cases: + with self.subTest(description): + self.assertEqual( + get_groups_from_jwt( + decoded_jwt, + claim_name=claim_name, + separator=separator, + regex_pattern=regex_pattern, + ), + expected, + ) diff --git a/timesketch/views/auth.py b/timesketch/views/auth.py index a7aef2b99b..441b842316 100644 --- a/timesketch/views/auth.py +++ b/timesketch/views/auth.py @@ -42,6 +42,7 @@ from timesketch.lib.google_auth import get_oauth2_discovery_document from timesketch.lib.google_auth import get_oauth2_authorize_url from timesketch.lib.google_auth import get_encoded_jwt_over_https +from timesketch.lib.google_auth import get_groups_from_jwt from timesketch.lib.google_auth import decode_jwt from timesketch.lib.google_auth import validate_jwt from timesketch.lib.google_auth import JwtValidationError @@ -450,7 +451,7 @@ def _get_oidc_csrf_validated_code(): """ error = request.args.get("error", None) if error: - current_app.logger.error(f"OAuth2 flow error: {error}") + current_app.logger.error("OAuth2 flow error: %s", error) return abort(HTTP_STATUS_CODE_BAD_REQUEST, f"OAuth2 flow error: {error!s}") try: @@ -472,7 +473,7 @@ def _fetch_validated_oidc_jwt(code): """Fetch and validate the OIDC JWT for an authorization code. Args: - code: Authorization code returned by the OIDC provider. + code (str): Authorization code returned by the OIDC provider. Returns: The decoded JWT as a dict. @@ -497,9 +498,7 @@ def _fetch_validated_oidc_jwt(code): # Fetch the public key and try to validate the JWT. try: - public_key = get_public_key_for_jwt( - encoded_jwt, discovery_document["jwks_uri"] - ) + public_key = get_public_key_for_jwt(encoded_jwt, discovery_document["jwks_uri"]) decoded_jwt = decode_jwt(encoded_jwt, public_key, algorithm, expected_audience) validate_jwt(decoded_jwt, expected_issuer, expected_domain) except (JwtValidationError, JwtKeyError) as e: @@ -522,6 +521,122 @@ def _check_oidc_user_allowed(validated_email): return None +def _get_oidc_token_groups(decoded_jwt): + """Extract the configured groups claim and the token's group names. + + Returns: + Tuple of (groups_claim, token_groups). + """ + groups_claim = current_app.config.get("GOOGLE_OIDC_GROUPS_CLAIM") + groups_regex = current_app.config.get("GOOGLE_OIDC_GROUPS_REGEX") + token_groups = get_groups_from_jwt( + decoded_jwt, + claim_name=groups_claim, + separator=current_app.config.get("GOOGLE_OIDC_GROUPS_SEPARATOR"), + regex_pattern=rf"{groups_regex}" if groups_regex else None, + ) + return groups_claim, token_groups + + +def _enforce_oidc_allowed_groups(groups_claim, token_groups, validated_email): + """Abort if GOOGLE_OIDC_ALLOWED_GROUPS is set and the user isn't a member. + + Returns: + The configured allowed groups list, or None if not configured. + """ + allowed_groups = current_app.config.get("GOOGLE_OIDC_ALLOWED_GROUPS") + if not allowed_groups: + return None + + if not groups_claim: + current_app.logger.warning( + "GOOGLE_OIDC_ALLOWED_GROUPS is set but GOOGLE_OIDC_GROUPS_CLAIM " + "is not configured. Access control will reject all users." + ) + + if not set(token_groups) & set(allowed_groups): + current_app.logger.warning( + "Unauthorized OIDC login attempt for user [%s]: not a member " + "of any allowed group %s (token groups: %s)", + validated_email, + allowed_groups, + token_groups, + ) + return abort( + HTTP_STATUS_CODE_UNAUTHORIZED, + "Unauthorized request, user is not a member of an allowed group", + ) + + return allowed_groups + + +def _resolve_oidc_groups_to_sync(token_groups, allowed_groups): + """Filter token groups down to those that should be synced to Timesketch. + + If GOOGLE_OIDC_ALLOWED_GROUPS is configured, only groups also present in + that allowlist are created/kept in Timesketch. + """ + if not allowed_groups: + return token_groups + return [group for group in token_groups if group in allowed_groups] + + +def _sync_user_groups(user, groups_to_sync): + """Add the user to any Timesketch groups in `groups_to_sync`.""" + for group_name in groups_to_sync: + group = Group.get_or_create(name=group_name, display_name=group_name) + if group not in user.groups: + user.groups.append(group) + + +def _remove_stale_user_groups(user, groups_to_sync): + """Remove the user from Timesketch groups no longer in `groups_to_sync`.""" + for group in list(user.groups): + if group.name not in groups_to_sync: + user.groups.remove(group) + + +def _sync_admin_privileges(user, is_admin_group_member, remove_stale, validated_email): + """Grant/revoke admin privileges based on OIDC admin group membership.""" + if is_admin_group_member and not user.admin: + user.admin = True + current_app.logger.info( + "User [%s] granted admin privileges via OIDC admin group membership", + validated_email, + ) + elif remove_stale and user.admin and not is_admin_group_member: + user.admin = False + current_app.logger.info( + "Admin privileges removed from user [%s] (no longer member of " + "admin group)", + validated_email, + ) + + +def _sync_oidc_groups( + user, groups_claim, token_groups, allowed_groups, validated_email +): + """Sync Timesketch group membership and admin privileges from the token. + + No-op unless a groups claim is configured and + GOOGLE_OIDC_GROUPS_SYNC_ENABLED is set. + """ + if not (groups_claim and current_app.config.get("GOOGLE_OIDC_GROUPS_SYNC_ENABLED")): + return + + groups_to_sync = _resolve_oidc_groups_to_sync(token_groups, allowed_groups) + _sync_user_groups(user, groups_to_sync) + + admin_group = current_app.config.get("GOOGLE_OIDC_ADMIN_GROUP", "") + is_admin_group_member = admin_group in token_groups + remove_stale = bool(current_app.config.get("GOOGLE_OIDC_GROUPS_REMOVE_STALE")) + + _sync_admin_privileges(user, is_admin_group_member, remove_stale, validated_email) + + if remove_stale: + _remove_stale_user_groups(user, groups_to_sync) + + @auth_views.route("/login/google_openid_connect/", methods=["GET"]) def google_openid_connect(): """Handler for the Google OpenID Connect callback. @@ -538,9 +653,19 @@ def google_openid_connect(): validated_email = decoded_jwt.get("email") _check_oidc_user_allowed(validated_email) + groups_claim, token_groups = _get_oidc_token_groups(decoded_jwt) + allowed_groups = _enforce_oidc_allowed_groups( + groups_claim, token_groups, validated_email + ) + user = User.get_or_create(username=validated_email, name=validated_email) login_user(user) + _sync_oidc_groups(user, groups_claim, token_groups, allowed_groups, validated_email) + + db_session.add(user) + db_session.commit() + # Log the user in and setup the session. if current_user.is_authenticated: next_url = session.get("next", "/") diff --git a/timesketch/views/auth_test.py b/timesketch/views/auth_test.py index c0358de585..caf06b7fc8 100644 --- a/timesketch/views/auth_test.py +++ b/timesketch/views/auth_test.py @@ -19,7 +19,10 @@ from timesketch.lib.definitions import HTTP_STATUS_CODE_REDIRECT from timesketch.lib.definitions import HTTP_STATUS_CODE_OK +from timesketch.lib.definitions import HTTP_STATUS_CODE_UNAUTHORIZED +from timesketch.lib.google_auth import CSRF_KEY from timesketch.lib.testlib import BaseTest +from timesketch.models.user import User class AuthViewTest(BaseTest): @@ -156,3 +159,188 @@ def side_effect(*_, **kwargs): # We expect 200 because scope mismatch is now relaxed self.assertEqual(response.status_code, HTTP_STATUS_CODE_OK) self.assertIn(b"Authenticated", response.data) + + +class GoogleOpenIdConnectViewTest(BaseTest): + """Tests for the google_openid_connect view.""" + + def _login_via_oidc(self, decoded_jwt, csrf_token="test-csrf-token"): + """Call the OIDC callback route with the OIDC dependencies mocked.""" + with self.client.session_transaction() as sess: + sess[CSRF_KEY] = csrf_token + + discovery_document = { + "id_token_signing_alg_values_supported": ["RS256"], + "issuer": "https://example-issuer.example.com", + "jwks_uri": "https://example-issuer.example.com/certs", + } + with mock.patch( + "timesketch.views.auth.get_encoded_jwt_over_https", + return_value="encoded-jwt", + ), mock.patch( + "timesketch.views.auth.get_oauth2_discovery_document", + return_value=discovery_document, + ), mock.patch( + "timesketch.views.auth.get_public_key_for_jwt", + return_value="public-key", + ), mock.patch( + "timesketch.views.auth.decode_jwt", return_value=decoded_jwt + ), mock.patch( + "timesketch.views.auth.validate_jwt" + ): + return self.client.get( + f"/login/google_openid_connect/?code=test-code&state={csrf_token}" + ) + + def test_login_allowed_users(self): + """Test GOOGLE_OIDC_ALLOWED_USERS restricts who can log in.""" + cases = [ + ("no allowlist configured allows any user", [], HTTP_STATUS_CODE_REDIRECT), + ( + "user present in the allowlist is allowed", + ["allowed@example.com"], + HTTP_STATUS_CODE_REDIRECT, + ), + ( + "user missing from the allowlist is rejected", + ["other@example.com"], + HTTP_STATUS_CODE_UNAUTHORIZED, + ), + ] + for description, allowed_users, expected_status in cases: + with self.subTest(description): + self.app.config["GOOGLE_OIDC_ALLOWED_USERS"] = allowed_users + response = self._login_via_oidc({"email": "allowed@example.com"}) + self.assertEqual(response.status_code, expected_status) + + def test_login_rejects_user_not_in_allowed_group(self): + """Test login is rejected for a user without an allowed group.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_ALLOWED_GROUPS"] = ["Admins"] + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Users"]} + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_UNAUTHORIZED) + self.assertIsNone(User.query.filter_by(username="newuser@example.com").first()) + + def test_login_allows_user_in_allowed_group(self): + """Test login succeeds for a user who is a member of an allowed group.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_ALLOWED_GROUPS"] = ["Admins", "Users"] + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Users"]} + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + self.assertIsNotNone( + User.query.filter_by(username="newuser@example.com").first() + ) + + def test_login_syncs_groups_and_grants_admin(self): + """Test a successful login syncs groups and grants admin rights.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_GROUPS_SYNC_ENABLED"] = True + self.app.config["GOOGLE_OIDC_ADMIN_GROUP"] = "Admins" + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Admins", "Analysts"]} + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertIsNotNone(user) + self.assertTrue(user.admin) + self.assertCountEqual( + [group.name for group in user.groups], ["Admins", "Analysts"] + ) + + def test_login_does_not_sync_groups_when_sync_disabled(self): + """Test groups/admin are untouched when GROUPS_SYNC_ENABLED is off.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_ADMIN_GROUP"] = "Admins" + self.app.config["GOOGLE_OIDC_GROUPS_SYNC_ENABLED"] = False + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Admins"]} + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertIsNotNone(user) + self.assertFalse(user.admin) + self.assertEqual(list(user.groups), []) + + def test_login_removes_stale_groups_and_admin_on_subsequent_login(self): + """Test groups/admin no longer present in the token get revoked.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_GROUPS_SYNC_ENABLED"] = True + self.app.config["GOOGLE_OIDC_ADMIN_GROUP"] = "Admins" + self.app.config["GOOGLE_OIDC_GROUPS_REMOVE_STALE"] = True + + self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Admins", "Analysts"]}, + csrf_token="csrf-token-1", + ) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertTrue(user.admin) + self.assertCountEqual( + [group.name for group in user.groups], ["Admins", "Analysts"] + ) + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Analysts"]}, + csrf_token="csrf-token-2", + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertFalse(user.admin) + self.assertEqual([group.name for group in user.groups], ["Analysts"]) + + def test_login_keeps_stale_groups_and_admin_when_remove_stale_disabled(self): + """Test groups/admin are kept when GROUPS_REMOVE_STALE is off.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_GROUPS_SYNC_ENABLED"] = True + self.app.config["GOOGLE_OIDC_ADMIN_GROUP"] = "Admins" + self.app.config["GOOGLE_OIDC_GROUPS_REMOVE_STALE"] = False + + self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Admins", "Analysts"]}, + csrf_token="csrf-token-1", + ) + + response = self._login_via_oidc( + {"email": "newuser@example.com", "groups": ["Analysts"]}, + csrf_token="csrf-token-2", + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertTrue(user.admin) + self.assertCountEqual( + [group.name for group in user.groups], ["Admins", "Analysts"] + ) + + def test_login_extracts_groups_using_separator_and_regex(self): + """Test the groups claim is parsed using the separator/regex config.""" + self.app.config["GOOGLE_OIDC_GROUPS_CLAIM"] = "groups" + self.app.config["GOOGLE_OIDC_GROUPS_SEPARATOR"] = ";" + self.app.config["GOOGLE_OIDC_GROUPS_REGEX"] = "CN=([^,]+)" + self.app.config["GOOGLE_OIDC_GROUPS_SYNC_ENABLED"] = True + + response = self._login_via_oidc( + { + "email": "newuser@example.com", + "groups": "CN=Admins,OU=X;CN=Analysts,OU=Y", + } + ) + + self.assertEqual(response.status_code, HTTP_STATUS_CODE_REDIRECT) + user = User.query.filter_by(username="newuser@example.com").first() + self.assertIsNotNone(user) + self.assertCountEqual( + [group.name for group in user.groups], ["Admins", "Analysts"] + )