CommitFiles

Source

import io
import os
from contextlib import redirect_stderr, redirect_stdout
from os.path import exists
from pathlib import Path

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
from demisto_sdk.commands.common.constants import ENTITY_TYPE_TO_DIR
from demisto_sdk.commands.common.tools import find_type
from demisto_sdk.commands.split.ymlsplitter import YmlSplitter

PR_TEMPLATE = (
    "### Pull Request created in Cortex XSOAR\n"
    "**Created by:** {}\n"
    "**Pack:** {}\n"
    "**Branch:** {}\n"
    "**Link to incident in Cortex XSOAR:** {}\n\n"
    "{}\n\n"
    "---"
)

file_path_to_sha: dict[str, str] = {}

files_path = []


class ContentFile:  # pragma: no cover
    path_to_file: str = ""
    file_name: str = ""
    content_type: str = ""
    file_text: str = ""
    entry_id: str = ""

    def __init__(self, pack_name=None, file=None):
        if not pack_name and not file:
            return

        # read the file from entry id
        file_object = demisto.getFilePath(file["EntryID"])
        with open(file_object["path"]) as f:
            file_contents = f.read()

        self.file_text = file_contents
        self.file_name = file["Name"]
        self.entry_id = file["EntryID"]

        if self.file_name == "metadata.json":
            self.file_name = "pack_metadata.json"
            self.path_to_file = os.path.join("Packs", pack_name)
            self.content_type = "metadata"
        else:
            with open(file_object["name"], "w") as f:
                f.write(file_contents)

            file_type = find_type(file_object["name"])
            os.remove(file_object["name"])
            self.content_type = file_type.value if file_type else file_type
            folder = ENTITY_TYPE_TO_DIR.get(self.content_type, "")
            self.path_to_file = os.path.join("Packs", pack_name, folder)


def get_file_sha(branch_name: str, content_file: ContentFile, get_files_command: str):
    full_path = os.path.join(content_file.path_to_file, content_file.file_name)

    sha = file_path_to_sha.get(full_path)
    if sha:
        return sha

    # try to get the file from branch
    status, list_files_res = execute_command(
        get_files_command, {"branch": branch_name, "path": content_file.path_to_file}, fail_on_error=False
    )
    if status:
        for file in list_files_res:
            file_path_to_sha[file["path"]] = file["sha"]

    return file_path_to_sha.get(full_path)


def escape_placeholders(content: str) -> str:
    """
    Escapes XSOAR-style placeholders ${somecontext.value} to prevent context substitution.

    Args:
        content (str): The input string containing placeholders.

    Returns:
        str: The string with placeholders escaped.
    """
    return re.sub(r"\$(\{[^}]+})", r"$\\\1", content)  # Prepend a backslash to preserve the placeholder


def commit_content_item(
    branch_name: str, content_file: ContentFile, new_files: List, modified_files: List, keep_placeholders_in_files: bool = False
):
    commit_args = {
        "commit_message": f"Added {content_file.file_name}",
        "path_to_file": f"{content_file.path_to_file}/{content_file.file_name}",
        "branch_name": branch_name,
        "file_text": content_file.file_text,
    }

    file_sha = get_file_sha(branch_name, content_file, "Github-list-files")

    # dont commit pack_metadata.json if already exists in the branch
    if file_sha and content_file.file_name == "pack_metadata.json":
        return
    elif file_sha:
        # update existing file
        commit_args["file_sha"] = file_sha
        commit_args["commit_message"] = f"Updated {content_file.file_name}"
        modified_files.append(content_file.file_name)
    else:
        # new file added
        if keep_placeholders_in_files:
            demisto.debug(f"Got {keep_placeholders_in_files=}, escaping placeholders with \\")
            commit_args["file_text"] = escape_placeholders(content_file.file_text)
            commit_args["placeholders_escaped"] = "True"
        new_files.append(content_file.file_name)

    status, commit_res = execute_command("Github-commit-file", commit_args, fail_on_error=False)
    if not status:
        raise DemistoException(commit_res)


def commit_content_item_gitlab(branch_name: str, content_file: ContentFile, new_files: List, modified_files: List):
    commit_args = {
        "commit_message": f"Added {content_file.file_name}",
        "file_path": f"{content_file.path_to_file}/{content_file.file_name}",
        "branch": branch_name,
        "file_content": content_file.file_text,
    }
    status, commit_res = execute_command("gitlab-file-create", commit_args, fail_on_error=False)
    if isinstance(commit_res, dict):
        new_files.append(content_file.file_name)
    elif isinstance(commit_res, str) and "already exists" in commit_res:
        demisto.debug(f"The file {content_file.file_name} already exist, running update command")
        if content_file.file_name == "pack_metadata.json":
            return
        commit_args["commit_message"] = f"Updated {content_file.file_name}"
        modified_files.append(content_file.file_name)
        status, commit_res = execute_command("gitlab-file-update", commit_args, fail_on_error=False)
    if not status:
        raise DemistoException(commit_res)


