ungleich-user/dal/tests/test_flows.py

"""
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 django.test import override_settings

from .base import TEST_LDAP_URI, 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_otp_requests_have_a_timeout(self):
        uid = self.make_ldap_user(password="cust-pass-123")
        self.login(uid, "cust-pass-123")
        self.client.get("/seeds/")
        self.client.post("/otp/", {"username": uid, "password": "cust-pass-123",
                                   "realm": "ungleich-auth"})
        calls = self.otp_get.call_args_list + self.otp_post.call_args_list
        self.assertTrue(calls)
        for call in calls:
            self.assertEqual(call.kwargs.get("timeout"), 10)

    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)


DEAD_LDAP = "ldap://127.0.0.1:1"


class MultipleLdapServerTests(LdapTestCase):

    @override_settings(LDAP_SERVERS=[DEAD_LDAP, TEST_LDAP_URI])
    def test_fails_over_to_second_server(self):
        uid = self.make_ldap_user(password="cust-pass-123")
        self.login(uid, "cust-pass-123")
        resp = self.client.post("/changedata/", {
            "firstname": "Fail", "lastname": "Over", "email": "fo@example.com"})
        self.assertTemplateUsed(resp, "changeddata.html")
        self.assertEqual(self.ldap_entry(uid).sn.value, "Over")

    @override_settings(LDAP_SERVERS=[DEAD_LDAP, TEST_LDAP_URI])
    def test_registration_with_first_server_down(self):
        uid = unique_uid("fo")
        self.track(uid)
        self.client.post("/register/", {
            "username": uid, "password1": "reg-pass-12345", "password2": "reg-pass-12345",
            "email": "fo@example.com", "firstname": "Fay", "lastname": "Lover"})
        self.assertTemplateUsed(self.client.get(self.last_mail_link()), "usercreated.html")
        self.assertTrue(self.can_bind(uid, "reg-pass-12345"))

    @override_settings(LDAP_SERVERS=[DEAD_LDAP])
    def test_all_servers_down_login_fails_cleanly(self):
        self.assertFalse(self.client.login(username="anyone", password="x"))