Add support for data filtering in webhook forwarder and introduce BasicFilter:

- Implement `match_filters` method in `webhook.py` for conditional forwarding based on filters.
- Extend configuration model with a `filters` field and the `BasicFilter` class for attribute-based filtering.
- Refactor filter application logic in `immediate`, `bulk`, and `periodic` modes.
- Add and adjust related test cases for filtering and configuration validation.

Author: U039b

mongoose/core/watchdogs.py

@@ -53,7 +53,7 @@ def on_any_event(self, event: FileSystemEvent):
             return
         elif event.event_type == "created":
             new_configuration = self._load_configuration(src_file)
-            if new_configuration:
+            if new_configuration and new_configuration.enable:
                 created.append(new_configuration)
         # Not supported yet
         # elif event.event_type == "modified":

mongoose/forward/webhook.py

@@ -134,6 +134,24 @@ def start(self):
         super().start()
         logger.info(f"Forwarding to {self.config.url} [mode={self.config.mode}]")
 
+    def match_filters(self, data):
+        """Check if data matches any configured filters.
+        Filters are applied using OR logic: if any filter matches, the data passes.
+        If no filters are configured, all data is allowed through.
+
+        Args:
+            data: The data object to check against configured filters.
+
+        Returns:
+            True if data passes filtering (matches at least one filter or no filters configured),
+            False otherwise.
+        """
+        matches = len(self.config.filters) == 0
+        for f in self.config.filters:
+            if f.matches(data):
+                matches = True
+        return matches
+
     def _run(self):
         """Main worker loop to process messages and forward them.
 
@@ -146,13 +164,15 @@ def _run(self):
                 if self.config.mode == "immediate":
                     data = self.queue.get(timeout=1.0)
                     if data is not None:
-                        self.forward(data)
+                        if self.match_filters(data):
+                            self.forward(data)
                         self.queue.task_done()
                 elif self.config.mode == "bulk":
                     try:
                         data = self.queue.get(timeout=1.0)
                         if data is not None:
-                            self._buffer.append(data)
+                            if self.match_filters(data):
+                                self._buffer.append(data)
                             if len(self._buffer) >= self.config.bulk_size:
                                 self._flush_buffer()
                             self.queue.task_done()
@@ -166,7 +186,8 @@ def _run(self):
                         while len(self._buffer) < self.config.periodic_rate * 2:  # Limit buffer growth
                             data = self.queue.get_nowait()
                             if data is not None:
-                                self._buffer.append(data)
+                                if self.match_filters(data):
+                                    self._buffer.append(data)
                                 self.queue.task_done()
                     except (Exception,):  # empty queue
                         pass
@@ -286,7 +307,3 @@ def _should_retry(self, exception: requests.exceptions.RequestException, attempt
                 f"Failed to forward data to {self.config.url} after {self.config.retry_count + 1} attempts: {exception}"
             )
         return False
-
-
-# Discord-specific forwarder implementation moved to `mongoose.forward.discord`.
-# See mongoose/forward/discord.py for the DiscordFormatter and DiscordForwarder

mongoose/models/configuration.py

@@ -1,10 +1,57 @@
 from pathlib import Path
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Optional, Union, Any
 
 from pydantic import BaseModel, Field, HttpUrl, SecretStr, validator
 
 
-class WebhookForwarderConfiguration(BaseModel):
+class BasicFilter(BaseModel):
+    """A filter that matches objects based on attribute values.
+
+    This filter checks if an object has a specific attribute and whether
+    the attribute's value is in a predefined list of allowed values.
+
+    Attributes:
+        attribute: The name of the attribute to check on the target object.
+        values: List of acceptable values for the attribute.
+    """
+
+    attribute: str
+    values: List[str]
+
+    def matches(self, obj: Any) -> bool:
+        """Check if the given object matches this filter.
+
+        The object matches if it has the specified attribute and the
+        attribute's value is in the list of allowed values.
+
+        Args:
+            obj: The object to check against this filter.
+
+        Returns:
+            True if the object has the attribute and its value is in the
+            allowed values list, False otherwise.
+        """
+        if not hasattr(obj, self.attribute):
+            return False
+        value = getattr(obj, self.attribute)
+        return value in self.values
+
+
+class ForwarderConfiguration(BaseModel):
+    enable: bool = Field(default=False)
+    """Enable the forwarder. Defaults to False."""
+
+    topics: List[str] = Field(default_factory=lambda: ["enriched-network-dpi", "enriched-network-alert"])
+    """List of topics to forward. Defaults to ["enriched-network-dpi", "enriched-network-alert"]."""
+
+    configuration_file: Path = None
+    """Configuration file path, only set when loaded from a drop-in configuration."""
+
+    filters: List[BasicFilter] = []
+    """List of filters. Defaults to []."""
+
+
+class WebhookForwarderConfiguration(ForwarderConfiguration):
     """Configuration for the Webhook Forwarder.
 
     This class defines the destination, authentication, and reliability settings
