infoblox.py

import ipaddress
from logging import getLogger

from infoblox_client import connector, objects
from infoblox_client.exceptions import InfobloxCannotCreateObject, InfobloxCannotUpdateObject

from gso.settings import IPAMParams, load_oss_params

logger = getLogger(__name__)


class AllocationError(Exception):
    pass


class DeletionError(Exception):
    pass


def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
    """Set up a new connection with an Infoblox instance.

    :return: A tuple that has an Infoblox `Connector` instance, and {term}`IPAM` parameters.
    :rtype: tuple[{class}`infoblox_client.connector.Connector`, IPAMParams]
    """
    oss = load_oss_params().IPAM
    options = {
        "host": oss.INFOBLOX.host,
        "username": oss.INFOBLOX.username,
        "password": oss.INFOBLOX.password,
        "wapi_version": oss.INFOBLOX.wapi_version,
        "ssl_verify": True if oss.INFOBLOX.scheme == "https" else False,
    }
    return connector.Connector(options), oss


def _allocate_network(
    conn: connector.Connector,
    dns_view: str,
    netmask: int,
    containers: list[str],
    comment: str | None = "",
) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
    """Allocate a new network in Infoblox.

    The function will go over all given containers, and try to allocate a network within the available IP space. If no
    space is available, this method raises an {class}`AllocationError`.

    :param conn: An active Infoblox connection.
    :type conn: {class}`infoblox_client.connector.Connector`
    :param dns_view: The Infoblox `dns_view` in which the network should be allocated.
    :type dns_view: str
    :param netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
    :type netmask: int
    :param containers: A list of network containers in which the network should be allocated, given in {term}`CIDR`
                       notation.
    :type containers: list[str]
    :param comment: Optionally, a comment can be added to the network allocation.
    :type comment: str, optional
    """
    for container in [ipaddress.ip_network(con) for con in containers]:
        for network in container.subnets(new_prefix=netmask):
            if objects.Network.search(conn, network=str(network)) is None:
                created_net = objects.Network.create(conn, network=str(network), dns_view=dns_view, comment=comment)
                if created_net.response != "Infoblox Object already Exists":
                    return ipaddress.ip_network(created_net.network)
        logger.warning(f"IP container {container} appears to be full.")

    raise AllocationError(f"Cannot allocate anything in {containers}, check whether any IP space is available.")


def hostname_available(hostname: str) -> bool:
    """Check whether a hostname is still available **in Infoblox**.

    Check whether Infoblox already contains a {class}`infoblox_client.objects.HostRecord` that matches the given
    hostname. Be aware that this method only checks within the Infoblox instance, and not the rest of the internet.
    The hostname could therefore still be taken elsewhere.

    :param hostname: The hostname to be checked.
    :type hostname: str
    """
    conn, _ = _setup_connection()
    return objects.HostRecord.search(conn, name=hostname) is None


def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddress.IPv4Network:
    """Allocate a new IPv4 network in Infoblox.

    Allocate an IPv4 network for a specific service type. The service type should be defined in the {term}`OSS`
    parameters of {term}`GSO`, from which the containers and netmask will be used.

    :param service_type: The service type for which the network is allocated.
    :type service_type: str
    :param comment: A comment to be added to the allocated network in Infoblox.
    :type comment: str, optional
    """
    conn, oss = _setup_connection()
    netmask = getattr(oss, service_type).V4.mask
    containers = getattr(oss, service_type).V4.containers
    dns_view = getattr(oss, service_type).dns_view

    return ipaddress.IPv4Network(_allocate_network(conn, dns_view, netmask, containers, comment))


def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddress.IPv6Network:
    """Allocate a new IPv6 network in Infoblox.

    Allocate an IPv6 network for a specific service type. The service type should be defined in the {term}`OSS`
    parameters of {term}`GSO`, from which the containers and netmask will be used.

    :param service_type: The service type for which the network is allocated.
    :type service_type: str
    :param comment: A comment to be added to the allocated network in Infoblox.
    :type comment: str, optional
    """
    conn, oss = _setup_connection()
    netmask = getattr(oss, service_type).V6.mask
    containers = getattr(oss, service_type).V6.containers
    dns_view = getattr(oss, service_type).dns_view

    return ipaddress.IPv6Network(_allocate_network(conn, dns_view, netmask, containers, comment))


