Some IPAM calls for A/AAAA records

gso/services/_ipam.py

@@ -4,6 +4,7 @@ import json
 import ipaddress
 from pydantic import BaseSettings
 from typing import Union
+import random
 
 from gso import settings
 
@@ -96,7 +97,7 @@ def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params:
     assert ip_version in [4, 6]
     endpoint = 'network' if ip_version == 4 else 'ipv6network'
 
-    _ipv4_req_payload = {
+    ipv4_req_payload = {
         "network": {
             "_object_function": "next_available_network",
             "_parameters": {
@@ -110,7 +111,7 @@ def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params:
         }
     }
 
-    _ipv6_req_payload = {
+    ipv6_req_payload = {
         "network": {
             "_object_function": "next_available_network",
             "_parameters": {
@@ -124,12 +125,12 @@ def _allocate_network(infoblox_params: settings.InfoBloxParams, network_params:
         }
     }
 
-    _req_payload = _ipv4_req_payload if ip_version == 4 else _ipv6_req_payload
+    req_payload = ipv4_req_payload if ip_version == 4 else ipv6_req_payload
 
     r = requests.post(
         f'{_wapi(infoblox_params)}/{endpoint}',
         params={'_return_fields': 'network'},
-        json=_req_payload,
+        json=req_payload,
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
         headers={'content-type': "application/json"},
         verify=False)
@@ -145,16 +146,14 @@ def allocate_service_ipv4_network(service_type):
     oss = settings.load_oss_params()
     assert oss.IPAM
     ipam_params = oss.IPAM
-    # sanity: verify the service_type is a proper one
-    assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX'
+    assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX', "Invalid service type."
     return _allocate_network(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V4, 4)
 
 def allocate_service_ipv6_network(service_type):
     oss = settings.load_oss_params()
     assert oss.IPAM
     ipam_params = oss.IPAM
-    # sanity: verify the service_type is a proper one
-    assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX'
+    assert hasattr(ipam_params, service_type) and service_type != 'INFOBLOX', "Invalid service type."
     return _allocate_network(ipam_params.INFOBLOX, getattr(ipam_params, service_type).V6, 6)
 
 def delete_network(network):
@@ -193,7 +192,7 @@ def _find_next_available_ip(infoblox_params, network_ref):
     assert len(received_ip) == 1
     return received_ip[0]
 
-def allocate_host(mac=None, hostname=None, addr=None, network=None):
+def allocate_host(mac=None, hostname=None, addr=None, network=None, service_type=None):
     oss = settings.load_oss_params()
     assert oss.IPAM.INFOBLOX
     infoblox_params = oss.IPAM.INFOBLOX
@@ -209,7 +208,7 @@ def allocate_host(mac=None, hostname=None, addr=None, network=None):
     else:
         ip_version = _ip_addr_version(addr)
 
-    _ipv4_req_payload = {
+    ipv4_req_payload = {
         "ipv4addrs": [
             {
                 "ipv4addr": addr,
@@ -221,7 +220,7 @@ def allocate_host(mac=None, hostname=None, addr=None, network=None):
         "view": "default"
     }
 
-    _ipv6_req_payload = {
+    ipv6_req_payload = {
         "ipv6addrs": [
             {
                 "ipv6addr": addr
@@ -232,17 +231,41 @@ def allocate_host(mac=None, hostname=None, addr=None, network=None):
         "view": "default"
     }
 
-    _req_payload = _ipv4_req_payload if ip_version == 4 else _ipv6_req_payload
+    ip_req_payload = ipv4_req_payload if ip_version == 4 else ipv6_req_payload
 
     r = requests.post(
         f'{_wapi(infoblox_params)}/record:host',
-        json=_req_payload,
+        json=ip_req_payload,
         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}"
     assert isinstance(r.json(), str)
     assert r.json().startswith("record:host/")
 
+    a_record_payload = {
+        "ipv4addr": addr,
+        "name": hostname,
+        "view": "default"
+    }
+
+    aaaa_record_payload = {
+        "ipv6addr": addr,
+        "name": hostname,
+        "view": "default"
+    }
+
+    endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
+    dns_req_payload = a_record_payload if ip_version == 4 else aaaa_record_payload
+
+    r = requests.post(
+        f'{_wapi(infoblox_params)}/{endpoint}',
+        json=dns_req_payload,
+        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}"
+    assert isinstance(r.json(), str)
+    assert r.json().startswith(f"{endpoint}/")
+
     if ip_version == 4:
         return V4HostAddress(v4=addr)
     else:
@@ -272,7 +295,26 @@ def delete_host_by_ip(addr):
         f'{_wapi(infoblox_params)}/{host_ref}',
         auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
         verify=False)
-    r.raise_for_status()
+    assert r.status_code >= 200 and r.status_code < 300, f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
+
+    # Also find and delete the associated dns a/aaaa record
+    endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
+
+    r = requests.get(
+        f'{_wapi(infoblox_params)}/{endpoint}',
+        params={ip_param: addr},
+        auth=HTTPBasicAuth(infoblox_params.username, infoblox_params.password),
+        verify=False)
+    dns_data = r.json()
+    assert len(dns_data) == 1, "DNS record does not exist."
+    assert '_ref' in dns_data[0]
+    dns_ref = dns_data[0]['_ref']
+
+    r = requests.delete(
+        f'{_wapi(infoblox_params)}/{dns_ref}',
+        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}"
 
     if ip_version == 4:
         return V4HostAddress(v4=addr)
@@ -287,8 +329,9 @@ if __name__ == '__main__':
         print("4. Delete network")
         print("5. Allocate host by IP")
         print("6. Allocate host by network CIDR")
-        print("7. Delete host by IP")
-        print("8. Exit")
+        print("7. Allocate host by service type")
+        print("8. Delete host by IP")
+        print("9. Exit")
 
         choice = input("Enter your choice: ")
 
@@ -322,23 +365,32 @@ if __name__ == '__main__':
         elif choice == '5':
             hostname = input("Enter host name: ")
             addr = input("Enter IP address to allocate: ")
-            mac = input("Enter mac address (you can leave empty if IPv6): ")
+            mac = ":".join(["{:02x}".format(random.randint(0x00, 0xff)) for i in range(6)]) #input("Enter MAC address (you can leave empty if IPv6): ")
             alloc_ip = allocate_host(hostname=hostname, addr=addr, mac=mac)
             print(json.dumps(str(alloc_ip), indent=2))
 
         elif choice == '6':
             hostname = input("Enter host name: ")
             network = input("Enter an existing network to allocate from (in CIDR notation): ")
-            mac = input("Enter mac address (you can leave empty if IPv6): ")
+            mac = ":".join(["{:02x}".format(random.randint(0x00, 0xff)) for i in range(6)]) #input("Enter MAC address (you can leave empty if IPv6): ")
             alloc_ip = allocate_host(hostname=hostname, network=network, mac=mac)
             print(json.dumps(str(alloc_ip), indent=2))
 
         elif choice == '7':
+            print("Not implemented.")
+            continue
+            hostname = input("Enter host name: ")
+            service_type = input("Enter service type: ")
+            mac = ":".join(["{:02x}".format(random.randint(0x00, 0xff)) for i in range(6)]) #input("Enter MAC address (you can leave empty if IPv6): ")
+            alloc_ip = allocate_host(hostname=hostname, service_type=service_type, mac=mac)
+            print(json.dumps(str(alloc_ip), indent=2))
+
+        elif choice == '8':
             addr = input("Enter IP address of host to delete: ")
             deleted_host = delete_host_by_ip(addr=addr)
             print(json.dumps(str(deleted_host), indent=2))
 
-        elif choice == '8':
+        elif choice == '9':
             print("Exiting...")
             break
 

gso/services/ipam.py

 import ipaddress
 from pydantic import BaseSettings
+
 from gso import settings
+import _ipam
 
 
-class ServiceNetworks(BaseSettings):
+class V4ServiceNetwork(BaseSettings):
     v4: ipaddress.IPv4Network
+
+
+class V6ServiceNetwork(BaseSettings):
     v6: ipaddress.IPv6Network
 
 
-class HostAddresses(BaseSettings):
+class ServiceNetworks(BaseSettings):
+    v4: V4ServiceNetwork
+    v6: V6ServiceNetwork
+
+
+class V4HostAddress(BaseSettings):
     v4: ipaddress.IPv4Address
-    v6: ipaddress.IPv6Address
 
 
-def new_service_networks(
-        service_params: settings.ServiceNetworkParams) -> ServiceNetworks:
-    oss = settings.load_oss_params()
-    assert oss.IPAM.INFOBLOX
-    # TODO: load from ipam
-    # cf. https://gitlab.geant.org/goat/gap-jenkins/-/blob/development/service-editor/gap_service_editor/ipam.py#L35-66  # noqa: E501
+class V6HostAddress(BaseSettings):
+     v6: ipaddress.IPv6Address
+
+
+class HostAddresses(BaseSettings):
+    v4: V4HostAddress
+    v6: V6HostAddress
+
+
+def new_service_networks(service_type) -> ServiceNetworks:
+    v4_service_network = _ipam.allocate_service_ipv4_network(service_type=service_type)
+    v6_service_network = _ipam.allocate_service_ipv6_network(service_type=service_type)
     return ServiceNetworks(
-        v4=ipaddress.IPv4Network('10.0.0.0/24'),
-        v6=ipaddress.IPv6Network('dead:beef::/120'))
+        v4=v4_service_network,
+        v6=v6_service_network)
 
 
 def new_device_lo_address() -> HostAddresses:
@@ -31,3 +46,7 @@ def new_device_lo_address() -> HostAddresses:
     return HostAddresses(
         v4=ipaddress.IPv4Address('10.10.10.10'),
         v6=ipaddress.IPv6Address('fc00:798:aa:1::10'))
+
+
+if __name__ == '__main__':
+    new_service_networks('TRUNK')
\ No newline at end of file