|
1 | 1 | import logging |
2 | 2 |
|
3 | | -try: |
4 | | - import ldap3 |
5 | | -except ModuleNotFoundError: |
6 | | - ldap3 = None |
7 | 3 |
|
8 | | -from ..config import get_current_config |
| 4 | +import ldap3 |
| 5 | + |
9 | 6 |
|
10 | 7 | logger = logging.getLogger(__name__) |
11 | 8 |
|
12 | 9 |
|
13 | 10 | class LdapManager: |
14 | | - @staticmethod |
15 | | - def try_auth(user, password): |
16 | | - config = get_current_config().LDAP |
17 | | - entrie = LdapManager.search_user(user, config["admin_filter"]) |
18 | | - if entrie: |
19 | | - logger.debug("{0} is admin".format(user)) |
20 | | - admin = True |
21 | | - else: |
22 | | - entrie = LdapManager.search_user(user, config["user_filter"]) |
| 11 | + |
| 12 | + def __init__(self, ldap_server, base_dn, user_filter, admin_filter, bind_dn, bind_password, username_attr, email_attr): |
| 13 | + self.ldap_server=ldap_server |
| 14 | + self.base_dn=base_dn |
| 15 | + self.user_filter=user_filter |
| 16 | + self.admin_filter=admin_filter |
| 17 | + self.bind_dn=bind_dn |
| 18 | + self.bind_password=bind_password |
| 19 | + self.username_attr=username_attr |
| 20 | + self.email_attr=email_attr |
| 21 | + if not self.ldap_server: |
| 22 | + raise ValueError("No LDAP configured") |
| 23 | + self.server = ldap3.Server(self.ldap_server, get_info="ALL") |
| 24 | + |
| 25 | + def try_auth(self,user, password): |
| 26 | + admin= False |
| 27 | + if self.admin_filter: |
| 28 | + entrie = self.search_user(user, self.admin_filter) |
23 | 29 | if entrie: |
24 | | - admin = False |
25 | | - else: |
| 30 | + logger.debug("{0} is admin".format(user)) |
| 31 | + admin = True |
| 32 | + if not admin: |
| 33 | + entrie = self.search_user(user, self.user_filter) |
| 34 | + if not entrie: |
26 | 35 | return False |
27 | | - server = ldap3.Server(config["ldap_server"], get_info="ALL") |
28 | 36 | try: |
29 | 37 | with ldap3.Connection( |
30 | | - server, entrie.entry_dn, password, read_only=True |
| 38 | + self.server, entrie.entry_dn, password, read_only=True |
31 | 39 | ) as conn: |
32 | 40 | return { |
33 | | - "uid": entrie[config["username_attr"]], |
34 | | - "mail": entrie[config["email_attr"]], |
| 41 | + "uid": entrie[self.username_attr], |
| 42 | + "mail": entrie[self.email_attr], |
35 | 43 | "admin": admin, |
36 | 44 | } |
37 | 45 | except ldap3.core.exceptions.LDAPBindError: |
38 | 46 | logger.warning("wrong password for user {0}".format(user)) |
39 | 47 | return False |
40 | 48 |
|
41 | | - @staticmethod |
42 | | - def search_user(user, filter): |
43 | | - if not ldap3: |
44 | | - logger.warning("module 'ldap2' is not installed") |
45 | | - return False |
46 | | - config = get_current_config().LDAP |
47 | | - if not config["ldap_server"]: |
48 | | - logger.info("No LDAP configured") |
49 | | - return False |
50 | | - server = ldap3.Server(config["ldap_server"], get_info="ALL") |
| 49 | + def search_user(self,user, filter): |
| 50 | + |
51 | 51 | try: |
52 | 52 | with ldap3.Connection( |
53 | | - server, config["bind_dn"], config["bind_password"], read_only=True |
| 53 | + self.server, self.bind_dn, self.bind_password, read_only=True |
54 | 54 | ) as conn: |
55 | 55 | conn.search( |
56 | | - config["base_dn"], |
| 56 | + self.base_dn, |
57 | 57 | filter, |
58 | | - attributes=[config["email_attr"], config["username_attr"]], |
| 58 | + attributes=[self.email_attr, self.username_attr], |
59 | 59 | ) |
60 | 60 | entries = conn.entries |
61 | 61 | except ldap3.core.exceptions.LDAPBindError: |
62 | | - logger.warning("wrong can't bind LDAP with {-1}".format(config["bind_dn"])) |
| 62 | + logger.warning( |
| 63 | + "wrong can't bind LDAP with {0}".format(self.bind_dn)) |
63 | 64 |
|
64 | 65 | for entrie in entries: |
65 | | - if entrie[config["username_attr"]] == user: |
| 66 | + if entrie[self.username_attr] == user: |
66 | 67 | return entrie |
67 | 68 | return False |
0 commit comments