sakshamandClaude Sonnet 5
3e1fa1845b
Was only ever settable via .env - now follows the same pattern as Cloudflare Turnstile: SystemSecretKey.RESEND_API_KEY + a META entry covers storage/encryption/frontend rendering automatically (the Settings UI's Server secrets box is fully data-driven off this list). Wired the register/login/resend-verification/request-password-reset route handlers to use get_effective_settings so an admin-set key actually reaches the emails those flows send, not just .env's value. Co-Authored-By: Claude Sonnet 5 <[email protected]>
313 lines
11 KiB
Python
313 lines
11 KiB
Python
"""Server-wide secrets (Turnstile site key/secret) admin-managed from the
|
|
Settings page instead of only .env: repository/service behavior, endpoint
|
|
auth/admin-gating (deliberately NOT localhost-gated, unlike the old
|
|
/system/api-keys this replaces), /system/status exposing the site key live,
|
|
and an end-to-end proof that a DB-only (no .env) secret actually drives
|
|
Turnstile enforcement on register."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.models.enums import SystemSecretKey
|
|
from app.repositories.system_secret_repository import SystemSecretRepository
|
|
from app.repositories.user_repository import UserRepository
|
|
from app.services import system_secret_service
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
async def _clean_system_secrets(db_session):
|
|
"""SystemSecret rows are true global singletons (one per key, not
|
|
per-user like UserApiKey) - unlike other tests in this suite that
|
|
dodge cross-test pollution by randomizing an id, there's no such trick
|
|
here. Every test starts from a clean slate, and leaves one behind for
|
|
whatever test file runs next in a full-suite run."""
|
|
|
|
async def _clear() -> None:
|
|
repo = SystemSecretRepository(db_session)
|
|
for row in await repo.list_all():
|
|
await db_session.delete(row)
|
|
await db_session.commit()
|
|
|
|
await _clear()
|
|
yield
|
|
await _clear()
|
|
|
|
|
|
def _unique_email() -> str:
|
|
return f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
|
|
|
|
def _register_and_login(client: TestClient) -> dict[str, str]:
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "T"},
|
|
)
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
async def _register_admin_and_login(client: TestClient, db_session: AsyncSession) -> dict[str, str]:
|
|
email = f"admin-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Admin"},
|
|
)
|
|
user = await UserRepository(db_session).get_by_email(email)
|
|
user.is_admin = True
|
|
await db_session.commit()
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
# --- Repository --------------------------------------------------------
|
|
|
|
|
|
async def test_repository_upsert_then_get_then_delete(db_session):
|
|
repo = SystemSecretRepository(db_session)
|
|
|
|
await repo.upsert(SystemSecretKey.TURNSTILE_SECRET, "encrypted-1")
|
|
row = await repo.get(SystemSecretKey.TURNSTILE_SECRET)
|
|
assert row is not None
|
|
assert row.encrypted_value == "encrypted-1"
|
|
|
|
await repo.upsert(SystemSecretKey.TURNSTILE_SECRET, "encrypted-2")
|
|
row = await repo.get(SystemSecretKey.TURNSTILE_SECRET)
|
|
assert row.encrypted_value == "encrypted-2" # updated in place, not duplicated
|
|
|
|
await repo.delete(SystemSecretKey.TURNSTILE_SECRET)
|
|
assert await repo.get(SystemSecretKey.TURNSTILE_SECRET) is None
|
|
|
|
|
|
# --- Service -------------------------------------------------------------
|
|
|
|
|
|
async def test_list_status_shows_all_keys_unconfigured_by_default(db_session):
|
|
settings = get_settings()
|
|
statuses = await system_secret_service.list_status(db_session, settings)
|
|
assert {s["key"] for s in statuses} == {
|
|
"turnstile_site_key",
|
|
"turnstile_secret",
|
|
"resend_api_key",
|
|
}
|
|
assert all(s["configured"] is False for s in statuses)
|
|
assert all(s["value"] is None for s in statuses)
|
|
|
|
|
|
async def test_set_secret_then_list_status_shows_it_configured(db_session):
|
|
settings = get_settings()
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SITE_KEY,
|
|
"0x-my-site-key",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
statuses = await system_secret_service.list_status(db_session, settings)
|
|
site_key = next(s for s in statuses if s["key"] == "turnstile_site_key")
|
|
assert site_key["configured"] is True
|
|
assert site_key["value"] == "0x-my-site-key"
|
|
|
|
|
|
async def test_set_blank_secret_clears_a_previously_set_one(db_session):
|
|
settings = get_settings()
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SECRET,
|
|
"0x-my-secret",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SECRET,
|
|
" ",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
statuses = await system_secret_service.list_status(db_session, settings)
|
|
secret = next(s for s in statuses if s["key"] == "turnstile_secret")
|
|
assert secret["configured"] is False
|
|
assert secret["value"] is None
|
|
|
|
|
|
async def test_get_effective_settings_falls_back_to_global_when_unset(db_session):
|
|
settings = get_settings().model_copy(update={"turnstile_site_key": "global-site-key"})
|
|
effective = await system_secret_service.get_effective_settings(db_session, settings)
|
|
assert effective.turnstile_site_key == "global-site-key"
|
|
|
|
|
|
async def test_get_effective_settings_overrides_only_the_keys_that_were_set(db_session):
|
|
settings = get_settings().model_copy(
|
|
update={"turnstile_site_key": "global-site-key", "turnstile_secret": "global-secret"}
|
|
)
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SITE_KEY,
|
|
"admin-site-key",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
effective = await system_secret_service.get_effective_settings(db_session, settings)
|
|
assert effective.turnstile_site_key == "admin-site-key"
|
|
assert effective.turnstile_secret == "global-secret" # untouched, no override set
|
|
|
|
|
|
async def test_admin_configured_resend_key_overrides_the_env_value(db_session):
|
|
settings = get_settings().model_copy(update={"resend_api_key": "global-resend-key"})
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.RESEND_API_KEY,
|
|
"admin-resend-key",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
effective = await system_secret_service.get_effective_settings(db_session, settings)
|
|
assert effective.resend_api_key == "admin-resend-key"
|
|
|
|
|
|
# --- Endpoints -----------------------------------------------------------
|
|
|
|
|
|
def test_list_system_secrets_requires_auth(client: TestClient):
|
|
resp = client.get("/api/v1/system/secrets")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_list_system_secrets_non_admin_forbidden(client: TestClient):
|
|
headers = _register_and_login(client)
|
|
resp = client.get("/api/v1/system/secrets", headers=headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_admin_can_list_and_set_secrets_even_from_a_non_loopback_client(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
"""Deliberately different from the old /system/api-keys this replaces:
|
|
admin-gated only, no additional is_localhost requirement - the default
|
|
TestClient here has a non-loopback fake peer."""
|
|
headers = await _register_admin_and_login(client, db_session)
|
|
|
|
initial = client.get("/api/v1/system/secrets", headers=headers)
|
|
assert initial.status_code == 200
|
|
assert all(not s["configured"] for s in initial.json())
|
|
|
|
set_resp = client.put(
|
|
"/api/v1/system/secrets/turnstile_secret",
|
|
json={"value": "sk-set-via-api"},
|
|
headers=headers,
|
|
)
|
|
assert set_resp.status_code == 200
|
|
assert set_resp.json()["configured"] is True
|
|
assert set_resp.json()["value"] == "sk-set-via-api"
|
|
|
|
after = client.get("/api/v1/system/secrets", headers=headers)
|
|
secret = next(s for s in after.json() if s["key"] == "turnstile_secret")
|
|
assert secret["configured"] is True
|
|
assert secret["value"] == "sk-set-via-api"
|
|
|
|
|
|
async def test_updating_a_server_secret_is_logged_to_the_acting_admins_account_activity(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
headers = await _register_admin_and_login(client, db_session)
|
|
|
|
client.put(
|
|
"/api/v1/system/secrets/turnstile_secret",
|
|
json={"value": "sk-set-via-api"},
|
|
headers=headers,
|
|
)
|
|
|
|
events = client.get("/api/v1/auth/security-events", headers=headers).json()
|
|
assert any(e["event_type"] == "server_secret_updated" for e in events)
|
|
|
|
|
|
async def test_set_system_secret_rejects_unknown_key(client: TestClient, db_session: AsyncSession):
|
|
headers = await _register_admin_and_login(client, db_session)
|
|
resp = client.put("/api/v1/system/secrets/not-a-real-key", json={"value": "x"}, headers=headers)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
# --- /system/status exposes the site key live -------------------------
|
|
|
|
|
|
async def test_system_status_exposes_admin_configured_turnstile_site_key(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
settings = get_settings()
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SITE_KEY,
|
|
"admin-set-site-key",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
resp = client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["turnstile_site_key"] == "admin-set-site-key"
|
|
|
|
|
|
def test_system_status_turnstile_site_key_is_null_when_unconfigured(client: TestClient):
|
|
resp = client.get("/api/v1/system/status")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["turnstile_site_key"] is None
|
|
|
|
|
|
# --- End-to-end: a DB-only secret (no .env value) drives enforcement ----
|
|
|
|
|
|
async def test_register_requires_turnstile_when_only_db_configured_secret_exists(
|
|
client: TestClient, db_session: AsyncSession
|
|
):
|
|
"""Proves _enforce_turnstile actually resolves effective (DB-aware)
|
|
settings, not just the raw .env-backed global Settings object."""
|
|
settings = get_settings()
|
|
assert not settings.turnstile_secret # sanity: nothing set in .env for this test run
|
|
await system_secret_service.set_secret(
|
|
db_session,
|
|
SystemSecretKey.TURNSTILE_SECRET,
|
|
"admin-set-secret",
|
|
settings,
|
|
admin_user_id=uuid.uuid4(),
|
|
client_ip="10.0.0.1",
|
|
)
|
|
|
|
no_token_resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": _unique_email(), "password": "correct-horse-1", "display_name": "T"},
|
|
)
|
|
assert no_token_resp.status_code == 400
|
|
|
|
with respx.mock:
|
|
respx.post("https://challenges.cloudflare.com/turnstile/v0/siteverify").mock(
|
|
return_value=httpx.Response(200, json={"success": True})
|
|
)
|
|
with_token_resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": _unique_email(),
|
|
"password": "correct-horse-1",
|
|
"display_name": "T",
|
|
"turnstile_token": "good-token",
|
|
},
|
|
)
|
|
assert with_token_resp.status_code == 201
|