文件上传安全实战：从恶意文件检测到存储隔离的纵深防御

攻击链：上传 → 校验绕过 → 文件落盘 → 解析执行 → 横向移动
          │        │            │          │           │
       入口控制   类型绕过    路径穿越     WebShell    内网探测
                  魔数伪装     0x00截断     存储XSS      RCE

# file_validation.py — 三层类型校验
import os
import imghdr  # 或使用 python-magic

# 允许的扩展名 → 魔数前缀（Magic Bytes）
ALLOWED_TYPES = {
    "png":  b"\x89PNG\r\n\x1a\n",
    "jpg":  b"\xff\xd8\xff",
    "gif":  b"GIF89a",
    "pdf":  b"%PDF-",
    "docx": b"PK\x03\x04",   # OOXML (zip)
}

def validate_file(filename: str, content: bytes) -> tuple[bool, str]:
    """三层校验：扩展名 → 魔数 → 返回结论。"""
    ext = os.path.splitext(filename)[-1].lower().lstrip(".")
    if ext not in ALLOWED_TYPES:
        return False, f"扩展名不允许: {ext}"

    expected_magic = ALLOWED_TYPES[ext]
    actual_magic = content[:len(expected_magic)]
    if actual_magic != expected_magic:
        return False, f"内容与扩展名不符: {ext}（疑似伪装文件）"
    return True, "ok"

# 黑名单思路不可靠（变体太多）：asp → aspx → ashx → asa → cer...
BLOCKED_EXTS = {".php", ".php5", ".phtml", ".asp", ".aspx", ".jsp",
                ".exe", ".bat", ".sh", ".py", ".pl", ".cgi", ".htaccess"}

def strong_ext_validation(filename: str, allowed: set) -> bool:
    """双保险：白名单允许 + 黑名单兜底。"""
    ext = os.path.splitext(filename)[-1].lower()
    if ext in BLOCKED_EXTS:            # 黑名单先拦
        return False
    if ext not in allowed:             # 白名单最终裁决
        return False
    # 双重扩展名陷阱：shell.php.jpg → 取最后一个扩展名
    return True

# svg 是 HTML 的一种，天然可携带 <script>
DANGEROUS_SVG = """<svg xmlns="http://www.w3.org/2000/svg">
  <script>alert(document.cookie)</script>
</svg>"""

def is_svg_with_script(content: bytes) -> bool:
    """检测 SVG 中是否含脚本（svg 应只允许纯矢量内容）。"""
    text = content.decode("utf-8", errors="ignore")
    return "<svg" in text.lower() and any(
        tag in text.lower() for tag in ("<script", "onload=", "onerror="))

def sanitize_svg(content: bytes) -> bytes:
    """白名单清洗 SVG：只保留允许的标签与属性。"""
    from xml.etree import ElementTree
    ALLOWED_TAGS = {"svg", "path", "circle", "rect", "line", "polyline",
                    "polygon", "g", "defs", "text", "tspan"}
    ALLOWED_ATTRS = {"viewbox", "fill", "stroke", "stroke-width", "d",
                     "cx", "cy", "r", "x", "y", "width", "height"}
    try:
        tree = ElementTree.fromstring(content)
        # 移除不安全的标签与属性
        for el in tree.iter():
            el.tag = el.tag.split("}")[-1]
            if el.tag not in ALLOWED_TAGS:
                return b""
            for attr in list(el.attrib):
                if attr.lower() not in ALLOWED_ATTRS:
                    del el.attrib[attr]
        return ElementTree.tostring(tree)
    except Exception:
        return b""

# filename_handling.py — 安全文件名生成
import re, uuid, os

def safe_filename(original: str, upload_dir: str) -> str:
    """
    核心原则：存储文件名与客户端文件名完全解耦。
    用 UUID 作为存储名，原始名只用于展示。
    """
    # 1. 生成随机存储名（杜绝路径穿越、覆盖、注入）
    storage_name = uuid.uuid4().hex
    ext = os.path.splitext(original)[-1].lower()
    # 扩展名也做白名单限制
    if ext not in {".png", ".jpg", ".pdf", ".docx"}:
        raise ValueError("不支持的扩展名")
    safe = storage_name + ext
    return os.path.join(upload_dir, safe)

MALICIOUS_FILENAMES = [
    "../../../etc/passwd",       # 路径穿越
    "..\\..\\windows\\system32", # Windows 变体
    "%2e%2e%2f%2e%2e%2f",       # URL 编码
    "shell.php\x00.jpg",         # 空字节截断（旧系统）
    "a" * 300 + ".png",          # 超长文件名 → DoS
]

