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

1"""Authentication throttle. 

2 

3Counts failed authentication attempts and blocks further attempts once a limit is reached. 

4Attempts are counted per remote address and, optionally, per login name. 

5""" 

6 

7from typing import Optional 

8 

9import gws 

10import gws.lib.datetimex as dtx 

11import gws.lib.sqlitex 

12 

13 

14class Config(gws.Config): 

15 """Authentication throttle options. (added in 8.4)""" 

16 

17 maxAttemptsPerIp: int = 10 

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

19 maxAttemptsPerUser: int = 0 

20 """Failed attempts for one login from all addresses before blocking, 0=no limit.""" 

21 windowTime: gws.Duration = '10m' 

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

23 blockTime: gws.Duration = '15m' 

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

25 allowFrom: Optional[list[str]] 

26 """Addresses exempt from throttling.""" 

27 path: Optional[str] 

28 """Throttle storage path.""" 

29 

30 

31_CLEANUP_INTERVAL = 600 

32_MAX_NAME_LENGTH = 128 

33 

34 

35class Object(gws.Node): 

36 """Authentication throttle.""" 

37 

38 maxAttemptsPerIp: int 

39 maxAttemptsPerUser: int 

40 windowTime: int 

41 blockTime: int 

42 allowFrom: set[str] 

43 dbPath: str 

44 

45 table = 'throttle' 

46 

47 def configure(self): 

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

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

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

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

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

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

54 

55 if self.blockTime <= self.windowTime: 

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

57 

58 ## 

59 

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

61 """Return the time in seconds the given attempt remains blocked, 0 if it is allowed.""" 

62 

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

64 if not u_addr and not u_user: 

65 return 0 

66 

67 now = gws.u.stime() 

68 rs = self._db().select( 

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

70 u_addr=u_addr, 

71 u_user=u_user, 

72 ) 

73 

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

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

76 

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

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

79 

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

81 if not u_addr and not u_user: 

82 return 

83 

84 if ok: 

85 self._db().execute( 

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

87 u_addr=u_addr, 

88 u_user=u_user, 

89 ) 

90 return 

91 

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

93 self.cleanup() 

94 

95 if u_addr: 

96 self._add_failure(u_addr, self.maxAttemptsPerIp) 

97 if u_user: 

98 self._add_failure(u_user, self.maxAttemptsPerUser) 

99 

100 _cleanupTime = 0 

101 

102 def cleanup(self): 

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

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

105 

106 now = gws.u.stime() 

107 self._db().execute( 

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

109 window_start=now - self.windowTime, 

110 now=now, 

111 ) 

112 self._cleanupTime = now 

113 

114 ## 

115 

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

117 now = gws.u.stime() 

118 

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

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

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

122 

123 expired = 'first_time < :window_start' 

124 

125 self._db().execute( 

126 f""" 

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

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

129 ON CONFLICT (uid) DO UPDATE SET 

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

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

132 blocked_until = 0 

133 """, 

134 uid=uid, 

135 now=now, 

136 window_start=now - self.windowTime, 

137 ) 

138 

139 self._db().execute( 

140 f""" 

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

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

143 """, 

144 uid=uid, 

145 until=now + self.blockTime, 

146 max_attempts=max_attempts, 

147 ) 

148 

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

150 ip = req.ip 

151 if ip and ip in self.allowFrom: 

152 return '', '' 

153 

154 u_addr = '' 

155 if ip and self.maxAttemptsPerIp > 0: 

156 u_addr = f'ip:{ip}' 

157 

158 u_user = '' 

159 if self.maxAttemptsPerUser > 0: 

160 name = self._login_name(credentials) 

161 if name: 

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

163 

164 return u_addr, u_user 

165 

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

167 s = credentials.get('username') 

168 if not isinstance(s, str): 

169 return '' 

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

171 

172 ## 

173 

174 _sqlitex: gws.lib.sqlitex.Object 

175 

176 def _db(self): 

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

178 ddl = f""" 

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

180 uid TEXT NOT NULL PRIMARY KEY, 

181 attempts INTEGER NOT NULL, 

182 first_time INTEGER NOT NULL, 

183 blocked_until INTEGER NOT NULL 

184 ) 

185 """ 

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

187 return self._sqlitex