Net

{
  "tshark_exe": "tshark",
  "capinfos_exe": "capinfos"
}

{
  "embedded_common_ports": [
    {"port": 20, "protocol": "FTP-Data", "transport": "TCP", "category": "file_transfer"},
    {"port": 21, "protocol": "FTP", "transport": "TCP", "category": "file_transfer"},
    {"port": 22, "protocol": "SSH", "transport": "TCP", "category": "remote_access"},
    {"port": 23, "protocol": "Telnet", "transport": "TCP", "category": "remote_access"},
    {"port": 25, "protocol": "SMTP", "transport": "TCP", "category": "email"},
    {"port": 53, "protocol": "DNS", "transport": "UDP/TCP", "category": "network"},
    {"port": 67, "protocol": "DHCP Server", "transport": "UDP", "category": "network"},
    {"port": 68, "protocol": "DHCP Client", "transport": "UDP", "category": "network"},
    {"port": 69, "protocol": "TFTP", "transport": "UDP", "category": "file_transfer"},
    {"port": 80, "protocol": "HTTP", "transport": "TCP", "category": "web"},
    {"port": 102, "protocol": "S7comm (Siemens)", "transport": "TCP", "category": "industrial"},
    {"port": 161, "protocol": "SNMP", "transport": "UDP", "category": "network_management"},
    {"port": 162, "protocol": "SNMP Trap", "transport": "UDP", "category": "network_management"},
    {"port": 443, "protocol": "HTTPS", "transport": "TCP", "category": "web"},
    {"port": 502, "protocol": "Modbus TCP", "transport": "TCP", "category": "industrial"},
    {"port": 554, "protocol": "RTSP", "transport": "TCP", "category": "streaming"},
    {"port": 1883, "protocol": "MQTT", "transport": "TCP", "category": "iot"},
    {"port": 2404, "protocol": "IEC 60870-5-104", "transport": "TCP", "category": "industrial"},
    {"port": 4840, "protocol": "OPC UA", "transport": "TCP", "category": "industrial"},
    {"port": 5060, "protocol": "SIP", "transport": "UDP/TCP", "category": "voip"},
    {"port": 5683, "protocol": "CoAP", "transport": "UDP", "category": "iot"},
    {"port": 8080, "protocol": "HTTP Alternate", "transport": "TCP", "category": "web"},
    {"port": 8443, "protocol": "HTTPS Alternate", "transport": "TCP", "category": "web"},
    {"port": 8883, "protocol": "MQTT over TLS", "transport": "TCP", "category": "iot"},
    {"port": 20000, "protocol": "DNP3", "transport": "TCP", "category": "industrial"},
    {"port": 44818, "protocol": "EtherNet/IP", "transport": "TCP/UDP", "category": "industrial"},
    {"port": 47808, "protocol": "BACnet/IP", "transport": "UDP", "category": "building_automation"}
  ],
  "industrial_protocols": {
    "modbus": {
      "name": "Modbus TCP/RTU",
      "default_port": 502,
      "tshark_filter": "modbus || mbtcp",
      "description": "工业自动化通信协议，常用于 PLC、HMI、传感器"
    },
    "s7comm": {
      "name": "S7comm (Siemens)",
      "default_port": 102,
      "tshark_filter": "s7comm",
      "description": "西门子 S7 系列 PLC 通信协议"
    },
    "ethernet_ip": {
      "name": "EtherNet/IP",
      "default_port": 44818,
      "tshark_filter": "enip || cip",
      "description": "罗克韦尔等厂商的工业以太网协议"
    },
    "opcua": {
      "name": "OPC UA",
      "default_port": 4840,
      "tshark_filter": "opcua",
      "description": "统一架构的工业通信标准"
    },
    "iec104": {
      "name": "IEC 60870-5-104",
      "default_port": 2404,
      "tshark_filter": "iec60870_104",
      "description": "电力系统远动通信协议"
    },
    "dnp3": {
      "name": "DNP3",
      "default_port": 20000,
      "tshark_filter": "dnp3",
      "description": "电力和水务系统 SCADA 协议"
    },
    "bacnet": {
      "name": "BACnet/IP",
      "default_port": 47808,
      "tshark_filter": "bacnet",
      "description": "楼宇自动化通信协议"
    }
  },
  "iot_protocols": {
    "mqtt": {
      "name": "MQTT",
      "default_port": 1883,
      "tls_port": 8883,
      "tshark_filter": "mqtt",
      "description": "轻量级 IoT 消息传输协议"
    },
    "coap": {
      "name": "CoAP",
      "default_port": 5683,
      "tshark_filter": "coap",
      "description": "受限应用协议，面向资源受限设备"
    }
  }
}

#!/usr/bin/env python3
"""基于 tshark/capinfos 的 pcap 离线分析工具。"""

import argparse
import io
import json
import os
import re
import subprocess
import sys
import tempfile

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from net_runtime import decode_text, load_local_config, resolve_tool_path


def load_config():
    return load_local_config()


def run_cmd(cmd, timeout=30):
    try:
        result = subprocess.run(cmd, capture_output=True, text=False, timeout=timeout)
        return decode_text(result.stdout), decode_text(result.stderr), result.returncode
    except FileNotFoundError:
        return "", f"命令未找到: {cmd[0]}", -1
    except subprocess.TimeoutExpired:
        return "", "命令超时", -2


def get_capinfos_summary(capinfos_exe, pcap_file):
    """通过 capinfos 获取文件级统计。"""
    stdout, stderr, rc = run_cmd([capinfos_exe, "-M", pcap_file])
    if rc != 0:
        return None

    info = {}
    for line in stdout.splitlines():
        if ":" not in line:
            continue
        key, _, val = line.partition(":")
        key = key.strip().lower()
        val = val.strip()
        if "number of packets" in key:
            info["packet_count"] = int(val) if val.isdigit() else val
        elif "capture duration" in key:
            info["duration"] = val
        elif "file size" in key:
            info["file_size"] = val
        elif "data size" in key:
            info["data_size"] = val
        elif "first packet time" in key:
            info["first_packet"] = val
        elif "last packet time" in key:
            info["last_packet"] = val
        elif "average packet size" in key:
            info["avg_packet_size"] = val
        elif "data byte rate" in key:
            info["byte_rate"] = val
    return info


