#442 - Review observable endpoints

Author: Kamforka

tests/test_observable_endpoint.py

@@ -3,6 +3,8 @@
 from typing import List
 
 import pytest
+
+from tests.utils import TestConfig
 from thehive4py import TheHiveApi
 from thehive4py.errors import TheHiveError
 from thehive4py.query.sort import Asc
@@ -16,7 +18,7 @@
 
 
 class TestObservableEndpoint:
-    def test_create_in_alert_and_get(
+    def test_create_single_in_alert_and_get(
         self, thehive: TheHiveApi, test_alert: OutputAlert
     ):
         created_observable = thehive.observable.create_in_alert(
@@ -33,6 +35,25 @@ def test_create_in_alert_and_get(
         )
         assert created_observable == fetched_observable
 
+    def test_create_multiple_in_alert(
+        self, thehive: TheHiveApi, test_alert: OutputAlert
+    ):
+
+        observable_count = 3
+        created_observables = thehive.observable.create_in_alert(
+            alert_id=test_alert["_id"],
+            observable={
+                "dataType": "domain",
+                "data": [f"{i}.example.com" for i in range(observable_count)],
+                "message": "test observable",
+            },
+        )
+
+        fetched_observables = thehive.alert.find_observables(alert_id=test_alert["_id"])
+
+        for created_observable in created_observables:
+            assert created_observable in fetched_observables
+
     def test_create_in_alert_from_file_and_download_as_zip(
         self, thehive: TheHiveApi, test_alert: OutputAlert, tmp_path: Path
     ):
@@ -66,7 +87,9 @@ def test_create_in_alert_from_file_and_download_as_zip(
             with archive_fp.open(observable_filename, pwd=b"malware") as downloaded_fp:
                 assert downloaded_fp.read().decode() == observable_content
 
-    def test_create_in_case_and_get(self, thehive: TheHiveApi, test_case: OutputCase):
+    def test_create_single_in_case_and_get(
+        self, thehive: TheHiveApi, test_case: OutputCase
+    ):
         created_observable = thehive.observable.create_in_case(
             case_id=test_case["_id"],
             observable={
@@ -81,6 +104,23 @@ def test_create_in_case_and_get(self, thehive: TheHiveApi, test_case: OutputCase
         )
         assert created_observable == fetched_observable
 
+    def test_create_multiple_in_case(self, thehive: TheHiveApi, test_case: OutputCase):
+
+        observable_count = 3
+        created_observables = thehive.observable.create_in_case(
+            case_id=test_case["_id"],
+            observable={
+                "dataType": "domain",
+                "data": [f"{i}.example.com" for i in range(observable_count)],
+                "message": "test observable",
+            },
+        )
+
+        fetched_observables = thehive.case.find_observables(case_id=test_case["_id"])
+
+        for created_observable in created_observables:
+            assert created_observable in fetched_observables
+
     def test_create_in_case_from_file_and_download_as_is(
         self, thehive: TheHiveApi, test_case: OutputCase, tmp_path: Path
     ):
@@ -153,25 +193,25 @@ def test_bulk_update(
             for key, value in expected_fields.items():
                 assert updated_task.get(key) == value
 
-    @pytest.mark.skip(
-        reason="documentation is unclear and implementation might be changed"
-    )
     def test_share_and_unshare(
-        self, thehive: TheHiveApi, test_observable: OutputObservable
+        self,
+        thehive: TheHiveApi,
+        test_observable: OutputObservable,
+        test_config: TestConfig,
     ):
-        organisation = "share-org"
 
         thehive.observable.share(
-            observable_id=test_observable["_id"], organisations=[organisation]
-        )
-        assert (
-            len(thehive.observable.list_shares(observable_id=test_observable["_id"]))
-            == 1
+            observable_id=test_observable["_id"], organisations=[test_config.main_org]
         )
 
-        thehive.observable.unshare(
-            observable_id=test_observable["_id"], organisations=[organisation]
-        )
+        # TODO: test `unshare` once a second organisation is allowed by the license
+        # thehive.observable.unshare(
+        #     observable_id=test_observable["_id"], organisations=[test_config.main_org]
+        # )
+
+        # TODO: test `list_shares` better once a second organisation is
+        # allowed by the license
+
         assert (
             len(thehive.observable.list_shares(observable_id=test_observable["_id"]))
             == 0

thehive4py/endpoints/observable.py

@@ -15,77 +15,198 @@
 
 
 class ObservableEndpoint(EndpointBase):
-    def create_in_alert(
+    def create_in_case(
         self,
-        alert_id: str,
+        case_id: str,
         observable: InputObservable,
         observable_path: Optional[str] = None,
     ) -> List[OutputObservable]:
+        """Create one or more observables in a case.
+
+        Args:
+            case_id: The id of the case.
+            observable: The fields of the observable to create.
+            observable_path: Optional path in case of file based observables.
+
+        Returns:
+            The created case observables.
+        """
         kwargs = self._build_observable_kwargs(
             observable=observable, observable_path=observable_path
         )
         return self._session.make_request(
-            "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
+            "POST", path=f"/api/v1/case/{case_id}/observable", **kwargs
         )
 
-    def create_in_case(
+    def create_in_alert(
         self,
-        case_id: str,
+        alert_id: str,
         observable: InputObservable,
         observable_path: Optional[str] = None,
     ) -> List[OutputObservable]:
+        """Create one or more observables in an alert.
+
+        Args:
+            alert_id: The id of the alert.
+            observable: The fields of the observable to create.
+            observable_path: Optional path in case of file based observables.
+
+        Returns:
+            The created alert observables.
+        """
         kwargs = self._build_observable_kwargs(
             observable=observable, observable_path=observable_path
         )
         return self._session.make_request(
-            "POST", path=f"/api/v1/case/{case_id}/observable", **kwargs
+            "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
         )
 
     def get(self, observable_id: str) -> OutputObservable:
+        """Get an observable by id.
+
+        Args:
+            observable_id: The id of the observable.
+
+        Returns:
+            The observable specified by the id.
+        """
         return self._session.make_request(
             "GET", path=f"/api/v1/observable/{observable_id}"
         )
 
     def delete(self, observable_id: str) -> None:
+        """Delete an observable.
+
+        Args:
+            observable_id: The id of the observable.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "DELETE", path=f"/api/v1/observable/{observable_id}"
         )
 
     def update(self, observable_id: str, fields: InputUpdateObservable) -> None:
+        """Update an observable.
+
+        Args:
+            observable_id: The id of the observable.
+            fields: The fields of the observable to update.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "PATCH", path=f"/api/v1/observable/{observable_id}", json=fields
         )
 
     def bulk_update(self, fields: InputBulkUpdateObservable) -> None:
+        """Update multiple observables with the same values.
+
+        Args:
+            fields: The ids and the fields of the observables to update.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "PATCH", path="/api/v1/observable/_bulk", json=fields
         )
 
+    def download_attachment(
+        self,
+        observable_id: str,
+        attachment_id: str,
+        observable_path: str,
+        as_zip: bool = False,
+    ) -> None:
+        """Download an observable attachment.
+
+        Args:
+            observable_id: The id of the observable.
+            attachment_id: The id of the observable attachment.
+            observable_path: The local path to download the observable attachment to.
+            as_zip: If `True`, the attachment will be sent as a zip file with a
+                password. Default password is 'malware'
+
+        Returns:
+            N/A
+        """
+        return self._session.make_request(
+            "GET",
+            path=(
+                f"/api/v1/observable/{observable_id}"
+                f"/attachment/{attachment_id}/download"
+            ),
+            params={"asZip": as_zip},
+            download_path=observable_path,
+        )
+
+    def list_shares(self, observable_id: str) -> List[OutputShare]:
+        """List all organisation shares of an observable.
+
+        Args:
+            observable_id: The id of the observable.
+
+        Returns:
+            The list of organisation shares of the observable.
+        """
+        return self._session.make_request(
+            "GET", path=f"/api/v1/case/{observable_id}/shares"
+        )
+
     def share(self, observable_id: str, organisations: List[str]) -> None:
+        """Share the observable with other organisations.
+
+        The case that owns the observable must already be shared with the target
+        organisations.
+
+        Args:
+            observable_id: The id of the observable.
+            organisations: The list of organisation names or ids.
+
+        Returns:
+            The list of organisation shares of the observable.
+        """
         return self._session.make_request(
             "POST",
             path=f"/api/v1/observable/{observable_id}/shares",
             json={"organisations": organisations},
         )
 
     def unshare(self, observable_id: str, organisations: List[str]) -> None:
+        """Unshare an observable from other organisations.
+
+        Args:
+            observable_id: The id of the observable.
+            organisations: The list of organisation names or ids.
+
+        Returns:
+            N/A
+        """
         return self._session.make_request(
             "DELETE",
             path=f"/api/v1/observable/{observable_id}/shares",
             json={"organisations": organisations},
         )
 
-    def list_shares(self, observable_id: str) -> List[OutputShare]:
-        return self._session.make_request(
-            "GET", path=f"/api/v1/case/{observable_id}/shares"
-        )
-
     def find(
         self,
         filters: Optional[FilterExpr] = None,
         sortby: Optional[SortExpr] = None,
         paginate: Optional[Paginate] = None,
     ) -> List[OutputObservable]:
+        """Find multiple observables.
+
+        Args:
+            filters: The filter expressions to apply in the query.
+            sortby: The sort expressions to apply in the query.
+            paginate: The pagination experssion to apply in the query.
+
+        Returns:
+            The list of observables matched by the query or an empty list.
+        """
         query: QueryExpr = [
             {"_name": "listObservable"},
             *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
@@ -99,6 +220,14 @@ def find(
         )
 
     def count(self, filters: Optional[FilterExpr] = None) -> int:
+        """Count observables.
+
+        Args:
+            filters: The filter expressions to apply in the query.
+
+        Returns:
+            The count of observables matched by the query.
+        """
         query: QueryExpr = [
             {"_name": "listObservable"},
             *self._build_subquery(filters=filters),
@@ -111,20 +240,3 @@ def count(self, filters: Optional[FilterExpr] = None) -> int:
             params={"name": "observable.count"},
             json={"query": query},
         )
-
-    def download_attachment(
-        self,
-        observable_id: str,
-        attachment_id: str,
-        observable_path: str,
-        as_zip=False,
-    ) -> None:
-        return self._session.make_request(
-            "GET",
-            path=(
-                f"/api/v1/observable/{observable_id}"
-                f"/attachment/{attachment_id}/download"
-            ),
-            params={"asZip": as_zip},
-            download_path=observable_path,
-        )

thehive4py/endpoints/user.py

@@ -113,7 +113,7 @@ def delete(self, user_id: str, organisation: Optional[str] = None) -> None:
 
         Args:
             user_id: The id of the user.
-            organisation: The organisation from which to delete the user from. Optional.
+            organisation: The organisation from which the user should be deleted.
 
         Returns:
             N/A