From a363d4a1c9ac95114b3399bde4737df0d0c9b525 Mon Sep 17 00:00:00 2001
From: Sistason <c3infra@sistason.de>
Date: Sat, 3 Apr 2021 15:42:20 +0200
Subject: [PATCH] started with unittests, but stuck on a mock issue
---
ldap_server_entries.json | 2 +-
tests/test_session.py | 3 +++
tests/utils.py | 6 +++++-
uffd/ldap.py | 7 ++++---
uffd/session/views.py | 14 ++++++++++----
5 files changed, 23 insertions(+), 9 deletions(-)
diff --git a/ldap_server_entries.json b/ldap_server_entries.json
index 5d98bb39..00d2aaea 100644
--- a/ldap_server_entries.json
+++ b/ldap_server_entries.json
@@ -1,7 +1,7 @@
{
"entries": [
{
- "dn": "uid=testuser,ou=users,dc=example,dc=com",
+ "dn": "uid=testuser,ou=users,dc=example,dc=com",
"raw": {
"cn": [
"Test User"
diff --git a/tests/test_session.py b/tests/test_session.py
index f9814ba6..dae41ab5 100644
--- a/tests/test_session.py
+++ b/tests/test_session.py
@@ -135,3 +135,6 @@ class TestSession(UffdTestCase):
class TestSessionOL(TestSession):
use_openldap = True
+
+class TestSessionOLUser(TestSessionOL):
+ use_userconnection = True
diff --git a/tests/utils.py b/tests/utils.py
index 8567d49d..abb0445f 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -25,6 +25,7 @@ def db_flush():
class UffdTestCase(unittest.TestCase):
use_openldap = False
+ use_userconnection = False
def setUp(self):
self.dir = tempfile.mkdtemp()
@@ -43,7 +44,10 @@ class UffdTestCase(unittest.TestCase):
self.skipTest('OPENLDAP_TESTING not set')
config['LDAP_SERVICE_MOCK'] = False
config['LDAP_SERVICE_URL'] = 'ldap://localhost'
- config['LDAP_SERVICE_BIND_DN'] = 'cn=uffd,ou=system,dc=example,dc=com'
+ if self.use_userconnection:
+ config['LDAP_SERVICE_BIND_DN'] = None
+ else:
+ config['LDAP_SERVICE_BIND_DN'] = 'cn=uffd,ou=system,dc=example,dc=com'
config['LDAP_SERVICE_BIND_PASSWORD'] = 'uffd-ldap-password'
os.system("ldapdelete -c -D 'cn=uffd,ou=system,dc=example,dc=com' -w 'uffd-ldap-password' -H 'ldap://localhost' -f ldap_server_entries_cleanup.ldif > /dev/null 2>&1")
os.system("ldapadd -c -D 'cn=uffd,ou=system,dc=example,dc=com' -w 'uffd-ldap-password' -H 'ldap://localhost' -f ldap_server_entries_add.ldif")
diff --git a/uffd/ldap.py b/uffd/ldap.py
index 390beca8..bfa79030 100644
--- a/uffd/ldap.py
+++ b/uffd/ldap.py
@@ -1,7 +1,7 @@
from flask import current_app, request, abort, session
import ldap3
-from ldap3.core.exceptions import LDAPBindError, LDAPPasswordIsMandatoryError
+from ldap3.core.exceptions import LDAPBindError, LDAPPasswordIsMandatoryError, LDAPInvalidDnError
from ldapalchemy import LDAPMapper, LDAPCommitError # pylint: disable=unused-import
from ldapalchemy.model import Query
@@ -42,8 +42,9 @@ def test_user_bind(bind_dn, bind_pw):
return False
conn.search(conn.user, encode_filter(current_app.config["LDAP_USER_SEARCH_FILTER"]))
- #conn.unbind()
- return len(conn.entries) == 1
+ lazy_entries = conn.entries
+ conn.unbind()
+ return len(lazy_entries) == 1
def connect_and_bind_to_ldap(server, bind_dn, bind_pw):
diff --git a/uffd/session/views.py b/uffd/session/views.py
index b8d71629..d27eb43a 100644
--- a/uffd/session/views.py
+++ b/uffd/session/views.py
@@ -5,7 +5,7 @@ import functools
from flask import Blueprint, render_template, request, url_for, redirect, flash, current_app, session, abort
from uffd.user.models import User
-from uffd.ldap import ldap, test_user_bind
+from uffd.ldap import ldap, test_user_bind, LDAPInvalidDnError
from uffd.ratelimit import Ratelimit, host_ratelimit, format_delay
bp = Blueprint("session", __name__, template_folder='templates', url_prefix='/')
@@ -16,7 +16,7 @@ def login_get_user(loginname, password):
dn = User(loginname=loginname).dn
# If we use a service connection, test user bind seperately
- if current_app.config['LDAP_SERVICE_BIND_DN']:
+ if current_app.config['LDAP_SERVICE_BIND_DN'] or current_app.config.get('LDAP_SERVICE_MOCK', False):
if not test_user_bind(dn, password):
return None
# If we use a user connection, just create the connection normally
@@ -25,9 +25,15 @@ def login_get_user(loginname, password):
session['user_dn'] = dn
session['user_pw'] = password
if not ldap.get_connection():
+ session.clear()
return None
-
- return User.query.get(dn)
+
+ try:
+ user = User.query.get(dn)
+ if user:
+ return user
+ except LDAPInvalidDnError:
+ return None
@bp.route("/logout")
def logout():
--
GitLab