feature: Implement provider interface in client role

jakub-vavra-cz · jakub-vavra-cz · commit 71b334d6394c · 2025-12-01T15:30:25.000+01:00
Allow Client to be used as provider by implementing user, group,
sudorule interfaces.
Implement LocalSudoRule class.
Extend BareClient with BareClientTopologyController that
setups local domain using either files-provider or proxy files.
diff --git a/sssd_test_framework/roles/client.py b/sssd_test_framework/roles/client.py
@@ -10,7 +10,7 @@
 from ..utils.automount import AutomountUtils
 from ..utils.gdm import GDM
 from ..utils.ldb import LDBUtils
-from ..utils.local_users import LocalUsersUtils
+from ..utils.local_users import LocalGroup, LocalSudoRule, LocalUser, LocalUsersUtils
 from ..utils.realmd import RealmUtils
 from ..utils.sbus import DBUSDestination, DBUSKnownBus
 from ..utils.smartcard import SmartCardUtils
@@ -156,3 +156,37 @@ def sss_ssh_authorizedkeys(self, *args: str) -> ProcessResult:
         :rtype: ProcessResult
         """
         return self.host.conn.exec(["sss_ssh_authorizedkeys", *args], raise_on_error=False)
+
+    def user(self, name: str) -> LocalUser:
+        """
+        Get user object.
+
+        :param name: User name.
+        :type name: str
+        :return: New user object.
+        :rtype: LocalUser
+        """
+
+        return LocalUser(self.local, name)
+
+    def group(self, name: str) -> LocalGroup:
+        """
+        Get group object.
+        :param name: Group name.
+        :type name: str
+        :return: New group object.
+        :rtype: LocalGroup
+        """
+
+        return LocalGroup(self.local, name)
+
+    def sudorule(self, name: str) -> LocalSudoRule:
+        """
+        Get sudo rule object.
+        :param name: Sudo rule name.
+        :type name: str
+        :return: New sudo rule object.
+        :rtype: LocalSudoRule
+        """
+
+        return LocalSudoRule(self.local, name)
diff --git a/sssd_test_framework/topology.py b/sssd_test_framework/topology.py
@@ -10,6 +10,7 @@
 from .config import SSSDTopologyMark
 from .topology_controllers import (
     ADTopologyController,
+    BareClientTopologyController,
     ClientTopologyController,
     GDMTopologyController,
     IPATopologyController,
@@ -45,7 +46,7 @@ def test_ldap(client: Client, ldap: LDAP):
     BareClient = SSSDTopologyMark(
         name="bare_client",
         topology=Topology(TopologyDomain("sssd", client=1)),
-        controller=ClientTopologyController(),
+        controller=BareClientTopologyController(),
         fixtures=dict(client="sssd.client[0]", provider="sssd.client[0]"),
     )
     """
diff --git a/sssd_test_framework/topology_controllers.py b/sssd_test_framework/topology_controllers.py
@@ -10,8 +10,11 @@
 from .hosts.keycloak import KeycloakHost
 from .hosts.samba import SambaHost
 from .misc.ssh import retry_command
