Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 91 additions & 7 deletions src/maasserver/models/iprange.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,12 @@ class IPRange(CleanSave, TimestampedModel):
"""Represents a range of IP addresses used for a particular purpose in
MAAS, such as a DHCP range or a range of reserved addresses."""

@classmethod
def from_db(cls, db, field_names, values):
instance = super().from_db(db, field_names, values)
instance._loaded_values = dict(zip(field_names, values))
return instance

objects = IPRangeManager()

subnet = ForeignKey(
Expand Down Expand Up @@ -274,18 +280,96 @@ def _validate_duplicates_and_overlaps(self):
else:
message += "IP address or range."

# Find unused range for start_ip
start_ip = IPAddress(self.start_ip)
end_ip = IPAddress(self.end_ip)

# Success when start and end are in the same unused range.
for unused_range in unused:
if IPAddress(self.start_ip) in unused_range:
if IPAddress(self.end_ip) in unused_range:
# Success, start and end IP are in an unused range.
return
else:
self._raise_validation_error(message)
if start_ip in unused_range and end_ip in unused_range:
return

if self._is_existing_dynamic_range_resize_allowed(
start_ip, end_ip, unused
):
return

self._raise_validation_error(message)

def _is_existing_dynamic_range_resize_allowed(
Comment thread
r00ta marked this conversation as resolved.
self, start_ip: IPAddress, end_ip: IPAddress, unused
) -> bool:
"""Allow resizing an existing dynamic range without re-querying the DB.

Uses cached original state (loaded_values) to compare the current
range against the persisted range, identifying only the newly added
segments (left/right expansion). Returns True if the range is merely
shrinking (no added segments) or if all newly added segments fit
within available unused ranges. Returns False if the original range
was not dynamic, the range moved to a different subnet, or any added
segment would overlap allocated IPs or other ranges.
Comment thread
r00ta marked this conversation as resolved.
"""
if self.id is None or self.type != IPRANGE_TYPE.DYNAMIC:
return False

loaded_values = getattr(self, "_loaded_values", None)
if not loaded_values:
return False

if loaded_values.get("type") != IPRANGE_TYPE.DYNAMIC:
return False

if loaded_values.get("subnet_id") != self.subnet_id:
return False

existing_start_ip = loaded_values.get("start_ip")
existing_end_ip = loaded_values.get("end_ip")
if existing_start_ip is None or existing_end_ip is None:
return False

existing_start_ip = IPAddress(existing_start_ip)
existing_end_ip = IPAddress(existing_end_ip)

start = int(start_ip)
end = int(end_ip)
existing_start = int(existing_start_ip)
existing_end = int(existing_end_ip)

added_segments = []

left_end = min(end, existing_start - 1)
if start <= left_end:
added_segments.append((start, left_end))

right_start = max(start, existing_end + 1)
if right_start <= end:
added_segments.append((right_start, end))

if not added_segments:
return True

version = start_ip.version
for segment_start, segment_end in added_segments:
added_start_ip = IPAddress(segment_start, version=version)
added_end_ip = IPAddress(segment_end, version=version)
if not any(
added_start_ip in unused_range and added_end_ip in unused_range
for unused_range in unused
):
return False

return True

def _cache_loaded_values(self):
self._loaded_values = {
"type": self.type,
"subnet_id": self.subnet_id,
"start_ip": self.start_ip,
"end_ip": self.end_ip,
}

def save(self, *args, **kwargs):
super().save(*args, **kwargs)
self._cache_loaded_values()

if self.subnet.vlan.dhcp_on:
post_commit_do(
Expand Down
235 changes: 235 additions & 0 deletions src/maasserver/models/tests/test_iprange.py
Original file line number Diff line number Diff line change
Expand Up @@ -682,6 +682,31 @@ def test_no_save_when_no_ranges_available(self):
},
)

def test_create_dynamic_range_spanning_assigned_ip_fails(self):
subnet = make_plain_subnet()
factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.30",
)
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.20",
end_ip="192.168.0.40",
)

with self.assertRaises(ValidationError) as cm:
iprange.clean()

