canaille-globuzma/canaille/models.py

210 lines
5.7 KiB
Python
Raw Normal View History

2021-12-20 22:57:27 +00:00
import ldap.filter
from flask import current_app
from flask import session
from .ldap_backend.ldapobject import LDAPObject
2020-08-14 11:18:08 +00:00
class User(LDAPObject):
2021-12-03 17:37:25 +00:00
DEFAULT_OBJECT_CLASS = "inetOrgPerson"
DEFAULT_FILTER = "(|(uid={login})(mail={login}))"
DEFAULT_ID_ATTRIBUTE = "cn"
2021-12-02 17:23:14 +00:00
def __init__(self, *args, **kwargs):
self.read = set()
self.write = set()
self.permissions = set()
2021-12-03 15:49:19 +00:00
self._groups = None
2021-12-02 17:23:14 +00:00
super().__init__(*args, **kwargs)
2020-08-19 14:20:57 +00:00
@classmethod
2020-10-21 15:15:33 +00:00
def get(cls, login=None, dn=None, filter=None, conn=None):
2020-08-19 14:20:57 +00:00
conn = conn or cls.ldap()
2020-10-21 15:15:33 +00:00
if login:
2021-12-06 14:40:30 +00:00
filter = (
current_app.config["LDAP"]
2021-12-03 17:37:25 +00:00
.get("USER_FILTER", User.DEFAULT_FILTER)
2021-12-06 14:40:30 +00:00
.format(login=ldap.filter.escape_filter_chars(login))
)
2020-10-21 15:15:33 +00:00
2020-08-19 14:20:57 +00:00
user = super().get(dn, filter, conn)
2021-06-03 13:00:11 +00:00
if user:
2021-12-02 17:23:14 +00:00
user.load_permissions(conn)
2021-06-03 13:00:11 +00:00
2020-08-19 14:20:57 +00:00
return user
2021-06-03 13:00:11 +00:00
def load_groups(self, conn=None):
try:
2021-12-03 17:37:25 +00:00
group_filter = (
current_app.config["LDAP"]
.get("GROUP_USER_FILTER", Group.DEFAULT_USER_FILTER)
.format(user=self)
)
2021-12-06 14:40:30 +00:00
escaped_group_filter = ldap.filter.escape_filter_chars(group_filter)
self._groups = Group.filter(filter=escaped_group_filter, conn=conn)
except KeyError:
pass
2021-06-03 13:00:11 +00:00
2020-08-20 08:45:33 +00:00
@classmethod
2020-08-21 08:23:39 +00:00
def authenticate(cls, login, password, signin=False):
2020-10-21 15:15:33 +00:00
user = User.get(login)
2020-08-20 08:45:33 +00:00
if not user or not user.check_password(password):
return None
2020-08-19 14:20:57 +00:00
2020-08-21 08:23:39 +00:00
if signin:
user.login()
2020-08-20 08:45:33 +00:00
return user
2020-08-14 11:18:08 +00:00
2020-08-21 08:23:39 +00:00
def login(self):
try:
2020-12-29 08:31:46 +00:00
previous = (
session["user_dn"]
if isinstance(session["user_dn"], list)
else [session["user_dn"]]
)
session["user_dn"] = previous + [self.dn]
except KeyError:
session["user_dn"] = [self.dn]
2020-08-21 08:23:39 +00:00
2020-12-29 08:31:46 +00:00
@classmethod
2020-08-21 08:23:39 +00:00
def logout(self):
try:
2020-12-29 08:31:46 +00:00
if isinstance(session["user_dn"], list):
session["user_dn"].pop()
else:
del session["user_dn"]
except (IndexError, KeyError):
2020-08-21 08:23:39 +00:00
pass
def has_password(self):
return bool(self.userPassword)
2020-08-14 11:18:08 +00:00
def check_password(self, password):
conn = ldap.initialize(current_app.config["LDAP"]["URI"])
if current_app.config["LDAP"].get("TIMEOUT"):
conn.set_option(
ldap.OPT_NETWORK_TIMEOUT, current_app.config["LDAP"]["TIMEOUT"]
)
try:
conn.simple_bind_s(self.dn, password)
return True
except ldap.INVALID_CREDENTIALS:
return False
finally:
conn.unbind_s()
2020-08-14 11:18:08 +00:00
2020-10-21 08:26:31 +00:00
def set_password(self, password, conn=None):
conn = conn or self.ldap()
try:
conn.passwd_s(
self.dn,
None,
password.encode("utf-8"),
2020-10-21 08:26:31 +00:00
)
except ldap.LDAPError:
return False
return True
2020-08-17 15:49:49 +00:00
@property
def name(self):
return self.cn[0]
2021-06-03 13:00:11 +00:00
@property
def groups(self):
2021-12-03 15:49:19 +00:00
if self._groups is None:
self.load_groups()
2021-06-03 13:00:11 +00:00
return self._groups
def set_groups(self, values):
2021-06-03 13:00:11 +00:00
before = self._groups
after = [v if isinstance(v, Group) else Group.get(dn=v) for v in values]
2021-06-03 13:00:11 +00:00
to_add = set(after) - set(before)
to_del = set(before) - set(after)
for group in to_add:
group.add_member(self)
2021-06-03 13:00:11 +00:00
for group in to_del:
group.remove_member(self)
2021-06-03 13:00:11 +00:00
self._groups = after
2021-12-02 17:23:14 +00:00
def load_permissions(self, conn=None):
conn = conn or self.ldap()
for access_group_name, details in current_app.config["ACL"].items():
if not details.get("FILTER") or (
self.dn
and conn.search_s(self.dn, ldap.SCOPE_SUBTREE, details["FILTER"])
):
self.permissions |= set(details.get("PERMISSIONS", []))
self.read |= set(details.get("READ", []))
self.write |= set(details.get("WRITE", []))
2021-12-08 17:06:50 +00:00
def can_read(self, field):
return field in self.read | self.write
2022-11-01 17:15:17 +00:00
def can_write(self, field):
2021-12-08 17:06:50 +00:00
return field in self.write
@property
def can_edit_self(self):
return "edit_self" in self.permissions
2021-12-06 23:07:32 +00:00
@property
def can_use_oidc(self):
return "use_oidc" in self.permissions
2021-12-02 17:23:14 +00:00
@property
def can_manage_users(self):
return "manage_users" in self.permissions
@property
def can_manage_groups(self):
return "manage_groups" in self.permissions
@property
def can_manage_oidc(self):
return "manage_oidc" in self.permissions
@property
def can_delete_account(self):
return "delete_account" in self.permissions
@property
def can_impersonate_users(self):
return "impersonate_users" in self.permissions
2021-06-03 13:00:11 +00:00
class Group(LDAPObject):
2021-12-03 17:37:25 +00:00
DEFAULT_OBJECT_CLASS = "groupOfNames"
DEFAULT_ID_ATTRIBUTE = "cn"
DEFAULT_NAME_ATTRIBUTE = "cn"
DEFAULT_USER_FILTER = "member={user.dn}"
2021-07-01 16:21:20 +00:00
@property
def name(self):
2021-12-03 17:37:25 +00:00
attribute = current_app.config["LDAP"].get(
"GROUP_NAME_ATTRIBUTE", Group.DEFAULT_NAME_ATTRIBUTE
)
2021-07-01 16:21:20 +00:00
return self[attribute][0]
2021-06-03 13:00:11 +00:00
def get_members(self, conn=None):
return [
User.get(dn=user_dn, conn=conn)
for user_dn in self.member
if User.get(dn=user_dn, conn=conn)
]
2021-06-03 13:00:11 +00:00
def add_member(self, user):
2021-06-03 13:00:11 +00:00
self.member = self.member + [user.dn]
self.save()
2021-06-03 13:00:11 +00:00
def remove_member(self, user):
2021-06-03 13:00:11 +00:00
self.member = [m for m in self.member if m != user.dn]
self.save()