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

Merge pull request #760 from ngoduykhanh/refactoring

Browse Source 
Code refactoring and bug fixes
This commit is contained in:
Khanh Ngo 2020-06-19 09:16:08 +07:00 committed by GitHub
commit 94da9198c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 152 additions and 166 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
powerdnsadmin/default_config.py
View File

@ -16,12 +16,12 @@ SQLA_DB_HOST = '127.0.0.1'
SQLA_DB_NAME = 'pda' SQLA_DB_NAME = 'pda'
SQLALCHEMY_TRACK_MODIFICATIONS = True SQLALCHEMY_TRACK_MODIFICATIONS = True
### DATBASE - MySQL ### DATABASE - MySQL
SQLALCHEMY_DATABASE_URI = 'mysql://'+SQLA_DB_USER+':'+SQLA_DB_PASSWORD+'@'+SQLA_DB_HOST+'/'+SQLA_DB_NAME 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') # SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'pdns.db')
# SAML Authnetication # SAML Authnetication
SAML_ENABLED = False SAML_ENABLED = False
SAML_ASSERTION_ENCRYPTED = True SAML_ASSERTION_ENCRYPTED = True

1
powerdnsadmin/lib/utils.py
View File

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

2
powerdnsadmin/models/account_user.py
View File

@ -14,4 +14,4 @@ class AccountUser(db.Model):
self.user_id = user_id self.user_id = user_id
def __repr__(self): def __repr__(self):
return '<Account_User {0} {1}>'.format(self.account_id, self.user_id) return '<Account_User {0} {1}>'.format(self.account_id, self.user_id)

4
powerdnsadmin/models/api_key.py
View File

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

215
powerdnsadmin/models/domain.py
View File

