身份认证踩坑记录

作者: 醉月思发布日期: 2021-11-08 分类: 技术实践

这次做身份认证改造，从 Session 到 JWT，再到 OAuth 2.0，。

Session 认证

基础 Session

from flask import Flask, session, request
from functools import wraps

app = Flask(__name__)
app.secret_key = 'your-secret-key'

def login_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_id' not in session:
            return {'error': 'Unauthorized'}, 401
        return f(*args, **kwargs)
    return decorated_function

@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    
    # 验证用户
    user = authenticate_user(username, password)
    if not user:
        return {'error': 'Invalid credentials'}, 401
    
    # 设置 Session
    session['user_id'] = user.id
    session['username'] = user.username
    
    return {'message': 'Logged in successfully'}

@app.route('/profile')
@login_required
def profile():
    return {'user_id': session['user_id'], 'username': session['username']}

@app.route('/logout')
def logout():
    session.clear()
    return {'message': 'Logged out'}

Session 存储

from flask_session import Session
import redis

# Redis 存储 Session
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.from_url('redis://localhost:6379/0')

Session(app)

# MySQL 存储 Session
app.config['SESSION_TYPE'] = 'sqlalchemy'
app.config['SESSION_SQLALCHEMY'] = db
app.config['SESSION_SQLALCHEMY_TABLE'] = 'sessions'

Session(app)

JWT 认证

基础 JWT

import jwt
from datetime import datetime, timedelta
from functools import wraps

SECRET_KEY = 'your-secret-key'
ALGORITHM = 'HS256'

def create_jwt_token(user_id, expires_in_hours=24):
    """创建 JWT Token"""
    expire = datetime.utcnow() + timedelta(hours=expires_in_hours)
    
    payload = {
        'user_id': user_id,
        'exp': expire,
        'iat': datetime.utcnow()
    }
    
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token

def verify_jwt_token(token):
    """验证 JWT Token"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        return {'error': 'Token has expired'}
    except jwt.InvalidTokenError:
        return {'error': 'Invalid token'}

def jwt_required(f):
    """JWT 认证装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        token = request.headers.get('Authorization')
        
        if not token:
            return {'error': 'Missing token'}, 401
        
        if token.startswith('Bearer '):
            token = token[7:]
        
        payload = verify_jwt_token(token)
        
        if 'error' in payload:
            return payload, 401
        
        request.user_id = payload['user_id']
        return f(*args, **kwargs)
    
    return decorated_function

# 使用
@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    
    user = authenticate_user(username, password)
    if not user:
        return {'error': 'Invalid credentials'}, 401
    
    token = create_jwt_token(user.id)
    return {'token': token}

@app.route('/profile')
@jwt_required
def profile():
    return {'user_id': request.user_id}

JWT 刷新令牌

def create_refresh_token(user_id):
    """创建刷新令牌"""
    expire = datetime.utcnow() + timedelta(days=7)
    
    payload = {
        'user_id': user_id,
        'exp': expire,
        'iat': datetime.utcnow(),
        'type': 'refresh'
    }
    
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token

def refresh_access_token(refresh_token):
    """刷新访问令牌"""
    payload = verify_jwt_token(refresh_token)
    
    if 'error' in payload:
        return payload
    
    if payload.get('type') != 'refresh':
        return {'error': 'Invalid refresh token'}
    
    user_id = payload['user_id']
    access_token = create_jwt_token(user_id)
    
    return {'access_token': access_token}

# 使用
@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    
    user = authenticate_user(username, password)
    if not user:
        return {'error': 'Invalid credentials'}, 401
    
    access_token = create_jwt_token(user.id)
    refresh_token = create_refresh_token(user.id)
    
    return {
        'access_token': access_token,
        'refresh_token': refresh_token
    }