@@ -23,9 +70,6 @@ class WebhookForwarderConfiguration(BaseModel):
     url: Union[HttpUrl, str]
     """The destination URL for the webhook (must be a valid HTTP/HTTPS URL)."""
 
-    configuration_file: Path = None
-    """Configuration file path, only set when loaded from a drop-in configuration."""
-
     headers: Dict[str, str] = Field(default_factory=dict)
     """Optional dictionary of additional HTTP headers to include in requests."""
 
@@ -50,12 +94,6 @@ class WebhookForwarderConfiguration(BaseModel):
     timeout: float = Field(default=10.0, gt=0)
     """Request timeout in seconds. Defaults to 10.0."""
 
-    enable: bool = Field(default=False)
-    """Enable the forwarder. Defaults to False."""
-
-    topics: List[str] = Field(default_factory=lambda: ["enriched-network-dpi", "enriched-network-alert"])
-    """List of topics to forward. Defaults to ["enriched-network-dpi", "enriched-network-alert"]."""
-
     # Forwarding modes
     mode: str = Field(default="immediate")  # immediate, bulk, periodic
     """Forwarding mode ('immediate', 'bulk', 'periodic'). Defaults to 'immediate'."""
@@ -94,7 +132,7 @@ def validate_auth_token(cls, v, values):
         return v
 
 
-class FileForwarderConfiguration(BaseModel):
+class FileForwarderConfiguration(ForwarderConfiguration):
     """Configuration for the File Forwarder."""
 
     output_dir: str = "output"

tests/test_processing_unsubscribe.py

@@ -22,7 +22,7 @@ def test_unsubscribe():
     assert len(pq.queues[topic]) == 1
 
     # Unsubscribe (method not yet implemented)
-    if hasattr(pq, 'unsubscribe'):
+    if hasattr(pq, "unsubscribe"):
         pq.unsubscribe(subscriber_id)
         assert subscriber_id not in pq.subscribers
         assert topic not in pq.queues or len(pq.queues[topic]) == 0

tests/test_webhook_forwarder.py

@@ -100,10 +100,7 @@ def test_webhook_formatter_alert():
 
 def test_webhook_forwarder_basic_flow(webhook_receiver):
     config = WebhookForwarderConfiguration(
-        enable=True,
-        url="http://example.com/webhook",
-        retry_count=0,
-        topics=["network-alert"]
+        enable=True, url="http://example.com/webhook", retry_count=0, topics=["network-alert"]
     )
 
     pq = ProcessingQueue()
@@ -147,7 +144,7 @@ def test_webhook_forwarder_auth_bearer():
         url="http://example.com/webhook",
         auth_type="bearer",
         auth_token=SecretStr("mytoken"),
-        topics=["network-alert"]
+        topics=["network-alert"],
     )
 
     forwarder = WebhookForwarder(config)
@@ -170,9 +167,7 @@ def test_webhook_forwarder_auth_header():
 
 def test_webhook_forwarder_retries(webhook_receiver):
     config = WebhookForwarderConfiguration(
-        enable=True,
-        url="http://example.com/webhook", retry_count=1, retry_delay=0.1,
-        topics=["network-alert"]
+        enable=True, url="http://example.com/webhook", retry_count=1, retry_delay=0.1, topics=["network-alert"]
     )
 
     pq = ProcessingQueue()
@@ -216,17 +211,14 @@ def test_webhook_configuration_validation():
         WebhookForwarderConfiguration(enable=True, url="http://example.com", auth_type="bearer")
 
     with pytest.raises(ValueError, match="must be in 'user:pass' format"):
-        WebhookForwarderConfiguration(enable=True, url="http://example.com", auth_type="basic",
-                                      auth_token=SecretStr("not-a-pair"))
+        WebhookForwarderConfiguration(
+            enable=True, url="http://example.com", auth_type="basic", auth_token=SecretStr("not-a-pair")
+        )
 
 
 def test_webhook_forwarder_bulk_mode(webhook_receiver):
     config = WebhookForwarderConfiguration(
-        enable=True,
-        url="http://example.com/webhook",
-        mode="bulk",
-        bulk_size=2,
-        topics=["network-alert"]
+        enable=True, url="http://example.com/webhook", mode="bulk", bulk_size=2, topics=["network-alert"]
     )
 
     pq = ProcessingQueue()