Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 66 additions & 71 deletions sssd_test_framework/roles/ad.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,21 +183,18 @@ def test_example_autofs(client: Client, ad: AD, nfs: NFS):
}
"""

#
# Disabling CA functionality until sssd-ci-container updated
#
# self._ca = ADCertificateAuthority(self.host)
# """
# AD Certificate Authority server management.
#
# Provides certificate operations:
# - Request certificates using templates
# - Request smartcard certificates with Enrollment Agent
# - Revoke certificates with configurable reasons
# - Manage certificate holds
# - Export certificates as PFX
# - Retrieve certificate and template details
# """
self._ca = ADCertificateAuthority(self.host)
"""
AD Certificate Authority server management.

Provides certificate operations:
- Request certificates using templates
- Request smartcard certificates with Enrollment Agent
- Revoke certificates with configurable reasons
- Manage certificate holds
- Export certificates as PFX
- Retrieve certificate and template details
"""

@property
def password_policy(self) -> ADPasswordPolicy:
Expand All @@ -217,47 +214,44 @@ def test_example(client: Client, ad: AD):
"""
return self._password_policy

#
# Disabling CA functionality until sssd-ci-container updated
#
# @property
# def ca(self) -> ADCertificateAuthority:
# """
# AD Certificate Authority management.
#
# Provides certificate operations:
#
# - Request certificates using templates
# - Request smartcard certificates with Enrollment Agent
# - Revoke certificates with configurable reasons
# - Manage certificate holds
# - Export certificates as PFX
# - Retrieve certificate and template details
#
# .. code-block:: python
# :caption: Example usage
#
# @pytest.mark.topology(KnownTopology.AD)
# def test_example(client: Client, ad: AD):
# # Request smartcard certificate
# cert, key, csr = ad.ca.request(
# template="SmartcardLogon",
# subject="CN=testuser"
# )
#
# # Get certificate details
# cert_details = ad.ca.get(cert)
#
# # Place certificate on hold (temporary revocation)
# ad.ca.revoke_hold(cert)
#
# # Remove hold (restore certificate)
# ad.ca.revoke_hold_remove(cert)
#
# # Permanently revoke certificate
# ad.ca.revoke(cert, reason="key_compromise")
# """
# return self._ca
@property
def ca(self) -> ADCertificateAuthority:
"""
AD Certificate Authority management.

Provides certificate operations:

- Request certificates using templates
- Request smartcard certificates with Enrollment Agent
- Revoke certificates with configurable reasons
- Manage certificate holds
- Export certificates as PFX
- Retrieve certificate and template details

.. code-block:: python
:caption: Example usage

@pytest.mark.topology(KnownTopology.AD)
def test_example(client: Client, ad: AD):
# Request smartcard certificate
cert, key, csr = ad.ca.request(
template="SmartcardLogon",
subject="CN=testuser"
)

# Get certificate details
cert_details = ad.ca.get(cert)

# Place certificate on hold (temporary revocation)
ad.ca.revoke_hold(cert)

# Remove hold (restore certificate)
ad.ca.revoke_hold_remove(cert)

# Permanently revoke certificate
ad.ca.revoke(cert, reason="key_compromise")
"""
return self._ca

@property
def naming_context(self) -> str:
Expand Down Expand Up @@ -2659,7 +2653,7 @@ def _setup_enrollment_agent(self) -> None:
FriendlyName = "Enrollment Agent"

[RequestAttributes]
CertificateTemplate = "EnrollmentAgent2"
CertificateTemplate = "EnrollmentAgent_IDMTEST"
"@
Set-Content -Path "{inf_path}" -Value $infContent
""")
Expand Down Expand Up @@ -2830,7 +2824,7 @@ def request(
self.export_pfx(cert_path, pfx_path)

self.host.conn.run(
f'"Secret123" | & openssl pkcs12 -in {pfx_path} -nocerts -nodes -out {extracted_key_path} -passin stdin'
f"openssl pkcs12 -in {pfx_path} -nocerts -nodes -out {extracted_key_path} -passin pass:Secret123"
)

self.host.conn.run(f"openssl rsa -in {extracted_key_path} -out {key_path}")
Expand Down Expand Up @@ -3074,24 +3068,25 @@ 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("""\
$ca = Get-ChildItem -Path Cert:\\LocalMachine\\Root | Where-Object {
$_.Subject -like '*CN=ad-RootCA*' -and $_.Issuer -eq $_.Subject
} | Select-Object -First 1
if ($ca) {
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=ad-RootCA*'
} | Select-Object -First 1
if ($ca) {
}} 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 {
}} else {{
Write-Error "CA certificate not found"
exit 1
}
}
}}
}}
"""),
raise_on_error=False,
)
Expand Down
19 changes: 18 additions & 1 deletion sssd_test_framework/topology_controllers.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import re
import tempfile
import textwrap

from pytest_mh import BackupTopologyController
from pytest_mh.conn import ProcessResult
Expand Down Expand Up @@ -66,10 +67,14 @@ def join_domain(self, client: ClientHost, provider: IPAHost | ADHost | SambaHost
"""
Helper method for joining domains
"""
client.fs.backup("/etc/krb5.conf")
client.fs.backup("/etc/krb5.keytab")
self.logger.info(f"Enrolling {client.hostname} into {provider.domain}")

# Remove any existing Kerberos configuration and keytab
client.fs.rm("/etc/krb5.conf")
# Skip removing krb5.conf when provider is AD
if not isinstance(provider, (ADHost)):
client.fs.rm("/etc/krb5.conf")
client.fs.rm("/etc/krb5.keytab")

# First check if joined. If so, leave.
Expand Down Expand Up @@ -100,6 +105,18 @@ def join_domain(self, client: ClientHost, provider: IPAHost | ADHost | SambaHost
client.conn.exec(["realm", "leave", "--unattended", provider.domain], input=provider.adminpw)
client.conn.exec(["realm", "join", provider.domain], input=provider.adminpw)

if isinstance(provider, (ADHost)):
ad_krb5_content = textwrap.dedent(f"""\
[realms]
{provider.realm} = {{
default_domain = {provider.domain}
pkinit_anchors = FILE:/etc/sssd/pki/sssd_auth_ca_db.pem
pkinit_eku_checking = kpServerAuth
pkinit_kdc_hostname = {provider.hostname}
}}
""")
client.fs.write("/etc/krb5.conf.d/ad_pkinit", ad_krb5_content)


class ClientTopologyController(ProvisionedBackupTopologyController):
"""
Expand Down