chore: 忽略离线部署包,提交安全加固、数据库初始化与文档

- .gitignore: 忽略 docker/offline 离线部署包(镜像/运行时等大文件)
- 安全加固: 新增 compute/api/security.py 及各端安全测试,补充 docs/security-hardening.md
- 数据库: 新增完整初始化 SQL 与 docs/database-config.md
- 数据转换与评测: 修复类型检查、增强校验并补充测试
- Docker 配置与环境变量更新

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-08-07 09:24:35 +08:00
parent e397bcc2ca
commit 75cc105ebc
24 changed files with 1850 additions and 50 deletions

3
.gitignore vendored
View File

@@ -208,3 +208,6 @@ docker/compute/data/yg-ft/logs/**
!docker/compute/data/yg-ft/logs/training/ !docker/compute/data/yg-ft/logs/training/
!docker/compute/data/yg-ft/**/.gitkeep !docker/compute/data/yg-ft/**/.gitkeep
!docker/compute/data/yg-ft/**/README.md !docker/compute/data/yg-ft/**/README.md
# Offline deployment bundle - 离线部署包(镜像、运行时等大文件,不提交)
docker/offline/

View File

@@ -1,6 +1,7 @@
from dataclasses import dataclass from dataclasses import dataclass
from functools import lru_cache from functools import lru_cache
import os import os
from typing import Any
try: try:
from pathlib import Path as _Path from pathlib import Path as _Path
@@ -29,6 +30,24 @@ def _list_env(name: str, default: list[str]) -> list[str]:
return [item.strip() for item in raw.split(",") if item.strip()] return [item.strip() for item in raw.split(",") if item.strip()]
def _bool_env(name: str, default: bool) -> bool:
raw = os.getenv(name)
if raw is None or raw.strip() == "":
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
def docs_kwargs(enabled: bool) -> dict[str, Any]:
"""Swagger UI / ReDoc / OpenAPI schema 路由开关。
关闭时 FastAPI 不注册 /docs、/redoc、/openapi.json访问一律返回 404
避免未授权访问泄露 API 结构。
"""
if enabled:
return {}
return {"docs_url": None, "redoc_url": None, "openapi_url": None}
@dataclass(frozen=True) @dataclass(frozen=True)
class Settings: class Settings:
app_name: str = os.getenv("APP_NAME", "YG Fine-Tune Platform API") app_name: str = os.getenv("APP_NAME", "YG Fine-Tune Platform API")
@@ -48,6 +67,7 @@ class Settings:
log_error_file_prefix: str = os.getenv("LOG_ERROR_FILE_PREFIX", "error") log_error_file_prefix: str = os.getenv("LOG_ERROR_FILE_PREFIX", "error")
log_max_bytes: int = _int_env("LOG_MAX_BYTES", 20 * 1024 * 1024) log_max_bytes: int = _int_env("LOG_MAX_BYTES", 20 * 1024 * 1024)
log_retention_days: int = _int_env("LOG_RETENTION_DAYS", 10) log_retention_days: int = _int_env("LOG_RETENTION_DAYS", 10)
enable_docs: bool = None # type: ignore[assignment]
def __post_init__(self) -> None: def __post_init__(self) -> None:
object.__setattr__( object.__setattr__(
@@ -63,6 +83,15 @@ class Settings:
], ],
), ),
) )
# Swagger UI / ReDoc / OpenAPI 文档路由开关:
# 未显式配置 ENABLE_DOCS 时,仅本地/开发环境开放,生产环境默认关闭,
# 避免未授权访问泄露 API 结构。从运行时环境读取 APP_ENV而非类定义时
# 缓存的默认值,保证生产默认关闭始终生效且便于测试。
object.__setattr__(
self,
"enable_docs",
_bool_env("ENABLE_DOCS", os.getenv("APP_ENV", "local") != "prod"),
)
@lru_cache @lru_cache

View File

