forked from ungleich-public/ungleich-user
Sajeyks
0cb0344ac8
The highest uidNumber was cached in dal/ldap_max_uid_file. That breaks with more than one replica and is lost on every container restart. The uniqueness check behind it searched base '' (the root DSE), which OpenLDAP never matches, so nothing caught a stale file. The error path also referenced a setting that does not exist. Now a paged search over ENTIRE_SEARCH_BASE finds the highest uidNumber on each activation. Numbering is unchanged: the next number is above both LDAP_DEFAULT_START_UID and every existing uidNumber. LDAP_MAX_UID_FILE_PATH is no longer used.
124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
"""
|
|
Shared helpers for the end-to-end tests.
|
|
|
|
These tests drive the real views over HTTP (Django test client) against a
|
|
real OpenLDAP server, started with:
|
|
|
|
docker compose -f docker-compose.dev.yml up -d --build
|
|
|
|
They are deliberately black-box: they assert on what a user sees (templates,
|
|
emails, status codes) and on what ends up in LDAP, so they keep passing while
|
|
the internals are rewritten during the upgrade.
|
|
"""
|
|
import re
|
|
import uuid
|
|
from unittest import mock
|
|
|
|
import ldap3
|
|
from django.conf import settings
|
|
from django.core import mail
|
|
from django.test import TestCase
|
|
|
|
from dal.ungleich_ldap import LdapManager
|
|
|
|
_ldap_checked = False
|
|
|
|
|
|
def admin_conn():
|
|
server = ldap3.Server(settings.AUTH_LDAP_SERVER)
|
|
return ldap3.Connection(
|
|
server, settings.LDAP_ADMIN_DN, settings.LDAP_ADMIN_PASSWORD,
|
|
auto_bind=True, raise_exceptions=False,
|
|
)
|
|
|
|
|
|
def ensure_ldap_up():
|
|
global _ldap_checked
|
|
if _ldap_checked:
|
|
return
|
|
try:
|
|
admin_conn().unbind()
|
|
except Exception as exc: # pragma: no cover - environment problem
|
|
raise RuntimeError(
|
|
"Test LDAP not reachable at {}. Start it with: docker compose -f "
|
|
"docker-compose.dev.yml up -d --build ({})".format(
|
|
settings.AUTH_LDAP_SERVER, exc)
|
|
)
|
|
_ldap_checked = True
|
|
|
|
|
|
def unique_uid(prefix="t"):
|
|
# Matches the app's username rule: lowercase letters, digits, - and _
|
|
return "{}-{}".format(prefix, uuid.uuid4().hex[:10])
|
|
|
|
|
|
class LdapTestCase(TestCase):
|
|
"""Base class: unique users per test, cleaned up afterwards."""
|
|
|
|
def setUp(self):
|
|
ensure_ldap_up()
|
|
self._created_dns = []
|
|
# OTP server is external; never hit the network from tests.
|
|
self.otp_post = mock.patch("dal.views.requests.post").start()
|
|
self.otp_post.return_value = mock.Mock(status_code=201, text="{}")
|
|
self.otp_get = mock.patch("dal.views.requests.get").start()
|
|
self.otp_get.return_value = mock.Mock(status_code=200, text="[]")
|
|
self.addCleanup(mock.patch.stopall)
|
|
|
|
def tearDown(self):
|
|
conn = admin_conn()
|
|
for dn in self._created_dns:
|
|
conn.delete(dn)
|
|
conn.unbind()
|
|
|
|
# LDAP helpers ---------------------------------------------------------
|
|
|
|
def make_ldap_user(self, uid=None, password="initial-pass-123",
|
|
ou="customers", email=None):
|
|
"""Create a user directly in LDAP, the same way the app does."""
|
|
uid = uid or unique_uid()
|
|
dn = "uid={},ou={},dc=ungleich,dc=ch".format(uid, ou)
|
|
conn = admin_conn()
|
|
ok = conn.add(dn, ["inetOrgPerson", "posixAccount", "ldapPublicKey"], {
|
|
"uid": uid, "cn": uid, "sn": "Tester", "givenName": "Tina",
|
|
"mail": email or "{}@example.com".format(uid),
|
|
"uidNumber": str(30000 + int(uuid.uuid4().int % 100000)),
|
|
"gidNumber": "2000", "homeDirectory": "/home/" + uid,
|
|
"userPassword": LdapManager()._ssha_password(password.encode()),
|
|
})
|
|
conn.unbind()
|
|
self.assertTrue(ok, "could not seed LDAP user {}".format(dn))
|
|
self._created_dns.append(dn)
|
|
return uid
|
|
|
|
def track(self, uid, ou="customers"):
|
|
"""Register a user created by the app for cleanup."""
|
|
self._created_dns.append("uid={},ou={},dc=ungleich,dc=ch".format(uid, ou))
|
|
|
|
def ldap_entry(self, uid, attributes=("*",)):
|
|
conn = admin_conn()
|
|
conn.search(settings.ENTIRE_SEARCH_BASE, "(uid={})".format(
|
|
ldap3.utils.conv.escape_filter_chars(uid)), attributes=list(attributes))
|
|
entries = list(conn.entries)
|
|
conn.unbind()
|
|
return entries[0] if entries else None
|
|
|
|
def can_bind(self, uid, password, ou="customers"):
|
|
server = ldap3.Server(settings.AUTH_LDAP_SERVER)
|
|
conn = ldap3.Connection(
|
|
server, "uid={},ou={},dc=ungleich,dc=ch".format(uid, ou), password)
|
|
ok = conn.bind()
|
|
conn.unbind()
|
|
return ok
|
|
|
|
# HTTP helpers ---------------------------------------------------------
|
|
|
|
def login(self, uid, password):
|
|
self.assertTrue(self.client.login(username=uid, password=password),
|
|
"login failed for {}".format(uid))
|
|
|
|
def last_mail_link(self):
|
|
self.assertTrue(mail.outbox, "no email was sent")
|
|
urls = re.findall(r"https?://testserver(/[^\s\"'<>]+)", mail.outbox[-1].body)
|
|
self.assertTrue(urls, "no link in email: " + mail.outbox[-1].body)
|
|
return urls[-1]
|