group canonicalization to avoid collisions, security config, integration test

Author: phact

cloud_securityconfig/roles.yml

@@ -33,6 +33,7 @@ openrag_user_role:
         {"bool":{"should":[
           {"term":{"owner":"${user.name}"}},
           {"term":{"allowed_users":"${user.name}"}},
+          {"terms":{"allowed_groups":[${user.roles}]}},
           {"bool":{"must_not":{"exists":{"field":"owner"}}}}
         ],"minimum_should_match":1}}
     - index_patterns: ["api_keys", "api_keys*"]

pyproject.toml

@@ -79,6 +79,9 @@ openrag = "tui.main:run_tui"
 filterwarnings = [
     "error::RuntimeWarning",
 ]
+markers = [
+    "openrag_skip_app_onboard: selected tests manage their own infrastructure clients and do not need the session app onboarding fixture",
+]
 
 [tool.uv]
 package = true
@@ -152,4 +155,3 @@ exclude = ["flows/components"]
 module = "api.settings.endpoints"
 disable_error_code = ["return-value"]
 
-

securityconfig/roles.yml

@@ -33,6 +33,7 @@ openrag_user_role:
         {"bool":{"should":[
           {"term":{"owner":"${user.name}"}},
           {"term":{"allowed_users":"${user.name}"}},
+          {"terms":{"allowed_groups":[${user.roles}]}},
           {"bool":{"must_not":{"exists":{"field":"owner"}}}}
         ],"minimum_should_match":1}}
     - index_patterns: ["api_keys", "api_keys*"]

src/app/container.py

@@ -23,6 +23,7 @@
 from services.docling_polling_service import DoclingPollingService
 from services.document_service import DocumentService
 from services.flows_service import FlowsService
+from services.group_acl_service import GroupACLService
 from services.knowledge_filter_service import KnowledgeFilterService
 from services.langflow_file_service import LangflowFileService
 from services.langflow_mcp_service import LangflowMCPService
@@ -121,6 +122,7 @@ async def initialize_services():
         langflow_connector_service=langflow_connector_service,
         openrag_connector_service=openrag_connector_service,
     )
