remove validation for event query filters, update file event v2 model

Author: tora-kozic

pyproject.toml

@@ -101,7 +101,7 @@ dependencies = [
 debug = "pytest ./tests/test_file_events.py -s -k saved_search"
 cov = "pytest --cov-report=term-missing --cov-config=pyproject.toml --cov=incydr"
 cov-report = "pytest --cov-report=xml:coverage.xml --cov-config=pyproject.toml --cov=incydr"
-no-cov = "cov --no-cov"
+no-cov = "pytest --disable-warnings"
 
 [tool.coverage.run]
 branch = true

src/_incydr_sdk/file_events/client.py

@@ -1,14 +1,29 @@
 from typing import List
 
 from pydantic import parse_obj_as
+from requests import HTTPError
 from requests.adapters import HTTPAdapter
 from urllib3 import Retry
 
+from ..exceptions import IncydrException
 from .models.response import FileEventsPage
 from .models.response import SavedSearch
 from _incydr_sdk.queries.file_events import EventQuery
 
 
+class InvalidQueryException(IncydrException):
+    """Raised when the file events search endpoint returns a 400."""
+
+    def __init__(self, query=None):
+        self.query = query
+        self.message = (
+            "400 Response Error: Invalid query. Please double check your query filters are valid. "
+            "\nTip: Make sure you're specifying your filter fields in dot notation. "
+            "\nFor example, filter by 'file.archiveId' to filter by the archiveId field within the file object.)"
+        )
+        super().__init__(self.message)
+
+
 class FileEventsV2:
     """
     Client for `/v2/file-events` endpoints.
@@ -46,7 +61,12 @@ def search(self, query: EventQuery) -> FileEventsPage:
         """
         self._mount_retry_adapter()
 
-        response = self._parent.session.post("/v2/file-events", json=query.dict())
+        try:
+            response = self._parent.session.post("/v2/file-events", json=query.dict())
+        except HTTPError as err:
+            if err.response.status_code == 400:
+                raise InvalidQueryException(query)
+            raise err
         page = FileEventsPage.parse_response(response)
         query.page_token = page.next_pg_token
         return page

src/_incydr_sdk/file_events/models/event.py

@@ -9,6 +9,28 @@
 from _incydr_sdk.enums.file_events import ReportType
 
 
+class UserJustification(Model):
+    reason: Optional[str] = Field(
+        None, title="User-select justification for temporarily allowing this action."
+    )
+    text: Optional[str] = Field(
+        None,
+        title="User-select justification for temporarily allowing this action. Only applies when reason is 'Other'.",
+    )
+
+
+class ResponseControls(Model):
+    preventative_control: Optional[str] = Field(
+        None,
+        alias="preventativeControl",
+        example="BLOCKED",
+        title="The preventative action applied to this event",
+    )
+    user_justification: Optional[UserJustification] = Field(
+        None, alias="userJustification"
+    )
+
+
 class AcquiredFromGit(Model):
     repository_email: Optional[str] = Field(
         None,
@@ -108,15 +130,33 @@ class Hash(Model):
     )
 
 
+class Extension(Model):
+    browser: Optional[str] = Field(
+        None, title="The web browser in which the event occurred."
+    )
+    version: Optional[str] = Field(
+        None,
+        title="The version of the Code42 Incydr extension installed when the event occurred.",
+    )
+    logged_in_user: Optional[str] = Field(
+        None,
+        alias="loggedInUser",
+        title="The user signed in to the active tab where the event occurred. For example, the user signed in to Gmail. This may differ from the user account signed in to the browser itself.",
+    )
+
+
 class Process(Model):
     executable: Optional[str] = Field(
+        None,
         description="The name of the process that accessed the file, as reported by the device’s operating system. Depending on your Code42 product plan, this value may be null for some event types.",
         example="bash",
     )
     owner: Optional[str] = Field(
+        None,
         description="The username of the process owner, as reported by the device’s operating system. Depending on your Code42 product plan, this value may be null for some event types.",
         example="root",
     )
+    extension: Optional[Extension]
 
 
 class RemovableMedia(Model):
@@ -454,6 +494,11 @@ class File(Model):
         description="Unique identifier reported by the cloud provider for the file associated with the event.",
         example="PUL5zWLRrdudiJZ1OCWw",
     )
