cwinfo/powerdns-admin
5
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2024-11-09 15:10:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Code refactoring and bug fixes

Browse Source
This commit is contained in:
Khanh Ngo 2020-06-19 08:47:51 +07:00
parent 5eb2edee2c
commit a3fd856dd8
No known key found for this signature in database
GPG Key ID: 29077342AA5648F6
14 changed files with 152 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
powerdnsadmin/default_config.py
View File

@ -16,10 +16,10 @@ SQLA_DB_HOST = '127.0.0.1'
SQLA_DB_NAME = 'pda'
SQLALCHEMY_TRACK_MODIFICATIONS = True
### DATBASE - MySQL
### DATABASE - MySQL
SQLALCHEMY_DATABASE_URI = 'mysql://'+SQLA_DB_USER+':'+SQLA_DB_PASSWORD+'@'+SQLA_DB_HOST+'/'+SQLA_DB_NAME
### DATABSE - SQLite
### DATABASE - SQLite
# SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'pdns.db')
# SAML Authnetication

1
powerdnsadmin/lib/utils.py
View File

@ -221,6 +221,7 @@ def ensure_list(l):
yield from l
class customBoxes:
boxes = {
"reverse": (" ", " "),

4
powerdnsadmin/models/api_key.py
View File

@ -91,9 +91,9 @@ class ApiKey(db.Model):
current_app.config.get('SALT').encode('utf-8'))
def check_password(self, hashed_password):
# Check hased password. Using bcrypt,
# Check hashed password. Using bcrypt,
# the salt is saved into the hash itself
if (self.plain_text_password):
if self.plain_text_password:
return bcrypt.checkpw(self.plain_text_password.encode('utf-8'),
hashed_password.encode('utf-8'))
return False

45
powerdnsadmin/models/domain.py
View File

