fix: restrict global system settings access

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_<tenant UUID>``; 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:

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()