设计素材网站永久苏州网站建设官网
前言
之前写了一篇关于chanlit如何实现登录授权文章 《Chainlit 实现自定Oauth2.0登录完整代码图文教程》 ，文章以为接入gitee OAuth授权为例，图文并茂介绍如何操作，并附带了完整代码示例。这次我们实现更有难度的事情，使用chanlit实现CAS中央用户登录授权，下面是完整教程。
教程
创建一个chainlit_sso的python项目
在项目文件下新建一个requirements.txt 依赖管理文件
配置如下：
chainlit~=2.2.1
requests
fastapi~=0.115.11
httpx~=0.28.1
starlette~=0.41.3
在命令行里输入安装依赖命令
(.venv) PS D:\PycharmProjects\chainlit_oauth> pip install -r .\requirements.txt
在项目文件下新建一个chainlit_app.py文件
代码如下：
from typing import Optional, Dict
from fastapi import Request, Responseimport chainlit as cl@cl.on_chat_start
async def on_chat_start():chat_profile = cl.user_session.get("chat_profile")print(chat_profile)pass@cl.oauth_callback
def oauth_callback(provider_id: str,token: str,raw_user_data: Dict[str, str],default_user: cl.User,
) -> Optional[cl.User]:return None@cl.on_logout
def login_out(provider_id: str,token: str,raw_user_data: Dict[str, str],default_user: cl.User,
) -> Optional[cl.User]:return None@cl.password_auth_callback
def auth_callback(username: str, password: str):# Fetch the user matching username from your database# and compare the hashed password with the value stored in the databaseif (username, password) == ("admin", "admin"):return cl.User(identifier="admin", metadata={"role": "admin", "provider": "credentials"})else:return None@cl.on_logout
def main(request: Request, response: Response):response.delete_cookie("access_token")- 代码中去掉
@cl.password_auth_callback注解 ，可以实现只有cas授权登录的方式
在项目文件下新建一个main.py文件
代码如下：
import os
import xml.etree.ElementTree as ETimport chainlit as cl
import httpx
from chainlit.auth import create_jwt
from chainlit.logger import logger
from chainlit.utils import mount_chainlit
from fastapi import FastAPI, Request, status, APIRouter
from fastapi.responses import RedirectResponseapp = FastAPI()
router = APIRouter()# 配置信息
CAS_LOGIN_URL = os.environ.get("CAS_LOGIN_URL")
CAS_VALIDATE_URL = os.environ.get("CAS_VALIDATE_URL")
SERVICE_URL = os.environ.get("SERVICE_URL")async def parse_cas_response(xml_str):"""解析CAS认证返回的XML响应返回包含用户信息和属性的字典"""namespaces = {'cas': 'http://www.yale.edu/tp/cas'}try:root = ET.fromstring(xml_str)except ET.ParseError as e:raise ValueError(f"XML解析失败: {str(e)}")# 查找认证成功节点auth_success = root.find('cas:authenticationSuccess', namespaces)if not auth_success:raise ValueError("CAS响应不包含认证成功信息")# 解析用户信息user_element = auth_success.find('cas:user', namespaces)if user_element is None:raise ValueError("缺少必要的用户字段")# 解析属性字段attributes = auth_success.find('cas:attributes', namespaces)if attributes:user_attribute = attributes.find('cas:userName', namespaces)return user_attribute.text.strip()return Noneasync def verify_ticket(ticket):if ticket is None:return False# 验证CAS Ticketasync with httpx.AsyncClient() as client:response = await client.get(CAS_VALIDATE_URL,params={"service": SERVICE_URL, "ticket": ticket})if response.status_code == 200:userName = await parse_cas_response(response.content)# 成功验证，可以在这里处理登录成功的逻辑return True, userNameelse:# 验证失败logger.error("Ticket verification failed")logger.error(response.content)return False, None@app.get("/sso/login")
async def cas_login(request: Request):cookies = request.cookiesaccess_token = cookies.get('access_token')logger.info('SERVICE_URL: %s', SERVICE_URL)logger.info('access_token %s', access_token)return RedirectResponse(url=f"{CAS_LOGIN_URL}?service={SERVICE_URL}")@app.get("/callback")
async def cas_callback(ticket: str = None):isOk, userName = await verify_ticket(ticket)if isOk:# 创建一个RedirectResponse实例redirect_response = RedirectResponse(url="/")access_token = create_jwt(cl.User(identifier=userName))# 使用set_cookie方法写入access_token到客户端的cookie中redirect_response.set_cookie(key="access_token", value=access_token) # 注意：这里应该是一个有效的token而非硬编码值return redirect_responseelse:return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)mount_chainlit(app=app, target="chainlit_app.py", path="/")在项目文件下新建一个.env环境变量文件
配置如下：
CHAINLIT_AUTH_SECRET="Wk92J%obBJneySV6x@HQo6W1by7/t,Bl42CJGm%12TIOzVkK1NIby?Phvrn25.qB"OAUTH_KEYCLOAK_CLIENT_ID="KEYCLOAK"
OAUTH_KEYCLOAK_CLIENT_SECRET="KEYCLOAK"
OAUTH_KEYCLOAK_REALM="KEYCLOAK"
OAUTH_KEYCLOAK_BASE_URL="KEYCLOAK"
OAUTH_KEYCLOAK_NAME="用户统一登陆"CAS_LOGIN_URL = "http://www.unityserver.cn/login"
CAS_VALIDATE_URL = "http://www.unityserver.cn/serviceValidate"
SERVICE_URL = "http://127.0.0.1:3000/callback"
CAS_LOGIN_URL用户统一登录地址CAS_VALIDATE_URL用户CAS验证地址SERVICE_URL服务回调地址OAUTH_KEYCLOAK_NAME复用了chainlit自带的KEYCLOAK用来自定义显示名称
修改chainlit前端源码
- 下载
chainlit源码，在编辑器里打开
- 依次点击 frontend>src>pages>Login.tsx
- 修改代码
# 修改前代码onOAuthSignIn={async (provider: string) => {window.location.href = apiClient.getOAuthEndpoint(provider);}}
# 修改后代码
onOAuthSignIn={async (provider: string) => {window.location.href = apiClient.getSsoEndpoint(provider);}}# 新增方法getSsoEndpoint(provider: string) {console.log(provider)return this.buildEndpoint(`/sso/login`);}
- 打包编译
执行以下命令，进行前端编译
PS D:\PycharmProjects\chainlit-2.2.1> npm run build
-
编译后，在
frontend文件夹下找到dist文件夹,拷贝到chaint_sso项目的public文件夹下，没有就新建。
-
chaint_sso项目下找到.chainlit文件夹下的config.toml文件修改配置，项目首次启动会生成这些文件，修改配置
custom_build = "./public/dist"
启动运行
执行一下命令运行
uvicorn main:app --host 0.0.0.0 --port 3000
- 可以自定义启动端口
效果展示
从git上获取的用户名和头像
