From 2b2b79ae9c3e6d73d143263cb486f289060ceb2b Mon Sep 17 00:00:00 2001 From: iliya Date: Thu, 14 May 2026 00:23:10 +0300 Subject: [PATCH] feat: add twitter session safety checks --- scripts/twitter_browser.py | 316 +++++++++++++++++++++++++++++++++++-- 1 file changed, 304 insertions(+), 12 deletions(-) diff --git a/scripts/twitter_browser.py b/scripts/twitter_browser.py index 791b56ba1..1f4e1c22d 100644 --- a/scripts/twitter_browser.py +++ b/scripts/twitter_browser.py @@ -5,6 +5,12 @@ Each function does all browser work internally and returns structured JSON. Usage: + # Check isolated X/Twitter browser profile login + python3 twitter_browser.py whoami + + # Open the isolated browser profile and wait for manual login + TWITTER_HEADLESS=0 python3 twitter_browser.py login + # Reply to a tweet python3 twitter_browser.py reply "https://x.com/user/status/123" "reply text" @@ -19,8 +25,8 @@ Requires: pip install playwright && playwright install chromium -Connects to the running twitter-agent MCP browser via CDP (Chrome DevTools Protocol) -to reuse the existing logged-in session. +Connects to the running twitter-agent MCP browser via CDP for automation commands, +or launches an isolated persistent browser profile for login/whoami. """ import atexit @@ -33,13 +39,55 @@ import time -PROFILE_DIR = os.path.expanduser("~/.claude/browser-profiles/twitter") +CONFIG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "config.json") +DEFAULT_PROFILE_DIR = "~/.claude/browser-profiles/twitter" +PROFILE_DIR = os.path.expanduser(os.environ.get("TWITTER_PROFILE_DIR", DEFAULT_PROFILE_DIR)) LOCK_FILE = os.path.expanduser("~/.claude/twitter-agent-lock.json") LOCK_EXPIRY = 300 # Must match twitter-agent-lock.sh LOCK_WAIT_MAX = 45 # seconds to wait for lock to free before giving up LOCK_POLL_INTERVAL = 2 VIEWPORT = {"width": 911, "height": 1016} -OUR_HANDLE = "m13v_" +HEADLESS = os.environ.get("TWITTER_HEADLESS", "1").lower() not in ("0", "false", "no") +DEFAULT_BROWSER_EXECUTABLE = "/Applications/Brave Browser.app/Contents/MacOS/Brave Browser" +BROWSER_EXECUTABLE = os.environ.get("TWITTER_BROWSER_EXECUTABLE", DEFAULT_BROWSER_EXECUTABLE) + + +def _normalize_twitter_handle(value): + if not value: + return "" + raw = str(value).strip() + url_match = re.search(r"(?:x|twitter)\.com/([^/?#]+)", raw, re.I) + if url_match: + raw = url_match.group(1) + raw = raw.strip().lstrip("@").rstrip("/") + raw = raw.split()[0] if raw.split() else raw + if raw.lower() in {"your-twitter-handle", "your_handle", "username"}: + return "" + if re.fullmatch(r"[A-Za-z0-9_]{1,15}", raw): + return raw + return "" + + +def _load_config_twitter_handle(): + if os.environ.get("TWITTER_ALLOW_ANY_HANDLE", "").lower() in ("1", "true", "yes"): + return "" + env_handle = _normalize_twitter_handle(os.environ.get("TWITTER_HANDLE", "")) + if env_handle: + return env_handle + try: + with open(CONFIG_PATH) as f: + config = json.load(f) + config_handle = ( + config.get("accounts", {}) + .get("twitter", {}) + .get("handle", "") + ) + return _normalize_twitter_handle(config_handle) or "m13v_" + except Exception: + return "m13v_" + + +OUR_HANDLE = _load_config_twitter_handle() # DM encryption passcode from .env DM_PASSCODE = os.environ.get("TWITTER_DM_PASSCODE", "") @@ -266,15 +314,22 @@ def _refresh_browser_lock(): pass -def get_browser_and_page(playwright): +def _browser_executable_path(): + path = os.path.expanduser(BROWSER_EXECUTABLE) if BROWSER_EXECUTABLE else "" + if path and os.path.exists(path): + return path + return None + + +def get_browser_and_page(playwright, prefer_cdp=True): """Connect to the twitter-agent MCP browser via CDP, or launch a new one. Returns (browser, page, is_cdp). When is_cdp=True, `page` is a reused existing Twitter tab (navigate it, don't close it). When is_cdp=False, - it's a new headless page. + it's a new page in the isolated persistent profile. """ _acquire_browser_lock() - cdp_port = find_twitter_cdp_port() + cdp_port = find_twitter_cdp_port() if prefer_cdp else None if cdp_port: try: @@ -299,12 +354,23 @@ def get_browser_and_page(playwright): deadline = time.time() + LOCK_WAIT_MAX while True: try: + launch_kwargs = { + "headless": HEADLESS, + "user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", + } + args = ["--disable-blink-features=AutomationControlled"] + if HEADLESS: + launch_kwargs["viewport"] = VIEWPORT + else: + launch_kwargs["no_viewport"] = True + args.extend(["--start-maximized", "--window-size=1280,1000"]) + launch_kwargs["args"] = args + executable_path = _browser_executable_path() + if executable_path: + launch_kwargs["executable_path"] = executable_path context = playwright.chromium.launch_persistent_context( PROFILE_DIR, - headless=True, - args=["--disable-blink-features=AutomationControlled"], - viewport=VIEWPORT, - user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", + **launch_kwargs, ) break except Exception as e: @@ -320,6 +386,217 @@ def get_browser_and_page(playwright): return context, page, False +def _twitter_login_state(page): + """Read X login state from the current page without posting anything.""" + return page.evaluate("""(expectedHandle) => { + const reserved = new Set([ + 'home', 'explore', 'notifications', 'messages', 'i', 'intent', + 'settings', 'search', 'compose', 'login', 'signup', 'privacy', + 'tos', 'download', 'jobs', 'help' + ]); + + function cleanHandle(raw) { + if (!raw) return ''; + let value = String(raw).trim(); + const at = value.match(/@([A-Za-z0-9_]{1,15})/); + if (at) { + value = at[1]; + } else { + value = value.replace(/^@/, '').split(/[\\s/?#]/)[0]; + } + if (!/^[A-Za-z0-9_]{1,15}$/.test(value)) return ''; + if (reserved.has(value.toLowerCase())) return ''; + return value; + } + + const candidates = []; + function addCandidate(raw, source) { + const handle = cleanHandle(raw); + if (!handle) return; + if (candidates.some(c => c.handle.toLowerCase() === handle.toLowerCase())) return; + candidates.push({handle, source}); + } + + const switcher = document.querySelector('[data-testid="SideNav_AccountSwitcher_Button"]'); + if (switcher) { + addCandidate(switcher.textContent, 'account_switcher_text'); + addCandidate(switcher.getAttribute('aria-label'), 'account_switcher_aria'); + } + + for (const el of document.querySelectorAll('[aria-label]')) { + const label = el.getAttribute('aria-label') || ''; + if (label.includes('@')) addCandidate(label, 'aria_label'); + } + + for (const link of document.querySelectorAll('a[href^="/"]')) { + const href = link.getAttribute('href') || ''; + const parts = href.split('?')[0].split('#')[0].split('/').filter(Boolean); + if (parts.length === 1) addCandidate(parts[0], 'profile_link'); + const text = link.textContent || ''; + if (text.includes('@')) addCandidate(text, 'link_text'); + } + + const expected = cleanHandle(expectedHandle); + let expectedFound = false; + if (expected) { + const lowerExpected = expected.toLowerCase(); + expectedFound = candidates.some(c => c.handle.toLowerCase() === lowerExpected); + if (!expectedFound) { + expectedFound = Boolean( + document.querySelector(`a[href="/${expected}"], a[href^="/${expected}/"]`) + ); + } + } + + const url = window.location.href; + const path = window.location.pathname || ''; + const bodyText = ((document.body && document.body.innerText) || '') + .replace(/\\s+/g, ' ') + .trim(); + const appNav = Boolean( + document.querySelector('a[href="/home"], a[data-testid="AppTabBar_Home_Link"], a[href="/notifications"], [data-testid="SideNav_AccountSwitcher_Button"], [data-testid="SideNav_NewTweet_Button"]') + ); + const loginUrl = /\\/login|\\/i\\/flow\\/login/i.test(path); + const challengeUrl = /\\/account\\/access|\\/challenge/i.test(url); + const loginCta = /Sign in|Log in|Create account/i.test(bodyText.slice(0, 800)); + const loggedIn = appNav && !loginUrl && !challengeUrl; + + let handle = ''; + if (expectedFound && expected) { + handle = expected; + } else if (candidates.length) { + handle = candidates[0].handle; + } + + return { + url, + logged_in: loggedIn, + login_url: loginUrl, + challenge_url: challengeUrl, + login_cta: loginCta, + handle, + expected_handle: expected, + handle_matches: Boolean( + loggedIn && expected && handle && + handle.toLowerCase() === expected.toLowerCase() + ), + candidates: candidates.slice(0, 8), + body_start: bodyText.slice(0, 300) + }; + }""", OUR_HANDLE) + + +def _twitter_handle_matches(state): + if not state or not state.get("logged_in"): + return False + expected = _normalize_twitter_handle(state.get("expected_handle", "")) + if not expected: + return True + return bool(state.get("handle_matches")) + + +def _profile_payload(extra=None): + payload = { + "profile_dir": PROFILE_DIR, + "expected_handle": OUR_HANDLE, + "headless": HEADLESS, + "browser_executable": _browser_executable_path(), + } + if extra: + payload.update(extra) + return payload + + +def whoami(): + """Check which X account is logged into the isolated browser profile.""" + from playwright.sync_api import sync_playwright + + with sync_playwright() as p: + browser, page, is_cdp = get_browser_and_page(p, prefer_cdp=False) + try: + try: + page.goto("https://x.com/home", wait_until="domcontentloaded", timeout=45000) + except Exception: + pass + page.wait_for_timeout(5000) + state = _twitter_login_state(page) + return _profile_payload({ + "ok": _twitter_handle_matches(state), + "state": state, + "using_cdp": is_cdp, + }) + finally: + if not is_cdp: + page.close() + browser.close() + + +def login(wait_seconds=600): + """Open a visible isolated browser and wait for manual X login.""" + if HEADLESS: + return _profile_payload({ + "ok": False, + "error": "login_requires_headed_browser", + "hint": "Run: TWITTER_HEADLESS=0 python3 scripts/twitter_browser.py login", + }) + + from playwright.sync_api import sync_playwright + + with sync_playwright() as p: + browser, page, is_cdp = get_browser_and_page(p, prefer_cdp=False) + try: + try: + page.goto("https://x.com/home", wait_until="domcontentloaded", timeout=45000) + except Exception: + pass + page.wait_for_timeout(3000) + state = _twitter_login_state(page) + if _twitter_handle_matches(state): + return _profile_payload({"ok": True, "state": state, "using_cdp": is_cdp}) + + if not state.get("logged_in"): + try: + page.goto("https://x.com/i/flow/login", wait_until="domcontentloaded", timeout=45000) + except Exception: + pass + + deadline = time.time() + max(1, int(wait_seconds)) + last_state = state + while time.time() < deadline: + page.wait_for_timeout(3000) + _refresh_browser_lock() + last_state = _twitter_login_state(page) + if _twitter_handle_matches(last_state): + return _profile_payload({ + "ok": True, + "state": last_state, + "using_cdp": is_cdp, + }) + if ( + last_state.get("logged_in") + and last_state.get("handle") + and last_state.get("expected_handle") + and not last_state.get("handle_matches") + ): + return _profile_payload({ + "ok": False, + "error": "wrong_twitter_account", + "state": last_state, + "using_cdp": is_cdp, + }) + + return _profile_payload({ + "ok": False, + "error": "login_timeout", + "state": last_state, + "using_cdp": is_cdp, + }) + finally: + if not is_cdp: + page.close() + browser.close() + + def _handle_dm_passcode(page): """Handle the DM encryption passcode dialog if it appears. @@ -1531,7 +1808,22 @@ def main(): cmd = sys.argv[1] - if cmd == "reply": + if cmd == "whoami": + result = whoami() + print(json.dumps(result, indent=2)) + + elif cmd == "login": + wait_seconds = 600 + if len(sys.argv) >= 3: + try: + wait_seconds = int(sys.argv[2]) + except ValueError: + print(f"login: wait_seconds must be int, got {sys.argv[2]!r}", file=sys.stderr) + sys.exit(1) + result = login(wait_seconds=wait_seconds) + print(json.dumps(result, indent=2)) + + elif cmd == "reply": if len(sys.argv) < 4: print( "Usage: twitter_browser.py reply ",