Add --apply-egress-mtu option and corresponding unittests to allow automatic MTU adjustment on egress interface before setting WireGuard MTU

See: https://chatgpt.com/share/68efc179-1a10-800f-9656-1e8731b40546 (German discussion)
This commit is contained in:
2025-10-16 10:27:53 +02:00
parent e5d553f783
commit c15127b6ab
4 changed files with 111 additions and 0 deletions

Binary file not shown.

Binary file not shown.

11
main.py
View File

@@ -207,6 +207,9 @@ def main():
ap.add_argument("--pmtu-max-payload", type=int, default=1472, help="Upper bound payload for PMTU search (default: 1472 ~ 1500-28).") ap.add_argument("--pmtu-max-payload", type=int, default=1472, help="Upper bound payload for PMTU search (default: 1472 ~ 1500-28).")
ap.add_argument("--pmtu-policy", choices=["min", "median", "max"], default="min", ap.add_argument("--pmtu-policy", choices=["min", "median", "max"], default="min",
help="How to choose effective PMTU across multiple targets (default: min).") help="How to choose effective PMTU across multiple targets (default: min).")
ap.add_argument("--apply-egress-mtu", action="store_true",
help="Apply the effective Path MTU to the detected egress interface (e.g. eth0).")
ap.add_argument("--dry-run", action="store_true", help="Show actions without applying changes.") ap.add_argument("--dry-run", action="store_true", help="Show actions without applying changes.")
# NEW: force a specific WireGuard MTU (overrides computed value) # NEW: force a specific WireGuard MTU (overrides computed value)
ap.add_argument("--set-wg-mtu", type=int, help="Force a specific MTU to apply on the WireGuard interface (overrides computed value).") ap.add_argument("--set-wg-mtu", type=int, help="Force a specific MTU to apply on the WireGuard interface (overrides computed value).")
@@ -289,6 +292,14 @@ def main():
else: else:
print("[wg-mtu] WARNING: All PMTU probes failed. Falling back to egress MTU.") print("[wg-mtu] WARNING: All PMTU probes failed. Falling back to egress MTU.")
# Optionally apply the effective MTU to the egress interface
if args.apply_egress_mtu:
if egress == args.wg_if:
print(f"[wg-mtu] INFO: Skipping egress MTU apply because egress == {args.wg_if}.")
else:
print(f"[wg-mtu] Applying effective MTU {effective_mtu} to egress {egress}")
set_mtu(egress, effective_mtu, args.dry_run)
# Compute WG MTU # Compute WG MTU
wg_mtu = max(args.wg_min, effective_mtu - args.wg_overhead) wg_mtu = max(args.wg_min, effective_mtu - args.wg_overhead)
print(f"[wg-mtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})") print(f"[wg-mtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})")

100
test.py
View File

