From bfae04723bd3155801f079ab5238018ec8a0cb09 Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Fri, 1 May 2026 14:34:04 +0200 Subject: Redesign API by introducing otp2289.OTPResponse type --- src/otp2289/__init__.py | 4 +- src/otp2289/__main__.py | 153 +++++++++++++---------- src/otp2289/generator.py | 314 ++++++++++++++++++++++++++--------------------- src/otp2289/server.py | 104 ++++++++-------- 4 files changed, 311 insertions(+), 264 deletions(-) (limited to 'src') diff --git a/src/otp2289/__init__.py b/src/otp2289/__init__.py index 8f8cf24..da65d9e 100644 --- a/src/otp2289/__init__.py +++ b/src/otp2289/__init__.py @@ -31,6 +31,7 @@ from .generator import ( OTPGenerator, OTPGeneratorError, OTPResponse, + OTPResponseError, ) from .server import ( OTPInvalidResponseError, @@ -41,7 +42,7 @@ from .server import ( ) __author__ = 'Simeon Simeonov' -__version__ = '2.0.0a' +__version__ = '2.0.0' __license__ = 'BSD 2-Clause' @@ -63,6 +64,7 @@ __all__ = [ 'OTPGeneratorError', 'OTPInvalidResponseError', 'OTPResponse', + 'OTPResponseError', 'OTPState', 'OTPStateError', 'OTPStore', diff --git a/src/otp2289/__main__.py b/src/otp2289/__main__.py index 0fdcc52..c5d28b6 100644 --- a/src/otp2289/__main__.py +++ b/src/otp2289/__main__.py @@ -51,6 +51,56 @@ def eprint( print(*value, sep=sep, end=end, file=sys.stderr, flush=True) +def get_rnd_seed() -> str: + """ + Returns a random seed in the format: + + 2 random letters (capitalize()) + 5 random digits + """ + rnd = secrets.SystemRandom() + return ''.join( + rnd.choices(string.ascii_lowercase, k=2) + ).capitalize() + ''.join(rnd.choices(string.digits, k=5)) + + +def initiate_new_sequence(args: argparse.Namespace) -> str: + """ + Generates a new sequence based on the parameters sent from the parser. + + :param args: The arguments assigned from argparse + :type args: argparse.Namespace + + :raises otp2289.OTPChallengeError: If the challenge is invalid + + :raises otp2289.OTPGeneratorError: If generator parameters are wrong + + :raises otp2289.OTPResponseError: If tokens or hex can not be generated + + :return: The response string + :rtype: str + """ + if not args.seed: + args.seed = get_rnd_seed() + + header = '' + if not args.quiet: + header = ( + f'Seed: {args.seed}, Step: {args.step}, ' + f'Hash: {args.hash_algo}{os.linesep}' + ) + generator = otp2289.OTPGenerator( + args.password.encode(), args.seed, args.hash_algo + ) + if args.challenge_string: + return ( + header + + generator.generate_otp_response_from_challenge( + args.challenge_string + ).hexdigest + ) + return header + generator.generate_otp_response(args.step).hexdigest + + def generate_otp_response(args: argparse.Namespace) -> str: """ Generates a response based on the parameters sent from the parser @@ -62,20 +112,22 @@ def generate_otp_response(args: argparse.Namespace) -> str: :raises otp2289.OTPGeneratorError: If generator parameters are wrong + :raises otp2289.OTPResponseError: If tokens or hex can not be generated + :return: The response string :rtype: str """ - generator = otp2289.generator.OTPGenerator( + generator = otp2289.OTPGenerator( args.password.encode(), args.seed, args.hash_algo ) if args.challenge_string: - if args.output_format == 'token': - return generator.generate_otp_words_from_challenge( - args.challenge_string - ) - return generator.generate_otp_hexdigest_from_challenge( - args.challenge_string + response = generator.generate_otp_response_from_challenge( + args.challange_string ) + if args.output_format == 'token': + return response.words + return response.hexdigest + # regular parameters header = '' if not args.quiet: @@ -83,9 +135,12 @@ def generate_otp_response(args: argparse.Namespace) -> str: f'Seed: {args.seed}, Step: {args.step}, ' f'Hash: {args.hash_algo}{os.linesep}' ) + + response = generator.generate_otp_response(args.step) + if args.output_format == 'token': - return header + generator.generate_otp_words(args.step) - return header + generator.generate_otp_hexdigest(args.step) + return header + response.words + return header + response.hexdigest def generate_otp_range(args: argparse.Namespace) -> str: @@ -99,20 +154,24 @@ def generate_otp_range(args: argparse.Namespace) -> str: :raises otp2289.OTPGeneratorError: If generator parameters are wrong + :raises otp2289.OTPResponseError: If tokens or hex can not be generated + :return: The responses string :rtype: str """ - generator = otp2289.generator.OTPGenerator( + generator = otp2289.OTPGenerator( args.password.encode(), args.seed, args.hash_algo ) - if args.output_format == 'token': - method = generator.generate_otp_words - else: - method = generator.generate_otp_hexdigest # handle most cases explicitly if args.range == 1: - return f'{args.step}: ' + method(args.step) + response = generator.generate_otp_response(args.step) + if args.output_format == 'token': + response_str = response.words + else: + response_str = response.hexdigest + return f'{args.step}: ' + response_str + args.range = min(args.range, args.step + 1) # any need for quiet? @@ -122,9 +181,19 @@ def generate_otp_range(args: argparse.Namespace) -> str: f'Seed: {args.seed}, Step: {args.step}, ' f'Hash: {args.hash_algo}, Range: {args.range}{os.linesep}' ) + + if args.output_format == 'token': + return header + os.linesep.join( + [ + f'{step}: ' + generator.generate_otp_response(step).words + for step in range(args.step, args.step - args.range, -1) + ] + ) + + # assume hex return header + os.linesep.join( [ - f'{step}: ' + method(step) + f'{step}: ' + generator.generate_otp_response(step).hexdigest for step in range(args.step, args.step - args.range, -1) ] ) @@ -172,50 +241,6 @@ def get_password(args: argparse.Namespace) -> str: return args.password -def get_rnd_seed() -> str: - """ - Returns a random seed in the format: - - 2 random letters (capitalize()) + 5 random digits - """ - rnd = secrets.SystemRandom() - return ''.join( - rnd.choices(string.ascii_lowercase, k=2) - ).capitalize() + ''.join(rnd.choices(string.digits, k=5)) - - -def initiate_new_sequence(args: argparse.Namespace) -> str: - """ - Generates a new sequence based on the parameters sent from the parser. - - :param args: The arguments assigned from argparse - :type args: argparse.Namespace - - :raises otp2289.OTPChallengeError: If the challenge is invalid - - :raises otp2289.OTPGeneratorError: If generator parameters are wrong - - :return: The response string - :rtype: str - """ - if not args.seed: - args.seed = get_rnd_seed() - header = '' - if not args.quiet: - header = ( - f'Seed: {args.seed}, Step: {args.step}, ' - f'Hash: {args.hash_algo}{os.linesep}' - ) - generator = otp2289.generator.OTPGenerator( - args.password.encode(), args.seed, args.hash_algo - ) - if args.challenge_string: - return header + generator.generate_otp_hexdigest_from_challenge( - args.challenge_string - ) - return header + generator.generate_otp_hexdigest(args.step) - - def main(inargs: list[str] | None = None) -> None: """the main entry point""" parser = argparse.ArgumentParser( @@ -278,8 +303,8 @@ def main(inargs: list[str] | None = None) -> None: parser.add_argument( '-a', '--hash-algorithm', - metavar='', type=str, + choices=['md5', 'sha1'], dest='hash_algo', default='md5', help='The hash algorithm to use. Possible values: md5 (default), sha1', @@ -296,8 +321,8 @@ def main(inargs: list[str] | None = None) -> None: parser.add_argument( '-f', '--output-format', - metavar='', type=str, + choices=['hex', 'token'], dest='output_format', default='hex', help='The output format to use. Possible values: hex (default), token', @@ -368,9 +393,9 @@ def main(inargs: list[str] | None = None) -> None: if args.generate_otp_response: print(generate_otp_response(args)) sys.exit(0) - except otp2289.generator.OTPGeneratorError as exp: + except otp2289.OTPGeneratorError as exp: eprint(f'GeneratorException: {exp}') - except otp2289.generator.OTPChallengeError as exp: + except otp2289.OTPChallengeError as exp: eprint(f'ChallengeException: {exp}') except Exception as exp: eprint(f'Unknown error: {exp}') diff --git a/src/otp2289/generator.py b/src/otp2289/generator.py index 32123b4..17024a2 100644 --- a/src/otp2289/generator.py +++ b/src/otp2289/generator.py @@ -24,16 +24,21 @@ # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """A pure Python implementation of the RFC-2289 OTP generator""" +from __future__ import annotations + import hashlib import string import typing -from collections.abc import Iterator + +if typing.TYPE_CHECKING: + from collections.abc import Iterator OTP_ALGO_MD5: typing.Final[int] = 1 OTP_ALGO_SHA1: typing.Final[int] = 2 # useful constants OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64 +OTP2289_HEX_DIGEST_SIZE: typing.Final[int] = 16 OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16 OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10 OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20 @@ -2094,12 +2099,16 @@ RFC1760_TOKENS = [ _ALGO_DICT = {OTP_ALGO_MD5: 'md5', OTP_ALGO_SHA1: 'sha1'} +class OTPChallengeError(Exception): + """OTPChallengeError class""" + + class OTPGeneratorError(Exception): """OTPGeneratorError class""" -class OTPChallengeError(Exception): - """OTPChallengeError class""" +class OTPResponseError(Exception): + """OTPResponseError class""" class OTPResponse: @@ -2120,10 +2129,24 @@ class OTPResponse: """bytes representation of the object""" return self._response_bytes + def __eq__(self, value: object, /) -> bool: + """definition for type equality""" + if not isinstance(value, OTPResponse): + return False + + return self._response_bytes == value.response_bytes + def __hash__(self) -> int: """Uses the hash value of _response_bytes""" return hash(self._response_bytes) + def __repr__(self) -> str: + """repr implementation""" + return ( + f'{self.__class__} at {id(self)} ' + f'(response_bytes={self._response_bytes!r})' + ) + @property def hexdigest(self) -> str: """Hexdigest representation of the OTP response""" @@ -2139,6 +2162,59 @@ class OTPResponse: """Tokens representation of the OTP response""" return self._words + @classmethod + def from_hex(cls, ot_hex: str) -> OTPResponse: + """ + Generates instance from the provided hexidigest with or without + leading 0x as specified by RFC-2289. + + :param ot_hex: The one-time hex to validate + :type ot_hex: str + + :raises otp2289.OTPResponseError: When the ot_hex is invalid + + :return: A new OTPResponse object + :rtype: otp2289.OTPResponse + """ + return cls(OTPResponse.hex_to_bytes(ot_hex)) + + @classmethod + def from_tokens(cls, tokens_str: str) -> OTPResponse: + """ + Generates instance from a 6 words token as specified by RFC-2289. + + :param tokens_str: String representing 6 words tokens + :type tokens_str: str + + :raises otp2289.OTPResponseError: When the tokens_str is invalid + + :return: A new OTPResponse object + :rtype: otp2289.OTPResponse + """ + return cls(OTPResponse.tokens_to_bytes(tokens_str)) + + @staticmethod + def bit_pair_sum(bit_stream: str) -> int: + """ + Split bit_stream in bit-pairs and sum them all together. + + :param bit_stream: The bit-stream object + :type bit_stream: str + + :return: The sum of all bit-pairs in bit_stream + :rtype: int + """ + if not isinstance(bit_stream, str): + raise OTPResponseError('bit_stream must be of type str') + if len(bit_stream) != OTP2289_BITSTREAM_SIZE: + raise OTPResponseError( + f'bit_stream must be of size {OTP2289_BITSTREAM_SIZE}' + ) + value = 0 + for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True): + value += int(''.join(pair), 2) + return value + @staticmethod def bytes_to_tokens(hash_bytes: bytes) -> str: """ @@ -2151,7 +2227,7 @@ class OTPResponse: :rtype: str """ bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) - bit_pair_sum = OTPGenerator.bit_pair_sum(bit_stream) + bit_pair_sum = OTPResponse.bit_pair_sum(bit_stream) tokens = [] tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)]) tokens.append(RFC1760_TOKENS[int(bit_stream[11:22], 2)]) @@ -2165,6 +2241,81 @@ class OTPResponse: ) return ' '.join(tokens) + @staticmethod + def hex_to_bytes(ot_hex: str) -> bytes: + """ + Returns bytes from the provided hexidigest. + + :param ot_hex: The one-time hex to validate + :type ot_hex: str + + :raises otp2289.OTPResponseError: If hex does not validate + + :return: The validated hex (without leading 0x) converted to bytes + :rtype: bytes + """ + if not isinstance(ot_hex, str): + raise OTPResponseError('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) != OTP2289_HEX_DIGEST_SIZE: + raise OTPResponseError( + f'The length of the hex should be {OTP2289_HEX_DIGEST_SIZE} ' + '(representing 64 bits digest)' + ) + try: + return bytes.fromhex(ot_hex) + except ValueError: + raise OTPResponseError('Invalid OT-hex') from None + + @staticmethod + def tokens_to_bytes(tokens_str: str) -> bytes: + """ + Returns bytes from a 6 words token as specified by RFC-2289. + + :param tokens_str: String representing 6 words tokens + :type tokens_str: str + + :raises otp2289.OTPResponseError: When the tokens_str is invalid + + :return: 6 words tokens + :rtype: bytes + """ + if not isinstance(tokens_str, str): + raise OTPResponseError('tokens must be a str') + tokens = tokens_str.split() + if len(tokens) != OTP2289_TOKENS_COUNT: + raise OTPResponseError( + f'Tokens-string does not contain {OTP2289_TOKENS_COUNT} tokens' + ) + token_ints = [] + try: + token_ints = [ + RFC1760_TOKENS.index(token.upper()) for token in tokens + ] + except ValueError: + raise OTPResponseError( + 'One or more words not present in RFC1760' + ) from None + # now we build a string of bits + bit_stream = format(token_ints[0], '011b') + bit_stream += format(token_ints[1], '011b') + bit_stream += format(token_ints[2], '011b') + bit_stream += format(token_ints[3], '011b') + bit_stream += format(token_ints[4], '011b') + bit_stream += format(token_ints[5], '011b') + # we have 66 bits: 64 digest + 2 bit pair sum (control number) + # RFC-2289: All OTP generators MUST calculate this checksum and all + # OTP servers MUST verify this checksum explicitly as part of the + # operation of decoding this representation of the one-time password. + if ( + f'{OTPResponse.bit_pair_sum(bit_stream[:64]):0>8b}'[-2:] + != bit_stream[-2:] + ): + raise OTPResponseError('Invalid bit checksum') + return int(bit_stream[:64], 2).to_bytes(8, 'big') + class OTPGenerator: """OTPGenerator class""" @@ -2210,28 +2361,6 @@ class OTPGenerator: f'hash_algo={self._hash_algo})' ) - @staticmethod - def bit_pair_sum(bit_stream: str) -> int: - """ - Split bit_stream in bit-pairs and sum them all together. - - :param bit_stream: The bit-stream object - :type bit_stream: str - - :return: The sum of all bit-pairs in bit_stream - :rtype: int - """ - if not isinstance(bit_stream, str): - raise OTPGeneratorError('bit_stream must be of type str') - if len(bit_stream) != OTP2289_BITSTREAM_SIZE: - raise OTPGeneratorError( - f'bit_stream must be of size {OTP2289_BITSTREAM_SIZE}' - ) - value = 0 - for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True): - value += int(''.join(pair), 2) - return value - @staticmethod def get_tokens_from_challenge(challenge: str) -> tuple[str, str, int]: """ @@ -2334,54 +2463,6 @@ class OTPGenerator: ) return bytes([byte_str1[i] ^ byte_str2[i] for i in range(length)]) - @staticmethod - def tokens_to_bytes(tokens_str: str) -> bytes: - """ - Returns bytes from a 6 words token as specified by RFC-2289. - - :param tokens_str: String representing 6 words tokens - :type tokens_str: str - - :raises otp2289.OTPGeneratorError: When the tokens_str is invalid - - :return: 6 words tokens - :rtype: bytes - """ - if not isinstance(tokens_str, str): - raise OTPGeneratorError('tokens must be a str') - tokens = tokens_str.split() - if len(tokens) != OTP2289_TOKENS_COUNT: - raise OTPGeneratorError( - f'Tokens-string does not contain {OTP2289_SHA1_DIGEST_SIZE} ' - 'tokens' - ) - token_ints = [] - try: - token_ints = [ - RFC1760_TOKENS.index(token.upper()) for token in tokens - ] - except ValueError: - raise OTPGeneratorError( - 'One or more words not present in RFC1760' - ) from None - # now we build a string of bits - bit_stream = format(token_ints[0], '011b') - bit_stream += format(token_ints[1], '011b') - bit_stream += format(token_ints[2], '011b') - bit_stream += format(token_ints[3], '011b') - bit_stream += format(token_ints[4], '011b') - bit_stream += format(token_ints[5], '011b') - # we have 66 bits: 64 digest + 2 bit pair sum (control number) - # RFC-2289: All OTP generators MUST calculate this checksum and all - # OTP servers MUST verify this checksum explicitly as part of the - # operation of decoding this representation of the one-time password. - if ( - f'{OTPGenerator.bit_pair_sum(bit_stream[:64]):0>8b}'[-2:] - != bit_stream[-2:] - ): - raise OTPGeneratorError('Invalid bit checksum') - return int(bit_stream[:64], 2).to_bytes(8, 'big') - @staticmethod def validate_hash_algo(hash_algo: int | str) -> str: """ @@ -2456,56 +2537,24 @@ class OTPGenerator: raise OTPGeneratorError('Step value MUST be >= 0') return step - def generate_otp_hexdigest(self, step: int) -> str: + def generate_otp_response(self, step: int) -> OTPResponse: """ - Generates the OTP hexdigest for the given step. + Generates OTPResponse instance for the given step :param step: The step to generate OTP for :type step: int - :return: Hexdigest for the given step - :rtype: str + :return: OTPResponse instance for the given step + :rtype: OTPResponse """ - response = OTPResponse(self._generate_otp_bytes(step)) - return response.hexdigest + return OTPResponse(self._generate_otp_bytes(step)) - def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: + def generate_otp_response_from_challenge( + self, challenge: str + ) -> OTPResponse: """ - Same as generate_otp_hexdigest, but it generates hex. from a challenge. - - RFC-2289 states: - The challenge MUST be in a standard syntax so - that automated generators can recognize the challenge in context and - extract these parameters. The syntax of the challenge is: - otp- - - :param challenge: The challenge string - :type challenge: str - - :return: Hexdigest for the given challenge - :rtype: str - """ - seed, hash_algo, step = self.get_tokens_from_challenge(challenge) - self._seed = self.validate_seed(seed) - self._hash_algo = self.validate_hash_algo(hash_algo) - return self.generate_otp_hexdigest(step) - - def generate_otp_words(self, step: int) -> str: - """ - Generates the OTP six words token for the given step. - - :param step: The step to generate OTP for - :type step: int - - :return: Six words (separated by single space) token for the given step - :rtype: str - """ - response = OTPResponse(self._generate_otp_bytes(step)) - return response.words - - def generate_otp_words_from_challenge(self, challenge: str) -> str: - """ - Same as generate_otp_words, but it generates words from a challenge. + Same as generate_otp_response, + but it generates OTPResponse from a challenge. RFC-2289 states: The challenge MUST be in a standard syntax so @@ -2522,35 +2571,14 @@ class OTPGenerator: seed, hash_algo, step = self.get_tokens_from_challenge(challenge) self._seed = self.validate_seed(seed) self._hash_algo = self.validate_hash_algo(hash_algo) - return self.generate_otp_words(step) + return self.generate_otp_response(step) - def hexdigest_range( + def otp_response_range( self, start: int = 499, stop: int = 0 - ) -> Iterator[str]: - """ - Returns an iterator that providing hexdigests corresponding to steps - from `start` to and including `stop`. - - :param start: The start of the range (default: 499) - :type start: int - - :param stop: The last step (default: 0) - :type stop: int - - :return: Iterator - :rtype: generator - """ - if not isinstance(start, int) and isinstance(stop, int): - raise OTPGeneratorError('Step value MUST be an int') - if start < stop: - raise OTPGeneratorError('Start value can not be lower than stop') - for step in range(start, stop - 1, -1): - yield self.generate_otp_hexdigest(step) - - def words_range(self, start: int = 499, stop: int = 0) -> Iterator[str]: + ) -> Iterator[OTPResponse]: """ - Returns an iterator that providing the words corresponding to steps - from `start` to and including `stop`. + Returns an iterator that is providing OTPResponse instances + corresponding to steps from `start` to and including `stop` :param start: The start of the range (default: 499) :type start: int @@ -2566,7 +2594,7 @@ class OTPGenerator: if start < stop: raise OTPGeneratorError('Start value can not be lower than stop') for step in range(start, stop - 1, -1): - yield self.generate_otp_words(step) + yield self.generate_otp_response(step) def _generate_otp_bytes(self, step: int) -> bytes: """ diff --git a/src/otp2289/server.py b/src/otp2289/server.py index 99ee460..62e6205 100644 --- a/src/otp2289/server.py +++ b/src/otp2289/server.py @@ -32,9 +32,13 @@ import typing if typing.TYPE_CHECKING: from collections.abc import Iterator -from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorError - -OTP2289_HEX_DIGEST_SIZE: typing.Final[int] = 16 +from .generator import ( + OTP_ALGO_MD5, + OTPGenerator, + OTPGeneratorError, + OTPResponse, + OTPResponseError, +) class OTPStateError(Exception): @@ -89,12 +93,15 @@ class OTPState: self._seed = OTPGenerator.validate_seed(seed) self._hash_algo = OTPGenerator.validate_hash_algo(hash_algo) self._step = OTPGenerator.validate_step(current_step) - except OTPGeneratorError as exp: - raise OTPStateError(exp.args[0]) from exp + except OTPGeneratorError as err: + raise OTPStateError(err.args[0]) from err self._current_digest = None if ot_hex is not None: - self._current_digest = self.validate_hex(ot_hex) + try: + self._current_digest = OTPResponse.hex_to_bytes(ot_hex) + except OTPResponseError as err: + raise OTPStateError(err.args[0]) from err self._new_digest_hex = None # set upon a successful validation def __repr__(self) -> str: @@ -154,70 +161,43 @@ class OTPState: :type dict_obj: dict :return: A new OTPState object - :rtype: otp2289.OTPStore + :rtype: otp2289.OTPState """ return cls(**dict_obj) @staticmethod - def response_to_bytes(response: str) -> bytes: + def response_string_to_otp_response(response_str: str) -> OTPResponse: """ 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 OTPInvalidResponseError is raised - It is up to the caller to run another iteration and compare the result - to an existing digest in this state. + The method first checks if the response string is a token and tries to + convert it to a OTPResponse instance. If that fails, the method assumes + that response string is a hex. If neither of those attempts succeeds + OTPInvalidResponseError 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 + :param response_str: The response string to this state (its challenge) + :type response_str: str :raises otp2289.OTPInvalidResponseError: If the response is corrupt/illegal, but not if it simply does not validate - :return: The bytes representation of response (if any) - :rtype: bytes + :return: OTPResponse instance + :rtype: otp2289.OTPResponse """ try: - return OTPGenerator.tokens_to_bytes(response) - except OTPGeneratorError: + return OTPResponse.from_tokens(response_str) + except OTPResponseError: # now assume hex... try: - return OTPState.validate_hex(response) - except OTPStateError: + return OTPResponse.from_hex(response_str) + except OTPResponseError: raise OTPInvalidResponseError( 'The response is neither a valid token or hex' ) from None - @staticmethod - def validate_hex(ot_hex: str) -> bytes: - """ - Validates the provided hexidigest. - - :param ot_hex: The one-time hex to validate - :type ot_hex: str - - :raises otp2289.OTPStateError: If hex does not validate - - :return: The validated hex (without leading 0x) converted to bytes - :rtype: bytes - """ - if not isinstance(ot_hex, str): - raise OTPStateError('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) != OTP2289_HEX_DIGEST_SIZE: - raise OTPStateError( - f'The length of the hex should be {OTP2289_HEX_DIGEST_SIZE} ' - '(representing 64 bits digest)' - ) - try: - return bytes.fromhex(ot_hex) - except ValueError: - raise OTPStateError('Invalid OT-hex') from None - def get_next_state(self) -> OTPState | None: """ Returns the next state for a validated OTPState. @@ -235,26 +215,38 @@ class OTPState: ) def response_validates( - self, response: str, *, store_valid_response: bool = True + self, response: str | OTPResponse, *, store_valid_response: bool = True ) -> bool: """ Validates the incoming response as specified by RFC-2289. :param response: The response to this state (its challenge) - :type response: str + :type response: str or OTPResponse :param store_valid_response: Should a valid response be stored :type store_valid_response: bool - :raises otp2289.OTPInvalidResponseError: If the response does not match - this state + :raises otp2289.OTPInvalidResponseError: If the response is corrupt + or invalid :return: Returns True if response validates, False otherwise :rtype: bool """ - # self.response_to_bytes raises OTPInvalidResponseError in case - # response is corrupt or in a wrong format - response_bytes = self.response_to_bytes(response) + # self.response_string_to_otp_response raises OTPInvalidResponseError + # in case response is corrupt or in a wrong format + if not isinstance(response, (str, OTPResponse)): + raise OTPInvalidResponseError( + 'response must be of type str or OTPResponse' + ) + + if isinstance(response, str): + response_bytes = self.response_string_to_otp_response( + response + ).response_bytes + else: + # assume OTPResponse + response_bytes = response.response_bytes + if self._hash_algo == 'md5': digest = hashlib.md5(response_bytes).digest() if ( -- cgit v1.3