summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/otp2289/__init__.py10
-rw-r--r--src/otp2289/__main__.py22
-rw-r--r--src/otp2289/generator.py165
-rw-r--r--src/otp2289/server.py88
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)
34from .server import ( 35from .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
47def int_or_str(value): 48def 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
36import errno 36import errno
37import getpass 37import getpass
38import os 38import os
39import pathlib
39import secrets 40import secrets
40import string 41import string
41import sys 42import sys
@@ -43,9 +44,11 @@ import sys
43import otp2289 44import otp2289
44 45
45 46
46def eprint(*arg, **kwargs): 47def 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
51def generate_otp_response(args: argparse.Namespace) -> str: 54def 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
216def main(args=None): 219def 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
27import binascii
28import hashlib 27import hashlib
29import string 28import string
29import typing
30from collections.abc import Iterator
30 31
31OTP_ALGO_MD5 = 1 32OTP_ALGO_MD5: typing.Final[int] = 1
32OTP_ALGO_SHA1 = 2 33OTP_ALGO_SHA1: typing.Final[int] = 2
34
35# useful constants
36OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64
37OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16
38OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10
39OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20
40OTP2289_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 #
35RFC1760_TOKENS = [ 43RFC1760_TOKENS = [
@@ -2094,12 +2102,79 @@ class OTPChallengeError(Exception):
2094 """OTPChallengeError class""" 2102 """OTPChallengeError class"""
2095 2103
2096 2104
2105class 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
2097class OTPGenerator: 2169class 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
27import binascii 27from __future__ import annotations
28
28import hashlib 29import hashlib
30import typing
31
32if typing.TYPE_CHECKING:
33 from collections.abc import Iterator
29 34
30from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorError 35from .generator import OTP_ALGO_MD5, OTPGenerator, OTPGeneratorError
31 36
37OTP2289_HEX_DIGEST_SIZE: typing.Final[int] = 16
38
32 39
33class OTPStateError(Exception): 40class 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