lou
af43b07dc0
- 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]
107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
"""HTTP 请求解析与重组(共享: 本地代理解析, 云端套壳重组)。
|
|
|
|
保持最小够用集: 请求行 + headers(保留顺序/重复) + Content-Length body。
|
|
chunked 请求体与 obs-fold 不做(第一版), 遇到按无 body 处理。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
_HEAD_END = b"\r\n\r\n"
|
|
|
|
# hop-by-hop 头, 套壳重组时剔除(不跨连接转发)
|
|
_HOP_BY_HOP = {
|
|
"connection",
|
|
"keep-alive",
|
|
"proxy-connection",
|
|
"proxy-authenticate",
|
|
"proxy-authorization",
|
|
"te",
|
|
"trailer",
|
|
"transfer-encoding",
|
|
"upgrade",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class HttpRequest:
|
|
method: str
|
|
path: str
|
|
host: str
|
|
headers: list[tuple[str, str]] = field(default_factory=lambda: [])
|
|
body: bytes = b""
|
|
|
|
def header(self, name: str) -> str | None:
|
|
low = name.lower()
|
|
for key, value in self.headers:
|
|
if key.lower() == low:
|
|
return value
|
|
return None
|
|
|
|
def without_hop_by_hop(self) -> list[tuple[str, str]]:
|
|
out: list[tuple[str, str]] = []
|
|
for key, value in self.headers:
|
|
if key.lower() not in _HOP_BY_HOP and key.lower() != "host":
|
|
out.append((key, value))
|
|
return out
|
|
|
|
|
|
def parse_request_head(data: bytes) -> tuple[HttpRequest, int] | None:
|
|
"""解析请求头部区, 返回 (请求, 头部区字节数); 数据不足/畸形返回 None。"""
|
|
idx = data.find(_HEAD_END)
|
|
if idx < 0:
|
|
return None
|
|
head = data[:idx]
|
|
lines = head.split(b"\r\n")
|
|
if not lines or len(lines[0].split(b" ")) != 3:
|
|
return None
|
|
method_b, path_b, _ver_b = lines[0].split(b" ", 2)
|
|
headers: list[tuple[str, str]] = []
|
|
for line in lines[1:]:
|
|
if not line:
|
|
continue
|
|
if b":" not in line:
|
|
continue
|
|
name_b, value_b = line.split(b":", 1)
|
|
name = name_b.decode("latin-1").strip()
|
|
value = value_b.decode("latin-1").strip()
|
|
headers.append((name, value))
|
|
try:
|
|
method = method_b.decode("latin-1")
|
|
path = path_b.decode("latin-1")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
host = ""
|
|
body_len = 0
|
|
for key, value in headers:
|
|
low = key.lower()
|
|
if low == "host" and not host:
|
|
host = value
|
|
elif low == "content-length":
|
|
try:
|
|
body_len = int(value)
|
|
except ValueError:
|
|
body_len = 0
|
|
if body_len < 0 or body_len > 64 * 1024 * 1024:
|
|
return None
|
|
return HttpRequest(method=method, path=path, host=host, headers=headers), idx + 4
|
|
|
|
|
|
def parse_request(data: bytes) -> HttpRequest | None:
|
|
"""解析完整请求(头部 + body), body 不足返回 None。"""
|
|
parsed = parse_request_head(data)
|
|
if parsed is None:
|
|
return None
|
|
req, head_len = parsed
|
|
content_length = 0
|
|
for key, value in req.headers:
|
|
if key.lower() == "content-length":
|
|
try:
|
|
content_length = int(value)
|
|
except ValueError:
|
|
content_length = 0
|
|
if len(data) < head_len + content_length:
|
|
return None
|
|
req.body = data[head_len : head_len + content_length]
|
|
return req
|