+    mime_type: Optional[str] = Field(
+        None,
+        alias="mimeType",
+        title="The MIME type of the file. For endpoint events, if the mimeTypeByBytes differs from mimeTypeByExtension, this indicates the most likely MIME type for the file. For activity observed by a web browser, this is the only MIME type reported.",
+    )
     mime_type_by_bytes: Optional[str] = Field(
         alias="mimeTypeByBytes",
         description="The MIME type of the file based on its contents.",
@@ -489,6 +534,21 @@ class File(Model):
         description="URL reported by the cloud provider at the time the event occurred.",
         example="https://example.com",
     )
+    archive_id: Optional[str] = Field(
+        None,
+        alias="archiveId",
+        title="Unique identifier for files identified as an archive, such as .zip files.",
+    )
+    parent_archive_id: Optional[str] = Field(
+        None,
+        alias="parentArchiveId",
+        title="For files contained within an archive (such as a .zip file), the unique identifier for that archive; searching on parentArchiveID returns events for all files contained within that archive",
+    )
+    password_protected: Optional[bool] = Field(
+        None,
+        alias="passwordProtected",
+        title="Indicates if this file is password protected.",
+    )
 
 
 class Risk(Model):
@@ -556,6 +616,11 @@ class Source(Model):
         description="The IP address of the user's device on your internal network, including Network interfaces, Virtual Network Interface controllers (NICs), and Loopback/non-routable addresses.",
         example=["127.0.0.1", "127.0.0.2"],
     )
+    remote_hostname: Optional[str] = Field(
+        None,
+        alias="remoteHostname",
+        title="For events where a file transfer tool was used, the source hostname.",
+    )
     removable_media: Optional[RemovableMedia] = Field(
         alias="removableMedia",
         description="Metadata about the removable media source.",
@@ -638,18 +703,44 @@ class Event(Model):
         description="Sharing types added by this event.",
         example=["SharedViaLink"],
     )
-    vector: Optional[str]
+    vector: Optional[str] = Field(
+        None,
+        example="GIT_PUSH",
+        title="The method of file movement. For example: UPLOADED, DOWNLOADED, EMAILED.",
+    )
 
 
 class Git(Model):
-    event_id: Optional[str] = Field(None, alias="eventId")
-    last_commit_hash: Optional[str] = Field(None, alias="lastCommitHash")
-    repository_email: Optional[str] = Field(None, alias="repositoryEmail")
+    event_id: Optional[str] = Field(
+        None,
+        alias="eventId",
+        title="A global unique identifier (GUID) generated by Incydr for this Git event. All files associated with this event have the same Git event ID. A single Git event can be associated with multiple file events.",
+    )
+    last_commit_hash: Optional[str] = Field(
+        None,
+        alias="lastCommitHash",
+        title="Hash value from the most recent commit in this Git event.",
+    )
+    repository_email: Optional[str] = Field(
+        None,
+        alias="repositoryEmail",
+        title="The email address specified by the user who performed the Git event. This is a user-defined value and may differ from the credentials used to sign in to Git.",
+    )
     repository_endpoint_path: Optional[str] = Field(
-        None, alias="repositoryEndpointPath"
+        None,
+        alias="repositoryEndpointPath",
+        title="File path of the local Git repository on the user's endpoint.",
+    )
+    repository_uri: Optional[str] = Field(
+        None,
+        alias="repositoryUri",
+        title="Uniform Resource Identifier (URI) for the Git repository.",
+    )
+    repository_user: Optional[str] = Field(
+        None,
+        alias="repositoryUser",
+        title="The username specified by the user who performed the Git event. This is a user-defined value and may differ from the credentials used to sign in to Git.",
     )
