feat(securitycenter): Add Resource SCC Org Mgt API ETD Custom Modules (GetEff, ListEff, ListDesc, Validate) (#13081)

* feat(securitycenter): Add Resource SCC Mgt API Org ETD Cust Modules (GetEff, ListEff, ListDesc, Validate)

* fix lint error and add nox file

* SKIP: only test latest due to concurrency issues

* remove unnecessary files

---------

Co-authored-by: Katie McLaughlin <[email protected]>

Author: vijaykanthm

securitycenter/snippets_management_api/event_threat_detection_custom_modules.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2024 Google LLC
+# Copyright 2025 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 
 import uuid
 
-from google.api_core.exceptions import GoogleAPICallError, NotFound
+from google.api_core.exceptions import GoogleAPICallError, NotFound, RetryError
 from google.cloud import securitycentermanagement_v1
 from google.protobuf.field_mask_pb2 import FieldMask
 from google.protobuf.struct_pb2 import Struct
@@ -220,3 +220,173 @@ def delete_event_threat_detection_custom_module(parent: str, module_id: str):
         print(f"Custom Module not found: {module_id}")
         raise e
 # [END securitycenter_delete_event_threat_detection_custom_module]
+
+
+# [START securitycenter_get_effective_event_threat_detection_custom_module]
+def get_effective_event_threat_detection_custom_module(parent: str, module_id: str):
+    """
+    Retrieves an Event Threat Detection custom module using parent and module id as parameters.
+    Args:
+        parent: Use any one of the following options:
+                - organizations/{organization_id}/locations/{location_id}
+                - folders/{folder_id}/locations/{location_id}
+                - projects/{project_id}/locations/{location_id}
+    Returns:
+        The retrieved Event Threat Detection custom module.
+    Raises:
+        NotFound: If the specified custom module does not exist.
+    """
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.GetEffectiveEventThreatDetectionCustomModuleRequest(
+            name=f"{parent}/effectiveEventThreatDetectionCustomModules/{module_id}",
+        )
+
+        response = client.get_effective_event_threat_detection_custom_module(request=request)
+        print(f"Retrieved Effective Event Threat Detection Custom Module: {response.name}")
+        return response
+    except NotFound as e:
+        print(f"Custom Module not found: {e.message}")
+        raise e
+# [END securitycenter_get_effective_event_threat_detection_custom_module]
+
+
+# [START securitycenter_list_effective_event_threat_detection_custom_module]
+def list_effective_event_threat_detection_custom_module(parent: str):
+    """
+    Retrieves list of Event Threat Detection custom module.
+    This includes resident modules defined at the scope of the parent,
+    and inherited modules, inherited from ancestor organizations, folders, and projects (no descendants).
+
+    Args:
+        parent: Use any one of the following options:
+                - organizations/{organization_id}/locations/{location_id}
+                - folders/{folder_id}/locations/{location_id}
+                - projects/{project_id}/locations/{location_id}
+    Returns:
+        List of retrieved all Event Threat Detection custom modules.
+    Raises:
+        NotFound: If the parent resource is not found.
+    """
+
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.ListEffectiveEventThreatDetectionCustomModulesRequest(
+            parent=parent,
+        )
+
+        response = client.list_effective_event_threat_detection_custom_modules(request=request)
+
+        custom_modules = []
+        for custom_module in response:
+            print(f"Custom Module: {custom_module.name}")
+            custom_modules.append(custom_module)
+        return custom_modules
+    except NotFound as e:
+        print(f"Parent resource not found: {parent}")
+        raise e
+    except Exception as e:
+        print(f"An error occurred while listing custom modules: {e}")
+        raise e
+
+# [END securitycenter_list_effective_event_threat_detection_custom_module]
+
+
+# [START securitycenter_list_descendant_event_threat_detection_custom_module]
+def list_descendant_event_threat_detection_custom_module(parent: str):
+    """
+    Retrieves list of all resident Event Threat Detection custom modules and all of its descendants.
+
+    Args:
+        parent: Use any one of the following options:
+                - organizations/{organization_id}/locations/{location_id}
+                - folders/{folder_id}/locations/{location_id}
+                - projects/{project_id}/locations/{location_id}
+    Returns:
+        List of retrieved all Event Threat Detection custom modules.
+    Raises:
+        NotFound: If the parent resource is not found.
+    """
+
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.ListDescendantEventThreatDetectionCustomModulesRequest(
+            parent=parent,
+        )
+
+        response = client.list_descendant_event_threat_detection_custom_modules(request=request)
+
+        custom_modules = []
+        for custom_module in response:
+            print(f"Custom Module: {custom_module.name}")
+            custom_modules.append(custom_module)
+        return custom_modules
+    except NotFound as e:
+        print(f"Parent resource not found: {parent}")
+        raise e
+    except Exception as e:
+        print(f"An error occurred while listing custom modules: {e}")
+        raise e
+
+# [END securitycenter_list_descendant_event_threat_detection_custom_module]
+
+
+# [START securitycenter_validate_event_threat_detection_custom_module]
+def validate_event_threat_detection_custom_module(parent: str):
+    """
+    Validates a custom module for Event Threat Detection.
+
+    Args:
+        parent (str): Use any one of the following options:
+            - organizations/{organization_id}/locations/{location_id}
+            - folders/{folder_id}/locations/{location_id}
+            - projects/{project_id}/locations/{location_id}
+    """
+    try:
+        # Define the raw JSON configuration for the Event Threat Detection custom module
+        raw_text = """
+        {
+            "ips": ["192.0.2.1"],
+            "metadata": {
+                "properties": {
+                    "someProperty": "someValue"
+                },
+                "severity": "MEDIUM"
+            }
+        }
+        """
+
+        # Initialize the client
+        client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+        # Create the request
+        request = securitycentermanagement_v1.ValidateEventThreatDetectionCustomModuleRequest(
+            parent=parent,
+            raw_text=raw_text,
+            type="CONFIGURABLE_BAD_IP"
+        )
+
+        # Perform validation
+        response = client.validate_event_threat_detection_custom_module(request=request)
+
+        # Handle the response and output validation results
+        if response.errors:
+            print("Validation errors:")
+            for error in response.errors:
+                print(f"Field: {error.field_path}, Description: {error.description}")
+            return response
+        else:
+            print("Validation successful: No errors found.")
+            return response
+
+    except GoogleAPICallError as api_error:
+        print(f"API call failed: {api_error}")
+    except RetryError as retry_error:
+        print(f"Retry error occurred: {retry_error}")
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+
+# [END securitycenter_validate_event_threat_detection_custom_module]

securitycenter/snippets_management_api/event_threat_detection_custom_modules_test.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2024 Google LLC
+# Copyright 2025 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -208,7 +208,8 @@ def test_get_event_threat_detection_custom_module():
 
     assert response is not None, "Failed to retrieve the custom module."
     assert response.display_name.startswith(PREFIX)
-    assert response.enablement_state == securitycentermanagement_v1.EventThreatDetectionCustomModule.EnablementState.ENABLED
+    response_org_id = response.name.split("/")[1]  # Extract organization ID from the name field
+    assert response_org_id == ORGANIZATION_ID, f"Organization ID mismatch: Expected {ORGANIZATION_ID}, got {response_org_id}."
 
 
 @backoff.on_exception(
@@ -258,3 +259,59 @@ def test_delete_event_threat_detection_custom_module():
 
     print(f"Custom module was deleted successfully: {module_id}")
     shared_modules.remove(module_id)
+
+
+@backoff.on_exception(
+    backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
+)
+def test_get_effective_event_threat_detection_custom_module():
+
+    module_id = get_random_shared_module()
+    parent = f"organizations/{ORGANIZATION_ID}/locations/{LOCATION}"
+
+    # Retrieve the custom module
+    response = event_threat_detection_custom_modules.get_effective_event_threat_detection_custom_module(parent, module_id)
+
+    assert response is not None, "Failed to retrieve the custom module."
+    assert response.display_name.startswith(PREFIX)
+    response_org_id = response.name.split("/")[1]  # Extract organization ID from the name field
+    assert response_org_id == ORGANIZATION_ID, f"Organization ID mismatch: Expected {ORGANIZATION_ID}, got {response_org_id}."
+
+
+@backoff.on_exception(
+    backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
+)
+def test_list_effective_event_threat_detection_custom_module():
+
+    parent = f"organizations/{ORGANIZATION_ID}/locations/{LOCATION}"
+    # Retrieve the custom modules
+    custom_modules = event_threat_detection_custom_modules.list_effective_event_threat_detection_custom_module(parent)
+
+    assert custom_modules is not None, "Failed to retrieve the custom modules."
+    assert len(custom_modules) > 0, "No custom modules were retrieved."
+
+
+@backoff.on_exception(
+    backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
+)
+def test_list_descendant_event_threat_detection_custom_module():
+
+    parent = f"organizations/{ORGANIZATION_ID}/locations/{LOCATION}"
+    # Retrieve the custom modules
+    custom_modules = event_threat_detection_custom_modules.list_descendant_event_threat_detection_custom_module(parent)
+
+    assert custom_modules is not None, "Failed to retrieve the custom modules."
+    assert len(custom_modules) > 0, "No custom modules were retrieved."
+
+
+@backoff.on_exception(
+    backoff.expo, (InternalServerError, ServiceUnavailable, NotFound), max_tries=3
+)
+def test_validate_event_threat_detection_custom_module():
+
+    parent = f"organizations/{ORGANIZATION_ID}/locations/{LOCATION}"
+
+    # Retrieve the custom module
+    response = event_threat_detection_custom_modules.validate_event_threat_detection_custom_module(parent)
+
+    assert response is not None, "Failed to retrieve the validte ETD custom module response."

securitycenter/snippets_management_api/noxfile_config.py

@@ -23,7 +23,9 @@
 
 TEST_CONFIG_OVERRIDE = {
     # You can opt out from the test for specific Python versions.
-    "ignored_versions": ["2.7", "3.7", "3.9", "3.10", "3.11"],
+    # SKIPPED VERSIONS: due to concurrency issues editing multiple org level
+    # custom modules, only test these samples on latest Python.
+    "ignored_versions": ["2.7", "3.7", "3.8", "3.9", "3.10", "3.11", "3.12"],
     # An envvar key for determining the project id to use. Change it
     # to 'BUILD_SPECIFIC_GCLOUD_PROJECT' if you want to opt in using a
     # build specific Cloud project. You can also use your own string