diff --git a/gso/oss-params-example.json b/gso/oss-params-example.json
index a13415dd26ddb0a58ad7eb65abcfd191c2e40de2..afc41967d1dbe3eb333eca86567f4546db71dcdd 100644
--- a/gso/oss-params-example.json
+++ b/gso/oss-params-example.json
@@ -11,13 +11,20 @@
"username": "robot-user",
"password": "robot-user-password"
},
+ "LO": {
+ "V4": {"containers": ["1.1.0.0/24"], "mask": 32},
+ "V6": {"containers": ["dead:beef::/64"], "mask": 128},
+ "domain_name": ".lo"
+ },
"TRUNK": {
- "V4": {"container": "1.1.0.0/24", "mask": 31},
- "V6": {"container": "dead:beef::/64", "mask": 96}
+ "V4": {"containers": ["1.1.1.0/24"], "mask": 31},
+ "V6": {"containers": ["dead:beef::/64"], "mask": 126},
+ "domain_name": ".trunk"
},
"GEANT_IP": {
- "V4": {"container": "1.1.8.0/24", "mask": 31},
- "V6": {"container": "dead:beef::/64", "mask": 96}
+ "V4": {"containers": ["1.1.2.0/24"], "mask": 31},
+ "V6": {"containers": ["dead:beef::/64"], "mask": 126},
+ "domain_name": ".geantip"
}
},
"PROVISIONING_PROXY": {
diff --git a/gso/services/_ipam.py b/gso/services/_ipam.py
new file mode 100644
index 0000000000000000000000000000000000000000..2707a53ddc0ef4c2111baca5b8ac640f80a9b614
--- /dev/null
+++ b/gso/services/_ipam.py
@@ -0,0 +1,681 @@
+import ipaddress
+import requests
+from enum import Enum
+from pydantic import BaseSettings
+from requests.auth import HTTPBasicAuth
+from typing import Union
+
+from gso import settings
+
+
+class V4ServiceNetwork(BaseSettings):
+ v4: ipaddress.IPv4Network
+
+
+class V6ServiceNetwork(BaseSettings):
+ v6: ipaddress.IPv6Network
+
+
+class ServiceNetworks(BaseSettings):
+ v4: ipaddress.IPv4Network
+ v6: ipaddress.IPv6Network
+
+
+class V4HostAddress(BaseSettings):
+ v4: ipaddress.IPv4Address
+
+
+class V6HostAddress(BaseSettings):
+ v6: ipaddress.IPv6Address
+
+
+class HostAddresses(BaseSettings):
+ v4: ipaddress.IPv4Address
+ v6: ipaddress.IPv6Address
+
+
+class IPAMErrors(Enum):
+ # HTTP error code, match in error message
+ CONTAINER_FULL = 400, "Can not find requested number of networks"
+ EXTATTR_UNKNOWN = 400, "Unknown extensible attribute"
+ EXTATTR_BADVALUE = 400, "Bad value for extensible attribute"
+
+
+# TODO: remove this!
+# lab infoblox cert is not valid for the ipv4 address
+# ... disable warnings for now
+requests.packages.urllib3.disable_warnings()
+
+
+def _match_error_code(response, error_code):
+ return response.status_code == error_code.value[0] \
+ and error_code.value[1] in response.text
+
+
+def _wapi(infoblox_params: settings.InfoBloxParams):
+ return (f'https://{infoblox_params.host}'
+ f'/wapi/{infoblox_params.wapi_version}')
+
+
+def _ip_addr_version(addr):
+ ip_version = None
+ ip_addr = ipaddress.ip_address(addr)
+ if isinstance(ip_addr, ipaddress.IPv4Address):
+ ip_version = 4
+ elif isinstance(ip_addr, ipaddress.IPv6Address):
+ ip_version = 6
+ assert ip_version in [4, 6]
+ return ip_version
+
+
+def _ip_network_version(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]
+ return ip_version
+
+
+def _find_networks(network_container=None, network=None, ip_version=4):
+ """
+ If network_container is not None, find all networks within the specified
+ container.
+ Otherwise, if network is not None, find the specified network.
+ Otherwise find all networks.
+ """
+ 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'
+ params = None
+ if network_container:
+ params = {'network_container': network_container}
+ elif network:
+ params = {'network': network}
+ r = requests.get(
+ f'{_wapi(infoblox_params)}/{endpoint}',
+ params=params,
+ auth=HTTPBasicAuth(infoblox_params.username,
+ infoblox_params.password),
+ verify=False
+ )
+ # TODO: propagate "network not found" error to caller
+ 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 _get_network_capacity(network=None):
+ """
+ Get utilization of a IPv4 network in a fraction of 1000.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM.INFOBLOX
+ infoblox_params = oss.IPAM.INFOBLOX
+
+ ip_version = _ip_network_version(network)
+ assert ip_version == 4, "Utilization is only available for IPv4 networks."
+ params = {
+ 'network': network,
+ '_return_fields': 'network,total_hosts,utilization'
+ }
+
+ r = requests.get(
+ f'{_wapi(infoblox_params)}/network',
+ params=params,
+ auth=HTTPBasicAuth(infoblox_params.username,
+ infoblox_params.password),
+ verify=False
+ )
+ # Utilization info takes several minutes to converge.
+ # The IPAM utilization bar in the GUI as well. Why?
+ assert r.status_code >= 200 and r.status_code < 300, \
+ f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
+ capacity_info = r.json()
+ assert len(capacity_info) == 1, "Requested IPv4 network doesn't exist."
+ assert 'utilization' in capacity_info[0]
+ utilization = capacity_info[0]['utilization']
+ return utilization
+
+
+def _allocate_network(
+ infoblox_params: settings.InfoBloxParams,
+ network_params: Union[settings.V4NetworkParams, settings.V6NetworkParams],
+ ip_version=4,
+ comment="",
+ extattrs={}
+) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+ assert ip_version in [4, 6]
+ endpoint = 'network' if ip_version == 4 else 'ipv6network'
+ ip_container = 'networkcontainer' if ip_version == 4 else \
+ 'ipv6networkcontainer'
+
+ # only return in the response the allocated network, not all available
+ # TODO: any validation needed for extrattrs wherever it's used?
+ req_payload = {
+ "network": {
+ "_object_function": "next_available_network",
+ "_parameters": {
+ "cidr": network_params.mask
+ },
+ "_object": ip_container,
+ "_object_parameters": {
+ "network": str(network_params.containers[0])
+ },
+ "_result_field": "networks",
+ },
+ "comment": comment,
+ "extattrs": extattrs
+ }
+
+ container_index = 0
+ while True:
+ r = requests.post(
+ f'{_wapi(infoblox_params)}/{endpoint}',
+ params={'_return_fields': 'network'},
+ json=req_payload,
+ auth=HTTPBasicAuth(infoblox_params.username,
+ infoblox_params.password),
+ headers={'content-type': "application/json"},
+ verify=False
+ )
+ if not _match_error_code(response=r,
+ error_code=IPAMErrors.CONTAINER_FULL):
+ break
+ # Container full: try with next valid container for service (if any)
+ container_index += 1
+ if len(network_params.containers) < (container_index + 1):
+ break
+ req_payload["network"]["_object_parameters"]["network"] = \
+ str(network_params.containers[container_index])
+
+ assert r.status_code >= 200 and r.status_code < 300, \
+ f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
+
+ assert 'network' in r.json()
+ allocated_network = r.json()['network']
+ if ip_version == 4:
+ return V4ServiceNetwork(v4=allocated_network)
+ else:
+ return V6ServiceNetwork(v6=allocated_network)
+
+
+def allocate_service_ipv4_network(service_type, comment="", extattrs={}
+ ) -> V4ServiceNetwork:
+ """
+ Allocate IPv4 network within the container of the specified service type.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM
+ ipam_params = oss.IPAM
+ 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,
+ comment,
+ extattrs)
+
+
+def allocate_service_ipv6_network(service_type, comment="", extattrs={}
+ ) -> V6ServiceNetwork:
+ """
+ Allocate IPv6 network within the container of the specified service type.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM
+ ipam_params = oss.IPAM
+ 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,
+ comment,
+ extattrs)
+
+
+def _find_next_available_ip(infoblox_params, network_ref):
+ r = requests.post(
+ f'{_wapi(infoblox_params)}/{network_ref}?_function=next_available_ip&num=1', # noqa: E501
+ auth=HTTPBasicAuth(infoblox_params.username,
+ infoblox_params.password),
+ verify=False
+ )
+ # TODO: propagate no more available IPs in the network
+ assert r.status_code >= 200 and r.status_code < 300, \
+ f"HTTP error {r.status_code}: {r.reason}\n\n{r.text}"
+ assert 'ips' in r.json()
+ received_ip = r.json()['ips']
+ assert len(received_ip) == 1
+ return received_ip[0]
+
+
+def _allocate_host(hostname=None, addr=None, network=None, extattrs={}
+ ) -> Union[V4HostAddress, V6HostAddress]:
+ """
+ If network is not None, allocate host in that network.
+ Otherwise if addr is not None, allocate host with that address.
+ hostname parameter must be full name including domain name.
+ """
+ # TODO: should hostnames be unique
+ # (i.e. fail if hostname already exists in this domain/service)?
+ assert addr or network, \
+ "You must specify either the host address or the network CIDR."
+ oss = settings.load_oss_params()
+ assert oss.IPAM.INFOBLOX
+ infoblox_params = oss.IPAM.INFOBLOX
+
+ if network:
+ ip_version = _ip_network_version(network)
+ # Find the next available IP address in the network
+ network_info = _find_networks(network=network, ip_version=ip_version)
+ assert len(network_info) == 1, \
+ "Network does not exist. Create it first."
+ assert '_ref' in network_info[0]
+ addr = _find_next_available_ip(infoblox_params,
+ network_info[0]["_ref"])
+
+ else:
+ ip_version = _ip_addr_version(addr)
+
+ ip_req_payload = {
+ f"ipv{ip_version}addrs": [
+ {
+ f"ipv{ip_version}addr": addr
+ }
+ ],
+ "name": hostname,
+ "configure_for_dns": False,
+ "view": "default",
+ "extattrs": extattrs
+ }
+
+ r = requests.post(
+ f'{_wapi(infoblox_params)}/record:host',
+ 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/")
+
+ dns_req_payload = {
+ f"ipv{ip_version}addr": addr,
+ "name": hostname,
+ "view": "default",
+ "extattrs": extattrs
+ }
+
+ endpoint = 'record:a' if ip_version == 4 else 'record:aaaa'
+
+ 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:
+ return V6HostAddress(v6=addr)
+
+
+def allocate_service_host(hostname=None,
+ service_type=None,
+ service_networks: ServiceNetworks = None,
+ host_addresses: HostAddresses = None,
+ extattrs={}
+ ) -> HostAddresses:
+ """
+ Allocate host with both IPv4 and IPv6 address (and respective DNS
+ records).
+ The domain name is also taken from the service type and appended to
+ specified hostname.
+ If service_networks is provided, that one is used.
+ If service_networks is not provided, and host_addresses is provided,
+ those specific addresses are used.
+ If neither is not provided, the first network with available space for
+ this service type is used.
+ Note that if WFO will always specify the network/addresses after
+ creating it, this mode won't be needed. Currently this mode doesn't
+ look further than the first container, so if needed, this will need
+ to be updated.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM
+ ipam_params = oss.IPAM
+
+ assert hasattr(ipam_params, service_type) \
+ and service_type != 'INFOBLOX', "Invalid service type."
+ ipv4_containers = getattr(ipam_params, service_type).V4.containers
+ ipv6_containers = getattr(ipam_params, service_type).V6.containers
+ domain_name = getattr(ipam_params, service_type).domain_name
+
+ # IPv4
+ if not service_networks and not host_addresses:
+ ipv4_networks_info = _find_networks(
+ network_container=str(ipv4_containers[0]), ip_version=4)
+ assert len(ipv4_networks_info) >= 1, \
+ "No IPv4 network exists in the container for this service type."
+ first_nonfull_ipv4_network = None
+ for ipv4_network_info in ipv4_networks_info:
+ assert 'network' in ipv4_network_info
+ capacity = _get_network_capacity(ipv4_network_info["network"])
+ if capacity < 1000:
+ first_nonfull_ipv4_network = ipv4_network_info["network"]
+ break
+ # Create a new network if the existing networks in the container for
+ # the service type are all full.
+ if not first_nonfull_ipv4_network:
+ first_nonfull_ipv4_network = str(allocate_service_ipv4_network(
+ service_type=service_type).v4)
+ assert first_nonfull_ipv4_network, \
+ "No available IPv4 addresses for this service type."
+ v4_host = _allocate_host(hostname=hostname+domain_name,
+ network=first_nonfull_ipv4_network,
+ extattrs=extattrs)
+ elif service_networks:
+ network = service_networks.v4
+ assert any(network.subnet_of(ipv4_container)
+ for ipv4_container in ipv4_containers)
+ v4_host = _allocate_host(hostname=hostname+domain_name,
+ network=str(network),
+ extattrs=extattrs)
+ elif host_addresses:
+ addr = host_addresses.v4
+ assert any(addr in ipv4_container
+ for ipv4_container in ipv4_containers)
+ v4_host = _allocate_host(hostname=hostname+domain_name,
+ addr=str(addr),
+ extattrs=extattrs)
+
+ # IPv6
+ if not service_networks and not host_addresses:
+ # ipv6 does not support capacity fetching (not even the GUI displays
+ # it). Maybe it's assumed that there is always available space?
+ ipv6_networks_info = _find_networks(
+ network_container=str(ipv6_containers[0]),
+ ip_version=6)
+ assert len(ipv6_networks_info) >= 1, \
+ "No IPv6 network exists in the container for this service type."
+ assert 'network' in ipv6_networks_info[0]
+ # TODO: if "no available IP" error, create a new network?
+ v6_host = _allocate_host(hostname=hostname+domain_name,
+ network=ipv6_networks_info[0]['network'],
+ extattrs=extattrs)
+ elif service_networks:
+ network = service_networks.v6
+ assert any(network.subnet_of(ipv6_container)
+ for ipv6_container in ipv6_containers)
+ v6_host = _allocate_host(hostname=hostname+domain_name,
+ network=str(network),
+ extattrs=extattrs)
+ elif host_addresses:
+ addr = host_addresses.v6
+ assert any(addr in ipv6_container
+ for ipv6_container in ipv6_containers)
+ v6_host = _allocate_host(hostname=hostname+domain_name,
+ addr=str(addr),
+ extattrs=extattrs)
+
+ return HostAddresses(v4=v4_host.v4, v6=v6_host.v6)
+
+
+"""
+Below methods are not used for supported outside calls
+"""
+
+'''
+def _find_containers(network=None, ip_version=4):
+ """
+ If network is not None, find that container.
+ Otherwise find all containers.
+ """
+ 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)}/{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 _delete_network(network) -> Union[V4ServiceNetwork, V6ServiceNetwork]:
+ """
+ Delete IPv4 or IPv6 network by CIDR.
+ """
+ # TODO: should we check that there are no hosts in this network before
+ # deleting? Deleting a network deletes the hosts in it, but not the
+ # associated DNS records.
+ oss = settings.load_oss_params()
+ assert oss.IPAM.INFOBLOX
+ infoblox_params = oss.IPAM.INFOBLOX
+
+ ip_version = _ip_network_version(network)
+
+ 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]
+
+ r = requests.delete(
+ f'{_wapi(infoblox_params)}/{network_info[0]["_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}"
+
+ # 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].replace("%3A", ":"))
+ if ip_version == 4:
+ return V4ServiceNetwork(v4=network_address)
+ else:
+ return V6ServiceNetwork(v6=network_address)
+
+
+def _delete_host_by_ip(addr) -> Union[V4HostAddress, V6HostAddress]:
+ """
+ Delete IPv4 or IPv6 host by its address.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM.INFOBLOX
+ infoblox_params = oss.IPAM.INFOBLOX
+
+ ip_version = _ip_addr_version(addr)
+ ip_param = 'ipv4addr' if ip_version == 4 else 'ipv6addr'
+
+ # Find host record reference
+ r = requests.get(
+ f'{_wapi(infoblox_params)}/record:host',
+ params={ip_param: addr},
+ auth=HTTPBasicAuth(infoblox_params.username,
+ infoblox_params.password),
+ verify=False
+ )
+ host_data = r.json()
+ assert len(host_data) == 1, "Host does not exist."
+ assert '_ref' in host_data[0]
+ host_ref = host_data[0]['_ref']
+
+ # Delete it
+ r = requests.delete(
+ f'{_wapi(infoblox_params)}/{host_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}"
+
+ # 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)
+ else:
+ return V6HostAddress(v6=addr)
+
+
+def _get_network_usage_status(network):
+ """
+ Get status and usage fields of all hosts in the specified ipv4 or ipv6
+ network.
+ """
+ oss = settings.load_oss_params()
+ assert oss.IPAM.INFOBLOX
+ infoblox_params = oss.IPAM.INFOBLOX
+
+ ip_version = _ip_network_version(network)
+ endpoint = 'ipv4address' if ip_version == 4 else 'ipv6address'
+
+ r = requests.get(
+ f'{_wapi(infoblox_params)}/{endpoint}',
+ params={
+ '_return_fields': 'ip_address,status,usage',
+ 'network': network},
+ 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()
+'''
+'''
+if __name__ == '__main__':
+ while True:
+ print("1. Find all containers")
+ print("2. Find all networks")
+ print("3. Get network capacity")
+ print("4. Create new network")
+ print("5. Delete network")
+ print("6. Allocate host by IP")
+ print("7. Allocate host by network CIDR")
+ print("8. Allocate host by service type")
+ print("9. Delete host by IP")
+ print("10. Get network usage status")
+ print("11. Exit")
+
+ choice = input("Enter your choice: ")
+
+ if choice == '1':
+ 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':
+ 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':
+ network = input("Enter network (in CIDR notation): ")
+ network_capacity = _get_network_capacity(network=network)
+ print(json.dumps(network_capacity, indent=2))
+
+ elif choice == '4':
+ service_type = input("Enter service type: ")
+ comment = input("Enter a comment for the network: ")
+ ip_version = int(input("Enter IP version (4 or 6): "))
+ if ip_version == 4:
+ new_network = allocate_service_ipv4_network(
+ comment=comment, service_type=service_type)
+ elif ip_version == 6:
+ new_network = allocate_service_ipv6_network(
+ comment=comment, service_type=service_type)
+ else:
+ print("Invalid IP version. Please enter either 4 or 6.")
+ continue
+ print(json.dumps(str(new_network), indent=2))
+
+ elif choice == '5':
+ network = input("Enter network to delete (in CIDR notation): ")
+ deleted_network = _delete_network(network=network)
+ print(json.dumps(str(deleted_network), indent=2))
+
+ elif choice == '6':
+ hostname = input("Enter host name (full name w/ domain name): ")
+ addr = input("Enter IP address to allocate: ")
+ alloc_ip = _allocate_host(hostname=hostname, addr=addr)
+ print(json.dumps(str(alloc_ip), indent=2))
+
+ elif choice == '7':
+ hostname = input("Enter host name (full name w/ domain name): ")
+ network = input(
+ "Enter existing network to allocate from (CIDR notation): ")
+ alloc_ip = _allocate_host(hostname=hostname, network=network)
+ print(json.dumps(str(alloc_ip), indent=2))
+
+ elif choice == '8':
+ hostname = input("Enter host name (w/o domain name): ")
+ service_type = input("Enter service type: ")
+ alloc_ip = allocate_service_host(
+ hostname=hostname,
+ service_type=service_type)
+ print(json.dumps(str(alloc_ip), indent=2))
+
+ elif choice == '9':
+ 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 == '10':
+ network = input(
+ "Enter network to get host usage status (CIDR notation): ")
+ usage_status_info = _get_network_usage_status(network=network)
+ print(json.dumps(usage_status_info, indent=2))
+
+ elif choice == '11':
+ print("Exiting...")
+ break
+
+ else:
+ print("Invalid choice. Please try again.")
+'''
diff --git a/gso/services/ipam.py b/gso/services/ipam.py
index 6e63130c6ed1cb00934b216bc6aee4ec8e2d80c5..c41a1239a14506e3104aaad5d5e4dcdfadd4dcae 100644
--- a/gso/services/ipam.py
+++ b/gso/services/ipam.py
@@ -1,6 +1,15 @@
import ipaddress
from pydantic import BaseSettings
-from gso import settings
+
+from gso.services import _ipam
+
+
+class V4ServiceNetwork(BaseSettings):
+ v4: ipaddress.IPv4Network
+
+
+class V6ServiceNetwork(BaseSettings):
+ v6: ipaddress.IPv6Network
class ServiceNetworks(BaseSettings):
@@ -8,26 +17,97 @@ class ServiceNetworks(BaseSettings):
v6: ipaddress.IPv6Network
+class V4HostAddress(BaseSettings):
+ v4: ipaddress.IPv4Address
+
+
+class V6HostAddress(BaseSettings):
+ v6: ipaddress.IPv6Address
+
+
class HostAddresses(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
+def new_service_networks(service_type,
+ comment="",
+ extattrs={}) -> ServiceNetworks:
+ v4_service_network = _ipam.allocate_service_ipv4_network(
+ service_type=service_type, comment=comment, extattrs=extattrs)
+ v6_service_network = _ipam.allocate_service_ipv6_network(
+ service_type=service_type, comment=comment, extattrs=extattrs)
return ServiceNetworks(
- v4=ipaddress.IPv4Network('10.0.0.0/24'),
- v6=ipaddress.IPv6Network('dead:beef::/120'))
+ v4=v4_service_network.v4,
+ v6=v6_service_network.v6)
+
+
+def new_service_host(hostname,
+ service_type,
+ service_networks: ServiceNetworks = None,
+ host_addresses: HostAddresses = None,
+ extattrs={}) -> HostAddresses:
+ return _ipam.allocate_service_host(
+ hostname=hostname,
+ service_type=service_type,
+ service_networks=service_networks,
+ host_addresses=host_addresses,
+ extattrs=extattrs)
+
+
+'''
+if __name__ == '__main__':
+ # sample call flow to allocate two loopback interfaces and a trunk service
+ # new_service_host can be called passing networks or addresses
+ # - host h1 for service LO uses a specific ipv4/ipv6 address pair
+ # - the rest use the ipv4/ipv6 network pair
+ # networks and hosts can be allocated with extensible attributes
+ # - host h2 for service LO uses extattrs for both network and address/DNS
+ # - the rest don't use extattrs
+
+ hostname_A = 'h1'
+ hostname_B = 'h2'
+
+ # h1 LO (loopback)
+ lo1_service_networks = new_service_networks(
+ service_type='LO',
+ comment="Network for h1 LO"
+ )
+ lo1_v4_host_address = lo1_service_networks.v4.network_address
+ lo1_v6_host_address = lo1_service_networks.v6.network_address
+ print(lo1_v4_host_address)
+ print(lo1_v6_host_address)
+ lo1_host_addresses = HostAddresses(v4=lo1_v4_host_address,
+ v6=lo1_v6_host_address)
+ new_service_host(hostname=hostname_A,
+ service_type='LO',
+ host_addresses=lo1_host_addresses)
+ # h2 LO (loopback)
+ lo2_network_extattrs = {
+ "vrf_name": {"value": "dummy_vrf"},
+ }
+ lo2_host_extattrs = {
+ "Site": {"value": "dummy_site"},
+ }
+ lo2_service_networks = \
+ new_service_networks(service_type='LO',
+ comment="Network for h2 LO",
+ extattrs=lo2_network_extattrs)
+ new_service_host(hostname=hostname_B,
+ service_type='LO',
+ service_networks=lo2_service_networks,
+ extattrs=lo2_host_extattrs)
-def new_device_lo_address() -> HostAddresses:
- oss = settings.load_oss_params()
- assert oss.IPAM.INFOBLOX
- # TODO: load from ipam
- return HostAddresses(
- v4=ipaddress.IPv4Address('10.10.10.10'),
- v6=ipaddress.IPv6Address('fc00:798:aa:1::10'))
+ # h1-h2 TRUNK
+ trunk12_service_networks = new_service_networks(
+ service_type='TRUNK',
+ comment="Network for h1-h2 TRUNK"
+ )
+ new_service_host(hostname=hostname_A,
+ service_type='TRUNK',
+ service_networks=trunk12_service_networks)
+ new_service_host(hostname=hostname_B,
+ service_type='TRUNK',
+ service_networks=trunk12_service_networks)
+'''
diff --git a/gso/settings.py b/gso/settings.py
index 11cebf9d63dcf14633cae189df201ab295d31532..76e1dccfd4cce663aa61aeb50c8ebebc51eabb56 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -17,22 +17,24 @@ class InfoBloxParams(BaseSettings):
class V4NetworkParams(BaseSettings):
- container: ipaddress.IPv4Network
+ containers: list[ipaddress.IPv4Network]
mask: int # TODO: validation on mask?
class V6NetworkParams(BaseSettings):
- container: ipaddress.IPv6Network
+ containers: list[ipaddress.IPv6Network]
mask: int # TODO: validation on mask?
class ServiceNetworkParams(BaseSettings):
V4: V4NetworkParams
V6: V6NetworkParams
+ domain_name: str
class IPAMParams(BaseSettings):
INFOBLOX: InfoBloxParams
+ LO: ServiceNetworkParams
TRUNK: ServiceNetworkParams
GEANT_IP: ServiceNetworkParams
diff --git a/requirements.txt b/requirements.txt
index 243292eec5c1bf77b11fc4166714d056f98fd313..3ac3e8394ad7a4649d8b87cd724f0d44ed8c1e03 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,6 @@
orchestrator-core==1.0.0
+pydantic
requests
pytest
+responses
diff --git a/setup.py b/setup.py
index 50d3dc63c04f6f68ec902ab6478942f252c4dc40..7062c773b0d214c582e7c5b9db49da806d1eae59 100644
--- a/setup.py
+++ b/setup.py
@@ -9,7 +9,8 @@ setup(
url=('https://gitlab.geant.org/goat/geant-service-orchestrator'),
packages=find_packages(),
install_requires=[
- 'requests',
'orchestrator-core==1.0.0',
+ 'pydantic',
+ 'requests',
]
)
diff --git a/test/conftest.py b/test/conftest.py
new file mode 100644
index 0000000000000000000000000000000000000000..2ef1475af538d31f0a235e78bd42fb79a834d859
--- /dev/null
+++ b/test/conftest.py
@@ -0,0 +1,76 @@
+import contextlib
+import json
+import os
+import socket
+import pytest
+import tempfile
+
+
+@pytest.fixture(scope='session')
+def configuration_data():
+ with contextlib.closing(
+ socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+ s.bind(('', 0))
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ yield {
+ "GENERAL": {
+ "public_hostname": "https://gap.geant.org"
+ },
+ "RESOURCE_MANAGER_API_PREFIX": "http://localhost:44444",
+ "IPAM": {
+ "INFOBLOX": {
+ "scheme": "https",
+ "wapi_version": "v2.12",
+ "host": "10.0.0.1",
+ "username": "robot-user",
+ "password": "robot-user-password"
+ },
+ "LO": {
+ "V4": {"containers": ["10.255.255.0/24"], "mask": 32},
+ "V6": {"containers": ["dead:beef::/64"], "mask": 128},
+ "domain_name": ".lo"
+ },
+ "TRUNK": {
+ "V4": {
+ "containers": ["10.255.255.0/24", "10.255.254.0/24"],
+ "mask": 31
+ },
+ "V6": {
+ "containers": ["dead:beef::/64", "dead:beee::/64"],
+ "mask": 126
+ },
+ "domain_name": ".trunk"
+ },
+ "GEANT_IP": {
+ "V4": {
+ "containers": ["10.255.255.0/24", "10.255.254.0/24"],
+ "mask": 31
+ },
+ "V6": {
+ "containers": ["dead:beef::/64", "dead:beee::/64"],
+ "mask": 126
+ },
+ "domain_name": ".geantip"
+ }
+ },
+ "PROVISIONING_PROXY": {
+ "scheme": "https",
+ "api_base": "localhost:44444",
+ "auth": "Bearer <token>",
+ "api_version": 1123
+ }
+ }
+
+
+@pytest.fixture(scope='session')
+def data_config_filename(configuration_data):
+ file_name = os.path.join(
+ tempfile.gettempdir(), os.urandom(24).hex())
+ open(file_name, 'x').close()
+ with open(file_name, 'wb') as f:
+ f.write(json.dumps(configuration_data).encode('utf-8'))
+ f.flush()
+
+ os.environ['OSS_PARAMS_FILENAME'] = f.name
+
+ yield f.name
diff --git a/test/test_ipam.py b/test/test_ipam.py
new file mode 100644
index 0000000000000000000000000000000000000000..2778148347a7fe84164305d86029efe3c5750ddc
--- /dev/null
+++ b/test/test_ipam.py
@@ -0,0 +1,118 @@
+import ipaddress
+import re
+import responses
+
+from gso.services import ipam
+
+
+@responses.activate
+def test_new_service_networks(data_config_filename):
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/network.*'),
+ json={
+ '_ref': 'network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default', # noqa: E501
+ 'network': '10.255.255.20/32'
+ }
+ )
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/ipv6network.*'),
+ json={
+ '_ref': 'ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default', # noqa: E501
+ 'network': 'dead:beef::18/128'
+ }
+ )
+
+ service_networks = ipam.new_service_networks(service_type='LO')
+ assert service_networks == ipam.ServiceNetworks(
+ v4=ipaddress.ip_network('10.255.255.20/32'),
+ v6=ipaddress.ip_network('dead:beef::18/128')
+ )
+
+
+@responses.activate
+def test_new_service_host(data_config_filename):
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/record:host$'),
+ json='record:host/ZG5zLmhvc3QkLm5vbl9ETlNfaG9zdF9yb290LjAuMTY4MzcwNTU4MzY3MC5nc28udGVzdA:test.lo/%20' # noqa: E501
+ )
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/record:a$'),
+ json='record:a/ZG5zLmJpbmRfYSQuX2RlZmF1bHQuZ3NvLHRlc3QsMTAuMjU1LjI1NS44:test.lo/default' # noqa: E501
+ )
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/record:aaaa$'),
+ json='record:aaaa/ZG5zLmJpbmRfYSQuX2RlZmF1bHQuZ3NvLHRlc3QsMTAuMjU1LjI1NS44:test.lo/default' # noqa: E501
+ )
+
+ responses.add(
+ method=responses.GET,
+ url=re.compile(r'.*/wapi.*/network.*'),
+ json=[
+ {
+ "_ref": "network/ZG5zLm5ldHdvcmskMTAuMjU1LjI1NS4yMC8zMi8w:10.255.255.20/32/default", # noqa: E501
+ "network": "10.255.255.20/32",
+ "network_view": "default"
+ }
+ ]
+
+ )
+
+ responses.add(
+ method=responses.GET,
+ url=re.compile(r'.*/wapi.*/ipv6network.*'),
+ json=[
+ {
+ "_ref": "ipv6network/ZG5zLm5ldHdvcmskZGVhZDpiZWVmOjoxOC8xMjgvMA:dead%3Abeef%3A%3A18/128/default", # noqa: E501
+ "network": "dead:beef::18/128",
+ "network_view": "default"
+ }
+ ]
+ )
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/network.*/.*?_function=next_available_ip&num=1.*'), # noqa: E501
+ json={'ips': ['10.255.255.20']}
+ )
+
+ responses.add(
+ method=responses.POST,
+ url=re.compile(r'.*/wapi.*/ipv6network.*/.*?_function=next_available_ip&num=1.*'), # noqa: E501
+ json={'ips': ['dead:beef::18']}
+ )
+
+ service_hosts = ipam.new_service_host(
+ hostname='test',
+ service_type='LO',
+ host_addresses=ipam.HostAddresses(
+ v4=ipaddress.ip_address('10.255.255.20'),
+ v6=ipaddress.ip_address('dead:beef::18')
+ )
+ )
+ assert service_hosts == ipam.HostAddresses(
+ v4=ipaddress.ip_address('10.255.255.20'),
+ v6=ipaddress.ip_address('dead:beef::18')
+ )
+
+ service_hosts = ipam.new_service_host(
+ hostname='test',
+ service_type='LO',
+ service_networks=ipam.ServiceNetworks(
+ v4=ipaddress.ip_network('10.255.255.20/32'),
+ v6=ipaddress.ip_network('dead:beef::18/128')
+ )
+ )
+ assert service_hosts == ipam.HostAddresses(
+ v4=ipaddress.ip_address('10.255.255.20'),
+ v6=ipaddress.ip_address('dead:beef::18')
+ )
diff --git a/test/test_placeholder.py b/test/test_placeholder.py
deleted file mode 100644
index e07460e2644214a04d3a149b0cf18c7de7367467..0000000000000000000000000000000000000000
--- a/test/test_placeholder.py
+++ /dev/null
@@ -1,4 +0,0 @@
-# just a placeholder to be able to run tests during ci
-
-def test_placeholder():
- pass