The AMP Leaf

class cmaple.amp.amp.AMP(**kwargs)

This class defines the API interface for the FMC.

Inherits generic REST functionality from RestBase.

Overrides methods in RestBase where necessary.

Method names not beginning with “_” are made available to cmaple_cli.py for use in operations config files.

__init__(**kwargs)

__init__ receives a kwargs dict to define parameters. This allows __init__ to pass these parameters to the superclass.

Returns an AMP leaf object.

Parameters

AMP_host: string, keyword, default=None

The ip address or fqdn of ThreatGrid

AMP_API_client_ID: string, keyword, default=None

The AMP API client ID

AMP_API_key: string, keyword, default=None

The AMP API key.

API_path_delimiter: string, keyword, default=’/’

The default delimiter for the API path.

API_version: string, keyword, default=’v2’

The API version supported by the target ThreatGrid.

verify: boolean, keyword, default=False

If True, verify the certificate. If False disable verification.

default_get_item_limit: integer, keyword, default=400

The default number of items to request in a GET request.

rpm_retries: integer, keyword, default=5

The number of times to retry in response to a 429 error.

backoff_timer: integer, keyword, default=30

The interval to wait between retry attempts

persist_responses: boolean, keyword, default=True

If True, responses will be pickle persisted by url into the leaf’s working directory.

restore_responses: boolean, keyword, default=False

If True, pickled persistent responses will be restored prior to all other operations.

leaf_dir: string, keyword, default=None

Provided by CMapleTree when this leaf type is instantiated. Contains the directory where working files for the leaf instance are stored.

_request_wrapper(recursed=False, **kwargs)

Wraps all requests for an AMP leaf in order to handle AMP specifics. This should only be called by internal methods.

Parameters

recursed: boolean, keyword, default=False

Signals if this is the top level call.

**kwargs: dictionary

Used to pass through arguments to wrapped methods.

_get_child_urls(response_dict, parent_url)

This method retrieves the url for all child objects of this response. This should only be called by internal methods.

Returns: child_url for this anomalous type

Parameters

response_dict: dictionary

The response for which to find child urls.

parent_url: string

The parent url of this response. Used to prevent circular object references.

GET_API_path(url, include_filter_regex=None, exclude_filter_regex=None, stop_on_error=False, use_cache=False, responses_dict=None, get_item_limit=None)

Wrapper for a REST GET request.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for the GET.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url : string

The starting url of the path to walk. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

use_cache: boolean, keyword, default=False

If set to True, any path that has already been requested will not generate a new request

include_filter_regex: string, keyword, default=None

A regex string defining which urls to include in walk.

exclude_filter_regex: string, keyword, default=None

A regex string defining which urls to exclude from walk.

stop_on_error: boolean, keyword, default=False

If set to True, walk will halt when a non positive status code response is received.

get_item_limit: integer, keyword, default=25

Specifies the number of items to return for each GET request.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

GET_responses_by_jsonpath(jsonpath, responses_dict=None)

Returns responses matching a jsonpath query.

***Inherited from RestBase…***

Returns - The matching responses.

Parameters

jsonpath: string

The jsonpath query to match responses.

responses_dict: dictionary, keyword, default=None

The response dictionary to query.

_collect_responses(url, response_dict, responses_dict)

Utility method called by wrappers to request all pages for a given url. Normally not called directly.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url: string

The url of the path to GET. Must be a fully valid FMC api “GET” path.

response_dict: dictionary

A dictionary reference updated by this method to include all responses.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

_prepare_url(url=None, params=None)

Optional override in parent class. Prepares the url

***Inherited from RestBase…***

Prepares the url by replacing parameter placeholders with values. This should only be called by internal methods.

Parameters

url: string, keyword, default=None

The url to prepare.

params: dictionary, keyword, default=None

The parameters dictionary.

_prepare_url_for_migration()

Must override in parent class

***Inherited from RestBase…***

_recurse_API_child_gets(url, use_cache=True, end_path_regex=None, include_filter_regex=None, exclude_filter_regex=None, stop_on_error=False, filtered=False, cache_hit=False, get_item_limit=None, responses_dict=None, parent_url='')

Handles recursion of a given url path. Normally not called directly but from a wrapper method. Begins at given API url path and recursively GET walks path and child paths until complete. Automatically handles pagination and discovery of child urls.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for all url path GETs.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url : string

The starting url of the path to recurse. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

use_cache: boolean, keyword, default=False

If set to True, any path that has already been requested will not generate a new request

include_filter_regex: string, keyword, default=None

A regex string defining which urls to include in walk.

exclude_filter_regex: string, keyword, default=None

A regex string defining which urls to exclude from walk.

stop_on_error: boolean, keyword, default=False

If set to True, walk will halt when a non positive status code response is received.

get_item_limit: integer, keyword, default=25

Specifies the number of items to return for each GET request.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

parent_url: string, keyword, default=’‘

Specifies the url of the parent of this child url. Used to prevent recursion loops where the API model contains a circular object.

_set_json_properties_by_objectpath(json_dict=None, properties_dict=None)

Sets properties in json_dict from properties_dict.

***Inherited from RestBase…***

Returns - The modified json_dict.

Parameters

json_dict: dictionary, keyword, default=None