def get_protocol_hierarchy(tshark_exe, pcap_file, display_filter="", decode_as=""):
    """获取协议层次统计。"""
    cmd = [tshark_exe, "-r", pcap_file, "-q", "-z", "io,phs"]
    if display_filter:
        cmd += ["-Y", display_filter]
    if decode_as:
        cmd += ["-d", decode_as]

    stdout, _, rc = run_cmd(cmd, timeout=60)
    if rc != 0:
        return []

    protocols = []
    for line in stdout.splitlines():
        line = line.strip()
        m = re.match(r"(\S+)\s+frames:(\d+)\s+bytes:(\d+)", line)
        if m:
            protocols.append({
                "protocol": m.group(1),
                "frames": int(m.group(2)),
                "bytes": int(m.group(3)),
            })
    return protocols


def get_conversations(tshark_exe, pcap_file, display_filter="", decode_as="", top=20):
    """获取会话统计。"""
    cmd = [tshark_exe, "-r", pcap_file, "-q", "-z", "conv,ip"]
    if display_filter:
        cmd += ["-Y", display_filter]
    if decode_as:
        cmd += ["-d", decode_as]

    stdout, _, rc = run_cmd(cmd, timeout=60)
    if rc != 0:
        return []

    conversations = []
    header_found = False
    for line in stdout.splitlines():
        line = line.strip()
        if not line or line.startswith("="):
            header_found = True
            continue
        if not header_found:
            continue
        # 格式: addr_a <-> addr_b  frames_a bytes_a frames_b bytes_b frames_total bytes_total ...
        parts = re.split(r"\s+", line)
        if len(parts) >= 8 and "<->" in parts:
            idx = parts.index("<->")
            if idx >= 1 and idx + 1 < len(parts):
                conversations.append({
                    "addr_a": parts[idx - 1],
                    "addr_b": parts[idx + 1],
                    "raw": line,
                })
    return conversations[:top]


def get_endpoints(tshark_exe, pcap_file, display_filter="", decode_as="", top=20):
    """获取端点统计。"""
    cmd = [tshark_exe, "-r", pcap_file, "-q", "-z", "endpoints,ip"]
    if display_filter:
        cmd += ["-Y", display_filter]
    if decode_as:
        cmd += ["-d", decode_as]

    stdout, _, rc = run_cmd(cmd, timeout=60)
    if rc != 0:
        return []

    endpoints = []
    header_found = False
    for line in stdout.splitlines():
        line = line.strip()
        if not line or line.startswith("="):
            header_found = True
            continue
        if not header_found or "Address" in line or line.startswith("|"):
            continue
        parts = re.split(r"\s+", line)
        if len(parts) >= 3:
            endpoints.append({
                "address": parts[0],
                "packets": parts[1] if len(parts) > 1 else "",
                "bytes": parts[2] if len(parts) > 2 else "",
                "raw": line,
            })
    return endpoints[:top]


def detect_anomalies(tshark_exe, pcap_file, display_filter="", decode_as=""):
    """检测常见网络异常（重传、RST、错误等）。"""
    checks = [
        ("tcp.analysis.retransmission", "TCP 重传"),
        ("tcp.analysis.fast_retransmission", "TCP 快速重传"),
        ("tcp.analysis.duplicate_ack", "TCP 重复 ACK"),
        ("tcp.flags.reset==1", "TCP RST"),
        ("icmp.type==3", "ICMP 不可达"),
        ("dns.flags.rcode!=0", "DNS 错误"),
        ("tcp.analysis.zero_window", "TCP 零窗口"),
    ]

    anomalies = []
    for filt, desc in checks:
        combined = f"({filt})"
        if display_filter:
            combined = f"({display_filter}) && ({filt})"

        cmd = [tshark_exe, "-r", pcap_file, "-Y", combined, "-T", "fields", "-e", "frame.number"]
        if decode_as:
            cmd += ["-d", decode_as]

        stdout, _, rc = run_cmd(cmd, timeout=30)
        if rc == 0:
            count = len([l for l in stdout.strip().splitlines() if l.strip()])
            if count > 0:
                anomalies.append({"type": desc, "filter": filt, "count": count})

    return anomalies


def get_io_stats(tshark_exe, pcap_file, display_filter="", decode_as=""):
    """获取 IO 统计。"""
    cmd = [tshark_exe, "-r", pcap_file, "-q", "-z", "io,stat,1"]
    if display_filter:
        cmd += ["-Y", display_filter]
    if decode_as:
        cmd += ["-d", decode_as]

    stdout, _, rc = run_cmd(cmd, timeout=60)
    if rc != 0:
        return []

    intervals = []
    for line in stdout.splitlines():
        m = re.match(r"\|\s*([\d.]+)\s*<>\s*([\d.]+)\s*\|\s*(\d+)\s*\|\s*(\d+)\s*\|", line)
        if m:
            intervals.append({
                "start": float(m.group(1)),
                "end": float(m.group(2)),
                "frames": int(m.group(3)),
                "bytes": int(m.group(4)),
            })
    return intervals


