95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""站点预览签名 URL：HMAC 短时效令牌，密钥首次使用时生成并持久化。
|
|
|
|
令牌只授予「该站点 dist/ 目录静态文件读」这一项能力，不携带任何其他 API 权限。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from deerflow.config.runtime_paths import runtime_home
|
|
|
|
SECRET_FILE_NAME = ".web-app-secret"
|
|
_DEFAULT_TTL_SECONDS = 7200
|
|
|
|
_secret_cache: bytes | None = None
|
|
|
|
|
|
def _secret_path() -> Path:
|
|
return runtime_home() / SECRET_FILE_NAME
|
|
|
|
|
|
def get_preview_secret() -> bytes:
|
|
"""加载或生成实例级 HMAC 密钥（0600 权限持久化在 .deer-flow 下）。"""
|
|
global _secret_cache
|
|
if _secret_cache is not None:
|
|
return _secret_cache
|
|
path = _secret_path()
|
|
try:
|
|
if path.is_file():
|
|
data = path.read_bytes().strip()
|
|
if len(data) >= 32:
|
|
_secret_cache = data
|
|
return _secret_cache
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = secrets.token_hex(32).encode("ascii")
|
|
path.write_bytes(data)
|
|
try:
|
|
path.chmod(0o600)
|
|
except OSError: # Windows 无 POSIX 权限，忽略
|
|
pass
|
|
_secret_cache = data
|
|
return _secret_cache
|
|
except OSError:
|
|
# 持久化失败时退化为进程内随机密钥（重启后旧令牌失效，可接受）
|
|
_secret_cache = secrets.token_hex(32).encode("ascii")
|
|
return _secret_cache
|
|
|
|
|
|
def mint_preview_token(thread_id: str, site_name: str, ttl_seconds: int = _DEFAULT_TTL_SECONDS) -> tuple[str, int]:
|
|
"""签发预览令牌，返回 (token, expires_at_epoch)。"""
|
|
expires_at = int(time.time()) + max(60, ttl_seconds)
|
|
payload = f"{thread_id}|{site_name}|{expires_at}"
|
|
encoded = base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=")
|
|
signature = hmac.new(get_preview_secret(), payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
return f"{encoded}.{signature}", expires_at
|
|
|
|
|
|
def verify_preview_token(token: str) -> tuple[str, str]:
|
|
"""校验令牌，返回 (thread_id, site_name)。
|
|
|
|
Raises:
|
|
ValueError: 令牌格式非法、签名不符或已过期。
|
|
"""
|
|
if not token or token.count(".") != 1:
|
|
raise ValueError("malformed token")
|
|
encoded, signature = token.split(".", 1)
|
|
if not encoded or not signature:
|
|
raise ValueError("malformed token")
|
|
padding = "=" * (-len(encoded) % 4)
|
|
try:
|
|
payload = base64.urlsafe_b64decode(encoded + padding).decode("utf-8")
|
|
except (ValueError, UnicodeDecodeError) as exc:
|
|
raise ValueError("malformed token") from exc
|
|
parts = payload.split("|")
|
|
if len(parts) != 3:
|
|
raise ValueError("malformed token")
|
|
thread_id, site_name, exp_raw = parts
|
|
expected = hmac.new(get_preview_secret(), payload.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(expected, signature):
|
|
raise ValueError("bad signature")
|
|
try:
|
|
expires_at = int(exp_raw)
|
|
except ValueError as exc:
|
|
raise ValueError("malformed token") from exc
|
|
if time.time() > expires_at:
|
|
raise ValueError("token expired")
|
|
if not thread_id or not site_name:
|
|
raise ValueError("malformed token")
|
|
return thread_id, site_name
|