bambuddy/backend/tests/unit/services/test_ldap_service.py

"""Tests for LDAP authentication service (#794).

Tests the pure logic functions in ldap_service.py:
- Config parsing from settings dict
- LDAP filter escaping (RFC 4515)
- Group mapping resolution
- LDAPConfig/LDAPUserInfo dataclass construction

Network-dependent functions (authenticate_ldap_user, test_ldap_connection)
are not tested here — they require a live LDAP server.
"""

from types import SimpleNamespace

import pytest
from ldap3.core.exceptions import (
    LDAPObjectClassError,
    LDAPSocketOpenError,
    LDAPStartTLSError,
    LDAPUnwillingToPerformResult,
)
from ldap3.utils.ciDict import CaseInsensitiveDict

from backend.app.services.ldap_service import (
    LDAPConfig,
    LDAPSearchResult,
    LDAPUserInfo,
    _ldap_escape,
    authenticate_ldap_user,
    lookup_ldap_user,
    parse_ldap_config,
    resolve_group_mapping,
    search_ldap_users,
    test_ldap_connection as check_ldap_connection,
)


class TestParseConfig:
    """Verify parse_ldap_config builds LDAPConfig from settings dict."""

    def test_returns_none_when_disabled(self):
        settings = {"ldap_enabled": "false", "ldap_server_url": "ldaps://example.com"}
        assert parse_ldap_config(settings) is None

    def test_returns_none_when_missing_enabled(self):
        settings = {"ldap_server_url": "ldaps://example.com"}
        assert parse_ldap_config(settings) is None

    def test_returns_none_when_no_server_url(self):
        settings = {"ldap_enabled": "true", "ldap_server_url": ""}
        assert parse_ldap_config(settings) is None

    def test_returns_none_when_server_url_whitespace(self):
        settings = {"ldap_enabled": "true", "ldap_server_url": "   "}
        assert parse_ldap_config(settings) is None

    def test_parses_minimal_config(self):
        settings = {
            "ldap_enabled": "true",
            "ldap_server_url": "ldaps://ldap.example.com:636",
        }
        config = parse_ldap_config(settings)
        assert config is not None
        assert config.server_url == "ldaps://ldap.example.com:636"
        assert config.bind_dn == ""
        assert config.search_base == ""
        assert config.user_filter == "(sAMAccountName={username})"
        assert config.security == "starttls"
        assert config.group_mapping == {}
        assert config.auto_provision is False
        assert config.ca_cert_path == ""
        assert config.default_group == ""

    def test_parses_full_config(self):
        settings = {
            "ldap_enabled": "true",
            "ldap_server_url": "ldaps://ldap.example.com:636",
            "ldap_bind_dn": "cn=admin,dc=example,dc=com",
            "ldap_bind_password": "secret",
            "ldap_search_base": "ou=users,dc=example,dc=com",
            "ldap_user_filter": "(uid={username})",
            "ldap_security": "ldaps",
            "ldap_group_mapping": '{"cn=admins,dc=example,dc=com": "Administrators"}',
            "ldap_auto_provision": "true",
            "ldap_ca_cert_path": "/path/to/ca.pem",
            "ldap_default_group": "Viewers",
        }
        config = parse_ldap_config(settings)
        assert config is not None
        assert config.bind_dn == "cn=admin,dc=example,dc=com"
        assert config.bind_password == "secret"
        assert config.search_base == "ou=users,dc=example,dc=com"
        assert config.user_filter == "(uid={username})"
        assert config.security == "ldaps"
        assert config.group_mapping == {"cn=admins,dc=example,dc=com": "Administrators"}
        assert config.auto_provision is True
        assert config.ca_cert_path == "/path/to/ca.pem"
        assert config.default_group == "Viewers"

    def test_handles_invalid_group_mapping_json(self):
        settings = {
            "ldap_enabled": "true",
            "ldap_server_url": "ldaps://ldap.example.com",
            "ldap_group_mapping": "not valid json",
        }
        config = parse_ldap_config(settings)
        assert config is not None
        assert config.group_mapping == {}

    def test_handles_non_dict_group_mapping(self):
        settings = {
            "ldap_enabled": "true",
            "ldap_server_url": "ldaps://ldap.example.com",
            "ldap_group_mapping": '["not", "a", "dict"]',
        }
        config = parse_ldap_config(settings)
        assert config is not None
        assert config.group_mapping == {}

    def test_enabled_case_insensitive(self):
        settings = {"ldap_enabled": "True", "ldap_server_url": "ldaps://ldap.example.com"}
        assert parse_ldap_config(settings) is not None

        settings = {"ldap_enabled": "TRUE", "ldap_server_url": "ldaps://ldap.example.com"}
        assert parse_ldap_config(settings) is not None

    def test_strips_whitespace(self):
        settings = {
            "ldap_enabled": "true",
            "ldap_server_url": "  ldaps://ldap.example.com  ",
            "ldap_bind_dn": "  cn=admin,dc=example,dc=com  ",
            "ldap_search_base": "  dc=example,dc=com  ",
            "ldap_default_group": "  Viewers  ",
        }
        config = parse_ldap_config(settings)
        assert config.server_url == "ldaps://ldap.example.com"
        assert config.bind_dn == "cn=admin,dc=example,dc=com"
        assert config.search_base == "dc=example,dc=com"
        assert config.default_group == "Viewers"


