Validate and sign mime types and file extensions for file uploads (#2860)

* Validate and sign mime types for file uploads

* add checks for file extensions

Author: sainak

care/emr/models/file_upload.py

@@ -7,6 +7,7 @@
 from care.emr.utils.file_manager import S3FilesManager
 from care.users.models import User
 from care.utils.csp.config import BucketType
+from care.utils.models.validators import parse_file_extension
 
 
 class FileUpload(EMRBaseModel):
@@ -32,9 +33,8 @@ class FileUpload(EMRBaseModel):
     files_manager = S3FilesManager(BucketType.PATIENT)
 
     def get_extension(self):
-        # TODO: improve this logic to handle files with multiple extensions
-        parts = self.internal_name.split(".")
-        return f".{parts[-1]}" if len(parts) > 1 else ""
+        extensions = parse_file_extension(self.internal_name)
+        return f".{".".join(extensions)}" if extensions else ""
 
     def save(self, *args, **kwargs):
         """

care/emr/resources/file_upload/spec.py

@@ -1,11 +1,14 @@
 import datetime
 from enum import Enum
 
-from pydantic import UUID4
+from django.conf import settings
+from django.core.exceptions import ValidationError
+from pydantic import UUID4, field_validator
 
 from care.emr.models import FileUpload
 from care.emr.resources.base import EMRResource
 from care.emr.resources.user.spec import UserSpec
+from care.utils.models.validators import file_name_validator
 
 
 class FileTypeChoices(str, Enum):
@@ -37,11 +40,33 @@ class FileUploadCreateSpec(FileUploadBaseSpec):
     file_type: FileTypeChoices
     file_category: FileCategoryChoices
     associating_id: str
+    mime_type: str
 
     def perform_extra_deserialization(self, is_update, obj):
         # Authz Performed in the request
         obj._just_created = True  # noqa SLF001
         obj.internal_name = self.original_name
+        obj.meta["mime_type"] = self.mime_type
+
+    @field_validator("mime_type")
+    @classmethod
+    def validate_mime_type(cls, mime_type: str):
+        if mime_type not in settings.ALLOWED_MIME_TYPES:
+            err = "Invalid mime type"
+            raise ValueError(err)
+        return mime_type
+
+    @field_validator("original_name")
+    @classmethod
+    def validate_original_name(cls, original_name: str):
+        if not original_name:
+            err = "File name cannot be empty"
+            raise ValueError(err)
+        try:
+            file_name_validator(original_name)
+        except ValidationError as e:
+            raise ValueError(e.message) from e
+        return original_name
 
 
 class FileUploadListSpec(FileUploadBaseSpec):
@@ -56,11 +81,13 @@ class FileUploadListSpec(FileUploadBaseSpec):
     created_date: datetime.datetime
     extension: str
     uploaded_by: dict
+    mime_type: str
 
     @classmethod
     def perform_extra_serialization(cls, mapping, obj):
         mapping["id"] = obj.external_id
         mapping["extension"] = obj.get_extension()
+        mapping["mime_type"] = obj.meta.get("mime_type")
         if obj.created_by:
             mapping["uploaded_by"] = UserSpec.serialize(obj.created_by)
 

care/emr/utils/file_manager.py

@@ -22,8 +22,10 @@ def signed_url(self, file_obj, duration=60 * 60, mime_type=None):
             "Bucket": bucket_name,
             "Key": f"{file_obj.file_type}/{file_obj.internal_name}",
         }
-        if mime_type:
-            params["ContentType"] = mime_type
+
+        _mime_type = file_obj.meta.get("mime_type") or mime_type
+        if _mime_type:
+            params["ContentType"] = _mime_type
         return s3.generate_presigned_url(
             "put_object",
             Params=params,

care/utils/models/validators.py

@@ -1,8 +1,10 @@
 import re
 from collections.abc import Iterable
 from fractions import Fraction
+from pathlib import Path
 
 import jsonschema
+from django.conf import settings
 from django.core import validators
 from django.core.exceptions import ValidationError
 from django.core.files.uploadedfile import UploadedFile
@@ -348,3 +350,89 @@ def _humanize_bytes(self, size: int) -> str:
 custom_image_extension_validator = validators.FileExtensionValidator(
     allowed_extensions=["jpg", "jpeg", "png"]
 )
+
+
+def parse_file_extension(file_name: str, max_extensions: int = 1) -> list[str]:
+    """
+    Extract up to `max_extensions` file extensions.
+
+    - "file.tar.gz" -> ['tar', 'gz']
+    - "file.tar.bz2.xz" (max 3) -> ['tar', 'bz2', 'xz']
+    - "file" -> []
+
+    Returns a list of extensions in lowercase (without the dot).
+    """
+    path = Path(file_name)
+    extensions = [ext[1:].lower() for ext in path.suffixes]  # Remove leading dots
+
+    return extensions[-max_extensions:]  # Keep only the last `max_extensions`
+
+
+class FileNameValidator:
+    """
+    This validator is used to validate the file name length and extension.
+    - File name length should not exceed `max_length` characters.
+    - File name should not start with a dot.
+    - File outermost extension should not be in `blocked_extensions` if provided.
+    - File outermost extension should be in `allowed_extensions` if provided.
+    """
+
+    def __init__(
+        self,
+        max_length: int = 255,
+        max_extensions: int = 1,
+        allowed_extensions: set[str] | None = None,
+        blocked_extensions: set[str] | None = None,
+    ):
+        self.max_length = max_length
+        self.max_extensions = max_extensions
+        self.allowed_extensions = allowed_extensions or set()
+        self.blocked_extensions = blocked_extensions or set()
+
+    def __call__(self, file_name: str):
+        if len(file_name) > self.max_length:
+            raise ValidationError(
+                _("File name cannot exceed %(max_length)d characters.")
+                % {"max_length": self.max_length}
+            )
+
+        if file_name.startswith("."):
+            raise ValidationError(
+                _("File name cannot start with a dot."),
+            )
+
+        extensions = parse_file_extension(file_name, self.max_extensions)
+        if not extensions:
+            raise ValidationError(_("Invalid file extension."))
+
+        extension = extensions[-1].lower()
+        if self.blocked_extensions and extension in self.blocked_extensions:
+            raise ValidationError(
+                _(
+                    "File extension not allowed. Blocked extensions are: %(blocked_extensions)s"
+                )
+                % {"blocked_extensions": ", ".join(self.blocked_extensions)},
+            )
+
+        if self.allowed_extensions and extension not in self.allowed_extensions:
+            raise ValidationError(
+                _(
+                    "File extension not allowed. Allowed extensions are: %(allowed_extensions)s"
+                )
+                % {"allowed_extensions": ", ".join(self.allowed_extensions)},
+            )
+
+    def __eq__(self, other):
+        return (
+            isinstance(other, FileNameValidator)
+            and self.max_length == other.max_length
+            and self.max_extensions == other.max_extensions
+            and self.allowed_extensions == other.allowed_extensions
+            and self.blocked_extensions == other.blocked_extensions
+        )
+
+
+file_name_validator = FileNameValidator(
+    allowed_extensions=settings.ALLOWED_FILE_EXTENSIONS,
+    blocked_extensions=settings.BLOCKED_FILE_EXTENSIONS,
+)

config/settings/base.py

@@ -569,46 +569,116 @@
     ),
 )
 
-ALLOWED_MIME_TYPES = env.list(
-    "ALLOWED_MIME_TYPES",
-    default=[
-        # Images
-        "image/jpeg",
-        "image/png",
-        "image/gif",
-        "image/bmp",
-        "image/webp",
-        "image/svg+xml",
-        # Videos
-        "video/mp4",
-        "video/mpeg",
-        "video/x-msvideo",
-        "video/quicktime",
-        "video/x-ms-wmv",
-        "video/x-flv",
-        "video/webm",
-        # Audio
-        "audio/mpeg",
-        "audio/wav",
-        "audio/aac",
-        "audio/ogg",
-        "audio/midi",
-        "audio/x-midi",
-        "audio/webm",
-        "audio/mp4",
-        # Documents
-        "text/plain",
-        "text/csv",
-        "application/rtf",
-        "application/msword",
-        "application/vnd.oasis.opendocument.text",
-        "application/pdf",
-        "application/vnd.ms-excel",
-        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
-        "application/vnd.oasis.opendocument.spreadsheet",
-    ],
+ALLOWED_MIME_TYPES = set(
+    env.list(
+        "ALLOWED_MIME_TYPES",
+        default=[
+            # Images
+            "image/jpeg",
+            "image/png",
+            "image/gif",
+            "image/bmp",
+            "image/webp",
+            "image/svg+xml",
+            # Videos
+            "video/mp4",
+            "video/mpeg",
+            "video/x-msvideo",
+            "video/quicktime",
+            "video/x-ms-wmv",
+            "video/x-flv",
+            "video/webm",
+            # Audio
+            "audio/mpeg",
+            "audio/wav",
+            "audio/aac",
+            "audio/ogg",
+            "audio/midi",
+            "audio/x-midi",
+            "audio/webm",
+            "audio/mp4",
+            # Documents
+            "text/plain",
+            "text/csv",
+            "application/rtf",
+            "application/msword",
+            "application/vnd.oasis.opendocument.text",
+            "application/pdf",
+            "application/vnd.ms-excel",
+            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+            "application/vnd.oasis.opendocument.spreadsheet",
+        ],
+    )
+)
+
+ALLOWED_FILE_EXTENSIONS = set(
+    env.list(
+        "ALLOWED_FILE_EXTENSIONS",
+        default=[
+            # Images
+            "jpg",
+            "jpeg",
+            "png",
+            "gif",
+            "bmp",
+            "webp",
+            "svg",
+            # Videos
+            "mp4",
+            "mpeg",
+            "avi",
+            "mov",
+            "wmv",
+            "flv",
+            "webm",
+            # Audio
+            "mp3",
+            "wav",
+            "aac",
+            "ogg",
+            "midi",
+            "mid",
+            "m4a",
+            # Documents
+            "txt",
+            "csv",
+            "rtf",
+            "doc",
+            "odt",
+            "pdf",
+            "xls",
+            "xlsx",
+            "ods",
+        ],
+    )
 )
 
+BLOCKED_FILE_EXTENSIONS = set(
+    env.list(
+        "BLOCKED_FILE_EXTENSIONS",
+        default=[
+            # Executable Files
+            "exe",
+            "dll",
+            "msi",
+            "msp",
+            "mst",
+            "com",
+            "scr",
+            "sys",
+            "pif",
+            # Registry Files
+            "reg",
+            # Script Files
+            "bat",
+            "cmd",
+            "wsf",
+            "sh",
+        ],
+    )
+)
+
+
 FACILITY_S3_BUCKET = env("FACILITY_S3_BUCKET", default="")
 FACILITY_S3_REGION = env("FACILITY_S3_REGION_CODE", default=BUCKET_REGION)
 FACILITY_S3_KEY = env("FACILITY_S3_KEY", default=BUCKET_KEY)