From a4b1f65deab68982d189e71de9ba2be23952a38b Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Mon, 24 Aug 2026 18:48:42 +0530 Subject: [PATCH 1/8] roles: add export_root_ca_certificate() to all providers Add export_root_ca_certificate() abstract method to GenericProvider and implement it in the AD, IPA, and Samba roles. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/roles/ad.py | 33 ++++++++++++++++++++++++++ sssd_test_framework/roles/generic.py | 35 ++++++++++++++++++++++++++++ sssd_test_framework/roles/ipa.py | 27 +++++++++++++++++++++ sssd_test_framework/roles/samba.py | 17 ++++++++++++++ 4 files changed, 112 insertions(+) diff --git a/sssd_test_framework/roles/ad.py b/sssd_test_framework/roles/ad.py index fb5cf22e..03fd3b22 100644 --- a/sssd_test_framework/roles/ad.py +++ b/sssd_test_framework/roles/ad.py @@ -252,6 +252,27 @@ def test_example(client: Client, ad: AD): """ return self._ca + def export_root_ca_certificate(self) -> str: + """ + Export the AD root CA certificate in PEM format. + + Uses ``certutil.exe`` to export the root CA certificate from the AD + Certificate Services store (requires AD CS to be installed on the DC). + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the root CA certificate cannot be exported. + """ + tmp = "C:\\Windows\\Temp" + self.host.conn.run(f'certutil.exe -f -"ca.cert" {tmp}\\ca.crt', raise_on_error=False) + self.host.conn.run(f"certutil.exe -f -encode {tmp}\\ca.crt {tmp}\\ca.pem", raise_on_error=False) + result = self.host.conn.run(f"Get-Content {tmp}\\ca.pem -Raw", raise_on_error=False) + + if result.rc != 0 or not result.stdout.strip(): + raise RuntimeError(f"Failed to export root CA certificate: {result.stderr}") + + return result.stdout.strip() + "\n" + @property def naming_context(self) -> str: """ @@ -3213,3 +3234,15 @@ def get_ca_cert(self) -> str: ca_cert_lines = [ca_cert_b64[i : i + 64] for i in range(0, len(ca_cert_b64), 64)] return "-----BEGIN CERTIFICATE-----\n" + "\n".join(ca_cert_lines) + "\n-----END CERTIFICATE-----\n" + + def export_root_ca_certificate(self) -> str: + """ + Export the AD root CA certificate in PEM format. + + Implements :meth:`GenericCertificateAuthority.export_root_ca_certificate`. + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If CA certificate cannot be retrieved. + """ + return self.get_ca_cert() diff --git a/sssd_test_framework/roles/generic.py b/sssd_test_framework/roles/generic.py index 58c887c5..67acb1bc 100644 --- a/sssd_test_framework/roles/generic.py +++ b/sssd_test_framework/roles/generic.py @@ -382,6 +382,30 @@ def test_certificate_operations(client: Client, provider: GenericProvider): """ pass + @abstractmethod + def export_root_ca_certificate(self) -> str: + """ + Export the root CA certificate in PEM format. + + Used to install the CA certificate on the client for LDAPS connections via + :meth:`~sssd_test_framework.utils.cert.CertUtils.install_ca` or + :meth:`~sssd_test_framework.utils.sssd.SSSDCommonConfiguration.use_ldaps`. + + .. code-block:: python + :caption: Example usage + + @pytest.mark.topology(KnownTopologyGroup.AnyDC) + def test_example(client: Client, provider: GenericProvider): + client.cert.install_ca(provider) + client.firewall.outbound.drop_port(389) + # ... join or LDAPS operation + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the certificate cannot be exported. + """ + pass + class GenericADProvider(GenericProvider): """ @@ -1694,3 +1718,14 @@ def get(self, cert_path: str) -> dict[str, list[str]]: :rtype: dict[str, list[str]] """ pass + + @abstractmethod + def export_root_ca_certificate(self) -> str: + """ + Export the root CA certificate in PEM format. + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the certificate cannot be exported. + """ + pass diff --git a/sssd_test_framework/roles/ipa.py b/sssd_test_framework/roles/ipa.py index 2d75e136..0c365c50 100644 --- a/sssd_test_framework/roles/ipa.py +++ b/sssd_test_framework/roles/ipa.py @@ -255,6 +255,18 @@ def test_example(client: Client, ipa: IPA): """ return self._ca + def export_root_ca_certificate(self) -> str: + """ + Export the IPA root CA certificate in PEM format. + + Delegates to :meth:`~IPACertificateAuthority.export_root_ca_certificate`. + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the certificate cannot be read. + """ + return self.ca.export_root_ca_certificate() + @property def naming_context(self) -> str: """ @@ -3360,6 +3372,21 @@ def get(self, cert_path: str) -> dict[str, list[str]]: raise ValueError(f"Certificate with serial '{serial}' not found in IPA: {result.stderr}!") return self._parse_cert_info(result.stdout) + def export_root_ca_certificate(self) -> str: + """ + Export the IPA root CA certificate in PEM format. + + Implements :meth:`GenericCertificateAuthority.export_root_ca_certificate`. + + Reads the CA certificate from ``/etc/ipa/ca.crt``, which is written to + the IPA server during ``ipa-server-install`` and is always present. + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the certificate cannot be read. + """ + return self.fs.read("/etc/ipa/ca.crt") + def _generate_csr(self, key_path: str, csr_path: str, subject: str, key_size: int = 2048) -> None: """ Generate a CSR and key using OpenSSL. diff --git a/sssd_test_framework/roles/samba.py b/sssd_test_framework/roles/samba.py index 135d63fd..36523007 100644 --- a/sssd_test_framework/roles/samba.py +++ b/sssd_test_framework/roles/samba.py @@ -186,6 +186,23 @@ def test_example(client: Client, samba: Samba): """ return self._password_policy + def export_root_ca_certificate(self) -> str: + """ + Export the Samba root CA certificate in PEM format. + + Reads the CA certificate from the path configured by ``ca_cert_path`` in + the host's ``mhc.yaml`` config section, defaulting to ``/var/data/certs/ca.crt``. + + :return: PEM-formatted root CA certificate. + :rtype: str + :raises RuntimeError: If the certificate cannot be read. + """ + ca_cert_path = self.host.config.get("ca_cert_path", "/var/data/certs/ca.crt") + result = self.host.conn.run(f"cat {ca_cert_path}", raise_on_error=False) + if result.rc != 0: + raise RuntimeError(f"Failed to export root CA certificate: {result.stderr}") + return result.stdout.strip() + @property def naming_context(self) -> str: """ From 4dbadd20321a5b4e141aa1dfd9b895c7ecda1cfe Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Tue, 25 Aug 2026 00:22:50 +0530 Subject: [PATCH 2/8] utils: add OpenSSLUtils and CA certificate install helpers Add OpenSSLUtils to utils/tools.py with install_ca_cert(), install_ca_cert_from_server(), and _configure_openldap() helpers. Add install_ca_cert() and ssl attribute to the Client role. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/roles/client.py | 172 +++++++++++++++++++++++++++ sssd_test_framework/roles/generic.py | 4 +- sssd_test_framework/utils/tools.py | 144 +++++++++++++++++++++- 3 files changed, 317 insertions(+), 3 deletions(-) diff --git a/sssd_test_framework/roles/client.py b/sssd_test_framework/roles/client.py index 5955e7f2..946f65cc 100644 --- a/sssd_test_framework/roles/client.py +++ b/sssd_test_framework/roles/client.py @@ -2,6 +2,8 @@ from __future__ import annotations +import os + from pytest_mh.conn import ProcessResult from ..hosts.client import ClientHost @@ -27,6 +29,7 @@ from ..utils.sssd import SSSDUtils from ..utils.vfido import Vfido from .base import BaseLinuxRole +from .generic import GenericProvider __all__ = [ "Client", @@ -222,3 +225,172 @@ def sudorule(self, name: str) -> LocalSudoRule: """ return self.local.sudorule(name) + + def install_ca_cert( + self, + provider: GenericProvider, + name: str = "test-ca.crt", + cert_path: str | None = None, + ) -> str: + """ + Install a CA certificate into the system trust store. + + Fetches the root CA certificate from ``provider``, writes it to the system + trust anchor directory (``/etc/pki/ca-trust/source/anchors/``), + runs ``update-ca-trust``, and configures ``/etc/openldap/ldap.conf`` + so that OpenLDAP clients (adcli, realmd, ldapsearch, etc.) use the + system trust store. + + All changes are tracked by the test framework and restored automatically + after the test. + + :param provider: Provider role to fetch the root CA certificate from. + :type provider: ~sssd_test_framework.roles.generic.GenericProvider + :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. + Ignored when ``cert_path`` is set. + :type name: str + :param cert_path: Full destination path on the client, overrides the default. + :type cert_path: str | None + :return: Path where the certificate was written on the client. + :rtype: str + """ + ca_cert = provider.export_root_ca_certificate() + + if cert_path is None: + cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" + + parent = os.path.dirname(cert_path) + if parent and len(parent.split("/")) > 2: + self.fs.mkdir_p(parent) + + self.fs.write(cert_path, ca_cert) + self.host.conn.run("update-ca-trust") + self._configure_tls_cacert() + + return cert_path + + def install_ca_cert_from_server( + self, + hostname: str, + port: int = 636, + name: str = "test-ca.crt", + cert_path: str | None = None, + ) -> str: + """ + Install a CA certificate by fetching it directly from a TLS server. + + Connects to ``hostname:port`` with ``openssl s_client``, captures the + certificate the server presents, writes it to the system trust anchor + directory, runs ``update-ca-trust``, and points ``TLS_CACERT`` in + ``/etc/openldap/ldap.conf`` at the updated system bundle. Useful when + the server uses a self-signed certificate that is difficult to export + from the host (e.g. AD DC with no AD CS). + + All changes are tracked by the test framework and restored automatically + after the test. + + :param hostname: Hostname or IP of the TLS server. + :type hostname: str + :param port: TLS port, defaults to 636. + :type port: int + :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. + Ignored when ``cert_path`` is set. + :type name: str + :param cert_path: Full destination path on the client, overrides the default. + :type cert_path: str | None + :return: Path where the certificate was written on the client. + :rtype: str + :raises RuntimeError: If the certificate cannot be fetched from the server. + """ + result = self.host.conn.run( + f"openssl s_client -connect {hostname}:{port} -showcerts /dev/null", + raise_on_error=False, + ) + + certs = [] + current: list[str] = [] + for line in result.stdout.splitlines(): + if "-----BEGIN CERTIFICATE-----" in line: + current = [line] + elif "-----END CERTIFICATE-----" in line: + current.append(line) + certs.append("\n".join(current)) + current = [] + elif current: + current.append(line) + + if not certs: + raise RuntimeError(f"Failed to fetch certificate from {hostname}:{port}: {result.stderr}") + + pem = certs[-1] + + if cert_path is None: + cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" + + parent = os.path.dirname(cert_path) + if parent and len(parent.split("/")) > 2: + self.fs.mkdir_p(parent) + + self.fs.write(cert_path, pem) + self.host.conn.run("update-ca-trust") + self._configure_tls_cacert() + + return cert_path + + def use_ldaps(self, provider: GenericProvider) -> None: + """ + Configure SSSD to connect to the provider over an encrypted LDAP channel. + + Installs the provider's root CA certificate into the system trust store via + :meth:`install_ca_cert`, then sets the appropriate SSSD domain option: + + - **AD / Samba**: sets ``ad_use_ldaps = True`` (port 636). + - **IPA**: sets ``ldap_id_use_start_tls = True`` (STARTTLS on port 389). + IPA's ``ipa_server`` option does not accept ``ldaps://`` URIs. + + Works with :data:`~sssd_test_framework.topology.KnownTopologyGroup.AnyDC` + so a single test covers AD, Samba, and IPA topologies. + + .. code-block:: python + :caption: Example usage + + @pytest.mark.topology(KnownTopologyGroup.AnyDC) + def test_example(client: Client, provider: GenericProvider): + client.use_ldaps(provider) + client.sssd.start() + + :param provider: Provider role to fetch the root CA certificate from. + :type provider: ~sssd_test_framework.roles.generic.GenericProvider + """ + cacert = self.install_ca_cert(provider) + self.sssd.domain["ldap_tls_cacert"] = cacert + if provider.name == "ad": + self.sssd.domain["ad_use_ldaps"] = "True" + elif provider.name == "ipa": + # IPA provider does not support ldaps:// URIs in ipa_server; use STARTTLS on + # port 389 instead, which IPA supports and provides equivalent transport security. + self.sssd.domain["ldap_id_use_start_tls"] = "True" + + def _configure_tls_cacert(self) -> None: + """ + Configure ``/etc/openldap/ldap.conf`` for system CA trust and channel binding. + + Removes any explicit ``TLS_CACERT`` and ``TLS_CACERTDIR`` directives so + libldap falls back to its compiled-in defaults (the system trust store). + Sets ``SASL_CBINDING tls-endpoint`` for channel binding support. + """ + ldap_conf = "/etc/openldap/ldap.conf" + + result = self.host.conn.run(f"cat {ldap_conf}", raise_on_error=False) + current = result.stdout if result.rc == 0 else "" + + lines = [ + line + for line in current.splitlines() + if not line.startswith("TLS_CACERT") + and not line.startswith("TLS_CACERTDIR") + and not line.startswith("SASL_CBINDING") + ] + lines.append("SASL_CBINDING tls-endpoint") + + self.fs.write(ldap_conf, "\n".join(lines) + "\n") diff --git a/sssd_test_framework/roles/generic.py b/sssd_test_framework/roles/generic.py index 67acb1bc..832bf1e1 100644 --- a/sssd_test_framework/roles/generic.py +++ b/sssd_test_framework/roles/generic.py @@ -388,7 +388,7 @@ def export_root_ca_certificate(self) -> str: Export the root CA certificate in PEM format. Used to install the CA certificate on the client for LDAPS connections via - :meth:`~sssd_test_framework.utils.cert.CertUtils.install_ca` or + :meth:`~sssd_test_framework.roles.client.Client.install_ca_cert` or :meth:`~sssd_test_framework.utils.sssd.SSSDCommonConfiguration.use_ldaps`. .. code-block:: python @@ -396,7 +396,7 @@ def export_root_ca_certificate(self) -> str: @pytest.mark.topology(KnownTopologyGroup.AnyDC) def test_example(client: Client, provider: GenericProvider): - client.cert.install_ca(provider) + client.install_ca_cert(provider) client.firewall.outbound.drop_port(389) # ... join or LDAPS operation diff --git a/sssd_test_framework/utils/tools.py b/sssd_test_framework/utils/tools.py index 22160e24..bb3be075 100644 --- a/sssd_test_framework/utils/tools.py +++ b/sssd_test_framework/utils/tools.py @@ -13,6 +13,7 @@ "GetentUtils", "GroupEntry", "LinuxToolsUtils", + "OpenSSLUtils", "PasswdEntry", "UnixGroup", "UnixObject", @@ -1158,4 +1159,145 @@ def generate( ) self.fs.chown(homedir, user=user, group=group, args=["-R"]) - return self.fs.read(f"{homedir}/.ssh/{file}.pub"), self.fs.read(f"{homedir}/.ssh/{file}") + return self.fs.read(f"{homedir}/.ssh/{file}.pub"), self.fs.read(f"{homedir}/.ssh/{file}") + + +class OpenSSLUtils: + """ + Manage CA certificates and TLS configuration on a remote Linux host. + + Provides helpers to install CA certificates into the system trust store + and to configure OpenLDAP client settings for TLS. Suitable for use + both from topology controllers (host-level setup) and from role + methods (per-test operations). + """ + + DEFAULT_CACERT_PATH = "/etc/pki/ca-trust/source/anchors/test-ca.crt" + + def __init__(self, host: MultihostHost, fs: LinuxFileSystem) -> None: + """ + :param host: Remote host. + :type host: MultihostHost + :param fs: Filesystem utility for the host. + :type fs: LinuxFileSystem + """ + self.host: MultihostHost = host + self.fs: LinuxFileSystem = fs + + def install_ca_cert( + self, + cert_pem: str, + name: str = "test-ca.crt", + cert_path: str | None = None, + ) -> str: + """ + Install a PEM-encoded CA certificate into the system trust store. + + Writes *cert_pem* to the system trust anchor directory + (``/etc/pki/ca-trust/source/anchors/``), runs + ``update-ca-trust``, and configures ``/etc/openldap/ldap.conf`` so + that OpenLDAP clients use the system trust store. + + All changes made via *fs* are tracked by the test framework and + restored automatically when the test or topology is torn down. + + :param cert_pem: PEM-encoded CA certificate content. + :type cert_pem: str + :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. + Ignored when *cert_path* is set. + :type name: str + :param cert_path: Full destination path on the client, overrides the default. + :type cert_path: str | None + :return: Path where the certificate was written on the client. + :rtype: str + """ + import os + + if cert_path is None: + cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" + + parent = os.path.dirname(cert_path) + if parent and len(parent.split("/")) > 2: + self.fs.mkdir_p(parent) + + self.fs.write(cert_path, cert_pem) + self.host.conn.run("update-ca-trust") + self._configure_tls_cacert() + + return cert_path + + def install_ca_cert_from_server( + self, + hostname: str, + port: int = 636, + name: str = "test-ca.crt", + cert_path: str | None = None, + ) -> str: + """ + Install a CA certificate by fetching it directly from a TLS server. + + Connects to ``hostname:port`` with ``openssl s_client``, captures the + last certificate in the chain (the root CA), writes it to the system + trust anchor directory, runs ``update-ca-trust``, and points + ``TLS_CACERT`` in ``/etc/openldap/ldap.conf`` at the system bundle. + Useful when the server uses a self-signed certificate that is + difficult to export from the host (e.g. AD DC with no AD CS). + + :param hostname: Hostname or IP of the TLS server. + :type hostname: str + :param port: TLS port, defaults to 636. + :type port: int + :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. + Ignored when *cert_path* is set. + :type name: str + :param cert_path: Full destination path on the client, overrides the default. + :type cert_path: str | None + :return: Path where the certificate was written on the client. + :rtype: str + :raises RuntimeError: If the certificate cannot be fetched from the server. + """ + result = self.host.conn.run( + f"openssl s_client -connect {hostname}:{port} -showcerts /dev/null", + raise_on_error=False, + ) + + certs: list[str] = [] + current: list[str] = [] + for line in result.stdout.splitlines(): + if "-----BEGIN CERTIFICATE-----" in line: + current = [line] + elif "-----END CERTIFICATE-----" in line: + current.append(line) + certs.append("\n".join(current)) + current = [] + elif current: + current.append(line) + + if not certs: + raise RuntimeError(f"Failed to fetch certificate from {hostname}:{port}: {result.stderr}") + + return self.install_ca_cert(certs[-1], name=name, cert_path=cert_path) + + def _configure_tls_cacert(self) -> None: + """ + Configure ``/etc/openldap/ldap.conf`` for system CA trust and channel binding. + + Removes any explicit ``TLS_CACERT`` and ``TLS_CACERTDIR`` directives so + libldap falls back to its compiled-in defaults (the system trust store). + Sets ``SASL_CBINDING tls-endpoint`` for channel binding support. + """ + ldap_conf = "/etc/openldap/ldap.conf" + + result = self.host.conn.run(f"cat {ldap_conf}", raise_on_error=False) + current = result.stdout if result.rc == 0 else "" + + lines = [ + line + for line in current.splitlines() + if not line.startswith("TLS_CACERT") + and not line.startswith("TLS_CACERTDIR") + and not line.startswith("SASL_CBINDING") + ] + lines.append("SASL_CBINDING tls-endpoint") + + self.fs.write(ldap_conf, "\n".join(lines) + "\n") From 4956b85b6e2fa1943801ec7d03d1a8e6f0360517 Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Wed, 30 Sep 2026 00:41:02 +0530 Subject: [PATCH 3/8] topology_controllers: install CA cert during topology setup ADTopologyController (Samba) and IPATopologyController now install the provider's CA certificate on the client via OpenSSLUtils before the provisioned check, so LDAPS/STARTTLS tests have the cert available without any per-test setup. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/topology_controllers.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py index 93342b5b..cf9d3c71 100644 --- a/sssd_test_framework/topology_controllers.py +++ b/sssd_test_framework/topology_controllers.py @@ -16,6 +16,7 @@ from .hosts.ldap import LDAPHost from .hosts.samba import SambaHost from .misc.ssh import retry_command +from .utils.tools import OpenSSLUtils __all__ = [ "LDAPTopologyController", @@ -252,6 +253,11 @@ def topology_setup(self, client: ClientHost, ipa: IPAHost) -> None: client.fs.backup("/etc/resolv.conf") + # Install IPA CA certificate so LDAPS/STARTTLS tests can use it without per-test setup. + # Done before the provisioned check so it runs even on already-provisioned containers. + cert_pem = ipa.fs.read("/etc/ipa/ca.crt") + OpenSSLUtils(client, client.fs).install_ca_cert(cert_pem) + if self.provisioned: self.logger.info(f"Topology '{self.name}' is already provisioned") return @@ -285,6 +291,14 @@ def topology_setup(self, client: ClientHost, provider: ADHost | SambaHost) -> No provider.fs.backup("/etc/resolv.conf") provider.fs.write("/etc/resolv.conf", f"search {provider.domain}\nnameserver 127.0.0.1\n\n") + # Install Samba CA certificate so LDAPS tests can use it without per-test setup. + # Done before the provisioned check so it runs even on already-provisioned containers. + if isinstance(provider, SambaHost): + ca_cert_path = provider.config.get("ca_cert_path", "/var/data/certs/ca.crt") + result = provider.conn.run(f"cat {ca_cert_path}", raise_on_error=False) + if result.rc == 0 and result.stdout.strip(): + OpenSSLUtils(client, client.fs).install_ca_cert(result.stdout) + if self.provisioned: self.logger.info(f"Topology '{self.name}' is already provisioned") return From 9701e20a32226ff60d74434f436a4987164164b1 Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Wed, 30 Sep 2026 00:41:09 +0530 Subject: [PATCH 4/8] utils/sssd: add ssl_tls() to SSSDCommonConfiguration Replace use_ldaps() with ssl_tls() in SSSDCommonConfiguration. The CA certificate is now pre-installed by the topology controller, so ssl_tls() only sets the SSSD domain options (ad_use_ldaps or ldap_id_use_start_tls) and ldap_tls_cacert path. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/utils/sssd.py | 41 +++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/sssd_test_framework/utils/sssd.py b/sssd_test_framework/utils/sssd.py index ef5ac17f..ca6b22c2 100644 --- a/sssd_test_framework/utils/sssd.py +++ b/sssd_test_framework/utils/sssd.py @@ -1231,3 +1231,44 @@ def smartcard_with_softhsm(self, smartcard: SmartCardUtils) -> None: self.sssd.pam["pam_cert_auth"] = "True" self.sssd.domain["local_auth_policy"] = "enable:smartcard" self.sssd.start() + + def ssl_tls( + self, + provider: GenericProvider, + cacert: str = "/etc/pki/ca-trust/source/anchors/test-ca.crt", + ) -> None: + """ + Configure SSSD to connect to the provider over an encrypted LDAP channel. + + The CA certificate must already be installed on the client (e.g. by + :meth:`~sssd_test_framework.utils.tools.OpenSSLUtils.install_ca_cert` or + automatically by the topology controller during setup). + + Sets the appropriate SSSD domain option based on the provider type: + + - **AD / Samba**: sets ``ad_use_ldaps = True`` (port 636). + - **IPA**: sets ``ldap_id_use_start_tls = True`` (STARTTLS on port 389). + + Works with :data:`~sssd_test_framework.topology.KnownTopologyGroup.AnyDC` + so a single test covers AD, Samba, and IPA topologies. + + .. code-block:: python + :caption: Example usage + + @pytest.mark.topology(KnownTopologyGroup.AnyDC) + def test_example(client: Client, provider: GenericProvider): + client.sssd.common.ssl_tls(provider) + client.sssd.start() + + :param provider: Provider role to determine the SSSD option to set. + :type provider: ~sssd_test_framework.roles.generic.GenericProvider + :param cacert: Path to the CA certificate on the client. + :type cacert: str + """ + self.sssd.domain["ldap_tls_cacert"] = cacert + if provider.name == "ad": + self.sssd.domain["ad_use_ldaps"] = "True" + elif provider.name == "ipa": + # IPA provider does not support ldaps:// URIs in ipa_server; use STARTTLS on + # port 389 instead, which IPA supports and provides equivalent transport security. + self.sssd.domain["ldap_id_use_start_tls"] = "True" From a527b02f62105d327eca895b672c2f1f5c2441da Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Wed, 30 Sep 2026 00:41:16 +0530 Subject: [PATCH 5/8] roles/client: replace cert helpers with OpenSSLUtils, remove use_ldaps() Remove install_ca_cert(), install_ca_cert_from_server(), use_ldaps(), and _configure_tls_cacert() from the Client role. Add self.ssl (OpenSSLUtils) for tests that still need to install a cert manually (e.g. AD with no AD CS). Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/roles/client.py | 178 +--------------------------- 1 file changed, 6 insertions(+), 172 deletions(-) diff --git a/sssd_test_framework/roles/client.py b/sssd_test_framework/roles/client.py index 946f65cc..420c60d2 100644 --- a/sssd_test_framework/roles/client.py +++ b/sssd_test_framework/roles/client.py @@ -2,8 +2,6 @@ from __future__ import annotations -import os - from pytest_mh.conn import ProcessResult from ..hosts.client import ClientHost @@ -27,9 +25,9 @@ from ..utils.sss_override import SSSOverrideUtils from ..utils.sssctl import SSSCTLUtils from ..utils.sssd import SSSDUtils +from ..utils.tools import OpenSSLUtils from ..utils.vfido import Vfido from .base import BaseLinuxRole -from .generic import GenericProvider __all__ = [ "Client", @@ -121,6 +119,11 @@ def __init__(self, *args, **kwargs) -> None: Managing virtual passkey device and service """ + self.ssl: OpenSSLUtils = OpenSSLUtils(self.host, self.fs) + """ + Managing CA certificates and TLS configuration on the client. + """ + def setup(self) -> None: """ Called before execution of each test. @@ -225,172 +228,3 @@ def sudorule(self, name: str) -> LocalSudoRule: """ return self.local.sudorule(name) - - def install_ca_cert( - self, - provider: GenericProvider, - name: str = "test-ca.crt", - cert_path: str | None = None, - ) -> str: - """ - Install a CA certificate into the system trust store. - - Fetches the root CA certificate from ``provider``, writes it to the system - trust anchor directory (``/etc/pki/ca-trust/source/anchors/``), - runs ``update-ca-trust``, and configures ``/etc/openldap/ldap.conf`` - so that OpenLDAP clients (adcli, realmd, ldapsearch, etc.) use the - system trust store. - - All changes are tracked by the test framework and restored automatically - after the test. - - :param provider: Provider role to fetch the root CA certificate from. - :type provider: ~sssd_test_framework.roles.generic.GenericProvider - :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. - Ignored when ``cert_path`` is set. - :type name: str - :param cert_path: Full destination path on the client, overrides the default. - :type cert_path: str | None - :return: Path where the certificate was written on the client. - :rtype: str - """ - ca_cert = provider.export_root_ca_certificate() - - if cert_path is None: - cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" - - parent = os.path.dirname(cert_path) - if parent and len(parent.split("/")) > 2: - self.fs.mkdir_p(parent) - - self.fs.write(cert_path, ca_cert) - self.host.conn.run("update-ca-trust") - self._configure_tls_cacert() - - return cert_path - - def install_ca_cert_from_server( - self, - hostname: str, - port: int = 636, - name: str = "test-ca.crt", - cert_path: str | None = None, - ) -> str: - """ - Install a CA certificate by fetching it directly from a TLS server. - - Connects to ``hostname:port`` with ``openssl s_client``, captures the - certificate the server presents, writes it to the system trust anchor - directory, runs ``update-ca-trust``, and points ``TLS_CACERT`` in - ``/etc/openldap/ldap.conf`` at the updated system bundle. Useful when - the server uses a self-signed certificate that is difficult to export - from the host (e.g. AD DC with no AD CS). - - All changes are tracked by the test framework and restored automatically - after the test. - - :param hostname: Hostname or IP of the TLS server. - :type hostname: str - :param port: TLS port, defaults to 636. - :type port: int - :param name: Certificate filename under ``/etc/pki/ca-trust/source/anchors/``. - Ignored when ``cert_path`` is set. - :type name: str - :param cert_path: Full destination path on the client, overrides the default. - :type cert_path: str | None - :return: Path where the certificate was written on the client. - :rtype: str - :raises RuntimeError: If the certificate cannot be fetched from the server. - """ - result = self.host.conn.run( - f"openssl s_client -connect {hostname}:{port} -showcerts /dev/null", - raise_on_error=False, - ) - - certs = [] - current: list[str] = [] - for line in result.stdout.splitlines(): - if "-----BEGIN CERTIFICATE-----" in line: - current = [line] - elif "-----END CERTIFICATE-----" in line: - current.append(line) - certs.append("\n".join(current)) - current = [] - elif current: - current.append(line) - - if not certs: - raise RuntimeError(f"Failed to fetch certificate from {hostname}:{port}: {result.stderr}") - - pem = certs[-1] - - if cert_path is None: - cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" - - parent = os.path.dirname(cert_path) - if parent and len(parent.split("/")) > 2: - self.fs.mkdir_p(parent) - - self.fs.write(cert_path, pem) - self.host.conn.run("update-ca-trust") - self._configure_tls_cacert() - - return cert_path - - def use_ldaps(self, provider: GenericProvider) -> None: - """ - Configure SSSD to connect to the provider over an encrypted LDAP channel. - - Installs the provider's root CA certificate into the system trust store via - :meth:`install_ca_cert`, then sets the appropriate SSSD domain option: - - - **AD / Samba**: sets ``ad_use_ldaps = True`` (port 636). - - **IPA**: sets ``ldap_id_use_start_tls = True`` (STARTTLS on port 389). - IPA's ``ipa_server`` option does not accept ``ldaps://`` URIs. - - Works with :data:`~sssd_test_framework.topology.KnownTopologyGroup.AnyDC` - so a single test covers AD, Samba, and IPA topologies. - - .. code-block:: python - :caption: Example usage - - @pytest.mark.topology(KnownTopologyGroup.AnyDC) - def test_example(client: Client, provider: GenericProvider): - client.use_ldaps(provider) - client.sssd.start() - - :param provider: Provider role to fetch the root CA certificate from. - :type provider: ~sssd_test_framework.roles.generic.GenericProvider - """ - cacert = self.install_ca_cert(provider) - self.sssd.domain["ldap_tls_cacert"] = cacert - if provider.name == "ad": - self.sssd.domain["ad_use_ldaps"] = "True" - elif provider.name == "ipa": - # IPA provider does not support ldaps:// URIs in ipa_server; use STARTTLS on - # port 389 instead, which IPA supports and provides equivalent transport security. - self.sssd.domain["ldap_id_use_start_tls"] = "True" - - def _configure_tls_cacert(self) -> None: - """ - Configure ``/etc/openldap/ldap.conf`` for system CA trust and channel binding. - - Removes any explicit ``TLS_CACERT`` and ``TLS_CACERTDIR`` directives so - libldap falls back to its compiled-in defaults (the system trust store). - Sets ``SASL_CBINDING tls-endpoint`` for channel binding support. - """ - ldap_conf = "/etc/openldap/ldap.conf" - - result = self.host.conn.run(f"cat {ldap_conf}", raise_on_error=False) - current = result.stdout if result.rc == 0 else "" - - lines = [ - line - for line in current.splitlines() - if not line.startswith("TLS_CACERT") - and not line.startswith("TLS_CACERTDIR") - and not line.startswith("SASL_CBINDING") - ] - lines.append("SASL_CBINDING tls-endpoint") - - self.fs.write(ldap_conf, "\n".join(lines) + "\n") From 4727a653d9c6e96fb8e2aa1002d442304cd99602 Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Fri, 2 Oct 2026 19:34:24 +0530 Subject: [PATCH 6/8] hosts/ad: move get_ca_cert() and _get_ca_config() to ADHost Move cert export logic from ADCertificateAuthority to ADHost so topology controllers can access it directly without going through role fixtures. ADCertificateAuthority._get_ca_config() and get_ca_cert() now delegate to self.host, keeping the role-level API unchanged. ADTopologyController gains an ADHost cert install branch with two-level fallback: ADCS/PowerShell first, then openssl s_client, so cert install works regardless of whether Windows Certificate Services is present. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/hosts/ad.py | 58 ++++++++++++++++++++ sssd_test_framework/roles/ad.py | 59 +++------------------ sssd_test_framework/topology_controllers.py | 12 ++++- 3 files changed, 75 insertions(+), 54 deletions(-) diff --git a/sssd_test_framework/hosts/ad.py b/sssd_test_framework/hosts/ad.py index ade95bd5..a4ba3479 100644 --- a/sssd_test_framework/hosts/ad.py +++ b/sssd_test_framework/hosts/ad.py @@ -2,6 +2,7 @@ from __future__ import annotations +import textwrap from pathlib import PureWindowsPath from typing import Any @@ -318,3 +319,60 @@ def restore(self, backup_data: Any | None) -> None: """, log_level=ProcessLogLevel.Error, ) + + def get_ca_config(self) -> str: + """ + Get CA configuration string. + + :return: CA configuration string. + :rtype: str + """ + result = self.conn.run("certutil -dump", raise_on_error=False) + if result.rc == 0: + for line in result.stdout_lines: + if "Config:" in line: + return line.split(":", 1)[1].strip() + + return f"{self.hostname}\\{self.domain}-CA" + + def get_ca_cert(self) -> str: + """ + Get the CA certificate in PEM format by querying the Windows certificate store. + + :return: CA certificate in PEM format. + :rtype: str + :raises RuntimeError: If CA certificate cannot be retrieved. + """ + ca_name = self.get_ca_config().split("\\", 1)[1].strip('"') + result = self.conn.run( + textwrap.dedent(f"""\ + $ca = Get-ChildItem -Path Cert:\\LocalMachine\\Root | Where-Object {{ + $_.Subject -like '*CN={ca_name}*' -and $_.Issuer -eq $_.Subject + }} | Select-Object -First 1 + if ($ca) {{ + [System.Convert]::ToBase64String($ca.Export('Cert')) + }} else {{ + $ca = Get-ChildItem -Path Cert:\\LocalMachine\\My | Where-Object {{ + $_.Subject -like '*CN={ca_name}*' + }} | Select-Object -First 1 + if ($ca) {{ + [System.Convert]::ToBase64String($ca.Export('Cert')) + }} else {{ + Write-Error "CA certificate not found" + exit 1 + }} + }} + """), + raise_on_error=False, + ) + + if result.rc != 0: + raise RuntimeError(f"Failed to get CA certificate: {result.stderr}!") + + ca_cert_b64 = result.stdout.strip() + + if not ca_cert_b64: + raise RuntimeError("CA certificate not found in certificate stores!") + + ca_cert_lines = [ca_cert_b64[i : i + 64] for i in range(0, len(ca_cert_b64), 64)] + return "-----BEGIN CERTIFICATE-----\n" + "\n".join(ca_cert_lines) + "\n-----END CERTIFICATE-----\n" diff --git a/sssd_test_framework/roles/ad.py b/sssd_test_framework/roles/ad.py index 03fd3b22..feeff2e9 100644 --- a/sssd_test_framework/roles/ad.py +++ b/sssd_test_framework/roles/ad.py @@ -2802,7 +2802,7 @@ def _setup_enrollment_agent(self) -> None: try: result = self.host.conn.run( - f'certreq -submit -config "{self._get_ca_config()}" "{req_path}" "{cert_path}"', + f'certreq -submit -config "{self.host.get_ca_config()}" "{req_path}" "{cert_path}"', raise_on_error=False, timeout=30, ) @@ -2882,7 +2882,7 @@ def request_basic( self.host.conn.run(f'certreq -q -new "{inf_path}" "{req_path}"') - self.host.conn.run(f'certreq -submit -config "{self._get_ca_config()}" "{req_path}" "{cert_path}"') + self.host.conn.run(f'certreq -submit -config "{self.host.get_ca_config()}" "{req_path}" "{cert_path}"') self.export_pfx(cert_path, pfx_path, password=password) @@ -2952,7 +2952,7 @@ def request( self.host.conn.run(f'certreq -q -sign -cert "{enrollment_agent_hash}" "{req_path}" "{signed_req_path}"') - self.host.conn.run(f'certreq -submit -config "{self._get_ca_config()}" "{signed_req_path}" "{cert_path}"') + self.host.conn.run(f'certreq -submit -config "{self.host.get_ca_config()}" "{signed_req_path}" "{cert_path}"') self.export_pfx(cert_path, pfx_path) @@ -3016,7 +3016,7 @@ def revoke(self, cert_path: str, reason: str = "unspecified") -> None: serial = self._get_cert_serial(cert_path) reason_code = self._revocation_reason_to_code(reason) - self.host.conn.run(f'certutil -config "{self._get_ca_config()}" -revoke {serial} {reason_code}') + self.host.conn.run(f'certutil -config "{self.host.get_ca_config()}" -revoke {serial} {reason_code}') def revoke_hold(self, cert_path: str) -> None: """ @@ -3040,7 +3040,7 @@ def revoke_hold_remove(self, cert_path: str) -> None: """ serial = self._get_cert_serial(cert_path) - self.host.conn.run(f'certutil -config "{self._get_ca_config()}" -revoke {serial} 8') # 8 = removeFromCRL + self.host.conn.run(f'certutil -config "{self.host.get_ca_config()}" -revoke {serial} 8') # 8 = removeFromCRL def get(self, cert_path: str) -> dict[str, list[str]]: """ @@ -3107,21 +3107,6 @@ def get_certificate_template(self, template_name: str) -> dict[str, list[str]]: return attrs_ad_parse(result.stdout) - def _get_ca_config(self) -> str: - """ - Get CA configuration string. - - :return: CA configuration string. - :rtype: str - """ - result = self.host.conn.run("certutil -dump", raise_on_error=False) - if result.rc == 0: - for line in result.stdout_lines: - if "Config:" in line: - return line.split(":", 1)[1].strip() - - return f"{self.host.hostname}\\{self.host.domain}-CA" - def _get_cert_serial(self, cert_path: str) -> str: """ Extract certificate serial number. @@ -3201,39 +3186,7 @@ def get_ca_cert(self) -> str: :rtype: str :raises RuntimeError: If CA certificate cannot be retrieved. """ - ca_name = self._get_ca_config().split("\\", 1)[1].strip('"') - result = self.host.conn.run( - textwrap.dedent(f"""\ - $ca = Get-ChildItem -Path Cert:\\LocalMachine\\Root | Where-Object {{ - $_.Subject -like '*CN={ca_name}*' -and $_.Issuer -eq $_.Subject - }} | Select-Object -First 1 - if ($ca) {{ - [System.Convert]::ToBase64String($ca.Export('Cert')) - }} else {{ - $ca = Get-ChildItem -Path Cert:\\LocalMachine\\My | Where-Object {{ - $_.Subject -like '*CN={ca_name}*' - }} | Select-Object -First 1 - if ($ca) {{ - [System.Convert]::ToBase64String($ca.Export('Cert')) - }} else {{ - Write-Error "CA certificate not found" - exit 1 - }} - }} - """), - raise_on_error=False, - ) - - if result.rc != 0: - raise RuntimeError(f"Failed to get CA certificate: {result.stderr}!") - - ca_cert_b64 = result.stdout.strip() - - if not ca_cert_b64: - raise RuntimeError("CA certificate not found in certificate stores!") - - ca_cert_lines = [ca_cert_b64[i : i + 64] for i in range(0, len(ca_cert_b64), 64)] - return "-----BEGIN CERTIFICATE-----\n" + "\n".join(ca_cert_lines) + "\n-----END CERTIFICATE-----\n" + return self.host.get_ca_cert() def export_root_ca_certificate(self) -> str: """ diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py index cf9d3c71..594117bc 100644 --- a/sssd_test_framework/topology_controllers.py +++ b/sssd_test_framework/topology_controllers.py @@ -291,13 +291,23 @@ def topology_setup(self, client: ClientHost, provider: ADHost | SambaHost) -> No provider.fs.backup("/etc/resolv.conf") provider.fs.write("/etc/resolv.conf", f"search {provider.domain}\nnameserver 127.0.0.1\n\n") - # Install Samba CA certificate so LDAPS tests can use it without per-test setup. + # Install CA certificate so LDAPS tests can use it without per-test setup. # Done before the provisioned check so it runs even on already-provisioned containers. if isinstance(provider, SambaHost): ca_cert_path = provider.config.get("ca_cert_path", "/var/data/certs/ca.crt") result = provider.conn.run(f"cat {ca_cert_path}", raise_on_error=False) if result.rc == 0 and result.stdout.strip(): OpenSSLUtils(client, client.fs).install_ca_cert(result.stdout) + elif isinstance(provider, ADHost): + try: + cert_pem = provider.get_ca_cert() + OpenSSLUtils(client, client.fs).install_ca_cert(cert_pem) + except RuntimeError: + self.logger.warning("AD CA certificate not found via cert store, falling back to openssl s_client") + try: + OpenSSLUtils(client, client.fs).install_ca_cert_from_server(provider.hostname) + except Exception: + self.logger.warning("Could not install AD CA cert; LDAPS tests will need per-test cert install") if self.provisioned: self.logger.info(f"Topology '{self.name}' is already provisioned") From d61eee723c75cdf247bce7f691b26665bf59c50b Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Sat, 3 Oct 2026 00:30:05 +0530 Subject: [PATCH 7/8] utils/tools: make install_ca_cert idempotent Skip the write and update-ca-trust call if the certificate file already exists with identical content. This avoids redundant re-installation on every topology setup cycle when using provisioned containers. Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/hosts/ad.py | 34 ++++++------------- sssd_test_framework/roles/ad.py | 13 ++------ sssd_test_framework/roles/generic.py | 4 +-- sssd_test_framework/topology_controllers.py | 36 ++++++++++++--------- sssd_test_framework/utils/sssd.py | 18 ++++++++--- sssd_test_framework/utils/tools.py | 13 +++++--- 6 files changed, 55 insertions(+), 63 deletions(-) diff --git a/sssd_test_framework/hosts/ad.py b/sssd_test_framework/hosts/ad.py index a4ba3479..56c66cc7 100644 --- a/sssd_test_framework/hosts/ad.py +++ b/sssd_test_framework/hosts/ad.py @@ -337,31 +337,17 @@ def get_ca_config(self) -> str: def get_ca_cert(self) -> str: """ - Get the CA certificate in PEM format by querying the Windows certificate store. + Get the CA certificate in PEM format using certutil. :return: CA certificate in PEM format. :rtype: str :raises RuntimeError: If CA certificate cannot be retrieved. """ - ca_name = self.get_ca_config().split("\\", 1)[1].strip('"') result = self.conn.run( - textwrap.dedent(f"""\ - $ca = Get-ChildItem -Path Cert:\\LocalMachine\\Root | Where-Object {{ - $_.Subject -like '*CN={ca_name}*' -and $_.Issuer -eq $_.Subject - }} | Select-Object -First 1 - if ($ca) {{ - [System.Convert]::ToBase64String($ca.Export('Cert')) - }} else {{ - $ca = Get-ChildItem -Path Cert:\\LocalMachine\\My | Where-Object {{ - $_.Subject -like '*CN={ca_name}*' - }} | Select-Object -First 1 - if ($ca) {{ - [System.Convert]::ToBase64String($ca.Export('Cert')) - }} else {{ - Write-Error "CA certificate not found" - exit 1 - }} - }} + textwrap.dedent("""\ + certutil.exe -f -"ca.cert" C:\\Windows\\Temp\\ca.crt + certutil.exe -f -encode C:\\Windows\\Temp\\ca.crt C:\\Windows\\Temp\\ca.pem + Get-Content C:\\Windows\\Temp\\ca.pem -Raw """), raise_on_error=False, ) @@ -369,10 +355,8 @@ def get_ca_cert(self) -> str: if result.rc != 0: raise RuntimeError(f"Failed to get CA certificate: {result.stderr}!") - ca_cert_b64 = result.stdout.strip() - - if not ca_cert_b64: - raise RuntimeError("CA certificate not found in certificate stores!") + cert_pem = result.stdout.strip() + if not cert_pem or "-----BEGIN CERTIFICATE-----" not in cert_pem: + raise RuntimeError("CA certificate not found in certutil output!") - ca_cert_lines = [ca_cert_b64[i : i + 64] for i in range(0, len(ca_cert_b64), 64)] - return "-----BEGIN CERTIFICATE-----\n" + "\n".join(ca_cert_lines) + "\n-----END CERTIFICATE-----\n" + return cert_pem + "\n" diff --git a/sssd_test_framework/roles/ad.py b/sssd_test_framework/roles/ad.py index feeff2e9..9762f0d2 100644 --- a/sssd_test_framework/roles/ad.py +++ b/sssd_test_framework/roles/ad.py @@ -256,22 +256,13 @@ def export_root_ca_certificate(self) -> str: """ Export the AD root CA certificate in PEM format. - Uses ``certutil.exe`` to export the root CA certificate from the AD - Certificate Services store (requires AD CS to be installed on the DC). + Delegates to :meth:`~sssd_test_framework.hosts.ad.ADHost.get_ca_cert`. :return: PEM-formatted root CA certificate. :rtype: str :raises RuntimeError: If the root CA certificate cannot be exported. """ - tmp = "C:\\Windows\\Temp" - self.host.conn.run(f'certutil.exe -f -"ca.cert" {tmp}\\ca.crt', raise_on_error=False) - self.host.conn.run(f"certutil.exe -f -encode {tmp}\\ca.crt {tmp}\\ca.pem", raise_on_error=False) - result = self.host.conn.run(f"Get-Content {tmp}\\ca.pem -Raw", raise_on_error=False) - - if result.rc != 0 or not result.stdout.strip(): - raise RuntimeError(f"Failed to export root CA certificate: {result.stderr}") - - return result.stdout.strip() + "\n" + return self.host.get_ca_cert() @property def naming_context(self) -> str: diff --git a/sssd_test_framework/roles/generic.py b/sssd_test_framework/roles/generic.py index 832bf1e1..0a6ef8f4 100644 --- a/sssd_test_framework/roles/generic.py +++ b/sssd_test_framework/roles/generic.py @@ -120,7 +120,7 @@ def password_policy(self) -> GenericPasswordPolicy: :caption: Example usage @pytest.mark.topology(KnownTopologyGroup.Any) - def test_example(client: Client, provider: GenericProvider): + def test_password_policy__lockout(client: Client, provider: GenericProvider): # Enable password complexity provider.password_policy.complexity(enable=True) @@ -395,7 +395,7 @@ def export_root_ca_certificate(self) -> str: :caption: Example usage @pytest.mark.topology(KnownTopologyGroup.AnyDC) - def test_example(client: Client, provider: GenericProvider): + def test_ldaps__certificate_install(client: Client, provider: GenericProvider): client.install_ca_cert(provider) client.firewall.outbound.drop_port(389) # ... join or LDAPS operation diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py index 594117bc..f4003333 100644 --- a/sssd_test_framework/topology_controllers.py +++ b/sssd_test_framework/topology_controllers.py @@ -255,8 +255,10 @@ def topology_setup(self, client: ClientHost, ipa: IPAHost) -> None: # Install IPA CA certificate so LDAPS/STARTTLS tests can use it without per-test setup. # Done before the provisioned check so it runs even on already-provisioned containers. - cert_pem = ipa.fs.read("/etc/ipa/ca.crt") - OpenSSLUtils(client, client.fs).install_ca_cert(cert_pem) + # Check the client first to avoid fetching the cert from the IPA server unnecessarily. + openssl = OpenSSLUtils(client, client.fs) + if not client.fs.exists("/etc/pki/ca-trust/source/anchors/ipa-ca.crt"): + openssl.install_ca_cert(ipa.fs.read("/etc/ipa/ca.crt"), name="ipa-ca.crt") if self.provisioned: self.logger.info(f"Topology '{self.name}' is already provisioned") @@ -291,23 +293,25 @@ def topology_setup(self, client: ClientHost, provider: ADHost | SambaHost) -> No provider.fs.backup("/etc/resolv.conf") provider.fs.write("/etc/resolv.conf", f"search {provider.domain}\nnameserver 127.0.0.1\n\n") - # Install CA certificate so LDAPS tests can use it without per-test setup. - # Done before the provisioned check so it runs even on already-provisioned containers. + # Install the provider's CA certificate on the client so LDAPS tests can use it + # without per-test setup. Done before the provisioned check so it runs even on + # already-provisioned containers. + openssl = OpenSSLUtils(client, client.fs) if isinstance(provider, SambaHost): - ca_cert_path = provider.config.get("ca_cert_path", "/var/data/certs/ca.crt") - result = provider.conn.run(f"cat {ca_cert_path}", raise_on_error=False) - if result.rc == 0 and result.stdout.strip(): - OpenSSLUtils(client, client.fs).install_ca_cert(result.stdout) + if not client.fs.exists("/etc/pki/ca-trust/source/anchors/samba-ca.crt"): + ca_cert_path = provider.config.get("ca_cert_path", "/var/data/certs/ca.crt") + result = provider.conn.run(f"cat {ca_cert_path}", raise_on_error=False) + if result.rc == 0 and result.stdout.strip(): + openssl.install_ca_cert(result.stdout, name="samba-ca.crt") elif isinstance(provider, ADHost): - try: - cert_pem = provider.get_ca_cert() - OpenSSLUtils(client, client.fs).install_ca_cert(cert_pem) - except RuntimeError: - self.logger.warning("AD CA certificate not found via cert store, falling back to openssl s_client") + if not client.fs.exists("/etc/pki/ca-trust/source/anchors/ad-ca.crt"): try: - OpenSSLUtils(client, client.fs).install_ca_cert_from_server(provider.hostname) - except Exception: - self.logger.warning("Could not install AD CA cert; LDAPS tests will need per-test cert install") + openssl.install_ca_cert(provider.get_ca_cert(), name="ad-ca.crt") + except RuntimeError: + try: + openssl.install_ca_cert_from_server(provider.hostname, name="ad-ca.crt") + except Exception as e: + self.logger.warning(f"Unable to install AD CA certificate on {client.hostname}: {e}") if self.provisioned: self.logger.info(f"Topology '{self.name}' is already provisioned") diff --git a/sssd_test_framework/utils/sssd.py b/sssd_test_framework/utils/sssd.py index ca6b22c2..bce5e4db 100644 --- a/sssd_test_framework/utils/sssd.py +++ b/sssd_test_framework/utils/sssd.py @@ -1235,7 +1235,7 @@ def smartcard_with_softhsm(self, smartcard: SmartCardUtils) -> None: def ssl_tls( self, provider: GenericProvider, - cacert: str = "/etc/pki/ca-trust/source/anchors/test-ca.crt", + cacert: str | None = None, ) -> None: """ Configure SSSD to connect to the provider over an encrypted LDAP channel. @@ -1256,15 +1256,23 @@ def ssl_tls( :caption: Example usage @pytest.mark.topology(KnownTopologyGroup.AnyDC) - def test_example(client: Client, provider: GenericProvider): + def test_ldaps__user_lookup_over_encrypted_channel(client: Client, provider: GenericProvider): + u = provider.user("tuser").add() client.sssd.common.ssl_tls(provider) client.sssd.start() + result = client.tools.id("tuser") + assert result is not None :param provider: Provider role to determine the SSSD option to set. :type provider: ~sssd_test_framework.roles.generic.GenericProvider - :param cacert: Path to the CA certificate on the client. - :type cacert: str - """ + :param cacert: Path to the CA certificate on the client. Defaults to the + topology-specific cert installed by the topology controller. + :type cacert: str | None + """ + if cacert is None: + _cert_names = {"ipa": "ipa-ca.crt", "samba": "samba-ca.crt", "ad": "ad-ca.crt"} + _provider_type = type(provider).__name__.lower() + cacert = f"/etc/pki/ca-trust/source/anchors/{_cert_names.get(_provider_type, 'test-ca.crt')}" self.sssd.domain["ldap_tls_cacert"] = cacert if provider.name == "ad": self.sssd.domain["ad_use_ldaps"] = "True" diff --git a/sssd_test_framework/utils/tools.py b/sssd_test_framework/utils/tools.py index bb3be075..831d5843 100644 --- a/sssd_test_framework/utils/tools.py +++ b/sssd_test_framework/utils/tools.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from typing import Any, Sequence import jc @@ -1211,18 +1212,21 @@ def install_ca_cert( :return: Path where the certificate was written on the client. :rtype: str """ - import os - if cert_path is None: cert_path = f"/etc/pki/ca-trust/source/anchors/{name}" + if self.fs.exists(cert_path) and self.fs.read(cert_path).strip() == cert_pem.strip(): + return cert_path + + self.fs.backup(cert_path) + parent = os.path.dirname(cert_path) if parent and len(parent.split("/")) > 2: self.fs.mkdir_p(parent) self.fs.write(cert_path, cert_pem) self.host.conn.run("update-ca-trust") - self._configure_tls_cacert() + self._configure_openldap() return cert_path @@ -1278,7 +1282,7 @@ def install_ca_cert_from_server( return self.install_ca_cert(certs[-1], name=name, cert_path=cert_path) - def _configure_tls_cacert(self) -> None: + def _configure_openldap(self) -> None: """ Configure ``/etc/openldap/ldap.conf`` for system CA trust and channel binding. @@ -1288,6 +1292,7 @@ def _configure_tls_cacert(self) -> None: """ ldap_conf = "/etc/openldap/ldap.conf" + self.fs.backup(ldap_conf) result = self.host.conn.run(f"cat {ldap_conf}", raise_on_error=False) current = result.stdout if result.rc == 0 else "" From beefd157fa7ace66f64a9425316fe37a796ecacf Mon Sep 17 00:00:00 2001 From: shridhargadekar Date: Fri, 9 Oct 2026 15:42:26 +0530 Subject: [PATCH 8/8] topology_controllers,utils/tools: address review feedback - topology_controllers: check client before fetching cert from provider to skip redundant SSH round-trips on re-provisioned containers; name certs by topology (ipa-ca.crt, samba-ca.crt, ad-ca.crt); replace conn.run cat with fs.exists/fs.read - utils/tools: add fs.backup() for /etc/openldap/ldap.conf so it is restored on teardown; replace conn.run cat with fs.exists/fs.read - utils/sssd: ssl_tls() derives default cacert path from provider type Co-Authored-By: Claude Sonnet 4.6 --- sssd_test_framework/topology_controllers.py | 5 ++--- sssd_test_framework/utils/tools.py | 3 +-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py index f4003333..86366970 100644 --- a/sssd_test_framework/topology_controllers.py +++ b/sssd_test_framework/topology_controllers.py @@ -300,9 +300,8 @@ def topology_setup(self, client: ClientHost, provider: ADHost | SambaHost) -> No if isinstance(provider, SambaHost): if not client.fs.exists("/etc/pki/ca-trust/source/anchors/samba-ca.crt"): ca_cert_path = provider.config.get("ca_cert_path", "/var/data/certs/ca.crt") - result = provider.conn.run(f"cat {ca_cert_path}", raise_on_error=False) - if result.rc == 0 and result.stdout.strip(): - openssl.install_ca_cert(result.stdout, name="samba-ca.crt") + if provider.fs.exists(ca_cert_path): + openssl.install_ca_cert(provider.fs.read(ca_cert_path), name="samba-ca.crt") elif isinstance(provider, ADHost): if not client.fs.exists("/etc/pki/ca-trust/source/anchors/ad-ca.crt"): try: diff --git a/sssd_test_framework/utils/tools.py b/sssd_test_framework/utils/tools.py index 831d5843..0c3797cb 100644 --- a/sssd_test_framework/utils/tools.py +++ b/sssd_test_framework/utils/tools.py @@ -1293,8 +1293,7 @@ def _configure_openldap(self) -> None: ldap_conf = "/etc/openldap/ldap.conf" self.fs.backup(ldap_conf) - result = self.host.conn.run(f"cat {ldap_conf}", raise_on_error=False) - current = result.stdout if result.rc == 0 else "" + current = self.fs.read(ldap_conf) if self.fs.exists(ldap_conf) else "" lines = [ line