UnzipFile

Unzip a file using fileName or entryID to specify a file. Unzipped files will be loaded to the War Room and names will be put into the context.

Type

python

Pack

CommonScripts

Source

import os
import shutil
import sys
import zipfile as z
from os.path import isdir, isfile
from pathlib import Path
from subprocess import PIPE, Popen
from tempfile import mkdtemp

import demistomock as demisto
from CommonServerPython import *

from CommonServerUserPython import *

MAX_FILENAME_SIZE_BYTES = 255
SLICE_FILENAME_SIZE_BYTES = 235


def get_zip_path(args):
    """
    :param args: arg from demisto
    :return: path of zip file
    """
    file_entry_id = ""
    if args.get("fileName") or args.get("lastZipFileInWarroom"):
        entries = demisto.executeCommand("getEntries", {})
        for entry in entries:
            fn = demisto.get(entry, "File")

            # We check the python version to prevent encoding issues. Effects Demisto 4.5+
            is_text = type(fn) is str  # pylint: disable=E0602

            is_correct_file = args.get("fileName", "").lower() == fn.lower()
            is_zip = fn.lower().endswith(".zip")

            if is_text and is_zip:
                if args.get("fileName") and is_correct_file:
                    file_entry_id = entry["ID"]
                    break
                if args.get("lastZipFileInWarroom"):
                    file_entry_id = entry["ID"]

        # after the for loop above checks if a entry was found
        if not file_entry_id:
            if args.get("fileName"):
                demisto.results(
                    {
                        "Type": entryTypes["error"],
                        "ContentsFormat": formats["text"],
                        "Contents": args.get("fileName", "") + " not such file in war room",
                    }
                )
            if args.get("lastZipFileInWarroom"):
                demisto.results(
                    {"Type": entryTypes["error"], "ContentsFormat": formats["text"], "Contents": "Not found zip file in war room"}
                )
            sys.exit(0)
    if "entryID" in args:
        file_entry_id = args.get("entryID")  # type: ignore

    if not file_entry_id:
        return_error("You must set entryID or fileName or lastZipFileInWarroom=true when executing Unzip script")

    res = demisto.executeCommand("getFilePath", {"id": file_entry_id})
    if res[0]["Type"] == entryTypes["error"]:
        demisto.results(
            {
                "Type": entryTypes["error"],
                "ContentsFormat": formats["text"],
                "Contents": "Failed to get the file path for entry: " + file_entry_id,
            }
        )
        sys.exit(0)

    return res[0]["Contents"]


def extract(file_info, dir_path, zip_tool="7z", password=None):
    """
    :param file_info: The file data.
    :param dir_path: directory that the file will be extract to
    :param zip_tool: The tool to extract files with (7z or zipfile).
    :param password: password if the zip file is encrypted
    :return:
        excluded_dirs: the excluded dirs which are in dir_path
        excluded_files: the excludedfiles which are in dir_path
    """
    # remembering which files and dirs we currently have so we add them later as newly extracted files.
    file_path = file_info["path"]
    file_name = file_info["name"]
    excluded_files = [f for f in os.listdir(".") if isfile(f)]
    excluded_dirs = [d for d in os.listdir(".") if isdir(d)]
    # extracting the zip file
    """
    We check the python version to ensure the docker image contains the necessary packages. 4.5+
    use the new docker image.
    """
    stdout = ""
    if ".rar" in file_name and sys.version_info > (3, 0):
        stdout = extract_using_unrar(file_path, dir_path, password=password)
    elif ".tar" in file_name:
        stdout = extract_using_tarfile(file_path, dir_path, file_name)
    else:
        if zip_tool == "7z":
            stdout = extract_using_7z(file_path, dir_path, password=password)
        elif zip_tool == "zipfile":
            if password:
                password = bytes(password, "utf-8")
            extract_using_zipfile(file_path, dir_path, password=password)
        else:
            return_error(f"There is no zipTool named: {zip_tool}")

    if "Wrong password?" in stdout:
        demisto.debug(stdout)
        return_error("Data Error in encrypted file. Wrong password?")

    return excluded_dirs, excluded_files


