Working with Orbital API

[dkrull]

• Getting Started
• Prerequisites
• Finding your Orbital Cloud
• Generate your API Credentials
• Using Python
• Generate your Token
• Confirming your token is valid
• Creating a Query
• Checking a Job
• Conclusion

Getting Started

You will need to make sure that you have Orbital access before we begin. You need to be an AMP for endpoints Administrator, Advantage tier licensing or higher and you'll need at least one host with Orbital installed and connected to the cloud. We are going to be using Python to craft our API but feel free to use any software you feel comfortable with.

Prerequisites

• AMP Administrator Account

• Advantage tier licensing or higher (only if you were not a paying customer prior to January 8th 2020. See this FAQ)

• Opt-In'd to Orbital under Accounts > Licenses with your AMP for Endpoints portal if not already.

• Orbital Enabled within your AMP for Endpoints Policy under Management > Policy > Advance Settings > Orbital > Enable Orbital Advanced Search

• Pycharm or another Interpreter

• Python3.7+

Helpful Links

• Orbital APIs

Finding your Orbital Cloud

• North America
• Europe
• Asia-Pacific

Generate your API Credentials

Before we can start using the API we need to generate our API credentials. This can be done in two different ways, either via the direct Orbital interface or via Cisco Threat Response interface. Since this guide is focused on Orbital we will be using that method.

After logging into your Orbital portal navigate to My Account you should see a option to "Create API Credentials" click on it.

This will open a new window where you can provide a API Credential Name and Description. These can be whatever you want, once filled out click 'Create'.

In your new window you will now see your 'Client ID' and Client 'Password'. The Client Password cannot be recovered once you close this window. Please store securely. You will be using this to request your Token.

A quick note about the token and the API credentials. Your client credentials are used for exactly one purpose, to generate access tokens. The tokens that are generated will expire after a brief period of time. Any application will need to periodically request a new token. See more here

Using Python

These scripts are provided as is and is to help those users who are maybe new to Python or coding in general. More experienced coders may not find this section useful but it'll give you a good idea what the Orbital API expects as far as formatting goes. If you do have a better way of coding the below scripts please feel free to share your code in the comments below!

Before we begin each of these code blocks uses some or if not all of the following modules and use them in each of the major code blocks. Its going to be assumed that all three are present at all times.

import json
import requests
import time

URLs that well be using to poll.

https://orbital.amp.cisco.com/v0/oauth2/token – Exchanges your client ID and secret for an access token.
https://orbital.amp.cisco.com/v0/ok – Confirms that credentials work and gets information about the client.
https://orbital.amp.cisco.com/v0/query – Schedules a query to be performed one or more times, returns a job object.
https://orbital.amp.cisco.com/v0/{jobid}/results – Retrieves the results of your queries.

And last little bit all results are returned in a dictionary so you can navigate through it. I will provide the keys at the bottom of each section if possible. Depending on your queries you may have a very large dictionary.

Generate your Token

This code block will generate a new token on each run and store it into a variable called new_token or access_token that you can call later on. Keys can be found bellow the code block.

# URL to query
token_url = 'https://orbital.amp.cisco.com/v0/oauth2/token'
 
# Generate Access Token
encoded_base64 = 'Your API username:password in base64 format'
headers = {
    'Authorization':'Basic {}'.format(encoded_base64),
    'Content-Type':'application/json', 'Accept':'application/json'
}
request_token = requests.post(token_url, headers=headers)
token_dict = request_token.json()
new_token = token_dict['token']
new_token_expiry = token_dict['expiry']
 
# Display token information
print('Your new token has been generated and can be found below:'
      '\n Your Token: {} '
      '\n Your Token Expires in ~10mins: {} UTC'
      '\n Current local time:            {}'
      '\n Current local time in Epoch:   {}'.format(new_token, new_token_expiry, time.asctime(), round(time.time())))
 
# Set new token to access token
access_token = new_token

After running you should get a similar output as below. I've shorted my token output for formatting but yours will be much longer:

Your new token has been generated and can be found below:
 Your Token: eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6IkhqVW4yNlBPUGZlWFFx... 
 Your Token Expires in ~10mins: 2020-06-29T18:17:10.465320081Z UTC
 Current local time:            Mon Jun 29 14:07:13 2020
 Current local time in Epoch:   1593454033

'token'	Your token
'expiry'	The date and time the token expires in UTC

 

Confirming your token is valid

After generating your token we need to make sure its valid and remains valid during the roughly 10 minutes we have with it. Now, in your code you can define a function to verify if your token is still valid and if so continue to use the token if not generate a new token. Or if you really don't care you can generate a new token for each request but if you have 10 billion requests it may not be the most efficient way. To save some time I've exported my token into a text file on disk that I write my token value to and read from between scripts. This isn't an elegant or recommend method but gets the job done for this but all I've done is added the follow few lines to the bottom of the previous section.

# Open file, write token, close file
temp_file = open('Temp_token.txt', 'w')
temp_file.write(access_token)
temp_file.close()
 
# Open file, write jobid, close file
temp_jobid = open('temp_jobid.txt', 'w')
jobid = temp_jobid.write(response_results['ID'])
temp_jobid.close()

And at the code to read at the top of the file:

# Open file, read token, close file
temp_file = open('Temp_token.txt', 'r')
new_token = temp_file.read(temp_file)
temp_file.close()
 
# Open file, read jobid, close file
temp_jobid = open('temp_jobid.txt', 'r')
jobid = temp_jobid.read()
temp_jobid.close()

Below we have the code block that will check to see if your token is valid or not and store that response in a dictionary for later use. Keys can be found the code block

# Open file, write token, close file
temp_file = open('Temp_token.txt', 'r')
new_token = temp_file.read()
temp_file.close()
 
# URL to query
ok_url = 'https://orbital.amp.cisco.com/v0/ok'
 
headers = {
    'Authorization': 'Bearer {}'.format(new_token),
    'Content-Type': 'application/json', 'Accept': 'application/json'
}
 
ok_token = requests.get(ok_url, headers=headers)
 
print("HTTP Response Code: ", ok_token)
print("Token Response: ", ok_token.json())
 
# Store response for later use
ok_response = ok_token.json()

And your results should be something similar to the below:

HTTP Response Code:  <Response [200]>
Token Response:  {'login': {'user': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXX', 'email': 'dkrull@cisco.com', 'userName': 'Dmitri Krull', 'organization': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', 'organizationName': 'Cisco - dkrull', 'accessLevel': 'write', 'admin': False, 'amp': False}, 'message': 'OK'}

'error'	Any related HTTP error code
'login'	Contains all user information
'user'	UUID for user authenticating
'email'	Email associated with authenticating user
'userName'	Username associated with authenticating user
'organization'	UUID of the organization of authenticating user
'organization name'	Name of the organization of authenticating user
'accessLevel'	Access Level of user authentication
'admin'	Was the API Credentials created with Admin privileges  (True/False)
'amp'	Does the API have AMP for Endpoints access (True/False).
'message'	HTTP Return Code.

Creating a Query

With all of the above we can now craft our very own query. This query will generate a job in the orbital dashboard. The results will not be returned to console and you will need to either periodically check the job for current status or completion then pull the results either back into your program/script or to another file to be read. Of course you cloud define some function to handle most of these but this is a one and done script. For this to work you will need to provide your API username:password in base64 format in the Generate Access Token section in the variable 'encoded_base64' and under the Query section the "nodes":["host:Name_of_your_host_here"] host name of your host dropping the domain name.

# URL to query
token_url = 'https://orbital.amp.cisco.com/v0/oauth2/token'
query_url = 'https://orbital.amp.cisco.com/v0/query'
 
