Coverage for gws-app/gws/base/auth/throttle.py: 99%

1"""Authentication throttle.""" 

2 

3from typing import Optional 

4 

5import gws 

6import gws.lib.datetimex as dtx 

7import gws.lib.sqlitex 

8 

9 

10class Config(gws.Config): 

11 """Blocking of repeated failed authentication attempts.""" 

12 

13 maxAttemptsPerIp: int = 10 

14 """Failed attempts from one address before blocking.""" 

15 maxAttemptsPerUser: int = 0 

16 """Failed attempts per login name from any address before blocking, 0 for no limit.""" 

17 windowTime: gws.Duration = '10m' 

18 """Time span in which failed attempts are counted.""" 

19 blockTime: gws.Duration = '15m' 

20 """How long to block once the limit is reached.""" 

21 allowFrom: Optional[list[str]] 

22 """IP addresses exempt from throttling.""" 

23 path: Optional[str] 

24 """File that stores failed attempts.""" 

25 

26 

27_CLEANUP_INTERVAL = 600 

28_MAX_NAME_LENGTH = 128 

29 

30 

31class Object(gws.Node): 

32 """Authentication throttle. 

33 

34 Counts failed authentication attempts per remote address and, optionally, 

35 per login name, and blocks further attempts once a limit is reached. 

36 """ 

37 

38 maxAttemptsPerIp: int 

39 """Failed attempts from one address before blocking, 0 for no limit.""" 

40 maxAttemptsPerUser: int 

41 """Failed attempts per login name before blocking, 0 for no limit.""" 

42 windowTime: int 

43 """Time span in seconds in which failed attempts are counted.""" 

44 blockTime: int 

45 """Blocking time in seconds.""" 

46 allowFrom: set[str] 

47 """IP addresses exempt from throttling.""" 

48 dbPath: str 

49 """Path to the sqlite database.""" 

50 

51 table = 'throttle' 

52 """Name of the database table.""" 

53 

54 def configure(self): 

55 self.maxAttemptsPerIp = self.cfg('maxAttemptsPerIp', default=10) 

56 self.maxAttemptsPerUser = self.cfg('maxAttemptsPerUser', default=0) 

57 self.windowTime = self.cfg('windowTime', default=dtx.parse_duration(Config.windowTime)) 

58 self.blockTime = self.cfg('blockTime', default=dtx.parse_duration(Config.blockTime)) 

59 self.allowFrom = set(self.cfg('allowFrom') or []) 

60 self.dbPath = self.cfg('path', default=f'{gws.c.MISC_DIR}/auth_throttle.sqlite') 

61 

62 if self.blockTime <= self.windowTime: 

63 raise gws.ConfigurationError(f'invalid blockTime={self.blockTime}, must be greater than windowTime={self.windowTime}') 

64 

65 ## 

66 

67 def blocked_for(self, req: gws.WebRequester, method: gws.AuthMethod, credentials: gws.Data) -> int: 

68 """Return how long an authentication attempt remains blocked. 

69 

70 Args: 

71 req: The web request, provides the remote address. 

72 method: The authentication method. 

73 credentials: The credentials, provide the login name as ``username``. 

74 

75 Returns: 

76 The remaining blocking time in seconds, 0 if the attempt is allowed. 

77 """ 

78 

79 u_addr, u_user = self._get_uids(req, credentials) 

80 if not u_addr and not u_user: 

81 return 0 

82 

83 now = gws.u.stime() 

84 rs = self._db().select( 

85 f'SELECT MAX(blocked_until) AS t FROM {self.table} WHERE uid IN (:u_addr, :u_user)', 

86 u_addr=u_addr, 

87 u_user=u_user, 

88 ) 

89 

90 t = rs[0]['t'] if rs else None 

91 return max(0, (t or 0) - now) 

92 

93 def register(self, ok: bool, req: gws.WebRequester, method: gws.AuthMethod, credentials: gws.Data): 

94 """Register the outcome of an authentication attempt. 

95 

96 A success removes the counts for the address and the login name. 

97 A failure increments them and starts a block once a limit is reached. 

98 

99 Args: 

100 ok: Whether the attempt was successful. 

101 req: The web request, provides the remote address. 

102 method: The authentication method. 

103 credentials: The credentials, provide the login name as ``username``. 

104 """ 

105 

