Added feature - TenableAD Attack API

fixed schema test

updated schema file

docs/api/ad/attacks.rst

@@ -0,0 +1 @@
+.. automodule:: tenable.ad.attacks.api

tenable/ad/__init__.py

@@ -14,6 +14,7 @@
 
     about
     api_keys
+    attacks
     attack_types
     category
     checker

tenable/ad/attacks/api.py

@@ -0,0 +1,75 @@
+'''
+Attacks
+=============
+
+Methods described in this section relate to the attacks API.
+These methods can be accessed at ``TenableAD.attacks``.
+
+.. rst-class:: hide-signature
+.. autoclass:: AttacksAPI
+    :members:
+'''
+from typing import List, Dict
+from tenable.ad.attacks.schema import AttackSchema
+from tenable.base.endpoint import APIEndpoint
+
+
+class AttacksAPI(APIEndpoint):
+    _schema = AttackSchema()
+
+    def list(self,
+             profile_id: str,
+             **kwargs
+             ) -> List[Dict]:
+        '''
+        Retrieve all attacks
+
+        Args:
+            profile_id (str):
+                The attack profile identifier.
+            resource_type (str):
+                The type of resource. possible values are ``infrastructure``,
+                ``directory``, ``hostname``, ``ip``.
+            resource_value (str):
+                The value of resource.
+            attack_type_ids (optional, list[str]):
+                The list of attack type ids.
+            date_end (optional, str):
+                The date before which the attack occurence should be
+                considered.
+            date_start (optional, str):
+                The date after which the attack occurence should be
+                considered.
+            include_closed (optional, str):
+                Whether closed attacks should be included?
+                Accepted values are ``true`` or ``false``
+            limit (optional, str):
+                The number of records user wants to return.
+            order (optional, str):
+                The order of response. Accepted values are
+                ``asc`` or ``desc``.
+            search (optional, str):
+                Search a value in response.
+
+        Returns:
+            list:
+                The list of attacks objects
+
+        Examples:
+            >>> tad.attacks.list(
+            ...     profile_id='1',
+            ...     resource_type='infrastructure',
+            ...     resource_value='1',
+            ...     attack_type_ids=[1, 2],
+            ...     include_closed='false',
+            ...     limit='10',
+            ...     order='asc',
+            ...     search='value',
+            ...     date_end='2022-12-31T18:30:00.000Z',
+            ...     date_start='2021-12-31T18:30:00.000Z'
+            ...     )
+        '''
+        params = self._schema.dump(self._schema.load(kwargs))
+        return self._schema.load(
+            self._api.get(f'profiles/{profile_id}/attacks', params=params),
+            many=True, partial=True)

tenable/ad/attacks/schema.py

@@ -0,0 +1,41 @@
+from marshmallow import fields, validate as v, pre_load
+from tenable.ad.base.schema import CamelCaseSchema, convert_keys_to_camel
+
+
+class AttackAttributesSchema(CamelCaseSchema):
+    name = fields.Str()
+    value = fields.Str()
+    value_type = fields.Str()
+
+
+class AttackVectorSchema(CamelCaseSchema):
+    template = fields.Str()
+    attributes = fields.Nested(AttackAttributesSchema, many=True)
+
+
+class AttackPathSchema(CamelCaseSchema):
+    ip = fields.Str()
+    hostname = fields.Str()
+    type = fields.Str()
+
+
+class AttackSchema(CamelCaseSchema):
+    id = fields.Int()
+    directory_id = fields.Int()
+    attack_type_id = fields.Int()
+    attack_type_ids = fields.List(fields.Int())
+    dc = fields.Str()
+    date = fields.DateTime()
+    vector = fields.Nested(AttackVectorSchema)
+    source = fields.Nested(AttackPathSchema)
+    destination = fields.Nested(AttackPathSchema)
+    is_closed = fields.Bool()
+    resource_type = fields.Str(required=True, validate=v.OneOf(
+        ['infrastructure', 'directory', 'hostname', 'ip']))
+    resource_value = fields.Str(required=True)
+    date_end = fields.DateTime()
+    date_start = fields.DateTime()
+    include_closed = fields.Str(validate=v.OneOf(['true', 'false']))
+    limit = fields.Str()
+    order = fields.Str(validate=v.OneOf(['asc', 'desc']))
+    search = fields.Str()

tenable/ad/session.py

@@ -9,6 +9,7 @@
 from .about import AboutAPI
 from .api_keys import APIKeyAPI
 from .attack_types.api import AttackTypesAPI
+from .attacks.api import AttacksAPI
 from .category.api import CategoryAPI
 from .checker.api import CheckerAPI
 from .checker_option.api import CheckerOptionAPI
@@ -68,6 +69,14 @@ def api_keys(self):
         '''
         return APIKeyAPI(self)
 
+    @property
+    def attacks(self):
+        '''
+        The interface object for the
+        :doc:`Tenable.ad Attacks APIs <attacks>`.
+        '''
+        return AttacksAPI(self)
+
     @property
     def attack_types(self):
         '''

tests/ad/attacks/test_attacks_api.py

@@ -0,0 +1,123 @@
+import pytest
+import responses
+from marshmallow import ValidationError
+
+from tests.ad.conftest import RE_BASE
+
+
+@responses.activate
+def test_attack_list_default(api):
+    responses.add(responses.GET,
+                  f'{RE_BASE}/profiles/1/attacks'
+                  f'?resourceType=infrastructure&resourceValue=1',
+                  json=[{
+                      'attackTypeId': 1,
+                      'date': '2022-01-14T07:24:50.424Z',
+                      'dc': 'dc',
+                      'destination': {
+                          'hostname': 'test',
+                          'ip': '192.168.1.1',
+                          'type': 'computer'
+                      },
+                      'directoryId': 1,
+                      'id': 1,
+                      'isClosed': False,
+                      'source': {
+                          'hostname': 'Unknown',
+                          'ip': '127.0.0.1',
+                          'type': 'computer'
+                      },
+                      'vector': {
+                          'attributes': [{
+                              'name': 'source_hostname',
+                              'value': 'Unknown',
+                              'valueType': 'string'
+                          }],
+                          'template': 'template'
+                      }
+                  }]
+                  )
+    resp = api.attacks.list(
+        profile_id=1,
+        resource_type='infrastructure',
+        resource_value='1'
+    )
+    assert isinstance(resp, list)
+    assert len(resp) == 1
+    assert resp[0]['id'] == 1
+    assert resp[0]['directory_id'] == 1
+    assert resp[0]['attack_type_id'] == 1
+
+
+@responses.activate
+def test_attack_list_parameterized(api):
+    responses.add(responses.GET,
+                  f'{RE_BASE}/profiles/1/attacks'
+                  f'?resourceValue=1'
+                  f'&includeClosed=false'
+                  f'&dateEnd=2022-12-31T18%3A30%3A00%2B00%3A00'
+                  f'&limit=10'
+                  f'&search=Something'
+                  f'&attackTypeIds=1'
+                  f'&attackTypeIds=2'
+                  f'&dateStart=2021-12-31T18%3A30%3A00%2B00%3A00'
+                  f'&resourceType=infrastructure'
+                  f'&order=desc',
+                  json=[{
+                      'attackTypeId': 1,
+                      'date': '2022-01-14T07:24:50.424Z',
+                      'dc': 'dc',
+                      'destination': {
+                          'hostname': 'test',
+                          'ip': '192.168.1.1',
+                          'type': 'computer'
+                      },
+                      'directoryId': 1,
+                      'id': 1,
+                      'isClosed': False,
+                      'source': {
+                          'hostname': 'Unknown',
+                          'ip': '127.0.0.1',
+                          'type': 'computer'
+                      },
+                      'vector': {
+                          'attributes': [{
+                              'name': 'source_hostname',
+                              'value': 'Unknown',
+                              'valueType': 'string'
+                          }],
+                          'template': 'template'
+                      }
+                  }]
+                  )
+    resp = api.attacks.list(
+        profile_id=1,
+        resource_type='infrastructure',
+        resource_value='1',
+        attack_type_ids=[1, 2],
+        include_closed='false',
+        limit='10',
+        order='desc',
+        search='Something',
+        date_end='2022-12-31T18:30:00.000Z',
+        date_start='2021-12-31T18:30:00.000Z'
+    )
+    assert isinstance(resp, list)
+    assert len(resp) == 1
+    assert resp[0]['id'] == 1
+    assert resp[0]['directory_id'] == 1
+    assert resp[0]['attack_type_id'] == 1
+    assert resp[0]['is_closed'] is False
+
+
+def test_attack_list_resource_type_validationerror(api):
+    '''
+    test to raise validation error when resource type does not match
+    the expected list of values
+    '''
+    with pytest.raises(ValidationError):
+        api.attacks.list(
+            profile_id=1,
+            resource_type='something',
+            resource_value='resource type value'
+        )

