# import ldap # from flask import current_app # from flask_login import UserMixin # # from app_common import make_ldap_connection, close_ldap_connection, login_manager # # users = {} # # # class User(UserMixin): # # def __init__(self, id_, name, email): # self.name = name # self.email = email # self.phone = None # self.group = None # self.id_ = None # ssh_pub_keys = [] # # @staticmethod # def authenticate(user_id, password): # ldap_connection = make_ldap_connection # result = ldap_connection.search_s(current_app.config['LDAP_BASE_DN'], ldap.SCOPE_SUBTREE, f'uid={user_id}')[0] # user = User.from_raw_ldap(result, password) # # try: # ldap_connection.bind_s(f'uid={user.id_},ou={user.group},{current_app.config["LDAP_BASE_DN"]}', password, ldap.AUTH_SIMPLE) # return user # except Exception as ex: # print(f'ERROR: {str(ex)}') # finally: # close_ldap_connection(ldap_connection) # # @staticmethod # def from_raw_ldap(raw_user, password=None): # raw_dn = raw_user[0] # raw_user_data = raw_user[1] # dn = parse_user_dn(raw_dn) # user = User() # user.id_ = raw_user_data.get('cn')[0].decode('utf-8') # user.group = dn.get('ou') # user.name = raw_user_data.get('displayName')[0].decode('utf-8') # user.phone = raw_user_data.get("mobile", [b''])[0].decode('utf-8') # user.ssh_pub_keys = [k.decode('utf-8') for k in raw_user_data.get('sshPublicKey', [])] # user.password = password # # if user.id_ not in users: # users[user.id_] = user # elif users[user.id_].password is None and password is not None: # users[user.id_] = user # # return user # # # def parse_user_dn(user_dn): # group = user_dn.split(',') # result = {} # for g in group: # split = g.split('=') # result[split[0]] = split[1] # return result # # # @login_manager.user_loader # def load_user(user_id): # ldap_connection = make_ldap_connection() # raw_user = ldap_connection.search_s(current_app.config['LDAP_BASE_DN'], ldap.SCOPE_SUBTREE, f'uid={user_id}')[0] # _users = users # user = User.from_raw_ldap(raw_user) # close_ldap_connection(ldap_connection) # return user