self.assertEqual(
cm.exception.message_dict,
{
"start_ip": [self.dynamic_overlaps],
"end_ip": [self.dynamic_overlaps],
},
)

def test_modify_existing_performs_validation(self):
subnet = make_plain_subnet()
IPRange(
Expand Down Expand Up @@ -719,6 +744,216 @@ def test_modify_existing_performs_validation(self):
iprange = reload_object(iprange)
self.assertEqual(iprange.id, instance_id)

def test_modify_existing_dynamic_range_can_shrink_with_allocated_ip(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.30",
)

iprange.end_ip = "192.168.0.49"
iprange.clean()
iprange.save()

def test_modify_existing_dynamic_range_can_expand_with_allocated_ip(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.30",
)

iprange.end_ip = "192.168.0.60"
iprange.clean()
iprange.save()
Comment thread
r00ta marked this conversation as resolved.

def test_modify_dynamic_range_expand_start_fails_on_busy_edge_ip(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.9",
)

iprange.start_ip = "192.168.0.9"
with self.assertRaises(ValidationError) as cm:
iprange.clean()

self.assertEqual(
cm.exception.message_dict,
{
"start_ip": [self.dynamic_overlaps],
"end_ip": [self.dynamic_overlaps],
},
)

def test_modify_dynamic_range_expand_end_fails_on_busy_edge_ip(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.51",
)

iprange.end_ip = "192.168.0.51"
with self.assertRaises(ValidationError) as cm:
iprange.clean()

self.assertEqual(
cm.exception.message_dict,
{
"start_ip": [self.dynamic_overlaps],
"end_ip": [self.dynamic_overlaps],
},
)

def test_modify_dynamic_range_shrink_without_original_query(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.30",
)

iprange = reload_object(iprange)
mock_filter = self.patch(IPRange.objects, "filter")

iprange.end_ip = "192.168.0.49"
iprange.clean()

mock_filter.assert_not_called()

def test_move_dynamic_range_to_new_subnet_disables_resize_shortcut(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

iprange = reload_object(iprange)
other_subnet = factory.make_Subnet(
cidr="192.168.1.0/24",
gateway_ip="192.168.1.1",
dns_servers=[],
)
iprange.subnet = other_subnet
iprange.start_ip = "192.168.1.20"
iprange.end_ip = "192.168.1.30"

unused = other_subnet.get_ipranges_available_for_dynamic_range(
exclude_ip_range_id=iprange.id
)

self.assertFalse(
iprange._is_existing_dynamic_range_resize_allowed(
iprange.netaddr_iprange.first,
iprange.netaddr_iprange.last,
unused,
)
)

def test_move_dynamic_range_to_new_subnet_with_clash_fails_clean(self):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.10",
end_ip="192.168.0.50",
)
iprange.save()

other_subnet = factory.make_Subnet(
cidr="192.168.1.0/24",
gateway_ip="192.168.1.1",
dns_servers=[],
)
IPRange(
subnet=other_subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.1.25",
end_ip="192.168.1.35",
).save()

iprange = reload_object(iprange)
iprange.subnet = other_subnet
iprange.start_ip = "192.168.1.20"
iprange.end_ip = "192.168.1.30"

with self.assertRaises(ValidationError) as cm:
iprange.clean()

self.assertEqual(
cm.exception.message_dict,
{
"start_ip": [self.dynamic_overlaps],
"end_ip": [self.dynamic_overlaps],
},
)

def test_modify_existing_dynamic_range_can_expand_start_with_allocated_ip(
self,
):
subnet = make_plain_subnet()
iprange = IPRange(
subnet=subnet,
type=IPRANGE_TYPE.DYNAMIC,
start_ip="192.168.0.20",
end_ip="192.168.0.50",
)
iprange.save()
factory.make_StaticIPAddress(
subnet=subnet,
alloc_type=IPADDRESS_TYPE.AUTO,
ip="192.168.0.30",
)
iprange.start_ip = "192.168.0.10"
iprange.clean()
iprange.save()

def test_dynamic_range_cant_overlap_gateway_ip(self):
subnet = make_plain_subnet()
iprange = IPRange(
Expand Down
Loading