Updated on February 11, 2025

Risk Exchange Custom Plugin Developers Guide

Risk Exchange Custom Plugin Developers Guide

This guide explains how to write a new Risk Exchange plugin and extract maximum value out of your risk ecosystem by leveraging the functionality provided within the Risk Exchange module. The developer should be able to write a new plugin independently without any technical issues by following this guide.

Prerequisites

  • Python 3.x programming experience (intermediate level).
  • Access to the Netskope Cloud Exchange platform.
  • API or Python SDK access to the product or solution for which you need to write the plugin.
  • An account with minimum permissions for the product.

Compatibility

This plugin development guide is specific for the development of Risk Exchange plugins supported for core version v5.1.0.

Risk Exchange Module

The Cloud Exchange platform, and its Risk Exchange module, comes with a rich set of features and functionality that allow for a high degree of customization, so we recommend that you familiarize yourself with the different aspects of the platform as listed below.

This module supports pulling records from third-party platforms and performs actions like add user to group on the platform.

Netskope Concepts & Terminology

  • Core: The Cloud Exchange core engine manages the 3rd party plugins and their life cycle methods. It has API endpoints for interacting with the platform to perform various tasks.
  • Module: The functional code areas that invoke modular-specific plugins to accomplish different workflows. Risk Exchange is one of the modules running in Cloud Exchange.
  • Plugin: Plugins are Python packages that have logic to fetch users, devices or applications details and risk/threat information from 3rd party Threat Intel systems, which will then be stored in Risk Exchange. Risk Exchange plugins also perform actions like add user to group, remove user from group, or other actions on device records or applications records if supported on a third-party platform.
  • Plugin Configurations: Plugin configurations are the plugin class objects which are configured with the required parameters and are scheduled by the Cloud Exchange core engine for fetching users and scores.
  • Schema Editor: Users with write access can manage Risk Exchange schemas and entity fields from Schema Editor.
  • Entity: Entity is collection of fields of specific schema and records are stored in this Entity. Entities can be managed from the Schema Editor page of the Risk Exchange module.
  • Records: Records (of users, devices or applications) are objects having fields mapped in Entity Sources step in plugin configuration gathered from third-party platforms and stored in the mapped Entity in Cloud Exchange database.

