文章

21 · 表单、会话与中间件原理

#020 · 2026-04-16 · Python
PythonPython教程进阶DjangoSessionCookie

🔗 知识图谱导航：阅读本文前，建议先掌握《20 · Django ORM：模型设计与数据迁移》中的模型层，以及《18 · HTTP 与 Web 框架基石》中的 Cookie 头格式——本文把这两块拼在一起，用零依赖代码演示 Session 管理和中间件链的完整工作原理。

运行环境：Python 3.12+ 标准库，零额外依赖，直接运行。

极客解析：Django 中间件的"魔法"本质上是函数组合：build_chain([mw1, mw2, mw3], view) 把中间件列表变成一个嵌套调用链，请求从外到内穿过，响应从内到外返回——这就是"洋葱模型"。

Cookie / Session 原理

  客户端                                          服务端
    │                                               │
    │──── POST /login {user, pass} ────────────────▶│
    │                                               │ 验证通过
    │                                               │ sid = secrets.token_hex(16)
    │                                               │ _sessions[sid] = {user: "alice"}
    │◀─── Set-Cookie: sessionid=sid; HttpOnly ──────│
    │                                               │
    │──── GET /me  Cookie: sessionid=sid ──────────▶│
    │                                               │ get_session(sid) → {user: "alice"}
    │◀─── 200 OK {"user": "alice"} ─────────────────│

中间件执行链（洋葱模型）

请求进入
    │
    ▼  logging_middleware（记录日志）
    │    ▼  timing_middleware（计时）
    │    │    ▼  auth_middleware（加载用户）
    │    │    │    ▼  router → view（业务逻辑）
    │    │    │    ◀  响应返回
    │    │    ◀  timing 记录耗时
    │    ◀  logging 打印日志
    ◀  响应返回客户端

安全要点速查

HttpOnly Cookie   JS 无法读取，防止 XSS 窃取 Session
Secure 标志       仅 HTTPS 传输，防止中间人截获
SameSite=Lax      限制跨站请求携带 Cookie，现代 CSRF 防护主力
CSRF Token        双重提交 Cookie 模式，服务端对比 Cookie 值与请求体 Token
Session 过期      服务端主动设置过期时间，不依赖客户端删除 Cookie

步步为营：核心逻辑自适应拆解

这一篇的核心是三件事：secrets.token_hex 生成不可预测的 Session ID → LoginForm 集中做字段校验 → build_chain 把中间件列表组合成洋葱调用链。下面每一步都聚焦一个机制，零依赖可直接运行。

Step 1：用 secrets.token_hex 实现内存 Session 管理

痛点与机制：

Session ID 必须不可预测——如果用 random.randint() 生成，攻击者可以暴力枚举猜出其他用户的 Session。secrets.token_hex(16) 生成 32 位密码学安全的随机十六进制字符串，暴力枚举的概率是 1/2^128，实际上不可能。get_session 检查过期时间（1小时），过期自动删除——服务端主动控制过期，不依赖客户端删除 Cookie。

核心源码（逐字来自文末完整源码）：

_sessions: Dict[str, Dict] = {}
_users: Dict[str, str] = {
    "alice": hashlib.sha256(b"pass123").hexdigest(),
    "bob":   hashlib.sha256(b"secret").hexdigest(),
}


def create_session(user: str) -> str:
    sid = secrets.token_hex(16)
    _sessions[sid] = {"user": user, "created_at": time.time()}
    return sid


def get_session(sid: str) -> Optional[Dict]:
    s = _sessions.get(sid)
    if s and time.time() - s["created_at"] < 3600:  # 1小时过期
        return s
    if s:
        del _sessions[sid]
    return None


def delete_session(sid: str) -> None:
    _sessions.pop(sid, None)


def parse_cookies(cookie_header: str) -> Dict[str, str]:
    cookies: Dict[str, str] = {}
    for part in cookie_header.split(";"):
        part = part.strip()
        if "=" in part:
            k, v = part.split("=", 1)
            cookies[k.strip()] = v.strip()
    return cookies

可运行演示（补齐 Mock 数据与 print 反馈）：

import hashlib
import secrets
import time
from typing import Dict, Optional

_sessions: Dict[str, Dict] = {}
_users: Dict[str, str] = {
    "alice": hashlib.sha256(b"pass123").hexdigest(),
    "bob":   hashlib.sha256(b"secret").hexdigest(),
}

def create_session(user: str) -> str:
    # secrets 生成不可预测的登录凭证，适合安全场景。
    sid = secrets.token_hex(16)
    _sessions[sid] = {"user": user, "created_at": time.time()}
    return sid

def get_session(sid: str) -> Optional[Dict]:
    s = _sessions.get(sid)
    if s and time.time() - s["created_at"] < 3600:
        return s
    if s:
        del _sessions[sid]
    return None

def delete_session(sid: str) -> None:
    _sessions.pop(sid, None)