class TestLDAPEscape:
    """Verify RFC 4515 escaping for LDAP search filter values."""

    def test_plain_string(self):
        assert _ldap_escape("testuser") == "testuser"

    def test_escapes_backslash(self):
        assert _ldap_escape("test\\user") == "test\\5cuser"

    def test_escapes_asterisk(self):
        assert _ldap_escape("test*user") == "test\\2auser"

    def test_escapes_open_paren(self):
        assert _ldap_escape("test(user") == "test\\28user"

    def test_escapes_close_paren(self):
        assert _ldap_escape("test)user") == "test\\29user"

    def test_escapes_null(self):
        assert _ldap_escape("test\x00user") == "test\\00user"

    def test_escapes_multiple_chars(self):
        assert _ldap_escape("a*b(c)d\\e") == "a\\2ab\\28c\\29d\\5ce"

    def test_empty_string(self):
        assert _ldap_escape("") == ""


class TestResolveGroupMapping:
    """Verify LDAP group DN to BamBuddy group name resolution."""

    def test_empty_mapping(self):
        assert resolve_group_mapping(["cn=admins,dc=example"], {}) == []

    def test_empty_groups(self):
        mapping = {"cn=admins,dc=example": "Administrators"}
        assert resolve_group_mapping([], mapping) == []

    def test_single_match(self):
        mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
        groups = ["cn=admins,dc=example,dc=com"]
        assert resolve_group_mapping(groups, mapping) == ["Administrators"]

    def test_multiple_matches(self):
        mapping = {
            "cn=admins,dc=example,dc=com": "Administrators",
            "cn=ops,dc=example,dc=com": "Operators",
        }
        groups = ["cn=admins,dc=example,dc=com", "cn=ops,dc=example,dc=com"]
        result = resolve_group_mapping(groups, mapping)
        assert set(result) == {"Administrators", "Operators"}

    def test_no_match(self):
        mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
        groups = ["cn=users,dc=example,dc=com"]
        assert resolve_group_mapping(groups, mapping) == []

    def test_case_insensitive_dn(self):
        mapping = {"CN=Admins,DC=Example,DC=Com": "Administrators"}
        groups = ["cn=admins,dc=example,dc=com"]
        assert resolve_group_mapping(groups, mapping) == ["Administrators"]

    def test_partial_match_not_matched(self):
        mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
        groups = ["cn=admins,dc=other,dc=com"]
        assert resolve_group_mapping(groups, mapping) == []

    def test_extra_groups_ignored(self):
        mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
        groups = ["cn=admins,dc=example,dc=com", "cn=users,dc=example,dc=com", "cn=devs,dc=example,dc=com"]
        assert resolve_group_mapping(groups, mapping) == ["Administrators"]


class TestDataclasses:
    """Verify dataclass construction."""

    def test_ldap_user_info(self):
        info = LDAPUserInfo(
            username="testuser",
            email="test@example.com",
            display_name="Test User",
            groups=["cn=admins,dc=example,dc=com"],
        )
        assert info.username == "testuser"
        assert info.email == "test@example.com"
        assert info.display_name == "Test User"
        assert info.groups == ["cn=admins,dc=example,dc=com"]

    def test_ldap_user_info_none_fields(self):
        info = LDAPUserInfo(username="testuser", email=None, display_name=None, groups=[])
        assert info.email is None
        assert info.display_name is None
        assert info.groups == []

    def test_ldap_config(self):
        config = LDAPConfig(
            server_url="ldaps://ldap.example.com:636",
            bind_dn="cn=admin,dc=example,dc=com",
            bind_password="secret",
            search_base="dc=example,dc=com",
            user_filter="(uid={username})",
            security="ldaps",
            group_mapping={"cn=admins": "Administrators"},
            auto_provision=True,
            ca_cert_path="",
            default_group="Viewers",
        )
        assert config.server_url == "ldaps://ldap.example.com:636"
        assert config.auto_provision is True
        assert config.default_group == "Viewers"


# ---------------------------------------------------------------------------
# Mocked authenticate_ldap_user group-discovery tests
# ---------------------------------------------------------------------------
# These tests mock ldap3.Connection to exercise the group-discovery logic in
# authenticate_ldap_user without a live LDAP server. Added after a bug where
# POSIX primary-group membership (via gidNumber) was ignored — see CHANGELOG.


class _MockAttr:
    """Minimal stand-in for ldap3 Attribute objects.

    Supports str(), bool(), .value, .values, and iteration — the operations
    used by ldap_service against user entry attributes.
    """

    def __init__(self, value):
        self._value = value

    @property
    def value(self):
        return self._value

    @property
    def values(self):
        return self._value if isinstance(self._value, list) else [self._value]

    def __str__(self):
        return str(self._value)

    def __bool__(self):
        return bool(self._value)

    def __iter__(self):
        if isinstance(self._value, list):
            return iter(self._value)
        return iter([self._value])


class _MockEntry:
    """Minimal stand-in for ldap3 Entry. Only attributes passed at construction exist."""

    def __init__(self, dn, **attrs):
        self.entry_dn = dn
        for key, val in attrs.items():
            setattr(self, key, _MockAttr(val))


