Browser Harness 与隐私遥测 深度解析

106class BrowserHarness:
107    """Expose Browser Harness helpers without connecting until first use."""
108
109    def __init__(self, *, name: str | None = None, wait: float = 60.0) -> None:
110        self.name = name
111        self.wait = wait
112        self._helpers: ModuleType | None = None
113
114    def connect(self) -> BrowserHarness:
115        if self._helpers is not None:
116            return self
117
118        if self.name:
119            os.environ["BU_NAME"] = self.name
120
121        try:
122            from browser_harness import admin, helpers
123        except ImportError as exc:  # pragma: no cover - packaging failure
124            raise RuntimeError(
125                "Browser Harness is unavailable. Install the project dependencies "
126                "with `uv sync`."
127            ) from exc
128
129        approver = subprocess.Popen(
130            [
131                sys.executable,
132                "-c",
133                (
134                    "from macos_harness.browser import _watch_chrome_prompts; "
135                    f"_watch_chrome_prompts({self.wait!r})"
136                ),
137            ],
138            stdout=subprocess.DEVNULL,
139            stderr=subprocess.DEVNULL,
140        )
141        try:
142            admin.ensure_daemon(wait=self.wait, name=self.name)
143        finally:
144            if approver.poll() is None:
145                approver.terminate()
146                try:
147                    approver.wait(timeout=0.2)
148                except subprocess.TimeoutExpired:
149                    approver.kill()
150        if self.name:
151            helpers.NAME = self.name
152        self._helpers = helpers
153        return self
154
155    def __getattr__(self, name: str) -> Any:
156        if name.startswith("_"):
157            raise AttributeError(name)
158        self.connect()
159        assert self._helpers is not None
160        try:
161            return getattr(self._helpers, name)
162        except AttributeError as exc:
163            raise AttributeError(f"Browser Harness has no helper {name!r}") from exc
164
165    def __repr__(self) -> str:
166        status = "connected" if self._helpers is not None else "lazy"
167        suffix = f", name={self.name!r}" if self.name else ""
168        return f"BrowserHarness({status}{suffix})"

15def _approve_chrome_prompt(stop: threading.Event, timeout: float) -> None:
16    """Press Chrome's exact remote-debugging Allow button through system-wide AX."""
17    if platform.system() != "Darwin":
18        return
19    try:
20        import ApplicationServices as AS
21    except ImportError:  # pragma: no cover - macOS dependency failure
22        return
23
24    deadline = time.monotonic() + timeout
25    while not stop.is_set() and time.monotonic() < deadline:
26        root = AS.AXUIElementCreateSystemWide()
27        pressed = False
28        windows = AS.CGWindowListCopyWindowInfo(
29            AS.kCGWindowListOptionOnScreenOnly, AS.kCGNullWindowID
30        )
31        for window in windows:
32            if window.get("kCGWindowOwnerName") != "Google Chrome":
33                continue
34            if window.get("kCGWindowName") != "Allow remote debugging?":
35                continue
36            bounds = window.get("kCGWindowBounds") or {}
37            left = float(bounds.get("X", 0))
38            top = float(bounds.get("Y", 0))
39            width = float(bounds.get("Width", 0))
40            height = float(bounds.get("Height", 0))
41            points = [(left + width - 58, top + height - 38)]
42            points.extend(
43                (x, y)
44                for y in range(int(top + height - 16), int(top + height - 96), -16)
45                for x in range(int(left + width - 16), int(left), -16)
46            )
47            for x, y in points:
48                error, element = AS.AXUIElementCopyElementAtPosition(root, x, y, None)
49                if error != 0 or element is None:
50                    continue
51                role_error, role = AS.AXUIElementCopyAttributeValue(
52                    element, "AXRole", None
53                )
54                description_error, description = AS.AXUIElementCopyAttributeValue(
55                    element, "AXDescription", None
56                )
57                if (
58                    role_error == 0
59                    and description_error == 0
60                    and str(role) == "AXButton"
61                    and str(description) == "Allow"
62                ):
63                    AS.AXUIElementPerformAction(element, "AXPress")
64                    pressed = True
65                    break
66            if pressed:
67                break
68        stop.wait(0.2 if pressed else 0.05)

