simplify SSO: read JWT from localStorage, bypass OAuth2 authorize flow entirely

This commit is contained in:
2026-08-31 17:39:29 +08:00
parent cb5573da73
commit b335430c18
2 changed files with 63 additions and 109 deletions

View File

@@ -7,6 +7,7 @@ import subprocess
import uuid
import urllib.request
import urllib.error
import base64
from pathlib import Path
from typing import Any, BinaryIO, Dict, List, Optional
import json
@@ -206,36 +207,43 @@ def create_app(
video_repository.delete_session(token)
return Response(status_code=204)
D1KT_USERINFO_URL = "https://d1kt.cn/api/oauth2/userinfo"
@application.post("/api/v1/auth/sso/exchange", response_model=SsoExchangeResponse)
def sso_exchange(payload: SsoExchangeRequest) -> SsoExchangeResponse:
# Call the 第一课堂 userinfo endpoint to validate the access_token.
req = urllib.request.Request(
D1KT_USERINFO_URL,
headers={"Authorization": f"Bearer {payload.access_token}"},
)
# Decode the JWT payload from 第一课堂 (without signature verification —
# the token was obtained directly from the 第一课堂 login page in a WebView).
try:
with urllib.request.urlopen(req, timeout=10) as resp:
userinfo = json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as exc:
parts = payload.access_token.split(".")
if len(parts) != 3:
raise ValueError("Invalid JWT format")
# Base64url decode the payload (second part)
padding = len(parts[1]) % 4
if padding:
parts[1] += "=" * (4 - padding)
decoded_bytes = base64.urlsafe_b64decode(parts[1])
userinfo = json.loads(decoded_bytes.decode("utf-8"))
except (ValueError, json.JSONDecodeError, IndexError) as exc:
raise HTTPException(
status_code=401,
detail=f"第一课堂 access_token validation failed: {exc}",
detail=f"Failed to decode 第一课堂 JWT token: {exc}",
)
d1kt_username = userinfo.get("username") or userinfo.get("name") or ""
if not d1kt_username:
raise HTTPException(status_code=401, detail="第一课堂 userinfo did not contain a username.")
# Check if the token is expired
import time
exp = userinfo.get("exp")
if exp and time.time() > exp:
raise HTTPException(status_code=401, detail="第一课堂 token has expired.")
fullname = userinfo.get("fullname") or userinfo.get("lastname") or d1kt_username
d1kt_username = userinfo.get("username") or ""
if not d1kt_username:
raise HTTPException(status_code=401, detail="第一课堂 token did not contain a username.")
fullname = userinfo.get("fullname") or d1kt_username
email = userinfo.get("email") or ""
# Try to find an existing local user with the same username.
# Find or create a local user with the same username.
existing = video_repository.get_user_by_username(d1kt_username)
if existing is None:
# Create a new SSO-linked user.
user = video_repository.create_sso_user(
username=d1kt_username,
nickname=fullname,