cwinfo/powerdns-admin
4
0
Fork 0
mirror of https://github.com/cwinfo/powerdns-admin.git synced 2025-09-16 07:12:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'fix-saml'

Browse Source
This commit is contained in:
Khanh Ngo
2018-08-07 09:09:34 +07:00
parent 501b817808 480989e86a
commit 47d5858fc6
4 changed files with 135 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
app/views.py
View File

@@ -17,7 +17,7 @@ from flask_login import login_user, logout_user, current_user, login_required
from werkzeug import secure_filename
from werkzeug.security import gen_salt
from .models import User, Account, Domain, Record, Server, History, Anonymous, Setting, DomainSetting, DomainTemplate, DomainTemplateRecord
from .models import User, Account, Domain, Record, Role, Server, History, Anonymous, Setting, DomainSetting, DomainTemplate, DomainTemplateRecord
from app import app, login_manager, github, google
from app.lib import utils
from app.decorators import admin_role_required, can_access_domain, can_configure_dnssec
@@ -227,20 +227,65 @@ def saml_authorized():
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']))
user = User.query.filter_by(username=session['samlNameId'].lower()).first()
if app.config.get('SAML_ATTRIBUTE_USERNAME', False):
username = session['samlUserdata'][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=session['samlNameId'],
user = User(username=username,
plain_text_password = None,
email=session['samlNameId'])
user.create_local_user()
session['user_id'] = user.id
if session['samlUserdata'].has_key("email"):
user.email = session['samlUserdata']["email"][0].lower()
if session['samlUserdata'].has_key("givenname"):
user.firstname = session['samlUserdata']["givenname"][0]
if session['samlUserdata'].has_key("surname"):
user.lastname = session['samlUserdata']["surname"][0]
email_attribute_name = app.config.get('SAML_ATTRIBUTE_EMAIL', 'email')
givenname_attribute_name = app.config.get('SAML_ATTRIBUTE_GIVENNAME', 'givenname')
surname_attribute_name = app.config.get('SAML_ATTRIBUTE_SURNAME', 'surname')
account_attribute_name = app.config.get('SAML_ATTRIBUTE_ACCOUNT', None)
admin_attribute_name = app.config.get('SAML_ATTRIBUTE_ADMIN', None)
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 admin_attribute_name:
user_accounts = set(user.get_account())
saml_accounts = []
for account_name in session['samlUserdata'].get(account_attribute_name, []):
clean_name = ''.join(c for c in account_name.lower() if c in "abcdefghijklmnopqrstuvwxyz0123456789")
if len(clean_name) > Account.name.type.length:
logging.error("Account name {0} too long. Truncated.".format(clean_name))
account = Account.query.filter_by(name=clean_name).first()
if not account:
account = Account(name=clean_name.lower(), description='', contact='', mail='')
account.create_account()
history = History(msg='Account {0} created'.format(account.name), created_by='SAML Assertion')
history.add()
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:
if 'true' in session['samlUserdata'].get(admin_attribute_name, []):
admin_role = Role.query.filter_by(name='Administrator').first().id
if user.role_id != admin_role:
user.role_id = admin_role
history = History(msg='Promoting {0} to administrator'.format(user.username), created_by='SAML Assertion')
history.add()
else:
user_role = Role.query.filter_by(name='User').first().id
if user.role_id != user_role:
user.role_id = user_role
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['external_auth'] = True