Gabriel Vidal
30a9d98c36
POST /accounts/{id}/login drives `claude auth login --claudeai` without a
TTY (URL scraped, code#state piped to stdin, #state checked first);
/login/code, DELETE /login and /logout complete the flow. Launch failures
are read from the CLI's JSON result line and classified (auth, rate_limit,
cli_outdated, …) into {message, kind, hint, account}.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
360 lines
14 KiB
Python
360 lines
14 KiB
Python
"""
|
|
The `claude` CLI's account plumbing, driven headlessly: log in, log out, and
|
|
turn what a failed run printed into a typed error.
|
|
|
|
Login (`claude auth login --claudeai`, CLI 2.1.x). With no TTY the CLI prints
|
|
an OAuth authorize URL whose redirect is Anthropic's *manual code* page
|
|
(platform.claude.com/oauth/code/callback), then blocks on stdin at
|
|
``Paste code here if prompted >``. Signing in on any device shows a
|
|
``<code>#<state>`` string there; written to the CLI's stdin it is exchanged for
|
|
tokens that land in the account's config dir. So the whole flow is:
|
|
|
|
POST /accounts/{id}/login → spawn the CLI, scrape the URL, return it
|
|
POST /accounts/{id}/login/code → pipe the pasted code, wait for the verdict
|
|
DELETE /accounts/{id}/login → give up (kill the CLI)
|
|
|
|
Observed CLI behaviour the code below leans on:
|
|
* a code without ``#state`` → ``Invalid code. Please make sure the full code
|
|
was copied.`` and the CLI prompts again (still alive — the user can retry);
|
|
* a wrong/expired code → exit 1, ``Login failed: Request failed with status
|
|
code 400``;
|
|
* success → exit 0, credentials written.
|
|
The CLI also opens a localhost callback listener; it's useless here (the
|
|
browser is on another device) and dies with the process.
|
|
|
|
A pending login lives only in this process: a sidecar restart drops it (the
|
|
CLI child is killed with it — unlike runs, it's not started as a daemon).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
|
|
# How long a started login waits for its code before the CLI is killed. The
|
|
# OAuth state inside the URL doesn't outlive the process anyway.
|
|
LOGIN_TTL_S = float(os.environ.get("SIDECAR_LOGIN_TTL_S", "600"))
|
|
# How long to wait for the CLI to print its authorize URL.
|
|
URL_WAIT_S = 20.0
|
|
# How long to wait for the token exchange after the code is written.
|
|
EXCHANGE_WAIT_S = 30.0
|
|
|
|
_ANSI = re.compile(r"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)|\x1b\[[0-9;?]*[A-Za-z]")
|
|
_URL = re.compile(r"https://\S+/oauth/authorize\?\S+")
|
|
# What the manual-code page shows: `<code>#<state>`, both base64url-ish.
|
|
_CODE = re.compile(r"^([A-Za-z0-9._~-]{8,})#([A-Za-z0-9._~-]{8,})$")
|
|
|
|
|
|
# ---- error classification --------------------------------------------------
|
|
# One vocabulary for "why did claude fail", shared by launch errors (the CLI's
|
|
# result line in a run log) and the transcript's API-error turns (the viewer
|
|
# maps the transcript's own `error` field the same way).
|
|
|
|
KINDS = ("auth", "rate_limit", "cli_outdated", "overloaded", "network",
|
|
"unknown")
|
|
|
|
_KIND_PATTERNS: list[tuple[str, re.Pattern]] = [
|
|
("auth", re.compile(
|
|
r"not logged in|please run /login|failed to authenticate|oauth"
|
|
r".*(expired|revoked|refresh)|invalid api key|authentication_failed"
|
|
r"|\b401\b", re.I)),
|
|
("rate_limit", re.compile(
|
|
r"session limit|usage limit|rate.?limit|weekly limit|\b429\b", re.I)),
|
|
("cli_outdated", re.compile(
|
|
r"does not support this model|newer is required|claude update", re.I)),
|
|
("overloaded", re.compile(r"overloaded|\b529\b|\b5\d\d\b", re.I)),
|
|
("network", re.compile(
|
|
r"connection (lost|error|refused)|ECONNRESET|ETIMEDOUT|ENOTFOUND"
|
|
r"|fetch failed", re.I)),
|
|
]
|
|
|
|
# The transcript's `error` field (CLI's own names) → our kind.
|
|
_API_KINDS = {"authentication_failed": "auth", "rate_limit": "rate_limit",
|
|
"server_error": "overloaded", "overloaded": "overloaded"}
|
|
|
|
HINTS = {
|
|
"auth": "The account's Claude login is missing or expired — log in again "
|
|
"from Settings → Claude accounts.",
|
|
"rate_limit": "The subscription's usage limit is reached — wait for the "
|
|
"reset or run on the other account.",
|
|
"cli_outdated": "The claude CLI on the host is too old for this model — "
|
|
"run `claude update`.",
|
|
"overloaded": "Anthropic's API is overloaded or erroring — retry shortly.",
|
|
"network": "The connection to Anthropic's API dropped — retry.",
|
|
}
|
|
|
|
|
|
def classify(text: str | None, api_error: str | None = None) -> str:
|
|
"""The error kind for a failure message (and the CLI's own error name)."""
|
|
if api_error in _API_KINDS:
|
|
return _API_KINDS[api_error]
|
|
for kind, rx in _KIND_PATTERNS:
|
|
if rx.search(text or ""):
|
|
return kind
|
|
return "unknown"
|
|
|
|
|
|
def error_detail(message: str, *, kind: str | None = None,
|
|
account: str | None = None, **extra) -> dict:
|
|
"""An HTTPException detail every caller can read the same way:
|
|
``{message, kind, hint?, account?}``."""
|
|
kind = kind or classify(message)
|
|
out = {"message": message, "kind": kind}
|
|
if HINTS.get(kind):
|
|
out["hint"] = HINTS[kind]
|
|
if account:
|
|
out["account"] = account
|
|
out.update({k: v for k, v in extra.items() if v is not None})
|
|
return out
|
|
|
|
|
|
def run_failure(text: str) -> tuple[str, str]:
|
|
"""(message, kind) for what a run printed before dying.
|
|
|
|
``claude -p --output-format json`` ends with a single JSON ``result`` line
|
|
(``{"type":"result","is_error":true,"result":"Not logged in · Please run
|
|
/login",…}``) — dumping that line raw is what the composer used to show.
|
|
Prefer its ``result`` text; fall back to the last plain line."""
|
|
lines = [ln.strip() for ln in (text or "").splitlines() if ln.strip()]
|
|
for ln in reversed(lines):
|
|
if not ln.startswith("{"):
|
|
continue
|
|
try:
|
|
rec = json.loads(ln)
|
|
except ValueError:
|
|
continue
|
|
if isinstance(rec, dict) and rec.get("type") == "result":
|
|
msg = str(rec.get("result") or rec.get("terminal_reason")
|
|
or "claude failed")[:400]
|
|
status = rec.get("api_error_status")
|
|
return msg, classify(f"{msg} {status or ''}")
|
|
msg = (lines[-1][:400] if lines else "(no output)")
|
|
return msg, classify(msg)
|
|
|
|
|
|
# ---- login sessions --------------------------------------------------------
|
|
|
|
class LoginError(Exception):
|
|
"""A login step failed; ``status`` is the HTTP code to answer with."""
|
|
|
|
def __init__(self, status: int, message: str, kind: str = "login"):
|
|
super().__init__(message)
|
|
self.status, self.message, self.kind = status, message, kind
|
|
|
|
|
|
class _Login:
|
|
"""One `claude auth login` child waiting for its code."""
|
|
|
|
def __init__(self, account: str, proc: subprocess.Popen):
|
|
self.id = secrets.token_hex(8)
|
|
self.account = account
|
|
self.proc = proc
|
|
self.started = time.time()
|
|
self.url: str | None = None
|
|
self.state: str | None = None
|
|
self._buf: list[str] = []
|
|
self._lock = threading.Lock()
|
|
threading.Thread(target=self._pump, daemon=True).start()
|
|
self._timer = threading.Timer(LOGIN_TTL_S, self.kill)
|
|
self._timer.daemon = True
|
|
self._timer.start()
|
|
|
|
def _pump(self) -> None:
|
|
# Raw os.read, not readline: the paste prompt has no trailing newline.
|
|
fd = self.proc.stdout.fileno()
|
|
while True:
|
|
try:
|
|
chunk = os.read(fd, 4096)
|
|
except OSError:
|
|
break
|
|
if not chunk:
|
|
break
|
|
with self._lock:
|
|
self._buf.append(chunk.decode(errors="replace"))
|
|
|
|
def output(self) -> str:
|
|
with self._lock:
|
|
return _ANSI.sub("", "".join(self._buf))
|
|
|
|
def wait_url(self, timeout: float) -> str | None:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
m = _URL.search(self.output())
|
|
if m:
|
|
self.url = m.group(0)
|
|
q = urllib.parse.parse_qs(urllib.parse.urlparse(self.url).query)
|
|
self.state = (q.get("state") or [None])[0]
|
|
return self.url
|
|
if self.proc.poll() is not None:
|
|
return None
|
|
time.sleep(0.1)
|
|
return None
|
|
|
|
def alive(self) -> bool:
|
|
return self.proc.poll() is None
|
|
|
|
def kill(self) -> None:
|
|
self._timer.cancel()
|
|
if self.proc.poll() is None:
|
|
try:
|
|
self.proc.kill()
|
|
self.proc.wait(timeout=5)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
def public(self) -> dict:
|
|
return {"loginId": self.id, "account": self.account, "url": self.url,
|
|
"startedAt": int(self.started),
|
|
"expiresAt": int(self.started + LOGIN_TTL_S)}
|
|
|
|
|
|
_logins: dict[str, _Login] = {}
|
|
_logins_lock = threading.Lock()
|
|
|
|
|
|
def normalize_code(raw: str) -> tuple[str, str] | None:
|
|
"""(code, state) from what the user pasted — the ``code#state`` string the
|
|
manual-code page shows, or the whole callback URL (``…?code=…&state=…``)
|
|
if they copied the address bar instead. None when it's neither."""
|
|
s = "".join((raw or "").split())
|
|
m = _CODE.match(s)
|
|
if m:
|
|
return m.group(1), m.group(2)
|
|
if s.startswith("http"):
|
|
q = urllib.parse.parse_qs(urllib.parse.urlparse(s).query)
|
|
code, state = (q.get("code") or [""])[0], (q.get("state") or [""])[0]
|
|
if code and state and _CODE.match(f"{code}#{state}"):
|
|
return code, state
|
|
return None
|
|
|
|
|
|
def start_login(account: str, cmd: list[str], env: dict[str, str],
|
|
email: str | None = None) -> dict:
|
|
"""Start (or restart) the login for ``account`` and return its URL.
|
|
|
|
One login per account: a second start kills the first — its OAuth state
|
|
would no longer match anything the user can paste."""
|
|
with _logins_lock:
|
|
old = _logins.pop(account, None)
|
|
if old:
|
|
old.kill()
|
|
args = [*cmd, "auth", "login", "--claudeai"]
|
|
if email:
|
|
args += ["--email", email]
|
|
# BROWSER=true: the CLI "opens" a browser via a no-op instead of trying
|
|
# xdg-open on a headless box; the URL still gets printed.
|
|
env = {**env, "BROWSER": "true", "NO_COLOR": "1"}
|
|
cfg = env.get("CLAUDE_CONFIG_DIR")
|
|
if cfg:
|
|
os.makedirs(cfg, exist_ok=True)
|
|
try:
|
|
proc = subprocess.Popen(args, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, env=env)
|
|
except OSError as e:
|
|
raise LoginError(500, f"could not start `claude auth login`: {e}")
|
|
login = _Login(account, proc)
|
|
if not login.wait_url(URL_WAIT_S):
|
|
out = login.output().strip().splitlines()
|
|
login.kill()
|
|
raise LoginError(502, "claude auth login printed no sign-in URL: "
|
|
+ (out[-1][:300] if out else "(no output)"))
|
|
with _logins_lock:
|
|
_logins[account] = login
|
|
return login.public()
|
|
|
|
|
|
def pending_login(account: str) -> dict | None:
|
|
with _logins_lock:
|
|
login = _logins.get(account)
|
|
if login and login.alive():
|
|
return login.public()
|
|
return None
|
|
|
|
|
|
def cancel_login(account: str) -> bool:
|
|
with _logins_lock:
|
|
login = _logins.pop(account, None)
|
|
if not login:
|
|
return False
|
|
login.kill()
|
|
return True
|
|
|
|
|
|
def submit_code(account: str, raw: str) -> None:
|
|
"""Hand the pasted code to the waiting CLI; returns once it's logged in.
|
|
|
|
Raises LoginError: 404 no pending login, 400 malformed/mismatched code (the
|
|
login stays pending — paste again), 502 the exchange failed (the login is
|
|
over — start again)."""
|
|
with _logins_lock:
|
|
login = _logins.get(account)
|
|
if not login or not login.alive():
|
|
with _logins_lock:
|
|
_logins.pop(account, None)
|
|
raise LoginError(404, "no login in progress for this account (it may "
|
|
"have expired) — start again", "login_expired")
|
|
parsed = normalize_code(raw)
|
|
if not parsed:
|
|
raise LoginError(400, "that doesn't look like a sign-in code — copy the "
|
|
"whole `code#state` string the page shows",
|
|
"bad_code")
|
|
code, state = parsed
|
|
if login.state and state != login.state:
|
|
raise LoginError(400, "this code belongs to a different sign-in "
|
|
"attempt — use the link from this one", "bad_code")
|
|
mark = len(login.output())
|
|
try:
|
|
login.proc.stdin.write(f"{code}#{state}\n".encode())
|
|
login.proc.stdin.flush()
|
|
except OSError as e:
|
|
raise LoginError(502, f"the login process went away: {e}")
|
|
|
|
deadline = time.monotonic() + EXCHANGE_WAIT_S
|
|
while time.monotonic() < deadline:
|
|
rc = login.proc.poll()
|
|
tail = login.output()[mark:]
|
|
if rc is not None:
|
|
time.sleep(0.2) # let the pump drain the last bytes
|
|
tail = login.output()[mark:]
|
|
with _logins_lock:
|
|
_logins.pop(account, None)
|
|
login.kill()
|
|
if rc == 0:
|
|
return
|
|
raise LoginError(502, _last_line(tail) or f"login failed (exit {rc})",
|
|
"login_failed")
|
|
if "Invalid code" in tail:
|
|
raise LoginError(400, _last_line(tail.split("Paste code")[0]),
|
|
"bad_code")
|
|
time.sleep(0.1)
|
|
raise LoginError(504, "the CLI didn't confirm the login in time — check "
|
|
"the account's status and retry", "login_timeout")
|
|
|
|
|
|
def _last_line(text: str) -> str:
|
|
lines = [ln.strip(" >") for ln in (text or "").splitlines()
|
|
if ln.strip(" >")]
|
|
return lines[-1][:300] if lines else ""
|
|
|
|
|
|
def logout(cmd: list[str], env: dict[str, str]) -> str:
|
|
"""`claude auth logout` under the account's env; its message on success."""
|
|
try:
|
|
r = subprocess.run([*cmd, "auth", "logout"], capture_output=True,
|
|
text=True, timeout=30, stdin=subprocess.DEVNULL,
|
|
env=env)
|
|
except (OSError, subprocess.TimeoutExpired) as e:
|
|
raise LoginError(502, f"claude auth logout failed: {e}")
|
|
out = _ANSI.sub("", (r.stdout or "") + (r.stderr or "")).strip()
|
|
if r.returncode != 0:
|
|
raise LoginError(502, _last_line(out) or
|
|
f"claude auth logout exited {r.returncode}")
|
|
return _last_line(out) or "logged out"
|