class _MockServer:
    """Stand-in for ldap3 Server: only the schema and root DSE info the service reads.

    `schema` None is a server that published no schema, where ldap3 checks no
    names client-side. Otherwise it carries the attribute types and object
    classes the server defines.
    """

    def __init__(self, attribute_types=None, object_classes=None, naming_contexts=None, active_directory=False):
        if attribute_types is None and object_classes is None:
            self.schema = None
        else:
            self.schema = SimpleNamespace(
                attribute_types=CaseInsensitiveDict(dict.fromkeys(attribute_types or ())),
                object_classes=CaseInsensitiveDict(dict.fromkeys(object_classes or ())),
            )
        features = [("1.2.840.113556.1.4.800", "FEATURE", "Active directory", "MICROSOFT")] if active_directory else []
        if naming_contexts is None and not active_directory:
            self.info = None
        else:
            self.info = SimpleNamespace(naming_contexts=naming_contexts, supported_features=features)


class _MockConnection:
    """Mock ldap3 Connection that returns pre-configured entries based on filter substring match.

    Every Connection() instance shares a class-level fixture dict so the service-account
    connection and the user-bind connection both see the same fake directory.
    """

    _search_fixture: dict[str, list] = {}
    _instances: list["_MockConnection"] = []
    # Filter substring that should raise LDAPObjectClassError instead of
    # searching, standing in for ldap3's client-side schema validation — it
    # rejects an object class the server's published schema doesn't define
    # before the request is ever built (#2769).
    _raise_object_class_error_on: str | None = None

    def __init__(self, server=None, *args, **kwargs):
        self.server = server
        self.entries: list = []
        self.search_calls: list[str] = []
        self.search_bases: list[str | None] = []
        self.search_attrs: list[list | None] = []
        self.last_attrs: list | None = None
        _MockConnection._instances.append(self)

    def open(self):
        pass

    def start_tls(self, read_server_info=True):
        self.start_tls_read_server_info = read_server_info

    def bind(self):
        return True

    def unbind(self):
        pass

    def search(self, search_base=None, search_filter=None, search_scope=None, attributes=None, **kwargs):
        # **kwargs absorbs ldap3 options like size_limit that the real client supports
        self.search_calls.append(search_filter or "")
        self.search_bases.append(search_base)
        self.last_attrs = list(attributes) if attributes is not None else None
        self.search_attrs.append(self.last_attrs)
        needle = _MockConnection._raise_object_class_error_on
        if needle and needle in (search_filter or ""):
            raise LDAPObjectClassError(f"invalid class in objectClass attribute: {needle}")
        for needle, entries in _MockConnection._search_fixture.items():
            if needle in (search_filter or ""):
                self.entries = entries
                return True
        self.entries = []
        return True


@pytest.fixture
def mock_ldap(monkeypatch):
    """Patch Connection + _create_server in ldap_service so authenticate_ldap_user can run offline."""
    _MockConnection._search_fixture = {}
    _MockConnection._instances = []
    _MockConnection._raise_object_class_error_on = None
    _MockConnection.server_fixture = _MockServer()
    monkeypatch.setattr("backend.app.services.ldap_service.Connection", _MockConnection)
    monkeypatch.setattr(
        "backend.app.services.ldap_service._create_server", lambda config: _MockConnection.server_fixture
    )
    return _MockConnection


def _base_config(**overrides):
    """Build a minimal LDAPConfig for mocked tests."""
    defaults = {
        "server_url": "ldaps://test.example.com:636",
        "bind_dn": "cn=admin,dc=test,dc=com",
        "bind_password": "x",
        "search_base": "dc=test,dc=com",
        "user_filter": "(uid={username})",
        "security": "ldaps",
        "group_mapping": {},
        "auto_provision": False,
        "ca_cert_path": "",
        "default_group": "",
    }
    defaults.update(overrides)
    return LDAPConfig(**defaults)


