Skip to content

Member API

The member API will need to provide routes to create members, and update aspects e.g. their password.

Creating the blueprint

The blueprint itself can be created with the following code in backend/src/backend/blueprints/members.py,

from quart import Blueprint

blueprint = Blueprint("members", __name__)
and activated by adding the following to backend/src/backend/run.py,

from backend.blueprints.members import blueprint as members_blueprint

def create_app() -> Quart:
    ...
    app.register_blueprint(members_blueprint)

Creating a member

For a RESTFul API the member creation route should use the POST method and require an email and a password. In the route we should check the passwod is sufficiently complex (see password strength), create a new member and send a welcome email. We can include a verification token in the welcome email in order for the email address owner to prove they signed up. We can do this by first creating an email template by adding the following to backend/src/backend/templates/emails/welcome.md,

Hello and welcome to tozo!

Please confirm you signed up by following this
[link]({{ config["BASE_URL"] }}/confirm-email/{{ token }}/).

Thanks

The route should return 201 on success, and if the email provided is already a member to prevent user enumeration. This is all achieved by adding the following to backend/src/backend/blueprints/members.py,

from dataclasses import dataclass
from datetime import timedelta

import asyncpg
import bcrypt
from itsdangerous import URLSafeTimedSerializer
from quart import current_app, render_template, ResponseReturnValue
from quart_schema import validate_request
from quart_rate_limiter import rate_limit
from zxcvbn import zxcvbn

from lib.api_error import APIError
from models.members import insert_member

MINIMUM_STRENGTH = 3
EMAIL_VERIFICATION_SALT = "email verify"

@dataclass
class MemberData:
    email: str
    password: str

@blueprint.post("/members/")
@rate_limit(10, timedelta(seconds=10))
@validate_request(MemberData)
async def register(data: MemberData) -> ResponseReturnValue:
    """Create a new Member.

    This allows a Member to be created.
    """
    strength = zxcvbn(data.password)
    if strength["score"] < MINIMUM_STRENGTH:
        raise APIError(400, "WEAK_PASSWORD")

    salt = bcrypt.gensalt(14)
    hashed_password = bcrypt.hashpw(data.password.encode("utf-8"), salt)

    try:
        member = await insert_member(
            current_app.db, data.email, hashed_password.decode()
        )
    except asyncpg.exceptions.UniqueViolationError:
        pass
    else:
        serializer = URLSafeTimedSerializer(
            current_app.secret_key, salt=EMAIL_VERIFICATION_SALT
        )
        token = serializer.dumps(member.id)
        body = await render_template("emails/welcome.md", token=token)
        await current_app.mail_client.send(
            member.email, "Welcome", body, "WELCOME"
        )

    return {}, 201

Changing password

A user will want to change their password which will require a route that accepts their new password whilst checking the old password also supplied is correct. This route should also inform the user that the password has changed, by adding the following to backend/src/backend/templates/emails/password_changed.md,

Hello,

Your Tozo password has been successfully changed.

Thanks

