diff --git a/docs/api/io/recast_rules.rst b/docs/api/io/recast_rules.rst new file mode 100644 index 000000000..b7bbef8be --- /dev/null +++ b/docs/api/io/recast_rules.rst @@ -0,0 +1 @@ +.. automodule:: tenable.io.recast_rules diff --git a/tenable/io/__init__.py b/tenable/io/__init__.py index 722a75b1e..627bb7a16 100644 --- a/tenable/io/__init__.py +++ b/tenable/io/__init__.py @@ -31,6 +31,7 @@ permissions plugins policies + recast_rules remediation_scans scanner_groups scanners @@ -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 @@ -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 `. + """ + return RecastRulesAPI(self) + @property def scanner_groups(self): """ diff --git a/tenable/io/recast_rules.py b/tenable/io/recast_rules.py new file mode 100644 index 000000000..09cea5977 --- /dev/null +++ b/tenable/io/recast_rules.py @@ -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() diff --git a/tests/io/test_recast_rules.py b/tests/io/test_recast_rules.py new file mode 100644 index 000000000..af29b7218 --- /dev/null +++ b/tests/io/test_recast_rules.py @@ -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)