From f33096211b4bfd682dd09f9145df325a51295b27 Mon Sep 17 00:00:00 2001
From: Julian Rother <julianr@fsmpi.rwth-aachen.de>
Date: Tue, 23 Feb 2021 01:17:57 +0100
Subject: [PATCH] Ported uffd code to new ldap mapper
---
ldap3_mapper_new/__init__.py | 3 ++-
ldap3_mapper_new/attribute.py | 6 ++---
ldap3_mapper_new/core.py | 26 +++++++++++---------
ldap3_mapper_new/dbutils.py | 14 ++++++-----
ldap3_mapper_new/model.py | 13 ++++++----
ldap3_mapper_new/relationship.py | 22 +++++++++++------
tests/test_mail.py | 4 +--
tests/test_mfa.py | 4 +--
tests/test_oauth2.py | 4 +--
tests/test_role.py | 2 +-
tests/test_selfservice.py | 22 ++++++++---------
tests/test_user.py | 42 +++++++++++++++++---------------
uffd/ldap.py | 11 ++++-----
uffd/mail/models.py | 6 ++---
uffd/mail/views.py | 8 +++---
uffd/mfa/models.py | 2 +-
uffd/mfa/views.py | 2 +-
uffd/oauth2/models.py | 4 +--
uffd/role/models.py | 6 ++---
uffd/role/views.py | 4 +--
uffd/selfservice/views.py | 8 +++---
uffd/session/views.py | 7 +++---
uffd/user/models.py | 30 +++++++++++------------
uffd/user/views_group.py | 4 +--
uffd/user/views_user.py | 10 ++++----
25 files changed, 140 insertions(+), 124 deletions(-)
diff --git a/ldap3_mapper_new/__init__.py b/ldap3_mapper_new/__init__.py
index 4eb8c8f4..3c0730e8 100644
--- a/ldap3_mapper_new/__init__.py
+++ b/ldap3_mapper_new/__init__.py
@@ -1,8 +1,9 @@
import ldap3
+from .core import LDAPCommitError
from . import model, attribute, relationship
-__all__ = ['LDAPMapper']
+__all__ = ['LDAPMapper', 'LDAPCommitError']
class LDAPMapper:
def __init__(self, server=None, bind_dn=None, bind_password=None):
diff --git a/ldap3_mapper_new/attribute.py b/ldap3_mapper_new/attribute.py
index 9e984025..bd776313 100644
--- a/ldap3_mapper_new/attribute.py
+++ b/ldap3_mapper_new/attribute.py
@@ -46,7 +46,7 @@ class Attribute:
def add_hook(self, obj):
if obj.ldap_object.getattr(self.name) == []:
- self.__set__(self.name, self.default() if callable(self.default) else self.default)
+ self.__set__(obj, self.default() if callable(self.default) else self.default)
def __set_name__(self, cls, name):
if self.default is not None:
@@ -57,10 +57,10 @@ class Attribute:
return self
if self.multi:
return AttributeList(obj.ldap_object, self.name, self.aliases)
- return obj.ldap_object.getattr(self.name)
+ return (obj.ldap_object.getattr(self.name) or [None])[0]
def __set__(self, obj, values):
if not self.multi:
values = [values]
- for name in self.aliases:
+ for name in [self.name] + self.aliases:
obj.ldap_object.setattr(name, values)
diff --git a/ldap3_mapper_new/core.py b/ldap3_mapper_new/core.py
index 11399220..c5fcff11 100644
--- a/ldap3_mapper_new/core.py
+++ b/ldap3_mapper_new/core.py
@@ -1,5 +1,3 @@
-from copy import deepcopy
-
from ldap3 import MODIFY_REPLACE, MODIFY_DELETE, MODIFY_ADD, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES
from ldap3.utils.conv import escape_filter_chars
@@ -10,9 +8,9 @@ def match_dn(dn, base):
return dn.endswith(base) # Probably good enougth for all valid dns
def make_cache_key(search_base, filter_params):
- res = [search_base]
+ res = (search_base,)
for attr, value in sorted(filter_params):
- res.append((attr, value))
+ res += ((attr, value),)
return res
class LDAPCommitError(Exception):
@@ -25,7 +23,10 @@ class SessionState:
self.references = references or {} # {(attr_name, value): {srcobj, ...}, ...}
def copy(self):
- return SessionState(deepcopy(self.objects), deepcopy(self.deleted_objects), deepcopy(self.references))
+ objects = self.objects.copy()
+ deleted_objects = self.deleted_objects.copy()
+ references = {key: objs.copy() for key, objs in self.references.items()}
+ return SessionState(objects, deleted_objects, references)
def ref(self, obj, attr, values):
for value in values:
@@ -46,18 +47,19 @@ class ObjectState:
self.dn = dn
def copy(self):
- return ObjectState(attributes=deepcopy(self.attributes), dn=self.dn, session=self.session)
+ attributes = {name: values.copy() for name, values in self.attributes.items()}
+ return ObjectState(attributes=attributes, dn=self.dn, session=self.session)
class AddOperation:
def __init__(self, obj, dn, object_classes):
self.obj = obj
self.dn = dn
self.object_classes = object_classes
- self.attributes = deepcopy(obj.state.attributes)
+ self.attributes = {name: values.copy() for name, values in obj.state.attributes.items()}
def apply_object(self, obj_state):
obj_state.dn = self.dn
- obj_state.attributes = deepcopy(self.attributes)
+ obj_state.attributes = {name: values.copy() for name, values in self.attributes.items()}
def apply_session(self, session_state):
assert self.dn not in session_state.objects
@@ -74,7 +76,7 @@ class DeleteOperation:
def __init__(self, obj):
self.dn = obj.state.dn
self.obj = obj
- self.attributes = deepcopy(obj.state.attributes)
+ self.attributes = {name: values.copy() for name, values in obj.state.attributes.items()}
def apply_object(self, obj_state):
obj_state.dn = None
@@ -94,8 +96,8 @@ class DeleteOperation:
class ModifyOperation:
def __init__(self, obj, changes):
self.obj = obj
- self.attributes = deepcopy(obj.state.attributes)
- self.changes = deepcopy(changes)
+ self.attributes = {name: values.copy() for name, values in obj.state.attributes.items()}
+ self.changes = changes
def apply_object(self, obj_state):
for attr, changes in self.changes.items():
@@ -181,7 +183,7 @@ class Session:
def get(self, dn, filter_params):
if dn in self.state.objects:
obj = self.state.objects[dn]
- return obj if obj.matches(filter_params) else None
+ return obj if obj.match(filter_params) else None
if dn in self.state.deleted_objects:
return None
conn = self.get_connection()
diff --git a/ldap3_mapper_new/dbutils.py b/ldap3_mapper_new/dbutils.py
index 21a62321..42fccb96 100644
--- a/ldap3_mapper_new/dbutils.py
+++ b/ldap3_mapper_new/dbutils.py
@@ -3,10 +3,11 @@ from collections.abc import MutableSet
from .model import add_to_session
class DBRelationshipSet(MutableSet):
- def __init__(self, dbobj, relattr, ldapcls):
+ def __init__(self, dbobj, relattr, ldapcls, mapcls):
self.__dbobj = dbobj
self.__relattr = relattr
self.__ldapcls = ldapcls
+ self.__mapcls = mapcls
def __get_dns(self):
return [mapobj.dn for mapobj in getattr(self.__dbobj, self.__relattr)]
@@ -28,10 +29,10 @@ class DBRelationshipSet(MutableSet):
def add(self, value):
if not isinstance(value, self.__ldapcls):
raise TypeError()
- if value.ldap_object.session is not None:
- add_to_session(value, self.__ldapcls.ldap_mapper.session)
+ if value.ldap_object.session is None:
+ add_to_session(value, self.__ldapcls.ldap_mapper.session.ldap_session)
if value.ldap_object.dn not in self.__get_dns():
- getattr(self.__dbobj, self.__relattr).append(self.__ldapcls(dn=value.ldap_object.dn))
+ getattr(self.__dbobj, self.__relattr).append(self.__mapcls(dn=value.ldap_object.dn))
def discard(self, value):
if not isinstance(value, self.__ldapcls):
@@ -57,7 +58,7 @@ class DBRelationship:
def __get__(self, obj, objtype=None):
if obj is None:
return self
- return DBRelationshipSet(obj, self.relattr, self.ldapcls)
+ return DBRelationshipSet(obj, self.relattr, self.ldapcls, self.mapcls)
def __set__(self, obj, values):
tmp = self.__get__(obj)
@@ -69,7 +70,7 @@ class DBBackreferenceSet(MutableSet):
def __init__(self, ldapobj, dbcls, relattr, mapcls, backattr):
self.__ldapobj = ldapobj
self.__dbcls = dbcls
- self.__relattr, = relattr
+ self.__relattr = relattr
self.__mapcls = mapcls
self.__backattr = backattr
@@ -94,6 +95,7 @@ class DBBackreferenceSet(MutableSet):
def add(self, value):
# TODO: add value to db session if necessary
+ assert self.__ldapobj.ldap_object.session is not None
if not isinstance(value, self.__dbcls):
raise TypeError()
rel = getattr(value, self.__relattr)
diff --git a/ldap3_mapper_new/model.py b/ldap3_mapper_new/model.py
index 4a0c20d7..0dc9f240 100644
--- a/ldap3_mapper_new/model.py
+++ b/ldap3_mapper_new/model.py
@@ -68,7 +68,8 @@ class ModelQuery:
return make_modelobjs(objs, self.model)
def filter_by(self, **kwargs):
- filter_params = self.model.ldap_filter_params + list(kwargs.items())
+ filter_params = self.model.ldap_filter_params
+ filter_params += tuple((getattr(self.model, attr).name, value) for attr, value in kwargs.items())
session = self.model.ldap_mapper.session.ldap_session
objs = session.filter(self.model.ldap_search_base, filter_params)
return make_modelobjs(objs, self.model)
@@ -81,12 +82,12 @@ class Model:
# Overwritten by mapper
ldap_mapper = None
query = ModelQueryWrapper()
- ldap_add_hooks = tuple()
+ ldap_add_hooks = ()
# Overwritten by models
ldap_search_base = None
- ldap_filter_params = None
- ldap_object_classes = None
+ ldap_filter_params = ()
+ ldap_object_classes = ()
ldap_dn_base = None
ldap_dn_attribute = None
@@ -106,7 +107,9 @@ class Model:
values = self.ldap_object.getattr(self.ldap_dn_attribute)
if not values:
return None
- return '%s=%s,%s'%(self.ldap_dn_attribute, escape_rdn(values[0]), self.ldap_dn_base)
+ # escape_rdn can't handle empty strings
+ rdn = escape_rdn(values[0]) if values[0] else ''
+ return '%s=%s,%s'%(self.ldap_dn_attribute, rdn, self.ldap_dn_base)
def __repr__(self):
cls_name = '%s.%s'%(type(self).__module__, type(self).__name__)
diff --git a/ldap3_mapper_new/relationship.py b/ldap3_mapper_new/relationship.py
index 06986266..d8e256aa 100644
--- a/ldap3_mapper_new/relationship.py
+++ b/ldap3_mapper_new/relationship.py
@@ -40,11 +40,15 @@ class RelationshipSet(MutableSet):
if value.ldap_object.session is None:
add_to_session(value, self.__ldap_object.session)
assert value.ldap_object.session == self.__ldap_object.session
- self.__ldap_object.attradd(self.__name, value.dn)
+ self.__ldap_object.attr_append(self.__name, value.dn)
def discard(self, value):
self.__modify_check(value)
- self.__ldap_object.attrdel(self.__name, value.dn)
+ self.__ldap_object.attr_remove(self.__name, value.dn)
+
+ def update(self, values):
+ for value in values:
+ self.add(value)
class Relationship:
def __init__(self, name, destmodel, backref=None):
@@ -59,7 +63,7 @@ class Relationship:
def __get__(self, obj, objtype=None):
if obj is None:
return self
- return RelationshipSet(obj, self.name, type(obj), self.destmodel)
+ return RelationshipSet(obj.ldap_object, self.name, type(obj), self.destmodel)
def __set__(self, obj, values):
tmp = self.__get__(obj)
@@ -83,7 +87,7 @@ class BackreferenceSet(MutableSet):
def __get(self):
if self.__ldap_object.session is None:
return set()
- filter_params = self.__srcmodel.filter_params + [(self.__name, self.__ldap_object.dn)]
+ filter_params = self.__srcmodel.ldap_filter_params + ((self.__name, self.__ldap_object.dn),)
objs = self.__ldap_object.session.filter(self.__srcmodel.ldap_search_base, filter_params)
return set(make_modelobjs(objs, self.__srcmodel))
@@ -105,11 +109,15 @@ class BackreferenceSet(MutableSet):
add_to_session(value, self.__ldap_object.session)
assert value.ldap_object.session == self.__ldap_object.session
if self.__ldap_object.dn not in value.ldap_object.getattr(self.__name):
- value.ldap_object.attradd(self.__name, self.__ldap_object.dn)
+ value.ldap_object.attr_append(self.__name, self.__ldap_object.dn)
def discard(self, value):
self.__modify_check(value)
- value.ldap_object.attrdel(self.__name, self.__ldap_object.dn)
+ value.ldap_object.attr_remove(self.__name, self.__ldap_object.dn)
+
+ def update(self, values):
+ for value in values:
+ self.add(value)
class Backreference:
def __init__(self, name, srcmodel):
@@ -119,7 +127,7 @@ class Backreference:
def __get__(self, obj, objtype=None):
if obj is None:
return self
- return BackreferenceSet(obj, self.name, type(obj), self.srcmodel)
+ return BackreferenceSet(obj.ldap_object, self.name, type(obj), self.srcmodel)
def __set__(self, obj, values):
tmp = self.__get__(obj)
diff --git a/tests/test_mail.py b/tests/test_mail.py
index 5d2f35d7..a5b9f88f 100644
--- a/tests/test_mail.py
+++ b/tests/test_mail.py
@@ -14,7 +14,7 @@ from uffd import create_app, db
from utils import dump, UffdTestCase
def get_mail():
- return Mail.ldap_get('uid=test,ou=postfix,dc=example,dc=com')
+ return Mail.query.get('uid=test,ou=postfix,dc=example,dc=com')
class TestMailViews(UffdTestCase):
def setUp(self):
@@ -68,7 +68,7 @@ class TestMailViews(UffdTestCase):
'mail-destinations': 'testuser@mail.example.com\ntestadmin@mail.example.com'}, follow_redirects=True)
dump('mail_create', r)
self.assertEqual(r.status_code, 200)
- m = Mail.ldap_get('uid=test1,ou=postfix,dc=example,dc=com')
+ m = Mail.query.get('uid=test1,ou=postfix,dc=example,dc=com')
self.assertEqual(m.uid, 'test1')
self.assertEqual(sorted(m.receivers), ['foo@bar.com', 'test@bar.com'])
self.assertEqual(sorted(m.destinations), ['testadmin@mail.example.com', 'testuser@mail.example.com'])
diff --git a/tests/test_mfa.py b/tests/test_mfa.py
index b5bd2a51..4ea98102 100644
--- a/tests/test_mfa.py
+++ b/tests/test_mfa.py
@@ -25,10 +25,10 @@ class TestMfaPrimitives(unittest.TestCase):
self.assertEqual(_hotp(2**64-1, b'abcde'), '899292')
def get_user():
- return User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testuser,ou=users,dc=example,dc=com')
def get_admin():
- return User.ldap_get('uid=testadmin,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
def get_fido2_test_cred():
try:
diff --git a/tests/test_oauth2.py b/tests/test_oauth2.py
index e68795ee..1dc7e1c5 100644
--- a/tests/test_oauth2.py
+++ b/tests/test_oauth2.py
@@ -14,10 +14,10 @@ from uffd import create_app, db, ldap
from utils import dump, UffdTestCase
def get_user():
- return User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testuser,ou=users,dc=example,dc=com')
def get_admin():
- return User.ldap_get('uid=testadmin,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
class TestOAuth2Client(UffdTestCase):
def setUpApp(self):
diff --git a/tests/test_role.py b/tests/test_role.py
index 2f4e0601..c6a42bdc 100644
--- a/tests/test_role.py
+++ b/tests/test_role.py
@@ -48,7 +48,7 @@ class TestRoleViews(UffdTestCase):
role = Role(name='base', description='Base role description')
db.session.add(role)
db.session.commit()
- role.groups.add(Group.ldap_get('cn=uffd_admin,ou=groups,dc=example,dc=com'))
+ role.groups.add(Group.query.get('cn=uffd_admin,ou=groups,dc=example,dc=com'))
db.session.commit()
self.assertEqual(role.name, 'base')
self.assertEqual(role.description, 'Base role description')
diff --git a/tests/test_selfservice.py b/tests/test_selfservice.py
index f46b4c1e..851f2d7b 100644
--- a/tests/test_selfservice.py
+++ b/tests/test_selfservice.py
@@ -14,7 +14,7 @@ from uffd import create_app, db, ldap
from utils import dump, UffdTestCase
def get_ldap_password():
- return User.ldap_get('uid=testuser,ou=users,dc=example,dc=com').pwhash
+ return User.query.get('uid=testuser,ou=users,dc=example,dc=com').pwhash
class TestSelfservice(UffdTestCase):
def setUpApp(self):
@@ -111,7 +111,7 @@ class TestSelfservice(UffdTestCase):
def test_token_mail_wrong_user(self):
self.login()
user = get_current_user()
- admin_user = User.ldap_get('uid=testadmin,ou=users,dc=example,dc=com')
+ admin_user = User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
db.session.add(MailToken(loginname=user.loginname, newmail='newusermail@example.com'))
admin_token = MailToken(loginname='testadmin', newmail='newadminmail@example.com')
db.session.add(admin_token)
@@ -120,7 +120,7 @@ class TestSelfservice(UffdTestCase):
dump('token_mail_wrong_user', r)
self.assertEqual(r.status_code, 200)
_user = get_current_user()
- _admin_user = User.ldap_get('uid=testadmin,ou=users,dc=example,dc=com')
+ _admin_user = User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
self.assertEqual(_user.mail, user.mail)
self.assertEqual(_admin_user.mail, admin_user.mail)
@@ -140,7 +140,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(len(tokens), 0)
def test_forgot_password(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
r = self.client.get(path=url_for('selfservice.forgot_password'))
dump('forgot_password', r)
self.assertEqual(r.status_code, 200)
@@ -153,7 +153,7 @@ class TestSelfservice(UffdTestCase):
self.assertIn(token.token, str(self.app.last_mail.get_content()))
def test_forgot_password_wrong_user(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
r = self.client.get(path=url_for('selfservice.forgot_password'))
self.assertEqual(r.status_code, 200)
r = self.client.post(path=url_for('selfservice.forgot_password'),
@@ -164,7 +164,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(len(PasswordToken.query.all()), 0)
def test_forgot_password_wrong_email(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
r = self.client.get(path=url_for('selfservice.forgot_password'), follow_redirects=True)
self.assertEqual(r.status_code, 200)
r = self.client.post(path=url_for('selfservice.forgot_password'),
@@ -184,7 +184,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(len(PasswordToken.query.all()), 0)
def test_token_password(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
oldpw = get_ldap_password()
token = PasswordToken(loginname=user.loginname)
db.session.add(token)
@@ -201,7 +201,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(len(PasswordToken.query.all()), 0)
def test_token_password_emptydb(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
oldpw = get_ldap_password()
r = self.client.get(path=url_for('selfservice.token_password', token='A'*128), follow_redirects=True)
dump('token_password_emptydb', r)
@@ -215,7 +215,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(oldpw, get_ldap_password())
def test_token_password_invalid(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
oldpw = get_ldap_password()
token = PasswordToken(loginname=user.loginname)
db.session.add(token)
@@ -232,7 +232,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(oldpw, get_ldap_password())
def test_token_password_expired(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
oldpw = get_ldap_password()
token = PasswordToken(loginname=user.loginname,
created=(datetime.datetime.now() - datetime.timedelta(days=10)))
@@ -250,7 +250,7 @@ class TestSelfservice(UffdTestCase):
self.assertEqual(oldpw, get_ldap_password())
def test_token_password_different_passwords(self):
- user = User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=testuser,ou=users,dc=example,dc=com')
oldpw = get_ldap_password()
token = PasswordToken(loginname=user.loginname)
db.session.add(token)
diff --git a/tests/test_user.py b/tests/test_user.py
index 5bc1e313..63e429f7 100644
--- a/tests/test_user.py
+++ b/tests/test_user.py
@@ -16,13 +16,13 @@ from uffd import create_app, db
from utils import dump, UffdTestCase
def get_user():
- return User.ldap_get('uid=testuser,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testuser,ou=users,dc=example,dc=com')
def get_user_password():
return get_user().pwhash
def get_admin():
- return User.ldap_get('uid=testadmin,ou=users,dc=example,dc=com')
+ return User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
class TestUserModel(UffdTestCase):
def test_has_permission(self):
@@ -74,19 +74,21 @@ class TestUserViews(UffdTestCase):
r = self.client.get(path=url_for('user.show'), follow_redirects=True)
dump('user_new', r)
self.assertEqual(r.status_code, 200)
- self.assertIsNone(User.ldap_get('uid=newuser,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
r = self.client.post(path=url_for('user.update'),
data={'loginname': 'newuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
f'role-{role1_id}': '1', 'password': 'newpassword'}, follow_redirects=True)
dump('user_new_submit', r)
self.assertEqual(r.status_code, 200)
- user = User.ldap_get('uid=newuser,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser')
self.assertEqual(user.displayname, 'New User')
self.assertEqual(user.mail, 'newuser@example.com')
self.assertTrue(user.uid)
+ role1 = Role(name='role1')
+ print('test_new', role1.db_members, role1.members, user.roles)
self.assertEqual(roles, ['base', 'role1'])
# TODO: check password hash
@@ -96,7 +98,7 @@ class TestUserViews(UffdTestCase):
'password': 'newpassword'}, follow_redirects=True)
dump('user_new_invalid_loginname', r)
self.assertEqual(r.status_code, 200)
- self.assertIsNone(User.ldap_get('uid=newuser,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
def test_new_empty_loginname(self):
r = self.client.post(path=url_for('user.update'),
@@ -104,7 +106,7 @@ class TestUserViews(UffdTestCase):
'password': 'newpassword'}, follow_redirects=True)
dump('user_new_empty_loginname', r)
self.assertEqual(r.status_code, 200)
- self.assertIsNone(User.ldap_get('uid=newuser,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
def test_new_empty_email(self):
r = self.client.post(path=url_for('user.update'),
@@ -112,7 +114,7 @@ class TestUserViews(UffdTestCase):
'password': 'newpassword'}, follow_redirects=True)
dump('user_new_empty_email', r)
self.assertEqual(r.status_code, 200)
- self.assertIsNone(User.ldap_get('uid=newuser,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
def test_new_invalid_display_name(self):
r = self.client.post(path=url_for('user.update'),
@@ -120,7 +122,7 @@ class TestUserViews(UffdTestCase):
'password': 'newpassword'}, follow_redirects=True)
dump('user_new_invalid_display_name', r)
self.assertEqual(r.status_code, 200)
- self.assertIsNone(User.ldap_get('uid=newuser,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
def test_update(self):
user = get_user()
@@ -255,59 +257,59 @@ newuser12,newuser12@example.com,{role1.id};{role1.id}
r = self.client.post(path=url_for('user.csvimport'), data={'csv': data}, follow_redirects=True)
dump('user_csvimport', r)
self.assertEqual(r.status_code, 200)
- user = User.ldap_get('uid=newuser1,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser1,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser1')
self.assertEqual(user.displayname, 'newuser1')
self.assertEqual(user.mail, 'newuser1@example.com')
self.assertEqual(roles, ['base'])
- user = User.ldap_get('uid=newuser2,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser2,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser2')
self.assertEqual(user.displayname, 'newuser2')
self.assertEqual(user.mail, 'newuser2@example.com')
self.assertEqual(roles, ['base', 'role1'])
- user = User.ldap_get('uid=newuser3,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser3,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser3')
self.assertEqual(user.displayname, 'newuser3')
self.assertEqual(user.mail, 'newuser3@example.com')
self.assertEqual(roles, ['base', 'role1', 'role2'])
- user = User.ldap_get('uid=newuser4,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser4,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser4')
self.assertEqual(user.displayname, 'newuser4')
self.assertEqual(user.mail, 'newuser4@example.com')
self.assertEqual(roles, ['base'])
- user = User.ldap_get('uid=newuser5,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser5,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser5')
self.assertEqual(user.displayname, 'newuser5')
self.assertEqual(user.mail, 'newuser5@example.com')
self.assertEqual(roles, ['base'])
- user = User.ldap_get('uid=newuser6,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser6,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser6')
self.assertEqual(user.displayname, 'newuser6')
self.assertEqual(user.mail, 'newuser6@example.com')
self.assertEqual(roles, ['base', 'role1', 'role2'])
- self.assertIsNone(User.ldap_get('uid=newuser7,ou=users,dc=example,dc=com'))
- self.assertIsNone(User.ldap_get('uid=newuser8,ou=users,dc=example,dc=com'))
- self.assertIsNone(User.ldap_get('uid=newuser9,ou=users,dc=example,dc=com'))
- user = User.ldap_get('uid=newuser10,ou=users,dc=example,dc=com')
+ self.assertIsNone(User.query.get('uid=newuser7,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser8,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.query.get('uid=newuser9,ou=users,dc=example,dc=com'))
+ user = User.query.get('uid=newuser10,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser10')
self.assertEqual(user.displayname, 'newuser10')
self.assertEqual(user.mail, 'newuser10@example.com')
self.assertEqual(roles, ['base'])
- user = User.ldap_get('uid=newuser11,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser11,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser11')
@@ -316,7 +318,7 @@ newuser12,newuser12@example.com,{role1.id};{role1.id}
# Currently the csv import is not very robust, imho newuser11 should have role1 and role2!
#self.assertEqual(roles, ['base', 'role1', 'role2'])
self.assertEqual(roles, ['base', 'role2'])
- user = User.ldap_get('uid=newuser12,ou=users,dc=example,dc=com')
+ user = User.query.get('uid=newuser12,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
self.assertEqual(user.loginname, 'newuser12')
diff --git a/uffd/ldap.py b/uffd/ldap.py
index 2365a81e..0dcd9fdf 100644
--- a/uffd/ldap.py
+++ b/uffd/ldap.py
@@ -2,20 +2,19 @@ from flask import current_app, request
import ldap3
-from ldap3_mapper import LDAP3Mapper, LDAPCommitError # pylint: disable=unused-import
-from ldap3_mapper.base import Session
+from ldap3_mapper_new import LDAPMapper, LDAPCommitError # pylint: disable=unused-import
-class FlaskLDAP3Mapper(LDAP3Mapper):
+class FlaskLDAPMapper(LDAPMapper):
def __init__(self):
super().__init__()
@property
def session(self):
if not hasattr(request, 'ldap_session'):
- request.ldap_session = Session()
+ request.ldap_session = self.Session(self.get_connection)
return request.ldap_session
- def connect(self):
+ def get_connection(self):
if current_app.config.get('LDAP_SERVICE_MOCK', False):
if not current_app.debug:
raise Exception('LDAP_SERVICE_MOCK cannot be enabled on production instances')
@@ -32,4 +31,4 @@ class FlaskLDAP3Mapper(LDAP3Mapper):
return ldap3.Connection(server, current_app.config["LDAP_SERVICE_BIND_DN"],
current_app.config["LDAP_SERVICE_BIND_PASSWORD"], auto_bind=True)
-ldap = FlaskLDAP3Mapper()
+ldap = FlaskLDAPMapper()
diff --git a/uffd/mail/models.py b/uffd/mail/models.py
index 22a7e0b2..322af9d4 100644
--- a/uffd/mail/models.py
+++ b/uffd/mail/models.py
@@ -2,11 +2,11 @@ from uffd.ldap import ldap
from uffd.lazyconfig import lazyconfig_str, lazyconfig_list
class Mail(ldap.Model):
- ldap_base = lazyconfig_str('LDAP_BASE_MAIL')
+ ldap_search_base = lazyconfig_str('LDAP_BASE_MAIL')
+ ldap_filter_params = (('objectClass', 'postfixVirtual'),)
+ ldap_object_classes = lazyconfig_list('MAIL_LDAP_OBJECTCLASSES')
ldap_dn_attribute = 'uid'
ldap_dn_base = lazyconfig_str('LDAP_BASE_MAIL')
- ldap_filter = '(objectClass=postfixVirtual)'
- ldap_object_classes = lazyconfig_list('MAIL_LDAP_OBJECTCLASSES')
uid = ldap.Attribute('uid')
receivers = ldap.Attribute('mailacceptinggeneralid', multi=True)
diff --git a/uffd/mail/views.py b/uffd/mail/views.py
index db401a3a..d7b68782 100644
--- a/uffd/mail/views.py
+++ b/uffd/mail/views.py
@@ -21,14 +21,14 @@ def mail_acl_check():
@bp.route("/")
@register_navbar('Mail', icon='envelope', blueprint=bp, visible=mail_acl_check)
def index():
- return render_template('mail_list.html', mails=Mail.ldap_all())
+ return render_template('mail_list.html', mails=Mail.query.all())
@bp.route("/<uid>")
@bp.route("/new")
def show(uid=None):
mail = Mail()
if uid is not None:
- mail = Mail.ldap_filter_by(uid=uid)[0]
+ mail = Mail.query.filter_by(uid=uid)[0]
return render_template('mail.html', mail=mail)
@bp.route("/<uid>/update", methods=['POST'])
@@ -36,7 +36,7 @@ def show(uid=None):
@csrf_protect(blueprint=bp)
def update(uid=None):
if uid is not None:
- mail = Mail.ldap_filter_by(uid=uid)[0]
+ mail = Mail.query.filter_by(uid=uid)[0]
else:
mail = Mail(uid=request.form.get('mail-uid'))
mail.receivers = request.form.get('mail-receivers', '').splitlines()
@@ -49,7 +49,7 @@ def update(uid=None):
@bp.route("/<uid>/del")
@csrf_protect(blueprint=bp)
def delete(uid):
- mail = Mail.ldap_filter_by(uid=uid)[0]
+ mail = Mail.query.filter_by(uid=uid)[0]
ldap.session.delete(mail)
ldap.session.commit()
flash('Deleted mail mapping.')
diff --git a/uffd/mfa/models.py b/uffd/mfa/models.py
index 4c20d3c8..92d20cfb 100644
--- a/uffd/mfa/models.py
+++ b/uffd/mfa/models.py
@@ -41,7 +41,7 @@ class MFAMethod(db.Model):
@property
def user(self):
- return User.ldap_get(self.dn)
+ return User.query.get(self.dn)
@user.setter
def user(self, new_user):
diff --git a/uffd/mfa/views.py b/uffd/mfa/views.py
index 50b6ebce..3ce9a9f0 100644
--- a/uffd/mfa/views.py
+++ b/uffd/mfa/views.py
@@ -46,7 +46,7 @@ def admin_disable(uid):
if not get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP']):
flash('Access denied')
return redirect(url_for('index'))
- user = User.ldap_filter_by(uid=uid)[0]
+ user = User.query.filter_by(uid=uid)[0]
MFAMethod.query.filter_by(dn=user.dn).delete()
db.session.commit()
flash('Two-factor authentication was reset')
diff --git a/uffd/oauth2/models.py b/uffd/oauth2/models.py
index 3afcb43c..2900367c 100644
--- a/uffd/oauth2/models.py
+++ b/uffd/oauth2/models.py
@@ -39,7 +39,7 @@ class OAuth2Grant(db.Model):
@property
def user(self):
- return User.ldap_get(self.user_dn)
+ return User.query.get(self.user_dn)
@user.setter
def user(self, newuser):
@@ -78,7 +78,7 @@ class OAuth2Token(db.Model):
user_dn = Column(String(128))
@property
def user(self):
- return User.ldap_get(self.user_dn)
+ return User.query.get(self.user_dn)
@user.setter
def user(self, newuser):
diff --git a/uffd/role/models.py b/uffd/role/models.py
index 44512b55..996b6b3f 100644
--- a/uffd/role/models.py
+++ b/uffd/role/models.py
@@ -2,7 +2,7 @@ from sqlalchemy import Column, String, Integer, Text, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declared_attr
-from ldap3_mapper.db_relation import DB2LDAPRelation
+from ldap3_mapper_new.dbutils import DBRelationship
from uffd.database import db
from uffd.user.models import User, Group
@@ -37,10 +37,10 @@ class Role(db.Model):
description = Column(Text(), default='')
db_members = relationship("RoleUser", backref="role", cascade="all, delete-orphan")
- members = DB2LDAPRelation('db_members', RoleUser, User, backattr='role', backref='roles')
+ members = DBRelationship('db_members', User, RoleUser, backattr='role', backref='roles')
db_groups = relationship("RoleGroup", backref="role", cascade="all, delete-orphan")
- groups = DB2LDAPRelation('db_groups', RoleGroup, Group, backattr='role', backref='roles')
+ groups = DBRelationship('db_groups', Group, RoleGroup, backattr='role', backref='roles')
def update_member_groups(self):
for user in self.members:
diff --git a/uffd/role/views.py b/uffd/role/views.py
index 7b16b024..e915ce5a 100644
--- a/uffd/role/views.py
+++ b/uffd/role/views.py
@@ -31,7 +31,7 @@ def show(roleid=False):
role = Role()
else:
role = Role.query.filter_by(id=roleid).one()
- return render_template('role.html', role=role, groups=Group.ldap_all())
+ return render_template('role.html', role=role, groups=Group.query.all())
@bp.route("/<int:roleid>/update", methods=['POST'])
@bp.route("/new", methods=['POST'])
@@ -45,7 +45,7 @@ def update(roleid=False):
role = Role.query.filter_by(id=roleid).one()
role.name = request.values['name']
role.description = request.values['description']
- for group in Group.ldap_all():
+ for group in Group.query.all():
if request.values.get('group-{}'.format(group.gid), False):
role.groups.add(group)
else:
diff --git a/uffd/selfservice/views.py b/uffd/selfservice/views.py
index c866d5ec..4ba81514 100644
--- a/uffd/selfservice/views.py
+++ b/uffd/selfservice/views.py
@@ -67,7 +67,7 @@ def forgot_password():
reset_ratelimit.log(loginname+'/'+mail)
host_ratelimit.log()
flash("We sent a mail to this users mail address if you entered the correct mail and login name combination")
- user = (User.ldap_filter_by(loginname=loginname) or [None])[0]
+ user = (User.query.filter_by(loginname=loginname) or [None])[0]
if user and user.mail == mail:
send_passwordreset(user)
return redirect(url_for('session.login'))
@@ -89,7 +89,7 @@ def token_password(token):
if not request.values['password1'] == request.values['password2']:
flash('Passwords do not match, please try again.')
return render_template('set_password.html', token=token)
- user = User.ldap_filter_by(loginname=dbtoken.loginname)[0]
+ user = User.query.filter_by(loginname=dbtoken.loginname)[0]
if not user.set_password(request.values['password1']):
flash('Password ist not valid, please try again.')
return render_template('set_password.html', token=token)
@@ -110,7 +110,7 @@ def token_mail(token):
db.session.commit()
return redirect(url_for('selfservice.index'))
- user = User.ldap_filter_by(loginname=dbtoken.loginname)[0]
+ user = User.query.filter_by(loginname=dbtoken.loginname)[0]
user.set_mail(dbtoken.newmail)
flash('New mail set')
db.session.delete(dbtoken)
@@ -129,7 +129,7 @@ def send_mail_verification(loginname, newmail):
db.session.add(token)
db.session.commit()
- user = User.ldap_filter_by(loginname=loginname)[0]
+ user = User.query.filter_by(loginname=loginname)[0]
msg = EmailMessage()
msg.set_content(render_template('mailverification.mail.txt', user=user, token=token.token))
diff --git a/uffd/session/views.py b/uffd/session/views.py
index cb28ee57..cad80b2d 100644
--- a/uffd/session/views.py
+++ b/uffd/session/views.py
@@ -18,7 +18,7 @@ login_ratelimit = Ratelimit('login', 1*60, 3)
def login_get_user(loginname, password):
dn = User(loginname=loginname).dn
if current_app.config.get('LDAP_SERVICE_MOCK', False):
- conn = ldap.connect()
+ conn = ldap.get_connection()
# Since we reuse the same conn for all calls to `user_conn()` we
# simulate the password check by rebinding. Note that ldap3's mocking
# implementation just compares the string in the objects's userPassword
@@ -38,7 +38,7 @@ def login_get_user(loginname, password):
conn.search(conn.user, '(objectClass=person)')
if len(conn.entries) != 1:
return None
- return User.ldap_get(dn)
+ return User.query.get(dn)
@bp.route("/logout")
def logout():
@@ -80,9 +80,8 @@ def login():
def get_current_user():
if 'user_dn' not in session:
- print(session)
return None
- return User.ldap_get(session['user_dn'])
+ return User.query.get(session['user_dn'])
def login_valid():
user = get_current_user()
diff --git a/uffd/user/models.py b/uffd/user/models.py
index e4507027..51dfe827 100644
--- a/uffd/user/models.py
+++ b/uffd/user/models.py
@@ -9,7 +9,7 @@ from uffd.lazyconfig import lazyconfig_str, lazyconfig_list
def get_next_uid():
max_uid = current_app.config['LDAP_USER_MIN_UID']
- for user in User.ldap_all():
+ for user in User.query.all():
if user.uid <= current_app.config['LDAP_USER_MAX_UID']:
max_uid = max(user.uid, max_uid)
next_uid = max_uid + 1
@@ -18,11 +18,11 @@ def get_next_uid():
return next_uid
class User(ldap.Model):
- ldap_base = lazyconfig_str('LDAP_BASE_USER')
- ldap_dn_attribute = 'uid'
- ldap_dn_base = lazyconfig_str('LDAP_BASE_USER')
- ldap_filter = '(objectClass=person)'
+ ldap_search_base = lazyconfig_str('LDAP_BASE_USER')
+ ldap_filter_params = (('objectClass', 'person'),)
ldap_object_classes = lazyconfig_list('LDAP_USER_OBJECTCLASSES')
+ ldap_dn_base = lazyconfig_str('LDAP_BASE_USER')
+ ldap_dn_attribute = 'uid'
uid = ldap.Attribute('uidNumber', default=get_next_uid)
loginname = ldap.Attribute('uid')
@@ -34,14 +34,14 @@ class User(ldap.Model):
roles = [] # Shuts up pylint, overwritten by back-reference
def dummy_attribute_defaults(self):
- if self.ldap_getattr('sn') == []:
- self.ldap_setattr('sn', [' '])
- if self.ldap_getattr('homeDirectory') == []:
- self.ldap_setattr('homeDirectory', ['/home/%s'%self.loginname])
- if self.ldap_getattr('gidNumber') == []:
- self.ldap_setattr('gidNumber', [current_app.config['LDAP_USER_GID']])
+ if self.ldap_object.getattr('sn') == []:
+ self.ldap_object.setattr('sn', [' '])
+ if self.ldap_object.getattr('homeDirectory') == []:
+ self.ldap_object.setattr('homeDirectory', ['/home/%s'%self.loginname])
+ if self.ldap_object.getattr('gidNumber') == []:
+ self.ldap_object.setattr('gidNumber', [current_app.config['LDAP_USER_GID']])
- ldap_pre_create_hooks = ldap.Model.ldap_pre_create_hooks + [dummy_attribute_defaults]
+ ldap_add_hooks = ldap.Model.ldap_add_hooks + (dummy_attribute_defaults,)
# Write-only property
def password(self, value):
@@ -102,12 +102,12 @@ class User(ldap.Model):
return True
class Group(ldap.Model):
- ldap_base = lazyconfig_str('LDAP_BASE_GROUPS')
- ldap_filter = '(objectClass=groupOfUniqueNames)'
+ ldap_search_base = lazyconfig_str('LDAP_BASE_GROUPS')
+ ldap_filter_params = (('objectClass', 'groupOfUniqueNames'),)
gid = ldap.Attribute('gidNumber')
name = ldap.Attribute('cn')
description = ldap.Attribute('description', default='')
- members = ldap.Relation('uniqueMember', User, backref='groups')
+ members = ldap.Relationship('uniqueMember', User, backref='groups')
roles = [] # Shuts up pylint, overwritten by back-reference
diff --git a/uffd/user/views_group.py b/uffd/user/views_group.py
index 3fb1b65f..d4e77741 100644
--- a/uffd/user/views_group.py
+++ b/uffd/user/views_group.py
@@ -19,8 +19,8 @@ def group_acl_check():
@bp.route("/")
@register_navbar('Groups', icon='layer-group', blueprint=bp, visible=group_acl_check)
def index():
- return render_template('group_list.html', groups=Group.ldap_all())
+ return render_template('group_list.html', groups=Group.query.all())
@bp.route("/<int:gid>")
def show(gid):
- return render_template('group.html', group=Group.ldap_filter_by(gid=gid)[0])
+ return render_template('group.html', group=Group.query.filter_by(gid=gid)[0])
diff --git a/uffd/user/views_user.py b/uffd/user/views_user.py
index 0d43feab..63f17cf4 100644
--- a/uffd/user/views_user.py
+++ b/uffd/user/views_user.py
@@ -27,12 +27,12 @@ def user_acl_check():
@bp.route("/")
@register_navbar('Users', icon='users', blueprint=bp, visible=user_acl_check)
def index():
- return render_template('user_list.html', users=User.ldap_all())
+ return render_template('user_list.html', users=User.query.all())
@bp.route("/<int:uid>")
@bp.route("/new")
def show(uid=None):
- user = User() if uid is None else User.ldap_filter_by(uid=uid)[0]
+ user = User() if uid is None else User.query.filter_by(uid=uid)[0]
return render_template('user.html', user=user, roles=Role.query.all())
@bp.route("/<int:uid>/update", methods=['POST'])
@@ -45,7 +45,7 @@ def update(uid=None):
flash('Login name does not meet requirements')
return redirect(url_for('user.show'))
else:
- user = User.ldap_filter_by(uid=uid)[0]
+ user = User.query.filter_by(uid=uid)[0]
if not user.set_mail(request.form['mail']):
flash('Mail is invalid')
return redirect(url_for('user.show', uid=uid))
@@ -56,12 +56,12 @@ def update(uid=None):
new_password = request.form.get('password')
if uid is not None and new_password:
user.set_password(new_password)
+ ldap.session.add(user)
user.roles.clear()
for role in Role.query.all():
if request.values.get('role-{}'.format(role.id), False) or role.name in current_app.config["ROLES_BASEROLES"]:
user.roles.add(role)
user.update_groups()
- ldap.session.add(user)
ldap.session.commit()
db.session.commit()
if uid is None:
@@ -74,7 +74,7 @@ def update(uid=None):
@bp.route("/<int:uid>/del")
@csrf_protect(blueprint=bp)
def delete(uid):
- user = User.ldap_filter_by(uid=uid)[0]
+ user = User.query.filter_by(uid=uid)[0]
user.roles.clear()
ldap.session.delete(user)
ldap.session.commit()
--
GitLab