fix(api): enforce tenant access for connector routes (#14747)

### What problem does this PR solve?

Fixes #14746.

Adds tenant access checks for connector-by-id REST routes before reading
connector details, mutating connector config/status, deleting
connectors, rebuilding, or listing sync logs. Unauthorized callers now
receive `RetCode.AUTHENTICATION_ERROR` with `No authorization.` without
reaching the connector/log mutation paths.

Validation:
- `python3 -m pytest
--confcutdir=test/testcases/test_web_api/test_connector_app
test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py`
- `uvx ruff check api/apps/restful_apis/connector_api.py
api/db/services/connector_service.py
test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py`

### Type of change

- [x] Bug Fix (non-breaking change which fixes an issue)

Co-authored-by: dev111-actor <dev111-actor@users.noreply.github.com>

api/apps/restful_apis/connector_api.py

@@ -35,9 +35,23 @@ from rag.utils.redis_conn import REDIS_CONN
 from api.apps import login_required, current_user
 from box_sdk_gen import BoxOAuth, OAuthConfig, GetAuthorizeUrlOptions
 
+
+LOGGER = logging.getLogger(__name__)
+
+
+def _connector_auth_error(connector_id: str, user_id: str):
+    """Return the connector authorization failure response and log the denial."""
+    LOGGER.warning("connector access denied: connector_id=%s user_id=%s", connector_id, user_id)
+    return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR)
+
+
 @manager.route("/connectors/<connector_id>", methods=["PATCH"])  # noqa: F821
 @login_required
 async def update_connector(connector_id):
+    """Update an accessible connector's polling configuration."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     req = await get_request_json()
     e, conn = ConnectorService.get_by_id(connector_id)
     if not e:
@@ -57,6 +71,7 @@ async def update_connector(connector_id):
 @manager.route("/connectors", methods=["POST"])  # noqa: F821
 @login_required
 async def create_connector():
+    """Create a connector owned by the current tenant."""
     req = await get_request_json()
     if req:
         req["id"] = get_uuid()
@@ -83,12 +98,17 @@ async def create_connector():
 @manager.route("/connectors", methods=["GET"])  # noqa: F821
 @login_required
 def list_connector():
+    """List connectors owned by the current tenant."""
     return get_json_result(data=ConnectorService.list(current_user.id))
 
 
 @manager.route("/connectors/<connector_id>", methods=["GET"])  # noqa: F821
 @login_required
 def get_connector(connector_id):
+    """Return connector details when the current user can access it."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     e, conn = ConnectorService.get_by_id(connector_id)
     if not e:
         return get_data_error_result(message="Can't find this Connector!")
@@ -98,6 +118,10 @@ def get_connector(connector_id):
 @manager.route("/connectors/<connector_id>/logs", methods=["GET"])  # noqa: F821
 @login_required
 def list_logs(connector_id):
+    """List sync logs for a connector the current user can access."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     req = request.args.to_dict(flat=True)
     arr, total = SyncLogsService.list_sync_tasks(connector_id, int(req.get("page", 1)), int(req.get("page_size", 15)))
     return get_json_result(data={"total": total, "logs": arr})
@@ -106,6 +130,10 @@ def list_logs(connector_id):
 @manager.route("/connectors/<connector_id>/resume", methods=["POST"])  # noqa: F821
 @login_required
 async def resume(connector_id):
+    """Resume or cancel sync for an accessible connector."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     req = await get_request_json()
     if req.get("resume"):
         ConnectorService.resume(connector_id, TaskStatus.SCHEDULE)
@@ -116,9 +144,15 @@ async def resume(connector_id):
 
 @manager.route("/connectors/<connector_id>/rebuild", methods=["POST"])  # noqa: F821
 @login_required
-@validate_request("kb_id")
 async def rebuild(connector_id):
+    """Schedule a rebuild for an accessible connector and knowledge base."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     req = await get_request_json()
+    if "kb_id" not in req:
+        return get_json_result(code=RetCode.ARGUMENT_ERROR, message="required argument is missing: kb_id")
+
     err = ConnectorService.rebuild(req["kb_id"], connector_id, current_user.id)
     if err:
         return get_json_result(data=False, message=err, code=RetCode.SERVER_ERROR)
@@ -128,6 +162,10 @@ async def rebuild(connector_id):
 @manager.route("/connectors/<connector_id>", methods=["DELETE"])  # noqa: F821
 @login_required
 def rm_connector(connector_id):
+    """Delete an accessible connector after canceling its sync tasks."""
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     ConnectorService.resume(connector_id, TaskStatus.CANCEL)
     ConnectorService.delete_by_id(connector_id)
     return get_json_result(data=True)
@@ -141,6 +179,9 @@ async def test_connector(connector_id):
     For the REST API connector, this uses `RestAPIConnector.validate_config`
     against the existing saved configuration.
     """
