from __future__ import annotations import asyncio import json import os import re import shutil import uuid from copy import deepcopy from pathlib import Path from urllib.parse import unquote, urlsplit from .learning_contracts import NAVIGATION_PROFILE, execution_metadata_matches FAILURE_PHASES = {"validate", "inspect_current", "capture_before", "resolve_target", "execute", "settle", "request_guard", "capture_after"} PRIVATE_PATH = re.compile(r"password|passwd|credential|token|secret|session|cookie|authorization|captcha|verification|phone|mobile|验证码|密码|手机号", re.I) LOGIN_EVIDENCE = {"visible_current_account_navigation", "visible_login_dialog", "visible_login_entry", "visible_verification_widget", "no_conclusive_visible_evidence", "unsupported_platform", "inspection_unavailable"} TIMING_PHASES = {"validate", "inspect_current", "capture_before", "resolve_target", "execute", "settle", "request_guard", "capture_after", "failure_capture", "batch_capture"} REQUEST_POLICY_RULES = { "xhs-search-experimental-v1": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa"}, "xhs-search-experimental-v2": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data"}, "xhs-search-experimental-v3": {"collect", "search_history", "search_notes", "search_onebox", "search_dqa", "apm_data", "apm_data_xhr"}, "xhs-note-open-experimental-v1": {"collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data"}, "xhs-note-open-experimental-v2": {"collect", "note_share_code", "note_widgets", "note_feed", "note_seo", "note_metrics", "apm_data", "note_history_report"}, "xhs-readonly-ui-experimental-v1": {"collect", "apm_data"}, "xhs-search-scroll-experimental-v1": {"collect", "apm_data", "search_notes"}, "xhs-page-native-v1": {"native_page_traffic"}, "xhs-page-navigation-v1": {"native_page_traffic"}, } REQUEST_RESOURCE_TYPES = {"document", "fetch", "xhr", "other", "script", "stylesheet", "image", "font", "media", "manifest", "websocket", "eventsource", "texttrack"} ACTIVE_REQUEST_POLICIES = {"xhs-search-experimental-v3", "xhs-note-open-experimental-v2", "xhs-readonly-ui-experimental-v1", "xhs-search-scroll-experimental-v1", "xhs-page-native-v1", "xhs-page-navigation-v1"} REQUEST_POLICY_REASONS = {"missing_policy", "inactive_policy", "invalid_time", "policy_expired", "automation_inactive", "session_out_of_scope", "page_out_of_scope", "not_main_frame", "method_not_allowed", "page_changed", "policy_not_started", "invalid_url", "request_out_of_scope", "resource_type_not_allowed", "navigation_out_of_scope"} def safe_login_detection(value) -> dict | None: if not isinstance(value, dict): return None state = value.get("state") if not isinstance(state, str) or state not in {"logged_in", "logged_out", "verification_required", "unknown"} or value.get("basis") != "visible_page": return None timestamp = value.get("observed_at") if not isinstance(timestamp, str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp): return None evidence = value.get("evidence") return {"state": state, "basis": "visible_page", "observed_at": timestamp, "evidence": [item for item in evidence[:7] if isinstance(item, str) and item in LOGIN_EVIDENCE] if isinstance(evidence, list) else []} def safe_worker_code(value) -> str: return value if isinstance(value, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", value) else "worker_error" def safe_request_summary(value, *, blocked: bool = False) -> dict | None: """A second boundary: never accept arbitrary worker error/request payloads.""" if not isinstance(value, dict) or not isinstance(value.get("method"), str) or value["method"] not in {"GET", "HEAD", "OPTIONS", "POST", "PUT", "PATCH", "DELETE", "CONNECT", "TRACE"}: return None if not isinstance(value.get("resource_type"), str) or value["resource_type"] not in REQUEST_RESOURCE_TYPES: return None location = safe_location(value.get("origin"), value.get("pathname")) if not location: return None summary = {"method": value["method"], **location, "resource_type": value["resource_type"]} if blocked: reason, version = value.get("policy_reason"), value.get("active_policy_version") if isinstance(reason, str) and reason in REQUEST_POLICY_REASONS: summary["policy_reason"] = reason if isinstance(version, str) and version in ACTIVE_REQUEST_POLICIES: summary["active_policy_version"] = version expected = value.get("expected_resource_types") if reason == "resource_type_not_allowed" and isinstance(expected, list): types = list(dict.fromkeys(item for item in expected if isinstance(item, str) and item in REQUEST_RESOURCE_TYPES))[:4] if types: summary["expected_resource_types"] = types return summary def safe_location(origin_value, raw_path) -> dict | None: try: parsed = urlsplit(origin_value) if isinstance(origin_value, str) else None if not parsed: return None if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}: return None parsed.port origin = f"{parsed.scheme}://{parsed.netloc}" except (ValueError, TypeError): return None if not isinstance(raw_path, str) or not raw_path.startswith("/"): return None segments = raw_path.split("?", 1)[0].split("#", 1)[0].split("/") path = [] for segment in segments: decoded = unquote(segment) if PRIVATE_PATH.search(decoded): path.append(":redacted") elif len(decoded) > 64 or "@" in decoded or re.search(r"1[3-9]\d{9}", decoded) or re.fullmatch(r"[a-f0-9-]{16,}|[A-Za-z0-9_=-]{32,}", decoded) or re.search(r"[\x00-\x1f]", decoded): path.append(":id") else: path.append(segment) return {"origin": origin, "pathname": "/".join(path)[:300]} def safe_popup_observations(value) -> list[dict]: records = [] for item in value[-10:] if isinstance(value, list) else []: if not isinstance(item, dict): continue allowed = {"phase": {"manual", "automation", "confirmed"}, "decision": {"adopted", "blocked"}, "reason": {"same_origin_manual", "same_origin_click", "unconfirmed_origin", "outside_scope", "unrequested_popup"}} if any(not isinstance(item.get(key), str) or item[key] not in options for key, options in allowed.items()): continue sequence, timestamp = item.get("sequence"), item.get("recorded_at") if type(sequence) is not int or not 0 <= sequence <= 2 ** 53 or not isinstance(timestamp, str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp): continue record = {"sequence": sequence, "recorded_at": timestamp, **{key: item[key] for key in allowed}} location = safe_location(item.get("origin"), item.get("pathname")) if location: record.update(location) records.append(record) return records def safe_request_observations(value) -> list[dict]: records = [] for item in value[-30:] if isinstance(value, list) else []: summary = safe_request_summary(item, blocked=isinstance(item, dict) and item.get("blocked") is True) if not summary: continue if type(item.get("sequence")) is int and 0 <= item["sequence"] <= 2 ** 53: summary["sequence"] = item["sequence"] timestamp = item.get("recorded_at") if isinstance(timestamp, str) and re.fullmatch(r"\d{4}-\d\d-\d\dT[0-9:.+Z-]{8,25}", timestamp): summary["recorded_at"] = timestamp for key in ("automating", "blocked"): if type(item.get(key)) is bool: summary[key] = item[key] rule, version = item.get("policy_rule"), item.get("policy_version") allowed_rules = REQUEST_POLICY_RULES.get(version) if isinstance(version, str) else None if allowed_rules and isinstance(rule, str) and rule in allowed_rules and item.get("automating") is True and item.get("blocked") is False: summary.update(policy_version=version, policy_rule=rule) records.append(summary) return records def safe_navigation_comparison(value) -> dict | None: """Only fixed relationship facts; never retain either navigation address.""" if not isinstance(value, dict): return None kinds = {"note", "profile", "notification", "explore", "search", "invalid"} booleans = ("same_origin", "same_pathname", "same_object_id", "same_query", "same_query_pairs", "same_hash", "other_query_equal") if any(not isinstance(value.get(key), str) or value[key] not in kinds for key in ("expected_kind", "actual_kind")): return None if any(type(value.get(key)) is not bool for key in booleans): return None result = {key: value[key] for key in ("expected_kind", "actual_kind", *booleans)} for key in ("xsec_token", "xsec_source"): item = value.get(key) if (not isinstance(item, dict) or type(item.get("values_equal")) is not bool or any(type(item.get(count)) is not int or not 0 <= item[count] <= 1000 for count in ("expected_count", "actual_count"))): return None result[key] = {field: item[field] for field in ("expected_count", "actual_count", "values_equal")} return result def safe_diagnostics(value, *, code=None, execution_profile=None) -> dict: value = value if isinstance(value, dict) else {} requests = [summary for item in value.get("blocked_requests", [])[:12] if (summary := safe_request_summary(item, blocked=True))] if isinstance(value.get("blocked_requests"), list) else [] count = value.get("blocked_request_count") phase = value.get("phase") result = {"phase": phase if isinstance(phase, str) and phase in FAILURE_PHASES else "unknown", "blocked_requests": requests, "blocked_request_count": min(count, 1_000_000) if type(count) is int and count >= 0 else len(requests)} if code == "NAVIGATION_BINDING_MISMATCH" and execution_profile == NAVIGATION_PROFILE: if comparison := safe_navigation_comparison(value.get("navigation_comparison")): result["navigation_comparison"] = comparison return result def safe_element_evidence(value: dict) -> dict: fields = {"selector", "tag", "label", "kind", "unique", "count", "sensitive", "in_viewport", "search_input"} result = {key: item for key, item in value.items() if key in fields and isinstance(item, (str, int, bool))} stable_label = value.get("stable_label") if isinstance(stable_label, str) and len(stable_label) <= 500: result["stable_label"] = stable_label basis = value.get("selector_basis") if isinstance(basis, str) and basis in {"id", "semantic", "scroll_container", "id_family", "sibling_class"}: result["selector_basis"] = basis if type(value.get("collection")) is bool: result["collection"] = value["collection"] sources = value.get("source_selectors") if isinstance(sources, list): result["source_selectors"] = [item[:1000] for item in sources[:3] if isinstance(item, str) and item.strip()] scroll = value.get("scroll_y") if isinstance(scroll, dict): cleaned = {key: item for key, item in scroll.items() if key in {"scroll_top", "scroll_height", "client_height"} and type(item) in {int, float} and 0 <= item <= 2 ** 53} if type(scroll.get("can_scroll")) is bool: cleaned["can_scroll"] = scroll["can_scroll"] result["scroll_y"] = cleaned hints = value.get("read_hints") if isinstance(hints, list): result["read_hints"] = [] for hint in hints[:12]: if not isinstance(hint, dict): continue cleaned = {key: item[:limit] for key, limit in (("selector", 1000), ("tag", 32), ("label", 100)) if isinstance(item := hint.get(key), str)} for key in ("unique", "in_viewport", "link"): if type(hint.get(key)) is bool: cleaned[key] = hint[key] if type(hint.get("count")) is int and 0 <= hint["count"] <= 2 ** 53: cleaned["count"] = hint["count"] source = hint.get("source") if isinstance(source, str) and source in {"text", "href", "attribute", "media"}: cleaned["source"] = source if source == "attribute" and hint.get("attribute") == "datetime": cleaned["attribute"] = "datetime" if source == "media" and isinstance(hint.get("media_kind"), str) and hint["media_kind"] in {"image", "video"} and type(hint.get("loaded")) is bool: cleaned.update(media_kind=hint["media_kind"], loaded=hint["loaded"]) elif source == "media": cleaned.pop("source", None) if cleaned.get("selector"): result["read_hints"].append(cleaned) return result def safe_failure_capture(value) -> dict | None: if not isinstance(value, dict) or not isinstance(value.get("url"), str) or not isinstance(value.get("elements"), list) or not isinstance(value.get("screenshot_path"), str): return None # These are already redacted page captures, not diagnostics. Preserve their # actual before/after state, while excluding arbitrary exception attributes. allowed = {"capture_id", "url", "title", "text", "viewport_text", "text_scope", "viewport", "elements", "screenshot_path", "captured_at", "identity_verification", "request_observations", "popup_observations"} capture = {key: deepcopy(item) for key, item in value.items() if key in allowed} if any(key in value for key in ("execution_profile", "execution_contract")): if value.get("execution_profile") is None or not execution_metadata_matches(value, value.get("execution_profile")): return None capture.update(execution_profile=value["execution_profile"], execution_contract=deepcopy(value["execution_contract"])) for key, limit in (("url", 4000), ("title", 250), ("text", 30000), ("viewport_text", 30000), ("capture_id", 128), ("text_scope", 64), ("captured_at", 64), ("identity_verification", 64), ("screenshot_path", 4000)): if key in capture: if not isinstance(capture[key], str): capture.pop(key) else: capture[key] = capture[key][:limit] capture["elements"] = [safe_element_evidence(element) for element in value["elements"][:300] if isinstance(element, dict)] capture["request_observations"] = safe_request_observations(capture.get("request_observations")) capture["popup_observations"] = safe_popup_observations(capture.get("popup_observations")) viewport = capture.get("viewport") if isinstance(viewport, dict): capture["viewport"] = {key: item for key, item in viewport.items() if key in {"width", "height", "scroll_x", "scroll_y"} and type(item) in {int, float}} else: capture.pop("viewport", None) return capture def safe_error_codes(value) -> list[str]: return list(dict.fromkeys(item for item in value[:10] if isinstance(item, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", item))) if isinstance(value, list) else [] def safe_failure_diagnostics(value, steps: list[dict], error_codes, *, passed=None, execution_profile=None) -> dict | None: """Keep a failed worker's bounded cause, never a new outcome or raw payload.""" if passed is not False or not isinstance(value, dict) or set(value) != {"step_position", "kind", "code", "diagnostics"}: return None position, kind, code = (value[key] for key in ("step_position", "kind", "code")) if (type(position) is not int or not 1 <= position <= len(steps) or not isinstance(kind, str) or kind != steps[position - 1].get("kind") or not isinstance(code, str) or code not in safe_error_codes(error_codes) or not isinstance(value["diagnostics"], dict)): return None return {"step_position": position, "kind": kind, "code": code, "diagnostics": safe_diagnostics(value["diagnostics"], code=code, execution_profile=execution_profile)} def safe_verify_timings(step_values, total_value, steps: list[dict], *, passed: bool = False) -> dict | None: def milliseconds(value): return type(value) is int and 0 <= value <= 120000 if not isinstance(total_value, dict) or set(total_value) != {"duration_ms", "entry_ms", "close_ms"}: return None if not milliseconds(total_value["duration_ms"]) or not milliseconds(total_value["entry_ms"]) or total_value["entry_ms"] > total_value["duration_ms"] + 1: return None if total_value["close_ms"] is not None and (not milliseconds(total_value["close_ms"]) or total_value["close_ms"] > total_value["duration_ms"] + 1): return None if not isinstance(step_values, list) or len(step_values) > min(30, len(steps)): return None if passed and (len(step_values) != len(steps) or any(not isinstance(item, dict) or item.get("status") != "passed" for item in step_values)): return None terminal = False for position, item in enumerate(step_values, 1): if terminal or not isinstance(item, dict) or set(item) != {"step_position", "kind", "status", "duration_ms", "phase_ms"}: return None if type(item["step_position"]) is not int or item["step_position"] != position or item["kind"] != steps[position - 1].get("kind") or not isinstance(item["status"], str) or item["status"] not in {"passed", "failed", "interrupted"}: return None terminal = item["status"] != "passed" if not milliseconds(item["duration_ms"]) or item["duration_ms"] > total_value["duration_ms"] + 1 or not isinstance(item["phase_ms"], dict) or not set(item["phase_ms"]) <= TIMING_PHASES: return None if any(not milliseconds(value) for value in item["phase_ms"].values()) or sum(item["phase_ms"].values()) > item["duration_ms"] + len(TIMING_PHASES): return None # Entry, step execution, and explicit closing are exclusive measurements; # allow only a small aggregate tolerance for independently rounded values. recorded_ms = total_value["entry_ms"] + sum(item["duration_ms"] for item in step_values) + (total_value["close_ms"] or 0) if recorded_ms > total_value["duration_ms"] + 30: return None return {"step_timings": deepcopy(step_values), "verify_timing": deepcopy(total_value)} def safe_batch_evidence(value) -> list[dict]: records = [] for item in value[:5] if isinstance(value, list) else []: if not isinstance(item, dict) or type(item.get("batch_index")) is not int or not 1 <= item["batch_index"] <= 5: continue if any(type(item.get(key)) is not int or not 0 <= item[key] <= 500 for key in ("added_count", "duplicate_count")): continue capture = safe_failure_capture(item.get("capture")) if capture: records.append({"batch_index": item["batch_index"], "added_count": item["added_count"], "duplicate_count": item["duplicate_count"], "capture": capture}) return records class WorkerError(Exception): def __init__(self, message: str, code: str = "worker_error", *, before=None, after=None, diagnostics=None, batch_evidence=None, error_codes=None, session_closed=False): super().__init__(message) self.code = safe_worker_code(code) self.before = safe_failure_capture(before) self.after = safe_failure_capture(after) captures = [capture for capture in (self.before, self.after) if capture is not None] profile = NAVIGATION_PROFILE if captures and all(execution_metadata_matches(capture, NAVIGATION_PROFILE) for capture in captures) else None self.diagnostics = safe_diagnostics(diagnostics, code=self.code, execution_profile=profile) self.batch_evidence = safe_batch_evidence(batch_evidence) self.error_codes = safe_error_codes(error_codes) self.session_closed = session_closed is True class BrowserWorker: """Serialized RPC to the single browser owner; never imports the old application.""" def __init__(self, root: Path, home: Path): self.root, self.home = root, home self.process: asyncio.subprocess.Process | None = None self.lock = asyncio.Lock() self.stderr_task: asyncio.Task | None = None self.closing = False self.close_task: asyncio.Task | None = None async def _start(self) -> None: if self.process and self.process.returncode is None: return node = os.environ.get("ASSISTANT_NODE") or shutil.which("node") if not node: raise WorkerError("找不到 Node.js，请通过项目启动脚本运行后台", "node_not_found") script = self.root / "tools" / "learning" / "browser-worker.mjs" if not script.is_file(): raise WorkerError("本机学习执行器尚未就绪", "worker_unavailable") environment = {**os.environ, "ASSISTANT_LEARNING_HOME": str(self.home)} self.process = await asyncio.create_subprocess_exec(node, str(script), cwd=self.root, env=environment, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=8 * 1024 * 1024) if self.closing: await self._stop() raise WorkerError("学习执行器正在正常退出", "worker_closed") process = self.process async def consume_stderr() -> None: assert process.stderr while await process.stderr.readline(): pass # Do not forward browser diagnostic text or URLs into business logs. self.stderr_task = asyncio.create_task(consume_stderr()) async def call(self, method: str, params: dict) -> dict: async with self.lock: if self.closing: raise WorkerError("学习执行器正在正常退出", "worker_closed") await self._start() process = self.process assert process and process.stdin and process.stdout request_id = uuid.uuid4().hex try: process.stdin.write((json.dumps({"id": request_id, "method": method, "params": params}, ensure_ascii=False) + "\n").encode()) await process.stdin.drain() async with asyncio.timeout(120): while True: line = await process.stdout.readline() if not line: raise WorkerError("本机学习浏览器已退出，请重新打开会话", "worker_closed") try: response = json.loads(line) except json.JSONDecodeError: continue if response.get("id") != request_id: continue if response.get("error"): error = response["error"] raise WorkerError(str(error.get("message", "学习执行失败")), error.get("code"), before=error.get("before"), after=error.get("after"), diagnostics=error.get("diagnostics"), batch_evidence=error.get("batch_evidence"), error_codes=error.get("error_codes"), session_closed=error.get("session_closed")) return response.get("result") or {} except (TimeoutError, BrokenPipeError, ConnectionResetError): await self._stop() raise WorkerError("学习执行器响应中断；请核对页面与步骤后重新打开，不会自动重放动作", "worker_timeout") from None async def _stop(self) -> None: process, self.process = self.process, None if not process: return if process.stdin: process.stdin.close() try: await asyncio.wait_for(process.wait(), 15) except TimeoutError: process.terminate() try: await asyncio.wait_for(process.wait(), 3) except TimeoutError: process.kill() await process.wait() if self.stderr_task: self.stderr_task.cancel() self.stderr_task = None async def close(self) -> None: # EOF must reach the child even while an RPC holds the request lock. # Node closes its contexts on EOF, unblocking that exact active action. self.closing = True if self.close_task is None: self.close_task = asyncio.create_task(self._stop()) await asyncio.shield(self.close_task)