feat(authz): complete release 1.12 RBAC deliverables

Author: erichare

.secrets.baseline

@@ -2243,7 +2243,7 @@
         "filename": "src/backend/tests/unit/services/variable/test_service.py",
         "hashed_secret": "30abc0a833efea23496b4d226fffa2f90c0855c0",
         "is_verified": false,
-        "line_number": 52,
+        "line_number": 53,
         "is_secret": false
       }
     ],
@@ -7222,5 +7222,5 @@
       }
     ]
   },
-  "generated_at": "2026-07-22T04:00:31Z"
+  "generated_at": "2026-07-22T21:43:55Z"
 }

src/backend/base/langflow/agentic/helpers/validation.py

@@ -241,6 +241,9 @@ async def _execute_output_methods_for_validation(cc_instance) -> str | None:
         return None
 
     try:
+        from lfx.services.model_provider_policy import ModelProviderPolicyPurpose
+
+        cc_instance.require_model_provider_policy(ModelProviderPolicyPurpose.USE)
         await cc_instance._build_results()  # noqa: SLF001 — sandbox bypass of send_error/tracing
     except ValidationError as exc:
         return _format_validation_error(exc)

src/backend/base/langflow/alembic/versions/a6c4e2f8b1d3_add_authz_audit_actor_identity.py

