feat: group acls#1611
Merged
Merged
Conversation
github-actions
Bot
added
backend
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds canonical group-role utilities and connector-specific group-role fetchers (Google Cloud Identity/Directory and Microsoft Graph), a GroupACLService with optional caching and create_opensearch_jwt, SessionManager JWT refactor and TTL helpers, connector flows updated to use ACL-aware sync JWTs, Google OAuth scope validation, Langflow/Docling call fixes, and tests/integration for ACL and onboarding flows. ChangesCore ACL & Auth
Langflow / Docling fixes
OAuth & Connector formatting
Tests, onboarding, and CI
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Suggested reviewers
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/connectors/google_drive/oauth.py (1)
22-29:⚠️ Potential issue | 🟠 Major | ⚡ Quick winForce re-consent when persisted scopes miss the new Admin Directory permission.
Adding
admin.directory.group.readonlytoSCOPESis insufficient for existing connections. Theload_credentials()method usestoken_data.get("scopes", self.SCOPES)(line 80), which retrieves stored scopes from older tokens that predate this change. Consequently, credentials are reconstructed without the new scope, and when the Directory group lookup fails with 401/403, the error is silently converted to[](lines 128–134 ingoogle_drive_acl.py), causing group ACLs to never activate until the user manually reconnects.Suggested fix
try: token_data = json.loads(raw_data) except Exception: logger.debug("[GoogleDrive] load_credentials: corrupted token file, removing") if os.path.exists(self.token_file): os.remove(self.token_file) return None + + stored_scopes = set(token_data.get("scopes") or []) + required_scopes = set(self.SCOPES) + if stored_scopes and not required_scopes.issubset(stored_scopes): + logger.info("[GoogleDrive] load_credentials: OAuth scopes changed, forcing re-auth") + if os.path.exists(self.token_file): + os.remove(self.token_file) + return None🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/connectors/google_drive/oauth.py` around lines 22 - 29, Persisted token scopes can omit the new "https://www.googleapis.com/auth/admin.directory.group.readonly", causing load_credentials() to rebuild credentials without it; update load_credentials() in oauth.py to read scopes = token_data.get("scopes") and if that list does not include the admin.directory.group.readonly scope, treat the token as missing scopes (e.g., ignore persisted scopes and force full re-consent by using self.SCOPES or by marking credentials invalid so OAuth flow runs). Also ensure any downstream logic in google_drive_acl.py that currently swallows 401/403 into an empty list will surface the auth failure so the re-consent flow is triggered (i.e., do not convert auth errors to [] silently; propagate an auth error or return a clear signal to re-authenticate).
🧹 Nitpick comments (3)
src/connectors/onedrive/connector.py (1)
1-16: ⚡ Quick winSwitch this connector to
get_logger.The touched import block still pulls
loggingfrom the stdlib, so this file bypasses the shared logger configuration used undersrc/.As per coding guidelines, "Use get_logger from utils.logging_config — never import stdlib logging directly."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/connectors/onedrive/connector.py` around lines 1 - 16, The module imports the stdlib logging and creates a module logger named logger; replace that by using the shared get_logger from utils.logging_config so the connector honors the application-wide logging configuration. Update the import to pull get_logger from utils.logging_config and change the module-level logger creation (symbol: logger) to call get_logger(__name__) instead of using logging.getLogger, ensuring all references to logger continue to work.src/connectors/base.py (1)
58-78: ⚡ Quick winCentralize credential lookup instead of calling
os.getenvhere.These base helpers still read env vars directly from
src/, which keeps config sourcing split across the codebase. Please route credential lookup throughconfig/settings.pyand keepBaseConnectorworking off injected values.As per coding guidelines, "Config values must come from config/settings.py (the only place os.environ is read); never access os.environ elsewhere in the codebase."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/connectors/base.py` around lines 58 - 78, The get_client_id and get_client_secret helpers currently call os.getenv directly; change them to use injected/config-sourced values from config/settings.py instead. Update BaseConnector (or its initializer) to accept a settings/config object or to reference a single settings accessor (e.g., a passed-in settings instance or a module-level settings import) and read the client id/secret from that injected source rather than os.getenv; then have get_client_id return the value from that injected settings (and get_client_secret similarly) while preserving the existing validation and error messages (raise NotImplementedError if the CLASS-level ENV_VAR names are not set and ValueError if the resolved secret/id is falsy). Ensure you reference get_client_id and get_client_secret to locate the code to change.src/services/group_acl_service.py (1)
55-60: ⚡ Quick winClear per-user locks when invalidating cache entries.
_locksretains entries indefinitely, so long-running processes can accumulate one lock per seen user ID.Proposed change
def invalidate_user(self, user_id: str) -> None: self._cache.pop(user_id, None) + self._locks.pop(user_id, None) def clear(self) -> None: self._cache.clear() + self._locks.clear()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/group_acl_service.py` around lines 55 - 60, invalidate_user currently removes entries from self._cache but leaves per-user locks in self._locks, causing unbounded growth; update invalidate_user to also remove the corresponding lock (e.g., self._locks.pop(user_id, None)) and update clear to call self._locks.clear() so locks are cleared when cache is cleared; ensure you use the safe pop variants to avoid KeyError and keep behavior consistent with existing methods.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/connectors/base.py`:
- Around line 93-95: The abstract method signature for list_files uses
Optional[...] which triggers UP045; update the parameter annotations to use PEP
604 union syntax instead (change Optional[str] to str | None and Optional[int]
to int | None) in the async def list_files method in class/base.py so the
signature reads using `| None` for page_token and max_files while preserving
return type Dict[str, Any] and behavior.
In `@src/connectors/onedrive/connector.py`:
- Around line 529-537: The fallback branches that construct DocumentACL after
using cached_info (and the shares-endpoint fallback) are passing
user_permissions and group_permissions which don't match the DocumentACL
signature; update those constructors to use the correct parameters allowed_users
and allowed_groups (e.g., empty collections) so instantiation succeeds (look for
the code around cached_info, _download_file_from_url, and the shares-endpoint
branch and replace user_permissions/group_permissions with
allowed_users/allowed_groups).
In `@src/services/group_acl_service.py`:
- Around line 6-24: The constructor of GroupACLService currently reads
OPENRAG_GROUP_ACL_CACHE_TTL directly from the environment; change it to read the
TTL from the centralized config module instead: import the appropriate value
from config.settings (e.g. settings.OPENRAG_GROUP_ACL_CACHE_TTL), parse/coerce
it to int with a default of 0 if missing, and assign self.cache_ttl_seconds =
max(parsed_value, 0); update the __init__ in GroupACLService to use that
settings value and remove the os.getenv usage (and unnecessary os import).
In `@src/services/langflow_file_service.py`:
- Around line 600-602: The call to submit_to_docling uses incorrect keyword
names (user_id, auth_header) that don't match the function signature (owner,
jwt_token), causing a TypeError; change the call at the task_id assignment to
pass matching parameters (either positional or use owner=owner and
jwt_token=jwt_token) so submit_to_docling(owner, jwt_token, ...) receives the
correct arguments and the ingestion flow won't break.
In `@src/session_manager.py`:
- Around line 188-194: The _get_oidc_issuer method (and the other env reads
around lines 252-260) currently reads os.environ directly; refactor to consume
the centralized config values from config.settings.py instead: import the
settings object (e.g. from config.settings import settings) and replace direct
os.getenv("OPENRAG_FQDN") and any direct os.environ access with
settings.OPENRAG_FQDN and settings.OPENRAG_OPENSEARCH_JWT_TTL (or the exact
names used in config/settings.py), preserving the same default behavior if the
setting is empty; update _get_oidc_issuer to build the issuer URL from
settings.OPENRAG_FQDN rather than os.getenv and change the other block to use
settings.OPENRAG_OPENSEARCH_JWT_TTL accordingly.
---
Outside diff comments:
In `@src/connectors/google_drive/oauth.py`:
- Around line 22-29: Persisted token scopes can omit the new
"https://www.googleapis.com/auth/admin.directory.group.readonly", causing
load_credentials() to rebuild credentials without it; update load_credentials()
in oauth.py to read scopes = token_data.get("scopes") and if that list does not
include the admin.directory.group.readonly scope, treat the token as missing
scopes (e.g., ignore persisted scopes and force full re-consent by using
self.SCOPES or by marking credentials invalid so OAuth flow runs). Also ensure
any downstream logic in google_drive_acl.py that currently swallows 401/403 into
an empty list will surface the auth failure so the re-consent flow is triggered
(i.e., do not convert auth errors to [] silently; propagate an auth error or
return a clear signal to re-authenticate).
---
Nitpick comments:
In `@src/connectors/base.py`:
- Around line 58-78: The get_client_id and get_client_secret helpers currently
call os.getenv directly; change them to use injected/config-sourced values from
config/settings.py instead. Update BaseConnector (or its initializer) to accept
a settings/config object or to reference a single settings accessor (e.g., a
passed-in settings instance or a module-level settings import) and read the
client id/secret from that injected source rather than os.getenv; then have
get_client_id return the value from that injected settings (and
get_client_secret similarly) while preserving the existing validation and error
messages (raise NotImplementedError if the CLASS-level ENV_VAR names are not set
and ValueError if the resolved secret/id is falsy). Ensure you reference
get_client_id and get_client_secret to locate the code to change.
In `@src/connectors/onedrive/connector.py`:
- Around line 1-16: The module imports the stdlib logging and creates a module
logger named logger; replace that by using the shared get_logger from
utils.logging_config so the connector honors the application-wide logging
configuration. Update the import to pull get_logger from utils.logging_config
and change the module-level logger creation (symbol: logger) to call
get_logger(__name__) instead of using logging.getLogger, ensuring all references
to logger continue to work.
In `@src/services/group_acl_service.py`:
- Around line 55-60: invalidate_user currently removes entries from self._cache
but leaves per-user locks in self._locks, causing unbounded growth; update
invalidate_user to also remove the corresponding lock (e.g.,
self._locks.pop(user_id, None)) and update clear to call self._locks.clear() so
locks are cleared when cache is cleared; ensure you use the safe pop variants to
avoid KeyError and keep behavior consistent with existing methods.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 042d933e-28e0-474d-ae8b-2615d8399b06
📒 Files selected for processing (21)
cloud_securityconfig/roles.ymlpyproject.tomlsecurityconfig/roles.ymlsrc/app/container.pysrc/connectors/base.pysrc/connectors/google_drive/connector.pysrc/connectors/google_drive/oauth.pysrc/connectors/google_drive_acl.pysrc/connectors/microsoft_graph_acl.pysrc/connectors/onedrive/connector.pysrc/connectors/sharepoint/connector.pysrc/dependencies.pysrc/services/docling_service.pysrc/services/group_acl_service.pysrc/services/langflow_file_service.pysrc/session_manager.pysrc/utils/group_acl.pytests/conftest.pytests/integration/core/test_group_acl_dls.pytests/unit/services/test_langflow_mcp_service.pytests/unit/test_group_acl.py
github-actions
Bot
added
enhancement
github-actions
Bot
added
enhancement
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/api/connectors.py (1)
1-1:⚠️ Potential issue | 🟠 Major | ⚡ Quick winUse builtin generics to fix the Ruff UP006 lint failure.
The backend Ruff job is failing with UP006 on this module. Replace
Dict/Listin the import andConnectorSyncBodyannotations with builtin generics.Suggested fix
-from typing import Any, Dict, List, Optional +from typing import Any, Optional @@ class ConnectorSyncBody(BaseModel): - max_files: Optional[int] = None - selected_files: Optional[List[Any]] = None + max_files: int | None = None + selected_files: list[Any] | None = None @@ - bucket_filter: Optional[List[str]] = None + bucket_filter: list[str] | None = None @@ - settings: Optional[Dict[str, Any]] = None + settings: dict[str, Any] | None = NoneAlso applies to: 80-89
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/api/connectors.py` at line 1, The lint UP006 is caused by using typing.Dict/typing.List; update the imports and type annotations to use built-in generics instead: remove Dict and List from the import line and replace their usages in the module (notably the ConnectorSyncBody type annotations around the ConnectorSyncBody definition and usages referenced at lines ~80-89) with dict[...] and list[...]; ensure Optional/Any remain as needed and adjust the import line to only import typing names still required.
🧹 Nitpick comments (1)
src/session_manager.py (1)
206-237: ⚡ Quick winMake the disabled-signing path explicit in the return types.
_create_signed_jwt_token()returnsNoneon Lines 234-236, but it and both public wrappers are annotated as always returningstr. That hides a reachableNonecontract from callers and from type-checking.Proposed fix
- def _create_signed_jwt_token( + def _create_signed_jwt_token( self, user: User, *, expires_delta: timedelta, backend_roles: Iterable[str] | None = None, - ) -> str: + ) -> Optional[str]: @@ - def create_jwt_token(self, user: User) -> str: + def create_jwt_token(self, user: User) -> Optional[str]: @@ - def create_opensearch_jwt_token( + def create_opensearch_jwt_token( self, user: User, group_roles: Iterable[str] | None = None, ttl_seconds: int | None = None, - ) -> str: + ) -> Optional[str]:Also applies to: 240-262
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/session_manager.py` around lines 206 - 237, The helper _create_signed_jwt_token currently returns None when signing is disabled but is annotated to return str; change its signature to return Optional[str] (import Optional from typing) and update both public wrapper methods that call it to also return Optional[str] (or propagate/handle None) so the type system reflects the disabled-signing path; ensure the code paths that log "JWT signing requested but signing is disabled" still return None explicitly and callers handle or raise on a None token rather than assuming a str.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/api/connectors.py`:
- Around line 514-523: The route handler block that instantiates GroupACLService
and calls create_opensearch_jwt should be removed and the JWT should be derived
once inside ConnectorService.sync_specific_files (use the existing
_get_effective_sync_jwt/connector_service logic) so we do not double-mint or
instantiate a service in the handler; instead ensure GroupACLService is created
in application startup and injected via dependencies (use connector_service and
session_manager passed into sync_specific_files to obtain the JWT via
_get_effective_sync_jwt or a new connector_service method) and delete the direct
GroupACLService instantiation and create_opensearch_jwt call from the route
handler.
In `@src/connectors/google_drive_acl.py`:
- Around line 42-55: The helper _email_from_id_token currently decodes id_token
without verification; change it to validate the token signature and audience
using the configured Google client ID (CLIENT_ID_ENV_VAR) before returning the
email claim, i.e., perform full jwt verification (signature + aud) against the
client ID and reject/return None if verification fails, and as an alternative
path use the authenticated Drive API call
drive_service.about().get(fields="user(emailAddress)") to obtain the user email
when verification is not possible; ensure errors are logged (logger.debug) with
details and that the function returns None on any verification or retrieval
failure.
---
Outside diff comments:
In `@src/api/connectors.py`:
- Line 1: The lint UP006 is caused by using typing.Dict/typing.List; update the
imports and type annotations to use built-in generics instead: remove Dict and
List from the import line and replace their usages in the module (notably the
ConnectorSyncBody type annotations around the ConnectorSyncBody definition and
usages referenced at lines ~80-89) with dict[...] and list[...]; ensure
Optional/Any remain as needed and adjust the import line to only import typing
names still required.
---
Nitpick comments:
In `@src/session_manager.py`:
- Around line 206-237: The helper _create_signed_jwt_token currently returns
None when signing is disabled but is annotated to return str; change its
signature to return Optional[str] (import Optional from typing) and update both
public wrapper methods that call it to also return Optional[str] (or
propagate/handle None) so the type system reflects the disabled-signing path;
ensure the code paths that log "JWT signing requested but signing is disabled"
still return None explicitly and callers handle or raise on a None token rather
than assuming a str.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 56f5389a-7003-430f-ac1a-052cd444214d
📒 Files selected for processing (12)
.env.examplesrc/api/connectors.pysrc/connectors/google_drive/oauth.pysrc/connectors/google_drive_acl.pysrc/connectors/langflow_connector_service.pysrc/connectors/onedrive/connector.pysrc/connectors/service.pysrc/dependencies.pysrc/services/group_acl_service.pysrc/services/langflow_file_service.pysrc/session_manager.pytests/unit/test_group_acl.py
🚧 Files skipped from review as they are similar to previous changes (3)
- src/dependencies.py
- src/connectors/google_drive/oauth.py
- src/connectors/onedrive/connector.py
github-actions
Bot
added
enhancement
github-actions
Bot
added
enhancement
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
tests/unit/test_group_acl.py (1)
72-72: 💤 Low valueConsider making the TTL calculation more explicit.
The assertion uses the magic number
3900whenINGESTION_TIMEOUTis3600. While the test correctly validates behavior, calculating the expected value fromconfig.settings.get_opensearch_jwt_ttl_seconds()or referencing the buffer constant would make the relationship more explicit and maintainable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_group_acl.py` at line 72, Replace the hardcoded magic number in the assertion with a computed value so the test documents the TTL relationship explicitly: use config.settings.get_opensearch_jwt_ttl_seconds() (or the INGESTION_TIMEOUT constant) plus the buffer/extra seconds used by the implementation to compute the expected TTL for payload["exp"] - payload["iat"]; update the assertion that currently references payload to compare against this computed_expected_ttl instead of 3900 (locate the assertion in tests/unit/test_group_acl.py referencing payload and exp/iat).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@tests/unit/test_group_acl.py`:
- Line 72: Replace the hardcoded magic number in the assertion with a computed
value so the test documents the TTL relationship explicitly: use
config.settings.get_opensearch_jwt_ttl_seconds() (or the INGESTION_TIMEOUT
constant) plus the buffer/extra seconds used by the implementation to compute
the expected TTL for payload["exp"] - payload["iat"]; update the assertion that
currently references payload to compare against this computed_expected_ttl
instead of 3900 (locate the assertion in tests/unit/test_group_acl.py
referencing payload and exp/iat).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: a3a0422f-2a24-4d1f-afb5-285e6339d3ed
📒 Files selected for processing (9)
src/config/settings.pysrc/connectors/google_drive/oauth.pysrc/connectors/onedrive/connector.pysrc/services/group_acl_service.pysrc/services/langflow_file_service.pysrc/session_manager.pytests/unit/test_group_acl.pytests/unit/test_langflow_file_service_two_phase.pytests/unit/test_oauth_encryption.py
🚧 Files skipped from review as they are similar to previous changes (2)
- src/services/group_acl_service.py
- src/connectors/onedrive/connector.py
Contributor
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: Length of output: 467 🏁 Script executed: Length of output: 17157
Overview
Functions Explained1.
|
| Field | Type | Description |
|---|---|---|
id |
str |
Langflow-assigned chunk ID (preserved as metadata only) |
text |
str |
Chunk text content |
vector |
list[float] |
Embedding vector (min 1 element) |
page |
int | None |
Source page number |
metadata |
dict |
Arbitrary metadata from Langflow |
2. LangflowIngestBatch (Pydantic Model)
Represents a batch of chunks for one ingestion run:
| Field | Type | Description |
|---|---|---|
ingest_run_id |
str |
Ties this batch to a specific document ingest run |
batch_id |
int |
Sequential batch number (default 0) |
final |
bool |
Whether this is the last batch in the run |
chunks |
list[LangflowIngestChunk] |
Chunks in this batch |
3. _extract_ingest_token(authorization, x_openrag_ingest_token) → str
Extracts the ingest token from either:
X-OpenRAG-Ingest-Tokenheader (preferred), orAuthorization: Bearer <token>header
Raises HTTP 401 if no token is found.
4. _authorized_chunk_id(body, document_id, index) → str
Builds a server-controlled chunk ID in the format:
{document_id}_{batch_id}_{index}
This prevents Langflow from injecting arbitrary chunk IDs. Langflow's original chunk id is preserved only as metadata["langflow_chunk_id"]. Raises HTTP 403 if document_id is empty.
5. ingest_langflow_chunks(...) — Main Endpoint Handler
The core async function registered at POST /internal/ingest/chunks. Steps:
- Extract token via
_extract_ingest_token() - Validate token via
token_service.validate_token()→ returns(context, jti), wherecontextholds authoritative doc/owner/ACL info - Validate run ID —
body.ingest_run_idmust matchcontext.ingest_run_id(HTTP 403 on mismatch) - Remap chunks — converts each
LangflowIngestChunkinto aDocumentIndexChunkwith:- Server-owned
chunk_id(via_authorized_chunk_id) - Authoritative
contextmetadata (owner, ACL, etc.) - Langflow's original chunk ID stored as
metadata["langflow_chunk_id"]
- Server-owned
- Write to OpenSearch via
writer.index_chunks(context, chunks, final=body.final) - Mark token finalized on
final=Truebatch (prevents token replay) - Return
{ "status": "ok", "batch_id": ..., ...result }
Sequence Diagram
sequenceDiagram
actor User
participant LFS as LangflowFileService<br/>(run_ingestion_flow)
participant TokSvc as LangflowIngestTokenService
participant Langflow as Langflow<br/>(embedding pipeline)
participant Endpoint as POST /internal/ingest/chunks<br/>(ingest_langflow_chunks)
participant Writer as DocumentIndexWriter
participant OS as OpenSearch
User->>LFS: Upload file (jwt_token, owner, ACL info)
LFS->>TokSvc: create_token(DocumentIndexContext)
TokSvc-->>LFS: signed JWT (ingest token)
LFS->>Langflow: Run ingestion flow<br/>with global vars:<br/>OPENRAG_INGEST_URL<br/>OPENRAG_INGEST_TOKEN<br/>OPENRAG_INGEST_RUN_ID<br/>OPENRAG_INGEST_BATCH_SIZE
loop For each batch of chunks
Langflow->>Endpoint: POST /internal/ingest/chunks<br/>Authorization: Bearer <token><br/>{ ingest_run_id, batch_id, final, chunks[] }
Endpoint->>Endpoint: _extract_ingest_token()<br/>(from header)
Endpoint->>TokSvc: validate_token(token)
TokSvc-->>Endpoint: (context, jti)<br/>context = {document_id, owner, allowed_users, ...}
Endpoint->>Endpoint: verify ingest_run_id matches context
loop For each chunk
Endpoint->>Endpoint: _authorized_chunk_id()<br/>→ "{doc_id}_{batch_id}_{i}"
Endpoint->>Endpoint: Build DocumentIndexChunk<br/>(chunk_id, text, vector, metadata)
end
Endpoint->>Writer: index_chunks(context, chunks, final=...)
Writer->>OS: Index chunk documents<br/>with owner/ACL fields from context
OS-->>Writer: acknowledgement
Writer-->>Endpoint: { indexed_chunks, document_id }
alt final == True
Endpoint->>TokSvc: mark_finalized(jti)<br/>(token cannot be reused)
end
Endpoint-->>Langflow: { "status": "ok", "batch_id": N, ... }
end
LFS-->>User: Ingestion response
Security Design Summary
| Threat | Mitigation |
|---|---|
| Langflow forging owner/ACL | All ACL metadata comes from the signed token context, not the request body |
| Langflow injecting arbitrary chunk IDs | Chunk IDs are server-generated (_authorized_chunk_id), Langflow IDs are demoted to metadata |
| Token replay after completion | Token is finalized (mark_finalized) after the last batch |
| Cross-run injection | ingest_run_id in body must match the token's context |
| Missing/invalid token | HTTP 401/403 raised immediately |
# Conflicts: # src/connectors/google_drive/connector.py
# Conflicts: # src/api/langflow_files.py
Introduce support for an 'openrag' authentication mode that delegates writes to an OpenRAG backend ingest callback. Adds new inputs (openrag_ingest_token, openrag_ingest_run_id) and exposes openrag_ingest_url/batch_size fields as configurable (and required when openrag is selected). Update auth_mode options/default to include 'openrag', validate required OPENRAG_* fields in _build_auth_kwargs, and adjust the dynamic UI build logic to show/require OpenRAG fields when appropriate. Update ingestion flow JSON to mirror these input/option changes. This enables ingestion via OpenRAG callbacks without direct OpenSearch credentials.
Improve observability and safety for OpenRAG ingest callbacks: trim callback inputs, mask tokens when logging, and emit structured debug payloads. Added request-level summaries and per-batch/response logging (including status, headers and body preview) and attempt to call self.log where available. Also include the callback URL in error messages to aid troubleshooting. Changes touch the OpenSearch multimodal component and the flow JSON that embeds the updated component code.
# Conflicts: # src/config/settings.py # src/connectors/service.py # src/dependencies.py # src/models/processors.py # src/utils/acl_utils.py
…nrag into group-dls-part-2
Import typing.Any and add explicit type annotations to several module-level and local variables for better static type checking: DLS_PRINCIPAL_INDEX_BODY in src/config/settings.py; query in src/services/file_service.py; filter_clauses and fallback_search_body in src/services/search_service.py. No functional changes; only type hints added for clarity and mypy/IDE support.
…nrag into group-dls-part-2
…o group-dls-part-2
…o group-dls-part-2
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
group canonicalization to avoid collisions, security config, integration test
Summary by CodeRabbit
New Features
Tests
Chores