Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/api/io/recast_rules.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
.. automodule:: tenable.io.recast_rules
10 changes: 10 additions & 0 deletions tenable/io/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
permissions
plugins
policies
recast_rules
remediation_scans
scanner_groups
scanners
Expand Down Expand Up @@ -71,6 +72,7 @@
from .permissions import PermissionsAPI
from .plugins import PluginsAPI
from .policies import PoliciesAPI
from .recast_rules import RecastRulesAPI
from .remediation_scans import RemediationScansAPI
from .scanner_groups import ScannerGroupsAPI
from .scanners import ScannersAPI
Expand Down Expand Up @@ -354,6 +356,14 @@ def policies(self):
"""
return PoliciesAPI(self)

@property
def recast_rules(self):
"""
The interface object for the
:doc:`Tenable Vulnerability Management Recast Rules APIs <recast_rules>`.
"""
return RecastRulesAPI(self)

@property
def scanner_groups(self):
"""
Expand Down
139 changes: 139 additions & 0 deletions tenable/io/recast_rules.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
"""
Recast Rules
============

The following methods allow for interaction with the Tenable Vulnerability
Management Recast Rules API endpoints.

Methods available on ``tio.recast_rules``:

.. rst-class:: hide-signature
.. autoclass:: RecastRulesAPI
:members:
"""

from typing import Any

from tenable.utils import scrub

from .base import TIOEndpoint


class RecastRulesAPI(TIOEndpoint):
"""The Tenable Vulnerability Management Recast Rules API."""

_resource_types = ['HOST', 'HOST_AUDIT', 'WEBAPP']

def _payload(
self,
resource_type: str,
rule_value: dict[str, Any],
rule_name: str | None = None,
description: str | None = None,
expires_at: str | None = None,
disabled_details: dict[str, Any] | None = None,
) -> dict[str, Any]:
payload = {
'resource_type': self._check(
'resource_type', resource_type, str, choices=self._resource_types
),
'rule_value': self._check('rule_value', rule_value, dict),
}
optional = {
'rule_name': (rule_name, str),
'description': (description, str),
'expires_at': (expires_at, str),
'disabled_details': (disabled_details, dict),
}
for name, (value, expected_type) in optional.items():
if value is not None:
payload[name] = self._check(name, value, expected_type)
return payload

def create(
self,
resource_type: str,
rule_value: dict[str, Any],
rule_name: str | None = None,
description: str | None = None,
expires_at: str | None = None,
disabled_details: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Create a recast rule."""
return self._api.post(
'recast/rules',
json=self._payload(
resource_type,
rule_value,
rule_name,
description,
expires_at,
disabled_details,
),
).json()

def search(
self,
resource_type: list[str] | None = None,
filter: dict[str, Any] | None = None,
limit: int = 100,
sort: list[str] | None = None,
next: str | None = None,
) -> dict[str, Any]:
"""Search recast rules."""
limit = self._check('limit', limit, int)
if not 1 <= limit <= 500:
raise ValueError('limit must be between 1 and 500')

payload: dict[str, Any] = {'limit': limit}
if resource_type is not None:
resource_type = self._check('resource_type', resource_type, list)
payload['resource_type'] = [
self._check('resource_type', item, str, choices=self._resource_types)
for item in resource_type
]
if filter is not None:
payload['filter'] = self._check('filter', filter, dict)
if sort is not None:
payload['sort'] = self._check('sort', sort, list)
if next is not None:
payload['next'] = self._check('next', next, str)

return self._api.post('recast/rules/search', json=payload).json()

def details(self, rule_id: str) -> dict[str, Any]:
"""Retrieve a recast rule by UUID."""
return self._api.get(f'recast/rules/{scrub(self._check("rule_id", rule_id, "uuid"))}').json()

def edit(
self,
rule_id: str,
resource_type: str,
rule_value: dict[str, Any],
rule_name: str | None = None,
description: str | None = None,
expires_at: str | None = None,
disabled_details: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Update a recast rule by UUID."""
return self._api.put(
f'recast/rules/{scrub(self._check("rule_id", rule_id, "uuid"))}',
json=self._payload(
resource_type,
rule_value,
rule_name,
description,
expires_at,
disabled_details,
),
).json()

def delete(self, rule_id: str) -> dict[str, Any]:
"""Delete a recast rule by UUID."""
return self._api.delete(
f'recast/rules/{scrub(self._check("rule_id", rule_id, "uuid"))}'
).json()

def filters(self) -> dict[str, Any]:
"""Retrieve the available recast rule filters."""
return self._api.get('recast/rules/filters').json()
113 changes: 113 additions & 0 deletions tests/io/test_recast_rules.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
"""Tests for the Tenable Vulnerability Management Recast Rules API."""

import pytest
import responses
from responses import matchers

from tenable.io import TenableIO
from tenable.errors import UnexpectedValueError


RULE_ID = '4c931fce-699c-4052-a43c-c953e71dd37b'
RULE_VALUE = {'filter': {'field': 'plugin_id', 'operator': 'eq', 'value': '19506'}}


@pytest.fixture
def api():
return TenableIO('access', 'secret')


@responses.activate
def test_recast_rules_create(api):
responses.add(
responses.POST,
'https://cloud.tenable.com/recast/rules',
json={'id': RULE_ID},
status=200,
match=[
matchers.json_params_matcher(
{'resource_type': 'HOST', 'rule_value': RULE_VALUE}
)
],
)

assert api.recast_rules.create('HOST', RULE_VALUE) == {'id': RULE_ID}


@responses.activate
def test_recast_rules_search(api):
responses.add(
responses.POST,
'https://cloud.tenable.com/recast/rules/search',
json={'items': [{'id': RULE_ID}]},
status=200,
match=[
matchers.json_params_matcher(
{
'resource_type': ['HOST'],
'filter': {'field': 'plugin_id'},
'limit': 25,
'sort': ['created_at:desc'],
'next': 'cursor',
}
)
],
)

result = api.recast_rules.search(
resource_type=['HOST'],
filter={'field': 'plugin_id'},
limit=25,
sort=['created_at:desc'],
next='cursor',
)
assert result == {'items': [{'id': RULE_ID}]}


@responses.activate
def test_recast_rules_details_edit_delete_and_filters(api):
responses.add(
responses.GET,
f'https://cloud.tenable.com/recast/rules/{RULE_ID}',
json={'id': RULE_ID},
status=200,
)
responses.add(
responses.PUT,
f'https://cloud.tenable.com/recast/rules/{RULE_ID}',
json={'id': RULE_ID, 'updated': True},
status=200,
match=[
matchers.json_params_matcher(
{'resource_type': 'HOST', 'rule_value': RULE_VALUE}
)
],
)
responses.add(
responses.DELETE,
f'https://cloud.tenable.com/recast/rules/{RULE_ID}',
json={'deleted': True},
status=200,
)
responses.add(
responses.GET,
'https://cloud.tenable.com/recast/rules/filters',
json={'filters': []},
status=200,
)

assert api.recast_rules.details(RULE_ID) == {'id': RULE_ID}
assert api.recast_rules.edit(RULE_ID, 'HOST', RULE_VALUE) == {
'id': RULE_ID,
'updated': True,
}
assert api.recast_rules.delete(RULE_ID) == {'deleted': True}
assert api.recast_rules.filters() == {'filters': []}


def test_recast_rules_validate_arguments(api):
with pytest.raises(UnexpectedValueError):
api.recast_rules.create('INVALID', RULE_VALUE)

with pytest.raises(ValueError):
api.recast_rules.search(limit=501)