170    class OverlayController(AppKit.NSObject):
171        def init(self) -> Any:
172            controller = objc.super(OverlayController, self).init()
173            if controller is None:
174                return None
175            frame = AppKit.NSMakeRect(-100, -100, POINTER_WIDTH, POINTER_HEIGHT)
176            style = (
177                AppKit.NSWindowStyleMaskBorderless
178                | AppKit.NSWindowStyleMaskNonactivatingPanel
179            )
180            controller.panel = (
181                AppKit.NSPanel.alloc().initWithContentRect_styleMask_backing_defer_(
182                    frame, style, AppKit.NSBackingStoreBuffered, False
183                )
184            )
185            controller.view = PointerView.alloc().initWithFrame_(
186                AppKit.NSMakeRect(0, 0, POINTER_WIDTH, POINTER_HEIGHT)
187            )
188            controller.panel.setContentView_(controller.view)
189            controller.panel.setTitle_("macOS Harness Pointer")
190            controller.panel.setOpaque_(False)
191            controller.panel.setBackgroundColor_(AppKit.NSColor.clearColor())
192            controller.panel.setHasShadow_(False)
193            controller.panel.setIgnoresMouseEvents_(True)
194            controller.panel.setHidesOnDeactivate_(False)
195            controller.panel.setReleasedWhenClosed_(False)
196            controller.panel.setLevel_(AppKit.NSStatusWindowLevel + 1)
197            controller.panel.setCollectionBehavior_(
198                AppKit.NSWindowCollectionBehaviorCanJoinAllSpaces
199                | AppKit.NSWindowCollectionBehaviorFullScreenAuxiliary
200                | AppKit.NSWindowCollectionBehaviorStationary
201            )
202            controller.shown = False
203            controller.animation_timer = None
204            controller.animation_started = 0.0
205            controller.animation_duration = 0.0
206            controller.animation_start = (0.0, 0.0)
207            controller.animation_end = (0.0, 0.0)
208            return controller

24def _config_dir() -> Path:
25    override = os.environ.get("MACOS_HARNESS_HOME")
26    if override:
27        return Path(override).expanduser()
28    return Path.home() / "Library" / "Application Support" / "macos-harness"
29
30
31def _config_path() -> Path:
32    return _config_dir() / "telemetry.json"
33
34
35def _load() -> dict:
36    try:
37        return json.loads(_config_path().read_text(encoding="utf-8"))
38    except (FileNotFoundError, OSError, ValueError):
39        return {}
40
41
42def _save(config: dict) -> None:
43    try:
44        path = _config_path()
45        path.parent.mkdir(parents=True, exist_ok=True)
46        os.chmod(path.parent, 0o700)
47        path.write_text(
48            json.dumps(config, indent=2, sort_keys=True) + "\n", encoding="utf-8"
49        )
50        os.chmod(path, 0o600)
51    except OSError:
52        pass

80def is_enabled() -> bool:
81    return not _env_disabled() and not bool(_load().get("disabled"))
82
83
84def status() -> dict:
85    config = _load()
86    env_disabled = _env_disabled()
87    enabled = not env_disabled and not bool(config.get("disabled"))
88    return {
89        "enabled": enabled,
90        "disabled_by_env": env_disabled,
91        "disabled_by_config": bool(config.get("disabled")),
92        "install_id": _install_id(config, create=enabled),
93        "config_path": str(_config_path()),
94    }
95
96
97def set_enabled(enabled: bool) -> dict:
98    config = _load()
99    config["disabled"] = not enabled
100    _save(config)
101    return status()