forked from ungleich-public/ungleich-user
Sajeyks
ab45535f38
python-ldap is a C extension that needs the OpenLDAP headers to build (and needed a workaround in the Dockerfile). ldap3, which the app already used for all writes, is pure Python. dal.backends.LdapBackend keeps the django-auth-ldap behaviour: search the uid in every LDAPSEARCH base as the admin DN, require exactly one match, log in with the user's DN, and mirror givenName, sn and mail into the local Django user. Empty passwords are rejected (LDAP treats them as anonymous logins, which succeed). LDAP_USE_TLS now applies StartTLS to every connection, including the admin connection used for writes, not just logins.
445 lines
20 KiB
Python
445 lines
20 KiB
Python
"""
|
|
End-to-end behaviour of the account webapp against a real LDAP server.
|
|
|
|
Several tests guard against security bugs found in the 2026 upgrade audit
|
|
(account takeover via /reset/, LDAP filter injection, passwords in links,
|
|
duplicate usernames across subtrees).
|
|
"""
|
|
from django.core import mail
|
|
|
|
from .base import LdapTestCase, unique_uid
|
|
|
|
REGISTER = "/register/"
|
|
RESET = "/resetpassword/"
|
|
|
|
|
|
class LoginTests(LdapTestCase):
|
|
|
|
def test_landing_page_renders(self):
|
|
resp = self.client.get("/")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertTemplateUsed(resp, "landing.html")
|
|
|
|
def test_customer_can_log_in(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
resp = self.client.post("/index/", {"username": uid, "password": "cust-pass-123"})
|
|
self.assertTemplateUsed(resp, "useroptions.html")
|
|
self.assertIn("_auth_user_id", self.client.session)
|
|
|
|
def test_staff_in_ou_users_can_log_in(self):
|
|
uid = self.make_ldap_user(password="staff-pass-123", ou="users")
|
|
resp = self.client.post("/index/", {"username": uid, "password": "staff-pass-123"})
|
|
self.assertTemplateUsed(resp, "useroptions.html")
|
|
|
|
def test_wrong_password_is_rejected(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
resp = self.client.post("/index/", {"username": uid, "password": "nope"})
|
|
self.assertNotIn("_auth_user_id", self.client.session)
|
|
self.assertContains(resp, "incorrect")
|
|
|
|
def test_empty_password_is_rejected(self):
|
|
"""An empty password is an anonymous LDAP bind, which succeeds."""
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.assertFalse(self.client.login(username=uid, password=""))
|
|
resp = self.client.post("/index/", {"username": uid, "password": ""})
|
|
self.assertNotIn("_auth_user_id", self.client.session)
|
|
|
|
def test_unknown_user_is_rejected(self):
|
|
self.assertFalse(self.client.login(username="nobody-here", password="x"))
|
|
|
|
def test_filter_characters_in_username_do_not_match(self):
|
|
self.make_ldap_user(password="cust-pass-123")
|
|
self.assertFalse(self.client.login(username="*", password="cust-pass-123"))
|
|
|
|
def test_uid_in_two_subtrees_cannot_log_in(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.make_ldap_user(uid=uid, password="cust-pass-123", ou="users")
|
|
self.assertFalse(self.client.login(username=uid, password="cust-pass-123"))
|
|
|
|
def test_login_mirrors_ldap_attributes(self):
|
|
from django.contrib.auth.models import User
|
|
uid = self.make_ldap_user(password="cust-pass-123", email="mirror@example.com")
|
|
self.login(uid, "cust-pass-123")
|
|
user = User.objects.get(username=uid)
|
|
self.assertEqual((user.first_name, user.last_name, user.email),
|
|
("Tina", "Tester", "mirror@example.com"))
|
|
self.assertFalse(user.has_usable_password())
|
|
|
|
def test_logged_in_user_gets_options_page(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.get("/")
|
|
self.assertTemplateUsed(resp, "useroptions.html")
|
|
self.assertIn("no-store", resp["Cache-Control"])
|
|
|
|
def test_logout(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.get("/logout/")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertNotIn("_auth_user_id", self.client.session)
|
|
|
|
|
|
class RegisterTests(LdapTestCase):
|
|
|
|
def form(self, **overrides):
|
|
data = {"username": unique_uid("reg"), "password1": "reg-pass-12345",
|
|
"password2": "reg-pass-12345", "email": "reg@example.com",
|
|
"firstname": "Rita", "lastname": "Register"}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
def assert_error(self, data, text):
|
|
resp = self.client.post(REGISTER, data)
|
|
self.assertTemplateUsed(resp, "error.html")
|
|
self.assertContains(resp, text)
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
|
|
def test_form_renders(self):
|
|
self.assertTemplateUsed(self.client.get(REGISTER), "registeruser.html")
|
|
|
|
def test_validation_errors(self):
|
|
self.assert_error(self.form(username=""), "Please supply a username")
|
|
self.assert_error(self.form(username="Bad Name"), "lowercase letters")
|
|
self.assert_error(self.form(username="pipe|name"), "lowercase letters")
|
|
self.assert_error(self.form(password2="different"), "match")
|
|
self.assert_error(self.form(email="not-an-email"), "email address is invalid")
|
|
self.assert_error(self.form(lastname=""), "firstname and lastname")
|
|
|
|
def test_full_registration_and_activation(self):
|
|
data = self.form()
|
|
uid = data["username"]
|
|
self.track(uid)
|
|
|
|
resp = self.client.post(REGISTER, data)
|
|
self.assertTemplateUsed(resp, "confirm_email.html")
|
|
self.assertEqual(len(mail.outbox), 1)
|
|
self.assertIsNone(self.ldap_entry(uid), "user must not exist before activation")
|
|
|
|
link = self.last_mail_link()
|
|
resp = self.client.get(link)
|
|
self.assertTemplateUsed(resp, "usercreated.html")
|
|
|
|
entry = self.ldap_entry(uid)
|
|
self.assertIsNotNone(entry)
|
|
self.assertEqual(entry.mail.value, "reg@example.com")
|
|
self.assertEqual(entry.givenName.value, "Rita")
|
|
self.assertEqual(int(entry.gidNumber.value), 2000)
|
|
self.assertIn("ldapPublicKey", entry.objectClass.values)
|
|
self.assertTrue(entry.userPassword.value.startswith(b"{SSHA}"))
|
|
self.assertTrue(self.can_bind(uid, "reg-pass-12345"))
|
|
self.otp_post.assert_called_once() # OTP seed requested for new user
|
|
|
|
# Activation link is single use
|
|
self.assertEqual(self.client.get(link).status_code, 404)
|
|
|
|
def test_tampered_activation_token_is_rejected(self):
|
|
data = self.form()
|
|
self.track(data["username"])
|
|
self.client.post(REGISTER, data)
|
|
self.assertEqual(self.client.get("/activate/not-the-token/").status_code, 404)
|
|
self.assertIsNone(self.ldap_entry(data["username"]))
|
|
|
|
def test_uid_numbers_are_unique(self):
|
|
uids = []
|
|
for _ in range(2):
|
|
data = self.form()
|
|
self.track(data["username"])
|
|
self.client.post(REGISTER, data)
|
|
self.client.get(self.last_mail_link())
|
|
uids.append(int(self.ldap_entry(data["username"]).uidNumber.value))
|
|
self.assertNotEqual(uids[0], uids[1])
|
|
|
|
def test_uid_number_is_above_every_existing_one(self):
|
|
from .base import admin_conn
|
|
self.make_ldap_user(ou="users") # random high uidNumber
|
|
conn = admin_conn()
|
|
conn.search("dc=ungleich,dc=ch", "(uidNumber=*)", attributes=["uidNumber"])
|
|
highest = max(int(e.uidNumber.value) for e in conn.entries)
|
|
conn.unbind()
|
|
data = self.form()
|
|
self.track(data["username"])
|
|
self.client.post(REGISTER, data)
|
|
self.client.get(self.last_mail_link())
|
|
self.assertEqual(int(self.ldap_entry(data["username"]).uidNumber.value), highest + 1)
|
|
|
|
def test_password_not_in_activation_link_or_db(self):
|
|
"""The plaintext password must never be stored or put in a URL."""
|
|
from dal.models import PendingRegistration
|
|
import base64
|
|
data = self.form(password1="Very-Secret-pw-1", password2="Very-Secret-pw-1")
|
|
self.track(data["username"])
|
|
self.client.post(REGISTER, data)
|
|
link = self.last_mail_link()
|
|
self.assertNotIn(base64.b64encode(b"Very-Secret-pw-1").decode(), link)
|
|
self.assertNotIn("Very-Secret-pw-1", mail.outbox[-1].body)
|
|
stored = PendingRegistration.objects.get(username=data["username"])
|
|
self.assertTrue(stored.password_hash.startswith("{SSHA}"))
|
|
self.assertNotIn("Very-Secret-pw-1", stored.password_hash)
|
|
|
|
def test_cannot_register_name_taken_by_staff(self):
|
|
"""Username in ou=users must not be re-registrable under ou=customers."""
|
|
staff = self.make_ldap_user(ou="users")
|
|
self.track(staff) # in case the bug creates a customer duplicate
|
|
self.client.post(REGISTER, self.form(username=staff))
|
|
if mail.outbox:
|
|
self.client.get(self.last_mail_link())
|
|
self.assertIsNone(
|
|
self.ldap_entry_in(staff, "ou=customers,dc=ungleich,dc=ch"))
|
|
|
|
def test_taken_username_rejected_at_form(self):
|
|
existing = self.make_ldap_user()
|
|
self.assert_error(self.form(username=existing), "already taken")
|
|
|
|
def test_name_taken_between_signup_and_activation(self):
|
|
data = self.form()
|
|
uid = data["username"]
|
|
self.client.post(REGISTER, data)
|
|
link = self.last_mail_link()
|
|
self.make_ldap_user(uid=uid, ou="users") # someone else got it first
|
|
resp = self.client.get(link)
|
|
self.assertTemplateUsed(resp, "error.html")
|
|
self.assertIsNone(self.ldap_entry_in(uid, "ou=customers,dc=ungleich,dc=ch"))
|
|
|
|
def ldap_entry_in(self, uid, base):
|
|
from .base import admin_conn
|
|
conn = admin_conn()
|
|
conn.search(base, "(uid={})".format(uid), attributes=["uid"])
|
|
found = list(conn.entries)
|
|
conn.unbind()
|
|
return found[0] if found else None
|
|
|
|
|
|
class ChangeDataTests(LdapTestCase):
|
|
|
|
def test_requires_login(self):
|
|
resp = self.client.get("/changedata/")
|
|
self.assertEqual(resp.status_code, 302)
|
|
|
|
def test_shows_and_updates_data(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.get("/changedata/")
|
|
self.assertTemplateUsed(resp, "changeuserdata.html")
|
|
self.assertContains(resp, "Tina")
|
|
|
|
resp = self.client.post("/changedata/", {
|
|
"firstname": "Neo", "lastname": "Name", "email": "new@example.com"})
|
|
self.assertTemplateUsed(resp, "changeddata.html")
|
|
entry = self.ldap_entry(uid)
|
|
self.assertEqual(entry.givenName.value, "Neo")
|
|
self.assertEqual(entry.sn.value, "Name")
|
|
self.assertEqual(entry.mail.value, "new@example.com")
|
|
|
|
def test_rejects_invalid_email(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.post("/changedata/", {
|
|
"firstname": "Neo", "lastname": "Name", "email": "broken"})
|
|
self.assertTemplateUsed(resp, "error.html")
|
|
self.assertEqual(self.ldap_entry(uid).mail.value, uid + "@example.com")
|
|
|
|
|
|
class ChangePasswordTests(LdapTestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.uid = self.make_ldap_user(password="old-pass-1234")
|
|
self.login(self.uid, "old-pass-1234")
|
|
|
|
def post(self, old, new1, new2=None):
|
|
return self.client.post("/changepassword/", {
|
|
"oldpassword": old, "password1": new1, "password2": new2 or new1})
|
|
|
|
def test_wrong_old_password(self):
|
|
resp = self.post("wrong", "new-pass-1234")
|
|
self.assertContains(resp, "Wrong password")
|
|
self.assertTrue(self.can_bind(self.uid, "old-pass-1234"))
|
|
|
|
def test_mismatch_and_too_short(self):
|
|
self.assertContains(self.post("old-pass-1234", "new-pass-1234", "other-pass-1234"),
|
|
"same password")
|
|
self.assertContains(self.post("old-pass-1234", "short"), "too short")
|
|
self.assertTrue(self.can_bind(self.uid, "old-pass-1234"))
|
|
|
|
def test_success_changes_ldap_password_and_logs_out(self):
|
|
resp = self.post("old-pass-1234", "new-pass-1234")
|
|
self.assertTemplateUsed(resp, "changedpassword.html")
|
|
self.assertTrue(self.can_bind(self.uid, "new-pass-1234"))
|
|
self.assertFalse(self.can_bind(self.uid, "old-pass-1234"))
|
|
self.assertNotIn("_auth_user_id", self.client.session)
|
|
|
|
|
|
class ResetPasswordTests(LdapTestCase):
|
|
|
|
def test_unknown_user(self):
|
|
resp = self.client.post(RESET, {"user": "does-not-exist"})
|
|
self.assertContains(resp, "does not exist")
|
|
self.assertEqual(len(mail.outbox), 0)
|
|
|
|
def test_full_reset_flow(self):
|
|
uid = self.make_ldap_user(password="old-pass-1234")
|
|
resp = self.client.post(RESET, {"user": uid})
|
|
self.assertTemplateUsed(resp, "send_resetrequest.html")
|
|
self.assertEqual(mail.outbox[0].to, ["{0} <{0}@example.com>".format(uid)])
|
|
|
|
link = self.last_mail_link()
|
|
resp = self.client.get(link)
|
|
self.assertTemplateUsed(resp, "resetpasswordnew.html")
|
|
self.assertEqual(self.client.get(link).status_code, 404, "link must be single use")
|
|
|
|
resp = self.client.post("/reset/", {
|
|
"user": uid, "password1": "brand-new-1234", "password2": "brand-new-1234"})
|
|
self.assertTemplateUsed(resp, "changedpassword.html")
|
|
self.assertTrue(self.can_bind(uid, "brand-new-1234"))
|
|
|
|
def test_verified_link_cannot_reset_another_user(self):
|
|
mine = self.make_ldap_user(password="old-pass-1234")
|
|
victim = self.make_ldap_user(password="victim-pass-1234")
|
|
self.client.post(RESET, {"user": mine})
|
|
self.client.get(self.last_mail_link())
|
|
self.client.post("/reset/", {
|
|
"user": victim, "password1": "attacker-1234", "password2": "attacker-1234"})
|
|
self.assertTrue(self.can_bind(victim, "victim-pass-1234"))
|
|
self.assertTrue(self.can_bind(mine, "attacker-1234"))
|
|
|
|
def test_verified_link_is_single_use(self):
|
|
uid = self.make_ldap_user(password="old-pass-1234")
|
|
self.client.post(RESET, {"user": uid})
|
|
self.client.get(self.last_mail_link())
|
|
self.client.post("/reset/", {
|
|
"user": uid, "password1": "first-new-1234", "password2": "first-new-1234"})
|
|
resp = self.client.post("/reset/", {
|
|
"user": uid, "password1": "second-new-1234", "password2": "second-new-1234"})
|
|
self.assertContains(resp, "invalid or has expired")
|
|
self.assertTrue(self.can_bind(uid, "first-new-1234"))
|
|
|
|
def test_garbage_reset_link_is_404(self):
|
|
self.assertEqual(self.client.get("/reset/%%%/abc/").status_code, 404)
|
|
|
|
def test_rejects_short_new_password(self):
|
|
uid = self.make_ldap_user(password="old-pass-1234")
|
|
self.client.post(RESET, {"user": uid})
|
|
self.client.get(self.last_mail_link())
|
|
resp = self.client.post("/reset/", {
|
|
"user": uid, "password1": "short", "password2": "short"})
|
|
self.assertContains(resp, "too short")
|
|
|
|
def test_reset_post_requires_a_valid_token(self):
|
|
"""POST /reset/ must not set a password without a verified link."""
|
|
victim = self.make_ldap_user(password="victim-pass-1234", ou="users")
|
|
self.client.post("/reset/", {
|
|
"user": victim, "password1": "attacker-1234", "password2": "attacker-1234"})
|
|
self.assertFalse(self.can_bind(victim, "attacker-1234", ou="users"),
|
|
"account takeover without any token")
|
|
|
|
def test_ldap_filter_injection_in_reset(self):
|
|
"""user='*' must not match the first entry in the whole tree."""
|
|
self.make_ldap_user(password="victim-pass-1234")
|
|
self.client.post("/reset/", {
|
|
"user": "*", "password1": "attacker-1234", "password2": "attacker-1234"})
|
|
from .base import admin_conn
|
|
conn = admin_conn()
|
|
conn.search("dc=ungleich,dc=ch", "(uid=*)", attributes=["uid"])
|
|
uids = [e.uid.value for e in conn.entries]
|
|
conn.unbind()
|
|
hijacked = [u for u in uids for ou in ("customers", "users")
|
|
if self.can_bind(u, "attacker-1234", ou=ou)]
|
|
self.assertEqual(hijacked, [])
|
|
|
|
|
|
class DeleteAccountTests(LdapTestCase):
|
|
|
|
def test_requires_login(self):
|
|
self.assertEqual(self.client.get("/deleteaccount/").status_code, 302)
|
|
|
|
def test_wrong_password_keeps_account(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.post("/deleteaccount/", {"username": uid, "password": "nope"})
|
|
self.assertContains(resp, "Wrong password")
|
|
self.assertIsNotNone(self.ldap_entry(uid))
|
|
|
|
def test_cannot_delete_someone_else(self):
|
|
me = self.make_ldap_user(password="cust-pass-123")
|
|
other = self.make_ldap_user(password="other-pass-123")
|
|
self.login(me, "cust-pass-123")
|
|
resp = self.client.post("/deleteaccount/", {"username": other, "password": "other-pass-123"})
|
|
self.assertContains(resp, "Unknown user")
|
|
self.assertIsNotNone(self.ldap_entry(other))
|
|
|
|
def test_delete_own_account(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.post("/deleteaccount/", {"username": uid, "password": "cust-pass-123"})
|
|
self.assertTemplateUsed(resp, "deleteduser.html")
|
|
self.assertIsNone(self.ldap_entry(uid))
|
|
self.assertNotIn("_auth_user_id", self.client.session)
|
|
|
|
|
|
class UserCreateApiTests(LdapTestCase):
|
|
|
|
def test_validation(self):
|
|
self.assertEqual(self.client.post("/create/", {}).status_code, 400)
|
|
taken = self.make_ldap_user()
|
|
self.assertEqual(self.client.post("/create/", {
|
|
"username": taken, "email": "a@example.com", "firstname": "a", "lastname": "b"}
|
|
).status_code, 400)
|
|
self.assertEqual(self.client.post("/create/", {
|
|
"username": "ok-name", "email": "bad", "firstname": "a", "lastname": "b"}
|
|
).status_code, 400)
|
|
|
|
def test_creates_account_via_emailed_link(self):
|
|
uid = unique_uid("api")
|
|
self.track(uid)
|
|
resp = self.client.post("/create/", {
|
|
"username": uid, "email": "api@example.com",
|
|
"firstname": "Api", "lastname": "User"})
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = mail.outbox[0].body
|
|
import re
|
|
password = re.search(r"Password: (\S+)", body).group(1)
|
|
self.assertTemplateUsed(self.client.get(self.last_mail_link()), "usercreated.html")
|
|
self.assertTrue(self.can_bind(uid, password))
|
|
|
|
|
|
class OtpSeedTests(LdapTestCase):
|
|
|
|
def test_seed_list_shows_only_own_seeds(self):
|
|
import json
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
self.otp_get.return_value.text = json.dumps([
|
|
{"name": uid, "realm": "ungleich-auth", "seed": "MINE"},
|
|
{"name": "someone-else", "realm": "ungleich-auth", "seed": "THEIRS"},
|
|
])
|
|
self.login(uid, "cust-pass-123")
|
|
resp = self.client.get("/seeds/")
|
|
self.assertContains(resp, "MINE")
|
|
self.assertNotContains(resp, "THEIRS")
|
|
|
|
def test_api_does_not_print_password(self):
|
|
import io
|
|
from contextlib import redirect_stdout
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
out = io.StringIO()
|
|
with redirect_stdout(out), self.assertLogs("dal", "DEBUG") as logs:
|
|
import logging
|
|
logging.getLogger("dal").debug("sentinel")
|
|
self.client.post("/otp/", {"username": uid, "password": "cust-pass-123",
|
|
"realm": "ungleich-auth"})
|
|
self.assertNotIn("cust-pass-123", out.getvalue())
|
|
self.assertNotIn("cust-pass-123", "\n".join(logs.output))
|
|
|
|
def test_api_rejects_bad_credentials(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
resp = self.client.post("/otp/", {"username": uid, "password": "x", "realm": "ungleich-auth"})
|
|
self.assertEqual(resp.status_code, 400)
|
|
|
|
def test_api_creates_seed_in_allowed_realm_only(self):
|
|
uid = self.make_ldap_user(password="cust-pass-123")
|
|
resp = self.client.post("/otp/", {"username": uid, "password": "cust-pass-123",
|
|
"realm": "evil-realm"})
|
|
self.assertEqual(resp.status_code, 403)
|
|
resp = self.client.post("/otp/", {"username": uid, "password": "cust-pass-123",
|
|
"realm": "ungleich-auth"})
|
|
self.assertEqual(resp.status_code, 201)
|