ReadPDFFileV2

Load a PDF file's content and metadata into context. Supports extraction of hashes, urls, and emails when available.

Type

python

Pack

CommonScripts

Source

import contextlib
import glob
import html
import io
import json
import os
import re
import shutil
import stat
import subprocess

import demistomock as demisto
import PyPDF2
from CommonServerPython import *
from pikepdf import PasswordError, Pdf

URL_EXTRACTION_REGEX = (
    r"(?:(?:https?|ftp|hxxps?):\/\/|www\[?\.\]?|ftp\[?\.\]?)(?:[-\w\d]+\[?\.\]?)+"
    r"[-\w\d]+(?::\d+)?(?:(?:\/|\?)[-\w\d+&@#\/%=~_$?!\-:,.\(\);]*[\w\d+&@#\/%=~_$\(\);])?"
)
INTEGRATION_NAME = "ReadPDFFileV2"
DEFAULT_NUM_IMAGES = 20


class PdfPermissionsException(Exception):
    """
    Every exception class that is in charge of catching errors that occur when trying to
    extract data from the PDF must inherit this class
    """


class PdfCopyingProtectedException(PdfPermissionsException):
    """
    This class is in charge of catching errors that occur when we try to extract data from
    a `copy-protected` file (Copy-protected files are files that prevent us from copy its content)
    This is relevant since we run a command that copies the content of the pdf file into a text file.
    """


class PdfInvalidCredentialsException(PdfPermissionsException):
    """
    This class is in charge of catching errors that occur when we try to decrypt an encrypted
    pdf file with the wrong password.
    """


# Error class for shell errors
class ShellException(Exception):
    pass


try:
    ROOT_PATH = os.getcwd()
except OSError:
    return_error(
        "The script failed to access the current working directory. This might happen if your docker isn't "
        "set up correctly. Please contact customer support"
    )

EMAIL_REGXEX = "[a-zA-Z0-9-_.]+@[a-zA-Z0-9-_.]+"
# Documentation claims png is enough for pdftohtml, but through testing we found jpg can be generated as well
IMG_FORMATS = ["jpg", "jpeg", "png", "gif"]


def handle_error_read_only(fun, path, exp) -> None:
    """
    Handling errors that can be encountered in `shutil.rmtree()` execution.
    """
    demisto.debug(exp)

    # Checking if the file is Read-Only
    if not os.access(path, os.W_OK):
        demisto.debug(f"The {path} file is read-only")
        # Change the file permission to the writing
        try:
            os.chmod(path, stat.S_IWUSR)
            fun(path)
        except Exception as e:
            raise ValueError(str(e))
    else:
        raise ValueError(str(exp))


def create_file_instance(entry_id: str, path: str, file_name: str, score: int | None) -> Common.File:
    dbot_score = Common.DBotScore(
        indicator=entry_id,
        indicator_type=DBotScoreType.FILE,
        integration_name="PDFx",
        score=score,
    )
    file = Common.File(
        dbot_score=dbot_score,
        extension="pdf",
        entry_id=entry_id,
        name=file_name,
        path=path,
    )
    return file


def mark_suspicious(suspicious_reason: str, entry_id: str, path: str, file_name: str) -> None:
    """Missing EOF, file may be corrupted or suspicious file"""

    human_readable = f"{suspicious_reason}\nFile marked as suspicious for entry id: {entry_id}"
    file_instance = create_file_instance(
        entry_id=entry_id,
        path=path,
        file_name=file_name,
        score=Common.DBotScore.SUSPICIOUS,
    )
    return_warning(message=human_readable, outputs=file_instance.to_context())


