diff --git a/lnbits/core/services/assets.py b/lnbits/core/services/assets.py index 2ed4aab20..ae9e96b00 100644 --- a/lnbits/core/services/assets.py +++ b/lnbits/core/services/assets.py @@ -1,7 +1,9 @@ import base64 import io +from urllib.parse import quote from uuid import uuid4 +import filetype from fastapi import UploadFile from loguru import logger from PIL import Image @@ -10,11 +12,48 @@ from lnbits.core.crud.assets import create_asset, get_user_assets_count from lnbits.core.models.assets import Asset from lnbits.settings import settings +IMAGE_MIME_TYPE_ALIASES = { + "heic": "image/heic", + "heics": "image/heics", + "heif": "image/heif", + "image/jpg": "image/jpeg", + "jpeg": "image/jpeg", + "jpg": "image/jpeg", + "png": "image/png", +} +PIL_IMAGE_FORMAT_MIME_TYPES = { + "JPEG": "image/jpeg", + "PNG": "image/png", +} +INLINE_ASSET_MIME_TYPES = { + "image/heic", + "image/heics", + "image/heif", + "image/jpeg", + "image/png", +} +ASSET_SECURITY_HEADERS = { + "X-Content-Type-Options": "nosniff", + "Content-Security-Policy": ( + "sandbox; default-src 'none'; script-src 'none'; " + "object-src 'none'; base-uri 'none'" + ), +} +THUMBNAIL_FORMAT_MIME_TYPES = { + "jpg": "image/jpeg", + "jpeg": "image/jpeg", + "png": "image/png", +} + async def create_user_asset(user_id: str, file: UploadFile, is_public: bool) -> Asset: if not file.content_type: raise ValueError("File must have a content type.") - if file.content_type.lower() not in settings.lnbits_assets_allowed_mime_types: + + content_type = normalize_asset_mime_type(file.content_type) + filename = file.filename or "unnamed" + + if content_type not in allowed_asset_mime_types(): raise ValueError(f"File type '{file.content_type}' not allowed.") if not settings.is_unlimited_assets_user(user_id): @@ -30,14 +69,26 @@ async def create_user_asset(user_id: str, file: UploadFile, is_public: bool) -> f"File limit of {settings.lnbits_max_asset_size_mb}MB exceeded." ) + stored_mime_type = detect_image_mime_type(contents) + if stored_mime_type != content_type: + logger.warning( + "Image MIME type mismatch: declared={}, detected={}", + content_type, + stored_mime_type, + ) + raise ValueError( + "Image file content does not match declared file type. " + f"Declared: '{content_type}', detected: '{stored_mime_type}'." + ) + thumb_buffer = thumbnail_from_bytes(contents) asset = Asset( id=uuid4().hex, user_id=user_id, - mime_type=file.content_type, + mime_type=stored_mime_type, is_public=is_public, - name=file.filename or "unnamed", + name=filename, size_bytes=len(contents), thumbnail_base64=( base64.b64encode(thumb_buffer.getvalue()).decode("utf-8") @@ -51,6 +102,79 @@ async def create_user_asset(user_id: str, file: UploadFile, is_public: bool) -> return asset +def normalize_asset_mime_type(content_type: str) -> str: + content_type = content_type.split(";", 1)[0].strip().lower() + return IMAGE_MIME_TYPE_ALIASES.get(content_type, content_type) + + +def normalize_media_type(media_type: str) -> str: + return media_type.split(";", 1)[0].strip().lower() or "application/octet-stream" + + +def thumbnail_media_type() -> str: + thumbnail_format = (settings.lnbits_asset_thumbnail_format or "png").strip().lower() + return THUMBNAIL_FORMAT_MIME_TYPES.get(thumbnail_format, "application/octet-stream") + + +def content_disposition(disposition: str, filename: str) -> str: + safe_filename = filename or "unnamed" + quoted_filename = quote(safe_filename, safe="") + if quoted_filename == safe_filename: + return f'{disposition}; filename="{safe_filename}"' + return f"{disposition}; filename*=utf-8''{quoted_filename}" + + +def allowed_asset_mime_types() -> set[str]: + return { + mime_type + for mime_type in ( + normalize_asset_mime_type(mime_type) + for mime_type in settings.lnbits_assets_allowed_mime_types + ) + if mime_type.startswith("image/") + } + + +def detect_image_mime_type(contents: bytes) -> str: + kind = filetype.guess(contents) + mime_type = normalize_asset_mime_type(kind.mime) if kind else None + + if mime_type and mime_type in PIL_IMAGE_FORMAT_MIME_TYPES.values(): + verify_pil_image(contents, mime_type) + return mime_type + + if mime_type and mime_type.startswith("image/"): + return mime_type + + try: + with Image.open(io.BytesIO(contents)) as image: + image.verify() + mime_type = PIL_IMAGE_FORMAT_MIME_TYPES.get(image.format or "") + except Exception as exc: + raise ValueError( + "Image file content does not match declared file type." + ) from exc + + if not mime_type: + raise ValueError("Image file content does not match declared file type.") + + return mime_type + + +def verify_pil_image(contents: bytes, mime_type: str) -> None: + try: + with Image.open(io.BytesIO(contents)) as image: + image.verify() + detected_mime_type = PIL_IMAGE_FORMAT_MIME_TYPES.get(image.format or "") + except Exception as exc: + raise ValueError( + "Image file content does not match declared file type." + ) from exc + + if detected_mime_type != mime_type: + raise ValueError("Image file content does not match declared file type.") + + def thumbnail_from_bytes(contents: bytes) -> io.BytesIO | None: try: image = Image.open(io.BytesIO(contents)) diff --git a/lnbits/core/views/asset_api.py b/lnbits/core/views/asset_api.py index afc0d1b00..86baf3ac7 100644 --- a/lnbits/core/views/asset_api.py +++ b/lnbits/core/views/asset_api.py @@ -16,7 +16,14 @@ from lnbits.core.crud.assets import ( from lnbits.core.models.assets import AssetFilters, AssetInfo, AssetUpdate from lnbits.core.models.misc import SimpleStatus from lnbits.core.models.users import AccountId -from lnbits.core.services.assets import create_user_asset +from lnbits.core.services.assets import ( + ASSET_SECURITY_HEADERS, + INLINE_ASSET_MIME_TYPES, + content_disposition, + create_user_asset, + normalize_media_type, + thumbnail_media_type, +) from lnbits.db import Filters, Page from lnbits.decorators import ( check_account_id_exists, @@ -75,11 +82,7 @@ async def api_get_asset_data( if not asset: raise HTTPException(HTTPStatus.NOT_FOUND, "Asset not found.") - return Response( - content=asset.data, - media_type=asset.mime_type, - headers={"Content-Disposition": f'inline; filename="{asset.name}"'}, - ) + return asset_response(asset.data, asset.mime_type, asset.name) @asset_router.get( @@ -101,14 +104,14 @@ async def api_get_asset_thumbnail( if not asset_info: raise HTTPException(HTTPStatus.NOT_FOUND, "Asset not found.") - return Response( + return asset_response( content=( base64.b64decode(asset_info.thumbnail_base64) if asset_info.thumbnail_base64 else b"" ), - media_type=asset_info.mime_type, - headers={"Content-Disposition": f'inline; filename="{asset_info.name}"'}, + media_type=thumbnail_media_type(), + filename=asset_info.name, ) @@ -172,3 +175,16 @@ async def api_delete_asset( await delete_user_asset(account_id.id, asset_id) return SimpleStatus(success=True, message="Asset deleted successfully.") + + +def asset_response(content: bytes, media_type: str, filename: str) -> Response: + media_type = normalize_media_type(media_type) + disposition = "inline" if media_type in INLINE_ASSET_MIME_TYPES else "attachment" + return Response( + content=content, + media_type=media_type, + headers={ + **ASSET_SECURITY_HEADERS, + "Content-Disposition": content_disposition(disposition, filename), + }, + ) diff --git a/lnbits/settings.py b/lnbits/settings.py index c92611f9a..fee51ca6b 100644 --- a/lnbits/settings.py +++ b/lnbits/settings.py @@ -323,11 +323,6 @@ class AssetSettings(LNbitsSettings): "heic", "heif", "heics", - "text/plain", - "text/json", - "text/xml", - "application/json", - "application/pdf", ] ) lnbits_asset_thumbnail_width: int = Field(default=128, ge=0) diff --git a/tests/api/test_asset_api.py b/tests/api/test_asset_api.py index 7707ad92e..40efe0977 100644 --- a/tests/api/test_asset_api.py +++ b/tests/api/test_asset_api.py @@ -13,14 +13,15 @@ async def test_asset_api_upload_list_update_and_delete( client: AsyncClient, user_headers_from: dict[str, str], ): + payload = get_png_bytes() upload = await client.post( "/api/v1/assets?public_asset=false", headers={"Authorization": user_headers_from["Authorization"]}, - files={"file": ("note.txt", b"hello world", "text/plain")}, + files={"file": ("note.png", payload, "image/png")}, ) assert upload.status_code == 200 asset = upload.json() - assert asset["name"] == "note.txt" + assert asset["name"] == "note.png" assert asset["is_public"] is False page = await client.get("/api/v1/assets/paginated", headers=user_headers_from) @@ -29,27 +30,30 @@ async def test_asset_api_upload_list_update_and_delete( info = await client.get(f"/api/v1/assets/{asset['id']}", headers=user_headers_from) assert info.status_code == 200 - assert info.json()["name"] == "note.txt" + assert info.json()["name"] == "note.png" data = await client.get( f"/api/v1/assets/{asset['id']}/data", headers=user_headers_from ) assert data.status_code == 200 - assert data.content == b"hello world" - assert data.headers["content-disposition"] == 'inline; filename="note.txt"' + assert data.content == payload + assert data.headers["content-type"] == "image/png" + assert data.headers["content-disposition"] == 'inline; filename="note.png"' + assert data.headers["x-content-type-options"] == "nosniff" + assert data.headers["content-security-policy"].startswith("sandbox") updated = await client.put( f"/api/v1/assets/{asset['id']}", headers=user_headers_from, - json={"name": "renamed.txt", "is_public": True}, + json={"name": "renamed.png", "is_public": True}, ) assert updated.status_code == 200 - assert updated.json()["name"] == "renamed.txt" + assert updated.json()["name"] == "renamed.png" assert updated.json()["is_public"] is True public_data = await client.get(f"/api/v1/assets/{asset['id']}/data") assert public_data.status_code == 200 - assert public_data.content == b"hello world" + assert public_data.content == payload deleted = await client.delete( f"/api/v1/assets/{asset['id']}", headers=user_headers_from @@ -98,15 +102,51 @@ async def test_asset_api_enforces_visibility_and_supports_admin_updates( assert admin_updated.json()["is_public"] is True assert admin_updated.json()["name"] == "admin-visible.png" + image_data = await client.get(f"/api/v1/assets/{private_asset.id}/data") + assert image_data.status_code == 200 + assert image_data.headers["content-type"] == "image/png" + assert image_data.headers["content-disposition"] == ( + 'inline; filename="admin-visible.png"' + ) + assert image_data.headers["x-content-type-options"] == "nosniff" + thumbnail = await client.get(f"/api/v1/assets/{private_asset.id}/thumbnail") assert thumbnail.status_code == 200 assert thumbnail.content assert thumbnail.headers["content-type"] == "image/png" + assert thumbnail.headers["content-disposition"] == ( + 'inline; filename="admin-visible.png"' + ) + + +@pytest.mark.anyio +async def test_asset_api_blocks_non_image_uploads( + client: AsyncClient, + user_headers_from: dict[str, str], +): + payload = ( + b'' + b'' + b'' + b"" + b"" + b"" + ) + + blocked = await client.post( + "/api/v1/assets", + headers={"Authorization": user_headers_from["Authorization"]}, + files={"file": ("payload.xsl", payload, "text/xml")}, + ) + + assert blocked.status_code == 400 + assert blocked.json()["detail"] == "File type 'text/xml' not allowed." @pytest.mark.anyio async def test_asset_api_validates_uploads_and_missing_assets( client: AsyncClient, + to_user, user_headers_from: dict[str, str], ): invalid = await client.post( @@ -117,6 +157,15 @@ async def test_asset_api_validates_uploads_and_missing_assets( assert invalid.status_code == 400 assert "not allowed" in invalid.json()["detail"] + fake_image_headers = await get_user_token_headers(client, to_user.id) + fake_image = await client.post( + "/api/v1/assets", + headers={"Authorization": fake_image_headers["Authorization"]}, + files={"file": ("fake.png", b"", "image/png")}, + ) + assert fake_image.status_code == 400 + assert "does not match declared file type" in fake_image.json()["detail"] + missing = await client.delete( f"/api/v1/assets/{uuid4().hex}", headers=user_headers_from, @@ -128,7 +177,9 @@ async def test_asset_api_validates_uploads_and_missing_assets( stored = await create_user_asset( "missing-user-check", - make_upload_file(b"content", filename="content.txt", content_type="text/plain"), + make_upload_file( + get_png_bytes(), filename="content.png", content_type="image/png" + ), is_public=True, ) fetched = await get_user_asset("missing-user-check", stored.id) diff --git a/tests/unit/test_services_assets.py b/tests/unit/test_services_assets.py index 20751ab6d..c952b7f58 100644 --- a/tests/unit/test_services_assets.py +++ b/tests/unit/test_services_assets.py @@ -15,7 +15,7 @@ from tests.helpers import make_upload_file @pytest.mark.anyio async def test_create_user_asset_validates_upload_constraints( - settings: Settings, mocker: MockerFixture + app, settings: Settings, mocker: MockerFixture ): file_without_type = make_upload_file(b"hello", filename="a.txt", content_type=None) with pytest.raises(ValueError, match="File must have a content type."): @@ -31,6 +31,40 @@ async def test_create_user_asset_validates_upload_constraints( ): await create_user_asset("user-1", bad_type, is_public=False) + xsl_upload = make_upload_file( + b"hello", + filename="style.xsl", + content_type="text/xml", + ) + with pytest.raises(ValueError, match="File type 'text/xml' not allowed."): + await create_user_asset("user-1", xsl_upload, is_public=False) + + original_allowed_mime_types = list(settings.lnbits_assets_allowed_mime_types) + try: + settings.lnbits_assets_allowed_mime_types = [ + *original_allowed_mime_types, + "text/xml", + ] + xsl_content = make_upload_file( + b'', + filename="style.xml", + content_type="text/xml", + ) + with pytest.raises(ValueError, match="File type 'text/xml' not allowed."): + await create_user_asset("user-1", xsl_content, is_public=False) + finally: + settings.lnbits_assets_allowed_mime_types = original_allowed_mime_types + + fake_image = make_upload_file( + b"", + filename="fake.png", + content_type="image/png", + ) + with pytest.raises( + ValueError, match="Image file content does not match declared file type." + ): + await create_user_asset("user-1", fake_image, is_public=False) + original_max_assets = settings.lnbits_max_assets_per_user original_max_size = settings.lnbits_max_asset_size_mb original_no_limit_users = list(settings.lnbits_assets_no_limit_users) @@ -40,14 +74,14 @@ async def test_create_user_asset_validates_upload_constraints( settings.lnbits_assets_no_limit_users = [] limited_user = await _create_user() allowed_type = make_upload_file( - b"hello", filename="ok.txt", content_type="text/plain" + _png_bytes(), filename="ok.png", content_type="image/png" ) await create_user_asset(limited_user, allowed_type, is_public=False) blocked_by_count = make_upload_file( - b"again", - filename="again.txt", - content_type="text/plain", + _png_bytes(), + filename="again.png", + content_type="image/png", ) with pytest.raises(ValueError, match="Max upload count of 1 exceeded."): await create_user_asset(limited_user, blocked_by_count, is_public=False) @@ -55,9 +89,9 @@ async def test_create_user_asset_validates_upload_constraints( settings.lnbits_max_asset_size_mb = 0.000001 oversized_user = await _create_user() large_file = make_upload_file( - b"0123456789", - filename="ok.txt", - content_type="text/plain", + _png_bytes(), + filename="ok.png", + content_type="image/png", ) with pytest.raises(ValueError, match="File limit of 1e-06MB exceeded."): await create_user_asset(oversized_user, large_file, is_public=False) @@ -68,29 +102,66 @@ async def test_create_user_asset_validates_upload_constraints( @pytest.mark.anyio -async def test_create_user_asset_success(mocker: MockerFixture): +async def test_create_user_asset_success(app, mocker: MockerFixture): user_id = await _create_user() mocker.patch( "lnbits.core.services.assets.thumbnail_from_bytes", return_value=None, ) - file = make_upload_file(b"hello", filename="hello.txt", content_type="text/plain") + contents = _png_bytes() + file = make_upload_file(contents, filename="hello.png", content_type="image/png") asset = await create_user_asset(user_id, file, is_public=True) stored = await get_user_asset(user_id, asset.id) assert asset.id assert asset.user_id == user_id - assert asset.name == "hello.txt" - assert asset.size_bytes == 5 - assert asset.data == b"hello" + assert asset.name == "hello.png" + assert asset.size_bytes == len(contents) + assert asset.data == contents assert asset.is_public is True assert stored is not None assert stored.id == asset.id - assert stored.data == b"hello" + assert stored.data == contents assert await get_user_assets_count(user_id) == 1 +@pytest.mark.anyio +async def test_create_user_asset_stores_detected_image_mime_type(app): + user_id = await _create_user() + buffer = BytesIO() + Image.new("RGB", (32, 32), color="blue").save(buffer, format="JPEG") + file = make_upload_file( + buffer.getvalue(), filename="photo.jpg", content_type="image/jpg" + ) + + asset = await create_user_asset(user_id, file, is_public=True) + stored = await get_user_asset(user_id, asset.id) + + assert asset.mime_type == "image/jpeg" + assert stored is not None + assert stored.mime_type == "image/jpeg" + + +@pytest.mark.anyio +async def test_create_user_asset_rejects_mismatched_image_content(app): + user_id = await _create_user() + buffer = BytesIO() + Image.new("RGB", (32, 32), color="blue").save(buffer, format="JPEG") + file = make_upload_file( + buffer.getvalue(), filename="photo.png", content_type="image/png" + ) + + with pytest.raises( + ValueError, + match=( + "Image file content does not match declared file type. " + "Declared: 'image/png', detected: 'image/jpeg'." + ), + ): + await create_user_asset(user_id, file, is_public=False) + + def test_thumbnail_from_bytes_success_and_failure(): image = Image.new("RGB", (512, 512), color="red") buffer = BytesIO() @@ -107,3 +178,9 @@ async def _create_user() -> str: user_id = uuid4().hex await create_account(Account(id=user_id, username=f"user_{user_id[:8]}")) return user_id + + +def _png_bytes() -> bytes: + buffer = BytesIO() + Image.new("RGB", (32, 32), color="green").save(buffer, format="PNG") + return buffer.getvalue()