ldaps: add LDAPS support

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,44 @@ 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 using certutil.
+
+        :return: CA certificate in PEM format.
+        :rtype: str
+        :raises RuntimeError: If CA certificate cannot be retrieved.
+        """
+        result = self.conn.run(
+            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,
+        )
+
+        if result.rc != 0:
+            raise RuntimeError(f"Failed to get CA certificate: {result.stderr}!")
+
+        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!")
+
+        return cert_pem + "\n"

sssd_test_framework/roles/ad.py

@@ -252,6 +252,18 @@ 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.
+
+        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.
+        """
+        return self.host.get_ca_cert()
+
     @property
     def naming_context(self) -> str:
         """
@@ -2781,7 +2793,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,
             )
@@ -2861,7 +2873,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)
 
@@ -2944,7 +2956,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)
 
@@ -3008,7 +3020,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:
         """
@@ -3032,7 +3044,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]]:
         """
@@ -3099,21 +3111,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.
@@ -3193,36 +3190,16 @@ 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,
-        )
+        return self.host.get_ca_cert()
 
-        if result.rc != 0:
-            raise RuntimeError(f"Failed to get CA certificate: {result.stderr}!")
-
-        ca_cert_b64 = result.stdout.strip()
+    def export_root_ca_certificate(self) -> str:
+        """
+        Export the AD root CA certificate in PEM format.
 
-        if not ca_cert_b64:
-            raise RuntimeError("CA certificate not found in certificate stores!")
+        Implements :meth:`GenericCertificateAuthority.export_root_ca_certificate`.
 
-        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: PEM-formatted root CA certificate.
+        :rtype: str
+        :raises RuntimeError: If CA certificate cannot be retrieved.
+        """
+        return self.get_ca_cert()

sssd_test_framework/roles/client.py

@@ -25,6 +25,7 @@
 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
 
@@ -118,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.

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)
 
@@ -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.roles.client.Client.install_ca_cert` or
+        :meth:`~sssd_test_framework.utils.sssd.SSSDCommonConfiguration.use_ldaps`.
+
+        .. code-block:: python
+            :caption: Example usage
+
+            @pytest.mark.topology(KnownTopologyGroup.AnyDC)
+            def test_ldaps__certificate_install(client: Client, provider: GenericProvider):
+                client.install_ca_cert(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

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.

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:
         """

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 the provider CA certificate on the client.
+        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")
             return
@@ -285,6 +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 the provider CA certificate on the client.
+        openssl = OpenSSLUtils(client, client.fs)
+        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")
+                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:
+                    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")
             return