Wiz Webhook with Netskope SSE

Wiz Webhook with Netskope SSE

Wiz works to secure cloud infrastructure, using agentless scanning to give cloud security teams immediate visibility into the most critical security risks that need to be addressed. Leveraging its seamless integration with Netskope, organizations gain real-time insights and proactive defense mechanisms against evolving cloud threats. Together, Wiz and Netskope form an unbeatable alliance in safeguarding cloud environments with unparalleled efficiency and precision. This document encompasses a detailed step by step guide for integrating Wiz Issues to influence Netskope NPA policies.


For this integration, you need:

  • An AWS Admin account for provisioning lambda.
  • A Netskope Tenant.
  • A Netskope API Key.
  • A Username for authenticating Wiz Webhooks.
  • A Password for authenticating Wiz Webhooks).

Wiz Overview

Wiz secures everything organizations build and run in the cloud. Wiz connects to every cloud environment, scans every layer, and covers every aspect of your cloud security, including elements that normally require installing agents. Its comprehensive approach has all of these cloud security solutions built in.

Joint Use Cases

  • Secure private access to AWS resources leveraging Wiz Issues.
  • Unified control plane between customer security providers.

Joint Solution

The integration takes advantage of the deep visibility that Wiz has into cloud environments and context of which exposures lead to critical attack paths and allows Netskope to enforce policy to the end user.

The solution leverages Wiz Webhooks along with the Netskope platform API.

A lambda function is deployed in the customer tenant along with an API Gateway to handle the Webhooks. The lambda function authenticates to the Netskope ZTNA (Private Access) APIs.

Wiz provide a list of hostnames that are currently experiencing any OPEN issues.

The lambda function will take the hostnames that is learned from Wiz Webhook and see if the hostname is configured for allow in any Real-time policies. If the hostname is seen there will be a API call placed to add the hostname to a private application and blocked by Netskope.

