feat(expectation): add LOGGING expectation support for collectors (#8746) (#374)

Merging, CI error regarding PR not linked to an issue is because the issue is in another repo.

Author: impolitepanda

pyoaev/apis/collector.py

@@ -25,6 +25,10 @@ class CollectorManager(GetMixin, ListMixin, CreateMixin, UpdateMixin, RESTManage
             # Source-declared author override for the collector's payloads and
             # contracts. When absent, the platform uses the collector name.
             "collector_author",
+            # Expectation types the collector can answer, e.g.
+            # ["DETECTION", "PREVENTION", "LOGGING"]. When absent, the platform
+            # assumes DETECTION and PREVENTION.
+            "collector_supported_expectation_types",
         ),
     )
 

pyoaev/apis/inject_expectation/inject_expectation.py

@@ -1,20 +1,61 @@
 import logging
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Type, TypeVar
 
 from pydantic import ValidationError
 
 from pyoaev import exceptions as exc
 from pyoaev.apis.inject_expectation.model import (
     DetectionExpectation,
     ExpectationTypeEnum,
+    LoggingExpectation,
     PreventionExpectation,
 )
+from pyoaev.apis.inject_expectation.model.expectation import Expectation
 from pyoaev.base import RESTManager, RESTObject
 from pyoaev.mixins import ListMixin, UpdateMixin
 from pyoaev.utils import RequiredOptional
 
 logger = logging.getLogger(__name__)
 
+ExpectationModel = TypeVar("ExpectationModel", bound=Expectation)
+
+# Expectation models built by expectations_models_for_source, by expectation type.
+# The /assets route only returns these two types.
+_ASSET_EXPECTATION_MODELS: Dict[str, Type[Expectation]] = {
+    ExpectationTypeEnum.Detection.value: DetectionExpectation,
+    ExpectationTypeEnum.Prevention.value: PreventionExpectation,
+}
+_LOGGING_EXPECTATION_MODELS: Dict[str, Type[LoggingExpectation]] = {
+    ExpectationTypeEnum.Logging.value: LoggingExpectation,
+}
+
+
+def _as_list_of_dicts(
+    raw: Any, path: str, label: str, item_label: str
+) -> List[Dict[str, Any]]:
+    """Checks that a server response is a JSON list of objects.
+
+    :param label: what the list holds, in the plural (e.g. "AI expectations")
+    :param item_label: the same, in the singular (e.g. "AI expectation")
+
+    :raises OpenAEVParsingError: if it is not
+    """
+    if not isinstance(raw, list):
+        raise exc.OpenAEVParsingError(
+            error_message=(
+                f"Expected a list of {label} from {path}, got {type(raw).__name__}"
+            )
+        )
+    for item in raw:
+        if not isinstance(item, dict):
+            raise exc.OpenAEVParsingError(
+                error_message=(
+                    f"Expected {item_label} objects (dicts) from {path}, "
+                    f"got a list element of type {type(item).__name__}"
+                )
+            )
+    return [item for item in raw]
+
 
 class InjectExpectation(RESTObject):
     _id_attr = "inject_expectation_id"
@@ -24,13 +65,21 @@ class InjectExpectationManager(ListMixin, UpdateMixin, RESTManager):
     _path = "/injects/expectations"
     _obj_cls = InjectExpectation
     _update_attrs = RequiredOptional(required=("collector_id", "result", "is_success"))
+    # Whether the missing LOGGING route of an older platform was already reported.
+    _logging_route_missing_reported = False
 
     @exc.on_http_error(exc.OpenAEVUpdateError)
     def expectations_assets_for_source(
         self, source_id: str, expiration_time: int = None, **kwargs: Any
-    ) -> Dict[str, Any]:
+    ) -> List[Dict[str, Any]]:
+        """Returns the DETECTION and PREVENTION expectations on assets not yet
+        filled for the given source.
+
+        :raises OpenAEVParsingError: if the server does not return a JSON list of dicts
+        """
         path = f"{self.path}/assets/" + source_id
-        result = self.openaev.http_get(
+        # The route returns a JSON list; http_get is annotated Union[Dict, Response].
+        raw: Any = self.openaev.http_get(
             path,
             query_data=(
                 {"expiration_time": expiration_time}
@@ -43,7 +92,7 @@ def expectations_assets_for_source(
             ),
             **kwargs,
         )
-        return result
+        return _as_list_of_dicts(raw, path, "asset expectations", "asset expectation")
 
     @exc.on_http_error(exc.OpenAEVListError)
     def ai_expectations_for_source(
@@ -60,27 +109,71 @@ def ai_expectations_for_source(
         """
         path = f"{self.path}/ai/{source_id}"
         # http_get is annotated Union[Dict, requests.Response]; widen to Any so the list
-        # validation below stays meaningful to type checkers, then build the result list
-        # explicitly to satisfy the declared return type.
+        # validation stays meaningful to type checkers.
         raw: Any = self.openaev.http_get(path, **kwargs)
-        if not isinstance(raw, list):
-            raise exc.OpenAEVParsingError(
-                error_message=(
-                    f"Expected a list of AI expectations from {path}, "
-                    f"got {type(raw).__name__}"
-                )
-            )
-        for item in raw:
-            if not isinstance(item, dict):
-                raise exc.OpenAEVParsingError(
-                    error_message=(
-                        f"Expected AI expectation objects (dicts) from {path}, "
-                        f"got a list element of type {type(item).__name__}"
-                    )
-                )
-        return [item for item in raw]
+        return _as_list_of_dicts(raw, path, "AI expectations", "AI expectation")
+
+    @exc.on_http_error(exc.OpenAEVListError)
+    def logging_expectations_for_source(
+        self, source_id: str, **kwargs: Any
+    ) -> List[Dict[str, Any]]:
+        """Returns the LOGGING expectations not yet filled for the given source.
+
+        Only collectors that declare LOGGING in their supported expectation
+        types receive them. An older platform without the LOGGING route answers
+        404: it is treated as "no LOGGING expectation", with a warning.
+
+        :param source_id: the identifier of the collector requesting expectations
+        :type source_id: str
+        :raises OpenAEVParsingError: if the server does not return a JSON list of dicts
+        :return: a list of logging expectation dicts
+        :rtype: list[dict]
+        """
+        path = f"{self.path}/logging/{source_id}"
+        try:
+            raw: Any = self.openaev.http_get(path, **kwargs)
+        except exc.OpenAEVHttpError as err:
+            if err.response_code != 404:
+                raise
+            self._report_missing_logging_route(path)
+            return []
+        return _as_list_of_dicts(
+            raw, path, "Logging expectations", "Logging expectation"
+        )
+
+    def _report_missing_logging_route(self, path: str) -> None:
+        if self._logging_route_missing_reported:
+            logger.debug("%s answered 404, no LOGGING expectation collected", path)
+            return
+        self._logging_route_missing_reported = True
+        logger.warning(
+            "%s answered 404: this OpenAEV platform does not support LOGGING "
+            "expectations yet, none will be collected",
+            path,
+        )
+
+    def logging_expectation_models_for_source(
+        self, source_id: str, **kwargs: Any
+    ) -> List[LoggingExpectation]:
+        """Returns the LOGGING expectations from OpenAEV that have had no result
+            yet from the source_id (e.g. collector), as models.
+
+        :param source_id: the identifier of the collector requesting expectations
+        :type source_id: str
+        :param kwargs: additional data to pass to the endpoint
+        :type kwargs: dict, optional
+
+        :return: a list of expectation objects
+        :rtype: list[LoggingExpectation]
+        """
+        return self._models_from_dicts(
+            self.logging_expectations_for_source(source_id=source_id, **kwargs),
+            _LOGGING_EXPECTATION_MODELS,
+        )
 
-    def expectations_models_for_source(self, source_id: str, **kwargs: Any):
+    def expectations_models_for_source(
+        self, source_id: str, **kwargs: Any
+    ) -> List[Expectation]:
         """Returns all expectations from OpenAEV that have had no result yet
             from the source_id (e.g. collector).
 
@@ -92,20 +185,36 @@ def expectations_models_for_source(self, source_id: str, **kwargs: Any):
         :return: a list of expectation objects
         :rtype: list[DetectionExpectation|PreventionExpectation]
         """
-        # TODO: we should implement a more clever mechanism to obtain
-        #   specialised Expectation instances rather than just if/elseing
-        #   through this list of possibilities.
-        expectations = []
-        for expectation_dict in self.expectations_assets_for_source(
-            source_id=source_id, **kwargs
-        ):
-            if (
-                expectation_dict["inject_expectation_type"]
-                == ExpectationTypeEnum.Detection.value
-            ):
-                expectation_class = DetectionExpectation
-            else:
-                expectation_class = PreventionExpectation
+        return self._models_from_dicts(
+            self.expectations_assets_for_source(source_id=source_id, **kwargs),
+            _ASSET_EXPECTATION_MODELS,
+        )
+
+    def _models_from_dicts(
+        self,
+        expectation_dicts: List[Dict[str, Any]],
+        model_by_type: Dict[str, Type[ExpectationModel]],
+    ) -> List[ExpectationModel]:
+        """Builds the model of each expectation dict, by expectation type.
+
+        A type missing from model_by_type is skipped with a warning, never
+        mapped to another model: a collector must not answer an expectation
+        type it does not handle.
+        """
+        expectations: List[ExpectationModel] = []
+        for expectation_dict in expectation_dicts:
+            expectation_type: str = (
+                expectation_dict.get("inject_expectation_type") or ""
+            )
+            expectation_class = model_by_type.get(expectation_type)
+            if expectation_class is None:
+                logger.warning(
+                    "Skipping expectation %s, its type %s is not one of %s",
+                    expectation_dict.get("inject_expectation_id"),
+                    expectation_type,
+                    ", ".join(model_by_type),
+                )
+                continue
             # One invalid expectation (e.g. an unknown signature type) must not
             # hide the valid ones from the collector.
             try:

pyoaev/apis/inject_expectation/model/__init__.py

@@ -1,7 +1,13 @@
 from .expectation import (
     DetectionExpectation,
     ExpectationTypeEnum,
+    LoggingExpectation,
     PreventionExpectation,
 )
 
-__all__ = ["DetectionExpectation", "ExpectationTypeEnum", "PreventionExpectation"]
+__all__ = [
+    "DetectionExpectation",
+    "ExpectationTypeEnum",
+    "LoggingExpectation",
+    "PreventionExpectation",
+]

pyoaev/apis/inject_expectation/model/expectation.py

@@ -14,6 +14,7 @@ class ExpectationTypeEnum(str, Enum):
 
     Detection = "DETECTION"
     Prevention = "PREVENTION"
+    Logging = "LOGGING"
     Vulnerability = "VULNERABILITY"
     Other = "other"
 
@@ -181,3 +182,16 @@ class PreventionExpectation(Expectation):
 
     success_label: str = "Prevented"
     failure_label: str = "Not Prevented"
+
+
+class LoggingExpectation(Expectation):
+    """An expectation that is specific to Logging, i.e. that is used
+    by OpenAEV to assert that an inject's execution was recorded by the
+    log sources of a security platform, whether it raised an alert or not.
+
+    Collectors only report successes for this type: "Not Logged" is written
+    by the platform when the expectation expires.
+    """
+
+    success_label: str = "Logged"
+    failure_label: str = "Not Logged"

pyoaev/apis/inject_expectation_trace.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from typing import Any, Dict, List
 
 from pyoaev import exceptions as exc
@@ -6,6 +7,19 @@
 from pyoaev.utils import RequiredOptional
 
 
+class TraceMatchType(str, Enum):
+    """Signal that matched a LOGGING expectation, sent as
+    inject_expectation_trace_match_type.
+
+    The platform drops an unknown value without error, so collectors should
+    use these members rather than raw strings.
+    """
+
+    MARKER = "marker"
+    SOURCE_IP = "source_ip"
+    TARGET_IP = "target_ip"
+
+
 class InjectExpectationTrace(RESTObject):
     _id_attr = "inject_expectation_trace_id"
 
@@ -21,6 +35,11 @@ class InjectExpectationTraceManager(CreateMixin, RESTManager):
             "inject_expectation_trace_alert_link",
             "inject_expectation_trace_date",
         ),
+        optional=(
+            # Signal that matched, for a LOGGING expectation: a TraceMatchType.
+            # Left out for the other types.
+            "inject_expectation_trace_match_type",
+        ),
     )
 
     @exc.on_http_error(exc.OpenAEVUpdateError)

pyoaev/contracts/contract_config.py

@@ -123,16 +123,17 @@ class ExpectationType(str, Enum):
     manual: str = "MANUAL"
     prevention: str = "PREVENTION"
     detection: str = "DETECTION"
+    logging: str = "LOGGING"
     vulnerability: str = "VULNERABILITY"
 
 
 class SecurityPlatformType(str, Enum):
     """Categories of security platform expected to fulfil a technical expectation.
 
     When an expectation declares one or more of these, the platform focuses the
-    technical (DETECTION / PREVENTION / VULNERABILITY) result on collectors of
-    those types only (instead of every connected security platform). An empty
-    list means "any security platform".
+    technical (DETECTION / PREVENTION / LOGGING / VULNERABILITY) result on
+    collectors of those types only (instead of every connected security
+    platform). An empty list means "any security platform".
     """
 
     EDR: str = "EDR"
@@ -156,7 +157,8 @@ class Expectation:
     expectation_is_multi_selectable: bool = False
     # Security platform types expected to fulfil this expectation. Empty = any
     # platform (unchanged behaviour). Typically set for technical expectations
-    # (DETECTION / PREVENTION / VULNERABILITY), left empty for MANUAL ones.
+    # (DETECTION / PREVENTION / LOGGING / VULNERABILITY), left empty for MANUAL
+    # ones.
     expectation_expected_security_platform_types: List[SecurityPlatformType] = field(
         default_factory=list
     )