Infoblox

The Infoblox service that allocates IPAM resources used in GSO products.

AllocationError

Bases: Exception

Raised when Infoblox failed to allocate a resource.

Source code in gso/services/infoblox.py

class AllocationError(Exception):
    """Raised when Infoblox failed to allocate a resource."""

_setup_connection()

Set up a new connection with an Infoblox instance.

Returns:

Type	Description
tuple[Connector, IPAMParams]	A tuple that has an Infoblox Connector instance, and IPAM parameters.

Source code in gso/services/infoblox.py

def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
    """Set up a new connection with an Infoblox instance.

    Returns:
        A tuple that has an Infoblox ``Connector`` instance, and IPAM parameters.
    """
    oss = load_oss_params().IPAM
    options = {
        "host": oss.INFOBLOX.host,
        "username": oss.INFOBLOX.username,
        "password": oss.INFOBLOX.password,
        "wapi_version": oss.INFOBLOX.wapi_version,
        "ssl_verify": oss.INFOBLOX.scheme == "https",
    }
    return connector.Connector(options), oss

_allocate_network(conn, dns_view, network_view, netmask, containers, comment='')

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 AllocationError.

Parameters:

Name	Type	Description	Default
conn	Connector	An active Infoblox connection.	required
dns_view	str	The Infoblox dns_view in which the network should be allocated.	required
network_view	str	The Infoblox network_view where the network should be allocated.	required
netmask	IPv4Netmask | IPv6Netmask	The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.	required
containers	list[str]	A list of network containers in which the network should be allocated, given in CIDR notation.	required
comment	str | None	Optionally, a comment can be added to the network allocation.	''

Source code in gso/services/infoblox.py

def _allocate_network(  # noqa: PLR0917
    conn: connector.Connector,
    dns_view: str,
    network_view: str,
    netmask: IPv4Netmask | IPv6Netmask,
    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 ``AllocationError``.

    Args:
        conn: An active Infoblox connection.
        dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
        network_view: The Infoblox ``network_view`` where the network should be allocated.
        netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
        containers: A list of network containers in which the network should be allocated, given in CIDR notation.
        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), 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."
        logger.warning(msg)

    msg = f"Cannot allocate anything in {containers}, check whether any IP space is available."
    raise AllocationError(msg)

create_v4_network_by_ip(dns_view, network_view, network, comment='')

Register an IPv4 network at the given location.

Raises:

Type	Description
AllocationError	on failure.

Source code in gso/services/infoblox.py

def create_v4_network_by_ip(
    dns_view: str, network_view: str, network: IPv4NetworkType, comment: str | None = ""
) -> None:
    """Register an IPv4 network at the given location.

    Raises:
        AllocationError: on failure.
    """
    conn, _ = _setup_connection()
    created_net = objects.NetworkV4.create(
        conn, network=network, view=dns_view, network_view=network_view, comment=comment
    )
    if created_net.response != "Infoblox Object created":
        msg = f"Failed to allocate network at {network}. Response from Netbox: {created_net.response}"
        raise AllocationError(msg)
    msg = f"Successfully registered new network at {network}"
    logger.debug(msg)

create_v6_network_by_ip(dns_view, network_view, network, comment='')

Register an IPv6 network at the given location.

Raises:

Type	Description
AllocationError	on failure.

Source code in gso/services/infoblox.py

def create_v6_network_by_ip(
    dns_view: str, network_view: str, network: IPv6NetworkType, comment: str | None = ""
) -> None:
    """Register an IPv6 network at the given location.

    Raises:
        AllocationError: on failure.
    """
    conn, _ = _setup_connection()
    created_net = objects.NetworkV6.create(
        conn, network=network, view=dns_view, network_view=network_view, comment=comment
    )
    if created_net.response != "Infoblox Object created":
        msg = f"Failed to allocate network at {network}. Response from Netbox: {created_net.response}"
        raise AllocationError(msg)
    msg = f"Successfully registered new network at {network}"
    logger.debug(msg)

hostname_available(hostname)

Check whether a hostname is still available in Infoblox.

Check whether Infoblox already has a infoblox_client.objects.HostRecord that matches the given hostname.

Danger

This method only checks within the Infoblox instance, and not the rest of the internet. The hostname could therefore still be taken elsewhere.

