Merge branch 'master' into TP2000-1327--upgrade-python-3.12

.copilot/config.yml

@@ -0,0 +1,4 @@
+repository: tariff-application/tamato
+builder:
+  name: paketobuildpacks/builder-jammy-full
+  version: 0.3.339

.copilot/image_build_run.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+set -e
+
+# Add commands below to run inside the container after all the other buildpacks have been applied

.copilot/phases/build.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+set -e

.copilot/phases/install.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+set -e

.copilot/phases/post_build.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+set -e

.copilot/phases/pre_build.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+set -e

common/tests/test_validators.py

@@ -1,4 +1,5 @@
 import pytest
+from django.core.exceptions import ValidationError
 
 from common.tests.util import check_validator
 from common.validators import AlphanumericValidator
@@ -8,6 +9,8 @@
 from common.validators import NumericValidator
 from common.validators import PasswordPolicyValidator
 from common.validators import SymbolValidator
+from common.validators import validate_filename
+from common.validators import validate_filepath
 
 
 @pytest.mark.parametrize(
@@ -116,3 +119,63 @@ def test_numeric_validator(value, expected_valid):
 def test_password_policy_validator(value, expected_valid):
     validator = PasswordPolicyValidator()
     check_validator(validator.validate, value, expected_valid)
+
+
+@pytest.mark.parametrize(
+    "filename",
+    [
+        "TGB123",
+        "TGB123-v2.xml",
+        "TGB123_v3.xml",
+    ],
+)
+def test_validate_filename_valid_name(filename):
+    try:
+        validate_filename(filename)
+    except ValidationError:
+        pytest.fail(
+            f'Unexpected ValidationError exception raised for filename "{filename}"',
+        )
+
+
+@pytest.mark.parametrize(
+    "filename",
+    [
+        "../dotdot/separator",
+        "$p£(!@lc|-|ars).txt",
+        "test.html%00.xml",
+    ],
+)
+def test_validate_filename_raises_exception(filename):
+    with pytest.raises(ValidationError):
+        validate_filename(filename)
+
+
+@pytest.mark.parametrize(
+    ("source", "base_path"),
+    [
+        ("common", ""),
+        ("common/tests/", "common/"),
+        ("common/tests/test_files", "common/"),
+        ("common/tests/test_files/dtd.xml", "common/tests/test_files"),
+    ],
+)
+def test_validate_filepath_valid_path(source, base_path):
+    try:
+        validate_filepath(source, base_path)
+    except ValidationError:
+        pytest.fail(
+            f'Unexpected ValidationError exception raised for source "{source}" and base_path "{base_path}"',
+        )
+
+
+@pytest.mark.parametrize(
+    ("source", "base_path"),
+    [
+        ("../../etc/passwd", ""),
+        ("common/tests/../../../outside", "common/"),
+    ],
+)
+def test_validate_filepath_raises_exception(source, base_path):
+    with pytest.raises(ValidationError):
+        validate_filepath(source, base_path)

common/util.py

@@ -8,7 +8,9 @@
 from datetime import timedelta
 from functools import lru_cache
 from functools import partial
+from pathlib import Path
 from platform import python_version_tuple
+from typing import IO
 from typing import Any
 from typing import Dict
 from typing import Optional
@@ -541,7 +543,7 @@ def check_docinfo(elementtree, forbid_dtd=False):
             raise DTDForbidden(docinfo.doctype, docinfo.system_url, docinfo.public_id)
 
 
-def parse_xml(source, forbid_dtd=True):
+def parse_xml(source: Union[str, Path, IO], forbid_dtd=True):
     parser = etree.XMLParser(resolve_entities=False)
     elementtree = etree.parse(source, parser)
     check_docinfo(elementtree, forbid_dtd=forbid_dtd)

common/validators.py

@@ -1,9 +1,16 @@
 """Common validators."""
 
+import os
+from pathlib import Path
+from typing import IO
+from typing import Union
+
+from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.core.validators import RegexValidator
 from django.db import models
 from django.utils.deconstruct import deconstructible
+from werkzeug.utils import secure_filename
 
 
 @deconstructible
@@ -127,3 +134,41 @@ def get_help_text(self):
     "tbody",
     "td",
 ]
+
+
+def validate_filename(filename: str) -> None:
+    """
+    Validates that `filename` only includes alphanumeric characters and special
+    characters such as spaces, hyphens and underscores.
+
+    Raises a `ValidationError` if `filename` includes non-permitted characters.
+    """
+
+    # filename might include spaces which secure_filename normalises to underscores
+    if filename.replace(" ", "_") != secure_filename(filename):
+        raise ValidationError(
+            f"File name must only include alphanumeric characters and special characters such as spaces, hyphens and underscores.",
+        )
+
+
+def validate_filepath(source: Union[str, Path, IO], base_path: str = "") -> None:
+    """
+    Validates that `source` normalises to an absolute path located within
+    `base_path` directory (`settings.BASE_DIR` by default).
+
+    Raises a `ValidationError` if the resulting path would be located outside the expected base path.
+    """
+
+    if isinstance(source, str):
+        path = source
+    elif hasattr(source, "name"):
+        path = source.name
+    else:
+        raise ValueError(f"Expected str, Path or File-like object, not {type(source)}")
+
+    if not base_path:
+        base_path = settings.BASE_DIR
+
+    full_path = os.path.normpath(os.path.join(base_path, path))
+    if not full_path.startswith(base_path):
+        raise ValidationError("Invalid file path: {path}")