class TestAuthenticateLdapUserGroups:
    """Group-discovery behaviour in authenticate_ldap_user.

    Covers the POSIX primary gidNumber lookup and case-insensitive dedupe added
    to fix a bug where users whose role came from their primary group were
    authenticated without the correct group membership.
    """

    def test_primary_gidnumber_group_found(self, mock_ldap):
        """Regression: POSIX primary group (gidNumber match) must be included in the result."""
        user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
        operators_group = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")

        mock_ldap._search_fixture = {
            "(uid=mz)": [user_entry],
            "memberUid=mz": [],  # no supplementary memberships
            "gidNumber=10002": [operators_group],
        }

        info = authenticate_ldap_user(_base_config(), "mz", "password")

        assert info is not None
        assert info.groups == ["cn=bambuddy-operators,ou=groups,dc=test,dc=com"]

    def test_dedupes_group_found_via_both_memberuid_and_primary_gid(self, mock_ldap):
        """A user in the same group via BOTH memberUid and primary gidNumber should appear once."""
        user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
        group_entry = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")

        mock_ldap._search_fixture = {
            "(uid=mz)": [user_entry],
            "memberUid=mz": [group_entry],  # supplementary membership
            "gidNumber=10002": [group_entry],  # primary group — same DN
        }

        info = authenticate_ldap_user(_base_config(), "mz", "password")

        assert info.groups == ["cn=bambuddy-operators,ou=groups,dc=test,dc=com"]

    def test_case_insensitive_dedupe(self, mock_ldap):
        """DNs differing only in case should collapse to a single entry (LDAP DNs are case-insensitive)."""
        user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
        upper_dn = _MockEntry("CN=Bambuddy-Operators,OU=Groups,DC=Test,DC=Com")
        lower_dn = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")

        mock_ldap._search_fixture = {
            "(uid=mz)": [user_entry],
            "memberUid=mz": [upper_dn],
            "gidNumber=10002": [lower_dn],
        }

        info = authenticate_ldap_user(_base_config(), "mz", "password")

        assert len(info.groups) == 1
        # The first-seen casing (memberUid result) is kept.
        assert info.groups[0] == "CN=Bambuddy-Operators,OU=Groups,DC=Test,DC=Com"

    def test_no_gidnumber_skips_primary_search(self, mock_ldap):
        """User entries without a gidNumber attribute should not crash and should not issue the primary-gid query."""
        user_entry = _MockEntry("cn=tester,dc=test,dc=com", uid="tester")  # no gidNumber
        viewers_group = _MockEntry("cn=bambuddy-viewers,ou=groups,dc=test,dc=com")

        mock_ldap._search_fixture = {
            "(uid=tester)": [user_entry],
            "memberUid=tester": [viewers_group],
        }

        info = authenticate_ldap_user(_base_config(), "tester", "password")

        assert info is not None
        assert info.groups == ["cn=bambuddy-viewers,ou=groups,dc=test,dc=com"]
        # Ensure the primary-gidNumber search was never issued — verifying the guard works.
        service_conn = _MockConnection._instances[0]
        gidnumber_searches = [call for call in service_conn.search_calls if "gidNumber=" in call]
        assert gidnumber_searches == []


class TestDirectoryWithoutPosixGroupClass:
    """A directory whose published schema defines no posixGroup class (#2769).

    ldap3 fetches the schema at connect time (get_info=ALL) and validates object
    class names in a filter against it before building the request, so both POSIX
    group searches raise client-side and nothing reaches the server. lldap is the
    case in the wild: it puts posixAccount on every account it creates, which
    gives each user a gidNumber, but defines no group class beyond groupOfNames.
    Left uncaught the exception escaped authenticate_ldap_user and the login route
    reported it as "Incorrect username or password", so LDAP login was impossible.
    """

    def test_authenticates_and_keeps_memberof_groups(self, mock_ldap):
        """The reporter's setup: the mapped membership comes from memberOf, which
        is read off the user entry and never touches a posixGroup filter."""
        user_entry = _MockEntry(
            "uid=peter,ou=people,dc=fablab,dc=test",
            uid="peter",
            gidNumber=1001,  # lldap gives every account one
            memberOf=["cn=AAUStudents,ou=groups,dc=fablab,dc=test"],
        )
        mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
        mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"

        info = authenticate_ldap_user(_base_config(), "peter", "password")

        assert info is not None
        assert info.groups == ["cn=AAUStudents,ou=groups,dc=fablab,dc=test"]

    def test_authenticates_with_no_groups_at_all(self, mock_ldap):
        """No memberOf either. The user still gets in — auto-provisioning assigns
        the configured default group, which is the whole point of that setting."""
        user_entry = _MockEntry("uid=peter,ou=people,dc=fablab,dc=test", uid="peter", gidNumber=1001)
        mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
        mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"

        info = authenticate_ldap_user(_base_config(), "peter", "password")

        assert info is not None
        assert info.username == "peter"
        assert info.groups == []

    def test_abandons_the_primary_gid_search_after_the_first_rejection(self, mock_ldap):
        """Both filters name the same class, so once one is rejected the other
        cannot succeed. Attempting it would only produce a second identical
        exception to swallow."""
        user_entry = _MockEntry("uid=peter,ou=people,dc=fablab,dc=test", uid="peter", gidNumber=1001)
        mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
        mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"

        authenticate_ldap_user(_base_config(), "peter", "password")

        service_conn = _MockConnection._instances[0]
        posix_searches = [call for call in service_conn.search_calls if "posixGroup" in call]
        assert len(posix_searches) == 1
        assert "memberUid=peter" in posix_searches[0]

    def test_a_directory_that_defines_the_class_is_untouched(self, mock_ldap):
        """The guard must not cost a normal directory its POSIX groups — both
        searches still run and both results still land."""
        user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
        supplementary = _MockEntry("cn=bambuddy-viewers,ou=groups,dc=test,dc=com")
        primary = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")

        mock_ldap._search_fixture = {
            "(uid=mz)": [user_entry],
            "memberUid=mz": [supplementary],
            "gidNumber=10002": [primary],
        }

        info = authenticate_ldap_user(_base_config(), "mz", "password")

        assert info.groups == [
            "cn=bambuddy-viewers,ou=groups,dc=test,dc=com",
            "cn=bambuddy-operators,ou=groups,dc=test,dc=com",
        ]


# ---------------------------------------------------------------------------
# Manual provisioning helpers — search_ldap_users + lookup_ldap_user (#1298)
# ---------------------------------------------------------------------------