Parameters:

Name	Type	Description	Default
hostname	str	The hostname to be checked.	required

Source code in gso/services/infoblox.py

def hostname_available(hostname: str) -> bool:
    """Check whether a hostname is still available **in Infoblox**.

    Check whether Infoblox already has a ``infoblox_client.objects.HostRecord`` that matches the given hostname.

    !!! danger
        This method only checks within the Infoblox instance, and not the rest of the internet. The hostname could
        therefore still be taken elsewhere.

    Args:
        hostname: The hostname to be checked.
    """
    conn, _ = _setup_connection()
    return objects.HostRecord.search(conn, name=hostname) is None

allocate_v4_network(service_type, comment='')

Allocate a new IPv4 network in Infoblox.

Allocate an IPv4 network for a specific service type. The service type should be defined in the OSS parameters of GSO, from which the containers and netmask will be used.

Parameters:

Name	Type	Description	Default
service_type	str	The service type for which the network is allocated.	required
comment	str | None	A comment to be added to the allocated network in Infoblox.	''

Source code in gso/services/infoblox.py

def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddress.IPv4Network:
    """Allocate a new IPv4 network in Infoblox.

    Allocate an IPv4 network for a specific service type. The service type should be defined in the OSS parameters of
    GSO, from which the containers and netmask will be used.

    Args:
        service_type: The service type for which the network is allocated.
        comment: A comment to be added to the allocated network in Infoblox.
    """
    conn, oss = _setup_connection()
    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, network_view, netmask, containers, comment))

allocate_v6_network(service_type, comment='')

Allocate a new IPv6 network in Infoblox.

Allocate an IPv6 network for a specific service type. The service type should be defined in the OSS parameters of GSO, from which the containers and netmask will be used.

Parameters:

Name	Type	Description	Default
service_type	str	The service type for which the network is allocated.	required
comment	str | None	A comment to be added to the allocated network in Infoblox.	''

Source code in gso/services/infoblox.py

def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddress.IPv6Network:
    """Allocate a new IPv6 network in Infoblox.

    Allocate an IPv6 network for a specific service type. The service type should be defined in the OSS parameters of
    GSO, from which the containers and netmask will be used.

    Args:
        service_type: The service type for which the network is allocated.
        comment: A comment to be added to the allocated network in Infoblox.
    """
    conn, oss = _setup_connection()
    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, network_view, netmask, containers, comment))

find_network_by_cidr(ip_network)

Find a network in Infoblox by its CIDR.

Parameters:

Name	Type	Description	Default
ip_network	IPv4Network | IPv6Network	The CIDR that is searched.	required

Source code in gso/services/infoblox.py

def find_network_by_cidr(
    ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network,
) -> objects.Network | None:
    """Find a network in Infoblox by its CIDR.

    Args:
        ip_network: The CIDR that is searched.
    """
    conn, _ = _setup_connection()
    return objects.Network.search(conn, cidr=str(ip_network))

delete_network(ip_network)

Delete a network in Infoblox.

Delete a network that is allocated in Infoblox, by passing the CIDR to be deleted. The CIDR must exactly match an existing entry in Infoblox.

Parameters:

Name	Type	Description	Default
ip_network	IPv4Network | IPv6Network	The network that should get deleted.	required

Source code in gso/services/infoblox.py

def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) -> None:
    """Delete a network in Infoblox.

    Delete a network that is allocated in Infoblox, by passing the CIDR to be deleted. The CIDR must exactly match an
    existing entry in Infoblox.

    Args:
        ip_network: The network that should get deleted.
    """
    network = find_network_by_cidr(ip_network)
    if network:
        network.delete()
    else:
        msg = f"Could not find network {ip_network}, nothing has been deleted."
        logger.warning(msg)

allocate_host(hostname, service_type, cname_aliases, comment)

Allocate a new host record in Infoblox.

Create a new host record in Infoblox, by providing a hostname, and the service type that is associated with this new host. Most likely to be a loopback interface. If the hostname is not available in Infoblox (due to a potential collision) this method raises an AllocationError.

Parameters:

Name	Type	Description	Default
hostname	str	The FQDN of the new host	required
service_type	str	The service type from which IP resources should be used.	required
cname_aliases	list[str]	A list of any CNAME aliases that should be associated with this host. Most often this will be a single loopback address.	required
comment	str	A comment that is added to the host record in Infoblox, should be the subscription_id of the new Router subscription.	required