def parse_cookies(cookie_header: str) -> Dict[str, str]:
    cookies: Dict[str, str] = {}
    for part in cookie_header.split(";"):
        part = part.strip()
        if "=" in part:
            k, v = part.split("=", 1)
            cookies[k.strip()] = v.strip()
    return cookies

sid = create_session("alice")
print("新建 session 前 8 位:", sid[:8])
print("服务端登记簿:", get_session(sid))
print("Cookie 解析:", parse_cookies(f"sessionid={sid}; theme=dark"))
delete_session(sid)
print("退出后再查:", get_session(sid))

Step 2：用 LoginForm 集中做字段校验，errors 字典收集所有错误

痛点与机制：

LoginForm.is_valid() 是"一次性体检"：把所有字段的校验逻辑集中在一个方法里，errors 字典收集所有错误，调用方只需检查 is_valid() 返回值，不需要写一堆 if/elif。这正是 Django Form.is_valid() 的设计思路——把校验逻辑和业务逻辑分离，视图函数只需关心"表单是否合法"，不需要关心"为什么不合法"。

核心源码（逐字来自文末完整源码）：

class LoginForm:
    def __init__(self, data: Dict):
        self.username = data.get("username", "").strip()
        self.password = data.get("password", "")
        self.errors: Dict[str, str] = {}

    def is_valid(self) -> bool:
        self.errors = {}
        if not self.username:
            self.errors["username"] = "用户名不能为空"
        elif len(self.username) < 2:
            self.errors["username"] = "用户名至少 2 个字符"
        if not self.password:
            self.errors["password"] = "密码不能为空"
        elif len(self.password) < 6:
            self.errors["password"] = "密码至少 6 位"
        return len(self.errors) == 0

可运行演示（补齐 Mock 数据与 print 反馈）：

from typing import Dict

class LoginForm:
    def __init__(self, data: Dict):
        self.username = data.get("username", "").strip()
        self.password = data.get("password", "")
        self.errors: Dict[str, str] = {}

    def is_valid(self) -> bool:
        self.errors = {}
        if not self.username:
            self.errors["username"] = "用户名不能为空"
        elif len(self.username) < 2:
            self.errors["username"] = "用户名至少 2 个字符"
        if not self.password:
            self.errors["password"] = "密码不能为空"
        elif len(self.password) < 6:
            self.errors["password"] = "密码至少 6 位"
        return len(self.errors) == 0

for data in [{"username": "alice", "password": "pass123"}, {"username": "a", "password": "123"}, {"username": "", "password": ""}]:
    form = LoginForm(data)
    print(f"输入={data} -> valid={form.is_valid()}, errors={form.errors}")

Step 3：用 MiddlewareContext 在中间件链中传递请求状态

痛点与机制：

MiddlewareContext 是请求在中间件链中传递的"行李箱"：每个中间件都能往里放东西（ctx.user = "alice"），后续的中间件和视图函数都能取到。timing_middleware 在调用 next_handler 前后各记一次时间，计算耗时——这个"前置操作 → 调用下一层 → 后置操作"的模式是所有中间件的标准结构。

核心源码（逐字来自文末完整源码）：

def timing_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    """计时中间件：记录请求耗时"""
    ctx.logs.append(f"[Timing] 请求开始 {ctx.method} {ctx.path}")
    resp = next_handler(ctx)
    elapsed = (time.time() - ctx.start_time) * 1000
    ctx.logs.append(f"[Timing] 耗时 {elapsed:.1f}ms → HTTP {resp.status}")
    return resp

可运行演示（补齐 Mock 数据与 print 反馈）：

import time
from typing import Callable, Dict, List, Optional

class MiddlewareContext:
    def __init__(self, method: str, path: str, headers: Dict, body: bytes, cookies: Dict):
        self.method = method; self.path = path; self.headers = headers
        self.body = body; self.cookies = cookies
        self.user: Optional[str] = None; self.session_id: Optional[str] = None
        self.start_time: float = time.time(); self.logs: List[str] = []

class MiddlewareResponse:
    def __init__(self, status: int, body: Dict):
        self.status = status; self.body = body

def timing_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    ctx.logs.append(f"[Timing] 请求开始 {ctx.method} {ctx.path}")
    resp = next_handler(ctx)
    elapsed = (time.time() - ctx.start_time) * 1000
    ctx.logs.append(f"[Timing] 耗时 {elapsed:.1f}ms → HTTP {resp.status}")
    return resp

def final_view(ctx: MiddlewareContext) -> MiddlewareResponse:
    time.sleep(0.01)
    return MiddlewareResponse(200, {"message": "ok"})

ctx = MiddlewareContext("GET", "/me", {}, b"", {})
resp = timing_middleware(ctx, final_view)
print("响应状态:", resp.status)
for log in ctx.logs:
    print("日志:", log)

Step 4：用 build_chain 把中间件列表组合成洋葱调用链

痛点与机制：

