Coverage for src/lexigram/auth/authn/mfa.py: 97%

61 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-26 00:58 +0800

1"""MFA utilities - TOTP (RFC 6238) and backup codes 

2 

3This module implements a small TOTP engine so we don't add a dependency 

4on `pyotp`. It provides helper functions to generate secrets, provisioning 

5URIs, verify codes, and to manage one-time backup codes. 

6 

7MFA data is stored on the `User.profile['mfa']` key to avoid immediate DB 

8schema changes. The helpers in AuthProvider operate on the `User` object and 

9use `AuthProvider.user_store.update_user()` to persist changes. 

10""" 

11 

12from __future__ import annotations 

13 

14import base64 

15import hashlib 

16import hmac 

17import secrets 

18import struct 

19from typing import TYPE_CHECKING 

20 

21from lexigram.primitives import clock as ambient_clock 

22 

23if TYPE_CHECKING: 

24 from collections.abc import Iterable 

25 

26DEFAULT_TOTP_DIGITS = 6 

27DEFAULT_TOTP_ALGORITHM = "SHA1" 

28DEFAULT_TOTP_PERIOD = 30 

29 

30 

31def _int_to_bytes(i: int) -> bytes: 

32 return struct.pack( 

33 ">Q", 

34 i, 

35 ) 

36 

37 

38def generate_totp_secret(length: int = 20) -> str: 

39 """Generate a base32-encoded secret for TOTP. 

40 

41 Args: 

42 length: number of random bytes to use before base32-encoding 

43 Returns: 

44 Base32 string without padding, upper-case (common otpauth format) 

45 """ 

46 b = secrets.token_bytes(length) 

47 s = base64.b32encode(b).decode("ascii") 

48 return s.strip("=") 

49 

50 

51def _hotp(secret: str, counter: int, digits: int = DEFAULT_TOTP_DIGITS) -> str: 

52 key = base64.b32decode(_normalize_base32(secret), casefold=True) 

53 msg = _int_to_bytes(counter) 

54 h = hmac.new(key, msg, hashlib.sha1).digest() 

55 offset = h[-1] & 0x0F 

56 code = struct.unpack( 

57 ">I", 

58 h[offset : offset + 4], 

59 )[0] 

60 code = code & 0x7FFFFFFF 

61 return str(code % (10**digits)).zfill(digits) 

62 

63 

64def _normalize_base32(secret: str) -> str: 

65 """Normalize base32 secret by adding padding if necessary""" 

66 s = secret.strip().replace(" ", "").upper() 

67 padding = "=" * ((8 - (len(s) % 8)) % 8) 

68 return s + padding 

69 

70 

71def generate_totp_code( 

72 secret: str, 

73 for_time: int | None = None, 

74 period: int = DEFAULT_TOTP_PERIOD, 

75 digits: int = DEFAULT_TOTP_DIGITS, 

76) -> str: 

77 """Generate TOTP code for given secret and time (unix seconds). 

78 

79 Args: 

80 secret: base32 secret 

81 for_time: unix timestamp; defaults to now 

82 period: time-step period in seconds 

83 digits: number of OTP digits (default: DEFAULT_TOTP_DIGITS = 6) 

84 

85 Returns: 

86 zero-padded numeric code string 

87 """ 

88 if for_time is None: 

89 for_time = int(ambient_clock.timestamp()) 

90 counter = int(for_time // period) 

91 return _hotp(secret, counter, digits=digits) 

92 

93 

94def verify_totp( 

95 secret: str, 

96 code: str, 

97 window: int = 1, 

98 period: int = DEFAULT_TOTP_PERIOD, 

99) -> bool: 

100 """Verify a TOTP code allowing a +/- window of time steps.""" 

101 try: 

102 code = str(code).zfill(DEFAULT_TOTP_DIGITS) 

103 except (TypeError, ValueError): 

104 return False 

105 

106 now = int(ambient_clock.timestamp()) 

107 for w in range(-window, window + 1): 

108 c = int((now // period) + w) 

109 if _hotp(secret, c) == code: 

110 return True 

111 return False 

112 

113 

114def get_provisioning_uri(secret: str, username: str, issuer: str) -> str: 

115 """Return an `otpauth://` provisioning URI suitable for authenticator apps.""" 

116 # Keep it simple; apps expect base32 secret without padding 

117 label = f"{issuer}:{username}" 

118 return f"otpauth://totp/{label}?secret={secret}&issuer={issuer}&algorithm=SHA1&digits={DEFAULT_TOTP_DIGITS}&period={DEFAULT_TOTP_PERIOD}" 

119 

120 

121def generate_backup_codes(count: int = 10, length: int = 8) -> list[str]: 

122 """Generate a list of human-friendly backup codes (plain strings). 

123 

124 These are returned to the caller once and should be hashed before storage. 

125 """ 

126 codes = [] 

127 alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # exclude confusing chars 

128 for _ in range(count): 

129 codes.append("".join(secrets.choice(alphabet) for _ in range(length))) 

130 return codes 

131 

132 

133def hash_backup_codes(codes: Iterable[str]) -> list[str]: 

134 """Return digest hashes for backup codes (store these, compare with digest).""" 

135 out = [] 

136 for c in codes: 

137 h = hashlib.sha256(c.encode("utf-8")).hexdigest() 

138 out.append(h) 

139 return out 

140 

141 

142__all__ = [ 

143 "generate_backup_codes", 

144 "generate_totp_code", 

145 "generate_totp_secret", 

146 "get_provisioning_uri", 

147 "hash_backup_codes", 

148 "verify_totp", 

149]