147 lines
5.7 KiB
Python
147 lines
5.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import mimetypes
|
|
import secrets
|
|
from http import HTTPStatus
|
|
from http.cookies import SimpleCookie
|
|
from typing import Any
|
|
from urllib.parse import unquote
|
|
|
|
from backend.bootstrap.config import SESSION_COOKIE, SESSION_MAX_AGE, STATIC_DIR
|
|
from backend.features.accounts.security import token_hash
|
|
from backend.http.context import correlation_id
|
|
from backend.http.errors import normalize_error_payload
|
|
|
|
|
|
class HttpTransportMixin:
|
|
"""Original HTTP transport, static-file, session and access behavior."""
|
|
|
|
application_service: Any
|
|
route_registry: Any
|
|
|
|
def session_token(self) -> str:
|
|
cookie = SimpleCookie()
|
|
try:
|
|
cookie.load(self.headers.get("Cookie", ""))
|
|
except Exception:
|
|
return ""
|
|
morsel = cookie.get(SESSION_COOKIE)
|
|
return morsel.value if morsel else ""
|
|
|
|
def require_auth(self, send_error: bool = True) -> bool:
|
|
raw_token = self.session_token()
|
|
service = self.application_service
|
|
user = service.database.session_user(token_hash(raw_token)) if raw_token else None
|
|
if not user:
|
|
if send_error:
|
|
self.send_json({"error": "请先登录。"}, HTTPStatus.UNAUTHORIZED)
|
|
return False
|
|
self.auth_user = user
|
|
service.bind_user(int(user["id"]))
|
|
return True
|
|
|
|
def require_csrf(self) -> bool:
|
|
supplied = self.headers.get("X-CSRF-Token", "")
|
|
expected = str(getattr(self, "auth_user", {}).get("csrf_token") or "")
|
|
if not supplied or not secrets.compare_digest(supplied, expected):
|
|
self.send_json({"error": "请求校验失败，请刷新页面后重试。"}, HTTPStatus.FORBIDDEN)
|
|
return False
|
|
return True
|
|
|
|
def require_admin(self) -> bool:
|
|
if str(getattr(self, "auth_user", {}).get("role") or "user") != "admin":
|
|
self.send_json({"error": "需要管理员权限。"}, HTTPStatus.FORBIDDEN)
|
|
return False
|
|
return True
|
|
|
|
def require_member(self) -> bool:
|
|
if self.application_service.membership()["active"]:
|
|
return True
|
|
self.send_json(
|
|
{"error": "该功能仅对有效会员开放，请联系管理员开通会员。", "code": "membership_required"},
|
|
HTTPStatus.FORBIDDEN,
|
|
)
|
|
return False
|
|
|
|
def require_access(self, method: str, path: str) -> bool:
|
|
route = self.route_registry.resolve(method, path)
|
|
if route is None:
|
|
self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)
|
|
return False
|
|
role = route.access
|
|
if role == "public":
|
|
return True
|
|
if role == "admin":
|
|
return self.require_admin()
|
|
if role == "member":
|
|
return self.require_member()
|
|
return True
|
|
|
|
def session_cookie(self, value: str, clear: bool = False) -> str:
|
|
max_age = 0 if clear else SESSION_MAX_AGE
|
|
cookie = (
|
|
f"{SESSION_COOKIE}={value}; Path=/; HttpOnly; SameSite=Lax; Max-Age={max_age}"
|
|
)
|
|
if self.headers.get("X-Forwarded-Proto", "").lower() == "https":
|
|
cookie += "; Secure"
|
|
return cookie
|
|
|
|
def read_json_body(self, allow_empty: bool = False) -> dict[str, Any]:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
if length == 0 and allow_empty:
|
|
return {}
|
|
if length <= 0 or length > 65536:
|
|
raise ValueError("请求内容为空或过大。")
|
|
return json.loads(self.rfile.read(length).decode("utf-8"))
|
|
|
|
def serve_static(self, request_path: str) -> None:
|
|
relative = unquote(request_path).lstrip("/") or "index.html"
|
|
candidate = (STATIC_DIR / relative).resolve()
|
|
try:
|
|
candidate.relative_to(STATIC_DIR.resolve())
|
|
except ValueError:
|
|
self.send_error(HTTPStatus.FORBIDDEN)
|
|
return
|
|
if not candidate.is_file():
|
|
candidate = STATIC_DIR / "index.html"
|
|
try:
|
|
content = candidate.read_bytes()
|
|
except OSError:
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
|
|
if content_type.startswith("text/") or content_type in {"application/javascript", "application/json"}:
|
|
content_type += "; charset=utf-8"
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
def send_json(
|
|
self,
|
|
payload: dict[str, Any],
|
|
status: HTTPStatus = HTTPStatus.OK,
|
|
headers: dict[str, str] | None = None,
|
|
) -> None:
|
|
request_id = getattr(self, "_correlation_id", "")
|
|
if not request_id:
|
|
request_id = correlation_id(self.headers.get("X-Request-ID", ""))
|
|
self._correlation_id = request_id
|
|
payload = normalize_error_payload(payload, status, request_id)
|
|
content = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("X-Request-ID", request_id)
|
|
for name, value in (headers or {}).items():
|
|
self.send_header(name, value)
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
def log_message(self, format_string: str, *args: Any) -> None:
|
|
print(f"[{self.log_date_time_string()}] {format_string % args}")
|