#434 - Review alert endpoints

Author: Kamforka

tests/test_alert_endpoint.py

@@ -99,6 +99,19 @@ def test_merge_into_case(
         merged_alert = thehive.alert.get(alert_id=alert_id)
         assert merged_alert.get("caseId") == merged_case["_id"]
 
+    def test_import_into_case(
+        self, thehive: TheHiveApi, test_alert: OutputAlert, test_case: OutputCase
+    ):
+        alert_id = test_alert["_id"]
+        case_id = test_case["_id"]
+
+        imported_case = thehive.alert.import_into_case(
+            alert_id=alert_id, case_id=case_id
+        )
+        imported_alert = thehive.alert.get(alert_id=alert_id)
+
+        assert imported_alert.get("caseId") == imported_case["_id"]
+
     def test_bulk_merge_into_case(
         self, thehive: TheHiveApi, test_alerts: List[OutputAlert], test_case: OutputCase
     ):
@@ -136,6 +149,16 @@ def test_bulk_delete(self, thehive: TheHiveApi, test_alerts: List[OutputAlert]):
             with pytest.raises(TheHiveError):
                 thehive.alert.get(alert_id)
 
+    def test_get_similar_observables(
+        self, thehive: TheHiveApi, test_alerts: List[OutputAlert]
+    ):
+
+        similar_observables = thehive.alert.get_similar_observables(
+            alert_id=test_alerts[0]["_id"], alert_or_case_id=test_alerts[1]["_id"]
+        )
+
+        assert similar_observables == []
+
     def test_create_and_get_observable(
         self, thehive: TheHiveApi, test_alert: OutputAlert
     ):
@@ -262,11 +285,12 @@ def test_add_and_download_attachment(
         )
 
         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,
-            )
+            with pytest.deprecated_call():
+                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:

thehive4py/endpoints/alert.py

@@ -1,4 +1,5 @@
 import json as jsonlib
+import warnings
 from typing import Any, Dict, List, Optional
 
 from thehive4py.endpoints._base import EndpointBase
@@ -56,30 +57,30 @@ def get(self, alert_id: str) -> OutputAlert:
 
         return self._session.make_request("GET", path=f"/api/v1/alert/{alert_id}")
 
-    def update(self, alert_id: str, fields: InputUpdateAlert) -> None:
-        """Update an alert.
+    def delete(self, alert_id: str) -> None:
+        """Delete an alert.
 
         Args:
             alert_id: The id of the alert.
-            fields: The fields of the alert to update.
 
         Returns:
             N/A
         """
-        return self._session.make_request(
-            "PATCH", path=f"/api/v1/alert/{alert_id}", json=fields
-        )
+        return self._session.make_request("DELETE", path=f"/api/v1/alert/{alert_id}")
 
-    def delete(self, alert_id: str) -> None:
-        """Delete an alert.
+    def update(self, alert_id: str, fields: InputUpdateAlert) -> None:
+        """Update an alert.
 
         Args:
             alert_id: The id of the alert.
+            fields: The fields of the alert to update.
 
         Returns:
             N/A
         """
-        return self._session.make_request("DELETE", path=f"/api/v1/alert/{alert_id}")
+        return self._session.make_request(
+            "PATCH", path=f"/api/v1/alert/{alert_id}", json=fields
+        )
 
     def bulk_update(self, fields: InputBulkUpdateAlert) -> None:
         """Update multiple alerts with the same values.
@@ -94,17 +95,22 @@ def bulk_update(self, fields: InputBulkUpdateAlert) -> None:
             "PATCH", path="/api/v1/alert/_bulk", json=fields
         )
 
-    def bulk_delete(self, ids: List[str]) -> None:
-        """Delete multiple alerts.
+    def promote_to_case(
+        self, alert_id: str, fields: InputPromoteAlert = {}
+    ) -> OutputCase:
+        """Promote an alert into a case.
 
         Args:
-            ids: The ids of the alerts to delete.
+            alert_id: The id of the alert.
+            fields: Override for the fields of the case created from the alert.
 
         Returns:
-            N/A
+            The case from the promoted alert.
         """
         return self._session.make_request(
-            "POST", path="/api/v1/alert/delete/_bulk", json={"ids": ids}
+            "POST",
+            path=f"/api/v1/alert/{alert_id}/case",
+            json=fields,
         )
 
     def follow(self, alert_id: str) -> None:
@@ -129,56 +135,93 @@ def unfollow(self, alert_id: str) -> None:
         """
         self._session.make_request("POST", path=f"/api/v1/alert/{alert_id}/unfollow")
 
-    def promote_to_case(
-        self, alert_id: str, fields: InputPromoteAlert = {}
-    ) -> OutputCase:
-        """Promote an alert into a case.
+    def merge_into_case(self, alert_id: str, case_id: str) -> OutputCase:
+        """Merge an alert into an existing case.
 
         Args:
-            alert_id: The id of the alert.
-            fields: Override for the fields of the case created from the alert.
+            alert_id: The id of the alert to merge.
+            case_id: The id of the case to merge the alert into.
 
         Returns:
-            The case from the promoted alert.
+            The case into which the alert was merged.
         """
         return self._session.make_request(
-            "POST",
-            path=f"/api/v1/alert/{alert_id}/case",
-            json=fields,
+            "POST", path=f"/api/v1/alert/{alert_id}/merge/{case_id}"
         )
 
-    def create_observable(
-        self,
-        alert_id: str,
-        observable: InputObservable,
-        observable_path: Optional[str] = None,
-    ) -> List[OutputObservable]:
-        """Create an observable in an alert.
+    def import_into_case(self, alert_id: str, case_id: str) -> OutputCase:
+        """Import alert observables and procedures into an existing case.
 
         Args:
-            alert_id: The id of the alert.
-            observable: The fields of the observable to create.
-            observable_path: Optional path in case of a file based observable.
+            alert_id: The id of the alert to merge.
+            case_id: The id of the case to merge the alert into.
 
         Returns:
-            The created alert observables.
+            The case into which the alert observables/procedures were imported.
         """
+        return self._session.make_request(
+            "POST", path=f"/api/v1/alert/{alert_id}/import/{case_id}"
+        )
 
-        kwargs = self._build_observable_kwargs(
-            observable=observable, observable_path=observable_path
+    def bulk_merge_into_case(self, case_id: str, alert_ids: List[str]) -> OutputCase:
+        """Merge an alert into an existing case.
+
+        Args:
+            case_id: The id of the case to merge the alerts into.
+            alert_ids: The list of alert ids to merge.
+
+        Returns:
+            The case into which the alerts were merged.
+        """
+        return self._session.make_request(
+            "POST",
+            path="/api/v1/alert/merge/_bulk",
+            json={"caseId": case_id, "alertIds": alert_ids},
+        )
+
+    def bulk_delete(self, ids: List[str]) -> None:
+        """Delete multiple alerts.
+
+        Args:
+            ids: The ids of the alerts to delete.
+
+        Returns:
+            N/A
+        """
+        return self._session.make_request(
+            "POST", path="/api/v1/alert/delete/_bulk", json={"ids": ids}
         )
+
+    def get_similar_observables(
+        self, alert_id: str, alert_or_case_id: str
+    ) -> List[OutputObservable]:
+        """Get similar observables between an alert and another alert or case.
+
+        Args:
+            alert_id: The id of the alert to use as base for observable similarity.
+            alert_or_case_id: The id of the alert/case to get similar observables from.
+
+        Returns:
+            The list of similar observables.
+        """
         return self._session.make_request(
-            "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
+            "GET",
+            path=f"/api/v1/alert/{alert_id}/similar/{alert_or_case_id}/observables",
         )
 
     def add_attachment(
-        self, alert_id: str, attachment_paths: List[str]
+        self,
+        alert_id: str,
+        attachment_paths: List[str],
+        can_rename: bool = True,
     ) -> List[OutputAttachment]:
         """Create an attachment in an alert.
 
         Args:
             alert_id: The id of the alert.
             attachment_paths: List of paths to the attachments to create.