For a RESTFul API the change member password should be PUT, returning 200 on success (we'll use the same rate limit). The following should be merged (removing duplicated imports) to backend/src/backend/blueprints/members.py,

from quart_auth import current_user, login_required

from backend.models.member import select_member_by_id, update_member_password

@dataclass
class PasswordData:
    current_password: str
    new_password: str

@blueprint.put("/members/password/")
@rate_limit(5, timedelta(minutes=1))
@login_required
@validate_request(PasswordData)
async def change_password(data: PasswordData) -> ResponseReturnValue:
    """Update the members password.

    This allows the user to update their password.
    """
    strength = zxcvbn(data.password)
    if strength["score"] < MINIMUM_STRENGTH:
        raise APIError(400, "WEAK_PASSWORD")

    member = await select_member_by_id(current_app.db, int(current_user.auth_id))
    passwords_match = bcrypt.checkpw(
        data.current_password.encode("utf-8"),
        member.password_hash.encode("utf-8"),
    )
    if not passwords_match:
        raise APIError(401, "INVALID_PASSWORD")

    salt = bcrypt.gensalt(14)
    hashed_password = bcrypt.hashpw(data.new_password.encode("utf-8"), salt)

    await update_member_password(
        current_app.db, int(current_user.auth_id), hashed_password
    )
    body = await render_template("emails/password_changed.md")
    await current_app.mail_client.send(
        member.email, "Password changed", body, "PASSWORD_CHANGED"
    )
    return {}

Confirming the email address

When a user creates their account they are sent a link back to the tozo app that includes a token. This route should test the token and if valid confirm the email address. This is achieved by adding the following to backend/src/backend/blueprints/members.py,

from backend.models.member import update_member_email_verified

@dataclass
class TokenData:
    token: str

@blueprint.put("/members/email/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(TokenData)
async def verify_email(data: TokenData) -> ResponseReturnValue:
    """Call to verify an email.

    This requires the user to supply a valid token.
    """
    serializer = URLSafeTimedSerializer(current_app.secret_key, salt=EMAIL_VERIFICATION_SALT)
    try:
        member_id = serializer.loads(data.token, max_age=ONE_MONTH)
    except SignatureExpired:
        raise APIError(403, "TOKEN_EXPIRED")
    except BadSignature:
        raise APIError(400, "TOKEN_INVALID")
    else:
        await update_member_email_verified(current_app.db, member_id)
    return {}

Forgotten password

If a user forgets their password they'll want us to send a password reset link to their email address. To do so we need a route that accepts the user's email address and sends out the following email, as added to backend/src/backend/templates/emails/forgotten_password.md,

Hello,

You can use this [link]({{ config["BASE_URL"] }}/reset-password/{{ token }}/)
to reset your password.

Thanks
from backend.models.member import select_member_by_email

FORGOTTEN_PASSWORD_SALT = "forgotten password"

@dataclass
class ForgottenPasswordData:
    email: str

@blueprint.put("/members/email/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(ForgottenPasswordData)
async def forgotten_password(data: ForgottenPasswordData) -> ResponseReturnValue:
    """Call to trigger a forgotten password email.

    This requires a valid member email.
    """
    member = await select_member_by_email(current_app.db, data.email)
    if member is not None:
        serializer = URLSafeTimedSerializer(
            current_app.secret_key, salt=FORGOTTEN_PASSWORD_SALT
        )
        token = serializer.dumps(member.id)
        body = await render_template("emails/forgotten_password.md", token=token)
        await current_app.mail_client.send(
            member.email, "Forgotten password", body, "FORGOTTEN_PASSWORD"
        )
    return {}

Reseting the password

PUT "/members/reset-password/"

@dataclass
class ResetPasswordData:
    password: str
    token: str

@blueprint.put("/members/reset-password/")
@rate_limit(5, timedelta(minutes=1))
@validate_request(ResetPasswordData)
async def reset_password(data: ResetPasswordData) -> ResponseReturnValue:
    """Call to reset a password using a token.

    This requires the user to supply a valid token and a
    new password.
    """
    serializer = URLSafeTimedSerializer(
        current_app.secret_key, salt=FORGOTTEN_PASSWORD_SALT
    )
    try:
        member_id = serializer.loads(data.token, max_age=ONE_MONTH)
    except SignatureExpired:
        raise APIError(403, "TOKEN_EXPIRED")
    except BadSignature:
        raise APIError(400, "TOKEN_INVALID")
    else:
        strength = zxcvbn(data.password)
        if strength["score"] < MINIMUM_STRENGTH:
            raise APIError(400, "WEAK_PASSWORD")

        salt = bcrypt.gensalt(14)
        hashed_password = bcrypt.hashpw(data.password.encode("utf-8"), salt)

        await update_member_password(
            current_app.db, member_id, hashed_password.decode()
        )
    return {}