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