Development Guidelines

  • Use the Plugin Directory Structure for all Python code.
  • Make sure all the thirdparty libraries packaged with the plugin package are checked for known vulnerabilities.
  • Make sure to follow standard Python code conventions. (https://peps.python.org/pep-0008/)
  • Run and verify that the flake8 Lint check passes with the docstring check enabled. The maximum length of a line should be 80.
  • Convert the time-stamp values to the human-readable format (from epoch to DateTime object). Make sure the time that is being displayed on the UI should be in the local timezone.
  • If possible, add a default value while adding a configuration parameter in the plugin.
  • Plugin architecture allows storing states, however, avoid storing huge objects for state management.
  • Check your python code for any vulnerabilities.
  • The plugin icon has to be under 10kb. Make sure to use the company logo(and not the product’s) with a transparent background. The recommended size for the logo is 300×50 or a similar aspect ratio.
  • Use the checkpoint provided by the CE core rather than implementing one on your own.
  • The logger messages and the Toast messages should not contain the API Token and the Password type field values.
  • Pagination should always be considered while developing any feature in a plugin.
  • Make sure to add a retry mechanism for the 429 status code.
  • Make sure to implement a proper logging mechanism with the logger object passed by the CE platform. Make sure enough logging is done which helps the Ops team in troubleshooting. Make sure any sensitive data is not logged or leaked in the notification.
  • Provide the proper help text (tooltip) for all the parameters. If feasible, make sure to explain the significance of the parameter in the tooltip.
  • Make sure to provide a meaningful name and description to plugin configuration parameters.
  • If the description contains a link, it should be a hyperlink instead of plaintext.
  • Make sure to provide an appropriate configuration type (text, number, password, choice, multi-choice) to the parameters.
  • Make sure to use the proxy configuration dict and the SSL certificate validation flag that is passed by the CE platform while making any outbound request (API/SDK).
  • Make sure to collect the value of a non-mandatory parameter using the .get() method and provide a default value while using the .get() method.
  • API Tokens and Password fields should not use strip().
  • The log messages should start with “<module> <plugin_name> [configuration_name]: “. Example: “CRE Crowdstrike [CrowdStrike Configuration Name]: <log_message>“. (This is a suggestion, we can avoid configuration names). [logger.info(“<module> <plugin_name>: <message>”)]
  • While logging an error log, if possible, we should add a traceback of the exception. USE: logger.error(error, details=traceback.format_exc())
  • The Toast message should not contain the <app_name> <module>: in the message.
  • Make sure to catch proper exceptions and status codes while and after making the API calls. If feasible, Devs can create a helper method from where the requests would be made and this method can be called with proper parameters wherever required.
  • Make sure to use API helper to execute the third party API calls. Refer code snippet section of fetch_records for API Helper example. Check reference for helper file (https://github.com/netskopeoss/ta_cloud_exchange_plugins/blob/main/crowdstrike_ztre/utils/helper.py)
    • For Bearer Token authentication, reload auth token on status code 401 Unauthorized for every API request, except for requests in validation methods. Maintain flag to indicate when to reload auth token in API Helper.
    • Handle the exceptions.ReadTimeout error in API helper.
  • The CHANGELOG.md file should be updated with proper tags such as Added, Changed, and Fixed along with a proper user-friendly message. Make sure the file name should exactly matches CHANGELOG.md
  • For Action parameters validation:
    • If the action parameter is selected from the Source field then validate value for that field in the execute_action
    • If the action parameter is of type `choice`, the validation message should be: “{Parameter} contains the Source Field. Please select {field} from the Static Field dropdown only.”
  • If the plugins support any datetime field, then convert that field into a datetime object in the fetch_records method while returning records.
  • Make sure to return records with the value of all the plugin supported fields received from API calls with. Fields like Score in Records fields make more sense to the SOC user when analyzing the data. Make sure to map the Netskope Normalized Score field which gives more context to the SOC analyst. The Netskope Normalized Score field should be an integer.
  • Use proper validation for the parameters passed to the validate method and provide the proper help text for all the parameters.
  • Make sure the plugin directory name (e.g sample_plugin) matches with the manifest.json’s id.
  • User Agent should be added to the headers while making any API call. Format for the User Agent: netskope-ce-<ce_version>-<module>-<plugin_name>-<plugin_version>.
  •  

Writing a Plugin

This section illustrates the process of writing a plugin from scratch.

Download the sample plugin from the NetskopeOSS public Github repository or from the Cloud Exchange Knowledge Base found here: https://support.netskope.com/hc/en-us/articles/360052128734-Cloud-Threat-Exchange.

Development setup

Python

Our system utilizes Python3 (v3.8 and above). Make sure to set up python3 in your development environment. Pytest is used to run unit tests.

Included Python Libraries

These Python libraries are included within the Netskope Cloud Exchange platform.

Library Name Version
aiofiles 22.1.0
amqp 5.1.1
annotated-types 0.5.0
anyio 3.6.2
asgiref 3.6.0
attrs 22.2.0
azure-core 1.26.2
azure-storage-blob 12.14.1
bcrypt 4.0.1
billiard 4.2.0
boto3 1.26.51
botocore 1.29.51
cabby 0.1.23
cachetools 5.2.1
celery 5.3.6
certifi 2024.7.4
cffi 1.15.1
chardet 5.1.0
charset-normalizer 3.0.1
click 8.1.3
click-didyoumean 0.3.0
click-plugins 1.1.1
click-repl 0.2.0
colorama 0.4.6
colorlog 6.7.0
cryptography 42.0.5
cybox 2.1.0.21
Cython 0.29.37
defusedxml 0.7.1
dnspython 2.6.1
docker 6.0.1
email_validator 2.2.0
fastapi 0.111.0
fastapi-cli 0.0.5
furl 2.1.3
google-api-core 2.11.0
google-auth 2.16.0
google-cloud-core 2.3.2
google-cloud-pubsub 2.13.12
google-cloud-pubsublite 1.6.0
google-cloud-storage 2.7.0
google-crc32c 1.5.0
google-resumable-media 2.4.0
googleapis-common-protos 1.58.0
grpc-google-iam-v1 0.12.6
grpcio 1.62.2
grpcio-status 1.51.1
gunicorn 22.0.0
h11 0.14.0
httpcore 1.0.5
httptools 0.6.1
httpx 0.27.0
hvac 1.1.0
idna 3.7
importlib-metadata 6.0.0
isodate 0.6.1
Jinja2 3.1.4
jmespath 1.0.1
jsonpath 0.82.2
jsonschema 4.17.3
kombu 5.3.7
libtaxii 1.1.119
lxml 5.3.0
markdown-it-py 3.0.0
MarkupSafe 2.1.2
mdurl 0.1.2
memory-profiler 0.61.0
mixbox 1.0.5
mongoengine 0.25.0
mongoquery 1.4.2
more-itertools 9.0.0
msrest 0.7.1
multidict 6.0.4
mypy-extensions 0.4.3
netskopesdk 0.0.38
numpy 1.26.2
oauthlib 3.2.2
onelogin 3.1.0
ordered-set 4.1.0
orderedmultidict 1.0.1
orjson 3.10.7
overrides 6.5.0
packaging 23.0
pandas 2.1.3
passlib 1.7.4
prompt-toolkit 3.0.36
proto-plus 1.22.2
protobuf 4.21.12
psutil 5.9.4
py-expression-eval 0.3.14
pyasn1 0.4.8
pyasn1-modules 0.2.8
pycparser 2.21
pydantic 2.4.0
pydantic_core 2.10.0
Pygments 2.18.0
pyhcl 0.4.4
PyJWT 2.6.0
pymongo 4.6.3
Pympler 1.0.1
pyparsing 3.0.9
pyrsistent 0.19.3
python-dateutil 2.8.2
python-dotenv 1.0.1
python-multipart 0.0.7
python3-saml 1.15.0
pytz 2022.7.1
PyYAML 6.0.1
requests 2.32.0
requests-oauthlib 1.3.1
rich 13.7.1
rsa 4.9
s3transfer 0.6.0
shellingham 1.5.4
six 1.16.0
sniffio 1.3.0
starlette 0.37.2
stix 1.2.0.11
taxii2-client 2.3.0
typer 0.12.3
typing-inspect 0.8.0
typing-utils 0.1.0
typing_extensions 4.12.2
tzdata 2023.3
ujson 5.10.0
urllib3 1.26.19
uvicorn 0.23.1
uvloop 0.19.0
vine 5.1.0
watchfiles 0.23.0
wcwidth 0.2.6
weakrefmethod 1.0.3
websocket-client 1.4.2
websockets 12.0
Werkzeug 3.0.3
wheel 0.44.0
xmlsec 1.3.14
zipp 3.19.1
Including Custom Plugin Libraries

Netskope advises bundling any of the third-party python libraries your plugin will need into the plugin package itself. Use the pip installer to achieve this bundling; it provides a switch that takes a directory as input. If the directory is provided, pip will install the packages into that directory.

For example, this command will install the “cowsay” package into the directory “lib”.

> pip install cowsay --target ./lib

For the official documentation on this, refer https://pip.pypa.io/en/stable/reference/pip_install/#cmdoption-t.

To import modules from the above lib folder, we should set the python system path for the custom module in __init__.py file of plugin package.

__init__.py
import sys
from pathlib import Path

src_path = Path(__file__).resolve()
src_dir = src_path.parent
sys.path.insert(0, str(src_dir / "lib"))

After setting the system path for the custom library, import the library as shown here:

import cowsay
IDE

Recommended IDEs are PyCharm or Visual Studio Code. Install flake8 linting and black formatter for plugin code consistency and readability. Refer to Flake8 Black.

Plugin Directory Structure

This section mentions the typical directory structure of a Risk Exchange Plugin.

/sample_plugin/
  ├── __init__.py
  ├── Changelog.md
  ├── icon.png
  ├── main.py
  └── manifest.json
  • __init__.py: Every plugin package is considered a python module by the CRE code. Make sure every plugin package contains the empty “__init__.py” file. Add code for setting system path for custom library for plugin when custom module is required for plugin. Refer to the __init__.py section above for more information.
  • CHANGELOG.md: This file contains the details about plugin updates and should be updated with proper tags such as Added, Changed, and Fixed along with a proper user-friendly message.
  • icon.png: Plugin icon logo, this will be visible in the plugin chiclet and configuration cards on the UI. The logo should have a transparent background with a recommended size of 300*50 pixels or a similar aspect ratio and under 10kb of size.
  • main.py: This python file contains the Plugin class containing the concrete implementation for the fetch records and update records, execute actions, validate, and validate actions methods.
  • manifest.json: Manifest file for the plugin package containing information about all the configurable parameters and their data types. This file has more information about the plugin integration as well.

The listed files here are mandatory for any plugin integration, but developers can add other files based on specific integration requirements.

The following files are not mandatory, but it is recommended to add them in the plugin folder.

  • /utils/constants.py: This file contains all the constants used in plugin implementation. Import constants in the required file from py file.
  • utils/helper.py: This file contains the plugin helper class containing the implementation for the api_helper, add user agent and other helper methods required in the plugin.

Check reference for helper file: https://github.com/netskopeoss/ta_cloud_exchange_plugins/blob/main/crowdstrike_ztre/utils/helper.py

Note

Make sure the plugin directory name (sample_plugin) matches with the manifest.json’s ID field.

Changelog.md

This is a file that contains details about plugin updates and should be updated with proper tags such as Added, Changed, and Fixed along with a proper user-friendly message.

  • Added: Use it when new features are added.
  • Fixed: Use it when any bug/error is fixed.
  • Changed: Use it when there is any change in the existing implementation of the plugin.
  • Removed: use it when any parameters or functionality is removed from the plugin

Sample Changelog.md

# 1.1.0
## Removed
- Removed priority from the syslog message for the logs that are not transformed in CEF.

# 1.0.1
## Changed
- Changed error logs to warning if a single field is skipped.
## Fixed
- Fixed JSON format of raw data. 

# 1.0.0
## Added
- Initial release.

Manifest.json

This is a JSON file that stores the meta-information related to the plugin, which is then read by the Risk Exchange module to render the plugin in the UI, as well as enabling the Risk Exchange module to know more about the plugin, including required configuration parameters, the plugin-id, the plugin-name, etc.

  • Every plugin must contain this file with the required information so that Risk Exchange can instantiate the Plugin object properly.
  • Common parameters for manifest.json include:
    • name: (string) Name of the plugin. (Required)
    • netskope:(Boolean) it is used to check if the plugin is a netskope plugin or not (example: True if it’s netskope plugin else False)
    • description: (string) Description of the plugin. Provide a detailed description that mentions the capabilities and instructions to use the plugin, (ex. Fetches users/devices details along with scores and performs actions. Also add Netskope Normalized Score formula if plugins support fetching scores from third party platforms.) This description would appear on the Plugin Configuration card. (Required) Check this(https://github.com/netskopeoss/ta_cloud_exchange_plugins/blob/main/crowdstrike_ztre/manifest.json) reference for description.
    • ID: (string) ID of the plugin package. Make sure it is unique across all the plugins installed in the Cloud Exchange. The ID has to match the directory name of the plugin package. (Required)
    • version: (string) Version of the plugin. Usage of a MAJOR.MINOR.PATCH (ex. 1.0.1) versioning scheme is encouraged, although there are no restrictions. (Required)
    • minimum_version: (string) Minimum Cloud Exchange version for this plugin to execute. 5.1.0 for Risk Exchange plugins. (Required)
    • module: (string) CE Module name for the plugin, like CRE, CTO, CTE, CLS. (Required)
    • configuration: (array) Array of JSON objects that contains information about all the parameters required by the plugin – their name, type, id, etc. The common parameters for the nested JSON objects are explained below.
      • label: Name of the parameter. This will be displayed on the plugin configuration page. (Required)
      • key: Unique parameter key, which will be used as a key in the python dict object where the plugin configuration is used. (Required)
      • type: Value type of the parameter. Allowed values are text, password, number, choice, and multichoice. (Required) Refer to the Plugin Configuration Parameter Types below for more details.
      • default: The default value for this parameter. This value will appear on the plugin configuration page on the Risk Exchange UI. Supported data-types are text, number, and list (for multichoice type). (Required)
      • mandatory: Boolean which indicates whether this parameter is mandatory or not. If a parameter is mandatory, the Risk Exchange UI won’t let you pass an empty value for the parameter. Allowed values are true and false. (Required)
      • description: Help text level description for the parameter which can give more details about the parameter and expected value. This string will appear on the plugin configuration page as a help text. (Required)
      • choices: A list of JSON objects containing key and value as JSON keys. This parameter is only supported by type: choice and multichoice.
Plugin Configuration Parameter types

Make sure all the required plugin configuration parameters are listed under the configuration section of manifest.json for the plugin.

Password Parameter

Use this parameter for storing any secrets/passwords for authentication with API endpoints. Parameters with type as the password will have a password text box in the Plugin configuration page and will be obfuscated and encrypted by the platform.
Sample JSON

"configuration": [
      {
          "label": "API Token",
          "key": "api_token",
          "type": "password",
          "mandatory": true,
          "default": "",
          "description": "API Token for the platform."
      },
]

Plugin Configuration View

Text Parameter

Use this parameter for storing any string information, such as base-url, username, etc. This parameter will have a normal text input on the plugin configuration page.
Sample JSON

"configuration": [
      {
          "label": "Tenant Name",
          "key": "tenant_name",
          "type": "text",
          "mandatory": true,
          "default": "",
          "description": "Tenant Name of the platform."
      },
]

Plugin Configuration View

Number Parameter

Use this parameter for storing number/float values. This parameter will have a number input box on the plugin configuration page. (The hash list is in the Threat Exchange module.)

"configuration": [
      {
               "label": "Maximum File hash list size in MB.",
               "key": "max_size",
               "type": "number"
               "mandatory": true,
               "default": 12,
               "description": "File hash list size in MB."
      },
]

Plugin Configuration View

Choice Parameter

Use this parameter for storing any enumeration parameter values. This parameter will have a dropdown box on the plugin configuration page.
Sample JSON

"configuration": [
      {
          "label": "Base URL",
          "key": "base_url",
          "type": "choice",
          "choices": [
              {
                  "key": "Commercial cloud (api.crowdstrike.com)",
                  "value": "https://api.crowdstrike.com"
              },
              {
                  "key": "US 2 (api.us-2.crowdstrike.com)",
                  "value": "https://api.us-2.crowdstrike.com"
              },
              {
                  "key": "Falcon on GovCloud (api.laggar.gcw.crowdstrike.com)",
                  "value": "https://api.laggar.gcw.crowdstrike.com"
              },
              {
                  "key": "EU cloud (api.eu-1.crowdstrike.com)",
                  "value": "https://api.eu-1.crowdstrike.com"
              }
          ],
            "default": "https://api.crowdstrike.com",
            "mandatory": true,
            "description": "API Base URL."
      },
]

Plugin Configuration View

After selecting the input:

Multichoice Parameter

Use this parameter for storing multiple choice values. This parameter will have a dropdown box on the plugin configuration page with the ability to select multiple values. (Severity is in the Threat Exchange module.)

Sample JSON

"configuration": [
      {
          "label": "Severity",
          "key": "severity",
          "type": "multichoice",
          "choices": [
              {
                  "key": "Unknown",
                  "value": "unknown"
              },
              {
                  "key": "Low",
                  "value": "low"
              },
              {
                  "key": "Medium",
                  "value": "medium"
              },
              {
                  "key": "High",
                  "value": "high"
              },
              {
                  "key": "Critical",
                  "value": "critical"
              }
          ],
            "default": [
                "critical",
                "high",
                "medium",
                "low",
                "unknown"
            ],
            "mandatory": false,
            "description": "Only indicators with matching severity will be saved."
     }
]

Plugin Configuration View

Toggle Parameter

This parameter stores a boolean value, toggle enabled is True and toggle disabled is False.
Use System Proxy (‘proxy’): Use system proxy configured in Settings. (Default: False)
Plugin Configuration View

Note

This parameter is provided by Core, and it is not allowed to add from the plugin’s manifest.json file.

Main.py

This python file contains the core implementation of the plugin.

Standard Imports
from netskope.integrations.crev2.plugin_base import (
   PluginBase,
   ValidationResult,
   Entity,
   EntityField,
   EntityFieldType
)

from netskope.integrations.crev2.models import (
   Action,
   ActionWithoutParams
)
PluginBase Variables

PluginBase provides access to variables that can be used during the plugin lifecycle Methods. Here is the list of variables.

Variable Name Usage Description
self.logger self.logger.error(“Message”)
self.logger.warn(“Message”)
self.logger.info(“Message”)		 Logger handle provided by core. Use this object to log important events. The logs would be visible in the Cloud Exchange Audit logs. Refer to the Logger Object documentation.
self.configuration self.configuration.get(<attribute-key-name>) JSON representation of the configuration object of the Plugin instance. Use this to access the configuration attributes like authentication credentials, server details, etc. Use the key name of the attribute mentioned in manifest.json.
self.last_run_at If self.last_run_at:
self.last_run_at.timestamp()
Use this format to convert the last run time in epoch format.		 Provides the timestamp of the last successful run time of the Plugin’s pull method. The Cloud Exchange core maintains the checkpoint time after each successful pull() execution. For the first execution, the value would be None. The datatype of the object is datetime.
self.notifier (optional) self.notifier.info(“message”)
self.notifier.warn(“message”)
self.notifier.error(“message”)		 This object provides a handle for Risk Exchange core’s notifier. Use this object to push any notification to the platform. The notifications would be visible in the Risk Exchange UI. Make sure the message contains summarized information for the user to read and take necessary actions.
Used notifier in Netskope plugin if the push() method exceeds 8MB limit of the product.
self.proxy requests.get(url=url, proxies=self.proxy) Handle of system’s proxy settings if configured, else {}.
self.ssl_validation requests.get(url=url, verify=self.ssl_validation) Boolean value which mentions if ssl validation is enforced for REST API calls.
Plugin Class
  • Plugin class has to be inherited from the PluginBase class. PluginBase class is defined in netskope.integrations.crev2.plugin_base.
  • Make sure the Plugin class provides implementation for the get_entities, fetch_records, update_records, validate, get_actions, get_action_params, validate_action, and execute_action.
  • Plugin class will contain all the necessary parameters to establish connection and authentication with the third-party API.
  • Pagination should always be considered while developing any feature in a plugin.
"""Sample plugin implementation.

This is a sample implementation of the base PluginBase class. Which explains the concrete implementation of the base class.
"""
from netskope.integrations.crev2.models import (
   Action,
   ActionWithoutParams
)
from netskope.integrations.crev2.plugin_base import (
   PluginBase,
   ValidationResult,
   Entity,
   EntityField,
   EntityFieldType
)
from .utils.helper import SamplePluginException, SamplePluginHelper

class SamplePlugin(PluginBase):
   """SamplePlugin class having concrete implementation for fetching information and performing actions.

   This class is responsible for implementing pull, perform actions and validate methods with proper return types, so that it's lifecycle execution can be scheduled by the CRE core engine.
   """
   def __init__(
       self,
       name,
       *args,
       **kwargs,
   ):
       """Init method.

       Args:
           name (str): Configuration name.
       """
       super().__init__(
           name,
           *args,
           **kwargs,
       )
       self.plugin_name, self.plugin_version = self._get_plugin_info()
       self.log_prefix = f"{MODULE_NAME} {self.plugin_name} [{name}]"
       self.sample_plugin_helper = SamplePluginHelper(
           logger=self.logger,
           log_prefix=self.log_prefix,
           plugin_name=self.plugin_name,
           plugin_version=self.plugin_version,
           configuration=self.configuration,
       )

   def _get_plugin_info(self) -> tuple:
       """Get plugin name and version from metadata.

       Returns:
           tuple: Tuple of plugin's name and version fetched from metadata.
       """
       try:
           metadata_json = SamplePlugin.metadata
           plugin_name = metadata_json.get("name", PLATFORM_NAME)
           plugin_version = metadata_json.get("version", PLUGIN_VERSION)
           return (plugin_name, plugin_version)
       except Exception as exp:
           self.logger.error(
               message=(
                   "{} {}: Error occurred while"
                   " getting plugin details. Error: {}".format(
                       MODULE_NAME, PLATFORM_NAME, exp
                   )
               ),
               details=traceback.format_exc(),
           )
       return (PLATFORM_NAME, PLUGIN_VERSION)
def validate()

Arguments:
configuration (dict): Dict object having all the Plugin configuration parameters.
Returns:
ValidationResult: ValidationResult object with success flag and message.
This is an abstract method of PluginBase Class.

  • This method validates the plugin configuration and authentication parameters passed while creating a plugin configuration.
  • This method will be called only when a new configuration is created or updated.
  • Validates against all the mandatory parameters are passed with the proper datatype.
  • Validate the authentication parameters and the API endpoint to ensure the smooth execution of the plugin lifecycle.
  • Return the object of ValidationResult (Refer to ValidationResult Data Model) with a success flag indicating validation success or failure and the validation message containing the validation failure reason.
def validate(self, configuration):
   """Validate the Plugin configuration parameters.

   Validation for all the parameters mentioned in the manifest.json
   for the existence and data type. Method returns the
   netskope.integrations.crev2.plugin_base.ValidationResult object
   with success = True in the case of successful validation and
   success = False and an error message in the case of failure.

   Args:
       configuration (dict): Dict object having all the Plugin configuration parameters.
   Returns:
       netskope.integrations.crev2.plugin_base.ValidationResult:
           ValidationResult object with success flag and message.
   """
   # Validate Base URL
   instance_url = configuration.get("base_url", "").strip().strip("/")
   if not instance_url:
       err_msg = "Base URL is required configuration parameter."
       self.logger.error(
           f"{self.log_prefix}: Validation error occurred. {err_msg}"
       )
       return ValidationResult(success=False, message=err_msg)
   elif not (isinstance(instance_url, str) and self._validate_url(instance_url)):
       err_msg = "Invalid Base URL provided in configuration parameters."
       self.logger.error(
           f"{self.log_prefix}: Validation error occurred. {err_msg}"
       )
       return ValidationResult(success=False, message=err_msg)

   # Validate client_id
   client_id = configuration.get("client_id", "").strip()
   if not client_id:
       err_msg = "Client ID is a required field."
       self.logger.error(f"{{self.log_prefix}}: Validation error occurred. Error: {err_msg}")
       return ValidationResult(success=False, message=err_msg,)

   elif not isinstance(client_id, str):
       err_msg = "Invalid Client ID value provided."
       self.logger.error(f"{{self.log_prefix}}: {err_msg}")
       return ValidationResult(success=False, message=err_msg)

   return ValidationResult(
       success=True,
       message="Validation successful for Sample Plugin."
   )

 def _validate_url(self, url: str) -> bool:
    """Validate the given URL.

    Args:
       url (str): URL to validate.

    Returns:
       bool: True if URL is valid else False.
    """
    parsed_url = urlparse(url.strip())
    return parsed_url.scheme and parsed_url.netloc
def get_entities()

Arguments:
None
Returns:
List[Entity]: List of Entity classes with plugin supported entity, fields and types.
This is an abstract method of PluginBase Class.

  • This method should return a list of Entity classes with plugin supported entities, fields and types.
  • If the plugin does not support any Entity, return an empty list.
def get_entities(self) -> list[Entity]:
   """Get available entities."""
   return [
       Entity(
          name="Users",
          fields=[
               EntityField(name="User Email", type=EntityFieldType.STRING, required=True),
               EntityField(name="Netskope Normalized Score", type=EntityFieldType.NUMBER),
          ],
       ),
       Entity(
          name="Applications",
          fields=[
              EntityField(name="Application ID", type=EntityFieldType.STRING, required=True),
              EntityField(name="Tags", type=EntityFieldType.LIST),
          ],
       )
   ]
def fetch_records()

Arguments:
entity (str): Entity to be fetched.
Returns:
List: List of records to be stored on the platform.
This is an abstract method of PluginBase Class.

  • This method implements the logic to fetch records () from the 3rd party API endpoints. This method is invoked periodically.
  • Use the proxy configuration passed by the CE platform by invoking self.proxy. It returns the python dict object which can be used directly with the requests module.
  • All the configuration parameters for API authentication are passed as python dict receives them by invoking self.configuration.
  • All the logs can be logged by the self.logger object with proper log level (info, warn, error). This object logs the logs to MongoDB and can be accessed via API calls.
  • Use self.ssl_validation bool to enable/disable validation of the SSL server certificate.
  • Handle pulling according to the entity if the plugin supports multiple entities.
  • Fetch risk information in this method if risk details are available in the same API response.
  • Return the list of dictionaries that contain the data received from the API endpoint mapped with plugin entity fields.
  • In the case of failure, raise an error or exception to the appropriate type with the proper message.
  • Make sure to use API helper for all API requests. Refer
  • Make sure to handle the case when the maximum payload size supported by the API endpoint is exceeded. There can be multiple ways to handle this case.
    • If the API endpoint supports multiple requests with a fixed payload size, send the data in chunks. This concept is called pagination.
def fetch_records(self, entity: str):
   """Fetches entity records from a third party API.

   Implement the logic of fetching data based on entity from third party apis 
   and return the list of dictionary objects containing plugin supported entity
   fields and their values on a successful fetch otherwise raises an exception.

   Returns:
       List[{}]: List of dictionary objects of entity data
       fetched from the third party platform.
   """

   # Load all the configured plugin parameters as python dict objects.
   # Use the key name provided in the manifest.json file for the
   # configuration parameters to get the value of that particular parameter.

   # See api_helper for API requests method to check
   # how to use proxy dict and ssl_validation flag.
	
   # check entity type and fetch records accordingly
   # if plugin supports multiple entity types

   headers = {
"Authorization": "<>",
   }
   if entity == "Users":
	resp_json = self.sample_plugin_helper.api_helper(
      url="www.example.com/users",
             method="GET",
             headers=headers,
             proxies=self.proxy,
             verify=self.ssl_validation,
             logger_msg=f"fetching {entity_name} from the platform",
       )
       records = resp_json.get("value", [])
   elif entity == "Applications":
	resp_json = self.sample_plugin_helper.api_helper(
      url="www.example.com/applications",
             method="GET",
             headers=headers,
             proxies=self.proxy,
             verify=self.ssl_validation,
             logger_msg=f"fetching {entity_name} from the platform",
       )
       records = resp_json.get("value", [])

   # Get the logger object for logging purposes. This logger object logs
   # all the logs to mongodb under the core database logs collection. 
   # Log timestamp is automatically recorded by the logger library.
   # Supported logging levels are info, warn and error.
   self.logger.info(f"{self.log_prefix}: Successfully fetched {
len(records)}{entity} from the platform.")
   return records
def update_records()

Arguments:
entity (str): Entity to be updated.
records (list): Records to be updated.
Returns:
List: List of updated records.
This is an abstract method of PluginBase Class.

  • This method implements the logic to update stored records from the 3rd party API endpoints. This method is invoked periodically.
  • Used to update fields values from the third party API of fetched records. All the fields should be updated in this method other than the field which is unique and required for the entity.
  • Calculate Netskope Normalized Score in update_records method and add field in record dictionary if risk information is fetched in fetch_record method. Make sure to map scores to an integer between 0-1000.
  • Make sure to not return None value for any field.
  • Use the proxy configuration passed by the CE platform by invoking self.proxy. It returns the python dict object which can be used directly with the requests module.
  • All the configuration parameters for API authentication are passed as python dict receives them by invoking self.configuration.
  • All the logs can be logged by the self.logger object with proper log level (info, warn, error). This object logs the logs to MongoDB and can be accessed via API calls.
  • Use self.ssl_validation bool to enable/disable validation of the SSL server certificate.
  • Return the list of scores which contain the data received from the API endpoint, which will be stored in the Record object. Note: The list will contain Record objects.
  • In the case of failure, raise an error or exception of the appropriate type with the proper message.
  • Make sure to handle the case when the maximum payload size supported by the API endpoint is exceeded. There can be multiple ways to handle this case.
    • If the API endpoint supports multiple requests with a fixed payload size, send the data in chunks. This concept is called pagination also.
  • Before updating records, first check that the unique/required field for plugin entity is available in fetched records and create a list. Example:
if entity == "Users":
user_email = []
for record in records:
    if record.get("User Email"):
           user_email.append(record.get("User Email"))

Note:
Make sure to calculate the normalized score in the update_record method and not in fetch_records when risk details are available.

def update_records(self, entity: str, records: list[dict]):
   """Update entity records from a third party API.

   Implement the logic of fetching updated data based on entity 
   from third party apis and return the list of updated values
   for all fetched records dictionary containing plugin supported entity
   fields and Netskope Normalize Score if risk information available on
   a successful fetch otherwise raises an exception.

   Returns:
       List[{}]: List of records with updated values of entity data
       fetched from the third party platform.
   """

   # Load all the configured plugin parameters as python dict objects.
   # Use the key name provided in the manifest.json file for the
   # configuration parameters to get the value of that particular parameter.

   # See api_helper for API requests method to check
   # how to use proxy dict and ssl_validation flag.
	
   # check entity type and fetch records accordingly
   # if plugin supports multiple entity types

   headers = {
"Authorization": "<>",
   }
   updated_records = {}
   if entity == "Users":
user_email = []
for record in records:
    if record.get("User Email"):
           user_email.append(record.get("User Email"))
	resp_json = self.sample_plugin_helper.api_helper(
      url="www.example.com/users",
             method="GET",
             headers=headers,
             proxies=self.proxy,
             verify=self.ssl_validation,
             logger_msg=f"updating {entity_name} from the platform",
       )
       updated_records = resp_json.get("value", [])
       for record in records:
           user_email = record.get("User Email")
           if user_email and user_email in updated_records:
               record.update(updated_records[user_email])
               count += 1
   elif entity == "Applications":
       app_id = []
for record in records:
    if app_id.get("Application ID"):
           app_id.append(record.get("Application ID"))
	resp_json = self.sample_plugin_helper.api_helper(
      url="www.example.com/applications",
             method="GET",
             headers=headers,
             proxies=self.proxy,
             verify=self.ssl_validation,
             logger_msg=f"updating {entity_name} from the platform",
       )
       updated_records = resp_json.get("value", [])
       for record in records:
           app_id = record.get("Application ID")
           if app_id and app_id in updated_records:
               record.update(updated_records[app_id])
               count += 1

   self.logger.info(
       f"{self.log_prefix}: Successfully updated {count} "
       f"{entity} from the platform."
   )
   return records
def get_actions()

Arguments:
None
Returns:
List[ActionWithoutParams]: List of ActionWithoutParams which has label and value defined.
This is an abstract method of PluginBase Class.

  • This method should return a list of all the supported actions.
  • Add all the supported actions in the ActionWithoutParams class and return a list of objects of ActionWithoutParams class.
  • If the plugin does not support any action, No action should be implemented.
def get_actions(self):
   """Get available actions.

   Returns:
       List[ActionWithoutParams]: List of ActionWithoutParams objects
       that are supported by the plugin.
   """
   return [
       ActionWithoutParams(label="Add to group", value="add"),
       ActionWithoutParams(label="Remove from group", value="remove"),
       ActionWithoutParams(label="No action", value="generate"),
   ]
def get_action_params():

Arguments:
action (Action): The type of action
Returns:
List: Returns a list of selected action parameters.
This is an abstract method of PluginBase Class.

  • This method should return the list of fields to be rendered in the UI when a target is selected from the dropdown.
  • This method should be called after the user selects any of the actions.
  • If the selected action requires any parameters then return a list of dictionaries (where each dictionary is a configurable input) otherwise return an empty list. Check Plugin Configuration Parameter types to see how fields are defined.
  • If the plugin does not support any action, No action should be implemented and return an empty list in the get_action_fields method.
def get_action_params(self, action: Action):
   """Get fields required for an action.
   Args:
       action (Action): Action object which is selected as Target.
   Return:
       List[Dict]: List of configurable fields based on selected action.
   """
   if action.value in ["generate"]:
       return []
   if action.value == "add":
       return [
           {
               "label": "Email",
               "key": "email",
               "type": "text",
               "default": "",
               "mandatory": True,
               "description": "Email ID of the user to perform the action on.",
           },
           {
               "label": "Group Name",
               "key": "group_name",
               "type": "choice",
               "choice": [{key: "name", value: "id"}, ...],
               "default": "",
               "mandatory": True,
               "description": "Name of group."
           }
       ]
   else:
       return []
def validate_action():

Arguments:
action (Action): The type of action with action parameters
Returns:
ValidationResult: ValidationResult object with success flag and message.
This is an abstract method of PluginBase Class.

  • This method validates the action parameters passed while creating an action configuration.
  • This method will be called only when a new action configuration is created or updated.
  • Separate validations should be made for empty field validation and type check validation in the validation method for plugin configuration.
  • Validates against all the mandatory parameters are passed with the proper data type.
  • While validating, use strip() for action parameters like URL, username, or text type etc. except API Tokens and Password fields.
  • Return the object of ValidationResult (Refer to ValidationResult Data Model) with a success flag indicating validation success or failure and the validation message containing the validation failure reason.
  • If the plugin does not have actions then return the ValidationResult object with a success flag otherwise check for validations.
def validate_action(self, action: Action):
   """Validate Action Parameters.

   Args:
       action (Action): Action object having all the configurable parameters.
   Return:
       netskope.integrations.crev2.plugin_base.ValidationResult:
           ValidationResult object with success flag and message.
   """
   action_params = action.parameters
   if action.value not in ["generate", "add", "remove"]:
	err_msg = "Unsupported action provided."
       self.logger.error(f"{self.log_prefix}: {err_msg}")
       return ValidationResult(
           success=False, message=err_msg
       )
   if action.value in ["generate"]:
        return ValidationResult(
            success=True, message="Validation successful."
        )

   email = action_params.get("email", "")
   if not email:
       err_msg = "Email is a required action parameter."
       self.logger.error(f"{self.log_prefix}: {err_msg}")
       return ValidationResult(success=False, message=err_msg)
   elif not isinstance(email, str):
       err_msg = "Invalid Email provided in action parameters."
       self.logger.error(f"{self.log_prefix}: {err_msg}")
       return ValidationResult(success=False, message=err_msg)
   if "$" in email:
log_msg = (
     "Email contains the Source Field"
     " hence validation for this field will be performed"
     " while executing the action."
       )
       self.logger.info(f"{self.log_prefix}: {log_msg}")

   if action.value == "add":
       if action_params.get("group_name") is None:
           err_msg = "Group Name can not be an empty field."
           self.logger.error(f"{self.log_prefix}: {err_msg}")
           return ValidationResult(success=False, message=err_msg)

   self.logger.debug(f"{self.log_prefix}: Validation successful.")
   return ValidationResult(
       success=True, message="Validation successful."
   )
def execute_action():

Arguments:
action (Action): The action that needs to be performed with action parameters.
Returns:
None
This is an abstract method of PluginBase Class.

  • This method implements the logic to execute action.
  • Make sure the method completes an action like add to the group or remove from the group.
  • A helper method is preferred to execute the third party API calls

Note:
For ‘add user to group’ and ‘remove user from group’ actions, first validate whether the user exists on the platform, if the platform supports an API for user validation in execute_action method.

def execute_action(self, action: Action):
   """Execute action on the record.

   Calls _add_to_group helper methods in the case of add action.
   Return when action is generate.

   Args:
       action (Action): Action object having all the configurable parameters.
   """
   action_label = action.label
   action_params = action.parameters
   if action.value == "generate":
	self.logger.debug(
           f'{self.log_prefix}: Successfully executed "{action_label}" action.'
           " Note: No processing will be done from plugin for "
           f'the "{action_label}" action.'
       )
       return

   email = action_params.get("email", "").strip()
   if not email:
       err_msg = (
           "Email not found in the action parameters. "
           f"Hence, skipping execution of '{action_label}' action."
       )
       self.logger.error(f"{self.log_prefix}: {err_msg}")
       return
   elif not isinstance(email, str):
       err_msg = (
           "Invalid Email found in the action parameters. "
           f"Hence, skipping execution of '{action_label}' action."
       )
       self.logger.error(f"{self.log_prefix}: {err_msg}")
       return
   if action.value == "add":
       group_id = action_params.get("group")
       email = action_params.get("email", "").strip()
      self.logger.info(f"{log_prefix}: Adding user with email '{email}' to group.")
       self._add_to_group(email, group_id)
       self.logger.info(f"{log_prefix}: Successfully added user with email '{email}' to group.")

Data Models

This section lists the Data Models and their properties.

Entity
  • This class represents the supported plugin entity.
  • Usually there will be Users, Devices or Applications but can be different based on the requirement.
  • It will be used to populate plugin supported entities and fields in the Risk Exchange plugin configuration page.

Data Model Properties

Name Type Description
name string Display label of the entity that is shown in UI
fields list[EntityField] Contains list of supported entity fields 
from netskope.integrations.crev2.plugin_base import (
   Entity,
   EntityField,
   EntityFieldType
)
def get_entities(self) -> list[Entity]:
    """Get available entities."""
    return [
       Entity(
          name="Users",
          fields=[
              EntityField(name="User Email", type=EntityFieldType.STRING, required=True),
              EntityField(name="Netskope Normalized Score", type=EntityFieldType.NUMBER),
          ],
       )
    ]
EntityField
  • This class represents the entity field and field type like string, number etc.
  • It will be used in the get_entities method.
  • It is used to show the supported fields on the plugin configuration page.

Data Model Properties

Name Type Description
name string Display label of the field that is shown in UI
type EntityFieldType Eum class to define field type
required bool Field is required to map in plugin configuration when true 
EntityField(name="User Email", type=EntityFieldType.STRING, required=True)
EntityFieldType
  • This class represents the entity field type like string, number etc.
  • It will be used in the get_entities method to define field type.

Data Model Properties (Enum)

Name
STRINGNUMBER
LIST
DATETIME
CALCULATED
IPV4
IPV6
EMAIL
REFERENCE
VALUE_MAP
RANGE_MAP		 
EntityField(name="Netskope Normalized Score", type=EntityFieldType.NUMBER)
Action
  • This class represents the action one wants to perform.
  • Usually there will be generate, add, remove, or more actions that one needs to implement
  • It will be used in validate_action, execute_actions, and get_action_params.

Data Model Properties

Name Type Description
label string Display label of the action that is shown in UI
value string Key to access it
parameters Dict It will contain all the required action fields for respective plugins
generateAlerts Bool Generates alerts to CTO when true
performLater Bool Synchronize actions when true
requireApproval Bool Action need approval from Action page when true
ActionWithoutParams
  • This class represents the action one wants to perform without using parameters.
  • It will be used in the get_actions method.
  • It is used to show the supported actions to the user.

Data Model Properties

Name Type Description
label string Display label of the action that is shown in UI
value string Key of action to access it in plugin code 
from netskope.integrations.crev2.models import ActionWithoutParams
return [
            ActionWithoutParams(label="Add to group", value="add"),
            ActionWithoutParams(label="Remove from group", value="remove"),
            ActionWithoutParams(label="No actions", value="generate"),
        ]
ValidationResult
  • This class contains the result of the validation process on the plugin configuration parameters passed to the Plugin object while creating a new configuration for the plugin.
  • Make sure that all the parameters passed to the validate method are validated against the data-type and value.
  • Validate method returns the object of this class with a success flag indicating the result of the validation operation and a message field containing the proper error message in the case of validation failure. (In a case of success a simple success message with the success flag True should be returned)
  • Note that Validate Action also returns the object of this class.

Data Model Properties

Name Type Description
success bool success flag indicates the result of the validation operation, whether it succeeded or failed.
message string Message field denotes the error in case of failure. In the case of success, it can be a simple success message. 
from netskope.integrations.crev2.plugin_base import ValidationResult
ValidationResult(
           success=True,
           message="Validation Successful for Sample plugin"
       )

Logging

Risk Exchange provides a handle of the logger object for logging.

  • Avoid print statements in the code.
  • This object logs to the central Cloud Exchange database with the timestamp field. Supported log levels are info, debug, warn, and error.
  • Make sure any API authentication secret or any sensitive information is not exposed in the log messages.
  • Make sure to implement a proper logging mechanism with the logger object passed by the Cloud Exchange platform.
  • Make sure enough logging is done, which helps the Ops team in troubleshooting.
  • Make sure any sensitive data is not logged or leaked in the notification or logs.
  • Refer log prefix format for all loggers.
    • The log messages should start with “<module> <plugin_name> [configuration_name]: “. Example: “CRE Crowdstrike [CrowdStrike Configuration Name]: <log_message>“. (This is a suggestion, we can avoid configuration names). [logger.info(“<module> <plugin_name>: <message>”)]
  • Make sure to add details in error logs wherever possible.
    • It is recommended to add error traceback in details for error logs.
import traceback

self.logger.error(
   message=f"{self.log_prefix}: Error log-message goes here.",
   details=str(traceback.format_exc())
)

self.logger.warn(
   f"{self.log_prefix}: Warning log-message goes here."
)

self.logger.info(
   f"{self.log_prefix}: Info log-message goes here."
)

self.logger.debug(
   f"{self.log_prefix}: Debug log-message goes here."
)

Notifications (Optional)

Risk Exchange provides a handle of notification objects that can be used to generate notifications in the Risk Exchange UI.

  • This object is passed from the Cloud Exchange platform to the Plugin object. Every plugin integration can use this object whenever there is a case of failure that has to be notified to the user immediately. The notification timestamp is managed by the Cloud Exchange platform.
  • This object will raise the notification in the UI with a color-coding for the severity of the failure. Supported notification severities are info, warn, and error.
  • Make sure any API authentication secret or any sensitive information is not exposed in the notification messages.
  • Use a notifier object to raise the notification for failures or critical situations (like rate-limiting, or exceeding payload size) to notify the status of the plugin to the user.
self.notifier.info(
f"{log_prefix}: Info notification-message goes here."
 )

 self.notifier.error(
f"{log_prefix}: Error notification-message goes here."
 )

 self.notifier.warn(
f"{log_prefix}: Warning notification-message goes here."
 )

Refer to https://github.com/netskopeoss/ta_cloud_exchange_plugins/tree/main/microsoft_entra_id_ztre  plugin for Risk Exchange plugin development of plugin methods, development best practices, and handling of multiple Entities.

Testing

Linting

As part of the build process, we run a few linters to catch common programming errors, stylistic errors, and possible security issues.

Flake8

This is a basic linter. It can be run without having all the dependencies available and will catch common errors. You can also use this linter to enforce the standard python pep8 formatting style. On rare occasions, you may encounter a need to disable an error/warning returned from this linter. Do this by adding an inline comment of the sort on the line you want to disable the error:

# noqa: <error-id>

For example:

example = lambda: 'example' # noqa: E731

When adding an inline comment, always include the error code you are disabling for. That way if there are other errors on the same line they will be reported.
For more info: https://flake8.pycqa.org/en/latest/user/violations.html#in-line-ignoring-errors
PEP8 style docstring check is also enabled with the flake8 linter. Make sure every function/module has a proper docstring added.

Plugin Deployment on Cloud Exchange

Package the Plugin

Cloud Exchange expects the developed plugin in zip or tar.gz format.
Execute this command to zip the package:

zip -r sample_plugin.zip sample_plugin

Execute this command to generate tar.gz package:

tar -zcvf sample_plugin.tar.gz sample_plugin

Add a Repo

To deploy your plugin in Cloud Exchange, add a repository first.

  1. Log in to Cloud Exchange.
  2. Go to Settings > Plugin Repository.
  3. Click Configure New Repository.
  4. Enter a Repository name, Repository URL, Username, and Personal Access Token.
  5. Click Save.
  6. Go to Settings > Plugins.
  7. Select your Repository name from Repository dropdown, and proceed to the next section.

Upload the Plugin zip or tar.gz File

  1. If you continuing from the last section, jump to step 2. Otherwise, go to Plugin Repositories.
  2. Click the Upload Plugin icon for your repository.
  3. Click Browse.
  4. Select the zip or tar.gz file you created.
  5. Click Upload.

Deliverables

Plugin Guide

  • Make sure to update the plugin guide with every release.
  • A Plugin guide should have this content:
    • Release Notes
    • Description
    • Prerequisites
    • CE version Compatibility
    • Plugin Scope
      • Type of data supported
      • Mapping
        • Pull Mapping for <Entity>
    • Permissions
    • API Details
      • List of APIs used
    • Performance Matrix
    • User Agent
    • Workflow
    • Configuration on third-party plugin platform
    • Configuration on Netskope CE
      • Third-party Plugin configuration
      • Configuration of business rule
      • Configuration of Action
    • Validation
    • Troubleshooting
    • Limitations

Refer to any published Risk Exchange plugin guide for examples.

Demo Video

After the successful development of the Risk Exchange plugin, create a demo video and show the end-to-end workflow of the plugin.
Note: Refer to Microsoft Entra ID plugin guide for reference.

Share this Doc

Risk Exchange Custom Plugin Developers Guide

Or copy link

In this topic ...