当前位置：首页 > news >正文

Dify-Token 应用实现

news 2025/11/16 5:13:53

源码github地址： dify github 地址

仅记录学习打卡，代码分析来自AI.

认证与安全机制

Cookie管理

# 控制台认证
COOKIE_NAME_ACCESS_TOKEN = "access_token"
COOKIE_NAME_REFRESH_TOKEN = "refresh_token" 
COOKIE_NAME_CSRF_TOKEN = "csrf_token"# Web应用认证
COOKIE_NAME_WEBAPP_ACCESS_TOKEN = "webapp_access_token"
COOKIE_NAME_PASSPORT = "passport"

CSRF防护

HEADER_NAME_CSRF_TOKEN = "X-CSRF-Token"
HEADER_NAME_APP_CODE = "X-App-Code"
HEADER_NAME_PASSPORT = "X-App-Passport"

整体架构设计

这是一个双令牌认证系统（Access Token + Refresh Token），结合了 CSRF 防护和多端认证机制。

核心令牌类型

1. Access Token（访问令牌）

def extract_access_token(request: Request) -> str | None:def _try_extract_from_cookie(request: Request) -> str | None:return request.cookies.get(_real_cookie_name(COOKIE_NAME_ACCESS_TOKEN))return _try_extract_from_cookie(request) or _try_extract_from_header(request)

特点：

双重提取策略：优先从 Cookie 提取，其次从 Authorization Header
短生命周期：ACCESS_TOKEN_EXPIRE_MINUTES 配置控制
安全存储：HttpOnly Cookie 防止 XSS 攻击

2. Refresh Token（刷新令牌）

def extract_refresh_token(request: Request) -> str | None:return request.cookies.get(_real_cookie_name(COOKIE_NAME_REFRESH_TOKEN))

特点：

仅 Cookie 存储：不通过 Header 传输
长生命周期：REFRESH_TOKEN_EXPIRE_DAYS 配置控制
专门用途：用于获取新的 Access Token

3. CSRF Token（跨站请求伪造防护令牌）

def extract_csrf_token(request: Request) -> str | None:return request.headers.get(HEADER_NAME_CSRF_TOKEN)def extract_csrf_token_from_cookie(request: Request) -> str | None:return request.cookies.get(_real_cookie_name(COOKIE_NAME_CSRF_TOKEN))

双重验证机制：

Cookie 存储：COOKIE_NAME_CSRF_TOKEN
Header 传输：X-CSRF-Token
比对验证：两者必须一致

安全实现细节

1. Cookie 安全配置

def set_access_token_to_cookie(request: Request, response: Response, token: str, samesite: str = "Lax"):response.set_cookie(_real_cookie_name(COOKIE_NAME_ACCESS_TOKEN),value=token,httponly=True,      # 防XSSdomain=_cookie_domain(),secure=is_secure(), # HTTPS onlysamesite=samesite,  # CSRF防护max_age=...,path="/",)

安全特性：

HttpOnly：JavaScript 无法访问
Secure：仅 HTTPS 传输
SameSite：限制跨站请求
域名控制：可配置的 Cookie 域

2. 动态 Cookie 命名

def _real_cookie_name(cookie_name: str) -> str:if is_secure() and _cookie_domain() is None:return "__Host-" + cookie_name  # 增强安全前缀else:return cookie_name

使用 __Host- 前缀提供额外的安全保护。

3. CSRF 白名单机制

CSRF_WHITE_LIST = [re.compile(r"/console/api/apps/[a-f0-9-]+/workflows/draft"),
]def check_csrf_token(request: Request, user_id: str):for pattern in CSRF_WHITE_LIST:if pattern.match(request.path):  # 白名单路径跳过CSRF检查return# ... CSRF验证逻辑

多端认证支持

1. 控制台认证（Console）

# 用于管理后台
COOKIE_NAME_ACCESS_TOKEN = "access_token"
COOKIE_NAME_REFRESH_TOKEN = "refresh_token"

2. WebApp 认证（终端用户）

# 用于面向用户的应用
COOKIE_NAME_WEBAPP_ACCESS_TOKEN = "webapp_access_token"
COOKIE_NAME_PASSPORT = "passport"

3. Passport 系统（应用级认证）

def extract_webapp_passport(app_code: str, request: Request) -> str | None:def _try_extract_passport_token_from_cookie(request: Request) -> str | None:return request.cookies.get(_real_cookie_name(COOKIE_NAME_PASSPORT + "-" + app_code))

特点：按应用代码动态生成 Cookie 名称，支持多应用隔离。

令牌验证流程

CSRF 令牌验证

def check_csrf_token(request: Request, user_id: str):csrf_token = extract_csrf_token(request)           # 从Header获取csrf_token_from_cookie = extract_csrf_token_from_cookie(request)  # 从Cookie获取# 1. 比对一致性if csrf_token != csrf_token_from_cookie:_unauthorized()# 2. 验证JWT签名和有效期verified = PassportService().verify(csrf_token)# 3. 验证用户身份if verified.get("sub") != user_id:_unauthorized()# 4. 验证过期时间exp: int | None = verified.get("exp")if not exp or exp < datetime.now().timestamp():_unauthorized()

令牌生成机制

CSRF 令牌生成

def generate_csrf_token(user_id: str) -> str:exp_dt = datetime.now(UTC) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)payload = {"exp": int(exp_dt.timestamp()),  # 过期时间"sub": user_id,                  # 用户标识}return PassportService().issue(payload)  # JWT签发

会话管理

1. 令牌清除

def clear_access_token_from_cookie(response: Response, samesite: str = "Lax"):_clear_cookie(response, COOKIE_NAME_ACCESS_TOKEN, samesite)def build_force_logout_cookie_headers() -> list[str]:"""强制登出：清除所有认证相关的Cookie"""response = Response()clear_access_token_from_cookie(response)clear_csrf_token_from_cookie(response)clear_refresh_token_from_cookie(response)return response.headers.getlist("Set-Cookie")