Skip to content
Snippets Groups Projects

Feature/ipam integration

Merged Feature/ipam integration
feature/ipam-integration into develop
All threads resolved!
Merged Aleksandr Kurbatov requested to merge feature/ipam-integration into develop
All threads resolved!
Viewing commit e3fab4ac
Next
Show latest version
4 files
+ 726
− 724
Compare changes
  • Side-by-side
  • Inline
Files
4
gso/services/_ipam.py
+ 678
− 681
import ipaddress
import ipaddress
import requests
import requests
from enum import Enum
from enum import Enum
from pydantic import BaseSettings
from pydantic import BaseSettings
from requests.auth import HTTPBasicAuth
from requests.auth import HTTPBasicAuth
from typing import Union
from typing import Union
from gso import settings
from gso import settings
class V4ServiceNetwork(BaseSettings):
class V4ServiceNetwork(BaseSettings):
v4: ipaddress.IPv4Network
v4: ipaddress.IPv4Network
class V6ServiceNetwork(BaseSettings):
class V6ServiceNetwork(BaseSettings):
v6: ipaddress.IPv6Network
v6: ipaddress.IPv6Network
class ServiceNetworks(BaseSettings):
class ServiceNetworks(BaseSettings):
v4: ipaddress.IPv4Network
v4: ipaddress.IPv4Network
v6: ipaddress.IPv6Network
v6: ipaddress.IPv6Network
class V4HostAddress(BaseSettings):
class V4HostAddress(BaseSettings):
v4: ipaddress.IPv4Address
v4: ipaddress.IPv4Address
class V6HostAddress(BaseSettings):
class V6HostAddress(BaseSettings):
v6: ipaddress.IPv6Address
v6: ipaddress.IPv6Address
class HostAddresses(BaseSettings):
class HostAddresses(BaseSettings):
v4: ipaddress.IPv4Address
v4: ipaddress.IPv4Address
v6: ipaddress.IPv6Address
v6: ipaddress.IPv6Address
class IPAMErrors(Enum):
class IPAMErrors(Enum):
# HTTP error code, match in error message
# HTTP error code, match in error message
CONTAINER_FULL = 400, "Can not find requested number of networks"
CONTAINER_FULL = 400, "Can not find requested number of networks"
EXTATTR_UNKNOWN = 400, "Unknown extensible attribute"
NETWORK_FULL = 400, \
EXTATTR_BADVALUE = 400, "Bad value for extensible attribute"
"Cannot find 1 available IP address(es) in this network"
EXTATTR_UNKNOWN = 400, "Unknown extensible attribute"
EXTATTR_BADVALUE = 400, "Bad value for extensible attribute"
# TODO: remove this!
# lab infoblox cert is not valid for the ipv4 address
# ... disable warnings for now
# TODO: remove this!
requests.packages.urllib3.disable_warnings()
# 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 _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 _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)
def _ip_addr_version(addr):
if isinstance(ip_addr, ipaddress.IPv4Address):
ip_version = None
ip_version = 4
ip_addr = ipaddress.ip_address(addr)
elif isinstance(ip_addr, ipaddress.IPv6Address):
if isinstance(ip_addr, ipaddress.IPv4Address):
ip_version = 6
ip_version = 4
assert ip_version in [4, 6]
elif isinstance(ip_addr, ipaddress.IPv6Address):
return ip_version
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)
def _ip_network_version(network):
if isinstance(ip_network, ipaddress.IPv4Network):
ip_version = None
ip_version = 4
ip_network = ipaddress.ip_network(network)
elif isinstance(ip_network, ipaddress.IPv6Network):
if isinstance(ip_network, ipaddress.IPv4Network):
ip_version = 6
ip_version = 4
assert ip_version in [4, 6]
elif isinstance(ip_network, ipaddress.IPv6Network):
return ip_version
ip_version = 6
assert ip_version in [4, 6]
return ip_version
def _find_networks(network_container=None, network=None, ip_version=4):
"""
If network_container is not None, find all networks within the specified
def _find_networks(network_container=None, network=None, ip_version=4):
container.
"""
Otherwise, if network is not None, find the specified network.
If network_container is not None, find all networks within the specified
Otherwise find all networks.
container.
"""
Otherwise, if network is not None, find the specified network.
assert ip_version in [4, 6]
Otherwise find all networks.
oss = settings.load_oss_params()
"""
assert oss.IPAM.INFOBLOX
assert ip_version in [4, 6]
infoblox_params = oss.IPAM.INFOBLOX
oss = settings.load_oss_params()
endpoint = 'network' if ip_version == 4 else 'ipv6network'
assert oss.IPAM.INFOBLOX
params = None
infoblox_params = oss.IPAM.INFOBLOX
if network_container:
endpoint = 'network' if ip_version == 4 else 'ipv6network'
params = {'network_container': network_container}
params = None
elif network:
if network_container:
params = {'network': network}
params = {'network_container': network_container}
r = requests.get(
elif network:
f'{_wapi(infoblox_params)}/{endpoint}',
params = {'network': network}
params=params,
r = requests.get(
auth=HTTPBasicAuth(infoblox_params.username,
f'{_wapi(infoblox_params)}/{endpoint}',
infoblox_params.password),
params=params,
verify=False
auth=HTTPBasicAuth(infoblox_params.username,
)
infoblox_params.password),
# TODO: propagate "network not found" error to caller
verify=False
assert r.status_code >= 200 and r.status_code < 300, \
)
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
# TODO: propagate "network not found" error to caller
return r.json()
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.
def _allocate_network(
"""
infoblox_params: settings.InfoBloxParams,
oss = settings.load_oss_params()
network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams],
assert oss.IPAM.INFOBLOX
ip_version=4,
infoblox_params = oss.IPAM.INFOBLOX
comment="",
extattrs={}
ip_version = _ip_network_version(network)
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
assert ip_version == 4, "Utilization is only available for IPv4 networks."
assert ip_version in [4, 6]
params = {
endpoint = 'network' if ip_version == 4 else 'ipv6network'
'network': network,
ip_container = 'networkcontainer' if ip_version == 4 else \
'_return_fields': 'network,total_hosts,utilization'
'ipv6networkcontainer'
}
assert network_params.containers, \
r = requests.get(
"No containers available to allocate networks for this service." \
f'{_wapi(infoblox_params)}/network',
"Maybe you want to allocate a host from a network directly?"
params=params,
auth=HTTPBasicAuth(infoblox_params.username,
# only return in the response the allocated network, not all available
infoblox_params.password),
# TODO: any validation needed for extrattrs wherever it's used?
verify=False
req_payload = {
)
"network": {
# Utilization info takes several minutes to converge.
"_object_function": "next_available_network",
# The IPAM utilization bar in the GUI as well. Why?
"_parameters": {
assert r.status_code >= 200 and r.status_code < 300, \
"cidr": network_params.mask
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
},
capacity_info = r.json()
"_object": ip_container,
assert len(capacity_info) == 1, "Requested IPv4 network doesn't exist."
"_object_parameters": {
assert 'utilization' in capacity_info[0]
"network": str(network_params.containers[0])
utilization = capacity_info[0]['utilization']
},
return utilization
"_result_field": "networks",
},
"comment": comment,
def _allocate_network(
"extattrs": extattrs
infoblox_params: settings.InfoBloxParams,
}
network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams],
ip_version=4,
container_index = 0
comment="",
while True:
extattrs={}
r = requests.post(
) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
f'{_wapi(infoblox_params)}/{endpoint}',
assert ip_version in [4, 6]
params={'_return_fields': 'network'},
endpoint = 'network' if ip_version == 4 else 'ipv6network'
json=req_payload,
ip_container = 'networkcontainer' if ip_version == 4 else \
auth=HTTPBasicAuth(infoblox_params.username,
'ipv6networkcontainer'
infoblox_params.password),
headers={'content-type': "application/json"},
# only return in the response the allocated network, not all available
verify=False
# TODO: any validation needed for extrattrs wherever it's used?
)
req_payload = {
if not _match_error_code(response=r,
"network": {
error_code=IPAMErrors.CONTAINER_FULL):
"_object_function": "next_available_network",
break
"_parameters": {
# Container full: try with next valid container for service (if any)
"cidr": network_params.mask
container_index += 1
},
if len(network_params.containers) < (container_index + 1):
"_object": ip_container,
break
"_object_parameters": {
req_payload["network"]["_object_parameters"]["network"] = \
"network": str(network_params.containers[0])
str(network_params.containers[container_index])
},
"_result_field": "networks",
assert r.status_code >= 200 and r.status_code < 300, \
},
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
"comment": comment,
"extattrs": extattrs
assert 'network' in r.json()
}
allocated_network = r.json()['network']
if ip_version == 4:
container_index = 0
return V4ServiceNetwork(v4=allocated_network)
while True:
else:
r = requests.post(
return V6ServiceNetwork(v6=allocated_network)
f'{_wapi(infoblox_params)}/{endpoint}',
params={'_return_fields': 'network'},
json=req_payload,
def allocate_service_ipv4_network(service_type, comment="", extattrs={}
auth=HTTPBasicAuth(infoblox_params.username,
) -> V4ServiceNetwork:
infoblox_params.password),
"""
headers={'content-type': "application/json"},
Allocate IPv4 network within the container of the specified service type.
verify=False
"""
)
oss = settings.load_oss_params()
if not _match_error_code(response=r,
assert oss.IPAM
error_code=IPAMErrors.CONTAINER_FULL):
ipam_params = oss.IPAM
break
assert hasattr(ipam_params, service_type) \
# Container full: try with next valid container for service (if any)
and service_type != 'INFOBLOX', "Invalid service type."
container_index += 1
return _allocate_network(ipam_params.INFOBLOX,
if len(network_params.containers) < (container_index + 1):
getattr(ipam_params, service_type).V4,
break
4,
req_payload["network"]["_object_parameters"]["network"] = \
comment,
str(network_params.containers[container_index])
extattrs)
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
def allocate_service_ipv6_network(service_type, comment="", extattrs={}
) -> V6ServiceNetwork:
assert 'network' in r.json()
"""
allocated_network = r.json()['network']
Allocate IPv6 network within the container of the specified service type.
if ip_version == 4:
"""
return V4ServiceNetwork(v4=allocated_network)
oss = settings.load_oss_params()
else:
assert oss.IPAM
return V6ServiceNetwork(v6=allocated_network)
ipam_params = oss.IPAM
assert hasattr(ipam_params, service_type) \
and service_type != 'INFOBLOX', "Invalid service type."
def allocate_service_ipv4_network(service_type, comment="", extattrs={}
return _allocate_network(ipam_params.INFOBLOX,
) -> V4ServiceNetwork:
getattr(ipam_params, service_type).V6,
"""
6,
Allocate IPv4 network within the container of the specified service type.
comment,
"""
extattrs)
oss = settings.load_oss_params()
assert oss.IPAM
ipam_params = oss.IPAM
def _find_next_available_ip(infoblox_params, network_ref):
assert hasattr(ipam_params, service_type) \
"""
and service_type != 'INFOBLOX', "Invalid service type."
Find the next available IP address from a network given its ref.
return _allocate_network(ipam_params.INFOBLOX,
Returns "NETWORK_FULL" if there's no space in the network.
getattr(ipam_params, service_type).V4,
Otherwise returns the next available IP address in the network.
4,
"""
comment,
r = requests.post(
extattrs)
f'{_wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1', # noqa: E501
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params.password),
def allocate_service_ipv6_network(service_type, comment="", extattrs={}
verify=False
) -> V6ServiceNetwork:
)
"""
Allocate IPv6 network within the container of the specified service type.
if _match_error_code(response=r,
"""
error_code=IPAMErrors.NETWORK_FULL):
oss = settings.load_oss_params()
return "NETWORK_FULL"
assert oss.IPAM
ipam_params = oss.IPAM
assert r.status_code >= 200 and r.status_code < 300, \
assert hasattr(ipam_params, service_type) \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
and service_type != 'INFOBLOX', "Invalid service type."
assert 'ips' in r.json()
return _allocate_network(ipam_params.INFOBLOX,
received_ip = r.json()['ips']
getattr(ipam_params, service_type).V6,
assert len(received_ip) == 1
6,
return received_ip[0]
comment,
extattrs)
def _allocate_host(hostname=None,
addrs=None,
def _find_next_available_ip(infoblox_params, network_ref):
networks=None,
r = requests.post(
cname_aliases=None,
f'{_wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1', # noqa: E501
extattrs={}
auth=HTTPBasicAuth(infoblox_params.username,
) -> Union[HostAddresses, str]:
infoblox_params.password),
"""
verify=False
If networks is not None, allocate host in those networks.
)
Otherwise if addrs is not None, allocate host with those addresses.
# TODO: propagate no more available IPs in the network
hostname parameter must be full name including domain name.
assert r.status_code >= 200 and r.status_code < 300, \
Return an error string if couldn't allocate host due to network full.
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
"""
assert 'ips' in r.json()
# TODO: should hostnames be unique
received_ip = r.json()['ips']
# (i.e. fail if hostname already exists in this domain/service)?
assert len(received_ip) == 1
assert addrs or networks, \
return received_ip[0]
"You must specify either the host addresses or the networks CIDR."
oss = settings.load_oss_params()
assert oss.IPAM.INFOBLOX
def _allocate_host(hostname=None, addr=None, network=None, extattrs={}
infoblox_params = oss.IPAM.INFOBLOX
) -> Union[V4HostAddress, V6HostAddress]:
"""
if networks:
If network is not None, allocate host in that network.
ipv4_network = networks[0]
Otherwise if addr is not None, allocate host with that address.
ipv6_network = networks[1]
hostname parameter must be full name including domain name.
assert _ip_network_version(ipv4_network) == 4
"""
assert _ip_network_version(ipv6_network) == 6
# TODO: should hostnames be unique
# (i.e. fail if hostname already exists in this domain/service)?
# Find the next available IP address in each network
assert addr or network, \
network_info = _find_networks(network=ipv4_network, ip_version=4)
"You must specify either the host address or the network CIDR."
assert len(network_info) == 1, \
oss = settings.load_oss_params()
"IPv4 Network does not exist. Create it first."
assert oss.IPAM.INFOBLOX
assert '_ref' in network_info[0]
infoblox_params = oss.IPAM.INFOBLOX
ipv4_addr = _find_next_available_ip(infoblox_params,
network_info[0]["_ref"])
if network:
ip_version = _ip_network_version(network)
network_info = _find_networks(network=ipv6_network, ip_version=6)
# Find the next available IP address in the network
assert len(network_info) == 1, \
network_info = _find_networks(network=network, ip_version=ip_version)
"IPv6 Network does not exist. Create it first."
assert len(network_info) == 1, \
assert '_ref' in network_info[0]
"Network does not exist. Create it first."
ipv6_addr = _find_next_available_ip(infoblox_params,
assert '_ref' in network_info[0]
network_info[0]["_ref"])
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":
else:
if ipv4_addr == "NETWORK_FULL":
ip_version = _ip_addr_version(addr)
return "IPV4_NETWORK_FULL"
if ipv6_addr == "NETWORK_FULL":
ip_req_payload = {
return "IPV6_NETWORK_FULL"
f"ipv{ip_version}addrs": [
{
else:
f"ipv{ip_version}addr": addr
ipv4_addr = addrs[0]
}
ipv6_addr = addrs[1]
],
assert _ip_addr_version(ipv4_addr) == 4
"name": hostname,
assert _ip_addr_version(ipv6_addr) == 6
"configure_for_dns": False,
"view": "default",
req_payload = {
"extattrs": extattrs
"ipv4addrs": [
}
{
"ipv4addr": ipv4_addr
r = requests.post(
}
f'{_wapi(infoblox_params)}/record:host',
],
json=ip_req_payload,
"ipv6addrs": [
auth=HTTPBasicAuth(infoblox_params.username,
{
infoblox_params.password),
"ipv6addr": ipv6_addr
verify=False
}
)
],
assert r.status_code >= 200 and r.status_code < 300, \
"name": hostname,
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
"configure_for_dns": True,
assert isinstance(r.json(), str)
"view": "default",
assert r.json().startswith("record:host/")
"extattrs": extattrs
}
dns_req_payload = {
f"ipv{ip_version}addr": addr,
r = requests.post(
"name": hostname,
f'{_wapi(infoblox_params)}/record:host',
"view": "default",
json=req_payload,
"extattrs": extattrs
auth=HTTPBasicAuth(infoblox_params.username,
}
infoblox_params.password),
verify=False
endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
)
assert r.status_code >= 200 and r.status_code < 300, \
r = requests.post(
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
f'{_wapi(infoblox_params)}/{endpoint}',
assert isinstance(r.json(), str)
json=dns_req_payload,
assert r.json().startswith("record:host/")
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params.password),
if cname_aliases:
verify=False
)
cname_req_payload = {
assert r.status_code >= 200 and r.status_code < 300, \
"name": "",
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
"canonical": hostname,
assert isinstance(r.json(), str)
"view": "default",
assert r.json().startswith(f"{endpoint}/")
"extattrs": extattrs
}
if ip_version == 4:
return V4HostAddress(v4=addr)
for alias in cname_aliases:
else:
cname_req_payload["name"] = alias
return V6HostAddress(v6=addr)
r = requests.post(
f'{_wapi(infoblox_params)}/record:cname',
json=cname_req_payload,
def allocate_service_host(hostname=None,
auth=HTTPBasicAuth(infoblox_params.username,
service_type=None,
infoblox_params.password),
service_networks: ServiceNetworks = None,
verify=False
host_addresses: HostAddresses = None,
)
extattrs={}
assert r.status_code >= 200 and r.status_code < 300, \
) -> HostAddresses:
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
"""
assert r.json().startswith("record:cname/")
Allocate host with both IPv4 and IPv6 address (and respective DNS
records).
return HostAddresses(v4=ipaddress.ip_address(ipv4_addr),
The domain name is also taken from the service type and appended to
v6=ipaddress.ip_address(ipv6_addr))
specified hostname.
If service_networks is provided, that one is used.
If service_networks is not provided, and host_addresses is provided,
def allocate_service_host(hostname=None,
those specific addresses are used.
service_type=None,
If neither is not provided, the first network with available space for
service_networks: ServiceNetworks = None,
this service type is used.
host_addresses: HostAddresses = None,
Note that if WFO will always specify the network/addresses after
cname_aliases=None,
creating it, this mode won't be needed. Currently this mode doesn't
extattrs={}
look further than the first container, so if needed, this will need
) -> HostAddresses:
to be updated.
"""
"""
Allocate host record with both IPv4 and IPv6 address, and respective DNS
oss = settings.load_oss_params()
A and AAAA records.
assert oss.IPAM
- If service_networks is provided, and it's in a valid container,
ipam_params = oss.IPAM
that one is used.
- If service_networks is not provided, and host_addresses is provided,
assert hasattr(ipam_params, service_type) \
those specific addresses are used.
and service_type != 'INFOBLOX', "Invalid service type."
- If neither is provided:
ipv4_containers = getattr(ipam_params, service_type).V4.containers
- If service has configured containers, new ipv4 and ipv6 networks are
ipv6_containers = getattr(ipam_params, service_type).V6.containers
created and those are used. Note that in this case extattrs is for the
domain_name = getattr(ipam_params, service_type).domain_name
hosts and not for the networks.
- If service doesn't have configured containers and has configured
# IPv4
networks instead, the configured networks are used (they are filled up
if not service_networks and not host_addresses:
in order of appearance in the configuration file).
ipv4_networks_info = _find_networks(
The domain name is taken from the service type and appended to the
network_container=str(ipv4_containers[0]), ip_version=4)
specified hostname.
assert len(ipv4_networks_info) >= 1, \
"""
"No IPv4 network exists in the container for this service type."
oss = settings.load_oss_params()
first_nonfull_ipv4_network = None
assert oss.IPAM
for ipv4_network_info in ipv4_networks_info:
ipam_params = oss.IPAM
assert 'network' in ipv4_network_info
capacity = _get_network_capacity(ipv4_network_info["network"])
assert hasattr(ipam_params, service_type) \
if capacity < 1000:
and service_type != 'INFOBLOX', "Invalid service type."
first_nonfull_ipv4_network = ipv4_network_info["network"]
oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
break
oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
# Create a new network if the existing networks in the container for
oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
# the service type are all full.
oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
if not first_nonfull_ipv4_network:
domain_name = getattr(ipam_params, service_type).domain_name
first_nonfull_ipv4_network = str(allocate_service_ipv4_network(
service_type=service_type).v4)
assert (oss_ipv4_containers and oss_ipv6_containers) \
assert first_nonfull_ipv4_network, \
or (oss_ipv4_networks and oss_ipv6_networks), \
"No available IPv4 addresses for this service type."
"This service is missing either containers or networks configuration."
v4_host = _allocate_host(hostname=hostname+domain_name,
assert domain_name, "This service is missing domain_name configuration."
network=first_nonfull_ipv4_network,
extattrs=extattrs)
if cname_aliases:
elif service_networks:
cname_aliases = [alias + domain_name for alias in cname_aliases]
network = service_networks.v4
assert any(network.subnet_of(ipv4_container)
if not service_networks and not host_addresses:
for ipv4_container in ipv4_containers)
if oss_ipv4_containers and oss_ipv6_containers:
v4_host = _allocate_host(hostname=hostname+domain_name,
# This service has configured containers.
network=str(network),
# Use them to allocate new networks that can allocate the hosts.
extattrs=extattrs)
elif host_addresses:
# IPv4
addr = host_addresses.v4
ipv4_network = str(allocate_service_ipv4_network(
assert any(addr in ipv4_container
service_type=service_type).v4)
for ipv4_container in ipv4_containers)
assert ipv4_network, \
v4_host = _allocate_host(hostname=hostname+domain_name,
"No available space for IPv4 networks for this service type."
addr=str(addr),
extattrs=extattrs)
# IPv6
ipv6_network = str(allocate_service_ipv6_network(
# IPv6
service_type=service_type).v6)
if not service_networks and not host_addresses:
assert ipv6_network, \
# ipv6 does not support capacity fetching (not even the GUI displays
"No available space for IPv6 networks for this service type."
# it). Maybe it's assumed that there is always available space?
ipv6_networks_info = _find_networks(
elif oss_ipv4_networks and oss_ipv6_networks:
network_container=str(ipv6_containers[0]),
# This service has configured networks.
ip_version=6)
# Allocate a host inside an ipv4 and ipv6 network from among them.
assert len(ipv6_networks_info) >= 1, \
ipv4_network = str(oss_ipv4_networks[0])
"No IPv6 network exists in the container for this service type."
ipv6_network = str(oss_ipv6_networks[0])
assert 'network' in ipv6_networks_info[0]
# TODO: if "no available IP" error, create a new network?
while True:
v6_host = _allocate_host(hostname=hostname+domain_name,
ipv4_network_index = 0
network=ipv6_networks_info[0]['network'],
ipv6_network_index = 0
extattrs=extattrs)
network_tuple = (ipv4_network, ipv6_network)
elif service_networks:
host = _allocate_host(hostname=hostname+domain_name,
network = service_networks.v6
networks=network_tuple,
assert any(network.subnet_of(ipv6_container)
cname_aliases=cname_aliases,
for ipv6_container in ipv6_containers)
extattrs=extattrs)
v6_host = _allocate_host(hostname=hostname+domain_name,
network=str(network),
if "NETWORK_FULL" not in host:
extattrs=extattrs)
break
elif host_addresses:
elif "IPV4" in host:
addr = host_addresses.v6
ipv4_network_index += 1
assert any(addr in ipv6_container
assert ipv4_network_index < len(oss_ipv4_networks), \
for ipv6_container in ipv6_containers)
"No available space in any IPv4 network for this service."
v6_host = _allocate_host(hostname=hostname+domain_name,
ipv4_network = str(oss_ipv4_networks[ipv4_network_index])
addr=str(addr),
else: # IPV6 in host
extattrs=extattrs)
ipv6_network_index += 1
assert ipv6_network_index < len(oss_ipv6_networks), \
return HostAddresses(v4=v4_host.v4, v6=v6_host.v6)
"No available space in any IPv6 network for this service."
ipv6_network = str(oss_ipv6_networks[ipv6_network_index])
"""
elif service_networks:
Below methods are not used for supported outside calls
# IPv4
"""
ipv4_network = service_networks.v4
if oss_ipv4_containers:
'''
assert any(ipv4_network.subnet_of(oss_ipv4_container)
def _find_containers(network=None, ip_version=4):
for oss_ipv4_container in oss_ipv4_containers)
"""
else:
If network is not None, find that container.
assert ipv4_network in oss_ipv4_networks
Otherwise find all containers.
"""
# IPv6
assert ip_version in [4, 6]
ipv6_network = service_networks.v6
oss = settings.load_oss_params()
if oss_ipv6_containers:
assert oss.IPAM.INFOBLOX
assert any(ipv6_network.subnet_of(oss_ipv6_container)
infoblox_params = oss.IPAM.INFOBLOX
for oss_ipv6_container in oss_ipv6_containers)
endpoint = 'networkcontainer' if ip_version == 4 \
else:
else 'ipv6networkcontainer'
assert ipv6_network in oss_ipv6_networks
r = requests.get(
f'{_wapi(infoblox_params)}/{endpoint}',
host = _allocate_host(
params={'network': network} if network else None,
hostname=hostname+domain_name,
auth=HTTPBasicAuth(infoblox_params.username,
networks=(str(ipv4_network), str(ipv6_network)),
infoblox_params.password),
cname_aliases=cname_aliases,
verify=False
extattrs=extattrs
)
)
assert r.status_code >= 200 and r.status_code < 300, \
assert "NETWORK_FULL" not in host
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
elif host_addresses:
# IPv4
ipv4_addr = host_addresses.v4
def _delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
if oss_ipv4_containers:
"""
assert any(ipv4_addr in oss_ipv4_container
Delete IPv4 or IPv6 network by CIDR.
for oss_ipv4_container in oss_ipv4_containers)
"""
else:
# TODO: should we check that there are no hosts in this network before
assert any(ipv4_addr in oss_ipv4_network
# deleting? Deleting a network deletes the hosts in it, but not the
for oss_ipv4_network in oss_ipv4_networks)
# associated DNS records.
oss = settings.load_oss_params()
# IPv6
assert oss.IPAM.INFOBLOX
ipv6_addr = host_addresses.v6
infoblox_params = oss.IPAM.INFOBLOX
if oss_ipv6_containers:
assert any(ipv6_addr in oss_ipv6_container
ip_version = _ip_network_version(network)
for oss_ipv6_container in oss_ipv6_containers)
else:
network_info = _find_networks(network=network, ip_version=ip_version)
assert any(ipv4_addr in oss_ipv6_network
assert len(network_info) == 1, "Network does not exist."
for oss_ipv6_network in oss_ipv6_networks)
assert '_ref' in network_info[0]
host = _allocate_host(
r = requests.delete(
hostname=hostname+domain_name,
f'{_wapi(infoblox_params)}/{network_info[0]["_ref"]}',
addrs=(str(ipv4_addr), str(ipv6_addr)),
auth=HTTPBasicAuth(infoblox_params.username,
cname_aliases=cname_aliases,
infoblox_params.password),
extattrs=extattrs
verify=False
)
)
assert "NETWORK_FULL" not in host
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return host
# Extract ipv4/ipv6 address from the network reference obtained in the
# response
"""
r_json = r.json()
Below methods are not used for supported outside calls
network_address = ipaddress.ip_network(
"""
r_json.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
if ip_version == 4:
'''
return V4ServiceNetwork(v4=network_address)
def _find_containers(network=None, ip_version=4):
else:
"""
return V6ServiceNetwork(v6=network_address)
If network is not None, find that container.
Otherwise find all containers.
"""
def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
assert ip_version in [4, 6]
"""
oss = settings.load_oss_params()
Delete IPv4 or IPv6 host by its address.
assert oss.IPAM.INFOBLOX
"""
infoblox_params = oss.IPAM.INFOBLOX
oss = settings.load_oss_params()
endpoint = 'networkcontainer' if ip_version == 4 \
assert oss.IPAM.INFOBLOX
else 'ipv6networkcontainer'
infoblox_params = oss.IPAM.INFOBLOX
r = requests.get(
f'{_wapi(infoblox_params)}/{endpoint}',
ip_version = _ip_addr_version(addr)
params={'network': network} if network else None,
ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params.password),
# Find host record reference
verify=False
r = requests.get(
)
f'{_wapi(infoblox_params)}/record:host',
assert r.status_code >= 200 and r.status_code < 300, \
params={ip_param: addr},
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
auth=HTTPBasicAuth(infoblox_params.username,
return r.json()
infoblox_params.password),
verify=False
)
def _get_network_capacity(network=None):
host_data = r.json()
"""
assert len(host_data) == 1, "Host does not exist."
Get utilization of a IPv4 network in a fraction of 1000.
assert '_ref' in host_data[0]
"""
host_ref = host_data[0]['_ref']
oss = settings.load_oss_params()
assert oss.IPAM.INFOBLOX
# Delete it
infoblox_params = oss.IPAM.INFOBLOX
r = requests.delete(
f'{_wapi(infoblox_params)}/{host_ref}',
ip_version = _ip_network_version(network)
auth=HTTPBasicAuth(infoblox_params.username,
assert ip_version == 4, "Utilization is only available for IPv4 networks."
infoblox_params.password),
params = {
verify=False
'network': network,
)
'_return_fields': 'network,total_hosts,utilization'
assert r.status_code >= 200 and r.status_code < 300, \
}
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
r = requests.get(
# Also find and delete the associated dns a/aaaa record
f'{_wapi(infoblox_params)}/network',
endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
params=params,
auth=HTTPBasicAuth(infoblox_params.username,
r = requests.get(
infoblox_params.password),
f'{_wapi(infoblox_params)}/{endpoint}',
verify=False
params={ip_param: addr},
)
auth=HTTPBasicAuth(infoblox_params.username,
# Utilization info takes several minutes to converge.
infoblox_params.password),
# The IPAM utilization bar in the GUI as well. Why?
verify=False
assert r.status_code >= 200 and r.status_code < 300, \
)
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
dns_data = r.json()
capacity_info = r.json()
assert len(dns_data) == 1, "DNS record does not exist."
assert len(capacity_info) == 1, "Requested IPv4 network doesn't exist."
assert '_ref' in dns_data[0]
assert 'utilization' in capacity_info[0]
dns_ref = dns_data[0]['_ref']
utilization = capacity_info[0]['utilization']
return utilization
r = requests.delete(
f'{_wapi(infoblox_params)}/{dns_ref}',
auth=HTTPBasicAuth(infoblox_params.username,
def _delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
infoblox_params.password),
"""
verify=False
Delete IPv4 or IPv6 network by CIDR.
)
"""
assert r.status_code >= 200 and r.status_code < 300, \
# TODO: should we check that there are no hosts in this network before
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
# deleting? Deleting a network deletes the hosts in it, but not the
# associated DNS records.
if ip_version == 4:
oss = settings.load_oss_params()
return V4HostAddress(v4=addr)
assert oss.IPAM.INFOBLOX
else:
infoblox_params = oss.IPAM.INFOBLOX
return V6HostAddress(v6=addr)
ip_version = _ip_network_version(network)
def _get_network_usage_status(network):
network_info = _find_networks(network=network, ip_version=ip_version)
"""
assert len(network_info) == 1, "Network does not exist."
Get status and usage fields of all hosts in the specified ipv4 or ipv6
assert '_ref' in network_info[0]
network.
"""
r = requests.delete(
oss = settings.load_oss_params()
f'{_wapi(infoblox_params)}/{network_info[0]["_ref"]}',
assert oss.IPAM.INFOBLOX
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params = oss.IPAM.INFOBLOX
infoblox_params.password),
verify=False
ip_version = _ip_network_version(network)
)
endpoint = 'ipv4address' if ip_version == 4 else 'ipv6address'
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
r = requests.get(
f'{_wapi(infoblox_params)}/{endpoint}',
# Extract ipv4/ipv6 address from the network reference obtained in the
params={
# response
'_return_fields': 'ip_address,status,usage',
r_json = r.json()
'network': network},
network_address = ipaddress.ip_network(
auth=HTTPBasicAuth(infoblox_params.username,
r_json.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
infoblox_params.password),
if ip_version == 4:
verify=False
return V4ServiceNetwork(v4=network_address)
)
else:
assert r.status_code >= 200 and r.status_code < 300, \
return V6ServiceNetwork(v6=network_address)
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
return r.json()
'''
def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
'''
"""
if __name__ == '__main__':
Delete IPv4 or IPv6 host by its address.
while True:
"""
print("1. Find all containers")
oss = settings.load_oss_params()
print("2. Find all networks")
assert oss.IPAM.INFOBLOX
print("3. Get network capacity")
infoblox_params = oss.IPAM.INFOBLOX
print("4. Create new network")
print("5. Delete network")
ip_version = _ip_addr_version(addr)
print("6. Allocate host by IP")
ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'
print("7. Allocate host by network CIDR")
print("8. Allocate host by service type")
# Find host record reference
print("9. Delete host by IP")
r = requests.get(
print("10. Get network usage status")
f'{_wapi(infoblox_params)}/record:host',
print("11. Exit")
params={ip_param: addr},
auth=HTTPBasicAuth(infoblox_params.username,
choice = input("Enter your choice: ")
infoblox_params.password),
verify=False
if choice == '1':
)
ip_version = int(input("Enter IP version (4 or 6): "))
host_data = r.json()
containers = _find_containers(ip_version=ip_version)
assert len(host_data) == 1, "Host does not exist."
print(json.dumps(containers, indent=2))
assert '_ref' in host_data[0]
host_ref = host_data[0]['_ref']
elif choice == '2':
ip_version = int(input("Enter IP version (4 or 6): "))
# Delete it
networks = _find_networks(ip_version=ip_version)
r = requests.delete(
print(json.dumps(networks, indent=2))
f'{_wapi(infoblox_params)}/{host_ref}',
auth=HTTPBasicAuth(infoblox_params.username,
elif choice == '3':
infoblox_params.password),
network = input("Enter network (in CIDR notation): ")
verify=False
network_capacity = _get_network_capacity(network=network)
)
print(json.dumps(network_capacity, indent=2))
assert r.status_code >= 200 and r.status_code < 300, \
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
elif choice == '4':
service_type = input("Enter service type: ")
# Also find and delete the associated dns a/aaaa record
comment = input("Enter a comment for the network: ")
endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
ip_version = int(input("Enter IP version (4 or 6): "))
if ip_version == 4:
r = requests.get(
new_network = allocate_service_ipv4_network(
f'{_wapi(infoblox_params)}/{endpoint}',
comment=comment, service_type=service_type)
params={ip_param: addr},
elif ip_version == 6:
auth=HTTPBasicAuth(infoblox_params.username,
new_network = allocate_service_ipv6_network(
infoblox_params.password),
comment=comment, service_type=service_type)
verify=False
else:
)
print("Invalid IP version. Please enter either 4 or 6.")
dns_data = r.json()
continue
assert len(dns_data) == 1, "DNS record does not exist."
print(json.dumps(str(new_network), indent=2))
assert '_ref' in dns_data[0]
dns_ref = dns_data[0]['_ref']
elif choice == '5':
network = input("Enter network to delete (in CIDR notation): ")
r = requests.delete(
deleted_network = _delete_network(network=network)
f'{_wapi(infoblox_params)}/{dns_ref}',
print(json.dumps(str(deleted_network), indent=2))
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params.password),
elif choice == '6':
verify=False
hostname = input("Enter host name (full name w/ domain name): ")
)
addr = input("Enter IP address to allocate: ")
assert r.status_code >= 200 and r.status_code < 300, \
alloc_ip = _allocate_host(hostname=hostname, addr=addr)
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
print(json.dumps(str(alloc_ip), indent=2))
if ip_version == 4:
elif choice == '7':
return V4HostAddress(v4=addr)
hostname = input("Enter host name (full name w/ domain name): ")
else:
network = input(
return V6HostAddress(v6=addr)
"Enter existing network to allocate from (CIDR notation): ")
alloc_ip = _allocate_host(hostname=hostname, network=network)
print(json.dumps(str(alloc_ip), indent=2))
def _get_network_usage_status(network):
"""
elif choice == '8':
Get status and usage fields of all hosts in the specified ipv4 or ipv6
hostname = input("Enter host name (w/o domain name): ")
network.
service_type = input("Enter service type: ")
"""
alloc_ip = allocate_service_host(
oss = settings.load_oss_params()
hostname=hostname,
assert oss.IPAM.INFOBLOX
service_type=service_type)
infoblox_params = oss.IPAM.INFOBLOX
print(json.dumps(str(alloc_ip), indent=2))
ip_version = _ip_network_version(network)
elif choice == '9':
endpoint = 'ipv4address' if ip_version == 4 else 'ipv6address'
addr = input("Enter IP address of host to delete: ")
deleted_host = _delete_host_by_ip(addr=addr)
r = requests.get(
print(json.dumps(str(deleted_host), indent=2))
f'{_wapi(infoblox_params)}/{endpoint}',
params={
elif choice == '10':
'_return_fields': 'ip_address,status,usage',
network = input(
'network': network},
"Enter network to get host usage status (CIDR notation): ")
auth=HTTPBasicAuth(infoblox_params.username,
usage_status_info = _get_network_usage_status(network=network)
infoblox_params.password),
print(json.dumps(usage_status_info, indent=2))
verify=False
)
elif choice == '11':
assert r.status_code >= 200 and r.status_code < 300, \
print("Exiting...")
f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
break
return r.json()
'''
else:
print("Invalid choice. Please try again.")
'''