def main():
    parser = argparse.ArgumentParser(description="分析 pcap 文件")
    parser.add_argument("pcap_file", help="pcap/pcapng 文件路径")
    parser.add_argument("--mode", default="summary",
                        choices=["summary", "protocols", "conversations", "endpoints", "io", "anomalies", "all"])
    parser.add_argument("--filter", default="", help="显示过滤器")
    parser.add_argument("--top", type=int, default=20, help="显示前 N 条")
    parser.add_argument("--decode-as", default="", help="解码规则")
    parser.add_argument("--export-fields", default="", help="导出字段列表")
    parser.add_argument("--output", default="", help="CSV 输出路径")
    parser.add_argument("--json", action="store_true", dest="output_json", help="JSON 输出")
    args = parser.parse_args()

    if not os.path.exists(args.pcap_file):
        error = {
            "status": "error",
            "action": "analyze",
            "error": {"code": "file_not_found", "message": f"文件不存在: {args.pcap_file}"},
        }
        print(json.dumps(error, ensure_ascii=False, indent=2))
        sys.exit(1)

    config = load_config()
    tshark_exe = resolve_tool_path(
        config.get("tshark_exe"),
        "tshark.exe" if sys.platform == "win32" else "tshark",
    )
    capinfos_exe = resolve_tool_path(
        config.get("capinfos_exe"),
        "capinfos.exe" if sys.platform == "win32" else "capinfos",
    )

    filtered_input = ""
    analysis_input = args.pcap_file

    if args.filter:
        fd, filtered_input = tempfile.mkstemp(prefix="net_analyze_filtered_", suffix=os.path.splitext(args.pcap_file)[1] or ".pcapng")
        os.close(fd)
        filter_cmd = [tshark_exe, "-r", args.pcap_file, "-Y", args.filter, "-w", filtered_input]
        if args.decode_as:
            filter_cmd += ["-d", args.decode_as]
        stdout, stderr, rc = run_cmd(filter_cmd, timeout=120)
        if rc != 0:
            error = {
                "status": "error",
                "action": "analyze",
                "error": {"code": "filter_failed", "message": stderr.strip() or "过滤失败"},
            }
            print(json.dumps(error, ensure_ascii=False, indent=2))
            sys.exit(1)
        analysis_input = filtered_input

    result = {
        "status": "ok",
        "action": "analyze",
        "summary": "",
        "details": {},
    }

    modes = [args.mode] if args.mode != "all" else [
        "summary", "protocols", "conversations", "endpoints", "io", "anomalies"
    ]

    for mode in modes:
        if mode == "summary":
            info = get_capinfos_summary(capinfos_exe, analysis_input)
            if info:
                result["details"]["summary"] = info
                result["summary"] = f"文件包含 {info.get('packet_count', '?')} 个数据包"
            else:
                # 回退用 tshark 统计
                stdout, _, rc = run_cmd([tshark_exe, "-r", analysis_input, "-q", "-z", "io,stat,0"])
                result["details"]["summary"] = {"raw": stdout}

        elif mode == "protocols":
            result["details"]["protocols"] = get_protocol_hierarchy(
                tshark_exe, analysis_input, "", args.decode_as
            )

        elif mode == "conversations":
            result["details"]["conversations"] = get_conversations(
                tshark_exe, analysis_input, "", args.decode_as, args.top
            )

        elif mode == "endpoints":
            result["details"]["endpoints"] = get_endpoints(
                tshark_exe, analysis_input, "", args.decode_as, args.top
            )

        elif mode == "io":
            result["details"]["io_stats"] = get_io_stats(
                tshark_exe, analysis_input, "", args.decode_as
            )

        elif mode == "anomalies":
            result["details"]["anomalies"] = detect_anomalies(
                tshark_exe, analysis_input, "", args.decode_as
            )

    # 导出字段
    if args.export_fields and args.output:
        fields = [f.strip() for f in args.export_fields.split(",")]
        cmd = [tshark_exe, "-r", analysis_input, "-T", "fields"]
        for f in fields:
            cmd += ["-e", f]
        cmd += ["-E", "header=y", "-E", "separator=,"]
        if args.decode_as:
            cmd += ["-d", args.decode_as]

        stdout, _, rc = run_cmd(cmd, timeout=120)
        if rc == 0:
            with open(args.output, "w", encoding="utf-8") as f:
                f.write(stdout)
            result["details"]["exported"] = {"file": args.output, "fields": fields}

    try:
        if args.output_json:
            print(json.dumps(result, ensure_ascii=False, indent=2))
        else:
            print(f"[net analyze] {result.get('summary', '分析完成')}")
            details = result["details"]
            if "summary" in details and isinstance(details["summary"], dict):
                for k, v in details["summary"].items():
                    if k != "raw":
                        print(f"  {k}: {v}")
            if "protocols" in details:
                print("\n  协议统计:")
                for p in details["protocols"][:args.top]:
                    print(f"    {p['protocol']}: {p['frames']} frames, {p['bytes']} bytes")
            if "anomalies" in details and details["anomalies"]:
                print("\n  异常检测:")
                for a in details["anomalies"]:
                    print(f"    {a['type']}: {a['count']} 次")
            if "conversations" in details:
                print("\n  会话:")
                for c in details["conversations"][:5]:
                    print(f"    {c['addr_a']} <-> {c['addr_b']}")
            if "endpoints" in details:
                print("\n  端点:")
                for e in details["endpoints"][:5]:
                    print(f"    {e['address']}")
    finally:
        if filtered_input and os.path.exists(filtered_input):
            os.remove(filtered_input)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""基于 tshark 的抓包工具，支持保存文件、过滤、解码规则和结构化输出。"""

import argparse
import io
import json
import os
import subprocess
import sys
import signal
import tempfile

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from net_runtime import (
    decode_text,
    get_net_config,
    save_project_config,
    update_state_entry,
    check_tshark,
)


def build_tshark_cmd(config, args, *, output_path="", include_display_filter=True):
    exe = config["tshark_exe"]
    cmd = [exe]

    # 接口
    iface = config["interface"]
    if iface:
        cmd += ["-i", str(iface)]

    # 抓包过滤器 (BPF)
    capture_filter = config["capture_filter"]
    if capture_filter:
        cmd += ["-f", capture_filter]

    # 显示过滤器
    display_filter = config["display_filter"]
    if display_filter and include_display_filter:
        cmd += ["-Y", display_filter]

    # 持续时间
    duration = config["duration"]
    cmd += ["-a", f"duration:{duration}"]

    # 输出文件
    if output_path:
        fmt = args.format or config["capture_format"]
        cmd += ["-w", output_path]
        if fmt == "pcap":
            cmd += ["-F", "pcap"]

    # 解码规则
    if args.decode_as:
        cmd += ["-d", args.decode_as]

    # JSON Lines 输出 (使用 -T ek)
    if args.output_json and not args.output:
        cmd += ["-T", "ek"]

    return cmd, exe


def main():
    parser = argparse.ArgumentParser(description="tshark 抓包")
    parser.add_argument("--interface", "-i", help="抓包接口")
    parser.add_argument("--duration", type=int, help="抓包时长(秒)")
    parser.add_argument("--capture-filter", "-f", help="抓包过滤器(BPF)")
    parser.add_argument("--display-filter", "-Y", help="显示过滤器")
    parser.add_argument("--output", "-o", default="", help="保存抓包文件路径")
    parser.add_argument("--format", choices=["pcapng", "pcap"], help="抓包文件格式")
    parser.add_argument("--decode-as", default="", help="自定义解码规则")
    parser.add_argument("--json", action="store_true", dest="output_json", help="JSON Lines 输出")
    args = parser.parse_args()

    # 获取配置
    config, sources = get_net_config(
        cli_interface=args.interface,
        cli_duration=args.duration,
        cli_capture_filter=args.capture_filter,
        cli_display_filter=args.display_filter,
    )

    exe = config["tshark_exe"]

    if not check_tshark(exe):
        error = {
            "status": "error",
            "action": "capture",
            "error": {
                "code": "tshark_not_found",
                "message": f"未找到 tshark ({exe})，请确认 Wireshark 已安装且已加入 PATH",
            },
        }
        print(json.dumps(error, ensure_ascii=False, indent=2))
        sys.exit(1)

    iface = config["interface"]
    if not iface:
        error = {
            "status": "error",
            "action": "capture",
            "error": {
                "code": "no_interface",
                "message": "未配置抓包接口，请用 --interface 指定或在 .embeddedskills/config.json 中配置",
            },
        }
        print(json.dumps(error, ensure_ascii=False, indent=2))
        sys.exit(1)

    # 保存确认的配置
    save_project_config(values={
        "interface": iface,
        "duration": config["duration"],
        "capture_filter": config["capture_filter"],
        "display_filter": config["display_filter"],
    })

    filter_after_capture = bool(args.output and config.get("display_filter"))
    temp_output = ""
    output_path = args.output
    if filter_after_capture:
        fd, temp_output = tempfile.mkstemp(
            prefix="net_capture_",
            suffix=".pcap" if (args.format or config["capture_format"]) == "pcap" else ".pcapng",
        )
        os.close(fd)
        output_path = temp_output

    cmd, _ = build_tshark_cmd(
        config,
        args,
        output_path=output_path,
        include_display_filter=not filter_after_capture,
    )
    duration = config["duration"]

    print(f"[net capture] 接口={iface}, 时长={duration}s", file=sys.stderr)
    if config.get("capture_filter"):
        print(f"  抓包过滤器: {config['capture_filter']}", file=sys.stderr)
    if config.get("display_filter"):
        print(f"  显示过滤器: {config['display_filter']}", file=sys.stderr)
    if args.output:
        print(f"  输出文件: {args.output}", file=sys.stderr)
        if filter_after_capture:
            print("  保存策略: 先原始抓包，再按显示过滤器离线筛选", file=sys.stderr)
    print(f"  命令: {' '.join(cmd)}", file=sys.stderr)

    try:
        proc = subprocess.Popen(
            cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False
        )
        stdout_data, stderr_data = proc.communicate()
        stdout_text = decode_text(stdout_data)
        stderr_output = decode_text(stderr_data)

        if stdout_text:
            print(stdout_text, end="")

        if proc.returncode != 0:
            error = {
                "status": "error",
                "action": "capture",
                "error": {
                    "code": "capture_failed",
                    "message": stderr_output.strip() or f"tshark 退出码 {proc.returncode}",
                },
            }
            print(json.dumps(error, ensure_ascii=False, indent=2))
            sys.exit(1)

        if filter_after_capture:
            filter_cmd = [exe, "-r", temp_output, "-Y", config["display_filter"], "-w", args.output]
            if (args.format or config["capture_format"]) == "pcap":
                filter_cmd += ["-F", "pcap"]
            if args.decode_as:
                filter_cmd += ["-d", args.decode_as]

            filtered = subprocess.run(
                filter_cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=False,
            )
            filtered_stderr = decode_text(filtered.stderr)
            if filtered.returncode != 0:
                error = {
                    "status": "error",
                    "action": "capture",
                    "error": {
                        "code": "capture_filter_failed",
                        "message": filtered_stderr.strip() or f"过滤失败，退出码 {filtered.returncode}",
                    },
                }
                print(json.dumps(error, ensure_ascii=False, indent=2))
                sys.exit(1)
            if filtered_stderr.strip():
                print(filtered_stderr, file=sys.stderr)

        if stderr_output.strip():
            print(stderr_output, file=sys.stderr)

        # 输出摘要
        summary = {"status": "ok", "action": "capture", "summary": f"抓包完成，时长 {duration}s"}
        if args.output and os.path.exists(args.output):
            size = os.path.getsize(args.output)
            summary["summary"] += f"，文件: {args.output} ({size} bytes)"
            summary["details"] = {"output_file": args.output, "file_size": size}

        print(json.dumps(summary, ensure_ascii=False, indent=2), file=sys.stderr)

        # 更新状态
        update_state_entry("last_observe", {
            "type": "net_capture",
            "interface": iface,
            "duration": duration,
            "output_file": args.output if args.output and os.path.exists(args.output) else None,
        })

    except KeyboardInterrupt:
        proc.terminate()
        print("\n[net capture] 用户中断抓包", file=sys.stderr)
    except Exception as e:
        error = {
            "status": "error",
            "action": "capture",
            "error": {"code": "capture_failed", "message": str(e)},
        }
        print(json.dumps(error, ensure_ascii=False, indent=2))
        sys.exit(1)
    finally:
        if temp_output and os.path.exists(temp_output):
            os.remove(temp_output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""网络接口发现工具，可关联 tshark 抓包接口列表。"""

