import os import re import json import traceback import datetime import ipaddress import base64 from distutils.util import strtobool from yaml import Loader, load from flask import Blueprint, render_template, make_response, url_for, current_app, g, session, request, redirect, abort from flask_login import login_user, logout_user, login_required, current_user from .base import captcha, csrf, login_manager from ..lib import utils from ..decorators import dyndns_login_required from ..models.base import db from ..models.user import User, Anonymous from ..models.role import Role from ..models.account import Account from ..models.account_user import AccountUser from ..models.domain import Domain from ..models.domain_user import DomainUser from ..models.domain_setting import DomainSetting from ..models.record import Record from ..models.setting import Setting from ..models.history import History from ..services.google import google_oauth from ..services.github import github_oauth from ..services.azure import azure_oauth from ..services.oidc import oidc_oauth from ..services.saml import SAML from ..services.token import confirm_token from ..services.email import send_account_verification google = None github = None azure = None oidc = None saml = None index_bp = Blueprint('index', __name__, template_folder='templates', url_prefix='/') @index_bp.before_app_first_request def register_modules(): global google global github global azure global oidc global saml google = google_oauth() github = github_oauth() azure = azure_oauth() oidc = oidc_oauth() saml = SAML() @index_bp.before_request def before_request(): # Check if user is anonymous g.user = current_user login_manager.anonymous_user = Anonymous # Check site is in maintenance mode maintenance = Setting().get('maintenance') if maintenance and current_user.is_authenticated and current_user.role.name not in [ 'Administrator', 'Operator' ]: return render_template('maintenance.html') # Manage session timeout session.permanent = True current_app.permanent_session_lifetime = datetime.timedelta( minutes=int(Setting().get('session_timeout'))) session.modified = True @index_bp.route('/', methods=['GET']) @login_required def index(): return redirect(url_for('dashboard.dashboard')) @index_bp.route('/ping', methods=['GET']) def ping(): return make_response('ok') @index_bp.route('/google/login') def google_login(): if not Setting().get('google_oauth_enabled') or google is None: current_app.logger.error( 'Google OAuth is disabled or you have not yet reloaded the pda application after enabling.' ) abort(400) else: redirect_uri = url_for('google_authorized', _external=True) return google.authorize_redirect(redirect_uri) @index_bp.route('/github/login') def github_login(): if not Setting().get('github_oauth_enabled') or github is None: current_app.logger.error( 'Github OAuth is disabled or you have not yet reloaded the pda application after enabling.' ) abort(400) else: redirect_uri = url_for('github_authorized', _external=True) return github.authorize_redirect(redirect_uri) @index_bp.route('/azure/login') def azure_login(): if not Setting().get('azure_oauth_enabled') or azure is None: current_app.logger.error( 'Microsoft OAuth is disabled or you have not yet reloaded the pda application after enabling.' ) abort(400) else: redirect_uri = url_for('azure_authorized', _external=True, _scheme='https') return azure.authorize_redirect(redirect_uri) @index_bp.route('/oidc/login') def oidc_login(): if not Setting().get('oidc_oauth_enabled') or oidc is None: current_app.logger.error( 'OIDC OAuth is disabled or you have not yet reloaded the pda application after enabling.' ) abort(400) else: redirect_uri = url_for('oidc_authorized', _external=True) return oidc.authorize_redirect(redirect_uri) @index_bp.route('/login', methods=['GET', 'POST']) def login(): SAML_ENABLED = current_app.config.get('SAML_ENABLED', False) if g.user is not None and current_user.is_authenticated: return redirect(url_for('dashboard.dashboard')) if 'google_token' in session: user_data = json.loads(google.get('userinfo').text) first_name = user_data['given_name'] surname = user_data['family_name'] email = user_data['email'] user = User.query.filter_by(username=email).first() if user is None: user = User.query.filter_by(email=email).first() if not user: user = User(username=email, firstname=first_name, lastname=surname, plain_text_password=None, email=email) result = user.create_local_user() if not result['status']: session.pop('google_token', None) return redirect(url_for('index.login')) session['user_id'] = user.id session['authentication_type'] = 'OAuth' return authenticate_user(user, 'Google OAuth') if 'github_token' in session: me = json.loads(github.get('user').text) github_username = me['login'] github_name = me['name'] github_email = me['email'] user = User.query.filter_by(username=github_username).first() if user is None: user = User.query.filter_by(email=github_email).first() if not user: user = User(username=github_username, plain_text_password=None, firstname=github_name, lastname='', email=github_email) result = user.create_local_user() if not result['status']: session.pop('github_token', None) return redirect(url_for('index.login')) session['user_id'] = user.id session['authentication_type'] = 'OAuth' return authenticate_user(user, 'Github OAuth') 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) me = json.loads(azure_info) azure_info = azure.post('me/getMemberGroups', json={'securityEnabledOnly': False}).text current_app.logger.info('Azure groups returned: ' + azure_info) grouplookup = json.loads(azure_info) # Groups are in mygroups['value'] which is an array if "value" in grouplookup: mygroups = grouplookup["value"] else: mygroups = [] azure_username = me["userPrincipalName"] azure_givenname = me["givenName"] azure_familyname = me["surname"] if "mail" in me: azure_email = me["mail"] else: azure_email = "" if not azure_email: azure_email = me["userPrincipalName"] # Handle foreign principals such as guest users azure_email = re.sub(r"#.*$", "", azure_email) azure_username = re.sub(r"#.*$", "", azure_username) user = User.query.filter_by(username=azure_username).first() if not user: user = User(username=azure_username, plain_text_password=None, firstname=azure_givenname, lastname=azure_familyname, email=azure_email) result = user.create_local_user() if not result['status']: current_app.logger.warning('Unable to create ' + azure_username) session.pop('azure_token', None) # note: a redirect to login results in an endless loop, so render the login page instead return render_template('login.html', saml_enabled=SAML_ENABLED, error=('User ' + azure_username + ' cannot be created.')) session['user_id'] = user.id session['authentication_type'] = 'OAuth' # Handle group memberships, if defined if Setting().get('azure_sg_enabled'): if Setting().get('azure_admin_group') in mygroups: current_app.logger.info('Setting role for user ' + azure_username + ' to Administrator due to group membership') user.set_role("Administrator") else: if Setting().get('azure_operator_group') in mygroups: current_app.logger.info('Setting role for user ' + azure_username + ' to Operator due to group membership') user.set_role("Operator") else: if Setting().get('azure_user_group') in mygroups: current_app.logger.info('Setting role for user ' + azure_username + ' to User due to group membership') user.set_role("User") else: current_app.logger.warning('User ' + azure_username + ' has no relevant group memberships') session.pop('azure_token', None) return render_template('login.html', saml_enabled=SAML_ENABLED, error=('User ' + azure_username + ' is not in any authorised groups.')) # Handle account/group creation, if enabled if Setting().get('azure_group_accounts_enabled') and mygroups: current_app.logger.info('Azure group account sync enabled') name_value = Setting().get('azure_group_accounts_name') description_value = Setting().get('azure_group_accounts_description') select_values = name_value if description_value != '': select_values += ',' + description_value mygroups = get_azure_groups( 'me/memberOf/microsoft.graph.group?$count=false&$securityEnabled=true&$select={}'.format(select_values)) description_pattern = Setting().get('azure_group_accounts_description_re') pattern = Setting().get('azure_group_accounts_name_re') # Loop through users security groups for azure_group in mygroups: if name_value in azure_group: group_name = azure_group[name_value] group_description = '' if description_value in azure_group: group_description = azure_group[description_value] # Do regex search if enabled for group description if description_pattern != '': current_app.logger.info('Matching group description {} against regex {}'.format( group_description, description_pattern)) matches = re.match( description_pattern, group_description) if matches: current_app.logger.info( 'Group {} matched regexp'.format(group_description)) group_description = matches.group(1) else: # Regexp didn't match, continue to next iteration continue # Do regex search if enabled for group name if pattern != '': current_app.logger.info( 'Matching group name {} against regex {}'.format(group_name, pattern)) matches = re.match(pattern, group_name) if matches: current_app.logger.info( 'Group {} matched regexp'.format(group_name)) group_name = matches.group(1) else: # Regexp didn't match, continue to next iteration continue account = Account() sanitized_group_name = Account.sanitize_name(group_name) account_id = account.get_id_by_name(account_name=sanitized_group_name) if account_id: account = Account.query.get(account_id) # check if user has permissions account_users = account.get_user() current_app.logger.info('Group: {} Users: {}'.format( group_name, account_users)) if user.id in account_users: current_app.logger.info('User id {} is already in account {}'.format( user.id, group_name)) else: account.add_user(user) history = History(msg='Update account {0}'.format( account.name), created_by='System') history.add() current_app.logger.info('User {} added to Account {}'.format( user.username, account.name)) else: account = Account( name=sanitized_group_name, description=group_description, contact='', mail='' ) account.create_account() history = History(msg='Create account {0}'.format( account.name), created_by='System') history.add() account.add_user(user) history = History(msg='Update account {0}'.format(account.name), created_by='System') history.add() current_app.logger.warning('group info: {} '.format(account_id)) return authenticate_user(user, 'Azure OAuth') if 'oidc_token' in session: me = json.loads(oidc.get('userinfo').text) oidc_username = me[Setting().get('oidc_oauth_username')] oidc_givenname = me[Setting().get('oidc_oauth_firstname')] oidc_familyname = me[Setting().get('oidc_oauth_last_name')] oidc_email = me[Setting().get('oidc_oauth_email')] user = User.query.filter_by(username=oidc_username).first() if not user: user = User(username=oidc_username, plain_text_password=None, firstname=oidc_givenname, lastname=oidc_familyname, email=oidc_email) result = user.create_local_user() else: user.firstname = oidc_givenname user.lastname = oidc_familyname user.email = oidc_email user.plain_text_password = None result = user.update_local_user() if not result['status']: 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'): #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 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. for i in range(len(accounts_name_prop)): description = '' if i < len(accounts_desc_prop): description = accounts_desc_prop[i] account = handle_account(accounts_name_prop[i], description) account_to_add.append(account) user_accounts = user.get_accounts() # Add accounts for account in account_to_add: if account not in user_accounts: account.add_user(user) # Remove accounts if the setting is enabled if Setting().get('delete_sso_accounts'): for account in user_accounts: if account not in account_to_add: account.remove_user(user) session['user_id'] = user.id session['authentication_type'] = 'OAuth' return authenticate_user(user, 'OIDC OAuth') if request.method == 'GET': return render_template('login.html', saml_enabled=SAML_ENABLED) elif request.method == 'POST': # process Local-DB authentication username = request.form['username'] password = request.form['password'] otp_token = request.form.get('otptoken') auth_method = request.form.get('auth_method', 'LOCAL') session[ 'authentication_type'] = 'LDAP' if auth_method != 'LOCAL' else 'LOCAL' remember_me = True if 'remember' in request.form else False if auth_method == 'LOCAL' and not Setting().get('local_db_enabled'): return render_template( 'login.html', saml_enabled=SAML_ENABLED, error='Local authentication is disabled') user = User(username=username, password=password, plain_text_password=password) try: if Setting().get('verify_user_email') and user.email and not user.confirmed: return render_template( 'login.html', saml_enabled=SAML_ENABLED, error='Please confirm your email address first') auth = user.is_validate(method=auth_method, src_ip=request.remote_addr) if auth == False: signin_history(user.username, auth_method, False) return render_template('login.html', saml_enabled=SAML_ENABLED, error='Invalid credentials') except Exception as e: current_app.logger.error( "Cannot authenticate user. Error: {}".format(e)) current_app.logger.debug(traceback.format_exc()) return render_template('login.html', saml_enabled=SAML_ENABLED, error=e) # check if user enabled OPT authentication if user.otp_secret: if otp_token and otp_token.isdigit(): good_token = user.verify_totp(otp_token) if not good_token: signin_history(user.username, auth_method, False) return render_template('login.html', saml_enabled=SAML_ENABLED, error='Invalid credentials') else: return render_template('login.html', 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'): user.set_role("User") user.revoke_privilege(True) 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') if Setting().get('purge'): user.set_role("User") user.revoke_privilege(True) 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(':')] for Entitlement in Entitlements: entArguments=Entitlement.split(':powerdns-admin') entArguments=[x.lower() for x in entArguments[0].split(':')] if (entArguments==urnArguments): return True return False def clear_session(): session.pop('user_id', None) session.pop('github_token', None) session.pop('google_token', None) session.pop('authentication_type', None) session.pop('remote_user', None) session.clear() logout_user() def signin_history(username, authenticator, success): # Get user ip address if request.headers.getlist("X-Forwarded-For"): request_ip = request.headers.getlist("X-Forwarded-For")[0] request_ip = request_ip.split(',')[0] else: request_ip = request.remote_addr # Write log if success: str_success = 'succeeded' current_app.logger.info( "User {} authenticated successfully via {} from {}".format( username, authenticator, request_ip)) else: str_success = 'failed' current_app.logger.warning( "User {} failed to authenticate via {} from {}".format( username, authenticator, request_ip)) # Write history History(msg='User {} authentication {}'.format(username, str_success), detail = json.dumps({ 'username': username, 'authenticator': authenticator, 'ip_address': request_ip, 'success': 1 if success else 0 }), 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 current_app.logger.info('Azure groups returned: ' + azure_info) grouplookup = json.loads(azure_info) if "value" in grouplookup: mygroups = grouplookup["value"] # If "@odata.nextLink" exists in the results, we need to get more groups if "@odata.nextLink" in grouplookup: # The additional groups are added to the existing array mygroups.extend(get_azure_groups(grouplookup["@odata.nextLink"])) else: 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']: 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'): req = saml.prepare_flask_request(request) auth = saml.init_saml_auth(req) if current_app.config.get('SAML_LOGOUT_URL'): return redirect( auth.logout( name_id_format= "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress", return_to=current_app.config.get('SAML_LOGOUT_URL'), session_index=session['samlSessionIndex'], name_id=session['samlNameId'])) return redirect( auth.logout( name_id_format= "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress", session_index=session['samlSessionIndex'], name_id=session['samlNameId'])) redirect_uri = url_for('index.login') oidc_logout = Setting().get('oidc_oauth_logout_url') if 'oidc_token' in session and oidc_logout: redirect_uri = "{}?redirect_uri={}".format( oidc_logout, url_for('index.login', _external=True)) # Clean cookies and flask session clear_session() # If remote user authentication is enabled and a logout URL is configured for it, # redirect users to that instead remote_user_logout_url = current_app.config.get('REMOTE_USER_LOGOUT_URL') if current_app.config.get('REMOTE_USER_ENABLED') and remote_user_logout_url: current_app.logger.debug( 'Redirecting remote user "{0}" to logout URL {1}' .format(current_user.username, remote_user_logout_url)) # Warning: if REMOTE_USER environment variable is still set and not cleared by # some external module, not defining a custom logout URL will trigger a loop # that will just log the user back in right after logging out res = make_response(redirect(remote_user_logout_url.strip())) # Remove any custom cookies the remote authentication mechanism may use # (e.g.: MOD_AUTH_CAS and MOD_AUTH_CAS_S) remote_cookies = current_app.config.get('REMOTE_USER_COOKIES') for r_cookie_name in utils.ensure_list(remote_cookies): res.delete_cookie(r_cookie_name) return res return redirect(redirect_uri) @index_bp.route('/register', methods=['GET', 'POST']) def register(): CAPTCHA_ENABLE = current_app.config.get('CAPTCHA_ENABLE') if Setting().get('signup_enabled'): if current_user.is_authenticated: return redirect(url_for('index.index')) if request.method == 'GET': return render_template('register.html', captcha_enable=CAPTCHA_ENABLE) elif request.method == 'POST': username = request.form.get('username', '').strip() password = request.form.get('password', '') firstname = request.form.get('firstname', '').strip() lastname = request.form.get('lastname', '').strip() email = request.form.get('email', '').strip() rpassword = request.form.get('rpassword', '') is_valid_email = re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$') error_messages = {} if not firstname: error_messages['firstname'] = 'First Name is required' if not lastname: error_messages['lastname'] = 'Last Name is required' if not username: error_messages['username'] = 'Username is required' if not password: error_messages['password'] = 'Password is required' if not rpassword: error_messages['rpassword'] = 'Password confirmation is required' if not email: error_messages['email'] = 'Email is required' if not is_valid_email.match(email): error_messages['email'] = 'Invalid email address' if password != rpassword: error_messages['password'] = 'Password confirmation does not match' error_messages['rpassword'] = 'Password confirmation does not match' if not captcha.validate(): return render_template( '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) user = User(username=username, plain_text_password=password, firstname=firstname, lastname=lastname, email=email ) try: result = user.create_local_user() if result and result['status']: if Setting().get('verify_user_email'): send_account_verification(email) if Setting().get('otp_force') and Setting().get('otp_field_enabled'): user.update_profile(enable_otp=True) prepare_welcome_user(user.id) return redirect(url_for('index.welcome')) else: return redirect(url_for('index.login')) else: return render_template('register.html', error=result['msg'], captcha_enable=CAPTCHA_ENABLE) except Exception as e: return render_template('register.html', error=e, captcha_enable=CAPTCHA_ENABLE) else: return render_template('errors/404.html'), 404 # Show welcome page on first login if otp_force is enabled @index_bp.route('/welcome', methods=['GET', 'POST']) def welcome(): if 'welcome_user_id' not in session: return redirect(url_for('index.index')) user = User(id=session['welcome_user_id']) encoded_img_data = base64.b64encode(user.get_qrcode_value()) if request.method == 'GET': return render_template('register_otp.html', qrcode_image=encoded_img_data.decode(), user=user) elif request.method == 'POST': otp_token = request.form.get('otptoken', '') 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") else: 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/', methods=['GET']) def confirm_email(token): email = confirm_token(token) if not email: # Cannot confirm email return render_template('email_confirmation.html', status=0) user = User.query.filter_by(email=email).first_or_404() if user.confirmed: # Already confirmed current_app.logger.info( "User email {} already confirmed".format(email)) return render_template('email_confirmation.html', status=2) else: # Confirm email is valid user.update_confirmed(confirmed=1) current_app.logger.info( "User email {} confirmed successfully".format(email)) return render_template('email_confirmation.html', status=1) @index_bp.route('/resend-confirmation-email', methods=['GET', 'POST']) def resend_confirmation_email(): if current_user.is_authenticated: return redirect(url_for('index.index')) if request.method == 'GET': return render_template('resend_confirmation_email.html') elif request.method == 'POST': email = request.form.get('email') user = User.query.filter(User.email == email).first() if not user: # Email not found status = 0 elif user.confirmed: # Email already confirmed status = 1 else: # Send new confirmed email send_account_verification(user.email) status = 2 return render_template('resend_confirmation_email.html', status=status) @index_bp.route('/nic/checkip.html', methods=['GET', 'POST']) @csrf.exempt def dyndns_checkip(): # This route covers the default ddclient 'web' setting for the checkip service return render_template('dyndns.html', response=request.environ.get( 'HTTP_X_REAL_IP', request.remote_addr)) @index_bp.route('/nic/update', methods=['GET', 'POST']) @csrf.exempt @dyndns_login_required def dyndns_update(): # dyndns protocol response codes in use are: # good: update successful # nochg: IP address already set to update address # nohost: hostname does not exist for this user account # 911: server error # have to use 200 HTTP return codes because ddclient does not read the return string if the code is other than 200 # reference: https://help.dyn.com/remote-access-api/perform-update/ # reference: https://help.dyn.com/remote-access-api/return-codes/ hostname = request.args.get('hostname') myip = request.args.get('myip') if not hostname: history = History(msg="DynDNS update: missing hostname parameter", created_by=current_user.username) history.add() return render_template('dyndns.html', response='nohost'), 200 try: if current_user.role.name in ['Administrator', 'Operator']: domains = Domain.query.all() else: # Get query for domain to which the user has access permission. # This includes direct domain permission AND permission through # account membership domains = db.session.query(Domain) \ .outerjoin(DomainUser, Domain.id == DomainUser.domain_id) \ .outerjoin(Account, Domain.account_id == Account.id) \ .outerjoin(AccountUser, Account.id == AccountUser.account_id) \ .filter( db.or_( DomainUser.user_id == current_user.id, AccountUser.user_id == current_user.id )).all() except Exception as e: current_app.logger.error('DynDNS Error: {0}'.format(e)) current_app.logger.debug(traceback.format_exc()) return render_template('dyndns.html', response='911'), 200 domain = None domain_segments = hostname.split('.') for _index in range(len(domain_segments)): full_domain = '.'.join(domain_segments) potential_domain = Domain.query.filter( Domain.name == full_domain).first() if potential_domain in domains: domain = potential_domain break domain_segments.pop(0) if not domain: history = History( msg= "DynDNS update: attempted update of {0} but it does not exist for this user" .format(hostname), created_by=current_user.username) history.add() return render_template('dyndns.html', response='nohost'), 200 myip_addr = [] if myip: for address in myip.split(','): myip_addr += utils.validate_ipaddress(address) remote_addr = utils.validate_ipaddress( request.headers.get('X-Forwarded-For', request.remote_addr).split(', ')[0]) response = 'nochg' for ip in myip_addr or remote_addr: if isinstance(ip, ipaddress.IPv4Address): rtype = 'A' else: rtype = 'AAAA' r = Record(name=hostname, type=rtype) # Check if the user requested record exists within this domain if r.exists(domain.name) and r.is_allowed_edit(): if r.data == str(ip): # Record content did not change, return 'nochg' history = History( msg= "DynDNS update: attempted update of {0} but record already up-to-date" .format(hostname), created_by=current_user.username, domain_id=domain.id) history.add() else: oldip = r.data result = r.update(domain.name, str(ip)) if result['status'] == 'ok': history = History( msg='DynDNS update: updated {} successfully'.format(hostname), detail = json.dumps({ 'domain': domain.name, 'record': hostname, 'type': rtype, 'old_value': oldip, 'new_value': str(ip) }), created_by=current_user.username, domain_id=domain.id) history.add() response = 'good' else: response = '911' break elif r.is_allowed_edit(): ondemand_creation = DomainSetting.query.filter( DomainSetting.domain == domain).filter( DomainSetting.setting == 'create_via_dyndns').first() if (ondemand_creation is not None) and (strtobool( ondemand_creation.value) == True): # Build the rrset rrset_data = [{ "changetype": "REPLACE", "name": hostname + '.', "ttl": 3600, "type": rtype, "records": [{ "content": str(ip), "disabled": False }], "comments": [] }] # Format the rrset rrset = {"rrsets": rrset_data} result = Record().add(domain.name, rrset) if result['status'] == 'ok': history = History( msg= 'DynDNS update: created record {0} in zone {1} successfully' .format(hostname, domain.name, str(ip)), detail = json.dumps({ 'domain': domain.name, 'record': hostname, 'value': str(ip) }), created_by=current_user.username, domain_id=domain.id) history.add() response = 'good' else: history = History( msg= 'DynDNS update: attempted update of {0} but it does not exist for this user' .format(hostname), created_by=current_user.username) history.add() return render_template('dyndns.html', response=response), 200 ### START SAML AUTHENTICATION ### @index_bp.route('/saml/login') def saml_login(): if not current_app.config.get('SAML_ENABLED', False): abort(400) from onelogin.saml2.utils import OneLogin_Saml2_Utils req = saml.prepare_flask_request(request) auth = saml.init_saml_auth(req) redirect_url = OneLogin_Saml2_Utils.get_self_url(req) + url_for( 'index.saml_authorized') return redirect(auth.login(return_to=redirect_url)) @index_bp.route('/saml/metadata') def saml_metadata(): if not current_app.config.get('SAML_ENABLED', False): current_app.logger.error("SAML authentication is disabled.") abort(400) from onelogin.saml2.utils import OneLogin_Saml2_Utils req = saml.prepare_flask_request(request) auth = saml.init_saml_auth(req) settings = auth.get_settings() metadata = settings.get_sp_metadata() errors = settings.validate_metadata(metadata) if len(errors) == 0: resp = make_response(metadata, 200) resp.headers['Content-Type'] = 'text/xml' else: resp = make_response(errors.join(', '), 500) return resp @index_bp.route('/saml/authorized', methods=['GET', 'POST']) @csrf.exempt def saml_authorized(): errors = [] if not current_app.config.get('SAML_ENABLED', False): current_app.logger.error("SAML authentication is disabled.") abort(400) from onelogin.saml2.utils import OneLogin_Saml2_Utils req = saml.prepare_flask_request(request) auth = saml.init_saml_auth(req) auth.process_response() current_app.logger.debug( auth.get_attributes() ) errors = auth.get_errors() if len(errors) == 0: session['samlUserdata'] = auth.get_attributes() session['samlNameId'] = auth.get_nameid() session['samlSessionIndex'] = auth.get_session_index() self_url = OneLogin_Saml2_Utils.get_self_url(req) self_url = self_url + req['script_name'] if 'RelayState' in request.form and self_url != request.form[ 'RelayState']: return redirect(auth.redirect_to(request.form['RelayState'])) if current_app.config.get('SAML_ATTRIBUTE_USERNAME', False): username = session['samlUserdata'][ current_app.config['SAML_ATTRIBUTE_USERNAME']][0].lower() else: username = session['samlNameId'].lower() user = User.query.filter_by(username=username).first() if not user: # create user user = User(username=username, plain_text_password=None, email=session['samlNameId']) user.create_local_user() session['user_id'] = user.id email_attribute_name = current_app.config.get('SAML_ATTRIBUTE_EMAIL', 'email') givenname_attribute_name = current_app.config.get( 'SAML_ATTRIBUTE_GIVENNAME', 'givenname') surname_attribute_name = current_app.config.get( 'SAML_ATTRIBUTE_SURNAME', 'surname') name_attribute_name = current_app.config.get('SAML_ATTRIBUTE_NAME', None) account_attribute_name = current_app.config.get( 'SAML_ATTRIBUTE_ACCOUNT', None) admin_attribute_name = current_app.config.get('SAML_ATTRIBUTE_ADMIN', None) group_attribute_name = current_app.config.get('SAML_ATTRIBUTE_GROUP', None) admin_group_name = current_app.config.get('SAML_GROUP_ADMIN_NAME', None) operator_group_name = current_app.config.get('SAML_GROUP_OPERATOR_NAME', None) group_to_account_mapping = create_group_to_account_mapping() if email_attribute_name in session['samlUserdata']: user.email = session['samlUserdata'][email_attribute_name][ 0].lower() if givenname_attribute_name in session['samlUserdata']: user.firstname = session['samlUserdata'][givenname_attribute_name][ 0] if surname_attribute_name in session['samlUserdata']: user.lastname = session['samlUserdata'][surname_attribute_name][0] if name_attribute_name in session['samlUserdata']: name = session['samlUserdata'][name_attribute_name][0].split(' ') user.firstname = name[0] user.lastname = ' '.join(name[1:]) if group_attribute_name: user_groups = session['samlUserdata'].get(group_attribute_name, []) else: user_groups = [] if admin_attribute_name or group_attribute_name: user_accounts = set(user.get_accounts()) saml_accounts = [] for group_mapping in group_to_account_mapping: mapping = group_mapping.split('=') group = mapping[0] account_name = mapping[1] if group in user_groups: account = handle_account(account_name) saml_accounts.append(account) for account_name in session['samlUserdata'].get( account_attribute_name, []): account = handle_account(account_name) saml_accounts.append(account) saml_accounts = set(saml_accounts) for account in saml_accounts - user_accounts: account.add_user(user) history = History(msg='Adding {0} to account {1}'.format( user.username, account.name), created_by='SAML Assertion') history.add() for account in user_accounts - saml_accounts: account.remove_user(user) history = History(msg='Removing {0} from account {1}'.format( user.username, account.name), created_by='SAML Assertion') history.add() if admin_attribute_name and 'true' in session['samlUserdata'].get( admin_attribute_name, []): uplift_to_admin(user) elif admin_group_name in user_groups: uplift_to_admin(user) elif operator_group_name in user_groups: uplift_to_operator(user) elif admin_attribute_name or group_attribute_name: if user.role.name != 'User': user.role_id = Role.query.filter_by(name='User').first().id history = History(msg='Demoting {0} to user'.format( user.username), created_by='SAML Assertion') history.add() user.plain_text_password = None user.update_profile() session['authentication_type'] = 'SAML' return authenticate_user(user, 'SAML') else: return render_template('errors/SAML.html', errors=errors) def create_group_to_account_mapping(): group_to_account_mapping_string = current_app.config.get( 'SAML_GROUP_TO_ACCOUNT_MAPPING', None) if group_to_account_mapping_string and len( group_to_account_mapping_string.strip()) > 0: group_to_account_mapping = group_to_account_mapping_string.split(',') else: group_to_account_mapping = [] return group_to_account_mapping def handle_account(account_name, account_description=""): clean_name = Account.sanitize_name(account_name) account = Account.query.filter_by(name=clean_name).first() if not account: account = Account(name=clean_name, description=account_description, contact='', mail='') account.create_account() history = History(msg='Account {0} created'.format(account.name), created_by='OIDC/SAML Assertion') history.add() else: account.description = account_description account.update_account() return account def uplift_to_admin(user): if user.role.name != 'Administrator': user.role_id = Role.query.filter_by(name='Administrator').first().id history = History(msg='Promoting {0} to administrator'.format( user.username), 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 history = History(msg='Promoting {0} to operator'.format( user.username), created_by='SAML Assertion') history.add() @index_bp.route('/saml/sls') def saml_logout(): req = saml.prepare_flask_request(request) auth = saml.init_saml_auth(req) url = auth.process_slo() errors = auth.get_errors() if len(errors) == 0: clear_session() if url is not None: return redirect(url) elif current_app.config.get('SAML_LOGOUT_URL') is not None: return redirect(current_app.config.get('SAML_LOGOUT_URL')) else: return redirect(url_for('login')) else: return render_template('errors/SAML.html', errors=errors) ### END SAML AUTHENTICATION ### @index_bp.route('/swagger', methods=['GET']) def swagger_spec(): try: spec_path = os.path.join(current_app.root_path, "swagger-spec.yaml") spec = open(spec_path, 'r') loaded_spec = load(spec.read(), Loader) except Exception as e: current_app.logger.error( 'Cannot view swagger spec. Error: {0}'.format(e)) current_app.logger.debug(traceback.format_exc()) abort(500) resp = make_response(json.dumps(loaded_spec), 200) resp.headers['Content-Type'] = 'application/json' return resp