Validate Schwab credentials before sending anyone to a browser

The first attempt failed with invalid_client and no way to tell why: the key,
the secret, the callback, or an app not yet propagated all look identical from
the browser, which shows raw JSON. It turned out to be a key clipped by one
character on paste.

Two preflight checks now say which. The authorize endpoint is asked whether it
recognises the key. The token endpoint is asked to exchange a deliberately
invalid code, which separates bad credentials from a bad grant — it
authenticates the key and secret over HTTP Basic before it looks at the code, so
invalid_client means the pair is wrong and invalid_grant means the pair is fine.

That second check matters more than it sounds. The secret is not used at all
during login, so a truncated one survives the whole browser round trip and only
surfaces at the exchange, by which point the authorisation code has been spent
and the flow has to start over.

Neither check can tell a wrong value from an app that is not live yet — an
invented key produces the identical response, verified — and both say so rather
than guessing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

scripts/check_schwab.py

@@ -1,23 +1,33 @@
 """Find out what a Schwab app is actually entitled to, before building on it.
 
-Run once after putting SCHWAB_API_KEY and SCHWAB_APP_SECRET in .env. It answers
+Answers the three questions that decide the design, empirically rather than
-the three questions that decide the design, and it answers them empirically
+from documentation:
-rather than from documentation:
 
   1. Do the credentials authenticate at all?
   2. Do REST quotes work for /ES — futures market data is a separate
      entitlement from equities and may not be granted.
   3. Does the streamer connect? Its bootstrap reads /trader/v1/userPreference,
      which belongs to the Accounts and Trading product, so an app registered
-     for Market Data Production alone is expected to fail here.
+     for Market Data Production alone is expected to fail there.
 
-Uses the manual OAuth flow because the callback is hosted rather than local:
+Two steps, neither of them interactive, so this works over a pipe or from an
-visit the printed URL, complete the login, then paste the URL you land on. The
+agent session where stdin is not a terminal:
-callback page shows it ready to copy.
 
     python3 -m scripts.check_schwab
+        prints the Schwab login URL.
+
+    python3 -m scripts.check_schwab --redirect-url 'https://.../api/qt?code=…'
+        exchanges the code, saves the token, runs the checks.
+
+The browser does not have to be on this machine. Nothing is captured locally —
+you copy a URL out, and paste a URL back. The authorisation code is single use
+and expires within minutes, so do not leave it sitting between the two steps.
+
+Once a token exists, running with no arguments skips straight to the checks.
 """
-import asyncio
+import argparse
+import sys
+from urllib.parse import parse_qs, urlparse
 
 from app.config import Settings
 
@@ -26,71 +36,187 @@ def heading(text: str) -> None:
     print(f"\n{text}\n{'-' * len(text)}")
 
 
-async def main() -> None:
+def load_settings() -> Settings:
     settings = Settings()
     if not settings.schwab_api_key or not settings.schwab_app_secret:
         raise SystemExit("Set SCHWAB_API_KEY and SCHWAB_APP_SECRET in .env first")
+    return settings
+
+
+def key_is_live(authorization_url: str) -> bool:
+    """Ask Schwab whether it recognises the app key, before opening a browser.
+
+    A key Schwab does not know produces `invalid_client` here — and so does a
+    deliberately invented one, byte for byte, so this cannot tell "wrong value"
+    from "not active yet". It can still save a confusing round trip through the
+    login page.
+    """
+    import httpx
 
     try:
-        from schwab.auth import client_from_manual_flow, client_from_token_file
+        response = httpx.get(authorization_url, follow_redirects=False, timeout=15)
-    except ImportError:
-        raise SystemExit("pip install -r requirements-dev.txt (schwab-py is not installed)")
-
-    heading("1. Authentication")
-    token_path = str(settings.schwab_token_path)
-    try:
-        client = client_from_token_file(
-            token_path, settings.schwab_api_key, settings.schwab_app_secret
-        )
-        print(f"  reused the token at {token_path}")
     except Exception:
