fix(exporting): F01 路径完整验证前不登记清理目标,清理不跟随符号链接
- 根目录/用户目录/目标文件先通过 lstat 完整校验(拒绝符号链接与非目录), 校验失败时不删除任何磁盘文件(此前父目录为链接时会沿链接 unlink 同名文件) - 失败清理只处理本次登记的受控文件,并通过 O_NOFOLLOW 目录描述符删除; 目录被换成符号链接时拒绝删除,保留任务编号与路径的对应关系 - 测试覆盖:用户目录链接、目标链接、运行中目录被换成链接(受保护文件字节不变)、 失败后正常重试成功
This commit is contained in:
@@ -198,6 +198,114 @@ def test_formula_export_worker_records_path_preparation_failure(workspace, setti
|
||||
assert job.status == "failed"
|
||||
|
||||
|
||||
def _queued_job(ds):
|
||||
return FormulaExport.objects.create(
|
||||
dataset=ds,
|
||||
user=ds.created_by,
|
||||
context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
|
||||
)
|
||||
|
||||
|
||||
def test_failed_export_rejects_symlinked_user_directory_without_touching_files(workspace, settings):
|
||||
from finance.exporting import process_formula_export
|
||||
|
||||
ds, *_ = workspace
|
||||
job = _queued_job(ds)
|
||||
root = Path(settings.MEDIA_ROOT) / "formula-exports"
|
||||
root.mkdir(parents=True)
|
||||
protected = Path(settings.MEDIA_ROOT) / "protected"
|
||||
protected.mkdir()
|
||||
victim = protected / f"{job.pk}.xlsx"
|
||||
victim.write_bytes(b"protected workbook")
|
||||
(root / str(job.user_id)).symlink_to(protected, target_is_directory=True)
|
||||
|
||||
with pytest.raises(ValueError, match="符号链接"):
|
||||
process_formula_export(job.pk)
|
||||
|
||||
job.refresh_from_db()
|
||||
assert job.status == "failed"
|
||||
assert not job.path
|
||||
assert victim.read_bytes() == b"protected workbook"
|
||||
assert list(protected.glob("*.tmp")) == []
|
||||
|
||||
|
||||
def test_failed_export_rejects_symlinked_target_and_keeps_link(workspace, settings):
|
||||
from finance.exporting import process_formula_export
|
||||
|
||||
ds, *_ = workspace
|
||||
job = _queued_job(ds)
|
||||
user_dir = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id)
|
||||
user_dir.mkdir(parents=True)
|
||||
protected_file = Path(settings.MEDIA_ROOT) / "protected.xlsx"
|
||||
protected_file.write_bytes(b"protected workbook")
|
||||
(user_dir / f"{job.pk}.xlsx").symlink_to(protected_file)
|
||||
|
||||
with pytest.raises(ValueError, match="普通文件"):
|
||||
process_formula_export(job.pk)
|
||||
|
||||
job.refresh_from_db()
|
||||
assert job.status == "failed"
|
||||
assert protected_file.read_bytes() == b"protected workbook"
|
||||
assert (user_dir / f"{job.pk}.xlsx").is_symlink()
|
||||
assert list(user_dir.glob("*.tmp")) == []
|
||||
|
||||
|
||||
def test_failed_export_refuses_cleanup_when_parent_directory_swapped_to_symlink(
|
||||
workspace, settings, monkeypatch
|
||||
):
|
||||
from finance.exporting import process_formula_export
|
||||
|
||||
ds, *_ = workspace
|
||||
job = _queued_job(ds)
|
||||
root = Path(settings.MEDIA_ROOT) / "formula-exports"
|
||||
user_dir = root / str(job.user_id)
|
||||
user_dir.mkdir(parents=True)
|
||||
protected = Path(settings.MEDIA_ROOT) / "protected"
|
||||
protected.mkdir()
|
||||
victim = protected / f"{job.pk}.xlsx"
|
||||
victim.write_bytes(b"protected workbook")
|
||||
|
||||
def swap_and_fail(*args, **kwargs):
|
||||
args[-1].write(b"partial")
|
||||
user_dir.rename(user_dir.with_name("swapped"))
|
||||
user_dir.symlink_to(protected, target_is_directory=True)
|
||||
raise ValueError("controlled failure after path validation")
|
||||
|
||||
monkeypatch.setattr("finance.exporting.build_formula_workbook", swap_and_fail)
|
||||
with pytest.raises(ValueError, match="controlled failure"):
|
||||
process_formula_export(job.pk)
|
||||
|
||||
job.refresh_from_db()
|
||||
assert job.status == "failed"
|
||||
assert victim.read_bytes() == b"protected workbook"
|
||||
assert user_dir.is_symlink()
|
||||
# 本次创建的临时文件残留在被换名的原目录中,由清理命令按保留期回收。
|
||||
assert list((root / "swapped").glob(f"{job.pk}.*.tmp"))
|
||||
|
||||
|
||||
def test_formula_export_retry_after_failure_succeeds(workspace, monkeypatch):
|
||||
from finance.exporting import process_formula_export
|
||||
|
||||
ds, *_ = workspace
|
||||
job = _queued_job(ds)
|
||||
|
||||
def fail(*args, **kwargs):
|
||||
args[-1].write(b"partial")
|
||||
raise ValueError("controlled export failure")
|
||||
|
||||
monkeypatch.setattr("finance.exporting.build_formula_workbook", fail)
|
||||
with pytest.raises(ValueError, match="controlled export failure"):
|
||||
process_formula_export(job.pk)
|
||||
monkeypatch.undo()
|
||||
|
||||
stats = process_formula_export(job.pk)
|
||||
job.refresh_from_db()
|
||||
assert stats["rows"] == 5
|
||||
assert job.status == "succeeded"
|
||||
assert Path(job.path).read_bytes()[:2] == b"PK"
|
||||
assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))
|
||||
|
||||
|
||||
def test_formula_workbook_rejects_unclassified_label_collision(workspace):
|
||||
from decimal import Decimal
|
||||
|
||||
|
||||
Reference in New Issue
Block a user