+    if not ConnectorService.accessible(connector_id, current_user.id):
+        return _connector_auth_error(connector_id, current_user.id)
+
     from common.data_source.rest_api_connector import RestAPIConnector
     from common.data_source.exceptions import ConnectorMissingCredentialError, ConnectorValidationError
 

api/db/services/connector_service.py

@@ -22,7 +22,7 @@ from anthropic import BaseModel
 from peewee import SQL, fn
 
 from api.db import InputType
-from api.db.db_models import Connector, SyncLogs, Connector2Kb, Knowledgebase
+from api.db.db_models import DB, Connector, SyncLogs, Connector2Kb, Knowledgebase
 from api.db.services.common_service import CommonService
 from api.db.services.document_service import DocumentService
 from api.db.services.document_service import DocMetadataService
@@ -32,9 +32,37 @@ from common.constants import TaskStatus
 from common.settings import TIMEZONE
 from common.time_utils import current_timestamp, timestamp_to_date
 
+LOGGER = logging.getLogger(__name__)
+
+
 class ConnectorService(CommonService):
     model = Connector
 
+    @classmethod
+    @DB.connection_context()
+    def accessible(cls, connector_id: str, user_id: str) -> bool:
+        """Return whether the user can access the connector's tenant."""
+        e, connector = cls.get_by_id(connector_id)
+        if not e:
+            LOGGER.warning("connector access denied: connector not found connector_id=%s user_id=%s", connector_id, user_id)
+            return False
+
+        if connector.tenant_id == user_id:
+            return True
+
+        from api.db.services.user_service import TenantService
+
+        joined_tenants = TenantService.get_joined_tenants_by_user_id(user_id)
+        has_access = any(tenant["tenant_id"] == connector.tenant_id for tenant in joined_tenants)
+        if not has_access:
+            LOGGER.warning(
+                "connector access denied: tenant mismatch connector_id=%s user_id=%s tenant_id=%s",
+                connector_id,
+                user_id,
+                connector.tenant_id,
+            )
+        return has_access
+
     @classmethod
     def resume(cls, connector_id, status):
         for c2k in Connector2KbService.query(connector_id=connector_id):
@@ -370,4 +398,3 @@ class Connector2KbService(CommonService):
                         cls.model.kb_id==kb_id
                     ).dicts()
         )
-

test/testcases/restful_api/test_connector_routes_unit.py

@@ -200,6 +200,10 @@ def _load_connector_app(monkeypatch):
         def list(_tenant_id):
             return []
 
+        @staticmethod
+        def accessible(*_args, **_kwargs):
+            return True
+
         @staticmethod
         def resume(*_args, **_kwargs):
             return True
@@ -246,6 +250,7 @@ def _load_connector_app(monkeypatch):
         SERVER_ERROR=500,
         RUNNING=102,
         PERMISSION_ERROR=403,
+        AUTHENTICATION_ERROR=109,
     )
     constants_mod.TaskStatus = SimpleNamespace(SCHEDULE="schedule", CANCEL="cancel")
     monkeypatch.setitem(sys.modules, "common.constants", constants_mod)
@@ -420,6 +425,42 @@ def test_connector_basic_routes_and_task_controls(monkeypatch):
     assert delete_calls == ["conn-rm"]
 
 
