nessi
e7d288ff12
Replaced boolean `smtp_use_tls` with explicit `smtp_security` enum field supporting "none", "starttls", and "ssl" modes. Added database migration to create `smtp_security` column with "starttls" default. Updated mailer to use SMTP_SSL client for direct SSL connections on port 465 and SMTP with STARTTLS for port 587. Modified admin settings UI to show dropdown selector with encryption options and auto-adjust
219 lines
8.0 KiB
Python
219 lines
8.0 KiB
Python
import hashlib
|
||
import secrets
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||
from sqlalchemy.orm import Session
|
||
from ..db import get_db
|
||
from ..mailer import send_html_email, send_user_invite
|
||
from ..models import AppSettings, InviteToken, User, Role
|
||
from ..security import hash_password, get_session_user_id
|
||
|
||
router = APIRouter(prefix="/admin", tags=["admin"])
|
||
|
||
def require_admin(req: Request, db: Session) -> User:
|
||
uid = get_session_user_id(req)
|
||
if not uid:
|
||
raise HTTPException(status_code=401, detail="not logged in")
|
||
user = db.query(User).filter(User.id == uid).first()
|
||
if not user or user.role != Role.admin.value:
|
||
raise HTTPException(status_code=403, detail="forbidden")
|
||
return user
|
||
|
||
@router.get("/users")
|
||
def list_users(req: Request, db: Session = Depends(get_db)):
|
||
require_admin(req, db)
|
||
users = db.query(User).order_by(User.created_at.desc()).all()
|
||
return [
|
||
{
|
||
"id": u.id,
|
||
"email": u.email,
|
||
"display_name": u.display_name,
|
||
"role": u.role,
|
||
"disabled": u.disabled,
|
||
}
|
||
for u in users
|
||
]
|
||
|
||
@router.post("/users")
|
||
def create_user(req: Request, data: dict, db: Session = Depends(get_db)):
|
||
require_admin(req, db)
|
||
email = (data.get("email") or "").lower().strip()
|
||
password = data.get("password") or ""
|
||
display_name = (data.get("display_name") or "").strip()
|
||
|
||
if not email or not display_name:
|
||
raise HTTPException(400, "name/email required")
|
||
if db.query(User).filter(User.email == email).first():
|
||
raise HTTPException(409, "email exists")
|
||
|
||
role = data.get("role")
|
||
if not role:
|
||
raise HTTPException(400, "role required")
|
||
if role not in (Role.admin.value, Role.user.value):
|
||
raise HTTPException(400, "invalid role")
|
||
|
||
settings = db.query(AppSettings).filter(AppSettings.id == "default").first() or AppSettings(id="default")
|
||
invite_sent = not password
|
||
raw_token = None
|
||
|
||
if invite_sent:
|
||
if not settings.smtp_host or not settings.smtp_from_email:
|
||
raise HTTPException(400, "configure SMTP settings before sending invites")
|
||
raw_token = secrets.token_urlsafe(32)
|
||
password = secrets.token_urlsafe(32)
|
||
|
||
u = User(email=email, password_hash=hash_password(password), role=role, display_name=display_name)
|
||
db.add(u)
|
||
db.flush()
|
||
|
||
if invite_sent:
|
||
invite = InviteToken(
|
||
user_id=u.id,
|
||
token_hash=hashlib.sha256(raw_token.encode()).hexdigest(),
|
||
expires_at=datetime.now(timezone.utc) + timedelta(hours=48),
|
||
)
|
||
db.add(invite)
|
||
invite_url = f"{(settings.app_base_url or 'http://localhost:8081').rstrip('/')}/invite/{raw_token}"
|
||
try:
|
||
send_user_invite(settings, u, invite_url)
|
||
except Exception as exc:
|
||
db.rollback()
|
||
raise HTTPException(502, f"invite email could not be sent: {exc}")
|
||
|
||
db.commit()
|
||
return {"ok": True, "id": u.id, "invite_sent": invite_sent}
|
||
|
||
@router.delete("/users/{user_id}")
|
||
def delete_user(req: Request, user_id: str, db: Session = Depends(get_db)):
|
||
admin = require_admin(req, db)
|
||
|
||
if admin.id == user_id:
|
||
raise HTTPException(400, "cannot delete yourself")
|
||
|
||
u = db.query(User).filter(User.id == user_id).first()
|
||
if not u:
|
||
raise HTTPException(404, "not found")
|
||
|
||
# soft delete
|
||
u.disabled = True
|
||
db.add(u)
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
@router.patch("/users/{user_id}")
|
||
def update_user(req: Request, user_id: str, data: dict, db: Session = Depends(get_db)):
|
||
admin = require_admin(req, db)
|
||
user = db.query(User).filter(User.id == user_id).first()
|
||
if not user:
|
||
raise HTTPException(404, "not found")
|
||
|
||
email = (data.get("email") or user.email).lower().strip()
|
||
display_name = (data.get("display_name") if "display_name" in data else user.display_name or "").strip()
|
||
role = data.get("role") or user.role
|
||
password = data.get("password") or ""
|
||
|
||
if not email or "@" not in email:
|
||
raise HTTPException(400, "valid email required")
|
||
if role not in (Role.admin.value, Role.user.value):
|
||
raise HTTPException(400, "invalid role")
|
||
if password and len(password) < 8:
|
||
raise HTTPException(400, "password too short (min 8)")
|
||
if admin.id == user_id and role != Role.admin.value:
|
||
raise HTTPException(400, "cannot demote yourself")
|
||
if admin.id == user_id and data.get("disabled") is True:
|
||
raise HTTPException(400, "cannot disable yourself")
|
||
|
||
duplicate = db.query(User).filter(User.email == email, User.id != user_id).first()
|
||
if duplicate:
|
||
raise HTTPException(409, "email exists")
|
||
|
||
user.email = email
|
||
user.display_name = display_name
|
||
user.role = role
|
||
if password:
|
||
user.password_hash = hash_password(password)
|
||
if "disabled" in data:
|
||
user.disabled = bool(data["disabled"])
|
||
|
||
db.add(user)
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
def get_settings(db: Session) -> AppSettings:
|
||
settings = db.query(AppSettings).filter(AppSettings.id == "default").first()
|
||
if not settings:
|
||
settings = AppSettings(id="default")
|
||
db.add(settings)
|
||
db.commit()
|
||
db.refresh(settings)
|
||
return settings
|
||
|
||
|
||
@router.get("/settings/smtp")
|
||
def read_smtp_settings(req: Request, db: Session = Depends(get_db)):
|
||
require_admin(req, db)
|
||
settings = get_settings(db)
|
||
return {
|
||
"smtp_host": settings.smtp_host,
|
||
"smtp_port": settings.smtp_port,
|
||
"smtp_username": settings.smtp_username,
|
||
"smtp_password_configured": bool(settings.smtp_password),
|
||
"smtp_from_email": settings.smtp_from_email,
|
||
"smtp_from_name": settings.smtp_from_name,
|
||
"smtp_security": getattr(settings, "smtp_security", "starttls") or ("starttls" if settings.smtp_use_tls else "none"),
|
||
"smtp_use_tls": settings.smtp_use_tls,
|
||
"app_base_url": settings.app_base_url,
|
||
}
|
||
|
||
|
||
@router.patch("/settings/smtp")
|
||
def update_smtp_settings(req: Request, data: dict, db: Session = Depends(get_db)):
|
||
require_admin(req, db)
|
||
settings = get_settings(db)
|
||
|
||
settings.smtp_host = (data.get("smtp_host") or "").strip()
|
||
try:
|
||
settings.smtp_port = int(data.get("smtp_port") or 587)
|
||
except (TypeError, ValueError):
|
||
raise HTTPException(400, "SMTP port must be a number")
|
||
settings.smtp_username = (data.get("smtp_username") or "").strip()
|
||
settings.smtp_from_email = (data.get("smtp_from_email") or "").strip()
|
||
settings.smtp_from_name = (data.get("smtp_from_name") or "Cluedo HP").strip()
|
||
security = data.get("smtp_security") or ("starttls" if data.get("smtp_use_tls", True) else "none")
|
||
if security not in ("none", "starttls", "ssl"):
|
||
raise HTTPException(400, "invalid SMTP security mode")
|
||
settings.smtp_security = security
|
||
settings.smtp_use_tls = security == "starttls"
|
||
settings.app_base_url = (data.get("app_base_url") or "http://localhost:8081").strip().rstrip("/")
|
||
if "smtp_password" in data and data.get("smtp_password"):
|
||
settings.smtp_password = data["smtp_password"]
|
||
|
||
if settings.smtp_host and not settings.smtp_from_email:
|
||
raise HTTPException(400, "from email required when SMTP is configured")
|
||
db.add(settings)
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
@router.post("/settings/smtp/test")
|
||
def test_smtp_settings(req: Request, data: dict, db: Session = Depends(get_db)):
|
||
require_admin(req, db)
|
||
settings = get_settings(db)
|
||
recipient = (data.get("recipient") or "").strip()
|
||
if not recipient:
|
||
raise HTTPException(400, "recipient required")
|
||
try:
|
||
send_html_email(
|
||
settings,
|
||
recipient,
|
||
"SMTP-Test – Cluedo HP",
|
||
"<div style='font-family:Georgia,serif;padding:24px;background:#17161b;color:#f5efdc'><h2 style='color:#e9d8a6'>SMTP funktioniert</h2><p>Der Einladungsversand ist bereit.</p></div>",
|
||
"SMTP funktioniert. Der Einladungsversand ist bereit.",
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(502, f"SMTP test failed: {exc}")
|
||
return {"ok": True}
|