Custom Integration with Falkonry Rules

To integrate with alerts generated by Falkonry Rules to perform condition-based actions, you can use Falkonry APIs. Here is a step-wise example on how to work with Falkonry APIs.

Step 1: Understanding Rules and the output signals

Each Falkonry Rule produces three types of output which are assessment, alert and explanation output. Each of these are represented as Falkonry generated signals.

Rule Output Type	Falkonry Generated Signal Name
Assessment	<rule_name>/rule - contains rule output each time rule is assessed. Data value true means the rule condition was true point in time.
Alert	<rule_name>/alert - contains alerts from a specific rule. All Rules/alerts - contains alerts from across all rules.
Explanation	All Rules/<rule_name>/Explanation - contains explanation data across all signals that are input to a rule. <rule_name>/Explanation-<signal_name> - contains explanation data for a specific signal that is input to a rule.

Step 2: Getting alert signal ID

Go to Signal Manager and search for the alert signal. You could use the All Rules/alerts signal ID for fetching alerts from across all the rules.

Step 3: Fetching alerts using the alert signal ID

Use :ref: [Alerts API \<Get Alerts>]{.title-ref} to fetch the alerts. You can poll the API to get recent alerts every 5 minutes. For time critical alerts, you can poll every 1 minute.

import requests
import time

TOKEN = '<token>'
HEADERS = {'Authorization': f'Bearer {TOKEN}', 'Accept': 'application/json'}

while True:

  # fetch alerts across all the rules

  end_time   = time.time() * 1000000000         # time in nanoseconds
  start_time = end_time - (5 * 60 * 1000000000) # now - 5m in nanoseconds

  URL = 'https://app3.falkonry.ai/api/1.2/data/accounts/<account_id>/connectedsources/<rule_alert_ID>/raw_data?start=' + start_time + '&end=' + end_time
  alerts = requests.get(URL, headers=HEADERS)
  print(alerts.json())

  time.sleep(5*60) # sleep for 5 minutes

Step 4: Getting rule configuration and explanation signal ID using rule ID

Use Signals API <Get Signals List> to get the explanation signal ID for the specific rule.

for alert in alerts.json():

  # get rule name and description using rule ID
  URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/assessments/' + alert["signal"]
  response = requests.get(URL, headers=HEADERS)
  rule = response.json()

  # get rule configuration
  URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/assessments/' + alert["signal"] + '/models?type=RULE&isCommonModel=true&offset=0&limit=1'
  response = requests.get(URL, headers=HEADERS)
  ruleConfiguration = response.json()

  # get explanation signal ID using rule name
  signal_name = 'All Rules/' + rule["name"] + '/Explanation'
  URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/connectedsources?sourceName="' + signal_name + '"&offset=0&limit=1'
  response = requests.get(URL, headers=HEADERS)
  explanation_signal = response.json()[0]

  # fetch explanation data using explanation signal ID
  URL = 'https://app3.falkonry.ai/api/1.2/data/accounts/<account_id>/connectedsources/' + explanation_signal["id"] + '/raw_data?start=' + start_time + '&end=' + end_time
  explanation_data = requests.get(URL, headers=HEADERS)

Step 5: Getting signal organization information using signal ID

Use Signals API <Get a Signal> to get signal object. The signal object has the tree assignment information as layouts. This can be used to identify the asset and equipment information.

for eachSignal in explanation_data.json():

  if eachSignal["value"] == True:

    # get signal information using signal ID
    URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/connectedsources/' + eachSignal["signal"]
    signal = requests.get(URL, headers=HEADERS)
    print(signal.json())

Step 6: Working with chained rules

If case the input to the rule is output of another rule, the signal ID in the step 5 is the rule assessment signal ID. To get the explanation data for the this rule, use the Rules API to get the rule name and then use the Signals API to get the explanation signal ID.

for eachSignal in explanation_data.json():

  if eachSignal["value"] == True:

    # get rule name from the assessment signal
    URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/connectedsources/' + eachSignal["signal"]
    rule_assessment = requests.get(URL, headers=HEADERS)
    rule_name = rule_assessment.json()["sourceName"].replace("/rule", "")

    # get explanation signal ID using rule name
    signal_name = 'All Rules/' + rule_name + '/Explanation'
    URL = 'https://app3.falkonry.ai/api/1.2/accounts/<account_id>/connectedsources?sourceName="' + signal_name + '"&offset=0&limit=1'
    response = requests.get(URL, headers=HEADERS)
    explanation_signal = response.json()[0]

    # fetch explanation data and signal information using signal ID as explained in *step 5*.