@ -74,23 +74,21 @@ class Domain(db.Model):
""" """
Get all domains which has in PowerDNS 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( jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain_name)), '/servers/localhost/zones/{0}'.format(domain_name)),
headers=headers, headers=headers,
timeout=int( timeout=int(
Setting().get('pdns_api_timeout')), Setting().get('pdns_api_timeout')),
verify=Setting().get('verify_ssl_connections')) verify=Setting().get('verify_ssl_connections'))
return jdata return jdata
def get_domains(self): def get_domains(self):
""" """
Get all domains which has in PowerDNS 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( jdata = utils.fetch_json(
urljoin(self.PDNS_STATS_URL, urljoin(self.PDNS_STATS_URL,
self.API_EXTENDED_URL + '/servers/localhost/zones'), 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) dict_db_domain = dict((x.name, x) for x in db_domain)
current_app.logger.info("Found {} entries in PowerDNS-Admin".format( current_app.logger.info("Found {} entries in PowerDNS-Admin".format(
len(list_db_domain))) len(list_db_domain)))
headers = {} headers = {'X-API-Key': self.PDNS_API_KEY}
headers['X-API-Key'] = self.PDNS_API_KEY
try: try:
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin(self.PDNS_STATS_URL, urljoin(self.PDNS_STATS_URL,
@ -211,8 +208,7 @@ class Domain(db.Model):
Add a domain to power dns 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_name = domain_name + '.'
domain_ns = [ns + '.' for ns in domain_ns] domain_ns = [ns + '.' for ns in domain_ns]
@ -262,15 +258,14 @@ class Domain(db.Model):
""" """
Read Domain from PowerDNS and add into PDNS-Admin 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: if not domain:
try: try:
domain = utils.fetch_json( domain = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format( '/servers/localhost/zones/{0}'.format(
domain_dict['name'])), domain_dict['name'])),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
verify=Setting().get('verify_ssl_connections')) verify=Setting().get('verify_ssl_connections'))
@ -315,8 +310,8 @@ class Domain(db.Model):
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if not domain: if not domain:
return {'status': 'error', 'msg': 'Domain does not exist.'} 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"]: if soa_edit_api not in ["DEFAULT", "INCREASE", "EPOCH", "OFF"]:
soa_edit_api = 'DEFAULT' soa_edit_api = 'DEFAULT'
@ -329,13 +324,13 @@ class Domain(db.Model):
try: try:
jdata = utils.fetch_json(urljoin( jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain.name)), '/servers/localhost/zones/{0}'.format(domain.name)),
headers=headers, headers=headers,
timeout=int( timeout=int(
Setting().get('pdns_api_timeout')), Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
verify=Setting().get('verify_ssl_connections'), verify=Setting().get('verify_ssl_connections'),
data=post_data) data=post_data)
if 'error' in jdata.keys(): if 'error' in jdata.keys():
current_app.logger.error(jdata['error']) current_app.logger.error(jdata['error'])
return {'status': 'error', 'msg': jdata['error']} return {'status': 'error', 'msg': jdata['error']}
@ -365,21 +360,21 @@ class Domain(db.Model):
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if not domain: if not domain:
return {'status': 'error', 'msg': 'Domain does not exist.'} 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} post_data = {"kind": kind, "masters": masters}
try: try:
jdata = utils.fetch_json(urljoin( jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain.name)), '/servers/localhost/zones/{0}'.format(domain.name)),
headers=headers, headers=headers,
timeout=int( timeout=int(
Setting().get('pdns_api_timeout')), Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
verify=Setting().get('verify_ssl_connections'), verify=Setting().get('verify_ssl_connections'),
data=post_data) data=post_data)
if 'error' in jdata.keys(): if 'error' in jdata.keys():
current_app.logger.error(jdata['error']) current_app.logger.error(jdata['error'])
return {'status': 'error', 'msg': jdata['error']} return {'status': 'error', 'msg': jdata['error']}
@ -410,27 +405,27 @@ class Domain(db.Model):
domain_obj = Domain.query.filter(Domain.name == domain_name).first() domain_obj = Domain.query.filter(Domain.name == domain_name).first()
domain_auto_ptr = DomainSetting.query.filter( domain_auto_ptr = DomainSetting.query.filter(
DomainSetting.domain == domain_obj).filter( DomainSetting.domain == domain_obj).filter(
DomainSetting.setting == 'auto_ptr').first() DomainSetting.setting == 'auto_ptr').first()
domain_auto_ptr = strtobool( domain_auto_ptr = strtobool(
domain_auto_ptr.value) if domain_auto_ptr else False domain_auto_ptr.value) if domain_auto_ptr else False
system_auto_ptr = Setting().get('auto_ptr') system_auto_ptr = Setting().get('auto_ptr')
self.name = domain_name self.name = domain_name
domain_id = self.get_id_by_name(domain_reverse_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 system_auto_ptr or
domain_auto_ptr domain_auto_ptr
): ):
result = self.add(domain_reverse_name, 'Master', 'DEFAULT', '', '') result = self.add(domain_reverse_name, 'Master', 'DEFAULT', [], [])
self.update() self.update()
if result['status'] == 'ok': if result['status'] == 'ok':
history = History(msg='Add reverse lookup domain {0}'.format( history = History(msg='Add reverse lookup domain {0}'.format(
domain_reverse_name), domain_reverse_name),
detail=str({ detail=str({
'domain_type': 'Master', 'domain_type': 'Master',
'domain_master_ips': '' 'domain_master_ips': ''
}), }),
created_by='System') created_by='System')
history.add() history.add()
else: else:
return { return {
@ -443,9 +438,9 @@ class Domain(db.Model):
self.grant_privileges(domain_user_ids) self.grant_privileges(domain_user_ids)
return { return {
'status': 'status':
'ok', 'ok',
'msg': 'msg':
'New reverse lookup domain created with granted privileges' 'New reverse lookup domain created with granted privileges'
} }
return { return {
'status': 'ok', 'status': 'ok',
@ -497,16 +492,15 @@ class Domain(db.Model):
""" """
Delete a single domain name from powerdns 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( utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain_name)), '/servers/localhost/zones/{0}'.format(domain_name)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='DELETE', method='DELETE',
verify=Setting().get('verify_ssl_connections')) verify=Setting().get('verify_ssl_connections'))
current_app.logger.info( current_app.logger.info(
'Deleted domain successfully from PowerDNS: {0}'.format( 'Deleted domain successfully from PowerDNS: {0}'.format(
domain_name)) domain_name))
@ -540,8 +534,8 @@ class Domain(db.Model):
user_ids = [] user_ids = []
query = db.session.query( query = db.session.query(
DomainUser, Domain).filter(User.id == DomainUser.user_id).filter( DomainUser, Domain).filter(User.id == DomainUser.user_id).filter(
Domain.id == DomainUser.domain_id).filter( Domain.id == DomainUser.domain_id).filter(
Domain.name == self.name).all() Domain.name == self.name).all()
for q in query: for q in query:
user_ids.append(q[0].user_id) user_ids.append(q[0].user_id)
return user_ids return user_ids
@ -566,7 +560,7 @@ class Domain(db.Model):
db.session.rollback() db.session.rollback()
current_app.logger.error( current_app.logger.error(
'Cannot revoke user privileges on domain {0}. DETAIL: {1}'. 'Cannot revoke user privileges on domain {0}. DETAIL: {1}'.
format(self.name, e)) format(self.name, e))
current_app.logger.debug(print(traceback.format_exc())) current_app.logger.debug(print(traceback.format_exc()))
try: try:
@ -578,7 +572,7 @@ class Domain(db.Model):
db.session.rollback() db.session.rollback()
current_app.logger.error( current_app.logger.error(
'Cannot grant user privileges to domain {0}. DETAIL: {1}'. 'Cannot grant user privileges to domain {0}. DETAIL: {1}'.
format(self.name, e)) format(self.name, e))
current_app.logger.debug(print(traceback.format_exc())) current_app.logger.debug(print(traceback.format_exc()))
def update_from_master(self, domain_name): def update_from_master(self, domain_name):
@ -587,27 +581,26 @@ class Domain(db.Model):
""" """
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if domain: if domain:
headers = {} headers = {'X-API-Key': self.PDNS_API_KEY}
headers['X-API-Key'] = self.PDNS_API_KEY
try: try:
r = utils.fetch_json(urljoin( r = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}/axfr-retrieve'.format( '/servers/localhost/zones/{0}/axfr-retrieve'.format(
domain.name)), domain.name)),
headers=headers, headers=headers,
timeout=int( timeout=int(
Setting().get('pdns_api_timeout')), Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
verify=Setting().get('verify_ssl_connections')) verify=Setting().get('verify_ssl_connections'))
return {'status': 'ok', 'msg': r.get('result')} return {'status': 'ok', 'msg': r.get('result')}
except Exception as e: except Exception as e:
current_app.logger.error( current_app.logger.error(
'Cannot update from master. DETAIL: {0}'.format(e)) 'Cannot update from master. DETAIL: {0}'.format(e))
return { return {
'status': 'status':
'error', 'error',
'msg': 'msg':
'There was something wrong, please contact administrator' 'There was something wrong, please contact administrator'
} }
else: else:
return {'status': 'error', 'msg': 'This domain does not exist'} return {'status': 'error', 'msg': 'This domain does not exist'}
@ -618,14 +611,13 @@ class Domain(db.Model):
""" """
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if domain: if domain:
headers = {} headers = {'X-API-Key': self.PDNS_API_KEY}
headers['X-API-Key'] = self.PDNS_API_KEY
try: try:
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}/cryptokeys'.format( '/servers/localhost/zones/{0}/cryptokeys'.format(
domain.name)), domain.name)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='GET', method='GET',
@ -642,9 +634,9 @@ class Domain(db.Model):
'Cannot get domain dnssec. DETAIL: {0}'.format(e)) 'Cannot get domain dnssec. DETAIL: {0}'.format(e))
return { return {
'status': 'status':
'error', 'error',
'msg': 'msg':
'There was something wrong, please contact administrator' 'There was something wrong, please contact administrator'
} }
else: else:
return {'status': 'error', 'msg': 'This domain does not exist'} return {'status': 'error', 'msg': 'This domain does not exist'}
@ -655,15 +647,14 @@ class Domain(db.Model):
""" """
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if domain: if domain:
headers = {} headers = {'X-API-Key': self.PDNS_API_KEY}
headers['X-API-Key'] = self.PDNS_API_KEY
try: try:
# Enable API-RECTIFY for domain, BEFORE activating DNSSEC # Enable API-RECTIFY for domain, BEFORE activating DNSSEC
post_data = {"api_rectify": True} post_data = {"api_rectify": True}
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain.name)), '/servers/localhost/zones/{0}'.format(domain.name)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
@ -673,7 +664,7 @@ class Domain(db.Model):
return { return {
'status': 'error', 'status': 'error',
'msg': 'msg':
'API-RECTIFY could not be enabled for this domain', 'API-RECTIFY could not be enabled for this domain',
'jdata': jdata 'jdata': jdata
} }
@ -682,8 +673,8 @@ class Domain(db.Model):
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}/cryptokeys'.format( '/servers/localhost/zones/{0}/cryptokeys'.format(
domain.name)), domain.name)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='POST', method='POST',
@ -692,12 +683,12 @@ class Domain(db.Model):
if 'error' in jdata: if 'error' in jdata:
return { return {
'status': 'status':
'error', 'error',
'msg': 'msg':
'Cannot enable DNSSEC for this domain. Error: {0}'. 'Cannot enable DNSSEC for this domain. Error: {0}'.
format(jdata['error']), format(jdata['error']),
'jdata': 'jdata':
jdata jdata
} }
return {'status': 'ok'} return {'status': 'ok'}
@ -708,9 +699,9 @@ class Domain(db.Model):
current_app.logger.debug(traceback.format_exc()) current_app.logger.debug(traceback.format_exc())
return { return {
'status': 'status':
'error', 'error',
'msg': 'msg':
'There was something wrong, please contact administrator' 'There was something wrong, please contact administrator'
} }
else: else:
@ -722,15 +713,14 @@ class Domain(db.Model):
""" """
domain = Domain.query.filter(Domain.name == domain_name).first() domain = Domain.query.filter(Domain.name == domain_name).first()
if domain: if domain:
headers = {} headers = {'X-API-Key': self.PDNS_API_KEY}
headers['X-API-Key'] = self.PDNS_API_KEY
try: try:
# Deactivate DNSSEC # Deactivate DNSSEC
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}/cryptokeys/{1}'.format( '/servers/localhost/zones/{0}/cryptokeys/{1}'.format(
domain.name, key_id)), domain.name, key_id)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='DELETE', method='DELETE',
@ -738,12 +728,12 @@ class Domain(db.Model):
if jdata != True: if jdata != True:
return { return {
'status': 'status':
'error', 'error',
'msg': 'msg':
'Cannot disable DNSSEC for this domain. Error: {0}'. 'Cannot disable DNSSEC for this domain. Error: {0}'.
format(jdata['error']), format(jdata['error']),
'jdata': 'jdata':
jdata jdata
} }
# Disable API-RECTIFY for domain, AFTER deactivating DNSSEC # Disable API-RECTIFY for domain, AFTER deactivating DNSSEC
@ -751,7 +741,7 @@ class Domain(db.Model):
jdata = utils.fetch_json( jdata = utils.fetch_json(
urljoin( urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain.name)), '/servers/localhost/zones/{0}'.format(domain.name)),
headers=headers, headers=headers,
timeout=int(Setting().get('pdns_api_timeout')), timeout=int(Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
@ -761,7 +751,7 @@ class Domain(db.Model):
return { return {
'status': 'error', 'status': 'error',
'msg': 'msg':
'API-RECTIFY could not be disabled for this domain', 'API-RECTIFY could not be disabled for this domain',
'jdata': jdata 'jdata': jdata
} }
@ -774,7 +764,7 @@ class Domain(db.Model):
return { return {
'status': 'error', 'status': 'error',
'msg': 'msg':
'There was something wrong, please contact administrator', 'There was something wrong, please contact administrator',
'domain': domain.name, 'domain': domain.name,
'id': key_id 'id': key_id
} }
@ -797,8 +787,7 @@ class Domain(db.Model):
if not domain: if not domain:
return {'status': False, 'msg': 'Domain does not exist'} 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) account_name = Account().get_name_by_id(account_id)
@ -807,13 +796,13 @@ class Domain(db.Model):
try: try:
jdata = utils.fetch_json(urljoin( jdata = utils.fetch_json(urljoin(
self.PDNS_STATS_URL, self.API_EXTENDED_URL + self.PDNS_STATS_URL, self.API_EXTENDED_URL +
'/servers/localhost/zones/{0}'.format(domain_name)), '/servers/localhost/zones/{0}'.format(domain_name)),
headers=headers, headers=headers,
timeout=int( timeout=int(
Setting().get('pdns_api_timeout')), Setting().get('pdns_api_timeout')),
method='PUT', method='PUT',
verify=Setting().get('verify_ssl_connections'), verify=Setting().get('verify_ssl_connections'),
data=post_data) data=post_data)
if 'error' in jdata.keys(): if 'error' in jdata.keys():
current_app.logger.error(jdata['error']) current_app.logger.error(jdata['error'])
@ -852,7 +841,7 @@ class Domain(db.Model):
.outerjoin(Account, Domain.account_id == Account.id) \ .outerjoin(Account, Domain.account_id == Account.id) \
.outerjoin(AccountUser, Account.id == AccountUser.account_id) \ .outerjoin(AccountUser, Account.id == AccountUser.account_id) \
.filter( .filter(
db.or_( db.or_(
DomainUser.user_id == user_id, DomainUser.user_id == user_id,
AccountUser.user_id == user_id AccountUser.user_id == user_id
)).filter(Domain.id == self.id).first() )).filter(Domain.id == self.id).first()

