316 lines
11 KiB
Python
316 lines
11 KiB
Python
import smtplib
|
|
from dataclasses import dataclass
|
|
from email.message import EmailMessage
|
|
|
|
from pydantic import EmailStr
|
|
from sqlalchemy.orm import Session
|
|
from starlette.requests import Request
|
|
|
|
from app.core.config import settings
|
|
from app.models.user import User
|
|
from app.repositories.system_settings_repository import SystemSettingsRepository
|
|
from app.schemas.system_setting import (
|
|
PublicLinksSettingsResponse,
|
|
PublicLinksSettingsUpdate,
|
|
SmtpSettingsResponse,
|
|
SmtpSettingsUpdate,
|
|
SmtpTestResponse,
|
|
SettingsSource,
|
|
)
|
|
from app.services.audit_service import write_audit_log
|
|
|
|
|
|
SMTP_KEYS = (
|
|
"smtp.host",
|
|
"smtp.port",
|
|
"smtp.username",
|
|
"smtp.password",
|
|
"smtp.from_email",
|
|
"smtp.from_name",
|
|
"smtp.use_tls",
|
|
"smtp.enabled",
|
|
)
|
|
|
|
SECRET_KEYS = {"smtp.password"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SmtpRuntimeConfig:
|
|
host: str
|
|
port: int
|
|
username: str
|
|
password: str
|
|
from_email: str
|
|
from_name: str
|
|
use_tls: bool
|
|
enabled: bool
|
|
source: SettingsSource
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
return self.enabled and bool(self.host and self.from_email)
|
|
|
|
@property
|
|
def password_is_set(self) -> bool:
|
|
return bool(self.password)
|
|
|
|
@property
|
|
def from_header(self) -> str:
|
|
if self.from_name:
|
|
return f"{self.from_name} <{self.from_email}>"
|
|
return self.from_email
|
|
|
|
|
|
def parse_bool(value: str | None, *, default: bool = False) -> bool:
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def parse_int(value: str | None, *, default: int) -> int:
|
|
try:
|
|
parsed = int(str(value or "").strip())
|
|
except ValueError:
|
|
return default
|
|
if parsed < 1 or parsed > 65535:
|
|
return default
|
|
return parsed
|
|
|
|
|
|
class SystemSettingsService:
|
|
@staticmethod
|
|
def get_smtp_settings(db: Session) -> SmtpSettingsResponse:
|
|
return SystemSettingsService.smtp_response(SystemSettingsService.get_smtp_runtime_config(db))
|
|
|
|
@staticmethod
|
|
def get_smtp_runtime_config(db: Session) -> SmtpRuntimeConfig:
|
|
values = SystemSettingsService._get_values(db, SMTP_KEYS)
|
|
database_enabled = parse_bool(values.get("smtp.enabled"), default=False)
|
|
database_host = values.get("smtp.host", "")
|
|
database_from_email = values.get("smtp.from_email", "")
|
|
|
|
if database_enabled and database_host and database_from_email:
|
|
return SmtpRuntimeConfig(
|
|
host=database_host,
|
|
port=parse_int(values.get("smtp.port"), default=587),
|
|
username=values.get("smtp.username", ""),
|
|
password=values.get("smtp.password", ""),
|
|
from_email=database_from_email,
|
|
from_name=values.get("smtp.from_name", "Funktechnik Schubert") or "Funktechnik Schubert",
|
|
use_tls=parse_bool(values.get("smtp.use_tls"), default=True),
|
|
enabled=True,
|
|
source="database",
|
|
)
|
|
|
|
if settings.smtp_host and settings.smtp_from_email:
|
|
return SmtpRuntimeConfig(
|
|
host=settings.smtp_host,
|
|
port=settings.smtp_port,
|
|
username=settings.smtp_username or "",
|
|
password=settings.smtp_password or "",
|
|
from_email=settings.smtp_from_email,
|
|
from_name=settings.smtp_from_name,
|
|
use_tls=settings.smtp_use_tls,
|
|
enabled=True,
|
|
source="environment",
|
|
)
|
|
|
|
return SmtpRuntimeConfig(
|
|
host=database_host or settings.smtp_host or "",
|
|
port=parse_int(values.get("smtp.port"), default=settings.smtp_port),
|
|
username=values.get("smtp.username", "") or settings.smtp_username or "",
|
|
password=values.get("smtp.password", "") or settings.smtp_password or "",
|
|
from_email=database_from_email or settings.smtp_from_email or "",
|
|
from_name=values.get("smtp.from_name", "") or settings.smtp_from_name,
|
|
use_tls=parse_bool(values.get("smtp.use_tls"), default=settings.smtp_use_tls),
|
|
enabled=False,
|
|
source="missing",
|
|
)
|
|
|
|
@staticmethod
|
|
def update_smtp_settings(
|
|
db: Session,
|
|
payload: SmtpSettingsUpdate,
|
|
*,
|
|
actor: User,
|
|
request: Request,
|
|
) -> SmtpSettingsResponse:
|
|
current_values = SystemSettingsService._get_values(db, SMTP_KEYS)
|
|
password = payload.password if payload.password else current_values.get("smtp.password", "")
|
|
|
|
updates = {
|
|
"smtp.host": payload.host,
|
|
"smtp.port": str(payload.port),
|
|
"smtp.username": payload.username,
|
|
"smtp.password": password,
|
|
"smtp.from_email": payload.from_email,
|
|
"smtp.from_name": payload.from_name or "Funktechnik Schubert",
|
|
"smtp.use_tls": "true" if payload.use_tls else "false",
|
|
"smtp.enabled": "true" if payload.enabled else "false",
|
|
}
|
|
|
|
for key, value in updates.items():
|
|
SystemSettingsRepository.upsert(db, key=key, value=value, is_secret=key in SECRET_KEYS)
|
|
db.commit()
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="system_settings.smtp.update",
|
|
entity_type="system_settings",
|
|
entity_label="SMTP-Konfiguration",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={
|
|
"enabled": payload.enabled,
|
|
"host": payload.host,
|
|
"from_email": payload.from_email,
|
|
"password_changed": bool(payload.password),
|
|
},
|
|
)
|
|
return SystemSettingsService.get_smtp_settings(db)
|
|
|
|
@staticmethod
|
|
def get_public_links_settings(db: Session) -> PublicLinksSettingsResponse:
|
|
value = SystemSettingsService.get_public_repair_status_base_url(db)
|
|
source: SettingsSource = "missing"
|
|
if value:
|
|
db_value = SystemSettingsRepository.get(db, "public.repair_status_base_url")
|
|
source = "database" if db_value and db_value.value.strip() else "environment"
|
|
return PublicLinksSettingsResponse(repair_status_base_url=value, source=source)
|
|
|
|
@staticmethod
|
|
def get_public_repair_status_base_url(db: Session) -> str:
|
|
setting = SystemSettingsRepository.get(db, "public.repair_status_base_url")
|
|
if setting and setting.value.strip():
|
|
return setting.value.strip().rstrip("/")
|
|
return (settings.public_repair_status_base_url or "").strip().rstrip("/")
|
|
|
|
@staticmethod
|
|
def update_public_links_settings(
|
|
db: Session,
|
|
payload: PublicLinksSettingsUpdate,
|
|
*,
|
|
actor: User,
|
|
request: Request,
|
|
) -> PublicLinksSettingsResponse:
|
|
SystemSettingsRepository.upsert(
|
|
db,
|
|
key="public.repair_status_base_url",
|
|
value=payload.repair_status_base_url,
|
|
is_secret=False,
|
|
)
|
|
db.commit()
|
|
write_audit_log(
|
|
db,
|
|
action="system_settings.public_links.update",
|
|
entity_type="system_settings",
|
|
entity_label="Öffentliche Links",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={"repair_status_base_url": payload.repair_status_base_url},
|
|
)
|
|
return SystemSettingsService.get_public_links_settings(db)
|
|
|
|
@staticmethod
|
|
def send_test_mail(
|
|
db: Session,
|
|
recipient: EmailStr,
|
|
*,
|
|
actor: User,
|
|
request: Request,
|
|
) -> SmtpTestResponse:
|
|
smtp_config = SystemSettingsService.get_smtp_runtime_config(db)
|
|
if not smtp_config.is_configured:
|
|
write_audit_log(
|
|
db,
|
|
action="system_settings.smtp.test_failed",
|
|
entity_type="system_settings",
|
|
entity_label="SMTP-Testmail",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={"reason": "smtp_not_configured", "source": smtp_config.source},
|
|
)
|
|
return SmtpTestResponse(success=False, message="SMTP ist nicht vollständig konfiguriert.", source=smtp_config.source)
|
|
|
|
try:
|
|
SystemSettingsService.send_email(
|
|
smtp_config,
|
|
recipient=str(recipient),
|
|
subject="Olympus CRM SMTP-Test",
|
|
text=(
|
|
"Hallo,\n\n"
|
|
"dies ist eine Testmail aus Olympus CRM. "
|
|
"Die SMTP-Konfiguration ist grundsätzlich erreichbar.\n\n"
|
|
"Olympus CRM"
|
|
),
|
|
html=(
|
|
"<p>Hallo,</p>"
|
|
"<p>dies ist eine Testmail aus Olympus CRM. "
|
|
"Die SMTP-Konfiguration ist grundsätzlich erreichbar.</p>"
|
|
"<p>Olympus CRM</p>"
|
|
),
|
|
)
|
|
except Exception:
|
|
write_audit_log(
|
|
db,
|
|
action="system_settings.smtp.test_failed",
|
|
entity_type="system_settings",
|
|
entity_label="SMTP-Testmail",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={"reason": "smtp_send_failed", "source": smtp_config.source},
|
|
)
|
|
return SmtpTestResponse(success=False, message="Testmail konnte nicht versendet werden.", source=smtp_config.source)
|
|
|
|
write_audit_log(
|
|
db,
|
|
action="system_settings.smtp.test_sent",
|
|
entity_type="system_settings",
|
|
entity_label="SMTP-Testmail",
|
|
actor=actor,
|
|
request=request,
|
|
metadata={"source": smtp_config.source},
|
|
)
|
|
return SmtpTestResponse(success=True, message="Testmail wurde versendet.", source=smtp_config.source)
|
|
|
|
@staticmethod
|
|
def send_email(
|
|
smtp_config: SmtpRuntimeConfig,
|
|
*,
|
|
recipient: str,
|
|
subject: str,
|
|
text: str,
|
|
html: str,
|
|
) -> None:
|
|
message = EmailMessage()
|
|
message["Subject"] = subject
|
|
message["From"] = smtp_config.from_header
|
|
message["To"] = recipient
|
|
message.set_content(text)
|
|
message.add_alternative(html, subtype="html")
|
|
|
|
with smtplib.SMTP(smtp_config.host, smtp_config.port, timeout=15) as smtp:
|
|
if smtp_config.use_tls:
|
|
smtp.starttls()
|
|
if smtp_config.username and smtp_config.password:
|
|
smtp.login(smtp_config.username, smtp_config.password)
|
|
smtp.send_message(message)
|
|
|
|
@staticmethod
|
|
def smtp_response(smtp_config: SmtpRuntimeConfig) -> SmtpSettingsResponse:
|
|
return SmtpSettingsResponse(
|
|
host=smtp_config.host,
|
|
port=smtp_config.port,
|
|
username=smtp_config.username,
|
|
password_is_set=smtp_config.password_is_set,
|
|
from_email=smtp_config.from_email,
|
|
from_name=smtp_config.from_name,
|
|
use_tls=smtp_config.use_tls,
|
|
enabled=smtp_config.enabled,
|
|
source=smtp_config.source,
|
|
)
|
|
|
|
@staticmethod
|
|
def _get_values(db: Session, keys: tuple[str, ...]) -> dict[str, str]:
|
|
settings_by_key = SystemSettingsRepository.get_many(db, keys)
|
|
return {key: settings_by_key[key].value.strip() for key in keys if key in settings_by_key}
|