VectraRUXEventsDetections
This integration allows the security operations center to create and manage incidents based on Vectra Events Detections.
Network Security · Vectra RUX
Details
| ID | VectraRUXEventsDetections |
|---|---|
| Provider | Vectra AI |
| Category | Network Security |
| From Version | 6.10.0 |
| Docker Image | demisto/python3:3.12.13.10116658 |
| Supported Modules | Agentix Attack Surface Management Cortex Cloud Application Security Cloud Posture Security Cloud Runtime Security EDR Email Security Exposure Management Threat Intelligence Management XSIAM |
README
This integration allows the security operations center to create and manage incidents based on Vectra Events Detections.
This integration was integrated and tested with Vectra API v3.5.
Configure Vectra RUX - Network Detection & Response in Cortex
| Parameter | Description | Required |
|---|---|---|
| Server URL | URL of the Vectra AI platform. | True |
| Client ID | Identifies a client or application for authentication and authorization in the Vectra AI platform. | True |
| Client Secret Key | Secret key used for secure communication with the Vectra AI platform. | True |
| Fetch incidents | False | |
| Max Fetch | The maximum number of events detections to fetch each time. If the value is greater than 200, it will be considered as 200. The maximum is 200. | False |
| First Fetch Time | The date or relative timestamp from which to begin fetching events detections. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 Oct 2025, 01 Mar 2021 04:45:33, 2025-12-17T14:05:44Z. |
False |
| Entity Types | Filter by entity type. If not selected, it will fetch all events detections. | False |
| Create Incidents for Prioritized Detections | Enabling this checkbox generates incidents for prioritized events detections. If not selected, incidents are created for all events detections. | False |
| Create Incidents for Escalated Detections | Enabling this checkbox generates incidents for escalated events detections. If not selected, incidents are created for all events detections. | False |
| Mirroring Direction | The mirroring direction in which to mirror the detections. You can mirror ‘Incoming’ (from Vectra to XSOAR), ‘Outgoing’ (from XSOAR to Vectra), or in both directions. | False |
| Mirror tag for notes | The tag value should be used to mirror the detection note by adding the same tag in the notes. | False |
| Open Detection on Incident Reopen | Enabling this checkbox opens the detection in Vectra when the incident is reopened in XSOAR. Note: This parameter is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’. |
False |
| Detection Status for Incident Reopen | Detection status to set in Vectra when incident is reopened in XSOAR. Default value is ‘Escalated’. Note: This parameter is only used when open detection on incident reopen is ‘checked’ and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’. |
False |
| Close Detection on Incident Closure | Enabling this checkbox closes the detection in Vectra when the incident is closed in XSOAR. Note: This parameter is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’. |
False |
| Detection Close Reason for Incident Closure | Detection close reason to set in Vectra when closing incidents in XSOAR. Default value is ‘Remediated’. Note: This parameter is only used when close detection on incident closer is ‘checked’ and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’. |
False |
| Incident type | False | |
| Trust any certificate (not secure) | When checked, no SSL certificates check will be done when interacting with the Vectra RUX API. It’s insecure. (Default - unchecked) | False |
| Use system proxy settings | Use the system proxy settings to reach with the Vectra RUX API. | False |
Configuration for fetching Vectra RUX Events Detections as an XSOAR Incident
To fetch Vectra RUX Events Detections follow the next steps:
- Select Fetches incidents.
- Under Classifier, select “N/A”.
- Under Incident type, select “Vectra RUX Events Detection”.
- Under Mapper (incoming), select “Vectra RUX - Incoming Mapper” for default mapping.
- Enter connection parameters. (Server URL, Client ID & Client Secret Key)
- Update “Max Fetch” & “First Fetch Time” based on your requirements.
- Filter the Detections by the “Entity Type”(Account and Host).
- Filter the Detections by “Create Incidents for Prioritized Detections”, “Create Incidents for Escalated Detections”:
- Default Behavior: By default, the integration retrieves all event detections across all entity types (Account and Host) and all detection statuses (Open, Acknowledged, Escalated, Paused). This includes both prioritized and non-prioritized detections.
- Fetch Only Prioritized Detections: Enable “Create Incidents for Prioritized Detections” to filter out non-prioritized detections. Incidents will be created only for prioritized event detections.
- Fetch Only Escalated Detections: Enable “Create Incidents for Escalated Detections” to retrieve all escalated detections, regardless of their priority level.
- Fetch Prioritized and Escalated Detections: Enable both “Create Incidents for Prioritized Detections” and “Create Incidents for Escalated Detections”. This configuration retrieves detections that are either prioritized or escalated.
- Select the Incident Mirroring Direction:
- Incoming - Mirrors changes from the Vectra RUX Detection into the Cortex XSOAR incident.
- Outgoing - Mirrors changes from the Cortex XSOAR incident to the Vectra RUX Detection.
- Incoming And Outgoing - Mirrors changes both Incoming and Outgoing directions on incidents.
- Enter the relevant tag name for mirror notes.
Note: This value is mapped to the dbotMirrorTags incident field in Cortex XSOAR, which defines how Cortex XSOAR handles notes when you tag them in the War Room. This is required for mirroring notes from Cortex XSOAR to Vectra RUX. - Uncheck the “Open Detection on Incident Reopen” option if you don’t want to open the detection in Vectra when the incident is reopened in XSOAR. This option is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Select the “Detection Status for Incident Reopen” option if you want to set the detection status in Vectra when the incident is reopened in XSOAR. Default value is ‘Escalated’. This option is only used when the “Open Detection on Incident Reopen” option is checked and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Uncheck the “Close Detection on Incident Closure” option if you don’t want to close the detection in Vectra when the incident is closed in XSOAR. This option is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Select the “Detection Close Reason for Incident Closure” option if you want to set the detection close reason in Vectra when the incident is closed in XSOAR. Default value is ‘Benign’. This option is only used when the “Close Detection on Incident Closure” option is checked and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Select SSL certificate validation and Proxy if required.
Notes for mirroring:
- This feature is compliant with XSOAR version 6.0 and above.
- When mirroring incidents, you can make changes in Vectra that will be reflected in Cortex XSOAR, or vice versa.
- Any tags removed from the Vectra entity will not be removed in the XSOAR incident, as XSOAR doesn’t allow the removal of the tags field via the backend. However, tags removed from the XSOAR incident UI will be removed from the Vectra entity.
- New notes from the XSOAR incident will be created as notes in the Vectra Detection. Updates to existing notes in the XSOAR incident will not be reflected in the Vectra Detection.
- New notes from the Vectra Detection will be created as notes in the XSOAR incident. Updates to existing notes in the Vectra Detection will create new notes in the XSOAR incident.
- If the Detection Status is updated in the Vectra Detection, it will be reflected in the XSOAR incident, or vice versa.
- If you want to reopen a detection in Vectra when the incident is reopened in XSOAR, check the “Open Detection on Incident Reopen” option. This option is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Set the “Detection Status for Incident Reopen” option to set the detection status in Vectra when the incident is reopened in XSOAR. Default value is ‘Escalated’. This option is only used when the “Open Detection on Incident Reopen” option is checked and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- If you want to close a detection in Vectra when the incident is closed in XSOAR, check the “Close Detection on Incident Closure” option. This option is only used when the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- Set the “Detection Close Reason for Incident Closure” option to set the detection close reason in Vectra when the incident is closed in XSOAR. Default value is ‘Benign’. This option is only used when the “Close Detection on Incident Closure” option is checked and the mirroring direction is set to ‘Outgoing’ or ‘Incoming And Outgoing’.
- The mirroring settings apply only for incidents that are fetched after applying the settings.
- The mirroring is strictly tied to Incident type “Vectra RUX Events Detection” & Incoming mapper “Vectra RUX - Incoming Mapper” If you want to change or use your custom incident type/mapper then make sure changes related to these are present.
- If you want to use the mirror mechanism and you’re using custom mappers, then the incoming mapper must contain the following fields: dbotMirrorDirection, dbotMirrorId, dbotMirrorInstance, and dbotMirrorTags.
- To use a custom mapper, you must first duplicate the mapper and update the fields in the copy of the mapper. (Refer to the “Create a custom mapper consisting of the default Vectra RUX mapper” section for more information.)
- Following new fields are introduced in the response of the incident to enable the mirroring:
- mirror_direction: This field determines the mirroring direction for the incident. It is a required field for XSOAR to enable mirroring support.
- mirror_tags: This field determines what would be the tag needed to mirror the XSOAR entry out to Vectra RUX. It is a required field for XSOAR to enable mirroring support.
- mirror_instance: This field determines from which instance the XSOAR incident was created. It is a required field for XSOAR to enable mirroring support.
Expire Inactive Detections
- Use the Expire Inactive Detections - Vectra RUX playbook to expire inactive detections that are fetched in XSOAR.
- You can also schedule a job with the Expire Inactive Detections - Vectra RUX playbook in Cortex XSOAR to expire inactive detections periodically. Refer to Cortex XSOAR documentation for more information. To create a job with a 24-hour recurring schedule, follow these steps:
- In Cortex XSOAR, navigate to Jobs (via the top menu or sidebar).
- Click New Job.
- Select Time triggered and enable Recurring.
- Set the schedule to Every 24 hours (or configure a specific daily time using a cron expression such as
0 0 * * *). - Set the Name for the job (e.g.,
Expire Inactive Detections - Daily). - Under Playbook, select Expire Inactive Detections - Vectra RUX.
- Click Create new job to activate the job.
Commands
You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.
vectra-detections-mark-asclosed
Mark detections as closed with provided detection IDs in the argument.
Base Command
vectra-detections-mark-asclosed
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_ids | Provide a list of detection IDs separated by commas or a single detection ID. | Required |
| close_reason | Provide the close reason. Possible values are: benign, remediated. | Required |
Context Output
There is no context output for this command.
Command example
!vectra-detections-mark-asclosed detection_ids=123,345 close_reason=remediated
Human Readable Output
The provided detection IDs have been successfully closed as remediated
vectra-user-list
Returns a list of users.
Base Command
vectra-user-list
Input
| Argument Name | Description | Required |
|---|---|---|
| Filter by email. | Optional | |
| role | Filter users with the specified role. Use the role standardized name. Possible values are: Admin, Auditor, Global Analyst, Read-Only, Restricted Admin, Security Analyst, Setting Admin, Super Admin. | Optional |
| last_login_timestamp | Return only the users which have a last login timestamp equal to or after the given timestamp. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 May 2023, 01 Mar 2021 04:45:33, 2022-04-17T14:05:44Z. |
Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.User.id | Number | The ID of the User. |
| Vectra.User.user_id | Number | The ID of the User. |
| Vectra.User.name | String | Username of the user. |
| Vectra.User.email | String | The email associated with the user. |
| Vectra.User.role | String | The role associated with the user. |
| Vectra.User.last_login_timestamp | String | Last login timestamp in UTC format of the user. |
| Vectra.User.last_login | String | Last login timestamp of the user. |
Command example
#### Context Example
```json
{
"Vectra": {
"User": [
{
"id": 59,
"user_id": 59,
"username": "user.name1",
"email": "",
"role": "Security Analyst",
"last_login_timestamp": "2023-08-22T09:24:44Z",
"last_login": "2023-08-22T09:24:44Z"
},
{
"id": 32,
"user_id": 32,
"username": "user.name2",
"email": "",
"role": "Super Admin",
"last_login_timestamp": "2023-07-02T18:41:19Z",
"last_login": "2023-07-02T18:41:19Z"
},
{
"id": 23,
"user_id": 23,
"username": "vectra_mdr",
"email": "",
"role": "Vectra MDR"
}
]
}
}
Human Readable Output
Users Table
User ID User Name Role Last Login Timestamp 59 user.name1 Security Analyst 2023-08-22T09:24:44Z 32 user.name2 Super Admin 2023-07-02T18:41:19Z 23 vectra_mdr Vectra MDR
vectra-entity-list
Returns a list of entities.
Base Command
vectra-entity-list
Input
| Argument Name | Description | Required |
|---|---|---|
| prioritized | Fetch only entities whose priority score is above the configured priority threshold will be included in the response. Possible values are: true, false. | Optional |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Optional |
| name | Filter by matching entity name. | Optional |
| tags | Filter by a tag or a comma-separated list of tags. | Optional |
| state | Filter on entity activation state. Possible values are: active, inactive. | Optional |
| ordering | Orders records by last timestamp or urgency score. Default sorting is by urgency score in descending order. Use the minus symbol (-) to sort scores in descending order. Multiple ordering fields can be specified with a comma-separated list (e.g., ordering=urgency_score,-name). | Optional |
| last_detection_timestamp | Return only the entities which have a last detection timestamp equal to or after the given timestamp. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 May 2023, 01 Mar 2021 04:45:33, 2022-04-17T14:05:44Z. |
Optional |
| page | Enables the caller to specify a particular page of results. Default is 1. | Optional |
| page_size | Specify the desired page size for the request. Maximum is 5000. Default is 50. | Optional |
| last_modified_timestamp | Return only the entities which have a last modified timestamp equal to or after the given timestamp. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 May 2023, 01 Mar 2021 04:45:33, 2022-04-17T14:05:44Z. |
Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.id | Number | ID of the entity. |
| Vectra.Entity.name | String | Name of the entity. |
| Vectra.Entity.breadth_contrib | Number | Breadth contribution of the entity. |
| Vectra.Entity.importance | Number | Entity importance. |
| Vectra.Entity.type | String | Type of the entity. |
| Vectra.Entity.is_prioritized | Boolean | Entity is prioritized or not. |
| Vectra.Entity.severity | String | Severity of the entity. |
| Vectra.Entity.urgency_score | Number | Urgency score of the entity. |
| Vectra.Entity.velocity_contrib | Number | Velocity contribution of the entity. |
| Vectra.Entity.detection_set | String | Set of detections related to entity. |
| Vectra.Entity.last_detection_timestamp | Date | Time of the last detection activity related to entity. |
| Vectra.Entity.notes.id | String | Notes of the entity. |
| Vectra.Entity.notes.dateCreated | String | Created date of the Note. |
| Vectra.Entity.notes.dateModified | String | Modified date of the Note. |
| Vectra.Entity.notes.createdBy | String | Created user of the Note. |
| Vectra.Entity.notes.ModifiedBy | String | Modified user of the Note. |
| Vectra.Entity.notes.note | String | Note of the entity. |
| Vectra.Entity.attack_rating | Number | Attack Ratting of the entity. |
| Vectra.Entity.privilege_level | String | Privilege Level of the entity. |
| Vectra.Entity.privilege_category | String | Privilege Category of the entity. |
| Vectra.Entity.attack_profile | String | Attack Profile of the entity. |
| Vectra.Entity.sensors | Unknown | Sensors of the entity. |
| Vectra.Entity.state | String | State of the entity. |
| Vectra.Entity.tags | Unknown | Tags of the entity. |
| Vectra.Entity.url | String | Url link of the entity. |
| Vectra.Entity.host_type | Unknown | Host type of the entity. |
| Vectra.Entity.account_type | String | Account type of the entity. |
Command example
!vectra-entity-list entity_type=account page=1 page_size=4 tags=test,test1 prioritized=true state=active
Context Example
{
[
{
"id": 334,
"name": "account_name",
"breadth_contrib": 2,
"entity_importance": 1,
"importance": 2,
"entity_type": "account",
"type": "account",
"is_prioritized": true,
"severity": "Critical",
"urgency_score": 100,
"velocity_contrib": 2,
"detection_set": [
"http://server_url.com/api/v3.3/detections/1933",
"http://server_url.com/api/v3.3/detections/1934"
],
"last_detection_timestamp": "2023-05-15T09:39:24Z",
"last_modified_timestamp": "2023-07-27T08:56:09Z",
"notes": [],
"attack_rating": 10,
"attack_profile": "AWS Threat Actor",
"sensors": [
"test"
],
"state": "active",
"tags": [
"test"
],
"url": "http://server_url.com/api/v3.3/accounts/334",
"account_type": [
"o365"
]
},
{
"id": 335,
"name": "account_name_1",
"breadth_contrib": 2,
"entity_importance": 1,
"importance": 2,
"entity_type": "account",
"type": "account",
"is_prioritized": true,
"severity": "Critical",
"urgency_score": 80,
"velocity_contrib": 2,
"detection_set": [
"http://server_url.com/api/v3.3/detections/1935",
"http://server_url.com/api/v3.3/detections/1937"
],
"last_detection_timestamp": "2023-05-15T09:41:24Z",
"last_modified_timestamp": "2023-07-27T08:56:09Z",
"notes": [],
"attack_rating": 6,
"attack_profile": "attack1",
"sensors": [],
"state": "active",
"tags": [
"test",
"test1"
],
"url": "http://server_url.com/api/v3.3/accounts/335",
"account_type": [
"o365"
]
},
{
"id": 337,
"name": "account_name_2",
"breadth_contrib": 2,
"entity_importance": 1,
"importance": 1,
"entity_type": "account",
"type": "account",
"is_prioritized": true,
"severity": "Critical",
"urgency_score": 40,
"velocity_contrib": 2,
"detection_set": [
"http://server_url.com/api/v3.3/detections/1835",
"http://server_url.com/api/v3.3/detections/1837"
],
"last_detection_timestamp": "2023-05-15T09:40:24Z",
"last_modified_timestamp": "2023-07-27T08:56:09Z",
"notes": [],
"attack_rating": 9,
"attack_profile": "attack2",
"sensors": [],
"state": "active",
"tags": [
"test1"
],
"url": "http://server_url.com/api/v3.3/accounts/337",
"account_type": [
"aws"
]
},
{
"id": 339,
"name": "account_name_3",
"breadth_contrib": 2,
"entity_importance": 1,
"importance": 2,
"entity_type": "account",
"type": "account",
"is_prioritized": true,
"severity": "Critical",
"urgency_score": 21,
"velocity_contrib": 2,
"detection_set": [
"http://server_url.com/api/v3.3/detections/1735",
"http://server_url.com/api/v3.3/detections/1737"
],
"last_detection_timestamp": "2023-05-15T09:44:24Z",
"last_modified_timestamp": "2023-07-27T08:56:09Z",
"notes": [],
"attack_rating": 5,
"attack_profile": "attack3",
"sensors": [],
"state": "active",
"tags": [
"test"
],
"url": "http://server_url.com/api/v3.3/accounts/339",
"account_type": [
"o365"
]
}
]
}
Human Readable Output
Entities Table (Showing Page 1 out of 1)
ID Name Entity Type Urgency Score Entity Importance Last Detection Timestamp Last Modified Timestamp Detections IDs Prioritize State Tags 334 account_name account 100 High 2023-05-15T09:39:24Z 2023-07-18T09:44:24Z 1933, 1934 true active test 335 account_name_1 account 80 High 2023-05-15T09:41:24Z 2023-07-17T09:44:24Z 1935, 1937 true active test, test1 337 account_name_2 account 40 Medium 2023-05-15T09:40:24Z 2023-07-16T09:44:24Z 1835, 1837 true active test1 339 account_name_3 account 21 High 2023-05-15T09:44:24Z 2023-07-15T09:44:24Z 1735, 1737 true active test
vectra-entity-describe
Describes an entity by ID.
Base Command
vectra-entity-describe
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: host, account. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.id | Number | ID of the entity. |
| Vectra.Entity.name | String | Name of the entity. |
| Vectra.Entity.breadth_contrib | Number | Breadth contribution of the entity. |
| Vectra.Entity.importance | Number | Entity importance. |
| Vectra.Entity.type | String | Type of the entity. |
| Vectra.Entity.is_prioritized | Boolean | Entity is prioritized or not. |
| Vectra.Entity.severity | String | Severity of the entity. |
| Vectra.Entity.urgency_score | Number | Urgency score of the entity. |
| Vectra.Entity.velocity_contrib | Number | Velocity contribution of the entity. |
| Vectra.Entity.detection_set | String | Set of detections related to the entity. |
| Vectra.Entity.last_detection_timestamp | Date | Time of the last detection activity related to the entity. |
| Vectra.Entity.last_modified_timestamp | Date | Time of the last modification activity related to the entity. |
| Vectra.Entity.notes.id | String | Notes of the entity. |
| Vectra.Entity.notes.dateCreated | String | Created date of the Note. |
| Vectra.Entity.notes.dateModified | String | Modified date of the Note. |
| Vectra.Entity.notes.createdBy | String | Created user of the Note. |
| Vectra.Entity.notes.ModifiedBy | String | Modified user of the Note. |
| Vectra.Entity.notes.note | String | Note of the entity. |
| Vectra.Entity.attack_rating | Number | Attack Ratting of the entity. |
| Vectra.Entity.privilege_level | String | Privilege Level of the entity. |
| Vectra.Entity.privilege_category | String | Privilege Category of the entity. |
| Vectra.Entity.attack_profile | String | Attack Profile of the entity. |
| Vectra.Entity.sensors | Unknown | Sensors of the entity. |
| Vectra.Entity.state | String | State of the entity. |
| Vectra.Entity.tags | Unknown | Tags of the entity. |
| Vectra.Entity.url | String | Url link of the entity. |
| Vectra.Entity.host_type | Unknown | Host type of the entity. |
| Vectra.Entity.account_type | Unknown | Account type of the entity. |
Command example
!vectra-entity-describe entity_type=account entity_id=334
Context Example
{
"id": 334,
"name": "account_name",
"breadth_contrib": 2,
"entity_importance": 1,
"importance": 2,
"entity_type": "account",
"type": "account",
"is_prioritized": true,
"severity": "Critical",
"urgency_score": 100,
"velocity_contrib": 2,
"detection_set": [
"http://server_url.com/api/v3.3/detections/1933",
"http://server_url.com/api/v3.3/detections/1934"
],
"last_detection_timestamp": "2023-05-15T09:39:24Z",
"last_modified_timestamp": "2023-07-28T05:25:47Z",
"notes": [],
"attack_rating": 10,
"attack_profile": "test_attack",
"sensors": [
"test"
],
"state": "active",
"tags": [
"test"
],
"url": "http://server_url.com/api/v3.3/accounts/334",
"account_type": [
"o365"
]
}
}
Human Readable Output
Entity detail
Entity ID: 334
Name Entity Type Urgency Score Entity Importance Last Detection Timestamp Last Modified Timestamp Detections IDs Prioritize State Tags account_name account 100 High 2023-05-15T09:39:24Z 2023-07-28T05:25:47Z 1933, 1934 true active test
vectra-entity-detection-list
Returns a list of detections for a specified entity.
Base Command
vectra-entity-detection-list
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| page | Enables the caller to specify a particular page of results. Default is 1. | Optional |
| page_size | Specify the desired page size for the request. Maximum is 5000. Default is 50. | Optional |
| detection_category | The category of the detection. Possible values are: Command & Control, Botnet, Reconnaissance, Lateral Movement, Exfiltration, Info. | Optional |
| detection_type | Filter by detection type. | Optional |
| last_timestamp | Return only the detections which have a last timestamp equal to or after the given timestamp. Formats: YYYY-MM-ddTHH:mm:ssZ, YYYY-MM-dd, N days, N hours. Example: 2023-04-25T00:00:00Z, 2023-04-25, 2 days, 5 hours, 01 Mar 2023, 01 Feb 2023 04:45:33, 15 Jun. |
Optional |
| detection_name | Filter by detection name. | Optional |
| state | Filter by state. Default is active. | Optional |
| tags | Filter by a tag or a comma-separated list of tags. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Detections.id | Number | Entity detection ID. |
| Vectra.Entity.Detections.assigned_date | Unknown | Date assigned to the detection. |
| Vectra.Entity.Detections.assigned_to | Unknown | User or entity assigned to the detection. |
| Vectra.Entity.Detections.category | String | Category of the detection. |
| Vectra.Entity.Detections.certainty | Number | Certainty level of the detection. |
| Vectra.Entity.Detections.c_score | Number | Confidence score of the detection. |
| Vectra.Entity.Detections.description | String | Description of the detection. |
| Vectra.Entity.Detections.detection | String | Detection information. |
| Vectra.Entity.Detections.detection_category | String | Category of the detection. |
| Vectra.Entity.Detections.detection_type | String | Type of the detection. |
| Vectra.Entity.Detections.grouped_details.external_target.ip | String | IP address of the external target in the detection group. |
| Vectra.Entity.Detections.grouped_details.external_target.name | String | Name of the external target in the detection group. |
| Vectra.Entity.Detections.grouped_details.num_sessions | Number | Number of sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.bytes_received | Number | Total bytes received in the detection group. |
| Vectra.Entity.Detections.grouped_details.bytes_sent | Number | Total bytes sent in the detection group. |
| Vectra.Entity.Detections.grouped_details.ja3_hashes | String | JA3 hashes in the detection group. |
| Vectra.Entity.Detections.grouped_details.ja3s_hashes | String | JA3S hashes in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.tunnel_type | String | Tunnel type used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.protocol | String | Protocol used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.app_protocol | String | Application protocol used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_port | Number | Destination port in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_ip | String | Destination IP address in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.bytes_received | Number | Total bytes received in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.bytes_sent | Number | Total bytes sent in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.first_timestamp | Date | First timestamp of the sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.last_timestamp | Date | Last timestamp of the sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo | Unknown | Geolocation of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo_lat | Unknown | Latitude of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo_lon | Unknown | Longitude of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.first_timestamp | Date | First timestamp of the detection group. |
| Vectra.Entity.Detections.grouped_details.last_timestamp | Date | Last timestamp of the detection group. |
| Vectra.Entity.Detections.grouped_details.dst_ips | String | Destination IP addresses in the detection group. |
| Vectra.Entity.Detections.grouped_details.dst_ports | Number | Destination ports in the detection group. |
| Vectra.Entity.Detections.grouped_details.target_domains | String | Target domains in the detection group. |
| Vectra.Entity.Detections.is_targeting_key_asset | Boolean | Indicates if the detection is targeting a key asset. |
| Vectra.Entity.Detections.last_timestamp | Date | Last timestamp of the detection. |
| Vectra.Entity.Detections.note | Unknown | Note associated with the detection. |
| Vectra.Entity.Detections.note_modified_by | Unknown | User or entity who last modified the note. |
| Vectra.Entity.Detections.note_modified_timestamp | Unknown | Timestamp when the note was last modified. |
| Vectra.Entity.Detections.notes | Unknown | Additional notes related to the detection. |
| Vectra.Entity.Detections.sensor_name | String | Name of the sensor associated with the detection. |
| Vectra.Entity.Detections.src_account.id | Number | ID of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.name | String | Name of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.url | String | URL of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.threat | Number | Threat level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.certainty | Number | Certainty level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.privilege_level | Number | Privilege level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.privilege_category | String | Privilege category of the source account associated with the detection. |
| Vectra.Entity.Detections.src_host.id | Number | ID of the source host in the detection. |
| Vectra.Entity.Detections.src_host.ip | String | IP address of the source host in the detection. |
| Vectra.Entity.Detections.src_host.name | String | Name of the source host in the detection. |
| Vectra.Entity.Detections.src_host.url | String | URL associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.is_key_asset | Boolean | Indicates if the source host is a key asset. |
| Vectra.Entity.Detections.src_host.groups | Unknown | Groups associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.threat | Number | Threat level associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.certainty | Number | Certainty level associated with the source host in the detection. |
| Vectra.Entity.Detections.src_ip | String | Source IP address in the detection. |
| Vectra.Entity.Detections.state | String | State of the detection. |
| Vectra.Entity.Detections.summary.bytes_received | Number | Total bytes received in the detection summary. |
| Vectra.Entity.Detections.summary.bytes_sent | Number | Total bytes sent in the detection summary. |
| Vectra.Entity.Detections.summary.cnc_server | String | CNC server associated with the detection summary. |
| Vectra.Entity.Detections.summary.num_events | Number | Total number of events related to the detection. |
| Vectra.Entity.Detections.summary.probable_owner | Unknown | Probable owner of the detection summary. |
| Vectra.Entity.Detections.summary.sessions | Number | Total sessions in the detection summary. |
| Vectra.Entity.Detections.tags | Unknown | Tags associated with the detection. |
| Vectra.Entity.Detections.threat | Number | Threat level of the detection. |
| Vectra.Entity.Detections.t_score | Number | T-score of the detection. |
| Vectra.Entity.Detections.type | String | Type of the detection. |
| Vectra.Entity.Detections.url | String | URL associated with the detection. |
Command example
!vectra-entity-detection-list entity_id=1
Context Example
{
[
{
"id": 132,
"category": "exfiltration",
"certainty": 70,
"c_score": 70,
"description": "",
"detection": "Data Smuggler",
"detection_category": "exfiltration",
"detection_type": "smuggler",
"grouped_details": [
{
"event_id": "ec2162c7-e526-4446-a549-71558743a1d7",
"event_name": "UpdateAssumeRolePolicy",
"aws_account_id": "aws_account_id",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\": \"stratus-red-team-backdoor-r-role\", \"policyDocument\": \"{\\\"Version\\\": \\\"2012-10-17\\\", \\\"Statement\\\": {\\\"Effect\\\": \\\"Allow\\\", \\\"Principal\\\": {\\\"AWS\\\": \\\"arn:aws:iam::123456789012:root\\\"}, \\\"Action\\\": \\\"sts:AssumeRole\\\"}}\"}"
],
"response_elements": [],
"role_sequence": [
"account_id",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_06e15b96-ee6b-482c-aff4-4f2f4a46a67c"
],
"last_timestamp": "2023-06-06T17:01:04Z"
},
{
"event_id": "89a098eb-1198-4e2a-9fa4-ef568ae39403",
"event_name": "UpdateAssumeRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"policyDocument\": \"{\\\"Version\\\": \\\"2012-10-17\\\", \\\"Statement\\\": {\\\"Effect\\\": \\\"Allow\\\", \\\"Principal\\\": {\\\"AWS\\\": \\\"arn:aws:iam::123456789012:root\\\"}, \\\"Action\\\": \\\"sts:AssumeRole\\\"}}\", \"roleName\": \"stratus-red-team-backdoor-r-role\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_a24eac3a-4fee-46ce-bc37-b4e675343fc9"
],
"last_timestamp": "2023-06-06T15:40:43Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T17:01:04Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "active",
"tags": [],
"threat": 80,
"t_score": 80,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/132"
},
{
"id": 135,
"category": "lateral_movement",
"certainty": 50,
"c_score": 50,
"description": "",
"detection": "AWS Suspect Admin Privilege Granting",
"detection_category": "lateral_movement",
"detection_type": "aws_admin_privilege_granted",
"grouped_details": [
{
"event_id": "85d88db5-cf2d-4b6e-9411-d3119d9920e0",
"event_name": "AttachRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\":\"stratus-red-team-backdoor-r-role\",\"policyArn\":\"arn:aws:iam::aws:policy/AdministratorAccess\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"APN/1.0 HashiCorp/1.0 Terraform/1.1.2 (+https://www.terraform.io) terraform-provider-aws/3.76.1 (+https://registry.terraform.io/providers/hashicorp/aws) aws-sdk-go/1.44.157 (go1.19.3; linux; amd64) stratus-red-team_5077134d-32ea-4403-996b-de30d7f278d7 HashiCorp-terraform-exec/0.17.3"
],
"last_timestamp": "2023-06-06T17:00:46Z"
},
{
"event_id": "ca157e7c-9a53-4012-9288-e6ac1c488fbc",
"event_name": "AttachRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\":\"stratus-red-team-backdoor-r-role\",\"policyArn\":\"arn:aws:iam::aws:policy/AdministratorAccess\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"APN/1.0 HashiCorp/1.0 Terraform/1.1.2 (+https://www.terraform.io) terraform-provider-aws/3.76.1 (+https://registry.terraform.io/providers/hashicorp/aws) aws-sdk-go/1.44.157 (go1.19.3; linux; amd64) stratus-red-team_01be8427-d1b5-4c18-8edb-0301c8e66c8e HashiCorp-terraform-exec/0.17.3"
],
"last_timestamp": "2023-06-06T15:40:07Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T17:00:46Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "fixed",
"summary": {
},
"tags": [],
"threat": 60,
"t_score": 60,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/135"
},
{
"id": 140,
"category": "reconnaissance",
"certainty": 40,
"c_score": 40,
"description": "",
"detection": "RPC Targeted Recon",
"detection_category": "reconnaissance",
"detection_type": "rpc_recon_1to1",
"grouped_details": [
{
"event_id": "cf9f469b-0a8e-47c6-85eb-5a0486292e58",
"event_name": "ModifySnapshotAttribute",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-west-2",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"snapshotId\":\"snap-0f7d022a2f4f67e08\",\"createVolumePermission\":{\"add\":{\"items\":[{\"userId\":\"012345678912\"}]}},\"attributeType\":\"CREATE_VOLUME_PERMISSION\"}"
],
"response_elements": [
"{\"requestId\":\"350c1eb8-b696-4d94-88d4-a764a0eed08b\",\"_return\":true}"
],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_a4dd596b-7a8d-4e77-a74d-13f19adf4403"
],
"last_timestamp": "2023-06-06T15:46:28Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T15:46:28Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "fixed",
"summary": {
},
"tags": [],
"threat": 60,
"t_score": 60,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/140"
}
]
}
Human Readable Output
Detections Table (Showing Page 1 out of 1)
ID Detection Name Detection Type Category Account Name Src IP Threat Score Certainty Score Number Of Events State Last Timestamp 132 Data Smuggler smuggler exfiltration account_name 0.0.0.0 80 70 0 active 2023-06-06T17:01:04Z 135 AWS Suspect Admin Privilege Granting aws_admin_privilege_granted lateral_movement account_name 0.0.0.0 60 50 0 fixed 2023-06-06T17:00:46Z 140 RPC Targeted Recon rpc_recon_1to1 reconnaissance account_name 0.0.0.0 60 40 0 fixed 2023-06-06T15:46:28Z
vectra-detection-describe
Returns a list of detections for the specified detection ID(s).
Base Command
vectra-detection-describe
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_ids | Specify the ID(s) of the detections. | Required |
| page | Enables the caller to specify a particular page of results. Default is 1. | Optional |
| page_size | Specify the desired page size for the request. Maximum is 5000. Default is 50. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Detections.id | Number | Entity detection ID. |
| Vectra.Entity.Detections.assigned_date | Unknown | Date assigned to the detection. |
| Vectra.Entity.Detections.assigned_to | Unknown | User or entity assigned to the detection. |
| Vectra.Entity.Detections.category | String | Category of the detection. |
| Vectra.Entity.Detections.certainty | Number | Certainty level of the detection. |
| Vectra.Entity.Detections.c_score | Number | Confidence score of the detection. |
| Vectra.Entity.Detections.description | String | Description of the detection. |
| Vectra.Entity.Detections.detection | String | Detection information. |
| Vectra.Entity.Detections.detection_category | String | Category of the detection. |
| Vectra.Entity.Detections.detection_type | String | Type of the detection. |
| Vectra.Entity.Detections.grouped_details.external_target.ip | String | IP address of the external target in the detection group. |
| Vectra.Entity.Detections.grouped_details.external_target.name | String | Name of the external target in the detection group. |
| Vectra.Entity.Detections.grouped_details.num_sessions | Number | Number of sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.bytes_received | Number | Total bytes received in the detection group. |
| Vectra.Entity.Detections.grouped_details.bytes_sent | Number | Total bytes sent in the detection group. |
| Vectra.Entity.Detections.grouped_details.ja3_hashes | String | JA3 hashes in the detection group. |
| Vectra.Entity.Detections.grouped_details.ja3s_hashes | String | JA3S hashes in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.tunnel_type | String | Tunnel type used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.protocol | String | Protocol used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.app_protocol | String | Application protocol used in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_port | Number | Destination port in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_ip | String | Destination IP address in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.bytes_received | Number | Total bytes received in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.bytes_sent | Number | Total bytes sent in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.first_timestamp | Date | First timestamp of the sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.last_timestamp | Date | Last timestamp of the sessions in the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo | Unknown | Geolocation of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo_lat | Unknown | Latitude of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.sessions.dst_geo_lon | Unknown | Longitude of the destination IP in the sessions of the detection group. |
| Vectra.Entity.Detections.grouped_details.first_timestamp | Date | First timestamp of the detection group. |
| Vectra.Entity.Detections.grouped_details.last_timestamp | Date | Last timestamp of the detection group. |
| Vectra.Entity.Detections.grouped_details.dst_ips | String | Destination IP addresses in the detection group. |
| Vectra.Entity.Detections.grouped_details.dst_ports | Number | Destination ports in the detection group. |
| Vectra.Entity.Detections.grouped_details.target_domains | String | Target domains in the detection group. |
| Vectra.Entity.Detections.is_targeting_key_asset | Boolean | Indicates if the detection is targeting a key asset. |
| Vectra.Entity.Detections.last_timestamp | Date | Last timestamp of the detection. |
| Vectra.Entity.Detections.note | Unknown | Note associated with the detection. |
| Vectra.Entity.Detections.note_modified_by | Unknown | User or entity who last modified the note. |
| Vectra.Entity.Detections.note_modified_timestamp | Unknown | Timestamp when the note was last modified. |
| Vectra.Entity.Detections.notes | Unknown | Additional notes related to the detection. |
| Vectra.Entity.Detections.sensor_name | String | Name of the sensor associated with the detection. |
| Vectra.Entity.Detections.src_account.id | Number | ID of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.name | String | Name of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.url | String | URL of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.threat | Number | Threat level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.certainty | Number | Certainty level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.privilege_level | Number | Privilege level of the source account associated with the detection. |
| Vectra.Entity.Detections.src_account.privilege_category | String | Privilege category of the source account associated with the detection. |
| Vectra.Entity.Detections.src_host.id | Number | ID of the source host in the detection. |
| Vectra.Entity.Detections.src_host.ip | String | IP address of the source host in the detection. |
| Vectra.Entity.Detections.src_host.name | String | Name of the source host in the detection. |
| Vectra.Entity.Detections.src_host.url | String | URL associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.is_key_asset | Boolean | Indicates if the source host is a key asset. |
| Vectra.Entity.Detections.src_host.groups | Unknown | Groups associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.threat | Number | Threat level associated with the source host in the detection. |
| Vectra.Entity.Detections.src_host.certainty | Number | Certainty level associated with the source host in the detection. |
| Vectra.Entity.Detections.src_ip | String | Source IP address in the detection. |
| Vectra.Entity.Detections.state | String | State of the detection. |
| Vectra.Entity.Detections.summary.bytes_received | Number | Total bytes received in the detection summary. |
| Vectra.Entity.Detections.summary.bytes_sent | Number | Total bytes sent in the detection summary. |
| Vectra.Entity.Detections.summary.cnc_server | String | CNC server associated with the detection summary. |
| Vectra.Entity.Detections.summary.num_events | Number | Total number of events related to the detection. |
| Vectra.Entity.Detections.summary.probable_owner | Unknown | Probable owner of the detection summary. |
| Vectra.Entity.Detections.summary.sessions | Number | Total sessions in the detection summary. |
| Vectra.Entity.Detections.tags | Unknown | Tags associated with the detection. |
| Vectra.Entity.Detections.threat | Number | Threat level of the detection. |
| Vectra.Entity.Detections.t_score | Number | T-score of the detection. |
| Vectra.Entity.Detections.type | String | Type of the detection. |
| Vectra.Entity.Detections.url | String | URL associated with the detection. |
Command example
!vectra-detection-describe detection_ids=132,135,140
Context Example
{
[
{
"id": 132,
"category": "exfiltration",
"certainty": 70,
"c_score": 70,
"description": "",
"detection": "Data Smuggler",
"detection_category": "exfiltration",
"detection_type": "smuggler",
"grouped_details": [
{
"event_id": "ec2162c7-e526-4446-a549-71558743a1d7",
"event_name": "UpdateAssumeRolePolicy",
"aws_account_id": "aws_account_id",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\": \"stratus-red-team-backdoor-r-role\", \"policyDocument\": \"{\\\"Version\\\": \\\"2012-10-17\\\", \\\"Statement\\\": {\\\"Effect\\\": \\\"Allow\\\", \\\"Principal\\\": {\\\"AWS\\\": \\\"arn:aws:iam::123456789012:root\\\"}, \\\"Action\\\": \\\"sts:AssumeRole\\\"}}\"}"
],
"response_elements": [],
"role_sequence": [
"account_id",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_06e15b96-ee6b-482c-aff4-4f2f4a46a67c"
],
"last_timestamp": "2023-06-06T17:01:04Z"
},
{
"event_id": "89a098eb-1198-4e2a-9fa4-ef568ae39403",
"event_name": "UpdateAssumeRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"policyDocument\": \"{\\\"Version\\\": \\\"2012-10-17\\\", \\\"Statement\\\": {\\\"Effect\\\": \\\"Allow\\\", \\\"Principal\\\": {\\\"AWS\\\": \\\"arn:aws:iam::123456789012:root\\\"}, \\\"Action\\\": \\\"sts:AssumeRole\\\"}}\", \"roleName\": \"stratus-red-team-backdoor-r-role\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_a24eac3a-4fee-46ce-bc37-b4e675343fc9"
],
"last_timestamp": "2023-06-06T15:40:43Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T17:01:04Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "active",
"summary": {
},
"tags": [],
"threat": 80,
"t_score": 80,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/132"
},
{
"id": 135,
"category": "lateral_movement",
"certainty": 50,
"c_score": 50,
"description": "",
"detection": "AWS Suspect Admin Privilege Granting",
"detection_category": "lateral_movement",
"detection_type": "aws_admin_privilege_granted",
"grouped_details": [
{
"event_id": "85d88db5-cf2d-4b6e-9411-d3119d9920e0",
"event_name": "AttachRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\":\"stratus-red-team-backdoor-r-role\",\"policyArn\":\"arn:aws:iam::aws:policy/AdministratorAccess\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"APN/1.0 HashiCorp/1.0 Terraform/1.1.2 (+https://www.terraform.io) terraform-provider-aws/3.76.1 (+https://registry.terraform.io/providers/hashicorp/aws) aws-sdk-go/1.44.157 (go1.19.3; linux; amd64) stratus-red-team_5077134d-32ea-4403-996b-de30d7f278d7 HashiCorp-terraform-exec/0.17.3"
],
"last_timestamp": "2023-06-06T17:00:46Z"
},
{
"event_id": "ca157e7c-9a53-4012-9288-e6ac1c488fbc",
"event_name": "AttachRolePolicy",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-east-1",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"roleName\":\"stratus-red-team-backdoor-r-role\",\"policyArn\":\"arn:aws:iam::aws:policy/AdministratorAccess\"}"
],
"response_elements": [],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"APN/1.0 HashiCorp/1.0 Terraform/1.1.2 (+https://www.terraform.io) terraform-provider-aws/3.76.1 (+https://registry.terraform.io/providers/hashicorp/aws) aws-sdk-go/1.44.157 (go1.19.3; linux; amd64) stratus-red-team_01be8427-d1b5-4c18-8edb-0301c8e66c8e HashiCorp-terraform-exec/0.17.3"
],
"last_timestamp": "2023-06-06T15:40:07Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T17:00:46Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "fixed",
"summary": {
},
"tags": [],
"threat": 60,
"t_score": 60,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/135"
},
{
"id": 140,
"category": "reconnaissance",
"certainty": 40,
"c_score": 40,
"description": "",
"detection": "RPC Targeted Recon",
"detection_category": "reconnaissance",
"detection_type": "rpc_recon_1to1",
"grouped_details": [
{
"event_id": "cf9f469b-0a8e-47c6-85eb-5a0486292e58",
"event_name": "ModifySnapshotAttribute",
"aws_account_id": "884414556547",
"src_external_host": {
"ip": "0.0.0.0"
},
"aws_region": "us-west-2",
"access_key_id": [
"123456"
],
"identity_type": "Federated Account",
"assumed_role": "AWSReservedSSO_AdministratorAccess_a670eb90f07e2960",
"request_parameters": [
"{\"snapshotId\":\"snap-0f7d022a2f4f67e08\",\"createVolumePermission\":{\"add\":{\"items\":[{\"userId\":\"012345678912\"}]}},\"attributeType\":\"CREATE_VOLUME_PERMISSION\"}"
],
"response_elements": [
"{\"requestId\":\"350c1eb8-b696-4d94-88d4-a764a0eed08b\",\"_return\":true}"
],
"role_sequence": [
"account_name",
"AWSReservedSSO_AdministratorAccess_a670eb90f07e2960"
],
"user_agent": [
"stratus-red-team_a4dd596b-7a8d-4e77-a74d-13f19adf4403"
],
"last_timestamp": "2023-06-06T15:46:28Z"
}
],
"is_targeting_key_asset": false,
"last_timestamp": "2023-06-06T15:46:28Z",
"notes": [],
"sensor_name": "mafosb50",
"src_account": {
"id": 21,
"name": "account_name",
"url": "http://server_url.com/api/v3.3/accounts/21",
"threat": 76,
"certainty": 35
},
"src_ip": "0.0.0.0",
"state": "fixed",
"summary": {
},
"tags": [],
"threat": 60,
"t_score": 60,
"type": "account",
"url": "http://server_url.com/api/v3.3/detections/140"
}
]
}
Human Readable Output
Detections Table (Showing Page 1 out of 1)
ID Detection Name Detection Type Category Account Name Src IP Threat Score Certainty Score Number Of Events State Last Timestamp 132 Data Smuggler smuggler exfiltration account_name 0.0.0.0 80 70 0 active 2023-06-06T17:01:04Z 135 AWS Suspect Admin Privilege Granting aws_admin_privilege_granted lateral_movement account_name 0.0.0.0 60 50 0 fixed 2023-06-06T17:00:46Z 140 RPC Targeted Recon rpc_recon_1to1 reconnaissance account_name 0.0.0.0 60 40 0 fixed 2023-06-06T15:46:28Z
vectra-entity-note-add
Add a note to the entity.
Base Command
vectra-entity-note-add
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| note | Note to be added in the specified entity_id. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Notes.entity_id | String | ID of the entity associated with the note. |
| Vectra.Entity.Notes.note_id | Number | ID of the note. |
| Vectra.Entity.Notes.date_created | Date | Date when the note was created. |
| Vectra.Entity.Notes.date_modified | Unknown | Date when the note was last modified. |
| Vectra.Entity.Notes.created_by | String | User who created the note. |
| Vectra.Entity.Notes.modified_by | Unknown | User who last modified the note. |
| Vectra.Entity.Notes.note | String | Content of the note. |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Notes.entity_id | String | The ID of the entity associated with the note. |
| Vectra.Entity.Notes.note_id | Number | The ID of the note. |
| Vectra.Entity.Notes.date_created | Date | The date when the note was created. |
| Vectra.Entity.Notes.date_modified | Unknown | The date when the note was last modified. |
| Vectra.Entity.Notes.created_by | String | The user who created the note. |
| Vectra.Entity.Notes.modified_by | Unknown | The user who last modified the note. |
| Vectra.Entity.Notes.note | String | The content of the note. |
Command example
!vectra-entity-note-add entity_id=1 entity_type=account note="test note"
Context Example
{
{
"date_created": "2023-06-21T06:19:15.224449Z",
"created_by": "test_user",
"note": "test_note",
"note_id": 19,
"entity_id": 1
}
}
Human Readable Output
The note has been successfully added to the entity
Returned Note ID: 19
vectra-entity-note-update
Update a note in the entity.
Base Command
vectra-entity-note-update
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| note_id | Specify the ID of the note. | Required |
| note | Note to be updated for the specified note_id. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Notes.entity_id | String | ID of the entity associated with the note. |
| Vectra.Entity.Notes.note_id | Number | ID of the note. |
| Vectra.Entity.Notes.date_created | Date | Date when the note was created. |
| Vectra.Entity.Notes.date_modified | Unknown | Date when the note was last modified. |
| Vectra.Entity.Notes.created_by | String | User who created the note. |
| Vectra.Entity.Notes.modified_by | Unknown | User who last modified the note. |
| Vectra.Entity.Notes.note | String | Content of the note. |
Command example
!vectra-entity-note-update entity_id=1 entity_type=account note_id=1 note="note modified"
Context Example
{
{
"date_created": "2023-06-16T04:55:58Z",
"date_modified": "2023-06-22T04:57:09Z",
"created_by": "test_user",
"modified_by": "test_user",
"note": "note modified",
"note_id": 8,
"entity_id": 1
}
}
Human Readable Output
The note has been successfully updated in the entity
vectra-entity-note-remove
Remove a note from the entity.
Base Command
vectra-entity-note-remove
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| note_id | Specify the ID of the note. | Required |
Context Output
There is no context output for this command.
Command Example
!vectra-entity-note-remove entity_id=1 entity_type=account note_id=1"
Context Example
{}
Human Readable Output
The note has been successfully removed from the entity
vectra-entity-tag-add
Add tags in the entity.
Base Command
vectra-entity-tag-add
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| tags | Comma-separated values of tags to be included in the entity. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Tags.tag_id | String | ID of the tag. |
| Vectra.Entity.Tags.entity_id | String | ID of the entity associated with the tag. |
| Vectra.Entity.Tags.entity_type | String | Type of the entity. |
| Vectra.Entity.Tags.tags | Unknown | A list of tags linked to an entity. |
Command example
!vectra-entity-tag-add entity_id=1 entity_type=host tags="tag1, tag2"
Context Example
{
{
"tag_id": "1",
"tags": [
"tag1",
"tag2"
],
"entity_type": "host",
"entity_id": 1
}
}
Human Readable Output
Tags have been successfully added to the entity
Updated list of tags: tag1, tag2
vectra-entity-tag-remove
Remove tags from the entity.
Base Command
vectra-entity-tag-remove
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| tags | Comma-separated values of tags to be removed from the entity. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Tags.tag_id | String | ID of the tag. |
| Vectra.Entity.Tags.entity_id | String | ID of the entity associated with the tag. |
| Vectra.Entity.Tags.entity_type | String | Type of the entity. |
| Vectra.Entity.Tags.tags | Unknown | A list of tags linked to an entity. |
Command example
!vectra-entity-tag-remove entity_id=1 entity_type=host tags="tag2"
Context Example
{
{
"tag_id": "1",
"tags": ["tag1"],
"entity_type": "host",
"entity_id": 1
}
}
Human Readable Output
Specified tags have been successfully removed for the entity
Updated list of tags: tag1
vectra-entity-tag-list
Returns a list of tags for a specified entity.
Base Command
vectra-entity-tag-list
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the id of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Tags.tag_id | String | ID of the tag. |
| Vectra.Entity.Tags.entity_id | String | ID of the entity associated with the tag. |
| Vectra.Entity.Tags.entity_type | String | Type of the entity. |
| Vectra.Entity.Tags.tags | Unknown | A list of tags linked to an entity. |
Command example
!vectra-entity-tag-list entity_id=1 entity_type=host
Context Example
{
"Vectra": {
"Entity": {
"Tags": {
"tag_id": "1",
"tags": [
"tag1",
"tag2"
],
"entity_type": "host",
"entity_id": 1
}
}
}
}
Human Readable Output
List of tags: tag1, tag2
vectra-entity-assignment-add
Add an assignment for the entity.
Base Command
vectra-entity-assignment-add
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| user_id | Specify the ID of the user. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Assignments.id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assignment_id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assigned_by.id | Number | ID of the user who assigned the entity. |
| Vectra.Entity.Assignments.assigned_by.username | String | Username of the user who assigned the entity. |
| Vectra.Entity.Assignments.date_assigned | Date | Date when the entity was assigned. |
| Vectra.Entity.Assignments.date_resolved | Date | Date when the entity was resolved. |
| Vectra.Entity.Assignments.events.assignment_id | Number | ID of the assignment event. |
| Vectra.Entity.Assignments.events.actor | Number | ID of the actor who performed the assignment event. |
| Vectra.Entity.Assignments.events.event_type | String | Type of assignment event. |
| Vectra.Entity.Assignments.events.datetime | Date | Date of the assignment event. |
| Vectra.Entity.Assignments.events.context.to | Number | ID of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.entity_t_score | Number | Threat score of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.entity_c_score | Number | Certainty score of the entity that was assigned to. |
| Vectra.Entity.Assignments.outcome.id | String | ID of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.builtin | String | Whether the assignment outcome is builtin or not. |
| Vectra.Entity.Assignments.outcome.user_selectable | String | Whether the assignment outcome is user selectable or not. |
| Vectra.Entity.Assignments.outcome.title | String | Title of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.category | String | Category of the assignment outcome. |
| Vectra.Entity.Assignments.resolved_by.id | Number | ID of the user who resolved the entity. |
| Vectra.Entity.Assignments.resolved_by.username | String | Username of the user who resolved the entity. |
| Vectra.Entity.Assignments.triaged_detections | Unknown | Number of detections that have been triaged for the entity. |
| Vectra.Entity.Assignments.host_id | Number | ID of the host that the entity is associated with. |
| Vectra.Entity.Assignments.account_id | Unknown | ID of the account that the entity is associated with. |
| Vectra.Entity.Assignments.assigned_to.id | Number | ID of the user who is currently assigned to the entity. |
| Vectra.Entity.Assignments.assigned_to.username | String | Username of the user who is currently assigned to the entity. |
Command Example
!vectra-entity-assignment-add entity_id=1 entity_type=account user_id=1
Context Example
{
{
"assigned_by": {
"id": 2,
"username": "test_user_2"
},
"date_assigned": "2023-07-24T08:52:59.367115Z",
"events": [
{
"assignment_id": 74,
"actor": 65,
"event_type": "created",
"datetime": "2023-07-24T08:52:59Z",
"context": {
"to": 60,
"entity_t_score": 0,
"entity_c_score": 0
}
}
],
"host_id": 10,
"assigned_to": {
"id": 1,
"username": "test.user@example.com"
},
"assignment_id": 1,
"id":1
}
}
Human Readable Output
The assignment has been successfully created
Assignment detail
Assignment ID Assigned By Assigned Date Assigned To Event Type 1 test_user_2 2023-07-24T08:52:59.367115Z test.user@example.com created
vectra-entity-assignment-update
Update an assignment in the entity.
Base Command
vectra-entity-assignment-update
Input
| Argument Name | Description | Required |
|---|---|---|
| assignment_id | Specify the ID of the assignment. | Required |
| user_id | Specify the ID of the user. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Assignments.id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assignment_id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assigned_by.id | Number | ID of the user who assigned the entity. |
| Vectra.Entity.Assignments.assigned_by.username | String | Username of the user who assigned the entity. |
| Vectra.Entity.Assignments.date_assigned | Date | Date when the entity was assigned. |
| Vectra.Entity.Assignments.date_resolved | Date | Date when the entity was resolved. |
| Vectra.Entity.Assignments.events.assignment_id | Number | ID of the assignment event. |
| Vectra.Entity.Assignments.events.actor | Number | ID of the actor who performed the assignment event. |
| Vectra.Entity.Assignments.events.event_type | String | Type of assignment event. |
| Vectra.Entity.Assignments.events.datetime | Date | Date of the assignment event. |
| Vectra.Entity.Assignments.events.context.to | Number | ID of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.from | Number | ID of the entity that was assigned. |
| Vectra.Entity.Assignments.events.context.entity_t_score | Number | Threat score of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.entity_c_score | Number | Certainty score of the entity that was assigned to. |
| Vectra.Entity.Assignments.outcome.id | String | ID of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.builtin | String | Whether the assignment outcome is builtin or not. |
| Vectra.Entity.Assignments.outcome.user_selectable | String | Whether the assignment outcome is user selectable or not. |
| Vectra.Entity.Assignments.outcome.title | String | Title of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.category | String | Category of the assignment outcome. |
| Vectra.Entity.Assignments.resolved_by.id | Number | ID of the user who resolved the entity. |
| Vectra.Entity.Assignments.resolved_by.username | String | Username of the user who resolved the entity. |
| Vectra.Entity.Assignments.triaged_detections | Unknown | Number of detections that have been triaged for the entity. |
| Vectra.Entity.Assignments.host_id | Number | ID of the host that the entity is associated with. |
| Vectra.Entity.Assignments.account_id | Unknown | ID of the account that the entity is associated with. |
| Vectra.Entity.Assignments.assigned_to.id | Number | ID of the user who is currently assigned to the entity. |
| Vectra.Entity.Assignments.assigned_to.username | String | Username of the user who is currently assigned to the entity. |
Command Example
!vectra-entity-assignment-update assignment_id=1 user_id=2
Context Example
{
{
"assigned_by": {
"id": 65,
"username": "api_client"
},
"date_assigned": "2023-07-21T12:44:10Z",
"events": [
{
"assignment_id": 1,
"actor": 65,
"event_type": "reassigned",
"datetime": "2023-07-25T06:26:10Z",
"context": {
"from": 1,
"to": 2,
"entity_t_score": 68,
"entity_c_score": 90
}
},
{
"assignment_id": 1,
"actor": 65,
"event_type": "created",
"datetime": "2023-07-21T12:44:10Z",
"context": {
"to": 1,
"entity_t_score": 68,
"entity_c_score": 90
}
}
],
"host_id": 97,
"assigned_to": {
"id": 2,
"username": "test_user_2"
},
"assignment_id": 1,
"id": 1
}
}
Human Readable Output
The assignment has been successfully updated
Assignment detail
Assignment ID Assigned By Assigned Date Assigned To Event Type 1 api_client 2023-07-21T12:44:10Z test_user_2 reassigned
vectra-detection-pcap-download
Download pcap of the detection.
Base Command
vectra-detection-pcap-download
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| File.Size | Number | The size of the file. |
| File.SHA1 | String | The SHA1 hash of the file. |
| File.SHA256 | String | The SHA256 hash of the file. |
| File.SHA512 | String | The SHA512 hash of the file. |
| File.Name | String | The name of the file. |
| File.SSDeep | String | The SSDeep hash of the file. |
| File.EntryID | String | The entry ID of the file. |
| File.Info | String | File information. |
| File.Type | String | The file type. |
| File.MD5 | String | The MD5 hash of the file. |
| File.Extension | String | The file extension. |
Command Example
!vectra-detection-pcap-download detection_id="116"
Context Example
{
"File": {
"EntryID": "1703@7e0f6637-f0a4-46b3-8c61-2f94b3432428",
"Extension": "pcap",
"Info": "pcap-ng capture file - version 1.0",
"MD5": "709db6e1f8f5054ca57caf43ba248ed6",
"Name": "IP-192.168.55.10_hidden_dns_tunnel_1382.pcap",
"SHA1": "49fe55c6aef85549261b46dd2e54f8d485306ee5",
"SHA256": "8615bde9332584b4fd4fe4dc2cc6fc4c75504f6d44667814456c089fd413aa4d",
"SHA512": "3fa29be0e20884c850b62d2a99aa09b24488289ba0bc9aff37ebe982c21d3a78fb26d9c9ac7fbf2a0839ba649dc0a845f30e7f13de3a0c6284c3c2ac54102143",
"SSDeep": "384:dN+Pm11R0XPmts64kZog9ZaikYngk+SnRxFyeyCEyuAOasucOcakca0/rHfcjOUI:dI+t25caEPjRSnmuNasxRana4DgOUDcX",
"Size": 23988,
"Type": "application/vnd.tcpdump.pcap"
}
}
Human Readable Output
Uploaded file: IP-192.168.55.10_hidden_dns_tunnel_1382.pcap
Property Type Size Info MD5 SHA1 SHA256 SHA512 SSDeep Value application/vnd.tcpdump.pcap 23,988 bytes pcap-ng capture file - version 1.0 709db6e1f8f5054ca57caf43ba248ed6 49fe55c6aef85549261b46dd2e54f8d485306ee5 8615bde9332584b4fd4fe4dc2cc6fc4c75504f6d44667814456c089fd413aa4d 3fa29be0e20884c850b62d2a99aa09b24488289ba0bc9aff37ebe982c21d3a78fb26d9c9ac7fbf2a0839ba649dc0a845f30e7f13de3a0c6284c3c2ac54102143 384:dN+Pm11R0XPmts64kZog9ZaikYngk+SnRxFyeyCEyuAOasucOcakca0/rHfcjOUI:dI+t25caEPjRSnmuNasxRana4DgOUDcX
vectra-assignment-list
Returns a list of all assignments.
Base Command
vectra-assignment-list
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_ids | Specify the IDs of the entities. Comma-separated values supported. | Optional |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Optional |
| resolved | Filter by resolved status. Possible values are: True, False. | Optional |
| assignees | Filter by user ids of the assignment. Comma-separated values supported. | Optional |
| resolution | Filter by outcome ids of the resolution. Comma-separated values supported. | Optional |
| created_after | Filter by created after the timestamp. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 May 2023, 01 Mar 2021 04:45:33, 2022-04-17T14:05:44Z. |
Optional |
| page | Enables the caller to specify a particular page of results. Default is 1. | Optional |
| page_size | Specify the desired page size for the request. Default is 50. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Assignments.id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assignment_id | Number | ID of the assignment. |
| Vectra.Entity.Assignments.assigned_by.id | Number | ID of the user who assigned the entity. |
| Vectra.Entity.Assignments.assigned_by.username | String | Username of the user who assigned the entity. |
| Vectra.Entity.Assignments.date_assigned | Date | Date when the entity was assigned. |
| Vectra.Entity.Assignments.date_resolved | Date | Date when the entity was resolved. |
| Vectra.Entity.Assignments.events.assignment_id | Number | ID of the assignment event. |
| Vectra.Entity.Assignments.events.actor | Number | ID of the actor who performed the assignment event. |
| Vectra.Entity.Assignments.events.event_type | String | Type of the assignment event. |
| Vectra.Entity.Assignments.events.datetime | Date | Date of the assignment event. |
| Vectra.Entity.Assignments.events.context.to | Number | ID of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.entity_t_score | Number | Threat score of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.entity_c_score | Number | Certainty score of the entity that was assigned to. |
| Vectra.Entity.Assignments.events.context.triage_as | String | Triage status of the entity. |
| Vectra.Entity.Assignments.events.context.triaged_detection_ids | Array | IDs of the detections that have been triaged for the entity. |
| Vectra.Entity.Assignments.events.context.fixed_detection_ids | Array | IDs of the detections that have been fixed. |
| Vectra.Entity.Assignments.events.context.created_rule_ids | Array | IDs of the rules that have been created for the entity. |
| Vectra.Entity.Assignments.outcome.id | Number | ID of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.builtin | Boolean | Whether the assignment outcome is builtin or not. |
| Vectra.Entity.Assignments.outcome.user_selectable | Boolean | Whether the assignment outcome is user selectable or not. |
| Vectra.Entity.Assignments.outcome.title | String | Title of the assignment outcome. |
| Vectra.Entity.Assignments.outcome.category | String | Category of the assignment outcome. |
| Vectra.Entity.Assignments.resolved_by.id | Number | ID of the user who resolved the entity. |
| Vectra.Entity.Assignments.resolved_by.username | String | Username of the user who resolved the entity. |
| Vectra.Entity.Assignments.triaged_detections | Array | Number of detections that have been triaged for the entity. |
| Vectra.Entity.Assignments.host_id | Number | ID of the host that the entity is associated with. |
| Vectra.Entity.Assignments.account_id | Number | ID of the account that the entity is associated with. |
| Vectra.Entity.Assignments.assigned_to.id | Number | ID of the user who is currently assigned to the entity. |
| Vectra.Entity.Assignments.assigned_to.username | String | Username of the user who is currently assigned to the entity. |
Command Example
#### Context Example
```json
{
"Vectra": {
"Entity": {
"Assignments": [
{
"id": 214,
"assigned_by": {
"id": 64,
"username": "test.user4@example.com"
},
"date_assigned": "2023-08-18T10:55:29Z",
"events": [
{
"assignment_id": 214,
"actor": 64,
"event_type": "reassigned",
"datetime": "2023-08-18T10:56:11Z",
"context": {
"from": 39,
"to": 59,
"entity_t_score": 0,
"entity_c_score": 0
}
},
{
"assignment_id": 214,
"actor": 64,
"event_type": "created",
"datetime": "2023-08-18T10:55:29Z",
"context": {
"to": 39,
"entity_t_score": 0,
"entity_c_score": 0
}
}
],
"host_id": 220,
"assigned_to": {
"id": 59,
"username": "test.user2@example.com"
},
"assignment_id": 214
},
{
"id": 212,
"assigned_by": {
"id": 65,
"username": "test.user4@example.com"
},
"date_assigned": "2023-08-18T06:29:56Z",
"date_resolved": "2023-08-18T06:32:09Z",
"events": [
{
"assignment_id": 212,
"actor": 65,
"event_type": "resolved",
"datetime": "2023-08-18T06:32:09Z",
"context": {
"entity_t_score": 77,
"entity_c_score": 53
}
},
{
"assignment_id": 212,
"actor": 65,
"event_type": "reassigned",
"datetime": "2023-08-18T06:31:02Z",
"context": {
"from": 59,
"to": 60,
"entity_t_score": 77,
"entity_c_score": 53
}
},
{
"assignment_id": 212,
"actor": 65,
"event_type": "created",
"datetime": "2023-08-18T06:29:56Z",
"context": {
"to": 59,
"entity_t_score": 77,
"entity_c_score": 53
}
}
],
"outcome": {
"id": 1,
"builtin": true,
"user_selectable": true,
"title": "Benign True Positive",
"category": "benign_true_positive"
},
"resolved_by": {
"id": 65,
"username": "test.user4@example.com"
},
"account_id": 108,
"assigned_to": {
"id": 60,
"username": "test.user1@example.com"
},
"assignment_id": 212
}
]
}
}
}
Human Readable Output
Assignments Table (Showing Page 1 out of 1)
Account ID Host ID Assignment ID Assigned By Assigned To Date Assigned Resolved By Date Resolved Outcome ID Outcome 220 214 test.user4@example.com test.user2@example.com 2023-08-18T10:55:29Z 108 212 test.user4@example.com test.user1@example.com 2023-08-18T06:29:56Z test.user4@example.com 2023-08-18T06:32:09Z 1 Benign True Positive
vectra-entity-note-list
Returns a list of notes for a specified entity.
Base Command
vectra-entity-note-list
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: host, account. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Notes.note_id | Number | ID of the note. |
| Vectra.Entity.Notes.id | Number | ID of the note. |
| Vectra.Entity.Notes.date_created | Date | Date when the note was created. |
| Vectra.Entity.Notes.date_modified | Unknown | Date when the note was last modified. |
| Vectra.Entity.Notes.created_by | String | User who created the note. |
| Vectra.Entity.Notes.modified_by | Unknown | User who last modified the note. |
| Vectra.Entity.Notes.note | String | Content of the note. |
| Vectra.Entity.Notes.entity_id | String | ID of the entity associated with the note. |
| Vectra.Entity.Notes.entity_type | String | Type of the entity associated with the note. |
Command Example
!vectra-entity-note-list entity_id="107" entity_type="account"
Context Example
{
"Vectra": {
"Entity": {
"Notes": [
{
"created_by": "test_user@example.com",
"date_created": "2023-08-25T07:09:08Z",
"entity_id": 107,
"entity_type": "account",
"id": 1070,
"modified_by": "test_user@example.com",
"note": "From XSOAR",
"note_id": 1070
},
{
"created_by": "test_user@example.com",
"date_created": "2023-08-25T07:08:58Z",
"entity_id": 107,
"entity_type": "account",
"id": 1069,
"modified_by": "test_user@example.com",
"note": "Test note",
"note_id": 1069
},
{
"created_by": "api_client",
"date_created": "2023-08-16T05:23:33Z",
"entity_id": 107,
"entity_type": "account",
"id": 922,
"note": "[Mirrored From XSOAR] XSOAR Incident ID: 14228\n\nNote: **bold**\n\n_Italic_\n\n+Underline+\n\n~~strikethrough~~\n\nAdded By: admin",
"note_id": 922
}
]
}
}
}
Human Readable Output
Entity Notes Table
Note ID Note Created By Created Date Modified By Modified Date 1070 From XSOAR test_user@example.com 2023-08-25T07:09:08Z test_user@example.com 2023-08-25T08:10:08Z 1069 Test note test_user@example.com 2023-08-25T07:08:58Z test_user@example.com 2023-08-25T08:10:08Z 922 [Mirrored From XSOAR] XSOAR Incident ID: 14228
Note:XSOAR note
Added By: adminapi_client 2023-08-16T05:23:33Z
vectra-group-list
Returns a list of all groups.
Base Command
vectra-group-list
Input
| Argument Name | Description | Required |
|---|---|---|
| group_type | Filter by group type. Possible values are: account, host, ip, domain. | Optional |
| account_names | Filter by Account Names. Supports comma-separated values. Note: Only valid when the group_type parameter is set to “account”. |
Optional |
| domains | Filter by Domains. Supports comma-separated values. Note: Only valid when the group_type parameter is set to “domain”. |
Optional |
| host_ids | Filter by Host IDs. Supports comma-separated values. Note: Only valid when the group_type parameter is set to “host”. |
Optional |
| host_names | Filter by Host Names. Supports comma-separated values. Note: Only valid when the group_type parameter is set to “host”. |
Optional |
| importance | Filter by group importance. Possible values are: high, medium, low, never_prioritize. | Optional |
| ips | Filter by IPs. Supports comma-separated values. Note: Only valid when the group_type parameter is set to “ip”. |
Optional |
| description | Filter by group description. | Optional |
| last_modified_timestamp | Return only the groups which have a last modification timestamp equal to or after the given timestamp. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 May 2023, 01 Mar 2023 04:45:33, 2023-04-17T14:05:44Z. |
Optional |
| last_modified_by | Filters by the user id who made the most recent modification to the group. | Optional |
| group_name | Filters by group name. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Group.group_id | Number | ID of the group. |
| Vectra.Group.id | Number | ID of the group. |
| Vectra.Group.name | String | Name of the group. |
| Vectra.Group.description | String | Description of the group. |
| Vectra.Group.last_modified | Date | Date when the group was last modified. |
| Vectra.Group.last_modified_by | String | Name of the user who last modified the group. |
| Vectra.Group.type | String | Type of the group. |
| Vectra.Group.members | Unknown | Members of the group. |
| Vectra.Group.members.id | Number | Entity ID of member. |
| Vectra.Group.members.name | String | Entity name of member. |
| Vectra.Group.members.is_key_asset | Boolean | Indicates key asset. |
| Vectra.Group.members.url | String | Entity URL of member. |
| Vectra.Group.members.uid | String | Entity UID of member. |
| Vectra.Group.rules.triage_category | String | Triage category of rule. |
| Vectra.Group.rules.id | Number | Id of the rule. |
| Vectra.Group.rules.description | String | Description of the rule. |
| Vectra.Group.importance | String | Importance level of the group. |
| Vectra.Group.cognito_managed | Boolean | Whether the group is managed by Cognito or not. |
Command Example
#### Context Example
```json
{
"Vectra": {
"Group": [
{
"id": 1,
"group_id": 1,
"name": "Cognito - Box",
"description": "Domains used by the Box service",
"last_modified": "2023-05-31T13:57:53Z",
"last_modified_by": "cognito",
"type": "domain",
"members": [
"*.abc.com",
"*.xyz.net"
],
"rules": [
{
"triage_category": "Box",
"id": 175,
"description": "data storage to Box service"
}
],
"importance": "medium",
"cognito_managed": true
},
{
"id": 8,
"group_id": 8,
"name": "Cognito - IPAM",
"description": "IPAM, created by Cognito",
"last_modified": "2023-08-18T09:16:54Z",
"last_modified_by": "cognito",
"type": "host",
"members": [
{
"is_key_asset": false,
"id": 97,
"name": "IP-0.0.0.0",
"url": "https://server_url.com/api/v3.3/hosts/97"
},
{
"is_key_asset": false,
"id": 212,
"name": "IP-0.0.0.1",
"url": "https://server_url.com/api/v3.3/hosts/212"
}
],
"rules": [
{
"triage_category": "Expected IPAM Behavior",
"id": 189,
"description": "Expected behavior from these devices"
},
{
"triage_category": "Expected IPAM Behavior",
"id": 193,
"description": "Expected behavior from these devices"
}
],
"importance": "medium"
},
{
"id": 16,
"group_id": 16,
"name": "Cognito - Guest Wifi",
"description": "IP space used by Guest Wifi",
"last_modified": "2023-08-18T08:55:54Z",
"last_modified_by": "cognito",
"type": "ip",
"members": [
"0.0.0.0",
"0.0.0.1"
],
"importance": "medium",
"cognito_managed": false
},
{
"id": 22,
"group_id": 22,
"name": "Dev-Group-Account-High",
"description": "",
"last_modified": "2023-08-25T10:17:37Z",
"last_modified_by": "cognito",
"type": "account",
"members": [
{
"uid": "O300:service-principal_00000000-0000-0000-0000-000000000001"
},
{
"uid": "administrator@fictotech.com"
}
],
"importance": "high"
}
]
}
}
Human Readable Output
Groups Table
Group ID Name Group Type Description Importance Members Last Modified Timestamp 1 Cognito - Box domain Domains used by the Box service medium *.abc.com, *.xyz.net 2023-05-31T13:57:53Z 8 Cognito - IPAM host IPAM, created by Cognito medium 97, 212 2023-08-18T09:16:54Z 16 Cognito - Guest Wifi ip IP space used by Guest Wifi medium 0.0.0.0, 0.0.0.1 2023-08-18T08:55:54Z 22 Dev-Group-Account-High account high O300:service-principal_00000000-0000-0000-0000-000000000001, administrator@fictotech.com 2023-08-25T10:17:37Z
vectra-group-unassign
Unassign members from the specified group.
Base Command
vectra-group-unassign
Input
| Argument Name | Description | Required |
|---|---|---|
| group_id | Specify Group ID to unassign members. | Required |
| members | Member values based on the group type. Supports comma-separated values. Note: If the group type is host, then the “Host IDs”. If the group type is account, then “Account Names”. If the group type is ip, then the list of “IPs”. If the group type is domain, then the list of “Domains” . |
Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Group.group_id | Number | ID of the group. |
| Vectra.Group.id | Number | ID of the group. |
| Vectra.Group.name | String | Name of the group. |
| Vectra.Group.description | String | Description of the group. |
| Vectra.Group.last_modified | Date | Date when the group was last modified. |
| Vectra.Group.last_modified_by | String | Name of the user who last modified the group. |
| Vectra.Group.type | String | Type of the group. |
| Vectra.Group.members | Unknown | Members of the group. |
| Vectra.Group.members.id | Number | Entity ID of member. |
| Vectra.Group.members.name | String | Entity name of member. |
| Vectra.Group.members.is_key_asset | Boolean | Indicates key asset. |
| Vectra.Group.members.url | String | Entity URL of member. |
| Vectra.Group.members.uid | String | Entity UID of member. |
| Vectra.Group.rules.triage_category | String | Triage category of rule. |
| Vectra.Group.rules.id | Number | Id of the rule. |
| Vectra.Group.rules.description | String | Description of the rule. |
Command Example
!vectra-group-unassign group_id=23 members="*.domain4.com,*.domain5.com"
Context Example
{
"Vectra": {
"Group": {
"cognito_managed": false,
"description": "xsoar-group-account-test",
"group_id": 23,
"id": 23,
"last_modified": "2023-09-04T12:03:02Z",
"last_modified_by": "API Client a7f5be37",
"members": ["*.domain1.net", "*.domain2.com", "*.domain3.com"],
"name": "xsoar-group-account-test",
"type": "domain"
}
}
}
Human Readable Output
Member(s) *.domain4.com, *.domain5.com have been unassigned from the group
Updated group details
Group ID Name Group Type Description Members Last Modified Timestamp 1 xsoar-group-account-test domain xsoar-group-account-test *.domain1.net, *.domain2.com, *.domain3.com 2023-09-04T07:30:01Z
vectra-group-assign
Assign members to the specified group.
Base Command
vectra-group-assign
Input
| Argument Name | Description | Required |
|---|---|---|
| group_id | Specify Group ID to assign members. | Required |
| members | Member values based on the group type. Supports comma-separated values. Note: If the group type is host, then the “Host IDs”. If the group type is account, then “Account Names”. If the group type is ip, then the list of “IPs”. If the group type is domain, then the list of “Domains” . |
Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Group.group_id | Number | ID of the group. |
| Vectra.Group.id | Number | ID of the group. |
| Vectra.Group.name | String | Name of the group. |
| Vectra.Group.description | String | Description of the group. |
| Vectra.Group.last_modified | Date | Date when the group was last modified. |
| Vectra.Group.last_modified_by | String | Name of the user who last modified the group. |
| Vectra.Group.type | String | Type of the group. |
| Vectra.Group.members | Unknown | Members of the group. |
| Vectra.Group.members.id | Number | Entity ID of member. |
| Vectra.Group.members.name | String | Entity name of member. |
| Vectra.Group.members.is_key_asset | Boolean | Indicates key asset. |
| Vectra.Group.members.url | String | Entity URL of member. |
| Vectra.Group.members.uid | String | Entity UID of member. |
| Vectra.Group.rules.triage_category | String | Triage category of rule. |
| Vectra.Group.rules.id | Number | Id of the rule. |
| Vectra.Group.rules.description | String | Description of the rule. |
Command Example
!vectra-group-assign group_id=23 members="*.domain4.com,*.domain5.com"
Context Example
{
"Vectra": {
"Group": {
"cognito_managed": false,
"description": "xsoar-group-account-test",
"group_id": 23,
"id": 23,
"last_modified": "2023-09-04T11:59:15Z",
"last_modified_by": "API Client a7f5be37",
"members": [
"*.domain1.net",
"*.domain2.com",
"*.domain3.com",
"*.domain4.com",
"*.domain5.com"
],
"name": "xsoar-group-account-test",
"type": "domain"
}
}
}
Human Readable Output
Member(s) *.domain4.com, *.domain5.com have been assigned to the group
Updated group details
Group ID Name Group Type Description Members Last Modified Timestamp 1 xsoar-group-account-test domain xsoar-group-account-test *.domain1.net, *.domain2.com, *.domain3.com, *.domain4.com, *.domain5.com 2023-09-04T06:30:01Z
vectra-entity-detections-mark-asclosed
Mark the detections of the entity as closed with the provided entity ID in the argument.
Base Command
vectra-entity-detections-mark-asclosed
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. | Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| close_reason | Specify the close reason. Possible values are: benign, remediated. | Required |
Context Output
There is no context output for this command.
Command example
!vectra-entity-detections-mark-asclosed entity_id=1 entity_type=account close_reason=benign
Human Readable Output
The detections (34122, 35097) of the provided entity ID have been successfully closed as benign
vectra-detections-mark-asopen
Open detections with provided detection IDs in the argument.
Base Command
vectra-detections-mark-asopen
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_ids | Provide a list of detection IDs separated by commas or a single detection ID. | Required |
Context Output
There is no context output for this command.
Command example
!vectra-detections-mark-asopen detection_ids=1,2,3
Human Readable Output
The provided detection IDs have been successfully re-opened
vectra-detection-tag-list
Returns a list of tags for a specified detection.
Base Command
vectra-detection-tag-list
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Tags.tag_id | String | The ID of the tag. |
| Vectra.Detection.Tags.detection_id | String | The ID of the Detection associated with the tag. |
| Vectra.Detection.Tags.tags | Unknown | A list of tags linked to a detection. |
Command example
!vectra-detection-tag-list detection_id=123
Context Example
{
"Vectra": {
"Detection": {
"Tags": {
"detection_id": 123,
"tag_id": "123",
"tags": [
"tag1",
"tag2"
]
}
}
}
}
Human Readable Output
List of tags: tag1, tag2
vectra-detection-tag-add
Add tags to a detection.
Base Command
vectra-detection-tag-add
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
| tags | Comma-separated values of tags to be added to the detection. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Tags.tag_id | String | The ID of the tag. |
| Vectra.Detection.Tags.detection_id | String | The ID of the detection associated with the tag. |
| Vectra.Detection.Tags.tags | Unknown | A list of tags linked to a detection. |
Command example
!vectra-detection-tag-add detection_id=1 tags="tag1,tag2"
Context Example
{
"Vectra": {
"Detection": {
"Tags": {
"detection_id": 1,
"tag_id": 1,
"tags": [
"tag",
"tag1",
"tag2"
]
}
}
}
}
Human Readable Output
Tags have been successfully added to the detection
Updated list of tags: tag, tag1, tag2
vectra-detection-tag-remove
Remove tags from the detection.
Base Command
vectra-detection-tag-remove
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
| tags | Comma-separated values of tags to be removed from the detection. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Tags.tag_id | String | The ID of the tag. |
| Vectra.Detection.Tags.detection_id | String | The ID of the detection associated with the tag. |
| Vectra.Detection.Tags.tags | Unknown | A list of tags linked to a detection. |
Command example
!vectra-detection-tag-remove detection_id="2" tags="tag3,tag4"
Context Example
{
"Vectra": {
"Detection": {
"Tags": {
"detection_id": 2,
"tag_id": "2",
"tags": [
"tag",
"tag1",
"tag2"
]
}
}
}
}
Human Readable Output
Specified tags have been successfully removed for the detection
Updated list of tags: tag, tag1, tag2
vectra-detection-note-list
Returns a list of notes for a specified detection.
Base Command
vectra-detection-note-list
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Notes.note_id | Number | ID of the note. |
| Vectra.Detection.Notes.id | Number | ID of the note. |
| Vectra.Detection.Notes.date_created | Date | Date when the note was created (ISO8601). |
| Vectra.Detection.Notes.date_modified | Date | Date when the note was last modified (ISO8601). |
| Vectra.Detection.Notes.created_by | String | User who created the note. |
| Vectra.Detection.Notes.modified_by | String | User who last modified the note. |
| Vectra.Detection.Notes.note | String | Content of the note. |
| Vectra.Detection.Notes.detection_id | String | ID of the detection associated with the note. |
Command example
!vectra-detection-note-list detection_id=1
Context Example
{
"Vectra": {
"Detection": {
"Notes": [
{
"created_by": "test_user@example.com",
"date_created": "2023-08-25T07:09:08Z",
"detection_id": 1,
"id": 1070,
"modified_by": "test_user@example.com",
"note": "From XSOAR",
"note_id": 1070
},
{
"created_by": "test_user@example.com",
"date_created": "2023-08-25T07:08:58Z",
"detection_id": 1,
"id": 1069,
"modified_by": "test_user@example.com",
"note": "Test note",
"note_id": 1069
},
{
"created_by": "api_client",
"date_created": "2023-08-16T05:23:33Z",
"detection_id": 1,
"id": 922,
"note": "[Mirrored From XSOAR] XSOAR Incident ID: 14228\n\nNote: **bold**\n\n_Italic_\n\n+Underline+\n\n~~strikethrough~~\n\nAdded By: admin",
"note_id": 922
}
]
}
}
}
Human Readable Output
Detection Notes Table
Note ID Note Created By Created Date Modified By Modified Date 1070 From XSOAR test_user@example.com 2023-08-25T07:08:58Z test_user@example.com 2023-08-25T07:08:58Z 1069 Test note test_user@example.com 2023-08-25T07:08:58Z test_user@example.com 2023-08-25T07:08:58Z 922 [Mirrored From XSOAR] XSOAR Incident ID: 14228\n\nNote: bold\n\n_Italic_\n\n+Underline+\n\nstrikethrough\n\nAdded By: admin api_client 2023-08-16T05:23:33Z
vectra-detection-note-add
Add a note to the detection.
Base Command
vectra-detection-note-add
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
| note | Note to be added in the specified detection_id. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Notes.detection_id | String | ID of the detection associated with the note. |
| Vectra.Detection.Notes.note_id | Number | ID of the note. |
| Vectra.Detection.Notes.id | Number | ID of the note. |
| Vectra.Detection.Notes.date_created | Date | Date when the note was created (ISO8601). |
| Vectra.Detection.Notes.created_by | String | User who created the note. |
| Vectra.Detection.Notes.note | String | Content of the note. |
Command example
!vectra-detection-note-add detection_id=1 note="test note"
Context Example
{
{
"date_created": "2023-06-21T06:19:15.224449Z",
"created_by": "test_user",
"note": "test note",
"note_id": 19,
"id": 19,
"detection_id": 1
}
}
Human Readable Output
The note has been successfully added to the detection
Returned Note ID: 19
vectra-detection-note-update
Update a note in the detection.
Base Command
vectra-detection-note-update
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
| note_id | Specify the ID of the note. | Required |
| note | Note to be updated for the specified note_id. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.Notes.detection_id | String | ID of the detection associated with the note. |
| Vectra.Detection.Notes.note_id | Number | ID of the note. |
| Vectra.Detection.Notes.id | Number | ID of the note. |
| Vectra.Detection.Notes.date_created | Date | Date when the note was created (ISO8601). |
| Vectra.Detection.Notes.date_modified | Date | Date when the note was last modified (ISO8601). |
| Vectra.Detection.Notes.created_by | String | User who created the note. |
| Vectra.Detection.Notes.modified_by | String | User who last modified the note. |
| Vectra.Detection.Notes.note | String | Content of the note. |
Command example
!vectra-detection-note-update detection_id=1 note_id=1 note="note modified"
Context Example
{
{
"date_created": "2023-06-16T04:55:58Z",
"date_modified": "2023-06-22T04:57:09Z",
"created_by": "test_user",
"modified_by": "test_user",
"note": "note modified",
"note_id": 8,
"id": 8,
"detection_id": 1
}
}
Human Readable Output
The note has been successfully updated in the detection
vectra-detection-note-remove
Remove a note from the detection.
Base Command
vectra-detection-note-remove
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_id | Specify the ID of the detection. | Required |
| note_id | Specify the ID of the note. | Required |
Context Output
There is no context output for this command.
Command Example
!vectra-detection-note-remove detection_id=1 note_id=1
Context Example
{}
Human Readable Output
The note has been successfully removed from the detection
vectra-entity-unresolved-priority-reset
Update the unresolved priority of an entity to false.
Base Command
vectra-entity-unresolved-priority-reset
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. Note: Users can get the entity ID by executing the “vectra-entity-list” command. |
Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.id | String | An ID of the entity. |
| Vectra.Entity.type | String | The type of the entity. |
| Vectra.Entity.unresolved_priority | Boolean | An entity unresolved priority status. |
Command Example
!vectra-entity-unresolved-priority-reset entity_id=1 entity_type=account
Context Example
{
"Vectra": {
"Entity": [
{
"id": "1",
"type": "account",
"unresolved_priority": false
}
]
}
}
Human Readable Output
The unresolved priority of the provided entity has been successfully changed as ‘false’
vectra-detection-investigation-status-update
Update the investigation status of the detection by detection ID(s).
Base Command
vectra-detection-investigation-status-update
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_ids | Provide a list of detection IDs separated by comma or a single detection ID. Note: Users can get the detection ID by executing the “vectra-detection-list” command. |
Required |
| investigation_status | Specify the investigation status. Possible values are: open, acknowledged, escalated, paused, closed, expired. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.id | String | The detection ID. |
| Vectra.Detection.investigation_status | String | The detection investigation status. |
Command Example
!vectra-detection-investigation-status-update detection_ids=1 investigation_status=escalated
Context Example
{
"Vectra": {
"Detection": [
{
"id": "1",
"investigation_status": "escalated"
}
]
}
}
Human Readable Output
The investigation Status for provided Detection ID(s) [‘1’] have been updated as escalated
vectra-detection-external-id-update
Update the external reference ID for the provided detection ID(s).
Base Command
vectra-detection-external-id-update
Input
| Argument Name | Description | Required |
|---|---|---|
| detection_ids | Provide a list of detection IDs separated by comma or a single detection ID. Note: Users can get the detection ID by executing the “vectra-detection-list” command. |
Required |
| external_reference_id | Provide the external reference ID. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.id | String | The detection ID. |
| Vectra.Detection.external_reference_id | String | The external reference ID of the detection. |
Command Example
!vectra-detection-external-id-update detection_ids=1 external_reference_id=12345
Context Example
{
"Vectra": {
"Detection": [
{
"id": "1",
"external_reference_id": "12345"
}
]
}
}
Human Readable Output
The external reference ID for provided Detection ID(s) [‘1’] have been updated as 12345
vectra-entity-external-id-update
Update the external reference ID for the provided entity.
Base Command
vectra-entity-external-id-update
Input
| Argument Name | Description | Required |
|---|---|---|
| entity_id | Specify the ID of the entity. Note: Users can get the entity ID by executing the “vectra-entity-list” command. |
Required |
| entity_type | Specify the type of the entity. Possible values are: account, host. | Required |
| external_reference_id | Provide the external reference ID. | Required |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.id | String | An ID of the entity. |
| Vectra.Entity.type | String | The type of the entity. |
| Vectra.Entity.external_reference_id | String | The external reference ID of the entity. |
Command Example
!vectra-entity-external-id-update entity_id=1 entity_type=account external_reference_id=12345
Context Example
{
"Vectra": {
"Entity": [
{
"id": "1",
"type": "account",
"external_reference_id": "12345"
}
]
}
}
Human Readable Output
The external reference ID for provided Entity have been updated as 12345
vectra-detection-list
Returns a list of detections based on the specified filters.
Base Command
vectra-detection-list
Input
| Argument Name | Description | Required |
|---|---|---|
| created_after | Filter the detections by created on or after the specified time. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 March 2026, 01 Mar 2026 04:45:33, 2026-04-17T14:05:44Z. |
Optional |
| created_before | Filter the detections by created on or before the specified time. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 March 2026, 01 Mar 2026 04:45:33, 2026-04-17T14:05:44Z. |
Optional |
| last_detected_after | Filter the detections by last detected on or after the specified time. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 March 2026, 01 Mar 2026 04:45:33, 2026-04-17T14:05:44Z. |
Optional |
| last_detected_before | Filter the detections by last detected on or before the specified time. Supported formats: 2 minutes, 2 hours, 2 days, 2 weeks, 2 months, 2 years, yyyy-mm-dd, yyyy-mm-ddTHH:MM:SSZ. For example: 01 March 2026, 01 Mar 2026 04:45:33, 2026-04-17T14:05:44Z. |
Optional |
| description | Filter by description containing specified value. | Optional |
| detection_name | Filter by detection name. | Optional |
| detection_type | Filter by detection type. | Optional |
| detection_category | Filter by detections category. Possible values are: Command & Control, Botnet, Reconnaissance, Lateral Movement, Exfiltration, Info. | Optional |
| include_info_category_detections | Include the info category detections which are excluded by default. Possible values are: true, false. Default is true. | Optional |
| close_reason | Filter by close reason of the detection. Possible values are: benign, remediated. | Optional |
| detection_state | Filter by detection state. Possible values are: active, inactive, fixed. | Optional |
| entity_type | Filter by Entity type. Possible values are: account, host. | Optional |
| tags | Filter by detection tags. Comma-separated values supported. | Optional |
| is_triaged | Filter by detection triage status. Possible values are: true, false. Default is false. | Optional |
| page | Provide page number to retrieve. Default is 1. | Optional |
| page_size | Provide a number of results per page. Default is 50. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Detection.id | Number | A unique identifier for the detection. |
| Vectra.Detection.assigned_date | Date | The date when the detection was assigned. |
| Vectra.Detection.assigned_to | String | The email or user to whom the detection is assigned. |
| Vectra.Detection.certainty | Number | The certainty level associated with the detection. |
| Vectra.Detection.created_timestamp | Date | The timestamp when the detection was created. |
| Vectra.Detection.custom_detection | Unknown | The custom detection configuration or settings. |
| Vectra.Detection.data_source.type | String | The type of data source for the detection. |
| Vectra.Detection.data_source.connection_name | String | The name of the connection used for data ingestion. |
| Vectra.Detection.data_source.connection_id | String | A unique identifier for the data source connection. |
| Vectra.Detection.description | String | The description of the detection. |
| Vectra.Detection.detection | String | The name of the detection. |
| Vectra.Detection.detection_category | String | The category of the detection. |
| Vectra.Detection.detection_type | String | The type of the detection. |
| Vectra.Detection.detection_url | String | The URL to access the detection details. |
| Vectra.Detection.filtered_by_ai | Boolean | Indicates if the detection was filtered by AI. |
| Vectra.Detection.filtered_by_rule | Boolean | Indicates if the detection was filtered by a rule. |
| Vectra.Detection.filtered_by_user | Boolean | Indicates if the detection was filtered by a user. |
| Vectra.Detection.first_timestamp | Date | The first timestamp when the detection was observed. |
| Vectra.Detection.grouped_details.role | String | The role associated with the detection group. |
| Vectra.Detection.grouped_details.last_timestamp | Date | The last timestamp of the detection group. |
| Vectra.Detection.groups.id | Number | A unique identifier for the group. |
| Vectra.Detection.groups.name | String | The name of the group. |
| Vectra.Detection.groups.description | String | The description of the group. |
| Vectra.Detection.groups.type | String | The type of the group. |
| Vectra.Detection.groups.last_modified | Date | The timestamp when the group was last modified. |
| Vectra.Detection.groups.last_modified_by | String | The email or user who last modified the group. |
| Vectra.Detection.is_custom_model | Boolean | Indicates if the detection uses a custom model. |
| Vectra.Detection.is_marked_custom | Boolean | Indicates if the detection is marked as custom. |
| Vectra.Detection.is_triaged | Boolean | Indicates if the detection has been triaged. |
| Vectra.Detection.last_timestamp | Date | The last timestamp when the detection was observed. |
| Vectra.Detection.note | String | A note associated with the detection. |
| Vectra.Detection.note_modified_by | String | The email or user who modified the note. |
| Vectra.Detection.note_modified_timestamp | Date | The timestamp when the note was last modified. |
| Vectra.Detection.notes.created_by | String | The email or user who created the note. |
| Vectra.Detection.notes.date_created | Date | The date when the note was created. |
| Vectra.Detection.notes.date_modified | Date | The date when the note was modified. |
| Vectra.Detection.notes.id | Number | A unique identifier for the note. |
| Vectra.Detection.notes.modified_by | String | The email or user who modified the note. |
| Vectra.Detection.notes.note | String | The content of the note. |
| Vectra.Detection.reason | String | The reason for the detection state or triage action. |
| Vectra.Detection.sensor | String | The sensor identifier that detected the activity. |
| Vectra.Detection.sensor_name | String | The name of the sensor that detected the activity. |
| Vectra.Detection.src_account.id | Number | A unique identifier for the source account. |
| Vectra.Detection.src_account.name | String | The name of the source account. |
| Vectra.Detection.src_account.url | String | The URL to access the source account details. |
| Vectra.Detection.src_account.threat | Number | The threat level associated with the source account. |
| Vectra.Detection.src_account.certainty | Number | The certainty level associated with the source account. |
| Vectra.Detection.src_account.privilege_level | Number | The privilege level associated with the source account. |
| Vectra.Detection.src_account.privilege_category | String | The privilege category associated with the source account. |
| Vectra.Detection.src_host.id | Number | A unique identifier for the source host. |
| Vectra.Detection.src_host.name | String | The name of the source host. |
| Vectra.Detection.src_host.ip | String | The IP address of the source host. |
| Vectra.Detection.src_host.url | String | The URL to access the source host details. |
| Vectra.Detection.src_host.is_key_asset | Boolean | Indicates if the source host is a key asset. |
| Vectra.Detection.src_host.group.id | Number | A unique identifier for the source host group. |
| Vectra.Detection.src_host.group.name | String | The name of the source host group. |
| Vectra.Detection.src_host.group.description | String | The description of the source host group. |
| Vectra.Detection.src_host.group.type | String | The type of the source host group. |
| Vectra.Detection.src_host.group.last_modified | Date | The timestamp when the source host group was last modified. |
| Vectra.Detection.src_host.group.last_modified_by | String | The email or user who last modified the source host group. |
| Vectra.Detection.src_host.threat | Number | The threat level associated with the source host. |
| Vectra.Detection.src_host.certainty | Number | The certainty level associated with the source host. |
| Vectra.Detection.src_ip | String | The source IP address in the detection. |
| Vectra.Detection.src_groups.id | Number | A unique identifier for the source group. |
| Vectra.Detection.src_groups.name | String | The name of the source group. |
| Vectra.Detection.src_groups.description | String | The description of the source group. |
| Vectra.Detection.src_groups.type | String | The type of the source group. |
| Vectra.Detection.src_groups.last_modified | Date | The timestamp when the source group was last modified. |
| Vectra.Detection.src_groups.last_modified_by | String | The email or user who last modified the source group. |
| Vectra.Detection.dst_groups.id | Number | A unique identifier for the destination group. |
| Vectra.Detection.dst_groups.name | String | The name of the destination group. |
| Vectra.Detection.dst_groups.description | String | The description of the destination group. |
| Vectra.Detection.dst_groups.type | String | The type of the destination group. |
| Vectra.Detection.dst_groups.last_modified | Date | The timestamp when the destination group was last modified. |
| Vectra.Detection.dst_groups.last_modified_by | String | The email or user who last modified the destination group. |
| Vectra.Detection.state | String | The current state of the detection. |
| Vectra.Detection.summary.artifact | Array | The artifacts associated with the detection summary. |
| Vectra.Detection.summary.last_timestamp | Date | The last timestamp in the detection summary. |
| Vectra.Detection.summary.description | String | The description in the detection summary. |
| Vectra.Detection.summary.roles | Array | The roles associated with the detection summary. |
| Vectra.Detection.tags | Array | The tags associated with the detection. |
| Vectra.Detection.is_targeting_key_asset | Boolean | Indicates if the detection is targeting a key asset. |
| Vectra.Detection.threat | Number | The threat level of the detection. |
| Vectra.Detection.triage_rule_id | Unknown | A unique identifier for the triage rule applied to the detection. |
| Vectra.Detection.type | String | The type of the detection. |
| Vectra.Detection.url | String | The URL to access the detection details. |
Command Example
!vectra-detection-list page=1 page_size=2
Context Example
{
"Vectra": {
"Detection": [
{
"summary": {
"app_name": "Exchange",
"operations": [
"Add-MailboxPermission"
],
"src_ips": [
"10.0.0.1"
],
"description": "This account performed Exchange operations that were unusual for the account."
},
"src_account": {
"id": 1001,
"name": "user@example.com",
"url": "https://example.vectra.ai/api/v3.5/accounts/1001",
"threat": 45,
"certainty": 60
},
"state": "active",
"created_timestamp": "2026-01-15T10:30:00Z",
"filtered_by_user": false,
"type": "account",
"detection_type": "M365 Risky Exchange Operation",
"data_source": {
"type": "o365",
"connection_name": "M365-Production",
"connection_id": "abc123"
},
"filtered_by_rule": false,
"detection": "M365 Risky Exchange Operation",
"url": "https://example.vectra.ai/api/v3.5/detections/5001",
"sensor": "abc123",
"threat": 50,
"is_custom_model": false,
"is_triaged": false,
"detection_category": "lateral_movement",
"filtered_by_ai": false,
"detection_url": "https://example.vectra.ai/api/v3.5/detections/5001",
"last_timestamp": "2026-01-15T12:00:00Z",
"first_timestamp": "2026-01-15T10:00:00Z",
"certainty": 50,
"is_marked_custom": false,
"id": 5001,
"sensor_name": "Vectra NDR",
"is_targeting_key_asset": false,
"grouped_details": [
{
"parameters": [
{
"data": [
{
"name": "Identity",
"value": "mailbox@example.com"
}
],
"timestamp": "2026-01-15T11:30:00Z"
}
],
"operation": "Add-MailboxPermission",
"behavior": "Mailbox management",
"user_type": "Admin",
"last_timestamp": "2026-01-15T12:00:00Z",
"src_ip": "10.0.0.1",
"app_name": "Exchange"
}
]
},
{
"state": "active",
"created_timestamp": "2026-01-15T09:00:00Z",
"filtered_by_user": false,
"type": "host",
"detection_type": "Suspicious Domain",
"groups": [
{
"id": 10,
"name": "Production Servers",
"description": "Production server subnet",
"type": "ip",
"last_modified": "2026-01-10T08:00:00Z",
"last_modified_by": "admin@example.com"
}
],
"data_source": {
"type": "sensor",
"connection_name": "Network Sensor 1",
"connection_id": "xyz789"
},
"filtered_by_rule": false,
"detection": "Suspicious Domain",
"url": "https://example.vectra.ai/api/v3.5/detections/5002",
"sensor": "xyz789",
"threat": 30,
"is_custom_model": false,
"is_triaged": false,
"detection_category": "command_and_control",
"filtered_by_ai": false,
"detection_url": "https://example.vectra.ai/api/v3.5/detections/5002",
"src_ip": "10.0.1.50",
"last_timestamp": "2026-01-15T09:45:00Z",
"first_timestamp": "2026-01-15T09:00:00Z",
"src_host": {
"id": 2001,
"ip": "10.0.1.50",
"name": "workstation-01",
"url": "https://example.vectra.ai/api/v3.5/hosts/2001",
"is_key_asset": false,
"groups": [
{
"id": 10,
"name": "Production Servers",
"description": "Production server subnet",
"last_modified": "2026-01-10T08:00:00Z",
"last_modified_by": "admin@example.com",
"type": "ip"
}
],
"threat": 35,
"certainty": 40
},
"certainty": 25,
"is_marked_custom": false,
"id": 5002,
"sensor_name": "Network Sensor 1",
"is_targeting_key_asset": false,
"grouped_details": [
{
"protocol": "dns",
"last_timestamp": "2026-01-15T09:45:00Z",
"grouping_field": "last_timestamp",
"response_code": "NXDomain",
"target_domains": [
"suspicious-domain.example"
],
"dst_ips": [
"8.8.8.8"
]
}
],
"summary": {
"num_failures": 5,
"num_successes": 0,
"num_sessions": 10
}
}
]
}
}
Human Readable Output
Detections Table (Showing Page 1 out of 100)
ID Detection Name Detection Type Account Name Host Name Src IP Threat Score Certainty Score Number Of Events State Last Timestamp 5001 M365 Risky Exchange Operation M365 Risky Exchange Operation user@example.com 50 50 0 active 2026-01-15T12:00:00Z 5002 Suspicious Domain Suspicious Domain workstation-01 10.0.1.50 30 25 0 active 2026-01-15T09:45:00Z
vectra-investigation-query-send
Submit an investigation query and receive a request ID for retrieving results.
Base Command
vectra-investigation-query-send
Input
| Argument Name | Description | Required |
|---|---|---|
| query | Provide an investigation query in the supported query language. | Required |
| version | Specify the version of the query language. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Investigation.request_id | String | The unique identifier for the query request. Use this to retrieve results. |
| Vectra.Investigation.searchable_range.searchable_days_allowed | Number | A Maximum number of days of data that can be searched. |
Command Example
!vectra-investigation-query-send query="SELECT * FROM detections" version=v1
Context Example
{
"Vectra": {
"Investigation": {
"request_id": "b57d7a27-28ad-4c0c-b28a-0e7b3",
"searchable_range": {
"searchable_days_allowed": 14
}
}
}
}
Human Readable Output
The Vectra investigation has started. You can view the results by executing the below command
!vectra-investigation-result-get id=b57d7a27-28ad-4c0c-b28a-0e7b3
vectra-investigation-result-get
Retrieve the results of a previously submitted investigation query using the request ID.
Base Command
vectra-investigation-result-get
Input
| Argument Name | Description | Required |
|---|---|---|
| id | Provide the unique request ID of investigation. | Required |
| page | Provide page number to retrieve. Default is 1. | Optional |
| page_size | Provide a number of results per page to retrieve. Default is 50. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Investigation.request_id | String | The unique identifier for the query request. |
| Vectra.Investigation.data | Unknown | An array of query results. |
| Vectra.Investigation.meta.query_status | String | The status of the query. |
| Vectra.Investigation.meta.num_rows_available | Number | Total rows returned by the query. |
| Vectra.Investigation.meta.page | Number | The current page number. |
| Vectra.Investigation.meta.page_size | Number | The rows returned on this page. |
| Vectra.Investigation.meta.estimated_file_size_bytes | Number | The estimated size of the full result set in bytes. |
| Vectra.Investigation.meta.columns | Unknown | An array of tuples describing the result schema. |
Command Example
!vectra-investigation-result-get id=b57d7a27-28ad-4c0c-b28a-0e7b3
Context Example
{
"Vectra": {
"Investigation": {
"request_id": "b57d7a27-28ad-4c0c-b28a-0e7b3",
"meta": {
"page": 1,
"page_size": 50,
"estimated_file_size_bytes": 0,
"num_rows_available": 0,
"query_status": "SUCCESS",
"columns": [
[
"timestamp",
[
{
"type": "timestamp"
},
""
]
],
[
"orig_h",
[
{
"type": "string"
},
""
]
],
[
"resp_h",
[
{
"type": "string"
},
""
]
],
[
"resp_p",
[
{
"type": "number"
},
""
]
]
]
}
}
}
}
Human Readable Output
Investigation Result for Request ID: b57d7a27-28ad-4c0c-b28a-0e7b3
Query Status Page Number Page size Total Rows File Size (bytes) Columns SUCCESS 1 50 0 0 - values: timestamp, [{‘type’: ‘timestamp’}, ‘’]
- values: orig_h, [{‘type’: ‘string’}, ‘’]
- values: resp_h, [{‘type’: ‘string’}, ‘’]
- values: resp_p, [{‘type’: ‘number’}, ‘’]
Investigation Results Data
No entries.
Configuration parameters
server_url— Server URL (required)credentials— Client ID (required)isFetch— Fetch incidentsmax_fetch— Max Fetchfirst_fetch— First Fetch Timeentity_types— Entity Typesonly_prioritized_detections— Create Incidents for Prioritized Detectionsonly_escalated_detections— Create Incidents for Escalated Detectionsmirror_direction— Mirroring Directionnote_tag— Mirror tag for notesopen_detection_on_incident_reopen— Open Detection on Incident Reopendetection_status_for_reopen— Detection Status for Incident Reopenclose_detection_on_incident_closure— Close Detection on Incident Closureclose_reason_of_detection— Detection Close Reason for Incident ClosureincidentType— Incident typeincidentFetchInterval— Incidents Fetch Intervalinsecure— Trust any certificate (not secure)proxy— Use system proxy settings
Commands (36)
-
vectra-assignment-listReturns a list of all assignments.
-
vectra-detection-describeReturns a list of detections for the specified detection ID(s).
-
vectra-detection-external-id-updateUpdate the external reference ID for the provided detection ID(s).
-
vectra-detection-investigation-status-updateUpdate the investigation status of the detection by detection ID(s).
-
vectra-detection-listReturns a list of detections based on the specified filters.
-
vectra-detection-note-addAdd a note to the detection.
-
vectra-detection-note-listReturns a list of notes for a specified detection.
-
vectra-detection-note-removeRemove a note from the detection.
-
vectra-detection-note-updateUpdate a note in the detection.
-
vectra-detection-pcap-downloadDownload pcap of the detection.
-
vectra-detection-tag-addAdd tags to a detection.
-
vectra-detection-tag-listReturns a list of tags for a specified detection.
-
vectra-detection-tag-removeRemove tags from the detection.
-
vectra-detections-mark-asclosedMark detections as closed with provided detection IDs in the argument.
-
vectra-detections-mark-asopenOpen detections with provided detection IDs in the argument.
-
vectra-entity-assignment-addAdd an assignment for the entity.
-
vectra-entity-assignment-updateUpdate an assignment in the entity.
-
vectra-entity-describeDescribes an entity by ID.
-
vectra-entity-detection-listReturns a list of detections for a specified entity.
-
vectra-entity-detections-mark-asclosedMark the detections of the entity as closed with the provided entity ID in the argument.
-
vectra-entity-external-id-updateUpdate the external reference ID for the provided entity.
-
vectra-entity-listReturns a list of entities.
-
vectra-entity-note-addAdd a note to the entity.
-
vectra-entity-note-listReturns a list of notes for a specified entity.
-
vectra-entity-note-removeRemove a note from the entity.
-
vectra-entity-note-updateUpdate a note in the entity.
-
vectra-entity-tag-addAdd tags in the entity.
-
vectra-entity-tag-listReturns a list of tags for a specified entity.
-
vectra-entity-tag-removeRemove tags from the entity.
-
vectra-entity-unresolved-priority-resetUpdate the unresolved priority of an entity to false.
-
vectra-group-assignAssign members to the specified group.
-
vectra-group-listReturns a list of all groups.
-
vectra-group-unassignUnassign members from the specified group.
-
vectra-investigation-query-sendSubmit an investigation query and receive a request ID for retrieving results.
-
vectra-investigation-result-getRetrieve the results of a previously submitted investigation query using the request ID.
-
vectra-user-listReturns a list of users.
import copy import urllib3 from collections.abc import Callable from datetime import datetime from requests.models import Response from typing import Any import demistomock as demisto from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # ISO8601 format with UTC, default in XSOAR STATUS_LIST_TO_RETRY = (429, *(status_code for status_code in requests.status_codes._codes if status_code >= 500)) # type: ignore OK_CODES = (200, 201, 204, 401) MAX_RETRIES = 4 BACKOFF_FACTOR = 7.5 FIRST_FETCH = "1 hour" MAX_FETCH = 200 PACK_VERSION = get_pack_version() or "1.0.0" USER_AGENT = f"Vectra-RUX-XSOAR-{PACK_VERSION}" UTM_PIVOT = f"?pivot=Vectra-RUX-XSOAR-{PACK_VERSION}" DEFAULT_ONLY_PRIORITIZED_DETECTIONS = False DEFAULT_ENTITY_TYPES = "Host,Account" VALID_ENTITY_TYPES = ("Host", "Account") VALID_DETECTION_STATUS = ("open", "acknowledged", "escalated", "paused", "closed", "expired") DEFAULT_FETCH_DETECTION_STATUS = ("open", "acknowledged", "escalated", "paused") VALID_CLOSE_REASON = ("benign", "remediated") MIRROR_DIRECTION = { "Incoming": "In", "Outgoing": "Out", "Incoming And Outgoing": "Both", } MAX_PAGE = 1 MAX_PAGE_SIZE = 50 ENTITY_AND_DETECTION_MAX_PAGE_SIZE = 5000 MAX_URGENCY_SCORE = 100 MIN_URGENCY_SCORE = 0 VALID_ENTITY_TYPE = ["account", "host"] VALID_GROUP_TYPE = ["account", "host", "ip", "domain"] VALID_IMPORTANCE_VALUE = ["high", "medium", "low", "never_prioritize"] VALID_ENTITY_STATE = ["active", "inactive"] VALID_BOOL_VALUES = ("y", "yes", "t", "true", "on", "1", "n", "no", "f", "false", "off", "0") MAX_MIRRORING_LIMIT = 5000 MAX_OUTGOING_NOTE_LIMIT = 8000 DETECTION_CATEGORY_TO_ARG = { "Command & Control": "command", "Botnet": "botnet", "Reconnaissance": "reconnaissance", "Lateral Movement": "lateral", "Exfiltration": "exfiltration", "Info": "info", } ENTITY_IMPORTANCE = {"low": 0, "medium": 1, "high": 2} ENTITY_IMPORTANCE_LABEL = {0: "Low", 1: "Medium", 2: "High"} SEVERITY = {"low": 1, "medium": 2, "high": 3, "critical": 4} MIRROR_DIRECTION = {"Incoming": "In", "Outgoing": "Out", "Incoming And Outgoing": "Both"} TAGS_REGEX = re.compile(r"^[\w:._ -]+$", re.U) DEFAULT_DETECTION_CLOSE_REASON = "Remediated" DEFAULT_DETECTION_STATUS_FOR_REOPEN = "Escalated" DEFAULT_ONLY_ESCALATED_DETECTIONS = False USER_ROLE_MAPPING = { "Admin": "admins", "Auditor": "auditor", "Global Analyst": "global_analyst", "Read-Only": "read_only", "Restricted Admin": "restricted_admins", "Security Analyst": "security_analyst", "Setting Admin": "setting_admins", "Super Admin": "super_admins", } ENDPOINTS = { "AUTH_ENDPOINT": "/oauth2/token", "EVENTS_DETECTIONS_ENDPOINT": "/api/v3.5/events/detections", "DETECTION_ENDPOINT": "/api/v3.5/detections", "ENTITY_ENDPOINT": "/api/v3.5/entities", "ENTITY_ENDPOINT_v34": "/api/v3.4/entities", "CLOSE_DETECTIONS_ENDPOINT": "/api/v3.5/detections/close", "ADD_NOTE_ENDPOINT": "/api/v3.5/detections/{}/notes", "LIST_TAGS_ENDPOINT": "/api/v3.5/tagging/detection/{}", "OPEN_DETECTIONS_ENDPOINT": "/api/v3.5/detections/open", "USER_ENDPOINT": "/api/v3.5/users", "GROUP_ENDPOINT": "/api/v3.5/groups", "ADD_AND_LIST_ENTITY_NOTE_ENDPOINT": "/api/v3.5/entities/{}/notes", "UPDATE_AND_REMOVE_ENTITY_NOTE_ENDPOINT": "/api/v3.5/entities/{}/notes/{}", "ENTITY_TAG_ENDPOINT": "/api/v3.5/tagging/entity/{}", "ASSIGNMENT_ENDPOINT": "/api/v3.5/assignments", "UPDATE_ASSIGNMENT_ENDPOINT": "/api/v3.5/assignments/{}", "RESOLVE_ASSIGNMENT_ENDPOINT": "/api/v3.5/assignments/{}/resolve", "ASSIGNMENT_OUTCOME_ENDPOINT": "/api/v3.5/assignment_outcomes/", "DOWNLOAD_DETECTION_PCAP": "/api/v3.5/detections/{}/pcap", "DETECTION_CLOSE_ENDPOINT": "/api/v3.5/detections/close", "DETECTION_OPEN_ENDPOINT": "/api/v3.5/detections/open", "DETECTION_TAG_ENDPOINT": "/api/v3.5/tagging/detection/{}", "ADD_AND_LIST_DETECTION_NOTE_ENDPOINT": "/api/v3.5/detections/{}/notes", "UPDATE_AND_REMOVE_DETECTION_NOTE_ENDPOINT": "/api/v3.5/detections/{}/notes/{}", "INVESTIGATION_ENDPOINT": "/api/v3.5/investigations", } ERRORS = { "INVALID_OBJECT": "Failed to parse {} object from response: {}", "INVALID_URGENCY_SCORE_THRESHOLD": "Invalid urgency score thresholds for severity mapping. Please ensure that the " "urgency score thresholds follow the correct order: " "urgency_score_low_threshold < urgency_score_medium_threshold < " "urgency_score_high_threshold.", "INVALID_COMMAND_ARG_VALUE": "Invalid '{}' value provided. Please ensure it is one of the values from the " "following options: {}.", "REQUIRED_ARGUMENT": "Please provide valid value of the '{}'. It is required field.", "INVALID_INTEGER_VALUE": "Invalid '{}' value. '{}' must be a non-zero and positive integer value.", "INVALID_NUMBER": '"{}" is not a valid number', "INVALID_PAGE_RESPONSE": "page contains no results", "INVALID_MAX_FETCH": "Invalid Max Fetch: {}. Max Fetch must be a positive integer ranging from 1 to 200.", "INVALID_PAGE_SIZE": "Invalid 'page size' provided. Please ensure that the page size value is between 1 and 5000.", "TRIAGE_AS_REQUIRED_WITH_DETECTION_IDS": "'triage_as' argument must be provided when using the 'detection_ids' argument. ", "INVALID_OUTCOME": "Invalid outcome value. Valid outcome values are: {}", "INVALID_SUPPORT_FOR_ARG": 'The argument "{}" must be set to "{}" when providing value for argument "{}".', "ENTITY_IDS_WITHOUT_TYPE": "When using the 'entity_ids' argument, 'entity_type' is required, and vice versa.", "INVALID_ARG_VALUE": "Invalid '{}' value provided. Please ensure it is one of the values from the following options: {}.", "INVALID_TIME_RANGE": "Invalid time range: '{}' ({}) must be earlier than '{}' ({}).", } """ CLIENT CLASS """ class VectraEventsDetectionsClient(BaseClient): """ Client class to interact with the Vectra Events Detections API. """ def __init__(self, server_url: str, client_id: str, client_secret_key: str, verify: bool, proxy: bool): """ Initializes the class instance. Args: server_url (str): The URL of the server. client_id (str): The client ID for authentication. client_secret_key (str): The client secret key for authentication. verify (bool): Indicates whether to verify the server's SSL certificate. proxy (bool): Indicates whether to use a proxy for the requests. """ super().__init__(base_url=server_url, verify=verify, proxy=proxy) self.client_id = client_id self.client_secret_key = client_secret_key # Fetch cached integration context. integration_context = get_integration_context() self._token = integration_context.get("access_token") or self._generate_tokens() def http_request( self, method: str, url_suffix: str = "", params: dict[str, Any] = None, data: dict[str, Any] = None, json_data: dict[str, Any] = None, response_type: str = "response", **kwargs, ): """ Makes an HTTP request to the server. Args: method (str): The HTTP method (e.g., GET, POST, PUT, DELETE). url_suffix (str): The URL suffix to be appended to the base URL. Defaults to an empty string. params (dict): Query parameters to be appended to the URL. Defaults to None. data (object): Data to be sent in the request body. Defaults to None. json_data (dict): JSON data to be sent in the request body. Defaults to None. response_type (str): The expected response type. Defaults to None. **kwargs: Additional keyword arguments. Returns: object: The response object or None. """ # Set the headers for the request, including the User-Agent and Authorization. headers = {"User-Agent": USER_AGENT, "Authorization": f"Bearer {self._token}"} demisto.debug(f"Making API request at {method} {url_suffix} with params: {params} and body: {data or json_data}") # Make the HTTP request using the _http_request method, passing the necessary parameters. res = self._http_request( method=method, url_suffix=url_suffix, headers=headers, data=data, json_data=json_data, params=params, retries=MAX_RETRIES, status_list_to_retry=STATUS_LIST_TO_RETRY, ok_codes=OK_CODES, backoff_factor=BACKOFF_FACTOR, resp_type="response", raise_on_status=True, **kwargs, ) # If the response status code indicates an authentication issue (e.g., 401), # generate a new access token using the refresh token and retry the request. if res.status_code in [401]: demisto.debug("Handling status code 401 by generating a new token using the refresh token.") self._token = self._generate_access_token_using_refresh_token() return self.http_request( method=method, url_suffix=url_suffix, params=params, response_type=response_type, data=data, json_data=json_data, **kwargs, ) try: result = None if response_type == "json": result = res.json() if response_type == "content": result = res.content() if response_type == "response": result = res if response_type == "text": result = res.text except ValueError as exception: raise DemistoException( f"Failed to parse {response_type} object from response: {res.content}", # type: ignore[str-bytes-safe] exception, res, ) # If the success response is received, then return it. if res.status_code in (200, 201, 204): return result # Return None if the response status code does not indicate success. return None def _generate_tokens(self) -> str: """ Generates access tokens using client credentials. Returns: str: The access token. """ demisto.info("Generating new access token.") payload = "grant_type=client_credentials" auth = requests.auth.HTTPBasicAuth(self.client_id, self.client_secret_key) headers = {"User-Agent": USER_AGENT, "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"} response = self._http_request( method="POST", url_suffix=ENDPOINTS["AUTH_ENDPOINT"], headers=headers, data=payload, auth=auth, retries=MAX_RETRIES, backoff_factor=BACKOFF_FACTOR, status_list_to_retry=STATUS_LIST_TO_RETRY, raise_on_status=True, ) access_token = response.get("access_token") refresh_token = response.get("refresh_token") set_integration_context({"access_token": access_token, "refresh_token": refresh_token}) return access_token def _generate_access_token_using_refresh_token(self) -> str: # type: ignore """ Generates a new access token using the refresh token. Returns: str: The access token. """ context = get_integration_context() refresh_token = context.get("refresh_token") demisto.info("Generating new access token using refresh token.") payload = f"grant_type=refresh_token&refresh_token={refresh_token}" headers = {"User-Agent": USER_AGENT, "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"} response = self._http_request( method="POST", url_suffix=ENDPOINTS["AUTH_ENDPOINT"], headers=headers, data=payload, ok_codes=OK_CODES, retries=MAX_RETRIES, backoff_factor=BACKOFF_FACTOR, raise_on_status=True, resp_type="response", ) if response.status_code in [401]: return self._generate_tokens() elif response.status_code in [200, 201]: access_token = response.json().get("access_token") # set new access token set_integration_context({"access_token": access_token, "refresh_token": refresh_token}) return access_token return "" def list_events_detections_request( self, params: dict[str, Any] = None, ) -> dict: """ List events detections. Args: params (dict[str, Any]): Fetch parameters. Returns: Dict: Response from the API containing the list of events detections. """ events = self.http_request( method="GET", url_suffix=ENDPOINTS["EVENTS_DETECTIONS_ENDPOINT"], params=params, response_type="json" ) return events def close_detections_by_ids_request(self, ids_list: list[str], reason: str) -> dict: """ Close detections by providing IDs of detections and close reason. Args: ids_list (list[str]): List of detection IDs. reason (str): Close reason. """ data = {"detectionIdList": ids_list, "reason": reason} return self.http_request( method="PATCH", url_suffix=ENDPOINTS["CLOSE_DETECTIONS_ENDPOINT"], json_data=data, response_type="json", ) def update_detection_status_request(self, ids_list: list[str], status: str) -> dict: """ Update detection status. Args: ids_list (list[str]): List of detection IDs to update. status (str): New status. """ data = {"detectionIdList": ids_list, "investigation_status": status} return self.http_request( method="PATCH", url_suffix=f"{ENDPOINTS['DETECTION_ENDPOINT']}", json_data=data, response_type="json", ) def update_detection_external_id_request(self, ids_list: list[str], external_reference_id: str) -> dict: """ Update detection external reference ID. Args: ids_list (list[str]): List of detection IDs to update. external_reference_id (str): New external reference ID. """ data = {"detectionIdList": ids_list, "external_reference_id": external_reference_id} return self.http_request( method="PATCH", url_suffix=f"{ENDPOINTS['DETECTION_ENDPOINT']}", json_data=data, response_type="json", ) def update_entity_external_id_request( self, entity_id: int, entity_type: str, external_reference_id: str, ) -> dict: """ Update entity external reference ID. Args: entity_id (int): Entity ID to update. entity_type (str): Entity type. external_reference_id (str): New external reference ID. """ params = {"type": entity_type} data = {"external_reference_id": external_reference_id} return self.http_request( method="PATCH", url_suffix=f"{ENDPOINTS['ENTITY_ENDPOINT']}/{entity_id}", params=params, json_data=data, response_type="json", ) def investigation_query_send(self, query, version) -> dict: """ Send investigation query. Args: query (str): Investigation query. version (str): Investigation version. """ data = {"query": query, "version": version} remove_nulls_from_dictionary(data) return self.http_request( method="POST", url_suffix=f"{ENDPOINTS['INVESTIGATION_ENDPOINT']}", json_data=data, response_type="json", ) def investigation_result_get(self, request_id: str, page: str, page_size: str) -> dict: """ Get investigation result. Args: request_id (str): Request ID. page (str): Page number. page_size (str): Page size. """ params = assign_params(page=page, page_size=page_size) return self.http_request( method="GET", url_suffix=f"{ENDPOINTS['INVESTIGATION_ENDPOINT']}/{request_id}", params=params, response_type="json", ) def update_entity_unresolved_priority_status_request( self, entity_id: str, entity_type: str, unresolved_priority: str, ) -> dict: """ Update entity unresolved priority status. Args: entity_id (str): Entity ID to update. entity_type (str): Entity type. unresolved_priority (str): Unresolved priority. """ params = {"type": entity_type} data = {"unresolved_priority": unresolved_priority} return self.http_request( method="PATCH", url_suffix=f"{ENDPOINTS['ENTITY_ENDPOINT']}/{entity_id}", params=params, json_data=data, response_type="json", ) def add_note_to_detection_request(self, detection_id: int, note: str) -> dict: """ Add note to detection. Args: detection_id (int): Detection ID to add note to. note (str): Note to add. """ body = {"note": note} return self.http_request( method="POST", url_suffix=ENDPOINTS["ADD_NOTE_ENDPOINT"].format(detection_id), json_data=body, response_type="json", ) def list_detection_tags_request(self, detection_id: int) -> dict: """ List detection tags. Args: detection_id (int): Detection ID to list tags for. """ params = {"type": "detection"} return self.http_request( method="GET", url_suffix=ENDPOINTS["LIST_TAGS_ENDPOINT"].format(detection_id), params=params, response_type="json", ) def update_detection_tags_request(self, detection_id: int, tags: list) -> dict: """ Update detection tags. Args: detection_id (int): Detection ID to update tags for. tags (list): List of tags to update. """ body = {"tags": tags} params = {"type": "detection"} return self.http_request( method="PATCH", url_suffix=ENDPOINTS["LIST_TAGS_ENDPOINT"].format(detection_id), params=params, json_data=body, response_type="json", ) def open_detections_by_ids_request(self, ids_list: list) -> dict: """ Open detections by providing IDs of detections. Args: ids_list (list[str]): List of detection IDs. """ data = {"detectionIdList": ids_list} return self.http_request( method="PATCH", url_suffix=ENDPOINTS["OPEN_DETECTIONS_ENDPOINT"], json_data=data, response_type="json", ) def list_detections_standalone_request( self, params: dict, ) -> dict: """ List detections. Args: params (dict): Parameters to filter detections. Returns: Dict: Response from the API containing the list of detections. """ detections = self.http_request( method="GET", url_suffix=ENDPOINTS["DETECTION_ENDPOINT"], params=params, response_type="json" ) return detections def list_users_request(self, email: str | None, role: str | None, last_login_timestamp: datetime | None) -> dict: """ List users. Args: email (str | None): The optional email to filter with (default: None). role (str | None): The optional user role to filter with (default: None). last_login_timestamp (datetime | None): Filter users after the specified last login timestamp (default: None). Returns: Dict: Response from the API containing the users. """ params = assign_params(email=email, role=role, last_login_gte=last_login_timestamp) return self.http_request(method="GET", url_suffix=ENDPOINTS["USER_ENDPOINT"], params=params, response_type="json") def list_entities_request( self, page: int = MAX_PAGE, page_size: int = MAX_PAGE_SIZE, is_prioritized: bool = None, entity_type: str = None, last_modified_timestamp: datetime | None = None, last_detection_timestamp: datetime | None = None, tags: str = None, ordering: str = None, state: str = "active", name: str = None, ) -> dict: """List entities. Args: page (int): The page number to retrieve (default: MAX_PAGE). page_size (int): The number of entities to retrieve per page (default: MAX_PAGE_SIZE). is_prioritized (bool): Filter entities by prioritization status (default: None). entity_type (str): Filter entities by type (default: None). last_modified_timestamp (str): Filter entities modified after the specified timestamp (default: None). last_detection_timestamp (str): Filter entities detected detection after the specified timestamp (default: None). tags (str): Filter entities by tags (default: None). ordering (str): Specify the ordering of the entities (default: None). state (str): Filter entities by state (default: 'active'). name (str): Filter entities by name (default: None). Returns: Dict: Response from the API containing the list of entities. """ params = assign_params( page=page, page_size=page_size, is_prioritized=is_prioritized, type=entity_type, last_modified_timestamp_gte=last_modified_timestamp, last_detection_timestamp_gte=last_detection_timestamp, tags=tags, state=state, ordering=ordering, name=name, ) entities = self.http_request( method="GET", url_suffix=ENDPOINTS["ENTITY_ENDPOINT_v34"], params=params, response_type="json", ) return entities def get_entity_request(self, entity_id: int = None, entity_type: str = None) -> dict: """Get entity by ID. Args: entity_id (int): The ID of the entity to retrieve. entity_type (str): Filter entity by type (default: None). Returns: Dict: Response from the API containing the entity information. """ params = assign_params(type=entity_type) entity = self.http_request( method="GET", url_suffix="{}/{}".format(ENDPOINTS["ENTITY_ENDPOINT_v34"], entity_id), params=params, response_type="json", ) return entity def list_detections_request( self, detection_category: str = None, detection_type: str = None, entity_id: int = None, entity_type: str = None, page: int = None, page_size: int = None, last_timestamp: datetime | None = None, tags: str = None, state: str = "active", detection_name: str = None, ids: str = None, ) -> dict: """ List detections. Args: detection_category (str, optional): Filter by detection category. detection_type (str, optional): Filter by detection type. entity_id (int, optional): Filter by entity ID. entity_type (str, optional): Filter by entity type. page (int, optional): Page number of the results. page_size (int, optional): Number of results per page. last_timestamp (str, optional): Filter by last timestamp greater than or equal to the provided value. tags (str, optional): Filter by tags. state (str, optional): Filter by detection state. detection_name (str, optional): Filter by detection name. ids(str, optional): Filter by detections ids. Returns: Dict: Response from the API containing the list of detections. """ params = assign_params( detection_category=detection_category, detection_type=detection_type, entity_id=entity_id, type=entity_type, page=page, page_size=page_size, last_timestamp_gte=last_timestamp, tags=tags, state=state, detection=detection_name, id=ids, ) detections = self.http_request( method="GET", url_suffix=ENDPOINTS["DETECTION_ENDPOINT"], params=params, response_type="json" ) return detections def list_entity_note_request(self, entity_id: int = None, entity_type: str = None) -> dict: """ List entity notes. Args: entity_id (int): The ID of the entity to add the note to. entity_type (str): The type of the entity. Returns: Dict: Response from the API. """ params = assign_params(type=entity_type) notes = self.http_request( method="GET", url_suffix=ENDPOINTS["ADD_AND_LIST_ENTITY_NOTE_ENDPOINT"].format(entity_id), params=params, response_type="json", ) return notes def add_entity_note_request(self, entity_id: int = None, entity_type: str = None, note: str = None) -> dict: """ Add a note to an entity. Args: entity_id (int): The ID of the entity to add the note to. entity_type (str): The type of the entity. note (str): The note to add. Returns: Dict: Response from the API containing the added note. """ params = assign_params(type=entity_type) data = {"note": note} notes = self.http_request( method="POST", url_suffix=ENDPOINTS["ADD_AND_LIST_ENTITY_NOTE_ENDPOINT"].format(entity_id), params=params, json_data=data, response_type="json", ) return notes def update_entity_note_request( self, entity_id: int = None, entity_type: str = None, note: str = None, note_id: int = None ) -> dict: """ Updates the note of an entity. Args: entity_id (int): The ID of the entity to update the note for. entity_type (str): The type of the entity. note (str): The updated note for the entity. note_id (int): The ID of the note to be updated. Returns: Dict: Response from the API containing the updated note details. """ params = assign_params(type=entity_type) data = {"note": note} notes = self.http_request( method="PATCH", url_suffix=ENDPOINTS["UPDATE_AND_REMOVE_ENTITY_NOTE_ENDPOINT"].format(entity_id, note_id), params=params, json_data=data, response_type="json", ) return notes def remove_entity_note_request(self, entity_id: int = None, entity_type: str = None, note_id: int = None): """ Removes a note from an entity. Args: entity_id (int): The ID of the entity to remove the note from. entity_type (str): The type of the entity. note_id (int): The ID of the note to be removed. Returns: Dict: Response from the API confirming the removal of the note. """ params = assign_params(type=entity_type) res = self.http_request( method="DELETE", url_suffix=ENDPOINTS["UPDATE_AND_REMOVE_ENTITY_NOTE_ENDPOINT"].format(entity_id, note_id), params=params, response_type="response", ) return res def update_entity_tags_request(self, entity_id: int = None, entity_type: str = None, tags: list = None) -> dict: """ Update tags to an entity. Args: entity_id (int): The ID of the entity to add the tags to. entity_type (str): The type of the entity. tags (List): Tags to set for entity. Returns: Dict: Response from the API containing the updated tags. """ params = assign_params(type=entity_type) data = {"tags": tags} res = self.http_request( method="PATCH", url_suffix=ENDPOINTS["ENTITY_TAG_ENDPOINT"].format(entity_id), params=params, json_data=data, response_type="json", ) return res def list_entity_tags_request(self, entity_id: int = None, entity_type: str = None) -> dict: """ List tags for the specified entity. Args: entity_id (int): The ID of the entity to add tags. entity_type (str): The type of the entity. Returns: Dict: Response from the API containing the tags. """ params = assign_params(type=entity_type) res = self.http_request( method="GET", url_suffix=ENDPOINTS["ENTITY_TAG_ENDPOINT"].format(entity_id), params=params, response_type="json" ) return res def list_assignments_request( self, account_ids: str = None, host_ids: str = None, resolution: str = None, resolved: bool = None, created_after: str = None, assignees: str = None, page: int = None, page_size: int = None, ) -> dict: """ Retrieve a list of assignments based on the provided account IDs and host IDs. Args: account_ids (str, optional): A string containing comma-separated account IDs to filter assignments. host_ids (str, optional): A string containing comma-separated host IDs to filter assignments. resolution (str, optional): The resolution status of the assignments. resolved (bool, optional): Whether the assignments are resolved (True) or unresolved (False). created_after (str, optional): Filter assignments created after this date and time. assignees (str, optional): A string containing comma-separated assignee usernames to filter assignments. page (int, optional): Page number of the results. page_size (int, optional): Number of results per page. Returns: dict: Response from the API. """ params = assign_params( accounts=account_ids, hosts=host_ids, resolution=resolution, resolved=resolved, created_after=created_after, assignees=assignees, page=page, page_size=page_size, ) res = self.http_request(method="GET", url_suffix=ENDPOINTS["ASSIGNMENT_ENDPOINT"], params=params, response_type="json") return res def add_entity_assignment_request( self, assign_to_user_id: int | None = None, assign_host_id: int | None = None, assign_account_id: int | None = None, ) -> dict: """ Send a request to add an entity assignment. Args: assign_to_user_id (str, optional): The ID of the user to whom the entity will be assigned. Defaults to None. assign_host_id (str, optional): The ID of the host to which the entity will be assigned. Defaults to None. assign_account_id (str, optional): The ID of the account to which the entity will be assigned. Defaults to None. Returns: dict: A dictionary containing the response from the API call. The structure of the dictionary depends on the specific implementation of the API. """ body = assign_params( assign_to_user_id=assign_to_user_id, assign_host_id=assign_host_id, assign_account_id=assign_account_id ) res = self.http_request(method="POST", url_suffix=ENDPOINTS["ASSIGNMENT_ENDPOINT"], json_data=body, response_type="json") return res def update_entity_assignment_request(self, assign_to_user_id: int | None = None, assignment_id: int | None = None) -> dict: """ Send a request to update an existing entity assignment. Args: assign_to_user_id (int, optional): The ID of the user to whom the entity will be reassigned. Defaults to None. assignment_id (int, optional): The ID of the assignment to be updated. Defaults to None. Returns: dict: Response from the API. """ body = assign_params(assign_to_user_id=assign_to_user_id) res = self.http_request( method="PUT", url_suffix=ENDPOINTS["UPDATE_ASSIGNMENT_ENDPOINT"].format(assignment_id), json_data=body, response_type="json", ) return res def download_detection_pcap_request(self, detection_id: str = None) -> Response: """ Send a request to download the packet capture (PCAP) associated with a Vectra detection. Args: detection_id (str, optional): The ID of the detection for which the PCAP should be downloaded. Returns: Response: Response from the API. """ res = self.http_request( method="GET", url_suffix=ENDPOINTS["DOWNLOAD_DETECTION_PCAP"].format(detection_id), response_type="response" ) return res def list_group_request( self, group_type: str, account_names: list[str], domains: list[str], host_ids: list[str], host_names: list[str], importance: str, ips: list[str], description: str, last_modified_timestamp: datetime | None, last_modified_by: str, group_name: str, ) -> dict: """ List groups as per the specified parameters. Args: group_type (str): Filter by group type. account_names (list[str]): Filter groups associated with accounts. domains (list[str]): Filter groups associated with domains. host_ids (list[str]): Filter groups associated with hosts. host_names (list[str]): Filter groups associated with hosts. importance (str): User defined group importance. ips (list[str]): Filter groups associated with ips. description (list[str]): Filter by group description. last_modified_timestamp (datetime | None): Filters for all groups modified on or after the given timestamp (GTE). last_modified_by (str): Filters groups by the user id who made the most recent modification. group_name (str): Filters by group name. Returns: Dict: Response from the API containing the tags. """ params = assign_params( type=group_type, account_names=",".join(account_names), domains=",".join(domains), host_ids=",".join(host_ids), host_names=",".join(host_names), importance=importance, ips=",".join(ips), description=description, name=group_name, last_modified_timestamp=last_modified_timestamp, last_modified_by=last_modified_by, ) res = self.http_request(method="GET", url_suffix=ENDPOINTS["GROUP_ENDPOINT"], params=params, response_type="json") return res def get_group_request(self, group_id: int = None) -> dict: """Get group by ID. Args: group_id (int): The ID of the group to retrieve. Returns: Dict: Response from the API containing the group information. """ group = self.http_request( method="GET", url_suffix="{}/{}".format(ENDPOINTS["GROUP_ENDPOINT"], group_id), response_type="json" ) return group def update_group_members_request(self, group_id: int = None, members: list = None) -> dict: """Update members in group. Args: group_id (int): The ID of the group to retrieve. members (List): The member list. Returns: Dict: Response from the API containing the group information. """ body = {"members": members} group = self.http_request( method="PATCH", url_suffix="{}/{}".format(ENDPOINTS["GROUP_ENDPOINT"], group_id), json_data=body, response_type="json" ) return group def close_detections_request(self, detection_ids: list[str], reason: str) -> dict: """ Close detections with a specific reason. Args: detection_ids (List[str]): List of detection IDs to close. reason (str): The close reason (benign or remediated). Returns: Dict: Response from the API. Raises: ValueError: If detection_ids is empty or reason is invalid. """ data = {"detectionIdList": detection_ids, "reason": reason} res = self.http_request( method="PATCH", url_suffix=ENDPOINTS["DETECTION_CLOSE_ENDPOINT"], json_data=data, response_type="json" ) return res def open_detections_request(self, detection_ids: list[str]) -> dict: """ Open detections with provided detection IDs. Args: detection_ids (List[str]): List of detection IDs to open. Returns: Dict: Response from the API. """ data = {"detectionIdList": detection_ids} res = self.http_request( method="PATCH", url_suffix=ENDPOINTS["DETECTION_OPEN_ENDPOINT"], json_data=data, response_type="json" ) return res def list_detection_note_request(self, detection_id: int) -> dict: """ List detection notes. Args: detection_id (int): The ID of the detection to get the notes for. Returns: Dict: Response from the API. """ notes = self.http_request( method="GET", url_suffix=ENDPOINTS["ADD_AND_LIST_DETECTION_NOTE_ENDPOINT"].format(detection_id), response_type="json", ) return notes def add_detection_note_request(self, detection_id: int = None, note: str = None) -> dict: """ Add a note to a detection. Args: detection_id (int): The ID of the detection to add the note to. note (str): The note to add. Returns: Dict: Response from the API containing the added note. """ data = {"note": note} notes = self.http_request( method="POST", url_suffix=ENDPOINTS["ADD_AND_LIST_DETECTION_NOTE_ENDPOINT"].format(detection_id), json_data=data, response_type="json", ) return notes def update_detection_note_request(self, detection_id: int = None, note: str = None, note_id: int = None) -> dict: """ Updates the note of a detection. Args: detection_id (int): The ID of the detection to update the note for. note (str): The updated note for the detection. note_id (int): The ID of the note to be updated. Returns: Dict: Response from the API containing the updated note details. """ data = {"note": note} notes = self.http_request( method="PATCH", url_suffix=ENDPOINTS["UPDATE_AND_REMOVE_DETECTION_NOTE_ENDPOINT"].format(detection_id, note_id), json_data=data, response_type="json", ) return notes def remove_detection_note_request(self, detection_id: int = None, note_id: int = None): """ Removes a note from a detection. Args: detection_id (int): The ID of the detection to remove the note from. note_id (int): The ID of the note to be removed. Returns: Dict: Response from the API confirming the removal of the note. """ res = self.http_request( method="DELETE", url_suffix=ENDPOINTS["UPDATE_AND_REMOVE_DETECTION_NOTE_ENDPOINT"].format(detection_id, note_id), response_type="response", ) return res """ HELPER FUNCTIONS """ def trim_spaces_from_args(args: dict) -> dict: """ Trim spaces from values of the args Dict. Args: args (Dict): Dictionary to trim spaces from. Returns: Dict: Arguments after trim spaces. """ for key, val in args.items(): if isinstance(val, str): args[key] = val.strip() val_list = argToList(val, transform=lambda x: x.strip()) args[key] = ",".join(val_list) return args def check_empty(x: Any) -> bool: """ Check if input is empty (None, empty dict, empty list, or empty string). :param x: Input to check. :type x: Any :return: True if x is empty, False otherwise. :rtype: bool """ return x is None or x == {} or x == [] or x == "" def remove_empty_elements_for_fetch(d: Any) -> Any: """ Recursively remove empty lists, empty dicts, or None elements from a dictionary or list. :param d: Input dictionary or list. :return: Dictionary or list with all empty lists, and empty dictionaries removed. """ if not isinstance(d, dict | list): return d elif isinstance(d, list): return [v for v in (remove_empty_elements_for_fetch(v) for v in d) if not check_empty(v)] return {k: v for k, v in ((k, remove_empty_elements_for_fetch(v)) for k, v in d.items()) if not check_empty(v)} def validate_positive_integer_arg(value: Any | None, arg_name: str, required: bool = False) -> bool: """ Validates whether the provided argument value is a valid positive integer. Args: value (int): The value to validate. arg_name (str): The name of the argument. required (bool): Flag indicating if the argument is required (default: False). Returns: bool: True if the value is a valid positive integer. Raises: ValueError: If the value is not a valid positive integer. """ if required and not value: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format(arg_name)) if value is not None and (not str(value).isdigit() or int(value) <= 0): raise ValueError(ERRORS["INVALID_INTEGER_VALUE"].format(arg_name, value)) return True def validate_urgency_score(urgency_score: str, score_name: str) -> int | None: """ Validates the urgency score to ensure it falls within the valid range of 0 to 100. Args: urgency_score (str): The urgency score to validate. score_name (str): The name of the urgency score. Raises: ValueError: If the urgency score is outside the valid range. """ score = arg_to_number(urgency_score, arg_name=score_name) if not MIN_URGENCY_SCORE <= score <= MAX_URGENCY_SCORE: # type: ignore raise ValueError(f"Please provide a valid {score_name} between 0 and 100.") return score def validate_entity_list_command_args(args: dict): """ Validate the arguments for the entity_list command. Args: args (Dict): The arguments passed to the entity_list command. Raises: ValueError: If any of the arguments are invalid. Returns: None """ entity_type = args.get("entity_type", "").lower() state = args.get("state", "").lower() page = args.get("page", "1") page_size = args.get("page_size", "50") # Validate entity_type value if entity_type and entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Validate state value if state and state not in VALID_ENTITY_STATE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("state", ", ".join(VALID_ENTITY_STATE))) validate_positive_integer_arg(page, arg_name="page") validate_positive_integer_arg(page_size, arg_name="page_size") if not 1 <= int(page_size) <= ENTITY_AND_DETECTION_MAX_PAGE_SIZE: raise ValueError(ERRORS["INVALID_PAGE_SIZE"]) def validate_list_entity_detections_args(args: dict[str, Any]): """ Validate the arguments for listing entity detections. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If the entity ID is not provided. ValueError: If the detection category is invalid. ValueError: If the page size is invalid. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type", "").lower() detection_category = args.get("detection_category") page = args.get("page", "1") page_size = args.get("page_size", "50") validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) if detection_category and detection_category not in DETECTION_CATEGORY_TO_ARG: raise ValueError( ERRORS["INVALID_COMMAND_ARG_VALUE"].format("detection_category", ", ".join(DETECTION_CATEGORY_TO_ARG.keys())) ) validate_positive_integer_arg(value=page, arg_name="page") validate_positive_integer_arg(value=page_size, arg_name="page_size") if not 1 <= int(page_size) <= ENTITY_AND_DETECTION_MAX_PAGE_SIZE: raise ValueError(ERRORS["INVALID_PAGE_SIZE"]) def validate_detection_describe_args(args: dict[str, Any]): """ Validate the arguments for detection describe. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If the detection IDs are not provided. ValueError: If the page size is invalid. """ detection_ids = args.get("detection_ids", "") page = args.get("page", "1") page_size = args.get("page_size", "50") detection_ids = argToList(detection_ids, transform=arg_to_number) found_valid_detection_ids = False for detection_id in detection_ids: if isinstance(detection_id, int): if detection_id < 1: raise ValueError(ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", detection_id)) found_valid_detection_ids = True if not found_valid_detection_ids: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) validate_positive_integer_arg(value=page, arg_name="page") validate_positive_integer_arg(value=page_size, arg_name="page_size") if not 1 <= int(page_size) <= ENTITY_AND_DETECTION_MAX_PAGE_SIZE: raise ValueError(ERRORS["INVALID_PAGE_SIZE"]) def validate_entity_note_list_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity list add command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ entity_type = args.get("entity_type", "").lower() entity_id = args.get("entity_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) def validate_entity_note_add_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity note add command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ entity_type = args.get("entity_type", "").lower() note = args.get("note") entity_id = args.get("entity_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Validate note value if not note: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("note")) def validate_entity_note_update_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity note update command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ entity_type = args.get("entity_type", "").lower() note = args.get("note") entity_id = args.get("entity_id") note_id = args.get("note_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate note_id value validate_positive_integer_arg(note_id, arg_name="note_id", required=True) if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Validate note value if not note: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("note")) def validate_entity_note_remove_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity note update command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ entity_type = args.get("entity_type", "").lower() entity_id = args.get("entity_id") note_id = args.get("note_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate note_id value validate_positive_integer_arg(note_id, arg_name="note_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type.lower() not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) def validate_entity_tag_add_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity tag add command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ validate_entity_tag_list_command_args(args) tags = argToList(args.get("tags", "")) # Validate Tags value if not [tag.strip() for tag in tags if isinstance(tag, str) and tag.strip()]: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("tags")) def validate_entity_tag_list_command_args(args: dict[str, Any]): """ Validates the arguments provided for the entity tag list command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ entity_type = args.get("entity_type", "").lower() entity_id = args.get("entity_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) def validate_assignment_list_command_args(args: dict): """ Validate the arguments provided for the assignment list command. Args: args (Dict): A dictionary containing the arguments for the assignment list command. Raises: ValueError: If the provided entity_type is not one of the valid types. ValueError: If entity_ids are provided without an entity_type and vice-versa. ValueError: If page or page_size values are not positive integers. """ entity_ids = args.get("entity_ids") entity_type = args.get("entity_type") page = args.get("page", "1") page_size = args.get("page_size", "50") # Validate entity type if entity_type and entity_type.lower() not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Validate entity ids without entity_type and vice-versa if (entity_ids and not entity_type) or (entity_type and not entity_ids): raise ValueError(ERRORS["ENTITY_IDS_WITHOUT_TYPE"]) # Validate pagination validate_positive_integer_arg(value=page, arg_name="page") validate_positive_integer_arg(value=page_size, arg_name="page_size") def validate_entity_assignment_add_command_args(args: dict): """ Validate the arguments provided for adding an entity assignment. Args: args (Dict): A dictionary containing the arguments for adding an entity assignment. Raises: ValueError: If the provided entity_id or user_id is not a positive integer. ValueError: If the entity_type is missing or not one of the valid types. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type") user_id = args.get("user_id") # Validate entity_id value validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate note_id value validate_positive_integer_arg(user_id, arg_name="user_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type.lower() not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) def validate_entity_assignment_update_command_args(args: dict): """ Validate the arguments provided for updating an entity assignment. Args: args (Dict): A dictionary containing the arguments for updating an entity assignment. Raises: ValueError: If the provided assignment_id or user_id is not a positive integer. """ assignment_id = args.get("assignment_id") user_id = args.get("user_id") # Validate assignment_id value validate_positive_integer_arg(assignment_id, arg_name="assignment_id", required=True) # Validate user_id value validate_positive_integer_arg(user_id, arg_name="user_id", required=True) def validate_group_list_command_args(args: dict[str, Any]): """ Validates the arguments provided for the group list command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ group_type = args.get("group_type") or "" if group_type and isinstance(group_type, str): group_type = group_type.lower() # Validate group_type value if group_type not in VALID_GROUP_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("group_type", ", ".join(VALID_GROUP_TYPE))) importance = args.get("importance") or "" # Validate importance value if importance and isinstance(importance, str) and importance.lower() not in VALID_IMPORTANCE_VALUE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("importance", ", ".join(VALID_IMPORTANCE_VALUE))) # Validate account_names value account_names = argToList(args.get("account_names") or "") if account_names and group_type != "account": raise ValueError(ERRORS["INVALID_SUPPORT_FOR_ARG"].format("group_type", "account", "account_names")) # Validate domains value domains = argToList(args.get("domains") or "") if domains and group_type != "domain": raise ValueError(ERRORS["INVALID_SUPPORT_FOR_ARG"].format("group_type", "domain", "domains")) # Validate host_ids value host_ids = argToList(args.get("host_ids") or "") if host_ids and group_type != "host": raise ValueError(ERRORS["INVALID_SUPPORT_FOR_ARG"].format("group_type", "host", "host_ids")) for host_id in host_ids: host_id = arg_to_number(host_id, "host_ids") validate_positive_integer_arg(host_id, arg_name="host_ids") # Validate host_names value host_names = argToList(args.get("host_names") or "") if host_names and group_type != "host": raise ValueError(ERRORS["INVALID_SUPPORT_FOR_ARG"].format("group_type", "host", "host_names")) # Validate ips value ips = argToList(args.get("ips") or "") if ips and group_type != "ip": raise ValueError(ERRORS["INVALID_SUPPORT_FOR_ARG"].format("group_type", "ip", "ips")) def validate_group_assign_and_unassign_command_args(args): """ Validate the arguments provided for assigning or unassigning members to/from a group. Args: args (Dict): A dictionary containing the arguments for the group assign and unassign command. Raises: ValueError: If the provided group_id is not a positive integer. ValueError: If members argument is missing. """ group_id = args.get("group_id") members = args.get("members") validate_positive_integer_arg(group_id, arg_name="group_id", required=True) if not members: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("members")) def validate_entity_detections_mark_asclosed_command_args(args): """ Validate the arguments for marking entity detections as closed. Args: args (Dict): The command arguments. Raises: ValueError: If entity_id, entity_type, or close_reason are invalid. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type", "").lower() close_reason = args.get("close_reason", "").lower() validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) if not close_reason: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("close_reason")) if close_reason not in VALID_CLOSE_REASON: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("close_reason", ", ".join(VALID_CLOSE_REASON))) def validate_detection_tag_add_command_args(args): """ Validates the arguments provided for the detection tag add command. Args: args (dict): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ detection_id = args.get("detection_id") tags = argToList(args.get("tags", "")) # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) # Validate Tags value if not [tag.strip() for tag in tags if isinstance(tag, str) and tag.strip()]: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("tags")) def validate_detection_note_list_command_args(args: dict[Any, Any]): """ Validates the arguments provided for the detection note list command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ detection_id = args.get("detection_id") # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) def validate_detection_note_add_command_args(args: dict[Any, Any]): """ Validates the arguments provided for the detection note add command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ note = args.get("note") detection_id = args.get("detection_id") # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) if not note: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("note")) def validate_detection_note_update_command_args(args: dict[Any, Any]): """ Validates the arguments provided for the detection note update command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ note = args.get("note") detection_id = args.get("detection_id") note_id = args.get("note_id") # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) # Validate note_id value validate_positive_integer_arg(note_id, arg_name="note_id", required=True) # Validate note value if not note: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("note")) def validate_detection_note_remove_command_args(args: dict[Any, Any]): """ Validates the arguments provided for the detection note remove command. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: If any of the arguments are invalid. """ detection_id = args.get("detection_id") note_id = args.get("note_id") # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) # Validate note_id value validate_positive_integer_arg(note_id, arg_name="note_id", required=True) def calc_pages(total_count: int, per_page_count: int): """ Calculates the number of pages required to display all the items, considering the number of items to be displayed per page Args: total_count (int): The total number of items. per_page_count (int): The count of items per page. Returns: int: The total number of pages. """ return -(-total_count // per_page_count) def trim_api_version(url: str) -> str: """ Trim the '/api/v3.x' portion from a URL. Args: url (str): The URL to trim. Returns: str: The trimmed URL. """ api_versions = ["/api/v3.5", "/api/v3.4", "/api/v3.3", "/api/v3"] for api_version in api_versions: if api_version in url: trimmed_url = url.replace(api_version, "") + UTM_PIVOT return trimmed_url return url + UTM_PIVOT def get_user_list_command_hr(users: list): """ Converts a list of users into a human-readable table format. Args: users (Dict): The list of entities to convert. Returns: str: The human-readable table in Markdown format. """ hr_dict = [] # Process detection_set and create detection_ids field for user in users: # type: ignore user["user_id"] = user["id"] hr_dict.append( { "User ID": user.get("user_id"), "User Name": user.get("name"), "Email": user.get("email"), "Role": user.get("role"), "Last Login Timestamp": user.get("last_login_timestamp"), } ) # Prepare human-readable output table human_readable = tableToMarkdown( "Users Table", hr_dict, ["User ID", "User Name", "Email", "Role", "Last Login Timestamp"], removeNull=True ) return human_readable def get_entity_list_command_hr(entities: dict, page: int, page_size: int, count: int): """ Converts a list of entities into a human-readable table format. Args: entities (Dict): The list of entities to convert. page (int): The current page number. page_size (int): The page size. count (int): The total count of entities. Returns: str: The human-readable table in Markdown format. """ hr_dict = [] entity_list = copy.deepcopy(entities) # Process detection_set and create detection_ids field for entity in entity_list: # type: ignore # Trim API version from url entity["url"] = trim_api_version(entity.get("url")) # Convert ID into clickable URL entity["id_url"] = f"[{entity['id']}]({entity['url']})" # Map entity importance entity["importance"] = ENTITY_IMPORTANCE_LABEL[entity.get("importance")] if "detection_set" in entity: entity["detection_ids"] = ", ".join( [ "[{}]({})".format(detection.split("/")[-1], trim_api_version(detection)) for detection in entity.get("detection_set") ] ) hr_dict.append( { "ID": entity.get("id_url"), "Name": entity.get("name"), "Entity Type": entity.get("type"), "Urgency Score": entity.get("urgency_score"), "Entity Importance": entity.get("importance"), "Last Modified Timestamp": entity.get("last_modified_timestamp"), "Last Detection Timestamp": entity.get("last_detection_timestamp"), "Detections IDs": entity.get("detection_ids"), "Prioritize": entity.get("is_prioritized"), "State": entity.get("state"), "Tags": ", ".join(entity.get("tags")) if entity.get("tags") else None, } ) # Prepare human-readable output table pages = calc_pages(per_page_count=page_size, total_count=count) # type: ignore human_readable = tableToMarkdown( f"Entities Table (Showing Page {page} out of {pages})", hr_dict, [ "ID", "Name", "Entity Type", "Urgency Score", "Entity Importance", "Last Detection Timestamp", "Last Modified Timestamp", "Detections IDs", "Prioritize", "State", "Tags", ], removeNull=True, ) return human_readable def get_entity_get_command_hr(entity: dict): """ Returns the human-readable output for the entity details. Args: entity (Dict): The entity details dictionary. Returns: str: The human-readable output. """ hr_dict = [] entity_res = copy.deepcopy(entity) # Trim API version from entity url entity_res["url"] = trim_api_version(entity_res.get("url")) # type: ignore entity_res["id"] = f"[{entity_res['id']}]({entity_res['url']})" # Process detection_set and create detection_ids field if "detection_set" in entity_res: entity_res["detection_ids"] = ", ".join( [ "[{}]({})".format(detection.split("/")[-1], trim_api_version(detection)) for detection in entity_res.get("detection_set", []) ] ) # type: ignore # Entity importance value to label entity_res["importance"] = ENTITY_IMPORTANCE_LABEL[entity_res.get("importance")] # type: ignore hr_dict.append( { "Name": entity_res.get("name"), "Entity Type": entity_res.get("type"), "Urgency Score": entity_res.get("urgency_score"), "Entity Importance": entity_res.get("importance"), "Last Modified Timestamp": entity_res.get("last_modified_timestamp"), "Last Detection Timestamp": entity_res.get("last_detection_timestamp"), "Detections IDs": entity_res.get("detection_ids"), "Prioritize": entity_res.get("is_prioritized"), "State": entity_res.get("state"), "Tags": ", ".join(entity_res.get("tags")) if entity_res.get("tags") else None, # type: ignore } ) # Prepare human-readable output table human_readable = tableToMarkdown( f"Entity detail:\n#### Entity ID: {entity_res.get('id')}", hr_dict, [ "Name", "Entity Type", "Urgency Score", "Entity Importance", "Last Detection Timestamp", "Last Modified Timestamp", "Detections IDs", "Prioritize", "State", "Tags", ], removeNull=True, ) return human_readable def get_list_entity_detections_command_hr(detections: dict[Any, Any], page: int | None, page_size: int | None, count: int): """ Converts the list of detections into a human-readable table format. Args: detections (Dict): Dictionary containing the list of detections. page (int): The current page number. page_size (int): The page size. count (int): The total count of detections. Returns: str: Human-readable table representation of the detections. """ hr_dict = [] detection_list = copy.deepcopy(detections) # Process detection_set and create detection_ids field for detection in detection_list: # type: ignore # Trim API version from url detection["url"] = trim_api_version(detection.get("url")) # Convert ID into clickable URL detection["id"] = f"[{detection['id']}]({detection['url']})" account_url = None host_url = None if detection.get("src_account"): account_url = ( f"[{detection.get('src_account').get('name')}]({trim_api_version(detection.get('src_account').get('url'))})" ) if detection.get("src_host"): host_url = f"[{detection.get('src_host').get('name')}]({trim_api_version(detection.get('src_host').get('url'))})" summary = detection.get("summary") num_events = 0 # For counting number of events if summary and isinstance(summary, dict): num_events = int(summary.get("num_events") or 0) hr_dict.append( { "ID": detection.get("id"), "Detection Name": detection.get("detection"), "Detection Type": detection.get("detection_type"), "Category": detection.get("detection_category"), "Account Name": account_url, "Host Name": host_url, "Src IP": detection.get("src_ip"), "Threat Score": detection.get("threat"), "Certainty Score": detection.get("certainty"), "Number Of Events": num_events, "State": detection.get("state"), "Tags": detection.get("tags"), "Last Timestamp": detection.get("last_timestamp"), } ) pages = calc_pages(per_page_count=page_size, total_count=count) # type: ignore human_readable = tableToMarkdown( f"Detections Table (Showing Page {page} out of {pages})", hr_dict, [ "ID", "Detection Name", "Detection Type", "Category", "Account Name", "Host Name", "Src IP", "Threat Score", "Certainty Score", "Number Of Events", "State", "Tags", "Last Timestamp", ], removeNull=True, ) return human_readable def get_assignment_list_command_hr(assignments: dict, page: int | None, page_size: int | None, count: int): """ Returns the human-readable output for the assignment. Args: assignments(Dict): The assignment details dictionary. page (int): The current page number. page_size (int): The page size. count (int): The total count of assignments. Returns: str: The human-readable output. """ hr_dict = [] for assignment in assignments: assignment["assignment_id"] = assignment["id"] hr_dict.append( { "Account ID": assignment.get("account_id"), "Host ID": assignment.get("host_id"), "Assignment ID": assignment.get("id"), "Assigned By": assignment.get("assigned_by", {}).get("username", ""), "Assigned To": assignment.get("assigned_to", {}).get("username", ""), "Date Assigned": assignment.get("date_assigned"), "Resolved By": assignment.get("resolved_by", {}).get("username", ""), "Date Resolved": assignment.get("date_resolved"), "Outcome ID": assignment.get("outcome", {}).get("id", ""), "Outcome": assignment.get("outcome", {}).get("title", ""), } ) pages = calc_pages(per_page_count=page_size, total_count=count) # type: ignore human_readable = tableToMarkdown( f"Assignments Table (Showing Page {page} out of {pages})", hr_dict, [ "Account ID", "Host ID", "Assignment ID", "Assigned By", "Assigned To", "Date Assigned", "Resolved By", "Date Resolved", "Outcome ID", "Outcome", ], removeNull=True, ) return human_readable, assignments def entity_assignment_add_command_hr(assignment: dict) -> str: """ Returns the human-readable output for the assignment. Args: assignment (Dict): The assignment details dictionary. Returns: str: The human-readable output. """ assigned_by = assignment.get("assigned_by", {}) assigned_to = assignment.get("assigned_to", {}) events = assignment.get("events", [{}]) hr_dict = [ { "Assignment ID": assignment.get("assignment_id"), "Assigned By": assigned_by.get("username") if isinstance(assigned_by, dict) else "", "Assigned Date": assignment.get("date_assigned"), "Assigned To": assigned_to.get("username") if isinstance(assigned_to, dict) else "", "Event Type": events[0].get("event_type") if isinstance(events, list) and len(events) > 0 else "", } ] # Prepare human-readable output table human_readable = tableToMarkdown( "Assignment detail", hr_dict, ["Assignment ID", "Assigned By", "Assigned Date", "Assigned To", "Event Type"], removeNull=True, ) return human_readable def get_list_entity_notes_command_hr(notes: dict, entity_id: int | None, entity_type: str) -> str: """ Returns the human-readable output for the entity notes. Args: notes (Dict): The assignment details dictionary. entity_id (int | None): Entity ID. entity_type (str): Entity Type. Returns: str: The human-readable output. """ hr_dict = [] for note in notes: note["note_id"] = note["id"] note.update({"entity_id": entity_id, "entity_type": entity_type}) hr_dict.append( { "Note ID": note.get("id"), "Note": note.get("note"), "Created By": note.get("created_by"), "Created Date": note.get("date_created"), "Modified By": note.get("modified_by"), "Modified Date": note.get("date_modified"), } ) # Prepare human-readable output table human_readable = tableToMarkdown( "Entity Notes Table", hr_dict, ["Note ID", "Note", "Created By", "Created Date", "Modified By", "Modified Date"], removeNull=True, ) return human_readable def get_group_list_command_hr(groups: list): """ Converts a list of groups into a human-readable table format. Args: groups (Dict): The list of groups to convert. Returns: str: The human-readable table in Markdown format. """ hr_dict = [] # Process members data from group and make HR for groups for group in groups: # type: ignore group["group_id"] = group["id"] members: list = group.get("members") members_hr = None if members and isinstance(members, list): # If the members are simple list of strings, then join them with comma. if isinstance(members[0], str): members_hr = ", ".join([re.escape(str(member)) for member in members]) # If the members are list of dictionaries, then extract important field from that and join it with comma. elif isinstance(members[0], dict): members_list = [] for member in members: if member.get("uid"): members_list.append(re.escape(str(member.get("uid")))) # type: ignore elif member.get("id"): members_list.append( # type: ignore "[{}]({})".format(member.get("id"), trim_api_version(member.get("url"))) ) members_hr = ", ".join(members_list) hr_dict.append( { "Group ID": group.get("group_id"), "Name": group.get("name"), "Group Type": group.get("type"), "Description": group.get("description"), "Importance": group.get("importance"), "Members": members_hr, "Last Modified Timestamp": group.get("last_modified"), } ) # Prepare human-readable output table human_readable = tableToMarkdown( "Groups Table", hr_dict, ["Group ID", "Name", "Group Type", "Description", "Importance", "Members", "Last Modified Timestamp"], removeNull=True, ) return human_readable def get_group_unassign_and_assign_command_hr(group: dict, changed_members: list, assign_flag: bool = False): """ Converts group into a human-readable table format. Args: group (Dict): The group to convert. changed_members (List): Removed/Added members from the group. assign_flag (bool): True for unassigning members, False for assigning members. Returns: str: The human-readable table in Markdown format. """ hr_dict = [] group["group_id"] = group["id"] members = group.get("members") members_hr = None if members and isinstance(members, list): # If the members are simple list of strings, then join them with comma. if isinstance(members[0], str): members_hr = ", ".join([re.escape(str(member)) for member in members]) # If the members are list of dictionaries, then extract important field from that and join it with comma. elif isinstance(members[0], dict): members_list = [] for member in members: if member.get("uid"): members_list.append(re.escape(str(member.get("uid")))) # type: ignore elif member.get("id"): members_list.append( # type: ignore "[{}]({})".format(member.get("id"), trim_api_version(member.get("url"))) ) members_hr = ", ".join(members_list) hr_dict.append( { "Group ID": group.get("group_id"), "Name": group.get("name"), "Group Type": group.get("type"), "Description": group.get("description"), "Members": members_hr, "Last Modified Timestamp": group.get("last_modified"), } ) # Prepare human-readable output table change_action = "assigned to" if assign_flag else "unassigned from" changed_members = [re.escape(member) for member in changed_members] human_readable = tableToMarkdown( f"Member(s) {', '.join(changed_members)} have been {change_action} the group.\n### Updated group details:", hr_dict, ["Group ID", "Name", "Group Type", "Description", "Members", "Last Modified Timestamp"], removeNull=True, ) return human_readable def get_list_detection_notes_command_hr(notes: dict, detection_id: int | None) -> str: """ Returns the human-readable output for the detection notes. Args: notes (Dict): list of detection notes. detection_id (int | None): Detection ID. Returns: str: The human-readable output. """ hr_dict = [] for note in notes: note["note_id"] = note["id"] note.update({"detection_id": detection_id}) hr_dict.append( { "Note ID": note.get("id"), "Note": note.get("note"), "Created By": note.get("created_by"), "Created Date": note.get("date_created"), "Modified By": note.get("modified_by"), "Modified Date": note.get("date_modified"), } ) # Prepare human-readable output table human_readable = tableToMarkdown( "Detection Notes Table", hr_dict, ["Note ID", "Note", "Created By", "Created Date", "Modified By", "Modified Date"], removeNull=True, ) return human_readable def merge_values(value1, value2): """ Merge two values based on their types. - Strings, Numbers: dict2 value takes priority - Lists of dicts: dict2 value takes priority - Lists of other types: combine and remove duplicates (preserves order) - Dicts: recursively merge - Other types: dict2 takes priority """ if isinstance(value1, dict) and isinstance(value2, dict): return update_dict_with_new_dict_values(value1, value2) # Both are lists if isinstance(value1, list) and isinstance(value2, list): # Check if list contains dicts if all(isinstance(item, dict) for item in value1 + value2): return value2 # Mixed or other types - append and remove duplicates merged = [] seen = [] for item in value1 + value2: if item not in seen: merged.append(item) seen.append(item) return merged # Different types or simple values - dict2 takes priority return value2 def update_dict_with_new_dict_values(dict1: dict, dict2: dict) -> dict: """ Recursively merge dict1 with dict2, handling different value types intelligently. - For strings: dict2 value replaces dict1 (if both valid) - For lists of strings: combine and remove duplicates - For dicts: recursively merge - For lists of dicts: dict2 value replaces dict1 (if both valid) - dict2 takes priority when both have valid values - Valid values: non-empty strings, non-empty lists, non-empty dicts - Preserves all valid data from both dictionaries Args: dict1 (dict): The base dictionary. dict2 (dict): The dictionary containing new values (priority). Returns: dict: A new merged dictionary with combined values. """ result = dict1.copy() for key, value2 in dict2.items() if dict2 else {}: if check_empty(value2): continue if key in result: value1 = result[key] if not check_empty(value1): result[key] = merge_values(value1, value2) else: result[key] = value2 else: result[key] = value2 return result def validate_time_range( after_time: datetime | None, before_time: datetime | None, after_arg_name: str, before_arg_name: str ) -> None: """ Validate that the 'after' timestamp is earlier than the 'before' timestamp. Args: after_time (datetime | None): The 'after' timestamp. before_time (datetime | None): The 'before' timestamp. after_arg_name (str): The name of the 'after' argument for error messages. before_arg_name (str): The name of the 'before' argument for error messages. Raises: ValueError: If after_time is not earlier than before_time. """ if after_time and before_time and after_time >= before_time: raise ValueError( ERRORS["INVALID_TIME_RANGE"].format( after_arg_name, after_time.strftime(DATE_FORMAT), before_arg_name, before_time.strftime(DATE_FORMAT), ) ) def validate_list_detections_args(args: dict[Any, Any]) -> dict[str, Any]: """ Validate the arguments for listing entity detections. Args: args (dict[Any, Any]): The arguments dictionary. Raises: ValueError: arg is invalid. return: params: return the dict values of params. """ created_after_dt = arg_to_datetime(args.get("created_after"), arg_name="created_after") created_before_dt = arg_to_datetime(args.get("created_before"), arg_name="created_before") # Validate that created_after is earlier than created_before validate_time_range(created_after_dt, created_before_dt, "created_after", "created_before") created_after = created_after_dt.strftime(DATE_FORMAT) if created_after_dt else None # type: ignore created_before = created_before_dt.strftime(DATE_FORMAT) if created_before_dt else None # type: ignore last_detected_after_dt = arg_to_datetime(args.get("last_detected_after"), arg_name="last_detected_after") last_detected_before_dt = arg_to_datetime(args.get("last_detected_before"), arg_name="last_detected_before") # Validate that updated_after is earlier than updated_before validate_time_range(last_detected_after_dt, last_detected_before_dt, "last_detected_after", "last_detected_before") last_detected_after = last_detected_after_dt.strftime(DATE_FORMAT) if last_detected_after_dt else None # type: ignore last_detected_before = last_detected_before_dt.strftime(DATE_FORMAT) if last_detected_before_dt else None # type: ignore description = args.get("description") detection_name = args.get("detection_name") detection_type = args.get("detection_type") detection_category = args.get("detection_category") include_info_category_detections = args.get("include_info_category_detections", "true") close_reason = args.get("close_reason") detection_state = args.get("detection_state") tags = argToList(args.get("tags")) is_triaged = args.get("is_triaged", "false") page = args.get("page", MAX_PAGE) page_size = args.get("page_size", MAX_PAGE_SIZE) entity_type = args.get("entity_type") if include_info_category_detections: if include_info_category_detections.lower() not in VALID_BOOL_VALUES: raise ValueError(ERRORS["INVALID_ARG_VALUE"].format("include_info_category_detections", ", ".join(VALID_BOOL_VALUES))) else: include_info_category_detections = argToBoolean(args.get("include_info_category_detections", "true")) if is_triaged: if is_triaged.lower() not in VALID_BOOL_VALUES: raise ValueError(ERRORS["INVALID_ARG_VALUE"].format("is_triaged", ", ".join(VALID_BOOL_VALUES))) else: is_triaged = argToBoolean(args.get("is_triaged", "false")) if entity_type and entity_type.capitalize() not in VALID_ENTITY_TYPES: raise ValueError(ERRORS["INVALID_ARG_VALUE"].format("entity_type", (", ".join(VALID_ENTITY_TYPES)).lower())) if close_reason and close_reason not in VALID_CLOSE_REASON: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("close_reason", ", ".join(VALID_CLOSE_REASON))) validate_positive_integer_arg(value=page, arg_name="page") validate_positive_integer_arg(value=page_size, arg_name="page_size") if not 1 <= int(page_size) <= ENTITY_AND_DETECTION_MAX_PAGE_SIZE: raise ValueError(ERRORS["INVALID_PAGE_SIZE"]) params = assign_params( created_timestamp_gte=created_after, created_timestamp_lte=created_before, last_timestamp_gte=last_detected_after, last_timestamp_lte=last_detected_before, description=description, detection=detection_name, detection_type=detection_type, detection_category=detection_category, include_info_category=include_info_category_detections, reason=close_reason, state=detection_state, tags=",".join(tags), type=entity_type, is_triaged=is_triaged, page=page, page_size=page_size, ) return params def investigation_result_get_command_hr(result: dict): """ Returns the human-readable output for the investigation results details. Args: entity (Dict): The entity details dictionary. Returns: str: The human-readable output. """ hr_dict = [] meta_data = result.get("meta", {}) or {} hr_dict.append( { "Query Status": meta_data.get("query_status", ""), "Page Number": meta_data.get("page", ""), "Page size": meta_data.get("page_size", ""), "Total Rows": meta_data.get("num_rows_available", ""), "File Size (bytes)": meta_data.get("estimated_file_size_bytes", ""), "Columns": meta_data.get("columns", ""), } ) # Prepare human-readable output table human_readable = tableToMarkdown( f"Investigation Result for Request ID: {result.get('request_id')}", hr_dict, [ "Query Status", "Page Number", "Page size", "Total Rows", "File Size (bytes)", "Columns", ], removeNull=True, json_transform_mapping={"Columns": JsonTransformer()}, ) return human_readable """ COMMAND FUNCTIONS """ def vectra_user_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Retrieves a list of users from the Vectra API. Args: client (VectraEventsDetectionsClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the entities. """ last_login_timestamp = arg_to_datetime(args.get("last_login_timestamp"), arg_name="last_login_timestamp") if last_login_timestamp: last_login_timestamp = last_login_timestamp.strftime(DATE_FORMAT) # type: ignore email = args.get("email", "") role = args.get("role", "") if role and role in USER_ROLE_MAPPING: role = USER_ROLE_MAPPING.get(role) # Call Vectra API to retrieve users response = client.list_users_request(email=email, role=role, last_login_timestamp=last_login_timestamp) count = response.get("count") if count == 0: return CommandResults(outputs={}, readable_output="##### Got the empty list of users.", raw_response=response) users = response.get("results") # Prepare context data human_readable = get_user_list_command_hr(users) # type: ignore context = [createContext(user) for user in remove_empty_elements(users)] # type: ignore return CommandResults( outputs_prefix="Vectra.User", outputs=context, readable_output=human_readable, raw_response=users, outputs_key_field=["user_id"], ) def vectra_entity_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Retrieves a list of entities from the Vectra API. Args: client (VectraEventsDetectionsClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the entities. Raises: ValueError: If an invalid entity_type or state value is provided. """ # Validate command args validate_entity_list_command_args(args) # Get function arguments entity_type = args.get("entity_type", "").lower() last_detection_timestamp = arg_to_datetime(args.get("last_detection_timestamp"), arg_name="last_detection_timestamp") last_modified_timestamp = arg_to_datetime(args.get("last_modified_timestamp"), arg_name="last_modified_timestamp") if last_detection_timestamp: last_detection_timestamp = last_detection_timestamp.strftime(DATE_FORMAT) # type: ignore if last_modified_timestamp: last_modified_timestamp = last_modified_timestamp.strftime(DATE_FORMAT) # type: ignore ordering = args.get("ordering", "") page = arg_to_number(args.get("page", "1"), arg_name="page") page_size = arg_to_number(args.get("page_size", "50"), arg_name="page_size") prioritized = args.get("prioritized", "") if prioritized: prioritized = argToBoolean(prioritized) state = args.get("state", "") tags = args.get("tags", "") name = args.get("name", "") # Call Vectra API to retrieve entities response = client.list_entities_request( entity_type=entity_type, last_detection_timestamp=last_detection_timestamp, last_modified_timestamp=last_modified_timestamp, ordering=ordering, page=page, # type: ignore page_size=page_size, # type: ignore is_prioritized=prioritized, state=state, tags=tags, name=name, ) count = response.get("count") if count == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching entities for provided filters.", raw_response=response ) entities = response.get("results") # Prepare context data human_readable = get_entity_list_command_hr(entities, page, page_size, count) # type: ignore context = [createContext(entity) for entity in remove_empty_elements(entities)] # type: ignore return CommandResults( outputs_prefix="Vectra.Entity", outputs=context, readable_output=human_readable, raw_response=entities, outputs_key_field=["id", "type"], ) def vectra_entity_describe_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Describes an entity from the Vectra API. Args: client (VectraEventsDetectionsClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the entity. Raises: ValueError: If an invalid entity_type is provided. """ # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id") entity_type = args.get("entity_type", "").lower() # Validate entity_id validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) # Validate entity_type value if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Call Vectra API to retrieve entity entity = client.get_entity_request(entity_id=entity_id, entity_type=entity_type) # type: ignore human_readable = get_entity_get_command_hr(entity) return CommandResults( outputs_prefix="Vectra.Entity", outputs=createContext(remove_empty_elements(entity)), readable_output=human_readable, raw_response=entity, outputs_key_field=["id", "type"], ) def vectra_entity_detection_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Retrieves a list of entity detections from the Vectra API. Args: client (VectraEventsDetectionsClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the entity detections. Raises: ValueError: If an invalid entity_type or state value is provided. """ # Validation for args validate_list_entity_detections_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id") entity_type = args.get("entity_type", "").lower() detection_category = args.get("detection_category") detection_type = args.get("detection_type") detection_name = args.get("detection_name") state = args.get("state", "active") tags = args.get("tags") last_timestamp = arg_to_datetime(args.get("last_timestamp"), arg_name="last_timestamp") if last_timestamp: last_timestamp = last_timestamp.strftime(DATE_FORMAT) # type: ignore page = arg_to_number(args.get("page", "1"), arg_name="page") page_size = arg_to_number(args.get("page_size", "50"), arg_name="page_size") if detection_category: detection_category = DETECTION_CATEGORY_TO_ARG[detection_category] entity = client.get_entity_request(entity_id=entity_id, entity_type=entity_type) detection_set = entity.get("detection_set", []) detections_ids = ",".join([url.split("/")[-1] for url in detection_set]) if detection_set else "" if len(detections_ids) == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching detections for provided entity ID and type.", raw_response={}, ) # Used entity_id and entity_type to list detections response = client.list_detections_request( page=page, page_size=page_size, detection_category=detection_category, detection_type=detection_type, detection_name=detection_name, last_timestamp=last_timestamp, state=state, tags=tags, entity_id=entity_id, entity_type=entity_type, ) count = response.get("count", 0) if count == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching entity detections for provided filters.", raw_response=response, ) detections = response.get("results", {}) # Remove empty elements from the response # Prepare HR hr = get_list_entity_detections_command_hr(detections, page, page_size, count) # Create context context = [createContext(remove_empty_elements(detection)) for detection in detections] # type: ignore return CommandResults( outputs_prefix="Vectra.Entity.Detections", outputs=context, readable_output=hr, raw_response=response, outputs_key_field="id", ) def vectra_detection_describe_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Describes a list of detections for provided detection IDs from the Vectra API. Args: client (VectraEventsDetectionsClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the detections. Raises: ValueError: If an invalid detection_ids or page value is provided. """ # Validation for args validate_detection_describe_args(args) # Get function arguments detection_ids = argToList(args.get("detection_ids"), transform=arg_to_number) detection_ids = [detection_id for detection_id in detection_ids if isinstance(detection_id, int)] page = arg_to_number(args.get("page", "1"), arg_name="page") page_size = arg_to_number(args.get("page_size", "50"), arg_name="page_size") # Call Vectra API to retrieve entities response = client.list_detections_request( ids=",".join([str(detection_id) for detection_id in detection_ids]), state="", page=page, page_size=page_size ) count = response.get("count", 0) if count == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching detections for provided detection ID(s).", raw_response=response, ) detections = response.get("results", {}) # Prepare HR hr = get_list_entity_detections_command_hr(detections, page, page_size, count) # Create context context = [createContext(remove_empty_elements(detection)) for detection in detections] # type: ignore return CommandResults( outputs_prefix="Vectra.Entity.Detections", outputs=context, readable_output=hr, raw_response=response, outputs_key_field="id", ) def vectra_entity_note_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ List entity notes. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_note_list_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() # Call Vectra API to add entity note notes = client.list_entity_note_request(entity_id=entity_id, entity_type=entity_type) # type: ignore notes = remove_empty_elements(notes) if notes: human_readable = get_list_entity_notes_command_hr(notes, entity_id, entity_type) context = [createContext(note) for note in notes] return CommandResults( outputs_prefix="Vectra.Entity.Notes", outputs=context, readable_output=human_readable, raw_response=notes, outputs_key_field=["entity_id", "entity_type", "note_id"], ) else: return CommandResults( outputs={}, readable_output="##### Couldn't find any notes for provided entity.", raw_response=notes ) def vectra_entity_note_add_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Adds a note to an entity in Vectra API. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_note_add_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() note = args.get("note") # Call Vectra API to add entity note notes = client.add_entity_note_request(entity_id=entity_id, entity_type=entity_type, note=note) # type: ignore if notes: notes["note_id"] = notes["id"] notes.update({"entity_id": entity_id, "entity_type": entity_type}) human_readable = "##### The note has been successfully added to the entity." human_readable += f"\nReturned Note ID: **{notes['note_id']}**" return CommandResults( outputs_prefix="Vectra.Entity.Notes", outputs=createContext(remove_empty_elements(notes)), readable_output=human_readable, raw_response=notes, outputs_key_field=["entity_id", "entity_type", "note_id"], ) def vectra_entity_note_update_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Updates a note to an entity in Vectra API. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_note_update_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() note = args.get("note") note_id = arg_to_number(args.get("note_id"), arg_name="note_id", required=True) # Call Vectra API to update entity note notes = client.update_entity_note_request( entity_id=entity_id, # type: ignore entity_type=entity_type, # type: ignore note=note, # type: ignore note_id=note_id, # type: ignore ) if notes: notes["note_id"] = notes["id"] notes.update({"entity_id": entity_id, "entity_type": entity_type}) human_readable = "##### The note has been successfully updated in the entity." return CommandResults( outputs_prefix="Vectra.Entity.Notes", outputs=createContext(remove_empty_elements(notes)), readable_output=human_readable, raw_response=notes, outputs_key_field=["entity_id", "entity_type", "note_id"], ) def vectra_entity_note_remove_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Updates a note to an entity in Vectra API. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_note_remove_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() note_id = arg_to_number(args.get("note_id"), arg_name="note_id", required=True) # Call Vectra API to remove note response = client.remove_entity_note_request( entity_id=entity_id, # type: ignore entity_type=entity_type, # type: ignore note_id=note_id, # type: ignore ) if response.status_code == 204: human_readable = "##### The note has been successfully removed from the entity." else: human_readable = "Something went wrong." return CommandResults(outputs={}, readable_output=human_readable) def vectra_entity_tag_add_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Add tags to an entity. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_tag_add_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() tags = [tag.strip() for tag in argToList(args.get("tags", "")) if isinstance(tag, str) and tag.strip()] # Call Vectra API to get existing entity tags existing_tag_res = client.list_entity_tags_request(entity_id=entity_id, entity_type=entity_type) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) tags = list(dict.fromkeys(tags_resp + tags)) res = existing_tag_res if len(dict.fromkeys(tags_resp)) != len(tags): # Call Vectra API to add entity tags res = client.update_entity_tags_request(entity_id=entity_id, entity_type=entity_type, tags=tags) # type: ignore res_status = res.get("status", "") if not res_status or not isinstance(res_status, str) or res_status.lower() != "success": message = "Something went wrong." if res.get("message"): message += f" Message: {res.get('message')}." raise DemistoException(message) human_readable = "##### Tags have been successfully added to the entity." tags_resp = res.get("tags", []) if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp = f"**{'**, **'.join(tags_resp)}**" human_readable += f"\nUpdated list of tags: {tags_resp}" res["entity_type"] = entity_type res["entity_id"] = entity_id del res["status"] return CommandResults( outputs_prefix="Vectra.Entity.Tags", outputs=createContext(remove_empty_elements(res)), readable_output=human_readable, raw_response=res, outputs_key_field=["tag_id", "entity_type", "entity_id"], ) def vectra_entity_tag_remove_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Removes associated tags for the specified entity using Vectra API. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_tag_add_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() input_tags = [tag.strip() for tag in argToList(args.get("tags", "")) if isinstance(tag, str) and tag.strip()] # Call Vectra API to get existing entity tags existing_tag_res = client.list_entity_tags_request(entity_id=entity_id, entity_type=entity_type) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) # Filtering set of tags from existing tags response with the provide set of input tags updated_tags = [tag_resp.strip() for tag_resp in tags_resp if tag_resp.strip() not in input_tags] res = existing_tag_res # Only update tags if there is any update required with the specified tags if len(dict.fromkeys(tags_resp)) != len(updated_tags): # Call Vectra API to update entity tags res = client.update_entity_tags_request(entity_id=entity_id, entity_type=entity_type, tags=updated_tags) # type: ignore res_status = res.get("status", "") if not res_status or not isinstance(res_status, str) or res_status.lower() != "success": message = "Something went wrong." if res.get("message"): message += f" Message: {res.get('message')}." raise DemistoException(message) human_readable = "##### Specified tags have been successfully removed for the entity." tags_resp = res.get("tags", []) if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp = f"**{'**, **'.join(tags_resp)}**" human_readable += f"\nUpdated list of tags: {tags_resp}" res["entity_type"] = entity_type res["entity_id"] = entity_id del res["status"] return CommandResults( outputs_prefix="Vectra.Entity.Tags", outputs=createContext(remove_empty_elements(res)), readable_output=human_readable, raw_response=res, outputs_key_field=["tag_id", "entity_type", "entity_id"], ) def vectra_entity_tag_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ List tags for an entity. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_entity_tag_list_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id", required=True) entity_type = args.get("entity_type", "").lower() # Call Vectra API to get existing entity tags existing_tag_res = client.list_entity_tags_request(entity_id=entity_id, entity_type=entity_type) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) human_readable = "##### No tags were found for the given entity ID and entity type." if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp = f"**{'**, **'.join(tags_resp)}**" human_readable = f"##### List of tags: {tags_resp}" existing_tag_res["entity_type"] = entity_type existing_tag_res["entity_id"] = entity_id del existing_tag_res["status"] return CommandResults( outputs_prefix="Vectra.Entity.Tags", outputs=createContext(remove_empty_elements(existing_tag_res)), readable_output=human_readable, raw_response=existing_tag_res, outputs_key_field=["tag_id", "entity_type", "entity_id"], ) def vectra_assignment_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ List assignments. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Returns: CommandResults: The command results. """ validate_assignment_list_command_args(args) # Get function arguments entity_ids = args.get("entity_ids") entity_type = args.get("entity_type", "").lower() resolved = args.get("resolved") page = arg_to_number(args.get("page", "1"), arg_name="page") page_size = arg_to_number(args.get("page_size", "50"), arg_name="page_size") assignees = args.get("assignees") resolution = args.get("resolution") # Convert argument to value if resolved: resolved = argToBoolean(resolved) created_after = arg_to_datetime(args.get("created_after"), arg_name="created_after") if created_after: created_after = created_after.strftime(DATE_FORMAT) # type: ignore accounts = None hosts = None if entity_type == "account": accounts = entity_ids elif entity_type == "host": hosts = entity_ids # Call Vectra API for assignment list response = client.list_assignments_request( account_ids=accounts, host_ids=hosts, resolved=resolved, assignees=assignees, resolution=resolution, created_after=created_after, # type: ignore page=page, # type: ignore page_size=page_size, ) # type: ignore response = remove_empty_elements(response) count = response.get("count", 0) assignments = response.get("results", []) if assignments: human_readable, assignments = get_assignment_list_command_hr(assignments, page=page, page_size=page_size, count=count) context = [createContext(assignment) for assignment in assignments] return CommandResults( outputs=context, readable_output=human_readable, raw_response=assignments, outputs_prefix="Vectra.Entity.Assignments", outputs_key_field=["assignment_id"], ) else: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching assignments for provided filters.", raw_response=response, ) def vectra_entity_assignment_add_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Create an assignment for specified entity id. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If detection_ids argument is missing or empty. Returns: CommandResults: The command results. """ # Validate command arguments validate_entity_assignment_add_command_args(args) # Get function arguments entity_id = arg_to_number(args.get("entity_id"), arg_name="entity_id") entity_type = args.get("entity_type", "").lower() user_id = arg_to_number(args.get("user_id"), arg_name="user_id") assign_account_id = None assign_host_id = None if entity_type == "account": assign_account_id = entity_id elif entity_type == "host": assign_host_id = entity_id # Call Vectra API to create an assignment response = client.add_entity_assignment_request( assign_account_id=assign_account_id, assign_host_id=assign_host_id, assign_to_user_id=user_id ) assignment = response.get("assignment", {}) # Update assignment response if assignment: assignment["assignment_id"] = assignment["id"] human_readable = "##### The assignment has been successfully created.\n" human_readable += entity_assignment_add_command_hr(assignment) return CommandResults( outputs_prefix="Vectra.Entity.Assignments", outputs=createContext(remove_empty_elements(assignment)), readable_output=human_readable, raw_response=assignment, outputs_key_field=["assignment_id"], ) def vectra_entity_assignment_update_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Updates an assignment for specified entity id. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If detection_ids argument is missing or empty. Returns: CommandResults: The command results. """ # Validate command arguments validate_entity_assignment_update_command_args(args) # Get function arguments assignment_id = arg_to_number(args.get("assignment_id"), arg_name="assignment_id") user_id = arg_to_number(args.get("user_id"), arg_name="user_id") # Call Vectra API to update an assignment response = client.update_entity_assignment_request(assignment_id=assignment_id, assign_to_user_id=user_id) updated_assignment = response.get("assignment", {}) # Update assignment response if updated_assignment: updated_assignment["assignment_id"] = updated_assignment["id"] human_readable = "##### The assignment has been successfully updated.\n" human_readable += entity_assignment_add_command_hr(updated_assignment) return CommandResults( outputs_prefix="Vectra.Entity.Assignments", outputs=createContext(remove_empty_elements(updated_assignment)), readable_output=human_readable, raw_response=updated_assignment, outputs_key_field=["assignment_id"], ) def vectra_detection_pcap_download_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Download the packet capture (PCAP) file associated with a Vectra detection. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): A dictionary containing the arguments for downloading the PCAP file. - detection_id (str): The ID of the detection associated with the PCAP file. Returns: fileResult: A fileResult object containing the downloaded PCAP file content. """ detection_id = args.get("detection_id") # Validate detection id validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) # Call Vectra API to download detection pcap response = client.download_detection_pcap_request(detection_id=detection_id) content_disposition = response.headers.get("Content-Disposition", "") file_name = content_disposition.split(";")[1].replace("filename=", "").replace('"', "") return fileResult(filename=file_name, data=response.content) def vectra_group_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Retrieves a list of groups. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_group_list_command_args(args) # Get function arguments group_type = args.get("group_type") or "" if group_type: group_type = group_type.lower() importance = args.get("importance") or "" if importance: importance = importance.lower() account_names = argToList(args.get("account_names") or "") domains = argToList(args.get("domains") or "") host_ids = argToList(args.get("host_ids") or "") host_names = argToList(args.get("host_names") or "") ips = argToList(args.get("ips") or "") description = args.get("description") or "" last_modified_timestamp = arg_to_datetime(args.get("last_modified_timestamp"), arg_name="last_modified_timestamp") last_modified_by = args.get("last_modified_by") or "" group_name = args.get("group_name") or "" # Call Vectra API to get groups response = client.list_group_request( group_type=group_type, account_names=account_names, domains=domains, host_ids=host_ids, host_names=host_names, importance=importance, ips=ips, description=description, last_modified_timestamp=last_modified_timestamp, last_modified_by=last_modified_by, group_name=group_name, ) # type: ignore count = response.get("count") if count == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any matching groups for provided filters.", raw_response=response ) groups = response.get("results") # Prepare context data human_readable = get_group_list_command_hr(groups) # type: ignore context = [createContext(group) for group in remove_empty_elements(groups)] # type: ignore return CommandResults( outputs_prefix="Vectra.Group", outputs=context, readable_output=human_readable, raw_response=groups, outputs_key_field=["group_id"], ) def vectra_group_unassign_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Unassign members in Group. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Returns: CommandResults: The command results. """ validate_group_assign_and_unassign_command_args(args) group_id = args.get("group_id") members = args.get("members") # Call to get group details group = client.get_group_request(group_id=group_id) group_type = group.get("type") updated_members = group_members = group.get("members") members_list = argToList(members) removed_members = [] if group_type.lower() == "ip" or group_type.lower() == "domain": # type: ignore for member in members_list: if member in group_members: # type: ignore removed_members.append(member) updated_members.remove(member) # type: ignore elif group_type.lower() == "account": # type: ignore uids = [i.get("uid") for i in group_members] # type: ignore for member in members_list: if member in uids: removed_members.append(member) uids.remove(member) updated_members = uids elif group_type.lower() == "host": # type: ignore ids = [str(i.get("id")) for i in group_members] # type: ignore for member in members_list: if member in ids: removed_members.append(member) ids.remove(member) updated_members = ids if not removed_members: members_list = [re.escape(member) for member in members_list] hr_string = f"##### Member(s) {', '.join(members_list)} do not exist in the group." return CommandResults(readable_output=hr_string) # Call Vectra API to unassign members in group res = client.update_group_members_request(group_id=group_id, members=updated_members) updated_group = remove_empty_elements(res) human_readable = get_group_unassign_and_assign_command_hr( group=updated_group, changed_members=removed_members, assign_flag=False ) return CommandResults( outputs_prefix="Vectra.Group", outputs=createContext(updated_group), readable_output=human_readable, raw_response=updated_group, outputs_key_field=["group_id"], ) def vectra_group_assign_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Assign members in Group. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Returns: CommandResults: The command results. """ validate_group_assign_and_unassign_command_args(args) group_id = args.get("group_id") members = args.get("members") # Call to get group details group = client.get_group_request(group_id=group_id) group_type = group.get("type") updated_members = group_members = group.get("members") members_list = argToList(members) added_members = [] if group_type.lower() == "ip" or group_type.lower() == "domain": # type: ignore for member in members_list: if member not in group_members: # type: ignore added_members.append(member) updated_members.append(member) # type: ignore elif group_type.lower() == "account": # type: ignore uids = [i.get("uid") for i in group_members] # type: ignore for member in members_list: if member not in uids: added_members.append(member) uids.append(member) updated_members = uids elif group_type.lower() == "host": # type: ignore ids = [str(i.get("id")) for i in group_members] # type: ignore for member in members_list: if member not in ids: added_members.append(member) ids.append(member) updated_members = ids if not added_members: members_list = [re.escape(member) for member in members_list] return CommandResults(readable_output=f"##### Member(s) {', '.join(members_list)} are already in the group.") # Call Vectra API to assign members in group res = client.update_group_members_request(group_id=group_id, members=updated_members) updated_group = remove_empty_elements(res) human_readable = get_group_unassign_and_assign_command_hr( group=updated_group, changed_members=added_members, assign_flag=True ) return CommandResults( outputs_prefix="Vectra.Group", outputs=createContext(updated_group), readable_output=human_readable, raw_response=updated_group, outputs_key_field=["group_id"], ) def vectra_entity_detections_mark_asclosed_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Mark the provided entity detections as closed. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If entity_id, entity_type, or close_reason arguments are missing or invalid. Returns: CommandResults: The command results. """ validate_entity_detections_mark_asclosed_command_args(args) # Get function arguments entity_id = args.get("entity_id") entity_type = args.get("entity_type", "").lower() close_reason = args.get("close_reason", "").lower() # Get entity details to retrieve detection IDs response = client.get_entity_request(entity_id=entity_id, entity_type=entity_type) detection_set = response.get("detection_set") detection_ids = [url.split("/")[-1] for url in detection_set] if detection_set else [] hr_string = f"There are no active detections to mark as closed for this entity ID: {entity_id}." if not detection_ids: return CommandResults(readable_output=hr_string) # Call Vectra API to close detections res = client.close_detections_request(detection_ids=detection_ids, reason=close_reason) res_message = res.get("_meta", {}).get("message", "") if res.get("_meta", {}).get("level").lower() == "success" and "successfully closed detections" in res_message.lower(): client.update_detection_status_request(ids_list=detection_ids, status="closed") human_readable = ( f"##### The detections ({', '.join(detection_ids)}) of the provided entity ID have been" f" successfully closed as {close_reason}." ) else: message = "Something went wrong." if res_message: message += f" Message: {res_message}." raise DemistoException(message) return CommandResults(readable_output=human_readable) def vectra_detections_mark_asopen_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Open detection with provided detection IDs. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If detection_ids argument is missing or empty. Returns: CommandResults: The command results. """ # Get function arguments detection_ids = args.get("detection_ids") # Convert string into list detection_ids_list = argToList(detection_ids) # Validate detection_ids if not detection_ids_list: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) all(validate_positive_integer_arg(detection_id, arg_name="detection_ids") for detection_id in detection_ids_list) # Call Vectra API to open detections res = client.open_detections_request(detection_ids_list) res_message = res.get("_meta", {}).get("message", "") if res.get("_meta", {}).get("level", "").lower() == "success" and "successfully re-opened detections" in res_message.lower(): client.update_detection_status_request(ids_list=detection_ids_list, status="open") human_readable = "##### The provided detection IDs have been successfully re-opened." else: message = "Something went wrong." if res_message: message += f" Message: {res_message}." raise DemistoException(message) return CommandResults(readable_output=human_readable) def vectra_detection_tag_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ List tags for a detection. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ detection_id = args.get("detection_id") # Validate detection_id value validate_positive_integer_arg(detection_id, arg_name="detection_id", required=True) # Get function arguments detection_id = arg_to_number(detection_id) # Call Vectra API to get existing detection tags existing_tag_res = client.list_detection_tags_request(detection_id=detection_id) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) human_readable = "##### No tags were found for the given detection ID." if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp_formatted = f"**{', '.join(tags_resp)}**" human_readable = f"##### List of tags: {tags_resp_formatted}" existing_tag_res["detection_id"] = detection_id del existing_tag_res["status"] return CommandResults( outputs_prefix="Vectra.Detection.Tags", outputs=createContext(remove_empty_elements(existing_tag_res)), readable_output=human_readable, raw_response=existing_tag_res, outputs_key_field=["tag_id", "detection_id"], ) def vectra_detection_tag_add_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Add tags to a detection. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_tag_add_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) tags = [tag.strip() for tag in argToList(args.get("tags", "")) if isinstance(tag, str) and tag.strip()] existing_tag_res = client.list_detection_tags_request(detection_id=detection_id) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) tags = list(dict.fromkeys(tags_resp + tags)) res = existing_tag_res if len(dict.fromkeys(tags_resp)) != len(tags): # Call Vectra API to add detection tags res = client.update_detection_tags_request(detection_id=detection_id, tags=tags) # type: ignore res_status = res.get("status", "") if not res_status or not isinstance(res_status, str) or res_status.lower() != "success": message = "Something went wrong." if res.get("message"): message += f" Message: {res.get('message')}." raise DemistoException(message) human_readable = "##### Tags have been successfully added to the detection." tags_resp = res.get("tags", []) if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp = f"**{'**, **'.join(tags_resp)}**" human_readable += f"\nUpdated list of tags: {tags_resp}" res["detection_id"] = detection_id del res["status"] return CommandResults( outputs_prefix="Vectra.Detection.Tags", outputs=createContext(remove_empty_elements(res)), readable_output=human_readable, raw_response=res, outputs_key_field=["tag_id", "detection_id"], ) def vectra_detection_tag_remove_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Removes associated tags for the specified detection using Vectra API. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_tag_add_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) input_tags = [tag.strip() for tag in argToList(args.get("tags", "")) if isinstance(tag, str) and tag.strip()] # Call Vectra API to get existing detection tags existing_tag_res = client.list_detection_tags_request(detection_id=detection_id) # type: ignore existing_tag_res_status = existing_tag_res.get("status", "") if ( not existing_tag_res_status or not isinstance(existing_tag_res_status, str) or existing_tag_res_status.lower() != "success" ): message = "Something went wrong." if existing_tag_res.get("message"): message += f" Message: {existing_tag_res.get('message')}." raise DemistoException(message) tags_resp = existing_tag_res.get("tags", []) # Filtering set of tags from existing tags response with the provide set of input tags updated_tags = [tag.strip() for tag in tags_resp if tag.strip() not in input_tags] res = existing_tag_res # Only update tags if there is any update required with the specified tags if len(dict.fromkeys(tags_resp)) != len(updated_tags): # Call Vectra API to update detection tags res = client.update_detection_tags_request(detection_id=detection_id, tags=updated_tags) # type: ignore res_status = res.get("status", "") if not res_status or not isinstance(res_status, str) or res_status.lower() != "success": message = "Something went wrong." if res.get("message"): message += f" Message: {res.get('message')}." raise DemistoException(message) human_readable = "##### Specified tags have been successfully removed for the detection." tags_resp = res.get("tags", []) if tags_resp and isinstance(tags_resp, list): tags_resp = [tag.strip() for tag in tags_resp if isinstance(tag, str) and tag.strip()] if tags_resp: tags_resp = f"**{'**, **'.join(tags_resp)}**" human_readable += f"\nUpdated list of tags: {tags_resp}" res["detection_id"] = detection_id del res["status"] return CommandResults( outputs_prefix="Vectra.Detection.Tags", outputs=createContext(remove_empty_elements(res)), readable_output=human_readable, raw_response=res, outputs_key_field=["tag_id", "detection_id"], ) def vectra_detection_note_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ List detection notes. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_note_list_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) # Call Vectra API to list detection notes notes = client.list_detection_note_request(detection_id=detection_id) # type: ignore notes = remove_empty_elements(notes) if notes: human_readable = get_list_detection_notes_command_hr(notes, detection_id) context = [createContext(note) for note in notes] return CommandResults( outputs_prefix="Vectra.Detection.Notes", outputs=context, readable_output=human_readable, raw_response=notes, outputs_key_field=["detection_id", "note_id"], ) else: return CommandResults( outputs={}, readable_output="##### Couldn't find any notes for provided detection.", raw_response=notes ) def vectra_detection_note_add_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Adds a note to a detection in Vectra API. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_note_add_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) note = args.get("note") # Call Vectra API to add detection note notes = client.add_detection_note_request(detection_id=detection_id, note=note) # type: ignore if notes: notes["note_id"] = notes["id"] notes.update({"detection_id": detection_id}) human_readable = "##### The note has been successfully added to the detection." human_readable += f"\nReturned Note ID: **{notes['note_id']}**" return CommandResults( outputs_prefix="Vectra.Detection.Notes", outputs=createContext(remove_empty_elements(notes)), readable_output=human_readable, raw_response=notes, outputs_key_field=["detection_id", "note_id"], ) def vectra_detection_note_update_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Updates a note to a detection in Vectra API. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_note_update_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) note = args.get("note") note_id = arg_to_number(args.get("note_id"), arg_name="note_id", required=True) # Call Vectra API to update detection note notes = client.update_detection_note_request( detection_id=detection_id, # type: ignore note=note, # type: ignore note_id=note_id, # type: ignore ) if notes: notes["note_id"] = notes["id"] notes.update({"detection_id": detection_id}) human_readable = "##### The note has been successfully updated in the detection." return CommandResults( outputs_prefix="Vectra.Detection.Notes", outputs=createContext(remove_empty_elements(notes)), readable_output=human_readable, raw_response=notes, outputs_key_field=["detection_id", "note_id"], ) def vectra_detection_note_remove_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Removes a note from a detection Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments provided by the user. Returns: CommandResults: The command results containing the outputs, readable output, raw response, and outputs key field. """ validate_detection_note_remove_command_args(args) # Get function arguments detection_id = arg_to_number(args.get("detection_id"), arg_name="detection_id", required=True) note_id = arg_to_number(args.get("note_id"), arg_name="note_id", required=True) # Call Vectra API to remove note response = client.remove_detection_note_request( detection_id=detection_id, # type: ignore note_id=note_id, # type: ignore ) if response.status_code == 204: human_readable = "##### The note has been successfully removed from the detection." else: human_readable = "Something went wrong." return CommandResults(outputs={}, readable_output=human_readable) def validate_fetch_params(params: dict[str, Any], last_run: dict[str, Any], is_test: bool = False) -> dict[str, Any]: """ Validates the fetch parameters. Args: params (dict[str, Any]): Fetch parameters. last_run (dict[str, Any]): Last run object. is_test (bool): Indicates whether to test the module. Returns: dict[str, Any]: Validated fetch parameters. """ first_fetch = params.get("first_fetch", FIRST_FETCH).strip() first_fetch_time = arg_to_datetime(first_fetch, arg_name="First Fetch Time").strftime(DATE_FORMAT) # type: ignore max_fetch_ = arg_to_number(params.get("max_fetch", MAX_FETCH), arg_name="Max Fetch") entity_types = argToList(params.get("entity_types", DEFAULT_ENTITY_TYPES), transform=lambda x: x.strip()) only_prioritized_detections = argToBoolean(params.get("only_prioritized_detections", DEFAULT_ONLY_PRIORITIZED_DETECTIONS)) only_escalated_detections = argToBoolean(params.get("only_escalated_detections", DEFAULT_ONLY_ESCALATED_DETECTIONS)) if max_fetch_ < 1: # type: ignore raise ValueError(ERRORS["INVALID_MAX_FETCH"].format(max_fetch_)) if max_fetch_ > MAX_FETCH: # type: ignore if is_test: raise ValueError(ERRORS["INVALID_MAX_FETCH"].format(max_fetch_)) else: demisto.debug( f"The max fetch value is {max_fetch_}, " "which is greater than the maximum allowed value of " f"{MAX_FETCH}. Setting it to {MAX_FETCH}." ) max_fetch = min(MAX_FETCH, max_fetch_) # type: ignore valid_entity_types = [] for entity_type in entity_types: if entity_type not in VALID_ENTITY_TYPES and is_test: raise ValueError(ERRORS["INVALID_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPES))) elif entity_type in VALID_ENTITY_TYPES: valid_entity_types.append(entity_type.lower()) else: demisto.debug(f"The entity type: {entity_type} is not valid. Skipping it.") if not valid_entity_types: valid_entity_types = argToList(DEFAULT_ENTITY_TYPES.lower()) if only_escalated_detections and not only_prioritized_detections: detection_statuses = ["escalated"] else: detection_statuses = list(DEFAULT_FETCH_DETECTION_STATUS) if only_prioritized_detections and not only_escalated_detections: unresolved_priority_status = True else: unresolved_priority_status = "" # type: ignore event_timestamp_gte = last_run.get("event_timestamp", first_fetch_time) _from = last_run.get("from", "") valid_entity_types.sort() detection_statuses.sort() params = assign_params( type=",".join(valid_entity_types), investigation_status=",".join(detection_statuses), unresolved_priority=unresolved_priority_status, limit=max_fetch, ordering="id", event_timestamp_gte=event_timestamp_gte, include_info_category=True, size="detailed", include_triaged=False, ) prev_entity_types = last_run.get("selected_types", "") prev_detection_statuses = last_run.get("selected_statuses", "") prev_unresolved_priority = last_run.get("unresolved_priority", "") if ( prev_entity_types == params.get("type", "") and prev_detection_statuses == params.get("investigation_status", "") and prev_unresolved_priority == params.get("unresolved_priority", "") ): params["from"] = _from else: demisto.debug("Change detected in filter configuration parameters. Resetting 'from' API parameter.") remove_nulls_from_dictionary(params) return params def map_severity(urgency_score: int) -> float: """ Maps the severity to the incident severity. Args: urgency_score (int): The urgency score to map. Returns: float: The incident severity. """ if urgency_score > 80: return 4 elif urgency_score > 50: return 3 elif urgency_score > 30: return 2 elif urgency_score > 0: return 1 else: return 0.5 def get_mirroring() -> dict: """ Get the mirroring configuration parameters from the Demisto integration parameters. :return: A dictionary containing the mirroring configuration parameters. :rtype: dict """ params = demisto.params() mirror_direction = params.get("mirror_direction", "None").strip() mirror_tags = params.get("note_tag", "").strip() return { "mirror_direction": MIRROR_DIRECTION.get(mirror_direction), "mirror_instance": demisto.integrationInstance(), "mirror_tags": mirror_tags, } def get_valid_and_dropped_tags(tags: list[str]) -> tuple[list[str], list[str]]: """ Return (valid_tags, dropped_tags) using TAG_REGEX.fullmatch(). Note: does not strip/mutate inputs. If you want trimming, do it before calling. """ valid: list[str] = [] invalid: list[str] = [] for t in tags: if TAGS_REGEX.fullmatch(t): valid.append(t) else: invalid.append(t) if invalid: demisto.debug(f"Dropping invalid tags which contains invalid characters: {invalid}") demisto.debug(f"Provided Valid tags(s): {valid}") return valid, invalid def multiline_logs_for_list(array: list, prefix: str = ""): """ Logs a list of items with a prefix, batched into 50 items per log message. :type array: list :param array: List of items to be logged. :type prefix: str :param prefix: String to be prefixed to the log message. """ for b in batch(array, batch_size=200): demisto.debug(f"{prefix}{b}") def vectra_entity_unresolved_priority_reset_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Updates the priority of an unresolved entity as false. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the updated entity. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type") validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type.lower() not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) result = client.update_entity_unresolved_priority_status_request( entity_id=str(entity_id), entity_type=entity_type.lower(), unresolved_priority="False", ) output_context = {"id": entity_id, "type": entity_type, "unresolved_priority": False} human_readable = "##### The unresolved priority of the provided entity has been successfully changed as 'false'." return CommandResults( outputs_prefix="Vectra.Entity", outputs_key_field=["id", "type"], outputs=output_context, readable_output=human_readable, raw_response=result, ) def vectra_detection_investigation_status_update_command( client: VectraEventsDetectionsClient, args: dict[str, Any] ) -> CommandResults: """ Update the investigation status of the detection by detection IDs. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the updated detection. """ detection_ids = argToList(args.get("detection_ids"), transform=lambda x: x.strip()) investigation_status = args.get("investigation_status") if not detection_ids: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) valid_detection_ids = [] invalid_detection_ids = [] for detection_id in detection_ids: if detection_id is not None and (not detection_id.isdigit() or int(detection_id) <= 0): invalid_detection_ids.append(detection_id) elif detection_id: valid_detection_ids.append(detection_id) if not valid_detection_ids: raise DemistoException(ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", ",".join(invalid_detection_ids))) if invalid_detection_ids: return_warning( message=ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", ",".join(invalid_detection_ids)), exit=(not valid_detection_ids), ) if not investigation_status: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("investigation_status")) if investigation_status and investigation_status.lower() not in [status.lower() for status in VALID_DETECTION_STATUS]: raise ValueError( ERRORS["INVALID_ARG_VALUE"].format( "investigation_status", ", ".join([status.lower() for status in VALID_DETECTION_STATUS]) ) ) result = client.update_detection_status_request( ids_list=valid_detection_ids, status=investigation_status, ) output_context = [{"id": detection_id, "investigation_status": investigation_status} for detection_id in valid_detection_ids] human_readable = ( f"##### The investigation status for provided Detection ID(s) {valid_detection_ids} " f"have been updated as {investigation_status}." ) return CommandResults( outputs_prefix="Vectra.Detection", outputs_key_field="id", outputs=output_context, readable_output=human_readable, raw_response=result, ) def vectra_detection_external_id_update_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Update the external reference ID for the detection by detection ID(s). Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the updated detection. """ detection_ids = argToList(args.get("detection_ids"), transform=lambda x: x.strip()) external_reference_id = args.get("external_reference_id") if not detection_ids: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) valid_detection_ids = [] invalid_detection_ids = [] for detection_id in detection_ids: if detection_id is not None and (not detection_id.isdigit() or int(detection_id) <= 0): invalid_detection_ids.append(detection_id) elif detection_id: valid_detection_ids.append(detection_id) if not valid_detection_ids: raise DemistoException(ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", ",".join(invalid_detection_ids))) if invalid_detection_ids: return_warning( message=ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", ",".join(invalid_detection_ids)), exit=(not valid_detection_ids), ) if not external_reference_id: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("external_reference_id")) result = client.update_detection_external_id_request( ids_list=valid_detection_ids, external_reference_id=external_reference_id, ) output_context = [ {"id": detection_id, "external_reference_id": external_reference_id} for detection_id in valid_detection_ids ] human_readable = ( f"##### The external reference ID for provided Detection ID(s) {valid_detection_ids} " f"have been updated as {external_reference_id}." ) return CommandResults( outputs_prefix="Vectra.Detection", outputs_key_field="id", outputs=output_context, readable_output=human_readable, raw_response=result, ) def vectra_entity_external_id_update_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Updates the external reference ID for the provided entity. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the updated entity. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type") external_reference_id = args.get("external_reference_id") validate_positive_integer_arg(entity_id, arg_name="entity_id", required=True) if not entity_type: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("entity_type")) if entity_type and entity_type.lower() not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) if not external_reference_id: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("external_reference_id")) result = client.update_entity_external_id_request( entity_id=int(entity_id), # type: ignore entity_type=entity_type, external_reference_id=external_reference_id, ) output_context = { "id": entity_id, "type": entity_type, "external_reference_id": external_reference_id, } human_readable = f"##### The external reference ID for provided Entity have been updated as {external_reference_id}." return CommandResults( outputs_prefix="Vectra.Entity", outputs_key_field=["id", "type"], outputs=output_context, readable_output=human_readable, raw_response=result, ) def vectra_investigation_query_send_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Submit an investigation query. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the updated entity. """ query = args.get("query") version = args.get("version") if not query: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("query")) result = client.investigation_query_send( query=query, version=version, ) remove_nulls_from_dictionary(result) human_readable = ( "##### The Vectra investigation has started. You can view the result by executing the below command:\n\n" f"!vectra-investigation-result-get id={result.get('request_id')}" ) return CommandResults( outputs_prefix="Vectra.Investigation", outputs_key_field="request_id", outputs=result, readable_output=human_readable, raw_response=result, ) def vectra_investigation_result_get_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Retrieve the results of a previously submitted investigation using the request ID. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict[str, Any]): Command arguments. Returns: CommandResults: A CommandResults object containing the investigation results. """ request_id = args.get("id") page = arg_to_number(args.get("page", MAX_PAGE)) page_size = arg_to_number(args.get("page_size", MAX_PAGE_SIZE)) validate_positive_integer_arg(page, arg_name="page") validate_positive_integer_arg(page_size, arg_name="page_size") if not request_id: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("id")) result = client.investigation_result_get( request_id=request_id, page=page, # type: ignore page_size=page_size, # type: ignore ) remove_nulls_from_dictionary(result) human_readable = investigation_result_get_command_hr(result) human_readable += tableToMarkdown("Investigation Results Data:", result.get("data", [])) return CommandResults( outputs_prefix="Vectra.Investigation", outputs_key_field="request_id", outputs=result, readable_output=human_readable, raw_response=result, ) def test_module(client: VectraEventsDetectionsClient, params: dict[str, Any]) -> str: """ Tests the connection to the Vectra server. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. params (Dict[str, Any]): Test module parameters. Returns: str: A message indicating the success of the test. """ if argToBoolean(params.get("isFetch", False)): fetch_incidents(client, params, last_run={}, is_test=True) else: client.list_events_detections_request(params=assign_params(limit=1)) return "ok" def fetch_incidents( client: VectraEventsDetectionsClient, params: dict[str, Any], last_run: dict[str, Any], is_test: bool = False, ) -> tuple[list, dict]: """ Fetches incidents from the Vectra Events Detections API. Args: client (VectraEventsDetectionsClient): Vectra client object. params (dict[str, Any]): Fetch incidents parameters. last_run (dict[str, Any]): Last run object. is_test (bool): Indicates whether to test the connection to the Vectra server. Returns: tuple[list, dict]: List of fetched incidents and the last run object. """ fetch_params = validate_fetch_params(params, last_run, is_test) demisto_incidents: list = [] new_last_run = last_run latest_timestamp = last_run.get("event_timestamp", "") only_prioritized_detections = argToBoolean(params.get("only_prioritized_detections", DEFAULT_ONLY_PRIORITIZED_DETECTIONS)) only_escalated_detections = argToBoolean(params.get("only_escalated_detections", DEFAULT_ONLY_ESCALATED_DETECTIONS)) try: response = client.list_events_detections_request(params=fetch_params) except DemistoException as e: demisto.debug(f"Error fetching events detections: {str(e)}") raise e if is_test: return [], {} # Retrieve the already fetched IDs from the last run already_fetched = new_last_run.get("was_fetched", []) events = remove_empty_elements_for_fetch(response.get("events", [])) # Process events and create incidents if events: for event in events: # Extract detection ID detection_id = event.get("detection_id") # Check if the detection is already fetched if detection_id in already_fetched: demisto.debug(f"Skipping event {detection_id} as it is already fetched") continue if ( only_prioritized_detections and only_escalated_detections and not event.get("unresolved_priority", "") and event.get("investigation_status", "").lower() != "escalated" ): demisto.debug(f"Skipping event {detection_id} as it is not escalated and not event prioritized") continue detection_href = event.get("detection_href", "") if detection_href: event["detection_href"] = trim_api_version(detection_href) entity_url = event.get("url", "") if entity_url: event["url"] = trim_api_version(entity_url) if event.get("dst_host") and event.get("dst_host", {}).get("url"): event["dst_host"]["url"] = trim_api_version(event.get("dst_host", {}).get("url")) if event.get("dst_account") and event.get("dst_account", {}).get("url"): event["dst_account"]["url"] = trim_api_version(event.get("dst_account", {}).get("url")) detection_timestamp = event.get("detail", {}).get("first_timestamp", "") occurred_time = detection_timestamp if detection_timestamp else event.get("event_timestamp") if occurred_time and occurred_time[-1].lower() != "z": occurred_time = occurred_time + "Z" # Updating mirroring fields mirroring_fields = get_mirroring() mirroring_fields.update({"mirror_id": detection_id}) event.update(mirroring_fields) # Create incident name detection_type = event.get("detection_type", "") entity_name = event.get("entity_name", "") category = event.get("category", "").title() incident_name = "Vectra RUX:" incident_name += f" {category}" if category else "" incident_name += " -" if detection_type else "" incident_name += f" {detection_type}" if detection_type else "" incident_name += " -" if entity_name else "" incident_name += f" {entity_name}" if entity_name else "" source = event.get("src_account", {}) if source: urgency_score = source.get("urgency_score", 0) else: urgency_score = event.get("src_host", {}).get("urgency_score", 0) demisto_incidents.append( { "name": incident_name, "occurred": occurred_time, "rawJSON": json.dumps(event), "severity": map_severity(urgency_score), } ) already_fetched.append(detection_id) latest_timestamp = event.get("event_timestamp") new_last_run["event_timestamp"] = latest_timestamp new_last_run["from"] = response.get("next_checkpoint") new_last_run["was_fetched"] = already_fetched new_last_run["selected_types"] = fetch_params.get("type", "") new_last_run["selected_statuses"] = fetch_params.get("investigation_status", "") new_last_run["unresolved_priority"] = fetch_params.get("unresolved_priority", "") demisto.debug(f"Fetch params of this interval: {fetch_params}") demisto.debug(f"New last run: {new_last_run}") multiline_logs_for_list(already_fetched, "Ingested Detections: ") return demisto_incidents, new_last_run def vectra_detections_mark_asclosed_command(client: VectraEventsDetectionsClient, args: dict[str, Any]) -> CommandResults: """ Mark the detections as closed by providing IDs of detections and close reason in the argument. Args: client (VectraEventsDetectionsClient): Vectra events detections client object. args (dict[str, Any]): Command arguments. Close reason must be one of the following: benign, remediated. Raises: ValueError: If detection_ids or close_reason arguments are missing or invalid. Returns: CommandResults: The command results. """ detection_ids = argToList(args.get("detection_ids"), transform=lambda x: x.strip()) close_reason = args.get("close_reason", "").lower() # Validate detection ids if not detection_ids: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) for detection_id in detection_ids: if (not detection_id.isdigit()) or int(detection_id) <= 0: raise ValueError(ERRORS["INVALID_INTEGER_VALUE"].format("detection_ids", detection_id)) # Validate close reason if not close_reason: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("close_reason")) if close_reason not in VALID_CLOSE_REASON: raise ValueError(ERRORS["INVALID_ARG_VALUE"].format("close_reason", ", ".join(VALID_CLOSE_REASON))) api_response = client.close_detections_by_ids_request(ids_list=detection_ids, reason=close_reason) if api_response.get("_meta", {}).get("level", "").lower() == "success": client.update_detection_status_request(ids_list=detection_ids, status="closed") readable_output = f"##### The provided detection IDs have been successfully closed as {close_reason}." else: res_message = api_response.get("_meta", {}).get("message", "") message = "Something went wrong." if res_message: message += f" Message: {res_message}." raise DemistoException(message) command_result = CommandResults(readable_output=readable_output, raw_response=api_response) return command_result def vectra_detection_list_command(client: VectraEventsDetectionsClient, args: dict[str, Any]): """ Retrieves a list of entity detections from the Vectra API. Args: client (VectraClient): The Vectra API client. args (Dict[str, Any]): Function arguments. Returns: CommandResults: The command results containing the entity detections. Raises: ValueError: If an invalid entity_type or state value is provided. """ # Validation for args params = validate_list_detections_args(args) # list detections response = client.list_detections_standalone_request(params=params) count = response.get("count", 0) if count == 0: return CommandResults( outputs={}, readable_output="##### Couldn't find any detections for provided filters.", raw_response=response, ) detections = response.get("results", []) # Remove empty elements from the response # Prepare HR hr = get_list_entity_detections_command_hr( detections=detections, page=int(params.get("page")), # type: ignore page_size=int(params.get("page_size")), # type: ignore count=count, ) return CommandResults( outputs_prefix="Vectra.Detection", outputs=remove_empty_elements(detections), readable_output=hr, raw_response=response, outputs_key_field="id", ) def get_modified_remote_data_command(client: VectraEventsDetectionsClient, args: dict) -> GetModifiedRemoteDataResponse: """ Get modified remote data from the Vectra platform and prepare it for mirroring in XSOAR. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict): A dictionary containing the arguments for retrieving modified remote data. Returns: GetModifiedRemoteDataResponse: List of incidents IDs which are modified since the last update. """ command_args = GetModifiedRemoteDataArgs(args) command_last_run_date = dateparser.parse( command_args.last_update, # type: ignore settings={"TIMEZONE": "UTC"}, # type: ignore ).strftime(DATE_FORMAT) modified_entities_ids = [] demisto.debug(f"Last update date of get-modified-remote-data command is {command_last_run_date}.") next_event_timestamp = None next_checkpoint = None page_size = 1000 while True: if next_event_timestamp: command_last_run_date = next_event_timestamp params = assign_params( limit=page_size, event_timestamp_gte=command_last_run_date, include_info_category=True, size="detailed", ordering="id", ) if next_checkpoint: params["from"] = next_checkpoint try: response = client.list_events_detections_request(params=params) except DemistoException as e: demisto.debug(f"Got the error in get-modified-remote-data command: {str(e)}") raise e events_detections = response.get("events", []) if not events_detections: break # Extract detection IDs and remove duplicates modified_entities_ids.extend([str(event.get("detection_id")) for event in events_detections]) # If there is no data on the next page if response.get("remaining_count") == 0: break # Mirroring limit if len(modified_entities_ids) > MAX_MIRRORING_LIMIT: demisto.debug("Max mirroring limit reached.") break next_event_timestamp = events_detections[-1].get("event_timestamp") next_checkpoint = response.get("next_checkpoint") # Filter out None values if there are any. modified_entities_ids: list[str] = list(filter(None, modified_entities_ids)) # type: ignore demisto.debug( f"Performing get-modified-remote-data command. Numbers Detections IDs to update in XSOAR: {len(modified_entities_ids)}" ) demisto.debug(f"Performing get-modified-remote-data command. Detections IDs to update in XSOAR: {modified_entities_ids}") # Filter out any duplicate incident IDs. updated_incident_ids = list(set(modified_entities_ids)) # At max 5,000 incidents should be updated. updated_incident_ids = updated_incident_ids[:5000] return GetModifiedRemoteDataResponse(modified_incident_ids=updated_incident_ids) def get_remote_data_command(client: VectraEventsDetectionsClient, args: dict) -> GetRemoteDataResponse: """ Get remote data for a specific detection from the Vectra platform and prepare it for mirroring in XSOAR. Args: client (VectraEventsDetectionsClient): An instance of the VectraEventsDetectionsClient class. args (Dict): A dictionary containing the arguments for retrieving remote data. - id (str): The ID of the detection to retrieve. - lastUpdate (str): The timestamp of the last update received for this detection. Returns: GetRemoteDataResponse: An object containing the remote incident data and any new entries to return to XSOAR. """ dbot_mirror_id: str = args.get("id") # type: ignore demisto.debug(f"dbot_mirror_id:{dbot_mirror_id}") demisto.debug(f"vectra_detection_id:{dbot_mirror_id}") command_last_run_dt = arg_to_datetime(args.get("lastUpdate"), arg_name="lastUpdate", required=True) command_last_run_timestamp = command_last_run_dt.strftime(DATE_FORMAT) # type: ignore demisto.debug( f"The time when the last time get-remote-data command is called for current incident is {command_last_run_timestamp}." ) # Retrieve the latest entity data from the Vectra platform. params_for_new_append_change_type = assign_params( detection_id=dbot_mirror_id, ordering="-id", limit=1, include_info_category=True, include_triaged=True, size="detailed", change_type="new,append", ) response_for_new_append_change_type = client.list_events_detections_request(params=params_for_new_append_change_type) params_for_other_change_type = assign_params( detection_id=dbot_mirror_id, ordering="-id", limit=1, include_info_category=True, include_triaged=True, size="detailed", change_type="adjust,triage,investigation_status", ) response_for_other_change_type = client.list_events_detections_request(params=params_for_other_change_type) event_for_new_append_chage_type = response_for_new_append_change_type.get("events", []) event_for_other_chage_type = response_for_other_change_type.get("events", []) if event_for_new_append_chage_type: event_for_new_append_chage_type = event_for_new_append_chage_type[0] remove_empty_elements_for_fetch(event_for_new_append_chage_type) if event_for_other_chage_type: event_for_other_chage_type = event_for_other_chage_type[0] remove_empty_elements_for_fetch(event_for_other_chage_type) remote_incident_data: dict = {} if event_for_new_append_chage_type and event_for_other_chage_type: if arg_to_datetime(event_for_new_append_chage_type.get("event_timestamp")) > arg_to_datetime( # type: ignore event_for_other_chage_type.get("event_timestamp") ): remote_incident_data = event_for_new_append_chage_type else: remote_incident_data = update_dict_with_new_dict_values(event_for_new_append_chage_type, event_for_other_chage_type) elif event_for_new_append_chage_type and not event_for_other_chage_type: remote_incident_data = event_for_new_append_chage_type elif not event_for_new_append_chage_type and event_for_other_chage_type: remote_incident_data = event_for_other_chage_type remove_nulls_from_dictionary(remote_incident_data) if not remote_incident_data: return "Incident was not found." # type: ignore detection_href = remote_incident_data.get("detection_href", "") if detection_href: remote_incident_data["detection_href"] = trim_api_version(detection_href) entity_url = remote_incident_data.get("url", "") if entity_url: remote_incident_data["url"] = trim_api_version(entity_url) if remote_incident_data.get("dst_host") and remote_incident_data.get("dst_host", {}).get("url"): remote_incident_data["dst_host"]["url"] = trim_api_version(remote_incident_data.get("dst_host", {}).get("url")) if remote_incident_data.get("dst_account") and remote_incident_data.get("dst_account", {}).get("url"): remote_incident_data["dst_account"]["url"] = trim_api_version(remote_incident_data.get("dst_account", {}).get("url")) detection_timestamp = demisto.get(remote_incident_data, "detail.first_timestamp", "") if detection_timestamp and detection_timestamp[-1].lower() != "z": remote_incident_data["detail"]["first_timestamp"] = detection_timestamp + "Z" event_timestamp = arg_to_datetime(remote_incident_data.get("event_timestamp")) if command_last_run_dt > event_timestamp: # type: ignore demisto.debug(f"Nothing new in the Vectra detection {dbot_mirror_id}.") else: demisto.debug(f"The Vectra detection {dbot_mirror_id} is updated.") new_entries_to_return: list[dict] = [] notes = remote_incident_data.get("notes") if notes: for note in notes: if "[Mirrored From XSOAR]" in note.get("note"): demisto.debug(f"Skipping the note {note.get('id')} as it is mirrored from XSOAR.") continue note_date_modified = arg_to_datetime(note.get("date_modified")) if note_date_modified: if note_date_modified <= command_last_run_dt: # type: ignore demisto.debug( f"Skipping the note {note.get('id')} as it was modified earlier than the command last run timestamp." ) continue else: note_date_created = arg_to_datetime(note.get("date_created"), arg_name="date_created", required=True) if note_date_created <= command_last_run_dt: # type: ignore demisto.debug(f"Skipping the note {note.get('id')} as it is older than the command last run timestamp.") continue new_entries_to_return.append( { "Type": EntryType.NOTE, "Contents": f"[Mirrored From Vectra]\n" f"Added By: {note.get('created_by')}\n" f"Added At: {note.get('date_created')} UTC\n" f"Note: {note.get('note')}", "ContentsFormat": EntryFormat.TEXT, "Note": True, } ) demisto.debug(f"remote_incident_data:{remote_incident_data} and new_entries_to_return:{new_entries_to_return}") return GetRemoteDataResponse(remote_incident_data, new_entries_to_return) def update_remote_system_command(client: VectraEventsDetectionsClient, args: dict, params: dict) -> str: """ Update a remote system based on changes in the XSOAR incident. Args: client (VectraClient): An instance of the VectraClient class. args (Dict): A dictionary containing the arguments required for updating the remote system. params (Dict): A dictionary containing the parameters required for updating the remote system. Returns: str: The ID of the updated remote entity. """ parsed_args = UpdateRemoteSystemArgs(args) # Get remote incident ID remote_incident_id = parsed_args.remote_incident_id mirror_detection_id = parsed_args.data.get("vectraruxdetectionid", "") demisto.debug(f"Remote Incident ID: {remote_incident_id}") detection_status = parsed_args.data.get("vectraruxinvestigationstatus", "") priority_status = parsed_args.data.get("vectraruxentityprioritystatus", "") unresolved_priority = parsed_args.data.get("vectraruxentityunresolvedprioritystatus", "") entity_id = parsed_args.data.get("vectraruxentityid", "") entity_type = parsed_args.data.get("vectraruxentitytype", "") external_reference_id = parsed_args.data.get("vectraruxexternalreferenceid") # Get XSOAR incident id xsoar_incident_id = parsed_args.data.get("id", "") demisto.debug(f"XSOAR Incident ID: {xsoar_incident_id}") # For status and unresolved_priority if detection_status: client.update_detection_status_request(ids_list=[mirror_detection_id], status=detection_status) demisto.debug(f"Updated detection investigation status for detection {mirror_detection_id} to {detection_status}") if priority_status == "Not Prioritized" and not unresolved_priority: client.update_entity_unresolved_priority_status_request( entity_id=entity_id, entity_type=entity_type.lower(), unresolved_priority="False", ) demisto.debug(f"Updated entity {entity_id} priority status to Not Prioritized") if external_reference_id: client.update_detection_external_id_request(ids_list=[mirror_detection_id], external_reference_id=external_reference_id) demisto.debug(f"Updated detection {mirror_detection_id} external reference id to {external_reference_id}") delta = parsed_args.delta new_entries = parsed_args.entries xsoar_tags: list = delta.get("tags") or [] # For notes if new_entries: for entry in new_entries: entry_id = entry.get("id") demisto.debug(f"Sending the entry with ID: {entry_id} and Type: {entry.get('type')}") # Get note content and user entry_content = re.sub(r"([^\n])\n", r"\1\n\n", entry.get("contents", "")) if len(entry_content) > MAX_OUTGOING_NOTE_LIMIT: demisto.info( f"Skipping outgoing mirroring for entity note with XSOAR Incident ID:{xsoar_incident_id}, " "because the note length exceeds 8000 characters." ) entry_user = "" else: entry_user = entry.get("user", "dbot") or "dbot" note_str = ( f"[Mirrored From XSOAR] XSOAR Incident ID: {xsoar_incident_id} \n\n" f"Note: {entry_content} \n\n" f"Added By: {entry_user}" ) # API request for adding notes client.add_note_to_detection_request(detection_id=mirror_detection_id, note=note_str) # For tags res = client.list_detection_tags_request(detection_id=mirror_detection_id) vectra_tags = res.get("tags") or [] if xsoar_tags: xsoar_tags = get_valid_and_dropped_tags(xsoar_tags)[0] demisto.debug(f"Sending the tags: {xsoar_tags}") client.update_detection_tags_request(detection_id=mirror_detection_id, tags=xsoar_tags) # Check if all tags from XSOAR removed elif not xsoar_tags and vectra_tags and "tags" in delta: demisto.debug(f"Sending the tags: {xsoar_tags}") client.update_detection_tags_request(detection_id=mirror_detection_id, tags=xsoar_tags) incident_reopened = False # Check if incident is reopened if delta and delta.get("closingUserId") == "" and delta.get("runStatus") == "": demisto.debug(f"Incident {xsoar_incident_id} is reopened.") incident_reopened = True detection_status_to_set = params.get("detection_status_for_reopen", DEFAULT_DETECTION_STATUS_FOR_REOPEN).lower() if incident_reopened and argToBoolean(params.get("open_detection_on_incident_reopen", False)): client.open_detections_by_ids_request(ids_list=[mirror_detection_id]) client.update_detection_status_request(ids_list=[mirror_detection_id], status=detection_status_to_set) # For Closing notes delta_keys = delta.keys() if "closingUserId" in delta_keys and parsed_args.incident_changed and parsed_args.inc_status == IncidentStatus.DONE: # Check if incident status is Done close_notes = parsed_args.data.get("closeNotes", "") close_reason = parsed_args.data.get("closeReason", "") close_user_id = parsed_args.data.get("closingUserId", "") # close detection detection_close_reason = params.get("close_reason_of_detection", DEFAULT_DETECTION_CLOSE_REASON).lower() if argToBoolean(params.get("close_detection_on_incident_closure", False)): client.update_detection_status_request(ids_list=[mirror_detection_id], status="closed") client.close_detections_by_ids_request(ids_list=[mirror_detection_id], reason=detection_close_reason) if len(close_notes) > MAX_OUTGOING_NOTE_LIMIT: demisto.info( f"Skipping outgoing mirroring for closing notes with XSOAR Incident ID {xsoar_incident_id}, " f"because the note length exceeds {MAX_OUTGOING_NOTE_LIMIT} characters." ) else: closing_note = ( f"[Mirrored From XSOAR] XSOAR Incident ID: {xsoar_incident_id}\n\n" f"Close Reason: {close_reason}\n\n" f"Closed By: {close_user_id}\n\n" f"Close Notes: {close_notes}" ) demisto.debug(f"Closing Comment: {closing_note}") client.add_note_to_detection_request(detection_id=mirror_detection_id, note=closing_note) return remote_incident_id def main(): params = demisto.params() remove_nulls_from_dictionary(params) # get connectivity parameters server_url = params.get("server_url", "").strip() client_id = str(dict_safe_get(params, ["credentials", "identifier"])).strip() client_secret_key = str(dict_safe_get(params, ["credentials", "password"])).strip() verify_certificate = not argToBoolean(params.get("insecure", False)) proxy = argToBoolean(params.get("proxy", False)) command = demisto.command() demisto.debug(f"Command being called is {command}") commands: dict[str, Callable] = { "vectra-user-list": vectra_user_list_command, "vectra-entity-list": vectra_entity_list_command, "vectra-entity-describe": vectra_entity_describe_command, "vectra-entity-detection-list": vectra_entity_detection_list_command, "vectra-detection-describe": vectra_detection_describe_command, "vectra-entity-note-list": vectra_entity_note_list_command, "vectra-entity-note-add": vectra_entity_note_add_command, "vectra-entity-note-update": vectra_entity_note_update_command, "vectra-entity-note-remove": vectra_entity_note_remove_command, "vectra-entity-tag-add": vectra_entity_tag_add_command, "vectra-entity-tag-remove": vectra_entity_tag_remove_command, "vectra-entity-tag-list": vectra_entity_tag_list_command, "vectra-assignment-list": vectra_assignment_list_command, "vectra-entity-assignment-add": vectra_entity_assignment_add_command, "vectra-entity-assignment-update": vectra_entity_assignment_update_command, "vectra-detection-pcap-download": vectra_detection_pcap_download_command, "vectra-group-list": vectra_group_list_command, "vectra-group-assign": vectra_group_assign_command, "vectra-group-unassign": vectra_group_unassign_command, "vectra-entity-detections-mark-asclosed": vectra_entity_detections_mark_asclosed_command, "vectra-detections-mark-asclosed": vectra_detections_mark_asclosed_command, "vectra-detections-mark-asopen": vectra_detections_mark_asopen_command, "vectra-detection-tag-list": vectra_detection_tag_list_command, "vectra-detection-tag-add": vectra_detection_tag_add_command, "vectra-detection-tag-remove": vectra_detection_tag_remove_command, "vectra-detection-note-list": vectra_detection_note_list_command, "vectra-detection-note-add": vectra_detection_note_add_command, "vectra-detection-note-update": vectra_detection_note_update_command, "vectra-detection-note-remove": vectra_detection_note_remove_command, "vectra-entity-unresolved-priority-reset": vectra_entity_unresolved_priority_reset_command, "vectra-detection-investigation-status-update": vectra_detection_investigation_status_update_command, "vectra-detection-external-id-update": vectra_detection_external_id_update_command, "vectra-entity-external-id-update": vectra_entity_external_id_update_command, "vectra-detection-list": vectra_detection_list_command, "vectra-investigation-query-send": vectra_investigation_query_send_command, "vectra-investigation-result-get": vectra_investigation_result_get_command, } try: result = None # Creates vectra client client = VectraEventsDetectionsClient( server_url=server_url, client_id=client_id, client_secret_key=client_secret_key, verify=verify_certificate, proxy=proxy, ) args = demisto.args() if command == "test-module": result = test_module(client, params) elif command == "fetch-incidents": last_run = demisto.getLastRun() incidents, next_run = fetch_incidents(client, params, last_run) demisto.setLastRun(next_run) demisto.debug(f"{len(incidents)} incidents are created successfully in XSOAR.") demisto.incidents(incidents) elif command in commands: # remove nulls from dictionary and trim space from args remove_nulls_from_dictionary(trim_spaces_from_args(args)) result = commands[command](client, args) elif command == "get-modified-remote-data": result = get_modified_remote_data_command(client, args) # type: ignore elif command == "get-remote-data": result = get_remote_data_command(client, args) # type: ignore elif command == "update-remote-system": result = update_remote_system_command(client, args, params) else: raise NotImplementedError(f"Command {command} is not implemented") return_results(result) # Returns either str, CommandResults and a list of CommandResults # Log exceptions and return errors except Exception as e: return_error(f"Failed to execute {command} command.\nError:\n{str(e)}") if __name__ in ("__main__", "__builtin__", "builtins"): # pragma: no cover main()