def run_shell_command(command: str, *args) -> bytes:
    """Runs shell command and returns the result if not encountered an error"""
    cmd = [command] + list(args)
    demisto.debug(f"Running the shell command {cmd=}")
    completed_process = subprocess.run(  # noqa: UP022
        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
    exit_codes = completed_process.returncode
    error_string = completed_process.stderr.decode("utf-8")
    demisto.debug(f"Got the following error: {exit_codes=},  {error_string=}")
    if exit_codes != 0:
        if "PDF file is damaged" in error_string or "Couldn't read xref table" in error_string:
            raise ShellException("PDf file is damaged/corrupted.")
        elif "Incorrect password" in error_string:
            raise PdfInvalidCredentialsException("Incorrect password. Please provide the correct password.")
        elif "Copying of text from this document is not allowed" in error_string:
            raise PdfCopyingProtectedException("Copying is not permitted")
        raise ShellException(f"Failed with the following error code: {exit_codes}.\n Error: {error_string}")
    elif error_string:
        demisto.debug(f"ReadPDFFilev2: exec of [{cmd}] completed with warnings: {error_string}")
    return completed_process.stdout


def get_files_names_in_path(path: str, name_of_file: str, full_path: bool = False) -> list:
    """Returns a list[str] of file names in path, will return full path if given full_path=True"""
    os.chdir(ROOT_PATH)
    os.chdir(path)
    res = []
    for file_path in glob.glob(name_of_file):
        if full_path:
            file_path = f"{path}/{file_path}"
        res.append(file_path)
    return res


def get_images_paths_in_path(path: str) -> list[str]:
    """Gets images paths from path"""
    res: list[str] = []
    for img_type in IMG_FORMATS:
        img_format = f"*.{img_type}"
        res.extend(get_files_names_in_path(path, img_format, True))
    return res


def get_pdf_metadata(file_path: str, user_or_owner_password: str | None = None) -> dict:
    """Gets the metadata from the pdf as a dictionary"""
    if user_or_owner_password:
        try:
            demisto.debug("Trying password as user password, using the [upw] flag")
            metadata_txt = run_shell_command("pdfinfo", "-upw", user_or_owner_password, file_path)
        except PdfInvalidCredentialsException:
            demisto.debug("Trying password as owner password, using the [opw] flag")
            metadata_txt = run_shell_command("pdfinfo", "-opw", user_or_owner_password, file_path)
        demisto.debug("PDF file has been successfully opened. Metadata has been retrieved.")
    else:
        metadata_txt = run_shell_command("pdfinfo", "-enc", "UTF-8", file_path)
    metadata = {}
    metadata_str = metadata_txt.decode("utf8", "replace")
    for line in metadata_str.split("\n"):
        # split to [key, value...]
        line_arr = line.split(":")
        if len(line_arr) > 1:
            key = line_arr[0]
            # camelize key
            if " " in key:
                if "PDF" in key:
                    key = key.title().replace("Pdf", "PDF").replace(" ", "")

                else:
                    key = key.title().replace(" ", "")

            # Handle values with and without ':'
            value = ""
            for i in range(1, len(line_arr)):
                value += line_arr[i].strip() + ":"
            # remove redundant ':'
            value = value[:-1]
            metadata[key] = value
    return metadata


def bypass_copy_protected_limitations(pdf_file: str) -> None:
    """
    This function is in charge of handling the situation when a pdf is `copy-protected`.
    Copy protected files prevent us from extracting content from the file, therefore we need a way to bypass this limitation.
    """
    with Pdf.open(pdf_file, allow_overwriting_input=True) as pdf:
        pdf.save(pdf_file)


def get_pdf_text(file_path: str, pdf_text_output_path: str) -> str:
    """Creates a txt file from the pdf in the pdf_text_output_path and returns the content of the txt file"""
    try:
        run_shell_command("pdftotext", file_path, pdf_text_output_path)
    except PdfCopyingProtectedException:
        bypass_copy_protected_limitations(pdf_file=file_path)
        run_shell_command("pdftotext", file_path, pdf_text_output_path)
    text = ""
    with open(pdf_text_output_path, "rb") as f:
        for line in f:
            text += line.decode("utf-8")
    return text


def get_pdf_htmls_content(pdf_path: str, output_folder: str, unescape_url: bool = True) -> str:
    """Creates an html file and images from the pdf in output_folder and returns the text content of the html files"""
    pdf_html_output_path = f"{output_folder}/PDF_html"
    try:
        run_shell_command("pdftohtml", pdf_path, pdf_html_output_path)
    except PdfCopyingProtectedException:
        bypass_copy_protected_limitations(pdf_file=pdf_path)
        run_shell_command("pdftohtml", pdf_path, pdf_html_output_path)
    html_file_names = get_files_names_in_path(output_folder, "*.html")
    html_content = ""
    for file_name in html_file_names:
        with open(file_name, "rb") as f:
            for line in f:
                html_content += html.unescape(str(line)) if unescape_url else str(line)
    return html_content


def build_readpdf_entry_object(
    entry_id: str,
    metadata: dict,
    text: str,
    urls: list,
    emails: list,
    images: list[str],
    max_images: int,
    hash_contexts: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
    """Builds an entry object for the main script flow"""
    pdf_file = {"EntryID": entry_id}
    # Add Text to file entity
    pdf_file["Text"] = text

    # Add Metadata to file entity
    for k in metadata:
        pdf_file[k] = metadata[k]

    md = "### Metadata\n"
    md += "* " if metadata else ""
    md += "\n* ".join([f"{k}: {v}" for k, v in metadata.items()])

    md += "\n### URLs\n"
    md += "* " if urls else ""
    md += "\n* ".join([f'{k["Data"]!s}' for k in urls])

    md += "\n### Text"
    md += f"\n{text}"
    results = [
        {
            "Type": entryTypes["note"],
            "ContentsFormat": formats["markdown"],
            "Contents": md,
            "HumanReadable": md,
            "EntryContext": {"File(val.EntryID == obj.EntryID)": pdf_file, "URL": urls},
        }
    ]
    if images:
        results[0]["HumanReadable"] = f"{results[0]['HumanReadable']}\n### Images"
        os.chdir(ROOT_PATH)
        for i, img in enumerate(images):
            if i >= max_images:
                break
            file = file_result_existing_file(img)
            results.append(file)
    all_pdf_data = ""
    if metadata:
        for _, v in metadata.items():
            all_pdf_data += str(v)
    if text:
        all_pdf_data += text
    if urls:
        for u in urls:
            u = u["Data"] + " "
            all_pdf_data += u

    # Extract indicators (omitting context output, letting auto-extract work)
    try:
        indicators_map = demisto.executeCommand("extractIndicators", {"text": all_pdf_data})[0]["Contents"]
        indicators_map = json.loads(indicators_map)
        if emails:
            indicators_map["Email"] = emails
        if hash_contexts:
            indicators_map["Hashes"] = hash_contexts
    except json.JSONDecodeError:
        pass
    ec = build_readpdf_entry_context(indicators_map)
    results.append(
        {
            "Type": entryTypes["note"],
            "ContentsFormat": formats["json"],
            "Contents": indicators_map,
            "HumanReadable": indicators_map,
            "EntryContext": ec,
        }
    )
    return results


def build_readpdf_entry_context(indicators_map: Any) -> dict:
    ec = {}
    if isinstance(indicators_map, dict):
        if "URL" in indicators_map:
            ec_url = []
            for url in indicators_map["URL"]:
                ec_url.append({"Data": url})
            ec["URL"] = ec_url
        if "Email" in indicators_map:
            ec_email = []
            for email in indicators_map["Email"]:
                ec_email.append({"Email": email})
            ec["Account"] = ec_email
        if "Hashes" in indicators_map:
            ec["Hashes"] = indicators_map["Hashes"]
    return ec


def get_urls_from_binary_file(file_path: str) -> set:
    """Reading from the binary pdf in the pdf_text_output_path and returns a list of the urls in the file"""
    with open(file_path, "rb") as file:
        # the urls usually appear in the form: '/URI (url)'
        urls = re.findall(r"/URI ?\((.*?)\)", str(file.read()))
    binary_file_urls = set()
    # make sure the urls match the url regex
    for url in urls:
        mached_url = re.findall(URL_EXTRACTION_REGEX, url)
        if len(mached_url) != 0:
            binary_file_urls.add(mached_url[0])
    return binary_file_urls


def get_urls_and_emails_from_pdf_html_content(
    cpy_file_path: str, output_folder: str, unescape_url: bool = True
) -> tuple[set, set]:
    """
    Extract the URLs and emails from the pdf html content.

    Args:
        cpy_file_path (str): the path of the PDF file.
        output_folder (str): the folder output to get the HTML files from.
    Returns:
        tuple[set, set]: The URLs and emails that were found.
    """
    pdf_html_content = get_pdf_htmls_content(cpy_file_path, output_folder, unescape_url)
    return set(re.findall(URL_EXTRACTION_REGEX, pdf_html_content)), set(re.findall(EMAIL_REGXEX, pdf_html_content))


def extract_url_from_annot_object(annot_object: Any):
    """
    Extracts the URLs from the Annot object (under key: '/A').

    Args:
        annot_object (PyPDF2.generic.DictionaryObject): An object contains annotations of a PDF.

    Returns:
         (PyPDF2.generic.TextStringObject): The extracted url if exists, else - None.

    """

    # Extracts the URLs from the Annot object (under key: '/A'):
    if a := annot_object.get("/A"):
        if isinstance(a, PyPDF2.generic.IndirectObject):
            a = a.get_object()

        if url := a.get("/URI"):
            if isinstance(url, PyPDF2.generic.IndirectObject):
                url = url.get_object()
            return url
        return None
    return None


def extract_url(extracted_object: Any):
    """
    Extracts URL (if exists) from the extracted object, according to the URL_EXTRACTION_REGEX.

    Args:
        extracted_object (PyPDF2.generic.TextStringObject): A TextStringObject object contains a url or an email.

    Returns:
         (str): The extracted url.
    """
    match = ""
    matched_url = re.findall(URL_EXTRACTION_REGEX, extracted_object)
    if len(matched_url) != 0:
        match = matched_url[0]

    return match


def extract_email(extracted_object: Any):
    """
    Extracts Email (if exists) from the extracted object, according to the EMAIL_REGXEX.

    Args:
        extracted_object (PyPDF2.generic.TextStringObject): A TextStringObject object contains a url or an email.

    Returns:
         (str): The extracted email.
    """
    match = ""
    matched_email = re.findall(EMAIL_REGXEX, extracted_object)
    if len(matched_email) != 0:
        match = matched_email[0]

    return match


def extract_urls_and_emails_from_annot_objects(annot_objects: list | Any):
    """
    Extracts URLs and Emails from the Annot objects, and separate them into two different sets.

    Args:
        annot_objects (List): A list of objects that contain annotations of a PDF.

    Returns:
         Tuple[set, set]: A set includes the extracted urls, A set includes the extracted emails.

    """

    urls = set()
    emails = set()

    for annot_object in annot_objects:
        if isinstance(annot_object, PyPDF2.generic.IndirectObject):
            try:
                annot_object = annot_object.get_object()
            except Exception as e:
                if "Could not find object" in str(e):
                    demisto.error(f"annot.get_object() encountered an error: {e}.\n Skipping without failure.")
                    continue
                else:
                    demisto.error(f"annot.get_object() encountered an error: {e}.")

        extracted_object = extract_url_from_annot_object(annot_object)
        # Separates URLs and Emails:
        if extracted_object:
            if isinstance(extracted_object, bytes):
                extracted_object = extracted_object.decode()
            if url := extract_url(extracted_object):
                urls.add(url)
            if email := extract_email(extracted_object):
                emails.add(email)

    return urls, emails


def get_urls_and_emails_from_pdf_annots(file_path: str) -> tuple[set, set]:
    """
    Extracts the URLs and Emails from the pdf's Annots (Annotations and Commenting) using PyPDF2 package.
    Args:
        file_path (str): The path of the PDF file.
    Returns:
        Tuple[set, set]: A set includes the URLs that were found, A set includes the Emails that were found.
    """
    all_urls: set[str] = set()
    all_emails: set[str] = set()
    output_capture = io.StringIO()
    with open(file_path, "rb") as pdf_file:
        # The following context manager was added so we could redirect error messages to the server logs since
        # PyPDF2 would sometimes return warnings on some files (warnings and not errors because strict=False), and these warnings
        # would be flushed to stderr, and therefore they would be returned as an error message to the user instead of being
        # flushed to the server logs.
        with contextlib.redirect_stderr(output_capture):
            pdf = PyPDF2.PdfReader(pdf_file, strict=False)
            pages_len = len(pdf.pages)

            # Goes over the PDF, page by page, and extracts urls and emails:
            for page in range(pages_len):
                page_sliced = pdf.pages[page]
                page_object = page_sliced.get_object()

                # Extracts the PDF's Annots (Annotations and Commenting):
                if annots := page_object.get("/Annots"):  # type: ignore[union-attr]
                    if not isinstance(annots, PyPDF2.generic.ArrayObject):
                        annots = [annots]

                    for annot in annots:
                        annot_objects = annot.get_object()
                        if not isinstance(annot_objects, PyPDF2.generic.ArrayObject):
                            annot_objects = [annot_objects]

                        # Extracts URLs and Emails:
                        urls_set, emails_set = extract_urls_and_emails_from_annot_objects(annot_objects)
                        all_urls = all_urls.union(urls_set)
                        all_emails = all_emails.union(emails_set)
        demisto.debug(output_capture.getvalue())
    # Logging:
    if len(all_urls) == 0:
        demisto.debug("No URLs were extracted from the PDF.")
    if len(all_emails) == 0:
        demisto.debug("No Emails were extracted from the PDF.")

    return all_urls, all_emails


def extract_urls_and_emails_from_pdf_file(file_path: str, output_folder: str, unescape_url: bool = True) -> tuple[list, list]:
    """
    Extract URLs and Emails from the PDF file.
    Args:
        file_path (str): The path of the PDF file.
        output_folder (str): The output folder for html files.
    Returns:
        tuple[set, set]: A set including the URLs and emails that were found, A set including only emails that were
         extracted from the html content.
    """

    # Get urls from the binary file:
    binary_file_urls = get_urls_from_binary_file(file_path)

    # Get URLS + emails:
    annots_urls, annots_emails = get_urls_and_emails_from_pdf_annots(file_path)
    html_urls, html_emails = get_urls_and_emails_from_pdf_html_content(file_path, output_folder, unescape_url)

    # This url might be generated with the pdf html file, if so, we remove it
    html_urls.discard("http://www.w3.org/1999/xhtml")

    # Unify urls:
    urls_set = annots_urls.union(html_urls, binary_file_urls)
    emails_set = annots_emails.union(html_emails)

    urls_ec = []
    emails_ec = []
    for url in urls_set:
        urls_ec.append({"Data": url})
    for email in emails_set:
        emails_ec.append(email)

    return urls_ec, emails_ec


def extract_hash_contexts_from_pdf_file(file_text: str) -> list[dict[str, Any]]:
    """Extracts the hashes from the file's text, and converts them to hash contexts.

    Args:
        file_text (str): The text extracted from the PDF.

    Returns:
        list[dict[str, Any]]: A list of hash contexts.
    """
    hash_contexts: list[dict[str, Any]] = []
    hashes_in_file = get_hashes_from_file(file_text)
    for hash_type, hashes in hashes_in_file.items():
        if hashes:
            hash_contexts.extend(convert_hash_to_context(hash_type, hashes))
    return hash_contexts


def convert_hash_to_context(hash_type: str, hashes: set[Any]) -> list[dict[str, Any]]:
    """Converts the given hashes to hash contexts

    Args:
        hash_type (str): The hash type of the given hashes.
        hashes (set[Any]): The set of hashes.

    Returns:
        list[dict[str, Any]]: A list of hash contexts that have the same hash type.
    """
    hash_context: list[dict[str, Any]] = [{"type": hash_type, "value": hash} for hash in hashes]
    return hash_context


def get_hashes_from_file(file_text: str) -> dict[str, set[Any]]:
    """Extracts all the hashes found in the file's text.

    Args:
        file_text (str): The file's text.

    Returns:
        dict[str, set[Any]]: A dictionary that holds the hash types as keys, and each key
        holds the set of hashes corresponding to that hash type.
    """
    demisto.debug("Extracting hashes from file")
    hashes: dict[str, set[Any]] = {}
    hashes["SHA1"] = set(re.findall(sha1Regex, file_text))
    hashes["SHA256"] = set(re.findall(sha256Regex, file_text))
    hashes["SHA512"] = set(re.findall(sha512Regex, file_text))
    hashes["MD5"] = set(re.findall(md5Regex, file_text))

    return hashes


def handling_pdf_credentials(cpy_file_path: str, dec_file_path: str, encrypted: str = "", user_password: str = "") -> str:
    """
    This function decrypts the pdf if needed.
    """
    try:
        if user_password or "yes" in encrypted:
            with Pdf.open(cpy_file_path, allow_overwriting_input=True, password=user_password) as pdf:
                pdf.save(dec_file_path)
                return dec_file_path
    except PasswordError:
        raise PdfInvalidCredentialsException("Incorrect password. Please provide the correct password.")
    return cpy_file_path


def extract_data_from_pdf(
    path: str, user_password: str, entry_id: str, max_images: int | None, working_dir: str, unescape_url: bool = True
) -> None:
    max_images = max_images if max_images else DEFAULT_NUM_IMAGES
    if path:
        cpy_file_path = f"{working_dir}/WorkingReadPDF.pdf"
        shutil.copy(path, cpy_file_path)
        metadata = get_pdf_metadata(path, user_password)
        encrypted = metadata.get("Encrypted", "")

        cpy_file_path = handling_pdf_credentials(
            cpy_file_path=cpy_file_path,
            dec_file_path=f"{working_dir}/DecWorkingReadPDF.pdf",
            encrypted=encrypted,
            user_password=user_password,
        )

        # Get text:
        pdf_text_output_path = f"{working_dir}/PDFText.txt"
        text = get_pdf_text(cpy_file_path, pdf_text_output_path)

        # Get hash contexts
        hash_contexts = extract_hash_contexts_from_pdf_file(text)

        # Get URLS + emails:
        urls_ec, emails_ec = extract_urls_and_emails_from_pdf_file(cpy_file_path, working_dir, unescape_url)

        # Get images:
        images = get_images_paths_in_path(working_dir)
        readpdf_entry_object = build_readpdf_entry_object(
            entry_id, metadata, text, urls_ec, emails_ec, images, max_images=max_images, hash_contexts=hash_contexts
        )

        return_results(readpdf_entry_object)
    else:
        raise Exception(f"EntryID {entry_id} path could not be found")


def main():  # pragma: no cover
    args = demisto.args()
    unescape_url: bool = argToBoolean(args.get("unescape_url", "true"))
    working_dir = "ReadPDFTemp"
    try:
        if not os.path.exists(working_dir):
            """ Check if the working directory does not exist and create it """
            os.makedirs(working_dir)
        entry_id = args.get("entryID")
        user_password = str(args.get("userPassword", ""))
        max_images = arg_to_number(args.get("maxImages", None))
        path = demisto.getFilePath(entry_id).get("path")

        extract_data_from_pdf(
            path=path,
            user_password=user_password,
            entry_id=entry_id,
            max_images=max_images,
            working_dir=working_dir,
            unescape_url=unescape_url,
        )
    except PdfPermissionsException as e:
        return_warning(str(e))
    except ShellException as e:
        file_name = demisto.getFilePath(entry_id).get("name")
        mark_suspicious(
            suspicious_reason=f"The script {INTEGRATION_NAME} failed due to an error\n{e!s}",
            entry_id=entry_id,
            path=path,
            file_name=file_name,
        )
    except Exception as e:
        demisto.error(traceback.format_exc())
        return_error(str(e))
    finally:
        os.chdir(ROOT_PATH)
        shutil.rmtree(working_dir, onerror=handle_error_read_only)


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

README