From 8f1eba8469820ac4a5084455a8ed3fe7f8025004 Mon Sep 17 00:00:00 2001 From: Jan-Kazlouski-elastic Date: Wed, 9 Sep 2026 13:58:18 +0300 Subject: [PATCH] fix(outlook): retry LDAP user searches with fresh connections (#4441) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of https://github.com/elastic/sdh-search/issues/1898 ## Problem On large on-prem Exchange deployments the Outlook Server connector enumerates users from Active Directory via LDAP before opening each mailbox over EWS. `ExchangeUsers` kept a single cached `ldap3` connection for both the normal-user and admin-user searches. `get_user_accounts()` consumes users lazily: it yields each normal user, the sync processes that mailbox (often for hours), then continues. The admin-user LDAP search can therefore run days after the first bind. AD or intermediate network gear often drops the idle TCP session, and the next search fails with errors such as: ``` UsersFetchFailed: Something went wrong while fetching users. Error: socket sending error[Errno 104] Connection reset by peer ``` Unlike the Office365 Graph path, LDAP had no retry layer. A single transient failure aborted the entire sync. A customer on SDH #1898 hit this after indexing 4.2M+ documents over ~45 hours. ## Fix - Replace the cached LDAP connection with a fresh `Connection` per search. - Add `_ldap_search()` with the existing `@retryable` settings (`RETRIES`, `RETRY_INTERVAL`, exponential backoff) and `skipped_exceptions=[UsersFetchFailed]` so empty LDAP results are not retried. - Unbind after each search attempt. - Refactor `_fetch_normal_users` / `_fetch_admin_users` to call the shared helper. ## Checklists #### Pre-Review Checklist - [x] this PR does NOT contain credentials of any kind, such as API keys or username/passwords (double check `config.yml.example`) - [x] this PR has a meaningful title - [x] this PR links to all relevant github issues that it fixes or partially addresses - [x] if there is no GH issue, please create it. Each PR should have a link to an issue - [x] this PR has a thorough description - [x] Covered the changes with automated tests - [x] Tested the changes locally - [ ] Added a label for each target release version (example: `v7.13.2`, `v7.14.0`, `v8.0.0`) - [ ] For bugfixes: backport safely to all minor branches still receiving patch releases - [x] Considered corresponding documentation changes - [x] Contributed any configuration settings changes to the configuration reference - [x] if you added or changed Rich Configurable Fields for a Native Connector, you made a corresponding PR in [Kibana](https://github.com/elastic/kibana/blob/main/packages/kbn-search-connectors/types/native_connectors.ts) #### Changes Requiring Extra Attention - [ ] Security-related changes (encryption, TLS, SSRF, etc) - [ ] New external service dependencies added. ## Related Pull Requests * https://github.com/elastic/connectors/pull/4406 — skip accounts with non-primary SMTP addresses * https://github.com/elastic/connectors/pull/4078 — skip AD users without a valid `mail` attribute * https://github.com/elastic/connectors/pull/4094 — per-account EWS error handling and TLS hardening ## Release Note **Outlook Server connector**: LDAP user enumeration now opens a fresh Active Directory connection per search and retries transient network failures instead of aborting long syncs when the admin-user query runs on a stale connection. Made with [Cursor](https://cursor.com) --------- Co-authored-by: Cursor Co-authored-by: Elastic Machine --- .../connectors/sources/outlook/client.py | 57 +++++++++--------- .../tests/sources/test_outlook.py | 60 +++++++++++++++++++ 2 files changed, 90 insertions(+), 27 deletions(-) diff --git a/app/connectors_service/connectors/sources/outlook/client.py b/app/connectors_service/connectors/sources/outlook/client.py index b6c3e1d2e..c66682511 100644 --- a/app/connectors_service/connectors/sources/outlook/client.py +++ b/app/connectors_service/connectors/sources/outlook/client.py @@ -145,8 +145,7 @@ def __init__( self.ssl_enabled = ssl_enabled self.ssl_ca = ssl_ca - @cached_property - def _create_connection(self): + def _create_ldap_connection(self): return Connection( server=self.ad_server, user=self.user, @@ -155,47 +154,51 @@ def _create_connection(self): auto_bind=True, # pyright: ignore ) - async def close(self): - pass - - def _fetch_normal_users(self, search_query): + @retryable( + retries=RETRIES, + interval=RETRY_INTERVAL, + strategy=RetryStrategy.EXPONENTIAL_BACKOFF, + skipped_exceptions=[UsersFetchFailed], + ) + def _ldap_search(self, search_query, search_filter): + connection = self._create_ldap_connection() try: - has_value_for_normal_users, _, response, _ = self._create_connection.search( + has_value, _, response, _ = connection.search( search_query, - SEARCH_FILTER_FOR_NORMAL_USERS, + search_filter, attributes=["mail"], ) - if not has_value_for_normal_users: + if not has_value: msg = "Error while fetching users from Exchange Active Directory." raise UsersFetchFailed(msg) - for user in response: - yield user + return response + finally: + try: + connection.unbind() + except Exception as exc: + logger.debug("Failed to unbind LDAP connection: %s", exc) + async def close(self): + pass + + def _fetch_normal_users(self, search_query): + try: + for user in self._ldap_search(search_query, SEARCH_FILTER_FOR_NORMAL_USERS): + yield user + except UsersFetchFailed: + raise except Exception as e: msg = f"Something went wrong while fetching users. Error: {e}" raise UsersFetchFailed(msg) from e def _fetch_admin_users(self, search_query): try: - ( - has_value_for_admin_users, - _, - response_for_admin, - _, - ) = self._create_connection.search( - search_query, - SEARCH_FILTER_FOR_ADMIN, - attributes=["mail"], - ) - - if not has_value_for_admin_users: - msg = "Error while fetching users from Exchange Active Directory." - raise UsersFetchFailed(msg) - - for user in response_for_admin: + for user in self._ldap_search(search_query, SEARCH_FILTER_FOR_ADMIN): yield user + except UsersFetchFailed: + raise except Exception as e: msg = f"Something went wrong while fetching users. Error: {e}" raise UsersFetchFailed(msg) from e diff --git a/app/connectors_service/tests/sources/test_outlook.py b/app/connectors_service/tests/sources/test_outlook.py index ab8573a8d..3deccd08e 100644 --- a/app/connectors_service/tests/sources/test_outlook.py +++ b/app/connectors_service/tests/sources/test_outlook.py @@ -732,6 +732,66 @@ async def test_fetch_admin_users(mock_connection): assert users == ["test.user@gmail.com", "dummy.user@gmail.com"] +@patch("connectors.utils.time_to_sleep_between_retries", return_value=0) +@patch("connectors.sources.outlook.client.Connection") +def test_ldap_search_retries_on_transient_error(mock_connection, _mock_sleep): + mock_connection_instance = mock_connection.return_value + mock_connection_instance.search.side_effect = [ + OSError("Connection reset by peer"), + (True, None, ["user@example.com"], None), + ] + + exchange_users = ExchangeUsers( + ad_server="ad.example.com", + domain="example.com", + exchange_server="exchange.example.com", + user="user", + password="password", + ssl_enabled=False, + ssl_ca=None, + ) + response = exchange_users._ldap_search("search_query", "filter") + assert list(response) == ["user@example.com"] + assert mock_connection.call_count == 2 + assert mock_connection_instance.search.call_count == 2 + assert mock_connection_instance.unbind.call_count == 2 + + +@patch("connectors.sources.outlook.client.Connection") +def test_exchange_ldap_search_uses_fresh_connection_per_search(mock_connection): + normal_user = {"mail": "normal@example.com"} + admin_user = {"mail": "admin@example.com"} + + first_connection = MagicMock() + first_connection.search.return_value = (True, None, [normal_user], None) + second_connection = MagicMock() + second_connection.search.return_value = (True, None, [admin_user], None) + mock_connection.side_effect = [first_connection, second_connection] + + exchange_users = ExchangeUsers( + ad_server="ad.example.com", + domain="example.com", + exchange_server="exchange.example.com", + user="user", + password="password", + ssl_enabled=False, + ssl_ca=None, + ) + + users = asyncio.run(_collect_exchange_users(exchange_users)) + assert users == [normal_user, admin_user] + assert mock_connection.call_count == 2 + first_connection.search.assert_called_once() + second_connection.search.assert_called_once() + + +async def _collect_exchange_users(exchange_users): + users = [] + async for user in exchange_users.get_users(): + users.append(user) + return users + + @pytest.mark.asyncio @pytest.mark.parametrize( "attachment, expected_content",