build_chain 用递归闭包实现"洋葱模型"：make_next(0) 返回一个函数，调用它会先执行 middlewares[0]，middlewares[0] 调用 next_handler 时会触发 make_next(1)，以此类推，直到所有中间件都执行完，最后调用 final（视图函数）。响应从内到外返回，每个中间件都能在 next_handler(ctx) 返回后做后置处理。这就是 Django MIDDLEWARE 列表的工作原理。

核心源码（逐字来自文末完整源码）：

def build_chain(middlewares: List[MiddlewareFunc], final: Callable) -> Callable:
    """将中间件列表组合成调用链"""
    def chain(ctx: MiddlewareContext) -> MiddlewareResponse:
        def make_next(idx: int) -> Callable:
            if idx >= len(middlewares):
                return lambda c: final(c)
            mw = middlewares[idx]
            return lambda c: mw(c, make_next(idx + 1))
        return make_next(0)(ctx)
    return chain

可运行演示（补齐 Mock 数据与 print 反馈）：

import time
from datetime import datetime
from typing import Callable, Dict, List, Optional

class MiddlewareContext:
    def __init__(self, method: str, path: str, headers: Dict, body: bytes, cookies: Dict):
        self.method = method; self.path = path; self.headers = headers
        self.body = body; self.cookies = cookies; self.user: Optional[str] = None
        self.session_id: Optional[str] = None; self.start_time = time.time(); self.logs: List[str] = []
class MiddlewareResponse:
    def __init__(self, status: int, body: Dict): self.status = status; self.body = body
MiddlewareFunc = Callable[[MiddlewareContext, Callable], MiddlewareResponse]

def timing_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    ctx.logs.append("进入 timing"); resp = next_handler(ctx); ctx.logs.append("离开 timing"); return resp
def logging_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    ctx.logs.append("进入 logging"); resp = next_handler(ctx)
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {ctx.method} {ctx.path} -> {resp.status}")
    ctx.logs.append("离开 logging"); return resp

def build_chain(middlewares: List[MiddlewareFunc], final: Callable) -> Callable:
    def chain(ctx: MiddlewareContext) -> MiddlewareResponse:
        def make_next(idx: int) -> Callable:
            if idx >= len(middlewares):
                return lambda c: final(c)
            mw = middlewares[idx]
            return lambda c: mw(c, make_next(idx + 1))
        return make_next(0)(ctx)
    return chain

def router(ctx: MiddlewareContext) -> MiddlewareResponse:
    ctx.logs.append("执行业务视图"); return MiddlewareResponse(200, {"ok": True})

ctx = MiddlewareContext("GET", "/demo", {}, b"", {})
resp = build_chain([logging_middleware, timing_middleware], router)(ctx)
print("最终状态:", resp.status)
print("执行顺序:", " -> ".join(ctx.logs))

Step 5：用 view_login 实现表单验证 + 密码哈希 + Session 创建

痛点与机制：

view_login 是三道关卡：① LoginForm.is_valid() 检查格式；② hashlib.sha256(password.encode()).hexdigest() 比对密码哈希——密码永远不明文存储，数据库泄露也不会暴露原始密码；③ create_session() 生成 Session ID，通过 Set-Cookie: sessionid=...; HttpOnly 响应头发给客户端。HttpOnly 标志让 JS 无法读取这个 Cookie，防止 XSS 攻击窃取 Session。

核心源码（逐字来自文末完整源码）：

def view_login(ctx: MiddlewareContext) -> MiddlewareResponse:
    if ctx.method != "POST":
        return MiddlewareResponse(405, {"error": "Method Not Allowed"})
    try:
        data = json.loads(ctx.body)
    except Exception:
        return MiddlewareResponse(400, {"error": "请求体必须是 JSON"})

    form = LoginForm(data)
    if not form.is_valid():
        return MiddlewareResponse(400, {"error": "表单验证失败", "fields": form.errors})

    pw_hash = hashlib.sha256(form.password.encode()).hexdigest()
    if _users.get(form.username) != pw_hash:
        return MiddlewareResponse(401, {"error": "用户名或密码错误"})

    sid = create_session(form.username)
    return MiddlewareResponse(
        200, {"message": f"欢迎，{form.username}！", "user": form.username},
        set_cookie=f"sessionid={sid}; HttpOnly; Path=/; Max-Age=3600"
    )

可运行演示（补齐 Mock 数据与 print 反馈）：

import hashlib, json, secrets, time
from typing import Dict, Optional

_sessions: Dict[str, Dict] = {}
_users: Dict[str, str] = {"alice": hashlib.sha256(b"pass123").hexdigest()}
def create_session(user: str) -> str:
    sid = secrets.token_hex(16); _sessions[sid] = {"user": user, "created_at": time.time()}; return sid
