Ship Steam login, D1 player sync, and cached「我」dashboard.

Players get a fast TTL-backed homepage (local profile / Cloudflare D1) with dense UI polish; login unlocks /home without blocking on every OpenDota refresh.

Co-authored-by: Cursor <cursoragent@cursor.com>

web/steam_auth.py

@@ -0,0 +1,179 @@
+"""Steam OpenID login + signed cookie sessions for local serve_relations.
+
+Env: STEAM_API_KEY, SESSION_SECRET.
+Mirrors web/frontend/functions/api/auth/* Pages Functions.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any
+
+COOKIE_NAME = "climperor_steam"
+SESSION_DAYS = 30
+STEAM_OPENID = "https://steamcommunity.com/openid/login"
+STEAM_ID_PREFIX = "https://steamcommunity.com/openid/id/"
+UA = "climperor-steam-auth"
+
+
+def _b64url_encode(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _b64url_decode(s: str) -> bytes:
+    pad = "=" * (-len(s) % 4)
+    return base64.urlsafe_b64decode(s + pad)
+
+
+def steam_id64_to_account_id(steam_id64: str | int) -> int | None:
+    try:
+        n = int(steam_id64)
+    except (TypeError, ValueError):
+        return None
+    account = n - 76561197960265728
+    return account if account > 0 else None
+
+
+def parse_steam_id_from_claimed(claimed_id: str | None) -> str | None:
+    if not claimed_id or not claimed_id.startswith(STEAM_ID_PREFIX):
+        return None
+    sid = claimed_id[len(STEAM_ID_PREFIX) :].rstrip("/")
+    return sid if sid.isdigit() and len(sid) == 17 else None
+
+
+def create_session_token(secret: str, payload: dict) -> str:
+    body = dict(payload)
+    body["exp"] = int(time.time()) + SESSION_DAYS * 86400
+    raw = _b64url_encode(json.dumps(body, separators=(",", ":"), ensure_ascii=False).encode())
+    sig = _b64url_encode(hmac.new(secret.encode(), raw.encode(), hashlib.sha256).digest())
+    return f"{raw}.{sig}"
+
+
+def verify_session_token(secret: str, token: str | None) -> dict | None:
+    if not secret or not token or "." not in token:
+        return None
+    raw, sig = token.split(".", 1)
+    expect = _b64url_encode(hmac.new(secret.encode(), raw.encode(), hashlib.sha256).digest())
+    if not hmac.compare_digest(sig, expect):
+        return None
+    try:
+        data = json.loads(_b64url_decode(raw).decode("utf-8"))
+    except (ValueError, UnicodeDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+    if int(data.get("exp") or 0) < int(time.time()):
+        return None
+    if not data.get("steamid") or not data.get("account_id"):
+        return None
+    return data
+
+
+def read_cookie(cookie_header: str | None, name: str = COOKIE_NAME) -> str | None:
+    if not cookie_header:
+        return None
+    for part in cookie_header.split(";"):
+        part = part.strip()
+        if part.startswith(name + "="):
+            return urllib.parse.unquote(part[len(name) + 1 :])
+    return None
+
+
+def cookie_header(token: str, *, clear: bool = False, secure: bool = False) -> str:
+    if clear:
+        base = f"{COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0"
+    else:
+        max_age = SESSION_DAYS * 86400
+        base = (
+            f"{COOKIE_NAME}={urllib.parse.quote(token)}; Path=/; HttpOnly; "
+            f"SameSite=Lax; Max-Age={max_age}"
+        )
+    if secure:
+        base = base.replace("SameSite=Lax", "Secure; SameSite=Lax")
+    return base
+
+
+def steam_login_redirect_url(origin: str) -> str:
+    return_to = f"{origin.rstrip('/')}/api/auth/steam/callback"
+    params = {
+        "openid.ns": "http://specs.openid.net/auth/2.0",
+        "openid.mode": "checkid_setup",
+        "openid.return_to": return_to,
+        "openid.realm": origin.rstrip("/"),
+        "openid.identity": "http://specs.openid.net/auth/2.0/identifier_select",
+        "openid.claimed_id": "http://specs.openid.net/auth/2.0/identifier_select",
+    }
+    return f"{STEAM_OPENID}?{urllib.parse.urlencode(params)}"
+
+
+def verify_steam_openid(query: dict[str, list[str] | str]) -> tuple[bool, str | None]:
+    def one(key: str) -> str | None:
+        v = query.get(key)
+        if isinstance(v, list):
+            return v[0] if v else None
+        return v
+
+    if one("openid.mode") != "id_res":
+        return False, None
+    steamid = parse_steam_id_from_claimed(one("openid.claimed_id"))
+    if not steamid:
+        return False, None
+    body: dict[str, str] = {}
+    for k, v in query.items():
+        if not k.startswith("openid."):
+            continue
+        body[k] = v[0] if isinstance(v, list) else str(v)
+    body["openid.mode"] = "check_authentication"
+    data = urllib.parse.urlencode(body).encode()
+    req = urllib.request.Request(
+        STEAM_OPENID,
+        data=data,
+        headers={"User-Agent": UA, "Content-Type": "application/x-www-form-urlencoded"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            text = resp.read().decode("utf-8", errors="replace")
+    except (urllib.error.URLError, TimeoutError, OSError):
+        return False, None
+    normalized = "".join(text.split()).lower()
+    if "is_valid:true" not in normalized:
+        return False, None
+    return True, steamid
+
+
+def fetch_steam_persona(api_key: str, steamid: str) -> dict[str, Any]:
+    if not api_key:
+        return {"personaname": None, "avatar": None}
+    url = (
+        "https://api.steampowered.com/ISteamUser/GetPlayerSummaries/v2/"
+        f"?key={urllib.parse.quote(api_key)}&steamids={urllib.parse.quote(steamid)}"
+    )
+    req = urllib.request.Request(url, headers={"User-Agent": UA})
+    try:
+        with urllib.request.urlopen(req, timeout=20) as resp:
+            data = json.loads(resp.read().decode())
+    except (urllib.error.URLError, TimeoutError, OSError, ValueError):
+        return {"personaname": None, "avatar": None}
+    players = (((data or {}).get("response") or {}).get("players")) or []
+    if not players:
+        return {"personaname": None, "avatar": None}
+    p = players[0]
+    return {
+        "personaname": p.get("personaname") or None,
+        "avatar": p.get("avatarfull") or p.get("avatarmedium") or p.get("avatar"),
+    }
+
+
+def env_secrets() -> tuple[str, str]:
+    key = (os.environ.get("STEAM_API_KEY") or "").strip()
+    secret = (os.environ.get("SESSION_SECRET") or "").strip()
+    return key, secret