ZeroFoxKeyIncidents

Cloud-based SaaS to detect risks found on social media and digital channels.

Data Enrichment & Threat Intelligence · ZeroFox

Details

ID	`ZeroFoxKeyIncidents`
Provider	Haveli Investments
Category	Data Enrichment & Threat Intelligence
From Version	6.1.0
Docker Image	`demisto/python3:3.12.13.10116658`
Supported Modules	Agentix XSIAM

README

Cloud-based SaaS to detect risks found on social media and digital channels.
This integration was integrated and tested with version 1.4.0 of ZeroFoxKey.

Configure ZeroFox Key Incidents in Cortex

Parameter	Required
URL (e.g., https://api.zerofox.com/)	True
Fetch incidents	False
Username	True
Password	True
First fetch timestamp (<number> <time unit>, e.g., 12 hours, 7 days)	False
Incident type	False

Commands

You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

zerofox-get-key-incident-attachment

Fetches a Key Incident Attachment by ID and uploads it to the current investigation War Room.

Base Command

zerofox-get-key-incident-attachment

Input

Argument Name	Description	Required
attachment_id	The ID of the Key Incident Attachment.	Required

Context Output

Path	Type	Description
File.Size	Number	The size of the file.
File.SHA1	String	The SHA1 hash of the file.
File.SHA256	String	The SHA256 hash of the file.
File.SHA512	String	The SHA512 hash of the file.
File.Name	String	The name of the file.
File.SSDeep	String	The SSDeep hash of the file.
File.EntryID	String	The entry ID of the file.
File.Info	String	File information.
File.Type	String	The file type.
File.MD5	String	The MD5 hash of the file.
File.Extension	String	The file extension.

Incident Mirroring

You can enable incident mirroring between Cortex XSOAR incidents and ZeroFox Key Incidents corresponding events (available from Cortex XSOAR version 6.0.0).
To set up the mirroring:

Enable Fetching incidents in your instance configuration.

Newly fetched incidents will be mirrored in the chosen direction. However, this selection does not affect existing incidents.
Important Note: To ensure the mirroring works as expected, mappers are required, both for incoming and outgoing, to map the expected fields in Cortex XSOAR and ZeroFox Key Incidents.

Configuration parameters

url — URL (e.g., https://api.zerofox.com/) (required)
isFetch — Fetch incidents
credentials — Username (required)
first_fetch — First fetch timestamp (<number> <time unit>, e.g., 12 hours, 7 days)
incidentType — Incident type
incidentFetchInterval — Incidents Fetch Interval
max_fetch — Maximum number of incidents per fetch
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings

Commands (1)

zerofox-get-key-incident-attachment
Fetches a Key Incident Attachment by ID and uploads it to the current investigation War Room.

Source

ZeroFoxKeyIncidents.py ZeroFoxKeyIncidents_test.py ZeroFoxKeyIncidents.yml ZeroFoxKeyIncidents_description.md

Download History Raw

import json
from pathlib import Path
from urllib.parse import urlencode

import freezegun
import pytest
from ZeroFoxKeyIncidents import (
    KeyIncident,
    ZeroFox,
    ZeroFoxKIAttachmentNotFoundException,
    ZeroFoxInternalException,
    demisto,
    fetch_incidents,
    get_key_incident_attachment_command,
    map_key_incident_to_xsoar,
)

BASE_URL = "https://api.zerofox.com"
OK_CODES = (200,)
TOKEN = "token"

KEY_INCIDENTS_ENDPOINT = "/cti/key-incidents/"
KEY_INCIDENT_ATTACHMENTS_ENDPOINT = "/cti/key-incident-attachment"
CTI_TOKEN_ENDPOINT = "/auth/token/"


@pytest.fixture
def zerofox() -> ZeroFox:
    return ZeroFox(
        base_url=BASE_URL,
        ok_codes=OK_CODES,
        username="",
        token=TOKEN,
    )


def load_json(file: str):
    with open(file) as f:
        return json.load(f)


def util_load_json(path):
    with open(path, encoding="utf-8") as f:
        return json.loads(f.read())


def test_get_key_incidents(requests_mock, zerofox):
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    start_time = "2022-05-24"
    end_time = "2022-05-25"
    first_page_response = load_json(
        "test_data/key_incidents/key_incidents_response_first_page.json",
    )
    first_url_params = urlencode(
        {"Tags": "Key Incident", "updated_after": start_time, "updated_before": end_time, "ordering": "updated"}
    )
    requests_mock.get(
        f"{KEY_INCIDENTS_ENDPOINT}?{first_url_params}",
        json=first_page_response,
    )
    second_page_response = load_json("test_data/key_incidents/key_incidents_response_second_page.json")
    second_url_params = urlencode(
        {
            "Tags": "Key Incident",
            "updated_after": start_time,
            "updated_before": end_time,
            "ordering": "updated",
            "cursor": "nextPageCursor",
        }
    )
    requests_mock.get(
        f"{KEY_INCIDENTS_ENDPOINT}?{second_url_params}",
        json=second_page_response,
    )

    expected = [
        KeyIncident.from_dict(ki) for ki in load_json("test_data/key_incidents/parsed_key_incidents.json").get("key_incidents")
    ]
    ki = zerofox.get_key_incidents(start_time=start_time, end_time=end_time)

    assert len(ki) == len(expected)
    assert sorted(ki, key=lambda x: x.incident_id) == sorted(expected, key=lambda x: x.incident_id)


def test_get_key_incident_attachment(requests_mock, zerofox, mocker):
    ATTACHMENT_ID = 123
    expected = load_json("test_data/key_incident_attachments/parsed_ki_attachment.json")
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(
        f"{KEY_INCIDENT_ATTACHMENTS_ENDPOINT}/{ATTACHMENT_ID}/",
        json=load_json("test_data/key_incident_attachments/ki_attachment.json"),
    )
    attachment = zerofox.get_key_incident_attachment(ATTACHMENT_ID)

    assert expected == attachment.to_dict()


def test_key_incident_attachment_not_found(requests_mock, zerofox):
    ATTACHMENT_ID = 123
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(
        f"{KEY_INCIDENT_ATTACHMENTS_ENDPOINT}/{ATTACHMENT_ID}/",
        status_code=404,
    )
    with pytest.raises(ZeroFoxKIAttachmentNotFoundException):
        zerofox.get_key_incident_attachment(ATTACHMENT_ID)


def test_create_xsoar_incidents():
    ki_list = [
        KeyIncident.from_dict(ki) for ki in load_json("test_data/key_incidents/parsed_key_incidents.json").get("key_incidents")
    ]

    xsoar_incidents = []
    for ki in ki_list:
        xsoar_incidents.append(map_key_incident_to_xsoar(ki))

    expected_names = [ki.incident_id + " " + ki.headline for ki in ki_list]
    actual_names = [incident.name for incident in xsoar_incidents]

    assert expected_names == actual_names
    assert len(xsoar_incidents) == 8


@freezegun.freeze_time("2022-05-25")
def test_fetch_incidents_zerofox_error(requests_mock, zerofox):
    start_time = "2022-05-24T00:00:00"
    end_time = "2022-05-25T00:00:00"
    last_run = {"time": start_time}
    url_params = urlencode(
        {"Tags": "Key Incident", "updated_after": start_time, "updated_before": end_time, "ordering": "updated"}
    )
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(f"{KEY_INCIDENTS_ENDPOINT}?{url_params}", status_code=500, json={"error": "test error"})
    first_fetch_time = ""
    with pytest.raises(ZeroFoxInternalException):
        fetch_incidents(zerofox, last_run, first_fetch_time)


@freezegun.freeze_time("2022-05-25")
def test_fetch_incidents_no_incidents(requests_mock, zerofox):
    start_time = "2022-05-24T00:00:00"
    end_time = "2022-05-25T00:00:00"
    last_run = {"time": start_time}
    url_params = urlencode(
        {"Tags": "Key Incident", "updated_after": start_time, "updated_before": end_time, "ordering": "updated"}
    )
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(f"{KEY_INCIDENTS_ENDPOINT}?{url_params}", json={"next": None, "results": []})
    first_fetch_time = ""
    last_run, incidents = fetch_incidents(zerofox, last_run, first_fetch_time)
    assert last_run == {"time": "2022-05-25T00:00:00"}
    assert incidents == []


@freezegun.freeze_time("2022-05-25")
def test_fetch_incidents(requests_mock, zerofox):
    requests_mock.post("/auth/token/", json={"access": "token"})
    start_time = "2022-05-24T00:00:00"
    end_time = "2022-05-25T00:00:00"
    first_page_response = load_json(
        "test_data/key_incidents/key_incidents_response_first_page.json",
    )
    first_url_params = urlencode(
        {"Tags": "Key Incident", "updated_after": start_time, "updated_before": end_time, "ordering": "updated"}
    )
    requests_mock.get(
        f"{KEY_INCIDENTS_ENDPOINT}?{first_url_params}",
        json=first_page_response,
    )
    second_page_response = load_json("test_data/key_incidents/key_incidents_response_second_page.json")
    second_url_params = urlencode(
        {
            "Tags": "Key Incident",
            "updated_after": start_time,
            "updated_before": end_time,
            "ordering": "updated",
            "cursor": "nextPageCursor",
        }
    )
    requests_mock.get(
        f"{KEY_INCIDENTS_ENDPOINT}?{second_url_params}",
        json=second_page_response,
    )

    last_run = {"time": start_time}
    first_fetch_time = ""

    expected_time = "2022-05-20T18:49:20.917000+00:00"

    last_run, incidents = fetch_incidents(zerofox, last_run, first_fetch_time)

    assert last_run == {"time": expected_time}

    expected_results = first_page_response.get("results", []) + second_page_response.get("results", [])
    for index, ki_expected in enumerate(expected_results):
        assert incidents[index].get("name") == f"{ki_expected.get('incident_id')} {ki_expected.get('headline')}"
        assert incidents[index].get("occurred") == ki_expected.get("created_at").replace("Z", "+00:00")
        assert incidents[index].get("dbotMirrorId") == ki_expected.get("incident_id")


@pytest.fixture
def mock_file_id(mocker):
    file_id = "dummyId"
    patcher = mocker.patch.object(demisto, "uniqueFile", return_value=file_id)

    yield file_id

    patcher.stop()
    try:
        file_path = Path(f"1_{file_id}")
        if file_path.exists():
            file_path.unlink()
    except Exception:
        pass


def test_get_key_incident_attachment_command(requests_mock, zerofox, mock_file_id):
    """
    Given
        A Key Incident Attachment Id
    When
        Calling get_key_incident_attachment_command
    Then
        It should return a file with attachment contents
    """
    attachment_id = 123
    ki_attachment = load_json("test_data/key_incident_attachments/ki_attachment.json")
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(
        f"/cti/key-incident-attachment/{attachment_id}/",
        json=ki_attachment,
    )
    args = {"attachment_id": attachment_id}

    results = get_key_incident_attachment_command(zerofox, args)

    expected = {
        "Contents": "",
        "ContentsFormat": "text",
        "File": ki_attachment.get("name"),
        "FileID": mock_file_id,
        "Type": 3,
    }

    assert results == expected


def test_key_incident_attachment_command_not_found(requests_mock, zerofox):
    """
    Given
        A Key Incident Attachment Id
    When
        Calling get_key_incident_attachment_command
    Then
        It should return a file with attachment contents
    """
    attachment_id = 123
    requests_mock.post(CTI_TOKEN_ENDPOINT, json={"access": "token"})
    requests_mock.get(
        f"/cti/key-incident-attachment/{attachment_id}/",
        status_code=404,
    )
    args = {"attachment_id": attachment_id}

    result = get_key_incident_attachment_command(zerofox, args)

    assert result.readable_output == f"Key Incident attachment {attachment_id} was not found"