116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""多租户中间件：从请求头 X-Tenant-Id 识别租户，注入 request.tenant。
|
||
|
||
策略：
|
||
1. 读取 Header `X-Tenant-Id`，默认为 settings.TENANT_DEFAULT
|
||
2. 查询 Tenant 对象；找不到则 400
|
||
3. 把 request.tenant 注入；让 ViewSet 通过 TenantScopedModel 自动加过滤
|
||
|
||
未识别租户的请求 = 400 Bad Request。
|
||
|
||
注意：Django middleware 在 ASGI 下也是同步调用的（除非是纯 async middleware），
|
||
但 ORM 查询必须走 sync_to_async，因为下游 view 是 async。
|
||
"""
|
||
|
||
from django.conf import settings
|
||
from django.http import JsonResponse
|
||
from asgiref.sync import sync_to_async
|
||
|
||
from .models import Tenant
|
||
|
||
|
||
class TenantMiddleware:
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
|
||
def __call__(self, request):
|
||
# 公开路径跳过租户识别（ping、admin、auth/token 等）
|
||
path = request.path
|
||
if path.startswith("/admin/") or path in (
|
||
"/api/v1/ping/",
|
||
"/api/v1/auth/token/",
|
||
"/api/v1/auth/token/refresh/",
|
||
"/api/v1/auth/register/",
|
||
):
|
||
return self.get_response(request)
|
||
|
||
tenant_code = request.headers.get(
|
||
settings.TENANT_HEADER, settings.TENANT_DEFAULT
|
||
)
|
||
|
||
# 在 sync_to_async 里跑 ORM 查询（即便下游 view 是 async，中间件本身在 ASGI 下也是同步调用）
|
||
# 但保险起见同步包一层
|
||
request.tenant = tenant_code # 先存 code，view 里再解析
|
||
# 真正解析 Tenant 由 async 视图完成（request.tenant_or_error）
|
||
return self.get_response(request)
|
||
|
||
|
||
def get_tenant_sync(tenant_code):
|
||
"""同步 ORM 查询 Tenant（供 async view 调用）。"""
|
||
return Tenant.objects.get(code=tenant_code, is_active=True)
|
||
|
||
|
||
async def resolve_tenant(tenant_code):
|
||
"""async 解析 Tenant；找不到抛 Tenant.DoesNotExist。"""
|
||
return await sync_to_async(get_tenant_sync)(tenant_code)
|
||
|
||
|
||
# ============================================================
|
||
# 演示租户只读保护（批次 C2）
|
||
# ============================================================
|
||
|
||
# 演示租户内允许的写操作白名单（登录/刷新/进入演示本身）
|
||
DEMO_WRITE_ALLOWLIST = {
|
||
"/api/v1/auth/token/",
|
||
"/api/v1/auth/token/refresh/",
|
||
"/api/v1/auth/register/",
|
||
"/api/v1/demo/enter/",
|
||
"/api/v1/ping/",
|
||
}
|
||
|
||
# 商城客户端允许写：客户在演示账套里要能体验"下单 → 业务员确认"闭环。
|
||
# 仅放行客户端端点；后台端的确认/驳回/授权/账号管理仍被只读拦截，
|
||
# 且订单只落在演示租户内（不影响其他租户数据）。
|
||
DEMO_WRITE_PREFIX_ALLOWLIST = (
|
||
"/api/v1/storefront/login/",
|
||
"/api/v1/storefront/orders/",
|
||
)
|
||
|
||
SAFE_METHODS = {"GET", "HEAD", "OPTIONS", "TRACE"}
|
||
|
||
|
||
class DemoReadOnlyMiddleware:
|
||
"""演示租户（code=demo）下拦截写入。
|
||
|
||
判定依据是请求头租户 + HTTP 方法（不依赖前端自觉、也不依赖 token 里的 claim），
|
||
这样即便有人拿演示 token 手工 curl 也无法改数据。
|
||
"""
|
||
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
|
||
def __call__(self, request):
|
||
if request.method in SAFE_METHODS:
|
||
return self.get_response(request)
|
||
|
||
path = request.path
|
||
if path.startswith("/admin/") or path in DEMO_WRITE_ALLOWLIST:
|
||
return self.get_response(request)
|
||
if path.startswith(DEMO_WRITE_PREFIX_ALLOWLIST):
|
||
return self.get_response(request)
|
||
|
||
from django.conf import settings as dj_settings
|
||
|
||
tenant_code = request.headers.get(
|
||
dj_settings.TENANT_HEADER, dj_settings.TENANT_DEFAULT
|
||
)
|
||
if tenant_code == "demo":
|
||
return JsonResponse(
|
||
{
|
||
"code": "demo_read_only",
|
||
"detail": "演示账套为只读模式，写入操作已被拒绝。"
|
||
"注册免费账号即可创建你自己的账套。",
|
||
},
|
||
status=403,
|
||
)
|
||
return self.get_response(request)
|