From 27650d57fa7e260a5e856043d5e321eeec52cb7c Mon Sep 17 00:00:00 2001 From: yaojin Date: Wed, 5 Aug 2026 16:59:24 +0800 Subject: [PATCH] fix: restrict global system settings access --- backend/app/api/enterprise.py | 26 +++++- .../test_enterprise_system_settings_access.py | 83 +++++++++++++++++++ 2 files changed, 106 insertions(+), 3 deletions(-) create mode 100644 backend/tests/test_enterprise_system_settings_access.py diff --git a/backend/app/api/enterprise.py b/backend/app/api/enterprise.py index 86eef7afd..a0e4e19ba 100644 --- a/backend/app/api/enterprise.py +++ b/backend/app/api/enterprise.py @@ -938,6 +938,27 @@ class RuntimeModelSettingsUpdate(BaseModel): compact_model_id: uuid.UUID +def _require_system_setting_access(key: str, current_user: User) -> None: + """Authorize access to a platform setting or a tenant company introduction. + + ``system_settings`` is a global key/value table and can contain credentials. + The sole tenant-scoped key family exposed through this API is + ``company_intro_``; organization administrators may manage + only their own tenant's entry. All other keys require a platform admin. + """ + company_intro_prefix = "company_intro_" + if key.startswith(company_intro_prefix): + try: + tenant_id = uuid.UUID(key.removeprefix(company_intro_prefix)) + except ValueError: + tenant_id = None + if tenant_id is not None and current_user.role == "org_admin" and current_user.tenant_id == tenant_id: + return + if _is_platform_admin_user(current_user): + return + raise HTTPException(status_code=403, detail="Platform admin access required for system settings") + + def _runtime_settings_tenant_id(current_user: User, requested_tenant_id: str | None) -> uuid.UUID: raw_tenant_id = requested_tenant_id or current_user.tenant_id if raw_tenant_id is None: @@ -1071,6 +1092,7 @@ async def get_system_setting( db: AsyncSession = Depends(get_db), ): """Get a system setting by key.""" + _require_system_setting_access(key, current_user) result = await db.execute(select(SystemSetting).where(SystemSetting.key == key)) setting = result.scalar_one_or_none() if not setting: @@ -1086,9 +1108,7 @@ async def update_system_setting( db: AsyncSession = Depends(get_db), ): """Create or update a system setting.""" - # Platform-level settings (e.g. PUBLIC_BASE_URL) require platform_admin - if key == "platform" and not _is_platform_admin_user(current_user): - raise HTTPException(status_code=403, detail="Only platform admin can modify platform settings") + _require_system_setting_access(key, current_user) result = await db.execute(select(SystemSetting).where(SystemSetting.key == key)) setting = result.scalar_one_or_none() if setting: diff --git a/backend/tests/test_enterprise_system_settings_access.py b/backend/tests/test_enterprise_system_settings_access.py new file mode 100644 index 000000000..451055c4f --- /dev/null +++ b/backend/tests/test_enterprise_system_settings_access.py @@ -0,0 +1,83 @@ +"""Regression coverage for global system-setting authorization.""" + +import uuid +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fastapi import HTTPException + +from app.api.enterprise import ( + SettingUpdate, + _require_system_setting_access, + get_system_setting, + update_system_setting, +) + + +def _user(*, role: str, tenant_id: uuid.UUID | None = None, platform_identity: bool = False) -> SimpleNamespace: + return SimpleNamespace( + role=role, + tenant_id=tenant_id, + identity=SimpleNamespace(is_platform_admin=platform_identity), + ) + + +def test_member_cannot_read_credential_system_setting() -> None: + with pytest.raises(HTTPException, match="Platform admin") as error: + _require_system_setting_access("system_email_platform", _user(role="member")) + + assert error.value.status_code == 403 + + +def test_org_admin_cannot_modify_global_system_setting() -> None: + with pytest.raises(HTTPException, match="Platform admin") as error: + _require_system_setting_access("jina_api_key", _user(role="org_admin", tenant_id=uuid.uuid4())) + + assert error.value.status_code == 403 + + +def test_org_admin_can_manage_own_company_intro_only() -> None: + tenant_id = uuid.uuid4() + _require_system_setting_access( + f"company_intro_{tenant_id}", + _user(role="org_admin", tenant_id=tenant_id), + ) + + +def test_org_admin_cannot_manage_another_tenant_company_intro() -> None: + with pytest.raises(HTTPException) as error: + _require_system_setting_access( + f"company_intro_{uuid.uuid4()}", + _user(role="org_admin", tenant_id=uuid.uuid4()), + ) + + assert error.value.status_code == 403 + + +def test_platform_admin_can_manage_global_and_tenant_scoped_settings() -> None: + platform_admin = _user(role="platform_admin") + + _require_system_setting_access("system_email_platform", platform_admin) + _require_system_setting_access(f"company_intro_{uuid.uuid4()}", platform_admin) + + +@pytest.mark.asyncio +async def test_endpoints_reject_unauthorized_credential_access_before_querying_database() -> None: + db = AsyncMock() + member = _user(role="member") + org_admin = _user(role="org_admin", tenant_id=uuid.uuid4()) + + with pytest.raises(HTTPException) as get_error: + await get_system_setting("jina_api_key", current_user=member, db=db) + with pytest.raises(HTTPException) as put_error: + await update_system_setting( + "system_email_platform", + SettingUpdate(value={"SYSTEM_SMTP_PASSWORD": "attempted-change"}), + current_user=org_admin, + db=db, + ) + + assert get_error.value.status_code == 403 + assert put_error.value.status_code == 403 + db.execute.assert_not_awaited()