Merge pull request #28 from web-push-libs/bug/27

bug: python3 byte string decoding was causing invalid headers

Author: jrconlin

python/py_vapid/__init__.py

@@ -4,6 +4,7 @@
 
 import os
 import logging
+import binascii
 import base64
 import time
 import hashlib
@@ -126,7 +127,8 @@ def validate(self, validation_token):
         :rtype: str
 
         """
-        sig = self.private_key.sign(validation_token, hashfunc=self._hasher)
+        sig = self.private_key.sign(validation_token,
+            hashfunc=self._hasher)
         verification_token = base64.urlsafe_b64encode(sig)
         return verification_token
 
@@ -154,6 +156,11 @@ def _base_sign(self, claims):
                 "'sub' is your admin email as a mailto: link.")
         return claims
 
+    def encode(self, bstring):
+        return binascii.b2a_base64(
+            bstring).decode('utf8').replace('\n', '').replace(
+            '+', '-').replace('/', '_').replace('=', '')
+
     def sign(self, claims, crypto_key=None):
         """Sign a set of claims.
         :param claims: JSON object containing the JWT claims to use.
@@ -169,7 +176,7 @@ def sign(self, claims, crypto_key=None):
         claims = self._base_sign(claims)
         sig = jws.sign(claims, self.private_key, algorithm="ES256")
         pkey = 'p256ecdsa='
-        pkey += base64.urlsafe_b64encode(self.public_key.to_string())
+        pkey += self.encode(self.public_key.to_string())
         if crypto_key:
             crypto_key = crypto_key + ',' + pkey
         else:
@@ -193,12 +200,12 @@ def sign(self, claims, crypto_key=None):
         pkey = self.public_key.to_string()
         # Make sure that the key is properly prefixed.
         if len(pkey) == 64:
-            pkey = '\04' + pkey
+            pkey = b'\04' + pkey
         return{
             "Authorization": "{schema} t={t},k={k}".format(
                 schema=self._schema,
                 t=sig,
-                k=base64.urlsafe_b64encode(pkey).strip('=')
+                k=self.encode(pkey)
             )
         }
 

python/py_vapid/tests/test_vapid.py

@@ -1,3 +1,4 @@
+import binascii
 import base64
 import hashlib
 import os
@@ -24,7 +25,7 @@
 """
 
 T_PUBLIC_RAW = """EJwJZq_GN8jJbo1GGpyU70hmP2hbWAUpQFKDBy\
-KB81yldJ9GTklBM5xqEwuPM7VuQcyiLDhvovthPIXx-gsQRQ=="""
+KB81yldJ9GTklBM5xqEwuPM7VuQcyiLDhvovthPIXx-gsQRQ==""".strip('=')
 
 
 def setUp(self):
@@ -44,21 +45,21 @@ def tearDown(self):
 class VapidTestCase(unittest.TestCase):
     def test_init(self):
         v1 = Vapid01(private_key_file="/tmp/private")
-        eq_(v1.private_key.to_pem(), T_PRIVATE)
-        eq_(v1.public_key.to_pem(), T_PUBLIC)
+        eq_(v1.private_key.to_pem(), T_PRIVATE.encode('utf8'))
+        eq_(v1.public_key.to_pem(), T_PUBLIC.encode('utf8'))
         v2 = Vapid01(private_key=T_PRIVATE)
-        eq_(v2.private_key.to_pem(), T_PRIVATE)
-        eq_(v2.public_key.to_pem(), T_PUBLIC)
+        eq_(v2.private_key.to_pem(), T_PRIVATE.encode('utf8'))
+        eq_(v2.public_key.to_pem(), T_PUBLIC.encode('utf8'))
         v3 = Vapid01(private_key=T_DER)
-        eq_(v3.private_key.to_pem(), T_PRIVATE)
-        eq_(v3.public_key.to_pem(), T_PUBLIC)
+        eq_(v3.private_key.to_pem(), T_PRIVATE.encode('utf8'))
+        eq_(v3.public_key.to_pem(), T_PUBLIC.encode('utf8'))
         no_exist = '/tmp/not_exist'
         Vapid01(private_key_file=no_exist)
         ok_(os.path.isfile(no_exist))
         os.unlink(no_exist)
 
     def repad(self, data):
-        return data + b"===="[:len(data) % 4]
+        return data + "===="[:len(data) % 4]
 
     @patch("ecdsa.SigningKey.from_pem", side_effect=Exception)
     def test_init_bad_priv(self, mm):
@@ -94,7 +95,7 @@ def test_save_public_key(self):
 
     def test_validate(self):
         v = Vapid01("/tmp/private")
-        msg = "foobar"
+        msg = "foobar".encode('utf8')
         vtoken = v.validate(msg)
         ok_(v.public_key.verify(base64.urlsafe_b64decode(vtoken),
                                 msg,
@@ -112,7 +113,7 @@ def test_sign_01(self):
         items = jws.verify(result['Authorization'].split(' ')[1],
                            v.public_key,
                            algorithms=["ES256"])
-        eq_(json.loads(items), claims)
+        eq_(json.loads(items.decode('utf8')), claims)
         result = v.sign(claims)
         eq_(result['Crypto-Key'],
             'p256ecdsa=' + T_PUBLIC_RAW)
@@ -131,9 +132,9 @@ def test_sign_02(self):
         eq_(len(parts), 2)
         t_val = json.loads(base64.urlsafe_b64decode(
             self.repad(parts[0][2:].split('.')[1])
-        ))
-        k_val = base64.urlsafe_b64decode(self.repad(parts[1][2:]))
-        eq_(k_val[0], "\04")
+        ).decode('utf8'))
+        k_val = binascii.a2b_base64(self.repad(parts[1][2:]))
+        eq_(binascii.hexlify(k_val)[:2], b'04')
         eq_(len(k_val), 65)
         for k in claims:
             eq_(t_val[k], claims[k])