Add Security Monitoring API to dogshell

Author: clementgbcn

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## Unreleased
+
+* [Added] Add Cloud SIEM rule management and security signals retrieval.
+* [Added] Add dogshell command for security monitoring rule and signal management.
+
 ## v0.51.0 / 2025-01-27
 
 * [Added] Add hosts endpoint. See [#884](https://github.com/DataDog/datadogpy/pull/884).

datadog/api/__init__.py

@@ -50,3 +50,5 @@
 from datadog.api.service_level_objectives import ServiceLevelObjective
 from datadog.api.synthetics import Synthetics
 from datadog.api.logs import Logs
+from datadog.api.security_monitoring_rules import SecurityMonitoringRule
+from datadog.api.security_monitoring_signals import SecurityMonitoringSignal

datadog/api/security_monitoring_rules.py

@@ -0,0 +1,93 @@
+# Unless explicitly stated otherwise all files in this repository are licensed under the BSD-3-Clause License.
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2015-Present Datadog, Inc
+"""
+Security Monitoring Rule API.
+"""
+
+from datadog.api.resources import (
+    GetableAPIResource,
+    CreateableAPIResource,
+    ListableAPIResource,
+    UpdatableAPIResource,
+    DeletableAPIResource,
+    ActionAPIResource,
+)
+
+
+class SecurityMonitoringRule(
+    GetableAPIResource,
+    CreateableAPIResource,
+    ListableAPIResource,
+    UpdatableAPIResource,
+    DeletableAPIResource,
+    ActionAPIResource,
+):
+    """
+    A wrapper around Security Monitoring Rule API.
+    """
+
+    _resource_name = "security_monitoring/rules"
+    _api_version = "v2"
+
+    @classmethod
+    def get_all(cls, **params):
+        """
+        Get all security monitoring rules.
+
+        :param params: additional parameters to filter security monitoring rules
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringRule, cls).get_all(**params)
+
+    @classmethod
+    def get(cls, rule_id, **params):
+        """
+        Get a security monitoring rule's details.
+
+        :param rule_id: ID of the security monitoring rule
+        :type rule_id: str
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringRule, cls).get(rule_id, **params)
+
+    @classmethod
+    def create(cls, **params):
+        """
+        Create a security monitoring rule.
+
+        :param params: Parameters to create the security monitoring rule with
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringRule, cls).create(**params)
+
+    @classmethod
+    def update(cls, rule_id, **params):
+        """
+        Update a security monitoring rule.
+
+        :param rule_id: ID of the security monitoring rule to update
+        :type rule_id: str
+        :param params: Parameters to update the security monitoring rule with
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringRule, cls).update(rule_id, **params)
+
+    @classmethod
+    def delete(cls, rule_id, **params):
+        """
+        Delete a security monitoring rule.
+
+        :param rule_id: ID of the security monitoring rule to delete
+        :type rule_id: str
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringRule, cls).delete(rule_id, **params)

datadog/api/security_monitoring_signals.py

@@ -0,0 +1,107 @@
+# Unless explicitly stated otherwise all files in this repository are licensed under the BSD-3-Clause License.
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2015-Present Datadog, Inc
+"""
+Security Monitoring Signals API.
+"""
+
+from datadog.api.resources import (
+    GetableAPIResource,
+    ListableAPIResource,
+    SearchableAPIResource,
+    ActionAPIResource,
+)
+
+
+class SecurityMonitoringSignal(
+    GetableAPIResource,
+    ListableAPIResource,
+    SearchableAPIResource,
+    ActionAPIResource,
+):
+    """
+    A wrapper around Security Monitoring Signal API.
+    """
+
+    _resource_name = "security_monitoring/signals"
+    _api_version = "v2"
+
+    @classmethod
+    def get(cls, signal_id, **params):
+        """
+        Get a security signal's details.
+
+        :param signal_id: ID of the security signal
+        :type signal_id: str
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringSignal, cls).get(signal_id, **params)
+
+    @classmethod
+    def get_all(cls, **params):
+        """
+        Get all security signals.
+
+        :param params: additional parameters to filter security signals
+            Valid options are:
+            - filter[query]: search query to filter security signals
+            - filter[from]: minimum timestamp for returned security signals
+            - filter[to]: maximum timestamp for returned security signals
+            - sort: sort order, can be 'timestamp', '-timestamp', etc.
+            - page[size]: number of signals to return per page
+            - page[cursor]: cursor to use for pagination
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return super(SecurityMonitoringSignal, cls).get_all(**params)
+
+    @classmethod
+    def search(cls, **params):
+        """
+        Search for security signals.
+
+        :param params: search parameters
+            Valid options are:
+            - filter[query]: search query to filter security signals
+            - filter[from]: minimum timestamp for returned security signals
+            - filter[to]: maximum timestamp for returned security signals
+            - sort: sort order, can be 'timestamp', '-timestamp', etc.
+            - page[size]: number of signals to return per page
+            - page[cursor]: cursor to use for pagination
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        return cls._search(**params)
+
+    @classmethod
+    def change_triage_state(cls, signal_ids, state, **params):
+        """
+        Change the triage state of security signals.
+
+        :param signal_ids: list of signal IDs to update
+        :type signal_ids: list of str
+        :param state: new triage state ('open', 'archived', 'under_review')
+        :type state: str
+        :param params: additional parameters
+        :type params: dict
+
+        :returns: Dictionary representing the API's JSON response
+        """
+        body = {
+            "data": {
+                "attributes": {
+                    "signals": signal_ids,
+                    "state": state,
+                },
+                "type": "signal_state_change",
+            }
+        }
+
+        params = params or {}
+
+        return cls._trigger_class_action(
+            "PATCH", "triage_state", params=params, **body
+        )

datadog/dogshell/__init__.py

@@ -27,6 +27,7 @@
 from datadog.dogshell.tag import TagClient
 from datadog.dogshell.timeboard import TimeboardClient
 from datadog.dogshell.dashboard import DashboardClient
+from datadog.dogshell.security_monitoring import SecurityMonitoringClient
 
 
 def main():
@@ -100,6 +101,7 @@ def main():
     DowntimeClient.setup_parser(subparsers)
     ServiceCheckClient.setup_parser(subparsers)
     ServiceLevelObjectiveClient.setup_parser(subparsers)
+    SecurityMonitoringClient.setup_parser(subparsers)
 
     args = parser.parse_args()