_ipam.py

import requests
from requests.auth import HTTPBasicAuth
import json
import ipaddress
from pydantic import BaseSettings
from typing import Union
import random

from gso import settings


class V4ServiceNetwork(BaseSettings):
    v4: ipaddress.IPv4Network


class V6ServiceNetwork(BaseSettings):
    v6: ipaddress.IPv6Network


class ServiceNetworks(BaseSettings):
    v4: V4ServiceNetwork
    v6: V6ServiceNetwork


class V4HostAddress(BaseSettings):
    v4: ipaddress.IPv4Address


class V6HostAddress(BaseSettings):
     v6: ipaddress.IPv6Address


class HostAddresses(BaseSettings):
    v4: V4HostAddress
    v6: V6HostAddress


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


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

def _ip_addr_version(addr):
    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):
    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 find_containers(network=None, ip_version=4):
    """
    If network is not None, find the container that contains specified network.
    Otherwise find all containers.
    """
    assert ip_version in [4, 6]
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX
    endpoint = 'networkcontainer' if ip_version == 4 else 'ipv6networkcontainer'
    r = requests.get(
        f'{_wapi(infoblox_params)}/{endpoint}',
        params={'network': network} if network else None,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    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 find_networks(network_container=None, network=None, ip_version=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.
    """
    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
    )
    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 get_network_capacity(network=None):
    """
    Get utilization of a IPv4 network in a fraction of 1000.
    """
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    ip_version = _ip_network_version(network)
    assert ip_version == 4, "Utilization is only available for IPv4 networks."
    params={'network': network, '_return_fields': 'network,total_hosts,utilization'}

    r = requests.get(
        f'{_wapi(infoblox_params)}/network',
        params=params,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
    capacity_info = r.json()
    assert len(capacity_info) == 1, "Requested IPv4 network doesn't exist."
    assert 'utilization' in capacity_info[0]
    utilization = capacity_info[0]['utilization']
    return utilization


def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams], ip_version=4):
    assert ip_version in [4, 6]
    endpoint = 'network' if ip_version == 4 else 'ipv6network'

    ipv4_req_payload = {
        "network": {
            "_object_function": "next_available_network",
            "_parameters": {
                "cidr": network_params.mask
            },
            "_object": "networkcontainer",
            "_object_parameters": {
                "network": str(network_params.container)
            },
            "_result_field": "networks", # only return in the response the allocated network, not all available
        }
    }

    ipv6_req_payload = {
        "network": {
            "_object_function": "next_available_network",
            "_parameters": {
                "cidr": network_params.mask
            },
            "_object": "ipv6networkcontainer",
            "_object_parameters": {
                "network": str(network_params.container)
            },
            "_result_field": "networks", # only return in the response the allocated network, not all available
        }
    }

    req_payload = ipv4_req_payload if ip_version == 4 else ipv6_req_payload

    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
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    allocated_network = r.json()['network']
    if ip_version==4:
        return V4ServiceNetwork(v4=allocated_network)
    else:
        return V6ServiceNetwork(v6=allocated_network)

def allocate_service_ipv4_network(service_type):
    """
    Allocate IPv4 network within the container of the specified service type.
    """
    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(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V4, 4)

def allocate_service_ipv6_network(service_type):
    """
    Allocate IPv6 network within the container of the specified service type.
    """
    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(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V6, 6)

def delete_network(network):
    """
    Delete IPv4 or IPv6 network by CIDR.
    """
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    ip_version = _ip_network_version(network)

    network_info = find_networks(network=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
    )
    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_json = r.json()
    network_address = ipaddress.ip_network(r_json.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
    if ip_version == 4:
        return V4ServiceNetwork(v4=network_address)
    else:
        return V6ServiceNetwork(v6=network_address)

def _find_next_available_ip(infoblox_params, network_ref):
    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
    )
    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(hostname=None, addr=None, network=None):
    """
    If network is not None, allocate host in that network.
    Otherwise if addr is not None, allocate host with that address.
    """
    assert addr or network, "You must specify either the host address or the network CIDR."
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    if network:
        ip_version = _ip_network_version(network)
        # Find the next available IP address in the network
        network_info = find_networks(network=network, ip_version=ip_version)
        assert len(network_info) == 1, "Network does not exist. Create it first."
        assert '_ref' in network_info[0]
        addr = _find_next_available_ip(infoblox_params, network_info[0]["_ref"])

    else:
        ip_version = _ip_addr_version(addr)

    # TODO: use same request for both IP and DNS records

    ipv4_req_payload = {
        "ipv4addrs": [
            {
                "ipv4addr": addr
            }
        ],
        "name": hostname,
        "configure_for_dns": False,
        "view": "default"
    }

    ipv6_req_payload = {
        "ipv6addrs": [
            {
                "ipv6addr": addr
            }
        ],
        "name": hostname,
        "configure_for_dns": False,
        "view": "default"
    }

    ip_req_payload = ipv4_req_payload if ip_version == 4 else ipv6_req_payload

    r = requests.post(
        f'{_wapi(infoblox_params)}/record:host',
        json=ip_req_payload,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    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/")

    a_record_payload = {
        "ipv4addr": addr,
        "name": hostname,
        "view": "default"
    }

    aaaa_record_payload = {
        "ipv6addr": addr,
        "name": hostname,
        "view": "default"
    }

    endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
    dns_req_payload = a_record_payload if ip_version == 4 else aaaa_record_payload

    r = requests.post(
        f'{_wapi(infoblox_params)}/{endpoint}',
        json=dns_req_payload,
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    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(f"{endpoint}/")

    if ip_version == 4:
        return V4HostAddress(v4=addr)
    else:
        return V6HostAddress(v6=addr)

def allocate_host_by_service_type(hostname=None, service_type=None):
    """
    Allocate host with both IPv4 and IPv6 address (and respective DNS records).
    The first network with available space for this service type is used.
    The domain name is also taken from the service type and appended to specified hostname.
    """
    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."
    ipv4_container = getattr(ipam_params, service_type).V4.container
    ipv6_container = getattr(ipam_params, service_type).V6.container
    domain_name = getattr(ipam_params, service_type).domain_name

    ipv4_networks_info = find_networks(network_container=str(ipv4_container), ip_version=4)
    assert len(ipv4_networks_info) >= 1, "No IPv4 network exists in the container for this service type."
    first_nonfull_ipv4_network = None
    for ipv4_network_info in ipv4_networks_info:
        assert 'network' in ipv4_network_info
        capacity = get_network_capacity(ipv4_network_info["network"])
        if capacity < 1000:
            first_nonfull_ipv4_network = ipv4_network_info["network"]
            break
    # TODO: create a new network if available space in the container.
    assert first_nonfull_ipv4_network, "No available IPv4 addresses for this service type."
    v4_host = allocate_host(hostname=hostname+domain_name, network=first_nonfull_ipv4_network)

    # ipv6 does not support capacity fetching (not even the GUI displays it).
    # Maybe it's assumed that there is always available space?
    ipv6_networks_info = find_networks(network_container=str(ipv6_container), ip_version=6)
    assert len(ipv6_networks_info) >= 1, "No IPv6 network exists in the container for this service type."
    assert 'network' in ipv6_networks_info[0]
    v6_host = allocate_host(hostname=hostname+domain_name, network=ipv6_networks_info[0]['network'])

    return HostAddresses(v4=v4_host,v6=v6_host)

def delete_host_by_ip(addr):
    """
    Delete IPv4 or IPv6 host by its address.
    """
    oss = settings.load_oss_params()
    assert oss.IPAM.INFOBLOX
    infoblox_params = oss.IPAM.INFOBLOX

    ip_version = _ip_addr_version(addr)
    ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'

    # Find host record reference
    r = requests.get(
        f'{_wapi(infoblox_params)}/record:host',
        params={ip_param: addr},
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    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']

    # Delete it
    r = requests.delete(
        f'{_wapi(infoblox_params)}/{host_ref}',
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    # Also find and delete the associated dns a/aaaa record
    endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'

    r = requests.get(
        f'{_wapi(infoblox_params)}/{endpoint}',
        params={ip_param: addr},
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    dns_data = r.json()
    assert len(dns_data) == 1, "DNS record does not exist."
    assert '_ref' in dns_data[0]
    dns_ref = dns_data[0]['_ref']

    r = requests.delete(
        f'{_wapi(infoblox_params)}/{dns_ref}',
        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
        verify=False
    )
    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"

    if ip_version == 4:
        return V4HostAddress(v4=addr)
    else:
        return V6HostAddress(v6=addr)

if __name__ == '__main__':
    while True:
        print("1. Find all containers")
        print("2. Find all networks")
        print("3. Get network capacity")
        print("4. Create new network")
        print("5. Delete network")
        print("6. Allocate host by IP")
        print("7. Allocate host by network CIDR")
        print("8. Allocate host by service type")
        print("9. Delete host by IP")
        print("10. Exit")

        choice = input("Enter your choice: ")

        if choice == '1':
            ip_version = int(input("Enter IP version (4 or 6): "))
            containers = find_containers(ip_version=ip_version)
            print(json.dumps(containers, indent=2))

        elif choice == '2':
            ip_version = int(input("Enter IP version (4 or 6): "))
            networks = find_networks(ip_version=ip_version) #, network_container="10.255.255.0/24"
            print(json.dumps(networks, indent=2))

        elif choice == '3':
            network = input("Enter network (in CIDR notation): ")
            network_capacity = get_network_capacity(network=network)
            print(json.dumps(network_capacity, indent=2))

        elif choice == '4':
            service_type = input("Enter service type: ")
            ip_version = int(input("Enter IP version (4 or 6): "))
            if ip_version == 4:
                new_network = allocate_service_ipv4_network(service_type=service_type)
            elif ip_version == 6:
                new_network = allocate_service_ipv6_network(service_type=service_type)
            else:
                print("Invalid IP version. Please enter either 4 or 6.")
                continue
            print(json.dumps(str(new_network), indent=2))

        elif choice == '5':
            network = input("Enter network to delete (in CIDR notation): ")
            deleted_network = delete_network(network=network)
            print(json.dumps(str(deleted_network), indent=2))

        elif choice == '6':
            hostname = input("Enter host name: ")
            addr = input("Enter IP address to allocate: ")
            alloc_ip = allocate_host(hostname=hostname, addr=addr)
            print(json.dumps(str(alloc_ip), indent=2))

        elif choice == '7':
            hostname = input("Enter host name: ")
            network = input("Enter an existing network to allocate from (in CIDR notation): ")
            alloc_ip = allocate_host(hostname=hostname, network=network)
            print(json.dumps(str(alloc_ip), indent=2))

        elif choice == '8':
            hostname = input("Enter host name: ")
            service_type = input("Enter service type: ")
            alloc_ip = allocate_host_by_service_type(hostname=hostname, service_type=service_type)
            print(json.dumps(str(alloc_ip), indent=2))

        elif choice == '9':
            addr = input("Enter IP address of host to delete: ")
            deleted_host = delete_host_by_ip(addr=addr)
            print(json.dumps(str(deleted_host), indent=2))

        elif choice == '10':
            print("Exiting...")
            break

        else:
            print("Invalid choice. Please try again.")