import argparse
import io
import json
import sys

# 确保 stdout 使用 UTF-8 编码
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from net_runtime import parse_ipconfig, parse_tshark_interfaces


def main():
    parser = argparse.ArgumentParser(description="列出网络接口")
    parser.add_argument("--filter", default="", help="按关键词筛选接口")
    parser.add_argument("--tshark", action="store_true", help="同时显示 tshark 抓包接口")
    parser.add_argument("--json", action="store_true", dest="output_json", help="JSON 输出")
    parser.add_argument("--tshark-exe", default="tshark", help="tshark 路径")
    args = parser.parse_args()

    interfaces = parse_ipconfig()

    if args.filter:
        kw = args.filter.lower()
        interfaces = [
            iface for iface in interfaces
            if kw in iface["name"].lower()
            or kw in iface["description"].lower()
            or kw in iface["type"].lower()
            or kw in iface.get("ipv4", "").lower()
            or any(kw in ip.lower() for ip in iface.get("ipv4_list", []))
        ]

    result = {
        "status": "ok",
        "action": "iface",
        "summary": f"发现 {len(interfaces)} 个网络接口",
        "details": {
            "interfaces": interfaces,
        },
    }

    if args.tshark:
        tshark_ifaces = parse_tshark_interfaces(args.tshark_exe)
        if tshark_ifaces is None:
            result["details"]["tshark_interfaces"] = []
            result["details"]["tshark_note"] = "tshark 不可用或未找到"
        else:
            result["details"]["tshark_interfaces"] = tshark_ifaces

    if args.output_json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(f"[net iface] {result['summary']}")
        for iface in interfaces:
            status_icon = "●" if iface["status"] == "up" else "○"
            print(f"  {status_icon} {iface['name']} ({iface['type']})")
            if iface["description"]:
                print(f"    描述: {iface['description']}")
            ipv4_list = iface.get("ipv4_list") or ([iface["ipv4"]] if iface["ipv4"] else [])
            subnet_list = iface.get("subnet_list") or ([iface["subnet"]] if iface["subnet"] else [])
            if ipv4_list:
                paired = []
                for index, ip in enumerate(ipv4_list):
                    subnet = subnet_list[index] if index < len(subnet_list) else iface.get("subnet", "")
                    paired.append(f"{ip}/{subnet}" if subnet else ip)
                print(f"    IPv4: {', '.join(paired)}")
            if iface["mac"]:
                print(f"    MAC:  {iface['mac']}")
            gateway_list = iface.get("gateway_list") or ([iface["gateway"]] if iface["gateway"] else [])
            if gateway_list:
                print(f"    网关: {', '.join(gateway_list)}")
        if args.tshark and result["details"].get("tshark_interfaces"):
            print("\n[tshark 抓包接口]")
            for ti in result["details"]["tshark_interfaces"]:
                print(f"  {ti['index']}. {ti['device']}")
                if ti["description"]:
                    print(f"     {ti['description']}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""连通性测试工具，支持 ICMP/TCP ping、批量测试和路由追踪。"""

import argparse
import io
import json
import os
import re
import socket
import subprocess
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from net_runtime import (
    get_net_config,
    save_project_config,
    update_state_entry,
)


def icmp_ping(target, count=4, timeout_ms=1000):
    """使用系统 ping 命令做 ICMP 测试。"""
    timeout_sec = max(1, timeout_ms // 1000)
    cmd = ["ping", "-n", str(count), "-w", str(timeout_ms), target]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, encoding="gbk",
                                errors="replace", timeout=count * timeout_sec + 10)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"target": target, "reachable": False, "error": "ping 命令超时或不可用"}

    output = result.stdout
    reachable = False
    sent = received = 0
    avg_ms = None

    for line in output.splitlines():
        # 统计行
        m = re.search(r"已发送\s*=\s*(\d+).*已接收\s*=\s*(\d+)", line)
        if not m:
            m = re.search(r"Sent\s*=\s*(\d+).*Received\s*=\s*(\d+)", line, re.IGNORECASE)
        if m:
            sent = int(m.group(1))
            received = int(m.group(2))
            reachable = received > 0

        # 平均延迟
        m2 = re.search(r"平均\s*=\s*(\d+)ms", line)
        if not m2:
            m2 = re.search(r"Average\s*=\s*(\d+)ms", line, re.IGNORECASE)
        if m2:
            avg_ms = int(m2.group(1))

    return {
        "target": target,
        "reachable": reachable,
        "sent": sent,
        "received": received,
        "loss_rate": f"{((sent - received) / sent * 100):.0f}%" if sent > 0 else "N/A",
        "avg_ms": avg_ms,
    }


