chore: enforce 2fa for legacy uploaders

tests/common/db/accounts.py

@@ -23,6 +23,17 @@ class UserFactory(WarehouseFactory):
     class Meta:
         model = User
 
+    class Params:
+        # Shortcut to create a user with a verified primary email
+        with_verified_primary_email = factory.Trait(
+            email=factory.RelatedFactory(
+                "tests.common.db.accounts.EmailFactory",
+                factory_related_name="user",
+                primary=True,
+                verified=True,
+            )
+        )
+
     username = factory.Faker("pystr", max_chars=12)
     name = factory.Faker("word")
     password = "!"

tests/unit/email/test_init.py

@@ -202,7 +202,7 @@ def test_doesnt_send_with_unverified(self, primary_email, address):
         assert request.task.calls == []
         assert task.delay.calls == []
 
-    def test_doesnt_send_within_reset_window(self, pyramid_request, pyramid_services):
+    def test_doesnt_send_within_repeat_window(self, pyramid_request, pyramid_services):
         email_service = pretend.stub(
             last_sent=pretend.call_recorder(
                 lambda to, subject: datetime.datetime.now()
@@ -226,6 +226,49 @@ def test_doesnt_send_within_reset_window(self, pyramid_request, pyramid_services
         assert pyramid_request.task.calls == []
         assert task.delay.calls == []
 
+    def test_sends_when_outside_repeat_window(self, db_request, pyramid_services):
+        email_service = pretend.stub(
+            last_sent=pretend.call_recorder(
+                lambda to, subject: datetime.datetime.now()
+                - datetime.timedelta(seconds=69)
+            )
+        )
+        pyramid_services.register_service(email_service, IEmailSender, None, name="")
+
+        task = pretend.stub(delay=pretend.call_recorder(lambda *a, **kw: None))
+        db_request.task = pretend.call_recorder(lambda x: task)
+
+        user = UserFactory.create(with_verified_primary_email=True)
+
+        msg = EmailMessage(subject="My Subject", body_text="My Body")
+
+        email._send_email_to_user(
+            db_request, user, msg, repeat_window=datetime.timedelta(seconds=42)
+        )
+
+        assert db_request.task.calls == [pretend.call(email.send_email)]
+        assert task.delay.calls == [
+            pretend.call(
+                f"{user.name} <{user.primary_email.email}>",
+                {
+                    "sender": None,
+                    "subject": "My Subject",
+                    "body_text": "My Body",
+                    "body_html": None,
+                },
+                {
+                    "tag": "account:email:sent",
+                    "user_id": user.id,
+                    "additional": {
+                        "from_": None,
+                        "to": user.email,
+                        "subject": "My Subject",
+                        "redact_ip": False,
+                    },
+                },
+            )
+        ]
+
     @pytest.mark.parametrize(
         ("username", "primary_email", "address", "expected"),
         [
@@ -1557,80 +1600,6 @@ def test_password_reset_by_admin_email(
         ]
 
 
-class Test2FAonUploadEmail:
-    def test_send_two_factor_not_yet_enabled_email(
-        self, pyramid_request, pyramid_config, monkeypatch
-    ):
-        stub_user = pretend.stub(
-            id="id",
-            username="username",
-            name="",
-            email="email@example.com",
-            primary_email=pretend.stub(email="email@example.com", verified=True),
-            has_2fa=False,
-        )
-        subject_renderer = pyramid_config.testing_add_renderer(
-            "email/two-factor-not-yet-enabled/subject.txt"
-        )
-        subject_renderer.string_response = "Email Subject"
-        body_renderer = pyramid_config.testing_add_renderer(
-            "email/two-factor-not-yet-enabled/body.txt"
-        )
-        body_renderer.string_response = "Email Body"
-        html_renderer = pyramid_config.testing_add_renderer(
-            "email/two-factor-not-yet-enabled/body.html"
-        )
-        html_renderer.string_response = "Email HTML Body"
-
-        send_email = pretend.stub(
-            delay=pretend.call_recorder(lambda *args, **kwargs: None)
-        )
-        pyramid_request.task = pretend.call_recorder(lambda *args, **kwargs: send_email)
-        monkeypatch.setattr(email, "send_email", send_email)
-
-        pyramid_request.db = pretend.stub(
-            query=lambda a: pretend.stub(
-                filter=lambda *a: pretend.stub(
-                    one=lambda: pretend.stub(user_id=stub_user.id)
-                )
-            ),
-        )
-        pyramid_request.user = stub_user
-        pyramid_request.registry.settings = {"mail.sender": "noreply@example.com"}
-
-        result = email.send_two_factor_not_yet_enabled_email(
-            pyramid_request,
-            stub_user,
-        )
-
-        assert result == {"username": stub_user.username}
-        assert pyramid_request.task.calls == [pretend.call(send_email)]
-        assert send_email.delay.calls == [
-            pretend.call(
-                f"{stub_user.username} <{stub_user.email}>",
-                {
-                    "sender": None,
-                    "subject": "Email Subject",
-                    "body_text": "Email Body",
-                    "body_html": (
-                        "<html>\n<head></head>\n"
-                        "<body><p>Email HTML Body</p></body>\n</html>\n"
-                    ),
-                },
-                {
-                    "tag": "account:email:sent",
-                    "user_id": stub_user.id,
-                    "additional": {
-                        "from_": "noreply@example.com",
-                        "to": stub_user.email,
-                        "subject": "Email Subject",
-                        "redact_ip": False,
-                    },
-                },
-            )
-        ]
-
-
 class TestAccountDeletionEmail:
     def test_account_deletion_email(
         self, pyramid_request, pyramid_config, metrics, monkeypatch

tests/unit/forklift/test_legacy.py

@@ -432,8 +432,9 @@ def test_fails_disallow_new_upload(self, pyramid_config, pyramid_request):
     def test_fails_invalid_version(self, pyramid_config, pyramid_request, version):
         pyramid_request.POST["protocol_version"] = version
         pyramid_request.flags = pretend.stub(enabled=lambda *a: False)
+        pyramid_request.help_url = pretend.call_recorder(lambda **kw: "/the/help/url/")
 
-        user = pretend.stub(primary_email=pretend.stub(verified=True))
+        user = UserFactory.create(with_verified_primary_email=True)
         pyramid_config.testing_securitypolicy(identity=user)
         pyramid_request.user = user
 
@@ -4248,11 +4249,10 @@ def test_upload_succeeds_with_gpg_signature_field(
 
         assert resp.status_code == 200
 
-    def test_upload_succeeds_without_two_factor(
+    def test_upload_fails_without_two_factor(
         self, pyramid_config, db_request, metrics, project_service, monkeypatch
     ):
-        user = UserFactory.create(totp_secret=None)
-        EmailFactory.create(user=user)
+        user = UserFactory.create(totp_secret=None, with_verified_primary_email=True)
 
         pyramid_config.testing_securitypolicy(identity=user)
         db_request.user = user
@@ -4278,19 +4278,23 @@ def test_upload_succeeds_without_two_factor(
             IProjectService: project_service,
         }.get(svc)
         db_request.user_agent = "warehouse-tests/6.6.6"
+        db_request.help_url = pretend.call_recorder(lambda **kw: "/the/help/url/")
 
-        send_email = pretend.call_recorder(lambda *a, **kw: None)
-        monkeypatch.setattr(legacy, "send_two_factor_not_yet_enabled_email", send_email)
+        with pytest.raises(HTTPBadRequest) as excinfo:
+            legacy.file_upload(db_request)
 
-        resp = legacy.file_upload(db_request)
+        resp = excinfo.value
 
-        assert resp.status_code == 200
-        assert resp.body == (
-            b"Two factor authentication is not enabled for your account."
+        assert resp.status_code == 400
+        assert resp.status == (
+            (
+                "400 User {!r} does not have two-factor authentication enabled. "
+                "Please enable two-factor authentication before attempting to "
+                "upload to PyPI. See /the/help/url/ for more information."
+            ).format(user.username)
         )
-
-        assert send_email.calls == [
-            pretend.call(db_request, user),
+        assert db_request.help_url.calls == [
+            pretend.call(_anchor="two-factor-authentication")
         ]
 
     @pytest.mark.parametrize(

warehouse/email/__init__.py

@@ -365,15 +365,6 @@ def send_account_recovery_initiated_email(
     }
 
 
-@_email(
-    "two-factor-not-yet-enabled",
-    allow_unverified=True,
-    repeat_window=datetime.timedelta(days=14),
-)
-def send_two_factor_not_yet_enabled_email(request, user):
-    return {"username": user.username}
-
-
 @_email("account-deleted")
 def send_account_deletion_email(request, user):
     return {"username": user.username}

warehouse/forklift/legacy.py

@@ -54,10 +54,7 @@
 from warehouse.authnz import Permissions
 from warehouse.classifiers.models import Classifier
 from warehouse.constants import MAX_FILESIZE, MAX_PROJECT_SIZE, ONE_GIB, ONE_MIB
-from warehouse.email import (
-    send_api_token_used_in_trusted_publisher_project_email,
-    send_two_factor_not_yet_enabled_email,
-)
+from warehouse.email import send_api_token_used_in_trusted_publisher_project_email
 from warehouse.events.tags import EventTag
 from warehouse.forklift import metadata
 from warehouse.forklift.forms import UploadForm, _filetype_extension_mapping
@@ -519,6 +516,19 @@ def file_upload(request):
                     project_help=request.help_url(_anchor="verified-email"),
                 ),
             ) from None
+        # Ensure user has enabled 2FA before they can upload a file.
+        if not request.user.has_two_factor:
+            raise _exc_with_message(
+                HTTPBadRequest,
+                (
+                    "User {!r} does not have two-factor authentication enabled. "
+                    "Please enable two-factor authentication before attempting to "
+                    "upload to PyPI. See {project_help} for more information."
+                ).format(
+                    request.user.username,
+                    project_help=request.help_url(_anchor="two-factor-authentication"),
+                ),
+            ) from None
 
     # Do some cleanup of the various form fields
     for key in list(request.POST):
@@ -1257,11 +1267,6 @@ def file_upload(request):
                 },
             )
 
-    # Check if the user has any 2FA methods enabled, and if not, email them.
-    if request.user and not request.user.has_two_factor:
-        warnings.append("Two factor authentication is not enabled for your account.")
-        send_two_factor_not_yet_enabled_email(request, request.user)
-
     request.db.flush()  # flush db now so server default values are populated for celery
 
     # Push updates to BigQuery