tests/ad/attacks/test_attacks_schema.py

@@ -0,0 +1,98 @@
+'''
+Testing the attack schema
+'''
+import datetime
+import pytest
+from marshmallow import ValidationError
+from tenable.ad.attacks.schema import AttackSchema
+
+
+@pytest.fixture()
+def attacks_schema_request_payload():
+    return {
+        'resource_type': 'infrastructure',
+        'resource_value': '1',
+        'attack_type_ids': [1, 2],
+        'include_closed': 'false',
+        'limit': '10',
+        'order': 'asc',
+        'search': 'Unknown',
+        'date_end': '2022-12-31T18:30:00.000Z',
+        'date_start': '2002-12-31T18:30:00.000Z'
+    }
+
+
+def test_attacks_schema_request_payload(attacks_schema_request_payload):
+    '''
+    test attacks schema list api request payload
+    '''
+    schema = AttackSchema()
+    req = schema.dump(schema.load(attacks_schema_request_payload))
+    assert req['resourceType'] == 'infrastructure'
+    assert req['resourceValue'] == '1'
+    assert req['attackTypeIds'] == [1, 2]
+    assert req['includeClosed'] == 'false'
+    assert req['limit'] == '10'
+    assert req['order'] == 'asc'
+    assert req['search'] == 'Unknown'
+    assert req['dateEnd'] == '2022-12-31T18:30:00+00:00'
+    assert req['dateStart'] == '2002-12-31T18:30:00+00:00'
+
+    with pytest.raises(ValidationError):
+        attacks_schema_request_payload['some_val'] = 'something'
+        schema.load(attacks_schema_request_payload)
+
+
+def test_attacks_schema_response_object():
+    '''
+    test attacks schema list api response object
+    '''
+    test_resp = [{
+        'attackTypeId': 1,
+        'date': '2022-01-14T07:24:50.424Z',
+        'dc': 'dc',
+        'destination': {
+            'hostname': 'test',
+            'ip': '192.168.1.1',
+            'type': 'computer'
+        },
+        'directoryId': 1,
+        'id': 1,
+        'isClosed': False,
+        'source': {
+            'hostname': 'Unknown',
+            'ip': '127.0.0.1',
+            'type': 'computer'
+        },
+        'vector': {
+            'attributes': [{
+                'name': 'source_hostname',
+                'value': 'Unknown',
+                'valueType': 'string'
+            }],
+            'template': 'template'
+        }
+    }]
+
+    schema = AttackSchema()
+    resp = schema.load(test_resp, many=True, partial=True)
+    assert resp[0]['attack_type_id'] == 1
+    assert resp[0]['source']['hostname'] == 'Unknown'
+    assert resp[0]['source']['type'] == 'computer'
+    assert resp[0]['source']['ip'] == '127.0.0.1'
+    assert resp[0]['date'] == datetime.datetime(
+        2022, 1, 14, 7, 24, 50, 424000, tzinfo=datetime.timezone.utc)
+    assert resp[0]['dc'] == 'dc'
+    assert resp[0]['destination']['hostname'] == 'test'
+    assert resp[0]['destination']['type'] == 'computer'
+    assert resp[0]['destination']['ip'] == '192.168.1.1'
+    assert resp[0]['vector']['template'] == 'template'
+    assert resp[0]['vector']['attributes'][0]['value'] == 'Unknown'
+    assert resp[0]['vector']['attributes'][0]['name'] == 'source_hostname'
+    assert resp[0]['vector']['attributes'][0]['value_type'] == 'string'
+    assert resp[0]['directory_id'] == 1
+    assert resp[0]['is_closed'] == False
+
+    with pytest.raises(ValidationError):
+        test_resp[0]['some_val'] = 'something'
+        schema.load(test_resp, many=True)