ipam.py

import ipaddress
from pydantic import BaseSettings
from typing import Union

from gso.services import _ipam


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


def new_service_networks(service_type,
                         comment="",
                         extattrs={}) -> ServiceNetworks:
    v4_service_network = _ipam.allocate_service_ipv4_network(
        service_type=service_type, comment=comment, extattrs=extattrs)
    v6_service_network = _ipam.allocate_service_ipv6_network(
        service_type=service_type, comment=comment, extattrs=extattrs)
    return ServiceNetworks(
        v4=v4_service_network.v4,
        v6=v6_service_network.v6)


def new_service_host(hostname,
                     service_type,
                     service_networks: ServiceNetworks = None,
                     host_addresses: HostAddresses = None,
                     cname_aliases=None,
                     extattrs={}) -> HostAddresses:
    return _ipam.allocate_service_host(
        hostname=hostname,
        service_type=service_type,
        service_networks=service_networks,
        host_addresses=host_addresses,
        cname_aliases=cname_aliases,
        extattrs=extattrs)


def delete_service_network(network, service_type
                           ) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
    return _ipam.delete_service_network(
        network=network,
        service_type=service_type
    )


if __name__ == '__main__':
    # sample call flow to allocate two loopback interfaces and a trunk service
    # new_service_host can be called passing networks, addresses, or nothing.
    # - host h1 for service TRUNK uses a specific ipv4/ipv6 address pair
    # - host h2 for service TRUNK uses the ipv4/ipv6 network pair
    # - service LO uses nothing
    # networks and hosts can be allocated with extensible attributes
    # networks can be created with a comment
    # CNAME records can be optionally created

    hostname_A = 'hA'
    hostname_B = 'hB'

    # hA LO (loopback)
    # loA_v4_host_address = ipaddress.ip_address('10.255.255.0')
    # loA_v6_host_address = ipaddress.ip_address('dead:beef::0')
    # loA_host_addresses = HostAddresses(v4=loA_v4_host_address,
    #                                    v6=loA_v6_host_address)
    # new_service_host(hostname=hostname_A+"_LO",
    #                  host_addresses=loA_host_addresses,
    #                  cname_aliases=["alias1.hA", "alias2.hA"],
    #                  service_type='LO')
    new_service_host(hostname=hostname_A+"_LO",
                     cname_aliases=["alias1.hA", "alias2.hA"],
                     service_type='LO')

    # hB LO (loopback)
    new_service_host(hostname=hostname_B+"_LO",
                     cname_aliases=["alias1.hB"],
                     service_type='LO')

    # hA-hB TRUNK
    trunkAB_network_extattrs = {
        "vrf_name": {"value": "dummy_vrf"},
    }
    trunkAB_host_extattrs = {
        "Site": {"value": "dummy_site"},
    }
    trunkAB_service_networks = new_service_networks(
        service_type='TRUNK',
        extattrs=trunkAB_network_extattrs,
        comment="Network for hA-hB TRUNK"
    )

    trunkAB_v4_host_address = trunkAB_service_networks.v4.network_address
    trunkAB_v6_host_address = trunkAB_service_networks.v6.network_address
    trunkAB_host_addresses = HostAddresses(v4=trunkAB_v4_host_address,
                                           v6=trunkAB_v6_host_address)

    new_service_host(hostname=hostname_A+"_TRUNK",
                     service_type='TRUNK',
                     host_addresses=trunkAB_host_addresses,
                     extattrs=trunkAB_host_extattrs)

    new_service_host(hostname=hostname_B+"_TRUNK",
                     service_type='TRUNK',
                     service_networks=trunkAB_service_networks,
                     extattrs=trunkAB_host_extattrs)