Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 4 additions & 7 deletions backend/app/api/enterprise.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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:
Expand Down
87 changes: 87 additions & 0 deletions backend/tests/test_system_settings_api.py
Original file line number Diff line number Diff line change
@@ -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