Skip to content

Collective Insights

Introduction

The CollectiveInsights class of the collective_insights module allows you to submit indicators from a detection coming from any of your internal tools to your Recorded Future enterprise. Collective Insights enriches submissions with Recorded Future intelligence to provide your organization's enterprise with enhanced and actionable intelligence.

When using the collective_insights module, you have to create an Insight object and then submit it. The CollectiveInsights class provides access to the modules needed to perform these actions. The examples below show how to use them.

See the API Reference for internal details of the module.

Notes

  • There are some limitations around the number of submissions allowed per day; see the Collective Insight API documentation.
  • The insights submitted require some fields to be filled with specific values. For example, the detection_type can be either sigma, yara, or snort. To help you, there are constant values defined in collective_insights.constants. See a usage example below.
  • The CollectiveInsights.create method has a default value of debug=True; this is used to make sure the whole workflow works, but the indicators are not submitted to Recorded Future. In production code, set debug=False when you are ready to share the detections with Recorded Future.

Examples

Warning

The following examples demonstrate how to use this module. Be sure to add appropriate error handling as needed; all possible errors for each method or function are listed in the API Reference page.

Additionally, you must configure the RF_TOKEN environment variable before getting started. For instructions, see Learn.

1: Submit a detection for a Wiper malware hash

The create method only requires the following arguments to be specified:

  • ioc_value
  • ioc_type
  • detection_type
  • timestamp

Every other value can optionally be provided and adds more context to the detection. In the example below, we have a detection coming from Symantec. The hash is coming from a Recorded Future Insikt note defined by the id doc:o6_lui. The other information is retrieved by both the note (for example, the malware type and MITRE codes) and the incident itself.

In our case, the timestamp has been mocked to "now", but in a real scenario it would be taken from the incident.

from datetime import UTC, datetime

from psengine.collective_insights import (
    DETECTION_SUB_TYPE_SIGMA,
    DETECTION_TYPE_RULE,
    ENTITY_HASH,
    CollectiveInsights,
)

ci = CollectiveInsights()

now = datetime.now(UTC).isoformat()

insight1 = ci.create(
    ioc_value='fbee00cb1d1ea4d7e0604436d9a36def71a9f3be804f1e2b8d117fd5d35aeabc',
    ioc_type=ENTITY_HASH,
    detection_type=DETECTION_TYPE_RULE,
    detection_sub_type=DETECTION_SUB_TYPE_SIGMA,
    detection_id='doc:o6_lui',
    detection_name='Instance of Alleged New Wiper Malware',
    ioc_field='hash',
    ioc_source_type='symantec',
    timestamp=now,
    incident_id='Incident 001',
    incident_name='Malware detected',
    incident_type='RF Sigma Rule',
    mitre_codes=['T1542', 'T1485'],
    malwares=['Aesthetic Wiper'],
)

insights = [insight1]
ci.submit(insight=insights)

The create method returns an Insight object, which can be passed to the submit method.

To see results from the API without actually submitting anything to the Recorded Future platform, use debug=True in your call to submit.

The submit method takes a single Insight or a list of Insight objects.

2: Search previously submitted enriched events

The search method queries events already processed in Collective Insights analysis and enriched with associated threat entities (threat actors, malware, MITRE codes). Every filter is optional; supplying a bare value (like indicator_type='ip') is coerced to a list, and the sentinel strings 'present'/'absent' filter for events where a given field is set or unset.

In the example below, we look for IP-typed events submitted via the api or integration methods over the last 7 days that have at least one associated malware entity, and print the resulting SearchEntry objects sorted from newest to oldest.

from datetime import UTC, datetime, timedelta

from psengine.collective_insights import CollectiveInsights

ci = CollectiveInsights()

last_week = (
    datetime.now(UTC) - timedelta(days=7)
).isoformat()

events = ci.search(
    indicator_type='ip',
    submission_method=['api', 'integration'],
    malware_id='present',
    detection_time_from=last_week,
    max_results=5,
)

for event in sorted(events, reverse=True):
    print(event)

This example will print something like:

1
2
3
Event ID: 16046941d44a85e4748a9e9c, IOC: 1.2.3.4, Detection Time: 2026-08-11 09:14:02, Detection Type: sandbox
Event ID: 16046941d44a85e4748a9e9b, IOC: 5.6.7.8, Detection Time: 2026-08-09 22:41:18, Detection Type: correlation
Event ID: 16046941d44a85e4748a9e9a, IOC: 9.10.11.12, Detection Time: 2026-08-06 15:05:23, Detection Type: detection_rule

Each SearchEntry also exposes the enriched fields (associated_threats, autonomous_threat_operations, indicator.risk, etc.); see the API Reference for the full schema.