ipam.py

# mypy: ignore-errors
import ipaddress
from enum import Enum
from typing import Optional, Tuple, Union

import requests
from pydantic import BaseSettings
from requests.auth import HTTPBasicAuth

from gso import settings


class V4ServiceNetwork(BaseSettings):
    v4: ipaddress.IPv4Network


class V6ServiceNetwork(BaseSettings):
    v6: ipaddress.IPv6Network


class ServiceNetworks(BaseSettings):
    v4: ipaddress.IPv4Network
    v6: ipaddress.IPv6Network


class V4HostAddress(BaseSettings):
    v4: ipaddress.IPv4Address


class V6HostAddress(BaseSettings):
    v6: ipaddress.IPv6Address


class HostAddresses(BaseSettings):
    v4: ipaddress.IPv4Address
    v6: ipaddress.IPv6Address


class IPAMErrors(Enum):
    # HTTP error code, match in error message
    CONTAINER_FULL = 400, "Can not find requested number of networks"
    NETWORK_FULL = 400, "Cannot find 1 available IP address(es) in this network"


REQUESTS_TIMEOUT = 20


# TODO: remove this!
# lab infoblox cert is not valid for the ipv4 address
#   ... disable warnings for now
requests.packages.urllib3.disable_warnings()


def match_error_code(response, error_code):
    return response.status_code == error_code.value[0] and error_code.value[1] in response.text


def wapi(infoblox_params: settings.InfoBloxParams):
    return f"https://{infoblox_params.host}" f"/wapi/{infoblox_params.wapi_version}"


def ip_addr_version(addr: str = ""):
    ip_version = None
    ip_addr = ipaddress.ip_address(addr)
    if isinstance(ip_addr, ipaddress.IPv4Address):
        ip_version = 4
    elif isinstance(ip_addr, ipaddress.IPv6Address):
        ip_version = 6
    assert ip_version in [4, 6]
    return ip_version


def ip_network_version(network: str = ""):
    ip_version = None
    ip_network = ipaddress.ip_network(network)
    if isinstance(ip_network, ipaddress.IPv4Network):
        ip_version = 4
    elif isinstance(ip_network, ipaddress.IPv6Network):
        ip_version = 6
    assert ip_version in [4, 6]
    return ip_version


def assert_host_in_service(
    ipv4_addr: str = "",
    ipv6_addr: str = "",
    oss_ipv4_containers=None,
    oss_ipv6_containers=None,
    oss_ipv4_networks=None,
    oss_ipv6_networks=None,
):
    # IPv4
    if oss_ipv4_containers:
        assert any(
            ipv4_addr in oss_ipv4_container for oss_ipv4_container in oss_ipv4_containers
        ), "Host's IPv4 address doesn't belong to service type."
    else:
        assert any(
            ipv4_addr in oss_ipv4_network for oss_ipv4_network in oss_ipv4_networks
        ), "Host's IPv4 address doesn't belong to service type."

    # IPv6
    if oss_ipv6_containers:
        assert any(
            ipv6_addr in oss_ipv6_container for oss_ipv6_container in oss_ipv6_containers
        ), "Host's IPv6 address doesn't belong to service type."
    else:
        assert any(
            ipv6_addr in oss_ipv6_network for oss_ipv6_network in oss_ipv6_networks
        ), "Host's IPv6 address doesn't belong to service type."


def assert_network_in_service(
    ipv4_network: Optional[V4ServiceNetwork] = None,
    ipv6_network: Optional[V6ServiceNetwork] = None,
    oss_ipv4_containers=None,
    oss_ipv6_containers=None,
    oss_ipv4_networks=None,
    oss_ipv6_networks=None,
):
    # IPv4
    if ipv4_network:
        if oss_ipv4_containers:
            assert any(
                ipv4_network.subnet_of(oss_ipv4_container) for oss_ipv4_container in oss_ipv4_containers
            ), "Network doesn't belong to service type."
        else:
            assert ipv4_network in oss_ipv4_networks, "Network doesn't belong to service type."

    # IPv6
    if ipv6_network:
        if oss_ipv6_containers:
            assert any(
                ipv6_network.subnet_of(oss_ipv6_container) for oss_ipv6_container in oss_ipv6_containers
            ), "Network doesn't belong to service type."
        else:
            assert ipv6_network in oss_ipv6_networks, "Network doesn't belong to service type."


