cwinfo/powerdns-admin
5
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2025-01-10 04:15:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Split the server statistics and configuration feature into separate pages.

Browse Source
This commit is contained in:
Matt Scott 2023-02-19 15:04:30 -05:00
parent 55e4f5f829
commit 65bfc53acb
5 changed files with 704 additions and 604 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

249
powerdnsadmin/routes/admin.py
View File

@ -4,7 +4,8 @@ import traceback
import re import re
from base64 import b64encode from base64 import b64encode
from ast import literal_eval from ast import literal_eval
from flask import Blueprint, render_template, render_template_string, make_response, url_for, current_app, request, redirect, jsonify, abort, flash, session from flask import Blueprint, render_template, render_template_string, make_response, url_for, current_app, request, \
redirect, jsonify, abort, flash, session
from flask_login import login_required, current_user from flask_login import login_required, current_user
from ..decorators import operator_role_required, admin_role_required, history_access_required from ..decorators import operator_role_required, admin_role_required, history_access_required
@ -44,6 +45,8 @@ change_type: "addition" or "deletion" or "status" for status change or "unchange
Note: A change in "content", is considered a deletion and recreation of the same record, Note: A change in "content", is considered a deletion and recreation of the same record,
holding the new content value. holding the new content value.
""" """
def get_record_changes(del_rrset, add_rrset): def get_record_changes(del_rrset, add_rrset):
changeSet = [] changeSet = []
delSet = del_rrset['records'] if 'records' in del_rrset else [] delSet = del_rrset['records'] if 'records' in del_rrset else []
@ -54,15 +57,15 @@ def get_record_changes(del_rrset, add_rrset):
if d['content'] == a['content']: if d['content'] == a['content']:
exists = True exists = True
if d['disabled'] != a['disabled']: if d['disabled'] != a['disabled']:
changeSet.append( ({"disabled":d['disabled'],"content":d['content']}, changeSet.append(({"disabled": d['disabled'], "content": d['content']},
{"disabled":a['disabled'],"content":a['content']}, {"disabled": a['disabled'], "content": a['content']},
"status") ) "status"))
break break
if not exists: # deletion if not exists: # deletion
changeSet.append( ({"disabled":d['disabled'],"content":d['content']}, changeSet.append(({"disabled": d['disabled'], "content": d['content']},
None, None,
"deletion") ) "deletion"))
for a in addSet: # get the additions for a in addSet: # get the additions
exists = False exists = False
@ -72,7 +75,7 @@ def get_record_changes(del_rrset, add_rrset):
# already checked for status change # already checked for status change
break break
if not exists: if not exists:
changeSet.append( (None, {"disabled":a['disabled'], "content":a['content']}, "addition") ) changeSet.append((None, {"disabled": a['disabled'], "content": a['content']}, "addition"))
continue continue
for a in addSet: # get the unchanged for a in addSet: # get the unchanged
@ -82,14 +85,15 @@ def get_record_changes(del_rrset, add_rrset):
exists = True exists = True
break break
if not exists: if not exists:
changeSet.append( ( {"disabled":a['disabled'], "content":a['content']}, {"disabled":a['disabled'], "content":a['content']}, "unchanged") ) changeSet.append(({"disabled": a['disabled'], "content": a['content']},
{"disabled": a['disabled'], "content": a['content']}, "unchanged"))
return changeSet return changeSet
# out_changes is a list of HistoryRecordEntry objects in which we will append the new changes # out_changes is a list of HistoryRecordEntry objects in which we will append the new changes
# a HistoryRecordEntry represents a pair of add_rrset and del_rrset # a HistoryRecordEntry represents a pair of add_rrset and del_rrset
def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_num, record_name=None, record_type=None): def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_num, record_name=None, record_type=None):
if history_entry.detail is None: if history_entry.detail is None:
return return
@ -101,7 +105,6 @@ def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_n
add_rrsets = detail_dict['add_rrsets'] add_rrsets = detail_dict['add_rrsets']
del_rrsets = detail_dict['del_rrsets'] del_rrsets = detail_dict['del_rrsets']
for add_rrset in add_rrsets: for add_rrset in add_rrsets:
exists = False exists = False
for del_rrset in del_rrsets: for del_rrset in del_rrsets:
@ -114,7 +117,8 @@ def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_n
if not exists: # this is a new record if not exists: # this is a new record
if change_num not in out_changes: if change_num not in out_changes:
out_changes[change_num] = [] out_changes[change_num] = []
out_changes[change_num].append(HistoryRecordEntry(history_entry, [], add_rrset, "+")) # (add_rrset, del_rrset, change_type) out_changes[change_num].append(
HistoryRecordEntry(history_entry, [], add_rrset, "+")) # (add_rrset, del_rrset, change_type)
for del_rrset in del_rrsets: for del_rrset in del_rrsets:
exists = False exists = False
for add_rrset in add_rrsets: for add_rrset in add_rrsets:
@ -126,7 +130,6 @@ def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_n
out_changes[change_num] = [] out_changes[change_num] = []
out_changes[change_num].append(HistoryRecordEntry(history_entry, del_rrset, [], "-")) out_changes[change_num].append(HistoryRecordEntry(history_entry, del_rrset, [], "-"))
# only used for changelog per record # only used for changelog per record
if record_name != None and record_type != None: # then get only the records with the specific (record_name, record_type) tuple if record_name != None and record_type != None: # then get only the records with the specific (record_name, record_type) tuple
if change_num in out_changes: if change_num in out_changes:
@ -134,15 +137,16 @@ def extract_changelogs_from_a_history_entry(out_changes, history_entry, change_n
else: else:
return return
for hre in changes_i: # for each history record entry in changes_i for hre in changes_i: # for each history record entry in changes_i
if 'type' in hre.add_rrset and hre.add_rrset['name'] == record_name and hre.add_rrset['type'] == record_type: if 'type' in hre.add_rrset and hre.add_rrset['name'] == record_name and hre.add_rrset[
'type'] == record_type:
continue continue
elif 'type' in hre.del_rrset and hre.del_rrset['name'] == record_name and hre.del_rrset['type'] == record_type: elif 'type' in hre.del_rrset and hre.del_rrset['name'] == record_name and hre.del_rrset[
'type'] == record_type:
continue continue
else: else:
out_changes[change_num].remove(hre) out_changes[change_num].remove(hre)
# records with same (name,type) are considered as a single HistoryRecordEntry # records with same (name,type) are considered as a single HistoryRecordEntry
# history_entry is of type History - used to extract created_by and created_on # history_entry is of type History - used to extract created_by and created_on
# add_rrset is a dictionary of replace # add_rrset is a dictionary of replace
@ -158,7 +162,6 @@ class HistoryRecordEntry:
self.changed_fields = [] # contains a subset of : [ttl, name, type] self.changed_fields = [] # contains a subset of : [ttl, name, type]
self.changeSet = [] # all changes for the records of this add_rrset-del_rrset pair self.changeSet = [] # all changes for the records of this add_rrset-del_rrset pair
if change_type == "+": # addition if change_type == "+": # addition
self.changed_fields.append("name") self.changed_fields.append("name")
self.changed_fields.append("type") self.changed_fields.append("type")
@ -175,22 +178,21 @@ class HistoryRecordEntry:
self.changed_fields.append("ttl") self.changed_fields.append("ttl")
self.changeSet = get_record_changes(del_rrset, add_rrset) self.changeSet = get_record_changes(del_rrset, add_rrset)
def toDict(self): def toDict(self):
return { return {
"add_rrset" : self.add_rrset, "add_rrset": self.add_rrset,
"del_rrset" : self.del_rrset, "del_rrset": self.del_rrset,
"changed_fields" : self.changed_fields, "changed_fields": self.changed_fields,
"created_on" : self.history_entry.created_on, "created_on": self.history_entry.created_on,
"created_by" : self.history_entry.created_by, "created_by": self.history_entry.created_by,
"change_type" : self.change_type, "change_type": self.change_type,
"changeSet" : self.changeSet "changeSet": self.changeSet
} }
def __eq__(self, obj2): # used for removal of objects from a list def __eq__(self, obj2): # used for removal of objects from a list
return True if obj2.toDict() == self.toDict() else False return True if obj2.toDict() == self.toDict() else False
@admin_bp.before_request @admin_bp.before_request
def before_request(): def before_request():
# Manage session timeout # Manage session timeout
@ -202,11 +204,10 @@ def before_request():
session.modified = True session.modified = True
@admin_bp.route('/server/statistics', methods=['GET'])
@admin_bp.route('/pdns', methods=['GET'])
@login_required @login_required
@operator_role_required @operator_role_required
def pdns_stats(): def server_statistics():
if not Setting().get('pdns_api_url') or not Setting().get( if not Setting().get('pdns_api_url') or not Setting().get(
'pdns_api_key') or not Setting().get('pdns_version'): 'pdns_api_key') or not Setting().get('pdns_version'):
return redirect(url_for('admin.setting_pdns')) return redirect(url_for('admin.setting_pdns'))
@ -215,7 +216,6 @@ def pdns_stats():
users = User.query.all() users = User.query.all()
server = Server(server_id='localhost') server = Server(server_id='localhost')
configs = server.get_config()
statistics = server.get_statistic() statistics = server.get_statistic()
history_number = History.query.count() history_number = History.query.count()
@ -226,12 +226,33 @@ def pdns_stats():
else: else:
uptime = 0 uptime = 0
return render_template('admin_pdns_stats.html', return render_template('admin_server_statistics.html',
domains=domains,
users=users,
statistics=statistics,
uptime=uptime,
history_number=history_number)
@admin_bp.route('/server/configuration', methods=['GET'])
@login_required
@operator_role_required
def server_configuration():
if not Setting().get('pdns_api_url') or not Setting().get(
'pdns_api_key') or not Setting().get('pdns_version'):
return redirect(url_for('admin.setting_pdns'))
domains = Domain.query.all()
users = User.query.all()
server = Server(server_id='localhost')
configs = server.get_config()
history_number = History.query.count()
return render_template('admin_server_configuration.html',
domains=domains, domains=domains,
users=users, users=users,
configs=configs, configs=configs,
statistics=statistics,
uptime=uptime,
history_number=history_number) history_number=history_number)
@ -296,6 +317,7 @@ def edit_user(user_username=None):
create=create, create=create,
error=result['msg']) error=result['msg'])
@admin_bp.route('/key/edit/<key_id>', methods=['GET', 'POST']) @admin_bp.route('/key/edit/<key_id>', methods=['GET', 'POST'])
@admin_bp.route('/key/edit', methods=['GET', 'POST']) @admin_bp.route('/key/edit', methods=['GET', 'POST'])
@login_required @login_required
@ -357,13 +379,13 @@ def edit_key(key_id=None):
try: try:
if role != "User": if role != "User":
domain_list, account_list = [], [] domain_list, account_list = [], []
apikey.update(role,description,domain_list, account_list) apikey.update(role, description, domain_list, account_list)
history_message = "Updated API key {0}".format(apikey.id) history_message = "Updated API key {0}".format(apikey.id)
except Exception as e: except Exception as e:
current_app.logger.error('Error: {0}'.format(e)) current_app.logger.error('Error: {0}'.format(e))
history = History(msg=history_message, history = History(msg=history_message,
detail = json.dumps({ detail=json.dumps({
'key': apikey.id, 'key': apikey.id,
'role': apikey.role.name, 'role': apikey.role.name,
'description': apikey.description, 'description': apikey.description,
@ -381,6 +403,7 @@ def edit_key(key_id=None):
create=create, create=create,
plain_key=plain_key) plain_key=plain_key)
@admin_bp.route('/manage-keys', methods=['GET', 'POST']) @admin_bp.route('/manage-keys', methods=['GET', 'POST'])
@login_required @login_required
@operator_role_required @operator_role_required
@ -404,7 +427,7 @@ def manage_keys():
history_apikey_id = apikey.id history_apikey_id = apikey.id
history_apikey_role = apikey.role.name history_apikey_role = apikey.role.name
history_apikey_description = apikey.description history_apikey_description = apikey.description
history_apikey_domains = [ domain.name for domain in apikey.domains] history_apikey_domains = [domain.name for domain in apikey.domains]
apikey.delete() apikey.delete()
except Exception as e: except Exception as e:
@ -412,7 +435,7 @@ def manage_keys():
current_app.logger.info('Delete API key {0}'.format(apikey.id)) current_app.logger.info('Delete API key {0}'.format(apikey.id))
history = History(msg='Delete API key {0}'.format(apikey.id), history = History(msg='Delete API key {0}'.format(apikey.id),
detail = json.dumps({ detail=json.dumps({
'key': history_apikey_id, 'key': history_apikey_id,
'role': history_apikey_role, 'role': history_apikey_role,
'description': history_apikey_description, 'description': history_apikey_description,
@ -427,6 +450,7 @@ def manage_keys():
'msg': 'Key has been removed.' 'msg': 'Key has been removed.'
}), 200) }), 200)
@admin_bp.route('/manage-user', methods=['GET', 'POST']) @admin_bp.route('/manage-user', methods=['GET', 'POST'])
@login_required @login_required
@operator_role_required @operator_role_required
@ -800,7 +824,9 @@ class DetailedHistory():
</table> </table>
""", """,
domaintype=detail_dict['domain_type'], domaintype=detail_dict['domain_type'],
account=Account.get_name_by_id(self=None, account_id=detail_dict['account_id']) if detail_dict['account_id'] != "0" else "None") account=Account.get_name_by_id(self=None, account_id=detail_dict[
'account_id']) if detail_dict[
'account_id'] != "0" else "None")
elif 'authenticator' in detail_dict: # this is a user authentication elif 'authenticator' in detail_dict: # this is a user authentication
self.detailed_msg = render_template_string(""" self.detailed_msg = render_template_string("""
@ -825,9 +851,11 @@ class DetailedHistory():
</tbody> </tbody>
</table> </table>
""", """,
background_rgba="68,157,68" if detail_dict['success'] == 1 else "201,48,44", background_rgba="68,157,68" if detail_dict[
'success'] == 1 else "201,48,44",
username=detail_dict['username'], username=detail_dict['username'],
auth_result="success" if detail_dict['success'] == 1 else "failure", auth_result="success" if detail_dict[
'success'] == 1 else "failure",
authenticator=detail_dict['authenticator'], authenticator=detail_dict['authenticator'],
ip_address=detail_dict['ip_address']) ip_address=detail_dict['ip_address'])
@ -844,7 +872,8 @@ class DetailedHistory():
</table> </table>
""", """,
template_name=DetailedHistory.get_key_val(detail_dict, "name"), template_name=DetailedHistory.get_key_val(detail_dict, "name"),
description=DetailedHistory.get_key_val(detail_dict, "description")) description=DetailedHistory.get_key_val(detail_dict,
"description"))
elif 'Change domain' in history.msg and 'access control' in history.msg: # added or removed a user from a domain elif 'Change domain' in history.msg and 'access control' in history.msg: # added or removed a user from a domain
users_with_access = DetailedHistory.get_key_val(detail_dict, "user_has_access") users_with_access = DetailedHistory.get_key_val(detail_dict, "user_has_access")
@ -868,9 +897,12 @@ class DetailedHistory():
""", """,
keyname=DetailedHistory.get_key_val(detail_dict, "key"), keyname=DetailedHistory.get_key_val(detail_dict, "key"),
rolename=DetailedHistory.get_key_val(detail_dict, "role"), rolename=DetailedHistory.get_key_val(detail_dict, "role"),
description=DetailedHistory.get_key_val(detail_dict, "description"), description=DetailedHistory.get_key_val(detail_dict,
linked_domains=DetailedHistory.get_key_val(detail_dict, "domains" if "domains" in detail_dict else "domain_acl"), "description"),
linked_accounts=DetailedHistory.get_key_val(detail_dict, "accounts")) linked_domains=DetailedHistory.get_key_val(detail_dict,
"domains" if "domains" in detail_dict else "domain_acl"),
linked_accounts=DetailedHistory.get_key_val(detail_dict,
"accounts"))
elif 'Delete API key' in history.msg: elif 'Delete API key' in history.msg:
self.detailed_msg = render_template_string(""" self.detailed_msg = render_template_string("""
@ -883,8 +915,10 @@ class DetailedHistory():
""", """,
keyname=DetailedHistory.get_key_val(detail_dict, "key"), keyname=DetailedHistory.get_key_val(detail_dict, "key"),
rolename=DetailedHistory.get_key_val(detail_dict, "role"), rolename=DetailedHistory.get_key_val(detail_dict, "role"),
description=DetailedHistory.get_key_val(detail_dict, "description"), description=DetailedHistory.get_key_val(detail_dict,
linked_domains=DetailedHistory.get_key_val(detail_dict, "domains")) "description"),
linked_domains=DetailedHistory.get_key_val(detail_dict,
"domains"))
elif 'Update type for domain' in history.msg: elif 'Update type for domain' in history.msg:
self.detailed_msg = render_template_string(""" self.detailed_msg = render_template_string("""
@ -905,8 +939,10 @@ class DetailedHistory():
<tr><td>Domain Master IPs:</td><td>{{ domain_master_ips }}</td></tr> <tr><td>Domain Master IPs:</td><td>{{ domain_master_ips }}</td></tr>
</table> </table>
""", """,
domain_type=DetailedHistory.get_key_val(detail_dict, "domain_type"), domain_type=DetailedHistory.get_key_val(detail_dict,
domain_master_ips=DetailedHistory.get_key_val(detail_dict, "domain_master_ips")) "domain_type"),
domain_master_ips=DetailedHistory.get_key_val(detail_dict,
"domain_master_ips"))
elif DetailedHistory.get_key_val(detail_dict, 'msg') and DetailedHistory.get_key_val(detail_dict, 'status'): elif DetailedHistory.get_key_val(detail_dict, 'msg') and DetailedHistory.get_key_val(detail_dict, 'status'):
self.detailed_msg = render_template_string(''' self.detailed_msg = render_template_string('''
@ -915,7 +951,8 @@ class DetailedHistory():
<tr><td>Message:</td><td>{{ history_msg }}</td></tr> <tr><td>Message:</td><td>{{ history_msg }}</td></tr>
</table> </table>
''', ''',
history_status=DetailedHistory.get_key_val(detail_dict, 'status'), history_status=DetailedHistory.get_key_val(detail_dict,
'status'),
history_msg=DetailedHistory.get_key_val(detail_dict, 'msg')) history_msg=DetailedHistory.get_key_val(detail_dict, 'msg'))
elif 'Update domain' in history.msg and 'associate account' in history.msg: # When an account gets associated or dissociate with domains elif 'Update domain' in history.msg and 'associate account' in history.msg: # When an account gets associated or dissociate with domains
@ -925,8 +962,10 @@ class DetailedHistory():
<tr><td>Dissociate:</td><td>{{ history_dissoc_account }}</td></tr> <tr><td>Dissociate:</td><td>{{ history_dissoc_account }}</td></tr>
</table> </table>
''', ''',
history_assoc_account=DetailedHistory.get_key_val(detail_dict, 'assoc_account'), history_assoc_account=DetailedHistory.get_key_val(detail_dict,
history_dissoc_account=DetailedHistory.get_key_val(detail_dict, 'dissoc_account')) 'assoc_account'),
history_dissoc_account=DetailedHistory.get_key_val(detail_dict,
'dissoc_account'))
# check for lower key as well for old databases # check for lower key as well for old databases
@staticmethod @staticmethod
@ -952,6 +991,7 @@ def convert_histories(histories):
detailedHistories.append(DetailedHistory(histories[i], None)) detailedHistories.append(DetailedHistory(histories[i], None))
return detailedHistories return detailedHistories
@admin_bp.route('/history', methods=['GET', 'POST']) @admin_bp.route('/history', methods=['GET', 'POST'])
@login_required @login_required
@history_access_required @history_access_required
@ -989,16 +1029,13 @@ def history():
'msg': 'Can not remove histories.' 'msg': 'Can not remove histories.'
}), 500) }), 500)
if request.method == 'GET': if request.method == 'GET':
doms = accounts = users = "" doms = accounts = users = ""
if current_user.role.name in [ 'Administrator', 'Operator']: if current_user.role.name in ['Administrator', 'Operator']:
all_domain_names = Domain.query.all() all_domain_names = Domain.query.all()
all_account_names = Account.query.all() all_account_names = Account.query.all()
all_user_names = User.query.all() all_user_names = User.query.all()
for d in all_domain_names: for d in all_domain_names:
doms += d.name + " " doms += d.name + " "
for acc in all_account_names: for acc in all_account_names:
@ -1026,7 +1063,6 @@ def history():
AccountUser.user_id == current_user.id AccountUser.user_id == current_user.id
)).all() )).all()
all_user_names = [] all_user_names = []
for a in all_account_names: for a in all_account_names:
temp = db.session.query(User) \ temp = db.session.query(User) \
@ -1051,17 +1087,21 @@ def history():
accounts += a.name + " " accounts += a.name + " "
for u in all_user_names: for u in all_user_names:
users += u.username + " " users += u.username + " "
return render_template('admin_history.html', all_domain_names=doms, all_account_names=accounts, all_usernames=users) return render_template('admin_history.html', all_domain_names=doms, all_account_names=accounts,
all_usernames=users)
# local_offset is the offset of the utc to the local time # local_offset is the offset of the utc to the local time
# offset must be int # offset must be int
# return the date converted and simplified # return the date converted and simplified
def from_utc_to_local(local_offset, timeframe): def from_utc_to_local(local_offset, timeframe):
offset = str(local_offset *(-1)) offset = str(local_offset * (-1))
date_split = str(timeframe).split(".")[0] date_split = str(timeframe).split(".")[0]
date_converted = datetime.datetime.strptime(date_split, '%Y-%m-%d %H:%M:%S') + datetime.timedelta(minutes=int(offset)) date_converted = datetime.datetime.strptime(date_split, '%Y-%m-%d %H:%M:%S') + datetime.timedelta(
minutes=int(offset))
return date_converted return date_converted
@admin_bp.route('/history_table', methods=['GET', 'POST']) @admin_bp.route('/history_table', methods=['GET', 'POST'])
@login_required @login_required
@history_access_required @history_access_required
@ -1097,7 +1137,7 @@ def history_table(): # ajax call data
lim = int(Setting().get('max_history_records')) # max num of records lim = int(Setting().get('max_history_records')) # max num of records
if request.method == 'GET': if request.method == 'GET':
if current_user.role.name in [ 'Administrator', 'Operator' ]: if current_user.role.name in ['Administrator', 'Operator']:
base_query = History.query base_query = History.query
else: else:
# if the user isn't an administrator or operator, # if the user isn't an administrator or operator,
@ -1115,28 +1155,35 @@ def history_table(): # ajax call data
)) ))
domain_name = request.args.get('domain_name_filter') if request.args.get('domain_name_filter') != None \ domain_name = request.args.get('domain_name_filter') if request.args.get('domain_name_filter') != None \
and len(request.args.get('domain_name_filter')) != 0 else None and len(
request.args.get('domain_name_filter')) != 0 else None
account_name = request.args.get('account_name_filter') if request.args.get('account_name_filter') != None \ account_name = request.args.get('account_name_filter') if request.args.get('account_name_filter') != None \
and len(request.args.get('account_name_filter')) != 0 else None and len(
request.args.get('account_name_filter')) != 0 else None
user_name = request.args.get('auth_name_filter') if request.args.get('auth_name_filter') != None \ user_name = request.args.get('auth_name_filter') if request.args.get('auth_name_filter') != None \
and len(request.args.get('auth_name_filter')) != 0 else None and len(request.args.get('auth_name_filter')) != 0 else None
min_date = request.args.get('min') if request.args.get('min') != None and len( request.args.get('min')) != 0 else None min_date = request.args.get('min') if request.args.get('min') != None and len(
request.args.get('min')) != 0 else None
if min_date != None: # get 1 day earlier, to check for timezone errors if min_date != None: # get 1 day earlier, to check for timezone errors
min_date = str(datetime.datetime.strptime(min_date, '%Y-%m-%d') - datetime.timedelta(days=1)) min_date = str(datetime.datetime.strptime(min_date, '%Y-%m-%d') - datetime.timedelta(days=1))
max_date = request.args.get('max') if request.args.get('max') != None and len( request.args.get('max')) != 0 else None max_date = request.args.get('max') if request.args.get('max') != None and len(
request.args.get('max')) != 0 else None
if max_date != None: # get 1 day later, to check for timezone errors if max_date != None: # get 1 day later, to check for timezone errors
max_date = str(datetime.datetime.strptime(max_date, '%Y-%m-%d') + datetime.timedelta(days=1)) max_date = str(datetime.datetime.strptime(max_date, '%Y-%m-%d') + datetime.timedelta(days=1))
tzoffset = request.args.get('tzoffset') if request.args.get('tzoffset') != None and len(request.args.get('tzoffset')) != 0 else None tzoffset = request.args.get('tzoffset') if request.args.get('tzoffset') != None and len(
request.args.get('tzoffset')) != 0 else None
changed_by = request.args.get('user_name_filter') if request.args.get('user_name_filter') != None \ changed_by = request.args.get('user_name_filter') if request.args.get('user_name_filter') != None \
and len(request.args.get('user_name_filter')) != 0 else None and len(
request.args.get('user_name_filter')) != 0 else None
""" """
Auth methods: LOCAL, Github OAuth, Azure OAuth, SAML, OIDC OAuth, Google OAuth Auth methods: LOCAL, Github OAuth, Azure OAuth, SAML, OIDC OAuth, Google OAuth
""" """
auth_methods = [] auth_methods = []
if (request.args.get('auth_local_only_checkbox') is None \ if (request.args.get('auth_local_only_checkbox') is None \
and request.args.get('auth_oauth_only_checkbox') is None \ and request.args.get('auth_oauth_only_checkbox') is None \
and request.args.get('auth_saml_only_checkbox') is None and request.args.get('auth_all_checkbox') is None): and request.args.get('auth_saml_only_checkbox') is None and request.args.get(
'auth_all_checkbox') is None):
auth_methods = [] auth_methods = []
if request.args.get('auth_all_checkbox') == "on": if request.args.get('auth_all_checkbox') == "on":
auth_methods.append("") auth_methods.append("")
@ -1152,11 +1199,8 @@ def history_table(): # ajax call data
else: else:
changelog_only = False changelog_only = False
# users cannot search for authentication # users cannot search for authentication
if user_name != None and current_user.role.name not in [ 'Administrator', 'Operator']: if user_name != None and current_user.role.name not in ['Administrator', 'Operator']:
histories = [] histories = []
elif domain_name != None: elif domain_name != None:
@ -1165,8 +1209,11 @@ def history_table(): # ajax call data
.filter( .filter(
db.and_( db.and_(
db.or_( db.or_(
History.msg.like("%domain "+ domain_name) if domain_name != "*" else History.msg.like("%domain%"), History.msg.like("%domain " + domain_name) if domain_name != "*" else History.msg.like(
History.msg.like("%domain "+ domain_name + " access control") if domain_name != "*" else History.msg.like("%domain%access control") "%domain%"),
History.msg.like(
"%domain " + domain_name + " access control") if domain_name != "*" else History.msg.like(
"%domain%access control")
), ),
History.created_on <= max_date if max_date != None else True, History.created_on <= max_date if max_date != None else True,
History.created_on >= min_date if min_date != None else True, History.created_on >= min_date if min_date != None else True,
@ -1214,14 +1261,19 @@ def history_table(): # ajax call data
) )
).order_by(History.created_on.desc()) \ ).order_by(History.created_on.desc()) \
.limit(lim).all() .limit(lim).all()
elif user_name != None and current_user.role.name in [ 'Administrator', 'Operator']: # only admins can see the user login-logouts elif user_name != None and current_user.role.name in ['Administrator',
'Operator']: # only admins can see the user login-logouts
histories = History.query \ histories = History.query \
.filter( .filter(
db.and_( db.and_(
db.or_( db.or_(
History.msg.like("User "+ user_name + " authentication%") if user_name != "*" and user_name != None else History.msg.like("%authentication%"), History.msg.like(
History.msg.like("User "+ user_name + " was not authorized%") if user_name != "*" and user_name != None else History.msg.like("User%was not authorized%") "User " + user_name + " authentication%") if user_name != "*" and user_name != None else History.msg.like(
"%authentication%"),
History.msg.like(
"User " + user_name + " was not authorized%") if user_name != "*" and user_name != None else History.msg.like(
"User%was not authorized%")
), ),
History.created_on <= max_date if max_date != None else True, History.created_on <= max_date if max_date != None else True,
History.created_on >= min_date if min_date != None else True, History.created_on >= min_date if min_date != None else True,
@ -1236,7 +1288,8 @@ def history_table(): # ajax call data
temp.append(h) temp.append(h)
break break
histories = temp histories = temp
elif (changed_by != None or max_date != None) and current_user.role.name in [ 'Administrator', 'Operator'] : # select changed by and date filters only elif (changed_by != None or max_date != None) and current_user.role.name in ['Administrator',
'Operator']: # select changed by and date filters only
histories = History.query \ histories = History.query \
.filter( .filter(
db.and_( db.and_(
@ -1246,7 +1299,8 @@ def history_table(): # ajax call data
) )
) \ ) \
.order_by(History.created_on.desc()).limit(lim).all() .order_by(History.created_on.desc()).limit(lim).all()
elif (changed_by != None or max_date != None): # special filtering for user because one user does not have access to log-ins logs elif (
changed_by != None or max_date != None): # special filtering for user because one user does not have access to log-ins logs
histories = base_query \ histories = base_query \
.filter( .filter(
db.and_( db.and_(
@ -1264,7 +1318,7 @@ def history_table(): # ajax call data
) )
).order_by(History.created_on.desc()).limit(lim).all() ).order_by(History.created_on.desc()).limit(lim).all()
else: # default view else: # default view
if current_user.role.name in [ 'Administrator', 'Operator']: if current_user.role.name in ['Administrator', 'Operator']:
histories = History.query.order_by(History.created_on.desc()).limit(lim).all() histories = History.query.order_by(History.created_on.desc()).limit(lim).all()
else: else:
histories = db.session.query(History) \ histories = db.session.query(History) \
@ -1289,14 +1343,15 @@ def history_table(): # ajax call data
max_date_split = max_date.split()[0] max_date_split = max_date.split()[0]
for i, history_rec in enumerate(detailedHistories): for i, history_rec in enumerate(detailedHistories):
local_date = str(from_utc_to_local(int(tzoffset), history_rec.history.created_on).date()) local_date = str(from_utc_to_local(int(tzoffset), history_rec.history.created_on).date())
if (min_date != None and local_date == min_date_split) or (max_date != None and local_date == max_date_split): if (min_date != None and local_date == min_date_split) or (
max_date != None and local_date == max_date_split):
detailedHistories[i] = None detailedHistories[i] = None
# Remove elements previously flagged as None # Remove elements previously flagged as None
detailedHistories = [h for h in detailedHistories if h is not None] detailedHistories = [h for h in detailedHistories if h is not None]
return render_template('admin_history_table.html', histories=detailedHistories, len_histories=len(detailedHistories), lim=lim) return render_template('admin_history_table.html', histories=detailedHistories,
len_histories=len(detailedHistories), lim=lim)
@admin_bp.route('/setting/basic', methods=['GET']) @admin_bp.route('/setting/basic', methods=['GET'])
@ -1528,7 +1583,7 @@ def setting_authentication():
Setting().set('autoprovisioning_attribute', Setting().set('autoprovisioning_attribute',
request.form.get('autoprovisioning_attribute')) request.form.get('autoprovisioning_attribute'))
if request.form.get('autoprovisioning')=='ON': if request.form.get('autoprovisioning') == 'ON':
if validateURN(request.form.get('urn_value')): if validateURN(request.form.get('urn_value')):
Setting().set('urn_value', Setting().set('urn_value',
request.form.get('urn_value')) request.form.get('urn_value'))
@ -1542,7 +1597,6 @@ def setting_authentication():
Setting().set('purge', True Setting().set('purge', True
if request.form.get('purge') == 'ON' else False) if request.form.get('purge') == 'ON' else False)
result = {'status': True, 'msg': 'Saved successfully'} result = {'status': True, 'msg': 'Saved successfully'}
elif conf_type == 'google': elif conf_type == 'google':
google_oauth_enabled = True if request.form.get( google_oauth_enabled = True if request.form.get(
@ -1738,7 +1792,7 @@ def create_template():
result = t.create() result = t.create()
if result['status'] == 'ok': if result['status'] == 'ok':
history = History(msg='Add domain template {0}'.format(name), history = History(msg='Add domain template {0}'.format(name),
detail = json.dumps({ detail=json.dumps({
'name': name, 'name': name,
'description': description 'description': description
}), }),
@ -1785,7 +1839,7 @@ def create_template_from_zone():
result = t.create() result = t.create()
if result['status'] == 'ok': if result['status'] == 'ok':
history = History(msg='Add domain template {0}'.format(name), history = History(msg='Add domain template {0}'.format(name),
detail = json.dumps({ detail=json.dumps({
'name': name, 'name': name,
'description': description 'description': description
}), }),
@ -1918,7 +1972,7 @@ def apply_records(template):
history = History( history = History(
msg='Apply domain template record changes to domain template {0}' msg='Apply domain template record changes to domain template {0}'
.format(template), .format(template),
detail = json.dumps(jdata), detail=json.dumps(jdata),
created_by=current_user.username) created_by=current_user.username)
history.add() history.add()
return make_response(jsonify(result), 200) return make_response(jsonify(result), 200)
@ -1948,7 +2002,7 @@ def delete_template(template):
if result['status'] == 'ok': if result['status'] == 'ok':
history = History( history = History(
msg='Deleted domain template {0}'.format(template), msg='Deleted domain template {0}'.format(template),
detail = json.dumps({'name': template}), detail=json.dumps({'name': template}),
created_by=current_user.username) created_by=current_user.username)
history.add() history.add()
return redirect(url_for('admin.templates')) return redirect(url_for('admin.templates'))
@ -2003,28 +2057,29 @@ def global_search():
return render_template('admin_global_search.html', domains=domains, records=records, comments=comments) return render_template('admin_global_search.html', domains=domains, records=records, comments=comments)
def validateURN(value): def validateURN(value):
NID_PATTERN = re.compile(r'^[0-9a-z][0-9a-z-]{1,31}$', flags=re.IGNORECASE) NID_PATTERN = re.compile(r'^[0-9a-z][0-9a-z-]{1,31}$', flags=re.IGNORECASE)
NSS_PCHAR = '[a-z0-9-._~]|%[a-f0-9]{2}|[!$&\'()*+,;=]|:|@' NSS_PCHAR = '[a-z0-9-._~]|%[a-f0-9]{2}|[!$&\'()*+,;=]|:|@'
NSS_PATTERN = re.compile(fr'^({NSS_PCHAR})({NSS_PCHAR}|/|\?)*$', re.IGNORECASE) NSS_PATTERN = re.compile(fr'^({NSS_PCHAR})({NSS_PCHAR}|/|\?)*$', re.IGNORECASE)
prefix=value.split(':') prefix = value.split(':')
if (len(prefix)<3): if (len(prefix) < 3):
current_app.logger.warning( "Too small urn prefix" ) current_app.logger.warning("Too small urn prefix")
return False return False
urn=prefix[0] urn = prefix[0]
nid=prefix[1] nid = prefix[1]
nss=value.replace(urn+":"+nid+":", "") nss = value.replace(urn + ":" + nid + ":", "")
if not urn.lower()=="urn": if not urn.lower() == "urn":
current_app.logger.warning( urn + ' contains invalid characters ' ) current_app.logger.warning(urn + ' contains invalid characters ')
return False return False
if not re.match(NID_PATTERN, nid.lower()): if not re.match(NID_PATTERN, nid.lower()):
current_app.logger.warning( nid + ' contains invalid characters ' ) current_app.logger.warning(nid + ' contains invalid characters ')
return False return False
if not re.match(NSS_PATTERN, nss): if not re.match(NSS_PATTERN, nss):
current_app.logger.warning( nss + ' contains invalid characters ' ) current_app.logger.warning(nss + ' contains invalid characters ')
return False return False
return True return True

127
powerdnsadmin/templates/admin_pdns_stats.html
View File

@ -1,127 +0,0 @@
{% extends "base.html" %}
{% set active_page = "admin_console" %}
{% block title %}
<title>
Admin Console - {{ SITE_NAME }}
</title>
{% endblock %}
{% block dashboard_stat %}
<div class="content-header">
<div class="container-fluid">
<div class="row mb-2">
<div class="col-sm-6">
<h1 class="m-0 text-dark">
PowerDNS
<small>Server Statistics & Configuration</small>
</h1>
</div>
<div class="col-sm-6">
<ol class="breadcrumb float-sm-right">
<li class="breadcrumb-item"><a href="{{ url_for('dashboard.dashboard') }}">Dashboard</a></li>
<li class="breadcrumb-item active">PowerDNS Server Statistics & Configuration</li>
</ol>
</div>
</div>
</div>
</div>
{% endblock %}
{% block content %}
<section class="content">
<div class="container-fluid">
<div class="row">
<div class="col-12">
<div class="card shadow">
<div class="card-header">
<h3 class="card-title">PowerDNS Server Statistics</h3>
</div>
<div class="card-body">
<table id="tbl_statistics" class="table table-bordered table-striped">
<thead>
<tr>
<th width="30%">Statistic</th>
<th>Value</th>
</tr>
</thead>
<tbody>
{% for statistic in statistics %}
<tr class="odd gradeX">
<td>
<a href="https://doc.powerdns.com/authoritative/search.html?q={{ statistic['name'] }}"
target="_blank" class="btn btn-primary">
<i class="fa fa-search"></i>&nbsp;{{ statistic['name'] }}
</a>
</td>
<td>{{ statistic['value'] }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>
<div class="row">
<div class="col-12">
<div class="card shadow">
<div class="card-header">
<h3 class="card-title">PowerDNS Server Configuration</h3>
</div>
<div class="card-body">
<table id="tbl_configuration" class="table table-bordered table-striped">
<thead>
<tr>
<th width="30%">Configuration</th>
<th>Value</th>
</tr>
</thead>
<tbody>
{% for config in configs %}
<tr class="odd gradeX">
<td>
<a href="https://doc.powerdns.com/authoritative/search.html?q={{ config['name'] }}"
target="_blank" class="btn btn-primary">
<i class="fa-solid fa-search"></i>&nbsp;{{ config['name'] }}
</a>
</td>
<td>
{{ config['value'] }}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>
</div>
</section>
{% endblock %}
{% block extrascripts %}
<script>
// set up statistics data table
$("#tbl_statistics").DataTable({
"paging": true,
"lengthChange": true,
"searching": true,
"ordering": true,
"info": true,
"autoWidth": false
});
// set up configuration data table
$("#tbl_configuration").DataTable({
"paging": true,
"lengthChange": true,
"searching": true,
"ordering": true,
"info": true,
"autoWidth": false
});
</script>
{% endblock %}

84
powerdnsadmin/templates/admin_server_configuration.html Normal file
View File

@ -0,0 +1,84 @@
{% extends "base.html" %}
{% set active_page = "server_configuration" %}
{% block title %}<title>Server Configuration - {{ SITE_NAME }}</title>{% endblock %}
{% block dashboard_stat %}
<div class="content-header">
<div class="container-fluid">
<div class="row mb-2">
<div class="col-sm-6">
<h1 class="m-0 text-dark">
Server Configuration
</h1>
</div>
<div class="col-sm-6">
<ol class="breadcrumb float-sm-right">
<li class="breadcrumb-item"><a href="{{ url_for('dashboard.dashboard') }}">Dashboard</a></li>
<li class="breadcrumb-item active">Server Configuration</li>
</ol>
</div>
</div>
</div>
</div>
{% endblock %}
{% block content %}
<section class="content">
<div class="container-fluid">
<div class="row">
<div class="col-12">
<div class="card shadow">
<div class="card-header">
<h3 class="card-title">Server Configuration</h3>
</div>
<!-- /.card-header -->
<div class="card-body table-responsive">
<table id="tbl_configuration" class="table table-bordered table-striped table-hover table-sm">
<thead>
<tr>
<th>Configuration</th>
<th>Value</th>
</tr>
</thead>
<tbody>
{% for config in configs %}
<tr class="odd gradeX">
<td>
<a href="https://doc.powerdns.com/authoritative/search.html?q={{ config['name'] }}"
target="_blank" class="btn btn-primary">
<i class="fa-solid fa-search"></i>&nbsp;{{ config['name'] }}
</a>
</td>
<td>
{{ config['value'] }}
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<!-- /.card-body -->
</div>
<!-- /.card -->
</div>
<!-- /.col -->
</div>
<!-- /.row -->
</div>
<!-- /.container-fluid -->
</section>
{% endblock %}
{% block extrascripts %}
<script>
// Initialize DataTables
$("#tbl_configuration").DataTable({
"paging": true,
"lengthChange": true,
"searching": true,
"ordering": true,
"info": true,
"autoWidth": false
});
</script>
{% endblock %}

82
powerdnsadmin/templates/admin_server_statistics.html Normal file
View File

@ -0,0 +1,82 @@
{% extends "base.html" %}
{% set active_page = "server_statistics" %}
{% block title %}<title>Server Statistics - {{ SITE_NAME }}</title>{% endblock %}
{% block dashboard_stat %}
<div class="content-header">
<div class="container-fluid">
<div class="row mb-2">
<div class="col-sm-6">
<h1 class="m-0 text-dark">
Server Statistics
</h1>
</div>
<div class="col-sm-6">
<ol class="breadcrumb float-sm-right">
<li class="breadcrumb-item"><a href="{{ url_for('dashboard.dashboard') }}">Dashboard</a></li>
<li class="breadcrumb-item active">Server Statistics</li>
</ol>
</div>
</div>
</div>
</div>
{% endblock %}
{% block content %}
<section class="content">
<div class="container-fluid">
<div class="row">
<div class="col-12">
<div class="card shadow">
<div class="card-header">
<h3 class="card-title">Server Statistics</h3>
</div>
<!-- /.card-header -->
<div class="card-body table-responsive">
<table id="tbl_statistics" class="table table-bordered table-striped table-hover table-sm">
<thead>
<tr>
<th width="30%">Statistic</th>
<th>Value</th>
</tr>
</thead>
<tbody>
{% for statistic in statistics %}
<tr class="odd gradeX">
<td>
<a href="https://doc.powerdns.com/authoritative/search.html?q={{ statistic['name'] }}"
target="_blank" class="btn btn-primary">
<i class="fa fa-search"></i>&nbsp;{{ statistic['name'] }}
</a>
</td>
<td>{{ statistic['value'] }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<!-- /.card-body -->
</div>
<!-- /.card -->
</div>
<!-- /.col -->
</div>
<!-- /.row -->
</div>
<!-- /.container-fluid -->
</section>
{% endblock %}
{% block extrascripts %}
<script>
// Initialize DataTables
$("#tbl_statistics").DataTable({
"paging": true,
"lengthChange": true,
"searching": true,
"ordering": true,
"info": true,
"autoWidth": false
});
</script>
{% endblock %}

14
powerdnsadmin/templates/base.html
View File

@ -115,10 +115,16 @@
{% endif %} {% endif %}
{% if current_user.role.name in ['Administrator', 'Operator'] %} {% if current_user.role.name in ['Administrator', 'Operator'] %}
<li class="nav-header">Administration</li> <li class="nav-header">Administration</li>
<li class="{{ 'nav-item active' if active_page == 'admin_console' else 'nav-item' }}"> <li class="{{ 'nav-item active' if active_page == 'server_statistics' else 'nav-item' }}">
<a href="{{ url_for('admin.pdns_stats') }}" class="nav-link"> <a href="{{ url_for('admin.server_statistics') }}" class="nav-link">
<i class="nav-icon fa-solid fa-info-circle"></i> <i class="nav-icon fa-solid fa-chart-simple"></i>
<p>PowerDNS Info</p> <p>Server Statistics</p>
</a>
</li>
<li class="{{ 'nav-item active' if active_page == 'server_configuration' else 'nav-item' }}">
<a href="{{ url_for('admin.server_configuration') }}" class="nav-link">
<i class="nav-icon fa-solid fa-cog"></i>
<p>Server Configuration</p>
</a> </a>
</li> </li>
<li class="{{ 'nav-item active' if active_page == 'admin_global_search' else 'nav-item' }}"> <li class="{{ 'nav-item active' if active_page == 'admin_global_search' else 'nav-item' }}">