def tcp_ping(target, port, timeout_ms=1000):
    """TCP 连通性测试。"""
    timeout_sec = timeout_ms / 1000.0
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout_sec)
        start = __import__("time").time()
        sock.connect((target, port))
        elapsed = (__import__("time").time() - start) * 1000
        sock.close()
        return {"target": target, "port": port, "reachable": True, "latency_ms": round(elapsed, 1)}
    except (socket.timeout, ConnectionRefusedError, OSError) as e:
        return {"target": target, "port": port, "reachable": False, "error": str(e)}


def traceroute(target, timeout_ms=1000):
    """路由追踪。"""
    timeout_sec = max(1, timeout_ms // 1000)
    cmd = ["tracert", "-d", "-w", str(timeout_ms), "-h", "30", target]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, encoding="gbk",
                                errors="replace", timeout=60)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"target": target, "hops": [], "error": "tracert 超时或不可用"}

    hops = []
    for line in result.stdout.splitlines():
        m = re.match(r"\s*(\d+)\s+(.+)", line)
        if m:
            hop_num = int(m.group(1))
            rest = m.group(2).strip()
            hops.append({"hop": hop_num, "detail": rest})

    reachable = any(target in hop["detail"] for hop in hops)
    return {"target": target, "hops": hops, "reachable": reachable}


