Skip to content

Manager

psengine.risk_rules.risk_rule_mgr.RiskRuleMgr 

RiskRuleMgr(rf_token: str | None = None)

Manages requests for Recorded Future risk rules.

PARAMETER DESCRIPTION
rf_token

Recorded Future API token.

TYPE: str | None DEFAULT: None

Source code in psengine/risk_rules/risk_rule_mgr.py 
32
33
34
35
36
37
38
def __init__(
    self,
    rf_token: Annotated[str | None, Doc('Recorded Future API token.')] = None,
):
    """Initializes the `RiskRuleMgr` object."""
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()

fetch 

fetch(entity_type: RiskRuleEntityType) -> list[RiskRule]

Fetch every risk rule defined for the given IOC type.

PARAMETER DESCRIPTION
entity_type

The IOC type to fetch risk rules for: ip, domain, hash, vulnerability, or url.

TYPE: RiskRuleEntityType

Endpoint

/v2/{entity_type}/riskrules

RAISES DESCRIPTION
ValidationError

If entity_type is not one of the supported IOC types.
RiskRuleFetchError

If an API error occurs while fetching risk rules.
RETURNS DESCRIPTION
list[RiskRule]

The list of risk rules for the given IOC type.
Source code in psengine/risk_rules/risk_rule_mgr.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=RiskRuleFetchError)
def fetch(
    self,
    entity_type: Annotated[
        RiskRuleEntityType,
        Doc('The IOC type to fetch risk rules for: ip, domain, hash, vulnerability, or url.'),
    ],
) -> Annotated[list[RiskRule], Doc('The list of risk rules for the given IOC type.')]:
    """Fetch every risk rule defined for the given IOC type.

    Endpoint:
        `/v2/{entity_type}/riskrules`

    Raises:
        ValidationError: If `entity_type` is not one of the supported IOC types.
        RiskRuleFetchError: If an API error occurs while fetching risk rules.
    """
    url = EP_RISK_RULES.format(entity_type.value)
    self.log.info(f'Fetching risk rules for entity type: {entity_type.value}')
    response = self.rf_client.request('get', url).json()
    results = response['data']['results']
    return validate_list(RiskRule, results, id_path='name', log=self.log)