Update Attachment structure

Problem:
Current attachment handling is not flexible and confusing and makes sharing attachments hard.

Solution:
Updated attachment structure to be more flexible and to allow easy sharing without any special handling.

Signed-off-by: Waldemar Parzonka <[email protected]>

Author: wp0pw

archivist/archivistpublic.py

@@ -228,7 +228,7 @@ def get_file(
 
         Args:
             url (str): e.g. assets/xxxxxxxxxxxxxxxxxxxxxx
-            identity (str): e.g. attachments/xxxxxxxxxxxxxxxxxxxxxxxxxxxx`
+            identity (str): e.g. blobs/xxxxxxxxxxxxxxxxxxxxxxxxxxxx`
             fd (file): an iterable representing a file (usually from open())
                 the file must be opened in binary mode
             headers (dict): optional REST headers

archivist/asset.py

@@ -21,6 +21,14 @@ def primary_image(self) -> str | None:
             :class:`Attachment` instance
 
         """
+
+        # try latest API
+        try:
+            return self["attributes"]["arc_primary_image"]
+        except (KeyError, TypeError):
+            pass
+
+        # try old API
         try:
             attachments = self["attributes"]["arc_attachments"]
         except (KeyError, TypeError):

archivist/assets.py

@@ -204,6 +204,7 @@ def create_if_not_exists(
                 attachments:
                   - filename: functests/test_resources/doors/assets/entry-terminal.jpg
                     content_type: image/jpg
+                    attachment: terminal entry
 
             The 'selector' value is required and will usually specify the 'arc_display_name' as a
             secondary key. The keys in 'selector' must exist in the attributes of the asset.
@@ -250,9 +251,15 @@ def create_if_not_exists(
 
         # any attachments ?
         if attachments is not None:
-            data["attributes"]["arc_attachments"] = [
-                self._archivist.attachments.create(a) for a in attachments
-            ]
+            for a in attachments:
+                # attempt to get attachment to use as a key
+                attachment_key = a.get("attachment", None)
+                if attachment_key is None:
+                    # failing that create a key from filename or url
+                    attachment_key = self._archivist.attachments.get_default_key(a)
+                data["attributes"][attachment_key] = self._archivist.attachments.create(
+                    a
+                )
 
         asset = self.create_from_data(
             data=data,

archivist/attachments.py

@@ -28,6 +28,7 @@
 from copy import deepcopy
 from io import BytesIO
 from logging import getLogger
+from os import path
 from typing import BinaryIO, Optional, Any
 
 from requests.models import Response
@@ -72,6 +73,18 @@ def __init__(self, archivist_instance: archivist.Archivist):
     def __str__(self) -> str:
         return f"AttachmentsClient({self._archivist.url})"
 
+    def get_default_key(self, data: dict[str, str]) -> str:
+        """
+        Return a key to use if no key was provided
+        either use filename or url as one of them is required
+        """
+        attachment_key = (
+            data.get("filename", "")
+            if data.get("filename", "")
+            else data.get("url", "")
+        )
+        return attachment_key.replace(".", "_")
+
     def create(self, data: dict[str, Any]) -> dict[str, Any]:  # pragma: no cover
         """
         Create an attachment and return struct suitable for use in an asset
@@ -108,14 +121,17 @@ def create(self, data: dict[str, Any]) -> dict[str, Any]:  # pragma: no cover
             .. code-block:: yaml
 
                 arc_display_name: Telephone
-                arc_attachment_identity: blobs/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
-                arc_hash_alg: SHA256
-                arc_hash_value: xxxxxxxxxxxxxxxxxxxxxxx
+                arc_blob_identity: blobs/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+                arc_blob_hash_alg: SHA256
+                arc_blob_hash_value: xxxxxxxxxxxxxxxxxxxxxxx
+                arc_file_name: gdn_front.jpg
 
         """
         result = None
+        file_part = None
         filename = data.get("filename")
         if filename is not None:
+            _, file_part = path.split(filename)
             with open(filename, "rb") as fd:
                 attachment = self.upload(fd, mtype=data.get("content_type"))
 
@@ -126,11 +142,15 @@ def create(self, data: dict[str, Any]) -> dict[str, Any]:  # pragma: no cover
             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"],
+            "arc_attribute_type": "arc_attachment",
+            "arc_blob_identity": attachment["identity"],
+            "arc_blob_hash_alg": attachment["hash"]["alg"],
+            "arc_blob_hash_value": attachment["hash"]["value"],
         }
 
+        if file_part:
+            result["arc_file_name"] = file_part
+
         display_name = data.get("display_name")
         if display_name is not None:
             result["arc_display_name"] = display_name
@@ -179,7 +199,7 @@ def download(
         fd iterator
 
         Args:
-            identity (str): attachment identity e.g. attachments/xxxxxxxxxxxxxxxxxxxxxxx
+            identity (str): attachment identity e.g. blobs/xxxxxxxxxxxxxxxxxxxxxxx
             fd (file): opened file descriptor or other file-type sink..
             params (dict): e.g. {"allow_insecure": "true"} OR {"strict": "true" }
 
@@ -202,7 +222,7 @@ def info(
         Reads attachment info
 
         Args:
-            identity (str): attachment identity e.g. attachments/xxxxxxxxxxxxxxxxxxxxxxx
+            identity (str): attachment identity e.g. blobs/xxxxxxxxxxxxxxxxxxxxxxx
 
         Returns:
             REST response

archivist/events.py

@@ -360,19 +360,20 @@ def create_from_data(
 
         attachments = data.pop("attachments", None)
         if attachments is not None:
-            result = [self._archivist.attachments.create(a) for a in attachments]
-            for i, a in enumerate(attachments):
+            for a in attachments:
+                result = self._archivist.attachments.create(a)
                 if a.get("type") == SBOM_RELEASE:
                     sbom_result = self._archivist.sboms.parse(a)
                     for k, v in sbom_result.items():
                         event_attributes[f"sbom_{k}"] = v
 
-                    event_attributes["sbom_identity"] = result[i][
-                        "arc_attachment_identity"
-                    ]
-                    break
+                    event_attributes["sbom_identity"] = result["arc_blob_identity"]
 
-            event_attributes["arc_attachments"] = result
+                attachment_key = a.get("attachment", None)
+                if attachment_key is None:
+                    # failing that create a key from filename or url
+                    attachment_key = self._archivist.attachments.get_default_key(a)
+                event_attributes[attachment_key] = result
 
         data["event_attributes"] = event_attributes
 

archivist/notebooks/Find Asset and Create Attachment.ipynb

@@ -132,9 +132,9 @@
     "        blob = arch.attachments.upload(fd)\n",
     "        attachment = {\n",
     "            \"arc_display_name\": name,\n",
-    "            \"arc_attachment_identity\": blob[\"identity\"],\n",
-    "            \"arc_hash_value\": blob[\"hash\"][\"value\"],\n",
-    "            \"arc_hash_alg\": blob[\"hash\"][\"alg\"],\n",
+    "            \"arc_blob_identity\": blob[\"identity\"],\n",
+    "            \"arc_blob_hash_value\": blob[\"hash\"][\"value\"],\n",
+    "            \"arc_blob_hash_alg\": blob[\"hash\"][\"alg\"],\n",
     "        }\n",
     "        return attachment"
    ]
@@ -173,15 +173,14 @@
     "    attachments = upload_attachment(\n",
     "        arch, \"pexels-andrea-turner-707697.jpeg\", \"arc_primary_image\"\n",
     "    )\n",
-    "    attachment_list = [attachments]\n",
     "\n",
     "    props = {\"operation\": \"Record\", \"behaviour\": \"RecordEvidence\"}\n",
     "    attrs = {\n",
     "        \"arc_description\": \"Attaching an image\",\n",
     "        \"arc_display_type\": \"Primary image\",\n",
     "    }\n",
     "\n",
-    "    asset_attrs = {\"arc_attachments\": attachment_list}\n",
+    "    asset_attrs = {\"pexels-andrea-turner-707697\": attachments}\n",
     "\n",
     "    return arch.events.create(\n",
     "        asset[\"identity\"], props=props, attrs=attrs, asset_attrs=asset_attrs\n",

archivist/sboms.py

@@ -174,7 +174,7 @@ def create(self, data: dict[str, Any]) -> dict[str, Any]:  # pragma: no cover
             .. code-block:: yaml
 
                 arc_display_name: Acme Generation1 SBOM
-                arc_attachment_identity: sboms/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+                arc_blob_identity: sboms/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
                 .....
 
         """

examples/scan_test.py

@@ -67,10 +67,12 @@ def scan_test(arch, datestring, scanned_expected=False):
                         "main/functests/test_resources/telephone.jpg"
                     ),
                     "content_type": "image/jpg",
