Improve scan_for_virus robustness, avoid per-file DB lookups, fix typo, and add missing-resource test

Signed-off-by: dikshaa2909 <dikshadeware@gmail.com>

Author: dikshaa2909

scanpipe/pipes/clamav.py

@@ -20,48 +20,92 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/aboutcode-org/scancode.io for support and download.
 
+import logging
 from pathlib import Path
 
+import clamd
 from django.conf import settings
 
-import clamd
+logger = logging.getLogger(__name__)
 
 
 def scan_for_virus(project):
     """
     Run a ClamAV scan to detect virus infection.
-    Create one Project error message per found virus and store the detection data
-    on the related codebase resource ``extra_data`` field.
+    - Avoid crashes when ClamAV reports files not indexed in CodebaseResource.
+    - Avoid per-file DB queries by preloading valid resource paths.
+    - Record a project-level error for any detected virus.
     """
     if settings.CLAMD_USE_TCP:
         clamd_socket = clamd.ClamdNetworkSocket(settings.CLAMD_TCP_ADDR)
     else:
         clamd_socket = clamd.ClamdUnixSocket()
 
     try:
-        scan_response = clamd_socket.multiscan(file=str(project.codebase_path))
+        scan_response = clamd_socket.multiscan(
+            file=str(project.codebase_path)
+        )
     except clamd.ClamdError as e:
-        raise Exception(f"Error with the ClamAV service: {e}")
+        raise Exception(
+            f"Error with the ClamAV service: {e}"
+        )
+
+    # Preload all valid indexed resource paths
+    valid_paths = set(
+        project.codebaseresources.values_list("path", flat=True)
+    )
+
+    missing_resources = []
 
     for resource_location, results in scan_response.items():
         status, reason = results
-        resource_path = Path(resource_location).relative_to(project.codebase_path)
 
-        resource = project.codebaseresources.get(path=resource_path)
+        if status != "FOUND":
+            continue
+
+        resource_path = Path(resource_location).relative_to(
+            project.codebase_path
+        )
+        resource_path_str = str(resource_path)
+
+        if resource_path_str not in valid_paths:
+            missing_resources.append(resource_path_str)
+            logger.warning(
+                "ClamAV detected virus in non-indexed file: %s",
+                resource_path_str,
+            )
+            continue
+
+        resource = project.codebaseresources.filter(
+            path=resource_path_str
+        ).first()
+
         virus_report = {
-            "calmav": {
+            "clamav": {
                 "status": status,
                 "reason": reason,
             }
         }
-        resource.update_extra_data({"virus_report": virus_report})
+
+        resource.update_extra_data(
+            {"virus_report": virus_report}
+        )
 
         project.add_error(
             description="Virus detected",
             model="ScanForVirus",
             details={
                 "status": status,
                 "reason": reason,
-                "resource_path": str(resource_path),
+                "resource_path": resource_path_str,
             },
         )
+
+    if missing_resources:
+        project.add_error(
+            description="ClamAV detected virus in files not indexed in DB",
+            model="ScanForVirus",
+            details={
+                "missing_resources": missing_resources
+            },
+        )

scanpipe/tests/pipes/test_clamav.py

@@ -34,22 +34,36 @@ class ScanPipeClamAVPipesTest(TestCase):
     data = Path(__file__).parent.parent / "data"
 
     @mock.patch("clamd.ClamdNetworkSocket.multiscan")
-    def test_scanpipe_pipes_clamav_scan_for_virus(self, mock_multiscan):
+    def test_scanpipe_pipes_clamav_scan_for_virus(
+        self, mock_multiscan
+    ):
         project = Project.objects.create(name="project")
         r1 = make_resource_file(project=project, path="eicar.zip")
-        r2 = make_resource_file(project=project, path="eicar.zip-extract/eicar.com")
+        r2 = make_resource_file(
+            project=project,
+            path="eicar.zip-extract/eicar.com",
+        )
 
         mock_multiscan.return_value = {
             r1.location: ("FOUND", "Win.Test.EICAR_HDB-1"),
             r2.location: ("FOUND", "Win.Test.EICAR_HDB-1"),
         }
 
         clamav.scan_for_virus(project)
+
         self.assertEqual(2, len(project.projectmessages.all()))
+
         error_message = project.projectmessages.all()[0]
         self.assertEqual("error", error_message.severity)
-        self.assertEqual("Virus detected", error_message.description)
-        self.assertEqual("ScanForVirus", error_message.model)
+        self.assertEqual(
+            "Virus detected",
+            error_message.description,
+        )
+        self.assertEqual(
+            "ScanForVirus",
+            error_message.model,
+        )
+
         expected_details = {
             "reason": "Win.Test.EICAR_HDB-1",
             "status": "FOUND",
@@ -60,10 +74,39 @@ def test_scanpipe_pipes_clamav_scan_for_virus(self, mock_multiscan):
         resource1 = project.codebaseresources.first()
         expected_virus_report_extra_data = {
             "virus_report": {
-                "calmav": {
+                "clamav": {
                     "status": "FOUND",
                     "reason": "Win.Test.EICAR_HDB-1",
                 }
             }
         }
-        self.assertEqual(expected_virus_report_extra_data, resource1.extra_data)
+        self.assertEqual(
+            expected_virus_report_extra_data,
+            resource1.extra_data,
+        )
+
+    @mock.patch("clamd.ClamdNetworkSocket.multiscan")
+    def test_scanpipe_pipes_clamav_missing_resource_does_not_crash(
+        self, mock_multiscan
+    ):
+        project = Project.objects.create(name="project")
+
+        r1 = make_resource_file(
+            project=project,
+            path="indexed.txt",
+        )
+
+        mock_multiscan.return_value = {
+            r1.location: ("FOUND", "Test.Virus"),
+            str(project.codebase_path / "non_indexed.txt"): (
+                "FOUND",
+                "Test.Virus",
+            ),
+        }
+
+        clamav.scan_for_virus(project)
+
+        self.assertEqual(
+            2,
+            len(project.projectmessages.all()),
+        )