common/views.py

@@ -14,6 +14,7 @@
 import kombu.exceptions
 from botocore.exceptions import ClientError
 from botocore.exceptions import EndpointConnectionError
+from dbt_copilot_python.utility import is_copilot
 from django.conf import settings
 from django.contrib.auth.mixins import LoginRequiredMixin
 from django.contrib.auth.mixins import PermissionRequiredMixin
@@ -302,13 +303,20 @@ def check_celery_broker(self) -> Tuple[str, int]:
 
     def check_s3(self) -> Tuple[str, int]:
         try:
-            client = boto3.client(
-                "s3",
-                aws_access_key_id=settings.HMRC_PACKAGING_S3_ACCESS_KEY_ID,
-                aws_secret_access_key=settings.HMRC_PACKAGING_S3_SECRET_ACCESS_KEY,
-                endpoint_url=settings.S3_ENDPOINT_URL,
-                region_name=settings.HMRC_PACKAGING_S3_REGION_NAME,
-            )
+            if is_copilot():
+                client = boto3.client(
+                    "s3",
+                    endpoint_url=settings.S3_ENDPOINT_URL,
+                    region_name=settings.HMRC_PACKAGING_S3_REGION_NAME,
+                )
+            else:
+                client = boto3.client(
+                    "s3",
+                    aws_access_key_id=settings.HMRC_PACKAGING_S3_ACCESS_KEY_ID,
+                    aws_secret_access_key=settings.HMRC_PACKAGING_S3_SECRET_ACCESS_KEY,
+                    endpoint_url=settings.S3_ENDPOINT_URL,
+                    region_name=settings.HMRC_PACKAGING_S3_REGION_NAME,
+                )
             client.head_bucket(Bucket=settings.HMRC_PACKAGING_STORAGE_BUCKET_NAME)
             return "OK", 200
         except (ClientError, EndpointConnectionError):

importer/forms.py

@@ -19,6 +19,8 @@
 
 from common.util import get_mime_type
 from common.util import parse_xml
+from common.validators import validate_filename
+from common.validators import validate_filepath
 from importer.chunker import chunk_taric
 from importer.management.commands.run_import_batch import run_batch
 from importer.models import ImportBatch
@@ -74,13 +76,8 @@ def clean_taric_file(self):
         if mime_type not in ["text/xml", "application/xml"]:
             raise ValidationError("The selected file must be XML")
 
-        secure_file_name = secure_filename(uploaded_taric_file.name)
-        if uploaded_taric_file.name.replace(" ", "_") == secure_file_name:
-            uploaded_taric_file.name == secure_file_name
-        else:
-            raise ValidationError(
-                "File name must only include alphanumeric characters and special characters such as spaces, hyphens and underscores.",
-            )
+        validate_filename(uploaded_taric_file.name)
+        validate_filepath(uploaded_taric_file)
 
         try:
             xml_file = parse_xml(uploaded_taric_file)
@@ -154,13 +151,8 @@ def clean_taric_file(self):
         if mime_type not in ["text/xml", "application/xml"]:
             raise ValidationError("The selected file must be XML")
 
-        secure_file_name = secure_filename(uploaded_taric_file.name)
-        if uploaded_taric_file.name.replace(" ", "_") == secure_file_name:
-            uploaded_taric_file.name == secure_file_name
-        else:
-            raise ValidationError(
-                "File name must only include alphanumeric characters and special characters such as spaces, hyphens and underscores.",
-            )
+        validate_filename(uploaded_taric_file.name)
+        validate_filepath(uploaded_taric_file)
 
         try:
             xml_file = parse_xml(uploaded_taric_file)
@@ -352,19 +344,20 @@ def save(self):
         make sense to use commit=False. process_file() should be moved into the
         view if this (common) behaviour becomes required.
         """
-        file_name = os.path.splitext(self.cleaned_data["name"])[0]
-        description = f"TARIC {file_name} commodity code changes"
+        taric_filename = secure_filename(self.cleaned_data["name"])
+        file_id = os.path.splitext(taric_filename)[0]
+        description = f"TARIC {file_id} commodity code changes"
 
         import_batch = ImportBatch(
             author=self.request.user,
-            name=self.cleaned_data["name"],
+            name=taric_filename,
             goods_import=True,
         )
 
         self.files["taric_file"].seek(0, os.SEEK_SET)
         import_batch.taric_file.save(
-            self.files["taric_file"].name,
-            ContentFile(self.files["taric_file"].read()),
+            name=taric_filename,
+            content=ContentFile(self.files["taric_file"].read()),
         )
 
         import_batch.save()

measures/forms.py

@@ -1379,10 +1379,11 @@ def serializable_data(self, remove_key_prefix: str = "") -> Dict:
         # Perculiarly, serializable data in this form keeps its prefix.
         return super().serializable_data()
 
-    def deserialize_init_kwargs(self, form_kwargs: Dict) -> Dict:
+    @classmethod
+    def deserialize_init_kwargs(cls, form_kwargs: Dict) -> Dict:
         # Perculiarly, this Form requires a prefix of "geographical_area".
         return {
-            "prefix": self.prefix,
+            "prefix": "geographical_area",
         }