"""Portable-credential transfer — the blob format `export-credential` writes and `import-credential` installs, for both providers, in ONE place so the two ends can never drift apart. Run (all paths absolute; the blob never passes through argv): python3 lib/credential.py export [--identity-only] [--out PATH] writes the blob to stdout, or to PATH (created 0600 and renamed into place, so a refusal leaves nothing behind and no other process ever sees it wider) python3 lib/credential.py inspect validates a blob and prints one 0x1F-separated record: class, id, email, home, added_at, cred_type. The separator is deliberately NOT whitespace: bash collapses runs of IFS whitespace, which would silently shift an empty field (a blob with no `home`) into the next one. python3 lib/credential.py install writes the credential material into (0600, atomic); no-op for an identity-only blob. Prints the file it wrote (or nothing). Exit codes are the contract a daemon branches on: 0 done 2 usage / unknown account / malformed blob 3 the credential is MACHINE-LOCAL — copying it would break both machines 4 the account has no credential material here 5 credential material is present but unusable (corrupt / wrong shape) WHAT IS PORTABLE claude: `server.token` — a subscription setup-token (`sk-ant-oat…`, ~1y life, inference-only). It authenticates from any machine and is exactly what the existing sync already mirrors to the server, so distributing it is safe. claude: `.credentials.json` — NOT portable. Its refresh token rotates on every refresh; a second machine refreshing the same grant strands the first. codex: `auth.json` — NOT portable, same rotating-refresh-token reason. Codex has no portable credential type at all: each machine signs in once with the device-code flow (`codex-accounts login `, works over SSH). """ import json import os import re import socket import sys import tempfile import time FORMAT = 'claude-multiacc/credential' VERSION = 1 VALID_ID = re.compile(r'acct-\d{2}') # Metadata crosses into a shell (the CLIs read it as a record) and into the manifest. # A control character there could truncate or shift a record, so it is refused at the # boundary rather than escaped downstream. CONTROL_CHARS = re.compile(r'[\x00-\x1f\x7f]') FIELD_SEP = '\x1f' # A subscription setup-token, and nothing else — API keys are refused pool-wide # (bin/claude-accounts: valid_subscription_token). Kept in sync with that check. SETUP_TOKEN = re.compile(r'sk-ant-oat\d{2}-[A-Za-z0-9_-]{40,}\Z') def die(msg, code=2): sys.stderr.write(msg.rstrip('\n') + '\n') sys.exit(code) def _load(path): try: with open(path) as f: return json.load(f) except Exception as e: die(f'not readable JSON: {path} ({str(e)[:120]})') def _clean(value, what): """Metadata that is safe to hand to a shell and to write into the manifest.""" text = '' if value is None else str(value) if CONTROL_CHARS.search(text): die(f'{what} contains a control character — refusing to transfer it') return text def _write_secure(path, data, nofollow_dir=False): """Create with the content, never following a symlink and never widening permissions: a unique 0600 temp file in the same directory (mkstemp is O_EXCL), then an atomic rename. os.replace does not follow a symlink at the destination, so a planted link is replaced rather than written through. nofollow_dir additionally pins the DIRECTORY: it is opened O_NOFOLLOW|O_DIRECTORY once and every later step runs relative to that descriptor, so swapping the directory for a symlink after the check cannot redirect the write. Used for credential material landing in a pool; a caller-named --out path does not need it (the caller chose the path) and may live on a filesystem without dir_fd support.""" d = os.path.dirname(os.path.abspath(path)) or '.' base = os.path.basename(path) if not os.path.isdir(d): die(f'destination directory does not exist: {d}') dir_fd = None if nofollow_dir: # Fail CLOSED: if the platform cannot pin the directory, refuse rather than # quietly downgrading to the path-based write this argument exists to avoid. # NB: the rename primitive is os.rename, not os.replace — on POSIX both are # renameat(2) (rename already replaces atomically), but CPython registers only # os.rename in supports_dir_fd on macOS, and demanding os.replace would make # this fail closed on the very platform the pool runs on. missing = [n for n in ('O_DIRECTORY', 'O_NOFOLLOW') if not hasattr(os, n)] if missing or os.open not in os.supports_dir_fd \ or os.rename not in os.supports_dir_fd \ or os.unlink not in os.supports_dir_fd: die(f'this platform cannot write a credential safely (no directory-fd ' f'support{": missing " + ", ".join(missing) if missing else ""}) — ' f'refusing to install it', 5) try: dir_fd = os.open(d, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) except OSError as e: die(f'could not open {d} safely ({str(e)[:120]})') tmp = None try: if dir_fd is not None: tmp = f'.cred.{os.getpid()}.{os.urandom(6).hex()}' fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=dir_fd) os.fchmod(fd, 0o600) with os.fdopen(fd, 'w') as f: f.write(data) os.rename(tmp, base, src_dir_fd=dir_fd, dst_dir_fd=dir_fd) tmp = None else: fd, tmp = tempfile.mkstemp(prefix='.cred.', dir=d) os.fchmod(fd, 0o600) with os.fdopen(fd, 'w') as f: f.write(data) os.replace(tmp, path) tmp = None except Exception as e: # Any exception, not just OSError: an interrupt or an unexpected TypeError # must not leave a file holding live credential material behind. die(f'could not write {path} ({str(e)[:120]})') finally: if tmp: try: if dir_fd is not None: os.unlink(tmp, dir_fd=dir_fd) else: os.unlink(tmp) except OSError: pass if dir_fd is not None: os.close(dir_fd) def _manifest_account(root, aid): doc = _load(os.path.join(root, 'accounts.json')) if not isinstance(doc, dict): die(f'manifest is not a JSON object: {os.path.join(root, "accounts.json")}') for a in doc.get('accounts', []): if isinstance(a, dict) and str(a.get('id', '')) == aid: return a return None def cmd_export(argv): if len(argv) < 3: die('usage: credential.py export ' '[--identity-only] [--out PATH]') root, provider, aid = argv[0], argv[1], argv[2] rest = list(argv[3:]) identity_only = '--identity-only' in rest out_path = None if '--out' in rest: i = rest.index('--out') if i + 1 >= len(rest): die('--out requires a path') out_path = rest[i + 1] def emit(blob): text = json.dumps(blob, indent=2) + '\n' if out_path: _write_secure(out_path, text) else: sys.stdout.write(text) if not VALID_ID.fullmatch(aid): die(f'not a valid account id: {aid}') acct = _manifest_account(root, aid) if acct is None: die(f'{aid} is not registered in {os.path.join(root, "accounts.json")}') d = os.path.join(root, aid) blob = { 'format': FORMAT, 'version': VERSION, 'provider': provider, 'class': 'identity', 'account': { 'id': aid, 'email': _clean(acct.get('email', ''), 'the account email'), 'home': _clean(acct.get('home', ''), 'the account home'), 'added_at': _clean(acct.get('added_at', ''), 'added_at') or None, }, 'exported_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'exported_from': {'host': socket.gethostname(), 'machine': 'mac' if sys.platform == 'darwin' else 'linux', 'pool_root': root}, } if identity_only: emit(blob) return if provider == 'codex': has_auth = os.path.isfile(os.path.join(d, 'auth.json')) \ and os.path.getsize(os.path.join(d, 'auth.json')) > 0 if not has_auth: die(f'{aid} has no codex credential in {d}\n' f' fix: sign in on the machine that should run it — ' f'codex-accounts login {aid}', 4) die(f'{aid} is a MACHINE-LOCAL codex credential (auth.json) and cannot be exported.\n' f' Why: auth.json carries a refresh token that ROTATES on every refresh; a\n' f' second machine using the same grant invalidates the first one.\n' f' Codex has no portable credential type — sign in once per machine:\n' f' codex-accounts login {aid} (device-code flow; works over SSH)\n' f' To move only the registry entry (id/email/home), add --identity-only.', 3) tpath = os.path.join(d, 'server.token') cpath = os.path.join(d, '.credentials.json') has_token = os.path.isfile(tpath) and os.path.getsize(tpath) > 0 if not has_token: where = None if os.path.isfile(cpath) and os.path.getsize(cpath) > 0: where = '.credentials.json' else: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import keychain if keychain.probe(d)['state'] in ('present', 'locked', 'corrupt'): where = 'macOS Keychain' if where: die(f'{aid} only has a MACHINE-LOCAL credential ({where}) and cannot ' f'be exported.\n' f' Why: the OAuth grant\'s refresh token ROTATES on every refresh; a second\n' f' machine refreshing the same grant strands the first one.\n' f' Make it portable (one interactive sign-in on THIS machine):\n' f' claude-accounts mint {aid} # mints a portable setup-token\n' f' then re-run export-credential. Or sign in on the other machine:\n' f' claude-accounts login {aid}\n' f' To move only the registry entry (id/email/home), add --identity-only.', 3) die(f'{aid} has no credential material in {d}\n' f' fix: claude-accounts login {aid} (or --token for a portable one)', 4) try: with open(tpath) as f: token = f.read().strip() except OSError as e: die(f'{aid}: server.token unreadable ({str(e)[:120]})', 5) if not SETUP_TOKEN.fullmatch(token): die(f'{aid}: server.token is not a subscription setup-token (sk-ant-oat…) — ' f'refusing to export unusable material.\n' f' fix: claude-accounts mint {aid}', 5) blob['class'] = 'portable' blob['credential'] = {'type': 'setup-token', 'value': token} emit(blob) def _validate(provider, path): blob = _load(path) if not isinstance(blob, dict): die('blob is not a JSON object') if blob.get('format') != FORMAT: die(f'not a claude-multiacc credential blob (format={blob.get("format")!r})') try: version = int(blob.get('version')) except (TypeError, ValueError): die(f'blob version is not a number: {blob.get("version")!r}') if version > VERSION: die(f'blob is version {version}; this claude-multiacc understands up to ' f'{VERSION} — update the addon (claude-accounts self-update)') if blob.get('provider') != provider: die(f'blob is for provider {blob.get("provider")!r}, not {provider!r} — ' f'use {blob.get("provider")}-accounts import-credential') cclass = blob.get('class') if cclass not in ('portable', 'identity'): die(f'unknown credential class {cclass!r} (expected portable or identity)') acct = blob.get('account') if not isinstance(acct, dict): die("blob has no 'account' object") aid = _clean(acct.get('id', ''), 'blob account id') if aid and not VALID_ID.fullmatch(aid): die(f'blob account id is malformed: {aid!r}') email = _clean(acct.get('email', ''), 'blob account email').strip() if not email: die('blob account has no email') cred_type = '' if cclass == 'portable': cred = blob.get('credential') if not isinstance(cred, dict): die("blob is class 'portable' but has no 'credential' object") cred_type = _clean(cred.get('type', ''), 'credential type') value = cred.get('value') if provider == 'codex': die('codex credentials are machine-local — a portable codex blob cannot be ' 'installed. Sign in on this machine instead: codex-accounts login ', 3) if cred_type != 'setup-token': die(f'unsupported credential type {cred_type!r} for provider {provider} ' f'(expected setup-token)', 3) if not isinstance(value, str) or not SETUP_TOKEN.fullmatch(value.strip()): die('credential value is not a subscription setup-token (sk-ant-oat…) — ' 'API keys and truncated tokens are refused', 5) return (blob, cclass, aid, email, _clean(acct.get('home', ''), 'blob account home'), _clean(acct.get('added_at', ''), 'blob added_at'), cred_type) def cmd_inspect(argv): if len(argv) < 2: die('usage: credential.py inspect ') _, cclass, aid, email, home, added_at, cred_type = _validate(argv[0], argv[1]) # 0x1F-separated: a NON-whitespace separator, so an empty field (a blob with no # `home`) stays an empty field when bash splits the record instead of collapsing # into the next one. _clean has already refused any control character in the data. print(FIELD_SEP.join((cclass, aid, email, home, added_at, cred_type))) def cmd_install(argv): if len(argv) < 3: die('usage: credential.py install ') provider, path, dest = argv[0], argv[1], argv[2] blob, cclass, _aid, _email, _home, _added, _type = _validate(provider, path) if cclass == 'identity': return # registry-only: nothing to write if os.path.islink(dest): die(f'destination is a symlink ({dest} -> {os.readlink(dest)}) — refusing to ' f'write a credential through it', 2) if not os.path.isdir(dest): die(f'destination is not a directory: {dest}') out = os.path.join(dest, 'server.token') _write_secure(out, blob['credential']['value'].strip(), nofollow_dir=True) print(out) if __name__ == '__main__': if len(sys.argv) < 2: die(__doc__.strip()) verb, rest = sys.argv[1], sys.argv[2:] if verb == 'export': cmd_export(rest) elif verb == 'inspect': cmd_inspect(rest) elif verb == 'install': cmd_install(rest) else: die(f'unknown verb: {verb}')