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/server.py | 104 +++++++++++++++++++++++--------------------------- 1 file changed, 48 insertions(+), 56 deletions(-) (limited to 'src/otp2289/server.py') 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