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