fix(outlook): skip mailbox-less accounts and stop SSL misconfig from aborting sync (#4085)

## Relates to elastic/sdh-search#1898

Fixes two independent failure modes in the Outlook Server connector that
caused an entire sync job to abort when a single account or the SSL
configuration was misconfigured. The guiding principle of this branch:
**per-account problems skip that account and continue; connection-wide
problems fail the sync loudly** (so they cannot silently empty the
index).

**Issue A — `ErrorNonExistentMailbox` kills the whole sync**

When iterating over AD user accounts, `get_mails()` accesses
`account.inbox`, which lazily resolves `account.root` via an EWS
`GetFolder` call. For AD users whose SMTP address has no associated
Exchange mailbox (a valid LDAP entry but no mailbox provisioned),
`exchangelib` raises `ErrorNonExistentMailbox`. With no exception
handling around the per-account block, this propagated up and terminated
the sync, losing all remaining accounts. This is the still-open part of
#2931.

**Fix:** the per-account block in `get_docs()` is wrapped in `try/except
ErrorNonExistentMailbox`. A missing mailbox is specific to a single
account, so the account is skipped with a warning and the sync continues
with the remaining accounts.

**Issue B — `NO_CERTIFICATE_OR_CRL_FOUND` SSL crash**

When `ssl_enabled=True` but no certificate is provided, `ssl_ca` is set
to `""`. The old code wrote that empty string to the cert file and still
selected `RootCAAdapter`. When urllib3 later called
`context.load_verify_locations()` on the empty file it raised
`NO_CERTIFICATE_OR_CRL_FOUND`, aborting the sync. The error occurs
inside the HTTP layer (not inside `cert_verify()`), so `RootCAAdapter`'s
own `try/except` never caught it.

**Fix:** in `ExchangeUsers.get_user_accounts()`, the cert file is
written and `RootCAAdapter` is selected only when `ssl_ca` is actually
populated. When SSL is enabled but no certificate is supplied, the
connector falls back to `NoVerifyHTTPAdapter` and logs a clear warning
instead of crashing.

### Why connection-wide SSL errors are NOT skipped per account

An earlier iteration of this branch also caught a custom `SSLFailed`
exception inside the per-account loop to "skip" accounts on SSL
problems. That approach was removed because it was both ineffective and
dangerous:

- **Ineffective:** `SSLFailed` is only raised inside
`RootCAAdapter.cert_verify`, and `requests`' `cert_verify` does not
actually load the certificate — it only records the CA path. A genuinely
bad/expired cert fails later, during the TLS handshake, where
`exchangelib` catches the underlying `requests.exceptions.SSLError` and
re-raises it as `exchangelib.errors.TransportError` (and explicitly does
**not** retry it). So `SSLFailed` was never raised in practice and the
`except SSLFailed` clause was dead code.
- **Dangerous:** an SSL/connection failure affects *every* account, not
one. Silently skipping all accounts would produce an empty but
"successful" sync, which the framework interprets as "all documents
deleted" — wiping previously indexed data.

Therefore only `ErrorNonExistentMailbox` (genuinely per-account) is
skipped. Connection-wide failures — TLS errors surfaced as
`TransportError`, and any other unexpected error — propagate and abort
the sync loudly, so the misconfiguration is surfaced to the operator and
existing indexed data is preserved.

## Checklists

#### Pre-Review Checklist
- [ ] 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
- [ ] 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
- [x] 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
- [ ] Considered corresponding documentation changes
- [ ] Contributed any configuration settings changes to the
configuration reference
- [ ] 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

- [x] Security-related changes (encryption, TLS, SSRF, etc) — the SSL
cert fallback behaviour changes: `ssl_enabled=True` with no cert now
falls back to `NoVerifyHTTPAdapter` (unverified connections) instead of
crashing. A genuinely invalid/expired certificate now fails the sync
loudly rather than being silently skipped. Reviewers should confirm the
no-cert fallback posture is desired.

## Related Pull Requests

* Partially addresses #2931

## Release Note

**Outlook Server connector**: sync jobs no longer abort when an Active
Directory user has a valid SMTP address but no associated Exchange
mailbox (`ErrorNonExistentMailbox`). The affected account is now skipped
with a warning and the sync continues. Additionally, configuring SSL
without providing a certificate no longer crashes with
`NO_CERTIFICATE_OR_CRL_FOUND`; the connector falls back to unverified
connections and logs a clear warning. Genuine certificate/connection
errors still fail the sync loudly rather than silently emptying the
index.

Made with [Cursor](https://cursor.com)

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Jan-Kazlouski-elastic

app/connectors_service/connectors/sources/outlook/datasource.py

@@ -4,14 +4,16 @@
 # you may not use this file except in compliance with the Elastic License 2.0.
 #
 
+import ssl
 from copy import copy
 from functools import cached_property, partial
 
-from connectors_sdk.source import BaseDataSource
+from connectors_sdk.source import BaseDataSource, ConfigurableFieldValueError
 from connectors_sdk.utils import (
     hash_id,
     iso_utc,
 )
+from exchangelib.errors import ErrorNonExistentMailbox
 
 from connectors.access_control import ACCESS_CONTROL, es_access_control_query
 from connectors.sources.outlook.client import OutlookClient
@@ -294,6 +296,41 @@ def get_default_configuration(cls):
             },
         }
 
