File size: 4,895 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """
中间件模块
提供请求指标、安全头、权限策略、ServiceLocator 作用域等中间件。
"""
import logging
import os
import time
from collections.abc import Callable, Iterable, Mapping
from django.conf import settings
from django.http import HttpRequest, HttpResponse
class RequestMetricsMiddleware:
"""请求指标记录中间件（WSGI callable 风格）"""
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
self.get_response = get_response
def __call__(self, request: HttpRequest) -> HttpResponse:
from apps.core.telemetry import metrics as metrics_module
enabled = not settings.DEBUG or os.environ.get("DJANGO_REQUEST_METRICS") == "1"
start = time.monotonic()
status_code = 500
try:
response = self.get_response(request)
status_code = response.status_code
return response
finally:
if enabled:
duration_ms = int((time.monotonic() - start) * 1000)
metrics_module.record_request(
method=request.method or "GET",
path=request.path,
status_code=status_code,
duration_ms=duration_ms,
)
class SecurityHeadersMiddleware:
"""按路径设置 Content-Security-Policy 响应头的中间件"""
# /api/v1/docs 等文档路径不使用 API 策略
_DOCS_SUFFIXES = ("/docs", "/schema", "/redoc", "/swagger")
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
self.get_response = get_response
def __call__(self, request: HttpRequest) -> HttpResponse:
response = self.get_response(request)
self._apply_csp(request, response)
return response
def _apply_csp(self, request: HttpRequest, response: HttpResponse) -> None:
path = request.path
if path.startswith("/admin"):
csp = getattr(settings, "CONTENT_SECURITY_POLICY_ADMIN", "")
csp_ro = getattr(settings, "CONTENT_SECURITY_POLICY_ADMIN_REPORT_ONLY", "")
elif path.startswith("/api/") and not any(path.endswith(s) for s in self._DOCS_SUFFIXES):
csp = getattr(settings, "CONTENT_SECURITY_POLICY_API", "")
csp_ro = getattr(settings, "CONTENT_SECURITY_POLICY_API_REPORT_ONLY", "")
else:
csp = getattr(settings, "CONTENT_SECURITY_POLICY", "")
csp_ro = getattr(settings, "CONTENT_SECURITY_POLICY_REPORT_ONLY", "")
if csp:
response["Content-Security-Policy"] = csp
if csp_ro:
response["Content-Security-Policy-Report-Only"] = csp_ro
class PermissionsPolicyMiddleware:
"""设置 Permissions-Policy 响应头的中间件"""
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
self.get_response = get_response
def __call__(self, request: HttpRequest) -> HttpResponse:
response = self.get_response(request)
policy = getattr(settings, "PERMISSIONS_POLICY", "")
if policy:
response["Permissions-Policy"] = self._serialize_policy(policy)
return response
def _serialize_policy(self, policy: str | Mapping[str, object]) -> str:
if isinstance(policy, str):
return policy
if not isinstance(policy, Mapping):
return str(policy)
directives: list[str] = []
for feature, allowlist in policy.items():
directives.append(f"{feature}={self._serialize_allowlist(allowlist)}")
return ", ".join(directives)
def _serialize_allowlist(self, allowlist: object) -> str:
if allowlist in (None, [], (), set()):
return "()"
if allowlist == "*":
return "*"
if isinstance(allowlist, str):
return f"({self._serialize_source(allowlist)})"
if isinstance(allowlist, Iterable):
values = " ".join(self._serialize_source(value) for value in allowlist)
return f"({values})" if values else "()"
return f"({self._serialize_source(allowlist)})"
def _serialize_source(self, value: object) -> str:
if value in {"self", "src", "*"}:
return str(value)
return f'"{value}"'
class ServiceLocatorScopeMiddleware:
"""
ServiceLocator 请求级作用域中间件
每个 HTTP 请求在独立的 ServiceLocator scope 中执行，
确保请求间服务实例不互相污染（基于 ContextVar 实现）。
"""
def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
self.get_response = get_response
def __call__(self, request: HttpRequest) -> HttpResponse:
from apps.core.interfaces import ServiceLocator
with ServiceLocator.scope():
return self.get_response(request)
|