Corvus/server/corvus/api/authentication_api.py

"""Authentication API blueprint and endpoint definitions."""
from flask import Blueprint, g, abort, request

from corvus.api.decorators import return_json
from corvus.api.model import APIMessage, APIResponse, APIPage
from corvus.middleware import authentication_middleware
from corvus.service import (
    user_token_service,
    authentication_service,
    user_service,
    transformation_service
)
from corvus.middleware.authentication_middleware import Auth
from corvus.service.role_service import Role
from corvus.model import UserToken
from corvus.utility.pagination_utility import get_pagination_params

AUTH_BLUEPRINT = Blueprint(
    name='auth', import_name=__name__, url_prefix='/auth')


@AUTH_BLUEPRINT.route('/login', methods=['POST'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.BASIC, required_role=Role.USER)
def login() -> APIResponse:
    """
    Get a token for continued authentication.

    :return: A login token for continued authentication
    """
    user_token = user_token_service.create(g.user)
    return APIResponse(user_token, 200)


@AUTH_BLUEPRINT.route('/bump', methods=['POST'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.TOKEN, required_role=Role.USER)
def login_bump() -> APIResponse:
    """
    Update the user last seen timestamp.

    :return: A time stamp for the bumped login
    """
    user_service.update_last_login_time(g.user)
    return APIResponse(g.user, 200, ['lastLoginTime'])


@AUTH_BLUEPRINT.route('/logout', methods=['POST'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.TOKEN, required_role=Role.USER)
def logout() -> APIResponse:
    """
    Logout and delete a token.

    :return:
    """
    authentication_service.logout(g.user_token)
    return APIResponse(APIMessage(True, None), 200)


@AUTH_BLUEPRINT.route('/token', methods=['GET'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.BASIC, required_role=Role.USER)
def get_tokens() -> APIResponse:
    """
    Get a list of all tokens for the current user.

    :return: a paginated list of user tokens
    """
    page, per_page = get_pagination_params(request.args)
    user_token_page = user_token_service.find_by_user(g.user, page, per_page)
    api_page = APIPage.from_page(user_token_page)
    if api_page is not None:
        return APIResponse(api_page, 200)
    return abort(404)


@AUTH_BLUEPRINT.route('/token', methods=['POST'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.BASIC, required_role=Role.USER)
def create_token() -> APIResponse:
    """
    Create a new token with optional parameters.

    note: String
    enabled: Boolean
    expirationTime: DateTime

    :return: The new token with the optional parameters
    """
    requested_token: UserToken = transformation_service.deserialize_model(
        UserToken, request.json, options=['note', 'enabled', 'expirationTime'])
    user_token = user_token_service.create(
        g.user, requested_token.note,
        requested_token.enabled, requested_token.expiration_time)
    return APIResponse(user_token, 200)


@AUTH_BLUEPRINT.route('/token/<token>', methods=['GET'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.BASIC, required_role=Role.USER)
def get_token(token: str) -> APIResponse:
    """
    Retrieve a specific token for this user.

    :param token: The token to retrieve for this user
    :return: The token if it exists
    """
    user_token = user_token_service.find_by_user_and_token(g.user, token)
    if user_token is None:
        return abort(404)
    return APIResponse(user_token, 200)


@AUTH_BLUEPRINT.route('/token/<token>', methods=['DELETE'])
@return_json
@authentication_middleware.require(
    required_auth=Auth.BASIC, required_role=Role.USER)
def delete_token(token: str) -> APIResponse:
    """
    Delete a specific token for this user.

    :param token: The token to delete for this user
    :return: Nothing on success
    """
    user_token = user_token_service.find_by_user_and_token(g.user, token)
    if user_token is None:
        return abort(404)
    user_token_service.delete(user_token)
    return APIResponse(APIMessage(True, None), 200)