Manager

psengine.links.links_mgr.LinksMgr

LinksMgr(rf_token: str | None = None)

Manager for interacting with the Recorded Future Links API.

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

Source code in psengine/links/links_mgr.py

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

list_sections

list_sections() -> list[IdName]

List all sections that can be used to filter a Link search.

Sections are the high-level categories the Links API groups results into, for example Actors, Tools & TTPs or Indicators & Detection Rules.

Endpoint

/links/metadata/sections

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
LinksMetadataError	If an API or connection error occurs.

RETURNS	DESCRIPTION
list[IdName]	List of section objects with id and name.

Source code in psengine/links/links_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=LinksMetadataError)
def list_sections(
    self,
) -> Annotated[list[IdName], Doc('List of section objects with id and name.')]:
    """List all sections that can be used to filter a Link search.

    Sections are the high-level categories the Links API groups results into,
    for example *Actors, Tools & TTPs* or *Indicators & Detection Rules*.

    Endpoint:
        `/links/metadata/sections`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        LinksMetadataError: If an API or connection error occurs.
    """
    response = self.rf_client.request(method='GET', url=EP_LINKS_METADATA_SECTIONS)
    return [IdName.model_validate(item) for item in response.json()['data']]

list_events

list_events() -> list[IdName]

List all event types that can be used to filter technical Link searches.

Event types describe the kind of analytical evidence that produced a technical link (for example TTPAnalysis or InfrastructureAnalysis).

Endpoint

/links/metadata/events

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
LinksMetadataError	If an API or connection error occurs.

RETURNS	DESCRIPTION
list[IdName]	List of event objects with id and name.

Source code in psengine/links/links_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=LinksMetadataError)
def list_events(
    self,
) -> Annotated[list[IdName], Doc('List of event objects with id and name.')]:
    """List all event types that can be used to filter technical Link searches.

    Event types describe the kind of analytical evidence that produced a
    technical link (for example `TTPAnalysis` or `InfrastructureAnalysis`).

    Endpoint:
        `/links/metadata/events`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        LinksMetadataError: If an API or connection error occurs.
    """
    response = self.rf_client.request(method='GET', url=EP_LINKS_METADATA_EVENTS)
    return [IdName.model_validate(item) for item in response.json()['data']]

list_entity_types

list_entity_types() -> list[IdName]

List all entity types that can be used to filter a Link search.

The returned values are the supported types for connected entities (for example Malware, Company, IpAddress).

Endpoint

/links/metadata/entities

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
LinksMetadataError	If an API or connection error occurs.

RETURNS	DESCRIPTION
list[IdName]	List of entity-type objects with id and name.

Source code in psengine/links/links_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=LinksMetadataError)
def list_entity_types(
    self,
) -> Annotated[list[IdName], Doc('List of entity-type objects with id and name.')]:
    """List all entity types that can be used to filter a Link search.

    The returned values are the supported types for connected entities
    (for example `Malware`, `Company`, `IpAddress`).

    Endpoint:
        `/links/metadata/entities`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        LinksMetadataError: If an API or connection error occurs.
    """
    response = self.rf_client.request(method='GET', url=EP_LINKS_METADATA_ENTITIES)
    return [IdName.model_validate(item) for item in response.json()['data']]

search

search(
    entities: str | list[str],
    sections: str | list[str] | None = None,
    entity_types: str | list[str] | None = None,
    sources: str | list[LinkSource] | None = None,
    timeframe: str | None = None,
    events: str | list[str] | None = None,
    connected_entities: str | list[str] | None = None,
    search_scope: SearchScope | None = 'medium',
    per_entity_type: int | None = None,
) -> list[EntityLinks]

Search for technically validated relationships between threat intelligence entities in the Recorded Future Intelligence Cloud — connections established through sandbox analysis, infrastructure analysis, network traffic analysis, and Insikt Group research.

Issues a single batched request: the response contains one EntityLinks per entity in entities, in the same order. If the API failed for a specific entity, that result's error is populated and links is empty — the rest of the batch still succeeds.

Entities must be supplied as Recorded Future entity IDs; if you only have a name, resolve it with EntityMatchMgr first.

