fix(agent,teacher): query live IPv4 snapshot on dialog open and harden Win7 WMI/route fallbacks

Author: misakano7545

agent/main.py

@@ -21,6 +21,7 @@
     MSG_ACK,
     MSG_COMMAND_NETWORK_RESTRICT,
     MSG_COMMAND_POWER,
+    MSG_COMMAND_QUERY_IPV4_DETAIL,
     MSG_COMMAND_RENAME_HOST,
     MSG_COMMAND_SET_IPV4,
     MSG_CONFIG_RESPONSE,
@@ -278,6 +279,24 @@ def _handle_command(self, cmd: Dict[str, Any]) -> None:
             if ok:
                 msg = "%s; auto reboot in 1s" % msg
                 self._schedule_reboot()
+        elif cmd_type == MSG_COMMAND_QUERY_IPV4_DETAIL:
+            snap = get_default_ipv4_detail_snapshot(ttl_sec=0.0)
+            ok = True
+            msg = "ipv4_detail"
+            detail_payload = snap
+            try:
+                self._send(
+                    {
+                        "type": MSG_RESULT,
+                        "cmd_id": cmd_id,
+                        "ok": bool(ok),
+                        "message": str(msg),
+                        "ipv4_detail": detail_payload if isinstance(detail_payload, dict) else {},
+                    }
+                )
+            except (ConnectionError, OSError, ValueError):
+                self._stop.set()
+            return
         elif cmd_type == MSG_COMMAND_SET_IPV4:
             name, diag = get_default_ipv4_interface_name()
             if not name:

agent/network_config.py

@@ -485,10 +485,13 @@ def _interface_from_route_print(interface_ip: str) -> Tuple[Optional[str], str]:
     return None, "no WMI row for interface IP %s" % interface_ip
 
 
-def _parse_route_print_default_interface_ip(route_text: str) -> Optional[str]:
-    """解析 route print -4 中 0.0.0.0/0.0.0.0 默认路由行，取「接口」列 IPv4（中英界面均多为数字列）。"""
+def _parse_route_print_default_route_row(route_text: str) -> Optional[Tuple[int, str, str]]:
+    """
+    解析 route print -4 中默认 IPv4 路由行（0.0.0.0 0.0.0.0 …）。
+    返回 (metric, gateway_ipv4, interface_ipv4)；无则 None。
+    """
     lines = route_text.replace("\r\n", "\n").split("\n")
-    best: Optional[Tuple[int, str]] = None  # (metric, interface_ip)
+    best: Optional[Tuple[int, str, str]] = None  # (metric, gw, iface_ip)
     for line in lines:
         line_stripped = line.strip()
         if not line_stripped.startswith("0.0.0.0"):
@@ -499,16 +502,91 @@ def _parse_route_print_default_interface_ip(route_text: str) -> Optional[str]:
             continue
         if parts[1] != "0.0.0.0":
             continue
+        gw_ip = parts[2]
         iface_ip = parts[3]
         if not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", iface_ip):
             continue
+        if not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", gw_ip):
+            continue
         try:
             metric = int(parts[4])
         except ValueError:
             continue
         if best is None or metric < best[0]:
