[pyobas] Update mandatory contract properties

.circleci/config.yml

@@ -1,7 +1,7 @@
 ---
 version: 2.1
 orbs:
-  slack: circleci/slack@5.1.1
+  slack: circleci/slack@5.2.0
 jobs:
   ensure_formatting:
     docker:

pyobas/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-__version__ = "1.18.7"
+__version__ = "1.18.9"
 
 from pyobas._version import (  # noqa: F401
     __author__,

pyobas/_version.py

@@ -3,4 +3,4 @@
 __email__ = "[email protected]"
 __license__ = "Apache 2.0"
 __title__ = "python-openbas"
-__version__ = "1.18.7"
+__version__ = "1.18.9"

pyobas/apis/__init__.py

@@ -1,5 +1,6 @@
 from .attack_pattern import *  # noqa: F401,F403
 from .collector import *  # noqa: F401,F403
+from .cve import *  # noqa: F401,F403
 from .document import *  # noqa: F401,F403
 from .endpoint import *  # noqa: F401,F403
 from .inject import *  # noqa: F401,F403

pyobas/apis/collector.py

@@ -1,3 +1,6 @@
+from typing import Any, Dict
+
+from pyobas import exceptions as exc
 from pyobas.base import RESTManager, RESTObject
 from pyobas.mixins import CreateMixin, GetMixin, ListMixin, UpdateMixin
 from pyobas.utils import RequiredOptional
@@ -18,3 +21,9 @@ class CollectorManager(GetMixin, ListMixin, CreateMixin, UpdateMixin, RESTManage
             "collector_period",
         )
     )
+
+    @exc.on_http_error(exc.OpenBASUpdateError)
+    def get(self, collector_id: str, **kwargs: Any) -> Dict[str, Any]:
+        path = f"{self.path}/" + collector_id
+        result = self.openbas.http_get(path, **kwargs)
+        return result

pyobas/apis/cve.py

@@ -0,0 +1,18 @@
+from typing import Any, Dict
+
+from pyobas import exceptions as exc
+from pyobas.base import RESTManager, RESTObject
+
+
+class Cve(RESTObject):
+    _id_attr = "cve_id"
+
+
+class CveManager(RESTManager):
+    _path = "/cves"
+
+    @exc.on_http_error(exc.OpenBASUpdateError)
+    def upsert(self, data: Dict[str, Any], **kwargs: Any) -> Dict[str, Any]:
+        path = f"{self.path}/bulk"
+        result = self.openbas.http_post(path, post_data=data, **kwargs)
+        return result

pyobas/client.py

@@ -59,6 +59,7 @@ def __init__(
         self.organization = apis.OrganizationManager(self)
         self.injector = apis.InjectorManager(self)
         self.collector = apis.CollectorManager(self)
+        self.cve = apis.CveManager(self)
         self.inject = apis.InjectManager(self)
         self.document = apis.DocumentManager(self)
         self.kill_chain_phase = apis.KillChainPhaseManager(self)

pyobas/contracts/contract_config.py

@@ -2,7 +2,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import List
+from typing import List, Dict
 
 from pyobas import utils
 from pyobas.contracts.contract_utils import ContractCardinality, ContractVariable
@@ -73,11 +73,12 @@ class ContractElement(ABC):
     key: str
     label: str
     type: str = field(default="", init=False)
-    mandatoryGroups: List[str] = None
-    mandatoryConditionField: str = None
-    mandatoryConditionValue: str = None
-    linkedFields: List["ContractElement"] = field(default_factory=list)
-    linkedValues: List[str] = field(default_factory=list)
+    mandatoryGroups: List[str] = field(default_factory=list)
+    mandatoryConditionFields: List[str] = field(default_factory=list)
+    mandatoryConditionValues: Dict[str, any] = field(default_factory=list)
+    visibleConditionFields: List[str] = field(default_factory=list)
+    visibleConditionValues: Dict[str, any] = field(default_factory=list)
+    linkedFields: List[str] = field(default_factory=list)
     mandatory: bool = False
     readOnly: bool = False
 

pyobas/signatures/types.py

@@ -8,9 +8,10 @@ class MatchTypes(str, Enum):
 
 class SignatureTypes(str, Enum):
     SIG_TYPE_PARENT_PROCESS_NAME = "parent_process_name"
-    SIG_TYPE_HOSTNAME = "hostname"
-    SIG_TYPE_PROCESS_NAME = "process_name"
-    SIG_TYPE_COMMAND_LINE = "command_line"
-    SIG_TYPE_FILE_NAME = "file_name"
-    SIG_TYPE_IPV4 = "ipv4_address"
-    SIG_TYPE_IPV6 = "ipv6_address"
+    SIG_TYPE_SOURCE_IPV4_ADDRESS = "source_ipv4_address"
+    SIG_TYPE_SOURCE_IPV6_ADDRESS = "source_ipv6_address"
+    SIG_TYPE_TARGET_IPV4_ADDRESS = "target_ipv4_address"
+    SIG_TYPE_TARGET_IPV6_ADDRESS = "target_ipv6_address"
+    SIG_TYPE_TARGET_HOSTNAME_ADDRESS = "target_hostname_address"
+    SIG_TYPE_START_DATE = "start_date"
+    SIG_TYPE_END_DATE = "end_date"

pyproject.toml

@@ -33,11 +33,11 @@ dependencies = [
     "PyYAML (>=6.0,<6.1)",
     "pydantic (>=2.11.3,<2.12.0)",
     "requests (>=2.32.3,<2.33.0)",
-    "setuptools (>=80.4.0,<80.5.0)",
+    "setuptools (>=80.9.0,<80.10.0)",
     "cachetools (>=5.5.0,<5.6.0)",
-    "prometheus-client (>=0.21.1,<0.22.0)",
-    "opentelemetry-api (>=1.32.0,<1.33.0)",
-    "opentelemetry-sdk (>=1.32.0,<1.33.0)",
+    "prometheus-client (>=0.22.1,<0.23.0)",
+    "opentelemetry-api (>=1.35.0,<1.36.0)",
+    "opentelemetry-sdk (>=1.35.0,<1.36.0)",
     # OpenBAS,
     "requests-toolbelt (>=1.0.0,<1.1.0)",
     "dataclasses-json (>=0.6.4,<0.7.0)",

test/apis/expectation/test_expectation.py

@@ -126,7 +126,7 @@ def test_when_no_expectation_signature_is_relevant_match_alert_return_false(self
 
         relevant_signature_types = [
             SignatureType(
-                label=SignatureTypes.SIG_TYPE_HOSTNAME,
+                label=SignatureTypes.SIG_TYPE_TARGET_HOSTNAME_ADDRESS,
                 match_type=MatchTypes.MATCH_TYPE_SIMPLE,
             )
         ]
@@ -251,7 +251,10 @@ def test_when_relevant_signatures_when_alert_data_missing_for_some_relevant_sign
                         "type": SignatureTypes.SIG_TYPE_PARENT_PROCESS_NAME,
                         "value": "parent.exe",
                     },
-                    {"type": SignatureTypes.SIG_TYPE_FILE_NAME, "value": "filename"},
+                    {
+                        "type": SignatureTypes.SIG_TYPE_SOURCE_IPV4_ADDRESS,
+                        "value": "231.102.107.38",
+                    },
                 ],
             },
             api_client=create_mock_api_client(),
@@ -263,7 +266,7 @@ def test_when_relevant_signatures_when_alert_data_missing_for_some_relevant_sign
             match_score=95,
         )
         file_name_signature_type = SignatureType(
-            label=SignatureTypes.SIG_TYPE_FILE_NAME,
+            label=SignatureTypes.SIG_TYPE_SOURCE_IPV4_ADDRESS,
             match_type=MatchTypes.MATCH_TYPE_FUZZY,
             match_score=95,
         )
@@ -294,8 +297,8 @@ def test_when_relevant_signatures_when_some_alert_data_dont_match_return_false(
                         "value": "parent.exe",
                     },
                     {
-                        "type": SignatureTypes.SIG_TYPE_FILE_NAME,
-                        "value": "some_file.odt",
+                        "type": SignatureTypes.SIG_TYPE_SOURCE_IPV4_ADDRESS,
+                        "value": "108.134.173.48",
                     },
                 ],
             },
@@ -308,7 +311,7 @@ def test_when_relevant_signatures_when_some_alert_data_dont_match_return_false(
             match_score=95,
         )
         file_name_signature_type = SignatureType(
-            label=SignatureTypes.SIG_TYPE_FILE_NAME,
+            label=SignatureTypes.SIG_TYPE_SOURCE_IPV4_ADDRESS,
             match_type=MatchTypes.MATCH_TYPE_FUZZY,
             match_score=95,
         )

test/signatures/test_signature_type.py

@@ -6,7 +6,7 @@
 
 class TestSignatureType(unittest.TestCase):
     def test_make_struct_create_expected_struct_for_simple_sig_type(self):
-        simple_signature_type_label = SignatureTypes.SIG_TYPE_HOSTNAME
+        simple_signature_type_label = SignatureTypes.SIG_TYPE_TARGET_HOSTNAME_ADDRESS
         simple_signature_type = SignatureType(
             label=simple_signature_type_label, match_type=MatchTypes.MATCH_TYPE_SIMPLE
         )
@@ -19,7 +19,7 @@ def test_make_struct_create_expected_struct_for_simple_sig_type(self):
         self.assertFalse("score" in simple_struct.keys())
 
     def test_make_struct_create_expected_struct_for_fuzzy_sig_type(self):
-        fuzzy_signature_type_label = SignatureTypes.SIG_TYPE_HOSTNAME
+        fuzzy_signature_type_label = SignatureTypes.SIG_TYPE_TARGET_HOSTNAME_ADDRESS
         fuzzy_signature_type_score = 50
         fuzzy_signature_type = SignatureType(
             label=fuzzy_signature_type_label,
@@ -37,7 +37,7 @@ def test_make_struct_create_expected_struct_for_fuzzy_sig_type(self):
     def test_make_struct_create_expected_struct_for_fuzzy_sig_type_when_score_is_0(
         self,
     ):
-        fuzzy_signature_type_label = SignatureTypes.SIG_TYPE_HOSTNAME
+        fuzzy_signature_type_label = SignatureTypes.SIG_TYPE_TARGET_HOSTNAME_ADDRESS
         fuzzy_signature_type_score = 0
         fuzzy_signature_type = SignatureType(
             label=fuzzy_signature_type_label,