# ❌ 全是错的
hashlib.md5(password.encode()).hexdigest()
hashlib.sha1(password.encode()).hexdigest()

# ❌ 自己造的"加密"（彩虹表秒破）
encrypt(password, my_own_algorithm)

import bcrypt

def hash_password(password: str) -> bytes:
    # cost=12 大约需要 250ms，OWASP 推荐的量级
    salt = bcrypt.gensalt(rounds=12)
    return bcrypt.hashpw(password.encode('utf-8'), salt)


def verify_password(password: str, hashed: bytes) -> bool:
    try:
        return bcrypt.checkpw(password.encode('utf-8'), hashed)
    except ValueError:
        return False

def login(request):
    user = db.find_user_by_email(request.email)

    if not user or not verify_password(request.password, user.password_hash):
        # 统一提示，不告诉用户到底是邮箱不存在还是密码错
        raise AuthenticationError('邮箱或密码错误')

    session_id = secrets.token_urlsafe(32)    # 32 字节随机，足够安全
    session_data = {
        'user_id': user.id,
        'created_at': time.time(),
        'ip': request.remote_addr,
    }
    redis.setex(f'session:{session_id}', 86400, serialize(session_data))

    response.set_cookie(
        'session_id', session_id,
        httponly=True,      # JavaScript 读不到，防 XSS 窃取
        secure=True,        # 只走 HTTPS
        samesite='lax',     # 防跨站请求伪造
        path='/',
    )
    return {'ok': True}

import jwt
from datetime import datetime, timedelta, timezone

SECRET = os.environ['JWT_SECRET']
ALGORITHM = 'HS256'


def create_access_token(user_id: int, roles: list) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        'sub': str(user_id),
        'roles': roles,
        'iat': now,
        'exp': now + timedelta(minutes=15),      # 短有效期
        'jti': str(uuid.uuid4()),                 # 唯一 ID，用于撤销
        'iss': 'my-api',                          # 签发者
        'aud': 'my-app',                          # 接收方
    }
    return jwt.encode(payload, SECRET, algorithm=ALGORITHM)


def verify_token(token: str) -> dict:
    options = {
        'verify_signature': True,
        'verify_exp': True,
        'verify_iss': True,
        'verify_aud': True,
        'require': ['exp', 'iat', 'sub'],
    }
    return jwt.decode(
        token, SECRET, algorithms=[ALGORITHM],
        issuer='my-api', audience='my-app', options=options,
    )

# ❌ 致命漏洞：攻击者可以把 alg 改成 "none"，或改成对称算法
jwt.decode(token, key, algorithms=jwt.algorithms.get_default_algorithms())

# ✅ 显式指定允许的算法
jwt.decode(token, SECRET, algorithms=['HS256'])

def revoke(token: str):
    payload = jwt.decode(token, options={'verify_signature': False})
    # 黑名单 TTL 设为 token 的剩余寿命
    redis.setex(
        f"jwt:blacklist:{payload['jti']}",
        int(payload['exp'] - payload['iat']),
        '1',
    )


def verify_and_check_revoked(token: str) -> dict:
    payload = jwt.decode(token, SECRET, algorithms=['HS256'])

    if redis.exists(f"jwt:blacklist:{payload['jti']}"):
        raise AuthenticationError('token 已撤销')

    return payload

def refresh(refresh_token: str):
    data = redis.get(f'refresh:{hash(refresh_token)}')

    if not data:
        raise AuthenticationError('refresh token 无效')

    session = deserialize(data)

    # 检测重放：如果同一个 token 用过两次，说明可能被窃取
    used = redis.getset(f'refresh_used:{hash(refresh_token)}', '1', ex=86400)
    if used:
        # 撤销整个 token 家族
        revoke_family(session['family_id'])
        raise AuthenticationError('检测到 token 重放，已撤销全部会话')

    user = db.get(User, session['user_id'])
    return create_access_token(user.id, user.roles)

1. 用户点击「用 Google 登录」
2. 前端把用户重定向到授权服务器的授权端点，带上 client_id、redirect_uri、state
3. 用户在授权服务器上登录并同意授权
4. 授权服务器重定向回 redirect_uri，附带 ?code=授权码&state=...
5. 前端把 code 发送到自己的后端
6. 后端用 code + client_secret 换取 access_token
7. 后端用 access_token 调用资源服务器获取用户信息

# 发起授权前
state = secrets.token_urlsafe(24)
session['oauth_state'] = state
redirect_to_authorize(client_id, redirect_uri, state, scope)

# 回调时
if request.args.get('state') != session.pop('oauth_state'):
    raise AuthenticationError('state 校验失败，可能是 CSRF 攻击')

def refresh_tokens(session_id: str):
    session = get_session(session_id)
    token = session['refresh_token']

    resp = http.post(
        f"{PROVIDER}/oauth/token",
        data={
            'grant_type': 'refresh_token',
            'refresh_token': token,
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
        },
        timeout=10,
    )

    new_token = resp.json()['access_token']

    if 'refresh_token' in resp.json():
        session['refresh_token'] = resp.json()['refresh_token']

    session['access_token'] = new_token
    session['expires_at'] = time.time() + resp.json()['expires_in']

    save_session(session_id, session)
    return new_token

ROLES = {
    'user':    ['post:read', 'comment:create'],
    'editor':  ['post:read', 'post:write', 'comment:create', 'comment:delete'],
    'admin':   ['*'],
}

def can(user, permission: str) -> bool:
    perms = ROLES.get(user.role, [])
    return '*' in perms or permission in perms

def can_edit_post(user, post) -> bool:
    if user.is_admin:
        return True
    if post.author_id != user.id:
        return False                    # 不是自己的文章
    return post.status in ('draft', 'published')   # 已归档的不能改

# ❌ 检查了是不是作者，但没检查是不是自己的文章
def update_post(post_id):
    post = db.get(Post, post_id)
    if not can_edit_post(current_user, post):     # 作者判断
        raise Forbidden()

    # 但是！post_id 是用户传的，他可以传别人的 post_id
    # 如果 post 查询没带 author_id 条件，作者判断就形同虚设

def update_post(post_id):
    post = db.query_one(
        "SELECT * FROM posts WHERE id = :id AND author_id = :author",
        {'id': post_id, 'author_id': current_user.id},
    )
    if not post:
        raise NotFound()          # 不区分「不存在」和「无权限」

SENSITIVE = {'authorization', 'cookie', 'set-cookie', 'x-api-key'}

def scrub(log_record: dict) -> dict:
    return {
        k: ('***' if k.lower() in SENSITIVE else v)
        for k, v in log_record.items()
    }