Make verify_email require Authenticated principal

Author: di

tests/unit/accounts/test_views.py

@@ -18,6 +18,7 @@
 import pytest
 
 from pyramid.httpexceptions import HTTPMovedPermanently, HTTPSeeOther
+from sqlalchemy.orm.exc import NoResultFound
 
 from warehouse.accounts import views
 from warehouse.accounts.interfaces import (
@@ -691,7 +692,9 @@ def test_reset_password_password_date_changed(self, pyramid_request):
 class TestVerifyEmail:
 
     def test_verify_email(self, db_request, user_service, token_service):
-        email = EmailFactory(verified=False)
+        user = UserFactory()
+        email = EmailFactory(user=user, verified=False)
+        db_request.user = user
         db_request.GET.update({"token": "RANDOM_KEY"})
         db_request.route_path = pretend.call_recorder(lambda name: "/")
         token_service.loads = pretend.call_recorder(
@@ -788,7 +791,7 @@ def test_verify_email_invalid_action(self, pyramid_request):
             ),
         ]
 
-    def test_verify_email_invalid_email(self, pyramid_request):
+    def test_verify_email_not_found(self, pyramid_request):
         data = {
             'action': 'email-verify',
             'email.id': 'invalid',
@@ -801,10 +804,11 @@ def test_verify_email_invalid_email(self, pyramid_request):
         pyramid_request.session.flash = pretend.call_recorder(
             lambda *a, **kw: None
         )
-        get = pretend.call_recorder(lambda a: None)
-        pyramid_request.db = pretend.stub(
-            query=lambda a: pretend.stub(get=get)
-        )
+
+        def raise_no_result(*a):
+            raise NoResultFound
+
+        pyramid_request.db = pretend.stub(query=raise_no_result)
 
         views.verify_email(pyramid_request)
 
@@ -814,35 +818,32 @@ def test_verify_email_invalid_email(self, pyramid_request):
         assert pyramid_request.session.flash.calls == [
             pretend.call('Email not found', queue='error')
         ]
-        assert get.calls == [pretend.call(data['email.id'])]
 
-    def test_verify_email_already_verified(self, pyramid_request):
+    def test_verify_email_already_verified(self, db_request):
+        user = UserFactory()
+        email = EmailFactory(user=user, verified=True)
         data = {
             'action': 'email-verify',
-            'email.id': 'valid',
+            'email.id': email.id,
         }
-        pyramid_request.find_service = (
+        db_request.user = user
+        db_request.find_service = (
             lambda *a, **kw: pretend.stub(loads=lambda a: data)
         )
-        pyramid_request.params = {"token": "RANDOM_KEY"}
-        pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
-        pyramid_request.session.flash = pretend.call_recorder(
+        db_request.params = {"token": "RANDOM_KEY"}
+        db_request.route_path = pretend.call_recorder(lambda name: "/")
+        db_request.session.flash = pretend.call_recorder(
             lambda *a, **kw: None
         )
-        get = pretend.call_recorder(lambda a: pretend.stub(verified=True))
-        pyramid_request.db = pretend.stub(
-            query=lambda a: pretend.stub(get=get)
-        )
 
-        views.verify_email(pyramid_request)
+        views.verify_email(db_request)
 
-        assert pyramid_request.route_path.calls == [
+        assert db_request.route_path.calls == [
             pretend.call('manage.profile'),
         ]
-        assert pyramid_request.session.flash.calls == [
+        assert db_request.session.flash.calls == [
             pretend.call('Email already verified', queue='error')
         ]
-        assert get.calls == [pretend.call(data['email.id'])]
 
 
 class TestProfileCallout:

warehouse/accounts/views.py

@@ -17,9 +17,10 @@
 from pyramid.httpexceptions import (
     HTTPMovedPermanently, HTTPSeeOther, HTTPTooManyRequests,
 )
-from pyramid.security import remember, forget
+from pyramid.security import Authenticated, remember, forget
 from pyramid.view import view_config
 from sqlalchemy.orm import joinedload
+from sqlalchemy.orm.exc import NoResultFound
 
 from warehouse.accounts import REDIRECT_FIELD_NAME
 from warehouse.accounts.forms import (
@@ -341,6 +342,7 @@ def _error(message):
 @view_config(
     route_name="accounts.verify-email",
     uses_session=True,
+    effective_principals=Authenticated,
 )
 def verify_email(request):
     token_service = request.find_service(ITokenService, name="email")
@@ -363,9 +365,13 @@ def _error(message):
     if data.get('action') != "email-verify":
         return _error("Invalid token - Not an email verification token")
 
-    email = request.db.query(Email).get(data['email.id'])
-
-    if not email:
+    try:
+        email = (
+            request.db.query(Email)
+            .filter(Email.id == data['email.id'], Email.user == request.user)
+            .one()
+        )
+    except NoResultFound:
         return _error("Email not found")
 
     if email.verified: