[client] add SignatureType supporting classes (#11)

Signed-off-by: Antoine MAZEAS <[email protected]>

Author: antoinemzs

.circleci/config.yml

@@ -32,6 +32,21 @@ jobs:
       - ms-teams/report:
           only_on_fail: true
           webhook_url: $MS_TEAMS_WEBHOOK_URL
+  test:
+    docker:
+      - image: cimg/python:3.13
+    working_directory: ~/repo
+    steps:
+      - checkout
+      - run:
+          name: install dependencies
+          command: pip3 install -r requirements.txt --user
+      - run:
+          name: install test-dependencies
+          command: pip3 install -r test-requirements.txt --user
+      - run:
+          name: run tests
+          command: python -m unittest
   linter:
     docker:
       - image: alpine/flake8
@@ -144,6 +159,10 @@ workflows:
           filters:
             tags:
               only: /.*/
+      - test:
+          filters:
+            tags:
+              only: /.*/
       - build:
           filters:
             tags:

pyobas/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-__version__ = "1.9.0"
+__version__ = "1.10.0"
 
 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.9.0"
+__version__ = "1.10.0"

pyobas/helpers.py

@@ -310,7 +310,11 @@ def get_conf(self, variable, is_number=None, default=None, required=None):
 
 class OpenBASCollectorHelper:
     def __init__(
-        self, config: OpenBASConfigHelper, icon, security_platform_type=None
+        self,
+        config: OpenBASConfigHelper,
+        icon,
+        security_platform_type=None,
+        connect_run_and_terminate: bool = False,
     ) -> None:
         self.config_helper = config
         self.api = OpenBAS(
@@ -351,11 +355,10 @@ def __init__(
 
         collector_icon = (icon_name, open(icon, "rb"), "image/png")
         self.api.collector.create(self.config, collector_icon)
-        self.connect_run_and_terminate = False
         # self.api.injector.create(self.config)
         self.scheduler = sched.scheduler(time.time, time.sleep)
         # Start ping thread
-        if not self.connect_run_and_terminate:
+        if not connect_run_and_terminate:
             self.ping = PingAlive(
                 self.api, self.config, self.collector_logger, "collector"
             )

pyobas/signatures/__init__.py

@@ -0,0 +1,12 @@
+from pyobas.exceptions import OpenBASError
+from pyobas.signatures.types import MatchTypes
+
+
+class SignatureMatch:
+    def __init__(self, match_type: MatchTypes, match_score: int | None):
+        if match_score is None and match_type != MatchTypes.MATCH_TYPE_SIMPLE:
+            raise OpenBASError(
+                f"Match type {match_type} requires score to be set, found score = {match_score}"
+            )
+        self.match_type = match_type
+        self.match_score = match_score

pyobas/signatures/signature_match.py

@@ -0,0 +1,32 @@
+from pyobas.signatures.signature_match import SignatureMatch
+from pyobas.signatures.types import MatchTypes, SignatureTypes
+
+
+class SignatureType:
+    def __init__(
+        self,
+        label: SignatureTypes,
+        match_type: MatchTypes = MatchTypes.MATCH_TYPE_SIMPLE,
+        match_score: int = None,
+    ):
+        self.label = label
+        self.match_policy = SignatureMatch(match_type, match_score)
+
+    # provided some `data`, formats a dictionary specifying the matching
+    # policy to use by the helper to match expected signatures (from expectations)
+    # with actual, alert signatures (from the security software)
+    # Output: {
+    #   "type": str,
+    #   "data": any,
+    #   "score": (optional) int
+    # }
+    def make_struct_for_matching(self, data):
+        struct = {
+            "type": self.match_policy.match_type,
+            "data": data,
+        }
+
+        if self.match_policy.match_score is not None:
+            struct["score"] = self.match_policy.match_score
+
+        return struct

pyobas/signatures/signature_type.py

@@ -0,0 +1,11 @@
+from enum import Enum
+
+
+class MatchTypes(Enum):
+    MATCH_TYPE_FUZZY = "fuzzy"
+    MATCH_TYPE_SIMPLE = "simple"
+
+
+class SignatureTypes(Enum):
+    SIG_TYPE_PARENT_PROCESS_NAME = "parent_process_name"
+    SIG_TYPE_HOSTNAME = "hostname"