feat: add secure invitation and reset links

tests/test_auth_registration.py

@@ -5,6 +5,7 @@ import unittest
 from datetime import timedelta
 from tempfile import TemporaryDirectory
 from unittest.mock import patch
+from urllib.parse import urlparse
 
 from flask import session
 
@@ -12,7 +13,7 @@ from app import create_app
 from app.config import Config
 from app.email import EmailDeliveryError
 from app.extensions import db
-from app.models import EmailVerificationCode, TrustedDevice, User
+from app.models import EmailVerificationCode, PasswordResetToken, RegistrationInvitation, TrustedDevice, User
 from app.routes import grant_fresh_authorization, has_fresh_authorization, valid_password
 from app.time_utils import utc_now
 
@@ -40,6 +41,8 @@ class EmailAuthenticationTest(unittest.TestCase):
             self.assertEqual(html.count("data-code-digit\n"), 6)
             self.assertIn("60 秒后可重新发送", html)
             self.assertIn('id="resendButton"', html)
+            self.assertIn("al****ce@example.com", html)
+            self.assertNotIn("alice@example.com", html)
 
     def test_fresh_authorization_is_bound_to_user_and_expires(self):
         with TemporaryDirectory() as directory:
@@ -296,8 +299,7 @@ class EmailAuthenticationTest(unittest.TestCase):
             with app.app_context(): self.assertEqual(TrustedDevice.query.count(), 1)
 
     @patch("app.routes.send_transactional_email")
-    @patch("app.routes.secrets.randbelow", return_value=123456)
-    def test_password_reset_revokes_trusted_devices(self, _random, _send):
+    def test_password_reset_link_revokes_trusted_devices(self, send):
         with TemporaryDirectory() as directory:
             app = self.create_app(directory)
             with app.app_context():
@@ -305,16 +307,49 @@ class EmailAuthenticationTest(unittest.TestCase):
             client = app.test_client()
             response = self.form(client, "/forgot-password", email="alice@example.com")
             self.assertEqual(response.status_code, 302)
-            self.assertEqual(response.location, "/verify/reset")
-            verify = client.get("/verify/reset")
-            response = client.post("/verify/reset", data={"csrf_token": self.csrf(verify), "code": "123456"})
-            self.assertEqual(response.location, "/set-password")
-            page = client.get("/set-password")
-            response = client.post("/set-password", data={"csrf_token": self.csrf(page), "password": "New-password-1234!", "password_confirm": "New-password-1234!"})
+            self.assertEqual(response.location, "/forgot-password")
+            self.assertEqual(send.call_count, 1)
+            html = send.call_args.kwargs["html"]
+            reset_path = urlparse(re.search(r'href="([^"]+)"', html).group(1)).path
+            page = client.get(reset_path)
+            response = client.post(reset_path, data={"csrf_token": self.csrf(page), "password": "New-password-1234!", "password_confirm": "New-password-1234!"})
             self.assertEqual(response.status_code, 302)
             with app.app_context():
                 user = User.query.filter_by(email="alice@example.com").one()
-                self.assertTrue(user.check_password("New-password-1234!")); self.assertEqual(TrustedDevice.query.count(), 0); self.assertEqual(user.auth_version, 2)
+                self.assertTrue(user.check_password("New-password-1234!")); self.assertEqual(TrustedDevice.query.count(), 0); self.assertEqual(user.auth_version, 2); self.assertIsNotNone(PasswordResetToken.query.one().used_at)
+
+    @patch("app.routes.send_transactional_email")
+    def test_admin_can_invite_user_while_self_registration_is_disabled(self, send):
+        with TemporaryDirectory() as directory:
+            app = self.create_app(directory)
+            with app.app_context():
+                admin = User(username="Admin", email="admin@example.com", is_admin=True, is_active_account=True)
+                admin.set_password("Password-1234!")
+                db.session.add(admin)
+                db.session.commit()
+                admin_id = admin.id
+            client = app.test_client()
+            with app.app_context():
+                self.login_as(client, db.session.get(User, admin_id))
+            admin_page = client.get("/admin")
+            response = client.post(
+                "/admin/invitations",
+                data={"csrf_token": self.csrf(admin_page), "email": "invitee@example.com"},
+            )
+            self.assertEqual(response.status_code, 302)
+            self.assertEqual(send.call_count, 1)
+            with app.app_context():
+                self.assertEqual(RegistrationInvitation.query.count(), 1)
+            invite_html = send.call_args.kwargs["html"]
+            invite_path = urlparse(re.search(r'href="([^"]+)"', invite_html).group(1)).path
+            page = client.get(invite_path)
+            response = client.post(invite_path, data={"csrf_token": self.csrf(page), "username": "Invited User", "password": "Invited-password-1234!", "password_confirm": "Invited-password-1234!"})
+            self.assertEqual(response.status_code, 302)
+            with app.app_context():
+                invited = User.query.filter_by(email="invitee@example.com").one()
+                self.assertTrue(invited.is_active_account)
+                self.assertIsNotNone(invited.email_verified_at)
+                self.assertIsNotNone(RegistrationInvitation.query.one().used_at)
 
     @patch("app.routes.send_transactional_email")
     @patch("app.routes.secrets.randbelow", return_value=123456)