Threat Exchange Custom Plugin Developers Guide

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

Sample Changelog.md

# 1.0.1
## Fixed
- Fixed pagination when there are more than 10k Logs
# 1.0.0
## Added
- Initial release.

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

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 field on the plugin configuration page.

Sample JSON

"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 on the plugin configuration page.

Sample JSON

"configuration": [
{
"label": "Type of Threat data to pull",
"key": "ioc_type",
"type": "choice",
"choices": [
    {
      "key": "Both",
      "value": ["malware", “malsite”]
    },
    {
      "key": "Malware",
      "value": "malware"
    },
    {
      "key": "Malsite",
      "value": "malsite"
    }
  ]
 },
]

Plugin configuration view:

After selecting the input.

Multichoice Parameter

Use this parameter for storing multiple choice values. This parameter will have a dropdown on the plugin configuration page with ability to select multiple values.

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.

Plugin Configuration View:

This python file contains the core implementation of the plugin.

Standard Imports

from netskope.integrations.cte.plugin_base import PluginBase, ValidationResult, PushResult
from netskope.integrations.cte.models import Indicator, IndicatorType
from netskope.integrations.cte.models.business_rule import Action, ActionWithoutParams

PluginBase Variables

PluginBase provides access to variables which can be used during the plugin lifecycle

Methods. Below is the list of variables.

"""Sample plugin implementation.
This is a sample implementation of base PluginBase class. Which explains the concrete implemetation
of the base class.
"""
from netskope.integrations.cte.plugin_base import PluginBase, ValidationResult, PushResult
from netskope.integrations.cte.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 pulling and pushing threat information.
   This class is responsible for implementing pull, push and validate methods with proper return types,
   so that it's lifecycle execution can be scheduled by the CTE core engine.
   """

Def Pull()

This is an abstract method of PluginBase Class.

def pull(self):
"""Pull the Threat information from the 3rd part Threat Intel systems.
Implement the logic of pulling Threat data from 3rd party apis and return the list of objects netskope.integrations.cte.models.Indicators on successful pull otherwise raises an exception.
Returns:
    List[netskope.integrations.cte.models.Indicators]: List of indicator objects received from the   3rd party Threat Intel Systems.
"""
# Load all the configured plugin parameters as python dict object.
# 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
start_time = self.last_run_at  # datetime.datetime object.
# How to use proxy dict and ssl_validation flag.
resp = requests.get("www.example.com", proxies=proxy_dict, verify=ssl_validation)  # noqa: F841
# Get the logger object for logging purpose. This logger object logs all the logs to mongodb
# under the cte 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}: Starting Pulling data for sample plugin.")
indicator_list = self.pull_data_from_3rd_party_api(config, logger)
logger.info(f"{PLUGIN_NAME}:logger.info("{PLUGIN_NAME}: Finished pulling data")
return indicator_list

Def Push()

This is an abstract method of PluginBase Class.

def push(self, indicators: List[Indicator]):
    """Push the Indicator list to the 3rd party Threat Intel systems.
    Implement the logic of spliting the indicators list according to their type and push the data
    to the 3rd party APIs. This method will be invoked while sharing the Threat information with 3rd party.
    Args:
        indicators (List[netskope.integrations.cte.models.Indicators]): List of Indicator objects to be pushed.
    Returns:
        netskope.integrations.cte.plugin_base.PushResult: PushResult object with success flag and Push result message.
    """
    # Load all the configured plugin parameters as python dict object.
    # 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)   # noqa: F841
    # Get the logger object for logging purpose. This logger object logs all the logs to mongodb
    # under the cte 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}: Starting Pulling data for sample plugin.")
    push_result = self.push_data_to_3rd_party_api(config, logger, indicators)
    logger.info("f"{PLUGIN_NAME}: Finished Pushing data for sample plugin.")
    return push_result

Def Validate()

This is an abstract method of PluginBase Class.

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.cte.plugin_base.ValidationResult object with success = True in the case
  of successful validation and success = False and a error message in the case of failure.
  Args:
      data (dict): Dict object having all the Plugin configuration parameters.
  Returns:
      netskope.integrations.cte.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.",
      )
else:
return ValidationResult(
        success=True, message="Validation Successful for Sample plugin"
   )

Def get_actions()

This is an abstract method of PluginBase Class.

If manifest has push_supported=false:

def get_actions(self):
"""Get available actions.
Returns:
 List[ActionWithoutParams]: List of ActionWithoutParams objects that are supported by the plugin.
