RecordedFutureASI

Attack Surface Intelligence Risk Rules help security teams take risk and vulnerability prioritization to the next level by helping organizations identify the biggest weaknesses within their attack surface in mere seconds.

Data Enrichment & Threat Intelligence · Recorded Future Attack Surface Intelligence

Details

IDRecordedFutureASI
ProviderMastercard
CategoryData Enrichment & Threat Intelligence
From Version6.5.0
Docker Imagedemisto/python3:3.12.13.10116658
Supported ModulesAgentix XSIAM

README

What does this pack do?

This pack enables security teams to

  • Access a unified risk management from the most popular SOAR platform.
  • Visualize to the most critical risks within your organization
  • Identify security incidents filtered by severity (critical, medium and low)
  • See the full context of the incident, including CVE id, name, description, and affected hostnames.

Configure RecordedFutureASI in Cortex

Get your Project ID

  • Log in to SecurityTrails SurfaceBrowser
  • Go to the Projects page by clicking the Projects link in the top right
  • Click on the Project that you want to use in XSOAR
  • Copy the ID from the URL (looks like c1234567-c123-4123-9123-0123456789ab)

Get your API Key

  • Log in to SecurityTrails SurfaceBrowser
  • Click the username in the top right corner
  • Click on Account
  • Go to API > API Keys
  • Create a new API key with a note that it is being used for the XSOAR Integration

Setting up the Integration

Parameter Required
API Key False
Project ID True
Min Severity False
Issue Grouping False
Expand Issues False
Fetch incidents False
Incidents Fetch Interval False
Incident type False
First fetch timestamp (<number> <time unit>, e.g., 12 hours, 7 days, 3 months, 1 year) False
Max Fetch 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.

asi-project-issues-fetch

Fetches all the current or added issues.

Base Command

asi-project-issues-fetch

Input

Argument Name Description Required
issues_start Timestamp to get added issues after Optional
group_by_host Whether to group results by host Optional
expand_issues Whether to expand grouped host issues by each issue Optional

Context Output

There is no context output for this command.

Configuration parameters

  • isFetch — Fetch incidents
  • apikey — API Key
  • credentials — API Key
  • project_id — Project ID (required)
  • incidentType — Incident type
  • incidentFetchInterval — Incidents Fetch Interval
  • min_severity — Minimum severity of alerts to fetch
  • issue_grouping — How to group new issues
  • expand_issues — Expand grouped By Host rules into separate Incidents (applicable if grouping By Host)
  • max_fetch — Fetch limit
  • first_fetch — First fetch time (<number> <time unit>, e.g., 12 hours, 7 days, 3 months, 1 year)

Commands (1)

  • asi-project-issues-fetch

    Gets the issues for a project from a particular snapshot (defaults to recent).

Source

RecordedFutureASI.py RecordedFutureASI_test.py RecordedFutureASI.yml RecordedFutureASI_description.md command_examples.txt
Download History Raw 
import json

import demistomock as demisto
import pytest
import RecordedFutureASI
from CommonServerPython import IncidentSeverity
from RecordedFutureASI import Client, fetch_incidents

TEST_PROJECT_ID = "fakeprojectid"


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


@pytest.fixture()
def client():
    client = Client(
        base_url="https://api.securitytrails.com/v1/asi",
        project_id=TEST_PROJECT_ID,
        verify=True,
        min_severity="Informational",
        host_incident_limit=10,
        headers={"APIKEY": "key"},
    )
    return client


def test_test_module_valid(requests_mock, client):
    """
    Tests that the test command correctly attempts to send a request
    """
    from RecordedFutureASI import test_module

    requests_mock.get(
        url=f"https://api.securitytrails.com/v1/asi/rules/{TEST_PROJECT_ID}/recent/issues", status_code=200, json="{}"
    )

    assert test_module(client) == "ok"


