Submit test malware to check attachment scanning

Problem:
A test is required that malware is scanned and
marked as bad.

Solution:
Create an asset if not existing
with 2 attachments - one good and one bad. Check that if they
have been scanned then each returns OK or BAD as required.
First attachment is clean, second attachment is test malware.

Signed-off-by: Paul Hewlett <phewlett76@gmail.com>

Author: eccles

archivist/attachments.py

@@ -25,6 +25,7 @@
 # pylint:disable=too-few-public-methods
 
 from copy import deepcopy
+from io import BytesIO
 from logging import getLogger
 from typing import BinaryIO, Dict, Optional
 
@@ -34,8 +35,15 @@
 # pylint:disable=cyclic-import      # but pylint doesn't understand this feature
 from . import archivist as type_helper
 
-from .constants import ATTACHMENTS_SUBPATH, ATTACHMENTS_LABEL
+from .constants import (
+    ASSETS_SUBPATH,
+    ATTACHMENTS_SUBPATH,
+    ATTACHMENTS_LABEL,
+    ATTACHMENTS_ASSETS_EVENTS_LABEL,
+    SEP,
+)
 from .dictmerge import _deepmerge
+from .utils import get_url
 from .type_aliases import NoneOnError
 
 LOGGER = getLogger(__name__)
@@ -81,7 +89,15 @@ def create(self, data: Dict) -> Dict:  # pragma: no cover
                 filename: functests/test_resources/doors/assets/gdn_front.jpg
                 content_type: image/jpg
 
-             Both 'filename' and 'content_type' are required.
+            OR
+
+            .. code-block:: yaml
+
+                url: https://secure.eicar.org/eicar.com"
+                content_type: image/jpg
+
+             Either 'filename' or 'url' is required.
+             'content_type' is required.
 
         Returns:
 
@@ -97,13 +113,22 @@ def create(self, data: Dict) -> Dict:  # pragma: no cover
 
         """
         result = None
-        with open(data["filename"], "rb") as fd:
+        filename = data.get("filename", None)
+        if filename is not None:
+            with open(filename, "rb") as fd:
+                attachment = self.upload(fd, mtype=data.get("content_type"))
+
+        else:
+            url = data["url"]
+            fd = BytesIO()
+            get_url(url, fd)
             attachment = self.upload(fd, mtype=data.get("content_type"))
-            result = {
-                "arc_attachment_identity": attachment["identity"],
-                "arc_hash_alg": attachment["hash"]["alg"],
-                "arc_hash_value": attachment["hash"]["value"],
-            }
+
+        result = {
+            "arc_attachment_identity": attachment["identity"],
+            "arc_hash_alg": attachment["hash"]["alg"],
+            "arc_hash_value": attachment["hash"]["value"],
+        }
 
         return result
 
@@ -141,6 +166,7 @@ def download(
         fd: BinaryIO,
         *,
         params: Optional[Dict] = None,
+        asset_or_event_id: Optional[str] = None,
     ) -> Response:
         """Read attachment
 
