Resend TOTP login confirmation emails that have expired

tests/unit/accounts/test_services.py

@@ -2081,6 +2081,41 @@ def test_device_is_pending_not_expired(self, user_service, monkeypatch):
         assert not user_service.device_is_known(user.id, user_service.request)
         assert send_email.calls == []
 
+    def test_device_is_pending_and_expired(self, user_service, monkeypatch):
+        user = UserFactory.create(with_verified_primary_email=True)
+        UserUniqueLoginFactory.create(
+            user=user,
+            ip_address=REMOTE_ADDR,
+            status="pending",
+            created=datetime.datetime(1970, 1, 1),
+            expires=datetime.datetime(1970, 1, 1),
+        )
+        send_email = pretend.call_recorder(lambda *a, **kw: None)
+        monkeypatch.setattr(services, "send_unrecognized_login_email", send_email)
+        token_service = pretend.stub(dumps=lambda d: "fake_token", max_age=60)
+        user_service.request = pretend.stub(
+            db=user_service.db,
+            remote_addr=REMOTE_ADDR,
+            headers={
+                "User-Agent": (
+                    "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) "
+                    "Gecko/20100101 Firefox/15.0.1"
+                )
+            },
+            find_service=lambda *a, **kw: token_service,
+        )
+
+        assert not user_service.device_is_known(user.id, user_service.request)
+        assert send_email.calls == [
+            pretend.call(
+                user_service.request,
+                user,
+                ip_address=REMOTE_ADDR,
+                user_agent="Firefox (Ubuntu)",
+                token="fake_token",
+            )
+        ]
+
     @pytest.mark.parametrize("ua_string", [None, "no bueno", "Python-urllib/3.7"])
     def test_device_is_not_known_bad_user_agent(
         self, user_service, monkeypatch, ua_string

warehouse/accounts/models.py

@@ -519,6 +519,7 @@ class UserUniqueLogin(db.Model):
         default=UniqueLoginStatus.PENDING,
         server_default=UniqueLoginStatus.PENDING.value,
     )
+    expires: Mapped[datetime.datetime | None] = mapped_column(TZDateTime)
 
     def __repr__(self):
         return (

warehouse/accounts/services.py

@@ -771,6 +771,17 @@ def device_is_known(self, userid, request):
             request.db.flush()  # To get the ID for the token
             should_send_email = True
 
+        # Check if the login had expired
+        if unique_login.expires and unique_login.expires < datetime.datetime.now(
+            datetime.UTC
+        ):
+            # The previous token has expired, update the expiry for
+            # the login and re-send the email
+            unique_login.expires = datetime.datetime.now(
+                datetime.UTC
+            ) + datetime.timedelta(seconds=token_service.max_age)
+            should_send_email = True
+
         # If we don't need to send an email, short-circuit
         if not should_send_email:
             return False

warehouse/migrations/versions/537b63a29cea_add_expires_column_to_useruniquelogin.py

@@ -0,0 +1,39 @@
+# SPDX-License-Identifier: Apache-2.0
+"""
+Add expires column to UserUniqueLogin
+
+Revision ID: 537b63a29cea
+Revises: 7cf64da2632a
+Create Date: 2025-11-18 14:38:31.355587
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+from warehouse.utils.db.types import TZDateTime
+
+revision = "537b63a29cea"
+down_revision = "7cf64da2632a"
+
+
+def upgrade():
+    op.add_column(
+        "user_unique_logins",
+        sa.Column("expires", TZDateTime(), nullable=True),
+    )
+
+    op.execute(
+        """
+        UPDATE user_unique_logins
+        SET expires =
+            CASE
+                WHEN status = 'confirmed' THEN NULL
+                ELSE created + INTERVAL '6 hours'
+            END
+        """
+    )
+
+
+def downgrade():
+    op.drop_column("user_unique_logins", "expires")

warehouse/migrations/versions/7cf64da2632a_add_reverse_id_index_for_journals.py

@@ -21,6 +21,9 @@ def upgrade():
     op.get_bind().commit()
 
     with op.get_context().autocommit_block():
+        op.execute(sa.text("SET statement_timeout = 200000"))
+        op.execute(sa.text("SET lock_timeout = 200000"))
+
         op.create_index(
             "journals_name_id_idx",
             "journals",