fbc_uploader/backend/tests/test_download_restrictions.py at 048676dc795ebb348541e3e7242ae8d221b087e4 · arabcoders/fbc_uploader · GitHub

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
"""Test download restrictions for expired/disabled tokens."""

import pytest
from datetime import UTC, datetime, timedelta
from fastapi import status
from unittest.mock import patch

from backend.app.main import app
from backend.app.config import settings
from backend.tests.utils import create_token, initiate_upload, upload_file_via_tus


@pytest.mark.asyncio
async def test_download_blocked_for_disabled_token(client):
    """Test that downloads are blocked when token is disabled and public downloads are off."""
    with patch("backend.app.security.settings.allow_public_downloads", False):
        token_data = await create_token(client, max_uploads=1)
        upload_token = token_data["token"]
        download_token = token_data["download_token"]

        upload_data = await initiate_upload(client, upload_token, "test.txt", 12)
        upload_id = upload_data["upload_id"]
        await upload_file_via_tus(client, upload_id, b"test content", upload_token)

        await client.patch(
            app.url_path_for("update_token", token_value=upload_token),
            json={"disabled": True},
            headers={"Authorization": f"Bearer {settings.admin_api_key}"},
        )

        download_url = app.url_path_for("download_file", download_token=download_token, upload_id=upload_id)
        response = await client.get(download_url)
        assert response.status_code in [status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN], (
            "Download should be blocked for disabled token without auth"
        )


@pytest.mark.asyncio
async def test_download_blocked_for_expired_token(client):
    """Test that downloads are blocked when token is expired and public downloads are off."""
    with patch("backend.app.security.settings.allow_public_downloads", False):
        token_data = await create_token(client, max_uploads=1)
        upload_token = token_data["token"]
        download_token = token_data["download_token"]

        upload_data = await initiate_upload(client, upload_token, "test.txt", 12)
        upload_id = upload_data["upload_id"]
        await upload_file_via_tus(client, upload_id, b"test content", upload_token)

        expired_time = datetime.now(UTC) - timedelta(hours=1)
        await client.patch(
            app.url_path_for("update_token", token_value=upload_token),
            json={"expiry_datetime": expired_time.isoformat()},
            headers={"Authorization": f"Bearer {settings.admin_api_key}"},
        )

        download_url = app.url_path_for("download_file", download_token=download_token, upload_id=upload_id)
        response = await client.get(download_url)
        assert response.status_code in [status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN], (
            "Download should be blocked for expired token without auth"
        )


@pytest.mark.asyncio
async def test_download_allowed_for_disabled_token_with_admin_key(client):
    """Test that admin can download from disabled tokens."""
    token_data = await create_token(client, max_uploads=1)
    upload_token = token_data["token"]
    download_token = token_data["download_token"]

    upload_data = await initiate_upload(client, upload_token, "test.txt", 12)
    upload_id = upload_data["upload_id"]
    await upload_file_via_tus(client, upload_id, b"test content", upload_token)

    await client.patch(
        app.url_path_for("update_token", token_value=upload_token),
        json={"disabled": True},
        headers={"Authorization": f"Bearer {settings.admin_api_key}"},
    )

    download_url = app.url_path_for("download_file", download_token=download_token, upload_id=upload_id)
    response = await client.get(download_url, headers={"Authorization": f"Bearer {settings.admin_api_key}"})
    assert response.status_code == status.HTTP_200_OK, "Admin should be able to download from disabled token"
    assert response.content == b"test content", "Downloaded content should match"


@pytest.mark.asyncio
async def test_get_file_info_blocked_for_disabled_token(client):
    """Test that file info is blocked when token is disabled and public downloads are off."""
    with patch("backend.app.security.settings.allow_public_downloads", False):
        token_data = await create_token(client, max_uploads=1)
        upload_token = token_data["token"]
        download_token = token_data["download_token"]

        upload_data = await initiate_upload(client, upload_token, "test.txt", 12)
        upload_id = upload_data["upload_id"]
        await upload_file_via_tus(client, upload_id, b"test content", upload_token)

        await client.patch(
            app.url_path_for("update_token", token_value=upload_token),
            json={"disabled": True},
            headers={"Authorization": f"Bearer {settings.admin_api_key}"},
        )

        info_url = app.url_path_for("get_file_info", download_token=download_token, upload_id=upload_id)
        response = await client.get(info_url)
        assert response.status_code in [status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN], (
            "File info should be blocked for disabled token without auth"
        )


@pytest.mark.asyncio
async def test_get_file_info_allowed_for_disabled_token_with_admin_key(client):
    """Test that admin can get file info from disabled tokens."""
    token_data = await create_token(client, max_uploads=1)
    upload_token = token_data["token"]
    download_token = token_data["download_token"]

    upload_data = await initiate_upload(client, upload_token, "test.txt", 12)
    upload_id = upload_data["upload_id"]
    await upload_file_via_tus(client, upload_id, b"test content", upload_token)

    await client.patch(
        app.url_path_for("update_token", token_value=upload_token),
        json={"disabled": True},
        headers={"Authorization": f"Bearer {settings.admin_api_key}"},
    )

    info_url = app.url_path_for("get_file_info", download_token=download_token, upload_id=upload_id)
    response = await client.get(info_url, headers={"Authorization": f"Bearer {settings.admin_api_key}"})
    assert response.status_code == status.HTTP_200_OK, "Admin should be able to get file info from disabled token"
    data = response.json()
    assert data["filename"] == "test.txt", "File info should be returned"