@@ -182,6 +182,106 @@ class TestWgMtuAutoExtended(unittest.TestCase):
self.assertIn("Forcing WireGuard MTU (override): 1280", s) self.assertIn("Forcing WireGuard MTU (override): 1280", s)
mock_set_mtu.assert_any_call("wg0", 1280, True) mock_set_mtu.assert_any_call("wg0", 1280, True)
# ---------- --apply-egress-mtu: setzt egress vor wg ----------
@patch("main.set_mtu")
@patch("main.probe_pmtu", return_value=1452)
@patch("main.read_mtu", return_value=1500)
@patch("main.exists_iface", return_value=True)
@patch("main.get_default_ifaces", return_value=["eth0"])
@patch("main.require_root", return_value=None)
def test_apply_egress_mtu_sets_egress_then_wg(
self, _req_root, _get_def, _exists, _read_mtu, _probe_pmtu, mock_set_mtu
):
"""
--apply-egress-mtu setzt egress=eth0 auf effective MTU (1452) und danach wg0 auf 1452-80=1372.
Reihenfolge der set_mtu-Aufrufe: erst eth0, dann wg0.
"""
argv = ["main.py", "--dry-run", "--apply-egress-mtu", "--pmtu-target", "46.4.224.77"]
with patch.object(sys, "argv", argv):
buf = io.StringIO()
with redirect_stdout(buf):
automtu.main()
out = buf.getvalue()
self.assertIn("Applying effective MTU 1452 to egress eth0", out)
self.assertIn("Computed wg0 MTU: 1372", out)
# Call order prüfen
calls = [
unittest.mock.call("eth0", 1452, True), # egress zuerst
unittest.mock.call("wg0", 1372, True), # dann wg0
]
mock_set_mtu.assert_has_calls(calls, any_order=False)
self.assertEqual(mock_set_mtu.call_count, 2)
# ---------- --apply-egress-mtu: egress == wg_if -> skip apply auf egress ----------
@patch("main.wg_default_is_active", return_value=True)
@patch("main.wg_is_active", return_value=True)
@patch("main.set_mtu")
@patch("main.probe_pmtu", return_value=1500)
@patch("main.read_mtu", return_value=1500)
@patch("main.exists_iface", return_value=True)
@patch("main.get_default_ifaces", return_value=["wg0"]) # wg0 wird als egress gewählt
@patch("main.require_root", return_value=None)
def test_apply_egress_mtu_skips_when_egress_is_wg(
self, _req_root, _get_def, _exists, _read_mtu, _probe, mock_set_mtu, _wg_active, _wg_def_active
):
"""
Wenn egress == wg0, soll das Skript den egress-Apply überspringen, aber wg0 ganz normal setzen.
"""
argv = ["main.py", "--dry-run", "--apply-egress-mtu", "--prefer-wg-egress", "--wg-if", "wg0"]
with patch.object(sys, "argv", argv):
out = io.StringIO()
with redirect_stdout(out):
automtu.main()
s = out.getvalue()
self.assertIn("Detected egress interface: wg0", s)
self.assertIn("Skipping egress MTU apply because egress == wg0", s)
# wg0 wird trotzdem gesetzt (1500-80=1420)
mock_set_mtu.assert_any_call("wg0", 1420, True)
# Es darf nur ein set_mtu-Aufruf erfolgen (kein separater egress-Apply)
self.assertEqual(mock_set_mtu.call_count, 1)
# ---------- --apply-egress-mtu plus --set-wg-mtu override ----------
@patch("main.set_mtu")
@patch("main.probe_pmtu", return_value=1452)
@patch("main.read_mtu", return_value=1500)
@patch("main.exists_iface", return_value=True)
@patch("main.get_default_ifaces", return_value=["eth0"])
@patch("main.require_root", return_value=None)
def test_apply_egress_mtu_with_forced_wg_override(
self, _req_root, _get_def, _exists, _read_mtu, _probe_pmtu, mock_set_mtu
):
"""
Egress wird auf 1452 gesetzt, aber wg0 wird mit --set-wg-mtu (z. B. 1300) überschrieben,
unabhängig vom berechneten Wert (1372).
"""
argv = [
"main.py", "--dry-run",
"--apply-egress-mtu",
"--pmtu-target", "1.1.1.1",
"--set-wg-mtu", "1300",
]
with patch.object(sys, "argv", argv):
buf = io.StringIO()
with redirect_stdout(buf):
automtu.main()
out = buf.getvalue()
self.assertIn("Applying effective MTU 1452 to egress eth0", out)
self.assertIn("Computed wg0 MTU: 1372", out) # erst berechnet
self.assertIn("Forcing WireGuard MTU (override): 1300", out) # dann überschrieben
calls = [
unittest.mock.call("eth0", 1452, True),
unittest.mock.call("wg0", 1300, True),
]
mock_set_mtu.assert_has_calls(calls, any_order=False)
self.assertEqual(mock_set_mtu.call_count, 2)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main(verbosity=2) unittest.main(verbosity=2)