Coverage for gws-app/gws/base/auth/manager.py: 88%

1"""Authorization manager.""" 

2 

3from typing import Optional, cast 

4 

5import gws 

6import gws.config 

7import gws.lib.jsonx 

8 

9from . import session, system_provider, throttle 

10 

11 

12class Config(gws.Config): 

13 """Authentication methods, providers, sessions and login throttling.""" 

14 

15 methods: Optional[list[gws.ext.config.authMethod]] 

16 """Login methods available to users.""" 

17 providers: Optional[list[gws.ext.config.authProvider]] 

18 """User sources that verify credentials.""" 

19 mfa: Optional[list[gws.ext.config.authMultiFactorAdapter]] 

20 """Multi-factor authentication adapters.""" 

21 session: Optional[gws.ext.config.authSessionManager] 

22 """Session storage and life time.""" 

23 throttle: Optional[throttle.Config] 

24 """Blocking of repeated failed login attempts, disabled if not set.""" 

25 

26 

27_DEFAULT_SESSION_TYPE = 'sqlite' 

28 

29 

30class Object(gws.AuthManager): 

31 """Authorization manager.""" 

32 

33 def configure(self): 

34 self.sessionMgr = self.create_child(gws.ext.object.authSessionManager, self.cfg('session'), type=_DEFAULT_SESSION_TYPE) 

35 

36 self.throttle = None 

37 p = self.cfg('throttle') 

38 if p: 

39 self.throttle = cast(throttle.Object, self.create_child(throttle.Object, p)) 

40 

41 self.providers = self.create_children(gws.ext.object.authProvider, self.cfg('providers')) 

42 

43 sys_provider = self.create_child(system_provider.Object) 

44 self.add_provider(sys_provider) 

45 

46 self.guestUser = sys_provider.get_user('guest') 

47 self.systemUser = sys_provider.get_user('system') 

48 

49 self.methods = self.create_children(gws.ext.object.authMethod, self.cfg('methods')) 

50 if not self.methods: 

51 # if no methods configured, enable the Web method 

52 self.add_method(self.create_child(gws.ext.object.authMethod, type='web')) 

53 

54 self.mfAdapters = self.create_children(gws.ext.object.authMultiFactorAdapter, self.cfg('mfa')) 

55 

56 self.guestSession = session.Object(uid='guest_session', method=None, user=self.guestUser) 

57 

58 self.root.app.middlewareMgr.register(self, 'auth', depends_on=['db']) 

59 

60 ## 

61 

62 def add_provider(self, provider): 

63 self.providers.append(provider) 

64 

65 def add_method(self, method): 

66 self.methods.append(method) 

67 

68 def add_multi_factor_adapter(self, adapter): 

69 self.mfAdapters.append(adapter) 

70 

71 ## 

72 

73 def enter_middleware(self, req: gws.WebRequester): 

74 sess = self._try_open_session(req) 

75 if sess: 

76 gws.log.debug(f'session: user={req.session.user.uid!r} roles={req.session.user.roles}') 

77 req.set_session(sess) 

78 else: 

79 gws.log.debug('session: guest') 

80 req.set_session(self.guestSession) 

81 

82 def _try_open_session(self, req): 

83 """Return the first session opened by a usable method, or ``None``. 

84 

85 A method that requires a secure context is skipped on insecure requests, 

86 unless the client address is in its ``allowInsecureFrom``. A session whose 

87 user no longer exists or whose method does not match the opening method is 

88 deleted, and ``None`` is returned, so the request continues as guest. 

89 """ 

90 for meth in self.methods: 

91 if not self.can_use_method(req, meth): 

92 gws.log.warning(f'open_session: {meth=}: insecure_context, ignore') 

93 continue 

94 

95 sess = meth.open_session(req) 

96 if not sess: 

97 continue 

98 

99 if not sess.user: 

100 gws.log.warning(f'open_session: {meth=}: {sess.uid=} user not found') 

101 self.sessionMgr.delete(sess) 

102 return 

103 

104 if not sess.method or sess.method.uid != meth.uid: 

105 gws.log.warning(f'open_session: {meth=}: {sess.uid=} wrong method {sess.method=}') 

106 self.sessionMgr.delete(sess) 

107 return 

108 

109 gws.log.debug(f'open_session: {meth=}: ok') 

110 return sess 

111 

112 def exit_middleware(self, req: gws.WebRequester, res: gws.WebResponder): 

113 sess = req.session 

114 if sess.method: 

115 sess.method.close_session(req, res) 

116 req.set_session(self.guestSession) 

117 

118 def can_use_method(self, req, meth): 

119 if not meth.secure or req.isSecure: 

120 return True 

121 if not meth.allowInsecureFrom: 

122 return False 

123 if req.ip not in meth.allowInsecureFrom: 

124 return False 

125 gws.log.warning(f'open_session: {meth=}: insecure_context allowed from {req.ip=}') 

126 return True 

127 

128 ## 

129 

130 def create_transient_session(self, method, user, data=None): 

131 return session.Object( 

132 uid=gws.u.random_string(64), 

133 method=method, 

134 user=user, 

135 data=data, 

136 is_changed=False, 

137 is_transient=True, 

138 ) 

139 

140 ## 

141 

142 def authenticate(self, method, credentials, req): 

143 if not self.throttle: 

144 return self._authenticate2(method, credentials) 

145 

146 sec = self.throttle.blocked_for(req, method, credentials) 

147 if sec > 0: 

148 gws.log.warning(f'authenticate: {method=}: throttled for {sec}s') 

149 exc = gws.TooManyRequestsError('too many authentication attempts') 

150 exc.retryAfter = sec 

151 raise exc 

152 

153 try: 

154 user = self._authenticate2(method, credentials) 

155 except gws.ForbiddenError: 

156 self.throttle.register(False, req, method, credentials) 

157 raise 

158 

159 self.throttle.register(user is not None, req, method, credentials) 

160 return user 

161 

162 def _authenticate2(self, method, credentials): 

163 """Try each provider that allows the method, return the first user found.""" 

164 for prov in self.providers: 

165 if prov.allowedMethods and method.extType not in prov.allowedMethods: 

166 continue 

167 gws.log.debug(f'trying provider {prov!r}') 

168 user = prov.authenticate(method, credentials) 

169 if user: 

170 gws.log.debug(f'ok provider {prov!r}') 

171 return user 

172 

173 ## 

174 

175 def get_user(self, user_uid): 

176 provider_uid, local_uid = gws.u.split_uid(user_uid) 

177 prov = self.get_provider(provider_uid) 

178 return prov.get_user(local_uid) if prov else None 

179 

180 def get_provider(self, uid): 

181 for obj in self.providers: 

182 if obj.uid == uid: 

183 return obj 

184 

185 def get_method(self, uid=None, ext_type=None): 

186 for obj in self.methods: 

187 if obj.uid == uid: 

188 return obj 

189 

190 def get_multi_factor_adapter(self, uid=None, ext_type=None): 

191 for obj in self.mfAdapters: 

192 if obj.uid == uid: 

193 return obj 

194 

195 def serialize_user(self, user): 

196 return gws.lib.jsonx.to_string([user.authProvider.uid, user.authProvider.serialize_user(user)]) 

197 

198 def unserialize_user(self, data): 

199 provider_uid, ds = gws.lib.jsonx.from_string(data) 

200 prov = self.get_provider(provider_uid) 

201 return prov.unserialize_user(ds) if prov else None 

202 

203 def is_public_object(self, obj, *context): 

204 return self.guestUser.can_read(obj, *context)