From 779e160b95b03f4b3ad7e017e08447ddf2ead5aa Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Wed, 29 Apr 2026 13:03:12 +0200 Subject: [PATCH] add ip change notification and refine sender display --- app/api/routes/public_ip.py | 5 ++- app/config.py | 1 + app/main.py | 4 +- app/services/config_page.py | 2 + app/services/email.py | 49 ++++++++++++++++++-- app/services/public_ip.py | 60 +++++++++++++++++++++++-- tests/test_config.py | 2 + tests/test_public_ip.py | 90 +++++++++++++++++++++++++++++++++++-- tests/test_smtp.py | 71 +++++++++++++++++++++++++++-- 9 files changed, 266 insertions(+), 18 deletions(-) diff --git a/app/api/routes/public_ip.py b/app/api/routes/public_ip.py index ee9d722..766525f 100644 --- a/app/api/routes/public_ip.py +++ b/app/api/routes/public_ip.py @@ -3,8 +3,9 @@ from sqlalchemy.orm import Session from app.dependencies import get_auth_db, get_current_auth_session from app.schemas.public_ip import PublicIPCheckResponse +from app.config import get_settings from app.services.auth import AuthenticatedSession -from app.services.public_ip import check_public_ipv4 +from app.services.public_ip import check_public_ipv4_and_notify router = APIRouter(tags=["public-ip"]) @@ -17,7 +18,7 @@ def run_public_ip_check( if current_auth is None: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="authentication required") - result = check_public_ipv4(session) + result = check_public_ipv4_and_notify(session, bootstrap_settings=get_settings()) return PublicIPCheckResponse( status=result.status, checked_at=result.checked_at, diff --git a/app/config.py b/app/config.py index 929b38a..29adab5 100644 --- a/app/config.py +++ b/app/config.py @@ -28,6 +28,7 @@ class Settings(BaseSettings): smtp_port: int = 587 smtp_username: str = "" smtp_password: str = "" + smtp_from_name: str = "" smtp_from_address: str = "" smtp_to_address: str = "" smtp_use_starttls: bool = True diff --git a/app/main.py b/app/main.py index 4a8ee2a..dd8a9ec 100644 --- a/app/main.py +++ b/app/main.py @@ -19,7 +19,7 @@ from app.api.routes.ticktick import router as ticktick_router from app.config import get_settings from app.services.auth import AuthBootstrapError, initialize_auth_schema from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap -from app.services.public_ip import check_public_ipv4 +from app.services.public_ip import check_public_ipv4_and_notify from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db from scripts.location_db_adopt import LocationDatabaseAdoptionError, validate_location_runtime_db from scripts.poo_db_adopt import PooDatabaseAdoptionError, validate_poo_runtime_db @@ -29,7 +29,7 @@ def _run_scheduled_public_ip_check() -> None: session_local = auth_db.get_auth_session_local() session: Session = session_local() try: - check_public_ipv4(session) + check_public_ipv4_and_notify(session, bootstrap_settings=get_settings()) finally: session.close() diff --git a/app/services/config_page.py b/app/services/config_page.py index 141f75d..db7a450 100644 --- a/app/services/config_page.py +++ b/app/services/config_page.py @@ -32,6 +32,7 @@ CONFIG_FIELDS: tuple[ConfigField, ...] = ( ConfigField("SMTP", "SMTP_PORT", "smtp_port", "SMTP Port"), ConfigField("SMTP", "SMTP_USERNAME", "smtp_username", "SMTP Username"), ConfigField("SMTP", "SMTP_PASSWORD", "smtp_password", "SMTP Password", secret=True), + ConfigField("SMTP", "SMTP_FROM_NAME", "smtp_from_name", "SMTP From Name"), ConfigField("SMTP", "SMTP_FROM_ADDRESS", "smtp_from_address", "SMTP From Address"), ConfigField("SMTP", "SMTP_TO_ADDRESS", "smtp_to_address", "SMTP To Address"), ConfigField("SMTP", "SMTP_USE_STARTTLS", "smtp_use_starttls", "SMTP Use STARTTLS"), @@ -273,6 +274,7 @@ def _settings_payload(settings: Settings) -> dict[str, Any]: "smtp_port": settings.smtp_port, "smtp_username": settings.smtp_username, "smtp_password": settings.smtp_password, + "smtp_from_name": settings.smtp_from_name, "smtp_from_address": settings.smtp_from_address, "smtp_to_address": settings.smtp_to_address, "smtp_use_starttls": settings.smtp_use_starttls, diff --git a/app/services/email.py b/app/services/email.py index 249b921..e8ef37e 100644 --- a/app/services/email.py +++ b/app/services/email.py @@ -1,7 +1,9 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import UTC, datetime from email.message import EmailMessage +from email.utils import formataddr import smtplib from app.config import Settings @@ -21,6 +23,7 @@ class SMTPConfig: port: int username: str password: str + from_name: str from_address: str to_address: str use_starttls: bool @@ -47,6 +50,7 @@ def get_smtp_config(settings: Settings, *, require_enabled: bool = True) -> SMTP port=settings.smtp_port, username=settings.smtp_username, password=settings.smtp_password, + from_name=settings.smtp_from_name, from_address=settings.smtp_from_address, to_address=settings.smtp_to_address, use_starttls=settings.smtp_use_starttls, @@ -72,7 +76,7 @@ def send_plaintext_email( smtp_config = get_smtp_config(settings, require_enabled=require_enabled) message = EmailMessage() message["Subject"] = subject - message["From"] = smtp_config.from_address + message["From"] = _build_from_header(smtp_config) message["To"] = recipient or smtp_config.to_address message.set_content(body) @@ -84,7 +88,11 @@ def send_plaintext_email( smtp.ehlo() if smtp_config.username: smtp.login(smtp_config.username, smtp_config.password) - smtp.send_message(message) + smtp.send_message( + message, + from_addr=smtp_config.from_address, + to_addrs=[recipient or smtp_config.to_address], + ) except (OSError, smtplib.SMTPException) as exc: error_message = _sanitize_error_message(str(exc), smtp_config.password) raise EmailDeliveryError(error_message or "SMTP delivery failed") from exc @@ -99,8 +107,43 @@ def send_smtp_test_email(settings: Settings) -> None: ) +def send_public_ip_changed_email( + settings: Settings, + *, + previous_ipv4: str, + current_ipv4: str, + detected_at: datetime, +) -> None: + send_plaintext_email( + settings, + subject="Public IP changed", + body=( + "Your public IPv4 address has changed.\n\n" + f"Previous IP: {previous_ipv4}\n" + f"Current IP: {current_ipv4}\n" + f"Detected at: {_format_utc_timestamp(detected_at)}\n\n" + "If you use Namecheap API trusted IP restrictions, you may need to " + "update the trusted IP manually.\n" + ), + ) + + def _sanitize_error_message(message: str, password: str) -> str: sanitized = message if password: sanitized = sanitized.replace(password, "[redacted]") - return sanitized \ No newline at end of file + return sanitized + + +def _format_utc_timestamp(value: datetime) -> str: + if value.tzinfo is None: + normalized = value.replace(tzinfo=UTC) + else: + normalized = value.astimezone(UTC) + return normalized.strftime("%Y-%m-%d %H:%M:%S UTC") + + +def _build_from_header(smtp_config: SMTPConfig) -> str: + if smtp_config.from_name: + return formataddr((smtp_config.from_name, smtp_config.from_address)) + return smtp_config.from_address \ No newline at end of file diff --git a/app/services/public_ip.py b/app/services/public_ip.py index a984f25..d13d6cc 100644 --- a/app/services/public_ip.py +++ b/app/services/public_ip.py @@ -10,7 +10,10 @@ import httpx from sqlalchemy import select from sqlalchemy.orm import Session +from app.config import Settings from app.models.public_ip import PublicIPHistory, PublicIPState +from app.services.config_page import build_runtime_settings +from app.services.email import EmailConfigurationError, EmailDeliveryError, send_public_ip_changed_email logger = logging.getLogger(__name__) @@ -31,6 +34,8 @@ class PublicIPCheckResult: status: PublicIPResultStatus checked_at: datetime changed: bool + previous_ipv4: str | None = None + current_ipv4: str | None = None def check_public_ipv4( @@ -77,7 +82,12 @@ def check_public_ipv4( ) ) session.commit() - return PublicIPCheckResult(status="first_seen", checked_at=checked_at, changed=False) + return PublicIPCheckResult( + status="first_seen", + checked_at=checked_at, + changed=False, + current_ipv4=current_ipv4, + ) if state.current_ipv4 == current_ipv4: state.last_checked_at = checked_at @@ -85,9 +95,15 @@ def check_public_ipv4( state.last_check_error = None state.last_provider = provider_name session.commit() - return PublicIPCheckResult(status="unchanged", checked_at=checked_at, changed=False) + return PublicIPCheckResult( + status="unchanged", + checked_at=checked_at, + changed=False, + current_ipv4=current_ipv4, + ) - state.previous_ipv4 = state.current_ipv4 + previous_ipv4 = state.current_ipv4 + state.previous_ipv4 = previous_ipv4 state.current_ipv4 = current_ipv4 state.last_checked_at = checked_at state.last_changed_at = checked_at @@ -103,7 +119,43 @@ def check_public_ipv4( ) ) session.commit() - return PublicIPCheckResult(status="changed", checked_at=checked_at, changed=True) + return PublicIPCheckResult( + status="changed", + checked_at=checked_at, + changed=True, + previous_ipv4=previous_ipv4, + current_ipv4=current_ipv4, + ) + + +def check_public_ipv4_and_notify( + session: Session, + *, + bootstrap_settings: Settings, + fetch_public_ipv4: PublicIPv4Fetcher | None = None, + provider_name: str = PUBLIC_IP_PROVIDER_NAME, +) -> PublicIPCheckResult: + result = check_public_ipv4( + session, + fetch_public_ipv4=fetch_public_ipv4, + provider_name=provider_name, + ) + + if result.status != "changed" or result.previous_ipv4 is None or result.current_ipv4 is None: + return result + + runtime_settings = build_runtime_settings(session, bootstrap_settings) + try: + send_public_ip_changed_email( + runtime_settings, + previous_ipv4=result.previous_ipv4, + current_ipv4=result.current_ipv4, + detected_at=result.checked_at, + ) + except (EmailConfigurationError, EmailDeliveryError) as exc: + logger.warning("Public IPv4 change notification failed: %s", exc) + + return result def fetch_public_ipv4_from_provider() -> str: diff --git a/tests/test_config.py b/tests/test_config.py index 1818de3..6aea8d0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -61,6 +61,7 @@ def test_settings_support_smtp_fields(monkeypatch) -> None: monkeypatch.setenv("SMTP_PORT", "2525") monkeypatch.setenv("SMTP_USERNAME", "smtp-user") monkeypatch.setenv("SMTP_PASSWORD", "smtp-password") + monkeypatch.setenv("SMTP_FROM_NAME", "Home Automation") monkeypatch.setenv("SMTP_FROM_ADDRESS", "sender@example.com") monkeypatch.setenv("SMTP_TO_ADDRESS", "recipient@example.com") monkeypatch.setenv("SMTP_USE_STARTTLS", "false") @@ -72,6 +73,7 @@ def test_settings_support_smtp_fields(monkeypatch) -> None: assert settings.smtp_port == 2525 assert settings.smtp_username == "smtp-user" assert settings.smtp_password == "smtp-password" + assert settings.smtp_from_name == "Home Automation" assert settings.smtp_from_address == "sender@example.com" assert settings.smtp_to_address == "recipient@example.com" assert settings.smtp_use_starttls is False diff --git a/tests/test_public_ip.py b/tests/test_public_ip.py index 23b39e1..a8a9558 100644 --- a/tests/test_public_ip.py +++ b/tests/test_public_ip.py @@ -6,7 +6,9 @@ from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker -from app.services.public_ip import PublicIPCheckResult, check_public_ipv4 +from app.config import Settings +from app.services.email import EmailDeliveryError +from app.services.public_ip import PublicIPCheckResult, check_public_ipv4, check_public_ipv4_and_notify def _make_session(database_url: str) -> Session: @@ -152,8 +154,8 @@ def test_public_ip_check_endpoint_hides_ip_values(client: TestClient, monkeypatc monkeypatch.setattr( public_ip_route, - "check_public_ipv4", - lambda session: PublicIPCheckResult( + "check_public_ipv4_and_notify", + lambda session, bootstrap_settings: PublicIPCheckResult( status="changed", checked_at=fixed_checked_at, changed=True, @@ -172,3 +174,85 @@ def test_public_ip_check_endpoint_hides_ip_values(client: TestClient, monkeypatc assert "current_ipv4" not in response.text assert "previous_ipv4" not in response.text assert "203.0.113.10" not in response.text + + +def _notification_settings() -> Settings: + return Settings( + _env_file=None, + app_env="development", + app_hostname="localhost:8000", + app_database_url="sqlite:///./data/app.db", + location_database_url="sqlite:///./data/locationRecorder.db", + poo_database_url="sqlite:///./data/pooRecorder.db", + auth_bootstrap_username="admin", + auth_bootstrap_password="secret-password", + smtp_enabled=True, + smtp_host="smtp.example.com", + smtp_port=587, + smtp_username="smtp-user", + smtp_password="super-secret-password", + smtp_from_address="sender@example.com", + smtp_to_address="recipient@example.com", + smtp_use_starttls=True, + ) + + +def test_public_ip_notification_sends_only_when_changed(auth_database, monkeypatch) -> None: + session = _make_session(auth_database["app_url"]) + sent = [] + monkeypatch.setattr( + "app.services.public_ip.send_public_ip_changed_email", + lambda settings, *, previous_ipv4, current_ipv4, detected_at: sent.append( + (previous_ipv4, current_ipv4, detected_at) + ), + ) + try: + first_seen = check_public_ipv4_and_notify( + session, + bootstrap_settings=_notification_settings(), + fetch_public_ipv4=lambda: "203.0.113.10", + ) + unchanged = check_public_ipv4_and_notify( + session, + bootstrap_settings=_notification_settings(), + fetch_public_ipv4=lambda: "203.0.113.10", + ) + changed = check_public_ipv4_and_notify( + session, + bootstrap_settings=_notification_settings(), + fetch_public_ipv4=lambda: "198.51.100.25", + ) + finally: + session.close() + + assert first_seen.status == "first_seen" + assert unchanged.status == "unchanged" + assert changed.status == "changed" + assert len(sent) == 1 + assert sent[0][0] == "203.0.113.10" + assert sent[0][1] == "198.51.100.25" + assert sent[0][2] == changed.checked_at + + +def test_public_ip_notification_failure_does_not_break_changed_result(auth_database, monkeypatch) -> None: + session = _make_session(auth_database["app_url"]) + monkeypatch.setattr( + "app.services.public_ip.send_public_ip_changed_email", + lambda settings, *, previous_ipv4, current_ipv4, detected_at: (_ for _ in ()).throw( + EmailDeliveryError("smtp down") + ), + ) + try: + check_public_ipv4(session, fetch_public_ipv4=lambda: "203.0.113.10") + result = check_public_ipv4_and_notify( + session, + bootstrap_settings=_notification_settings(), + fetch_public_ipv4=lambda: "198.51.100.25", + ) + finally: + session.close() + + assert result.status == "changed" + assert result.changed is True + assert result.previous_ipv4 == "203.0.113.10" + assert result.current_ipv4 == "198.51.100.25" diff --git a/tests/test_smtp.py b/tests/test_smtp.py index f004b2e..2881da6 100644 --- a/tests/test_smtp.py +++ b/tests/test_smtp.py @@ -5,7 +5,13 @@ import smtplib from fastapi.testclient import TestClient from app.config import Settings -from app.services.email import EmailDeliveryError, get_smtp_config, is_smtp_ready, send_smtp_test_email +from app.services.email import ( + EmailDeliveryError, + get_smtp_config, + is_smtp_ready, + send_public_ip_changed_email, + send_smtp_test_email, +) def _extract_csrf_token(html: str) -> str: @@ -43,6 +49,7 @@ def _smtp_settings(**overrides) -> Settings: "smtp_port": 587, "smtp_username": "smtp-user", "smtp_password": "super-secret-password", + "smtp_from_name": "Home Automation", "smtp_from_address": "sender@example.com", "smtp_to_address": "recipient@example.com", "smtp_use_starttls": True, @@ -60,6 +67,7 @@ def test_get_smtp_config_reads_runtime_values() -> None: assert smtp_config.port == 2525 assert smtp_config.username == "smtp-user" assert smtp_config.password == "super-secret-password" + assert smtp_config.from_name == "Home Automation" assert smtp_config.from_address == "sender@example.com" assert smtp_config.to_address == "recipient@example.com" assert smtp_config.use_starttls is False @@ -96,11 +104,13 @@ def test_send_smtp_test_email_success(monkeypatch) -> None: sent["username"] = username sent["password"] = password - def send_message(self, message): + def send_message(self, message, from_addr=None, to_addrs=None): sent["subject"] = message["Subject"] sent["from"] = message["From"] sent["to"] = message["To"] sent["body"] = message.get_content() + sent["envelope_from"] = from_addr + sent["envelope_to"] = to_addrs monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) @@ -113,8 +123,10 @@ def test_send_smtp_test_email_success(monkeypatch) -> None: assert sent["username"] == "smtp-user" assert sent["password"] == "super-secret-password" assert sent["subject"] == "Home Automation SMTP Test" - assert sent["from"] == "sender@example.com" + assert sent["from"] == "Home Automation " assert sent["to"] == "recipient@example.com" + assert sent["envelope_from"] == "sender@example.com" + assert sent["envelope_to"] == ["recipient@example.com"] assert "This is a test email" in sent["body"] @@ -140,8 +152,10 @@ def test_send_smtp_test_email_does_not_require_smtp_enabled(monkeypatch) -> None def login(self, username, password): return None - def send_message(self, message): + def send_message(self, message, from_addr=None, to_addrs=None): sent["subject"] = message["Subject"] + sent["from"] = message["From"] + sent["envelope_from"] = from_addr monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) @@ -149,6 +163,8 @@ def test_send_smtp_test_email_does_not_require_smtp_enabled(monkeypatch) -> None assert sent["host"] == "smtp.example.com" assert sent["subject"] == "Home Automation SMTP Test" + assert sent["from"] == "Home Automation " + assert sent["envelope_from"] == "sender@example.com" def test_send_smtp_test_email_failure_sanitizes_password(monkeypatch) -> None: @@ -178,6 +194,53 @@ def test_send_smtp_test_email_failure_sanitizes_password(monkeypatch) -> None: assert "[redacted]" in str(exc) +def test_send_public_ip_changed_email_contains_expected_english_content(monkeypatch) -> None: + sent = {} + + class FakeSMTP: + def __init__(self, host, port, timeout): + sent["host"] = host + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return None + + def ehlo(self): + return None + + def starttls(self): + return None + + def login(self, username, password): + return None + + def send_message(self, message, from_addr=None, to_addrs=None): + sent["subject"] = message["Subject"] + sent["body"] = message.get_content() + sent["from"] = message["From"] + sent["envelope_from"] = from_addr + + monkeypatch.setattr("app.services.email.smtplib.SMTP", FakeSMTP) + + send_public_ip_changed_email( + _smtp_settings(), + previous_ipv4="203.0.113.10", + current_ipv4="198.51.100.25", + detected_at=__import__("datetime").datetime(2026, 4, 29, 10, 0, tzinfo=__import__("datetime").UTC), + ) + + assert sent["subject"] == "Public IP changed" + assert sent["from"] == "Home Automation " + assert sent["envelope_from"] == "sender@example.com" + assert "Your public IPv4 address has changed." in sent["body"] + assert "Previous IP: 203.0.113.10" in sent["body"] + assert "Current IP: 198.51.100.25" in sent["body"] + assert "Detected at: 2026-04-29 10:00:00 UTC" in sent["body"] + assert "update the trusted IP manually" in sent["body"] + + def test_config_update_does_not_clear_existing_smtp_password( client: TestClient, test_database_urls ) -> None: