Skip to content

Commit

Permalink
Merge 0883dfc into c4a1683
Browse files Browse the repository at this point in the history
  • Loading branch information
@nemesifier
nemesifier authored May 29, 2024
2 parents c4a1683 + 0883dfc commit 2a7a2bb
Show file tree
Hide file tree
Showing 3 changed files with 196 additions and 20 deletions.
20 changes: 13 additions & 7 deletions openwisp_controller/subnet_division/base/models.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ def _validate_existing_fields(self):

def _validate_master_subnet_consistency(self):
master_subnet = self.master_subnet.subnet
# Validate size of generated subnet is not greater than size of master subnet
# Ensure that the size of the generated subnet
# is not greater than size of master subnet
try:
next(master_subnet.subnets(new_prefix=self.size))
except ValueError:
Expand All @@ -118,13 +119,19 @@ def _validate_master_subnet_consistency(self):
}
)

# Validate master subnet can accommodate required number of generated subnets
if self.number_of_subnets > (2 ** (self.size - master_subnet.prefixlen)):
# Ensure that master subnet can accommodate
# the required number of generated subnets
available = 2 ** (self.size - master_subnet.prefixlen)
# Account for the reserved subnet
available -= 1
if self.number_of_subnets >= available:
raise ValidationError(
{
'number_of_subnets': _(
f'Master subnet cannot accommodate {self.number_of_subnets} '
f'subnets of size /{self.size}'
'The master subnet is too small to acommodate the '
'requested "number of subnets" plus the reserved '
'subnet, please increase the size of the master '
'subnet or decrease the "size of subnets" field.'
)
}
)
Expand All @@ -139,11 +146,10 @@ def _validate_master_subnet_consistency(self):
)

def _validate_ip_address_consistency(self):
# Validate individual generated subnet can accommodate required number of IPs
try:
next(
ip_network(str(self.master_subnet.subnet)).subnets(new_prefix=self.size)
)[self.number_of_ips]
)[self.number_of_ips - 1]
except IndexError:
raise ValidationError(
{
Expand Down
26 changes: 19 additions & 7 deletions openwisp_controller/subnet_division/rule_types/base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -194,8 +194,7 @@ def get_max_subnet(master_subnet, division_rule):
subnet_obj.full_clean()
subnet_obj.save()
max_subnet = subnet_obj.subnet
finally:
return max_subnet
return max_subnet

@staticmethod
def create_subnets(config, division_rule, max_subnet, generated_indexes):
Expand Down Expand Up @@ -234,24 +233,37 @@ def create_subnets(config, division_rule, max_subnet, generated_indexes):
def create_ips(config, division_rule, generated_subnets, generated_indexes):
generated_ips = []
for subnet_obj in generated_subnets:
for ip_id in range(1, division_rule.number_of_ips + 1):
# don't assign first ip address of a subnet,
# unless the rule is designed to use the whole
# address space of the subnet
if subnet_obj.subnet.num_addresses != division_rule.number_of_ips:
index_start = 1
index_end = division_rule.number_of_ips + 1
# this allows handling /32, /128 or cases in which
# the number of requested ip addresses matches exactly
# what is available in the subnet
else:
index_start = 0
index_end = division_rule.number_of_ips
# generate IPs and indexes accordingly
for ip_index in range(index_start, index_end):
ip_obj = IpAddress(
subnet_id=subnet_obj.id,
ip_address=str(subnet_obj.subnet[ip_id]),
ip_address=str(subnet_obj.subnet[ip_index]),
)
ip_obj.full_clean()
generated_ips.append(ip_obj)

# ensure human friendly labels (starting from 1 instead of 0)
keyword_index = ip_index if index_start == 1 else ip_index + 1
generated_indexes.append(
SubnetDivisionIndex(
keyword=f'{subnet_obj.name}_ip{ip_id}',
keyword=f'{subnet_obj.name}_ip{keyword_index}',
subnet_id=subnet_obj.id,
ip_id=ip_obj.id,
rule_id=division_rule.id,
config=config,
)
)

IpAddress.objects.bulk_create(generated_ips)
return generated_ips

Expand Down
170 changes: 164 additions & 6 deletions openwisp_controller/subnet_division/tests/test_models.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,11 @@ def test_field_validations(self):
rule.full_clean()
expected_message_dict = {
'number_of_subnets': [
'Master subnet cannot accommodate 99999999 subnets of size /28'
'The master subnet is too small to acommodate '
'the requested "number of subnets" plus the '
'reserved subnet, please increase the size of '
'the master subnet or decrease the '
'"size of subnets" field.'
]
}
self.assertDictEqual(
Expand Down Expand Up @@ -246,6 +250,73 @@ def test_field_validations(self):
{'organization': ['Organization should be same as the subnet']},
)

def test_slash_32_rule_ipv4(self):
rule = self._get_vpn_subdivision_rule(
size=32, number_of_ips=1, number_of_subnets=1
)
self.config.templates.add(self.template)
rule.subnetdivisionindex_set.count()
index_queryset = rule.subnetdivisionindex_set.filter(
config_id=self.config.id, subnet_id__isnull=False, ip_id__isnull=False
)
self.assertEqual(index_queryset.count(), 1)
index = index_queryset.first()
self.assertEqual(str(index.subnet.subnet), '10.0.0.1/32')
self.assertEqual(index.ip.ip_address, '10.0.0.1')

def test_slash_32_rule_ipv4_error(self):
master_ipv4 = self._get_master_subnet(subnet='192.168.1.1/32')
self.vpn_server.subnet = master_ipv4
self.vpn_server.save()
try:
self._get_vpn_subdivision_rule(
size=32, number_of_ips=1, number_of_subnets=1, master_subnet=master_ipv4
)
except ValidationError as e:
self.assertIn('number_of_subnets', e.message_dict)
self.assertIn(
'The master subnet is too small to acommodate',
e.message_dict['number_of_subnets'][0],
)
else:
self.fail('Expected error not raised')

def test_slash_128_rule_ipv6(self):
master_ipv6 = self._get_master_subnet(subnet='fd12:3456:7890::/48')
self.vpn_server.subnet = master_ipv6
self.vpn_server.save()
rule = self._get_vpn_subdivision_rule(
size=128, number_of_ips=1, number_of_subnets=1, master_subnet=master_ipv6
)
self.config.templates.add(self.template)
index_queryset = rule.subnetdivisionindex_set.filter(
config_id=self.config.id, subnet_id__isnull=False, ip_id__isnull=False
)
self.assertEqual(index_queryset.count(), 1)
index = index_queryset.first()
self.assertEqual(str(index.subnet.subnet), 'fd12:3456:7890::1/128')
self.assertEqual(index.ip.ip_address, 'fd12:3456:7890::1')

def test_slash_128_rule_ipv6_error(self):
master_ipv6 = self._get_master_subnet(subnet='fd12:3456:7890::/128')
self.vpn_server.subnet = master_ipv6
self.vpn_server.save()
try:
self._get_vpn_subdivision_rule(
size=128,
number_of_ips=1,
number_of_subnets=1,
master_subnet=master_ipv6,
)
except ValidationError as e:
self.assertIn('number_of_subnets', e.message_dict)
self.assertIn(
'The master subnet is too small to acommodate',
e.message_dict['number_of_subnets'][0],
)
else:
self.fail('Expected error not raised')

def test_rule_label_updated(self):
new_rule_label = 'TSDR'
rule = self._get_vpn_subdivision_rule(label='VPN_OW')
Expand Down Expand Up @@ -418,11 +489,20 @@ def test_multiple_vpnclient_delete(self):
@patch('logging.Logger.error')
def test_subnets_exhausted(self, mocked_logger):
subnet = self._get_master_subnet(
'10.0.0.0/28', master_subnet=self.master_subnet
)
'10.0.0.0/29', master_subnet=self.master_subnet
)
# The master subnet can acommodate
# this rule only once:
# A /29 has 4 /31 slots available
# Minus the reserved subnet = 3
# Each run will eat 2 slots.
# Hence we expect this to run fine the
# first time but fail the second time.
self._get_vpn_subdivision_rule(
master_subnet=subnet,
size=29,
size=31,
number_of_ips=2,
number_of_subnets=2,
)
self.vpn_server.subnet = subnet
self.vpn_server.save()
Expand Down Expand Up @@ -577,16 +657,94 @@ def test_device_subnet_division_rule(self):
'backend': 'netjsonconfig.OpenWrt',
}
response = self.client.post(reverse('controller:device_register'), options)
lines = response.content.decode().split('\n')
self.assertEqual(lines[0], 'registration-result: success')
self.assertEqual(response.status_code, 201)

# Verify generated subnets and IP addresses match expectations
self.assertEqual(
subnet_query.count(),
rule.number_of_subnets,
)
self.assertEqual(
self.ip_query.count(), (rule.number_of_subnets * rule.number_of_ips)
)
subnets = subnet_query.order_by('created')
subnet1 = subnets[0]
subnet2 = subnets[1]
self.assertEqual(str(subnet1.subnet), '10.0.0.16/28')
self.assertEqual(str(subnet2.subnet), '10.0.0.32/28')
self.assertEqual(subnet1.ipaddress_set.count(), 2)
self.assertEqual(subnet2.ipaddress_set.count(), 2)
subnet1_ips = list(subnet1.ipaddress_set.order_by('created').all())
with self.subTest('Check IP addresses of subnet1'):
self.assertEqual(str(subnet1_ips[0].ip_address), '10.0.0.17')
self.assertEqual(str(subnet1_ips[1].ip_address), '10.0.0.18')
subnet2_ips = list(subnet2.ipaddress_set.order_by('created').all())
with self.subTest('Check IP addresses of subnet2'):
self.assertEqual(str(subnet2_ips[0].ip_address), '10.0.0.33')
self.assertEqual(str(subnet2_ips[1].ip_address), '10.0.0.34')

# Verify context of config
device = Device.objects.get(mac_address='FF:FF:FF:FF:FF:FF')
context = get_subnet_division_config_context(device.config)
self.assertIn(f'{rule.label}_prefixlen', context)
for subnet_id in range(1, rule.number_of_subnets + 1):
self.assertIn(f'{rule.label}_subnet{subnet_id}', context)
for ip_id in range(1, rule.number_of_ips + 1):
self.assertIn(f'{rule.label}_subnet{subnet_id}_ip{ip_id}', context)

# Verify working of delete handler
device.delete()
self.assertEqual(
subnet_query.count(),
0,
)
self.assertEqual(self.ip_query.count(), 0)

def test_device_rule_use_entire_subnet(self):
self.config.delete()
rule = self._get_device_subdivision_rule(size=29, number_of_ips=8)
OrganizationConfigSettings.objects.create(
organization=self.org, shared_secret='shared_secret'
)
subnet_query = self.subnet_query.filter(organization_id=self.org.id).exclude(
id=self.master_subnet.id
)
self.assertEqual(subnet_query.count(), 0)

# Register device
options = {
'hardware_id': '1234',
'secret': 'shared_secret',
'name': 'FF:FF:FF:FF:FF:FF',
'mac_address': 'FF:FF:FF:FF:FF:FF',
'backend': 'netjsonconfig.OpenWrt',
}
response = self.client.post(reverse('controller:device_register'), options)
self.assertEqual(response.status_code, 201)

# Verify generated subnets and IP addresses match expectations
self.assertEqual(
subnet_query.count(),
rule.number_of_subnets,
)
self.assertEqual(
self.ip_query.count(), (rule.number_of_subnets * rule.number_of_ips)
)
subnets = subnet_query.order_by('created')
subnet1 = subnets[0]
subnet2 = subnets[1]
self.assertEqual(str(subnet1.subnet), '10.0.0.8/29')
self.assertEqual(str(subnet2.subnet), '10.0.0.16/29')
self.assertEqual(subnet1.ipaddress_set.count(), 8)
self.assertEqual(subnet2.ipaddress_set.count(), 8)

number = 8
for subnet in [subnet1, subnet2]:
for ip in subnet.ipaddress_set.order_by('created').all():
expected_ip = f'10.0.0.{number}'
with self.subTest(f'Expect IP address: {expected_ip}'):
self.assertEqual(str(ip.ip_address), expected_ip)
number += 1

# Verify context of config
device = Device.objects.get(mac_address='FF:FF:FF:FF:FF:FF')
Expand Down

0 comments on commit 2a7a2bb

Please sign in to comment.