The json_dict to modify.

properties_dict: dictionary, keyword, default=None

The properties to modify {property:value}.

chained_smart_get(base_paths=None, params=None, responses_dict=None, query_dict=None)

Gets threatgrid samples.

Returns: a responses dictionary

Parameters

sample_search_paths: dictionary, keyword, default=None

Defines the search parameters (e.g. checksum=<sha256>).

params: dictionary, keyword, default=None

Defines the search scope parameters (e.g. before=<strftime>).

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

get_all_items(url, use_cache=True, end_path_regex=None, include_filter_regex=None, exclude_filter_regex=None, stop_on_error=False, filtered=False, cache_hit=False, get_item_limit=None, responses_dict=None)

Performs a get to retrieve the “Items” listing for the url.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url : string

The url for which to retrieve the items list. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

use_cache: boolean, keyword, default=False

If set to True, any path that has already been requested will not generate a new request

include_filter_regex: string, keyword, default=None

A regex string defining which urls to include in walk.

exclude_filter_regex: string, keyword, default=None

A regex string defining which urls to exclude from walk.

stop_on_error: boolean, keyword, default=False

If set to True, walk will halt when a non positive status code response is received.

get_item_limit: integer, keyword, default=25

Specifies the number of items to return for each GET request.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

get_json_request(url, responses_dict=None)

Generic wrapper for a REST API GET request.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for the Post.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url: string

The url of the path to GET. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

or_migrate_config()

Must override in parent class

***Inherited from RestBase…***

post_csv_template(url=None, file_path=None)

Reads a csv file containing flatlined records (flattened with output_transforms.flatten_json(json_dict) and posts each record individually to the target.

***Inherited from RestBase…***

Returns - No return value.

Parameters

url: string, keyword, default=None

The target url to post records.

file_path: string, keyword, default=None

The full path to the file containing the csv records.

post_json_request(url, json_dict, responses_dict=None)

Generic wrapper for a REST API Post request.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for the Post.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url: string

The url of the path to POST. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

json_dict: dictionary, argument

The Python dictionary containing the request json

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

put_json_request(url, json_dict, responses_dict=None)

Generic wrapper for a REST API Put request.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for the Post.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url: string

The url of the path to POST. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

json_dict: dictionary, argument

The Python dictionary containing the request json

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

query_json_field(query_field=None, json_to_query=None)

Returns a list of objects matching the query_field query.

***Inherited from RestBase…***

Returns - A list of the matching objects

Parameters

query_field: string, keyword, default=None

The field for which to query fields.

json_to_query: dictionary, keyword, default=None

The json dictionary to query.

query_json_field_from_url(query_url=None, json_to_query=None)

Returns a list of json fields matching an objectpath query.

***Inherited from RestBase…***

Returns - A list of the matching fields.

Parameters

query_url: string, keyword, default=None

The url for which to query fields.

json_to_query: dictionary, keyword, default=None

The json dictionary to query.

query_with_list(query_url=None, query_list=None, responses_dict=None)

Iterates over and substitutes the values in query_list in query_url

***Inherited from RestBase…***

Returns - All responses obtained.

Parameters

query_url: string, keyword, default=None

The url for which to query fields.

query_list: list, keyword, default=None

The list of values to iterate over and substitute in query_url.

set_json_properties(json_dict=None, properties_dict=None)

Sets properties in json_dict from properties_dict.

***Inherited from RestBase…***

Returns - The modified json_dict.

Parameters

json_dict: dictionary, keyword, default=None

The json_dict to modify.

properties_dict: dictionary, keyword, default=None

The properties to modify {property:value}.

walk_API_path_gets(url, end_path_regex=None, include_filter_regex=None, exclude_filter_regex=None, use_cache=True, stop_on_error=False, get_item_limit=None, responses_dict=None)

Begins at given API url path and recursively GET walks path and child paths until complete.

***Inherited from RestBase…***

Returns a Python dictionary object containing the response results for all url path GETs.

By default stores all responses in self.responses_dict unless a dictionary is passed in using the responses_dict parameter.

Parameters

url : string

The starting url of the path to walk. Must be a fully valid FMC api “GET” path. url can include the host prefix or start from the resource path. If the host prefix is missing, it will be added automatically.

use_cache: boolean, keyword, default=False

If set to True, any path that has already been requested will not generate a new request

include_filter_regex: string, keyword, default=None

A regex string defining which urls to include in walk.

exclude_filter_regex: string, keyword, default=None

A regex string defining which urls to exclude from walk.

stop_on_error: boolean, keyword, default=False

If set to True, walk will halt when a non positive status code response is received.

get_item_limit: integer, keyword, default=25

Specifies the number of items to return for each GET request.

responses_dict: dictionary, keyword, default=None

Allows the caller to override the default behavior to store responses in the self.responses_dict. Useful if caller would like to keep the responses isolated.

write_csv_template_from_response(response_json=None, file=<colorama.ansitowin32.StreamWrapper object>)

Flatlines response_json and writes to a csv record.

***Inherited from RestBase…***

Returns - The csv records.

Parameters

response_json: list, keyword, default=None

The list of responses to write to csv.

file: file_handle, keyword, default=sys.stdout

The file_handle target for the csv records.