Examples: Using the API to Perform an ElasticSearch Query

This section describes how to use the Stellar Cyber API to perform an ElasticSearch Query on a specified index.

Refer to Configuring API Authentication for general requirements to use the API.

ElasticSearch API Example

The following example uses a Python script to make an ElasticSearch DSL query. Your call must have the application/json header and the /connect/api/data path. Enter your own information for the arguments listed below:

Argument Description
userid User name of the admin making the call. Set to myuser@stellarcyber.ai in this example.
refresh_token API key for the user name (edit a user on the Admin | User Management page to generate an API key). The script generates a JWT using the provided API key as a refresh token. Set to 2iRpBAyQYEfv77R2QtATlJN6Nvq6uzftBdzotSy2pjT-IvJTLw9aiHyh7Y2mo12IDSWc-FfHwUyPpmiHQnJrSH in this example.
Host The URL or IP address of your Stellar Cyber server. Set to myserver.stellarcyber.cloud in this example.
Index The index to be queried.
Query What you're looking for in DSL

Understanding the Script

This script 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 call to the data API in the getData procedure for a search of the aella-eventsummary-* index (Step 3 in the sample).

  • 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).

Copy
#!/usr/bin/python3

import requests
import base64
import json
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 getData(token):
    headers = {"Authorization": "Bearer " + token, 'content-type': 'application/json'}
    url = urlunparse(("https", HOST, "/connect/api/data/aella-eventsummary-*/_search", "", "", ""))
    res = requests.get(url, headers=headers, verify=False)
    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 = getData(jwt)
    print("------------ call result of /connect/api/data -------------")
    print(data)
    print("------------ end api results -------------")

Finding the Index and ID

To find the index and _id:

  1. Log in to Stellar Cyber.
  2. Navigate to the event.
  3. Click More Info for more information on the event. The Event Details screen appears.
  4. Click the Details entry in the left navigation panel (or JSON) and scroll to the bottom of the display.
  5. The _index field is the index, and _id field is the ID.