feat: Do not do VAPID auth on trusted connections

Closes #938

autopush/jwt.py

@@ -78,7 +78,21 @@ def extract_signature(auth):
         return payload, encoded
 
     @staticmethod
-    def decode(token, key):
+    def extract_assertion(token):
+        # type (str) -> JSONDict
+        """Extract the assertion dictionary from the passed token. This does
+        NOT do validation.
+
+        :param token: Partial or full VAPID auth token
+        :return dict of the VAPID claims
+
+        """
+        return json.loads(
+            base64.urlsafe_b64decode(
+                repad(token.split('.')[1]).encode('utf8')))
+
+    @staticmethod
+    def validate_and_extract_assertion(token, key):
         # type (str, str) -> JSONDict
         """Decode a web token into a assertion dictionary.
 
@@ -119,9 +133,7 @@ def decode(token, key):
                 signature,
                 sig_material.encode('utf8'),
                 ec.ECDSA(hashes.SHA256()))
-            return json.loads(
-                base64.urlsafe_b64decode(
-                    repad(sig_material.split('.')[1]).encode('utf8')))
+            return VerifyJWT.extract_assertion(sig_material)
         except InvalidSignature:
             raise
         except (ValueError, TypeError, binascii.Error, PyAsn1Error):

autopush/tests/test_utils.py

@@ -3,6 +3,8 @@
 from mock import patch, Mock
 from nose.tools import eq_
 
+from autopush.tests.test_integration import _get_vapid
+
 
 class TestUserAgentParser(unittest.TestCase):
     def _makeFUT(self, *args):
@@ -31,6 +33,12 @@ def test_other_os_and_browser(self):
         eq_(dd["ua_browser_family"], "Other")
         eq_(raw["ua_browser_family"], "BlackBerry")
 
+    def test_trusted_vapid(self):
+        from autopush.utils import extract_jwt
+        vapid_info = _get_vapid(payload={'sub': 'mailto:[email protected]'})
+        data = extract_jwt(vapid_info['auth'], 'invalid_key', is_trusted=True)
+        eq_(data['sub'], 'mailto:[email protected]')
+
     @patch("requests.get")
     def test_get_ec2_instance_id_unknown(self, request_mock):
         import requests

autopush/utils.py

@@ -188,13 +188,17 @@ def decipher_public_key(key_data):
     raise ValueError("Unknown public key format specified")
 
 
-def extract_jwt(token, crypto_key):
-    # type: (str, str) -> Dict[str, str]
+def extract_jwt(token, crypto_key, is_trusted=False):
+    # type: (str, str, bool) -> Dict[str, str]
     """Extract the claims from the validated JWT. """
     # first split and convert the jwt.
     if not token or not crypto_key:
         return {}
-    return jwt.decode(token, decipher_public_key(crypto_key.encode('utf8')))
+    if is_trusted:
+        return jwt.extract_assertion(token)
+    return jwt.validate_and_extract_assertion(
+        token,
+        decipher_public_key(crypto_key.encode('utf8')))
 
 
 def parse_user_agent(agent_string):

autopush/web/webpush.py

@@ -367,7 +367,11 @@ def validate_auth(self, d):
                 public_key = vapid_auth['k']
             else:
                 public_key = d["subscription"].get("public_key")
-            jwt = extract_jwt(token, public_key)
+            jwt = extract_jwt(
+                token,
+                public_key,
+                is_trusted=self.context['settings'].enable_tls_auth
+            )
         except (KeyError, ValueError, InvalidSignature, TypeError,
                 VapidAuthException):
             raise InvalidRequest("Invalid Authorization Header",