1import hmac
2from datetime import UTC, datetime
3from typing import TYPE_CHECKING, Any
4
5from app.users.models import User
6from plain.auth.sessions import login as auth_login
7from plain.auth.sessions import update_session_auth_hash
8from plain.auth.views import AuthView
9from plain.forms import BaseForm
10from plain.http import (
11 BadRequestError400,
12 RedirectResponse,
13)
14from plain.signing import BadSignature, SignatureExpired, TimestampSigner
15from plain.templates.views import CreateView, FormView
16from plain.urls import reverse
17from plain.utils.cache import add_never_cache_headers
18from plain.utils.encoding import force_bytes
19
20from .forms import (
21 PasswordChangeForm,
22 PasswordLoginForm,
23 PasswordResetForm,
24 PasswordSetForm,
25 PasswordSignupForm,
26)
27
28if TYPE_CHECKING:
29 from plain.http import Response
30
31
32class PasswordForgotView(FormView[PasswordResetForm]):
33 form_class = PasswordResetForm
34 reset_confirm_url_name: str
35
36 def generate_password_reset_token(self, user: Any) -> str:
37 return TimestampSigner(salt="password-reset").sign_object(
38 {
39 "id": user.id,
40 "email": user.email,
41 "password": user.password, # Hashed password
42 "timestamp": datetime.now(UTC).timestamp(), # Makes each token unique
43 },
44 compress=True,
45 )
46
47 def generate_password_reset_url(self, user: Any) -> str:
48 token = self.generate_password_reset_token(user)
49 url = reverse(self.reset_confirm_url_name) + f"?token={token}"
50 return self.request.build_absolute_uri(url)
51
52 def form_valid(self, form: PasswordResetForm) -> Response:
53 form.save(
54 generate_reset_url=self.generate_password_reset_url,
55 )
56 return super().form_valid(form)
57
58
59class PasswordResetView(AuthView, FormView[PasswordSetForm]):
60 form_class = PasswordSetForm
61 reset_token_max_age = 60 * 60 # 1 hour
62 _reset_token_session_key = "_password_reset_token"
63
64 def check_password_reset_token(self, token: str) -> User | None:
65 max_age = self.reset_token_max_age
66
67 try:
68 data = TimestampSigner(salt="password-reset").unsign_object(
69 token, max_age=max_age
70 )
71 except SignatureExpired:
72 return None
73 except BadSignature:
74 return None
75
76 try:
77 user = User.query.get(data["id"])
78 except TypeError, ValueError, OverflowError, User.DoesNotExist:
79 return None
80
81 # If the password has changed since the token was generated, the token is invalid.
82 # (These are the hashed passwords, not the raw passwords.)
83 if not hmac.compare_digest(
84 force_bytes(user.password),
85 force_bytes(data["password"]),
86 ):
87 return None
88
89 # If the email has changed since the token was generated, the token is invalid.
90 if not hmac.compare_digest(force_bytes(user.email), force_bytes(data["email"])):
91 return None
92
93 return user
94
95 def get(self) -> Response:
96 if self.user:
97 # Redirect if the user is already logged in
98 return RedirectResponse(
99 str(self.success_url) if self.success_url else "/", status_code=302
100 )
101
102 # Tokens are initially passed as GET parameters and we
103 # immediately store them in the session and remove it from the URL.
104 if token := self.request.query_params.get("token", ""):
105 # Store the token in the session and redirect to the
106 # password reset form at a URL without the token. That
107 # avoids the possibility of leaking the token in the
108 # HTTP Referer header.
109 self.session[self._reset_token_session_key] = token
110 # Redirect to the path itself, without the GET parameters
111 response = RedirectResponse(self.request.path, status_code=302)
112 add_never_cache_headers(response)
113 return response
114
115 return super().get()
116
117 def get_user(self) -> User:
118 session_token = self.session.get(self._reset_token_session_key, "")
119 if not session_token:
120 # No token in the session, so we can't check the password reset token.
121 raise BadRequestError400("No password reset token found.")
122
123 user = self.check_password_reset_token(session_token)
124 if not user:
125 # Remove it from the session if it is invalid.
126 del self.session[self._reset_token_session_key]
127 raise BadRequestError400("Password reset token is no longer valid.")
128
129 return user
130
131 def get_form_kwargs(self) -> dict:
132 kwargs = super().get_form_kwargs()
133 kwargs["user"] = self.get_user()
134 return kwargs
135
136 def form_valid(self, form: PasswordSetForm) -> Response:
137 form.save()
138 del self.session[self._reset_token_session_key]
139 # If you wanted, you could log in the user here so they don't have to
140 # go through the log in form again.
141 return super().form_valid(form)
142
143
144class PasswordChangeView(AuthView, FormView[PasswordChangeForm]):
145 # Change to PasswordSetForm if you want to set new passwords
146 # without confirming the old one.
147 form_class = PasswordChangeForm
148
149 def get_form_kwargs(self) -> dict:
150 kwargs = super().get_form_kwargs()
151 kwargs["user"] = self.user
152 return kwargs
153
154 def form_valid(self, form: PasswordChangeForm) -> Response:
155 form.save()
156 # Updating the password logs out all other sessions for the user
157 # except the current one.
158 update_session_auth_hash(self.request, form.user)
159 return super().form_valid(form)
160
161
162class PasswordLoginView(AuthView, FormView[PasswordLoginForm]):
163 form_class = PasswordLoginForm
164 success_url = "/"
165
166 def get(self) -> Response:
167 # Redirect if the user is already logged in
168 if self.user:
169 return RedirectResponse(self.success_url, status_code=302)
170
171 return super().get()
172
173 def form_valid(self, form: PasswordLoginForm) -> Response:
174 # Log the user in and redirect
175 auth_login(self.request, form.get_user())
176
177 return super().form_valid(form)
178
179
180class PasswordSignupView(CreateView):
181 form_class = PasswordSignupForm
182 success_url = "/"
183
184 def form_valid(self, form: BaseForm) -> Response:
185 # # Log the user in and redirect
186 # auth_login(self.request, form.create())
187
188 return super().form_valid(form)