User Risk Exchange Custom Plugin Developers Guide
User Risk Exchange Custom Plugin Developers Guide
This guide explains how to write a new User Risk Exchange plugin and extract maximum value out of your risk ecosystem by leveraging the functionality provided within the User 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.
User Risk Exchange Module
The Cloud Exchange platform, and its User 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 sharing of data from Netskope to third-party tools and vice versa.
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. User Risk Exchange is one of the modules running in Cloud Exchange.
- Plugin: Plugins are Python packages that have logic to fetch users and risk scores from 3rd party Threat Intel systems, which will then be stored in User Risk Exchange. It can also perform actions.
- 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.
- Users (UBA alerts): Records (of users) are objects having email and their risk scores gathered from various platforms and stored in the Cloud Exchange database.
Development Guidelines
- Use the Plugin Directory Structure for all Python code.
- Make sure all the 3rd party 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.
- For Scripts/Integrations written in Python, make sure to create unit tests. Refer to Unit Testing.
- 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 Cloud Exchange 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.
- 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.
- 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.
- 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.
- 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 Cloud Exchange 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 .get() method.
- API Tokens and Password fields should not use strip().
- The log messages should start with “<module> <app name> Plugin [configuration_name]: “. Example: “URE Crowdstrike Plugin [CrowdStrike Configuration Name]: <log_message>“. [This is a suggestion, we can avoid configuration name]. (logger.info(“<module> <plugin_name> Plugin: <message>”))
- While logging an error log, if possible, we should add a traceback of the exception. USE: self.logger.error(error, details=traceback.format_exc())
- The Toast message should not contain the <app_name> <module> Plugin: 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.
- 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.
- Follow the Plugin Directory Structure guidelines.
- If the description contains a link, it should be a hyperlink instead of plaintext.
- Make sure to map the various fields received from API calls with the User and Score data model to leverage the full benefit of the system. Fields like Score in Records fields make more sense to the SOC user when analyzing the data. Make sure to map the score field which gives more context to the SOC analyst. The 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.
- Use a notifier object to raise the notification for failures or critical situations (like rate-limiting, exceeding payload size) to notify the status of the plugin to the user.
- 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.
- Make sure the plugin directory name (sample_plugin) matches with the manifest.json’s ID field.
- 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>.
- The logger statement should follow this format:
- logger.info(“<module> <plugin_name> Plugin: <message>”)
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.7 and above). Make sure to set up python3 in your development environment. Pytest is used to run unit tests.
Included Python Libraries
Following Python libraries are included within the Netskope Cloud Exchange platform.
Library Name | Version |
---|---|
aiofiles | 22.1.0 |
amqp | 5.1.1 |
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 |
boto3 | 1.26.51 |
botocore | 1.29.51 |
billiard | 3.6.4.0 |
celery | 5.2.7 |
cabby | 0.1.23 |
cachetools | 5.2.1 |
celerybeat-mongo | 0.2.0 |
certifi | 2022.12.7 |
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 | 39.0.0 |
cybox | 2.1.0.21 |
defusedxml | 0.7.1 |
dnspython | 2.3.0 |
docker | 6.0.1 |
fastapi | 0.89.1 |
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.51.1 |
grpcio-status | 1.51.1 |
gunicorn | 20.1.0 |
h11 | 0.14.0 |
idna | 3.4 |
importlib-metadata | 6.0.0 |
isodate | 0.6.1 |
jmespath | 1.0.1 |
jsonpath | 0.82 |
jsonschema | 4.17.3 |
kombu | 5.2.4 |
libcst | 0.3.21 |
libtaxii | 1.1.119 |
lxml | 4.9.2 |
mongoengine | 0.25.0 |
more-itertools | 9.0.0 |
MarkupSafe | 2.1.2 |
memory-profiler | 0.61.0 |
mixbox | 1.0.5 |
mongoquery | 1.4.2 |
msrest | 0.7.1 |
multidict | 6.0.4 |
mypy-extensions | 0.4.3 |
netskopesdk | 0.0.25 |
numpy | 1.23.5 |
oauthlib | 3.2.2 |
onelogin | 3.1.0 |
ordered-set | 4.1.0 |
orderedmultidict | 1.0.1 |
overrides | 6.5.0 |
pandas | 1.5.0 |
packaging | 23.0 |
passlib | 1.7.4 |
pycparser | 2.21 |
prompt-toolkit | 3.0.36 |
proto-plus | 1.22.2 |
protobuf | 4.21.12 |
psutil | 5.9.4 |
pydantic | 1.10.4 |
pyasn1 | 0.4.8 |
pyasn1-modules | 0.2.8 |
PyJWT | 2.6.0 |
pymongo | 4.3.3 |
pyparsing | 3.0.9 |
python-dateutil | 2.8.2 |
pyrsistent | 0.19.3 |
python-multipart | 0.0.5 |
python3-saml | 1.15.0 |
pytz | 2022.7.1 |
PyYAML | 6.0 |
requests | 2.28.2 |
requests-oauthlib | 1.3.1 |
rsa | 4.9 |
six | 1.16.0 |
starlette | 0.22.0 |
sniffio | 1.3.0 |
s3transfer | 0.6.0 |
stix | 1.2.0.11 |
taxii2-client | 2.3.0 |
typing-inspect | 0.8.0 |
typing-utils | 0.1.0 |
typing_extensions | 4.4.0 |
urllib3 | 1.26.14 |
uvicorn | 0.20.0 |
vine | 5.0.0 |
wcwidth | 0.2.6 |
weakrefmethod | 1.0.3 |
websocket-client | 1.4.2 |
Werkzeug | 2.2.2 |
xmlsec | 1.3.11 |
zipp | 3.11.0 |
requests-mock | 1.7.0 |
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, the command shown below 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.
While importing modules from above lib folder, you should be using a relative import instead of an absolute import.
IDE
Recommended IDEs are PyCharm or Visual Studio Code.
Plugin Directory Structure
This section mentions the typical directory structure of a User 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 User Risk Exchange code. Make sure every plugin package contains the empty “__init__.py” file.
- 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.
- main.py: This python file contains the Plugin class containing the concrete implementation for the fetch users and scores, execute actions, validate, and validate actions method.
- 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.
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.
Sample Changelog.md
# 1.0.1 ## Fixed - Fixed pagination while fetching records. # 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 URE module to render the plugin in the UI as well as enabling the URE 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 User 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 and user scores and performs actions for members in group/groups) This description would appear on the Plugin Configuration card. (Required)
- 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)
- Type: ( [String] ): The type of users the data fetches. Can be Host or User or Both.
- 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)
- 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 URE 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 User 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" }, ]
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" }, ]
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" }, ]
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.cre.plugin_base import PluginBase, ValidationResult, PushResult
from netskope.integrations.cre.models.business_rule import Action, ActionWithoutParams, ActionWithoutParams, Action.
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.storage | Cloud Exchange provides the plugin a mechanism to maintain the state. Use this object to persist in any state that would be required during subsequent calls. The datatype of this object is python dict. | |
self.notifier | self.notifier.info(“message”) self.notifier.warn(“message”) self.notifier.error(“message”) |
This object provides a handle for User Risk Exchange core’s notifier. Use this object to push any notification to the platform. The notifications would be visible in the User 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.cre.plugin_base.
- Make sure the Plugin class provides implementation for the fetch_records, fetch_scores, validate, get_actions, get_action_fields, validate_actions, and execute_actions.
- Plugin class will contain all the necessary parameters to establish connection and authentication with the 3rd 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.cre.plugin_base import PluginBase, ValidationResult, PushResult from netskope.integrations.cre.models import Indicator, IndicatorType from typing import List from datetime import datetime import requests PLUGIN_NAME = "<module> <plugin_name> Plugin" 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 URE core engine. """
def fetch_records()
This is an abstract method of PluginBase Class.
- This method implements the logic to fetch users (indicators) from the 3rd party API endpoints. This method is invoked periodically.
- Make sure it’s unit-testable.
- Use the proxy configuration passed by the Cloud Exchange 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 the self.ssl_validation boolean to enable/disable validation of the SSL server certificate.
- Return the list of records that contain the data received from the API endpoint. Note: The list will contain Record objects.
- In the case of failure, raise an error or exception to 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.
- If the API endpoint does not support multiple requests (you can push it in one API call only), either plugin can skip the remaining indicators and raise a notification to the user to adjust the sharing filters or it can fail with the error of exceeded payload size.
def fetch_records(self): """Fetches users from a 3rd party API. Implement the logic of fetching users from 3rd party apis and return the list of objects netskope.integrations.cre.models.Records on successful fetch otherwise raises an exception. Returns: List[netskope.integrations.cre.models.Records]: List of Record objects received from the 3rd 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. config = self.configuration # get proxy settings dict, just the way requests module requires. proxy_dict = self.proxy # get the ssl_validation bool for enabling/disabling validation of SSL server certificates. ssl_validation = self.ssl_validation # How to use proxy dict and ssl_validation flag. resp = requests.get("www.example.com", proxies=proxy_dict, verify=ssl_validation) # Get the logger object for logging purpose. This logger object logs all the logs to mongodb # under the cre database logs collection. Log timestamp is automatically recorded by the logger library. # Supported logging levels are info, warn and error. logger = self.logger logger.info(f"{PLUGIN_NAME}: Finished fetching records") return records
def fetch_scores()
This is an abstract method of PluginBase Class.
- This method implements the logic to fetch scores(indicators) from the 3rd party API endpoints. This method is invoked periodically.
- Make sure it’s unit-testable.
- 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 the self.ssl_validation boolean 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.
- If the API endpoint does not support multiple requests (i.e. we can push it in one API call only) either plugin can skip the remaining indicators and raise a notification to the user to adjust the sharing filters or it can fail with the error of exceeded payload size.
- Make sure to map scores to an integer between 0-1000.
def fetch_scores(self): """Fetches users from a 3rd party API. Implement the logic of fetching risk scores from 3rd party apis and return the list of objects netskope.integrations.cre.models. Records a successful fetch otherwise raises an exception. Returns: List[netskope.integrations.cre.models.Records]: List of Record objects received from the 3rd 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. config = self.configuration # get proxy settings dict, just the way requests module requires. proxy_dict = self.proxy # get the ssl_validation bool for enabling/disabling validation of SSL server certificates. ssl_validation = self.ssl_validation # How to use proxy dict and ssl_validation flag. resp = requests.get("www.example.com", proxies=proxy_dict, verify=ssl_validation) # Get the logger object for logging purpose. This logger object logs all the logs to mongodb # under the cre database logs collection. Log timestamp is automatically recorded by the logger library. # Supported logging levels are info, warn and error. logger = self.logger logger.info(f"{PLUGIN_NAME}: Finished fetching records") return scores
def validate()
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.
- Validate 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, data): """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.cre.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: data (dict): Dict object having all the Plugin configuration parameters. Returns: netskope.integrations.cre.plugin_base.ValidateResult: ValidateResult object with success flag and message. """ self.logger.info(f"{PLUGIN_NAME}: Executing validate method for Sample plugin") if ( "secret_field_id1" not in data or not data["secret_field_id1"] or type(data["secret_field_id1"]) != str ): self.logger.error( "f{PLUGIN_NAME}: Validation error occurred Error: Secret Field1 is required with type string." ) return ValidationResult( success=False, message="Invalid Secret Field 1 provided.",
def get_actions()
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.
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 actions”, value=”generate”) ]
def get_action_fields()
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.
- Go to Manifest.json to see how fields are defined.
def get_action_fields(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 == “add”: return [ { “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()
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.
- 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.
- Validate the authentication parameters and the API endpoint to ensure the smooth execution of the plugin lifecycle.
- While validating, use strip() for configuration parameters like base URL, email, username, 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.cre.plugin_base.ValidateResult: ValidateResult object with success flag and message. """ if action.value not in [“add”, “remove”]: return ValidationResult( success=False, message=”Unsupported action provided.” ) if action.value == “add”: if action.parameters.get(“group_name”) is None: return ValidationResult( success=False, message=”Group Name should not be empty.” ) return ValidationResult( success=True, message=”Validation successful.” )
def execute_action()
This is an abstract method of PluginBase Class.
- This method validates executed actions.
- 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 3rd party API calls.
def execute_action(self, record: Record, action: Action): """Execute action on the user. Calls _add_to_group or _remove_to_group helper methods in the case of add or remove action Passes when action is generate if action.value == "generate": return if action.value == "add": group_id = action.parameters.get("group") self._add_to_group(user, group_id) self.logger.info(f"Added {user} to group to group ") )
Data Models
This section lists down the Data Models and their properties.
RecordType Model
- The RecordType model contains type of record ie user or host
- The record model has 2 fields: USER, and HOST which is a type of Record.
- You will interact with the model in the fetch_records and fetch_scores methods as you return a list of records.
Data Model Properties
Name | Type | Description |
---|---|---|
USER | string | Can be a user. |
HOST | string | Can be a host. |
from netskope.integrations.cre.models import RecordType, Record for user in users: res.append( Record(uid=user["email"], type=RecordType.USER, score=None) )
Record Model
- The record model contains user (indicator) records.
- The record model has 3 fields: uid, type, and score, which are email, type of user, and integer score value respectively.
- You will interact with the model in the fetch_users and fetch_scores methods as you return a list of records.
Data Model Properties
Name | Type | Description |
---|---|---|
uid | string | Email of the user. |
type | list[string] | Can be a user, a host, or both. |
score | int | Score of the user. It should be mapped to an integer that is in the range of 0 – 1000. |
from netskope.integrations.cre.models import Record Record(uid=each_user.get("userPrincipalName"), type=RecordType.USER, score=None )
Action Class
- 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 fetch scores, fetch users, validate_action, execute_actions, and get_action_fields.
Data Model Properties
Name | Type | Description |
---|---|---|
label | string | Display label of the action that is shown in the 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. |
syncAction | Bool | Synchronize actions when true. |
from netskope.integrations.cre.models import Action if action.value == "generate": return
ActionWithoutParams Class
- 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 the UI. |
value | string | Key to access it. |
from netskope.integrations.cre.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 Class
- 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 has to 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.cre.plugin_base import ValidationResult ValidationResult( success=True, message="Validation Successful for Sample plugin" )
Logging
URE 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, 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.
self.logger.error( f"{PLUGIN_NAME}: Error log-message goes here." ) self.logger.warn( f"{PLUGIN_NAME}: Warning log-message goes here." ) self.logger.info( f"{PLUGIN_NAME}: Info log-message goes here." )
Notifications
URE provides a handle of notification objects, which can be used to generate notifications on URE 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 which 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"{PLUGIN_NAME}: Info notification-message goes here." ) self.notifier.error( f"{PLUGIN_NAME}: Error notification-message goes here." ) self.notifier.warn( f"{PLUGIN_NAME}: Warning notification-message goes here." )
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. We 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 also 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. So make sure every function/module has a proper docstring added.
Unit Testing
Ensure unit testing to test small units of code in an isolated and deterministic fashion. Make sure that unit tests avoid performing communication with external APIs and use mocking. Make sure unit tests ensure the code coverage is more than 70%.
Environment Setup
In order to work with unit testing, the integration or automation script needs to be developed in Plugin Directory Structure. Use PIP to install all the required Python module dependencies required to run the setup. Before running the tests, make sure you install all the required dependencies mentioned in the requirements.txt of the l core repository.
Write Your Unit Tests
Make sure unit tests are written in a separate Python file named: <your_plugin_name>_test.py. Within the unit test file, each unit test function should be named: test_<your test case>. More information on writing unit tests and their format is available at the PyTest Docs.
Mocking
Use pytest-mock for mocking. pytest-mock is enabled by default and installed in the base environment mentioned above. To use a mocker object, simply pass it as a parameter to your test function. The mocker can then be used to mock both the plugin class object and also external APIs.
Example:
def test_netskope_fetch_users(mocker): mocker.patch("cre.plugins.sample.main.Sample.fetch_users") fetch_users_return_result = [ Record(uid="a@def.com", type=Record.user, score=400), Record(uid="b@pqr.com", type=Record.user, score=300), Record(uid="c@xyz.com", type=Record.user, score=750), ] samplePlugin.fetch_users.return_value = fetch_users_return_result sp = samplePlugin(None, None, None, logger) actual_fetch = sp.fetch_users() assert fetch_return_result == actual_fetch
To mock request module response for API calls (requests_mock Pytest plugin is installed with all the dependencies).
def test_fetch_users(requests_mock): endpoint_url = "https://example-api.com" mock_response_json = { { "@odata.context": "https://graph.microsoft.com/v1.0/$metadata#directoryObjects", "value": [ { "@odata.type": "#microsoft.graph.user", "id": "48a97c4b-65af-4cb7-90f9-c7835d5ecd8b", "userPrincipalName": "adaml@xyz.com" } ] } requests_mock.get(endpoint_url, json=mock_response_json) config_dict = { "tenant_name": "ecre-ccec-jpvasq", "client_id": "ue4y-3id8-mnx87q", "client_secret": "d9oe-ple7-nfi34w", "api_token": "token", } sp = SamplePlugin(config_dict, None, None, logger) actual_users_lists = sp.fetch_users( endpoint_url, config_dict['api_token'] ) users_mock = [ Record(uid="a@def.com", type=Record.user, score=400), Record(uid="b@pqr.com", type=Record.user, score=300), Record(uid="c@xyz.com", type=Record.user, score=750), ] assert len(actual_users_lists) == len(users_mock) for i in range(len(actual_users_lists)): assert actual_ind_lists[i].value == users_mock[i].value
Running Your Unit Tests
$ PYTHONPATH=. pytest
Plugin Deployment on CE
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
Upload the Plugin
To deploy this zip or tar.gz on Cloud Exchange platform:
- Log in to the Cloud Exchange platform.
- Goto Settings > Plugin.
- Click Add New Plugin.
- Click Browse.
- Select your zip or tar.gz file.
- Click Upload.
Add a Repo
To deploy your plugin in the Cloud Exchange platform, you can add your repo in the Cloud Exchange platform by following these steps:
- Log in to Cloud Exchange platform.
- Go to Settings > Plugin Repository.
- Click Configure New Repository.
- Provide a Repository name, Repository URL, Username, and Personal Access Token.
- Click Save.
- Go to Settings > Plugins.
- Select a Repository name from the Repository dropdown.
Deliverables
Plugin Guide
- Make sure to update the plugin guide with every release.
- The Plugin guide should have this content:
- Compatibility
- Release Notes
- Description
- Prerequisites
- Permissions
- Authorization of the product
- Configuration of Netskope User Risk Exchange plugin
- Configuration of the third-party User Risk Exchange plugin that we have developed.
- Configuration of a business rule
- Configuration of an action
- Validation.
Demo Video
After the successful development of the User Risk Exchange plugin, create a demo video and show the end-to-end workflow of the plugin.