adding dns server and zone utilities to the samba role

Author: Dan Lavu

sssd_test_framework/roles/samba.py

@@ -11,7 +11,7 @@
 from pytest_mh.conn import ProcessResult
 
 from ..hosts.samba import SambaHost
-from ..misc import attrs_parse, to_list_of_strings
+from ..misc import attrs_parse, ip_version, to_list_of_strings
 from ..utils.ldap import LDAPRecordAttributes
 from .base import BaseLinuxLDAPRole, BaseObject, DeleteAttribute
 from .generic import GenericPasswordPolicy
@@ -28,6 +28,8 @@
     "SambaAutomount",
     "SambaSudoRule",
     "SambaGPO",
+    "SambaDNSServer",
+    "SambaDNSZone",
 ]
 
 
@@ -301,6 +303,38 @@ def test_example(client: Client, samba: Samba):
         """
         return SambaComputer(self, name)
 
+    def dns(self) -> SambaDNSServer:
+        """
+        Get DNS server object.
+
+            Get methods use dig and is parsed by jc. The data from jc contains several nested dict,
+            but two are returned as a tuple, ``answer, authority``.
+
+        .. code-block:: python
+            :caption: Example usage
+
+            # Create forward zone and add forward record
+            zone = samba.dns().zone("example.test").create()
+            zone.add_record("client", "172.16.200.15")
+
+            # Create reverse zone and add reverse record
+            zone = samba.dns().zone("10.0.10.in-addr.arpa").create()
+            zone.add_ptr_record("client.example.test", 15)
+
+            # Add forward record to default domain
+            samba.dns().zone(samba.domain).add_record("client", "1.2.3.4")
+
+            # Add a global forwarder
+            samba.dns().add_forwarder("1.1.1.1")
+
+            # Remove a global forwarder
+            samba.dns().remove_forwarder("1.1.1.1")
+
+             # Clear all forwarders
+             samba.dns().clear_forwarders()
+        """
+        return SambaDNSServer(self)
+
     def gpo(self, name: str) -> SambaGPO:
         """
         Get group policy object.
@@ -887,8 +921,8 @@ class SambaComputer(SambaObject):
 
     def __init__(self, role: Samba, name: str) -> None:
         """
-        :param role: AD role object.
-        :type role: AD
+        :param role: Samba role object.
+        :type role: Samba
         :param name: Computer name.
         :type name: str
         """
@@ -1172,6 +1206,206 @@ def lockout(self, duration: int, attempts: int) -> SambaPasswordPolicy:
         return self
 
 
