diff --git a/src/fastmcp/apps/file_upload.py b/src/fastmcp/apps/file_upload.py index 8cc91e039..b3acc90e2 100644 --- a/src/fastmcp/apps/file_upload.py +++ b/src/fastmcp/apps/file_upload.py @@ -71,6 +71,15 @@ _TEXT_EXTENSIONS = frozenset( ) +def _b64_decoded_size(b64: str) -> int: + """Return the exact decoded byte-length of a base64 string without decoding it.""" + n = len(b64) + if n == 0: + return 0 + padding = b64.count("=", max(0, n - 2)) + return n * 3 // 4 - padding + + def _format_size(size: int) -> str: if size < 1024: return f"{size} B" @@ -274,10 +283,13 @@ class FileUpload(FastMCPApp): def store_files(files: list[dict], ctx: Context) -> list[dict]: """Store uploaded files. Receives file objects with name, size, type, data (base64).""" for f in files: - if f.get("size", 0) > provider._max_file_size: + # Compute actual data size from the base64 payload rather + # than trusting the client-reported ``size`` field. + actual_size = _b64_decoded_size(f.get("data", "")) + if actual_size > provider._max_file_size: raise ValueError( f"File {f.get('name', '?')!r} exceeds max size " - f"({_format_size(f['size'])} > " + f"({_format_size(actual_size)} > " f"{_format_size(provider._max_file_size)})" ) return provider.on_store(files, ctx) diff --git a/tests/apps/test_file_upload.py b/tests/apps/test_file_upload.py index c8bd7470e..0ef107a3d 100644 --- a/tests/apps/test_file_upload.py +++ b/tests/apps/test_file_upload.py @@ -5,7 +5,32 @@ import base64 import pytest from fastmcp import FastMCP -from fastmcp.apps.file_upload import FileUpload +from fastmcp.apps.file_upload import FileUpload, _b64_decoded_size + + +class TestB64DecodedSize: + """Unit tests for the _b64_decoded_size helper.""" + + @pytest.mark.parametrize("size", [0, 1, 2, 3, 4, 50, 99, 100, 101, 1000]) + def test_matches_actual_decode(self, size: int): + data = b"x" * size + b64 = base64.b64encode(data).decode() + assert _b64_decoded_size(b64) == size + + def test_empty_string(self): + assert _b64_decoded_size("") == 0 + + def test_no_padding(self): + # 3 bytes → 4 base64 chars, no padding + assert _b64_decoded_size(base64.b64encode(b"abc").decode()) == 3 + + def test_one_pad(self): + # 2 bytes → 4 base64 chars with 1 '=' + assert _b64_decoded_size(base64.b64encode(b"ab").decode()) == 2 + + def test_two_pads(self): + # 1 byte → 4 base64 chars with 2 '=' + assert _b64_decoded_size(base64.b64encode(b"a").decode()) == 1 def _make_file( @@ -123,6 +148,23 @@ class TestFileUploadProvider: with pytest.raises(Exception, match="exceeds max size"): await server.call_tool("Files___store_files", {"files": [big_file]}) + async def test_max_file_size_checks_actual_data_not_reported_size(self): + """Size limit should be enforced on actual base64 payload, not the + client-reported ``size`` field which can be spoofed.""" + server = FastMCP("test", providers=[FileUpload(max_file_size=100)]) + + big_content = "x" * 200 + big_b64 = base64.b64encode(big_content.encode()).decode() + spoofed_file = { + "name": "spoofed.bin", + "size": 1, # lies about size + "type": "application/octet-stream", + "data": big_b64, + } + + with pytest.raises(Exception, match="exceeds max size"): + await server.call_tool("Files___store_files", {"files": [spoofed_file]}) + class TestFileUploadSubclass: async def test_custom_storage(self):