+                    "attachment": "telephone1",
                 },
                 {
                     "url": "https://secure.eicar.org/eicarcom2.zip",
                     "content_type": "application/zip",
+                    "attachment": "zipfile",
                 },
             ],
         },
@@ -82,7 +84,7 @@ def scan_test(arch, datestring, scanned_expected=False):
     fails = []
 
     # first attachment should be clean....
-    attachment_id = asset["attributes"]["arc_attachments"][0]["arc_attachment_identity"]
+    attachment_id = asset["attributes"]["telephone1"]["arc_blob_identity"]
     info = arch.attachments.info(
         attachment_id,
         asset_or_event_id=asset["identity"],
@@ -100,7 +102,7 @@ def scan_test(arch, datestring, scanned_expected=False):
         fails.append("Yesterday's first attachment has not been scanned.")
 
     # second attachment should be bad when scanned....
-    attachment_id = asset["attributes"]["arc_attachments"][1]["arc_attachment_identity"]
+    attachment_id = asset["attributes"]["zipfile"]["arc_blob_identity"]
     info = arch.attachments.info(
         attachment_id,
         asset_or_event_id=asset["identity"],

functests/execassets.py

@@ -59,10 +59,12 @@
         {
             "filename": "functests/test_resources/telephone.jpg",
             "content_type": "image/jpg",
+            "attachment": "telephone",
         },
         {
             "url": "https://secure.eicar.org/eicarcom2.zip",
             "content_type": "application/zip",
+            "attachment": "zipfile",
         },
     ],
 }
