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

1"""Authorization and session 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 and authorization options""" 

14 

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

16 """Authorization methods.""" 

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

18 """Authorization providers.""" 

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

20 """Authorization providers.""" 

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

22 """Session options.""" 

23 throttle: Optional[throttle.Config] 

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

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 for meth in self.methods: 

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

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

86 continue 

87 

88 sess = meth.open_session(req) 

89 if not sess: 

90 continue 

91 

92 if not sess.user: 

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

94 self.sessionMgr.delete(sess) 

95 return 

96 

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

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

99 self.sessionMgr.delete(sess) 

100 return 

101 

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

103 return sess 

104 

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

106 sess = req.session 

107 if sess.method: 

108 sess.method.close_session(req, res) 

109 req.set_session(self.guestSession) 

110 

111 def can_use_method(self, req, meth): 

112 if not meth.secure or req.isSecure: 

113 return True 

114 if not meth.allowInsecureFrom: 

115 return False 

116 if req.ip not in meth.allowInsecureFrom: 

117 return False 

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

119 return True 

120 

121 ## 

122 

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

124 return session.Object( 

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

126 method=method, 

127 user=user, 

128 data=data, 

129 is_changed=False, 

130 is_transient=True, 

131 ) 

132 

133 ## 

134 

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

136 if not self.throttle: 

137 return self._authenticate2(method, credentials) 

138 

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

140 if sec > 0: 

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

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

143 exc.retryAfter = sec 

144 raise exc 

145 

146 try: 

147 user = self._authenticate2(method, credentials) 

148 except gws.ForbiddenError: 

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

150 raise 

151 

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

153 return user 

154 

155 def _authenticate2(self, method, credentials): 

156 for prov in self.providers: 

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

158 continue 

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

160 user = prov.authenticate(method, credentials) 

161 if user: 

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

163 return user 

164 

165 ## 

166 

167 def get_user(self, user_uid): 

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

169 prov = self.get_provider(provider_uid) 

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

171 

172 def get_provider(self, uid): 

173 for obj in self.providers: 

174 if obj.uid == uid: 

175 return obj 

176 

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

178 for obj in self.methods: 

179 if obj.uid == uid: 

180 return obj 

181 

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

183 for obj in self.mfAdapters: 

184 if obj.uid == uid: 

185 return obj 

186 

187 def serialize_user(self, user): 

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

189 

190 def unserialize_user(self, data): 

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

192 prov = self.get_provider(provider_uid) 

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

194 

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

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