diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py index 132387f235b886b7130579d4f2ae8346405832df..c8775d0ead11a51b6c7d73143997fba2ed2bda17 100644 --- a/gso/services/infoblox.py +++ b/gso/services/infoblox.py @@ -18,6 +18,11 @@ class DeletionError(Exception): def _setup_connection() -> tuple[connector.Connector, IPAMParams]: + """Set up a new connection with an Infoblox instance. + + :return: A tuple that contains an Infoblox `Connector` instance, and + :rtype: tuple[{class}`infoblox_client.connector.Connector`, IPAMParams] + """ oss = load_oss_params().IPAM options = { "host": oss.INFOBLOX.host, @@ -36,6 +41,22 @@ def _allocate_network( 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 CIDR notation. + :type containers: list[str] + :param comment: Optionally, a comment can be added to the network allocation. + :type comment: str, optional + """ 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: @@ -48,11 +69,30 @@ def _allocate_network( def hostname_available(hostname: str) -> bool: + """Check whether a hostname is still available **in Infoblox**. + + Check whether Infoblox already contains a {class}`infoblox_client.objects.HostRecord` that matches the given + hostname. Be aware that this method only checks within the Infoblox instance, and not the rest of the internet. + The hostname could therefore still be taken elsewhere. + + :param hostname: The hostname to be checked. + :type hostname: str + """ conn, _ = _setup_connection() return objects.HostRecord.search(conn, name=hostname) is None 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. + + :param service_type: The service type for which the network is allocated. + :type service_type: str + :param comment: A comment to be added to the allocated network in Infoblox. + :type comment: str, optional + """ conn, oss = _setup_connection() netmask = getattr(oss, service_type).V4.mask containers = getattr(oss, service_type).V4.containers @@ -62,6 +102,16 @@ def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddres 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. + + :param service_type: The service type for which the network is allocated. + :type service_type: str + :param comment: A comment to be added to the allocated network in Infoblox. + :type comment: str, optional + """ conn, oss = _setup_connection() netmask = getattr(oss, service_type).V6.mask containers = getattr(oss, service_type).V6.containers @@ -71,6 +121,14 @@ def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddres 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, otherwise this method raises a {class}`DeletionError` + + :param ip_network: The network that should get deleted. + :type ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network + """ conn, _ = _setup_connection() network = objects.Network.search(conn, cidr=str(ip_network)) if network: @@ -82,6 +140,22 @@ def delete_network(ip_network: ipaddress.IPv4Network | ipaddress.IPv6Network) -> def allocate_host( hostname: str, service_type: str, cname_aliases: list[str], comment: str | None = "" ) -> 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 {class}`AllocationError`. + + :param hostname: The FQDN of the new host + :type hostname: str + :param service_type: The service type from which IP resources should be used. + :type service_type: str + :param cname_aliases: A list of any CNAME aliases that should be associated with this host. Most often this will + be a single loopback address. + :type cname_aliases: list[str] + :param comment: Optionally, a comment can be added to the host record in Infoblox. + :type comment: str, optional + """ if not hostname_available(hostname): raise AllocationError(f"Cannot allocate new host, FQDN {hostname} already taken.") @@ -125,6 +199,14 @@ def allocate_host( def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | 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. Raises a + {class}`DeletionError` if no record can be found in Infoblox. + + :param ip_addr: The IP address of the host record that should get deleted. + :type ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address + """ conn, _ = _setup_connection() host = objects.HostRecord.search(conn, ipv4addr=ip_addr) if host: @@ -134,6 +216,14 @@ def delete_host_by_ip(ip_addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> 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. Raises a + {class}`DeletionError` if no record can be found in Infoblox. + + :param fqdn: The FQDN of the host record that should get deleted. + :type fqdn: str + """ conn, _ = _setup_connection() host = objects.HostRecord.search(conn, name=fqdn) host.delete() diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py index 265e98b409c3e3ae632c21ac57c1f485142d4ed9..593ef2d3253ec88d10662d73ed0b45ec2581e5aa 100644 --- a/gso/workflows/router/create_router.py +++ b/gso/workflows/router/create_router.py @@ -1,4 +1,3 @@ -import re from typing import Optional # noinspection PyProtectedMember