climperor/web/steam_auth.py

"""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