fix: 清 Pylance 类型检查提示(老板标注)

- client.py: 删未使用导入 TYPE_AUTH_OK/TYPE_DNS_RESP/TYPE_HTTP_RESP
- dnsmsg.py: flags 解包后未用 -> 补 QR=1 响应拒绝(符合'非 query 返回 None'语义)
- httpmsg.py: headers default_factory=list 推断 Unknown -> lambda: [] 显式类型
- smoke_channel.py: 删未使用导入 json/os/socket/FrameReader/pack_frame; cfg 显式 dict[str, object]
- analyze_capture.py: proto Counter() 无类型参数 -> Counter[int]
This commit is contained in:
lou
2026-09-09 16:32:31 +08:00
parent 3809f76c29
commit af43b07dc0
5 changed files with 5 additions and 10 deletions
-3
View File
@@ -18,12 +18,9 @@ from typing import Any
from .proto import (
FrameReader,
TYPE_AUTH,
TYPE_AUTH_OK,
TYPE_DNS_REQ,
TYPE_DNS_RESP,
TYPE_ERROR,
TYPE_HTTP_REQ,
TYPE_HTTP_RESP,
TYPE_PING,
TYPE_PONG,
pack_frame,
+2
View File
@@ -19,6 +19,8 @@ def parse_qname(pkt: bytes) -> str | None:
if len(pkt) < DNS_HEADER_LEN + 5:
return None
flags = struct.unpack("!H", pkt[2:4])[0]
if flags & 0x8000: # QR=1 是响应报文, 非 query, 拒绝
return None
qdcount = struct.unpack("!H", pkt[4:6])[0]
if qdcount != 1:
return None
+1 -1
View File
@@ -28,7 +28,7 @@ class HttpRequest:
method: str
path: str
host: str
headers: list[tuple[str, str]] = field(default_factory=list)
headers: list[tuple[str, str]] = field(default_factory=lambda: [])
body: bytes = b""
def header(self, name: str) -> str | None:
+1 -5
View File
@@ -9,9 +9,6 @@ DNS 目标走公网 223.5.5.5, 学校网络不通时只验证帧链路(ERROR/超
from __future__ import annotations
import http.server
import json
import os
import socket
import subprocess
import tempfile
import threading
@@ -19,7 +16,6 @@ from pathlib import Path
from channel.client import ChannelClient
from channel.httpmsg import parse_request
from channel.proto import FrameReader, pack_frame
from channel.server import CloudConfig, CloudServer
WORK = Path(tempfile.mkdtemp(prefix="smoke-channel-"))
@@ -64,7 +60,7 @@ def main() -> int:
target = threading.Thread(target=_run_target, daemon=True)
target.start()
cfg = {
cfg: dict[str, object] = {
"cloud_side": {
"listen_port": CLOUD_PORT,
"token": TOKEN,
+1 -1
View File
@@ -53,7 +53,7 @@ def main() -> int:
bytes_total = sum(len(bytes(p)) for p in pkts)
print(f"== 总览: {total} 包 / {bytes_total/1024:.1f} KB")
proto = Counter()
proto: Counter[int] = Counter()
out_syn: list[tuple[int, int, int, bool, list[str], int, str]] = []
cloud_bytes = 0
cloud_pkts = 0