diff --git a/sssd_test_framework/hosts/ad.py b/sssd_test_framework/hosts/ad.py index ade95bd5..56c66cc7 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,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" diff --git a/sssd_test_framework/roles/ad.py b/sssd_test_framework/roles/ad.py index fb5cf22e..9762f0d2 100644 --- a/sssd_test_framework/roles/ad.py +++ b/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) @@ -2931,7 +2943,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) @@ -2995,7 +3007,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: """ @@ -3019,7 +3031,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]]: """ @@ -3086,21 +3098,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. @@ -3180,36 +3177,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() diff --git a/sssd_test_framework/roles/client.py b/sssd_test_framework/roles/client.py index 5955e7f2..420c60d2 100644 --- a/sssd_test_framework/roles/client.py +++ b/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. diff --git a/sssd_test_framework/roles/generic.py b/sssd_test_framework/roles/generic.py index 58c887c5..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) @@ -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 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: """ diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py index 93342b5b..86366970 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,13 @@ 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. + # 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") return @@ -285,6 +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 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): + 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 diff --git a/sssd_test_framework/utils/sssd.py b/sssd_test_framework/utils/sssd.py index ef5ac17f..bce5e4db 100644 --- a/sssd_test_framework/utils/sssd.py +++ b/sssd_test_framework/utils/sssd.py @@ -1231,3 +1231,52 @@ 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 | None = None, + ) -> 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_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. 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" + 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" diff --git a/sssd_test_framework/utils/tools.py b/sssd_test_framework/utils/tools.py index 22160e24..0c3797cb 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 @@ -13,6 +14,7 @@ "GetentUtils", "GroupEntry", "LinuxToolsUtils", + "OpenSSLUtils", "PasswdEntry", "UnixGroup", "UnixObject", @@ -1158,4 +1160,148 @@ 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 + """ + 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_openldap() + + 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_openldap(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" + + self.fs.backup(ldap_conf) + current = self.fs.read(ldap_conf) if self.fs.exists(ldap_conf) 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")