test_ldap_service.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706
  1. """Tests for LDAP authentication service (#794).
  2. Tests the pure logic functions in ldap_service.py:
  3. - Config parsing from settings dict
  4. - LDAP filter escaping (RFC 4515)
  5. - Group mapping resolution
  6. - LDAPConfig/LDAPUserInfo dataclass construction
  7. Network-dependent functions (authenticate_ldap_user, test_ldap_connection)
  8. are not tested here — they require a live LDAP server.
  9. """
  10. import pytest
  11. from ldap3.core.exceptions import LDAPObjectClassError
  12. from backend.app.services.ldap_service import (
  13. LDAPConfig,
  14. LDAPSearchResult,
  15. LDAPUserInfo,
  16. _ldap_escape,
  17. authenticate_ldap_user,
  18. lookup_ldap_user,
  19. parse_ldap_config,
  20. resolve_group_mapping,
  21. search_ldap_users,
  22. )
  23. class TestParseConfig:
  24. """Verify parse_ldap_config builds LDAPConfig from settings dict."""
  25. def test_returns_none_when_disabled(self):
  26. settings = {"ldap_enabled": "false", "ldap_server_url": "ldaps://example.com"}
  27. assert parse_ldap_config(settings) is None
  28. def test_returns_none_when_missing_enabled(self):
  29. settings = {"ldap_server_url": "ldaps://example.com"}
  30. assert parse_ldap_config(settings) is None
  31. def test_returns_none_when_no_server_url(self):
  32. settings = {"ldap_enabled": "true", "ldap_server_url": ""}
  33. assert parse_ldap_config(settings) is None
  34. def test_returns_none_when_server_url_whitespace(self):
  35. settings = {"ldap_enabled": "true", "ldap_server_url": " "}
  36. assert parse_ldap_config(settings) is None
  37. def test_parses_minimal_config(self):
  38. settings = {
  39. "ldap_enabled": "true",
  40. "ldap_server_url": "ldaps://ldap.example.com:636",
  41. }
  42. config = parse_ldap_config(settings)
  43. assert config is not None
  44. assert config.server_url == "ldaps://ldap.example.com:636"
  45. assert config.bind_dn == ""
  46. assert config.search_base == ""
  47. assert config.user_filter == "(sAMAccountName={username})"
  48. assert config.security == "starttls"
  49. assert config.group_mapping == {}
  50. assert config.auto_provision is False
  51. assert config.ca_cert_path == ""
  52. assert config.default_group == ""
  53. def test_parses_full_config(self):
  54. settings = {
  55. "ldap_enabled": "true",
  56. "ldap_server_url": "ldaps://ldap.example.com:636",
  57. "ldap_bind_dn": "cn=admin,dc=example,dc=com",
  58. "ldap_bind_password": "secret",
  59. "ldap_search_base": "ou=users,dc=example,dc=com",
  60. "ldap_user_filter": "(uid={username})",
  61. "ldap_security": "ldaps",
  62. "ldap_group_mapping": '{"cn=admins,dc=example,dc=com": "Administrators"}',
  63. "ldap_auto_provision": "true",
  64. "ldap_ca_cert_path": "/path/to/ca.pem",
  65. "ldap_default_group": "Viewers",
  66. }
  67. config = parse_ldap_config(settings)
  68. assert config is not None
  69. assert config.bind_dn == "cn=admin,dc=example,dc=com"
  70. assert config.bind_password == "secret"
  71. assert config.search_base == "ou=users,dc=example,dc=com"
  72. assert config.user_filter == "(uid={username})"
  73. assert config.security == "ldaps"
  74. assert config.group_mapping == {"cn=admins,dc=example,dc=com": "Administrators"}
  75. assert config.auto_provision is True
  76. assert config.ca_cert_path == "/path/to/ca.pem"
  77. assert config.default_group == "Viewers"
  78. def test_handles_invalid_group_mapping_json(self):
  79. settings = {
  80. "ldap_enabled": "true",
  81. "ldap_server_url": "ldaps://ldap.example.com",
  82. "ldap_group_mapping": "not valid json",
  83. }
  84. config = parse_ldap_config(settings)
  85. assert config is not None
  86. assert config.group_mapping == {}
  87. def test_handles_non_dict_group_mapping(self):
  88. settings = {
  89. "ldap_enabled": "true",
  90. "ldap_server_url": "ldaps://ldap.example.com",
  91. "ldap_group_mapping": '["not", "a", "dict"]',
  92. }
  93. config = parse_ldap_config(settings)
  94. assert config is not None
  95. assert config.group_mapping == {}
  96. def test_enabled_case_insensitive(self):
  97. settings = {"ldap_enabled": "True", "ldap_server_url": "ldaps://ldap.example.com"}
  98. assert parse_ldap_config(settings) is not None
  99. settings = {"ldap_enabled": "TRUE", "ldap_server_url": "ldaps://ldap.example.com"}
  100. assert parse_ldap_config(settings) is not None
  101. def test_strips_whitespace(self):
  102. settings = {
  103. "ldap_enabled": "true",
  104. "ldap_server_url": " ldaps://ldap.example.com ",
  105. "ldap_bind_dn": " cn=admin,dc=example,dc=com ",
  106. "ldap_search_base": " dc=example,dc=com ",
  107. "ldap_default_group": " Viewers ",
  108. }
  109. config = parse_ldap_config(settings)
  110. assert config.server_url == "ldaps://ldap.example.com"
  111. assert config.bind_dn == "cn=admin,dc=example,dc=com"
  112. assert config.search_base == "dc=example,dc=com"
  113. assert config.default_group == "Viewers"
  114. class TestLDAPEscape:
  115. """Verify RFC 4515 escaping for LDAP search filter values."""
  116. def test_plain_string(self):
  117. assert _ldap_escape("testuser") == "testuser"
  118. def test_escapes_backslash(self):
  119. assert _ldap_escape("test\\user") == "test\\5cuser"
  120. def test_escapes_asterisk(self):
  121. assert _ldap_escape("test*user") == "test\\2auser"
  122. def test_escapes_open_paren(self):
  123. assert _ldap_escape("test(user") == "test\\28user"
  124. def test_escapes_close_paren(self):
  125. assert _ldap_escape("test)user") == "test\\29user"
  126. def test_escapes_null(self):
  127. assert _ldap_escape("test\x00user") == "test\\00user"
  128. def test_escapes_multiple_chars(self):
  129. assert _ldap_escape("a*b(c)d\\e") == "a\\2ab\\28c\\29d\\5ce"
  130. def test_empty_string(self):
  131. assert _ldap_escape("") == ""
  132. class TestResolveGroupMapping:
  133. """Verify LDAP group DN to BamBuddy group name resolution."""
  134. def test_empty_mapping(self):
  135. assert resolve_group_mapping(["cn=admins,dc=example"], {}) == []
  136. def test_empty_groups(self):
  137. mapping = {"cn=admins,dc=example": "Administrators"}
  138. assert resolve_group_mapping([], mapping) == []
  139. def test_single_match(self):
  140. mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
  141. groups = ["cn=admins,dc=example,dc=com"]
  142. assert resolve_group_mapping(groups, mapping) == ["Administrators"]
  143. def test_multiple_matches(self):
  144. mapping = {
  145. "cn=admins,dc=example,dc=com": "Administrators",
  146. "cn=ops,dc=example,dc=com": "Operators",
  147. }
  148. groups = ["cn=admins,dc=example,dc=com", "cn=ops,dc=example,dc=com"]
  149. result = resolve_group_mapping(groups, mapping)
  150. assert set(result) == {"Administrators", "Operators"}
  151. def test_no_match(self):
  152. mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
  153. groups = ["cn=users,dc=example,dc=com"]
  154. assert resolve_group_mapping(groups, mapping) == []
  155. def test_case_insensitive_dn(self):
  156. mapping = {"CN=Admins,DC=Example,DC=Com": "Administrators"}
  157. groups = ["cn=admins,dc=example,dc=com"]
  158. assert resolve_group_mapping(groups, mapping) == ["Administrators"]
  159. def test_partial_match_not_matched(self):
  160. mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
  161. groups = ["cn=admins,dc=other,dc=com"]
  162. assert resolve_group_mapping(groups, mapping) == []
  163. def test_extra_groups_ignored(self):
  164. mapping = {"cn=admins,dc=example,dc=com": "Administrators"}
  165. groups = ["cn=admins,dc=example,dc=com", "cn=users,dc=example,dc=com", "cn=devs,dc=example,dc=com"]
  166. assert resolve_group_mapping(groups, mapping) == ["Administrators"]
  167. class TestDataclasses:
  168. """Verify dataclass construction."""
  169. def test_ldap_user_info(self):
  170. info = LDAPUserInfo(
  171. username="testuser",
  172. email="test@example.com",
  173. display_name="Test User",
  174. groups=["cn=admins,dc=example,dc=com"],
  175. )
  176. assert info.username == "testuser"
  177. assert info.email == "test@example.com"
  178. assert info.display_name == "Test User"
  179. assert info.groups == ["cn=admins,dc=example,dc=com"]
  180. def test_ldap_user_info_none_fields(self):
  181. info = LDAPUserInfo(username="testuser", email=None, display_name=None, groups=[])
  182. assert info.email is None
  183. assert info.display_name is None
  184. assert info.groups == []
  185. def test_ldap_config(self):
  186. config = LDAPConfig(
  187. server_url="ldaps://ldap.example.com:636",
  188. bind_dn="cn=admin,dc=example,dc=com",
  189. bind_password="secret",
  190. search_base="dc=example,dc=com",
  191. user_filter="(uid={username})",
  192. security="ldaps",
  193. group_mapping={"cn=admins": "Administrators"},
  194. auto_provision=True,
  195. ca_cert_path="",
  196. default_group="Viewers",
  197. )
  198. assert config.server_url == "ldaps://ldap.example.com:636"
  199. assert config.auto_provision is True
  200. assert config.default_group == "Viewers"
  201. # ---------------------------------------------------------------------------
  202. # Mocked authenticate_ldap_user group-discovery tests
  203. # ---------------------------------------------------------------------------
  204. # These tests mock ldap3.Connection to exercise the group-discovery logic in
  205. # authenticate_ldap_user without a live LDAP server. Added after a bug where
  206. # POSIX primary-group membership (via gidNumber) was ignored — see CHANGELOG.
  207. class _MockAttr:
  208. """Minimal stand-in for ldap3 Attribute objects.
  209. Supports str(), bool(), .value, .values, and iteration — the operations
  210. used by ldap_service against user entry attributes.
  211. """
  212. def __init__(self, value):
  213. self._value = value
  214. @property
  215. def value(self):
  216. return self._value
  217. @property
  218. def values(self):
  219. return self._value if isinstance(self._value, list) else [self._value]
  220. def __str__(self):
  221. return str(self._value)
  222. def __bool__(self):
  223. return bool(self._value)
  224. def __iter__(self):
  225. if isinstance(self._value, list):
  226. return iter(self._value)
  227. return iter([self._value])
  228. class _MockEntry:
  229. """Minimal stand-in for ldap3 Entry. Only attributes passed at construction exist."""
  230. def __init__(self, dn, **attrs):
  231. self.entry_dn = dn
  232. for key, val in attrs.items():
  233. setattr(self, key, _MockAttr(val))
  234. class _MockConnection:
  235. """Mock ldap3 Connection that returns pre-configured entries based on filter substring match.
  236. Every Connection() instance shares a class-level fixture dict so the service-account
  237. connection and the user-bind connection both see the same fake directory.
  238. """
  239. _search_fixture: dict[str, list] = {}
  240. _instances: list["_MockConnection"] = []
  241. # Filter substring that should raise LDAPObjectClassError instead of
  242. # searching, standing in for ldap3's client-side schema validation — it
  243. # rejects an object class the server's published schema doesn't define
  244. # before the request is ever built (#2769).
  245. _raise_object_class_error_on: str | None = None
  246. def __init__(self, *args, **kwargs):
  247. self.entries: list = []
  248. self.search_calls: list[str] = []
  249. self.last_attrs: list | None = None
  250. _MockConnection._instances.append(self)
  251. def open(self):
  252. pass
  253. def start_tls(self):
  254. pass
  255. def bind(self):
  256. return True
  257. def unbind(self):
  258. pass
  259. def search(self, search_base=None, search_filter=None, search_scope=None, attributes=None, **kwargs):
  260. # **kwargs absorbs ldap3 options like size_limit that the real client supports
  261. self.search_calls.append(search_filter or "")
  262. self.last_attrs = list(attributes) if attributes is not None else None
  263. needle = _MockConnection._raise_object_class_error_on
  264. if needle and needle in (search_filter or ""):
  265. raise LDAPObjectClassError(f"invalid class in objectClass attribute: {needle}")
  266. for needle, entries in _MockConnection._search_fixture.items():
  267. if needle in (search_filter or ""):
  268. self.entries = entries
  269. return True
  270. self.entries = []
  271. return True
  272. @pytest.fixture
  273. def mock_ldap(monkeypatch):
  274. """Patch Connection + _create_server in ldap_service so authenticate_ldap_user can run offline."""
  275. _MockConnection._search_fixture = {}
  276. _MockConnection._instances = []
  277. _MockConnection._raise_object_class_error_on = None
  278. monkeypatch.setattr("backend.app.services.ldap_service.Connection", _MockConnection)
  279. monkeypatch.setattr("backend.app.services.ldap_service._create_server", lambda config: None)
  280. return _MockConnection
  281. def _base_config(**overrides):
  282. """Build a minimal LDAPConfig for mocked tests."""
  283. defaults = {
  284. "server_url": "ldaps://test.example.com:636",
  285. "bind_dn": "cn=admin,dc=test,dc=com",
  286. "bind_password": "x",
  287. "search_base": "dc=test,dc=com",
  288. "user_filter": "(uid={username})",
  289. "security": "ldaps",
  290. "group_mapping": {},
  291. "auto_provision": False,
  292. "ca_cert_path": "",
  293. "default_group": "",
  294. }
  295. defaults.update(overrides)
  296. return LDAPConfig(**defaults)
  297. class TestAuthenticateLdapUserGroups:
  298. """Group-discovery behaviour in authenticate_ldap_user.
  299. Covers the POSIX primary gidNumber lookup and case-insensitive dedupe added
  300. to fix a bug where users whose role came from their primary group were
  301. authenticated without the correct group membership.
  302. """
  303. def test_primary_gidnumber_group_found(self, mock_ldap):
  304. """Regression: POSIX primary group (gidNumber match) must be included in the result."""
  305. user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
  306. operators_group = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")
  307. mock_ldap._search_fixture = {
  308. "(uid=mz)": [user_entry],
  309. "memberUid=mz": [], # no supplementary memberships
  310. "gidNumber=10002": [operators_group],
  311. }
  312. info = authenticate_ldap_user(_base_config(), "mz", "password")
  313. assert info is not None
  314. assert info.groups == ["cn=bambuddy-operators,ou=groups,dc=test,dc=com"]
  315. def test_dedupes_group_found_via_both_memberuid_and_primary_gid(self, mock_ldap):
  316. """A user in the same group via BOTH memberUid and primary gidNumber should appear once."""
  317. user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
  318. group_entry = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")
  319. mock_ldap._search_fixture = {
  320. "(uid=mz)": [user_entry],
  321. "memberUid=mz": [group_entry], # supplementary membership
  322. "gidNumber=10002": [group_entry], # primary group — same DN
  323. }
  324. info = authenticate_ldap_user(_base_config(), "mz", "password")
  325. assert info.groups == ["cn=bambuddy-operators,ou=groups,dc=test,dc=com"]
  326. def test_case_insensitive_dedupe(self, mock_ldap):
  327. """DNs differing only in case should collapse to a single entry (LDAP DNs are case-insensitive)."""
  328. user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
  329. upper_dn = _MockEntry("CN=Bambuddy-Operators,OU=Groups,DC=Test,DC=Com")
  330. lower_dn = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")
  331. mock_ldap._search_fixture = {
  332. "(uid=mz)": [user_entry],
  333. "memberUid=mz": [upper_dn],
  334. "gidNumber=10002": [lower_dn],
  335. }
  336. info = authenticate_ldap_user(_base_config(), "mz", "password")
  337. assert len(info.groups) == 1
  338. # The first-seen casing (memberUid result) is kept.
  339. assert info.groups[0] == "CN=Bambuddy-Operators,OU=Groups,DC=Test,DC=Com"
  340. def test_no_gidnumber_skips_primary_search(self, mock_ldap):
  341. """User entries without a gidNumber attribute should not crash and should not issue the primary-gid query."""
  342. user_entry = _MockEntry("cn=tester,dc=test,dc=com", uid="tester") # no gidNumber
  343. viewers_group = _MockEntry("cn=bambuddy-viewers,ou=groups,dc=test,dc=com")
  344. mock_ldap._search_fixture = {
  345. "(uid=tester)": [user_entry],
  346. "memberUid=tester": [viewers_group],
  347. }
  348. info = authenticate_ldap_user(_base_config(), "tester", "password")
  349. assert info is not None
  350. assert info.groups == ["cn=bambuddy-viewers,ou=groups,dc=test,dc=com"]
  351. # Ensure the primary-gidNumber search was never issued — verifying the guard works.
  352. service_conn = _MockConnection._instances[0]
  353. gidnumber_searches = [call for call in service_conn.search_calls if "gidNumber=" in call]
  354. assert gidnumber_searches == []
  355. class TestDirectoryWithoutPosixGroupClass:
  356. """A directory whose published schema defines no posixGroup class (#2769).
  357. ldap3 fetches the schema at connect time (get_info=ALL) and validates object
  358. class names in a filter against it before building the request, so both POSIX
  359. group searches raise client-side and nothing reaches the server. lldap is the
  360. case in the wild: it puts posixAccount on every account it creates, which
  361. gives each user a gidNumber, but defines no group class beyond groupOfNames.
  362. Left uncaught the exception escaped authenticate_ldap_user and the login route
  363. reported it as "Incorrect username or password", so LDAP login was impossible.
  364. """
  365. def test_authenticates_and_keeps_memberof_groups(self, mock_ldap):
  366. """The reporter's setup: the mapped membership comes from memberOf, which
  367. is read off the user entry and never touches a posixGroup filter."""
  368. user_entry = _MockEntry(
  369. "uid=peter,ou=people,dc=fablab,dc=test",
  370. uid="peter",
  371. gidNumber=1001, # lldap gives every account one
  372. memberOf=["cn=AAUStudents,ou=groups,dc=fablab,dc=test"],
  373. )
  374. mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
  375. mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"
  376. info = authenticate_ldap_user(_base_config(), "peter", "password")
  377. assert info is not None
  378. assert info.groups == ["cn=AAUStudents,ou=groups,dc=fablab,dc=test"]
  379. def test_authenticates_with_no_groups_at_all(self, mock_ldap):
  380. """No memberOf either. The user still gets in — auto-provisioning assigns
  381. the configured default group, which is the whole point of that setting."""
  382. user_entry = _MockEntry("uid=peter,ou=people,dc=fablab,dc=test", uid="peter", gidNumber=1001)
  383. mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
  384. mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"
  385. info = authenticate_ldap_user(_base_config(), "peter", "password")
  386. assert info is not None
  387. assert info.username == "peter"
  388. assert info.groups == []
  389. def test_abandons_the_primary_gid_search_after_the_first_rejection(self, mock_ldap):
  390. """Both filters name the same class, so once one is rejected the other
  391. cannot succeed. Attempting it would only produce a second identical
  392. exception to swallow."""
  393. user_entry = _MockEntry("uid=peter,ou=people,dc=fablab,dc=test", uid="peter", gidNumber=1001)
  394. mock_ldap._search_fixture = {"(uid=peter)": [user_entry]}
  395. mock_ldap._raise_object_class_error_on = "objectClass=posixGroup"
  396. authenticate_ldap_user(_base_config(), "peter", "password")
  397. service_conn = _MockConnection._instances[0]
  398. posix_searches = [call for call in service_conn.search_calls if "posixGroup" in call]
  399. assert len(posix_searches) == 1
  400. assert "memberUid=peter" in posix_searches[0]
  401. def test_a_directory_that_defines_the_class_is_untouched(self, mock_ldap):
  402. """The guard must not cost a normal directory its POSIX groups — both
  403. searches still run and both results still land."""
  404. user_entry = _MockEntry("cn=mz,dc=test,dc=com", uid="mz", gidNumber=10002)
  405. supplementary = _MockEntry("cn=bambuddy-viewers,ou=groups,dc=test,dc=com")
  406. primary = _MockEntry("cn=bambuddy-operators,ou=groups,dc=test,dc=com")
  407. mock_ldap._search_fixture = {
  408. "(uid=mz)": [user_entry],
  409. "memberUid=mz": [supplementary],
  410. "gidNumber=10002": [primary],
  411. }
  412. info = authenticate_ldap_user(_base_config(), "mz", "password")
  413. assert info.groups == [
  414. "cn=bambuddy-viewers,ou=groups,dc=test,dc=com",
  415. "cn=bambuddy-operators,ou=groups,dc=test,dc=com",
  416. ]
  417. # ---------------------------------------------------------------------------
  418. # Manual provisioning helpers — search_ldap_users + lookup_ldap_user (#1298)
  419. # ---------------------------------------------------------------------------
  420. class TestSearchLdapUsers:
  421. """Admin directory search for the manual-provision flow."""
  422. def test_returns_empty_when_query_too_short(self, mock_ldap):
  423. """Queries under 2 chars must not hit the directory at all."""
  424. results = search_ldap_users(_base_config(), "a")
  425. assert results == []
  426. # No connection was opened — no Connection instance recorded.
  427. assert _MockConnection._instances == []
  428. def test_returns_empty_when_query_whitespace(self, mock_ldap):
  429. results = search_ldap_users(_base_config(), " ")
  430. assert results == []
  431. assert _MockConnection._instances == []
  432. def test_filter_covers_all_common_attributes(self, mock_ldap):
  433. """The fixed OR filter must cover sAMAccountName, uid, mail, displayName, cn."""
  434. _MockConnection._search_fixture = {} # any matching attr; empty result is fine
  435. search_ldap_users(_base_config(), "jdoe")
  436. assert len(_MockConnection._instances) == 1
  437. sent = _MockConnection._instances[0].search_calls[0]
  438. for attr in ("sAMAccountName=*jdoe*", "uid=*jdoe*", "mail=*jdoe*", "displayName=*jdoe*", "cn=*jdoe*"):
  439. assert attr in sent, f"filter missing {attr}: {sent}"
  440. def test_wildcard_in_query_is_escaped(self, mock_ldap):
  441. """A typed * in the query must not enumerate the whole directory."""
  442. _MockConnection._search_fixture = {}
  443. search_ldap_users(_base_config(), "j*")
  444. sent = _MockConnection._instances[0].search_calls[0]
  445. # _ldap_escape replaces * with \2a; the outer wildcards (from our filter)
  446. # must remain, but the user-supplied * must be escaped.
  447. assert "*j\\2a*" in sent
  448. def test_picks_samaccountname_first(self, mock_ldap):
  449. entry = _MockEntry(
  450. "cn=John Doe,dc=test,dc=com",
  451. sAMAccountName="jdoe",
  452. uid="jdoe-uid",
  453. mail="jdoe@test.com",
  454. displayName="John Doe",
  455. cn="John Doe",
  456. )
  457. _MockConnection._search_fixture = {"sAMAccountName=*jdoe*": [entry]}
  458. results = search_ldap_users(_base_config(), "jdoe")
  459. assert len(results) == 1
  460. assert isinstance(results[0], LDAPSearchResult)
  461. assert results[0].username == "jdoe" # sAMAccountName preferred
  462. assert results[0].email == "jdoe@test.com"
  463. assert results[0].display_name == "John Doe"
  464. assert results[0].dn == "cn=John Doe,dc=test,dc=com"
  465. def test_falls_back_to_uid_when_no_samaccountname(self, mock_ldap):
  466. entry = _MockEntry("uid=alice,ou=people,dc=test,dc=com", uid="alice", cn="Alice")
  467. _MockConnection._search_fixture = {"uid=*alice*": [entry]}
  468. results = search_ldap_users(_base_config(), "alice")
  469. assert len(results) == 1
  470. assert results[0].username == "alice"
  471. def test_falls_back_to_cn_when_neither_samaccountname_nor_uid(self, mock_ldap):
  472. """Some OpenLDAP layouts only have cn — make sure we still surface them."""
  473. entry = _MockEntry("cn=Bob,ou=people,dc=test,dc=com", cn="Bob")
  474. _MockConnection._search_fixture = {"cn=*Bob*": [entry]}
  475. results = search_ldap_users(_base_config(), "Bob")
  476. assert len(results) == 1
  477. assert results[0].username == "Bob"
  478. def test_raises_when_service_bind_fails(self, mock_ldap, monkeypatch):
  479. """Bind failures must propagate so the route can return 503 instead of [] (which
  480. would look indistinguishable from 'no matches found' to the admin)."""
  481. class _BindFailConn(_MockConnection):
  482. def bind(self):
  483. raise RuntimeError("simulated bind failure")
  484. monkeypatch.setattr("backend.app.services.ldap_service.Connection", _BindFailConn)
  485. with pytest.raises(RuntimeError):
  486. search_ldap_users(_base_config(), "anyone")
  487. def test_connection_skips_client_side_attribute_validation(self, mock_ldap, monkeypatch):
  488. """OpenLDAP directories don't define sAMAccountName/displayName in their schema,
  489. so ldap3 would raise LDAPAttributeError client-side before sending the query
  490. — break the regression by asserting Connection is opened with check_names=False
  491. for directory search."""
  492. captured_kwargs: dict = {}
  493. class _CapturingConn(_MockConnection):
  494. def __init__(self, *args, **kwargs):
  495. captured_kwargs.update(kwargs)
  496. super().__init__(*args, **kwargs)
  497. monkeypatch.setattr("backend.app.services.ldap_service.Connection", _CapturingConn)
  498. search_ldap_users(_base_config(), "anyone")
  499. assert captured_kwargs.get("check_names") is False, (
  500. "search_ldap_users must open the connection with check_names=False — "
  501. "otherwise ldap3 rejects sAMAccountName/displayName on OpenLDAP schemas"
  502. )
  503. def test_requests_all_user_attributes_to_bypass_schema_check(self, mock_ldap):
  504. """ldap3's `build_attribute_selection` validates each named attribute against
  505. the server schema regardless of check_names; only the `*` wildcard is in
  506. its hard-coded exclusion list. So search_ldap_users MUST request `["*"]`
  507. — not the explicit AD-flavoured names — or OpenLDAP servers raise
  508. `LDAPAttributeError: invalid attribute type in attribute list: sAMAccountName`."""
  509. _MockConnection._search_fixture = {}
  510. search_ldap_users(_base_config(), "anyone")
  511. # The mock's search() captures search_filter in search_calls but not
  512. # attributes — so monkeypatch its signature briefly to capture both.
  513. # Easier: re-grep ldap3 here. The mock's search() accepts kwargs via
  514. # **kwargs; we just need to verify the attributes arg was the wildcard.
  515. sent_attrs = _MockConnection._instances[0].last_attrs # set by patched search
  516. assert sent_attrs == ["*"], (
  517. f"Expected attributes=['*'] to bypass ldap3 schema validation; got {sent_attrs!r}. "
  518. "Explicit AD attribute names (sAMAccountName, displayName) make ldap3 throw on "
  519. "OpenLDAP directories whose schema doesn't define them."
  520. )
  521. class TestLookupLdapUser:
  522. """Service-bind lookup used by the manual-provision route."""
  523. def test_returns_none_when_user_missing(self, mock_ldap):
  524. _MockConnection._search_fixture = {} # nothing matches
  525. result = lookup_ldap_user(_base_config(), "nobody")
  526. assert result is None
  527. def test_returns_user_info_with_groups(self, mock_ldap):
  528. user_entry = _MockEntry(
  529. "cn=John Doe,dc=test,dc=com",
  530. uid="jdoe",
  531. mail="jdoe@test.com",
  532. displayName="John Doe",
  533. memberOf=["cn=ops,ou=groups,dc=test,dc=com", "cn=qa,ou=groups,dc=test,dc=com"],
  534. )
  535. _MockConnection._search_fixture = {"(uid=jdoe)": [user_entry]}
  536. info = lookup_ldap_user(_base_config(), "jdoe")
  537. assert info is not None
  538. assert info.username == "jdoe"
  539. assert info.email == "jdoe@test.com"
  540. assert info.display_name == "John Doe"
  541. assert set(info.groups) == {"cn=ops,ou=groups,dc=test,dc=com", "cn=qa,ou=groups,dc=test,dc=com"}
  542. def test_does_not_attempt_password_bind(self, mock_ldap):
  543. """lookup_ldap_user MUST NOT call the user-DN bind that authenticate_ldap_user
  544. does — admins are using their own session, not the LDAP user's password."""
  545. user_entry = _MockEntry("cn=jdoe,dc=test,dc=com", uid="jdoe")
  546. _MockConnection._search_fixture = {"(uid=jdoe)": [user_entry]}
  547. lookup_ldap_user(_base_config(), "jdoe")
  548. # authenticate_ldap_user creates TWO Connection objects (service + user-bind).
  549. # lookup_ldap_user must create only ONE.
  550. assert len(_MockConnection._instances) == 1
  551. def test_raises_when_service_bind_fails(self, mock_ldap, monkeypatch):
  552. class _BindFailConn(_MockConnection):
  553. def bind(self):
  554. raise RuntimeError("simulated bind failure")
  555. monkeypatch.setattr("backend.app.services.ldap_service.Connection", _BindFailConn)
  556. with pytest.raises(RuntimeError):
  557. lookup_ldap_user(_base_config(), "anyone")