ValidateContent

Runs validation and linting using the Demisto SDK on content items, such as integrations, automations and content packs. This automation script is used as part of the content validation that runs as part of the contribution flow.

python · Base

Source

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()

README

Runs validation and linting using the Demisto SDK on content items, such as integrations, automations and content packs.
This automation script is used as part of the content validation that runs as part of the contribution flow.

Script Data


Name Description
Script Type python3
Cortex XSOAR Version 5.5.0

Inputs


Argument Name Description
filename Name of the file to validate.
data Base64 encoded contents of the file to validate.
entry_id ID of War Room file entry for the ZIP content pack or the integration/automation YAML.
use_system_proxy Use system proxy settings to download required models from GitHub.
trust_any_certificate Trust any certificate (not secure) to download required models from GitHub.

Outputs


Path Description Type
ValidationResult.Name Name of validated item. String
ValidationResult.Error The validation error message. String
ValidationResult.Line The code line number in which the error was found in the lint. String
ValidationResult.ErrorCode The error code or the name of the linter that identified the issue. String

Script Example

!ValidateContent entry_id=G2SUaH9ZPmfw7QHWQNk2pa@6

Context Example

    "ValidationResult": [
        {
            "Name": "MyScript",
            "Error": "{unterminated string literal (detected at line 166)  [syntax]"
            "Line": 165,
            "Error Code/Linter": "mypy"
        },
        {
            "Name": "MyScript",
            "Error": "The following commands contain duplicated arguments:Command example-my-command, contains multiple appearances of the following arguments message.Please make sure to remove the duplications.",
            "Error Code/Linter": "IN113"
        }
    ]
}

Human Readable Output

Validation Results

Name Error Line Error Code/Linter
MyScript The following commands contain duplicated arguments: Command example-my-command, contains multiple appearances of the following arguments message. Please make sure to remove the duplications.   IN113
MyScript unterminated string literal (detected at line 166) [syntax] 166 mypy