Continuing the discussion from NEW Workflows API Functionality!:
Along with the NEW Workflow API Functionality, we have provided a couple of useful python scripts to get you started with Get, List, Activate & Deactivate Workflow!
List Workflow: This script accepts a user key and any filters you’d like and returns a list of workflows from your organization using the List Workflow endpoint.
import requests, json
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
def main():
# Parse command line arguments
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
parser.add_argument("-n", "--name", help="Filter by workflow name", required=False)
parser.add_argument("-s", "--state", help="Filter by the state of the workflow", required=False)
parser.add_argument("-t", "--tags", help="Filter by tag names. Parameter may be included multiple times, and workflow will only be included if all of the given tags match", required=False)
parser.add_argument("-l", "--limit", default=30, type=int, help="The number of items to return", required=False)
parser.add_argument("-o", "--offset", default=0, type=int, help="The number of records to skip", required=False)
required = parser.add_argument_group('required arguments')
required.add_argument("-k", "--key", help="The user api key needed to authenticate class1 API requests", required=True)
args = vars(parser.parse_args())
# Set up User API key
headers = {'X-API-KEY': args["key"]}
# List of Tags
tags = args["tags"]
if tags:
listOfTags = tags.split(',')
args["tags"] = listOfTags
#Build URL
regionInput = input("What region?\n1) eu\n2) us\n3) ap\n4) ca\n5) au\n6) us2\n7) us3\nEnter a number 1-7: ")
if regionInput == "1":
region = "eu"
elif regionInput == "2":
region = "us"
elif regionInput == "3":
region = "ap"
elif regionInput == "4":
region = "ca"
elif regionInput == "5":
region = "au"
elif regionInput == "6":
region = "us2"
elif regionInput == "7":
region = "us3"
else:
print("A region must be provided. Please try again.")
return
# Get Workflow V2 API
url = "https://%s.api.insight.rapid7.com/connect/v2/workflows" % region
# List Workflows V2 API
r = requests.get(url, params=args, headers=headers)
print("Print the url to check the URL has been correctly encoded or not!")
print("URL: ", r.url)
print("Status Code: ", r.status_code)
response = json.dumps(r.json(), indent=4, ensure_ascii=False)
print(response)
main()
Get, Activate, Deactivate Workflow: This interactive script accepts a user key and a workflow id. First it will show you details about the workflow with the Get Workflow endpoint and proceeds to ask if you would like to activate, deactivate, or reactivate the inactive published version using the Activate and Deactivate Workflow endpoints.
import requests, json, re
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
def main():
# Parse command line arguments
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
required = parser.add_argument_group('required arguments')
required.add_argument("-i", "--id", help="The unique ID of the workflow", required=True)
required.add_argument("-k", "--key", help="The user api key needed to authenticate API requests", required=True)
args = vars(parser.parse_args())
# Set up User API key
headers = {'X-API-KEY': args["key"]}
# Build URL
workflowID = args["id"]
regionInput = input("What region?\n1) eu\n2) us\n3) ap\n4) ca\n5) au\n6) us2\n7) us3\nEnter a number 1-7: ")
if regionInput == "1":
region = "eu"
elif regionInput == "2":
region = "us"
elif regionInput == "3":
region = "ap"
elif regionInput == "4":
region = "ca"
elif regionInput == "5":
region = "au"
elif regionInput == "6":
region = "us2"
elif regionInput == "7":
region = "us3"
else:
print("A region must be provided. Please try again.")
return
# Get Workflow V2 API
getWorkflowUrl = "https://{0}.api.insight.rapid7.com/connect/v2/workflows/{1}".format(region, workflowID)
r = requests.get(getWorkflowUrl, headers=headers)
print("Print the url to check the URL has been correctly encoded or not!")
print("URL: ", r.url)
print("Status Code: ", r.status_code)
getWorkflowResponse = json.dumps(r.json(), indent=4, ensure_ascii=False)
print(getWorkflowResponse)
# Bail out of Get Workflow returns anything besides a 200
if r.status_code != 200:
return
# User Decision about activate/deactivate/reactivate the workflow
activateOrDeactivate = None
params = None
userDecision = input("\nWhat would you like to do with this workflow?\n1) Activate\n2) Deactivate\n3) Reactivate\nPlease enter 1, 2 or 3: ")
if userDecision == "1":
print("Are you sure you want to activate this workflow?")
activateDecision = input("Y or N?").upper()
if activateDecision.upper() == "Y":
activateOrDeactivate = "activate"
elif userDecision == "2":
print("Are you sure you want to deactivate this workflow?")
deactivateDecision = input("Y or N?").upper()
if deactivateDecision == "Y":
activateOrDeactivate = "deactivate"
elif userDecision == "3":
print("Are you sure you want to reactivate?\nNote: If the workflow has an inactive version and unpublished changes," +
"the unpublished changes will be published and activated; to bypass publishing unpublished changes on activate, " +
"set the reactivate query parameter to true and this will reactivate the inactive published version.")
reactivateDecision = input("Y or N?").upper()
if reactivateDecision == "Y":
activateOrDeactivate = "activate"
params = {"reactivate": True}
else:
print("Do you want to active this workflow?")
activateDecision = input("Y or N?").upper()
if activateDecision == "Y":
activateOrDeactivate = "activate"
else:
print("Exit")
return
if activateOrDeactivate:
# List Workflows V2 API
url = "https://{0}.api.insight.rapid7.com/connect/v2/workflows/{1}/{2}".format(region, workflowID, activateOrDeactivate)
r = requests.post(url, params=params, headers=headers)
print("Print the url to check the URL has been correctly encoded or not!")
print("URL: ", r.url)
print("Status Code: ", r.status_code)
response = json.dumps(r.json(), indent=4, ensure_ascii=False)
print(response)
main()