From 2a6299c32e0baf3df06f8e6d6bc8695b951d630d Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Mon, 30 Mar 2020 21:57:53 +0200 Subject: Implement OTPState in the server module --- otp2289/server.py | 171 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 171 insertions(+) (limited to 'otp2289/server.py') diff --git a/otp2289/server.py b/otp2289/server.py index 096ebd7..4151026 100644 --- a/otp2289/server.py +++ b/otp2289/server.py @@ -24,3 +24,174 @@ # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """A pure Python implementation of the RFC-2289 OTP server""" +import binascii +import hashlib + +from .generator import (OTP_ALGO_MD5, + OTPGenerator, + OTPGeneratorException) + + +class OTPStateException(Exception): + """OTPStateException class""" + + +class OTPInvalidResponse(Exception): + """OTPInvalidResponse class""" + + +class OTPState: + """ + OTPState class + + The OTPState class represents a single state on the server side that can: + - generate a challenge + - validate the corresponding generated response from the generator + """ + + def __init__(self, ot_hex, current_step, seed, hash_algo=OTP_ALGO_MD5): + """ + Constructs an OTPState object with the given arguments. + + Keyword Arguments: + :param ot_hex: The one-time hex from the last successful + authentication or the first OTP of a newly + initialized sequence + :type ot_hex: str + + :param current_step: The current step that is sent with the challenge + :type current_step: int + + :param seed: The seed that is sent with the challenge + :type seed: str + + :param hash_algo: The hash algo, defaults to OTP_ALGO_MD5 + :type hash_algo: int or str + + :raises OTPStateException: In case the input does not validate + """ + # enforce the rfc2289 constraints + try: + self._seed = OTPGenerator.validate_seed(seed) + self._hash_algo = OTPGenerator.validate_hash_algo(hash_algo) + self._step = OTPGenerator.validate_step(current_step) + except OTPGeneratorException as exp: + raise OTPStateException(exp.args[0]) + self._current_digest = self.validate_hex(ot_hex) + self._new_digest_hex = None # set upon a successful validation + + def __str__(self): + """Duplicate the challenge string""" + return f'otp-{self._hash_algo} {self._step} {self._seed} ' + + @property + def challenge_string(self): + """challenge_string-property""" + # RFC-2289: "...the entire challenge string MUST be + # terminated with either a space or a new line." + return f'otp-{self._hash_algo} {self._step} {self._seed} ' + + @property + def validated(self): + """validated-property""" + return bool(self._new_digest_hex) + + @staticmethod + def response_to_bytes(response): + """ + A wrapper that handles/validates the response as specified by RFC-2289. + + The method first checks if response is a token and tries to convert + it to bytes. If that fails, the method assumes that response is a hex. + If neither of those attempts succeeds OTPInvalidResponse is raised. + It is up to the caller to run another iteration and compare the result + to an existing digest in this state. + + :param response: The response to this state (its challenge) + :type response: str + + :raises OTPInvalidResponse: If the response is corrupt / illegal, + but not if it simply does not validate + + :return: The bytes representation of response (if any) + :rtype: bytes + """ + try: + return OTPGenerator.tokens_to_bytes(response) + except OTPGeneratorException: + # now assume hex... + try: + return OTPState.validate_hex(response) + except OTPStateException: + raise OTPInvalidResponse( + 'The response is neither a valid token or hex') + + @staticmethod + def validate_hex(ot_hex): + """ + Validates the provided hexidigest. + + :param ot_hex: The one-time hex to validate + :type ot_hex: str + + :raises OTPStateException: In case hex does not validate + + :return: The validated hex (without leading 0x) converted to bytes + :rtype: bytes + """ + if not isinstance(ot_hex, str): + raise OTPStateException('OT-hex must be a str') + if ot_hex.startswith('0x'): + ot_hex = ot_hex[2:] + ot_hex = ot_hex.strip().lower() + if len(ot_hex) != 16: + raise OTPStateException('The length of the hex should be 16 ' + '(representing 64 bits digest)') + try: + return binascii.unhexlify(ot_hex) + except binascii.Error: + raise OTPStateException('Invalid OT-hex') + + def response_validates(self, response, store_valid_response=True): + """ + Validates the incoming response as specified by RFC-2289. + + :param response: The response to this state (its challenge) + :type response: str + + :param store_valid_response: Should a valid response be stored + :type store_valid_response: bool + + :raises OTPInvalidResponse: If the response does not match this state + + :return: Returns True if response validates, False otherwise + :rtype: bool + """ + # self.response_to_bytes raises OTPInvalidResponse in case response + # is corrupt or in a wrong format + response_bytes = self.response_to_bytes(response) + if self._hash_algo == 'md5': + digest = hashlib.md5(response_bytes).digest() + if ( + OTPGenerator.strxor(digest[0:8], digest[8:]) == + self._current_digest + ): + if store_valid_response: + self._new_digest_hex = binascii.hexlify( + response_bytes).decode() + return True + return False + if self._hash_algo == 'sha1': + digest = hashlib.sha1(response_bytes).digest() + if ( + OTPGenerator.sha1_digest_folding( + hashlib.sha1( + response_bytes).digest()) == self._current_digest + ): + if store_valid_response: + self._new_digest_hex = binascii.hexlify( + response_bytes).decode() + return True + return False + # this should not happen since the hash_algo is validated by the caller + raise OTPInvalidResponse(f'Ivalid hash_algo: {self._hash_algo}') -- cgit v1.3