Merge pull request #321 from TheHive-Project/319-attachement-to-an-alert

319 attachement to an alert

.github/workflows/_build-package.yml

@@ -6,7 +6,7 @@ jobs:
     name: Build wheel and sdist
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Set up Python
         uses: actions/setup-python@v4
         with: 

.github/workflows/_integration-tests.yml

@@ -9,7 +9,7 @@ jobs:
     name: Run integration tests
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Set up Python
         uses: actions/setup-python@v4
         with:

.github/workflows/_static-checks.yml

@@ -9,7 +9,7 @@ jobs:
       matrix:
         python-version: ["3.8", "3.9", "3.10", "3.11"]
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v4
         with:

.github/workflows/_upload-package.yml

@@ -9,7 +9,7 @@ jobs:
     name: Upload wheel and sdist
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Compare tag and package version
         run: |
           TAG=${GITHUB_REF#refs/*/}

tests/conftest.py

@@ -23,7 +23,7 @@
 @pytest.fixture(scope="session")
 def test_config():
     return TestConfig(
-        image_name="kamforka/thehive4py-integrator:thehive-5.2.10",
+        image_name="kamforka/thehive4py-integrator:thehive-5.2.11",
         container_name="thehive4py-integration-tests",
         user="[email protected]",
         password="secret",

tests/test_alert_endpoint.py

@@ -235,3 +235,48 @@ def test_create_and_find_procedure(
         )
         alert_procedures = thehive.alert.find_procedures(alert_id=test_alert["_id"])
         assert [created_procedure] == alert_procedures
+
+    def test_add_and_download_attachment(
+        self, thehive: TheHiveApi, test_alert: OutputAlert, tmp_path: Path
+    ):
+        attachment_paths = [str(tmp_path / f"attachment-{i}.txt") for i in range(2)]
+        download_attachment_paths = [
+            str(tmp_path / f"dl-attachment-{i}.txt") for i in range(2)
+        ]
+
+        for path in attachment_paths:
+            with open(path, "w") as attachment_fp:
+                attachment_fp.write(f"content of {path}")
+
+        added_attachments = thehive.alert.add_attachment(
+            alert_id=test_alert["_id"], attachment_paths=attachment_paths
+        )
+
+        for attachment, path in zip(added_attachments, download_attachment_paths):
+            thehive.alert.download_attachment(
+                alert_id=test_alert["_id"],
+                attachment_id=attachment["_id"],
+                attachment_path=path,
+            )
+
+        for original, downloaded in zip(attachment_paths, download_attachment_paths):
+            with open(original) as original_fp, open(downloaded) as downloaded_fp:
+                assert original_fp.read() == downloaded_fp.read()
+
+    def test_add_and_delete_attachment(
+        self, thehive: TheHiveApi, test_alert: OutputAlert, tmp_path: Path
+    ):
+        attachment_path = str(tmp_path / "my-attachment.txt")
+        with open(attachment_path, "w") as attachment_fp:
+            attachment_fp.write("some content...")
+
+        added_attachments = thehive.alert.add_attachment(
+            alert_id=test_alert["_id"], attachment_paths=[attachment_path]
+        )
+
+        for attachment in added_attachments:
+            thehive.alert.delete_attachment(
+                alert_id=test_alert["_id"], attachment_id=attachment["_id"]
+            )
+
+        assert thehive.alert.find_attachments(alert_id=test_alert["_id"]) == []

thehive4py/endpoints/alert.py

@@ -9,10 +9,11 @@
 from thehive4py.types.alert import (
     InputAlert,
     InputBulkUpdateAlert,
-    InputUpdateAlert,
     InputPromoteAlert,
+    InputUpdateAlert,
     OutputAlert,
 )
+from thehive4py.types.attachment import OutputAttachment
 from thehive4py.types.case import OutputCase
 from thehive4py.types.comment import OutputComment
 from thehive4py.types.observable import InputObservable, OutputObservable
@@ -83,6 +84,31 @@ def create_observable(
             "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
         )
 
+    def add_attachment(
+        self, alert_id: str, attachment_paths: List[str]
+    ) -> List[OutputAttachment]:
+        files = [
+            ("attachments", self._fileinfo_from_filepath(attachment_path))
+            for attachment_path in attachment_paths
+        ]
+        return self._session.make_request(
+            "POST", f"/api/v1/alert/{alert_id}/attachments", files=files
+        )["attachments"]
+
+    def download_attachment(
+        self, alert_id: str, attachment_id: str, attachment_path: str
+    ) -> None:
+        return self._session.make_request(
+            "GET",
+            path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}/download",
+            download_path=attachment_path,
+        )
+
+    def delete_attachment(self, alert_id: str, attachment_id: str) -> None:
+        return self._session.make_request(
+            "DELETE", path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}"
+        )
+
     def merge_into_case(self, alert_id: str, case_id: str) -> OutputCase:
         return self._session.make_request(
             "POST", path=f"/api/v1/alert/{alert_id}/merge/{case_id}"
@@ -191,3 +217,22 @@ def find_procedures(
             params={"name": "alert-procedures"},
             json={"query": query},
         )
+
+    def find_attachments(
+        self,
+        alert_id: str,
+        filters: Optional[FilterExpr] = None,
+        sortby: Optional[SortExpr] = None,
+        paginate: Optional[Paginate] = None,
+    ) -> List[OutputAttachment]:
+        query: QueryExpr = [
+            {"_name": "getAlert", "idOrName": alert_id},
+            {"_name": "attachments"},
+            *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
+        ]
+        return self._session.make_request(
+            "POST",
+            path="/api/v1/query",
+            params={"name": "alert-attachments"},
+            json={"query": query},
+        )