@@ -0,0 +1,71 @@
+"""Add first-class actor identity to authorization audit rows.
+
+Phase: EXPAND
+Revision ID: a6c4e2f8b1d3
+Revises: d19e7b3c5a42
+Create Date: 2026-07-22 00:00:00.000000
+
+Both columns are nullable for N-1 compatibility and to preserve legacy rows.
+``actor_id`` intentionally has no foreign key: audit attribution must survive
+deletion of an API key or other credential record.
+"""
+
+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 identifiers, used by Alembic.
+revision: str = "a6c4e2f8b1d3"  # pragma: allowlist secret
+down_revision: str | None = "d19e7b3c5a42"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+TABLE_NAME = "authz_audit_log"
+ACTOR_INDEX = "ix_authz_audit_log_actor_timestamp"
+ACTOR_TYPE_INDEX = "ix_authz_audit_log_actor_type_timestamp"
+
+
+def _index_exists(conn, index_name: str) -> bool:
+    return index_name in {index["name"] for index in sa.inspect(conn).get_indexes(TABLE_NAME)}
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists(TABLE_NAME, conn):
+        return
+
+    with op.batch_alter_table(TABLE_NAME, schema=None) as batch_op:
+        if not migration.column_exists(TABLE_NAME, "actor_type", conn):
+            batch_op.add_column(sa.Column("actor_type", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
+        if not migration.column_exists(TABLE_NAME, "actor_id", conn):
+            batch_op.add_column(sa.Column("actor_id", sa.Uuid(), nullable=True))
+
+    # Re-inspect after the batch because SQLite recreates the table.
+    if not _index_exists(conn, ACTOR_INDEX):
+        op.create_index(ACTOR_INDEX, TABLE_NAME, ["actor_id", "timestamp"], unique=False)
+    if not _index_exists(conn, ACTOR_TYPE_INDEX):
+        op.create_index(ACTOR_TYPE_INDEX, TABLE_NAME, ["actor_type", "timestamp"], unique=False)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if not migration.table_exists(TABLE_NAME, conn):
+        return
+
+    for index_name in (ACTOR_INDEX, ACTOR_TYPE_INDEX):
+        if _index_exists(conn, index_name):
+            op.drop_index(index_name, table_name=TABLE_NAME)
+
+    with op.batch_alter_table(TABLE_NAME, schema=None) as batch_op:
+        if migration.column_exists(TABLE_NAME, "actor_id", conn):
+            batch_op.drop_column("actor_id")
+        if migration.column_exists(TABLE_NAME, "actor_type", conn):
+            batch_op.drop_column("actor_type")

src/backend/base/langflow/alembic/versions/b7d5f9a3c2e4_sync_flow_workspace_from_project.py

@@ -0,0 +1,60 @@
+"""Synchronize denormalized flow workspace ids from their projects.
+
+Phase: MIGRATE
+Revision ID: b7d5f9a3c2e4
+Revises: a6c4e2f8b1d3
+Create Date: 2026-07-22 00:00:00.000000
+
+Project membership is the authorization source of truth. Older rows could
+carry a caller-supplied ``flow.workspace_id`` that disagreed with the owning
+folder, so repair every project-scoped row before workspace grants are used as
+a SQL list prefilter.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+revision: str = "b7d5f9a3c2e4"  # pragma: allowlist secret
+down_revision: str | None = "a6c4e2f8b1d3"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if not (migration.table_exists("flow", conn) and migration.table_exists("folder", conn)):
+        return
+    required = (
+        migration.column_exists("flow", "folder_id", conn)
+        and migration.column_exists("flow", "workspace_id", conn)
+        and migration.column_exists("folder", "workspace_id", conn)
+    )
+    if not required:
+        return
+
+    conn.execute(
+        sa.text(
+            """
+            UPDATE flow
+            SET workspace_id = (
+                SELECT folder.workspace_id
+                FROM folder
+                WHERE folder.id = flow.folder_id
+            )
+            WHERE flow.folder_id IS NOT NULL
+              AND EXISTS (SELECT 1 FROM folder WHERE folder.id = flow.folder_id)
+            """
+        )
+    )
+
+
+def downgrade() -> None:
+    """Data repair is intentionally irreversible."""

src/backend/base/langflow/api/v1/authz_audit.py

@@ -15,6 +15,7 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
+from sqlalchemy import or_
 from sqlmodel import col, select
 
 from langflow.api.utils import DbSession
@@ -32,6 +33,8 @@ class AuthzAuditLogRead(BaseModel):
 
     id: UUID
     user_id: UUID | None
+    actor_type: str | None = None
+    actor_id: UUID | None = None
     action: str
     resource_type: str | None
     resource_id: UUID | None
@@ -58,6 +61,14 @@ async def list_audit_log(
     session: DbSession,
     _admin: Annotated[User, Depends(get_current_active_superuser)],
     user_id: Annotated[UUID | None, Query(description="Filter by acting user id.")] = None,
+    actor_type: Annotated[
+        str | None,
+        Query(description="Filter by credential actor type; ``unknown`` also includes legacy rows without a type."),
+    ] = None,
+    actor_id: Annotated[
+        UUID | None,
+        Query(description="Filter by first-class credential actor UUID."),
+    ] = None,
     resource_type: Annotated[
         str | None,
         Query(description="Filter by resource type slug, e.g. ``flow`` or ``deployment``."),
@@ -88,6 +99,21 @@ async def list_audit_log(
     base = select(AuthzAuditLog)
     if user_id is not None:
         base = base.where(AuthzAuditLog.user_id == user_id)
+    if actor_type == "unknown":
+        # Rows written before first-class actor identity was added retain a
+        # NULL actor_type. The UI presents one "Legacy / unknown" filter, so
+        # its backend meaning must include both historical NULLs and new
+        # events explicitly classified as unknown.
+        base = base.where(
+            or_(
+                AuthzAuditLog.actor_type == actor_type,
+                col(AuthzAuditLog.actor_type).is_(None),
+            )
+        )
+    elif actor_type is not None:
+        base = base.where(AuthzAuditLog.actor_type == actor_type)
+    if actor_id is not None:
+        base = base.where(AuthzAuditLog.actor_id == actor_id)
     if resource_type is not None:
         base = base.where(AuthzAuditLog.resource_type == resource_type)
     if resource_id is not None:
@@ -110,7 +136,14 @@ async def list_audit_log(
     total_stmt = select(func.count()).select_from(base.subquery())
     total = int((await session.exec(total_stmt)).first() or 0)
 
-    page_stmt = base.order_by(col(AuthzAuditLog.timestamp).desc()).offset((page - 1) * size).limit(size)
+    page_stmt = (
+        base.order_by(
+            col(AuthzAuditLog.timestamp).desc(),
+            col(AuthzAuditLog.id).desc(),
+        )
+        .offset((page - 1) * size)
+        .limit(size)
+    )
     rows = list(await session.exec(page_stmt))
 
     items = [AuthzAuditLogRead.model_validate(row, from_attributes=True) for row in rows]

src/backend/base/langflow/api/v1/authz_route_dependencies.py

@@ -8,7 +8,7 @@
 from fastapi import Depends, HTTPException, status
 
 from langflow.api.utils import CurrentActiveUser, DbSession
-from langflow.api.v1.flows_helpers import _read_flow
+from langflow.api.v1.flows_helpers import _canonicalize_flow_destination, _read_flow
 from langflow.services.authorization import FlowAction, ensure_flow_permission
 from langflow.services.authorization.fetch import deny_to_404
 from langflow.services.database.models.flow.model import Flow, FlowCreate
@@ -86,17 +86,20 @@ async def get_authorized_flow_for_delete(
 async def require_flow_create_permission(
     current_user: CurrentActiveUser,
     flow: FlowCreate,
-) -> None:
+    session: DbSession,
+) -> tuple[UUID | None, UUID]:
     """Authorize CREATE at the destination workspace/folder before inserting a flow."""
+    destination = await _canonicalize_flow_destination(session, flow, current_user.id)
     await ensure_flow_permission(
         current_user,
         FlowAction.CREATE,
         workspace_id=flow.workspace_id,
         folder_id=flow.folder_id,
     )
+    return destination
 
 
 AuthorizedReadFlow = Annotated[Flow, Depends(get_authorized_flow_for_read)]
 AuthorizedWriteFlow = Annotated[Flow, Depends(get_authorized_flow_for_write)]
 AuthorizedDeleteFlow = Annotated[Flow, Depends(get_authorized_flow_for_delete)]
-RequireFlowCreate = Annotated[None, Depends(require_flow_create_permission)]
+RequireFlowCreate = Annotated[tuple[UUID | None, UUID], Depends(require_flow_create_permission)]