pyceo/ceod/api/database.py

from flask import Blueprint
from zope import component
from functools import wraps

from ceod.api.utils import authz_restrict_to_syscom, user_is_in_group, requires_authentication_no_realm, \
    development_only
from ceo_common.errors import UserNotFoundError, DatabaseConnectionError, DatabasePermissionError, \
    InvalidUsernameError, UserAlreadyExistsError
from ceo_common.interfaces import ILDAPService, IDatabaseService


bp = Blueprint('db', __name__)


def db_exception_handler(func):
    @wraps(func)
    def function(db_type: str, username: str):
        try:
            if not username.isalnum():  # username should not contain symbols
                raise InvalidUsernameError()
            ldap_srv = component.getUtility(ILDAPService)
            ldap_srv.get_user(username)  # make sure user exists
            return func(db_type, username)
        except UserNotFoundError:
            return {'error': 'user not found'}, 404
        except UserAlreadyExistsError:
            return {'error': 'database user is already created'}, 409
        except InvalidUsernameError:
            return {'error': 'username contains invalid characters'}, 400
        except DatabaseConnectionError:
            return {'error': 'unable to connect or authenticate to sql server'}, 500
        except DatabasePermissionError:
            return {'error': 'unable to perform action due to permissions'}, 500
    return function


@db_exception_handler
def create_db_from_type(db_type: str, username: str):
    db_srv = component.getUtility(IDatabaseService, db_type)
    password = db_srv.create_db(username)
    return {'password': password}


@db_exception_handler
def reset_db_passwd_from_type(db_type: str, username: str):
    db_srv = component.getUtility(IDatabaseService, db_type)
    password = db_srv.reset_passwd(username)
    return {'password': password}


@db_exception_handler
def delete_db_from_type(db_type: str, username: str):
    db_srv = component.getUtility(IDatabaseService, db_type)
    db_srv.delete_db(username)


@bp.route('/mysql/<username>', methods=['POST'])
@requires_authentication_no_realm
def create_mysql_db(auth_user: str, username: str):
    if not (auth_user == username or user_is_in_group(auth_user, 'syscom')):
        return {'error': "not authorized to create databases for others"}, 403
    return create_db_from_type('mysql', username)


@bp.route('/postgresql/<username>', methods=['POST'])
@requires_authentication_no_realm
def create_postgresql_db(auth_user: str, username: str):
    if not (auth_user == username or user_is_in_group(auth_user, 'syscom')):
        return {'error': "not authorized to create databases for others"}, 403
    return create_db_from_type('postgresql', username)


@bp.route('/mysql/<username>/pwreset', methods=['POST'])
@requires_authentication_no_realm
def reset_mysql_db_passwd(auth_user: str, username: str):
    if not (auth_user == username or user_is_in_group(auth_user, 'syscom')):
        return {'error': "not authorized to request password reset for others"}, 403
    return reset_db_passwd_from_type('mysql', username)


@bp.route('/postgresql/<username>/pwreset', methods=['POST'])
@requires_authentication_no_realm
def reset_postgresql_db_passwd(auth_user: str, username: str):
    if not (auth_user == username or user_is_in_group(auth_user, 'syscom')):
        return {'error': "not authorized to request password reset for others"}, 403
    return reset_db_passwd_from_type('postgresql', username)


@bp.route('/mysql/<username>', methods=['DELETE'])
@authz_restrict_to_syscom
@development_only
def delete_mysql_db(username: str):
    delete_db_from_type('mysql', username)


@bp.route('/postgresql/<username>', methods=['DELETE'])
@authz_restrict_to_syscom
@development_only
def delete_postgresql_db(username: str):
    delete_db_from_type('postgresl', username)