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/generator.py | 314 ++++++++++++++++++++++++++--------------------- 1 file changed, 171 insertions(+), 143 deletions(-) (limited to 'src/otp2289/generator.py') 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: """ -- cgit v1.3