feat(authz): add source-aware role assignment grants

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)

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

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

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,