Add IPA IPA Trust Topology Controller

sssd_test_framework/roles/ad.py

@@ -138,6 +138,13 @@ def fqn(self, name: str) -> str:
         """
         return f"{name}@{self.domain}"
 
+    @property
+    def admin_fqn(self) -> str:
+        """
+        Return fully qualified administrator name in form name@domain.
+        """
+        return f"administrator@{self.domain}"
+
     def user(self, name: str, basedn: ADObject | str | None = "cn=users") -> ADUser:
         """
         Get user object.

sssd_test_framework/roles/ipa.py

@@ -137,6 +137,19 @@ def setup(self) -> None:
         super().setup()
         self.host.kinit()
 
+    def fqn(self, name: str) -> str:
+        """
+        Return fully qualified name in form name@domain.
+        """
+        return f"{name}@{self.domain}"
+
+    @property
+    def admin_fqn(self) -> str:
+        """
+        Return fully qualified admin name in form name@domain.
+        """
+        return f"admin@{self.domain}"
+
     def user(self, name: str) -> IPAUser:
         """
         Get user object.

sssd_test_framework/roles/samba.py

@@ -134,6 +134,13 @@ def fqn(self, name: str) -> str:
         """
         return f"{name}@{self.domain}"
 
+    @property
+    def admin_fqn(self) -> str:
+        """
+        Return fully qualified administrator name in form name@domain.
+        """
+        return f"administrator@{self.domain}"
+
     def user(self, name: str) -> SambaUser:
         """
         Get user object.

sssd_test_framework/topology.py

@@ -13,6 +13,7 @@
     ClientTopologyController,
     IPATopologyController,
     IPATrustADTopologyController,
+    IPATrustIPATopologyController,
     IPATrustSambaTopologyController,
     LDAPTopologyController,
     SambaTopologyController,
@@ -118,6 +119,17 @@ def test_ldap(client: Client, ldap: LDAP):
     .. topology-mark:: KnownTopology.IPATrustSamba
     """
 
+    IPATrustIPA = SSSDTopologyMark(
+        name="ipa-trust-ipa",
+        topology=Topology(TopologyDomain("sssd", client=1, ipa=1), TopologyDomain("ipa2", ipa=1)),
+        controller=IPATrustIPATopologyController(),
+        domains=dict(test="sssd.ipa[0]"),
+        fixtures=dict(client="sssd.client[0]", ipa="sssd.ipa[0]", trusted="ipa2.ipa[0]"),
+    )
+    """
+    .. topology-mark:: KnownTopology.IPATrustIPA
+    """
+
 
 class KnownTopologyGroup(KnownTopologyGroupBase):
     """
@@ -145,7 +157,12 @@ def test_ldap(client: Client, provider: GenericProvider):
     .. topology-mark:: KnownTopologyGroup.AnyAD
     """
 
-    IPATrust = [KnownTopology.IPATrustAD, KnownTopology.IPATrustSamba]
+    IPATrustAD = [KnownTopology.IPATrustAD, KnownTopology.IPATrustSamba]
+    """
+    .. topology-mark:: KnownTopologyGroup.IPATrustAD
+    """
+
+    AnyIPATrust = [KnownTopology.IPATrustAD, KnownTopology.IPATrustSamba, KnownTopology.IPATrustIPA]
     """
-    .. topology-mark:: KnownTopologyGroup.IPATrust
+    .. topology-mark:: KnownTopologyGroup.AnyIPATrust
     """

sssd_test_framework/topology_controllers.py

@@ -18,6 +18,7 @@
     "SambaTopologyController",
     "IPATrustADTopologyController",
     "IPATrustSambaTopologyController",
+    "IPATrustIPATopologyController",
 ]
 
 
@@ -174,3 +175,68 @@ class IPATrustSambaTopologyController(IPATrustADTopologyController):
     """
 
     pass
+
+
+class IPATrustIPATopologyController(ProvisionedBackupTopologyController):
+    """
+    IPA trust IPA Topology Controller.
+    """
+
+    @BackupTopologyController.restore_vanilla_on_error
+    def topology_setup(self, client: ClientHost, ipa: IPAHost, trusted: IPAHost) -> None:
+        if self.provisioned:
+            self.logger.info(f"Topology '{self.name}' is already provisioned")
+            return
+
+        # Add ipa-ipa trust COPR and update packages
+        self.logger.info("Adding COPR and updating packages")
+        ipa.conn.exec(["dnf", "copr", "enable", "abbra/wip-ipa-trust", "-y"])
+        client.conn.exec(["dnf", "copr", "enable", "abbra/wip-ipa-trust", "-y"])
+        trusted.conn.exec(["dnf", "copr", "enable", "abbra/wip-ipa-trust", "-y"])
+
+        ipa.conn.exec(["dnf", "update", "freeipa-server", "sssd-client", "-y"])
+        trusted.conn.exec(["dnf", "update", "freeipa-server", "sssd-client", "-y"])
+        client.conn.exec(["dnf", "update", "sssd-client", "-y"])
+
+        # F40 sssd-kcm fails to start with 'Invalid option --genconf-section=kcm:'
+        ipa.conn.exec(["systemctl", "restart", "sssd-kcm"])
+        trusted.conn.exec(["systemctl", "restart", "sssd-kcm"])
+
+        # Create trust
+        self.logger.info(f"Establishing trust between {ipa.domain} and {trusted.domain}")
+
+        ipa.kinit()
+        ipa.conn.exec(
+            [
+                "ipa",
+                "trust-add",
+                trusted.domain,
+                "--admin",
+                "admin",
+                "--password",
+                "--range-type=ipa-ad-trust-posix",
+                "--type=ipa",
+                "--two-way=true",
+            ],
+            input=trusted.adminpw,
+        )
+
+        # Do not enroll client into IPA domain if it is already joined
+        if "ipa" not in self.multihost.provisioned_topologies:
+            self.logger.info(f"Enrolling {client.hostname} into {ipa.domain}")
+
+            # Remove any existing Kerberos configuration and keytab
+            client.fs.rm("/etc/krb5.conf")
+            client.fs.rm("/etc/krb5.keytab")
+
+            # Backup ipa-client-install files
+            client.fs.backup("/etc/ipa")
+            client.fs.backup("/var/lib/ipa-client")
+
+            # Join IPA domain)
+            client.conn.exec(["realm", "join", ipa.domain], input=ipa.adminpw)
+
+        # Backup so we can restore to this state after each test
+        self.backup_data[ipa] = ipa.backup()
+        self.backup_data[trusted] = trusted.backup()
+        self.backup_data[client] = client.backup()