@@ -239,10 +239,18 @@ def llama_dataset_info(dataset_key: str, file_names: list[str], formatting: str
fmt = str(formatting).lower() fmt = str(formatting).lower()
for key, file_name in zip(llama_dataset_keys(dataset_key, file_names), file_names): for key, file_name in zip(llama_dataset_keys(dataset_key, file_names), file_names):
if fmt == "sharegpt": if fmt == "sharegpt":
# 平台校验按 OpenAI 风格消息role/content故 tags 用 role/content
# 与 LLaMA-Factory 默认的 from/value 不同,需显式声明避免解析失败。
result[key] = { result[key] = {
"file_name": file_name, "file_name": file_name,
"formatting": "sharegpt", "formatting": "sharegpt",
"columns": {"messages": "messages"}, "columns": {"messages": "messages"},
"tags": {
"role_tag": "role",
"content_tag": "content",
"user_tag": "user",
"assistant_tag": "assistant",
},
} }
elif fmt == "dpo": elif fmt == "dpo":
result[key] = { result[key] = {
@@ -273,6 +281,47 @@ def llama_dataset_info(dataset_key: str, file_names: list[str], formatting: str
return result return result
def _sniff_dataset_format(sample_text: str, max_samples: int = 20) -> str:
"""嗅探数据集内容格式(兼容 jsonl返回 sharegpt / dpo / cpt / alpaca。
按内容而非文件名判断,纯 jsonl 数据集(如 ShareGPT messages、缺省 input 的
Alpaca都能被正确识别避免训练任务误按 alpaca 解析而失败。
"""
text = (sample_text or "").strip()
if not text:
return ""
records: list[dict[str, Any]] = []
try:
value = json.loads(text)
except (TypeError, ValueError, json.JSONDecodeError):
value = None
if isinstance(value, list):
records = [item for item in value[:max_samples] if isinstance(item, dict)]
elif isinstance(value, dict):
records = [value]
else:
for line in text.splitlines()[:max_samples]:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(obj, dict):
records.append(obj)
records = records[:max_samples]
if not records:
return ""
if all("messages" in record for record in records):
return "sharegpt"
if all(record.get("chosen") and record.get("rejected") for record in records):
return "dpo"
if all(record.get("text") and not (record.get("instruction") or record.get("output")) for record in records):
return "cpt"
return "alpaca"
PASSWORD_HASH_ITERATIONS = 390_000 PASSWORD_HASH_ITERATIONS = 390_000
@@ -2076,18 +2125,38 @@ class PlatformStore:
runtime_keys = llama_dataset_keys(dataset_key, runtime_file_names) runtime_keys = llama_dataset_keys(dataset_key, runtime_file_names)
training_keys = runtime_keys[: len(training_files)] training_keys = runtime_keys[: len(training_files)]
validation_keys = runtime_keys[len(training_files) :] validation_keys = runtime_keys[len(training_files) :]
dataset_format = str(task.get("dataset_format") or (dataset and dataset.get("formatting")) or "alpaca").lower() # P0-2: 推导数据集格式并校验内容(兼容 jsonl按内容嗅探 ShareGPT/DPO/CPT/Alpaca
# P0-2: Validate dataset content against declared format
train_type = str(task.get("train_type", task.get("train_method", ""))).upper() train_type = str(task.get("train_type", task.get("train_method", ""))).upper()
expected_format = { expected_format = {"DPO": "dpo", "CPT": "cpt"}.get(train_type)
"DPO": "dpo", raw_format = str(
"CPT": "cpt", task.get("dataset_format")
}.get(train_type) or dataset_metadata.get("format")
or (dataset and dataset.get("formatting"))
or "alpaca"
).lower()
sniffed_format = ""
content_samples: dict[str, str] = {}
if training_files:
with self.connect() as conn:
for file_entry in training_files:
sample_row = conn.execute(
"SELECT substr(content, 1, 400000) AS sample FROM dataset_files WHERE id=?",
(str(file_entry["id"]),),
).fetchone()
sample = (sample_row or {}).get("sample") or ""
content_samples[str(file_entry["id"])] = sample
if not sniffed_format:
sniffed_format = _sniff_dataset_format(sample)
known_formats = {"sharegpt", "dpo", "cpt", "pt", "pretrain"}
if expected_format: if expected_format:
dataset_format = expected_format dataset_format = expected_format
elif raw_format in known_formats:
dataset_format = raw_format
else:
dataset_format = sniffed_format or raw_format or "alpaca"
format_errors: list[str] = [] format_errors: list[str] = []
for file_entry in training_files: for file_entry in training_files:
content = file_entry.get("content") or "" content = content_samples.get(str(file_entry["id"])) or ""
if content: if content:
from app.modules.data_process.dataset_format import validate_dataset_format from app.modules.data_process.dataset_format import validate_dataset_format
file_errors = validate_dataset_format(dataset_format, content=content) file_errors = validate_dataset_format(dataset_format, content=content)

View File

@@ -0,0 +1,755 @@
-- ============================================================================
-- YG Fine-Tune Platform — PostgreSQL 完整初始化脚本(一键建库建表)
-- ============================================================================
-- 用途:切换到新的 PG 数据集时,一次性创建平台运行所需的全部数据库对象与
-- 基础种子数据(幂等,可重复执行)。
--
-- 覆盖范围(与运行时代码实际使用的表一致):
-- 001_platform_runtime.sql 平台核心表
-- 002_governance.sql 治理表(租户 / 审批 / 审计 / 留存)
-- 003_tenant_quota.sql 租户配额列
-- 003_model_path_governance.sql 模型可训练标识列
-- 002_data_process.sql 数据处理表 + 数据集扩展列
-- 本文件补充data_convert_tasks数据转换任务运行时代码引用但原脚本缺失
-- 种子数据admin / operator 两个初始用户
--
-- 说明:
-- * 本脚本通过 psql 执行,包含 DO $$ ... $$ 块与事务,不能用应用的
-- executescript()(按分号切分)执行。
-- * 应用启动时 PlatformStore.ensure_schema() 只会自动执行
-- 001 / 002_governance / 003_tenant_quota数据处理表需另跑
-- 002_data_process.sql本脚本已包含。应用首次启动还会自动补充
-- admin/operator 种子用户(本脚本已包含,二选一即可)。
-- * 脚本内所有 DDL 均使用 IF NOT EXISTS / ADD COLUMN IF NOT EXISTS
-- 可在已初始化的库上安全重复执行。
--
-- 执行步骤(详见 docs/database-config.md
-- 1. 以超级用户创建角色与数据库(必须单独执行,不能放进事务):
-- CREATE ROLE yg_ft LOGIN PASSWORD '请改为强密码';
-- CREATE DATABASE yg_ft OWNER yg_ft;
-- 2. 连接目标库执行本脚本:
-- psql "postgresql://yg_ft:密码@<host>:5432/yg_ft" -f backend/app/db/sql/000_full_init.sql
-- 3. 可选:为 superuser 授权
-- ALTER ROLE yg_ft SUPERUSER; -- 仅当需要执行 CREATE EXTENSION 等
-- ============================================================================
BEGIN;
-- ============================================================================
-- 一、平台核心表来源001_platform_runtime.sql
-- ============================================================================
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
display_name TEXT NOT NULL,
role TEXT NOT NULL,
status TEXT NOT NULL,
permissions TEXT NOT NULL,
create_time TEXT NOT NULL,
last_login TEXT,
protected INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS models (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
type TEXT NOT NULL,
purpose TEXT NOT NULL,
model_source TEXT NOT NULL,
description TEXT,
path TEXT,
api_url TEXT,
api_key TEXT,
online_model_name TEXT,
can_train INTEGER NOT NULL DEFAULT 0,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS trained_models (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
train_methods TEXT NOT NULL,
base_model_path TEXT,
create_time TEXT NOT NULL,
merged INTEGER NOT NULL DEFAULT 0,
merging INTEGER NOT NULL DEFAULT 0,
merged_path TEXT,
artifact_dir TEXT,
compute_node_id TEXT,
compute_node_name TEXT
);
CREATE TABLE IF NOT EXISTS model_lineage (
id TEXT PRIMARY KEY,
child_resource_type TEXT NOT NULL,
child_resource_id TEXT NOT NULL,
parent_resource_type TEXT NOT NULL,
parent_resource_id TEXT NOT NULL,
relation_type TEXT NOT NULL,
compute_job_id TEXT,
payload TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS model_artifacts (
id TEXT PRIMARY KEY,
model_id TEXT NOT NULL,
model_kind TEXT NOT NULL,
artifact_type TEXT NOT NULL,
path TEXT NOT NULL,
size_bytes BIGINT NOT NULL DEFAULT 0,
checksum_sha256 TEXT,
metadata TEXT NOT NULL,
compute_job_id TEXT,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS model_export_jobs (
id TEXT PRIMARY KEY,
trained_model_id TEXT,
compute_job_id TEXT NOT NULL,
node_id TEXT,
export_type TEXT NOT NULL,
quantization_bit INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL,
output_dir TEXT,
payload TEXT NOT NULL,
create_time TEXT NOT NULL,
completed_at TEXT
);
CREATE TABLE IF NOT EXISTS datasets (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
type TEXT NOT NULL,
storage_type TEXT NOT NULL,
source TEXT NOT NULL,
task_id TEXT,
size TEXT,
count INTEGER NOT NULL DEFAULT 0,
description TEXT,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS dataset_files (
id TEXT PRIMARY KEY,
dataset_id TEXT NOT NULL REFERENCES datasets(id) ON DELETE CASCADE,
name TEXT NOT NULL,
size TEXT,
content TEXT NOT NULL,
active_version_id TEXT NOT NULL,
versions TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS compute_nodes (
id TEXT PRIMARY KEY,
code TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
api_base_url TEXT NOT NULL,
file_gateway_url TEXT NOT NULL,
enabled INTEGER NOT NULL DEFAULT 1,
scheduler_status TEXT NOT NULL,
scheduler_weight INTEGER NOT NULL DEFAULT 100,
tags TEXT NOT NULL,
gpu_count INTEGER NOT NULL DEFAULT 0,
current_running_jobs INTEGER NOT NULL DEFAULT 0,
max_parallel_jobs INTEGER NOT NULL DEFAULT 2,
data_root TEXT NOT NULL,
model_root TEXT NOT NULL,
log_root TEXT NOT NULL,
api_version TEXT NOT NULL DEFAULT 'v1',
capabilities TEXT NOT NULL DEFAULT '[]',
description TEXT,
last_health_check_at TEXT,
health_detail TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS gpus (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE,
gpu_index INTEGER NOT NULL,
uuid TEXT NOT NULL,
name TEXT NOT NULL,
memory_total_gb DOUBLE PRECISION NOT NULL,
power_limit_w DOUBLE PRECISION NOT NULL,
base_temperature INTEGER NOT NULL,
last_seen_at TEXT
);
CREATE TABLE IF NOT EXISTS fine_tune_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
payload TEXT NOT NULL,
status TEXT NOT NULL,
progress INTEGER NOT NULL DEFAULT 0,
process_id INTEGER,
create_time TEXT NOT NULL,
start_time TEXT,
completed_at TEXT,
compute_node_id TEXT REFERENCES compute_nodes(id) ON DELETE SET NULL,
gpus TEXT NOT NULL,
sync_job_id TEXT,
compute_job_id TEXT
);
CREATE TABLE IF NOT EXISTS fine_tune_metrics (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES fine_tune_tasks(id) ON DELETE CASCADE,
step INTEGER NOT NULL,
epoch DOUBLE PRECISION,
loss DOUBLE PRECISION,
grad_norm DOUBLE PRECISION,
learning_rate DOUBLE PRECISION,
raw TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS fine_tune_checkpoints (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES fine_tune_tasks(id) ON DELETE CASCADE,
step INTEGER NOT NULL,
name TEXT NOT NULL,
path TEXT NOT NULL,
size_bytes BIGINT NOT NULL DEFAULT 0,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS compute_jobs (
id TEXT PRIMARY KEY,
task_id TEXT REFERENCES fine_tune_tasks(id) ON DELETE SET NULL,
node_id TEXT REFERENCES compute_nodes(id) ON DELETE SET NULL,
engine TEXT NOT NULL,
status TEXT NOT NULL,
command TEXT NOT NULL,
output_dir TEXT,
log_file TEXT,
payload TEXT NOT NULL,
create_time TEXT NOT NULL,
update_time TEXT NOT NULL,
completed_at TEXT
);
CREATE TABLE IF NOT EXISTS gpu_allocations (
id TEXT PRIMARY KEY,
task_id TEXT REFERENCES fine_tune_tasks(id) ON DELETE CASCADE,
compute_job_id TEXT,
node_id TEXT REFERENCES compute_nodes(id) ON DELETE CASCADE,
gpu_index INTEGER NOT NULL,
status TEXT NOT NULL,
create_time TEXT NOT NULL,
released_at TEXT
);
CREATE TABLE IF NOT EXISTS scheduler_locks (
lock_key TEXT PRIMARY KEY,
owner TEXT NOT NULL,
expires_at TEXT NOT NULL,
create_time TEXT NOT NULL,
update_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS resource_replicas (
id TEXT PRIMARY KEY,
node_id TEXT NOT NULL REFERENCES compute_nodes(id) ON DELETE CASCADE,
resource_type TEXT NOT NULL,
resource_id TEXT NOT NULL,
local_path TEXT NOT NULL,
status TEXT NOT NULL,
sync_status TEXT NOT NULL,
checksum_sha256 TEXT,
byte_size BIGINT NOT NULL DEFAULT 0,
last_checked_at TEXT,
last_error TEXT,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS resource_sync_jobs (
id TEXT PRIMARY KEY,
target_node_id TEXT NOT NULL,
resources TEXT NOT NULL,
status TEXT NOT NULL,
progress INTEGER NOT NULL DEFAULT 0,
create_time TEXT NOT NULL,
completed_at TEXT
);
CREATE TABLE IF NOT EXISTS eval_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS eval_dimensions (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
is_active INTEGER NOT NULL DEFAULT 1,
is_default INTEGER NOT NULL DEFAULT 0,
create_time TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS compare_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT NOT NULL,
create_time TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_fine_tune_status ON fine_tune_tasks(status);
CREATE INDEX IF NOT EXISTS idx_fine_tune_compute_job ON fine_tune_tasks(compute_job_id);
CREATE INDEX IF NOT EXISTS idx_fine_tune_compute_node_status ON fine_tune_tasks(compute_node_id, status);
CREATE INDEX IF NOT EXISTS idx_model_lineage_child ON model_lineage(child_resource_type, child_resource_id);
CREATE INDEX IF NOT EXISTS idx_model_lineage_parent ON model_lineage(parent_resource_type, parent_resource_id);
CREATE INDEX IF NOT EXISTS idx_model_artifacts_model ON model_artifacts(model_kind, model_id, artifact_type);
CREATE INDEX IF NOT EXISTS idx_model_export_jobs_model ON model_export_jobs(trained_model_id, create_time DESC);
CREATE INDEX IF NOT EXISTS idx_model_export_jobs_compute ON model_export_jobs(compute_job_id);
CREATE INDEX IF NOT EXISTS idx_fine_tune_metrics_task_step ON fine_tune_metrics(task_id, step);
CREATE UNIQUE INDEX IF NOT EXISTS uq_fine_tune_metrics_task_step_epoch ON fine_tune_metrics(task_id, step, epoch);
CREATE INDEX IF NOT EXISTS idx_fine_tune_checkpoints_task_step ON fine_tune_checkpoints(task_id, step);
CREATE UNIQUE INDEX IF NOT EXISTS uq_fine_tune_checkpoints_task_path ON fine_tune_checkpoints(task_id, path);
CREATE INDEX IF NOT EXISTS idx_compute_jobs_task ON compute_jobs(task_id);
CREATE INDEX IF NOT EXISTS idx_compute_jobs_node_status ON compute_jobs(node_id, status);
CREATE INDEX IF NOT EXISTS idx_gpu_allocations_node_status ON gpu_allocations(node_id, status);
CREATE UNIQUE INDEX IF NOT EXISTS uq_gpu_allocations_active ON gpu_allocations(node_id, gpu_index) WHERE status IN ('allocated','running');
CREATE INDEX IF NOT EXISTS idx_scheduler_locks_expires ON scheduler_locks(expires_at);
CREATE INDEX IF NOT EXISTS idx_dataset_files_dataset ON dataset_files(dataset_id);
CREATE INDEX IF NOT EXISTS idx_gpus_node ON gpus(node_id);
CREATE UNIQUE INDEX IF NOT EXISTS uq_gpus_node_index ON gpus(node_id, gpu_index);
CREATE INDEX IF NOT EXISTS idx_replicas_resource ON resource_replicas(resource_type, resource_id);
CREATE UNIQUE INDEX IF NOT EXISTS uq_replicas_node_resource ON resource_replicas(node_id, resource_type, resource_id);
CREATE INDEX IF NOT EXISTS idx_sync_jobs_node_status ON resource_sync_jobs(target_node_id, status);
CREATE INDEX IF NOT EXISTS idx_eval_tasks_status ON eval_tasks(status);
CREATE INDEX IF NOT EXISTS idx_eval_dimensions_active ON eval_dimensions(is_active);
CREATE INDEX IF NOT EXISTS idx_compare_tasks_status ON compare_tasks(status);
-- ---- 项目 / 租户 ----
CREATE TABLE IF NOT EXISTS projects (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL DEFAULT 'default',
name TEXT NOT NULL,
code TEXT NOT NULL,
description TEXT,
quota TEXT,
status TEXT NOT NULL DEFAULT 'active',
create_time TEXT NOT NULL,
create_by TEXT,
updated_at TEXT
);
CREATE TABLE IF NOT EXISTS project_members (
project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
role TEXT NOT NULL DEFAULT 'member',
create_time TEXT NOT NULL,
PRIMARY KEY (project_id, user_id)
);
CREATE TABLE IF NOT EXISTS roles (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
permissions TEXT NOT NULL DEFAULT '[]',
create_time TEXT
);
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
issued_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
ip TEXT
);
CREATE TABLE IF NOT EXISTS acls (
id TEXT PRIMARY KEY,
resource_type TEXT NOT NULL,
resource_id TEXT NOT NULL,
principal_type TEXT NOT NULL,
principal_id TEXT NOT NULL,
permission TEXT NOT NULL,
create_time TEXT
);
-- ============================================================================
-- 二、治理表来源002_governance.sql
-- ============================================================================
CREATE TABLE IF NOT EXISTS tenants (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
code TEXT,
status TEXT DEFAULT 'active',
owner_user_id TEXT,
quota TEXT,
retention_policy_id TEXT,
create_time TEXT
);
CREATE TABLE IF NOT EXISTS approval_templates (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
steps TEXT,
create_time TEXT
);
CREATE TABLE IF NOT EXISTS approval_instances (
id TEXT PRIMARY KEY,
template_id TEXT,
resource_type TEXT,
resource_id TEXT,
applicant_id TEXT,
status TEXT DEFAULT 'pending',
current_step INTEGER DEFAULT 0,
create_time TEXT
);
CREATE TABLE IF NOT EXISTS approval_steps (
id TEXT PRIMARY KEY,
instance_id TEXT,
step_index INTEGER,
approver_id TEXT,
status TEXT DEFAULT 'pending',
comment TEXT,
time TEXT
);
CREATE TABLE IF NOT EXISTS audit_logs (
id TEXT PRIMARY KEY,
tenant_id TEXT,
project_id TEXT,
actor_id TEXT,
action TEXT,
target_type TEXT,
target_id TEXT,
detail TEXT,
client_ip TEXT,
time TEXT
);
CREATE INDEX IF NOT EXISTS idx_audit_tenant ON audit_logs(tenant_id);
CREATE INDEX IF NOT EXISTS idx_audit_project ON audit_logs(project_id);
CREATE INDEX IF NOT EXISTS idx_audit_action ON audit_logs(action);
CREATE INDEX IF NOT EXISTS idx_audit_time ON audit_logs(time);
CREATE TABLE IF NOT EXISTS retention_policies (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
scope TEXT,
rule TEXT,
status TEXT DEFAULT 'active',
create_time TEXT,
create_by TEXT,
updated_at TEXT
);
-- ============================================================================
-- 三、租户配额扩展来源003_tenant_quota.sql
-- ============================================================================
ALTER TABLE tenants ADD COLUMN IF NOT EXISTS gpu_quota TEXT;
ALTER TABLE tenants ADD COLUMN IF NOT EXISTS storage_quota TEXT;
-- ============================================================================
-- 四、模型路径治理来源003_model_path_governance.sql
-- models.can_train 已在建表语句中声明;以下为兼容旧库的幂等语句。
-- ============================================================================
ALTER TABLE models ADD COLUMN IF NOT EXISTS can_train INTEGER NOT NULL DEFAULT 0;
ALTER TABLE trained_models ADD COLUMN IF NOT EXISTS artifact_dir TEXT;
-- 按规则推定已有模型的 can_train新库为空表此语句为 no-op
UPDATE models
SET can_train = CASE
WHEN path IS NOT NULL AND path != '' AND model_source IS NOT NULL AND model_source != 'api' THEN 1
ELSE 0
END;
-- ============================================================================
-- 五、数据处理来源002_data_process.sql去掉其外层 BEGIN/COMMIT
-- 数据集扩展列
-- ============================================================================
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS source_task_id TEXT;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS size_bytes BIGINT NOT NULL DEFAULT 0;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS record_count BIGINT NOT NULL DEFAULT 0;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS tenant_id TEXT;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS project_id TEXT;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS owner_id TEXT;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_by TEXT;
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now();
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now();
ALTER TABLE datasets ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS storage_object_id TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS current_version_id TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS size_bytes BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS record_count BIGINT NOT NULL DEFAULT 0;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS file_format VARCHAR(40);
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS checksum_sha256 CHAR(64);
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS version_no INTEGER NOT NULL DEFAULT 1;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS source_task_id TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS tenant_id TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS project_id TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS created_by TEXT;
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now();
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now();
ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
-- ---- 数据处理任务 / 源文件 / 预览 / 结果 ----
CREATE TABLE IF NOT EXISTS data_process_tasks (
id TEXT PRIMARY KEY,
name VARCHAR(150) NOT NULL,
description TEXT,
status VARCHAR(20) NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'running', 'completed', 'failed', 'stopped')),
process_type VARCHAR(20) NOT NULL
CHECK (process_type IN ('structured', 'unstructured', 'external')),
source_dataset_id TEXT REFERENCES datasets(id) ON DELETE SET NULL,
output_dataset_id TEXT REFERENCES datasets(id) ON DELETE SET NULL,
config TEXT NOT NULL DEFAULT '{}',
progress NUMERIC(5,2) NOT NULL DEFAULT 0 CHECK (progress >= 0 AND progress <= 100),
input_count BIGINT NOT NULL DEFAULT 0 CHECK (input_count >= 0),
output_count BIGINT NOT NULL DEFAULT 0 CHECK (output_count >= 0),
filtered_count BIGINT NOT NULL DEFAULT 0 CHECK (filtered_count >= 0),
duplicate_count BIGINT NOT NULL DEFAULT 0 CHECK (duplicate_count >= 0),
error_count BIGINT NOT NULL DEFAULT 0 CHECK (error_count >= 0),
failure_reason TEXT,
generation_run_id TEXT,
results_confirmed BOOLEAN NOT NULL DEFAULT TRUE,
workflow_step VARCHAR(20) NOT NULL DEFAULT 'create'
CHECK (workflow_step IN ('create', 'model', 'upload', 'preview', 'generate', 'results')),
preview_status VARCHAR(20) NOT NULL DEFAULT 'idle'
CHECK (preview_status IN ('idle', 'queued', 'running', 'completed', 'failed', 'cancelled')),
preview_progress NUMERIC(5,2) NOT NULL DEFAULT 0
CHECK (preview_progress >= 0 AND preview_progress <= 100),
preview_run_id TEXT,
preview_failure_reason TEXT,
preview_total_files INTEGER NOT NULL DEFAULT 0 CHECK (preview_total_files >= 0),
preview_completed_files INTEGER NOT NULL DEFAULT 0 CHECK (preview_completed_files >= 0),
tenant_id TEXT,
project_id TEXT,
owner_id TEXT,
approval_status VARCHAR(30) NOT NULL DEFAULT 'not_required',
created_by TEXT,
updated_by TEXT,
deleted_by TEXT,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ
);
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS generation_run_id TEXT;
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS results_confirmed BOOLEAN NOT NULL DEFAULT TRUE;
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS workflow_step VARCHAR(20);
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_status VARCHAR(20);
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_progress NUMERIC(5,2);
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_run_id TEXT;
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_failure_reason TEXT;
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_total_files INTEGER;
ALTER TABLE data_process_tasks ADD COLUMN IF NOT EXISTS preview_completed_files INTEGER;
ALTER TABLE data_process_tasks ALTER COLUMN workflow_step SET DEFAULT 'create';
ALTER TABLE data_process_tasks ALTER COLUMN workflow_step SET NOT NULL;
ALTER TABLE data_process_tasks ALTER COLUMN preview_status SET DEFAULT 'idle';
ALTER TABLE data_process_tasks ALTER COLUMN preview_status SET NOT NULL;
ALTER TABLE data_process_tasks ALTER COLUMN preview_progress SET DEFAULT 0;
ALTER TABLE data_process_tasks ALTER COLUMN preview_progress SET NOT NULL;
ALTER TABLE data_process_tasks ALTER COLUMN preview_total_files SET DEFAULT 0;
ALTER TABLE data_process_tasks ALTER COLUMN preview_total_files SET NOT NULL;
ALTER TABLE data_process_tasks ALTER COLUMN preview_completed_files SET DEFAULT 0;
ALTER TABLE data_process_tasks ALTER COLUMN preview_completed_files SET NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS uq_data_process_tasks_name_alive
ON data_process_tasks(name) WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_data_process_tasks_scope_status
ON data_process_tasks(tenant_id, project_id, status, created_at DESC)
WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_data_process_tasks_creator_created
ON data_process_tasks(created_by, created_at DESC) WHERE deleted_at IS NULL;
CREATE TABLE IF NOT EXISTS data_process_source_files (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
storage_object_id TEXT,
name TEXT NOT NULL,
size_bytes BIGINT NOT NULL DEFAULT 0 CHECK (size_bytes >= 0),
record_count BIGINT NOT NULL DEFAULT 0 CHECK (record_count >= 0),
file_format VARCHAR(40),
checksum_sha256 CHAR(64) NOT NULL,
version_no INTEGER NOT NULL DEFAULT 1 CHECK (version_no > 0),
content TEXT NOT NULL,
content_preview TEXT,
metadata TEXT NOT NULL DEFAULT '{}',
tenant_id TEXT,
project_id TEXT,
created_by TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_data_process_source_files_task
ON data_process_source_files(task_id, created_at) WHERE deleted_at IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS uq_data_process_source_checksum_alive
ON data_process_source_files(task_id, checksum_sha256) WHERE deleted_at IS NULL;
CREATE TABLE IF NOT EXISTS data_process_preview_items (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
source_file_id TEXT REFERENCES data_process_source_files(id) ON DELETE CASCADE,
original_content TEXT NOT NULL DEFAULT '',
edited_content TEXT NOT NULL DEFAULT '',
source_start INTEGER CHECK (source_start IS NULL OR source_start >= 0),
source_end INTEGER CHECK (source_end IS NULL OR source_end >= 0),
source_start_line INTEGER CHECK (source_start_line IS NULL OR source_start_line > 0),
source_end_line INTEGER CHECK (source_end_line IS NULL OR source_end_line > 0),
token_count INTEGER NOT NULL DEFAULT 0 CHECK (token_count >= 0),
status VARCHAR(20) NOT NULL DEFAULT 'original'
CHECK (status IN ('original', 'modified', 'manual', 'invalid')),
quality_score TEXT NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CHECK (source_start IS NULL OR source_end IS NULL OR source_end >= source_start),
CHECK (source_start_line IS NULL OR source_end_line IS NULL OR source_end_line >= source_start_line)
);
CREATE INDEX IF NOT EXISTS idx_data_process_preview_task_file
ON data_process_preview_items(task_id, source_file_id, created_at);
CREATE TABLE IF NOT EXISTS data_process_results (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES data_process_tasks(id) ON DELETE CASCADE,
preview_item_id TEXT REFERENCES data_process_preview_items(id) ON DELETE SET NULL,
instruction TEXT NOT NULL,
input TEXT NOT NULL DEFAULT '',
output TEXT NOT NULL,
original_instruction TEXT,
original_input TEXT,
original_output TEXT,
status VARCHAR(20) NOT NULL DEFAULT 'valid'
CHECK (status IN ('valid', 'modified', 'invalid')),
error TEXT,
split VARCHAR(20) CHECK (split IS NULL OR split IN ('train', 'validation', 'test')),
quality_score TEXT NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_status
ON data_process_results(task_id, status, id);
CREATE INDEX IF NOT EXISTS idx_data_process_results_task_split
ON data_process_results(task_id, split);
-- ---- 数据集版本 / 记录 ----
CREATE TABLE IF NOT EXISTS dataset_file_versions (
id TEXT PRIMARY KEY,
dataset_file_id TEXT NOT NULL REFERENCES dataset_files(id) ON DELETE CASCADE,
version_no INTEGER NOT NULL CHECK (version_no > 0),
storage_object_id TEXT NOT NULL,
content_preview TEXT,
description TEXT,
base_version_id TEXT REFERENCES dataset_file_versions(id) ON DELETE SET NULL,
size_bytes BIGINT NOT NULL DEFAULT 0 CHECK (size_bytes >= 0),
record_count BIGINT NOT NULL DEFAULT 0 CHECK (record_count >= 0),
checksum_sha256 CHAR(64) NOT NULL,
source_task_id TEXT REFERENCES data_process_tasks(id) ON DELETE SET NULL,
metadata TEXT NOT NULL DEFAULT '{}',
created_by TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE dataset_file_versions ADD COLUMN IF NOT EXISTS source_task_id TEXT;
ALTER TABLE dataset_file_versions ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}';
CREATE UNIQUE INDEX IF NOT EXISTS uq_dataset_file_versions_no_002
ON dataset_file_versions(dataset_file_id, version_no);
CREATE INDEX IF NOT EXISTS idx_dataset_file_versions_source_task_002
ON dataset_file_versions(source_task_id) WHERE source_task_id IS NOT NULL;
CREATE TABLE IF NOT EXISTS dataset_records (
id TEXT PRIMARY KEY,
dataset_id TEXT NOT NULL REFERENCES datasets(id) ON DELETE CASCADE,
dataset_file_id TEXT REFERENCES dataset_files(id) ON DELETE CASCADE,
version_id TEXT REFERENCES dataset_file_versions(id) ON DELETE CASCADE,
line_no INTEGER,
split VARCHAR(20) CHECK (split IS NULL OR split IN ('train', 'validation', 'test')),
instruction TEXT,
input TEXT,
output TEXT,
raw TEXT NOT NULL DEFAULT '{}',
status VARCHAR(20) NOT NULL DEFAULT 'valid'
CHECK (status IN ('valid', 'modified', 'invalid')),
source_task_id TEXT REFERENCES data_process_tasks(id) ON DELETE SET NULL,
source_result_id TEXT REFERENCES data_process_results(id) ON DELETE SET NULL,
preview_item_id TEXT REFERENCES data_process_preview_items(id) ON DELETE SET NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS source_task_id TEXT;
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS source_result_id TEXT;
ALTER TABLE dataset_records ADD COLUMN IF NOT EXISTS preview_item_id TEXT;
CREATE INDEX IF NOT EXISTS idx_dataset_records_dataset_002
ON dataset_records(dataset_id, id);
CREATE INDEX IF NOT EXISTS idx_dataset_records_source_task_002
ON dataset_records(source_task_id, source_result_id);
CREATE INDEX IF NOT EXISTS idx_datasets_source_task_002
ON datasets(source_task_id) WHERE source_task_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_dataset_files_source_task_002
ON dataset_files(source_task_id) WHERE source_task_id IS NOT NULL;
-- ============================================================================
-- 六、数据转换任务data_convert_tasks
-- 运行时 routerapp/modules/data_convert/router.py引用但原脚本缺失本文件补齐。
-- ============================================================================
CREATE TABLE IF NOT EXISTS data_convert_tasks (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
output_filename TEXT DEFAULT 'converted-data.jsonl',
status TEXT NOT NULL DEFAULT 'pending',
input_count INTEGER NOT NULL DEFAULT 0,
output_count INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
create_time TEXT NOT NULL DEFAULT (to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')),
update_time TEXT NOT NULL DEFAULT (to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')),
deleted_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_data_convert_tasks_status ON data_convert_tasks(status);
CREATE INDEX IF NOT EXISTS idx_data_convert_tasks_create_time ON data_convert_tasks(create_time DESC);
-- ============================================================================
-- 七、种子数据:初始管理员 / 操作员
-- 应用首次启动ensure_seed_data也会自动创建此处提供以便脱离应用直接初始化。
-- 密码admin / admin123operator / operator123上线前请改密
-- ============================================================================
INSERT INTO users
(id, username, password_hash, display_name, role, status, permissions, create_time, protected)
VALUES
(
'u_admin', 'admin', 'pbkdf2_sha256$390000$ygft_init_salt_admin$2b6f31f22968c4f5a30bcf0acf066b7a0f58d4773d15c5ab898ba715ea87b5bd',
'Platform Admin', 'admin', 'active',
'["dashboard","fine-tune","model-eval","model-inference","model-manage","dataset","data-process","data-convert","compute","hardware","logs","user-settings"]',
to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), 1
),
(
'u_operator', 'operator', 'pbkdf2_sha256$390000$ygft_init_salt_op$525bf35d02ed26f37952cbd6862b0ae358b9d1a7fa0cbbf0217aa2b5dd544125',
'Platform Operator', 'operator', 'active',
'["dashboard","fine-tune","model-eval","model-inference","model-manage","dataset","data-process","data-convert","compute","hardware","logs"]',
to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), 0
)
ON CONFLICT (username) DO NOTHING;
COMMIT;

View File

@@ -5,7 +5,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.router import api_router from app.api.v1.router import api_router
from app.core.config import get_settings from app.core.config import docs_kwargs, get_settings
from app.core.logging import configure_logging, setup_request_logging from app.core.logging import configure_logging, setup_request_logging
from app.workers.compute_poller import run_compute_poller from app.workers.compute_poller import run_compute_poller
@@ -14,7 +14,7 @@ def create_app() -> FastAPI:
settings = get_settings() settings = get_settings()
configure_logging(settings) configure_logging(settings)
app = FastAPI(title=settings.app_name) app = FastAPI(title=settings.app_name, **docs_kwargs(settings.enable_docs))
app.add_middleware( app.add_middleware(
CORSMiddleware, CORSMiddleware,
allow_origins=settings.cors_allow_origins, allow_origins=settings.cors_allow_origins,

View File

@@ -5,10 +5,11 @@ import os
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from fastapi import APIRouter, Body, UploadFile, File from fastapi import APIRouter, Body, Depends, File, UploadFile
from fastapi.responses import FileResponse from fastapi.responses import FileResponse
from app.api.v1.endpoints.platform import ok, fail from app.api.v1.endpoints.platform import ok, fail
from app.core.auth import get_current_user
from app.db.platform_store import get_platform_store, new_id from app.db.platform_store import get_platform_store, new_id
@@ -18,6 +19,30 @@ router = APIRouter(prefix="/data-convert", tags=["data-convert"])
STORAGE_ROOT = Path(__file__).resolve().parents[3] / "storage" / "data-convert" STORAGE_ROOT = Path(__file__).resolve().parents[3] / "storage" / "data-convert"
def _safe_output_filename(value: Any) -> str:
"""输出文件名白名单校验:仅允许普通文件名,阻断 ``../``、``/``、``\\`` 等路径穿越。
转换结果始终写入 ``STORAGE_ROOT/<task_id>/output/<output_filename>``
若文件名可被注入路径分隔符,将导致任意文件读写/删除。
"""
name = str(value or "converted-data.jsonl").strip()
if (
not name
or name in {".", ".."}
or name != Path(name).name
or "/" in name
or "\\" in name
or any(ord(character) < 32 or ord(character) == 127 for character in name)
):
raise fail(400, "output filename must be a plain file name")
return name
def _task_output_path(task: dict[str, Any]) -> Path:
"""返回经过白名单校验的转换输出文件路径(始终位于任务 output 目录内)。"""
return _output_dir(task["id"]) / _safe_output_filename(task.get("output_filename"))
def _task_dir(task_id: str) -> Path: def _task_dir(task_id: str) -> Path:
return STORAGE_ROOT / task_id return STORAGE_ROOT / task_id
@@ -31,7 +56,11 @@ def _output_dir(task_id: str) -> Path:
@router.get("") @router.get("")
def list_tasks(page: int = 1, page_size: int = 20) -> dict[str, Any]: def list_tasks(
page: int = 1,
page_size: int = 20,
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
store = get_platform_store() store = get_platform_store()
with store.connect() as conn: with store.connect() as conn:
rows = conn.execute( rows = conn.execute(
@@ -46,12 +75,15 @@ def list_tasks(page: int = 1, page_size: int = 20) -> dict[str, Any]:
@router.post("") @router.post("")
def create_task(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: def create_task(
payload: dict[str, Any] = Body(...),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
name = str(payload.get("name") or "").strip() name = str(payload.get("name") or "").strip()
if not name: if not name:
raise fail(400, "name is required") raise fail(400, "name is required")
task_id = new_id("dct") task_id = new_id("dct")
output_filename = str(payload.get("output_filename") or "converted-data.jsonl").strip() output_filename = _safe_output_filename(payload.get("output_filename"))
description = str(payload.get("description") or "").strip() description = str(payload.get("description") or "").strip()
store = get_platform_store() store = get_platform_store()
with store.connect() as conn: with store.connect() as conn:
@@ -67,7 +99,10 @@ def create_task(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
@router.get("/{task_id}") @router.get("/{task_id}")
def get_task(task_id: str) -> dict[str, Any]: def get_task(
task_id: str,
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
task = _get_task(task_id) task = _get_task(task_id)
if not task: if not task:
raise fail(404, "task not found") raise fail(404, "task not found")
@@ -86,6 +121,7 @@ def get_task(task_id: str) -> dict[str, Any]:
async def upload_source_files( async def upload_source_files(
task_id: str, task_id: str,
files: list[UploadFile] = File(...), files: list[UploadFile] = File(...),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]: ) -> dict[str, Any]:
task = _get_task(task_id) task = _get_task(task_id)
if not task: if not task:
@@ -114,7 +150,7 @@ async def upload_source_files(
try: try:
output_dir = _output_dir(task_id) output_dir = _output_dir(task_id)
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / (task["output_filename"] or "converted-data.jsonl") output_path = _task_output_path(task)
# 清空旧输出(如果重新上传) # 清空旧输出(如果重新上传)
if output_path.exists(): if output_path.exists():
output_path.unlink() output_path.unlink()
@@ -157,7 +193,7 @@ async def upload_source_files(
}) })
dataset_id = dataset["id"] dataset_id = dataset["id"]
with store.connect() as conn: with store.connect() as conn:
store.add_dataset_file(conn, dataset_id, task["output_filename"] or "converted-data.jsonl", content) store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content)
return ok({ return ok({
"staged_files": staged, "staged_files": staged,
"auto_converted": True, "auto_converted": True,
@@ -175,7 +211,10 @@ async def upload_source_files(
@router.post("/{task_id}/run") @router.post("/{task_id}/run")
def run_convert(task_id: str) -> dict[str, Any]: def run_convert(
task_id: str,
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
task = _get_task(task_id) task = _get_task(task_id)
if not task: if not task:
raise fail(404, "task not found") raise fail(404, "task not found")
@@ -192,7 +231,7 @@ def run_convert(task_id: str) -> dict[str, Any]:
input_dir = _input_dir(task_id) input_dir = _input_dir(task_id)
output_dir = _output_dir(task_id) output_dir = _output_dir(task_id)
output_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / (task["output_filename"] or "converted-data.jsonl") output_path = _task_output_path(task)
input_count = 0 input_count = 0
output_count = 0 output_count = 0
for json_file in sorted(input_dir.iterdir()): for json_file in sorted(input_dir.iterdir()):
@@ -229,19 +268,22 @@ def run_convert(task_id: str) -> dict[str, Any]:
@router.get("/{task_id}/download") @router.get("/{task_id}/download")
def download_result(task_id: str): def download_result(
task_id: str,
current_user: dict = Depends(get_current_user),
):
task = _get_task(task_id) task = _get_task(task_id)
if not task: if not task:
raise fail(404, "task not found") raise fail(404, "task not found")
if task["status"] != "completed": if task["status"] != "completed":
raise fail(400, "task is not completed") raise fail(400, "task is not completed")
output_path = _output_dir(task_id) / (task["output_filename"] or "converted-data.jsonl") output_path = _task_output_path(task)
if not output_path.exists(): if not output_path.exists():
raise fail(404, "output file not found") raise fail(404, "output file not found")
return FileResponse( return FileResponse(
str(output_path), str(output_path),
media_type="application/octet-stream", media_type="application/octet-stream",
filename=task["output_filename"] or "converted-data.jsonl", filename=_safe_output_filename(task.get("output_filename")),
) )
@@ -249,6 +291,7 @@ def download_result(task_id: str):
def import_as_dataset( def import_as_dataset(
task_id: str, task_id: str,
payload: dict[str, Any] = Body(default={}), payload: dict[str, Any] = Body(default={}),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]: ) -> dict[str, Any]:
"""把已转换的 JSONL 文件导入为数据集管理中的上传任务记录source='task')。""" """把已转换的 JSONL 文件导入为数据集管理中的上传任务记录source='task')。"""
task = _get_task(task_id) task = _get_task(task_id)
@@ -256,7 +299,7 @@ def import_as_dataset(
raise fail(404, "task not found") raise fail(404, "task not found")
if task["status"] != "completed": if task["status"] != "completed":
raise fail(400, "task is not completed") raise fail(400, "task is not completed")
output_path = _output_dir(task_id) / (task["output_filename"] or "converted-data.jsonl") output_path = _task_output_path(task)
if not output_path.exists(): if not output_path.exists():
raise fail(404, "output file not found") raise fail(404, "output file not found")
content = output_path.read_text(encoding="utf-8") content = output_path.read_text(encoding="utf-8")
@@ -277,12 +320,15 @@ def import_as_dataset(
}) })
dataset_id = dataset["id"] dataset_id = dataset["id"]
with store.connect() as conn: with store.connect() as conn:
store.add_dataset_file(conn, dataset_id, task["output_filename"] or "converted-data.jsonl", content) store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content)
return ok({"dataset_id": dataset_id, "name": dataset_name}) return ok({"dataset_id": dataset_id, "name": dataset_name})
@router.delete("/{task_id}") @router.delete("/{task_id}")
def delete_task(task_id: str) -> dict[str, Any]: def delete_task(
task_id: str,
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
task = _get_task(task_id) task = _get_task(task_id)
if not task: if not task:
raise fail(404, "task not found") raise fail(404, "task not found")

View File

@@ -26,6 +26,16 @@ def _load_sample(path: str | None, content: str | None = None, max_samples: int
if not text: if not text:
return [] return []
# 先按整文件 JSON数组/单对象)解析,兼容 .json失败再按 jsonl 逐行解析
try:
value = json.loads(text)
except json.JSONDecodeError:
value = None
if isinstance(value, list):
return [item for item in value[:max_samples] if isinstance(item, dict)]
if isinstance(value, dict):
return [value]
lines = text.splitlines()[:max_samples] lines = text.splitlines()[:max_samples]
records: list[dict[str, Any]] = [] records: list[dict[str, Any]] = []
for line in lines: for line in lines:

View File

@@ -0,0 +1,57 @@
"""data_convert 模块安全回归测试:输出文件名路径穿越与鉴权。
- ``output_filename`` 必须通过白名单校验,阻断 ``../``、``/``、``\\`` 及控制字符,
否则转换结果可被写出到存储根目录之外(任意文件读写/删除)。
- 所有 data_convert 路由必须挂载 ``get_current_user`` 鉴权依赖。
"""
from __future__ import annotations
import pytest
from fastapi import HTTPException
from app.core.auth import get_current_user
from app.modules.data_convert.router import _safe_output_filename, router
def test_safe_output_filename_defaults() -> None:
assert _safe_output_filename(None) == "converted-data.jsonl"
assert _safe_output_filename("") == "converted-data.jsonl"
def test_safe_output_filename_valid() -> None:
assert _safe_output_filename("converted-data.jsonl") == "converted-data.jsonl"
assert _safe_output_filename("my-data.v1.jsonl") == "my-data.v1.jsonl"
assert _safe_output_filename(" 报告.jsonl ") == "报告.jsonl"
@pytest.mark.parametrize(
"bad",
[
"../../etc/passwd",
"../x.jsonl",
"a/b.jsonl",
r"a\b.jsonl",
"a\\b.jsonl",
"..",
".",
"x\x00.jsonl",
"x\n.jsonl",
"x\t.jsonl",
],
)
def test_safe_output_filename_rejects_traversal(bad: str) -> None:
with pytest.raises(HTTPException):
_safe_output_filename(bad)
def test_all_data_convert_routes_require_auth() -> None:
for route in router.routes:
node = getattr(route, "dependant", None)
assert node is not None, f"route {route.path} has no dependency graph"
stack = list(node.dependencies)
calls: list = []
while stack:
dep = stack.pop()
stack.extend(getattr(dep, "dependencies", []))
calls.append(getattr(dep, "call", None))
assert get_current_user in calls, f"route {route.path} is missing get_current_user auth"

View File

@@ -0,0 +1,76 @@
"""Swagger / ReDoc / OpenAPI 文档路由安全开关测试。
生产环境APP_ENV=prod默认关闭 /docs、/redoc、/openapi.json
避免未授权访问泄露 API 结构;本地开发环境默认开放,可用 ENABLE_DOCS 覆盖。
"""
from __future__ import annotations
import pytest
from app.core.config import docs_kwargs, get_settings
@pytest.fixture(autouse=True)
def _reset_settings_cache():
"""每次测试前后清空 get_settings 的 lru_cache避免环境变量互相污染。"""
get_settings.cache_clear()
yield
get_settings.cache_clear()
def test_docs_kwargs_enabled() -> None:
assert docs_kwargs(True) == {}
def test_docs_kwargs_disabled() -> None:
assert docs_kwargs(False) == {"docs_url": None, "redoc_url": None, "openapi_url": None}
def test_docs_disabled_by_default_in_prod(monkeypatch) -> None:
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("APP_ENV", "prod")
assert get_settings().enable_docs is False
def test_docs_enabled_by_default_outside_prod(monkeypatch) -> None:
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("APP_ENV", "local")
assert get_settings().enable_docs is True
def test_docs_env_override_enables_in_prod(monkeypatch) -> None:
monkeypatch.setenv("ENABLE_DOCS", "true")
monkeypatch.setenv("APP_ENV", "prod")
assert get_settings().enable_docs is True
def test_docs_env_override_disables_outside_prod(monkeypatch) -> None:
monkeypatch.setenv("ENABLE_DOCS", "false")
monkeypatch.setenv("APP_ENV", "local")
assert get_settings().enable_docs is False
def test_create_app_disables_docs_in_prod(monkeypatch, tmp_path) -> None:
pytest.importorskip("fastapi")
from app.main import create_app
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("APP_ENV", "prod")
monkeypatch.setenv("LOG_DIR", str(tmp_path))
app = create_app()
assert app.docs_url is None
assert app.redoc_url is None
assert app.openapi_url is None
def test_create_app_enables_docs_outside_prod(monkeypatch, tmp_path) -> None:
pytest.importorskip("fastapi")
from app.main import create_app
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("APP_ENV", "local")
monkeypatch.setenv("LOG_DIR", str(tmp_path))
app = create_app()
assert app.docs_url == "/docs"
assert app.redoc_url == "/redoc"
assert app.openapi_url == "/openapi.json"

View File

@@ -15,12 +15,13 @@ from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFi
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from compute.agent.process_manager import ProcessManager from compute.agent.process_manager import ProcessManager
from compute.api.security import docs_kwargs
from compute.engines.llama_factory.adapter import build_command, parse_log_line, prepare_runtime_files from compute.engines.llama_factory.adapter import build_command, parse_log_line, prepare_runtime_files
from compute.engines.llama_factory.inference import get_inference_session from compute.engines.llama_factory.inference import get_inference_session
def create_app() -> FastAPI: def create_app() -> FastAPI:
app = FastAPI(title="YG Fine-Tune Compute API") app = FastAPI(title="YG Fine-Tune Compute API", **docs_kwargs())
jobs: dict[str, dict[str, Any]] = {} jobs: dict[str, dict[str, Any]] = {}
route_prefix = os.getenv("MODELTF_ROUTE_PREFIX", "/modelTF").rstrip("/") or "/modelTF" route_prefix = os.getenv("MODELTF_ROUTE_PREFIX", "/modelTF").rstrip("/") or "/modelTF"
process_manager = ProcessManager(os.getenv("TRAINING_LOG_ROOT", "/opt/yg-ft/logs/training")) process_manager = ProcessManager(os.getenv("TRAINING_LOG_ROOT", "/opt/yg-ft/logs/training"))
@@ -860,10 +861,17 @@ def create_app() -> FastAPI:
@app.get(f"{route_prefix}/compute/files/{{file_id}}/download") @app.get(f"{route_prefix}/compute/files/{{file_id}}/download")
async def download_file(file_id: str) -> FileResponse: async def download_file(file_id: str) -> FileResponse:
upload_root = Path(os.getenv("YG_FT_DATA_ROOT", "/data/yg-ft")) / "uploads" upload_root = Path(os.getenv("YG_FT_DATA_ROOT", "/data/yg-ft")) / "uploads"
# file_id 仅允许普通标识符,拒绝 ../、/、\ 等路径穿越字符。
if not file_id or not all(character.isalnum() or character in {"_", "-"} for character in file_id):
raise HTTPException(status_code=400, detail="invalid file id")
matches = list(upload_root.glob(f"{file_id}_*")) matches = list(upload_root.glob(f"{file_id}_*"))
if not matches: if not matches:
raise HTTPException(status_code=404, detail="file not found") raise HTTPException(status_code=404, detail="file not found")
return FileResponse(matches[0]) # 解析符号链接后仍必须位于 upload 根目录内,防止符号链接指向目录外文件。
resolved = matches[0].resolve()
if not _path_inside(upload_root, resolved):
raise HTTPException(status_code=404, detail="file not found")
return FileResponse(resolved)
return app return app

29
compute/api/security.py Normal file
View File

@@ -0,0 +1,29 @@
"""计算节点 API 安全配置Swagger / ReDoc / OpenAPI 文档路由开关。"""
from __future__ import annotations
import os
from typing import Any
def docs_enabled() -> bool:
"""判断 FastAPI 文档路由(/docs、/redoc、/openapi.json是否开放。
显式配置 ENABLE_DOCS 时以之为准;否则仅在关闭 token 鉴权
COMPUTE_AUTH_ENABLED=false本地开发时开放生产环境默认关闭
避免未授权访问泄露 API 结构。
"""
raw = os.getenv("ENABLE_DOCS", "").strip().lower()
if raw in {"true", "false"}:
return raw == "true"
auth_enabled = os.getenv("COMPUTE_AUTH_ENABLED", "true").lower() == "true"
return not auth_enabled
def docs_kwargs() -> dict[str, Any]:
"""返回传入 FastAPI 的文档路由参数。
关闭时 FastAPI 不注册 /docs、/redoc、/openapi.json访问一律返回 404。
"""
if docs_enabled():
return {}
return {"docs_url": None, "redoc_url": None, "openapi_url": None}

View File

@@ -15,27 +15,58 @@ class LlamaFactoryCommand:
def _load_dataset_preview(path: Path) -> list[dict[str, Any]]: def _load_dataset_preview(path: Path) -> list[dict[str, Any]]:
"""Load a preview of JSON/JSONL records from a dataset file.
Content-sniffs instead of trusting the extension so that BOM-prefixed files,
JSONL files containing a single JSON array, and mislabeled extensions all work.
"""
if not path.exists(): if not path.exists():
return [] return []
text = path.read_text(encoding="utf-8", errors="replace").strip() text = path.read_text(encoding="utf-8-sig", errors="replace").strip()
if not text: if not text:
return [] return []
if path.suffix.lower() == ".jsonl": try:
items: list[dict[str, Any]] = [] value = json.loads(text)
for line in text.splitlines()[:20]: except json.JSONDecodeError:
line = line.strip() value = None
if not line:
continue
value = json.loads(line)
if isinstance(value, dict):
items.append(value)
return items
value = json.loads(text)
if isinstance(value, list): if isinstance(value, list):
return [item for item in value[:20] if isinstance(item, dict)] return [item for item in value[:20] if isinstance(item, dict)]
if isinstance(value, dict): if isinstance(value, dict):
return [value] return [value]
return [] items: list[dict[str, Any]] = []
for line in text.splitlines()[:20]:
line = line.strip()
if not line:
continue
try:
parsed = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(parsed, list):
items.extend(item for item in parsed[:20] if isinstance(item, dict))
elif isinstance(parsed, dict):
items.append(parsed)
if len(items) >= 20:
break
return items[:20]
def _required_columns_for(formatting: str, columns: dict[str, Any]) -> list[str]:
"""Required data columns per dataset format.
Mirrors LLaMA-Factory's leniency: optional columns (e.g. ``input`` / ``query``
in Alpaca) are never required, only fields the format structurally needs.
"""
fmt = str(formatting or "").lower()
if fmt == "sharegpt":
return [str(columns.get("messages") or "messages")]
if fmt in {"dpo", "rm", "kto", "ppo"}:
return [str(columns[key]) for key in ("chosen", "rejected") if columns.get(key)]
if fmt in {"cpt", "pt", "pretrain"}:
return [str(columns.get("prompt") or columns.get("text") or "text")]
# alpaca family: prompt (instruction) + response (output) required,
# query (input) / history are optional and common to omit in jsonl datasets.
return [str(columns[key]) for key in ("prompt", "response") if columns.get(key)]
def _validate_dataset_columns(config: dict[str, Any]) -> list[str]: def _validate_dataset_columns(config: dict[str, Any]) -> list[str]:
@@ -51,7 +82,7 @@ def _validate_dataset_columns(config: dict[str, Any]) -> list[str]:
file_name = item.get("file_name") file_name = item.get("file_name")
file_names = file_name if isinstance(file_name, list) else [file_name] file_names = file_name if isinstance(file_name, list) else [file_name]
columns = item.get("columns") if isinstance(item.get("columns"), dict) else {} columns = item.get("columns") if isinstance(item.get("columns"), dict) else {}
required_columns = [str(value) for value in columns.values() if value] required_columns = _required_columns_for(str(item.get("formatting") or ""), columns)
for name in file_names: for name in file_names:
if not name: if not name:
continue continue

View File

@@ -22,7 +22,10 @@ from typing import Any
def _load_dataset(path: str) -> list[dict[str, Any]]: def _load_dataset(path: str) -> list[dict[str, Any]]:
"""Load a JSON or JSONL dataset file. """Load a JSON or JSONL dataset file (jsonl-compatible).
Content-sniffs instead of trusting the extension so jsonl files with a BOM,
a single JSON array on one line, or mislabeled extensions all load correctly.
Supports common field names used across the platform: Supports common field names used across the platform:
* ``instruction`` + ``input`` + ``output`` (Alpaca-style) * ``instruction`` + ``input`` + ``output`` (Alpaca-style)
@@ -30,14 +33,17 @@ def _load_dataset(path: str) -> list[dict[str, Any]]:
* ``messages`` (ShareGPT-style the last assistant message is treated as reference) * ``messages`` (ShareGPT-style the last assistant message is treated as reference)
""" """
file_path = Path(path) file_path = Path(path)
text = file_path.read_text(encoding="utf-8", errors="replace").strip() text = file_path.read_text(encoding="utf-8-sig", errors="replace").strip()
if not text: if not text:
return [] return []
if file_path.suffix.lower() == ".json": try:
value = json.loads(text) value = json.loads(text)
if isinstance(value, list): except json.JSONDecodeError:
return [item for item in value if isinstance(item, dict)] value = None
return [value] if isinstance(value, dict) else [] if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
if isinstance(value, dict):
return [value]
samples: list[dict[str, Any]] = [] samples: list[dict[str, Any]] = []
for line in text.splitlines(): for line in text.splitlines():

View File

@@ -0,0 +1,42 @@
from __future__ import annotations
import json
from compute.engines.llama_factory.eval_runner import _load_dataset
def _write(tmp_path, name: str, text: str) -> str:
path = tmp_path / name
path.write_text(text, encoding="utf-8")
return str(path)
def test_load_jsonl_multiline(tmp_path) -> None:
path = _write(
tmp_path,
"eval.jsonl",
'{"question": "q1", "answer": "a1"}\n{"question": "q2", "answer": "a2"}\n',
)
assert _load_dataset(path) == [
{"question": "q1", "answer": "a1"},
{"question": "q2", "answer": "a2"},
]
def test_load_json_array(tmp_path) -> None:
path = _write(
tmp_path,
"eval.json",
json.dumps([{"question": "x", "answer": "y"}]),
)
assert _load_dataset(path) == [{"question": "x", "answer": "y"}]
def test_load_jsonl_with_bom_and_embedded_array(tmp_path) -> None:
"""jsonl 带 BOM 且单行内嵌 JSON 数组,都应正常加载。"""
path = _write(
tmp_path,
"eval.jsonl",
"" + json.dumps([{"question": "a", "answer": "b"}, {"question": "c", "answer": "d"}]),
)
assert len(_load_dataset(path)) == 2

View File

@@ -0,0 +1,63 @@
"""compute ``download_file`` 端点安全回归测试。
修复前 ``file_id`` 直接拼进 glob 模式且不校验路径包含关系,可通过 ``../``
穿越出 upload 目录,并在 Linux 上跟随符号链接读取任意文件。
修复后file_id 仅允许字母/数字/下划线/连字符,返回前对解析后的路径
做 upload 根目录包含性校验。
"""
from __future__ import annotations
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
def _make_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient:
monkeypatch.setenv("TRAINING_LOG_ROOT", str(tmp_path / "logs"))
monkeypatch.setenv("YG_FT_DATA_ROOT", str(tmp_path / "data"))
monkeypatch.setenv("COMPUTE_EXECUTION_MODE", "simulator")
monkeypatch.setenv("COMPUTE_AUTH_ENABLED", "false")
monkeypatch.delenv("ENABLE_DOCS", raising=False)
from compute.api.main import create_app
return TestClient(create_app())
def _upload_root(tmp_path: Path) -> Path:
root = tmp_path / "data" / "uploads"
root.mkdir(parents=True, exist_ok=True)
return root
def test_download_legit_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client(tmp_path, monkeypatch)
(_upload_root(tmp_path) / "file_123456_hello.txt").write_text("HELLO-DOWNLOAD", encoding="utf-8")
response = client.get("/modelTF/compute/files/file_123456/download")
assert response.status_code == 200
assert response.content == b"HELLO-DOWNLOAD"
def test_download_rejects_traversal_file_id(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client(tmp_path, monkeypatch)
outside = tmp_path / "secret" / "passwd_1.txt"
outside.parent.mkdir(parents=True, exist_ok=True)
outside.write_text("TOP-SECRET", encoding="utf-8")
for file_id in ["..", "file.123", "..%2F..%2Fsecret%2Fpasswd", "file%20name"]:
response = client.get(f"/modelTF/compute/files/{file_id}/download")
assert response.status_code in (400, 404), f"file_id={file_id!r} -> {response.status_code}"
assert b"TOP-SECRET" not in response.content
def test_download_blocks_symlink_escape(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
client = _make_client(tmp_path, monkeypatch)
upload_root = _upload_root(tmp_path)
outside = tmp_path / "secret.txt"
outside.write_text("TOP-SECRET", encoding="utf-8")
try:
(upload_root / "file_999999_link.txt").symlink_to(outside)
except OSError:
pytest.skip("symlink creation not permitted on this platform")
response = client.get("/modelTF/compute/files/file_999999/download")
assert response.status_code == 404
assert b"TOP-SECRET" not in response.content

View File

@@ -1,6 +1,8 @@
from __future__ import annotations from __future__ import annotations
from compute.engines.llama_factory.adapter import build_command import json
from compute.engines.llama_factory.adapter import _validate_dataset_columns, build_command
def test_build_command_uses_explicit_validation_dataset_without_resplitting() -> None: def test_build_command_uses_explicit_validation_dataset_without_resplitting() -> None:
@@ -21,3 +23,70 @@ def test_build_command_uses_explicit_validation_dataset_without_resplitting() ->
) )
assert "--do_eval" in result.command assert "--do_eval" in result.command
assert "--val_size" not in result.command assert "--val_size" not in result.command
def _write(tmp_path, name: str, lines: list[dict]) -> object:
path = tmp_path / name
path.write_text(
"".join(json.dumps(line, ensure_ascii=False) + "\n" for line in lines),
encoding="utf-8",
)
return path
def test_jsonl_alpaca_without_input_column_passes_validation(tmp_path) -> None:
"""纯 jsonl Alpaca 数据缺省 input 字段(常见),不应被校验拦截。"""
_write(tmp_path, "train.jsonl", [{"instruction": "hi", "output": "hello"}])
errors = _validate_dataset_columns(
{
"dataset_dir": str(tmp_path),
"dataset_info": {
"ygft_a": {
"file_name": "train.jsonl",
"formatting": "alpaca",
"columns": {"prompt": "instruction", "query": "input", "response": "output"},
}
},
}
)
assert errors == []
def test_jsonl_sharegpt_passes_validation(tmp_path) -> None:
"""ShareGPT 格式 jsonlmessages应通过校验。"""
_write(
tmp_path,
"msg.jsonl",
[{"messages": [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}]}],
)
errors = _validate_dataset_columns(
{
"dataset_dir": str(tmp_path),
"dataset_info": {
"ygft_m": {
"file_name": "msg.jsonl",
"formatting": "sharegpt",
"columns": {"messages": "messages"},
}
},
}
)
assert errors == []
def test_jsonl_missing_response_still_rejected(tmp_path) -> None:
"""缺 outputresponse仍应报错——没有答案无法做有监督微调。"""
_write(tmp_path, "train.jsonl", [{"instruction": "hi"}])
errors = _validate_dataset_columns(
{
"dataset_dir": str(tmp_path),
"dataset_info": {
"ygft_a": {
"file_name": "train.jsonl",
"formatting": "alpaca",
"columns": {"prompt": "instruction", "query": "input", "response": "output"},
}
},
}
)
assert errors and "output" in errors[0]

View File

@@ -0,0 +1,40 @@
"""计算节点文档路由(/docs、/redoc、/openapi.json安全开关测试。
生产默认COMPUTE_AUTH_ENABLED=true关闭文档路由避免未授权泄露 API 结构;
显式配置 ENABLE_DOCS 可覆盖默认行为。
"""
from __future__ import annotations
from compute.api.security import docs_enabled, docs_kwargs
def test_docs_disabled_when_auth_enabled(monkeypatch) -> None:
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("COMPUTE_AUTH_ENABLED", "true")
assert docs_enabled() is False
assert docs_kwargs() == {"docs_url": None, "redoc_url": None, "openapi_url": None}
def test_docs_enabled_when_auth_disabled(monkeypatch) -> None:
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.setenv("COMPUTE_AUTH_ENABLED", "false")
assert docs_enabled() is True
assert docs_kwargs() == {}
def test_docs_env_override_enables_with_auth(monkeypatch) -> None:
monkeypatch.setenv("ENABLE_DOCS", "true")
monkeypatch.setenv("COMPUTE_AUTH_ENABLED", "true")
assert docs_enabled() is True
def test_docs_env_override_disables_without_auth(monkeypatch) -> None:
monkeypatch.setenv("ENABLE_DOCS", "false")
monkeypatch.setenv("COMPUTE_AUTH_ENABLED", "false")
assert docs_enabled() is False
def test_docs_default_when_auth_env_missing(monkeypatch) -> None:
monkeypatch.delenv("ENABLE_DOCS", raising=False)
monkeypatch.delenv("COMPUTE_AUTH_ENABLED", raising=False)
assert docs_enabled() is False

View File

@@ -1,6 +1,8 @@
APP_ENV=prod APP_ENV=prod
APP_NAME=YG Fine-Tune Platform API APP_NAME=YG Fine-Tune Platform API
MODELTF_ROUTE_PREFIX=/modelTF MODELTF_ROUTE_PREFIX=/modelTF
# 禁止未授权访问 Swagger/ReDoc/OpenAPI 文档;调试时可临时改为 true
ENABLE_DOCS=false
CORS_ALLOW_ORIGINS=http://localhost:16801,http://127.0.0.1:16801 CORS_ALLOW_ORIGINS=http://localhost:16801,http://127.0.0.1:16801
FRONTEND_IMAGE=yg-ft-frontend-runtime:latest FRONTEND_IMAGE=yg-ft-frontend-runtime:latest
@@ -17,7 +19,9 @@ POSTGRES_USER=root
POSTGRES_PASSWORD=8811614287327Leo POSTGRES_PASSWORD=8811614287327Leo
DATABASE_URL=postgresql+psycopg://root:8811614287327Leo@www.caoxiaozhu.com:5432/yg_ft DATABASE_URL=postgresql+psycopg://root:8811614287327Leo@www.caoxiaozhu.com:5432/yg_ft
REDIS_URL=redis://redis:6379/0 # Redis 访问鉴权requirepass 密码;REDIS_URL 已内嵌密码(redis://:<密码>@redis:6379/0
REDIS_PASSWORD=Tvhrf659WaX-S1B8FG6c2kSZK07XTv82
REDIS_URL=redis://:Tvhrf659WaX-S1B8FG6c2kSZK07XTv82@redis:6379/0
# PostgreSQL uses the shared external database. The local postgres service is disabled in docker-compose.yml. # PostgreSQL uses the shared external database. The local postgres service is disabled in docker-compose.yml.
# Redis still uses the built-in service during current development. # Redis still uses the built-in service during current development.

View File

@@ -42,9 +42,10 @@ services:
APP_ENV: ${APP_ENV:-prod} APP_ENV: ${APP_ENV:-prod}
APP_NAME: ${APP_NAME:-YG Fine-Tune Platform API} APP_NAME: ${APP_NAME:-YG Fine-Tune Platform API}
MODELTF_ROUTE_PREFIX: ${MODELTF_ROUTE_PREFIX:-/modelTF} MODELTF_ROUTE_PREFIX: ${MODELTF_ROUTE_PREFIX:-/modelTF}
ENABLE_DOCS: ${ENABLE_DOCS:-false}
CORS_ALLOW_ORIGINS: ${CORS_ALLOW_ORIGINS:-http://localhost:16801,http://127.0.0.1:16801} CORS_ALLOW_ORIGINS: ${CORS_ALLOW_ORIGINS:-http://localhost:16801,http://127.0.0.1:16801}
DATABASE_URL: ${DATABASE_URL:-postgresql+psycopg://root:8811614287327Leo@www.caoxiaozhu.com:5432/yg_ft} DATABASE_URL: ${DATABASE_URL:-postgresql+psycopg://root:8811614287327Leo@www.caoxiaozhu.com:5432/yg_ft}
REDIS_URL: ${REDIS_URL:-redis://redis:6379/0} REDIS_URL: ${REDIS_URL:-redis://:${REDIS_PASSWORD:-change_me}@redis:6379/0}
USE_BUILTIN_POSTGRES: ${USE_BUILTIN_POSTGRES:-false} USE_BUILTIN_POSTGRES: ${USE_BUILTIN_POSTGRES:-false}
USE_BUILTIN_REDIS: ${USE_BUILTIN_REDIS:-true} USE_BUILTIN_REDIS: ${USE_BUILTIN_REDIS:-true}
LOG_LEVEL: ${LOG_LEVEL:-INFO} LOG_LEVEL: ${LOG_LEVEL:-INFO}
@@ -103,7 +104,10 @@ services:
redis: redis:
image: redis:7-alpine image: redis:7-alpine
container_name: yg-ft-redis container_name: yg-ft-redis
command: ["redis-server", "--appendonly", "yes"] command: ["redis-server", "--appendonly", "yes", "--requirepass", "${REDIS_PASSWORD:-change_me}"]
environment:
# redis-cli 健康检查免命令行传密码(避免 -a 泄露进程参数)
REDISCLI_AUTH: ${REDIS_PASSWORD:-change_me}
volumes: volumes:
- redis_data:/data - redis_data:/data
ports: ports:

View File

@@ -2,6 +2,8 @@ COMPUTE_ENV=prod
COMPUTE_HOST_ID=gpu-node-01 COMPUTE_HOST_ID=gpu-node-01
COMPUTE_EXECUTION_MODE=real COMPUTE_EXECUTION_MODE=real
MODELTF_ROUTE_PREFIX=/modelTF MODELTF_ROUTE_PREFIX=/modelTF
# 禁止未授权访问 Swagger/ReDoc/OpenAPI 文档;调试时可临时改为 true
ENABLE_DOCS=false
# Five-digit host ports exposed outside the compute server. # Five-digit host ports exposed outside the compute server.
COMPUTE_API_PORT=19100 COMPUTE_API_PORT=19100
FILE_GATEWAY_PORT=19101 FILE_GATEWAY_PORT=19101

View File

@@ -2,6 +2,8 @@ COMPUTE_ENV=prod
COMPUTE_HOST_ID=gpu-node-01 COMPUTE_HOST_ID=gpu-node-01
COMPUTE_EXECUTION_MODE=real COMPUTE_EXECUTION_MODE=real
MODELTF_ROUTE_PREFIX=/modelTF MODELTF_ROUTE_PREFIX=/modelTF
# 禁止未授权访问 Swagger/ReDoc/OpenAPI 文档;调试时可临时改为 true
ENABLE_DOCS=false
# Five-digit host ports exposed outside the compute server. # Five-digit host ports exposed outside the compute server.
COMPUTE_API_PORT=19100 COMPUTE_API_PORT=19100
FILE_GATEWAY_PORT=19101 FILE_GATEWAY_PORT=19101

View File

@@ -11,6 +11,7 @@ services:
COMPUTE_HOST_ID: ${COMPUTE_HOST_ID:-gpu-node-01} COMPUTE_HOST_ID: ${COMPUTE_HOST_ID:-gpu-node-01}
COMPUTE_EXECUTION_MODE: ${COMPUTE_EXECUTION_MODE:-real} COMPUTE_EXECUTION_MODE: ${COMPUTE_EXECUTION_MODE:-real}
MODELTF_ROUTE_PREFIX: ${MODELTF_ROUTE_PREFIX:-/modelTF} MODELTF_ROUTE_PREFIX: ${MODELTF_ROUTE_PREFIX:-/modelTF}
ENABLE_DOCS: ${ENABLE_DOCS:-false}
COMPUTE_AUTH_ENABLED: ${COMPUTE_AUTH_ENABLED:-true} COMPUTE_AUTH_ENABLED: ${COMPUTE_AUTH_ENABLED:-true}
COMPUTE_SERVICE_TOKEN: ${COMPUTE_SERVICE_TOKEN:-change_me} COMPUTE_SERVICE_TOKEN: ${COMPUTE_SERVICE_TOKEN:-change_me}
ENABLE_APP_CALLBACK: ${ENABLE_APP_CALLBACK:-false} ENABLE_APP_CALLBACK: ${ENABLE_APP_CALLBACK:-false}

164
docs/database-config.md Normal file
View File

@@ -0,0 +1,164 @@
# 数据库配置与初始化说明PostgreSQL / Redis
> 记录平台的 **PostgreSQL 账号密码**、**Redis 账号密码**、**数据库地址在代码中的配置位置**
> 以及**切换 PG 数据集时如何执行完整初始化 SQL**。
---
## 1. 账号密码速查表
### 1.1 PostgreSQL
| 环境 | 地址 | 用户 | 密码 | 数据库 | 来源 |
|------|------|------|------|--------|------|
| 代码默认值 | `localhost:15432` | `yg_ft` | `change_me` | `yg_ft` | `config.py` / `session.py``DATABASE_URL` 兜底 |
| Docker 部署 | `www.caoxiaozhu.com:5432` | `root` | `8811614287327Leo` | `yg_ft` | `docker/app/.env``DATABASE_URL` |
| Docker 内置 Postgres已注释 | `localhost:15432` | `root` | `8811614287327Leo` | `yg_ft` | `docker/app/docker-compose.yml` 注释掉的 postgres 服务 |
> ⚠️ `change_me` 与 `8811614287327Leo` 均为默认/示例凭据,生产环境务必更换。
### 1.2 Redis
| 项 | 值 | 说明 |
|----|-----|------|
| 连接串 | `redis://:<REDIS_PASSWORD>@redis:6379/0` | 已内嵌密码docker 网络内服务名 `redis`,端口 6379db 0 |
| 对外端口 | `16379``REDIS_PORT` | 宿主机映射 |
| 密码 | `docker/app/.env``REDIS_PASSWORD` | 已启用 `requirepass` 鉴权 |
| 镜像 | `redis:7-alpine` | 已开启 AOF`--appendonly yes`+ `requirepass` |
> **当前后端代码未使用 Redis**`redis` 包已列入 `requirements.txt``REDIS_URL` 通过
> docker-compose 注入容器,但全仓库 `backend/`、`compute/` 没有任何 `import redis` /
> `Redis(...)` 连接代码。Redis 为后续功能预留;**即便如此仍已配置鉴权**
> 避免无密码实例对外暴露(纵深防御)。未来启用时按 `REDIS_URL` 连接即可。
>
> 健康检查通过容器环境变量 `REDISCLI_AUTH` 认证,不在进程参数中泄露密码。
---
## 2. 数据库地址在代码中的配置位置
### 2.1 后端backend
| 文件 | 作用 | 取值 |
|------|------|------|
| `backend/app/core/config.py` | **唯一权威配置**`Settings.database_url` | `os.getenv("DATABASE_URL", "postgresql+psycopg://yg_ft:change_me@localhost:15432/yg_ft")` |
| `backend/app/db/session.py` | SQLAlchemy 引擎(`get_db` / `session_scope` | `os.getenv("DATABASE_URL", ...)` 同样兜底 |
| `backend/app/db/platform_store.py` | **平台主存储**,直接用 psycopg 连接池 | 取 `settings.database_url``_psycopg_url()``postgresql+psycopg://` 转成 `postgresql://` |
| `backend/app/modules/data_process/store.py` | 数据处理存储 | `get_settings().database_url` |
> **环境变量加载顺序**`config.py` 导入时会 `load_dotenv(backend/.env, override=True)`
> 即 **`backend/.env` 会覆盖系统环境变量**docker 部署则直接由 compose 注入 `DATABASE_URL`。
> 最终优先级:`backend/.env` / compose 注入的环境变量 > 代码内默认值。
### 2.2 部署配置docker
| 文件 | 关键项 |
|------|--------|
| `docker/app/.env` | `DATABASE_URL``POSTGRES_USER``POSTGRES_PASSWORD``REDIS_URL``REDIS_PORT` |
| `docker/app/docker-compose.yml` | `backend-api` 环境透传上述变量;`redis` 服务定义 |
```ini
# docker/app/.env节选
DATABASE_URL=postgresql+psycopg://root:8811614287327Leo@www.caoxiaozhu.com:5432/yg_ft
POSTGRES_USER=root
POSTGRES_PASSWORD=8811614287327Leo
REDIS_PASSWORD=<强密码> # Redis requirepass新增鉴权
REDIS_URL=redis://:<REDIS_PASSWORD>@redis:6379/0 # 连接串内嵌密码
REDIS_PORT=16379
USE_BUILTIN_POSTGRES=false # 当前用共享外部库,内置 postgres 服务被注释
USE_BUILTIN_REDIS=true # Redis 用内置服务
```
---
## 3. 完整初始化 SQL
### 3.1 脚本位置
| 脚本 | 用途 |
|------|------|
| **`backend/app/db/sql/000_full_init.sql`** | **一键初始化脚本(新增)**:建库表 + 索引 + 种子数据,幂等,覆盖全部 39 张运行表 |
| `backend/app/db/sql/001_platform_runtime.sql` | 平台核心表(应用启动自动执行) |
| `backend/app/db/sql/002_governance.sql` | 治理表(应用启动自动执行) |
| `backend/app/db/sql/003_tenant_quota.sql` | 租户配额列(应用启动自动执行) |
| `backend/app/db/sql/003_model_path_governance.sql` | 模型可训练列(**应用不自动执行**,已并入完整脚本) |
| `backend/app/db/sql/002_data_process.sql` | 数据处理表(**应用不自动执行**,已并入完整脚本) |
| `docs/postgres-schema.sql` | ⚠️ **目标设计稿**UUID/JSONB与运行时代码不兼容**不要用于初始化** |
> **重要**`docs/postgres-schema.sql` 是规划中的“目标 schema”UUID 主键、`ft_platform` schema 等),
> 运行时代码明确拒绝该结构(`002_data_process.sql` 检测到 `datasets.id` 非 TEXT 会直接报错)。
> 初始化请使用 **`000_full_init.sql`**。
### 3.2 执行步骤(全新 PG 环境)
**第 1 步:创建角色与数据库**(必须单独执行,不能放进事务)
```sql
-- 以超级用户(如 postgres连接
CREATE ROLE yg_ft LOGIN PASSWORD '请改为强密码';
CREATE DATABASE yg_ft OWNER yg_ft;
-- 如需应用执行 CREATE EXTENSION 等,可再授予超级用户(按需):
-- ALTER ROLE yg_ft SUPERUSER;
```
**第 2 步:执行完整初始化脚本**
```bash
psql "postgresql://yg_ft:密码@<host>:5432/yg_ft" \
-f backend/app/db/sql/000_full_init.sql
```
脚本特点:
- 全程一个事务(`BEGIN; ... COMMIT;`),失败自动回滚
- 所有 DDL 使用 `IF NOT EXISTS` / `ADD COLUMN IF NOT EXISTS`**可重复执行**
-`DO $$...$$` 语句块,必须用 `psql` 执行(应用内部的按分号切分 `executescript()` 不适用)
- 自动写入种子用户:`admin / admin123``operator / operator123`(登录后请改密)
**第 3 步:校验**
```sql
SELECT count(*) FROM pg_tables WHERE schemaname = 'public'; -- 应 ≥ 39
SELECT username, role, status FROM users; -- 应有 admin / operator
```
### 3.3 执行方式对比(三种途径)
| 方式 | 覆盖范围 | 命令 |
|------|----------|------|
| **A. 完整脚本(推荐,切换新库)** | 全部 39 表 + 索引 + 种子 | `psql ... -f 000_full_init.sql` |
| B. 应用自动初始化 | 001 + 002_governance + 003_tenant_quota + 种子用户;**不含**数据处理表、`models.can_train``data_convert_tasks` | 应用首次调用 `get_platform_store()``ensure_schema()` 自动执行 |
| C. 数据处理表单独安装 | `002_data_process.sql` 全部内容 | 在 `backend/` 目录下:`python -m app.modules.data_process.schema_cli --apply --yes`(或 `--check` 只读检查) |
> **缺口说明**
> - `models.can_train`(训练预检用)只在 `003_model_path_governance.sql` 中创建,应用启动**不会**自动执行;
> - `data_convert_tasks`(数据转换任务表)运行时代码引用但**原 SQL 脚本缺失**
> 已统一并入 `000_full_init.sql` 补齐。若现有库缺这两项,执行一次完整脚本即可幂等补上。
### 3.4 完整脚本包含的表39 张)
**核心**users、models、trained_models、model_lineage、model_artifacts、model_export_jobs、
datasets、dataset_files、compute_nodes、gpus、fine_tune_tasks、fine_tune_metrics、
fine_tune_checkpoints、compute_jobs、gpu_allocations、scheduler_locks、resource_replicas、
resource_sync_jobs、eval_tasks、eval_dimensions、compare_tasks、projects、project_members、
roles、sessions、acls
**治理**tenants、approval_templates、approval_instances、approval_steps、audit_logs、retention_policies
**数据处理**data_process_tasks、data_process_source_files、data_process_preview_items、
data_process_results、dataset_file_versions、dataset_records
**数据转换**data_convert_tasks新增补齐
---
## 4. 安全注意事项
1. **更换默认密码**`change_me`(代码兜底)、`8811614287327Leo`(部署)、`admin123`/`operator123`(种子用户)、`REDIS_PASSWORD` 上线前必须更换。
2. **Redis 已加鉴权**:已配置 `requirepass` + `REDISCLI_AUTH` 健康检查;`REDIS_URL` 内嵌密码。若端口需暴露公网,仍建议用防火墙/安全组限制来源。
3. **`docker/*/.env` 已入库,含明文凭据**
- `backend/.env` 已被 `.gitignore` 排除;
-`docker/app/.env``docker/compute/.env` 目前被 git 跟踪(`git ls-files` 可见),
其中的 `DATABASE_URL``POSTGRES_PASSWORD``COMPUTE_SERVICE_TOKEN` 等均为明文。
- **建议**:轮换这些凭据,将 `docker/*/.env` 移出版本库(`git rm --cached`)并改用
部署侧机密注入(如 docker secrets / CI 变量 / 环境变量模板),保留 `.env.example` 作为模板。
4. **最小权限**:应用角色只需对业务库的 DML/DDL 权限,尽量避免 SUPERUSER。

290
docs/security-hardening.md Normal file
View File

@@ -0,0 +1,290 @@
# 安全加固总结(前端 / 后端 / 算力节点)
> 记录 2026-08-06 对本平台的漏洞修复。核心目标:修复 **FastAPI 文档接口未授权访问**、
> **Swagger 泄露 API 结构**、以及两类**任意文件读取**漏洞(路径穿越 + 符号链接跟随),
> 修复过程不改变正常业务流程。
>
> 其中 **FastAPI 文档开关(`ENABLE_DOCS`** 的详细用法见
> [§4 FastAPI 文档开关使用说明](#4-fastapi-文档开关使用说明enabledocs)。
---
## 1. 漏洞总览
| # | 影响面 | 漏洞 | 风险等级 | 修复 |
|---|--------|------|----------|------|
| 1 | 后端 + 算力节点 | FastAPI 默认暴露 `/docs``/redoc``/openapi.json`**未授权**泄露全部 API 结构、参数、内部路由 | 中 | 生产环境关闭文档路由,访问返回 404`ENABLE_DOCS` 可覆盖) |
| 2 | 后端 | `data-convert` 模块 `output_filename` **路径穿越**:可任意文件读 / 写 / 删,且整个模块**无鉴权** | **严重** | 输出文件名白名单校验 + 全部端点补鉴权 |
| 3 | 算力节点 | `compute/files/{file_id}/download``file_id` 直接拼进 glob 模式可 `../` **穿越出上传目录**`FileResponse` 在 Linux 上**跟随符号链接**读取任意文件 | 中高 | `file_id` 字符白名单 + 解析后路径包含性二次校验 |
| 4 | 前端Vue + nginx | 无文件服务代码nginx 仅服务受控静态目录,无 `alias` | 无 | 审计确认,无需修复 |
---
## 2. 前端Vue 3 + nginx
**审计结论:不构成 ComfyUI `follow_symlinks` 类文件读取漏洞。**
- nginx`docker/nginx.conf.template`)只服务受控的 `dist/` 静态目录,使用 `try_files`
`alias` 指令、无用户可控文件路径,不存在路径穿越面。
- Vue SPA 自身没有任何文件服务逻辑;文件下载全部走后端/算力节点 API。
- 前端 axios 拦截器(`frontend/src/api/request.ts`)对**每个请求**自动附加
`Authorization: Bearer platform-token-{user_id}`,因此给后端接口补鉴权不会影响页面功能。
---
## 3. 后端FastAPI
### 3.1 data_convert 输出文件名路径穿越(严重)
**问题**`backend/app/modules/data_convert/router.py`
- `output_filename` 由请求体传入后**原样入库**,随后拼进
`output_dir / output_filename` 用于写/读/删:
- `if output_path.exists(): output_path.unlink()` → 任意文件删除
- `open(output_path, "a")` → 任意文件追加写
- `download_result``FileResponse(output_path)` → 任意文件读取
- 整个 router **无任何鉴权依赖**(后端无全局鉴权中间件),任意网络访问者可利用。
**修复**
```python
def _safe_output_filename(value: Any) -> str:
"""输出文件名白名单:拒绝 ../、/、\ 及控制字符,仅允许普通文件名。"""
name = str(value or "converted-data.jsonl").strip()
if (
not name
or name in {".", ".."}
or name != Path(name).name
or "/" in name
or "\\" in name
or any(ord(c) < 32 or ord(c) == 127 for c in name)
):
raise fail(400, "output filename must be a plain file name")
return name
def _task_output_path(task: dict[str, Any]) -> Path:
"""统一构造转换输出路径,始终位于任务 output 目录内。"""
return _output_dir(task["id"]) / _safe_output_filename(task.get("output_filename"))
```
- `create_task` 创建时即校验(恶意值直接 400
- 全部 4 处使用点(`upload_source_files` 自动转换、`run_convert``download_result`
`import_as_dataset`)统一改用 `_task_output_path()`,历史任务同样受保护
- 全部 **8 个** `/data-convert` 端点补充 `current_user: dict = Depends(get_current_user)` 鉴权
**功能影响**:正常转换流程(前端 `outputName + '.jsonl'` 这类纯文件名)不受影响;
接口现在要求登录态,未登录调用返回 401。
---
## 4. FastAPI 文档开关使用说明(`ENABLE_DOCS`
### 4.1 为什么需要这个开关
FastAPI 默认注册 3 个**无需鉴权**的路由,直接泄露全部 API 结构:
| 路由 | 说明 |
|------|------|
| `/docs` | Swagger UI 交互文档 |
| `/redoc` | ReDoc 文档 |
| `/openapi.json` | OpenAPI Schema含全部接口、参数、模型定义 |
修复方式是:**关闭时让 FastAPI 不注册这 3 个路由**,访问一律返回 404而不是返回空页面。
### 4.2 核心实现
关闭的本质是向 `FastAPI(...)` 传入三个 `None` 参数:
```python
# docs 关闭时等价于:
FastAPI(
title=...,
docs_url=None, # /docs → 404
redoc_url=None, # /redoc → 404
openapi_url=None, # /openapi.json → 404
)
```
### 4.3 后端开关逻辑(`backend/app/core/config.py` + `backend/app/main.py`
```python
# config.py —— Settings.enable_docs 在 __post_init__ 中计算
object.__setattr__(
self,
"enable_docs",
_bool_env("ENABLE_DOCS", os.getenv("APP_ENV", "local") != "prod"),
)
# config.py —— 返回传给 FastAPI 的文档参数
def docs_kwargs(enabled: bool) -> dict[str, Any]:
if enabled:
return {}
return {"docs_url": None, "redoc_url": None, "openapi_url": None}
# main.py —— 接入
app = FastAPI(title=settings.app_name, **docs_kwargs(settings.enable_docs))
```
**判定顺序(优先级从高到低)**
1. 显式设置 `ENABLE_DOCS=true/false` → 以显式值为准
2. 未设置 → `APP_ENV != "prod"` 时开放,`APP_ENV=prod` 时**关闭**
> 注意:`enable_docs` 从**运行时环境**读取 `APP_ENV`(而非类定义时缓存的默认值),
> 确保生产环境默认关闭始终生效且便于测试。
### 4.4 算力节点开关逻辑(`compute/api/security.py` + `compute/api/main.py`
```python
# security.py
def docs_enabled() -> bool:
raw = os.getenv("ENABLE_DOCS", "").strip().lower()
if raw in {"true", "false"}:
return raw == "true"
auth_enabled = os.getenv("COMPUTE_AUTH_ENABLED", "true").lower() == "true"
return not auth_enabled # 开启 token 鉴权(生产)时默认关闭文档
def docs_kwargs() -> dict[str, Any]:
if docs_enabled():
return {}
return {"docs_url": None, "redoc_url": None, "openapi_url": None}
# main.py
app = FastAPI(title="YG Fine-Tune Compute API", **docs_kwargs())
```
**判定顺序(优先级从高到低)**
1. 显式设置 `ENABLE_DOCS=true/false` → 以显式值为准
2. 未设置 → `COMPUTE_AUTH_ENABLED=true`(生产默认)时**关闭**
`COMPUTE_AUTH_ENABLED=false`(本地开发)时开放
### 4.5 环境变量速查表
| 服务 | 环境变量 | 取值 | 默认行为 |
|------|----------|------|----------|
| 后端 | `ENABLE_DOCS` | `true` / `false` | 未设置时按 `APP_ENV != "prod"` 判定 |
| 后端 | `APP_ENV` | `local` / `prod` 等 | `prod` 时关闭文档 |
| 算力节点 | `ENABLE_DOCS` | `true` / `false` | 未设置时按 `COMPUTE_AUTH_ENABLED` 判定 |
| 算力节点 | `COMPUTE_AUTH_ENABLED` | `true` / `false` | `true` 时关闭文档 |
### 4.6 Docker 部署配置
已在以下文件加入 `ENABLE_DOCS=false`,并通过 docker-compose 透传(默认 `false`
```
docker/app/.env → ENABLE_DOCS=false
docker/compute/.env → ENABLE_DOCS=false
docker/compute/.env.example → ENABLE_DOCS=false
docker/app/docker-compose.yml → ENABLE_DOCS: ${ENABLE_DOCS:-false}
docker/compute/docker-compose.yml → ENABLE_DOCS: ${ENABLE_DOCS:-false}
```
### 4.7 如何临时开启(排查/调试)
```bash
# 后端:非 prod 环境默认已开启prod 环境临时开启
ENABLE_DOCS=true docker compose -f docker/app/docker-compose.yml up -d backend-api
# 算力节点:临时开启(生产默认关闭)
ENABLE_DOCS=true docker compose -f docker/compute/docker-compose.yml up -d compute-api
```
> ⚠️ 仅在可信内网调试时开启,用毕改回 `false`。
### 4.8 验证方法
```bash
# 关闭状态下三个地址均应返回 404
curl -s -o /dev/null -w "%{http_code}\n" http://<host>/docs # 404
curl -s -o /dev/null -w "%{http_code}\n" http://<host>/redoc # 404
curl -s -o /dev/null -w "%{http_code}\n" http://<host>/openapi.json # 404
# 健康检查不受影响
curl -s http://<host>/modelTF/health
```
---
## 5. 算力节点FastAPI
### 5.1 文档开关
见 [§4.4](#44-算力节点开关逻辑computeapisecuritypy--computeapimainpy),逻辑与后端一致,
生产(`COMPUTE_AUTH_ENABLED=true`)默认关闭。
> 补充:原 token 鉴权中间件已覆盖全部非 health 路径;现在文档路由同时被 FastAPI 层关闭,
> 属于**纵深防御**(双重保护)。
### 5.2 `download_file` glob 穿越 + 符号链接跟随(`compute/api/main.py`
**问题**
```python
matches = list(upload_root.glob(f"{file_id}_*")) # file_id 来自 URL直接拼进 glob
return FileResponse(matches[0]) # 跟随符号链接
```
- `Path.glob` 支持 `..` 段,`file_id` 注入 `../` 可**穿越出 upload 目录**(已实测确认)
- Linux 上目录内符号链接可被 `FileResponse` 跟随 → 读取任意文件
**修复**
```python
# 1) file_id 字符白名单:仅字母/数字/_/-,含 .、/、% 等一律 400
if not file_id or not all(c.isalnum() or c in {"_", "-"} for c in file_id):
raise HTTPException(status_code=400, detail="invalid file id")
matches = list(upload_root.glob(f"{file_id}_*"))
if not matches:
raise HTTPException(status_code=404, detail="file not found")
# 2) 解析符号链接后必须仍位于 upload 根目录内
resolved = matches[0].resolve()
if not _path_inside(upload_root, resolved):
raise HTTPException(status_code=404, detail="file not found")
return FileResponse(resolved)
```
**功能影响**:服务端生成的 `file_<时间戳>` 格式完全兼容;外部工具使用正常 `file_id` 下载不受影响。
### 5.3 已确认安全的同类文件端点(无需改动)
| 端点 | 保护机制 |
|------|----------|
| `compute/files/list` | `_path_inside()` + `.resolve()`,符号链接逃逸被阻断 |
| `compute/files/read` | 同上 |
| `compute/files/upload` | 同上 + 文件名取 `.name` |
| `compute/files/import-local` | 目标路径 `_path_inside()` 校验 |
| 后端 `data-process` 存储 `LocalDataProcessStorage` | `lstat` + `S_ISLNK` + `O_NOFOLLOW` + 规范化引用校验,彻底防符号链接 |
---
## 6. 测试与验证结果
| 验证项 | 结果 |
|--------|------|
| 计算节点全量测试 | **22 passed, 1 skipped**(跳过项为 Windows 无权限建符号链接Linux 生产环境会执行) |
| 新增 compute 下载安全测试 | 正常下载 200穿越样本 400/404符号链接逃逸 404 |
| 新增后端 data_convert 安全测试 | **13 passed**(穿越样本 9 项全拦截 + 鉴权覆盖检查) |
| 后端文档开关测试 | 6 passed2 项 `create_app` 集成测试需完整依赖,在 WSL 下运行) |
| 实时验证 | 生产环境 `/docs` `/redoc` `/openapi.json` 均返回 **404**`download_file` 合法 `file_123456` 返回 200 |
---
## 7. 变更文件清单
**后端**
- `backend/app/core/config.py` — 新增 `_bool_env``docs_kwargs()``Settings.enable_docs`
- `backend/app/main.py``FastAPI(...)` 接入 `docs_kwargs`
- `backend/app/modules/data_convert/router.py` — 输出文件名白名单 + 全部端点补鉴权
**算力节点**
- `compute/api/security.py` — 新增文档开关模块(`docs_enabled` / `docs_kwargs`
- `compute/api/main.py` — 文档开关接入 + `download_file` 加固
**Docker 配置**
- `docker/app/.env``docker/compute/.env``docker/compute/.env.example``ENABLE_DOCS=false`
- `docker/app/docker-compose.yml``docker/compute/docker-compose.yml` — 透传 `ENABLE_DOCS`
**测试**
- `backend/tests/test_docs_security.py``backend/tests/test_data_convert_security.py`
- `compute/tests/test_security.py``compute/tests/test_file_download_security.py`