import json from datetime import datetime, timedelta import requests from . import exit_message, get_env_variable_required PRODUCT_TYPE_URL = "/product_types/" PRODUCTS_URL = "/products/" ENGAGEMENTS_URL = "/engagements/" TESTS_URL = "/tests/" REIMPORT_SCAN_URL = "/reimport-scan/" IMPORT_SCAN_URL = "/import-scan/" FINDINGS = "/findings/" class DefectDojoHelper(object): def __init__(self): self.base_url = get_env_variable_required("SCI_DEFECTDOJO_URL") self.base_api_url = f"{self.base_url}/api/v2" self.session = requests.Session() self.session.headers.update( { "Authorization": f'Token {get_env_variable_required("SCI_DEFECTDOJO_API_KEY")}', } ) self.ci_project_namespace = get_env_variable_required("CI_PROJECT_NAMESPACE") self.ci_project_name = get_env_variable_required("CI_PROJECT_NAME") self.ci_project_url = get_env_variable_required("CI_PROJECT_URL") self.ci_commit_branch = get_env_variable_required("CI_COMMIT_REF_NAME") self.ci_commit_sha = get_env_variable_required("CI_COMMIT_SHA") self.ci_pipeline_id = get_env_variable_required("CI_PIPELINE_ID") self.branch_url = f"{self.ci_project_url}/-/tree/{self.ci_commit_branch}" self.environment = ( "Production" if self.ci_commit_branch in ["master", "main"] else "Development" ) self.environment_mapping = { "Production": 3, "Development": 1, } self.environment_id = self.environment_mapping.get(self.environment, 1) def _post(self, url: str, data: dict): return self.session.post( f"{self.base_api_url}{url}", json=data, timeout=30.0, ) def _post_form_data(self, url: str, data: dict, files): return self.session.post( f"{self.base_api_url}{url}", data=data, files=files, timeout=60.0, ) def _get(self, url: str, params: dict): return self.session.get( f"{self.base_api_url}{url}", params=params, timeout=30.0, ) def _patch( self, url: str, item_id: str, data: dict, ): return self.session.patch( f"{self.base_api_url}{url}{item_id}", json=data, timeout=30.0, ) def _get_product_type_id(self): response = self._get(PRODUCT_TYPE_URL, {"name": self.ci_project_namespace}) if response.status_code == 200 and response.json()["count"] > 0: return response.json()["results"][0]["id"] return None def _create_product_type(self): data = { "name": self.ci_project_namespace, } response = self._post(PRODUCT_TYPE_URL, data) if response.status_code != 201: exit_message( f'Falha ao criar Product Type: "{self.ci_project_namespace}": {response.content}' ) return response.json()["id"] def _upsert_product_type(self): product_type_id = self._get_product_type_id() if not product_type_id: product_type_id = self._create_product_type() return product_type_id def _get_product_id(self, product_type_id): data = {"name": self.ci_project_name, "product_type": product_type_id} response = self._get(PRODUCTS_URL, data) if response.status_code == 200 and response.json()["count"] > 0: results = response.json()["results"] for result in results: if result["name"] == data["name"]: return result["id"] return None def _create_product(self, product_type_id): description = ( "Para mais informações consulte o repositório git do projeto " f"[{self.ci_project_name}]({self.ci_project_url})" ) data = { "name": self.ci_project_name, "description": description, "lifecycle": "production", "prod_type": product_type_id, } response = self._post(PRODUCTS_URL, data) if response.status_code != 201: exit_message( f'Falha ao criar Product "{self.ci_project_name,}": {response.content}' ) return response.json()["id"] def _upsert_product(self, product_type_id): product_id = self._get_product_id(product_type_id) if not product_id: product_id = self._create_product(product_type_id) return product_id def _get_engagement(self, product_id): params = {"name": self.ci_commit_branch, "product": product_id} response = self._get(ENGAGEMENTS_URL, params) if response.status_code == 200 and response.json()["count"] > 0: return response.json()["results"][0] return None def _create_engagement(self, product_id, data): data.update( { "name": self.ci_commit_branch, "product": product_id, "status": "In Progress", "target_start": datetime.now().strftime("%Y-%m-%d"), "target_end": (datetime.now() + timedelta(days=365 * 10)).strftime( "%Y-%m-%d" ), "engagement_type": "CI/CD", "source_code_management_uri": self.branch_url, } ) response = self._post(ENGAGEMENTS_URL, data) if response.status_code != 201: exit_message( f'Falha ao criar Engagement "{self.ci_commit_branch}": {response.content}' ) return response.json() def _edit_engagement(self, engagement_id, data): response = self._patch(ENGAGEMENTS_URL, engagement_id, data) if response.status_code != 200: exit_message( f'Falha ao editar Engagement "{self.ci_commit_branch}": {response.content}' ) def _upsert_engagement(self, product_id): base_engagement_data = { "environment": self.environment, "commit_hash": self.ci_commit_sha, "branch_tag": self.ci_commit_branch, "build_id": self.ci_pipeline_id, "build_server": 1, "source_code_management_server": 1, "orchestration_engine": 1, } engagement = self._get_engagement(product_id) if not engagement: engagement = self._create_engagement(product_id, base_engagement_data) else: self._edit_engagement(engagement["id"], base_engagement_data) return engagement def _get_test_id(self, engagement_id, test_type_id, title): params = { "engagement": engagement_id, "test_type": test_type_id, "title": title, } response = self._get(TESTS_URL, params) if response.status_code == 200 and response.json(): results = response.json()["results"] first_item = results[0] if results else {} return first_item.get("id", None) return None def _create_test(self, engagement_id, test_type_id, test_title): data = { "engagement": engagement_id, "test_type": test_type_id, "title": test_title, "environment": self.environment_id, "target_start": datetime.now().strftime("%Y-%m-%d"), "target_end": (datetime.now() + timedelta(days=365 * 10)).strftime( "%Y-%m-%d" ), } response = self._post(TESTS_URL, data) if response.status_code != 201: exit_message(f'Falha ao criar Test "{test_title}": {response.content}') return response.json()["id"] def _upsert_findings(self, engagement_id, data, files): test_id = self._get_test_id( engagement_id, data["test_type"], data["test_title"] ) if test_id: data.update({"test": test_id}) response = self._post_form_data(REIMPORT_SCAN_URL, data, files) else: if data.get("scan_type") == "Generic Findings Import": test_id = self._create_test( engagement_id=engagement_id, test_type_id=data["test_type"], test_title=data["test_title"], ) data.update({"test": test_id}) response = self._post_form_data(REIMPORT_SCAN_URL, data, files) else: data.update({"engagement": engagement_id}) response = self._post_form_data(IMPORT_SCAN_URL, data, files) if response.status_code != 201: exit_message( f"Falha ao importar os findings de {files['file'][0]}. {response} {response.text}" ) return test_id def prepare_import(self): product_type_id = self._upsert_product_type() product_id = self._upsert_product(product_type_id) return self._upsert_engagement(product_id) def import_scan(self, engagement_id, data, files): data.update( { "environment": self.environment, "commit_hash": self.ci_commit_sha, "branch_tag": self.ci_commit_branch, "build_id": self.ci_pipeline_id, } ) return self._upsert_findings(engagement_id, data, files) def get_product_expired_findings(self, engagement, severities_str_list): params = { "active": "true", "test__engagement__product": engagement["product"], "outside_of_sla": 1, "severity": ",".join(severities_str_list), "test__engagement": engagement["id"], } response = self._get(FINDINGS, params) if response.status_code == 200: return response.json(), params exit_message( f'Falha ao verificar Findings do Product "{self.ci_project_name}": {response.content}' )