diff --git a/pycroft/model/net.py b/pycroft/model/net.py index 710f149f3..f644bcfe5 100644 --- a/pycroft/model/net.py +++ b/pycroft/model/net.py @@ -56,18 +56,27 @@ class Subnet(IntegerIdModel): # /backrefs @property - def usable_ip_range(self) -> netaddr.IPRange | None: - """All IPs in this subnet which are not reserved.""" + def reserved_ipset(self) -> netaddr.IPSet: + res_bottom = self.reserved_addresses_bottom or 0 + res_top = self.reserved_addresses_top or 0 # takes care of host- and broadcast domains plus edge-cases (e.g. /32) first_usable, last_usable = self.address._usable_range() + return netaddr.IPSet( + [ + netaddr.IPRange(self.address[0], first_usable + res_bottom), + netaddr.IPRange(last_usable - res_top, self.address[-1]), + ] + ) - res_bottom = self.reserved_addresses_bottom or 0 - res_top = self.reserved_addresses_top or 0 - first_usable = first_usable + res_bottom - last_usable = last_usable + res_top - if last_usable < first_usable: - return None - return netaddr.IPRange(first_usable, last_usable) + def reserved_ip_ranges_iter(self) -> t.Iterator[netaddr.IPRange]: + return self.reserved_ipset.iter_ipranges() + + @property + def usable_ip_range(self) -> netaddr.IPRange | None: + """All IPs in this subnet which are not reserved.""" + usable = netaddr.IPSet(self.address) - self.reserved_ipset + assert usable.iscontiguous(), f"Complement of reserved ranges in {self} is not contiguous" + return usable.iprange() @property def usable_size(self) -> int: diff --git a/tests/frontend/test_infrastructure.py b/tests/frontend/test_infrastructure.py index f50ff30ea..e5cb4d729 100644 --- a/tests/frontend/test_infrastructure.py +++ b/tests/frontend/test_infrastructure.py @@ -5,7 +5,7 @@ import typing as t import pytest -from netaddr import IPAddress, IPNetwork +from netaddr import IPNetwork from sqlalchemy.orm import Session from flask import url_for @@ -15,7 +15,6 @@ from pycroft.model.port import PatchPort from tests import factories as f from tests.assertions import assert_one -from web.blueprints.infrastructure import format_address_range from .assertions import TestClient @@ -31,11 +30,6 @@ def client(module_test_client: TestClient) -> TestClient: return module_test_client -def test_format_empty_address_range(): - with pytest.raises(ValueError): - format_address_range(IPAddress("141.30.228.39"), amount=0) - - @pytest.mark.usefixtures("admin_logged_in", "session") class TestSubnets: @pytest.fixture(scope="class", autouse=True) diff --git a/web/blueprints/host/__init__.py b/web/blueprints/host/__init__.py index 7924a59ba..2b2ae4225 100644 --- a/web/blueprints/host/__init__.py +++ b/web/blueprints/host/__init__.py @@ -4,7 +4,7 @@ from flask.typing import ResponseReturnValue from flask_login import current_user from flask_wtf import FlaskForm -from ipaddr import IPv4Address +from netaddr import IPAddress from pycroft.exc import PycroftException from pycroft.helpers.net import mac_regex, get_interface_manufacturer @@ -269,7 +269,7 @@ def default_response() -> ResponseReturnValue: if not form.validate(): return default_response() - ips = {IPv4Address(ip) for ip in form.ips.data} + ips = {IPAddress(ip) for ip in form.ips.data} with abort_on_error(default_response), session.session.begin_nested(): lib_host.interface_edit( @@ -306,7 +306,7 @@ def default_response() -> ResponseReturnValue: if not form.validate(): return default_response() - ips = {IPv4Address(ip) for ip in form.ips.data} + ips = {IPAddress(ip) for ip in form.ips.data} try: lib_host.interface_create( diff --git a/web/blueprints/infrastructure/__init__.py b/web/blueprints/infrastructure/__init__.py index fb971e9cb..1e303a965 100644 --- a/web/blueprints/infrastructure/__init__.py +++ b/web/blueprints/infrastructure/__init__.py @@ -14,7 +14,7 @@ from flask.typing import ResponseValue from flask_login import current_user from flask_wtf import FlaskForm as Form -from ipaddr import IPAddress, _BaseIP +from netaddr import IPAddress from sqlalchemy.orm import joinedload from pycroft.lib.infrastructure import create_switch, \ @@ -56,27 +56,8 @@ def subnets() -> ResponseValue: subnet_table=SubnetTable(data_url=url_for(".subnets_json"))) -def format_address_range(base_address: _BaseIP, amount: int) -> str: - if amount == 0: - raise ValueError - if abs(amount) == 1: - return str(base_address) - if amount > 1: - return f'{str(base_address)} - {str(base_address + amount - 1)}' - return f'{str(base_address + amount + 1)} - {str(base_address)}' - - def format_reserved_addresses(subnet: Subnet) -> list[str]: - reserved = [] - if subnet.reserved_addresses_bottom: - reserved.append( - format_address_range(subnet.address.network + 1, - subnet.reserved_addresses_bottom)) - if subnet.reserved_addresses_top: - reserved.append( - format_address_range(subnet.address.broadcast - 1, - -subnet.reserved_addresses_top)) - return reserved + return [str(range) for range in subnet.reserved_ip_ranges_iter()] @bp.route('/subnets/json') diff --git a/web/blueprints/user/log.py b/web/blueprints/user/log.py index 889197887..68ccac3c4 100644 --- a/web/blueprints/user/log.py +++ b/web/blueprints/user/log.py @@ -7,7 +7,7 @@ import logging import typing as t -from ipaddr import _BaseIP +from netaddr import IPAddress from sqlalchemy import select, Row from sqlalchemy.orm import Query @@ -24,7 +24,7 @@ logger = logging.getLogger(__name__) -def iter_hades_switch_ports(room: Room) -> t.Sequence[Row[tuple[str, _BaseIP]]]: +def iter_hades_switch_ports(room: Room) -> t.Sequence[Row[tuple[str, IPAddress]]]: """Return all tuples of (nasportid, nasipaddress) for a room. :param room: The room to filter by