cwinfo/powerdns-admin
5
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2025-04-21 02:57:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Completed the implementation of the SERVER_EXTERNAL_SSL environment setting into the app config files.

Browse Source 
Completed the implementation of the aforementioned environment setting into the OAuth workflows.

Documented the aforementioned setting in the Environment-variables.md wiki document.
This commit is contained in:
Matt Scott 2023-04-08 17:05:27 -04:00
parent cacfc042e2
commit ab4495dc46
No known key found for this signature in database
GPG Key ID: A9A0AFFC0E079001
8 changed files with 270 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
configs/docker_config.py
View File

@ -1,3 +1,8 @@
# import everything from environment variables
import os
import sys
import json
# Defaults for Docker image
BIND_ADDRESS = '0.0.0.0'
PORT = 80
@ -23,6 +28,7 @@ legal_envvars = (
'OIDC_OAUTH_EMAIL',
'BIND_ADDRESS',
'PORT',
'SERVER_EXTERNAL_SSL',
'LOG_LEVEL',
'SALT',
'SQLALCHEMY_TRACK_MODIFICATIONS',
@ -97,21 +103,18 @@ legal_envvars_bool = (
'SESSION_COOKIE_SECURE',
'CSRF_COOKIE_SECURE',
'CAPTCHA_ENABLE',
'SERVER_EXTERNAL_SSL',
)
legal_envvars_dict = (
'SQLALCHEMY_ENGINE_OPTIONS',
)
# import everything from environment variables
import os
import sys
import json
def str2bool(v):
return v.lower() in ("true", "yes", "1")
def dictfromstr(v,ret):
def dictfromstr(v, ret):
try:
return json.loads(ret)
except Exception as e:
@ -119,10 +122,11 @@ def dictfromstr(v,ret):
print(e)
raise ValueError
for v in legal_envvars:
ret = None
# _FILE suffix will allow to read value from file, usefull for Docker's
# _FILE suffix will allow to read value from file, useful for Docker containers.
# secrets feature
if v + '_FILE' in os.environ:
if v in os.environ:

5
docs/wiki/configuration/Environment-variables.md
View File

@ -1,10 +1,10 @@
# Supported environment variables
| Variable | Description | Required | Default value |
| ---------| ----------- | -------- | ------------- |
|--------------------------------|--------------------------------------------------------------------------|------------|---------------|
| BIND_ADDRESS |
| CSRF_COOKIE_SECURE |
| SESSION_TYPE | null|filesystem|sqlalchemy | | filesystem |
| SESSION_TYPE | null | filesystem | sqlalchemy | | filesystem |
| LDAP_ENABLED |
| LOCAL_DB_ENABLED |
| LOG_LEVEL |
@ -22,6 +22,7 @@
| OIDC_OAUTH_TOKEN_URL | | | |
| OIDC_OAUTH_METADATA_URL | | | |
| PORT |
| SERVER_EXTERNAL_SSL | Forceful override of URL schema detection when using the url_for method. | False | None |
| REMOTE_USER_COOKIES |
| REMOTE_USER_LOGOUT_URL |
| SALT |

1
powerdnsadmin/default_config.py
View File

@ -8,6 +8,7 @@ SECRET_KEY = 'e951e5a1f4b94151b360f47edf596dd2'
BIND_ADDRESS = '0.0.0.0'
PORT = 9191
HSTS_ENABLED = False
SERVER_EXTERNAL_SSL = None
SESSION_TYPE = 'sqlalchemy'
SESSION_COOKIE_SAMESITE = 'Lax'

107
powerdnsadmin/routes/index.py
View File

@ -45,6 +45,7 @@ index_bp = Blueprint('index',
template_folder='templates',
url_prefix='/')
@index_bp.before_app_first_request
def register_modules():
global google
@ -98,7 +99,11 @@ def google_login():
)
abort(400)
else:
redirect_uri = url_for('google_authorized', _external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
redirect_uri = url_for('google_authorized', **params)
return google.authorize_redirect(redirect_uri)
@ -110,7 +115,11 @@ def github_login():
)
abort(400)
else:
redirect_uri = url_for('github_authorized', _external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
redirect_uri = url_for('github_authorized', **params)
return github.authorize_redirect(redirect_uri)
@ -122,9 +131,11 @@ def azure_login():
)
abort(400)
else:
redirect_uri = url_for('azure_authorized',
_external=True,
_scheme='https')
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
redirect_uri = url_for('azure_authorized', **params)
return azure.authorize_redirect(redirect_uri)
@ -136,7 +147,11 @@ def oidc_login():
)
abort(400)
else:
redirect_uri = url_for('oidc_authorized', _external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
redirect_uri = url_for('oidc_authorized', **params)
return oidc.authorize_redirect(redirect_uri)
@ -198,7 +213,7 @@ def login():
if 'azure_token' in session:
azure_info = azure.get('me?$select=displayName,givenName,id,mail,surname,userPrincipalName').text
current_app.logger.info('Azure login returned: '+azure_info)
current_app.logger.info('Azure login returned: ' + azure_info)
me = json.loads(azure_info)
azure_info = azure.post('me/getMemberGroups',
@ -394,20 +409,21 @@ def login():
session.pop('oidc_token', None)
return redirect(url_for('index.login'))
#This checks if the account_name_property and account_description property were included in settings.
if Setting().get('oidc_oauth_account_name_property') and Setting().get('oidc_oauth_account_description_property'):
# This checks if the account_name_property and account_description property were included in settings.
if Setting().get('oidc_oauth_account_name_property') and Setting().get(
'oidc_oauth_account_description_property'):
#Gets the name_property and description_property.
# Gets the name_property and description_property.
name_prop = Setting().get('oidc_oauth_account_name_property')
desc_prop = Setting().get('oidc_oauth_account_description_property')
account_to_add = []
#If the name_property and desc_property exist in me (A variable that contains all the userinfo from the IdP).
# If the name_property and desc_property exist in me (A variable that contains all the userinfo from the IdP).
if name_prop in me and desc_prop in me:
accounts_name_prop = [me[name_prop]] if type(me[name_prop]) is not list else me[name_prop]
accounts_desc_prop = [me[desc_prop]] if type(me[desc_prop]) is not list else me[desc_prop]
#Run on all groups the user is in by the index num.
# Run on all groups the user is in by the index num.
for i in range(len(accounts_name_prop)):
description = ''
if i < len(accounts_desc_prop):
@ -490,34 +506,36 @@ def login():
saml_enabled=SAML_ENABLED,
error='Token required')
if Setting().get('autoprovisioning') and auth_method!='LOCAL':
urn_value=Setting().get('urn_value')
Entitlements=user.read_entitlements(Setting().get('autoprovisioning_attribute'))
if len(Entitlements)==0 and Setting().get('purge'):
if Setting().get('autoprovisioning') and auth_method != 'LOCAL':
urn_value = Setting().get('urn_value')
Entitlements = user.read_entitlements(Setting().get('autoprovisioning_attribute'))
if len(Entitlements) == 0 and Setting().get('purge'):
user.set_role("User")
user.revoke_privilege(True)
elif len(Entitlements)!=0:
elif len(Entitlements) != 0:
if checkForPDAEntries(Entitlements, urn_value):
user.updateUser(Entitlements)
else:
current_app.logger.warning('Not a single powerdns-admin record was found, possibly a typo in the prefix')
current_app.logger.warning(
'Not a single powerdns-admin record was found, possibly a typo in the prefix')
if Setting().get('purge'):
user.set_role("User")
user.revoke_privilege(True)
current_app.logger.warning('Procceding to revoke every privilige from ' + user.username + '.' )
current_app.logger.warning('Procceding to revoke every privilige from ' + user.username + '.')
return authenticate_user(user, auth_method, remember_me)
def checkForPDAEntries(Entitlements, urn_value):
"""
Run through every record located in the ldap attribute given and determine if there are any valid powerdns-admin records
"""
urnArguments=[x.lower() for x in urn_value.split(':')]
urnArguments = [x.lower() for x in urn_value.split(':')]
for Entitlement in Entitlements:
entArguments=Entitlement.split(':powerdns-admin')
entArguments=[x.lower() for x in entArguments[0].split(':')]
if (entArguments==urnArguments):
entArguments = Entitlement.split(':powerdns-admin')
entArguments = [x.lower() for x in entArguments[0].split(':')]
if (entArguments == urnArguments):
return True
return False
@ -553,7 +571,7 @@ def signin_history(username, authenticator, success):
# Write history
History(msg='User {} authentication {}'.format(username, str_success),
detail = json.dumps({
detail=json.dumps({
'username': username,
'authenticator': authenticator,
'ip_address': request_ip,
@ -561,6 +579,7 @@ def signin_history(username, authenticator, success):
}),
created_by='System').add()
# Get a list of Azure security groups the user is a member of
def get_azure_groups(uri):
azure_info = azure.get(uri).text
@ -576,30 +595,33 @@ def get_azure_groups(uri):
mygroups = []
return mygroups
# Handle user login, write history and, if set, handle showing the register_otp QR code.
# if Setting for OTP on first login is enabled, and OTP field is also enabled,
# but user isn't using it yet, enable OTP, get QR code and display it, logging the user out.
def authenticate_user(user, authenticator, remember=False):
login_user(user, remember=remember)
signin_history(user.username, authenticator, True)
if Setting().get('otp_force') and Setting().get('otp_field_enabled') and not user.otp_secret and session['authentication_type'] not in ['OAuth']:
if Setting().get('otp_force') and Setting().get('otp_field_enabled') and not user.otp_secret \
and session['authentication_type'] not in ['OAuth']:
user.update_profile(enable_otp=True)
user_id = current_user.id
prepare_welcome_user(user_id)
return redirect(url_for('index.welcome'))
return redirect(url_for('index.login'))
# Prepare user to enter /welcome screen, otherwise they won't have permission to do so
def prepare_welcome_user(user_id):
logout_user()
session['welcome_user_id'] = user_id
@index_bp.route('/logout')
def logout():
if current_app.config.get(
'SAML_ENABLED'
) and 'samlSessionIndex' in session and current_app.config.get(
'SAML_LOGOUT'):
) and 'samlSessionIndex' in session and current_app.config.get('SAML_LOGOUT'):
req = saml.prepare_flask_request(request)
auth = saml.init_saml_auth(req)
if current_app.config.get('SAML_LOGOUT_URL'):
@ -651,13 +673,12 @@ def logout():
def password_policy_check(user, password):
def check_policy(chars, user_password, setting):
lenreq = int(Setting().get(setting))
setting_as_int = int(Setting().get(setting))
test_string = user_password
for c in chars:
test_string = test_string.replace(c, '')
return (lenreq, len(user_password) - len(test_string))
return (setting_as_int, len(user_password) - len(test_string))
def matches_policy(item, policy_fails):
return "*" if item in policy_fails else ""
@ -704,12 +725,14 @@ def password_policy_check(user, password):
(pwd_min_lowercase_setting, pwd_lowercase) = check_policy(string.digits, password, 'pwd_min_lowercase')
if pwd_lowercase < pwd_min_lowercase_setting:
policy_fails["lowercase"] = True
policy.append(f"{matches_policy('lowercase', policy_fails)}lowercase={pwd_lowercase}/{pwd_min_lowercase_setting}")
policy.append(
f"{matches_policy('lowercase', policy_fails)}lowercase={pwd_lowercase}/{pwd_min_lowercase_setting}")
# Uppercase
(pwd_min_uppercase_setting, pwd_uppercase) = check_policy(string.digits, password, 'pwd_min_uppercase')
if pwd_uppercase < pwd_min_uppercase_setting:
policy_fails["uppercase"] = True
policy.append(f"{matches_policy('uppercase', policy_fails)}uppercase={pwd_uppercase}/{pwd_min_uppercase_setting}")
policy.append(
f"{matches_policy('uppercase', policy_fails)}uppercase={pwd_uppercase}/{pwd_min_uppercase_setting}")
# Special
(pwd_min_special_setting, pwd_special) = check_policy(string.digits, password, 'pwd_min_special')
if pwd_special < pwd_min_special_setting:
@ -728,7 +751,8 @@ def password_policy_check(user, password):
pwd_complexity = result['guesses_log10']
if pwd_complexity < pwd_min_complexity_setting:
policy_fails["complexity"] = True
policy.append(f"{matches_policy('complexity', policy_fails)}complexity={pwd_complexity:.0f}/{pwd_min_complexity_setting}")
policy.append(
f"{matches_policy('complexity', policy_fails)}complexity={pwd_complexity:.0f}/{pwd_min_complexity_setting}")
policy_str = {"password": f"Fails policy: {', '.join(policy)}. Items prefixed with '*' failed."}
@ -775,7 +799,8 @@ def register():
if not captcha.validate():
return render_template(
'register.html', error='Invalid CAPTCHA answer', error_messages=error_messages, captcha_enable=CAPTCHA_ENABLE)
'register.html', error='Invalid CAPTCHA answer', error_messages=error_messages,
captcha_enable=CAPTCHA_ENABLE)
if error_messages:
return render_template('register.html', error_messages=error_messages, captcha_enable=CAPTCHA_ENABLE)
@ -827,12 +852,15 @@ def welcome():
if otp_token and otp_token.isdigit():
good_token = user.verify_totp(otp_token)
if not good_token:
return render_template('register_otp.html', qrcode_image=encoded_img_data.decode(), user=user, error="Invalid token")
return render_template('register_otp.html', qrcode_image=encoded_img_data.decode(), user=user,
error="Invalid token")
else:
return render_template('register_otp.html', qrcode_image=encoded_img_data.decode(), user=user, error="Token required")
return render_template('register_otp.html', qrcode_image=encoded_img_data.decode(), user=user,
error="Token required")
session.pop('welcome_user_id')
return redirect(url_for('index.index'))
@index_bp.route('/confirm/<token>', methods=['GET'])
def confirm_email(token):
email = confirm_token(token)
@ -982,7 +1010,7 @@ def dyndns_update():
if result['status'] == 'ok':
history = History(
msg='DynDNS update: updated {} successfully'.format(hostname),
detail = json.dumps({
detail=json.dumps({
'domain': domain.name,
'record': hostname,
'type': rtype,
@ -1024,7 +1052,7 @@ def dyndns_update():
msg=
'DynDNS update: created record {0} in zone {1} successfully'
.format(hostname, domain.name, str(ip)),
detail = json.dumps({
detail=json.dumps({
'domain': domain.name,
'record': hostname,
'value': str(ip)
@ -1088,7 +1116,7 @@ def saml_authorized():
req = saml.prepare_flask_request(request)
auth = saml.init_saml_auth(req)
auth.process_response()
current_app.logger.debug( auth.get_attributes() )
current_app.logger.debug(auth.get_attributes())
errors = auth.get_errors()
if len(errors) == 0:
session['samlUserdata'] = auth.get_attributes()
@ -1237,6 +1265,7 @@ def uplift_to_admin(user):
created_by='SAML Assertion')
history.add()
def uplift_to_operator(user):
if user.role.name != 'Operator':
user.role_id = Role.query.filter_by(name='Operator').first().id

10
powerdnsadmin/services/azure.py
View File

@ -38,14 +38,16 @@ def azure_oauth():
@current_app.route('/azure/authorized')
def azure_authorized():
session['azure_oauthredir'] = url_for('.azure_authorized',
_external=True,
_scheme='https')
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
session['azure_oauthredir'] = url_for('.azure_authorized', **params)
token = azure.authorize_access_token()
if token is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error'], request.args['error_description'])
session['azure_token'] = (token)
return redirect(url_for('index.login', _external=True, _scheme='https'))
return redirect(url_for('index.login', **params))
return azure

11
powerdnsadmin/services/github.py
View File

@ -40,13 +40,16 @@ def github_oauth():
@current_app.route('/github/authorized')
def github_authorized():
session['github_oauthredir'] = url_for('.github_authorized',
_external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
session['github_oauthredir'] = url_for('.github_authorized', **params)
token = github.authorize_access_token()
if token is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error'], request.args['error_description'])
session['github_token'] = (token)
return redirect(url_for('index.login'))
session['github_token'] = token
return redirect(url_for('index.login', **params))
return github

12
powerdnsadmin/services/google.py
View File

@ -39,16 +39,18 @@ def google_oauth():
@current_app.route('/google/authorized')
def google_authorized():
session['google_oauthredir'] = url_for(
'.google_authorized', _external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
session['google_oauthredir'] = url_for('.google_authorized', **params)
token = google.authorize_access_token()
if token is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error_reason'],
request.args['error_description']
)
session['google_token'] = (token)
return redirect(url_for('index.login'))
session['google_token'] = token
return redirect(url_for('index.login', **params))
return google

11
powerdnsadmin/services/oidc.py
View File

@ -39,13 +39,16 @@ def oidc_oauth():
@current_app.route('/oidc/authorized')
def oidc_authorized():
session['oidc_oauthredir'] = url_for('.oidc_authorized',
_external=True)
use_ssl = current_app.config.get('SERVER_EXTERNAL_SSL')
params = {'_external': True}
if isinstance(use_ssl, bool):
params['_scheme'] = 'https' if use_ssl else 'http'
session['oidc_oauthredir'] = url_for('.oidc_authorized', **params)
token = oidc.authorize_access_token()
if token is None:
return 'Access denied: reason=%s error=%s' % (
request.args['error'], request.args['error_description'])
session['oidc_token'] = (token)
return redirect(url_for('index.login'))
session['oidc_token'] = token
return redirect(url_for('index.login', **params))
return oidc