@ -74,8 +74,7 @@ class Domain(db.Model):
"""
Get all domains which has in PowerDNS
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain_name)),
@ -89,8 +88,7 @@ class Domain(db.Model):
"""
Get all domains which has in PowerDNS
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
jdata = utils.fetch_json(
urljoin(self.PDNS_STATS_URL,
self.API_EXTENDED_URL + '/servers/localhost/zones'),
@ -120,8 +118,7 @@ class Domain(db.Model):
dict_db_domain = dict((x.name, x) for x in db_domain)
current_app.logger.info("Found {} entries in PowerDNS-Admin".format(
len(list_db_domain)))
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(
urljoin(self.PDNS_STATS_URL,
@ -211,8 +208,7 @@ class Domain(db.Model):
Add a domain to power dns
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
domain_name = domain_name + '.'
domain_ns = [ns + '.' for ns in domain_ns]
@ -262,8 +258,7 @@ class Domain(db.Model):
"""
Read Domain from PowerDNS and add into PDNS-Admin
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
if not domain:
try:
domain = utils.fetch_json(
@ -315,8 +310,8 @@ class Domain(db.Model):
domain = Domain.query.filter(Domain.name == domain_name).first()
if not domain:
return {'status': 'error', 'msg': 'Domain does not exist.'}
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
if soa_edit_api not in ["DEFAULT", "INCREASE", "EPOCH", "OFF"]:
soa_edit_api = 'DEFAULT'
@ -365,8 +360,8 @@ class Domain(db.Model):
domain = Domain.query.filter(Domain.name == domain_name).first()
if not domain:
return {'status': 'error', 'msg': 'Domain does not exist.'}
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
post_data = {"kind": kind, "masters": masters}
@ -416,12 +411,12 @@ class Domain(db.Model):
system_auto_ptr = Setting().get('auto_ptr')
self.name = domain_name
domain_id = self.get_id_by_name(domain_reverse_name)
if None == domain_id and \
if domain_id is None and \
(
system_auto_ptr or
domain_auto_ptr
):
result = self.add(domain_reverse_name, 'Master', 'DEFAULT', '', '')
result = self.add(domain_reverse_name, 'Master', 'DEFAULT', [], [])
self.update()
if result['status'] == 'ok':
history = History(msg='Add reverse lookup domain {0}'.format(
@ -497,8 +492,7 @@ class Domain(db.Model):
"""
Delete a single domain name from powerdns
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL +
@ -587,8 +581,7 @@ class Domain(db.Model):
"""
domain = Domain.query.filter(Domain.name == domain_name).first()
if domain:
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
r = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL +
@ -618,8 +611,7 @@ class Domain(db.Model):
"""
domain = Domain.query.filter(Domain.name == domain_name).first()
if domain:
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(
urljoin(
@ -655,8 +647,7 @@ class Domain(db.Model):
"""
domain = Domain.query.filter(Domain.name == domain_name).first()
if domain:
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
# Enable API-RECTIFY for domain, BEFORE activating DNSSEC
post_data = {"api_rectify": True}
@ -722,8 +713,7 @@ class Domain(db.Model):
"""
domain = Domain.query.filter(Domain.name == domain_name).first()
if domain:
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
# Deactivate DNSSEC
jdata = utils.fetch_json(
@ -797,8 +787,7 @@ class Domain(db.Model):
if not domain:
return {'status': False, 'msg': 'Domain does not exist'}
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
account_name = Account().get_name_by_id(account_id)

2
powerdnsadmin/models/domain_setting.py
View File

@ -1,3 +1,5 @@
import traceback
from flask import current_app
from .base import db

2
powerdnsadmin/models/history.py
View File

@ -1,3 +1,5 @@
import traceback
from flask import current_app
from datetime import datetime

65
powerdnsadmin/models/record.py
View File

@ -14,12 +14,11 @@ from .setting import Setting
from .domain import Domain
from .domain_setting import DomainSetting
def byRecordContent(e):
return e['content']
def byRecordContentPair(e):
def by_record_content_pair(e):
return e[0]['content']
class Record(object):
"""
This is not a model, it's just an object
@ -49,8 +48,7 @@ class Record(object):
"""
Query domain's rrsets via PDNS API
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL +
@ -68,8 +66,8 @@ class Record(object):
rrsets=[]
for r in jdata['rrsets']:
while len(r['comments'])<len(r['records']):
r['comments'].append({"content":"", "account":""})
r['records'], r['comments'] = (list(t) for t in zip(*sorted(zip(r['records'],r['comments']),key=byRecordContentPair)))
r['comments'].append({"content": "", "account": ""})
r['records'], r['comments'] = (list(t) for t in zip(*sorted(zip(r['records'], r['comments']), key=by_record_content_pair)))
rrsets.append(r)
return rrsets
@ -98,8 +96,7 @@ class Record(object):
}
# Continue if the record is ready to be added
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(urljoin(
@ -129,26 +126,26 @@ class Record(object):
"""
Merge the rrsets that has same "name" and
"type".
Return: a new rrest which has multiple "records"
Return: a new rrset which has multiple "records"
and "comments"
"""
if not rrsets:
raise Exception("Empty rrsets to merge")
elif len(rrsets) == 1:
# It is unique rrest already
# It is unique rrset already
return rrsets[0]
else:
# Merge rrsets into one
rrest = rrsets[0]
rrset = rrsets[0]
for r in rrsets[1:]:
rrest['records'] = rrest['records'] + r['records']
rrest['comments'] = rrest['comments'] + r['comments']
while len(rrest['comments'])<len(rrest['records']):
rrest['comments'].append({"content":"", "account":""})
zipped_list = zip(rrest['records'],rrest['comments'])
tuples = zip(*sorted(zipped_list,key=byRecordContentPair))
rrest['records'], rrest['comments'] = [list(t) for t in tuples]
return rrest
rrset['records'] = rrset['records'] + r['records']
rrset['comments'] = rrset['comments'] + r['comments']
while len(rrset['comments']) < len(rrset['records']):
rrset['comments'].append({"content": "", "account": ""})
zipped_list = zip(rrset['records'], rrset['comments'])
tuples = zip(*sorted(zipped_list, key=by_record_content_pair))
rrset['records'], rrset['comments'] = [list(t) for t in tuples]
return rrset
def build_rrsets(self, domain_name, submitted_records):
"""
@ -159,7 +156,7 @@ class Record(object):
submitted_records(list): List of records submitted from PDA datatable
Returns:
transformed_rrsets(list): List of rrests converted from PDA datatable
transformed_rrsets(list): List of rrsets converted from PDA datatable
"""
rrsets = []
for record in submitted_records:
@ -192,7 +189,7 @@ class Record(object):
] and record["record_data"].strip()[-1:] != '.':
record["record_data"] += '.'
record_conntent = {
record_content = {
"content": record["record_data"],
"disabled":
False if record['record_status'] == 'Active' else True
@ -212,12 +209,12 @@ class Record(object):
"name": record_name,
"type": record["record_type"],
"ttl": int(record["record_ttl"]),
"records": [record_conntent],
"records": [record_content],
"comments": record_comments
})
# Group the records which has the same name and type.
# The rrest then has multiple records inside.
# The rrset then has multiple records inside.
transformed_rrsets = []
# Sort the list before using groupby
@ -238,8 +235,8 @@ class Record(object):
submitted_records(list): List of records submitted from PDA datatable
Returns:
new_rrsets(list): List of rrests to be added
del_rrsets(list): List of rrests to be deleted
new_rrsets(list): List of rrsets to be added
del_rrsets(list): List of rrsets to be deleted
"""
# Create submitted rrsets from submitted records
submitted_rrsets = self.build_rrsets(domain_name, submitted_records)
@ -285,7 +282,7 @@ class Record(object):
"""
Apply record changes to a domain. This function
will make 2 calls to the PDNS API to DELETE and
REPLACE records (rrests)
REPLACE records (rrsets)
"""
current_app.logger.debug(
"submitted_records: {}".format(submitted_records))
@ -293,7 +290,7 @@ class Record(object):
# Get the list of rrsets to be added and deleted
new_rrsets, del_rrsets = self.compare(domain_name, submitted_records)
# Remove blank comments from rrsets for compatability with some backends
# Remove blank comments from rrsets for compatibility with some backends
for r in new_rrsets['rrsets']:
if not r['comments']:
del r['comments']
@ -304,8 +301,7 @@ class Record(object):
# Submit the changes to PDNS API
try:
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
if del_rrsets["rrsets"]:
jdata1 = utils.fetch_json(urljoin(
@ -467,8 +463,7 @@ class Record(object):
"""
Delete a record from domain
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
data = {
"rrsets": [{
"name": self.name.rstrip('.') + '.',
@ -530,8 +525,7 @@ class Record(object):
"""
Update single record
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
data = {
"rrsets": [{
@ -572,8 +566,7 @@ class Record(object):
}
def update_db_serial(self, domain):
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain)),

9
powerdnsadmin/models/server.py
View File

@ -24,8 +24,7 @@ class Server(object):
"""
Get server config
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(urljoin(
@ -46,8 +45,7 @@ class Server(object):
"""
Get server statistics
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(urljoin(
@ -68,8 +66,7 @@ class Server(object):
"""
Search zone/record/comment directly from PDNS API
"""
headers = {}
headers['X-API-Key'] = self.PDNS_API_KEY
headers = {'X-API-Key': self.PDNS_API_KEY}
try:
jdata = utils.fetch_json(urljoin(

2
powerdnsadmin/models/setting.py
View File

@ -1,4 +1,6 @@
import sys
import traceback
import pytimeparse
from ast import literal_eval
from distutils.util import strtobool

2
powerdnsadmin/routes/user.py
View File

@ -50,7 +50,7 @@ def profile():
'password'] if 'password' in request.form else ''
else:
firstname = lastname = email = new_password = ''
logging.warning(
current_app.logger.warning(
'Authenticated externally. User {0} information will not allowed to update the profile'
.format(current_user.username))