feat(authentication.digest): add support for SHA-512-256 algorithm

Author: spaceone

httoop/authentication/digest.py

@@ -1,38 +1,43 @@
 from __future__ import annotations
 
-from hashlib import md5, sha256
+from hashlib import md5, new, sha256
 from hmac import compare_digest
+from time import time
 from typing import Callable
+from uuid import uuid4
 
 from httoop.exceptions import InvalidHeader
 from httoop.header.element import HeaderElement
 from httoop.util import ByteUnicodeDict, _
 
 
 class DigestAuthScheme:
-
     algorithms = {
-        'MD5': lambda val: md5(val).hexdigest().encode('ASCII'),  # nosec
-        'MD5-sess': lambda val: md5(val).hexdigest().encode('ASCII'),  # nosec
-        'SHA-256': lambda val: sha256(val).hexdigest().encode('ASCII'),
-        'SHA-256-sess': lambda val: sha256(val).hexdigest().encode('ASCII'),
-        # 'SHA-512-256': lambda val: sha256(val).hexdigest().encode('ASCII'), TODO: ??
-        # 'SHA-512-256-sess': lambda val: sha256(val).hexdigest().encode('ASCII'), TODO: ??
-    }
+        'MD5': lambda: md5(),  # noqa: S324
+        'SHA-256': lambda: sha256(),
+        'SHA-512-256': lambda: new('sha512_256'),
+    }  # not case insensitive per RFC
+    algorithms['MD5-sess'] = algorithms['MD5']
+    algorithms['SHA-256-sess'] = algorithms['SHA-256']
+    algorithms['SHA-512-256-sess'] = algorithms['SHA-512-256']
     qops = (b'auth', b'auth-int')  # quality of protection
 
     @classmethod
-    def get_algorithm(cls, algorithm: bytes | str) -> Callable:
+    def get_algorithm(cls, algorithm: bytes | str) -> Callable[bytes, bytes]:
         try:
-            return cls.algorithms[algorithm.decode('ASCII', 'ignore') if isinstance(algorithm, bytes) else algorithm]
+            H = cls.algorithms[algorithm.decode('ASCII', 'ignore') if isinstance(algorithm, bytes) else algorithm]
         except KeyError:
-            raise InvalidHeader(_('Unknown digest authentication algorithm: %r'), algorithm)
+            raise InvalidHeader(_('Unknown digest authentication algorithm: %r'), algorithm) from None
+
+        def _algo(value) -> bytes:
+            h = H()
+            h.update(value)
+            return h.hexdigest().encode('ASCII')
+
+        return _algo
 
     @classmethod
     def generate_nonce(cls, authinfo: ByteUnicodeDict) -> bytes:
