Coverage for src/lexigram/admin/controllers/auth/password_reset.py: 89%

1from __future__ import annotations 

2 

3import secrets 

4from typing import TYPE_CHECKING 

5from urllib.parse import quote_plus 

6 

7from starlette.requests import Request 

8from starlette.responses import HTMLResponse, RedirectResponse 

9 

10from lexigram.admin.controllers.auth.core import ( 

11 _CACHE_CONTROL_NO_STORE, 

12 AuthCoreMixin, 

13 _humanize_error, 

14 logger, 

15) 

16from lexigram.admin.lib.template import ( 

17 render_password_reset_confirm_page, 

18 render_password_reset_request_page, 

19) 

20from lexigram.contracts.web import get, post 

21 

22if TYPE_CHECKING: 

23 from lexigram.admin.auth.protocols import ( 

24 AdminCsrfServiceProtocol, 

25 AdminPasswordResetServiceProtocol, 

26 ) 

27 

28 

29class AuthPasswordResetMixin(AuthCoreMixin): 

30 """AuthController password_reset endpoints.""" 

31 

32 _csrf_service: AdminCsrfServiceProtocol 

33 _password_reset_service: AdminPasswordResetServiceProtocol | None 

34 

35 @get("/password-reset") 

36 async def password_reset_request_form( 

37 self, request: Request 

38 ) -> HTMLResponse | RedirectResponse: 

39 """Display the standalone password reset request form. 

40 

41 Redirects authenticated users to the admin home. A fresh CSRF 

42 token is embedded for unauthenticated visitors. 

43 

44 Args: 

45 request: Incoming HTTP request. 

46 

47 Returns: 

48 HTMLResponse with the rendered request page, or a 

49 RedirectResponse when already authenticated. 

50 """ 

51 user = getattr(request.state, "user", None) 

52 if user and user.user_id != "guest": 

53 return RedirectResponse(url="/admin/", status_code=302) 

54 

55 error = request.query_params.get("error", "") 

56 sent = request.query_params.get("sent", "") == "1" 

57 

58 csrf_session_id = secrets.token_urlsafe(16) 

59 request.session["csrf_session_id"] = csrf_session_id 

60 csrf_token = self._csrf_service.generate_token(csrf_session_id) 

61 

62 html = render_password_reset_request_page( 

63 error=error, csrf_token=csrf_token, sent=sent 

64 ) 

65 return HTMLResponse(content=html, headers=_CACHE_CONTROL_NO_STORE) 

66 

67 @post("/password-reset") 

68 async def password_reset_request_submit(self, request: Request) -> RedirectResponse: 

69 """Process the password reset request form. 

70 

71 Always redirects to the generic ``sent`` notice — the response is 

72 identical whether or not the email exists (anti-enumeration). 

73 

74 Args: 

75 request: Incoming HTTP request carrying form data. 

76 

77 Returns: 

78 RedirectResponse to the request page with ``sent=1``. 

79 """ 

80 form_data = request.scope.get("admin_form_data") or await request.form() 

81 email = str(form_data.get("email", "")) 

82 csrf_token = str(form_data.get("csrf_token", "")) 

83 

84 csrf_session_id = request.session.get("csrf_session_id", "") 

85 if not csrf_session_id or not self._csrf_service.validate_token( 

86 csrf_session_id, csrf_token 

87 ): 

88 return RedirectResponse( 

89 url=f"/admin/password-reset?error={quote_plus('Invalid or expired security token. Please try again.')}", 

90 status_code=302, 

91 ) 

92 

93 if not email: 

94 return RedirectResponse( 

95 url=f"/admin/password-reset?error={quote_plus('Email is required.')}", 

96 status_code=302, 

97 ) 

98 

99 if self._password_reset_service is not None: 

100 result = await self._password_reset_service.request_reset( 

101 email=email, 

102 ip_address=self._get_client_ip(request), 

103 user_agent=request.headers.get("user-agent", ""), 

104 base_url=str(request.base_url), 

105 ) 

106 if result.is_err(): 

107 return RedirectResponse( 

108 url=f"/admin/password-reset?error={quote_plus(_humanize_error(str(result.unwrap_err())))}", 

109 status_code=302, 

110 ) 

111 return RedirectResponse(url="/admin/password-reset?sent=1", status_code=302) 

