Merge lp:~maxiberta/canonical-identity-provider/secure-email-links into lp:canonical-identity-provider/release

[Maximiliano Bertacchini]

Use the scheme of settings.SSO_ROOT_URL in all links in emails.

IOW, uses http for dev/testing and https for staging/production.

[Daniel Manrique (roadmr)]

Looks good to me; FWIW it's even more elegant than my initial thought of just clobbering the scheme with https in all cases (but I missed the fact that it would screw things up for development/testing).

review: Approve

[Celso Providelo (cprov)]

Thanks Maxi,

Even if the hack itself is not pretty, code and tests are more structured now.

review: Approve

=== modified file 'src/identityprovider/emailutils.py'
--- src/identityprovider/emailutils.py	2018-08-17 18:38:29 +0000
+++ src/identityprovider/emailutils.py	2018-08-22 22:34:52 +0000
@@ -13,6 +13,7 @@
 
 from identityprovider.models import AuthToken, EmailAddress
 from identityprovider.models.const import AuthTokenType, EmailStatus
+from identityprovider.utils import use_root_url_scheme
 
 
 logger = logging.getLogger(__name__)
@@ -117,6 +118,7 @@
     token_path = token.get_absolute_url()
     if oid_token:
         token_path = oid_token + token_path
+    root_url = use_root_url_scheme(root_url)
     token_url = urljoin(root_url, token_path)
     context = {
         'requester': name,
@@ -135,6 +137,7 @@
             requester=account, requester_email=requester_email,
             email=email, token_type=AuthTokenType.INVALIDATEEMAIL,
             redirection_url=redirection_url)
+        root_url = use_root_url_scheme(root_url)
         token_url = urljoin(
             root_url, invalidate_email_token.get_absolute_url())
         context['invalidate_email_url'] = token_url
@@ -144,6 +147,7 @@
 
 def send_impersonation_email(root_url, email):
     """Send an email to user warning of attempted registration"""