+    async def validate_config(self):
+        """Validate the configuration and the SSL certificate content.
+
+        The base field checks only confirm the certificate is present; this also
+        confirms it actually loads, so a bad certificate fails here instead of
+        mid-sync with an opaque SSL error.
+
+        Raises:
+            ConfigurableFieldValueError: if SSL is enabled for an Exchange server
+                source but the certificate is not a loadable PEM certificate.
+        """
+        await super().validate_config()
+        self._validate_ssl_certificate()
+
+    def _validate_ssl_certificate(self):
+        if (
+            self.configuration["data_source"] != OUTLOOK_SERVER
+            or not self.configuration["ssl_enabled"]
+        ):
+            return
+
+        # Load the exact PEM string used at sync time through the same OpenSSL
+        # loader: load_verify_locations raises ssl.SSLError for malformed content
+        # and ValueError for empty data.
+        try:
+            ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT).load_verify_locations(
+                cadata=self.client.ssl_ca
+            )
+        except (ssl.SSLError, ValueError) as exception:
+            msg = (
+                "The provided SSL certificate is not valid. Provide a valid "
+                "PEM-encoded certificate."
+            )
+            raise ConfigurableFieldValueError(msg) from exception
+
     def _dls_enabled(self):
         """Check if document level security is enabled. This method checks whether document level security (DLS) is enabled based on the provided configuration.
 
@@ -562,24 +599,34 @@ async def get_docs(self, filtering=None):
         """
         async for account in self.client._get_user_instance.get_user_accounts():
             timezone = account.default_timezone or DEFAULT_TIMEZONE
+            try:
+                async for mail in self._fetch_mails(account=account, timezone=timezone):
+                    yield mail
 
-            async for mail in self._fetch_mails(account=account, timezone=timezone):
-                yield mail
-
-            async for contact in self._fetch_contacts(
-                account=account, timezone=timezone
-            ):
-                yield contact
+                async for contact in self._fetch_contacts(
+                    account=account, timezone=timezone
+                ):
+                    yield contact
 
-            async for task in self._fetch_tasks(account=account, timezone=timezone):
-                yield task
+                async for task in self._fetch_tasks(account=account, timezone=timezone):
+                    yield task
 
-            async for calendar in self._fetch_calendars(
-                account=account, timezone=timezone
-            ):
-                yield calendar
+                async for calendar in self._fetch_calendars(
+                    account=account, timezone=timezone
+                ):
+                    yield calendar
 
