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