Skip to content

Commit

Permalink
Added 'filters' option for sc plugins #718
Browse files Browse the repository at this point in the history
  • Loading branch information
@SteveMcGrath
SteveMcGrath committed Nov 21, 2024
1 parent 7e13674 commit a290aaa
Show file tree
Hide file tree
Showing 4 changed files with 266 additions and 131 deletions.
1 change: 1 addition & 0 deletions pyproject.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ dev-dependencies = [
# VCR for the test suite.
"urllib3==1.26.20",
"rich>=13.8.1",
"ptpython>=3.0.29",
]


Expand Down
165 changes: 105 additions & 60 deletions tenable/sc/plugins.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
'''
"""
Plugins
=======
Expand All @@ -11,64 +11,87 @@
.. rst-class:: hide-signature
.. autoclass:: PluginAPI
:members:
'''
from .base import SCEndpoint, SCResultsIterator
"""

import json

from tenable.errors import UnexpectedValueError

from .base import SCEndpoint, SCResultsIterator


class PluginResultsIterator(SCResultsIterator):
pass


class PluginAPI(SCEndpoint):
def _constructor(self, **kwargs):
'''
"""
Constructs the plugin query.
'''
"""

if 'fields' in kwargs:
kwargs['fields'] = ','.join([self._check('field', f, str)
for f in self._check('fields', kwargs['fields'], list)])
kwargs['fields'] = ','.join(
[
self._check('field', f, str)
for f in self._check('fields', kwargs['fields'], list)
]
)

if 'filter' in kwargs:
# break down the filter tuple into the various query parameters
# that the plugin api expects.
self._check('filter', kwargs['filter'], tuple)
if len(kwargs['filter']) != 3:
raise UnexpectedValueError(
'the filter tuple must be name, operator, value.')
kwargs['filterField'] = self._check('filter:field', kwargs['filter'][0], str)
kwargs['op'] = self._check('filter:operator', kwargs['filter'][1], str,
choices=['eq', 'gt', 'gte', 'like', 'lt', 'lte'])
'the filter tuple must be name, operator, value.'
)
kwargs['filterField'] = self._check(
'filter:field', kwargs['filter'][0], str
)
kwargs['op'] = self._check(
'filter:operator',
kwargs['filter'][1],
str,
choices=['eq', 'gt', 'gte', 'like', 'lt', 'lte'],
)
kwargs['value'] = self._check('filter:value', kwargs['filter'][2], str)
del kwargs['filter']

#filters = kwargs.pop('filters', [])
#kwargs['filters'] = []
#for filter in filters:
# if isinstance(filter, tuple):
# kwargs['filters'].append({
# 'filterField': filter[0],
# 'filterOperator': filter[1],
# 'filterString': filter[2],
# })
# elif isinstance(filter, dict):
# kwargs['filters'].append(filter)

if 'filters' in kwargs:
filters = self._check('filters', kwargs.pop('filters', []), list)
kwargs['filters'] = []
for filter in filters:
self._check('filters:filter', filter, tuple)
f = {
'filterField': self._check('filter:name', filter[0], str),
'filterOperator': self._check(
'filter:op',
filter[1],
str,
choices=['eq', 'gt', 'gte', 'like', 'lt', 'lte'],
),
'filterString': self._check('filter:value', filter[2], str),
}
kwargs['filters'].append(f)
kwargs['filters'] = json.dumps(kwargs['filters'])

if 'sort_field' in kwargs:
# convert the snake_cased variant of the parameter to the camelCased
# variant that the API expects to see.
kwargs['sortField'] = self._check(
'sort_field', kwargs['sort_field'], str)
kwargs['sortField'] = self._check('sort_field', kwargs['sort_field'], str)
del kwargs['sort_field']

if 'sort_direction' in kwargs:
# convert the snake_cased variant of the parameter to the camelCased
# variant that the API expects to see.
kwargs['sortDirection'] = self._check(
'sort_direction', kwargs['sort_direction'], str,
choices=['ASC', 'DESC'], case='upper')
'sort_direction',
kwargs['sort_direction'],
str,
choices=['ASC', 'DESC'],
case='upper',
)
del kwargs['sort_direction']

if 'since' in kwargs:
Expand All @@ -77,10 +100,21 @@ def _constructor(self, **kwargs):

if 'type' in kwargs:
# Validate that the plugin type is what's expected.
self._check('type', kwargs['type'], str, choices=[
'active', 'all', 'compliance', 'custom',
'lce', 'notPassive', 'passive'
], default='all')
self._check(
'type',
kwargs['type'],
str,
choices=[
'active',
'all',
'compliance',
'custom',
'lce',
'notPassive',
'passive',
],
default='all',
)

# While the iterator will handle the offset & limits, a raw json result
# may be requested instead.
Expand All @@ -90,7 +124,8 @@ def _constructor(self, **kwargs):

if 'limit' in kwargs:
kwargs['endOffset'] = self._check(
'limit', kwargs['limit'], int) + kwargs.get('startOffset', 0)
'limit', kwargs['limit'], int
) + kwargs.get('startOffset', 0)
del kwargs['limit']

# Pages and json_result parameters should be removed from the document
Expand All @@ -105,7 +140,7 @@ def _constructor(self, **kwargs):
return kwargs

def list(self, **kwargs):
'''
"""
Retrieves the list of plugins.
:sc-api:`plugins: list <Plugin.htm#PluginRESTReference-/plugin>`
Expand All @@ -117,6 +152,8 @@ def list(self, **kwargs):
A filter tuple for which to filter the plugins. Filter tuples
must be ``('name', 'operator', 'value')`` and follow a similar
yet different format to the analysis filters.
filters (list[tuple], optional):
A list of filter tuples. Filters are treated as a logical ``OR``.
limit (int, optional):
How many records should be returned in each page of data. If
none is specified, the default is 1000 records.
Expand Down Expand Up @@ -156,7 +193,7 @@ def list(self, **kwargs):
For just the active plugins, we'd run:
>>> plugins = sc.plugins.list(type='active')
'''
"""
offset = self._check('offset', kwargs.get('offset', 0), int)
limit = self._check('limit', kwargs.get('limit', 1000), int)
pages = self._check('pages', kwargs.get('pages'), int)
Expand All @@ -165,15 +202,17 @@ def list(self, **kwargs):