class LoginForm:
    def __init__(self, data: Dict):
        self.username = data.get("username", "").strip(); self.password = data.get("password", ""); self.errors: Dict[str, str] = {}
    def is_valid(self) -> bool:
        self.errors = {}
        if not self.username: self.errors["username"] = "用户名不能为空"
        elif len(self.username) < 2: self.errors["username"] = "用户名至少 2 个字符"
        if not self.password: self.errors["password"] = "密码不能为空"
        elif len(self.password) < 6: self.errors["password"] = "密码至少 6 位"
        return len(self.errors) == 0
class MiddlewareContext:
    def __init__(self, method: str, path: str, headers: Dict, body: bytes, cookies: Dict):
        self.method = method; self.path = path; self.headers = headers; self.body = body; self.cookies = cookies; self.user: Optional[str] = None; self.session_id: Optional[str] = None
class MiddlewareResponse:
    def __init__(self, status: int, body: Dict, set_cookie: Optional[str] = None, clear_cookie: bool = False):
        self.status = status; self.body = body; self.set_cookie = set_cookie; self.clear_cookie = clear_cookie

def view_login(ctx: MiddlewareContext) -> MiddlewareResponse:
    if ctx.method != "POST": return MiddlewareResponse(405, {"error": "Method Not Allowed"})
    try: data = json.loads(ctx.body)
    except Exception: return MiddlewareResponse(400, {"error": "请求体必须是 JSON"})
    form = LoginForm(data)
    if not form.is_valid(): return MiddlewareResponse(400, {"error": "表单验证失败", "fields": form.errors})
    pw_hash = hashlib.sha256(form.password.encode()).hexdigest()
    if _users.get(form.username) != pw_hash: return MiddlewareResponse(401, {"error": "用户名或密码错误"})
    sid = create_session(form.username)
    return MiddlewareResponse(200, {"message": f"欢迎，{form.username}！", "user": form.username}, set_cookie=f"sessionid={sid}; HttpOnly; Path=/; Max-Age=3600")

for payload in [{"username": "alice", "password": "pass123"}, {"username": "alice", "password": "wrong"}, {"username": "a", "password": "1"}]:
    resp = view_login(MiddlewareContext("POST", "/login", {}, json.dumps(payload).encode(), {}))
    print(f"登录 {payload['username']!r} -> {resp.status}, body={resp.body}")
    if resp.set_cookie: print("Set-Cookie 前 45 字符:", resp.set_cookie[:45])

痛点与机制：

HTTP 是无状态协议——每个请求都是独立的，服务端不记得上一个请求是谁发的。auth_middleware 解决了这个问题：从请求的 Cookie 头里取出 sessionid，查 _sessions 字典，把用户名写入 ctx.user。后续的视图函数只需检查 ctx.user is None 就能判断是否已登录，不需要每个视图都重复写 Cookie 解析逻辑——这正是中间件的价值所在。

核心源码（逐字来自文末完整源码）：

def auth_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    """认证中间件：从 Cookie 加载用户"""
    sid = ctx.cookies.get("sessionid")
    if sid:
        session = get_session(sid)
        if session:
            ctx.user = session["user"]
            ctx.session_id = sid
            ctx.logs.append(f"[Auth] 已认证用户: {ctx.user}")
        else:
            ctx.logs.append("[Auth] Session 已过期")
    else:
        ctx.logs.append("[Auth] 未登录")
    return next_handler(ctx)

可运行演示（补齐 Mock 数据与 print 反馈）：

import secrets, time
from typing import Callable, Dict, List, Optional
_sessions: Dict[str, Dict] = {}
def create_session(user: str) -> str:
    sid=secrets.token_hex(16); _sessions[sid]={"user":user,"created_at":time.time()}; return sid
def get_session(sid: str) -> Optional[Dict]:
    s=_sessions.get(sid)
    if s and time.time()-s["created_at"]<3600: return s
    if s: del _sessions[sid]
    return None
class MiddlewareContext:
    def __init__(self, method: str, path: str, headers: Dict, body: bytes, cookies: Dict):
        self.method=method; self.path=path; self.headers=headers; self.body=body; self.cookies=cookies; self.user: Optional[str]=None; self.session_id: Optional[str]=None; self.logs: List[str]=[]
class MiddlewareResponse:
    def __init__(self, status: int, body: Dict): self.status=status; self.body=body
def auth_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    sid = ctx.cookies.get("sessionid")
    if sid:
        session = get_session(sid)
        if session:
            ctx.user = session["user"]; ctx.session_id = sid; ctx.logs.append(f"[Auth] 已认证用户: {ctx.user}")
        else: ctx.logs.append("[Auth] Session 已过期")
    else: ctx.logs.append("[Auth] 未登录")
    return next_handler(ctx)
def view_me(ctx: MiddlewareContext) -> MiddlewareResponse:
    return MiddlewareResponse(200, {"user": ctx.user}) if ctx.user else MiddlewareResponse(401, {"error": "请先登录"})
sid=create_session("alice")
for cookies in [{"sessionid": sid}, {}, {"sessionid": "bad-id"}]:
    ctx=MiddlewareContext("GET","/me",{},b"",cookies); resp=auth_middleware(ctx, view_me)
    print(f"cookies={list(cookies.keys())} -> status={resp.status}, body={resp.body}, logs={ctx.logs}")