def test_current_issues_command(requests_mock, client):
    """
    Test !asi-project-issues-fetch correctly gets issues from the current issues endpoint and returns
    incidents
    """
    mock_response = util_load_json("test_data/current-issues.json")
    requests_mock.get(f"https://api.securitytrails.com/v1/asi/rules/{TEST_PROJECT_ID}/recent/issues", json=mock_response)
    last_run, incidents = fetch_incidents(client, {}, False, False)
    assert len(incidents) == 3
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"] == mock_response["data"][0]["name"]
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[1]["name"] == mock_response["data"][1]["name"]
    assert incidents[2]["severity"] == IncidentSeverity.LOW
    assert incidents[2]["name"] == mock_response["data"][2]["name"]


def test_added_issues_command(requests_mock, client):
    """
    Test !asi-project-issues-fetch issues_start=1646769704 correctly uses the activity endpoint and returns
    added incidents
    """
    last_run = 1234
    mock_response = util_load_json("test_data/added-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity?rule_action=added&start={last_run}",
        json=mock_response,
    )
    last_run, incidents = fetch_incidents(client, {"last_fetch": last_run}, False, False)
    assert len(incidents) == 3
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"] == mock_response["data"][0]["added_rules"][0]["name"]
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[1]["name"] == mock_response["data"][0]["added_rules"][1]["name"]
    assert incidents[2]["severity"] == IncidentSeverity.LOW
    assert incidents[2]["name"] == mock_response["data"][0]["added_rules"][2]["name"]


def test_min_severity_filtering(requests_mock, client):
    """
    Test that a min_severity of Moderate correctly filters out informational rules
    """
    mock_response = util_load_json("test_data/current-issues.json")
    requests_mock.get(f"https://api.securitytrails.com/v1/asi/rules/{TEST_PROJECT_ID}/recent/issues", json=mock_response)
    client.min_severity = "Moderate"
    last_run, incidents = fetch_incidents(client, {}, False, False)
    assert len(incidents) == 2
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"] == mock_response["data"][0]["name"]
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[1]["name"] == mock_response["data"][1]["name"]


def test_incident_count_limit(requests_mock, client):
    """
    Test that a setting a max incident number to 2 returns 2 incidents with the last one being
    a warning that too many incidents were available
    """
    mock_response = util_load_json("test_data/current-issues.json")
    requests_mock.get(f"https://api.securitytrails.com/v1/asi/rules/{TEST_PROJECT_ID}/recent/issues", json=mock_response)
    client.host_incident_limit = 2
    last_run, incidents = fetch_incidents(client, {}, False, False)
    assert len(incidents) == 2
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"] == mock_response["data"][0]["name"]
    assert incidents[1]["severity"] == IncidentSeverity.LOW
    assert incidents[1]["name"] == "❗Attack Surface Intelligence: 3+ Changes"


def test_incident_by_host_recent(requests_mock, client):
    """
    Test that By Host issues returns nothing if last_fetch is past any acans
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    last_run, incidents = fetch_incidents(client, {"last_fetch": 99999999999}, True, False)
    assert len(incidents) == 0


def test_incident_by_host(requests_mock, client):
    """
    Test that By Host issues are loaded correctly
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, False)
    assert len(incidents) == 3
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"] == "Attack Surface Risk Increase: registration.example.com (0 --> 95)"
    assert len(json.loads(incidents[0]["rawJSON"])["rules"]) == 1
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[1]["name"] == "Attack Surface Risk Increase: ip.example.com (0 --> 65)"
    assert len(json.loads(incidents[1]["rawJSON"])["rules"]) == 1
    assert incidents[2]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[2]["name"] == "Attack Surface Risk Increase: stage.example.com (20 --> 26)"
    assert len(json.loads(incidents[2]["rawJSON"])["rules"]) == 2


def test_incident_by_host_partial_filter(requests_mock, client):
    """
    Test that By Host issues filter severity correctly and can eliminate a subset of rules from a host
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    client.min_severity = "Moderate"
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, False)
    assert len(incidents) == 3
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert len(json.loads(incidents[0]["rawJSON"])["rules"]) == 1
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert len(json.loads(incidents[1]["rawJSON"])["rules"]) == 1
    assert incidents[2]["severity"] == IncidentSeverity.MEDIUM
    assert len(json.loads(incidents[2]["rawJSON"])["rules"]) == 1


def test_incident_by_host_full_filter(requests_mock, client):
    """
    Test that By Host issues filter severity correctly and eliminates hosts entirely
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    client.min_severity = "Critical"
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, False)
    assert len(incidents) == 1
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert len(json.loads(incidents[0]["rawJSON"])["rules"]) == 1


