Implement log shipping to Graylog via GELF

Author: bnazare

pycti/api/opencti_api_client.py

@@ -98,6 +98,18 @@ class OpenCTIApiClient:
     :type cert: str, tuple, optional
     :param auth: Add a AuthBase class with custom authentication for you OpenCTI infrastructure.
     :type auth: requests.auth.AuthBase, optional
+    :param graylog_host: Graylog host name or IP address
+    :type graylog_host: str, optional
+    :param graylog_port: Graylog port
+    :type graylog_port: int, optional
+    :param graylog_adapter: the Graylog adapter to use. Valid values are "udp" and "tcp". Uses UDP by default.
+    :type graylog_adapter: str, optional
+    :param log_shipping_level: log level when shipping logs remotely
+    :type log_shipping_level: str, optional
+    :param log_shipping_env_var_prefix: The prefix used to match environment variables. Matching variables will be added
+        as meta info to the log data. The value of this property will be stripped from the name of the environment
+        variable.
+    :type log_shipping_env_var_prefix: str, optional
     """
 
     def __init__(
@@ -112,6 +124,11 @@ def __init__(
         cert=None,
         auth=None,
         perform_health_check=True,
+        graylog_host=None,
+        graylog_port=None,
+        graylog_adapter=None,
+        log_shipping_level=None,
+        log_shipping_env_var_prefix=None,
     ):
         """Constructor method"""
 
@@ -126,7 +143,9 @@ def __init__(
             raise ValueError("A TOKEN must be set")
 
         # Configure logger
-        self.logger_class = logger(log_level.upper(), json_logging)
+        self.logger_class = logger(log_level.upper(), json_logging, graylog_host, graylog_port, graylog_adapter,
+                                   log_shipping_level.upper() if log_shipping_level is not None else None,
+                                   log_shipping_env_var_prefix)
         self.app_logger = self.logger_class("api")
 
         # Define API

pycti/connector/opencti_connector_helper.py

@@ -878,6 +878,21 @@ def __init__(self, config: Dict, playbook_compatible=False) -> None:
         self.log_level = get_config_variable(
             "CONNECTOR_LOG_LEVEL", ["connector", "log_level"], config, default="INFO"
         ).upper()
+        self.graylog_host = get_config_variable(
+            "CONNECTOR_GRAYLOG_HOST", ["connector", "graylog_host"], config
+        )
+        self.graylog_port = get_config_variable(
+            "CONNECTOR_GRAYLOG_PORT", ["connector", "graylog_port"], config, True, 12201
+        )
+        self.graylog_adapter = get_config_variable(
+            "CONNECTOR_GRAYLOG_ADAPTER", ["connector", "graylog_adapter"], config, default="udp"
+        )
+        self.log_shipping_level = get_config_variable(
+            "CONNECTOR_LOG_SHIPPING_LEVEL", ["connector", "log_shipping_level"], config, default="INFO"
+        ).upper()
+        self.log_shipping_env_var_prefix = get_config_variable(
+            "CONNECTOR_LOG_SHIPPING_ENV_VAR_PREFIX", ["connector", "log_shipping_env_var_prefix"], config
+        )
         self.connect_run_and_terminate = get_config_variable(
             "CONNECTOR_RUN_AND_TERMINATE",
             ["connector", "run_and_terminate"],
@@ -915,6 +930,11 @@ def __init__(self, config: Dict, playbook_compatible=False) -> None:
             self.opencti_ssl_verify,
             json_logging=self.opencti_json_logging,
             bundle_send_to_queue=self.bundle_send_to_queue,
+            graylog_host=self.graylog_host,
+            graylog_port=self.graylog_port,
+            graylog_adapter=self.graylog_adapter,
+            log_shipping_level=self.log_shipping_level,
+            log_shipping_env_var_prefix=self.log_shipping_env_var_prefix,
         )
         # - Impersonate API that will use applicant id
         # Behave like standard api if applicant not found
@@ -925,6 +945,11 @@ def __init__(self, config: Dict, playbook_compatible=False) -> None:
             self.opencti_ssl_verify,
             json_logging=self.opencti_json_logging,
             bundle_send_to_queue=self.bundle_send_to_queue,
+            graylog_host=self.graylog_host,
+            graylog_port=self.graylog_port,
+            graylog_adapter=self.graylog_adapter,
+            log_shipping_level=self.log_shipping_level,
+            log_shipping_env_var_prefix=self.log_shipping_env_var_prefix,
         )
         self.connector_logger = self.api.logger_class(self.connect_name)
         # For retro compatibility

pycti/utils/opencti_logger.py

@@ -1,6 +1,8 @@
 import datetime
 import logging
+import os
 
+from pygelf import GelfUdpHandler, GelfTcpHandler
 from pythonjsonlogger import jsonlogger
 
 
@@ -17,7 +19,23 @@ def add_fields(self, log_record, record, message_dict):
             log_record["level"] = record.levelname
 
 
-def logger(level, json_logging=True):
+class ContextFilter(logging.Filter):
+    def __init__(self, context_vars):
+        """
+        :param context_vars: the extra properties to add to the LogRecord
+        :type context_vars: list[tuple[str, str]]
+        """
+        super().__init__()
+        self.context_vars = context_vars
+
+    def filter(self, record):
+        for key, value in self.context_vars:
+            setattr(record, key, value)
+        return True
+
+
+def logger(level, json_logging=True, graylog_host=None, graylog_port=None, graylog_adapter=None,
+           log_shipping_level=None, log_shipping_env_var_prefix=None):
     # Exceptions
     logging.getLogger("urllib3").setLevel(logging.WARNING)
     logging.getLogger("pika").setLevel(logging.ERROR)
@@ -31,6 +49,21 @@ def logger(level, json_logging=True):
     else:
         logging.basicConfig(level=level)
 
+    if graylog_host is not None:
+        if graylog_adapter == "tcp":
+            shipping_handler = GelfTcpHandler(host=graylog_host, port=graylog_port, include_extra_fields=True)
+        else:
+            shipping_handler = GelfUdpHandler(host=graylog_host, port=graylog_port, include_extra_fields=True)
+        shipping_handler.setLevel(log_shipping_level)
+
+        if log_shipping_env_var_prefix is not None:
+            filtered_env = [(k.removeprefix(log_shipping_env_var_prefix), v) for k, v in os.environ.items()
+                            if k.startswith(log_shipping_env_var_prefix)]
+            shipping_filter = ContextFilter(filtered_env)
+            shipping_handler.addFilter(shipping_filter)
+
+        logging.getLogger().addHandler(shipping_handler)
+
     class AppLogger:
         def __init__(self, name):
             self.local_logger = logging.getLogger(name)

requirements.txt

@@ -15,4 +15,5 @@ opentelemetry-sdk~=1.22.0
 deprecation~=2.1.0
 # OpenCTI
 filigran-sseclient~=1.0.0
-stix2~=3.0.1
+stix2~=3.0.1
+pygelf~=0.4.2

setup.cfg

@@ -52,6 +52,7 @@ install_requires =
     # OpenCTI
     filigran-sseclient~=1.0.0
     stix2~=3.0.1
+    pygelf~=0.4.2
 
 [options.extras_require]
 dev =

tests/cases/connectors.py

@@ -63,6 +63,7 @@ def __init__(self, config_file_path: str, api_client: OpenCTIApiClient, data: Di
         os.environ["OPENCTI_JSON_LOGGING"] = "true"
         os.environ["CONNECTOR_EXPOSE_METRICS"] = "true"
         os.environ["CONNECTOR_METRICS_PORT"] = "9096"
+        os.environ["GRAYLOG_DUMMY_VAR"] = "dummy_value"
 
         config = (
             yaml.load(open(config_file_path), Loader=yaml.FullLoader)

tests/data/external_import_config.yml

@@ -6,6 +6,11 @@ connector:
   confidence_level: 80 # From 0 (Unknown) to 100 (Fully trusted)
   update_existing_data: True
   log_level: 'debug'
+  graylog_host: '127.0.0.1'
+  graylog_port: 12201
+  graylog_adapter: 'tcp'
+  log_shipping_level: 'warn'
+  log_shipping_env_var_prefix: 'GRAYLOG_'
 
 test:
   interval: 1

tests/data/internal_import_config.yml

@@ -7,3 +7,4 @@ connector:
   only_contextual: true # Only extract data related to an entity (a report, a threat actor, etc.)
   confidence_level: 15 # From 0 (Unknown) to 100 (Fully trusted)
   log_level: 'debug'
+  graylog_host: '127.0.0.1'