da7d2fb4b4
安全审计发现： - WS receive_text 无应用层大小限制，恶意客户端可发超大帧（uvicorn 默认 16MB 才拦截）占内存。加 512KB 单帧检查（与 content 256KB 限制对齐留余量）。 - 单 board 无连接数上限，恶意脚本可开海量连接耗尽资源。config 加 max_connections_per_board（默认 50），hub.register 超限返回 False， controller 回 error 帧并关闭连接。 审计结论（无需修复）： - SQL：全部 SQLAlchemy ORM 参数化，无注入。 - 路径穿越：storage_path 服务端生成（uuid+basename(ext)），用户不可控分隔符。 - 鉴权：管理类操作均有 require_docs_auth，公开写接口符合设计。 - 下载 filename CRLF：Starlette FileResponse 用 quote() 编码，无响应拆分。 - XSS：前端 el() 用 createTextNode，textarea 纯文本，innerHTML 仅用于静态文案。
149 lines
4.6 KiB
Python
149 lines
4.6 KiB
Python
"""运行时配置：所有参数与凭据均从项目根目录的 config.yaml 读取。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
from pydantic import BaseModel, Field, field_validator
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
DEFAULT_CONFIG_PATH = PROJECT_ROOT / "config.yaml"
|
||
|
||
|
||
class ServerConfig(BaseModel):
|
||
host: str = "127.0.0.1"
|
||
port: int = 6867
|
||
workers: int = 1
|
||
|
||
|
||
class DatabaseConfig(BaseModel):
|
||
host: str = "127.0.0.1"
|
||
port: int = 3306
|
||
user: str = "zikai_filesvc"
|
||
password: str = ""
|
||
database: str = "zikai_filesvc"
|
||
pool_size: int = 5
|
||
pool_recycle: int = 1800
|
||
|
||
|
||
class StorageConfig(BaseModel):
|
||
upload_dir: str = "./uploads"
|
||
chunk_bytes: int = 1024 * 1024
|
||
sha256_on_upload: bool = True
|
||
# 分片上传会话的暂存目录（相对 upload_dir），完整文件拼接在此完成
|
||
chunk_session_dir: str = "./.work"
|
||
# 分片上传会话的存活秒数：超过该时长无活动的 pending 会话由后台 reaper 清理
|
||
chunk_session_ttl_seconds: int = 300
|
||
|
||
|
||
class SftpUser(BaseModel):
|
||
username: str
|
||
password_hash: str = "" # bcrypt
|
||
public_key: str | None = None # 可选的内联 OpenSSH 公钥
|
||
|
||
|
||
class SftpConfig(BaseModel):
|
||
enabled: bool = True
|
||
host: str = "0.0.0.0"
|
||
port: int = 2022
|
||
host_key_path: str = "./keys/ssh_host_ed25519_key"
|
||
authorized_keys_path: str = "./keys/authorized_keys"
|
||
users: list[SftpUser] = Field(default_factory=list)
|
||
|
||
|
||
class DocsConfig(BaseModel):
|
||
"""/docs、/redoc、/openapi.json 的 Basic Auth 凭据（明文）。"""
|
||
|
||
enabled: bool = True
|
||
username: str = "admin"
|
||
password: str = "" # 留空 → /docs 返回 503，避免误暴露
|
||
realm: str = "zikai docs"
|
||
|
||
@field_validator("password", mode="before")
|
||
@classmethod
|
||
def _coerce(cls, v):
|
||
# 兼容 YAML 把纯数字密码解析成 int 的情况。
|
||
return "" if v is None else str(v)
|
||
|
||
|
||
class TunnelUser(BaseModel):
|
||
"""一个反向隧道用户：凭据 + 固定的隧道端口与本地端口。"""
|
||
|
||
username: str
|
||
password_hash: str = "" # bcrypt，与 SFTP 用户同款
|
||
# 该 user 在 server 侧绑定的隧道端口（SSH remote forward 的 listen port）
|
||
tunnel_port: int = 0
|
||
# 该 user 要暴露的本地服务端口（仅用于记录，实际转发由 user 端完成）
|
||
local_port: int = 0
|
||
|
||
|
||
class TunnelConfig(BaseModel):
|
||
"""反向隧道总开关与用户列表。"""
|
||
|
||
enabled: bool = False
|
||
users: list[TunnelUser] = Field(default_factory=list)
|
||
|
||
def find_user(self, username: str) -> TunnelUser | None:
|
||
return next((u for u in self.users if u.username == username), None)
|
||
|
||
|
||
class WhiteboardConfig(BaseModel):
|
||
"""共享白板配置。
|
||
|
||
白板本身无鉴权（任何人凭 /whiteboard/{id} 即可访问并实时协作）；
|
||
管理页（/whiteboard-admin、/api/admin/whiteboards）走 docs 同款 Basic Auth。
|
||
心跳按 heartbeat_interval_seconds 发送，连续丢失 heartbeat_miss_threshold 次即判失活。
|
||
"""
|
||
|
||
heartbeat_interval_seconds: int = 3
|
||
heartbeat_miss_threshold: int = 5
|
||
# board_id 合法字符集与长度上限，防路径/注入
|
||
max_board_id_length: int = 64
|
||
# 单 board 并发连接上限，防资源耗尽（同 board 同时在线人数）
|
||
max_connections_per_board: int = 50
|
||
# 列表/管理页分页默认值
|
||
list_limit: int = 100
|
||
|
||
|
||
class Settings(BaseModel):
|
||
server: ServerConfig = ServerConfig()
|
||
database: DatabaseConfig = DatabaseConfig()
|
||
storage: StorageConfig = StorageConfig()
|
||
sftp: SftpConfig = SftpConfig()
|
||
docs: DocsConfig = DocsConfig()
|
||
tunnel: TunnelConfig = TunnelConfig()
|
||
whiteboard: WhiteboardConfig = WhiteboardConfig()
|
||
|
||
def db_url(self) -> str:
|
||
c = self.database
|
||
return (
|
||
f"mysql+pymysql://{c.user}:{c.password}@{c.host}:{c.port}/{c.database}"
|
||
"?charset=utf8mb4"
|
||
)
|
||
|
||
def resolved_upload_dir(self) -> Path:
|
||
return (PROJECT_ROOT / self.storage.upload_dir).resolve()
|
||
|
||
|
||
def _load_yaml(path: Path) -> dict:
|
||
if not path.exists():
|
||
raise FileNotFoundError(
|
||
f"未找到配置文件 {path}；请先 cp config.example.yaml config.yaml 并填值。"
|
||
)
|
||
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def get_settings() -> Settings:
|
||
path = Path(os.getenv("CONFIG_PATH", str(DEFAULT_CONFIG_PATH)))
|
||
return Settings.model_validate(_load_yaml(path))
|
||
|
||
|
||
def reload_settings() -> Settings:
|
||
"""清缓存并重新读取，供脚本与测试使用。"""
|
||
get_settings.cache_clear()
|
||
return get_settings()
|