+class SambaDNSServer(BaseObject[SambaHost, Samba]):
+    """
+    DNS management utilities.
+    """
+
+    def __init__(self, role: Samba):
+        """
+        :param role: Samba host object.
+        :type role: SambaHost
+        """
+        super().__init__(role)
+
+        self.domain: str = role.domain
+        """Domain name."""
+
+        self.server: str = role.server
+        """Server name."""
+
+        self.naming_context: str = role.naming_context
+        """Naming context."""
+
+        self.credentials: str = f" --username={self.role.host.adminuser} --password={self.role.host.adminpw}"
+        """Credentials to manage GPOs."""
+
+        self.smb_conf: str = "/etc/samba/smb.conf"
+
+    def zone(self, name: str) -> SambaDNSZone:
+        """
+        Get SambaDNSZone object.
+
+        :param name: Zone name.
+        :type name: str
+        :return: SambaDNSZone object.
+        :rtype: SambaDNSZone
+        """
+        return SambaDNSZone(self.role, name)
+
+    def get_forwarders(self) -> list[str] | None:
+        """
+        Get DNS global forwarders.
+
+        Global forwarders are configured in /etc/smb.conf
+
+        :return: DNS global forwarders.
+        :rtype: list[str]
+        """
+        result = [line.strip() for line in self.host.fs.read(self.smb_conf).split("\n")]
+        if result is not None and isinstance(result, list):
+            for i in result:
+                if "dns forwarder" in i:
+                    # The additional split is to support more than one server
+                    return i.split("=")[1].strip().split(" ")
+        return None
+
+    def add_forwarder(self, ip_address: str) -> SambaDNSServer:
+        """
+        Add a DNS server forwarder.
+
+        :param ip_address: IP address.
+        :type ip_address: str
+        :return:  Self.
+        :rtype: SambaDNSServer
+        """
+        self.host.fs.backup(self.smb_conf)
+        self.host.fs.sed(f"s/dns forwarder = .*/ & {ip_address}/", self.smb_conf, ["-i"])
+        self.host.svc.reload("samba.service")
+        return self
+
+    def remove_forwarder(self, ip_address: str) -> None:
+        """
+        Remove a DNS server forwarder.
+
+        :param ip_address: IP address.
+        :type ip_address: str
+        """
+        ip_address = ip_address.replace(".", "\\.").strip()
+        self.host.fs.backup(self.smb_conf)
+        self.host.fs.sed(f"/dns forwarder/s/ {ip_address}//", self.smb_conf, ["-i"])
+        self.host.svc.reload("samba.service")
+
+    def clear_forwarders(self) -> None:
+        """
+        Clear all DNS server forwarders.
+
+        Samba has one global forwarder enabled  by default.
+        """
+        forwarders = self.get_forwarders()
+
+        if isinstance(forwarders, list) and not None:
+            for forwarder in forwarders:
+                self.remove_forwarder(forwarder)
+
+    def list_zones(self) -> list[str]:
+        """
+        List zones.
+
+        :return: List of zones.
+        :rtype: list[str]
+        """
+        result = self.host.conn.run(f"samba-tool dns zonelist {self.server} {self.credentials}").stdout_lines
+        result = [i for i in result if "pszZoneName" in i]
+        result = [z.split(":")[1].strip() for z in result]
+
+        return result
+
+
+class SambaDNSZone(SambaDNSServer):
+    """
+    DNS zone management.
+    """
+
+    def __init__(self, role: Samba, name: str):
+        """
+        :param role: Samba host object.
+        :type role: SambaHost
+        :param name: DNS zone name.
+        :type name: str
+        """
+        super().__init__(role)
+
+        self.zone_name: str = name
+        """Zone name."""
+
+    def create(self) -> SambaDNSZone:
+        """
+        Create new zone.
+
+        :return: SambaDNSServer object.
+        :rtype: SambaDNSServer
+        """
+        self.host.conn.run(f"samba-tool dns zonecreate {self.server} {self.zone_name} {self.credentials}")
+        return self
+
+    def delete(self) -> None:
+        """
+        Delete zone.
+        """
+        self.host.conn.run(f"samba-tool dns zonedelete {self.server} {self.zone_name} {self.credentials}")
+
+    def add_record(self, name: str, data: str) -> SambaDNSZone:
+        """
+        Add DNS record.
+
+        If ``data`` is a str, a forward record will be added.
+        If an integer a reverse record will be added.
+
+        :param name: Record name.
+        :type name: str | int
+        :param data: Record data.
+        :type data: str
+        :return: SambaDNSZone object.
+        :rtype: SambaDNSZone
+        """
+        args = ""
+
+        if isinstance(data, int):
+            args = f" {name} PTR {str(data)} {self.credentials}"
+        elif isinstance(data, str) and ip_version(data) == 4:
+            args = f" {name} A {data} {self.credentials}"
+        elif isinstance(data, str) and ip_version(data) == 6:
+            args = f" {name} AAAA {data} {self.credentials}"
+
+        self.host.conn.run(f"samba-tool dns add {self.server} {self.zone_name} {args}")
+        return self
+
+    def delete_record(self, name: str) -> None:
+        """
+        Delete DNS record.
+
+        :param name: Name of the record.
+        :type name: str
+        """
+        if "in-addr" in self.zone_name:
+            record_type = "PTR"
+            data = self.host.conn.run(f"dig -x +short {name}").stdout.strip()
+        else:
+            data = self.host.conn.run(f"dig +short {name}").stdout.strip()
+            record_type = "AAAA" if ":" in data else "A"
+
+        self.role.host.conn.run(
+            f"samba-tool dns delete "
+            f"{self.server} {self.zone_name} "
+            f"{name} {record_type} {data} "
+            f"{self.credentials}"
+        )
+
+    def print(self) -> str:
+        """
+        Prints all dns records in a zone as text.
+
+        :return: Print zone data.
+        :rtype: str
+        """
+        result = self.host.conn.run(
+            f"samba-tool dns query {self.server} {self.zone_name} @ ALL {self.credentials}"
+        ).stdout
+
+        return result
+
+
 SambaOrganizationalUnit: TypeAlias = LDAPOrganizationalUnit[SambaHost, Samba]
 SambaAutomount: TypeAlias = LDAPAutomount[SambaHost, Samba]
 SambaSudoRule: TypeAlias = LDAPSudoRule[SambaHost, Samba, SambaUser, SambaGroup]