def main():
    parser = argparse.ArgumentParser(description="连通性测试")
    parser.add_argument("--target", "-t", help="目标地址")
    parser.add_argument("--tcp", type=int, default=0, help="TCP 端口")
    parser.add_argument("--count", type=int, default=4, help="ping 次数")
    parser.add_argument("--traceroute", action="store_true", help="路由追踪")
    parser.add_argument("--concurrent", type=int, default=4, help="并发线程数")
    parser.add_argument("--timeout", type=int, help="超时(毫秒)")
    parser.add_argument("--json", action="store_true", dest="output_json", help="JSON 输出")
    args = parser.parse_args()

    # 获取配置
    config, sources = get_net_config(
        cli_target=args.target,
        cli_timeout_ms=args.timeout,
    )

    target = config["target"]
    timeout_ms = config["timeout_ms"]

    if not target:
        error = {
            "status": "error",
            "action": "ping",
            "error": {"code": "no_target", "message": "未配置目标地址，请用 --target 指定或在 .embeddedskills/config.json 中配置"},
        }
        print(json.dumps(error, ensure_ascii=False, indent=2))
        sys.exit(1)

    # 保存确认的配置
    save_project_config(values={
        "target": target,
        "timeout_ms": timeout_ms,
    })

    # 支持逗号分隔的多目标
    targets = [t.strip() for t in target.split(",") if t.strip()]

    results = []

    if args.traceroute:
        for t in targets:
            results.append(traceroute(t, timeout_ms))
    elif args.tcp > 0:
        with ThreadPoolExecutor(max_workers=args.concurrent) as pool:
            futures = {pool.submit(tcp_ping, t, args.tcp, timeout_ms): t for t in targets}
            for future in as_completed(futures):
                results.append(future.result())
    else:
        with ThreadPoolExecutor(max_workers=args.concurrent) as pool:
            futures = {pool.submit(icmp_ping, t, args.count, timeout_ms): t for t in targets}
            for future in as_completed(futures):
                results.append(future.result())

    reachable_count = sum(1 for r in results if r.get("reachable", False))
    total = len(results)
    action = "traceroute" if args.traceroute else ("tcp_ping" if args.tcp > 0 else "ping")

    output = {
        "status": "ok",
        "action": action,
        "summary": {
            "total": total,
            "reachable": reachable_count,
            "description": f"{reachable_count}/{total} 目标可达",
        },
        "details": {"results": results},
    }

    if args.output_json:
        print(json.dumps(output, ensure_ascii=False, indent=2))
    else:
        print(f"[net {action}] {output['summary']['description']}")
        for r in results:
            if args.traceroute:
                print(f"\n  追踪 {r['target']}:")
                for h in r.get("hops", []):
                    print(f"    {h['hop']:>3}  {h['detail']}")
            else:
                icon = "+" if r.get("reachable") else "x"
                line = f"  [{icon}] {r['target']}"
                if r.get("port"):
                    line += f":{r['port']}"
                if r.get("avg_ms") is not None:
                    line += f"  延迟={r['avg_ms']}ms"
                elif r.get("latency_ms") is not None:
                    line += f"  延迟={r['latency_ms']}ms"
                if r.get("loss_rate"):
                    line += f"  丢包={r['loss_rate']}"
                if r.get("error"):
                    line += f"  ({r['error']})"
                print(line)

    # 更新状态
    update_state_entry("last_net_ping", {
        "target": target,
        "reachable_count": reachable_count,
        "total": total,
    })


if __name__ == "__main__":
    main()

"""net skill 私有运行时工具。"""

from __future__ import annotations

import json
import re
import shutil
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


SKILL_DIR = Path(__file__).resolve().parent.parent
SKILL_NAME = "net"
STATE_DIR_NAME = ".embeddedskills"
STATE_FILE_NAME = "state.json"
PROJECT_CONFIG_FILE = "config.json"

WINDOWS_TOOL_DIRS = [
    Path(r"C:\Program Files\Wireshark"),
    Path(r"C:\Program Files (x86)\Wireshark"),
]


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def is_missing(value: Any) -> bool:
    return value is None or value == ""


def decode_text(data: bytes | str | None) -> str:
    """以稳健方式解码命令输出，兼容 Windows 下工具的混合编码。"""
    if data is None:
        return ""
    if isinstance(data, str):
        return data

    for encoding in ("utf-8", "gbk", "cp1252", sys.getdefaultencoding()):
        try:
            return data.decode(encoding)
        except UnicodeDecodeError:
            continue
    return data.decode("utf-8", errors="replace")


def looks_like_ipv4(value: str) -> bool:
    return bool(re.fullmatch(r"(?:\d{1,3}\.){3}\d{1,3}", value))


def looks_like_ip(value: str) -> bool:
    return looks_like_ipv4(value) or (":" in value and bool(re.fullmatch(r"[0-9a-fA-F:]+(?:%\d+)?", value)))


def resolve_tool_path(configured: str | None, default_name: str) -> str:
    """解析工具路径，优先使用配置，其次 PATH，最后尝试常见安装目录。"""
    candidates: list[str] = []
    if configured and configured.strip():
        candidates.append(configured.strip())
    candidates.append(default_name)

    seen: set[str] = set()
    for candidate in candidates:
        if candidate in seen:
            continue
        seen.add(candidate)

        expanded = str(Path(candidate).expanduser())
        if Path(expanded).exists():
            return expanded

        resolved = shutil.which(candidate)
        if resolved:
            return resolved

    for base_dir in WINDOWS_TOOL_DIRS:
        candidate_path = base_dir / default_name
        if candidate_path.exists():
            return str(candidate_path)

    return configured.strip() if configured and configured.strip() else default_name


