summaryrefslogtreecommitdiff
path: root/src/otp2289/generator.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/otp2289/generator.py')
-rw-r--r--src/otp2289/generator.py314
1 files changed, 171 insertions, 143 deletions
diff --git a/src/otp2289/generator.py b/src/otp2289/generator.py
index 32123b4..17024a2 100644
--- a/src/otp2289/generator.py
+++ b/src/otp2289/generator.py
@@ -24,16 +24,21 @@
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
27from __future__ import annotations
28
27import hashlib 29import hashlib
28import string 30import string
29import typing 31import typing
30from collections.abc import Iterator 32
33if typing.TYPE_CHECKING:
34 from collections.abc import Iterator
31 35
32OTP_ALGO_MD5: typing.Final[int] = 1 36OTP_ALGO_MD5: typing.Final[int] = 1
33OTP_ALGO_SHA1: typing.Final[int] = 2 37OTP_ALGO_SHA1: typing.Final[int] = 2
34 38
35# useful constants 39# useful constants
36OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64 40OTP2289_BITSTREAM_SIZE: typing.Final[int] = 64
41OTP2289_HEX_DIGEST_SIZE: typing.Final[int] = 16
37OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16 42OTP2289_MAX_SEED_LENGTH: typing.Final[int] = 16
38OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10 43OTP2289_MIN_PASSWORD_LENGTH: typing.Final[int] = 10
39OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20 44OTP2289_SHA1_DIGEST_SIZE: typing.Final[int] = 20
@@ -2094,12 +2099,16 @@ RFC1760_TOKENS = [
2094_ALGO_DICT = {OTP_ALGO_MD5: 'md5', OTP_ALGO_SHA1: 'sha1'} 2099_ALGO_DICT = {OTP_ALGO_MD5: 'md5', OTP_ALGO_SHA1: 'sha1'}
2095 2100
2096 2101
2102class OTPChallengeError(Exception):
2103 """OTPChallengeError class"""
2104
2105
2097class OTPGeneratorError(Exception): 2106class OTPGeneratorError(Exception):
2098 """OTPGeneratorError class""" 2107 """OTPGeneratorError class"""
2099 2108
2100 2109
2101class OTPChallengeError(Exception): 2110class OTPResponseError(Exception):
2102 """OTPChallengeError class""" 2111 """OTPResponseError class"""
2103 2112
2104 2113
2105class OTPResponse: 2114class OTPResponse:
@@ -2120,10 +2129,24 @@ class OTPResponse:
2120 """bytes representation of the object""" 2129 """bytes representation of the object"""
2121 return self._response_bytes 2130 return self._response_bytes
2122 2131
2132 def __eq__(self, value: object, /) -> bool:
2133 """definition for type equality"""
2134 if not isinstance(value, OTPResponse):
2135 return False
2136
2137 return self._response_bytes == value.response_bytes
2138
2123 def __hash__(self) -> int: 2139 def __hash__(self) -> int:
2124 """Uses the hash value of _response_bytes""" 2140 """Uses the hash value of _response_bytes"""
2125 return hash(self._response_bytes) 2141 return hash(self._response_bytes)
2126 2142
2143 def __repr__(self) -> str:
2144 """repr implementation"""
2145 return (
2146 f'{self.__class__} at {id(self)} '
2147 f'(response_bytes={self._response_bytes!r})'
2148 )
2149
2127 @property 2150 @property
2128 def hexdigest(self) -> str: 2151 def hexdigest(self) -> str:
2129 """Hexdigest representation of the OTP response""" 2152 """Hexdigest representation of the OTP response"""
@@ -2139,6 +2162,59 @@ class OTPResponse:
2139 """Tokens representation of the OTP response""" 2162 """Tokens representation of the OTP response"""
2140 return self._words 2163 return self._words
2141 2164
2165 @classmethod
2166 def from_hex(cls, ot_hex: str) -> OTPResponse:
2167 """
2168 Generates instance from the provided hexidigest with or without
2169 leading 0x as specified by RFC-2289.
2170
2171 :param ot_hex: The one-time hex to validate
2172 :type ot_hex: str
2173
2174 :raises otp2289.OTPResponseError: When the ot_hex is invalid
2175
2176 :return: A new OTPResponse object
2177 :rtype: otp2289.OTPResponse
2178 """
2179 return cls(OTPResponse.hex_to_bytes(ot_hex))
2180
2181 @classmethod
2182 def from_tokens(cls, tokens_str: str) -> OTPResponse:
2183 """
2184 Generates instance from a 6 words token as specified by RFC-2289.
2185
2186 :param tokens_str: String representing 6 words tokens
2187 :type tokens_str: str
2188
2189 :raises otp2289.OTPResponseError: When the tokens_str is invalid
2190
2191 :return: A new OTPResponse object
2192 :rtype: otp2289.OTPResponse
2193 """
2194 return cls(OTPResponse.tokens_to_bytes(tokens_str))
2195
2196 @staticmethod
2197 def bit_pair_sum(bit_stream: str) -> int:
2198 """
2199 Split bit_stream in bit-pairs and sum them all together.
2200
2201 :param bit_stream: The bit-stream object
2202 :type bit_stream: str
2203
2204 :return: The sum of all bit-pairs in bit_stream
2205 :rtype: int
2206 """
2207 if not isinstance(bit_stream, str):
2208 raise OTPResponseError('bit_stream must be of type str')
2209 if len(bit_stream) != OTP2289_BITSTREAM_SIZE:
2210 raise OTPResponseError(
2211 f'bit_stream must be of size {OTP2289_BITSTREAM_SIZE}'
2212 )
2213 value = 0
2214 for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True):
2215 value += int(''.join(pair), 2)
2216 return value
2217
2142 @staticmethod 2218 @staticmethod
2143 def bytes_to_tokens(hash_bytes: bytes) -> str: 2219 def bytes_to_tokens(hash_bytes: bytes) -> str:
2144 """ 2220 """
@@ -2151,7 +2227,7 @@ class OTPResponse:
2151 :rtype: str 2227 :rtype: str
2152 """ 2228 """
2153 bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes]) 2229 bit_stream = ''.join([f'{byte:0>8b}' for byte in hash_bytes])
2154 bit_pair_sum = OTPGenerator.bit_pair_sum(bit_stream) 2230 bit_pair_sum = OTPResponse.bit_pair_sum(bit_stream)
2155 tokens = [] 2231 tokens = []
2156 tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)]) 2232 tokens.append(RFC1760_TOKENS[int(bit_stream[:11], 2)])
2157 tokens.append(RFC1760_TOKENS[int(bit_stream[11:22], 2)]) 2233 tokens.append(RFC1760_TOKENS[int(bit_stream[11:22], 2)])
@@ -2165,6 +2241,81 @@ class OTPResponse:
2165 ) 2241 )
2166 return ' '.join(tokens) 2242 return ' '.join(tokens)
2167 2243
2244 @staticmethod
2245 def hex_to_bytes(ot_hex: str) -> bytes:
2246 """
2247 Returns bytes from the provided hexidigest.
2248
2249 :param ot_hex: The one-time hex to validate
2250 :type ot_hex: str
2251
2252 :raises otp2289.OTPResponseError: If hex does not validate
2253
2254 :return: The validated hex (without leading 0x) converted to bytes
2255 :rtype: bytes
2256 """
2257 if not isinstance(ot_hex, str):
2258 raise OTPResponseError('OT-hex must be a str')
2259 if ot_hex.startswith('0x'):
2260 ot_hex = ot_hex[2:]
2261 ot_hex = ot_hex.strip().lower()
2262 if len(ot_hex) != OTP2289_HEX_DIGEST_SIZE:
2263 raise OTPResponseError(
2264 f'The length of the hex should be {OTP2289_HEX_DIGEST_SIZE} '
2265 '(representing 64 bits digest)'
2266 )
2267 try:
2268 return bytes.fromhex(ot_hex)
2269 except ValueError:
2270 raise OTPResponseError('Invalid OT-hex') from None
2271
2272 @staticmethod
2273 def tokens_to_bytes(tokens_str: str) -> bytes:
2274 """
2275 Returns bytes from a 6 words token as specified by RFC-2289.
2276
2277 :param tokens_str: String representing 6 words tokens
2278 :type tokens_str: str
2279
2280 :raises otp2289.OTPResponseError: When the tokens_str is invalid
2281
2282 :return: 6 words tokens
2283 :rtype: bytes
2284 """
2285 if not isinstance(tokens_str, str):
2286 raise OTPResponseError('tokens must be a str')
2287 tokens = tokens_str.split()
2288 if len(tokens) != OTP2289_TOKENS_COUNT:
2289 raise OTPResponseError(
2290 f'Tokens-string does not contain {OTP2289_TOKENS_COUNT} tokens'
2291 )
2292 token_ints = []
2293 try:
2294 token_ints = [
2295 RFC1760_TOKENS.index(token.upper()) for token in tokens
2296 ]
2297 except ValueError:
2298 raise OTPResponseError(
2299 'One or more words not present in RFC1760'
2300 ) from None
2301 # now we build a string of bits
2302 bit_stream = format(token_ints[0], '011b')
2303 bit_stream += format(token_ints[1], '011b')
2304 bit_stream += format(token_ints[2], '011b')
2305 bit_stream += format(token_ints[3], '011b')
2306 bit_stream += format(token_ints[4], '011b')
2307 bit_stream += format(token_ints[5], '011b')
2308 # we have 66 bits: 64 digest + 2 bit pair sum (control number)
2309 # RFC-2289: All OTP generators MUST calculate this checksum and all
2310 # OTP servers MUST verify this checksum explicitly as part of the
2311 # operation of decoding this representation of the one-time password.
2312 if (
2313 f'{OTPResponse.bit_pair_sum(bit_stream[:64]):0>8b}'[-2:]
2314 != bit_stream[-2:]
2315 ):
2316 raise OTPResponseError('Invalid bit checksum')
2317 return int(bit_stream[:64], 2).to_bytes(8, 'big')
2318
2168 2319
2169class OTPGenerator: 2320class OTPGenerator:
2170 """OTPGenerator class""" 2321 """OTPGenerator class"""
@@ -2211,28 +2362,6 @@ class OTPGenerator:
2211 ) 2362 )
2212 2363
2213 @staticmethod 2364 @staticmethod
2214 def bit_pair_sum(bit_stream: str) -> int:
2215 """
2216 Split bit_stream in bit-pairs and sum them all together.
2217
2218 :param bit_stream: The bit-stream object
2219 :type bit_stream: str
2220
2221 :return: The sum of all bit-pairs in bit_stream
2222 :rtype: int
2223 """
2224 if not isinstance(bit_stream, str):
2225 raise OTPGeneratorError('bit_stream must be of type str')
2226 if len(bit_stream) != OTP2289_BITSTREAM_SIZE:
2227 raise OTPGeneratorError(
2228 f'bit_stream must be of size {OTP2289_BITSTREAM_SIZE}'
2229 )
2230 value = 0
2231 for pair in zip(bit_stream[::2], bit_stream[1::2], strict=True):
2232 value += int(''.join(pair), 2)
2233 return value
2234
2235 @staticmethod
2236 def get_tokens_from_challenge(challenge: str) -> tuple[str, str, int]: 2365 def get_tokens_from_challenge(challenge: str) -> tuple[str, str, int]:
2237 """ 2366 """
2238 Returns tokens (seed, hash_algo and step) from a challenge string. 2367 Returns tokens (seed, hash_algo and step) from a challenge string.
@@ -2335,54 +2464,6 @@ class OTPGenerator:
2335 return bytes([byte_str1[i] ^ byte_str2[i] for i in range(length)]) 2464 return bytes([byte_str1[i] ^ byte_str2[i] for i in range(length)])
2336 2465
2337 @staticmethod 2466 @staticmethod
2338 def tokens_to_bytes(tokens_str: str) -> bytes:
2339 """
2340 Returns bytes from a 6 words token as specified by RFC-2289.
2341
2342 :param tokens_str: String representing 6 words tokens
2343 :type tokens_str: str
2344
2345 :raises otp2289.OTPGeneratorError: When the tokens_str is invalid
2346
2347 :return: 6 words tokens
2348 :rtype: bytes
2349 """
2350 if not isinstance(tokens_str, str):
2351 raise OTPGeneratorError('tokens must be a str')
2352 tokens = tokens_str.split()
2353 if len(tokens) != OTP2289_TOKENS_COUNT:
2354 raise OTPGeneratorError(
2355 f'Tokens-string does not contain {OTP2289_SHA1_DIGEST_SIZE} '
2356 'tokens'
2357 )
2358 token_ints = []
2359 try:
2360 token_ints = [
2361 RFC1760_TOKENS.index(token.upper()) for token in tokens
2362 ]
2363 except ValueError:
2364 raise OTPGeneratorError(
2365 'One or more words not present in RFC1760'
2366 ) from None
2367 # now we build a string of bits
2368 bit_stream = format(token_ints[0], '011b')
2369 bit_stream += format(token_ints[1], '011b')
2370 bit_stream += format(token_ints[2], '011b')
2371 bit_stream += format(token_ints[3], '011b')
2372 bit_stream += format(token_ints[4], '011b')
2373 bit_stream += format(token_ints[5], '011b')
2374 # we have 66 bits: 64 digest + 2 bit pair sum (control number)
2375 # RFC-2289: All OTP generators MUST calculate this checksum and all
2376 # OTP servers MUST verify this checksum explicitly as part of the
2377 # operation of decoding this representation of the one-time password.
2378 if (
2379 f'{OTPGenerator.bit_pair_sum(bit_stream[:64]):0>8b}'[-2:]
2380 != bit_stream[-2:]
2381 ):
2382 raise OTPGeneratorError('Invalid bit checksum')
2383 return int(bit_stream[:64], 2).to_bytes(8, 'big')
2384
2385 @staticmethod
2386 def validate_hash_algo(hash_algo: int | str) -> str: 2467 def validate_hash_algo(hash_algo: int | str) -> str:
2387 """ 2468 """
2388 Validates the provided hash-algorithm. 2469 Validates the provided hash-algorithm.
@@ -2456,56 +2537,24 @@ class OTPGenerator:
2456 raise OTPGeneratorError('Step value MUST be >= 0') 2537 raise OTPGeneratorError('Step value MUST be >= 0')
2457 return step 2538 return step
2458 2539
2459 def generate_otp_hexdigest(self, step: int) -> str: 2540 def generate_otp_response(self, step: int) -> OTPResponse:
2460 """ 2541 """
2461 Generates the OTP hexdigest for the given step. 2542 Generates OTPResponse instance for the given step
2462 2543
2463 :param step: The step to generate OTP for 2544 :param step: The step to generate OTP for
2464 :type step: int 2545 :type step: int
2465 2546
2466 :return: Hexdigest for the given step 2547 :return: OTPResponse instance for the given step
2467 :rtype: str 2548 :rtype: OTPResponse
2468 """ 2549 """
2469 response = OTPResponse(self._generate_otp_bytes(step)) 2550 return OTPResponse(self._generate_otp_bytes(step))
2470 return response.hexdigest
2471 2551
2472 def generate_otp_hexdigest_from_challenge(self, challenge: str) -> str: 2552 def generate_otp_response_from_challenge(
2553 self, challenge: str
2554 ) -> OTPResponse:
2473 """ 2555 """
2474 Same as generate_otp_hexdigest, but it generates hex. from a challenge. 2556 Same as generate_otp_response,
2475 2557 but it generates OTPResponse from a challenge.
2476 RFC-2289 states:
2477 The challenge MUST be in a standard syntax so
2478 that automated generators can recognize the challenge in context and
2479 extract these parameters. The syntax of the challenge is:
2480 otp-<algorithm identifier> <sequence integer> <seed>
2481
2482 :param challenge: The challenge string
2483 :type challenge: str
2484
2485 :return: Hexdigest for the given challenge
2486 :rtype: str
2487 """
2488 seed, hash_algo, step = self.get_tokens_from_challenge(challenge)
2489 self._seed = self.validate_seed(seed)
2490 self._hash_algo = self.validate_hash_algo(hash_algo)
2491 return self.generate_otp_hexdigest(step)
2492
2493 def generate_otp_words(self, step: int) -> str:
2494 """
2495 Generates the OTP six words token for the given step.
2496
2497 :param step: The step to generate OTP for
2498 :type step: int
2499
2500 :return: Six words (separated by single space) token for the given step
2501 :rtype: str
2502 """
2503 response = OTPResponse(self._generate_otp_bytes(step))
2504 return response.words
2505
2506 def generate_otp_words_from_challenge(self, challenge: str) -> str:
2507 """
2508 Same as generate_otp_words, but it generates words from a challenge.
2509 2558
2510 RFC-2289 states: 2559 RFC-2289 states:
2511 The challenge MUST be in a standard syntax so 2560 The challenge MUST be in a standard syntax so
@@ -2522,35 +2571,14 @@ class OTPGenerator:
2522 seed, hash_algo, step = self.get_tokens_from_challenge(challenge) 2571 seed, hash_algo, step = self.get_tokens_from_challenge(challenge)
2523 self._seed = self.validate_seed(seed) 2572 self._seed = self.validate_seed(seed)
2524 self._hash_algo = self.validate_hash_algo(hash_algo) 2573 self._hash_algo = self.validate_hash_algo(hash_algo)
2525 return self.generate_otp_words(step) 2574 return self.generate_otp_response(step)
2526 2575
2527 def hexdigest_range( 2576 def otp_response_range(
2528 self, start: int = 499, stop: int = 0 2577 self, start: int = 499, stop: int = 0
2529 ) -> Iterator[str]: 2578 ) -> Iterator[OTPResponse]:
2530 """
2531 Returns an iterator that providing hexdigests corresponding to steps
2532 from `start` to and including `stop`.
2533
2534 :param start: The start of the range (default: 499)
2535 :type start: int
2536
2537 :param stop: The last step (default: 0)
2538 :type stop: int
2539
2540 :return: Iterator
2541 :rtype: generator
2542 """
2543 if not isinstance(start, int) and isinstance(stop, int):
2544 raise OTPGeneratorError('Step value MUST be an int')
2545 if start < stop:
2546 raise OTPGeneratorError('Start value can not be lower than stop')
2547 for step in range(start, stop - 1, -1):
2548 yield self.generate_otp_hexdigest(step)
2549
2550 def words_range(self, start: int = 499, stop: int = 0) -> Iterator[str]:
2551 """ 2579 """
2552 Returns an iterator that providing the words corresponding to steps 2580 Returns an iterator that is providing OTPResponse instances
2553 from `start` to and including `stop`. 2581 corresponding to steps from `start` to and including `stop`
2554 2582
2555 :param start: The start of the range (default: 499) 2583 :param start: The start of the range (default: 499)
2556 :type start: int 2584 :type start: int
@@ -2566,7 +2594,7 @@ class OTPGenerator:
2566 if start < stop: 2594 if start < stop:
2567 raise OTPGeneratorError('Start value can not be lower than stop') 2595 raise OTPGeneratorError('Start value can not be lower than stop')
2568 for step in range(start, stop - 1, -1): 2596 for step in range(start, stop - 1, -1):
2569 yield self.generate_otp_words(step) 2597 yield self.generate_otp_response(step)
2570 2598
2571 def _generate_otp_bytes(self, step: int) -> bytes: 2599 def _generate_otp_bytes(self, step: int) -> bytes:
2572 """ 2600 """