fix: share policy refresh and owner permissions (langflow-ai#13830)

Fix share policy refresh and owner permissions

Author: erichare

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

@@ -15,10 +15,19 @@
 
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel, Field, field_validator
+from sqlmodel import col, select
+from sqlmodel.ext.asyncio.session import AsyncSession
 
-from langflow.api.utils import CurrentActiveUser
+from langflow.api.utils import CurrentActiveUser, DbSessionReadOnly
 from langflow.services.auth.context import current_auth_context_for_authz
 from langflow.services.authorization.access_ceiling import filter_actions_by_external_access_ceiling
+from langflow.services.authorization.guards import should_apply_owner_override
+from langflow.services.database.models.deployment.model import Deployment
+from langflow.services.database.models.file.model import File as UserFile
+from langflow.services.database.models.flow.model import Flow
+from langflow.services.database.models.folder.model import Folder
+from langflow.services.database.models.knowledge_base.model import KnowledgeBaseRecord
+from langflow.services.database.models.variable.model import Variable
 from langflow.services.deps import get_authorization_service
 
 router = APIRouter(prefix="/authz/me", tags=["Authorization"])
@@ -43,6 +52,15 @@
 # letting a client request `["read"] * 100000` to flood the enforcer.
 _MAX_ACTIONS = 10
 
+_RESOURCE_OWNER_LOOKUPS: dict[str, tuple[type, str]] = {
+    "flow": (Flow, "user_id"),
+    "deployment": (Deployment, "user_id"),
+    "project": (Folder, "user_id"),
+    "knowledge_base": (KnowledgeBaseRecord, "user_id"),
+    "variable": (Variable, "user_id"),
+    "file": (UserFile, "user_id"),
+}
+
 
 class EffectivePermissionsRequest(BaseModel):
     """Body for :func:`get_effective_permissions`."""
@@ -102,10 +120,57 @@ class EffectivePermissionsResponse(BaseModel):
     permissions: dict[UUID, list[str]]
 
 
+async def _owned_resource_ids(
+    *,
+    session: AsyncSession,
+    resource_type: str,
+    resource_ids: list[UUID],
+    user_id: UUID,
+) -> set[UUID]:
+    """Return requested resource IDs owned by ``user_id``."""
+    lookup = _RESOURCE_OWNER_LOOKUPS.get(resource_type)
+    if lookup is None or not resource_ids:
+        return set()
+    model, owner_attr = lookup
+    stmt = select(model.id).where(
+        col(model.id).in_(resource_ids),
+        getattr(model, owner_attr) == user_id,
+    )
+    return set((await session.exec(stmt)).all())
+
+
+async def _apply_owner_permissions(
+    *,
+    session: AsyncSession,
+    permissions: dict[UUID, list[str]],
+    resource_type: str,
+    resource_ids: list[UUID],
+    actions: tuple[str, ...],
+    user_id: UUID,
+) -> dict[UUID, list[str]]:
+    """Mirror route guard owner override for the UI permission endpoint."""
+    normalized = {resource_id: list(permissions.get(resource_id, [])) for resource_id in resource_ids}
+    if not await should_apply_owner_override():
+        return normalized
+
+    owned_ids = await _owned_resource_ids(
+        session=session,
+        resource_type=resource_type,
+        resource_ids=resource_ids,
+        user_id=user_id,
+    )
+    for resource_id in owned_ids:
+        allowed = dict.fromkeys(normalized.get(resource_id, []))
+        allowed.update(dict.fromkeys(actions))
+        normalized[resource_id] = list(allowed)
+    return normalized
+
+
 @router.post("/permissions", response_model=EffectivePermissionsResponse)
 async def get_effective_permissions(
     body: EffectivePermissionsRequest,
     current_user: CurrentActiveUser,
+    session: DbSessionReadOnly,
 ) -> EffectivePermissionsResponse:
     """Return per-resource allowed actions for the current user.
 
@@ -134,6 +199,14 @@ async def get_effective_permissions(
             "is_superuser": current_user.is_superuser,
         },
     )
+    permissions = await _apply_owner_permissions(
+        session=session,
+        permissions=permissions,
+        resource_type=body.resource_type,
+        resource_ids=body.resource_ids,
+        actions=actions,
+        user_id=current_user.id,
+    )
     permissions = {
         resource_id: filter_actions_by_external_access_ceiling(allowed_actions)
         for resource_id, allowed_actions in permissions.items()

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

@@ -8,11 +8,13 @@
 
 from fastapi import APIRouter, HTTPException, Query, status
 from lfx.log.logger import logger
+from lfx.services.authorization.base import BaseAuthorizationService
 from sqlmodel import select
 
 from langflow.api.utils import CurrentActiveUser, DbSession
 from langflow.api.v1.schemas.authz_shares import ShareCreate, ShareRead, ShareUpdate
 from langflow.services.authorization import ShareAction, ensure_share_permission
+from langflow.services.authorization.invalidation import safe_invalidate_all, safe_invalidate_user
 from langflow.services.authorization.utils import audit_decision
 from langflow.services.database.models.auth import AuthzShare, AuthzTeamMember, SharePermissionLevel, ShareScope
 from langflow.services.database.models.deployment.model import Deployment
@@ -104,13 +106,32 @@ async def _user_can_see_share(
     )
 
 
-async def _invalidate_for_share(scope: str, target_id: UUID | None) -> None:
+async def _invalidate_for_share(scope: str, target_id: UUID | None, *, op: str = "share:write") -> None:
     """Invalidate cached policy after a share write (user scope vs invalidate_all)."""
     authz = get_authorization_service()
     if scope == ShareScope.USER.value and target_id is not None:
-        await authz.invalidate_user(target_id)
+        await safe_invalidate_user(authz, target_id, op=op)
     else:
-        await authz.invalidate_all()
+        await safe_invalidate_all(authz, op=op)
+
+
+def _uses_base_sync_shares(authz: BaseAuthorizationService) -> bool:
+    """Return True when the service only has the OSS no-op sync_shares hook."""
+    return getattr(type(authz), "sync_shares", None) is BaseAuthorizationService.sync_shares
+
+
+async def _refresh_policy_for_share(scope: str, target_id: UUID | None, *, op: str) -> None:
+    """Refresh share-derived policy after the share DB transaction is durable."""
+    authz = get_authorization_service()
+    sync_shares = getattr(authz, "sync_shares", None)
+    if sync_shares is not None and not _uses_base_sync_shares(authz):
+        try:
+            await sync_shares()
+        except Exception as exc:  # noqa: BLE001 - plugin hooks are best-effort post-commit work
+            logger.warning("sync_shares failed after %s; falling back to targeted invalidation: %s", op, exc)
+        else:
+            return
+    await _invalidate_for_share(scope, target_id, op=op)
 
 
 async def _ensure_can_administer_share(
@@ -189,23 +210,26 @@ async def create_share(
             detail="Share could not be created: it may already exist or conflict with an existing share.",
         ) from exc
     await session.refresh(row)
+    response = ShareRead.model_validate(row, from_attributes=True)
+    await session.commit()
 
-    # Invalidate policy cache for the share audience.
-    await _invalidate_for_share(payload.scope, payload.target_id)
+    # Refresh policy after commit so plugins using a separate DB connection see
+    # the durable authz_share row instead of the pre-commit transaction state.
+    await _refresh_policy_for_share(payload.scope, payload.target_id, op="share:create")
 
     await audit_decision(
         user_id=current_user.id,
         action="share:create",
         obj=f"{payload.resource_type}:{payload.resource_id}",
         result="allow",
         details={
-            "share_id": str(row.id),
+            "share_id": str(response.id),
             "scope": payload.scope,
             "target_id": str(payload.target_id) if payload.target_id else None,
             "permission_level": payload.permission_level,
         },
     )
-    return ShareRead.model_validate(row, from_attributes=True)
+    return response
 
 
 _LIST_SHARES_MAX_LIMIT = 200
@@ -381,20 +405,22 @@ async def update_share(
             detail="Share could not be updated: it may conflict with an existing share.",
         ) from exc
     await session.refresh(row)
+    response = ShareRead.model_validate(row, from_attributes=True)
+    await session.commit()
 
-    await _invalidate_for_share(row.scope, row.target_id)
+    await _refresh_policy_for_share(response.scope, response.target_id, op="share:update")
 
     await audit_decision(
         user_id=current_user.id,
         action="share:update",
-        obj=f"{row.resource_type}:{row.resource_id}",
+        obj=f"{response.resource_type}:{response.resource_id}",
         result="allow",
         details={
-            "share_id": str(row.id),
-            "permission_level": row.permission_level,
+            "share_id": str(response.id),
+            "permission_level": response.permission_level,
         },
     )
-    return ShareRead.model_validate(row, from_attributes=True)
+    return response
 
 
 @router.delete("/{share_id}", status_code=status.HTTP_204_NO_CONTENT)
@@ -429,8 +455,9 @@ async def delete_share(
     scope = row.scope
     await session.delete(row)
     await session.flush()
+    await session.commit()
 
-    await _invalidate_for_share(scope, target_id)
+    await _refresh_policy_for_share(scope, target_id, op="share:delete")
 
     await audit_decision(
         user_id=current_user.id,

src/backend/base/langflow/services/authorization/__init__.py

@@ -24,6 +24,7 @@
     ensure_project_permission,
     ensure_share_permission,
     ensure_variable_permission,
+    should_apply_owner_override,
 )
 from langflow.services.authorization.listing import (
     filter_visible_resources,
@@ -57,5 +58,6 @@
     "requires_flow_permission",
     "requires_resource_permission",
     "restrict_to_owned_or_visible",
+    "should_apply_owner_override",
     "visible_id_prefilter",
 ]

src/backend/base/langflow/services/authorization/guards.py

@@ -96,6 +96,11 @@ async def _api_key_scopes_require_plugin_enforcement() -> bool:
         return False
 
 
+async def should_apply_owner_override() -> bool:
+    """Return True when Langflow should apply the built-in owner override."""
+    return not await _api_key_scopes_require_plugin_enforcement()
+
+
 def _coerce_action(
     act: DeploymentAction
     | FlowAction
@@ -224,11 +229,7 @@ async def _ensure_resource_permission(
             detail="External credentials do not allow this action",
         )
 
-    if (
-        owner_id is not None
-        and getattr(user, "id", None) == owner_id
-        and not await _api_key_scopes_require_plugin_enforcement()
-    ):
+    if owner_id is not None and getattr(user, "id", None) == owner_id and await should_apply_owner_override():
         await _audit.audit_decision(
             user_id=user.id,
             action=f"{resource_type}:{act_str}",

src/backend/base/langflow/services/authorization/utils.py

@@ -51,6 +51,7 @@
     ensure_project_permission,
     ensure_share_permission,
     ensure_variable_permission,
+    should_apply_owner_override,
 )
 from langflow.services.authorization.listing import (
     _default_resource_id_getter,
@@ -104,4 +105,5 @@ def permission_denied_to_http(exc):
     "get_authorization_service",
     "get_settings_service",
     "permission_denied_to_http",
+    "should_apply_owner_override",
 ]

src/backend/tests/unit/api/v1/test_authz_admin_routes.py

@@ -995,12 +995,55 @@ async def test_me_permissions_returns_per_resource_actions(stub_authz):
         resource_type="flow",
         resource_ids=resource_ids,
     )
-    result = await authz_me.get_effective_permissions(body=body, current_user=user)
+    result = await authz_me.get_effective_permissions(body=body, current_user=user, session=_FakeAsyncSession())
     assert result.resource_type == "flow"
     assert set(result.permissions[resource_ids[0]]) == {"read", "execute"}
     assert "delete" in result.permissions[resource_ids[1]]
 
 
+@pytest.mark.asyncio
+async def test_me_permissions_adds_owner_override_actions(stub_authz):
+    """Owners get the same built-in owner override in the UI gate as route guards."""
+    from langflow.api.v1 import authz_me
+    from langflow.api.v1.authz_me import EffectivePermissionsRequest
+
+    authz = stub_authz()
+    owned_flow_id = uuid4()
+    other_flow_id = uuid4()
+    authz.effective_perms_payload = {
+        owned_flow_id: [],
+        other_flow_id: [],
+    }
+    user = _make_user()
+    session = _FakeAsyncSession(exec_results=[[owned_flow_id]])
+
+    body = EffectivePermissionsRequest(
+        resource_type="flow",
+        resource_ids=[owned_flow_id, other_flow_id],
+        actions=["read", "write", "execute", "delete"],
+    )
+    result = await authz_me.get_effective_permissions(body=body, current_user=user, session=session)
+
+    assert set(result.permissions[owned_flow_id]) == {"read", "write", "execute", "delete"}
+    assert result.permissions[other_flow_id] == []
+
+
+@pytest.mark.asyncio
+async def test_me_permissions_returns_empty_entries_for_plugin_omissions(stub_authz):
+    """A plugin omission is normalized to [] so the response shape stays stable."""
+    from langflow.api.v1 import authz_me
+    from langflow.api.v1.authz_me import EffectivePermissionsRequest
+
+    stub_authz()
+    flow_id = uuid4()
+    user = _make_user()
+    body = EffectivePermissionsRequest(resource_type="flow", resource_ids=[flow_id])
+
+    result = await authz_me.get_effective_permissions(body=body, current_user=user, session=_FakeAsyncSession())
+
+    assert result.permissions == {flow_id: []}
+
+
 @pytest.mark.asyncio
 async def test_me_permissions_caps_resource_ids_at_500(stub_authz):
     from langflow.api.v1 import authz_me
@@ -1014,7 +1057,7 @@ async def test_me_permissions_caps_resource_ids_at_500(stub_authz):
         resource_ids=[uuid4() for _ in range(501)],
     )
     with pytest.raises(HTTPException) as excinfo:
-        await authz_me.get_effective_permissions(body=body, current_user=user)
+        await authz_me.get_effective_permissions(body=body, current_user=user, session=_FakeAsyncSession())
     assert excinfo.value.status_code == 400
     assert "capped at 500" in excinfo.value.detail
 
@@ -1028,7 +1071,7 @@ async def test_me_permissions_empty_request_returns_empty(stub_authz):
     user = _make_user()
 
     body = EffectivePermissionsRequest(resource_type="flow", resource_ids=[])
-    result = await authz_me.get_effective_permissions(body=body, current_user=user)
+    result = await authz_me.get_effective_permissions(body=body, current_user=user, session=_FakeAsyncSession())
     assert result.permissions == {}
 
 
@@ -1116,7 +1159,7 @@ async def _capture(**kwargs):
         resource_ids=[rid],
         actions=["READ", "read", " Write "],
     )
-    await authz_me.get_effective_permissions(body=body, current_user=user)
+    await authz_me.get_effective_permissions(body=body, current_user=user, session=_FakeAsyncSession())
     # Normalization happened at the model layer; handler sees the bounded set.
     assert tuple(captured["actions"]) == ("read", "write")