allow reprocessing multiple files/reports at once

.env.example

@@ -15,6 +15,7 @@ CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP=
 # stixify settings
 MAX_PAGE_SIZE=
 DEFAULT_PAGE_SIZE=
+REPROCESS_MAX_FAILED_PROCESSES=10
 # stix2arango settings
 ARANGODB_HOST_URL=
 ARANGODB_USERNAME=

.env.markdown

@@ -36,6 +36,8 @@ These define how the API behaves.
 	* This is the maximum number of results the API will ever return before pagination
 * `DEFAULT_PAGE_SIZE`: `50`
 	* The default page size of result returned by the API
+* `REPROCESS_MAX_FAILED_PROCESSES`: default `10`
+	* Stop a detached File reprocessing job before starting another File after this many failures.
 
 ## ArangoDB settings
 

stixify/settings.py

@@ -280,3 +280,4 @@
 CLASSIFIER_MODEL_PATH = os.getenv("CLASSIFIER_MODEL_PATH", os.path.join(BASE_DIR, "classifier_hdbscan.joblib"))
 CLASSIFIER_CONCURRENCY = int(os.getenv("CLASSIFIER_CONCURRENCY", 12))
 CREATE_EMBEDDING_INCLUDE_NON_INCIDENT = bool(os.getenv("CREATE_EMBEDDING_INCLUDE_NON_INCIDENT", False))
+REPROCESS_MAX_FAILED_PROCESSES = int(os.getenv("REPROCESS_MAX_FAILED_PROCESSES", 10))

stixify/web/migrations/0027_file_added_file_updated.py

@@ -0,0 +1,48 @@
+from django.db import migrations, models
+
+
+def populate_file_timestamps(apps, schema_editor):
+    File = apps.get_model("stixify_core", "File")
+    Job = apps.get_model("stixify_core", "Job")
+    first_job_times = dict(
+        Job.objects.filter(file_id__isnull=False)
+        .order_by("file_id", "run_datetime", "id")
+        .distinct("file_id")
+        .values_list("file_id", "run_datetime")
+    )
+    files = list(File.objects.all())
+    for file in files:
+        timestamp = first_job_times.get(file.pk, file.modified)
+        file.added = timestamp
+        file.updated = timestamp
+    File.objects.bulk_update(files, ["added", "updated"])
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("stixify_core", "0026_file_pap_level"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="file",
+            name="added",
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.AddField(
+            model_name="file",
+            name="updated",
+            field=models.DateTimeField(null=True),
+        ),
+        migrations.RunPython(populate_file_timestamps, migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name="file",
+            name="added",
+            field=models.DateTimeField(auto_now_add=True),
+        ),
+        migrations.AlterField(
+            model_name="file",
+            name="updated",
+            field=models.DateTimeField(auto_now=True),
+        ),
+    ]

stixify/web/migrations/0028_rename_reprocess_job_type.py

@@ -0,0 +1,36 @@
+from django.db import migrations, models
+
+
+def rename_reprocess_job_type(apps, schema_editor):
+    Job = apps.get_model("stixify_core", "Job")
+    Job.objects.filter(type="reprocess-posts").update(type="reprocess-files")
+
+
+def restore_reprocess_job_type(apps, schema_editor):
+    Job = apps.get_model("stixify_core", "Job")
+    Job.objects.filter(type="reprocess-files").update(type="reprocess-posts")
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("stixify_core", "0027_file_added_file_updated"),
+    ]
+
+    operations = [
+        migrations.RunPython(rename_reprocess_job_type, restore_reprocess_job_type),
+        migrations.AlterField(
+            model_name="job",
+            name="type",
+            field=models.CharField(
+                choices=[
+                    ("import-file", "Import File"),
+                    ("reprocess-files", "Reprocess Files"),
+                    ("sync-knowledgebase", "Sync Knowledgebase"),
+                    ("build-clusters", "Build Clusters"),
+                    ("build-embeddings", "Build Embeddings"),
+                ],
+                default="import-file",
+                max_length=64,
+            ),
+        ),
+    ]

stixify/web/models.py

@@ -118,6 +118,8 @@ def validate_file(file: InMemoryUploadedFile, mode: str):
 
 class File(CommonSTIXProps):
     id = models.UUIDField(unique=True, max_length=64, primary_key=True, default=uuid.uuid4)
