feat: filter cookies to only store required 10 based on HAR analysis
- Define REQUIRED_COOKIE_NAMES in login_helper.py (10 cookies from HAR) - login_helper.py: import-time filtering + strict all-or-nothing validation - pickup_api.py: save_session() filters to whitelist only (prevents accumulation) - engine.py: unified strict check, all 10 cookies required Based on portal-to-pickup-full-flow.har analysis: - All /pickup-web/a/* APIs use exactly 9 cookies - JSESSIONID additionally needed for portal page fetch (Referer) - enable_HFZrs6pXKulz is portal UI only, excluded
This commit is contained in:
@@ -11,6 +11,7 @@ from typing import Any, Callable
|
|||||||
from log import debug as _ld, trace as _lt, user as _lu
|
from log import debug as _ld, trace as _lt, user as _lu
|
||||||
from pickup_api import PickupAPI, PickupSession, TokenBridge
|
from pickup_api import PickupAPI, PickupSession, TokenBridge
|
||||||
from field_mapping import get_fixed_value_str, translate_value
|
from field_mapping import get_fixed_value_str, translate_value
|
||||||
|
from login_helper import REQUIRED_COOKIE_NAMES
|
||||||
|
|
||||||
# ─── Submit 表单构建器 ─────────────────────────────────────────────────
|
# ─── Submit 表单构建器 ─────────────────────────────────────────────────
|
||||||
|
|
||||||
@@ -368,11 +369,12 @@ class PickupEngine:
|
|||||||
session_path = self._get_session_path()
|
session_path = self._get_session_path()
|
||||||
if self._sess.load_session(session_path):
|
if self._sess.load_session(session_path):
|
||||||
cookie_names = {c.name for c in self._sess._http.cookies}
|
cookie_names = {c.name for c in self._sess._http.cookies}
|
||||||
missing = [n for n in ("JSESSIONID", "jdpt.session.id") if n not in cookie_names]
|
missing = [n for n in REQUIRED_COOKIE_NAMES if n not in cookie_names]
|
||||||
if missing:
|
if missing:
|
||||||
self._ulog(f"缺少关键 cookie: {', '.join(missing)},请重新登录并手动填写")
|
self._ulog(f"缺少必需 cookie ({len(missing)}/{len(REQUIRED_COOKIE_NAMES)}): "
|
||||||
|
f"{', '.join(missing)},请重新登录并导入 cookies")
|
||||||
else:
|
else:
|
||||||
self._dlog("已加载登录会话")
|
self._dlog(f"已加载登录会话 ({len(REQUIRED_COOKIE_NAMES)} 个 cookie 齐全)")
|
||||||
else:
|
else:
|
||||||
self._ulog("未找到登录会话,请先登录")
|
self._ulog("未找到登录会话,请先登录")
|
||||||
|
|
||||||
|
|||||||
+53
-4
@@ -1,4 +1,10 @@
|
|||||||
"""登录助手 — 解析浏览器导出的 cookies JSON,保存为 session.json。"""
|
"""登录助手 — 解析浏览器导出的 cookies JSON,保存为 session.json。
|
||||||
|
|
||||||
|
Cookie 白名单基于 HAR 抓包分析:
|
||||||
|
- pickup-web 所有业务 API 统一使用 9 个 cookie(见 _REQUIRED_COOKIE_NAMES)
|
||||||
|
- JSESSIONID 仅 portal 主页面需要(加载 pickup 页面时作为 Referer)
|
||||||
|
- enable_HFZrs6pXKulz 仅 portal UI 配置,pickup-web 不需要
|
||||||
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
@@ -6,6 +12,31 @@ import json
|
|||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
|
# 基于 portal-to-pickup-full-flow.har 分析得出的必需 cookie
|
||||||
|
# pickup-web (/pickup-web/a/*) 所有业务 API 请求都携带这 10 个 cookie
|
||||||
|
# 缺失任何一个都会导致请求失败
|
||||||
|
REQUIRED_COOKIE_NAMES: set[str] = {
|
||||||
|
# 会话认证(核心 — 缺失则 401/重定向到登录页)
|
||||||
|
"jdpt.session.id",
|
||||||
|
# 组织层级(固定值,session 生命周期内不变,缺失则权限异常)
|
||||||
|
"CORAL_GRAYLEVEL",
|
||||||
|
"PROV_GRAYLEVEL",
|
||||||
|
"CITY_GRAYLEVEL",
|
||||||
|
"COUNTY_GRAYLEVEL",
|
||||||
|
"ORG_GRAYLEVEL",
|
||||||
|
# $_ts 安全框架 token(缺失则 c93l5Fya 校验失败,接口拒绝服务)
|
||||||
|
"HFZrs6pXKulzS", # 会话级 token,固定不变
|
||||||
|
"HFZrs6pXKulzO", # 组织级 token,固定不变
|
||||||
|
"HFZrs6pXKulzP", # 请求级动态 token,服务端 Set-Cookie 持续更新
|
||||||
|
# portal 主页面会话(缺失则无法加载 pickup 页面,fetch_page_resources 失败)
|
||||||
|
"JSESSIONID",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _is_cookie_required(name: str) -> bool:
|
||||||
|
"""判断 cookie 是否为 pickup 流程所必需。"""
|
||||||
|
return name in REQUIRED_COOKIE_NAMES
|
||||||
|
|
||||||
|
|
||||||
def get_session_path() -> str:
|
def get_session_path() -> str:
|
||||||
if getattr(sys, "frozen", False):
|
if getattr(sys, "frozen", False):
|
||||||
@@ -16,17 +47,20 @@ def get_session_path() -> str:
|
|||||||
|
|
||||||
|
|
||||||
def parse_json_and_save(json_text: str) -> int:
|
def parse_json_and_save(json_text: str) -> int:
|
||||||
"""解析浏览器导出的 cookies JSON 数组,提取目标域名的 cookie 并保存。
|
"""解析浏览器导出的 cookies JSON 数组,过滤并保存必需 cookie。
|
||||||
|
|
||||||
|
只保存 _REQUIRED_COOKIE_NAMES 白名单中的 cookie(基于 HAR 分析)。
|
||||||
返回保存的 cookie 数量。
|
返回保存的 cookie 数量。
|
||||||
"""
|
"""
|
||||||
raw = json.loads(json_text)
|
raw = json.loads(json_text)
|
||||||
if not isinstance(raw, list):
|
if not isinstance(raw, list):
|
||||||
raise ValueError("JSON 必须是数组格式")
|
raise ValueError("JSON 必须是数组格式")
|
||||||
|
|
||||||
# 过滤目标域名
|
# 过滤目标域名 + 必需 cookie 名称
|
||||||
target = "10.4.188.1"
|
target = "10.4.188.1"
|
||||||
cookies = []
|
cookies = []
|
||||||
seen = set()
|
seen = set()
|
||||||
|
skipped = []
|
||||||
for c in raw:
|
for c in raw:
|
||||||
domain = c.get("domain", "")
|
domain = c.get("domain", "")
|
||||||
if target not in domain:
|
if target not in domain:
|
||||||
@@ -35,6 +69,9 @@ def parse_json_and_save(json_text: str) -> int:
|
|||||||
path = c.get("path", "/")
|
path = c.get("path", "/")
|
||||||
if not name:
|
if not name:
|
||||||
continue
|
continue
|
||||||
|
if not _is_cookie_required(name):
|
||||||
|
skipped.append(name)
|
||||||
|
continue
|
||||||
key = (name, domain, path)
|
key = (name, domain, path)
|
||||||
if key in seen:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
@@ -46,7 +83,19 @@ def parse_json_and_save(json_text: str) -> int:
|
|||||||
"path": path,
|
"path": path,
|
||||||
})
|
})
|
||||||
|
|
||||||
data = {"cookies": cookies, "saved_from": f"https://{target}"}
|
# 检查必需 cookie 是否齐全(全部 10 个缺一不可)
|
||||||
|
found_names = {c["name"] for c in cookies}
|
||||||
|
missing = REQUIRED_COOKIE_NAMES - found_names
|
||||||
|
if missing:
|
||||||
|
raise ValueError(
|
||||||
|
f"缺少必需 cookie ({len(missing)} 个): {', '.join(sorted(missing))}"
|
||||||
|
)
|
||||||
|
|
||||||
|
data = {
|
||||||
|
"cookies": cookies,
|
||||||
|
"saved_from": f"https://{target}",
|
||||||
|
"filtered_count": len(skipped),
|
||||||
|
}
|
||||||
session_path = get_session_path()
|
session_path = get_session_path()
|
||||||
os.makedirs(os.path.dirname(session_path) or ".", exist_ok=True)
|
os.makedirs(os.path.dirname(session_path) or ".", exist_ok=True)
|
||||||
with open(session_path, "w", encoding="utf-8") as f:
|
with open(session_path, "w", encoding="utf-8") as f:
|
||||||
|
|||||||
+16
-5
@@ -194,12 +194,23 @@ class PickupSession:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
def save_session(self, session_path: str):
|
def save_session(self, session_path: str):
|
||||||
"""保存 cookies 到文件。"""
|
"""保存 cookies 到文件(仅保存必需 cookie,基于 HAR 分析的白名单)。
|
||||||
|
|
||||||
|
requests.Session 会自动合并服务端 Set-Cookie 响应头,
|
||||||
|
可能累积非必需 cookie。保存时只保留白名单中的 cookie。
|
||||||
|
"""
|
||||||
|
from login_helper import REQUIRED_COOKIE_NAMES
|
||||||
|
all_cookies = [
|
||||||
|
{"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
|
||||||
|
for c in self._http.cookies
|
||||||
|
]
|
||||||
|
required = [c for c in all_cookies if c["name"] in REQUIRED_COOKIE_NAMES]
|
||||||
|
skipped = len(all_cookies) - len(required)
|
||||||
data = {
|
data = {
|
||||||
"cookies": [
|
"cookies": required,
|
||||||
{"name": c.name, "value": c.value, "domain": c.domain, "path": c.path}
|
"total_in_session": len(all_cookies),
|
||||||
for c in self._http.cookies
|
"saved": len(required),
|
||||||
]
|
"skipped": skipped,
|
||||||
}
|
}
|
||||||
with open(session_path, "w") as f:
|
with open(session_path, "w") as f:
|
||||||
json.dump(data, f, ensure_ascii=False, indent=2)
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
||||||
|
|||||||
Reference in New Issue
Block a user