def does_file_exist(branch_name: str, content_file: ContentFile) -> bool:
    full_path = os.path.join(content_file.path_to_file, content_file.file_name)

    if full_path in files_path:
        return True
    # try to get the file from branch
    status, list_commit_res = execute_command(
        "bitbucket-commit-list", {"included_branches": branch_name, "file_path": full_path}, fail_on_error=False
    )
    if list_commit_res and len(list_commit_res) > 0:
        files_path.append(full_path)
        return True
    return False


def searched_file_path(branch_name: str, content_file: ContentFile) -> bool:
    full_path = Path(content_file.path_to_file, content_file.file_name)

    if str(full_path) in files_path:  # the files list, check if the file already exists in the list
        return True
    # try to get the file from branch
    response = execute_command(
        "azure-devops-file-list", args={"branch_name": branch_name.split("/")[-1], "recursion_level": "Full"}
    )
    files_set = {file["path"] for file in response.get("value", [])}
    for file in files_set:
        if str(full_path) in file:
            files_path.append(str(full_path))
            return True
    return False


def commit_content_item_bitbucket(branch_name: str, content_file: ContentFile, new_files: List, modified_files: List):
    commit_args = {
        "message": f"Added {content_file.file_name}",
        "file_name": f"{content_file.path_to_file}/{content_file.file_name}",
        "branch": f"{branch_name}",
        "file_content": f"{content_file.file_text}",
    }
    file_status = does_file_exist(branch_name, content_file)
    # dont commit pack_metadata.json if already exists in the branch
    if file_status and content_file.file_name == "pack_metadata.json":
        return
    elif file_status:
        # update existing file
        commit_args["message"] = f"Updated {content_file.file_name}"
        modified_files.append(content_file.file_name)
    else:
        # new file added
        new_files.append(content_file.file_name)
    try:
        demisto.executeCommand("bitbucket-commit-create", args=commit_args)
    except DemistoException as e:
        raise DemistoException(f"Failed to execute bitbucket-commit-create command. Error: {e}, {traceback.format_exc()}")


def commit_content_item_azure_devops(branch_name: str, content_file: ContentFile, new_files: List, modified_files: List):
    file_args = {
        "commit_comment": f"{content_file.file_name} was added.",
        "file_path": f"{content_file.path_to_file}/{content_file.file_name}",
        "branch_name": f"{branch_name}",
        "file_content": f"{content_file.file_text}",
    }

    file_exists = searched_file_path(branch_name, content_file)

    # don't commit pack_metadata.json if already exists in the branch
    if file_exists and content_file.file_name == "pack_metadata.json":
        return
    response = demisto.executeCommand("azure-devops-branch-list", args={})
    branches_list = response[0].get("Contents", {}).get("value", []) if response and isinstance(response, list) else []
    for branch in branches_list:
        if branch["name"] == branch_name:
            branch_id = branch["objectId"]
            break
    else:
        raise DemistoException("Failed to find a corresponding branch id to the given branch name.")
    file_args["branch_id"] = branch_id

    if file_exists:
        # update existing file
        file_args["commit_comment"] = f"{content_file.file_name} was updated."
        modified_files.append(content_file.file_name)
        command_to_execute = "azure-devops-file-update"
    else:
        # new file added
        new_files.append(content_file.file_name)
        command_to_execute = "azure-devops-file-create"

    try:
        demisto.executeCommand(command_to_execute, args=file_args)
    except DemistoException as e:
        raise DemistoException(
            f"Failed to execute azure-devops-file-update or azure-devops-file-create commands. Error: {e}, "
        ) from e


