import json import sys from abc import ABC, abstractmethod from pathlib import Path from colorama import Fore from common import ( ExitCode, exec_command, exit_message, get_env_variable, get_env_variable_required, get_version, is_semantic_released, print_message, ) from common.defectdojo_helper import DefectDojoHelper from common.dependency_track_helper import DependencyTrackHelper from common.trivy_helper import scan_vulnerabilities from common.validations.buildable import BuildableProject from common.validations.changelog.validate import ValidateChangelog from common.validations.issues import check_and_validate_issues from common.validations.release_notes import ValidateCommitMessage from src.enum.defect_dojo_severities import DefectdojoFindingsSeverity class ProjectInterface(ABC): def __init__(self) -> None: self.buildable_project = BuildableProject() self.version = get_version() self.sci_debug = get_env_variable("SCI_DEBUG") self.ci_project_dir = get_env_variable_required("CI_PROJECT_DIR") def validate(self): skip_changelog_validation = get_env_variable("SKIP_CHANGELOG_VALIDATION") skip_release_notes_validation = get_env_variable( "SKIP_RELEASE_NOTES_VALIDATION" ) if not skip_changelog_validation and not is_semantic_released(): validate_changelog = ValidateChangelog() validate_changelog.validate() elif is_semantic_released() and not skip_release_notes_validation: validator = ValidateCommitMessage() validator.validate() check_and_validate_issues() @abstractmethod def compile(self): pass @abstractmethod def unit_test(self): pass def trivy_scan_vulnerabilities(self): self._docker_login() scan_vulnerabilities() @abstractmethod def sonar_scanner(self): pass @abstractmethod def package(self): pass def defect_dojo_import(self): defectdojo_helper = DefectDojoHelper() engagement = self.import_scans(defectdojo_helper) self.print_engagement_url(defectdojo_helper, engagement) def print_engagement_url(self, defectdojo_helper, engagement): print(" ") print_message( "Report importado com sucesso, para verificar clique no link: " f"{defectdojo_helper.base_url}/engagement/{engagement['id']}" ) print(" ") message, color = self._check_expired_findings(defectdojo_helper, engagement) print_message(message, color) print(" ") def import_scans(self, defectdojo_helper): engagement = defectdojo_helper.prepare_import() for report in Path("trivy").glob("*.json"): with report.open("rb") as file: # ID correspondente ao test type do CycloneDX https://defectdojo.devops.senior.com.br/test_type/63/edit cyclonedx_test_type_id = 63 data = { "scan_type": "CycloneDX Scan", "test_title": report.stem.replace("_", " ").title(), "test_type": cyclonedx_test_type_id, } files = { "file": (report.name, file, "application/json"), } defectdojo_helper.import_scan(engagement["id"], data, files) return engagement def _check_expired_findings(self, defectdojo_helper, engagement): sci_defectdojo_severity = get_env_variable( "SCI_DEFECTDOJO_SEVERITY", DefectdojoFindingsSeverity.LOW ) sci_skip_defectdojo_fail = get_env_variable("SCI_SKIP_DEFECTDOJO_FAIL") severity = DefectdojoFindingsSeverity[sci_defectdojo_severity.upper()] severities = list(DefectdojoFindingsSeverity) severity_index = severities.index(severity) severity_levels = severities[severity_index:] severities_str_list = [ str(severity_level) for severity_level in severity_levels ] result, params = defectdojo_helper.get_product_expired_findings( engagement, severities_str_list ) filtered_url = ( f"{defectdojo_helper.base_url}/product/{engagement['product']}/finding/open?severity={params['severity']}&" f"active={params['active']}&outside_of_sla={params['outside_of_sla']}&test__engagement={params['test__engagement']}" ) if result["count"] > 0: message = ( f"Você tem {result['count']} findings com SLA expirado." f" Incluindo as severidades: {self.severity_message(severities_str_list)}." f" Para visualiza-las entre no link: {filtered_url}" ) if not sci_skip_defectdojo_fail: exit_message(message) return (f"{message} Isso não irá quebrar sua pipeline.", Fore.YELLOW) return ( "Parabéns! Você não tem nenhuma finding com SLA expirado (nas severidades " f"{self.severity_message(severities_str_list)}). Continue o bom trabalho!", Fore.GREEN, ) @staticmethod def severity_message(severities_str_list): if len(severities_str_list) > 1: severities_str = ( ", ".join(severities_str_list[:-1]) + " e " + severities_str_list[-1] ) else: severities_str = severities_str_list[0] return severities_str def _docker_login(self): if not Path(Path.home() / ".docker/config.json").is_file(): dockerhub_username = get_env_variable_required("DOCKERHUB_USERNAME") dockerhub_pass = get_env_variable_required("DOCKERHUB_PASS") result = exec_command( f"docker login -u {dockerhub_username} -p {dockerhub_pass}", print_output=False, ) if result.exit_code == ExitCode.ERROR: exit_message("Falha ao realizar o login do docker.") def dependency_track_import(self): dependency_track_helper = DependencyTrackHelper() has_violations = False project_uuid = None current_branch = get_env_variable_required("CI_COMMIT_REF_NAME") is_default_branch = current_branch in ["master", "develop"] files = list(Path("trivy").glob("*.json")) if not files: print_message("Nenhum relatório SBOM encontrado para importar", Fore.YELLOW) return for report in files: print_message(f"Processando {report.name}...") self._load_and_fix_sbom(str(report)) project_uuid, violation_report = ( dependency_track_helper.analyze_dependencies( str(report), skip_violations_for_new=True ) ) if violation_report and violation_report.get("count", 0) > 0: has_violations = True self._display_violation_details(violation_report) if has_violations: if is_default_branch: exit_message( "Pipeline bloqueada - Componentes com violações encontradas.\n" "Em caso de dúvida, contate o time de Cybersecurity para mais informações." ) else: print(" ") print_message( "Atenção: Foram encontradas violações de componentes.\n" "Como esta é uma branch de desenvolvimento, o pipeline não será bloqueado,\n" "mas recomenda-se revisar e corrigir as vulnerabilidades encontradas.", Fore.YELLOW, ) print(" ") print_message( "Relatório importado com sucesso, para verificar acesse: " f"{dependency_track_helper.base_url}/projects/{project_uuid}" ) print(" ") def _display_violation_details(self, violation_report): components = violation_report["components"] project_url = violation_report["project_url"] print(" ") print(f"{Fore.LIGHTRED_EX}Componentes com violações:") for i, (component_key, data) in enumerate(sorted(components.items()), 1): policies = ", ".join(sorted(data["policies"])) print( f" {i}. {component_key} - Policies: {policies}", Fore.LIGHTRED_EX, ) print( f"Total: {len(components)} componente(s) afetado(s)", Fore.LIGHTRED_EX, ) print(" ") print(f"Mais detalhes: {project_url}") @staticmethod def _load_and_fix_sbom(filepath): with open(filepath, encoding="utf-8") as f: sbom = json.load(f) for component in sbom.get("components", []): props = {p["name"]: p["value"] for p in component.get("properties", [])} if ( props.get("aquasecurity:trivy:Class") == "lang-pkgs" and "purl" not in component ): component_name = component.get("name", "unknown") component["purl"] = f"pkg:file/{component_name}" with open(filepath, "w", encoding="utf-8") as f: json.dump(sbom, f, indent=2)