Manager

psengine.collective_insights.collective_insights.CollectiveInsights

CollectiveInsights(rf_token: str | None = None)

Class for interacting with the Recorded Future Collective Insights API.

PARAMETER	DESCRIPTION
rf_token	Recorded Future API token. TYPE: str | None DEFAULT: None

Source code in psengine/collective_insights/collective_insights.py

def __init__(
    self,
    rf_token: Annotated[str | None, Doc('Recorded Future API token.')] = None,
):
    """Initializes the CollectiveInsights object."""
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()

submit

submit(
    insight: Insight | list[Insight],
    debug: bool = True,
    organization_ids: list | None = None,
) -> InsightsIn

Submit a detection or insight to the Recorded Future Collective Insights API.

PARAMETER	DESCRIPTION
insight	A detection or list of detections to submit. TYPE: Insight | list[Insight]
debug	Whether the submission should appear in the SecOPS dashboard. TYPE: bool DEFAULT: True
organization_ids	List of organization IDs. TYPE: list | None DEFAULT: None

Endpoint

collective-insights/detections

RAISES	DESCRIPTION
CollectiveInsightsError	If connection error occurs.
ValidationError	If any supplied parameter is of incorrect type.

RETURNS	DESCRIPTION
InsightsIn	Response from the Recorded Future API.

Source code in psengine/collective_insights/collective_insights.py

@validate_call
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=CollectiveInsightsError)
def submit(
    self,
    insight: Annotated[
        Insight | list[Insight], Doc('A detection or list of detections to submit.')
    ],
    debug: Annotated[
        bool, Doc('Whether the submission should appear in the SecOPS dashboard.')
    ] = True,
    organization_ids: Annotated[list | None, Doc('List of organization IDs.')] = None,
) -> Annotated[InsightsIn, Doc('Response from the Recorded Future API.')]:
    """Submit a detection or insight to the Recorded Future Collective Insights API.

    Endpoint:
        `collective-insights/detections`

    Raises:
        CollectiveInsightsError: If connection error occurs.
        ValidationError: If any supplied parameter is of incorrect type.
    """
    if not insight:
        raise ValueError('Insight cannot be empty')

    insight = insight if isinstance(insight, list) else [insight]

    ci_data = self._prepare_ci_request(insight, debug, organization_ids)
    response = self.rf_client.request(
        'post',
        url=EP_COLLECTIVE_INSIGHTS_DETECTIONS,
        data=ci_data.json(),
    )

    return InsightsIn.model_validate(response.json())

create

create(
    ioc_value: str,
    ioc_type: str,
    timestamp: str,
    detection_type: str,
    detection_sub_type: str | None = None,
    detection_id: str | None = None,
    detection_name: str | None = None,
    ioc_field: str | None = None,
    ioc_source_type: str | None = None,
    incident_id: str | None = None,
    incident_name: str | None = None,
    incident_type: str | None = None,
    mitre_codes: list[str] | str | None = None,
    malwares: list[str] | str | None = None,
    **kwargs,
) -> Insight

Create a new Insight object.

PARAMETER	DESCRIPTION
ioc_value	The value of the IOC. TYPE: str
ioc_type	The type of the IOC. TYPE: str
timestamp	The timestamp associated with the detection as ISO 8601. TYPE: str
detection_type	The type of the detection. TYPE: str
detection_sub_type	The subtype of the detection. TYPE: str | None DEFAULT: None
detection_id	The ID of the detection. TYPE: str | None DEFAULT: None
detection_name	The name of the detection. TYPE: str | None DEFAULT: None
ioc_field	The field in which the IOC was detected. TYPE: str | None DEFAULT: None
ioc_source_type	The source type of the IOC. TYPE: str | None DEFAULT: None
incident_id	The ID of the incident. TYPE: str | None DEFAULT: None
incident_name	The name of the incident. TYPE: str | None DEFAULT: None
incident_type	The type of the incident. TYPE: str | None DEFAULT: None
mitre_codes	MITRE ATT&CK technique or tactic codes. TYPE: list[str] | str | None DEFAULT: None
malwares	Associated malware family or names. TYPE: list[str] | str | None DEFAULT: None

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.

RETURNS	DESCRIPTION
Insight	The created Insight object.

Source code in psengine/collective_insights/collective_insights.py

