Security backport (#11584)

* Improvements for get_bulk_queryset (#11581)

* Improvements for get_bulk_queryset

- Limit scope to base view queryset
- Remove ability to provide arbitrary filters
- Remove feedback if zero items are found

* Adjust unit test

* Remove filter test

* Update CHANGELOG.md

* Refactoring for report helper functions (#11579)

* Refactoring for media file report helper functions

* Updated unit tests

* Improved error handling

* Generic path return for asset

* Fix return type annotations

* Fix existing test

* Tweaked unit test

* Collect static files in CI

* Run static step for all DB tests

* Update action.yaml

* Fix for action.yaml

* Updated unit tests

.github/actions/setup/action.yaml

@@ -15,6 +15,10 @@ inputs:
         required: false
         description: 'Install the InvenTree requirements?'
         default: 'false'
+    static:
+        required: false
+        description: 'Should the static files be built?'
+        default: 'false'
     dev-install:
         required: false
         description: 'Install the InvenTree development requirements?'
@@ -103,3 +107,7 @@ runs:
         if: ${{ inputs.update == 'true' }}
         shell: bash
         run: invoke update --skip-backup --skip-static
+      - name: Collect static files
+        if: ${{ inputs.static == 'true' }}
+        shell: bash
+        run: invoke static --skip-plugins

.github/workflows/qc_checks.yaml

@@ -322,6 +322,7 @@ jobs:
           apt-dependency: gettext poppler-utils
           dev-install: true
           update: true
+          static: true
           npm: true
       - name: Download Python Code For `${WRAPPER_NAME}`
         run: git clone --depth 1 https://github.com/inventree/${WRAPPER_NAME} ./${WRAPPER_NAME}
@@ -378,6 +379,7 @@ jobs:
           apt-dependency: gettext poppler-utils
           dev-install: true
           update: true
+          static: true
       - name: Data Export Test
         uses: ./.github/actions/migration
       - name: Test Translations
@@ -480,6 +482,7 @@ jobs:
           pip-dependency: psycopg django-redis>=5.0.0
           dev-install: true
           update: true
+          static: true
       - name: Run Tests
         run: invoke dev.test --check --translations
       - name: Data Export Test
@@ -528,6 +531,7 @@ jobs:
           pip-dependency: mysqlclient
           dev-install: true
           update: true
+          static: true
       - name: Run Tests
         run: invoke dev.test --check --translations
       - name: Data Export Test

src/backend/InvenTree/InvenTree/api.py

@@ -423,23 +423,19 @@ class BulkOperationMixin:
     def get_bulk_queryset(self, request):
         """Return a queryset based on the selection made in the request.
 
-        Selection can be made by providing either:
+        Selection can be made by providing a list of primary key values,
-
+        which will be used to filter the queryset.
-        - items: A list of primary key values
-        - filters: A dictionary of filter values
         """
-        model = self.serializer_class.Meta.model
-
         items = request.data.pop('items', None)
-        filters = request.data.pop('filters', None)
         all_filter = request.GET.get('all', None)
 
-        queryset = model.objects.all()
+        # Return the base queryset for this model
+        queryset = self.get_queryset()
 
-        if not items and not filters and all_filter is None:
+        if not items and all_filter is None:
             raise ValidationError({
                 'non_field_errors': _(
-                    'List of items or filters must be provided for bulk operation'
+                    'List of items must be provided for bulk operation'
                 )
             })
 
@@ -457,19 +453,6 @@ class BulkOperationMixin:
                     'non_field_errors': _('Invalid items list provided')
                 })
 
-        if filters:
-            if type(filters) is not dict:
-                raise ValidationError({
-                    'non_field_errors': _('Filters must be provided as a dict')
-                })
-
-            try:
-                queryset = queryset.filter(**filters)
-            except Exception:
-                raise ValidationError({
-                    'non_field_errors': _('Invalid filters provided')
-                })
-
         if all_filter and not helpers.str2bool(all_filter):
             raise ValidationError({
                 'non_field_errors': _('All filter must only be used with true')

src/backend/InvenTree/InvenTree/helpers.py

@@ -229,8 +229,15 @@ def getStaticUrl(filename):
     return os.path.join(STATIC_URL, str(filename))
 
 
-def TestIfImage(img):
+def TestIfImage(img) -> bool:
-    """Test if an image file is indeed an image."""
+    """Test if an image file is indeed an image.
+
+    Arguments:
+        img: A file-like object
+
+    Returns:
+        True if the file is a valid image, False otherwise
+    """
     try:
         Image.open(img).verify()
         return True

src/backend/InvenTree/InvenTree/test_api.py

@@ -278,8 +278,7 @@ class BulkDeleteTests(InvenTreeAPITestCase):
         response = self.delete(url, {}, expected_code=400)
 
         self.assertIn(
-            'List of items or filters must be provided for bulk operation',
+            'List of items must be provided for bulk operation', str(response.data)
-            str(response.data),
         )
 
         # DELETE with invalid 'items'
@@ -287,11 +286,6 @@ class BulkDeleteTests(InvenTreeAPITestCase):
 
         self.assertIn('Items must be provided as a list', str(response.data))
 
-        # DELETE with invalid 'filters'
-        response = self.delete(url, {'filters': [1, 2, 3]}, expected_code=400)
-
-        self.assertIn('Filters must be provided as a dict', str(response.data))
-
 
 class SearchTests(InvenTreeAPITestCase):
     """Unit tests for global search endpoint."""

src/backend/InvenTree/common/tests.py

@@ -1317,12 +1317,18 @@ class NotificationTest(InvenTreeAPITestCase):
 
         # Now, let's bulk delete all 'unread' notifications via the API,
         # but only associated with the logged in user
-        response = self.delete(url, {'filters': {'read': False}}, expected_code=200)
+        read_notifications = NotificationMessage.objects.filter(read=True)
+        response = self.delete(
+            url, {'items': [ntf.pk for ntf in read_notifications]}, expected_code=200
+        )
 
-        # Only 7 notifications should have been deleted,
+        # Only 3 notifications should have been deleted,
         # as the notifications associated with other users must remain untouched
-        self.assertEqual(NotificationMessage.objects.count(), 13)
+        self.assertEqual(NotificationMessage.objects.count(), 17)
-        self.assertEqual(NotificationMessage.objects.filter(user=self.user).count(), 3)
+        self.assertEqual(NotificationMessage.objects.filter(user=self.user).count(), 7)
+        self.assertEqual(
+            NotificationMessage.objects.filter(user=self.user, read=True).count(), 0
+        )
 
     def test_simple(self):
         """Test that a simple notification can be created."""

src/backend/InvenTree/part/test_api.py

@@ -74,7 +74,6 @@ class PartImageTestMixin:
                 {'image': img_file},
                 expected_code=200,
             )
-            print(response.data)
             image_name = response.data['image']
             self.assertTrue(image_name.startswith('/media/part_images/part_image'))
         return image_name
@@ -1838,7 +1837,7 @@ class PartDetailTests(PartImageTestMixin, PartAPITestBase):
 
         # Part should not have an image!
         with self.assertRaises(ValueError):
-            print(p.image.file)
+            _x = p.image.file
 
         # Try to upload a non-image file
         test_path = get_testfolder_dir() / 'dummy_image'

src/backend/InvenTree/report/templatetags/report.py

@@ -5,12 +5,16 @@ import logging
 import os
 from datetime import date, datetime
 from decimal import Decimal, InvalidOperation
+from io import BytesIO
+from pathlib import Path
 from typing import Any, Optional
 
 from django import template
 from django.apps.registry import apps
 from django.conf import settings
-from django.core.exceptions import ValidationError
+from django.contrib.staticfiles.storage import staticfiles_storage
+from django.core.exceptions import SuspiciousFileOperation, ValidationError
+from django.core.files.storage import default_storage
 from django.db.models import Model
 from django.db.models.query import QuerySet
 from django.utils.safestring import SafeString, mark_safe
@@ -145,6 +149,111 @@ def getkey(container: dict, key: str, backup_value: Optional[Any] = None) -> Any
     return container.get(key, backup_value)
 
 
+def media_file_exists(path: Path | str) -> bool:
+    """Check if a media file exists at the specified path.
+
+    Arguments:
+        path: The path to the media file, relative to the media storage root
+
+    Returns:
+        True if the file exists, False otherwise
+    """
+    if not path:
+        return False
+
+    try:
+        return default_storage.exists(str(path))
+    except SuspiciousFileOperation:
+        # Prevent path traversal attacks
+        raise ValidationError(_('Invalid media file path') + f": '{path}'")
+
+
+def static_file_exists(path: Path | str) -> bool:
+    """Check if a static file exists at the specified path.
+
+    Arguments:
+        path: The path to the static file, relative to the static storage root
+
+    Returns:
+        True if the file exists, False otherwise
+    """
+    if not path:
+        return False
+
+    try:
+        return staticfiles_storage.exists(str(path))
+    except SuspiciousFileOperation:
+        # Prevent path traversal attacks
+        raise ValidationError(_('Invalid static file path') + f": '{path}'")
+
+
+def get_static_file_contents(
+    path: Path | str, raise_error: bool = True
+) -> bytes | None:
+    """Return the contents of a static file.
+
+    Arguments:
+        path: The path to the static file, relative to the static storage root
+        raise_error: If True, raise an error if the file cannot be found (default = True)
+
+    Returns:
+        The contents of the static file, or None if the file cannot be found
+    """
+    if not path:
+        if raise_error:
+            raise ValueError('No media file specified')
+        else:
+            return None
+
+    if not staticfiles_storage.exists(path):
+        if raise_error:
+            raise FileNotFoundError(f'Static file does not exist: {path!s}')
+        else:
+            return None
+
+    with staticfiles_storage.open(str(path)) as f:
+        file_data = f.read()
+
+    return file_data
+
+
+def get_media_file_contents(path: Path | str, raise_error: bool = True) -> bytes | None:
+    """Return the fully qualified file path to an uploaded media file.
+
+    Arguments:
+        path: The path to the media file, relative to the media storage root
+        raise_error: If True, raise an error if the file cannot be found (default = True)
+
+    Returns:
+        The contents of the media file, or None if the file cannot be found
+
+    Raises:
+        FileNotFoundError: If the requested media file cannot be loaded
+        PermissionError: If the requested media file is outside of the media root
+        ValidationError: If the provided path is invalid
+
+    Notes:
+        - The resulting path is resolved against the media root directory
+    """
+    if not path:
+        if raise_error:
+            raise ValueError('No media file specified')
+        else:
+            return None
+
+    if not media_file_exists(path):
+        if raise_error:
+            raise FileNotFoundError(f'Media file does not exist: {path!s}')
+        else:
+            return None
+
+    # Load the file - and return the contents
+    with default_storage.open(str(path)) as f:
+        file_data = f.read()
+
+    return file_data
+
+
 @register.simple_tag()
 def asset(filename):
     """Return fully-qualified path for an upload report asset file.
@@ -159,18 +268,21 @@ def asset(filename):
         # Prepend an empty string to enforce 'stringiness'
         filename = '' + filename
 
-    # If in debug mode, return URL to the image, not a local file
+    # Remove any leading slash characters from the filename, to prevent path traversal attacks
-    debug_mode = get_global_setting('REPORT_DEBUG_MODE', cache=False)
+    filename = str(filename).lstrip('/\\')
 
-    # Test if the file actually exists
+    full_path = Path('report', 'assets', filename)
-    full_path = settings.MEDIA_ROOT.joinpath('report', 'assets', filename).resolve()
 
-    if not full_path.exists() or not full_path.is_file():
+    if not media_file_exists(full_path):
-        raise FileNotFoundError(_('Asset file does not exist') + f": '{filename}'")
+        raise FileNotFoundError(_('Asset file not found') + f": '{filename}'")
 
-    if debug_mode:
+    # In debug mode, return a web URL to the asset file (rather than a local file path)
-        return os.path.join(settings.MEDIA_URL, 'report', 'assets', filename)
+    if get_global_setting('REPORT_DEBUG_MODE', cache=False):
-    return f'file://{full_path}'
+        return str(Path(settings.MEDIA_URL, 'report', 'assets', filename))
+
+    storage_path = default_storage.path(str(full_path))
+
+    return f'file://{storage_path}'
 
 
 @register.simple_tag()
@@ -187,7 +299,7 @@ def uploaded_image(
     """Return raw image data from an 'uploaded' image.
 
     Arguments:
-        filename: The filename of the image relative to the MEDIA_ROOT directory
+        filename: The filename of the image relative to the media root directory
         replace_missing: Optionally return a placeholder image if the provided filename does not exist (default = True)
         replacement_file: The filename of the placeholder image (default = 'blank_image.png')
         validate: Optionally validate that the file is a valid image file
@@ -205,38 +317,43 @@ def uploaded_image(
         # Prepend an empty string to enforce 'stringiness'
         filename = '' + filename
 
+    # Strip out any leading slash characters from the filename, to prevent path traversal attacks
+    filename = str(filename).lstrip('/\\')
+
     # If in debug mode, return URL to the image, not a local file
     debug_mode = get_global_setting('REPORT_DEBUG_MODE', cache=False)
 
-    # Check if the file exists
+    # Load image data - this will check if the file exists
-    if not filename:
+    exists = bool(filename) and media_file_exists(filename)
-        exists = False
-    else:
-        try:
-            full_path = settings.MEDIA_ROOT.joinpath(filename).resolve()
-            exists = full_path.exists() and full_path.is_file()
-        except Exception:  # pragma: no cover
-            exists = False  # pragma: no cover
-
-    if exists and validate and not InvenTree.helpers.TestIfImage(full_path):
-        logger.warning("File '%s' is not a valid image", filename)
-        exists = False
 
     if not exists and not replace_missing:
         raise FileNotFoundError(_('Image file not found') + f": '{filename}'")
 
+    if exists:
+        img_data = get_media_file_contents(filename, raise_error=False)
+
+        # Check if the image data is valid
+        if (
+            img_data
+            and validate
+            and not InvenTree.helpers.TestIfImage(BytesIO(img_data))
+        ):
+            logger.warning("File '%s' is not a valid image", filename)
+            img_data = None
+            exists = False
+    else:
+        # Load the backup image from the static files directory
+        replacement_file_path = Path('img', replacement_file)
+        img_data = get_static_file_contents(replacement_file_path)
+
     if debug_mode:
         # In debug mode, return a web path (rather than an encoded image blob)
         if exists:
             return os.path.join(settings.MEDIA_URL, filename)
         return os.path.join(settings.STATIC_URL, 'img', replacement_file)
 
-    elif not exists:
+    if img_data:
-        full_path = settings.STATIC_ROOT.joinpath('img', replacement_file).resolve()
+        img = Image.open(BytesIO(img_data))
-
-    # Load the image, check that it is valid
-    if full_path.exists() and full_path.is_file():
-        img = Image.open(full_path)
     else:
         # A placeholder image showing that the image is missing
         img = Image.new('RGB', (64, 64), color='red')
@@ -288,22 +405,15 @@ def encode_svg_image(filename: str) -> str:
         # Prepend an empty string to enforce 'stringiness'
         filename = '' + filename
 
-    # Check if the file exists
+    # Remove any leading slash characters from the filename, to prevent path traversal attacks
+    filename = str(filename).lstrip('/\\')
+
     if not filename:
-        exists = False
+        raise FileNotFoundError(_('No image file specified'))
-    else:
-        try:
-            full_path = settings.MEDIA_ROOT.joinpath(filename).resolve()
-            exists = full_path.exists() and full_path.is_file()
-        except Exception:
-            exists = False
 
-    if not exists:
+    # Read out the file contents
-        raise FileNotFoundError(_('Image file not found') + f": '{filename}'")
+    # Note: This will check if the file exists, and raise an error if it does not
-
+    data = get_media_file_contents(filename)
-    # Read the file data
-    with open(full_path, 'rb') as f:
-        data = f.read()
 
     # Return the base64-encoded data
     return 'data:image/svg+xml;charset=utf-8;base64,' + base64.b64encode(data).decode(
@@ -323,8 +433,15 @@ def part_image(part: Part, preview: bool = False, thumbnail: bool = False, **kwa
     Raises:
         TypeError: If provided part is not a Part instance
     """
-    if type(part) is not Part:
+    if not part or not isinstance(part, Part):
         raise TypeError(_('part_image tag requires a Part instance'))
+
+    image_filename = InvenTree.helpers.image2name(part.image, preview, thumbnail)
+
+    if kwargs.get('check_exists'):
+        if not media_file_exists(image_filename):
+            raise FileNotFoundError(_('Image file not found') + f": '{image_filename}'")
+
     return uploaded_image(
         InvenTree.helpers.image2name(part.image, preview, thumbnail), **kwargs
     )

src/backend/InvenTree/report/test_tags.py

@@ -14,7 +14,6 @@ from djmoney.money import Money
 from PIL import Image
 
 from common.models import InvenTreeSetting, Parameter, ParameterTemplate
-from InvenTree.config import get_testfolder_dir
 from InvenTree.unit_test import InvenTreeTestCase
 from part.models import Part  # TODO fix import: PartParameter, PartParameterTemplate
 from part.test_api import PartImageTestMixin
@@ -81,7 +80,27 @@ class ReportTagTest(PartImageTestMixin, InvenTreeTestCase):
 
         self.debug_mode(False)
         asset = report_tags.asset('test.txt')
-        self.assertEqual(asset, f'file://{asset_dir}/test.txt')
+
+        # Test for attempted path traversal
+        with self.assertRaises(ValidationError):
+            report_tags.asset('../../../report/assets/test.txt')
+
+    def test_file_access(self):
+        """Tests for media and static file access."""
+        for fn in [None, '', '@@@@@@', 'fake_file.txt']:
+            self.assertFalse(report_tags.media_file_exists(fn))
+            self.assertFalse(report_tags.static_file_exists(fn))
+
+        with self.assertRaises(FileNotFoundError):
+            report_tags.get_media_file_contents('dummy_file.txt')
+
+        with self.assertRaises(ValueError):
+            report_tags.get_static_file_contents(None)
+
+        # Try again, without throwing an error
+        self.assertIsNone(
+            report_tags.get_media_file_contents('dummy_file.txt', raise_error=False)
+        )
 
     def test_uploaded_image(self):
         """Tests for retrieving uploaded images."""
@@ -148,6 +167,10 @@ class ReportTagTest(PartImageTestMixin, InvenTreeTestCase):
         )
         self.assertTrue(img.startswith('data:image/png;charset=utf-8;base64,'))
 
+        # Attempted path traversal
+        with self.assertRaises(ValidationError):
+            report_tags.uploaded_image('../../../part/images/test.jpg')
+
     def test_part_image(self):
         """Unit tests for the 'part_image' tag."""
         with self.assertRaises(TypeError):
@@ -157,8 +180,10 @@ class ReportTagTest(PartImageTestMixin, InvenTreeTestCase):
         self.create_test_image()
         obj.refresh_from_db()
 
-        report_tags.part_image(obj, preview=True)
+        r = report_tags.part_image(obj, preview=True)
-        report_tags.part_image(obj, thumbnail=True)
+        self.assertIn('data:image/png;charset=utf-8;base64,', r)
+        r = report_tags.part_image(obj, thumbnail=True)
+        self.assertIn('data:image/png;charset=utf-8;base64,', r)
 
     def test_company_image(self):
         """Unit tests for the 'company_image' tag."""
@@ -392,12 +417,16 @@ class ReportTagTest(PartImageTestMixin, InvenTreeTestCase):
     def test_encode_svg_image(self):
         """Test the encode_svg_image template tag."""
         # Generate smallest possible SVG for testing
-        svg_path = get_testfolder_dir() / 'part_image_123abc.png'
+        # Store it in the media directory
+
+        img_path = 'part_image_123abc.png'
+        svg_path = settings.MEDIA_ROOT / img_path
+
         with open(svg_path, 'w', encoding='utf8') as f:
             f.write('<svg xmlns="http://www.w3.org/2000/svg>')
 
         # Test with a valid SVG file
-        svg = report_tags.encode_svg_image(svg_path)
+        svg = report_tags.encode_svg_image(img_path)
         self.assertTrue(svg.startswith('data:image/svg+xml;charset=utf-8;base64,'))
         self.assertIn('svg', svg)
         self.assertEqual(

src/backend/InvenTree/report/tests.py

@@ -571,7 +571,9 @@ class TestReportTest(PrintTestMixins, ReportTest):
 
     def test_mdl_salesorder(self):
         """Test the SalesOrder model."""
-        self.run_print_test(SalesOrder, 'salesorder', label=False)
+        for enabled in [True, False]:
+            set_global_setting('REPORT_DEBUG_MODE', enabled)
+            self.run_print_test(SalesOrder, 'salesorder', label=False)
 
 
 class AdminTest(AdminTestCase):

src/backend/InvenTree/stock/test_api.py

@@ -2407,17 +2407,7 @@ class StockTestResultTest(StockAPITestCase):
         self.delete(url, {}, expected_code=400)
 
         # Now, let's delete all the newly created items with a single API request
-        # However, we will provide incorrect filters
+        response = self.delete(url, {'items': tests}, expected_code=200)
-        response = self.delete(
-            url, {'items': tests, 'filters': {'stock_item': 10}}, expected_code=400
-        )
-
-        self.assertEqual(StockItemTestResult.objects.count(), n + 50)
-
-        # Try again, but with the correct filters this time
-        response = self.delete(
-            url, {'items': tests, 'filters': {'stock_item': 1}}, expected_code=200
-        )
 
         self.assertEqual(StockItemTestResult.objects.count(), n)
 

src/frontend/src/components/editors/TemplateEditor/PdfPreview/PdfPreview.tsx

@@ -112,9 +112,9 @@ export const PdfPreviewComponent: PreviewAreaComponent = forwardRef(
             style={{
               display: 'flex',
               justifyContent: 'center',
-              alignItems: 'center',
               height: '100%',
-              width: '100%'
+              width: '100%',
+              paddingTop: '50px'
             }}
           >
             <Trans>Preview not available, click "Reload Preview".</Trans>