@@ -150,16 +176,64 @@ def download(
 
         Args:
             identity (str): attachment identity e.g. attachments/xxxxxxxxxxxxxxxxxxxxxxx
-            fd (file): opened file escriptor or other file-type sink..
+            fd (file): opened file descriptor or other file-type sink..
             params (dict): e.g. {"allow_insecure": "true"} OR {"strict": "true" }
+            asset_or_event_id (str): optional asset identity
+                                     e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx
+                                          assets/xxxxx/events/xxxx
 
         Returns:
             REST response
 
         """
+        if asset_or_event_id is not None:
+            uuid = identity.split(SEP)[1]
+            identity = SEP.join((asset_or_event_id, uuid))
+            return self._archivist.get_file(
+                SEP.join((ASSETS_SUBPATH, ATTACHMENTS_ASSETS_EVENTS_LABEL)),
+                identity,
+                fd,
+                params=self.__params(params),
+            )
+
         return self._archivist.get_file(
             ATTACHMENTS_SUBPATH,
             identity,
             fd,
             params=self.__params(params),
         )
+
+    def info(
+        self,
+        identity: str,
+        *,
+        asset_or_event_id: Optional[str] = None,
+    ) -> Response:
+        """Read attachment info
+
+        Reads attachment info
+
+        Args:
+            identity (str): attachment identity e.g. attachments/xxxxxxxxxxxxxxxxxxxxxxx
+            asset_or_event_id (str): optional asset identity
+                                     e.g. assets/xxxxxxxxxxxxxxxxxxxxxxxxxx
+                                          assets/xxxxx/events/xxxx
+
+        Returns:
+            REST response
+
+        """
+        if asset_or_event_id is not None:
+            uuid = identity.split(SEP)[1]
+            identity = SEP.join((asset_or_event_id, uuid))
+            return self._archivist.get(
+                SEP.join((ASSETS_SUBPATH, ATTACHMENTS_ASSETS_EVENTS_LABEL)),
+                identity,
+                tail="info",
+            )
+
+        return self._archivist.get(
+            ATTACHMENTS_SUBPATH,
+            identity,
+            tail="info",
+        )

archivist/constants.py

@@ -32,6 +32,7 @@
 
 ATTACHMENTS_SUBPATH = "v1"
 ATTACHMENTS_LABEL = "blobs"
+ATTACHMENTS_ASSETS_EVENTS_LABEL = "attachments"
 
 ACCESS_POLICIES_SUBPATH = "iam/v1"
 ACCESS_POLICIES_LABEL = "access_policies"

archivist/runner.py

@@ -32,6 +32,10 @@ class _ActionMap(dict):
 
     def __init__(self, archivist: "type_helper.Archivist"):
         super().__init__()
+        self["ASSETS_ATTACHMENT_INFO"] = {
+            "action": archivist.attachments.info,
+            "keywords": ("asset_or_event_id",),
+        }
         self["ASSETS_CREATE_IF_NOT_EXISTS"] = {
             "action": archivist.assets.create_if_not_exists,
             "keywords": ("confirm",),
@@ -357,7 +361,7 @@ def set_deletions(self, response: Dict, delete_method):
     def delete(self):
         """Deletes all entities"""
         for identity, delete_method in self.deletions.items():
-            LOGGER.info("Delete %s -> %s", identity, delete_method)
+            LOGGER.info("Delete %s", identity)
             delete_method(identity)
 
     def asset_id(self, name: str) -> Optional[str]:

archivist/utils.py

@@ -1,7 +1,9 @@
-"""Some covenience for confirmers
+"""Some convenience for confirmers
 """
 
+from io import BytesIO
 from logging import getLogger
+from requests import get as requests_get
 
 LOGGER = getLogger(__name__)
 
@@ -29,3 +31,18 @@ def backoff_handler(details):
         client,
         identity,
     )
+
+
+def get_url(url: str, fd: BytesIO):  # pragma no cover
+    """GET method (REST) - chunked
+
+    Downloads a binary object from upstream storage.
+    """
+    response = requests_get(
+        url,
+        stream=True,
+    )
+
+    for chunk in response.iter_content(chunk_size=4096):
+        if chunk:
+            fd.write(chunk)

functests/execassets.py

@@ -3,12 +3,13 @@
 """
 
 from copy import copy, deepcopy
-import json
+from json import dumps as json_dumps
 from os import environ
 from unittest import skip, TestCase
 from uuid import uuid4
 
 from archivist.archivist import Archivist
+from archivist.assets import BEHAVIOURS
 from archivist.logger import set_logger
 from archivist.proof_mechanism import ProofMechanism
 
@@ -27,6 +28,35 @@
     "some_custom_attribute": "value",
 }
 
+ASSET_NAME = "Telephone with 2 attachments - one bad or not scanned 2022-03-01"
+REQUEST_EXISTS_ATTACHMENTS = {
+    "signature": {
+        "attributes": {
+            "arc_display_name": ASSET_NAME,
+            "arc_namespace": "namespace",
+        },
+    },
+    "behaviours": BEHAVIOURS,
+    "proof_mechanism": ProofMechanism.SIMPLE_HASH.name,
+    "attributes": {
+        "arc_firmware_version": "1.0",
+        "arc_serial_number": "vtl-x4-07",
+        "arc_description": "Traffic flow control light at A603 North East",
+        "arc_display_type": "Traffic light with violation camera",
+        "some_custom_attribute": "value",
+    },
+    "attachments": [
+        {
+            "filename": "functests/test_resources/telephone.jpg",
+            "content_type": "image/jpg",
+        },
+        {
+            "url": "https://secure.eicar.org/eicarcom2.zip",
+            "content_type": "application/zip",
+        },
+    ],
+}
+
 
 class TestAssetCreate(TestCase):
     """
