scripts/nbhelper/service.py

#!/usr/bin/env python3

# SPDX-FileCopyrightText: © 2021 Open Networking Foundation <support@opennetworking.org>
# SPDX-License-Identifier: Apache-2.0

# service.py

import netaddr

from .container import ServiceInfoContainer


def getIPaddress(addressWithMask):
    return str(netaddr.IPNetwork(addressWithMask).ip)


def dhcpSubnetConfigGenerator():
    dhcpSubnetConfigs = list()

    serviceInfoContainer = ServiceInfoContainer()
    serviceInfoContainer.initialize()

    for domainName, domain in serviceInfoContainer.all():
        subnetConfig = {
            "dns_search": [domainName],
            "dns_servers": [domain["dnsServer"]["address"]],
            "ntp_servers": [domain["ntpServer"]["address"]],
            "tftpd_server": domain["dhcpServer"]["address"],
            "range": domain["dhcprange"][0],
            "subnet": domain["subnet"],
            "routers": domain["router"],
            "hosts": list(),
        }

        for address, host in domain["hosts"].items():
            subnetConfig["hosts"].append(
                {
                    "ip_addr": getIPaddress(address),
                    "mac_addr": host["macaddr"],
                    "name": host["hostname"],
                }
            )

        subnetConfig["hosts"] = sorted(
            subnetConfig["hosts"], key=lambda x: int(x["ip_addr"].split(".")[-1])
        )
        dhcpSubnetConfigs.append(subnetConfig)

    return dhcpSubnetConfigs


def dnsFowardZoneConfigGenerator():
    def getDomainNameByIP(ip_address):
        """
        getDomainNameByIP will return the corresponding domain name of an IP address
        In the ntpServer, dhcpServer, dnsServer we only have the IP addresses of them,
        But we can use the dhcp subnet configuration to find the FQDN
        """
        dhcpSubnetConfigs = dhcpSubnetConfigGenerator()

        for domain in dhcpSubnetConfigs:
            domainName = domain["dns_search"][0]
            for host in domain["hosts"]:
                if ip_address == host["ip_addr"]:
                    return f"{host['name']}.{domainName}."
        return ""

    dnsForwardZoneConfigs = dict()

    serviceInfoContainer = ServiceInfoContainer()
    serviceInfoContainer.initialize()

    for domainName, domain in serviceInfoContainer.all():
        forwardZoneConfig = {
            "cname": dict(),
            "a": dict(),
            "ns": list(),
            "srv": dict(),
            "txt": dict(),
        }

        # Get the services set to this Tenant network
        ntpServer = domain["ntpServer"] or None
        dhcpServer = domain["dhcpServer"] or None
        dnsServer = domain["dnsServer"] or None

        # If service exists, set the FQDN to CNAME records
        if ntpServer:
            forwardZoneConfig["cname"]["ntp"] = getDomainNameByIP(ntpServer["address"])
        if dhcpServer:
            forwardZoneConfig["cname"]["tftp"] = getDomainNameByIP(
                dhcpServer["address"]
            )
        if dnsServer:
            forwardZoneConfig["cname"]["ns"] = getDomainNameByIP(dnsServer["address"])
            forwardZoneConfig["ns"].append(getDomainNameByIP(dnsServer["address"]))

        for address, host in domain["hosts"].items():
            # Add exist IP address into dnsReverseZoneConfigs,
            hostname = host["hostname"]
            forwardZoneConfig["a"][hostname] = address

        for address in netaddr.IPSet(domain["dhcprange"]):
            # If address exists in ServiceInfoContainer's host dictionary,
            # Use the pre-generated hostname as A record's name
            hostname = "dhcp%03d" % address.words[-1]
            forwardZoneConfig["a"][hostname] = str(address)

        dnsForwardZoneConfigs[domainName] = forwardZoneConfig

    return dnsForwardZoneConfigs


def dnsReverseZoneConfigGenerator():
    def canonicalize_rfc1918_prefix(prefix):
        """
        RFC1918 prefixes need to be expanded to their widest canonical range to
        group all reverse lookup domains together for reverse DNS with NSD/Unbound.
        """

        pnet = netaddr.IPNetwork(prefix)
        (o1, o2, o3, o4) = pnet.network.words  # Split ipv4 octets
        cidr_plen = pnet.prefixlen

        if o1 == 10:
            o2 = o3 = o4 = 0
            cidr_plen = 8
        elif (o1 == 172 and 16 <= o2 <= 31) or (o1 == 192 and o2 == 168):
            o3 = o4 = 0
            cidr_plen = 16

        return "%s/%d" % (".".join(map(str, [o1, o2, o3, o4])), cidr_plen)

    serviceInfoContainer = ServiceInfoContainer()
    serviceInfoContainer.initialize()

    # dnsReverseZoneConfigs contains all reverse zone records.
    dnsReverseZoneConfigs = dict()
    widedomain = None

    for domainName, domain in serviceInfoContainer.all():

        # Expand network range to group all tenant domains
        widedomain = widedomain or canonicalize_rfc1918_prefix(domain["subnet"])

        # Create the basic structure of reverse zone config
        # {"10.0.0.0/8": {"ns": list(), "ptr": dict()}}
        dnsReverseZoneConfigs.setdefault(widedomain, dict())
        dnsReverseZoneConfigs[widedomain].setdefault("ns", list())
        dnsReverseZoneConfigs[widedomain].setdefault("ptr", dict())

        # Get the DNS services set to this Tenant network
        dnsServer = domain["dnsServer"]["name"] if domain["dnsServer"] else None

        # If service exists, set the FQDN to CNAME records
        if dnsServer:
            dnsReverseZoneConfigs[widedomain]["ns"].append(f"{domainName}.{dnsServer}.")

        for address, host in domain["hosts"].items():
            # Add exist IP address into dnsReverseZoneConfigs,
            hostname = host["hostname"]
            dnsReverseZoneConfigs[widedomain]["ptr"][
                address
            ] = f"{hostname}.{domainName}."

        for address in netaddr.IPSet(domain["dhcprange"]):
            # Add DHCP range IP address into dnsReverseZoneConfigs,
            # Use the pre-generated hostname as A record's name
            hostname = "dhcp%03d" % address.words[3]
            dnsReverseZoneConfigs[widedomain]["ptr"][
                str(address)
            ] = f"{hostname}.{domainName}."

    dnsReverseZoneConfigs[widedomain]["ptr"] = dict(
        sorted(
            dnsReverseZoneConfigs[widedomain]["ptr"].items(),
            key=lambda x: (int(x[0].split(".")[-2]), int(x[0].split(".")[-1])),
        )
    )

    return dnsReverseZoneConfigs