Coverage for gws-app/gws/lib/otp/__init__.py: 96%
74 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""Generate and check HOTP and TOTP one-time passwords.
3This package implements HMAC-based (HOTP, RFC 4226) and time-based (TOTP, RFC 6238)
4one-time passwords, as used by multi-factor authentication. It also creates ``otpauth://``
5key URIs for authenticator apps and random secrets.
7All functions accept an optional ``Options`` object. Options that are not set are taken
8from ``DEFAULTS`` (30 second step, 6 digits, SHA-1, tolerance of one step).
10Example::
12 import time
13 import gws.lib.otp
15 secret = gws.lib.otp.random_secret()
16 uri = gws.lib.otp.totp_key_uri(secret, 'GWS', 'user@example.com')
17 ok = gws.lib.otp.check_totp(user_input, secret, int(time.time()))
19References:
20 https://datatracker.ietf.org/doc/html/rfc4226
21 https://datatracker.ietf.org/doc/html/rfc6238
22 https://github.com/google/google-authenticator/wiki/Key-Uri-Format
23"""
25from typing import Optional, cast
27import base64
28import hashlib
29import hmac
30import random
32import gws
33import gws.lib.net
36class Options(gws.Data):
37 """OTP generation options."""
39 start: int
40 """Start time (Unix timestamp) for TOTP counting."""
41 step: int
42 """TOTP time step in seconds."""
43 length: int
44 """Number of digits in a token."""
45 tolerance: int
46 """Number of time steps before and after the current one that are also accepted."""
47 algo: str
48 """Hash algorithm name, as in ``hashlib``, e.g. ``sha1``."""
51DEFAULTS = Options(
52 start=0,
53 step=30,
54 length=6,
55 tolerance=1,
56 algo='sha1',
57)
60def new_hotp(secret: str | bytes, counter: int, options: Optional[Options] = None) -> str:
61 """Generate an HOTP token as per RFC 4226 section 5.3.
63 Args:
64 secret: Shared secret.
65 counter: Counter value.
66 options: Generation options.
68 Returns:
69 The token as a string of digits.
70 """
72 options = cast(Options, gws.u.merge(DEFAULTS, options))
73 return _raw_otp(_to_bytes(secret), counter, options)
76def new_totp(secret: str | bytes, timestamp: int, options: Optional[Options] = None) -> str:
77 """Generate a TOTP token as per RFC 6238 section 4.2.
79 Args:
80 secret: Shared secret.
81 timestamp: Unix timestamp.
82 options: Generation options.
84 Returns:
85 The token as a string of digits.
86 """
88 options = cast(Options, gws.u.merge(DEFAULTS, options))
89 counter = (timestamp - options.start) // options.step
90 return _raw_otp(_to_bytes(secret), counter, options)
93def check_totp(input: str, secret: str, timestamp: int, options: Optional[Options] = None) -> bool:
94 """Check if a TOTP token is valid.
96 Compares the input against the TOTP tokens within the tolerance window
97 ``(timestamp-step*tolerance...timestamp+step*tolerance)``.
99 Args:
100 input: Token entered by the user.
101 secret: Shared secret.
102 timestamp: Unix timestamp.
103 options: Generation options.
105 Returns:
106 ``True`` if the input matches one of the tokens in the window.
107 """
109 options = cast(Options, gws.u.merge(DEFAULTS, options))
111 if len(input) != options.length:
112 return False
114 ok = False
116 for window in range(-options.tolerance, options.tolerance + 1):
117 ts = timestamp + options.step * window
118 counter = (ts - options.start) // options.step
119 totp = _raw_otp(_to_bytes(secret), counter, options)
120 if hmac.compare_digest(_to_bytes(input), _to_bytes(totp)):
121 ok = True
123 return ok
126def totp_key_uri(
127 secret: str | bytes,
128 issuer_name: str,
129 account_name: str,
130 options: Optional[Options] = None
131) -> str:
132 """Create a TOTP key URI for authenticator apps.
134 Args:
135 secret: Shared secret, encoded as base32 in the URI.
136 issuer_name: Issuer name, e.g. the application name.
137 account_name: Account name, e.g. the user login.
138 options: Generation options. Only non-default values are included in the URI.
140 Returns:
141 An ``otpauth://totp/...`` URI.
142 """
143 return _key_uri('totp', secret, issuer_name, account_name, None, options)
146def hotp_key_uri(
147 secret: str | bytes,
148 issuer_name: str,
149 account_name: str,
150 counter: int,
151 options: Optional[Options] = None
152) -> str:
153 """Create an HOTP key URI for authenticator apps.
155 Args:
156 secret: Shared secret, encoded as base32 in the URI.
157 issuer_name: Issuer name, e.g. the application name.
158 account_name: Account name, e.g. the user login.
159 counter: Initial counter value.
160 options: Generation options. Only non-default values are included in the URI.
162 Returns:
163 An ``otpauth://hotp/...`` URI.
164 """
165 return _key_uri('hotp', secret, issuer_name, account_name, counter, options)
168def _key_uri(
169 method: str,
170 secret: str | bytes,
171 issuer_name: str,
172 account_name: str,
173 counter: Optional[int] = None,
174 options: Optional[Options] = None
175) -> str:
176 """Create a key URI for authenticator apps (Google Authenticator Key Uri Format)."""
178 params: dict = {
179 'secret': base32_encode(secret),
180 'issuer': issuer_name,
181 }
183 options = cast(Options, gws.u.merge(DEFAULTS, options))
185 if options.algo != DEFAULTS.algo:
186 params['algorithm'] = options.algo
187 if options.length != DEFAULTS.length:
188 params['digits'] = options.length
190 if method == 'hotp':
191 params['counter'] = counter
192 elif options.step != DEFAULTS.step:
193 params['period'] = options.step
195 return 'otpauth://{}/{}:{}?{}'.format(
196 method,
197 gws.lib.net.quote_param(issuer_name),
198 gws.lib.net.quote_param(account_name),
199 gws.lib.net.make_qs(params)
200 )
203def base32_decode(s: str) -> bytes:
204 """Decode a base32 string.
206 Args:
207 s: Base32 string.
209 Returns:
210 Decoded bytes.
211 """
212 return base64.b32decode(s)
215def base32_encode(s: str | bytes) -> str:
216 """Encode a string or bytes as base32.
218 Args:
219 s: Value to encode. Strings are encoded as UTF-8 first.
221 Returns:
222 Base32 string.
223 """
224 return base64.b32encode(_to_bytes(s)).decode('ascii')
227def random_secret(base32_length: int = 32) -> str:
228 """Generate a random secret of printable ASCII characters.
230 The secret length is chosen so that its base32 encoding is exactly ``base32_length`` characters long.
232 Args:
233 base32_length: Length of the base32-encoded secret, must be a multiple of 8.
235 Returns:
236 The secret.
238 Raises:
239 ``ValueError``: If ``base32_length`` is not a multiple of 8.
240 """
242 if (base32_length & 7) != 0:
243 raise ValueError('invalid length')
245 size = (base32_length >> 3) * 5
246 r = random.SystemRandom()
247 return ''.join(chr(r.randint(0x21, 0x7f)) for _ in range(size))
250##
252def _raw_otp(key: bytes, counter: int, options: Options) -> str:
253 # https://www.rfc-editor.org/rfc/rfc4226#section-5.3
254 #
255 # Step 1: Generate an HMAC-SHA-1 value
256 # Let HS = HMAC-SHA-1(K,C) // HS is a 20-byte string
257 #
258 # Step 2: Generate a 4-byte string (Dynamic Truncation)
259 # Let Sbits = DT(HS) // DT, defined below, returns a 31-bit string
260 #
261 # Let OffsetBits be the low-order 4 bits of String[19]
262 # Offset = StToNum(OffsetBits) // 0 <= OffSet <= 15
263 # Let P = String[OffSet]...String[OffSet+3]
264 # Return the Last 31 bits of P
265 #
266 # Let Snum = StToNum(Sbits) // Convert S to a number in 0...2^{31}-1
267 #
268 # Step 3: Compute an HOTP value
269 # Return D = Snum mod 10^Digit // D is a number in the range 0...10^{Digit}-1
271 c = counter.to_bytes(8, byteorder='big')
273 digestmod = getattr(hashlib, options.algo.lower())
274 hs = hmac.new(key, c, digestmod).digest()
276 offset = hs[-1] & 0xf
277 p = hs[offset:offset + 4]
278 snum = int.from_bytes(p, byteorder='big', signed=False) & 0x7fffffff
280 d = snum % (10 ** options.length)
282 return f'{d:0{options.length}d}'
285def _to_bytes(s):
286 return s.encode('utf8') if isinstance(s, str) else s
289def _option(options, key, default):
290 if not options:
291 return default
292 return getattr(options, key, default)