Step 7：用 show_chain 可视化中间件执行顺序

痛点与机制：

show_chain 打印两张图：请求方向（从外到内）和响应方向（从内到外）。这个"洋葱"结构是理解中间件的关键——logging_middleware 在最外层，所以它能看到最终的响应状态码；auth_middleware 在最内层（靠近视图），所以它能在视图执行前把用户信息写入 ctx。Django 的 MIDDLEWARE 列表顺序就是这个洋葱的层次顺序。

核心源码（逐字来自文末完整源码）：

def show_chain() -> None:
    names = ["logging_middleware", "timing_middleware", "auth_middleware", "router(view)"]
    print(f"\n{'─'*55}")
    print("  中间件执行顺序（请求方向 →，响应方向 ←）")
    print(f"{'─'*55}")
    print("  请求 →")
    for i, name in enumerate(names):
        indent = "  " + "  " * i
        print(f"{indent}└─▶ {name}")
    print()
    print("  响应 ←")
    for i, name in enumerate(reversed(names)):
        indent = "  " + "  " * i
        print(f"{indent}└─◀ {name}")
    print(f"{'─'*55}")

    print(f"\n{'─'*55}")
    print(f"  {'中间件':<25} {'职责'}")
    print(f"{'─'*55}")
    rows = [
        ("logging_middleware", "记录访问日志"),
        ("timing_middleware",  "统计请求耗时"),
        ("auth_middleware",    "解析 Cookie，加载用户"),
        ("router",             "路由分发到视图函数"),
    ]
    for name, role in rows:
        print(f"  {name:<25} {role}")
    print(f"{'─'*55}")

可运行演示（补齐 Mock 数据与 print 反馈）：

def show_chain() -> None:
    print(f"\n{'─'*54}")
    print("  中间件洋葱模型执行顺序")
    print(f"{'─'*54}")
    steps = [
        "1. 请求进入 logging_middleware",
        "2. logging 调用 next -> timing_middleware",
        "3. timing 调用 next -> auth_middleware",
        "4. auth 调用 next -> router/view",
        "5. view 返回响应",
        "6. auth 返回响应",
        "7. timing 记录耗时后返回响应",
        "8. logging 打印访问日志后返回响应",
    ]
    for step in steps: print(" ", step)
    print(f"{'─'*54}")
show_chain()

Step 8：用 main 做 server/demo/chain 三种模式的 CLI 总入口

痛点与机制：

main 用 argparse 做 CLI 入口：--mode server 启动真实 HTTP 服务器，--mode demo 用 urllib 发真实请求演示完整的登录→鉴权→退出流程，--mode chain 打印中间件链可视化。demo_client 里的 jar 字典模拟浏览器的 Cookie 存储，解析 Set-Cookie 响应头并在后续请求里带上 Cookie 请求头——这正是浏览器自动做的事情。

核心源码（逐字来自文末完整源码）：

def main() -> None:
    parser = argparse.ArgumentParser(description="Cookie/Session/中间件演示（零依赖）")
    parser.add_argument("--mode", choices=["server", "demo", "chain"], default="chain",
                        help="server=启动服务器, demo=演示客户端, chain=中间件链可视化")
    parser.add_argument("--port", type=int, default=8001)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "demo":
        demo_client(f"http://127.0.0.1:{args.port}")
    else:
        show_chain()

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse, sys

def run_server(host: str = "127.0.0.1", port: int = 8001) -> None:
    print(f"server 模式: 将监听 http://{host}:{port}")
def demo_client(base: str = "http://127.0.0.1:8001") -> None:
    print(f"demo 模式: 将请求 {base}/login、{base}/me、{base}/logout")
def show_chain() -> None:
    print("chain 模式: 展示 logging -> timing -> auth -> router 的洋葱顺序")
def main() -> None:
    parser = argparse.ArgumentParser(description="Django 表单/Session/中间件演示")
    parser.add_argument("--mode", choices=["server", "demo", "chain"], default="chain")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8001)
    args = parser.parse_args()
    if args.mode == "server": run_server(args.host, args.port)
    elif args.mode == "demo": demo_client(f"http://{args.host}:{args.port}")
    else: show_chain()
for mode in ["chain", "server", "demo"]:
    sys.argv = ["prog", "--mode", mode, "--port", "18081"]
    print(f">>> python3 21-django-advanced.py --mode {mode}")
    main()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

#!/usr/bin/env python3
"""
21-django-advanced.py
用 Python 内置 http.server 模拟：
  - Cookie / Session 机制
  - 中间件执行链（计时 / 认证 / 日志）
  - 简化版表单验证

用法：
  python3 21-django-advanced.py --mode server   # 启动服务器
  python3 21-django-advanced.py --mode demo     # 演示（需先启动服务器）
  python3 21-django-advanced.py --mode chain    # 打印中间件链执行顺序
"""

