Chris Amow
cc25871032
The live socket never made a REST call, so the seven-day refresh token expired while the chart still looked fine. A deploy then could not log in. Ping user preferences every six hours, and when the grant is already dead offer a one-click reconnect that writes the token on the existing callback.
108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
"""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, RedirectResponse
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
|
|
def begin_login(settings) -> object:
|
|
from schwab.auth import get_auth_context
|
|
|
|
return get_auth_context(settings.schwab_api_key, settings.schwab_callback_url)
|
|
|
|
|
|
def complete_login(settings, context, redirect_url: str) -> None:
|
|
from schwab import auth
|
|
from schwab.auth import client_from_received_url
|
|
|
|
path = settings.schwab_token_path
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
write_token = getattr(auth, "__make_update_token_func")(str(path))
|
|
client_from_received_url(
|
|
settings.schwab_api_key,
|
|
settings.schwab_app_secret,
|
|
context,
|
|
redirect_url,
|
|
write_token,
|
|
)
|
|
|
|
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>
|
|
"""
|
|
|
|
|
|
FAILED = """
|
|
<p>The authorisation code could not be exchanged. Start again from the chart.</p>
|
|
<p class="muted">Nothing was stored.</p>
|
|
"""
|
|
|
|
|
|
@router.get("/qt")
|
|
def callback(request: Request):
|
|
runtime = getattr(request.app.state, "runtime", None)
|
|
if runtime is not None and runtime.schwab_login is not None and request.query_params.get("code"):
|
|
try:
|
|
runtime.finish_schwab_login(str(request.url))
|
|
except Exception:
|
|
page = PAGE.format(heading="Authorisation failed", body=FAILED)
|
|
return HTMLResponse(page, headers={"Cache-Control": "no-store"}, status_code=400)
|
|
return RedirectResponse("/", headers={"Cache-Control": "no-store"})
|
|
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"})
|