单点登录原理详解
2026/3/20大约 11 分钟
单点登录原理详解
什么是单点登录
基本概念
单点登录(Single Sign-On,简称 SSO)是一种身份认证机制,允许用户使用一组凭证(用户名/密码)登录一次,就能访问多个相互信任的应用系统,无需重复认证。
SSO 的核心术语
| 术语 | 全称 | 说明 |
|---|---|---|
| IdP | Identity Provider | 身份提供者,负责认证用户身份(如 SSO 服务器) |
| SP | Service Provider | 服务提供者,依赖 IdP 进行用户认证的应用系统 |
| Principal | - | 被认证的主体,通常指用户 |
| Credential | - | 凭证,用于证明身份的信息(密码、Token 等) |
| Assertion | - | 断言,IdP 发给 SP 的身份声明 |
为什么需要单点登录
没有 SSO 的痛点:
SSO 带来的价值:
| 维度 | 传统方式 | SSO 方式 |
|---|---|---|
| 用户体验 | 多次登录,体验割裂 | 一次登录,处处通行 |
| 安全性 | 密码分散,难以统一管控 | 集中认证,统一安全策略 |
| 运维成本 | 账号分散,管理复杂 | 集中管理,降低成本 |
| 密码管理 | 多个弱密码 | 一个强密码 + MFA |
| 审计追踪 | 日志分散,难以追溯 | 统一审计,完整追踪 |
SSO 核心原理
同域 SSO(最简单)
当所有应用都在同一个父域下时,可以通过共享 Cookie 实现 SSO:
from flask import Flask, request, make_response, redirect, jsonify
import jwt
import time
app = Flask(__name__)
SECRET_KEY = "your-secret-key"
DOMAIN = ".company.com"
@app.route("/sso/login", methods=["POST"])
def sso_login():
"""SSO 认证中心登录"""
data = request.get_json()
username = data.get("username")
password = data.get("password")
# 验证用户凭证(实际应查询数据库)
if username == "admin" and password == "admin123":
# 生成全局会话 Token
token = jwt.encode({
"user_id": 1,
"username": username,
"exp": time.time() + 3600
}, SECRET_KEY, algorithm="HS256")
response = make_response(jsonify({"message": "登录成功"}))
# 关键:设置 Cookie 到父域
response.set_cookie(
key="sso_token",
value=token,
domain=DOMAIN, # 父域,所有子域可访问
max_age=3600,
httponly=True,
secure=True,
samesite="Lax"
)
return response
return jsonify({"error": "认证失败"}), 401
@app.route("/sso/verify")
def verify_token():
"""验证 Token(供各子系统调用)"""
token = request.cookies.get("sso_token")
if not token:
return jsonify({"valid": False, "error": "无Token"}), 401
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return jsonify({
"valid": True,
"user_id": payload["user_id"],
"username": payload["username"]
})
except jwt.ExpiredSignatureError:
return jsonify({"valid": False, "error": "Token已过期"}), 401
except jwt.InvalidTokenError:
return jsonify({"valid": False, "error": "无效Token"}), 401
@app.route("/sso/logout", methods=["POST"])
def sso_logout():
"""统一登出"""
response = make_response(jsonify({"message": "登出成功"}))
response.delete_cookie("sso_token", domain=DOMAIN)
return response
跨域 SSO(主流方案)
当应用分布在不同域名下,Cookie 无法直接共享,需要更复杂的机制:
SSO 登录流程详解
完整登录流程(CAS 风格)
核心概念解释
已登录用户访问新应用
当用户已经在 SSO 登录过,访问其他应用时的流程:
SSO 登出机制
单点登出的挑战
单点登出(Single Logout,SLO)比单点登录更复杂,需要协调多个系统:
单点登出流程
登出通知方式
| 方式 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| 后端通道 | SSO 服务端直接调用各 SP 的登出接口 | 可靠性高 | SP 需暴露接口 |
| 前端通道 | 通过重定向或 iframe 访问各 SP 登出 URL | 实现简单 | 可能被浏览器拦截 |
| 被动检测 | SP 定期向 SSO 验证会话有效性 | 无需推送 | 有延迟 |
import requests
from typing import List
from concurrent.futures import ThreadPoolExecutor
class SSOLogoutService:
"""SSO 登出服务"""
def __init__(self):
# 存储已登录的 SP 信息
# 实际应存储在 Redis 中
self.active_sessions = {} # {tgt_id: [{sp_url, session_id}]}
def register_sp_session(self, tgt_id: str, sp_url: str, session_id: str):
"""注册 SP 会话(SP 登录成功时调用)"""
if tgt_id not in self.active_sessions:
self.active_sessions[tgt_id] = []
self.active_sessions[tgt_id].append({
"sp_url": sp_url,
"session_id": session_id
})
def logout(self, tgt_id: str) -> dict:
"""执行单点登出"""
results = {"success": [], "failed": []}
# 获取所有需要登出的 SP
sessions = self.active_sessions.get(tgt_id, [])
if not sessions:
return results
# 并行通知所有 SP 登出
with ThreadPoolExecutor(max_workers=10) as executor:
futures = []
for session in sessions:
future = executor.submit(
self._notify_sp_logout,
session["sp_url"],
session["session_id"]
)
futures.append((session["sp_url"], future))
for sp_url, future in futures:
try:
if future.result():
results["success"].append(sp_url)
else:
results["failed"].append(sp_url)
except Exception as e:
results["failed"].append(sp_url)
# 清除 SSO 会话记录
del self.active_sessions[tgt_id]
return results
def _notify_sp_logout(self, sp_url: str, session_id: str) -> bool:
"""通知 SP 登出(后端通道)"""
try:
response = requests.post(
f"{sp_url}/sso/logout-callback",
json={"session_id": session_id},
timeout=5
)
return response.status_code == 200
except Exception:
return False
SSO 架构模式
代理模式(Proxy)
所有请求都经过 SSO 代理,由代理进行认证和转发:
代理模式(Agent)
在每个应用中嵌入 Agent,拦截请求进行认证:
from flask import Flask, request, redirect, g
from functools import wraps
import requests
app = Flask(__name__)
# SSO 配置
SSO_SERVER = "https://sso.company.com"
APP_URL = "https://app1.company.com"
class SSOAgent:
"""SSO Agent - 嵌入应用的认证代理"""
def __init__(self, sso_server: str, app_url: str):
self.sso_server = sso_server
self.app_url = app_url
def check_authentication(self, func):
"""认证检查装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
# 1. 检查本地会话
session_id = request.cookies.get("app_session")
if session_id and self._validate_local_session(session_id):
return func(*args, **kwargs)
# 2. 检查是否携带 ST(从 SSO 回调)
ticket = request.args.get("ticket")
if ticket:
user_info = self._validate_ticket(ticket)
if user_info:
# 创建本地会话
response = func(*args, **kwargs)
self._create_local_session(response, user_info)
return response
# 3. 重定向到 SSO 登录
login_url = f"{self.sso_server}/login?service={self.app_url}"
return redirect(login_url)
return wrapper
def _validate_local_session(self, session_id: str) -> bool:
"""验证本地会话"""
# 实际应查询 Session 存储
return False
def _validate_ticket(self, ticket: str) -> dict | None:
"""向 SSO 验证票据"""
try:
response = requests.get(
f"{self.sso_server}/validate",
params={"ticket": ticket, "service": self.app_url},
timeout=5
)
if response.status_code == 200:
return response.json()
except Exception:
pass
return None
def _create_local_session(self, response, user_info: dict):
"""创建本地会话"""
# 实际应存储到 Redis 并设置 Cookie
pass
# 使用示例
sso_agent = SSOAgent(SSO_SERVER, APP_URL)
@app.route("/dashboard")
@sso_agent.check_authentication
def dashboard():
return "欢迎访问仪表盘!"
令牌模式(Token)
基于 Token 的无状态 SSO,常与 OAuth2/OIDC 结合:
本章小结
核心要点
- SSO 本质:一处登录,处处通行,基于信任关系实现会话共享
- 票据体系:TGT(全局会话)→ ST(服务票据)→ 应用会话
- 跨域难点:Cookie 无法跨域,需要重定向 + 票据机制
- 登出复杂:需要协调多个系统,保证一致性