import argparse
import hashlib
import http.server
import json
import secrets
import time
import urllib.request
import urllib.error
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, Tuple



_sessions: Dict[str, Dict] = {}
_users: Dict[str, str] = {
    "alice": hashlib.sha256(b"pass123").hexdigest(),
    "bob":   hashlib.sha256(b"secret").hexdigest(),
}


def create_session(user: str) -> str:
    sid = secrets.token_hex(16)
    _sessions[sid] = {"user": user, "created_at": time.time()}
    return sid


def get_session(sid: str) -> Optional[Dict]:
    s = _sessions.get(sid)
    if s and time.time() - s["created_at"] < 3600:  # 1小时过期
        return s
    if s:
        del _sessions[sid]
    return None


def delete_session(sid: str) -> None:
    _sessions.pop(sid, None)


def parse_cookies(cookie_header: str) -> Dict[str, str]:
    cookies: Dict[str, str] = {}
    for part in cookie_header.split(";"):
        part = part.strip()
        if "=" in part:
            k, v = part.split("=", 1)
            cookies[k.strip()] = v.strip()
    return cookies


# ─── 简化版表单验证 ───────────────────────────────────────────────────────────

dataclass_like = None  # 用普通类代替

class LoginForm:
    def __init__(self, data: Dict):
        self.username = data.get("username", "").strip()
        self.password = data.get("password", "")
        self.errors: Dict[str, str] = {}

    def is_valid(self) -> bool:
        self.errors = {}
        if not self.username:
            self.errors["username"] = "用户名不能为空"
        elif len(self.username) < 2:
            self.errors["username"] = "用户名至少 2 个字符"
        if not self.password:
            self.errors["password"] = "密码不能为空"
        elif len(self.password) < 6:
            self.errors["password"] = "密码至少 6 位"
        return len(self.errors) == 0


# ─── 中间件框架 ───────────────────────────────────────────────────────────────

class MiddlewareContext:
    """在中间件链中传递的请求上下文"""
    def __init__(self, method: str, path: str, headers: Dict, body: bytes, cookies: Dict):
        self.method = method
        self.path = path
        self.headers = headers
        self.body = body
        self.cookies = cookies
        self.user: Optional[str] = None
        self.session_id: Optional[str] = None
        self.start_time: float = time.time()
        self.logs: List[str] = []


class MiddlewareResponse:
    def __init__(self, status: int, body: Dict,
                 set_cookie: Optional[str] = None,
                 clear_cookie: bool = False):
        self.status = status
        self.body = body
        self.set_cookie = set_cookie
        self.clear_cookie = clear_cookie


MiddlewareFunc = Callable[[MiddlewareContext, Callable], MiddlewareResponse]


def timing_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    """计时中间件：记录请求耗时"""
    ctx.logs.append(f"[Timing] 请求开始 {ctx.method} {ctx.path}")
    resp = next_handler(ctx)
    elapsed = (time.time() - ctx.start_time) * 1000
    ctx.logs.append(f"[Timing] 耗时 {elapsed:.1f}ms → HTTP {resp.status}")
    return resp


def auth_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    """认证中间件：从 Cookie 加载用户"""
    sid = ctx.cookies.get("sessionid")
    if sid:
        session = get_session(sid)
        if session:
            ctx.user = session["user"]
            ctx.session_id = sid
            ctx.logs.append(f"[Auth] 已认证用户: {ctx.user}")
        else:
            ctx.logs.append("[Auth] Session 已过期")
    else:
        ctx.logs.append("[Auth] 未登录")
    return next_handler(ctx)


def logging_middleware(ctx: MiddlewareContext, next_handler: Callable) -> MiddlewareResponse:
    """日志中间件：打印访问记录"""
    resp = next_handler(ctx)
    user_str = ctx.user or "anonymous"
    print(f"  [{datetime.now().strftime('%H:%M:%S')}] {ctx.method} {ctx.path}"
          f" → {resp.status} ({user_str})")
    return resp


def build_chain(middlewares: List[MiddlewareFunc], final: Callable) -> Callable:
    """将中间件列表组合成调用链"""
    def chain(ctx: MiddlewareContext) -> MiddlewareResponse:
        def make_next(idx: int) -> Callable:
            if idx >= len(middlewares):
                return lambda c: final(c)
            mw = middlewares[idx]
            return lambda c: mw(c, make_next(idx + 1))
        return make_next(0)(ctx)
    return chain


# ─── 视图函数 ─────────────────────────────────────────────────────────────────

