Combine User and UserTokenContext for user-backed identities in requests (#15757)

* Combine User and UserTokenContext for user-backed identities

We have three types that a request.identity can be:

- `User`: when the identity is a user backed by a login session
- `UserTokenContext`: when the identity is a user backed by an API token (i.e. macaroon)
- `PublisherTokenContext`: when the identity is an OIDCPublisher backed by an API token

Of these, User and UserTokenContext are confusable and prone to error. This change
collapses them into a single UserContext type.

* Add extra check

* Update warehouse/accounts/security_policy.py

---------

Co-authored-by: William Woodruff <[email protected]>
Co-authored-by: Dustin Ingram <[email protected]>

Author: 3 people

tests/unit/accounts/test_core.py

@@ -29,7 +29,7 @@
     database_login_factory,
 )
 from warehouse.accounts.tasks import compute_user_metrics
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.oidc.interfaces import SignedClaims
 from warehouse.oidc.models import OIDCPublisher
 from warehouse.oidc.utils import PublisherTokenContext
@@ -40,15 +40,16 @@
 
 
 class TestUser:
-    def test_with_user(self, db_request):
+    def test_with_user_context_no_macaroon(self, db_request):
         user = UserFactory.create()
-        request = pretend.stub(identity=user)
+        user_ctx = UserContext(user, None)
+        request = pretend.stub(identity=user_ctx)
 
         assert accounts._user(request) is user
 
-    def test_with_user_token_context(self, db_request):
+    def test_with_user_token_context_macaroon(self, db_request):
         user = UserFactory.create()
-        user_ctx = UserTokenContext(user, pretend.stub())
+        user_ctx = UserContext(user, pretend.stub())
         request = pretend.stub(identity=user_ctx)
 
         assert accounts._user(request) is user
@@ -107,7 +108,7 @@ class TestOrganizationAccess:
     def test_organization_access(self, db_session, identity, flag, orgs, expected):
         user = None if not identity else UserFactory()
         request = pretend.stub(
-            identity=user,
+            identity=UserContext(user, None),
             find_service=lambda interface, context=None: pretend.stub(
                 get_organizations_by_user=lambda x: orgs
             ),

tests/unit/accounts/test_security_policy.py

@@ -18,7 +18,7 @@
 from pyramid.interfaces import ISecurityPolicy
 from zope.interface.verify import verifyClass
 
-from warehouse.accounts import security_policy
+from warehouse.accounts import UserContext, security_policy
 from warehouse.accounts.interfaces import IUserService
 from warehouse.utils.security_policy import AuthenticationMethod
 
@@ -451,7 +451,7 @@ def test_identity(self, monkeypatch):
             remote_addr="1.2.3.4",
         )
 
-        assert policy.identity(request) is user
+        assert policy.identity(request).user is user
         assert request.authentication_method == AuthenticationMethod.SESSION
         assert session_helper_obj.authenticated_userid.calls == [pretend.call(request)]
         assert session_helper_cls.calls == [pretend.call()]
@@ -518,14 +518,15 @@ class TestPermits:
         "principals,expected", [("user:5", True), ("user:1", False)]
     )
     def test_acl(self, monkeypatch, policy_class, principals, expected):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
             flags=pretend.stub(enabled=lambda flag: False),
-            identity=pretend.stub(
-                __principals__=lambda: principals,
-                has_primary_verified_email=True,
-                has_two_factor=True,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: principals,
+                    has_primary_verified_email=True,
+                    has_two_factor=True,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name="random.route"),
         )
@@ -535,13 +536,14 @@ def test_acl(self, monkeypatch, policy_class, principals, expected):
         assert bool(policy.permits(request, context, "myperm")) == expected
 
     def test_permits_with_unverified_email(self, monkeypatch, policy_class):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
-            identity=pretend.stub(
-                __principals__=lambda: ["user:5"],
-                has_primary_verified_email=False,
-                has_two_factor=False,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: ["user:5"],
+                    has_primary_verified_email=False,
+                    has_two_factor=False,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name="manage.projects"),
         )
@@ -551,13 +553,14 @@ def test_permits_with_unverified_email(self, monkeypatch, policy_class):
         assert not policy.permits(request, context, "myperm")
 
     def test_permits_manage_projects_with_2fa(self, monkeypatch, policy_class):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
-            identity=pretend.stub(
-                __principals__=lambda: ["user:5"],
-                has_primary_verified_email=True,
-                has_two_factor=True,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: ["user:5"],
+                    has_primary_verified_email=True,
+                    has_two_factor=True,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name="manage.projects"),
         )
@@ -567,14 +570,15 @@ def test_permits_manage_projects_with_2fa(self, monkeypatch, policy_class):
         assert policy.permits(request, context, "myperm")
 
     def test_deny_manage_projects_without_2fa(self, monkeypatch, policy_class):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
             flags=pretend.stub(enabled=lambda flag: False),
