UpgradeCheck

Check content pack upgrade impacts and display changes. Packs argument is comma separated list of content pack names, or blank for all content packs to assess.

Type

python

Pack

ContentTesting

Source

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401

INDENT = "##### "


def NewVersion(curver):
    parts = curver.split(".")
    parts[2] = str(int(parts[2]) + 1)
    return ".".join(parts)


def NewMinVersion(curver):
    parts = curver.split(".")
    parts[2] = "0"
    parts[1] = str(int(parts[1]) + 1)
    return ".".join(parts)


def NewMajVersion(curver):
    parts = curver.split(".")
    parts[2] = "0"
    parts[1] = "0"
    parts[0] = str(int(parts[0]) + 1)
    return ".".join(parts)


def CleanMarkdown(text):
    return re.sub('["*","#","-", ":"]', "", text)


def ChangeList(text):
    commits = text.split("\n")
    changes = []

    for line in commits:
        if line != "" and line != "- ":
            changes.append(line.strip().replace("- ", "").replace("-", ""))
    return changes


def CheckUpdates(packs, types, playbook, layout):
    pbupdated = False
    loupdated = False
    for p in packs:
        if p in types:
            if types[p]["playbook"] == playbook:
                pbupdated = True
            if types[p]["layout"] == layout:
                loupdated = True

    return pbupdated, loupdated


def GetUpgradedPacks():
    response = execute_command("core-api-get", {"uri": "/contentpacks/installed-expired", "body": ""})["response"]
    upgradePacks = {}
    changesPacks = {}

    for r in response:
        if "updateAvailable" in r and r["updateAvailable"]:
            packid = r["packID"]
            upgradePacks[packid] = r
            changes = ""
            breaking = ""

            newver = NewVersion(r["currentVersion"])
            if newver in r["changelog"]:
                changes = r["changelog"][newver]["releaseNotes"]
                breaking = r["changelog"][newver]["breakingChangesNotes"]
            newver = NewMinVersion(r["currentVersion"])
            if newver in r["changelog"]:
                changes = r["changelog"][newver]["releaseNotes"]
                breaking = r["changelog"][newver]["breakingChangesNotes"]
            newver = NewMajVersion(r["currentVersion"])
            if newver in r["changelog"]:
                changes = r["changelog"][newver]["releaseNotes"]
                breaking = r["changelog"][newver]["breakingChangesNotes"]

            changesPacks[packid] = {"changes": changes, "breaking": breaking}

    return upgradePacks, changesPacks


def FilterPacks(packs, upgradePacks, changesPacks):
    packlist = [p.lower().replace(" ", "") for p in packs.split(",")]
    if len(packlist) == 1 and packlist[0] == "":
        return upgradePacks, changesPacks

    newUpgrade = {}
    for packid, pack in upgradePacks.items():
        if packid.lower().replace(" ", "") in packlist:
            newUpgrade[packid] = pack
        else:
            del changesPacks[packid]

    return newUpgrade, changesPacks


def GetUpgradedIntegrations(packs):
    response = execute_command("core-api-post", {"uri": "/settings/integration/search", "body": {}})["response"]
    configs = response["configurations"]
    instances = response["instances"]
    integrations = {}
    integmap = {}

    for packid, p in packs.items():
        integs = p["contentItems"].get("integration", [])
        if integs is not None:
            for i in integs:
                integmap[i["name"]] = packid

    for c in configs:
        instid = c["id"]
        if c["packName"] == "Palo Alto Networks Cortex XDR - Investigation and Response":
            instid = "Palo Alto Networks Cortex XDR - Investigation and Response"
        if instid in integmap:
            packid = integmap[instid]
            integrations[packid] = {
                "classifier": c.get("defaultClassifier", ""),
                "mapperin": c.get("defaultMapperIn", ""),
                "mapperout": c.get("defaultMapperOut", ""),
                "instance": "",
            }

    for i in instances:
        instid = i["brand"]
        name = i["name"]
        if instid == "Cortex XDR - IR":
            instid = "Palo Alto Networks Cortex XDR - Investigation and Response"
        if instid in integmap:
            packid = integmap[instid]
            integrations[packid]["instance"] = name

    return integrations


