ldap-python-webui/app.py

1005 lines
33 KiB
Python
Raw Normal View History

2022-04-09 17:38:00 +02:00
#!/usr/bin/env python3
'''
ldap-python-webui :: LDAP kudeaketarako Web Interfazea - Web UI for LDAP management.
Copyright (C) 2022 Aitzol Berasategi - Wproject
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
'''
import bottle
from bottle import get, post, static_file, request, route, template
from bottle import SimpleTemplate
from bottle.ext import beaker
from configparser import ConfigParser
from ldap3 import Server, Connection, ALL
2023-04-06 16:40:24 +02:00
from ldap3 import SIMPLE, SUBTREE, MODIFY_REPLACE, MODIFY_ADD, ALL_ATTRIBUTES
2022-04-09 17:38:00 +02:00
from ldap3.core.exceptions import LDAPBindError, LDAPConstraintViolationResult, \
LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError, \
2023-04-04 11:02:56 +02:00
LDAPSocketOpenError, LDAPExceptionError, LDAPAttributeOrValueExistsResult
2022-04-09 17:38:00 +02:00
import logging
from os import getenv, environ, path
2022-04-16 23:13:00 +02:00
from libs import flist, slist
from libs.localization import *
2023-04-07 15:21:44 +02:00
from libs.helper import tools
2022-04-09 17:38:00 +02:00
import random
2023-04-29 11:46:01 +02:00
from user_agents import parse as ua_parse
2023-04-06 16:40:24 +02:00
from datetime import datetime
2022-04-09 17:38:00 +02:00
BASE_DIR = path.dirname(__file__)
LOG = logging.getLogger(__name__)
LOG_FORMAT = '%(asctime)s %(levelname)s: %(message)s'
2023-04-04 13:25:10 +02:00
VERSION = '0.0.2'
2022-04-09 17:38:00 +02:00
@get('/')
def get_index():
try:
return user_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
@get('/user')
def get_index():
try:
2023-04-04 13:25:10 +02:00
print(newSession().get())
2022-04-09 17:38:00 +02:00
return user_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
@get('/signup')
def get_index():
newSession()
2022-09-11 21:58:25 +02:00
if allowed:
2022-09-11 19:28:52 +02:00
return signup_tpl(str=i18n.str)
else:
return index_tpl(alerts=[('error', i18n.msg[28], 'upDown')], str=i18n.str)
2022-04-09 17:38:00 +02:00
@get('/change_pwd')
def get_index():
try:
return change_pwd_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
@get('/edit_fullname')
def get_index():
try:
return edit_fullname_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
@get('/edit_email')
def get_index():
try:
return edit_email_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
2023-04-06 22:09:26 +02:00
@get('/logs')
def get_index():
try:
return logs_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
2022-04-09 17:38:00 +02:00
@get('/delete')
def get_index():
try:
return delete_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
@get('/logout')
def get_index():
def error(msg):
return index_tpl(alerts=[('error', msg, 'fadeOut')], str=i18n.str)
try:
username = newSession().get()['username']
if(username is not None):
logout(username)
except Error as e:
LOG.warning("Unsuccessful attempt to log out: %s" % e)
return error(str(e))
return index_tpl(alerts=[('success', i18n.msg[0], 'fadeOut')], str=i18n.str)
2023-11-10 13:12:15 +01:00
@get('/_2fa')
def get_index():
try:
return _2fa_tpl(data=newSession().get(), str=i18n.str)
except Exception as e:
return index_tpl(str=i18n.str)
2022-04-09 17:38:00 +02:00
@post('/user')
def post_user():
form = request.forms.getunicode
def error(msg):
return index_tpl(alerts=[('error', msg, 'fadeOut')], str=i18n.str)
if len(form('username')) < 3:
2022-04-25 17:20:59 +02:00
return error(i18n.msg[2])
2022-04-25 13:10:25 +02:00
elif not tools.input_validation(form('username')):
2022-04-25 17:20:59 +02:00
return error(i18n.msg[3])
2022-04-09 17:38:00 +02:00
2022-04-25 13:10:25 +02:00
if not tools.pwd_validation(form('password')):
return error(i18n.msg[21])
2022-04-09 17:38:00 +02:00
try:
login(form('username'), form('password'))
except Error as e:
LOG.warning("Unsuccessful attempt to login %s: %s" % (form('username'), e))
return error(str(e))
2023-11-10 13:12:15 +01:00
#if 2fa not chekced || (2fa checked & success)
2022-04-25 17:20:59 +02:00
return user_tpl(alerts=[('success', '%s %s' % (i18n.msg[1], form('username').capitalize()), 'fadeOut' )], data=newSession().get(), str=i18n.str)
2023-11-10 13:12:15 +01:00
#elif 2fa checked
#return _2fa_tpl(alerts=[('success', '%s %s' % (i18n.msg[1], form('username').capitalize()), 'fadeOut' )], data=newSession().get(), str=i18n.str)
2022-04-09 17:38:00 +02:00
@post('/signup')
def post_signup():
2022-04-25 13:10:25 +02:00
2022-04-09 17:38:00 +02:00
#ensure that i18n exists
if 'i18n' not in globals():
newSession()
form = request.forms.getunicode
isFake = False
2022-04-16 23:13:00 +02:00
db = 'data/invite-codes.db'
2022-04-09 17:38:00 +02:00
def auto_complete(arg):
if arg == 'firstname':
result = random.choice(flist.firstname)
elif arg == 'surname':
result = random.choice(slist.surname)
2023-04-04 11:02:56 +02:00
return(result.capitalize())
2022-04-09 17:38:00 +02:00
def error(msg):
return signup_tpl(alerts=[('error', msg, 'fadeOut')], str=i18n.str)
2022-04-25 13:10:25 +02:00
if not tools.code_is_valid(form('invite_code'), db):
2022-04-25 17:20:59 +02:00
return(error(i18n.msg[6]))
2022-04-09 17:38:00 +02:00
if len(form('username')) < 3:
2022-04-25 17:20:59 +02:00
return error(i18n.msg[2])
2022-04-09 17:38:00 +02:00
username = form('username').lower()
2022-04-25 13:10:25 +02:00
if not tools.input_validation(username):
2022-04-25 17:20:59 +02:00
return error(i18n.msg[3])
2022-04-09 17:38:00 +02:00
if len(form('firstname')) == 0:
firstname = auto_complete('firstname')
isFake = True
else:
2023-04-04 11:02:56 +02:00
firstname = form('firstname').title()
2022-04-09 17:38:00 +02:00
if len(form('surname')) == 0:
surname = auto_complete('surname')
isFake = True
else:
2023-04-04 11:02:56 +02:00
surname = form('surname').title()
2022-04-09 17:38:00 +02:00
email = form('email').lower()
2022-04-25 13:10:25 +02:00
if not tools.email_validation(email):
2022-04-25 17:20:59 +02:00
return error(i18n.msg[15])
2022-04-09 17:38:00 +02:00
2022-04-25 13:10:25 +02:00
if not tools.pwd_validation(form('password')):
return error(i18n.msg[8]) #mezua ALDATU egin behar da
elif form('password') != form('confirm-password'):
2022-04-09 17:38:00 +02:00
return error(i18n.msg[7])
2023-04-04 20:26:56 +02:00
try:
account_request(username, firstname, surname, form('password'), email, isFake, get_dev())
2022-04-09 17:38:00 +02:00
except Error as e:
LOG.warning("Unsuccessful attempt to create an account for %s: %s" % (form('username'), e))
return error(str(e))
try:
2022-04-25 13:10:25 +02:00
tools.mark_code_as_used(form('invite_code'), db)
2022-04-09 17:38:00 +02:00
except Error as e:
LOG.warning("There was a problem verifying the invitation code, please try again later.", e)
return error(str(e))
LOG.info("New account successfully created for %s" % form('username'))
return index_tpl(alerts=[('success', i18n.msg[9], 'fadeOut')], str=i18n.str)
@post('/edit_fullname')
def post_edit_fullname():
form = request.forms.getunicode
try:
username = newSession().get()['username']
old_firstname = newSession().get()['firstname']
old_surname = newSession().get()['surname']
except Error as e:
return index_tpl(alerts=[('error', str(e), 'fadeOut')], str=i18n.str)
def error(msg):
return edit_fullname_tpl(alerts=[('error', msg, 'fadeOut')], data=newSession().get(), str=i18n.str)
if len(form('firstname')) < 3:
2022-04-16 23:13:00 +02:00
return error(i18n.msg[11])
2023-04-04 11:02:56 +02:00
elif not tools.input_validation(form('firstname'), True):
2022-04-25 17:20:59 +02:00
return error(i18n.msg[4])
2022-04-09 17:38:00 +02:00
if len(form('surname')) < 3:
2022-04-16 23:13:00 +02:00
return error(i18n.msg[12])
2023-04-04 11:02:56 +02:00
elif not tools.input_validation(form('surname'), True):
2022-04-25 17:20:59 +02:00
return error(i18n.msg[5])
2022-04-09 17:38:00 +02:00
try:
2023-04-04 11:02:56 +02:00
edit_fullname(username, old_firstname, old_surname, form('firstname').title(), form('surname').title())
2022-04-09 17:38:00 +02:00
except Error as e:
LOG.warning("Unsuccessful attempt to edit fullname for %s: %s" % (username, e))
return error(str(e))
2022-04-16 23:13:00 +02:00
return user_tpl(alerts=[('success', i18n.msg[13], 'fadeOut' )], data=newSession().get(), str=i18n.str)
2022-04-09 17:38:00 +02:00
@post('/edit_email')
def post_edit_email():
form = request.forms.getunicode
try:
username = newSession().get()['username']
old_email = newSession().get()['mail']
except Error as e:
return index_tpl(alerts=[('error', str(e), 'fadeOut')], str=i18n.str)
def error(msg):
return edit_email_tpl(alerts=[('error', msg, 'fadeOut')], data=newSession().get(), str=i18n.str)
2022-04-25 13:10:25 +02:00
if not tools.email_validation(form('email')):
2022-04-16 23:13:00 +02:00
return(error(i18n.msg[14]))
2022-04-09 17:38:00 +02:00
try:
edit_email(username, old_email, form('email').lower())
except Error as e:
LOG.warning("Unsuccessful attempt to change email addres for %s: %s" % (username, e))
return error(str(e))
2022-04-16 23:13:00 +02:00
return user_tpl(alerts=[('success', i18n.msg[16], 'fadeOut' )], data=newSession().get(), str=i18n.str)
2022-04-09 17:38:00 +02:00
@post('/change_pwd')
def post_change_pwd():
form = request.forms.getunicode
2022-04-25 13:10:25 +02:00
2022-04-09 17:38:00 +02:00
try:
username=newSession().get()['username']
except Error as e:
return index_tpl(alerts=[('error', str(e), 'fadeOut')], str=i18n.str)
def error(msg):
return change_pwd_tpl(username=username, alerts=[('error', msg, 'fadeOut')], str=i18n.str)
2022-04-25 17:20:59 +02:00
if not tools.pwd_validation(form('old-password')):
return error(i18n.msg[26])
elif (not tools.pwd_validation(form('new-password')) or not tools.pwd_validation(form('confirm-password'))):
2022-04-25 13:10:25 +02:00
return error(i18n.msg[8]) #mezua aldatu egin behar da
elif form('new-password') != form('confirm-password'):
2022-04-09 17:38:00 +02:00
return error(i18n.msg[7])
2022-04-25 13:10:25 +02:00
elif form('old-password') == form('confirm-password'):
2022-04-16 23:13:00 +02:00
return error(i18n.msg[17])
2022-04-09 17:38:00 +02:00
try:
change_passwords(username, form('old-password'), form('new-password'))
logout(username)
except Error as e:
LOG.warning("Unsuccessful attempt to change password for %s: %s" % (username, e))
return error(str(e))
LOG.info("Password successfully changed for: %s" % username)
2022-04-16 23:13:00 +02:00
return index_tpl(alerts=[('success', i18n.msg[18], 'fadeOut')], username=username, str=i18n.str)
2022-04-09 17:38:00 +02:00
@post('/delete')
def post_delete():
form = request.forms.getunicode
def error(msg):
return delete_tpl(alerts=[('error', msg, 'fadeOut')], str=i18n.str)
try:
username = newSession().get()['username']
2022-04-26 07:13:51 +02:00
if(tools.input_validation(form('username')) and form('username').lower() == username):
2022-04-09 17:38:00 +02:00
del_user(username)
else:
2022-04-16 23:13:00 +02:00
return(error(i18n.msg[19]))
2022-04-26 07:13:51 +02:00
2022-04-09 17:38:00 +02:00
except Error as e:
LOG.warning("Unsuccessful attempt to delete the account: %s" % e)
return error(str(e))
LOG.info("Account successfully deleted")
2022-04-16 23:13:00 +02:00
return index_tpl(alerts=[('success', i18n.msg[20], 'fadeOut')], str=i18n.str)
2022-04-09 17:38:00 +02:00
@route('/static/<filename>', name='static')
def serve_static(filename):
return static_file(filename, root=path.join(BASE_DIR, 'static'))
def index_tpl(**kwargs):
return template('index', **kwargs)
def user_tpl(**kwargs):
return template('user', **kwargs)
def signup_tpl(**kwargs):
return template('signup', **kwargs)
def change_pwd_tpl(**kwargs):
return template('change_pwd', **kwargs)
def edit_email_tpl(**kwargs):
return template('edit_email', **kwargs)
def edit_fullname_tpl(**kwargs):
return template('edit_fullname', **kwargs)
def delete_tpl(**kwargs):
return template('delete', **kwargs)
2023-04-06 22:09:26 +02:00
def logs_tpl(**kwargs):
return template('logs', **kwargs)
2023-11-10 13:12:15 +01:00
def _2fa_tpl(**kwargs):
return template('_2fa', **kwargs)
2022-04-09 17:38:00 +02:00
def connect_ldap(conf, **kwargs):
server = Server(host=conf['host'],
port=conf.getint('port', None),
use_ssl=conf.getboolean('use_ssl', False),
connect_timeout=5)
return Connection(server, raise_exceptions=True, **kwargs)
#LOGIN
def login(username, password):
n = N
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("%s is trying to logging in %s" % (username, key))
n -= 1
try:
login_user(CONF[key], username, password)
except Error as e:
if n >=1:
e = []
continue
else:
raise e
break
def login_user(conf, *args):
try:
login_user_ldap(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[21])
2022-04-09 17:38:00 +02:00
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def login_user_ldap(conf, username, password):
#set current LDAP
superUser = SuperUsers(conf)
#with connect_ldap(conf) as c:
with connect_ldap(conf, user=superUser.readonly_dn, password=superUser.readonly_pwd) as c:
user_dn = find_user_dn(conf, c, username)
# Note: raises LDAPUserNameIsMandatoryError when user_dn is None.
with connect_ldap(conf, authentication=SIMPLE, user=user_dn, password=password) as c:
c.bind()
2023-04-06 22:09:26 +02:00
if is_trusted_device(conf, user_dn):
newSession().set(get_user_data(user_dn, c))
#update timestamp + ip address
update_login_info(conf, user_dn)
LOG.debug("%s logged in to %s" % (username, conf['base']))
2022-04-09 17:38:00 +02:00
#LOGOUT
def logout(username):
n = N
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Logging out %s from %s" % (username, key))
n -= 1
try:
logout_user(CONF[key], username)
except Error as e:
if n >=1:
e = []
continue
else:
raise e
break
def logout_user(conf, *args):
try:
logout_user_ldap(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[21])
2022-04-09 17:38:00 +02:00
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def logout_user_ldap(conf, username):
#set current LDAP
superUser = SuperUsers(conf)
#with connect_ldap(conf) as c:
with connect_ldap(conf, user=superUser.readonly_dn, password=superUser.readonly_pwd) as c:
user_dn = find_user_dn(conf, c, username)
c.unbind()
#newSession().close()
newSession().delete()
LOG.info("%s LOGED OUT" % (username))
#SIGN UP
2023-04-04 20:26:56 +02:00
def account_request(username, firstname, surname, password, email, isFake, device):
2022-04-09 17:38:00 +02:00
created = []
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Creating account for %s on %s server" % (username, key))
try:
2023-04-04 20:26:56 +02:00
new_user_account(CONF[key], username, firstname, surname, password, email, isFake, device)
2022-04-09 17:38:00 +02:00
created.append(key)
except Error as e:
for key in reversed(created):
LOG.info("Reverting account creation in %s for %s" % (key, username))
try:
#Akatsen bat gertatzen bada LDAP instantzia guztietan kontua ezabatu
del_account(CONF[key], username)
except Error as e2:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e2))
raise e
def new_user_account(conf, *args):
try:
register(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[21])
2022-04-09 17:38:00 +02:00
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[22])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
2023-04-04 20:26:56 +02:00
def register(conf, username, firstname, surname, password, email, isFake, device):
2022-04-09 17:38:00 +02:00
def to_ascii(str):
ascii_str=""
for c in str:
if 0 <= ord(c) <= 127:
ascii_str=ascii_str+c
else:
ascii_str=ascii_str+"X"
return(ascii_str)
#set current LDAP
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
try:
if (find_user_dn(conf,c,username) is not None):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[24])
2022-04-09 17:38:00 +02:00
if (find_email(conf,c,email)):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[25])
2022-04-09 17:38:00 +02:00
except Error as e:
raise e
else:
#create new account
uidNumber = find_uid_number(conf,c)+1
directory = 'home/user/'+to_ascii(username)
OBJECT_CLASS = ['top', 'inetOrgPerson', 'posixAccount', 'accountsManagement']
2023-04-11 18:19:16 +02:00
ts = datetime.now().strftime('%Y%m%d%H%M%S')+'Z'
2023-04-04 20:26:56 +02:00
attributes = {'gidNumber': '501', 'uidNumber': uidNumber, 'homeDirectory': directory, 'givenName':
2023-04-05 13:23:26 +02:00
firstname, 'sn': surname, 'uid' : username, 'mail': email, 'active': False, 'fakeCn': isFake,
2023-11-10 13:12:15 +01:00
'devices':device, 'ip':request.environ.get('HTTP_X_REAL_IP', request.remote_addr), 'lastLogin': ts,
'secureAuth': False}
2022-04-29 10:16:13 +02:00
new_user_dn = "cn="+firstname+" "+surname+" - "+username+",cn=users,"+conf['base']
2022-04-09 17:38:00 +02:00
c.add(dn=new_user_dn,object_class=OBJECT_CLASS, attributes=attributes)
#create/change user password
c.extend.standard.modify_password(new_user_dn, '', password)
LOG.info("%s has registered on %s" % (username, conf))
#EDIT FULLNAME
def edit_fullname(username, old_firstname, old_surname, firstname, surname,):
changed = []
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Changing fullname in %s for %s" % (key, username))
try:
new_fullname(CONF[key], username, firstname, surname)
changed.append(key)
LOG.debug("%s changed fullname on %s" % (username, key))
except Error as e:
for key in reversed(changed):
LOG.info("Reverting fullname change in %s for %s" % (key, username))
try:
new_fullname(CONF[key], username, old_firstname, old_surname)
except Error as e2:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e2))
raise e
def new_fullname(conf, *args):
try:
update_fullname(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[26])
2022-04-09 17:38:00 +02:00
except LDAPConstraintViolationResult as e:
# Extract useful part of the error message (for Samba 4 / AD).
msg = e.message.split('check_password_restrictions: ')[-1].capitalize()
raise Error(msg)
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def update_fullname(conf, username, firstname, surname):
#set current LDAP
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
#with connect_ldap(conf) as c:
user_dn = find_user_dn(conf, c, username)
c.modify(user_dn, {'givenName': [( MODIFY_REPLACE, firstname )], 'sn': [( MODIFY_REPLACE, surname )]})
2022-04-29 10:16:13 +02:00
new_cn = "cn="+firstname+" "+ surname+" - "+ username
2022-04-09 17:38:00 +02:00
c.modify_dn(user_dn, new_cn)
new_user_dn = new_cn+",cn=users,"+conf['base']
base = ",cn=users," + conf['base']
fakeFullName = user_dn[3:-len(base)].split(" ")
if(user_dn == new_user_dn):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[10])
2022-04-09 17:38:00 +02:00
c.modify(new_user_dn, {'fakeCn': [(MODIFY_REPLACE, 'false' )]})
newSession().set(get_user_data(new_user_dn, c))
#EDIT EMAIL
def edit_email(username, old_email, new_email):
changed = []
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Changing email in %s for %s" % (key, username))
try:
new_email_address(CONF[key], username, old_email, new_email)
changed.append(key)
LOG.debug("%s changed email address on %s" % (username, key))
except Error as e:
for key in reversed(changed):
LOG.info("Reverting email change in %s for %s" % (key, username))
try:
new_email_address(CONF[key], username, new_email, old_email)
except Error as e2:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e2))
raise e
def new_email_address(conf, *args):
try:
update_email_address(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[26])
2022-04-09 17:38:00 +02:00
except LDAPConstraintViolationResult as e:
# Extract useful part of the error message (for Samba 4 / AD).
msg = e.message.split('check_password_restrictions: ')[-1].capitalize()
raise Error(msg)
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def update_email_address(conf, username, old_email, new_email):
if(old_email == new_email):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[15])
2022-04-09 17:38:00 +02:00
#set current LDAP
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
user_dn = find_user_dn(conf, c, username)
new_email_addresses = get_user_email_array(user_dn, c, old_email, new_email)
c.modify(user_dn, {'mail': [( MODIFY_REPLACE, new_email_addresses )]})
newSession().set(get_user_data(user_dn, c))
#CHANGE PASSWORD
def change_passwords(username, old_pass, new_pass):
changed = []
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Changing password in %s for %s" % (key, username))
try:
change_password(CONF[key], username, old_pass, new_pass)
changed.append(key)
LOG.debug("%s changed pwd on %s" % (username, key))
except Error as e:
for key in reversed(changed):
LOG.info("Reverting password change in %s for %s" % (key, username))
try:
change_password(CONF[key], username, new_pass, old_pass)
except Error as e2:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e2))
raise e
def change_password(conf, *args):
try:
if conf.get('type') == 'ad':
change_password_ad(conf, *args)
else:
change_password_ldap(conf, *args)
except (LDAPBindError, LDAPInvalidCredentialsResult, LDAPUserNameIsMandatoryError):
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[26])
2022-04-09 17:38:00 +02:00
except LDAPConstraintViolationResult as e:
# Extract useful part of the error message (for Samba 4 / AD).
msg = e.message.split('check_password_restrictions: ')[-1].capitalize()
raise Error(msg)
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def change_password_ldap(conf, username, old_pass, new_pass):
#set current LDAP
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.readonly_dn, password=superUser.readonly_pwd) as c:
#with connect_ldap(conf) as c:
user_dn = find_user_dn(conf, c, username)
# Note: raises LDAPUserNameIsMandatoryError when user_dn is None.
with connect_ldap(conf, authentication=SIMPLE, user=user_dn, password=old_pass) as c:
c.bind()
c.extend.standard.modify_password(user_dn, old_pass, new_pass)
def change_password_ad(conf, username, old_pass, new_pass):
user = username + '@' + conf['ad_domain']
with connect_ldap(conf, authentication=SIMPLE, user=user, password=old_pass) as c:
c.bind()
user_dn = find_user_dn(conf, c, username)
c.extend.microsoft.modify_password(user_dn, new_pass, old_pass)
#DELETE ACCOUNT
def del_user(username):
n = N
for key in (key for key in CONF.sections()
if key == 'ldap' or key.startswith('ldap:')):
LOG.debug("Deleting account for %s from %s" % (username, key))
n -= 1
try:
del_account(CONF[key], username)
LOG.debug("Account for %s deleted on -> %s" % (username, CONF[key]))
if(n == 0 and newSession().get()['username'] is not None):
newSession().delete()
except Error as e:
raise e
def del_account(conf, *args):
try:
delete(conf, *args)
except LDAPSocketOpenError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-25 17:20:59 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
except LDAPExceptionError as e:
LOG.error('{}: {!s}'.format(e.__class__.__name__, e))
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[23])
2022-04-09 17:38:00 +02:00
def delete(conf, username):
#set current LDAP
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
try:
user_dn = find_user_dn(conf, c, username)
c.delete(user_dn)
except Error as e:
raise e
#AUXILIARY FUNCTIONS
#find user
def find_user_dn(conf, conn, uid):
search_filter = conf['search_filter'].replace('{uid}', uid)
conn.search(conf['base'], "(%s)" % search_filter, SUBTREE)
return conn.response[0]['dn'] if conn.response else None
#find email
def find_email(conf, conn, email):
search_filter = '(uid=*)'
if conn.search(conf['base'], search_filter, attributes=['mail']):
for i in conn.response:
for j in i['attributes']['mail']:
if(j == email):
return True
return False
2023-04-11 18:19:16 +02:00
#find device
def find_device(user_dn, conn, device):
search_filter = '(objectClass=*)'
if conn.search(user_dn, search_filter, attributes=['devices']):
2023-04-05 13:23:26 +02:00
for i in conn.response:
for j in i['attributes']['devices']:
if(j == device):
return True
2023-04-11 18:19:16 +02:00
return False
2023-04-05 13:23:26 +02:00
2022-04-09 17:38:00 +02:00
#find highest uidNumber
def find_uid_number(conf, conn):
search_filter = '(uid=*)'
if conn.search(conf['base'], search_filter, attributes=['uidNumber']):
uidNumbersList=[]
for i in conn.response:
uidNumbersList.append(i['attributes']['uidNumber'])
2022-04-24 18:39:17 +02:00
uidNumbersList = list(filter(lambda i: type(i) is int, uidNumbersList))
2022-04-09 17:38:00 +02:00
return max(uidNumbersList)
else:
return(999)
def get_user_email_array(user_dn, conn, old_email, new_email):
search_filter = '(objectClass=*)'
conn.search(user_dn, search_filter, attributes=['mail'])
emails = conn.entries[0].mail.values
for i in range(len(emails)):
if(emails[i] == old_email):
emails[i] = new_email
return(emails)
def get_user_data(user_dn, conn):
search_filter = '(objectClass=*)'
2023-11-10 13:12:15 +01:00
conn.search(user_dn, search_filter,
attributes=['active','fakeCn','givenName','sn','uid','mail','devices','ip','lastLogin','secureAuth'])
2022-04-09 17:38:00 +02:00
data = []
data.append(conn.entries[0].active.values[0])
data.append(conn.entries[0].fakeCn.values[0])
data.append(conn.entries[0].givenName.values[0])
data.append(conn.entries[0].sn.values[0])
data.append(conn.entries[0].uid.values[0])
data.append(conn.entries[0].mail.values[0])
2023-04-05 13:23:26 +02:00
data.append(conn.entries[0].devices.values)
2023-04-06 22:09:26 +02:00
data.append(conn.entries[0].ip.values[0])
#ts = conn.entries[0].lastLogin.values[0]
#ts = datetime.strptime(ts, '%Y-%m-%d %H:%M:%S%z')
#ts = datetime.strftime(t, '%Y-%m-%d %H:%M:%S')
data.append(str(conn.entries[0].lastLogin.values[0])[:-6])
2023-11-10 13:12:15 +01:00
data.append(conn.entries[0].secureAuth.values[0])
2023-04-06 22:09:26 +02:00
2022-04-09 17:38:00 +02:00
return(data)
def read_config():
config = ConfigParser()
config.read([path.join(BASE_DIR, 'settings.ini'), getenv('CONF_FILE', '')])
return config
CONF = read_config()
def ldaps_count():
keys = []
for i, key in enumerate(CONF.sections()):
if key == 'ldap' or key.startswith('ldap:'):
keys.append(key)
#n=len(keys)
return(len(keys))
N = ldaps_count()
2022-09-11 19:28:52 +02:00
def reg():
if CONF['general']['allow_registration'] == 'True':
return True
2022-09-11 21:58:25 +02:00
allowed = reg()
2022-09-11 19:28:52 +02:00
2023-04-04 20:26:56 +02:00
def get_dev():
ua_string = bottle.request.environ.get('HTTP_USER_AGENT')
2023-04-29 11:46:01 +02:00
user_agent = ua_parse(ua_string)
2023-04-04 20:26:56 +02:00
return str(user_agent)
2023-04-06 22:09:26 +02:00
def is_trusted_device(conf, user_dn):
2023-04-06 16:40:24 +02:00
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
d = get_dev()
2023-04-11 18:19:16 +02:00
try:
if not find_device(user_dn, c, d):
OBJECT_CLASS = ['top', 'inetOrgPerson', 'posixAccount', 'accountsManagement']
c.modify(user_dn, {'devices': [( MODIFY_ADD, d )] })
'''
if find_device(user_dn, c, 'unknown'):
OBJECT_CLASS = ['top', 'inetOrgPerson', 'posixAccount', 'accountsManagement']
c.modify(user_dn, {'devices': [( MODIFY_REPLACE, d )] })
else:
OBJECT_CLASS = ['top', 'inetOrgPerson', 'posixAccount', 'accountsManagement']
c.modify(user_dn, {'devices': [( MODIFY_ADD, d )] })
'''
c.unbind()
return True
except Exception as e:
print(e)
return True
2023-04-11 18:22:56 +02:00
2023-04-06 22:09:26 +02:00
def update_login_info(conf, user_dn):
superUser = SuperUsers(conf)
with connect_ldap(conf, user=superUser.admin_dn, password=superUser.admin_pwd) as c:
ip = request.environ.get('HTTP_X_REAL_IP', request.remote_addr)
2023-04-11 18:19:16 +02:00
ts = datetime.now().strftime('%Y%m%d%H%M%S')+'Z'
c.modify(user_dn, {'ip': [( MODIFY_REPLACE, str(ip) )], 'lastLogin': [( MODIFY_REPLACE, ts )] })
2023-04-06 22:09:26 +02:00
c.unbind()
2023-04-06 16:40:24 +02:00
2022-04-09 17:38:00 +02:00
class Error(Exception):
pass
#SESSIONS MANAGEMENT
def newSession():
class Session(object):
"""docstring for Session"""
def __init__(self):
super(Session, self).__init__()
self.data = bottle.request.environ.get('beaker.session')
self.lang = self.get_lang()
#localization
2022-04-16 23:13:00 +02:00
self.lang = self.get_lang()
2022-04-09 17:38:00 +02:00
global i18n
2022-04-16 23:13:00 +02:00
i18n = LocalizeTo(self.lang, CONF)
2022-04-09 17:38:00 +02:00
def get_lang(self):
if 'HTTP_ACCEPT_LANGUAGE' in bottle.request.environ:
lang = bottle.request.get('HTTP_ACCEPT_LANGUAGE')
return str(lang[:2])
else:
return CONF['locale']['lang']
def get(self):
if 'username' in self.data:
return(self.data)
else:
2022-04-16 23:13:00 +02:00
raise Error(i18n.msg[27])
2022-04-09 17:38:00 +02:00
def set(self, data):
self.active = data[0]
self.fakeCn = data[1]
self.firstname = data[2]
self.surname = data[3]
self.username = data[4]
self.mail = data[5]
2023-04-05 13:23:26 +02:00
self.devices = data[6]
2023-04-06 22:09:26 +02:00
self.ip = data[7]
self.lastLogin = data[8]
2023-11-10 13:12:15 +01:00
self.secureAuth = data[9]
2022-04-09 17:38:00 +02:00
self.data['active'] = self.active
self.data['fakeCn'] = self.fakeCn
self.data['firstname'] = self.firstname
self.data['surname'] = self.surname
self.data['username'] = self.username
self.data['mail'] = self.mail
2023-04-05 13:23:26 +02:00
self.data['devices'] = self.devices
2023-04-06 22:09:26 +02:00
self.data['ip'] = self.ip
self.data['lastLogin'] = self.lastLogin
2023-11-10 13:12:15 +01:00
self.data['secureAuth'] = self.secureAuth
2022-04-09 17:38:00 +02:00
def close(self):
self.data.pop('username')
def delete(self):
self.data.delete()
s=Session()
return s
if environ.get('DEBUG'):
bottle.debug(True)
# Set up logging.
logging.basicConfig(format=LOG_FORMAT)
LOG.setLevel(logging.INFO)
LOG.info("Starting ldap-python-webui %s" % VERSION)
session_opts = {
'session.type': CONF['session']['type'],
'session.cookie_expires': CONF['session']['expire'],
'session.data_dir': CONF['session']['data_dir'],
'session.auto': CONF['session']['auto']
}
class SuperUsers(object):
"""docstring for Session"""
def __init__(self, conf):
super(SuperUsers, self).__init__()
self.domain=conf['base'][conf['base'].find(","):]
self.admin_dn="cn=admin"+self.domain
self.admin_pwd=environ['LDAP_ADMIN_PASSWORD']
self.readonly_dn="cn=readonly"+self.domain
self.readonly_pwd=environ['LDAP_READONLY_PASSWORD']
superUser = SuperUsers(CONF['ldap:0'])
app = beaker.middleware.SessionMiddleware(bottle.app(), session_opts)
bottle.TEMPLATE_PATH = [BASE_DIR]
# Set default attributes to pass into templates.
#SimpleTemplate.defaults = dict(CONF['html'])
SimpleTemplate.defaults['url'] = bottle.url
# Run bottle internal server when invoked directly (mainly for development).
if __name__ == '__main__':
bottle.run(app, **CONF['server'])
# Run bottle in application mode (in production under uWSGI server).
else:
#application = bottle.default_app()
application = app