def view_login(ctx: MiddlewareContext) -> MiddlewareResponse:
    if ctx.method != "POST":
        return MiddlewareResponse(405, {"error": "Method Not Allowed"})
    try:
        data = json.loads(ctx.body)
    except Exception:
        return MiddlewareResponse(400, {"error": "请求体必须是 JSON"})

    form = LoginForm(data)
    if not form.is_valid():
        return MiddlewareResponse(400, {"error": "表单验证失败", "fields": form.errors})

    pw_hash = hashlib.sha256(form.password.encode()).hexdigest()
    if _users.get(form.username) != pw_hash:
        return MiddlewareResponse(401, {"error": "用户名或密码错误"})

    sid = create_session(form.username)
    return MiddlewareResponse(
        200, {"message": f"欢迎，{form.username}！", "user": form.username},
        set_cookie=f"sessionid={sid}; HttpOnly; Path=/; Max-Age=3600"
    )


def view_me(ctx: MiddlewareContext) -> MiddlewareResponse:
    if not ctx.user:
        return MiddlewareResponse(401, {"error": "请先登录"})
    return MiddlewareResponse(200, {
        "user": ctx.user,
        "session_age": f"{time.time() - _sessions[ctx.session_id]['created_at']:.0f}s"
    })


def view_logout(ctx: MiddlewareContext) -> MiddlewareResponse:
    if ctx.session_id:
        delete_session(ctx.session_id)
    return MiddlewareResponse(
        200, {"message": "已退出登录"},
        clear_cookie=True
    )


ROUTES = {
    ("POST", "/login"):  view_login,
    ("GET",  "/me"):     view_me,
    ("POST", "/logout"): view_logout,
}


def router(ctx: MiddlewareContext) -> MiddlewareResponse:
    handler = ROUTES.get((ctx.method, ctx.path))
    if handler is None:
        return MiddlewareResponse(404, {"error": f"路径 {ctx.path} 不存在"})
    return handler(ctx)


# ─── HTTP 服务器 ──────────────────────────────────────────────────────────────

MIDDLEWARES: List[MiddlewareFunc] = [logging_middleware, timing_middleware, auth_middleware]
handle_request = build_chain(MIDDLEWARES, router)


class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *args): pass  # 禁用默认日志

    def _handle(self) -> None:
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length) if length else b""
        cookies = parse_cookies(self.headers.get("Cookie", ""))
        ctx = MiddlewareContext(
            method=self.command,
            path=self.path,
            headers=dict(self.headers),
            body=body,
            cookies=cookies,
        )
        resp = handle_request(ctx)
        payload = json.dumps(resp.body, ensure_ascii=False).encode()
        self.send_response(resp.status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        if resp.set_cookie:
            self.send_header("Set-Cookie", resp.set_cookie)
        if resp.clear_cookie:
            self.send_header("Set-Cookie", "sessionid=; Max-Age=0; Path=/")
        self.end_headers()
        self.wfile.write(payload)

    do_GET = do_POST = _handle


def run_server(port: int = 8001) -> None:
    print(f"🚀 Session 服务器启动：http://127.0.0.1:{port}")
    print("   路由：POST /login | GET /me | POST /logout")
    print("   测试账号：alice/pass123, bob/secret\n")
    try:
        http.server.HTTPServer(("127.0.0.1", port), Handler).serve_forever()
    except KeyboardInterrupt:
        print("\n服务器已停止。")


# ─── 演示客户端 ───────────────────────────────────────────────────────────────

def demo_client(base: str = "http://127.0.0.1:8001") -> None:
    jar: Dict[str, str] = {}  # 简单 Cookie 存储

    def request(method: str, path: str, data: Dict = None) -> Tuple[int, Dict]:
        payload = json.dumps(data).encode() if data else b""
        cookie_str = "; ".join(f"{k}={v}" for k, v in jar.items())
        headers = {"Content-Type": "application/json", "Content-Length": str(len(payload))}
        if cookie_str:
            headers["Cookie"] = cookie_str
        req = urllib.request.Request(f"{base}{path}", data=payload or None,
                                     headers=headers, method=method)
        try:
            with urllib.request.urlopen(req) as r:
                # 解析 Set-Cookie
                sc = r.headers.get("Set-Cookie", "")
                if sc:
                    for part in sc.split(";"):
                        part = part.strip()
                        if "=" in part and not any(
                            part.lower().startswith(x)
                            for x in ["httponly", "path=", "max-age=", "secure"]
                        ):
                            k, v = part.split("=", 1)
                            if v:
                                jar[k] = v
                            else:
                                jar.pop(k, None)
                return r.status, json.loads(r.read())
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read())

    print("\n=== Cookie/Session 演示 ===\n")

    steps = [
        ("未登录访问 /me",          "GET",  "/me",      None),
        ("错误密码登录",             "POST", "/login",   {"username": "alice", "password": "wrong"}),
        ("表单验证失败（密码太短）", "POST", "/login",   {"username": "alice", "password": "123"}),
        ("正确登录",                 "POST", "/login",   {"username": "alice", "password": "pass123"}),
        ("登录后访问 /me",           "GET",  "/me",      None),
        ("退出登录",                 "POST", "/logout",  None),
        ("退出后访问 /me",           "GET",  "/me",      None),
    ]

    for desc, method, path, data in steps:
        status, body = request(method, path, data)
        icon = "✅" if status < 400 else "❌"
        print(f"  {icon} {desc}")
        print(f"     HTTP {status} → {body}")
        if jar:
            print(f"     Cookie jar: {jar}")
        print()