class TestSearchLdapUsers:
    """Admin directory search for the manual-provision flow."""

    def test_returns_empty_when_query_too_short(self, mock_ldap):
        """Queries under 2 chars must not hit the directory at all."""
        results = search_ldap_users(_base_config(), "a")
        assert results == []
        # No connection was opened — no Connection instance recorded.
        assert _MockConnection._instances == []

    def test_returns_empty_when_query_whitespace(self, mock_ldap):
        results = search_ldap_users(_base_config(), "   ")
        assert results == []
        assert _MockConnection._instances == []

    def test_filter_covers_all_common_attributes(self, mock_ldap):
        """The fixed OR filter must cover sAMAccountName, uid, mail, displayName, cn."""
        _MockConnection._search_fixture = {}  # any matching attr; empty result is fine
        search_ldap_users(_base_config(), "jdoe")

        assert len(_MockConnection._instances) == 1
        sent = _MockConnection._instances[0].search_calls[0]
        for attr in ("sAMAccountName=*jdoe*", "uid=*jdoe*", "mail=*jdoe*", "displayName=*jdoe*", "cn=*jdoe*"):
            assert attr in sent, f"filter missing {attr}: {sent}"

    def test_wildcard_in_query_is_escaped(self, mock_ldap):
        """A typed * in the query must not enumerate the whole directory."""
        _MockConnection._search_fixture = {}
        search_ldap_users(_base_config(), "j*")

        sent = _MockConnection._instances[0].search_calls[0]
        # _ldap_escape replaces * with \2a; the outer wildcards (from our filter)
        # must remain, but the user-supplied * must be escaped.
        assert "*j\\2a*" in sent

    def test_picks_samaccountname_first(self, mock_ldap):
        entry = _MockEntry(
            "cn=John Doe,dc=test,dc=com",
            sAMAccountName="jdoe",
            uid="jdoe-uid",
            mail="jdoe@test.com",
            displayName="John Doe",
            cn="John Doe",
        )
        _MockConnection._search_fixture = {"sAMAccountName=*jdoe*": [entry]}

        results = search_ldap_users(_base_config(), "jdoe")

        assert len(results) == 1
        assert isinstance(results[0], LDAPSearchResult)
        assert results[0].username == "jdoe"  # sAMAccountName preferred
        assert results[0].email == "jdoe@test.com"
        assert results[0].display_name == "John Doe"
        assert results[0].dn == "cn=John Doe,dc=test,dc=com"

    def test_falls_back_to_uid_when_no_samaccountname(self, mock_ldap):
        entry = _MockEntry("uid=alice,ou=people,dc=test,dc=com", uid="alice", cn="Alice")
        _MockConnection._search_fixture = {"uid=*alice*": [entry]}

        results = search_ldap_users(_base_config(), "alice")

        assert len(results) == 1
        assert results[0].username == "alice"

    def test_falls_back_to_cn_when_neither_samaccountname_nor_uid(self, mock_ldap):
        """Some OpenLDAP layouts only have cn — make sure we still surface them."""
        entry = _MockEntry("cn=Bob,ou=people,dc=test,dc=com", cn="Bob")
        _MockConnection._search_fixture = {"cn=*Bob*": [entry]}

        results = search_ldap_users(_base_config(), "Bob")

        assert len(results) == 1
        assert results[0].username == "Bob"

    def test_raises_when_service_bind_fails(self, mock_ldap, monkeypatch):
        """Bind failures must propagate so the route can return 503 instead of [] (which
        would look indistinguishable from 'no matches found' to the admin)."""

        class _BindFailConn(_MockConnection):
            def bind(self):
                raise RuntimeError("simulated bind failure")

        monkeypatch.setattr("backend.app.services.ldap_service.Connection", _BindFailConn)

        with pytest.raises(RuntimeError):
            search_ldap_users(_base_config(), "anyone")

    def test_connection_skips_client_side_attribute_validation(self, mock_ldap, monkeypatch):
        """OpenLDAP directories don't define sAMAccountName/displayName in their schema,
        so ldap3 would raise LDAPAttributeError client-side before sending the query
        — break the regression by asserting Connection is opened with check_names=False
        for directory search."""
        captured_kwargs: dict = {}

        class _CapturingConn(_MockConnection):
            def __init__(self, *args, **kwargs):
                captured_kwargs.update(kwargs)
                super().__init__(*args, **kwargs)

        monkeypatch.setattr("backend.app.services.ldap_service.Connection", _CapturingConn)

        search_ldap_users(_base_config(), "anyone")

        assert captured_kwargs.get("check_names") is False, (
            "search_ldap_users must open the connection with check_names=False — "
            "otherwise ldap3 rejects sAMAccountName/displayName on OpenLDAP schemas"
        )

    def test_requests_all_user_attributes_to_bypass_schema_check(self, mock_ldap):
        """ldap3's `build_attribute_selection` validates each named attribute against
        the server schema regardless of check_names; only the `*` wildcard is in
        its hard-coded exclusion list. So search_ldap_users MUST request `["*"]`
        — not the explicit AD-flavoured names — or OpenLDAP servers raise
        `LDAPAttributeError: invalid attribute type in attribute list: sAMAccountName`."""
        _MockConnection._search_fixture = {}
        search_ldap_users(_base_config(), "anyone")

        # The mock's search() captures search_filter in search_calls but not
        # attributes — so monkeypatch its signature briefly to capture both.
        # Easier: re-grep ldap3 here. The mock's search() accepts kwargs via
        # **kwargs; we just need to verify the attributes arg was the wildcard.
        sent_attrs = _MockConnection._instances[0].last_attrs  # set by patched search
        assert sent_attrs == ["*"], (
            f"Expected attributes=['*'] to bypass ldap3 schema validation; got {sent_attrs!r}. "
            "Explicit AD attribute names (sAMAccountName, displayName) make ldap3 throw on "
            "OpenLDAP directories whose schema doesn't define them."
        )