def find_networks(network_container: Optional[str] = "", network: Optional[str] = "", ip_version: int = 4):
    """Get all networks optinally filtering by a container or by a network.

    Args:
    ----
    network_container (str, optional): container to filter by
    network (str, optional): network to filter by
    ip_version (int): 4 or 6

    Returns:
    -------
    (list) all found networks mathing the args, which may be empty.

    """
    assert ip_version in [4, 6]
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX
    endpoint = "network" if ip_version == 4 else "ipv6network"
    params = None
    if network_container:
        params = {"network_container": network_container}
    elif network:
        params = {"network": network, "_return_fields": "comment"}
    r = requests.get(
        f"{wapi(infoblox_params)}/{endpoint}",
        params=params,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
    return r.json()


def allocate_network_inner(
    infoblox_params: settings.InfoBloxParams,
    network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams],
    ip_version: int = 4,
    comment: Optional[str] = "",
    extattrs: Optional[dict] = None,
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
    if extattrs is None:
        extattrs = {}
    assert ip_version in [4, 6]
    endpoint = "network" if ip_version == 4 else "ipv6network"
    ip_container = "networkcontainer" if ip_version == 4 else "ipv6networkcontainer"

    assert network_params.containers, (
        "No containers available to allocate networks for this service."
        "Maybe you want to allocate a host from a network directly?"
    )

    # only return in the response the allocated network, not all available
    # TODO: any validation needed for extrattrs wherever it's used?
    req_payload = {
        "network": {
            "_object_function": "next_available_network",
            "_parameters": {"cidr": network_params.mask},
            "_object": ip_container,
            "_object_parameters": {"network": str(network_params.containers[0])},
            "_result_field": "networks",
        },
        "comment": comment,
        "extattrs": extattrs,
    }

    container_index = 0
    while True:
        r = requests.post(
            f"{wapi(infoblox_params)}/{endpoint}",
            params={"_return_fields": "network"},
            json=req_payload,
            auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
            headers={"content-type": "application/json"},
            verify=False,  # noqa: S501
            timeout=REQUESTS_TIMEOUT,
        )
        if not match_error_code(response=r, error_code=IPAMErrors.CONTAINER_FULL):
            break
        # Container full: try with next valid container for service (if any)
        container_index += 1
        if len(network_params.containers) < (container_index + 1):
            break
        req_payload["network"]["_object_parameters"]["network"] = str(network_params.containers[container_index])

    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    assert "network" in r.json()
    allocated_network = r.json()["network"]
    if ip_version == 4:
        return V4ServiceNetwork(v4=ipaddress.ip_network(allocated_network))
    return V6ServiceNetwork(v6=ipaddress.ip_network(allocated_network))


def allocate_ipv4_network(
    service_type: str = "", comment: Optional[str] = "", extattrs: Optional[dict] = None
) -> V4ServiceNetwork:
    """Allocate IPv4 network within the container of the specified service type.

    Args:
    ----
    service_type (str): the name of the service type (e.g. "TRUNK")
    comment (str, optional): a custom comment to write in the comment field in IPAM
    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})

    Returns:
    -------
    (V4ServiceNetwork): the allocated network

    """
    if extattrs is None:
        extattrs = {}
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM
    assert hasattr(ipam_params, service_type) and service_type != "INFOBLOX", "Invalid service type."
    return allocate_network_inner(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V4, 4, comment, extattrs)


def allocate_ipv6_network(
    service_type: str = "", comment: Optional[str] = "", extattrs: Optional[dict] = None
) -> V6ServiceNetwork:
    """Allocate IPv6 network within the container of the specified service type.

    Args:
    ----
    service_type (str): the name of the service type (e.g. "TRUNK")
    comment (str, optional): a custom comment to write in the comment field in IPAM
    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})

    Returns:
    -------
    (V4ServiceNetwork): the allocated network

    """
    if extattrs is None:
        extattrs = {}
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM
    assert hasattr(ipam_params, service_type) and service_type != "INFOBLOX", "Invalid service type."
    return allocate_network_inner(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V6, 6, comment, extattrs)


def allocate_networks(
    service_type: str = "", comment: Optional[str] = "", extattrs: Optional[dict] = None
) -> ServiceNetworks:
    """Allocate IPv4 and IPv6 network for the specified service type."""
    if extattrs is None:
        extattrs = {}
    v4_service_network = allocate_ipv4_network(service_type=service_type, comment=comment, extattrs=extattrs)
    v6_service_network = allocate_ipv6_network(service_type=service_type, comment=comment, extattrs=extattrs)
    return ServiceNetworks(v4=v4_service_network.v4, v6=v6_service_network.v6)


def find_next_available_ip(infoblox_params, network_ref: str = ""):
    """Find the next available IP address from a network given its ref.

    Args:
    ----
    infoblox_params (settings.InfoBloxParams): infoblox params
    network_ref (str): the network to find the next available ip, in InfoBlox reference format

    Returns:
    -------
    (str): next available ip in the network, or "NETWORK_FULL" if there's no space in the network

    """
    r = requests.post(
        f"{wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1",
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )

    if match_error_code(response=r, error_code=IPAMErrors.NETWORK_FULL):
        return "NETWORK_FULL"

    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
    assert "ips" in r.json()
    received_ip = r.json()["ips"]
    assert len(received_ip) == 1
    return received_ip[0]


def allocate_host_inner(  # noqa: C901
    hostname: str = "",
    addrs: Optional[Tuple] = None,
    networks: Optional[Tuple] = None,
    cname_aliases: Optional[list] = None,
    dns_view: Optional[str] = "default",
    extattrs: Optional[dict] = None,
) -> Union[HostAddresses, str]:
    # TODO: should hostnames be unique
    # (i.e. fail if hostname already exists in this domain/service)?
    if cname_aliases is None:
        cname_aliases = []
    if extattrs is None:
        extattrs = {}
    assert addrs or networks, "Neither networks nor host addresses could be derived to allocate host."
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    # If networks is not None, allocate host in those networks.
    if networks:
        ipv4_network = networks[0]
        ipv6_network = networks[1]
        assert ip_network_version(ipv4_network) == 4
        assert ip_network_version(ipv6_network) == 6

        # Find the next available IP address in each network
        # If requested error doesn't exist, return error
        network_info = find_networks(network=ipv4_network, ip_version=4)
        if len(network_info) != 1:
            return "IPV4_NETWORK_NOT_FOUND"
        assert "_ref" in network_info[0]
        ipv4_addr = find_next_available_ip(infoblox_params, network_info[0]["_ref"])

        network_info = find_networks(network=ipv6_network, ip_version=6)
        if len(network_info) != 1:
            return "IPV6_NETWORK_NOT_FOUND"
        assert "_ref" in network_info[0]
        ipv6_addr = find_next_available_ip(infoblox_params, network_info[0]["_ref"])

        # If couldn't find next available IPs, return error
        if ipv4_addr == "NETWORK_FULL" or ipv6_addr == "NETWORK_FULL":
            if ipv4_addr == "NETWORK_FULL":
                return "IPV4_NETWORK_FULL"
            if ipv6_addr == "NETWORK_FULL":
                return "IPV6_NETWORK_FULL"

    # Otherwise if addrs is not None, allocate host with those addresses.
    else:
        ipv4_addr = addrs[0]
        ipv6_addr = addrs[1]
        assert ip_addr_version(ipv4_addr) == 4
        assert ip_addr_version(ipv6_addr) == 6

    # hostname parameter must be full name including domain name.
    req_payload = {
        "ipv4addrs": [{"ipv4addr": ipv4_addr}],
        "ipv6addrs": [{"ipv6addr": ipv6_addr}],
        "name": hostname,
        "configure_for_dns": True,
        "view": dns_view,
        "extattrs": extattrs,
    }

    r = requests.post(
        f"{wapi(infoblox_params)}/record:host",
        json=req_payload,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
    assert isinstance(r.json(), str)
    assert r.json().startswith("record:host/")

    if cname_aliases:
        cname_req_payload = {"name": "", "canonical": hostname, "view": dns_view, "extattrs": extattrs}

        for alias in cname_aliases:
            cname_req_payload["name"] = alias
            r = requests.post(
                f"{wapi(infoblox_params)}/record:cname",
                json=cname_req_payload,
                auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
                verify=False,  # noqa: S501
                timeout=REQUESTS_TIMEOUT,
            )
            assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
            assert r.json().startswith("record:cname/")

    return HostAddresses(v4=ipaddress.ip_address(ipv4_addr), v6=ipaddress.ip_address(ipv6_addr))


def allocate_host(  # noqa: C901
    hostname: str = "",
    service_type: str = "",
    service_networks: Optional[ServiceNetworks] = None,
    host_addresses: Optional[HostAddresses] = None,
    cname_aliases: Optional[list] = None,
    extattrs: Optional[dict] = None,
) -> HostAddresses:
    """Allocate host record with both IPv4 and IPv6 address, and respective DNS A and AAAA records.

    Args:
    ----
    hostname (str): hostname of the host (without domain name, which is taken from the service type)
    service_type (str): the name of the service type (e.g. "TRUNK")
    service_networks (ServiceNetworks, optional): ipv4 and ipv6 network to allocate host
    host_addresses (HostAddresses, optional): ipv4 and ipv6 addresses to allocate host (service_networks has precedence)
    cname_aliases (list, optional): to create cname records in addition to the host record
    extattrs (dict, optional): any extensible attributes to add in IPAM (e.g. "Site": {"value": "dummy"})

    Returns:
    -------
    (HostAddresses): ipv4 and ipv6 addresses of the allocated host

    """
    if cname_aliases is None:
        cname_aliases = []
    if extattrs is None:
        extattrs = {}
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM

    assert hasattr(ipam_params, service_type) and service_type != "INFOBLOX", "Invalid service type."
    oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
    oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
    oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
    oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
    domain_name = getattr(ipam_params, service_type).domain_name
    dns_view = getattr(ipam_params, service_type).dns_view

    assert (oss_ipv4_containers and oss_ipv6_containers) or (
        oss_ipv4_networks and oss_ipv6_networks
    ), "This service is missing either containers or networks configuration."
    assert domain_name, "This service is missing domain_name configuration."
    assert dns_view, "This service is missing dns_view configuration."

    if cname_aliases:
        cname_aliases = [alias + domain_name for alias in cname_aliases]

    # When neither service_networks not host_addresses are provided:
    # If service has configured containers, new ipv4 and ipv6 networks are created and those are used.
    # Note that in this case extattrs is for the hosts and not for the networks.
    # If service doesn't have configured containers and has configured networks instead, the configured
    # networks are used (they are filled up in order of appearance in the configuration file).
    if not service_networks and not host_addresses:
        if oss_ipv4_containers and oss_ipv6_containers:
            # This service has configured containers.
            # Use them to allocate new networks that can allocate the hosts.

            # IPv4
            ipv4_network = str(allocate_ipv4_network(service_type=service_type).v4)
            assert ipv4_network, "No available space for IPv4 networks for this service type."

            # IPv6
            ipv6_network = str(allocate_ipv6_network(service_type=service_type).v6)
            assert ipv6_network, "No available space for IPv6 networks for this service type."

        elif oss_ipv4_networks and oss_ipv6_networks:
            # This service has configured networks.
            # Allocate a host inside an ipv4 and ipv6 network from among them.
            ipv4_network = str(oss_ipv4_networks[0])
            ipv6_network = str(oss_ipv6_networks[0])

        ipv4_network_index = 0
        ipv6_network_index = 0
        while True:
            network_tuple = (ipv4_network, ipv6_network)
            host = allocate_host_inner(
                hostname=hostname + domain_name,
                networks=network_tuple,
                cname_aliases=cname_aliases,
                dns_view=dns_view,
                extattrs=extattrs,
            )

            if "NETWORK_FULL" not in host and "NETWORK_NOT_FOUND" not in host:
                break
            if "IPV4" in host:
                ipv4_network_index += 1
                assert oss_ipv4_networks, "No available space in any IPv4 network for this service."
                assert ipv4_network_index < len(
                    oss_ipv4_networks
                ), "No available space in any IPv4 network for this service."
                ipv4_network = str(oss_ipv4_networks[ipv4_network_index])
            else:  # "IPV6" in host
                ipv6_network_index += 1
                assert oss_ipv6_networks, "No available space in any IPv6 network for this service."
                assert ipv6_network_index < len(
                    oss_ipv6_networks
                ), "No available space in any IPv6 network for this service."
                ipv6_network = str(oss_ipv6_networks[ipv6_network_index])

    elif service_networks:
        ipv4_network = service_networks.v4
        ipv6_network = service_networks.v6
        assert_network_in_service(
            ipv4_network, ipv6_network, oss_ipv4_containers, oss_ipv6_containers, oss_ipv4_networks, oss_ipv6_networks
        )

        host = allocate_host_inner(
            hostname=hostname + domain_name,
            networks=(str(ipv4_network), str(ipv6_network)),
            cname_aliases=cname_aliases,
            dns_view=dns_view,
            extattrs=extattrs,
        )
        assert "NETWORK_FULL" not in host, "Network is full."
        assert "NETWORK_NOT_FOUND" not in host, "Network does not exist in IPAM. Create it first."

    elif host_addresses:
        ipv4_addr = host_addresses.v4
        ipv6_addr = host_addresses.v6
        assert_host_in_service(
            ipv4_addr, ipv6_addr, oss_ipv4_containers, oss_ipv6_containers, oss_ipv4_networks, oss_ipv6_networks
        )

        host = allocate_host_inner(
            hostname=hostname + domain_name,
            addrs=(str(ipv4_addr), str(ipv6_addr)),
            cname_aliases=cname_aliases,
            dns_view=dns_view,
            extattrs=extattrs,
        )
        assert "NETWORK_FULL" not in host

    return host


def delete_network(
    network: ipaddress.ip_network = None, service_type: str = ""
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
    """Delete IPv4 or IPv6 network by CIDR."""
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM
    assert ipam_params.INFOBLOX
    infoblox_params = ipam_params.INFOBLOX

    assert network, "No network specified to delete."
    assert hasattr(ipam_params, service_type) and service_type != "INFOBLOX", "Invalid service type."

    ip_version = ip_network_version(str(network))

    # Ensure that the network to be deleted is under the service type.
    # Otherwise user is not allowed to delete it
    oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
    oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
    oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
    oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
    ipv4_network = None
    ipv6_network = None
    if ip_version == 4:
        ipv4_network = network
    else:
        ipv6_network = network
    assert_network_in_service(
        ipv4_network, ipv6_network, oss_ipv4_containers, oss_ipv6_containers, oss_ipv4_networks, oss_ipv6_networks
    )

    network_info = find_networks(network=str(network), ip_version=ip_version)
    assert len(network_info) == 1, "Network to delete does not exist in IPAM."
    assert "_ref" in network_info[0], "Network to delete does not exist in IPAM."

    r = requests.delete(
        f'{wapi(infoblox_params)}/{network_info[0]["_ref"]}',
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    # Extract ipv4/ipv6 address from the network reference obtained in the
    # response
    r_text = r.text
    network_address = ipaddress.ip_network(r_text.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
    if ip_version == 4:
        return V4ServiceNetwork(v4=ipaddress.ip_network(network_address))
    return V6ServiceNetwork(v6=ipaddress.ip_network(network_address))


def delete_host(
    hostname: str = "",
    host_addresses: HostAddresses = None,
    cname_aliases: Optional[list] = None,
    service_type: str = "",
) -> HostAddresses:
    """Delete host record and associated CNAME records.

    All arguments passed to this function must match together a host record in
    IPAM, and all CNAME records associated to it must also be passed exactly.
    """
    if cname_aliases is None:
        cname_aliases = []
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM
    assert ipam_params.INFOBLOX
    infoblox_params = ipam_params.INFOBLOX

    assert host_addresses, "No host specified to delete."
    assert hasattr(ipam_params, service_type) and service_type != "INFOBLOX", "Invalid service type."
    oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
    oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
    oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
    oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
    domain_name = getattr(ipam_params, service_type).domain_name
    dns_view = getattr(ipam_params, service_type).dns_view
    ipv4_addr = str(host_addresses.v4)
    ipv6_addr = str(host_addresses.v6)

    assert_host_in_service(
        host_addresses.v4,
        host_addresses.v6,
        oss_ipv4_containers,
        oss_ipv6_containers,
        oss_ipv4_networks,
        oss_ipv6_networks,
    )

    # Find host record reference
    r = requests.get(
        f"{wapi(infoblox_params)}/record:host",
        params={
            "name": (hostname + domain_name).lower(),  # hostnames are lowercase
            "ipv4addr": ipv4_addr,
            "ipv6addr": ipv6_addr,
            "view": dns_view,
        },
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    host_data = r.json()
    assert len(host_data) == 1, "Host to delete does not exist in IPAM."
    assert "_ref" in host_data[0], "Host to delete does not exist in IPAM."
    host_ref = host_data[0]["_ref"]

    # Find cname records reference
    r = requests.get(
        f"{wapi(infoblox_params)}/record:cname",
        params={
            "canonical": hostname + domain_name,
            "view": dns_view,
        },
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    cname_data = r.json()
    provided_cnames = [item + domain_name for item in cname_aliases]
    found_cnames = [item["name"] for item in cname_data if "name" in item]
    assert provided_cnames == found_cnames, "Provided CNAME alias names don't match the ones poiting to hostname."

    # Delete the host record
    r = requests.delete(
        f"{wapi(infoblox_params)}/{host_ref}",
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    # Delete the CNAME records
    cname_refs = [item["_ref"] for item in cname_data if "name" in item]
    for cname_ref in cname_refs:
        r = requests.delete(
            f"{wapi(infoblox_params)}/{cname_ref}",
            auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
            verify=False,  # noqa: S501
            timeout=REQUESTS_TIMEOUT,
        )
        assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    return host_addresses


def validate_network(
    gso_subscription_id: str = "", network: ipaddress.ip_network = None
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
    """Validate IPv4 or IPv6 network.

    Check if the specified network exist, and, if it does, check if its comment field contains gso_subscription_id.
    Returns the network if validation successful.
    """
    assert network, "No network specified to validate."

    ip_version = ip_network_version(str(network))
    network_info = find_networks(network=str(network), ip_version=ip_version)
    assert len(network_info) == 1 and "_ref" in network_info[0], "Network to validate not found in IPAM."
    assert "comment" in network_info[0], "Network to validate does not have comment in IPAM."
    assert (
        gso_subscription_id in network_info[0]["comment"]
    ), "GSO subscription ID does not match the one in the comment field of the IPAM network."

    if ip_version == 4:
        return V4ServiceNetwork(v4=network)
    return V6ServiceNetwork(v6=network)


def validate_host(
    hostname: str = "",
    host_addresses: HostAddresses = None,
    cname_aliases: Optional[list] = None,
    service_type: str = "",
) -> HostAddresses:
    """Validate host.

    Check if all arguments passed to this function match together a host record in
    IPAM, and all CNAME records associated to it also match exactly.
    Returns the host if validation successful.
    """
    if cname_aliases is None:
        cname_aliases = []
    oss = settings.load_oss_params()
    assert oss.IPAM
    ipam_params = oss.IPAM
    assert ipam_params.INFOBLOX
    infoblox_params = ipam_params.INFOBLOX

    assert hostname and host_addresses, "No host specified to validate. Either hostname or host_addresses missing."
    domain_name = getattr(ipam_params, service_type).domain_name
    ipv4_addr = str(host_addresses.v4)
    ipv6_addr = str(host_addresses.v6)
    dns_view = getattr(ipam_params, service_type).dns_view

    # Find host record reference
    r = requests.get(
        f"{wapi(infoblox_params)}/record:host",
        params={
            "name": (hostname + domain_name).lower(),  # hostnames are lowercase
            "ipv4addr": ipv4_addr,
            "ipv6addr": ipv6_addr,
            "view": dns_view,
        },
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    host_data = r.json()
    assert len(host_data) == 1, "Host to validate does not exist in IPAM."
    assert "_ref" in host_data[0], "Host to validate does not exist in IPAM."

    # Find cname records reference
    r = requests.get(
        f"{wapi(infoblox_params)}/record:cname",
        params={
            "canonical": hostname + domain_name,
            "view": dns_view,
        },
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,  # noqa: S501
        timeout=REQUESTS_TIMEOUT,
    )
    cname_data = r.json()
    provided_cnames = [item + domain_name for item in cname_aliases]
    found_cnames = [item["name"] for item in cname_data if "name" in item]
    assert provided_cnames == found_cnames, "Provided CNAME alias names don't match the ones poiting to hostname."

    return host_addresses