cwinfo/powerdns-admin
5
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2024-11-08 22:50:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cosmetic

Browse Source
This commit is contained in:
Ymage 2022-12-22 22:55:05 +01:00
parent 7d153932b3
commit 1d885278d4
1 changed files with 51 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
powerdnsadmin/routes/api.py
View File

@ -1,22 +1,21 @@
import json import json
from urllib.parse import urljoin import secrets
import string
from base64 import b64encode from base64 import b64encode
from flask import ( from urllib.parse import urljoin
Blueprint, g, request, abort, current_app, make_response, jsonify,
) from flask import (Blueprint, g, request, abort, current_app, make_response, jsonify)
from flask_login import current_user from flask_login import current_user
from .base import csrf from .base import csrf
from ..models.base import db from ..decorators import (
from ..models import ( api_basic_auth, api_can_create_domain, is_json, apikey_auth,
User, Domain, DomainUser, Account, AccountUser, History, Setting, ApiKey, apikey_can_create_domain, apikey_can_remove_domain,
Role, 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 import utils, helper
from ..lib.schema import (
ApiKeySchema, DomainSchema, ApiPlainKeySchema, UserSchema, AccountSchema,
UserDetailedSchema,
)
from ..lib.errors import ( from ..lib.errors import (
StructuredException, StructuredException,
DomainNotExists, DomainAlreadyExists, DomainAccessForbidden, DomainNotExists, DomainAlreadyExists, DomainAccessForbidden,
@ -26,15 +25,15 @@ from ..lib.errors import (
UserCreateFail, UserCreateDuplicate, UserUpdateFail, UserDeleteFail, UserCreateFail, UserCreateDuplicate, UserUpdateFail, UserDeleteFail,
UserUpdateFailEmail, InvalidAccountNameException UserUpdateFailEmail, InvalidAccountNameException
) )
from ..decorators import ( from ..lib.schema import (
api_basic_auth, api_can_create_domain, is_json, apikey_auth, ApiKeySchema, DomainSchema, ApiPlainKeySchema, UserSchema, AccountSchema,
apikey_can_create_domain, apikey_can_remove_domain, UserDetailedSchema,
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
) )
import secrets from ..models import (
import string User, Domain, DomainUser, Account, AccountUser, History, Setting, ApiKey,
Role,
)
from ..models.base import db
api_bp = Blueprint('api', __name__, url_prefix='/api/v1') api_bp = Blueprint('api', __name__, url_prefix='/api/v1')
apilist_bp = Blueprint('apilist', __name__, url_prefix='/') apilist_bp = Blueprint('apilist', __name__, url_prefix='/')
@ -167,12 +166,7 @@ def handle_request_is_not_json(err):
def before_request(): def before_request():
# Check site is in maintenance mode # Check site is in maintenance mode
maintenance = Setting().get('maintenance') maintenance = Setting().get('maintenance')
if ( if (maintenance and current_user.is_authenticated and current_user.role.name not in ['Administrator', 'Operator']):
maintenance and current_user.is_authenticated and
current_user.role.name not in [
'Administrator', 'Operator'
]
):
return make_response( return make_response(
jsonify({ jsonify({
"status": False, "status": False,
@ -230,8 +224,7 @@ def api_login_create_zone():
history.add() history.add()
if current_user.role.name not in ['Administrator', 'Operator']: if current_user.role.name not in ['Administrator', 'Operator']:
current_app.logger.debug( current_app.logger.debug("User is ordinary user, assigning created domain")
"User is ordinary user, assigning created domain")
domain = Domain(name=data['name'].rstrip('.')) domain = Domain(name=data['name'].rstrip('.'))
domain.update() domain.update()
domain.grant_privileges([current_user.id]) domain.grant_privileges([current_user.id])
@ -326,14 +319,14 @@ def api_generate_apikey():
if 'domains' not in data: if 'domains' not in data:
domains = [] domains = []
elif not isinstance(data['domains'], (list, )): elif not isinstance(data['domains'], (list,)):
abort(400) abort(400)
else: else:
domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']] domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']]
if 'accounts' not in data: if 'accounts' not in data:
accounts = [] accounts = []
elif not isinstance(data['accounts'], (list, )): elif not isinstance(data['accounts'], (list,)):
abort(400) abort(400)
else: else:
accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']] accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']]
@ -385,8 +378,7 @@ def api_generate_apikey():
user_domain_list = [item.name for item in user_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("Input domain list: {0}".format(domain_list))
current_app.logger.debug( current_app.logger.debug("User domain list: {0}".format(user_domain_list))
"User domain list: {0}".format(user_domain_list))
inter = set(domain_list).intersection(set(user_domain_list)) inter = set(domain_list).intersection(set(user_domain_list))
@ -539,14 +531,14 @@ def api_update_apikey(apikey_id):
if 'domains' not in data: if 'domains' not in data:
domains = None domains = None
elif not isinstance(data['domains'], (list, )): elif not isinstance(data['domains'], (list,)):
abort(400) abort(400)
else: else:
domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']] domains = [d['name'] if isinstance(d, dict) else d for d in data['domains']]
if 'accounts' not in data: if 'accounts' not in data:
accounts = None accounts = None
elif not isinstance(data['accounts'], (list, )): elif not isinstance(data['accounts'], (list,)):
abort(400) abort(400)
else: else:
accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']] accounts = [a['name'] if isinstance(a, dict) else a for a in data['accounts']]
@ -963,9 +955,7 @@ def api_delete_account(account_id):
account = account_list[0] account = account_list[0]
else: else:
abort(404) abort(404)
current_app.logger.debug( current_app.logger.debug(f'Deleting Account {account.name}')
f'Deleting Account {account.name}'
)
# Remove account association from domains first # Remove account association from domains first
if len(account.domains) > 0: if len(account.domains) > 0:
@ -1192,17 +1182,13 @@ def api_get_zones(server_id):
return jsonify(domain_schema.dump(domain_obj_list)), 200 return jsonify(domain_schema.dump(domain_obj_list)), 200
else: else:
resp = helper.forward_request() resp = helper.forward_request()
if ( if (g.apikey.role.name not in ['Administrator', 'Operator'] and resp.status_code == 200):
g.apikey.role.name not in ['Administrator', 'Operator']
and resp.status_code == 200
):
domain_list = [d['name'] domain_list = [d['name']
for d in domain_schema.dump(g.apikey.domains)] for d in domain_schema.dump(g.apikey.domains)]
accounts_domains = [d.name for a in g.apikey.accounts for d in a.domains] accounts_domains = [d.name for a in g.apikey.accounts for d in a.domains]
allowed_domains = set(domain_list + accounts_domains) allowed_domains = set(domain_list + accounts_domains)
current_app.logger.debug("Account domains: {}".format( current_app.logger.debug("Account domains: {}".format('/'.join(accounts_domains)))
'/'.join(accounts_domains)))
content = json.dumps([i for i in json.loads(resp.content) content = json.dumps([i for i in json.loads(resp.content)
if i['name'].rstrip('.') in allowed_domains]) if i['name'].rstrip('.') in allowed_domains])
return content, resp.status_code, resp.headers.items() return content, resp.status_code, resp.headers.items()
@ -1223,6 +1209,7 @@ def api_server_config_forward(server_id):
resp = helper.forward_request() resp = helper.forward_request()
return resp.content, resp.status_code, resp.headers.items() return resp.content, resp.status_code, resp.headers.items()
# The endpoint to synchronize Domains in background # The endpoint to synchronize Domains in background
@api_bp.route('/sync_domains', methods=['GET']) @api_bp.route('/sync_domains', methods=['GET'])
@apikey_or_basic_auth @apikey_or_basic_auth
@ -1231,6 +1218,7 @@ def sync_domains():
domain.update() domain.update()
return 'Finished synchronization in background', 200 return 'Finished synchronization in background', 200
@api_bp.route('/health', methods=['GET']) @api_bp.route('/health', methods=['GET'])
@apikey_auth @apikey_auth
def health(): def health():
@ -1244,7 +1232,8 @@ def health():
try: try:
domain.get_domain_info(domain_to_query.name) domain.get_domain_info(domain_to_query.name)
except Exception as e: except Exception as e:
current_app.logger.error("Health Check - Failed to query authoritative server for domain {}".format(domain_to_query.name)) current_app.logger.error(
"Health Check - Failed to query authoritative server for domain {}".format(domain_to_query.name))
return make_response("Down", 503) return make_response("Down", 503)
return make_response("Up", 200) return make_response("Up", 200)