diff --git a/src/handler/control/login.py b/src/handler/control/login.py index 8e026c2..008dde3 100644 --- a/src/handler/control/login.py +++ b/src/handler/control/login.py @@ -9,7 +9,8 @@ handoff: the pane for the ``claude.com`` / ``claude.ai`` authorization URL. The URL is returned to the UI (which opens it in an iframe) and the tmux session is *left alive*. 2. ``login_submit`` sends the authorization code the operator pastes back into that same - still-alive session, waits for claude to exchange it, and reports success. + still-alive session, then confirms the login by watching for claude to write its + credentials file (with a success-text fallback). Everything shells out through the :mod:`~handler.control.tmux` seam, so the whole flow is unit-testable with a fake tmux and never needs a real ``claude`` binary — the same pattern @@ -22,6 +23,7 @@ onboarding (theme/trust prompts) before the ``/login`` menu, bump ``boot_wait``. from __future__ import annotations +import glob import os import re import time @@ -38,18 +40,23 @@ LOGIN_SESSION = "handler__login" LOGIN_COLS = 500 LOGIN_ROWS = 50 -# Any http(s) URL in the pane; we then prefer the OAuth/authorize link among them. The -# character class stops at whitespace, quotes, and — importantly — box-drawing glyphs -# (U+2500–U+257F) the TUI may render flush against the link, so a bordered URL isn't -# captured with a trailing "│". -_URL_RE = re.compile(r"https?://[^\s\"'<>`|─-╿]+") -_OAUTH_HINTS = ("oauth", "authorize", "claude.ai", "claude.com", "console.anthropic") +# Strip ANSI CSI + OSC escape sequences so success-text matching sees plain text. +_ANSI_RE = re.compile( + r"\x1b\[[0-9;?]*[ -/]*[@-~]" # CSI (colors, cursor moves) + r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC (…terminated by BEL or ST) + r"|\x1b[@-Z\\-_]" # two-char escapes +) +# An http(s) URL. The class excludes whitespace, quotes, box-drawing glyphs the TUI may +# render flush against the link, *and* control/escape bytes — so a URL sitting inside an +# OSC-8 hyperlink escape (``\x1b]8;;\x1b\\``) is recovered cleanly, cut at the ESC. +_URL_RE = re.compile(r"https?://[^\s\"'<>`|\x00-\x1f─-╿]+") _SUCCESS_HINTS = ( "login successful", "logged in", "successfully authenticated", "authentication successful", "you are now logged in", + "welcome back", ) @@ -66,28 +73,76 @@ def _sleep(seconds: float) -> None: time.sleep(seconds) +def _strip_ansi(text: str) -> str: + return _ANSI_RE.sub("", text or "") + + +def _is_complete_oauth_url(url: str) -> bool: + """A *usable* Claude OAuth URL, not a partial/garbled capture. + + Requiring the scheme + the OAuth query markers rejects a mid-render capture like + ``ttps://claude.com/cai/oauth?…`` (dropped scheme chars) or a URL cut before its + query string — handing either to the iframe would send the operator to a broken page. + """ + low = url.lower() + return ( + low.startswith("https://") + and "oauth" in low + and "client_id=" in low + and "redirect_uri=" in low + and "state=" in low + ) + + def _extract_url(pane: str) -> str | None: - """Pull the login URL out of a captured pane, preferring the OAuth link.""" + """Return the first *complete* OAuth URL found in a captured pane, else ``None``.""" if not pane: return None - candidates = [c.rstrip(".,);]") for c in _URL_RE.findall(pane)] - for c in candidates: - if any(hint in c.lower() for hint in _OAUTH_HINTS): - return c - return candidates[0] if candidates else None + for raw in _URL_RE.findall(pane): + candidate = raw.rstrip(".,);]}>") + if _is_complete_oauth_url(candidate): + return candidate + return None + + +def _credentials_fingerprint() -> tuple: + """A fingerprint of claude's on-disk credentials, to detect a login writing them. + + Claude Code stores its OAuth credentials under the user's home; the exact filename has + drifted across versions, so we watch every likely location and any ``*credential*`` + file under ``~/.claude``. The fingerprint is ``(path, mtime, size)`` tuples — it + changes when a login creates or rewrites the credentials, which is a far more reliable + "did it work" signal than scraping the TUI for a success string. + """ + home = _home() + paths = { + os.path.join(home, ".claude", ".credentials.json"), + os.path.join(home, ".claude", "credentials.json"), + os.path.join(home, ".claude.json"), + os.path.join(home, ".config", "claude", "credentials.json"), + } + paths.update(glob.glob(os.path.join(home, ".claude", "*credential*"))) + fp = [] + for p in sorted(paths): + try: + st = os.stat(p) + fp.append((p, st.st_mtime_ns, st.st_size)) + except OSError: + continue + return tuple(fp) def start( *, - boot_wait: float = 4.0, - menu_wait: float = 1.5, - url_timeout: float = 30.0, + boot_wait: float = 6.0, + menu_wait: float = 2.0, + url_timeout: float = 45.0, poll_interval: float = 0.5, ) -> dict: """Open ``claude`` in tmux, drive ``/login`` to the subscription account, return the URL. Leaves the tmux session alive for :func:`submit_code`. Raises :class:`LoginError` if - no authorization URL appears within ``url_timeout`` seconds. + no complete authorization URL appears within ``url_timeout`` seconds. """ claude = get_settings().claude_bin # A stale session from a previous, abandoned attempt would swallow our keystrokes. @@ -97,7 +152,7 @@ def start( tmux.new_session( LOGIN_SESSION, cwd=_home(), command=claude, env={}, width=LOGIN_COLS, height=LOGIN_ROWS ) - _sleep(boot_wait) # let claude boot to its prompt + _sleep(boot_wait) # let claude finish its splash/boot and reach a prompt tmux.send_keys(LOGIN_SESSION, "/login") _sleep(menu_wait) @@ -108,26 +163,43 @@ def start( deadline = time.monotonic() + url_timeout url: str | None = None + last_pane = "" while url is None and time.monotonic() < deadline: - url = _extract_url(tmux.capture_pane(LOGIN_SESSION)) + # Capture with escapes so an OSC-8 hyperlink href is recoverable; require a + # *complete* URL so a still-rendering pane keeps us polling instead of returning + # a garbled fragment. + last_pane = tmux.capture_pane(LOGIN_SESSION, escapes=True) + url = _extract_url(last_pane) if url is None: _sleep(poll_interval) if url is None: - # Don't leave a half-driven session lying around on failure. + # Don't leave a half-driven session lying around on failure. Surface what claude + # actually rendered so a wrong menu/onboarding state is diagnosable, not opaque. + tail = _tail(_strip_ansi(last_pane)) if tmux.has_session(LOGIN_SESSION): tmux.kill_session(LOGIN_SESSION) - raise LoginError( - "timed out waiting for the claude login URL — is the 'claude' binary installed " - "in the control container and does '/login' open the subscription flow?" + message = ( + "timed out waiting for a complete claude login URL — is the 'claude' binary " + "installed in the control container and does '/login' open the subscription flow?" ) + if tail: + message += f" Last screen:\n{tail}" + raise LoginError(message) return {"session": LOGIN_SESSION, "url": url} -def submit_code(code: str, *, settle_wait: float = 3.0) -> dict: - """Feed the pasted authorization ``code`` into the live login session. +def submit_code( + code: str, + *, + poll_timeout: float = 40.0, + poll_interval: float = 1.0, +) -> dict: + """Feed the pasted authorization ``code`` into the live login session and confirm. - Returns ``{"success": bool, "output": }``. Kills the session on success. - Raises :class:`LoginError` if there is no active login session to submit to. + Confirms by polling (up to ``poll_timeout`` seconds) for any of: claude's credentials + file changing on disk (the authoritative signal), a success line in the pane, or the + session exiting cleanly. Returns ``{"success": bool, "output": }`` and kills + the session on success. Raises :class:`LoginError` if there is no session to submit to. """ code = (code or "").strip() if not code: @@ -135,14 +207,29 @@ def submit_code(code: str, *, settle_wait: float = 3.0) -> dict: if not tmux.has_session(LOGIN_SESSION): raise LoginError("no active claude login session — start the login flow again") + baseline = _credentials_fingerprint() tmux.send_keys(LOGIN_SESSION, code) - _sleep(settle_wait) - pane = tmux.capture_pane(LOGIN_SESSION) - success = _looks_successful(pane) + deadline = time.monotonic() + poll_timeout + success = False + pane = "" + while time.monotonic() < deadline: + _sleep(poll_interval) + pane = tmux.capture_pane(LOGIN_SESSION, escapes=True) + if _credentials_fingerprint() != baseline: + success = True + break + if _looks_successful(_strip_ansi(pane)): + success = True + break + if not tmux.has_session(LOGIN_SESSION): + # claude exited on its own after a successful login. + success = True + break + if success and tmux.has_session(LOGIN_SESSION): tmux.kill_session(LOGIN_SESSION) - return {"success": success, "output": _tail(pane)} + return {"success": success, "output": _tail(_strip_ansi(pane))} def _looks_successful(pane: str) -> bool: diff --git a/src/handler/control/tmux.py b/src/handler/control/tmux.py index 5a5f1da..f308c11 100644 --- a/src/handler/control/tmux.py +++ b/src/handler/control/tmux.py @@ -80,19 +80,21 @@ def send_keys(name: str, keys: str) -> None: subprocess.run([tmux, "send-keys", "-t", name, keys, "Enter"], check=True) -def capture_pane(name: str) -> str: +def capture_pane(name: str, escapes: bool = False) -> str: """Return the visible text of a session's pane. ``-p`` prints to stdout, ``-J`` joins wrapped lines so a long URL split across the pane width comes back on one logical line (the login flow relies on this to recover - the claude.com authorization link). Returns an empty string if the session is gone. + the claude.com authorization link). ``escapes=True`` adds ``-e`` to keep ANSI/OSC + escape sequences — the login URL extractor uses this so it can also recover a URL that + the TUI renders as an OSC-8 hyperlink (where the visible text differs from the href). + Returns an empty string if the session is gone. """ tmux = get_settings().tmux_bin - result = subprocess.run( - [tmux, "capture-pane", "-t", name, "-p", "-J"], - capture_output=True, - text=True, - ) + argv = [tmux, "capture-pane", "-t", name, "-p", "-J"] + if escapes: + argv.append("-e") + result = subprocess.run(argv, capture_output=True, text=True) if result.returncode != 0: return "" return result.stdout diff --git a/tests/test_control_login.py b/tests/test_control_login.py index 4fb2656..dae0f38 100644 --- a/tests/test_control_login.py +++ b/tests/test_control_login.py @@ -1,8 +1,9 @@ -"""The claude web-login seam: driving ``claude /login`` through tmux and scraping the URL. +"""The claude web-login seam: driving ``claude /login`` through tmux, scraping the URL, +and confirming the login. Uses the shared ``fake_tmux`` fixture (extended here with a scripted ``capture_pane``) and -patches out the real sleeps, so no live claude/tmux is touched — the same approach as the -spawn tests. +patches out the real sleeps + the on-disk credentials check, so no live claude/tmux/FS is +touched — the same approach as the spawn tests. """ from __future__ import annotations @@ -17,20 +18,32 @@ def no_sleep(monkeypatch): monkeypatch.setattr(login, "_sleep", lambda *_a, **_k: None) +@pytest.fixture +def stable_creds(monkeypatch): + """No credentials change on disk — success must come from the pane/session signals.""" + monkeypatch.setattr(login, "_credentials_fingerprint", lambda: ()) + + def _pane(monkeypatch, *frames): """Make ``capture_pane`` return each frame in turn, then repeat the last one.""" seq = list(frames) - def capture(_name): + def capture(_name, escapes=False): return seq[0] if len(seq) == 1 else seq.pop(0) monkeypatch.setattr(tmux, "capture_pane", capture) -AUTH_URL = "https://claude.ai/oauth/authorize?code=true&client_id=abc&state=xyz" +# A complete Claude OAuth URL (scheme + client_id + redirect_uri + state) — extraction +# deliberately rejects anything less, so the fixtures must use the real shape. +AUTH_URL = ( + "https://claude.ai/oauth/authorize?code=true&client_id=abc123&response_type=code" + "&redirect_uri=https%3A%2F%2Fplatform.claude.com%2Foauth%2Fcode%2Fcallback" + "&scope=user%3Aprofile&code_challenge=chal&code_challenge_method=S256&state=st42" +) -def test_extract_url_prefers_oauth_link(): +def test_extract_url_prefers_complete_oauth_link(): pane = f"Visit https://example.com/help or\n{AUTH_URL}\nand paste the code." assert login._extract_url(pane) == AUTH_URL @@ -43,22 +56,24 @@ def test_extract_url_none_when_no_link(): assert login._extract_url("no link here") is None +def test_extract_url_rejects_incomplete_url(): + # A garbled/partial capture (dropped scheme char, or no query string) must be refused + # so the iframe never opens a broken page. + assert login._extract_url("ttps://claude.com/cai/oauth/authorize?client_id=x") is None + assert login._extract_url("https://claude.ai/oauth/authorize") is None + + def test_extract_url_stops_at_box_border(): # claude may draw the URL inside a rounded box; a "│" flush against the link must not # be captured as part of the URL. assert login._extract_url(f"│{AUTH_URL}│") == AUTH_URL -def test_extract_url_captures_full_long_url_with_redirect_uri(): - # The real login URL carries redirect_uri + PKCE; on a wide pane it arrives intact and - # extraction must not clip it (the truncation-at-80-cols bug was in capture, not here). - long_url = ( - "https://claude.ai/oauth/authorize?code=true&client_id=9d1c250a-e61b-44d9-88ab-" - "0123456789ab&response_type=code&redirect_uri=https%3A%2F%2Fconsole.anthropic.com" - "%2Foauth%2Fcode%2Fcallback&scope=org%3Acreate_api_key+user%3Aprofile&" - "code_challenge=abcDEF123&code_challenge_method=S256&state=xyz789" - ) - assert login._extract_url(f"Use this URL to sign in:\n{long_url}") == long_url +def test_extract_url_recovers_href_from_osc8_hyperlink(): + # claude renders the URL as an OSC-8 hyperlink: the visible text can be styled/garbled + # while the real href sits in the escape. Capturing with escapes lets us recover it. + pane = f"\x1b]8;;{AUTH_URL}\x1b\\click here\x1b]8;;\x1b\\" + assert login._extract_url(pane) == AUTH_URL def test_start_launches_claude_selects_subscription_and_returns_url( @@ -103,11 +118,11 @@ def test_start_times_out_and_cleans_up_when_no_url(env, fake_tmux, no_sleep, mon assert login.LOGIN_SESSION not in fake_tmux["live"] -def test_submit_code_sends_code_and_reports_success(env, fake_tmux, no_sleep, monkeypatch): +def test_submit_code_confirmed_by_success_text(env, fake_tmux, no_sleep, stable_creds, monkeypatch): fake_tmux["live"].add(login.LOGIN_SESSION) _pane(monkeypatch, "Login successful. Welcome back!") - result = login.submit_code("my-auth-code") + result = login.submit_code("my-auth-code", poll_timeout=1.0) assert result["success"] is True assert "Login successful" in result["output"] @@ -116,13 +131,32 @@ def test_submit_code_sends_code_and_reports_success(env, fake_tmux, no_sleep, mo assert login.LOGIN_SESSION not in fake_tmux["live"] +def test_submit_code_confirmed_by_credentials_file(env, fake_tmux, no_sleep, monkeypatch): + fake_tmux["live"].add(login.LOGIN_SESSION) + # The pane never prints a success string, but claude writes its credentials — the + # authoritative signal. First call = baseline, later calls = changed. + calls = {"n": 0} + + def fingerprint(): + calls["n"] += 1 + return () if calls["n"] == 1 else (("~/.claude/.credentials.json", 123, 45),) + + monkeypatch.setattr(login, "_credentials_fingerprint", fingerprint) + _pane(monkeypatch, "still on the paste-code screen, no success text") + + result = login.submit_code("code", poll_timeout=1.0) + + assert result["success"] is True + assert login.LOGIN_SESSION not in fake_tmux["live"] + + def test_submit_code_reports_failure_without_killing_session( - env, fake_tmux, no_sleep, monkeypatch + env, fake_tmux, no_sleep, stable_creds, monkeypatch ): fake_tmux["live"].add(login.LOGIN_SESSION) _pane(monkeypatch, "Invalid code, please try again") - result = login.submit_code("wrong") + result = login.submit_code("wrong", poll_timeout=0.05) assert result["success"] is False assert login.LOGIN_SESSION in fake_tmux["live"] # left up for a retry