Enforce API Token requirement for 2FA users (#13830)

* Enforce API Token requirement for 2FA users

* spelling and a lil tweak

Author: ewdurbin

docs/blog/assets/2023-06-01-2fa-notice-email.png

@@ -0,0 +1,53 @@
+---
+title: Enforcement of 2FA for upload.pypi.org begins today
+description: PyPI now requires all uploads from accounts with 2FA enabled
+  to use an API token or Trusted Publisher configuration.
+author: Ee Durbin
+publish_date: 2023-06-01
+date: "2023-06-01 00:00"
+tags:
+  - security
+  - 2fa
+---
+
+Beginning today, all uploads from user accounts with 2FA enabled
+will be required to use an [API Token](https://pypi.org/help/#apitoken)
+or [Trusted Publisher](https://docs.pypi.org/trusted-publishers/) configuration
+in place of their password.
+
+This change has [been planned](https://github.com/pypi/warehouse/issues/7265)
+since 2FA was rolled out in 2019.
+In [February of 2022](https://github.com/pypi/warehouse/pull/10836)
+we began notifying users on upload that this change was coming.
+
+If you have 2FA enabled and have been using only your password to upload,
+the following email is likely familiar to you:
+
+<figure markdown>
+  ![Sample notice email](/assets/2023-06-01-2fa-notice-email.png)
+  <figcaption>
+      A sample notice email sent when users with 2FA enabled
+      upload using only their password.
+  </figcaption>
+</figure>
+
+Initially, we intended for this notice to live for six months before
+we began enforcement.
+
+However, some valid concerns were raised regarding
+the use of user-scoped API tokens for new project creation.
+
+With the [introduction of Trusted Publishers](/posts/2023-04-20-introducing-trusted-publishers/)
+PyPI now provides a way for users to publish **new** projects without 
+provisioning a user-scoped token, and to continue publishing without
+ever provisioning a long lived API token whatsoever.
+
+Given this, and our [commitment to further rolling out 2FA across PyPI](/posts/2023-05-25-securing-pypi-with-2fa/),
+we are now enforcing this policy.
+
+---
+
+_Ee Durbin is the Director of Infrastructure at
+the Python Software Foundation.
+They have been contributing to keeping PyPI online, available, and
+secure since 2013._

docs/blog/posts/2023-06-01-2fa-enforcement-for-upload.md

@@ -22,6 +22,8 @@ theme:
 markdown_extensions:
   - footnotes
   - pymdownx.superfences
+  - attr_list
+  - md_in_html
 extra_css:
   - stylesheets/extra.css
 plugins:

docs/mkdocs-blog.yml

@@ -34,7 +34,11 @@
     TokenServiceFactory,
     database_login_factory,
 )
-from warehouse.errors import BasicAuthBreachedPassword, BasicAuthFailedPassword
+from warehouse.errors import (
+    BasicAuthBreachedPassword,
+    BasicAuthFailedPassword,
+    BasicAuthTwoFactorEnabled,
+)
 from warehouse.events.tags import EventTag
 from warehouse.oidc.models import OIDCPublisher
 from warehouse.rate_limiting import IRateLimiter, RateLimit
@@ -76,6 +80,7 @@ def test_with_invalid_password(self, pyramid_request, pyramid_services):
                 lambda userid, password, tags=None: False
             ),
             is_disabled=pretend.call_recorder(lambda user_id: (False, None)),
+            has_two_factor=pretend.call_recorder(lambda uid: False),
         )
         pyramid_services.register_service(service, IUserService, None)
         pyramid_services.register_service(
@@ -212,6 +217,7 @@ def test_with_valid_password(self, monkeypatch, pyramid_request, pyramid_service
             ),
             update_user=pretend.call_recorder(lambda userid, last_login: None),
             is_disabled=pretend.call_recorder(lambda user_id: (False, None)),
+            has_two_factor=pretend.call_recorder(lambda uid: False),
         )
         breach_service = pretend.stub(
             check_password=pretend.call_recorder(lambda pw, tags=None: False)
@@ -232,6 +238,7 @@ def test_with_valid_password(self, monkeypatch, pyramid_request, pyramid_service
         assert service.find_userid.calls == [pretend.call("myuser")]
         assert service.get_user.calls == [pretend.call(2)]
         assert service.is_disabled.calls == [pretend.call(2)]
+        assert service.has_two_factor.calls == [pretend.call(2)]
         assert service.check_password.calls == [
             pretend.call(
                 2,
@@ -252,6 +259,52 @@ def test_with_valid_password(self, monkeypatch, pyramid_request, pyramid_service
             )
         ]
 
+    def test_via_basic_auth_2fa_enforced(
+        self, monkeypatch, pyramid_request, pyramid_services
+    ):
+        send_email = pretend.call_recorder(lambda *a, **kw: None)
+
+        user = pretend.stub(id=2, username="multiFactor")
+        service = pretend.stub(
+            get_user=pretend.call_recorder(lambda user_id: user),
+            find_userid=pretend.call_recorder(lambda username: 2),
+            check_password=pretend.call_recorder(
+                lambda userid, password, tags=None: True
+            ),
+            is_disabled=pretend.call_recorder(lambda user_id: (False, None)),
+            disable_password=pretend.call_recorder(
+                lambda user_id, request, reason=None: None
+            ),
+            has_two_factor=pretend.call_recorder(lambda uid: True),
+        )
+        breach_service = pretend.stub(
+            check_password=pretend.call_recorder(lambda pw, tags=None: False),
+        )
+
+        pyramid_services.register_service(service, IUserService, None)
+        pyramid_services.register_service(
+            breach_service, IPasswordBreachedService, None
+        )
+
+        pyramid_request.matched_route = pretend.stub(name="forklift.legacy.file_upload")
+
+        with pytest.raises(BasicAuthTwoFactorEnabled) as excinfo:
+            _basic_auth_check("multiFactor", "mypass", pyramid_request)
+
+        assert excinfo.value.status == (
+            "401 User multiFactor has two factor auth enabled, "
+            "an API Token or Trusted Publisher must be used "
+            "to upload in place of password."
+        )
+        assert service.find_userid.calls == [pretend.call("multiFactor")]
+        assert service.get_user.calls == [pretend.call(2)]
+        assert service.is_disabled.calls == [pretend.call(2)]
+        assert service.has_two_factor.calls == [pretend.call(2)]
+        assert service.check_password.calls == []
+        assert breach_service.check_password.calls == []
+        assert service.disable_password.calls == []
+        assert send_email.calls == []
+
     def test_via_basic_auth_compromised(
         self, monkeypatch, pyramid_request, pyramid_services
     ):
@@ -271,6 +324,7 @@ def test_via_basic_auth_compromised(
             disable_password=pretend.call_recorder(
                 lambda user_id, request, reason=None: None
             ),
+            has_two_factor=pretend.call_recorder(lambda uid: False),
         )
         breach_service = pretend.stub(
             check_password=pretend.call_recorder(lambda pw, tags=None: True),

tests/unit/accounts/test_core.py

@@ -29,6 +29,7 @@
     BasicAuthAccountFrozen,
     BasicAuthBreachedPassword,
     BasicAuthFailedPassword,
+    BasicAuthTwoFactorEnabled,
     WarehouseDenied,
 )
 from warehouse.events.tags import EventTag
@@ -75,6 +76,15 @@ def _basic_auth_check(username, password, request):
                 raise _format_exc_status(BasicAuthAccountFrozen(), "Account is frozen.")
             else:
                 raise _format_exc_status(HTTPUnauthorized(), "Account is disabled.")
+        elif login_service.has_two_factor(user.id):
+            raise _format_exc_status(
+                BasicAuthTwoFactorEnabled(),
+                (
+                    f"User {user.username} has two factor auth enabled, "
+                    "an API Token or Trusted Publisher must be used to upload "
+                    "in place of password."
+                ),
+            )
         elif login_service.check_password(
             user.id,
             password,

warehouse/accounts/security_policy.py

@@ -26,6 +26,10 @@ class BasicAuthAccountFrozen(HTTPUnauthorized):
     pass
 
 
+class BasicAuthTwoFactorEnabled(HTTPUnauthorized):
+    pass
+
+
 class WarehouseDenied(Denied):
     def __new__(cls, s, *args, reason=None, **kwargs):
         inner = super().__new__(cls, s, *args, **kwargs)