# ─── 中间件链可视化 ───────────────────────────────────────────────────────────

def show_chain() -> None:
    names = ["logging_middleware", "timing_middleware", "auth_middleware", "router(view)"]
    print(f"\n{'─'*55}")
    print("  中间件执行顺序（请求方向 →，响应方向 ←）")
    print(f"{'─'*55}")
    print("  请求 →")
    for i, name in enumerate(names):
        indent = "  " + "  " * i
        print(f"{indent}└─▶ {name}")
    print()
    print("  响应 ←")
    for i, name in enumerate(reversed(names)):
        indent = "  " + "  " * i
        print(f"{indent}└─◀ {name}")
    print(f"{'─'*55}")

    print(f"\n{'─'*55}")
    print(f"  {'中间件':<25} {'职责'}")
    print(f"{'─'*55}")
    rows = [
        ("logging_middleware", "记录访问日志"),
        ("timing_middleware",  "统计请求耗时"),
        ("auth_middleware",    "解析 Cookie，加载用户"),
        ("router",             "路由分发到视图函数"),
    ]
    for name, role in rows:
        print(f"  {name:<25} {role}")
    print(f"{'─'*55}")


# ─── 入口 ─────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(description="Cookie/Session/中间件演示（零依赖）")
    parser.add_argument("--mode", choices=["server", "demo", "chain"], default="chain",
                        help="server=启动服务器, demo=演示客户端, chain=中间件链可视化")
    parser.add_argument("--port", type=int, default=8001)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "demo":
        demo_client(f"http://127.0.0.1:{args.port}")
    else:
        show_chain()


if __name__ == "__main__":
    main()

$ python3 21-python-django-advanced.py --mode chain

───────────────────────────────────────────────────────
  中间件执行顺序（请求方向 →，响应方向 ←）
───────────────────────────────────────────────────────
  请求 →
  └─▶ logging_middleware
    └─▶ timing_middleware
      └─▶ auth_middleware
        └─▶ router(view)

  响应 ←
  └─◀ router(view)
    └─◀ auth_middleware
      └─◀ timing_middleware
        └─◀ logging_middleware
───────────────────────────────────────────────────────

$ python3 21-python-django-advanced.py --mode server &
$ python3 21-python-django-advanced.py --mode demo

=== Cookie/Session 演示 ===

  ❌ 未登录访问 /me
     HTTP 401 → {'error': '请先登录'}

  ❌ 错误密码登录
     HTTP 401 → {'error': '用户名或密码错误'}

  ❌ 表单验证失败（密码太短）
     HTTP 400 → {'error': '表单验证失败', 'fields': {'password': '密码至少 6 位'}}

  ✅ 正确登录
     HTTP 200 → {'message': '欢迎，alice！', 'user': 'alice'}
     Cookie jar: {'sessionid': 'a3f2...'}

  ✅ 登录后访问 /me
     HTTP 200 → {'user': 'alice', 'session_age': '0s'}
     Cookie jar: {'sessionid': 'a3f2...'}

  ✅ 退出登录
     HTTP 200 → {'message': '已退出登录'}

  ❌ 退出后访问 /me
     HTTP 401 → {'error': '请先登录'}

小结

概念 一句话记忆
secrets.token_hex(16) 密码学安全的随机 Session ID，暴力枚举概率 1/2^128
HttpOnly Cookie JS 无法读取，防止 XSS 窃取 Session
LoginForm.is_valid() 集中校验所有字段，errors 字典收集错误，视图函数只看返回值
MiddlewareContext 请求在中间件链中传递的"行李箱"，每层都能读写
build_chain 递归闭包实现洋葱模型，请求从外到内，响应从内到外
auth_middleware 从 Cookie 加载用户，写入 ctx.user，视图函数只需检查 ctx.user
hashlib.sha256 密码哈希存储，数据库泄露不暴露原始密码
Set-Cookie: sessionid=; Max-Age=0 退出登录时让浏览器删除 Cookie

⏱ NexDo Time（5 分钟）

挑战：给中间件链加一个限流中间件 rate_limit_middleware，限制同一路径每秒最多 5 次请求，超过返回 429。

具体步骤：

  1. 用字典 _rate: Dict[str, List[float]] 记录每个路径的请求时间戳列表
  2. 在 rate_limit_middleware 里：清理 1 秒前的时间戳，如果剩余数量 >= 5 就返回 MiddlewareResponse(429, {"error": "Too Many Requests"})，否则记录当前时间戳并调用 next_handler
  3. 把它加入 MIDDLEWARES 列表（放在 logging_middleware 之后）
  4. 用循环发 6 次请求验证第 6 次返回 429

Don’t wait for next time, do it in the next moment.