def test_incident_by_host_by_issue(requests_mock, client):
    """
    Test that By Host By Issue expands each array of rules for each host
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, True)
    assert len(incidents) == 4
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[0]["name"].endswith("[registration.example.com]")
    assert json.loads(incidents[0]["rawJSON"])["rules"][0]["name"] in incidents[0]["name"]
    # NOTE :: Make sure classification titles are used
    assert json.loads(incidents[0]["rawJSON"])["rules"][0]["classification"] == "Critical"
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[1]["name"].endswith("[ip.example.com]")
    assert json.loads(incidents[1]["rawJSON"])["rules"][0]["name"] in incidents[1]["name"]
    assert incidents[2]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[2]["name"].endswith("[stage.example.com]")
    assert json.loads(incidents[2]["rawJSON"])["rules"][0]["name"] in incidents[2]["name"]
    assert incidents[3]["severity"] == IncidentSeverity.LOW
    assert incidents[3]["name"].endswith("[stage.example.com]")
    assert json.loads(incidents[3]["rawJSON"])["rules"][0]["name"] in incidents[3]["name"]


def test_incident_by_host_by_issue_filter(requests_mock, client):
    """
    Test that By Host By Issue expands each array of rules and applies min_severity correcctly
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    client.min_severity = "Moderate"
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, True)
    assert len(incidents) == 3
    assert incidents[0]["severity"] == IncidentSeverity.CRITICAL
    assert incidents[1]["severity"] == IncidentSeverity.MEDIUM
    assert incidents[2]["severity"] == IncidentSeverity.MEDIUM


def test_incident_total_limit(requests_mock, client):
    """
    Some APIs limit how many results are returned. So if the total is over the XSOAR limit, a warning Incident should
    be created
    """
    mock_response = util_load_json("test_data/by-host-issues.json")
    requests_mock.get(
        f"https://api.securitytrails.com/v1/asi/rules/history/{TEST_PROJECT_ID}/activity/by_host/compare", json=mock_response
    )
    client.host_incident_limit = 5
    last_run, incidents = fetch_incidents(client, {"last_fetch": 1234}, True, True)
    assert len(incidents) == 5
    assert incidents[4]["severity"] == IncidentSeverity.LOW
    assert incidents[4]["name"] == "❗Attack Surface Intelligence: 10+ Changes"


@pytest.mark.parametrize(
    "demisto_params_result, expected_result",
    [
        ({"credentials": {"password": "api_key"}, "apikey": "old_api_key"}, {"APIKEY": "api_key"}),
        ({"credentials": {"password": ""}, "apikey": "old_api_key"}, {"APIKEY": "old_api_key"}),
        ({"apikey": "old_api_key"}, {"APIKEY": "old_api_key"}),
    ],
)
def test_get_api_key(mocker, demisto_params_result, expected_result):
    """Test get API key.
    Given: Input parameters to the main function, including the API key configured
           in credentials or passed directly via the apikey parameter.
    When: The main function is called, which instantiates a client.
    Then: Ensure the API key passed to the client constructor matches the expected API key based on the input parameters.
    """
    mocker.patch.object(demisto, "params", return_value=demisto_params_result)
    mock_client = mocker.patch("RecordedFutureASI.Client")
    # Call main()
    RecordedFutureASI.main()

    # Get the client that was instantiated
    assert mock_client.call_args[1].get("headers") == expected_result


def test_get_api_key_invalid_key(mocker):
    """Test get API key.
    Given: Input parameters to the main function, including empty API key and empty credentials object.
    When: The main function is called, which instantiates a client.
    Then: Ensure that error message was raised.
    """
    mocker.patch.object(demisto, "params", return_value={"credentials": {"password": ""}, "apikey": ""})
    mocker.patch.object(demisto, "results")

    # Get the client that was instantiated
    with pytest.raises(SystemExit):
        RecordedFutureASI.main()
    assert demisto.results.call_args[0][0]["Contents"] == "Please provide a valid API token"