Skip to content
Snippets Groups Projects
Commit 551aa18c authored by JORGE SASIAIN's avatar JORGE SASIAIN
Browse files

NAT-199: implement delete host in IPAM

parent 4ce8282e
No related branches found
No related tags found
2 merge requests!27Merge develop into NAT-185,!15Nat 185
......@@ -81,6 +81,32 @@ def _ip_network_version(network):
return ip_version
def _assert_host_in_service(
ipv4_addr='', ipv6_addr='',
oss_ipv4_containers=None, oss_ipv6_containers=None,
oss_ipv4_networks=None, oss_ipv6_networks=None
):
# IPv4
if oss_ipv4_containers:
assert any(ipv4_addr in oss_ipv4_container
for oss_ipv4_container in oss_ipv4_containers), \
"Host's IPv4 address doesn't belong to service type."
else:
assert any(ipv4_addr in oss_ipv4_network
for oss_ipv4_network in oss_ipv4_networks), \
"Host's IPv4 address doesn't belong to service type."
# IPv6
if oss_ipv6_containers:
assert any(ipv6_addr in oss_ipv6_container
for oss_ipv6_container in oss_ipv6_containers), \
"Host's IPv6 address doesn't belong to service type."
else:
assert any(ipv6_addr in oss_ipv6_network
for oss_ipv6_network in oss_ipv6_networks), \
"Host's IPv6 address doesn't belong to service type."
def _find_networks(network_container=None, network=None, ip_version=4):
"""
If network_container is not None, find all networks within the specified
......@@ -248,7 +274,10 @@ def _allocate_host(hostname='',
If networks is not None, allocate host in those networks.
Otherwise if addrs is not None, allocate host with those addresses.
hostname parameter must be full name including domain name.
Return an error string if couldn't allocate host due to network full.
Return "IPV4_NETWORK_FULL" or "IPV6_NETWORK_FULL"
if couldn't allocate host due to requested network being full.
Return "IPV4_NETWORK_NOT_FOUND" or "IPV6_NETWORK_NOT_FOUND"
if couldn't allocate host due to requested network not existing.
"""
# TODO: should hostnames be unique
# (i.e. fail if hostname already exists in this domain/service)?
......@@ -473,23 +502,13 @@ def allocate_service_host(hostname='',
"Network does not exist. Create it first."
elif host_addresses:
# IPv4
ipv4_addr = host_addresses.v4
if oss_ipv4_containers:
assert any(ipv4_addr in oss_ipv4_container
for oss_ipv4_container in oss_ipv4_containers)
else:
assert any(ipv4_addr in oss_ipv4_network
for oss_ipv4_network in oss_ipv4_networks)
# IPv6
ipv6_addr = host_addresses.v6
if oss_ipv6_containers:
assert any(ipv6_addr in oss_ipv6_container
for oss_ipv6_container in oss_ipv6_containers)
else:
assert any(ipv6_addr in oss_ipv6_network
for oss_ipv6_network in oss_ipv6_networks)
_assert_host_in_service(
ipv4_addr, ipv6_addr,
oss_ipv4_containers, oss_ipv6_containers,
oss_ipv4_networks, oss_ipv6_networks
)
host = _allocate_host(
hostname=hostname+domain_name,
......@@ -569,73 +588,98 @@ def delete_service_network(ipnetwork=None, service_type=''
return V6ServiceNetwork(v6=ipaddress.ip_network(network_address))
# def delete_service_host(
# hostname='',
# host_addresses: HostAddresses = None,
# cname_aliases=[],
# service_type=''
# ) -> 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 delete_service_host(
hostname='',
host_addresses: HostAddresses = None,
cname_aliases=[],
service_type=''
) -> Union[V4HostAddress, V6HostAddress]:
"""
Delete host record and associated CNAME records.
All arguments passed to this function must match together a host record in
IPAM, and all CNAME records associated to it must also be passed exactly.
"""
oss = settings.load_oss_params()
assert oss.IPAM
ipam_params = oss.IPAM
assert ipam_params.INFOBLOX
infoblox_params = ipam_params.INFOBLOX
assert hasattr(ipam_params, service_type) \
and service_type != 'INFOBLOX', "Invalid service type."
oss_ipv4_containers = getattr(ipam_params, service_type).V4.containers
oss_ipv6_containers = getattr(ipam_params, service_type).V6.containers
oss_ipv4_networks = getattr(ipam_params, service_type).V4.networks
oss_ipv6_networks = getattr(ipam_params, service_type).V6.networks
domain_name = getattr(ipam_params, service_type).domain_name
dns_view = getattr(ipam_params, service_type).dns_view
ipv4_addr = str(host_addresses.v4)
ipv6_addr = str(host_addresses.v6)
_assert_host_in_service(
host_addresses.v4, host_addresses.v6,
oss_ipv4_containers, oss_ipv6_containers,
oss_ipv4_networks, oss_ipv6_networks
)
# Find host record reference
r = requests.get(
f'{_wapi(infoblox_params)}/record:host',
params={
'name': (hostname+domain_name).lower(), # hostnames are lowercase
'ipv4addr': ipv4_addr,
'ipv6addr': ipv6_addr,
'view': dns_view,
},
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']
# Find cname records reference
r = requests.get(
f'{_wapi(infoblox_params)}/record:cname',
params={
'canonical': hostname+domain_name,
"view": dns_view,
},
auth=HTTPBasicAuth(infoblox_params.username,
infoblox_params.password),
verify=False
)
cname_data = r.json()
provided_cnames = [item + domain_name for item in cname_aliases]
found_cnames = [item['name'] for item in cname_data if 'name' in item]
assert provided_cnames == found_cnames, \
"Provided CNAME alias names don't match the ones poiting to hostname."
# Delete the host record
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}"
# Delete the CNAME records
cname_refs = [item['_ref'] for item in cname_data if 'name' in item]
for cname_ref in cname_refs:
r = requests.delete(
f'{_wapi(infoblox_params)}/{cname_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}"
return host_addresses
"""
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment