cwinfo/powerdns-admin
4
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2025-07-31 17:53:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactored zone history retrieval, parsing and displaying code.

Browse Source
This commit is contained in:
Rauno Tuul
2023-04-04 15:32:52 +03:00
parent 53cfa4fdaa
commit fe10665e19
5 changed files with 224 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
powerdnsadmin/routes/admin.py
View File

@@ -107,66 +107,58 @@ def get_record_changes(del_rrset, add_rrset):
return changeset
def filter_rr_list_by_name_and_type(rrset, record_name, record_type):
return list(filter(lambda rr: rr['name'] == record_name and rr['type'] == record_type, rrset))
# 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
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:
return
def extract_changelogs_from_history(histories, record_name=None, record_type=None):
out_changes = []
if "add_rrsets" in history_entry.detail:
detail_dict = json.loads(history_entry.detail)
else: # not a record entry
return
for entry in histories:
changes = []
add_rrsets = detail_dict['add_rrsets']
del_rrsets = detail_dict['del_rrsets']
for add_rrset in add_rrsets:
exists = False
for del_rrset in del_rrsets:
if del_rrset['name'] == add_rrset['name'] and del_rrset['type'] == add_rrset['type']:
exists = True
if change_num not in out_changes:
out_changes[change_num] = []
out_changes[change_num].append(HistoryRecordEntry(history_entry, del_rrset, add_rrset, "*"))
break
if not exists: # this is a new record
if change_num not in out_changes:
out_changes[change_num] = []
out_changes[change_num].append(
HistoryRecordEntry(history_entry, {}, add_rrset, "+")) # (add_rrset, del_rrset, change_type)
for del_rrset in del_rrsets:
exists = False
for add_rrset in add_rrsets:
if del_rrset['name'] == add_rrset['name'] and del_rrset['type'] == add_rrset['type']:
exists = True # no need to add in the out_changes set
break
if not exists: # this is a deletion
if change_num not in out_changes:
out_changes[change_num] = []
out_changes[change_num].append(HistoryRecordEntry(history_entry, del_rrset, {}, "-"))
# Sort them by the record name
if change_num in out_changes:
out_changes[change_num].sort(key=lambda change:
change.del_rrset['name'] if change.del_rrset else change.add_rrset['name']
)
# 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 change_num in out_changes:
changes_i = out_changes[change_num]
else:
return
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 "add_rrsets" in entry.detail:
details = json.loads(entry.detail)
if not details['add_rrsets'] and not details['del_rrsets']:
continue
elif 'type' in hre.del_rrset and hre.del_rrset['name'] == record_name and hre.del_rrset[
'type'] == record_type:
else: # not a record entry
continue
# filter only the records with the specific record_name, record_type
if record_name != None and record_type != None:
details['add_rrsets'] = list(filter_rr_list_by_name_and_type(details['add_rrsets'], record_name, record_type))
details['del_rrsets'] = list(filter_rr_list_by_name_and_type(details['del_rrsets'], record_name, record_type))
if not details['add_rrsets'] and not details['del_rrsets']:
continue
else:
out_changes[change_num].remove(hre)
# same record name and type RR are being deleted and created in same entry.
del_add_changes = set([(r['name'], r['type']) for r in details['add_rrsets']]).intersection([(r['name'], r['type']) for r in details['del_rrsets']])
for del_add_change in del_add_changes:
changes.append(HistoryRecordEntry(
entry,
filter_rr_list_by_name_and_type(details['del_rrsets'], del_add_change[0], del_add_change[1]).pop(0),
filter_rr_list_by_name_and_type(details['add_rrsets'], del_add_change[0], del_add_change[1]).pop(0),
"*")
)
for rrset in details['add_rrsets']:
if (rrset['name'], rrset['type']) not in del_add_changes:
changes.append(HistoryRecordEntry(entry, {}, rrset, "+"))
for rrset in details['del_rrsets']:
if (rrset['name'], rrset['type']) not in del_add_changes:
changes.append(HistoryRecordEntry(entry, rrset, {}, "-"))
# sort changes by the record name
if changes:
changes.sort(key=lambda change:
change.del_rrset['name'] if change.del_rrset else change.add_rrset['name']
)
out_changes.extend(changes)
return out_changes
# records with same (name,type) are considered as a single HistoryRecordEntry
@@ -184,21 +176,16 @@ class HistoryRecordEntry:
self.changed_fields = [] # contains a subset of : [ttl, name, type]
self.changeSet = [] # all changes for the records of this add_rrset-del_rrset pair
if change_type == "+": # addition
if change_type == "+" or change_type == "-":
self.changed_fields.append("name")
self.changed_fields.append("type")
self.changed_fields.append("ttl")
self.changeSet = get_record_changes(del_rrset, add_rrset)
elif change_type == "-": # removal
self.changed_fields.append("name")
self.changed_fields.append("type")
self.changed_fields.append("ttl")
self.changeSet = get_record_changes(del_rrset, add_rrset)
elif change_type == "*": # edit of unchanged
if add_rrset['ttl'] != del_rrset['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):
return {
@@ -882,9 +869,7 @@ class DetailedHistory():
ip_address=detail_dict['ip_address'])
elif 'add_rrsets' in detail_dict: # this is a zone record change
# changes_set = []
self.detailed_msg = ""
# extract_changelogs_from_a_history_entry(changes_set, history, 0)
elif 'name' in detail_dict and 'template' in history.msg: # template creation / deletion
self.detailed_msg = render_template_string("""
@@ -997,20 +982,14 @@ class DetailedHistory():
# convert a list of History objects into DetailedHistory objects
def convert_histories(histories):
changes_set = dict()
detailedHistories = []
j = 0
for i in range(len(histories)):
if histories[i].detail and ('add_rrsets' in histories[i].detail or 'del_rrsets' in histories[i].detail):
extract_changelogs_from_a_history_entry(changes_set, histories[i], j)
if j in changes_set:
detailedHistories.append(DetailedHistory(histories[i], changes_set[j]))
else: # no changes were found
detailedHistories.append(DetailedHistory(histories[i], None))
j += 1
for history in histories:
if history.detail and ('add_rrsets' in history.detail or 'del_rrsets' in history.detail):
history_as_list = list()
history_as_list.append(history)
detailedHistories.append(DetailedHistory(history, extract_changelogs_from_history(history_as_list)))
else:
detailedHistories.append(DetailedHistory(histories[i], None))
detailedHistories.append(DetailedHistory(history, None))
return detailedHistories

106
powerdnsadmin/routes/domain.py
View File

@@ -25,7 +25,7 @@ from ..models.domain_setting import DomainSetting
from ..models.base import db
from ..models.domain_user import DomainUser
from ..models.account_user import AccountUser
from .admin import extract_changelogs_from_a_history_entry
from .admin import extract_changelogs_from_history
from ..decorators import history_access_required
domain_bp = Blueprint('domain',
__name__,
@@ -201,17 +201,6 @@ def changelog(domain_name):
if not domain:
abort(404)
# Query domain's rrsets from PowerDNS API
rrsets = Record().get_rrsets(domain.name)
current_app.logger.debug("Fetched rrsets: \n{}".format(pretty_json(rrsets)))
# API server might be down, misconfigured
if not rrsets and domain.type != 'Slave':
abort(500)
records_allow_to_edit = Setting().get_records_allow_to_edit()
records = []
# get all changelogs for this domain, in descening order
if current_user.role.name in [ 'Administrator', 'Operator' ]:
histories = History.query.filter(History.domain_id == domain.id).order_by(History.created_on.desc()).all()
@@ -230,49 +219,13 @@ def changelog(domain_name):
DomainUser.user_id == current_user.id,
AccountUser.user_id == current_user.id
),
History.domain_id == domain.id
History.domain_id == domain.id,
History.detail.isnot(None)
)
).all()
if StrictVersion(Setting().get('pdns_version')) >= StrictVersion('4.0.0'):
for r in rrsets:
if r['type'] in records_allow_to_edit:
r_name = r['name'].rstrip('.')
changes_set = extract_changelogs_from_history(histories)
# If it is reverse zone and pretty_ipv6_ptr setting
# is enabled, we reformat the name for ipv6 records.
if Setting().get('pretty_ipv6_ptr') and r[
'type'] == 'PTR' and 'ip6.arpa' in r_name and '*' not in r_name:
r_name = dns.reversename.to_address(
dns.name.from_text(r_name))
# Create the list of records in format that
# PDA jinja2 template can understand.
index = 0
for record in r['records']:
if (len(r['comments'])>index):
c=r['comments'][index]['content']
else:
c=''
record_entry = RecordEntry(
name=r_name,
type=r['type'],
status='Disabled' if record['disabled'] else 'Active',
ttl=r['ttl'],
data=record['content'],
comment=c,
is_allowed_edit=True)
index += 1
records.append(record_entry)
else:
# Unsupported version
abort(500)
changes_set = dict()
for i in range(len(histories)):
extract_changelogs_from_a_history_entry(changes_set, histories[i], i)
if i in changes_set and len(changes_set[i]) == 0: # if empty, then remove the key
changes_set.pop(i)
return render_template('domain_changelog.html', domain=domain, allHistoryChanges=changes_set)
"""
@@ -289,17 +242,18 @@ def record_changelog(domain_name, record_name, record_type):
domain = Domain.query.filter(Domain.name == domain_name).first()
if not domain:
abort(404)
# Query domain's rrsets from PowerDNS API
rrsets = Record().get_rrsets(domain.name)
current_app.logger.debug("Fetched rrsets: \n{}".format(pretty_json(rrsets)))
# API server might be down, misconfigured
if not rrsets and domain.type != 'Slave':
abort(500)
# get all changelogs for this domain, in descening order
if current_user.role.name in [ 'Administrator', 'Operator' ]:
histories = History.query.filter(History.domain_id == domain.id).order_by(History.created_on.desc()).all()
histories = History.query \
.filter(
db.and_(
History.domain_id == domain.id,
History.detail.like("%{}%".format(record_name))
)
) \
.order_by(History.created_on.desc()) \
.all()
else:
# if the user isn't an administrator or operator,
# allow_user_view_history must be enabled to get here,
@@ -309,43 +263,23 @@ def record_changelog(domain_name, record_name, record_type):
.outerjoin(DomainUser, Domain.id == DomainUser.domain_id) \
.outerjoin(Account, Domain.account_id == Account.id) \
.outerjoin(AccountUser, Account.id == AccountUser.account_id) \
.order_by(History.created_on.desc()) \
.filter(
db.and_(db.or_(
DomainUser.user_id == current_user.id,
AccountUser.user_id == current_user.id
),
History.domain_id == domain.id
History.domain_id == domain.id,
History.detail.like("%{}%".format(record_name))
)
).all()
) \
.order_by(History.created_on.desc()) \
.all()
changes_set_of_record = dict()
for i in range(len(histories)):
extract_changelogs_from_a_history_entry(changes_set_of_record, histories[i], i, record_name, record_type)
if i in changes_set_of_record and len(changes_set_of_record[i]) == 0: # if empty, then remove the key
changes_set_of_record.pop(i)
changes_set = extract_changelogs_from_history(histories, record_name, record_type)
indexes_to_pop = []
for change_num in changes_set_of_record:
changes_i = changes_set_of_record[change_num]
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:
continue
elif 'type' in hre.del_rrset and hre.del_rrset['name'] == record_name and hre.del_rrset['type'] == record_type:
continue
else:
changes_set_of_record[change_num].remove(hre)
if change_num in changes_set_of_record and len(changes_set_of_record[change_num]) == 0: # if empty, then remove the key
indexes_to_pop.append(change_num)
for i in indexes_to_pop:
changes_set_of_record.pop(i)
return render_template('domain_changelog.html', domain=domain, allHistoryChanges=changes_set_of_record,
return render_template('domain_changelog.html', domain=domain, allHistoryChanges=changes_set,
record_name = record_name, record_type = record_type)
@domain_bp.route('/add', methods=['GET', 'POST'])
@login_required
@can_create_domain