path: root/gn_auth/auth/authorisation/users/views.py

"""User authorisation endpoints."""
import sqlite3
import secrets
import datetime
import traceback
from typing import Any
from functools import partial
from dataclasses import asdict
from email.headerregistry import Address
from email_validator import validate_email, EmailNotValidError
from flask import (
    request,
    jsonify,
    Response,
    Blueprint,
    current_app,
    render_template)

from gn_auth.smtp import send_message, build_email_message

from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.db.sqlite3 import with_db_connection

from gn_auth.auth.authorisation.resources.models import (
    user_resources as _user_resources)
from gn_auth.auth.authorisation.roles.models import (
    assign_default_roles, user_roles as _user_roles)
from gn_auth.auth.authorisation.resources.groups.models import (
    user_group as _user_group)

from gn_auth.auth.errors import (
    NotFoundError,
    UsernameError,
    PasswordError,
    UserVerificationError,
    UserRegistrationError)

from gn_auth.auth.authentication.users import valid_login, user_by_email
from gn_auth.auth.authentication.oauth2.resource_server import require_oauth
from gn_auth.auth.authentication.users import User, save_user, set_user_password
from gn_auth.auth.authentication.oauth2.models.oauth2token import (
    token_by_access_token)

from .models import list_users
from .masquerade.views import masq
from .collections.views import collections

users = Blueprint("users", __name__)
users.register_blueprint(masq, url_prefix="/masquerade")
users.register_blueprint(collections, url_prefix="/collections")

@users.route("/", methods=["GET"])
@require_oauth("profile")
def user_details() -> Response:
    """Return user's details."""
    with require_oauth.acquire("profile") as the_token:
        user = the_token.user
        user_dets = {
            "user_id": user.user_id, "email": user.email, "name": user.name,
            "group": False
        }
        with db.connection(current_app.config["AUTH_DB"]) as conn:
            the_group = _user_group(conn, user).maybe(# type: ignore[misc]
                False, lambda grp: grp)# type: ignore[arg-type]
            return jsonify({
                **user_dets,
                "group": asdict(the_group) if the_group else False
            })

@users.route("/roles", methods=["GET"])
@require_oauth("role")
def user_roles() -> Response:
    """Return the non-resource roles assigned to the user."""
    with require_oauth.acquire("role") as token:
        with db.connection(current_app.config["AUTH_DB"]) as conn:
            return jsonify(tuple(
                {**role, "roles": tuple(asdict(rol) for rol in role["roles"])}
                for role in _user_roles(conn, token.user)))

def validate_password(password, confirm_password) -> str:
    """Validate the provided password."""
    if len(password) < 8:
        raise PasswordError("The password must be at least 8 characters long.")

    if password != confirm_password:
        raise PasswordError("Mismatched password values")

    return password

def validate_username(name: str) -> str:
    """Validate the provides name."""
    if name == "":
        raise UsernameError("User's name not provided.")

    return name

def __assert_not_logged_in__(conn: db.DbConnection):
    bearer = request.headers.get('Authorization')
    if bearer:
        token = token_by_access_token(conn, bearer.split(None)[1]).maybe(# type: ignore[misc]
            False, lambda tok: tok)
        if token:
            raise UserRegistrationError(
                "Cannot register user while authenticated")

def user_address(user: User) -> Address:
    """Compute the `email.headerregistry.Address` from a `User`"""
    return Address(display_name=user.name, addr_spec=user.email)

def send_verification_email(conn, user: User) -> None:
    """Send an email verification message."""
    subject="GeneNetwork: Please Verify Your Email"
    verification_code = secrets.token_urlsafe(64)
    generated = datetime.datetime.now()
    expiration_minutes = 15
    def __render__(template):
        return render_template(template,
                               subject=subject,
                               verification_code=verification_code,
                               verification_uri="https://please/change/this/",
                               expiration_minutes=expiration_minutes)
    with db.cursor(conn) as cursor:
        cursor.execute(
            ("INSERT INTO "
             "user_verification_codes(user_id, code, generated, expires) "
             "VALUES (:user_id, :code, :generated, :expires)"),
            {
                "user_id": str(user.user_id),
                "code": verification_code,
                "generated": int(generated.timestamp()),
                "expires": int(
                    (generated +
                     datetime.timedelta(
                         minutes=expiration_minutes)).timestamp())
            })
        send_message(smtp_user=current_app.config["SMTP_USER"],
                     smtp_passwd=current_app.config["SMTP_PASSWORD"],
                     message=build_email_message(
                         to_addresses=(user_address(user),),
                         subject=subject,
                         txtmessage=__render__("emails/verify-email.txt"),
                         htmlmessage=__render__("emails/verify-email.html")),
                     host=current_app.config["SMTP_HOST"],
                     port=current_app.config["SMTP_PORT"])

