Merge pull request #22 from trendmicro/feature/api_oat

Added support for OAT events API

src/pytmv1/__init__.py

@@ -25,6 +25,10 @@
     MsDataApiKey,
     MsDataUrl,
     MsError,
+    OatEndpoint,
+    OatEvent,
+    OatFilter,
+    OatObject,
     SaeAlert,
     SaeIndicator,
     SandboxSuspiciousObject,
@@ -43,6 +47,10 @@
     Iam,
     IntegrityLevel,
     InvestigationStatus,
+    OatDataSource,
+    OatEntityType,
+    OatRiskLevel,
+    OatType,
     ObjectType,
     OperatingSystem,
     ProductCode,
@@ -95,6 +103,7 @@
     ListEndpointActivityResp,
     ListEndpointDataResp,
     ListExceptionsResp,
+    ListOatsResp,
     ListSandboxSuspiciousResp,
     ListSuspiciousResp,
     MultiApiKeyResp,
@@ -172,6 +181,7 @@
     "ListEndpointActivityResp",
     "ListEndpointDataResp",
     "ListExceptionsResp",
+    "ListOatsResp",
     "ListSandboxSuspiciousResp",
     "ListSuspiciousResp",
     "MatchedEvent",
@@ -187,6 +197,14 @@
     "MultiResp",
     "MultiUrlResp",
     "NoContentResp",
+    "OatDataSource",
+    "OatEndpoint",
+    "OatEntityType",
+    "OatEvent",
+    "OatFilter",
+    "OatObject",
+    "OatRiskLevel",
+    "OatType",
     "ObjectRequest",
     "ObjectType",
     "OperatingSystem",

src/pytmv1/api/__init__.py

@@ -4,6 +4,7 @@
 from .email import Email
 from .endpoint import Endpoint
 from .note import Note
+from .oat import Oat
 from .object import Object
 from .sandbox import Sandbox
 from .script import CustomScript
@@ -18,6 +19,7 @@
     "Email",
     "Endpoint",
     "Note",
+    "Oat",
     "Object",
     "Sandbox",
     "System",

src/pytmv1/api/oat.py

@@ -0,0 +1,119 @@
+from typing import Callable, Optional
+
+from .. import utils
+from ..core import Core
+from ..model.common import OatEvent
+from ..model.enum import Api, QueryOp
+from ..model.response import ListOatsResp
+from ..result import Result
+
+
+class Oat:
+    _core: Core
+
+    def __init__(self, core: Core):
+        self._core = core
+
+    def list(
+        self,
+        detected_start_date_time: Optional[str] = None,
+        detected_end_date_time: Optional[str] = None,
+        ingested_start_date_time: Optional[str] = None,
+        ingested_end_date_time: Optional[str] = None,
+        top: int = 50,
+        op: QueryOp = QueryOp.AND,
+        **fields: str,
+    ) -> Result[ListOatsResp]:
+        """Retrieves Observed Attack Techniques events in a paginated list.
+
+        :param detected_start_date_time: Date that indicates the start of
+        the event detection data retrieval time range (yyyy-MM-ddThh:mm:ssZ).
+        Defaults to 24 hours before the request is made.
+        :type detected_start_date_time: Optional[str]
+        :param detected_end_date_time: Date that indicates the end of
+        the event detection data retrieval time range (yyyy-MM-ddThh:mm:ssZ).
+        Defaults to the time the request is made.
+        :type detected_end_date_time: Optional[str]
+        :param ingested_start_date_time: Date that indicates the start of
+        the data ingestion time range (yyyy-MM-ddThh:mm:ssZ).
+        :type ingested_start_date_time: Optional[str]
+        :param ingested_end_date_time: Date that indicates the end of
+        the data ingestion time range (yyyy-MM-ddThh:mm:ssZ).
+        :type ingested_end_date_time: Optional[str]
+        :param top: Number of records displayed on a page.
+        :type top: int
+        :param op: Operator to apply between fields (ie: ... OR ...).
+        :type op: QueryOp
+        :param fields: Field/value used to filter result (i.e:uuid="123"),
+        check Vision One API documentation for full list of supported fields.
+        :type fields: Dict[str, str]
+        :rtype: Result[ListOatsResp]
+        """
+        return self._core.send(
+            ListOatsResp,
+            Api.GET_OAT_LIST,
+            params=utils.filter_none(
+                {
+                    "detectedStartDateTime": detected_start_date_time,
+                    "detectedEndDateTime": detected_end_date_time,
+                    "ingestedStartDateTime": ingested_start_date_time,
+                    "ingestedEndDateTime": ingested_end_date_time,
+                    "top": top,
+                }
+            ),
+            headers=utils.tmv1_filter(op, fields),
+        )
+
+    def consume(
+        self,
+        consumer: Callable[[OatEvent], None],
+        detected_start_date_time: Optional[str] = None,
+        detected_end_date_time: Optional[str] = None,
+        ingested_start_date_time: Optional[str] = None,
+        ingested_end_date_time: Optional[str] = None,
+        top: int = 50,
+        op: QueryOp = QueryOp.AND,
+        **fields: str,
+    ) -> Result[ListOatsResp]:
+        """Retrieves and consume OAT events.
+
+        :param consumer: Function which will consume every record in result.
+        :type consumer: Callable[[Oat], None]
+        :param detected_start_date_time: Date that indicates the start of
+        the event detection data retrieval time range (yyyy-MM-ddThh:mm:ssZ).
+        Defaults to 24 hours before the request is made.
+        :type detected_start_date_time: Optional[str]
+        :param detected_end_date_time: Date that indicates the end of
+        the event detection data retrieval time range (yyyy-MM-ddThh:mm:ssZ).
+        Defaults to the time the request is made.
+        :type detected_end_date_time: Optional[str]
+        :param ingested_start_date_time: Date that indicates the start of
+        the data ingestion time range (yyyy-MM-ddThh:mm:ssZ).
+        :type ingested_start_date_time: Optional[str]
+        :param ingested_end_date_time: Date that indicates the end of
+        the data ingestion time range (yyyy-MM-ddThh:mm:ssZ).
+        :type ingested_end_date_time: Optional[str]
+        :param top: Number of records displayed on a page.
+        :type top: int
+        :param op: Operator to apply between fields (ie: ... OR ...).
+        :type op: QueryOp
+        :param fields: Field/value used to filter result (i.e:uuid="123"),
+        check Vision One API documentation for full list of supported fields.
+        :type fields: Dict[str, str]
+        :rtype: Result[ListOatsResp]
+        """
+        return self._core.send_linkable(
+            ListOatsResp,
+            Api.GET_OAT_LIST,
+            consumer,
+            params=utils.filter_none(
+                {
+                    "detectedStartDateTime": detected_start_date_time,
+                    "detectedEndDateTime": detected_end_date_time,
+                    "ingestedStartDateTime": ingested_start_date_time,
+                    "ingestedEndDateTime": ingested_end_date_time,
+                    "top": top,
+                }
+            ),
+            headers=utils.tmv1_filter(op, fields),
+        )

