try to restore previous ip lease
This commit is contained in:
parent
c89ec7f957
commit
db9119b88a
1 changed files with 62 additions and 12 deletions
|
@ -5,13 +5,22 @@ from typing import Any, Iterator
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Address:
|
||||||
|
address: str
|
||||||
|
family: str
|
||||||
|
prefixlen: int
|
||||||
|
preferred_life_time: int = 0
|
||||||
|
valid_life_time: int = 0
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class Interface:
|
class Interface:
|
||||||
name: str
|
name: str
|
||||||
ifname: str | None
|
ifname: str | None
|
||||||
mac_address: str
|
mac_address: str
|
||||||
dynamic_addresses: list[str]
|
dynamic_addresses: list[Address]
|
||||||
static_addresses: list[dict[str, Any]]
|
static_addresses: list[Address]
|
||||||
static_routes: list[dict[str, Any]]
|
static_routes: list[dict[str, Any]]
|
||||||
|
|
||||||
|
|
||||||
|
@ -25,14 +34,35 @@ def filter_interfaces(network: list[dict[str, Any]]) -> list[Interface]:
|
||||||
continue
|
continue
|
||||||
static_addresses = []
|
static_addresses = []
|
||||||
dynamic_addresses = []
|
dynamic_addresses = []
|
||||||
for addr in net.get("addr_info", []):
|
for info in net.get("addr_info", []):
|
||||||
# no link-local ipv4/ipv6
|
# no link-local ipv4/ipv6
|
||||||
if addr.get("scope") == "link":
|
if info.get("scope") == "link":
|
||||||
continue
|
continue
|
||||||
if addr.get("dynamic", False):
|
if (preferred_life_time := info.get("preferred_life_time")) is None:
|
||||||
dynamic_addresses.append(addr["local"])
|
continue
|
||||||
|
if (valid_life_time := info.get("valid_life_time")) is None:
|
||||||
|
continue
|
||||||
|
if (prefixlen := info.get("prefixlen")) is None:
|
||||||
|
continue
|
||||||
|
if (family := info.get("family")) not in ["inet", "inet6"]:
|
||||||
|
continue
|
||||||
|
if (local := info.get("local")) is None:
|
||||||
|
continue
|
||||||
|
if (dynamic := info.get("dynamic", False)) is None:
|
||||||
|
continue
|
||||||
|
|
||||||
|
address = Address(
|
||||||
|
address=local,
|
||||||
|
family=family,
|
||||||
|
prefixlen=prefixlen,
|
||||||
|
preferred_life_time=preferred_life_time,
|
||||||
|
valid_life_time=valid_life_time,
|
||||||
|
)
|
||||||
|
|
||||||
|
if dynamic:
|
||||||
|
dynamic_addresses.append(address)
|
||||||
else:
|
else:
|
||||||
static_addresses.append(addr)
|
static_addresses.append(address)
|
||||||
interfaces.append(
|
interfaces.append(
|
||||||
Interface(
|
Interface(
|
||||||
name=net.get("ifname", mac_address.replace(":", "-")),
|
name=net.get("ifname", mac_address.replace(":", "-")),
|
||||||
|
@ -58,6 +88,19 @@ def filter_routes(routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||||
return filtered
|
return filtered
|
||||||
|
|
||||||
|
|
||||||
|
def find_most_recent_v4_lease(addresses: list[Address]) -> Address | None:
|
||||||
|
most_recent_address = None
|
||||||
|
most_recent_lifetime = -1
|
||||||
|
for addr in addresses:
|
||||||
|
if addr.family == "inet6":
|
||||||
|
continue
|
||||||
|
lifetime = max(addr.preferred_life_time, addr.valid_life_time)
|
||||||
|
if lifetime > most_recent_lifetime:
|
||||||
|
most_recent_lifetime = lifetime
|
||||||
|
most_recent_address = addr
|
||||||
|
return most_recent_address
|
||||||
|
|
||||||
|
|
||||||
def generate_routes(
|
def generate_routes(
|
||||||
interface: Interface, routes: list[dict[str, Any]]
|
interface: Interface, routes: list[dict[str, Any]]
|
||||||
) -> Iterator[str]:
|
) -> Iterator[str]:
|
||||||
|
@ -83,7 +126,6 @@ def generate_routes(
|
||||||
yield f"Gateway = {gateway}"
|
yield f"Gateway = {gateway}"
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def generate_networkd_units(
|
def generate_networkd_units(
|
||||||
interfaces: list[Interface], routes: list[dict[str, Any]], directory: Path
|
interfaces: list[Interface], routes: list[dict[str, Any]], directory: Path
|
||||||
) -> None:
|
) -> None:
|
||||||
|
@ -103,16 +145,24 @@ LLDP = yes
|
||||||
# ipv6 router advertisements
|
# ipv6 router advertisements
|
||||||
IPv6AcceptRA = yes
|
IPv6AcceptRA = yes
|
||||||
# allows us to ping "nixos.local"
|
# allows us to ping "nixos.local"
|
||||||
MulticastDNS = yes
|
MulticastDNS = yes"""
|
||||||
"""
|
|
||||||
]
|
]
|
||||||
unit_sections.extend(
|
unit_sections.extend(
|
||||||
f"Address = {addr['local']}/{addr['prefixlen']}"
|
f"Address = {addr.address}/{addr.prefixlen}"
|
||||||
for addr in interface.static_addresses
|
for addr in interface.static_addresses
|
||||||
)
|
)
|
||||||
unit_sections.extend(generate_routes(interface, routes))
|
unit_sections.extend(generate_routes(interface, routes))
|
||||||
|
most_recent_v4_lease = find_most_recent_v4_lease(interface.dynamic_addresses)
|
||||||
|
if most_recent_v4_lease:
|
||||||
|
unit_sections.append("[DHCPv4]")
|
||||||
|
unit_sections.append(f"RequestAddress = {most_recent_v4_lease.address}")
|
||||||
|
|
||||||
(directory / f"00-{interface.name}.network").write_text("\n".join(unit_sections))
|
# trailing newline at the end
|
||||||
|
unit_sections.append("")
|
||||||
|
|
||||||
|
(directory / f"00-{interface.name}.network").write_text(
|
||||||
|
"\n".join(unit_sections)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
|
|
Loading…
Reference in a new issue