+    root_url = use_root_url_scheme(root_url)
     url = urljoin(root_url, reverse('forgot_password'))
     context = {
         'forgotten_password_url': url, 'to_email': email,
@@ -244,6 +248,7 @@
         return
     email = email.email
     device_urls = []
+    root_url = use_root_url_scheme(root_url)
     for device in devices:
         device_urls.append(
             urljoin(root_url,
@@ -289,6 +294,7 @@
         to_email=email,
     )
     if account.unverified_emails().count() > 0:
+        root_url = use_root_url_scheme(root_url)
         url = urljoin(root_url, reverse('account-emails'))
         context['verify_emails_link'] = url
     send_fancy_email(
@@ -320,6 +326,7 @@
 
 
 def send_invitation_after_password_reset(root_url, email):
+    root_url = use_root_url_scheme(root_url)
     url = urljoin(root_url, reverse('new_account'))
     send_fancy_email(
         _("Password reset request"), 'email/invitation.txt',
@@ -345,6 +352,7 @@
     context, token, invalidate_email_token = _context_for_email_request(
         root_url, account, email, AuthTokenType.VALIDATEEMAIL,
         redirection_url=None)
+    root_url = use_root_url_scheme(root_url)
     context.update(dict(
         emails_url=urljoin(root_url, reverse('account-emails')),
         created=account.date_created, action=action,

=== modified file 'src/identityprovider/tests/test_emailutils.py'
--- src/identityprovider/tests/test_emailutils.py	2018-08-17 18:38:29 +0000
+++ src/identityprovider/tests/test_emailutils.py	2018-08-22 22:34:52 +0000
@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 
-# Copyright 2010-2017 Canonical Ltd.  This software is licensed under the
+# Copyright 2010-2018 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 import unittest
@@ -969,3 +969,70 @@
         self.factory.make_email_for_account(account, email=email,
                                             status=EmailStatus.NEW)
         self.assertEqual(True, _should_add_invalidation_link(email))
+
+
+@override_settings(SSO_ROOT_URL='https://testserver')
+class SecureLinksTestCase(SSOBaseTestCase):
+
+    email = '[e-mail]'
+    http_root_url = 'http://testserver'
+
+    def setUp(self):
+        super(SecureLinksTestCase, self).setUp()
+        self.account = self.factory.make_account(email='[e-mail]')
+
+    def test_send_impersonation_email(self):
+        send_impersonation_email(self.http_root_url, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_new_user_email(self):
+        send_new_user_email(self.http_root_url, self.account, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_password_reset_email(self):
+        send_password_reset_email(self.http_root_url, self.account, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_printed_codes_renewal_email(self):
+        counter = (settings.TWOFACTOR_PAPER_CODES -
+                   settings.TWOFACTOR_PAPER_CODES_WARN_RENEWAL + 1)
+        device = self.factory.make_device(
+            account=self.account, device_type='paper', counter=counter)
+        send_printed_codes_renewal_email(
+            self.http_root_url, self.account, [device])
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_validation_email_request(self):
+        send_validation_email_request(
+            self.http_root_url, self.account, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_invalidation_email_notice(self):
+        send_invalidation_email_notice(
+            self.http_root_url, self.account, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_invitation_after_password_reset(self):
+        send_invitation_after_password_reset(self.http_root_url, self.email)
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)
+
+    def test_send_action_required_warning(self):
+        send_action_required_warning(
+            self.http_root_url, self.account, 8, 'suspend')
+        email = mail.outbox[0]
+        self.assertNotIn('http://', email.body)
+        self.assertIn('https://', email.body)

=== modified file 'src/identityprovider/tests/test_utils.py'
--- src/identityprovider/tests/test_utils.py	2018-02-09 20:56:16 +0000
+++ src/identityprovider/tests/test_utils.py	2018-08-22 22:34:52 +0000
@@ -1,4 +1,4 @@
-# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# Copyright 2010-2018 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 import os
 import random
@@ -21,6 +21,7 @@
     get_provider_url,
     polite_form_errors,
     redirection_url_for_token,
+    use_root_url_scheme,
     validate_launchpad_password,
 )
 
@@ -39,6 +40,58 @@
         self.assertEqual(url, "https://launchpad.net/~test/TEST")
 
 
+class UseRootURLSchemeTestCase(SSOBaseTestCase):
+
+    @override_settings(SSO_ROOT_URL='https://testserver')
+    def test_with_https_root_url(self):
+        self.assertEqual(
+            use_root_url_scheme('http://0.0.0.0:8000'),
+            'https://0.0.0.0:8000')
+        self.assertEqual(
+            use_root_url_scheme('https://0.0.0.0:8000'),
+            'https://0.0.0.0:8000')
+        self.assertEqual(
+            use_root_url_scheme('http://testserver/a/b/c?x=1;y=2'),
+            'https://testserver/a/b/c?x=1;y=2')
+        self.assertEqual(
+            use_root_url_scheme('https://testserver/a/b/c?x=1;y=2'),
+            'https://testserver/a/b/c?x=1;y=2')
+
+    @override_settings(SSO_ROOT_URL='http://testserver')
+    def test_with_http_root_url(self):
+        self.assertEqual(
+            use_root_url_scheme('http://0.0.0.0:8000'), 'http://0.0.0.0:8000')
+        self.assertEqual(
+            use_root_url_scheme('https://0.0.0.0:8000'), 'http://0.0.0.0:8000')
+        self.assertEqual(
+            use_root_url_scheme('http://testserver/a/b/c?x=1;y=2'),
+            'http://testserver/a/b/c?x=1;y=2')
+        self.assertEqual(
+            use_root_url_scheme('https://testserver/a/b/c?x=1;y=2'),
+            'http://testserver/a/b/c?x=1;y=2')
+
+    def test_no_scheme(self):
+        for url in (
+                '0.0.0.0:8000',
+                'ftp://testserver/a/b/c',
+                'x/y/z',
+                ''):
+            self.assertEqual(url, use_root_url_scheme(url), url)
+
+    def test_idempotency(self):
+        for url in (
+                'http://0.0.0.0:8000',
+                'https://0.0.0.0:8000',
+                'http://testserver/a/b/c?x=1;y=2',
+                'https://testserver/a/b/c?x=1;y=2',
+                'x/y/z',
+                ''):
+            self.assertEqual(
+                use_root_url_scheme(url),
+                use_root_url_scheme(use_root_url_scheme(url)),
+                url)
+
+
 class PoliteFormErrorsTestCase(SSOBaseTestCase):
 
     def test_polite_errors(self):

=== modified file 'src/identityprovider/utils.py'
--- src/identityprovider/utils.py	2018-05-28 16:55:25 +0000
+++ src/identityprovider/utils.py	2018-08-22 22:34:52 +0000
@@ -1,4 +1,4 @@
-# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# Copyright 2010-2018 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
 from __future__ import absolute_import
@@ -8,6 +8,7 @@
 import hashlib
 import logging
 import string
+import urlparse
 from datetime import timedelta
 
 
@@ -42,6 +43,17 @@
         return None
 
 
+def use_root_url_scheme(url):
+    """Replace the given URL's scheme by the one of the root URL, if needed."""
+    sso_root_scheme = urlparse.urlparse(settings.SSO_ROOT_URL).scheme
+    parsed_url = urlparse.urlparse(url)
+    if parsed_url.scheme not in ('http', 'https'):
+        return url
+    if parsed_url.scheme == sso_root_scheme:
+        return url
+    return urlparse.urlunparse(parsed_url._replace(scheme=sso_root_scheme))
+
+
 # TODO: Remove after migrating all users' passwords to the Django cypher
 def encrypt_launchpad_password(plaintext, salt=None):
     plaintext = str(plaintext)