Source code in gso/services/infoblox.py

def allocate_host(
    hostname: str, service_type: str, cname_aliases: list[str], comment: str
) -> tuple[ipaddress.IPv4Address, ipaddress.IPv6Address]:
    """Allocate a new host record in Infoblox.

    Create a new host record in Infoblox, by providing a hostname, and the service type that is associated with this new
    host. Most likely to be a loopback interface. If the hostname is not available in Infoblox (due to a potential
    collision) this method raises an ``AllocationError``.

    Args:
        hostname: The FQDN of the new host
        service_type: The service type from which IP resources should be used.
        cname_aliases: A list of any CNAME aliases that should be associated with this host. Most often this will be a
            single loopback address.
        comment: A comment that is added to the host record in Infoblox, should be the ``subscription_id`` of the new
            ``Router`` subscription.
    """
    if not hostname_available(hostname):
        msg = f"Cannot allocate new host, FQDN {hostname} already taken."
        raise AllocationError(msg)

    conn, oss = _setup_connection()
    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(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(
                conn,
                ip=ipv6_object,
                name=hostname,
                aliases=cname_aliases,
                comment=comment,
                view=dns_view,
                network_view=network_view,
            )
            created_v6 = ipaddress.IPv6Address(new_host.ipv6addr)
        except InfobloxCannotCreateObject as e:
            msg = f"Cannot find 1 available IP address in network {ipv6_range}."
            logger.warning(msg, exc_info=e)

    if created_v6 is None:
        msg = f"Cannot find 1 available IP address in networks {allocation_networks_v6}."
        raise AllocationError(msg)

    created_v4 = None
    for ipv4_range in allocation_networks_v4:
        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]
        try:
            new_host.update()
            new_host = objects.HostRecord.search(conn, name=hostname)
            created_v4 = ipaddress.IPv4Address(new_host.ipv4addr)
        except InfobloxCannotUpdateObject as e:
            msg = f"Cannot find 1 available IP address in network {ipv4_range}."
            logger.warning(msg, exc_info=e)

    if created_v4 is None:
        msg = f"Cannot find 1 available IP address in networks {allocation_networks_v4}."
        raise AllocationError(msg)

    return created_v4, created_v6

create_host_by_ip(hostname, service_type, comment, *, ipv4_address=None, ipv6_address=None)

Create a new host record with a given IPv4 and IPv6 address.

Parameters:

Name	Type	Description	Default
hostname	str	The FQDN of the new host.	required
service_type	str	The relevant service type, used to deduce the correct dns_view and network_view in Infoblox.	required
comment	str	The comment stored in this Infoblox record, most likely the relevant subscription_id in GSO.	required
ipv4_address	IPv4AddressType | None	The IPv4 address of the new host.	None
ipv6_address	IPv6AddressType | None	The IPv6 address of the new host.	None

Source code in gso/services/infoblox.py

def create_host_by_ip(
    hostname: str,
    service_type: str,
    comment: str,
    *,
    ipv4_address: IPv4AddressType | None = None,
    ipv6_address: IPv6AddressType | None = None,
) -> None:
    """Create a new host record with a given IPv4 and IPv6 address.

    Args:
        hostname: The FQDN of the new host.
        service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
            Infoblox.
        comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in GSO.
        ipv4_address: The IPv4 address of the new host.
        ipv6_address: The IPv6 address of the new host.
    """
    if not hostname_available(hostname):
        msg = f"FQDN '{hostname}' is already in use, allocation aborted."
        raise AllocationError(msg)
    if not (ipv4_address or ipv6_address):
        msg = "At least one IP address must be given to create a record for. Received None."
        raise AllocationError(msg)

    conn, oss = _setup_connection()
    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.
    if ipv6_address:  # We first register the IPv6 record.
        ipv6_object = objects.IP.create(ip=str(ipv6_address), mac=NULL_MAC, configure_for_dhcp=False)
        objects.HostRecord.create(
            conn, ip=ipv6_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
        )
        if ipv4_address:  # If set, we also register the IPv4 record.
            new_host = find_host_by_fqdn(hostname)
            ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
            new_host.ipv4addrs = [ipv4_object]
            new_host.update()
    else:  # IPv6 is not set, it must therefore be IPv4 only.
        ipv4_object = objects.IP.create(ip=str(ipv4_address), mac=NULL_MAC, configure_for_dhcp=False)
        objects.HostRecord.create(
            conn, ip=ipv4_object, name=hostname, comment=comment, view=dns_view, network_view=network_view
        )