if json_result:
return self._api.get('plugin', params=query).json()['response']
return PluginResultsIterator(self._api,
_resource='plugin',
_offset=offset,
_limit=limit,
_query=query,
_pages_total=pages)
return PluginResultsIterator(
self._api,
_resource='plugin',
_offset=offset,
_limit=limit,
_query=query,
_pages_total=pages,
)

def details(self, plugin_id, fields=None):
'''
"""
Returns the details for a specific plugin.
:sc-api:`plugins: details <Plugin.htm#PluginRESTReference-/plugin/{id}>`
Expand All @@ -188,16 +227,17 @@ def details(self, plugin_id, fields=None):
Examples:
>>> plugin = sc.plugins.detail(19506)
>>> pprint(plugin)
'''
"""
params = dict()
if fields:
params['fields'] = ','.join([self._check('field', f, str) for f in fields])

return self._api.get('plugin/{}'.format(self._check('plugin_id', plugin_id, int)),
params=params).json()['response']
return self._api.get(
'plugin/{}'.format(self._check('plugin_id', plugin_id, int)), params=params
).json()['response']

def family_list(self, **kwargs):
'''
"""
Returns the list of plugin families.
:sc-api:`plugin-families: list <Plugin-Family.htm#PluginFamilyRESTReference-/pluginFamily>`
Expand Down Expand Up @@ -226,12 +266,12 @@ def family_list(self, **kwargs):
Examples:
>>> for fam in sc.plugins.family_list():
... pprint(fam)
'''
"""
query = self._constructor(**kwargs)
return self._api.get('pluginFamily', params=query).json()['response']

def family_details(self, plugin_id, fields=None):
'''
"""
Returns the details for the specified plugin family.
:sc-api:`plugin-family: details <Plugin-Family.htm#PluginFamilyRESTReference-/pluginFamily/{id}>`
Expand All @@ -248,16 +288,18 @@ def family_details(self, plugin_id, fields=None):
Examples:
>>> family = sc.plugins.family_details(10)
>>> pprint(family)
'''
"""
params = dict()
if fields:
params['fields'] = ','.join([self._check('field', f, str)
for f in fields])
params['fields'] = ','.join([self._check('field', f, str) for f in fields])

return self._api.get('pluginFamily/{}'.format(self._check('plugin_id', plugin_id, int)), params=params).json()['response']
return self._api.get(
'pluginFamily/{}'.format(self._check('plugin_id', plugin_id, int)),
params=params,
).json()['response']

def family_plugins(self, plugin_id, **kwargs):
'''
"""
Retrieves the plugins for the specified family.
:sc-api:`plugin-family: plugins <Plugin-Family.htm#PluginFamilyRESTReference-/pluginFamily/{id}/plugins::GET>`
Expand Down Expand Up @@ -296,7 +338,7 @@ def family_plugins(self, plugin_id, **kwargs):
>>> plugins = sc.plugins.family_plugins(10)
>>> for plugin in plugins:
... pprint(plugin)
'''
"""
offset = self._check('offset', kwargs.get('offset', 0), int)
limit = self._check('limit', kwargs.get('limit', 1000), int)
pages = self._check('pages', kwargs.get('pages'), int)
Expand All @@ -305,10 +347,13 @@ def family_plugins(self, plugin_id, **kwargs):

if json_result:
return self._api.get('plugin', params=query).json()['response']
return PluginResultsIterator(self._api,
_resource='pluginFamily/{}/plugins'.format(
self._check('plugin_id', plugin_id, int)),
_offset=offset,
_limit=limit,
_query=query,
_pages_total=pages)
return PluginResultsIterator(
self._api,
_resource='pluginFamily/{}/plugins'.format(
self._check('plugin_id', plugin_id, int)
),
_offset=offset,
_limit=limit,
_query=query,
_pages_total=pages,
)
Loading

0 comments on commit a290aaa

Please sign in to comment.