diff --git a/README.md b/README.md index cda2349..8d6d73e 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,9 @@ - 配置模板：保存常用网卡配置、套用模板到表单、删除模板。 - Ping 探测：支持持续/指定次数、间隔、超时、包大小、TTL、禁止分片、本地源 IP 下拉、CIDR/范围/列表批量探活、目标导入、CSV 导出、实时丢包率/平均延迟/抖动/质量统计。 - 端口扫描：支持单端口测试、端口列表/范围、常用端口预设、批量主机/CIDR/IP 段扫描、并发/超时控制、只显示开放端口、服务名识别、可选 Banner 探测、开放端口复制与 CSV 导出。 -- 路由追踪：支持设置最大跳数和单跳超时时间。 +- 路由追踪：支持 IPv4/IPv6、主机名解析开关、单次/指定次数/持续追踪、最大跳数、单跳超时、高延迟阈值、逐跳 RTT 表格、超时/高延迟/波动诊断、原始输出保留、诊断摘要复制与 CSV 导出。 +- 环网检测：基于本机网卡统计、广播/组播流量、ARP/邻居表变化和网关 Ping 稳定性，判断疑似二层环路/广播风暴风险，支持单网卡/全部活动网卡、诊断摘要复制与 CSV 导出。 +- IP 冲突：支持本机 IP 冲突检测与安全限速网段扫描，结合地址状态、Tcpip 系统事件、ARP/邻居表和 IP/MAC 异常判断冲突风险，支持摘要复制与 CSV 导出。 ## 运行 diff --git a/core/Function/ip_conflict_fun.py b/core/Function/ip_conflict_fun.py new file mode 100644 index 0000000..7b5bba9 --- /dev/null +++ b/core/Function/ip_conflict_fun.py @@ -0,0 +1,696 @@ +import concurrent.futures +import csv +import ipaddress +import json +import re +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 = "全部活动网卡" +MODE_BOTH = "两者都做" +MODE_LOCAL = "本机 IP 检测" +MODE_SUBNET = "网段扫描" + + +@dataclass +class IpConflictOptions: + mode: str = MODE_BOTH + scan_range: str = "" + workers: int = 64 + timeout_ms: int = 500 + max_hosts: int = 254 + neighbor_samples: int = 3 + + +@dataclass +class IpConflictEvidence: + address_state: str = "" + local_mac: str = "" + observed_macs: list[str] = field(default_factory=list) + conflict_macs: list[str] = field(default_factory=list) + tcpip_event_count: int = 0 + gateway_mac_changes: int = 0 + ip_mac_changes: int = 0 + shared_mac_ips: int = 0 + scanned_hosts: int = 0 + notes: str = "" + + def to_dict(self) -> dict: + data = asdict(self) + data["observed_macs"] = ", ".join(self.observed_macs) + data["conflict_macs"] = ", ".join(self.conflict_macs) + return data + + +@dataclass +class IpConflictResult: + adapter: str + ip: str + mac: str + conflict_macs: str + evidence_type: str + risk_level: str + verdict: str + evidence: IpConflictEvidence + checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")) + + def to_dict(self) -> dict: + data = { + "adapter": self.adapter, + "ip": self.ip, + "mac": self.mac, + "conflict_macs": self.conflict_macs, + "evidence_type": self.evidence_type, + "risk_level": self.risk_level, + "verdict": self.verdict, + "checked_at": self.checked_at, + } + data.update(self.evidence.to_dict()) + return data + + +class IpConflictDetector: + 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_detection(self, adapter_name: str = ALL_ADAPTERS, options: Optional[dict] = None) -> None: + if self.is_running(): + raise RuntimeError("IP 冲突检测正在运行，请先停止当前任务") + + conflict_options = self.normalize_options(options) + adapters = self._select_adapters(adapter_name) + if not adapters: + raise ValueError("没有找到可检测的活动网卡") + + self.stop_event.clear() + self.last_results = [] + self.last_summary = "" + self.output(f"开始 IP 冲突检测: {adapter_name or ALL_ADAPTERS}，模式: {conflict_options.mode}\n", "muted") + self.output("说明: 检测结果基于本机可见 ARP/邻居表、地址状态和系统事件，不替代交换机侧排查。\n\n", "muted") + self.status(self._status("运行中", len(adapters), "", 0, 0, 0, 0)) + self.worker = threading.Thread(target=self._run_detection, args=(adapters, conflict_options), daemon=True) + self.worker.start() + + def _run_detection(self, adapters: list[dict], options: IpConflictOptions) -> None: + started = time.perf_counter() + completed = 0 + max_risk = "正常" + + try: + address_states = self.get_address_states() + events = self.get_tcpip_conflict_events() + neighbor_samples = [self.get_neighbors()] + + if options.mode in (MODE_BOTH, MODE_SUBNET): + targets = self.build_scan_targets(adapters, options) + self.output(f"准备刷新 ARP: {len(targets)} 个地址，并发 {options.workers}，超时 {options.timeout_ms}ms\n", "muted") + self.status(self._status("扫描中", len(adapters), "", 0, len(targets), 0, time.perf_counter() - started)) + self.refresh_arp_targets(targets, options) + neighbor_samples.append(self.get_neighbors()) + + for _index in range(max(options.neighbor_samples - len(neighbor_samples), 0)): + if self.stop_event.is_set(): + break + self.stop_event.wait(0.3) + neighbor_samples.append(self.get_neighbors()) + + rows = [] + for adapter in adapters: + if self.stop_event.is_set(): + break + completed += 1 + self.status(self._status("分析中", len(adapters), adapter["name"], 0, 0, completed, time.perf_counter() - started)) + results = self.evaluate_adapter(adapter, address_states, events, neighbor_samples, options) + if not results: + results = [normal_result(adapter, "未发现本机 IP 冲突证据")] + for item in results: + row = item.to_dict() + rows.append(row) + self.last_results.append(row) + max_risk = worse_risk(max_risk, item.risk_level) + self.result(row) + + self.last_summary = build_summary(rows, stopped=self.stop_event.is_set()) + self.output("\nIP 冲突检测已停止\n" if self.stop_event.is_set() else "\nIP 冲突检测完成\n", "warning" if self.stop_event.is_set() else "success") + self.output(self.last_summary + "\n", "success" if max_risk == "正常" else "warning") + self.status(self._status("已停止" if self.stop_event.is_set() else "已完成", len(adapters), "", risk_rank(max_risk), 0, completed, time.perf_counter() - started)) + except Exception as exc: + self.output(f"\nIP 冲突检测失败: {exc}\n", "error") + self.status(self._status("失败", len(adapters), "", 0, 0, completed, time.perf_counter() - started)) + finally: + self.done() + + def evaluate_adapter( + self, + adapter: dict, + address_states: dict, + events: list[dict], + neighbor_samples: list[list[dict]], + options: IpConflictOptions, + ) -> list[IpConflictResult]: + local_mac = normalize_mac(adapter.get("mac", "")) + local_ip = adapter.get("ipv4", "") + gateway = adapter.get("gateway", "") + interface_index = str(adapter.get("interface_index", "")) + local_neighbors = filter_neighbors(neighbor_samples, interface_index, local_ip) + ip_to_macs, mac_to_ips = build_neighbor_maps(local_neighbors) + results = [] + + state = address_states.get(local_ip, "") + event_hits = event_count_for_ip(events, local_ip) + observed_macs = sorted(ip_to_macs.get(local_ip, set())) + conflict_macs = sorted([mac for mac in observed_macs if local_mac and mac != local_mac and not is_special_mac(mac)]) + if state in {"Duplicate", "Tentative", "DadFailed"} or event_hits or conflict_macs: + evidence = IpConflictEvidence( + address_state=state, + local_mac=local_mac, + observed_macs=observed_macs, + conflict_macs=conflict_macs, + tcpip_event_count=event_hits, + scanned_hosts=options.max_hosts, + notes="本机 IP 冲突证据", + ) + level = "冲突风险" + reasons = [] + if state: + reasons.append(f"地址状态 {state}") + if event_hits: + reasons.append(f"系统 Tcpip 冲突事件 {event_hits} 条") + if conflict_macs: + reasons.append("本机 IP 被其它 MAC 响应") + results.append( + IpConflictResult( + adapter=adapter["name"], + ip=local_ip, + mac=local_mac, + conflict_macs=", ".join(conflict_macs), + evidence_type="本机 IP", + risk_level=level, + verdict="疑似本机 IP 冲突，请检查是否有其它设备使用相同地址。证据: " + "、".join(reasons), + evidence=evidence, + ) + ) + + multi_mac_ips = sorted(ip for ip, macs in ip_to_macs.items() if len([mac for mac in macs if not is_special_mac(mac)]) > 1) + for ip in multi_mac_ips: + if ip in {local_ip, gateway}: + continue + macs = sorted(ip_to_macs[ip]) + evidence = IpConflictEvidence( + local_mac=local_mac, + observed_macs=macs, + conflict_macs=macs, + ip_mac_changes=1, + scanned_hosts=options.max_hosts, + notes="同一 IP 出现多个 MAC", + ) + results.append( + IpConflictResult( + adapter=adapter["name"], + ip=ip, + mac="", + conflict_macs=", ".join(macs), + evidence_type="网段 IP", + risk_level="冲突风险", + verdict="同一 IP 在邻居表中出现多个 MAC，疑似 IP 地址冲突。", + evidence=evidence, + ) + ) + + if gateway: + gateway_macs = sorted(ip_to_macs.get(gateway, set())) + if len([mac for mac in gateway_macs if not is_special_mac(mac)]) > 1: + evidence = IpConflictEvidence( + local_mac=local_mac, + observed_macs=gateway_macs, + conflict_macs=gateway_macs, + gateway_mac_changes=max(len(gateway_macs) - 1, 0), + scanned_hosts=options.max_hosts, + notes="网关 MAC 发生变化", + ) + results.append( + IpConflictResult( + adapter=adapter["name"], + ip=gateway, + mac="", + conflict_macs=", ".join(gateway_macs), + evidence_type="网关", + risk_level="可疑", + verdict="网关 IP 对应多个 MAC，可能是网关冗余、ARP 欺骗或地址冲突，建议复核网关设备。", + evidence=evidence, + ) + ) + + proxy_rows = [] + for mac, ips in mac_to_ips.items(): + clean_ips = [ip for ip in ips if not is_multicast_or_broadcast_ip(ip)] + if len(clean_ips) >= 8 and mac != local_mac: + proxy_rows.append((mac, sorted(clean_ips))) + for mac, ips in proxy_rows[:3]: + evidence = IpConflictEvidence( + local_mac=local_mac, + observed_macs=[mac], + shared_mac_ips=len(ips), + scanned_hosts=options.max_hosts, + notes="同一 MAC 对应大量 IP，可能是 Proxy ARP 或网关代理，不直接判定为 IP 冲突", + ) + results.append( + IpConflictResult( + adapter=adapter["name"], + ip=", ".join(ips[:5]) + ("..." if len(ips) > 5 else ""), + mac=mac, + conflict_macs="", + evidence_type="Proxy ARP 提示", + risk_level="可疑", + verdict="同一 MAC 对应大量 IP，更像 Proxy ARP/网关代理，需结合网络拓扑判断。", + evidence=evidence, + ) + ) + + return results + + def build_scan_targets(self, adapters: list[dict], options: IpConflictOptions) -> 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 refresh_arp_targets(self, targets: list[str], options: IpConflictOptions) -> None: + done = 0 + + def task(ip: str) -> None: + if self.stop_event.is_set(): + return + run_hidden(["ping", ip, "-n", "1", "-w", str(options.timeout_ms)], timeout=max(2, options.timeout_ms / 1000 + 2)) + + 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): + done += 1 + if self.stop_event.is_set(): + for item in futures: + item.cancel() + break + try: + future.result() + except Exception: + pass + if done % 20 == 0 or done == len(targets): + self.status(self._status("扫描中", 0, futures[future], risk_rank("正常"), len(targets), done, 0)) + + def get_address_states(self) -> dict[str, str]: + script = r""" +$ErrorActionPreference = "SilentlyContinue" +Get-NetIPAddress -AddressFamily IPv4 | Select-Object IPAddress,InterfaceIndex,AddressState,PrefixLength | 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 {item.get("IPAddress", ""): str(item.get("AddressState", "")) for item in data if item.get("IPAddress")} + except Exception as exc: + self.output(f"读取本机地址状态失败: {exc}\n", "warning") + return {} + + def get_tcpip_conflict_events(self) -> list[dict]: + script = r""" +$ErrorActionPreference = "SilentlyContinue" +Get-WinEvent -FilterHashtable @{LogName='System'; ProviderName='Tcpip'; StartTime=(Get-Date).AddDays(-7)} -MaxEvents 80 | +Select-Object TimeCreated,Id,ProviderName,Message | ConvertTo-Json -Depth 4 -Compress +""" + try: + result = run_hidden(["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script], timeout=12) + if result.returncode != 0 or not result.stdout.strip(): + return [] + data = extract_json(result.stdout) + if isinstance(data, dict): + data = [data] + events = [] + for item in data: + message = str(item.get("Message", "")) + if re.search(r"conflict|duplicate|冲突|重复", message, re.IGNORECASE): + events.append( + { + "time": str(item.get("TimeCreated", "")), + "id": str(item.get("Id", "")), + "message": message, + } + ) + return events + except Exception as exc: + self.output(f"读取 Tcpip 系统事件失败: {exc}\n", "warning") + return [] + + 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 stop_detection(self) -> None: + if not self.is_running(): + raise RuntimeError("当前没有正在运行的 IP 冲突检测") + self.stop_event.set() + self.output("\n正在停止 IP 冲突检测...\n", "warning") + + def export_results(self, path: str) -> None: + if not self.last_results: + raise RuntimeError("还没有可导出的 IP 冲突检测结果") + fields = [ + "adapter", + "ip", + "mac", + "conflict_macs", + "evidence_type", + "risk_level", + "verdict", + "address_state", + "local_mac", + "observed_macs", + "tcpip_event_count", + "gateway_mac_changes", + "ip_mac_changes", + "shared_mac_ips", + "scanned_hosts", + "notes", + "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 + + def normalize_options(self, options: Optional[dict]) -> IpConflictOptions: + options = options or {} + mode = options.get("mode", MODE_BOTH) + if mode not in {MODE_BOTH, MODE_LOCAL, MODE_SUBNET}: + mode = MODE_BOTH + return IpConflictOptions( + mode=mode, + 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, "最大扫描地址数"), + neighbor_samples=clamp_int(options.get("neighbor_samples", 3), 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, total: int, current: str, max_risk_rank: int, total_targets: int, completed: int, elapsed: float) -> dict: + return { + "state": state, + "total_adapters": total, + "current": current, + "max_risk": risk_from_rank(max_risk_rank), + "total_targets": total_targets, + "completed": completed, + "elapsed": elapsed, + } + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) + + +def normal_result(adapter: dict, verdict: str) -> IpConflictResult: + evidence = IpConflictEvidence(address_state="", local_mac=normalize_mac(adapter.get("mac", "")), notes="未发现冲突证据") + return IpConflictResult( + adapter=adapter.get("name", ""), + ip=adapter.get("ipv4", ""), + mac=normalize_mac(adapter.get("mac", "")), + conflict_macs="", + evidence_type="本机 IP", + risk_level="正常", + verdict=verdict, + evidence=evidence, + ) + + +def filter_neighbors(samples: list[list[dict]], interface_index: str, interface_ip: str) -> list[dict]: + rows = [] + for sample in samples: + for item in sample: + if item.get("ifIndex"): + if str(item.get("ifIndex")) == str(interface_index): + rows.append(item) + elif item.get("interface_ip") == interface_ip: + rows.append(item) + return rows + + +def build_neighbor_maps(neighbors: list[dict]) -> tuple[dict[str, set[str]], dict[str, set[str]]]: + ip_to_macs: dict[str, set[str]] = {} + mac_to_ips: dict[str, set[str]] = {} + for item in neighbors: + ip = item.get("ip", "") + mac = normalize_mac(item.get("mac", "")) + if not ip or not mac or is_special_mac(mac): + continue + ip_to_macs.setdefault(ip, set()).add(mac) + mac_to_ips.setdefault(mac, set()).add(ip) + return ip_to_macs, mac_to_ips + + +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 + network = ipaddress.ip_network(f"{ip}/{prefix}", strict=False) + return str(network) + 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 event_count_for_ip(events: list[dict], ip: str) -> int: + return len([event for event in events if ip and ip in event.get("message", "")]) + + +def build_summary(rows: list[dict], stopped: bool = False) -> str: + if stopped: + return "检测已停止，当前结果仅代表已完成部分。" + if not rows: + return "没有生成 IP 冲突检测结果。" + ordered = sorted(rows, key=lambda row: risk_rank(row.get("risk_level", "正常")), reverse=True) + highest = ordered[0] + conflict = [row for row in ordered if row.get("risk_level") == "冲突风险"] + suspicious = [row for row in ordered if row.get("risk_level") == "可疑"] + lines = [ + "==== IP 冲突诊断摘要 ====", + f"最高风险: {highest.get('adapter', '')} {highest.get('ip', '')} {highest.get('risk_level', '')}", + "结论基于本机 Windows 可见证据，建议结合交换机 MAC 地址表和现场设备确认。", + ] + if conflict: + lines.append(f"发现 {len(conflict)} 条冲突风险。") + for row in conflict[:5]: + lines.append(f"{row['adapter']} {row['ip']}: {row['verdict']}") + elif suspicious: + lines.append(f"发现 {len(suspicious)} 条可疑信号，建议复测或结合交换机侧信息确认。") + for row in suspicious[:5]: + lines.append(f"{row['adapter']} {row['ip']}: {row['verdict']}") + else: + lines.append("整体正常，未发现本机 IP 被其它 MAC 占用、同 IP 多 MAC 或 Tcpip 冲突事件。") + return "\n".join(lines) + + +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 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 输出") + return json.loads(output[json_start:]) + + +def is_special_mac(mac: str) -> bool: + if not mac or mac == "00-00-00-00-00-00": + return True + return mac.startswith("FF-FF-FF") or mac.startswith("01-00-5E") + + +def is_multicast_or_broadcast_ip(ip: str) -> bool: + try: + parsed = ipaddress.ip_address(ip) + return parsed.is_multicast or str(parsed).endswith(".255") + except ValueError: + return False + + +def worse_risk(left: str, right: str) -> str: + return left if risk_rank(left) >= risk_rank(right) else right + + +def risk_rank(level: str) -> int: + return {"正常": 0, "可疑": 1, "冲突风险": 2}.get(level, 0) + + +def risk_from_rank(value: int) -> str: + if value >= 2: + return "冲突风险" + if value == 1: + return "可疑" + return "正常" + + +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/loop_fun.py b/core/Function/loop_fun.py new file mode 100644 index 0000000..df75f5d --- /dev/null +++ b/core/Function/loop_fun.py @@ -0,0 +1,631 @@ +import csv +import json +import re +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 = "全部活动网卡" + + +@dataclass +class LoopOptions: + duration_sec: int = 15 + interval_sec: int = 1 + ping_timeout_ms: int = 800 + + +@dataclass +class LoopEvidence: + non_unicast_pps: float = 0.0 + broadcast_pps: float = 0.0 + multicast_pps: float = 0.0 + non_unicast_ratio: float = 0.0 + error_delta: int = 0 + discard_delta: int = 0 + gateway_ping_sent: int = 0 + gateway_ping_loss: float = 0.0 + gateway_avg_ms: float = 0.0 + gateway_jitter_ms: float = 0.0 + gateway_mac_changes: int = 0 + ip_mac_changes: int = 0 + shared_mac_count: int = 0 + neighbor_unreachable: int = 0 + notes: str = "" + + def to_dict(self) -> dict: + data = asdict(self) + for key, value in data.items(): + if isinstance(value, float): + data[key] = round(value, 2) + return data + + +@dataclass +class LoopResult: + adapter: str + ipv4: str + gateway: str + interface_index: str + link_speed: str + risk_score: int + risk_level: str + verdict: str + evidence: LoopEvidence + checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")) + + def to_dict(self) -> dict: + data = { + "adapter": self.adapter, + "ipv4": self.ipv4, + "gateway": self.gateway, + "interface_index": self.interface_index, + "link_speed": self.link_speed, + "risk_score": self.risk_score, + "risk_level": self.risk_level, + "verdict": self.verdict, + "checked_at": self.checked_at, + } + data.update(self.evidence.to_dict()) + return data + + +class LoopDetector: + 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 start_detection(self, adapter_name: str = ALL_ADAPTERS, options: Optional[dict] = None) -> None: + if self.is_running(): + raise RuntimeError("环网检测正在运行，请先停止当前任务") + + loop_options = self.normalize_options(options) + adapters = self._select_adapters(adapter_name) + if not adapters: + raise ValueError("没有找到可检测的活动网卡") + + self.stop_event.clear() + self.last_results = [] + self.last_summary = "" + self.output( + f"开始环网风险检测: {adapter_name or ALL_ADAPTERS}，" + f"检测 {loop_options.duration_sec}s，采样间隔 {loop_options.interval_sec}s\n", + "muted", + ) + self.output("说明: 本功能基于本机侧证据判断疑似二层环路风险，不等同于交换机 STP/SNMP 的绝对结论。\n\n", "muted") + self.status(self._status("运行中", len(adapters), "", 0, 0, 0)) + self.worker = threading.Thread(target=self._run_detection, args=(adapters, loop_options), daemon=True) + self.worker.start() + + def _run_detection(self, adapters: list[dict], options: LoopOptions) -> None: + started = time.perf_counter() + ping_samples = {adapter["name"]: [] for adapter in adapters} + neighbor_samples = [] + start_stats = {} + end_stats = {} + + try: + start_stats = self.get_adapter_statistics() + neighbor_samples.append(self.get_neighbors()) + end_time = started + options.duration_sec + sample_index = 0 + + while not self.stop_event.is_set() and time.perf_counter() < end_time: + sample_index += 1 + for adapter in adapters: + if self.stop_event.is_set(): + break + self.status(self._status("运行中", len(adapters), adapter["name"], 0, sample_index, time.perf_counter() - started)) + gateway = adapter.get("gateway", "") + if gateway: + ping_samples[adapter["name"]].append(self.ping_gateway(gateway, options.ping_timeout_ms)) + neighbor_samples.append(self.get_neighbors()) + remaining = end_time - time.perf_counter() + if remaining <= 0: + break + self.stop_event.wait(min(options.interval_sec, remaining)) + + end_stats = self.get_adapter_statistics() + rows = [] + max_score = 0 + for index, adapter in enumerate(adapters, start=1): + result = self.evaluate_adapter( + adapter, + start_stats.get(adapter["name"], {}), + end_stats.get(adapter["name"], {}), + neighbor_samples, + ping_samples.get(adapter["name"], []), + max(time.perf_counter() - started, 1.0), + ) + row = result.to_dict() + rows.append(row) + self.last_results.append(row) + max_score = max(max_score, result.risk_score) + self.result(row) + self.status(self._status("汇总中", len(adapters), adapter["name"], max_score, index, time.perf_counter() - started)) + + self.last_summary = build_summary(rows, 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" if max_score < 30 else "warning") + self.status(self._status("已停止" if self.stop_event.is_set() else "已完成", len(adapters), "", max_score, len(adapters), time.perf_counter() - started)) + except Exception as exc: + self.output(f"\n环网检测失败: {exc}\n", "error") + self.status(self._status("失败", len(adapters), "", 0, 0, time.perf_counter() - started)) + finally: + self.done() + + def evaluate_adapter( + self, + adapter: dict, + start_stats: dict, + end_stats: dict, + neighbor_samples: list[list[dict]], + pings: list[dict], + duration_sec: float, + ) -> LoopResult: + evidence = build_evidence(adapter, start_stats, end_stats, neighbor_samples, pings, duration_sec) + score, reasons = score_evidence(evidence) + level = risk_level(score) + verdict = build_verdict(level, reasons) + return LoopResult( + adapter=adapter.get("name", ""), + ipv4=adapter.get("ipv4", ""), + gateway=adapter.get("gateway", ""), + interface_index=str(adapter.get("interface_index", "")), + link_speed=adapter.get("link_speed", ""), + risk_score=score, + risk_level=level, + verdict=verdict, + evidence=evidence, + ) + + def get_adapter_statistics(self) -> dict[str, dict]: + script = r""" +$ErrorActionPreference = "SilentlyContinue" +Get-NetAdapterStatistics | Select-Object InterfaceAlias, +ReceivedBroadcastPackets,SentBroadcastPackets, +ReceivedMulticastPackets,SentMulticastPackets, +ReceivedUnicastPackets,SentUnicastPackets, +ReceivedPacketErrors,OutboundPacketErrors, +ReceivedDiscardedPackets,OutboundDiscardedPackets, +ReceivedBytes,SentBytes | 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()) + output = result.stdout.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:]) + if isinstance(data, dict): + data = [data] + return {item.get("InterfaceAlias", ""): normalize_stat_item(item) for item in data if item.get("InterfaceAlias")} + except Exception as exc: + self.output(f"读取网卡统计失败，相关证据将降级: {exc}\n", "warning") + return {} + + 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()) + output = result.stdout.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:]) + 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 ping_gateway(self, gateway: str, timeout_ms: int) -> dict: + try: + result = run_hidden(["ping", gateway, "-n", "1", "-w", str(timeout_ms)], timeout=max(2, timeout_ms / 1000 + 2)) + return parse_ping_result(gateway, result.stdout) + except Exception as exc: + return {"gateway": gateway, "ok": False, "rtt": 0.0, "message": str(exc)} + + def stop_detection(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 = [ + "adapter", + "ipv4", + "gateway", + "interface_index", + "link_speed", + "risk_score", + "risk_level", + "verdict", + "non_unicast_pps", + "broadcast_pps", + "multicast_pps", + "non_unicast_ratio", + "error_delta", + "discard_delta", + "gateway_ping_sent", + "gateway_ping_loss", + "gateway_avg_ms", + "gateway_jitter_ms", + "gateway_mac_changes", + "ip_mac_changes", + "shared_mac_count", + "neighbor_unreachable", + "notes", + "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 + + def normalize_options(self, options: Optional[dict]) -> LoopOptions: + options = options or {} + return LoopOptions( + duration_sec=clamp_int(options.get("duration_sec", 15), 5, 300, "检测时长"), + interval_sec=clamp_int(options.get("interval_sec", 1), 1, 30, "采样间隔"), + ping_timeout_ms=clamp_int(options.get("ping_timeout_ms", 800), 100, 10000, "Ping 超时"), + ) + + 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, total: int, current: str, max_score: int, completed: int, elapsed: float) -> dict: + return { + "state": state, + "total_adapters": total, + "current_adapter": current, + "max_score": max_score, + "risk_level": risk_level(max_score), + "completed": completed, + "elapsed": elapsed, + } + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) + + +def build_evidence( + adapter: dict, + start_stats: dict, + end_stats: dict, + neighbor_samples: list[list[dict]], + pings: list[dict], + duration_sec: float, +) -> LoopEvidence: + broadcast_delta = positive_delta(start_stats, end_stats, "broadcast_packets") + multicast_delta = positive_delta(start_stats, end_stats, "multicast_packets") + unicast_delta = positive_delta(start_stats, end_stats, "unicast_packets") + error_delta = positive_delta(start_stats, end_stats, "packet_errors") + discard_delta = positive_delta(start_stats, end_stats, "discarded_packets") + non_unicast = broadcast_delta + multicast_delta + total_packets = non_unicast + unicast_delta + + ping_rtts = [item["rtt"] for item in pings if item.get("ok")] + sent = len(pings) + received = len(ping_rtts) + loss = (sent - received) / sent * 100 if sent else 0.0 + avg_ms = sum(ping_rtts) / len(ping_rtts) if ping_rtts else 0.0 + jitter = 0.0 + if len(ping_rtts) > 1: + deltas = [abs(ping_rtts[index] - ping_rtts[index - 1]) for index in range(1, len(ping_rtts))] + jitter = sum(deltas) / len(deltas) + + arp = analyze_neighbors(adapter, neighbor_samples) + notes = [] + if not start_stats or not end_stats: + notes.append("网卡统计不可用") + if not adapter.get("gateway"): + notes.append("未发现默认网关，跳过网关 Ping") + + return LoopEvidence( + non_unicast_pps=non_unicast / duration_sec, + broadcast_pps=broadcast_delta / duration_sec, + multicast_pps=multicast_delta / duration_sec, + non_unicast_ratio=(non_unicast / total_packets * 100) if total_packets else 0.0, + error_delta=error_delta, + discard_delta=discard_delta, + gateway_ping_sent=sent, + gateway_ping_loss=loss, + gateway_avg_ms=avg_ms, + gateway_jitter_ms=jitter, + gateway_mac_changes=arp["gateway_mac_changes"], + ip_mac_changes=arp["ip_mac_changes"], + shared_mac_count=arp["shared_mac_count"], + neighbor_unreachable=arp["neighbor_unreachable"], + notes="; ".join(notes), + ) + + +def analyze_neighbors(adapter: dict, samples: list[list[dict]]) -> dict: + interface_index = str(adapter.get("interface_index", "")) + interface_ip = adapter.get("ipv4", "") + gateway = adapter.get("gateway", "") + gateway_macs = set() + ip_to_macs: dict[str, set[str]] = {} + mac_to_ips: dict[str, set[str]] = {} + unreachable = 0 + + for sample in samples: + for item in sample: + if not neighbor_belongs_to_adapter(item, interface_index, interface_ip): + continue + ip = item.get("ip", "") + mac = normalize_mac(item.get("mac", "")) + state = str(item.get("state", "")) + if not ip or not mac or is_special_mac(mac): + continue + if state.lower() in {"unreachable", "incomplete"} or state in {"不可达", "不完整"}: + unreachable += 1 + ip_to_macs.setdefault(ip, set()).add(mac) + mac_to_ips.setdefault(mac, set()).add(ip) + if gateway and ip == gateway: + gateway_macs.add(mac) + + return { + "gateway_mac_changes": max(len(gateway_macs) - 1, 0), + "ip_mac_changes": len([ip for ip, macs in ip_to_macs.items() if len(macs) > 1]), + "shared_mac_count": len([mac for mac, ips in mac_to_ips.items() if len(ips) >= 6]), + "neighbor_unreachable": unreachable, + } + + +def score_evidence(evidence: LoopEvidence) -> tuple[int, list[str]]: + score = 0 + reasons = [] + if evidence.non_unicast_pps >= 2000: + score += 35 + reasons.append("非单播流量速率极高") + elif evidence.non_unicast_pps >= 500: + score += 25 + reasons.append("非单播流量速率偏高") + elif evidence.non_unicast_pps >= 100: + score += 15 + reasons.append("非单播流量明显增加") + elif evidence.non_unicast_pps >= 20: + score += 8 + reasons.append("非单播流量轻微偏高") + + if evidence.non_unicast_ratio >= 60: + score += 20 + reasons.append("非单播占比过高") + elif evidence.non_unicast_ratio >= 30: + score += 12 + reasons.append("非单播占比较高") + elif evidence.non_unicast_ratio >= 10: + score += 6 + reasons.append("非单播占比偏高") + + if evidence.error_delta + evidence.discard_delta >= 20: + score += 20 + reasons.append("错误/丢弃包明显增加") + elif evidence.error_delta + evidence.discard_delta > 0: + score += 10 + reasons.append("出现错误/丢弃包") + + if evidence.gateway_ping_sent: + if evidence.gateway_ping_loss >= 50: + score += 20 + reasons.append("网关 Ping 丢包严重") + elif evidence.gateway_ping_loss > 0: + score += 10 + reasons.append("网关 Ping 存在丢包") + if evidence.gateway_jitter_ms >= 50: + score += 10 + reasons.append("网关延迟波动较大") + if evidence.gateway_avg_ms >= 100: + score += 8 + reasons.append("网关平均延迟偏高") + + if evidence.gateway_mac_changes: + score += 25 + reasons.append("网关 MAC 发生变化") + if evidence.ip_mac_changes: + score += 20 + reasons.append("同一 IP 出现多个 MAC") + if evidence.shared_mac_count: + score += 8 + reasons.append("部分 MAC 对应过多 IP") + if evidence.neighbor_unreachable >= 5: + score += 8 + reasons.append("邻居表不可达项较多") + + return min(score, 100), reasons + + +def build_verdict(level: str, reasons: list[str]) -> str: + if not reasons: + return "未发现明显环网风险，仅基于本机侧证据判断。" + prefix = { + "高风险": "疑似二层环路或广播风暴，请优先检查交换机端口、网线回接和 STP 状态。", + "可疑": "存在环网风险迹象，建议结合交换机端口流量和 STP 日志复核。", + "正常": "存在轻微信号，但尚不足以判断为环网风险。", + }[level] + return prefix + " 证据: " + "、".join(reasons) + + +def build_summary(rows: list[dict], stopped: bool = False) -> str: + if stopped: + return "检测已停止，当前结果仅代表已完成采样。" + if not rows: + return "没有生成检测结果。" + ordered = sorted(rows, key=lambda row: row.get("risk_score", 0), reverse=True) + highest = ordered[0] + lines = [ + "==== 环网风险诊断摘要 ====", + f"最高风险: {highest['adapter']} {highest['risk_level']} {highest['risk_score']} 分", + "结论基于本机网卡统计、ARP/邻居表和网关 Ping，不能替代交换机 STP/SNMP 侧确认。", + ] + risky = [row for row in ordered if row.get("risk_score", 0) >= 30] + if not risky: + lines.append("整体正常，未发现明显广播风暴、ARP 抖动或网关稳定性异常。") + else: + for row in risky: + lines.append(f"{row['adapter']}: {row['verdict']}") + return "\n".join(lines) + + +def normalize_stat_item(item: dict) -> dict: + return { + "broadcast_packets": to_int(item.get("ReceivedBroadcastPackets")) + to_int(item.get("SentBroadcastPackets")), + "multicast_packets": to_int(item.get("ReceivedMulticastPackets")) + to_int(item.get("SentMulticastPackets")), + "unicast_packets": to_int(item.get("ReceivedUnicastPackets")) + to_int(item.get("SentUnicastPackets")), + "packet_errors": to_int(item.get("ReceivedPacketErrors")) + to_int(item.get("OutboundPacketErrors")), + "discarded_packets": to_int(item.get("ReceivedDiscardedPackets")) + to_int(item.get("OutboundDiscardedPackets")), + "bytes": to_int(item.get("ReceivedBytes")) + to_int(item.get("SentBytes")), + } + + +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 neighbor_belongs_to_adapter(item: dict, interface_index: str, interface_ip: str) -> bool: + if item.get("ifIndex"): + return str(item.get("ifIndex")) == str(interface_index) + if item.get("interface_ip"): + return item.get("interface_ip") == interface_ip + return False + + +def parse_ping_result(gateway: str, output: str) -> dict: + if re.search(r"\bTTL=", output, re.IGNORECASE): + match = re.search(r"(?:time|时间)[=<]?\s*(\d+(?:\.\d+)?)\s*(?:ms|毫秒)", output, re.IGNORECASE) + return {"gateway": gateway, "ok": True, "rtt": float(match.group(1)) if match else 0.0, "message": "在线"} + if re.search(r"请求超时|timed out|timeout", output, re.IGNORECASE): + return {"gateway": gateway, "ok": False, "rtt": 0.0, "message": "超时"} + return {"gateway": gateway, "ok": False, "rtt": 0.0, "message": "无响应"} + + +def positive_delta(start: dict, end: dict, key: str) -> int: + return max(to_int(end.get(key)) - to_int(start.get(key)), 0) + + +def to_int(value) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + +def risk_level(score: int) -> str: + if score >= 60: + return "高风险" + if score >= 30: + return "可疑" + return "正常" + + +def normalize_mac(value: str) -> str: + raw = str(value or "").strip().replace(":", "-").upper() + return raw + + +def is_special_mac(mac: str) -> bool: + if not mac or mac == "00-00-00-00-00-00": + return True + return mac.startswith("FF-FF-FF") or mac.startswith("01-00-5E") + + +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/tracert_fun.py b/core/Function/tracert_fun.py index d9ff0b9..e00655d 100644 --- a/core/Function/tracert_fun.py +++ b/core/Function/tracert_fun.py @@ -1,4 +1,8 @@ +import csv +import re import threading +import time +from dataclasses import asdict, dataclass, field from typing import Callable, Optional from core.Function.common import popen_hidden, validate_host @@ -6,53 +10,208 @@ from core.Function.common import popen_hidden, validate_host OutputCallback = Callable[[str, Optional[str]], None] DoneCallback = Callable[[], None] +StatusCallback = Callable[[dict], None] +ResultCallback = Callable[[dict], None] + + +TRACE_STATUS = { + "ok": "正常", + "timeout": "超时", + "high_latency": "高延迟", + "jitter": "波动大", +} + + +@dataclass +class TraceOptions: + max_hops: int = 20 + timeout_ms: int = 800 + address_family: str = "自动" + resolve_names: bool = False + mode: str = "单次" + repeat_count: int = 1 + interval_ms: int = 1000 + high_latency_ms: int = 100 + + +@dataclass +class TraceHop: + run: int + hop: int + probe1: str + probe2: str + probe3: str + avg_ms: float + jitter_ms: float + host: str + ip: str + status: str + status_text: str + raw_line: 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["avg_ms"] = round(self.avg_ms, 1) + data["jitter_ms"] = round(self.jitter_ms, 1) + return data + + +@dataclass +class TraceStats: + total_runs: int = 1 + current_run: int = 0 + max_hops: int = 20 + current_hop: int = 0 + timeout_hops: int = 0 + high_latency_hops: int = 0 + jitter_hops: int = 0 + max_latency_ms: float = 0.0 + avg_latency_ms: float = 0.0 + samples: list[float] = field(default_factory=list) + started_at: float = field(default_factory=time.perf_counter) + + def record(self, hop: TraceHop) -> None: + self.current_hop = max(self.current_hop, hop.hop) + if hop.status == "timeout": + self.timeout_hops += 1 + elif hop.status == "high_latency": + self.high_latency_hops += 1 + elif hop.status == "jitter": + self.jitter_hops += 1 + if hop.avg_ms: + self.samples.append(hop.avg_ms) + self.max_latency_ms = max(self.max_latency_ms, hop.avg_ms) + self.avg_latency_ms = sum(self.samples) / len(self.samples) + + def snapshot(self, state: str) -> dict: + return { + "state": state, + "current_run": self.current_run, + "total_runs": self.total_runs, + "current_hop": self.current_hop, + "max_hops": self.max_hops, + "timeout_hops": self.timeout_hops, + "high_latency_hops": self.high_latency_hops, + "jitter_hops": self.jitter_hops, + "max_latency_ms": self.max_latency_ms, + "avg_latency_ms": self.avg_latency_ms, + "elapsed": time.perf_counter() - self.started_at, + } class TracertFun: - def __init__(self, output: OutputCallback, done: Optional[DoneCallback] = None): + 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.process = None self.stop_event = threading.Event() self.worker = None + self.last_results: list[dict] = [] + self.last_summary = "" - def start_tracert(self, target: str, max_hops: int = 20, timeout_ms: int = 800) -> None: + def start_tracert(self, target: str, max_hops: int = 20, timeout_ms: int = 800, options: Optional[dict] = None) -> None: target = validate_host(target) - max_hops = max(1, min(int(max_hops), 64)) - timeout_ms = max(100, min(int(timeout_ms), 10000)) + trace_options = self.normalize_options(options or {}) + trace_options.max_hops = clamp_int(max_hops, 1, 64, "最大跳数") + trace_options.timeout_ms = clamp_int(timeout_ms, 100, 60000, "单跳超时") if self.is_running(): raise RuntimeError("路由追踪正在运行，请先停止当前任务") self.stop_event.clear() - command = ["tracert", "-d", "-w", str(timeout_ms), "-h", str(max_hops), target] - self.output(f"开始路由追踪: {target}，最大 {max_hops} 跳，超时 {timeout_ms}ms\n\n", "muted") - self.worker = threading.Thread(target=self._run, args=(command,), daemon=True) + self.last_results = [] + self.last_summary = "" + total_runs = 0 if trace_options.mode == "持续" else trace_options.repeat_count + stats = TraceStats(total_runs=total_runs, max_hops=trace_options.max_hops) + self.status(stats.snapshot("运行中")) + self.output(f"开始路由追踪: {target}\n", "muted") + self.output(self.describe_options(trace_options), "muted") + self.worker = threading.Thread(target=self._run_loop, args=(target, trace_options, stats), daemon=True) self.worker.start() - def _run(self, command) -> None: + def _run_loop(self, target: str, options: TraceOptions, stats: TraceStats) -> None: + run = 0 + max_runs = None if options.mode == "持续" else options.repeat_count + try: + while not self.stop_event.is_set() and (max_runs is None or run < max_runs): + run += 1 + stats.current_run = run + stats.current_hop = 0 + self.output(f"\n==== 第 {run} 次追踪 ====\n", "muted") + self.status(stats.snapshot("运行中")) + self._run_once(target, options, stats, run) + if self.stop_event.is_set() or (max_runs is not None and run >= max_runs): + break + self.stop_event.wait(options.interval_ms / 1000) + finally: + state = "已停止" if self.stop_event.is_set() else "已完成" + self.status(stats.snapshot(state)) + self._write_overall_summary(state) + self.done() + + def _run_once(self, target: str, options: TraceOptions, stats: TraceStats, run: int) -> None: + command = self.build_command(target, options) + run_results = [] try: self.process = popen_hidden(command) if not self.process.stdout: return - for line in self.process.stdout: + for raw_line in self.process.stdout: if self.stop_event.is_set(): break - self.output(line, None) + line = raw_line.rstrip() + self.output(raw_line, None) + hop = parse_trace_line(line, run, options.high_latency_ms) + if not hop: + continue + row = hop.to_dict() + self.last_results.append(row) + run_results.append(row) + stats.record(hop) + self.result(row) + self.status(stats.snapshot("运行中")) except Exception as exc: self.output(f"\n路由追踪失败: {exc}\n", "error") finally: if self.process: try: - self.process.terminate() + if self.stop_event.is_set(): + self.process.terminate() + else: + self.process.wait(timeout=1) except Exception: pass self.process = None - if self.stop_event.is_set(): - self.output("\n路由追踪已停止\n", "warning") - else: - self.output("\n路由追踪完成\n", "success") - self.done() + self._write_run_summary(run_results, run) + + def build_command(self, target: str, options: TraceOptions) -> list[str]: + command = ["tracert"] + if not options.resolve_names: + command.append("-d") + if options.address_family == "IPv4": + command.append("-4") + elif options.address_family == "IPv6": + command.append("-6") + command += ["-w", str(options.timeout_ms), "-h", str(options.max_hops), target] + return command + + def describe_options(self, options: TraceOptions) -> str: + runs = "持续" if options.mode == "持续" else f"{options.repeat_count} 次" + names = "开启" if options.resolve_names else "关闭" + return ( + f"模式: {options.mode}({runs}) 地址族: {options.address_family} 最大跳数: {options.max_hops} " + f"单跳超时: {options.timeout_ms}ms 间隔: {options.interval_ms}ms " + f"解析主机名: {names} 高延迟阈值: {options.high_latency_ms}ms\n" + ) def stop_tracert(self) -> None: if not self.is_running(): @@ -65,5 +224,193 @@ class TracertFun: pass self.output("\n正在停止路由追踪...\n", "warning") + def export_results(self, path: str) -> None: + if not self.last_results: + raise RuntimeError("还没有可导出的路由追踪结果") + fields = [ + "run", + "hop", + "probe1", + "probe2", + "probe3", + "avg_ms", + "jitter_ms", + "host", + "ip", + "status", + "status_text", + "raw_line", + "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: row.get(field, "") for field in fields}) + + def copy_summary(self) -> str: + return self.last_summary + + def normalize_options(self, options: dict) -> TraceOptions: + mode = options.get("mode", "单次") + if mode not in {"单次", "指定次数", "持续"}: + mode = "单次" + repeat_count = 1 if mode == "单次" else clamp_int(options.get("repeat_count", 3), 1, 1000, "追踪次数") + address_family = options.get("address_family", "自动") + if address_family not in {"自动", "IPv4", "IPv6"}: + address_family = "自动" + return TraceOptions( + address_family=address_family, + resolve_names=bool(options.get("resolve_names", False)), + mode=mode, + repeat_count=repeat_count, + interval_ms=clamp_int(options.get("interval_ms", 1000), 0, 60000, "追踪间隔"), + high_latency_ms=clamp_int(options.get("high_latency_ms", 100), 1, 10000, "高延迟阈值"), + ) + + def _write_run_summary(self, rows: list[dict], run: int) -> None: + if not rows: + return + summary = build_diagnosis(rows) + self.last_summary = summary + self.output(f"\n==== 第 {run} 次诊断摘要 ====\n{summary}\n", "success" if "整体正常" in summary else "warning") + + def _write_overall_summary(self, state: str) -> None: + if self.stop_event.is_set(): + self.output("\n路由追踪已停止\n", "warning") + self.last_summary = "路由追踪已停止。" + return + if not self.last_results: + self.output("\n路由追踪完成，但没有解析到跳点结果\n", "warning") + self.last_summary = "路由追踪完成，但没有解析到跳点结果。" + return + self.last_summary = build_diagnosis(self.last_results) + self.output(f"\n路由追踪{state}\n", "success") + def is_running(self) -> bool: return bool(self.worker and self.worker.is_alive()) + + +def parse_trace_line(line: str, run: int = 1, high_latency_ms: int = 100) -> Optional[TraceHop]: + match = re.match(r"^\s*(\d+)\s+(.+)$", line) + if not match: + return None + + hop = int(match.group(1)) + rest = match.group(2) + probe_matches = list(re.finditer(r"\*|<\s*\d+\s*(?:ms|毫秒)|\d+\s*(?:ms|毫秒)", rest, flags=re.IGNORECASE)) + if not probe_matches: + return None + + probes = [] + values = [] + for probe_match in probe_matches[:3]: + raw = normalize_probe_text(probe_match.group(0)) + probes.append(raw) + value = probe_to_ms(raw) + if value is not None: + values.append(value) + while len(probes) < 3: + probes.append("*") + + endpoint = rest[probe_matches[min(len(probe_matches), 3) - 1].end() :].strip() + host, ip = parse_endpoint(endpoint) + avg_ms = sum(values) / len(values) if values else 0.0 + jitter_ms = max(values) - min(values) if len(values) > 1 else 0.0 + status = classify_hop(values, avg_ms, jitter_ms, high_latency_ms) + return TraceHop( + run=run, + hop=hop, + probe1=probes[0], + probe2=probes[1], + probe3=probes[2], + avg_ms=avg_ms, + jitter_ms=jitter_ms, + host=host, + ip=ip, + status=status, + status_text=TRACE_STATUS[status], + raw_line=line, + ) + + +def normalize_probe_text(text: str) -> str: + text = re.sub(r"\s+", "", text.strip()) + text = text.replace("毫秒", "ms") + return text.replace("MS", "ms").replace("Ms", "ms") + + +def probe_to_ms(text: str) -> Optional[float]: + if text == "*": + return None + if text.startswith("<"): + return 1.0 + match = re.search(r"\d+(?:\.\d+)?", text) + return float(match.group(0)) if match else None + + +def parse_endpoint(endpoint: str) -> tuple[str, str]: + cleaned = endpoint.strip() + if not cleaned or re.search(r"请求超时|request timed out", cleaned, re.IGNORECASE): + return "", "" + bracket_match = re.match(r"(.+?)\s+\[([^\]]+)\]$", cleaned) + if bracket_match: + return bracket_match.group(1).strip(), bracket_match.group(2).strip() + return "", cleaned + + +def classify_hop(values: list[float], avg_ms: float, jitter_ms: float, high_latency_ms: int) -> str: + if not values: + return "timeout" + if avg_ms >= high_latency_ms: + return "high_latency" + if jitter_ms >= max(30, high_latency_ms * 0.5): + return "jitter" + return "ok" + + +def build_diagnosis(rows: list[dict]) -> str: + if not rows: + return "没有解析到可诊断的跳点。" + + latest_run = max(int(row.get("run") or 1) for row in rows) + latest = [row for row in rows if int(row.get("run") or 1) == latest_run] + latest.sort(key=lambda row: int(row.get("hop") or 0)) + + timeout_hops = [row for row in latest if row.get("status") == "timeout"] + high_hops = [row for row in latest if row.get("status") == "high_latency"] + jitter_hops = [row for row in latest if row.get("status") == "jitter"] + ok_after_timeout = False + for index, row in enumerate(latest): + if row.get("status") == "timeout" and any(next_row.get("status") != "timeout" for next_row in latest[index + 1 :]): + ok_after_timeout = True + break + + lines = [ + f"本次共解析 {len(latest)} 跳，超时 {len(timeout_hops)} 跳，高延迟 {len(high_hops)} 跳，波动 {len(jitter_hops)} 跳。" + ] + if latest and latest[0].get("status") == "timeout": + lines.append("第一跳无响应，优先检查本机网关、防火墙或本地网络。") + if ok_after_timeout: + lines.append("中间跳点超时但后续恢复，通常是中间路由器限制 ICMP，不一定代表链路故障。") + if len(latest) >= 2 and latest[-1].get("status") == "timeout" and latest[-2].get("status") == "timeout": + lines.append("末段连续超时，目标侧网络、跨网链路或目标防火墙可能存在限制。") + if high_hops: + detail = ", ".join(f"第 {row['hop']} 跳 {row.get('avg_ms', 0):.1f}ms" for row in high_hops[:5]) + lines.append(f"发现高延迟跳点: {detail}。") + if jitter_hops: + detail = ", ".join(f"第 {row['hop']} 跳抖动 {row.get('jitter_ms', 0):.1f}ms" for row in jitter_hops[:5]) + lines.append(f"发现延迟波动: {detail}。") + if not timeout_hops and not high_hops and not jitter_hops: + lines.append("整体正常，未发现明显超时、高延迟或波动。") + return "\n".join(lines) + + +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/logger_config.py b/core/logger_config.py deleted file mode 100644 index fbd1144..0000000 --- a/core/logger_config.py +++ /dev/null @@ -1,26 +0,0 @@ -import logging -import os -import sys - - -def get_base_dir() -> str: - if getattr(sys, "frozen", False): - return os.path.dirname(sys.executable) - return os.path.dirname(os.path.abspath(sys.argv[0])) - - -def setup_logger() -> logging.Logger: - log_file = os.path.join(get_base_dir(), "app.log") - - logging.basicConfig( - level=logging.INFO, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", - handlers=[ - logging.FileHandler(log_file, encoding="utf-8"), - logging.StreamHandler(), - ], - ) - - logger = logging.getLogger(__name__) - logger.info("日志系统初始化完成: %s", log_file) - return logger diff --git a/core/ui/tab_ip_conflict.py b/core/ui/tab_ip_conflict.py new file mode 100644 index 0000000..a055938 --- /dev/null +++ b/core/ui/tab_ip_conflict.py @@ -0,0 +1,224 @@ +import threading +import tkinter as tk +from tkinter import filedialog, messagebox, ttk + +from core.Function.ip_conflict_fun import ALL_ADAPTERS, MODE_BOTH, MODE_LOCAL, MODE_SUBNET, IpConflictDetector +from core.ui.components import Console, Page, action_bar, button, combo, field + + +class IpConflictTab(Page): + def __init__(self, parent): + super().__init__(parent, "IP 冲突", "检测本机 IP 是否被占用，并安全扫描网段内 IP/MAC 异常。") + self.body.rowconfigure(3, weight=1) + + status = self.section("实时状态", 0, columns=6) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.adapter_total = field(status, "网卡数", 1, 1, "0", 8) + self.current = field(status, "当前对象", 1, 2, "-", 18) + self.max_risk = field(status, "最高风险", 1, 3, "正常", 10) + self.progress = field(status, "进度", 1, 4, "0/0", 10) + self.elapsed = field(status, "耗时", 1, 5, "0.0s", 10) + for item in (self.state, self.adapter_total, self.current, self.max_risk, self.progress, 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.mode = combo(params, "检测模式", 1, 1, [MODE_BOTH, MODE_LOCAL, MODE_SUBNET], MODE_BOTH, 14) + self.scan_range = field(params, "扫描范围", 1, 2, "", 30, colspan=2) + self.workers = field(params, "并发数", 1, 4, "64", 10) + self.timeout = field(params, "超时 ms", 1, 5, "500", 10) + self.max_hosts = field(params, "最大地址数", 2, 0, "254", 10) + actions = action_bar(params, 3, 6) + self.start_btn = button(actions, "开始检测", self.start_detection, "Primary.TButton") + self.stop_btn = button(actions, "停止", self.stop_detection, "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_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=("adapter", "ip", "mac", "conflict", "type", "level", "verdict"), + show="headings", + height=9, + ) + headings = { + "adapter": "网卡", + "ip": "IP", + "mac": "本机/观察 MAC", + "conflict": "冲突 MAC", + "type": "证据类型", + "level": "风险", + "verdict": "判断", + } + widths = { + "adapter": 130, + "ip": 150, + "mac": 150, + "conflict": 180, + "type": 110, + "level": 90, + "verdict": 420, + } + 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.detector = IpConflictDetector(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": "等待", + "total_adapters": 0, + "current": "-", + "max_risk": "正常", + "total_targets": 0, + "completed": 0, + "elapsed": 0, + } + ) + + def load_adapters(self): + def worker(): + try: + values = self.detector.get_adapter_choices() + default_range = self.detector.default_scan_range(values[0] if len(values) == 1 else 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[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.detector.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_detection(self): + try: + self.clear() + self.start_btn.configure(state="disabled") + self.detector.start_detection(self.adapter["var"].get(), self.options()) + except Exception as exc: + self.start_btn.configure(state="normal") + messagebox.showwarning("无法开始 IP 冲突检测", str(exc)) + + def stop_detection(self): + try: + self.detector.stop_detection() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) + + def options(self): + return { + "mode": self.mode["var"].get(), + "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="导出 IP 冲突检测结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: + return + try: + self.detector.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_summary(self): + text = self.detector.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_targets", 0) + done = stats.get("completed", 0) + values = [ + (self.state, stats.get("state", "等待")), + (self.adapter_total, str(stats.get("total_adapters", 0))), + (self.current, stats.get("current") or "-"), + (self.max_risk, stats.get("max_risk", "正常")), + (self.progress, f"{done}/{total}" if total else str(done)), + (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(): + self.results_tree.insert( + "", + "end", + values=( + row.get("adapter", ""), + row.get("ip", ""), + row.get("mac", ""), + row.get("conflict_macs", ""), + row.get("evidence_type", ""), + row.get("risk_level", ""), + row.get("verdict", ""), + ), + tags=(row.get("risk_level", ""),), + ) + + 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_loop.py b/core/ui/tab_loop.py new file mode 100644 index 0000000..f5d492b --- /dev/null +++ b/core/ui/tab_loop.py @@ -0,0 +1,215 @@ +import threading +import tkinter as tk +from tkinter import filedialog, messagebox, ttk + +from core.Function.loop_fun import ALL_ADAPTERS, LoopDetector +from core.ui.components import Console, Page, action_bar, button, combo, field + + +class LoopTab(Page): + def __init__(self, parent): + super().__init__(parent, "环网检测", "基于本机证据判断疑似二层环路、广播风暴和网关抖动风险。") + self.body.rowconfigure(3, weight=1) + + status = self.section("实时状态", 0, columns=6) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.adapter_total = field(status, "网卡数", 1, 1, "0", 8) + self.current_adapter = field(status, "当前网卡", 1, 2, "-", 18) + self.risk_level = field(status, "风险等级", 1, 3, "正常", 10) + self.max_score = field(status, "最高分", 1, 4, "0", 8) + self.elapsed = field(status, "耗时", 1, 5, "0.0s", 10) + for item in (self.state, self.adapter_total, self.current_adapter, self.risk_level, self.max_score, self.elapsed): + item["entry"].configure(state="disabled") + + params = self.section("检测参数", 1, columns=5) + self.adapter = combo(params, "检测网卡", 1, 0, [ALL_ADAPTERS], ALL_ADAPTERS, 24) + self.duration = field(params, "检测时长 s", 1, 1, "15", 10) + self.interval = field(params, "采样间隔 s", 1, 2, "1", 10) + self.ping_timeout = field(params, "网关 Ping 超时 ms", 1, 3, "800", 12) + actions = action_bar(params, 2, 5) + self.start_btn = button(actions, "开始检测", self.start_detection, "Primary.TButton") + self.stop_btn = button(actions, "停止", self.stop_detection, "Danger.TButton") + self.refresh_btn = button(actions, "刷新网卡", self.load_adapters, "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=("adapter", "ipv4", "gateway", "non_unicast", "ratio", "loss", "jitter", "arp", "score", "level", "verdict"), + show="headings", + height=9, + ) + headings = { + "adapter": "网卡", + "ipv4": "IPv4", + "gateway": "网关", + "non_unicast": "非单播/s", + "ratio": "非单播占比", + "loss": "网关丢包", + "jitter": "网关抖动", + "arp": "ARP 异常", + "score": "分数", + "level": "等级", + "verdict": "判断", + } + widths = { + "adapter": 140, + "ipv4": 120, + "gateway": 120, + "non_unicast": 90, + "ratio": 90, + "loss": 80, + "jitter": 80, + "arp": 80, + "score": 60, + "level": 80, + "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.detector = LoopDetector(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": "等待", + "total_adapters": 0, + "current_adapter": "-", + "risk_level": "正常", + "max_score": 0, + "elapsed": 0, + } + ) + + def load_adapters(self): + def worker(): + try: + values = self.detector.get_adapter_choices() + except Exception as exc: + values = [ALL_ADAPTERS] + self.write(f"读取检测网卡失败: {exc}\n", "warning") + self.after(0, lambda: self.apply_adapters(values)) + + threading.Thread(target=worker, daemon=True).start() + + def apply_adapters(self, values): + 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[0]) + + def start_detection(self): + try: + self.clear() + self.start_btn.configure(state="disabled") + self.detector.start_detection( + self.adapter["var"].get(), + { + "duration_sec": self.duration["var"].get(), + "interval_sec": self.interval["var"].get(), + "ping_timeout_ms": self.ping_timeout["var"].get(), + }, + ) + except Exception as exc: + self.start_btn.configure(state="normal") + messagebox.showwarning("无法开始环网检测", str(exc)) + + def stop_detection(self): + try: + self.detector.stop_detection() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) + + def export_results(self): + path = filedialog.asksaveasfilename( + title="导出环网检测结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: + return + try: + self.detector.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_summary(self): + text = self.detector.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(): + current = stats.get("current_adapter") or "-" + values = [ + (self.state, stats.get("state", "等待")), + (self.adapter_total, str(stats.get("total_adapters", 0))), + (self.current_adapter, current), + (self.risk_level, stats.get("risk_level", "正常")), + (self.max_score, str(stats.get("max_score", 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(): + arp_score = int(row.get("gateway_mac_changes", 0)) + int(row.get("ip_mac_changes", 0)) + self.results_tree.insert( + "", + "end", + values=( + row.get("adapter", ""), + row.get("ipv4", ""), + row.get("gateway", ""), + f"{row.get('non_unicast_pps', 0):.1f}", + f"{row.get('non_unicast_ratio', 0):.1f}%", + f"{row.get('gateway_ping_loss', 0):.1f}%", + f"{row.get('gateway_jitter_ms', 0):.1f} ms", + str(arp_score), + row.get("risk_score", 0), + row.get("risk_level", ""), + row.get("verdict", ""), + ), + tags=(row.get("risk_level", ""),), + ) + + 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_tracert.py b/core/ui/tab_tracert.py index fac24c5..aaae9c2 100644 --- a/core/ui/tab_tracert.py +++ b/core/ui/tab_tracert.py @@ -1,43 +1,141 @@ -from tkinter import messagebox +import tkinter as tk +from tkinter import filedialog, messagebox, ttk from core.Function.tracert_fun import TracertFun -from core.ui.components import Console, Page, action_bar, button, field +from core.ui.components import Console, Page, action_bar, button, combo, field class TracertTab(Page): def __init__(self, parent): - super().__init__(parent, "路由追踪", "查看从本机到目标地址的网络跳点和响应时间。") - self.body.rowconfigure(1, weight=1) + super().__init__(parent, "路由追踪", "结构化查看跳点、延迟、超时、波动和诊断摘要。") + self.body.rowconfigure(3, weight=1) - target = self.section("追踪参数", 0, columns=4) - self.target = field(target, "目标 IP / 域名", 1, 0, "8.8.8.8", 28) - self.max_hops = field(target, "最大跳数", 1, 1, "20", 12) - self.timeout_ms = field(target, "单跳超时（ms）", 1, 2, "800", 14) - actions = action_bar(target, 2, 4) - self.start_btn = button(actions, "↗ 开始追踪", self.start_trace, "Primary.TButton") - self.stop_btn = button(actions, "■ 停止", self.stop_trace, "Danger.TButton") + status = self.section("实时状态", 0, columns=7) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.run_state = field(status, "轮次", 1, 1, "0/0", 8) + self.current_hop = field(status, "当前跳", 1, 2, "0", 8) + self.timeout_hops = field(status, "超时跳", 1, 3, "0", 8) + self.high_hops = field(status, "高延迟", 1, 4, "0", 8) + self.max_latency = field(status, "最大延迟", 1, 5, "0.0 ms", 12) + self.elapsed = field(status, "耗时", 1, 6, "0.0s", 10) + for item in ( + self.state, + self.run_state, + self.current_hop, + self.timeout_hops, + self.high_hops, + self.max_latency, + self.elapsed, + ): + item["entry"].configure(state="disabled") - output = self.section("输出控制台", 1, columns=1) + params = self.section("追踪参数", 1, columns=7) + self.target = field(params, "目标 IP / 域名", 1, 0, "8.8.8.8", 26) + self.address_family = combo(params, "地址族", 1, 1, ["自动", "IPv4", "IPv6"], "自动", 10) + self.mode = combo(params, "模式", 1, 2, ["单次", "指定次数", "持续"], "单次", 12) + self.repeat_count = field(params, "次数", 1, 3, "3", 8) + self.interval_ms = field(params, "间隔 ms", 1, 4, "1000", 10) + self.max_hops = field(params, "最大跳数", 1, 5, "20", 10) + self.timeout_ms = field(params, "单跳超时 ms", 1, 6, "800", 12) + self.high_latency_ms = field(params, "高延迟阈值 ms", 2, 0, "100", 12) + self.resolve_names_var = tk.BooleanVar(value=False) + self._check(params, "解析主机名", self.resolve_names_var, 2, 1) + actions = action_bar(params, 3, 7) + self.start_btn = button(actions, "开始追踪", self.start_trace, "Primary.TButton") + self.stop_btn = button(actions, "停止", self.stop_trace, "Danger.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=("run", "hop", "probe1", "probe2", "probe3", "avg", "jitter", "host", "ip", "status"), + show="headings", + height=9, + ) + headings = { + "run": "轮次", + "hop": "跳数", + "probe1": "RTT 1", + "probe2": "RTT 2", + "probe3": "RTT 3", + "avg": "平均", + "jitter": "波动", + "host": "主机名", + "ip": "IP", + "status": "判断", + } + widths = { + "run": 60, + "hop": 60, + "probe1": 70, + "probe2": 70, + "probe3": 70, + "avg": 80, + "jitter": 80, + "host": 200, + "ip": 150, + "status": 90, + } + 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("ok", foreground="#15803d") + self.results_tree.tag_configure("timeout", foreground="#b7791f") + self.results_tree.tag_configure("high_latency", foreground="#dc2626") + self.results_tree.tag_configure("jitter", foreground="#b7791f") + 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=22) + self.console = Console(output, height=12) self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) - self.tracert_fun = TracertFun(self.write, self.on_task_done) + self.tracert_fun = TracertFun(self.write, self.on_task_done, self.update_status, self.add_result) + + def _check(self, parent, text, variable, row, column): + frame = ttk.Frame(parent, style="Panel.TFrame") + frame.grid(row=row, column=column, sticky="ew", padx=18, pady=(22, 14)) + ttk.Checkbutton(frame, text=text, variable=variable).pack(anchor="w") 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_run": 0, + "total_runs": 0, + "current_hop": 0, + "timeout_hops": 0, + "high_latency_hops": 0, + "max_latency_ms": 0, + "elapsed": 0, + } + ) + def start_trace(self): try: - self.console.clear() + self.clear() self.tracert_fun.start_tracert( self.target["var"].get(), int(self.max_hops["var"].get()), int(self.timeout_ms["var"].get()), + self.trace_options(), ) self.start_btn.configure(state="disabled") except Exception as exc: + self.start_btn.configure(state="normal") messagebox.showwarning("无法开始路由追踪", str(exc)) def stop_trace(self): @@ -46,5 +144,82 @@ class TracertTab(Page): except Exception as exc: messagebox.showinfo("提示", str(exc)) + def trace_options(self): + return { + "address_family": self.address_family["var"].get(), + "resolve_names": self.resolve_names_var.get(), + "mode": self.mode["var"].get(), + "repeat_count": self.repeat_count["var"].get(), + "interval_ms": self.interval_ms["var"].get(), + "high_latency_ms": self.high_latency_ms["var"].get(), + } + + def export_results(self): + path = filedialog.asksaveasfilename( + title="导出路由追踪结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: + return + try: + self.tracert_fun.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_summary(self): + text = self.tracert_fun.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_runs = stats.get("total_runs", 0) + run_text = f"{stats.get('current_run', 0)}/持续" if total_runs == 0 else f"{stats.get('current_run', 0)}/{total_runs}" + values = [ + (self.state, stats.get("state", "等待")), + (self.run_state, run_text), + (self.current_hop, f"{stats.get('current_hop', 0)}/{stats.get('max_hops', 0)}"), + (self.timeout_hops, str(stats.get("timeout_hops", 0))), + (self.high_hops, str(stats.get("high_latency_hops", 0))), + (self.max_latency, f"{stats.get('max_latency_ms', 0):.1f} ms"), + (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(): + avg = f"{row.get('avg_ms', 0):.1f} ms" if row.get("avg_ms") else "-" + jitter = f"{row.get('jitter_ms', 0):.1f} ms" if row.get("jitter_ms") else "-" + self.results_tree.insert( + "", + "end", + values=( + row.get("run", ""), + row.get("hop", ""), + row.get("probe1", ""), + row.get("probe2", ""), + row.get("probe3", ""), + avg, + jitter, + row.get("host", ""), + row.get("ip", ""), + row.get("status_text", ""), + ), + 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 46bd097..7a363bf 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_ip_conflict import IpConflictTab +from core.ui.tab_loop import LoopTab from core.ui.tab_network import NetworkTab from core.ui.tab_ping import PingTab from core.ui.tab_telnet import TelnetTab @@ -62,6 +64,8 @@ class MainUI: ("ping", "⌁", "Ping 探测", "单点与批量探活"), ("ports", "⌕", "端口扫描", "TCP 连通性检测"), ("trace", "↗", "路由追踪", "跳点路径分析"), + ("loop", "◇", "环网检测", "二层环路风险"), + ("ip_conflict", "≠", "IP 冲突", "地址占用排查"), ] for key, icon, title, subtitle in items: self.nav_buttons[key] = self._nav_button(key, icon, title, subtitle) @@ -103,6 +107,8 @@ class MainUI: "ping": PingTab(self.content), "ports": TelnetTab(self.content), "trace": TracertTab(self.content), + "loop": LoopTab(self.content), + "ip_conflict": IpConflictTab(self.content), } for page in self.pages.values(): page.grid(row=0, column=0, sticky="nsew") diff --git a/main.py b/main.py index 40c565c..e410935 100644 --- a/main.py +++ b/main.py @@ -2,7 +2,6 @@ import os import sys import tkinter as tk -from core.logger_config import setup_logger from core.ui.ui_main import MainUI @@ -13,8 +12,6 @@ def get_base_dir() -> str: def main() -> None: - setup_logger() - root = tk.Tk() root.minsize(960, 600) MainUI(root, base_dir=get_base_dir())