-            async for child_calendar in self._fetch_child_calendars(
-                account=account, timezone=timezone
-            ):
-                yield child_calendar
+                async for child_calendar in self._fetch_child_calendars(
+                    account=account, timezone=timezone
+                ):
+                    yield child_calendar
+            except ErrorNonExistentMailbox:
+                # A missing mailbox is specific to this account, so skip it and
+                # keep syncing the rest. Connection-wide failures (e.g. TLS
+                # errors) are intentionally not caught here: they affect every
+                # account, and silently skipping them would yield an empty but
+                # "successful" sync that deletes previously indexed documents.
+                self._logger.warning(
+                    f"Skipping account {account.primary_smtp_address}: "
+                    "the SMTP address has no associated mailbox."
+                )

app/connectors_service/connectors/utils.py

@@ -558,11 +558,18 @@ def is_expired(expires_at):
 
 
 def get_pem_format(key, postfix="-----END CERTIFICATE-----"):
-    """Convert key into PEM format.
+    """Convert a key/certificate into PEM format.
+
+    Handles both formats users provide:
+
+    * Single-line, where PEM newlines were replaced by spaces: reflowed back
+      into newlines.
+    * Already multi-line PEM: only whitespace-normalized, since reflowing would
+      break the space inside the BEGIN/END markers.
 
     Args:
-        key (str): Key in raw format.
-        postfix (str): Certificate footer.
+        key (str): Key/certificate in raw format.
+        postfix (str): PEM footer used to detect single-line blocks.
 
     Returns:
         string: PEM format