src/pytmv1/client.py

@@ -73,6 +73,7 @@ def __init__(self, core: Core):
         self.email = api.Email(self._core)
         self.endpoint = api.Endpoint(self._core)
         self.note = api.Note(self._core)
+        self.oat = api.Oat(self._core)
         self.object = api.Object(self._core)
         self.sandbox = api.Sandbox(self._core)
         self.script = api.CustomScript(self._core)

src/pytmv1/model/common.py

@@ -16,6 +16,9 @@
     Iam,
     IntegrityLevel,
     InvestigationStatus,
+    OatDataSource,
+    OatEntityType,
+    OatRiskLevel,
     ObjectType,
     OperatingSystem,
     ProductCode,
@@ -325,6 +328,41 @@ def values(self) -> List[int]:
         return self.root
 
 
+class OatEndpoint(BaseModel):
+    endpoint_name: str
+    agent_guid: str
+    ips: List[str]
+
+
+class OatObject(BaseModel):
+    type: str
+    field: str
+    value: Union[int, str, List[str]]
+
+
+class OatFilter(BaseModel):
+    id: str
+    name: str
+    description: Optional[str] = None
+    mitre_tactic_ids: List[str]
+    mitre_technique_ids: List[str]
+    highlighted_objects: List[OatObject]
+    risk_level: OatRiskLevel
+    type: str
+
+
+class OatEvent(BaseConsumable):
+    source: OatDataSource
+    uuid: str
+    filters: List[OatFilter]
+    endpoint: Optional[OatEndpoint] = None
+    entity_type: OatEntityType
+    entity_name: str
+    detected_date_time: str
+    ingested_date_time: Optional[str] = None
+    detail: Union[EndpointActivity, EmailActivity]
+
+
 class SaeAlert(Alert):
     description: str
     matched_rules: List[MatchedRule]

src/pytmv1/model/enum.py

@@ -58,6 +58,7 @@ class Api(str, Enum):
     GET_ALERT_NOTE = "/workbench/alerts/{0}/notes/{1}"
     UPDATE_ALERT_NOTE = "/workbench/alerts/{0}/notes/{1}"
     DELETE_ALERT_NOTE = "/workbench/alerts/{0}/notes/delete"
+    GET_OAT_LIST = "/oat/detections"
 
 
 class ApiExpInMonths(int, Enum):
@@ -184,6 +185,40 @@ class HttpMethod(str, Enum):
     DELETE = "DELETE"
 
 
+class OatDataSource(str, Enum):
+    DETECTIONS = "detections"
+    ENDPOINT_ACTIVITY_DATA = "endpointActivityData"
+    CLOUD_ACTIVITY_DATA = "cloudActivityData"
+    EMAIL_ACTIVITY_DATA = "emailActivityData"
+    MOBILE_ACTIVITY_DATA = "mobileActivityData"
+    NETWORK_ACTIVITY_DATA = "networkActivityData"
+    CONTAINER_ACTIVITY_DATA = "containerActivityData"
+
+
+class OatEntityType(str, Enum):
+    ENDPOINT = "endpoint"
+    MAILBOX = "mailbox"
+    CLOUDTRAIL = "cloudtrail"
+    MESSAGING = "messaging"
+    NETWORK = "network"
+    ICS = "ics"
+    CONTAINER = "container"
+
+
+class OatRiskLevel(str, Enum):
+    UNDEFINED = "undefined"
+    INFO = "info"
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class OatType(str, Enum):
+    CUSTOM = "custom"
+    PRESET = "preset"
+
+
 class ObjectType(str, Enum):
     IP = "ip"
     URL = "url"

src/pytmv1/model/response.py

@@ -20,6 +20,7 @@
     MsData,
     MsDataApiKey,
     MsDataUrl,
+    OatEvent,
     SaeAlert,
     SandboxSuspiciousObject,
     Script,
@@ -207,6 +208,11 @@ class ListExceptionsResp(BaseLinkableResp[ExceptionObject]):
     ...
 
 
+class ListOatsResp(BaseLinkableResp[OatEvent]):
+    total_count: int
+    count: int
+
+
 class ListSuspiciousResp(BaseLinkableResp[SuspiciousObject]):
     ...