Coverage for gws-app/gws/plugin/auth_method/web/core.py: 78%

1"""Web authorisation method.""" 

2 

3import re 

4from typing import Optional, cast 

5 

6import gws 

7import gws.base.auth 

8import gws.base.web 

9 

10gws.ext.new.authMethod('web') 

11 

12 

13class LoginRedirectRule(gws.Data): 

14 """Login redirect rule.""" 

15 

16 pattern: Optional[gws.Regex] 

17 """URL matching pattern for pages that must be redirected.""" 

18 target: str 

19 """Target url.""" 

20 

21 

22class Config(gws.base.auth.method.Config): 

23 """Web-based authorization options""" 

24 

25 cookieName: str = 'auth' 

26 """Name for the cookie.""" 

27 cookiePath: str = '/' 

28 """Cookie path.""" 

29 cookieSameSite: str = 'Lax' 

30 """Cookie SameSite attribute.""" 

31 loginRedirect: Optional[LoginRedirectRule] 

32 """Rule to redirect if a login is required.""" 

33 

34 

35## 

36 

37 

38class UserResponse(gws.Response): 

39 user: Optional[gws.base.auth.user.Props] 

40 

41 

42class LogoutResponse(gws.Response): 

43 pass 

44 

45 

46class LoginRequest(gws.Request): 

47 username: str 

48 password: str 

49 to: Optional[str] 

50 

51 

52class LoginResponse(gws.Response): 

53 user: Optional[gws.base.auth.user.Props] 

54 mfaState: Optional[gws.AuthMultiFactorState] 

55 mfaMessage: str = '' 

56 mfaCanRestart: bool = False 

57 redirectTo: str = '' 

58 

59 

60class MfaVerifyRequest(gws.Request): 

61 payload: dict 

62 to: Optional[str] 

63 

64 

65## 

66 

67_DELETED_SESSION = 'web:deleted' 

68 

69 

70class Object(gws.base.auth.method.Object): 

71 cookieName: str 

72 cookiePath: str 

73 cookieSameSite: str 

74 loginRedirect: Optional[LoginRedirectRule] 

75 

76 deletedSession: gws.base.auth.session.Object 

77 

78 def configure(self): 

79 self.uid = 'gws.plugin.auth.method.web' 

80 self.cookieName = self.cfg('cookieName', default=Config.cookieName) 

81 self.cookiePath = self.cfg('cookiePath', default=Config.cookiePath) 

82 self.cookieSameSite = self.cfg('cookieSameSite', default=Config.cookieSameSite) 

83 self.loginRedirect = self.cfg('loginRedirect') 

84 self.root.app.middlewareMgr.register(self, self.uid, depends_on=['auth']) 

85 

86 ## 

87 

88 def exit_middleware(self, req, res): 

89 if res.status in (403, 401) and req.isGet: 

90 self._check_login_redirect(req, res) 

91 

92 def _check_login_redirect(self, req: gws.WebRequester, res: gws.WebResponder): 

93 lr = self.loginRedirect 

94 if not lr: 

95 return 

96 request_uri = req.env('REQUEST_URI', '') 

97 if not request_uri: 

98 return 

99 if lr.pattern and not re.match(lr.pattern, request_uri): 

100 return 

101 redir = req.relative_url_for(lr.target, to=request_uri) 

102 res.set_status(302) 

103 res.add_header('Location', redir) 

104 res.set_body(f'Redirecting to {redir}...') 

105 gws.log.debug(f'auth web: redirect {res.status=} {redir=}') 

106 

107 def activate(self): 

108 am = self.root.app.authMgr 

109 self.deletedSession = gws.base.auth.session.Object( 

110 uid=_DELETED_SESSION, 

111 method=self, 

112 user=am.guestUser, 

113 ) 

114 

115 def open_session(self, req): 

116 am = self.root.app.authMgr 

117 

118 sid = req.cookie(self.cookieName) 

119 if not sid: 

120 return 

121 

122 sess = am.sessionMgr.get(sid) 

123 

124 if not sess: 

125 gws.log.debug(f'open_session: {sid=} not found or invalid') 

126 return self.deletedSession 

127 

128 return sess 

129 

130 def close_session(self, req, res): 

131 am = self.root.app.authMgr 

132 

133 sess = getattr(req, 'session') 

134 if not sess: 

135 return 

136 

137 if sess.uid == _DELETED_SESSION: 

138 gws.log.debug('session cookie=deleted') 

139 res.delete_cookie( 

140 self.cookieName, 

141 path=self.cookiePath, 

142 ) 

143 return 

144 

145 if res.status < 400: 

146 gws.log.debug(f'session cookie={sess.uid!r}') 

147 res.set_cookie( 

148 self.cookieName, 

149 sess.uid, 

150 path=self.cookiePath, 

151 secure=self.secure, 

152 samesite=self.cookieSameSite, 

153 httponly=True, 

154 ) 

155 am.sessionMgr.touch(sess) 

156 

157 def handle_login(self, req: gws.WebRequester, p: LoginRequest) -> LoginResponse: 

158 if not req.user.isGuest: 

159 raise gws.ForbiddenError(f'login: already logged-in {req.user.uid=}') 

160 

161 if self.secure and not req.isSecure: 

162 raise gws.ForbiddenError('login: insecure_context, ignored') 

163 

164 user = self.root.app.authMgr.authenticate(self, p, req) 

165 if not user: 

166 raise gws.ForbiddenError('login: user not found') 

167 

168 if user.mfaUid: 

169 mfa = self._mfa_start(req, user) 

170 gws.log.info(f'LOGGED_IN (MFA pending): {user.uid=} {user.roles=}') 

171 return self._mfa_response(mfa) 