@users.route("/register", methods=["POST"])
def register_user() -> Response:
    """Register a user."""
    with db.connection(current_app.config["AUTH_DB"]) as conn:
        __assert_not_logged_in__(conn)

        try:
            form = request.form
            email = validate_email(form.get("email", "").strip(),
                                   check_deliverability=True)
            password = validate_password(
                form.get("password", "").strip(),
                form.get("confirm_password", "").strip())
            user_name = validate_username(form.get("user_name", "").strip())
            with db.cursor(conn) as cursor:
                user, _hashed_password = set_user_password(
                    cursor, save_user(
                        cursor, email["email"], user_name), password)
                assign_default_roles(cursor, user)
                send_verification_email(conn, user)
                return jsonify(asdict(user))
        except sqlite3.IntegrityError as sq3ie:
            current_app.logger.debug(traceback.format_exc())
            raise UserRegistrationError(
                "A user with that email already exists") from sq3ie
        except EmailNotValidError as enve:
            current_app.logger.debug(traceback.format_exc())
            raise(UserRegistrationError(f"Email Error: {str(enve)}")) from enve

    raise Exception(
        "unknown_error", "The system experienced an unexpected error.")

def delete_verification_code(cursor, code: str):
    """Delete verification code from db."""
    cursor.execute("DELETE FROM user_verification_codes "
                   "WHERE code=:code",
                   {"code": code})

@users.route("/verify", methods=["GET", "POST"])
def verify_user():
    """Verify users are not bots."""
    code = request.args.get("verificationcode",
                            request.form.get("verificationcode",
                                             "nosuchcode"))
    with (db.connection(current_app.config["AUTH_DB"]) as conn,
          db.cursor(conn) as cursor):
        cursor.execute("SELECT * FROM user_verification_codes "
                       "WHERE code=:code",
                       {"code": code})
        results = tuple(dict(row) for row in cursor.fetchall())

        if not bool(results):
            raise UserVerificationError(
                "Invalid verification code: code not found.")

        if len(results) > 1:
            delete_verification_code(cursor, code)
            raise UserVerificationError(
                "Invalid verification code: code is duplicated.")

        results = results[0]
        if (datetime.datetime.fromtimestamp(
                int(results["expires"])) < datetime.datetime.now()):
            delete_verification_code(cursor, code)
            raise UserVerificationError(
                "Invalid verification code: code has expired.")

        # Code is good!
        delete_verification_code(cursor, code)
        cursor.execute("UPDATE users SET verified=1 WHERE user_id=:user_id",
                       {"user_id": results["user_id"]})
        return jsonify({
            "status": "success",
            "message": "User verification successful!"
        })


@users.route("/group", methods=["GET"])
@require_oauth("profile group")
def user_group() -> Response:
    """Retrieve the group in which the user is a member."""
    with require_oauth.acquire("profile group") as the_token:
        db_uri = current_app.config["AUTH_DB"]
        with db.connection(db_uri) as conn:
            group = _user_group(conn, the_token.user).maybe(# type: ignore[misc]
                False, lambda grp: grp)# type: ignore[arg-type]

        if group:
            return jsonify(asdict(group))
        raise NotFoundError("User is not a member of any group.")

@users.route("/resources", methods=["GET"])
@require_oauth("profile resource")
def user_resources() -> Response:
    """Retrieve the resources a user has access to."""
    with require_oauth.acquire("profile resource") as the_token:
        db_uri = current_app.config["AUTH_DB"]
        with db.connection(db_uri) as conn:
            return jsonify([
                asdict(resource) for resource in
                _user_resources(conn, the_token.user)])

@users.route("group/join-request", methods=["GET"])
@require_oauth("profile group")
def user_join_request_exists():
    """Check whether a user has an active group join request."""
    def __request_exists__(conn: db.DbConnection, user: User) -> dict[str, Any]:
        with db.cursor(conn) as cursor:
            cursor.execute(
                "SELECT * FROM group_join_requests WHERE requester_id=? AND "
                "status = 'PENDING'",
                (str(user.user_id),))
            res = cursor.fetchone()
            if res:
                return {
                    "request_id": res["request_id"],
                    "exists": True
                }
        return{
            "status": "Not found",
            "exists": False
        }
    with require_oauth.acquire("profile group") as the_token:
        return jsonify(with_db_connection(partial(
            __request_exists__, user=the_token.user)))

@users.route("/list", methods=["GET"])
@require_oauth("profile user")
def list_all_users() -> Response:
    """List all the users."""
    with require_oauth.acquire("profile group") as _the_token:
        return jsonify(tuple(
            asdict(user) for user in with_db_connection(list_users)))

@users.route("/handle-unverified", methods=["POST"])
def handle_unverified():
    """Handle case where user tries to login but is unverified"""
    form = request.form
    # TODO: Maybe have a GN2_URI setting here?
    #       or pass the client_id here?
    return render_template(
        "users/unverified-user.html", email=form.get("user:email"))

@users.route("/send-verification", methods=["POST"])
def send_verification_code():
    """Send verification code email."""
    form = request.form
    with (db.connection(current_app.config["AUTH_DB"]) as conn,
          db.cursor(conn) as cursor):
        user = user_by_email(conn, form["user_email"])
        if valid_login(conn, user, form.get("user_password", "")):
            cursor.execute(
                "DELETE FROM user_verification_codes WHERE user_id=:user_id",
                {"user_id": str(user.user_id)})
            send_verification_email(conn, user)
            return jsonify({
                "status": "success",
                "message": "Sent a verification code to your email."
            })

    resp = jsonify({
        "error": "InvalidLogin",
        "error-description": "Invalid email or password."
    })
    resp.code = 400
    return resp