[Integration][CheckmarxOne] Add Support for DAST Scan

[emekanwaoma]

User description

Description

What - Added DAST (Dynamic Application Security Testing) functionality to Checkmarx One integration

Why - To provide comprehensive security testing coverage including dynamic application security testing alongside existing SAST, KICS, and API security capabilities

How - Implemented new DAST exporters for scan environments, scans, and results with proper pagination, filtering, and async handling

Type of change

Please leave one option from the following and delete the rest:

• New feature (non-breaking change which adds functionality)

All tests should be run against the port production environment(using a testing org).

Core testing checklist

• Integration able to create all default resources from scratch
• Resync finishes successfully
• Resync able to create entities
• Resync able to update entities
• Resync able to detect and delete entities
• Scheduled resync able to abort existing resync and start a new one
• Tested with at least 2 integrations from scratch
• Tested with Kafka and Polling event listeners
• Tested deletion of entities that don't pass the selector

Integration testing checklist

• Integration able to create all default resources from scratch
• Completed a full resync from a freshly installed integration and it completed successfully
• Resync able to create entities
• Resync able to update entities
• Resync able to detect and delete entities
• Resync finishes successfully
• If new resource kind is added or updated in the integration, add example raw data, mapping and expected result to the examples folder in the integration directory.
• If resource kind is updated, run the integration with the example data and check if the expected result is achieved
• If new resource kind is added or updated, validate that live-events for that resource are working as expected
• Docs PR link here

Preflight checklist

• Handled rate limiting
• Handled pagination
• Implemented the code in async
• Support Multi account

Screenshots

Include screenshots from your environment showing how the resources of the integration will look.

API Documentation

Provide links to the API documentation used for this integration.

---

PR Type

Enhancement

---

Description

• Add DAST (Dynamic Application Security Testing) functionality

• Implement three new exporters for environments, scans, and results

• Add comprehensive filtering and pagination support

• Include async handling and proper error management

---

Diagram Walkthrough

flowchart LR
  A["DAST Scan Environment"] --> B["DAST Scan"]
  B --> C["DAST Scan Result"]
  D["Exporter Factory"] --> A
  D --> B
  D --> C
  E["Main Integration"] --> D
  F["Fetcher Module"] --> B
  F --> C

Loading

File Walkthrough

	Relevant files
Enhancement	9 files dast_scan_environment_exporter.py Add DAST scan environment exporter                                              +28/-0    dast_scan_exporter.py Add DAST scan exporter with environment enrichment              +40/-0    dast_scan_result_exporter.py Add DAST scan result exporter with filtering                          +54/-0    options.py Add DAST-related option types and filters                                +36/-0    exporter_factory.py Add factory functions for DAST exporters                                  +27/-0    utils.py Add DAST object kinds to enum                                                        +3/-0      fetcher.py Add DAST scan result fetching logic                                            +84/-0    integration.py Add DAST resource configurations and selectors                      +61/-1    main.py Add DAST resync handlers and imports                                          +61/-0
Bug fix	1 files client.py Improve error logging order in HTTP client                              +3/-4
Tests	6 files test_dast_scan_environment_exporter.py Add comprehensive tests for DAST environment exporter        +179/-0  test_dast_scan_exporter.py Add comprehensive tests for DAST scan exporter                      +277/-0  test_dast_scan_result_exporter.py Add comprehensive tests for DAST result exporter                  +332/-0  test_exporter_factory.py Add tests for new DAST exporter factories                                +50/-0    test_utils.py Update enum tests for new DAST kinds                                          +25/-1    test_fetcher.py Add tests for DAST fetcher functionality                                  +153/-0
Documentation	1 files CHANGELOG.md Document DAST feature additions in changelog                          +10/-0
Configuration changes	1 files pyproject.toml Bump version to 0.1.2                                                                        +1/-1

---

[qodo-merge-pro]

PR Compliance Guide 🔍

Below is a summary of compliance checks for this PR:

Security Compliance
🟢	No security concerns identified No security vulnerabilities detected by AI analysis. Human verification advised for critical code.
Ticket Compliance
⚪	🎫 No ticket provided Create ticket/issue
Codebase Duplication Compliance
⚪	Codebase context is not defined Follow the guide to enable codebase context checks.
Custom Compliance
⚪	No custom compliance provided Follow the guide to enable custom compliance check.
Update

Compliance status legend

🟢 - Fully Compliant
🟡 - Partial Compliant
🔴 - Not Compliant
⚪ - Requires Further Human Verification
🏷️ - Compliance label

