VectraXDR
This integration allows to create incidents based on Vectra XDR Entities.
Network Security · Vectra XDR
Details
| ID | VectraXDR |
|---|---|
| Provider | Vectra AI |
| Category | Network Security |
| From Version | 6.8.0 |
| Docker Image | demisto/python3:3.12.13.10116658 |
| Supported Modules | Agentix XSIAM |
README
Overview
Vectra XDR pack empowers the SOC to create incidents using Vectra AI’s Attack Signal Intelligence. This integration was integrated and tested with Vectra API v3.3.
This integration supports only cloud instances of Vectra XDR. To configure an instance provide Server URL, Client ID and Client Secret Key.
Use cases
- Fetch entities and their detections from Vectra XDR.
- List and Describe Entities and Detections.
- List, Create, Update, and Resolve Entity Assignments.
- List Assignment Outcomes.
- List, Create, Update, and Delete Entity notes.
- List, Update, and Remove Entity tags.
- List, Assign, and Unassign members in Group.
- Mark and Unmark Entity’s detections as fixed.
- Download PCAP of detection.
Configure Vectra XDR on Cortex XSOAR
- Navigate to Settings > Integrations > Servers & Services.
- Search for Vectra XDR.
- Click Add instance to create and configure a new integration instance.
| 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 |
| Trust any certificate (not secure) | When checked, no SSL certificates check will be done when interacting with the Vectra XDR API. It’s insecure. (Default - unchecked) | False |
| Use system proxy settings | Use the system proxy settings to reach with the Vectra XDR API. | False |
| Fetch incidents | False | |
| Max Fetch | The maximum number of entities 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 entities. 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. |
False |
| Mirroring Direction | The mirroring direction in which to mirror the entities. You can mirror “Incoming” (from Vectra to XSOAR), “Outgoing” (from XSOAR to Vectra), or in both directions. Cortex XSOAR only parameter. | False |
| Re-Fetch closed incidents via mirroring | If selected, new incidents will be created (via Outgoing Mirroring). If not selected, it reopens previously closed incidents (via Incoming Mirroring). Note: This flow is triggered only when the relevant entity is still active and the previously fetched incident is closed. |
False |
| Mirror tag for notes | The tag value should be used to mirror the entity note by adding the same tag in the notes. | False |
| Entity Type | Entity Type(Host, Account). | False |
| Prioritized | Retrieve only prioritize entities based on the configuration on the Vectra platform. If not selected will fetch all entities. | False |
| Tags | Retrieve entities that contain any of the tags specified. Supports comma-separated values. | False |
| Detection Category | Retrieve detections belonging to a specified category. | False |
| Detection Type | Retrieve detections belonging to a specified detection type. | False |
| Specify the numeric value of “Urgency Score” for mapping the Low Incident Severity. | If the entity’s urgency score is equal to or less than the configured threshold, it would be considered as a “Low” Severity Incident. Default is 30. | False |
| Specify the numeric value of “Urgency Score” for mapping the Medium Incident Severity. | If the entity’s urgency score is equal to or less than the configured threshold, it would be considered as a “Medium” Severity Incident.Default is 50. | False |
| Specify the numeric value of “Urgency Score” for mapping the High Incident Severity. | If the entity’s urgency score is equal to or less than the configured threshold, it would be considered as a “High” Severity Incident and if urgency score greater than threshold, it would be considered as a “Critical” Severity Incident. Default is 80. |
False |
| Incident type | False |
- Click Test to validate the URLs, token, and connection.
Configuration for fetching Vectra XDR Entity as an XSOAR Incident
To fetch Vectra XDR Entity follow the next steps:
- Select Fetches incidents.
- Under Classifier, select “N/A”.
- Under Incident type, select “Vectra XDR Entity”.
- Under Mapper (incoming), select “Vectra XDR - Incoming Mapper” for default mapping.
- Enter connection parameters. (Server URL, Client ID & Client Secret Key)
- Select SSL certificate validation and Proxy if required.
- Update “Max Fetch” & “First Fetch Time” based on your requirements.
- Select the Incident Mirroring Direction:
- Incoming - Mirrors changes from the Vectra XDR Entity into the Cortex XSOAR incident.
- Outgoing - Mirrors changes from the Cortex XSOAR incident to the Vectra XDR Entity.
- Incoming And Outgoing - Mirrors changes both Incoming and Outgoing directions on incidents.
- Check the “Re-Fetch closed incidents via mirroring” option if you want to prevent reopening of closed incidents and refetch them via mirroring on modification of an entity.
- 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 XDR. - Provide appropriate values for filtering Entities, such as Entity Type, Prioritization, and Tags. Additionally, specify filters for detections, including Detection Category and Detection Type.
Note: Filters for Entities and Detections are combined using ‘OR’ logic, while filters within the same category(Entity, Detections) are combined using ‘AND’. - Adjust the Urgency Score to categorize Entity severity in Cortex XSOAR. There are three fields for this mapping:
- Input a value for ‘Low’ severity. Scores up to this limit are labelled as Low.
- The next value is for ‘Medium’ severity. Scores up to this limit are labelled as Medium.
- The third value is for ‘High’ severity. Scores up to this limit are labelled as High. Any score above this is marked as ‘Critical’ severity.
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 entity. Updates to existing notes in the XSOAR incident will not be reflected in the Vectra entity.
- New notes from the Vectra entity will be created as notes in the XSOAR incident. Updates to existing notes in the Vectra entity will create new notes in the XSOAR incident.
- If a closed XSOAR incident is tied to a specific entity and new detections for that entity arise or existing detections become active again:
- If “Re-Fetch closed Incidents while Mirroring” checkbox is not selected and “Incoming Mirroring” is enabled, the incident will be automatically reopened.
- If “Re-Fetch closed Incidents while Mirroring” checkbox is selected and “Outgoing Mirroring” is enabled, a new incident will be created for the entity.
- When a XSOAR incident is closed but there are still active detections on the Vectra side, and the entity is subsequently updated:
- If “Re-Fetch closed Incidents while Mirroring” checkbox is not selected and “Incoming Mirroring” is enabled, the corresponding XSOAR incident for that entity will be reopened.
- If “Re-Fetch closed Incidents while Mirroring” checkbox is selected and “Outgoing Mirroring” is enabled, a new incident will be created for the entity.
- The mirroring settings apply only for incidents that are fetched after applying the settings.
- The mirroring is strictly tied to Incident type “Vectra XDR Entity” & Incoming mapper “Vectra XDR - 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 XDR 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 XDR. 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.
Create a custom mapper consisting of the default Vectra XDR mapper
- Go to the settings -> Object setup -> Incidents.
- Navigate to the “Classification and Mapping” tab.
- Select the Mapper “Vectra XDR - Incoming Mapper”.
- Create a copy of that mapper and click on it. (You can rename the mapper.)
- Under the Incident Type dropdown, verify that the type of Mapper is “Vectra XDR Entity”.
- Click on “Choose data path” and map it to the custom field:
- Find the context field you want to map to this incident field on the right side and click on its value.
- Then you will see the path you’ve selected under your newly added field
- Note: You can also type the path manually.
- Click “Save Version”.
- Created mapper will appear in the drop-down for the “Mapper (incoming)” integration instance settings fields.
- Select the newly added mapper at the time of instance configuration.
Create a custom layout consisting of the default Vectra XDR layout
- Go to the settings -> Object setup -> Incidents.
- Navigate to the “Layouts” tab.
- Select the layout “Vectra XDR Entity”.
- Create a copy of that layout and click on it. (You can rename the layout.)
- Select the newly created layout and click on edit.
- To create a new section, drag and drop the “New Section” widget into the layout.
- To add a new field to the layout, navigate to the “Fields and Buttons” section and search for the field. Drag and drop the field widget in the layout.
- Once done, select “Save Version”.
- Navigate to the “Incident Type” tab and select “Vectra XDR Entity” type and detach it.
- Attach the newly created layout.
- Reattach the same “Incident Type” again else this incident type will not receive any new updates.
Note: It is recommended to use out-of-the-box mappers, layout & incident types for better visualization and meaningful mappings. If you are changing any out-of-the-box mappers/layout then it might not render all the fields as per the expectation.
Troubleshooting
Receive Notification on an Incident Fetch Error
The administrator and Cortex XSOAR users on the recipient’s list receive a notification when an integration experiences an incident fetch error. Cortex XSOAR users can select their notification method, such as email, from their user preferences. Refer to Cortex XSOAR 6.13 documentation or Cortex XSOAR 8 Cloud documentation or Cortex XSOAR 8.7 On-prem documentation for more information.
The following are tips for handling issues with mirroring incidents between Vectra XDR and Cortex XSOAR.
| Issue | Recommendation |
|---|---|
| Mirroring is not working. | Open Context Data and search for dbot. Confirm the dbot fields are configured correctly either through the mapper for that specific incident type or using setIncident. Specifically, make sure the integration instance is configured correctly for the mirroring direction (incoming, outgoing, both) - dbotMirrorId, dbotMirrorDirection, dbotMirrorInstance, dbotMirrorTags. |
| Required fields are not getting sent or not visible in UI. | This may be a mapping issue, specifically if you have used a custom mapper make sure you’ve covered all the out of box mapper fields. |
| Notes from XSOAR have not been mirrored in Vectra XDR | Tag is required for mirroring notes from Cortex XSOAR to Vectra XDR. There might be a reason the note is not tagged as the tag needs to be added manually in XSOAR. Click Actions > Tags and add the “note” tag (OR the specific tag name which was set up on Instance Configuration). |
Commands
You can execute these commands from the Cortex XSOAR 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-user-list
Returns a list of users.
Base Command
vectra-user-list
Input
| Argument Name | Description | Required |
|---|---|---|
| username | Filter by username. | Optional |
| role | Filter users with the specified role. Possible values are: Admin, 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.username | 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
{
"Vectra.Entity(val.id && val.id == obj.id)": [
{
"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 and 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
{
"Vectra.Entity(val.id && val.id == obj.id && val.type && val.type == obj.type)": {
"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.num_events | Number | Total number of events related to 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
{
"Vectra.Entity.Detections(val.id && val.id == obj.id)": [
{
"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.num_events | Number | Total number of events related to 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
{
"Vectra.Entity.Detections(val.id && val.id == obj.id)": [
{
"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 | 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
{
"Vectra.Entity.Notes(val.entity_id && val.entity_id == obj.entity_id && val.note_id && val.note_id == obj.note_id)": {
"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-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
{
"Vectra.Detection.Notes(val.detection_id && val.detection_id == obj.detection_id && val.note_id && val.note_id == obj.note_id)": {
"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-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
{
"Vectra.Entity.Notes(val.entity_id && val.entity_id == obj.entity_id && val.note_id && val.note_id == obj.note_id)": {
"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-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
{
"Vectra.Detection.Notes(val.detection_id && val.detection_id == obj.detection_id && val.note_id && val.note_id == obj.note_id)": {
"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-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-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-detections-mark-fixed
Mark detection as fixed with provided detection IDs in argument.
Base Command
vectra-detections-mark-fixed
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-fixed detection_ids=1,2,3
Context Example
{}
Human Readable Output
The provided detection IDs have been successfully marked as fixed
vectra-detections-unmark-fixed
Unmark detection as fixed with provided detection IDs in argument.
Base Command
vectra-detections-unmark-fixed
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-unmark-fixed detection_ids=1,2,3
Context Example
{}
Human Readable Output
The provided detection IDs have been successfully unmarked as fixed
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: host, account. | 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
{
"Vectra.Entity.Tags(val.tag_id && val.tag_id == obj.tag_id && val.entity_type && val.entity_type == obj.entity_type && val.entity_id && val.entity_id == obj.entity_id)": {
"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: host, account. | 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
{
"Vectra.Entity.Tags(val.tag_id && val.tag_id == obj.tag_id && val.entity_type && val.entity_type == obj.entity_type && val.entity_id && val.entity_id == obj.entity_id)": {
"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: host, account. | 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 | Certainnity 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
{
"Vectra.Entity.Assignments(val.assignment_id && val.assignment_id == obj.assignment_id)": {
"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@mail.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@mail.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
{
"Vectra.Entity.Assignments(val.assignment_id && val.assignment_id == obj.assignment_id)": {
"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-entity-assignment-resolve
Resolve an assignment in the entity.
Base Command
vectra-entity-assignment-resolve
Input
| Argument Name | Description | Required |
|---|---|---|
| assignment_id | Specify the ID of the assignment. | Required |
| outcome | Specify the Outcome for resolving an assignment in the entity. The custom outcome is allowed. Possible values are: Benign True Positive, Malicious True Positive, False Positive. | Required |
| note | A note to be added for resolving an assignment in the entity. Default is Updated by XSOAR. | Optional |
| triage_as | Triage rule for resolving an assignment in the entity. | Optional |
| detection_ids | Provide a list of detection IDs separated by commas or a single detection ID. | 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 account that the entity is associated with. |
| Vectra.Entity.Assignments.account_id | Number | ID of the host 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-resolve assignment_id=116 outcome="Custom outcome" detection_ids=1431,1432,1433 triage_as="triage rule"
Context Example
{
"Vectra.Entity.Assignments(val.assignment_id && val.assignment_id == obj.assignment_id)": {
"assigned_by": {
"id": 1,
"username": "test_user@mail.com"
},
"date_assigned": "2023-07-27T12:33:44Z",
"date_resolved": "2023-07-27T12:36:11Z",
"events": [
{
"assignment_id": 116,
"actor": 65,
"event_type": "resolved",
"datetime": "2023-07-27T12:36:11Z",
"context": {
"entity_t_score": 83,
"entity_c_score": 84,
"triage_as": "Triage by XSOAR",
"triaged_detection_ids": [
1432,
1433,
1431
],
"created_rule_ids": [
243,
244,
245
]
}
},
{
"assignment_id": 116,
"actor": 65,
"event_type": "created",
"datetime": "2023-07-27T12:33:44Z",
"context": {
"to": 60,
"entity_t_score": 63,
"entity_c_score": 42
}
}
],
"outcome": {
"id": 6,
"builtin": false,
"user_selectable": true,
"title": "Custom outcome",
"category": "benign_true_positive"
},
"resolved_by": {
"id": 2,
"username": "test_user2@gmail.com"
},
"triaged_detections": [
1432,
1433,
1431
],
"account_id": 100,
"assigned_to": {
"id": 2,
"username": "test_user2@gmail.com"
},
"assignment_id": 116
}
}
Human Readable Output
The assignment has been successfully resolved
vectra-entity-detections-mark-fixed
Mark the detections of the entity as fixed with the provided entity ID in the argument.
Base Command
vectra-entity-detections-mark-fixed
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
There is no context output for this command.
Command Example
!vectra-entity-detections-mark-fixed entity_id=1 entity_type="account"
Context Example
{}
Human Readable Output
The detections (1431, 1432) of the provided entity ID have been successfully marked as fixed
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@mail.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@mail.com"
},
"assignment_id": 214
},
{
"id": 212,
"assigned_by": {
"id": 65,
"username": "test.user4@mail.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@mail.com"
},
"account_id": 108,
"assigned_to": {
"id": 60,
"username": "test.user1@mail.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@mail.com test.user2@mail.com 2023-08-18T10:55:29Z 108 212 test.user4@mail.com test.user1@mail.com 2023-08-18T06:29:56Z test.user4@mail.com 2023-08-18T06:32:09Z 1 Benign True Positive
vectra-assignment-outcome-list
Returns a list of all entity assignment outcomes.
Base Command
vectra-assignment-outcome-list
Input
| Argument Name | Description | 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. Default is 50. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| Vectra.Entity.Assignments.Outcomes.id | Number | ID of the assignment outcome. |
| Vectra.Entity.Assignments.Outcomes.builtin | Boolean | Whether the assignment outcome is builtin or not. |
| Vectra.Entity.Assignments.Outcomes.user_selectable | Boolean | Whether the assignment outcome is user selectable or not. |
| Vectra.Entity.Assignments.Outcomes.title | String | Title of the assignment outcome. |
| Vectra.Entity.Assignments.Outcomes.category | String | Category of the assignment outcome. |
Command Example
!vectra-assignment-outcome-list page=1 page_size=5
Context Example
{
"Vectra": {
"Entity": {
"Assignments": {
"Outcomes": [
{
"builtin": false,
"category": "benign_true_positive",
"id": 7,
"title": "Custom outcome1",
"user_selectable": true
},
{
"builtin": true,
"category": "false_positive",
"id": 3,
"title": "False Positive",
"user_selectable": true
},
{
"builtin": false,
"category": "benign_true_positive",
"id": 6,
"title": "Custom outcome",
"user_selectable": true
},
{
"builtin": true,
"category": "benign_true_positive",
"id": 1,
"title": "Benign True Positive",
"user_selectable": true
},
{
"builtin": true,
"category": "malicious_true_positive",
"id": 2,
"title": "Malicious True Positive",
"user_selectable": true
}
]
}
}
}
}
Human Readable Output
Assignment Outcomes Table (Showing Page 1 out of 1)
ID Title Category Built IN User Selectable 1 Benign True Positive benign_true_positive true true 2 Malicious True Positive malicious_true_positive true true 3 False Positive false_positive true true 6 Custom outcome benign_true_positive false true 7 Custom outcome1 benign_true_positive false true
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@mail.com",
"date_created": "2023-08-25T07:09:08Z",
"entity_id": 107,
"entity_type": "account",
"id": 1070,
"modified_by": "test_user@mail.com",
"note": "From XSOAR",
"note_id": 1070
},
{
"created_by": "test_user@mail.com",
"date_created": "2023-08-25T07:08:58Z",
"entity_id": 107,
"entity_type": "account",
"id": 1069,
"modified_by": "test_user@mail.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@mail.com 2023-08-25T07:09:08Z test_user@mail.com 2023-08-25T08:10:08Z 1069 Test note test_user@mail.com 2023-08-25T07:08:58Z test_user@mail.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-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-accout-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-accout-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-accout-test domain xsoar-group-accout-test *.domain1.net, *.domain2.com, *.domain3.com, *.domain4.com, *.domain5.com 2023-09-04T06:30:01Z
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-accout-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-accout-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-accout-test domain xsoar-group-accout-test *.domain1.net, *.domain2.com, *.domain3.com 2023-09-04T07: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-asclosed
Mark detections as close 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 | Specify 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=1,2,3 close_reason=benign
Human Readable Output
The provided detection IDs 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@mail.com",
"date_created": "2023-08-25T07:09:08Z",
"detection_id": 1,
"id": 1070,
"modified_by": "test_user@mail.com",
"note": "From XSOAR",
"note_id": 1070
},
{
"created_by": "test_user@mail.com",
"date_created": "2023-08-25T07:08:58Z",
"detection_id": 1,
"id": 1069,
"modified_by": "test_user@mail.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
}
]
}
}
}
vectra-entity-reset-fetch
Resets the given entity to refetch incidents.
Base Command
vectra-entity-reset-fetch
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
There is no context output for this command.
Command example
!vectra-entity-reset-fetch entity_id=1 entity_type=host
Human Readable Output
Reset fetch status for 1-host
Configuration parameters
server_url— Server URL (required)credentials— Client ID (required)insecure— Trust any certificate (not secure)proxy— Use system proxy settingsisFetch— Fetch incidentsmax_fetch— Max Fetchfirst_fetch— First Fetch Timemirror_direction— Mirroring Directionrefetch_closed_incidents— Re-Fetch closed incidents via mirroringnote_tag— Mirror tag for notesentity_type— Entity Typeis_prioritized— Prioritizedtags— Tagsdetection_category— Detection Categorydetection_type— Detection Typeurgency_score_low_threshold— Specify the numeric value of "Urgency Score" for mapping the Low Incident Severity.urgency_score_medium_threshold— Specify the numeric value of "Urgency Score" for mapping the Medium Incident Severity.urgency_score_high_threshold— Specify the numeric value of "Urgency Score" for mapping the High Incident Severity.incidentType— Incident typeincidentFetchInterval— Incidents Fetch Interval
Commands (35)
-
vectra-assignment-listReturns a list of all assignments.
-
vectra-assignment-outcome-listReturns a list of all entity assignment outcomes.
-
vectra-detection-describeReturns a list of detections for the specified detection ID(s).
-
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 close with provided detection IDs in the argument.
-
vectra-detections-mark-asopenOpen detections with provided detection IDs in the argument.
-
vectra-detections-mark-fixedMark detection as fixed with provided detection IDs in argument.
-
vectra-detections-unmark-fixedUnmark detection as fixed with provided detection IDs in argument.
-
vectra-entity-assignment-addAdd an assignment for the entity.
-
vectra-entity-assignment-resolveResolve an assignment in 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-detections-mark-fixedMark the detections of the entity as fixed with the provided entity ID in the argument.
-
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-reset-fetchResets the given entity to refetch incidents.
-
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-group-assignAssign members to the specified group.
-
vectra-group-listReturns a list of all groups.
-
vectra-group-unassignUnassign members from the specified group.
-
vectra-user-listReturns a list of users.
import copy from collections.abc import Callable import requests.auth import urllib3 from urllib.parse import urlparse, parse_qs from datetime import datetime from requests.models import Response 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] + list(range(500, 600)) OK_CODES = (200, 201, 204, 401) MAX_RETRIES = 3 BACKOFF_FACTOR = 15 FIRST_FETCH = "1 hour" 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_CLOSE_REASON = ["benign", "remediated"] DEFAULT_URGENCY_SCORE_LOW_THRESHOLD = 30 DEFAULT_URGENCY_SCORE_MEDIUM_THRESHOLD = 50 DEFAULT_URGENCY_SCORE_HIGH_THRESHOLD = 80 MAX_MIRRORING_LIMIT = 5000 MAX_OUTGOING_NOTE_LIMIT = 8000 PACK_VERSION = get_pack_version() or "1.2.0" UTM_PIVOT = f"?pivot=Vectra-XSOAR-{PACK_VERSION}" EMPTY_ASSIGNMENT = [ { "id": "", "date_assigned": "", "date_resolved": "", "assigned_to": {"username": ""}, "resolved_by": {"username": ""}, "assigned_by": {"username": ""}, "outcome": {"title": ""}, } ] DETECTION_CATEGORY_TO_ARG = { "Command & Control": "command", "Botnet": "botnet", "Reconnaissance": "reconnaissance", "Lateral Movement": "lateral", "Exfiltration": "exfiltration", "Info": "info", } 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": "'{}' 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.", } ENDPOINTS = { "AUTH_ENDPOINT": "/oauth2/token", "USER_ENDPOINT": "/api/v3.3/users", "GROUP_ENDPOINT": "/api/v3.3/groups", "ENTITY_ENDPOINT": "/api/v3.3/entities", "DETECTION_ENDPOINT": "/api/v3.3/detections", "ADD_AND_LIST_DETECTION_NOTE_ENDPOINT": "/api/v3.3/detections/{}/notes", "ADD_AND_LIST_ENTITY_NOTE_ENDPOINT": "/api/v3.3/entities/{}/notes", "UPDATE_AND_REMOVE_DETECTION_NOTE_ENDPOINT": "/api/v3.3/detections/{}/notes/{}", "UPDATE_AND_REMOVE_ENTITY_NOTE_ENDPOINT": "/api/v3.3/entities/{}/notes/{}", "ENTITY_TAG_ENDPOINT": "/api/v3.3/tagging/entity/{}", "ASSIGNMENT_ENDPOINT": "/api/v3.3/assignments", "UPDATE_ASSIGNMENT_ENDPOINT": "/api/v3.3/assignments/{}", "RESOLVE_ASSIGNMENT_ENDPOINT": "/api/v3.3/assignments/{}/resolve", "ASSIGNMENT_OUTCOME_ENDPOINT": "/api/v3.3/assignment_outcomes/", "DOWNLOAD_DETECTION_PCAP": "/api/v3.3/detections/{}/pcap", "DETECTION_CLOSE_ENDPOINT": "/api/v3.4/detections/close", "DETECTION_OPEN_ENDPOINT": "/api/v3.4/detections/open", "DETECTION_TAG_ENDPOINT": "/api/v3.3/tagging/detection/{}", } USER_AGENT = f"VectraXDR-XSOAR-{PACK_VERSION}" PAGE_SIZE = 200 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) """ CLIENT CLASS """ class VectraClient(BaseClient): """ Client class to interact with the Vectra 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( ERRORS["INVALID_OBJECT"].format(response_type, 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_users_request(self, username: Optional[str], role=Optional[str], last_login_timestamp=Optional[datetime]) -> dict: """ List users. Args: username (Optional[str]): The optional username to filter with (default: None). role (Optional[str]): The optional user role to filter with (default: None). last_login_timestamp (Optional[str]): Filter users after the specified last login timestamp (default: None). Returns: Dict: Response from the API containing the users. """ params = assign_params(username=username, 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: Optional[datetime] = None, last_detection_timestamp: Optional[datetime] = 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"], 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"], 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: Optional[datetime] = 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 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_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, data=data, 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), 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, 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), 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 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 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 mark_or_unmark_detection_fixed_request(self, detection_ids: List[str], mark: str) -> dict: """ Mark or unmark detections as fixed. Args: detection_ids (List[str]): List of detection IDs to mark or unmark as fixed. mark (str): True to mark as fixed, False to unmark as fixed. Returns: Dict: Response from the API. Raises: ValueError: If detection_ids is empty. """ data = {"detectionIdList": detection_ids, "mark_as_fixed": mark} res = self.http_request(method="PATCH", url_suffix=ENDPOINTS["DETECTION_ENDPOINT"], json_data=data, 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: Optional[int] = None, assign_host_id: Optional[int] = None, assign_account_id: Optional[int] = 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"], data=body, response_type="json") return res def update_entity_assignment_request( self, assign_to_user_id: Optional[int] = None, assignment_id: Optional[int] = None ) -> dict: """ Send a request to update an existing entity assignment. Args: assign_to_user_id (Optional[int], optional): The ID of the user to whom the entity will be reassigned. Defaults to None. assignment_id (Optional[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), data=body, response_type="json", ) return res def resolve_entity_assignment_request( self, assignment_id: Optional[int] = None, outcome: int = None, note: str = "Updated by XSOAR.", triage_as: str = None, detection_ids=None, ) -> dict: """ Resolves an entity assignment. Args: assignment_id (str, optional): The ID of the assignment to resolve. outcome (int, optional): The outcome of the resolved assignment. note (str, optional): A note to add to the resolution (default is "Updated by XSOAR."). triage_as (str, optional): The triage status to set for the resolved assignment. detection_ids (List[str], optional): The IDs of detections associated with the assignment. Returns: dict: Response from the API. """ body = assign_params(outcome=outcome, note=note, triage_as=triage_as, detection_ids=detection_ids) res = self.http_request( method="PUT", url_suffix=ENDPOINTS["RESOLVE_ASSIGNMENT_ENDPOINT"].format(assignment_id), data=body, response_type="json", ) return res def list_assignment_outcomes_request(self, page: int = None, page_size: int = None) -> dict: """ Send a request to retrieve a list of assignment outcomes. Args: 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(page=page, page_size=page_size) res = self.http_request( method="GET", url_suffix=ENDPOINTS["ASSIGNMENT_OUTCOME_ENDPOINT"], params=params, 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: Optional[datetime], 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 (Optional[datetime]): 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_tags_request(self, detection_id: int) -> dict: """ List tags for the specified detection. Args: detection_id (int): The ID of the detection to list tags for. Returns: Dict: Response from the API containing the tags. """ res = self.http_request( method="GET", url_suffix=ENDPOINTS["DETECTION_TAG_ENDPOINT"].format(detection_id), response_type="json" ) return res def update_detection_tags_request(self, detection_id: int = None, tags: List = None) -> dict: """ Update tags to a detection. Args: detection_id (int): The ID of the detection to add the tags to. tags (List): Tags to set for detection. Returns: Dict: Response from the API containing the updated tags. """ data = {"tags": tags} res = self.http_request( method="PATCH", url_suffix=ENDPOINTS["DETECTION_TAG_ENDPOINT"].format(detection_id), json_data=data, response_type="json", ) return res """ HELPER FUNCTIONS """ def validate_urgency_score(urgency_score: str, score_name: str) -> Optional[int]: """ 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_configuration_parameters(params: dict[str, Any]): """ Validates the configuration parameters provided. Args: params (Dict[str, Any]): A dictionary containing the configuration parameters. Raises: ValueError: Raised when required parameters are missing or have invalid values. """ fetch_time = params.get("first_fetch") max_fetch = params.get("max_fetch") urgency_score = params.get("urgency_score") urgency_score_low_threshold = params.get("urgency_score_low_threshold", "30").strip() urgency_score_medium_threshold = params.get("urgency_score_medium_threshold", "50").strip() urgency_score_high_threshold = params.get("urgency_score_high_threshold", "80").strip() # Validate empty values if not fetch_time: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("First Fetch Time")) if not max_fetch: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("Max Fetch")) # validate max_fetch max_fetch = arg_to_number(max_fetch, arg_name="Max Fetch", required=True) if not 1 <= max_fetch <= 200: # type: ignore raise ValueError(ERRORS["INVALID_MAX_FETCH"].format(max_fetch)) # validate first_fetch parameter arg_to_datetime(fetch_time, arg_name="first_fetch", required=True) # validate urgency score if urgency_score: validate_urgency_score(urgency_score, "urgency_score") # type: ignore uslt = validate_urgency_score(urgency_score_low_threshold, "urgency_score_low_threshold") # type: ignore usmt = validate_urgency_score(urgency_score_medium_threshold, "urgency_score_medium_threshold") # type: ignore usht = validate_urgency_score(urgency_score_high_threshold, "urgency_score_high_threshold") # type: ignore # validating urgency score threshold if not (uslt < usmt < usht): # type: ignore raise ValueError(ERRORS["INVALID_URGENCY_SCORE_THRESHOLD"]) def validate_positive_integer_arg(value: Optional[Any], 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)) return True 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[Any, 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[Any, 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")) 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[Any, 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_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_entity_note_add_command_args(args: dict[Any, 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_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_entity_note_update_command_args(args: dict[Any, 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_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_entity_note_remove_command_args(args: dict[Any, 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_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 validate_entity_tag_add_command_args(args: dict[Any, 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[Any, 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_detections_mark_and_unmark_args(detection_ids: List): """ Validate the arguments for mark and unmark detections as fixed. Args: detection_ids (List[int]): The list of detection IDs. Raises: ValueError: If the detection IDs are empty or contain invalid values. """ # Validate detection_ids if not detection_ids: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("detection_ids")) all(validate_positive_integer_arg(i, arg_name="detection_ids") for i in detection_ids) 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_entity_assignment_resolve_command_args(args): """ Validate the arguments provided for resolving an entity assignment. Args: args (Dict): A dictionary containing the arguments for resolving an entity assignment. Raises: ValueError: If the provided assignment_id is not a positive integer. ValueError: If the outcome argument is missing. ValueError: If detection_ids are provided without triage_as. """ outcome = args.get("outcome") assignment_id = args.get("assignment_id") triage_as = args.get("triage_as") detection_ids = args.get("detection_ids") # Validate assignment_id value validate_positive_integer_arg(assignment_id, arg_name="assignment_id", required=True) # Validate outcome value if not outcome: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("outcome")) if detection_ids and not triage_as: raise ValueError(ERRORS["TRIAGE_AS_REQUIRED_WITH_DETECTION_IDS"]) def validate_entity_detections_mark_fix_command_args(args): """ Validate the arguments provided for marking entity detections as fixed. Args: args (Dict): A dictionary containing the arguments for marking entity detections as fixed. Raises: ValueError: If the provided entity_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") 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))) def validate_group_list_command_args(args: dict[Any, 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 urgency_score_to_severity(entity_urgency_score: Optional[int], params: dict[str, Any]) -> int: """ Maps the urgency score to severity levels. Demisto's severity levels are 4 - Critical, 3 - High, 2 - Medium, 1 - Low Args: entity_urgency_score(int): The urgency score to be mapped. params (Dict[str, Any]): Fetch incidents parameters. Returns severity(int):The corresponding severity level based on the urgency score. """ try: urgency_score_lt = validate_urgency_score( urgency_score=params.get("urgency_score_low_threshold", "30").strip(), score_name="urgency_score_lt" ) except ValueError: demisto.error("Provided urgency_score_low_threshold is not a valid number. Falling back to default threshold value 30.") urgency_score_lt = DEFAULT_URGENCY_SCORE_LOW_THRESHOLD try: urgency_score_mt = validate_urgency_score( urgency_score=params.get("urgency_score_medium_threshold", "50").strip(), score_name="urgency_score_mt" ) except ValueError: demisto.error( "Provided urgency_score_medium_threshold is not a valid number. Falling back to default threshold value 50." ) urgency_score_mt = DEFAULT_URGENCY_SCORE_MEDIUM_THRESHOLD try: urgency_score_ht = validate_urgency_score( urgency_score=params.get("urgency_score_high_threshold", "80").strip(), score_name="urgency_score_ht" ) except ValueError: demisto.error("Provided urgency_score_high_threshold is not a valid number. Falling back to default threshold value 80.") urgency_score_ht = DEFAULT_URGENCY_SCORE_HIGH_THRESHOLD if entity_urgency_score > urgency_score_ht: # type: ignore return SEVERITY["critical"] elif entity_urgency_score > urgency_score_mt: # type: ignore return SEVERITY["high"] elif entity_urgency_score > urgency_score_lt: # type: ignore return SEVERITY["medium"] else: return SEVERITY["low"] 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) if len(val_list) > 1: val_list = [item for item in val_list if item.strip()] args[key] = ",".join(val_list) return args 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.3' portion from a URL. Args: url (str): The URL to trim. Returns: str: The trimmed URL. """ api_versions = ["/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 def get_mirroring(): """ Get the mirroring configuration parameters from the Demisto integration parameters. Returns: dict: A dictionary containing the mirroring configuration parameters. """ 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_tags": mirror_tags, "mirror_instance": demisto.integrationInstance(), } def reopen_in_xsoar(entries: list, entity_id_type: list): """Reopen the XSOAR incident for the given entity. Args: entries (list): List of entries where the reopening entry will be appended. entity_id_type (list): Indicates the entity ID and type. """ demisto.debug(f"Reopening the incident with remote entity ID: {entity_id_type}.") entries.append({"Type": EntryType.NOTE, "Contents": {"dbotIncidentReopen": True}, "ContentsFormat": EntryFormat.JSON}) 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("username"), "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: Optional[int], page_size: Optional[int], 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("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: Optional[int], page_size: Optional[int], 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_assignment_outcome_list_command_hr(outcomes: dict, page: Optional[int], page_size: Optional[int], count: int): """ Returns the human-readable output for the assignment outcome list. Args: outcomes(Dict): The assignment outcomes list dictionary. page (int): The current page number. page_size (int): The page size. count (int): The total count of outcomes. Returns: str: The human-readable output. dict: Outcomes dictionary """ hr_dict = [] for outcome in outcomes: outcome["outcome_id"] = outcome["id"] hr_dict.append( { "Outcome ID": outcome.get("id"), "Built IN": outcome.get("builtin"), "User Selectable": outcome.get("user_selectable"), "Title": outcome.get("title"), "Category": outcome.get("category"), } ) pages = calc_pages(per_page_count=page_size, total_count=count) # type: ignore # Prepare human-readable output table human_readable = tableToMarkdown( f"Assignment Outcomes Table (Showing Page {page} out of {pages})", hr_dict, ["Outcome ID", "Title", "Category", "Built IN", "User Selectable"], removeNull=True, ) return human_readable, outcomes def get_list_entity_notes_command_hr(notes: dict, entity_id: Optional[int], entity_type: str) -> str: """ Returns the human-readable output for the entity notes. Args: notes (Dict): The assignment details dictionary. entity_id (Optional[int]): 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_list_detection_notes_command_hr(notes: dict, detection_id: Optional[int]) -> str: """ Returns the human-readable output for the detection notes. Args: notes (Dict): list of detection notes. detection_id (Optional[int]): 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 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 add_refetch_id_to_integration_context(entity_id: str, entity_type: str): """ Adds the entity ID and type to the integration context. Args: entity_id (str): The ID of the entity. entity_type (str): The type of the entity. """ if not entity_id or not entity_type: raise ValueError("Both 'entity_id' and 'entity_type' arguments are required.") if entity_type not in VALID_ENTITY_TYPE: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("entity_type", ", ".join(VALID_ENTITY_TYPE))) # Get current integration context integration_context = get_integration_context() # Initialize refetch_ids if it doesn't exist if "refetch_ids" not in integration_context: integration_context["refetch_ids"] = [] # Create the refetch entry refetch_entry = f"{entity_id}-{entity_type}" demisto.debug(f"Adding entity id to the integration context: {refetch_entry}") # Add to refetch list if not already present if refetch_entry not in integration_context["refetch_ids"]: integration_context["refetch_ids"].append(refetch_entry) # Update integration context set_integration_context(integration_context) demisto.debug(f"Updated entity ids list in the integration context: {integration_context['refetch_ids']}") def validate_entity_detections_mark_asclosed_command_args(args: dict[str, Any]): """ Validate the arguments provided for marking entity detections as closed. Args: args (Dict): A dictionary containing the arguments for marking entity detections as closed. Raises: ValueError: If the provided entity_id is not a positive integer. ValueError: If the entity_type is missing or not one of the valid types. ValueError: If the close_reason is missing or not one of the valid values. """ entity_id = args.get("entity_id") entity_type = args.get("entity_type") close_reason = args.get("close_reason") 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 close_reason: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("close_reason")) if close_reason and close_reason.lower() not in VALID_CLOSE_REASON: raise ValueError(ERRORS["INVALID_COMMAND_ARG_VALUE"].format("close_reason", ", ".join(VALID_CLOSE_REASON))) def validate_detections_mark_asclosed_args(args: Dict[str, Any]): """ Validate the arguments for marking detections as closed. Args: args (Dict[str, Any]): The arguments for marking detections as closed. close_reason (str): The close reason. Raises: ValueError: If the detection IDs are empty or contain invalid values. ValueError: If the close_reason is missing or not one of the valid values. """ detection_ids = args.get("detection_ids") close_reason = args.get("close_reason", "").lower() # Convert string into list detection_ids = argToList(detection_ids) # Validate detection_ids if not detection_ids: 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) # Validate close_reason if not close_reason: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("close_reason")) if close_reason and close_reason.lower() 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: dict[Any, Any]): """ Validates the arguments provided for the detection tag add 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) tags = argToList(args.get("tags", "")) valid_tags = [tag.strip() for tag in tags if isinstance(tag, str) and tag.strip()] # Validate Tags value if not valid_tags: raise ValueError(ERRORS["REQUIRED_ARGUMENT"].format("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 """ COMMAND FUNCTIONS """ def fetch_incidents(client: VectraClient, params: dict[str, Any]) -> List: """ Fetches incidents from the Vectra server. Args: client (VectraClient): Vectra client object. params (Dict[str, Any]): Fetch incidents parameters. Returns: List[Dict[str, Any]]: List of fetched incidents. """ # Fetch parameters from the integration configuration first_fetch_time = params.get("first_fetch", FIRST_FETCH).strip() max_fetch_ = arg_to_number(params.get("max_fetch", str(MAX_PAGE_SIZE)).strip(), arg_name="Max Fetch", required=True) # Ensure max_fetch is not greater than 200 if max_fetch_ < 1: # type: ignore raise ValueError(ERRORS["INVALID_MAX_FETCH"].format(max_fetch_)) if max_fetch_ > 200: # type: ignore demisto.debug( f"The max fetch value is {max_fetch_}, which is greater than the maximum allowed value of 200. Setting it to 200." ) max_fetch = min(200, max_fetch_) # type: ignore # Default page is 1 page = MAX_PAGE # Retrieve last run data from the previous execution last_run = json.loads(demisto.getLastRun().get("value", "{}")) # Retrieve and preprocess entity_type parameter entity_type = params.get("entity_type", "") if isinstance(entity_type, list): entity_type = ",".join(entity_type).lower() else: entity_type = entity_type.lower() # Process is_prioritized parameter is_prioritized = params.get("is_prioritized", "") if not is_prioritized: is_prioritized = None else: is_prioritized = argToBoolean(is_prioritized.lower()) # Fetch and process tags parameter tags = params.get("tags", "").strip() # Fetch detection_category and detection_type parameters detection_category = params.get("detection_category", "") detection_type = params.get("detection_type", "").strip() if not last_run: # If it's the first time running # Create a new last run object with the first fetch time new_last_run = { "time": arg_to_datetime(first_fetch_time, arg_name="First Fetch Time").strftime(DATE_FORMAT) # type: ignore } demisto.debug(f"No last run object found, creating new last run object with value: {json.dumps(new_last_run)}") else: # Use the last run object from the previous run new_last_run = last_run demisto.debug("Using the last run object obtained from the previous run.") # Retrieve the from_timestamp from the new last run object from_timestamp = new_last_run.get("time") # Initialize the demisto_incidents list demisto_incidents: List = [] # Retrieve the page number from the last run object or use the default value of 1 next_url_last_run = last_run.get("next_url") if next_url_last_run: # Parse the URL parsed_url = urlparse(next_url_last_run) # Extract the query parameters query_params = parse_qs(parsed_url.query) page = arg_to_number(query_params.get("page", [""])[0], arg_name="page") # type: ignore max_fetch = arg_to_number(query_params.get("page_size", [""])[0], arg_name="max_fetch") entity_type = query_params.get("type", [""])[0] is_prioritized = query_params.get("is_prioritized", [""])[0] if is_prioritized: is_prioritized = argToBoolean(is_prioritized) from_timestamp = query_params.get("last_modified_timestamp_gte", [""])[0] tags = query_params.get("tags", [""])[0] try: # Fetch the entities list from the server using the provided parameters response = client.list_entities_request( page=page, page_size=max_fetch, # type: ignore entity_type=entity_type, # type: ignore is_prioritized=is_prioritized, last_modified_timestamp=from_timestamp, # type: ignore tags=tags, ordering="last_modified_timestamp", ) except DemistoException as e: # If the status code is 404 and the message is invalid page number, then return the empty list of incidents. if str(e.res) == "<Response [404]>" and ( e.res.status_code == 404 and ERRORS["INVALID_PAGE_RESPONSE"] in str(e.message).lower() ): demisto.debug(f"Returning 0 incidents in fetch incidents due to the end of page: {str(e.message)}") return demisto_incidents raise e # Retrieve the next page URL for pagination next_url = response.get("next") # Get the entities from the response entities = response.get("results") # Retrieve the already fetched IDs from the last run already_fetched = last_run.get("already_fetched", []) # Get the refetch IDs from the integration context integration_context = get_integration_context() refetch_ids = integration_context.get("refetch_ids", []) # Remove refetch IDs from already_fetched and create the removed_ids list removed_ids = [] for refetch_id in refetch_ids: if refetch_id in already_fetched: removed_ids.append(refetch_id) already_fetched.remove(refetch_id) demisto.debug(f"Removed closed entity ids from last run checkpoint: {removed_ids}") # Update integration context with empty refetch_ids integration_context["refetch_ids"] = [] # Set the updated integration context set_integration_context(integration_context) # Extract the IDs to check from the current response ids_to_check = [f"{item['id']}-{item['type']}" for item in response["results"]] # Check if entities exist and not all IDs to check are already fetched if entities and not all(id_to_check in already_fetched for id_to_check in ids_to_check): # Iterate over each entity in the entities list for entity in entities: # Retrieve entity details detection_set = entity.get("detection_set", []) entity_id = entity.get("id") entity_name = entity.get("name") entity_type = entity.get("type") last_modified_timestamp = entity.get("last_modified_timestamp") entity_urgency_score = entity.get("urgency_score") entity_checkpoint = f"{entity_id}-{entity_type}" # Check if the entity has detections if len(detection_set) != 0: # Fetch detections data using detections API call # Used entity_id and entity_type to list detections detections_data = client.list_detections_request( detection_type=detection_type, detection_category=detection_category, entity_id=entity_id, entity_type=entity_type, ) detections = detections_data.get("results", []) # Add detection details to the entity entity.update({"detection_details": detections}) # Fetch entity assignment if entity_type == "account": response = client.list_assignments_request(account_ids=entity_id) elif entity_type == "host": response = client.list_assignments_request(host_ids=entity_id) assignment_details = response.get("results", []) assignment_details = assignment_details[0] if len(assignment_details) > 0 else assignment_details entity.update({"assignment_details": assignment_details}) # Add configuration filter entity.update({"filter_tags": tags}) entity["url"] = f"{entity.get('url') or ''}{UTM_PIVOT}" # Create an incident if the entity is not already fetched if entity_checkpoint not in already_fetched: # Updating mirroring fields mirroring_fields = get_mirroring() mirroring_fields.update({"mirror_id": str(entity_id) + "-" + entity_type}) entity.update(mirroring_fields) incident_name = f"Vectra XDR Entity {entity_name}:{entity_id}" # Calculate severity based on urgency score using the urgency_score_to_severity function. severity = urgency_score_to_severity(entity_urgency_score, params) # type: ignore entity["urgency_score_based_severity"] = severity demisto_incidents.append( { "name": incident_name, "occurred": last_modified_timestamp, "rawJSON": json.dumps(entity), "severity": severity, } ) already_fetched.append(entity_checkpoint) # Update the last run object with the latest timestamp and page information if incidents were found if len(entities) == max_fetch and next_url: new_last_run.update({"next_url": next_url, "already_fetched": already_fetched}) else: new_last_run.update( {"time": entities[-1].get("last_modified_timestamp"), "next_url": None, "already_fetched": already_fetched} ) else: # If no incidents were found, update the last run object with the current time and reset the page to 1 now = datetime.now().strftime(DATE_FORMAT) new_last_run.update({"page": 1, "time": now}) # Save the new last run object demisto.setLastRun({"value": json.dumps(new_last_run)}) return demisto_incidents def vectra_user_list_command(client: VectraClient, args: dict[str, Any]): """ Retrieves a list of users from the Vectra API. Args: client (VectraClient): 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 username = args.get("username", "") role = args.get("role", "") # Call Vectra API to retrieve users response = client.list_users_request(username=username, 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: VectraClient, args: dict[str, Any]): """ Retrieves a list of entities from the Vectra API. Args: client (VectraClient): 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: VectraClient, args: dict[str, Any]): """ Describes an entity 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. 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: VectraClient, 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 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: VectraClient, args: dict[str, Any]): """ Describes a list of detections for provided detection IDs from the Vectra API. Args: client (VectraClient): 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: VectraClient, args: dict[str, Any]): """ List entity 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_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_detection_note_list_command(client: VectraClient, 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_entity_note_add_command(client: VectraClient, args: dict[str, Any]): """ Adds a note to an entity 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_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_detection_note_add_command(client: VectraClient, 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_entity_note_update_command(client: VectraClient, args: dict[str, Any]): """ Updates a note to an entity 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_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_detection_note_update_command(client: VectraClient, 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_entity_note_remove_command(client: VectraClient, args: dict[str, Any]): """ Updates a note to an entity 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_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_detection_note_remove_command(client: VectraClient, 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 vectra_entity_tag_add_command(client: VectraClient, args: dict[str, Any]): """ Add tags to an entity. 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_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: VectraClient, args: dict[str, Any]): """ Removes associated tags for the specified entity using 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_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: VectraClient, args: dict[str, Any]): """ List tags for an entity. 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_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_detections_mark_fixed_command(client: VectraClient, args: dict[str, Any]): """ Mark the provided detection IDs as fixed. Args: client (VectraClient): An instance of the VectraClient 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 validate_detections_mark_and_unmark_args(detection_ids_list) # Call Vectra API to mark detection as fixed res = client.mark_or_unmark_detection_fixed_request(detection_ids_list, mark="True") if ( res.get("_meta", {}).get("level") == "Success" and res.get("_meta").get( # type: ignore "message" ) == "Successfully marked detections" ): human_readable = "##### The provided detection IDs have been successfully marked as fixed." else: raise DemistoException("Something went wrong.") return CommandResults(outputs={}, readable_output=human_readable) def vectra_detections_unmark_fixed_command(client: VectraClient, args: dict[str, Any]): """ Unmark the provided detection IDs as fixed. Args: client (VectraClient): An instance of the VectraClient 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 validate_detections_mark_and_unmark_args(detection_ids_list) # Call Vectra API to unmark detection as fixed res = client.mark_or_unmark_detection_fixed_request(detection_ids_list, mark="False") if ( res.get("_meta", {}).get("level") == "Success" and res.get("_meta").get( # type: ignore "message" ) == "Successfully marked detections" ): human_readable = "##### The provided detection IDs have been successfully unmarked as fixed." else: raise DemistoException("Something went wrong.") return CommandResults(outputs={}, readable_output=human_readable) def vectra_assignment_list_command(client: VectraClient, args: dict[str, Any]): """ List assignments. Args: client (VectraClient): An instance of the VectraClient 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: VectraClient, args: dict[str, Any]): """ Create an assignment for specified entity id. Args: client (VectraClient): An instance of the VectraClient 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: VectraClient, args: dict[str, Any]): """ Updates an assignment for specified entity id. Args: client (VectraClient): An instance of the VectraClient 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_entity_assignment_resolve_command(client: VectraClient, args: dict[str, Any]): """ Resolve an assignment for specified assignment id. Args: client (VectraClient): An instance of the VectraClient 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_resolve_command_args(args) # Get function arguments assignment_id = arg_to_number(args.get("assignment_id"), arg_name="assignment_id") outcome = args.get("outcome") note = args.get("note") triage_as = args.get("triage_as") detection_ids = argToList(args.get("detection_ids")) # list outcome assignment res = client.list_assignment_outcomes_request() outcome_list = res.get("results") title_to_id = {item["title"]: item["id"] for item in outcome_list} # type: ignore if outcome in title_to_id: outcome_id = title_to_id[outcome] else: outcome_list = ", ".join([item["title"] for item in outcome_list]) # type: ignore raise ValueError(ERRORS["INVALID_OUTCOME"].format(outcome_list)) # Call Vectra API to resolve an assignment response = client.resolve_entity_assignment_request( assignment_id=assignment_id, outcome=outcome_id, note=note, # type: ignore triage_as=triage_as, # type: ignore detection_ids=detection_ids, ) assignment = response.get("assignment", {}) # resolve assignment response if assignment: assignment["assignment_id"] = assignment["id"] human_readable = "##### The assignment has been successfully resolved." 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_assignment_outcome_list_command(client: VectraClient, args: dict[str, Any]): """ List assignment outcomes. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments. Returns: CommandResults: The command results. """ page = args.get("page", "1") page_size = args.get("page_size", "50") # Validate page and page_size validate_positive_integer_arg(value=page, arg_name="page") validate_positive_integer_arg(value=page_size, arg_name="page_size") # Get command args page = arg_to_number(page, arg_name="page") page_size = arg_to_number(page_size, arg_name="page_size") # Call Vectra API for assignment outcome list response = client.list_assignment_outcomes_request(page=page, page_size=page_size) response = remove_empty_elements(response) count = response.get("count", 0) outcomes = response.get("results", []) human_readable, outcomes = get_assignment_outcome_list_command_hr( outcomes=outcomes, page=page, page_size=page_size, count=count ) context = [createContext(outcome) for outcome in outcomes] return CommandResults( outputs=context, readable_output=human_readable, raw_response=outcomes, outputs_prefix="Vectra.Entity.Assignments.Outcomes", outputs_key_field=["id"], ) def vectra_detection_pcap_download_command(client: VectraClient, args: dict[str, Any]): """ Download the packet capture (PCAP) file associated with a Vectra detection. Args: client (VectraClient): An instance of the VectraClient 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_entity_detections_mark_fixed_command(client: VectraClient, args: dict[str, Any]): """ Mark the provided entity detections as fixed. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If detection_ids argument is missing or empty. Returns: CommandResults: The command results. """ validate_entity_detections_mark_fix_command_args(args) # Get function arguments entity_id = args.get("entity_id") entity_type = args.get("entity_type", "").lower() 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 detections to mark as fixed for this entity ID:{entity_id}." if not detection_ids: return CommandResults(readable_output=hr_string) # Call Vectra API to mark detection as fixed res = client.mark_or_unmark_detection_fixed_request(detection_ids=detection_ids, mark="True") # type: ignore if ( res.get("_meta", {}).get("level") == "Success" and res.get("_meta").get( # type: ignore "message" ) == "Successfully marked detections" ): human_readable = ( f"##### The detections ({', '.join(detection_ids)}) of the provided entity ID have been successfully marked as fixed." ) else: raise DemistoException("Something went wrong.") return CommandResults(readable_output=human_readable) def vectra_group_list_command(client: VectraClient, args: dict[str, Any]): """ Retrieves a list of groups. 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_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_entity_reset_fetch_command(client: VectraClient, args: dict[str, Any]): """Reset the Already_fetched state for the given entity Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments. """ entity_id = args.get("entity_id", "") entity_type = args.get("entity_type", "") add_refetch_id_to_integration_context(entity_id=entity_id, entity_type=entity_type) return CommandResults(readable_output=f"Reset fetch status for {entity_id}-{entity_type}") def vectra_group_unassign_command(client: VectraClient, args: dict[str, Any]): """ Unassign members in Group. Args: client (VectraClient): An instance of the VectraClient 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: VectraClient, args: dict[str, Any]): """ Assign members in Group. Args: client (VectraClient): An instance of the VectraClient 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: VectraClient, args: dict[str, Any]): """ Mark the provided entity detections as closed. Args: client (VectraClient): An instance of the VectraClient 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(): 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_asclosed_command(client: VectraClient, args: dict[str, Any]): """ Mark the provided detection IDs as closed. Args: client (VectraClient): An instance of the VectraClient class. args (Dict[str, Any]): The command arguments. Raises: ValueError: If detection_ids or close_reason arguments are missing or invalid. Returns: CommandResults: The command results. """ # Validate arguments validate_detections_mark_asclosed_args(args) # Get function arguments detection_ids = args.get("detection_ids") close_reason = args.get("close_reason", "").lower() # Convert string into list detection_ids_list = argToList(detection_ids) # Call Vectra API to close detections res = client.close_detections_request(detection_ids=detection_ids_list, 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(): human_readable = f"##### The provided detection IDs have been 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: VectraClient, args: dict[str, Any]): """ Open detection with provided detection IDs. Args: client (VectraClient): An instance of the VectraClient 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(): 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: VectraClient, args: dict[str, Any]): """ List tags for 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. """ 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: VectraClient, args: dict[str, Any]): """ Add tags to 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_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: VectraClient, args: dict[str, Any]): """ Removes associated tags for the specified detection using 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_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 test_module(client: VectraClient) -> str: """ Tests the connection to the Vectra server. Args: client (VectraClient): An instance of the VectraClient class. Returns: str: A message indicating the success of the test. """ params = demisto.params() if params.get("isFetch"): validate_configuration_parameters(params) client.list_entities_request(page_size=1) return "ok" def get_modified_remote_data_command(client: VectraClient, args: dict) -> GetModifiedRemoteDataResponse: """ Get modified remote data from the Vectra platform and prepare it for mirroring in XSOAR. Args: client (VectraClient): An instance of the VectraClient 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_url = None page = 1 page_size = 500 while True: if next_url: # Parse the URL parsed_url = urlparse(next_url) # Extract the query parameters query_params = parse_qs(parsed_url.query) page = arg_to_number(query_params.get("page", [""])[0], arg_name="page") # type: ignore page_size = arg_to_number(query_params.get("page_size", [""])[0], arg_name="page_size") # type: ignore command_last_run_date = query_params.get("last_modified_timestamp_gte", [""])[0] try: response = client.list_entities_request( last_modified_timestamp=command_last_run_date, # type: ignore page=page, page_size=page_size, state="", ) except DemistoException as e: # If the status code is 404 and the message is invalid page number, then return the empty list of incidents. if str(e.res) == "<Response [404]>" and ( e.res.status_code == 404 and ERRORS["INVALID_PAGE_RESPONSE"] in str(e.message) ): demisto.debug(f"Got the 404 error in get-modified-remote-data command: {str(e.message)}") break raise e entities = response.get("results", []) next_url = response.get("next_url") if len(entities) == 0: break # Extra ID and type of the entities modified_entities_ids.extend([str(entity.get("id")) + "-" + entity.get("type") for entity in entities]) # If there is no data on the next page if not next_url: break # Mirroring limit if len(modified_entities_ids) > MAX_MIRRORING_LIMIT: demisto.debug("Max mirroring limit reached.") break # 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 Entity IDs to update in XSOAR: {len(modified_entities_ids)}" ) demisto.debug(f"Performing get-modified-remote-data command. Entity 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: VectraClient, args: dict, params: dict) -> GetRemoteDataResponse: """ Get remote data for a specific entity from the Vectra platform and prepare it for mirroring in XSOAR. Args: client (VectraClient): An instance of the VectraClient class. args (Dict): A dictionary containing the arguments for retrieving remote data. - id (str): The ID of the entity to retrieve. - lastUpdate (str): The timestamp of the last update received for this entity. params (Dict): A dictionary containing the parameters required for updating the remote system. Returns: GetRemoteDataResponse: An object containing the remote incident data and any new entries to return to XSOAR. """ new_entries_to_return: List[dict] = [] dbot_mirror_id: str = args.get("id") # type: ignore demisto.debug(f"dbot_mirror_id:{dbot_mirror_id}") entity_id_type = dbot_mirror_id.split("-") vectra_entity_id = entity_id_type[0] if entity_id_type else "" vectra_entity_type = entity_id_type[1] if entity_id_type else "" demisto.debug(f"vectra_entity_id:{vectra_entity_type}") demisto.debug(f"Getting update for remote {vectra_entity_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. remote_incident_data = client.get_entity_request(entity_id=int(vectra_entity_id), entity_type=vectra_entity_type) if not remote_incident_data: return "Incident was not found." # type: ignore # Get detection set. detection_set = remote_incident_data.get("detection_set", []) entity_urgency_score = remote_incident_data.get("urgency_score") # Calculate severity based on urgency score using the urgency_score_to_severity function. severity = urgency_score_to_severity(entity_urgency_score, params) # type: ignore remote_incident_data["urgency_score_based_severity"] = severity # Collect the detections if the detection set is not empty. # Used entity_id and entity_type to list detections if len(detection_set) != 0: detections_data = client.list_detections_request( detection_type=params.get("detection_type"), # type: ignore detection_category=params.get("detection_category"), # type: ignore entity_id=int(vectra_entity_id), # type: ignore entity_type=vectra_entity_type, ) detections = detections_data.get("results", []) else: detections = [] response = {} # Add detection details to the entity. remote_incident_data.update({"detection_details": detections}) if vectra_entity_type == "account": response = client.list_assignments_request(account_ids=vectra_entity_id) elif vectra_entity_type == "host": response = client.list_assignments_request(host_ids=vectra_entity_id) assignment_details = response.get("results", []) assignment_details = assignment_details[0] if len(assignment_details) > 0 else {} if assignment_details: if not assignment_details.get("resolved_by"): assignment_details["resolved_by"] = {"username": ""} assignment_details["outcome"] = {"title": ""} assignment_details["date_resolved"] = "" else: assignment_details = EMPTY_ASSIGNMENT remote_incident_data.update({"assignment_details": assignment_details}) last_modified_timestamp = arg_to_datetime(remote_incident_data.get("last_modified_timestamp")) if command_last_run_dt > last_modified_timestamp: # type: ignore demisto.debug(f"Nothing new in the Vectra entity {entity_id_type}.") else: demisto.debug(f"The Vectra entity {entity_id_type} is updated.") if detections and not argToBoolean(params.get("refetch_closed_incidents", False)): reopen_in_xsoar(new_entries_to_return, entity_id_type) 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}") return GetRemoteDataResponse(remote_incident_data, new_entries_to_return) def update_remote_system_command(client: VectraClient, 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_entity_id = parsed_args.remote_incident_id mirror_entity_id = parsed_args.data.get("vectraxdrentityid", "") demisto.debug(f"Remote Incident ID: {remote_entity_id}") delta = parsed_args.delta # Get XSOAR incident id xsoar_incident_id = parsed_args.data.get("id", "") demisto.debug(f"XSOAR Incident ID: {xsoar_incident_id}") new_entries = parsed_args.entries xsoar_tags: List = parsed_args.delta.get("tags") or [] remote_entity_type = parsed_args.data.get("vectraxdrentitytype", "").lower() # 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_entity_note_request(entity_id=mirror_entity_id, entity_type=remote_entity_type, note=note_str) # For tags res = client.list_entity_tags_request(entity_id=mirror_entity_id, entity_type=remote_entity_type) 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_entity_tags_request(entity_id=mirror_entity_id, entity_type=remote_entity_type, 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_entity_tags_request(entity_id=mirror_entity_id, entity_type=remote_entity_type, tags=xsoar_tags) # For Closing notes delta_keys = parsed_args.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", "") if argToBoolean(params.get("refetch_closed_incidents", False)): add_refetch_id_to_integration_context(mirror_entity_id, remote_entity_type) 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_entity_note_request(entity_id=mirror_entity_id, entity_type=remote_entity_type, note=closing_note) return remote_entity_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-detections-mark-fixed": vectra_detections_mark_fixed_command, "vectra-detections-unmark-fixed": vectra_detections_unmark_fixed_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-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-entity-assignment-resolve": vectra_entity_assignment_resolve_command, "vectra-assignment-outcome-list": vectra_assignment_outcome_list_command, "vectra-detection-pcap-download": vectra_detection_pcap_download_command, "vectra-entity-detections-mark-fixed": vectra_entity_detections_mark_fixed_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-entity-reset-fetch": vectra_entity_reset_fetch_command, } try: result = None # Creates vectra client client = VectraClient( server_url=server_url, client_id=client_id, client_secret_key=client_secret_key, verify=verify_certificate, proxy=proxy, ) # Get Command args args = demisto.args() if command == "test-module": result = test_module(client) elif command == "fetch-incidents": # Fetch incidents incidents = fetch_incidents(client, params) # Ingest incidents in XSOAR demisto.debug(f"{len(incidents)} incidents are created successfully in XSOAR.") demisto.incidents(incidents) 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, params) # type: ignore elif command == "update-remote-system": result = update_remote_system_command(client, args, params) 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) 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()