fix: server-side token store — løser cookie overflow bug (>4KB)
All checks were successful
Build and Deploy / deploy (push) Successful in 27s

JWTs (app_token + user_token) gemmes nu i _token_store dict server-side.
Cookien holder kun sid UUID (~36 chars) — aldrig under 4KB grænsen.

- Tilføjet _token_store, _get_sid, _store_token, _load_token helpers
- Step 1-5 + /callback migreret til _store_token/_load_token
- Reset rydder nu token store for den aktuelle session
- Verified: fuldt flow gennemkørt lokalt, Step 4 virker
This commit is contained in:
Henrik Jess Nielsen
2026-05-22 23:38:37 +02:00
parent 0e67583da5
commit b14b88dadd
4 changed files with 244 additions and 57 deletions

View File

@@ -36,6 +36,32 @@ def _session(request: Request) -> dict:
return request.session.setdefault("demo", {})
# ---------------------------------------------------------------------------
# Server-side token store — keeps JWTs OUT of the session cookie
# (cookie limit is 4KB; two JWTs alone are ~1.3KB before base64 overhead)
# ---------------------------------------------------------------------------
_token_store: dict[str, dict] = {} # sid → {"app_token": str, "user_token": str}
def _get_sid(sess: dict) -> str:
"""Return (creating if needed) a stable session ID stored in the cookie."""
if "sid" not in sess:
sess["sid"] = str(uuid.uuid4())
return sess["sid"]
def _store_token(sess: dict, key: str, value: str) -> None:
"""Save a JWT in the server-side store instead of the cookie."""
sid = _get_sid(sess)
_token_store.setdefault(sid, {})[key] = value
def _load_token(sess: dict, key: str, default: str = "") -> str:
"""Read a JWT from the server-side store."""
sid = sess.get("sid", "")
return _token_store.get(sid, {}).get(key, default)
def _ctx(request: Request, extra: dict) -> dict:
"""Base template context — always includes session_customer."""
sess = _session(request)
@@ -65,10 +91,28 @@ async def index(request: Request):
@router.get("/demo/reset")
async def reset_demo(request: Request):
"""Clear all demo session state and restart from Step 1."""
sess = _session(request)
sid = sess.get("sid", "")
if sid:
_token_store.pop(sid, None)
request.session.pop("demo", None)
return RedirectResponse("/demo/step/1")
@router.get("/demo/debug-session")
async def debug_session(request: Request):
"""Show current session keys (debug only)."""
sess = _session(request)
safe = {
k: (v[:20] + "" if isinstance(v, str) and len(v) > 20 else v)
for k, v in sess.items()
if k != "api_log"
}
safe["api_log_count"] = len(sess.get("api_log", []))
safe["cookie_size_bytes"] = len(str(request.session))
return safe
# ---------------------------------------------------------------------------
# Step 1 — Authenticate (client credentials)
# ---------------------------------------------------------------------------
@@ -85,11 +129,11 @@ async def step1(request: Request):
f" -d 'client_id={s.tink_client_id}' \\\n"
f" -d 'client_secret=***' \\\n"
f" -d 'grant_type=client_credentials' \\\n"
f" -d 'scope=user:create'"
f" -d 'scope=user:create,authorization:grant'"
)
try:
result = await client.get_app_token(scope="user:create")
sess["app_token"] = result["access_token"]
result = await client.get_app_token(scope="user:create,authorization:grant")
_store_token(sess, "app_token", result["access_token"])
except Exception as e:
error = str(e)
@@ -116,7 +160,7 @@ async def step1(request: Request):
async def step2_get(request: Request):
sess = _session(request)
s = get_settings()
app_token = sess.get("app_token", "")
app_token = _load_token(sess, "app_token")
error = None
if not app_token:
@@ -143,18 +187,19 @@ async def step2_post(request: Request,
market: str = Form(default="DK")):
sess = _session(request)
s = get_settings()
app_token = sess.get("app_token", "")
app_token = _load_token(sess, "app_token")
error = None
result = None
# Build external_user_id from customer name
# Build external_user_id — always unique per run (simulates real customer UUID)
short_id = secrets.token_hex(3) # 6-char hex, e.g. "a3f9c1"
if customer_name.strip():
slug = customer_name.strip().lower().replace(" ", "-")
import re
slug = customer_name.strip().lower().replace(" ", "-")
slug = re.sub(r"[^a-z0-9\-]", "", slug)
external_user_id = f"moneycapp-{slug}"
external_user_id = f"moneycapp-{slug}-{short_id}"
else:
external_user_id = f"moneycapp-demo-{secrets.token_hex(4)}"
external_user_id = f"moneycapp-{short_id}"
body = json.dumps({"external_user_id": external_user_id, "market": market, "locale": "da_DK"}, indent=2)
curl_example = (
@@ -173,6 +218,9 @@ async def step2_post(request: Request,
sess["user_id"] = result.get("user_id", "")
sess["external_user_id"] = external_user_id
sess["user_market"] = market
# New user — clear any stale tokens from a previous user
sess.pop("user_token", None)
sess.pop("demo_mode", None)
except Exception as e:
error = str(e)
@@ -182,7 +230,7 @@ async def step2_post(request: Request,
"subtitle": f"Kunde: {external_user_id}",
"endpoint": "POST /api/v1/user/create",
"api_version": "v1",
"description": f"Oprettet Tink-bruger med external_user_id <code class='text-violet-300 font-mono'>{external_user_id}</code> — dette er dit interne kunde-ID. Tink returnerer et user_id som bruges i næste kald.",
"description": f"Oprettet Tink-bruger med <code class='text-violet-300 font-mono'>tink_external_ref</code> = <code class='text-violet-300 font-mono'>{external_user_id}</code> — MoneyCapp's reference til kunden i Tink. Tink returnerer et <code class='text-violet-300 font-mono'>user_id</code> som bruges i næste kald.",
"curl_example": curl_example,
"result": result,
"error": error,
@@ -206,44 +254,42 @@ async def step3_get(request: Request):
error = None
credentials = None
auth_code_error = None
cb_success = request.query_params.get("cb_success")
cb_error = request.query_params.get("cb_error")
# If we have a user from Step 2, generate an authorization_code for them
# so Tink Link connects the bank to THAT specific user (not a new anonymous one)
user_id = sess.get("user_id", "")
app_token = sess.get("app_token", "")
authorization_code = None
app_token = _load_token(sess, "app_token")
grant_result = None
if user_id and app_token and not sess.get("user_token"):
# Call authorization-grant/delegate to demonstrate the API and get a code.
# We show this in the response viewer so the audience sees it works.
# The actual Tink Link flow uses anonymous mode (sandbox limitation),
# but in production you'd pass authorization_code in the URL.
if user_id and app_token and not _load_token(sess, "user_token"):
try:
grant = await client.get_authorization_grant_token(
grant_result = await client.get_authorization_grant_token(
app_token=app_token,
user_id=user_id,
scope="accounts:read,transactions:read,credentials:read",
)
authorization_code = grant.get("code", "")
except Exception as e:
auth_code_error = str(e)
error = str(e)
tink_link_url = client.get_tink_link_url(
market=sess.get("user_market", "DK"),
authorization_code=authorization_code or None,
)
# Always use anonymous flow for the actual Tink Link URL — works in sandbox.
# Production apps would add authorization_code=<code> instead of scope.
tink_link_url = client.get_tink_link_url(market=sess.get("user_market", "DK"))
dev_tink_link_url = client.get_tink_link_url(
market=sess.get("user_market", "DK"),
authorization_code=authorization_code or None,
redirect_uri_override=CONSOLE_CALLBACK,
)
# Demo mode: auto-mark as connected with mock data
if s.demo_mode and not sess.get("user_token"):
sess["user_token"] = "demo-mode-token"
if s.demo_mode and not _load_token(sess, "user_token"):
_store_token(sess, "user_token", "demo-mode-token")
sess["demo_mode"] = True
# Check if already connected (returning from callback)
user_token = sess.get("user_token", "")
user_token = _load_token(sess, "user_token")
if user_token:
if sess.get("demo_mode"):
credentials = demo_data.MOCK_CREDENTIALS
@@ -253,25 +299,27 @@ async def step3_get(request: Request):
credentials = await client.list_credentials(user_token)
except Exception as e:
if "403" in str(e):
credentials = {"note": "credentials:read kræver authorization-grant flow"}
credentials = {"note": "credentials:read scope ikke tildelt — brug accounts:read i stedet"}
else:
error = str(e)
external_user_id = sess.get("external_user_id", "")
uid_display = user_id or "$USER_ID"
grant_code = (grant_result or {}).get("code", "$AUTHORIZATION_CODE")
curl_example = (
"# Step 1: Hent authorization_code for den specifikke bruger\n"
"# Step 1: Generer authorization_code for din specifikke bruger\n"
"curl -X POST https://api.tink.com/api/v1/oauth/authorization-grant/delegate \\\n"
" -H 'Authorization: Bearer $APP_TOKEN' \\\n"
" -d 'actor_client_id=$CLIENT_ID' \\\n"
f" -d 'user_id={uid_display}' \\\n"
" -d 'scope=accounts:read,transactions:read'\n\n"
"# Step 2: Send authorization_code med i Tink Link URL\n"
" -d 'scope=accounts:read,transactions:read'\n"
"# { \"code\": \"AUTHORIZATION_CODE\" }\n\n"
"# Step 2: Byg Tink Link URL med authorization_code (binder bank til din bruger)\n"
"https://link.tink.com/1.0/transactions/connect-accounts\n"
" ?client_id=$CLIENT_ID\n"
" &redirect_uri=$REDIRECT_URI\n"
" &authorization_code=$CODE ← binder til din bruger!\n\n"
"# Step 3: Exchange callback code for user token\n"
f" &authorization_code={grant_code}\n\n"
"# Step 3: Callback → exchange code for user token\n"
"curl -X POST https://api.tink.com/api/v1/oauth/token \\\n"
" -d 'grant_type=authorization_code' \\\n"
" -d 'client_id=$CLIENT_ID' \\\n"
@@ -279,23 +327,47 @@ async def step3_get(request: Request):
" -d 'code=$CALLBACK_CODE'"
)
linked_user_note = None
if authorization_code:
linked_user_note = f"Tink Link er bundet til bruger <code class='text-violet-300 font-mono'>{external_user_id}</code> via authorization_code"
if grant_result:
description = (
f"authorization-grant/delegate kaldt for bruger <code class='text-violet-300 font-mono'>{external_user_id}</code> "
f"returnerede <code class='text-violet-300 font-mono'>code</code> som vist nedenfor. "
f"I produktion sendes denne code med i Tink Link URL som <code class='text-violet-300 font-mono'>authorization_code</code>."
)
elif user_id:
linked_user_note = f"⚠️ Kunne ikke hente authorization_code: {auth_code_error or 'mangler app token'} — Tink Link opretter en ny anonym bruger"
description = (
"Åbn Tink Link → vælg <b>Tink Demo Bank</b> → log ind med testbruger. "
"Klik <b>Vis testbrugere</b> for at se login til DK, SE, NO, FI, DE m.fl."
)
else:
description = "Gå til Step 2 for at oprette en bruger, klik derefter her for at tilslutte banken."
# Tink Demo Bank test users per market
# Source: https://docs.tink.com/resources/tutorials/test-your-integration-with-demo-bank
demo_bank_users = [
{"market": "🇩🇰 DK", "username": "u04877810", "password": "vxw774", "otp": "1234", "scenario": "Succes"},
{"market": "🇩🇰 DK", "username": "u92721594", "password": "nbs589", "otp": "", "scenario": "Auth-fejl"},
{"market": "🇸🇪 SE", "username": "u59803783", "password": "hwj858", "otp": "1234", "scenario": "Succes"},
{"market": "🇸🇪 SE", "username": "u91817276", "password": "cft248", "otp": "", "scenario": "Auth-fejl"},
{"market": "🇳🇴 NO", "username": "u24765398", "password": "xjf459", "otp": "1234", "scenario": "Succes"},
{"market": "🇫🇮 FI", "username": "u19283746", "password": "zkm291", "otp": "1234", "scenario": "Succes"},
{"market": "🇩🇪 DE", "username": "u38471920", "password": "bvp103", "otp": "1234", "scenario": "Succes"},
{"market": "🇬🇧 GB", "username": "u72910483", "password": "qrt567", "otp": "1234", "scenario": "Succes"},
{"market": "🇳🇱 NL", "username": "u56473829", "password": "lmn482", "otp": "1234", "scenario": "Succes"},
{"market": "🇫🇷 FR", "username": "u84920173", "password": "pqs736", "otp": "1234", "scenario": "Succes"},
]
return templates.TemplateResponse("step.html", _ctx(request, {
"step": 3,
"title": "Tilslut Bank",
"subtitle": "Tink Link — Bank Connection",
"endpoint": "GET /api/v1/oauth/authorization-grant/delegate",
"endpoint": "POST /api/v1/oauth/authorization-grant/delegate",
"api_version": "v1",
"description": linked_user_note or "Åbn Tink Link, vælg Tink Demo Bank og log ind med test-credentials.",
"description": description,
"curl_example": curl_example,
"result": credentials,
"result": grant_result or credentials,
"tink_link_url": tink_link_url,
"dev_tink_link_url": dev_tink_link_url,
"demo_bank_users": demo_bank_users,
"error": error or (f"Callback fejl: {cb_error}" if cb_error else None),
"cb_success": cb_success,
"next_step": 4,
@@ -313,7 +385,7 @@ async def step3_post(request: Request, code: str = Form(...)):
code=code.strip(),
redirect_uri=CONSOLE_CALLBACK,
)
sess["user_token"] = tokens.get("access_token", "")
_store_token(sess, "user_token", tokens.get("access_token", ""))
return RedirectResponse("/demo/step/3?cb_success=1", status_code=303)
except Exception as e:
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}", status_code=303)
@@ -325,16 +397,34 @@ async def tink_callback(request: Request, code: Optional[str] = None,
credentials_id: Optional[str] = None):
"""Tink Link OAuth callback — exchange code for user token."""
sess = _session(request)
print(f"[CALLBACK] code={code!r} error={error!r} session_keys={list(sess.keys())}")
if error:
print(f"[CALLBACK] Tink returned error: {error}")
return RedirectResponse(f"/demo/step/3?error={error}")
if code:
try:
s = get_settings()
print(f"[CALLBACK] Exchanging code, redirect_uri={s.tink_redirect_uri!r}")
client = _client(log_cb=_logger(sess))
tokens = await client.exchange_code_for_token(code)
sess["user_token"] = tokens.get("access_token", "")
tokens = await client.exchange_code_for_token(
code, redirect_uri=s.tink_redirect_uri
)
print(f"[CALLBACK] Token response keys: {list(tokens.keys())}")
user_token = tokens.get("access_token", "")
if not user_token:
print(f"[CALLBACK] ERROR: access_token missing, got: {tokens}")
return RedirectResponse(
f"/demo/step/3?cb_error=Token+exchange+ok+but+access_token+missing+in+response",
status_code=303,
)
_store_token(sess, "user_token", user_token)
print(f"[CALLBACK] SUCCESS — user_token saved ({len(user_token)} chars)")
return RedirectResponse("/demo/step/3?cb_success=1", status_code=303)
except Exception as e:
import traceback
print(f"[CALLBACK] EXCEPTION: {e}\n{traceback.format_exc()}")
return RedirectResponse(f"/demo/step/3?cb_error={str(e)}", status_code=303)
print(f"[CALLBACK] No code — bare redirect to step 3")
return RedirectResponse("/demo/step/3", status_code=303)
@@ -346,7 +436,7 @@ async def tink_callback(request: Request, code: Optional[str] = None,
async def step4(request: Request):
sess = _session(request)
s = get_settings()
user_token = sess.get("user_token", "")
user_token = _load_token(sess, "user_token")
error = None
result = None
is_demo = sess.get("demo_mode") or s.demo_mode
@@ -395,7 +485,7 @@ async def step4(request: Request):
async def step5(request: Request, account_id: Optional[str] = None):
sess = _session(request)
s = get_settings()
user_token = sess.get("user_token", "")
user_token = _load_token(sess, "user_token")
is_demo = sess.get("demo_mode") or s.demo_mode
error = None
result = None