feat(admin): SUPER_ADMIN TLS-Zertifikat-Status + Renewal-Trigger
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
This commit is contained in:
Claude Sonnet 5
@@ -20,6 +20,7 @@ from app.routers import hours_payouts
|
||||
from app.routers import public_stamp
|
||||
from app.routers import ical
|
||||
from app.routers import reseller, tenants
|
||||
from app.routers import tls_admin
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
@@ -112,6 +113,7 @@ app.include_router(special_assignments.router, prefix=API_PREFIX)
|
||||
app.include_router(hours_payouts.router, prefix=API_PREFIX)
|
||||
app.include_router(reseller.router, prefix=API_PREFIX)
|
||||
app.include_router(tenants.router, prefix=API_PREFIX)
|
||||
app.include_router(tls_admin.router, prefix=API_PREFIX)
|
||||
|
||||
|
||||
# ── Health ────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,96 @@
|
||||
"""TLS-Status + Renewal für SUPER_ADMIN.
|
||||
|
||||
Server-lokal – jede Installation (137/164) kennt nur ihr eigenes Zertifikat,
|
||||
kein Cross-Server-Wissen (siehe CLAUDE.md: Server sind entkoppelt).
|
||||
|
||||
Erkennt drei Modi:
|
||||
- "proxy" – kein lokal verwaltetes Zertifikat, vorgeschalteter Proxy
|
||||
terminiert TLS (aktueller Standardfall)
|
||||
- "certbot" – öffentliches Let's-Encrypt-Zertifikat (setup-tls.sh)
|
||||
- "internal" – eigene interne CA (setup-tls-internal.sh)
|
||||
"""
|
||||
import re
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
LETSENCRYPT_LIVE = Path("/etc/letsencrypt/live")
|
||||
INTERNAL_CA_DIR = Path("/etc/nginx/internal-ca")
|
||||
SETUP_TLS_SCRIPT = Path("/opt/timemaster/setup-tls.sh")
|
||||
SETUP_TLS_INTERNAL_SCRIPT = Path("/opt/timemaster/setup-tls-internal.sh")
|
||||
|
||||
_DOMAIN_RE = re.compile(r"^[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)+$")
|
||||
_IP_RE = re.compile(r"^\d{1,3}(\.\d{1,3}){3}$")
|
||||
|
||||
|
||||
def _cert_enddate(cert_path: Path) -> str | None:
|
||||
try:
|
||||
out = subprocess.run(
|
||||
["openssl", "x509", "-enddate", "-noout", "-in", str(cert_path)],
|
||||
capture_output=True, text=True, timeout=5, check=True,
|
||||
)
|
||||
except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
|
||||
return None
|
||||
# Format: "notAfter=Aug 5 12:00:00 2027 GMT"
|
||||
return out.stdout.strip().removeprefix("notAfter=") or None
|
||||
|
||||
|
||||
def get_tls_status() -> dict[str, Any]:
|
||||
if LETSENCRYPT_LIVE.is_dir():
|
||||
for entry in LETSENCRYPT_LIVE.iterdir():
|
||||
cert = entry / "cert.pem"
|
||||
if cert.exists():
|
||||
return {
|
||||
"mode": "certbot",
|
||||
"domain": entry.name,
|
||||
"valid_until": _cert_enddate(cert),
|
||||
"renewable": True,
|
||||
}
|
||||
|
||||
internal_cert = INTERNAL_CA_DIR / "server.crt"
|
||||
if internal_cert.exists():
|
||||
return {
|
||||
"mode": "internal",
|
||||
"domain": None,
|
||||
"valid_until": _cert_enddate(internal_cert),
|
||||
"renewable": True,
|
||||
"ca_cert_path": str(INTERNAL_CA_DIR / "ca.crt"),
|
||||
}
|
||||
|
||||
return {
|
||||
"mode": "proxy",
|
||||
"domain": None,
|
||||
"valid_until": None,
|
||||
"renewable": False,
|
||||
}
|
||||
|
||||
|
||||
def renew_certbot(domain: str) -> str:
|
||||
if not _DOMAIN_RE.match(domain):
|
||||
raise ValueError("Ungültiger Domain-Name")
|
||||
if not SETUP_TLS_SCRIPT.exists():
|
||||
raise RuntimeError("setup-tls.sh nicht gefunden")
|
||||
result = subprocess.run(
|
||||
[str(SETUP_TLS_SCRIPT), domain],
|
||||
capture_output=True, text=True, timeout=120,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(result.stderr[-2000:] or result.stdout[-2000:])
|
||||
return result.stdout[-4000:]
|
||||
|
||||
|
||||
def renew_internal(hostname: str, ip: str | None) -> str:
|
||||
if not _DOMAIN_RE.match(hostname) and not _IP_RE.match(hostname):
|
||||
raise ValueError("Ungültiger Hostname")
|
||||
if ip and not _IP_RE.match(ip):
|
||||
raise ValueError("Ungültige IP")
|
||||
if not SETUP_TLS_INTERNAL_SCRIPT.exists():
|
||||
raise RuntimeError("setup-tls-internal.sh nicht gefunden")
|
||||
args = [str(SETUP_TLS_INTERNAL_SCRIPT), hostname]
|
||||
if ip:
|
||||
args.append(ip)
|
||||
result = subprocess.run(args, capture_output=True, text=True, timeout=60)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(result.stderr[-2000:] or result.stdout[-2000:])
|
||||
subprocess.run(["systemctl", "reload", "nginx"], capture_output=True, timeout=10)
|
||||
return result.stdout[-4000:]
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Tests für SUPER_ADMIN TLS-Status/Renewal – Test-Server hat kein echtes
|
||||
Zertifikat, daher fällt get_tls_status() auf mode="proxy" zurück. Renewal-
|
||||
Endpunkte werden nur auf Rollen-Gate + Input-Validierung geprüft (echtes
|
||||
Skript-Ausführen ist Server-Ops, kein Unit-Test-Ziel)."""
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.core.security import hash_password
|
||||
from app.models.user import User, UserRole
|
||||
|
||||
|
||||
async def _make_user(db, *, email, role, company_id=None, password="Secret123"):
|
||||
await db.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
||||
user = User(
|
||||
id=uuid.uuid4(), company_id=company_id, email=email,
|
||||
password_hash=hash_password(password),
|
||||
first_name=role.title(), last_name="User",
|
||||
role=UserRole(role), is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
await db.commit()
|
||||
return user.id
|
||||
|
||||
|
||||
async def _login(client, email, password="Secret123"):
|
||||
r = await client.post("/api/v1/auth/login", json={"email": email, "password": password})
|
||||
assert r.status_code == 200, r.text
|
||||
return {"Authorization": f"Bearer {r.json()['access_token']}"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_tls_status_superadmin_only(client: AsyncClient, db_session):
|
||||
await _make_user(db_session, email="sa-tls@platform.de", role="SUPER_ADMIN")
|
||||
h = await _login(client, "sa-tls@platform.de")
|
||||
|
||||
r = await client.get("/api/v1/admin/tls/status", headers=h)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["mode"] == "proxy"
|
||||
assert r.json()["renewable"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_tls_status_forbidden_for_non_superadmin(client: AsyncClient, db_session):
|
||||
company_id = uuid.uuid4()
|
||||
from app.models.company import Company
|
||||
await db_session.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
||||
db_session.add(Company(id=company_id, name="Acme", slug=f"acme-{company_id.hex[:8]}"))
|
||||
await db_session.commit()
|
||||
await _make_user(db_session, email="admin-tls@acme.de", role="COMPANY_ADMIN", company_id=company_id)
|
||||
h = await _login(client, "admin-tls@acme.de")
|
||||
|
||||
r = await client.get("/api/v1/admin/tls/status", headers=h)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="session")
|
||||
async def test_tls_renew_certbot_rejects_invalid_domain(client: AsyncClient, db_session):
|
||||
await _make_user(db_session, email="sa-tls2@platform.de", role="SUPER_ADMIN")
|
||||
h = await _login(client, "sa-tls2@platform.de")
|
||||
|
||||
r = await client.post("/api/v1/admin/tls/renew/certbot", json={"domain": "; rm -rf /"}, headers=h)
|
||||
assert r.status_code == 422
|
||||
Reference in New Issue
Block a user