add methods for xfc id downloading

Author: ceciliastevens

CHANGELOG.md

@@ -17,6 +17,9 @@
   - `is_none`
   - `date_range`
   - `subquery`
+- New methods to download files by XFC content ID.
+  - `sdk.files.download_file_by_xfc_content_id` and `sdk.files.stream_file_by_xfc_content_id`
+  - `incydr files download-by-xfc-id`
 
 ### Fixed
 - An issue where in some cases saved searches could not be retrieved.

src/_incydr_cli/cmds/files.py

@@ -31,3 +31,17 @@ def download(sha256: str, path: str):
     """
     client = Client()
     client.files.v1.download_file_by_sha256(sha256, path)
+
+
+@files.command(cls=IncydrCommand)
+@click.argument("XFC_ID")
+@path_option
+@logging_options
+def download_by_xfc_id(xfc_id: str, path: str):
+    """
+    Download the file matching the given XFC content ID hash to the target path.
+    """
+    client = Client()
+    client.files.v1.download_file_by_xfc_content_id(
+        xfc_content_id=xfc_id, target_path=path
+    )

src/_incydr_sdk/file_events/client.py

@@ -21,7 +21,7 @@ def __init__(self, query=None, exception=None):
             "\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.)"
         )
-        if 'problems' in exception.response.json().keys():
+        if "problems" in exception.response.json().keys():
             self.message += f"\nRaw problem data from the response: {exception.response.json()['problems']}"
         self.original_exception = exception
         super().__init__(self.message)

src/_incydr_sdk/file_events/models/event.py

@@ -709,6 +709,11 @@ class Event(Model):
         example="GIT_PUSH",
         title="The method of file movement. For example: UPLOADED, DOWNLOADED, EMAILED.",
     )
+    xfc_event_id: Optional[str] = Field(
+        None,
+        alias="xfcEventId",
+        description="The identifier for the exfiltrated file collection data associated with this event.",
+    )
 
 
 class Git(Model):

src/_incydr_sdk/files/client.py

@@ -65,3 +65,46 @@ def stream_file_by_sha256(self, sha256: str):
         return self._parent.session.get(
             f"/v1/files/get-file-by-sha256/{sha256}", stream=True
         )
+
+    def download_file_by_xfc_content_id(
+        self, xfc_content_id: str, target_path: Path
+    ) -> Path:
+        """Download a file that matches the given XFC content ID.
+
+        **Parameters:**
+
+        * **xfc_content_id**: `str` (required) The XFC content ID for file you wish to download.
+        * **target_path**: `Path | str` a string or `pathlib.Path` object that represents the target file path and
+            name to which the file will be saved to.
+
+        **Returns**: A `pathlib.Path` object representing the location of the downloaded file.
+        """
+        target = Path(
+            target_path
+        )  # ensure that target is a path even if we're given a string
+        response = self._parent.session.get(
+            f"/v1/files/get-file-by-xfc-content-id/{xfc_content_id}"
+        )
+        target.write_bytes(response.content)
+        return target
+
+    def stream_file_by_xfc_content_id(self, xfc_content_id: str):
+        """Stream a file that matches the given XFC content ID.
+
+        **Example usage:**
+        ```
+        >>> with sdk.files.v1.stream_file_by_xfc_content_id("content_id_example") as response:
+        >>>     with open("./testfile.zip", "wb") as file:
+        >>>         for chunk in response.iter_content(chunk_size=128):
+        >>>             file.write(chunk)
+        ```
+
+        **Parameters:**
+
+        * **xfc_content_id**: `str` (required) The XFC content ID for file you wish to download.
+
+        **Returns**: A `requests.Response` object with a stream of the requested file.
+        """
+        return self._parent.session.get(
+            f"/v1/files/get-file-by-xfc-content-id/{xfc_content_id}", stream=True
+        )

tests/test_files.py

@@ -7,6 +7,7 @@
 
 
 TEST_SHA256 = "38acb15d02d5ac0f2a2789602e9df950c380d2799b4bdb59394e4eeabdd3a662"
+TEST_XFC_ID = "test"
 BAD_SHA256 = "asdf"
 TEST_DATA = b"test data"
 
@@ -19,6 +20,14 @@ def mock_file_download(httpserver_auth: HTTPServer):
     return httpserver_auth
 
 
+@pytest.fixture
+def mock_file_download_by_xfc(httpserver_auth: HTTPServer):
+    httpserver_auth.expect_request(
+        f"/v1/files/get-file-by-xfc-content-id/{TEST_XFC_ID}"
+    ).respond_with_data(response_data=TEST_DATA, status=200)
+    return httpserver_auth
+
+
 @pytest.fixture
 def mock_bad_sha256(httpserver_auth: HTTPServer):
     httpserver_auth.expect_request(
@@ -35,6 +44,14 @@ def mock_file_not_found(httpserver_auth: HTTPServer):
     return httpserver_auth
 
 
+@pytest.fixture
+def mock_file_not_found_by_xfc_id(httpserver_auth: HTTPServer):
+    httpserver_auth.expect_request(
+        f"/v1/files/get-file-by-xfc-content-id/{TEST_XFC_ID}"
+    ).respond_with_data(response_data="", status=404)
+    return httpserver_auth
+
+
 def test_download_file_by_sha256_calls_with_correct_parameter(
     mock_file_download, tmp_path
 ):
@@ -70,6 +87,35 @@ def test_stream_file_by_sha256_streams_file(mock_file_download):
     assert content == TEST_DATA
 
 
+def test_download_file_by_xfc_id_calls_with_correct_parameter(
+    mock_file_download_by_xfc, tmp_path
+):
+    c = Client()
+    p = tmp_path / "testfile.test"
+    f = c.files.v1.download_file_by_xfc_content_id(TEST_XFC_ID, p)
+    with open(f, "rb") as file:
+        content = file.read()
+    assert content == TEST_DATA
+    mock_file_download_by_xfc.check()
+
+
+def test_download_file_by_xfc_id_raises_error_when_file_not_found(
+    mock_file_not_found_by_xfc_id,
+):
+    c = Client()
+    with pytest.raises(HTTPError) as error:
+        c.files.v1.download_file_by_xfc_content_id(TEST_XFC_ID, "testpath.text")
+    assert error.value.response.status_code == 404
+    mock_file_not_found_by_xfc_id.check()
+
+
+def test_stream_file_by_xfc_id_streams_file(mock_file_download_by_xfc):
+    c = Client()
+    response = c.files.v1.stream_file_by_xfc_content_id(TEST_XFC_ID)
+    content = b"".join([c for c in response.iter_content(chunk_size=128) if c])
+    assert content == TEST_DATA
+
+
 # ************************************************ CLI ************************************************