Manage emails (#2898)

* Break out new email field into mixin

* Add an AddEmailForm

* Show errors from SaveProfileForm

* Add an email TokenService

* Verification email sending and templates

* Add email verification view

* Mange email views

Adding emails, deleting emails, changing primary emails, resending
verification emails.

* Use a default response property

* Make verify_email require Authenticated principal

* Break out new email field into mixin

* Add an AddEmailForm

* Show errors from SaveProfileForm

* Add an email TokenService

* Verification email sending and templates

* Add email verification view

* Use a default response property

* Mange email views

Adding emails, deleting emails, changing primary emails, resending
verification emails.

* Make verify_email require Authenticated principal

* Remove old/unused verify_email function

* Activate user on email verification

* Style email page, update messages

* Fix linting errors

dev/environment

@@ -34,3 +34,4 @@ MAIL_SSL=false
 STATUSPAGE_URL=https://2p66nmmycsj3.statuspage.io
 
 TOKEN_PASSWORD_SECRET="an insecure password reset secret key"
+TOKEN_EMAIL_SECRET="an insecure email verification secret key"

tests/common/db/accounts.py

@@ -46,3 +46,4 @@ class Meta:
  user = factory.SubFactory(UserFactory)
  email = FuzzyEmail()
  verified = True
+ primary = True

tests/unit/accounts/test_services.py

@@ -162,16 +162,6 @@ def test_update_user(self, user_service):
  assert password != user_from_db.password
  assert user_service.hasher.verify(password, user_from_db.password)
 
- def test_verify_email(self, user_service):
- user = UserFactory.create()
- EmailFactory.create(user=user, primary=True,
- verified=False)
- EmailFactory.create(user=user, primary=False,
- verified=False)
- user_service.verify_email(user.id, user.emails[0].email)
- assert user.emails[0].verified
- assert not user.emails[1].verified
-
  def test_find_by_email(self, user_service):
  user = UserFactory.create()
  EmailFactory.create(user=user, primary=True, verified=False)

tests/unit/accounts/test_views.py

@@ -18,14 +18,15 @@
 import pytest
 
 from pyramid.httpexceptions import HTTPMovedPermanently, HTTPSeeOther
+from sqlalchemy.orm.exc import NoResultFound
 
 from warehouse.accounts import views
 from warehouse.accounts.interfaces import (
  IUserService, ITokenService, TokenExpired, TokenInvalid, TokenMissing,
  TooManyFailedLogins
 )
 
-from ...common.db.accounts import UserFactory
+from ...common.db.accounts import EmailFactory, UserFactory
 
 
 class TestFailedLoginView:
@@ -695,6 +696,166 @@ def test_reset_password_password_date_changed(self, pyramid_request):
  ]
 
 
+class TestVerifyEmail:
+
+ def test_verify_email(self, db_request, user_service, token_service):
+ user = UserFactory(is_active=False)
+ 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(
+ lambda token: {
+ 'action': 'email-verify',
+ 'email.id': str(email.id),
+ }
+ )
+ db_request.find_service = pretend.call_recorder(
+ lambda *a, **kwargs: token_service,
+ )
+ db_request.session.flash = pretend.call_recorder(
+ lambda *a, **kw: None
+ )
+
+ result = views.verify_email(db_request)
+
+ db_request.db.flush()
+ assert email.verified
+ assert user.is_active
+ assert isinstance(result, HTTPSeeOther)
+ assert result.headers["Location"] == "/"
+ assert db_request.route_path.calls == [pretend.call('manage.profile')]
+ assert token_service.loads.calls == [pretend.call('RANDOM_KEY')]
+ assert db_request.session.flash.calls == [
+ pretend.call(
+ f"Email address {email.email} verified. " +
+ "You can now set this email as your primary address.",
+ queue="success"
+ ),
+ ]
+ assert db_request.find_service.calls == [
+ pretend.call(ITokenService, name="email"),
+ ]
+
+ @pytest.mark.parametrize(
+ ("exception", "message"),
+ [
+ (
+ TokenInvalid,
+ "Invalid token - Request a new verification link",
+ ), (
+ TokenExpired,
+ "Expired token - Request a new verification link",
+ ), (
+ TokenMissing,
+ "Invalid token - No token supplied"
+ ),
+ ],
+ )
+ def test_verify_email_loads_failure(
+ self, pyramid_request, exception, message):
+
+ def loads(token):
+ raise exception
+
+ pyramid_request.find_service = (
+ lambda *a, **kw: pretend.stub(loads=loads)
+ )
+ pyramid_request.params = {"token": "RANDOM_KEY"}
+ pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
+ pyramid_request.session.flash = pretend.call_recorder(
+ lambda *a, **kw: None
+ )
+
+ views.verify_email(pyramid_request)
+
+ assert pyramid_request.route_path.calls == [
+ pretend.call('manage.profile'),
+ ]
+ assert pyramid_request.session.flash.calls == [
+ pretend.call(message, queue='error'),
+ ]
+
+ def test_verify_email_invalid_action(self, pyramid_request):
+ data = {
+ 'action': 'invalid-action',
+ }
+ pyramid_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(
+ lambda *a, **kw: None
+ )
+
+ views.verify_email(pyramid_request)
+
+ assert pyramid_request.route_path.calls == [
+ pretend.call('manage.profile'),
+ ]
+ assert pyramid_request.session.flash.calls == [
+ pretend.call(
+ "Invalid token - Not an email verification token",
+ queue='error',
+ ),
+ ]
+
+ def test_verify_email_not_found(self, pyramid_request):
+ data = {
+ 'action': 'email-verify',
+ 'email.id': 'invalid',
+ }
+ pyramid_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(
+ lambda *a, **kw: None
+ )
+
+ def raise_no_result(*a):
+ raise NoResultFound
+
+ pyramid_request.db = pretend.stub(query=raise_no_result)
+
+ views.verify_email(pyramid_request)
+
+ assert pyramid_request.route_path.calls == [
+ pretend.call('manage.profile'),
+ ]
+ assert pyramid_request.session.flash.calls == [
+ pretend.call('Email not found', queue='error')
+ ]
+
+ def test_verify_email_already_verified(self, db_request):
+ user = UserFactory()
+ email = EmailFactory(user=user, verified=True)
+ data = {
+ 'action': 'email-verify',
+ 'email.id': email.id,
+ }
+ db_request.user = user
+ db_request.find_service = (
+ lambda *a, **kw: pretend.stub(loads=lambda a: data)
+ )
+ 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
+ )
+
+ views.verify_email(db_request)
+
+ assert db_request.route_path.calls == [
+ pretend.call('manage.profile'),
+ ]
+ assert db_request.session.flash.calls == [
+ pretend.call('Email already verified', queue='error')
+ ]
+
+
 class TestProfileCallout:
 
  def test_profile_callout_returns_user(self):

tests/unit/manage/test_forms.py

@@ -69,3 +69,12 @@ def test_validate_role_name_fails(self, value, expected):
 
  assert not form.validate()
  assert form.role_name.errors == [expected]
+
+
+class TestAddEmailForm:
+
+ def test_creation(self):
+ user_service = pretend.stub()
+ form = forms.AddEmailForm(user_service=user_service)
+
+ assert form.user_service is user_service