def test_no_path_traversal(candidate: str) -> bool:
    """检测路径穿越。生产靠服务端重命名兜底，此处做防御性校验。"""
    decoded = candidate
    if ".." in decoded or decoded.startswith("/") or decoded.startswith("\\"):
        return False
    if "\x00" in decoded:          # 空字节
        return False
    if len(decoded) > 255:         # 超长
        return False
    return True

def escape_display_filename(name: str) -> str:
    """展示用文件名做 HTML 转义，防止注入。"""
    import html
    return html.escape(name)

# 前端展示时用 textContent 而非 innerHTML

# webshell_detector.py — Web Shell 启发式检测
import re

# 常见 Web Shell 特征
SHELL_PATTERNS = [
    r"eval\s*\(",                      # eval 执行
    r"system\s*\(\s*['\"\$_]",         # system 命令
    r"exec\s*\(\s*['\"\$_]",           # exec
    r"passthru\s*\(",                   # 直接输出命令结果
    r"shell_exec\s*\(",                 # shell_exec
    r"\$_GET\s*\[|_\$POST\s*\[",        # 直接取用户输入执行
    r"base64_decode\s*\(",              # 编码混淆
    r"assert\s*\(\s*\$",                # assert 动态执行
    r"popen\s*\(",                      # popen
    r"proc_open\s*\(",                  # proc_open
]

def score_webshell_likelihood(content: bytes) -> tuple[float, list[str]]:
    """统计脚本文件中的 Web Shell 特征命中数。"""
    try:
        text = content.decode("utf-8", errors="ignore")
    except Exception:
        text = str(content)
    hits = [p for p in SHELL_PATTERNS if re.search(p, text, re.I)]
    score = len(hits) / len(SHELL_PATTERNS)
    return score, hits

def is_webshell(content: bytes, threshold: float = 0.3) -> bool:
    score, hits = score_webshell_likelihood(content)
    if score >= threshold:
        print(f"疑似 Web Shell，命中特征: {hits}")
        return True
    return False

# 攻击示例：JPG + PHP 代码复合体
POLYGLOT = b"\xff\xd8\xff\xe0" + b"\x00" * 100 + \
           b"<?php system($_GET['cmd']); ?>"

def detect_polyglot(content: bytes, ext: str) -> bool:
    """图片中嵌入可执行代码的检测。"""
    if ext in {"jpg", "png", "gif"}:
        # 图片文件里出现 PHP 标签 = 可疑
        if b"<?php" in content or b"<?=" in content:
            return True
        # 图片尾部有大量非图片数据
        if ext == "jpg" and b"\xff\xd9" in content:   # EOI 标记
            idx = content.rfind(b"\xff\xd9")
            if len(content) - idx > 4096:   # EOI 后还有大量数据
                return True
    return False

# zip_checker.py — 压缩包安全校验
import zipfile, io

MAX_FILES = 100            # 文件数上限
MAX_TOTAL_SIZE = 100 * 1024 * 1024   # 解压后总大小上限（100MB）
MAX_RATIO = 100            # 单文件压缩比上限

def validate_zip(content: bytes) -> tuple[bool, str]:
    """解压前扫描压缩包元数据，拦截 Zip 炸弹。"""
    try:
        with zipfile.ZipFile(io.BytesIO(content)) as zf:
            infos = zf.infolist()
            if len(infos) > MAX_FILES:
                return False, f"文件数超限: {len(infos)}"

            total = 0
            for info in infos:
                if info.file_size > MAX_TOTAL_SIZE:
                    return False, f"单文件解压过大: {info.file_size}"
                total += info.file_size
                if info.compress_type == zipfile.ZIP_STORED:  # 未压缩 = 无炸弹风险
                    continue
                ratio = info.file_size / max(info.compress_size, 1)
                if ratio > MAX_RATIO:
                    return False, f"压缩比异常({ratio:.0f}): 疑似 Zip 炸弹"
                # 对每个成员递归检测 Web Shell
                data = zf.read(info.filename)
                if is_webshell(data):
                    return False, f"压缩包内含恶意文件: {info.filename}"
            if total > MAX_TOTAL_SIZE:
                return False, f"解压总大小超限: {total}"
            return True, "ok"
    except zipfile.BadZipFile:
        return False, "非法压缩包"

# clamav_scan.py — 病毒扫描集成
import subprocess, tempfile, os

CLAMSCAN = "/usr/bin/clamscan"

def scan_with_clamav(content: bytes) -> tuple[bool, str]:
    """用 ClamAV 扫描内存中的文件内容。"""
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        tmp.write(content)
        tmp_path = tmp.name
    try:
        proc = subprocess.run(
            [CLAMSCAN, "--no-summary", tmp_path],
            capture_output=True, text=True, timeout=60)
        if proc.returncode == 0:
            return True, "clean"
        elif proc.returncode == 1:     # 发现病毒
            return False, proc.stdout.strip()
        else:
            return False, f"扫描异常: {proc.stderr.strip()}"
    finally:
        os.unlink(tmp_path)