[qodo-merge-pro]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Specify resync kinds to prevent errors Update the @ocean.on_resync() decorator for on_scan_result_resync to explicitly handle only sca and containers kinds. This prevents it from incorrectly processing the new dast-scan-result kind and causing errors. integrations/checkmarx-one/main.py [203-244] -@ocean.on_resync() +@ocean.on_resync(ObjectKind.SCA) +@ocean.on_resync(ObjectKind.CONTAINERS) async def on_scan_result_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE: """Resync scan results from Checkmarx One.""" scan_result_exporter = create_scan_result_exporter() scan_exporter = create_scan_exporter() selector = cast(CheckmarxOneScanResultResourcesConfig, event.resource_config).selector async for scans_batch in scan_exporter.get_paginated_resources( ListScanOptions(from_date=selector.from_date) ): for scan in scans_batch: options = ListScanResultOptions( scan_id=scan["id"], result_type=ScanResultObjectKind(kind), filter=selector.filter, ) async for results_batch in scan_result_exporter.get_paginated_resources( options ): yield results_batch [To ensure code accuracy, apply this suggestion manually] Suggestion importance[1-10]: 9 __ Why: The suggestion correctly identifies a critical bug where the generic @ocean.on_resync() handler would incorrectly process the new dast-scan-result kind, leading to a runtime error.	High
High-level	Refactor DAST result resync to stream data Refactor on_dast_scan_result_resync to stream DAST results instead of accumulating them in memory. This avoids high memory usage by yielding results as they are fetched, improving performance and scalability. Examples: integrations/checkmarx-one/main.py [285-298] @ocean.on_resync(ObjectKind.DAST_SCAN_RESULT) async def on_dast_scan_result_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE: """Resync DAST scan results from Checkmarx One.""" dast_scan_environment_exporter = create_dast_scan_environment_exporter() selector = cast( CheckmarxOneDastScanResultResourcesConfig, event.resource_config ).selector async for env_batch in dast_scan_environment_exporter.get_paginated_resources(): ... (clipped 4 lines) integrations/checkmarx-one/fetcher.py [40-84] async def fetch_dast_scan_results_for_environment( dast_scan_environment: Dict[str, Any], selector: "CheckmarxOneDastScanResultSelector", ) -> list[Dict[str, Any]]: """Fetch all scans and their results for a given DAST environment.""" dast_scan_exporter = create_dast_scan_exporter() dast_scan_result_exporter = create_dast_scan_result_exporter() env_id = dast_scan_environment["environmentId"] ... (clipped 35 lines) Solution Walkthrough: Before: @ocean.on_resync(ObjectKind.DAST_SCAN_RESULT) async def on_dast_scan_result_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE: # ... setup exporters and selector async for env_batch in dast_scan_environment_exporter.get_paginated_resources(): results = [] for env in env_batch: # fetch_dast_scan_results_for_environment returns a full list, not a generator results.extend(await fetch_dast_scan_results_for_environment(env, selector)) yield results After: @ocean.on_resync(ObjectKind.DAST_SCAN_RESULT) async def on_dast_scan_result_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE: # ... setup exporters and selector async for env_batch in dast_scan_environment_exporter.get_paginated_resources(): tasks = [ # fetch_dast_scan_results_for_environment should be refactored to be an async generator fetch_dast_scan_results_for_environment(env, selector) for env in env_batch ] async for results_batch in stream_async_iterators_tasks(*tasks): yield results_batch Suggestion importance[1-10]: 8 __ Why: The suggestion correctly identifies a significant performance issue in on_dast_scan_result_resync where results are fully buffered in memory, which could lead to scalability problems, and proposes a valid streaming-based solution.	Medium
General	Improve memory usage by streaming results Refactor on_dast_scan_result_resync to process environments concurrently using asyncio.as_completed. This will stream results as they become available, improving memory efficiency. integrations/checkmarx-one/main.py [285-298] @ocean.on_resync(ObjectKind.DAST_SCAN_RESULT) async def on_dast_scan_result_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE: """Resync DAST scan results from Checkmarx One.""" + import asyncio dast_scan_environment_exporter = create_dast_scan_environment_exporter() selector = cast( CheckmarxOneDastScanResultResourcesConfig, event.resource_config ).selector async for env_batch in dast_scan_environment_exporter.get_paginated_resources(): - results = [] - for env in env_batch: - results.extend(await fetch_dast_scan_results_for_environment(env, selector)) - yield results + if not env_batch: + continue + tasks = [ + asyncio.create_task(fetch_dast_scan_results_for_environment(env, selector)) + for env in env_batch + ] + + for task in asyncio.as_completed(tasks): + try: + results = await task + if results: + yield results + except Exception as e: + logger.warning(f"Failed to fetch DAST scan results for an environment: {e}") + Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly identifies a performance issue where results are aggregated in memory and proposes a valid concurrent processing model, which improves memory efficiency and responsiveness.	Medium
Update

[aws-amplify-eu-west-1]

This pull request is automatically being deployed by Amplify Hosting (learn more).

Access this pull request here: https://pr-2259.d1ftd8v2gowp8w.amplifyapp.com

[mk-armah]

• Sort results by timeframe not exceeding since date,
• Expose since selector 90 days by default
• Expose max_results selector on the kind. (default max count = 3000)
• Cap the results by a max_results not exceeding the since date

[mk-armah]

LGTM