diff --git a/src/backend/base/langflow/alembic/versions/cp01a2b3c4d5_add_role_assignment_grant_sources.py b/src/backend/base/langflow/alembic/versions/cp01a2b3c4d5_add_role_assignment_grant_sources.py new file mode 100644 index 000000000000..64271f8f5e17 --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/cp01a2b3c4d5_add_role_assignment_grant_sources.py @@ -0,0 +1,97 @@ +"""Add independent provenance sources for effective role assignments. + +Phase: EXPAND +Revision ID: cp01a2b3c4d5 +Revises: b7d5f9a3c2e4 +Create Date: 2026-07-30 + +The table is additive. Downgrade removes provenance but deliberately preserves +all effective ``authz_role_assignment`` rows. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import sqlalchemy as sa +import sqlmodel +from alembic import op +from langflow.utils import migration + +if TYPE_CHECKING: + from collections.abc import Sequence + +revision: str = "cp01a2b3c4d5" # pragma: allowlist secret +down_revision: str | None = "b7d5f9a3c2e4" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +TABLE_NAME = "authz_role_assignment_grant" + + +def upgrade() -> None: + conn = op.get_bind() + if migration.table_exists(TABLE_NAME, conn): + return + + op.create_table( + TABLE_NAME, + sa.Column("id", sa.Uuid(), nullable=False), + sa.Column("assignment_id", sa.Uuid(), nullable=False), + sa.Column("source_kind", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("provider_id", sqlmodel.sql.sqltypes.AutoString(length=256), nullable=True), + sa.Column("external_group", sqlmodel.sql.sqltypes.AutoString(length=256), nullable=True), + sa.Column("administrative_actor", sa.Uuid(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint( + "(source_kind = 'manual' AND provider_id IS NULL AND external_group IS NULL) " + "OR (source_kind = 'idp' AND provider_id IS NOT NULL AND external_group IS NOT NULL)", + name="ck_authz_role_assignment_grant_source", + ), + sa.ForeignKeyConstraint( + ["administrative_actor"], + ["user.id"], + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["assignment_id"], + ["authz_role_assignment.id"], + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index( + "ix_authz_role_assignment_grant_assignment_id", + TABLE_NAME, + ["assignment_id"], + unique=False, + ) + op.create_index( + "ix_authz_role_assignment_grant_provider_group", + TABLE_NAME, + ["provider_id", "external_group"], + unique=False, + ) + op.create_index( + "uq_authz_role_assignment_grant_manual", + TABLE_NAME, + ["assignment_id"], + unique=True, + postgresql_where=sa.text("source_kind = 'manual'"), + sqlite_where=sa.text("source_kind = 'manual'"), + ) + op.create_index( + "uq_authz_role_assignment_grant_idp", + TABLE_NAME, + ["assignment_id", "provider_id", "external_group"], + unique=True, + postgresql_where=sa.text("source_kind = 'idp'"), + sqlite_where=sa.text("source_kind = 'idp'"), + ) + + +def downgrade() -> None: + conn = op.get_bind() + if migration.table_exists(TABLE_NAME, conn): + op.drop_table(TABLE_NAME) diff --git a/src/backend/base/langflow/alembic/versions/cp02a2b3c4d5_backfill_manual_role_grant_sources.py b/src/backend/base/langflow/alembic/versions/cp02a2b3c4d5_backfill_manual_role_grant_sources.py new file mode 100644 index 000000000000..361de46b960b --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/cp02a2b3c4d5_backfill_manual_role_grant_sources.py @@ -0,0 +1,78 @@ +"""Backfill existing effective role assignments with manual grant sources. + +Phase: MIGRATE +Revision ID: cp02a2b3c4d5 +Revises: cp01a2b3c4d5 +Create Date: 2026-07-30 + +The data migration is idempotent and leaves effective assignments untouched. +Its downgrade is intentionally a no-op: removing provenance before the schema +downgrade would make a partially downgraded application misclassify grants. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from uuid import uuid4 + +import sqlalchemy as sa +from alembic import op + +if TYPE_CHECKING: + from collections.abc import Sequence + +revision: str = "cp02a2b3c4d5" # pragma: allowlist secret +down_revision: str | None = "cp01a2b3c4d5" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_BATCH_SIZE = 1000 + + +def upgrade() -> None: + conn = op.get_bind() + metadata = sa.MetaData() + assignment = sa.Table("authz_role_assignment", metadata, autoload_with=conn) + grant = sa.Table("authz_role_assignment_grant", metadata, autoload_with=conn) + + existing_source = sa.exists(sa.select(grant.c.id).where(grant.c.assignment_id == assignment.c.id)) + last_assignment_id = None + + def new_grant_id(): + value = uuid4() + return value.hex if conn.dialect.name == "sqlite" else value + + while True: + query = sa.select( + assignment.c.id, + assignment.c.assigned_by, + assignment.c.assigned_at, + ).where(~existing_source) + if last_assignment_id is not None: + query = query.where(assignment.c.id > last_assignment_id) + + rows = conn.execute(query.order_by(assignment.c.id).limit(_BATCH_SIZE)).all() + if not rows: + return + + conn.execute( + grant.insert(), + [ + { + "id": new_grant_id(), + "assignment_id": row.id, + "source_kind": "manual", + "provider_id": None, + "external_group": None, + "administrative_actor": row.assigned_by, + "created_at": row.assigned_at, + "updated_at": row.assigned_at, + } + for row in rows + ], + ) + last_assignment_id = rows[-1].id + + +def downgrade() -> None: + pass diff --git a/src/backend/base/langflow/alembic/versions/cp03a2b3c4d5_allow_skipped_authz_audit_results.py b/src/backend/base/langflow/alembic/versions/cp03a2b3c4d5_allow_skipped_authz_audit_results.py new file mode 100644 index 000000000000..37f9d8527e82 --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/cp03a2b3c4d5_allow_skipped_authz_audit_results.py @@ -0,0 +1,52 @@ +"""Allow skipped authorization reconciliation audit outcomes. + +Phase: EXPAND +Revision ID: cp03a2b3c4d5 +Revises: cp02a2b3c4d5 +Create Date: 2026-07-31 + +The wider result vocabulary is backward compatible with existing services. +Downgrade intentionally preserves both the wider constraint and append-only +``skip`` evidence rather than rewriting historical audit rows. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import sqlalchemy as sa +from alembic import op + +if TYPE_CHECKING: + from collections.abc import Sequence + +revision: str = "cp03a2b3c4d5" # pragma: allowlist secret +down_revision: str | None = "cp02a2b3c4d5" # pragma: allowlist secret +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_TABLE_NAME = "authz_audit_log" +_CONSTRAINT_NAME = "ck_authz_audit_log_result_enum" +_RENDERED_CONSTRAINT_NAME = f"ck_{_TABLE_NAME}_{_CONSTRAINT_NAME}" +_RESULT_CHECK = "result IN ('allow', 'deny', 'owner_override', 'skip')" + + +def upgrade() -> None: + conn = op.get_bind() + checks = sa.inspect(conn).get_check_constraints(_TABLE_NAME) + constraint_names = {_CONSTRAINT_NAME, _RENDERED_CONSTRAINT_NAME} + existing = next((check for check in checks if check.get("name") in constraint_names), None) + if existing is not None and "'skip'" in (existing.get("sqltext") or ""): + return + + with op.batch_alter_table(_TABLE_NAME, schema=None) as batch_op: + if existing is not None: + constraint_name = batch_op.f(existing["name"]) + batch_op.drop_constraint(constraint_name, type_="check") + else: + constraint_name = _CONSTRAINT_NAME + batch_op.create_check_constraint(constraint_name, _RESULT_CHECK) + + +def downgrade() -> None: + pass diff --git a/src/backend/base/langflow/api/v1/authz_audit.py b/src/backend/base/langflow/api/v1/authz_audit.py index 689cd8f40957..f8cc3d7aaff5 100644 --- a/src/backend/base/langflow/api/v1/authz_audit.py +++ b/src/backend/base/langflow/api/v1/authz_audit.py @@ -80,7 +80,7 @@ async def list_audit_log( ] = None, result: Annotated[ str | None, - Query(description="Filter by decision result (``allow`` / ``deny`` / ``owner_override``)."), + Query(description="Filter by audit result (``allow`` / ``deny`` / ``owner_override`` / ``skip``)."), ] = None, since: Annotated[datetime | None, Query(description="Inclusive lower bound on ``timestamp``.")] = None, until: Annotated[datetime | None, Query(description="Exclusive upper bound on ``timestamp``.")] = None, diff --git a/src/backend/base/langflow/api/v1/authz_role_assignments.py b/src/backend/base/langflow/api/v1/authz_role_assignments.py index 3e03e0fb3501..7dad89ed3a26 100644 --- a/src/backend/base/langflow/api/v1/authz_role_assignments.py +++ b/src/backend/base/langflow/api/v1/authz_role_assignments.py @@ -12,7 +12,7 @@ from typing import Annotated from uuid import UUID -from fastapi import APIRouter, HTTPException, Query, status +from fastapi import APIRouter, HTTPException, Query, Response, status from lfx.log.logger import logger from lfx.services.authorization import ( AuthorizationMutation, @@ -25,6 +25,7 @@ from langflow.api.utils import CurrentActiveUser, DbSession from langflow.api.v1.schemas.authz_role_assignments import ( RoleAssignmentCreate, + RoleAssignmentGrantSummary, RoleAssignmentRead, ) from langflow.services.authorization.lifecycle import ( @@ -33,7 +34,7 @@ validate_identity_mutation, ) from langflow.services.authorization.utils import audit_decision -from langflow.services.database.models.auth import AuthzRole, AuthzRoleAssignment +from langflow.services.database.models.auth import AuthzRole, AuthzRoleAssignment, AuthzRoleAssignmentGrant from langflow.services.database.models.user.model import User from langflow.services.deps import get_authorization_service @@ -52,6 +53,50 @@ def _require_superuser(user) -> None: ) +async def _assignment_reads(session, assignments: list[AuthzRoleAssignment]) -> list[RoleAssignmentRead]: + """Serialize effective assignments with source summaries in two queries.""" + if not assignments: + return [] + assignment_ids = [assignment.id for assignment in assignments] + grants = ( + await session.exec( + select(AuthzRoleAssignmentGrant) + .where(AuthzRoleAssignmentGrant.assignment_id.in_(assignment_ids)) + .order_by( + AuthzRoleAssignmentGrant.assignment_id, + AuthzRoleAssignmentGrant.source_kind, + AuthzRoleAssignmentGrant.provider_id, + AuthzRoleAssignmentGrant.external_group, + ) + ) + ).all() + grants_by_assignment: dict[UUID, list[RoleAssignmentGrantSummary]] = {} + for grant in grants: + grants_by_assignment.setdefault(grant.assignment_id, []).append( + RoleAssignmentGrantSummary.model_validate(grant) + ) + return [ + RoleAssignmentRead.model_validate(assignment).model_copy( + update={"grant_sources": grants_by_assignment.get(assignment.id, [])} + ) + for assignment in assignments + ] + + +def _assignment_match(payload: RoleAssignmentCreate): + domain_match = ( + AuthzRoleAssignment.domain_id.is_(None) + if payload.domain_id is None + else AuthzRoleAssignment.domain_id == payload.domain_id + ) + return ( + AuthzRoleAssignment.user_id == payload.user_id, + AuthzRoleAssignment.role_id == payload.role_id, + AuthzRoleAssignment.domain_type == payload.domain_type, + domain_match, + ) + + @router.get("", response_model=list[RoleAssignmentRead]) @router.get("/", response_model=list[RoleAssignmentRead]) async def list_assignments( @@ -87,7 +132,7 @@ async def list_assignments( stmt = stmt.where(AuthzRoleAssignment.domain_id == domain_id) stmt = stmt.order_by(AuthzRoleAssignment.assigned_at.desc(), AuthzRoleAssignment.id).offset(offset).limit(limit) rows = (await session.exec(stmt)).all() - return [RoleAssignmentRead.model_validate(row) for row in rows] + return await _assignment_reads(session, list(rows)) @router.post("", response_model=RoleAssignmentRead, status_code=status.HTTP_201_CREATED) @@ -107,15 +152,41 @@ async def create_assignment( if role is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="role_id not found") - assignment = AuthzRoleAssignment( - user_id=payload.user_id, - role_id=payload.role_id, - domain_type=payload.domain_type, - domain_id=payload.domain_id, - assigned_at=datetime.now(timezone.utc), - assigned_by=current_user.id, + assignment = (await session.exec(select(AuthzRoleAssignment).where(*_assignment_match(payload)))).first() + effective_assignment_created = assignment is None + if assignment is None: + assignment = AuthzRoleAssignment( + user_id=payload.user_id, + role_id=payload.role_id, + domain_type=payload.domain_type, + domain_id=payload.domain_id, + assigned_at=datetime.now(timezone.utc), + assigned_by=current_user.id, + ) + session.add(assignment) + await session.flush() + else: + existing_manual = ( + await session.exec( + select(AuthzRoleAssignmentGrant).where( + AuthzRoleAssignmentGrant.assignment_id == assignment.id, + AuthzRoleAssignmentGrant.source_kind == "manual", + ) + ) + ).first() + if existing_manual is not None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Manual assignment already exists for this user/role/domain", + ) + + session.add( + AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=current_user.id, + ) ) - session.add(assignment) authorization_service = get_authorization_service() mutation = AuthorizationMutation( kind=AuthorizationMutationKind.ROLE_ASSIGNMENT_CREATED, @@ -129,7 +200,8 @@ async def create_assignment( ) try: await session.flush() - await stage_identity_mutation(authorization_service, session, mutation) + if effective_assignment_created: + await stage_identity_mutation(authorization_service, session, mutation) await session.commit() except IntegrityError as exc: await session.rollback() @@ -137,7 +209,8 @@ async def create_assignment( status_code=status.HTTP_409_CONFLICT, detail="Assignment already exists for this user/role/domain", ) from exc - await safe_identity_mutation_committed(authorization_service, mutation) + if effective_assignment_created: + await safe_identity_mutation_committed(authorization_service, mutation) await session.refresh(assignment) await audit_decision( user_id=current_user.id, @@ -159,20 +232,65 @@ async def create_assignment( payload.domain_type, payload.domain_id, ) - return RoleAssignmentRead.model_validate(assignment) + return (await _assignment_reads(session, [assignment]))[0] -@router.delete("/{assignment_id}", status_code=status.HTTP_204_NO_CONTENT) +@router.delete( + "/{assignment_id}", + response_model=RoleAssignmentRead, + status_code=status.HTTP_200_OK, + responses={status.HTTP_204_NO_CONTENT: {"description": "Manual assignment fully revoked."}}, +) async def delete_assignment( assignment_id: UUID, current_user: CurrentActiveUser, session: DbSession, -) -> None: - """Revoke a role assignment. Superuser-only.""" +) -> RoleAssignmentRead | Response: + """Remove a manual grant, returning the assignment when another source preserves it.""" _require_superuser(current_user) assignment = await session.get(AuthzRoleAssignment, assignment_id) if assignment is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Assignment not found") + grants = ( + await session.exec( + select(AuthzRoleAssignmentGrant).where(AuthzRoleAssignmentGrant.assignment_id == assignment_id) + ) + ).all() + manual_grant = next((grant for grant in grants if grant.source_kind == "manual"), None) + if grants and manual_grant is None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="IdP-derived assignments cannot be deleted through the manual assignment API", + ) + if manual_grant is not None and len(grants) > 1: + surviving_grants = [grant for grant in grants if grant is not manual_grant] + await session.delete(manual_grant) + await session.commit() + await audit_decision( + user_id=current_user.id, + action="role_assignment:delete_manual_source", + obj=f"user:{assignment.user_id}", + result="allow", + details={ + "assignment_id": str(assignment_id), + "role_id": str(assignment.role_id), + "domain_type": assignment.domain_type, + "domain_id": str(assignment.domain_id) if assignment.domain_id else None, + "effective_assignment_preserved": True, + "surviving_grant_sources": [ + { + "source_kind": grant.source_kind, + "provider_id": grant.provider_id, + "external_group": grant.external_group, + } + for grant in surviving_grants + ], + }, + ) + return RoleAssignmentRead.model_validate(assignment).model_copy( + update={"grant_sources": [RoleAssignmentGrantSummary.model_validate(grant) for grant in surviving_grants]} + ) + user_id = assignment.user_id role_id = assignment.role_id domain_type = assignment.domain_type @@ -210,3 +328,4 @@ async def delete_assignment( }, ) logger.info("Revoked role assignment id=%s (user=%s)", assignment_id, user_id) + return Response(status_code=status.HTTP_204_NO_CONTENT) diff --git a/src/backend/base/langflow/api/v1/schemas/authz_role_assignments.py b/src/backend/base/langflow/api/v1/schemas/authz_role_assignments.py index 2d485a771c23..261c20fa3da6 100644 --- a/src/backend/base/langflow/api/v1/schemas/authz_role_assignments.py +++ b/src/backend/base/langflow/api/v1/schemas/authz_role_assignments.py @@ -48,8 +48,21 @@ def _check_domain_id_consistency(self) -> RoleAssignmentCreate: return self +class RoleAssignmentGrantSummary(BaseModel): + """Non-secret provenance returned with an effective role assignment.""" + + source_kind: Literal["manual", "idp"] + provider_id: str | None + external_group: str | None + administrative_actor: UUID | None + created_at: datetime + updated_at: datetime + + model_config = {"from_attributes": True} + + class RoleAssignmentRead(BaseModel): - """Serialized authz_role_assignment row returned by the API.""" + """Serialized effective assignment plus its independent grant sources.""" id: UUID user_id: UUID @@ -58,5 +71,6 @@ class RoleAssignmentRead(BaseModel): domain_id: UUID | None assigned_at: datetime assigned_by: UUID | None + grant_sources: list[RoleAssignmentGrantSummary] = Field(default_factory=list) model_config = {"from_attributes": True} diff --git a/src/backend/base/langflow/services/auth/service.py b/src/backend/base/langflow/services/auth/service.py index 0e4db398fad7..75d14c130d5c 100644 --- a/src/backend/base/langflow/services/auth/service.py +++ b/src/backend/base/langflow/services/auth/service.py @@ -2,7 +2,7 @@ import hashlib import warnings -from collections.abc import Coroutine +from collections.abc import Awaitable, Callable, Coroutine, Mapping from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING from uuid import UUID @@ -54,6 +54,60 @@ from langflow.services.deps import session_scope from langflow.services.schema import ServiceType +_MAX_EXTERNAL_AUTHORIZATION_GROUPS = 500 +_MAX_EXTERNAL_AUTHORIZATION_GROUP_LENGTH = 256 + + +def _has_external_group_overage(claims: Mapping[str, object], claim_name: str) -> bool: + """Return whether an Entra-style overage pointer replaces the group claim.""" + claim_names = claims.get("_claim_names") + return isinstance(claim_names, Mapping) and claim_name in claim_names + + +def _audit_audience(claims: Mapping[str, object]) -> str | list[str] | None: + """Normalize a verified audience claim to a JSON-safe audit value.""" + audience = claims.get("aud") + if isinstance(audience, str): + return audience + if isinstance(audience, (list, tuple)) and all(isinstance(value, str) for value in audience): + return list(audience) + return None + + +async def _safe_audit_directory_reconciliation( + audit: Callable[..., Awaitable[None]], + *, + identity: ExternalIdentity, + user: User, + issuer: str | None, + result: str, + details: dict[str, object], +) -> None: + """Record reconciliation without letting audit outages fail authentication.""" + try: + await audit( + user_id=user.id, + action="directory_membership:reconcile", + obj=f"user:{user.id}", + result=result, + details={ + "provider_id": identity.provider, + "issuer": issuer, + "subject": identity.subject, + "audience": _audit_audience(identity.claims), + "source": "external_bearer", + **details, + }, + ) + except Exception: # noqa: BLE001 + await logger.aexception( + "Authorization directory reconciliation audit failed for provider=%s user=%s result=%s", + identity.provider, + user.id, + result, + ) + + if TYPE_CHECKING: from cryptography.fernet import Fernet, MultiFernet from lfx.services.settings.service import SettingsService @@ -292,7 +346,188 @@ async def _authenticate_with_external_token(self, token: str, db: AsyncSession) AuthCredentialContext(method=AUTH_METHOD_EXTERNAL, external_provider=identity.provider) ) set_current_external_access_context(access_context_from_identity(identity, self.settings.auth_settings)) - return await self._materialize_external_user(identity, db) + user = await self._materialize_external_user(identity, db) + await self._reconcile_verified_external_groups(identity=identity, user=user, db=db) + return user + + async def _reconcile_verified_external_groups( + self, + *, + identity: ExternalIdentity, + user: User, + db: AsyncSession, + ) -> None: + """Send a complete verified group claim through the authorization seam. + + Missing, overage, or malformed claims are not authoritative zero-group + snapshots and therefore skip reconciliation. Provider configuration and + transactional ingest failures intentionally remain fail-closed; audit + enqueue and post-commit publication are isolated from authentication. + """ + from lfx.services.authorization import DirectoryMembershipSnapshot + + from langflow.services.authorization.audit import AUDIT_ALLOW, AUDIT_SKIP, audit_decision + from langflow.services.authorization.lifecycle import safe_directory_membership_committed + from langflow.services.deps import get_authorization_service + + authorization_service = get_authorization_service() + issuer_value = identity.claims.get("iss") + issuer = issuer_value.strip() if isinstance(issuer_value, str) and issuer_value.strip() else None + claim_name = await authorization_service.external_groups_claim( + provider_id=identity.provider, + issuer=issuer, + ) + if not claim_name: + return + + async def audit_reconciliation(*, result: str, details: dict[str, object]) -> None: + await _safe_audit_directory_reconciliation( + audit_decision, + identity=identity, + user=user, + issuer=issuer, + result=result, + details=details, + ) + + async def audit_skip(reason: str) -> None: + # JIT user/profile and last-login updates share this transaction. + # Commit them before the independent audit writer resolves the + # audit row's user foreign key. + await db.commit() + await audit_reconciliation( + result=AUDIT_SKIP, + details={ + "claim_name": claim_name, + "reason": reason, + "authoritative": False, + "complete": False, + }, + ) + + if _has_external_group_overage(identity.claims, claim_name): + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: claim=%s uses an overage pointer", + identity.provider, + user.id, + claim_name, + ) + await audit_skip("overage") + return + if claim_name not in identity.claims: + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: claim=%s is absent", + identity.provider, + user.id, + claim_name, + ) + await audit_skip("absent") + return + + raw_groups = identity.claims[claim_name] + if isinstance(raw_groups, str): + candidates = (raw_groups,) + elif isinstance(raw_groups, (list, tuple, set, frozenset)): + candidates = raw_groups + else: + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: claim=%s has an invalid type", + identity.provider, + user.id, + claim_name, + ) + await audit_skip("malformed") + return + + normalized_groups: set[str] = set() + for candidate in candidates: + if not isinstance(candidate, str): + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: " + "claim=%s contains a non-string entry", + identity.provider, + user.id, + claim_name, + ) + await audit_skip("malformed") + return + group = candidate.strip() + if not group or len(group) > _MAX_EXTERNAL_AUTHORIZATION_GROUP_LENGTH: + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: " + "claim=%s contains an invalid group identifier", + identity.provider, + user.id, + claim_name, + ) + await audit_skip("malformed") + return + normalized_groups.add(group) + + groups = tuple(sorted(normalized_groups)) + if len(groups) > _MAX_EXTERNAL_AUTHORIZATION_GROUPS: + logger.warning( + "Skipping external group reconciliation for provider=%s user=%s: claim=%s exceeds the %d-group limit", + identity.provider, + user.id, + claim_name, + _MAX_EXTERNAL_AUTHORIZATION_GROUPS, + ) + await audit_skip("too_many") + return + + result = await authorization_service.ingest_directory_membership_snapshot( + session=db, + snapshot=DirectoryMembershipSnapshot( + provider_id=identity.provider, + source="external_bearer", + observed_at=datetime.now(timezone.utc), + user_id=user.id, + provider_user_id=identity.subject, + memberships=groups, + authoritative=True, + complete=True, + ), + ) + await db.commit() + if result is None: + # Compatibility with a plugin built against the initial untyped + # seam: an unknown result must invalidate, never preserve stale + # policy by assuming nothing changed. + logger.warning( + "Authorization plugin returned no directory ingest result for provider=%s user=%s; " + "invalidating conservatively", + identity.provider, + user.id, + ) + changed = True + added = None + removed = None + else: + # The initial seam only documented ``changed`` through caller-side + # duck typing. Keep older plugin result objects safe after commit + # while the explicit result contract rolls out. + changed = bool(getattr(result, "changed", True)) + added = getattr(result, "added", None) + removed = getattr(result, "removed", None) + + await audit_reconciliation( + result=AUDIT_ALLOW, + details={ + "membership_count": len(groups), + "membership_sha256": hashlib.sha256("\0".join(groups).encode()).hexdigest(), + "changed": changed, + "added": added, + "removed": removed, + "authoritative": True, + "complete": True, + }, + ) + await safe_directory_membership_committed( + authorization_service, + user_id=user.id, + changed=changed, + ) async def _authenticate_with_api_key(self, api_key: str, db: AsyncSession) -> UserRead | None: """Internal method to authenticate with API key (raises generic exceptions). diff --git a/src/backend/base/langflow/services/authorization/audit.py b/src/backend/base/langflow/services/authorization/audit.py index 36e87416f6d3..b01de3f968eb 100644 --- a/src/backend/base/langflow/services/authorization/audit.py +++ b/src/backend/base/langflow/services/authorization/audit.py @@ -33,6 +33,7 @@ AUDIT_ALLOW = "allow" AUDIT_DENY = "deny" AUDIT_OWNER_OVERRIDE = "owner_override" +AUDIT_SKIP = "skip" AUDIT_ACTOR_API_KEY = "api_key" # pragma: allowlist secret AUDIT_ACTOR_UNKNOWN = "unknown" diff --git a/src/backend/base/langflow/services/authorization/lifecycle.py b/src/backend/base/langflow/services/authorization/lifecycle.py index 8a89ef252b0e..3666325f2ac3 100644 --- a/src/backend/base/langflow/services/authorization/lifecycle.py +++ b/src/backend/base/langflow/services/authorization/lifecycle.py @@ -7,6 +7,8 @@ from lfx.log.logger import logger if TYPE_CHECKING: + from uuid import UUID + from lfx.services.authorization.base import AuthorizationMutation, BaseAuthorizationService from sqlmodel.ext.asyncio.session import AsyncSession @@ -54,3 +56,19 @@ async def safe_identity_mutation_committed( mutation.kind.value, mutation.entity_id, ) + + +async def safe_directory_membership_committed( + service: BaseAuthorizationService, + *, + user_id: UUID, + changed: bool, +) -> None: + """Publish committed directory membership without misreporting durable success.""" + try: + await service.directory_membership_committed(user_id=user_id, changed=changed) + except Exception: # noqa: BLE001 + logger.exception( + "Authorization directory publication failed after commit for user=%s", + user_id, + ) diff --git a/src/backend/base/langflow/services/authorization/utils.py b/src/backend/base/langflow/services/authorization/utils.py index 64cb595825ac..2202479c4cbe 100644 --- a/src/backend/base/langflow/services/authorization/utils.py +++ b/src/backend/base/langflow/services/authorization/utils.py @@ -35,6 +35,9 @@ from langflow.services.authorization.audit import ( AUDIT_OWNER_OVERRIDE as _AUDIT_OWNER_OVERRIDE, ) +from langflow.services.authorization.audit import ( + AUDIT_SKIP as _AUDIT_SKIP, +) from langflow.services.authorization.guards import ( _ACTION_ENUMS, _OWNER_CONTEXT_KEYS, @@ -81,6 +84,7 @@ def permission_denied_to_http(exc): "_AUDIT_DENY", "_AUDIT_OWNER_OVERRIDE", "_AUDIT_QUEUE_MAX", + "_AUDIT_SKIP", "_OWNER_CONTEXT_KEYS", "_AuditEntry", "_audit_writer_loop", diff --git a/src/backend/base/langflow/services/database/models/auth/__init__.py b/src/backend/base/langflow/services/database/models/auth/__init__.py index 22623bdf4eae..1ef3e8d1ec9f 100644 --- a/src/backend/base/langflow/services/database/models/auth/__init__.py +++ b/src/backend/base/langflow/services/database/models/auth/__init__.py @@ -3,6 +3,7 @@ AuthzEditLock, AuthzRole, AuthzRoleAssignment, + AuthzRoleAssignmentGrant, AuthzShare, AuthzTeam, AuthzTeamMember, @@ -17,6 +18,7 @@ "AuthzEditLock", "AuthzRole", "AuthzRoleAssignment", + "AuthzRoleAssignmentGrant", "AuthzShare", "AuthzTeam", "AuthzTeamMember", diff --git a/src/backend/base/langflow/services/database/models/auth/authz.py b/src/backend/base/langflow/services/database/models/auth/authz.py index bb850b545028..e13ee2ebee0c 100644 --- a/src/backend/base/langflow/services/database/models/auth/authz.py +++ b/src/backend/base/langflow/services/database/models/auth/authz.py @@ -146,6 +146,64 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg] ) +class AuthzRoleAssignmentGrant(SQLModel, table=True): # type: ignore[call-arg] + """Independent provenance source that keeps an effective assignment alive. + + An assignment may have one manual source and any number of IdP group + sources. This avoids making provenance a lossy property of the effective + assignment when a manual grant overlaps an externally-derived grant. + """ + + __tablename__ = "authz_role_assignment_grant" + __table_args__ = ( + CheckConstraint( + "(source_kind = 'manual' AND provider_id IS NULL AND external_group IS NULL) " + "OR (source_kind = 'idp' AND provider_id IS NOT NULL AND external_group IS NOT NULL)", + name="ck_authz_role_assignment_grant_source", + ), + Index( + "uq_authz_role_assignment_grant_manual", + "assignment_id", + unique=True, + postgresql_where=text("source_kind = 'manual'"), + sqlite_where=text("source_kind = 'manual'"), + ), + Index( + "uq_authz_role_assignment_grant_idp", + "assignment_id", + "provider_id", + "external_group", + unique=True, + postgresql_where=text("source_kind = 'idp'"), + sqlite_where=text("source_kind = 'idp'"), + ), + Index( + "ix_authz_role_assignment_grant_provider_group", + "provider_id", + "external_group", + ), + ) + + id: UUIDstr = Field(default_factory=uuid4, primary_key=True) + assignment_id: UUIDstr = Field( + sa_column=Column( + sa.Uuid(), + ForeignKey("authz_role_assignment.id", ondelete="CASCADE"), + nullable=False, + index=True, + ), + ) + source_kind: str = Field(description="manual or idp") + provider_id: str | None = Field(default=None, max_length=256) + external_group: str | None = Field(default=None, max_length=256) + administrative_actor: UUIDstr | None = Field( + default=None, + sa_column=Column(sa.Uuid(), ForeignKey("user.id", ondelete="SET NULL"), nullable=True), + ) + created_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column()) + updated_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column()) + + class AuthzTeam(SQLModel, table=True): # type: ignore[call-arg] """Logical grouping of users for share scopes and bulk role assignments.""" @@ -282,12 +340,11 @@ class AuthzAuditLog(SQLModel, table=True): # type: ignore[call-arg] Index("ix_authz_audit_log_actor_timestamp", "actor_id", "timestamp"), Index("ix_authz_audit_log_actor_type_timestamp", "actor_type", "timestamp"), Index("ix_authz_audit_log_resource", "resource_type", "resource_id"), - # ``owner_override`` is the third value the framework writes (see - # ``_AUDIT_OWNER_OVERRIDE`` in services/authorization/utils.py); it - # must be in the CHECK set or owner-shortcut audit rows would - # silently fail the constraint. + # Keep this vocabulary aligned with ``services.authorization.audit``. + # ``skip`` records an operator-visible reconciliation attempt that did + # not apply an authoritative directory snapshot. CheckConstraint( - "result IN ('allow', 'deny', 'owner_override')", + "result IN ('allow', 'deny', 'owner_override', 'skip')", name="ck_authz_audit_log_result_enum", ), ) diff --git a/src/backend/tests/unit/alembic/test_allow_skipped_authz_audit_results.py b/src/backend/tests/unit/alembic/test_allow_skipped_authz_audit_results.py new file mode 100644 index 000000000000..1d294cdfabdf --- /dev/null +++ b/src/backend/tests/unit/alembic/test_allow_skipped_authz_audit_results.py @@ -0,0 +1,133 @@ +"""Tests for the skipped authorization-audit result expansion.""" + +from __future__ import annotations + +import importlib + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +_MIGRATION = importlib.import_module("langflow.alembic.versions.cp03a2b3c4d5_allow_skipped_authz_audit_results") + +_CONSTRAINT_NAME = "ck_authz_audit_log_result_enum" +_NAMING_CONVENTION = {"ck": "ck_%(table_name)s_%(constraint_name)s"} +_RENDERED_CONSTRAINT_NAME = f"ck_authz_audit_log_{_CONSTRAINT_NAME}" + + +def _make_engine(naming_convention): + engine = sa.create_engine("sqlite:///:memory:") + metadata = sa.MetaData(naming_convention=naming_convention) + sa.Table( + "authz_audit_log", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("result", sa.String(), nullable=False), + sa.CheckConstraint( + "result IN ('allow', 'deny', 'owner_override')", + name=_CONSTRAINT_NAME, + ), + ) + metadata.create_all(engine) + return engine + + +def _run_upgrade(engine) -> None: + with engine.begin() as conn: + original_op = _MIGRATION.op + try: + _MIGRATION.op = Operations(MigrationContext.configure(conn)) + _MIGRATION.upgrade() + finally: + _MIGRATION.op = original_op + + +@pytest.mark.parametrize( + ("naming_convention", "expected_constraint_name"), + [ + pytest.param(None, _CONSTRAINT_NAME, id="literal-name"), + pytest.param(_NAMING_CONVENTION, _RENDERED_CONSTRAINT_NAME, id="langflow-convention"), + ], +) +def test_upgrade_allows_skip_without_weakening_other_result_validation(naming_convention, expected_constraint_name): + engine = _make_engine(naming_convention) + audit_log = sa.table("authz_audit_log", sa.column("result", sa.String())) + + _run_upgrade(engine) + _run_upgrade(engine) + + with engine.begin() as conn: + conn.execute(sa.insert(audit_log).values(result="skip")) + + with pytest.raises(sa.exc.IntegrityError), engine.begin() as conn: + conn.execute(sa.insert(audit_log).values(result="unexpected")) + + result_checks = [ + check + for check in sa.inspect(engine).get_check_constraints("authz_audit_log") + if "result IN" in (check.get("sqltext") or "") + ] + assert len(result_checks) == 1 + result_check = result_checks[0] + assert result_check["name"] == expected_constraint_name + assert "'skip'" in result_check["sqltext"] + + _MIGRATION.downgrade() + with engine.begin() as conn: + conn.execute(sa.insert(audit_log).values(result="skip")) + + +def test_upgrade_reuses_convention_rendered_constraint_name(monkeypatch): + class FakeBatchOperations: + def __init__(self): + self.dropped = [] + self.created = [] + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def f(self, name): + return f"fixed:{name}" + + def drop_constraint(self, name, *, type_): + self.dropped.append((name, type_)) + + def create_check_constraint(self, name, condition): + self.created.append((name, condition)) + + class FakeOperations: + def __init__(self, batch): + self.batch = batch + + def get_bind(self): + return object() + + def batch_alter_table(self, table_name, *, schema): + assert table_name == "authz_audit_log" + assert schema is None + return self.batch + + checks = [ + { + "name": _RENDERED_CONSTRAINT_NAME, + "sqltext": "result IN ('allow', 'deny', 'owner_override')", + } + ] + inspector = type("Inspector", (), {"get_check_constraints": lambda _self, _table: checks})() + batch = FakeBatchOperations() + monkeypatch.setattr(_MIGRATION.sa, "inspect", lambda _conn: inspector) + monkeypatch.setattr(_MIGRATION, "op", FakeOperations(batch)) + + _MIGRATION.upgrade() + + fixed_name = f"fixed:{_RENDERED_CONSTRAINT_NAME}" + assert batch.dropped == [(fixed_name, "check")] + assert batch.created == [(fixed_name, _MIGRATION._RESULT_CHECK)] + + +def test_migration_follows_role_grant_backfill(): + assert _MIGRATION.down_revision == "cp02a2b3c4d5" diff --git a/src/backend/tests/unit/alembic/test_backfill_manual_role_grant_sources.py b/src/backend/tests/unit/alembic/test_backfill_manual_role_grant_sources.py new file mode 100644 index 000000000000..f2e2be4dfeb7 --- /dev/null +++ b/src/backend/tests/unit/alembic/test_backfill_manual_role_grant_sources.py @@ -0,0 +1,117 @@ +"""Tests for the manual role-assignment grant-source backfill.""" + +from __future__ import annotations + +import importlib +import types +from datetime import datetime, timedelta, timezone +from uuid import UUID + +import sqlalchemy as sa + +_MIGRATION = importlib.import_module("langflow.alembic.versions.cp02a2b3c4d5_backfill_manual_role_grant_sources") + + +def _make_tables(): + engine = sa.create_engine("sqlite:///:memory:") + metadata = sa.MetaData() + assignment = sa.Table( + "authz_role_assignment", + metadata, + sa.Column("id", sa.Uuid(), primary_key=True), + sa.Column("assigned_by", sa.Uuid(), nullable=True), + sa.Column("assigned_at", sa.DateTime(timezone=True), nullable=False), + ) + grant = sa.Table( + "authz_role_assignment_grant", + metadata, + sa.Column("id", sa.Uuid(), primary_key=True), + sa.Column("assignment_id", sa.Uuid(), nullable=False), + sa.Column("source_kind", sa.String(), nullable=False), + sa.Column("provider_id", sa.String(), nullable=True), + sa.Column("external_group", sa.String(), nullable=True), + sa.Column("administrative_actor", sa.Uuid(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + metadata.create_all(engine) + return engine, assignment, grant + + +def _run_upgrade(engine) -> None: + with engine.begin() as conn: + original_op = _MIGRATION.op + try: + _MIGRATION.op = types.SimpleNamespace(get_bind=lambda: conn) + _MIGRATION.upgrade() + finally: + _MIGRATION.op = original_op + + +def test_upgrade_backfills_deterministic_batches_without_replacing_existing_sources(monkeypatch): + engine, assignment, grant = _make_tables() + monkeypatch.setattr(_MIGRATION, "_BATCH_SIZE", 2) + assignment_ids = [UUID(int=value) for value in range(1, 7)] + actors = [UUID(int=100 + value) for value in range(1, 7)] + assigned_at = [datetime(2026, 7, 30, 12, tzinfo=timezone.utc) + timedelta(minutes=value) for value in range(1, 7)] + + with engine.begin() as conn: + conn.execute( + assignment.insert(), + [ + { + "id": assignment_id, + "assigned_by": actor, + "assigned_at": timestamp, + } + for assignment_id, actor, timestamp in zip(assignment_ids, actors, assigned_at, strict=True) + ], + ) + conn.execute( + grant.insert(), + { + "id": UUID(int=1000), + "assignment_id": assignment_ids[2], + "source_kind": "idp", + "provider_id": "example-idp", + "external_group": "engineering", + "administrative_actor": None, + "created_at": assigned_at[2], + "updated_at": assigned_at[2], + }, + ) + + batch_sizes: list[int] = [] + + def record_grant_insert(_conn, _cursor, statement, parameters, _context, executemany): + if statement.lstrip().startswith("INSERT INTO authz_role_assignment_grant"): + batch_sizes.append(len(parameters) if executemany else 1) + + sa.event.listen(engine, "before_cursor_execute", record_grant_insert) + + _run_upgrade(engine) + _run_upgrade(engine) + + with engine.connect() as conn: + rows = conn.execute(sa.select(grant).order_by(grant.c.assignment_id)).mappings().all() + + assert batch_sizes == [2, 2, 1] + assert len(rows) == len(assignment_ids) + assert rows[2]["source_kind"] == "idp" + assert rows[2]["provider_id"] == "example-idp" + assert rows[2]["external_group"] == "engineering" + + manual_rows = [row for row in rows if row["source_kind"] == "manual"] + expected = { + assignment_id: (actor, timestamp.replace(tzinfo=None)) + for assignment_id, actor, timestamp in zip(assignment_ids, actors, assigned_at, strict=True) + if assignment_id != assignment_ids[2] + } + assert {row["assignment_id"] for row in manual_rows} == expected.keys() + for row in manual_rows: + actor, timestamp = expected[row["assignment_id"]] + assert row["provider_id"] is None + assert row["external_group"] is None + assert row["administrative_actor"] == actor + assert row["created_at"] == timestamp + assert row["updated_at"] == timestamp diff --git a/src/backend/tests/unit/api/v1/test_authz_admin_routes.py b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py index 717f331ca735..1654a3a90964 100644 --- a/src/backend/tests/unit/api/v1/test_authz_admin_routes.py +++ b/src/backend/tests/unit/api/v1/test_authz_admin_routes.py @@ -775,6 +775,38 @@ async def test_list_assignments_no_user_id_defaults_to_self(stub_authz): assert result == [] +@pytest.mark.asyncio +async def test_list_assignments_serializes_grant_sources(stub_authz): + """Assignment reads include every durable provenance source.""" + from langflow.api.v1 import authz_role_assignments + from langflow.services.database.models.auth import AuthzRoleAssignment, AuthzRoleAssignmentGrant + + stub_authz() + user = _make_user(is_superuser=False) + assignment = AuthzRoleAssignment(user_id=user.id, role_id=uuid4(), assigned_by=user.id) + manual_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=user.id, + ) + idp_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="idp", + provider_id="entra", + external_group="corp-dev", + ) + session = _FakeAsyncSession(exec_results=[[assignment], [manual_grant, idp_grant]]) + + result = await authz_role_assignments.list_assignments(session=session, current_user=user) + + assert len(result) == 1 + sources = {source.source_kind: source for source in result[0].grant_sources} + assert sources.keys() == {"manual", "idp"} + assert sources["manual"].administrative_actor == user.id + assert sources["idp"].provider_id == "entra" + assert sources["idp"].external_group == "corp-dev" + + @pytest.mark.asyncio async def test_create_assignment_invalid_user_404(stub_authz): from langflow.api.v1 import authz_role_assignments @@ -816,7 +848,9 @@ async def test_create_assignment_emits_lifecycle_for_target_user(stub_authz): current_user=actor, session=session, ) - assert len(session.added) == 1 + assert len(session.added) == 2 + assert session.added[1].source_kind == "manual" + assert session.added[1].assignment_id == session.added[0].id assert session.committed == 1 assert authz.staged_mutations == authz.committed_mutations assert authz.staged_mutations[0].affected_user_ids == (target_user.id,) @@ -824,6 +858,249 @@ async def test_create_assignment_emits_lifecycle_for_target_user(stub_authz): assert authz.staged_mutations[0].domain_id is None +@pytest.mark.asyncio +async def test_create_assignment_duplicate_manual_source_is_409(stub_authz): + from langflow.api.v1 import authz_role_assignments + from langflow.api.v1.schemas.authz_role_assignments import RoleAssignmentCreate + from langflow.services.database.models.auth import AuthzRole, AuthzRoleAssignment, AuthzRoleAssignmentGrant + from langflow.services.database.models.user.model import User + + authz = stub_authz() + actor = _make_user(is_superuser=True) + target_user = SimpleNamespace(id=uuid4()) + role = SimpleNamespace(id=uuid4(), name="viewer") + assignment = AuthzRoleAssignment(user_id=target_user.id, role_id=role.id, assigned_by=actor.id) + manual_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=actor.id, + ) + session = _FakeAsyncSession( + {(User, target_user.id): target_user, (AuthzRole, role.id): role}, + exec_results=[[assignment], [manual_grant]], + ) + + with pytest.raises(HTTPException) as excinfo: + await authz_role_assignments.create_assignment( + payload=RoleAssignmentCreate(user_id=target_user.id, role_id=role.id), + current_user=actor, + session=session, + ) + + assert excinfo.value.status_code == 409 + assert "Manual assignment already exists" in excinfo.value.detail + assert session.added == [] + assert session.committed == 0 + assert authz.staged_mutations == [] + + +@pytest.mark.asyncio +async def test_create_assignment_adds_manual_source_to_idp_assignment_without_lifecycle_mutation(stub_authz): + from langflow.api.v1 import authz_role_assignments + from langflow.api.v1.schemas.authz_role_assignments import RoleAssignmentCreate + from langflow.services.database.models.auth import AuthzRole, AuthzRoleAssignment, AuthzRoleAssignmentGrant + from langflow.services.database.models.user.model import User + + authz = stub_authz() + actor = _make_user(is_superuser=True) + target_user = SimpleNamespace(id=uuid4()) + role = SimpleNamespace(id=uuid4(), name="viewer") + original_actor_id = uuid4() + assignment = AuthzRoleAssignment( + user_id=target_user.id, + role_id=role.id, + assigned_by=original_actor_id, + ) + idp_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="idp", + provider_id="entra", + external_group="corp-dev", + ) + persisted_manual = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=actor.id, + ) + session = _FakeAsyncSession( + {(User, target_user.id): target_user, (AuthzRole, role.id): role}, + exec_results=[[assignment], [], [idp_grant, persisted_manual]], + ) + + result = await authz_role_assignments.create_assignment( + payload=RoleAssignmentCreate(user_id=target_user.id, role_id=role.id), + current_user=actor, + session=session, + ) + + assert len(session.added) == 1 + assert session.added[0].source_kind == "manual" + assert session.added[0].assignment_id == assignment.id + assert session.committed == 1 + assert authz.staged_mutations == [] + assert authz.committed_mutations == [] + assert result.id == assignment.id + assert result.assigned_by == original_actor_id + assert {source.source_kind for source in result.grant_sources} == {"idp", "manual"} + + +@pytest.mark.asyncio +async def test_delete_assignment_rejects_idp_only_source(stub_authz): + from langflow.api.v1 import authz_role_assignments + from langflow.services.database.models.auth import AuthzRoleAssignment, AuthzRoleAssignmentGrant + + authz = stub_authz() + actor = _make_user(is_superuser=True) + assignment = AuthzRoleAssignment(user_id=uuid4(), role_id=uuid4()) + idp_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="idp", + provider_id="entra", + external_group="corp-dev", + ) + session = _FakeAsyncSession( + {(AuthzRoleAssignment, assignment.id): assignment}, + exec_results=[[idp_grant]], + ) + + with pytest.raises(HTTPException) as excinfo: + await authz_role_assignments.delete_assignment( + assignment_id=assignment.id, + current_user=actor, + session=session, + ) + + assert excinfo.value.status_code == 409 + assert "IdP-derived assignments" in excinfo.value.detail + assert session.deleted == [] + assert session.committed == 0 + assert authz.staged_mutations == [] + + +@pytest.mark.asyncio +async def test_delete_assignment_returns_surviving_idp_assignment(stub_authz, monkeypatch): + from langflow.api.v1 import authz_role_assignments + from langflow.services.database.models.auth import AuthzRoleAssignment, AuthzRoleAssignmentGrant + + authz = stub_authz() + actor = _make_user(is_superuser=True) + domain_id = uuid4() + assignment = AuthzRoleAssignment( + user_id=uuid4(), + role_id=uuid4(), + domain_type="workspace", + domain_id=domain_id, + assigned_by=actor.id, + ) + manual_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=actor.id, + ) + idp_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="idp", + provider_id="entra", + external_group="corp-dev", + ) + session = _FakeAsyncSession( + {(AuthzRoleAssignment, assignment.id): assignment}, + exec_results=[[manual_grant, idp_grant]], + ) + audit_calls: list[dict[str, Any]] = [] + + async def capture_audit(**kwargs): + audit_calls.append(kwargs) + + monkeypatch.setattr(authz_role_assignments, "audit_decision", capture_audit) + + result = await authz_role_assignments.delete_assignment( + assignment_id=assignment.id, + current_user=actor, + session=session, + ) + + delete_route = next(route for route in authz_role_assignments.router.routes if "DELETE" in route.methods) + assert delete_route.status_code == 200 + assert result.id == assignment.id + assert len(result.grant_sources) == 1 + assert result.grant_sources[0].source_kind == "idp" + assert result.grant_sources[0].provider_id == "entra" + assert result.grant_sources[0].external_group == "corp-dev" + assert session.deleted == [manual_grant] + assert session.committed == 1 + assert authz.staged_mutations == [] + assert audit_calls[0]["action"] == "role_assignment:delete_manual_source" + assert audit_calls[0]["details"] == { + "assignment_id": str(assignment.id), + "role_id": str(assignment.role_id), + "domain_type": "workspace", + "domain_id": str(domain_id), + "effective_assignment_preserved": True, + "surviving_grant_sources": [ + { + "source_kind": "idp", + "provider_id": "entra", + "external_group": "corp-dev", + } + ], + } + + +@pytest.mark.asyncio +async def test_delete_assignment_manual_only_returns_204(stub_authz): + from langflow.api.v1 import authz_role_assignments + from langflow.services.database.models.auth import AuthzRoleAssignment, AuthzRoleAssignmentGrant + + authz = stub_authz() + actor = _make_user(is_superuser=True) + assignment = AuthzRoleAssignment(user_id=uuid4(), role_id=uuid4()) + manual_grant = AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=actor.id, + ) + session = _FakeAsyncSession( + {(AuthzRoleAssignment, assignment.id): assignment}, + exec_results=[[manual_grant]], + ) + + result = await authz_role_assignments.delete_assignment( + assignment_id=assignment.id, + current_user=actor, + session=session, + ) + + assert result.status_code == 204 + assert session.deleted == [assignment] + assert authz.staged_mutations == authz.committed_mutations + + +@pytest.mark.asyncio +async def test_delete_legacy_assignment_without_grant_rows_returns_204(stub_authz): + from langflow.api.v1 import authz_role_assignments + from langflow.services.database.models.auth import AuthzRoleAssignment + + authz = stub_authz() + actor = _make_user(is_superuser=True) + assignment = AuthzRoleAssignment(user_id=uuid4(), role_id=uuid4()) + session = _FakeAsyncSession( + {(AuthzRoleAssignment, assignment.id): assignment}, + exec_results=[[]], + ) + + result = await authz_role_assignments.delete_assignment( + assignment_id=assignment.id, + current_user=actor, + session=session, + ) + + assert result.status_code == 204 + assert session.deleted == [assignment] + assert session.committed == 1 + assert authz.staged_mutations == authz.committed_mutations + + # ===================================================================== # /authz/teams # ===================================================================== diff --git a/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py b/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py index 30c6022990a9..3e9bb83926c4 100644 --- a/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py +++ b/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py @@ -166,6 +166,46 @@ async def test_audit_query_filters_and_returns_first_class_actor_fields(): assert "ORDER BY authz_audit_log.timestamp DESC, authz_audit_log.id DESC" in page_sql +@pytest.mark.anyio +async def test_audit_query_filters_skipped_reconciliation_results(): + from langflow.api.v1.authz_audit import list_audit_log + + row = SimpleNamespace( + id=uuid4(), + user_id=uuid4(), + actor_type="user", + actor_id=uuid4(), + action="directory_membership:reconcile", + resource_type="user", + resource_id=uuid4(), + result="skip", + details={"reason": "overage"}, + timestamp=datetime.now(timezone.utc), + ) + session = _Session(row) + + result = await list_audit_log( + session=session, + _admin=SimpleNamespace(), + user_id=None, + actor_type=None, + actor_id=None, + resource_type=None, + resource_id=None, + action="directory_membership:reconcile", + result="skip", + since=None, + until=None, + page=1, + size=50, + ) + + assert result.items[0].result == "skip" + assert result.items[0].details == {"reason": "overage"} + assert "authz_audit_log.action" in str(session.statements[0]) + assert "authz_audit_log.result" in str(session.statements[0]) + + @pytest.mark.anyio async def test_unknown_actor_filter_includes_legacy_null_and_explicit_unknown_rows(): from langflow.api.v1.authz_audit import list_audit_log diff --git a/src/backend/tests/unit/api/v1/test_authz_lifecycle_contract.py b/src/backend/tests/unit/api/v1/test_authz_lifecycle_contract.py index 189c2dadc766..c087ad392692 100644 --- a/src/backend/tests/unit/api/v1/test_authz_lifecycle_contract.py +++ b/src/backend/tests/unit/api/v1/test_authz_lifecycle_contract.py @@ -336,6 +336,8 @@ async def test_assignment_delete_validates_live_row_before_mutation_and_stage(mo ) session = SimpleNamespace() session.get = AsyncMock(return_value=assignment) + grant_result = SimpleNamespace(all=list) + session.exec = AsyncMock(return_value=grant_result) session.delete = AsyncMock(side_effect=lambda _row: events.append("mutate")) session.flush = AsyncMock(side_effect=lambda: events.append("flush")) session.commit = AsyncMock(side_effect=lambda: events.append("commit")) @@ -376,6 +378,8 @@ async def test_assignment_delete_policy_rejection_is_409_without_mutation(monkey ) session = SimpleNamespace() session.get = AsyncMock(return_value=assignment) + grant_result = SimpleNamespace(all=list) + session.exec = AsyncMock(return_value=grant_result) session.delete = AsyncMock() session.flush = AsyncMock() session.commit = AsyncMock() diff --git a/src/backend/tests/unit/services/auth/test_auth_service.py b/src/backend/tests/unit/services/auth/test_auth_service.py index 46c1e99ce7cd..f45330997c69 100644 --- a/src/backend/tests/unit/services/auth/test_auth_service.py +++ b/src/backend/tests/unit/services/auth/test_auth_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib from contextlib import asynccontextmanager from datetime import datetime, timedelta, timezone from types import SimpleNamespace @@ -1061,6 +1062,297 @@ async def test_materialize_external_user_preserves_email_when_token_omits_it( assert profile.email == "alice2@example.com" +# ============================================================================= +# Verified external-group reconciliation +# ============================================================================= + + +class _DirectoryAuthorizationStub: + """Authorization seam double for verified external-group reconciliation.""" + + def __init__(self, *, result, claim_name: str | None = "groups") -> None: + self.external_groups_claim = AsyncMock(return_value=claim_name) + self.ingest_directory_membership_snapshot = AsyncMock(return_value=result) + self.directory_membership_committed = AsyncMock() + + +def _external_identity(claims: dict): + from langflow.services.auth.external import ExternalIdentity + + return ExternalIdentity( + provider="customer-idp", + subject="external-subject", + username="external-user", + claims=claims, + ) + + +@pytest.mark.anyio +async def test_external_group_reconciliation_normalizes_commits_and_audits(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult(changed=True, added=2, removed=1)) + audit = AsyncMock() + identity = _external_identity( + { + "iss": " https://issuer.example ", + "aud": ["langflow-api"], + "groups": [" reviewers ", "engineering", "reviewers"], + } + ) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=audit), + ): + await auth_service._reconcile_verified_external_groups(identity=identity, user=user, db=db) + + snapshot = authz.ingest_directory_membership_snapshot.await_args.kwargs["snapshot"] + assert snapshot.provider_id == "customer-idp" + assert snapshot.provider_user_id == "external-subject" + assert snapshot.user_id == user.id + assert snapshot.memberships == ("engineering", "reviewers") + assert snapshot.authoritative is True + assert snapshot.complete is True + db.commit.assert_awaited_once() + authz.directory_membership_committed.assert_awaited_once_with(user_id=user.id, changed=True) + + audit_details = audit.await_args.kwargs + assert audit_details["action"] == "directory_membership:reconcile" + assert audit_details["obj"] == f"user:{user.id}" + assert audit_details["result"] == "allow" + assert audit_details["details"] == { + "provider_id": "customer-idp", + "issuer": "https://issuer.example", + "subject": "external-subject", + "audience": ["langflow-api"], + "source": "external_bearer", + "membership_count": 2, + "membership_sha256": hashlib.sha256(b"engineering\0reviewers").hexdigest(), + "changed": True, + "added": 2, + "removed": 1, + "authoritative": True, + "complete": True, + } + + +@pytest.mark.anyio +async def test_external_group_reconciliation_accepts_present_empty_claim(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult()) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=AsyncMock()), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({"groups": []}), + user=user, + db=db, + ) + + snapshot = authz.ingest_directory_membership_snapshot.await_args.kwargs["snapshot"] + assert snapshot.memberships == () + db.commit.assert_awaited_once() + authz.directory_membership_committed.assert_awaited_once_with(user_id=user.id, changed=False) + + +@pytest.mark.parametrize( + ("claims", "expected_reason"), + [ + ({"iss": "https://issuer.example"}, "absent"), + ( + {"_claim_names": {"groups": "src1"}, "_claim_sources": {"src1": {"endpoint": "https://graph"}}}, + "overage", + ), + ({"groups": {"unexpected": "mapping"}}, "malformed"), + ({"groups": ["engineering", 7]}, "malformed"), + ({"groups": [" "]}, "malformed"), + ({"groups": ["x" * 257]}, "malformed"), + ({"groups": [f"group-{index}" for index in range(501)]}, "too_many"), + ], + ids=["absent", "entra-overage", "invalid-type", "non-string", "blank", "overlong", "too-many"], +) +@pytest.mark.anyio +async def test_incomplete_external_group_claim_skips_authoritative_reconciliation( + auth_service: AuthService, + claims: dict, + expected_reason: str, +): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult(changed=True)) + events: list[str] = [] + db.commit.side_effect = lambda: events.append("commit") + audit = AsyncMock(side_effect=lambda **_kwargs: events.append("audit")) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=audit), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity(claims), + user=user, + db=db, + ) + + authz.ingest_directory_membership_snapshot.assert_not_awaited() + db.commit.assert_awaited_once() + authz.directory_membership_committed.assert_not_awaited() + audit.assert_awaited_once() + assert events == ["commit", "audit"] + audit_call = audit.await_args.kwargs + assert audit_call["action"] == "directory_membership:reconcile" + assert audit_call["obj"] == f"user:{user.id}" + assert audit_call["result"] == "skip" + assert audit_call["details"] == { + "provider_id": "customer-idp", + "issuer": claims.get("iss"), + "subject": "external-subject", + "audience": None, + "source": "external_bearer", + "claim_name": "groups", + "reason": expected_reason, + "authoritative": False, + "complete": False, + } + + +@pytest.mark.anyio +async def test_external_group_reconciliation_plugin_opt_out_is_not_a_skip(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub( + result=DirectoryMembershipIngestResult(changed=True), + claim_name=None, + ) + audit = AsyncMock() + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=audit), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({}), + user=user, + db=db, + ) + + authz.ingest_directory_membership_snapshot.assert_not_awaited() + db.commit.assert_not_awaited() + audit.assert_not_awaited() + + +@pytest.mark.parametrize( + "legacy_result", + [None, SimpleNamespace(changed=True)], + ids=["none", "changed-only"], +) +@pytest.mark.anyio +async def test_legacy_directory_ingest_result_invalidates_conservatively( + auth_service: AuthService, + legacy_result, +): + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=legacy_result) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=AsyncMock()), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({"groups": ["engineering"]}), + user=user, + db=db, + ) + + db.commit.assert_awaited_once() + authz.directory_membership_committed.assert_awaited_once_with(user_id=user.id, changed=True) + + +@pytest.mark.anyio +async def test_directory_post_commit_failure_does_not_fail_authentication(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult(changed=True, added=1)) + authz.directory_membership_committed.side_effect = RuntimeError("replica unavailable") + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=AsyncMock()), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({"groups": ["engineering"]}), + user=user, + db=db, + ) + + db.commit.assert_awaited_once() + authz.directory_membership_committed.assert_awaited_once_with(user_id=user.id, changed=True) + + +@pytest.mark.anyio +async def test_directory_post_commit_audit_failure_does_not_fail_authentication(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult(changed=True, added=1)) + audit = AsyncMock(side_effect=RuntimeError("audit settings unavailable")) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=audit), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({"groups": ["engineering"]}), + user=user, + db=db, + ) + + db.commit.assert_awaited_once() + audit.assert_awaited_once() + authz.directory_membership_committed.assert_awaited_once_with(user_id=user.id, changed=True) + + +@pytest.mark.anyio +async def test_directory_skip_audit_failure_does_not_fail_authentication(auth_service: AuthService): + from lfx.services.authorization import DirectoryMembershipIngestResult + + user = _dummy_user(uuid4()) + db = AsyncMock() + authz = _DirectoryAuthorizationStub(result=DirectoryMembershipIngestResult(changed=True)) + audit = AsyncMock(side_effect=RuntimeError("audit settings unavailable")) + + with ( + patch("langflow.services.deps.get_authorization_service", return_value=authz), + patch("langflow.services.authorization.audit.audit_decision", new=audit), + ): + await auth_service._reconcile_verified_external_groups( + identity=_external_identity({}), + user=user, + db=db, + ) + + authz.ingest_directory_membership_snapshot.assert_not_awaited() + db.commit.assert_awaited_once() + audit.assert_awaited_once() + authz.directory_membership_committed.assert_not_awaited() + + # ============================================================================= # External fallback (F2/F14): a valid external credential is tried when native fails # ============================================================================= diff --git a/src/backend/tests/unit/test_authz_models.py b/src/backend/tests/unit/test_authz_models.py index c02e530fd80b..70456888fc50 100644 --- a/src/backend/tests/unit/test_authz_models.py +++ b/src/backend/tests/unit/test_authz_models.py @@ -11,6 +11,7 @@ AuthzEditLock, AuthzRole, AuthzRoleAssignment, + AuthzRoleAssignmentGrant, AuthzShare, AuthzTeam, AuthzTeamMember, @@ -106,6 +107,66 @@ async def test_authz_role_assignment_persists(authz_async_session: AsyncSession) assert stored.domain_id is None +@pytest.mark.anyio +async def test_role_assignment_grants_preserve_independent_manual_and_idp_sources( + authz_async_session: AsyncSession, +): + user = User(username="sourced_assignee", password=_TEST_PASSWORD) + actor = User(username="sourced_actor", password=_TEST_PASSWORD) + role = AuthzRole(name="sourced_editor", permissions=["flow:write"]) + authz_async_session.add_all([user, actor, role]) + await authz_async_session.commit() + + assignment = AuthzRoleAssignment(user_id=user.id, role_id=role.id) + authz_async_session.add(assignment) + await authz_async_session.flush() + authz_async_session.add_all( + [ + AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="manual", + administrative_actor=actor.id, + ), + AuthzRoleAssignmentGrant( + assignment_id=assignment.id, + source_kind="idp", + provider_id="customer-idp", + external_group="corp_dev", + ), + ] + ) + await authz_async_session.commit() + + grants = ( + await authz_async_session.exec( + select(AuthzRoleAssignmentGrant).where(AuthzRoleAssignmentGrant.assignment_id == assignment.id) + ) + ).all() + assert {(grant.source_kind, grant.provider_id, grant.external_group) for grant in grants} == { + ("manual", None, None), + ("idp", "customer-idp", "corp_dev"), + } + + +@pytest.mark.anyio +async def test_role_assignment_grant_rejects_duplicate_manual_source(authz_async_session: AsyncSession): + from sqlalchemy.exc import IntegrityError + + user = User(username="duplicate_source_user", password=_TEST_PASSWORD) + role = AuthzRole(name="duplicate_source_role", permissions=[]) + authz_async_session.add_all([user, role]) + await authz_async_session.commit() + assignment = AuthzRoleAssignment(user_id=user.id, role_id=role.id) + authz_async_session.add(assignment) + await authz_async_session.flush() + authz_async_session.add(AuthzRoleAssignmentGrant(assignment_id=assignment.id, source_kind="manual")) + await authz_async_session.commit() + authz_async_session.add(AuthzRoleAssignmentGrant(assignment_id=assignment.id, source_kind="manual")) + with pytest.raises(IntegrityError): + await authz_async_session.commit() + await authz_async_session.rollback() + + @pytest.mark.anyio async def test_authz_role_assignment_blocks_duplicate_global(authz_async_session: AsyncSession): """Two global assignments with the same (user_id, role_id) must conflict. @@ -619,6 +680,24 @@ async def test_authz_audit_log_persists(authz_async_session: AsyncSession): assert stored.timestamp is not None +@pytest.mark.anyio +async def test_authz_audit_log_accepts_skipped_reconciliation(authz_async_session: AsyncSession): + """Skipped reconciliation is a first-class, queryable audit outcome.""" + entry = AuthzAuditLog( + action="directory_membership:reconcile", + resource_type="user", + result="skip", + details={"reason": "overage"}, + ) + authz_async_session.add(entry) + await authz_async_session.commit() + + stored = (await authz_async_session.exec(select(AuthzAuditLog).where(AuthzAuditLog.id == entry.id))).first() + assert stored is not None + assert stored.result == "skip" + assert stored.details == {"reason": "overage"} + + def test_authz_audit_actor_identity_has_no_fk_and_composite_timestamp_index(): """Credential attribution survives API-key deletion and supports actor history scans.""" table = AuthzAuditLog.__table__ diff --git a/src/lfx/src/lfx/services/authorization/__init__.py b/src/lfx/src/lfx/services/authorization/__init__.py index 2769f923a017..637b2e63d79a 100644 --- a/src/lfx/src/lfx/services/authorization/__init__.py +++ b/src/lfx/src/lfx/services/authorization/__init__.py @@ -5,6 +5,7 @@ AuthorizationMutationKind, AuthorizationMutationRejected, BaseAuthorizationService, + DirectoryMembershipIngestResult, DirectoryMembershipSnapshot, ResourceVisibilityScope, ShareRuleSnapshot, @@ -18,6 +19,7 @@ "AuthorizationMutationRejected", "AuthorizationService", "BaseAuthorizationService", + "DirectoryMembershipIngestResult", "DirectoryMembershipSnapshot", "ResourceVisibilityScope", "ShareRuleSnapshot", diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index b318807931f9..e7e4fd504a47 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -145,6 +145,15 @@ class DirectoryMembershipSnapshot: complete: bool = True +@dataclass(frozen=True, slots=True) +class DirectoryMembershipIngestResult: + """Outcome of ingesting a provider directory-membership snapshot.""" + + changed: bool = False + added: int = 0 + removed: int = 0 + + @dataclass(frozen=True, slots=True) class ResourceVisibilityScope: """Compact SQL-prefilter contract for resource-list authorization. @@ -441,13 +450,35 @@ async def ingest_directory_membership_snapshot( *, session: Any, snapshot: DirectoryMembershipSnapshot, - ) -> None: + ) -> DirectoryMembershipIngestResult: """Ingest one complete provider snapshot in the caller's transaction. The base implementation is intentionally inert. Directory polling and provider-specific pagination remain plugin responsibilities. """ _ = (session, snapshot) + return DirectoryMembershipIngestResult() + + async def external_groups_claim( + self, + *, + provider_id: str, + issuer: str | None, + ) -> str | None: + """Return the configured claim to normalize after external verification. + + The authentication service calls this only after it has verified the + external credential. It normalizes that one claim to string group + identifiers before invoking ``ingest_directory_membership_snapshot``; + raw tokens and raw claim dictionaries never cross this plugin seam. + """ + _ = (provider_id, issuer) + return None + + async def directory_membership_committed(self, *, user_id: UUID, changed: bool = True) -> None: + """Publish a committed membership change to authorization replicas.""" + if changed: + await self.invalidate_user(user_id) async def teardown(self) -> None: """No resources to release in the base implementation.""" diff --git a/src/lfx/tests/unit/services/authorization/test_default_authorization_service.py b/src/lfx/tests/unit/services/authorization/test_default_authorization_service.py index 3f4beb5e9098..f2701c5040ac 100644 --- a/src/lfx/tests/unit/services/authorization/test_default_authorization_service.py +++ b/src/lfx/tests/unit/services/authorization/test_default_authorization_service.py @@ -2,11 +2,13 @@ from __future__ import annotations +from dataclasses import FrozenInstanceError from datetime import datetime, timezone from unittest.mock import AsyncMock from uuid import uuid4 import pytest +from lfx.services.authorization import DirectoryMembershipIngestResult from lfx.services.authorization.base import ( AuthorizationMutation, AuthorizationMutationKind, @@ -175,7 +177,14 @@ async def test_identity_and_directory_contracts_are_default_noops(service: Autho assert await service.validate_identity_mutation(session=session, mutation=mutation) is None assert await service.stage_identity_mutation(session=session, event=mutation) is None assert await service.identity_mutation_committed(mutation) is None - assert await service.ingest_directory_membership_snapshot(session=session, snapshot=snapshot) is None + result = await service.ingest_directory_membership_snapshot(session=session, snapshot=snapshot) + assert result == DirectoryMembershipIngestResult() + assert result.changed is False + assert result.added == 0 + assert result.removed == 0 + assert not hasattr(result, "__dict__") + with pytest.raises(FrozenInstanceError): + result.changed = True # type: ignore[misc] async def test_identity_committed_adapts_to_legacy_invalidation_hooks(