Implement a ScanForVirus Pipeline #1182 (#1193)

Signed-off-by: tdruez <tdruez@nexb.com>

Author: tdruez

CHANGELOG.rst

@@ -4,6 +4,9 @@ Changelog
 v34.6.0 (unreleased)
 --------------------
 
+- Add a new ``scan_for_virus`` add-on pipeline based on ClamAV scan.
+  https://github.com/nexB/scancode.io/issues/1182
+
 - Add ability to filter by tag on the resource list view.
   https://github.com/nexB/scancode.io/issues/1217
 

docker-compose.yml

@@ -66,8 +66,18 @@ services:
       - web
     restart: always
 
+  clamav:
+    image: clamav/clamav
+    volumes:
+      - clamav_data:/var/lib/clamav
+      - workspace:/var/scancodeio/workspace/
+    ports:
+      - "3310:3310"
+    restart: always
+
 volumes:
   db_data:
   redis_data:
+  clamav_data:
   static:
   workspace:

docs/built-in-pipelines.rst

@@ -182,6 +182,14 @@ Scan Codebase
     :members:
     :member-order: bysource
 
+.. _pipeline_scan_for_virus:
+
+Scan For Virus
+--------------
+.. autoclass:: scanpipe.pipelines.scan_for_virus.ScanForVirus()
+    :members:
+    :member-order: bysource
+
 .. _pipeline_scan_single_package:
 
 Scan Single Package

docs/scanpipe-pipes.rst

@@ -8,6 +8,11 @@ Generic
 .. automodule:: scanpipe.pipes
     :members:
 
+ClamAV
+------
+.. automodule:: scanpipe.pipes.clamav
+    :members:
+
 Codebase
 --------
 .. automodule:: scanpipe.pipes.codebase

scancodeio/settings.py

@@ -364,6 +364,10 @@
     for queue_config in RQ_QUEUES.values():
         queue_config["ASYNC"] = False
 
+# ClamAV virus scan
+CLAMD_USE_TCP = env.bool("CLAMD_USE_TCP", default=True)
+CLAMD_TCP_ADDR = env.str("CLAMD_TCP_ADDR", default="clamav")
+
 # Django restframework
 
 REST_FRAMEWORK = {

scanpipe/pipelines/scan_for_virus.py

@@ -0,0 +1,39 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+from scanpipe.pipelines import Pipeline
+from scanpipe.pipes import clamav
+
+
+class ScanForVirus(Pipeline):
+    """Run a ClamAV scan on the codebase directory to detect virus infection."""
+
+    download_inputs = False
+    is_addon = True
+
+    @classmethod
+    def steps(cls):
+        return (cls.scan_for_virus,)
+
+    def scan_for_virus(self):
+        """Run a ClamAV scan to detect virus infection."""
+        clamav.scan_for_virus(self.project)

scanpipe/pipes/clamav.py

@@ -0,0 +1,57 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+from pathlib import Path
+
+from django.conf import settings
+
+import clamd
+
+
+def scan_for_virus(project):
+    """
+    Run a ClamAV scan to detect virus infection.
+    Create one Project error message per found 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))
+    except clamd.ClamdError as e:
+        raise Exception(f"Error with the ClamAV service: {e}")
+
+    for resource_location, results in scan_response.items():
+        status, reason = results
+        resource_path = Path(resource_location).relative_to(project.codebase_path)
+        details = {
+            "status": status,
+            "reason": reason,
+            "resource_path": str(resource_path),
+        }
+        project.add_error(
+            description="Virus detected",
+            model="ScanForVirus",
+            details=details,
+        )

scanpipe/tests/pipes/test_clamav.py

@@ -0,0 +1,58 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# ScanCode.io is a free software code scanning tool from nexB Inc. and others.
+# Visit https://github.com/nexB/scancode.io for support and download.
+
+from pathlib import Path
+from unittest import mock
+
+from django.test import TestCase
+
+from scanpipe.models import Project
+from scanpipe.pipes import clamav
+from scanpipe.tests import make_resource_file
+
+
+class ScanPipeClamAVPipesTest(TestCase):
+    data_location = Path(__file__).parent.parent / "data"
+
+    @mock.patch("clamd.ClamdNetworkSocket.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")
+
+        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)
+        expected_details = {
+            "reason": "Win.Test.EICAR_HDB-1",
+            "status": "FOUND",
+            "resource_path": "eicar.zip",
+        }
+        self.assertEqual(expected_details, error_message.details)

setup.cfg

@@ -102,6 +102,8 @@ install_requires =
     # Markdown
     markdown-it-py==3.0.0
     bleach==6.1.0
+    # Antivirus
+    clamd==1.0.2
 
 [options.extras_require]
 dev =
@@ -146,6 +148,7 @@ scancodeio_pipelines =
     populate_purldb = scanpipe.pipelines.populate_purldb:PopulatePurlDB
     resolve_dependencies = scanpipe.pipelines.resolve_dependencies:ResolveDependencies
     scan_codebase = scanpipe.pipelines.scan_codebase:ScanCodebase
+    scan_for_virus = scanpipe.pipelines.scan_for_virus:ScanForVirus
     scan_single_package = scanpipe.pipelines.scan_single_package:ScanSinglePackage
 
 [isort]