@app.route('/refresh', methods=['POST'])
def refresh():
    refresh_token = request.json.get('refresh_token')
    
    if not refresh_token:
        return {'error': 'Missing refresh token'}, 401
    
    result = refresh_access_token(refresh_token)
    
    if 'error' in result:
        return result, 401
    
    return result

OAuth 2.0

授权码模式

from authlib.integrations.flask_client import OAuth

app = Flask(__name__)
app.secret_key = 'your-secret-key'

oauth = OAuth(app)

# 配置 OAuth
oauth.register(
    name='google',
    client_id='your-google-client-id',
    client_secret='your-google-client-secret',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={
        'scope': 'openid email profile'
    }
)

@app.route('/login/google')
def login_google():
    redirect_uri = url_for('authorize', _external=True)
    return oauth.google.authorize_redirect(redirect_uri)

@app.route('/authorize')
def authorize():
    token = oauth.google.authorize_access_token()
    user = oauth.google.parse_id_token(token)
    
    # 处理用户信息
    user_info = user.get('userinfo')
    
    return {
        'user_id': user_info.get('sub'),
        'name': user_info.get('name'),
        'email': user_info.get('email')
    }

客户端凭证模式

@app.route('/token', methods=['POST'])
def get_token():
    client_id = request.form.get('client_id')
    client_secret = request.form.get('client_secret')
    
    # 验证客户端
    client = authenticate_client(client_id, client_secret)
    if not client:
        return {'error': 'Invalid client credentials'}, 401
    
    # 生成 Token
    token = create_jwt_token(client.id, expires_in_hours=1)
    
    return {
        'access_token': token,
        'token_type': 'Bearer',
        'expires_in': 3600
    }

权限控制

RBAC（基于角色的访问控制）

class Permission:
    READ = 'read'
    WRITE = 'write'
    DELETE = 'delete'
    ADMIN = 'admin'

class Role:
    USER = 'user'
    MODERATOR = 'moderator'
    ADMIN = 'admin'

ROLE_PERMISSIONS = {
    Role.USER: [Permission.READ],
    Role.MODERATOR: [Permission.READ, Permission.WRITE],
    Role.ADMIN: [Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN]
}