106 u_addr, u_user = self._get_uids(req, credentials) 

107 if not u_addr and not u_user: 

108 return 

109 

110 if ok: 

111 self._db().execute( 

112 f'DELETE FROM {self.table} WHERE uid IN (:u_addr, :u_user)', 

113 u_addr=u_addr, 

114 u_user=u_user, 

115 ) 

116 return 

117 

118 if gws.u.stime() > self._cleanupTime + _CLEANUP_INTERVAL: 

119 self.cleanup() 

120 

121 if u_addr: 

122 self._add_failure(u_addr, self.maxAttemptsPerIp) 

123 if u_user: 

124 self._add_failure(u_user, self.maxAttemptsPerUser) 

125 

126 _cleanupTime = 0 

127 

128 def cleanup(self): 

129 """Remove the counts whose window has elapsed and which hold no active block.""" 

130 

131 # a row may only be dropped when its window has elapsed *and* it holds no live block, 

132 # the row is the only place a block is recorded 

133 

134 now = gws.u.stime() 

135 self._db().execute( 

136 f'DELETE FROM {self.table} WHERE first_time < :window_start AND blocked_until <= :now', 

137 window_start=now - self.windowTime, 

138 now=now, 

139 ) 

140 self._cleanupTime = now 

141 

142 ## 

143 

144 def _add_failure(self, uid: str, max_attempts: int): 

145 """Increment the failure count of a key, starting a new window if needed, and block if the limit is reached.""" 

146 now = gws.u.stime() 

147 

148 # a new attempt starts a new window if the current one has elapsed. 

149 # an expired block needs no test of its own: since blockTime is greater than windowTime, 

150 # the window has always elapsed by the time a block runs out 

151 

152 expired = 'first_time < :window_start' 

153 

154 self._db().execute( 

155 f""" 

156 INSERT INTO {self.table} (uid, attempts, first_time, blocked_until) 

157 VALUES (:uid, 1, :now, 0) 

158 ON CONFLICT (uid) DO UPDATE SET 

159 attempts = CASE WHEN {expired} THEN 1 ELSE attempts + 1 END, 

160 first_time = CASE WHEN {expired} THEN :now ELSE first_time END, 

161 blocked_until = 0 

162 """, 

163 uid=uid, 

164 now=now, 

165 window_start=now - self.windowTime, 

166 ) 

167 

168 self._db().execute( 

169 f""" 

170 UPDATE {self.table} SET blocked_until = :until 

171 WHERE uid = :uid AND attempts >= :max_attempts 

172 """, 

173 uid=uid, 

174 until=now + self.blockTime, 

175 max_attempts=max_attempts, 

176 ) 

177 

178 def _get_uids(self, req: gws.WebRequester, credentials: gws.Data) -> tuple[str, str]: 

179 """Return the address and login name keys of an attempt, empty if not counted.""" 

180 ip = req.ip 

181 if ip and ip in self.allowFrom: 

182 return '', '' 

183 

184 u_addr = '' 

185 if ip and self.maxAttemptsPerIp > 0: 

186 u_addr = f'ip:{ip}' 

187 

188 u_user = '' 

189 if self.maxAttemptsPerUser > 0: 

190 name = self._login_name(credentials) 

191 if name: 

192 u_user = f'user:{gws.u.sha256(name)}' 

193 

194 return u_addr, u_user 

195 

196 def _login_name(self, credentials: gws.Data) -> str: 

197 """Return the normalized login name from the credentials.""" 

198 s = credentials.get('username') 

199 if not isinstance(s, str): 

200 return '' 

201 return s.strip().casefold()[:_MAX_NAME_LENGTH] 

202 

203 ## 

204 

205 _sqlitex: gws.lib.sqlitex.Object 

206 

207 def _db(self): 

208 """Return the sqlite database, creating the table on first use.""" 

209 if getattr(self, '_sqlitex', None) is None: 

210 ddl = f""" 

211 CREATE TABLE IF NOT EXISTS {self.table} ( 

212 uid TEXT NOT NULL PRIMARY KEY, 

213 attempts INTEGER NOT NULL, 

214 first_time INTEGER NOT NULL, 

215 blocked_until INTEGER NOT NULL 

216 ) 

217 """ 

218 self._sqlitex = gws.lib.sqlitex.Object(self.dbPath, ddl) 

219 return self._sqlitex