diff options
Diffstat (limited to 'src/otp2289/generator.py')
| -rw-r--r-- | src/otp2289/generator.py | 165 |
1 files changed, 114 insertions, 51 deletions
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`. |