@validate_call
@debug_call
def create(
    self,
    ioc_value: Annotated[str, Doc('The value of the IOC.')],
    ioc_type: Annotated[str, Doc('The type of the IOC.')],
    timestamp: Annotated[str, Doc('The timestamp associated with the detection as ISO 8601.')],
    detection_type: Annotated[str, Doc('The type of the detection.')],
    detection_sub_type: Annotated[str | None, Doc('The subtype of the detection.')] = None,
    detection_id: Annotated[str | None, Doc('The ID of the detection.')] = None,
    detection_name: Annotated[str | None, Doc('The name of the detection.')] = None,
    ioc_field: Annotated[str | None, Doc('The field in which the IOC was detected.')] = None,
    ioc_source_type: Annotated[str | None, Doc('The source type of the IOC.')] = None,
    incident_id: Annotated[str | None, Doc('The ID of the incident.')] = None,
    incident_name: Annotated[str | None, Doc('The name of the incident.')] = None,
    incident_type: Annotated[str | None, Doc('The type of the incident.')] = None,
    mitre_codes: Annotated[
        list[str] | str | None, Doc('MITRE ATT&CK technique or tactic codes.')
    ] = None,
    malwares: Annotated[
        list[str] | str | None, Doc('Associated malware family or names.')
    ] = None,
    **kwargs,
) -> Annotated[Insight, Doc('The created Insight object.')]:
    """Create a new Insight object.

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
    """
    incident = {'id': incident_id, 'type': incident_type, 'name': incident_name}
    detection = {
        'id': detection_id,
        'name': detection_name,
        'type': detection_type,
        'sub_type': detection_sub_type,
    }
    ioc = {
        'type': ioc_type,
        'value': ioc_value,
        'source_type': ioc_source_type,
        'field': ioc_field,
    }
    data = {
        'timestamp': timestamp,
        'ioc': ioc,
        'incident': incident,
        'detection': detection,
        'mitre_codes': mitre_codes,
        'malwares': malwares,
    }
    data['incident'] = (
        None
        if isinstance(data['incident'], dict)
        and all(sub_v is None for sub_v in data['incident'].values())
        else data['incident']
    )
    if kwargs:
        data.update(kwargs)

    return Insight.model_validate(data)

search

search(
    indicator_type: list[str]
    | str
    | Presence
    | None = None,
    detection_type: list[str]
    | str
    | Presence
    | None = None,
    submission_method: list[str]
    | str
    | Presence
    | None = None,
    organizations: list[str] | str | None = None,
    detection_rule_id: list[str]
    | str
    | Presence
    | None = None,
    detection_time_from: str | datetime | None = None,
    detection_time_to: str | datetime | None = None,
    malware_id: list[str] | str | Presence | None = None,
    mitre_code_id: list[str] | str | Presence | None = None,
    threat_actor_id: list[str]
    | str
    | Presence
    | None = None,
    atop_use_case: list[str] | str | Presence | None = None,
    atop_profile_id: list[str]
    | str
    | Presence
    | None = None,
    atop_job_id: list[str] | str | Presence | None = None,
    integration_type_id: list[str]
    | str
    | Presence
    | None = None,
    indicator_risk_score: dict
    | Literal['present', 'absent']
    | None = None,
    max_results: int = Field(ge=1, default=DEFAULT_LIMIT),
    page_size: int = Field(
        ge=1, le=SEARCH_MAX_LIMIT, default=SEARCH_PAGE_SIZE
    ),
) -> list[SearchEntry]

Search enriched Collective Insights events.

PARAMETER	DESCRIPTION
indicator_type	IOC type filter (ip, domain, hash, url, vulnerability). TYPE: list[str] | str | Presence | None DEFAULT: None
detection_type	Detection method filter (correlation, playbook, detection_rule, sandbox, threat_hunt, vulnerability_scan). TYPE: list[str] | str | Presence | None DEFAULT: None
submission_method	Submission method filter (api, integration, sandbox). TYPE: list[str] | str | Presence | None DEFAULT: None
organizations	Filter by organization IDs (uhash). TYPE: list[str] | str | None DEFAULT: None
detection_rule_id	Filter by associated detection rule IDs. TYPE: list[str] | str | Presence | None DEFAULT: None
detection_time_from	Start of the detection time range (inclusive). TYPE: str | datetime | None DEFAULT: None
detection_time_to	End of the detection time range (inclusive). TYPE: str | datetime | None DEFAULT: None
malware_id	Filter by associated malware entity IDs. TYPE: list[str] | str | Presence | None DEFAULT: None
mitre_code_id	Filter by associated MITRE ATT&CK IDs (prefixed with mitre:). TYPE: list[str] | str | Presence | None DEFAULT: None
threat_actor_id	Filter by associated threat actor entity IDs. TYPE: list[str] | str | Presence | None DEFAULT: None
atop_use_case	Filter by Autonomous Threat Operations use case (hunting, detection, prevention). TYPE: list[str] | str | Presence | None DEFAULT: None
atop_profile_id	Filter by Autonomous Threat Operations profile ID. TYPE: list[str] | str | Presence | None DEFAULT: None
atop_job_id	Filter by Autonomous Threat Operations job ID. TYPE: list[str] | str | Presence | None DEFAULT: None
integration_type_id	Filter by integration type entity IDs. TYPE: list[str] | str | Presence | None DEFAULT: None
indicator_risk_score	Filter by indicator risk score at detection time. Pass present/absent or a range dict such as {"gte": 50, "lt": 90}. TYPE: dict | Literal['present', 'absent'] | None DEFAULT: None
max_results	Maximum number of events to return. TYPE: int DEFAULT: Field(ge=1, default=DEFAULT_LIMIT)
page_size	Number of events per page (max 1000). TYPE: int DEFAULT: Field(ge=1, le=SEARCH_MAX_LIMIT, default=SEARCH_PAGE_SIZE)

