Require sign_algorithm for 'hs2019' algorith in Verifier

Author: fulder

httpsig/verify.py

@@ -4,7 +4,7 @@
 import base64
 import six
 
-from .sign import Signer
+from .sign import Signer, DEFAULT_SIGN_ALGORITHM
 from .utils import *
 
 
@@ -37,6 +37,11 @@ def _verify(self, data, signature):
             s = base64.b64decode(signature)
             return ct_bytes_compare(h, s)
 
+        elif self.sign_algorithm == 'PSS':
+            h = self._hash.new()
+            h.update(data)
+            return self._rsa.verify(h, base64.b64decode(signature))
+
         else:
             raise HttpSigException("Unsupported algorithm.")
 
@@ -47,7 +52,7 @@ class HeaderVerifier(Verifier):
     """
 
     def __init__(self, headers, secret, required_headers=None, method=None,
-                 path=None, host=None, sign_header='authorization'):
+                 path=None, host=None, sign_header='authorization', sign_algorithm=None):
         """
         Instantiate a HeaderVerifier object.
 
@@ -66,6 +71,8 @@ def __init__(self, headers, secret, required_headers=None, method=None,
             header, if not supplied in :param:headers.
         :param sign_header:         Optional. The header where the signature is.
             Default is 'authorization'.
+        :param sign_algorithm:      Required for 'hs2019' algorithm, specifies the
+                                    digital signature algorithm (derived from keyId) to use.
         """
         required_headers = required_headers or ['date']
         self.headers = CaseInsensitiveDict(headers)
@@ -84,8 +91,13 @@ def __init__(self, headers, secret, required_headers=None, method=None,
         self.path = path
         self.host = host
 
+        if self.auth_dict['algorithm'] != DEFAULT_SIGN_ALGORITHM:
+            print("Algorithm: {} is deprecated please update to {}".format(self.auth_dict['algorithm'], DEFAULT_SIGN_ALGORITHM))
+        elif self.auth_dict['algorithm'] == DEFAULT_SIGN_ALGORITHM and self.sign_algorithm is None:
+            raise HttpSigException("Required sign algorithm for {} algorithm not set".format(DEFAULT_SIGN_ALGORITHM))
+
         super(HeaderVerifier, self).__init__(
-                secret, algorithm=self.auth_dict['algorithm'])
+                secret, algorithm=self.auth_dict['algorithm'], sign_algorithm=sign_algorithm)
 
     def verify(self):
         """