def extract_using_unrar(file_path, dir_path, password=None):
    """
    :param file_path: The file path.
    :param dir_path: directory that the file will be extract to
    :param password: password if the zip file is encrypted
    """
    if password:
        cmd = ["unrar", "x", f"-p{password}", file_path, dir_path]
    else:
        cmd = ["unrar", "x", "-p-", file_path, dir_path]
    process = Popen(cmd, stdout=PIPE, stderr=PIPE)
    # process = Popen([cmd], shell=True, stdout=PIPE, stderr=PIPE)
    stdout, stderr = process.communicate()
    stdout = str(stdout)
    if stderr:
        if "Incorrect password" in str(stderr):
            return_error("The .rar file provided requires a password.")
        else:
            return_error(str(stderr))
    return stdout


def extract_using_tarfile(file_path: str, dir_path: str, file_name: str) -> str:
    if ".tar.gz" in file_name:
        cmd = ["tar", "-xzvf", file_path, "-C", dir_path, "--no-same-owner", "--no-same-permissions"]
    elif file_name.endswith(".tar"):
        cmd = ["tar", "-xf", file_path, "-C", dir_path, "--no-same-owner", "--no-same-permissions"]
    else:
        cmd = []
        demisto.debug(f"{file_name=} didn't match any condition. {cmd=}")
    process = Popen(cmd, stdout=PIPE, stderr=PIPE)
    stdout, stderr = process.communicate()
    stdout = str(stdout)
    if stderr:
        demisto.info(str(stderr))
    return stdout


def extract_using_7z(file_path, dir_path, password=None):
    """
    :param file_path: The file path.
    :param dir_path: directory that the file will be extract to
    :param password: password if the zip file is encrypted
    """
    cmd = ["7z", "x", f"-p{password}", f"-o{dir_path}", file_path]
    process = Popen(cmd, stdout=PIPE, stderr=PIPE)
    # process = Popen([cmd], shell=True, stdout=PIPE, stderr=PIPE)
    stdout, stderr = process.communicate()
    stdout = str(stdout)

    if "Errors" in stdout:
        return_error(
            "7z couldn't extract this file - try using zipTool=zipfile\n"
            "If you already tried both zipfile and 7z check that the zip file is valid."
        )
    return stdout


def extract_using_zipfile(file_path, dir_path, password=None):
    """
    :param file_path: The file path.
    :param dir_path: directory that the file will be extract to
    :param password: password if the zip file is encrypted
    """
    try:
        with z.ZipFile(file_path, "r") as given_zip:
            for f in given_zip.filelist:
                full_filename = f.filename
                filename_length = len(full_filename.encode("utf-8"))
                if filename_length > MAX_FILENAME_SIZE_BYTES:
                    file_name_splited = full_filename.rsplit(".", 1)
                    file_name = file_name_splited[0]
                    extension = file_name_splited[1]
                    slice_object = slice(SLICE_FILENAME_SIZE_BYTES)
                    file_name_bytes = file_name.encode("utf-8")[slice_object]
                    # Rename the filename to a shorten filename.
                    f.filename = f"{file_name_bytes.decode('utf-8', errors='ignore')}_shortened_.{extension}"
                    demisto.results(f"The filename {full_filename} is too long - change file name to:\n{f.filename}")
            for filename in given_zip.filelist:
                given_zip.extract(filename, path=dir_path, pwd=password)
        given_zip.close()
    except Exception as e:
        return_error(
            f"zipfile couldn't extract this file - try using zipTool=7z\n"
            f"If you already tried both zipfile and 7z check that the zip file is valid. {e}"
        )


