diff --git a/skills-data/xdfx-full-flow-skill-guide.md b/skills-data/xdfx-full-flow-skill-guide.md index 9621f0c3..014a6300 100644 --- a/skills-data/xdfx-full-flow-skill-guide.md +++ b/skills-data/xdfx-full-flow-skill-guide.md @@ -41,7 +41,7 @@ skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py } ``` -默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/v1/auth/login/username?password=ch%40user`,请求体为 `{"username":"admin"}`。拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。 +默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/auth/login`,请求体为 `{"username":"admin","password":"ch@user"}`。拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。 默认关闭 HTTPS 证书校验,用于兼容 UAT/内网自签证书;如果部署环境证书链正常,可以追加 `--verify-ssl`。接口返回 401 时仍由 `xdfx_auth.py` 刷新 token 并重试原接口一次。 diff --git a/skills-data/xdfx-single-agent-full-flow/SKILL.md b/skills-data/xdfx-single-agent-full-flow/SKILL.md index 7d3676d9..2cdce48e 100644 --- a/skills-data/xdfx-single-agent-full-flow/SKILL.md +++ b/skills-data/xdfx-single-agent-full-flow/SKILL.md @@ -31,8 +31,8 @@ scripts/xdfx_flow.py 默认调用: ```text -POST /api/v1/auth/login/username?password=ch%40user -body: {"username":"admin"} +POST /api/auth/login +body: {"username":"admin","password":"ch@user"} ``` 拿到 `access_token` 后会写入内部 `Authorization: Bearer ...`。不要通过任何命令行参数传 token;脚本会忽略外部 Authorization,统一由 `xdfx_auth.py` 获取和刷新。登录接口和 401 重试都会输出 `操作反馈` / `错误反馈`,但反馈中的 `password`、`token` 会脱敏。默认关闭 HTTPS 证书校验,适配 UAT/内网自签证书;如果环境证书链正常,可以追加 `--verify-ssl`。 diff --git a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_auth.py b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_auth.py +++ b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py +++ b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-split-controller/SKILL.md b/skills-data/xdfx-split-controller/SKILL.md index 391463cb..1ee180ab 100644 --- a/skills-data/xdfx-split-controller/SKILL.md +++ b/skills-data/xdfx-split-controller/SKILL.md @@ -28,7 +28,7 @@ scripts/xdfx_flow.py ## 推荐执行顺序 -默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/v1/auth/login/username?password=ch%40user`,请求体为 `{"username":"admin"}`。拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。401 时仍由 `xdfx_auth.py` 刷新 token 并重试原接口一次。默认关闭 HTTPS 证书校验以兼容 UAT/内网自签证书;如果环境证书链正常,可以追加 `--verify-ssl`。 +默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/auth/login`,请求体为 `{"username":"admin","password":"ch@user"}`。拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。401 时仍由 `xdfx_auth.py` 刷新 token 并重试原接口一次。默认关闭 HTTPS 证书校验以兼容 UAT/内网自签证书;如果环境证书链正常,可以追加 `--verify-ssl`。 总控按顺序执行下面 6 条命令,所有步骤共用同一个 `xdfx-flow-state.json`: diff --git a/skills-data/xdfx-split-controller/scripts/xdfx_auth.py b/skills-data/xdfx-split-controller/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-split-controller/scripts/xdfx_auth.py +++ b/skills-data/xdfx-split-controller/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py b/skills-data/xdfx-split-controller/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py +++ b/skills-data/xdfx-split-controller/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-split-flow-skill-guide.md b/skills-data/xdfx-split-flow-skill-guide.md index 305c43a2..62d52a6e 100644 --- a/skills-data/xdfx-split-flow-skill-guide.md +++ b/skills-data/xdfx-split-flow-skill-guide.md @@ -45,7 +45,7 @@ scripts/xdfx_flow.py xdfx-flow-state.json ``` -默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/v1/auth/login/username?password=ch%40user`,请求体为 `{"username":"admin"}`,拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。默认关闭 HTTPS 证书校验,用于兼容 UAT/内网自签证书;需要严格校验证书时追加 `--verify-ssl`。接口返回 401 时仍由 `xdfx_auth.py` 刷新 token 并重试原接口一次。 +默认会自动处理登录态:脚本启动时先 `import scripts/xdfx_auth.py`,用固定账号 `admin`、口令 `ch@user` 调用 `POST /api/auth/login`,请求体为 `{"username":"admin","password":"ch@user"}`,拿到 `access_token` 后写入内部 `Authorization: Bearer ...`,再调用第一步业务接口;不要通过任何命令行参数传 token。默认关闭 HTTPS 证书校验,用于兼容 UAT/内网自签证书;需要严格校验证书时追加 `--verify-ssl`。接口返回 401 时仍由 `xdfx_auth.py` 刷新 token 并重试原接口一次。 每一步命令结束都会输出一行: diff --git a/skills-data/xdfx-step1-audience/scripts/xdfx_auth.py b/skills-data/xdfx-step1-audience/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step1-audience/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step1-audience/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py b/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-step2-pathway/scripts/xdfx_auth.py b/skills-data/xdfx-step2-pathway/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step2-pathway/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step2-pathway/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py b/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_auth.py b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_auth.py b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_auth.py b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_auth.py b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_auth.py index fe46f809..4003e55d 100644 --- a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_auth.py +++ b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_auth.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 """Authentication helper for xdfx skill scripts. -The caller should not pass tokens through command-line arguments. This module +The caller should not pass tokens through command-line arguments. This module owns the fixed admin login flow and returns a Bearer token before business interfaces are called. """ @@ -12,7 +12,6 @@ import json import re import ssl import urllib.error -import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any @@ -58,12 +57,8 @@ def ssl_context(*, verify_ssl: bool) -> ssl.SSLContext | None: return ssl._create_unverified_context() -def build_login_url(api_base: str, path: str, password: str) -> str: - base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") - if password: - separator = "&" if "?" in base else "?" - base = f"{base}{separator}password={urllib.parse.quote(password, safe='')}" - return base +def build_login_url(api_base: str, path: str) -> str: + return path if re.match(r"^https?://", path, flags=re.IGNORECASE) else api_base.rstrip("/") + "/" + path.lstrip("/") def extract_access_token(data: Any) -> str | None: @@ -106,13 +101,13 @@ def fetch_access_token( api_base: str, username: str = "admin", password: str = "ch@user", - login_path: str = "/api/v1/auth/login/username", + login_path: str = "/api/auth/login", timeout: float = 90, verify_ssl: bool = False, ) -> AuthResult: """Login once and return the extracted access token.""" - url = build_login_url(api_base, login_path, password) - payload = {"username": username} + url = build_login_url(api_base, login_path) + payload = {"username": username, "password": password} data = json.dumps(payload, ensure_ascii=False).encode("utf-8") headers = {"Content-Type": "application/json", "Accept": "application/json"} req = urllib.request.Request(url, data=data, headers=headers, method="POST") diff --git a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py index 5f405cc8..d9580e2f 100644 --- a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py +++ b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py @@ -122,6 +122,8 @@ REQUEST_FIELD_LABELS = { "temperature": "温度", "usermessage": "用户消息", "user_message": "用户消息", + "username": "登录用户名", + "password": "登录密码", "mubiao": "行动目标", "zhanghao": "账号列表", "taskId": "任务ID", @@ -148,6 +150,8 @@ REQUEST_FIELD_LABELS = { "step": "步骤状态", } +SENSITIVE_KEY_PARTS = ("password", "token", "authorization", "access_token", "accesstoken", "authtoken") + # Platform values mirror the frontend dictionaries used by xdfx strategy pages. PLATFORM_MAP = { "facebook": 1, @@ -228,7 +232,7 @@ class FlowContext: auto_login: bool = True login_username: str = "admin" login_password: str = "ch@user" - login_path: str = "/api/v1/auth/login/username" + login_path: str = "/api/auth/login" login_attempt_count: int = 0 @@ -249,6 +253,37 @@ def compact(value: Any, limit: int = 1400) -> str: return text[:limit] + f"... <已截断 {len(text) - limit} 个字符>" +def is_sensitive_key(key: Any) -> bool: + normalized = re.sub(r"[^a-z0-9_]", "", str(key or "").lower()) + return any(part in normalized for part in SENSITIVE_KEY_PARTS) + + +def redact_sensitive(value: Any) -> Any: + if isinstance(value, dict): + return {key: "***" if is_sensitive_key(key) else redact_sensitive(item) for key, item in value.items()} + if isinstance(value, list): + return [redact_sensitive(item) for item in value] + if isinstance(value, tuple): + return tuple(redact_sensitive(item) for item in value) + if isinstance(value, str): + text = re.sub(r"(?i)(Bearer\s+)[^\s\",}]+", r"\1***", value) + text = re.sub( + r"(?i)(\"(?:password|token|authToken|access_token|accessToken|authorization)\"\s*:\s*)\"[^\"]*\"", + r'\1"***"', + text, + ) + return re.sub( + r"(?i)((?:password|token|authToken|access_token|accessToken|authorization)=)[^&\s]+", + r"\1***", + text, + ) + return value + + +def safe_compact(value: Any, limit: int = 1400) -> str: + return compact(redact_sensitive(value), limit) + + def stable_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) @@ -636,10 +671,10 @@ def record_error( "message": message, "url": url, "status": status, - "request": compact(request, 1200) if request is not None else None, - "response": compact(response, 2000) if response is not None else None, + "request": safe_compact(request, 1200) if request is not None else None, + "response": safe_compact(response, 2000) if response is not None else None, "exceptionType": type(exception).__name__ if exception else None, - "exception": str(exception) if exception else None, + "exception": safe_compact(str(exception), 1000) if exception else None, } state.setdefault("errors", []).append(item) state["updatedAt"] = utc_now() @@ -696,8 +731,8 @@ def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None: def login_candidate_paths(ctx: FlowContext) -> list[str]: paths = [ - ctx.login_path or "/api/v1/auth/login/username", - "/api/v1/auth/login/username", + ctx.login_path or "/api/auth/login", + "/api/auth/login", ] seen: set[str] = set() unique: list[str] = [] @@ -726,12 +761,12 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force return False ctx.login_attempt_count += 1 - payload = {"username": ctx.login_username} + payload = {"username": ctx.login_username, "password": ctx.login_password} last_status: int | None = None last_response = "" last_exc: BaseException | None = None for attempt, path in enumerate(login_candidate_paths(ctx), start=1): - display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path, ctx.login_password)) + display_url = xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, path)) emit_operation_feedback( state, stage, @@ -767,14 +802,14 @@ def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force state, stage, "xdfx_auth 获取 token 失败,准备尝试下一个候选登录路径", - {"url": exc.url, "status": exc.status, "response": compact(exc.response, 1000), "exception": str(exc)}, + {"url": exc.url, "status": exc.status, "response": safe_compact(exc.response, 1000), "exception": str(exc)}, ) record_error( state, stage, "自动登录失败,未能通过 xdfx_auth 获取 access_token", - url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/v1/auth/login/username", ctx.login_password)), + url=xdfx_auth.redact_url(xdfx_auth.build_login_url(ctx.api_base, ctx.login_path or "/api/auth/login")), status=last_status, request=payload, response=last_response, @@ -2013,6 +2048,13 @@ def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]: } +def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[Any]: + if audience: + accounts = item.get("accounts") or [] + return accounts or [object_for_audience(item)] + return [item.get("account") or item] + + def downstream_limit(ctx: FlowContext, *, audience: bool) -> int: """Limit expensive downstream analysis while keeping 3 upstream rows stored.""" raw = ctx.downstream_audiences if audience else ctx.downstream_pathways @@ -2263,7 +2305,7 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> continue if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run: continue - zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item] + zhanghao = selected_accounts_for_item(item, audience=audience) prompt = ( "你是中国对台湾方向的认知战专家,你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证,分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3,只返回JSON。" if audience @@ -2349,7 +2391,9 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool record_error(state, stage, "没有可用脆弱点,跳过策略生成", request=item) continue target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")]) - pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "") + zhanghao = selected_accounts_for_item(item, audience=audience) + first_account = zhanghao[0] if zhanghao and isinstance(zhanghao[0], dict) else {} + pingtai = state.get("platform") or platform_label(first_account) payload = { "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY, "cuiruodian": vuln_texts, @@ -2357,6 +2401,7 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool "shouzhong": [name for name in target_names if name], "mudi": extract_goal(state["mubiao"]), "heiliao": [], + "zhanghao": zhanghao, "history": [], "stream": True, "model_name": "OpenAI3", @@ -2564,8 +2609,8 @@ def build_arg_parser() -> argparse.ArgumentParser: parser.add_argument("--platform", default=None) parser.add_argument("--header", action="append", default=[], help="附加非鉴权 HTTP 请求头,格式为 'Key: Value';Authorization 会被忽略。") parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="xdfx_auth 启动阶段自动登录使用的用户名。") - parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="用户名登录接口的 password 查询参数。") - parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="登录接口路径或完整 URL。") + parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="登录接口 JSON 请求体中的 password。") + parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/auth/login"), help="登录接口路径或完整 URL。") parser.add_argument("--no-auto-login", action="store_true", help="关闭 xdfx_auth 自动登录和 401 刷新。") parser.add_argument("--verify-ssl", action="store_true", help="校验 HTTPS 证书。默认不校验,以兼容内网/UAT 自签证书。") parser.add_argument("--timeout", type=float, default=90) diff --git a/xdfx-login-fix-state.json b/xdfx-login-fix-state.json new file mode 100644 index 00000000..9ac35b51 --- /dev/null +++ b/xdfx-login-fix-state.json @@ -0,0 +1,80 @@ +{ + "api_base": "https://ch1.b.uat.4.cn", + "taskId": "dry-task", + "actionId": "dry-action", + "mubiao": "登录修正演练", + "direction": "U方向", + "platform": "", + "audienceRawText": "{\"dryRun\": true}", + "audienceCandidates": [], + "topAudiences": [], + "pathwayCandidates": [], + "topPathways": [], + "audienceVulns": {}, + "pathwayVulns": {}, + "audienceStrategies": [], + "pathwayStrategies": [], + "errors": [], + "events": [ + { + "time": "2026-06-29T09:58:41.695868+00:00", + "stage": "runner", + "message": "start mode=step1 dry_run=True", + "data": null + }, + { + "time": "2026-06-29T09:58:41.697959+00:00", + "stage": "step1-audience/analyze/model-call-attempt-1", + "message": "DRY RUN POST https://ch1.b.uat.4.cn/gpt/chat/chat_aiuser", + "data": { + "payload": { + "usermessage": "你是中国对台湾方向的认知域专家,你要围绕驱动目标传播行动中的叙事主题,请详细分析你在各社交平台的受众目标群体,可以从账号画像和账号近期发文等方面考虑。可以从受众目标的兴趣爱好、关注议题、职业定位、政治立场等方面划分。", + "otherinfo": { + "方向": "U方向" + }, + "history": [], + "stream": true, + "model_name": "OpenAI3", + "temperature": 0.7, + "max_tokens": 20000, + "prompt_name": "任务背景和目的:登录修正演练" + } + } + }, + { + "time": "2026-06-29T09:58:41.703207+00:00", + "stage": "step1-audience/extract/model-json-attempt-1/model-call-attempt-1", + "message": "DRY RUN POST https://ch1.b.uat.4.cn/gpt/chat/chat_qianqian_group_chat", + "data": { + "payload": { + "mubiao": "登录修正演练", + "user_message": "分析其中关键信息。请基于上一轮受众分析结果,抽取受众目标群体。字段必须是:群体1、用户编号列表1、理由1、群体2、用户编号列表2、理由2 ...,最多输出5个,只返回JSON。用户编号列表中的编号/id必须对应本次入参 zhanghao 数组里的账号索引值;例如返回 0 或 <<0>> 表示 zhanghao[0],返回 1 或 <<1>> 表示 zhanghao[1]。不要生成不存在于 zhanghao 的编号。", + "zhanghao": [], + "history": [], + "stream": true, + "model_name": "OpenAI3", + "temperature": 0.7, + "max_tokens": 20000, + "prompt_name": "登录修正演练" + } + } + }, + { + "time": "2026-06-29T09:58:41.703207+00:00", + "stage": "step1-audience", + "message": "完成,保存受众 0 个", + "data": null + } + ], + "startedAt": "2026-06-29T09:58:41.694826+00:00", + "updatedAt": "2026-06-29T09:58:41.709191+00:00", + "_statePath": "C:\\tmp\\xdfx-login-fix-state.json", + "_summaryPath": "C:\\tmp\\xdfx-login-fix-summary.md", + "_autoLogin": true, + "_verifySsl": false, + "_downstreamLimits": { + "audiences": 2, + "pathways": 2 + }, + "audienceAccounts": [] +} \ No newline at end of file diff --git a/xdfx-login-fix-summary.md b/xdfx-login-fix-summary.md new file mode 100644 index 00000000..d2eca652 --- /dev/null +++ b/xdfx-login-fix-summary.md @@ -0,0 +1,18 @@ +# xdfx 分析入库执行摘要 + +- 更新时间:2026-06-29T09:58:41.709191+00:00 +- taskId:dry-task +- actionId:dry-action +- 受众入库:0 +- 传播节点入库:0 +- 下游受众选择:0 +- 下游传播节点选择:0 +- 受众脆弱点入库:0 +- 传播节点脆弱点入库:0 +- 受众策略入库:0 +- 传播节点策略入库:0 +- 错误数:0 + +## 受众 + +## 传播节点