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

1"""Generate and check HOTP and TOTP one-time passwords. 

2 

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. 

6 

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). 

9 

10Example:: 

11 

12 import time 

13 import gws.lib.otp 

14 

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())) 

18 

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""" 

24 

25from typing import Optional, cast 

26 

27import base64 

28import hashlib 

29import hmac 

30import random 

31 

32import gws 

33import gws.lib.net 

34 

35 

36class Options(gws.Data): 

37 """OTP generation options.""" 

38 

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``.""" 

49 

50 

51DEFAULTS = Options( 

52 start=0, 

53 step=30, 

54 length=6, 

55 tolerance=1, 

56 algo='sha1', 

57) 

58 

59 

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. 

62 

63 Args: 

64 secret: Shared secret. 

65 counter: Counter value. 

66 options: Generation options. 

67 

68 Returns: 

69 The token as a string of digits. 

70 """ 

71 

72 options = cast(Options, gws.u.merge(DEFAULTS, options)) 

73 return _raw_otp(_to_bytes(secret), counter, options) 

74 

75 

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. 

78 

79 Args: 

80 secret: Shared secret. 

81 timestamp: Unix timestamp. 

82 options: Generation options. 

83 

84 Returns: 

85 The token as a string of digits. 

86 """ 

87 

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) 

91 

92 

93def check_totp(input: str, secret: str, timestamp: int, options: Optional[Options] = None) -> bool: 

94 """Check if a TOTP token is valid. 

95 

96 Compares the input against the TOTP tokens within the tolerance window 

97 ``(timestamp-step*tolerance...timestamp+step*tolerance)``. 

98 

99 Args: 

100 input: Token entered by the user. 

101 secret: Shared secret. 

102 timestamp: Unix timestamp. 

103 options: Generation options. 

104 

105 Returns: 

106 ``True`` if the input matches one of the tokens in the window. 

107 """ 

108 

109 options = cast(Options, gws.u.merge(DEFAULTS, options)) 

110 

111 if len(input) != options.length: 

112 return False 

113 

114 ok = False 

115 

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 

122 

123 return ok 

124 

125 

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. 

133 

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. 

139 

140 Returns: 

141 An ``otpauth://totp/...`` URI. 

142 """ 

143 return _key_uri('totp', secret, issuer_name, account_name, None, options) 

144 

145 

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. 

154 

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. 

161 

162 Returns: 

163 An ``otpauth://hotp/...`` URI. 

164 """ 

165 return _key_uri('hotp', secret, issuer_name, account_name, counter, options) 

166 

167 

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).""" 

177 

178 params: dict = { 

179 'secret': base32_encode(secret), 

180 'issuer': issuer_name, 

181 } 

182 

183 options = cast(Options, gws.u.merge(DEFAULTS, options)) 

184 

185 if options.algo != DEFAULTS.algo: 

186 params['algorithm'] = options.algo 

187 if options.length != DEFAULTS.length: 

188 params['digits'] = options.length 

189 

190 if method == 'hotp': 

191 params['counter'] = counter 

192 elif options.step != DEFAULTS.step: 

193 params['period'] = options.step 

194 

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 ) 

201 

202 

203def base32_decode(s: str) -> bytes: 

204 """Decode a base32 string. 

205 

206 Args: 

207 s: Base32 string. 

208 

209 Returns: 

210 Decoded bytes. 

211 """ 

212 return base64.b32decode(s) 

213 

214 

215def base32_encode(s: str | bytes) -> str: 

216 """Encode a string or bytes as base32. 

217 

218 Args: 

219 s: Value to encode. Strings are encoded as UTF-8 first. 

220 

221 Returns: 

222 Base32 string. 

223 """ 

224 return base64.b32encode(_to_bytes(s)).decode('ascii') 

225 

226 

227def random_secret(base32_length: int = 32) -> str: 

228 """Generate a random secret of printable ASCII characters. 

229 

230 The secret length is chosen so that its base32 encoding is exactly ``base32_length`` characters long. 

231 

232 Args: 

233 base32_length: Length of the base32-encoded secret, must be a multiple of 8. 

234 

235 Returns: 

236 The secret. 

237 

238 Raises: 

239 ``ValueError``: If ``base32_length`` is not a multiple of 8. 

240 """ 

241 

242 if (base32_length & 7) != 0: 

243 raise ValueError('invalid length') 

244 

245 size = (base32_length >> 3) * 5 

246 r = random.SystemRandom() 

247 return ''.join(chr(r.randint(0x21, 0x7f)) for _ in range(size)) 

248 

249 

250## 

251 

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 

270 

271 c = counter.to_bytes(8, byteorder='big') 

272 

273 digestmod = getattr(hashlib, options.algo.lower()) 

274 hs = hmac.new(key, c, digestmod).digest() 

275 

276 offset = hs[-1] & 0xf 

277 p = hs[offset:offset + 4] 

278 snum = int.from_bytes(p, byteorder='big', signed=False) & 0x7fffffff 

279 

280 d = snum % (10 ** options.length) 

281 

282 return f'{d:0{options.length}d}' 

283 

284 

285def _to_bytes(s): 

286 return s.encode('utf8') if isinstance(s, str) else s 

287 

288 

289def _option(options, key, default): 

290 if not options: 

291 return default 

292 return getattr(options, key, default)