Merge branch 'update-input-validation'

archamoney · Aug 25, 2020 · 18138b8 · 18138b8
2 parents 7198be9 + 78b17af
commit 18138b8
Show file tree

Hide file tree

Showing 5 changed files with 255 additions and 6 deletions.
diff --git a/httpsig/sign.py b/httpsig/sign.py
@@ -107,6 +107,17 @@ class HeaderSigner(Signer):
     def __init__(self, key_id, secret, algorithm=None, sign_algorithm=None, headers=None, sign_header='authorization'):
         if algorithm is None:
             algorithm = DEFAULT_ALGORITHM
+        if not key_id:
+            raise ValueError("key_id can't be empty")
+
+        if len(key_id) > 100000:
+            raise ValueError("key_id cant be larger than 100000 chars")
+
+        if not secret:
+            raise ValueError("secret can't be empty")
+
+        if len(secret) > 100000:
+            raise ValueError("secret cant be larger than 100000 chars")
 
         super(HeaderSigner, self).__init__(secret=secret, algorithm=algorithm, sign_algorithm=sign_algorithm)
         self.headers = headers or ['date']

diff --git a/httpsig/tests/test_signature.py b/httpsig/tests/test_signature.py
@@ -140,4 +140,220 @@ def test_unsupported_hash_algorithm(self):
     def test_deprecated_hash_algorithm(self):
         with pytest.raises(HttpSigException) as e:
             sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha256", salt_length=0))
-        self.assertEqual(str(e.value), "Hash algorithm: sha256 is deprecated. Please use: sha512")
+        self.assertEqual(str(e.value), "Hash algorithm: sha256 is deprecated. Please use: sha512")
+
+    def test_empty_secret(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id='Test', secret='', headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "secret can't be empty")
+
+    def test_none_secret(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id='Test', secret=None, headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "secret can't be empty")
+
+    def test_huge_secret(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id='Test', secret='x' * 1000000, headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "secret cant be larger than 100000 chars")
+
+    def test_empty_key_id(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id='', secret=self.key_2048, headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "key_id can't be empty")
+
+    def test_none_key_id(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id=None, secret=self.key_2048, headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "key_id can't be empty")
+
+    def test_huge_key_id(self):
+        with self.assertRaises(ValueError) as e:
+            sign.HeaderSigner(key_id='x' * 1000000, secret=self.key_2048, headers=[
+                '(request-target)',
+                'host',
+                'date',
+                'content-type',
+                'digest',
+                'content-length'
+            ])
+        self.assertEqual(str(e.exception), "key_id cant be larger than 100000 chars")
+
+    def test_empty_method(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method='', path=self.test_path)
+        self.assertEqual(str(e.exception), 'method and path arguments required when using "(request-target)"')
+
+    def test_none_method(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=None, path=self.test_path)
+        self.assertEqual(str(e.exception), 'method and path arguments required when using "(request-target)"')
+
+    def test_empty_path(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=self.test_method, path='')
+        self.assertEqual(str(e.exception), 'method and path arguments required when using "(request-target)"')
+
+    def test_none_path(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=self.test_method, path=None)
+        self.assertEqual(str(e.exception), 'method and path arguments required when using "(request-target)"')
+
+    def test_missing_header_host(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=self.test_method, path=self.test_path)
+        self.assertEqual(str(e.exception), 'missing required header "host"')
+
+    def test_missing_header_date(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Content-Type': self.header_content_type,
+            'Digest': self.header_digest,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=self.test_method, path=self.test_path)
+        self.assertEqual(str(e.exception), 'missing required header "date"')
+
+    def test_missing_header_digest(self):
+        hs = sign.HeaderSigner(key_id='Test', secret=self.key_2048, sign_algorithm=PSS("sha512", salt_length=0), headers=[
+            '(request-target)',
+            'host',
+            'date',
+            'content-type',
+            'digest',
+            'content-length'
+        ])
+        unsigned = {
+            'Host': self.header_host,
+            'Date': self.header_date,
+            'Content-Type': self.header_content_type,
+            'Content-Length': self.header_content_length,
+        }
+
+        with self.assertRaises(ValueError) as e:
+            hs.sign(unsigned, method=self.test_method, path=self.test_path)
+        self.assertEqual(str(e.exception), 'missing required header "digest"')
diff --git a/httpsig/tests/test_verify.py b/httpsig/tests/test_verify.py
@@ -133,8 +133,9 @@ def test_incorrect_headers(self):
                             required_headers=["some-other-header"],
                             host=HOST, method=METHOD, path=PATH,
                             sign_header=self.sign_header, sign_algorithm=self.sign_algorithm)
