feat(securitycenter): Add Resource SCC Org Mgmt API ETD Custom Modules (Create, Get, List, Delete, Update) #13040
Conversation
…Create, Get, Delete, List, Update)
|
Here is the summary of changes. You are about to add 5 region tags.
This comment is generated by snippet-bot.
|
![code-review-assist[bot]](https://avatars.githubusercontent.com/in/1008254?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hi team,
I'm currently reviewing this pull request and will provide feedback shortly. In the meantime, here's a summary for you and other reviewers to quickly get up to speed:
This pull request, authored by vijaykanthm, aims to add Security Health Analytics (SHA) Organization Management API Event Threat Detection (ETD) Custom Module samples. The samples cover the full CRUD (Create, Read, Update, Delete) operations.
Here's a breakdown of the changes:
- New file:
securitycenter/snippets_management_api/event_threat_detection_custom_modules.pyThis file contains the core Python code for creating, getting, listing, updating, and deleting ETD custom modules using the Security Health Analytics API. The functions are well-documented with clear examples and error handling. - New file:
securitycenter/snippets_management_api/event_threat_detection_custom_modules_test.pyThis file includes comprehensive unit tests to verify the functionality of the custom module operations. It usespytestand includes robust error handling and backoff mechanisms for transient API issues. - Updated files:
securitycenter/snippets_management_api/noxfile_config.py,securitycenter/snippets_management_api/requirements-test.txt,securitycenter/snippets_management_api/requirements.txtThese files were updated to reflect the new dependencies and test configurations required for the added samples.
The pull request addresses several internal Google bugs (b/347347902, b/347347344, b/347347881, b/347347878, b/347347934) and includes a checklist to ensure the samples meet the required quality standards.
I'll be looking at the code for correctness, efficiency, and adherence to Google's style guidelines. I'll also be running the tests to ensure everything works as expected.
A little poem to lighten the mood:
Code flows like a river,
Tests run, a green light appears,
Bugs flee in the night.
Thanks,
Gemini
product-auto-label
bot
added
api: securitycenter
| 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. |
There was a problem hiding this comment.
The description mentions Cloud KMS CryptoKeys, but the actual module created is of type CONFIGURABLE_BAD_IP. Update the description to reflect the functionality of the created 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. | |
| This custom module creates a configurable bad IP type custom module, which can be used to detect and block malicious IP addresses. |
| response = client.create_event_threat_detection_custom_module(request=request) | ||
|
|
||
| print(f"Created EventThreatDetectionCustomModule: {response.name}") | ||
| return response |
There was a problem hiding this comment.
The return type of the function should be securitycentermanagement_v1.EventThreatDetectionCustomModule instead of Dict.
return response| except Exception as e: | ||
| print(f"Failed to create EventThreatDetectionCustomModule: {e}") | ||
| raise |
There was a problem hiding this comment.
Use a more specific exception instead of a bare except. This will help in debugging and avoid catching unexpected exceptions.
except google.api_core.exceptions.GoogleAPICallError as e:
print(f"Failed to create EventThreatDetectionCustomModule: {e}")
raise| print(f"Retrieved Event Threat Detection Custom Module: {response.name}") | ||
| return response | ||
| except NotFound as e: | ||
| print(f"Custom Module not found: {response.name}") |
There was a problem hiding this comment.
The variable "response" is not defined in the except block. Use the exception "e" instead.
print(f"Custom Module not found: {e.message}")| def add_custom_module(org_id: str): | ||
|
|
||
| parent = f"organizations/{org_id}/locations/global" | ||
| client = securitycentermanagement_v1.SecurityCenterManagementClient() | ||
|
|
||
| # Generate a unique display name | ||
| unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}" | ||
| display_name = f"python_sample_etd_custom_module_test_{unique_suffix}" | ||
|
|
||
| # Define the metadata and other config parameters as a dictionary | ||
| config_map = { | ||
| "metadata": { | ||
| "severity": "MEDIUM", | ||
| "description": "Sample custom module for testing purposes. Please do not delete.", | ||
| "recommendation": "na", | ||
| }, | ||
| "ips": ["0.0.0.0"], | ||
| } | ||
|
|
||
| # Convert the dictionary to a Struct | ||
| config_struct = Struct() | ||
| config_struct.update(config_map) | ||
|
|
||
| # Define the custom module configuration | ||
| custom_module = { | ||
| "display_name": display_name, | ||
| "enablement_state": "ENABLED", | ||
| "type_": "CONFIGURABLE_BAD_IP", | ||
| "config": config_struct, | ||
| } | ||
|
|
||
| request = securitycentermanagement_v1.CreateEventThreatDetectionCustomModuleRequest( | ||
| parent=parent, | ||
| event_threat_detection_custom_module=custom_module, | ||
| ) | ||
| response = client.create_event_threat_detection_custom_module(request=request) | ||
| print(f"Created Event Threat Detection Custom Module: {response.name}") | ||
| module_name = response.name | ||
| module_id = module_name.split("/")[-1] | ||
| return module_name, module_id |
There was a problem hiding this comment.
Consider using the client library's create method directly in the test instead of creating a separate helper function. This will simplify the test and make it easier to understand.
parent = f"organizations/{org_id}/locations/global"
client = securitycentermanagement_v1.SecurityCenterManagementClient()
# Generate a unique display name
unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}"
display_name = f"python_sample_etd_custom_module_test_{unique_suffix}"
# Define the metadata and other config parameters as a dictionary
config_map = {
"metadata": {
"severity": "MEDIUM",
"description": "Sample custom module for testing purposes. Please do not delete.",
"recommendation": "na",
},
"ips": ["0.0.0.0"],
}
# Convert the dictionary to a Struct
config_struct = Struct()
config_struct.update(config_map)
# Define the custom module configuration
custom_module = securitycentermanagement_v1.EventThreatDetectionCustomModule(
display_name=display_name,
enablement_state=securitycentermanagement_v1.EventThreatDetectionCustomModule.EnablementState.ENABLED,
type_="CONFIGURABLE_BAD_IP",
config=config_struct,
)
request = securitycentermanagement_v1.CreateEventThreatDetectionCustomModuleRequest(
parent=parent,
event_threat_detection_custom_module=custom_module,
)
response = client.create_event_threat_detection_custom_module(request=request)
print(f"Created Event Threat Detection Custom Module: {response.name}")
module_name = response.name
module_id = module_name.split('/')[-1]
return module_name, module_id|
|
||
|
|
||
| # [START securitycenter_create_event_threat_detection_custom_module] | ||
| def create_event_threat_detection_custom_module(parent: str) -> Dict: |
There was a problem hiding this comment.
The return type annotation should be securitycentermanagement_v1.EventThreatDetectionCustomModule instead of Dict.
def create_event_threat_detection_custom_module(parent: str) -> securitycentermanagement_v1.EventThreatDetectionCustomModule:| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
Update the return type to securitycentermanagement_v1.EventThreatDetectionCustomModule.
EventThreatDetectionCustomModule| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
Update the return type to securitycentermanagement_v1.EventThreatDetectionCustomModule.
EventThreatDetectionCustomModule
There was a problem hiding this comment.
Hi team,
I'm currently reviewing this pull request and will provide my feedback in a few minutes. In the meantime, I'm providing this summary of the pull request's content to help you and other reviewers quickly understand the changes and intent.
This pull request, authored by vijaykanthm, aims to add Security Health Analytics (SHA) samples for the Event Threat Detection (ETD) custom modules. Specifically, it adds functionality for creating, getting, listing, deleting, and updating ETD custom modules using the Security Center Org Management API. This addresses several internal Google bugs (b/347347902, b/347347344, b/347347881, b/347347878, b/347347934).
Here's a breakdown of the changes:
securitycenter/snippets_management_api/event_threat_detection_custom_modules.py: This file contains the core Python code for interacting with the Security Center Org Management API's ETD custom modules. The changes introduce functions for each CRUD operation (Create, Read, Update, Delete) and list operations. The code is well-structured and includes comprehensive docstrings.securitycenter/snippets_management_api/event_threat_detection_custom_modules_test.py: This file contains unit tests for the functions inevent_threat_detection_custom_modules.py. The tests cover all the added functionalities and include robust error handling usingbackofffor transient errors. The tests also include a fixture to ensure a clean environment before and after running the tests.securitycenter/snippets_management_api/noxfile_config.py: This file is updated to include environment variables for testing, specifically for the organization ID, project ID, and other relevant resources.securitycenter/snippets_management_api/requirements-test.txtandsecuritycenter/snippets_management_api/requirements.txt: These files are updated to include the necessary dependencies for running the tests and the samples, respectively.
The author has also included a checklist to ensure that all necessary steps have been followed before merging the PR.
And now, a little haiku to lighten the mood:
Code flows like a stream,
Tests run green, a joyful sight,
Bugs flee in the night.
Thanks,
Gemini
| 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. |
There was a problem hiding this comment.
The description mentions Cloud KMS CryptoKeys, but the created module is of type CONFIGURABLE_BAD_IP. Update the description to reflect the actual functionality of the 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. | |
| This custom module creates a configurable bad IP custom module, which can be used to detect and block malicious IP addresses. |
| response = client.create_event_threat_detection_custom_module(request=request) | ||
|
|
||
| print(f"Created EventThreatDetectionCustomModule: {response.name}") | ||
| return response |
There was a problem hiding this comment.
The return type is Dict, but the method is returning an EventThreatDetectionCustomModule object. It should return a dictionary.
return {'name': response.name, 'displayName': response.display_name, 'enablementState': response.enablement_state, 'config': response.config}| except Exception as e: | ||
| print(f"Failed to create EventThreatDetectionCustomModule: {e}") | ||
| raise |
There was a problem hiding this comment.
Use a more specific exception instead of a bare except. This will help in debugging and prevent unintended errors from being masked.
except google.api_core.exceptions.GoogleAPICallError as e:
print(f"Failed to create EventThreatDetectionCustomModule: {e}")
raise| print(f"Retrieved Event Threat Detection Custom Module: {response.name}") | ||
| return response | ||
| except NotFound as e: | ||
| print(f"Custom Module not found: {response.name}") |
There was a problem hiding this comment.
The variable "response" is not defined within the except block. Use the exception object "e" instead.
print(f"Custom Module not found: {e.message}")| def add_custom_module(org_id: str): | ||
|
|
||
| parent = f"organizations/{org_id}/locations/global" | ||
| client = securitycentermanagement_v1.SecurityCenterManagementClient() | ||
|
|
||
| # Generate a unique display name | ||
| unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}" | ||
| display_name = f"python_sample_etd_custom_module_test_{unique_suffix}" | ||
|
|
||
| # Define the metadata and other config parameters as a dictionary | ||
| config_map = { | ||
| "metadata": { | ||
| "severity": "MEDIUM", | ||
| "description": "Sample custom module for testing purposes. Please do not delete.", | ||
| "recommendation": "na", | ||
| }, | ||
| "ips": ["0.0.0.0"], | ||
| } | ||
|
|
||
| # Convert the dictionary to a Struct | ||
| config_struct = Struct() | ||
| config_struct.update(config_map) | ||
|
|
||
| # Define the custom module configuration | ||
| custom_module = { | ||
| "display_name": display_name, | ||
| "enablement_state": "ENABLED", | ||
| "type_": "CONFIGURABLE_BAD_IP", | ||
| "config": config_struct, | ||
| } | ||
|
|
||
| request = securitycentermanagement_v1.CreateEventThreatDetectionCustomModuleRequest( | ||
| parent=parent, | ||
| event_threat_detection_custom_module=custom_module, | ||
| ) | ||
| response = client.create_event_threat_detection_custom_module(request=request) | ||
| print(f"Created Event Threat Detection Custom Module: {response.name}") | ||
| module_name = response.name | ||
| module_id = module_name.split("/")[-1] | ||
| return module_name, module_id |
There was a problem hiding this comment.
Consider using the client library's create method directly instead of recreating the logic. This will ensure consistency and reduce code duplication.
The add_custom_module function duplicates the logic for creating a custom module, which is already present in the create_event_threat_detection_custom_module function in event_threat_detection_custom_modules.py. This duplication can lead to inconsistencies and make it harder to maintain the code. It's better to reuse the existing function to ensure consistency and reduce code duplication.
return event_threat_detection_custom_modules.create_event_threat_detection_custom_module(parent)| try: | ||
| response = event_threat_detection_custom_modules.delete_event_threat_detection_custom_module(parent, module_id) | ||
| except Exception as e: | ||
| pytest.fail(f"delete_event_threat_detection_custom_module() failed: {e}") | ||
| return |
There was a problem hiding this comment.
Instead of returning after pytest.fail, let the exception propagate. This will provide more context in the test output.
event_threat_detection_custom_modules.delete_event_threat_detection_custom_module(parent, module_id)
There was a problem hiding this comment.
Hi team,
I'm currently reviewing this pull request and will provide my feedback in a few minutes. In the meantime, I'm providing a summary of the pull request's content for you and other reviewers to quickly get up to speed on the changes and intent.
This pull request, authored by vijaykanthm, aims to add Security Health Analytics (SHA) samples for the Google Cloud Security Command Center's Organization Management API. Specifically, it adds samples for creating, getting, listing, deleting, and updating Event Threat Detection Custom Modules. The changes address several internal Google bugs (b/347347902, b/347347344, b/347347881, b/347347878, b/347347934).
Here's a breakdown of the changes:
- New file:
securitycenter/snippets_management_api/event_threat_detection_custom_modules.pyThis file contains the core Python code for the five SHA API operations (create, get, list, delete, update). - New file:
securitycenter/snippets_management_api/event_threat_detection_custom_modules_test.pyThis file contains comprehensive unit tests for the functions inevent_threat_detection_custom_modules.py. The tests include error handling forNotFound,InternalServerError, andServiceUnavailableexceptions. The tests also usebackoffto retry failed API calls. - Updated file:
securitycenter/snippets_management_api/noxfile_config.pyThis file was updated to include configuration for the test environment. - Updated file:
securitycenter/snippets_management_api/requirements-test.txtThis file was updated to include the necessary test dependencies (backoff, pytest, google-cloud-bigquery, google-cloud-securitycentermanagement). - Updated file:
securitycenter/snippets_management_api/requirements.txtThis file was updated to include the necessary runtime dependencies (google-cloud-securitycentermanagement, google-cloud-bigquery, google-cloud-pubsub).
The author has also included a checklist to ensure all necessary steps have been taken before merging. The README file is updated and code owners are specified.
I'll provide a more detailed review shortly.
A little poem for your enjoyment:
Code flows like a stream,
Tests run, a gentle cascade,
Bugs vanish in mist.
| 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. |
There was a problem hiding this comment.
The description mentions Cloud KMS CryptoKeys, but the created module is of type CONFIGURABLE_BAD_IP. Update the description to reflect the actual functionality of the 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. | |
| This custom module creates a configurable bad IP custom module for Event Threat Detection. |
| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
The return type should be EventThreatDetectionCustomModule instead of Dict for better type hinting.
| Dict: Created custom module details. | |
| EventThreatDetectionCustomModule |
| response = client.create_event_threat_detection_custom_module(request=request) | ||
|
|
||
| print(f"Created EventThreatDetectionCustomModule: {response.name}") | ||
| return response |
There was a problem hiding this comment.
It is a good practice to return the created object itself, rather than just printing its name. This allows for easier integration with other functions and modules.
return response| except Exception as e: | ||
| print(f"Failed to create EventThreatDetectionCustomModule: {e}") |
There was a problem hiding this comment.
Use a more specific exception instead of a bare Exception for better error handling.
except google.api_core.exceptions.GoogleAPICallError as e:
print(f"Failed to create EventThreatDetectionCustomModule: {e}")
raise| print(f"Retrieved Event Threat Detection Custom Module: {response.name}") | ||
| return response | ||
| except NotFound as e: | ||
| print(f"Custom Module not found: {response.name}") |
There was a problem hiding this comment.
The variable within the f-string should be module_id instead of response.name, as the response might not be available in case of a NotFound exception.
print(f"Custom Module not found: {module_id}")| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
The return type should be EventThreatDetectionCustomModule instead of Dict for better type hinting.
EventThreatDetectionCustomModule| def add_custom_module(org_id: str): | ||
|
|
||
| parent = f"organizations/{org_id}/locations/global" | ||
| client = securitycentermanagement_v1.SecurityCenterManagementClient() | ||
|
|
||
| # Generate a unique display name | ||
| unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}" | ||
| display_name = f"python_sample_etd_custom_module_test_{unique_suffix}" | ||
|
|
||
| # Define the metadata and other config parameters as a dictionary | ||
| config_map = { | ||
| "metadata": { | ||
| "severity": "MEDIUM", | ||
| "description": "Sample custom module for testing purposes. Please do not delete.", | ||
| "recommendation": "na", | ||
| }, | ||
| "ips": ["0.0.0.0"], | ||
| } | ||
|
|
||
| # Convert the dictionary to a Struct | ||
| config_struct = Struct() | ||
| config_struct.update(config_map) | ||
|
|
||
| # Define the custom module configuration | ||
| custom_module = { | ||
| "display_name": display_name, | ||
| "enablement_state": "ENABLED", | ||
| "type_": "CONFIGURABLE_BAD_IP", | ||
| "config": config_struct, | ||
| } | ||
|
|
||
| request = securitycentermanagement_v1.CreateEventThreatDetectionCustomModuleRequest( | ||
| parent=parent, | ||
| event_threat_detection_custom_module=custom_module, | ||
| ) | ||
| response = client.create_event_threat_detection_custom_module(request=request) | ||
| print(f"Created Event Threat Detection Custom Module: {response.name}") | ||
| module_name = response.name | ||
| module_id = module_name.split("/")[-1] | ||
| return module_name, module_id |
There was a problem hiding this comment.
This helper function duplicates code from the main create_event_threat_detection_custom_module function. Consider refactoring the common code into a separate helper function and reusing it in both places. Also, consider adding a finally block to ensure cleanup of the created custom module in case of test failure.
from event_threat_detection_custom_modules import create_event_threat_detection_custom_module
def add_custom_module(org_id: str):
parent = f"organizations/{org_id}/locations/global"
# Generate a unique display name
unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}"
display_name = f"python_sample_etd_custom_module_test_{unique_suffix}"
try:
response = create_event_threat_detection_custom_module(parent, display_name)
print(f"Created Event Threat Detection Custom Module: {response.name}")
module_name = response.name
module_id = module_name.split('/')[-1]
return module_name, module_id
except Exception as e:
print(f"An error occurred while creating custom module: {e}")
raise
finally:
try:
delete_event_threat_detection_custom_module(parent, module_id)
except NotFound:
pass
There was a problem hiding this comment.
Hi team,
I'm currently reviewing this pull request and will provide my feedback shortly. In the meantime, here's a summary for you and other reviewers to quickly get up to speed:
This pull request, authored by vijaykanthm, aims to add Security Health Analytics (SHA) Organization Management API Event Threat Detection (ETD) custom module samples for creating, getting, listing, deleting, and updating modules. The changes address several internal Google bugs (b/347347902, b/347347344, b/347347881, b/347347878, b/347347934).
Here's a breakdown of the changes:
securitycenter/snippets_management_api/event_threat_detection_custom_modules.py: This file contains the core Python code for interacting with the SHA Organization Management API's ETD custom modules. The changes introduce functions for each CRUD operation (Create, Read, Update, Delete) and listing. The code includes comprehensive comments and error handling.securitycenter/snippets_management_api/event_threat_detection_custom_modules_test.py: This file contains unit tests for the functions inevent_threat_detection_custom_modules.py. The tests use pytest and include robust error handling and cleanup mechanisms to ensure a clean testing environment. The tests cover each CRUD operation and listing.securitycenter/snippets_management_api/noxfile_config.py: This file configures the testing environment, specifying the required Python versions, environment variables, and project IDs. The changes update the configuration to reflect the new requirements for the added samples.securitycenter/snippets_management_api/requirements-test.txtandsecuritycenter/snippets_management_api/requirements.txt: These files specify the project dependencies. The changes add necessary libraries for testing and running the samples.
The author has indicated that the samples require new API enablement and environment variables in testing projects. This information should be clarified before merging.
I'll provide a more detailed review shortly. Thanks!
And now, a little haiku to lighten the mood:
Code flows like a stream,
Tests run, bugs swim downstream,
Clean code, peace of mind.
| 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. |
There was a problem hiding this comment.
The description mentions Cloud KMS CryptoKeys, but the created module is of type CONFIGURABLE_BAD_IP. Update the description to reflect the actual functionality of the 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. | |
| This custom module creates a configurable bad IP custom module, which can be used to detect and block malicious IP addresses. |
| except Exception as e: | ||
| print(f"Failed to create EventThreatDetectionCustomModule: {e}") | ||
| raise |
There was a problem hiding this comment.
Use a more specific exception for the try-except block. Catching a generic Exception can mask unexpected errors. Consider catching specific exceptions like InvalidArgument or PermissionDenied and handle them accordingly.
except google.api_core.exceptions.GoogleAPICallError as e:
print(f"Failed to create EventThreatDetectionCustomModule: {e}")
raise| print(f"Retrieved Event Threat Detection Custom Module: {response.name}") | ||
| return response | ||
| except NotFound as e: | ||
| print(f"Custom Module not found: {response.name}") |
There was a problem hiding this comment.
The variable "response" might not be defined within the NotFound exception block. Use the exception object 'e' instead to print the error message.
print(f"Custom Module not found: {e}")| 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 |
There was a problem hiding this comment.
The generic exception except Exception as e should be removed. It hides potential errors and makes debugging harder. The NotFound exception is already handled, so this is redundant. If there are other expected exceptions, catch them explicitly.
except NotFound as e:
print(f"Parent resource not found: {parent}")
raise e| try: | ||
| response = event_threat_detection_custom_modules.delete_event_threat_detection_custom_module(parent, module_id) | ||
| except Exception as e: | ||
| pytest.fail(f"delete_event_threat_detection_custom_module() failed: {e}") | ||
| return |
There was a problem hiding this comment.
Instead of returning after pytest.fail, let the exception propagate naturally. This will provide a clearer error message in the test output.
except Exception as e:
pytest.fail(f"delete_event_threat_detection_custom_module() failed: {e}")
There was a problem hiding this comment.
Hi team,
I'm currently reviewing this pull request and will provide my feedback shortly. In the meantime, I wanted to give you all a quick summary of the changes and intent to help speed up the review process.
This pull request, authored by vijaykanthm, aims to add Security Health Analytics (SHA) samples for managing Event Threat Detection (ETD) custom modules. Specifically, it adds functionality for creating, getting, listing, deleting, and updating ETD custom modules using the Security Center Management API.
Here's a breakdown of the changes:
- Added
event_threat_detection_custom_modules.py: This file contains the core Python code for interacting with the Security Center Management API to manage ETD custom modules. The code includes functions for each of the CRUD operations (Create, Read, Update, Delete) and List. The functions are well-documented and include examples of how to use them. - Added
event_threat_detection_custom_modules_test.py: This file contains unit tests for the functions inevent_threat_detection_custom_modules.py. The tests use thepytestframework and include tests for each of the CRUD operations and List. The tests are well-structured and include assertions to verify that the functions are working correctly. The tests also include a cleanup function to remove any test modules created during the tests. - Updated
noxfile_config.py: This file was updated to include configuration settings for the tests. Specifically, theenvsdictionary was updated to include theGCLOUD_ORGANIZATIONenvironment variable. This variable is used by the tests to specify the organization ID to use when creating and managing ETD custom modules. - Added
requirements-test.txt: This file was added to specify the dependencies required for running the tests. The dependencies includebackoff,pytest, andgoogle-cloud-bigquery. - Updated
requirements.txt: This file was updated to include the dependencies required for running the samples. The dependencies includegoogle-cloud-securitycentermanagement,google-cloud-bigquery, andgoogle-cloud-pubsub.
The pull request also addresses several internal Google bugs (b/347347902, b/347347344, b/347347881, b/347347878, b/347347934) related to SHA and ETD custom modules.
I'll be checking the code for style, functionality, and test coverage. I'll also be verifying that the samples are easy to use and well-documented.
And now, a little haiku to lighten the mood:
Code flows like a stream,
Tests run, bugs swim downstream,
A cleaner code base.
| 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. |
There was a problem hiding this comment.
The description mentions Cloud KMS CryptoKeys, but the created module is of type CONFIGURABLE_BAD_IP. Update the description to reflect the actual functionality of the 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. | |
| This custom module creates a configurable bad IP custom module, which can be used to detect and block malicious IP addresses. |
| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
The return type should be EventThreatDetectionCustomModule instead of Dict for better type hinting.
| Dict: Created custom module details. | |
| EventThreatDetectionCustomModule |
| response = client.create_event_threat_detection_custom_module(request=request) | ||
|
|
||
| print(f"Created EventThreatDetectionCustomModule: {response.name}") | ||
| return response |
There was a problem hiding this comment.
The return value should be explicitly converted to a dictionary.
return response._pb| except Exception as e: | ||
| print(f"Failed to create EventThreatDetectionCustomModule: {e}") |
There was a problem hiding this comment.
Use a more specific exception instead of a bare Exception for better error handling.
except google.api_core.exceptions.GoogleAPICallError as e:
print(f"Failed to create EventThreatDetectionCustomModule: {e}")
raise| print(f"Retrieved Event Threat Detection Custom Module: {response.name}") | ||
| return response | ||
| except NotFound as e: | ||
| print(f"Custom Module not found: {response.name}") |
There was a problem hiding this comment.
The variable within the f-string should be module_id instead of response.name, as the response might not be available in case of a NotFound exception.
print(f"Custom Module not found: {module_id}")| - folders/{folder_id}/locations/{location_id} | ||
| - projects/{project_id}/locations/{location_id} | ||
| Returns: | ||
| Dict: Created custom module details. |
There was a problem hiding this comment.
The return type should be EventThreatDetectionCustomModule instead of Dict for better type hinting.
EventThreatDetectionCustomModule|
|
||
| # Make the API call | ||
| response = client.update_event_threat_detection_custom_module(request=request) | ||
|
|
There was a problem hiding this comment.
The return value should be explicitly converted to a dictionary.
return response._pb| def add_custom_module(org_id: str): | ||
|
|
||
| parent = f"organizations/{org_id}/locations/global" | ||
| client = securitycentermanagement_v1.SecurityCenterManagementClient() | ||
|
|
||
| # Generate a unique display name | ||
| unique_suffix = f"{int(time.time())}_{random.randint(0, 999)}" | ||
| display_name = f"python_sample_etd_custom_module_test_{unique_suffix}" | ||
|
|
||
| # Define the metadata and other config parameters as a dictionary | ||
| config_map = { | ||
| "metadata": { | ||
| "severity": "MEDIUM", | ||
| "description": "Sample custom module for testing purposes. Please do not delete.", | ||
| "recommendation": "na", | ||
| }, | ||
| "ips": ["0.0.0.0"], | ||
| } | ||
|
|
||
| # Convert the dictionary to a Struct | ||
| config_struct = Struct() | ||
| config_struct.update(config_map) | ||
|
|
||
| # Define the custom module configuration | ||
| custom_module = { | ||
| "display_name": display_name, | ||
| "enablement_state": "ENABLED", | ||
| "type_": "CONFIGURABLE_BAD_IP", | ||
| "config": config_struct, | ||
| } | ||
|
|
||
| request = securitycentermanagement_v1.CreateEventThreatDetectionCustomModuleRequest( | ||
| parent=parent, | ||
| event_threat_detection_custom_module=custom_module, | ||
| ) | ||
| response = client.create_event_threat_detection_custom_module(request=request) | ||
| print(f"Created Event Threat Detection Custom Module: {response.name}") | ||
| module_name = response.name | ||
| module_id = module_name.split("/")[-1] | ||
| return module_name, module_id |
There was a problem hiding this comment.
Consider using the create_event_threat_detection_custom_module function from the main module instead of reimplementing the logic here.
module_name, module_id = event_threat_detection_custom_modules.create_event_threat_detection_custom_module(parent)
return module_name, module_id.split('/')[-1]| try: | ||
| response = event_threat_detection_custom_modules.delete_event_threat_detection_custom_module(parent, module_id) | ||
| except Exception as e: | ||
| pytest.fail(f"delete_event_threat_detection_custom_module() failed: {e}") | ||
| return |
There was a problem hiding this comment.
The try-except block is unnecessary as the delete_event_threat_detection_custom_module function already handles the NotFound exception.
event_threat_detection_custom_modules.delete_event_threat_detection_custom_module(parent, module_id)
Description
Fixes b/347347902, b/347347344, b/347347881, b/347347878, b/347347934
This PR adds SCC Org Managament API ETD Custom Module Samples to Create, Get, List, Delete & Update.
Note: Before submitting a pull request, please open an issue for discussion if you are not associated with Google.
Checklist
nox -s py-3.9(see Test Environment Setup)nox -s lint(see Test Environment Setup)