find_host_by_ip(ip_addr)

Find a host record in Infoblox by its associated IP address.

Parameters:

Name	Type	Description	Default
ip_addr	IPv4Address | IPv6Address	The IP address of a host that is searched for.	required

Source code in gso/services/infoblox.py

def find_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> objects.HostRecord | None:
    """Find a host record in Infoblox by its associated IP address.

    Args:
        ip_addr: The IP address of a host that is searched for.
    """
    conn, _ = _setup_connection()
    if ip_addr.version == 4:  # noqa: PLR2004, the 4 in IPv4 is well-known and not a "magic value."
        return objects.HostRecord.search(
            conn,
            ipv4addr=ip_addr,
            return_fields=["ipv4addrs", "name", "view", "aliases", "comment"],
        )
    return objects.HostRecordV6.search(
        conn,
        ipv6addr=ip_addr,
        return_fields=["ipv6addrs", "name", "view", "aliases", "comment"],
    )

find_host_by_fqdn(fqdn)

Find a host record by its associated FQDN.

Parameters:

Name	Type	Description	Default
fqdn	str	The FQDN of a host that is searched for.	required

Source code in gso/services/infoblox.py

def find_host_by_fqdn(fqdn: str) -> objects.HostRecord:
    """Find a host record by its associated FQDN.

    Args:
        fqdn: The FQDN of a host that is searched for.
    """
    conn, _ = _setup_connection()
    return objects.HostRecord.search(conn, name=fqdn, return_fields=["ipv4addrs", "name", "view", "aliases", "comment"])

find_v6_host_by_fqdn(fqdn)

Find a host record by its associated FQDN.

This specific method will return the IPv6 variant of a record, if it exists.

Parameters:

Name	Type	Description	Default
fqdn	str	The FQDN of a host that is searched for.	required

Source code in gso/services/infoblox.py

def find_v6_host_by_fqdn(fqdn: str) -> objects.HostRecordV6:
    """Find a host record by its associated FQDN.

    This specific method will return the IPv6 variant of a record, if it exists.

    Args:
        fqdn: The FQDN of a host that is searched for.
    """
    conn, _ = _setup_connection()
    return objects.HostRecordV6.search(
        conn, name=fqdn, return_fields=["ipv6addrs", "name", "view", "aliases", "comment"]
    )

delete_host_by_ip(ip_addr)

Delete a host from Infoblox.

Delete a host record in Infoblox, by providing the IP address that is associated with the record.

Parameters:

Name	Type	Description	Default
ip_addr	IPv4AddressType | IPv6Address	The IP address of the host record that should get deleted.	required

Source code in gso/services/infoblox.py

def delete_host_by_ip(ip_addr: IPv4AddressType | ipaddress.IPv6Address) -> None:
    """Delete a host from Infoblox.

    Delete a host record in Infoblox, by providing the IP address that is associated with the record.

    Args:
        ip_addr: The IP address of the host record that should get deleted.
    """
    host = find_host_by_ip(ip_addr)
    if host:
        host.delete()
    else:
        msg = f"Could not find host at {ip_addr}, nothing has been deleted."
        logger.warning(msg)

delete_host_by_fqdn(fqdn)

Delete a host from Infoblox.

Delete a host record in Infoblox, by providing the FQDN that is associated with the record.

Parameters:

Name	Type	Description	Default
fqdn	str	The FQDN of the host record that should get deleted.	required

Source code in gso/services/infoblox.py

def delete_host_by_fqdn(fqdn: str) -> None:
    """Delete a host from Infoblox.

    Delete a host record in Infoblox, by providing the FQDN that is associated with the record.

    Args:
        fqdn: The FQDN of the host record that should get deleted.
    """
    host = find_host_by_fqdn(fqdn)
    if host:
        host.delete()
    else:
        msg = f"Could not find host at {fqdn}, nothing has been deleted."
        logger.warning(msg)