Add wg-mtu-auto initial implementation, documentation, and unit tests

- Added main.py: automatic WireGuard MTU calculation and PMTU probing
- Added test.py: unittests covering base, PMTU, and fallback scenarios
- Added Makefile: includes test target and install guidance
- Added README.md: usage, pkgmgr installation, and MIT license

Reference: https://chatgpt.com/share/68efc179-1a10-800f-9656-1e8731b40546

Makefile

@@ -0,0 +1,19 @@
+PY ?= python3
+
+.PHONY: test install help
+
+help:
+	@echo "Targets:"
+	@echo "  make test      - run unit tests"
+	@echo "  make install   - print installation guidance"
+	@echo "  make help      - this help"
+
+test:
+	$(PY) -m unittest -v test.py
+
+install:
+	@echo "Installation is provided via your package manager:"
+	@echo "  pkgmgr install automtu"
+	@echo ""
+	@echo "Alternatively, run the tool directly:"
+	@echo "  $(PY) main.py [--options]"

README.md

@@ -0,0 +1,98 @@
+# wg-mtu-auto
+
+Automatically detect the optimal WireGuard MTU by analyzing your local egress interface and optionally probing the Path MTU (PMTU) to one or more remote hosts.  
+The tool ensures stable and efficient VPN connections by preventing fragmentation and latency caused by mismatched MTU settings.
+
+---
+
+## ✨ Features
+
+- **Automatic Egress Detection** — Finds your primary internet interface automatically.  
+- **WireGuard MTU Calculation** — Computes `wg0` MTU based on egress MTU minus overhead (default 80 bytes).  
+- **Optional Path MTU Probing** — Uses ICMP “Don’t Fragment” (`ping -M do`) to find the real usable MTU across network paths.  
+- **Multi-Target PMTU Support** — Test multiple remote hosts and choose an effective value via policy (`min`, `median`, `max`).  
+- **Dry-Run Mode** — Simulate changes without applying them.  
+- **Safe for Automation** — Integrates well with WireGuard systemd services or Ansible setups.
+
+---
+
+## 🚀 Installation
+
+### Option 1 — Using [pkgmgr](https://github.com/kevinveenbirkenbach/package-manager)
+
+If you use Kevin Veen-Birkenbach’s package manager (`pkgmgr`):
+
+```bash
+pkgmgr install automtu
+````
+
+This will automatically fetch and install `wg-mtu-auto` system-wide.
+
+### Option 2 — Run directly from source
+
+Clone this repository and execute the script manually:
+
+```bash
+git clone https://github.com/kevinveenbirkenbach/wg-mtu-auto.git
+cd wg-mtu-auto
+sudo python3 main.py --help
+```
+
+---
+
+## 🧩 Usage Examples
+
+### Basic detection (no PMTU)
+
+```bash
+sudo automtu
+```
+
+### Specify egress interface and force MTU
+
+```bash
+sudo automtu --egress-if eth0 --force-egress-mtu 1452
+```
+
+### Probe multiple PMTU targets (safe policy: `min`)
+
+```bash
+sudo automtu --pmtu-target 46.4.224.77 --pmtu-target 2a01:4f8:2201:4695::2
+```
+
+### Choose median or max policy
+
+```bash
+sudo automtu --pmtu-target 46.4.224.77,1.1.1.1 --pmtu-policy median
+sudo automtu --pmtu-target 46.4.224.77,1.1.1.1 --pmtu-policy max
+```
+
+### Dry-run (no system changes)
+
+```bash
+automtu --dry-run
+```
+
+---
+
+## 🧪 Development
+
+Run unit tests using:
+
+```bash
+make test
+```
+
+To see installation guidance (does not install anything):
+
+```bash
+make install
+```
+
+---
+
+## 👤 Author
+
+**Kevin Veen-Birkenbach**
+[https://www.veen.world](https://www.veen.world)
+

main.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""
+wg_mtu_auto.py — Auto-detect egress IF, optionally probe Path MTU to one or more targets,
+compute the correct WireGuard MTU, and apply it.
+
+Examples:
+  sudo ./wg_mtu_auto.py
+  sudo ./wg_mtu_auto.py --force-egress-mtu 1452
+  sudo ./wg_mtu_auto.py --pmtu-target 46.4.224.77 --pmtu-target 2a01:4f8:2201:4695::2
+  sudo ./wg_mtu_auto.py --pmtu-target 46.4.224.77,2a01:4f8:2201:4695::2 --pmtu-policy min
+  ./wg_mtu_auto.py --dry-run
+"""
+import argparse, os, re, subprocess, sys, pathlib, ipaddress, statistics
+
+def run(cmd):  # -> str
+    return subprocess.run(cmd, check=False, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True).stdout.strip()
+
+def rc(cmd):  # -> int
+    return subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode
+
+def exists_iface(iface):  # -> bool
+    return pathlib.Path(f"/sys/class/net/{iface}").exists()
+
+def get_default_ifaces():  # -> list[str]
+    devs = []
+    for cmd in (["ip","-4","route","show","default"], ["ip","-6","route","show","default"]):
+        out = run(cmd)
+        for line in out.splitlines():
+            m = re.search(r"\bdev\s+(\S+)", line)
+            if m: devs.append(m.group(1))
+    if not devs:
+        for cmd in (["ip","route","get","1.1.1.1"], ["ip","-6","route","get","2606:4700:4700::1111"]):
+            out = run(cmd)
+            m = re.search(r"\bdev\s+(\S+)", out)
+            if m: devs.append(m.group(1))
+    uniq = []
+    for d in devs:
+        if not d or d == "lo" or re.match(r"^(wg|tun)\d*$", d) or not exists_iface(d): continue
+        if d not in uniq: uniq.append(d)
+    return uniq
+
+def read_mtu(iface):  # -> int
+    with open(f"/sys/class/net/{iface}/mtu","r") as f:
+        return int(f.read().strip())
+
+def set_mtu(iface, mtu, dry):
+    if dry:
+        print(f"[wg-mtu] DRY-RUN: ip link set mtu {mtu} dev {iface}")
+    else:
+        subprocess.run(["ip","link","set","mtu",str(mtu),"dev",iface], check=True)
+
+def require_root(dry):
+    if not dry and os.geteuid() != 0:
+        print("[wg-mtu][ERROR] Please run as root (sudo) or use --dry-run.", file=sys.stderr)
+        sys.exit(1)
+
+def is_ipv6(addr):  # -> bool
+    try:
+        return isinstance(ipaddress.ip_address(addr), ipaddress.IPv6Address)
+    except ValueError:
+        return ":" in addr
+
+def ping_ok(payload, target, timeout_s):  # -> bool
+    base = ["ping","-M","do","-c","1","-s",str(payload),"-W",str(max(1, int(round(timeout_s))))]
+    if is_ipv6(target):
+        base.insert(1, "-6")
+    return rc(base + [target]) == 0
+
+def probe_pmtu(target, lo_payload=1200, hi_payload=1472, timeout=1.0):  # -> int|None
+    """Binary-search the largest payload that passes with DF. Return Path-MTU (payload + hdr) or None."""
+    hdr = 48 if is_ipv6(target) else 28
+    # ensure the lower bound works; if not, try slightly smaller floors
+    if not ping_ok(lo_payload, target, timeout):
+        for p in (1180, 1160, 1140):
+            if ping_ok(p, target, timeout):
+                lo_payload = p
+                break
+        else:
+            return None
+    lo, hi, best = lo_payload, hi_payload, None
+    while lo <= hi:
+        mid = (lo + hi) // 2
+        if ping_ok(mid, target, timeout):
+            best = mid
+            lo = mid + 1
+        else:
+            hi = mid - 1
+    return (best + hdr) if best is not None else None
+
+def choose_effective(pmtus, policy="min"):  # -> int
+    """Pick an effective PMTU from a list of successful PMTUs."""
+    if not pmtus:
+        raise ValueError("no PMTUs to choose from")
+    if policy == "min":
+        return min(pmtus)
+    if policy == "max":
+        return max(pmtus)
+    if policy == "median":
+        return int(statistics.median(sorted(pmtus)))
+    raise ValueError(f"unknown policy {policy}")
+
+def main():
+    ap = argparse.ArgumentParser(description="Compute/apply WireGuard MTU based on egress MTU and optional multi-target PMTU probing.")
+    ap.add_argument("--egress-if", help="Explicit egress interface (auto-detected if omitted).")
+    ap.add_argument("--force-egress-mtu", type=int, help="Force this MTU on the egress interface before computing wg MTU.")
+    ap.add_argument("--wg-if", default=os.environ.get("WG_IF","wg0"), help="WireGuard interface name (default: wg0).")
+    ap.add_argument("--wg-overhead", type=int, default=int(os.environ.get("WG_OVERHEAD","80")), help="Bytes of WG overhead to subtract (default: 80).")
+    ap.add_argument("--wg-min", type=int, default=int(os.environ.get("WG_MIN","1280")), help="Minimum allowed WG MTU (default: 1280).")
+    # PMTU (multi-target)
+    ap.add_argument("--pmtu-target", action="append", help="Target hostname/IP to probe PMTU. Can be given multiple times OR comma-separated.")
+    ap.add_argument("--pmtu-timeout", type=float, default=1.0, help="Timeout (seconds) per ping probe (default: 1.0).")
+    ap.add_argument("--pmtu-min-payload", type=int, default=1200, help="Lower bound payload for PMTU search (default: 1200).")
+    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",
+                    help="How to choose effective PMTU across multiple targets (default: min).")
+    ap.add_argument("--dry-run", action="store_true", help="Show actions without applying changes.")
+    args = ap.parse_args()
+
+    require_root(args.dry_run)
+
+    # Detect egress
+    egress = args.egress_if or (get_default_ifaces()[0] if get_default_ifaces() else None)
+    if not egress:
+        print("[wg-mtu][ERROR] Could not detect egress interface (use --egress-if).", file=sys.stderr)
+        sys.exit(2)
+    if not exists_iface(egress):
+        print(f"[wg-mtu][ERROR] Interface {egress} does not exist.", file=sys.stderr); sys.exit(3)
+    print(f"[wg-mtu] Detected egress interface: {egress}")
+
+    # Egress MTU
+    if args.force_egress_mtu:
+        print(f"[wg-mtu] Forcing egress MTU {args.force_egress_mtu} on {egress}")
+        set_mtu(egress, args.force_egress_mtu, args.dry_run)
+        base_mtu = args.force_egress_mtu
+    else:
+        base_mtu = read_mtu(egress)
+    print(f"[wg-mtu] Egress base MTU: {base_mtu}")
+
+    # PMTU over multiple targets
+    effective_mtu = base_mtu
+    pmtu_targets = []
+    if args.pmtu_target:
+        # flatten comma-separated + repeated flags
+        for item in args.pmtu_target:
+            pmtu_targets.extend([x.strip() for x in item.split(",") if x.strip()])
+
+    if pmtu_targets:
+        results = {}
+        good = []
+        print(f"[wg-mtu] Probing Path MTU for: {', '.join(pmtu_targets)} (policy={args.pmtu_policy})")
+        for t in pmtu_targets:
+            p = probe_pmtu(t, args.pmtu_min_payload, args.pmtu_max_payload, args.pmtu_timeout)
+            results[t] = p
+            if p:
+                good.append(p)
+            print(f"[wg-mtu]  - {t}: {'%s' % p if p else 'probe failed'}")
+        if good:
+            chosen = choose_effective(good, args.pmtu_policy)
+            print(f"[wg-mtu] Selected Path MTU (policy={args.pmtu_policy}): {chosen}")
+            effective_mtu = min(base_mtu, chosen)
+        else:
+            print("[wg-mtu] WARNING: All PMTU probes failed. Falling back to egress MTU.")
+
+    # Compute WG MTU
+    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})")
+
+    # Apply
+    if exists_iface(args.wg_if):
+        set_mtu(args.wg_if, wg_mtu, args.dry_run)
+        print(f"[wg-mtu] Applied: {args.wg_if} MTU {wg_mtu}")
+    else:
+        print(f"[wg-mtu] NOTE: {args.wg_if} not present yet. Start WireGuard first, then re-run this script.")
+
+    print(f"[wg-mtu] Done. Summary: egress={egress} mtu={base_mtu}, effective_mtu={effective_mtu}, {args.wg_if}_mtu={wg_mtu}")
+
+if __name__ == "__main__":
+    main()

test.py

@@ -0,0 +1,134 @@
+import io
+import sys
+import unittest
+from unittest.mock import patch, call
+from contextlib import redirect_stdout
+
+# Import the script as a module
+import main as automtu
+
+
+class TestWgMtuAuto(unittest.TestCase):
+
+    @patch("main.set_mtu")
+    @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_no_pmtu_uses_egress_minus_overhead(
+        self, _req_root, _get_def, _exists, _read_mtu, mock_set_mtu
+    ):
+        """
+        Without PMTU probing, wg MTU should be base_mtu - overhead (clamped by min).
+        With base=1500, overhead=80 ⇒ wg_mtu=1420.
+        """
+        argv = ["main.py", "--dry-run"]
+        with patch.object(sys, "argv", argv):
+            buf = io.StringIO()
+            with redirect_stdout(buf):
+                automtu.main()
+
+        out = buf.getvalue()
+        self.assertIn("Detected egress interface: eth0", out)
+        self.assertIn("Egress base MTU: 1500", out)
+        self.assertIn("Computed wg0 MTU: 1420", out)
+
+        # dry-run still calls set_mtu (but prints DRY-RUN); ensure it targeted wg0 with 1420
+        mock_set_mtu.assert_any_call("wg0", 1420, True)
+
+    @patch("main.set_mtu")
+    @patch("main.exists_iface", return_value=True)
+    @patch("main.get_default_ifaces", return_value=["eth0"])
+    @patch("main.require_root", return_value=None)
+    def test_force_egress_mtu_and_pmtu_multiple_targets_min_policy(
+        self, _req_root, _get_def, _exists, mock_set_mtu
+    ):
+        """
+        base_mtu forced=1452; PMTU results: 1452, 1420 -> policy=min => 1420 chosen.
+        effective=min(1452,1420)=1420; wg_mtu=1420-80=1340
+        """
+        with patch("main.read_mtu", return_value=9999):  # should be ignored because we force
+            with patch("main.probe_pmtu", side_effect=[1452, 1420]):
+                argv = [
+                    "main.py",
+                    "--dry-run",
+                    "--force-egress-mtu", "1452",
+                    "--pmtu-target", "t1",
+                    "--pmtu-target", "t2",
+                    "--pmtu-policy", "min",
+                ]
+                with patch.object(sys, "argv", argv):
+                    buf = io.StringIO()
+                    with redirect_stdout(buf):
+                        automtu.main()
+
+        out = buf.getvalue()
+        self.assertIn("Forcing egress MTU 1452 on eth0", out)
+        self.assertIn("Probing Path MTU for: t1, t2 (policy=min)", out)
+        self.assertIn("Selected Path MTU (policy=min): 1420", out)
+        self.assertIn("Computed wg0 MTU: 1340", out)
+        mock_set_mtu.assert_any_call("wg0", 1340, True)
+
+    @patch("main.set_mtu")
+    @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_pmtu_policy_median(
+        self, _req_root, _get_def, _exists, _read_mtu, mock_set_mtu
+    ):
+        """
+        base=1500; PMTUs: 1500, 1452, 1472 -> median=1472.
+        effective=min(1500,1472)=1472; wg_mtu=1472-80=1392
+        """
+        with patch("main.probe_pmtu", side_effect=[1500, 1452, 1472]):
+            argv = [
+                "main.py",
+                "--dry-run",
+                "--pmtu-target", "a",
+                "--pmtu-target", "b",
+                "--pmtu-target", "c",
+                "--pmtu-policy", "median",
+            ]
+            with patch.object(sys, "argv", argv):
+                buf = io.StringIO()
+                with redirect_stdout(buf):
+                    automtu.main()
+
+        out = buf.getvalue()
+        self.assertIn("Probing Path MTU for: a, b, c (policy=median)", out)
+        self.assertIn("Selected Path MTU (policy=median): 1472", out)
+        self.assertIn("Computed wg0 MTU: 1392", out)
+        mock_set_mtu.assert_any_call("wg0", 1392, True)
+
+    @patch("main.set_mtu")
+    @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_pmtu_all_fail_falls_back_to_base(
+        self, _req_root, _get_def, _exists, _read_mtu, mock_set_mtu
+    ):
+        """
+        If all PMTU probes fail, fall back to base MTU (1500) => wg_mtu=1420.
+        """
+        with patch("main.probe_pmtu", side_effect=[None, None]):
+            argv = [
+                "main.py",
+                "--dry-run",
+                "--pmtu-target", "bad1",
+                "--pmtu-target", "bad2",
+            ]
+            with patch.object(sys, "argv", argv):
+                buf = io.StringIO()
+                with redirect_stdout(buf):
+                    automtu.main()
+
+        out = buf.getvalue()
+        self.assertIn("WARNING: All PMTU probes failed. Falling back to egress MTU.", out)
+        self.assertIn("Computed wg0 MTU: 1420", out)
+        mock_set_mtu.assert_any_call("wg0", 1420, True)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)