saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
"""Server-wide secrets an admin can configure from the Settings page
|
|
instead of only via .env - today just the Cloudflare Turnstile site key
|
|
and secret (app/services/turnstile_service.py). Storage is encrypted at
|
|
rest (app/core/crypto.py). Unlike per-user API keys
|
|
(user_api_key_service.py), there's exactly one value per key, shared by
|
|
the whole app - visible/editable only to admins (see
|
|
app/api/v1/system.py's require_admin gate), never per-user.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import Settings
|
|
from app.core.crypto import decrypt_secret, encrypt_secret
|
|
from app.models.enums import SecurityEventType, SystemSecretKey
|
|
from app.repositories.system_secret_repository import SystemSecretRepository
|
|
from app.repositories.user_security_event_repository import UserSecurityEventRepository
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SecretMeta:
|
|
label: str
|
|
settings_field: str
|
|
|
|
|
|
META: dict[SystemSecretKey, SecretMeta] = {
|
|
SystemSecretKey.TURNSTILE_SITE_KEY: SecretMeta(
|
|
label="Cloudflare Turnstile Site Key", settings_field="turnstile_site_key"
|
|
),
|
|
SystemSecretKey.TURNSTILE_SECRET: SecretMeta(
|
|
label="Cloudflare Turnstile Secret Key", settings_field="turnstile_secret"
|
|
),
|
|
}
|
|
|
|
|
|
async def list_status(db: AsyncSession, settings: Settings) -> list[dict[str, Any]]:
|
|
repo = SystemSecretRepository(db)
|
|
stored = {row.key: row for row in await repo.list_all()}
|
|
|
|
results: list[dict[str, Any]] = []
|
|
for key, meta in META.items():
|
|
row = stored.get(key)
|
|
value = decrypt_secret(row.encrypted_value, settings) if row is not None else None
|
|
results.append(
|
|
{
|
|
"key": key.value,
|
|
"label": meta.label,
|
|
"configured": row is not None,
|
|
"value": value,
|
|
}
|
|
)
|
|
return results
|
|
|
|
|
|
async def set_secret(
|
|
db: AsyncSession,
|
|
key: SystemSecretKey,
|
|
plaintext_value: str,
|
|
settings: Settings,
|
|
*,
|
|
admin_user_id: uuid.UUID,
|
|
client_ip: str,
|
|
) -> None:
|
|
"""A blank value clears the stored override, falling back to the
|
|
server's .env-configured value again. Every update - including a clear -
|
|
is logged to the acting admin's own Account activity, since this is a
|
|
security-sensitive, app-wide change (Turnstile keys today)."""
|
|
repo = SystemSecretRepository(db)
|
|
stripped = plaintext_value.strip()
|
|
if not stripped:
|
|
await repo.delete(key)
|
|
else:
|
|
await repo.upsert(key, encrypt_secret(stripped, settings))
|
|
await UserSecurityEventRepository(db).create(
|
|
user_id=admin_user_id,
|
|
event_type=SecurityEventType.SERVER_SECRET_UPDATED,
|
|
ip_address=client_ip,
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def get_effective_settings(db: AsyncSession, settings: Settings) -> Settings:
|
|
"""A copy of the global settings with any admin-stored secret
|
|
substituted in for the matching field - keys with no stored override
|
|
keep using the server's .env-configured default."""
|
|
rows = await SystemSecretRepository(db).list_all()
|
|
if not rows:
|
|
return settings
|
|
overrides = {
|
|
META[row.key].settings_field: decrypt_secret(row.encrypted_value, settings) for row in rows
|
|
}
|
|
return settings.model_copy(update=overrides)
|