diff --git a/backend/app/api/enterprise.py b/backend/app/api/enterprise.py index 86eef7afd..4c46a6fc3 100644 --- a/backend/app/api/enterprise.py +++ b/backend/app/api/enterprise.py @@ -1067,10 +1067,10 @@ async def get_notification_bar_public( @router.get("/system-settings/{key}") async def get_system_setting( key: str, - current_user: User = Depends(get_current_user), + current_user: User = Depends(require_role("platform_admin")), db: AsyncSession = Depends(get_db), ): - """Get a system setting by key.""" + """Get a platform-level system setting (platform admin only).""" result = await db.execute(select(SystemSetting).where(SystemSetting.key == key)) setting = result.scalar_one_or_none() if not setting: @@ -1082,13 +1082,10 @@ async def get_system_setting( async def update_system_setting( key: str, data: SettingUpdate, - current_user: User = Depends(get_current_admin), + current_user: User = Depends(require_role("platform_admin")), 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") + """Create or update a platform-level system setting (platform admin only).""" result = await db.execute(select(SystemSetting).where(SystemSetting.key == key)) setting = result.scalar_one_or_none() if setting: diff --git a/backend/tests/test_system_settings_api.py b/backend/tests/test_system_settings_api.py new file mode 100644 index 000000000..36b29f189 --- /dev/null +++ b/backend/tests/test_system_settings_api.py @@ -0,0 +1,87 @@ +"""Authorization regression coverage for platform system settings.""" + +import uuid +from types import SimpleNamespace + +import httpx +import pytest +from fastapi import FastAPI + +from app.api.enterprise import router +from app.core.security import get_current_user +from app.database import get_db + + +app = FastAPI() +app.include_router(router, prefix="/api") + + +class _Result: + def scalar_one_or_none(self): + return None + + +class _Session: + async def execute(self, _statement: object) -> _Result: + return _Result() + + +async def _get_db(): + yield _Session() + + +@pytest.fixture +def client(): + transport = httpx.ASGITransport(app=app) + + async def _build(): + return httpx.AsyncClient(transport=transport, base_url="http://test") + + return _build + + +@pytest.fixture(autouse=True) +def clear_dependency_overrides(): + yield + app.dependency_overrides.clear() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["get", "put"]) +async def test_org_admin_cannot_read_or_modify_platform_system_settings(client, method: str) -> None: + app.dependency_overrides[get_current_user] = lambda: SimpleNamespace( + id=uuid.uuid4(), + role="org_admin", + identity=None, + tenant_id=uuid.uuid4(), + is_active=True, + ) + app.dependency_overrides[get_db] = _get_db + + async with await client() as ac: + if method == "get": + response = await ac.get("/api/enterprise/system-settings/system_email_platform") + else: + response = await ac.put( + "/api/enterprise/system-settings/system_email_platform", + json={"value": {"SYSTEM_SMTP_PASSWORD": "attacker-value"}}, + ) + + assert response.status_code == 403 + + +@pytest.mark.asyncio +async def test_member_cannot_read_platform_system_settings(client) -> None: + app.dependency_overrides[get_current_user] = lambda: SimpleNamespace( + id=uuid.uuid4(), + role="member", + identity=None, + tenant_id=uuid.uuid4(), + is_active=True, + ) + app.dependency_overrides[get_db] = _get_db + + async with await client() as ac: + response = await ac.get("/api/enterprise/system-settings/jina_api_key") + + assert response.status_code == 403