4
powerdnsadmin/models/domain_setting.py
View File

@ -1,3 +1,5 @@
import traceback
from flask import current_app from flask import current_app
from .base import db from .base import db
@ -32,4 +34,4 @@ class DomainSetting(db.Model):
'Unable to set DomainSetting value. DETAIL: {0}'.format(e)) 'Unable to set DomainSetting value. DETAIL: {0}'.format(e))
current_app.logger.debug(traceback.format_exc()) current_app.logger.debug(traceback.format_exc())
db.session.rollback() db.session.rollback()
return False return False

2
powerdnsadmin/models/domain_user.py
View File

@ -14,4 +14,4 @@ class DomainUser(db.Model):
self.user_id = user_id self.user_id = user_id
def __repr__(self): def __repr__(self):
return '<Domain_User {0} {1}>'.format(self.domain_id, self.user_id) return '<Domain_User {0} {1}>'.format(self.domain_id, self.user_id)

2
powerdnsadmin/models/history.py
View File

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

2
powerdnsadmin/models/record_entry.py
View File

@ -24,4 +24,4 @@ class RecordEntry(object):
return self._is_allowed_edit return self._is_allowed_edit
def is_allowed_delete(self): def is_allowed_delete(self):
return self._is_allowed_delete return self._is_allowed_delete

2
powerdnsadmin/models/role.py
View File

@ -20,4 +20,4 @@ class Role(db.Model):
self.description = description self.description = description
def __repr__(self): def __repr__(self):
return '<Role {0}r>'.format(self.name) return '<Role {0}r>'.format(self.name)

9
powerdnsadmin/models/server.py
View File

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

2
powerdnsadmin/models/setting.py
View File

@ -1,4 +1,6 @@
import sys import sys
import traceback
import pytimeparse import pytimeparse
from ast import literal_eval from ast import literal_eval
from distutils.util import strtobool 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 '' 'password'] if 'password' in request.form else ''
else: else:
firstname = lastname = email = new_password = '' firstname = lastname = email = new_password = ''
logging.warning( current_app.logger.warning(
'Authenticated externally. User {0} information will not allowed to update the profile' 'Authenticated externally. User {0} information will not allowed to update the profile'
.format(current_user.username)) .format(current_user.username))