PARAMETER	DESCRIPTION
entities	One or more Recorded Future entity IDs to look up links for. TYPE: str | list[str]
sections	Filter results to these link section IDs. TYPE: str | list[str] | None DEFAULT: None
entity_types	Restrict linked entities to these entity types (e.g. "type:IpAddress"). TYPE: str | list[str] | None DEFAULT: None
sources	Limit to source type(s): "technical", "insikt", or both if argument omitted. TYPE: str | list[LinkSource] | None DEFAULT: None
timeframe	Technical-link timeframe (e.g. "-30d", default "-30d", max "-90d"). TYPE: str | None DEFAULT: None
events	Restrict technical links to these event types (e.g. "type:MalwareAnalysis"). TYPE: str | list[str] | None DEFAULT: None
connected_entities	Only return technical links that connect to these entities. TYPE: str | list[str] | None DEFAULT: None
search_scope	Result-volume scope: "small", "medium" (default), or "large". TYPE: SearchScope | None DEFAULT: 'medium'
per_entity_type	Max linked entities returned per entity type (>= 1 <= 1,000,000,000). TYPE: int | None DEFAULT: None

Endpoint

/links/search

If the API failed for a specific entity in the batch, its result looks like:

    EntityLinks(
        entity=IdNameType(id_='QCwdoU', name='...', type_='...'),
        links=[],
        error=EntitySearchError(message='...', status_code=404),
    )

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
LinksSearchError	If an API or connection error occurs at the request level.

RETURNS	DESCRIPTION
list[EntityLinks]	A list of EntityLinks objects

Source code in psengine/links/links_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=LinksSearchError)
def search(
    self,
    entities: Annotated[
        Annotated[str | list[str], AfterValidator(Validators.convert_str_to_list)],
        Doc('One or more Recorded Future entity IDs to look up links for.'),
    ],
    sections: Annotated[
        str | list[str] | None, Doc('Filter results to these link section IDs.')
    ] = None,
    entity_types: Annotated[
        str | list[str] | None,
        Doc('Restrict linked entities to these entity types (e.g. "type:IpAddress").'),
    ] = None,
    sources: Annotated[
        str | list[LinkSource] | None,
        Doc('Limit to source type(s): "technical", "insikt", or both if argument omitted.'),
    ] = None,
    timeframe: Annotated[
        str | None,
        Doc('Technical-link timeframe (e.g. "-30d", default "-30d", max "-90d").'),
    ] = None,
    events: Annotated[
        str | list[str] | None,
        Doc('Restrict technical links to these event types (e.g. "type:MalwareAnalysis").'),
    ] = None,
    connected_entities: Annotated[
        str | list[str] | None,
        Doc('Only return technical links that connect to these entities.'),
    ] = None,
    search_scope: Annotated[
        SearchScope | None,
        Doc('Result-volume scope: "small", "medium" (default), or "large".'),
    ] = 'medium',
    per_entity_type: Annotated[
        int | None,
        Field(ge=1, le=1_000_000_000),
        Doc('Max linked entities returned per entity type (>= 1 <= 1,000,000,000).'),
    ] = None,
) -> Annotated[
    list[EntityLinks],
    Doc('A list of EntityLinks objects'),
]:
    """Search for technically validated relationships between threat intelligence
    entities in the Recorded Future Intelligence Cloud — connections established
    through sandbox analysis, infrastructure analysis, network traffic analysis,
    and Insikt Group research.

    Issues a single batched request: the response contains one
    `EntityLinks` per entity in `entities`, in the same order. If the
    API failed for a specific entity, that result's `error` is populated
    and `links` is empty — the rest of the batch still succeeds.

    Entities must be supplied as Recorded Future entity IDs; if you only have
    a name, resolve it with `EntityMatchMgr` first.

    Endpoint:
        `/links/search`

    If the API failed for a specific entity in the batch, its result looks like:
    ```python
        EntityLinks(
            entity=IdNameType(id_='QCwdoU', name='...', type_='...'),
            links=[],
            error=EntitySearchError(message='...', status_code=404),
        )
    ```

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        LinksSearchError: If an API or connection error occurs at the request level.
    """
    technical_filters = FilterTechnical(
        timeframe=timeframe, events=events, connected_entities=connected_entities
    ).json()

    filters = LinksFilterObjects(
        sections=sections,
        entity_types=entity_types,
        sources=sources,
        technical=technical_filters or None,
    ).json()
    limits = LinksLimitsObjects(
        search_scope=search_scope, per_entity_type=per_entity_type
    ).json()
    payload = LinksSearchIn(
        entities=entities,
        filters=filters or None,
        limits=limits or None,
    ).json()

    self.log.info(f'Executing links search for {len(entities)} entities.')

    response = self.rf_client.request(method='POST', url=EP_LINKS_SEARCH, data=payload)
    return [EntityLinks.model_validate(item) for item in response.json()['data']]