+            can_rename: If set to True, the files can be renamed if they already exist
+                with the same name.
 
         Returns:
             The created alert attachments.
@@ -188,71 +231,81 @@ def add_attachment(
             for attachment_path in attachment_paths
         ]
         return self._session.make_request(
-            "POST", f"/api/v1/alert/{alert_id}/attachments", files=files
+            "POST",
+            f"/api/v1/alert/{alert_id}/attachments",
+            data={"canRename": can_rename},
+            files=files,
         )["attachments"]
 
-    def download_attachment(
-        self, alert_id: str, attachment_id: str, attachment_path: str
-    ) -> None:
-        """Download an alert attachment.
+    def delete_attachment(self, alert_id: str, attachment_id: str) -> None:
+        """Delete an alert attachment.
 
         Args:
             alert_id: The id of the alert.
             attachment_id: The id of the alert attachment.
-            attachment_path: The local path to download the attachment to.
 
         Returns:
             N/A
         """
+
         return self._session.make_request(
-            "GET",
-            path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}/download",
-            download_path=attachment_path,
+            "DELETE", path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}"
         )
 
-    def delete_attachment(self, alert_id: str, attachment_id: str) -> None:
-        """Delete an alert attachment.
+    def download_attachment(
+        self, alert_id: str, attachment_id: str, attachment_path: str
+    ) -> None:
+        """Download an alert attachment.
+
+        !!! warning
+            Deprecated: use [organisation.download_attachment]
+            [thehive4py.endpoints.organisation.OrganisationEndpoint.download_attachment]
+            instead
 
         Args:
             alert_id: The id of the alert.
             attachment_id: The id of the alert attachment.
+            attachment_path: The local path to download the attachment to.
 
         Returns:
             N/A
         """
 
-        return self._session.make_request(
-            "DELETE", path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}"
+        warnings.warn(
+            message=(
+                "Deprecated: use the organisation.download_attachment method instead"
+            ),
+            category=DeprecationWarning,
+            stacklevel=2,
         )
-
-    def merge_into_case(self, alert_id: str, case_id: str) -> OutputCase:
-        """Merge an alert into an existing case.
-
-        Args:
-            alert_id: The id of the alert to merge.
-            case_id: The id of the case to merge the alert into.
-
-        Returns:
-            The case into which the alert was merged.
-        """
         return self._session.make_request(
-            "POST", path=f"/api/v1/alert/{alert_id}/merge/{case_id}"
+            "GET",
+            path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}/download",
+            download_path=attachment_path,
         )
 
-    def bulk_merge_into_case(self, case_id: str, alert_ids: List[str]) -> OutputCase:
-        """Merge an alert into an existing case.
+    def create_observable(
+        self,
+        alert_id: str,
+        observable: InputObservable,
+        observable_path: Optional[str] = None,
+    ) -> List[OutputObservable]:
+        """Create an observable in an alert.
 
         Args:
-            case_id: The id of the case to merge the alerts into.
-            alert_ids: The list of alert ids to merge.
+            alert_id: The id of the alert.
+            observable: The fields of the observable to create.
+            observable_path: Optional path in case of a file based observable.
 
         Returns:
-            The case into which the alerts were merged.
+            The created alert observables.
         """
+
+        kwargs = self._build_observable_kwargs(
+            observable=observable, observable_path=observable_path
+        )
         return self._session.make_request(
-            "POST",
-            path="/api/v1/alert/merge/_bulk",
-            json={"caseId": case_id, "alertIds": alert_ids},
+            "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
         )
 
     def find(