Deployment Steps

  1. Generate a REST API v2 token from your Netskope tenant. You need to add the following endpoints to your scope:
    For more information about creating a Netskope API token and adding endpoints to the scope, go to REST API v2 Overview.
  2. When you have the token information, you can deploy the lambda function in your AWS tenant.
    1. Log in to your AWS console (https://console.aws.amazon.com/).
    2. Go to the AWS service Lambda.
    3. Deploy a new Lambda function in management console of AWS. Create from scratch or choose the latest python version, and leave all default configuration variables.
    4. After the Function is created, go to Code and paste the following code in to your function:
      import os
      import json
      import urllib3
      # fetching environmental variables
      netskope_api_token = os.environ['netskope_api_token'] 
      netskope_tenant = os.environ['netskope_tenant']
      username = os.environ['username']
      password = os.environ['password']
      http = urllib3.PoolManager()
      # creating python function to get all NPA Publishers from the tenant
      def get_npa_publishers():
          url = f"https://{netskope_tenant}.goskope.com/api/v2/infrastructure/publishers"
          headers = {'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}
          response = http.request('GET', url, headers=headers)
          response_json = json.loads(response.data.decode('utf-8'))
          return [[x['publisher_id'],x['connected_apps']] for x in response_json['data']['publishers'] if x['apps_count'] > 0]
      # creating python function to get all NPA Apps for a given publisher ID
      def get_npa_publisher_apps(pub_id):
          url = f"https://{netskope_tenant}.goskope.com/api/v2/infrastructure/publishers/{pub_id}/apps"
          headers = {'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}
          response = http.request('GET', url, headers=headers)
          response_json = json.loads(response.data.decode('utf-8'))
          return [response_json['data'][0]['host']]
      # creating python function to get all NPA rules from the tenant
      def get_npa_rules():
          url = f"https://{netskope_tenant}.goskope.com/api/v2/policy/npa/rules"
          headers = {'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}
          response = http.request('GET', url, headers=headers)
          return json.loads(response.data.decode('utf-8'))
      # creating python function to get all NPA private apps from the tenant
      def get_npa_apps():
          url = f"https://{netskope_tenant}.goskope.com/api/v2/steering/apps/private"
          headers = {'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}
          response = http.request('GET', url, headers=headers)
          return json.loads(response.data.decode('utf-8'))
      # creating python function to update a NPA rule for the tenant
      def create_npa_rule(create_npa_rule_data):
          url = f"https://{netskope_tenant}.goskope.com/api/v2/policy/npa/rules"
          headers = {'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}
          response = http.request('POST', url, headers=headers, body=json.dumps(npa_policy_data))
          return json.loads(response.data.decode('utf-8'))
      # creating python function to patch a private app to add hostnames
      def post_wiz_private_app():
          # Make the request
          response = http.request("POST", f"https://{netskope_tenant}.goskope.com/api/v2/steering/apps/private",\
          headers={'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}, \
              "app_name": "Wiz_Hosts_w_Issues",
              "host": "wiz_default.io",
              "protocols": [
                  {"type": "tcp", "port": "80"},
                  {"type": "tcp", "port": "443"}
          # Decode and return the response
          return response.data.decode('utf-8')
      # creating python function to update private app in NPA
      def update_wiz_private_app(host_list, app_id_to_update):
          # Make the request
          response = http.request("PATCH", f"https://{netskope_tenant}.goskope.com/api/v2/steering/apps/private/{app_id_to_update}",\
          headers={'Accept': 'application/json', 'Netskope-api-token': netskope_api_token}, \
              "app_name": "Wiz_Hosts_w_Issues",
              "host": host_list,
              "protocols": [
                  {"type": "tcp", "port": "80"},
                  {"type": "tcp", "port": "443"}
          # Decode and return the response
          return response.data.decode('utf-8')
      # creating function to look for username and password in webhook 
      def check_http_basic_auth(headers):
          if str(username) not in str(headers) or str(password) not in str(headers):
              return {
                  'statusCode': 401,  # Unauthorized status code
                  'body': 'Unauthorized access'
      def lambda_handler(event,context):
          # Check if 'body' key exists in the event dictionary
          if 'body' in event:
              # Check if the 'body' is not None and not an empty string
              if event['body'] and event['body'].strip():
                  # 'body' is not empty
                  # Proceed with processing the payload
                  payload = event['body']
                  # Your code to handle the non-empty payload
                  # 'body' is empty
                  # Handle the case where payload is empty
                  return {
                      'statusCode': 400,
                      'body': 'Empty payload received'
              # 'body' key does not exist in event dictionary
              # Handle the case where 'body' is missing
              return {
                  'statusCode': 400,
                  'body': 'No payload found in the request'
          # Parse the 'body' JSON data
          body_data = json.loads(event['body'])
          # Access elements from the parsed JSON data
          issueSeverity = body_data.get('issue', {}).get('severity')
          resource = body_data.get('resource', {}).get('id')
          issueStatus = body_data.get('issue', {}).get('status')
          # executing function to validate correct username and password for Webhook  Basic Auth
          # executing function to get all npa private apps for tenant
          npa_apps = get_npa_apps()
          # executing function to get all NPA policies for Netskope tenant
          npa_rules = get_npa_rules()
          # creating list of app IDs to update that will later be appended to
          app_id_to_update = []
          # conditional statement to validate that private app Wiz_Hosts_w_Issues already exists
          if "Wiz_Hosts_w_Issues" in str(json.dumps(npa_apps['data']['private_apps'])):
              print("Wiz_Hosts_w_Issues private app exists, continuing")
              # using list comprehension to get all existing application IDs (hostnames), will need these later
              app_id_to_update = [x['app_id'] for x in npa_apps['data']['private_apps'] if 'Wiz_Hosts_w_Issues' in str(x)]
              # if Wiz_Hosts_w_Issues private app does not exist we will create the app using the Netwskope API
              print(" need to create Wiz_Hosts_w_Issues private app")
              # executing function to create Wiz_Hosts_w_Issues private app
              private_app_response = json.loads(json.dumps(post_wiz_private_app()))
              # still need to collect placeholder private app names as discussed in post_wiz_private_app func
              app_id_to_update = json.loads(private_app_response)['data']['app_id']
          # validating the private app Wiz_Blocked_Hosts is inside the npa_rules
          if "Wiz_Hosts_w_Issues" in str(json.dumps(npa_rules)):
              print("Wiz Block Rule Exists, continuing")
              print("need to createWiz Block Rule")
              # need to create NPA Rule with Wiz_Hosts_w_Issues private app (executing create_npa_rule func)
              create_npa_rule_data = {"enabled": "1", "policy_type": "private-app", 
              "rule_data": {"access_method": ["Client"], "external_dlp": False, "json_version": 3, 
              "match_criteria_action": {"action_name": "block", "template": "wiztest"}, "policy_type": "private-app", 
              "privateApps": ["[Wiz_Hosts_w_Issues]"], "privateAppsWithActivities": [{"activities": [{"activity": "any", "list_of_constraints": []}], 
              "appName": "[Wiz_Hosts_w_Issues]"}], "show_dlp_profile_action_table": False, "userType": "user", "version": 1}, 
              "rule_id": "3", "rule_name": "Wiz_Hosts_w_Issues", "rule_order": {"order": "top"}}
              create_rule_response = create_npa_rule(create_npa_rule_data)
          # creating clean list of all hostnames from npa_apps
          npa_apps_hostnames = [hostname['host'] for hostname in npa_apps['data']['private_apps']]
          npa_apps_hostnames_list = []
          final_host_list = []
          for host in npa_apps_hostnames:
              hostname = host.split(',')
              if len(hostname) > 1:
                  for host2 in hostname:
          # checking if resource is in the npa_apps_hostnames_list
          resource_id_from_webhook = str(resource).split('instance/')
          # Take the last part of the resulting list
          resource = resource_id_from_webhook[-1]
          if str(resource) in str(npa_apps_hostnames_list) and 'open' == str(issueStatus).lower():
              # if resource is in apps list and the issue status is OPEN see if it is already in Wiz_Host_With_Issues private app
              wiz_hosts_with_issues_list = [x['host'] for x in npa_apps['data']['private_apps'] if 'Wiz_Hosts_w_Issues' in str(x)]
              if str(resource) in str(wiz_hosts_with_issues_list):
                  print(f"resource {resource} is already blocked, no action needed")
                  print(f"resource {resource} is not blocked, updating NPA private app for Wiz_Hosts_w_Issues")
                  wiz_hosts_with_issues = str(wiz_hosts_with_issues_list[0]).split(',') + [resource]
                  wiz_hosts = ', '.join(wiz_hosts_with_issues)
                  patch_host_list_response = update_wiz_private_app(wiz_hosts, app_id_to_update[0])
          # now the function will perform a update to NPA private Apps if issueStatus = CLOSED
          if str(issueStatus).lower() == 'closed':
              # updating npa private apps to remove resource
              wiz_hosts_with_issues_list = [x['host'] for x in npa_apps['data']['private_apps'] if 'Wiz_Hosts_w_Issues' in str(x)]
              wiz_hosts_with_issues = str(wiz_hosts_with_issues_list[0]).replace(resource, '').replace(', ,', ',').strip(', ')
              patch_host_list_response = update_wiz_private_app(wiz_hosts_with_issues, app_id_to_update[0])
          # prepare response body
          response_body = {}
          #response_body['npa_apps_hostnames_list'] = npa_apps_hostnames_list
          response_body['issueSeverity'] = issueSeverity
          response_body['resource'] = resource
          response_body['header_info'] = event['headers']
          # prepare http response
          http_result = {}
          http_result['statusCode'] = 200
          http_result['headers'] = {}
          http_result['headers']['Content-Type'] = 'application/json'
          http_result['body'] = json.dumps(response_body)
          return http_result
    5. Configure the environment variables to authenticate the Wiz Webhook and Netskope API:
  3. Click Edit.
  4. Build a Rest API Gateway:
    1. Select Add Trigger in the lambda dashboard.
    2. Select API Gateway as the source.


      BASIC HTTP Authentication is done on lines 93 in the code pasted in the previous step.

    3. Go to the API Gateway to make sure the API Endpoint Wiz will use in the integration.
    4. When there, click the hyperlink for the API endpoint.

    Test the Integration

    Now that the function and API Gateway are deployed, you can test the endpoint to validate to see the policy change in the Netskope Dashboard, as well as authentication.

    Lambda is looking for 3 things inside the webhook: issueSeverity , issueStatus , and resource.

    It also tries to authenticate the Webhook using Basic Auth, so you need the username and Password specified in the lambda environment variable to be used.

    If the Username and Password in the Headers does not match then a 401 unauthorized error is returned as shown:

Validate the Integration

After successfully integrating a webhook in the API GW Dashboard, you can validate the policies and private apps are seen in the Netskope One platform.

Log in to the Netskope tenant and go to Policies > Real-Time Protection and ensure you see Wiz_Hosts_w_Issues rule at the top.

Also check under the Destination that the private application Wiz_Hosts_w_Issues exists. (The function should automatically create this rule and private application.)

Go to Settings and Security Cloud Platform > App Definition > Private Apps and ensure that you see Wiz_Hosts_w_Issues.

Click into the private app Wiz_Hosts_w_Issues and ensure you see the default placeholder hostname along with the hostname of the resource from the test performed in the API Gateway in previous step.

In the Wiz dashboard you can fill in the Basic Auth parameters and the API Gateway URL from AWS.

For the URL you need to locate that from the API Gateway in the AWS console: (Invoke URL)

After inputting the Username and Password for Basic Auth, you can also filter to what Issue Types will be sent in the Webhooks. For this we recommend anything with a hostname that Netskope clients can have access to, like vulnerability, malware, unpatched software, External Misconfiguration, etc.

Since this toolkit only cares about issues when they are OPEN or RESOLVED, you can filter the amount of webhooks that are sent to the API GW. However, the code has been made to ignore other webhooks as well:

Share this Doc

Wiz Webhook with Netskope SSE

Or copy link

In this topic ...