def load_json_file(path: str | Path) -> dict:
    """加载 JSON 文件，不存在返回空字典"""
    file_path = Path(path)
    if not file_path.exists():
        return {}
    try:
        return json.loads(file_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def save_json_file(path: str | Path, data: dict) -> None:
    """保存 JSON 文件，自动创建目录"""
    file_path = Path(path)
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def load_local_config() -> dict:
    """加载 skill/config.json（环境级配置）"""
    return load_json_file(SKILL_DIR / "config.json")


def save_local_config(data: dict) -> None:
    """保存环境级配置到 skill/config.json"""
    save_json_file(SKILL_DIR / "config.json", data)


def workspace_root(workspace: str | None = None) -> Path:
    if not is_missing(workspace):
        return Path(str(workspace)).expanduser().resolve()
    return Path.cwd().resolve()


def load_project_config(workspace: str | None = None) -> dict:
    """从 workspace/.embeddedskills/config.json 读取本 skill 的工程级配置"""
    proj_config = load_json_file(workspace_root(workspace) / STATE_DIR_NAME / PROJECT_CONFIG_FILE)
    return proj_config.get(SKILL_NAME, {})


def save_project_config(workspace: str | None = None, values: dict | None = None) -> None:
    """写回工程级配置，只更新本 skill 的部分"""
    if values is None:
        return
    proj_path = workspace_root(workspace) / STATE_DIR_NAME / PROJECT_CONFIG_FILE
    proj_config = load_json_file(proj_path)
    proj_config[SKILL_NAME] = {**proj_config.get(SKILL_NAME, {}), **values}
    save_json_file(proj_path, proj_config)


def load_workspace_state(workspace: str | None = None) -> dict:
    """从 workspace/.embeddedskills/state.json 读取状态"""
    return load_json_file(workspace_root(workspace) / STATE_DIR_NAME / STATE_FILE_NAME)


def save_workspace_state(state: dict, workspace: str | None = None) -> Path:
    """保存状态"""
    file_path = workspace_root(workspace) / STATE_DIR_NAME / STATE_FILE_NAME
    save_json_file(file_path, state)
    return file_path


def update_state_entry(category: str, record: dict, workspace: str | None = None) -> dict:
    """更新状态条目"""
    state = load_workspace_state(workspace)
    state[category] = {**record, "timestamp": record.get("timestamp") or now_iso()}
    file_path = save_workspace_state(state, workspace)
    return {
        "workspace": str(workspace_root(workspace)),
        "file": str(file_path),
        "updated_keys": [category],
        category: state[category],
    }


def normalize_path(value: str | None, base: str | Path | None = None) -> str:
    """路径规范化"""
    if is_missing(value):
        return ""
    path = Path(str(value)).expanduser()
    if base and not path.is_absolute():
        path = Path(base) / path
    return str(path.resolve()) if path.is_absolute() else str(path)


def _first_resolved(mapping: dict, keys: list[str]) -> tuple[Any, str | None]:
    for key in keys:
        value = mapping.get(key)
        if not is_missing(value):
            return value, key
    return None, None


def resolve_param(
    name: str,
    cli_value: Any = None,
    local_config: dict | None = None,
    local_keys: list[str] | None = None,
    project_config: dict | None = None,
    project_keys: list[str] | None = None,
    state: dict | None = None,
    state_keys: list[str] | None = None,
    default: Any = None,
) -> tuple[Any, str]:
    """统一参数解析，优先级: CLI > 环境级 > 工程级 > state > default"""
    if not is_missing(cli_value):
        return cli_value, "cli"

    if local_config and local_keys:
        value, key = _first_resolved(local_config, local_keys)
        if not is_missing(value):
            return value, f"local:{key}"

    if project_config and project_keys:
        value, key = _first_resolved(project_config, project_keys)
        if not is_missing(value):
            return value, f"project:{key}"

    if state and state_keys:
        value, key = _first_resolved(state, state_keys)
        if not is_missing(value):
            return value, f"state:{key}"

    if not is_missing(default):
        return default, "default"

    return None, ""


def parameter_context(name: str, value: Any, source: str) -> dict:
    """记录参数来源"""
    return {"name": name, "value": value, "source": source}


def make_result(
    success: bool = True,
    action: str = "",
    summary: str = "",
    details: dict | None = None,
    error: dict | None = None,
) -> dict:
    """统一结果格式"""
    result = {
        "status": "ok" if success else "error",
        "action": action,
        "summary": summary,
    }
    if details:
        result["details"] = details
    if error:
        result["error"] = error
    return result


def make_timing(start_time: float) -> dict:
    """执行时间记录"""
    elapsed = datetime.now().timestamp() - start_time
    return {
        "started_at": datetime.fromtimestamp(start_time).astimezone().isoformat(timespec="seconds"),
        "finished_at": now_iso(),
        "elapsed_ms": int(elapsed * 1000),
    }


def check_tshark(exe: str = "tshark") -> bool:
    """检查 tshark 是否可用"""
    resolved_exe = resolve_tool_path(exe, "tshark.exe" if sys.platform == "win32" else "tshark")
    try:
        result = subprocess.run([resolved_exe, "--version"], capture_output=True, text=False, timeout=5)
        return result.returncode == 0
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False


def parse_tshark_interfaces(tshark_exe: str = "tshark") -> list[dict] | None:
    """解析 tshark -D 获取抓包接口列表"""
    resolved_exe = resolve_tool_path(tshark_exe, "tshark.exe" if sys.platform == "win32" else "tshark")
    try:
        result = subprocess.run(
            [resolved_exe, "-D"], capture_output=True, text=False, timeout=10
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return None

    if result.returncode != 0:
        return None

    interfaces = []
    for line in decode_text(result.stdout).splitlines():
        line = line.strip()
        if not line:
            continue
        # 格式: 1. \Device\NPF_{...} (描述)
        m = re.match(r"(\d+)\.\s+(.+?)(?:\s+\((.+?)\))?\s*$", line)
        if m:
            interfaces.append({
                "index": int(m.group(1)),
                "device": m.group(2).strip(),
                "description": m.group(3).strip() if m.group(3) else "",
            })
    return interfaces


def parse_ipconfig() -> list[dict]:
    """解析 ipconfig /all 获取网络接口信息"""
    try:
        result = subprocess.run(
            ["ipconfig", "/all"], capture_output=True, text=True, encoding="gbk", errors="replace"
        )
    except FileNotFoundError:
        return []

    interfaces = []
    current = None
    current_label = ""

    for line in result.stdout.splitlines():
        # 适配器标题行
        adapter_match = re.match(r"^(\S.*?)\s*适配器\s+(.+?)\s*[:：]", line)
        if not adapter_match:
            adapter_match = re.match(r"^(\S.*?)\s+adapter\s+(.+?)\s*[:：]", line, re.IGNORECASE)
        if adapter_match:
            if current:
                interfaces.append(current)
            current = {
                "type": adapter_match.group(1).strip(),
                "name": adapter_match.group(2).strip(),
                "description": "",
                "mac": "",
                "ipv4": "",
                "ipv4_list": [],
                "subnet": "",
                "subnet_list": [],
                "gateway": "",
                "gateway_list": [],
                "dhcp": "",
                "status": "up",
            }
            current_label = ""
            continue

        if current is None:
            continue

        line_stripped = line.strip()
        key, sep, value = line_stripped.partition(":")
        if not sep:
            key, sep, value = line_stripped.partition("：")
        key = key.strip()
        value = value.strip()
        continuation_value = value if sep else line_stripped

        if re.match(r"(媒体状态|Media State)", line_stripped, re.IGNORECASE):
            if "断开" in line_stripped or "disconnected" in line_stripped.lower():
                current["status"] = "down"
            current_label = ""
        elif re.match(r"(描述|Description)", line_stripped, re.IGNORECASE):
            current["description"] = value
            current_label = ""
        elif re.match(r"(物理地址|Physical Address)", line_stripped, re.IGNORECASE):
            current["mac"] = value
            current_label = ""
        elif re.match(r"(IPv4 地址|IPv4 Address)", line_stripped, re.IGNORECASE):
            ipv4 = re.sub(r"\(.*?\)", "", value).strip()
            if looks_like_ipv4(ipv4):
                current["ipv4_list"].append(ipv4)
                current["ipv4"] = current["ipv4_list"][0]
            current_label = "ipv4"
        elif re.match(r"(子网掩码|Subnet Mask)", line_stripped, re.IGNORECASE):
            if looks_like_ipv4(value):
                current["subnet_list"].append(value)
                current["subnet"] = current["subnet_list"][0]
            current_label = "subnet"
        elif re.match(r"(默认网关|Default Gateway)", line_stripped, re.IGNORECASE):
            if looks_like_ip(value):
                current["gateway_list"].append(value)
                current["gateway"] = current["gateway_list"][0]
            current_label = "gateway"
        elif re.match(r"DHCP", line_stripped, re.IGNORECASE) and ("已启用" in line_stripped or "Yes" in line_stripped):
            current["dhcp"] = "enabled"
            current_label = ""
        elif current_label == "gateway" and line.startswith(" ") and looks_like_ip(continuation_value):
            current["gateway_list"].append(continuation_value)
        elif current_label == "ipv4" and line.startswith(" ") and looks_like_ipv4(continuation_value):
            ipv4 = re.sub(r"\(.*?\)", "", continuation_value).strip()
            if ipv4:
                current["ipv4_list"].append(ipv4)
        elif current_label == "subnet" and line.startswith(" ") and looks_like_ipv4(continuation_value):
            current["subnet_list"].append(continuation_value)
        elif current_label in {"ipv4", "subnet", "gateway"} and value == "" and key:
            # 避免误判下一行标题
            current_label = ""

    for iface in interfaces + ([current] if current else []):
        if iface["ipv4_list"] and not iface["ipv4"]:
            iface["ipv4"] = iface["ipv4_list"][0]
        if iface["subnet_list"] and not iface["subnet"]:
            iface["subnet"] = iface["subnet_list"][0]
        if iface["gateway_list"] and not iface["gateway"]:
            iface["gateway"] = iface["gateway_list"][0]

    if current:
        interfaces.append(current)

    return interfaces


def get_net_config(
    cli_interface: str | None = None,
    cli_target: str | None = None,
    cli_capture_filter: str | None = None,
    cli_display_filter: str | None = None,
    cli_duration: int | None = None,
    cli_timeout_ms: int | None = None,
    cli_scan_ports: str | None = None,
    cli_capture_format: str | None = None,
    workspace: str | None = None,
) -> tuple[dict, dict]:
    """
    获取网络配置，按优先级解析参数。
    返回 (config_dict, sources_dict)
    """
    local_cfg = load_local_config()
    proj_cfg = load_project_config(workspace)
    state = load_workspace_state(workspace)

    sources = {}

    # 解析各个参数
    interface, src = resolve_param(
        "interface", cli_interface,
        project_config=proj_cfg, project_keys=["interface"],
        state=state, state_keys=["last_net_interface"],
    )
    sources["interface"] = src or "unknown"

    target, src = resolve_param(
        "target", cli_target,
        project_config=proj_cfg, project_keys=["target"],
        state=state, state_keys=["last_net_target"],
    )
    sources["target"] = src or "unknown"

    capture_filter, src = resolve_param(
        "capture_filter", cli_capture_filter,
        project_config=proj_cfg, project_keys=["capture_filter"],
        state=state, state_keys=["last_capture_filter"],
        default="",
    )
    sources["capture_filter"] = src or "default"

    display_filter, src = resolve_param(
        "display_filter", cli_display_filter,
        project_config=proj_cfg, project_keys=["display_filter"],
        state=state, state_keys=["last_display_filter"],
        default="",
    )
    sources["display_filter"] = src or "default"

    duration, src = resolve_param(
        "duration", cli_duration,
        project_config=proj_cfg, project_keys=["duration"],
        state=state, state_keys=["last_duration"],
        default=30,
    )
    sources["duration"] = src or "default"

    timeout_ms, src = resolve_param(
        "timeout_ms", cli_timeout_ms,
        project_config=proj_cfg, project_keys=["timeout_ms"],
        state=state, state_keys=["last_timeout_ms"],
        default=1000,
    )
    sources["timeout_ms"] = src or "default"

    scan_ports, src = resolve_param(
        "scan_ports", cli_scan_ports,
        project_config=proj_cfg, project_keys=["scan_ports"],
        state=state, state_keys=["last_scan_ports"],
        default="",
    )
    sources["scan_ports"] = src or "default"

    capture_format, src = resolve_param(
        "capture_format", cli_capture_format,
        project_config=proj_cfg, project_keys=["capture_format"],
        state=state, state_keys=["last_capture_format"],
        default="pcapng",
    )
    sources["capture_format"] = src or "default"

    log_dir, src = resolve_param(
        "log_dir", None,
        project_config=proj_cfg, project_keys=["log_dir"],
        default=".embeddedskills/logs/net",
    )
    sources["log_dir"] = src or "default"

    # 获取工具路径（环境级配置）
    default_tshark = "tshark.exe" if sys.platform == "win32" else "tshark"
    default_capinfos = "capinfos.exe" if sys.platform == "win32" else "capinfos"
    tshark_exe = resolve_tool_path(local_cfg.get("tshark_exe"), default_tshark)
    capinfos_exe = resolve_tool_path(local_cfg.get("capinfos_exe"), default_capinfos)

    config = {
        "interface": interface,
        "target": target,
        "capture_filter": capture_filter,
        "display_filter": display_filter,
        "duration": duration,
        "timeout_ms": timeout_ms,
        "scan_ports": scan_ports,
        "capture_format": capture_format,
        "log_dir": log_dir,
        "tshark_exe": tshark_exe,
        "capinfos_exe": capinfos_exe,
    }

    return config, sources


def output_json(data: dict, *, indent: int = 2) -> None:
    """输出 JSON 到 stdout"""
    sys.stdout.reconfigure(encoding="utf-8")
    print(json.dumps(data, ensure_ascii=False, indent=indent), flush=True)