+    group_acl_service = GroupACLService(connector_service)
 
     auth_service = AuthService(
         session_manager,
@@ -205,6 +207,7 @@ def _lazy_session_factory():
         "langflow_file_service": langflow_file_service,
         "auth_service": auth_service,
         "connector_service": connector_service,
+        "group_acl_service": group_acl_service,
         "knowledge_filter_service": knowledge_filter_service,
         "models_service": models_service,
         "monitor_service": monitor_service,

src/connectors/base.py

@@ -143,3 +143,12 @@ async def _detect_base_url(self) -> Optional[str]:
         Subclasses (OneDrive, SharePoint) should override this method.
         """
         return None
+
+    async def get_current_user_group_roles(self) -> List[str]:
+        """Return OpenSearch backend roles for the current connector user.
+
+        Connectors that support upstream group ACLs can override this hook.
+        The core ACL service calls it generically so new connectors only need
+        to implement their own provider-specific group lookup.
+        """
+        return []

src/connectors/google_drive/connector.py

@@ -13,6 +13,10 @@
 from utils.logging_config import get_logger
 
 from ..base import BaseConnector, ConnectorDocument, DocumentACL
+from ..google_drive_acl import (
+    get_current_user_google_group_roles,
+    google_drive_group_role,
+)
 from .oauth import GoogleDriveOAuth
 
 logger = get_logger(__name__)
@@ -90,6 +94,8 @@ def emit(self, doc: ConnectorDocument) -> None:
         logger.debug(f"Emitting document: {doc.id} ({doc.filename})")
 
     def __init__(self, config: Dict[str, Any]) -> None:
+        super().__init__(config)
+
         # Read from config OR env (backend env, not NEXT_PUBLIC_*):
         env_client_id = os.getenv(self.CLIENT_ID_ENV_VAR)
         env_client_secret = os.getenv(self.CLIENT_SECRET_ENV_VAR)
@@ -570,6 +576,12 @@ async def authenticate(self) -> bool:
             logger.error("[GoogleDrive] authenticate failed: %s", e)
             return False
 
+    async def get_current_user_group_roles(self) -> List[str]:
+        """Return canonical group ACL roles for the connected Google user."""
+        if not self._authenticated and not await self.authenticate():
+            return []
+        return await get_current_user_google_group_roles(self.service, self.creds)
+
     async def list_files(
         self,
         page_token: Optional[str] = None,
@@ -651,7 +663,9 @@ def _extract_google_drive_acl(self, file_meta: Dict) -> DocumentACL:
 
                 # Add allowed groups
                 elif perm_type == "group" and email:
-                    allowed_groups.append(email)
+                    group_role = google_drive_group_role(email)
+                    if group_role:
+                        allowed_groups.append(group_role)
 
             # Fallback to file owners if no owner found in permissions
             if not owner and file_meta.get("owners"):

src/connectors/google_drive/oauth.py

@@ -25,6 +25,7 @@ class GoogleDriveOAuth:
         "profile",
         "https://www.googleapis.com/auth/drive.readonly",
         "https://www.googleapis.com/auth/drive.metadata.readonly",
+        "https://www.googleapis.com/auth/admin.directory.group.readonly",
     ]
 
     AUTH_ENDPOINT = "https://accounts.google.com/o/oauth2/v2/auth"

src/connectors/google_drive_acl.py

@@ -0,0 +1,141 @@
+"""Google Drive group ACL helpers."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import jwt
+from googleapiclient.discovery import build
+from googleapiclient.errors import HttpError
+
+from utils.group_acl import canonical_group_role
+from utils.logging_config import get_logger
+
+
+logger = get_logger(__name__)
+
+GOOGLE_DRIVE_GROUP_PROVIDER = "gdrive"
+
+
+def _group_tenant(group_email: str) -> str:
+    if "@" in group_email:
+        return group_email.rsplit("@", 1)[1].lower()
+    return "global"
+
+
+def google_drive_group_role(group_email: str | None) -> str | None:
+    """Return the canonical OpenSearch role for a Google Drive group email."""
+    if not group_email:
+        return None
+    email = group_email.strip().lower()
+    if not email:
+        return None
+    return canonical_group_role(
+        GOOGLE_DRIVE_GROUP_PROVIDER,
+        _group_tenant(email),
+        email,
+    )
+
+
+def _email_from_id_token(id_token: str | None) -> str | None:
+    if not id_token:
+        return None
+    try:
+        claims = jwt.decode(
+            id_token,
+            options={"verify_signature": False, "verify_aud": False},
+        )
+        email = claims.get("email")
+        if email:
+            return str(email)
+    except Exception as e:
+        logger.debug("Could not decode Google id_token email", error=str(e))
+    return None
+
+
+async def _execute_google_request(request: Any) -> dict[str, Any]:
+    return await asyncio.to_thread(request.execute)
+
+
+async def _get_drive_user_email(drive_service: Any, credentials: Any) -> str | None:
+    email = _email_from_id_token(getattr(credentials, "id_token", None))
+    if email:
+        return email
+
+    if drive_service is None:
+        return None
+
+    try:
+        about = await _execute_google_request(
+            drive_service.about().get(fields="user(emailAddress)")
+        )
+        return about.get("user", {}).get("emailAddress")
+    except Exception as e:
+        logger.warning("Google Drive group ACL lookup could not resolve user email", error=str(e))
+        return None
+
+
+async def get_current_user_google_group_roles(
+    drive_service: Any,
+    credentials: Any,
+) -> list[str]:
+    """Fetch Google Workspace groups for the connected Drive user."""
+    if credentials is None:
+        return []
+
+    user_email = await _get_drive_user_email(drive_service, credentials)
+    if not user_email:
+        return []
+
+    try:
+        directory_service = build(
+            "admin",
+            "directory_v1",
+            credentials=credentials,
+            cache_discovery=False,
+        )
+    except Exception as e:
+        logger.warning("Could not create Google Directory client for group ACLs", error=str(e))
+        return []
+
+    roles: list[str] = []
+    seen: set[str] = set()
+    page_token: str | None = None
+
+    try:
+        while True:
+            request = directory_service.groups().list(
+                customer="my_customer",
+                userKey=user_email,
+                maxResults=200,
+                pageToken=page_token,
+                fields="nextPageToken,groups(email)",
+            )
+            response = await _execute_google_request(request)
+
+            for group in response.get("groups", []) or []:
+                role = google_drive_group_role(group.get("email"))
+                if role and role not in seen:
+                    seen.add(role)
+                    roles.append(role)
+
+            page_token = response.get("nextPageToken")
+            if not page_token:
+                break
+    except HttpError as e:
+        status = getattr(getattr(e, "resp", None), "status", None)
+        if status in (401, 403):
+            logger.warning(
+                "Google Directory group ACL lookup denied; check Admin SDK API and "
+                "admin.directory.group.readonly consent",
+                status_code=status,
+            )
+            return []
+        logger.warning("Google Directory group ACL lookup failed", error=str(e))
+        return []
+    except Exception as e:
+        logger.warning("Google Directory group ACL lookup errored", error=str(e))
+        return []
+
+    return roles

src/connectors/microsoft_graph_acl.py

@@ -0,0 +1,132 @@
+"""Microsoft Graph group ACL helpers shared by Microsoft connectors."""
+
+from __future__ import annotations
+
+import inspect
+from typing import Any
+
+import httpx
+import jwt
+
+from utils.group_acl import canonical_group_role, canonical_group_roles
+from utils.logging_config import get_logger
+
+
+logger = get_logger(__name__)
+
+MICROSOFT_GRAPH_GROUP_PROVIDER = "m365"
+
+
+def tenant_id_from_access_token(access_token: str | None, fallback: str | None = None) -> str:
+    """Read the tenant id from a Microsoft access token without validating it."""
+    if access_token:
+        raw_token = access_token.removeprefix("Bearer ").strip()
+        try:
+            claims = jwt.decode(
+                raw_token,
+                options={"verify_signature": False, "verify_aud": False},
+            )
+            token_tenant = claims.get("tid")
+            if token_tenant:
+                return token_tenant
+        except Exception as e:
+            logger.debug("Could not decode Microsoft access token tenant", error=str(e))
+    return fallback or "common"
+
+
+def microsoft_group_role(
+    group_id: str | None,
+    *,
+    access_token: str | None = None,
+    tenant_id: str | None = None,
+) -> str | None:
+    """Return the canonical OpenSearch role for a Microsoft group id."""
+    if not group_id:
+        return None
+    resolved_tenant = tenant_id_from_access_token(access_token, fallback=tenant_id)
+    return canonical_group_role(
+        MICROSOFT_GRAPH_GROUP_PROVIDER,
+        resolved_tenant,
+        group_id,
+    )
+
+
+async def _maybe_await(value: Any) -> Any:
+    if inspect.isawaitable(value):
+        return await value
+    return value
+
+
+async def get_oauth_access_token(oauth: Any) -> str | None:
+    """Return an access token string from either old dict or current string APIs."""
+    if oauth is None:
+        return None
+    token_value = await _maybe_await(oauth.get_access_token())
+    if isinstance(token_value, dict):
+        return token_value.get("access_token")
+    if isinstance(token_value, str):
+        return token_value.removeprefix("Bearer ").strip()
+    return None
+
+
+async def get_current_user_microsoft_group_roles(
+    oauth: Any,
+    graph_base_url: str,
+    *,
+    tenant_id: str | None = None,
+    timeout_seconds: float = 10.0,
+) -> list[str]:
+    """Fetch transitive Microsoft group memberships for the current OAuth user."""
+    if oauth is None:
+        return []
+
+    try:
+        access_token = await get_oauth_access_token(oauth)
+    except Exception as e:
+        logger.warning("Unable to get Microsoft Graph token for group ACLs", error=str(e))
+        return []
+
+    if not access_token:
+        return []
+
+    resolved_tenant = tenant_id_from_access_token(access_token, fallback=tenant_id)
+    headers = {"Authorization": f"Bearer {access_token}"}
+    url = f"{graph_base_url}/me/transitiveMemberOf/microsoft.graph.group"
+    params: dict[str, str] | None = {"$select": "id"}
+    group_ids: list[str] = []
+
+    try:
+        async with httpx.AsyncClient(timeout=timeout_seconds) as client:
+            while url:
+                response = await client.get(url, headers=headers, params=params)
+                params = None
+                if response.status_code in (401, 403):
+                    logger.warning(
+                        "Microsoft Graph group ACL lookup denied",
+                        status_code=response.status_code,
+                        response_text=response.text[:500],
+                    )
+                    return []
+                if response.status_code != 200:
+                    logger.warning(
+                        "Microsoft Graph group ACL lookup failed",
+                        status_code=response.status_code,
+                        response_text=response.text[:500],
+                    )
+                    return []
+
+                data = response.json()
+                for entry in data.get("value", []):
+                    group_id = entry.get("id")
+                    if group_id:
+                        group_ids.append(group_id)
+                url = data.get("@odata.nextLink")
+    except Exception as e:
+        logger.warning("Microsoft Graph group ACL lookup errored", error=str(e))
+        return []
+
+    return canonical_group_roles(
+        MICROSOFT_GRAPH_GROUP_PROVIDER,
+        resolved_tenant,
+        group_ids,
+    )