+from .utils.authselect import AuthselectUtils
+from .utils.sssd import SSSDUtils
 
 __all__ = [
+    "BareClientTopologyController",
     "LDAPTopologyController",
     "IPATopologyController",
     "ADTopologyController",
@@ -58,6 +61,22 @@ class ClientTopologyController(ProvisionedBackupTopologyController):
     pass
 
 
+class BareClientTopologyController(BackupTopologyController[SSSDMultihostConfig]):
+    """
+    Bare Client Topology Controller.
+    """
+
+    def topology_setup(self, client: ClientHost) -> None:
+        sssd = SSSDUtils(client, client.fs, client.svc, AuthselectUtils(client), load_config=False)
+        if client.features["files-provider"]:
+            # Use sssd with files provider when available
+            sssd.sssd["enable_files_domain"] = "true"
+        else:
+            # Use sssd with proxy-files provider
+            sssd.common.local()
+        sssd.start()
+
+
 class LDAPTopologyController(ProvisionedBackupTopologyController):
     """
     LDAP Topology Controller.
diff --git a/sssd_test_framework/utils/local_users.py b/sssd_test_framework/utils/local_users.py
@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 import jc
 from pytest_mh import MultihostHost, MultihostUtility
 from pytest_mh.cli import CLIBuilder, CLIBuilderArgs
@@ -12,6 +14,7 @@
     "LocalGroup",
     "LocalUser",
     "LocalUsersUtils",
+    "LocalSudoRule",
 ]
 
 
@@ -35,6 +38,7 @@ def __init__(self, host: MultihostHost, fs: LinuxFileSystem) -> None:
         self.fs: LinuxFileSystem = fs
         self._users: list[str] = []
         self._groups: list[str] = []
+        self._sudorules: list[LocalSudoRule] = []
 
     def teardown(self) -> None:
         """
@@ -53,6 +57,9 @@ def teardown(self) -> None:
         if cmd:
             self.host.conn.run("set -e\n\n" + cmd)
 
+        for rule in self._sudorules[:]:
+            rule.delete()
+
         super().teardown()
 
     def user(self, name: str) -> LocalUser:
@@ -129,6 +136,12 @@ def __init__(self, util: LocalUsersUtils, name: str) -> None:
         self.util = util
         self.name = name
 
+    def __str__(self):
+        """
+        Returns a string representation of the LocalUser.
+        """
+        return self.name
+
     def add(
         self,
         *,
@@ -276,6 +289,12 @@ def __init__(self, util: LocalUsersUtils, name: str) -> None:
         self.util = util
         self.name = name
 
+    def __str__(self):
+        """
+        Returns a string representation of the LocalGroup.
+        """
+        return self.name
+
     def add(
         self,
         *,
@@ -421,3 +440,174 @@ def remove_members(self, members: list[LocalUser]) -> LocalGroup:
         self.util.host.conn.run("set -ex\n" + cmd, log_level=ProcessLogLevel.Error)
 
         return self
+
+
+class LocalSudoRule(object):
+    """
+    Local sudo rule management.
+    """
+
+    def __init__(self, util: LocalUsersUtils, name: str) -> None:
+        """
+        :param util: LocalUsersUtils util object.
+        :param name: Sudo rule name.
+        :type name: str
+        """
+        self.name = name
+        self.util = util
+        self.__rule: dict[str, Any] = dict()
+        self.filename: str | None = None
+        self.rule_str: str | None = None
+
+    def __str__(self):
+        """
+        Returns a string representation of the LocalSudoRule.
+        """
+        if self.rule_str:
+            return self.rule_str
+        else:
+            return self.name
+
+    @staticmethod
+    def _format_list(item: str | Any | list[str | Any], add_percent: bool = False) -> str:
+        """
+        Format the item as a string.
+
+        :param item: object to be formatted
+        :type item: str | Any| list[str | Any]
+        :param add_percent: If true, prepend % to the item, defaults to False
+        :type add_percent: bool, optional
+        :return: Formatted string.
+        :rtype: str
+        """
+        if isinstance(item, list):
+            result = ", ".join(f"%{str(x)}" if isinstance(x, LocalGroup) and add_percent else str(x) for x in item)
+        else:
+            if isinstance(item, LocalGroup) and add_percent:
+                result = f"%{str(item)}"
+            else:
+                result = str(item)
+        return result
+
+    def add(
+        self,
+        *,
+        user: str | LocalUser | LocalGroup | list[str | LocalUser | LocalGroup] = "ALL",
+        host: str | list[str] = "ALL",
+        command: str | list[str] = "ALL",
+        option: str | list[str] | None = None,
+        runasuser: str | LocalUser | list[str | LocalUser] | None = None,
+        runasgroup: str | LocalGroup | list[str | LocalGroup] | None = None,
+        order: int | None = None,
+        nopasswd: bool | None = None,
+    ) -> LocalSudoRule:
+        """
+        Create new sudo rule.
+
+        :param user: sudoUser attribute, defaults to ALL
+        :type user: str | LocalUser | LocalGroup | list[str | LocalUser | LocalGroup]
+        :param host: sudoHost attribute, defaults to ALL
+        :type host: str | list[str],
+        :param command: sudoCommand attribute, defaults to ALL
+        :type command: str | list[str],
+        :param option: sudoOption attribute, defaults to None
+        :type option: str | list[str] | None, optional
+        :param runasuser: sudoRunAsUser attribute, defaults to None
+        :type runasuser: str | LocalUser | list[str | LocalUser] | None, optional
+        :param runasgroup: sudoRunAsGroup attribute, defaults to None
+        :type runasgroup: str | LocalGroup | list[str | LocalGroup] | None, optional
+        :param order: sudoOrder attribute, defaults to None
+        :type order: int | None, optional
+        :param nopasswd: If true, no authentication is required (NOPASSWD), defaults to None (no change)
+        :type nopasswd: bool | None, optional
+        :return: New sudo rule object.
+        :rtype: LocalSudoRule
+        """
+        orderstr = f"{order:02d}" if order is not None else str(len(self.util._sudorules))
+        if self.filename is None:
+            self.filename = f"{orderstr}_{self.name}"
+
+        # Remember arguments so we can use them in modify if needed
+        self.__rule = dict[str, Any](
+            user=user,
+            host=host,
+            command=command,
+            option=option,
+            runasuser=runasuser,
+            runasgroup=runasgroup,
+            order=order,
+            nopasswd=nopasswd,
+        )
+        run_as_str = ""
+        if runasuser or runasgroup:
+            run_as_str += "("
+            if runasuser:
+                run_as_str += LocalSudoRule._format_list(runasuser)
+            if runasgroup:
+                run_as_str += f":{LocalSudoRule._format_list(runasgroup)}"
+            run_as_str += ")"
+        user_str = LocalSudoRule._format_list(user, add_percent=True)
+        host_str = LocalSudoRule._format_list(host)
+        tagspec_str = "NOPASSWD:" if nopasswd else ""
+        command_str = LocalSudoRule._format_list(command)
+        rule_str = f"{user_str} {host_str}={run_as_str} {tagspec_str} {command_str}\n"
+        self.rule_str = rule_str
+        self.util.fs.write(f"/etc/sudoers.d/{self.filename}", self.rule_str)
+        self.util._sudorules.append(self)
+        return self
+
+    def modify(
+        self,
+        *,
+        user: str | LocalUser | LocalGroup | list[str | LocalUser | LocalGroup] | None = None,
+        host: str | list[str] | None = None,
+        command: str | list[str] | None = None,
+        option: str | list[str] | None = None,
+        runasuser: str | LocalUser | list[str | LocalUser] | None = None,
+        runasgroup: str | LocalGroup | list[str | LocalGroup] | None = None,
+        order: int | None = None,
+        nopasswd: bool | None = None,
+    ) -> LocalSudoRule:
+        """
+        Modify existing Local sudo rule.
+
+        :param user: sudoUser attribute, defaults to None
+        :type user: str | LocalUser | LocalGroup | list[str | LocalUser | LocalGroup] | None, optional
+        :param host: sudoHost attribute, defaults to None
+        :type host: str | list[str] | None, optional
+        :param command: sudoCommand attribute defaults to None
+        :type command: str | list[str] | None, optional
+        :param option: sudoOption attribute, defaults to None
+        :type option: str | list[str] | None, optional
+        :param runasuser: sudoRunAsUser attribute, defaults to None
+        :type runasuser: str | LocalUser | list[str | LocalUser] | None, optional
+        :param runasgroup: sudoRunAsGroup attribute, defaults to None
+        :type runasgroup: str | LocalGroup | list[str | LocalGroup] | None, optional
+        :param order: sudoOrder attribute, defaults to None
+        :type order: int | None, optional
+        :param nopasswd: If true, no authentication is required (NOPASSWD), defaults to None (no change)
+        :type nopasswd: bool | None, optional
+        :return:  New sudo rule object.
+        :rtype: LocalSudoRule
+        """
+        self.delete()
+        self.add(
+            user=user if user is not None else self.__rule.get("user", "ALL"),
+            host=host if host is not None else self.__rule.get("host", "ALL"),
+            command=command if command is not None else self.__rule.get("command", "ALL"),
+            option=option if option is not None else self.__rule.get("option", None),
+            runasuser=runasuser if runasuser is not None else self.__rule.get("runasuser", None),
+            runasgroup=runasgroup if runasgroup is not None else self.__rule.get("runasgroup", None),
+            order=order if order is not None else self.__rule.get("order", None),
+            nopasswd=nopasswd if nopasswd is not None else self.__rule.get("nopasswd", None),
+        )
+        return self
+
+    def delete(self) -> None:
+        """
+        Delete local sudo rule.
+        """
+        if self.filename:
+            self.util.fs.rm(f"/etc/sudoers.d/{self.filename}")
+        if self in self.util._sudorules:
+            self.util._sudorules.remove(self)
