Files
re-xxm/reaction_board/web/server.py
T

749 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""只读看板 HTTP 服务。
只注册 GET(以及登录所需的 POST /login)。数据管理仍由 Telegram、CLI 或
数据库完成,Web 不提供任何业务写接口。
"""
from __future__ import annotations
import hashlib
import hmac
import json
import mimetypes
import time
from dataclasses import dataclass
from http import HTTPStatus
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Callable, Mapping
from urllib.parse import parse_qs, quote, unquote, urlsplit
from ..analytics import is_loopback
from ..client_ip import extract_client_ip, is_bot, normalize_route, parse_networks
from ..config import ConfigStore
from ..database import Database
from .queries import DashboardQueries
from .visits import VisitCollector
STATIC_ROOT = Path(__file__).resolve().parent / "static"
COOKIE_NAME = "rb_session"
SESSION_TTL = 7 * 86400
PROTECTED_API = {"/api/candidates", "/api/system", "/api/stats"}
PROTECTED_PAGES = {"/candidates", "/system"}
# 显式放行的 POST 端点:访问统计埋点(业务写操作仍统一 405)。
POST_API = {"/api/visit"}
VISIT_BODY_LIMIT = 2048
class ApiError(Exception):
def __init__(self, status: int, code: str, message: str) -> None:
super().__init__(message)
self.status = status
self.code = code
self.message = message
@dataclass
class WebContext:
queries: DashboardQueries
config_store: ConfigStore
preview: bool = False
logger: Any = None
clock: Callable[[], float] = time.time
db_path: str | None = None
schema_version: int | None = None
visits: VisitCollector | None = None
@property
def auth_token(self) -> str:
return self.config_store.get().web.auth_token
class DashboardHTTPServer(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = True
def __init__(self, address, context: WebContext) -> None:
self.context = context
super().__init__(address, DashboardHandler)
def server_close(self) -> None:
# 关闭前停止访问统计刷新线程,保证最后一次缓冲落库。
collector = getattr(self.context, "visits", None)
if collector is not None:
collector.stop()
super().server_close()
class DashboardHandler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "ReactionBoard/0.1"
# ------------------------------------------------------------ logging
def log_message(self, fmt: str, *args: Any) -> None:
# 请求级日志容易刷屏,只保留错误在 _send_error 里记录。
return
@property
def context(self) -> WebContext:
return self.server.context # type: ignore[attr-defined]
# ------------------------------------------------------------ HTTP
def do_GET(self) -> None:
self._dispatch("GET")
def do_HEAD(self) -> None:
self._dispatch("HEAD")
def do_POST(self) -> None:
self._dispatch("POST")
def do_PUT(self) -> None:
self._method_not_allowed()
def do_PATCH(self) -> None:
self._method_not_allowed()
def do_DELETE(self) -> None:
self._method_not_allowed()
def do_OPTIONS(self) -> None:
self._method_not_allowed()
def _method_not_allowed(self) -> None:
self._send_error(HTTPStatus.METHOD_NOT_ALLOWED, "method_not_allowed", "不支持该操作")
# ------------------------------------------------------------ dispatch
def _dispatch(self, method: str) -> None:
parsed = urlsplit(self.path)
path = unquote(parsed.path)
query = parse_qs(parsed.query, keep_blank_values=True)
try:
if path == "/api/health":
self._send_json(HTTPStatus.OK, {
"ok": True,
"preview": self.context.preview,
"time": int(self.context.clock()),
})
return
if path == "/api/session":
self._send_json(HTTPStatus.OK, {
"auth_enabled": bool(self.context.auth_token),
"authenticated": self._authorized(),
"preview": self.context.preview,
})
return
if path == "/login":
if method == "POST":
self._login_post(query)
else:
self._login_page(query)
return
if path == "/logout":
self._logout()
return
if path.startswith("/api/"):
if path in PROTECTED_API and not self._authorized():
self._send_error(HTTPStatus.UNAUTHORIZED, "unauthorized", "需要登录")
return
if method == "POST":
if path in POST_API:
self._api_post(path, query)
return
self._send_error(
HTTPStatus.METHOD_NOT_ALLOWED,
"method_not_allowed",
"不支持该操作",
)
return
if path in POST_API:
self._send_error(
HTTPStatus.METHOD_NOT_ALLOWED,
"method_not_allowed",
"不支持该操作",
)
return
self._api(path, query, method)
return
if path in PROTECTED_PAGES and not self._authorized():
self._redirect(f"/login?next={quote(path)}")
return
self._static(path, method)
except ApiError as exc:
self._send_error(exc.status, exc.code, exc.message)
except ValueError as exc:
self._send_error(HTTPStatus.BAD_REQUEST, "bad_request", str(exc))
except BrokenPipeError:
return
except Exception as exc: # noqa: BLE001
if self.context.logger is not None:
self.context.logger.error("web.error", path=path, error=str(exc))
self._send_error(
HTTPStatus.INTERNAL_SERVER_ERROR, "internal_error", "服务端错误"
)
# ------------------------------------------------------------ auth
def _session_value(self, expires_at: int) -> str:
token = self.context.auth_token
payload = f"rb:{int(expires_at)}"
signature = hmac.new(
token.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256
).hexdigest()
return f"{int(expires_at)}.{signature}"
def _valid_session(self, value: str) -> bool:
token = self.context.auth_token
if not token or "." not in value:
return False
raw_exp, signature = value.split(".", 1)
try:
expires_at = int(raw_exp)
except ValueError:
return False
if expires_at < int(self.context.clock()):
return False
payload = f"rb:{expires_at}"
expected = hmac.new(
token.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected)
def _authorized(self) -> bool:
token = self.context.auth_token
if not token:
return True
header = self.headers.get("Authorization", "")
if header.startswith("Bearer "):
return hmac.compare_digest(header[7:].strip(), token)
cookie_header = self.headers.get("Cookie")
if not cookie_header:
return False
cookie = SimpleCookie()
try:
cookie.load(cookie_header)
except Exception: # noqa: BLE001
return False
morsel = cookie.get(COOKIE_NAME)
return morsel is not None and self._valid_session(morsel.value)
def _login_page(self, query: Mapping[str, list[str]]) -> None:
if not self.context.auth_token:
self._redirect("/")
return
next_path = _safe_next(_param(query, "next", ""))
html = """<!doctype html>
<html lang="zh-CN"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<link rel="icon" href="/static/favicon.svg" type="image/svg+xml">
<title>登录 · Reaction 看板</title>
<style>
:root{color-scheme:light dark}body{margin:0;min-height:100vh;display:grid;
place-items:center;font-family:system-ui,-apple-system,"Segoe UI",sans-serif;
background:#f7f2ea;color:#2a201a}main{width:min(92vw,360px);padding:28px;
border-radius:16px;background:#fffdf9;box-shadow:0 12px 40px rgba(42,32,26,.12)}
h1{font-size:20px;margin:0 0 8px}p{color:#7b6a58;font-size:14px;margin:0 0 18px}
input{width:100%;box-sizing:border-box;padding:12px 14px;border:1px solid #e6d8c4;
border-radius:10px;font-size:16px}button{width:100%;margin-top:12px;padding:12px;
border:0;border-radius:10px;background:#a9702f;color:#fff;font-size:16px;
font-weight:600;cursor:pointer}@media(prefers-color-scheme:dark){body{background:#1b1411;
color:#f3e9dc}main{background:#221a15}p{color:#c9b6a0}input{background:#1b1411;
border-color:#3a2c22;color:#f3e9dc}button{background:#e9b97c;color:#2a1a10}}
</style></head><body><main>
<h1>Reaction 看板</h1><p>输入配置中的 web.auth_token 以继续。</p>
<form method="post" action="/login">
<input type="password" name="token" autocomplete="current-password" placeholder="访问令牌" required autofocus>
<button type="submit">进入看板</button>
</form></main></body></html>"""
html = html.replace(
'action="/login"', f'action="/login?next={quote(next_path)}"'
)
self._send_html(HTTPStatus.OK, html)
def _login_post(self, query: Mapping[str, list[str]]) -> None:
if not self.context.auth_token:
self._redirect("/")
return
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
length = 0
if length <= 0 or length > 4096:
self._send_error(HTTPStatus.BAD_REQUEST, "bad_request", "请求体过大")
return
body = self.rfile.read(length).decode("utf-8", errors="replace")
token = (parse_qs(body).get("token") or [""])[0]
if not hmac.compare_digest(token, self.context.auth_token):
self._send_error(HTTPStatus.UNAUTHORIZED, "unauthorized", "令牌不正确")
return
expires_at = int(self.context.clock()) + SESSION_TTL
value = self._session_value(expires_at)
target = _safe_next(_param(query, "next", ""))
self.send_response(HTTPStatus.SEE_OTHER)
self.send_header("Location", target)
self.send_header("Set-Cookie", (
f"{COOKIE_NAME}={value}; Path=/; HttpOnly; SameSite=Lax; "
f"Max-Age={SESSION_TTL}"
))
self.send_header("Content-Length", "0")
self._security_headers()
self.end_headers()
def _logout(self) -> None:
self.send_response(HTTPStatus.SEE_OTHER)
self.send_header("Location", "/login" if self.context.auth_token else "/")
self.send_header("Set-Cookie", f"{COOKIE_NAME}=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax")
self.send_header("Content-Length", "0")
self._security_headers()
self.end_headers()
# ------------------------------------------------------------ API
def _api_post(self, path: str, query: Mapping[str, list[str]]) -> None:
if path == "/api/visit":
self._visit_post(query)
return
raise ApiError(HTTPStatus.METHOD_NOT_ALLOWED, "method_not_allowed", "不支持该操作")
def _visit_post(self, query: Mapping[str, list[str]]) -> None:
"""记录一次页面浏览;前端 sendBeacon 上报,写内存后定期批量落库。"""
track = self.context.config_store.get().web.track
collector = self.context.visits
if not track.enabled or collector is None:
self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "disabled"})
return
path_value = _param(query, "path", "")
body = self._read_small_body()
if not path_value and body:
path_value = _json_path(body)
user_agent = self.headers.get("User-Agent")
if not track.count_bots and is_bot(user_agent):
self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "bot"})
return
peer = self.client_address[0] if self.client_address else None
ip = extract_client_ip(
self.headers,
peer,
trusted_proxies=parse_networks(track.trusted_proxies),
header_order=track.header_order,
)
if not ip:
self._send_json(HTTPStatus.OK, {"recorded": False, "reason": "no_ip"})
return
route = normalize_route(path_value)
recorded = collector.record(ip, route)
self._send_json(HTTPStatus.OK, {
"recorded": bool(recorded),
"route": route,
"reason": None if recorded else "capped",
})
def _read_small_body(self) -> bytes:
try:
length = int(self.headers.get("Content-Length", "0"))
except (TypeError, ValueError):
length = 0
if length <= 0:
return b""
if length > VISIT_BODY_LIMIT:
# 不读取超长请求体,直接关闭连接,避免 keep-alive 上残留字节错位。
self.close_connection = True
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "请求体过大")
return self.rfile.read(length)
def _api(self, path: str, query: Mapping[str, list[str]], method: str) -> None:
queries = self.context.queries
if path == "/api/overview":
self._send_json(HTTPStatus.OK, queries.overview())
return
if path == "/api/videos":
self._send_json(HTTPStatus.OK, queries.list_videos(
sort=_param(query, "sort", "heat"),
within_days=_optional_int(query, "within_days"),
year=_year_param(query),
growth_days=_int_param(query, "growth_days", 7, 1, 365),
folder_id=_optional_int(query, "folder"),
q=_optional_str(query, "q"),
page=_int_param(query, "page", 1, 1, 1_000_000),
page_size=_int_param(
query, "page_size",
_optional_int(query, "limit") or 20,
1, 200,
),
))
return
if path.startswith("/api/videos/"):
parts = [part for part in path.split("/") if part]
if len(parts) == 3:
aid = _path_int(parts[2], "aid")
data = queries.video_detail(aid)
if data is None:
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "视频不存在")
self._send_json(HTTPStatus.OK, data)
return
if len(parts) == 4 and parts[3] == "trend":
aid = _path_int(parts[2], "aid")
days = _range_days(query)
self._send_json(HTTPStatus.OK, queries.video_trend(aid, days=days))
return
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
if path == "/api/rankings":
scope = _param(query, "scope", "video")
if scope == "up":
self._send_json(HTTPStatus.OK, queries.list_ups(
sort=_param(query, "sort", "efficiency"),
within_days=_optional_int(query, "within_days"),
year=_year_param(query),
q=_optional_str(query, "q"),
min_videos=_int_param(query, "min_videos", 1, 1, 10000),
page=_int_param(query, "page", 1, 1, 1_000_000),
page_size=_int_param(
query, "page_size",
_optional_int(query, "limit") or 20,
1, 200,
),
))
return
if scope != "video":
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "scope 必须是 video 或 up")
self._send_json(HTTPStatus.OK, queries.list_videos(
sort=_param(query, "sort", "heat"),
within_days=_optional_int(query, "within_days"),
year=_year_param(query),
growth_days=_int_param(query, "growth_days", 7, 1, 365),
folder_id=_optional_int(query, "folder"),
q=_optional_str(query, "q"),
page=_int_param(query, "page", 1, 1, 1_000_000),
page_size=_int_param(
query, "page_size",
_optional_int(query, "limit") or 20,
1, 200,
),
))
return
if path == "/api/ups":
self._send_json(HTTPStatus.OK, queries.list_ups(
sort=_param(query, "sort", "efficiency"),
within_days=_optional_int(query, "within_days"),
year=_year_param(query),
q=_optional_str(query, "q"),
min_videos=_int_param(query, "min_videos", 1, 1, 10000),
page=_int_param(query, "page", 1, 1, 1_000_000),
page_size=_int_param(
query, "page_size",
_optional_int(query, "limit") or 20,
1, 200,
),
))
return
if path.startswith("/api/ups/"):
parts = [part for part in path.split("/") if part]
if len(parts) != 3:
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
mid = _path_int(parts[2], "mid")
data = queries.up_detail(mid)
if data is None:
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "UP 不存在")
self._send_json(HTTPStatus.OK, data)
return
if path == "/api/fans":
self._send_json(HTTPStatus.OK, queries.fans(
_optional_int(query, "mid"),
days=_int_param(query, "days", 90, 1, 3650),
start=_optional_str(query, "start"),
end=_optional_str(query, "end"),
recent_days=_int_param(query, "recent_days", 14, 1, 3650),
))
return
if path == "/api/about":
self._send_json(HTTPStatus.OK, queries.about())
return
if path == "/api/stats":
pending = (
self.context.visits.pending_rows()
if self.context.visits is not None else []
)
self._send_json(HTTPStatus.OK, queries.stats(
days=_int_param(query, "days", 30, 1, 3650),
pending=pending,
))
return
if path == "/api/candidates":
self._send_json(HTTPStatus.OK, queries.candidates())
return
if path == "/api/system":
self._send_json(HTTPStatus.OK, queries.system())
return
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
# ------------------------------------------------------------ static
def _static(self, path: str, method: str) -> None:
if path in {"/", "/index.html"}:
target = STATIC_ROOT / "index.html"
elif path.startswith("/static/"):
relative = path[len("/static/"):]
target = (STATIC_ROOT / relative).resolve()
try:
target.relative_to(STATIC_ROOT.resolve())
except ValueError:
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "文件不存在")
else:
# 伪静态前端路由:/rank、/video/123、/up/123 等直接返回 index.html,
# 由前端根据 location.pathname 渲染。带扩展名的路径仍按文件处理并返回 404。
last_segment = path.rstrip("/").rsplit("/", 1)[-1]
if "." in last_segment:
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "文件不存在")
target = STATIC_ROOT / "index.html"
if not target.is_file():
raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "页面不存在")
content_type = mimetypes.guess_type(str(target))[0] or "application/octet-stream"
if content_type.startswith("text/") or content_type in {
"application/javascript", "application/json", "image/svg+xml",
}:
content_type += "; charset=utf-8"
payload = target.read_bytes()
vendor_root = (STATIC_ROOT / "vendor").resolve()
try:
target.relative_to(vendor_root)
cache_control = "public, max-age=86400"
except ValueError:
# 应用自身的 JS/CSS/HTML 每次重新校验,保证升级后浏览器立刻拿到新版。
cache_control = "no-cache"
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(payload)))
self.send_header("Cache-Control", cache_control)
self._security_headers()
self.end_headers()
if method != "HEAD":
self.wfile.write(payload)
# ------------------------------------------------------------ responses
def _security_headers(self) -> None:
self.send_header(
"Content-Security-Policy",
"default-src 'self'; img-src 'self' https: data:; "
"style-src 'self' 'unsafe-inline'; script-src 'self'; "
"connect-src 'self'; base-uri 'none'; frame-ancestors 'none'",
)
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("X-Frame-Options", "DENY")
def _send_json(self, status: int, data: Any) -> None:
payload = json.dumps({"ok": True, "data": data}, ensure_ascii=False).encode("utf-8")
self.send_response(int(status))
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.send_header("Cache-Control", "no-store")
self._security_headers()
self.end_headers()
if self.command != "HEAD":
self.wfile.write(payload)
def _send_error(self, status: int, code: str, message: str) -> None:
payload = json.dumps(
{"ok": False, "error": {"code": code, "message": message}},
ensure_ascii=False,
).encode("utf-8")
self.send_response(int(status))
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.send_header("Cache-Control", "no-store")
self._security_headers()
self.end_headers()
if self.command != "HEAD":
self.wfile.write(payload)
def _send_html(self, status: int, html: str) -> None:
payload = html.encode("utf-8")
self.send_response(int(status))
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.send_header("Cache-Control", "no-store")
self._security_headers()
self.end_headers()
if self.command != "HEAD":
self.wfile.write(payload)
def _redirect(self, location: str) -> None:
self.send_response(HTTPStatus.SEE_OTHER)
self.send_header("Location", location)
self.send_header("Content-Length", "0")
self._security_headers()
self.end_headers()
# ------------------------------------------------------------ helpers
def _safe_next(value: str | None) -> str:
text = (value or "").strip()
if not text.startswith("/") or text.startswith("//") or text.startswith("/login"):
return "/"
return text
def _json_path(body: bytes) -> str:
"""从 JSON(或表单编码)请求体里取 ``path`` 字段;失败返回空串。"""
text = body.decode("utf-8", errors="replace").strip()
if not text:
return ""
try:
data = json.loads(text)
except (ValueError, TypeError):
data = None
if isinstance(data, Mapping):
value = data.get("path")
return value if isinstance(value, str) else ""
values = parse_qs(text).get("path")
return values[0] if values else ""
def _param(query: Mapping[str, list[str]], name: str, default: str) -> str:
values = query.get(name)
return default if not values else str(values[0])
def _optional_str(query: Mapping[str, list[str]], name: str) -> str | None:
value = _param(query, name, "").strip()
return value or None
def _optional_int(query: Mapping[str, list[str]], name: str) -> int | None:
value = _optional_str(query, name)
if value is None:
return None
try:
return int(value)
except ValueError as exc:
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", f"{name} 必须是整数") from exc
def _year_param(query: Mapping[str, list[str]]) -> int | None:
value = _optional_int(query, "year")
if value is None:
return None
if value < 2000 or value > 2100:
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "year 必须在 2000~2100 之间")
return value
def _int_param(
query: Mapping[str, list[str]],
name: str,
default: int,
minimum: int,
maximum: int,
) -> int:
value = _optional_int(query, name)
if value is None:
return int(default)
if value < minimum or value > maximum:
raise ApiError(
HTTPStatus.BAD_REQUEST, "bad_request",
f"{name} 必须在 {minimum}~{maximum} 之间",
)
return value
def _path_int(value: str, name: str) -> int:
try:
return int(value)
except ValueError as exc:
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", f"{name} 必须是整数") from exc
def _range_days(query: Mapping[str, list[str]]) -> int | None:
raw = _optional_str(query, "range")
if raw in {None, "all", "0"}:
return None
if raw in {"7", "30", "90"}:
return int(raw)
days = _optional_int(query, "days")
if days is None:
return 30
if days < 1 or days > 3650:
raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "days 超出范围")
return days
def _preview_marker(db: Database) -> bool:
def run(conn):
row = conn.execute("SELECT v FROM kv WHERE k = 'preview:seeded'").fetchone()
return row is not None and str(row["v"]) == "1"
return bool(db.read(run))
def create_server(
config_store: ConfigStore,
db: Database,
*,
host: str | None = None,
port: int | None = None,
preview: bool = False,
logger: Any = None,
clock: Callable[[], float] = time.time,
) -> DashboardHTTPServer:
cfg = config_store.get()
bind_host = cfg.web.host if host is None else host
bind_port = cfg.web.port if port is None else int(port)
if not cfg.web.auth_token and not is_loopback(bind_host):
raise ValueError(
"非回环地址必须配置 web.auth_token,或只绑定 127.0.0.1"
)
if preview and not _preview_marker(db):
raise ValueError("目标数据库不是预览库(缺少 preview:seeded=1 标记)")
track = cfg.web.track
visits = VisitCollector(
db,
clock=clock,
flush_seconds=track.flush_seconds,
keep_days=track.keep_days,
max_hits_per_key=track.max_hits_per_ip_path_per_day,
logger=logger,
)
context = WebContext(
queries=DashboardQueries(db, config_store, clock=clock),
config_store=config_store,
preview=preview,
logger=logger,
clock=clock,
db_path=str(db.path),
schema_version=db.schema_version,
visits=visits,
)
server = DashboardHTTPServer((bind_host, bind_port), context)
visits.start()
return server
def serve_forever(server: DashboardHTTPServer, *, logger: Any = None) -> None:
host, port = server.server_address[:2]
if logger is not None:
logger.info(
"web.start",
url=f"http://{host}:{port}",
preview=server.context.preview,
auth=bool(server.context.auth_token),
)
try:
server.serve_forever()
finally:
server.server_close()