172 

173 self._finalize_login(req, user) 

174 return LoginResponse(user=gws.props_of(user, user), redirectTo=self._redirect_target(p.to)) 

175 

176 def handle_mfa_verify(self, req: gws.WebRequester, p: MfaVerifyRequest) -> LoginResponse: 

177 try: 

178 mfa = self._mfa_verify(req, p.payload) 

179 except gws.ForbiddenError: 

180 self._delete_session(req) 

181 raise 

182 

183 if mfa.state == gws.AuthMultiFactorState.ok: 

184 self._finalize_login(req, mfa.user) 

185 return self._mfa_response(mfa, self._redirect_target(p.to)) 

186 

187 if mfa.state == gws.AuthMultiFactorState.retry: 

188 return self._mfa_response(mfa) 

189 

190 self._delete_session(req) 

191 raise gws.ForbiddenError(f'MFA: verify failed {mfa.state=}') 

192 

193 def handle_mfa_restart(self, req: gws.WebRequester, p: gws.Request) -> LoginResponse: 

194 try: 

195 mfa = self._mfa_restart(req) 

196 except gws.ForbiddenError: 

197 self._delete_session(req) 

198 raise 

199 

200 return self._mfa_response(mfa) 

201 

202 def handle_logout(self, req: gws.WebRequester) -> LogoutResponse: 

203 if req.user.isGuest: 

204 self._delete_session(req) 

205 return LogoutResponse() 

206 

207 if req.session.method != self: 

208 raise gws.ForbiddenError(f'wrong method for logout: {req.session.method!r}') 

209 

210 self._delete_session(req) 

211 

212 gws.log.info(f'LOGGED_OUT: user={req.user.uid!r}') 

213 return LogoutResponse() 

214 

215 ## 

216 

217 def _delete_session(self, req: gws.WebRequester): 

218 am = self.root.app.authMgr 

219 am.sessionMgr.delete(req.session) 

220 req.set_session(self.deletedSession) 

221 

222 def _finalize_login(self, req: gws.WebRequester, user: gws.User): 

223 self._delete_session(req) 

224 am = self.root.app.authMgr 

225 req.set_session(am.sessionMgr.create(self, user)) 

226 gws.log.info(f'LOGGED_IN: {user.uid=} {user.roles=}') 

227 

228 def _redirect_target(self, s: str | None) -> str: 

229 """Convert a client-provided redirect target to a local url path.""" 

230 

231 s = (s or '').strip() 

232 if not s: 

233 return '' 

234 if '\\' in s or any(c < ' ' or c == '\x7f' for c in s): 

235 return '' 

236 return '/' + s.lstrip('/') 

237 

238 ## 

239 

240 def _mfa_start(self, req: gws.WebRequester, user: gws.User) -> gws.AuthMultiFactorTransaction: 

241 am = self.root.app.authMgr 

242 

243 adapter = am.get_multi_factor_adapter(user.mfaUid) 

244 if not adapter: 

245 raise gws.ForbiddenError(f'MFA: {user.mfaUid=} unknown') 

246 

247 mfa = adapter.start(user) 

248 if not mfa: 

249 raise gws.ForbiddenError(f'MFA: {user.mfaUid=} start failed') 

250 

251 req.set_session(am.sessionMgr.create(self, am.guestUser)) 

252 

253 self._mfa_store(req, mfa) 

254 return mfa 

255 

256 def _mfa_verify(self, req: gws.WebRequester, payload: dict) -> gws.AuthMultiFactorTransaction: 

257 mfa = self._mfa_load(req) 

258 mfa = mfa.adapter.verify(mfa, payload) 

259 

260 self._mfa_store(req, mfa) 

261 return mfa 

262 

263 def _mfa_restart(self, req: gws.WebRequester) -> gws.AuthMultiFactorTransaction: 

264 mfa = self._mfa_load(req) 

265 mfa = mfa.adapter.restart(mfa) 

266 if not mfa: 

267 raise gws.ForbiddenError(f'MFA: restart failed') 

268 

269 self._mfa_store(req, mfa) 

270 return mfa 

271 

272 def _mfa_store(self, req: gws.WebRequester, mfa: gws.AuthMultiFactorTransaction): 

273 am = self.root.app.authMgr 

274 

275 sess_mfa = gws.u.merge({}, mfa) 

276 sess_mfa['user'] = am.serialize_user(mfa.user) 

277 sess_mfa['adapter'] = mfa.adapter.uid 

278 req.session.set('AuthMultiFactorTransaction', sess_mfa) 

279 

280 def _mfa_load(self, req: gws.WebRequester) -> gws.AuthMultiFactorTransaction: 

281 am = self.root.app.authMgr 

282 

283 sess_mfa = req.session.get('AuthMultiFactorTransaction') 

284 if not sess_mfa: 

285 raise gws.ForbiddenError(f'MFA: transaction not found') 

286 

287 mfa = gws.AuthMultiFactorTransaction(sess_mfa) 

288 mfa.adapter = gws.u.require(am.get_multi_factor_adapter(sess_mfa['adapter'])) 

289 mfa.user = gws.u.require(am.unserialize_user(sess_mfa['user'])) 

290 

291 if not mfa.adapter.check_state(mfa): 

292 raise gws.ForbiddenError(f'MFA: invalid transaction in session') 

293 

294 return mfa 

295 

296 def _mfa_response(self, mfa: gws.AuthMultiFactorTransaction, redirect_to: str = '') -> LoginResponse: 

297 return LoginResponse( 

298 mfaState=mfa.state, 

299 mfaMessage=mfa.message, 

300 mfaCanRestart=mfa.adapter.check_restart(mfa), 

301 redirectTo=redirect_to, 

302 )