diff --git a/modules/sasl/scram.py b/modules/sasl/scram.py index 463843e4..487a2091 100644 --- a/modules/sasl/scram.py +++ b/modules/sasl/scram.py @@ -1,4 +1,5 @@ import base64, enum, hashlib, hmac, os, typing +from src import utils # IANA Hash Function Textual Names # https://tools.ietf.org/html/rfc5802#section-4 @@ -101,7 +102,7 @@ class SCRAM(object): server_key = self._hmac(self._salted_password, b"Server Key") server_signature = self._hmac(server_key, self._auth_message) - if self._constant_time_compare(server_signature, verifier): + if utils.security.constant_time_compare(server_signature, verifier): self.state = SCRAMState.Success return True else: diff --git a/src/utils/security.py b/src/utils/security.py index 266a767a..1c8f2cf4 100644 --- a/src/utils/security.py +++ b/src/utils/security.py @@ -1,4 +1,4 @@ -import socket, ssl +import hmac, socket, ssl, typing def ssl_context(cert: str=None, key: str=None, verify: bool=True ) -> ssl.SSLContext: @@ -21,3 +21,8 @@ def ssl_wrap(sock: socket.socket, cert: str=None, key: str=None, context = ssl_context(cert=cert, key=key, verify=verify) return context.wrap_socket(sock, server_side=server_side, server_hostname=hostname) + +def constant_time_compare( + a: typing.Union[str, bytes], + b: typing.Union[str, bytes]) -> bool: + return hmac.compare_digest(a, b)