78 lines
3.0 KiB
Python
78 lines
3.0 KiB
Python
from flask import Blueprint, request
|
|
from zope import component
|
|
from ceod.api.utils import authz_restrict_to_staff, authz_restrict_to_syscom, \
|
|
user_is_in_group, requires_authentication_no_realm, \
|
|
create_streaming_response, development_only
|
|
from ceo_common.errors import UserNotFoundError, DatabaseConnectionError, DatabasePermissionError
|
|
from ceo_common.interfaces import ILDAPService, IDatabaseService
|
|
|
|
|
|
bp = Blueprint('db', __name__)
|
|
|
|
|
|
def create_db_from_type(db_type: str, username: str):
|
|
try:
|
|
if not username.isalnum(): # username should not contain symbols
|
|
raise UserNotFoundError
|
|
ldap_srv = component.getUtility(ILDAPService)
|
|
ldap_srv.get_user(username) # make sure user exists
|
|
db_srv = component.getUtility(IDatabaseService, db_type)
|
|
password = db_srv.create_db(username)
|
|
return {'password': password}
|
|
except UserNotFoundError:
|
|
return {'error': 'user not found'}, 404
|
|
except DatabaseConnectionError:
|
|
return {'error': 'unable to connect or authenticate to sql server'}, 400
|
|
except DatabasePermissionError:
|
|
return {'error': 'unable to perform action due to permissions'}, 502
|
|
except:
|
|
return {'error': 'Unexpected error'}, 500
|
|
|
|
|
|
def delete_db_from_type(db_type: str, username: str):
|
|
try:
|
|
if not username.isalnum(): # username should not contain symbols
|
|
raise UserNotFoundError
|
|
ldap_srv = component.getUtility(ILDAPService)
|
|
ldap_srv.get_user(username) # make sure user exists
|
|
db_srv = component.getUtility(IDatabaseService, db_type)
|
|
db_srv.delete_db(username)
|
|
except UserNotFoundError:
|
|
return {'error': 'user not found'}, 404
|
|
except DatabaseConnectionError:
|
|
return {'error': 'unable to connect or authenticate to sql server'}, 400
|
|
except DatabasePermissionError:
|
|
return {'error': 'unable to perform action due to permissions'}, 502
|
|
except:
|
|
return {'error': 'Unexpected error'}, 500
|
|
|
|
|
|
@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('/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=['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('/postgresql/<username>', methods=['DELETE'])
|
|
@authz_restrict_to_syscom
|
|
@development_only
|
|
def delete_postgresql_db(username: str):
|
|
delete_db_from_type('postgresl', username)
|