| 1 | """agentcookie sidecar reader — an X cookie source for the bird backend. |
| 2 | |
| 3 | ``agentcookie`` is an external, user-installed CLI that can deliver browser |
| 4 | cookies on Linux (where Chrome's SQLite cookie store cannot be decrypted by |
| 5 | this engine's stdlib extractor). This module shells out to it and pulls the |
| 6 | ``auth_token`` + ``ct0`` pair that the bird backend needs. |
| 7 | |
| 8 | Deliberate constraints (see docs/plans/2026-08-31 X plan): |
| 9 | |
| 10 | * **Soft dependency.** Activation is gated on ``shutil.which("agentcookie")`` |
| 11 | resolving on the agent subprocess PATH, exactly like the other CLI-gated |
| 12 | optional sources (Digg, yt-dlp). A binary that is absent is not an error. |
| 13 | * **``AGENTCOOKIE=off`` disables it** regardless of PATH. |
| 14 | * **Independent of ``FROM_BROWSER``.** Reading the sidecar is not a browser |
| 15 | extraction, so it runs even when ``FROM_BROWSER`` is unset (the state in |
| 16 | which the in-process browser extractor stays off). |
| 17 | * We call ``agentcookie cookies --domain .x.com --json`` and parse its stdout. |
| 18 | We never open ``cookies-plain.db`` directly, never import an ``agentcookie`` |
| 19 | Python package, and never add a Python dependency on it. |
| 20 | * Cookie **values are never logged**. Only counts / names are ever emitted. |
| 21 | * First complete pair wins: a lone ``auth_token`` or a lone ``ct0`` is not a |
| 22 | usable result — both must be present. |
| 23 | """ |
| 24 | |
| 25 | from __future__ import annotations |
| 26 | |
| 27 | import json |
| 28 | import os |
| 29 | import shutil |
| 30 | import subprocess |
| 31 | from pathlib import Path |
| 32 | from typing import Any, Dict, List, Optional |
| 33 | |
| 34 | from . import log |
| 35 | |
| 36 | AGENTCOOKIE_BIN = "agentcookie" |
| 37 | X_COOKIE_DOMAIN = ".x.com" |
| 38 | X_COOKIE_NAMES = ("auth_token", "ct0") |
| 39 | |
| 40 | # agentcookie writes its config (including its role: "source" or "sink") here. |
| 41 | # A SINK receives cookies delivered from another machine; on Darwin, a sink is |
| 42 | # an "extra host" that gets the extra cookie lookups even though it is not a |
| 43 | # Mac mini. Reading this file is a plain filesystem read — NEVER a subprocess — |
| 44 | # so a MacBook (source/unknown role) never spawns agentcookie just to be |
| 45 | # classified (see the AE8 no-subprocess contract). Override the path for tests |
| 46 | # or non-default installs with AGENTCOOKIE_CONFIG. |
| 47 | _DEFAULT_CONFIG_PATH = Path.home() / ".config" / "agentcookie" / "config.json" |
| 48 | |
| 49 | # The sidecar read shells out to another process; keep it bounded so a hung |
| 50 | # agentcookie never stalls config loading. |
| 51 | _TIMEOUT_SECONDS = 10 |
| 52 | |
| 53 | |
| 54 | def _log(msg: str) -> None: |
| 55 | log.source_log("agentcookie", msg, tty_only=False) |
| 56 | |
| 57 | |
| 58 | def is_disabled(config: Optional[Dict[str, Any]] = None) -> bool: |
| 59 | """True when ``AGENTCOOKIE=off`` in config/env disables the sidecar.""" |
| 60 | raw = "" |
| 61 | if config is not None: |
| 62 | raw = config.get("AGENTCOOKIE") or "" |
| 63 | if not raw: |
| 64 | from . import env |
| 65 | |
| 66 | raw = env.read_secret_env("AGENTCOOKIE") or "" |
| 67 | return str(raw).strip().lower() == "off" |
| 68 | |
| 69 | |
| 70 | def is_available(config: Optional[Dict[str, Any]] = None) -> bool: |
| 71 | """True when the sidecar could be used: on PATH and not disabled. |
| 72 | |
| 73 | PATH-only, side-effect free (no subprocess). Safe for the doctor / |
| 74 | preflight prediction path — it reads no cookies, only whether the binary |
| 75 | would be reachable at run time. |
| 76 | """ |
| 77 | if is_disabled(config): |
| 78 | return False |
| 79 | return shutil.which(AGENTCOOKIE_BIN) is not None |
| 80 | |
| 81 | |
| 82 | def _config_path(config: Optional[Dict[str, Any]] = None) -> Path: |
| 83 | """Path to agentcookie's config file (AGENTCOOKIE_CONFIG override wins).""" |
| 84 | override = "" |
| 85 | if config is not None: |
| 86 | override = config.get("AGENTCOOKIE_CONFIG") or "" |
| 87 | override = override or os.environ.get("AGENTCOOKIE_CONFIG") or "" |
| 88 | return Path(override) if override else _DEFAULT_CONFIG_PATH |
| 89 | |
| 90 | |
| 91 | def role(config: Optional[Dict[str, Any]] = None) -> Optional[str]: |
| 92 | """agentcookie's configured role ("source"/"sink"), or None. |
| 93 | |
| 94 | Reads and parses the config FILE only — no subprocess, so it is safe on a |
| 95 | MacBook that must not spawn agentcookie. Any failure (missing file, bad |
| 96 | JSON, no ``role`` key, wrong type) returns None. "parse failure = not sink". |
| 97 | """ |
| 98 | path = _config_path(config) |
| 99 | try: |
| 100 | data = json.loads(path.read_text(encoding="utf-8")) |
| 101 | except (OSError, ValueError): |
| 102 | return None |
| 103 | if not isinstance(data, dict): |
| 104 | return None |
| 105 | raw = data.get("role") |
| 106 | return raw.strip().lower() if isinstance(raw, str) else None |
| 107 | |
| 108 | |
| 109 | def role_is_sink(config: Optional[Dict[str, Any]] = None) -> bool: |
| 110 | """True only when agentcookie's configured role parses as ``sink``.""" |
| 111 | return role(config) == "sink" |
| 112 | |
| 113 | |
| 114 | def _consume_cookie_list(items: List[Any], names: tuple, found: Dict[str, str]) -> None: |
| 115 | """Collect ``name -> value`` for wanted cookie names from a cookie list.""" |
| 116 | for item in items: |
| 117 | if not isinstance(item, dict): |
| 118 | continue |
| 119 | name = item.get("name") |
| 120 | value = item.get("value") |
| 121 | if name in names and isinstance(value, str) and value and name not in found: |
| 122 | found[name] = value |
| 123 | |
| 124 | |
| 125 | def _pair_from_json(data: Any, names: tuple) -> Dict[str, str]: |
| 126 | """Extract wanted cookies from agentcookie JSON, tolerant of its shape. |
| 127 | |
| 128 | Accepts a list of ``{"name","value",...}`` objects, a ``{"cookies": [...]}`` |
| 129 | wrapper, or a flat ``{name: value}`` / ``{name: {"value": ...}}`` mapping. |
| 130 | """ |
| 131 | found: Dict[str, str] = {} |
| 132 | if isinstance(data, list): |
| 133 | _consume_cookie_list(data, names, found) |
| 134 | elif isinstance(data, dict): |
| 135 | cookies = data.get("cookies") |
| 136 | if isinstance(cookies, list): |
| 137 | _consume_cookie_list(cookies, names, found) |
| 138 | else: |
| 139 | for name in names: |
| 140 | raw = data.get(name) |
| 141 | if isinstance(raw, str) and raw: |
| 142 | found[name] = raw |
| 143 | elif isinstance(raw, dict) and isinstance(raw.get("value"), str) and raw["value"]: |
| 144 | found[name] = raw["value"] |
| 145 | return found |
| 146 | |
| 147 | |
| 148 | def read_x_cookies(config: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, str]]: |
| 149 | """Return the complete X cookie pair from agentcookie, or None. |
| 150 | |
| 151 | Returns ``{"auth_token": ..., "ct0": ...}`` only when BOTH cookies are |
| 152 | present (no half-pair). Any failure (binary absent, disabled, non-zero |
| 153 | exit, unparsable JSON, timeout, incomplete pair) returns None so the |
| 154 | caller falls through to the next cookie source. Never raises. |
| 155 | """ |
| 156 | if is_disabled(config): |
| 157 | return None |
| 158 | binary = shutil.which(AGENTCOOKIE_BIN) |
| 159 | if binary is None: |
| 160 | return None |
| 161 | |
| 162 | try: |
| 163 | result = subprocess.run( |
| 164 | [binary, "cookies", "--domain", X_COOKIE_DOMAIN, "--json"], |
| 165 | capture_output=True, |
| 166 | text=True, |
| 167 | timeout=_TIMEOUT_SECONDS, |
| 168 | ) |
| 169 | except subprocess.TimeoutExpired: |
| 170 | _log(f"timed out after {_TIMEOUT_SECONDS}s; skipping") |
| 171 | return None |
| 172 | except OSError as exc: |
| 173 | _log(f"could not run agentcookie: {type(exc).__name__}") |
| 174 | return None |
| 175 | |
| 176 | if result.returncode != 0: |
| 177 | # stderr may carry a reason; do not echo it (it can contain values). |
| 178 | _log(f"exited {result.returncode}; skipping") |
| 179 | return None |
| 180 | |
| 181 | output = (result.stdout or "").strip() |
| 182 | if not output: |
| 183 | return None |
| 184 | try: |
| 185 | data = json.loads(output) |
| 186 | except json.JSONDecodeError: |
| 187 | _log("returned non-JSON output; skipping") |
| 188 | return None |
| 189 | |
| 190 | found = _pair_from_json(data, X_COOKIE_NAMES) |
| 191 | if all(name in found for name in X_COOKIE_NAMES): |
| 192 | _log(f"delivered a complete X cookie pair ({len(found)} of {len(X_COOKIE_NAMES)} names)") |
| 193 | return {name: found[name] for name in X_COOKIE_NAMES} |
| 194 | if found: |
| 195 | _log(f"returned an incomplete pair ({sorted(found)}); ignoring per no-half-pair rule") |
| 196 | return None |
| 197 |