@@ -574,6 +581,13 @@ def get_pem_format(key, postfix="-----END CERTIFICATE-----"):
                     PrivateKey
                     -----END PRIVATE KEY-----"
     """
+    key = key.strip()
+
+    # Already multi-line: normalize whitespace instead of reflowing, which would
+    # break the space inside the BEGIN/END markers.
+    if "\n" in key:
+        return "\n".join(line.strip() for line in key.splitlines() if line.strip())
+
     pem_format = ""
     reverse_split = postfix.count(" ")
     if key.count(postfix) == 1:

app/connectors_service/tests/sources/test_outlook.py

@@ -12,7 +12,11 @@
 import pytest
 from aiohttp import StreamReader
 from connectors_sdk.source import ConfigurableFieldValueError
-from exchangelib.errors import ErrorFolderNotFound
+from exchangelib.errors import (
+    ErrorFolderNotFound,
+    ErrorNonExistentMailbox,
+    TransportError,
+)
 
 from connectors.sources.outlook import OutlookDataSource
 from connectors.sources.outlook.client import (
@@ -770,6 +774,70 @@ async def test_exchange_get_user_accounts_normalizes_ldap_mail_list(mock_account
         )
 
 
+# Real self-signed certificate in the single-line form the connector receives.
+# load_verify_locations ignores validity dates, so it never expires for tests.
+VALID_SSL_CERTIFICATE = (
+    "-----BEGIN CERTIFICATE----- "
+    "MIICsjCCAZqgAwIBAgIUDznyN9v5Tk8muCxnL/Z2EFwtcBwwDQYJKoZIhvcNAQELBQAwEjEQMA4GA1UEAwwHdGVzdC1jYTAgFw0wMDAxMDEwMDAwMDBaGA8yMDk5MDEwMTAwMDAwMFowEjEQMA4GA1UEAwwHdGVzdC1jYTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKmQAqm3+hDpc9+OzTjhY4W/AASWa41qyeuKNL+K8kA6oh9TmT20YhPikxPzQCUxp/prm9pi9eym5VLh2GhNCCE8LR+TsrwZr2MpYGZph1Y/y4U5PVNZCOboCee44F/6f8huYtHRPSrOC1OHehvMwdfAC63MueN6oBtxIIOwktxlkuBbK5wY97QlY/utxMa72APdUh3TAyzA6GWum7rLvEafj1v7WRpJWkTpklFXhaGVm4u/SWeFiMfgIK+ciJgT04k0qbk8APwuPmLR5VmUNyMDOgLMtSLu9sbntVv+eLoAAiOFLk5ZpHs0Q8UPANdNMV03tgxaDnvtgzh7W0Qgvo8CAwEAATANBgkqhkiG9w0BAQsFAAOCAQEAGPI/7KUIjHsNuRHUALIRtNVlhD80gdzKN27IEFLTu/jbiNEIGY59oV0qvx+iCPrLTLDnJkxHlnwApwB2WulXNg7+nYGHPP03jSLXKA+61GAN/ghPULl1DcA5Q+gunhPA4ITyqOr70i/3fphSXWjWfcX8hcym3pDcKzPIY3wV+dVeVdRdi9C1cTRuZ7zh2Chm7e4vM1SagLybMA4F8yckPJsRdVV5hZ+W6cI1H9fhjq/G1N0TyH4wG3FffRVniYVgAxY9m9RgMiQ5qCuc2PdktO7ovmNybijVG1aLVcHcYAS285f4JnZPIAJJMvCvW0NXDDphBNQPG5Nt1PHVgzUiNA== "
+    "-----END CERTIFICATE-----"
+)
+
+
+@pytest.mark.asyncio
+async def test_validate_config_raises_when_ssl_enabled_without_certificate():
+    # SSL enabled without a certificate must be rejected up front, not silently
+    # downgraded or failed mid-sync.
+    async with create_outlook_source(
+        data_source=OUTLOOK_SERVER,
+        username="foo.bar@gmail.com",
+        password="abc@123",
+        exchange_server="127.0.0.1",
+        domain="gmail.com",
+        active_directory_server="127.0.0.1",
+        ssl_enabled=True,
+        ssl_ca="",
+    ) as source:
+        with pytest.raises(ConfigurableFieldValueError) as exc_info:
+            await source.validate_config()
+
+        assert "SSL certificate" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_validate_config_raises_when_certificate_is_invalid():
+    # A present but unloadable certificate would otherwise crash mid-sync with an
+    # opaque X509 error.
+    async with create_outlook_source(
+        data_source=OUTLOOK_SERVER,
+        username="foo.bar@gmail.com",
+        password="abc@123",
+        exchange_server="127.0.0.1",
+        domain="gmail.com",
+        active_directory_server="127.0.0.1",
+        ssl_enabled=True,
+        ssl_ca="this is not a certificate",
+    ) as source:
+        with pytest.raises(ConfigurableFieldValueError) as exc_info:
+            await source.validate_config()
+
+        assert "SSL certificate is not valid" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_validate_config_passes_when_ssl_enabled_with_valid_certificate():
+    async with create_outlook_source(
+        data_source=OUTLOOK_SERVER,
+        username="foo.bar@gmail.com",
+        password="abc@123",
+        exchange_server="127.0.0.1",
+        domain="gmail.com",
+        active_directory_server="127.0.0.1",
+        ssl_enabled=True,
+        ssl_ca=VALID_SSL_CERTIFICATE,
+    ) as source:
+        await source.validate_config()
+
+
 @pytest.mark.asyncio
 async def test_get_docs():
     async with create_outlook_source() as source:
@@ -780,6 +848,61 @@ async def test_get_docs():
             assert document in EXPECTED_RESPONSE
 
 
+def _account_raising_on_inbox(exception, smtp):
+    """Build a mock account whose first folder access (inbox) raises."""
+    account = MagicMock()
+    account.default_timezone = "UTC"
+    account.primary_smtp_address = smtp
+    type(account).inbox = mock.PropertyMock(side_effect=exception)
+    return account
+
+
+@pytest.mark.asyncio
+async def test_get_docs_skips_account_without_mailbox_and_continues():
+    async with create_outlook_source() as source:
+        bad_account = _account_raising_on_inbox(
+            ErrorNonExistentMailbox("no mailbox"), smtp="no.mailbox@example.com"
+        )
+        source.client._get_user_instance.get_user_accounts = AsyncIterator(
+            [bad_account, MockAccount()]
+        )
+        source._logger = MagicMock()
+
+        documents = [document async for document, _ in source.get_docs()]
+
+        # The healthy account is still fully synced past the mail stage.
+        assert all(document in EXPECTED_RESPONSE for document in documents)
+        assert any(document["_id"] == "contact_1" for document in documents)
+
+        source._logger.warning.assert_called_once()
+        warning_message = source._logger.warning.call_args.args[0]
+        assert "no.mailbox@example.com" in warning_message
+        assert "no associated mailbox" in warning_message
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "exception",
+    [
+        # Connection-wide failures (e.g. exchangelib wraps TLS errors as
+        # TransportError) must abort the sync rather than be silently skipped,
+        # which would empty the index.
+        TransportError("TLS verification failed"),
+        RuntimeError("boom"),
+    ],
+)
+async def test_get_docs_reraises_connection_wide_error(exception):
+    async with create_outlook_source() as source:
+        bad_account = _account_raising_on_inbox(exception, smtp="broken@example.com")
+        source.client._get_user_instance.get_user_accounts = AsyncIterator(
+            [bad_account, MockAccount()]
+        )
+
+        with pytest.raises(type(exception)):
+            async for _document, _ in source.get_docs():
+                pass
+
+
 @pytest.mark.asyncio
 async def test_get_contacts_resolves_via_distinguished_folder_id():
     async with create_outlook_source() as source:

app/connectors_service/tests/test_utils.py

@@ -847,6 +847,12 @@ def test_evaluate_timedelta():
     assert expected_response == "2023-02-19T14:25:05.158843"
 
 
+MULTILINE_PEM_CERTIFICATE = """-----BEGIN CERTIFICATE-----
+Certificate1
+Certificate2
+-----END CERTIFICATE-----"""
+
+
 def test_get_pem_format_with_postfix():
     expected_formatted_pem_key = """-----BEGIN PRIVATE KEY-----
 PrivateKey