"""
 return []

If manifest has push_supported=true:

def get_actions(self): 
"""Get available actions.
Returns:
List[ActionWithoutParams]: List of ActionWithoutParams objects that are supported by the plugin.
"""
return [
ActionWithoutParams(label=”Share Indicators”, value=”share”)
ActionWithoutParams(label=”Add to Group”, value=”add”)
]

Def get_action_fields():

This is an abstract method of PluginBase Class.

If manifest has push_supported=false:

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.
"""
 return []

If manifest has push_supported=true:

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”: “text”,
       “default”: “”,
       “mandatory”: True,
       “description”: “Name of group.”
     }
 ]
else:
 return []

Def validate_action():

This is an abstract method of PluginBase Class.

If manifest has push_supported=false:

def validate_action(self, action: Action):
"""Validate Action Parameters.
Args: 
action (Action): Action object having all the configurable parameters.
Return:
netskope.integrations.cte.plugin_base.ValidateResult: ValidateResult object with success flag and message.
"""
return ValidationResult(success=True, message=”Validation successful.”)

If manifest has push_supported=true:

def validate_action(self, action: Action):
"""Validate Action Parameters.
Args: 
action (Action): Action object having all the configurable parameters.
Return:
netskope.integrations.cte.plugin_base.ValidateResult: ValidateResult object with success flag and message.
"""
if action.value not in [“share”, “add”]:
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.”
 )

This section lists down the Data Models and their properties.

IndicatorType Model

Data Model Properties

from netskope.integrations.cte.models import ( Indicator,
IndicatorType,
SeverityType,
)
Indicator(
value=behavior_info.get("ioc_value"),
type=IndicatorType.SHA256,
comments=behavior_info.get("ioc_description", ""),
firstSeen=datetime.datetime.strptime(
behavior_info.get("timestamp"),
"%Y-%m-%dT%H:%M:%SZ",
),
lastSeen=datetime.datetime.strptime(
behavior_info.get("timestamp"),
"%Y-%m-%dT%H:%M:%SZ",
),
severity=self.get_severity_from_int(behavior_info.get("severity", 0)),
)

SeverityType Model

Data Model Properties

from netskope.integrations.cte.models import SeverityType
if type(severity) is not int or severity == 0:
return SeverityType.UNKNOWN
if 10 <= severity <= 39:
return SeverityType.LOW
if 40 <= severity <= 69:
return SeverityType.MEDIUM
if 70 <= severity <= 89:
return SeverityType.HIGH
if 90 <= severity <= 100:
return SeverityType.CRITICAL
return SeverityType.UNKNOWN

ActionWithoutParams Model

Data Model Properties

from netskope.integrations.cte.models.business_rule import
ActionWithoutParams
ActionWithoutParams(
label="Add to Suspicious Object List",
value="suspicious_object",
)

Data Model Properties

from netskope.integrations.cte.models import Indicator, IndicatorType  
Indicator(
      value="md5hash",         # md5 hash value of the indicator.
      type=IndicatorType.MD5,  # Type of indicator.
      test=True,               # Indicates whether it's test indicator or not. Defaults to False.
      reputation=7,            # Reputation score of Indicator. Defaults to 5.
      # Time after which Indicator will be marked as inactive.
      expiresAt=datetime.datetime(2022, 12, 23, 15, 22, 52, 667126),
      # Time when the Indicator was discovered first time. Defaults to current time.
      firstSeen=datetime.datetime(2017, 1, 11, 15, 22, 52, 667126),
      # Time when the Indicator was discovered last time. Defaults to current time.
      lastSeen=datetime.datetime(2019, 12, 23, 15, 22, 52, 667126),
      # Comment which gives more information about the indicator.
      comments="Indicator explanation",
  ),

PushResult Class

Data Model Properties

from netskope.integrations.cte.plugin_base import PushResult
PushResult(
    success=True, 
    message=" Successfully pushed data to 3rd party."
)

ValidationResult Class

Data Model Properties

from netskope.integrations.cte.plugin_base import ValidationResult
ValidationResult(
           success=True,
           message="Validation Successfull for Sample plugin"
       )

Cloud Exchange provides a handle of logger object for logging.

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."
 )