-    repository_uri: Optional[str] = Field(None, alias="repositoryUri")
-    repository_user: Optional[str] = Field(None, alias="repositoryUser")
 
 
 class FileEventV2(ResponseModel):
@@ -688,6 +779,10 @@ class FileEventV2(ResponseModel):
     report: Optional[Report] = Field(
         description="Metadata for reports from 3rd party sources, such Salesforce downloads.",
     )
+    response_controls: Optional[ResponseControls] = Field(
+        alias="responseControls",
+        description="Metadata about preventative actions applied to file activity. Only applies to events for users on a preventative watchlist.",
+    )
     risk: Optional[Risk] = Field(
         description="Risk factor metadata.",
     )

src/_incydr_sdk/queries/file_events.py

@@ -40,7 +40,7 @@
 
 
 class Filter(BaseModel):
-    term: EventSearchTerm
+    term: str
     operator: Operator
     value: Optional[Union[int, str]]
 
@@ -53,8 +53,9 @@ def _validate_enums(cls, values: dict):  # noqa `root_validator` is a classmetho
         operator = values.get("operator")
         value = values.get("value")
 
+        # 11-13-2024 - Removing strict filter term requirements to avoid breaking on new fields
         # make sure `term` is valid enum value
-        EventSearchTerm(term)
+        # EventSearchTerm(term)
 
         if operator in (Operator.EXISTS, Operator.DOES_NOT_EXIST):
             values["value"] = None
@@ -66,15 +67,16 @@ def _validate_enums(cls, values: dict):  # noqa `root_validator` is a classmetho
                     f"`IS` and `IS_NOT` filters require a `str | int` value, got term={term}, operator={operator}, value={value}."
                 )
 
+        # 11-13-2024 - Removing strict filter term requirements to avoid breaking on new fields
         # check that value is a valid enum for that search term
-        enum = _term_enum_map.get(term)
-        if enum:
-            try:
-                values.update(
-                    {"value": enum[value.upper()]}
-                )  # check if enum name is passed as a value
-            except KeyError:
-                enum(value)
+        # enum = _term_enum_map.get(term)
+        # if enum:
+        #     try:
+        #         values.update(
+        #             {"value": enum[value.upper()]}
+        #         )  # check if enum name is passed as a value
+        #     except KeyError:
+        #         enum(value)
 
         return values
 

tests/queries/test_event_query.py

@@ -132,6 +132,10 @@ def test_event_query_is_when_no_values_raises_error():
     assert e.value.args[0] == "equals() requires at least one value."
 
 
+# TODO
+@pytest.mark.skip(
+    reason="11-13-2024 - Removing strict filter term requirements to avoid breaking on new fields"
+)
 def test_event_query_is_when_invalid_value_for_term_raises_type_error():
     with pytest.raises(ValueError) as e:
 
@@ -144,6 +148,31 @@ def test_event_query_is_when_invalid_value_for_term_raises_type_error():
     )
 
 
+def test_event_query_allows_any_search_term_and_creates_expected_filter_group():
+    q = EventQuery(TEST_START_DATE).equals("file-event-field", "value")
+    expected = FilterGroup(
+        filterClause="AND",
+        filters=[
+            Filter(term="file-event-field", operator="IS", value="value"),
+        ],
+    )
+    assert q.groups.pop() == expected
+
+
+def test_event_query_allows_any_string_values_and_creates_expected_filter_group():
+    q = EventQuery(TEST_START_DATE).equals(
+        "file.category", ["Document", "string-value"]
+    )
+    expected = FilterGroup(
+        filterClause="OR",
+        filters=[
+            Filter(term="file.category", operator="IS", value="Document"),
+            Filter(term="file.category", operator="IS", value="string-value"),
+        ],
+    )
+    assert q.groups.pop() == expected
+
+
 def test_event_query_exists_creates_expected_filter_group():
     q = EventQuery(start_date=TEST_START_DATE).exists("event.action")
     expected = FilterGroup(