@@ -140,7 +170,7 @@ def test_asset_create_event(self):
         # get identity of first asset
         identity = None
         for asset in self.arch.assets.list():
-            print("asset", json.dumps(asset, sort_keys=True, indent=4))
+            print("asset", json_dumps(asset, sort_keys=True, indent=4))
             identity = asset["identity"]
             break
 
@@ -176,4 +206,80 @@ def test_asset_create_event(self):
         event = self.arch.events.create(
             identity, props=props, attrs=attrs, confirm=True
         )
-        print("event", json.dumps(event, sort_keys=True, indent=4))
+        print("event", json_dumps(event, sort_keys=True, indent=4))
+
+
+class TestAssetCreateIfNotExists(TestCase):
+    """
+    Test Archivist Asset CreateIfNotExists method
+    """
+
+    maxDiff = None
+
+    def setUp(self):
+        with open(environ["TEST_AUTHTOKEN_FILENAME"], encoding="utf-8") as fd:
+            auth = fd.read().strip()
+        self.arch = Archivist(
+            environ["TEST_ARCHIVIST"], auth, verify=False, max_time=300
+        )
+
+    def tearDown(self):
+        self.arch = None
+
+    def test_asset_create_if_not_exists_with_bad_attachment(self):
+        """
+        Test asset creation if not exists - check attachment for scanned status.
+
+        Because we use create_if_not_exists the asset and attachments will persist.
+
+        The test checks the scanned timestamp and checks scanned status.
+        The first attachment should return OK after 24 hours and the second attachment
+        should return bad after 24 hours.
+
+        """
+        asset, existed = self.arch.assets.create_if_not_exists(
+            REQUEST_EXISTS_ATTACHMENTS,
+            confirm=True,
+        )
+        print("asset", json_dumps(asset, indent=4))
+        print("existed", existed)
+
+        # first attachment is ok....
+        attachment_id = asset["attributes"]["arc_attachments"][0][
+            "arc_attachment_identity"
+        ]
+        info = self.arch.attachments.info(
+            attachment_id,
+            asset_or_event_id=asset["identity"],
+        )
+        print("info attachment1", json_dumps(info, indent=4))
+        timestamp = info["scanned_timestamp"]
+        if timestamp:
+            print(attachment_id, "scanned last at", timestamp)
+            print(attachment_id, "scanned status", info["scanned_status"])
+            print(attachment_id, "scanned reason", info["scanned_reason"])
+            self.assertEqual(
+                info["scanned_status"],
+                "SCANNED_OK",
+                msg="First attachment is not clean",
+            )
+
+        # second attachment is bad when scanned....
+        attachment_id = asset["attributes"]["arc_attachments"][1][
+            "arc_attachment_identity"
+        ]
+        info = self.arch.attachments.info(
+            attachment_id,
+            asset_or_event_id=asset["identity"],
+        )
+        print("info attachment1", json_dumps(info, indent=4))
+        timestamp = info["scanned_timestamp"]
+        if timestamp:
+            print(attachment_id, "scanned last at", timestamp)
+            print(attachment_id, "scanned status", info["scanned_status"])
+            print(attachment_id, "scanned reason", info["scanned_reason"])
+            self.assertEqual(
+                info["scanned_status"],
+                "SCANNED_BAD",
+                msg="First attachment should not be clean",
+            )