From c71e95bcdf327cb05ce074e522bed47e96175a3d Mon Sep 17 00:00:00 2001 From: justin Date: Mon, 25 May 2026 18:56:48 -0400 Subject: [PATCH] Restore oauth-headers scrape script to fix broken LaunchAgent The LaunchAgent plist (com.justin.codexmeter.claude-plan-scrape) was updated to reference claude-plan-usage-oauth-headers.py while the feat/claude-oauth-headers-scrape branch was active, but the script was never merged to main. After switching back to main the LaunchAgent began failing every 15s (exit code 2) causing claude-plan-usage.json to go stale and the daemon to fall back to "needs login". Restored from feeda27 on feat/claude-oauth-headers-scrape. Verified the script reads the Claude Code OAuth token from Keychain and returns fresh session/weekly percentages via Anthropic rate-limit headers. Co-Authored-By: Claude Sonnet 4.6 --- scripts/claude-plan-usage-oauth-headers.py | 283 +++++++++++++++++++++ 1 file changed, 283 insertions(+) create mode 100644 scripts/claude-plan-usage-oauth-headers.py diff --git a/scripts/claude-plan-usage-oauth-headers.py b/scripts/claude-plan-usage-oauth-headers.py new file mode 100644 index 0000000..aec83e0 --- /dev/null +++ b/scripts/claude-plan-usage-oauth-headers.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +"""Read Claude plan usage from anthropic-ratelimit-unified-* response headers. + +Approach (adapted from Hermann Bjorgvin's Clawdmeter +https://github.com/HermannBjorgvin/Clawdmeter): + + 1. Read the Claude Code OAuth access token from the macOS Keychain + (service "Claude Code-credentials") or from ~/.claude/.credentials.json + on Linux. + 2. POST a minimal /v1/messages request to api.anthropic.com (1 token of + Haiku, ~$0.0001 per scrape). + 3. Ignore the body. Parse: + anthropic-ratelimit-unified-5h-utilization (0.0-1.0 float) + anthropic-ratelimit-unified-5h-reset (epoch seconds) + anthropic-ratelimit-unified-7d-utilization + anthropic-ratelimit-unified-7d-reset + anthropic-ratelimit-unified-5h-status (string) + 4. Write the normalized JSON to claude-plan-usage.json — same shape the + daemon already reads. + +This is more durable than the cookie scrapes because: + * No browser or Cloudflare in the loop. + * Uses the documented /v1/messages endpoint. + * Claude Code rotates the OAuth token in Keychain on its own. + +If anything fails, the script falls back to claude-plan-usage-desktop.py +(which itself falls through to the Firefox cookie scraper). +""" + +from __future__ import annotations + +import argparse +import getpass +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request +from datetime import datetime +from pathlib import Path + +CONFIG_DIR = Path.home() / ".config" / "codexmeter" +CACHE_PATH = Path( + os.getenv( + "CODEXMETER_CLAUDE_PLAN_USAGE_FILE", str(CONFIG_DIR / "claude-plan-usage.json") + ) +) +HEALTH_PATH = Path( + os.getenv( + "CODEXMETER_CLAUDE_SCRAPE_HEALTH", str(CONFIG_DIR / "claude-scrape-health.json") + ) +) +KEYCHAIN_SERVICE = os.getenv( + "CODEXMETER_CLAUDE_CODE_KEYCHAIN_SERVICE", "Claude Code-credentials" +) +CREDENTIALS_PATH = Path( + os.getenv( + "CODEXMETER_CLAUDE_CODE_CREDENTIALS", + str(Path.home() / ".claude" / ".credentials.json"), + ) +) +API_URL = "https://api.anthropic.com/v1/messages" +API_HEADERS_TEMPLATE = { + "anthropic-version": "2023-06-01", + "anthropic-beta": "oauth-2025-04-20", + "content-type": "application/json", + "user-agent": os.getenv("CODEXMETER_CLAUDE_UA", "claude-code/2.1.5"), + "accept": "application/json", +} +API_BODY = { + "model": os.getenv("CODEXMETER_CLAUDE_PROBE_MODEL", "claude-haiku-4-5-20251001"), + "max_tokens": 1, + "messages": [{"role": "user", "content": "hi"}], +} + + +def _read_token_keychain() -> str | None: + try: + out = subprocess.run( + [ + "security", + "find-generic-password", + "-w", + "-s", + KEYCHAIN_SERVICE, + "-a", + getpass.getuser(), + ], + check=False, + capture_output=True, + text=True, + timeout=10, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return None + if out.returncode != 0 or not out.stdout.strip(): + return None + return _extract_access_token(out.stdout) + + +def _read_token_file() -> str | None: + try: + raw = CREDENTIALS_PATH.read_text() + except OSError: + return None + return _extract_access_token(raw) + + +def _extract_access_token(blob: str) -> str | None: + blob = blob.strip() + if not blob: + return None + try: + data = json.loads(blob) + except json.JSONDecodeError: + return None + if isinstance(data, dict): + # nested: {"claudeAiOauth": {"accessToken": "..."}} + for value in data.values(): + if isinstance(value, dict) and isinstance(value.get("accessToken"), str): + return value["accessToken"] + if isinstance(data.get("accessToken"), str): + return data["accessToken"] + return None + + +def read_oauth_token() -> str: + token = _read_token_keychain() if sys.platform == "darwin" else _read_token_file() + if not token: + # Try the other source as a last-ditch. + token = _read_token_file() if sys.platform == "darwin" else _read_token_keychain() + if not token: + raise SystemExit( + "Could not read Claude Code OAuth token (Keychain service " + f"'{KEYCHAIN_SERVICE}' or {CREDENTIALS_PATH}). Sign in to " + "Claude Code so it stores a fresh token." + ) + return token + + +def probe_headers(token: str) -> dict[str, str]: + body = json.dumps(API_BODY).encode() + headers = dict(API_HEADERS_TEMPLATE) + headers["authorization"] = f"Bearer {token}" + req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers) + try: + with urllib.request.urlopen(req, timeout=20) as resp: + return {k.lower(): v for k, v in resp.getheaders()} + except urllib.error.HTTPError as exc: + # Some 4xx responses still return ratelimit headers; keep them in case + # we still get a useful snapshot. But error out on auth failures so the + # fallback chain runs. + body_text = exc.read().decode("utf-8", errors="replace")[:200] + if exc.code == 401: + raise SystemExit( + f"OAuth bearer rejected by /v1/messages (HTTP 401). Token may " + f"have expired. Body: {body_text}" + ) from exc + raise SystemExit( + f"/v1/messages returned HTTP {exc.code}: {body_text}" + ) from exc + + +def pct(util: str | None) -> int: + if util is None: + return 0 + try: + return int(round(float(util) * 100)) + except ValueError: + return 0 + + +def reset_minutes(reset_ts: str | None) -> int: + if reset_ts is None: + return -1 + try: + when = float(reset_ts) + except ValueError: + return -1 + delta = (when - time.time()) / 60.0 + return max(0, int(round(delta))) + + +def normalize(headers: dict[str, str]) -> dict: + five_util = headers.get("anthropic-ratelimit-unified-5h-utilization") + five_reset = headers.get("anthropic-ratelimit-unified-5h-reset") + week_util = headers.get("anthropic-ratelimit-unified-7d-utilization") + week_reset = headers.get("anthropic-ratelimit-unified-7d-reset") + if five_util is None or week_util is None: + raise SystemExit( + f"Expected anthropic-ratelimit-unified-* headers; got: " + f"{sorted(k for k in headers if 'ratelimit' in k)}" + ) + return { + "source": "claude_api_oauth_headers", + "captured_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "session": { + "used_pct": pct(five_util), + "reset_mins": reset_minutes(five_reset), + }, + "weekly": { + "used_pct": pct(week_util), + "reset_mins": reset_minutes(week_reset), + }, + } + + +def write_atomic(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(payload, separators=(",", ":"))) + tmp.replace(path) + + +def write_health(ok: bool, message: str, payload: dict | None = None) -> None: + health: dict = { + "ok": ok, + "source": "oauth_headers", + "checked_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "message": message, + } + if payload: + health["last_payload"] = payload + try: + write_atomic(HEALTH_PATH, health) + except OSError: + pass + + +def try_fallback(command: str) -> int: + fallback = Path(__file__).parent / "claude-plan-usage-desktop.py" + if not fallback.exists(): + return 1 + print( + f"oauth-headers scrape failed; falling back to {fallback.name} {command}", + file=sys.stderr, + ) + proc = subprocess.run([sys.executable, str(fallback), command], check=False) + return proc.returncode + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument( + "command", nargs="?", default="scrape", choices=["scrape", "print"] + ) + ap.add_argument( + "--no-fallback", + action="store_true", + help="Don't try the desktop-cookie script if this one fails.", + ) + args = ap.parse_args() + + try: + token = read_oauth_token() + headers = probe_headers(token) + payload = normalize(headers) + except SystemExit as exc: + msg = f"oauth-headers scrape failed: {exc}" + write_health(False, str(exc)) + if args.no_fallback: + raise + print(msg, file=sys.stderr) + return try_fallback(args.command) + except Exception as exc: # noqa: BLE001 + msg = f"oauth-headers scrape failed: {type(exc).__name__}: {exc}" + write_health(False, msg) + if args.no_fallback: + raise SystemExit(msg) from exc + print(msg, file=sys.stderr) + return try_fallback(args.command) + + if args.command == "scrape": + write_atomic(CACHE_PATH, payload) + write_health(True, "ok", payload) + print(json.dumps(payload, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())