class TestLookupLdapUser:
    """Service-bind lookup used by the manual-provision route."""

    def test_returns_none_when_user_missing(self, mock_ldap):
        _MockConnection._search_fixture = {}  # nothing matches

        result = lookup_ldap_user(_base_config(), "nobody")

        assert result is None

    def test_returns_user_info_with_groups(self, mock_ldap):
        user_entry = _MockEntry(
            "cn=John Doe,dc=test,dc=com",
            uid="jdoe",
            mail="jdoe@test.com",
            displayName="John Doe",
            memberOf=["cn=ops,ou=groups,dc=test,dc=com", "cn=qa,ou=groups,dc=test,dc=com"],
        )
        _MockConnection._search_fixture = {"(uid=jdoe)": [user_entry]}

        info = lookup_ldap_user(_base_config(), "jdoe")

        assert info is not None
        assert info.username == "jdoe"
        assert info.email == "jdoe@test.com"
        assert info.display_name == "John Doe"
        assert set(info.groups) == {"cn=ops,ou=groups,dc=test,dc=com", "cn=qa,ou=groups,dc=test,dc=com"}

    def test_does_not_attempt_password_bind(self, mock_ldap):
        """lookup_ldap_user MUST NOT call the user-DN bind that authenticate_ldap_user
        does — admins are using their own session, not the LDAP user's password."""
        user_entry = _MockEntry("cn=jdoe,dc=test,dc=com", uid="jdoe")
        _MockConnection._search_fixture = {"(uid=jdoe)": [user_entry]}

        lookup_ldap_user(_base_config(), "jdoe")

        # authenticate_ldap_user creates TWO Connection objects (service + user-bind).
        # lookup_ldap_user must create only ONE.
        assert len(_MockConnection._instances) == 1

    def test_raises_when_service_bind_fails(self, mock_ldap, monkeypatch):
        class _BindFailConn(_MockConnection):
            def bind(self):
                raise RuntimeError("simulated bind failure")

        monkeypatch.setattr("backend.app.services.ldap_service.Connection", _BindFailConn)

        with pytest.raises(RuntimeError):
            lookup_ldap_user(_base_config(), "anyone")


# ---------------------------------------------------------------------------
# Group membership on directories where `*` doesn't return memberOf (#3197)
# ---------------------------------------------------------------------------

_USER_DN = "uid=tofm,ou=people,dc=example,dc=com"
_ADMINS_DN = "cn=bambuddy-admins,ou=groups,dc=example,dc=com"
_PEOPLE_BASE = "ou=people,dc=example,dc=com"


def _user_search_attrs(conn: _MockConnection) -> list | None:
    """The attribute list sent with the user search (the first search on the service connection)."""
    return conn.search_attrs[0]


def _member_searches(conn: _MockConnection) -> list[tuple[str, str | None]]:
    return [
        (flt, base)
        for flt, base in zip(conn.search_calls, conn.search_bases, strict=True)
        if "(member=" in flt or "(uniqueMember=" in flt
    ]


