diff --git a/api/apps/restful_apis/connector_api.py b/api/apps/restful_apis/connector_api.py index 5e428d1a22..21ab7fd4d0 100644 --- a/api/apps/restful_apis/connector_api.py +++ b/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/", 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/", 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//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//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//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/", 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 diff --git a/api/db/services/connector_service.py b/api/db/services/connector_service.py index ab754101e1..4d9fd2258a 100644 --- a/api/db/services/connector_service.py +++ b/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() ) - diff --git a/test/testcases/restful_api/test_connector_routes_unit.py b/test/testcases/restful_api/test_connector_routes_unit.py index ad47aef379..33c4d7a8f1 100644 --- a/test/testcases/restful_api/test_connector_routes_unit.py +++ b/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) diff --git a/test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py b/test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py index 9d9e1c9c14..3807fb8e15 100644 --- a/test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py +++ b/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)