feat(securitycenter): Add Resource SCC Mgmt API Org SHA Custom Modules (Create, Get, List, Delete, Update) (GoogleCloudPlatform#13004)

vijaykanthm · web-flow · commit 2cb8866994ee · 2025-02-07T09:36:29.000Z
* feat(securitycenter): Add Resource SCC Mgt API Org SHA Cust Modules (Create, Get, Delete, List, Update)

* fix lint and address comments

* fix lint errors

* fix the ci python version errors

* lint fix

* fix linting

* Refactor the filename of the testfile

* Trigger CI pipeline

* Refactor the cleaning up of created custom modules in the current test session

* Refactor the module creation and clean up

* Remove commented code
diff --git a/securitycenter/snippets_management_api/noxfile_config.py b/securitycenter/snippets_management_api/noxfile_config.py
@@ -0,0 +1,41 @@
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Default TEST_CONFIG_OVERRIDE for python repos.
+
+# You can copy this file into your directory, then it will be inported from
+# the noxfile.py.
+
+# The source of truth:
+# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/master/noxfile_config.py
+
+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"],
+    # 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
+    # to use your own Cloud project.
+    "gcloud_project_env": "GOOGLE_CLOUD_PROJECT",
+    # "gcloud_project_env": "BUILD_SPECIFIC_GCLOUD_PROJECT",
+    # A dictionary you want to inject into your test. Don't put any
+    # secrets here. These values will override predefined values.
+    "envs": {
+        "GCLOUD_ORGANIZATION": "1081635000895",
+        "GCLOUD_PROJECT": "project-a-id",
+        "GCLOUD_PUBSUB_TOPIC": "projects/project-a-id/topics/notifications-sample-topic",
+        "GCLOUD_PUBSUB_SUBSCRIPTION": "projects/project-a-id/subscriptions/notification-sample-subscription",
+        "GCLOUD_LOCATION": "global",
+    },
+}
diff --git a/securitycenter/snippets_management_api/requirements-test.txt b/securitycenter/snippets_management_api/requirements-test.txt
@@ -0,0 +1,5 @@
+backoff==2.2.1
+pytest==8.2.0
+google-cloud-bigquery==3.11.4
+google-cloud-securitycentermanagement==0.1.17
+
diff --git a/securitycenter/snippets_management_api/requirements.txt b/securitycenter/snippets_management_api/requirements.txt
@@ -0,0 +1,3 @@
+google-cloud-securitycentermanagement==0.1.17
+google-cloud-bigquery==3.11.4
+google-cloud-pubsub==2.21.5
diff --git a/securitycenter/snippets_management_api/security_health_analytics_custom_modules.py b/securitycenter/snippets_management_api/security_health_analytics_custom_modules.py
@@ -0,0 +1,240 @@
+#!/usr/bin/env python
+#
+# 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.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import uuid
+
+from google.api_core.exceptions import GoogleAPICallError, NotFound
+from google.cloud import securitycentermanagement_v1
+
+
+# [START securitycenter_create_security_health_analytics_custom_module]
+def create_security_health_analytics_custom_module(parent: str) -> securitycentermanagement_v1.SecurityHealthAnalyticsCustomModule:
+    """
+    Creates a Security Health Analytics custom module.
+
+    This custom module evaluates Cloud KMS CryptoKeys to ensure their rotation period exceeds 30 days (2592000 seconds),
+    as per security best practices. A shorter rotation period helps reduce the risk of exposure in the event of a compromise.
+
+    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:
+        Dict: Created custom module details.
+    """
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        # Generate a unique suffix
+        unique_suffix = str(uuid.uuid4()).replace("-", "_")
+        # Generate a unique display name
+        display_name = f"python_sample_sha_custom_module_{unique_suffix}"
+
+        # Define the custom module configuration
+        custom_module = {
+            "display_name": display_name,
+            "enablement_state": "ENABLED",
+            "custom_config": {
+                "description": (
+                    "Sample custom module for testing purposes. This custom module evaluates "
+                    "Cloud KMS CryptoKeys to ensure their rotation period exceeds 30 days (2592000 seconds)."
+                ),
+                "predicate": {
+                    "expression": "has(resource.rotationPeriod) && (resource.rotationPeriod > duration('2592000s'))",
+                    "title": "Cloud KMS CryptoKey Rotation Period",
+                    "description": (
+                        "Evaluates whether the rotation period of a Cloud KMS CryptoKey exceeds 30 days. "
+                        "A longer rotation period might increase the risk of exposure."
+                    ),
+                },
+                "recommendation": (
+                    "Review and adjust the rotation period for Cloud KMS CryptoKeys to align with your security policies. "
+                    "Consider setting a shorter rotation period if possible."
+                ),
+                "resource_selector": {"resource_types": ["cloudkms.googleapis.com/CryptoKey"]},
+                "severity": "CRITICAL",
+                "custom_output": {
+                    "properties": [
+                        {
+                            "name": "example_property",
+                            "value_expression": {
+                                "description": "The resource name of the CryptoKey being evaluated.",
+                                "expression": "resource.name",
+                                "location": "global",
+                                "title": "CryptoKey Resource Name",
+                            },
+                        }
+                    ]
+                },
+            },
+        }
+
+        request = securitycentermanagement_v1.CreateSecurityHealthAnalyticsCustomModuleRequest(
+            parent=parent,
+            security_health_analytics_custom_module=custom_module,
+        )
+
+        response = client.create_security_health_analytics_custom_module(request=request)
+        print(f"Created SecurityHealthAnalytics Custom Module: {response.name}")
+        return response
+
+    except GoogleAPICallError as e:
+        print(f"Failed to create EventThreatDetectionCustomModule: {e}")
+        raise
+# [END securitycenter_create_security_health_analytics_custom_module]
+
+
+# [START securitycenter_get_security_health_analytics_custom_module]
+def get_security_health_analytics_custom_module(parent: str, module_id: str):
+    """
+    Retrieves a Security Health Analytics custom module.
+    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 Security Health Analytics custom module.
+    Raises:
+        NotFound: If the specified custom module does not exist.
+    """
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.GetSecurityHealthAnalyticsCustomModuleRequest(
+            name=f"{parent}/securityHealthAnalyticsCustomModules/{module_id}",
+        )
+
+        response = client.get_security_health_analytics_custom_module(request=request)
+        print(f"Retrieved Security Health Analytics Custom Module: {response.name}")
+        return response
+    except NotFound as e:
+        print(f"Custom Module not found: {response.name}")
+        raise e
+# [END securitycenter_get_security_health_analytics_custom_module]
+
+
+# [START securitycenter_list_security_health_analytics_custom_module]
+def list_security_health_analytics_custom_module(parent: str):
+    """
+    Retrieves list of Security Health Analytics custom module.
+    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 Security Health Analytics custom modules.
+    Raises:
+        NotFound: If the specified custom module does not exist.
+    """
+
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.ListSecurityHealthAnalyticsCustomModulesRequest(
+            parent=parent,
+        )
+
+        response = client.list_security_health_analytics_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_security_health_analytics_custom_module]
+
+
+# [START securitycenter_delete_security_health_analytics_custom_module]
+def delete_security_health_analytics_custom_module(parent: str, module_id: str):
+    """
+    Deletes a Security Health Analytics custom module.
+    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 deleted Security Health Analytics custom module.
+    Raises:
+        NotFound: If the specified custom module does not exist.
+    """
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+
+    try:
+        request = securitycentermanagement_v1.DeleteSecurityHealthAnalyticsCustomModuleRequest(
+            name=f"{parent}/securityHealthAnalyticsCustomModules/{module_id}",
+        )
+
+        client.delete_security_health_analytics_custom_module(request=request)
+        print(f"Deleted SecurityHealthAnalyticsCustomModule Successfully: {module_id}")
+    except NotFound as e:
+        print(f"Custom Module not found: {module_id}")
+        raise e
+# [END securitycenter_delete_security_health_analytics_custom_module]
+
+
+# [START securitycenter_update_security_health_analytics_custom_module]
+def update_security_health_analytics_custom_module(parent: str, module_id: str):
+    """
+    Updates Security Health Analytics custom module.
+    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 updated Security Health Analytics custom module.
+    Raises:
+        NotFound: If the specified custom module does not exist.
+    """
+    from google.protobuf.field_mask_pb2 import FieldMask
+
+    client = securitycentermanagement_v1.SecurityCenterManagementClient()
+    try:
+        # Define the custom module configuration
+        custom_module = securitycentermanagement_v1.SecurityHealthAnalyticsCustomModule(
+            name=f"{parent}/securityHealthAnalyticsCustomModules/{module_id}",
+            enablement_state=securitycentermanagement_v1.SecurityHealthAnalyticsCustomModule.EnablementState.DISABLED,
+        )
+
+        # Prepare the update request
+        request = securitycentermanagement_v1.UpdateSecurityHealthAnalyticsCustomModuleRequest(
+            security_health_analytics_custom_module=custom_module,
+            update_mask=FieldMask(paths=["enablement_state"]),
+        )
+
+        # Execute the update request
+        response = client.update_security_health_analytics_custom_module(request=request)
+
+        print(f"Updated Security Health Analytics Custom Module: {response.name}")
+        return response
+    except NotFound:
+        print(f"Custom Module not found: {custom_module.name}")
+        raise
+    except Exception as e:
+        print(f"An error occurred while updating the custom module: {e}")
+        raise
+
+# [END securitycenter_update_security_health_analytics_custom_module]
diff --git a/securitycenter/snippets_management_api/security_health_analytics_custom_modules_test.py b/securitycenter/snippets_management_api/security_health_analytics_custom_modules_test.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+google-cloud-securitycentermanagement==0.1.17`
	`2`	`+google-cloud-bigquery==3.11.4`
	`3`	`+google-cloud-pubsub==2.21.5`