diff options
Diffstat (limited to 'src/otp2289')
| -rw-r--r-- | src/otp2289/__init__.py | 10 | ||||
| -rw-r--r-- | src/otp2289/__main__.py | 22 | ||||
| -rw-r--r-- | src/otp2289/generator.py | 165 | ||||
| -rw-r--r-- | src/otp2289/server.py | 88 |
4 files changed, 185 insertions, 100 deletions
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 @@ | |||
| 1 | # SPDX-License-Identifier: BSD-2-Clause-FreeBSD | 1 | # SPDX-License-Identifier: BSD-2-Clause |
| 2 | # | 2 | # |
| 3 | # Copyright (c) 2020-2025 Simeon Simeonov | 3 | # Copyright (c) 2020-2026 Simeon Simeonov |
| 4 | # All rights reserved. | 4 | # All rights reserved. |
| 5 | # | 5 | # |
| 6 | # Redistribution and use in source and binary forms, with or without | 6 | # Redistribution and use in source and binary forms, with or without |
| @@ -30,6 +30,7 @@ from .generator import ( | |||
| 30 | OTPChallengeError, | 30 | OTPChallengeError, |
| 31 | OTPGenerator, | 31 | OTPGenerator, |
| 32 | OTPGeneratorError, | 32 | OTPGeneratorError, |
| 33 | OTPResponse, | ||
| 33 | ) | 34 | ) |
| 34 | from .server import ( | 35 | from .server import ( |
| 35 | OTPInvalidResponseError, | 36 | OTPInvalidResponseError, |
| @@ -40,11 +41,11 @@ from .server import ( | |||
| 40 | ) | 41 | ) |
| 41 | 42 | ||
| 42 | __author__ = 'Simeon Simeonov' | 43 | __author__ = 'Simeon Simeonov' |
| 43 | __version__ = '1.2.2' | 44 | __version__ = '2.0.0a' |
| 44 | __license__ = 'BSD 2-Clause' | 45 | __license__ = 'BSD 2-Clause' |
| 45 | 46 | ||
| 46 | 47 | ||
| 47 | def int_or_str(value): | 48 | def int_or_str(value: int | str) -> int | str: |
| 48 | """Returns int value of value when possible""" | 49 | """Returns int value of value when possible""" |
| 49 | try: | 50 | try: |
| 50 | return int(value) | 51 | return int(value) |
| @@ -61,6 +62,7 @@ __all__ = [ | |||
| 61 | 'OTPGenerator', | 62 | 'OTPGenerator', |
| 62 | 'OTPGeneratorError', | 63 | 'OTPGeneratorError', |
| 63 | 'OTPInvalidResponseError', | 64 | 'OTPInvalidResponseError', |
| 65 | 'OTPResponse', | ||
| 64 | 'OTPState', | 66 | 'OTPState', |
| 65 | 'OTPStateError', | 67 | 'OTPStateError', |
| 66 | 'OTPStore', | 68 | '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 @@ | |||
| 1 | # SPDX-License-Identifier: BSD-2-Clause-FreeBSD | 1 | # SPDX-License-Identifier: BSD-2-Clause |
| 2 | # | 2 | # |
| 3 | # Copyright (c) 2020-2025 Simeon Simeonov | 3 | # Copyright (c) 2020-2026 Simeon Simeonov |
| 4 | # All rights reserved. | 4 | # All rights reserved. |
| 5 | # | 5 | # |
| 6 | # Redistribution and use in source and binary forms, with or without | 6 | # Redistribution and use in source and binary forms, with or without |
| @@ -36,6 +36,7 @@ import argparse | |||
| 36 | import errno | 36 | import errno |
| 37 | import getpass | 37 | import getpass |
| 38 | import os | 38 | import os |
| 39 | import pathlib | ||
| 39 | import secrets | 40 | import secrets |
| 40 | import string | 41 | import string |
| 41 | import sys | 42 | import sys |
| @@ -43,9 +44,11 @@ import sys | |||
| 43 | import otp2289 | 44 | import otp2289 |
| 44 | 45 | ||
| 45 | 46 | ||
| 46 | def eprint(*arg, **kwargs): | 47 | def eprint( |
| 48 | *value: object, sep: str | None = ' ', end: str | None = '\n' | ||
| 49 | ) -> None: | ||
| 47 | """stdderr print wrapper""" | 50 | """stdderr print wrapper""" |
| 48 | print(*arg, file=sys.stderr, flush=True, **kwargs) | 51 | print(*value, sep=sep, end=end, file=sys.stderr, flush=True) |
| 49 | 52 | ||
| 50 | 53 | ||
| 51 | def generate_otp_response(args: argparse.Namespace) -> str: | 54 | def generate_otp_response(args: argparse.Namespace) -> str: |
| @@ -162,8 +165,8 @@ def get_password(args: argparse.Namespace) -> str: | |||
| 162 | eprint('The passwords do not match') | 165 | eprint('The passwords do not match') |
| 163 | return password | 166 | return password |
| 164 | 167 | ||
| 165 | if os.path.isfile(args.password): | 168 | if pathlib.Path(args.password).is_file(): |
| 166 | with open(args.password, encoding='utf-8') as fp: | 169 | with pathlib.Path(args.password).open(encoding='utf-8') as fp: |
| 167 | return fp.readline().strip() | 170 | return fp.readline().strip() |
| 168 | 171 | ||
| 169 | return args.password | 172 | return args.password |
| @@ -213,7 +216,7 @@ def initiate_new_sequence(args: argparse.Namespace) -> str: | |||
| 213 | return header + generator.generate_otp_hexdigest(args.step) | 216 | return header + generator.generate_otp_hexdigest(args.step) |
| 214 | 217 | ||
| 215 | 218 | ||
| 216 | def main(args=None): | 219 | def main(inargs: list[str] | None = None) -> None: |
| 217 | """the main entry point""" | 220 | """the main entry point""" |
| 218 | parser = argparse.ArgumentParser( | 221 | parser = argparse.ArgumentParser( |
| 219 | prog=__package__, | 222 | prog=__package__, |
| @@ -344,8 +347,10 @@ def main(args=None): | |||
| 344 | version=f'%(prog)s {otp2289.__version__}', | 347 | version=f'%(prog)s {otp2289.__version__}', |
| 345 | help='display program-version and exit', | 348 | help='display program-version and exit', |
| 346 | ) | 349 | ) |
| 347 | args = parser.parse_args(args) | 350 | |
| 351 | args = parser.parse_args(inargs) | ||
| 348 | # handle the password before everything else | 352 | # handle the password before everything else |
| 353 | |||
| 349 | try: | 354 | try: |
| 350 | args.password = get_password(args) | 355 | args.password = get_password(args) |
| 351 | except KeyboardInterrupt: | 356 | except KeyboardInterrupt: |
| @@ -354,6 +359,7 @@ def main(args=None): | |||
| 354 | except Exception as exp: | 359 | except Exception as exp: |
| 355 | eprint(f'Unable to fetch password: {exp}') | 360 | eprint(f'Unable to fetch password: {exp}') |
| 356 | sys.exit(1) | 361 | sys.exit(1) |
| 362 | |||
| 357 | try: | 363 | try: |
| 358 | if args.initiate_new_sequence: | 364 | if args.initiate_new_sequence: |
| 359 | print(initiate_new_sequence(args)) | 365 | 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 @@ | |||
| 1 | # SPDX-License-Identifier: BSD-2-Clause-FreeBSD | 1 | # SPDX-License-Identifier: BSD-2-Clause |
| 2 | # | 2 | # |
| 3 | # Copyright (c) 2020-2025 Simeon Simeonov | 3 | # Copyright (c) 2020-2026 Simeon Simeonov |
| 4 | # All rights reserved. | 4 | # All rights reserved. |
| 5 | # | 5 | # |
| 6 | # Redistribution and use in source and binary forms, with or without | 6 | # Redistribution and use in source and binary forms, with or without |
| @@ -24,12 +24,20 @@ | |||
| 24 | # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. | 24 | # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| 25 | """A pure Python implementation of the RFC-2289 OTP generator""" | 25 | """A pure Python implementation of the RFC-2289 OTP generator""" |
| 26 | 26 | ||
| 27 | import binascii | ||
| 28 | import hashlib | 27 | import hashlib |
| 29 | import string | 28 | import string |
| 29 | import typing | ||
| 30 | from collections.abc import Iterator | ||
| 30 | 31 | ||
| 31 | OTP_ALGO_MD5 = 1 | 32 | OTP_ALGO_MD5: typing.Final[int] = 1 |
| 32 | OTP_ALGO_SHA1 = 2 | 33 | OTP_ALGO_SHA1: typing.Final[int] = 2 |
| 34 | |||
| 35 | # useful constants | ||
| 36 | OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64 | ||
| 37 | OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16 | ||
| 38 | OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10 | ||
| 39 | OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20 | ||
| 40 | OTP2289_TOKENS_COUNT: typing.Final[int] = 6 | ||
| 33 | 41 | ||
| 34 | # the tokens are defined in https://tools.ietf.org/html/rfc2289 # | 42 | # the tokens are defined in https://tools.ietf.org/html/rfc2289 # |
| 35 | RFC1760_TOKENS = [ | 43 | RFC1760_TOKENS = [ |
| @@ -2094,12 +2102,79 @@ class OTPChallengeError(Exception): | |||
| 2094 | """OTPChallengeError class""" | 2102 | """OTPChallengeError class""" |
| 2095 | 2103 | ||
| 2096 | 2104 | ||
| 2105 | class OTPResponse: | ||
| 2106 | """Encapsulates the functionality for a single OTP response""" | ||
| 2107 | |||
| 2108 | def __init__(self, response_bytes: bytes) -> None: | ||
| 2109 | """ | ||
| 2110 | Constructs a single OTP response | ||
| 2111 | |||
| 2112 | :param response_bytes: The response state | ||
| 2113 | :type response_bytes: bytes | ||
| 2114 | """ | ||
| 2115 | self._response_bytes = response_bytes | ||
| 2116 | self._hexdigest = '0x' + response_bytes.hex() | ||
| 2117 | self._words = self.bytes_to_tokens(response_bytes) | ||
| 2118 | |||
| 2119 | def __bytes__(self) -> bytes: | ||
| 2120 | """bytes representation of the object""" | ||
| 2121 | return self._response_bytes | ||
| 2122 | |||
| 2123 | def __hash__(self) -> int: | ||
| 2124 | """Uses the hash value of _response_bytes""" | ||
| 2125 | return hash(self._response_bytes) | ||
| 2126 | |||
| 2127 | @property | ||
| 2128 | def hexdigest(self) -> str: | ||
| 2129 | """Hexdigest representation of the OTP response""" | ||
| 2130 | return self._hexdigest | ||
| 2131 | |||
| 2132 | @property | ||
| 2133 | def response_bytes(self) -> bytes: | ||
| 2134 | """response_bytes read-only property""" | ||
| 2135 | return self._response_bytes | ||
| 2136 | |||
| 2137 | @property | ||
| 2138 | def words(self) -> str: | ||
| 2139 | """Tokens representation of the OTP response""" | ||
| 2140 | return self._words | ||
| 2141 | |||
| 2142 | @staticmethod | ||
| 2143 | def bytes_to_tokens(hash_bytes: bytes) -> str: | ||
| 2144 | """ | ||
| 2145 | Returns a 6 words token from bytes as specified by RFC-2289. | ||
| 2146 | |||
| 2147 | :param hash_bytes: The input bytes | ||
| 2148 | :type hash_bytes: bytes | ||
| 2149 | |||
| 2150 | :return: 6 words tokens | ||
| 2151 | :rtype: str | ||
| 2152 | """ | ||
| 2153 | bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) | ||
| 2154 | bit_pair_sum = OTPGenerator.bit_pair_sum(bit_stream) | ||
| 2155 | tokens = [] | ||
| 2156 | tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)]) | ||
| 2157 | tokens.append(RFC1760_TOKENS[int(bit_stream[11:22], 2)]) | ||
| 2158 | tokens.append(RFC1760_TOKENS[int(bit_stream[22:33], 2)]) | ||
| 2159 | tokens.append(RFC1760_TOKENS[int(bit_stream[33:44], 2)]) | ||
| 2160 | tokens.append(RFC1760_TOKENS[int(bit_stream[44:55], 2)]) | ||
| 2161 | tokens.append( | ||
| 2162 | RFC1760_TOKENS[ | ||
| 2163 | int(bit_stream[55:64] + f'{bit_pair_sum:0>8b}'[-2:], 2) | ||
| 2164 | ] | ||
| 2165 | ) | ||
| 2166 | return ' '.join(tokens) | ||
| 2167 | |||
| 2168 | |||
| 2097 | class OTPGenerator: | 2169 | class OTPGenerator: |
| 2098 | """OTPGenerator class""" | 2170 | """OTPGenerator class""" |
| 2099 | 2171 | ||
| 2100 | def __init__( | 2172 | def __init__( |
| 2101 | self, password: bytes, seed: str = '', hash_algo=OTP_ALGO_MD5 | 2173 | self, |
| 2102 | ): | 2174 | password: bytes, |
| 2175 | seed: str = '', | ||
| 2176 | hash_algo: int | str = OTP_ALGO_MD5, | ||
| 2177 | ) -> None: | ||
| 2103 | """ | 2178 | """ |
| 2104 | Constructs an OTPGenerator object with a given password and seed. | 2179 | Constructs an OTPGenerator object with a given password and seed. |
| 2105 | 2180 | ||
| @@ -2121,11 +2196,14 @@ class OTPGenerator: | |||
| 2121 | self._hash_algo = self.validate_hash_algo(hash_algo) | 2196 | self._hash_algo = self.validate_hash_algo(hash_algo) |
| 2122 | if not isinstance(password, bytes): | 2197 | if not isinstance(password, bytes): |
| 2123 | raise OTPGeneratorError('Password must be a byte-string') | 2198 | raise OTPGeneratorError('Password must be a byte-string') |
| 2124 | if len(password) < 10: | 2199 | if len(password) < OTP2289_MIN_PASSWORD_LENGTH: |
| 2125 | raise OTPGeneratorError('Password must be longer than 10 bytes') | 2200 | raise OTPGeneratorError( |
| 2201 | f'Password must be longer than {OTP2289_MIN_PASSWORD_LENGTH} ' | ||
| 2202 | 'bytes' | ||
| 2203 | ) | ||
| 2126 | self._password = password | 2204 | self._password = password |
| 2127 | 2205 | ||
| 2128 | def __repr__(self): | 2206 | def __repr__(self) -> str: |
| 2129 | """repr implementation""" | 2207 | """repr implementation""" |
| 2130 | return ( | 2208 | return ( |
| 2131 | f'{self.__class__} at {id(self)} (seed={self._seed}, ' | 2209 | f'{self.__class__} at {id(self)} (seed={self._seed}, ' |
| @@ -2145,41 +2223,17 @@ class OTPGenerator: | |||
| 2145 | """ | 2223 | """ |
| 2146 | if not isinstance(bit_stream, str): | 2224 | if not isinstance(bit_stream, str): |
| 2147 | raise OTPGeneratorError('bit_stream must be of type str') | 2225 | raise OTPGeneratorError('bit_stream must be of type str') |
| 2148 | if len(bit_stream) != 64: | 2226 | if len(bit_stream) != OTP2289_BITSTREAM_SIZE: |
| 2149 | raise OTPGeneratorError('bit_stream must be of size 64') | 2227 | raise OTPGeneratorError( |
| 2228 | f'bit_stream must be of size {OTP2289_BITSTREAM_SIZE}' | ||
| 2229 | ) | ||
| 2150 | value = 0 | 2230 | value = 0 |
| 2151 | for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True): | 2231 | for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True): |
| 2152 | value += int(''.join(pair), 2) | 2232 | value += int(''.join(pair), 2) |
| 2153 | return value | 2233 | return value |
| 2154 | 2234 | ||
| 2155 | @staticmethod | 2235 | @staticmethod |
| 2156 | def bytes_to_tokens(hash_bytes: bytes) -> str: | 2236 | def get_tokens_from_challenge(challenge: str) -> tuple[str, str, int]: |
| 2157 | """ | ||
| 2158 | Returns a 6 words token from bytes as specified by RFC-2289. | ||
| 2159 | |||
| 2160 | :param hash_bytes: The input bytes | ||
| 2161 | :type hash_bytes: bytes | ||
| 2162 | |||
| 2163 | :return: 6 words tokens | ||
| 2164 | :rtype: str | ||
| 2165 | """ | ||
| 2166 | bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) | ||
| 2167 | bit_pair_sum = OTPGenerator.bit_pair_sum(bit_stream) | ||
| 2168 | tokens = [] | ||
| 2169 | tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)]) | ||
| 2170 | tokens.append(RFC1760_TOKENS[int(bit_stream[11:22], 2)]) | ||
| 2171 | tokens.append(RFC1760_TOKENS[int(bit_stream[22:33], 2)]) | ||
| 2172 | tokens.append(RFC1760_TOKENS[int(bit_stream[33:44], 2)]) | ||
| 2173 | tokens.append(RFC1760_TOKENS[int(bit_stream[44:55], 2)]) | ||
| 2174 | tokens.append( | ||
| 2175 | RFC1760_TOKENS[ | ||
| 2176 | int(bit_stream[55:64] + f'{bit_pair_sum:0>8b}'[-2:], 2) | ||
| 2177 | ] | ||
| 2178 | ) | ||
| 2179 | return ' '.join(tokens) | ||
| 2180 | |||
| 2181 | @staticmethod | ||
| 2182 | def get_tokens_from_challenge(challenge: str) -> tuple: | ||
| 2183 | """ | 2237 | """ |
| 2184 | Returns tokens (seed, hash_algo and step) from a challenge string. | 2238 | Returns tokens (seed, hash_algo and step) from a challenge string. |
| 2185 | 2239 | ||
| @@ -2218,9 +2272,10 @@ class OTPGenerator: | |||
| 2218 | """ | 2272 | """ |
| 2219 | if not isinstance(sha1_digest, bytes): | 2273 | if not isinstance(sha1_digest, bytes): |
| 2220 | raise OTPGeneratorError('sha1_digest must be of type bytes') | 2274 | raise OTPGeneratorError('sha1_digest must be of type bytes') |
| 2221 | if len(sha1_digest) != 20: | 2275 | if len(sha1_digest) != OTP2289_SHA1_DIGEST_SIZE: |
| 2222 | raise OTPGeneratorError( | 2276 | raise OTPGeneratorError( |
| 2223 | 'sha1_digest must be 160 bits (20 bytes) long' | 2277 | f'sha1_digest must be {OTP2289_SHA1_DIGEST_SIZE * 2} bits ' |
| 2278 | f'({OTP2289_SHA1_DIGEST_SIZE} bytes) long' | ||
| 2224 | ) | 2279 | ) |
| 2225 | digested = list(5 * b'i') # 5 bytes (40 bits) | 2280 | digested = list(5 * b'i') # 5 bytes (40 bits) |
| 2226 | result = list(8 * b'x') # 8 bytes (64 bits) | 2281 | result = list(8 * b'x') # 8 bytes (64 bits) |
| @@ -2295,8 +2350,11 @@ class OTPGenerator: | |||
| 2295 | if not isinstance(tokens_str, str): | 2350 | if not isinstance(tokens_str, str): |
| 2296 | raise OTPGeneratorError('tokens must be a str') | 2351 | raise OTPGeneratorError('tokens must be a str') |
| 2297 | tokens = tokens_str.split() | 2352 | tokens = tokens_str.split() |
| 2298 | if len(tokens) != 6: | 2353 | if len(tokens) != OTP2289_TOKENS_COUNT: |
| 2299 | raise OTPGeneratorError('Tokens-string does not contain 6 tokens') | 2354 | raise OTPGeneratorError( |
| 2355 | f'Tokens-string does not contain {OTP2289_SHA1_DIGEST_SIZE} ' | ||
| 2356 | 'tokens' | ||
| 2357 | ) | ||
| 2300 | token_ints = [] | 2358 | token_ints = [] |
| 2301 | try: | 2359 | try: |
| 2302 | token_ints = [ | 2360 | token_ints = [ |
| @@ -2325,7 +2383,7 @@ class OTPGenerator: | |||
| 2325 | return int(bit_stream[:64], 2).to_bytes(8, 'big') | 2383 | return int(bit_stream[:64], 2).to_bytes(8, 'big') |
| 2326 | 2384 | ||
| 2327 | @staticmethod | 2385 | @staticmethod |
| 2328 | def validate_hash_algo(hash_algo) -> str: | 2386 | def validate_hash_algo(hash_algo: int | str) -> str: |
| 2329 | """ | 2387 | """ |
| 2330 | Validates the provided hash-algorithm. | 2388 | Validates the provided hash-algorithm. |
| 2331 | 2389 | ||
| @@ -2342,7 +2400,7 @@ class OTPGenerator: | |||
| 2342 | raise OTPGeneratorError( | 2400 | raise OTPGeneratorError( |
| 2343 | 'hash_algo is not among the known algorithms' | 2401 | 'hash_algo is not among the known algorithms' |
| 2344 | ) | 2402 | ) |
| 2345 | hash_algo = _ALGO_DICT.get(hash_algo) | 2403 | hash_algo = _ALGO_DICT[hash_algo] |
| 2346 | if not isinstance(hash_algo, str): | 2404 | if not isinstance(hash_algo, str): |
| 2347 | raise OTPGeneratorError('hash_algo must be an int or a str') | 2405 | raise OTPGeneratorError('hash_algo must be an int or a str') |
| 2348 | if hash_algo not in hashlib.algorithms_available: | 2406 | if hash_algo not in hashlib.algorithms_available: |
| @@ -2367,9 +2425,10 @@ class OTPGenerator: | |||
| 2367 | """ | 2425 | """ |
| 2368 | if not isinstance(seed, str): | 2426 | if not isinstance(seed, str): |
| 2369 | raise OTPGeneratorError('Seed must be a string') | 2427 | raise OTPGeneratorError('Seed must be a string') |
| 2370 | if not seed or len(seed) > 16: | 2428 | if not seed or len(seed) > OTP2289_MAX_SEED_LENGTH: |
| 2371 | raise OTPGeneratorError( | 2429 | raise OTPGeneratorError( |
| 2372 | 'The seed MUST be of 1 to 16 characters in length' | 2430 | f'The seed MUST be of 1 to {OTP2289_MAX_SEED_LENGTH} ' |
| 2431 | 'characters in length' | ||
| 2373 | ) | 2432 | ) |
| 2374 | for char in seed: | 2433 | for char in seed: |
| 2375 | if char not in string.ascii_letters + string.digits: | 2434 | if char not in string.ascii_letters + string.digits: |
| @@ -2407,7 +2466,8 @@ class OTPGenerator: | |||
| 2407 | :return: Hexdigest for the given step | 2466 | :return: Hexdigest for the given step |
| 2408 | :rtype: str | 2467 | :rtype: str |
| 2409 | """ | 2468 | """ |
| 2410 | return '0x' + binascii.hexlify(self._generate_otp_bytes(step)).decode() | 2469 | response = OTPResponse(self._generate_otp_bytes(step)) |
| 2470 | return response.hexdigest | ||
| 2411 | 2471 | ||
| 2412 | def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: | 2472 | def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: |
| 2413 | """ | 2473 | """ |
| @@ -2440,7 +2500,8 @@ class OTPGenerator: | |||
| 2440 | :return: Six words (separated by single space) token for the given step | 2500 | :return: Six words (separated by single space) token for the given step |
| 2441 | :rtype: str | 2501 | :rtype: str |
| 2442 | """ | 2502 | """ |
| 2443 | return self.bytes_to_tokens(self._generate_otp_bytes(step)) | 2503 | response = OTPResponse(self._generate_otp_bytes(step)) |
| 2504 | return response.words | ||
| 2444 | 2505 | ||
| 2445 | def generate_otp_words_from_challenge(self, challenge: str) -> str: | 2506 | def generate_otp_words_from_challenge(self, challenge: str) -> str: |
| 2446 | """ | 2507 | """ |
| @@ -2463,7 +2524,9 @@ class OTPGenerator: | |||
| 2463 | self._hash_algo = self.validate_hash_algo(hash_algo) | 2524 | self._hash_algo = self.validate_hash_algo(hash_algo) |
| 2464 | return self.generate_otp_words(step) | 2525 | return self.generate_otp_words(step) |
| 2465 | 2526 | ||
| 2466 | def hexdigest_range(self, start: int = 499, stop: int = 0): | 2527 | def hexdigest_range( |
| 2528 | self, start: int = 499, stop: int = 0 | ||
| 2529 | ) -> Iterator[str]: | ||
| 2467 | """ | 2530 | """ |
| 2468 | Returns an iterator that providing hexdigests corresponding to steps | 2531 | Returns an iterator that providing hexdigests corresponding to steps |
| 2469 | from `start` to and including `stop`. | 2532 | from `start` to and including `stop`. |
| @@ -2484,7 +2547,7 @@ class OTPGenerator: | |||
| 2484 | for step in range(start, stop - 1, -1): | 2547 | for step in range(start, stop - 1, -1): |
| 2485 | yield self.generate_otp_hexdigest(step) | 2548 | yield self.generate_otp_hexdigest(step) |
| 2486 | 2549 | ||
| 2487 | def words_range(self, start: int = 499, stop: int = 0): | 2550 | def words_range(self, start: int = 499, stop: int = 0) -> Iterator[str]: |
| 2488 | """ | 2551 | """ |
| 2489 | Returns an iterator that providing the words corresponding to steps | 2552 | Returns an iterator that providing the words corresponding to steps |
| 2490 | from `start` to and including `stop`. | 2553 | 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 @@ | |||
| 1 | # SPDX-License-Identifier: BSD-2-Clause-FreeBSD | 1 | # SPDX-License-Identifier: BSD-2-Clause |
| 2 | # | 2 | # |
| 3 | # Copyright (c) 2020-2025 Simeon Simeonov | 3 | # Copyright (c) 2020-2026 Simeon Simeonov |
| 4 | # All rights reserved. | 4 | # All rights reserved. |
| 5 | # | 5 | # |
| 6 | # Redistribution and use in source and binary forms, with or without | 6 | # Redistribution and use in source and binary forms, with or without |
| @@ -24,11 +24,18 @@ | |||
| 24 | # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. | 24 | # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| 25 | """A pure Python implementation of the RFC-2289 OTP server""" | 25 | """A pure Python implementation of the RFC-2289 OTP server""" |
| 26 | 26 | ||
| 27 | import binascii | 27 | from __future__ import annotations |
| 28 | |||
| 28 | import hashlib | 29 | import hashlib |
| 30 | import typing | ||
| 31 | |||
| 32 | if typing.TYPE_CHECKING: | ||
| 33 | from collections.abc import Iterator | ||
| 29 | 34 | ||
| 30 | from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorError | 35 | from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorError |
| 31 | 36 | ||
| 37 | OTP2289_HEX_DIGEST_SIZE: typing.Final[int] = 16 | ||
| 38 | |||
| 32 | 39 | ||
| 33 | class OTPStateError(Exception): | 40 | class OTPStateError(Exception): |
| 34 | """OTPStateError class""" | 41 | """OTPStateError class""" |
| @@ -52,8 +59,12 @@ class OTPState: | |||
| 52 | """ | 59 | """ |
| 53 | 60 | ||
| 54 | def __init__( | 61 | def __init__( |
| 55 | self, ot_hex: str, current_step: int, seed: str, hash_algo=OTP_ALGO_MD5 | 62 | self, |
| 56 | ): | 63 | ot_hex: str | None, |
| 64 | current_step: int, | ||
| 65 | seed: str, | ||
| 66 | hash_algo: int | str = OTP_ALGO_MD5, | ||
| 67 | ) -> None: | ||
| 57 | """ | 68 | """ |
| 58 | Constructs an OTPState object with the given arguments. | 69 | Constructs an OTPState object with the given arguments. |
| 59 | 70 | ||
| @@ -79,13 +90,14 @@ class OTPState: | |||
| 79 | self._hash_algo = OTPGenerator.validate_hash_algo(hash_algo) | 90 | self._hash_algo = OTPGenerator.validate_hash_algo(hash_algo) |
| 80 | self._step = OTPGenerator.validate_step(current_step) | 91 | self._step = OTPGenerator.validate_step(current_step) |
| 81 | except OTPGeneratorError as exp: | 92 | except OTPGeneratorError as exp: |
| 82 | raise OTPStateError(exp.args[0]) from None | 93 | raise OTPStateError(exp.args[0]) from exp |
| 94 | |||
| 83 | self._current_digest = None | 95 | self._current_digest = None |
| 84 | if ot_hex is not None: | 96 | if ot_hex is not None: |
| 85 | self._current_digest = self.validate_hex(ot_hex) | 97 | self._current_digest = self.validate_hex(ot_hex) |
| 86 | self._new_digest_hex = None # set upon a successful validation | 98 | self._new_digest_hex = None # set upon a successful validation |
| 87 | 99 | ||
| 88 | def __repr__(self): | 100 | def __repr__(self) -> str: |
| 89 | """repr implementation""" | 101 | """repr implementation""" |
| 90 | return ( | 102 | return ( |
| 91 | f'{self.__class__} at {id(self)} ' | 103 | f'{self.__class__} at {id(self)} ' |
| @@ -102,7 +114,7 @@ class OTPState: | |||
| 102 | return f'otp-{self._hash_algo} {self._step} {self._seed} ' | 114 | return f'otp-{self._hash_algo} {self._step} {self._seed} ' |
| 103 | 115 | ||
| 104 | @property | 116 | @property |
| 105 | def current_digest(self) -> bytes: | 117 | def current_digest(self) -> bytes | None: |
| 106 | """current_digest-property""" | 118 | """current_digest-property""" |
| 107 | return self._current_digest | 119 | return self._current_digest |
| 108 | 120 | ||
| @@ -116,7 +128,7 @@ class OTPState: | |||
| 116 | """ot_hex-property""" | 128 | """ot_hex-property""" |
| 117 | if self._current_digest is None: | 129 | if self._current_digest is None: |
| 118 | return '' | 130 | return '' |
| 119 | return binascii.hexlify(self._current_digest).decode() | 131 | return self._current_digest.hex() |
| 120 | 132 | ||
| 121 | @property | 133 | @property |
| 122 | def seed(self) -> str: | 134 | def seed(self) -> str: |
| @@ -134,7 +146,7 @@ class OTPState: | |||
| 134 | return bool(self._new_digest_hex) | 146 | return bool(self._new_digest_hex) |
| 135 | 147 | ||
| 136 | @classmethod | 148 | @classmethod |
| 137 | def from_dict(cls, dict_obj: dict): | 149 | def from_dict(cls, dict_obj: dict) -> OTPState: |
| 138 | """ | 150 | """ |
| 139 | Returns an OTPState object from the dict-object | 151 | Returns an OTPState object from the dict-object |
| 140 | 152 | ||
| @@ -196,17 +208,17 @@ class OTPState: | |||
| 196 | if ot_hex.startswith('0x'): | 208 | if ot_hex.startswith('0x'): |
| 197 | ot_hex = ot_hex[2:] | 209 | ot_hex = ot_hex[2:] |
| 198 | ot_hex = ot_hex.strip().lower() | 210 | ot_hex = ot_hex.strip().lower() |
| 199 | if len(ot_hex) != 16: | 211 | if len(ot_hex) != OTP2289_HEX_DIGEST_SIZE: |
| 200 | raise OTPStateError( | 212 | raise OTPStateError( |
| 201 | 'The length of the hex should be 16 ' | 213 | f'The length of the hex should be {OTP2289_HEX_DIGEST_SIZE} ' |
| 202 | '(representing 64 bits digest)' | 214 | '(representing 64 bits digest)' |
| 203 | ) | 215 | ) |
| 204 | try: | 216 | try: |
| 205 | return binascii.unhexlify(ot_hex) | 217 | return bytes.fromhex(ot_hex) |
| 206 | except binascii.Error: | 218 | except ValueError: |
| 207 | raise OTPStateError('Invalid OT-hex') from None | 219 | raise OTPStateError('Invalid OT-hex') from None |
| 208 | 220 | ||
| 209 | def get_next_state(self): | 221 | def get_next_state(self) -> OTPState | None: |
| 210 | """ | 222 | """ |
| 211 | Returns the next state for a validated OTPState. | 223 | Returns the next state for a validated OTPState. |
| 212 | 224 | ||
| @@ -223,7 +235,7 @@ class OTPState: | |||
| 223 | ) | 235 | ) |
| 224 | 236 | ||
| 225 | def response_validates( | 237 | def response_validates( |
| 226 | self, response: str, store_valid_response: str = True | 238 | self, response: str, *, store_valid_response: bool = True |
| 227 | ) -> bool: | 239 | ) -> bool: |
| 228 | """ | 240 | """ |
| 229 | Validates the incoming response as specified by RFC-2289. | 241 | Validates the incoming response as specified by RFC-2289. |
| @@ -251,9 +263,7 @@ class OTPState: | |||
| 251 | == self._current_digest | 263 | == self._current_digest |
| 252 | ): | 264 | ): |
| 253 | if store_valid_response: | 265 | if store_valid_response: |
| 254 | self._new_digest_hex = binascii.hexlify( | 266 | self._new_digest_hex = response_bytes.hex() |
| 255 | response_bytes | ||
| 256 | ).decode() | ||
| 257 | return True | 267 | return True |
| 258 | return False | 268 | return False |
| 259 | if self._hash_algo == 'sha1': | 269 | if self._hash_algo == 'sha1': |
| @@ -266,9 +276,7 @@ class OTPState: | |||
| 266 | == self._current_digest | 276 | == self._current_digest |
| 267 | ): | 277 | ): |
| 268 | if store_valid_response: | 278 | if store_valid_response: |
| 269 | self._new_digest_hex = binascii.hexlify( | 279 | self._new_digest_hex = response_bytes.hex() |
| 270 | response_bytes | ||
| 271 | ).decode() | ||
| 272 | return True | 280 | return True |
| 273 | return False | 281 | return False |
| 274 | # this should not happen since the hash_algo is validated by the caller | 282 | # this should not happen since the hash_algo is validated by the caller |
| @@ -283,9 +291,11 @@ class OTPState: | |||
| 283 | :return: The dict representation of the object | 291 | :return: The dict representation of the object |
| 284 | :rtype: dict | 292 | :rtype: dict |
| 285 | """ | 293 | """ |
| 286 | ot_hex = self._current_digest | 294 | ot_hex = ( |
| 287 | if ot_hex is not None: | 295 | self._current_digest.hex() |
| 288 | ot_hex = binascii.hexlify(self._current_digest).decode() | 296 | if self._current_digest is not None |
| 297 | else None | ||
| 298 | ) | ||
| 289 | return { | 299 | return { |
| 290 | 'ot_hex': ot_hex, | 300 | 'ot_hex': ot_hex, |
| 291 | 'current_step': self._step, | 301 | 'current_step': self._step, |
| @@ -304,27 +314,27 @@ class OTPStore: | |||
| 304 | The class could serve as a base class when implementing store backends. | 314 | The class could serve as a base class when implementing store backends. |
| 305 | """ | 315 | """ |
| 306 | 316 | ||
| 307 | def __init__(self, data=None): | 317 | def __init__(self, data: dict | None = None) -> None: |
| 308 | """ | 318 | """ |
| 309 | Constructs an OTPStore object from data | 319 | Constructs an OTPStore object from data |
| 310 | 320 | ||
| 311 | :param data: The data object, defaults to None | 321 | :param data: The data dict, defaults to None |
| 312 | :type data: object or None | 322 | :type data: dict or None |
| 313 | """ | 323 | """ |
| 314 | self._data = {} # {key1: {state1-data...}, key2: {state2-data...}} | 324 | self._data = {} # {key1: {state1-data...}, key2: {state2-data...}} |
| 315 | self._states = {} # OTPState: (domain, key) - dict | 325 | self._states = {} # OTPState: (domain, key) - dict |
| 316 | if data is not None: | 326 | if data is not None: |
| 317 | self._add_data(data) | 327 | self._add_data(data) |
| 318 | 328 | ||
| 319 | def __contains__(self, state): | 329 | def __contains__(self, state: OTPState) -> bool: |
| 320 | """membership test""" | 330 | """membership test""" |
| 321 | return state in self._states | 331 | return state in self._states |
| 322 | 332 | ||
| 323 | def __iter__(self): | 333 | def __iter__(self) -> Iterator: |
| 324 | """iterator for OTPStore""" | 334 | """iterator for OTPStore""" |
| 325 | return iter(self._data) | 335 | return iter(self._data) |
| 326 | 336 | ||
| 327 | def __len__(self): | 337 | def __len__(self) -> int: |
| 328 | """len() implementation""" | 338 | """len() implementation""" |
| 329 | return len(self._data) | 339 | return len(self._data) |
| 330 | 340 | ||
| @@ -348,7 +358,7 @@ class OTPStore: | |||
| 348 | """ | 358 | """ |
| 349 | return self._states | 359 | return self._states |
| 350 | 360 | ||
| 351 | def add_state(self, key: str, state: OTPState): | 361 | def add_state(self, key: str, state: OTPState) -> None: |
| 352 | """ | 362 | """ |
| 353 | Adds an OTPState object with a given key. | 363 | Adds an OTPState object with a given key. |
| 354 | 364 | ||
| @@ -367,11 +377,13 @@ class OTPStore: | |||
| 367 | self._data[key] = state | 377 | self._data[key] = state |
| 368 | self._states[state] = key | 378 | self._states[state] = key |
| 369 | 379 | ||
| 370 | def get(self, key, default=None): | 380 | def get( |
| 381 | self, key: str, default: OTPState | None = None | ||
| 382 | ) -> OTPState | None: | ||
| 371 | """A wrapper for dict.get""" | 383 | """A wrapper for dict.get""" |
| 372 | return self._data.get(key, default) | 384 | return self._data.get(key, default) |
| 373 | 385 | ||
| 374 | def items(self): | 386 | def items(self) -> typing.ItemsView: |
| 375 | """A wrapper for dict.items""" | 387 | """A wrapper for dict.items""" |
| 376 | return self._data.items() | 388 | return self._data.items() |
| 377 | 389 | ||
| @@ -396,7 +408,7 @@ class OTPStore: | |||
| 396 | return state | 408 | return state |
| 397 | 409 | ||
| 398 | def response_validates( | 410 | def response_validates( |
| 399 | self, key: str, response: str, store_valid_response: bool = True | 411 | self, key: str, response: str, *, store_valid_response: bool = True |
| 400 | ) -> bool: | 412 | ) -> bool: |
| 401 | """ | 413 | """ |
| 402 | A method that wraps around OTPState.response_validates and | 414 | A method that wraps around OTPState.response_validates and |
| @@ -424,7 +436,9 @@ class OTPStore: | |||
| 424 | :rtype: bool | 436 | :rtype: bool |
| 425 | """ | 437 | """ |
| 426 | state = self._data[key] | 438 | state = self._data[key] |
| 427 | rvalue = state.response_validates(response, store_valid_response) | 439 | rvalue = state.response_validates( |
| 440 | response, store_valid_response=store_valid_response | ||
| 441 | ) | ||
| 428 | if rvalue and store_valid_response: | 442 | if rvalue and store_valid_response: |
| 429 | next_state = state.get_next_state() | 443 | next_state = state.get_next_state() |
| 430 | self._data[key] = next_state | 444 | self._data[key] = next_state |
| @@ -443,7 +457,7 @@ class OTPStore: | |||
| 443 | """ | 457 | """ |
| 444 | return {key: state.to_dict() for key, state in self._data.items()} | 458 | return {key: state.to_dict() for key, state in self._data.items()} |
| 445 | 459 | ||
| 446 | def _add_data(self, dict_obj: dict) -> dict: | 460 | def _add_data(self, dict_obj: dict) -> None: |
| 447 | """ | 461 | """ |
| 448 | Adds data from a dict object (dict_obj). | 462 | Adds data from a dict object (dict_obj). |
| 449 | 463 | ||
