diff --git a/src/netmover/mail.py b/src/netmover/mail.py new file mode 100644 index 0000000..335350d --- /dev/null +++ b/src/netmover/mail.py @@ -0,0 +1,91 @@ +"""E-Mail-Benachrichtigung per SMTP. + +* Verschlüsselung: ``starttls`` (Port 587, Standard), ``ssl`` (Port 465) oder ``none`` + (nur für interne Relays ohne TLS). Zertifikate werden immer geprüft. +* Das Passwort liegt Fernet-verschlüsselt in den Einstellungen. +""" + +from __future__ import annotations + +import re +import smtplib +import socket +import ssl +from datetime import datetime +from email.message import EmailMessage +from email.utils import formatdate, make_msgid + +from netmover.settings_store import AppSettings + +SECURITY_MODES = ("starttls", "ssl", "none") +_ADDRESS = re.compile(r"^[^@\s,;<>]+@[^@\s,;<>]+\.[^@\s,;<>]+$") + + +class MailConfigError(ValueError): + pass + + +def parse_recipients(raw: str) -> list[str]: + recipients = [r.strip() for r in re.split(r"[,;\n]", raw) if r.strip()] + bad = [r for r in recipients if not _ADDRESS.fullmatch(r)] + if bad: + raise MailConfigError(f"Ungültige Adresse: {', '.join(bad)}") + return recipients + + +def valid_address(value: str) -> bool: + return bool(_ADDRESS.fullmatch(value.strip())) + + +def mail_configured(cfg: AppSettings) -> bool: + return bool(cfg.email_enabled and cfg.smtp_host and cfg.smtp_from and cfg.smtp_to) + + +def build_message(cfg: AppSettings, subject: str, body: str) -> EmailMessage: + msg = EmailMessage() + msg["Subject"] = subject + msg["From"] = cfg.smtp_from + msg["To"] = ", ".join(parse_recipients(cfg.smtp_to)) + msg["Date"] = formatdate(localtime=True) + msg["Message-ID"] = make_msgid(domain=cfg.smtp_from.rsplit("@", 1)[-1]) + msg["X-Mailer"] = "NetMover" + footer = ( + f"\n\n-- \nNetMover auf {socket.gethostname()}, " + f"{datetime.now().astimezone():%d.%m.%Y %H:%M:%S %Z}" + ) + msg.set_content(body + footer) + return msg + + +def tls_context() -> ssl.SSLContext: + """Zertifikatsprüfung gegen die System-CAs (eigene Funktion, damit Tests ihr + Test-Zertifikat einsetzen können, ohne die Prüfung abzuschalten).""" + return ssl.create_default_context() + + +def smtp_send( + cfg: AppSettings, password: str | None, subject: str, body: str, timeout: float = 20 +) -> None: + """Verschickt eine Nachricht; Fehler werden als Ausnahme weitergereicht.""" + if cfg.smtp_security not in SECURITY_MODES: + raise MailConfigError(f"Unbekannte Verschlüsselung: {cfg.smtp_security}") + msg = build_message(cfg, subject, body) + context = tls_context() + smtp: smtplib.SMTP + if cfg.smtp_security == "ssl": + smtp = smtplib.SMTP_SSL(cfg.smtp_host, cfg.smtp_port, timeout=timeout, context=context) + else: + smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=timeout) + try: + smtp.ehlo() + if cfg.smtp_security == "starttls": + smtp.starttls(context=context) + smtp.ehlo() + if cfg.smtp_user: + smtp.login(cfg.smtp_user, password or "") + smtp.send_message(msg) + finally: + try: + smtp.quit() + except smtplib.SMTPException: + smtp.close() diff --git a/src/netmover/notify.py b/src/netmover/notify.py index fc44602..e900a79 100644 --- a/src/netmover/notify.py +++ b/src/netmover/notify.py @@ -22,7 +22,9 @@ from string import Template import httpx from sqlalchemy.orm import Session, sessionmaker +from netmover.crypto import SecretBox from netmover.logging_setup import event, get_logger +from netmover.mail import mail_configured, smtp_send from netmover.settings_store import AppSettings, NotifyMode, load_settings log = get_logger(__name__) @@ -60,34 +62,71 @@ def http_sender(url: str, content_type: str, body: str) -> None: resp.raise_for_status() +Mailer = Callable[[AppSettings, "str | None", str, str], None] # (cfg, passwort, betreff, text) + + class Notifier: + """Verteilt Benachrichtigungen auf die eingerichteten Kanäle (Webhook, E-Mail).""" + def __init__( self, session_factory: sessionmaker[Session], *, sender: Sender = http_sender, + mailer: Mailer | None = None, + secrets: SecretBox | None = None, background: bool = True, ) -> None: self.session_factory = session_factory self._sender = sender + self._mailer: Mailer = mailer or smtp_send + self._secrets = secrets self._background = background def settings(self) -> AppSettings: return load_settings(self.session_factory) + def smtp_password(self, cfg: AppSettings) -> str | None: + if not cfg.smtp_password_enc or self._secrets is None: + return None + return self._secrets.decrypt(cfg.smtp_password_enc) + + def send_mail_now(self, subject: str, body: str, settings: AppSettings | None = None) -> None: + """Synchron senden (Test-Knopf); Fehler werden weitergereicht.""" + cfg = settings or self.settings() + self._mailer(cfg, self.smtp_password(cfg), subject, body) + def send(self, n: Notification, settings: AppSettings | None = None) -> None: cfg = settings or self.settings() - if cfg.notify_mode is NotifyMode.OFF or not cfg.webhook_url: + if cfg.notify_mode is NotifyMode.OFF: + return + jobs: list[tuple[str, Callable[[], None]]] = [] + if cfg.webhook_url: + body = render(cfg.webhook_template, cfg.webhook_content_type, n) + jobs.append( + ("webhook", lambda: self._sender(cfg.webhook_url, cfg.webhook_content_type, body)) + ) + if mail_configured(cfg): + password = self.smtp_password(cfg) + jobs.append(("email", lambda: self._mailer(cfg, password, n.title, n.message))) + if not jobs: return - body = render(cfg.webhook_template, cfg.webhook_content_type, n) def deliver() -> None: - try: - self._sender(cfg.webhook_url, cfg.webhook_content_type, body) - event(log, "notify.sent", kind=n.event, rule=n.rule) - except Exception as exc: # Benachrichtigung darf nie den Dienst stören - # URL nicht protokollieren – sie enthält oft ein Token. - event(log, "notify.failed", level=30, kind=n.event, error=type(exc).__name__) + for channel, job in jobs: + try: + job() + event(log, "notify.sent", kind=n.event, rule=n.rule, channel=channel) + except Exception as exc: # Benachrichtigung darf nie den Dienst stören + # Weder URL noch Zugangsdaten protokollieren. + event( + log, + "notify.failed", + level=30, + kind=n.event, + channel=channel, + error=type(exc).__name__, + ) if self._background: threading.Thread(target=deliver, name="nm-notify", daemon=True).start() diff --git a/tests/test_mail.py b/tests/test_mail.py new file mode 100644 index 0000000..0c8eecf --- /dev/null +++ b/tests/test_mail.py @@ -0,0 +1,304 @@ +"""E-Mail-Benachrichtigung gegen einen echten SMTP-Server im Testprozess (aiosmtpd).""" + +from __future__ import annotations + +import datetime +import socket +import ssl +from collections.abc import Iterator +from email import message_from_bytes +from email.policy import default +from pathlib import Path +from typing import Any + +import pytest +from aiosmtpd.controller import Controller +from aiosmtpd.smtp import AuthResult, LoginPassword +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session, sessionmaker + +from netmover.crypto import SecretBox +from netmover.mail import MailConfigError, parse_recipients, smtp_send +from netmover.notify import Notification, Notifier +from netmover.settings_store import AppSettings, NotifyMode, load_settings, save_settings +from tests.webutil import post + + +class Inbox: + def __init__(self) -> None: + self.messages: list[tuple[list[str], Any]] = [] + + async def handle_DATA(self, server: Any, session: Any, envelope: Any) -> str: + self.messages.append( + (envelope.rcpt_tos, message_from_bytes(envelope.content, policy=default)) + ) + return "250 OK" + + +def _free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port: int = s.getsockname()[1] + s.close() + return port + + +def _auth(server: Any, session: Any, envelope: Any, mechanism: str, data: Any) -> AuthResult: + ok = ( + isinstance(data, LoginPassword) + and data.login == b"netmover" + and data.password == b"Pa$$,wort" + ) + return AuthResult(success=ok, handled=False) + + +@pytest.fixture(scope="module") +def tls_files(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, Path]: + d = tmp_path_factory.mktemp("tls") + key = ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) + now = datetime.datetime.now(datetime.UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=1)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .sign(key, hashes.SHA256()) + ) + (d / "cert.pem").write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + (d / "key.pem").write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + return d / "cert.pem", d / "key.pem" + + +@pytest.fixture +def trust(tls_files: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch) -> None: + """Client vertraut unserem Test-Zertifikat (Prüfung bleibt eingeschaltet).""" + monkeypatch.setattr( + "netmover.mail.tls_context", lambda: ssl.create_default_context(cafile=str(tls_files[0])) + ) + + +def _server(mode: str, tls_files: tuple[Path, Path]) -> tuple[Controller, Inbox, int]: + inbox, port = Inbox(), _free_port() + ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ctx.load_cert_chain(*map(str, tls_files)) + kw: dict[str, Any] = { + "hostname": "localhost", + "port": port, + "authenticator": _auth, + "auth_exclude_mechanism": ["LOGIN"], + } + if mode == "starttls": + kw.update(tls_context=ctx, require_starttls=True) + elif mode == "ssl": + # implizites TLS erkennt aiosmtpd nicht als „verschlüsselt“ – AUTH trotzdem anbieten + kw.update(ssl_context=ctx, auth_require_tls=False) + else: + kw.update(auth_require_tls=False) + controller = Controller(inbox, **kw) + controller.start() + return controller, inbox, port + + +@pytest.fixture(params=["starttls", "ssl", "none"]) +def smtp_server( + request: pytest.FixtureRequest, tls_files: tuple[Path, Path], trust: None +) -> Iterator[tuple[str, Inbox, int]]: + controller, inbox, port = _server(request.param, tls_files) + yield request.param, inbox, port + controller.stop() + + +def _cfg(mode: str, port: int, **kw: Any) -> AppSettings: + data: dict[str, Any] = { + "email_enabled": True, + "smtp_host": "localhost", + "smtp_port": port, + "smtp_security": mode, + "smtp_user": "netmover", + "smtp_from": "netmover@example.org", + "smtp_to": "ich@example.org; admin@example.org", + "notify_mode": NotifyMode.EVERY, + } + data.update(kw) + return AppSettings(**data) + + +def test_smtp_send_all_security_modes(smtp_server: tuple[str, Inbox, int]) -> None: + mode, inbox, port = smtp_server + smtp_send(_cfg(mode, port), "Pa$$,wort", "NetMover: Anbindung „NAS“ gestört", "Host is down") + ((rcpts, msg),) = inbox.messages + assert rcpts == ["ich@example.org", "admin@example.org"] + assert msg["Subject"] == "NetMover: Anbindung „NAS“ gestört" + assert msg["From"] == "netmover@example.org" + body = msg.get_content() + assert "Host is down" in body and "NetMover auf" in body + + +def test_wrong_password_raises(smtp_server: tuple[str, Inbox, int]) -> None: + import smtplib + + mode, inbox, port = smtp_server + with pytest.raises(smtplib.SMTPAuthenticationError): + smtp_send(_cfg(mode, port), "falsch", "x", "y") + assert inbox.messages == [] + + +def test_certificate_is_verified(tls_files: tuple[Path, Path]) -> None: + """Ohne Vertrauen in das Zertifikat wird abgelehnt – keine stille Herabstufung.""" + controller, inbox, port = _server("starttls", tls_files) + try: + with pytest.raises(ssl.SSLCertVerificationError): + smtp_send(_cfg("starttls", port), "Pa$$,wort", "x", "y") + finally: + controller.stop() + assert inbox.messages == [] + + +def test_recipients() -> None: + assert parse_recipients("a@b.de, c@d.org;e@f.net\n") == ["a@b.de", "c@d.org", "e@f.net"] + with pytest.raises(MailConfigError, match="kein-mail"): + parse_recipients("a@b.de, kein-mail") + + +def test_notifier_sends_webhook_and_mail( + session_factory: sessionmaker[Session], secretbox: SecretBox +) -> None: + hooks: list[Any] = [] + mails: list[Any] = [] + cfg = _cfg( + "starttls", + 587, + smtp_password_enc=secretbox.encrypt("Pa$$,wort"), + webhook_url="http://hook.invalid/x", + ) + save_settings(session_factory, cfg) + n = Notifier( + session_factory, + sender=lambda *a: hooks.append(a), + mailer=lambda *a: mails.append(a), + secrets=secretbox, + background=False, + ) + n.send(Notification("mount.error", "NetMover: Anbindung „NAS“ gestört", "weg")) + assert len(hooks) == 1 and len(mails) == 1 + _cfg_used, password, subject, body = mails[0] + assert password == "Pa$$,wort" and subject.endswith("gestört") and body == "weg" + + +def test_mail_only_and_mode_off( + session_factory: sessionmaker[Session], secretbox: SecretBox +) -> None: + mails: list[Any] = [] + n = Notifier( + session_factory, mailer=lambda *a: mails.append(a), secrets=secretbox, background=False + ) + save_settings(session_factory, _cfg("starttls", 587, notify_mode=NotifyMode.OFF)) + n.send(Notification("x", "t", "m")) + save_settings(session_factory, _cfg("starttls", 587, email_enabled=False)) + n.send(Notification("x", "t", "m")) + assert mails == [] + save_settings(session_factory, _cfg("starttls", 587)) # nur E-Mail, kein Webhook + n.send(Notification("x", "t", "m")) + assert len(mails) == 1 + + +def test_mail_failure_is_swallowed( + session_factory: sessionmaker[Session], secretbox: SecretBox +) -> None: + def broken(*a: Any) -> None: + raise ConnectionRefusedError("weg") + + save_settings(session_factory, _cfg("starttls", 587)) + Notifier(session_factory, mailer=broken, secrets=secretbox, background=False).send( + Notification("x", "t", "m") + ) + + +# -- Einstellungsseite ------------------------------------------------------------------------ + + +def _form(**kw: str) -> dict[str, str]: + data = { + "retention_days": "30", + "notify_mode": "every", + "notify_after": "3", + "webhook_url": "", + "webhook_content_type": "application/json", + "webhook_template": "$message", + "email_enabled": "on", + "smtp_host": "smtp.example.org", + "smtp_port": "587", + "smtp_security": "starttls", + "smtp_user": "netmover", + "smtp_password": "Geheim,123!", + "smtp_from": "netmover@example.org", + "smtp_to": "ich@example.org", + } + data.update(kw) + return data + + +def test_settings_store_encrypted_password(client: TestClient, state: Any) -> None: + assert post(client, "/settings", _form()).status_code == 303 + cfg = load_settings(state.session_factory) + assert cfg.email_enabled and "Geheim" not in cfg.smtp_password_enc + assert state.secrets.decrypt(cfg.smtp_password_enc) == "Geheim,123!" + page = client.get("/settings").text + assert "Geheim,123!" not in page and 'placeholder="••••••"' in page + # leer lassen = unverändert + post(client, "/settings", _form(smtp_password="")) + assert ( + state.secrets.decrypt(load_settings(state.session_factory).smtp_password_enc) + == "Geheim,123!" + ) + + +def test_settings_validation(client: TestClient) -> None: + r = post( + client, + "/settings", + _form(smtp_to="ich@example.org, quatsch", smtp_port="99999", smtp_from="kein-absender"), + ) + assert r.status_code == 400 + for msg in ("Ungültige Adresse: quatsch", "Port zwischen", "Keine gültige E-Mail-Adresse"): + assert msg in r.text + + +def test_mail_test_button( + client: TestClient, state: Any, tls_files: tuple[Path, Path], trust: None +) -> None: + controller, inbox, port = _server("starttls", tls_files) + try: + post( + client, + "/settings", + _form(smtp_host="localhost", smtp_port=str(port), smtp_password="Pa$$,wort"), + ) + r = post(client, "/settings/mail-test") + assert "Gesendet an ich@example.org" in r.text + post( + client, + "/settings", + _form(smtp_host="localhost", smtp_port=str(port), smtp_password="falsch"), + ) + r = post(client, "/settings/mail-test") + assert "Fehlgeschlagen" in r.text and "falsch" not in r.text + finally: + controller.stop() + assert len(inbox.messages) == 1