Adding upstream version 2:4.20.0+dfsg.upstream/2%4.20.0+dfsg

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 229 insertions, 0 deletions

diff --git a/python/samba/tests/samba_tool/silo_base.py b/python/samba/tests/samba_tool/silo_base.py
new file mode 100644
index 0000000..451d330
--- /dev/null
+++ b/python/samba/tests/samba_tool/silo_base.py
@@ -0,0 +1,229 @@
+# Unix SMB/CIFS implementation.
+#
+# Base test class for samba-tool domain auth policy and silo commands.
+#
+# Copyright (C) Catalyst.Net Ltd. 2023
+#
+# Written by Rob van der Linde <rob@catalyst.net.nz>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import os
+
+from ldb import SCOPE_ONELEVEL
+
+from samba.netcmd.domain.models import Group
+
+from .base import SambaToolCmdTest
+
+HOST = "ldap://{DC_SERVER}".format(**os.environ)
+CREDS = "-U{DC_USERNAME}%{DC_PASSWORD}".format(**os.environ)
+
+
+class SiloTest(SambaToolCmdTest):
+    """Base test class for silo and policy related commands."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.samdb = cls.getSamDB("-H", HOST, CREDS)
+        super().setUpClass()
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.create_authentication_policy(name="User Policy")
+        cls.create_authentication_policy(name="Service Policy")
+        cls.create_authentication_policy(name="Computer Policy")
+
+        cls.create_authentication_silo(
+            name="Developers",
+            description="Developers, Developers, Developers!",
+            user_authentication_policy="User Policy")
+        cls.create_authentication_silo(
+            name="Managers",
+            description="Managers",
+            user_authentication_policy="User Policy")
+        cls.create_authentication_silo(
+            name="QA",
+            description="Quality Assurance",
+            user_authentication_policy="User Policy",
+            service_authentication_policy="Service Policy",
+            computer_authentication_policy="Computer Policy")
+
+        cls.device_group = Group(name="device-group")
+        cls.device_group.save(cls.samdb)
+        cls.addClassCleanup(cls.device_group.delete, cls.samdb)
+
+    def get_services_dn(self):
+        """Returns Services DN."""
+        services_dn = self.samdb.get_config_basedn()
+        services_dn.add_child("CN=Services")
+        return services_dn
+
+    def get_authn_configuration_dn(self):
+        """Returns AuthN Configuration DN."""
+        authn_policy_configuration = self.get_services_dn()
+        authn_policy_configuration.add_child("CN=AuthN Policy Configuration")
+        return authn_policy_configuration
+
+    def get_authn_silos_dn(self):
+        """Returns AuthN Silos DN."""
+        authn_silos_dn = self.get_authn_configuration_dn()
+        authn_silos_dn.add_child("CN=AuthN Silos")
+        return authn_silos_dn
+
+    def get_authn_policies_dn(self):
+        """Returns AuthN Policies DN."""
+        authn_policies_dn = self.get_authn_configuration_dn()
+        authn_policies_dn.add_child("CN=AuthN Policies")
+        return authn_policies_dn
+
+    def get_users_dn(self):
+        """Returns Users DN."""
+        users_dn = self.samdb.get_root_basedn()
+        users_dn.add_child("CN=Users")
+        return users_dn
+
+    def get_user(self, username):
+        """Get a user by username."""
+        users_dn = self.get_users_dn()
+
+        result = self.samdb.search(base=users_dn,
+                                   scope=SCOPE_ONELEVEL,
+                                   expression=f"(sAMAccountName={username})")
+
+        if len(result) == 1:
+            return result[0]
+
+    @classmethod
+    def _run(cls, *argv):
+        """Override _run, so we don't always have to pass host and creds."""
+        args = list(argv)
+        args.extend(["-H", HOST, CREDS])
+        return super()._run(*args)
+
+    runcmd = _run
+    runsubcmd = _run
+
+    @classmethod
+    def create_authentication_policy(cls, name, description=None, audit=False,
+                                     protect=False):
+        """Create an authentication policy."""
+
+        # base command for create authentication policy
+        cmd = ["domain", "auth", "policy", "create", "--name", name]
+
+        # optional attributes
+        if description is not None:
+            cmd.append(f"--description={description}")
+        if audit:
+            cmd.append("--audit")
+        if protect:
+            cmd.append("--protect")
+
+        # Run command and store name in self.silos for tearDownClass to clean
+        # up.
+        result, out, err = cls.runcmd(*cmd)
+        assert result is None
+        assert out.startswith("Created authentication policy")
+        cls.addClassCleanup(cls.delete_authentication_policy,
+                            name=name, force=True)
+        return name
+
+    @classmethod
+    def delete_authentication_policy(cls, name, force=False):
+        """Delete authentication policy by name."""
+        cmd = ["domain", "auth", "policy", "delete", "--name", name]
+
+        # Force-delete protected authentication policy.
+        if force:
+            cmd.append("--force")
+
+        result, out, err = cls.runcmd(*cmd)
+        assert result is None
+        assert "Deleted authentication policy" in out
+
+    @classmethod
+    def create_authentication_silo(cls, name, description=None,
+                                   user_authentication_policy=None,
+                                   service_authentication_policy=None,
+                                   computer_authentication_policy=None,
+                                   audit=False, protect=False):
+        """Create an authentication silo using the samba-tool command."""
+
+        # Base command for create authentication policy.
+        cmd = ["domain", "auth", "silo", "create", "--name", name]
+
+        # Authentication policies.
+        if user_authentication_policy:
+            cmd += ["--user-authentication-policy",
+                    user_authentication_policy]
+        if service_authentication_policy:
+            cmd += ["--service-authentication-policy",
+                    service_authentication_policy]
+        if computer_authentication_policy:
+            cmd += ["--computer-authentication-policy",
+                    computer_authentication_policy]
+
+        # Other optional attributes.
+        if description is not None:
+            cmd.append(f"--description={description}")
+        if protect:
+            cmd.append("--protect")
+        if audit:
+            cmd.append("--audit")
+
+        # Run command and store name in self.silos for tearDownClass to clean
+        # up.
+        result, out, err = cls.runcmd(*cmd)
+        assert result is None
+        assert out.startswith("Created authentication silo")
+        cls.addClassCleanup(cls.delete_authentication_silo,
+                            name=name, force=True)
+        return name
+
+    @classmethod
+    def delete_authentication_silo(cls, name, force=False):
+        """Delete authentication silo by name."""
+        cmd = ["domain", "auth", "silo", "delete", "--name", name]
+
+        # Force-delete protected authentication silo.
+        if force:
+            cmd.append("--force")
+
+        result, out, err = cls.runcmd(*cmd)
+        assert result is None
+        assert "Deleted authentication silo" in out
+
+    def get_authentication_silo(self, name):
+        """Get authentication silo by name."""
+        authn_silos_dn = self.get_authn_silos_dn()
+
+        result = self.samdb.search(base=authn_silos_dn,
+                                   scope=SCOPE_ONELEVEL,
+                                   expression=f"(CN={name})")
+
+        if len(result) == 1:
+            return result[0]
+
+    def get_authentication_policy(self, name):
+        """Get authentication policy by name."""
+        authn_policies_dn = self.get_authn_policies_dn()
+
+        result = self.samdb.search(base=authn_policies_dn,
+                                   scope=SCOPE_ONELEVEL,
+                                   expression=f"(CN={name})")
+
+        if len(result) == 1:
+            return result[0]