Compare commits
4 Commits
a0db5f797f
...
8378591520
| Author | SHA1 | Date | |
|---|---|---|---|
| 8378591520 | |||
| 75c1164ad7 | |||
| 6392131577 | |||
| 499e61346d |
10
.claude/settings.json
Normal file
10
.claude/settings.json
Normal file
@@ -0,0 +1,10 @@
|
|||||||
|
{
|
||||||
|
"permissions": {
|
||||||
|
"ask": [
|
||||||
|
"Edit(CHANGELOG.md)",
|
||||||
|
"Write(CHANGELOG.md)",
|
||||||
|
"Edit(pyproject.toml)",
|
||||||
|
"Write(pyproject.toml)"
|
||||||
|
]
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,3 +1,12 @@
|
|||||||
|
# Changelog
|
||||||
|
|
||||||
|
## [1.1.1] - 2026-09-08
|
||||||
|
|
||||||
|
* Fixed interface detection with dead /sys/class/net symlinks (sysbox)
|
||||||
|
* Added netlink fallback for reading MTU when sysfs is unreadable
|
||||||
|
* Fixed Docker bridge detection in the same environments
|
||||||
|
* Fixed lint against the widened ruff default rule set
|
||||||
|
|
||||||
## [1.1.0] - 2026-01-23
|
## [1.1.0] - 2026-01-23
|
||||||
|
|
||||||
* * Added persistent MTU configuration via systemd (install & uninstall)
|
* * Added persistent MTU configuration via systemd (install & uninstall)
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "automtu"
|
name = "automtu"
|
||||||
version = "1.1.0"
|
version = "1.1.1"
|
||||||
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
|
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
|
|||||||
@@ -2,8 +2,8 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import statistics
|
import statistics
|
||||||
import sys
|
import sys
|
||||||
|
from collections.abc import Iterable
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Iterable, Optional
|
|
||||||
|
|
||||||
from .docker import detect_docker_ifaces
|
from .docker import detect_docker_ifaces
|
||||||
from .net import (
|
from .net import (
|
||||||
@@ -28,7 +28,7 @@ class Result:
|
|||||||
wg_mtu: int
|
wg_mtu: int
|
||||||
|
|
||||||
|
|
||||||
def _split_targets(items: Optional[list[str]]) -> list[str]:
|
def _split_targets(items: list[str] | None) -> list[str]:
|
||||||
raw: list[str] = []
|
raw: list[str] = []
|
||||||
for item in items or []:
|
for item in items or []:
|
||||||
raw.extend([x.strip() for x in item.split(",") if x.strip()])
|
raw.extend([x.strip() for x in item.split(",") if x.strip()])
|
||||||
@@ -119,7 +119,11 @@ def run_automtu(args) -> int:
|
|||||||
set_iface_mtu(egress, args.force_egress_mtu, args.dry_run)
|
set_iface_mtu(egress, args.force_egress_mtu, args.dry_run)
|
||||||
base_mtu = int(args.force_egress_mtu)
|
base_mtu = int(args.force_egress_mtu)
|
||||||
else:
|
else:
|
||||||
base_mtu = int(read_iface_mtu(egress))
|
try:
|
||||||
|
base_mtu = int(read_iface_mtu(egress))
|
||||||
|
except RuntimeError as exc:
|
||||||
|
print(f"[automtu][ERROR] {exc}", file=sys.stderr)
|
||||||
|
return 3
|
||||||
log(f"[automtu] Egress base MTU: {base_mtu}")
|
log(f"[automtu] Egress base MTU: {base_mtu}")
|
||||||
|
|
||||||
# Targets (explicit + optional WG auto targets)
|
# Targets (explicit + optional WG auto targets)
|
||||||
@@ -140,8 +144,8 @@ def run_automtu(args) -> int:
|
|||||||
|
|
||||||
# PMTU probing
|
# PMTU probing
|
||||||
effective_mtu = base_mtu
|
effective_mtu = base_mtu
|
||||||
probe_results: dict[str, Optional[int]] = {}
|
probe_results: dict[str, int | None] = {}
|
||||||
chosen_pmtu: Optional[int] = None
|
chosen_pmtu: int | None = None
|
||||||
|
|
||||||
if targets:
|
if targets:
|
||||||
log(
|
log(
|
||||||
@@ -186,7 +190,7 @@ def run_automtu(args) -> int:
|
|||||||
f"[automtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})"
|
f"[automtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})"
|
||||||
)
|
)
|
||||||
|
|
||||||
wg_mtu_set: Optional[int] = None
|
wg_mtu_set: int | None = None
|
||||||
wg_mtu_clamped = False
|
wg_mtu_clamped = False
|
||||||
|
|
||||||
if args.set_wg_mtu is not None:
|
if args.set_wg_mtu is not None:
|
||||||
|
|||||||
@@ -1,15 +1,13 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import re
|
import re
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
from .net import iface_exists, list_ifaces
|
from .net import iface_exists, list_ifaces
|
||||||
|
|
||||||
|
|
||||||
_BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE)
|
_BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE)
|
||||||
|
|
||||||
|
|
||||||
def _split_items(items: Optional[list[str]]) -> list[str]:
|
def _split_items(items: list[str] | None) -> list[str]:
|
||||||
raw: list[str] = []
|
raw: list[str] = []
|
||||||
for item in items or []:
|
for item in items or []:
|
||||||
raw.extend([x.strip() for x in item.split(",") if x.strip()])
|
raw.extend([x.strip() for x in item.split(",") if x.strip()])
|
||||||
@@ -18,7 +16,7 @@ def _split_items(items: Optional[list[str]]) -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
def detect_docker_ifaces(
|
def detect_docker_ifaces(
|
||||||
docker_if_args: Optional[list[str]], *, include_user_bridges: bool
|
docker_if_args: list[str] | None, *, include_user_bridges: bool
|
||||||
) -> list[str]:
|
) -> list[str]:
|
||||||
"""
|
"""
|
||||||
Determine Docker-related interfaces to apply MTU to.
|
Determine Docker-related interfaces to apply MTU to.
|
||||||
|
|||||||
@@ -5,7 +5,6 @@ import pathlib
|
|||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
|
|
||||||
def _run(cmd: list[str]) -> str:
|
def _run(cmd: list[str]) -> str:
|
||||||
@@ -14,27 +13,55 @@ def _run(cmd: list[str]) -> str:
|
|||||||
).stdout.strip()
|
).stdout.strip()
|
||||||
|
|
||||||
|
|
||||||
|
SYSFS_NET = pathlib.Path("/sys/class/net")
|
||||||
|
|
||||||
|
|
||||||
|
def _ip_link_show(iface: str) -> str:
|
||||||
|
return _run(["ip", "link", "show", "dev", iface])
|
||||||
|
|
||||||
|
|
||||||
def iface_exists(iface: str) -> bool:
|
def iface_exists(iface: str) -> bool:
|
||||||
return pathlib.Path(f"/sys/class/net/{iface}").exists()
|
"""
|
||||||
|
True if the interface exists. Does not follow the /sys/class/net symlink;
|
||||||
|
falls back to netlink when sysfs is unavailable.
|
||||||
|
"""
|
||||||
|
if os.path.lexists(SYSFS_NET / iface):
|
||||||
|
return True
|
||||||
|
return bool(_ip_link_show(iface))
|
||||||
|
|
||||||
|
|
||||||
def list_ifaces() -> list[str]:
|
def list_ifaces() -> list[str]:
|
||||||
"""
|
"""
|
||||||
Return a sorted list of all network interfaces visible under /sys/class/net.
|
Return a sorted list of all network interfaces, netlink as fallback.
|
||||||
|
Plain files in /sys/class/net (e.g. bonding_masters) are not interfaces.
|
||||||
"""
|
"""
|
||||||
base = pathlib.Path("/sys/class/net")
|
try:
|
||||||
if not base.exists():
|
return sorted(
|
||||||
return []
|
p.name for p in SYSFS_NET.iterdir() if p.is_symlink() or p.is_dir()
|
||||||
names: list[str] = []
|
)
|
||||||
for p in base.iterdir():
|
except OSError:
|
||||||
if p.is_dir():
|
out = _run(["ip", "-o", "link", "show"])
|
||||||
names.append(p.name)
|
return sorted(re.findall(r"^\d+:\s+([^:@\s]+)", out, flags=re.MULTILINE))
|
||||||
names.sort()
|
|
||||||
return names
|
|
||||||
|
|
||||||
|
|
||||||
def read_iface_mtu(iface: str) -> int:
|
def read_iface_mtu(iface: str) -> int:
|
||||||
return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip())
|
"""
|
||||||
|
Read the interface MTU from sysfs, falling back to netlink.
|
||||||
|
Raises RuntimeError if neither source reports an MTU.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return int((SYSFS_NET / iface / "mtu").read_text().strip())
|
||||||
|
except (OSError, ValueError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
m = re.search(r"\bmtu\s+(\d+)\b", _ip_link_show(iface))
|
||||||
|
if not m:
|
||||||
|
raise RuntimeError(
|
||||||
|
f"Could not read MTU of {iface}: "
|
||||||
|
f"{SYSFS_NET / iface / 'mtu'} is unreadable and "
|
||||||
|
f"'ip link show dev {iface}' reported no MTU."
|
||||||
|
)
|
||||||
|
return int(m.group(1))
|
||||||
|
|
||||||
|
|
||||||
def set_iface_mtu(iface: str, mtu: int, dry: bool) -> None:
|
def set_iface_mtu(iface: str, mtu: int, dry: bool) -> None:
|
||||||
@@ -53,7 +80,7 @@ def require_root(*, dry: bool, needs_root: bool) -> None:
|
|||||||
raise SystemExit(1)
|
raise SystemExit(1)
|
||||||
|
|
||||||
|
|
||||||
def detect_egress_iface(ignore_vpn: bool = True) -> Optional[str]:
|
def detect_egress_iface(ignore_vpn: bool = True) -> str | None:
|
||||||
devs: list[str] = []
|
devs: list[str] = []
|
||||||
for cmd in (
|
for cmd in (
|
||||||
["ip", "-4", "route", "show", "default"],
|
["ip", "-4", "route", "show", "default"],
|
||||||
|
|||||||
@@ -4,19 +4,18 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import sys
|
import sys
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
class OutputMode:
|
class OutputMode:
|
||||||
print_mtu: Optional[str] # "egress" | "effective" | "wg" | None
|
print_mtu: str | None # "egress" | "effective" | "wg" | None
|
||||||
print_json: bool
|
print_json: bool
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def machine(self) -> bool:
|
def machine(self) -> bool:
|
||||||
return bool(self.print_mtu or self.print_json)
|
return bool(self.print_mtu or self.print_json)
|
||||||
|
|
||||||
def validate(self) -> Optional[str]:
|
def validate(self) -> str | None:
|
||||||
if self.print_mtu and self.print_json:
|
if self.print_mtu and self.print_json:
|
||||||
return "--print-mtu and --print-json are mutually exclusive."
|
return "--print-mtu and --print-json are mutually exclusive."
|
||||||
return None
|
return None
|
||||||
@@ -67,24 +66,24 @@ def emit_json(
|
|||||||
egress_iface: str,
|
egress_iface: str,
|
||||||
base_mtu: int,
|
base_mtu: int,
|
||||||
effective_mtu: int,
|
effective_mtu: int,
|
||||||
egress_forced_mtu: Optional[int],
|
egress_forced_mtu: int | None,
|
||||||
egress_applied: bool,
|
egress_applied: bool,
|
||||||
pmtu_targets: list[str],
|
pmtu_targets: list[str],
|
||||||
pmtu_auto_targets_added: list[str],
|
pmtu_auto_targets_added: list[str],
|
||||||
pmtu_policy: str,
|
pmtu_policy: str,
|
||||||
pmtu_chosen: Optional[int],
|
pmtu_chosen: int | None,
|
||||||
pmtu_results: dict[str, Optional[int]],
|
pmtu_results: dict[str, int | None],
|
||||||
wg_iface: str,
|
wg_iface: str,
|
||||||
wg_mtu: int,
|
wg_mtu: int,
|
||||||
wg_overhead: int,
|
wg_overhead: int,
|
||||||
wg_min: int,
|
wg_min: int,
|
||||||
wg_set_mtu: Optional[int],
|
wg_set_mtu: int | None,
|
||||||
wg_clamped: bool,
|
wg_clamped: bool,
|
||||||
wg_present: bool,
|
wg_present: bool,
|
||||||
wg_active: bool,
|
wg_active: bool,
|
||||||
wg_applied: bool,
|
wg_applied: bool,
|
||||||
docker_ifaces: Optional[list[str]] = None,
|
docker_ifaces: list[str] | None = None,
|
||||||
docker_applied: Optional[list[str]] = None,
|
docker_applied: list[str] | None = None,
|
||||||
dry_run: bool,
|
dry_run: bool,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""
|
"""
|
||||||
|
|||||||
@@ -4,14 +4,12 @@ import shlex
|
|||||||
import shutil
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import List
|
|
||||||
|
|
||||||
|
|
||||||
_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu.service")
|
_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu.service")
|
||||||
_DOCKER_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu-docker.service")
|
_DOCKER_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu-docker.service")
|
||||||
|
|
||||||
|
|
||||||
def _strip_persist_args(argv: List[str]) -> List[str]:
|
def _strip_persist_args(argv: list[str]) -> list[str]:
|
||||||
"""
|
"""
|
||||||
Remove persistence-only arguments from argv:
|
Remove persistence-only arguments from argv:
|
||||||
- --persist systemd|docker
|
- --persist systemd|docker
|
||||||
@@ -51,7 +49,7 @@ def _resolve_exec(argv0: str) -> str:
|
|||||||
return argv0
|
return argv0
|
||||||
|
|
||||||
|
|
||||||
def _needs_docker_ordering(filtered_argv: List[str]) -> bool:
|
def _needs_docker_ordering(filtered_argv: list[str]) -> bool:
|
||||||
"""
|
"""
|
||||||
Heuristic: If we apply docker MTU (directly or via --apply-all), order after docker.service.
|
Heuristic: If we apply docker MTU (directly or via --apply-all), order after docker.service.
|
||||||
"""
|
"""
|
||||||
@@ -115,7 +113,7 @@ def _uninstall_unit(unit_path: Path, *, dry: bool) -> None:
|
|||||||
print(f"[automtu] Uninstalled systemd service: {unit_path.name}")
|
print(f"[automtu] Uninstalled systemd service: {unit_path.name}")
|
||||||
|
|
||||||
|
|
||||||
def persist_systemd(argv: List[str], *, dry: bool) -> None:
|
def persist_systemd(argv: list[str], *, dry: bool) -> None:
|
||||||
"""
|
"""
|
||||||
Install a systemd oneshot service that re-runs automtu with the same arguments.
|
Install a systemd oneshot service that re-runs automtu with the same arguments.
|
||||||
Adds docker ordering automatically if docker MTU is applied.
|
Adds docker ordering automatically if docker MTU is applied.
|
||||||
@@ -142,7 +140,7 @@ def uninstall_systemd(*, dry: bool) -> None:
|
|||||||
_uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry)
|
_uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry)
|
||||||
|
|
||||||
|
|
||||||
def persist_docker(argv: List[str], *, dry: bool) -> None:
|
def persist_docker(argv: list[str], *, dry: bool) -> None:
|
||||||
"""
|
"""
|
||||||
Docker-focused persistence backend:
|
Docker-focused persistence backend:
|
||||||
always orders after docker.service (even if args don't include docker flags),
|
always orders after docker.service (even if args don't include docker flags),
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import ipaddress
|
import ipaddress
|
||||||
import subprocess
|
import subprocess
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
|
|
||||||
def _is_ipv6(target: str) -> bool:
|
def _is_ipv6(target: str) -> bool:
|
||||||
@@ -14,7 +13,7 @@ def _is_ipv6(target: str) -> bool:
|
|||||||
|
|
||||||
def _rc(cmd: list[str]) -> int:
|
def _rc(cmd: list[str]) -> int:
|
||||||
return subprocess.run(
|
return subprocess.run(
|
||||||
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
|
cmd, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
|
||||||
).returncode
|
).returncode
|
||||||
|
|
||||||
|
|
||||||
@@ -28,7 +27,7 @@ def _ping_ok(payload: int, target: str, timeout_s: float) -> bool:
|
|||||||
"-s",
|
"-s",
|
||||||
str(payload),
|
str(payload),
|
||||||
"-W",
|
"-W",
|
||||||
str(max(1, int(round(timeout_s)))),
|
str(max(1, round(timeout_s))),
|
||||||
]
|
]
|
||||||
if _is_ipv6(target):
|
if _is_ipv6(target):
|
||||||
cmd.insert(1, "-6")
|
cmd.insert(1, "-6")
|
||||||
@@ -37,7 +36,7 @@ def _ping_ok(payload: int, target: str, timeout_s: float) -> bool:
|
|||||||
|
|
||||||
def probe_pmtu(
|
def probe_pmtu(
|
||||||
target: str, lo_payload: int = 1200, hi_payload: int = 1472, timeout: float = 1.0
|
target: str, lo_payload: int = 1200, hi_payload: int = 1472, timeout: float = 1.0
|
||||||
) -> Optional[int]:
|
) -> int | None:
|
||||||
hdr = 48 if _is_ipv6(target) else 28
|
hdr = 48 if _is_ipv6(target) else 28
|
||||||
|
|
||||||
if not _ping_ok(lo_payload, target, timeout):
|
if not _ping_ok(lo_payload, target, timeout):
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
from typing import List
|
|
||||||
|
|
||||||
from .net import iface_exists
|
from .net import iface_exists
|
||||||
|
|
||||||
@@ -15,7 +14,7 @@ def _run(cmd: list[str]) -> str:
|
|||||||
|
|
||||||
def _rc(cmd: list[str]) -> int:
|
def _rc(cmd: list[str]) -> int:
|
||||||
return subprocess.run(
|
return subprocess.run(
|
||||||
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
|
cmd, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
|
||||||
).returncode
|
).returncode
|
||||||
|
|
||||||
|
|
||||||
@@ -23,7 +22,7 @@ def wg_is_active(wg_if: str) -> bool:
|
|||||||
return iface_exists(wg_if) and _rc(["wg", "show", wg_if]) == 0
|
return iface_exists(wg_if) and _rc(["wg", "show", wg_if]) == 0
|
||||||
|
|
||||||
|
|
||||||
def wg_peer_endpoints(wg_if: str) -> List[str]:
|
def wg_peer_endpoints(wg_if: str) -> list[str]:
|
||||||
targets: list[str] = []
|
targets: list[str] = []
|
||||||
|
|
||||||
out = _run(["wg", "show", wg_if, "endpoints"])
|
out = _run(["wg", "show", wg_if, "endpoints"])
|
||||||
|
|||||||
@@ -1,12 +1,46 @@
|
|||||||
import io
|
import io
|
||||||
|
import pathlib
|
||||||
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
from contextlib import redirect_stdout
|
from contextlib import redirect_stdout
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from automtu import net
|
||||||
from automtu.core import run_automtu
|
from automtu.core import run_automtu
|
||||||
|
|
||||||
|
|
||||||
|
def _args(**over) -> SimpleNamespace:
|
||||||
|
base = {
|
||||||
|
"dry_run": True,
|
||||||
|
"egress_if": None,
|
||||||
|
"prefer_wg_egress": False,
|
||||||
|
"force_egress_mtu": None,
|
||||||
|
"pmtu_target": None,
|
||||||
|
"auto_pmtu_from_wg": False,
|
||||||
|
"pmtu_min_payload": 1200,
|
||||||
|
"pmtu_max_payload": 1472,
|
||||||
|
"pmtu_timeout": 1.0,
|
||||||
|
"pmtu_policy": "min",
|
||||||
|
"apply_egress_mtu": False,
|
||||||
|
"apply_wg_mtu": False,
|
||||||
|
"apply_docker_mtu": False,
|
||||||
|
"apply_all": False,
|
||||||
|
"docker_if": None,
|
||||||
|
"docker_no_user_bridges": False,
|
||||||
|
"wg_if": "wg0",
|
||||||
|
"wg_overhead": 80,
|
||||||
|
"wg_min": 1280,
|
||||||
|
"set_wg_mtu": None,
|
||||||
|
"persist": None,
|
||||||
|
"uninstall": False,
|
||||||
|
"print_mtu": None,
|
||||||
|
"print_json": False,
|
||||||
|
}
|
||||||
|
base.update(over)
|
||||||
|
return SimpleNamespace(**base)
|
||||||
|
|
||||||
|
|
||||||
class TestCore(unittest.TestCase):
|
class TestCore(unittest.TestCase):
|
||||||
def test_run_automtu_happy_path_all_mocked(self) -> None:
|
def test_run_automtu_happy_path_all_mocked(self) -> None:
|
||||||
args = SimpleNamespace(
|
args = SimpleNamespace(
|
||||||
@@ -164,5 +198,53 @@ class TestCore(unittest.TestCase):
|
|||||||
mock_set.assert_not_called()
|
mock_set.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
class TestCoreInSysboxContainer(unittest.TestCase):
|
||||||
|
"""Container with dead /sys/class/net symlinks: routing and ip link work."""
|
||||||
|
|
||||||
|
def setUp(self) -> None:
|
||||||
|
tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.addCleanup(tmp.cleanup)
|
||||||
|
root = pathlib.Path(tmp.name)
|
||||||
|
self.netdir = root / "class" / "net"
|
||||||
|
self.netdir.mkdir(parents=True)
|
||||||
|
for name in ("eth0", "lo"):
|
||||||
|
(self.netdir / name).symlink_to(root / "devices" / "virtual" / "net" / name)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _fake_run(cmd: list[str]) -> str:
|
||||||
|
if cmd[:5] == ["ip", "-4", "route", "show", "default"]:
|
||||||
|
return "default via 172.28.0.1 dev eth0"
|
||||||
|
if cmd[:3] == ["ip", "link", "show"] and cmd[-1] == "eth0":
|
||||||
|
return (
|
||||||
|
"2: eth0@if5: <BROADCAST,MULTICAST,UP> mtu 1450 qdisc noqueue state UP"
|
||||||
|
)
|
||||||
|
return ""
|
||||||
|
|
||||||
|
def test_print_mtu_effective_yields_number_and_rc0(self) -> None:
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", self.netdir),
|
||||||
|
patch.object(net, "_run", side_effect=self._fake_run),
|
||||||
|
patch("automtu.core.probe_pmtu", return_value=1400),
|
||||||
|
):
|
||||||
|
buf = io.StringIO()
|
||||||
|
with redirect_stdout(buf):
|
||||||
|
rc = run_automtu(_args(pmtu_target=["1.1.1.1"], print_mtu="effective"))
|
||||||
|
|
||||||
|
self.assertEqual(rc, 0)
|
||||||
|
self.assertEqual(int(buf.getvalue().strip()), 1400)
|
||||||
|
|
||||||
|
def test_explicit_egress_if_is_accepted(self) -> None:
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", self.netdir),
|
||||||
|
patch.object(net, "_run", side_effect=self._fake_run),
|
||||||
|
):
|
||||||
|
buf = io.StringIO()
|
||||||
|
with redirect_stdout(buf):
|
||||||
|
rc = run_automtu(_args(egress_if="eth0", print_mtu="egress"))
|
||||||
|
|
||||||
|
self.assertEqual(rc, 0)
|
||||||
|
self.assertEqual(int(buf.getvalue().strip()), 1450)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main(verbosity=2)
|
unittest.main(verbosity=2)
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.docker as docker
|
from automtu import docker
|
||||||
|
|
||||||
|
|
||||||
class TestDocker(unittest.TestCase):
|
class TestDocker(unittest.TestCase):
|
||||||
|
|||||||
@@ -1,8 +1,120 @@
|
|||||||
|
import pathlib
|
||||||
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
from pathlib import Path
|
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.net as net
|
from automtu import net
|
||||||
|
|
||||||
|
IP_LINK_ETH0 = (
|
||||||
|
"2: eth0@if5: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1420 qdisc noqueue state UP "
|
||||||
|
"mode DEFAULT group default\n link/ether 02:42:ac:1c:00:02 brd ff:ff:ff:ff:ff:ff"
|
||||||
|
)
|
||||||
|
IP_LINK_ALL = (
|
||||||
|
"1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN\n"
|
||||||
|
"2: eth0@if5: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1420 qdisc noqueue state UP"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _sysbox_sysfs(root: pathlib.Path, ifaces=("eth0", "lo")) -> pathlib.Path:
|
||||||
|
"""/sys replacement as seen inside a sysbox container: dead symlinks."""
|
||||||
|
netdir = root / "class" / "net"
|
||||||
|
netdir.mkdir(parents=True)
|
||||||
|
for name in ifaces:
|
||||||
|
(netdir / name).symlink_to(root / "devices" / "virtual" / "net" / name)
|
||||||
|
return netdir
|
||||||
|
|
||||||
|
|
||||||
|
class SysboxSysfsBase(unittest.TestCase):
|
||||||
|
def setUp(self) -> None:
|
||||||
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
|
self.root = pathlib.Path(self._tmp.name)
|
||||||
|
self.addCleanup(self._tmp.cleanup)
|
||||||
|
|
||||||
|
|
||||||
|
class TestDeadSymlinkSysfs(SysboxSysfsBase):
|
||||||
|
def test_iface_exists_true_for_dead_symlink_without_netlink(self) -> None:
|
||||||
|
netdir = _sysbox_sysfs(self.root)
|
||||||
|
self.assertFalse((netdir / "eth0").exists())
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", return_value=""),
|
||||||
|
):
|
||||||
|
self.assertTrue(net.iface_exists("eth0"))
|
||||||
|
self.assertFalse(net.iface_exists("eth9"))
|
||||||
|
|
||||||
|
def test_iface_exists_falls_back_to_netlink_without_sysfs(self) -> None:
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", self.root / "absent" / "net"),
|
||||||
|
patch.object(net, "_run", return_value=IP_LINK_ETH0),
|
||||||
|
):
|
||||||
|
self.assertTrue(net.iface_exists("eth0"))
|
||||||
|
|
||||||
|
def test_detect_egress_iface_accepts_dead_symlink_iface(self) -> None:
|
||||||
|
netdir = _sysbox_sysfs(self.root)
|
||||||
|
|
||||||
|
def fake_run(cmd: list[str]) -> str:
|
||||||
|
if cmd[:5] == ["ip", "-4", "route", "show", "default"]:
|
||||||
|
return "default via 172.28.0.1 dev eth0"
|
||||||
|
return ""
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", side_effect=fake_run),
|
||||||
|
):
|
||||||
|
self.assertEqual(net.detect_egress_iface(), "eth0")
|
||||||
|
|
||||||
|
def test_list_ifaces_keeps_dead_symlinks_and_skips_plain_files(self) -> None:
|
||||||
|
netdir = _sysbox_sysfs(self.root, ("eth0", "lo", "br-abc123"))
|
||||||
|
(netdir / "bonding_masters").write_text("")
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", return_value=""),
|
||||||
|
):
|
||||||
|
self.assertEqual(net.list_ifaces(), ["br-abc123", "eth0", "lo"])
|
||||||
|
|
||||||
|
def test_list_ifaces_falls_back_to_netlink_without_sysfs(self) -> None:
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", self.root / "absent" / "net"),
|
||||||
|
patch.object(net, "_run", return_value=IP_LINK_ALL),
|
||||||
|
):
|
||||||
|
self.assertEqual(net.list_ifaces(), ["eth0", "lo"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestReadIfaceMtu(SysboxSysfsBase):
|
||||||
|
def test_prefers_sysfs_over_netlink(self) -> None:
|
||||||
|
netdir = self.root / "class" / "net"
|
||||||
|
(netdir / "eth0").mkdir(parents=True)
|
||||||
|
(netdir / "eth0" / "mtu").write_text("1500\n")
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", return_value=IP_LINK_ETH0) as run,
|
||||||
|
):
|
||||||
|
self.assertEqual(net.read_iface_mtu("eth0"), 1500)
|
||||||
|
run.assert_not_called()
|
||||||
|
|
||||||
|
def test_netlink_fallback_on_dead_symlink(self) -> None:
|
||||||
|
netdir = _sysbox_sysfs(self.root)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", return_value=IP_LINK_ETH0),
|
||||||
|
):
|
||||||
|
self.assertEqual(net.read_iface_mtu("eth0"), 1420)
|
||||||
|
|
||||||
|
def test_raises_when_neither_sysfs_nor_netlink_answers(self) -> None:
|
||||||
|
netdir = _sysbox_sysfs(self.root)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(net, "SYSFS_NET", netdir),
|
||||||
|
patch.object(net, "_run", return_value=""),
|
||||||
|
self.assertRaises(RuntimeError) as ctx,
|
||||||
|
):
|
||||||
|
net.read_iface_mtu("eth0")
|
||||||
|
|
||||||
|
self.assertIn("eth0", str(ctx.exception))
|
||||||
|
|
||||||
|
|
||||||
class TestNet(unittest.TestCase):
|
class TestNet(unittest.TestCase):
|
||||||
@@ -46,18 +158,10 @@ class TestNet(unittest.TestCase):
|
|||||||
self.assertFalse(net.default_route_uses_iface("wg0"))
|
self.assertFalse(net.default_route_uses_iface("wg0"))
|
||||||
|
|
||||||
def test_list_ifaces_returns_sorted_names(self) -> None:
|
def test_list_ifaces_returns_sorted_names(self) -> None:
|
||||||
fake = [
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
Path("/sys/class/net/eth0"),
|
netdir = _sysbox_sysfs(pathlib.Path(tmp), ("wg0", "eth0", "lo"))
|
||||||
Path("/sys/class/net/lo"),
|
with patch.object(net, "SYSFS_NET", netdir):
|
||||||
Path("/sys/class/net/wg0"),
|
self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])
|
||||||
]
|
|
||||||
|
|
||||||
with (
|
|
||||||
patch("automtu.net.pathlib.Path.exists", return_value=True),
|
|
||||||
patch("automtu.net.pathlib.Path.iterdir", return_value=fake),
|
|
||||||
patch.object(Path, "is_dir", return_value=True),
|
|
||||||
):
|
|
||||||
self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
import io
|
import io
|
||||||
import json
|
import json
|
||||||
import unittest
|
import unittest
|
||||||
from contextlib import redirect_stdout, redirect_stderr
|
from contextlib import redirect_stderr, redirect_stdout
|
||||||
|
|
||||||
from automtu.output import OutputMode, emit_json, emit_single_number, Logger
|
from automtu.output import Logger, OutputMode, emit_json, emit_single_number
|
||||||
|
|
||||||
|
|
||||||
class TestOutput(unittest.TestCase):
|
class TestOutput(unittest.TestCase):
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ from contextlib import redirect_stdout
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.persist as persist
|
from automtu import persist
|
||||||
|
|
||||||
|
|
||||||
class TestPersist(unittest.TestCase):
|
class TestPersist(unittest.TestCase):
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.pmtu as pmtu
|
from automtu import pmtu
|
||||||
|
|
||||||
|
|
||||||
class TestPmtu(unittest.TestCase):
|
class TestPmtu(unittest.TestCase):
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.wg as wg
|
from automtu import wg
|
||||||
|
|
||||||
|
|
||||||
class TestWg(unittest.TestCase):
|
class TestWg(unittest.TestCase):
|
||||||
|
|||||||
Reference in New Issue
Block a user