From 10522d591a98af18aaa60a5c0d46d2458eef6c6d Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Wed, 15 Oct 2025 19:56:43 +0200 Subject: [PATCH] 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 | 19 ++++ README.md | 98 +++++++++++++++++ __pycache__/main.cpython-313.pyc | Bin 0 -> 12383 bytes __pycache__/test.cpython-313.pyc | Bin 0 -> 6833 bytes main.py | 178 +++++++++++++++++++++++++++++++ test.py | 134 +++++++++++++++++++++++ 6 files changed, 429 insertions(+) create mode 100644 Makefile create mode 100644 README.md create mode 100644 __pycache__/main.cpython-313.pyc create mode 100644 __pycache__/test.cpython-313.pyc create mode 100755 main.py create mode 100644 test.py diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b1cfa35 --- /dev/null +++ b/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]" diff --git a/README.md b/README.md new file mode 100644 index 0000000..5d97fb4 --- /dev/null +++ b/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) + diff --git a/__pycache__/main.cpython-313.pyc b/__pycache__/main.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..7bf40492f4a258f8e1e3a1c2bf2aed01b539dade GIT binary patch literal 12383 zcmb_ieQX=YmER?oA0nx5IhJL|UR!b^+WP*oOxf-64_=bawXB0mdq|G zS!|l9y^qSt6_J|+R(r)&X@gqD1$WAwaZ!N+O@RXS;R4*!R6}(u^opKwhulBsI7Qpb zU-z3`?vj#aH+SeroSmI{`{vD?H*em&d7tIyJ1C?-to`>`Y9~egJ9*KQDO33D4hnBm z93@g5&6!TqBCS46qDg(4MYH;3L`HpDM2q^gidOY$6K(3#F4{SB1tsQj3~~o&LGI+N z$X%Qbc|K=HUclucFXSA^i#R9pVy=jDRZv%Hv4kt;^3_r)SAdqXOIx@?&mctw>=bpW zY!B6%A(^E{O`|4vN&N3e;^)<)J3P+-^{$lcA#7v;$wU;#`2>gFG*}~ zcMThf#=;RHFg`xbM#ac5&z=j!#@Mq1J~kF%BLdGxM0O$~@@y<1j`A_7#u1E6MB_13 zjPYzRB8mjz#jwbqj0eOJ`Lc!$2q88QjRI#lR_AbZT@Oq|$9c))U|A_1im-L{>j`{Z zJzHBl5)p%ZtwwMy#>6(Zv^G5fJ_?vxjaIg$t*)i6si~#zzyV;&ZD|TLGU;8!GUsi(D1uX8O7Z z+>Bxgj`6{3idBk*BJr5|pj<{*gCr|^!d)5 z5iLNfSfrR3j{bmF%)yC}M3})U&)1JdCiwbm{A5^|IkjXp)K3J$0+xiBhYr-N5Cy3j*$lPj-15fJ*N)CtF8J5XNafG?*#3fMpA`_Q0 zF;AYhO48IjDbHfS=VibLdMaZw0HLcExd3XS0+VTNK7 zgJLNPC8YjYWI3F9_iFl?gUM$O{Icus-60rh3BaD^if6+FHmH>7InPV%y|1AKi)4e64PTxQuikq31pIt1oLF^QQ+?d z&}#Kr_3;Fn(1(??kOx;C&6ui(7hXVvM|afGlUhRB*y6WTK~Z?uzHIZ9E3 z{9&}Ha|HhSESt`e@in*)nh5Pa3zm9bpWC9C!qJ4OwnedsP{6#xNMn&H#TMd60`c*f z+am4&1)rO0+r+0)5MlRF;x1%~{7b_je)5vEzxool-~9=p56Xf-tB{Hrdw#;!Sf~DF z@QAG5Ebc)mQP$Mf(B^45(9ob|o<=0aG#*LhU#XrNeK{5v+?PW8s@!%}0bx6cj)!C8 zVS$&RYj|uId_W9hslsI7g^*$y#UP@JWg-v@j;YJiZBrPKDTxK>Pcf4{h9Tc8^n}91 zh44!X9g@iUW3{9$?-J1N&#*vS}FTYWeCJ}#@8?OZD=xs~^gJh@`uVrQzTakk^%3X5i6nyb3C|K|Srz;{N! zIl2(|>#HA^xIeQ{TlRixp^7fizjIKoBBiY9UP~xd%FmuyE7>v^e6t#bEV2`+qK3yK z`_x92jnEJ5l&fH-^OyO>v-GbDi{?7!%iipLx8!#Dcgy8{p5>18?xV@wM}Imla|1t% z{3MdR=wEfZzTQ4x^_~6S+`ka`n>vtFQ1ZYEYGzJ9s-TVS>ib{^6m&*DDd;VtiaArW znKSf>iRdmvNBQU{sb?>ljSWfHW%??ROdL#Xj?z~ROI1<3shX@t0W~8gmf8hLhj~l2 zn|DzoG;9u$sObb#k4;^nMc5xq1cIZO!vf?&upiOJDXWD6P$LOoHnXU*57Dm$!D^9Z1=|kBT+18&tl( zivswb6{%kE9d%mT+?dwIl%qXUIA_)43xZEgoN36gw>C4EVMc@3IWX#3Q~z3fVyjp; z$D?F!EPW!G>cnAvf`+=h-hS_8_S`rhka!kWGaHatF%pTf)#ME2CI_=PT8wi<9c*W2 zBs|WGyMYO2w<#hiHt0Y;9uA4^sKfM3OR7PrFfe^$n8;(OUS|@C{7Z2h=KQ25u?MX! z1fmTYrpQ_FR?{udH#~11S!hg^R+0HCIKATarM;Ju-b?rFm)GphH|($3=Ze4Xd|;u< z_RcmxVRBvr(zmIdISb*BiMx-|bw>`QF(E*gNxDWM`eewi*V@!WGpp)0BAk+G7c|a@ zu2)Qb3hlvG>Ih3=L5c;0Ag|AmB|p8 zT#nkpl0O`sY!lC;%R{7FZ1`a21C!aZeXVfYx83QAx_cFMi?LKi%bjq#?R2v3bgJ-- zY(Mh{97sN+Xv_9hXra)JFWs|OJTT++bBWj#U#)GRmI{gDp=spn6b~`o)dirS07nL# z7@V#upjEY$U*&rvXjd9Kfe&Taj@J`Z-PVP$3&Qm*pznK!jKp{#Ul zSVrVvqNc&ddb$=Ry$8TJ3f+i@gx^JVC1~afSzT_M zNO%ub7Pu}FxmL|3A*4=?V#TcUVoW{yK)2x>F+Cm$gcPghO;GY;;Rzn5hhI{dVXR4s z0S2fwU9S;3{E=(oWmK!n>O~Y_Vz^4?TA;iJ7AhA)?=>$@{q2!u=11I5OFumSk@=%8 z**lavb1BtxIn{9`)$W&_FV1wX*$QSSzxL`^UR|}j-f-S<&UMViR_r_0ocXh%xxpKe zlym38?xeF)W-7JiMJUmev87uv(YafCcDjv;08==+AHpFp*=Zjv!V!W950VZUg0aY% zVI;$L5-d5e=Qe=_*0WC$GghEks^3fu)lp`!gkz##6Rd-y9uKrC4C=vAGNqiQ7Vtt| z4$iqpaR%(bmnYbLd7PDeC0P3{Y8m=fuw`(Peu39#9|k`L|D3>8LDxb|YzhDIun>U1 zMKd*3-_(Kkzv5=@L;XZiD z>OLWcSNy$})9lD5G*R6m}dxVyxwXvc|ylD>s(hPq$0b>4cb z@MhtSJqzBM6Zf4ZbDo8%w6i|xtp8Q$4i0-O ztE@QV0L{}r$g%1fB@h|;g)|sSqYB4u&TK#!d-?_lMumXjXH{Ny3-J<$Z%BJ`6jc_N zuB!BBRW8*QAMCxXpM$5}*2IBx;ovoRPFa3r1TNL^BwVpX;zq>C1RH>j22qWSz<(Mf zezcMJxY{h$CAJC?tw=n!!LdjLMgjSVNQfZfR7)w=2|g4K2#LbDa7~Cz32eqS35D2? zc5$3!1X9zA4(xt0^cN0F;)MFW6cJ;5$Zdkf2eFbAtA=2mVO^y4?KkKbMSM$STpC3- zL;Z)dXsw`J-qw^VXqN5GtG4{PEr0eZv9McTPssLdn^u&kY})|xmM2|Wb+5GQ-3yCd z-@BYDZB7-o$o7^sN7>w~3zG|D@|OCPqd{gGG*Lo-Mj~ecRg?-ad0)f{#hDJ#0`Zg> zB4Xfz5+nvUG4dYMZp|8h9E|Tn=Ep+jJz1>0`Pp9e66fHY`u@ndxcnzzFUZpIMFn}& z9t*i5y+>V?aV1f5hG`QA_PmuV=eBVbPvFY-A>u($6SsqV`iaJwhCUdz2|A}8SYuG`U!k=bNjgc#%PLly0cIHpn_95ao4H| zr}8qroPFvuhdHCY*qS${cG}qJ;TnSFMGnqt()-Zc1CYu`Te%vtRA5L@#E+@NhH$zY z+MYo~akWEadsLH`IG{Epu1-BVWntJ20C>gGsXjZFp-LUXXV|1h-|*h$^xfby;eg%P zZ2ptb(3C?%c6`mGkE~Kxd0nMbn0@LL271;}*n*ka1R7Mgs(5pbj6=`r*amlR_RV8t zfapM^$I&I2`<+8>4TgfNLLVb+`Ro|meC7>fGz*OCL8ZO>^l!6IW2|GqxG4loO87G2 zceQwly|y9wtX>?yHzw|YPFpE;!6upNagqZb&>orcHg1H@-dE|=sD4&RxX<<^=yRsC z`!z4CWD;z8kMO->N7H<-I=;YvQB(sZ*ujUq99|vd4tbj&rN(E`aba#f;0ig2Y)EK1 z+-$&M#w-{gHc`PbM4VP?8hpn~{8k$h@Od^=;T5&S$+97}GFpY;99;G}h*0^Q#+aaQ ze0g4Sj@giWPMv->cZD7CIY4!L4hE(df?Ma7R$8kfKY);)^ za?}09&?PzJJGnUqP}xJqmpl9Y+j73^ee8Fqn$0@R>tyY;s?E<3UBsRBF~|=+7Du1j zwaQL%kNTur88;CFlGQ!XWgB31X0w;08Tvf}OziN?4CK zQ0;Yc`hq8*4!$5~h?j!6)Fu%Y31_Vy#Sk7zw02#`9Y`>2gkr#r%8TTFlC37!)R{mH z^c@~yBNO2mvAzjPW{e+?CQ2TSi%1;nCXo_2#fBv{d;>(22#rjrM;8AukB|xyxj~{h zKqxykO5p2Y+%vK1C@;n#n+aQOEj-2H5yf)xq`$XYF;9&)B&xH_$>I|N6CgTs&Ef&J z1`ujeZV4rx2LP4Q!yQENYo#nj^W6Sv(3emtPV!bGKd<8 zi2+2HWU(J4$j#y;NX+TMuJ(Y_j$#G!v%PQ%H8uhBvuDFXcp{GD)A)E~3d}?Y>LK2s zmVmkI5+E@rIz%LEB5{5|4R~S4B2tWuvA*{lAu1CjLpa8sz-eumXGd{X6EsOclnYDF z?GoGXB`lr@z~c}NhylzW;jk6MfiP%{DbRb>{7V0+>4XQz5)A`d4LrSkFd~E`H;Z{; ziGwNAf0iSSbz}e#2z3D9L2bsF@W~R?d!pqG1}?J0kvODQ_gz2$w6kT#825WA!1GaRl< zS#54K=MVgRS_Bi}d2lQsz*SVIx#&ALy68_t;^QG!fMdniw*>` zh4~POv8V%26dK6-`UkpD-=R)W6xRg25M!tLSRKp3X`CAtHxd@5SPjw00xR+)>`T+s z+Wt_N!{|=pPGq?_bcZNUE0r6dx&NWD@LPm)pv!Bhm<2=e|6IPpV1N7{(#syK1IFjpf`)%qWB=C^n$5Wql z#ex?s$^yj{h~feVBS+Yfx*QZcSr3H6O1`0o)YnW`07g)x!h~b|ghF3eY!CvN1gIH} zL_!K36GzCXtx!b}4y}7Q(1;g8_uA(AZW$7uCf-IX`AoVB6;3Xui_>}Ry*w5%3w0+= zpCZ1q#{?fNWBpqHSNiA9%GLgv{v`8ara~^OeovAQB}m)e+70->W@w*SEUO|G-RH%09)- zS%B7iKfUOZ4+jaZ5Q0A5Xur|^`mveL`%KYX*{vNncdRfw5hPJi^2U`LSJL?}tQTvhVqs?q4~J)mzVjm7y!?AwD_8-?NVxx1N9HTyw(1o_6j^ zI(MnJ+4~mT7OUl^<8r}?k1GFl-#_oWXZNjKd?9_&fA6ASelf6ead?f%lU)rku%gWPbRVyUkWz|)lcI`^Kb}e;&aO(Y2%hH|x6np+3`WIjN+0;*_em4ER zZ41S>x4yIW``g~{PqODfc6mQFBk<*Spu23fq-@rPi-B22rM-TM`5^E8yycoy!%4Zm z=Wd?ty^!_}C%wb+`Jfy`pas8TANghR)3|yls+ey~?`%%)Y+f2jx1LJ2p1RwaK65#F z=5ng&%52B#wu-rq8>bKivszSk%X!l|-|<$#nyYy3%tG;kd7erfn_FruqS!2M?TPdxAg9X zyZFqv;<~W5W7mQ`wWIOI+4}%1d49$9!fzoP#gLTkH@DAEr0b3*@xSwED*xC_x60j} z^Re$FzL}UGUwnp4yA`_RtN9R1K`rpq2W7<&6d u7eiP5)>TTk{dS*&ZhTnYOVh8?4=$J}#xu%*C6pRk%(kEOQ;c>sEOsx z)C{rttmTZATF=<1O~iW3S^F7|ax85+%Tq@qIY=aH6On8a=22+p@u$yXlu+k{OR^9C zkdR0$aJQ7Ms$ou}CWAGwB|x#FJEsX|k%t zGg&RRS(PK7S-)w*u}u&h4~)e`EfZE~!Q2Ps_^_NsvP!lAGfUf{^+KAL99^bjPLQ0K zmRwDw&v?0FOl@UatK5RSatqX6M)U5*QB}E>!Z<1SAGhY&MQgmYhdL5YU$yQYQ&jDI z;+&Q}k<~H}oG|Y>*K{PA4ycMIPixr?H_K5a636U=RoK-xOuCCIT9&5e*+?p@WvSR5>{ri78{b23D2;EpcTsl1*twg{oT>ZK^x}3RT|B!XA@pCZmnS?DB3XuQrgdwCCT2%+2f~U5xC7lvf-0&iCexZiaW@Z&Y|qx8G|I7G*wsN1Tio!5GSf3z zeD6}CJ6{fs=@!U7_ywGou^Uc`ds(+gHrd=OvB8*iV0-qOs7q2pmG_ ztZv0#>O7ntcEEH#qN)njPN#K`k&+|nIT=BH!=YPMB{iu#WH?OhAnClEKXy7N8;6qB z8YftHO=n^k<^1ucxDRzpG&`wVU`15qx>$JhH@c=W(@NJxWj2|fRg`ozNiRYeiA{Bl zDl-{1sby$VQM(ebp4n(1mYFUqdnw_ELHTiTT^jHvFf*sOV?s`6!YggETrc7?`)*Xza|h*%?GKUr`}IrwtU6A zK5+lseb?K6v-UneT+I3x)cgG5Vm7d7xzG2m*@>h6OTP6U-+Fy~A-veS!uNky3woiU z4W=79v7tT(9sN~HsMd1Z>YT;r9j{?LAt`r00+klym z$ptu)#)GJoECXhStWqghV9DNPmN=uPi%9(IrnGGrKy{4U3P9KuQ0<$5YJWzc3KINE z@Wb40^5^Y40+-tcrcN*bMTl=B<9Z4i_fyDJZIQ8-g|bK5BUKmH0H7@FfZX|tQ2w-{ z#1|DMev=1#P3|3`i0jD=jVW;;ZeQ?4gYke<@BmZ*CX{tT80;4P$HbXTDjAy_h7&Ii zA2Uj0QyEoB2Lxqu5=fbxRj{^$y@qf37{_1S40Zp}?%|%pz1;%EBa!s96GG8~Z+HyU zgLzRHWgh$*&JRWJ85|g=nv5&5(TY!?9w?>gEaz(ce%61#fg>cBZpHRD+Ehnj>GXRZ>!hg(}K!uE*&OcIV33BbXnGE=6((?GZ(b@8hWm5pBWXAO?sb z+J*sI1^PS&Fq=e0hpAu`C^`Z9AmcEbs3Nz4U&q4_I-ZnlS7~SuY^|jb}ZDc?DelU9{Ak#iR&|je^USe?d?NchfR>L z>R;Rc6n5amt;6^Emy32_=w|Fb|57n)n1!$Sy2p(|HM9pEK%?O8gU)Rj?XFaZHdxy$ z?jsxO>oC^$-Qh0FwU+RH+wWV#ZtixIH#Fq9-OXeAC=bPdBs@qg4GCeveaFs0`5m_q zZg$>j;Gp)-ej(iHxYNOb%$+VFJj`uzyK$wNRjGTzs({5Xziuj67RXt_e(c5}xBU+t za=Qe-63a2SQ~ntp^sSC{c_*IV;(V8vacq$(@1Wz?vTKtqusI zLh*)$`XNZ(LVyRbbQ7@+2%94LIGT@Rv=;<*yU~k0jQ!YL4?vf;`q&0e8yHTRD{MegHoC31m0{(RVy~duKR^H=Yy=SA zt!@2tFud$O3GAz@x*P`f)ihk)H^1+**!3d|y(`t-tKP<|7w0d2IDN&kTHktITB-M~ z)-_*!b^g_lMu0co7l1=GJ;0!L|j<*;}17Wlhoi(or`3%ucz#q746^MGz>3^`ZR z106xmK^rUXlN;(FjNq3op#jTpTS6VS+a7PIfxF$tW7^L{@lt~jYI82Na8R<;A%u=N zmVz9X4+vpy8;-4k2mFM03|9&NryMhbZ*Shju?qQTU$>mEHCp>J$Cm8Z`|j9RGB00Cpf~iuDFjEIns_C@iLCs7)Kxt9BH>K z*@RH7bIHR&?UGjrwK|qW4$IqxP#<>#tP5PB$mz(mBFnlU%hQ>77F=0bcggab*+?p1 zHE1;g2vQ zKsiidfL|}_8vw4Kk-yo5%g2^&O@D#-_(I)1TL4mbo%cUj>ngF#AUm$j8L33q&&1Lm9ukldvaB!2V-E?wqS*3+eLhZvN zY62hHmO)Y3DBxyF^o??hRCfM5FXe-8&-HAeAnYtHAUjVRXufzD^sNe$1N?k=PP zc#~$Vx)nu)SO#tk-+~PhW%qfs8Q*SVZ{raR!C^F3`6O20@*5ptpH%#ivCodtQK-O+ z8}&2!Fzd literal 0 HcmV?d00001 diff --git a/main.py b/main.py new file mode 100755 index 0000000..2ee1927 --- /dev/null +++ b/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() diff --git a/test.py b/test.py new file mode 100644 index 0000000..1e2a657 --- /dev/null +++ b/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)