From 751c74e7de1c78a151fdd8b76f220d8411a2108a Mon Sep 17 00:00:00 2001 From: Simeon Simeonov Date: Tue, 28 Apr 2026 14:27:05 +0200 Subject: Restructure the entire project, enforce linting and add support for type checkers --- src/otp2289/generator.py | 165 ++++++++++++++++++++++++++++++++--------------- 1 file changed, 114 insertions(+), 51 deletions(-) (limited to 'src/otp2289/generator.py') diff --git a/src/otp2289/generator.py b/src/otp2289/generator.py index c03d289..32123b4 100644 --- a/src/otp2289/generator.py +++ b/src/otp2289/generator.py @@ -1,6 +1,6 @@ -# SPDX-License-Identifier: BSD-2-Clause-FreeBSD +# SPDX-License-Identifier: BSD-2-Clause # -# Copyright (c) 2020-2025 Simeon Simeonov +# Copyright (c) 2020-2026 Simeon Simeonov # All rights reserved. # # Redistribution and use in source and binary forms, with or without @@ -24,12 +24,20 @@ # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """A pure Python implementation of the RFC-2289 OTP generator""" -import binascii import hashlib import string +import typing +from collections.abc import Iterator -OTP_ALGO_MD5 = 1 -OTP_ALGO_SHA1 = 2 +OTP_ALGO_MD5: typing.Final[int] = 1 +OTP_ALGO_SHA1: typing.Final[int] = 2 + +# useful constants +OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64 +OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16 +OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10 +OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20 +OTP2289_TOKENS_COUNT: typing.Final[int] = 6 # the tokens are defined in https://tools.ietf.org/html/rfc2289 # RFC1760_TOKENS = [ @@ -2094,12 +2102,79 @@ class OTPChallengeError(Exception): """OTPChallengeError class""" +class OTPResponse: + """Encapsulates the functionality for a single OTP response""" + + def __init__(self, response_bytes: bytes) -> None: + """ + Constructs a single OTP response + + :param response_bytes: The response state + :type response_bytes: bytes + """ + self._response_bytes = response_bytes + self._hexdigest = '0x' + response_bytes.hex() + self._words = self.bytes_to_tokens(response_bytes) + + def __bytes__(self) -> bytes: + """bytes representation of the object""" + return self._response_bytes + + def __hash__(self) -> int: + """Uses the hash value of _response_bytes""" + return hash(self._response_bytes) + + @property + def hexdigest(self) -> str: + """Hexdigest representation of the OTP response""" + return self._hexdigest + + @property + def response_bytes(self) -> bytes: + """response_bytes read-only property""" + return self._response_bytes + + @property + def words(self) -> str: + """Tokens representation of the OTP response""" + return self._words + + @staticmethod + def bytes_to_tokens(hash_bytes: bytes) -> str: + """ + Returns a 6 words token from bytes as specified by RFC-2289. + + :param hash_bytes: The input bytes + :type hash_bytes: bytes + + :return: 6 words tokens + :rtype: str + """ + bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) + bit_pair_sum = OTPGenerator.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)]) + tokens.append(RFC1760_TOKENS[int(bit_stream[22:33], 2)]) + tokens.append(RFC1760_TOKENS[int(bit_stream[33:44], 2)]) + tokens.append(RFC1760_TOKENS[int(bit_stream[44:55], 2)]) + tokens.append( + RFC1760_TOKENS[ + int(bit_stream[55:64] + f'{bit_pair_sum:0>8b}'[-2:], 2) + ] + ) + return ' '.join(tokens) + + class OTPGenerator: """OTPGenerator class""" def __init__( - self, password: bytes, seed: str = '', hash_algo=OTP_ALGO_MD5 - ): + self, + password: bytes, + seed: str = '', + hash_algo: int | str = OTP_ALGO_MD5, + ) -> None: """ Constructs an OTPGenerator object with a given password and seed. @@ -2121,11 +2196,14 @@ class OTPGenerator: self._hash_algo = self.validate_hash_algo(hash_algo) if not isinstance(password, bytes): raise OTPGeneratorError('Password must be a byte-string') - if len(password) < 10: - raise OTPGeneratorError('Password must be longer than 10 bytes') + if len(password) < OTP2289_MIN_PASSWORD_LENGTH: + raise OTPGeneratorError( + f'Password must be longer than {OTP2289_MIN_PASSWORD_LENGTH} ' + 'bytes' + ) self._password = password - def __repr__(self): + def __repr__(self) -> str: """repr implementation""" return ( f'{self.__class__} at {id(self)} (seed={self._seed}, ' @@ -2145,41 +2223,17 @@ class OTPGenerator: """ if not isinstance(bit_stream, str): raise OTPGeneratorError('bit_stream must be of type str') - if len(bit_stream) != 64: - raise OTPGeneratorError('bit_stream must be of size 64') + 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 bytes_to_tokens(hash_bytes: bytes) -> str: - """ - Returns a 6 words token from bytes as specified by RFC-2289. - - :param hash_bytes: The input bytes - :type hash_bytes: bytes - - :return: 6 words tokens - :rtype: str - """ - bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) - bit_pair_sum = OTPGenerator.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)]) - tokens.append(RFC1760_TOKENS[int(bit_stream[22:33], 2)]) - tokens.append(RFC1760_TOKENS[int(bit_stream[33:44], 2)]) - tokens.append(RFC1760_TOKENS[int(bit_stream[44:55], 2)]) - tokens.append( - RFC1760_TOKENS[ - int(bit_stream[55:64] + f'{bit_pair_sum:0>8b}'[-2:], 2) - ] - ) - return ' '.join(tokens) - - @staticmethod - def get_tokens_from_challenge(challenge: str) -> tuple: + def get_tokens_from_challenge(challenge: str) -> tuple[str, str, int]: """ Returns tokens (seed, hash_algo and step) from a challenge string. @@ -2218,9 +2272,10 @@ class OTPGenerator: """ if not isinstance(sha1_digest, bytes): raise OTPGeneratorError('sha1_digest must be of type bytes') - if len(sha1_digest) != 20: + if len(sha1_digest) != OTP2289_SHA1_DIGEST_SIZE: raise OTPGeneratorError( - 'sha1_digest must be 160 bits (20 bytes) long' + f'sha1_digest must be {OTP2289_SHA1_DIGEST_SIZE * 2} bits ' + f'({OTP2289_SHA1_DIGEST_SIZE} bytes) long' ) digested = list(5 * b'i') # 5 bytes (40 bits) result = list(8 * b'x') # 8 bytes (64 bits) @@ -2295,8 +2350,11 @@ class OTPGenerator: if not isinstance(tokens_str, str): raise OTPGeneratorError('tokens must be a str') tokens = tokens_str.split() - if len(tokens) != 6: - raise OTPGeneratorError('Tokens-string does not contain 6 tokens') + if len(tokens) != OTP2289_TOKENS_COUNT: + raise OTPGeneratorError( + f'Tokens-string does not contain {OTP2289_SHA1_DIGEST_SIZE} ' + 'tokens' + ) token_ints = [] try: token_ints = [ @@ -2325,7 +2383,7 @@ class OTPGenerator: return int(bit_stream[:64], 2).to_bytes(8, 'big') @staticmethod - def validate_hash_algo(hash_algo) -> str: + def validate_hash_algo(hash_algo: int | str) -> str: """ Validates the provided hash-algorithm. @@ -2342,7 +2400,7 @@ class OTPGenerator: raise OTPGeneratorError( 'hash_algo is not among the known algorithms' ) - hash_algo = _ALGO_DICT.get(hash_algo) + hash_algo = _ALGO_DICT[hash_algo] if not isinstance(hash_algo, str): raise OTPGeneratorError('hash_algo must be an int or a str') if hash_algo not in hashlib.algorithms_available: @@ -2367,9 +2425,10 @@ class OTPGenerator: """ if not isinstance(seed, str): raise OTPGeneratorError('Seed must be a string') - if not seed or len(seed) > 16: + if not seed or len(seed) > OTP2289_MAX_SEED_LENGTH: raise OTPGeneratorError( - 'The seed MUST be of 1 to 16 characters in length' + f'The seed MUST be of 1 to {OTP2289_MAX_SEED_LENGTH} ' + 'characters in length' ) for char in seed: if char not in string.ascii_letters + string.digits: @@ -2407,7 +2466,8 @@ class OTPGenerator: :return: Hexdigest for the given step :rtype: str """ - return '0x' + binascii.hexlify(self._generate_otp_bytes(step)).decode() + response = OTPResponse(self._generate_otp_bytes(step)) + return response.hexdigest def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: """ @@ -2440,7 +2500,8 @@ class OTPGenerator: :return: Six words (separated by single space) token for the given step :rtype: str """ - return self.bytes_to_tokens(self._generate_otp_bytes(step)) + response = OTPResponse(self._generate_otp_bytes(step)) + return response.words def generate_otp_words_from_challenge(self, challenge: str) -> str: """ @@ -2463,7 +2524,9 @@ class OTPGenerator: self._hash_algo = self.validate_hash_algo(hash_algo) return self.generate_otp_words(step) - def hexdigest_range(self, start: int = 499, stop: int = 0): + def hexdigest_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`. @@ -2484,7 +2547,7 @@ class OTPGenerator: for step in range(start, stop - 1, -1): yield self.generate_otp_hexdigest(step) - def words_range(self, start: int = 499, stop: int = 0): + def words_range(self, start: int = 499, stop: int = 0) -> Iterator[str]: """ Returns an iterator that providing the words corresponding to steps from `start` to and including `stop`. -- cgit v1.3