From 2ee9af4bd02b0a1e99e5a139c4a419245e50cc0a Mon Sep 17 00:00:00 2001 From: admin Date: Tue, 15 Sep 2026 23:11:27 +0800 Subject: [PATCH] =?UTF-8?q?fix(bioflow):=20=E8=B7=AF=E5=BE=84=E7=99=BB?= =?UTF-8?q?=E8=AE=B0=E7=9B=AE=E5=BD=95=E5=B1=95=E5=BC=80+=E9=BB=91?= =?UTF-8?q?=E5=90=8D=E5=8D=95+=E8=87=AA=E5=8A=A8=E6=8C=82=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 登记目录自动展开多文件(扩展名白名单过滤,上限 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 --- bioflow_service/app/config.py | 15 ++++ bioflow_service/app/context.py | 11 +++ bioflow_service/app/routers/biojobs.py | 85 ++++++++++++++++--- .../app/services/bioflow/kernel.py | 10 ++- .../app/services/bioflow/runner.py | 26 +++++- bioflow_service/tests/test_api.py | 45 ++++++++-- bioflow_service/tests/test_kernel.py | 9 ++ 7 files changed, 184 insertions(+), 17 deletions(-) diff --git a/bioflow_service/app/config.py b/bioflow_service/app/config.py index f9d60d9..ceb17fa 100644 --- a/bioflow_service/app/config.py +++ b/bioflow_service/app/config.py @@ -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 可配） diff --git a/bioflow_service/app/context.py b/bioflow_service/app/context.py index 4bfc967..be579d2 100644 --- a/bioflow_service/app/context.py +++ b/bioflow_service/app/context.py @@ -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"], diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index 94c883f..26d925a 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/bioflow_service/app/routers/biojobs.py @@ -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") diff --git a/bioflow_service/app/services/bioflow/kernel.py b/bioflow_service/app/services/bioflow/kernel.py index ee70239..550dd29 100644 --- a/bioflow_service/app/services/bioflow/kernel.py +++ b/bioflow_service/app/services/bioflow/kernel.py @@ -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) diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index 58d4871..2732f54 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -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"]), diff --git a/bioflow_service/tests/test_api.py b/bioflow_service/tests/test_api.py index fdad853..130919f 100644 --- a/bioflow_service/tests/test_api.py +++ b/bioflow_service/tests/test_api.py @@ -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 # --------------------------------------------------------------------------- diff --git a/bioflow_service/tests/test_kernel.py b/bioflow_service/tests/test_kernel.py index aaa2316..9fa72ab 100644 --- a/bioflow_service/tests/test_kernel.py +++ b/bioflow_service/tests/test_kernel.py @@ -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",