document-management/backend/tests/dms/test_b5_cors_headers.py

from __future__ import annotations

import io
import json
from urllib.parse import quote

from dms.api.v1.blueprint import _merge_exposed_headers

ORIGIN = "http://127.0.0.1:9346"
EXPOSED = {"content-disposition", "x-request-id"}


def _headers(token: str | None = None) -> dict[str, str]:
    headers = {"Origin": ORIGIN}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def _exposed(response) -> list[str]:
    return [
        value.strip().lower()
        for value in response.headers["Access-Control-Expose-Headers"].split(",")
        if value.strip()
    ]


def test_dms_json_success_exposes_headers_and_request_id_matches(client):
    response = client.get("/api/v1/health", headers=_headers())
    payload = response.get_json()
    # 白名单化后默认不启用跨域：未配置 DMS_CORS_ORIGINS 时不回显 Allow-Origin
    assert "Access-Control-Allow-Origin" not in response.headers
    assert EXPOSED <= set(_exposed(response))
    assert "*" not in _exposed(response)
    assert response.headers["X-Request-Id"] == payload["requestId"]


def test_dms_401_and_routing_404_expose_request_id(client):
    unauthorized = client.get("/api/v1/documents", headers=_headers())
    assert unauthorized.status_code == 401
    assert EXPOSED <= set(_exposed(unauthorized))
    assert (
        unauthorized.headers["X-Request-Id"]
        == unauthorized.get_json()["requestId"]
    )
    missing = client.get("/api/v1/not-implemented", headers=_headers())
    assert missing.status_code == 404
    assert EXPOSED <= set(_exposed(missing))
    assert missing.headers["X-Request-Id"] == missing.get_json()["requestId"]


def test_expose_header_merge_is_case_insensitive_and_deduplicated():
    value = _merge_exposed_headers(
        "ETag, x-request-id, etag, CONTENT-DISPOSITION"
    )
    parts = [item.strip().lower() for item in value.split(",")]
    assert parts == ["etag", "x-request-id", "content-disposition"]
    assert len(parts) == len(set(parts))
    assert "*" not in parts


def test_preview_and_chinese_download_expose_headers_once(
    b5_app, b5_client, token_for
):
    from dms.extensions import db
    from dms.models import AuditLog, Document

    token = token_for("admin")
    filename = "中文下载文件.pdf"
    created = b5_client.post(
        "/api/v1/attachments",
        data={
            "file": (io.BytesIO(b"%PDF-1.4\nCORS"), filename),
            "metadata": json.dumps(
                {
                    "documentName": "B5_C1_TEST_中文下载",
                    "attachmentType": "SUPPORT_ACHIEVEMENT",
                    "summary": "CORS响应头验证",
                    "tags": ["CORS"],
                },
                ensure_ascii=False,
            ),
        },
        headers=_headers(token),
        content_type="multipart/form-data",
    ).get_json()["data"]
    preview = b5_client.get(
        f"/api/v1/documents/{created['id']}/preview",
        headers=_headers(token),
    )
    assert preview.status_code == 200
    assert EXPOSED <= set(_exposed(preview))
    assert preview.headers["Content-Disposition"].startswith("inline")
    preview.close()

    download = b5_client.get(
        f"/api/v1/documents/{created['id']}/download",
        headers=_headers(token),
    )
    disposition = download.headers["Content-Disposition"]
    assert download.status_code == 200
    assert download.headers["Access-Control-Allow-Origin"] == ORIGIN
    assert EXPOSED <= set(_exposed(download))
    assert "filename=" in disposition
    assert "filename*=UTF-8''" in disposition
    assert quote(filename, safe="") in disposition
    assert download.headers["X-Request-Id"]
    download.close()
    with b5_app.app_context():
        document = db.session.get(Document, int(created["id"]))
        audit_count = db.session.scalar(
            db.select(db.func.count(AuditLog.id)).where(
                AuditLog.target_id == document.id,
                AuditLog.action_type == "DOWNLOAD_DOCUMENT",
            )
        )
        assert document.download_count == 1
        assert audit_count == 1


def test_ai_response_does_not_gain_dms_expose_headers(client):
    """非 DMS 蓝图路由（/api/health）不带 DMS Expose-Headers；默认亦不开放跨域。"""
    response = client.get("/api/health", headers=_headers())
    assert response.status_code == 200
    assert "Access-Control-Allow-Origin" not in response.headers
    assert "Access-Control-Expose-Headers" not in response.headers