Examples: Using the API to Insert a Security Event Document to aella-ser Index

Inserting a Security Event with Basic Authentication

headers={'Accept': 'application/json', 'Content-type': 'application/json'}

elastic_url ='https:// YourStellarCyberServer/connect/api/v1/insert_ser'

json_data = {

"event_name": "EventName",

"timestamp": Timestamp,

"event_status": "EventStatus",

"event_source": "EventSource",

}

query = json.dumps(json_data)

response = requests.post(elastic_url, auth=('Username','API Key'), data=query, verify=False, headers = headers)

print (response.text)

Your call must have the header and the /connect/api/v1/insert_ser path, and must have at least the following fields, in addition to any optional aella-ser index fields:

• event_name

• timestamp

• event_status

• event_source

Enter your own information for the arguments in bold.

The contents of response.text indicate success or failure and will be similar to:

• OK 200—the API call was successful.
• Invalid event status value: closed—the API call failed because the supplied status was invalid.
• Missing fields to be updated for the event—the API call failed because you left out one of the mandatory fields.

Sample insert_ser Call

Here is an example with arguments filled in:

headers={'Accept': 'application/json', 'Content-type': 'application/json'}

elastic_url ='https://192.168.1.24/connect/api/v1/insert_ser'

json_data = {

"event_name": "demo test",

"detected_field": "my field test",

"tryout": "test",

"timestamp": long(time.time() * 1000),

"event_status": "New",

"comments": "Test Comment",

"event_source": "ActZero",

}

query = json.dumps(json_data)

response = requests.post(elastic_url, auth=('admin','APISquared''), data=query, verify=False, headers = headers)

print (response.text)

This example performs a POST call as the user admin with the obviously fake API key of APISquared. The call is to the Stellar Cyber server at 192.168.1.24 to insert a document in the aella-ser index with the following fields:

• event_name—demo test

• detected_field—my field test

• tryout—test

• timestamp—Current time in Python (long(time.time() * 1000))

• event_status—New

• comments—Test Comment

• event_source—ActZero

Inserting a Security Event with JWT Authentication

You can also use JWT authentication, as in the example below.

Your call must have the header and the /connect/api/v1/insert_ser path, and must have at least the following fields, in addition to any optional aella-ser index fields:

• event_name

• timestamp

• event_status

• event_source

Enter your own information in the script for the arguments in the table below.

The contents of response.text indicate success or failure and will be similar to:

• OK 200—the API call was successful.
• Invalid event status value: closed—the API call failed because the supplied status was invalid.
• Missing fields to be updated for the event—the API call failed because you left out one of the mandatory fields.

Sample insert_ser Script

The sample script below works as follows:

• The script sets the host, userid, and refresh_token parameters in Step 1 in the sample.

• Because JWTs expire ten minutes after they are generated, this script includes logic that generates and uses a fresh JWT every time the script is run. The script runs the getAccessToken procedure to generate the new JWT (Step 2 in the sample).

• The script uses the generated JWT to make a POST to the insert_ser API in the putCase procedure (Step 3 in the sample). The script posts the following JSON data:

  • event_name—demo test

  • detected_field—my field test

  • tryout—test

  • timestamp—Current time in Python (long(time.time() * 1000))

  • event_status—New

  • comments—Test Comment

  • event_source—ActZero

• The script also prints the generated JWT to the screen. This, however, is not strictly necessary since the getAccessToken procedure already prints the status code for the call to the access_token API (200 for success; 401 for failure).

#!/usr/bin/python3

import requests
import base64
import json
import time
from urllib.parse import urlunparse
requests.packages.urllib3.disable_warnings()

# Step 1
# Add DP IP/hostname, userid, and refresh token from GUI here
HOST = "myserver.stellarcyber.cloud"
userid = "myuser@stellarcyber.ai"
refresh_token ="2iRpBAyQYEfv77R2QtATlJN6Nvq6uzftBdzotSy2pjT-IvJTLw9aiHyh7Y2mo12IDSWc-FfHwUyPpmiHQnJrSH"

def getAccessToken(userid, refresh_token):
    auth = base64.b64encode(bytes(userid + ":" + refresh_token, "utf-8")).decode("utf-8")
    headers = {
        "Authorization": "Basic " + auth,
        "Content-Type": "application/x-www-form-urlencoded",
    }
    url = urlunparse(("https", HOST, "/connect/api/v1/access_token", "", "", ""))
    res = requests.post(url, headers=headers, verify=False)
    print(res.status_code)
    return res.json()["access_token"]
    
def putCase(token):
    headers = {"Authorization": "Bearer " + token, 'content-type': 'application/json' }
    url = urlunparse(("https", HOST, "/connect/api/v1/insert_ser", "", "", ""))
    json_data = {
      "event_name": "demo test",
      "detected_field": "my field test",
      "tryout": "test",
      "timestamp": time.time() * 1000,
      "event_status": "New",
      "comments": "Test Comment",
      "event_source": "ActZero",
    }
    query = json.dumps(json_data)
    res = requests.post(url, data=query, headers=headers, verify=False)
    print(res.text)
    print(res.status_code)
    return res.json()
    
if __name__ == "__main__":

    # Step 2: Use getAccessToken with supplied credentials to generate JWT
    jwt = getAccessToken(userid, refresh_token)
    print("------------ jwt -------------")
    print(jwt)
    print("------------ jwt  end -------------")
    
    # Step 3: use JWT token to call public API
    data = putCase(jwt)
    print("------------ show result of put case -------------")
    print(data)
    print("------------ end api results -------------")