fix: fix agent owner can not edit agent (#45040)

api/controllers/common/rbac/locators.py

@@ -88,7 +88,7 @@ class AgentId(_ParamLocator):
         return ResourceIdentity(self.scope, _required(path_args, self.param))
 
     def owner_id(self, tenant_id: str, identity: ResourceIdentity) -> str | None:
-        return None
+        return application_services().rbac.queries.get_agent_creator(tenant_id, identity.id)
 
 
 class PlainApp(_ParamLocator):

api/repositories/app/console_repository.py

@@ -129,6 +129,16 @@ class ConsoleAppRepository(ConsoleApps, DiscoveryApps, RelatedApps):
             binding = app.agent_app_binding_with_session(session=session, include_archived=True)
             return AppAgentBinding(binding.id, binding.scope == AgentScope.WORKFLOW_ONLY) if binding else None
 
+    def get_agent_creator_id(self, workspace_id: str, agent_id: str) -> str | None:
+        with self._session_factory() as session:
+            return session.scalar(
+                select(Agent.created_by).where(
+                    Agent.id == agent_id,
+                    Agent.tenant_id == workspace_id,
+                    Agent.status == AgentStatus.ACTIVE,
+                )
+            )
+
     def get_maintainer_id(self, workspace_id: str, resource_id: str, *, normal_only: bool = False) -> str | None:
         statement = select(App.maintainer).where(App.id == resource_id, App.tenant_id == workspace_id)
         if normal_only:

api/services/rbac/resource_queries.py

@@ -9,6 +9,10 @@ from services.rbac.ports import ResourceMaintainers
 class AppResourceQueries(Protocol):
     def get_agent_binding(self, workspace_id: str, app_id: str) -> AppAgentBinding | None: ...
 
+    def get_agent_creator_id(self, workspace_id: str, agent_id: str) -> str | None:
+        """Return an active Agent's creator in this workspace, or None if absent or unset."""
+        ...
+
     def get_maintainer_id(self, workspace_id: str, resource_id: str, *, normal_only: bool = False) -> str | None: ...
 
 
@@ -26,6 +30,9 @@ class ResourceQueryService:
     def get_app_agent_binding(self, workspace_id: str, app_id: str) -> AppAgentBinding | None:
         return self._apps.get_agent_binding(workspace_id, app_id)
 
+    def get_agent_creator(self, workspace_id: str, agent_id: str) -> str | None:
+        return self._apps.get_agent_creator_id(workspace_id, agent_id)
+
     def get_app_maintainer(self, workspace_id: str, app_id: str) -> str | None:
         return self._apps.get_maintainer_id(workspace_id, app_id, normal_only=True)
 

api/tests/unit_tests/controllers/common/rbac/test_locators.py

@@ -3,11 +3,13 @@
 import pytest
 from flask import Flask
 from sqlalchemy import Engine, event, text
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import Forbidden, NotFound
 
+from configs import dify_config
 from controllers.common.errors import InvalidArgumentError
 from controllers.common.rbac import locators
-from core.rbac import RBACResourceScope
+from controllers.common.rbac.checks import RBACCheck, enforce_rbac_checks
+from core.rbac import RBACPermission, RBACResourceScope
 from models.agent import Agent, AgentScope, AgentSource, AgentStatus
 from models.model import AppMode
 from tests.helpers.unit.model_factories import make_app, make_dataset, make_document
@@ -124,6 +126,78 @@ def test_app_owner_resolution_preserves_normal_app_requirement(
         assert locator.owner_id("other-workspace", identity) is None
 
 
+@pytest.mark.parametrize("status", [AgentStatus.ACTIVE, AgentStatus.ARCHIVED])
+def test_agent_owner_resolution_is_workspace_scoped(rbac_domain: RBACDomain, status: AgentStatus) -> None:
+    with rbac_domain.sessions.begin() as session:
+        session.add(
+            Agent(
+                id="agent",
+                tenant_id="workspace",
+                name="Agent",
+                scope=AgentScope.ROSTER,
+                source=AgentSource.AGENT_APP,
+                created_by="owner",
+                status=status,
+            )
+        )
+    locator = locators.AgentId()
+    identity = locators.ResourceIdentity(RBACResourceScope.AGENT, "agent")
+    assert locator.owner_id("workspace", identity) == ("owner" if status == AgentStatus.ACTIVE else None)
+    assert locator.owner_id("other-workspace", identity) is None
+    assert locator.owner_id("workspace", locators.ResourceIdentity(RBACResourceScope.AGENT, "missing")) is None
+
+
+@pytest.mark.parametrize(
+    ("workspace_id", "account_id", "is_owner"),
+    [("workspace", "owner", True), ("workspace", "other", False), ("other-workspace", "owner", False)],
+)
+def test_agent_creator_bypasses_rbac_only_in_own_workspace(
+    app: Flask,
+    rbac_domain: RBACDomain,
+    monkeypatch: pytest.MonkeyPatch,
+    workspace_id: str,
+    account_id: str,
+    is_owner: bool,
+) -> None:
+    monkeypatch.setattr(dify_config, "RBAC_ENABLED", True)
+    rbac_domain.transport.response = {"allowed": False}
+    with rbac_domain.sessions.begin() as session:
+        session.add(
+            Agent(
+                id="agent",
+                tenant_id="workspace",
+                name="Agent",
+                scope=AgentScope.ROSTER,
+                source=AgentSource.AGENT_APP,
+                created_by="owner",
+            )
+        )
+    with app.test_request_context():
+        if is_owner:
+            enforce_rbac_checks(
+                tenant_id=workspace_id,
+                account_id=account_id,
+                checks=[RBACCheck(RBACPermission.AGENT_PREVIEW, locators.AgentId())],
+                path_args={"agent_id": "agent"},
+            )
+            assert not rbac_domain.transport.requests
+        else:
+            with pytest.raises(Forbidden):
+                enforce_rbac_checks(
+                    tenant_id=workspace_id,
+                    account_id=account_id,
+                    checks=[RBACCheck(RBACPermission.AGENT_PREVIEW, locators.AgentId())],
+                    path_args={"agent_id": "agent"},
+                )
+            assert rbac_domain.transport.only_request.json == {
+                "tenant_id": workspace_id,
+                "account_id": account_id,
+                "scene": "agent_preview",
+                "resource_type": "agent",
+                "resource_id": "agent",
+            }
+
+
 @pytest.mark.parametrize("locator", [locators.DatasetByDocument(), locators.DatasetByPipeline()])
 def test_nested_dataset_resolution_checks_ownership_and_maps_missing_resources(
     app: Flask, rbac_domain: RBACDomain, locator: locators.DatasetByDocument | locators.DatasetByPipeline