def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) -> None:
    """Delete a network in Infoblox.

    Delete a network that is allocated in Infoblox, by passing the {term}`CIDR` to be deleted. The {term}`CIDR` must
    exactly match an existing entry in Infoblox, otherwise this method raises a {class}`DeletionError`

    :param ip_network: The network that should get deleted.
    :type ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network
    """
    conn, _ = _setup_connection()
    network = objects.Network.search(conn, cidr=str(ip_network))
    if network:
        network.delete()
    else:
        raise DeletionError(f"Could not find network {ip_network}, nothing has been deleted.")


def allocate_host(
    hostname: str, service_type: str, cname_aliases: list[str], comment: str | None = ""
) -> tuple[ipaddress.IPv4Address, ipaddress.IPv6Address]:
    """Allocate a new host record in Infoblox.

    Create a new host record in Infoblox, by providing a hostname, and the service type that is associated with this new
    host. Most likely to be a loopback interface. If the hostname is not available in Infoblox (due to a potential
    collision) this method raises an {class}`AllocationError`.

    :param hostname: The {term}`FQDN` of the new host
    :type hostname: str
    :param service_type: The service type from which IP resources should be used.
    :type service_type: str
    :param cname_aliases: A list of any {term}`CNAME` aliases that should be associated with this host. Most often this
                          will be a single loopback address.
    :type cname_aliases: list[str]
    :param comment: Optionally, a comment can be added to the host record in Infoblox.
    :type comment: str, optional
    """
    if not hostname_available(hostname):
        raise AllocationError(f"Cannot allocate new host, FQDN {hostname} already taken.")

    conn, oss = _setup_connection()
    allocation_networks_v4 = getattr(oss, service_type).V4.networks
    allocation_networks_v6 = getattr(oss, service_type).V6.networks
    dns_view = getattr(oss, service_type).dns_view

    created_v6 = None
    for ipv6_range in allocation_networks_v6:
        v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv6_range))
        ipv6_object = objects.IP.create(ip=v6_alloc, mac="00:00:00:00:00:00", configure_for_dhcp=False)
        try:
            new_host = objects.HostRecord.create(
                conn, ip=ipv6_object, name=hostname, aliases=cname_aliases, comment=comment, dns_view=dns_view
            )
            created_v6 = ipaddress.IPv6Address(new_host.ipv6addr)
        except InfobloxCannotCreateObject:
            logger.warning(f"Cannot find 1 available IP address in network {ipv6_range}.")

    if created_v6 is None:
        raise AllocationError(f"Cannot find 1 available IP address in networks {allocation_networks_v6}.")

    created_v4 = None
    for ipv4_range in allocation_networks_v4:
        v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv4_range))
        ipv4_object = objects.IP.create(ip=v4_alloc, mac="00:00:00:00:00:00", configure_for_dhcp=False)
        new_host = objects.HostRecord.search(conn, name=hostname)
        new_host.ipv4addrs = [ipv4_object]
        try:
            new_host.update()
            new_host = objects.HostRecord.search(conn, name=hostname)
            created_v4 = ipaddress.IPv4Address(new_host.ipv4addr)
        except InfobloxCannotUpdateObject:
            logger.warning(f"Cannot find 1 available IP address in network {ipv4_range}.")

    if created_v4 is None:
        raise AllocationError(f"Cannot find 1 available IP address in networks {allocation_networks_v4}.")

    return created_v4, created_v6


def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> None:
    """Delete a host from Infoblox.

    Delete a host record in Infoblox, by providing the IP address that is associated with the record. Raises a
    {class}`DeletionError` if no record can be found in Infoblox.

    :param ip_addr: The IP address of the host record that should get deleted.
    :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address
    """
    conn, _ = _setup_connection()
    host = objects.HostRecord.search(conn, ipv4addr=ip_addr)
    if host:
        host.delete()
    else:
        raise DeletionError(f"Could not find host at {ip_addr}, nothing has been deleted.")


def delete_host_by_fqdn(fqdn: str) -> None:
    """Delete a host from Infoblox.

    Delete a host record in Infoblox, by providing the {term}`FQDN` that is associated with the record. Raises a
    {class}`DeletionError` if no record can be found in Infoblox.

    :param fqdn: The FQDN of the host record that should get deleted.
    :type fqdn: str
    """
    conn, _ = _setup_connection()
    host = objects.HostRecord.search(conn, name=fqdn)
    host.delete()