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

1"""The ``web`` authentication method and its API types.""" 

2 

3import re 

4from typing import Optional, cast 

5 

6import gws 

7import gws.base.auth 

8import gws.base.web 

9 

10 

11class LoginRedirectRule(gws.Data): 

12 """Redirect to a login page for page requests that are denied.""" 

13 

14 pattern: Optional[gws.Regex] 

15 """Regular expression for URLs to redirect.""" 

16 target: str 

17 """Login page URL.""" 

18 

19 

20@gws.ext.config.authMethod('web') 

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

22 """Authentication with a login form and a session cookie.""" 

23 

24 cookieName: str = 'auth' 

25 """Name of the session cookie.""" 

26 cookiePath: str = '/' 

27 """Path attribute of the session cookie.""" 

28 cookieSameSite: str = 'Lax' 

29 """SameSite attribute of the session cookie.""" 

30 loginRedirect: Optional[LoginRedirectRule] 

31 """Redirect denied page requests to a login page.""" 

32 

33 

34## 

35 

36 

37class UserResponse(gws.Response): 

38 """Response with the current user.""" 

39 

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

41 """Properties of the user, ``None`` for guests.""" 

42 

43 

44class LogoutResponse(gws.Response): 

45 """Response to a logout request.""" 

46 

47 pass 

48 

49 

50class LoginRequest(gws.Request): 

51 """Login request.""" 

52 

53 username: str 

54 """Login name.""" 

55 password: str 

56 """Password.""" 

57 to: Optional[str] 

58 """URL path to return to after the login.""" 

59 

60 

61class LoginResponse(gws.Response): 

62 """Response to a login or multi-factor request.""" 

63 

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

65 """Properties of the logged-in user, set when the login is completed without a second step.""" 

66 mfaState: Optional[gws.AuthMultiFactorState] 

67 """State of the multi-factor transaction, if there is one.""" 

68 mfaMessage: str = '' 

69 """Message for the user in the multi-factor step.""" 

70 mfaCanRestart: bool = False 

71 """The multi-factor transaction can be restarted.""" 

72 redirectTo: str = '' 

73 """URL path to go to after a completed login, empty if none.""" 

74 

75 

76class MfaVerifyRequest(gws.Request): 

77 """Request to verify a multi-factor code.""" 

78 

79 payload: dict 

80 """Data entered by the user, for example ``{"code": "123456"}``.""" 

81 to: Optional[str] 

82 """URL path to return to after the login.""" 

83 

84 

85## 

86 

87_DELETED_SESSION = 'web:deleted' 

88 

89 

90@gws.ext.object.authMethod('web') 

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

92 """Web authentication method. 

93 

94 Authenticates users with a login form and keeps their sessions in the 

95 session manager, identified by a session cookie. Handles logins, logouts 

96 and multi-factor verification for the ``auth`` action. 

97 """ 

98 

99 cookieName: str 

100 """Name of the session cookie.""" 

101 cookiePath: str 

102 """Path attribute of the session cookie.""" 

103 cookieSameSite: str 

104 """SameSite attribute of the session cookie.""" 

105 loginRedirect: Optional[LoginRedirectRule] 

106 """Redirect rule for denied page requests, or ``None``.""" 

107 

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

109 """Placeholder session for requests with an invalid session cookie.""" 

110 

111 def configure(self): 

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

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

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

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

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

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

118 

119 ## 

120 

121 def exit_middleware(self, req, res): 

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

123 self._check_login_redirect(req, res) 

124 

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

126 """Redirect a denied page request to the login page, if configured.""" 

127 lr = self.loginRedirect 

128 if not lr: 

129 return 

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

131 if not request_uri: 

132 return 

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

134 return 

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

136 res.set_status(302) 

137 res.add_header('Location', redir) 

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

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

140 

141 def activate(self): 

142 am = self.root.app.authMgr 

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

144 uid=_DELETED_SESSION, 

145 method=self, 

146 user=am.guestUser, 

147 ) 

148 

149 def open_session(self, req): 

150 am = self.root.app.authMgr 

151 

152 sid = req.cookie(self.cookieName) 

153 if not sid: 

154 return 

155 

156 sess = am.sessionMgr.get(sid) 

157 

158 if not sess: 

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

160 return self.deletedSession 

161 

162 return sess 

163 

164 def close_session(self, req, res): 

165 am = self.root.app.authMgr 

166 

167 sess = getattr(req, 'session') 

168 if not sess: 

169 return 

170 

171 if sess.uid == _DELETED_SESSION: 

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

173 res.delete_cookie( 

174 self.cookieName, 

175 path=self.cookiePath, 

176 ) 

177 return 

178 

179 if res.status < 400: 

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

181 res.set_cookie( 

182 self.cookieName, 

183 sess.uid, 

184 path=self.cookiePath, 

185 secure=self.secure, 

186 samesite=self.cookieSameSite, 

187 httponly=True, 

188 ) 

189 am.sessionMgr.touch(sess) 

190 

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