-        from time import time
-        from uuid import uuid4
-
         nonce = b'%d:%s:%s' % (
             time(),
             authinfo.get('etag', authinfo.get('realm', b'')),
@@ -176,7 +181,7 @@ def calculate_request_digest(cls, authinfo: ByteUnicodeDict) -> bytes:
         algorithm = authinfo.get('algorithm', b'MD5').decode('ASCII', 'replace')
         H = cls.get_algorithm(algorithm)
 
-        if algorithm == 'MD5-sess' and authinfo.get('A1'):  # noqa: SIM108
+        if algorithm.endswith('-sess') and authinfo.get('A1'):  # noqa: SIM108
             secret = H(authinfo['A1'])
         else:
             secret = H(cls.A1(authinfo))
@@ -198,18 +203,17 @@ def A2(cls, params: ByteUnicodeDict) -> bytes:
         if not qop or qop == b'auth':
             return b'%s:%s' % (params['method'], params['uri'])
         if qop == b'auth-int':
-            H = cls.get_algorithm(params['algorithm'])
+            H = cls.get_algorithm(params.get('algorithm', b'MD5'))
             return b'%s:%s:%s' % (params['method'], params['uri'], H(params['entity_body']))
         raise NotImplementedError(f'Unknown quality of protection: {qop!r}')  # pragma: no cover
 
     @classmethod
     def A1(cls, params: ByteUnicodeDict) -> bytes:
         algorithm = params.get('algorithm', b'')
 
-        if not algorithm or algorithm == b'MD5':
+        if not algorithm or not algorithm.endswith(b'-sess'):
             return b'%s:%s:%s' % (params['username'], params['realm'], params['password'])
-        if algorithm == b'MD5-sess':
-            H = cls.get_algorithm(algorithm)
-            s = b'%s:%s:%s' % (params['username'], params['realm'], params['password'])
-            return b'%s:%s:%s' % (H(s), params['nonce'], params['cnonce'])
-        raise NotImplementedError(f'Unknown algorithm: {algorithm}')  # pragma: no cover
+
+        H = cls.get_algorithm(algorithm)
+        s = b'%s:%s:%s' % (params['username'], params['realm'], params['password'])
+        return b'%s:%s:%s' % (H(s), params['nonce'], params['cnonce'])

tests/authentication/test_digest.py

@@ -7,6 +7,31 @@
 from httoop.header import Authorization, WWWAuthenticate
 
 
+RFC7616_MUFASA = {
+    'username': 'Mufasa',
+    'realm': 'http-auth@example.org',
+    'password': 'Circle of Life',
+    'method': 'GET',
+    'uri': '/dir/index.html',
+    'nonce': '7ypf/xlj9XXwfDPEoM4URrv/xwf94BcCAzFZH4GiTo0v',
+    'nc': '00000001',
+    'cnonce': 'f2/wE4q74E6zIJEtWaHKaf5wv/H5QzzpXusqGemxURZJ',
+    'qop': 'auth',
+}
+
+RFC7616_JASON_DOE = {
+    'username': 'Jäsøn Doe',
+    'realm': 'api@example.org',
+    'password': 'Secret, or not?',
+    'method': 'GET',
+    'uri': '/doe.json',
+    'nonce': '5TsQWLVdgBdmrQ0XsxbDODV+57QdFR34I9HAbC/RVvkK',
+    'nc': '00000001',
+    'cnonce': 'NTg6RKcb9boFIAS3KrFK9BGeh+iDa/sm6jUMp2wds69v',
+    'qop': 'auth',
+}
+
+
 def test_digest_www_authentication(headers):
     www_auth = WWWAuthenticate('Digest', {
         'realm': 'testrealm@host.com',
@@ -112,6 +137,35 @@ def test_digest_authorization_auth_int(headers):
     assert auth.params == headers.element('Authorization').params
 
 
+@pytest.mark.parametrize(
+    ('algorithm', 'expected'),
+    [
+        ('MD5', '02c9d6f0ab6dfc20fc9e2105c8fc728b'),
+        ('MD5-sess', '4c187ba5e8ff03c06627fc4e3940fc97'),
+        ('SHA-256', '22be276ffb2b1acc389119cac518f32fb2db8a419f31ec8ba3f395d711920c6e'),
+        ('SHA-256-sess', '28a76fe8f6141a8d4d868ed1c8ff383edf29c5f03b50e1d53e7251654befbe83'),
+        ('SHA-512-256', 'd54bd6b5b9fc948b8135e347402e314d14f62d9041cb6745c7f5331c6809d221'),
+        ('SHA-512-256-sess', '87bad8ef67556e3c82f765be811beeb7c5bc61d4d7c2e538f8561a7cc04027f3'),
+    ],
+)
+def test_digest_auth_int_all_algorithms(algorithm, expected):
+    auth = {
+        'username': b'Mufasa',
+        'realm': b'testrealm@host.com',
+        'password': b'Circle Of Life',
+        'method': b'GET',
+        'uri': b'/dir/index.html',
+        'nonce': b'dcd98b7102dd2f0e8b11d0f600bfb0c093',
+        'nc': b'00000001',
+        'cnonce': b'0a4f113b',
+        'qop': b'auth-int',
+        'algorithm': algorithm.encode(),
+        'entity_body': b'foo',
+    }
+
+    assert DigestAuthRequestScheme.calculate_request_digest(auth) == expected.encode()
+
+
 def test_digest_authorization_md5_sess_a1(headers):
     auth = Authorization('Digest', {
         'username': 'Mufasa',
@@ -211,3 +265,20 @@ def test_required_parameter(params, headers):
     with pytest.raises(InvalidHeader) as excinfo:
         bytes(element)
     assert 'Missing parameter' in str(excinfo)
+
+
+def test_auth_int_defaults_to_md5_when_algorithm_omitted():
+    authinfo = {
+        'username': b'Mufasa',
+        'realm': b'testrealm@host.com',
+        'password': b'Circle Of Life',
+        'method': b'GET',
+        'uri': b'/dir/index.html',
+        'nonce': b'dcd98b7102dd2f0e8b11d0f600bfb0c093',
+        'nc': b'00000001',
+        'cnonce': b'0a4f113b',
+        'qop': b'auth-int',
+        'entity_body': b'',
+    }
+
+    DigestAuthRequestScheme.calculate_request_digest(authinfo)