Neuer Router /admin/tls (SUPER_ADMIN only, AuditLog, Rate-Limit 5/hour): - GET /admin/tls/status – erkennt proxy/certbot/internal-Modus, liest Ablaufdatum via openssl x509 -enddate - POST /admin/tls/renew/certbot – ruft setup-tls.sh <domain> auf - POST /admin/tls/renew/internal – ruft setup-tls-internal.sh <hostname> [ip] auf, reloaded nginx danach Läuft mit den Root-Rechten des bestehenden timemaster.service (User=root, unverändert) - Angriffsfläche dadurch begrenzt auf SUPER_ADMIN-Auth + Domain/Hostname-Validierung (Regex, kein Shell-Interpolieren, subprocess mit Argument-Liste statt shell=True). Frontend: neuer Tab "Server / TLS" in TenantsPage – Status-Anzeige + zwei Formulare (öffentlich/intern). 3 neue Tests in test_tls_admin.py (Rollen-Gate, Status im Testcontext, Input-Validierung). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LTxkZEUdfgMxZvHPiZJ8bV
93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""SUPER_ADMIN: TLS-Zertifikat-Status + manuelles Renewal (server-lokal).
|
||
|
||
Läuft mit Root-Rechten (timemaster.service User=root), deshalb strikt auf
|
||
SUPER_ADMIN begrenzt + AuditLog + Rate-Limit. Kein Cross-Server-Wissen –
|
||
zeigt nur das Zertifikat DIESES Servers.
|
||
"""
|
||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.database import get_db
|
||
from app.core.dependencies import get_client_ip, require_role
|
||
from app.core.limiter import limiter
|
||
from app.models.audit_log import AuditLog
|
||
from app.models.user import User, UserRole
|
||
from app.services import tls_service
|
||
|
||
router = APIRouter(prefix="/admin/tls", tags=["Admin · TLS"])
|
||
|
||
_sa = (UserRole.SUPER_ADMIN,)
|
||
|
||
|
||
class TlsStatusOut(BaseModel):
|
||
mode: str
|
||
domain: str | None
|
||
valid_until: str | None
|
||
renewable: bool
|
||
ca_cert_path: str | None = None
|
||
|
||
|
||
class TlsRenewCertbot(BaseModel):
|
||
domain: str
|
||
|
||
|
||
class TlsRenewInternal(BaseModel):
|
||
hostname: str
|
||
ip: str | None = None
|
||
|
||
|
||
@router.get("/status", response_model=TlsStatusOut)
|
||
async def tls_status(current_user: User = require_role(*_sa)):
|
||
return tls_service.get_tls_status()
|
||
|
||
|
||
@router.post("/renew/certbot", response_model=TlsStatusOut)
|
||
@limiter.limit("5/hour")
|
||
async def tls_renew_certbot(
|
||
request: Request,
|
||
data: TlsRenewCertbot,
|
||
current_user: User = require_role(*_sa),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
try:
|
||
tls_service.renew_certbot(data.domain)
|
||
except ValueError as e:
|
||
raise HTTPException(422, str(e))
|
||
except RuntimeError as e:
|
||
raise HTTPException(500, f"Zertifikat-Renewal fehlgeschlagen: {e}")
|
||
|
||
db.add(AuditLog(
|
||
company_id=None, user_id=current_user.id,
|
||
action="tls_renewed_certbot", entity_type="server", entity_id=None,
|
||
new_value={"domain": data.domain},
|
||
ip=get_client_ip(request),
|
||
))
|
||
await db.commit()
|
||
return tls_service.get_tls_status()
|
||
|
||
|
||
@router.post("/renew/internal", response_model=TlsStatusOut)
|
||
@limiter.limit("5/hour")
|
||
async def tls_renew_internal(
|
||
request: Request,
|
||
data: TlsRenewInternal,
|
||
current_user: User = require_role(*_sa),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
try:
|
||
tls_service.renew_internal(data.hostname, data.ip)
|
||
except ValueError as e:
|
||
raise HTTPException(422, str(e))
|
||
except RuntimeError as e:
|
||
raise HTTPException(500, f"Zertifikat-Renewal fehlgeschlagen: {e}")
|
||
|
||
db.add(AuditLog(
|
||
company_id=None, user_id=current_user.id,
|
||
action="tls_renewed_internal", entity_type="server", entity_id=None,
|
||
new_value={"hostname": data.hostname, "ip": data.ip},
|
||
ip=get_client_ip(request),
|
||
))
|
||
await db.commit()
|
||
return tls_service.get_tls_status()
|