import io import json import os import re import shutil import traceback import types import zipfile from base64 import b64decode from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import contextmanager, redirect_stderr from dataclasses import asdict, dataclass from datetime import datetime, timedelta from pathlib import Path from shutil import copy import demistomock as demisto # noqa: F401 import git import requests from CommonServerPython import * # noqa: F401 from demisto_sdk.commands.common.constants import ENTITY_TYPE_TO_DIR, FileType from demisto_sdk.commands.common.logger import logging_setup from demisto_sdk.commands.common.tools import find_type from demisto_sdk.commands.split.ymlsplitter import YmlSplitter from importlib.metadata import version from ruamel.yaml import YAML DEFAULT_CONFIG_CATEGORY = "xsoar_best_practices_path_based_validations" CONTENT_DIR_PATH = "/tmp/content" PACKS_DIR_NAME = "Packs" CONTENT_REPO_URL = "https://github.com/demisto/content.git" CACHED_MODULES_DIR = "/tmp/cached_modules" PRE_COMMIT_TEMPLATE_PATH = os.path.join(CONTENT_DIR_PATH, ".pre-commit-config_template.yaml") PRE_COMMIT_MODE = "validate_content" BRANCH_MASTER = "master" DEFAULT_ERROR_PATTERN = { "regex": re.compile(r"(\/[\w\/\.-]+):(\d+):(\d+): .* : (.*)"), "groups": ["file", "line", "column", "details"], } HOOK_ID_TO_PATTERN = { "xsoar-lint": DEFAULT_ERROR_PATTERN, "debug-statements": { "regex": re.compile(r'File\s+"(.+)",\s+line\s+(\d+)(?:.*?\n.*?\^+\n)\s+([^\n]+)'), "groups": ["file", "line", "details"], }, "check-ast": {"regex": re.compile(r'File\s"(Packs/.*?)",\sline\s(\d+)'), "groups": ["file", "line"]}, "mypy": { "regex": re.compile(r"(.*?\.py):(\d+): error: ([\s\S]*?)(?=\nPacks\/|\nFound \d+ error)"), "groups": ["file", "line", "details"], }, } FILE_TYPE_TO_ERROR_TYPE = {"py": "Code", "ps1": "Code", "yml": "Settings", "json": "Settings", "md": "Settings"} ALLOWED_FILE_TYPES = ["py", "yml", "yaml", "json", "ps1", "zip"] SKIPPED_HOOKS = [ "validate-deleted-files", "pwsh-test-in-docker", "pwsh-analyze-in-docker", "coverage-pytest-analyze", "merge-pytest-reports", "format", "validate", "validate-content-paths", "validate-conf-json", "check-merge-conflict", "name-tests-test", "check-added-large-files", "check-case-conflict", "poetry-check", "autopep8", "pycln", "ruff", "xsoar-lint", "check-yaml", "check-json", ] class FormattedResultFields: NAME = "Name" ERROR = "Error" LINE = "Line" ERROR_CODE_OR_LINTER = "Error Code/Linter" @dataclass class ValidationResult: filePath: str = "" fileType: str = "" errorCode: str = "" errorType: str = "" message: str = "" name: str = "" linter: str = "" severity: str = "error" entityType: str = "" col: int = 0 row: int = 0 relatedField: str = "" ui: bool = True def to_dict(self): return asdict(self) @contextmanager def ConstantTemporaryDirectory(path): """Creates a temporary directory with a constant name.""" def cleanup(): # Cleanup: Remove the directory if exists. if os.path.exists(path): shutil.rmtree(path) demisto.debug(f"Temporary directory {path} cleaned up.") cleanup() os.makedirs(path, exist_ok=True) yield path def log_demisto_sdk_version(): try: demisto.debug(f'Using demisto-sdk version {version("demisto-sdk")}') except Exception as e: demisto.debug(f"Could not get demisto-sdk version. Error: {e}") def setup_proxy(_args: dict): if _args.get("use_system_proxy") == "no": # Remove proxy environment variables if they exist for proxy_var in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]: os.environ.pop(proxy_var, None) def strip_ansi_codes(text): ansi_escape = re.compile( r""" \x1B # ESC \[ # [ [0-?]* # Parameter bytes [ -/]* # Intermediate bytes [@-~] # Final byte """, re.VERBOSE, ) return ansi_escape.sub("", text) def extract_hook_id(output: str) -> str: """ Extracts the hook id from a pre-commit hook's output. Args: output (str): The raw output string. Returns: str: The extracted hook id or None if not found. """ pattern = r"- hook id:\s+([\w-]+)" match = re.search(pattern, output) return match.group(1) if match else "" def parse_pre_commit_output(output: str, pattern_obj: dict) -> list[dict]: """ Extracts information from pre-commit hook's output lines based on a given pattern. Args: output (str): Hook's output string to be processed. pattern_obj (str): The regular expression pattern object, including match groups to match against each line. Returns: dict: Extracted information for a single matching line. """ results = [] demisto.debug(f"parse_pre_commit_output got {pattern_obj}") regex = pattern_obj["regex"] group_names = pattern_obj["groups"] for match in re.finditer(regex, output): match_data = match.groups() # Build the result dictionary based on the group names result = {} result.update({name: match_data[i] for i, name in enumerate(group_names) if i < len(match_data)}) if result not in results: results.append(result) return results def get_skipped_hooks(): return SKIPPED_HOOKS def resolve_entity_type(file_path: str): """Resolve entity type from file path.""" parts = file_path.split("/") if parts[0] == "Packs" and len(parts) > 2 and (entity_type_directory_name := parts[2].lower()): entity_type = entity_type_directory_name[:-1] if entity_type_directory_name.endswith("s") else entity_type_directory_name return entity_type entity_type = "contentpack" return entity_type def get_pack_name(zip_filepath: str) -> str: """Returns the pack name from the zipped contribution file's metadata.json file.""" with zipfile.ZipFile(zip_filepath) as zipped_contrib, zipped_contrib.open("metadata.json") as metadata_file: metadata = json.loads(metadata_file.read()) if pack_name := metadata.get("name"): return pack_name demisto.error("Could not find pack name in metadata.json") return "TmpPack" def _create_pack_base_files(self): """ Creates empty 'README.md', '.secrets-ignore', and '.pack-ignore' files that are expected to be in the base directory of a pack """ fp = open(os.path.join(self.pack_dir_path, "README.md"), "a") fp.close() fp = open(os.path.join(self.pack_dir_path, ".secrets-ignore"), "a") fp.close() fp = open(os.path.join(self.pack_dir_path, ".pack-ignore"), "a") fp.close() def content_item_to_package_format( self, content_item_dir: str, del_unified: bool = True, source_mapping: dict | None = None, # noqa: F841 ) -> None: from demisto_sdk.commands.init.contribution_converter import AUTOMATION, INTEGRATION, SCRIPT, get_child_files child_files = get_child_files(content_item_dir) for child_file in child_files: cf_name_lower = os.path.basename(child_file).lower() if cf_name_lower.startswith((SCRIPT, AUTOMATION, INTEGRATION)) and cf_name_lower.endswith("yml"): content_item_file_path = child_file file_type = find_type(content_item_file_path) file_type = file_type.value if file_type else file_type try: extractor = YmlSplitter( input=content_item_file_path, output=content_item_dir, file_type=file_type, ) extractor.extract_to_package_format() except Exception as e: err_msg = ( f'Error occurred while trying to split the unified YAML "{content_item_file_path}" ' f'into its component parts.\nError: "{e}"' ) self.contrib_conversion_errs.append(err_msg) if del_unified: os.remove(content_item_file_path) def convert_contribution_to_pack(contrib_converter): """ Creates or updates a pack in the Content repo from the contents of a contributed zip file. Args: contrib_converter (ContributionConverter): Contribution contributor object """ from demisto_sdk.commands.init.contribution_converter import INTEGRATIONS_DIR, SCRIPTS_DIR, get_child_directories # Only create pack_metadata.json and base pack files if creating a new pack. if contrib_converter.create_new: if contrib_converter.contribution: # Create pack metadata file. with ( zipfile.ZipFile(contrib_converter.contribution) as zipped_contrib, zipped_contrib.open("metadata.json") as metadata_file, ): metadata = json.loads(metadata_file.read()) demisto.debug(f"convert_contribution_to_pack {metadata=}") contrib_converter.create_metadata_file(metadata) # Create base files. contrib_converter.create_pack_base_files = types.MethodType(_create_pack_base_files, contrib_converter) contrib_converter.create_pack_base_files() # Unpack. contrib_converter.unpack_contribution_to_dst_pack_directory() # Convert. unpacked_contribution_dirs = get_child_directories(contrib_converter.pack_dir_path) for unpacked_contribution_dir in unpacked_contribution_dirs: contrib_converter.convert_contribution_dir_to_pack_contents(unpacked_contribution_dir) # Extract to package format. for pack_subdir in get_child_directories(contrib_converter.pack_dir_path): basename = os.path.basename(pack_subdir) if basename in {SCRIPTS_DIR, INTEGRATIONS_DIR}: contrib_converter.content_item_to_package_format = types.MethodType(content_item_to_package_format, contrib_converter) contrib_converter.content_item_to_package_format( pack_subdir, del_unified=False, source_mapping=None, ) def prepare_content_pack_for_validation(filename: str, data: bytes, content_dir_path: str) -> str: from demisto_sdk.commands.init.contribution_converter import ContributionConverter # Write zip file data to file system. zip_path = os.path.abspath(os.path.join(content_dir_path, filename)) with open(zip_path, "wb") as fp: fp.write(data) pack_name = get_pack_name(zip_path) contrib_converter = ContributionConverter(name=pack_name, contribution=zip_path, base_dir=content_dir_path) convert_contribution_to_pack(contrib_converter) os.remove(zip_path) return contrib_converter.pack_dir_path def prepare_single_content_item_for_validation(file_name: str, data: bytes, packs_path: str) -> str: from demisto_sdk.commands.init.contribution_converter import ContributionConverter pack_name = "TmpPack" pack_path = os.path.join(packs_path, pack_name) demisto.debug(f"Pack name: {pack_name}") # create pack_metadata.json file in TmpPack contrib_converter = ContributionConverter(name=pack_name, pack_dir_name=pack_name, contribution=pack_name) contrib_converter.create_metadata_file({"description": "Temporary Pack", "author": "xsoar"}) # Determine entity type by filename prefix. file_name_prefix = "-".join(file_name.split("-")[:-1]) containing_dir = os.path.join(pack_path, ENTITY_TYPE_TO_DIR.get(file_name_prefix, "Integrations")) os.makedirs(containing_dir, exist_ok=True) is_json = file_name.casefold().endswith(".json") data_as_string = data.decode() yaml = YAML() loaded_data = json.loads(data_as_string) if is_json else yaml.load(data_as_string) if is_json: data_as_string = json.dumps(loaded_data) else: buff = io.StringIO() yaml.dump(loaded_data, buff) data_as_string = buff.getvalue() # Write content item file to file system. file_path = Path(os.path.join(containing_dir, file_name)) file_path.write_text(data_as_string) file_type = find_type(str(file_path)) file_type = file_type.value if file_type else file_type if is_json or file_type in (FileType.PLAYBOOK.value, FileType.TEST_PLAYBOOK.value): return str(file_path) extractor = YmlSplitter(input=str(file_path), file_type=file_type, output=containing_dir) # Validate the resulting package files, ergo set path_to_validate to the package directory that results # from extracting the unified yaml to a package format extractor.extract_to_package_format() output_path = extractor.get_output_path() demisto.debug(f"prepare_single_content_item_for_validation {output_path=}") return output_path def run_validate(path_to_validate: str, json_output_file: str) -> int: """ Runs demisto-sdk validations on a specified file path and writes the results to a JSON file. Args: path_to_validate (str): The path of the file or directory to be validated. json_output_file (str): The file path where validation results will be written in JSON format. Returns: int: An exit code indicating the validation status; 0 for success and non-zero for failures. """ from demisto_sdk.commands.common.constants import ExecutionMode from demisto_sdk.commands.validate.config_reader import ConfigReader from demisto_sdk.commands.validate.initializer import Initializer from demisto_sdk.commands.validate.validate_manager import ValidateManager from demisto_sdk.commands.validate.validation_results import ResultWriter result_writer = ResultWriter(json_output_file) config_reader = ConfigReader(category=DEFAULT_CONFIG_CATEGORY) initializer = Initializer( staged=False, committed_only=False, file_path=str(path_to_validate), execution_mode=ExecutionMode.SPECIFIC_FILES ) validate_manager = ValidateManager(result_writer, config_reader, initializer, allow_autofix=False, ignore=["BA106"]) demisto.debug(f"run_validate validate_manager initialized. Running validations: {validate_manager.validators=}") err_file = io.StringIO() with redirect_stderr(err_file): exit_code: int = validate_manager.run_validations() demisto.debug(f"run_validate {exit_code=}") return exit_code def get_content_modules(content_path: str, verify_ssl: bool = True) -> None: """ Copies the required content modules for validation and pre-commit from the cached dir. The cached dir is updated once a day Args: content_path (str): Path to Content directory verify_ssl (bool): Whether to verify SSL """ modules = [ { "file": "CommonServerPython.py", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Packs/Base/Scripts" "/CommonServerPython/CommonServerPython.py", "content_path": "Packs/Base/Scripts/CommonServerPython", }, { "file": "CommonServerPowerShell.ps1", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Packs/Base/Scripts" "/CommonServerPowerShell/CommonServerPowerShell.ps1", "content_path": "Packs/Base/Scripts/CommonServerPowerShell", }, { "file": "demistomock.py", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Tests/demistomock/demistomock.py", "content_path": "Tests/demistomock", }, { "file": "demistomock.ps1", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Tests/demistomock/demistomock.ps1", "content_path": "Tests/demistomock", }, { "file": "tox.ini", "github_url": "https://raw.githubusercontent.com/demisto/content/master/tox.ini", "content_path": ".", }, { "file": "conftest.py", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Tests/scripts/dev_envs/pytest/conftest.py", "content_path": "Tests/scripts/dev_envs/pytest", }, { "file": "approved_usecases.json", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Config/approved_usecases.json", "content_path": "Config", }, { "file": "approved_tags.json", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Config/approved_tags.json", "content_path": "Config", }, { "file": "approved_categories.json", "github_url": "https://raw.githubusercontent.com/demisto/content/master/Config/approved_categories.json", "content_path": "Config", }, { "file": ".pre-commit-config_template.yaml", "github_url": "https://raw.githubusercontent.com/demisto/content/master/.pre-commit-config_template.yaml", "content_path": "", }, { "file": ".pre-commit-config_template.yaml", "github_url": "https://raw.githubusercontent.com/demisto/content/master/.pre-commit-config_template.yaml", "content_path": "", }, ] for module in modules: demisto.debug(f'get_content_modules getting {module["file"]=}') module_path = os.path.join(content_path, module["content_path"]) os.makedirs(Path(module_path), exist_ok=True) try: cached_module_path = os.path.join(CACHED_MODULES_DIR, module["file"]) fname = Path(cached_module_path) modified_time = ( datetime.fromtimestamp(fname.stat().st_mtime) if os.path.isfile(cached_module_path) else datetime(1970, 1, 1) ) if modified_time + timedelta(days=1) < datetime.utcnow(): demisto.debug(f'Downloading {module["file"]} from git') res = requests.get(module["github_url"], verify=verify_ssl, timeout=10) res.raise_for_status() with open(cached_module_path, "wb") as f: f.write(res.content) demisto.debug(f"Copying from {cached_module_path} to {module_path}") copy(cached_module_path, module_path) except Exception as e: fallback_path = f'/home/demisto/{module["file"]}' demisto.debug(f'Failed downloading content module {module["github_url"]} - {e}. Copying from {fallback_path}') copy(fallback_path, module_path) def run_pre_commit(output_path: Path) -> int: """ Runs demisto-sdk pre-commit. Args: output_path (str): The file path where validation results will be written in JSON format. Returns: int: An exit code indicating the validation status; 0 for success and non-zero for failures. """ from demisto_sdk.commands.pre_commit.pre_commit_command import pre_commit_manager os.environ["DEMISTO_SDK_DISABLE_MULTIPROCESSING"] = "true" demisto.debug(f"run_pre_commit | {get_skipped_hooks()=} | {PRE_COMMIT_TEMPLATE_PATH=} | {output_path=}") exit_code = pre_commit_manager( skip_hooks=get_skipped_hooks(), all_files=True, run_docker_hooks=False, pre_commit_template_path=Path(PRE_COMMIT_TEMPLATE_PATH), json_output_path=output_path, mode=PRE_COMMIT_MODE, ) demisto.debug(f"run_pre_commit {exit_code=}") return exit_code def read_json_results(json_path: Path, results: list = None) -> list: """ Process JSON results file and append items to results list. Args: json_path: JSON file path results: Existing results list Returns: Updated results with 'file_name' added to each result """ if results is None: results = [] content = json.loads(json_path.read_text()) if not content: return results file_name = json_path.stem if isinstance(content, list): for item in content: item["file_name"] = file_name results.extend(content) else: content["file_name"] = file_name results.append(content) return results def read_validate_results(json_path: Path): if not json_path.exists(): raise DemistoException("Validation Results file does not exist.") raw_outputs = read_json_results(json_path) demisto.debug(f"read_validate_results: {raw_outputs=}") results = [] for output in raw_outputs: for validation in output.get("validations", []): file_path = validation.get("file path", "") file_type = "yml" if file_path.endswith((".yml", ".yaml")) else "" error_code = validation.get("error code", "") message = validation.get("message", "") results.append( ValidationResult( filePath=str(Path(file_path).absolute()) if file_path else "", name=Path(file_path).stem, fileType=file_type, errorCode=error_code, errorType="Code" if file_type in {"py", "ps1"} else "Settings", entityType=resolve_entity_type(file_path), message=message, linter="validate", ) ) return results def read_pre_commit_results(pre_commit_dir: Path): results = [] for output_file in pre_commit_dir.iterdir(): raw_outputs = read_json_results(output_file) for output in raw_outputs: stdout: str = strip_ansi_codes(output.get("stdout", "")) demisto.debug(f"stripped-output: {stdout}") hook_id: str = extract_hook_id(stdout) or output.get("file_name", "") pattern_obj: dict = HOOK_ID_TO_PATTERN.get(hook_id, DEFAULT_ERROR_PATTERN) parsed_results: list[dict] = parse_pre_commit_output(stdout, pattern_obj) demisto.debug(f"extracted_data={json.dumps(parsed_results, indent=4)}") for result in parsed_results: file_path = result.get("file", "") # Isolating file's extension. file_type = "" if not file_path else os.path.splitext(f"{file_path}")[1].lstrip(".") error_type = FILE_TYPE_TO_ERROR_TYPE.get(file_type, "") # 'check-ast' details value has to be treated individually as regex does not capture it properly. if hook_id == "check-ast": result["details"] = stdout.splitlines()[5:] # Trimming error metadata info (5 lines of it). details = result.get("details", "") results.append( ValidationResult( filePath=file_path, fileType=file_type, name=Path(file_path).stem, entityType=resolve_entity_type(file_path), errorType=error_type, message=details, linter=hook_id, col=result.get("column", 0), row=int(result.get("line", 0)) - 1, # Normalizing as UI adds a module registration line at row=1. ) ) return results def validate_content(path_to_validate: str) -> tuple[list, list]: """ Validate the content items in the given `path_to_validate`, using demisto-sdk's ValidateManager and PreCommitManager. Arguments: path_to_validate: Path to the file/directory to validate. Returns: Tuple[list, list]: Formatted validation results, and raw validation results. """ demisto.info(f"Starting to validate content at {path_to_validate}.") output_base_dir = Path("ValidateContentOutput") / f'run-{datetime.now().strftime("%Y%m%d-%H%M%S")}' os.makedirs(output_base_dir, exist_ok=True) validations_output_path = output_base_dir / "validation_res.json" pre_commit_dir = output_base_dir / "pre-commit-output/" os.makedirs(pre_commit_dir, exist_ok=True) # One thread for `validate` execution & one for `pre_commit`. with ThreadPoolExecutor(max_workers=2) as executor: validate_future = executor.submit(run_validate, path_to_validate, str(validations_output_path)) demisto.info("Submitting `run_validate` future.") pre_commit_future = executor.submit(run_pre_commit, pre_commit_dir) demisto.info("Submitting `pre_commit` future.") for future in as_completed([validate_future, pre_commit_future]): if future == validate_future: validate_exit_code = future.result(timeout=60) # One minute timeout. demisto.info(f"Finished running `demisto-sdk validate` with exit code {validate_exit_code}.") else: pre_commit_exit_code = future.result(timeout=60) # One minute timeout. demisto.info(f"Finished running `demisto-sdk pre-commit` with exit code {pre_commit_exit_code}.") # If no errors were found. if not (validate_exit_code or pre_commit_exit_code): return [], [] raw_validation_results: list[ValidationResult] = [] raw_validation_results += read_validate_results(validations_output_path) raw_validation_results += read_pre_commit_results(pre_commit_dir) demisto.debug(f"{json.dumps([output.to_dict() for output in raw_validation_results], indent=4)}") formatted_results = [] for result in raw_validation_results: formatted_results.append( { FormattedResultFields.NAME: result.name, FormattedResultFields.ERROR: result.message, FormattedResultFields.LINE: result.row if int(result.row) > 0 else None, FormattedResultFields.ERROR_CODE_OR_LINTER: result.errorCode or result.linter, } ) return formatted_results, [output.to_dict() for output in raw_validation_results] def setup_content_repo(content_path: str): """Set up local Content git repository to run demisto-sdk commands against.""" content_repo = git.Repo.init(content_path) demisto.debug(f"main created content_repo {os.listdir(content_path)=}") # Check if the repository has any commits, make an initial commit if needed. if not content_repo.head.is_valid(): # Make an empty initial commit to create the master branch. content_repo.index.commit("Initial commit") # Set up the remote branch and fetch it. content_repo.create_remote("origin", CONTENT_REPO_URL) content_repo.remotes.origin.fetch("master", depth=1) # Ensure 'master' branch exists, and checkout. if BRANCH_MASTER not in content_repo.heads: content_repo.create_head(BRANCH_MASTER) content_repo.heads.master.checkout() return content_repo def get_file_name_and_contents( filename: str | None = None, data: str | None = None, entry_id: str | None = None, ): if filename and data: return filename, b64decode(data) elif entry_id: file_object = demisto.getFilePath(entry_id) demisto.debug(f"{file_object=}") with open(file_object["path"], "rb") as f: file_contents = f.read() return file_object["name"], file_contents return None def setup_content_dir(file_name: str, file_contents: bytes | str, entry_id: str, verify_ssl=False) -> str: """Sets up the content directory to validate the content items in it.""" # Set up the content directory path globally, required for demisto-sdk logic. os.environ["DEMISTO_SDK_CONTENT_PATH"] = CONTENT_DIR_PATH packs_path = os.path.join(CONTENT_DIR_PATH, PACKS_DIR_NAME) Path.mkdir(Path(packs_path)) demisto.debug(f"created packs directory in {packs_path}") content_repo = setup_content_repo(CONTENT_DIR_PATH) file_name, file_contents = get_file_name_and_contents(file_name, str(file_contents), entry_id) file_type = file_name.split(".")[-1] if file_type not in ALLOWED_FILE_TYPES: demisto.debug(f"resolved {file_type=}") raise DemistoException( f"{file_name} does not define a content item. Files defining content items can be of types: {ALLOWED_FILE_TYPES}" ) if isinstance(file_contents, str): size_in_bytes = len(file_contents.encode("utf-8")) else: # Assuming it's already bytes size_in_bytes = len(file_contents) demisto.debug( f"setup_content_dir preparing content_items for validateion: " f"{file_name=}\n|" f" file_content size in bytes={size_in_bytes} \n|" f" {packs_path if packs_path else CONTENT_DIR_PATH}" ) if file_name.endswith(".zip"): path_to_validate = prepare_content_pack_for_validation(file_name, file_contents, CONTENT_DIR_PATH) else: path_to_validate = prepare_single_content_item_for_validation(file_name, file_contents, packs_path) demisto.debug(f"setup_content_dir {path_to_validate=}") # "git add packs_path" content_repo.index.add(os.path.join(packs_path)) os.makedirs(CACHED_MODULES_DIR, exist_ok=True) get_content_modules(CONTENT_DIR_PATH, verify_ssl=verify_ssl) return path_to_validate def setup_envvars(): os.environ["DEMISTO_SDK_IGNORE_CONTENT_WARNING"] = "false" os.environ["DEMISTO_SDK_OFFLINE_ENV"] = "False" os.environ["ARTIFACTS_FOLDER"] = "/tmp/artifacts" os.environ["DEMISTO_SDK_LOG_NO_COLORS"] = "true" demisto.debug(f"setup_envvars: {os.environ}") def main(): setup_envvars() # Save working directory for later return, as working directory changes during runtime. cwd = os.getcwd() demisto.debug(f"{cwd=}") try: args = demisto.args() demisto.debug(f"Got {args=}") setup_proxy(args) verify_ssl = argToBoolean(args.get("trust_any_certificate")) # Whether `filename` & `data` will be provided, or an `entry_id`. filename: str = args.get("filename", "") data: bytes | str = args.get("data", b"") entry_id: str = args.get("entry_id", "") with ConstantTemporaryDirectory(CONTENT_DIR_PATH) as tmp_dir: demisto.info("Setting up content validation environment.") demisto.debug(f"created {tmp_dir=}") # Setup Demisto SDK's logging. logging_setup( calling_function="ValidateContent", console_threshold="DEBUG" if is_debug_mode() else "ERROR", propagate=True, ) demisto.debug("Finished setting logger.") path_to_validate: str = setup_content_dir(filename, data, entry_id, verify_ssl) demisto.debug("Finished setting content dir.") # Got to be in content dir when running demisto-sdk commands. os.chdir(CONTENT_DIR_PATH) validation_results, raw_outputs = validate_content(path_to_validate) os.chdir(cwd) if not raw_outputs: readable_output = "All validations passed." else: readable_output = tableToMarkdown( name="Validation Results", t=validation_results, headers=[ FormattedResultFields.NAME, FormattedResultFields.ERROR, FormattedResultFields.LINE, FormattedResultFields.ERROR_CODE_OR_LINTER, ], ) return_results( CommandResults( readable_output=readable_output, outputs_prefix="ValidationResult", outputs=validation_results, raw_response=raw_outputs, ) ) demisto.info("Finished validating content.") except Exception as e: demisto.error(traceback.format_exc()) return_error(f"Failed to execute ValidateContent. Error: {e!s}") finally: os.chdir(cwd) if __name__ in ("__main__", "__builtin__", "builtins"): log_demisto_sdk_version() main()