def test_scan_detects_eicar():
    """验证扫描器可用：EICAR 测试文件应被检出。"""
    eicar = (b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$"
             b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*")
    ok, result = scan_with_clamav(eicar)
    assert not ok and "EICAR" in result.upper(), "扫描器应检出 EICAR"

模式 A：同步扫描（低延迟场景慎用）
  上传 → ClamAV 扫描 → 通过则入库 / 不通过则拒绝
  优点：入库必干净  缺点：阻塞上传，扫描慢

模式 B：异步扫描 + 隔离（推荐）
  上传 → 先落隔离区 → 后台扫描
  通过 → 移至公开区 / 不通过 → 隔离删除
  优点：上传体验好  缺点：短窗口内恶意文件可能被访问

模式 C：混合（生产推荐）
  小文件同步扫，大文件/批文件异步扫

# 异步隔离架构
def async_scan_and_publish(file_id, content, quarantine_bucket,
                           public_bucket, scan_fn):
    """异步扫描：先隔离，扫描通过后发布。"""
    quarantine_bucket.put(file_id, content)
    ok, reason = scan_fn(content)
    if ok:
        public_bucket.publish(file_id)
        quarantine_bucket.remove(file_id)
        notify_upload_ready(file_id)
    else:
        quarantine_bucket.mark_infected(file_id, reason)
        notify_security_team(file_id, reason)

安全存储架构：
┌──────────────────────────────────────────────┐
│ 隔离区（Quarantine）   → 新上传文件先落此     │
│   · 独立目录/桶，无外部访问                   │
├──────────────────────────────────────────────┤
│ 公开区（Public）       → 扫描通过后发布       │
│   · 独立域（uploads.example.com）            │
│   · 禁止脚本执行                             │
├──────────────────────────────────────────────┤
│ 私有区（Private）      → 敏感文件             │
│   · 权限控制 + 签名 URL 访问                 │
└──────────────────────────────────────────────┘

# nginx — 上传目录禁止 PHP/脚本执行 + 强制下载
location /uploads/ {
    root /var/www/uploads;
    # 关键：不解析任何脚本
    location ~ \.(php|pl|py|jsp|asp|aspx|sh|cgi)$ {
        return 403;
    }
    # 强制作为文件下载，不参与内容解析
    add_header Content-Disposition "attachment";
    add_header X-Content-Type-Options "nosniff";
    # 禁止缓存执行类文件
    add_header Cache-Control "private, no-store";
}

# Flask 中禁止脚本执行的安全返回
def serve_upload(file_id):
    """上传文件以 attachment 下载形式返回，杜绝浏览器解析。"""
    return send_file(
        get_upload_path(file_id),
        as_attachment=True,           # 强制下载而非内联
        download_name=display_name(file_id),   # 用安全展示名
    )

def validate_no_external_refs(content: bytes, content_type: str) -> bool:
    """禁止上传内容引用外部资源（SVG/HTML 场景）。"""
    if "image/svg" in content_type or b"<svg" in content:
        text = content.decode("utf-8", errors="ignore")
        if "href=" in text.lower() or "xlink:href" in text.lower():
            return False   # SVG 引用外部 → 拦截（可能 SSRF）
    return True

# upload_pipeline.py — 完整上传校验管线
class UploadPipeline:
    def __init__(self, quarantine, public, scanner):
        self.q = quarantine
        self.pub = public
        self.scanner = scanner

    def handle_upload(self, filename: str, content_type: str,
                      content: bytes, user_id: str) -> dict:
        """按顺序执行全部校验，任一失败即拒绝。"""
        checks = [
            ("大小限制", lambda: len(content) <= MAX_UPLOAD_SIZE),
            ("类型白名单", lambda: self._check_ext(content, filename)),
            ("魔数校验", lambda: self._check_magic(filename, content)),
            ("WebShell 检测", lambda: not is_webshell(content)),
            ("Polyglot 检测", lambda: not detect_polyglot(content, ext(filename))),
            ("SVG 外部引用", lambda: validate_no_external_refs(content, content_type)),
        ]
        for name, check in checks:
            if not check():
                self._audit(user_id, filename, name, "REJECT")
                return {"status": "rejected", "reason": name}

        # 安全命名 + 隔离存储
        storage_name = safe_filename(filename, "/quarantine")
        file_id = self.q.store(storage_name, content, user_id)

        # 异步病毒扫描后发布
        self.scanner.submit(file_id, on_pass=self._publish, on_fail=self._quarantine)
        self._audit(user_id, filename, "upload", "ACCEPTED")
        return {"status": "accepted", "file_id": file_id}

    def _publish(self, file_id):
        self.pub.move_from_quarantine(file_id)
    def _quarantine(self, file_id, reason):
        self.q.mark_infected(file_id, reason)
        notify_security(file_id, reason)

def _audit(user_id, filename, action, result, extra=""):
    log_structured({
        "event": "file_upload",
        "user_id": user_id,
        "filename_hash": sha256(filename.encode()).hexdigest(),  # 不存原名
        "action": action,
        "result": result,
        "extra": extra,
        "ts": time.time(),
    })

import pytest

def test_rejects_webshell_disguised_as_png():
    pipeline = UploadPipeline(...)
    fake_png = PNG_MAGIC + b"\x00" * 50 + b"<?php system($_GET['c']); ?>"
    result = pipeline.handle_upload("photo.png", "image/png", fake_png, "u1")
    assert result["status"] == "rejected"

def test_rejects_path_traversal():
    result = pipeline.handle_upload("../../etc/passwd", "text/plain", b"x", "u1")
    assert result["status"] == "rejected"

def test_rejects_polyglot():
    result = pipeline.handle_upload("cat.jpg", "image/jpeg", POLYGLOT_JPG, "u1")
    assert result["status"] == "rejected"

def test_accepts_clean_png():
    result = pipeline.handle_upload("logo.png", "image/png", CLEAN_PNG, "u1")
    assert result["status"] == "accepted"

# AWS S3 预签名上传（避免服务器代理文件）
import boto3

def create_presigned_upload(bucket: str, key: str,
                            content_type: str, expires=600) -> str:
    """客户端直传 S3，服务器只签发受限的预签名 URL。"""
    s3 = boto3.client("s3")
    return s3.generate_presigned_post(
        Bucket=bucket,
        Key=key,
        Fields={"Content-Type": content_type,
                "x-amz-meta-userid": "user-123"},
        Conditions=[{"x-amz-meta-userid": "user-123"}],
        ExpiresIn=expires)

def test_presigned_url_restricts_key():
    url = create_presigned_upload("uploads", "files/${filename}", "image/png")
    # 预签名中 key 用 ${filename} 占位，S3 会替换但限制在 files/ 前缀
    # 攻击者无法把 key 改到 files/ 之外

// Spring Boot 上传安全示例
@PostMapping("/upload")
public ResponseEntity<?> upload(@RequestParam("file") MultipartFile file) {
    // 1. 大小限制
    if (file.getSize() > 5 * 1024 * 1024) {
        return ResponseEntity.badRequest().body("文件过大");
    }
    // 2. 类型白名单（魔数级）
    String ext = getExtension(file.getOriginalFilename());
    byte[] magic = getMagicBytes(file);
    if (!isAllowedType(ext, magic)) {
        return ResponseEntity.badRequest().body("非法文件类型");
    }
    // 3. 服务端命名
    String safeName = UUID.randomUUID() + "." + ext;
    // 4. 存储到隔离目录（非静态资源目录）
    file.transferTo(Path.of("/data/quarantine/", safeName));
    return ResponseEntity.ok(safeName);
}

# upload_security_testset.py — 上传攻击样本库
ATTACK_SAMPLES = {
    "webshell_php_jpg":  b"\xff\xd8\xff\xe0" + b"<?php eval($_POST['x']);?>",
    "double_ext":        b"\x89PNG\r\n\x1a\n" + b"pixel data",  # shell.php.png
    "path_traversal":    b"content",   # filename: ../../etc/passwd
    "null_byte":         b"content",   # filename: evil.php\x00.jpg
    "svg_xss":           SVG_MALICIOUS,
    "zip_bomb":          make_zip_bomb(compress_ratio=1000),
    "polyglot":          PNG_WITH_PHP_TAIL,
    "oversized":         b"\x00" * (10 * 1024 * 1024),
}

def run_upload_security_suite(pipeline) -> dict:
    """跑全部攻击样本，统计拦截率。"""
    results = {}
    for name, content in ATTACK_SAMPLES.items():
        filename = name + ".png"  # 统一伪装
        r = pipeline.handle_upload(filename, "image/png", content, "tester")
        results[name] = r["status"] == "rejected"
    intercept_rate = sum(results.values()) / len(results)
    return {"intercept_rate": intercept_rate, "details": results}

def monitor_upload_metrics(recent_uploads):
    """监控上传安全指标：拦截率、异常来源、误杀率。"""
    return {
        "intercept_rate": rejected / total,       # 拦截率基线
        "top_blocked_reason": most_common(reasons), # 主要拦截原因
        "false_positive_rate": human_approved_rejects / rejected,  # 误杀
        "suspicious_ips": top_sources(blocked),    # 异常来源
    }