add basic injector_contract client

Signed-off-by: Antoine MAZEAS <antoine.mazeas@filigran.io>

Author: antoinemzs

pyobas/apis/__init__.py

@@ -7,6 +7,7 @@
 from .inject_expectation import *  # noqa: F401,F403
 from .inject_expectation_trace import *  # noqa: F401,F403
 from .injector import *  # noqa: F401,F403
+from .injector_contract import *  # noqa: F401,F403
 from .kill_chain_phase import *  # noqa: F401,F403
 from .me import *  # noqa: F401,F403
 from .organization import *  # noqa: F401,F403

pyobas/apis/injector_contract.py

@@ -0,0 +1,66 @@
+import json
+from typing import Any, Dict
+
+from pyobas import exceptions as exc
+from pyobas.apis.inputs.search import InjectorContractSearchPaginationInput
+from pyobas.base import RESTManager, RESTObject
+from pyobas.mixins import CreateMixin, DeleteMixin, UpdateMixin
+from pyobas.utils import RequiredOptional
+
+
+class InjectorContract(RESTObject):
+    pass
+
+
+class InjectorContractManager(CreateMixin, UpdateMixin, DeleteMixin, RESTManager):
+    _path = "/injector_contracts"
+    _obj_cls = InjectorContract
+    _create_attrs = (
+        RequiredOptional(
+            required=(
+                "contract_content",
+                "contract_id",
+                "contract_labels",
+                "injector_id",
+            ),
+            optional=(
+                "contract_attack_patterns_ids",
+                "contract_attack_patterns_external_ids",
+                "contract_manual",
+                "contract_platforms",
+                "external_contract_id",
+                "is_atomic_testing",
+            ),
+        ),
+    )
+    _update_attrs = (
+        RequiredOptional(
+            required=(
+                "contract_content",
+                "contract_labels",
+            ),
+            optional=(
+                "contract_attack_patterns_ids",
+                "contract_manual",
+                "contract_platforms",
+                "is_atomic_testing",
+            ),
+        ),
+    )
+
+    @exc.on_http_error(exc.OpenBASUpdateError)
+    def search(
+        self, input: InjectorContractSearchPaginationInput, **kwargs: Any
+    ) -> Dict[str, Any]:
+        path = f"{self.path}/search"
+        # force the serialisation here since we only need a naive serialisation to json
+        result = self.openbas.http_post(
+            path, post_data=json.dumps(input, default=lambda x: x.__dict__), **kwargs
+        )
+        return result
+
+    @exc.on_http_error(exc.OpenBASUpdateError)
+    def upsert(self, user: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]:
+        path = f"{self.path}/upsert"
+        result = self.openbas.http_post(path, post_data=user, **kwargs)
+        return result

pyobas/apis/inputs/__init__.py

@@ -0,0 +1,45 @@
+from typing import Dict, List
+
+
+class Filter:
+    def __init__(self, key: str, mode: str, operator: str, values: List[str]):
+        self.key = key
+        self.mode = mode
+        self.operator = operator
+        self.values = values
+
+
+class FilterGroup:
+    def __init__(self, mode: str, filters: List[Filter]):
+        self.mode = mode
+        self.filters = filters
+
+
+class SearchPaginationInput:
+    def __init__(
+        self,
+        page: int,
+        size: int,
+        filter_group: FilterGroup,
+        text_search: str,
+        sorts: Dict[str, str],
+    ):
+        self.size = size
+        self.page = page
+        self.filter_group = filter_group
+        self.text_search = text_search
+        self.sorts = sorts
+
+
+class InjectorContractSearchPaginationInput(SearchPaginationInput):
+    def __init__(
+        self,
+        page: int,
+        size: int,
+        filter_group: FilterGroup,
+        text_search: str,
+        sorts: Dict[str, str],
+        full_details: bool = True,
+    ):
+        super().__init__(page, size, filter_group, text_search, sorts)
+        self.full_details = full_details

pyobas/apis/inputs/search.py

@@ -61,6 +61,7 @@ def __init__(
         self.collector = apis.CollectorManager(self)
         self.cve = apis.CveManager(self)
         self.inject = apis.InjectManager(self)
+        self.injector_contract = apis.InjectorContractManager(self)
         self.document = apis.DocumentManager(self)
         self.kill_chain_phase = apis.KillChainPhaseManager(self)
         self.attack_pattern = apis.AttackPatternManager(self)

pyobas/client.py

@@ -216,3 +216,27 @@ def create(
             assert not isinstance(server_data, requests.Response)
             assert self._obj_cls is not None
         return self._obj_cls(self, server_data)
+
+
+class DeleteMixin(_RestManagerBase):
+    _computed_path: Optional[str]
+    _from_parent_attrs: Dict[str, Any]
+    _obj_cls: Optional[Type[base.RESTObject]]
+    _parent: Optional[base.RESTObject]
+    _parent_attrs: Dict[str, Any]
+    _path: Optional[str]
+    openbas: pyobas.OpenBAS
+
+    @exc.on_http_error(exc.OpenBASCreateError)
+    def delete(
+        self, id: Optional[Union[str, int]] = None, **kwargs: Any
+    ) -> requests.Response:
+        if id is None:
+            path = self.path
+        else:
+            path = f"{self.path}/{utils.EncodedId(id)}"
+
+        result = self.openbas.http_delete(path, **kwargs)
+        if TYPE_CHECKING:
+            assert isinstance(result, requests.Response)
+        return result

pyobas/mixins.py

@@ -0,0 +1,63 @@
+import json
+from unittest import TestCase, main, mock
+from unittest.mock import ANY
+
+from pyobas import OpenBAS
+from pyobas.apis.inputs.search import (
+    Filter,
+    FilterGroup,
+    InjectorContractSearchPaginationInput,
+)
+
+
+def create_mock_api_client():
+    return mock.MagicMock()
+
+
+def mock_response(**kwargs):
+    class MockResponse:
+        def __init__(self, json_data, status_code):
+            self.json_data = json_data
+            self.status_code = status_code
+            self.history = None
+            self.content = None
+            self.headers = {"Content-Type": "application/json"}
+
+        def json(self):
+            return self.json_data
+
+    return MockResponse(None, 200)
+
+
+class TestInjectorContract(TestCase):
+    @mock.patch("requests.Session.request", side_effect=mock_response)
+    def test_search_input_correctly_serialised(self, mock_request):
+        api_client = OpenBAS("url", "token")
+
+        search_input = InjectorContractSearchPaginationInput(
+            0,
+            20,
+            FilterGroup("or", [Filter("prop", "and", "eq", ["titi", "toto"])]),
+            None,
+            None,
+        )
+
+        expected_json = json.dumps(search_input, default=lambda x: x.__dict__)
+        api_client.injector_contract.search(search_input)
+
+        mock_request.assert_called_once_with(
+            method="post",
+            url="url/api/injector_contracts/search",
+            params={},
+            data=None,
+            timeout=None,
+            stream=False,
+            verify=True,
+            json=expected_json,
+            headers=ANY,
+            auth=ANY,
+        )
+
+
+if __name__ == "__main__":
+    main()