class TestMemberOfIsRequestedByName:
    """lldap fills in memberOf only when it is asked for by name, and OpenLDAP's
    memberof overlay makes it operational, so `*` alone returns no groups. The
    reporter's lldap user was a member of a mapped group and always got the
    default group instead."""

    def test_login_asks_for_memberof_when_the_schema_has_it(self, mock_ldap):
        mock_ldap.server_fixture = _MockServer(
            attribute_types=["uid", "memberOf"], object_classes=["groupOfUniqueNames"]
        )
        user_entry = _MockEntry(_USER_DN, uid="tofm", memberOf=[_ADMINS_DN])
        mock_ldap._search_fixture = {"(uid=tofm)": [user_entry]}

        info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        service_conn = _MockConnection._instances[0]
        assert _user_search_attrs(service_conn) == ["*", "memberOf"]
        assert info.groups == [_ADMINS_DN]

    def test_schema_match_is_case_insensitive(self, mock_ldap):
        """Schemas spell it memberof, memberOf or MemberOf; ldap3's schema dict ignores case."""
        mock_ldap.server_fixture = _MockServer(attribute_types=["memberof"], object_classes=[])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(), "tofm", "password")

        assert _user_search_attrs(_MockConnection._instances[0]) == ["*", "memberOf"]

    def test_groups_are_asked_as_well_as_memberof(self, mock_ldap):
        """OpenLDAP's memberof overlay tracks only the group class it was set up
        for (osixia's image: groupOfUniqueNames), so a groupOfNames group is
        missing from memberOf even though the schema has the attribute."""
        mock_ldap.server_fixture = _MockServer(
            attribute_types=["memberOf"],
            object_classes=["groupOfNames", "groupOfUniqueNames"],
            naming_contexts=["dc=example,dc=com"],
        )
        operators_dn = "cn=bambuddy-operators,ou=groups,dc=example,dc=com"
        mock_ldap._search_fixture = {
            "(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm", memberOf=[operators_dn])],
            f"(member={_USER_DN})": [_MockEntry(_ADMINS_DN), _MockEntry(operators_dn)],
        }

        info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        assert info.groups == [operators_dn, _ADMINS_DN]

    def test_no_group_side_search_on_active_directory(self, mock_ldap):
        """AD keeps memberOf complete, and its groups are objectClass=group, so a
        subtree search from the domain root would find nothing."""
        mock_ldap.server_fixture = _MockServer(
            attribute_types=["memberOf", "member"],
            object_classes=["group", "groupOfNames"],
            naming_contexts=["dc=example,dc=com"],
            active_directory=True,
        )
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm", memberOf=[_ADMINS_DN])]}

        info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        assert info.groups == [_ADMINS_DN]
        assert _member_searches(_MockConnection._instances[0]) == []

    def test_admin_lookup_asks_for_memberof_too(self, mock_ldap):
        mock_ldap.server_fixture = _MockServer(attribute_types=["memberOf"], object_classes=[])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm", memberOf=[_ADMINS_DN])]}

        info = lookup_ldap_user(_base_config(), "tofm")

        assert _user_search_attrs(_MockConnection._instances[0]) == ["*", "memberOf"]
        assert info.groups == [_ADMINS_DN]

    def test_memberof_not_requested_when_the_schema_lacks_it(self, mock_ldap):
        """ldap3 rejects a requested attribute the schema doesn't define before
        sending anything, even with check_names off. Asking anyway would make
        every login on such a directory fail."""
        mock_ldap.server_fixture = _MockServer(attribute_types=["uid", "cn"], object_classes=[])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(), "tofm", "password")

        assert _user_search_attrs(_MockConnection._instances[0]) == ["*"]

    def test_memberof_requested_when_the_server_publishes_no_schema(self, mock_ldap):
        """Without a schema ldap3 checks no names, and the server ignores an
        attribute it doesn't know."""
        mock_ldap.server_fixture = _MockServer()
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(), "tofm", "password")

        assert _user_search_attrs(_MockConnection._instances[0]) == ["*", "memberOf"]


class TestGroupsListingTheUserByDn:
    """Group membership asked from the group side. Plain OpenLDAP without the
    memberof overlay can answer it no other way."""

    def _server(self, object_classes=("groupOfNames", "groupOfUniqueNames"), naming_contexts=("dc=example,dc=com",)):
        return _MockServer(
            attribute_types=["uid", "cn", "member", "uniqueMember"],
            object_classes=list(object_classes),
            naming_contexts=list(naming_contexts),
        )

    def test_finds_a_group_outside_the_user_search_base(self, mock_ldap):
        """The reporter's layout: users under ou=people, groups under ou=groups.
        The group search starts at the naming context, not the user search base."""
        mock_ldap.server_fixture = self._server()
        mock_ldap._search_fixture = {
            "(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")],
            f"(member={_USER_DN})": [_MockEntry(_ADMINS_DN)],
        }

        info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        assert info.groups == [_ADMINS_DN]
        searches = _member_searches(_MockConnection._instances[0])
        assert len(searches) == 1
        flt, base = searches[0]
        assert base == "dc=example,dc=com"
        assert flt == (
            f"(|(&(objectClass=groupOfNames)(member={_USER_DN}))"
            f"(&(objectClass=groupOfUniqueNames)(uniqueMember={_USER_DN})))"
        )

    def test_only_classes_the_schema_defines_go_into_the_filter(self, mock_ldap):
        """Naming an undefined class raises client-side, the #2769 failure."""
        mock_ldap.server_fixture = self._server(object_classes=["groupOfUniqueNames"])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        (flt, _base) = _member_searches(_MockConnection._instances[0])[0]
        assert flt == f"(&(objectClass=groupOfUniqueNames)(uniqueMember={_USER_DN}))"

    def test_no_search_when_the_schema_has_neither_class(self, mock_ldap):
        mock_ldap.server_fixture = self._server(object_classes=["posixGroup"])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        info = authenticate_ldap_user(_base_config(), "tofm", "password")

        assert info.groups == []
        assert _member_searches(_MockConnection._instances[0]) == []

    def test_dn_is_escaped_in_the_filter(self, mock_ldap):
        """A DN may carry filter metacharacters (an escaped comma, a parenthesis)."""
        dn = r"cn=Doe\, John (ops),ou=people,dc=example,dc=com"
        mock_ldap.server_fixture = self._server(object_classes=["groupOfNames"])
        mock_ldap._search_fixture = {"(uid=jdoe)": [_MockEntry(dn, uid="jdoe")]}

        authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "jdoe", "password")

        (flt, _base) = _member_searches(_MockConnection._instances[0])[0]
        assert flt == r"(&(objectClass=groupOfNames)(member=cn=Doe\5c, John \28ops\29,ou=people,dc=example,dc=com))"

    def test_search_base_is_used_when_no_naming_context_contains_it(self, mock_ldap):
        mock_ldap.server_fixture = self._server(naming_contexts=["dc=other,dc=org"])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        (_flt, base) = _member_searches(_MockConnection._instances[0])[0]
        assert base == _PEOPLE_BASE

    def test_naming_context_match_is_on_whole_components(self, mock_ldap):
        """dc=ample,dc=com is not a suffix of ou=people,dc=example,dc=com."""
        mock_ldap.server_fixture = self._server(naming_contexts=["dc=ample,dc=com", "DC=Example,DC=Com"])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}

        authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        (_flt, base) = _member_searches(_MockConnection._instances[0])[0]
        assert base == "DC=Example,DC=Com"

    def test_dedupes_against_posix_groups(self, mock_ldap):
        """A group can be both a groupOfNames and a posixGroup (OpenLDAP rfc2307bis)."""
        mock_ldap.server_fixture = self._server(object_classes=["groupOfNames", "posixGroup"])
        mock_ldap._search_fixture = {
            "(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")],
            f"(member={_USER_DN})": [_MockEntry(_ADMINS_DN)],
            "memberUid=tofm": [_MockEntry(_ADMINS_DN.upper())],
        }

        info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")

        assert info.groups == [_ADMINS_DN]

    def test_a_failed_group_search_still_logs_the_user_in(self, mock_ldap, caplog):
        """The user gets the groups found by other means, and the log names the
        failure without the DN it may contain (#2681)."""

        class _GroupSearchFails(_MockConnection):
            def search(self, search_base=None, search_filter=None, **kwargs):
                if "(member=" in (search_filter or ""):
                    raise LDAPObjectClassError(f"size limit on {_USER_DN}")
                return super().search(search_base=search_base, search_filter=search_filter, **kwargs)

        import backend.app.services.ldap_service as ldap_service

        mock_ldap.server_fixture = self._server(object_classes=["groupOfNames"])
        mock_ldap._search_fixture = {"(uid=tofm)": [_MockEntry(_USER_DN, uid="tofm")]}
        original = ldap_service.Connection
        ldap_service.Connection = _GroupSearchFails
        try:
            with caplog.at_level("WARNING", logger="backend.app.services.ldap_service"):
                info = authenticate_ldap_user(_base_config(search_base=_PEOPLE_BASE), "tofm", "password")
        finally:
            ldap_service.Connection = original

        assert info is not None
        assert info.groups == []
        assert "LDAP group membership lookup failed (LDAPObjectClassError)" in caplog.text
        assert _USER_DN not in caplog.text


