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/__init__.py | 10 +-- src/otp2289/__main__.py | 22 ++++--- src/otp2289/generator.py | 165 ++++++++++++++++++++++++++++++++--------------- src/otp2289/server.py | 88 ++++++++++++++----------- 4 files changed, 185 insertions(+), 100 deletions(-) (limited to 'src/otp2289') diff --git a/src/otp2289/__init__.py b/src/otp2289/__init__.py index c9e3c74..8f8cf24 100644 --- a/src/otp2289/__init__.py +++ b/src/otp2289/__init__.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 @@ -30,6 +30,7 @@ from .generator import ( OTPChallengeError, OTPGenerator, OTPGeneratorError, + OTPResponse, ) from .server import ( OTPInvalidResponseError, @@ -40,11 +41,11 @@ from .server import ( ) __author__ = 'Simeon Simeonov' -__version__ = '1.2.2' +__version__ = '2.0.0a' __license__ = 'BSD 2-Clause' -def int_or_str(value): +def int_or_str(value: int | str) -> int | str: """Returns int value of value when possible""" try: return int(value) @@ -61,6 +62,7 @@ __all__ = [ 'OTPGenerator', 'OTPGeneratorError', 'OTPInvalidResponseError', + 'OTPResponse', 'OTPState', 'OTPStateError', 'OTPStore', diff --git a/src/otp2289/__main__.py b/src/otp2289/__main__.py index 9f1aab8..0fdcc52 100644 --- a/src/otp2289/__main__.py +++ b/src/otp2289/__main__.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 @@ -36,6 +36,7 @@ import argparse import errno import getpass import os +import pathlib import secrets import string import sys @@ -43,9 +44,11 @@ import sys import otp2289 -def eprint(*arg, **kwargs): +def eprint( + *value: object, sep: str | None = ' ', end: str | None = '\n' +) -> None: """stdderr print wrapper""" - print(*arg, file=sys.stderr, flush=True, **kwargs) + print(*value, sep=sep, end=end, file=sys.stderr, flush=True) def generate_otp_response(args: argparse.Namespace) -> str: @@ -162,8 +165,8 @@ def get_password(args: argparse.Namespace) -> str: eprint('The passwords do not match') return password - if os.path.isfile(args.password): - with open(args.password, encoding='utf-8') as fp: + if pathlib.Path(args.password).is_file(): + with pathlib.Path(args.password).open(encoding='utf-8') as fp: return fp.readline().strip() return args.password @@ -213,7 +216,7 @@ def initiate_new_sequence(args: argparse.Namespace) -> str: return header + generator.generate_otp_hexdigest(args.step) -def main(args=None): +def main(inargs: list[str] | None = None) -> None: """the main entry point""" parser = argparse.ArgumentParser( prog=__package__, @@ -344,8 +347,10 @@ def main(args=None): version=f'%(prog)s {otp2289.__version__}', help='display program-version and exit', ) - args = parser.parse_args(args) + + args = parser.parse_args(inargs) # handle the password before everything else + try: args.password = get_password(args) except KeyboardInterrupt: @@ -354,6 +359,7 @@ def main(args=None): except Exception as exp: eprint(f'Unable to fetch password: {exp}') sys.exit(1) + try: if args.initiate_new_sequence: print(initiate_new_sequence(args)) 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`. diff --git a/src/otp2289/server.py b/src/otp2289/server.py index e5ee0f2..99ee460 100644 --- a/src/otp2289/server.py +++ b/src/otp2289/server.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,11 +24,18 @@ # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """A pure Python implementation of the RFC-2289 OTP server""" -import binascii +from __future__ import annotations + import hashlib +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 + class OTPStateError(Exception): """OTPStateError class""" @@ -52,8 +59,12 @@ class OTPState: """ def __init__( - self, ot_hex: str, current_step: int, seed: str, hash_algo=OTP_ALGO_MD5 - ): + self, + ot_hex: str | None, + current_step: int, + seed: str, + hash_algo: int | str = OTP_ALGO_MD5, + ) -> None: """ Constructs an OTPState object with the given arguments. @@ -79,13 +90,14 @@ class OTPState: 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 None + raise OTPStateError(exp.args[0]) from exp + self._current_digest = None if ot_hex is not None: self._current_digest = self.validate_hex(ot_hex) self._new_digest_hex = None # set upon a successful validation - def __repr__(self): + def __repr__(self) -> str: """repr implementation""" return ( f'{self.__class__} at {id(self)} ' @@ -102,7 +114,7 @@ class OTPState: return f'otp-{self._hash_algo} {self._step} {self._seed} ' @property - def current_digest(self) -> bytes: + def current_digest(self) -> bytes | None: """current_digest-property""" return self._current_digest @@ -116,7 +128,7 @@ class OTPState: """ot_hex-property""" if self._current_digest is None: return '' - return binascii.hexlify(self._current_digest).decode() + return self._current_digest.hex() @property def seed(self) -> str: @@ -134,7 +146,7 @@ class OTPState: return bool(self._new_digest_hex) @classmethod - def from_dict(cls, dict_obj: dict): + def from_dict(cls, dict_obj: dict) -> OTPState: """ Returns an OTPState object from the dict-object @@ -196,17 +208,17 @@ class OTPState: if ot_hex.startswith('0x'): ot_hex = ot_hex[2:] ot_hex = ot_hex.strip().lower() - if len(ot_hex) != 16: + if len(ot_hex) != OTP2289_HEX_DIGEST_SIZE: raise OTPStateError( - 'The length of the hex should be 16 ' + f'The length of the hex should be {OTP2289_HEX_DIGEST_SIZE} ' '(representing 64 bits digest)' ) try: - return binascii.unhexlify(ot_hex) - except binascii.Error: + return bytes.fromhex(ot_hex) + except ValueError: raise OTPStateError('Invalid OT-hex') from None - def get_next_state(self): + def get_next_state(self) -> OTPState | None: """ Returns the next state for a validated OTPState. @@ -223,7 +235,7 @@ class OTPState: ) def response_validates( - self, response: str, store_valid_response: str = True + self, response: str, *, store_valid_response: bool = True ) -> bool: """ Validates the incoming response as specified by RFC-2289. @@ -251,9 +263,7 @@ class OTPState: == self._current_digest ): if store_valid_response: - self._new_digest_hex = binascii.hexlify( - response_bytes - ).decode() + self._new_digest_hex = response_bytes.hex() return True return False if self._hash_algo == 'sha1': @@ -266,9 +276,7 @@ class OTPState: == self._current_digest ): if store_valid_response: - self._new_digest_hex = binascii.hexlify( - response_bytes - ).decode() + self._new_digest_hex = response_bytes.hex() return True return False # this should not happen since the hash_algo is validated by the caller @@ -283,9 +291,11 @@ class OTPState: :return: The dict representation of the object :rtype: dict """ - ot_hex = self._current_digest - if ot_hex is not None: - ot_hex = binascii.hexlify(self._current_digest).decode() + ot_hex = ( + self._current_digest.hex() + if self._current_digest is not None + else None + ) return { 'ot_hex': ot_hex, 'current_step': self._step, @@ -304,27 +314,27 @@ class OTPStore: The class could serve as a base class when implementing store backends. """ - def __init__(self, data=None): + def __init__(self, data: dict | None = None) -> None: """ Constructs an OTPStore object from data - :param data: The data object, defaults to None - :type data: object or None + :param data: The data dict, defaults to None + :type data: dict or None """ self._data = {} # {key1: {state1-data...}, key2: {state2-data...}} self._states = {} # OTPState: (domain, key) - dict if data is not None: self._add_data(data) - def __contains__(self, state): + def __contains__(self, state: OTPState) -> bool: """membership test""" return state in self._states - def __iter__(self): + def __iter__(self) -> Iterator: """iterator for OTPStore""" return iter(self._data) - def __len__(self): + def __len__(self) -> int: """len() implementation""" return len(self._data) @@ -348,7 +358,7 @@ class OTPStore: """ return self._states - def add_state(self, key: str, state: OTPState): + def add_state(self, key: str, state: OTPState) -> None: """ Adds an OTPState object with a given key. @@ -367,11 +377,13 @@ class OTPStore: self._data[key] = state self._states[state] = key - def get(self, key, default=None): + def get( + self, key: str, default: OTPState | None = None + ) -> OTPState | None: """A wrapper for dict.get""" return self._data.get(key, default) - def items(self): + def items(self) -> typing.ItemsView: """A wrapper for dict.items""" return self._data.items() @@ -396,7 +408,7 @@ class OTPStore: return state def response_validates( - self, key: str, response: str, store_valid_response: bool = True + self, key: str, response: str, *, store_valid_response: bool = True ) -> bool: """ A method that wraps around OTPState.response_validates and @@ -424,7 +436,9 @@ class OTPStore: :rtype: bool """ state = self._data[key] - rvalue = state.response_validates(response, store_valid_response) + rvalue = state.response_validates( + response, store_valid_response=store_valid_response + ) if rvalue and store_valid_response: next_state = state.get_next_state() self._data[key] = next_state @@ -443,7 +457,7 @@ class OTPStore: """ return {key: state.to_dict() for key, state in self._data.items()} - def _add_data(self, dict_obj: dict) -> dict: + def _add_data(self, dict_obj: dict) -> None: """ Adds data from a dict object (dict_obj). -- cgit v1.3