@@ -271,9 +273,7 @@ def test_asset_create_if_not_exists_with_bad_attachment(self):
         LOGGER.debug("existed %s", existed)
 
         # first attachment is ok....
-        attachment_id = asset["attributes"]["arc_attachments"][0][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["telephone"]["arc_blob_identity"]
         info = self.arch.attachments.info(
             attachment_id,
         )
@@ -290,9 +290,7 @@ def test_asset_create_if_not_exists_with_bad_attachment(self):
             )
 
         # second attachment is bad when scanned....
-        attachment_id = asset["attributes"]["arc_attachments"][1][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["zipfile"]["arc_blob_identity"]
         info = self.arch.attachments.info(
             attachment_id,
         )
@@ -327,9 +325,7 @@ def test_asset_create_if_not_exists_with_bad_attachment_assetattachment(self):
         LOGGER.debug("existed %s", existed)
 
         # first attachment is ok....
-        attachment_id = asset["attributes"]["arc_attachments"][0][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["telephone"]["arc_blob_identity"]
         info = self.arch.assetattachments.info(
             asset["identity"],
             attachment_id,
@@ -347,9 +343,7 @@ def test_asset_create_if_not_exists_with_bad_attachment_assetattachment(self):
             )
 
         # second attachment is bad when scanned....
-        attachment_id = asset["attributes"]["arc_attachments"][1][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["zipfile"]["arc_blob_identity"]
         info = self.arch.assetattachments.info(
             asset["identity"],
             attachment_id,

functests/execpublicassets.py

@@ -59,10 +59,12 @@
         {
             "filename": "functests/test_resources/telephone.jpg",
             "content_type": "image/jpg",
+            "attachment": "telephone",
         },
         {
             "url": "https://secure.eicar.org/eicarcom2.zip",
             "content_type": "application/zip",
+            "attachment": "zipfile",
         },
     ],
     "public": True,
@@ -231,9 +233,7 @@ def test_asset_create_if_not_exists_with_bad_attachment_assetattachment(self):
 
         asset_id = asset["identity"]
         # first attachment is ok....
-        attachment_id = asset["attributes"]["arc_attachments"][0][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["telephone"]["arc_blob_identity"]
         public_asset_id = self.arch.assets.publicurl(asset_id)
         LOGGER.debug("public asset id %s", public_asset_id)
         public = self.arch.Public
@@ -252,9 +252,7 @@ def test_asset_create_if_not_exists_with_bad_attachment_assetattachment(self):
             )
 
         # second attachment is bad when scanned....
-        attachment_id = asset["attributes"]["arc_attachments"][1][
-            "arc_attachment_identity"
-        ]
+        attachment_id = asset["attributes"]["zipfile"]["arc_blob_identity"]
         info = public.assetattachments.info(public_asset_id, attachment_id)
         LOGGER.debug("info attachment1 %s", json_dumps(info, indent=4))
         timestamp = info["scanned_timestamp"]