powerdns-admin/powerdnsadmin/routes/api.py

1242 lines
42 KiB
Python
Raw Normal View History

2019-03-01 22:49:31 +00:00
import json
2022-12-22 21:55:05 +00:00
import secrets
import string
from base64 import b64encode
2022-12-22 21:55:05 +00:00
from urllib.parse import urljoin
from flask import (Blueprint, g, request, abort, current_app, make_response, jsonify)
from flask_login import current_user
2019-03-01 22:49:31 +00:00
from .base import csrf
2022-12-22 21:55:05 +00:00
from ..decorators import (
api_basic_auth, api_can_create_domain, is_json, apikey_auth,
apikey_can_create_domain, apikey_can_remove_domain,
apikey_is_admin, apikey_can_access_domain, apikey_can_configure_dnssec,
api_role_can, apikey_or_basic_auth,
callback_if_request_body_contains_key, allowed_record_types, allowed_record_ttl
)
from ..lib import utils, helper
from ..lib.errors import (
StructuredException,
DomainNotExists, DomainAlreadyExists, DomainAccessForbidden,
RequestIsNotJSON, ApiKeyCreateFail, ApiKeyNotUsable, NotEnoughPrivileges,
AccountCreateFail, AccountUpdateFail, AccountDeleteFail,
AccountCreateDuplicate, AccountNotExists,
2021-03-16 18:39:53 +00:00
UserCreateFail, UserCreateDuplicate, UserUpdateFail, UserDeleteFail,
UserUpdateFailEmail, InvalidAccountNameException
)
2022-12-22 21:55:05 +00:00
from ..lib.schema import (
ApiKeySchema, DomainSchema, ApiPlainKeySchema, UserSchema, AccountSchema,
UserDetailedSchema,
)
2022-12-22 21:55:05 +00:00
from ..models import (
User, Domain, DomainUser, Account, AccountUser, History, Setting, ApiKey,
Role,
)
from ..models.base import db
api_bp = Blueprint('api', __name__, url_prefix='/api/v1')
2022-06-17 14:48:23 +00:00
apilist_bp = Blueprint('apilist', __name__, url_prefix='/')
2019-03-01 22:49:31 +00:00
apikey_schema = ApiKeySchema(many=True)
2021-03-16 18:39:53 +00:00
apikey_single_schema = ApiKeySchema()
2019-03-01 22:49:31 +00:00
domain_schema = DomainSchema(many=True)
2021-03-16 18:39:53 +00:00
apikey_plain_schema = ApiPlainKeySchema()
user_schema = UserSchema(many=True)
2021-03-16 18:39:53 +00:00
user_single_schema = UserSchema()
user_detailed_schema = UserDetailedSchema()
account_schema = AccountSchema(many=True)
2021-03-16 18:39:53 +00:00
account_single_schema = AccountSchema()
2019-03-01 22:49:31 +00:00
2022-06-17 14:48:23 +00:00
def get_user_domains():
domains = db.session.query(Domain) \
.outerjoin(DomainUser, Domain.id == DomainUser.domain_id) \
.outerjoin(Account, Domain.account_id == Account.id) \
.outerjoin(AccountUser, Account.id == AccountUser.account_id) \
.filter(
2022-12-22 21:55:05 +00:00
db.or_(
DomainUser.user_id == current_user.id,
AccountUser.user_id == current_user.id
)).all()
return domains
2019-12-22 03:06:02 +00:00
def get_user_apikeys(domain_name=None):
info = []
apikey_query = db.session.query(ApiKey) \
.join(Domain.apikeys) \
.outerjoin(DomainUser, Domain.id == DomainUser.domain_id) \
.outerjoin(Account, Domain.account_id == Account.id) \
.outerjoin(AccountUser, Account.id == AccountUser.account_id) \
.filter(
2022-12-22 21:55:05 +00:00
db.or_(
DomainUser.user_id == User.id,
AccountUser.user_id == User.id
)
2019-12-22 03:06:02 +00:00
) \
.filter(User.id == current_user.id)
if domain_name:
info = apikey_query.filter(Domain.name == domain_name).all()
else:
info = apikey_query.all()
return info
def get_role_id(role_name, role_id=None):
if role_id:
if role_name:
role = Role.query.filter(Role.name == role_name).first()
if not role or role.id != role_id:
role_id = None
else:
role = Role.query.filter(Role.id == role_id).first()
if not role:
role_id = None
else:
role = Role.query.filter(Role.name == role_name).first()
role_id = role.id if role else None
return role_id
@api_bp.errorhandler(400)
2019-03-01 22:49:31 +00:00
def handle_400(err):
return jsonify({"msg": "Bad Request"}), 400
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(401)
2019-03-01 22:49:31 +00:00
def handle_401(err):
return jsonify({"msg": "Unauthorized"}), 401
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(409)
def handle_409(err):
return jsonify({"msg": "Conflict"}), 409
@api_bp.errorhandler(500)
2019-03-01 22:49:31 +00:00
def handle_500(err):
return jsonify({"msg": "Internal Server Error"}), 500
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(StructuredException)
def handle_StructuredException(err):
return jsonify(err.to_dict()), err.status_code
@api_bp.errorhandler(DomainNotExists)
2019-03-01 22:49:31 +00:00
def handle_domain_not_exists(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(DomainAlreadyExists)
def handle_domain_already_exists(err):
return jsonify(err.to_dict()), err.status_code
@api_bp.errorhandler(DomainAccessForbidden)
2019-03-01 22:49:31 +00:00
def handle_domain_access_forbidden(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(ApiKeyCreateFail)
2019-03-01 22:49:31 +00:00
def handle_apikey_create_fail(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(ApiKeyNotUsable)
2019-03-01 22:49:31 +00:00
def handle_apikey_not_usable(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(NotEnoughPrivileges)
2019-03-01 22:49:31 +00:00
def handle_not_enough_privileges(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.errorhandler(RequestIsNotJSON)
2019-03-01 22:49:31 +00:00
def handle_request_is_not_json(err):
return jsonify(err.to_dict()), err.status_code
2019-03-01 22:49:31 +00:00
@api_bp.before_request
2019-03-01 22:49:31 +00:00
@is_json
def before_request():
# Check site is in maintenance mode
maintenance = Setting().get('maintenance')
2022-12-22 21:55:05 +00:00
if (maintenance and current_user.is_authenticated and current_user.role.name not in ['Administrator', 'Operator']):
return make_response(
jsonify({
"status": False,
"msg": "Site is in maintenance mode"
}))
2019-03-01 22:49:31 +00:00
2022-06-17 14:48:23 +00:00
@apilist_bp.route('/api', methods=['GET'])
def index():
return '[{"url": "/api/v1", "version": 1}]', 200
@api_bp.route('/pdnsadmin/zones', methods=['POST'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
@api_can_create_domain
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_login_create_zone():
pdns_api_url = Setting().get('pdns_api_url')
pdns_api_key = Setting().get('pdns_api_key')
pdns_version = Setting().get('pdns_version')
api_uri_with_prefix = utils.pdns_api_extended_uri(pdns_version)
api_full_uri = api_uri_with_prefix + '/servers/localhost/zones'
headers = {}
headers['X-API-Key'] = pdns_api_key
headers['Content-Type'] = 'application/json'
2019-03-01 22:49:31 +00:00
msg_str = "Sending request to powerdns API {0}"
msg = msg_str.format(request.get_json(force=True))
current_app.logger.debug(msg)
2019-03-01 22:49:31 +00:00
try:
resp = utils.fetch_remote(urljoin(pdns_api_url, api_full_uri),
2019-12-22 03:06:02 +00:00
method='POST',
data=request.get_json(force=True),
headers=headers,
accept='application/json; q=1',
verify=Setting().get('verify_ssl_connections'))
except Exception as e:
current_app.logger.error("Cannot create domain. Error: {}".format(e))
abort(500)
2019-03-01 22:49:31 +00:00
if resp.status_code == 201:
current_app.logger.debug("Request to powerdns API successful")
2019-03-01 22:49:31 +00:00
data = request.get_json(force=True)
2021-03-27 18:33:11 +00:00
domain = Domain()
domain.update()
domain_id = domain.get_id_by_name(data['name'].rstrip('.'))
history = History(msg='Add domain {0}'.format(
data['name'].rstrip('.')),
2022-12-22 21:55:05 +00:00
detail=json.dumps(data),
created_by=current_user.username,
domain_id=domain_id)
2019-03-01 22:49:31 +00:00
history.add()
if current_user.role.name not in ['Administrator', 'Operator']:
2022-12-22 21:55:05 +00:00
current_app.logger.debug("User is ordinary user, assigning created domain")
2019-03-01 22:49:31 +00:00
domain = Domain(name=data['name'].rstrip('.'))
domain.update()
domain.grant_privileges([current_user.id])
2019-03-01 22:49:31 +00:00
if resp.status_code == 409:
2019-12-22 03:06:02 +00:00
raise (DomainAlreadyExists)
2019-03-01 22:49:31 +00:00
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/pdnsadmin/zones', methods=['GET'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
def api_login_list_zones():
if current_user.role.name not in ['Administrator', 'Operator']:
domain_obj_list = get_user_domains()
2019-03-01 22:49:31 +00:00
else:
domain_obj_list = Domain.query.all()
domain_obj_list = [] if domain_obj_list is None else domain_obj_list
return jsonify(domain_schema.dump(domain_obj_list)), 200
2019-03-01 22:49:31 +00:00
@api_bp.route('/pdnsadmin/zones/<string:domain_name>', methods=['DELETE'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
@api_can_create_domain
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_login_delete_zone(domain_name):
pdns_api_url = Setting().get('pdns_api_url')
pdns_api_key = Setting().get('pdns_api_key')
pdns_version = Setting().get('pdns_version')
api_uri_with_prefix = utils.pdns_api_extended_uri(pdns_version)
api_full_uri = api_uri_with_prefix + '/servers/localhost/zones'
api_full_uri += '/' + domain_name
headers = {}
headers['X-API-Key'] = pdns_api_key
domain = Domain.query.filter(Domain.name == domain_name)
if not domain:
abort(404)
if current_user.role.name not in ['Administrator', 'Operator']:
user_domains_obj_list = get_user_domains()
2019-03-01 22:49:31 +00:00
user_domains_list = [item.name for item in user_domains_obj_list]
if domain_name not in user_domains_list:
raise DomainAccessForbidden()
msg_str = "Sending request to powerdns API {0}"
current_app.logger.debug(msg_str.format(domain_name))
2019-03-01 22:49:31 +00:00
try:
resp = utils.fetch_remote(urljoin(pdns_api_url, api_full_uri),
method='DELETE',
headers=headers,
accept='application/json; q=1',
verify=Setting().get('verify_ssl_connections'))
2019-03-01 22:49:31 +00:00
if resp.status_code == 204:
current_app.logger.debug("Request to powerdns API successful")
2019-03-01 22:49:31 +00:00
2021-03-27 18:33:11 +00:00
domain = Domain()
domain_id = domain.get_id_by_name(domain_name)
domain.update()
2021-03-16 18:37:05 +00:00
history = History(msg='Delete domain {0}'.format(
2022-01-19 12:50:12 +00:00
utils.pretty_domain_name(domain_name)),
2022-12-22 21:55:05 +00:00
detail='',
created_by=current_user.username,
domain_id=domain_id)
2019-03-01 22:49:31 +00:00
history.add()
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
abort(500)
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/pdnsadmin/apikeys', methods=['POST'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_generate_apikey():
data = request.get_json()
description = None
role_name = None
apikey = None
domain_obj_list = []
account_obj_list = []
2019-03-01 22:49:31 +00:00
abort(400) if 'role' not in data else None
if 'domains' not in data:
domains = []
2022-12-22 21:55:05 +00:00
elif not isinstance(data['domains'], (list,)):
abort(400)
else:
domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']]
if 'accounts' not in data:
accounts = []
2022-12-22 21:55:05 +00:00
elif not isinstance(data['accounts'], (list,)):
abort(400)
else:
accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']]
2019-03-01 22:49:31 +00:00
description = data['description'] if 'description' in data else None
if isinstance(data['role'], str):
role_name = data['role']
elif isinstance(data['role'], dict) and 'name' in data['role'].keys():
role_name = data['role']['name']
else:
abort(400)
2019-03-01 22:49:31 +00:00
if role_name == 'User' and len(domains) == 0 and len(accounts) == 0:
current_app.logger.error("Apikey with User role must have domains or accounts")
2019-03-01 22:49:31 +00:00
raise ApiKeyNotUsable()
if role_name == 'User' and len(domains) > 0:
2019-03-01 22:49:31 +00:00
domain_obj_list = Domain.query.filter(Domain.name.in_(domains)).all()
if len(domain_obj_list) == 0:
2020-01-26 23:05:28 +00:00
msg = "One of supplied domains does not exist"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise DomainNotExists(message=msg)
if role_name == 'User' and len(accounts) > 0:
account_obj_list = Account.query.filter(Account.name.in_(accounts)).all()
if len(account_obj_list) == 0:
msg = "One of supplied accounts does not exist"
current_app.logger.error(msg)
raise AccountNotExists(message=msg)
if current_user.role.name not in ['Administrator', 'Operator']:
2019-03-01 22:49:31 +00:00
# domain list of domain api key should be valid for
# if not any domain error
# role of api key, user cannot assign role above for api key
if role_name != 'User':
msg = "User cannot assign other role than User"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise NotEnoughPrivileges(message=msg)
if len(accounts) > 0:
msg = "User cannot assign accounts"
current_app.logger.error(msg)
raise NotEnoughPrivileges(message=msg)
user_domain_obj_list = get_user_domains()
2019-03-01 22:49:31 +00:00
domain_list = [item.name for item in domain_obj_list]
user_domain_list = [item.name for item in user_domain_obj_list]
current_app.logger.debug("Input domain list: {0}".format(domain_list))
2022-12-22 21:55:05 +00:00
current_app.logger.debug("User domain list: {0}".format(user_domain_list))
2019-03-01 22:49:31 +00:00
inter = set(domain_list).intersection(set(user_domain_list))
if not (len(inter) == len(domain_list)):
msg = "You don't have access to one of domains"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise DomainAccessForbidden(message=msg)
apikey = ApiKey(desc=description,
role_name=role_name,
domains=domain_obj_list,
accounts=account_obj_list)
2019-03-01 22:49:31 +00:00
try:
apikey.create()
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
raise ApiKeyCreateFail(message='Api key create failed')
apikey.plain_key = b64encode(apikey.plain_key.encode('utf-8')).decode('utf-8')
2021-03-16 18:39:53 +00:00
return jsonify(apikey_plain_schema.dump(apikey)), 201
2019-03-01 22:49:31 +00:00
@api_bp.route('/pdnsadmin/apikeys', defaults={'domain_name': None})
@api_bp.route('/pdnsadmin/apikeys/<string:domain_name>')
2019-03-01 22:49:31 +00:00
@api_basic_auth
def api_get_apikeys(domain_name):
apikeys = []
current_app.logger.debug("Getting apikeys")
2019-03-01 22:49:31 +00:00
if current_user.role.name not in ['Administrator', 'Operator']:
2019-03-01 22:49:31 +00:00
if domain_name:
2019-12-22 03:06:02 +00:00
msg = "Check if domain {0} exists and is allowed for user.".format(
domain_name)
current_app.logger.debug(msg)
2019-12-22 03:06:02 +00:00
apikeys = get_user_apikeys(domain_name)
2019-03-01 22:49:31 +00:00
if not apikeys:
raise DomainAccessForbidden(name=domain_name)
current_app.logger.debug(apikey_schema.dump(apikeys))
2019-03-01 22:49:31 +00:00
else:
msg_str = "Getting all allowed domains for user {0}"
msg = msg_str.format(current_user.username)
current_app.logger.debug(msg)
2019-03-01 22:49:31 +00:00
try:
2019-12-22 03:06:02 +00:00
apikeys = get_user_apikeys()
current_app.logger.debug(apikey_schema.dump(apikeys))
2019-03-01 22:49:31 +00:00
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
abort(500)
else:
current_app.logger.debug("Getting all domains for administrative user")
2019-03-01 22:49:31 +00:00
try:
apikeys = ApiKey.query.all()
current_app.logger.debug(apikey_schema.dump(apikeys))
2019-03-01 22:49:31 +00:00
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
abort(500)
return jsonify(apikey_schema.dump(apikeys)), 200
2019-03-01 22:49:31 +00:00
@api_bp.route('/pdnsadmin/apikeys/<int:apikey_id>', methods=['GET'])
@api_basic_auth
def api_get_apikey(apikey_id):
apikey = ApiKey.query.get(apikey_id)
if not apikey:
abort(404)
current_app.logger.debug(current_user.role.name)
if current_user.role.name not in ['Administrator', 'Operator']:
if apikey_id not in [a.id for a in get_user_apikeys()]:
raise DomainAccessForbidden()
2021-03-16 18:39:53 +00:00
return jsonify(apikey_single_schema.dump(apikey)), 200
@api_bp.route('/pdnsadmin/apikeys/<int:apikey_id>', methods=['DELETE'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_delete_apikey(apikey_id):
apikey = ApiKey.query.get(apikey_id)
if not apikey:
abort(404)
current_app.logger.debug(current_user.role.name)
2019-03-01 22:49:31 +00:00
if current_user.role.name not in ['Administrator', 'Operator']:
2019-12-22 03:06:02 +00:00
apikeys = get_user_apikeys()
user_domains_obj_list = current_user.get_domain().all()
2019-03-01 22:49:31 +00:00
apikey_domains_obj_list = apikey.domains
user_domains_list = [item.name for item in user_domains_obj_list]
apikey_domains_list = [item.name for item in apikey_domains_obj_list]
apikeys_ids = [apikey_item.id for apikey_item in apikeys]
inter = set(apikey_domains_list).intersection(set(user_domains_list))
if not (len(inter) == len(apikey_domains_list)):
msg = "You don't have access to some domains apikey belongs to"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise DomainAccessForbidden(message=msg)
if apikey_id not in apikeys_ids:
raise DomainAccessForbidden()
try:
apikey.delete()
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
abort(500)
return '', 204
@api_bp.route('/pdnsadmin/apikeys/<int:apikey_id>', methods=['PUT'])
2019-03-01 22:49:31 +00:00
@api_basic_auth
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_update_apikey(apikey_id):
# if role different and user is allowed to change it, update
# if apikey domains are different and user is allowed to handle
# that domains update domains
domain_obj_list = None
account_obj_list = None
apikey = ApiKey.query.get(apikey_id)
if not apikey:
abort(404)
2019-03-01 22:49:31 +00:00
data = request.get_json()
description = data['description'] if 'description' in data else None
if 'role' in data:
if isinstance(data['role'], str):
role_name = data['role']
elif isinstance(data['role'], dict) and 'name' in data['role'].keys():
role_name = data['role']['name']
else:
abort(400)
target_role = role_name
else:
role_name = None
target_role = apikey.role.name
if 'domains' not in data:
domains = None
2022-12-22 21:55:05 +00:00
elif not isinstance(data['domains'], (list,)):
abort(400)
else:
domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']]
if 'accounts' not in data:
accounts = None
2022-12-22 21:55:05 +00:00
elif not isinstance(data['accounts'], (list,)):
abort(400)
else:
accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']]
2019-03-01 22:49:31 +00:00
current_app.logger.debug('Updating apikey with id {0}'.format(apikey_id))
2019-03-01 22:49:31 +00:00
if target_role == 'User':
current_domains = [item.name for item in apikey.domains]
current_accounts = [item.name for item in apikey.accounts]
if domains is not None:
domain_obj_list = Domain.query.filter(Domain.name.in_(domains)).all()
if len(domain_obj_list) != len(domains):
msg = "One of supplied domains does not exist"
current_app.logger.error(msg)
raise DomainNotExists(message=msg)
target_domains = domains
else:
target_domains = current_domains
if accounts is not None:
account_obj_list = Account.query.filter(Account.name.in_(accounts)).all()
if len(account_obj_list) != len(accounts):
msg = "One of supplied accounts does not exist"
current_app.logger.error(msg)
raise AccountNotExists(message=msg)
target_accounts = accounts
else:
target_accounts = current_accounts
if len(target_domains) == 0 and len(target_accounts) == 0:
current_app.logger.error("Apikey with User role must have domains or accounts")
raise ApiKeyNotUsable()
if domains is not None and set(domains) == set(current_domains):
current_app.logger.debug(
"Domains are the same, apikey domains won't be updated")
domains = None
if accounts is not None and set(accounts) == set(current_accounts):
current_app.logger.debug(
"Accounts are the same, apikey accounts won't be updated")
accounts = None
2019-03-01 22:49:31 +00:00
if current_user.role.name not in ['Administrator', 'Operator']:
2019-03-01 22:49:31 +00:00
if role_name != 'User':
msg = "User cannot assign other role than User"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise NotEnoughPrivileges(message=msg)
if len(accounts) > 0:
msg = "User cannot assign accounts"
current_app.logger.error(msg)
raise NotEnoughPrivileges(message=msg)
2019-12-22 03:06:02 +00:00
apikeys = get_user_apikeys()
2019-03-01 22:49:31 +00:00
apikeys_ids = [apikey_item.id for apikey_item in apikeys]
user_domain_obj_list = current_user.get_domain().all()
2019-03-01 22:49:31 +00:00
domain_list = [item.name for item in domain_obj_list]
user_domain_list = [item.name for item in user_domain_obj_list]
current_app.logger.debug("Input domain list: {0}".format(domain_list))
current_app.logger.debug(
"User domain list: {0}".format(user_domain_list))
2019-03-01 22:49:31 +00:00
inter = set(domain_list).intersection(set(user_domain_list))
if not (len(inter) == len(domain_list)):
msg = "You don't have access to one of domains"
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise DomainAccessForbidden(message=msg)
if apikey_id not in apikeys_ids:
msg = 'Apikey does not belong to domain to which user has access'
current_app.logger.error(msg)
2019-03-01 22:49:31 +00:00
raise DomainAccessForbidden()
if role_name == apikey.role.name:
current_app.logger.debug("Role is same, apikey role won't be updated")
2019-03-01 22:49:31 +00:00
role_name = None
if description == apikey.description:
msg = "Description is same, apikey description won't be updated"
current_app.logger.debug(msg)
2019-03-01 22:49:31 +00:00
description = None
if target_role != "User":
domains, accounts = [], []
2019-03-01 22:49:31 +00:00
try:
apikey.update(role_name=role_name,
domains=domains,
accounts=accounts,
description=description)
2019-03-01 22:49:31 +00:00
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
2019-03-01 22:49:31 +00:00
abort(500)
return '', 204
@api_bp.route('/pdnsadmin/users', defaults={'username': None})
@api_bp.route('/pdnsadmin/users/<string:username>')
@api_basic_auth
@api_role_can('list users', allow_self=True)
def api_list_users(username=None):
if username is None:
user_list = [] or User.query.all()
2021-03-16 18:39:53 +00:00
return jsonify(user_schema.dump(user_list)), 200
else:
2021-03-16 18:39:53 +00:00
user = User.query.filter(User.username == username).first()
if user is None:
abort(404)
2021-03-16 18:39:53 +00:00
return jsonify(user_detailed_schema.dump(user)), 200
@api_bp.route('/pdnsadmin/users', methods=['POST'])
@api_basic_auth
@api_role_can('create users', allow_self=True)
@csrf.exempt
def api_create_user():
"""
Create new user
"""
data = request.get_json()
username = data['username'] if 'username' in data else None
password = data['password'] if 'password' in data else None
plain_text_password = (
data['plain_text_password']
if 'plain_text_password' in data
else None
)
firstname = data['firstname'] if 'firstname' in data else None
lastname = data['lastname'] if 'lastname' in data else None
email = data['email'] if 'email' in data else None
otp_secret = data['otp_secret'] if 'otp_secret' in data else None
confirmed = data['confirmed'] if 'confirmed' in data else None
role_name = data['role_name'] if 'role_name' in data else None
role_id = data['role_id'] if 'role_id' in data else None
# Create user
if not username:
current_app.logger.debug('Invalid username {}'.format(username))
abort(400)
if not confirmed:
confirmed = False
elif confirmed is not True:
current_app.logger.debug('Invalid confirmed {}'.format(confirmed))
abort(400)
if not plain_text_password and not password:
plain_text_password = ''.join(
secrets.choice(string.ascii_letters + string.digits)
for _ in range(15))
if not role_name and not role_id:
role_name = 'User'
role_id = get_role_id(role_name, role_id)
if not role_id:
current_app.logger.debug(
'Invalid role {}/{}'.format(role_name, role_id))
abort(400)
user = User(
username=username,
password=password,
plain_text_password=plain_text_password,
firstname=firstname,
lastname=lastname,
role_id=role_id,
email=email,
otp_secret=otp_secret,
confirmed=confirmed,
)
try:
result = user.create_local_user()
except Exception as e:
current_app.logger.error('Create user ({}, {}) error: {}'.format(
username, email, e))
raise UserCreateFail(message='User create failed')
if not result['status']:
current_app.logger.warning('Create user ({}, {}) error: {}'.format(
username, email, result['msg']))
2021-03-16 18:39:53 +00:00
raise UserCreateDuplicate(message=result['msg'])
history = History(msg='Created user {0}'.format(user.username),
created_by=current_user.username)
history.add()
2021-03-16 18:39:53 +00:00
return jsonify(user_single_schema.dump(user)), 201
@api_bp.route('/pdnsadmin/users/<int:user_id>', methods=['PUT'])
@api_basic_auth
@api_role_can('update users', allow_self=True)
@csrf.exempt
def api_update_user(user_id):
"""
Update existing user
"""
data = request.get_json()
username = data['username'] if 'username' in data else None
password = data['password'] if 'password' in data else None
plain_text_password = (
data['plain_text_password']
if 'plain_text_password' in data
else None
)
firstname = data['firstname'] if 'firstname' in data else None
lastname = data['lastname'] if 'lastname' in data else None
email = data['email'] if 'email' in data else None
otp_secret = data['otp_secret'] if 'otp_secret' in data else None
confirmed = data['confirmed'] if 'confirmed' in data else None
role_name = data['role_name'] if 'role_name' in data else None
role_id = data['role_id'] if 'role_id' in data else None
user = User.query.get(user_id)
if not user:
current_app.logger.debug("User not found for id {}".format(user_id))
abort(404)
if username:
if username != user.username:
current_app.logger.error(
'Cannot change username for {}'.format(user.username)
)
abort(400)
if password is not None:
user.password = password
user.plain_text_password = plain_text_password or ''
if firstname is not None:
user.firstname = firstname
if lastname is not None:
user.lastname = lastname
if email is not None:
user.email = email
if otp_secret is not None:
user.otp_secret = otp_secret
if confirmed is not None:
user.confirmed = confirmed
if role_name is not None:
user.role_id = get_role_id(role_name, role_id)
elif role_id is not None:
user.role_id = role_id
current_app.logger.debug(
"Updating user {} ({})".format(user_id, user.username))
try:
result = user.update_local_user()
except Exception as e:
current_app.logger.error('Create user ({}, {}) error: {}'.format(
username, email, e))
raise UserUpdateFail(message='User update failed')
if not result['status']:
current_app.logger.warning('Update user ({}, {}) error: {}'.format(
username, email, result['msg']))
2021-03-16 18:39:53 +00:00
if result['msg'].startswith('New email'):
raise UserUpdateFailEmail(message=result['msg'])
else:
raise UserCreateFail(message=result['msg'])
history = History(msg='Updated user {0}'.format(user.username),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route('/pdnsadmin/users/<int:user_id>', methods=['DELETE'])
@api_basic_auth
@api_role_can('delete users')
@csrf.exempt
def api_delete_user(user_id):
user = User.query.get(user_id)
if not user:
current_app.logger.debug("User not found for id {}".format(user_id))
abort(404)
if user.id == current_user.id:
current_app.logger.debug("Cannot delete self (id {})".format(user_id))
msg = "Cannot delete self"
raise UserDeleteFail(message=msg)
# Remove account associations first
user_accounts = Account.query.join(AccountUser).join(
User).filter(AccountUser.user_id == user.id,
AccountUser.account_id == Account.id).all()
for uc in user_accounts:
uc.revoke_privileges_by_id(user.id)
# Then delete the user
result = user.delete()
if not result:
raise UserDeleteFail("Failed to delete user {}".format(
user.username))
history = History(msg='Delete user {0}'.format(user.username),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route('/pdnsadmin/accounts', defaults={'account_name': None})
@api_bp.route('/pdnsadmin/accounts/<string:account_name>')
@api_basic_auth
@api_role_can('list accounts')
def api_list_accounts(account_name):
if current_user.role.name not in ['Administrator', 'Operator']:
msg = "{} role cannot list accounts".format(current_user.role.name)
raise NotEnoughPrivileges(message=msg)
else:
if account_name is None:
account_list = [] or Account.query.all()
2021-03-16 18:39:53 +00:00
return jsonify(account_schema.dump(account_list)), 200
else:
2021-03-16 18:39:53 +00:00
account = Account.query.filter(
Account.name == account_name).first()
if account is None:
abort(404)
2021-03-16 18:39:53 +00:00
return jsonify(account_single_schema.dump(account)), 200
@api_bp.route('/pdnsadmin/accounts', methods=['POST'])
@api_basic_auth
@csrf.exempt
def api_create_account():
if current_user.role.name not in ['Administrator', 'Operator']:
msg = "{} role cannot create accounts".format(current_user.role.name)
raise NotEnoughPrivileges(message=msg)
data = request.get_json()
name = data['name'] if 'name' in data else None
description = data['description'] if 'description' in data else None
contact = data['contact'] if 'contact' in data else None
mail = data['mail'] if 'mail' in data else None
if not name:
current_app.logger.debug("Account creation failed: name missing")
raise InvalidAccountNameException(message="Account name missing")
sanitized_name = Account.sanitize_name(name)
account_exists = Account.query.filter(Account.name == sanitized_name).all()
2021-03-16 18:39:53 +00:00
if len(account_exists) > 0:
msg = ("Requested Account {} would be translated to {}"
" which already exists").format(name, sanitized_name)
2021-03-16 18:39:53 +00:00
current_app.logger.debug(msg)
raise AccountCreateDuplicate(message=msg)
account = Account(name=name,
description=description,
contact=contact,
mail=mail)
try:
result = account.create_account()
except Exception as e:
current_app.logger.error('Error: {0}'.format(e))
raise AccountCreateFail(message='Account create failed')
if not result['status']:
raise AccountCreateFail(message=result['msg'])
history = History(msg='Create account {0}'.format(account.name),
created_by=current_user.username)
history.add()
2021-03-16 18:39:53 +00:00
return jsonify(account_single_schema.dump(account)), 201
@api_bp.route('/pdnsadmin/accounts/<int:account_id>', methods=['PUT'])
@api_basic_auth
@api_role_can('update accounts')
@csrf.exempt
def api_update_account(account_id):
data = request.get_json()
name = data['name'] if 'name' in data else None
description = data['description'] if 'description' in data else None
contact = data['contact'] if 'contact' in data else None
mail = data['mail'] if 'mail' in data else None
account = Account.query.get(account_id)
if not account:
abort(404)
if name and Account.sanitize_name(name) != account.name:
msg = "Account name is immutable"
raise AccountUpdateFail(message=msg)
if current_user.role.name not in ['Administrator', 'Operator']:
msg = "User role update accounts"
raise NotEnoughPrivileges(message=msg)
if description is not None:
account.description = description
if contact is not None:
account.contact = contact
if mail is not None:
account.mail = mail
current_app.logger.debug(
"Updating account {} ({})".format(account_id, account.name))
result = account.update_account()
if not result['status']:
raise AccountUpdateFail(message=result['msg'])
history = History(msg='Update account {0}'.format(account.name),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route('/pdnsadmin/accounts/<int:account_id>', methods=['DELETE'])
@api_basic_auth
@api_role_can('delete accounts')
@csrf.exempt
def api_delete_account(account_id):
account_list = [] or Account.query.filter(Account.id == account_id).all()
if len(account_list) == 1:
account = account_list[0]
else:
abort(404)
2022-12-22 21:55:05 +00:00
current_app.logger.debug(f'Deleting Account {account.name}')
# Remove account association from domains first
if len(account.domains) > 0:
for domain in account.domains:
current_app.logger.info(f"Disassociating domain {domain.name} with {account.name}")
Domain(name=domain.name).assoc_account(None, update=False)
current_app.logger.info("Syncing all domains")
Domain().update()
current_app.logger.debug(
"Deleting account {} ({})".format(account_id, account.name))
result = account.delete_account()
if not result:
raise AccountDeleteFail(message=result['msg'])
history = History(msg='Delete account {0}'.format(account.name),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route('/pdnsadmin/accounts/users/<int:account_id>', methods=['GET'])
2021-03-16 18:39:53 +00:00
@api_bp.route('/pdnsadmin/accounts/<int:account_id>/users', methods=['GET'])
@api_basic_auth
@api_role_can('list account users')
def api_list_account_users(account_id):
account = Account.query.get(account_id)
if not account:
abort(404)
user_list = User.query.join(AccountUser).filter(
AccountUser.account_id == account_id).all()
return jsonify(user_schema.dump(user_list)), 200
@api_bp.route(
'/pdnsadmin/accounts/users/<int:account_id>/<int:user_id>',
methods=['PUT'])
2021-03-16 18:39:53 +00:00
@api_bp.route(
'/pdnsadmin/accounts/<int:account_id>/users/<int:user_id>',
methods=['PUT'])
@api_basic_auth
@api_role_can('add user to account')
@csrf.exempt
def api_add_account_user(account_id, user_id):
account = Account.query.get(account_id)
if not account:
abort(404)
user = User.query.get(user_id)
if not user:
abort(404)
if not account.add_user(user):
raise AccountUpdateFail("Cannot add user {} to {}".format(
user.username, account.name))
history = History(
msg='Add {} user privileges on {}'.format(
user.username, account.name),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route(
'/pdnsadmin/accounts/users/<int:account_id>/<int:user_id>',
methods=['DELETE'])
2021-03-16 18:39:53 +00:00
@api_bp.route(
'/pdnsadmin/accounts/<int:account_id>/users/<int:user_id>',
methods=['DELETE'])
@api_basic_auth
@api_role_can('remove user from account')
@csrf.exempt
def api_remove_account_user(account_id, user_id):
account = Account.query.get(account_id)
if not account:
abort(404)
user = User.query.get(user_id)
if not user:
abort(404)
user_list = User.query.join(AccountUser).filter(
AccountUser.account_id == account_id,
AccountUser.user_id == user_id,
2022-12-22 21:55:05 +00:00
).all()
if not user_list:
abort(404)
if not account.remove_user(user):
raise AccountUpdateFail("Cannot remove user {} from {}".format(
user.username, account.name))
history = History(
msg='Revoke {} user privileges on {}'.format(
user.username, account.name),
created_by=current_user.username)
history.add()
return '', 204
@api_bp.route(
'/servers/<string:server_id>/zones/<string:zone_id>/cryptokeys',
methods=['GET', 'POST'])
@apikey_auth
@apikey_can_access_domain
@apikey_can_configure_dnssec(http_methods=['POST'])
@csrf.exempt
def api_zone_cryptokeys(server_id, zone_id):
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route(
'/servers/<string:server_id>/zones/<string:zone_id>/cryptokeys/<string:cryptokey_id>',
methods=['GET', 'PUT', 'DELETE'])
@apikey_auth
@apikey_can_access_domain
@apikey_can_configure_dnssec()
@csrf.exempt
def api_zone_cryptokey(server_id, zone_id, cryptokey_id):
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route(
2019-03-01 22:49:31 +00:00
'/servers/<string:server_id>/zones/<string:zone_id>/<path:subpath>',
methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'])
2019-03-01 22:49:31 +00:00
@apikey_auth
@apikey_can_access_domain
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_zone_subpath_forward(server_id, zone_id, subpath):
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/servers/<string:server_id>/zones/<string:zone_id>',
methods=['GET', 'PUT', 'PATCH', 'DELETE'])
2019-03-01 22:49:31 +00:00
@apikey_auth
@allowed_record_types
@allowed_record_ttl
2019-03-01 22:49:31 +00:00
@apikey_can_access_domain
@apikey_can_remove_domain(http_methods=['DELETE'])
@callback_if_request_body_contains_key(apikey_can_configure_dnssec()(),
http_methods=['PUT'],
keys=['dnssec', 'nsec3param'])
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_zone_forward(server_id, zone_id):
resp = helper.forward_request()
if not Setting().get('bg_domain_updates'):
domain = Domain()
domain.update()
status = resp.status_code
if 200 <= status < 300:
current_app.logger.debug("Request to powerdns API successful")
if Setting().get('enable_api_rr_history'):
2022-06-17 14:48:23 +00:00
if request.method in ['POST', 'PATCH']:
data = request.get_json(force=True)
history = History(
msg='Apply record changes to domain {0}'.format(zone_id.rstrip('.')),
detail = json.dumps({
'domain': zone_id.rstrip('.'),
'add_rrsets': list(filter(lambda r: r['changetype'] == "REPLACE", data['rrsets'])),
'del_rrsets': list(filter(lambda r: r['changetype'] == "DELETE", data['rrsets']))
}),
created_by=g.apikey.description,
domain_id=Domain().get_id_by_name(zone_id.rstrip('.')))
history.add()
elif request.method == 'DELETE':
history = History(msg='Deleted zone {0}'.format(zone_id.rstrip('.')),
detail='',
created_by=g.apikey.description,
domain_id=Domain().get_id_by_name(zone_id.rstrip('.')))
history.add()
elif request.method != 'GET':
history = History(msg='Updated zone {0}'.format(zone_id.rstrip('.')),
detail='',
created_by=g.apikey.description,
domain_id=Domain().get_id_by_name(zone_id.rstrip('.')))
history.add()
2019-03-01 22:49:31 +00:00
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/servers/<path:subpath>', methods=['GET', 'PUT'])
2019-03-01 22:49:31 +00:00
@apikey_auth
@apikey_is_admin
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_server_sub_forward(subpath):
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/servers/<string:server_id>/zones', methods=['POST'])
2019-03-01 22:49:31 +00:00
@apikey_auth
@apikey_can_create_domain
@csrf.exempt
2019-03-01 22:49:31 +00:00
def api_create_zone(server_id):
resp = helper.forward_request()
if resp.status_code == 201:
current_app.logger.debug("Request to powerdns API successful")
2019-03-01 22:49:31 +00:00
data = request.get_json(force=True)
if g.apikey.role.name not in ['Administrator', 'Operator']:
current_app.logger.debug(
"Apikey is user key, assigning created domain")
2019-03-01 22:49:31 +00:00
domain = Domain(name=data['name'].rstrip('.'))
g.apikey.domains.append(domain)
domain = Domain()
domain.update()
2021-03-27 18:33:11 +00:00
history = History(msg='Add domain {0}'.format(
data['name'].rstrip('.')),
detail=json.dumps(data),
created_by=g.apikey.description,
domain_id=domain.get_id_by_name(data['name'].rstrip('.')))
history.add()
2019-03-01 22:49:31 +00:00
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/servers/<string:server_id>/zones', methods=['GET'])
2019-03-01 22:49:31 +00:00
@apikey_auth
def api_get_zones(server_id):
if server_id == 'pdnsadmin':
2019-12-04 04:50:46 +00:00
if g.apikey.role.name not in ['Administrator', 'Operator']:
domain_obj_list = g.apikey.domains
else:
domain_obj_list = Domain.query.all()
return jsonify(domain_schema.dump(domain_obj_list)), 200
2019-03-01 22:49:31 +00:00
else:
2019-12-04 04:50:46 +00:00
resp = helper.forward_request()
2022-12-22 21:55:05 +00:00
if (g.apikey.role.name not in ['Administrator', 'Operator'] and resp.status_code == 200):
domain_list = [d['name']
for d in domain_schema.dump(g.apikey.domains)]
accounts_domains = [d.name for a in g.apikey.accounts for d in a.domains]
allowed_domains = set(domain_list + accounts_domains)
2022-12-22 21:55:05 +00:00
current_app.logger.debug("Account domains: {}".format('/'.join(accounts_domains)))
content = json.dumps([i for i in json.loads(resp.content)
if i['name'].rstrip('.') in allowed_domains])
return content, resp.status_code, resp.headers.items()
else:
return resp.content, resp.status_code, resp.headers.items()
@api_bp.route('/servers', methods=['GET'])
@apikey_auth
def api_server_forward():
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
2022-06-17 14:48:23 +00:00
@api_bp.route('/servers/<string:server_id>', methods=['GET'])
@apikey_auth
def api_server_config_forward(server_id):
resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items()
2022-12-22 21:55:05 +00:00
2022-01-20 11:49:37 +00:00
# The endpoint to synchronize Domains in background
@api_bp.route('/sync_domains', methods=['GET'])
@apikey_or_basic_auth
def sync_domains():
domain = Domain()
domain.update()
return 'Finished synchronization in background', 200
2022-04-18 09:01:22 +00:00
2022-12-22 21:55:05 +00:00
2022-04-18 09:01:22 +00:00
@api_bp.route('/health', methods=['GET'])
@apikey_auth
2022-04-18 09:01:22 +00:00
def health():
domain = Domain()
domain_to_query = domain.query.first()
2022-04-18 09:01:22 +00:00
if not domain_to_query:
current_app.logger.error("No domain found to query a health check")
2022-05-23 16:46:11 +00:00
return make_response("Unknown", 503)
2022-04-18 09:01:22 +00:00
try:
domain.get_domain_info(domain_to_query.name)
2022-04-18 09:01:22 +00:00
except Exception as e:
2022-12-22 21:55:05 +00:00
current_app.logger.error(
"Health Check - Failed to query authoritative server for domain {}".format(domain_to_query.name))
return make_response("Down", 503)
2022-04-18 09:01:22 +00:00
return make_response("Up", 200)