@@ -860,14 +866,9 @@ def test_get_pem_format_with_postfix():
 
 
 def test_get_pem_format_multiline():
-    expected_formatted_certificate = """-----BEGIN CERTIFICATE-----
-Certificate1
-Certificate2
------END CERTIFICATE-----"""
     certificate = "-----BEGIN CERTIFICATE----- Certificate1 Certificate2 -----END CERTIFICATE-----"
 
-    formatted_certificate = get_pem_format(key=certificate)
-    assert formatted_certificate == expected_formatted_certificate
+    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE
 
 
 def test_get_pem_format_multiple_certificates():
@@ -884,6 +885,44 @@ def test_get_pem_format_multiple_certificates():
     assert formatted_multi_certificate == expected_formatted_multiple_certificates
 
 
+def test_get_pem_format_already_multiline_certificate_is_preserved():
+    # Copied verbatim from a .pem file: the intact END marker must not be broken.
+    assert get_pem_format(key=MULTILINE_PEM_CERTIFICATE) == MULTILINE_PEM_CERTIFICATE
+
+
+def test_get_pem_format_already_multiline_private_key_is_preserved():
+    private_key = """-----BEGIN PRIVATE KEY-----
+PrivateKey
+-----END PRIVATE KEY-----"""
+
+    assert (
+        get_pem_format(key=private_key, postfix="-----END PRIVATE KEY-----")
+        == private_key
+    )
+
+
+def test_get_pem_format_multiline_with_surrounding_whitespace_is_normalized():
+    certificate = (
+        "\n  -----BEGIN CERTIFICATE-----  \n"
+        "Certificate1\n"
+        "\n"
+        "Certificate2\n"
+        "  -----END CERTIFICATE-----  \n"
+    )
+
+    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE
+
+
+def test_get_pem_format_single_line_with_trailing_newline_is_reflowed():
+    # A trailing newline alone must not be mistaken for multi-line content.
+    certificate = (
+        "-----BEGIN CERTIFICATE----- Certificate1 Certificate2 "
+        "-----END CERTIFICATE-----\n"
+    )
+
+    assert get_pem_format(key=certificate) == MULTILINE_PEM_CERTIFICATE
+
+
 def test_truncate_id():
     long_id = "something-12341361361-21905128510263"
     truncated_id = truncate_id(long_id)