Michilis/lnbits
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stricter asset upload (#3982)

Browse Source
This commit is contained in:
Vlad Stan
2026-05-21 12:44:23 +02:00
committed by GitHub
parent 8f033a6047
commit 30e0522419
5 changed files with 303 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lnbits/core/services/assets.py
+127 -3
View File
@@ -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))
lnbits/core/views/asset_api.py
+25 -9
View File
@@ -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),
},
)
lnbits/settings.py
-5
View File
@@ -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)
tests/api/test_asset_api.py
+60 -9
View File
@@ -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'<?xml version="1.0"?>'
b'<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform">'
b'<xsl:template match="/">'
b"<script>alert(1)</script>"
b"</xsl:template>"
b"</xsl:stylesheet>"
)
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"<root></root>", "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)
tests/unit/test_services_assets.py
+91 -14
View File
@@ -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'<stylesheet xmlns="http://www.w3.org/1999/XSL/Transform"></stylesheet>',
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"<?xml version='1.0'?><root></root>",
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()