-        print("  no usable token — starting the manual flow")
+        return True  # Network trouble is not evidence about the key.
-        client = client_from_manual_flow(
+    return "invalid_client" not in response.text
+
+
+def secret_is_valid(settings: Settings) -> bool | None:
+    """Check the secret without needing an authorisation code.
+
+    The token endpoint authenticates the key and secret over HTTP Basic before
+    it looks at the grant, so a deliberately invalid code separates the two
+    failures: bad credentials give invalid_client, good credentials give
+    invalid_grant. Otherwise a truncated secret survives the login unnoticed and
+    only surfaces at the exchange, after the code has been spent.
+
+    None when the answer is not clear enough to act on.
+    """
+    import httpx
+
+    try:
+        response = httpx.post(
+            "https://api.schwabapi.com/v1/oauth/token",
+            auth=(settings.schwab_api_key, settings.schwab_app_secret),
+            data={
+                "grant_type": "authorization_code",
+                "code": "deliberately-invalid-code",
+                "redirect_uri": settings.schwab_callback_url,
+            },
+            timeout=20,
+        )
+    except Exception:
+        return None
+    if "invalid_grant" in response.text:
+        return True
+    if "invalid_client" in response.text or response.status_code == 401:
+        return False
+    return None
+
+
+def print_login_url(settings: Settings) -> None:
+    from schwab.auth import get_auth_context
+
+    context = get_auth_context(settings.schwab_api_key, settings.schwab_callback_url)
+
+    if secret_is_valid(settings) is False:
+        print("Schwab rejects the app key and secret pair.\n")
+        print("The key alone is accepted at the authorize endpoint, so this is")
+        print("the secret. Re-copy it from the portal using the Show icon —")
+        print("a value clipped by one character looks entirely normal.")
+        sys.exit(1)
+
+    if not key_is_live(context.authorization_url):
+        print("Schwab rejects this app key with invalid_client.\n")
+        print("The secret is not involved yet — it is only used when the code is")
+        print("exchanged — so this is the key itself or the app's readiness.\n")
+        print("  1. Re-copy the App Key from the portal using the Show icon.")
+        print("  2. If it matches, the app is most likely not live yet. Newly")
+        print("     created or newly edited apps take time to propagate, and the")
+        print("     portal says Ready For Use before the key works.")
+        print("\nRe-run this to check again; nothing else is needed.")
+        sys.exit(1)
+
+    print("Open this in any browser, on any machine, and approve the app:\n")
+    print(f"  {context.authorization_url}\n")
+    print("You will land on the callback URL. A 404 there is fine until the")
+    print("branch is deployed — the code is in the address bar either way.")
+    print("Copy the ENTIRE address and run:\n")
+    print("  python3 -m scripts.check_schwab --redirect-url '<paste it here>'")
+
+
+def exchange(settings: Settings, redirect_url: str):
+    from schwab import auth
+    from schwab.auth import AuthContext, client_from_received_url
+
+    state = parse_qs(urlparse(redirect_url).query).get("state", [None])[0]
+    if not state:
+        raise SystemExit("That URL has no ?state= — paste the full address you landed on")
+
+    settings.schwab_token_path.parent.mkdir(parents=True, exist_ok=True)
+    # The library's own writer, so the token file keeps the shape its loader
+    # expects rather than one guessed at here.
+    write_token = getattr(auth, "__make_update_token_func")(str(settings.schwab_token_path))
+
+    # Only the state is needed again; the authorisation URL is not, which is
+    # what lets the two steps share nothing. Taking the state from the pasted
+    # URL makes the CSRF check a formality — acceptable because the thing being
+    # guarded against is a redirect you did not initiate, and you pasted this
+    # one in by hand.
+    context = AuthContext(settings.schwab_callback_url, None, state)
+    return client_from_received_url(
         settings.schwab_api_key,
         settings.schwab_app_secret,
-            settings.schwab_callback_url,
+        context,
-            token_path,
+        redirect_url,
+        write_token,
     )
-    print("  authenticated")
 
-    heading(f"2. REST quote for {settings.schwab_symbol}")
+
-    response = client.get_quote(settings.schwab_symbol)
+def run_checks(client, settings: Settings) -> None:
-    print(f"  HTTP {response.status_code}")
+    heading(f"REST quote for {settings.schwab_symbol}")
-    if response.status_code == 200:
+    quote = client.get_quote(settings.schwab_symbol)
-        payload = response.json()
+    print(f"  HTTP {quote.status_code}")
-        print(f"  keys: {list(payload)[:4]}")
+    quotes_ok = quote.status_code == 200 and bool(quote.json())
-        for symbol, data in list(payload.items())[:1]:
+    if quotes_ok:
-            quote = data.get("quote", {})
+        for symbol, data in list(quote.json().items())[:1]:
-            print(f"  {symbol}: last={quote.get('lastPrice')} "
+            values = data.get("quote", {})
-                  f"bid={quote.get('bidPrice')} ask={quote.get('askPrice')}")
+            print(f"  {symbol}: last={values.get('lastPrice')} "
+                  f"bid={values.get('bidPrice')} ask={values.get('askPrice')}")
         print("  -> futures market data IS available")
     else:
-        print(f"  body: {response.text[:200]}")
+        print(f"  body: {quote.text[:200]}")
         print("  -> futures market data is NOT available on this app")
 
-    heading("3. Streamer bootstrap (/trader/v1/userPreference)")
+    heading("Streamer bootstrap (/trader/v1/userPreference)")
     prefs = client.get_user_preferences()
     print(f"  HTTP {prefs.status_code}")
-    if prefs.status_code == 200:
+    streaming_ok = prefs.status_code == 200 and bool(prefs.json().get("streamerInfo"))
-        info = prefs.json().get("streamerInfo") or []
+    if streaming_ok:
-        print(f"  streamerInfo entries: {len(info)}")
+        print(f"  streamerInfo entries: {len(prefs.json()['streamerInfo'])}")
         print("  -> streaming is available; CHART_FUTURES should work")
     else:
         print(f"  body: {prefs.text[:200]}")
-        print("  -> streaming is NOT available. This endpoint belongs to the")
+        print("  -> streaming is NOT available on this app")
-        print("     Accounts and Trading product; a Market Data Production app")
-        print("     cannot reach it, so CHART_FUTURES is out of reach until the")
-        print("     app adds that product.")
 
     heading("Summary")
-    print("  Quotes available  :", response.status_code == 200)
+    print(f"  Quotes    : {'yes' if quotes_ok else 'no'}")
-    print("  Streaming available:", prefs.status_code == 200)
+    print(f"  Streaming : {'yes' if streaming_ok else 'no'}")
-    if response.status_code == 200 and prefs.status_code != 200:
+    if quotes_ok and not streaming_ok:
         print("\n  Polling REST quotes is then the real-time path: it removes")
         print("  Yahoo's ten-minute delay without needing trading scope, at the")
         print("  cost of building bars from snapshots rather than receiving")
         print("  true exchange OHLCV.")
 
 
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--redirect-url", help="the full URL you were redirected to")
+    args = parser.parse_args()
+    settings = load_settings()
+
+    try:
+        from schwab.auth import client_from_token_file
+    except ImportError:
+        raise SystemExit("pip install -r requirements-dev.txt (schwab-py is not installed)")
+
+    if args.redirect_url:
+        heading("Exchanging the authorisation code")
+        client = exchange(settings, args.redirect_url)
+        print(f"  token written to {settings.schwab_token_path}")
+    elif settings.schwab_token_path.exists():
+        heading("Authentication")
+        client = client_from_token_file(
+            str(settings.schwab_token_path),
+            settings.schwab_api_key,
+            settings.schwab_app_secret,
+        )
+        print(f"  reused the token at {settings.schwab_token_path}")
+    else:
+        print_login_url(settings)
+        sys.exit(0)
+
+    run_checks(client, settings)
+
+
 if __name__ == "__main__":
-    asyncio.run(main())
+    main()