|
2 | 2 | import datetime |
3 | 3 | import hashlib |
4 | 4 | import time |
5 | | -from typing import Any, Optional, Union |
| 5 | +from typing import Any, Literal, Optional, Union |
6 | 6 |
|
7 | 7 | from . import utils |
8 | 8 | from .otp import OTP |
@@ -40,7 +40,7 @@ def __init__( |
40 | 40 |
|
41 | 41 | def at(self, for_time: Union[int, datetime.datetime], counter_offset: int = 0) -> str: |
42 | 42 | """ |
43 | | - Accepts either a Unix timestamp integer or a datetime object. |
| 43 | + Accepts either a Unix timestamp or a datetime object. |
44 | 44 |
|
45 | 45 | To get the time until the next timecode change (seconds until the current OTP expires), use this instead: |
46 | 46 |
|
@@ -85,6 +85,35 @@ def verify(self, otp: str, for_time: Optional[datetime.datetime] = None, valid_w |
85 | 85 |
|
86 | 86 | return utils.strings_equal(str(otp), str(self.at(for_time))) |
87 | 87 |
|
| 88 | + def verify_and_get_timecode( |
| 89 | + self, otp: str, for_time: Optional[Union[int, datetime.datetime]] = None, valid_window: int = 0 |
| 90 | + ) -> int | Literal[False]: |
| 91 | + """ |
| 92 | + Verifies the OTP passed in against the current time OTP and returns the matching timecode. |
| 93 | +
|
| 94 | + This is useful when callers need to prevent TOTP reuse. An |
| 95 | + application can store the timecode from the last successful |
| 96 | + authentication and reject future OTPs whose matching timecode |
| 97 | + is not greater than the stored one. |
| 98 | +
|
| 99 | + :param otp: the OTP to check against |
| 100 | + :param for_time: Time to check OTP at (defaults to now); accepts datetime or Unix timestamp |
| 101 | + :param valid_window: extends the validity to this counter ticks before and after the current one |
| 102 | + :returns: matching timecode if verification succeeded, False otherwise |
| 103 | + """ |
| 104 | + if for_time is None: |
| 105 | + for_time = datetime.datetime.now() |
| 106 | + elif isinstance(for_time, float): |
| 107 | + for_time = datetime.datetime.fromtimestamp(int(for_time)) |
| 108 | + |
| 109 | + if valid_window < 0: |
| 110 | + raise ValueError("valid_window cannot be negative") |
| 111 | + |
| 112 | + for i in range(-valid_window, valid_window + 1): |
| 113 | + if utils.strings_equal(str(otp), str(self.at(for_time, i))): |
| 114 | + return self.timecode(for_time) + i |
| 115 | + return False |
| 116 | + |
88 | 117 | def provisioning_uri(self, name: Optional[str] = None, issuer_name: Optional[str] = None, **kwargs) -> str: |
89 | 118 | """ |
90 | 119 | Returns the provisioning URI for the OTP. This can then be |
|
0 commit comments