Gabriel Vidal
cf01bc00f4
config / pids / validate / claude_accounts / launch / fork / routes_runs /
routes_accounts, mounted by a thin sidecar.py (same `sidecar:app`, same 24
routes, same startup hook). Mutable state keeps one owner module
(pids._procs, claude_accounts._account_status_cache). Includes the
_claude_args builder from 1de9426 and its test; the Dockerfile's explicit
COPY list gains the new modules.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
112 lines
4.4 KiB
Python
112 lines
4.4 KiB
Python
"""
|
|
``/health`` and the account routes: ``/accounts`` (login health) and the
|
|
headless login / logout flow under ``/accounts/{account}/…``.
|
|
"""
|
|
|
|
import re
|
|
|
|
from fastapi import APIRouter, Header, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
import claude_accounts
|
|
import claude_cli
|
|
import pairing
|
|
from claude_accounts import (_account_env, _auth_status, _login_account,
|
|
_login_error)
|
|
from config import (ACCOUNTS, CLAUDE_BIN, DEFAULT_ACCOUNT, DEFAULT_CWD,
|
|
DEFAULT_MODEL, PERMISSION_MODE, TOKEN)
|
|
from validate import _auth
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/health")
|
|
def health() -> dict:
|
|
"""Open (no bearer): the deploy probe and the hub's worker poller read it.
|
|
The worker fields say who this runner is — ``paired`` is whether a hub
|
|
holds its token (SIDECAR_TOKEN counts, so the host sidecar reads paired)."""
|
|
return {"ok": True, "cwd": DEFAULT_CWD, "claude": CLAUDE_BIN,
|
|
"defaultModel": DEFAULT_MODEL, "accounts": list(ACCOUNTS),
|
|
"defaultAccount": DEFAULT_ACCOUNT,
|
|
"permissionMode": PERMISSION_MODE,
|
|
**pairing.identity(), "paired": bool(TOKEN or pairing.token())}
|
|
|
|
|
|
@router.get("/accounts")
|
|
def accounts(authorization: str | None = Header(default=None)) -> dict:
|
|
"""Each account's login health (who it is, which org, which plan)."""
|
|
_auth(authorization)
|
|
return {"accounts": [{**_auth_status(a),
|
|
"pendingLogin": claude_cli.pending_login(a)}
|
|
for a in ACCOUNTS],
|
|
"default": DEFAULT_ACCOUNT}
|
|
|
|
|
|
# ---- login / logout ---------------------------------------------------------
|
|
# `claude auth login` driven headlessly — see claude_cli.py for the flow. Each
|
|
# account logs in under its own env (_account_env), so the credentials land in
|
|
# that account's config dir; the default account writes the host user's own
|
|
# ~/.claude login.
|
|
|
|
class LoginBody(BaseModel):
|
|
email: str | None = None # pre-fills the sign-in page (--email)
|
|
|
|
|
|
class LoginCodeBody(BaseModel):
|
|
code: str # the `code#state` the sign-in page shows
|
|
|
|
|
|
@router.post("/accounts/{account}/login")
|
|
def login_start(account: str, body: LoginBody | None = None,
|
|
authorization: str | None = Header(default=None)) -> dict:
|
|
"""Start a subscription login: returns the sign-in URL to open (on any
|
|
device); the code it ends with goes to ``/login/code``."""
|
|
_auth(authorization)
|
|
account = _login_account(account)
|
|
email = ((body.email if body else None) or "").strip() or None
|
|
if email and not re.fullmatch(r"[^@\s]+@[^@\s]+", email):
|
|
raise HTTPException(400, "invalid email")
|
|
try:
|
|
return claude_cli.start_login(account, [CLAUDE_BIN],
|
|
_account_env(account), email)
|
|
except claude_cli.LoginError as e:
|
|
raise _login_error(e, account)
|
|
|
|
|
|
@router.post("/accounts/{account}/login/code")
|
|
def login_code(account: str, body: LoginCodeBody,
|
|
authorization: str | None = Header(default=None)) -> dict:
|
|
"""Finish the pending login with the pasted code; answers the fresh
|
|
``auth status`` of the account."""
|
|
_auth(authorization)
|
|
account = _login_account(account)
|
|
try:
|
|
claude_cli.submit_code(account, body.code)
|
|
except claude_cli.LoginError as e:
|
|
raise _login_error(e, account)
|
|
claude_accounts._account_status_cache.pop(account, None)
|
|
return _auth_status(account)
|
|
|
|
|
|
@router.delete("/accounts/{account}/login")
|
|
def login_cancel(account: str,
|
|
authorization: str | None = Header(default=None)) -> dict:
|
|
_auth(authorization)
|
|
return {"cancelled": claude_cli.cancel_login(_login_account(account))}
|
|
|
|
|
|
@router.post("/accounts/{account}/logout")
|
|
def logout(account: str,
|
|
authorization: str | None = Header(default=None)) -> dict:
|
|
"""`claude auth logout` for the account. New launches on it fail with a
|
|
typed ``auth`` error until it logs back in; runs already in flight may
|
|
keep going on the access token they hold in memory."""
|
|
_auth(authorization)
|
|
account = _login_account(account)
|
|
claude_cli.cancel_login(account)
|
|
try:
|
|
msg = claude_cli.logout([CLAUDE_BIN], _account_env(account))
|
|
except claude_cli.LoginError as e:
|
|
raise _login_error(e, account)
|
|
claude_accounts._account_status_cache.pop(account, None)
|
|
return {**_auth_status(account), "message": msg}
|