continous deployment to splunk anywhere

Recently we build an app which should be deployed on splunk Enterprise as well as on Splunk Cloud. Since we were among the first engineers to use the splunk Cloud Victoria Experience to deploy our components independently – this was a good occasion to explore the state of the art again.

First point: the integration with gitlab via a python deploy skript works like a charm, it is worth the effort.

The CI/CD process to splunk Cloud contains several steps and follows the Splunk App Inspect documentation i.e. the app is tested first with cypress on sample data, generated by eventgen, then it is tested by appinspect and then the splunk cloud service will test it before you get a token to deploy it. There are sample projects on github, that we used as a starting point.

Our contribution is the working python script which performs the deployment steps above to on premise installations or splunk cloud.

The first is on premise:


import splunklib.client as client

def deployOnPremise( file_path, user, pwd, token ):
   print("This deploys:" + file_path)
   uri = "https://splunk.bln.uneedsecurity.com"

   # check arguments...

   # show if gitlab variable masking works
   print("authenticate to splunk rest api with:", user, " pwd:", pwd, " token:", token)

   service = client.connect(host="splunk1.bln.uneedsecurity.com", port=8089, username=user, password=pwd)
   assert isinstance(service, client.Service)
   print("auth successfull")
   params = {'name':'/opt/splunk/gitlab-apps/my_new_app.tgz','filename':'true','update':'true'}
   service.post('apps/local/create',**params)
   print("deployment successfull")

   return 0

The second is for the cloud deployment:


import sys, getopt
import os
from time import sleep
from requests_toolbelt.multipart.encoder import MultipartEncoder
import requests
from requests.auth import HTTPBasicAuth

def deploy2Cloud( file_path, user, pwd, token ):
   print("This deploys:" + file_path)
   print("authenticate to use splunk rest api")
   
   uri = "https://api.splunk.com/2.0/rest/login/splunk"

   # check arguments...

   # show if gitlab variable masking works
   print("authenticate to use splunk rest api with:", user, " pwd:", pwd, " token:", token)
   
   response = requests.get(uri, auth=HTTPBasicAuth(user, pwd))
   print("response:" + response.content.decode())
   user_token = response.json().get("data").get("token")

   print("get validation from appinspect")

   url = "https://appinspect.splunk.com/v1/app/validate"
   app_name = os.path.basename(file_path)

   fields = {}
   fields.update({"app_package": (app_name, open(file_path, "rb"))})
   fields.update({"included_tags": "private_app"})
   payload = MultipartEncoder(fields=fields)

   headers = {"Authorization": "Bearer {}".format(user_token), "Content-Type": payload.content_type}
   response = requests.request("POST", url, data=payload, headers=headers)

   print(response.status_code)
   print(response.json())
   request_id = response.json().get("request_id")

   print("wait on validation status for request_id " + request_id)
   counter = 0
   while True:
      if (counter > 20): 
         return 10
      
      uri = "https://appinspect.splunk.com/v1/app/validate/status/" + request_id
      headers = {"Authorization": "bearer {}".format(user_token), "Content-Type": payload.content_type}
      response = requests.request("GET", uri, headers=headers)
   
      status = response.json().get("status")
      print(response.json())

      if status == "ERROR":
          print("error on validation status.")
          return 11
      if( status == "SUCCESS"):
          print("success on validation status.")
          break
      sleep(10) # wait another 10 seconds
      counter += 1

   print("deploy with appinspect_token and bearer token")

   url = "https://admin.splunk.com/uneedsecurity-cb/adminconfig/v2/apps/victoria?splunkbase=false"
   auth = "Bearer " + token
   headers = {"Authorization": auth, "ACS-Legal-Ack":"Y", "X-Splunk-Authorization": "{}".format(user_token),"Content-Type": "application/x-www-form-urlencoded"}
   response = requests.post(url, data=open(file_path, 'rb'), headers=headers)
   
   print(response.status_code)
   print(response.json())
   return 0

For splunk cloud deploymentd you get already the status information:


{'appID': 'my_new_app', 'label': 'My new App', 'name': 'rhebo_app', 'status': 'installed', 'version': '1.0.0'}

The status ‘installed’ means, that the app passed the splunk cloud tests.

We as UNeedSecurity support customers who enhance their security automation on their own. This is in our opinion the first recommended way. But, if there are no inside specialists available, we do it for you as part of your threat management contract based on our daily rates.