def GetCustomPlaybooks():
    response = execute_command(
        "core-api-post", {"uri": "/playbook/search", "body": {"query": "system:F AND hidden:F AND deprecated:F"}}
    )["response"]["playbooks"]
    playbooks = []

    for r in response:
        playbooks.append(r)
    return playbooks


def GetFieldKey(inoutfield):
    if type(inoutfield) is list:
        output = []
        for item in inoutfield:
            if "key" in item:
                output.append(item["key"])
            elif "contextPath" in item:
                output.append(item["contextPath"])
            else:
                output.append(item)
        output = ",".join(output)
    else:
        output = re.sub('["$","{","}"]', "", inoutfield)
    return output


def GetFieldsUsed(playbooks):
    usedfields: dict[str, list[str]]
    usedfields = {}
    regex = re.compile(r"\$\{incident\.[^}]+\}")

    for p in playbooks:
        name = p["name"]
        usedfields[name] = []
        if p["inputs"] is not None:
            usedfields[name].append(GetFieldKey(p["inputs"]))
        if p["outputs"] is not None:
            usedfields[name].append(GetFieldKey(p["outputs"]))

        for _key, t in p["tasks"].items():
            if "scriptArguments" in t:
                for m in regex.findall(json.dumps(t)):
                    usedfields[name].append(GetFieldKey(m))

    return usedfields


def GetSubplaybooksUsed(playbooks):
    usedplaybooks = []
    for p in playbooks:
        for _key, t in p["tasks"].items():
            if t["type"] == "playbook" and "name" in p and "name" in t["task"]:
                usedplaybooks.append({"parent": p["name"], "child": t["task"]["name"]})

    return usedplaybooks


def GetAutomationsUsed(playbooks):
    automations = []
    for p in playbooks:
        if len(p["scriptIds"]) != 0:
            for s in p["scriptIds"]:
                if "name" in p:
                    automations.append({"playbook": p["name"], "scripts": s})

    return automations


def GetUpgradedScripts(packs, scripts):
    scriptdict = {}
    for _index, value in enumerate(scripts):
        scriptdict[value["scripts"]] = value
    upgscripts = []

    for p in packs:
        response = execute_command("core-api-get", {"uri": f"/contentpacks/marketplace/{p}", "body": ""})["response"]
        automations = response["contentItems"]["automation"]

        if automations is not None:
            for a in automations:
                if a["name"] in scriptdict:
                    s = scriptdict[a["name"]]
                    upgscripts.append({"playbook": s["playbook"], "pack": p, "script": s["scripts"]})

    return upgscripts


def GetUpgradedIncidentTypes(packs):
    response = execute_command("core-api-get", {"uri": "/incidenttype", "body": ""})["response"]
    uptypes = {}
    custtypes = []
    for r in response:
        pb = r.get("playbookId", "<none>")
        lo = r.get("layout", "<none>")
        if not r["system"]:
            custype = {"id": r["id"], "playbook": pb, "layout": lo, "packid": "<none>"}
            custtypes.append(custype)
        pi = r.get("packID", "<none>")
        if pi in packs:
            uptypes[pi] = {"playbook": pb, "layout": lo, "packid": pi}
    return (uptypes, custtypes)


def BuildItem(pack, key):
    md = f"{key}:\n"
    items = pack["contentItems"][key]
    if items is not None:
        for i in items:
            if "name" in i:
                md += f"{i['name']}\n"
    if md != f"{key}:\n":
        return md
    return ""