+    added = models.DateTimeField(auto_now_add=True)
+    updated = models.DateTimeField(auto_now=True)
     file = models.FileField(max_length=1024, upload_to=upload_to_func)
     identity = models.ForeignKey(Identity, on_delete=models.CASCADE, default=None)
     profile = models.ForeignKey(Profile, on_delete=models.PROTECT)
@@ -174,6 +176,12 @@ def report_id(self, value):
     def clean(self) -> None:
         validate_file(self.file, self.mode)
         return super().clean()
+
+    def save(self, *args, **kwargs):
+        update_fields = kwargs.get("update_fields")
+        if update_fields is not None:
+            kwargs["update_fields"] = set(update_fields) | {"updated"}
+        return super().save(*args, **kwargs)
 
     def __str__(self) -> str:
         return f"File(id={self.id})"
@@ -298,7 +306,7 @@ class JobState(models.TextChoices):
 
 class JobType(models.TextChoices):
     IMPORT_FILE = "import-file"
-    REPROCESS_POSTS = "reprocess-posts"
+    REPROCESS_FILES = "reprocess-files"
     SYNC_KNOWLEDGEBASE = "sync-knowledgebase"
     BUILD_CLUSTERS = "build-clusters"
     BUILD_EMBEDDINGS = "build-embeddings"

stixify/web/serializers.py

@@ -144,7 +144,7 @@ class FileSerializer(serializers.ModelSerializer):
     class Meta:
         model = File
         exclude = ['profile', "markdown_file", "txt2stix_data", "pdf_file", "identity", "embedding"]
-        read_only_fields = []
+        read_only_fields = ["added", "updated"]
 
     def validate(self, attrs):
         return super().validate(attrs)
@@ -240,6 +240,50 @@ def validate(self, attrs):
         return super().validate(attrs)
 
 
+class ReprocessFilesSerializer(ReprocessSingleFileSerializer):
+    file_ids = serializers.ListField(
+        child=serializers.UUIDField(),
+        required=False,
+        help_text="IDs of the Files to reprocess.",
+    )
+    identity_id = IdentityIDField(
+        required=False,
+        help_text="Reprocess Files owned by this Identity.",
+    )
+    added_after = serializers.DateTimeField(
+        required=False,
+        help_text="Only reprocess Files added at or after this time.",
+    )
+    added_before = serializers.DateTimeField(
+        required=False,
+        help_text="Only reprocess Files added at or before this time.",
+    )
+
+    def validate(self, attrs):
+        attrs = super().validate(attrs)
+        if not attrs.get("file_ids") and not attrs.get("identity_id"):
+            raise serializers.ValidationError(
+                {
+                    "non_field_errors": [
+                        "At least one of file_ids or identity_id must be provided"
+                    ]
+                }
+            )
+        if (
+            attrs.get("added_after")
+            and attrs.get("added_before")
+            and attrs["added_after"] > attrs["added_before"]
+        ):
+            raise serializers.ValidationError(
+                {
+                    "non_field_errors": [
+                        "added_after cannot be later than added_before"
+                    ]
+                }
+            )
+        return attrs
+
+
 class AttackNavigatorDomainSerializer(JSONSchemaSerializer):
     json_schema = {
         "$schema": "http://json-schema.org/draft-07/schema#",

stixify/web/views.py

@@ -63,6 +63,7 @@
     HealthCheckSerializer,
     ImageSerializer,
     JobSerializer,
+    ReprocessFilesSerializer,
     ReprocessSingleFileSerializer,
 )
 from .topics import SimilarFileSerializer
