ipam.py

import ipaddress
from pydantic import BaseSettings

from gso import settings
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) -> ServiceNetworks:
    v4_service_network = _ipam.allocate_service_ipv4_network(service_type=service_type)
    v6_service_network = _ipam.allocate_service_ipv6_network(service_type=service_type)
    return ServiceNetworks(
        v4=v4_service_network.v4,
        v6=v6_service_network.v6)


def new_service_host(hostname, service_type, service_networks: ServiceNetworks) -> HostAddresses:
    return _ipam.allocate_service_host(hostname=hostname, service_type=service_type, service_networks=service_networks)


if __name__ == '__main__':
    # sample call flow to allocate two loopback interfaces and a trunk service
    # new_service_host can be called passing networks or addresses
    # - the first time is called by passing a specific ipv4/ipv6 address pair
    # - the rest are called by passing the ipv4/ipv6 network pair

    hostname_A = 'newhost1'
    hostname_B = 'newhost2'
    # h1 loopback
    lo1_service_networks = new_service_networks('LO')
    lo1_v4_host_address = lo1_service_networks.v4.network_address
    lo1_v6_host_address = lo1_service_networks.v6.network_address
    lo1_host_addresses = HostAddresses(v4=lo1_v4_host_address, v6=lo1_v6_host_address)
    new_service_host(hostname_A, 'LO', lo1_service_networks)
    # h2 loopback
    lo2_service_networks = new_service_networks('LO')
    new_service_host(hostname_B, 'LO', lo2_service_networks)
    # h1-h2 trunk
    trunk12_service_networks = new_service_networks('TRUNK')
    new_service_host(hostname_A, 'TRUNK', trunk12_service_networks)
    new_service_host(hostname_B, 'TRUNK', trunk12_service_networks)