def split_yml_file(content_file: ContentFile):  # pragma: no cover
    content_files = []

    if content_file.content_type == "script":
        base_name = content_file.file_name.replace("automation-", "").replace(".yml", "")
    else:
        base_name = content_file.file_name.replace("integration-", "").replace(".yml", "")

    # create the yml file from entry id
    with open(content_file.file_name, "w") as f:
        f.write(content_file.file_text)

    output_capture = io.StringIO()

    # split the yml file
    yml_splitter = YmlSplitter(
        content_file.file_name,
        base_name=base_name,
        output=base_name,  # pylint: disable=E1123
        file_type=content_file.content_type,
        no_pipenv=True,
        no_code_formatting=True,
        no_logging=True,
        no_readme=True,
    )

    script_type = yml_splitter.yml_data.get("type")
    if not script_type:
        script_type = yml_splitter.yml_data.get("script", {}).get("type")

    if script_type == "python":
        script_extention = "py"
    elif script_type == "javascript":
        script_extention = "js"
    elif script_type == "powershell":
        script_extention = "ps1"
    else:
        script_extention = ""
        demisto.debug(f"{script_type=} didn't match any condition. {script_extention=}")

    with redirect_stdout(output_capture), redirect_stderr(output_capture):
        yml_splitter.extract_to_package_format()

    yml_file_path = f"{base_name}/{base_name}.yml"
    script_file_path = f"{base_name}/{base_name}.{script_extention}"
    path_to_file = os.path.join(content_file.path_to_file, base_name)

    # read the py and yml files content
    with open(yml_file_path) as f:
        yml_txt = f.read()

    with open(script_file_path) as f:
        script_txt = f.read()

    # create the yml file
    yml_file = ContentFile()
    yml_file.file_text = yml_txt
    yml_file.file_name = f"{base_name}.yml"
    yml_file.path_to_file = path_to_file
    content_files.append(yml_file)

    # create the script file
    script_file = ContentFile()
    script_file.file_text = script_txt
    script_file.file_name = f"{base_name}.{script_extention}"
    script_file.path_to_file = path_to_file
    content_files.append(script_file)

    # create the description file
    description_file_path = f"{base_name}/{base_name}_description.md"
    if exists(description_file_path):
        with open(description_file_path) as f:
            description_txt = f.read()

        description_file = ContentFile()
        description_file.file_text = description_txt
        description_file.file_name = f"{base_name}_description.md"
        description_file.path_to_file = path_to_file
        content_files.append(description_file)

    return content_files


def commit_git(
    git_integration: str,
    branch_name: str,
    content_file: ContentFile,
    new_files: List,
    modified_files: List,
    keep_placeholders_in_files: bool,
):
    if git_integration == "Gitlab":
        commit_content_item_gitlab(branch_name, content_file, new_files, modified_files)
    elif git_integration == "GitHub":
        commit_content_item(branch_name, content_file, new_files, modified_files, keep_placeholders_in_files)
    elif git_integration == "Bitbucket":
        commit_content_item_bitbucket(branch_name, content_file, new_files, modified_files)
    elif git_integration == "AzureDevOps":
        commit_content_item_azure_devops(branch_name, content_file, new_files, modified_files)
    else:
        raise DemistoException(f"Unexpected {git_integration=}. Possible values: Gitlab, GitHub, Bitbucket and AzureDevOps.")


""" MAIN FUNCTION """


def main():  # pragma: no cover
    try:
        files = demisto.getArg("files")
        branch_name = demisto.getArg("branch")
        pack_name = demisto.getArg("pack")
        user = demisto.getArg("user")
        comment = demisto.getArg("comment")
        template = demisto.getArg("template")
        git_integration = demisto.getArg("git_integration")
        keep_placeholders_in_files = demisto.getArg("cicd_keep_placeholders_in_files") or False
        new_files: List[str] = []
        modified_files: List[str] = []

        if not template:
            template = PR_TEMPLATE

        if not comment:
            comment = ""

        username = user.get("username")
        if user.get("email"):
            username = f'{username} ({user.get("email")})'

        # commit the files from the input
        for file in files:
            if file.get("Unzipped"):
                continue

            # create ContentFile item
            content_file = ContentFile(pack_name=pack_name, file=file)

            if content_file.content_type in ("script", "integration"):
                # split automation file to yml and script files
                content_files = split_yml_file(content_file)
                for file_to_commit in content_files:
                    commit_git(
                        git_integration, branch_name, file_to_commit, new_files, modified_files, keep_placeholders_in_files
                    )

            else:
                commit_git(git_integration, branch_name, content_file, new_files, modified_files, keep_placeholders_in_files)
        incident_url = demisto.demistoUrls().get("investigation")

        # create the PR text
        pr_body = template.format(f"{username}\n", f"{pack_name}\n", f"{branch_name}\n", f"{incident_url}\n", f"{comment}\n")
        if new_files:
            pr_body = f"{pr_body}\n\n### New files\n- "
            pr_body = pr_body + "\n- ".join(new_files)

        if modified_files:
            pr_body = f"{pr_body}\n\n### Modified files\n- "
            pr_body = pr_body + "\n- ".join(modified_files)

        return_results(CommandResults(readable_output=pr_body, outputs_prefix="PR_text", outputs=pr_body))

    except Exception as ex:
        return_error(f"Failed to execute CommitFiles script. Error: {ex}", error=ex)


""" ENTRY POINT """

if __name__ in ("__main__", "__builtin__", "builtins"):
    main()

README