@@ -202,6 +203,7 @@ def _get_schema_response(self, request):
         ),
     ),
     reprocess=extend_schema(
+        operation_id="v1_files_reprocess_single",
         summary="Reprocess an uploaded File",
         description=textwrap.dedent(
             """
@@ -221,6 +223,24 @@ def _get_schema_response(self, request):
         },
         request=ReprocessSingleFileSerializer,
     ),
+    reprocess_files=extend_schema(
+        operation_id="v1_files_reprocess_multiple",
+        summary="Reprocess uploaded Files",
+        description=textwrap.dedent(
+            """
+            Reprocess multiple Files selected by `file_ids`, `identity_id`, or both.
+
+            `added_after` and `added_before` can be used to restrict the resolved Files
+            by their immutable `added` timestamp. The response is one detached Job whose
+            `extra.file_ids` contains the complete resolved set of Files.
+            """
+        ),
+        responses={
+            201: JobSerializer,
+            400: DEFAULT_400_ERROR,
+        },
+        request=ReprocessFilesSerializer,
+    ),
 )
 class FileView(
     mixins.CreateModelMixin,
@@ -536,7 +556,110 @@ def reprocess(self, request, file_id=None, **kwargs):
             raise exceptions.ValidationError(
                 {"error": "Cannot skip extraction on unprocessed file"}
             )
-        job = tasks.create_reprocessing_job(file_obj, s.validated_data)
+        job = tasks.create_reprocessing_job(
+            [file_obj.id], self._reprocess_options(s.validated_data)
+        )
+        return Response(
+            JobSerializer(job, context={"request": request}).data,
+            status=status.HTTP_201_CREATED,
+        )
+
+    @staticmethod
+    def _reprocess_options(validated_data):
+        options = {}
+        for key in (
+            "identity_id",
+            "profile_id",
+            "skip_extraction",
+            "added_after",
+            "added_before",
+        ):
+            if key not in validated_data:
+                continue
+            value = validated_data[key]
+            if hasattr(value, "isoformat"):
+                value = value.isoformat()
+            elif value is not None:
+                value = str(value)
+            options[key] = value
+        return options
+
+    @decorators.action(
+        methods=["PATCH"], detail=False, url_path="reprocess"
+    )
+    def reprocess_files(self, request, **kwargs):
+        serializer = ReprocessFilesSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        data = serializer.validated_data
+
+        requested_ids = [str(file_id) for file_id in data.get("file_ids", [])]
+        existing_ids = {
+            str(file_id)
+            for file_id in File.objects.filter(pk__in=requested_ids).values_list(
+                "pk", flat=True
+            )
+        }
+        missing_ids = [
+            file_id for file_id in requested_ids if file_id not in existing_ids
+        ]
+        if missing_ids:
+            raise exceptions.ValidationError(
+                {"file_ids": [f"Unknown File IDs: {', '.join(missing_ids)}"]}
+            )
+
+        files = File.objects.all()
+        if data.get("added_after"):
+            files = files.filter(added__gte=data["added_after"])
+        if data.get("added_before"):
+            files = files.filter(added__lte=data["added_before"])
+
+        eligible_explicit_ids = {
+            str(file_id)
+            for file_id in files.filter(pk__in=requested_ids).values_list("pk", flat=True)
+        }
+        resolved_ids = []
+        seen = set()
+        for file_id in requested_ids:
+            if file_id in eligible_explicit_ids and file_id not in seen:
+                resolved_ids.append(file_id)
+                seen.add(file_id)
+
+        if identity_id := data.get("identity_id"):
+            identity_file_ids = files.filter(identity_id=identity_id).order_by(
+                "added", "id"
+            ).values_list("pk", flat=True)
+            for file_id in identity_file_ids:
+                file_id = str(file_id)
+                if file_id not in seen:
+                    resolved_ids.append(file_id)
+                    seen.add(file_id)
+
+        if not resolved_ids:
+            raise exceptions.ValidationError(
+                {"file_ids": ["No Files matched the supplied selectors"]}
+            )
+
+        if data["skip_extraction"]:
+            unprocessed_ids = [
+                str(file_id)
+                for file_id, extraction_data in File.objects.filter(
+                    pk__in=resolved_ids
+                ).values_list("pk", "txt2stix_data")
+                if not extraction_data
+            ]
+            if unprocessed_ids:
+                raise exceptions.ValidationError(
+                    {
+                        "file_ids": [
+                            "Cannot skip extraction for unprocessed Files: "
+                            + ", ".join(unprocessed_ids)
+                        ]
+                    }
+                )
+
+        job = tasks.create_reprocessing_job(
+            resolved_ids, self._reprocess_options(data)
+        )
         return Response(
             JobSerializer(job, context={"request": request}).data,
             status=status.HTTP_201_CREATED,
@@ -589,11 +712,18 @@ def get_queryset(self):
         return Job.objects.all()
 
     class filterset_class(FilterSet):
-        file_id = Filter("file_id", label="Filter Jobs by File `id`")
+        file_id = filters.UUIDFilter(
+            method="filter_file_id", label="Filter Jobs by File `id`"
+        )
         state = filters.BaseCSVFilter(
             help_text="Filter Jobs by their state.", lookup_expr="in"
         )
 
+        def filter_file_id(self, queryset, name, value):
+            return queryset.filter(
+                Q(file_id=value) | Q(extra__file_ids__contains=[str(value)])
+            )
+
 
 @extend_schema_view(
     list=extend_schema(