mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 06:16:24 +00:00
fix(login): confirm via credentials file + validate the OAuth URL
Testing showed the wide-window fix captured the full URL, but login still
failed at submit ("login not confirmed"): the old check snapshotted the pane
once after 3s and only matched a few success strings, so an in-progress or
differently-worded exchange read as failure. Two hardening changes:
- login_submit now polls (up to 40s) and confirms by the authoritative signal —
claude's credentials file changing on disk (any of the known locations /
~/.claude/*credential*) — with success-text and clean-exit as fallbacks.
- login_start captures with escape sequences (-e) and accepts only a *complete*
OAuth URL (https:// + client_id + redirect_uri + state). This recovers the
real href when claude renders the link as an OSC-8 hyperlink (whose visible
text can be garbled, e.g. the "ttps://claude.com/cai/..." seen in testing) and
refuses partial/garbled captures. On timeout the error now includes the actual
last screen so a wrong menu/onboarding state is diagnosable.
tmux.capture_pane gains an `escapes` flag. Tests cover URL completeness,
OSC-8 href recovery, and credentials-file confirmation. Suite green (199).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YKVyBmKvWDVgrFC9WER2f2
This commit is contained in:
+118
-31
@@ -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;;<URL>\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": <pane tail>}``. 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": <pane tail>}`` 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:
|
||||
|
||||
@@ -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
|
||||
|
||||
+54
-20
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user