def ImpactMD(upgradePacks, integinstances, types, custypes, subplaybooks, upgradescripts, usedfields):
    md = ""
    md += "#### Custom_Playbooks:\n"
    md += f"{INDENT}Automations_Used:\n"
    md += "|Custom Playbook|Content Pack|Automation|\n"
    md += "|---|---|---|\n"
    for s in upgradescripts:
        md += f"| {s['playbook']} | {s['pack']} | {s['script']} |\n"

    md += f"{INDENT}Subplaybooks_Used:\n"
    md += "|Playbook|Sub-Playbook|\n"
    md += "|---|---|\n"
    for p in subplaybooks:
        md += f"| {p['parent']} | {p['child']} |\n"

    md += f"{INDENT}Fields_Used_Inputs_Outputs:\n"
    md += "|Playbook|Fields|\n"
    md += "|---|---|\n"
    for key, val in usedfields.items():
        for f in val:
            md += f"| {key} | {f} |\n"

    md += "#### Custom_Incident_Types:\n"
    md += "|Incident Type|Object Type|Object|\n"
    md += "|---|---|---|\n"
    for c in custypes:
        if c["id"] != "Unclassified":
            pbupd, loupd = CheckUpdates(upgradePacks, types, c["playbook"], c["layout"])
            if pbupd:
                md += f"| {c['id']} | Playbook | {c['playbook']} |\n"
            if loupd:
                md += f"| {c['id']} | Layout   | {c['layout']}   |\n"

    md += "#### Integration_Instances:\n"
    for _key, i in integinstances.items():
        if i["instance"] != "":
            md += f"{INDENT}Instance: {i['instance']}\n"
            if i["classifier"] != "":
                md += f"Classifier: {i['classifier']}\n"
            if i["mapperin"] != "":
                md += f"Mapperin: {i['mapperin']}\n"
            if i["mapperout"] != "":
                md += f"Mapperout: {i['mapperout']}\n"

    return md


def UpgradeMD(upgradePacks, changes):
    md = ""

    for p in upgradePacks:
        md += f"\n#### {p}:\n"

        newmd = BuildItem(upgradePacks[p], "incidentfield")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        newmd = BuildItem(upgradePacks[p], "integration")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        newmd = BuildItem(upgradePacks[p], "incidenttype")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        newmd = BuildItem(upgradePacks[p], "playbook")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        newmd = BuildItem(upgradePacks[p], "layoutscontainer")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        newmd = BuildItem(upgradePacks[p], "automation")
        if newmd != "":
            md += f"{INDENT} {newmd}"

        md += f"{INDENT}changes:\n"
        chglist = ChangeList(CleanMarkdown(changes[p]["changes"]))
        if chglist is not None:
            for c in chglist:
                md += f"{c}\n"

        breaking = CleanMarkdown(changes[p]["breaking"])
        if breaking.strip() != "":
            md += f"{INDENT}breaking:\n"
            md += f"{breaking}\n"

    return md


def main():
    try:
        packs = demisto.args().get("packs", "")
        upgradePacks, changesPacks = GetUpgradedPacks()
        upgradePacks, changesPacks = FilterPacks(packs, upgradePacks, changesPacks)
        upgradeIntegs = GetUpgradedIntegrations(upgradePacks)
        upgradeTypes, customTypes = GetUpgradedIncidentTypes(upgradePacks)
        playbooks = GetCustomPlaybooks()
        subplaybooks = GetSubplaybooksUsed(playbooks)
        scripts = GetAutomationsUsed(playbooks)
        fields = GetFieldsUsed(playbooks)
        upgradeScripts = GetUpgradedScripts(changesPacks, scripts)

        impact = ImpactMD(upgradePacks, upgradeIntegs, upgradeTypes, customTypes, subplaybooks, upgradeScripts, fields)
        demisto.executeCommand("setIncident", {"customFields": json.dumps({"contenttestingcontentimpacts": impact})})
        details = UpgradeMD(upgradePacks, changesPacks)
        demisto.executeCommand("setIncident", {"customFields": json.dumps({"contenttestingcontentdetails": details})})

    except Exception as ex:
        demisto.error(traceback.format_exc())
        return_error(f"UpgradeCheck: Exception failed to execute. Error: {ex!s}")


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

README