Add the OAuth callback endpoint at /api/qt

Schwab requires an HTTPS callback. The usual answer is https://127.0.0.1:8182
behind a self-signed certificate, which means clicking through a browser warning
on every re-authentication — and the refresh token expires weekly. There are
also reports of Schwab refusing to register apps whose callback is a loopback
address. This app already terminates real HTTPS, so it can take the redirect
itself.

Unauthenticated by necessity: the provider redirects a browser here and cannot
attach the chart token, so it sits alongside /health and /version. It is inert —
nothing is stored, and the page echoes only the query string of the request that
produced it, which the caller already has in their address bar. Retaining the
code would let a later anonymous visitor read it.

The path and the page are both deliberately unrevealing. That is not a security
control; it just avoids advertising which brokerage this host talks to. Treat
the path as fixed — changing a registered callback means editing the app, which
can send it back through approval.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Chris Amow 2026-08-10 04:45:31 -05:00
parent e69d1c190f
commit 0bafe9de01
3 changed files with 115 additions and 0 deletions

72
app/api/schwab_auth.py Normal file
View file

@ -0,0 +1,72 @@
"""Schwab OAuth callback.
Schwab requires an HTTPS callback URL. The usual answer is
``https://127.0.0.1:8182`` with a self-signed certificate, which means clicking
through a browser warning on every re-authentication — and the refresh token
expires weekly. There are also reports of Schwab refusing to register apps whose
callback is a loopback address.
This app already terminates real HTTPS, so it can receive the redirect itself.
Deliberately unauthenticated: Schwab redirects a browser here and cannot attach
the chart token. Nothing is stored — the page only echoes the query string of
the request that produced it, which the caller already has in their address bar.
Storing the code would mean a later, unauthenticated visitor could read it.
The path is deliberately unrevealing. That is not a security control — the
endpoint's safety is that it is inert — it simply avoids advertising which
brokerage this host talks to. Treat it as fixed: changing a registered callback
URL means editing the Schwab app, which can send it back through approval.
"""
from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse
router = APIRouter(prefix="/api")
PAGE = """<!doctype html>
<meta charset="utf-8">
<title>Callback</title>
<style>
body {{ font: 15px/1.6 ui-sans-serif, system-ui, sans-serif; max-width: 46rem;
margin: 3rem auto; padding: 0 1.5rem; background: #14161a; color: #e8eaed; }}
h1 {{ font-size: 1.2rem; }}
code, textarea {{ font-family: ui-monospace, monospace; font-size: 13px; }}
textarea {{ width: 100%; height: 7rem; padding: .7rem; border-radius: 6px;
border: 1px solid #2a2e35; background: #0e1013; color: #e8eaed; }}
.warn {{ color: #efb643; }}
.muted {{ color: #9aa1ab; }}
</style>
<h1>{heading}</h1>
{body}
"""
RECEIVED = """
<p>Paste this entire URL into the waiting login prompt:</p>
<textarea readonly onclick="this.select()">{url}</textarea>
<p class="warn">Single use, and it expires within minutes. Do not share it.</p>
<p class="muted">Nothing was stored on the server. This page shows only the URL
you just arrived with.</p>
"""
IDLE = """
<p>OAuth callback endpoint. Register this exact URL with the provider:</p>
<p><code>{url}</code></p>
<p class="muted">Arriving here directly is expected and harmless — the useful
version of this page is the one you are redirected to.</p>
"""
@router.get("/qt", response_class=HTMLResponse)
def callback(request: Request) -> HTMLResponse:
if request.query_params.get("code"):
page = PAGE.format(
heading="Authorisation received",
body=RECEIVED.format(url=str(request.url)),
)
else:
page = PAGE.format(
heading="Callback endpoint",
body=IDLE.format(url=str(request.url).split("?")[0]),
)
# Never cached: it carries a single-use authorisation code.
return HTMLResponse(page, headers={"Cache-Control": "no-store"})

View file

@ -12,6 +12,7 @@ from fastapi.staticfiles import StaticFiles
from app.api.meta import router as meta_router from app.api.meta import router as meta_router
from app.api.routes import router as api_router from app.api.routes import router as api_router
from app.api.schwab_auth import router as schwab_auth_router
from app.api.ws import router as ws_router from app.api.ws import router as ws_router
from app.config import Settings from app.config import Settings
from app.runtime import Runtime from app.runtime import Runtime
@ -37,6 +38,7 @@ app = FastAPI(title="chart", lifespan=lifespan)
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.include_router(meta_router) app.include_router(meta_router)
app.include_router(schwab_auth_router)
app.include_router(api_router) app.include_router(api_router)
app.include_router(ws_router) app.include_router(ws_router)

View file

@ -0,0 +1,41 @@
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.schwab_auth import router
def client() -> TestClient:
app = FastAPI()
app.include_router(router)
return TestClient(app)
def test_callback_needs_no_chart_token():
# Schwab redirects a browser here and cannot attach the token, so this
# endpoint has to stay open the way /health and /version do.
assert client().get("/api/qt").status_code == 200
def test_page_does_not_name_the_brokerage():
# The path is neutral so the host does not advertise who it trades with;
# the page saying it anyway would defeat that.
assert "chwab" not in client().get("/api/qt").text
def test_landing_here_directly_explains_itself():
body = client().get("/api/qt").text
assert "Register this exact URL" in body
assert "code" not in body.split("<style>")[0]
def test_authorisation_code_is_echoed_for_the_manual_flow():
response = client().get("/api/qt", params={"code": "abc123", "session": "s"})
assert "abc123" in response.text
assert response.headers["cache-control"] == "no-store"
def test_the_code_is_not_retained_for_a_later_visitor():
session = client()
session.get("/api/qt", params={"code": "secret-code"})
# A second, code-less request must not replay the first one's code.
assert "secret-code" not in session.get("/api/qt").text