懒猫微服进阶心得（十四）：接入 Casdoor，玩转OpenID Connect（OIDC）

在之前的文章中，我们演示了如何基于 **懒猫自带的 OpenID Connect（OIDC）** 来实现身份认证。那属于「平台内置」的简化方案，主要是帮助大家快速理解 OIDC 的基本使用场景。 https://appstore.lazycat.cloud/#/shop/detail/cloud.lazycat.app.casdoor 这一次，我们换一个更通用、更贴近生产实践的方式：使用应用商店里的 **OpenID Connect（OIDC） Provider —— Casdoor**。Casdoor 是一个开源的统一身份认证平台，支持完整的 OIDC 协议，可以作为独立的 IdP（Identity Provider）对接到任何应用。通过它，我们不仅能跑通最标准的授权码流程，还能深入理解 OIDC 的关键环节：授权跳转、Token 换取、ID Token 验签以及用户信息获取。  我们经常听到 **单点登录（SSO）**、**OAuth2**、**JWT** 这些词。 OIDC（OpenID Connect）正是基于 OAuth2 的标准化身份认证协议。它的核心作用是： - 帮助应用确认用户是谁（认证） - 不需要你自己维护密码和用户库（交给 IdP） - 与 OAuth2 完全兼容，可以同时获取访问 API 的能力（授权） 一个形象的比喻： - OAuth2 提供的是“门禁卡”功能（你能不能进某个房间） - OIDC 在此基础上加了“身份证”功能（你是谁） ### OIDC 基本流程 OIDC 的标准授权码流程（Authorization Code Flow）： 1. **用户访问应用** → 应用把用户跳转到 IdP 登录页 2. **用户在 IdP 登录** → IdP 返回一个授权码（code） 3. **应用后端用授权码换 token**（包括 access_token 和 id_token） 4. **应用验证 id_token** → 确认用户身份 5. **可选：调用 userinfo 接口** 获取更详细的用户信息 #### 进入管理后台 1. 登录到你的 Casdoor 管理控制台（通常是 https://casdoor.<name>.heiyu.space/ 或者部署时设定的管理地址）。 2. 用管理员账号（admin/123）进入后台。 3. OIDC 应用必须挂在某个 **Organization** 下。 4. 默认有一个 `built-in` 组织，可以直接用，也可以新建一个。  #### 创建应用 (Application) 1. 在左侧菜单选择 **Application** → 点击 **Add**。 2. 填写基本信息： - **Name**：应用名称（例如 `my-oidc-app`）。 - **Display name**：显示名称。 - **Organization**：选择上一步的组织。 - **Logo**：可选。  3. 在 **Authentication** 部分： - 设置 **Redirect URIs**： - 必须和你应用里写的一致，例如： ``` http://localhost:5001/auth/callback ```  4. 在 **OAuth 授权类型** 部分：勾选 `authorization_code`（最标准的流程）。  #### 获取 OIDC 参数 保存后，在应用详情页可以看到： - **Client ID** - **Client Secret** - **Redirect URI**（你填的）  同时，Casdoor 服务提供一个 OIDC Discovery 地址： ``` https://<casdoor-server-domain>/.well-known/openid-configuration ``` 这个地址返回 JSON，里面包括： - `issuer` - `authorization_endpoint` - `token_endpoint` - `userinfo_endpoint` - `jwks_uri` - `end_session_endpoint` 这些就是在后端应用里要配置的参数。 #### 验证配置 1. 在浏览器里直接访问： ``` https://<casdoor-server-domain>/.well-known/openid-configuration ``` 如果能看到 JSON，说明 OIDC 服务已开启。 ```json { "issuer": "https://casdoor.xxxx.heiyu.space", "authorization_endpoint": "https://casdoor.xxxx.heiyu.space/login/oauth/authorize", "token_endpoint": "https://casdoor.xxxx.heiyu.space/api/login/oauth/access_token", "userinfo_endpoint": "https://casdoor.xxxx.heiyu.space/api/userinfo", "jwks_uri": "https://casdoor.xxx.heiyu.space/.well-known/jwks", "introspection_endpoint": "https://casdoor.xxx.heiyu.space/api/login/oauth/introspect", "response_types_supported": [ "code", "token", "id_token", "code token", "code id_token", "token id_token", "code token id_token", "none" ], "response_modes_supported": ["query", "fragment", "login", "code", "link"], "grant_types_supported": ["password", "authorization_code"], "subject_types_supported": ["public"], "id_token_signing_alg_values_supported": [ "RS256", "RS512", "ES256", "ES384", "ES512" ], "scopes_supported": [ "openid", "email", "profile", "address", "phone", "offline_access" ], "claims_supported": [ "iss", "ver", "sub", "aud", "iat", "exp", "id", "type", "displayName", "avatar", "permanentAvatar", "email", "phone", "location", "affiliation", "title", "homepage", "bio", "tag", "region", "language", "score", "ranking", "isOnline", "isAdmin", "isForbidden", "signupApplication", "ldap" ], "request_parameter_supported": true, "request_object_signing_alg_values_supported": ["HS256", "HS384", "HS512"], "end_session_endpoint": "https://casdoor.xxxxx.heiyu.space/api/logout" } ``` 2. 用 Postman 或者 oidc-client 测试一下授权流程，看看能不能拿到 `code`、`access_token`、`id_token`。 ✅ 至此，Casdoor 端就配置好了。剩下的就是在应用端（RP）写 OIDC 客户端代码 ### OpenID Connect(OIDC)代码 ```python import os, requests, jwt from urllib.parse import urlencode from flask import Flask, redirect, request, session, url_for, jsonify, abort from dotenv import load_dotenv from jwt import PyJWKClient # ---------------- Load env ---------------- load_dotenv() app = Flask(__name__) app.secret_key = os.getenv("FLASK_SECRET_KEY", "dev-secret") ISSUER = os.getenv("OIDC_ISSUER") CLIENT_ID = os.getenv("OIDC_CLIENT_ID") CLIENT_SECRET = os.getenv("OIDC_CLIENT_SECRET") REDIRECT_URI = os.getenv("OIDC_REDIRECT_URI") # ---------------- Discover OIDC endpoints ---------------- discovery = requests.get(f"{ISSUER}/.well-known/openid-configuration").json() AUTH_ENDPOINT = discovery["authorization_endpoint"] TOKEN_ENDPOINT = discovery["token_endpoint"] USERINFO_ENDPOINT = discovery["userinfo_endpoint"] JWKS_URI = discovery["jwks_uri"] # ---------------- Routes ---------------- @app.route("/") def index(): if "user" in session: return f"Hi, {session['user'].get('name') or session['user']['sub']}<br><a href='/logout'>退出</a>" return "<a href='/login'>使用 Casdoor 登录</a>" @app.route("/login") def login(): params = { "client_id": CLIENT_ID, "response_type": "code", "scope": "openid profile email", "redirect_uri": REDIRECT_URI, "state": "xyz123", # 可以生成随机数并存到 session "nonce": "abc456" } return redirect(f"{AUTH_ENDPOINT}?{urlencode(params)}") @app.route("/auth/callback") def callback(): if "error" in request.args: return f"Error: {request.args['error']}" code = request.args.get("code") if not code: abort(400, "Missing code") # 1. 换取 Token data = { "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, } token_resp = requests.post(TOKEN_ENDPOINT, data=data).json() id_token = token_resp.get("id_token") access_token = token_resp.get("access_token") # 2. 验证并解码 ID Token jwks_client = PyJWKClient(JWKS_URI) signing_key = jwks_client.get_signing_key_from_jwt(id_token) claims = jwt.decode( id_token, signing_key.key, algorithms=["RS256"], audience=CLIENT_ID, issuer=ISSUER, ) # 3. 获取用户信息 userinfo = requests.get( USERINFO_ENDPOINT, headers={"Authorization": f"Bearer {access_token}"} ).json() session["user"] = { "sub": claims["sub"], "name": userinfo.get("name", claims.get("name")), "email": userinfo.get("email", claims.get("email")), } return redirect(url_for("profile")) @app.route("/profile") def profile(): if "user" not in session: return redirect("/") return jsonify(session["user"]) @app.route("/logout") def logout(): session.clear() return redirect("/") if __name__ == "__main__": app.run("0.0.0.0", 5001, debug=True) ``` 这个是.env 的环境变量： ``` FLASK_SECRET_KEY=replace-with-a-random-32-bytes-string OIDC_ISSUER=https://casdoor.name.heiyu.space OIDC_CLIENT_ID= OIDC_CLIENT_SECRET= OIDC_REDIRECT_URI=http://localhost:5001/auth/callback ``` 下面是代码解读： Flask 应用 = OIDC 客户端（RP）；Casdoor = 身份提供方（IdP）。 - `/login`：把浏览器重定向到 IdP 的 **授权端点**。 - `/callback`：IdP 回调携带 `code` → 后端用 `code`去 **令牌端点** 换 `access_token` + `id_token` → 用 **JWKS 公钥**校验 `id_token` → 用 `access_token` 拉 **userinfo**。 - `/profile`：展示从 userinfo/ID Token 得到的用户信息。 - `/logout`：清空本地会话。 #### 依赖与配置 ```python import os, requests, jwt from urllib.parse import urlencode from flask import Flask, redirect, request, session, url_for, jsonify, abort from dotenv import load_dotenv from jwt import PyJWKClient ``` - `requests`：调 OIDC 的 HTTP 端点。 - `PyJWT` + `cryptography`：验证 `id_token` 的数字签名。 - `PyJWKClient`：根据 JWT 头里的 `kid`，自动从 `jwks_uri` 拉对应公钥。 - `urlencode`：把授权请求的参数拼到 URL 上（避免手写字符串拼接出错）。 ```python load_dotenv() app = Flask(__name__) app.secret_key = os.getenv("FLASK_SECRET_KEY", "dev-secret") ISSUER = os.getenv("OIDC_ISSUER") CLIENT_ID = os.getenv("OIDC_CLIENT_ID") CLIENT_SECRET = os.getenv("OIDC_CLIENT_SECRET") REDIRECT_URI = os.getenv("OIDC_REDIRECT_URI") ``` - 从 `.env` 读取 Issuer / Client / Secret / Redirect URI，**和 Casdoor 后台登记的必须完全一致**（协议、域名、端口、路径一字不差）。 --- #### 通过 Discovery 自动找端点 ```python discovery = requests.get(f"{ISSUER}/.well-known/openid-configuration").json() AUTH_EP = discovery["authorization_endpoint"] TOKEN_EP = discovery["token_endpoint"] USERINFO_EP = discovery["userinfo_endpoint"] JWKS_URI = discovery["jwks_uri"] ``` - OIDC 规范要求 IdP 暴露**发现文档**，里面告诉你：授权端点、令牌端点、用户信息端点、JWKS 公钥地址等。 - 这么做的好处：**不写死 URL**，换 IdP/升级版本也不怕路径差异。 #### 首页 & 登录 ```python @app.route("/") def index(): if "user" in session: return f"欢迎 {session['user'].get('name') or session['user']['sub']} <br><a href='/logout'>退出</a>" return "<a href='/login'>使用 Casdoor 登录</a>" ``` - 简单展示：有会话就显示用户名，否则给一个“登录”链接。   ```python @app.route("/login") def login(): params = { "client_id": CLIENT_ID, "response_type": "code", # 标准授权码流程 "scope": "openid profile email", # 至少要有 openid；加 profile/email 便于拿到名字/邮箱 "redirect_uri": REDIRECT_URI, "state": "xyz123", # 防 CSRF（下面会给“随机+校验”的升级版） "nonce": "abc456" # 防重放（也建议随机+校验） } return redirect(f"{AUTH_EP}?{urlencode(params)}") ``` - **state**：浏览器去 IdP 再回来时要原样带回；用于确认这真是你发起的请求（防 CSRF）。 - **nonce**：IdP 会把它放进 `id_token`，回调后核对一致（防重放/混淆响应）。 > Demo 为了短小，先写了固定值。**写文章时要强调：生产必须随机、并在回调里校验**。 ##### 回调：换令牌 → 验签 ID Token → 拉用户信息 ```python @app.route("/auth/callback") def callback(): if "error" in request.args: return f"Error: {request.args['error']}" code = request.args.get("code") if not code: abort(400, "Missing code") ``` - IdP 会带着 `?code=...&state=...` 回来。这里先取出 `code`，并处理错误场景（用户取消授权等）。 ```python # 1) 用 code 换 token token_resp = requests.post(TOKEN_EP, data={ "grant_type": "authorization_code", "code": code, "redirect_uri": REDIRECT_URI, "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, }).json() id_token = token_resp.get("id_token") access_token = token_resp.get("access_token") ``` - 这一步是**服务端到 IdP**的 POST。两种常见做法： - 把 `client_id/client_secret` 放表单（本例）； - 或用 HTTP Basic（IdP 要求不同，文章里可顺带提一嘴）。 - 返回里关键是 `id_token`（JWT，证明“你是谁”）和 `access_token`（调用 `userinfo` 的 Bearer Token）。 ```python # 2) 验证 id_token（必须做） jwks_client = PyJWKClient(JWKS_URI) signing_key = jwks_client.get_signing_key_from_jwt(id_token) claims = jwt.decode( id_token, signing_key.key, algorithms=["RS256"], # Casdoor 也可能配 ES256/ECDSA，按实际配置来 audience=CLIENT_ID, # aud 必须包含你的 client_id issuer=ISSUER, # iss 必须等于你的 Issuer ) ``` - **为什么一定要验签？**否则任何人都可以伪造一个 “自称由 IdP 签发”的 JWT。 - `PyJWKClient` 会读 JWT 头部的 `kid`，去 `JWKS_URI` 拉这把钥匙的公钥。 - 同时校验标准字段：`iss/aud/exp/iat/...`。如果你还用了 `nonce`，建议对 `claims["nonce"]` 做一致性校验。 ```python # 3) 拉取用户信息 userinfo = requests.get( USERINFO_EP, headers={"Authorization": f"Bearer {access_token}"} ).json() ``` - 不是所有 IdP 都在 `id_token` 里带全资料，所以通常再拉一次 `userinfo`。 - 这一步需要前面的 `access_token`。 ```python # 4) 缩小会话，只存最小字段（避免 Cookie > 4KB） session["user"] = { "sub": claims["sub"], # 唯一ID "name": userinfo.get("name", claims.get("name")), # 没有就退回到ID Token "email": userinfo.get("email", claims.get("email")), } return redirect(url_for("profile")) ``` - **强烈建议**：只把少量字段放进 session（Flask 默认把 session 放加密 Cookie，4KB 有上限）。不要把整个 token/JWT/raw 塞进去。 ##### 查看资料 & 退出 ```python @app.route("/profile") def profile(): if "user" not in session: return redirect("/") return jsonify(session["user"]) ``` - 只从会话里取“精简后的用户档案”返回给前端。  ```python @app.route("/logout") def logout(): session.clear() return redirect("/") ``` - 清本地会话即可。若要**单点登出**，可以再调用 IdP 的 `end_session_endpoint`（用 Discovery 取到）并带 `post_logout_redirect_uri`。  ### 小结：懒猫微服 + Casdoor OIDC 通过这次实战，我们完成了一个从 Casdoor 配置 到 应用端代码实现 的 OIDC 流程。 这说明在懒猫微服中，大家不仅可以直接用内置的 OIDC 功能，还可以自由选择商店里的 OIDC Provider（如 Casdoor） 来扩展身份认证能力。 懒猫微服不仅能用自带的 OIDC，更能灵活调用商店里的 Casdoor 等 Provider，满足更灵活的认证与单点登录需求。