Password Router¶
The password router is a router use to provide different state of passwords, for example forget password, reset password, etc.
- POST
/forgot_password
- GET
/password
- POST
password
- POST
/password/{token}
- PUT
/password
Setup the Password Service¶
To Setup the Password service, you will need to add all requirements to the
object PasswordService
:
from authx.services.password import PasswordService
from authx.backend import UsersRepo
from authx.core.jwt import JWTBackend
PasswordService.setup(
repo = UsersRepo,
auth_backend = JWTBackend,
debug = True,
base_url= "http://localhost:8000",
site= "http://localhost:8000",
recaptcha_secret = None,
smtp_username= None,
smtp_password= None,
smtp_host= None,
smtp_tls= None,
display_name= "authx",
)
This one gonna help use to use the Password service, that we provide.
from authx import Authentication
from fastapi import FastAPI
app = FastAPI()
auth = Authentication()
app.include_router(auth.password_router, prefix="/api/users")
Forgot Password¶
This is a POST
request to /forgot_password
that will send a request getting
the Data(Login, Password) and IP address of the user, this one work only for
accounts with a pre-configured password.
@router.post("/forgot_password", name="auth:forgot_password")
async def forgot_password(*, request: Request):
data = await request.json()
ip = request.client.host
service = PasswordService()
return await service.forgot_password(data, ip)
Now lets try to understand the concept that we have in
service.forgot_password
:
- This one return a
Response
object, that we can use to send a response to the user. - If the user is not found, we will raise an HTTPException with the status code
404, we could get also
400
if the data is not valid or time is expired.
async def forgot_password(self, data: dict, ip: str) -> None:
try:
email = UserInForgotPassword(**data).email
except ValidationError:
raise HTTPException(400, detail=get_error_message("validation"))
item = await self._repo.get_by_email(email)
if item is None:
raise HTTPException(404, detail=get_error_message("email not found"))
if item.get("password") is None:
raise HTTPException(406)
id = item.get("id")
if not await self._repo.is_password_reset_available(id):
raise HTTPException(400, detail=get_error_message("reset before"))
logger.info(f"forgot_password ip={ip} email={email}")
token = create_random_string()
token_hash = hash_string(token)
await self._repo.set_password_reset_token(id, token_hash)
email_client = self._create_email_client()
await email_client.send_forgot_password_email(email, token)
return None
Password Status¶
Using a GET
Method to get the status of the password, based on the logged
user.
@router.get("/password", name="auth:password_status")
async def password_status(*, user: User = Depends(get_authenticated_user)):
service = PasswordService(user)
return await service.password_status()
This one gonna return a dict where the response gonna show up the status of the password.
Service.password_status
will return a dict with the following keys:
async def password_status(self) -> dict:
status = await self._repo.get_password_status(self._user.id)
return {"status": status}
This one also relate to the UsersPasswordMixin
Class.
Password Set¶
To Set a Password that require a POST
request to /password
that will send a
request with the token and the new password, here we use as a parameter the
request where we have the token and also the Authenticated User (Optional for
Admins).
@router.post("/password", name="auth:password_set")
async def password_set(
*, request: Request,
user: User = Depends(get_authenticated_user)
):
data = await request.json()
service = PasswordService(user)
return await service.password_set(data)
As always we use a Pre-configured Services (Service.password_set
) where we
have 2 ways to set the password.
- First Example:
async def password_set(self, data: dict) -> None:
item = await self._repo.get(self._user.id)
return {
"password": item.get("password") is not None,
"provider": item.get("provider") is not None,
"reset_available": await self._repo.is_password_reset_available(
self._user.id
),
}
- Second Example:
async def password_set(self, data: dict) -> None:
item = await self._repo.get(self._user.id)
if item.get("provider") is not None and item.get("password") is None:
user_model = self._validate_user_model(UserInSetPassword, data)
password_hash = get_password_hash(user_model.password1)
await self._repo.set_password(self._user.id, password_hash)
return None
else:
raise HTTPException(400, get_error_message("password already exists"))
Both of them gonna return a Response
object, that we can use to send a
response to the user.
Password Reset¶
Reset the password using a POST
request to /password/{token}
that will send
a request with the new password, here we use as a parameter the request and a
string token.
@router.post(
"/password/{token}",
name="auth:password_reset")
async def password_reset(*, token: str, request: Request):
data = await request.json()
service = PasswordService()
return await service.password_reset(data, token)
To reset the password we pass data under this format:
data: {password1: "password", password2: "password"}
with the token that we have in the forgot_password
request, to validate and
reset the password.
async def password_reset(self, data: dict, token: str) -> None:
token_hash = hash_string(token)
id = await self._repo.get_id_for_password_reset(token_hash)
if id is None:
raise HTTPException(404)
user_model = self._validate_user_model(UserInSetPassword, data)
password_hash = get_password_hash(user_model.password1)
await self._repo.set_password(id, password_hash)
return None
Warning
You could get the HTTPException if:
* `404` : The token is not found.
* `400` : Token Validation Error.
Password Change¶
After Passing all this steps, Now we can use PUT
to change the password of the
user, this one will send a request with the new password, here we use as a
parameter the request and User (Defaults to Depends get_authenticated_user
).
@router.put("/password", name="auth:password_change")
async def password_change(
*, request: Request, user: User = Depends(get_authenticated_user)
):
data = await request.json()
service = PasswordService(user)
return await service.password_change(data)
return router
Equivalent to password_set
but with the PUT
request, we pass the data under
the same format.
async def password_change(self, data: dict) -> None:
user_model = self._validate_user_model(UserInChangePassword, data)
item = await self._repo.get(self._user.id)
if not verify_password(user_model.old_password, item.get("password")):
raise HTTPException(400, detail=get_error_message("password invalid"))
password_hash = get_password_hash(user_model.password1)
await self._repo.set_password(self._user.id, password_hash)
return None
Warning
You could get the HTTPException if:
* `404` : The token is not found.
* `400` : Token Validation Error.