############ Generate Access Token ##################
encoded_base64 = 'Your API username:password in base64 format'
headers = {
    'Authorization':'Basic {}'.format(encoded_base64),
    'Content-Type':'application/json', 'Accept':'application/json'
}
request_token = requests.post(token_url, headers=headers)
token_dict = request_token.json()
new_token = token_dict['token']
new_token_expiry = token_dict['expiry']
 
# Display token information
print('\nYour new token has been generated and can be found below:'
      '\n Your Token: {} '
      '\n Your Token Expires in ~10mins: {} UTC'
      '\n Current local time:            {}'
      '\n Current local time in Epoch:   {}'.format(new_token, new_token_expiry, time.asctime(), round(time.time())))
 
# Set new token to access token
access_token = new_token
###################### END ############################
 
##################### Query ##########################
# Headers to send with POST
headers = {
    'Authorization':'Bearer {}'.format(access_token),
    'Content-Type':'application/json', 'Accept':'application/json'
}
 
# Craft query/job expiry, must be 60 seconds or longer
timer_expiry = int(time.time())+60
 
payload = {
    "osQuery":[{"sql": "SELECT * FROM processes;"}],
    "nodes":["host:Name_of_your_host_here"],
    "expiry": timer_expiry,
}
# Format Payload before and after
payload = json.dumps(inspect_payload)
response = requests.post(query_url, headers=headers, data=payload)
 
# Display response code and json formatted result
print("\nHTTP response code: ", response)
 
#Store Response in dictionary for later parsing
response_results = response.json()
 
# Print JOB ID and Expiry
print("Here is the Job ID: ", response_results['ID'])
print("Epoch Job Expiry (60 seconds): ", response_results['expiry'])
###################### END ##############################

Your results should be as below. My Token has been truncated.

Your new token has been generated and can be found below:
 Your Token: eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCIsImtpZCI6IkhqVW4yNlBPUGZlWFFxeDEtcEc3TFU1MnBNRTRVMVlySWlJa29fUTJMV0kifQ...
 Your Token Expires in ~10mins: 2020-06-30T14:28:19.917169068Z UTC
 Current local time:            Tue Jun 30 10:18:20 2020
 Current local time in Epoch:   1593526701
 
HTTP response code:  <Response [200]>
Here is the Job ID:  SzhmRE_SUW6ENhtdO8k0Eg
Epoch Job Expiry (60 seconds):  1593526760

Some things I want to point out here. You can see the the that we can tell the token expires after 10 minutes by subtracting our current time from the UTC time. I am located in EST/EDT (-4 GMT) so make sure you adjust the calculation for your timezone. You can also verify how long the job is suppose to run by subtracting the return Epoch time.

'ID'	The Job ID for the query
'done_count'	How many times the job has been completed
'submission'	Epoch time job was submitted in seconds
'update'	Epoch time job was updated in seconds
'expiry'	Epoch time job will expire in seconds
'internal'	How many times the job should run

Checking a Job

Now that we have a job running let us go ahead and check in on it!

# Open file, read, close file
temp_file = open('Temp_token.txt', 'r')
new_token = temp_file.read()
temp_file.close()
 
temp_jobid = open('temp_jobid.txt', 'r')
jobid = temp_jobid.read()
temp_jobid.close()
 
# URL to query
job_url = 'https://orbital.amp.cisco.com/v0/jobs/{}/results'.format(jobid)
print("Completed URL: ", job_url)
 
# Headers
headers = {
    'Authorization': 'Bearer {}'.format(new_token),
    'Content-Type': 'application/json', 'Accept': 'application/json'
}
 
# Send GET
job_request = requests.get(job_url, headers=headers)
 
# Print Results
print("HTTP Response Code: ", job_request)
print("Job Response: ", job_request.json())
 
# Store response for later use
job_results = job_request.json()

This job is running the osQuery":[{"sql": "SELECT * FROM processes;"}] which will return a list of all running process from your host. I've had to truncate it a lot in order to maintain space but still show you what you should get back. Its a lot so this section wont have a Key list.

