Skip to content

Commit

Permalink
Fix and simplify ipaddr dependent web code
Browse files Browse the repository at this point in the history
  • Loading branch information
lukasjuhrich committed Jul 5, 2024
1 parent b9f4407 commit 5a8b0f8
Show file tree
Hide file tree
Showing 5 changed files with 26 additions and 42 deletions.
27 changes: 18 additions & 9 deletions pycroft/model/net.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
8 changes: 1 addition & 7 deletions tests/frontend/test_infrastructure.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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


Expand All @@ -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)
Expand Down
6 changes: 3 additions & 3 deletions web/blueprints/host/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
23 changes: 2 additions & 21 deletions web/blueprints/infrastructure/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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, \
Expand Down Expand Up @@ -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')
Expand Down
4 changes: 2 additions & 2 deletions web/blueprints/user/log.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand Down

0 comments on commit 5a8b0f8

Please sign in to comment.