192 """Log in with a username and a password. 

193 

194 The credentials are checked by the authentication providers. If the 

195 user has an ``mfaUid``, a multi-factor transaction is started and kept 

196 in a new guest session; ``handle_mfa_verify`` completes the login. 

197 Otherwise the current session is replaced by a new session for the user. 

198 

199 Args: 

200 req: Web requester. 

201 p: Login request. 

202 

203 Returns: 

204 The user and the redirect target, or the multi-factor state. 

205 

206 Raises: 

207 ``gws.ForbiddenError``: If the user is already logged in, the method is secure and the request is not, or the multi-factor transaction cannot be started. 

208 ``gws.AuthenticationError``: If the credentials are not accepted. 

209 """ 

210 if not req.user.isGuest: 

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

212 

213 if self.secure and not req.isSecure: 

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

215 

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

217 if not user: 

218 raise gws.AuthenticationError('login: user not found') 

219 

220 if user.mfaUid: 

221 mfa = self._mfa_start(req, user) 

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

223 return self._mfa_response(mfa) 

224 

225 self._finalize_login(req, user) 

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

227 

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

229 """Verify a multi-factor payload. 

230 

231 On success, the session is replaced by a new session for the user. If 

232 the adapter allows another attempt, the transaction is kept. Otherwise 

233 the session is deleted. 

234 

235 Args: 

236 req: Web requester. 

237 p: Verification request. 

238 

239 Returns: 

240 The multi-factor state, with the redirect target on success. 

241 

242 Raises: 

243 ``gws.ForbiddenError``: If the session has no valid multi-factor transaction. 

244 ``gws.AuthenticationError``: If the verification failed. 

245 """ 

246 try: 

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

248 except gws.ForbiddenError: 

249 self._delete_session(req) 

250 raise 

251 

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

253 self._finalize_login(req, mfa.user) 

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

255 

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

257 return self._mfa_response(mfa) 

258 

259 self._delete_session(req) 

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

261 

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

263 """Restart the multi-factor transaction of the current session. 

264 

265 Args: 

266 req: Web requester. 

267 p: Request parameters. 

268 

269 Returns: 

270 The state of the new transaction. 

271 

272 Raises: 

273 ``gws.ForbiddenError``: If the session has no valid transaction or it cannot be restarted. The session is deleted in this case. 

274 """ 

275 try: 

276 mfa = self._mfa_restart(req) 

277 except gws.ForbiddenError: 

278 self._delete_session(req) 

279 raise 

280 

281 return self._mfa_response(mfa) 

282 

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

284 """Log out the current user and delete the session. 

285 

286 Args: 

287 req: Web requester. 

288 

289 Returns: 

290 An empty response. 

291 

292 Raises: 

293 ``gws.ForbiddenError``: If the session was opened by a different method. 

294 """ 

295 if req.user.isGuest: 

296 self._delete_session(req) 

297 return LogoutResponse() 

298 

299 if req.session.method != self: 

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

301 

302 self._delete_session(req) 

303 

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

305 return LogoutResponse() 

306 

307 ## 

308 

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

310 """Delete the current session and replace it with the placeholder session.""" 

311 am = self.root.app.authMgr 

312 am.sessionMgr.delete(req.session) 

313 req.set_session(self.deletedSession) 

314 

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

316 """Replace the current session with a new session for the user.""" 

317 self._delete_session(req) 

318 am = self.root.app.authMgr 

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

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

321 

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

323 """Convert a redirect target sent by the client to a local URL path.""" 

324 

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

326 if not s: 

327 return '' 

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

329 return '' 

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

331 

332 ## 

333 

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

335 """Start a multi-factor transaction and keep it in a new guest session.""" 

336 am = self.root.app.authMgr 

337 

338 adapter = am.get_multi_factor_adapter(user.mfaUid) 

339 if not adapter: 

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

341 

342 mfa = adapter.start(user) 

343 if not mfa: 

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

345 

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

347 

348 self._mfa_store(req, mfa) 

349 return mfa 

350 

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

352 """Verify a payload against the transaction stored in the session.""" 

353 mfa = self._mfa_load(req) 

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

355 

356 self._mfa_store(req, mfa) 

357 return mfa 

358 

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

360 """Restart the transaction stored in the session.""" 

361 mfa = self._mfa_load(req) 

362 mfa = mfa.adapter.restart(mfa) 

363 if not mfa: 

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

365 

366 self._mfa_store(req, mfa) 

367 return mfa 

368 

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

370 """Store the transaction in the session.""" 

371 am = self.root.app.authMgr 

372 

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

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

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

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

377 

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

379 """Load the transaction from the session and check its state.""" 

380 am = self.root.app.authMgr 

381 

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

383 if not sess_mfa: 

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

385 

386 mfa = gws.AuthMultiFactorTransaction(sess_mfa) 

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

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

389 

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

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

392 

393 return mfa 

394 

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

396 """Create a login response from a multi-factor transaction.""" 

397 return LoginResponse( 

398 mfaState=mfa.state, 

399 mfaMessage=mfa.message, 

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

401 redirectTo=redirect_to, 

402 )