-            best = (metric, iface_ip)
-    return best[1] if best else None
+            best = (metric, gw_ip, iface_ip)
+    return best
+
+
+def _parse_route_print_default_interface_ip(route_text: str) -> Optional[str]:
+    """解析 route print -4 中 0.0.0.0/0.0.0.0 默认路由行，取「接口」列 IPv4（中英界面均多为数字列）。"""
+    row = _parse_route_print_default_route_row(route_text)
+    return row[2] if row else None
+
+
+def _parse_route_print_default_gateway(route_text: str) -> Optional[str]:
+    """默认路由行中的 IPv4 网关。"""
+    row = _parse_route_print_default_route_row(route_text)
+    return row[1] if row else None
+
+
+def _route_print_ipv4_default_gateway() -> str:
+    try:
+        p = subprocess.run(
+            ["route", "print", "-4"],
+            capture_output=True,
+            text=True,
+            creationflags=_subprocess_flags(),
+            timeout=30,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return ""
+    if p.returncode != 0:
+        return ""
+    g = _parse_route_print_default_gateway(p.stdout or "")
+    return g if g else ""
+
+
+def _wmic_nic_configuration_row_for_ipv4(target_ipv4: str) -> Optional[Dict[str, str]]:
+    """
+    Win7 等环境下路由表 InterfaceIndex 与 NICConfiguration.InterfaceIndex 可能不一致；
+    按本机 IPv4 在 WMI 全部已启用配置中反查对应行。
+    """
+    target = (target_ipv4 or "").strip().lower()
+    if not target or not _is_dotted_ipv4(target):
+        return None
+    try:
+        p = subprocess.run(
+            [
+                "wmic",
+                "path",
+                "Win32_NetworkAdapterConfiguration",
+                "where",
+                "IPEnabled=true",
+                "get",
+                "InterfaceIndex,IPAddress,IPSubnet,DefaultIPGateway,DNSServerSearchOrder,DHCPEnabled,IPEnabled",
+                "/format:list",
+            ],
+            capture_output=True,
+            text=True,
+            creationflags=_subprocess_flags(),
+            timeout=45,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    if p.returncode != 0:
+        return None
+    for row in _parse_wmic_list(p.stdout or ""):
+        if row.get("ipenabled", "").strip().lower() not in ("true", "1"):
+            continue
+        raw_ip = row.get("ipaddress", "")
+        tokens = [t.strip().lower() for t in _wmic_multivalue_tokens(raw_ip)]
+        if not tokens:
+            t0 = _clean_wmic_value(raw_ip).strip().lower()
+            if t0:
+                tokens = [t0]
+        if target in tokens:
+            return row
+    return None
 
 
 def _interface_from_route_print_only() -> Tuple[Optional[str], str]:
@@ -744,32 +822,36 @@ def _wmic_interface_index_for_netconnection_id(target: str) -> Optional[int]:
 
 
 def _wmic_nic_configuration_by_interface_index(if_idx: int) -> Optional[Dict[str, str]]:
-    try:
-        p = subprocess.run(
-            [
-                "wmic",
-                "path",
-                "Win32_NetworkAdapterConfiguration",
-                "where",
-                "InterfaceIndex=%d" % if_idx,
-                "get",
-                "IPAddress,IPSubnet,DefaultIPGateway,DNSServerSearchOrder,DHCPEnabled,IPEnabled",
-                "/format:list",
-            ],
-            capture_output=True,
-            text=True,
-            creationflags=_subprocess_flags(),
-            timeout=30,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        return None
-    if p.returncode != 0:
-        return None
-    rows = _parse_wmic_list(p.stdout or "")
-    for row in rows:
-        if row.get("ipenabled", "").strip().lower() in ("true", "1"):
-            return row
-    return rows[0] if rows else None
+    """Win7 上部分环境需用 Index= 而非 InterfaceIndex=，两种都试。"""
+    for where in ("InterfaceIndex=%d" % if_idx, "Index=%d" % if_idx):
+        try:
+            p = subprocess.run(
+                [
+                    "wmic",
+                    "path",
+                    "Win32_NetworkAdapterConfiguration",
+                    "where",
+                    where,
+                    "get",
+                    "IPAddress,IPSubnet,DefaultIPGateway,DNSServerSearchOrder,DHCPEnabled,IPEnabled",
+                    "/format:list",
+                ],
+                capture_output=True,
+                text=True,
+                creationflags=_subprocess_flags(),
+                timeout=30,
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            continue
+        if p.returncode != 0:
+            continue
+        rows = _parse_wmic_list(p.stdout or "")
+        for row in rows:
+            if row.get("ipenabled", "").strip().lower() in ("true", "1"):
+                return row
+        if rows:
+            return rows[0]
+    return None
 
 
 def _ipv4_subnet_pairs_from_wmi_row(row: Dict[str, str]) -> List[Tuple[str, str]]:
@@ -1017,6 +1099,22 @@ def get_default_ipv4_detail_snapshot(ttl_sec: float = 20.0) -> Dict[str, Any]:
     if not str(out.get("ip") or "").strip() and pref:
         out["ip"] = pref
 
+    hip = str(out.get("ip") or pref).strip()
+    if hip:
+        need_more = (
+            not str(out.get("mask") or "").strip()
+            or not str(out.get("gateway") or "").strip()
+            or not str(out.get("dns_primary") or "").strip()
+        )
+        if need_more:
+            row_fix = _wmic_nic_configuration_row_for_ipv4(hip)
+            if row_fix:
+                _merge_ipv4_detail_fields(out, _detail_row_to_dict(row_fix, hip))
+    if not str(out.get("gateway") or "").strip():
+        gw_rp = _route_print_ipv4_default_gateway()
+        if gw_rp:
+            out["gateway"] = gw_rp
+
     _detail_mono = now
     _detail_payload = dict(out)
     return dict(out)

common/protocol.py

@@ -15,6 +15,7 @@
 MSG_ACK = "ack"
 MSG_COMMAND_RENAME_HOST = "command_rename_host"
 MSG_COMMAND_SET_IPV4 = "command_set_ipv4"
+MSG_COMMAND_QUERY_IPV4_DETAIL = "command_query_ipv4_detail"
 MSG_COMMAND_POWER = "command_power"
 MSG_COMMAND_NETWORK_RESTRICT = "command_network_restrict"
 MSG_RESULT = "result"

teacher/main.py

@@ -17,6 +17,7 @@
 from common.protocol import (
     MSG_COMMAND_NETWORK_RESTRICT,
     MSG_COMMAND_POWER,
+    MSG_COMMAND_QUERY_IPV4_DETAIL,
     MSG_COMMAND_RENAME_HOST,
     MSG_COMMAND_SET_IPV4,
 )
@@ -68,6 +69,15 @@ def enqueue_set_ipv4(session_id: str, cmd_id: str, payload: Dict[str, Any]) -> b
         body.update(payload)
         return server.enqueue_command(session_id, body)
 
+    def enqueue_query_ipv4_detail(session_id: str, cmd_id: str) -> bool:
+        return server.enqueue_command(
+            session_id,
+            {
+                "type": MSG_COMMAND_QUERY_IPV4_DETAIL,
+                "cmd_id": cmd_id,
+            },
+        )
+
     def enqueue_power(session_id: str, cmd_id: str, action: str) -> bool:
         return server.enqueue_command(
             session_id,
@@ -83,7 +93,14 @@ def enqueue_network_restrict(session_id: str, cmd_id: str, payload: Dict[str, An
         body.update(payload)
         return server.enqueue_command(session_id, body)
 
-    app = TeacherApp(server, enqueue_rename, enqueue_set_ipv4, enqueue_power, enqueue_network_restrict)
+    app = TeacherApp(
+        server,
+        enqueue_rename,
+        enqueue_set_ipv4,
+        enqueue_query_ipv4_detail,
+        enqueue_power,
+        enqueue_network_restrict,
+    )
     app.log_line(
         "教师端已启动 监听 %s:%s token=%s"
         % (listen_host, listen_port, "已启用" if token else "未启用")
@@ -114,6 +131,7 @@ def poll_queue() -> None:
                         % (data.get("hostname") or "-", data.get("session_id"))
                     )
                 elif kind == "command_result":
+                    app.handle_ipv4_detail_query_result(data)
                     app.log_line(
                         "结果 pc_name=%s cmd_id=%s ok=%s %s"
                         % (

teacher/server.py

@@ -293,16 +293,17 @@ def _client_loop(self, conn: socket.socket, peer_ip: str) -> None:
                         s_res = self._sessions.get(session_id)
                         if s_res:
                             res_host = str(s_res.hostname or "")
-                    self._emit(
-                        "command_result",
-                        {
-                            "session_id": session_id,
-                            "hostname": res_host,
-                            "cmd_id": msg.get("cmd_id"),
-                            "ok": bool(msg.get("ok")),
-                            "message": str(msg.get("message") or ""),
-                        },
-                    )
+                    payload_cr: Dict[str, Any] = {
+                        "session_id": session_id,
+                        "hostname": res_host,
+                        "cmd_id": msg.get("cmd_id"),
+                        "ok": bool(msg.get("ok")),
+                        "message": str(msg.get("message") or ""),
+                    }
+                    det_r = msg.get("ipv4_detail")
+                    if isinstance(det_r, dict):
+                        payload_cr["ipv4_detail"] = det_r
+                    self._emit("command_result", payload_cr)
                 elif mtype == MSG_REQUEST_CONFIG:
                     config_content = ""
                     if self._yaml_file_path and os.path.exists(self._yaml_file_path):