Add FileFieldType

This field type takes the string value from the CSV as a reference to
a file, and stores this file's contents into the target FileField.

It is taken care that the path references a file below the base
directory where the ZIPfile was extracted.

aleksis/apps/csv_import/field_types.py

+import os
 import re
 from typing import Any, Callable, Optional, Sequence, Tuple, Type, Union
 from uuid import uuid4
 
 from django.apps import apps
+from django.core.files import File
 from django.db.models import Model
 from django.utils.translation import gettext as _
 
@@ -74,8 +76,9 @@ class FieldType:
         return model in cls.models
 
     @classmethod
-    def prepare(cls, school_term: SchoolTerm):
+    def prepare(cls, school_term: SchoolTerm, base_path: str):
         cls.school_term = school_term
+        cls.base_path = os.path.realpath(base_path)
 
 
 class MatchFieldType(FieldType):
@@ -504,3 +507,33 @@ class ChildByUniqueReference(ProcessFieldType):
     def process(self, instance: Model, value):
         child = Person.objects.get(import_ref_csv=value)
         instance.children.add(child)
+
+
+@field_type_registry.register
+class FileFieldType(ProcessFieldType):
+    """Field type that stores a referenced file on a file field."""
+
+    def process(self, instance: Model, value: str):
+        # Test path for unwanted path traversal
+        abs_path = os.path.realpath(value)
+        if not self.base_path == os.path.commonpath((self.base_path, abs_path)):
+            raise ValueError(f"Disallowed path traversal importing file from {value}")
+
+        # Get target FileField and save content
+        field_name = self.get_db_field()
+        file_field = getattr(instance, field_name)
+        file_field.save(os.path.basename(abs_path), File(open(abs_path, "rb")))
+
+        instance.save()
+
+
+@field_type_registry.register
+class AvatarFieldType(FileFieldType):
+    name = "avatar"
+    db_field = "avatar"
+
+
+@field_type_registry.register
+class PhotoFieldType(FileFieldType):
+    name = "photo"
+    db_field = "photo"

aleksis/apps/csv_import/util/process.py

@@ -33,39 +33,39 @@ def import_csv(
     import_job: ImportJob,
     recorder: Optional[ProgressRecorder] = None,
 ) -> None:
-    template = import_job.template
-    model = template.content_type.model_class()
-    school_term = import_job.school_term
+    with TemporaryDirectory() as temp_dir:
+        template = import_job.template
+        model = template.content_type.model_class()
+        school_term = import_job.school_term
 
-    data_types = {}
-    cols = []
-    cols_for_multiple_fields = {}
-    converters = {}
-    match_field_types = []
-    for field in template.fields.all():
-        field_type = field.field_type_class
-        column_name = field_type.get_column_name()
+        data_types = {}
+        cols = []
+        cols_for_multiple_fields = {}
+        converters = {}
+        match_field_types = []
+        for field in template.fields.all():
+            field_type = field.field_type_class
+            column_name = field_type.get_column_name()
 
-        # Get column header
-        if issubclass(field_type, MultipleValuesFieldType):
-            cols_for_multiple_fields.setdefault(field_type, [])
-            cols_for_multiple_fields[field_type].append(column_name)
+            # Get column header
+            if issubclass(field_type, MultipleValuesFieldType):
+                cols_for_multiple_fields.setdefault(field_type, [])
+                cols_for_multiple_fields[field_type].append(column_name)
 
-        # Get data type
-        data_types[column_name] = field_type.get_data_type()
+            # Get data type
+            data_types[column_name] = field_type.get_data_type()
 
-        if field_type.get_converter():
-            converters[column_name] = field_type.get_converter()
+            if field_type.get_converter():
+                converters[column_name] = field_type.get_converter()
 
-        if issubclass(field_type, MatchFieldType):
-            match_field_types.append((field_type.priority, field_type))
+            if issubclass(field_type, MatchFieldType):
+                match_field_types.append((field_type.priority, field_type))
 
-        # Prepare field type for import
-        field_type.prepare(school_term)
+            # Prepare field type for import
+            field_type.prepare(school_term, temp_dir)
 
-        cols.append(column_name)
+            cols.append(column_name)
 
-    with TemporaryDirectory() as temp_dir:
         # Determine whether the data file is a plain CSV or an archive
         if import_job.data_file.name.endswith(".zip"):
             # Unpack to temporary directory