fix(bioflow): 路径登记目录展开+黑名单+自动挂载

- 登记目录自动展开多文件(扩展名白名单过滤,上限 200)
- 移除白名单根校验(用户数据可在任意位置),改敏感路径黑名单
  (/etc /root /proc /var/lib/docker 等前缀 + .env/*.key/id_ed25519 等文件名)
- 登记文件 symlink 进任务 input + server_registered.json 持久化
- 执行容器自动挂载登记路径(-v src:src:ro),跨重启恢复
- 前端 chat.js 兼容 items[] 响应(单文件亦在 items[0])
- 336 用例全绿

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-15 23:11:27 +08:00
co-authored by Claude Fable 5
parent 957f3cb062
commit 2ee9af4bd0
7 changed files with 184 additions and 17 deletions
+15
View File
@@ -56,6 +56,21 @@ class Settings(BaseSettings):
max_upload_gb: float = 100.0 # 上传大小上限(GB;超限 413)
upload_ext_allow: str = ".fasta,.fa,.faa,.fna,.pep,.cds,.gz,.gff,.gff3,.gtf,.txt,.tsv,.csv,.xlsx,.xls,.bed,.vcf,.vcf.gz,.bam,.cram,.sam,.hmm,.nwk,.newick,.treefile,.json,.yaml,.yml,.pdf,.zip,.py,.r,.sh"
# ---- 服务器路径登记(FR-BIO-007;用户数据可在任意位置)----
max_register_files_per_job: int = 200 # 目录展开单任务文件数上限(防误填大目录)
# 敏感路径黑名单:命中前缀即 403(白名单根校验已按用户要求移除)
sensitive_path_prefixes: tuple[str, ...] = (
"/etc", "/root", "/proc", "/sys", "/dev", "/run", "/boot",
"/bin", "/sbin", "/lib", "/lib64", "/usr/bin", "/usr/sbin",
"/usr/lib", "/usr/lib64", "/usr/local/bin", "/var/lib", "/var/log",
"/opt/neo4j", "/var/lib/docker", "/srv/plantai/install",
)
# 敏感文件名(正则,re.match 整名);.env 等含密钥配置、密钥文件不可登记
sensitive_filename_patterns: tuple[str, ...] = (
r"\.env.*", r".*\.key$", r".*\.pem$", r"id_rsa.*", r"id_ed25519.*",
r".*\.p12$", r"\.gitconfig.*",
)
# ---- 用户自定义扩展分析(并行链路;存储根 + AI 转换 LLM 键 + 修复上限)----
extensions_dir: str = "bioflow_data/extensions" # 扩展存储根({ext_id}/extension.json...)
extension_max_repair_rounds: int = 20 # AI 修复循环上限(契约 §6 可配)
+11
View File
@@ -205,6 +205,17 @@ def _record_from_job(job: dict[str, Any], stages: list[dict[str, Any]]) -> Any:
auto_continue=bool(job.get("auto_continue")),
archive_id=job.get("archive_id"),
)
# 登记路径恢复(server_registered.json;供执行容器自动挂载)
try:
import json as _json
reg_file = record.job_dir / "server_registered.json"
if reg_file.is_file():
record.registered_server_paths = (
_json.loads(reg_file.read_text(encoding="utf-8"))
.get("registered_server_paths") or []
)
except Exception: # noqa: BLE001 - 恢复失败不阻断(挂载缺失→阶段报错可查)
pass
for s in stages:
record._stage_states[s["stage"]] = {
"status": s["status"],
+75 -10
View File
@@ -239,22 +239,87 @@ async def upload_file(
async def _register_path(record: Any, server_path: str, settings: Any) -> dict[str, Any]:
"""服务器路径登记(FR-BIO-007):限 BIOFLOW_DATA_ROOT/任务目录内。"""
"""服务器路径登记(FR-BIO-007):单文件或目录展开,任意数据位置(敏感路径黑名单拦截)。
目录:递归收集目录内可登记文件(扩展名白名单),登记为多条记录。
白名单根校验已按用户要求移除,改为敏感路径黑名单(见 settings)。
"""
path = Path(server_path).expanduser().resolve()
if not path.exists():
return _err("PATH_NOT_EXIST", f"路径不存在: {server_path}", 400)
if not path.is_file():
return _err("NOT_A_FILE", f"非文件: {server_path}", 400)
allowed_roots = [Path(settings.data_root_abs).resolve(), record.job_dir.resolve()]
if not any(path.is_relative_to(root) for root in allowed_roots):
return _err("PATH_NOT_ALLOWED", f"路径不在白名单根内: {server_path}", 403)
if not path.is_file() and not path.is_dir():
return _err("NOT_A_FILE", f"非文件或目录: {server_path}", 400)
blocked = _blocked_reason(path, settings)
if blocked:
return _err("PATH_NOT_ALLOWED", f"路径不允许: {server_path}({blocked})", 403)
files: list[Path] = []
if path.is_file():
files = [path]
else:
ext_set = settings.upload_ext_set
for p in sorted(path.rglob("*")):
if p.is_file():
if ext_set and p.suffix.lower() not in ext_set:
continue
if _blocked_reason(p, settings):
continue
files.append(p)
if len(files) >= settings.max_register_files_per_job:
break
if not files:
return _err("VALIDATION",
f"目录内无可登记文件(允许扩展名 {settings.upload_ext_allow[:80]}…)", 400)
# 登记文件 symlink 进任务 input(CLI 扫 /job/input 发现数据)+ 持久化挂载清单
input_dir = record.input_dir
input_dir.mkdir(parents=True, exist_ok=True)
for p in files:
link = input_dir / p.name
try:
if not link.exists():
link.symlink_to(p)
except OSError as exc: # noqa: BLE001
logger.warning("登记 symlink 失败(不阻断): %s -> %s (%s)", p, link, exc)
registered = [str(p) for p in files]
try:
_persist_registered_paths(record, registered)
except Exception as exc: # noqa: BLE001
logger.warning("登记路径持久化失败(不阻断登记): %s", exc)
# BUG-P7-01:路径登记后自动续跑(await_inputs 任务重推;非等待态 no-op)
ctx: Context = context_mod.context
await ctx.runner.retry_after_files(record)
return {"ok": True, "data": {
"file_id": f"{record.id}:{path.name}", "name": path.name,
"size": path.stat().st_size, "path": str(path), "source": "server_path",
}}
items = [
{"file_id": f"{record.id}:{p.name}", "name": p.name,
"size": p.stat().st_size, "path": str(p), "source": "server_path"}
for p in files
]
return {"ok": True, "data": {"items": items, "total": len(items)}}
def _persist_registered_paths(record: Any, registered: list[str]) -> None:
"""登记路径落盘:任务目录 server_registered.json(跨重启可恢复挂载清单)。"""
record.registered_server_paths = list(
dict.fromkeys((getattr(record, "registered_server_paths", None) or []) + registered)
)
payload = {"registered_server_paths": record.registered_server_paths}
(record.job_dir / "server_registered.json").write_text(
json.dumps(payload, ensure_ascii=False), encoding="utf-8"
)
def _blocked_reason(path: Path, settings: Any) -> str | None:
"""敏感路径黑名单:命中返回原因字符串,未命中返回 None。"""
s = str(path)
for prefix in settings.sensitive_path_prefixes:
if s == prefix or s.startswith(prefix + "/"):
return f"敏感路径前缀 {prefix}"
name = path.name
for pat in settings.sensitive_filename_patterns:
if re.match(pat, name):
return f"敏感文件名 {name}"
return None
@router.get("/jobs/{job_id}/files")
@@ -71,6 +71,7 @@ class KernelConfig:
bba_env: dict[str, str] = field(default_factory=dict)
cpu_allow: str = "4,8,16,32"
memory_allow: str = "16g,32g,64g"
extra_mounts: list[str] = field(default_factory=list) # 登记路径只读挂载(src==dst)
class Kernel:
@@ -118,6 +119,7 @@ class Kernel:
cpu: str | None = None,
memory: str | None = None,
module: str | None = None,
extra_mounts: list[str] | None = None,
) -> list[str]:
"""拼装 docker run 命令(list,无 shell;TC-BIO-KERN-001 基元)。
@@ -140,6 +142,9 @@ class Kernel:
cmd += ["-v", f"{job_workspace_dir}:{self.cfg.container_workspace_dir}"]
if self.cfg.data_root:
cmd += ["-v", f"{self.cfg.data_root}:{self.cfg.container_data_dir}:ro"]
# 登记路径只读挂载(src==dst,容器内原路径直接可见;登记时已过黑名单)
for m in (extra_mounts or self.cfg.extra_mounts):
cmd += ["-v", f"{m}:{m}:ro"]
eff_cpu = cpu or str(params.get("cpu") or "")
eff_memory = memory or str(params.get("memory") or "")
cmd += ["--cpus", self._safe_cpu(eff_cpu or self.cfg.default_cpus)]
@@ -205,9 +210,11 @@ class Kernel:
log_path: str | Path | None = None,
cpu: str | None = None,
memory: str | None = None,
extra_mounts: list[str] | None = None,
) -> dict[str, Any]:
"""执行一个阶段 docker:返回 {exit_code, timed_out, log_tail, cmd}。
- extra_mounts:本次执行额外只读挂载的宿主绝对路径(登记路径,src==dst)
- 超时:terminate → 3s → kill(TC-BIO-KERN-009)
- 日志全量写 log_path(任务 logs/),返回尾部 log_tail_lines 行
- 退出码非 0 / 超时 → 失败语义由 runner 判定(本函数只报事实)
@@ -216,7 +223,8 @@ class Kernel:
cpu = cpu or str(params.get("cpu") or "")
memory = memory or str(params.get("memory") or "")
cmd = self.build_command(stage, job_input_dir, job_workspace_dir, params,
cpu=cpu or None, memory=memory or None)
cpu=cpu or None, memory=memory or None,
extra_mounts=extra_mounts)
log_path = Path(log_path) if log_path else Path(job_workspace_dir) / "logs" / f"{stage}.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
+25 -1
View File
@@ -110,6 +110,7 @@ class JobRecord:
auto_continue: bool = False
archive_id: str | None = None
pause_requested: bool = False
registered_server_paths: list[str] = field(default_factory=list) # 登记路径(自动挂载)
_stage_states: dict[str, dict[str, Any]] = field(default_factory=dict)
@property
@@ -347,7 +348,8 @@ class Runner:
"job_id": record.id, "stage": stage, "attempt": attempt,
})
result = await self.executor(stage, record.input_dir, record.workspace_dir,
params, log_path=record.logs_dir / f"{stage}.log")
params, log_path=record.logs_dir / f"{stage}.log",
extra_mounts=self._registered_mounts(record))
log_tail = str(result.get("log_tail") or "")
self._set_stage(record, stage, log_tail=log_tail)
await self._persist_stage(record, stage)
@@ -611,6 +613,17 @@ class Runner:
await self._persist_job(record)
self.spawn(self._resume_loop(record))
def _registered_mounts(self, record: JobRecord) -> list[str]:
"""登记路径 → 挂载清单:取每个登记文件的父目录(去重)。"""
parents: list[str] = []
seen: set[str] = set()
for p in getattr(record, "registered_server_paths", None) or []:
parent = str(Path(p).parent)
if parent not in seen:
seen.add(parent)
parents.append(parent)
return parents
async def retry_after_files(self, record: JobRecord) -> None:
"""文件上传/路径登记后的自动续跑(BUG-P7-01:文件引导闭环 FR-BIO-005/006)。
@@ -680,6 +693,17 @@ class Runner:
auto_continue=bool(row["auto_continue"]),
archive_id=row["archive_id"],
)
# 登记路径恢复(server_registered.json)
try:
import json as _json
reg_file = record.job_dir / "server_registered.json"
if reg_file.is_file():
record.registered_server_paths = (
_json.loads(reg_file.read_text(encoding="utf-8"))
.get("registered_server_paths") or []
)
except Exception: # noqa: BLE001
pass
for s in stages:
record._stage_states[s["stage"]] = {
"status": s["status"], "params": params_dict(s["params"]),
+40 -5
View File
@@ -529,7 +529,7 @@ def test_files_upload_no_job(api):
def test_path_register_valid_and_whitelist(api, tmp_path):
"""TC-BIO-API-042/045:路径登记合法 + 白名单外 403。"""
"""TC-BIO-API-042/045:路径登记合法 + 敏感黑名单 403。"""
resp = _create_job(api)
job_id = _job_id(resp)
# api_context 的 settings 已指向 tmp_path/data(conftest _isolated_env 派生)
@@ -542,11 +542,13 @@ def test_path_register_valid_and_whitelist(api, tmp_path):
f.write(">x\nACGT")
ok = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": allowed})
assert ok.status_code == 200
assert ok.json()["data"]["source"] == "server_path"
# 白名单外
outside = str(tmp_path / "outside.fa")
body = ok.json()["data"]
assert body["items"][0]["source"] == "server_path"
assert body["items"][0]["name"] == "SPL.fa"
# 敏感黑名单(.env 文件名形态)
outside = str(tmp_path / ".env")
with open(outside, "w") as f:
f.write("x")
f.write("SECRET=1")
forbidden = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": outside})
assert forbidden.status_code == 403
assert forbidden.json()["error"]["code"] == "PATH_NOT_ALLOWED"
@@ -557,6 +559,39 @@ def test_path_register_valid_and_whitelist(api, tmp_path):
assert missing.json()["error"]["code"] == "PATH_NOT_EXIST"
def test_path_register_directory_expand(api, tmp_path):
"""目录登记:递归展开多文件,逐个 items;挂载清单落 server_registered.json。"""
resp = _create_job(api)
job_id = _job_id(resp)
d = tmp_path / "rnaseq_fastq"
(d / "s1").mkdir(parents=True)
(d / "s2").mkdir(parents=True)
(d / "s1" / "CRR1_1.fastq.gz").write_text("g1", encoding="utf-8")
(d / "s1" / "CRR1_2.fastq.gz").write_text("g2", encoding="utf-8")
(d / "s2" / "CRR2_1.fastq.gz").write_text("g3", encoding="utf-8")
(d / "notes.txt").write_text("n", encoding="utf-8") # 允许扩展名内
(d / "skip.exe").write_text("x", encoding="utf-8") # 扩展名白名单外,跳过
ok = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": str(d)})
assert ok.status_code == 200
body = ok.json()["data"]
names = {it["name"] for it in body["items"]}
assert names == {"CRR1_1.fastq.gz", "CRR1_2.fastq.gz", "CRR2_1.fastq.gz", "notes.txt"}
assert body["total"] == 4
# 挂载清单落盘
import json as _json
from pathlib import Path as _Path
from app.config import get_settings as _gs
job_dir = _Path(_gs().jobs_dir_abs)
reg = _json.loads((job_dir / job_id / "server_registered.json").read_text(encoding="utf-8"))
assert len(reg["registered_server_paths"]) == 4
# 空目录报错
empty = tmp_path / "empty_dir"
empty.mkdir()
err = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": str(empty)})
assert err.status_code == 400
assert err.json()["error"]["code"] == "VALIDATION"
# ---------------------------------------------------------------------------
# stages 控制 + manifest
# ---------------------------------------------------------------------------
+9
View File
@@ -66,6 +66,15 @@ def test_build_command_basics(kernel, tmp_path):
assert "--workspace" in cmd and "/job/workspace" in cmd
def test_build_command_extra_mounts(kernel, tmp_path):
"""登记路径自动挂载:extra_mounts 每个追加 -v src:src:ro。"""
extra = [str(tmp_path / "rnaseq"), str(tmp_path / "wgs")]
cmd = kernel.build_command("identify", tmp_path / "in", tmp_path / "ws",
extra_mounts=extra)
for m in extra:
assert f"{m}:{m}:ro" in cmd
def test_build_command_params(kernel, tmp_path):
cmd = kernel.build_command("identify", tmp_path / "in", tmp_path / "ws",
{"prefix": "Ft.SPL", "required_pfams": "PF03110",