112 

113 @get("/password-reset/{token}") 

114 async def password_reset_confirm_form( 

115 self, request: Request 

116 ) -> HTMLResponse | RedirectResponse: 

117 """Display the standalone confirm form for a reset token. 

118 

119 Args: 

120 request: Incoming HTTP request (``token`` from path params). 

121 

122 Returns: 

123 HTMLResponse with the rendered confirm page, or a 

124 RedirectResponse when already authenticated. 

125 """ 

126 user = getattr(request.state, "user", None) 

127 if user and user.user_id != "guest": 

128 return RedirectResponse(url="/admin/", status_code=302) 

129 

130 token = request.path_params.get("token", "") 

131 error = request.query_params.get("error", "") 

132 

133 csrf_session_id = secrets.token_urlsafe(16) 

134 request.session["csrf_session_id"] = csrf_session_id 

135 csrf_token = self._csrf_service.generate_token(csrf_session_id) 

136 

137 html = render_password_reset_confirm_page( 

138 token=token, 

139 error=error, 

140 csrf_token=csrf_token, 

141 password_err=request.query_params.get("password_err", ""), 

142 confirmation_err=request.query_params.get("confirmation_err", ""), 

143 ) 

144 return HTMLResponse(content=html, headers=_CACHE_CONTROL_NO_STORE) 

145 

146 @post("/password-reset/{token}") 

147 async def password_reset_confirm_submit(self, request: Request) -> RedirectResponse: 

148 """Process the new-password form. 

149 

150 Validates CSRF and password/confirmation match, then delegates to 

151 the reset service. On success the user is redirected to the login 

152 page with a success notice; on failure back to the confirm form 

153 with an error message. 

154 

155 Args: 

156 request: Incoming HTTP request carrying form data. 

157 

158 Returns: 

159 RedirectResponse to the login page or back to the confirm form. 

160 """ 

161 token = request.path_params.get("token", "") 

162 form_data = request.scope.get("admin_form_data") or await request.form() 

163 password = str(form_data.get("password", "")) 

164 password_confirmation = str(form_data.get("password_confirmation", "")) 

165 csrf_token = str(form_data.get("csrf_token", "")) 

166 

167 csrf_session_id = request.session.get("csrf_session_id", "") 

168 if not csrf_session_id or not self._csrf_service.validate_token( 

169 csrf_session_id, csrf_token 

170 ): 

171 return RedirectResponse( 

172 url=f"/admin/password-reset/{token}?error={quote_plus('Invalid or expired security token. Please try again.')}", 

173 status_code=302, 

174 ) 

175 

176 if not password or not password_confirmation: 

177 password_err = "New password is required." if not password else "" 

178 confirmation_err = ( 

179 "Please confirm your password." if not password_confirmation else "" 

180 ) 

181 return RedirectResponse( 

182 url=f"/admin/password-reset/{token}?password_err={quote_plus(password_err)}&confirmation_err={quote_plus(confirmation_err)}", 

183 status_code=302, 

184 ) 

185 

186 if password != password_confirmation: 

187 return RedirectResponse( 

188 url=f"/admin/password-reset/{token}?confirmation_err={quote_plus('Passwords do not match.')}", 

189 status_code=302, 

190 ) 

191 

192 if self._password_reset_service is None: 

193 return RedirectResponse( 

194 url=f"/admin/password-reset/{token}?error={quote_plus('Password reset is not available.')}", 

195 status_code=302, 

196 ) 

197 

198 result = await self._password_reset_service.confirm_reset( 

199 token=token, 

200 new_password=password, 

201 ip_address=self._get_client_ip(request), 

202 user_agent=request.headers.get("user-agent", ""), 

203 ) 

204 if result.is_ok(): 

205 logger.info("admin.password_reset_confirm_success", token_prefix=token[:8]) 

206 return RedirectResponse( 

207 url=f"/admin/login?notice={quote_plus('Password reset successful. Please sign in.')}", 

208 status_code=302, 

209 ) 

210 

211 logger.warning( 

212 "admin.password_reset_confirm_failed", error=str(result.unwrap_err()) 

213 ) 

214 return RedirectResponse( 

215 url=f"/admin/password-reset/{token}?error={quote_plus(_humanize_error(str(result.unwrap_err())))}", 

216 status_code=302, 

217 )