summaryrefslogtreecommitdiff
path: root/src/otp2289/server.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/otp2289/server.py')
-rw-r--r--src/otp2289/server.py88
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
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