-        with self.assertRaises(Exception):
+        with self.assertRaises(ValueError) as e:
             hv.verify()
+        self.assertEqual(str(e.exception), 'some-other-header is a required header(s)')
 
     def test_extra_auth_headers(self):
         HOST = "example.com"
@@ -171,6 +172,21 @@ def test_extra_auth_headers(self):
             sign_algorithm=self.sign_algorithm)
         self.assertTrue(hv.verify())
 
+    def test_empty_secret(self):
+        with self.assertRaises(ValueError) as e:
+            HeaderVerifier(secret='', headers={})
+        self.assertEqual(str(e.exception), 'secret cant be empty')
+
+    def test_none_secret(self):
+        with self.assertRaises(ValueError) as e:
+            HeaderVerifier(secret=None, headers={})
+        self.assertEqual(str(e.exception), 'secret cant be empty')
+
+    def test_huge_secret(self):
+        with self.assertRaises(ValueError) as e:
+            HeaderVerifier(secret='x' * 1000000, headers={})
+        self.assertEqual(str(e.exception), 'secret cant be larger than 100000 chars')
+
 
 class TestVerifyHMACSHA256(TestVerifyHMACSHA1):
 

diff --git a/httpsig/utils.py b/httpsig/utils.py
@@ -65,7 +65,7 @@ def generate_message(required_headers, headers, host=None, method=None,
         h = h.lower()
         if h == '(request-target)':
             if not method or not path:
-                raise Exception('method and path arguments required when ' +
+                raise ValueError('method and path arguments required when ' +
                                 'using "(request-target)"')
             signable_list.append('%s: %s %s' % (h, method.lower(), path))
 
@@ -77,11 +77,11 @@ def generate_message(required_headers, headers, host=None, method=None,
                 if 'host' in headers:
                     host = headers[h]
                 else:
-                    raise Exception('missing required header "%s"' % h)
+                    raise ValueError('missing required header "%s"' % h)
             signable_list.append('%s: %s' % (h, host))
         else:
             if h not in headers:
-                raise Exception('missing required header "%s"' % h)
+                raise ValueError('missing required header "%s"' % h)
 
             signable_list.append('%s: %s' % (h, headers[h]))
 

diff --git a/httpsig/verify.py b/httpsig/verify.py
@@ -73,6 +73,12 @@ def __init__(self, headers, secret, required_headers=None, method=None,
         :param sign_algorithm:      Required for 'hs2019' algorithm, specifies the
                                     digital signature algorithm (derived from keyId) to use.
         """
+        if not secret:
+            raise ValueError("secret cant be empty")
+
+        if len(secret) > 100000:
+            raise ValueError("secret cant be larger than 100000 chars")
+
         required_headers = required_headers or ['date']
         self.headers = CaseInsensitiveDict(headers)
 
@@ -112,7 +118,7 @@ def verify(self):
         if len(set(self.required_headers) - set(auth_headers)) > 0:
             error_headers = ', '.join(
                     set(self.required_headers) - set(auth_headers))
-            raise Exception(
+            raise ValueError(
                     '{} is a required header(s)'.format(error_headers))
 
         signing_str = generate_message(