You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
126 lines
4.0 KiB
126 lines
4.0 KiB
"""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.BASIC, 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.BASIC, 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)
|
|
if user_token_page is not None:
|
|
return APIResponse(APIPage.from_page(user_token_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():
|
|
"""
|
|
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):
|
|
"""
|
|
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):
|
|
"""
|
|
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)
|