From 2d67146487f6889ddcfe6705746601333565c4eb Mon Sep 17 00:00:00 2001 From: qsc Date: Sat, 6 Jun 2026 12:34:35 +0800 Subject: [PATCH] OK --- .gitignore | 7 + README.md | 2 + assets/app.ico | Bin 0 -> 8827 bytes core/Function/device_discovery_fun.py | 618 ++++++++++++++++++++++++++ core/Function/dns_diag_fun.py | 584 ++++++++++++++++++++++++ core/ui/basic_ui.py | 71 --- core/ui/tab_device_discovery.py | 233 ++++++++++ core/ui/tab_dns.py | 229 ++++++++++ core/ui/ui_main.py | 6 + 打包.bat | 5 +- 10 files changed, 1682 insertions(+), 73 deletions(-) create mode 100644 assets/app.ico create mode 100644 core/Function/device_discovery_fun.py create mode 100644 core/Function/dns_diag_fun.py delete mode 100644 core/ui/basic_ui.py create mode 100644 core/ui/tab_device_discovery.py create mode 100644 core/ui/tab_dns.py diff --git a/.gitignore b/.gitignore index 7109433..18a7a33 100644 --- a/.gitignore +++ b/.gitignore @@ -24,6 +24,13 @@ dist/ *.spec *.manifest +# Packaged binaries and release archives +*.exe +*.msi +*.7z +*.rar +*.zip + # Local runtime data network_profiles.json *.db diff --git a/README.md b/README.md index 8d6d73e..4410b7f 100644 --- a/README.md +++ b/README.md @@ -6,11 +6,13 @@ - 网卡配置：读取网卡、IPv4/IPv6、网关、DNS、DHCP、链路速率、接口索引、DHCP 服务器与租约时间。 - 配置模板：保存常用网卡配置、套用模板到表单、删除模板。 +- DNS 诊断：读取本机 DNS 并补充常用 DNS 服务器做对比，支持多域名、多记录类型解析、耗时/失败原因统计、诊断摘要复制与 CSV 导出。 - Ping 探测：支持持续/指定次数、间隔、超时、包大小、TTL、禁止分片、本地源 IP 下拉、CIDR/范围/列表批量探活、目标导入、CSV 导出、实时丢包率/平均延迟/抖动/质量统计。 - 端口扫描：支持单端口测试、端口列表/范围、常用端口预设、批量主机/CIDR/IP 段扫描、并发/超时控制、只显示开放端口、服务名识别、可选 Banner 探测、开放端口复制与 CSV 导出。 - 路由追踪：支持 IPv4/IPv6、主机名解析开关、单次/指定次数/持续追踪、最大跳数、单跳超时、高延迟阈值、逐跳 RTT 表格、超时/高延迟/波动诊断、原始输出保留、诊断摘要复制与 CSV 导出。 - 环网检测：基于本机网卡统计、广播/组播流量、ARP/邻居表变化和网关 Ping 稳定性，判断疑似二层环路/广播风暴风险，支持单网卡/全部活动网卡、诊断摘要复制与 CSV 导出。 - IP 冲突：支持本机 IP 冲突检测与安全限速网段扫描，结合地址状态、Tcpip 系统事件、ARP/邻居表和 IP/MAC 异常判断冲突风险，支持摘要复制与 CSV 导出。 +- 设备发现：支持当前网卡安全限速局域网扫描，发现在线/ARP 可见设备，展示 IP、MAC、主机名、厂商、来源网卡、延迟与发现方式，支持设备清单复制与 CSV 导出。 ## 运行 diff --git a/assets/app.ico b/assets/app.ico new file mode 100644 index 0000000000000000000000000000000000000000..64d661c743de3dc3e1c2692781ab7c0e58939119 GIT binary patch literal 8827 zcmd6M2T)X9knSC3fFTbVMPNWO0z;A@IU|yD2FajgNkbSiNK}%9At?e%PJ%>9Dj6k8 z&XRLZ%m4ef-tN}h{a?L$`|5Rd)ji+s>No3jh#^Ke!1K0Pf=gfTZLfjB{(7BLV;v>JK)D0>B#iU+cdF z4)EUkE71Z#l)9<{0q)&fPk=y4QC9Qz`Bwxm=&e6McW`3?02l;HvQpaKnX!p2j@nYN zuM$-|3T~ABjCNcUI1bWJUP7-6H6x!|5U|M!k+`btFXXBIfa~JS`4uojd-zv zc%S&qXk0HfnB9WwOQgA0h)Uz@Hr7y(Ra$-)B*HsPJR@%u7K>}Uy@}8iojcfI>*g-7 z$4nV%2v3$9f6%^V8(h6iz|pEq%XXb7WaI6UFX>1;4CN@Gr);0{AX+qg3@dx)<^M7w zHE#E*n0eKbV@sfpaNuKodyP=!tyhp0VQ!lYn;o;x(p{mNxT+V<;CwioTG-v&Fo)s+qT7@=_UEW{j%-Lt<&u|BR<=>%_2e& z{aR}Hf_n5Y?iS6X4dbG3-p!HEgIr_DRwS>q!c8viYh)?nDh@jH6phX`#`sDbU#eFT zdW1*e+iz(R1s}ezkogMQ(k=WY%sidz(yvMyRn|PvDhswdKiG{p^1EJgmYn&*tEttL zew!bFlANk+skC|U|00Y2%3G?xvuMss>=XdNM1QiVBzeG5TaWCk#Oln64KrWj69)PU za}_-~MVOQnt3eG#ZkL>t%Vewx+2b(ybUh{Ig0vvlRT!S&HR$=P;OckNg$?}ADPD&{ zz6z-(rkyN)^LOTzFhe*rXPkJSOzoUx?FgTU-nrVW`w?C)F?gIYb|E%}?FQm8zzANF zmk`Gz7%HDx*?Ub%K)PQsF{$VT;SUI;kA=;?tvJUnyJ?I(%@4JOgTE&?a85naCN_fw zp3(@Yt0Xl%gCsAx3pSR%_(Wk3Stl5-(1TZW8ZbkSC_*r>8iQ-^=TSY(?BhJ8=nfm5 z5qG?RXi6_~XjM^DN%S!m`xIXZghL_3PqgP2GGPt+K{uoG^Q7M{Li!1ybsjW(!$ks@ zPeYHADV2S)u6e@+)SX*9pqHmAwgYV>Q%L-<1zLz_CDegd3ZJ1n|GP}+L1_zpT&^|> z&iK7`UT3<~n%5eyPG5^qNS!w`1?Od6G1g0R`WOpVqUMhU?>mwgwqph1a}PU$<|nSy z(!}kA-J5%hj0CbSu{txEF-i*sTO!JALVJjkQk`?=3W9g~TLm;86yKOw_~YJvmDkh$ zB!*Q+7=fj0_N7f?w{z8Hm51y={a2ObN4|`Q%m{NHez{2sArjk>1eHbh<^*@oEw__C zlo)Ql=+xDBm#s&M9{raFxy_g(BHDssE4vhnEP_;=?L%}aLYGH(UF~@B&=c>sQv!s+ zrW;9B3g+9g4)G0zD=3To!~}&{fvy5GtyrcC0gf;ePTwlDX9P4RdA=*G-|3a8igRYR zV>BeLaV_TCL$(Xr1Vi$AG2Cblm+cQvIc0;?GoQI^wzcqEc4}pGZ3WYUVl87RH7+Dp zCOOh$K-hw`>jWzO1?|I&VBK2m1EGi)7hJa z-qz8|pE~MH6?=A12lh4a;CO4MAVebaUPC1c3ukChzAadhRZ*#3Aq0{KXuxq3G^G-; z7#};pG;ncY_N;}9a@Y&)Brl0088w49JA~o}Ru87P3Dc)*+*hl1xZ()zqmQ?z8`5qr zS8rDOzMoEMb>G-E2P(x0y(0wWxX!-E7G(1ifeAuU@S#)Ua`P{kKM2$Lk(U=Vp}(Z^ zaF|t1W%Dd^*5*M&WKkFo%lwrW3yAg^1=g?zdr-1R?IpqNBsdF>_Z=jU67DEUUa5|T zfZ$DKO^D6*xXl?ypTo0zxH)6|Yd5dz0w6`wpE;yh=6vM>vjIq(gWnDrx&CY21RZjIFo7`AgJ z2}xqymiulW(e5^;bsp|o2m&_xR1rcxJ{X#8^VpKW7?Rt%KkauRGh&M6u$+mOT2UB6 zH7!6GR=tbdGHq2Os;TENDrV%X2w3O_LeeNU?eQ6}@>&7`vIPP(Vp-a>RIUB;da{I3 z7mLGkwnzwyl4N{))!B`2>!9W5y~>`-oe6I?k;f${c50n0n6EiQ-lZD3G>w+eg`^kx z&;JzGP46yRmHBW0N|liK!rc;USBk>)mwM_xSYu)x`ZsGD-u4mL^L4?G_f_$-WUJUJ&A z4ai*R&cD-EDGq=Ic!9}>y5Gl&*Qhu~Gm9C&ZnV92+`C%2yXBVQ#$+M(IKp%zV0X-l zfVBZF(4xQTvu6D1{ij*ApSsIUv9ku-9zqntKZxVr(bZab$yA)oa<&@s?Rbv6#)O?e zi>{cUFlBT!tQGGTniobj-Go)T&^Y8vg9+w`_`JL~H)l^(*&Izp78I9H410ZjfDKOX zdCNY&t;!>nH|Sm)n~^1!q|YIW?Gy5{bp0{7wAh z=DNy;(S?S#l@Bz@qV8ZDxN*Wo&v1II4;ZA#la&4|7{6wyky54sr?zs(+JI6!UKXGF zG0nT@hGWgaIX@m@&~)i<(`*?@tjfqzLf5XG3+)Pq$S=!n4%$+q6VOXo*L))Ele8xh zQ3*Hw;rM+D&Nmp!`0h@ISg+&%fgb)E{}6xEgC53N1OQ-${-KAFw11@sPv-&T>Vd)n zjN@H-u#);a*&KBc2Cgi25lWpQK~sYsm;ElI9Taon?OQ1ZYer4UXjw2rR45+aoqY{R zWFl;}zrJkR-_G;owb;e>Lc_RM{i`hhhKrf)-i)&w+ZiNpl}j&g*HbEYo|yW&X*wLD z_;C<^xs-^@aQHZ*ddHz93O@7X!(|~47i_d#qiNB~2_d{Jn&Gdw2LJ)!Jvnmxk2L>>{QiN<{4CAdCwm}kTL!5%vt`hQ0Z}mN%BBF zB5^iz_PMXgbIGQON#&3`3GAr{;JB*%sd#((>6plFBF>0pj=ToBDL~>n*pKe< z6;@8Z{>DF(h!scvzQbD6IG7Ww45M~r}X$r2)53AJcaR)>}bj{4qvs|k4f4-U3fE8%_`rG4jX9Km3lLg+xQQ1ZN5@`3j zV-~71nvV1wGOz|jA~T|Z(xMb{Q~IU8%K&k+fAVKk(UiO`cC~Rhb=%;DaCNR4| zMXi_4udHh772$IdF22rCULCjev}h$K=ug*2+1O;y7~%Z^y-xFs=R#N< zXem)LiXXc}cRbL@ImhV%Q!lb*qtLfoXY>zG{-Pau+j$+?SIqgj-_f5Lj zlOT7=`xe$NFzmGPRr(ta_{yu!QAUOzQUcGxyBp}(IFKmuSz-+Zb@A^GVxGoVS)weuK7xo5L zi@OpeZTBGoHDR@2Dplu8U79KwqA55M%&s0zZ9V@(7H#moViX^Tol9jSXL>~WYN3^e ztYdv%SA~?)4kO~EUBQ{z+BM~|Tqr9L*uVH{LHvV7D~n^QJ^?k^J^Gj7kOik7?6Gti zJNKTRbbyb$TDbVxjVlCRnjDKpI7oUb719sen|z#KU3#!>MOq=yLC-!kX~L1 zcjUwgG!6gMQr&B3^Bj*}S)H$!)_C=sQ7X>5G;6mEGfs^zbvVfI*>hu*ou<}MTG=S+ z-D${z2${5<2z`*{Li3MxSWsP)7eDPc)c6&%0ZG>B6UW^Ct715Qi&K~KXfDWl0JLa2NKdFgazleE4BxFO3 z-m-i2%}KI?i@D5Vq(pkk3S}b`Lg1*KA};R{#@ihtp0Vrhespi)l!i#;-GNe5=fj?; z4S7FBdmYR%g(tS$hxoY>QjFJ={sZ}fkzXZm*(jWclFtE?R{I*;1^^152mg@wvQdlkk}TQvjQ8eM^0XEBhZ5ZNx2gve*YFy!DuP_EIKyLA_AOllYiZBQ0qk9!;BKDgFMc10_nN{}LXV zsY(B|7KGsM?=*;sPN;`;obu)IdO|MB9Lzv$nOJrU&2RibSZB!|WJH-%EVZq;~S4TM0&qE$P+h4HC4_Xpubg%)hNw#&yF zn;_q;>&65V)MC55t?eN)BXzSldM%r7b3ss~&CEI(V78)VxB;GimR%#g8z#o0uj zBxj_I**1k1JEFNapYp36jm8_Fe%%IRHaq#~Af#Owubv%3c|( z!8&!e<(zmjX7?Alh;pR(Y4F}6;oyfOG4l)YLHvH?*Go~9ERu2aYMpG+3Tud%ZmR%- zeRB5{iTB&-n|Q<9HO(lK3X18%z_f7?q{y zfk>;q=*h`6?u0Vs9@U?n!PRBodZYL3F=8TH^n>A0Py2G6x7hMaWJi0R7(D7R$}&^N z{VWOwx4C>+P4mwE*n=?8)Bkl0m+~lGv2aQ*74BlU3T+2-XK($65;1?5xycJ*m;O5U zW5nXeCw%lM&2^PZ0zwKtl3rTJhOMoo>NkRs&2xf@w#Yrn4dP3Rc4weFrhqqD6(G=Z z*4)k_7AMrbjzBw=7egWUIW%!2Yur zN~O@G^bK~zY|l50w@|}57HktL$tzFKWV)W^jl!KLmB9g77Cv#00%dAC|G<*Zo9XD0 zX_OC8&`7~n1lLCpzg%IBU$t~f(m;?5iWtvVG)5=cmR`tvbZFOKhk&+r{t)66FZdG}oK-(%=;dg4wlblj6lSNTt8H zr>rdQ9WiU9?IWJkYpcQ@XrO&$t<%A`^n*r5ke@GD>dcVUEOI{76+7n*^1!otJBBNa zhlERd`#ac$QS#;cBQeQd2+I8Z33P3LCr)+A*wh#Odas4L| zhrGQ?wT?O3DQU#FNSz=Mfg~F~dZyxu(Pa=lOl;z)!df z6j;c5UeqoaDi1dbOx84F!#UdmwEVK(i2fk_v5sw5^bP!MU%1K1ho}|R=7)!Z((sPI z#CZgU)7F8_CtWGZmG(&hcAJX?>BgHWokP}gS`;{aQ$_kXd~!FOs@inVr$=#@1m#0C zh^~EH;9X<|v}U!Uim`>-a>5@_*F0N6Vz*WdT^qN=ymd5mcrXMc4F%ZGLb?LDAMInv zi9vVRhUY+euoCShqiP8n$o|PWS#LNp{g(p~z#@0yvz#ggv^gh&FZq2!xuhYk;StFD zqEmDXF)GVjx(fxkV2?+wlV!vUh38J@KZorA3b_j==(9^Tx~m|b`{e>(W=eX-Tf}7> ztux6(V-(Y-pJE(SO#}yi>ScEj1ER;(aq^@v62%+5jbyaK->69FtOy=JUzNWt*JjE#*`g~Em-6eX?uq`+47wS_Z+XddP}Awe8n&#hE4{)=K9>*5QFe#wy{md59ALh3m#nXzHLL<2rd(_Tp%U zC$rL&K_GT**rWE^RNfTiVz`60R$Xx^h;?tjOy@5tJ5e20KDc6a+BGMNy4SKcK=U{+ zAiOZ!tfAjjeg-Ia*g-F*t{t4w$ zz?cu4#vb=yc(YJlG);H*h0n^x(@R|tBAe2s%zWh| z@9E+Wa~C{cl=V=h1dP(@35WFs7-Xp&4lH2+`U>Kx^dq4_n;JsYPu0~d9kSl-nX~h$ zSzkc)5gO^K!I5=ggx6YG;I5ZKkMup;-v&1XX?0Or!+uY?2O>fGdmh^kK0wHH>Nfp4 zi9jn-mLkfrn_GIHO^ou>>#$PWJw5&kpP8_eergoTt4{TFe>Fa*+-!n$ z@U*)IH-+yTbKu6I*~fidJF+8pfPDsut=t;sAXBKTcTG|BSI?8macFY;?Q+DU{_g|% zcTxREa89R;!Tuf_e$)3~&ljLuGYI&5zA$zxw7LC>aP~jui|GoL`x<1DLp$V{Lgwxy zB%}-^hB?r?IoM0IQ>_$Ic&d@~xscyMDp4r{zm9rc29%x~sMUHL$seB-kJ_g0fECYq zj(}DV2+;Q5q^j>}=MGm~*NNu}XJ|{ijK=}^lmBmY)2hQ{RwBNB(pk`w&D8TFkJqZ# zj#fN^`XK-%{Xf#48fG2bkFx)B^qdTBc+2 zy(dcE(*}X-ASeXrKWh2N`R2ZVH%Qf)_N&)^^3Iom9p)=&T=xZE*JS(18$4EwQd=J)+$Hz_vvPBmcBpO#Njwkyh{F|7nK$35Oe^C8v1nldJi*x=XsGm9b z@1vyxc!-Fb2dBu0>=Q2o1kL|W{KELv!0g6KPH)sd-tWFGDCO|c1=F~*pjLb<=otWG k{=cfVhS?^cg=u9c{$C`uxyKcj=_-H4=hUBL|CTZS1rLHR1ONa4 literal 0 HcmV?d00001 diff --git a/core/Function/device_discovery_fun.py b/core/Function/device_discovery_fun.py new file mode 100644 index 0000000..1d8f26f --- /dev/null +++ b/core/Function/device_discovery_fun.py @@ -0,0 +1,618 @@ +import concurrent.futures +import csv +import ipaddress +import json +import re +import socket +import threading +import time +from dataclasses import asdict, dataclass, field +from typing import Callable, Optional + +from core.Function.common import run_hidden +from core.Function.loop_fun import normalize_mac +from core.Function.network_fun import NetworkManager + + +OutputCallback = Callable[[str, Optional[str]], None] +DoneCallback = Callable[[], None] +StatusCallback = Callable[[dict], None] +ResultCallback = Callable[[dict], None] + + +ALL_ADAPTERS = "全部活动网卡" + + +OUI_VENDOR_MAP = { + "00-05-69": "VMware", + "00-0C-29": "VMware", + "00-1C-14": "VMware", + "00-50-56": "VMware", + "00-15-5D": "Microsoft Hyper-V", + "00-03-FF": "Microsoft", + "00-1A-A0": "Dell", + "00-1B-21": "Intel", + "00-1E-67": "Intel", + "00-21-5C": "Intel", + "00-24-D7": "Intel", + "3C-A8-2A": "Intel", + "48-7D-2E": "TP-Link", + "50-C7-BF": "TP-Link", + "60-E3-27": "TP-Link", + "A0-F3-C1": "TP-Link", + "B0-A7-B9": "TP-Link", + "D8-07-B6": "TP-Link", + "00-1D-0F": "Cisco", + "00-22-BD": "Cisco", + "00-25-9C": "Cisco", + "A4-18-75": "Cisco", + "F4-4E-05": "Cisco", + "00-16-EA": "HPE", + "00-1F-29": "HPE", + "3C-A8-2A": "HPE/Intel", + "00-1E-C2": "Apple", + "00-25-00": "Apple", + "3C-15-C2": "Apple", + "7C-D1-C3": "Apple", + "A4-C3-F0": "Apple", + "F0-18-98": "Apple", + "18-C0-4D": "Realtek", + "52-54-00": "QEMU/KVM", + "08-00-27": "VirtualBox", + "BC-24-11": "Proxmox/QEMU", +} + + +@dataclass +class DeviceDiscoveryOptions: + scan_range: str = "" + workers: int = 64 + timeout_ms: int = 500 + max_hosts: int = 254 + + +@dataclass +class DeviceInfo: + ip: str + mac: str + hostname: str + vendor: str + adapter: str + interface_index: str + latency_ms: float = 0.0 + method: str = "ARP 可见" + note: str = "" + checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")) + + def to_dict(self) -> dict: + data = asdict(self) + data["latency_ms"] = round(self.latency_ms, 1) + return data + + +class DeviceDiscovery: + def __init__( + self, + output: OutputCallback, + done: Optional[DoneCallback] = None, + status: Optional[StatusCallback] = None, + result: Optional[ResultCallback] = None, + ): + self.output = output + self.done = done or (lambda: None) + self.status = status or (lambda _stats: None) + self.result = result or (lambda _row: None) + self.network = NetworkManager(lambda _text, _tag=None: None) + self.stop_event = threading.Event() + self.worker = None + self.last_results: list[dict] = [] + self.last_summary = "" + + def get_adapter_choices(self) -> list[str]: + adapters = self._active_adapters(self.network.get_network_info()) + return [ALL_ADAPTERS] + [adapter["name"] for adapter in adapters] + + def default_scan_range(self, adapter_name: str = ALL_ADAPTERS) -> str: + adapters = self._select_adapters(adapter_name) + if not adapters: + return "" + return adapter_to_safe_range(adapters[0]) + + def start_discovery(self, adapter_name: str = ALL_ADAPTERS, options: Optional[dict] = None) -> None: + if self.is_running(): + raise RuntimeError("设备发现正在运行，请先停止当前任务") + + discovery_options = self.normalize_options(options) + adapters = self._select_adapters(adapter_name) + if not adapters: + raise ValueError("没有找到可扫描的活动网卡") + + targets = self.build_scan_targets(adapters, discovery_options) + if not targets: + raise ValueError("没有可扫描的目标地址") + + self.stop_event.clear() + self.last_results = [] + self.last_summary = "" + self.output( + f"开始局域网设备发现: {adapter_name or ALL_ADAPTERS}，" + f"目标 {len(targets)} 个，并发 {discovery_options.workers}，超时 {discovery_options.timeout_ms}ms\n", + "muted", + ) + self.status(self._status("扫描中", adapter_name or ALL_ADAPTERS, describe_targets(targets), len(targets), 0, 0, 0)) + self.worker = threading.Thread(target=self._run_discovery, args=(adapters, targets, discovery_options), daemon=True) + self.worker.start() + + def _run_discovery(self, adapters: list[dict], targets: list[str], options: DeviceDiscoveryOptions) -> None: + started = time.perf_counter() + ping_results: dict[str, dict] = {} + + try: + ping_results = self.scan_targets(targets, options, started) + neighbors = self.get_neighbors() + devices = self.build_devices(adapters, targets, ping_results, neighbors) + + for index, device in enumerate(sorted(devices, key=lambda item: ip_sort_key(item.ip)), start=1): + if self.stop_event.is_set(): + break + row = device.to_dict() + self.last_results.append(row) + self.result(row) + self.status( + self._status( + "汇总中", + device.adapter, + describe_targets(targets), + len(targets), + len(targets), + index, + time.perf_counter() - started, + ) + ) + + self.last_summary = build_summary(self.last_results, stopped=self.stop_event.is_set()) + if self.stop_event.is_set(): + self.output("\n设备发现已停止\n", "warning") + else: + self.output("\n设备发现完成\n", "success") + self.output(self.last_summary + "\n", "success") + self.status( + self._status( + "已停止" if self.stop_event.is_set() else "已完成", + "", + describe_targets(targets), + len(targets), + len(targets), + len(self.last_results), + time.perf_counter() - started, + ) + ) + except Exception as exc: + self.output(f"\n设备发现失败: {exc}\n", "error") + self.status(self._status("失败", "", describe_targets(targets), len(targets), len(ping_results), 0, time.perf_counter() - started)) + finally: + self.done() + + def scan_targets(self, targets: list[str], options: DeviceDiscoveryOptions, started: float) -> dict[str, dict]: + results = {} + completed = 0 + + def task(ip: str) -> dict: + if self.stop_event.is_set(): + return {"ip": ip, "ok": False, "rtt": 0.0} + return ping_once(ip, options.timeout_ms) + + with concurrent.futures.ThreadPoolExecutor(max_workers=options.workers) as executor: + futures = {executor.submit(task, ip): ip for ip in targets} + for future in concurrent.futures.as_completed(futures): + completed += 1 + ip = futures[future] + if self.stop_event.is_set(): + for item in futures: + item.cancel() + break + try: + result = future.result() + except Exception: + result = {"ip": ip, "ok": False, "rtt": 0.0} + results[ip] = result + if completed % 20 == 0 or completed == len(targets): + self.status(self._status("扫描中", ip, describe_targets(targets), len(targets), completed, 0, time.perf_counter() - started)) + return results + + def build_devices( + self, + adapters: list[dict], + targets: list[str], + ping_results: dict[str, dict], + neighbors: list[dict], + ) -> list[DeviceInfo]: + adapter_by_index = {str(adapter.get("interface_index", "")): adapter for adapter in adapters} + adapter_by_ip = {adapter.get("ipv4", ""): adapter for adapter in adapters} + target_set = set(targets) + local_ips = {adapter.get("ipv4", "") for adapter in adapters} + gateways = {adapter.get("gateway", "") for adapter in adapters if adapter.get("gateway")} + devices: dict[str, DeviceInfo] = {} + + for adapter in adapters: + local_ip = adapter.get("ipv4", "") + if not local_ip: + continue + devices[local_ip] = DeviceInfo( + ip=local_ip, + mac=normalize_mac(adapter.get("mac", "")), + hostname=resolve_hostname(local_ip), + vendor=vendor_name(adapter.get("mac", "")), + adapter=adapter.get("name", ""), + interface_index=str(adapter.get("interface_index", "")), + latency_ms=ping_results.get(local_ip, {}).get("rtt", 0.0), + method="在线" if ping_results.get(local_ip, {}).get("ok") else "本机", + note="本机", + ) + + for item in neighbors: + ip = item.get("ip", "") + mac = normalize_mac(item.get("mac", "")) + if ip not in target_set and ip not in gateways and ip not in local_ips: + continue + if not is_valid_device_mac(mac) or is_multicast_or_broadcast_ip(ip): + continue + adapter = adapter_by_index.get(str(item.get("ifIndex", ""))) or adapter_by_ip.get(item.get("interface_ip", "")) or adapter_for_ip(ip, adapters) + if not adapter: + continue + ping = ping_results.get(ip, {}) + method = "在线" if ping.get("ok") else "ARP 可见" + note = [] + if ip in local_ips: + note.append("本机") + if ip in gateways: + note.append("网关") + devices[ip] = DeviceInfo( + ip=ip, + mac=mac, + hostname=resolve_hostname(ip), + vendor=vendor_name(mac), + adapter=adapter.get("name", ""), + interface_index=str(adapter.get("interface_index", "")), + latency_ms=ping.get("rtt", 0.0), + method=method, + note="; ".join(note), + ) + + for ip, ping in ping_results.items(): + if not ping.get("ok") or ip in devices: + continue + adapter = adapter_for_ip(ip, adapters) + if not adapter: + continue + # Ping succeeded but no valid ARP MAC was visible; keep it out of + # the asset list unless it is a known local/gateway address. + if ip not in gateways and ip not in local_ips: + continue + devices[ip] = DeviceInfo( + ip=ip, + mac="", + hostname=resolve_hostname(ip), + vendor="未知", + adapter=adapter.get("name", ""), + interface_index=str(adapter.get("interface_index", "")), + latency_ms=ping.get("rtt", 0.0), + method="在线", + note="网关" if ip in gateways else "本机", + ) + + return list(devices.values()) + + def get_neighbors(self) -> list[dict]: + script = r""" +$ErrorActionPreference = "SilentlyContinue" +Get-NetNeighbor -AddressFamily IPv4 | Select-Object ifIndex,IPAddress,LinkLayerAddress,State | ConvertTo-Json -Depth 4 -Compress +""" + try: + result = run_hidden(["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script], timeout=10) + if result.returncode != 0: + raise RuntimeError(result.stdout.strip()) + data = extract_json(result.stdout) + if isinstance(data, dict): + data = [data] + return [normalize_neighbor(item) for item in data] + except Exception as exc: + self.output(f"PowerShell 邻居表读取失败，尝试 arp -a: {exc}\n", "warning") + return self.get_neighbors_from_arp() + + def get_neighbors_from_arp(self) -> list[dict]: + result = run_hidden(["arp", "-a"], timeout=10) + if result.returncode != 0: + return [] + neighbors = [] + current_interface = "" + for line in result.stdout.splitlines(): + header = re.search(r"Interface:\s+([^\s]+)", line, re.IGNORECASE) + if header: + current_interface = header.group(1) + continue + match = re.match(r"\s*(\d+\.\d+\.\d+\.\d+)\s+([0-9a-fA-F-]+)\s+(\w+)", line) + if match: + neighbors.append( + { + "ifIndex": "", + "interface_ip": current_interface, + "ip": match.group(1), + "mac": normalize_mac(match.group(2)), + "state": match.group(3), + } + ) + return neighbors + + def build_scan_targets(self, adapters: list[dict], options: DeviceDiscoveryOptions) -> list[str]: + if options.scan_range: + return parse_target_range(options.scan_range, options.max_hosts) + targets = [] + for adapter in adapters: + targets.extend(parse_target_range(adapter_to_safe_range(adapter), options.max_hosts)) + return list(dict.fromkeys(targets))[: options.max_hosts] + + def stop_discovery(self) -> None: + if not self.is_running(): + raise RuntimeError("当前没有正在运行的设备发现") + self.stop_event.set() + self.output("\n正在停止设备发现...\n", "warning") + + def export_results(self, path: str) -> None: + if not self.last_results: + raise RuntimeError("还没有可导出的设备发现结果") + fields = ["ip", "mac", "hostname", "vendor", "adapter", "interface_index", "latency_ms", "method", "note", "checked_at"] + with open(path, "w", newline="", encoding="utf-8-sig") as file: + writer = csv.DictWriter(file, fieldnames=fields) + writer.writeheader() + for row in self.last_results: + writer.writerow({field_name: row.get(field_name, "") for field_name in fields}) + + def copy_summary(self) -> str: + if not self.last_results: + return self.last_summary + return "\n".join( + f"{row.get('ip', '')}\t{row.get('mac', '')}\t{row.get('hostname', '')}\t{row.get('vendor', '')}\t{row.get('method', '')}" + for row in sorted(self.last_results, key=lambda item: ip_sort_key(item.get("ip", ""))) + ) + + def normalize_options(self, options: Optional[dict]) -> DeviceDiscoveryOptions: + options = options or {} + return DeviceDiscoveryOptions( + scan_range=str(options.get("scan_range", "")).strip(), + workers=clamp_int(options.get("workers", 64), 1, 256, "并发数"), + timeout_ms=clamp_int(options.get("timeout_ms", 500), 100, 10000, "超时"), + max_hosts=clamp_int(options.get("max_hosts", 254), 1, 254, "最大扫描地址数"), + ) + + def _select_adapters(self, adapter_name: str) -> list[dict]: + adapters = self._active_adapters(self.network.get_network_info()) + if not adapter_name or adapter_name == ALL_ADAPTERS: + return adapters + return [adapter for adapter in adapters if adapter.get("name") == adapter_name] + + def _active_adapters(self, adapters: list[dict]) -> list[dict]: + active = [] + for adapter in adapters: + if not adapter.get("ipv4"): + continue + status = str(adapter.get("status", "")).lower() + if "disconnect" in status or "断开" in status: + continue + active.append(adapter) + return active + + def _status(self, state: str, current: str, scan_range: str, total: int, scanned: int, found: int, elapsed: float) -> dict: + return { + "state": state, + "current": current, + "scan_range": scan_range, + "total": total, + "scanned": scanned, + "found": found, + "elapsed": elapsed, + } + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) + + +def ping_once(ip: str, timeout_ms: int) -> dict: + try: + result = run_hidden(["ping", ip, "-n", "1", "-w", str(timeout_ms)], timeout=max(2, timeout_ms / 1000 + 2)) + output = result.stdout + if re.search(r"\bTTL=", output, re.IGNORECASE): + match = re.search(r"(?:time|时间)[=<]?\s*(\d+(?:\.\d+)?)\s*(?:ms|毫秒)", output, re.IGNORECASE) + return {"ip": ip, "ok": True, "rtt": float(match.group(1)) if match else 0.0} + except Exception: + pass + return {"ip": ip, "ok": False, "rtt": 0.0} + + +def parse_target_range(text: str, max_hosts: int = 254) -> list[str]: + raw = text.strip() + if not raw: + return [] + targets = [] + for part in re.split(r"[,，;\s]+", raw): + item = part.strip() + if not item: + continue + if "/" in item: + network = ipaddress.ip_network(item, strict=False) + targets.extend(str(ip) for ip in network.hosts()) + elif re.match(r"^\d{1,3}(?:\.\d{1,3}){3}-\d{1,3}$", item): + prefix, tail = item.rsplit(".", 1) + start_text, end_text = tail.split("-", 1) + start = int(start_text) + end = int(end_text) + if start > end: + raise ValueError("IP 范围起始值不能大于结束值") + targets.extend(str(ipaddress.ip_address(f"{prefix}.{value}")) for value in range(start, end + 1)) + elif re.match(r"^\d{1,3}(?:\.\d{1,3}){3}-\d{1,3}(?:\.\d{1,3}){3}$", item): + start_text, end_text = item.split("-", 1) + start_ip = ipaddress.ip_address(start_text) + end_ip = ipaddress.ip_address(end_text) + if start_ip.version != end_ip.version: + raise ValueError("IP 范围两端必须是同一 IP 版本") + if int(start_ip) > int(end_ip): + raise ValueError("IP 范围起始值不能大于结束值") + targets.extend(str(ipaddress.ip_address(value)) for value in range(int(start_ip), int(end_ip) + 1)) + else: + targets.append(str(ipaddress.ip_address(item))) + return list(dict.fromkeys(targets))[:max_hosts] + + +def adapter_to_safe_range(adapter: dict) -> str: + ip = adapter.get("ipv4", "") + if not ip: + return "" + prefix = adapter.get("prefix_length") + if prefix in ("", None): + prefix = netmask_to_prefix(adapter.get("netmask", "")) or 24 + try: + prefix = int(prefix) + if prefix < 24: + prefix = 24 + return str(ipaddress.ip_network(f"{ip}/{prefix}", strict=False)) + except Exception: + parts = ip.split(".") + return ".".join(parts[:3]) + ".0/24" if len(parts) == 4 else "" + + +def netmask_to_prefix(netmask: str) -> Optional[int]: + if not netmask: + return None + try: + return ipaddress.IPv4Network(f"0.0.0.0/{netmask}").prefixlen + except Exception: + return None + + +def normalize_neighbor(item: dict) -> dict: + return { + "ifIndex": str(item.get("ifIndex", "")), + "interface_ip": "", + "ip": str(item.get("IPAddress", "")), + "mac": normalize_mac(item.get("LinkLayerAddress", "")), + "state": str(item.get("State", "")), + } + + +def resolve_hostname(ip: str) -> str: + try: + old_timeout = socket.getdefaulttimeout() + socket.setdefaulttimeout(0.8) + try: + return socket.gethostbyaddr(ip)[0] + finally: + socket.setdefaulttimeout(old_timeout) + except Exception: + pass + try: + result = run_hidden(["nbtstat", "-A", ip], timeout=2) + for line in result.stdout.splitlines(): + match = re.match(r"\s*([^\s<]+)\s+<00>\s+UNIQUE", line, re.IGNORECASE) + if match: + return match.group(1).strip() + except Exception: + pass + return "" + + +def vendor_name(mac: str) -> str: + normalized = normalize_mac(mac) + if len(normalized) < 8: + return "未知" + return OUI_VENDOR_MAP.get(normalized[:8], "未知") + + +def build_summary(rows: list[dict], stopped: bool = False) -> str: + if stopped: + return "设备发现已停止，当前结果仅代表已完成扫描。" + if not rows: + return "未发现有效局域网设备。" + online = len([row for row in rows if row.get("method") == "在线"]) + arp = len([row for row in rows if row.get("method") == "ARP 可见"]) + gateway = len([row for row in rows if "网关" in row.get("note", "")]) + local = len([row for row in rows if "本机" in row.get("note", "")]) + vendors = sorted({row.get("vendor", "未知") for row in rows if row.get("vendor") and row.get("vendor") != "未知"}) + lines = [ + "==== 局域网设备发现摘要 ====", + f"发现设备: {len(rows)} 在线: {online} ARP 可见: {arp} 本机: {local} 网关: {gateway}", + ] + if vendors: + lines.append("识别厂商: " + ", ".join(vendors[:8])) + else: + lines.append("未识别到已知厂商，可能需要后续扩展 OUI 数据。") + return "\n".join(lines) + + +def describe_targets(targets: list[str]) -> str: + if not targets: + return "" + if len(targets) == 1: + return targets[0] + return f"{targets[0]} - {targets[-1]}" + + +def adapter_for_ip(ip: str, adapters: list[dict]) -> Optional[dict]: + try: + address = ipaddress.ip_address(ip) + except ValueError: + return None + for adapter in adapters: + local_ip = adapter.get("ipv4", "") + if not local_ip: + continue + prefix = adapter.get("prefix_length") or netmask_to_prefix(adapter.get("netmask", "")) or 24 + try: + network = ipaddress.ip_network(f"{local_ip}/{prefix}", strict=False) + if address in network: + return adapter + except Exception: + continue + return adapters[0] if adapters else None + + +def is_valid_device_mac(mac: str) -> bool: + normalized = normalize_mac(mac) + if not normalized or normalized == "00-00-00-00-00-00": + return False + if normalized.startswith("FF-FF-FF") or normalized.startswith("01-00-5E"): + return False + return bool(re.match(r"^[0-9A-F]{2}(?:-[0-9A-F]{2}){5}$", normalized)) + + +def is_multicast_or_broadcast_ip(ip: str) -> bool: + try: + parsed = ipaddress.ip_address(ip) + return parsed.is_multicast or str(parsed).endswith(".255") or str(parsed) == "255.255.255.255" + except ValueError: + return True + + +def ip_sort_key(value: str) -> tuple: + try: + return (0, int(ipaddress.ip_address(value))) + except ValueError: + return (1, value) + + +def extract_json(text: str): + output = text.strip() + json_start = min([idx for idx in (output.find("["), output.find("{")) if idx >= 0], default=-1) + if json_start < 0: + raise RuntimeError("未获取到 JSON 输出") + data = json.loads(output[json_start:]) + return [] if data is None else data + + +def clamp_int(value, min_value: int, max_value: int, label: str) -> int: + try: + number = int(str(value).strip()) + except ValueError as exc: + raise ValueError(f"{label}必须是整数") from exc + if number < min_value or number > max_value: + raise ValueError(f"{label}必须在 {min_value}-{max_value} 之间") + return number diff --git a/core/Function/dns_diag_fun.py b/core/Function/dns_diag_fun.py new file mode 100644 index 0000000..6c8eaf1 --- /dev/null +++ b/core/Function/dns_diag_fun.py @@ -0,0 +1,584 @@ +import concurrent.futures +import csv +import ipaddress +import json +import re +import subprocess +import threading +import time +from dataclasses import asdict, dataclass, field +from typing import Callable, Optional + +from core.Function.common import run_hidden +from core.Function.network_fun import NetworkManager + + +OutputCallback = Callable[[str, Optional[str]], None] +DoneCallback = Callable[[], None] +StatusCallback = Callable[[dict], None] +ResultCallback = Callable[[dict], None] + + +ALL_ADAPTERS = "全部活动网卡" +DEFAULT_DOMAINS = "www.baidu.com,www.qq.com" +DEFAULT_RECORD_TYPES = "A,AAAA,CNAME,MX,TXT" +PUBLIC_DNS_SERVERS = ["223.5.5.5", "114.114.114.114", "8.8.8.8"] +SUPPORTED_RECORD_TYPES = {"A", "AAAA", "CNAME", "MX", "TXT", "NS"} + + +@dataclass +class DnsDiagOptions: + domains: list[str] = field(default_factory=lambda: parse_list(DEFAULT_DOMAINS)) + record_types: list[str] = field(default_factory=lambda: parse_record_types(DEFAULT_RECORD_TYPES)) + dns_servers: list[str] = field(default_factory=list) + timeout_ms: int = 2000 + repeat_count: int = 1 + + +@dataclass +class DnsQueryResult: + domain: str + record_type: str + dns_server: str + adapter: str + status: str + elapsed_ms: float + values: str = "" + error: str = "" + verdict: str = "" + attempt: int = 1 + resolver: str = "Resolve-DnsName" + checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")) + + def to_dict(self) -> dict: + data = asdict(self) + data["elapsed_ms"] = round(self.elapsed_ms, 1) + return data + + +class DnsDiagnostic: + def __init__( + self, + output: OutputCallback, + done: Optional[DoneCallback] = None, + status: Optional[StatusCallback] = None, + result: Optional[ResultCallback] = None, + ): + self.output = output + self.done = done or (lambda: None) + self.status = status or (lambda _stats: None) + self.result = result or (lambda _row: None) + self.network = NetworkManager(lambda _text, _tag=None: None) + self.stop_event = threading.Event() + self.worker = None + self.last_results: list[dict] = [] + self.last_summary = "" + self.local_dns_servers: set[str] = set() + + def get_adapter_choices(self) -> list[str]: + adapters = self._active_adapters(self.network.get_network_info()) + return [ALL_ADAPTERS] + [adapter["name"] for adapter in adapters] + + def default_dns_servers(self, adapter_name: str = ALL_ADAPTERS) -> str: + adapters = self._select_adapters(adapter_name) + servers = collect_adapter_dns(adapters) + return ",".join(list(dict.fromkeys(servers + PUBLIC_DNS_SERVERS))) + + def start_diagnosis(self, adapter_name: str = ALL_ADAPTERS, options: Optional[dict] = None) -> None: + if self.is_running(): + raise RuntimeError("DNS 诊断正在运行，请先停止当前任务") + + adapters = self._select_adapters(adapter_name) + if not adapters: + raise ValueError("没有找到可用于 DNS 诊断的活动网卡") + + diag_options = self.normalize_options(options, adapters) + tasks = build_tasks(diag_options) + if not tasks: + raise ValueError("没有可执行的 DNS 查询任务") + + self.stop_event.clear() + self.last_results = [] + self.last_summary = "" + self.local_dns_servers = set(collect_adapter_dns(adapters)) + current_dns = ",".join(sorted(self.local_dns_servers)) or "未读取到本机 DNS" + self.output( + f"开始 DNS 诊断: {adapter_name or ALL_ADAPTERS}，" + f"域名 {len(diag_options.domains)} 个，类型 {len(diag_options.record_types)} 个，" + f"DNS {len(diag_options.dns_servers)} 个，重复 {diag_options.repeat_count} 次\n", + "muted", + ) + self.status(self._status("查询中", current_dns, len(tasks), 0, 0, 0)) + self.worker = threading.Thread( + target=self._run_diagnosis, + args=(adapter_name or ALL_ADAPTERS, current_dns, tasks, diag_options), + daemon=True, + ) + self.worker.start() + + def _run_diagnosis(self, adapter_name: str, current_dns: str, tasks: list[dict], options: DnsDiagOptions) -> None: + started = time.perf_counter() + completed = 0 + abnormal = 0 + + try: + workers = min(8, max(1, len(tasks))) + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: + futures = { + executor.submit(self.query_one, adapter_name, task, options.timeout_ms): task + for task in tasks + } + for future in concurrent.futures.as_completed(futures): + if self.stop_event.is_set(): + for item in futures: + item.cancel() + break + completed += 1 + try: + row = future.result().to_dict() + except Exception as exc: + task = futures[future] + row = DnsQueryResult( + domain=task["domain"], + record_type=task["record_type"], + dns_server=task["dns_server"], + adapter=adapter_name, + status="失败", + elapsed_ms=0, + error=str(exc), + verdict="查询失败，建议与其它 DNS 对比", + attempt=task["attempt"], + ).to_dict() + self.last_results.append(row) + if row.get("status") != "正常": + abnormal += 1 + self.result(row) + self.status(self._status("查询中", current_dns, len(tasks), completed, abnormal, time.perf_counter() - started)) + + self.last_summary = build_summary(self.last_results, self.local_dns_servers, stopped=self.stop_event.is_set()) + if self.stop_event.is_set(): + self.output("\nDNS 诊断已停止\n", "warning") + else: + self.output("\nDNS 诊断完成\n", "success") + self.output(self.last_summary + "\n", "success") + self.status( + self._status( + "已停止" if self.stop_event.is_set() else "已完成", + current_dns, + len(tasks), + completed, + abnormal, + time.perf_counter() - started, + ) + ) + except Exception as exc: + self.output(f"\nDNS 诊断失败: {exc}\n", "error") + self.status(self._status("失败", current_dns, len(tasks), completed, abnormal, time.perf_counter() - started)) + finally: + self.done() + + def query_one(self, adapter_name: str, task: dict, timeout_ms: int) -> DnsQueryResult: + if self.stop_event.is_set(): + return DnsQueryResult( + domain=task["domain"], + record_type=task["record_type"], + dns_server=task["dns_server"], + adapter=adapter_name, + status="失败", + elapsed_ms=0, + error="任务已停止", + verdict="任务停止，结果不完整", + attempt=task["attempt"], + ) + + started = time.perf_counter() + domain = task["domain"] + record_type = task["record_type"] + dns_server = task["dns_server"] + try: + values = resolve_with_powershell(domain, record_type, dns_server, timeout_ms) + resolver = "Resolve-DnsName" + except Exception as ps_error: + try: + values = resolve_with_nslookup(domain, record_type, dns_server, timeout_ms) + resolver = "nslookup" + except Exception as ns_error: + elapsed = (time.perf_counter() - started) * 1000 + error = classify_error(f"{ps_error}; {ns_error}") + return DnsQueryResult( + domain=domain, + record_type=record_type, + dns_server=dns_server, + adapter=adapter_name, + status="失败", + elapsed_ms=elapsed, + error=error, + verdict=verdict_for_error(error), + attempt=task["attempt"], + resolver="fallback", + ) + + elapsed = (time.perf_counter() - started) * 1000 + if values: + return DnsQueryResult( + domain=domain, + record_type=record_type, + dns_server=dns_server, + adapter=adapter_name, + status="正常", + elapsed_ms=elapsed, + values="; ".join(values), + verdict="解析成功", + attempt=task["attempt"], + resolver=resolver, + ) + return DnsQueryResult( + domain=domain, + record_type=record_type, + dns_server=dns_server, + adapter=adapter_name, + status="无记录", + elapsed_ms=elapsed, + verdict="未返回该类型记录，建议结合其它记录类型判断", + attempt=task["attempt"], + resolver=resolver, + ) + + def stop_diagnosis(self) -> None: + if not self.is_running(): + raise RuntimeError("当前没有正在运行的 DNS 诊断") + self.stop_event.set() + self.output("\n正在停止 DNS 诊断...\n", "warning") + + def export_results(self, path: str) -> None: + if not self.last_results: + raise RuntimeError("还没有可导出的 DNS 诊断结果") + fields = [ + "domain", + "record_type", + "dns_server", + "adapter", + "status", + "elapsed_ms", + "values", + "error", + "verdict", + "attempt", + "resolver", + "checked_at", + ] + with open(path, "w", newline="", encoding="utf-8-sig") as file: + writer = csv.DictWriter(file, fieldnames=fields) + writer.writeheader() + for row in self.last_results: + writer.writerow({field_name: row.get(field_name, "") for field_name in fields}) + + def copy_summary(self) -> str: + return self.last_summary or "\n".join( + f"{row.get('domain', '')}\t{row.get('record_type', '')}\t{row.get('dns_server', '')}\t" + f"{row.get('status', '')}\t{row.get('elapsed_ms', '')}ms\t{row.get('values', '') or row.get('error', '')}" + for row in self.last_results + ) + + def normalize_options(self, options: Optional[dict], adapters: list[dict]) -> DnsDiagOptions: + options = options or {} + domains = parse_list(str(options.get("domains", "")).strip() or DEFAULT_DOMAINS) + record_types = parse_record_types(str(options.get("record_types", "")).strip() or DEFAULT_RECORD_TYPES) + dns_servers = parse_dns_servers(str(options.get("dns_servers", "")).strip()) + if not dns_servers: + dns_servers = collect_adapter_dns(adapters) + PUBLIC_DNS_SERVERS + dns_servers = list(dict.fromkeys(dns_servers)) + return DnsDiagOptions( + domains=domains, + record_types=record_types, + dns_servers=dns_servers, + timeout_ms=clamp_int(options.get("timeout_ms", 2000), 300, 10000, "超时"), + repeat_count=clamp_int(options.get("repeat_count", 1), 1, 5, "重复次数"), + ) + + def _select_adapters(self, adapter_name: str) -> list[dict]: + adapters = self._active_adapters(self.network.get_network_info()) + if not adapter_name or adapter_name == ALL_ADAPTERS: + return adapters + return [adapter for adapter in adapters if adapter.get("name") == adapter_name] + + def _active_adapters(self, adapters: list[dict]) -> list[dict]: + active = [] + for adapter in adapters: + if not adapter.get("ipv4"): + continue + status = str(adapter.get("status", "")).lower() + if "disconnect" in status or "断开" in status: + continue + active.append(adapter) + return active + + def _status(self, state: str, current_dns: str, total: int, completed: int, abnormal: int, elapsed: float) -> dict: + return { + "state": state, + "current_dns": current_dns, + "total": total, + "completed": completed, + "abnormal": abnormal, + "elapsed": elapsed, + } + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) + + +def build_tasks(options: DnsDiagOptions) -> list[dict]: + tasks = [] + for attempt in range(1, options.repeat_count + 1): + for domain in options.domains: + for record_type in options.record_types: + for dns_server in options.dns_servers: + tasks.append( + { + "domain": domain, + "record_type": record_type, + "dns_server": dns_server, + "attempt": attempt, + } + ) + return tasks + + +def resolve_with_powershell(domain: str, record_type: str, dns_server: str, timeout_ms: int) -> list[str]: + script = f""" +$ErrorActionPreference = "Stop" +$items = Resolve-DnsName -Name {ps_quote(domain)} -Type {ps_quote(record_type)} -Server {ps_quote(dns_server)} -DnsOnly -ErrorAction Stop +$items | Select-Object Name,Type,QueryType,IPAddress,NameHost,NameExchange,Preference,Strings,CharacterStrings | ConvertTo-Json -Depth 5 -Compress +""" + result = run_hidden( + ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script], + timeout=max(3, timeout_ms / 1000 + 2), + ) + if result.returncode != 0: + raise RuntimeError(result.stdout.strip() or "Resolve-DnsName 查询失败") + data = extract_json(result.stdout) + if isinstance(data, dict): + data = [data] + return normalize_record_values(data) + + +def resolve_with_nslookup(domain: str, record_type: str, dns_server: str, timeout_ms: int) -> list[str]: + seconds = max(1, int(round(timeout_ms / 1000))) + try: + result = run_hidden(["nslookup", f"-timeout={seconds}", f"-type={record_type}", domain, dns_server], timeout=seconds + 3) + except subprocess.TimeoutExpired as exc: + raise RuntimeError("nslookup 查询超时") from exc + output = result.stdout.strip() + if result.returncode != 0 and not output: + raise RuntimeError("nslookup 查询失败") + if is_nslookup_error(output): + raise RuntimeError(output) + return parse_nslookup_output(output, record_type) + + +def normalize_record_values(items) -> list[str]: + values = [] + for item in items or []: + if not isinstance(item, dict): + continue + if item.get("IPAddress"): + values.append(str(item["IPAddress"])) + if item.get("NameHost"): + values.append(str(item["NameHost"]).rstrip(".")) + if item.get("NameExchange"): + exchange = str(item["NameExchange"]).rstrip(".") + preference = item.get("Preference") + values.append(f"{preference} {exchange}" if preference not in ("", None) else exchange) + for key in ("Strings", "CharacterStrings"): + text = item.get(key) + if isinstance(text, list): + values.append(" ".join(str(part) for part in text)) + elif text: + values.append(str(text)) + return sorted(dict.fromkeys(value.strip() for value in values if value and value.strip())) + + +def parse_nslookup_output(output: str, record_type: str) -> list[str]: + values = [] + answer_started = False + for raw_line in output.splitlines(): + line = raw_line.strip() + lower = line.lower() + if not line: + continue + if lower.startswith("name:") or line.startswith("名称:") or "canonical name =" in lower or "mail exchanger =" in lower: + answer_started = True + if record_type in {"A", "AAAA"}: + match = re.search(r"(?:address|addresses|地址)\s*:\s*([0-9a-fA-F:.]+)", line, re.IGNORECASE) + if match and answer_started: + value = match.group(1) + if ":" in value or "." in value: + values.append(value) + elif record_type == "CNAME": + match = re.search(r"canonical name\s*=\s*(.+)$", line, re.IGNORECASE) + if match: + values.append(match.group(1).strip().rstrip(".")) + elif record_type == "MX": + match = re.search(r"mail exchanger\s*=\s*(.+)$", line, re.IGNORECASE) + if match: + values.append(match.group(1).strip().rstrip(".")) + elif record_type == "TXT": + match = re.search(r'text\s*=\s*"?(.*?)"?$', line, re.IGNORECASE) + if match: + values.append(match.group(1).strip()) + elif record_type == "NS": + match = re.search(r"nameserver\s*=\s*(.+)$", line, re.IGNORECASE) + if match: + values.append(match.group(1).strip().rstrip(".")) + if answer_started and record_type in {"A", "AAAA"}: + bare_ip = re.match(r"^([0-9a-fA-F:.]+)$", line) + if bare_ip: + values.append(bare_ip.group(1)) + return sorted(dict.fromkeys(value for value in values if value)) + + +def build_summary(rows: list[dict], local_dns_servers: set[str], stopped: bool = False) -> str: + if stopped: + return "DNS 诊断已停止，当前结果仅代表已完成查询。" + if not rows: + return "DNS 诊断未产生结果。" + + total = len(rows) + ok_rows = [row for row in rows if row.get("status") == "正常"] + failed_rows = [row for row in rows if row.get("status") != "正常"] + avg_elapsed = sum(float(row.get("elapsed_ms") or 0) for row in ok_rows) / len(ok_rows) if ok_rows else 0 + lines = [ + "==== DNS 诊断摘要 ====", + f"查询总数: {total} 成功: {len(ok_rows)} 异常: {len(failed_rows)} 成功平均耗时: {avg_elapsed:.1f} ms", + ] + + for (domain, record_type), group in group_by_domain_type(rows).items(): + local = [row for row in group if row.get("dns_server") in local_dns_servers] + public = [row for row in group if row.get("dns_server") not in local_dns_servers] + local_ok = any(row.get("status") == "正常" for row in local) + public_ok = any(row.get("status") == "正常" for row in public) + any_ok = any(row.get("status") == "正常" for row in group) + if local and not local_ok and public_ok: + lines.append(f"{domain} {record_type}: 本机 DNS 失败但对比 DNS 成功，本机 DNS 服务器疑似异常。") + elif not any_ok: + lines.append(f"{domain} {record_type}: 所有 DNS 均未成功，可能域名不存在、网络不可达或上游 DNS 异常。") + elif values_are_inconsistent(group): + lines.append(f"{domain} {record_type}: 不同 DNS 返回结果不一致，可能存在 CDN、缓存、污染或策略差异。") + + if len(lines) == 2: + lines.append("未发现明显 DNS 异常；如故障偶发，建议增加重复次数复测。") + return "\n".join(lines) + + +def group_by_domain_type(rows: list[dict]) -> dict[tuple[str, str], list[dict]]: + grouped: dict[tuple[str, str], list[dict]] = {} + for row in rows: + grouped.setdefault((row.get("domain", ""), row.get("record_type", "")), []).append(row) + return grouped + + +def values_are_inconsistent(rows: list[dict]) -> bool: + value_sets = { + tuple(sorted(value.strip() for value in str(row.get("values", "")).split(";") if value.strip())) + for row in rows + if row.get("status") == "正常" and row.get("values") + } + return len(value_sets) > 1 + + +def collect_adapter_dns(adapters: list[dict]) -> list[str]: + servers = [] + for adapter in adapters: + for key in ("dns1", "dns2"): + value = str(adapter.get(key, "")).strip() + if value and is_ip_address(value): + servers.append(value) + return list(dict.fromkeys(servers)) + + +def parse_list(text: str, max_items: int = 20) -> list[str]: + items = [item.strip() for item in re.split(r"[,，;\s]+", text.strip()) if item.strip()] + return list(dict.fromkeys(items))[:max_items] + + +def parse_record_types(text: str) -> list[str]: + values = [item.upper() for item in parse_list(text, 10)] + invalid = [item for item in values if item not in SUPPORTED_RECORD_TYPES] + if invalid: + raise ValueError("不支持的 DNS 记录类型: " + ", ".join(invalid)) + return values + + +def parse_dns_servers(text: str) -> list[str]: + servers = [] + for item in parse_list(text, 12): + if not is_ip_address(item): + raise ValueError(f"DNS 服务器必须是 IP 地址: {item}") + servers.append(item) + return list(dict.fromkeys(servers)) + + +def classify_error(text: str) -> str: + clean = compact_text(text) + lower = clean.lower() + if "timed out" in lower or "timeout" in lower or "超时" in clean: + return "查询超时" + if "non-existent" in lower or "nxdomain" in lower or "不存在" in clean: + return "域名或记录不存在" + if "refused" in lower or "拒绝" in clean: + return "DNS 服务器拒绝查询" + if "server failed" in lower or "servfail" in lower: + return "上游 DNS 返回失败" + return clean[:240] if clean else "解析失败" + + +def verdict_for_error(error: str) -> str: + if "超时" in error: + return "查询超时，可能 DNS 服务器不可达或网络阻塞" + if "不存在" in error: + return "未查询到该记录，建议与其它 DNS 或记录类型对比" + if "拒绝" in error: + return "DNS 服务器拒绝查询，可能存在策略限制" + return "解析失败，建议与其它 DNS 对比" + + +def is_nslookup_error(output: str) -> bool: + lower = output.lower() + return any(token in lower for token in ("timed out", "can't find", "non-existent", "servfail", "refused")) + + +def is_ip_address(value: str) -> bool: + try: + ipaddress.ip_address(value) + return True + except ValueError: + return False + + +def extract_json(text: str): + output = text.strip() + if not output: + return [] + json_start = min([idx for idx in (output.find("["), output.find("{")) if idx >= 0], default=-1) + if json_start < 0: + return [] + data = json.loads(output[json_start:]) + return [] if data is None else data + + +def ps_quote(value: str) -> str: + return "'" + str(value).replace("'", "''") + "'" + + +def compact_text(text: str) -> str: + return re.sub(r"\s+", " ", text or "").strip() + + +def clamp_int(value, min_value: int, max_value: int, label: str) -> int: + try: + number = int(str(value).strip()) + except ValueError as exc: + raise ValueError(f"{label}必须是整数") from exc + if number < min_value or number > max_value: + raise ValueError(f"{label}必须在 {min_value}-{max_value} 之间") + return number diff --git a/core/ui/basic_ui.py b/core/ui/basic_ui.py deleted file mode 100644 index ab126b6..0000000 --- a/core/ui/basic_ui.py +++ /dev/null @@ -1,71 +0,0 @@ -import tkinter as tk -from tkinter import ttk - - -class BasicUI: - """旧版页面的兼容辅助类。 - - 新界面主要使用 core.ui.components；保留这个类是为了兼容后续可能仍引用 - add_input/add_combobox/add_button 的小工具页。 - """ - - def add_input( - self, - parent, - label_text, - row, - col=0, - inivar="", - label_width=8, - entry_width=20, - colspan=1, - sticky="w", - ): - group_frame = ttk.Frame(parent) - group_frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - - label = ttk.Label(group_frame, text=f"{label_text}:", width=label_width, anchor="w") - label.grid(row=0, column=0, sticky="w", padx=(0, 5)) - - var = tk.StringVar(value=inivar) - entry = ttk.Entry(group_frame, textvariable=var, width=entry_width) - entry.grid(row=0, column=1, sticky="w") - - return {"frame": group_frame, "label": label, "entry": entry, "var": var} - - def add_combobox( - self, - parent, - label_text, - row, - col=0, - listbox=None, - inivar=0, - width=17, - colspan=1, - sticky="w", - label_width=8, - state="readonly", - ): - values = listbox or [] - frame = ttk.Frame(parent) - frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - - label = ttk.Label(frame, text=f"{label_text}:", width=label_width, anchor="w") - label.grid(row=0, column=0, sticky="w", padx=(0, 5)) - - var = tk.StringVar() - combobox = ttk.Combobox(frame, textvariable=var, values=values, width=width, state=state) - combobox.grid(row=0, column=1, sticky="w") - if values and inivar >= 0: - var.set(values[inivar]) - - return {"frame": frame, "label": label, "combobox": combobox, "var": var} - - def add_button(self, parent, button_text, row, col=0, command=None, width=5, colspan=1, sticky="w"): - group_frame = ttk.Frame(parent) - group_frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - - btn = ttk.Button(group_frame, text=button_text, command=command, width=width) - btn.grid(row=0, column=0, sticky="w") - return {"frame": group_frame, "btn": btn} diff --git a/core/ui/tab_device_discovery.py b/core/ui/tab_device_discovery.py new file mode 100644 index 0000000..0d259a8 --- /dev/null +++ b/core/ui/tab_device_discovery.py @@ -0,0 +1,233 @@ +import threading +import tkinter as tk +from tkinter import filedialog, messagebox, ttk + +from core.Function.device_discovery_fun import ALL_ADAPTERS, DeviceDiscovery +from core.ui.components import Console, Page, action_bar, button, combo, field + + +class DeviceDiscoveryTab(Page): + def __init__(self, parent): + super().__init__(parent, "设备发现", "发现局域网在线与 ARP 可见设备，整理 IP、MAC、主机名、厂商与来源网卡。") + self.body.rowconfigure(3, weight=1) + + status = self.section("实时状态", 0, columns=6) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.current = field(status, "当前网卡", 1, 1, "-", 18) + self.scan_scope = field(status, "扫描范围", 1, 2, "-", 20) + self.progress = field(status, "进度", 1, 3, "0/0", 10) + self.found = field(status, "发现设备", 1, 4, "0", 8) + self.elapsed = field(status, "耗时", 1, 5, "0.0s", 10) + for item in (self.state, self.current, self.scan_scope, self.progress, self.found, self.elapsed): + item["entry"].configure(state="disabled") + + params = self.section("发现参数", 1, columns=6) + self.adapter = combo(params, "检测网卡", 1, 0, [ALL_ADAPTERS], ALL_ADAPTERS, 22) + self.scan_range = field(params, "扫描范围", 1, 1, "", 34, colspan=2) + self.workers = field(params, "并发数", 1, 3, "64", 10) + self.timeout = field(params, "超时 ms", 1, 4, "500", 10) + self.max_hosts = field(params, "最大地址数", 1, 5, "254", 10) + actions = action_bar(params, 2, 6) + self.start_btn = button(actions, "开始发现", self.start_discovery, "Primary.TButton") + self.stop_btn = button(actions, "停止", self.stop_discovery, "Danger.TButton") + self.refresh_btn = button(actions, "刷新网卡", self.load_adapters, "Secondary.TButton") + self.auto_range_btn = button(actions, "自动范围", self.fill_default_range, "Secondary.TButton") + self.copy_btn = button(actions, "复制清单", self.copy_inventory, "Secondary.TButton") + self.export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton") + + results = self.section("发现结果", 2, columns=1) + results.rowconfigure(1, weight=1) + results.columnconfigure(0, weight=1) + self.results_tree = ttk.Treeview( + results, + columns=("ip", "mac", "hostname", "vendor", "adapter", "latency", "method", "note"), + show="headings", + height=10, + ) + headings = { + "ip": "IP", + "mac": "MAC", + "hostname": "主机名", + "vendor": "厂商", + "adapter": "来源网卡", + "latency": "延迟", + "method": "发现方式", + "note": "备注", + } + widths = { + "ip": 130, + "mac": 150, + "hostname": 180, + "vendor": 130, + "adapter": 150, + "latency": 85, + "method": 90, + "note": 130, + } + for column, title in headings.items(): + self.results_tree.heading(column, text=title) + self.results_tree.column(column, width=widths[column], anchor="w") + self.results_tree.tag_configure("在线", foreground="#15803d") + self.results_tree.tag_configure("ARP 可见", foreground="#b7791f") + self.results_tree.tag_configure("本机", foreground="#246bfe") + self.results_tree.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) + result_scroll = ttk.Scrollbar(results, orient="vertical", command=self.results_tree.yview) + result_scroll.grid(row=1, column=1, sticky="ns", pady=(0, 18)) + self.results_tree.configure(yscrollcommand=result_scroll.set) + + output = self.section("发现控制台", 3, columns=1) + output.rowconfigure(1, weight=1) + output.columnconfigure(0, weight=1) + self.console = Console(output, height=12) + self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) + + self.discovery = DeviceDiscovery(self.write, self.on_task_done, self.update_status, self.add_result) + self.after(350, self.load_adapters) + + def write(self, text, tag=None): + self.console.write(text, tag) + + def clear(self): + self.console.clear() + for item in self.results_tree.get_children(): + self.results_tree.delete(item) + self.update_status( + { + "state": "等待", + "current": "-", + "scan_range": "-", + "total": 0, + "scanned": 0, + "found": 0, + "elapsed": 0, + } + ) + + def load_adapters(self): + def worker(): + try: + values = self.discovery.get_adapter_choices() + default_range = self.discovery.default_scan_range(ALL_ADAPTERS) + except Exception as exc: + values = [ALL_ADAPTERS] + default_range = "" + self.write(f"读取网卡失败: {exc}\n", "warning") + self.after(0, lambda: self.apply_adapters(values, default_range)) + + threading.Thread(target=worker, daemon=True).start() + + def apply_adapters(self, values, default_range): + values = values or [ALL_ADAPTERS] + values = list(dict.fromkeys(values)) + self.adapter["combobox"]["values"] = values + if self.adapter["var"].get() not in values: + self.adapter["var"].set(values[1] if len(values) > 1 else values[0]) + if not self.scan_range["var"].get() and default_range: + self.scan_range["var"].set(default_range) + + def fill_default_range(self): + try: + value = self.discovery.default_scan_range(self.adapter["var"].get()) + if not value: + messagebox.showinfo("提示", "未能根据当前网卡生成扫描范围") + return + self.scan_range["var"].set(value) + self.write(f"已生成安全扫描范围: {value}\n", "success") + except Exception as exc: + messagebox.showwarning("生成失败", str(exc)) + + def start_discovery(self): + try: + self.clear() + self.start_btn.configure(state="disabled") + self.discovery.start_discovery(self.adapter["var"].get(), self.options()) + except Exception as exc: + self.start_btn.configure(state="normal") + messagebox.showwarning("无法开始设备发现", str(exc)) + + def stop_discovery(self): + try: + self.discovery.stop_discovery() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) + + def options(self): + return { + "scan_range": self.scan_range["var"].get(), + "workers": self.workers["var"].get(), + "timeout_ms": self.timeout["var"].get(), + "max_hosts": self.max_hosts["var"].get(), + } + + def export_results(self): + path = filedialog.asksaveasfilename( + title="导出设备发现结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: + return + try: + self.discovery.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_inventory(self): + text = self.discovery.copy_summary() + if not text: + messagebox.showinfo("提示", "还没有可复制的设备清单") + return + self.clipboard_clear() + self.clipboard_append(text) + self.write("已复制设备清单到剪贴板\n", "success") + + def update_status(self, stats): + def apply(): + total = stats.get("total", 0) + scanned = stats.get("scanned", 0) + values = [ + (self.state, stats.get("state", "等待")), + (self.current, stats.get("current") or "-"), + (self.scan_scope, stats.get("scan_range") or "-"), + (self.progress, f"{scanned}/{total}" if total else "0/0"), + (self.found, str(stats.get("found", 0))), + (self.elapsed, f"{stats.get('elapsed', 0):.1f}s"), + ] + for item, value in values: + item["entry"].configure(state="normal") + item["var"].set(value) + item["entry"].configure(state="disabled") + + self.after(0, apply) + + def add_result(self, row): + def apply(): + method = row.get("method", "") + note = row.get("note", "") + tag = "本机" if "本机" in note else method + latency = row.get("latency_ms", 0) + try: + latency_text = f"{float(latency):.1f} ms" if float(latency) > 0 else "" + except (TypeError, ValueError): + latency_text = "" + self.results_tree.insert( + "", + "end", + values=( + row.get("ip", ""), + row.get("mac", ""), + row.get("hostname", ""), + row.get("vendor", ""), + row.get("adapter", ""), + latency_text, + method, + note, + ), + tags=(tag,), + ) + + self.after(0, apply) + + def on_task_done(self): + self.after(0, lambda: self.start_btn.configure(state="normal")) diff --git a/core/ui/tab_dns.py b/core/ui/tab_dns.py new file mode 100644 index 0000000..2c55153 --- /dev/null +++ b/core/ui/tab_dns.py @@ -0,0 +1,229 @@ +import threading +import tkinter as tk +from tkinter import filedialog, messagebox, ttk + +from core.Function.dns_diag_fun import ALL_ADAPTERS, DEFAULT_DOMAINS, DEFAULT_RECORD_TYPES, DnsDiagnostic +from core.ui.components import Console, Page, action_bar, button, combo, field + + +class DnsTab(Page): + def __init__(self, parent): + super().__init__(parent, "DNS 诊断", "对比本机 DNS 与常用 DNS 的解析结果、耗时和失败原因。") + self.body.rowconfigure(3, weight=1) + + status = self.section("实时状态", 0, columns=6) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.current_dns = field(status, "当前 DNS", 1, 1, "-", 24, colspan=2) + self.target_total = field(status, "目标数", 1, 3, "0", 8) + self.progress = field(status, "进度", 1, 4, "0/0", 10) + self.abnormal = field(status, "异常数", 1, 5, "0", 8) + self.elapsed = field(status, "耗时", 2, 0, "0.0s", 10) + for item in (self.state, self.current_dns, self.target_total, self.progress, self.abnormal, self.elapsed): + item["entry"].configure(state="disabled") + + params = self.section("诊断参数", 1, columns=6) + self.adapter = combo(params, "检测网卡", 1, 0, [ALL_ADAPTERS], ALL_ADAPTERS, 22) + self.domains = field(params, "目标域名", 1, 1, DEFAULT_DOMAINS, 34, colspan=2) + self.record_types = field(params, "记录类型", 1, 3, DEFAULT_RECORD_TYPES, 18) + self.timeout = field(params, "超时 ms", 1, 4, "2000", 10) + self.repeat_count = field(params, "重复次数", 1, 5, "1", 10) + self.dns_servers = field(params, "DNS 服务器", 2, 0, "", 60, colspan=6) + actions = action_bar(params, 3, 6) + self.start_btn = button(actions, "开始诊断", self.start_diagnosis, "Primary.TButton") + self.stop_btn = button(actions, "停止", self.stop_diagnosis, "Danger.TButton") + self.refresh_btn = button(actions, "刷新网卡", self.load_adapters, "Secondary.TButton") + self.auto_dns_btn = button(actions, "自动 DNS", self.fill_default_dns, "Secondary.TButton") + self.copy_btn = button(actions, "复制摘要", self.copy_summary, "Secondary.TButton") + self.export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton") + + results = self.section("诊断结果", 2, columns=1) + results.rowconfigure(1, weight=1) + results.columnconfigure(0, weight=1) + self.results_tree = ttk.Treeview( + results, + columns=("domain", "type", "server", "status", "elapsed", "values", "verdict"), + show="headings", + height=10, + ) + headings = { + "domain": "域名", + "type": "类型", + "server": "DNS 服务器", + "status": "状态", + "elapsed": "耗时", + "values": "解析结果", + "verdict": "错误 / 判断", + } + widths = { + "domain": 160, + "type": 70, + "server": 130, + "status": 80, + "elapsed": 85, + "values": 300, + "verdict": 360, + } + for column, title in headings.items(): + self.results_tree.heading(column, text=title) + self.results_tree.column(column, width=widths[column], anchor="w") + self.results_tree.tag_configure("正常", foreground="#15803d") + self.results_tree.tag_configure("无记录", foreground="#b7791f") + self.results_tree.tag_configure("失败", foreground="#dc2626") + self.results_tree.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) + result_scroll = ttk.Scrollbar(results, orient="vertical", command=self.results_tree.yview) + result_scroll.grid(row=1, column=1, sticky="ns", pady=(0, 18)) + self.results_tree.configure(yscrollcommand=result_scroll.set) + + output = self.section("诊断控制台", 3, columns=1) + output.rowconfigure(1, weight=1) + output.columnconfigure(0, weight=1) + self.console = Console(output, height=12) + self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) + + self.diagnostic = DnsDiagnostic(self.write, self.on_task_done, self.update_status, self.add_result) + self.after(350, self.load_adapters) + + def write(self, text, tag=None): + self.console.write(text, tag) + + def clear(self): + self.console.clear() + for item in self.results_tree.get_children(): + self.results_tree.delete(item) + self.update_status( + { + "state": "等待", + "current_dns": "-", + "total": 0, + "completed": 0, + "abnormal": 0, + "elapsed": 0, + } + ) + + def load_adapters(self): + def worker(): + try: + values = self.diagnostic.get_adapter_choices() + default_dns = self.diagnostic.default_dns_servers(values[1] if len(values) > 1 else ALL_ADAPTERS) + except Exception as exc: + values = [ALL_ADAPTERS] + default_dns = "" + self.write(f"读取网卡 DNS 失败: {exc}\n", "warning") + self.after(0, lambda: self.apply_adapters(values, default_dns)) + + threading.Thread(target=worker, daemon=True).start() + + def apply_adapters(self, values, default_dns): + values = values or [ALL_ADAPTERS] + values = list(dict.fromkeys(values)) + self.adapter["combobox"]["values"] = values + if self.adapter["var"].get() not in values: + self.adapter["var"].set(values[1] if len(values) > 1 else values[0]) + if not self.dns_servers["var"].get() and default_dns: + self.dns_servers["var"].set(default_dns) + + def fill_default_dns(self): + try: + value = self.diagnostic.default_dns_servers(self.adapter["var"].get()) + if not value: + messagebox.showinfo("提示", "未读取到本机 DNS，已保留常用 DNS 对比服务器") + value = "223.5.5.5,114.114.114.114,8.8.8.8" + self.dns_servers["var"].set(value) + self.write(f"已生成 DNS 对比列表: {value}\n", "success") + except Exception as exc: + messagebox.showwarning("生成失败", str(exc)) + + def start_diagnosis(self): + try: + self.clear() + self.start_btn.configure(state="disabled") + self.diagnostic.start_diagnosis(self.adapter["var"].get(), self.options()) + except Exception as exc: + self.start_btn.configure(state="normal") + messagebox.showwarning("无法开始 DNS 诊断", str(exc)) + + def stop_diagnosis(self): + try: + self.diagnostic.stop_diagnosis() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) + + def options(self): + return { + "domains": self.domains["var"].get(), + "record_types": self.record_types["var"].get(), + "dns_servers": self.dns_servers["var"].get(), + "timeout_ms": self.timeout["var"].get(), + "repeat_count": self.repeat_count["var"].get(), + } + + def export_results(self): + path = filedialog.asksaveasfilename( + title="导出 DNS 诊断结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: + return + try: + self.diagnostic.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_summary(self): + text = self.diagnostic.copy_summary() + if not text: + messagebox.showinfo("提示", "还没有可复制的 DNS 诊断摘要") + return + self.clipboard_clear() + self.clipboard_append(text) + self.write("已复制 DNS 诊断摘要到剪贴板\n", "success") + + def update_status(self, stats): + def apply(): + total = stats.get("total", 0) + completed = stats.get("completed", 0) + values = [ + (self.state, stats.get("state", "等待")), + (self.current_dns, stats.get("current_dns") or "-"), + (self.target_total, str(total)), + (self.progress, f"{completed}/{total}" if total else "0/0"), + (self.abnormal, str(stats.get("abnormal", 0))), + (self.elapsed, f"{stats.get('elapsed', 0):.1f}s"), + ] + for item, value in values: + item["entry"].configure(state="normal") + item["var"].set(value) + item["entry"].configure(state="disabled") + + self.after(0, apply) + + def add_result(self, row): + def apply(): + elapsed = row.get("elapsed_ms", 0) + try: + elapsed_text = f"{float(elapsed):.1f} ms" if float(elapsed) > 0 else "" + except (TypeError, ValueError): + elapsed_text = "" + detail = row.get("error") or row.get("verdict", "") + self.results_tree.insert( + "", + "end", + values=( + row.get("domain", ""), + row.get("record_type", ""), + row.get("dns_server", ""), + row.get("status", ""), + elapsed_text, + row.get("values", ""), + detail, + ), + tags=(row.get("status", ""),), + ) + + self.after(0, apply) + + def on_task_done(self): + self.after(0, lambda: self.start_btn.configure(state="normal")) diff --git a/core/ui/ui_main.py b/core/ui/ui_main.py index 7a363bf..4155265 100644 --- a/core/ui/ui_main.py +++ b/core/ui/ui_main.py @@ -1,6 +1,8 @@ import tkinter as tk from tkinter import ttk +from core.ui.tab_device_discovery import DeviceDiscoveryTab +from core.ui.tab_dns import DnsTab from core.ui.tab_ip_conflict import IpConflictTab from core.ui.tab_loop import LoopTab from core.ui.tab_network import NetworkTab @@ -61,6 +63,8 @@ class MainUI: items = [ ("network", "◎", "网卡配置", "IP / DNS / DHCP"), + ("dns", "◇", "DNS 诊断", "解析与服务器对比"), + ("devices", "▦", "设备发现", "局域网资产"), ("ping", "⌁", "Ping 探测", "单点与批量探活"), ("ports", "⌕", "端口扫描", "TCP 连通性检测"), ("trace", "↗", "路由追踪", "跳点路径分析"), @@ -104,11 +108,13 @@ class MainUI: def _build_pages(self) -> None: self.pages = { "network": NetworkTab(self.content), + "dns": DnsTab(self.content), "ping": PingTab(self.content), "ports": TelnetTab(self.content), "trace": TracertTab(self.content), "loop": LoopTab(self.content), "ip_conflict": IpConflictTab(self.content), + "devices": DeviceDiscoveryTab(self.content), } for page in self.pages.values(): page.grid(row=0, column=0, sticky="nsew") diff --git a/打包.bat b/打包.bat index 3a916de..4180b76 100644 --- a/打包.bat +++ b/打包.bat @@ -3,6 +3,7 @@ setlocal set "PYTHON_EXE=C:\Users\Administrator\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe" set "PYINSTALLER_PATH=%CD%\.packaging\pyinstaller" +set "APP_ICON=%CD%\assets\app.ico" if exist "%PYTHON_EXE%" ( if not exist "%PYINSTALLER_PATH%\PyInstaller\__main__.py" ( @@ -10,9 +11,9 @@ if exist "%PYTHON_EXE%" ( if errorlevel 1 exit /b 1 ) set "PYTHONPATH=%PYINSTALLER_PATH%" - "%PYTHON_EXE%" -m PyInstaller -F -w main.py -n NetworkTool + "%PYTHON_EXE%" -m PyInstaller -F -w main.py -n NetworkTool --icon "%APP_ICON%" ) else ( - pyinstaller -F -w main.py -n NetworkTool + pyinstaller -F -w main.py -n NetworkTool --icon "%APP_ICON%" ) endlocal