class TestStartTlsRefused:
    """lldap offers LDAPS only. Its answer to StartTLS is "Unsupported extended
    operation" plus the StartTLS OID, which the reporter had to decode."""

    def _refusing(self, error):
        class _Conn(_MockConnection):
            def start_tls(self, read_server_info=True):
                raise error

        return _Conn

    def test_connection_test_says_what_to_change(self, mock_ldap, monkeypatch):
        refusal = LDAPUnwillingToPerformResult(
            result=53,
            description="unwillingToPerform",
            message="Unsupported extended operation: 1.3.6.1.4.1.1466.20037",
            response_type="extendedResp",
        )
        monkeypatch.setattr("backend.app.services.ldap_service.Connection", self._refusing(refusal))

        ok, message = check_ldap_connection(_base_config(server_url="ldap://lldap:3890", security="starttls"))

        assert ok is False
        assert message == (
            "LDAP connection failed: the server refused StartTLS (unwillingToPerform). "
            "If it only offers LDAPS, choose LDAPS and use its ldaps:// URL and port"
        )

    def test_login_with_refused_starttls_fails_cleanly(self, mock_ldap, monkeypatch):
        refusal = LDAPUnwillingToPerformResult(result=53, description="unwillingToPerform")
        monkeypatch.setattr("backend.app.services.ldap_service.Connection", self._refusing(refusal))

        assert authenticate_ldap_user(_base_config(server_url="ldap://x", security="starttls"), "u", "p") is None

    @pytest.mark.parametrize(
        "error",
        [LDAPStartTLSError("wrap socket error: certificate verify failed"), LDAPSocketOpenError("reset")],
    )
    def test_tls_failures_keep_their_own_message(self, mock_ldap, monkeypatch, error):
        """Only a refusal by the server is reworded; a certificate problem is not a missing feature."""
        monkeypatch.setattr("backend.app.services.ldap_service.Connection", self._refusing(error))

        ok, message = check_ldap_connection(_base_config(server_url="ldap://x", security="starttls"))

        assert ok is False
        assert message == f"LDAP connection failed: {error}"
        assert "refused StartTLS" not in message

    def test_ldaps_never_sends_starttls(self, mock_ldap, monkeypatch):
        refusal = LDAPUnwillingToPerformResult(result=53, description="unwillingToPerform")
        monkeypatch.setattr("backend.app.services.ldap_service.Connection", self._refusing(refusal))

        ok, _message = check_ldap_connection(_base_config(server_url="ldaps://x:636", security="starttls"))

        assert ok is True

    def test_server_info_is_not_read_before_the_bind(self, mock_ldap):
        """ldap3's default re-reads the schema right after StartTLS, before any
        bind; Active Directory and Samba AD refuse that anonymous read with
        operationsError, so StartTLS never worked against them. bind() reads it
        once authenticated."""
        mock_ldap._search_fixture = {"(uid=u)": [_MockEntry("uid=u,dc=test,dc=com", uid="u")]}

        authenticate_ldap_user(_base_config(server_url="ldap://ad:389", security="starttls"), "u", "p")

        service_conn, user_conn = _MockConnection._instances
        assert service_conn.start_tls_read_server_info is False
        assert user_conn.start_tls_read_server_info is False