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 find_networks(network_container: Optional[str] = "", network: Optional[str] = "", ip_version: int = 4):
    """If network_container is not None, find all networks within the specified container.
    Otherwise, if network is not None, find the specified network.
    Otherwise find all networks.
    A list of all found networks is returned (an HTTP 200 code
    may be returned with an empty list).
    """
    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}
    r = requests.get(
        f"{wapi(infoblox_params)}/{endpoint}",
        params=params,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,
        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,
            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."""
    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."""
    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.
    Returns "NETWORK_FULL" if there's no space in the network.
    Otherwise returns the next available IP address in the network.
    """
    r = requests.post(
        f"{wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1",  # noqa: E501
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,
        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(
    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]:
    """If networks is not None, allocate host in those networks.
    Otherwise if addrs is not None, allocate host with those addresses.
    hostname parameter must be full name including domain name.
    Return "IPV4_NETWORK_FULL" or "IPV6_NETWORK_FULL"
    if couldn't allocate host due to requested network being full.
    Return "IPV4_NETWORK_NOT_FOUND" or "IPV6_NETWORK_NOT_FOUND"
    if couldn't allocate host due to requested network not existing.
    """
    # 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, "You must specify either the host addresses or the networks CIDR."
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    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
        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"

    else:
        ipv4_addr = addrs[0]
        ipv6_addr = addrs[1]
        assert ip_addr_version(ipv4_addr) == 4
        assert ip_addr_version(ipv6_addr) == 6

    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,
        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,
                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(
    hostname: str = "",
    service_type: str = "",
    service_networks: Optional[ServiceNetworks] = None,
    host_addresses: Optional[ServiceNetworks] = 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.
    - If service_networks is provided, and it's in a valid container,
    that one is used.
    - If service_networks is not provided, and host_addresses is provided,
    those specific addresses are used.
    - If neither is 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).
    The domain name is taken from the service type and appended to the
    specified hostname.
    """
    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]

    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
        ipv4_network = service_networks.v4
        if oss_ipv4_containers:
            assert any(ipv4_network.subnet_of(oss_ipv4_container) for oss_ipv4_container in oss_ipv4_containers)
        else:
            assert ipv4_network in oss_ipv4_networks

        # IPv6
        ipv6_network = service_networks.v6
        if oss_ipv6_containers:
            assert any(ipv6_network.subnet_of(oss_ipv6_container) for oss_ipv6_container in oss_ipv6_containers)
        else:
            assert ipv6_network in 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. 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: Union[V4ServiceNetwork, V6ServiceNetwork] = 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 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
    if ip_version == 4:
        oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
        oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
        if oss_ipv4_containers:
            assert any(
                network.subnet_of(oss_ipv4_container) for oss_ipv4_container in oss_ipv4_containers
            ), "Can't delete: network doesn't belong to service type."
        else:
            assert network in oss_ipv4_networks, "Can't delete: network doesn't belong to service type."

    else:
        oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
        oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
        if oss_ipv6_containers:
            assert any(
                network.subnet_of(oss_ipv6_container) for oss_ipv6_container in oss_ipv6_containers
            ), "Can't delete: network doesn't belong to service type."
        else:
            assert network in oss_ipv6_networks, "Can't delete: network doesn't belong to service type."

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

    r = requests.delete(
        f'{wapi(infoblox_params)}/{network_info[0]["_ref"]}',
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False,
        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 = ""
) -> Union[V4HostAddress, V6HostAddress]:
    """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 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,
        timeout=REQUESTS_TIMEOUT
    )
    host_data = r.json()
    assert len(host_data) == 1, "Host does not exist."
    assert "_ref" in host_data[0]
    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,
        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,
        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,
            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