diff --git a/docs/Docusaurus/docs/life_cycle/getting_started.md b/docs/Docusaurus/docs/life_cycle/getting_started.md index 32df692dc..a6cdc94cf 100644 --- a/docs/Docusaurus/docs/life_cycle/getting_started.md +++ b/docs/Docusaurus/docs/life_cycle/getting_started.md @@ -83,10 +83,14 @@ For more information about structure remote config visit [Structure Remote Confi Free - ENGINE_CODE + ENGINE_CODE BEARER Free + + KIUWAN + Paid + ### Scan running - (CLI) - Flags diff --git a/docs/Docusaurus/docs/life_cycle/remote_config_structure.md b/docs/Docusaurus/docs/life_cycle/remote_config_structure.md index 3a468fbff..7b07ad2b4 100644 --- a/docs/Docusaurus/docs/life_cycle/remote_config_structure.md +++ b/docs/Docusaurus/docs/life_cycle/remote_config_structure.md @@ -173,7 +173,7 @@ Configuration of the driven adapters in the main layer and management of on/off }, "ENGINE_CODE": { "ENABLED": true, - "TOOL": "BEARER" + "TOOL": "BEARER|KIUWAN" }, "ENGINE_RISK": { "ENABLED": false diff --git a/docs/Docusaurus/docs/modules/engine_sast/engine_code.md b/docs/Docusaurus/docs/modules/engine_sast/engine_code.md index 178f7f9bc..b9dae43b7 100644 --- a/docs/Docusaurus/docs/modules/engine_sast/engine_code.md +++ b/docs/Docusaurus/docs/modules/engine_sast/engine_code.md @@ -6,7 +6,7 @@ The `engine_code` module is responsible for orchestrating Static Application Sec ## Main Responsibilities -- **SAST Orchestration:** Executes SAST tools (e.g., Bearer) on source code and pull requests. +- **SAST Orchestration:** Executes SAST tools (e.g., Bearer, Kiuwan) on source code and pull requests. - **Configuration Management:** Loads and processes scan configurations and exclusions from remote repositories. - **Pull Request Analysis:** Identifies and filters files changed in pull requests for targeted scanning. - **Exclusions Management:** Applies exclusion rules based on configuration and DevSecOps policy. @@ -17,7 +17,7 @@ The `engine_code` module is responsible for orchestrating Static Application Sec - `runner_engine_code.py`: Main entry point for SAST scan orchestration. - `entry_point_tool.py`: Initializes the SAST engine and triggers the scan process. - `code_scan.py`: Core use case for executing the scan, handling configuration, exclusions, and result aggregation. -- **Adapters:** Integrations for SAST tools (Bearer) and Git operations. +- **Adapters:** Integrations for SAST tools (Bearer, Kiuwan) and Git operations. ## Supported Tools and Features @@ -26,6 +26,11 @@ The `engine_code` module is responsible for orchestrating Static Application Sec - **Configurable Exclusions:** Supports exclusion of files/folders and custom ignore patterns. - **Thresholds and Policies:** Handles custom thresholds and build-breaking policies. +- **Kiuwan:** Aditional SAST tool for scanning source code for security issues and compliances. +- **Pull Request Scanning:** Supports scanning files changed in pull requests and files in a "folder_path" directory. +- **Configurable Exclusions:** Supports exclusion of files/folders and custom ignore patterns. +- **Thresholds and Policies:** Handles custom thresholds and build-breaking policies. + ## Example Usage The SAST engine is typically invoked as part of the overall DevSecOps pipeline, after code changes are detected: @@ -40,6 +45,16 @@ devsecops-engine-tools \ --folder_path path/to/source ``` +```sh +devsecops-engine-tools \ + --platform_devops azure \ + --remote_config_source azure \ + --remote_config_repo my-org/devsecops-config \ + --module engine_code \ + --tool kiuwan \ + --folder_path path/to/source +``` + ## Extensibility - New SAST tools or exclusion policies can be added by extending the adapters and use cases. @@ -47,4 +62,4 @@ devsecops-engine-tools \ ## Testing -- Unit tests are provided in the `test/` directory, covering orchestration logic, configuration parsing, and exclusion handling. +- Unit tests are provided in the `test/` directory, covering orchestration logic, configuration parsing, and exclusion handling. \ No newline at end of file diff --git a/example_remote_config_local/engine_core/ConfigTool.json b/example_remote_config_local/engine_core/ConfigTool.json index 06b204c19..d01847321 100644 --- a/example_remote_config_local/engine_core/ConfigTool.json +++ b/example_remote_config_local/engine_core/ConfigTool.json @@ -123,7 +123,7 @@ }, "ENGINE_CODE": { "ENABLED": true, - "TOOL": "BEARER" + "TOOL": "BEARER|KIUWAN" }, "ENGINE_RISK": { "ENABLED": false diff --git a/example_remote_config_local/engine_sast/engine_code/ConfigTool.json b/example_remote_config_local/engine_sast/engine_code/ConfigTool.json index 18b5bef33..f5db5f09f 100644 --- a/example_remote_config_local/engine_sast/engine_code/ConfigTool.json +++ b/example_remote_config_local/engine_sast/engine_code/ConfigTool.json @@ -1,22 +1,50 @@ { - "IGNORE_SEARCH_PATTERN": [ - ".git" - ], - "EXCLUDE_FOLDER": [], - "MESSAGE_INFO_ENGINE_CODE": "engine_code run successfully", - "THRESHOLD": { - "VULNERABILITY": { - "Critical": 999, - "High": 999, - "Medium": 999, - "Low": 999 - }, - "COMPLIANCE": { - "Critical": 0 - } + "IGNORE_SEARCH_PATTERN": [ + ".git" + ], + "EXCLUDE_FOLDER": [], + "MESSAGE_INFO_ENGINE_CODE": "test message", + "THRESHOLD": { + "VULNERABILITY": { + "Critical": 999, + "High": 999, + "Medium": 999, + "Low": 999 }, - "BEARER": { - "NUMBER_THREADS": 4 + "COMPLIANCE": { + "Critical": 999 + } + }, + "TARGET_BRANCHES": [ + "trunk", + "develop" + ], + "BEARER": { + "NUMBER_THREADS": 4 + }, + "KIUWAN": { + "SERVER": { + "BASE_URL": "https://api.kiuwan.com", + "USER": "user", + "DOMAIN_ID": "12345678910111213141516171819" }, - "TARGET_BRANCHES": ["trunk", "develop"] + "MODELOS": { + "Maven": "Reglas Java", + "Gradle": "Reglas Java", + "Npm": "Reglas Node", + "Python": "Reglas Python", + "Angular": "Reglas Angular", + ".Net": "ReglasNet", + "PHP": "PHP Symfony", + "General": "CQM", + "OWASP": "OWASP-benchmark" + }, + "SEVERITY": { + "Very High": "critical", + "High": "high", + "Normal": "medium", + "Low": "low", + "Very Low": "low" + } + } } \ No newline at end of file diff --git a/example_remote_config_local/engine_sast/engine_code/Exclusions.json b/example_remote_config_local/engine_sast/engine_code/Exclusions.json index 6dba25caf..6cdd0c6d8 100644 --- a/example_remote_config_local/engine_sast/engine_code/Exclusions.json +++ b/example_remote_config_local/engine_sast/engine_code/Exclusions.json @@ -9,6 +9,16 @@ "severity": "Low", "hu": "0000000" } + ], + "KIUWAN": [ + { + "id": "test_id", + "where": "all", + "create_date": "18112023", + "expired_date": "18032024", + "severity": "Low", + "hu": "0000000" + } ] }, "Repository_Test": { @@ -22,5 +32,17 @@ "hu": "0000000" } ] + }, + "Repository_Test_2": { + "KIUWAN": [ + { + "id": "35278412059", + "where": "all", + "create_date": "01082025", + "expired_date": "10082025", + "severity": "high", + "hu": "0000000" + } + ] } } \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py b/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py index 87728e308..3d05bc3f6 100755 --- a/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py +++ b/tools/devsecops_engine_tools/engine_core/src/applications/runner_engine_core.py @@ -35,7 +35,6 @@ from devsecops_engine_tools.engine_utilities import settings from devsecops_engine_tools.version import version - logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() @@ -109,6 +108,7 @@ def get_inputs_from_cli(args): "trivy", "xray", "dependency_check", + "kiuwan", ], type=str, required=False, @@ -199,6 +199,12 @@ def get_inputs_from_cli(args): required=False, help="Token for downloading external checks from engine_iac or engine_secret if is necessary. Ej: github_token:token, github_app:private_key, ssh:privatekey:pass", ) + parser.add_argument( + "--token_engine_code", + type=str, + required=False, + help="Password for connecting with the kiuwan platform. In order to get a kiuwan pass, go to the platform and select the pass of the account selected for the engine." + ) parser.add_argument( "--xray_mode", choices=["scan", "audit","build-scan"], @@ -226,13 +232,20 @@ def get_inputs_from_cli(args): default="false", help="Enable or disable context creation. Applies to engine_iac, engine_container and engine_dependencies. Default is false." ) + parser.add_argument( + "-repo", + "--repo_name", + type=str, + required=False, + help="Repository name, used when the repository name should not be taken from environment variable. Apply to kiuwan" + ) TOOLS = { "engine_iac": ["checkov", "kics", "kubescape"], "engine_secret": ["trufflehog", "gitleaks"], "engine_container": ["prisma", "trivy"], "engine_dependencies": ["xray", "dependency_check", "trivy"], - "engine_code": ["bearer"], + "engine_code": ["bearer", "kiuwan"], "engine_dast": ["nuclei"], "engine_risk": None, } @@ -264,10 +277,12 @@ def get_inputs_from_cli(args): "token_engine_container": args.token_engine_container, "token_engine_dependencies": args.token_engine_dependencies, "token_external_checks": args.token_external_checks, + "token_engine_code": args.token_engine_code, "xray_mode": args.xray_mode, "image_to_scan": args.image_to_scan, "dast_file_path": args.dast_file_path, - "context": args.context + "context": args.context, + "repo_name": args.repo_name } @@ -317,4 +332,4 @@ def application_core(): if __name__ == "__main__": - application_core() + application_core() \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/domain/model/finding.py b/tools/devsecops_engine_tools/engine_core/src/domain/model/finding.py index 0b161259c..e70b8ce5f 100644 --- a/tools/devsecops_engine_tools/engine_core/src/domain/model/finding.py +++ b/tools/devsecops_engine_tools/engine_core/src/domain/model/finding.py @@ -17,4 +17,12 @@ class Finding: module: str category: Category requirements: str - tool: str \ No newline at end of file + tool: str + +@dataclass +class EngineCodeFinding(Finding): + analysis_url: str + analysis_code: str + label: str + application_business_value: str + defect_type: str \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/domain/usecases/break_build.py b/tools/devsecops_engine_tools/engine_core/src/domain/usecases/break_build.py index d531e8716..ab614650e 100644 --- a/tools/devsecops_engine_tools/engine_core/src/domain/usecases/break_build.py +++ b/tools/devsecops_engine_tools/engine_core/src/domain/usecases/break_build.py @@ -50,16 +50,16 @@ def process(self, findings_list: "list[Finding]", input_core: InputCore, args: a findings_excluded, findings_without_exclusions = self._filter_findings(findings_list, exclusions) scan_result["findings_excluded"] = [self._map_finding_excluded(item) for item in findings_excluded] - + vulnerabilities = [v for v in findings_without_exclusions if v.category == Category.VULNERABILITY] compliances = [v for v in findings_without_exclusions if v.category == Category.COMPLIANCE] vulnerability_counts = self._count_severities(vulnerabilities) compliance_counts = self._count_severities(compliances) - self._handle_vulnerabilities(vulnerability_counts, vulnerabilities, threshold, warning_release, scan_result) + self._handle_vulnerabilities(vulnerability_counts, vulnerabilities, threshold, warning_release, scan_result, args) self._handle_cve_policy(vulnerabilities, threshold) - self._handle_compliances(compliance_counts, compliances, threshold, warning_release, scan_result) + self._handle_compliances(compliance_counts, compliances, threshold, warning_release, scan_result, args) self._handle_exclusions(findings_excluded, exclusions) else: print(devops_platform_gateway.message("succeeded", "There are no findings")) @@ -109,7 +109,7 @@ def _map_finding_excluded(self, item): return { "id": item.id, "severity": item.severity, - "category": item.category.value, + "category": item.category, } def _count_severities(self, findings_list): @@ -125,7 +125,7 @@ def _count_severities(self, findings_list): counts[severity] += 1 return counts - def _handle_vulnerabilities(self, counts, vulnerabilities_list, threshold, warning_release, scan_result): + def _handle_vulnerabilities(self, counts, vulnerabilities_list, threshold, warning_release, scan_result, args): devops_platform_gateway = self.devops_platform_gateway printer_table_gateway = self.printer_table_gateway print() @@ -142,6 +142,8 @@ def _handle_vulnerabilities(self, counts, vulnerabilities_list, threshold, warni counts["low"] >= threshold.vulnerability.low): print("Below are all vulnerabilities detected.") + if vulnerabilities_list and args.get("tool", None) == "kiuwan": + print(f"Analysis url: {vulnerabilities_list[0].analysis_url}") printer_table_gateway.print_table_findings(vulnerabilities_list) print(devops_platform_gateway.message( "error", @@ -194,13 +196,15 @@ def _handle_cve_policy(self, vulnerabilities_list: "list[Finding]", threshold): )) print(devops_platform_gateway.result_pipeline("failed")) - def _handle_compliances(self, counts, compliances_list, threshold, warning_release, scan_result): + def _handle_compliances(self, counts, compliances_list, threshold, warning_release, scan_result, args): devops_platform_gateway = self.devops_platform_gateway printer_table_gateway = self.printer_table_gateway print() if compliances_list: print("Below are all compliances issues detected.") + if compliances_list and args.get("tool", None) == "kiuwan": + print(f"Analysis url: {compliances_list[0].analysis_url}") printer_table_gateway.print_table_findings(compliances_list) status = "succeeded" if counts["critical"] >= threshold.compliance.critical: @@ -251,4 +255,4 @@ def _handle_exclusions(self, findings_excluded_list, exclusions): printer_table_gateway.print_table_exclusions(exclusions_list) for reason, total in Counter(x["reason"] for x in exclusions_list).items(): - print("{0} findings count: {1}".format(reason, total)) + print("{0} findings count: {1}".format(reason, total)) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/azure/azure_devops.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/azure/azure_devops.py index 306be4cc8..fd85b9616 100644 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/azure/azure_devops.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/azure/azure_devops.py @@ -8,6 +8,7 @@ ReleaseVariables, AgentVariables, VMVariables, + ApplicationVariables, ) from devsecops_engine_tools.engine_utilities.azuredevops.infrastructure.azure_devops_api import ( AzureDevopsApi, @@ -117,8 +118,9 @@ def get_variable(self, variable): "vm_product_type_name": VMVariables.Vm_Product_Type_Name, "vm_product_name": VMVariables.Vm_Product_Name, "vm_product_description": VMVariables.Vm_Product_Description, + "build_task": ApplicationVariables.Application_Build_Task, } try: return variable_map.get(variable).value() except ValueError: - return None + return None \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/defect_dojo/defect_dojo.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/defect_dojo/defect_dojo.py index 733e0c160..fcd6308c7 100755 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/defect_dojo/defect_dojo.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/defect_dojo/defect_dojo.py @@ -40,7 +40,6 @@ logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() - @dataclass class DefectDojoPlatform(VulnerabilityManagementGateway): @@ -70,6 +69,7 @@ class DefectDojoPlatform(VulnerabilityManagementGateway): "SONARQUBE": "SonarQube API Import", "GITLEAKS": "Gitleaks Scan", "NUCLEI": "Nuclei Scan", + "KIUWAN": "Kiuwan Scan" } def send_vulnerability_management( @@ -88,6 +88,7 @@ def send_vulnerability_management( else vulnerability_management.secret_tool["token_cmdb"] ) + tags = [] if any( branch in str(vulnerability_management.branch_tag) for branch in vulnerability_management.config_tool[ @@ -95,12 +96,14 @@ def send_vulnerability_management( ]["BRANCH_FILTER"] ) or (vulnerability_management.dict_args["module"] == "engine_secret"): tags = [vulnerability_management.dict_args["module"]] + if vulnerability_management.dict_args["module"] == "engine_iac": tags = [ f"{vulnerability_management.dict_args['module']}_{'_'.join(vulnerability_management.dict_args['platform'])}" ] if vulnerability_management.input_core.scope_service != vulnerability_management.input_core.scope_pipeline: tags.append(vulnerability_management.input_core.scope_service.replace(f"{vulnerability_management.input_core.scope_pipeline}_", "")) + if ( vulnerability_management.dict_args["module"] == "engine_container" and sum( @@ -122,47 +125,46 @@ def send_vulnerability_management( f"{vulnerability_management.dict_args['module']}_{tag_suffix}" ] - use_cmdb = vulnerability_management.config_tool[ - "VULNERABILITY_MANAGER" - ]["DEFECT_DOJO"]["CMDB"]["USE_CMDB"] - - request = self._build_request_importscan( - vulnerability_management, - token_cmdb, - token_dd, - tags, - use_cmdb, - ) + use_cmdb = vulnerability_management.config_tool[ + "VULNERABILITY_MANAGER" + ]["DEFECT_DOJO"]["CMDB"]["USE_CMDB"] - def request_func(): - return DefectDojo.send_import_scan(request) + request = self._build_request_importscan( + vulnerability_management, + token_cmdb, + token_dd, + tags, + use_cmdb, + ) - response = Utils().retries_requests( - request_func, - vulnerability_management.config_tool["VULNERABILITY_MANAGER"][ - "DEFECT_DOJO" - ]["MAX_RETRIES_QUERY"], - retry_delay=5, - ) + def request_func(): + return DefectDojo.send_import_scan(request) + + response = Utils().retries_requests( + request_func, + vulnerability_management.config_tool["VULNERABILITY_MANAGER"][ + "DEFECT_DOJO" + ]["MAX_RETRIES_QUERY"], + retry_delay=5, + ) - if hasattr(response, "url"): - if vulnerability_management.config_tool.get("VULNERABILITY_MANAGER").get("DEFECT_DOJO").get("PRINT_DOMAIN"): - response.url = response.url.replace(vulnerability_management.config_tool["VULNERABILITY_MANAGER"][ - "DEFECT_DOJO"]["HOST_DEFECT_DOJO" - ], vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["PRINT_DOMAIN"]) - url_parts = response.url.split("//") - test_string = "//".join([url_parts[0] + "/", url_parts[1]]) - print( - "Report sent to vulnerability management: ", - f"{test_string}?tags={vulnerability_management.dict_args['module']}", + if hasattr(response, "url"): + if vulnerability_management.config_tool.get("VULNERABILITY_MANAGER").get("DEFECT_DOJO").get("PRINT_DOMAIN"): + response.url = response.url.replace( + vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["HOST_DEFECT_DOJO"], + vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["PRINT_DOMAIN"] ) - else: - raise ExceptionVulnerabilityManagement(response) + url_parts = response.url.split("//") + test_string = "//".join([url_parts[0] + "/", url_parts[1]]) + print( + "Report sent to vulnerability management: ", + f"{test_string}?tags={vulnerability_management.dict_args['module']}", + ) + else: + raise ExceptionVulnerabilityManagement(response) except Exception as ex: raise ExceptionVulnerabilityManagement( - "Error sending report to vulnerability management with the following error: {0} ".format( - ex - ) + f"Error sending report to vulnerability management with the following error: {str(ex)}" ) def get_product_type_pipeline(self, service, dict_args, secret_tool, config_tool): @@ -198,10 +200,11 @@ def request_func(): except Exception as ex: raise ExceptionVulnerabilityManagement( - "Error getting product type with the following error: {0} ".format(ex) + f"Error getting product type with the following error: {str(ex)}" ) def get_findings_excepted(self, service, dict_args, secret_tool, config_tool): + logger.info("Starting get_findings_excepted") try: session_manager = self._get_session_manager( dict_args, secret_tool, config_tool @@ -213,7 +216,6 @@ def get_findings_excepted(self, service, dict_args, secret_tool, config_tool): dd_max_retries = config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"][ "MAX_RETRIES_QUERY" ] - tool = dict_args["module"] risk_accepted_query_params = { @@ -302,18 +304,18 @@ def get_findings_excepted(self, service, dict_args, secret_tool, config_tool): white_list=white_list, ) - return ( + result = ( list(exclusions_risk_accepted) + list(exclusions_false_positive) + list(exclusions_out_of_scope) + list(exclusions_transfer_finding) + list(exclusions_white_list) ) + return result + except Exception as ex: raise ExceptionFindingsExcepted( - "Error getting excepted findings with the following error: {0} ".format( - ex - ) + f"Error getting excepted findings with the following error: {str(ex)}" ) def get_all(self, service, dict_args, secret_tool, config_tool): @@ -368,7 +370,7 @@ def get_all(self, service, dict_args, secret_tool, config_tool): except Exception as ex: raise ExceptionGettingFindings( - "Error getting all findings with the following error: {0} ".format(ex) + f"Error getting all findings with the following error: {str(ex)}" ) def get_active_engagements( @@ -411,7 +413,7 @@ def get_active_engagements( except Exception as ex: raise ExceptionGettingEngagements( - "Error getting engagements with the following error: {0} ".format(ex) + f"Error getting engagements with the following error: {str(ex)}" ) def send_sbom_components( @@ -429,7 +431,7 @@ def send_sbom_components( ) with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor: - _ = [ + futures = [ executor.submit( self._process_component, sbom_component, @@ -438,12 +440,11 @@ def send_sbom_components( ) for sbom_component in sbom_components ] + concurrent.futures.wait(futures) except Exception as ex: raise ExceptionVulnerabilityManagement( - "Error sending components sbom to vulnerability management with the following error: {0} ".format( - ex - ) + f"Error sending components sbom to vulnerability management with the following error: {str(ex)}" ) def get_black_list(self, dict_args, secret_tool, config_tool): @@ -462,10 +463,12 @@ def get_black_list(self, dict_args, secret_tool, config_tool): }, ) - return [entry.unique_id_from_tool for entry in exclusions_black_list] + result = [entry.unique_id_from_tool for entry in exclusions_black_list] + return result + except Exception as ex: raise ExceptionVulnerabilityManagement( - "Error getting black list with the following error: {0} ".format(ex) + f"Error getting black list with the following error: {str(ex)}" ) def _build_request_importscan( @@ -478,6 +481,7 @@ def _build_request_importscan( ): tool_scm_conf_mapping = vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["TOOL_SCM_MAPPING"] tool_sonar_conf_mapping = vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["TOOL_SONAR_MAPPING"] + common_fields = { "scan_type": self.scan_type_mapping[vulnerability_management.scan_type], "file": vulnerability_management.input_core.path_file_results, @@ -532,7 +536,7 @@ def _build_request_importscan( cmdb_mapping = vulnerability_management.config_tool[ "VULNERABILITY_MANAGER" ]["DEFECT_DOJO"]["CMDB"]["CMDB_MAPPING"] - return Connect.cmdb( + request = Connect.cmdb( generate_auth_cmdb=vulnerability_management.config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["CMDB"]["GENERATE_AUTH_CMDB"], auth_cmdb_request_response=vulnerability_management.config_tool[ "VULNERABILITY_MANAGER" @@ -556,7 +560,7 @@ def _build_request_importscan( **common_fields, ) else: - request: ImportScanRequest = ImportScanSerializer().load( + request = ImportScanSerializer().load( { "product_type_name": vulnerability_management.vm_product_type_name, "product_name": vulnerability_management.vm_product_name, @@ -565,7 +569,8 @@ def _build_request_importscan( **common_fields, } ) - return request + + return request def _process_component(self, component_sbom, session_manager, engagement): request = { @@ -586,10 +591,11 @@ def _get_session_manager(self, dict_args, secret_tool, config_tool): token_dd = dict_args.get("token_vulnerability_management") or secret_tool.get( "token_defect_dojo" ) - return SessionManager( + session_manager = SessionManager( token_dd, config_tool["VULNERABILITY_MANAGER"]["DEFECT_DOJO"]["HOST_DEFECT_DOJO"], ) + return session_manager def _get_report_exclusions(self, total_findings, date_fn, host_dd, **kwargs): exclusions = [] @@ -665,8 +671,8 @@ def _get_findings_with_exclusions( findings = self._get_findings( session_manager, service, max_retries, query_params ) - - return map( + + result = map( partial( self._create_exclusion, date_fn=date_fn, @@ -676,6 +682,7 @@ def _get_findings_with_exclusions( ), findings, ) + return result def _get_findings(self, session_manager, service, max_retries, query_params): def request_func(): @@ -683,7 +690,8 @@ def request_func(): session=session_manager, service=service, **query_params ).results - return Utils().retries_requests(request_func, max_retries, retry_delay=5) + findings = Utils().retries_requests(request_func, max_retries, retry_delay=5) + return findings def _get_finding_exclusion(self, session_manager, max_retries, query_params): def request_func(): @@ -691,31 +699,33 @@ def request_func(): session=session_manager, **query_params ).results - return Utils().retries_requests(request_func, max_retries, retry_delay=5) + exclusions = Utils().retries_requests(request_func, max_retries, retry_delay=5) + return exclusions def _date_reason_based(self, finding, date_fn, reason, tool, **kwargs): def get_vuln_id(finding, tool): if tool == "engine_risk": - return ( + vuln_id = ( finding.id[0]["vulnerability_id"] if finding.id else finding.vuln_id_from_tool ) else: - return ( + vuln_id = ( finding.vulnerability_ids[0]["vulnerability_id"] if finding.vulnerability_ids else finding.vuln_id_from_tool ) + return vuln_id def get_dates_from_whitelist(vuln_id, white_list): matching_finding = next( filter(lambda x: x.unique_id_from_tool == vuln_id, white_list), None ) if matching_finding: - return date_fn(matching_finding.create_date), date_fn( - matching_finding.expiration_date - ) + create_date = date_fn(matching_finding.create_date) + expiration_date = date_fn(matching_finding.expiration_date) + return create_date, expiration_date return date_fn(None), date_fn(None) reason_to_dates = { @@ -749,8 +759,7 @@ def _create_exclusion(self, finding, date_fn, tool, reason, **kwargs): create_date, expired_date = self._date_reason_based( finding, date_fn, reason, tool, **kwargs ) - - return Exclusions( + exclusion = Exclusions( id=( finding.vuln_id_from_tool if finding.vuln_id_from_tool @@ -766,6 +775,7 @@ def _create_exclusion(self, finding, date_fn, tool, reason, **kwargs): severity=finding.severity.lower(), reason=reason, ) + return exclusion def _create_report_exclusion( self, finding, date_fn, tool, reason, host_dd, **kwargs @@ -773,8 +783,7 @@ def _create_report_exclusion( create_date, expired_date = self._date_reason_based( finding, date_fn, reason, tool, **kwargs ) - - return Exclusions( + exclusion = Exclusions( id=( finding.vuln_id_from_tool if finding.vuln_id_from_tool @@ -790,9 +799,10 @@ def _create_report_exclusion( service=finding.service, tags=finding.tags, ) + return exclusion def _create_report(self, finding, host_dd): - return Report( + report = Report( vm_id=str(finding.id), vm_id_url=f"{host_dd}/finding/{finding.id}", id=finding.vulnerability_ids, @@ -824,29 +834,33 @@ def _create_report(self, finding, host_dd): service=finding.service, unique_id_from_tool=finding.unique_id_from_tool, ) + return report def _format_date_to_dd_format(self, date_string): - return ( + result = ( format_date(date_string.split("T")[0], "%Y-%m-%d", "%d%m%Y") if date_string else None ) + return result def _get_where(self, finding, tool): if tool == "engine_dependencies": - return ( + result = ( finding.component_name.replace("_", ":") + ":" + finding.component_version ) elif tool == "engine_container": - return finding.component_name + ":" + finding.component_version + result = finding.component_name + ":" + finding.component_version elif tool == "engine_dast": - return finding.endpoints + result = finding.endpoints elif tool == "engine_risk": for tag in finding.tags: - return self._get_where(finding, tag) - return finding.file_path + result = self._get_where(finding, tag) + return result + result = finding.file_path else: + return finding.file_path \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/github/github_actions.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/github/github_actions.py index e8add039e..50a0e349d 100755 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/github/github_actions.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/github/github_actions.py @@ -7,7 +7,8 @@ SystemVariables, ReleaseVariables, AgentVariables, - VMVariables + VMVariables, + ApplicationVariables ) from devsecops_engine_tools.engine_utilities.github.infrastructure.github_api import ( GithubApi, @@ -94,6 +95,7 @@ def get_variable(self, variable): "vm_product_type_name": VMVariables.Vm_Product_Type_Name, "vm_product_name": VMVariables.Vm_Product_Name, "vm_product_description": VMVariables.Vm_Product_Description, + "build_task": ApplicationVariables.Application_Build_Task, } try: return variable_map.get(variable).value() diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_pretty_table/printer_pretty_table.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_pretty_table/printer_pretty_table.py index 0d5ef512b..50e22e144 100644 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_pretty_table/printer_pretty_table.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/printer_pretty_table/printer_pretty_table.py @@ -19,7 +19,7 @@ class PrinterPrettyTable(PrinterTableGateway): def _create_table(self, headers, finding_list): table = PrettyTable(headers) - + for finding in finding_list: row_data = [ finding.severity, @@ -31,7 +31,9 @@ def _create_table(self, headers, finding_list): finding.module == "engine_dependencies" ): row_data.append(finding.requirements) - + elif finding.module == "engine_code": + row_data.append(finding.cvss) + row_data.append(finding.defect_type) table.add_row(row_data) severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unknown": 4} @@ -50,12 +52,17 @@ def _create_table(self, headers, finding_list): def print_table_findings(self, finding_list: "list[Finding]"): if ( finding_list - and (finding_list[0].module != "engine_container") - and (finding_list[0].module != "engine_dependencies") + and ( + (finding_list[0].module == "engine_container") + or (finding_list[0].module == "engine_dependencies") + ) ): - headers = ["Severity", "ID", "Description", "Where"] - else: headers = ["Severity", "ID", "Description", "Where", "Fixed in"] + elif finding_list and finding_list[0].module == "engine_code": + headers = ["Severity", "ID", "Description", "Where", "Rule", "Defect Type"] + + else: + headers = ["Severity", "ID", "Description", "Where"] sorted_table = self._create_table(headers, finding_list) @@ -165,4 +172,4 @@ def _check_spaces(self, value): new_value = "\n".join(values) else: new_value = f"{values[0]}" - return new_value + return new_value \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/runtime_local/runtime_local.py b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/runtime_local/runtime_local.py index 7a148b19c..5c2ed9414 100755 --- a/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/runtime_local/runtime_local.py +++ b/tools/devsecops_engine_tools/engine_core/src/infrastructure/driven_adapters/runtime_local/runtime_local.py @@ -76,5 +76,6 @@ def get_variable(self, variable): "vm_product_type_name" : "DET_VM_PRODUCT_TYPE_NAME", "vm_product_name" : "DET_VM_PRODUCT_NAME", "vm_product_description" : "DET_VM_PRODUCT_DESCRIPTION", + "build_task": "DET_APPLICATION_BUILD_TASK", } return os.environ.get(env_variables[variable], None) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_core/test/applications/test_runner_engine_core.py b/tools/devsecops_engine_tools/engine_core/test/applications/test_runner_engine_core.py index 1a7b10645..6c1c92237 100644 --- a/tools/devsecops_engine_tools/engine_core/test/applications/test_runner_engine_core.py +++ b/tools/devsecops_engine_tools/engine_core/test/applications/test_runner_engine_core.py @@ -110,6 +110,8 @@ def test_get_inputs_from_cli(mock_parse_args): mock_args.image_to_scan = "image" mock_args.dast_file_path = "dast_file_path" mock_args.context = "false" + mock_args.repo_name = None + mock_args.token_engine_code=None # Mock the parse_args method mock_parse_args.return_value = mock_args @@ -140,6 +142,8 @@ def test_get_inputs_from_cli(mock_parse_args): "image_to_scan": "image", "dast_file_path": "dast_file_path", "context": "false", + "repo_name" : None, + "token_engine_code": None } diff --git a/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_break_build.py b/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_break_build.py index eadf77c48..b9e24241e 100644 --- a/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_break_build.py +++ b/tools/devsecops_engine_tools/engine_core/test/domain/usecases/test_break_build.py @@ -329,11 +329,11 @@ def test_process_with_findings_succeeded(self): result_compare = { "findings_excluded": [ - {"id": "CKV_DOCKER_3", "severity": "high", "category": "vulnerability"}, - {"id": "CKV_K8S_20", "severity": "high", "category": "vulnerability"}, + {"id": "CKV_DOCKER_3", "severity": "high", "category": Category.VULNERABILITY}, + {"id": "CKV_K8S_20", "severity": "high", "category": Category.VULNERABILITY}, ], "vulnerabilities": {}, "compliances": {}, } - assert result == result_compare + self.assertEqual(result, result_compare) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_config.py b/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_config.py index 5ef02b39c..5e199b597 100644 --- a/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_config.py +++ b/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_config.py @@ -1,3 +1,4 @@ +import os import unittest from unittest.mock import Mock, patch, mock_open from devsecops_engine_tools.engine_dast.src.infrastructure.driven_adapters.nuclei.nuclei_config import NucleiConfig @@ -80,5 +81,6 @@ def test_process_template_file(self, mock_dump, mock_load, mock_open): def test_customize_templates(self, mock_process_templates_folder): directory = "dummy_directory" self.nuclei_api.customize_templates(directory) - self.assertEqual(self.nuclei_api.custom_templates_dir, "dummy_directory/customized-nuclei-templates") - mock_process_templates_folder.assert_any_call(base_folder=directory) \ No newline at end of file + self.assertEqual(self.nuclei_api.custom_templates_dir, os.path.join("dummy_directory","customized-nuclei-templates")) + mock_process_templates_folder.assert_any_call(base_folder=directory) + \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_tool.py b/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_tool.py index 7bce994f3..71e8fce42 100644 --- a/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_tool.py +++ b/tools/devsecops_engine_tools/engine_dast/test/infrastructure/driven_adapters/nuclei/test_nuclei_tool.py @@ -1,3 +1,4 @@ +import platform import unittest from unittest.mock import Mock, patch, mock_open from devsecops_engine_tools.engine_dast.src.infrastructure.driven_adapters.nuclei.nuclei_tool import ( @@ -60,10 +61,13 @@ def test_install_tool_linux(self, mock_download_tool, mock_expanduser, mock_subp mock_download_tool.return_value = 0 result = self.nuclei_tool.install_tool(self.version, '/home/user') - - self.assertEqual(result["status"], 201) + os_type = platform.system().lower() + if os_type == "windows": + self.assertEqual(result["status"], 202) + else: + self.assertEqual(result["status"], 201) + mock_subprocess_run.assert_called_once_with(["chmod", "+x", "/home/user/nuclei"], check=True) mock_download_tool.assert_called_once() - mock_subprocess_run.assert_called_once_with(["chmod", "+x", "/home/user/nuclei"], check=True) @patch('builtins.open', new_callable=mock_open, read_data='{"key": "value"}') @patch('json.load', return_value={"key": "value"}) diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/applications/runner_engine_code.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/applications/runner_engine_code.py index dac72e446..68e80fe18 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_code/src/applications/runner_engine_code.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/applications/runner_engine_code.py @@ -7,13 +7,29 @@ from devsecops_engine_tools.engine_utilities.git_cli.infrastructure.git_run import ( GitRun ) +from devsecops_engine_tools.engine_utilities.utils.logger_info import MyLogger +from devsecops_engine_tools.engine_utilities import settings +from devsecops_engine_tools.engine_sast.engine_code.src.infrastructure.driven_adapters.kiuwan.kiuwan_tool import get_kiuwan_instance + + +logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() def runner_engine_code(dict_args, tool, devops_platform_gateway, remote_config_source_gateway): try: + logger.info("Selecting tool...") tool_gateway = None git_gateway = GitRun() if (tool == "BEARER"): + logger.info("Bearer tool selected...") tool_gateway = BearerTool() + elif (tool == "KIUWAN"): + logger.info("Kiuwan tool selected...") + tool_gateway = get_kiuwan_instance( + dict_args=dict_args, + devops_platform_gateway=devops_platform_gateway, + ) + + logger.info("Tool has been selected successfully") return init_engine_sast_code( devops_platform_gateway=devops_platform_gateway, @@ -29,4 +45,4 @@ def runner_engine_code(dict_args, tool, devops_platform_gateway, remote_config_s if __name__ == "__main__": - runner_engine_code() + runner_engine_code() \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/model/gateways/tool_gateway.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/model/gateways/tool_gateway.py index 4ecc620d2..135fc91f7 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/model/gateways/tool_gateway.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/model/gateways/tool_gateway.py @@ -1,15 +1,28 @@ from abc import ABCMeta, abstractmethod -from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.config_tool import ( - ConfigTool, -) +from typing import List, Optional, Any, Tuple +from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.config_tool import ConfigTool class ToolGateway(metaclass=ABCMeta): - @abstractmethod - def run_tool(self, - folder_to_scan: str, - pull_request_files: list, - agent_work_folder: str, - repository: str, - config_tool: ConfigTool): - "run code scan tool" \ No newline at end of file + def run_tool( + self, + folder_to_scan: str, + pull_request_files: List[str], + agent_work_folder: str, + repository: str, + config_tool: ConfigTool + ) -> Tuple[List[Any], Optional[str]]: + """ + Run the code scan tool. + Args: + folder_to_scan: Path to the folder to scan + pull_request_files: List of files modified in the pull request + agent_work_folder: Directory for storing temporary files + repository: Name of the repository + config_tool: Configuration object for the tool + Returns: + Tuple containing: + - List of findings (vulnerabilities or defects). + - Path to the results file, or None if no file is generated. + """ + pass \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/usecases/code_scan.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/usecases/code_scan.py index fcb3dc3c0..38af49489 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/usecases/code_scan.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/domain/usecases/code_scan.py @@ -1,4 +1,5 @@ import re +from typing import Any, Dict, List, Tuple from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.gateways.tool_gateway import ( ToolGateway, ) @@ -33,9 +34,9 @@ def __init__( self.remote_config_source_gateway = remote_config_source_gateway self.git_gateway = git_gateway - def set_config_tool(self, dict_args): + def set_config_tool(self, dict_args: Dict[str, Any], config_tool_path: str): init_config_tool = self.remote_config_source_gateway.get_remote_config( - dict_args["remote_config_repo"], "engine_sast/engine_code/ConfigTool.json", dict_args["remote_config_branch"] + dict_args["remote_config_repo"], config_tool_path, dict_args["remote_config_branch"] ) scope_pipeline = self.devops_platform_gateway.get_variable("pipeline_name") return ConfigTool(json_data=init_config_tool, scope=scope_pipeline) @@ -88,35 +89,112 @@ def apply_exclude_path( return True return False - def process(self, dict_args, tool): - config_tool = self.set_config_tool(dict_args) + def _get_config_tool_and_exclusions_data(self, dict_args, tool) -> Tuple[Dict[str, Any], Dict[str, str]]: + + """ + Get the information related to configuracion and exclusiones for the selected tool. + + Parameters: + dict_args: Dictionary with properties setting up from the extension. + tool: String name of the tool that will be used to run scan. + + Returns: + config_tool: Dictionary with the configuration of the tool. + excusions_data: Dictionary with the exclusions configured for an specific tool and pipelines. + """ + logger.info("Getting engine_code config tool and exclusions...") + config_tool = self.set_config_tool(dict_args, "engine_sast/engine_code/ConfigTool.json") exclusions_data = self.remote_config_source_gateway.get_remote_config( - dict_args["remote_config_repo"], "engine_sast/engine_code/Exclusions.json" + dict_args["remote_config_repo"], "engine_sast/engine_code/Exclusions.json", dict_args["remote_config_branch"] + ) + return config_tool, exclusions_data + + def _get_filtered_pr_files(self, dict_args: Dict[str,str], config_tool: Dict[str, Any]) -> List[str]: + """ + Retrieve and filter pull request files based on exclusion rules. + + Parameters: + config_tool: Configuration dictionary for the SAST tool. + + Returns: + List of filtered pull request file paths. + """ + pull_request_files = [] + if not dict_args["folder_path"]: + pull_request_files = self.get_pull_request_files( + config_tool.target_branches + ) + pull_request_files = [ + pf + for pf in pull_request_files + if not self.apply_exclude_path( + config_tool.exclude_folder, + config_tool.ignore_search_pattern, + pf, + ) + ] + return pull_request_files + + def _create_input_core(self, list_exclusions: List[str], config_tool: Dict[str, Any], exclusions_data: Dict[str, str], path_file_results: str) -> 'InputCore': + """ + Create an InputCore object with pipeline and tool configuration. + + Parameters: + list_exclusions: List of excluded files or patterns. + config_tool: Configuration dictionary for the SAST tool. + exclusions_data: Exclusion data for the tool. + path_file_results: Path to the results file. + + Returns: + InputCore object with pipeline and tool configuration. + """ + return InputCore( + totalized_exclusions=list_exclusions, + threshold_defined=Utils.update_threshold( + self, + config_tool.threshold, + exclusions_data, + config_tool.scope_pipeline, + ), + path_file_results=path_file_results, + custom_message_break_build=config_tool.message_info_engine_code, + scope_pipeline=config_tool.scope_pipeline, + scope_service=config_tool.scope_pipeline, + stage_pipeline=self.devops_platform_gateway.get_variable("stage").capitalize(), ) + + def process(self, dict_args, tool): + + """ + This function process the request to a new scan code. + + Parameters: + dict_args: Dictionary with properties setting up from the extension. + tool: String name of the tool that will be used to run scan. + + Returns: + findings_list: List with the defects founded during analysis. + input_core: InputCore instance with properties related to scan proccess. + """ + + # Retrieve the config tool and exclusions data for the tool used during scan + config_tool, exclusions_data = self._get_config_tool_and_exclusions_data( + dict_args, tool + ) + list_exclusions, skip_tool = self.get_exclusions(tool, exclusions_data) findings_list, path_file_results = [], "" if not skip_tool: - pull_request_files = [] - if not dict_args["folder_path"]: - pull_request_files = self.get_pull_request_files( - config_tool.target_branches - ) - pull_request_files = [ - pf - for pf in pull_request_files - if not self.apply_exclude_path( - config_tool.exclude_folder, - config_tool.ignore_search_pattern, - pf, - ) - ] + # Retrieve the pull requests files + # If folder path was not added in dict args, the pr files will be downloaded excluding the paths excluded + pull_request_files = self._get_filtered_pr_files(dict_args, config_tool) findings_list, path_file_results = self.tool_gateway.run_tool( dict_args["folder_path"], pull_request_files, self.devops_platform_gateway.get_variable("path_directory"), - self.devops_platform_gateway.get_variable("repository"), + dict_args.get("repo_name", False) if dict_args.get("repo_name", False) else self.devops_platform_gateway.get_variable("repository"), config_tool, ) @@ -125,21 +203,6 @@ def process(self, dict_args, tool): dict_args["send_metrics"] = "false" dict_args["use_vulnerability_management"] = "false" - input_core = InputCore( - totalized_exclusions=list_exclusions, - threshold_defined=Utils.update_threshold( - self, - config_tool.threshold, - exclusions_data, - config_tool.scope_pipeline, - ), - path_file_results=path_file_results, - custom_message_break_build=config_tool.message_info_engine_code, - scope_pipeline=config_tool.scope_pipeline, - scope_service=config_tool.scope_pipeline, - stage_pipeline=self.devops_platform_gateway.get_variable( - "stage" - ).capitalize(), - ) + input_core = self._create_input_core(list_exclusions, config_tool, exclusions_data, path_file_results) - return findings_list, input_core + return findings_list, input_core \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/bearer/bearer_deserealizator.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/bearer/bearer_deserealizator.py index e72c55e9a..249f5f3d8 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/bearer/bearer_deserealizator.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/bearer/bearer_deserealizator.py @@ -1,6 +1,6 @@ from devsecops_engine_tools.engine_core.src.domain.model.finding import ( Category, - Finding, + EngineCodeFinding, ) from datetime import datetime from dataclasses import dataclass @@ -12,7 +12,7 @@ class BearerDeserealizator: @classmethod def get_list_finding(cls, scan_result_path, - agent_work_folder) -> "list[Finding]": + agent_work_folder) -> "list[EngineCodeFinding]": findings = [] with open(scan_result_path, encoding='utf-8') as arc: try: @@ -30,7 +30,7 @@ def get_list_finding(cls, chunks = [description[i : i + 70] for i in range(0, len(description), 70)] formatted_description = "\n".join(chunks) + "\n" - finding = Finding( + finding = EngineCodeFinding( id=vul["id"], cvss="", where=vul["full_filename"].replace(agent_work_folder, "").replace("/copy_files_bearer", ""), @@ -41,7 +41,12 @@ def get_list_finding(cls, module="engine_code", category=Category.VULNERABILITY, requirements="", - tool="Bearer" + tool="Bearer", + analysis_url=None, + analysis_code=None, + label=None, + application_business_value=None, + defect_type=None ) findings.append(finding) diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/__init__.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_deserealizator.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_deserealizator.py new file mode 100644 index 000000000..5ef85a56f --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_deserealizator.py @@ -0,0 +1,59 @@ +from datetime import datetime +from typing import Dict, Any, List + +from devsecops_engine_tools.engine_core.src.domain.model.finding import Category, EngineCodeFinding + + +class KiuwanDeserealizator: + + """ + This class has the functions to deserealize the defects found in kiuwan analysis scan. + """ + + @staticmethod + def get_findings( + last_analysis: Dict[str, Any], + defects_data: Dict[str, Any], + analysis_code: str, + severity_mapper: Dict[str,str], + ) -> List[EngineCodeFinding]: + """ + Map Kiuwan defects to KiuwanFinding objects. + + Args: + last_analysis: Dictionary containing the last analysis data. + defects_data: Dictionary containing the defects data. + analysis_code: The code of the analysis. + + Returns: + List of KiuwanFinding objects representing the mapped defects. + """ + analysis_date = last_analysis.get("date", datetime.now().strftime("%d/%m/%Y")) + application_business_value = last_analysis.get("applicationBusinessValue", "") + + findings = [] + for defect in defects_data.get("defects", []): + # Mapear characteristic a Category + characteristic = defect.get("characteristic", "").lower() + + finding = EngineCodeFinding( + id=str(defect.get("defectId", "")), + cvss=defect.get("ruleCode", ""), + where=f"{defect.get('file', '')}:{defect.get('line', '')}", + description=defect.get("rule", ""), + severity=severity_mapper.get(defect.get("priority", ""), "Undefined"), + identification_date=analysis_date, + published_date_cve="", + module="engine_code", + category=Category.VULNERABILITY if characteristic == "security" else Category.COMPLIANCE, + requirements="", + tool="kiuwan", + analysis_url=last_analysis.get("analysisURL", "No available"), + analysis_code=analysis_code, + label=defect.get("label", ""), + application_business_value=application_business_value, + defect_type=characteristic + ) + findings.append(finding) + + return findings \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_tool.py b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_tool.py new file mode 100644 index 000000000..ce6eacb8a --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/src/infrastructure/driven_adapters/kiuwan/kiuwan_tool.py @@ -0,0 +1,537 @@ +import os +from pathlib import Path +import subprocess +import platform +import time +import fnmatch +import urllib.request +import zipfile +import stat +import shutil +from typing import Dict, Any, List, Optional, Union +from urllib.parse import urlparse + +import requests +from requests.exceptions import RequestException, Timeout, ConnectionError + +from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.config_tool import ConfigTool +from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.gateways.tool_gateway import ToolGateway +from devsecops_engine_tools.engine_utilities.utils.logger_info import MyLogger +from devsecops_engine_tools.engine_utilities import settings +from devsecops_engine_tools.engine_core.src.domain.model.finding import EngineCodeFinding +from devsecops_engine_tools.engine_sast.engine_code.src.infrastructure.driven_adapters.kiuwan.kiuwan_deserealizator import KiuwanDeserealizator + +logger = MyLogger.__call__(**settings.SETTING_LOGGER).get_logger() + +class KiuwanTool(ToolGateway): + + """ + Kiuwan class to make analysis. This class install kiuwan CLI in different OS(Linux/MacOs/Windows). + This class make an analysis and promote to baseline if the specified requirements are supply. + """ + + def __init__(self, config: Dict[str, Any]): + self.user: str = config.get("user_engine_code", "") + self.password: str = config.get("token_engine_code", "") + self.base_url: str = config.get("host_engine_code", "") + self.build_execution_id = config.get("build_execution_id", "") + self.source_branch_name: str = config.get("source_branch_name", "") + self.target_branch: str = config.get("target_branch", "") + self.build_task: str = config.get("build_task", "") + self.modelo_regla: dict = config.get("MODELOS", {}).get(self.build_task, "General") + self.domain_id: str = config.get("domain_id_engine_code", "") + self.headers = {"X-KW-CORPORATE-DOMAIN-ID": self.domain_id} + self.working_directory: str = os.getcwd() + self.kiuwan_agent_path: Optional[str] = self._find_or_download_kiuwan_agent() + self.repository_name: str = "" + + + def run_tool( + self, + folder_to_scan: str, + pull_request_files: List[str], + agent_work_folder: str, + repository: str, + config_tool: ConfigTool + ) -> tuple[List[Any], Optional[str]]: + """ + Run the code scan tool. + """ + # Validar target branch + if not self._validate_target_branch(config_tool): + logger.warning("Target branch %s is not allowed for analysis", self.target_branch) + return [], None + + # Configurar el repositorio desde el parámetro + self.repository_name = repository + + # Preparar el directorio de escaneo + scan_directory = self._prepare_scan_directory( + folder_to_scan, + pull_request_files, + agent_work_folder, + config_tool + ) + + logger.info( + """== Context: == + - Repository: %s + - Source branch: %s + - Target branch: %s + - Scan directory: %s + == == == ==""", + self.repository_name, + self.source_branch_name, + self.target_branch, + scan_directory + ) + + analysis_type = self._determine_analysis_type() + logger.info("Analysis selected: %s\n", analysis_type) + logger.info("Analysis %s started", analysis_type) + + self._execute_kiuwan_scan(analysis_type, scan_directory) + self._validate_results(analysis_type) + last_analysis = self._fetch_last_analysis() + self._promote_to_baseline(last_analysis) + if not last_analysis: + last_analysis = self._fetch_last_analysis() + defects = self._fetch_defects_for_analysis(last_analysis.get("analysisCode", "")) + findings= self._map_defects_to_findings(last_analysis, defects, last_analysis.get("analysisCode", ""), config_tool.data["KIUWAN"]["SEVERITY"]) + + defects_file_path = self._download_kiuwan_csv_official(last_analysis.get("analysisCode", ""), "kiuwan_findings.csv") + + return findings, defects_file_path + + def _validate_target_branch(self, config_tool: ConfigTool) -> bool: + """Validate if the target branch is allowed for analysis.""" + target_branches = config_tool.target_branches + if target_branches and self.target_branch not in target_branches: + return False + return True + + def _prepare_scan_directory( + self, + folder_to_scan: str, + pull_request_files: List[str], + agent_work_folder: str, + config_tool: ConfigTool + ) -> str: + """Prepare the directory that will be scanned by Kiuwan.""" + + # Crear directorio temporal para el escaneo + temp_scan_dir = os.path.join(agent_work_folder, "temp_folder_to_scan") + os.makedirs(temp_scan_dir, exist_ok=True) + + if folder_to_scan is None: + # Copiar solo los archivos del PR + logger.info("Copying PR files to temporary scan directory") + for file_path in pull_request_files: + if os.path.exists(file_path): + # Mantener la estructura de directorios + relative_path = os.path.relpath(file_path, self.working_directory) + dest_path = os.path.join(temp_scan_dir, relative_path) + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + shutil.copy2(file_path, dest_path) + else: + # Copiar todo el folder_to_scan + logger.info("Copying folder_to_scan to temporary scan directory") + if os.path.exists(folder_to_scan): + shutil.copytree(folder_to_scan, temp_scan_dir, dirs_exist_ok=True) + + # Manejar exclusiones + if config_tool.exclude_folder: + self._handle_exclusions(temp_scan_dir, config_tool.exclude_folder, agent_work_folder) + + return temp_scan_dir + + def _handle_exclusions(self, scan_dir: str, exclude_folders: List[str], agent_work_folder: str): + """Move excluded files/folders to a separate directory.""" + + exclude_dir = os.path.join(agent_work_folder, "exclude_to_scan") + os.makedirs(exclude_dir, exist_ok=True) + + logger.info("Processing exclusions: %s", exclude_folders) + + for exclude_pattern in exclude_folders: + # Buscar archivos/carpetas que coincidan con el patrón + for root, dirs, files in os.walk(scan_dir): + # Verificar directorios + for dir_name in dirs[:]: # Usar slice copy para modificar durante iteración + if self._matches_exclude_pattern(dir_name, exclude_pattern): + source_path = os.path.join(root, dir_name) + relative_path = os.path.relpath(source_path, scan_dir) + dest_path = os.path.join(exclude_dir, relative_path) + + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + shutil.move(source_path, dest_path) + dirs.remove(dir_name) # No procesar este directorio más + logger.info("Excluded directory: %s", relative_path) + + # Verificar archivos + for file_name in files[:]: + if self._matches_exclude_pattern(file_name, exclude_pattern): + source_path = os.path.join(root, file_name) + relative_path = os.path.relpath(source_path, scan_dir) + dest_path = os.path.join(exclude_dir, relative_path) + + os.makedirs(os.path.dirname(dest_path), exist_ok=True) + shutil.move(source_path, dest_path) + logger.info("Excluded file: %s", relative_path) + + def _matches_exclude_pattern(self, item_name: str, pattern: str) -> bool: + """Check if an item matches the exclusion pattern.""" + return fnmatch.fnmatch(item_name, pattern) or pattern in item_name + + def _determine_analysis_type(self) -> str: + """Determine if baseline or delivery analysis is needed.""" + url = f"{self.base_url}/applications/list?applicationName={self.repository_name}" + try: + response = requests.get(url, headers=self.headers, auth=(self.user, self.password), timeout=60) + response.raise_for_status() + apps = response.json() + app_names = [app["name"] for app in apps] + return "baseline" if self.repository_name not in app_names else "delivery" + except (requests.RequestException, ValueError) as e: + raise RuntimeError(f"Failed to determine analysis type in engine code: {e}") from e + + def _execute_kiuwan_scan(self, analysis_type: str, scan_directory: str) -> Union[subprocess.CompletedProcess, Dict[str, Any]]: + """Execute Kiuwan baseline or delivery analysis.""" + + cmd = [ + self.kiuwan_agent_path, + "--user", self.user, + "--pass", self.password, + "--domain-id", self.domain_id, + "--softwareName", self.repository_name, + "--sourcePath", scan_directory, # Usar el directorio preparado + "--label", self.build_execution_id, + "--wait-for-results", + "--model-name", self.modelo_regla + ] + + if analysis_type == "baseline": + cmd.extend(["--create", "--analysis-scope", "baseline"]) + else: + cmd.extend(["--analysis-scope", "completeDelivery", "--change-request", "inprogress", + "--branch-name", self.source_branch_name]) + + logger.info("Kiuwan analysis will be executed using model %s", self.modelo_regla) + try: + result = subprocess.run(cmd, capture_output=True, text=True, check=True, errors="ignore") + logger.info("Scan results: %s", result) + return result + except subprocess.CalledProcessError as e: + error = {"status": "failed", "output": e.stderr} + logger.error("Scan results: %s", error) + return error + + def _validate_results(self, analysis_type: str): + + """Validate analysis results.""" + try: + if analysis_type == "baseline": + logger.info("Validate Baseline Results with applicationBusinessValue") + url = f"{self.base_url}/applications/list?applicationName={self.repository_name}" + response = requests.get(url, headers=self.headers, auth=(self.user, self.password), timeout=60) + response.raise_for_status() + application_business_value = response.json()[0].get("applicationBusinessValue") + if application_business_value in ["CRITICAL", "HIGH"]: + logger.warning("Baseline analysis failed: Business Value is %s", application_business_value) + else: + logger.info("Validate Delivery Results with AuditResult") + url = f"{self.base_url}/applications/deliveries?application={self.repository_name}" + response = requests.get(url, headers=self.headers, auth=(self.user, self.password), timeout=60) + response.raise_for_status() + results = response.json() + + for result in results: + audit_result = result.get("auditResult", "") + if audit_result != "OK": + logger.warning("Delivery analysis failed: Audit Result is %s", audit_result) + + except (subprocess.CalledProcessError, ValueError) as e: + logger.error("Analysis result failed:\nRepository: %s", self.repository_name) + raise RuntimeError(f"Validation analysis results failed: {e}") from e + + def _fetch_last_analysis(self) -> Dict[str, Any]: + """ + Fetch the last analysis for the repository and wait until it is finished. + + Returns: + Dictionary containing the last analysis data. + + Raises: + RuntimeError: If there's an error fetching the analysis or if the response is invalid. + """ + last_analysis_url = f"{self.base_url}/applications/last_analysis?application={self.repository_name}" + logger.info("Getting last analysis...") + max_tries = 10 + tried = 0 + try: + while tried <= max_tries: + response = requests.get( + last_analysis_url, + headers=self.headers, + auth=(self.user, self.password), + timeout=60, + ) + response.raise_for_status() + last_analysis = response.json() + + if last_analysis.get("analysisStatus") == "FINISHED": + return last_analysis + + logger.info("Analysis status %s, waiting status FINISHED", last_analysis.get('analysisStatus')) + time.sleep(5) + tried += 1 + return None + except (requests.RequestException, ValueError) as e: + raise RuntimeError(f"Failed to fetch last analysis: {e}") from e + + def _promote_to_baseline(self, last_analysis): + + """Promote delivery to baseline if on master branch.""" + + if self.target_branch == "master" or not last_analysis: + logger.info("Promoting delivery to baseline...") + cmd = [ + self.kiuwan_agent_path, + "--promote-to-baseline", + "--user", self.user, + "--pass", self.password, + "--domain-id", self.domain_id, + "--softwareName", self.repository_name, + "--change-request", "inprogress", + "--label", self.build_execution_id, + ] + try: + subprocess.run(cmd, capture_output=True, text=True, check=True, errors="ignore") + logger.info("Promotion completed") + except subprocess.CalledProcessError as e: + logger.error("Promotion failed: %s", e) + + else: + logger.info("No promotion needed") + + + def _fetch_defects_for_analysis(self, analysis_code: str) -> Dict[str, Any]: + """ + Fetch all defects for a given analysis code, handling pagination. + + Args: + analysis_code: The code of the analysis to fetch defects for. + + Returns: + Dictionary containing all defects and metadata. + + Raises: + RuntimeError: If there's an error fetching defects or if the response is invalid. + """ + base_defects_url = f"{self.base_url}/apps/analysis/{analysis_code}/defects" + first_defects_page = "?page=1&count=5000" + last_analysis_defects_url = base_defects_url + first_defects_page + + logger.info("Getting defects...") + try: + response = requests.get( + last_analysis_defects_url, + headers=self.headers, + auth=(self.user, self.password), + timeout=60, + ) + response.raise_for_status() + last_analysis_defects: Dict[str, Any] = response.json() + + all_defects = last_analysis_defects.get("defects", []) + total_defects = last_analysis_defects.get("defects_count", 0) + + if total_defects > 5000: + total_pages = (total_defects + 4999) // 5000 + for page in range(2, total_pages + 1): + paginated_url = f"{base_defects_url}?page={page}&count=5000" + logger.info("Fetching page %s of defects...", page) + response = requests.get( + paginated_url, + headers=self.headers, + auth=(self.user, self.password), + timeout=60, + ) + if response.status_code == 200: + page_data = response.json() + all_defects.extend(page_data.get("defects", [])) + else: + logger.warning("Failed to fetch page %s: %s", page, response.status_code) + + last_analysis_defects["defects"] = all_defects + return last_analysis_defects + + except (requests.RequestException, ValueError) as e: + raise RuntimeError(f"Failed to fetch defects: {e}") from e + + def _download_kiuwan_csv_official(self, analysis_code: str, output_path: str = "kiuwan_findings.csv") -> str: + """ + Download the official Kiuwan SAST CSV report using the vulnerabilities/export API. + Compatible with DefectDojo's 'Kiuwan Scan' parser. + + Args: + analysis_code (str): The analysis code to export. + output_path (str): Path to save the CSV file. + + Returns: + str: Path to the downloaded CSV file. + """ + csv_url = ( + f"{self.base_url}/applications/analysis/vulnerabilities/export?" + f"application={self.repository_name}&code={analysis_code}&type=CSV" + ) + + try: + logger.info("Downloading official Kiuwan CSV from: %s", csv_url) + response = requests.get( + csv_url, + auth=(self.user, self.password), + headers=self.headers, + timeout=60, + stream=True + ) + response.raise_for_status() + + if 'text/csv' not in response.headers.get('Content-Type', ''): + logger.warning("Response Content-Type is not CSV: %s", response.headers.get('Content-Type')) + + with open(output_path, "wb") as f: + for chunk in response.iter_content(chunk_size=8192): + f.write(chunk) + + logger.info("Official Kiuwan CSV downloaded successfully: %s", output_path) + return output_path + + except RequestException as e: + logger.error("Failed to download Kiuwan CSV: %s", e) + raise RuntimeError(f"Error downloading Kiuwan CSV from {csv_url}: {e}") from e + + def _map_defects_to_findings( + self, + last_analysis: Dict[str, Any], + defects_data: Dict[str, Any], + analysis_code: str, + severity_mapper: Dict[str,str] + ) -> List[EngineCodeFinding]: + return KiuwanDeserealizator.get_findings(last_analysis, defects_data, analysis_code, severity_mapper) + + def _find_or_download_kiuwan_agent(self) -> str: + """Ensure Kiuwan agent is available and return its path.""" + + system = platform.system() + agent_script = { + "Windows": "agent.cmd", + "Linux": "agent.sh", + "Darwin": "agent.sh" # macOS + }.get(system) + + if not agent_script: + raise RuntimeError(f"Unsupported OS: {system}") + + agent_path = self._search_agent_script(agent_script) + if agent_path: + return agent_path + + self._download_and_extract_kiuwan() + agent_path = self._search_agent_script(agent_script) + + if not agent_path: + raise FileNotFoundError(f"{agent_script} was not found.") + + logger.info("Kiuwan agent path is: %s", agent_path) + return agent_path + + def _search_agent_script(self, script_name: str) -> str: + """Search for the Kiuwan script in the tools directory.""" + for root, _, files in os.walk(self.working_directory): + if script_name in files: + return os.path.join(root, script_name) + return "" + + def _set_execution_permissions(self, extracted_files): + + """Set execution permissions to extracted files""" + + # Set execution permissions on .sh files if on Unix-like system + if platform.system().lower() in ["linux", "darwin"]: + for file_path in extracted_files: + if file_path.endswith(".sh"): + try: + os.chmod(file_path, os.stat(file_path).st_mode | stat.S_IEXEC) + logger.info("Execution permissions granted to: %s", file_path) + except Exception as e: + logger.warning("Failed to set execution permissions for %s : %s", file_path, e) + + def _download_and_extract_kiuwan(self): + """Download and extract the Kiuwan CLI, flattening the internal folder structure and setting execution permissions.""" + kiuwan_url = "https://www.kiuwan.com/pub/analyzer/KiuwanLocalAnalyzer.zip" + zip_path = os.path.join(self.working_directory, "KiuwanLocalAnalyzer.zip") + + try: + + parsed_url = urlparse(kiuwan_url) + if parsed_url.scheme != "https": + raise ValueError("Only HTTPS URLs are allowed for downloading Kiuwan CLI.") + + logger.info("Downloading Kiuwan CLI...") + urllib.request.urlretrieve(kiuwan_url, zip_path) # nosec + + extracted_files = self._extract_zip(zip_path) + + os.remove(zip_path) + logger.info("Kiuwan CLI extracted successfully into the root tools directory.") + + self._set_execution_permissions(extracted_files) + + except Exception as e: + raise RuntimeError(f"Error downloading or extracting the Kiuwan agent: {e}") from e + + def _extract_zip(self, zip_path) -> list: + + """Extract zip files""" + + extracted_files = [] + + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + for member in zip_ref.namelist(): + if member.startswith("KiuwanLocalAnalyzer/") and not member.endswith("/"): + relative_path = os.path.relpath(member, "KiuwanLocalAnalyzer") + target_path = os.path.join(self.working_directory, relative_path) + + Path(os.path.dirname(target_path)).mkdir(parents=True, exist_ok=True) + with zip_ref.open(member) as source, open(target_path, "wb") as target: + shutil.copyfileobj(source, target) + extracted_files.append(target_path) + + return extracted_files + + +def get_kiuwan_instance(dict_args: Dict, devops_platform_gateway) -> KiuwanTool: + """ + Create a kiuwan instance to scan + """ + logger.info("Retrieving kiuwan configuration file...") + kiuwan_config_tool = devops_platform_gateway.get_remote_config( + dict_args["remote_config_repo"], + "/engine_sast/engine_code/ConfigTool.json", + dict_args["remote_config_branch"] + ) + logger.info("Kiuwan configuration file retrieved") + logger.info("Settings config dictionary to scan tool...") + config = { + "host_engine_code": kiuwan_config_tool["KIUWAN"]["SERVER"]["BASE_URL"], + "user_engine_code": kiuwan_config_tool["KIUWAN"]["SERVER"]["USER"], + "domain_id_engine_code": kiuwan_config_tool["KIUWAN"]["SERVER"]["DOMAIN_ID"], + "token_engine_code": dict_args["token_engine_code"], + "build_execution_id": devops_platform_gateway.get_variable("build_execution_id"), + "source_branch_name": devops_platform_gateway.get_variable("branch_name"), + "target_branch": devops_platform_gateway.get_variable("target_branch"), + "build_task": devops_platform_gateway.get_variable("build_task"), + "MODELOS": kiuwan_config_tool["KIUWAN"]["MODELOS"] + } + return KiuwanTool(config) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/test/domain/usecases/test_code_scan.py b/tools/devsecops_engine_tools/engine_sast/engine_code/test/domain/usecases/test_code_scan.py index 0185014c9..12c5d87e7 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_code/test/domain/usecases/test_code_scan.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/test/domain/usecases/test_code_scan.py @@ -41,7 +41,7 @@ def test_set_config_tool(self, mock_config_tool): self.mock_devops_platform_gateway.get_variable.return_value = "pipeline_test_name" # Act - self.code_scan.set_config_tool({"remote_config_repo": "test_repo", "remote_config_branch": ""}) + self.code_scan.set_config_tool({"remote_config_repo": "test_repo", "remote_config_branch": "", "tool": "BEARER"}, "engine_sast/engine_code/ConfigTool.json") # Assert self.mock_remote_config_source_gateway.get_remote_config.assert_called_once_with( @@ -179,7 +179,7 @@ def test_process(self, mock_input_core): self.mock_devops_platform_gateway.get_variable.side_effect = ["test_work_folder", "test_repo", "test_stage"] # Act - findings_list, _ = self.code_scan.process({"folder_path": None, "remote_config_repo": "some_repo", "remote_config_branch": ""}, "TOOL_NAME") + findings_list, _ = self.code_scan.process({"folder_path": None, "remote_config_repo": "some_repo", "remote_config_branch": ""}, "BEARER") # Assert self.code_scan.set_config_tool.assert_called_once() @@ -205,7 +205,7 @@ def test_process_skip_tool(self, mock_input_core): self.mock_devops_platform_gateway.get_variable.return_value = "test_stage" # Act - findings_list, _ = self.code_scan.process({"folder_path": None, "remote_config_repo": "some_repo", "remote_config_branch": ""}, "TOOL_NAME") + findings_list, _ = self.code_scan.process({"folder_path": None, "remote_config_repo": "some_repo", "remote_config_branch": ""}, "BEARER") # Assert self.code_scan.set_config_tool.assert_called_once() @@ -216,4 +216,4 @@ def test_process_skip_tool(self, mock_input_core): totalized_exclusions=["exclusion1"], threshold_defined=self.code_scan.set_config_tool.return_value.threshold, path_file_results="", custom_message_break_build=self.code_scan.set_config_tool.return_value.message_info_engine_code, scope_pipeline="test_scope", scope_service="test_scope", stage_pipeline="Test_stage" - ) + ) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_code/test/infrastructure/driven_adapters/test_kiuwan_tool.py b/tools/devsecops_engine_tools/engine_sast/engine_code/test/infrastructure/driven_adapters/test_kiuwan_tool.py new file mode 100644 index 000000000..76b2992ab --- /dev/null +++ b/tools/devsecops_engine_tools/engine_sast/engine_code/test/infrastructure/driven_adapters/test_kiuwan_tool.py @@ -0,0 +1,857 @@ +import unittest +from unittest.mock import Mock, mock_open, patch +import os +import tempfile +import shutil +import subprocess +import zipfile +from pathlib import Path + +import requests + +from devsecops_engine_tools.engine_sast.engine_code.src.domain.model.config_tool import ConfigTool +from devsecops_engine_tools.engine_sast.engine_code.src.infrastructure.driven_adapters.kiuwan import kiuwan_tool + +class TestKiuwanToolBase(unittest.TestCase): + + def setUp(self): + self.config = { + "user_engine_code": "test_user", + "token_engine_code": "test_token", + "host_engine_code": "https://test.kiuwan.com", + "app_name": "test_app", + "build_execution_id": "test_build_123", + "source_branch_name": "feature/test", + "target_branch": "master", + "build_task": "test_task", + "MODELOS": {"test_task": "TestModel"}, + "domain_id_engine_code": "test_domain_id" + } + + self.mock_config_tool = Mock(spec=ConfigTool) + self.mock_config_tool.target_branches = ["master", "develop"] + self.mock_config_tool.exclude_folder = ["node_modules", "*.test.js"] + self.mock_config_tool.data = {"SEVERITY": {"HIGH": "Critical", "MEDIUM": "High"}} + + self.temp_directory = tempfile.mkdtemp() + + def tearDown(self): + if os.path.exists(self.temp_directory): + shutil.rmtree(self.temp_directory) + + def create_kiuwan_tool(self, config=None, path_mock="/path/to/agent.sh"): + with patch.object(kiuwan_tool.KiuwanTool, '_find_or_download_kiuwan_agent', return_value=path_mock): + return kiuwan_tool.KiuwanTool(config or self.config) + + +class TestKiuwanToolInit(TestKiuwanToolBase): + + def test_init_with_complete_config(self): + with patch.object(kiuwan_tool.KiuwanTool, '_find_or_download_kiuwan_agent', return_value='/path/to/agent.sh'): + tool = kiuwan_tool.KiuwanTool(self.config) + + self.assertEqual(tool.user, "test_user") + self.assertEqual(tool.password, "test_token") + self.assertEqual(tool.base_url, "https://test.kiuwan.com") + self.assertEqual(tool.build_execution_id, "test_build_123") + self.assertEqual(tool.source_branch_name, "feature/test") + self.assertEqual(tool.target_branch, "master") + self.assertEqual(tool.build_task, "test_task") + self.assertEqual(tool.modelo_regla, "TestModel") + self.assertEqual(tool.domain_id, "test_domain_id") + self.assertEqual(tool.repository_name, "") + self.assertEqual(tool.kiuwan_agent_path, '/path/to/agent.sh') + + def test_init_with_partial_config(self): + partial_config = { + "user_engine_code": "user", + "host_engine_code": "https://test.com" + } + with patch.object(kiuwan_tool.KiuwanTool, '_find_or_download_kiuwan_agent', return_value='/path/to/agent.sh'): + tool = kiuwan_tool.KiuwanTool(partial_config) + + self.assertEqual(tool.user, "user") + self.assertEqual(tool.password, "") + self.assertEqual(tool.modelo_regla, "General") + self.assertEqual(tool.kiuwan_agent_path, '/path/to/agent.sh') + + +class TestFindOrDownloadKiuwanAgent(TestKiuwanToolBase): + + @patch('platform.system') + def test_find_agent_windows(self, mock_system): + mock_system.return_value = "Windows" + tool = self.create_kiuwan_tool() + tool._search_agent_script = Mock(return_value="/path/to/agent.cmd") + + result = tool._find_or_download_kiuwan_agent() + + self.assertEqual(result, "/path/to/agent.cmd") + tool._search_agent_script.assert_called_with("agent.cmd") + + @patch('platform.system') + def test_find_agent_linux(self, mock_system): + mock_system.return_value = "Linux" + tool = self.create_kiuwan_tool() + tool._search_agent_script = Mock(return_value="/path/to/agent.sh") + + result = tool._find_or_download_kiuwan_agent() + + self.assertEqual(result, "/path/to/agent.sh") + tool._search_agent_script.assert_called_with("agent.sh") + + @patch('platform.system') + def test_find_agent_macos(self, mock_system): + mock_system.return_value = "Darwin" + tool = self.create_kiuwan_tool() + tool._search_agent_script = Mock(return_value="/path/to/agent.sh") + + result = tool._find_or_download_kiuwan_agent() + + self.assertEqual(result, "/path/to/agent.sh") + tool._search_agent_script.assert_called_with("agent.sh") + + @patch('platform.system') + def test_unsupported_os(self, mock_system): + mock_system.return_value = "UnsupportedOS" + + with self.assertRaises(RuntimeError) as context: + kiuwan_tool.KiuwanTool(self.config) + + self.assertIn("Unsupported OS", str(context.exception)) + + @patch('platform.system') + def test_download_agent_when_not_found(self, mock_system): + mock_system.return_value = "Linux" + tool = self.create_kiuwan_tool() + tool._search_agent_script = Mock(side_effect=["", "/path/to/agent.sh"]) + tool._download_and_extract_kiuwan = Mock() + + result = tool._find_or_download_kiuwan_agent() + + self.assertEqual(result, "/path/to/agent.sh") + tool._download_and_extract_kiuwan.assert_called_once() + + @patch('platform.system') + def test_agent_not_found_after_download(self, mock_system): + mock_system.return_value = "Linux" + tool = self.create_kiuwan_tool() + tool._search_agent_script = Mock(return_value="") + tool._download_and_extract_kiuwan = Mock() + + with self.assertRaises(FileNotFoundError) as context: + tool._find_or_download_kiuwan_agent() + + self.assertIn("agent.sh was not found", str(context.exception)) + + +class TestSearchAgentScript(TestKiuwanToolBase): + + def test_search_agent_script_found(self): + tool = self.create_kiuwan_tool() + # Crear estructura de directorios + Path(os.path.join(self.temp_directory, "tools")).mkdir(parents=True, exist_ok=True) + agent_path = os.path.join(self.temp_directory, "tools", "agent.sh") + with open(agent_path, 'w') as f: + f.write("#!/bin/bash") + + tool.working_directory = self.temp_directory + result = tool._search_agent_script("agent.sh") + + self.assertEqual(result, agent_path) + + def test_search_agent_script_not_found(self): + tool = self.create_kiuwan_tool() + tool.working_directory = self.temp_directory + result = tool._search_agent_script("agent.sh") + + self.assertEqual(result, "") + + +class TestSetExecutionPermissions(TestKiuwanToolBase): + + def setUp(self): + + self.config = { + "user_engine_code": "test_user", + "token_engine_code": "test_token", + "host_engine_code": "https://test.kiuwan.com", + "app_name": "test_app", + "build_execution_id": "test_build_123", + "source_branch_name": "feature/test", + "target_branch": "master", + "build_task": "test_task", + "MODELOS": {"test_task": "TestModel"}, + "domain_id_engine_code": "test_domain_id" + } + + self.mock_config_tool = Mock(spec=ConfigTool) + self.mock_config_tool.target_branches = ["master", "develop"] + self.mock_config_tool.exclude_folder = ["node_modules", "*.test.js"] + self.mock_config_tool.data = {"SEVERITY": {"HIGH": "Critical", "MEDIUM": "High"}} + + self.temp_directory = tempfile.mkdtemp() + Path(os.path.join(self.temp_directory, "tools")).mkdir(parents=True, exist_ok=True) + self.agent_path = os.path.join(self.temp_directory, "tools", "agent.sh") + with open(self.agent_path, 'w') as f: + f.write("#!/bin/bash") + + + def tearDown(self): + if os.path.exists(self.temp_directory): + shutil.rmtree(self.temp_directory) + + @patch('platform.system') + def test_set_execution_permissions_linux(self, mock_system): + mock_system.return_value = "Linux" + + tool = self.create_kiuwan_tool() + + files = [self.agent_path, f'{os.path.join(self.temp_directory, "tools", "file.txt")}'] + tool._set_execution_permissions(files) + permission_sh = os.access(files[0], os.X_OK) + permission_txt = os.access(files[1], os.X_OK) + self.assertTrue(permission_sh) + self.assertFalse(permission_txt) + self.assertNotEqual(permission_sh, permission_txt) + + @patch('platform.system') + def test_set_execution_permissions_darwin(self, mock_system): + mock_system.return_value = "darwin" + + tool = self.create_kiuwan_tool() + + files = [self.agent_path, f'{os.path.join(self.temp_directory, "tools", "file.txt")}'] + tool._set_execution_permissions(files) + permission_sh = os.access(files[0], os.X_OK) + permission_txt = os.access(files[1], os.X_OK) + self.assertTrue(permission_sh) + self.assertFalse(permission_txt) + self.assertNotEqual(permission_sh, permission_txt) + + +class TestExtractZip(TestKiuwanToolBase): + + def test_extract_zip_success(self): + self.temp_directory = tempfile.mkdtemp() + # Rutas locales + bin_dir = os.path.join(self.temp_directory, "KiuwanLocalAnalyzer", "bin") + lib_dir = os.path.join(self.temp_directory, "KiuwanLocalAnalyzer", "lib") + Path(bin_dir).mkdir(parents=True, exist_ok=True) + Path(lib_dir).mkdir(parents=True, exist_ok=True) + + agent_sh_path = os.path.join(bin_dir, "agent.sh") + kiuwan_jar_path = os.path.join(lib_dir, "kiuwan.jar") + + Path(agent_sh_path).write_text("#!/bin/bash") + Path(kiuwan_jar_path).write_text("fake jar content", encoding="utf-8") + + # Crear ZIP + zip_path = os.path.join(self.temp_directory, "test_kiuwan.zip") + with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: + # ✅ Directorios (opcionales, pero ayudan) + zipf.writestr("KiuwanLocalAnalyzer/", "") + zipf.writestr("KiuwanLocalAnalyzer/bin/", "") + zipf.writestr("KiuwanLocalAnalyzer/lib/", "") + + # Archivos + info = zipfile.ZipInfo("KiuwanLocalAnalyzer/bin/agent.sh") + info.external_attr = 0o755 << 16 + with open(agent_sh_path, 'rb') as f: + zipf.writestr(info, f.read()) + + info = zipfile.ZipInfo("KiuwanLocalAnalyzer/lib/kiuwan.jar") + info.external_attr = 0o644 << 16 + with open(kiuwan_jar_path, 'rb') as f: + zipf.writestr(info, f.read()) + + # Imprimir contenido del ZIP para depurar + print("\n=== ZIP CONTENT ===") + with zipfile.ZipFile(zip_path, 'r') as zf: + for f in zf.namelist(): + print(f" {f}") + print("==================\n") + + # Preparar herramienta + tool = self.create_kiuwan_tool() + extracted_dir = os.path.join(self.temp_directory, "extracted") + tool.working_directory = extracted_dir + + # Llamar al método + result = tool._extract_zip(zip_path) + + # Validar + self.assertEqual(len(result), 2) + self.assertTrue(os.path.exists(os.path.join(extracted_dir, "bin", "agent.sh"))) + self.assertTrue(os.path.exists(os.path.join(extracted_dir, "lib", "kiuwan.jar"))) + + + @patch('zipfile.ZipFile') + def test_extract_zip_error(self, mock_zipfile): + mock_zipfile.side_effect = Exception("Zip error") + + tool = self.create_kiuwan_tool() + with self.assertRaises(Exception) as context: + tool._extract_zip("/path/to/zip") + + self.assertIn("Zip error", str(context.exception)) + + +class TestDownloadAndExtractKiuwan(TestKiuwanToolBase): + + @patch('urllib.request.urlretrieve') + @patch('os.remove') + def test_download_and_extract_success(self, mock_remove, mock_urlretrieve): + tool = self.create_kiuwan_tool() + tool._extract_zip = Mock(return_value=["/path/to/agent.sh"]) + tool._set_execution_permissions = Mock() + + tool._download_and_extract_kiuwan() + + mock_urlretrieve.assert_called_once() + tool._extract_zip.assert_called_once() + tool._set_execution_permissions.assert_called_once_with(["/path/to/agent.sh"]) + mock_remove.assert_called_once() + + @patch('urllib.request.urlretrieve') + def test_download_error(self, mock_urlretrieve): + mock_urlretrieve.side_effect = Exception("Download failed") + + tool = self.create_kiuwan_tool() + with self.assertRaises(RuntimeError) as context: + tool._download_and_extract_kiuwan() + + self.assertIn("Error downloading or extracting", str(context.exception)) + + +class TestValidateTargetBranch(TestKiuwanToolBase): + + def test_validate_target_branch_allowed(self): + tool = self.create_kiuwan_tool() + tool.target_branch = "master" + result = tool._validate_target_branch(self.mock_config_tool) + self.assertTrue(result) + + def test_validate_target_branch_not_allowed(self): + tool = self.create_kiuwan_tool() + tool.target_branch = "forbidden_branch" + + result = tool._validate_target_branch(self.mock_config_tool) + + self.assertFalse(result) + + def test_validate_target_branch_empty_list(self): + tool = self.create_kiuwan_tool() + config_tool = Mock(spec=ConfigTool) + config_tool.target_branches = [] + + result = tool._validate_target_branch(config_tool) + + self.assertTrue(result) + + def test_validate_target_branch_none(self): + tool = self.create_kiuwan_tool() + config_tool = Mock(spec=ConfigTool) + config_tool.target_branches = None + + result = tool._validate_target_branch(config_tool) + + self.assertTrue(result) + + +class TestPrepareScanDirectory(TestKiuwanToolBase): + + @patch('os.makedirs') + @patch('os.path.exists') + @patch('shutil.copy2') + @patch('os.path.relpath') + def test_prepare_scan_directory_pr_files(self, mock_relpath, mock_copy2, mock_exists, mock_makedirs): + mock_exists.return_value = True + mock_relpath.return_value = "src/file.py" + self.mock_config_tool.exclude_folder = [] + + tool = self.create_kiuwan_tool() + pr_files = ["/repo/src/file.py", "/repo/src/test.py"] + + result = tool._prepare_scan_directory( + None, pr_files, self.temp_directory, self.mock_config_tool + ) + + expected_path = os.path.join(self.temp_directory, "temp_folder_to_scan") + self.assertEqual(result, expected_path) + mock_makedirs.assert_called() + self.assertEqual(mock_copy2.call_count, 2) + + @patch('os.makedirs') + @patch('os.path.exists') + @patch('shutil.copytree') + def test_prepare_scan_directory_folder_to_scan(self, mock_copytree, mock_exists, mock_makedirs): + mock_exists.return_value = True + self.mock_config_tool.exclude_folder = [] + + tool = self.create_kiuwan_tool() + folder_to_scan = "/source/folder" + + result = tool._prepare_scan_directory( + folder_to_scan, [], self.temp_directory, self.mock_config_tool + ) + + expected_path = os.path.join(self.temp_directory, "temp_folder_to_scan") + self.assertEqual(result, expected_path) + mock_copytree.assert_called_with(folder_to_scan, expected_path, dirs_exist_ok=True) + + @patch('os.makedirs') + def test_prepare_scan_directory_with_exclusions(self, mock_makedirs): + self.mock_config_tool.exclude_folder = ["node_modules"] + + tool = self.create_kiuwan_tool() + tool._handle_exclusions = Mock() + + result = tool._prepare_scan_directory( + None, [], self.temp_directory, self.mock_config_tool + ) + + tool._handle_exclusions.assert_called_once() + + +class TestHandleExclusions(TestKiuwanToolBase): + + @patch('os.walk') + @patch('os.makedirs') + @patch('shutil.move') + @patch('os.path.relpath') + def test_handle_exclusions_directory(self, mock_relpath, mock_move, mock_makedirs, mock_walk): + mock_walk.return_value = [ + ("/scan", ["node_modules", "src"], ["file.py"]), + ("/scan/src", [], ["test.py"]) + ] + mock_relpath.return_value = "node_modules" + + tool = self.create_kiuwan_tool() + tool._matches_exclude_pattern = Mock(side_effect=lambda name, pattern: name == "node_modules") + + tool._handle_exclusions("/scan", ["node_modules"], "/agent") + + mock_move.assert_called() + mock_makedirs.assert_called() + + @patch('os.walk') + @patch('os.makedirs') + @patch('shutil.move') + @patch('os.path.relpath') + def test_handle_exclusions_file(self, mock_relpath, mock_move, mock_makedirs, mock_walk): + mock_walk.return_value = [ + ("/scan", [], ["test.spec.js", "app.js"]) + ] + mock_relpath.return_value = "test.spec.js" + + tool = self.create_kiuwan_tool() + tool._matches_exclude_pattern = Mock(side_effect=lambda name, pattern: "test" in name) + + tool._handle_exclusions("/scan", ["*test*"], "/agent") + + mock_move.assert_called() + + +class TestMatchesExcludePattern(TestKiuwanToolBase): + + def test_matches_exclude_pattern_fnmatch(self): + tool = self.create_kiuwan_tool() + result = tool._matches_exclude_pattern("test.spec.js", "*.spec.js") + self.assertTrue(result) + + def test_matches_exclude_pattern_substring(self): + tool = self.create_kiuwan_tool() + result = tool._matches_exclude_pattern("node_modules", "node_modules") + self.assertTrue(result) + + def test_matches_exclude_pattern_no_match(self): + tool = self.create_kiuwan_tool() + result = tool._matches_exclude_pattern("app.js", "*.spec.js") + self.assertFalse(result) + + +class TestDetermineAnalysisType(TestKiuwanToolBase): + + @patch('requests.get') + def test_determine_analysis_type_baseline(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"name": "other_app"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + tool.repository_name = "new_app" + result = tool._determine_analysis_type() + + self.assertEqual(result, "baseline") + + @patch('requests.get') + def test_determine_analysis_type_delivery(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"name": "existing_app"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + tool.repository_name = "existing_app" + result = tool._determine_analysis_type() + + self.assertEqual(result, "delivery") + + +class TestExecuteKiuwanScan(TestKiuwanToolBase): + + @patch('subprocess.run') + def test_execute_kiuwan_scan_baseline_success(self, mock_run): + mock_result = Mock() + mock_result.stdout = "Analysis completed" + mock_run.return_value = mock_result + + tool = self.create_kiuwan_tool() + result = tool._execute_kiuwan_scan("baseline", "/scan/dir") + + self.assertEqual(result, mock_result) + # Verificar que se llamó con los argumentos correctos + call_args = mock_run.call_args[0][0] + self.assertIn("--create", call_args) + self.assertIn("--analysis-scope", call_args) + self.assertIn("baseline", call_args) + + @patch('subprocess.run') + def test_execute_kiuwan_scan_delivery_success(self, mock_run): + mock_result = Mock() + mock_run.return_value = mock_result + + tool = self.create_kiuwan_tool() + result = tool._execute_kiuwan_scan("delivery", "/scan/dir") + + self.assertEqual(result, mock_result) + call_args = mock_run.call_args[0][0] + self.assertIn("completeDelivery", call_args) + self.assertIn("--change-request", call_args) + + @patch('subprocess.run') + def test_execute_kiuwan_scan_error(self, mock_run): + mock_run.side_effect = subprocess.CalledProcessError(1, "cmd", stderr="Error") + + tool = self.create_kiuwan_tool() + result = tool._execute_kiuwan_scan("baseline", "/scan/dir") + + self.assertEqual(result["status"], "failed") + self.assertIn("Error", result["output"]) + + +class TestValidateResults(TestKiuwanToolBase): + + @patch('requests.get') + def test_validate_results_baseline_success(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"applicationBusinessValue": "LOW"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + #No debería lanzar excepción + try: + tool._validate_results("baseline") + except Exception: + self.fail("_validate_results raised an unexpected exception") + + @patch('requests.get') + def test_validate_results_baseline_critical(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"applicationBusinessValue": "CRITICAL"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + tool._validate_results("baseline") + mock_get.assert_called_once() + + @patch('requests.get') + def test_validate_results_delivery_success(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"auditResult": "OK"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + try: + tool._validate_results("delivery") + except Exception: + self.fail("_validate_results raised an unexpected exception") + + @patch('requests.get') + def test_validate_results_delivery_failed(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = [{"auditResult": "FAIL"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + tool._validate_results("delivery") + mock_get.assert_called_once() + +class TestFetchLastAnalysis(TestKiuwanToolBase): + + @patch('requests.get') + @patch('time.sleep') + def test_fetch_last_analysis_finished(self, mock_sleep, mock_get): + mock_response = Mock() + mock_response.json.return_value = {"analysisStatus": "FINISHED"} + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + result = tool._fetch_last_analysis() + + self.assertEqual(result["analysisStatus"], "FINISHED") + mock_sleep.assert_not_called() + + @patch('requests.get') + @patch('time.sleep') + def test_fetch_last_analysis_waiting(self, mock_sleep, mock_get): + responses = [ + {"analysisStatus": "RUNNING"}, + {"analysisStatus": "RUNNING"}, + {"analysisStatus": "FINISHED"} + ] + mock_response = Mock() + mock_response.json.side_effect = responses + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + result = tool._fetch_last_analysis() + + self.assertEqual(result["analysisStatus"], "FINISHED") + self.assertEqual(mock_sleep.call_count, 2) + + @patch('requests.get') + def test_fetch_last_analysis_error(self, mock_get): + mock_get.side_effect = RuntimeError("Failed to fetch last analysis") + + tool = self.create_kiuwan_tool() + with self.assertRaises(RuntimeError) as context: + tool._fetch_last_analysis() + + self.assertIn("Failed to fetch last analysis", str(context.exception)) + + +class TestFetchDefectsForAnalysis(TestKiuwanToolBase): + + @patch('requests.get') + def test_fetch_defects_single_page(self, mock_get): + mock_response = Mock() + mock_response.json.return_value = { + "defects": [{"id": 1}, {"id": 2}], + "defects_count": 2 + } + mock_response.raise_for_status = Mock() + mock_response.status_code = 200 + mock_get.return_value = mock_response + + tool = self.create_kiuwan_tool() + result = tool._fetch_defects_for_analysis("analysis123") + + self.assertEqual(len(result["defects"]), 2) + self.assertEqual(mock_get.call_count, 1) + + @patch('requests.get') + def test_fetch_defects_multiple_pages(self, mock_get): + # Primera página + first_response = Mock() + first_response.json.return_value = { + "defects": [{"id": i} for i in range(5000)], + "defects_count": 7000 + } + first_response.raise_for_status = Mock() + first_response.status_code = 200 + + # Segunda página + second_response = Mock() + second_response.json.return_value = { + "defects": [{"id": i} for i in range(5000, 7000)] + } + second_response.status_code = 200 + + mock_get.side_effect = [first_response, second_response] + + tool = self.create_kiuwan_tool() + result = tool._fetch_defects_for_analysis("analysis123") + + self.assertEqual(len(result["defects"]), 7000) + self.assertEqual(mock_get.call_count, 2) + + @patch('requests.get') + def test_fetch_defects_error(self, mock_get): + mock_get.side_effect = RuntimeError("Failed to fetch defects") + + tool = self.create_kiuwan_tool() + with self.assertRaises(RuntimeError) as context: + tool._fetch_defects_for_analysis("analysis123Test") + + self.assertIn("Failed to fetch defects", str(context.exception)) + + +class TestMapDefectsToFindings(TestKiuwanToolBase): + + @patch('devsecops_engine_tools.engine_sast.engine_code.src.infrastructure.driven_adapters.kiuwan.kiuwan_deserealizator.KiuwanDeserealizator.get_findings') + def test_map_defects_to_findings(self, mock_get_findings): + last_analysis = {"analysisCode": "123"} + defects_data = {"defects": []} + severity_mapper = {"HIGH": "Critical"} + expected_findings = [Mock()] + + mock_get_findings.return_value = expected_findings + + tool = self.create_kiuwan_tool() + result = tool._map_defects_to_findings( + last_analysis, defects_data, "123", severity_mapper + ) + + self.assertEqual(result, expected_findings) + mock_get_findings.assert_called_once_with( + last_analysis, defects_data, "123", severity_mapper + ) + + + +class TestGetKiuwanInstance(unittest.TestCase): + + @patch('devsecops_engine_tools.engine_sast.engine_code.src.infrastructure.driven_adapters.kiuwan.kiuwan_tool.KiuwanTool') + def test_get_kiuwan_instance(self, mock_kiuwan_tool): + # Mock dependencies + devops_platform_gateway = Mock() + + # Mock configuration + kiuwan_config = { + "KIUWAN":{ + "SERVER": { + "BASE_URL": "https://test.kiuwan.com", + "USER": "test_user", + "DOMAIN_ID": "test_domain" + }, + "MODELOS": {"task1": "Model1"} + } + } + + devops_platform_gateway.get_remote_config.return_value = kiuwan_config + devops_platform_gateway.get_variable.side_effect = lambda var: { + "app_name": "test_app", + "build_execution_id": "build123", + "branch_name": "feature/test", + "target_branch": "main", + "build_task": "task1" + }.get(var) + + dict_args = { + "remote_config_repo": "config_repo", + "remote_config_branch": "main", + "token_engine_code": "test_token", + "tool": "KIUWAN" + } + + # Mock KiuwanTool instance + mock_instance = Mock() + mock_kiuwan_tool.return_value = mock_instance + + result = kiuwan_tool.get_kiuwan_instance(dict_args, devops_platform_gateway) + + self.assertEqual(result, mock_instance) + devops_platform_gateway.get_remote_config.assert_called_once() + mock_kiuwan_tool.assert_called_once() + + # Verificar configuración pasada + call_args = mock_kiuwan_tool.call_args[0][0] + self.assertEqual(call_args["host_engine_code"], "https://test.kiuwan.com") + self.assertEqual(call_args["user_engine_code"], "test_user") + self.assertEqual(call_args["token_engine_code"], "test_token") + + +class TestDownloadKiuwanCsvOfficial(TestKiuwanToolBase): + + def setUp(self): + super().setUp() + self.tool = kiuwan_tool.KiuwanTool(self.config) + + @patch("requests.get") + @patch("builtins.open", new_callable=mock_open) + def test_download_csv_success(self, mock_file, mock_requests_get): + # Configurar respuesta simulada + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.headers = {"Content-Type": "text/csv"} + mock_response.iter_content = Mock(return_value=[b"sep=,\n", b"Vulnerability,Line\n", b"XSS,42\n"]) + + mock_requests_get.return_value = mock_response + + analysis_code = "ANALYSIS123" + output_path = "kiuwan_findings.csv" + + # Ejecutar el método + result = self.tool._download_kiuwan_csv_official(analysis_code, output_path) + + # Construir URL esperada + expected_url = ( + f"{self.tool.base_url}/applications/analysis/vulnerabilities/export?" + f"application={self.tool.repository_name}&code=ANALYSIS123&type=CSV" + ) + + # Verificaciones + mock_requests_get.assert_called_once() + call_args = mock_requests_get.call_args + self.assertEqual(call_args.args[0], expected_url) + self.assertEqual(call_args.kwargs["auth"], (self.tool.user, self.tool.password)) + self.assertIn("stream", call_args.kwargs) + self.assertTrue(call_args.kwargs["stream"]) + + # Verificar que se abrió el archivo y se escribió + mock_file.assert_called_once_with(output_path, "wb") + mock_file().write.assert_any_call(b"sep=,\n") + mock_file().write.assert_any_call(b"Vulnerability,Line\n") + mock_file().write.assert_any_call(b"XSS,42\n") + + # Verificar retorno + self.assertEqual(result, output_path) + + @patch("requests.get") + def test_download_csv_non_csv_content_type_warning(self, mock_requests_get): + # Simular respuesta sin Content-Type CSV + mock_response = Mock() + mock_response.raise_for_status.return_value = None + mock_response.headers = {"Content-Type": "application/octet-stream"} # No es CSV + mock_response.iter_content = Mock(return_value=[b"data"]) + + mock_requests_get.return_value = mock_response + + with patch("builtins.open", mock_open()) as mock_file: + with patch.object(kiuwan_tool.logger,"warning") as mock_warn: + result = self.tool._download_kiuwan_csv_official("ANALYSIS123", "kiuwan_findings.csv") + mock_warn.assert_called_once() + self.assertEqual(result, "kiuwan_findings.csv") + + @patch("requests.get") + def test_download_csv_request_failed(self, mock_requests_get): + # Simular error en la solicitud + mock_requests_get.side_effect = requests.RequestException("Connection timeout") + + with self.assertRaises(RuntimeError) as context: + self.tool._download_kiuwan_csv_official("ANALYSIS123", "kiuwan_findings.csv") + + self.assertIn("Error downloading Kiuwan CSV", str(context.exception)) + self.assertIn("Connection timeout", str(context.exception)) + + @patch("requests.get") + def test_download_csv_http_error(self, mock_requests_get): + # Simular error 404 + mock_response = Mock() + mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") + mock_requests_get.return_value = mock_response + + with self.assertRaises(RuntimeError) as context: + self.tool._download_kiuwan_csv_official("ANALYSIS123", "kiuwan_findings.csv") + + self.assertIn("Error downloading Kiuwan CSV", str(context.exception)) + self.assertIn("404 Not Found", str(context.exception)) \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_sast/engine_secret/src/infrastructure/driven_adapters/gitleaks/gitleaks_tool.py b/tools/devsecops_engine_tools/engine_sast/engine_secret/src/infrastructure/driven_adapters/gitleaks/gitleaks_tool.py index 40b0d2be3..5a3e18e09 100644 --- a/tools/devsecops_engine_tools/engine_sast/engine_secret/src/infrastructure/driven_adapters/gitleaks/gitleaks_tool.py +++ b/tools/devsecops_engine_tools/engine_sast/engine_secret/src/infrastructure/driven_adapters/gitleaks/gitleaks_tool.py @@ -67,7 +67,7 @@ def _create_report(self, output_file, combined_data): json.dump(combined_data, f, ensure_ascii=False, indent=4) def _check_path(self, path, excluded_paths): - parts = path.split(os.sep) + parts = [p for p in path.replace('\\', '/').split('/') if p] for part in parts: if part in excluded_paths: return True return False diff --git a/tools/devsecops_engine_tools/engine_utilities/azuredevops/models/AzurePredefinedVariables.py b/tools/devsecops_engine_tools/engine_utilities/azuredevops/models/AzurePredefinedVariables.py index 92d3db89c..31cae4ef7 100644 --- a/tools/devsecops_engine_tools/engine_utilities/azuredevops/models/AzurePredefinedVariables.py +++ b/tools/devsecops_engine_tools/engine_utilities/azuredevops/models/AzurePredefinedVariables.py @@ -68,3 +68,6 @@ class VMVariables(BaseEnum): Vm_Product_Type_Name = "Vm.Product.Type.Name" Vm_Product_Name = "Vm.Product.Name" Vm_Product_Description = "Vm.Product.Description" + +class ApplicationVariables(BaseEnum): + Application_Build_Task = "BUILDTASK" \ No newline at end of file diff --git a/tools/devsecops_engine_tools/engine_utilities/github/models/GithubPredefinedVariables.py b/tools/devsecops_engine_tools/engine_utilities/github/models/GithubPredefinedVariables.py index 720afa642..ec038aa90 100644 --- a/tools/devsecops_engine_tools/engine_utilities/github/models/GithubPredefinedVariables.py +++ b/tools/devsecops_engine_tools/engine_utilities/github/models/GithubPredefinedVariables.py @@ -60,3 +60,6 @@ class VMVariables(BaseEnum): Vm_Product_Type_Name = "Vm.Product.Type.Name" Vm_Product_Name = "Vm.Product.Name" Vm_Product_Description = "Vm.Product.Description" + +class ApplicationVariables(BaseEnum): + Application_Build_Task = "BUILDTASK" \ No newline at end of file