from __future__ import annotations import queue import fcntl import json import os import subprocess import tempfile from collections.abc import Callable, Iterator from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path import threading import time from account_registry import Account, AccountRef, AccountRegistry, StagedAccount from provider_services import LifecycleEvent STATUS_TIMEOUT_SECS = 5.0 LOGIN_TIMEOUT_SECS = 900.0 CODE_ENTRY_FALLBACK_SECS = 2.0 @dataclass(frozen=True) class CommandResult: returncode: int stdout: str = "" stderr: str = "" @dataclass(frozen=True) class VerifiedIdentity: email: str org_id: str subscription_type: str | None @dataclass(frozen=True) class ClaudeLoginPrompt: url: str raw_text: str @dataclass(frozen=True) class ClaudeLoginSession: process: subprocess.Popen account_home: Path @dataclass(frozen=True) class LoginEvent: kind: str url: str | None = None raw_text: str | None = None session: ClaudeLoginSession | None = None Runner = Callable[[list[str], Path, float | None], CommandResult] class ClaudeAuthOperationError(RuntimeError): pass class ClaudeAuthOperation: def __init__( self, registry: AccountRegistry, runner: Callable[..., CommandResult] | None = None, login_popen: Callable[..., subprocess.Popen] | None = None, ) -> None: self.registry = registry self._runner = runner or self._run_command self._login_popen = login_popen or subprocess.Popen def add( self, alias: str, *, login_hint: str | None = None, slug: str | None = None, ) -> Iterator[LifecycleEvent]: staged: StagedAccount | None = None pending_account: Account | None = None try: staged = self.registry.stage_add(alias, slug=slug) pending_account = self._pending_account(staged) except Exception as exc: yield LifecycleEvent( kind="failure", message=self._failure_message(f"Adding account {alias} failed", exc), error=exc, ) return yield LifecycleEvent(kind="started", account=pending_account) yield LifecycleEvent(kind="browser-waiting", account=pending_account) try: login_url = "" for event in self._login(staged.account_home, login_hint=login_hint): if event.kind == "url_ready": if event.url is not None: login_url = event.url continue if event.kind == "code_required": prompt = ClaudeLoginPrompt( url=login_url, raw_text=event.raw_text or "", ) yield LifecycleEvent( kind="code_required", account=pending_account, prompt=prompt, session=event.session, ) continue verified = self._verify_login(staged.account_home) collision = self._find_collision(verified, exclude_slug=None) if collision is not None: staged.rollback() yield LifecycleEvent(kind="rollback", account=pending_account) yield LifecycleEvent( kind="collision", account=pending_account, collision=collision, message=f"Adding account {alias} collided with existing account {collision.alias}", ) return self._write_identity_file(staged.account_home / "account_identity.json", verified) committed = staged.commit() yield LifecycleEvent(kind="success", account=self._with_verified_identity(committed, verified)) except Exception as exc: if staged is not None and not staged._closed: staged.rollback() yield LifecycleEvent(kind="rollback", account=pending_account) yield LifecycleEvent( kind="failure", account=pending_account, message=self._failure_message(f"Adding account {alias} failed", exc), error=exc, ) def reauthenticate(self, account: Account) -> Iterator[LifecycleEvent]: yield LifecycleEvent(kind="started", account=account) try: with self._account_lock(account.account_home): snapshots = self._snapshot_files(account.account_home) baseline = self._read_persisted_identity(account.account_home) yield LifecycleEvent(kind="browser-waiting", account=account) try: login_url = "" for event in self._login(account.account_home): if event.kind == "url_ready": if event.url is not None: login_url = event.url continue if event.kind == "code_required": prompt = ClaudeLoginPrompt( url=login_url, raw_text=event.raw_text or "", ) yield LifecycleEvent( kind="code_required", account=account, prompt=prompt, session=event.session, ) continue verified = self._verify_login(account.account_home) collision = self._find_collision(verified, exclude_slug=account.slug) if collision is not None: self._restore_files(account.account_home, snapshots) yield LifecycleEvent(kind="rollback", account=account) yield LifecycleEvent( kind="collision", account=account, collision=collision, message=( f"Re-authentication for {account.alias} collided with existing account " f"{collision.alias}" ), ) return if baseline is not None and self._identity_key(baseline) != self._identity_key(verified): self._restore_files(account.account_home, snapshots) yield LifecycleEvent(kind="rollback", account=account) yield LifecycleEvent( kind="failure", account=account, message=f"Re-authentication failed for {account.alias}", error=ClaudeAuthOperationError("verified identity changed"), ) return self._write_identity_file(account.account_home / "account_identity.json", verified) except Exception as exc: self._restore_files(account.account_home, snapshots) yield LifecycleEvent(kind="rollback", account=account) yield LifecycleEvent( kind="failure", account=account, message=self._failure_message(f"Re-authentication failed for {account.alias}", exc), error=exc, ) return except Exception as exc: yield LifecycleEvent( kind="failure", account=account, message=self._failure_message(f"Re-authentication failed for {account.alias}", exc), error=exc, ) return yield LifecycleEvent(kind="success", account=self._with_verified_identity(account, verified)) def submit_code(self, session: ClaudeLoginSession, code: str) -> None: process = session.process stdin = process.stdin if stdin is None or process.poll() is not None: return try: stdin.write(code.strip() + "\n") stdin.flush() except (BrokenPipeError, ValueError): return def _login(self, account_home: Path, *, login_hint: str | None = None) -> Iterator[LoginEvent]: command = ["claude", "auth", "login", "--claudeai"] normalized_hint = (login_hint or "").strip() if normalized_hint: command.extend(["--email", normalized_hint]) env = os.environ.copy() env["CLAUDE_CONFIG_DIR"] = str(account_home) process = self._login_popen( command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, env=env, ) stdout = process.stdout if stdout is None: process.kill() process.wait() raise ClaudeAuthOperationError("claude auth login exited with code -1") output_queue: queue.Queue[str | object] = queue.Queue() sentinel = object() started_at = time.monotonic() deadline = started_at + LOGIN_TIMEOUT_SECS code_entry_deadline = started_at + CODE_ENTRY_FALLBACK_SECS session = ClaudeLoginSession(process=process, account_home=account_home) def reader() -> None: try: while True: chunk = stdout.read(1) if chunk == "": break output_queue.put(chunk) finally: output_queue.put(sentinel) reader_thread = threading.Thread(target=reader, daemon=True) reader_thread.start() def cleanup() -> None: if process.poll() is None: process.kill() process.wait() reader_thread.join() if process.stdin is not None: process.stdin.close() stdout.close() transcript = "" seen_url = False seen_code_prompt = False url_prefix = "If the browser didn't open, visit: " code_prompt = "Paste code here if prompted > " try: while True: now = time.monotonic() remaining = deadline - now if remaining <= 0: raise ClaudeAuthOperationError( f"Claude login timed out after {int(LOGIN_TIMEOUT_SECS)}s waiting for browser sign-in" ) if not seen_code_prompt and now >= code_entry_deadline and process.poll() is None: seen_code_prompt = True yield LoginEvent( kind="code_required", raw_text=transcript, session=session, ) continue wait_timeout = remaining if not seen_code_prompt: wait_timeout = min(wait_timeout, max(0.0, code_entry_deadline - now)) try: chunk = output_queue.get(timeout=wait_timeout) except queue.Empty as exc: if ( not seen_code_prompt and time.monotonic() < deadline and process.poll() is None ): seen_code_prompt = True yield LoginEvent( kind="code_required", raw_text=transcript, session=session, ) continue raise ClaudeAuthOperationError( f"Claude login timed out after {int(LOGIN_TIMEOUT_SECS)}s waiting for browser sign-in" ) from exc if not isinstance(chunk, str): break transcript += chunk if not seen_url and transcript.endswith("\n"): for line in transcript.splitlines(): if line.startswith(url_prefix): seen_url = True yield LoginEvent(kind="url_ready", url=line[len(url_prefix) :].strip()) if not seen_code_prompt and process.poll() is None: seen_code_prompt = True yield LoginEvent( kind="code_required", raw_text=transcript, session=session, ) break if not seen_code_prompt and code_prompt in transcript: seen_code_prompt = True yield LoginEvent( kind="code_required", raw_text=transcript, session=session, ) try: result = process.wait(timeout=max(0.0, deadline - time.monotonic())) except subprocess.TimeoutExpired as exc: raise ClaudeAuthOperationError( f"Claude login timed out after {int(LOGIN_TIMEOUT_SECS)}s waiting for browser sign-in" ) from exc if result != 0: raise ClaudeAuthOperationError(f"claude auth login exited with code {result}") finally: cleanup() def _verify_login(self, account_home: Path) -> VerifiedIdentity: credentials_path = account_home / ".credentials.json" if not credentials_path.exists(): raise ClaudeAuthOperationError("Claude credentials file is missing after login") result = self._call_runner( ["claude", "auth", "status", "--json"], config_dir=account_home, timeout=STATUS_TIMEOUT_SECS, ) if result.returncode != 0: raise ClaudeAuthOperationError("Claude auth status failed") try: payload = json.loads(result.stdout or "{}") except json.JSONDecodeError as exc: raise ClaudeAuthOperationError("Claude auth status returned invalid JSON") from exc if not isinstance(payload, dict) or payload.get("loggedIn") is not True: raise ClaudeAuthOperationError("Claude auth status did not confirm a logged-in session") metadata = self._read_json_object(account_home / "claude.json") credentials = self._read_json_object(credentials_path) oauth_account = metadata.get("oauthAccount") if not isinstance(oauth_account, dict): oauth_account = {} oauth_credentials = credentials.get("claudeAiOauth") if not isinstance(oauth_credentials, dict): oauth_credentials = {} email = self._first_text( payload.get("email"), payload.get("emailAddress"), oauth_account.get("emailAddress"), metadata.get("emailAddress"), ) org_id = self._first_text( payload.get("orgId"), payload.get("org_id"), payload.get("organizationUuid"), oauth_account.get("organizationUuid"), metadata.get("organizationUuid"), ) subscription_type = self._first_text( payload.get("subscriptionType"), payload.get("subscription_type"), payload.get("rateLimitTier"), oauth_account.get("organizationRateLimitTier"), oauth_account.get("seatTier"), metadata.get("plan"), oauth_credentials.get("subscriptionType"), oauth_credentials.get("rateLimitTier"), ) if email is None or org_id is None: raise ClaudeAuthOperationError("Claude identity metadata is incomplete after login") return VerifiedIdentity( email=email.lower(), org_id=org_id.lower(), subscription_type=subscription_type, ) def _find_collision( self, verified: VerifiedIdentity, *, exclude_slug: str | None, ) -> Account | None: verified_key = self._identity_key(verified) for account in self.registry.list(): if exclude_slug is not None and account.slug == exclude_slug: continue if account.email is None or account.account_id is None: continue if (account.email.strip().lower(), account.account_id.strip().lower()) == verified_key: return account return None def _pending_account(self, staged: StagedAccount) -> Account: return Account( ref=AccountRef("claude", staged.slug), alias=staged.alias, account_home=staged.account_home, email=None, plan=None, account_id=None, ) def _with_verified_identity(self, account: Account, verified: VerifiedIdentity) -> Account: return Account( ref=account.ref, alias=account.alias, account_home=account.account_home, email=verified.email, plan=verified.subscription_type, account_id=verified.org_id, ) def _read_persisted_identity(self, account_home: Path) -> VerifiedIdentity | None: path = account_home / "account_identity.json" if not path.exists(): return None payload = self._read_json_object(path) email = self._first_text(payload.get("email")) org_id = self._first_text(payload.get("org_id")) if email is None or org_id is None: return None subscription_type = self._first_text(payload.get("subscription_type")) return VerifiedIdentity( email=email.lower(), org_id=org_id.lower(), subscription_type=subscription_type, ) def _snapshot_files(self, account_home: Path) -> dict[str, tuple[bytes | None, int | None]]: snapshots: dict[str, tuple[bytes | None, int | None]] = {} for name in (".credentials.json", "account_identity.json", "claude.json"): path = account_home / name if path.exists(): snapshots[name] = (path.read_bytes(), path.stat().st_mode) else: snapshots[name] = (None, None) return snapshots def _restore_files( self, account_home: Path, snapshots: dict[str, tuple[bytes | None, int | None]], ) -> None: for name, (content, mode) in snapshots.items(): path = account_home / name if content is None: if path.exists(): path.unlink() continue self._atomic_write_bytes(path, content) if mode is not None: os.chmod(path, mode) def _write_identity_file(self, path: Path, verified: VerifiedIdentity) -> None: payload = { "email": verified.email, "org_id": verified.org_id, "subscription_type": verified.subscription_type, } self._atomic_write_bytes( path, (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8"), ) def _call_runner( self, command: list[str], *, config_dir: Path, timeout: float | None, ) -> CommandResult: return self._runner(command, config_dir=config_dir, timeout=timeout) @staticmethod def _run_command( command: list[str], *, config_dir: Path, timeout: float | None = None, ) -> CommandResult: env = os.environ.copy() env["CLAUDE_CONFIG_DIR"] = str(config_dir) completed = subprocess.run( command, check=False, capture_output=True, text=True, env=env, timeout=timeout, ) return CommandResult( returncode=completed.returncode, stdout=completed.stdout, stderr=completed.stderr, ) @staticmethod def _read_json_object(path: Path) -> dict[str, object]: try: payload = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return {} except json.JSONDecodeError as exc: raise ClaudeAuthOperationError(f"{path.name} is not valid JSON") from exc if not isinstance(payload, dict): raise ClaudeAuthOperationError(f"{path.name} must contain a JSON object") return payload @staticmethod def _first_text(*values: object) -> str | None: for value in values: if isinstance(value, str): stripped = value.strip() if stripped: return stripped return None @staticmethod def _identity_key(identity: VerifiedIdentity) -> tuple[str, str]: return (identity.email, identity.org_id) @staticmethod def _failure_message(prefix: str, exc: Exception) -> str: detail = str(exc) if not detail: return prefix return f"{prefix}: {detail}" @staticmethod def _atomic_write_bytes(path: Path, content: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "wb", dir=path.parent, delete=False, ) as handle: handle.write(content) temp_name = handle.name os.replace(temp_name, path) @staticmethod @contextmanager def _account_lock(account_home: Path) -> Iterator[Path]: lock_path = account_home / ".claude-auth.lock" with lock_path.open("a+b") as handle: fcntl.flock(handle.fileno(), fcntl.LOCK_EX) try: yield lock_path finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN)