File size: 4,331 Bytes
e86dfae
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""
Unit tests for the direct upload endpoint.

Regressions locked down here:
  * the whole body was buffered in memory before the size check ran
  * any content type was accepted (ALLOWED_FILE_TYPES was never consulted)
  * errors returned {"error": ...} with a 202, not a 4xx
  * the on-disk extension came from the attacker-controlled filename
"""

import io
import os

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api.deps import get_current_user_id
from app.api.v1 import upload as upload_module
from app.db.session import get_db

pytestmark = pytest.mark.unit

PDF_TYPE = "application/pdf"
DOCX_TYPE = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"


@pytest.fixture
def client(tmp_path, monkeypatch, db_session):
    """An app with just the upload router, auth stubbed, storage in tmp_path."""
    monkeypatch.setattr(upload_module, "UPLOAD_DIR", str(tmp_path))
    monkeypatch.setattr(upload_module, "MAX_FILE_SIZE", 1024)  # 1KB ceiling
    monkeypatch.setattr(upload_module, "_CHUNK_SIZE", 256)

    app = FastAPI()
    app.include_router(upload_module.router, prefix="/upload")
    app.dependency_overrides[get_current_user_id] = lambda: "user_test"
    app.dependency_overrides[get_db] = lambda: db_session
    return TestClient(app), tmp_path


# ── Content type enforcement ───────────────────────────────────────────────────


@pytest.mark.parametrize(
    "content_type",
    [
        "application/x-msdownload",
        "image/png",
        "application/octet-stream",
        "text/html",
        "",
    ],
)
def test_rejects_disallowed_content_types(client, content_type):
    c, _ = client
    r = c.post(
        "/upload",
        files={"file": ("payload.pdf", io.BytesIO(b"data"), content_type)},
    )
    assert r.status_code == 415


def test_accepts_allowed_content_type(client):
    c, storage = client
    r = c.post(
        "/upload",
        files={"file": ("report.pdf", io.BytesIO(b"%PDF-1.4 hello"), PDF_TYPE)},
    )
    assert r.status_code == 202
    assert len(os.listdir(storage)) == 1


# ── Size ceiling ───────────────────────────────────────────────────────────────


def test_rejects_oversized_file_with_413(client):
    c, storage = client
    big = b"x" * 5000  # ceiling is 1KB
    r = c.post("/upload", files={"file": ("big.pdf", io.BytesIO(big), PDF_TYPE)})

    assert r.status_code == 413
    # And nothing is left behind on the volume.
    assert os.listdir(storage) == []


def test_rejects_empty_file(client):
    c, storage = client
    r = c.post("/upload", files={"file": ("empty.pdf", io.BytesIO(b""), PDF_TYPE)})

    assert r.status_code == 400
    assert os.listdir(storage) == []


# ── Filename handling ──────────────────────────────────────────────────────────


def test_stored_extension_comes_from_content_type_not_filename(client):
    """A doubled extension must not survive to the filesystem."""
    c, storage = client
    r = c.post(
        "/upload",
        files={"file": ("invoice.pdf.exe", io.BytesIO(b"%PDF-1.4"), PDF_TYPE)},
    )

    assert r.status_code == 202
    stored = os.listdir(storage)
    assert len(stored) == 1
    assert stored[0].endswith(".pdf")
    assert "exe" not in stored[0]


def test_traversal_in_filename_does_not_escape_upload_dir(client):
    c, storage = client
    r = c.post(
        "/upload",
        files={"file": ("../../../evil.pdf", io.BytesIO(b"%PDF-1.4"), PDF_TYPE)},
    )

    assert r.status_code == 202
    stored = os.listdir(storage)
    assert len(stored) == 1
    assert ".." not in stored[0] and "/" not in stored[0]


def test_document_row_records_validated_type_and_real_size(client):
    c, _ = client
    body = b"%PDF-1.4 some content"
    r = c.post("/upload", files={"file": ("doc.pdf", io.BytesIO(body), PDF_TYPE)})

    assert r.status_code == 202
    payload = r.json()
    assert payload["status"] == "pending"
    assert payload["file_name"] == "doc.pdf"