+@pytest.mark.p2
+def test_connector_by_id_routes_reject_cross_tenant_access(monkeypatch):
+    """Verify per-id connector routes stop before body parsing or service access."""
+    module = _load_connector_app(monkeypatch)
+
+    touched = []
+    monkeypatch.setattr(module.ConnectorService, "accessible", lambda cid, uid: False)
+    monkeypatch.setattr(module.ConnectorService, "get_by_id", lambda *_args: touched.append("get_by_id"))
+    monkeypatch.setattr(module.SyncLogsService, "list_sync_tasks", lambda *_args: touched.append("list_sync_tasks"))
+    monkeypatch.setattr(module.ConnectorService, "resume", lambda *_args: touched.append("resume"))
+    monkeypatch.setattr(module.ConnectorService, "delete_by_id", lambda *_args: touched.append("delete_by_id"))
+    monkeypatch.setattr(module.ConnectorService, "update_by_id", lambda *_args: touched.append("update_by_id"))
+    monkeypatch.setattr(module.ConnectorService, "rebuild", lambda *_args: touched.append("rebuild"))
+
+    def _get_request_json():
+        touched.append("get_request_json")
+        return _AwaitableValue({"resume": True, "config": {"x": 1}})
+
+    monkeypatch.setattr(module, "get_request_json", _get_request_json)
+
+    responses = [
+        _run(module.update_connector("conn-victim")),
+        module.get_connector("conn-victim"),
+        module.list_logs("conn-victim"),
+        _run(module.resume("conn-victim")),
+        _run(module.rebuild("conn-victim")),
+        module.rm_connector("conn-victim"),
+        _run(module.test_connector("conn-victim")),
+    ]
+
+    assert all(res["code"] == module.RetCode.AUTHENTICATION_ERROR for res in responses)
+    assert all(res["message"] == "No authorization." for res in responses)
+    assert all(res["data"] is False for res in responses)
+    assert touched == []
+
+
 @pytest.mark.p2
 def test_connector_oauth_helper_functions(monkeypatch):
     module = _load_connector_app(monkeypatch)

test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py

@@ -200,6 +200,10 @@ def _load_connector_app(monkeypatch):
         def list(_tenant_id):
             return []
 
+        @staticmethod
+        def accessible(*_args, **_kwargs):
+            return True
+
         @staticmethod
         def resume(*_args, **_kwargs):
             return True
@@ -246,6 +250,7 @@ def _load_connector_app(monkeypatch):
         SERVER_ERROR=500,
         RUNNING=102,
         PERMISSION_ERROR=403,
+        AUTHENTICATION_ERROR=109,
     )
     constants_mod.TaskStatus = SimpleNamespace(SCHEDULE="schedule", CANCEL="cancel")
     monkeypatch.setitem(sys.modules, "common.constants", constants_mod)
@@ -420,6 +425,42 @@ def test_connector_basic_routes_and_task_controls(monkeypatch):
     assert delete_calls == ["conn-rm"]
 
 
+@pytest.mark.p2
+def test_connector_by_id_routes_reject_cross_tenant_access(monkeypatch):
+    """Verify per-id connector routes stop before body parsing or service access."""
+    module = _load_connector_app(monkeypatch)
+
+    touched = []
+    monkeypatch.setattr(module.ConnectorService, "accessible", lambda cid, uid: False)
+    monkeypatch.setattr(module.ConnectorService, "get_by_id", lambda *_args: touched.append("get_by_id"))
+    monkeypatch.setattr(module.SyncLogsService, "list_sync_tasks", lambda *_args: touched.append("list_sync_tasks"))
+    monkeypatch.setattr(module.ConnectorService, "resume", lambda *_args: touched.append("resume"))
+    monkeypatch.setattr(module.ConnectorService, "delete_by_id", lambda *_args: touched.append("delete_by_id"))
+    monkeypatch.setattr(module.ConnectorService, "update_by_id", lambda *_args: touched.append("update_by_id"))
+    monkeypatch.setattr(module.ConnectorService, "rebuild", lambda *_args: touched.append("rebuild"))
+
+    def _get_request_json():
+        touched.append("get_request_json")
+        return _AwaitableValue({"resume": True, "config": {"x": 1}})
+
+    monkeypatch.setattr(module, "get_request_json", _get_request_json)
+
+    responses = [
+        _run(module.update_connector("conn-victim")),
+        module.get_connector("conn-victim"),
+        module.list_logs("conn-victim"),
+        _run(module.resume("conn-victim")),
+        _run(module.rebuild("conn-victim")),
+        module.rm_connector("conn-victim"),
+        _run(module.test_connector("conn-victim")),
+    ]
+
+    assert all(res["code"] == module.RetCode.AUTHENTICATION_ERROR for res in responses)
+    assert all(res["message"] == "No authorization." for res in responses)
+    assert all(res["data"] is False for res in responses)
+    assert touched == []
+
+
 @pytest.mark.p2
 def test_connector_oauth_helper_functions(monkeypatch):
     module = _load_connector_app(monkeypatch)