From b22888c530458ce9871e55017255188816f93410 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CJan-Kazlouski-elastic=E2=80=9C?= Date: Fri, 4 Sep 2026 18:22:29 +0300 Subject: [PATCH 1/3] fix(outlook): retry LDAP user searches with fresh connections On long on-prem Exchange syncs the admin-user LDAP query can run days after the normal-user search. Reusing a cached ldap3 connection caused Connection reset by peer and aborted the sync. Use a new connection per search, unbind after each attempt, and retry transient failures with the same backoff settings as Office365Users. Co-authored-by: Cursor --- .../connectors/sources/outlook/client.py | 54 ++++++++--------- .../tests/sources/test_outlook.py | 59 +++++++++++++++++++ 2 files changed, 86 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..99d08fa90 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,48 @@ 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: + connection.unbind() + + 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..e8f3a67b3 100644 --- a/app/connectors_service/tests/sources/test_outlook.py +++ b/app/connectors_service/tests/sources/test_outlook.py @@ -732,6 +732,65 @@ 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 + + +@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", From dfce859941e45ce96f864991cd7e4c6243eda0b3 Mon Sep 17 00:00:00 2001 From: Elastic Machine Date: Fri, 4 Sep 2026 15:25:54 +0000 Subject: [PATCH 2/3] Update NOTICE.txt --- app/connectors_service/NOTICE.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/connectors_service/NOTICE.txt b/app/connectors_service/NOTICE.txt index f24893e0b..1c0ba0ae1 100644 --- a/app/connectors_service/NOTICE.txt +++ b/app/connectors_service/NOTICE.txt @@ -1750,7 +1750,7 @@ SOFTWARE. anyio -4.14.2 +4.15.0 MIT The MIT License (MIT) @@ -4241,7 +4241,7 @@ Apache Software License google-auth -2.57.0 +2.57.1 Apache Software License Apache License Version 2.0, January 2004 From a3d3add6f9b27b617dfaf4d696f5f690b7f5545b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CJan-Kazlouski-elastic=E2=80=9C?= Date: Sat, 5 Sep 2026 01:06:17 +0300 Subject: [PATCH 3/3] fix(outlook): harden LDAP cleanup and test unbind on retry Wrap LDAP unbind in a defensive try/except so cleanup failures do not mask search errors or block retries, and assert unbind runs on each retry attempt in the transient-error test. --- app/connectors_service/connectors/sources/outlook/client.py | 5 ++++- app/connectors_service/tests/sources/test_outlook.py | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/app/connectors_service/connectors/sources/outlook/client.py b/app/connectors_service/connectors/sources/outlook/client.py index 99d08fa90..c66682511 100644 --- a/app/connectors_service/connectors/sources/outlook/client.py +++ b/app/connectors_service/connectors/sources/outlook/client.py @@ -175,7 +175,10 @@ def _ldap_search(self, search_query, search_filter): return response finally: - connection.unbind() + try: + connection.unbind() + except Exception as exc: + logger.debug("Failed to unbind LDAP connection: %s", exc) async def close(self): pass diff --git a/app/connectors_service/tests/sources/test_outlook.py b/app/connectors_service/tests/sources/test_outlook.py index e8f3a67b3..3deccd08e 100644 --- a/app/connectors_service/tests/sources/test_outlook.py +++ b/app/connectors_service/tests/sources/test_outlook.py @@ -754,6 +754,7 @@ def test_ldap_search_retries_on_transient_error(mock_connection, _mock_sleep): 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")