Coverage for gws-app/gws/base/auth/mfa.py: 73%

1"""Base multi-factor authentication adapter.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.lib.otp 

7 

8 

9class OtpConfig: 

10 """Options for one-time password generation.""" 

11 

12 start: Optional[int] 

13 """Start time for TOTP counting, as a Unix timestamp.""" 

14 step: Optional[int] 

15 """TOTP time step in seconds.""" 

16 length: Optional[int] 

17 """Number of digits in the code.""" 

18 tolerance: Optional[int] 

19 """Accepted time steps before and after the current one.""" 

20 algo: Optional[str] 

21 """Hash algorithm for code generation.""" 

22 

23 

24class Config(gws.Config): 

25 """Multi-factor authorization configuration.""" 

26 

27 message: str = '' 

28 """Message shown to the user during the MFA step.""" 

29 lifeTime: Optional[gws.Duration] = '120' 

30 """Time allowed to complete the MFA step.""" 

31 maxVerifyAttempts: int = 3 

32 """Code entries allowed before the MFA step fails.""" 

33 maxRestarts: int = 0 

34 """How often a new code can be requested.""" 

35 otp: Optional[OtpConfig] 

36 """OTP generation options.""" 

37 

38 

39class Object(gws.AuthMultiFactorAdapter): 

40 """Base multi-factor authentication adapter. 

41 

42 Implements the transaction life cycle (start, state checks, restarts, 

43 counting of verification attempts) and provides TOTP code generation and 

44 checking. Subclasses implement ``verify`` and usually extend ``start``. 

45 """ 

46 

47 otpOptions: gws.lib.otp.Options 

48 """Options for one-time password generation.""" 

49 

50 def configure(self): 

51 self.message = self.cfg('message', default='') 

52 self.lifeTime = self.cfg('lifeTime', default=120) 

53 self.maxVerifyAttempts = self.cfg('maxVerifyAttempts', default=3) 

54 self.maxRestarts = self.cfg('maxRestarts', default=0) 

55 self.otpOptions = gws.u.merge(gws.lib.otp.DEFAULTS, self.cfg('otp')) 

56 

57 def start(self, user): 

58 return gws.AuthMultiFactorTransaction( 

59 state=gws.AuthMultiFactorState.open, 

60 restartCount=0, 

61 verifyCount=0, 

62 secret='', 

63 startTime=self.current_timestamp(), 

64 generateTime=0, 

65 message=self.message, 

66 adapter=self, 

67 user=user, 

68 ) 

69 

70 def check_state(self, mfa): 

71 ts = self.current_timestamp() 

72 if ts - mfa.startTime >= self.lifeTime: 

73 mfa.state = gws.AuthMultiFactorState.failed 

74 return False 

75 if mfa.verifyCount > self.maxVerifyAttempts: 

76 mfa.state = gws.AuthMultiFactorState.failed 

77 return False 

78 if mfa.state == gws.AuthMultiFactorState.failed: 

79 return False 

80 return True 

81 

82 def check_restart(self, mfa): 

83 return mfa.restartCount < self.maxRestarts 

84 

85 def restart(self, mfa): 

86 rc = mfa.restartCount + 1 

87 if rc > self.maxRestarts: 

88 return 

89 

90 mfa = self.start(mfa.user) 

91 if not mfa: 

92 return 

93 

94 mfa.restartCount = rc 

95 return mfa 

96 

97 ## 

98 

99 def verify_attempt(self, mfa, payload_valid: bool): 

100 """Count a verification attempt and update the transaction state. 

101 

102 The state becomes ``ok`` if the payload is valid, ``failed`` if the transaction 

103 is no longer valid or the attempts are used up, and ``retry`` otherwise. 

104 

105 Args: 

106 mfa: The transaction. 

107 payload_valid: Whether the submitted payload was valid. 

108 

109 Returns: 

110 The same transaction, updated. 

111 """ 

112 mfa.verifyCount += 1 

113 

114 if not self.check_state(mfa): 

115 return mfa 

116 

117 if payload_valid: 

118 mfa.state = gws.AuthMultiFactorState.ok 

119 return mfa 

120 

121 if mfa.verifyCount >= self.maxVerifyAttempts: 

122 mfa.state = gws.AuthMultiFactorState.failed 

123 return mfa 

124 

125 mfa.state = gws.AuthMultiFactorState.retry 

126 return mfa 

127 

128 def generate_totp(self, mfa: gws.AuthMultiFactorTransaction) -> str: 

129 """Generate a TOTP code from the transaction secret for the current time. 

130 

131 Also records the generation time in the transaction. 

132 

133 Args: 

134 mfa: The transaction. 

135 

136 Returns: 

137 The code. 

138 """ 

139 ts = self.current_timestamp() 

140 totp = gws.lib.otp.new_totp(mfa.secret, ts, self.otpOptions) 

141 mfa.generateTime = ts 

142 gws.log.debug(f'generate_totp {ts=} {totp=} {mfa.generateTime=}') 

143 return totp 

144 

145 def check_totp(self, mfa: gws.AuthMultiFactorTransaction, input: str) -> bool: 

146 """Check a TOTP code against the transaction secret for the current time. 

147 

148 Args: 

149 mfa: The transaction. 

150 input: The code entered by the user. 

151 

152 Returns: 

153 ``True`` if the code is valid. 

154 """ 

155 return gws.lib.otp.check_totp( 

156 str(input or ''), 

157 mfa.secret, 

158 self.current_timestamp(), 

159 self.otpOptions, 

160 ) 

161 

162 def current_timestamp(self): 

163 """Return the current time. 

164 

165 Returns: 

166 The current time as a Unix timestamp in seconds. 

167 """ 

168 return gws.u.stime()