OIDC Group/Role-Based Access Control and Membership Sync

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.

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.
 

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,
+                )