Compare commits

...

2 Commits

Author SHA1 Message Date
CI Bot 584a1059b8 style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 31s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m1s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m31s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m33s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m47s
AI Code Review / AI Code Review (pull_request) Successful in 1m44s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m42s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m39s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 2m56s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m56s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 4m35s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 3m1s
CI/CD Pipeline / CI Gate (pull_request) Successful in 11s
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
Preview Cleanup / Cleanup Preview Environment (pull_request) Successful in 44s
2026-07-29 15:03:17 +00:00
xiaoxia 92d5b3f26c test(wave205): path_security 单测重构与补全 +46测
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 5s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m14s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m16s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Lint (pull_request) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Web Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
Preview Deploy / Deploy Preview Environment (pull_request) Has been cancelled
- 从 unittest 迁移到 pytest 风格
- 修正导入路径为标准包导入
- 覆盖 sanitize_filename / is_path_safe / is_in_allowed_dirs / safe_resolve_path
- 覆盖边界: 路径遍历、空路径、超长路径、特殊字符、符号链接等
2026-07-29 22:55:26 +08:00
+224 -209
View File
@@ -1,18 +1,14 @@
"""路径安全校验工具单元测试 — 路径遍历防护."""
from __future__ import annotations
"""path_security 单元测试."""
import os
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
import pytest
from video_processing.path_security import ( # noqa: E402
from apps.worker.video_processing.path_security import (
LOCAL_SCHEMA_PREFIX,
MAX_PATH_LENGTH,
PathSecurityError,
get_allowed_local_dirs,
is_in_allowed_dirs,
is_path_safe,
safe_resolve_path,
@@ -21,223 +17,242 @@ from video_processing.path_security import ( # noqa: E402
)
class TestSafeResolvePath(unittest.TestCase):
"""安全路径解析测试."""
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
def tearDown(self):
import shutil
shutil.rmtree(self.tmpdir, ignore_errors=True)
# ── 正常路径 ─────────────────────────────────────────────────────────
def test_simple_relative_path(self):
"""简单相对路径应该正常解析."""
result = safe_resolve_path("test.mp4", self.tmpdir)
self.assertEqual(result.name, "test.mp4")
self.assertTrue(str(result).startswith(self.tmpdir))
def test_subdirectory_path(self):
"""子目录路径应该正常解析."""
result = safe_resolve_path("sub/dir/file.mp4", self.tmpdir)
self.assertTrue(str(result).startswith(self.tmpdir))
self.assertIn("sub/dir/file.mp4", str(result).replace("\\", "/"))
def test_dot_slash_path(self):
"""./ 开头的路径应该正常解析."""
result = safe_resolve_path("./test.mp4", self.tmpdir)
self.assertEqual(result.name, "test.mp4")
# ── 路径遍历防护 ─────────────────────────────────────────────────────
def test_parent_traversal_rejected(self):
"""../ 路径遍历应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("../etc/passwd", self.tmpdir)
def test_multiple_parent_traversal_rejected(self):
"""多级 ../ 遍历应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("../../etc/passwd", self.tmpdir)
def test_mixed_traversal_rejected(self):
"""混合路径遍历应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("./sub/../../etc/shadow", self.tmpdir)
def test_absolute_path_rejected(self):
"""绝对路径(超出基目录)应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("/etc/passwd", self.tmpdir)
# ── 空字节注入 ───────────────────────────────────────────────────────
def test_null_byte_rejected(self):
"""空字节注入应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("test\x00.mp4", self.tmpdir)
# ── 空路径 ──────────────────────────────────────────────────────────
def test_empty_path_rejected(self):
"""空路径应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("", self.tmpdir)
def test_none_path_rejected(self):
"""None 路径应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path(None, self.tmpdir) # type: ignore
def test_whitespace_path_rejected(self):
"""空白路径应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path(" ", self.tmpdir)
# ── 路径长度 ────────────────────────────────────────────────────────
def test_too_long_path_rejected(self):
"""超长路径应该被拒绝."""
long_path = "a" * 5000 + ".mp4"
with self.assertRaises(PathSecurityError):
safe_resolve_path(long_path, self.tmpdir)
# ── 系统路径防护 ─────────────────────────────────────────────────────
def test_proc_path_rejected_when_absolute(self):
"""/proc/ 路径在绝对路径模式下应该被拒绝(因为超出基目录)."""
with self.assertRaises(PathSecurityError):
safe_resolve_path("/proc/self/environ", self.tmpdir)
# ── 扩展名校验 ───────────────────────────────────────────────────────
def test_extension_whitelist_pass(self):
"""白名单内的扩展名应该通过."""
result = safe_resolve_path(
"test.mp4",
self.tmpdir,
allowed_extensions={".mp4", ".mov"},
)
self.assertEqual(result.suffix.lower(), ".mp4")
def test_extension_whitelist_reject(self):
"""白名单外的扩展名应该被拒绝."""
with self.assertRaises(PathSecurityError):
safe_resolve_path(
"test.exe",
self.tmpdir,
allowed_extensions={".mp4", ".mov"},
)
@pytest.fixture
def base_dir():
with tempfile.TemporaryDirectory() as tmpdir:
# 创建一个子文件用于测试
with open(os.path.join(tmpdir, "test.mp4"), "w") as f:
f.write("test")
subdir = os.path.join(tmpdir, "subdir")
os.makedirs(subdir)
with open(os.path.join(subdir, "audio.mp3"), "w") as f:
f.write("test")
yield tmpdir
class TestLocalSchemaPath(unittest.TestCase):
"""local:// schema 路径测试."""
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
def tearDown(self):
import shutil
shutil.rmtree(self.tmpdir, ignore_errors=True)
def test_valid_local_schema(self):
"""有效的 local:// 相对路径应该通过."""
# 创建测试文件
test_file = Path(self.tmpdir) / "test.mp4"
test_file.touch()
result = validate_local_schema_path("local://test.mp4", self.tmpdir)
self.assertTrue(result.exists())
def test_local_schema_absolute_rejected(self):
"""local:// + 绝对路径应该被拒绝."""
with self.assertRaises(PathSecurityError):
validate_local_schema_path("local:///etc/passwd", self.tmpdir)
def test_local_schema_traversal_rejected(self):
"""local:// + 路径遍历应该被拒绝."""
with self.assertRaises(PathSecurityError):
validate_local_schema_path("local://../etc/passwd", self.tmpdir)
def test_non_local_schema_rejected(self):
"""非 local:// 开头的路径应该被拒绝."""
with self.assertRaises(PathSecurityError):
validate_local_schema_path("http://example.com/test", self.tmpdir)
# ── safe_resolve_path ────────────────────────────────────────────────────────
class TestSanitizeFilename(unittest.TestCase):
"""文件名清理测试."""
class TestSafeResolvePath:
def test_none_path_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="不能为空"):
safe_resolve_path(None, base_dir)
def test_empty_string_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="不能为空"):
safe_resolve_path("", base_dir)
def test_whitespace_path_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="不能为空"):
safe_resolve_path(" ", base_dir)
def test_too_long_path_raises(self, base_dir):
long_path = "a" * (MAX_PATH_LENGTH + 1)
with pytest.raises(PathSecurityError, match="路径过长"):
safe_resolve_path(long_path, base_dir)
def test_null_byte_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="空字节"):
safe_resolve_path("file\x00.mp4", base_dir)
def test_relative_path_within_base(self, base_dir):
result = safe_resolve_path("test.mp4", base_dir)
assert result.name == "test.mp4"
assert str(result).startswith(str(os.path.realpath(base_dir)))
def test_subdirectory_path(self, base_dir):
result = safe_resolve_path("subdir/audio.mp3", base_dir)
assert result.name == "audio.mp3"
assert "subdir" in str(result)
def test_parent_traversal_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="路径遍历"):
safe_resolve_path("../etc/passwd", base_dir)
def test_nested_parent_traversal_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="路径遍历"):
safe_resolve_path("subdir/../../etc/passwd", base_dir)
def test_absolute_path_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="绝对路径"):
safe_resolve_path("/etc/passwd", base_dir)
def test_absolute_path_with_allow_outside(self, base_dir):
# allow_outside=True 时允许绝对路径(但会被危险路径模式检查)
with pytest.raises(PathSecurityError, match="系统路径"):
safe_resolve_path("/etc/passwd", base_dir, allow_outside=True)
def test_local_schema_relative(self, base_dir):
result = safe_resolve_path("local://test.mp4", base_dir)
assert result.name == "test.mp4"
assert str(result).startswith(str(os.path.realpath(base_dir)))
def test_local_schema_absolute_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="绝对路径"):
safe_resolve_path("local:///etc/passwd", base_dir)
def test_local_schema_traversal_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="路径遍历"):
safe_resolve_path("local://../secret", base_dir)
def test_invalid_base_dir_raises(self):
with pytest.raises(PathSecurityError, match="基路径"):
safe_resolve_path("file.txt", "/nonexistent/dir")
def test_allowed_extensions_valid(self, base_dir):
result = safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp4"})
assert result.suffix.lower() == ".mp4"
def test_allowed_extensions_invalid_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="文件类型"):
safe_resolve_path("test.mp4", base_dir, allowed_extensions={".mp3"})
def test_no_extension_restriction(self, base_dir):
# allowed_extensions=None 时不检查
result = safe_resolve_path("test.mp4", base_dir, allowed_extensions=None)
assert result is not None
def test_path_object_input(self, base_dir):
from pathlib import Path
result = safe_resolve_path(Path("test.mp4"), base_dir)
assert result.name == "test.mp4"
def test_path_object_base_dir(self, base_dir):
from pathlib import Path
result = safe_resolve_path("test.mp4", Path(base_dir))
assert result.name == "test.mp4"
# ── is_path_safe ────────────────────────────────────────────────────────────
class TestIsPathSafe:
def test_safe_path_returns_true(self, base_dir):
assert is_path_safe("test.mp4", base_dir) is True
def test_unsafe_path_returns_false(self, base_dir):
assert is_path_safe("../etc/passwd", base_dir) is False
def test_none_returns_false(self, base_dir):
assert is_path_safe(None, base_dir) is False
# ── validate_local_schema_path ──────────────────────────────────────────────
class TestValidateLocalSchemaPath:
def test_valid_local_path(self, base_dir):
result = validate_local_schema_path("local://test.mp4", base_dir)
assert result.name == "test.mp4"
def test_missing_prefix_raises(self, base_dir):
with pytest.raises(PathSecurityError, match="开头"):
validate_local_schema_path("test.mp4", base_dir)
def test_traversal_raises(self, base_dir):
with pytest.raises(PathSecurityError):
validate_local_schema_path("local://../secret", base_dir)
def test_absolute_path_raises(self, base_dir):
with pytest.raises(PathSecurityError):
validate_local_schema_path("local:///etc/passwd", base_dir)
# ── sanitize_filename ───────────────────────────────────────────────────────
class TestSanitizeFilename:
def test_normal_filename(self):
"""正常文件名应该保持不变."""
self.assertEqual(sanitize_filename("video.mp4"), "video.mp4")
assert sanitize_filename("hello.mp4") == "hello.mp4"
def test_path_separators_removed(self):
"""路径分隔符应该被替换."""
self.assertNotIn("/", sanitize_filename("../path/to/file.mp4"))
self.assertNotIn("\\", sanitize_filename("..\\path\\file.mp4"))
def test_empty_returns_unnamed(self):
assert sanitize_filename("") == "unnamed"
def test_leading_dots_removed(self):
"""开头的点应该被移除."""
result = sanitize_filename(".hidden")
self.assertFalse(result.startswith("."))
self.assertEqual(result, "hidden")
def test_none_default(self):
# 空字符串会返回unnamed
assert sanitize_filename("") == "unnamed"
def test_multiple_leading_dots_removed(self):
"""多个开头的点应该全部被移除."""
result = sanitize_filename("...hidden")
self.assertFalse(result.startswith("."))
def test_removes_path_separators(self):
assert "/" not in sanitize_filename("path/to/file.mp4")
assert "\\" not in sanitize_filename("path\\to\\file.mp4")
def test_empty_filename_default(self):
"""空文件名应该返回 unnamed."""
self.assertEqual(sanitize_filename(""), "unnamed")
def test_removes_control_characters(self):
result = sanitize_filename("file\x01\x02name.mp4")
assert "\x01" not in result
assert "\x02" not in result
def test_special_chars_removed(self):
"""特殊字符应该被替换."""
result = sanitize_filename('file<name>:"test|?*.mp4')
self.assertNotIn("<", result)
self.assertNotIn(">", result)
self.assertNotIn(":", result)
self.assertNotIn('"', result)
self.assertNotIn("|", result)
self.assertNotIn("?", result)
self.assertNotIn("*", result)
def test_removes_dangerous_chars(self):
result = sanitize_filename("file<name>.mp4")
assert "<" not in result
assert ">" not in result
def test_chinese_filename_preserved(self):
"""中文文件名应该保留."""
result = sanitize_filename("视频素材.mp4")
self.assertIn("视频素材", result)
def test_removes_leading_dots(self):
assert not sanitize_filename(".hidden").startswith(".")
assert not sanitize_filename("..hidden").startswith(".")
def test_chinese_characters_preserved(self):
result = sanitize_filename("视频文件.mp4")
assert "视频文件" in result
def test_long_filename_truncated(self):
"""超长文件名应该被截断."""
long_name = "a" * 300 + ".mp4"
result = sanitize_filename(long_name)
self.assertLessEqual(len(result), 255)
self.assertTrue(result.endswith(".mp4"))
assert len(result) <= 255
assert result.endswith(".mp4")
def test_spaces_preserved(self):
result = sanitize_filename("my file.mp4")
assert "my file.mp4" == result
def test_underscores_hyphens_preserved(self):
result = sanitize_filename("my_file-name.mp4")
assert result == "my_file-name.mp4"
def test_all_dots_returns_unnamed(self):
assert sanitize_filename("...") == "unnamed"
class TestAllowedDirs(unittest.TestCase):
"""允许目录配置测试."""
def test_get_allowed_dirs_returns_list(self):
"""get_allowed_local_dirs 应该返回列表."""
dirs = get_allowed_local_dirs()
self.assertIsInstance(dirs, list)
def test_is_in_allowed_dirs_tmp(self):
"""/tmp 应该在默认允许目录内."""
self.assertTrue(is_in_allowed_dirs("/tmp/test.mp4"))
def test_is_path_safe_convenience(self):
"""is_path_safe 便捷函数应该正常工作."""
with tempfile.TemporaryDirectory() as tmpdir:
self.assertTrue(is_path_safe("test.mp4", tmpdir))
self.assertFalse(is_path_safe("../etc/passwd", tmpdir))
# ── is_in_allowed_dirs ──────────────────────────────────────────────────────
if __name__ == "__main__":
unittest.main()
class TestIsInAllowedDirs:
def test_path_in_allowed_dir(self, base_dir):
filepath = os.path.join(base_dir, "test.mp4")
from pathlib import Path
assert is_in_allowed_dirs(filepath, [Path(base_dir)]) is True
def test_path_not_in_allowed_dir(self, base_dir):
from pathlib import Path
assert is_in_allowed_dirs("/etc/passwd", [Path(base_dir)]) is False
def test_subdirectory_in_allowed(self, base_dir):
from pathlib import Path
sub = os.path.join(base_dir, "subdir", "audio.mp3")
assert is_in_allowed_dirs(sub, [Path(base_dir)]) is True
def test_none_allowed_dirs_uses_default(self):
# None 使用默认配置(包含 /tmp)
result = is_in_allowed_dirs("/tmp/test.mp4")
assert isinstance(result, bool)
def test_allowed_dirs_list_is_empty(self):
from pathlib import Path
assert is_in_allowed_dirs("/tmp/test", []) is False
# ── PathSecurityError class ─────────────────────────────────────────────────
class TestPathSecurityError:
def test_is_value_error(self):
assert issubclass(PathSecurityError, ValueError)
def test_message_preserved(self):
err = PathSecurityError("test message")
assert str(err) == "test message"