Add secure password reset email delivery

tests/test_email.py

@@ -0,0 +1,173 @@
+"""Tests for transactional email and password-reset delivery safety."""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+from unittest.mock import patch
+
+import requests
+from flask import Flask
+
+from cmdforge.registry.app import create_app, password_hasher
+from cmdforge.registry.db import connect_db
+from cmdforge.web.email import send_email
+
+
+def _email_app(**config) -> Flask:
+    app = Flask(__name__)
+    defaults = {
+        "CMDFORGE_ENV": "development",
+        "MAIL_TRANSPORT": "disabled",
+        "MAIL_FROM": "CmdForge <[email protected]>",
+        "MAIL_TIMEOUT": 10,
+        "RESEND_API_KEY": "",
+    }
+    defaults.update(config)
+    app.config.update(defaults)
+    return app
+
+
+def test_disabled_delivery_does_not_log_message_or_recipient(caplog):
+    app = _email_app()
+    with app.app_context(), caplog.at_level(logging.INFO):
+        assert not send_email(
+            "[email protected]", "Reset subject", "secret reset body", "secret text"
+        )
+    assert "[email protected]" not in caplog.text
+    assert "secret reset body" not in caplog.text
+    assert "secret text" not in caplog.text
+
+
+def test_console_transport_is_rejected_in_production(caplog):
+    app = _email_app(CMDFORGE_ENV="production", MAIL_TRANSPORT="console")
+    with app.app_context(), caplog.at_level(logging.INFO):
+        assert not send_email("[email protected]", "subject", "body")
+    assert "[email protected]" not in caplog.text
+    assert "body" not in caplog.text
+
+
+def test_resend_transport_sends_expected_request(monkeypatch):
+    app = _email_app(MAIL_TRANSPORT="resend", RESEND_API_KEY="test-api-key")
+    captured = {}
+
+    class Response:
+        def raise_for_status(self):
+            return None
+
+    def fake_post(url, **kwargs):
+        captured["url"] = url
+        captured.update(kwargs)
+        return Response()
+
+    monkeypatch.setattr(requests, "post", fake_post)
+    with app.app_context():
+        assert send_email("[email protected]", "subject", "<p>body</p>", "body")
+
+    assert captured["url"] == "https://api.resend.com/emails"
+    assert captured["headers"]["Authorization"] == "Bearer test-api-key"
+    assert captured["json"]["to"] == ["[email protected]"]
+    assert captured["timeout"] == 10.0
+
+
+def test_resend_failure_does_not_log_sensitive_data(monkeypatch, caplog):
+    app = _email_app(MAIL_TRANSPORT="resend", RESEND_API_KEY="test-api-key")
+
+    def fail_post(*args, **kwargs):
+        raise requests.ConnectionError("provider unavailable")
+
+    monkeypatch.setattr(requests, "post", fail_post)
+    with app.app_context(), caplog.at_level(logging.INFO):
+        assert not send_email("[email protected]", "subject", "reset-secret")
+    assert "[email protected]" not in caplog.text
+    assert "reset-secret" not in caplog.text
+    assert "test-api-key" not in caplog.text
+
+
+def _registry_client(tmp_path, monkeypatch):
+    db_path = tmp_path / "registry.db"
+    monkeypatch.setenv("CMDFORGE_REGISTRY_DB", str(db_path))
+    monkeypatch.setenv("CMDFORGE_ENV", "production")
+    monkeypatch.setenv("CMDFORGE_PUBLIC_URL", "https://cmdforge.example")
+    app = create_app()
+    app.config["TESTING"] = True
+    with connect_db() as db:
+        db.execute(
+            """
+            INSERT INTO publishers (email, password_hash, slug, display_name)
+            VALUES (?, ?, ?, ?)
+            """,
+            ["[email protected]", password_hasher.hash("initial-password"), "owner", "Owner"],
+        )
+        db.commit()
+    return app.test_client(), db_path
+
+
+def test_reset_uses_canonical_url_and_leaves_delivered_token_valid(tmp_path, monkeypatch):
+    client, db_path = _registry_client(tmp_path, monkeypatch)
+    captured = {}
+
+    def delivered(to, token, base_url):
+        captured.update(to=to, token=token, base_url=base_url)
+        return True
+
+    with patch("cmdforge.web.email.send_password_reset_email", delivered):
+        response = client.post(
+            "/api/v1/password-reset/request",
+            json={"email": "[email protected]"},
+            headers={"Host": "attacker.example"},
+        )
+
+    assert response.status_code == 200
+    assert captured["base_url"] == "https://cmdforge.example"
+    with connect_db() as db:
+        row = db.execute(
+            "SELECT token_hash, used_at FROM password_reset_tokens"
+        ).fetchone()
+    assert row["token_hash"] == hashlib.sha256(captured["token"].encode()).hexdigest()
+    assert row["used_at"] is None
+
+
+def test_reset_invalidates_token_when_delivery_fails(tmp_path, monkeypatch):
+    client, db_path = _registry_client(tmp_path, monkeypatch)
+
+    with patch("cmdforge.web.email.send_password_reset_email", return_value=False):
+        response = client.post(
+            "/api/v1/password-reset/request", json={"email": "[email protected]"}
+        )
+
+    assert response.status_code == 200
+    with connect_db() as db:
+        row = db.execute("SELECT used_at FROM password_reset_tokens").fetchone()
+    assert row["used_at"] is not None
+
+
+def test_reset_invalidates_token_without_safe_production_base_url(tmp_path, monkeypatch):
+    client, db_path = _registry_client(tmp_path, monkeypatch)
+    client.application.config["PUBLIC_BASE_URL"] = "http://insecure.example"
+
+    with patch("cmdforge.web.email.send_password_reset_email") as sender:
+        response = client.post(
+            "/api/v1/password-reset/request", json={"email": "[email protected]"}
+        )
+
+    assert response.status_code == 200
+    sender.assert_not_called()
+    with connect_db() as db:
+        row = db.execute("SELECT used_at FROM password_reset_tokens").fetchone()
+    assert row["used_at"] is not None
+
+
+def test_unknown_email_creates_no_token_and_sends_nothing(tmp_path, monkeypatch):
+    client, db_path = _registry_client(tmp_path, monkeypatch)
+
+    with patch("cmdforge.web.email.send_password_reset_email") as sender:
+        response = client.post(
+            "/api/v1/password-reset/request", json={"email": "[email protected]"}
+        )
+
+    assert response.status_code == 200
+    sender.assert_not_called()
+    with connect_db() as db:
+        count = db.execute("SELECT COUNT(*) FROM password_reset_tokens").fetchone()[0]
+    assert count == 0