Endpoint

collective-insights/search

RAISES	DESCRIPTION
CollectiveInsightsSearchError	If connection error occurs.
ValidationError	If any supplied parameter is of incorrect type.

RETURNS	DESCRIPTION
list[SearchEntry]	Enriched events matching the search criteria.

Source code in psengine/collective_insights/collective_insights.py

@validate_call
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=CollectiveInsightsSearchError)
def search(
    self,
    indicator_type: Annotated[
        list[str] | str | Presence | None,
        Doc('IOC type filter (`ip`, `domain`, `hash`, `url`, `vulnerability`).'),
    ] = None,
    detection_type: Annotated[
        list[str] | str | Presence | None,
        Doc(
            'Detection method filter (`correlation`, `playbook`, `detection_rule`, '
            '`sandbox`, `threat_hunt`, `vulnerability_scan`).',
        ),
    ] = None,
    submission_method: Annotated[
        list[str] | str | Presence | None,
        Doc('Submission method filter (`api`, `integration`, `sandbox`).'),
    ] = None,
    organizations: Annotated[
        list[str] | str | None, Doc('Filter by organization IDs (uhash).')
    ] = None,
    detection_rule_id: Annotated[
        list[str] | str | Presence | None, Doc('Filter by associated detection rule IDs.')
    ] = None,
    detection_time_from: Annotated[
        str | datetime | None, Doc('Start of the detection time range (inclusive).')
    ] = None,
    detection_time_to: Annotated[
        str | datetime | None, Doc('End of the detection time range (inclusive).')
    ] = None,
    malware_id: Annotated[
        list[str] | str | Presence | None,
        Doc('Filter by associated malware entity IDs.'),
    ] = None,
    mitre_code_id: Annotated[
        list[str] | str | Presence | None,
        Doc('Filter by associated MITRE ATT&CK IDs (prefixed with `mitre:`).'),
    ] = None,
    threat_actor_id: Annotated[
        list[str] | str | Presence | None,
        Doc('Filter by associated threat actor entity IDs.'),
    ] = None,
    atop_use_case: Annotated[
        list[str] | str | Presence | None,
        Doc(
            'Filter by Autonomous Threat Operations use case '
            '(`hunting`, `detection`, `prevention`).',
        ),
    ] = None,
    atop_profile_id: Annotated[
        list[str] | str | Presence | None,
        Doc('Filter by Autonomous Threat Operations profile ID.'),
    ] = None,
    atop_job_id: Annotated[
        list[str] | str | Presence | None,
        Doc('Filter by Autonomous Threat Operations job ID.'),
    ] = None,
    integration_type_id: Annotated[
        list[str] | str | Presence | None, Doc('Filter by integration type entity IDs.')
    ] = None,
    indicator_risk_score: Annotated[
        dict | Literal['present', 'absent'] | None,
        Doc(
            'Filter by indicator risk score at detection time. Pass `present`/`absent` '
            'or a range dict such as `{"gte": 50, "lt": 90}`.',
        ),
    ] = None,
    max_results: Annotated[int, Doc('Maximum number of events to return.')] = Field(
        ge=1, default=DEFAULT_LIMIT
    ),
    page_size: Annotated[int, Doc('Number of events per page (max 1000).')] = Field(
        ge=1, le=SEARCH_MAX_LIMIT, default=SEARCH_PAGE_SIZE
    ),
) -> Annotated[
    list[SearchEntry],
    Doc('Enriched events matching the search criteria.'),
]:
    """Search enriched Collective Insights events.

    Endpoint:
        `collective-insights/search`

    Raises:
        CollectiveInsightsSearchError: If connection error occurs.
        ValidationError: If any supplied parameter is of incorrect type.
    """
    data = {
        'filters': {
            'organizations': organizations,
            'indicator_type': indicator_type,
            'detection_rule': detection_rule_id,
            'detection_type': detection_type,
            'submission_method': submission_method,
            'detection_time': {'from': detection_time_from, 'to': detection_time_to},
            'associated_threats': {
                'malware': malware_id,
                'mitre_code': mitre_code_id,
                'threat_actor': threat_actor_id,
            },
            'autonomous_threat_operations': {
                'use_case': atop_use_case,
                'profile': atop_profile_id,
                'job': atop_job_id,
            },
            'integration_type': integration_type_id,
            'indicator': {'risk': {'score': {'at_detection': indicator_risk_score}}},
        },
        'limit': min(page_size, max_results),
    }
    search_data = SearchIn.model_validate(data)
    self.log.info(f'Searching Collective Insights events with query: {search_data.json()}')

    results = self.rf_client.request_paged(
        method='post',
        url=EP_COLLECTIVE_INSIGHTS_SEARCH,
        data=search_data.json(),
        results_path='data',
        offset_key='offset',
        max_results=max_results,
    )

    return [SearchEntry.model_validate(r) for r in results]