def upload_files(excluded_dirs, excluded_files, dir_path):
    """
    :param excluded_dirs: excluded dirs
    :param excluded_files: excluded files
    :param dir_path: dir path for the files
    :return:
    """
    filenames = []  # type: ignore
    file_entry_id = dir_path
    # recursive call over the file system top down
    for root, directories, files in os.walk(dir_path):
        # removing the previously existing dirs from the search
        directories[:] = [d for d in directories if d not in excluded_dirs]
        for f in files:
            # skipping previously existing files and verifying that the current file is a file and
            # then adding it to the extracted files list
            if f not in excluded_files and isfile(os.path.join(root, f)):
                filenames.append(os.path.join(root, f))

    if len(filenames) == 0:
        demisto.results(
            {"Type": entryTypes["error"], "ContentsFormat": formats["text"], "Contents": "Could not find files in archive"}
        )
    else:
        results = []
        # extracted files can be in sub directories so we save the base names of
        # the files and also the full path of the file
        files_dic = {file_path: os.path.basename(file_path) for file_path in filenames}
        # Filter out symlinks and out-of-tree entries before building outputs
        safe_base = os.path.realpath(dir_path) + os.sep
        safe_files_dic = {}
        for file_path, file_name in files_dic.items():
            real = os.path.realpath(file_path)
            if Path(file_path).is_symlink() or not real.startswith(safe_base):
                demisto.debug(f"Skipping out-of-tree entry: {file_path} -> {real}")
                continue
            safe_files_dic[file_path] = file_name
            with open(file_path, "rb") as _file:
                demisto.results(fileResult(file_name, _file.read()))
        # Build outputs from the filtered (safe) set only
        files_base_names = list(safe_files_dic.values())  # noqa: F812
        results.append(
            {
                "Type": entryTypes["note"],
                "ContentsFormat": formats["json"],
                "Contents": {"extractedFiles": files_base_names},
                "EntryContext": {"ExtractedFiles": files_base_names, 'File(val.EntryID=="' + file_entry_id + '").Unzipped': True},
                "ReadableContentsFormat": formats["markdown"],
                "HumanReadable": tableToMarkdown(
                    "Extracted Files",
                    [{"name": file_name, "path": file_path} for file_path, file_name in safe_files_dic.items()],
                ),
            }
        )

        demisto.results(results)


def get_password(args):
    """
    Get the file's password argument inserted by the user. The password can be inserted either in the sensitive
    argument (named 'password') or nonsensitive argument (named 'nonsensitive_password). This function asserts these
    arguments are used properly and raises an error if both are inserted and have a different value.
    so this
    Args:
        args: script's arguments

    Returns:
        the password given for the file.
    """
    sensitive_password = args.get("password")
    nonsensitive_password = args.get("nonsensitive_password")
    if sensitive_password and nonsensitive_password and sensitive_password != nonsensitive_password:
        raise ValueError("Please use either the password argument or the non_sensitive_password argument, and not both.")

    return sensitive_password or nonsensitive_password


def main():
    dir_path = mkdtemp()
    try:
        args = demisto.args()
        zip_tool = args.get("zipTool", "7z")
        file_info = get_zip_path(args)
        password = get_password(args)
        excluded_dirs, excluded_files = extract(file_info=file_info, dir_path=dir_path, password=password, zip_tool=zip_tool)
        # Remove symlinks before uploading - use pure Python to avoid dependency on 'find' command
        for root, dirs, files_in_dir in os.walk(dir_path):
            for name in files_in_dir + dirs:
                full_path = Path(root) / name
                if full_path.is_symlink():
                    try:
                        full_path.unlink()
                        demisto.debug(f"Removed symlink: {full_path}")
                    except OSError:
                        demisto.debug(f"Failed to remove symlink: {full_path}")
        upload_files(excluded_dirs, excluded_files, dir_path)

    except Exception as e:
        return_error(str(e))
    finally:
        shutil.rmtree(dir_path)


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

README