from __future__ import annotations import hashlib import html import platform import re import ssl import time import webbrowser import ctypes import ctypes.wintypes from dataclasses import dataclass, field from html.parser import HTMLParser from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urldefrag, urljoin, urlparse from urllib.request import Request, urlopen TITLE_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) UNVERIFIED_HTTPS_CONTEXT = ssl._create_unverified_context() MAX_HTML_BYTES = 1_500_000 MAX_CHILD_TASKS = 8 def network_error_message(exc: BaseException, timeout_seconds: float) -> str: reason = getattr(exc, "reason", exc) if isinstance(exc, TimeoutError) or isinstance(reason, TimeoutError): return f"timed out after {timeout_seconds:g}s" return str(reason) or type(reason).__name__ @dataclass class ToolResult: ok: bool result: dict[str, Any] = field(default_factory=dict) error: str = "" duration_seconds: float = 0.0 def clean_text(value: Any, max_length: int = 160) -> str: text = html.unescape(re.sub(r"\s+", " ", str(value or ""))).strip() return text[:max_length].strip() def decode_body(raw: bytes, content_type: str) -> str: charset = "utf-8" match = re.search(r"charset=([^;\s]+)", content_type or "", re.IGNORECASE) if match: charset = match.group(1).strip("\"'") for candidate in [charset, "utf-8", "gbk", "latin-1"]: try: return raw.decode(candidate, errors="replace") except LookupError: continue return raw.decode("utf-8", errors="replace") def extract_title(raw: bytes, content_type: str) -> str: text = decode_body(raw, content_type) match = TITLE_RE.search(text) if not match: return "" return clean_text(match.group(1), 120) def normalize_url(value: Any) -> str: text = str(value or "").strip() if text and "://" not in text: text = f"http://{text}" return text def origin_tuple(url: str) -> tuple[str, str, int]: parsed = urlparse(url) port = parsed.port or (443 if parsed.scheme == "https" else 80) return parsed.scheme, (parsed.hostname or "").lower(), port def same_origin(base_url: str, candidate_url: str) -> bool: try: return origin_tuple(base_url) == origin_tuple(candidate_url) except ValueError: return False def safe_join_url(base_url: str, href: str) -> str | None: text = str(href or "").strip() if not text or text.startswith(("#", "javascript:", "mailto:", "tel:", "data:")): return None joined = urljoin(base_url, text) joined, _fragment = urldefrag(joined) parsed = urlparse(joined) if parsed.scheme not in {"http", "https"} or not parsed.netloc: return None if not same_origin(base_url, joined): return None return joined def stable_id(*parts: Any, prefix: str = "agent") -> str: raw = "|".join(str(part or "") for part in parts) return f"{prefix}-{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:12]}" def risk_for_text(text: str) -> str: lowered = text.lower() high = ["delete", "remove", "drop", "pay", "refund", "logout", "删除", "清空", "支付", "付款", "退款", "注销"] medium = ["save", "submit", "create", "update", "edit", "upload", "保存", "提交", "新增", "修改", "上传"] if any(word in lowered for word in high): return "high" if any(word in lowered for word in medium): return "medium" return "low" class WebPageParser(HTMLParser): def __init__(self) -> None: super().__init__(convert_charrefs=True) self.title_parts: list[str] = [] self.links: list[dict[str, Any]] = [] self.forms: list[dict[str, Any]] = [] self.buttons: list[dict[str, Any]] = [] self._in_title = False self._anchors: list[dict[str, Any]] = [] self._form: dict[str, Any] | None = None self._button: dict[str, Any] | None = None def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: tag = tag.lower() attr = {name.lower(): value or "" for name, value in attrs} if tag == "title": self._in_title = True elif tag == "a": self._anchors.append({"href": attr.get("href", ""), "text_parts": []}) elif tag == "form": self._form = { "action": attr.get("action", ""), "method": (attr.get("method") or "get").upper(), "id": attr.get("id", ""), "name": attr.get("name", ""), "class": attr.get("class", ""), "text_parts": [], "fields": [], } elif self._form is not None and tag in {"input", "select", "textarea"}: self._form["fields"].append( { "tag": tag, "type": (attr.get("type") or tag).lower(), "name": attr.get("name", ""), "id": attr.get("id", ""), "placeholder": attr.get("placeholder", ""), "aria_label": attr.get("aria-label", ""), } ) elif tag == "button": self._button = { "type": attr.get("type", ""), "id": attr.get("id", ""), "name": attr.get("name", ""), "class": attr.get("class", ""), "text_parts": [], "in_form": self._form is not None, } def handle_data(self, data: str) -> None: if self._in_title: self.title_parts.append(data) if self._anchors: self._anchors[-1]["text_parts"].append(data) if self._form is not None: self._form["text_parts"].append(data) if self._button is not None: self._button["text_parts"].append(data) def handle_endtag(self, tag: str) -> None: tag = tag.lower() if tag == "title": self._in_title = False elif tag == "a" and self._anchors: anchor = self._anchors.pop() self.links.append({"href": anchor.get("href", ""), "text": clean_text(" ".join(anchor.get("text_parts") or []), 120)}) elif tag == "button" and self._button is not None: button = self._button button["text"] = clean_text(" ".join(button.get("text_parts") or []), 120) self.buttons.append(button) self._button = None elif tag == "form" and self._form is not None: form = self._form form["text"] = clean_text(" ".join(form.get("text_parts") or []), 160) self.forms.append(form) self._form = None def looks_like_login_form(form: dict[str, Any]) -> bool: fields = form.get("fields") or [] text = " ".join( [ str(form.get("id") or ""), str(form.get("name") or ""), str(form.get("class") or ""), str(form.get("text") or ""), *[ " ".join( [ str(field.get("type") or ""), str(field.get("name") or ""), str(field.get("id") or ""), str(field.get("placeholder") or ""), str(field.get("aria_label") or ""), ] ) for field in fields ], ] ).lower() return "password" in text or any(keyword in text for keyword in ["login", "signin", "sign in", "登录", "账号", "密码"]) class WebTools: def open_in_browser(self, url: str, wait_seconds: float = 1.5) -> ToolResult: target_url = normalize_url(url) if not target_url: return ToolResult(ok=False, error="Missing target URL.") started = time.monotonic() try: opened = bool(webbrowser.open(target_url, new=2, autoraise=True)) if wait_seconds > 0: time.sleep(min(max(wait_seconds, 0.0), 5.0)) return ToolResult( ok=opened, result={"url": target_url, "opened": opened}, error="" if opened else "Default browser did not accept the URL.", duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult( ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3), ) def observe_page(self, url: str, timeout_seconds: float = 8) -> ToolResult: target_url = normalize_url(url) if not target_url: return ToolResult(ok=False, error="Missing target URL.") request = Request( target_url, headers={ "Accept": "text/html,application/xhtml+xml,*/*", "User-Agent": "SystemTest-Agent/0.1 WebTools", }, ) started = time.monotonic() try: with urlopen(request, timeout=timeout_seconds, context=UNVERIFIED_HTTPS_CONTEXT) as response: raw = response.read(MAX_HTML_BYTES) status = response.status final_url = response.geturl() content_type = response.headers.get("Content-Type", "") except HTTPError as exc: raw = exc.read(MAX_HTML_BYTES) status = exc.code final_url = exc.geturl() content_type = exc.headers.get("Content-Type", "") except (URLError, TimeoutError, OSError) as exc: return ToolResult( ok=False, error=f"Target URL is not reachable: {network_error_message(exc, timeout_seconds)}", duration_seconds=round(time.monotonic() - started, 3), ) text = decode_body(raw, content_type) parser = WebPageParser() try: parser.feed(text) except Exception: pass title = clean_text(" ".join(parser.title_parts), 120) or extract_title(raw, content_type) duration = round(time.monotonic() - started, 3) has_login_form = any(looks_like_login_form(form) for form in parser.forms) auth_challenge = status in {401, 403} return ToolResult( ok=status < 500, duration_seconds=duration, result={ "url": target_url, "final_url": final_url, "http_status": status, "content_type": content_type, "page_title": title, "elapsed_seconds": duration, "links": parser.links[:50], "forms": parser.forms[:20], "buttons": parser.buttons[:50], "requires_auth": auth_challenge, "auth_challenge": auth_challenge, "has_login_form": has_login_form, }, ) def discover_child_tasks( self, parent_task: dict[str, Any], observation: dict[str, Any], *, max_children: int = MAX_CHILD_TASKS, max_depth: int = 2, ) -> list[dict[str, Any]]: payload = parent_task.get("payload") if isinstance(parent_task.get("payload"), dict) else {} parent_external_id = str(parent_task.get("external_id") or parent_task.get("id") or "") root_external_id = str(parent_task.get("root_external_id") or payload.get("root_external_id") or parent_external_id) parent_depth = int(parent_task.get("depth") or payload.get("depth") or 0) discovery_max_depth = int(payload.get("discovery_max_depth") or max_depth) child_depth = parent_depth + 1 if child_depth > discovery_max_depth: return [] page_url = observation.get("final_url") or observation.get("url") or payload.get("target_url") or "" children: list[dict[str, Any]] = [] seen_urls: set[str] = set() for link in observation.get("links") or []: if len(children) >= max_children: break href = safe_join_url(page_url, str(link.get("href") or "")) if not href or href in seen_urls: continue seen_urls.add(href) label = clean_text(link.get("text"), 80) or urlparse(href).path.strip("/") or href risk = risk_for_text(label) children.append( self._child_task( parent_external_id, root_external_id, child_depth, label, f"探索页面:{label}", f"访问并检查页面“{label}”是否可达,记录标题、状态码和可交互入口。", href, "module", risk, "html_link", discovery_max_depth, ) ) for index, form in enumerate(observation.get("forms") or [], start=1): if len(children) >= max_children: break action_url = safe_join_url(page_url, str(form.get("action") or "")) or page_url is_login = looks_like_login_form(form) label = "登录表单" if is_login else clean_text(form.get("name") or form.get("id") or form.get("text"), 60) or f"表单 {index}" risk = "medium" if is_login else risk_for_text(label) child = self._child_task( parent_external_id, root_external_id, child_depth, label, f"识别功能:{label}", f"分析{label}的输入项与提交入口;P0 只确认元素存在,不执行真实提交。", action_url, "feature", risk, "html_form", discovery_max_depth, ) child["payload"]["preconditions"] = ["需要测试账号/密码"] if is_login else [] children.append(child) for button in observation.get("buttons") or []: if len(children) >= max_children: break label = clean_text(button.get("text") or button.get("name") or button.get("id"), 60) if not label: continue risk = risk_for_text(label) child = self._child_task( parent_external_id, root_external_id, child_depth, label, f"识别入口:{label}", f"确认入口“{label}”存在;高风险入口只记录,不执行破坏性动作。", page_url, "feature", risk, "html_button", discovery_max_depth, ) if risk == "high": child["failure_policy"] = "pause" children.append(child) unique: dict[str, dict[str, Any]] = {} for child in children: unique.setdefault(str(child["external_id"]), child) return list(unique.values())[:max_children] def _child_task( self, parent_external_id: str, root_external_id: str, depth: int, module: str, title: str, objective: str, target_url: str, level: str, risk: str, source: str, discovery_max_depth: int, ) -> dict[str, Any]: external_id = stable_id(root_external_id, parent_external_id, source, target_url, module) payload = { "task_type": "smart_discovery_p0", "instruction": objective, "objective": objective, "target_url": target_url, "source": source, "level": level, "depth": depth, "risk_level": risk, "allow_child_discovery": level == "module" and depth < discovery_max_depth, "discovery_max_depth": discovery_max_depth, "open_in_browser": True, "capture_screenshot": True, "hierarchy": { "id": external_id, "parent_id": parent_external_id, "root_id": root_external_id, "depth": depth, "level": level, }, } return { "external_id": external_id, "title": title, "module": module, "parent_external_id": parent_external_id, "root_external_id": root_external_id, "depth": depth, "task_level": level, "risk_level": risk, "priority": 40 + depth * 10 + (40 if risk == "high" else 20 if risk == "medium" else 0), "failure_policy": "pause" if risk == "high" else "skip", "max_retries": 2, "timeout_seconds": 60, "tags": ["agent-discovered", "p0"], "payload": payload, } class GuiTools: def __init__(self) -> None: self._user32 = ctypes.windll.user32 if self.available() else None self._enum_proc_type = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.wintypes.HWND, ctypes.wintypes.LPARAM) if self.available() else None if self._user32 and self._enum_proc_type: self._user32.EnumWindows.argtypes = [ctypes.c_void_p, ctypes.wintypes.LPARAM] self._user32.EnumWindows.restype = ctypes.c_bool self._user32.EnumChildWindows.argtypes = [ctypes.wintypes.HWND, ctypes.c_void_p, ctypes.wintypes.LPARAM] self._user32.EnumChildWindows.restype = ctypes.c_bool def available(self) -> bool: return platform.system() == "Windows" and hasattr(ctypes, "windll") def backend_status(self) -> dict[str, Any]: if not self.available(): return { "available": False, "backend": "ctypes-win32", "reason": "Windows GUI APIs are unavailable on this platform.", } return { "available": True, "backend": "ctypes-win32", "capabilities": [ "list_windows", "inspect_window_controls", "capture_desktop", "capture_window", "focus_window", "click_window_point", ], } def list_windows(self, limit: int = 80) -> ToolResult: started = time.monotonic() if not self.available(): return ToolResult(ok=False, error=self.backend_status()["reason"]) try: windows = self._enum_windows(limit=limit) return ToolResult( ok=True, result={"windows": windows, "backend": self.backend_status()}, duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def inspect_window(self, window_id: str, limit: int = 220) -> ToolResult: started = time.monotonic() if not self.available(): return ToolResult(ok=False, error=self.backend_status()["reason"]) try: hwnd = self._parse_hwnd(window_id) if not self._user32.IsWindow(hwnd): return ToolResult(ok=False, error=f"Window not found: {window_id}") return ToolResult( ok=True, result={ "window": self._window_info(hwnd), "controls": self._enum_child_windows(hwnd, limit=limit), "backend": self.backend_status(), }, duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def capture_desktop(self, save_dir: Path, prefix: str = "desktop") -> ToolResult: started = time.monotonic() try: from PIL import ImageGrab save_dir.mkdir(parents=True, exist_ok=True) image = ImageGrab.grab(all_screens=True) path = save_dir / f"{prefix}_{int(time.time() * 1000)}.png" image.save(path) return ToolResult( ok=True, result={"image_path": str(path), "width": image.width, "height": image.height}, duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def capture_window(self, window_id: str, save_dir: Path, prefix: str = "window") -> ToolResult: started = time.monotonic() if not self.available(): return ToolResult(ok=False, error=self.backend_status()["reason"]) try: from PIL import ImageGrab hwnd = self._parse_hwnd(window_id) if not self._user32.IsWindow(hwnd): return ToolResult(ok=False, error=f"Window not found: {window_id}") rect = self._rect(hwnd) if rect["width"] <= 0 or rect["height"] <= 0: return ToolResult(ok=False, error=f"Window has invalid bounds: {window_id}") save_dir.mkdir(parents=True, exist_ok=True) image = ImageGrab.grab(bbox=(rect["left"], rect["top"], rect["right"], rect["bottom"])) path = save_dir / f"{prefix}_{window_id}_{int(time.time() * 1000)}.png" image.save(path) return ToolResult( ok=True, result={ "image_path": str(path), "window": self._window_info(hwnd), "width": image.width, "height": image.height, }, duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def focus_window(self, window_id: str) -> ToolResult: started = time.monotonic() if not self.available(): return ToolResult(ok=False, error=self.backend_status()["reason"]) try: hwnd = self._parse_hwnd(window_id) if not self._user32.IsWindow(hwnd): return ToolResult(ok=False, error=f"Window not found: {window_id}") self._user32.ShowWindow(hwnd, 9) # SW_RESTORE ok = bool(self._user32.SetForegroundWindow(hwnd)) return ToolResult( ok=ok, result={"window": self._window_info(hwnd), "focused": ok}, error="" if ok else "SetForegroundWindow returned false.", duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def click_window_point(self, window_id: str, x: float, y: float, normalized: bool = True) -> ToolResult: started = time.monotonic() if not self.available(): return ToolResult(ok=False, error=self.backend_status()["reason"]) try: hwnd = self._parse_hwnd(window_id) if not self._user32.IsWindow(hwnd): return ToolResult(ok=False, error=f"Window not found: {window_id}") rect = self._rect(hwnd) if normalized: target_x = int(rect["left"] + max(0.0, min(1.0, float(x))) * rect["width"]) target_y = int(rect["top"] + max(0.0, min(1.0, float(y))) * rect["height"]) else: target_x = int(x) target_y = int(y) self.focus_window(window_id) self._user32.SetCursorPos(target_x, target_y) self._user32.mouse_event(0x0002, 0, 0, 0, 0) # MOUSEEVENTF_LEFTDOWN self._user32.mouse_event(0x0004, 0, 0, 0, 0) # MOUSEEVENTF_LEFTUP return ToolResult( ok=True, result={"window_id": str(window_id), "x": target_x, "y": target_y, "normalized": normalized}, duration_seconds=round(time.monotonic() - started, 3), ) except Exception as exc: return ToolResult(ok=False, error=str(exc), duration_seconds=round(time.monotonic() - started, 3)) def _parse_hwnd(self, window_id: str) -> int: text = str(window_id).strip() if text.lower().startswith("0x"): return int(text, 16) return int(text) def _window_text(self, hwnd: int, max_length: int = 512) -> str: length = min(max(self._user32.GetWindowTextLengthW(hwnd), 0) + 1, max_length) buffer = ctypes.create_unicode_buffer(length) self._user32.GetWindowTextW(hwnd, buffer, length) return clean_text(buffer.value, 200) def _class_name(self, hwnd: int) -> str: buffer = ctypes.create_unicode_buffer(256) self._user32.GetClassNameW(hwnd, buffer, 256) return clean_text(buffer.value, 120) def _rect(self, hwnd: int) -> dict[str, int]: rect = ctypes.wintypes.RECT() self._user32.GetWindowRect(hwnd, ctypes.byref(rect)) return { "left": int(rect.left), "top": int(rect.top), "right": int(rect.right), "bottom": int(rect.bottom), "width": max(0, int(rect.right - rect.left)), "height": max(0, int(rect.bottom - rect.top)), } def _process_id(self, hwnd: int) -> int: pid = ctypes.wintypes.DWORD() self._user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) return int(pid.value) def _window_info(self, hwnd: int) -> dict[str, Any]: return { "id": str(int(hwnd)), "hex_id": hex(int(hwnd)), "title": self._window_text(hwnd), "class_name": self._class_name(hwnd), "process_id": self._process_id(hwnd), "visible": bool(self._user32.IsWindowVisible(hwnd)), "enabled": bool(self._user32.IsWindowEnabled(hwnd)), "minimized": bool(self._user32.IsIconic(hwnd)), "rect": self._rect(hwnd), } def _enum_windows(self, limit: int = 80) -> list[dict[str, Any]]: windows: list[dict[str, Any]] = [] enum_proc = self._enum_proc_type @enum_proc def callback(hwnd: int, _lparam: int) -> bool: if len(windows) >= limit: return False if not self._user32.IsWindowVisible(hwnd): return True info = self._window_info(hwnd) rect = info["rect"] if rect["width"] < 40 or rect["height"] < 40: return True if not info["title"]: return True if info["title"] or info["class_name"]: info["index"] = len(windows) + 1 windows.append(info) return True self._user32.EnumWindows(ctypes.cast(callback, ctypes.c_void_p), ctypes.wintypes.LPARAM(0)) return windows def _enum_child_windows(self, hwnd: int, limit: int = 220) -> list[dict[str, Any]]: controls: list[dict[str, Any]] = [] enum_proc = self._enum_proc_type @enum_proc def callback(child_hwnd: int, _lparam: int) -> bool: if len(controls) >= limit: return False rect = self._rect(child_hwnd) if rect["width"] <= 0 or rect["height"] <= 0: return True text = self._window_text(child_hwnd) class_name = self._class_name(child_hwnd) if not text and not class_name: return True controls.append( { "id": str(int(child_hwnd)), "hex_id": hex(int(child_hwnd)), "index": len(controls) + 1, "text": text, "class_name": class_name, "visible": bool(self._user32.IsWindowVisible(child_hwnd)), "enabled": bool(self._user32.IsWindowEnabled(child_hwnd)), "rect": rect, } ) return True self._user32.EnumChildWindows(hwnd, ctypes.cast(callback, ctypes.c_void_p), ctypes.wintypes.LPARAM(0)) return controls class AgentToolRegistry: def __init__(self) -> None: self.web = WebTools() self.gui = GuiTools() def descriptions(self) -> list[dict[str, Any]]: gui_status = self.gui.backend_status() return [ { "name": "open_in_browser", "scope": "web", "available": True, "description": "Open the target URL in the controlled device default browser so the operator can see the test start.", }, { "name": "observe_web_page", "scope": "web", "available": True, "description": "Open a web page, read status/title, and list links/forms/buttons without side effects.", }, { "name": "discover_web_subtasks", "scope": "web", "available": True, "description": "Create hierarchy-aware P0 child tasks from observed links/forms/buttons.", }, { "name": "list_windows", "scope": "gui", "available": gui_status["available"], "description": "List visible Windows desktop application windows using a dependency-free ctypes Win32 backend.", }, { "name": "inspect_window_controls", "scope": "gui", "available": gui_status["available"], "description": "Enumerate Win32 child controls for a selected window.", }, { "name": "capture_desktop", "scope": "gui", "available": True, "description": "Capture the desktop screenshot with PIL ImageGrab and save it as local evidence.", }, { "name": "capture_window", "scope": "gui", "available": gui_status["available"], "description": "Capture a selected window region and save it as local evidence.", }, { "name": "focus_window", "scope": "gui", "available": gui_status["available"], "description": "Bring a window to foreground. This is an action tool and should be used only when the task allows it.", }, { "name": "click_window_point", "scope": "gui", "available": gui_status["available"], "description": "Click a normalized or absolute point in a selected window. This is guarded and not used by default in P0 observation.", }, { "name": "no_action", "scope": "agent", "available": True, "description": "Finish, fail, or wait for user input without executing a UI action.", }, ]