def require_permission(permission):
    """权限检查装饰器"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if 'user_id' not in session:
                return {'error': 'Unauthorized'}, 401
            
            user = get_user(session['user_id'])
            user_role = user.role
            role_permissions = ROLE_PERMISSIONS.get(user_role, [])
            
            if permission not in role_permissions:
                return {'error': 'Forbidden'}, 403
            
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# 使用
@app.route('/admin')
@login_required
@require_permission(Permission.ADMIN)
def admin():
    return {'message': 'Admin dashboard'}

ABAC（基于属性的访问控制）

class ABAC:
    def __init__(self):
        self.policies = []
    
    def add_policy(self, policy):
        self.policies.append(policy)
    
    def check_access(self, user, resource, action):
        """检查访问权限"""
        for policy in self.policies:
            if self._evaluate_policy(policy, user, resource, action):
                return True
        return False
    
    def _evaluate_policy(self, policy, user, resource, action):
        """评估策略"""
        # 实现策略评估逻辑
        return True

# 使用
abac = ABAC()

# 添加策略
abac.add_policy({
    'name': 'document_read_policy',
    'conditions': {
        'user.role': 'user',
        'resource.type': 'document',
        'action': 'read'
    }
})

# 检查权限
if abac.check_access(user, 'document', 'read'):
    return document

踩过的坑

坑一：Session 固定攻击

攻击者固定用户的 Session ID。

解决：登录后重新生成 Session ID。

@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    
    user = authenticate_user(username, password)
    if not user:
        return {'error': 'Invalid credentials'}, 401
    
    # 重新生成 Session ID
    session.regenerate()
    
    session['user_id'] = user.id
    
    return {'message': 'Logged in successfully'}

坑二：JWT 泄露

JWT 被泄露，攻击者可以冒充用户。

解决：使用短期 Token + 刷新 Token。

# 短期访问令牌（1 小时）
access_token = create_jwt_token(user.id, expires_in_hours=1)

# 长期刷新令牌（7 天）
refresh_token = create_refresh_token(user.id)

坑三：CSRF 攻击

攻击者伪造用户请求。

解决：使用 CSRF Token。

from flask_wtf.csrf import CSRFProtect

app = Flask(__name__)
app.secret_key = 'your-secret-key'
csrf = CSRFProtect(app)

@app.route('/api/data', methods=['POST'])
@login_required
def get_data():
    # CSRF 保护自动启用
    return {'data': '...'}

坑四：OAuth 配置错误

OAuth 配置错误，导致认证失败。

解决：仔细配置 OAuth 参数。

# 正确的 OAuth 配置
oauth.register(
    name='google',
    client_id='your-google-client-id',
    client_secret='your-google-client-secret',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={
        'scope': 'openid email profile',
        'redirect_uri': 'https://your-domain.com/authorize'
    },
    redirect_uri='https://your-domain.com/authorize'
)

最佳实践

安全措施

# 1. 使用 HTTPS
# 2. 设置合理的 Token 过期时间
# 3. 使用强密钥
# 4. 定期轮换密钥
# 5. 实施速率限制
# 6. 记录认证事件
# 7. 实施多因素认证

# 使用强密钥
import secrets
SECRET_KEY = secrets.token_hex(32)

# 设置合理的过期时间
def create_jwt_token(user_id):
    expire = datetime.utcnow() + timedelta(hours=1)
    
    payload = {
        'user_id': user_id,
        'exp': expire,
        'iat': datetime.utcnow()
    }
    
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

# 记录认证事件
def log_auth_event(event_type, user_id, details):
    logger.info({
        'event_type': event_type,
        'user_id': user_id,
        'details': details,
        'timestamp': datetime.utcnow()
    })

用户管理

class UserManager:
    def __init__(self, db):
        self.db = db
    
    def create_user(self, username, password, email):
        """创建用户"""
        # 密码哈希
        hashed_password = hash_password(password)
        
        # 创建用户
        user = User(
            username=username,
            password=hashed_password,
            email=email
        )
        
        self.db.add(user)
        self.db.commit()
        
        return user
    
    def update_password(self, user_id, old_password, new_password):
        """更新密码"""
        user = self.db.query(User).get(user_id)
        
        if not verify_password(old_password, user.password):
            return False
        
        user.password = hash_password(new_password)
        self.db.commit()
        
        return True
    
    def reset_password(self, user_id, new_password):
        """重置密码（管理员功能）"""
        user = self.db.query(User).get(user_id)
        user.password = hash_password(new_password)
        self.db.commit()
        
        return True

写在最后

身份认证这东西，不只是技术问题，是安全问题和用户体验问题。

解决了：

用户认证
权限控制
安全保护

带来了：

复杂度增加
用户体验变化
运维成本

实施之前先评估：

安全要求
用户体验
技术能力
预算

不是所有场景都需要复杂的认证，有时候简单的 Session 就够用。

这次身份认证改造花了一个月，从 Session 到 JWT，再到 OAuth 2.0。改造完成后，安全事件减少了 90%，用户体验也有提升。

版权声明: 本文首发于指尖魔法屋-身份认证踩坑记录（https://blog.thinkmoon.cn/post/86-authentication-session-jwt-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

身份认证踩坑记录

Session 认证

基础 Session

Session 存储

JWT 认证

基础 JWT

JWT 刷新令牌

OAuth 2.0

授权码模式

客户端凭证模式

权限控制

RBAC（基于角色的访问控制）

ABAC（基于属性的访问控制）

踩过的坑

坑一：Session 固定攻击

坑二：JWT 泄露

坑三：CSRF 攻击

坑四：OAuth 配置错误

最佳实践

安全措施

用户管理

写在最后

评论

相关阅读