HTTP Response Code:  <Response [200]>
Job Response:  {'results': [{'node': 'ZsWurlZxOsYARYqXDsFsFA', 'osQuery': [{'sql': 'SELECT * FROM processes;'}], 'osQueryResult': [{'types': ['', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', '', 'user', '', '', '', '', '', '', '', '', '', '', '', ''], 'columns': ['pid', 'name', 'path', 'cmdline', 'state', 'cwd', 'root', 'uid', 'gid', 'euid', 'egid', 'suid', 'sgid', 'on_disk', 'wired_size', 'resident_size', 'total_size', 'user_time', 'system_time', 'disk_bytes_read', 'disk_bytes_written', 'start_time', 'parent', 'pgroup', 'threads', 'nice', 'is_elevated_token', 'elapsed_time', 'handle_count', 'percent_processor_time'], 'values': ['0', '[System Process]', '', '', '', '', '', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '-1', '0', '-1', '4', '-1', '', '-1', '-1', '-1', '4', 'System', '', '', 'STILL_ACTIVE', '', '', '18', '18', '-1', '-1', '-1', '-1', '-1', '272', '32768', '196608', '0', '6878640', '34195004209', '115241623524', '1592484661', '0', '-1', '153', '32', '1', '1049653', '3031', '68786406250', '112', 'Registry', '', '', 'STILL_ACTIVE', '', '', '18', '18', '-1', '-1', '-1', '-1', '-1', '10472', '16621568', '4825088', '0', '20390', '2048', '2050449408', '1592484652', '4', '-1', '4', '32', '0', '1049662', '0', '203906250', '420', 'smss.exe', 'C:\\Windows\\System32\\smss.exe', '\\SystemRoot\\System32\\smss.exe', 'STILL_ACTIVE', 'C:\\Windows\\System32\\smss.exe', 'C:\\Windows\\System32\\smss.exe', '18', '18', '-1', '-1', '-1', '-1', '1', '3224', '397312', '1081344', '15', '296', '1297412', '25056', '1592484661', '4', '-1', '2', '32', '1', '1049653', '53', '3125000', '520', 'csrss.exe', 'C:\\Windows\\System32\\csrss.exe', '%SystemRoot%\\system32\\csrss.exe ObjectDirectory=\\Windows SharedSection=1024,20480,768 Windows=On SubSystemType=Windows ServerDll=basesrv,1 ServerDll=winsrv:UserServerDllInitialization,3 ServerDll=sxssrv,4 ProfileControl=Off MaxRequestThreads=16', 'STILL_ACTIVE', 'C:\\Windows\\System32\\csrss.exe', 'C:\\Windows\\System32\\csrss.exe', '18', '18', '-1', '-1', '-1', '-1', '1', '21952', '1826816', '1822720', '31812', '250687', '630004', '0', '1592484665', '508', '-1', '11', '32', '1', '1049649', '631', '2825000000', '600', 'wininit.exe', 'C:\\Windows\\System32\\wininit.exe', 'wininit.exe', 'STILL_ACTIVE', 'C:\\Windows\\System32\\wininit.exe', 'C:\\Windows\\System32\\wininit.exe', '18', '18', '-1', '-1', '-1', '-1', '1', '11512', '1773568', '1470464', '15', '109', '1599296', '0', '1592484665', '508', '-1', '1', '128', '1', '1049649', '161', '1250000', '616', 'csrss.exe', 'C:\\Windows\\System32\\csrss.exe', '%SystemRoot%\\system32\\csrss.exe ObjectDirectory=\\Windows SharedSection=1024,20480,768 Windows=On SubSystemType=Windows ServerDll=basesrv,1 ServerDll=winsrv:UserServerDllInitialization,3 ServerDll=sxssrv,4 ProfileControl=Off MaxRequestThreads=16'....

Conclusion

There you have it. You've gone through all of the API for orbital. You've generated your API credentials, generated your token, verified that token, created a job and returned that result! I highly recommend that you check out the Orbital API for a more formal read.