UploadFile
Copies a file from this incident to the specified incident. The file is recorded as an entry in the specified incident’s War Room.
python · Cortex REST API
Details
| ID | UploadFile |
|---|---|
| Language | python |
| From Version | 6.5.0 |
| Docker Image | demisto/python3:3.12.13.10404775 |
| Tags | DemistoAPI |
README
Copies a file from this incident to the specified incident. The file is recorded as an entry in the specified incident’s War Room.
Script Data
| Name | Description |
|---|---|
| Script Type | python3 |
| Tags | DemistoAPI |
| Cortex XSOAR Version | 5.0.0 |
Dependencies
This script uses the following commands and scripts.
- core-api-multipart
Inputs
| Argument Name | Description |
|---|---|
| entryID | File entry ID. |
| incidentID | Incident ID to upload the file to. |
| body | Request body. |
| target | Where to upload the file - Available options are: - `war room entry`: the file will be uploaded as War Room entry. - `incident attachment`: the file will be uploaded as incident attachment. - default are `war room entry` |
| using | Integration instance to use to run the command. |
Outputs
There are no outputs for this script.
Troubleshooting
Multi-tenant environments should be configured with the Cortex Rest API instance when using this
automation. Make sure the Use tenant parameter (in the Cortex Rest API integration) is checked
to ensure that API calls are made to the current tenant instead of the master tenant.
import pytest from CommonServerPython import EntryType from UploadFile import is_transient_error, upload_file, upload_file_command RAW_RESPONSE = [ { "Brand": "Core REST API", "Category": "Utilities", "Contents": { "response": { "ShardID": 0, "category": "", "childs": None, "closed": "0001-01-01T00:00:00Z", "created": "2020-07-08T11:30:14.386972+03:00", "creatingUserId": "admin", "dbotCreatedBy": "admin", "details": "", "entries": [ { "IndicatorTimeline": None, "ShardID": 0, "brand": "none", "category": "artifact", "contents": "", "contentsSize": 0, "created": "2020-07-13T10:57:32.956671+03:00", "cronView": False, "dbotCreatedBy": "admin", "endingDate": "0001-01-01T00:00:00Z", "entryTask": None, "errorSource": "", "file": "test_name.pdf", "fileID": "94b6eed9-fd5d-412e-87dc-5e2a7cdc8457", "fileMetadata": { "info": "PDF document, version 1.3", "isMediaFile": False, "md5": "be357a5a72c1236a42590ea894866884", "sha1": "29f4d33b0895aa46a1dad97b54096d4e68e3d91a", "sha256": "6577fb6b9a5f4779f835fec594d21267aedf3c7f3ee183c4bc2fd3f9fca52df4", "sha512": "174c4574cbe429bb646e19ed1973b2ccfc314bc951f251684591105d1d0f5fe8aa9f67beae0d742" "0becec61bcfd51e17d270683242b77dc552e58b2f95a0625b", "size": 250301, "ssdeep": "6144:IwUbC/ok4IeyUvtRCRvhp2h8C+GYPcX6PorXMido+vWap81:z7ok4XjtIaxrKArXBoqw", "type": "application/pdf", }, "format": "", "hasRole": False, "id": "121@899", "investigationId": "899", "isTodo": False, "mirrored": False, "modified": "2020-07-13T10:57:33.011106+03:00", "note": False, "parentContent": None, "parentEntryTruncated": False, "parentId": "", "pinned": False, "playbookId": "", "previousRoles": None, "recurrent": False, "reputationSize": 0, "reputations": None, "retryTime": "0001-01-01T00:00:00Z", "roles": None, "scheduled": False, "sortValues": None, "startDate": "0001-01-01T00:00:00Z", "system": "", "tags": [], "tagsRaw": [], "taskId": "", "times": 0, "timezoneOffset": 0, "type": 3, "user": "admin", "version": 2, } ], "entryUsers": ["admin"], "hasRole": False, "highPriority": False, "id": "899", "invContext": None, "lastOpen": "0001-01-01T00:00:00Z", "mirrorAutoClose": None, "mirrorTypes": None, "modified": "2020-07-08T11:31:30.594058+03:00", "name": "office365_Feed", "previousRoles": None, "rawCategory": "", "reason": None, "roles": None, "runStatus": "completed", "slackMirrorAutoClose": False, "slackMirrorType": "", "sortValues": None, "status": 0, "systems": None, "tags": ["Automation"], "totalEntries": 0, "type": 0, "users": ["admin", "analyst"], "version": 8, } }, "ContentsFormat": "json", "EntryContext": None, "Evidence": False, "EvidenceID": "", "File": "", "FileID": "", "FileMetadata": None, "HumanReadable": None, "ID": "", "IgnoreAutoExtract": False, "ImportantEntryContext": None, "IndicatorTimeline": None, "Metadata": { "IndicatorTimeline": None, "ShardID": 0, "brand": "Core REST API", "category": "", "contents": "", "contentsSize": 0, "created": "2020-07-13T10:57:33.044787+03:00", "cronView": False, "dbotCreatedBy": "", "endingDate": "0001-01-01T00:00:00Z", "entryTask": None, "errorSource": "", "file": "", "fileID": "", "fileMetadata": None, "format": "json", "hasRole": False, "id": "", "instance": "Core REST API_instance_1", "investigationId": "737", "isTodo": False, "mirrored": False, "modified": "0001-01-01T00:00:00Z", "note": False, "parentContent": '!core-api-multipart uri="entry/upload/899" entryID="722@737" body="test_bark"', "parentEntryTruncated": False, "parentId": "726@737", "pinned": False, "playbookId": "", "previousRoles": None, "recurrent": False, "reputationSize": 0, "reputations": None, "retryTime": "0001-01-01T00:00:00Z", "roles": None, "scheduled": False, "sortValues": None, "startDate": "0001-01-01T00:00:00Z", "system": "", "tags": None, "tagsRaw": None, "taskId": "", "times": 0, "timezoneOffset": 0, "type": 1, "user": "", "version": 0, }, "ModuleName": "Core REST API_instance_1", "Note": False, "ReadableContentsFormat": "", "System": "", "Tags": None, "Type": 1, "Version": 0, } ] def test_upload_file(mocker): """Unit test Given - upload_file_command command - command args When - mock the raw response. Then - run the upload_file Validate the content of the HumanReadable. """ mocker.patch("UploadFile.upload_file", return_value=RAW_RESPONSE) command_results = upload_file_command({"incidentId": "1", "entryID": "12@12", "body": "test_bark"}) assert "test_bark" in command_results[0].readable_output def test_upload_file_multiple_entry_ids(mocker): """Unit test Given - Command args with multiple entry IDs. When - Running the upload_file_command function. Then - Validate that the API request was called for each entry ID. """ execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand") upload_file_command({"incidentId": "1", "entryID": "1,2"}) assert execute_command_mocker.call_args_list[0][0][1]["entryID"] == "1" assert execute_command_mocker.call_args_list[1][0][1]["entryID"] == "2" def test_upload_file_one_entry_id(mocker): """Unit test Given - Command args with one entry ID. When - Running the upload_file_command function. Then - Validate that the API request was called only one entry ID. """ execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand") upload_file_command({"incidentId": "1", "entryID": "1"}) assert len(execute_command_mocker.call_args_list) == 1 assert execute_command_mocker.call_args_list[0][0][1]["entryID"] == "1" RAW_RESPONSE_ERROR = [ { "Brand": "Core REST API", "Category": "Utilities", "Contents": { "response": { "ShardID": 0, "category": "", "childs": None, "closed": "0001-01-01T00:00:00Z", "created": "2020-07-08T11:30:14.386972+03:00", "creatingUserId": "admin", "dbotCreatedBy": "admin", "details": "", "entries": [ { "IndicatorTimeline": None, "ShardID": 0, "brand": "none", "category": "artifact", "contents": "", "contentsSize": 0, "created": "2020-07-13T10:57:32.956671+03:00", "cronView": False, "dbotCreatedBy": "admin", "endingDate": "0001-01-01T00:00:00Z", "entryTask": None, "errorSource": "", "file": "test_name.pdf", "fileID": "94b6eed9-fd5d-412e-87dc-5e2a7cdc8457", "fileMetadata": { "info": "PDF document, version 1.3", "isMediaFile": False, "md5": "be357a5a72c1236a42590ea894866884", "sha1": "29f4d33b0895aa46a1dad97b54096d4e68e3d91a", "sha256": "6577fb6b9a5f4779f835fec594d21267aedf3c7f3ee183c4bc2fd3f9fca52df4", "sha512": "174c4574cbe429bb646e19ed1973b2ccfc314bc951f251684591105d1d0f5fe8aa9f67beae0d7420" "becec61bcfd51e17d270683242b77dc552e58b2f95a0625b", "size": 250301, "ssdeep": "6144:IwUbC/ok4IeyUvtRCRvhp2h8C+GYPcX6PorXMido+vWap81:z7ok4XjtIaxrKArXBoqw", "type": "application/pdf", }, "format": "", "hasRole": False, "id": "121@899", "investigationId": "899", "isTodo": False, "mirrored": False, "modified": "2020-07-13T10:57:33.011106+03:00", "note": False, "parentContent": None, "parentEntryTruncated": False, "parentId": "", "pinned": False, "playbookId": "", "previousRoles": None, "recurrent": False, "reputationSize": 0, "reputations": None, "retryTime": "0001-01-01T00:00:00Z", "roles": None, "scheduled": False, "sortValues": None, "startDate": "0001-01-01T00:00:00Z", "system": "", "tags": [], "tagsRaw": [], "taskId": "", "times": 0, "timezoneOffset": 0, "type": 3, "user": "admin", "version": 2, } ], "entryUsers": ["admin"], "hasRole": False, "highPriority": False, "id": "899", "invContext": None, "lastOpen": "0001-01-01T00:00:00Z", "mirrorAutoClose": None, "mirrorTypes": None, "modified": "2020-07-08T11:31:30.594058+03:00", "name": "office365_Feed", "previousRoles": None, "rawCategory": "", "reason": None, "roles": None, "runStatus": "completed", "slackMirrorAutoClose": False, "slackMirrorType": "", "sortValues": None, "status": 0, "systems": None, "tags": ["Automation"], "totalEntries": 0, "type": 0, "users": ["admin", "analyst"], "version": 8, } }, "ContentsFormat": "json", "EntryContext": None, "Evidence": False, "EvidenceID": "", "File": "", "FileID": "", "FileMetadata": None, "HumanReadable": None, "ID": "", "IgnoreAutoExtract": False, "ImportantEntryContext": None, "IndicatorTimeline": None, "Metadata": { "IndicatorTimeline": None, "ShardID": 0, "brand": "Core REST API", "category": "", "contents": "", "contentsSize": 0, "created": "2020-07-13T10:57:33.044787+03:00", "cronView": False, "dbotCreatedBy": "", "endingDate": "0001-01-01T00:00:00Z", "entryTask": None, "errorSource": "", "file": "", "fileID": "", "fileMetadata": None, "format": "json", "hasRole": False, "id": "", "instance": "Core REST API_instance_1", "investigationId": "737", "isTodo": False, "mirrored": False, "modified": "0001-01-01T00:00:00Z", "note": False, "parentContent": '!core-api-multipart uri="entry/upload/899" entryID="722@737" body="test_bark"', "parentEntryTruncated": False, "parentId": "726@737", "pinned": False, "playbookId": "", "previousRoles": None, "recurrent": False, "reputationSize": 0, "reputations": None, "retryTime": "0001-01-01T00:00:00Z", "roles": None, "scheduled": False, "sortValues": None, "startDate": "0001-01-01T00:00:00Z", "system": "", "tags": None, "tagsRaw": None, "taskId": "", "times": 0, "timezoneOffset": 0, "type": 1, "user": "", "version": 0, }, "ModuleName": "Core REST API_instance_1", "Note": False, "ReadableContentsFormat": "", "System": "", "Tags": None, "Type": 4, "Version": 0, } ] def test_demisto_upload_file_error(mocker): """Unit test Given - upload_file_command command - command args When - mock the raw response as error - mock the Client's send_request. Then - run the upload_file Validate that the correct error was raised """ mocker.patch("UploadFile.upload_file", return_value=RAW_RESPONSE_ERROR) with pytest.raises(Exception, match="There was an issue uploading the file."): upload_file_command({"incidentId": "1", "entryID": "12@12", "body": "test_bark"}) @pytest.mark.parametrize( argnames="target, service", argvalues=[ ("incident attachment", "incident"), ("war room entry", "entry"), ("alert attachment", "incident"), ("issue attachment", "incident"), ], ) def test_demisto_upload_file_as_attachment(mocker, target, service): """ Given: - target where to upload the file When: - Running the upload_file_command command Then: - Validate the correct Uri was sent to the executeCommand """ import UploadFile mocker.patch("UploadFile.demisto.executeCommand") upload_file_command({"target": target, "entryID": "1"}) assert f"{service}/upload/" in UploadFile.demisto.executeCommand.call_args[0][1]["uri"] def test_upload_with_using_argument(mocker): """Unit test Given - Command args with one entry ID. When - Running the upload_file_command function. Then - Validate that the API request was called only one entry ID. """ execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand") upload_file_command({"incidentId": "1", "entryID": "1", "using": "instance_1"}) assert len(execute_command_mocker.call_args_list) == 1 assert execute_command_mocker.call_args_list[0][0][1]["using"] == "instance_1" SUCCESS_ENTRY = [{"Type": EntryType.NOTE, "Contents": {"response": {"entries": [{"id": "1@1"}]}}}] def _transient_entry(message: str) -> list: """Build an error entry whose contents match a transient error marker.""" return [{"Type": EntryType.ERROR, "Contents": message}] def _permanent_entry(message: str = "Item not found (8) - incident does not exist") -> list: """Build an error entry that is NOT transient (should not be retried).""" return [{"Type": EntryType.ERROR, "Contents": message}] TRANSIENT_MESSAGES = [ 'Post "https://api/xsoar/incident/upload/1": context deadline exceeded (Client.Timeout exceeded while awaiting headers)', "502 Bad Gateway", "The service is Under Maintenance, please try again later", "read tcp 1.2.3.4->5.6.7.8: connection reset by peer", ] def test_is_transient_error_transient_markers(): """ Given: error entries whose contents contain a known transient marker (any case). When: calling is_transient_error. Then: it returns True. """ for message in TRANSIENT_MESSAGES: assert is_transient_error({"Type": EntryType.ERROR, "Contents": message}) is True # case-insensitive check assert is_transient_error({"Type": EntryType.ERROR, "Contents": "CONTEXT DEADLINE EXCEEDED"}) is True def test_is_transient_error_non_transient(): """ Given: a permanent error entry, a success entry, and a None entry. When: calling is_transient_error. Then: it returns False for all of them. """ assert is_transient_error({"Type": EntryType.ERROR, "Contents": "incident does not exist"}) is False assert is_transient_error(SUCCESS_ENTRY[0]) is False assert is_transient_error(None) is False def test_upload_file_success_first_try(mocker): """ Given: executeCommand succeeds on the first attempt. When: calling upload_file. Then: executeCommand is called once, no sleep occurs, and the success response is returned. """ execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand", return_value=SUCCESS_ENTRY) sleep_mocker = mocker.patch("UploadFile.time.sleep") result = upload_file("1", "1@1") assert result == SUCCESS_ENTRY assert execute_command_mocker.call_count == 1 assert sleep_mocker.call_count == 0 @pytest.mark.parametrize("num_failures", [1, 2, 3]) def test_upload_file_success_after_transient_failures(mocker, num_failures): """ Given: executeCommand returns N transient errors then succeeds. When: calling upload_file. Then: executeCommand is called N+1 times, sleep is called N times, and success is returned. """ side_effects = [_transient_entry("context deadline exceeded")] * num_failures + [SUCCESS_ENTRY] execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand", side_effect=side_effects) sleep_mocker = mocker.patch("UploadFile.time.sleep") result = upload_file("1", "1@1") assert result == SUCCESS_ENTRY assert execute_command_mocker.call_count == num_failures + 1 assert sleep_mocker.call_count == num_failures @pytest.mark.parametrize("message", TRANSIENT_MESSAGES) def test_upload_file_retries_each_transient_marker(mocker, message): """ Given: executeCommand returns a specific transient marker once, then succeeds. When: calling upload_file. Then: the call is retried and ultimately succeeds. """ execute_command_mocker = mocker.patch( "UploadFile.demisto.executeCommand", side_effect=[_transient_entry(message), SUCCESS_ENTRY] ) mocker.patch("UploadFile.time.sleep") result = upload_file("1", "1@1") assert result == SUCCESS_ENTRY assert execute_command_mocker.call_count == 2 def test_upload_file_permanent_error_not_retried(mocker): """ Given: executeCommand returns a permanent (non-transient) error. When: calling upload_file. Then: it is NOT retried - executeCommand is called once, no sleep, and the error is returned. """ permanent = _permanent_entry() execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand", return_value=permanent) sleep_mocker = mocker.patch("UploadFile.time.sleep") result = upload_file("1", "1@1") assert result == permanent assert execute_command_mocker.call_count == 1 assert sleep_mocker.call_count == 0 def test_upload_file_retries_exhausted(mocker): """ Given: executeCommand always returns a transient error. When: calling upload_file. Then: it retries MAX_RETRIES times (4 total calls, 3 sleeps) and returns the last error response. """ from UploadFile import MAX_RETRIES transient = _transient_entry("502 Bad Gateway") execute_command_mocker = mocker.patch("UploadFile.demisto.executeCommand", return_value=transient) sleep_mocker = mocker.patch("UploadFile.time.sleep") result = upload_file("1", "1@1") assert result == transient assert execute_command_mocker.call_count == MAX_RETRIES + 1 assert sleep_mocker.call_count == MAX_RETRIES def test_upload_file_exponential_backoff_values(mocker): """ Given: executeCommand always returns a transient error. When: calling upload_file. Then: sleep is called with exponentially increasing, capped delays (2, 4, 8). """ from UploadFile import BASE_BACKOFF, MAX_BACKOFF, MAX_RETRIES mocker.patch("UploadFile.demisto.executeCommand", return_value=_transient_entry("connection reset by peer")) sleep_mocker = mocker.patch("UploadFile.time.sleep") upload_file("1", "1@1") expected_delays = [min(BASE_BACKOFF * 2**i, MAX_BACKOFF) for i in range(MAX_RETRIES)] actual_delays = [call.args[0] for call in sleep_mocker.call_args_list] assert actual_delays == expected_delays def test_upload_file_command_transient_then_success(mocker): """ Given: the first upload attempt hits a transient error, the retry succeeds. When: running upload_file_command. Then: no error is raised and a success readable output is returned. """ mocker.patch( "UploadFile.demisto.executeCommand", side_effect=[_transient_entry("context deadline exceeded"), SUCCESS_ENTRY], ) mocker.patch("UploadFile.time.sleep") command_results = upload_file_command({"incID": "1", "entryID": "1@1"}) assert "File uploaded successfully." in command_results[0].readable_output def test_upload_file_command_transient_exhausted_raises(mocker): """ Given: every upload attempt hits a transient error and retries are exhausted. When: running upload_file_command. Then: the original upload error is raised (error surface preserved). """ mocker.patch("UploadFile.demisto.executeCommand", return_value=_transient_entry("502 Bad Gateway")) mocker.patch("UploadFile.time.sleep") with pytest.raises(Exception, match="There was an issue uploading the file."): upload_file_command({"incID": "1", "entryID": "1@1"})