-            identity=pretend.stub(
-                __principals__=lambda: ["user:5"],
-                has_primary_verified_email=True,
-                has_two_factor=False,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: ["user:5"],
+                    has_primary_verified_email=True,
+                    has_two_factor=False,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name="manage.projects"),
         )
@@ -584,14 +588,15 @@ def test_deny_manage_projects_without_2fa(self, monkeypatch, policy_class):
         assert not policy.permits(request, context, "myperm")
 
     def test_deny_forklift_file_upload_without_2fa(self, monkeypatch, policy_class):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
             flags=pretend.stub(enabled=lambda flag: False),
-            identity=pretend.stub(
-                __principals__=lambda: ["user:5"],
-                has_primary_verified_email=True,
-                has_two_factor=False,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: ["user:5"],
+                    has_primary_verified_email=True,
+                    has_two_factor=False,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name="forklift.legacy.file_upload"),
         )
@@ -614,13 +619,14 @@ def test_deny_forklift_file_upload_without_2fa(self, monkeypatch, policy_class):
     def test_permits_2fa_routes_without_2fa(
         self, monkeypatch, policy_class, matched_route
     ):
-        monkeypatch.setattr(security_policy, "User", pretend.stub)
-
         request = pretend.stub(
-            identity=pretend.stub(
-                __principals__=lambda: ["user:5"],
-                has_primary_verified_email=True,
-                has_two_factor=False,
+            identity=UserContext(
+                user=pretend.stub(
+                    __principals__=lambda: ["user:5"],
+                    has_primary_verified_email=True,
+                    has_two_factor=False,
+                ),
+                macaroon=None,
             ),
             matched_route=pretend.stub(name=matched_route),
         )

tests/unit/forklift/test_legacy.py

@@ -30,7 +30,7 @@
 from trove_classifiers import classifiers
 from webob.multidict import MultiDict
 
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.admin.flags import AdminFlag, AdminFlagValue
 from warehouse.classifiers.models import Classifier
 from warehouse.forklift import legacy, metadata
@@ -968,7 +968,7 @@ def test_upload_escapes_nul_characters(self, pyramid_config, db_request):
 
         assert "\x00" not in db_request.POST["summary"]
 
-    @pytest.mark.parametrize("token_context", [True, False])
+    @pytest.mark.parametrize("macaroon_in_user_context", [True, False])
     @pytest.mark.parametrize(
         ("digests",),
         [
@@ -997,7 +997,7 @@ def test_successful_upload(
         pyramid_config,
         db_request,
         digests,
-        token_context,
+        macaroon_in_user_context,
         metrics,
     ):
         monkeypatch.setattr(tempfile, "tempdir", str(tmpdir))
@@ -1013,11 +1013,10 @@ def test_successful_upload(
         filename = f"{project.name}-{release.version}.tar.gz"
 
         db_request.user = user
-        if token_context:
-            user_context = UserTokenContext(user, pretend.stub())
-            pyramid_config.testing_securitypolicy(identity=user_context)
-        else:
-            pyramid_config.testing_securitypolicy(identity=user)
+        user_context = UserContext(
+            user, pretend.stub() if macaroon_in_user_context else None
+        )
+        pyramid_config.testing_securitypolicy(identity=user_context)
 
         db_request.user_agent = "warehouse-tests/6.6.6"
 
@@ -3977,7 +3976,7 @@ def test_upload_with_token_api_warns_if_trusted_publisher_configured(
                 [caveats.RequestUser(user_id=str(maintainer.id))],
                 user_id=maintainer.id,
             )
-            identity = UserTokenContext(maintainer, macaroon)
+            identity = UserContext(maintainer, macaroon)
         else:
             claims = {"sha": "somesha"}
             identity = PublisherTokenContext(publisher, SignedClaims(claims))

tests/unit/macaroons/test_caveats.py

@@ -20,7 +20,7 @@
 from pymacaroons import Macaroon
 
 from warehouse.accounts import _oidc_publisher
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.macaroons import caveats
 from warehouse.macaroons.caveats import (
     Caveat,
@@ -271,21 +271,32 @@ def test_verify_no_identity(self):
 
         assert result == Failure("token with user restriction without a user")
 
-    def test_verify_invalid_identity(self):
+    def test_verify_invalid_identity_no_user(self):
         caveat = RequestUser(user_id="invalid")
         result = caveat.verify(
             pretend.stub(identity=pretend.stub()), pretend.stub(), pretend.stub()
         )
 
         assert result == Failure("token with user restriction without a user")
 
+    def test_verify_invalid_identity_no_macaroon(self, db_request):
+        user = UserFactory.create()
+        user_context = UserContext(user, None)
+
+        caveat = RequestUser(user_id=str(user.id))
+        result = caveat.verify(
+            pretend.stub(identity=user_context), pretend.stub(), pretend.stub()
+        )
+
+        assert result == Failure("token with user restriction without a macaroon")
+
     def test_verify_invalid_user_id(self, db_request):
         user = UserFactory.create()
-        user_token_context = UserTokenContext(user, pretend.stub())
+        user_context = UserContext(user, pretend.stub())
 
         caveat = RequestUser(user_id="invalid")
         result = caveat.verify(
-            pretend.stub(identity=user_token_context), pretend.stub(), pretend.stub()
+            pretend.stub(identity=user_context), pretend.stub(), pretend.stub()
         )
 
         assert result == Failure(
@@ -294,11 +305,11 @@ def test_verify_invalid_user_id(self, db_request):
 
     def test_verify_ok(self, db_request):
         user = UserFactory.create()
-        user_token_context = UserTokenContext(user, pretend.stub())
+        user_context = UserContext(user, pretend.stub())
 
         caveat = RequestUser(user_id=str(user.id))
         result = caveat.verify(
-            pretend.stub(identity=user_token_context), pretend.stub(), pretend.stub()
+            pretend.stub(identity=user_context), pretend.stub(), pretend.stub()
         )
 
         assert result == Success()

tests/unit/macaroons/test_security_policy.py

@@ -20,7 +20,7 @@
 from zope.interface.verify import verifyClass
 
 from warehouse.accounts.interfaces import IUserService
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.authnz import Permissions
 from warehouse.macaroons import security_policy
 from warehouse.macaroons.interfaces import IMacaroonService
@@ -215,7 +215,7 @@ def test_identity_user(self, monkeypatch):
             ),
         )
 
-        assert policy.identity(request) == UserTokenContext(user, macaroon)
+        assert policy.identity(request) == UserContext(user, macaroon)
         assert extract_http_macaroon.calls == [pretend.call(request)]
         assert request.find_service.calls == [
             pretend.call(IMacaroonService, context=None),

tests/unit/macaroons/test_utils.py

@@ -13,12 +13,12 @@
 import pretend
 
 from tests.common.db.accounts import UserFactory
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.utils.security_policy import principals_for
 
 
-def test_usertoken_context_principals(db_request):
+def test_user_context_principals(db_request):
     user = UserFactory.create()
     assert principals_for(
-        UserTokenContext(user=user, macaroon=pretend.stub())
+        UserContext(user=user, macaroon=pretend.stub())
     ) == principals_for(user)

tests/unit/utils/test_security_policy.py

@@ -16,7 +16,7 @@
 from zope.interface.verify import verifyClass
 
 from tests.common.db.accounts import UserFactory
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.utils import security_policy
 
 
@@ -88,9 +88,9 @@ def test_authenticated_userid_nonuser_identity(self, db_request):
 
         assert policy.authenticated_userid(request) is None
 
-    def test_authenticated_userid_user_token_context(self, db_request):
+    def test_authenticated_userid_user_contex_macaroon(self, db_request):
         user = UserFactory.create()
-        user_ctx = UserTokenContext(user, pretend.stub())
+        user_ctx = UserContext(user, pretend.stub())
 
         request = pretend.stub(add_finished_callback=lambda *a, **kw: None)
         subpolicies = [pretend.stub(identity=lambda r: user_ctx)]
@@ -102,11 +102,12 @@ def test_authenticated_userid_user_token_context(self, db_request):
             == str(user_ctx.user.id)
         )
 
-    def test_authenticated_userid(self, db_request):
+    def test_authenticated_userid_user_context_no_macaroon(self, db_request):
         user = UserFactory.create()
+        user_ctx = UserContext(user, None)
 
         request = pretend.stub(add_finished_callback=lambda *a, **kw: None)
-        subpolicies = [pretend.stub(identity=lambda r: user)]
+        subpolicies = [pretend.stub(identity=lambda r: user_ctx)]
         policy = security_policy.MultiSecurityPolicy(subpolicies)
 
         assert policy.authenticated_userid(request) == str(user.id)

warehouse/accounts/__init__.py

@@ -18,7 +18,6 @@
     ITokenService,
     IUserService,
 )
-from warehouse.accounts.models import User
 from warehouse.accounts.security_policy import (
     BasicAuthSecurityPolicy,
     SessionSecurityPolicy,
@@ -32,7 +31,7 @@
     database_login_factory,
 )
 from warehouse.accounts.tasks import compute_user_metrics
-from warehouse.accounts.utils import UserTokenContext
+from warehouse.accounts.utils import UserContext
 from warehouse.admin.flags import AdminFlagValue
 from warehouse.macaroons.security_policy import MacaroonSecurityPolicy
 from warehouse.oidc.utils import PublisherTokenContext
@@ -55,12 +54,8 @@ def _user(request):
     if request.identity is None:
         return None
 
-    if isinstance(request.identity, UserTokenContext):
-        # A UserTokenContext signals a user-created API token;
-        # take the underlying user.
+    if isinstance(request.identity, UserContext):
         return request.identity.user
-    elif isinstance(request.identity, User):
-        return request.identity
     else:
         return None