Skip to content
Snippets Groups Projects

Add network_view to infoblox client

Merged Karel van Klink requested to merge feature/fix-infoblox-service into develop
4 files
+ 46
24
Compare changes
  • Side-by-side
  • Inline
Files
4
+ 30
19
@@ -41,29 +41,32 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
def _allocate_network(
conn: connector.Connector, dns_view: str, netmask: int, containers: list[str], comment: str | None = ""
conn: connector.Connector,
dns_view: str,
network_view: str,
netmask: int,
containers: list[str],
comment: str | None = "",
) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
"""Allocate a new network in Infoblox.
The function will go over all given containers, and try to allocate a network within the available IP space. If no
space is available, this method raises an :class:`AllocationError`.
:param conn: An active Infoblox connection.
:type conn: :class:`infoblox_client.connector.Connector`
:param dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
:type dns_view: str
:param netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
:type netmask: int
:param containers: A list of network containers in which the network should be allocated, given in :term:`CIDR`
notation.
:type containers: list[str]
:param comment: Optionally, a comment can be added to the network allocation.
:type comment: str, optional
:param :class:`infoblox_client.connector.Connector` conn: An active Infoblox connection.
:param str dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
:param str network_view: The Infoblox ``network_view`` where the network should be allocated.
:param int netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
:param list [str] containers: A list of network containers in which the network should be allocated, given in
:term:`CIDR` notation.
:param str comment: Optionally, a comment can be added to the network allocation.
"""
for container in [ipaddress.ip_network(con) for con in containers]:
for network in container.subnets(new_prefix=netmask):
if objects.Network.search(conn, network=str(network)) is None:
created_net = objects.Network.create(conn, network=str(network), dns_view=dns_view, comment=comment)
created_net = objects.Network.create(
conn, network=str(network), dns_view=dns_view, network_view=network_view, comment=comment
)
if created_net.response != "Infoblox Object already Exists":
return ipaddress.ip_network(created_net.network)
msg = f"IP container {container} appears to be full."
@@ -104,8 +107,9 @@ def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddres
netmask = getattr(oss, service_type).V4.mask
containers = getattr(oss, service_type).V4.containers
dns_view = getattr(oss, service_type).dns_view
network_view = getattr(oss, service_type).network_view
return ipaddress.IPv4Network(_allocate_network(conn, dns_view, netmask, containers, comment))
return ipaddress.IPv4Network(_allocate_network(conn, dns_view, network_view, netmask, containers, comment))
def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddress.IPv6Network:
@@ -123,8 +127,9 @@ def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddres
netmask = getattr(oss, service_type).V6.mask
containers = getattr(oss, service_type).V6.containers
dns_view = getattr(oss, service_type).dns_view
network_view = getattr(oss, service_type).network_view
return ipaddress.IPv6Network(_allocate_network(conn, dns_view, netmask, containers, comment))
return ipaddress.IPv6Network(_allocate_network(conn, dns_view, network_view, netmask, containers, comment))
def find_network_by_cidr(
@@ -184,10 +189,11 @@ def allocate_host(
allocation_networks_v4 = getattr(oss, service_type).V4.networks
allocation_networks_v6 = getattr(oss, service_type).V6.networks
dns_view = getattr(oss, service_type).dns_view
network_view = getattr(oss, service_type).network_view
created_v6 = None
for ipv6_range in allocation_networks_v6:
v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv6_range))
v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(network_view, str(ipv6_range))
ipv6_object = objects.IP.create(ip=v6_alloc, mac=NULL_MAC, configure_for_dhcp=False)
try:
new_host = objects.HostRecord.create(
@@ -197,6 +203,7 @@ def allocate_host(
aliases=cname_aliases,
comment=comment,
dns_view=dns_view,
network_view=network_view,
)
created_v6 = ipaddress.IPv6Address(new_host.ipv6addr)
except InfobloxCannotCreateObject:
@@ -209,7 +216,7 @@ def allocate_host(
created_v4 = None
for ipv4_range in allocation_networks_v4:
v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv4_range))
v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(network_view, str(ipv4_range))
ipv4_object = objects.IP.create(ip=v4_alloc, mac=NULL_MAC, configure_for_dhcp=False)
new_host = objects.HostRecord.search(conn, name=hostname)
new_host.ipv4addrs = [ipv4_object]
@@ -240,7 +247,8 @@ def create_host_by_ip(
:param str hostname: The :term:`FQDN` of the new host.
:param IPv4Address ipv4_address: The IPv4 address of the new host.
:param IPv6Address ipv6_address: The IPv6 address of the new host.
:param str service_type: The relevant service type, used to deduce the correct ``dns_view`` in Infoblox.
:param str service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
Infoblox.
:param str comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in
:term:`GSO`.
"""
@@ -252,9 +260,12 @@ def create_host_by_ip(
ipv6_object = objects.IP.create(ip=ipv6_address, mac=NULL_MAC, configure_for_dhcp=False)
ipv4_object = objects.IP.create(ip=ipv4_address, mac=NULL_MAC, configure_for_dhcp=False)
dns_view = getattr(oss, service_type).dns_view
network_view = getattr(oss, service_type).network_view
# This needs to be done in two steps, otherwise only one of the IP addresses is stored.
objects.HostRecord.create(conn, ip=ipv6_object, name=hostname, comment=comment, dns_view=dns_view)
objects.HostRecord.create(
conn, ip=ipv6_object, name=hostname, comment=comment, dns_view=dns_view, network_view=network_view
)
new_host = find_host_by_fqdn(hostname)
new_host.ipv4addrs = [ipv4_object]
new_host.update()
Loading