IPAM network endpoints with ipv6 support

gso/services/_ipam.py

@@ -44,25 +44,28 @@ def _wapi(infoblox_params: settings.InfoBloxParams):
     return (f'https://{infoblox_params.host}'
             f'/wapi/{infoblox_params.wapi_version}')
 
-def find_containers(network=None):
+def find_containers(network=None, ip_version=4):
+    assert ip_version in [4, 6]
     oss = settings.load_oss_params()
     assert oss.IPAM.INFOBLOX
     infoblox_params = oss.IPAM.INFOBLOX
+    endpoint = 'networkcontainer' if ip_version == 4 else 'ipv6networkcontainer'
     r = requests.get(
-        f'{_wapi(infoblox_params)}/networkcontainer',
+        f'{_wapi(infoblox_params)}/{endpoint}',
         params={'network': network} if network else None,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
         verify=False)
     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 find_networks(network=None):
+def find_networks(network=None, ip_version=4):
+    assert ip_version in [4, 6]
     oss = settings.load_oss_params()
     assert oss.IPAM.INFOBLOX
     infoblox_params = oss.IPAM.INFOBLOX
+    endpoint = 'network' if ip_version == 4 else 'ipv6network'
     r = requests.get(
-        f'{_wapi(infoblox_params)}/network',
+        f'{_wapi(infoblox_params)}/{endpoint}',
         params={'network': network} if network else None,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
         verify=False)
@@ -71,8 +74,9 @@ def find_networks(network=None):
 
 def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams], ip_version=4):
     assert ip_version in [4, 6]
+    endpoint = 'network' if ip_version == 4 else 'ipv6network'
 
-    _req_payload = {
+    _ipv4_req_payload = {
         "network": {
             "_object_function": "next_available_network",
             "_parameters": {
@@ -82,12 +86,28 @@ def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params:
             "_object_parameters": {
                 "network": str(network_params.container)
             },
-            "_result_field": "networks",
+            "_result_field": "networks", # only return in the response the allocated network, not all available
+        }
+    }
+
+    _ipv6_req_payload = {
+        "network": {
+            "_object_function": "next_available_network",
+            "_parameters": {
+                "cidr": network_params.mask
+            },
+            "_object": "ipv6networkcontainer",
+            "_object_parameters": {
+                "network": str(network_params.container)
+            },
+            "_result_field": "networks", # only return in the response the allocated network, not all available
         }
     }
 
+    _req_payload = _ipv4_req_payload if ip_version == 4 else _ipv6_req_payload
+
     r = requests.post(
-        f'{_wapi(infoblox_params)}/network',
+        f'{_wapi(infoblox_params)}/{endpoint}',
         params={'_return_fields': 'network'},
         json=_req_payload,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
@@ -121,7 +141,16 @@ def delete_network(network):
     oss = settings.load_oss_params()
     assert oss.IPAM.INFOBLOX
     infoblox_params = oss.IPAM.INFOBLOX
-    network_info = find_networks(network=network)
+
+    ip_version = None
+    ip_network = ipaddress.ip_network(network)
+    if isinstance(ip_network, ipaddress.IPv4Network):
+        ip_version = 4
+    elif isinstance(ip_network, ipaddress.IPv6Network):
+        ip_version = 6
+    assert ip_version in [4, 6]
+
+    network_info = find_networks(network=network, ip_version=ip_version)
     assert len(network_info) == 1, "Network does not exist."
     assert '_ref' in network_info[0]
 
@@ -133,10 +162,10 @@ def delete_network(network):
 
     # Extract ipv4/ipv6 address from the network reference obtained in the response
     r_json = r.json()
-    network_address = ipaddress.ip_network(r_json.rsplit("/", 1)[0].split(":")[1])
-    if isinstance(network_address, ipaddress.IPv4Network):
+    network_address = ipaddress.ip_network(r_json.rsplit("/", 1)[0].split(":")[1].replace("%3A", ":"))
+    if ip_version == 4:
         return V4ServiceNetwork(v4=network_address)
-    elif isinstance(network_address, ipaddress.IPv6Network):
+    else:
         return V6ServiceNetwork(v6=network_address)
 
 def _find_next_available_ip(infoblox_params, network_ref):
@@ -224,12 +253,14 @@ if __name__ == '__main__':
         choice = input("Enter your choice: ")
 
         if choice == '1':
-            containers = find_containers()
+            ip_version = int(input("Enter IP version (4 or 6): "))
+            containers = find_containers(ip_version=ip_version)
             print(json.dumps(containers, indent=2))
 
         elif choice == '2':
-            all_networks = find_networks()
-            print(json.dumps(all_networks, indent=2))
+            ip_version = int(input("Enter IP version (4 or 6): "))
+            networks = find_networks(ip_version=ip_version)
+            print(json.dumps(networks, indent=2))
 
         elif choice == '3':
             service_type = input("Enter service type: ")