RST Cloud - Threat Feed API
This is the RST Threat Feed integration for interacting with API.
Data Enrichment & Threat Intelligence · RST Threat Feed
Details
| ID | RST Cloud - Threat Feed API |
|---|---|
| Provider | RSTCloud |
| Category | Data Enrichment & Threat Intelligence |
| From Version | 6.0.0 |
| Docker Image | demisto/python3:3.12.13.10116658 |
| Supported Modules | Agentix XSIAM |
README
This is the RST Threat Feed integration for interacting with API
This integration was integrated and tested with RST Cloud - Threat Feed API v1
Please contact the RST Cloud team via email support@rstcloud.net to obtain a key and ask any questions you have.
Also, the following contact details can be used:
Each indicator is ranked from 0 to 100. Indicators are being collected from multiple sources and are cross-verified using multiple criteria.
Please check indicator tags and malware family fields. An indicator may describe a known malware or a scanning host. Therefore, different actions may be required based on the context.
Configure RST Cloud - Threat Feed API in Cortex
| Parameter | Description | Required |
|---|---|---|
| Server URL (e.g. https://api.rstcloud.net/v1) | True | |
| API Key | True | |
| Score threshold for IP reputation command | Set this to determine the RST Threat Feed score that will determine if an IP is malicious (0-100) | True |
| Score threshold for domain reputation command | Set this to determine the RST Threat Feed score that will determine if a domain is malicious (0-100) | True |
| Score threshold for url reputation command | Set this to determine the RST Threat Feed score that will determine if a url is malicious (0-100) | True |
| Score threshold for file reputation command | Set this to determine the RST Threat Feed score that will determine if a file is malicious (0-100) | True |
| IP Indicator Expiration (days) | Mark IP indicators older than indicator_expiration_ip value in days as Suspicious ignoring the last available score | True |
| Domain Indicator Expiration (days) | Mark domain indicators older than indicator_expiration_domain value in days as Suspicious ignoring the last available score | True |
| URL Indicator Expiration (days) | Mark URL indicators older than indicator_expiration_url value in days as Suspicious ignoring the last available score | True |
| Hash Indicator Expiration (days) | Mark Hash indicators older than indicator_expiration_url value in days as Suspicious ignoring the last available score | True |
| Use system proxy settings | False | |
| Trust any certificate (not secure) | False |
Commands
You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.
ip
Returns IP information and reputation.
Base Command
ip
Input
| Argument Name | Description | Required |
|---|---|---|
| ip | List of IPs. | Required |
| threshold | If the IP has reputation above the threshold then the IP defined as malicious. If threshold not set, then threshold from instance configuration is used. Default is 45. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| DBotScore.Indicator | String | The indicator that was tested. |
| DBotScore.Score | Number | The actual score. |
| DBotScore.Type | String | The indicator type. |
| DBotScore.Vendor | String | The vendor used to calculate the score. |
| IP.Address | String | IP address. |
| IP.Geo.Country | String | Country of origin. |
| IP.Tags | String | The associated tags. |
| IP.Malicious.Vendor | String | The vendor reporting the IP address as malicious. |
| IP.Malicious.Description | String | A description explaining why the IP address was reported as malicious. |
| IP.Malicious.Score | String | The score calculated for the indicator by the vendor. |
| RST.IP.Address | String | The actual IP address. |
| RST.IP.Geo.Country | String | The country name. |
| RST.IP.Geo.Region | String | The geo region name. |
| RST.IP.Geo.City | String | The city name. |
| RST.IP.ASN | String | The autonomous system name for the IP address. |
| RST.IP.Organization | String | The organisation name for the autonomous system name for the IP address. |
| RST.IP.ISP | String | The Internet Service Provider name for the autonomous system name for the IP address. |
| RST.IP.CloudHosting | String | The Cloud Provider name for the IP address. |
| RST.IP.NumberOfDomainInASN | String | The number of domain names for the IP address. |
| RST.IP.FirstSeen | Date | First Seen. |
| RST.IP.LastSeen | Date | Last Seen. |
| RST.IP.Tags | String | The associated tags. |
| RST.IP.Threat | String | The associated Malware Family or threat name. |
| RST.IP.Score | Number | The total score. |
| RST.IP.UUID | String | The unique ID for the indicator. |
| RST.IP.RSTReference | String | The link to the raw JSON indicator. |
| RST.IP.Related | String | The associated domains. |
| RST.IP.FalsePositive | String | true if it is likely a False Positive. |
| RST.IP.FalsePositiveDesc | String | Description why we think it may be a False Positive. |
| RST.IP.CVE | String | Related CVE (vulnerabilities) |
| RST.IP.Industry | String | Related Industry. |
| RST.IP.Report | String | Collected from. |
Command Example
!ip ip=1.2.3.4 threshold=50
Context Example
{
"DBotScore": {
"Indicator": "1.2.3.4",
"Score": 2,
"Type": "ip",
"Vendor": "RST Cloud"
},
"IP": {
"ASN": "4788",
"Address": "1.2.3.4",
"Geo": {
"Country": "Malaysia"
},
"Tags": [
"c2",
"generic"
]
},
"RST": {
"IP": {
"ASN": "4788",
"Address": "1.2.3.4",
"CloudHosting": "",
"FalsePositive": "false",
"FalsePositiveDesc": "",
"FirstSeen": "2019-12-05T00:00:00.000Z",
"Geo": {
"city": "Batang Kali",
"country": "Malaysia",
"region": "Selangor"
},
"ISP": "TMNETASAP",
"LastSeen": "2021-01-26T00:00:00.000Z",
"NumberOfDomainInASN": "9615",
"Organization": "TM Net Internet Service Provider",
"RSTReference": "https://rstcloud.net/uuid?id=8f10a17d-9931-3329-b97f-db3953c093e2",
"Related": [],
"Score": "3",
"Tags": [
"c2",
"generic"
],
"Threat": [
"emotet"
],
"Type": "IP",
"UUID": "8f10a17d-9931-3329-b97f-db3953c093e2"
}
}
}
Human Readable Output
RST Threat Feed IP Reputation for: 1.2.3.4
Description Last Seen Relevance Score Tags Threat Ioc with tags: c2, generic. related threats: emotet 2021-01-26 Suspicious 3 c2, generic emotet
domain
Returns Domain information and reputation.
Base Command
domain
Input
| Argument Name | Description | Required |
|---|---|---|
| domain | List of Domains. | Required |
| threshold | If the domain has reputation above the threshold then the domain defined as malicious. If threshold not set, then threshold from instance configuration is used. Default is 45. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| DBotScore.Indicator | String | The indicator that was tested. |
| DBotScore.Score | Number | The actual score. |
| DBotScore.Type | String | The indicator type. |
| DBotScore.Vendor | String | The vendor used to calculate the score. |
| Domain.Name | String | The domain name. |
| Domain.Tags | String | The associated tags |
| Domain.Malicious.Vendor | String | The vendor reporting the domain as malicious. |
| Domain.Malicious.Description | String | A description explaining why the domain was reported as malicious. |
| Domain.Malicious.Score | String | The score calculated for the indicator by the vendor. |
| RST.Domain.Name | String | The domain name. |
| RST.Domain.WhoisAge | Number | Days since creation. |
| RST.Domain.WhoisDomainCreationDate | Date | Creation date. Format is ISO8601. |
| RST.Domain.WhoisDomainUpdateDate | Date | Update date. Format is ISO8601. |
| RST.Domain.WhoisDomainExpireDate | Date | Expiration date. Format is ISO8601. |
| RST.Domain.WhoisRegistrar | String | Domain Registrar. |
| RST.Domain.WhoisRegistrant | String | Domain Registrant. |
| RST.Domain.FirstSeen | Date | First Seen. |
| RST.Domain.LastSeen | Date | Last Seen. |
| RST.Domain.Tags | String | The associated tags. |
| RST.Domain.Threat | String | The associated Malware Family or threat name. |
| RST.Domain.Score | Number | The total score. |
| RST.Domain.UUID | String | The unique ID for the indicator. |
| RST.Domain.RSTReference | String | The link to the raw JSON indicator. |
| RST.Domain.Related | String | The associated IP addresses. |
| RST.Domain.FalsePositive | String | true if it is likely a False Positive. |
| RST.Domain.FalsePositiveDesc | String | Description why we think it may be a False Positive. |
| RST.Domain.CVE | String | Related CVE (vulnerabilities) |
| RST.Domain.Industry | String | Related Industry. |
| RST.Domain.Report | String | Collected from. |
Command Example
!domain domain="domaintovalidate.local" threshold=40
Context Example
{
"DBotScore": {
"Indicator": "domaintovalidate.local",
"Score": 2,
"Type": "domain",
"Vendor": "RST Cloud"
},
"Domain": {
"Name": "domaintovalidate.local",
"Tags": [
"malware"
]
},
"RST": {
"Domain": {
"FalsePositive": "true",
"FalsePositiveDesc": "Domain not resolved. Whois records not found",
"FirstSeen": "2020-06-26T00:00:00.000Z",
"LastSeen": "2021-01-25T00:00:00.000Z",
"Name": "domaintovalidate.local",
"RSTReference": "https://rstcloud.net/uuid?id=552fdbe7-7265-3a9d-b364-83426d1c2dbc",
"Related": {
"a": [],
"alias": [],
"cname": []
},
"Score": "10",
"Tags": [
"malware"
],
"Threat": [],
"Type": "Domain",
"UUID": "552fdbe7-7265-3a9d-b364-83426d1c2dbc",
"WhoisAge": "",
"WhoisDomainCreationDate": "",
"WhoisDomainExpireDate": "",
"WhoisDomainUpdateDate": "",
"WhoisRegistrant": "",
"WhoisRegistrar": ""
}
}
}
Human Readable Output
RST Threat Feed Domain Reputation for: domaintovalidate.local
Description Last Seen Relevance: Score Tags Ioc with tags: malware 2021-01-25 Suspicious 10 malware
url
Returns URL information and reputation.
Base Command
url
Input
| Argument Name | Description | Required |
|---|---|---|
| url | List of URLs. | Required |
| threshold | If the URL has reputation above the threshold then the URL defined as malicious. If threshold not set, then threshold from instance configuration is used. Default is 30. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| DBotScore.Indicator | String | The indicator that was tested. |
| DBotScore.Score | Number | The actual score. |
| DBotScore.Type | String | The indicator type. |
| DBotScore.Vendor | String | The vendor used to calculate the score. |
| URL.Data | String | The URL. |
| URL.Tags | String | The associated tags. |
| URL.Malicious.Vendor | String | The vendor reporting the URL as malicious. |
| URL.Malicious.Description | String | A description explaining why the URL was reported as malicious. |
| URL.Malicious.Score | String | The score calculated for the URL indicator by the vendor. |
| RST.URL.Data | String | The URL. |
| RST.URL.Status | String | Last HTTP status code. |
| RST.URL.FirstSeen | Date | First Seen. |
| RST.URL.LastSeen | Date | Last Seen. |
| RST.URL.Tags | String | The associated tags. |
| RST.URL.Threat | String | The associated Malware Family or threat name. |
| RST.URL.Score | Number | The total score. |
| RST.URL.UUID | String | The unique ID for the indicator |
| RST.URL.Description | String | The associated Description provided by the vendor. |
| RST.URL.FalsePositive | String | true if it is likely a False Positive. |
| RST.URL.FalsePositiveDesc | String | Description why we think it may be a False Positive. |
| RST.URL.Parsed | String | Parsed URL components. |
| RST.URL.CVE | String | Related CVE (vulnerabilities) |
| RST.URL.Industry | String | Related Industry. |
| RST.URL.Report | String | Collected from. |
Command Example
!url url="https://domain.local/testurl" threshold=30
Context Example
{
"DBotScore": {
"Indicator": "https://domain.local/testurl",
"Score": 2,
"Type": "url",
"Vendor": "RST Cloud"
},
"RST": {
"URL": {
"CVE": [],
"Data": "https://domain.local/testurl",
"FalsePositive": "true",
"FalsePositiveDesc": "Resource unavailable",
"FirstSeen": "2021-01-05T00:00:00.000Z",
"LastSeen": "2021-01-26T00:00:00.000Z",
"Parsed": {
"anchor": null,
"domain": "domain.local",
"params": null,
"path": "/testurl",
"port": "443",
"schema": "https"
},
"RSTReference": "https://rstcloud.net/uuid?id=f64f7a99-068b-3fec-b572-598f9d11d4d6",
"Score": "14",
"Status": "503",
"Tags": [
"malware"
],
"Threat": [
"emotet"
],
"Type": "URL",
"UUID": "f64f7a99-068b-3fec-b572-598f9d11d4d6"
}
},
"URL": {
"Data": "https://domain.local/testurl",
"Tags": [
"malware"
]
}
}
Human Readable Output
RST Threat Feed URL Reputation for:
https://domain.local/testurl
Description Last Seen Relevance Score Tags Threat Ioc with tags: malware. related threats: emotet 2021-01-26 Suspicious 14 malware emotet
file
Returns File information and reputation.
Base Command
file
Input
| Argument Name | Description | Required |
|---|---|---|
| file | List of Files. | Required |
| threshold | If the File has reputation above the threshold then the File defined as malicious. If threshold not set, then threshold from instance configuration is used. Default is 5. | Optional |
Context Output
| Path | Type | Description |
|---|---|---|
| DBotScore.Indicator | String | The indicator that was tested. |
| DBotScore.Score | Number | The actual score. |
| DBotScore.Type | String | The indicator type. |
| DBotScore.Vendor | String | The vendor used to calculate the score. |
| File.Name | String | The file name. |
| File.MD5 | String | MD5 for the the file name. |
| File.SHA1 | String | SHA1 for the the file name. |
| File.SHA256 | String | The URL. |
| File.Tags | String | The associated tags. |
| File.Malicious.Vendor | String | The vendor reporting the File as malicious. |
| File.Malicious.Description | String | A description explaining why the File was reported as malicious. |
| File.Malicious.Score | String | The score calculated for the File indicator by the vendor. |
| RST.File.Name | String | The file name. |
| RST.File.MD5 | String | MD5 for the the file name. |
| RST.File.SHA1 | String | SHA1 for the the file name. |
| RST.File.SHA256 | String | SHA256 for the the file name. |
| RST.File.FirstSeen | Date | First Seen. |
| RST.File.LastSeen | Date | Last Seen. |
| RST.File.Tags | String | The associated tags. |
| RST.File.Threat | String | The associated Malware Family or threat name. |
| RST.File.Score | Number | The total score. |
| RST.File.UUID | String | The unique ID for the indicator. |
| RST.File.Description | String | The associated Description provided by the vendor. |
| RST.File.FalsePositive | String | true if it is likely a False Positive. |
| RST.File.FalsePositiveDesc | String | Description why we think it may be a False Positive. |
| RST.File.CVE | String | Related CVE (vulnerabilities) |
| RST.File.Industry | String | Related Industry. |
| RST.File.Report | String | Collected from. |
Command Example
!file file="fe3d38316dc38a4ec63eac80e34cb157c9d896460f9b7b3bfbd2cec4e2cb8cdc"threshold=5
Context Example
{
"DBotScore": {
"Indicator": "fe3d38316dc38a4ec63eac80e34cb157c9d896460f9b7b3bfbd2cec4e2cb8cdc",
"Score": 3,
"Type": "file",
"Vendor": "RST Cloud"
},
"RST": {
"File": {
"CVE": [],
"FalsePositive": "false",
"FalsePositiveDesc": "",
"FirstSeen": "2021-05-11T00:00:00.000Z",
"Industry": [],
"LastSeen": "2022-03-11T00:00:00.000Z",
"Name": [],
"RSTReference": "https://rstcloud.net/uuid?id=c86948e7-eb72-3fe1-96e9-429e885cea3b",
"Report": [
"https://www.threatfabric.com/blogs/partners-in-crime-medusa-cabassous.html"
],
"SHA256": "fe3d38316dc38a4ec63eac80e34cb157c9d896460f9b7b3bfbd2cec4e2cb8cdc",
"Score": "6",
"Tags": [
"malware"
],
"Threat": [
"medusa",
"flubot"
],
"Type": "File",
"UUID": "c86948e7-eb72-3fe1-96e9-429e885cea3b"
}
},
"File": {
"SHA256": "fe3d38316dc38a4ec63eac80e34cb157c9d896460f9b7b3bfbd2cec4e2cb8cdc",
"Tags": [
"malware"
]
}
}
Human Readable Output
RST Threat Feed File Reputation for:
https://domain.local/testurl
Description Last Seen Relevance Score Tags Threat Ioc with tags: malware. related threats: emotet 2021-01-26 Suspicious 14 malware emotet
rst-submit-new
Submits an indicator to RST Threat Feed.
Base Command
rst-submit-new
Input
| Argument Name | Description | Required |
|---|---|---|
| ioc | List of IOCs (URLs, domains or IPs). | Required |
| description | Any context to pass to RST Cloud. | Optional |
Context Output
There is no context output for this command.
Command Example
!rst-submit-new ioc="thisisamaliciouswebsite.com" description="a user downloaded a trojan"
Human Readable Output
Indicator: thisisamaliciouswebsite.com was submitted as a potential threat indicator to RST Cloud
rst-submit-fp
Submits a potential False Positive to RST Threat Feed.
Base Command
rst-submit-fp
Input
| Argument Name | Description | Required |
|---|---|---|
| ioc | List of IOCs (URLs, domains or IPs). | Required |
| description | Any context to pass to RST Cloud. | Optional |
Context Output
There is no context output for this command.
Command Example
!rst-submit-fp ioc="thisisnotamaliciousdomain.com" description="a decent website"
Human Readable Output
Indicator: thisisnotamaliciousdomain.com was submitted as False Positive to RST Cloud
Configuration parameters
url— Server URL (e.g. https://api.rstcloud.net/v1) (required)apikey— API Key (required)threshold_ip— Score threshold for IP reputation command (required)threshold_domain— Score threshold for domain reputation command (required)threshold_url— Score threshold for url reputation command (required)threshold_hash— Score threshold for hash reputation commandindicator_expiration_ip— IP Indicator Expiration (days) (required)indicator_expiration_domain— Domain Indicator Expiration (days) (required)indicator_expiration_url— URL Indicator Expiration (days) (required)indicator_expiration_hash— File Hash Indicator Expiration (days)proxy— Use system proxy settingsinsecure— Trust any certificate (not secure)integrationReliability— Source ReliabilityfeedExpirationPolicy—feedExpirationInterval—
Commands (6)
-
domainReturns Domain information and reputation.
-
fileReturns File information and reputation.
-
ipReturns IP information and reputation.
-
rst-submit-fpSubmits a potential False Positive to RST Threat Feed.
-
rst-submit-newSubmits an indicator to RST Threat Feed.
-
urlReturns URL information and reputation.
import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 urllib3.disable_warnings() """ CONSTANTS """ RSTCLOUD_URL = "https://api.rstcloud.net" RST_URL = "https://rstcloud.net/" DEFAULT_THRESHOLD = 50 DEFAULT_EXPIRATION = 180 DBOT_SCORE_KEY = "DBotScore(val.Indicator == obj.Indicator && val.Vendor == obj.Vendor)" IPV4REGEX = r"^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$" DOMAINREGEX = r"(((?=[a-z0-9\-_]{1,63}\.)(xn--)?[a-z0-9_\-]+(-[a-z0-9_]+)*\.)+[a-z-0-9]{2,63})(:\d+)?" URLREGEX = ( r"^(?:(?:(?:https?|ftps?):)?\/\/)?((?:\S+(?::\S*)?@)?(?:(?!(?:10|127)(?:\.\d{1,3}){3})(?!(?:169\.254|192\.168)" + r"(?:\.\d{1,3}){2})(?!172\.(?:1[6-9]|2\d|3[0-1])(?:\.\d{1,3}){2})(?:[1-9]\d?|1\d\d|2[01]\d|22[0-3])" + r"(?:\.(?:1?\d{1,2}|2[0-4]\d|25[0-5])){2}(?:\.(?:[1-9]\d?|1\d\d|2[0-4]\d|25[0-4]))|" + r"(?:(?:[a-z0-9\\u00a1-\\uffff][a-z0-9\\u00a1-\\uffff_-]{0,62})?" + r"[a-z0-9\\u00a1-\\uffff]\.)+(?:[a-z\\u00a1-\\uffff]{2,}|xn--[a-z0-9]+\.?))(?::\d{2,5})?(?:[\/?#]\S*)?)$" ) MD5REGEX = r"^[a-f0-9]{32}$" SHA1REGEX = r"^[0-9a-f]{40}$" SHA256REGEX = r"^[A-Fa-f0-9]{64}$" """ CLIENT CLASS """ class RSTIP(Common.IP): CONTEXT_PATH = "IP(val.Address && val.Address == obj.Address)" def __init__( self, ip, dbot_score, asn=None, hostname=None, geo_latitude=None, geo_longitude=None, geo_country=None, geo_description=None, detection_engines=None, positive_engines=None, rstscore=None, tags=None, malwarefamily=None, firstseenbysource=None, lastseenbysource=None, ): super().__init__( ip, dbot_score, asn, hostname, geo_latitude, geo_longitude, geo_country, geo_description, detection_engines, positive_engines, ) self.rstscore = rstscore self.tags = tags self.malwarefamily = malwarefamily self.firstseenbysource = firstseenbysource self.lastseenbysource = lastseenbysource def to_context(self): ret_value = super().to_context() ip_context = {"Address": self.ip} if self.rstscore: ip_context["RST Score"] = self.rstscore if self.tags: ip_context["Tags"] = self.tags if self.malwarefamily: ip_context["MalwareFamily"] = self.malwarefamily if self.lastseenbysource: ip_context["FirstSeenBySource"] = self.firstseenbysource if self.lastseenbysource: ip_context["LastSeenBySource"] = self.lastseenbysource ret_value[Common.IP.CONTEXT_PATH].update(ip_context) return ret_value class RSTDomain(Common.Domain): CONTEXT_PATH = "Domain(val.Name && val.Name == obj.Name)" def __init__( self, domain, dbot_score, dns=None, detection_engines=None, positive_detections=None, organization=None, sub_domains=None, creation_date=None, updated_date=None, expiration_date=None, domain_status=None, name_servers=None, registrar_name=None, registrar_abuse_email=None, registrar_abuse_phone=None, registrant_name=None, registrant_email=None, registrant_phone=None, registrant_country=None, admin_name=None, admin_email=None, admin_phone=None, admin_country=None, rstscore=None, tags=None, malwarefamily=None, firstseenbysource=None, lastseenbysource=None, ): super().__init__( domain, dbot_score, dns, detection_engines, positive_detections, organization, sub_domains, creation_date, updated_date, expiration_date, domain_status, name_servers, registrar_name, registrar_abuse_email, registrar_abuse_phone, registrant_name, registrant_email, registrant_phone, registrant_country, admin_name, admin_email, admin_phone, admin_country, ) self.rstscore = rstscore self.tags = tags self.malwarefamily = malwarefamily self.firstseenbysource = firstseenbysource self.lastseenbysource = lastseenbysource def to_context(self): ret_value = super().to_context() domain_context = {"Name": self.domain} if self.rstscore: domain_context["RST Score"] = self.rstscore if self.tags: domain_context["Tags"] = self.tags if self.malwarefamily: domain_context["MalwareFamily"] = self.malwarefamily if self.lastseenbysource: domain_context["FirstSeenBySource"] = self.firstseenbysource if self.lastseenbysource: domain_context["LastSeenBySource"] = self.lastseenbysource ret_value[Common.Domain.CONTEXT_PATH].update(domain_context) return ret_value class RSTUrl(Common.URL): CONTEXT_PATH = "URL(val.Data && val.Data == obj.Data)" def __init__( self, url, dbot_score, detection_engines=None, positive_detections=None, category=None, rstscore=None, tags=None, malwarefamily=None, firstseenbysource=None, lastseenbysource=None, ): super().__init__(url, dbot_score, detection_engines, positive_detections, category) self.rstscore = rstscore self.tags = tags self.malwarefamily = malwarefamily self.firstseenbysource = firstseenbysource self.lastseenbysource = lastseenbysource def to_context(self): ret_value = super().to_context() url_context = {"Data": self.url} if self.rstscore: url_context["RST Score"] = self.rstscore if self.tags: url_context["Tags"] = self.tags if self.malwarefamily: url_context["MalwareFamily"] = self.malwarefamily if self.lastseenbysource: url_context["FirstSeenBySource"] = self.firstseenbysource if self.lastseenbysource: url_context["LastSeenBySource"] = self.lastseenbysource ret_value[Common.URL.CONTEXT_PATH].update(url_context) return ret_value class RSTFile(Common.File): CONTEXT_PATH = ( "File(val.MD5 && val.MD5 == obj.MD5 || val.SHA1 && val.SHA1 == obj.SHA1 || " "val.SHA256 && val.SHA256 == obj.SHA256 || val.SHA512 && val.SHA512 == obj.SHA512 || " "val.CRC32 && val.CRC32 == obj.CRC32 || val.CTPH && val.CTPH == obj.CTPH || " "val.SSDeep && val.SSDeep == obj.SSDeep)" ) def __init__( self, dbot_score, md5=None, sha1=None, sha256=None, detection_engines=None, positive_detections=None, category=None, rstscore=None, tags=None, malwarefamily=None, firstseenbysource=None, lastseenbysource=None, ): super().__init__(md5, sha1, sha256, dbot_score, detection_engines, positive_detections, category) self.rstscore = rstscore self.tags = tags self.malwarefamily = malwarefamily self.firstseenbysource = firstseenbysource self.lastseenbysource = lastseenbysource def to_context(self): ret_value = super().to_context() file_context = {"MD5": self.md5, "SHA1": self.sha1, "SHA256": self.sha256} if self.rstscore: file_context["RST Score"] = self.rstscore if self.tags: file_context["Tags"] = self.tags if self.malwarefamily: file_context["MalwareFamily"] = self.malwarefamily if self.lastseenbysource: file_context["FirstSeenBySource"] = self.firstseenbysource if self.lastseenbysource: file_context["LastSeenBySource"] = self.lastseenbysource ret_value[Common.File.CONTEXT_PATH].update(file_context) return ret_value class Client: def __init__(self, apikey, api_url="https://api.rstcloud.net/v1", verify=False, proxy=False): self.apikey = apikey self.api_url = api_url self.verify = verify self.proxy = proxy def get_indicator(self, value): """Gets reputation data using the '/ioc' API endpoint Args: value (str): an indicator value to get the reputation for Returns: Dict - dict containing the IOC reputation as returned from the API """ endpoint = "/ioc" apiurl = self.api_url + endpoint + "?value=" + value headers = {"Accept": "*/*", "X-Api-Key": self.apikey} r = requests.get(apiurl, headers=headers, verify=self.verify, proxies=self.proxy) return r.json() def submit_indicator(self, value, desc="manual submission"): """Submits an indicator using the '/ioc' API endpoint Args: value (str): an indicator value to submit as a new indicator desc (str): an indicator description why it is considered as malicious Returns: dict - contains the confirmation or error """ endpoint = "/ioc" apiurl = self.api_url + endpoint payload = {"ioc_value": value, "description": desc} headers = {"Accept": "*/*", "X-Api-Key": self.apikey} r = requests.post(apiurl, json=payload, headers=headers, verify=self.verify, proxies=self.proxy) return r.json() def submit_falsepositive(self, value, desc="manual submission"): """Submits an indicator using the '/ioc' API endpoint Args: value (str): an indicator value to submit as a False Positive desc (str): an indicator description why it is considered as False Positive Returns: dict - contains the confirmation or error """ endpoint = "/ioc" payload = {"ioc_value": value, "description": desc} apiurl = self.api_url + endpoint headers = {"Accept": "*/*", "X-Api-Key": self.apikey} r = requests.put(apiurl, json=payload, headers=headers, verify=self.verify, proxies=self.proxy) return r.json() """ HELPER FUNCTIONS """ def check_arg_type(arg_name: str, arg_value: str): """ Checks that RST Threat Feed API parameters are valid. Args: arg_name (str): paramater name arg_value (str): paramater value to verify Returns: (str): a null string means OK while any text is an error """ output = "" try: isinstance(int(arg_value), int) value = int(arg_value) if "threshold" in arg_name and value < 0 or value > 100: output = str(arg_name) + ": the value must be between 0 and 100; " if "indicator_expiration" in arg_name and value < 0: output = str(arg_name) + ": the value must be positive (>0); " except Exception: return str(arg_name) + ": bad format, must be a number; " return output def calculate_score(score: int, threshold: int, itype: str, lseen: int) -> int: """ Calculates and converts RST Threat Feed score into XSOAR score. Args: score (int): the score from RST Threat Feed for certain indicator (0-100). threshold (int): the score threshold configured by the user. itype (str): indicator type - ip, domain, url lseen (int): last seen in epoch (sec) Returns: int - XSOAR's score for the indicator """ expiry_limit = int(demisto.params().get("indicator_expiration_" + itype, DEFAULT_EXPIRATION)) lseendays = (int(time.time()) - lseen) / 24 / 60 / 60 if score > threshold and expiry_limit > lseendays: return Common.DBotScore.BAD else: return Common.DBotScore.SUSPICIOUS def parse_indicator_response(res, indicator_type): """ Parses responses from RST Threat Feed API. Args: res (Dict[str, str]): RST Threat Feed response indicator_type (str): IP, Domain, URL or File Returns: (Dict[str, str]): a result to return into XSOAR's context. """ name = {"IP": "Address", "Domain": "Name", "URL": "Data"} if indicator_type != "File": indicator = {name[indicator_type]: res.get("ioc_value", ""), "Type": indicator_type} else: indicator = {} try: indicator = {"MD5": res.get("md5", ""), "Type": indicator_type} except Exception: pass try: indicator = {"SHA1": res.get("sha1", ""), "Type": indicator_type} except Exception: pass try: indicator = {"SHA256": res.get("sha256", ""), "Type": indicator_type} except Exception: pass if "error" not in res: first_seen = str(int(res.get("fseen", "")) * 1000) last_seen = str(int(res.get("lseen", "")) * 1000) indicator["Score"] = res.get("score").get("total") if first_seen: indicator["FirstSeen"] = timestamp_to_datestring(first_seen) if last_seen: indicator["LastSeen"] = timestamp_to_datestring(last_seen) if "tags" in res: indicator["Tags"] = res.get("tags").get("str", "") if "threat" in res: indicator["Threat"] = res.get("threat", "") if "fp" in res: indicator["FalsePositive"] = res.get("fp").get("alarm", "") indicator["FalsePositiveDesc"] = res.get("fp").get("descr", "") if "cve" in res: indicator["CVE"] = res.get("cve", "") if "industry" in res: indicator["Industry"] = res.get("industry", "") if "src" in res and "report" in res["src"]: try: indicator["Report"] = res.get("src").get("report", "").split(",") except Exception: indicator["Report"] = res.get("src").get("report", "") if "id" in res: indicator["UUID"] = res.get("id", "") indicator["RSTReference"] = "https://rstcloud.net/uuid?id=" + res.get("id", "") if indicator_type == "Domain": indicator["WhoisDomainCreationDate"] = res.get("whois", {}).get("created", "") indicator["WhoisDomainExpireDate"] = res.get("whois", {}).get("expires", "") indicator["WhoisDomainUpdateDate"] = res.get("whois", {}).get("updated", "") indicator["WhoisRegistrar"] = res.get("whois", {}).get("registrar", "") indicator["WhoisRegistrant"] = res.get("whois", {}).get("registrant", "") indicator["WhoisAge"] = res.get("whois", {}).get("age", "") indicator["Related"] = res.get("resolved").get("ip", "") if indicator_type == "IP": indicator["ASN"] = res.get("asn", {}).get("num", "") indicator["CloudHosting"] = res.get("asn", {}).get("cloud", "") indicator["NumberOfDomainInASN"] = res.get("asn", {}).get("domains", "") indicator["Organization"] = res.get("asn", {}).get("org", "") indicator["ISP"] = res.get("asn", {}).get("isp", "") indicator["Geo"] = res.get("geo") indicator["Related"] = res.get("related").get("domains", "") if indicator_type == "URL": indicator["Parsed"] = res.get("parsed", "") indicator["Status"] = res.get("resolved").get("status", "") if indicator_type == "File": indicator["Name"] = res.get("filename", "") else: indicator["error"] = res.get("error", "") return indicator """ COMMAND FUNCTIONS """ def test_module(client: Client) -> str: """ Returning 'ok' indicates that the integration works like it is supposed to. Connection to the service is successful. Args: client (Client): RST Threat Feed client. Returns: str: 'ok' if test passed, anything else will fail the test. """ result = "" params = [ "threshold_ip", "threshold_domain", "threshold_url", "threshold_hash", "indicator_expiration_ip", "indicator_expiration_domain", "indicator_expiration_url", "indicator_expiration_hash", ] for param in params: result += check_arg_type(param, demisto.params().get(param)) if result == "": if "ioc_value" in client.get_indicator("1.1.1.1"): result += "ok" else: result += "Connection failed." return result def ip_command(client: Client, args: dict[str, str]) -> tuple[list, list, list]: """ Executes IP enrichment against RST Threat Feed. Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the IP indicators. list(dict): the results to return into XSOAR's context. list(dict): the raw results to return into XSOAR's context. """ addresses = argToList(args.get("ip", "")) threshold = int(args.get("threshold", DEFAULT_THRESHOLD)) if check_arg_type("threshold_ip", str(threshold)) != "": raise Exception(str(threshold) + ": threshold must be from 0 to 100") markdown = [] raw_results = [] indicators = [] for ip in addresses: markdown_item = "" ipv4regex = re.compile(IPV4REGEX) ipv4match = ipv4regex.fullmatch(ip) if ipv4match: indicator = client.get_indicator(ip) else: raise Exception("is not valid IP") if "error" in indicator: if indicator["error"] == "Not Found": score = Common.DBotScore( indicator=ip, indicator_type=DBotScoreType.IP, integration_name="RST Cloud", score=Common.DBotScore.NONE, reliability=demisto.params().get("integrationReliability"), ) markdown_item += f"IP: {ip} not found\n" raw_results.append(parse_indicator_response(indicator, "IP")) indicators.append(RSTIP(ip=indicator["ioc_value"], dbot_score=score)) markdown.append(markdown_item) continue else: raise Exception( f"RST Threat Feed API error while getting a response for {indicator['ioc_value']}: {indicator['error']}\n" ) else: total_score = int(indicator.get("score", {}).get("total")) calc_score = calculate_score(total_score, threshold, "ip", int(indicator.get("lseen", ""))) dbot_score = Common.DBotScore( indicator=indicator["ioc_value"], indicator_type=DBotScoreType.IP, integration_name="RST Cloud", score=calc_score, malicious_description=indicator.get("description", ""), reliability=demisto.params().get("integrationReliability"), ) human_readable_score = "" if calc_score == 3: human_readable_score = "Malicious" if calc_score == 2: human_readable_score = "Suspicious" result = RSTIP( ip=indicator["ioc_value"], asn=indicator.get("asn", {}).get("num", ""), geo_country=indicator.get("geo", {}).get("country", ""), dbot_score=dbot_score, tags=indicator.get("tags", "").get("str", ""), ) table = { "Score": total_score, "Relevance": human_readable_score, "Threat": ", ".join(threat for threat in indicator.get("threat", "")), "Last Seen": time.strftime("%Y-%m-%d", time.localtime(int(indicator.get("lseen", "")))), "Description": f"{string_to_context_key(indicator.get('description', ''))}\n", "Tags": ", ".join(tag for tag in indicator.get("tags", "").get("str", "")), } markdown_item += tableToMarkdown( f'RST Threat Feed IP Reputation for: {indicator["ioc_value"]}\n', table, removeNull=True ) markdown.append(markdown_item) raw_results.append(parse_indicator_response(indicator, "IP")) indicators.append(result) return markdown, raw_results, indicators def domain_command(client: Client, args: dict[str, str]) -> tuple[list, list, list]: """ Executes Domain enrichment against RST Threat Feed. Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the Domain indicators. list(dict): the results to return into XSOAR's context. list(dict): the raw results to return into XSOAR's context. """ domains = argToList(args.get("domain", "")) threshold = int(args.get("threshold", DEFAULT_THRESHOLD)) if check_arg_type("threshold_domain", str(threshold)) != "": raise Exception(str(threshold) + ": threshold must be from 0 to 100") markdown = [] raw_results = [] indicators = [] for domain in domains: markdown_item = "" domainregex = re.compile(DOMAINREGEX) domainmatch = domainregex.fullmatch(domain) if domainmatch: indicator = client.get_indicator(domain) else: raise Exception("is not valid Domain name") if "error" in indicator: if indicator["error"] == "Not Found": score = Common.DBotScore( indicator=domain, indicator_type=DBotScoreType.DOMAIN, integration_name="RST Cloud", score=Common.DBotScore.NONE, reliability=demisto.params().get("integrationReliability"), ) markdown_item += f"Domain: {domain} not found\n" raw_results.append(parse_indicator_response(indicator, "Domain")) indicators.append(RSTDomain(domain=indicator["ioc_value"], dbot_score=score)) markdown.append(markdown_item) continue else: raise Exception( f"RST Threat Feed API error while getting a response for {indicator['ioc_value']}: {indicator['error']}\n" ) else: total_score = int(indicator.get("score", {}).get("total")) calc_score = calculate_score(total_score, threshold, "domain", int(indicator.get("lseen", ""))) dbot_score = Common.DBotScore( indicator=indicator["ioc_value"], indicator_type=DBotScoreType.DOMAIN, integration_name="RST Cloud", score=calc_score, malicious_description=indicator.get("description", ""), reliability=demisto.params().get("integrationReliability"), ) human_readable_score = "" if calc_score == 3: human_readable_score = "Malicious" if calc_score == 2: human_readable_score = "Suspicious" result = RSTDomain( domain=indicator["ioc_value"], dns=indicator.get("resolved", {}).get("ip", "").get("a", ""), creation_date=indicator.get("whois", {}).get("created", ""), updated_date=indicator.get("whois", {}).get("updated", ""), expiration_date=indicator.get("whois", {}).get("expires", ""), registrar_name=indicator.get("whois", {}).get("registrar", ""), registrant_name=indicator.get("whois", {}).get("registrant", ""), dbot_score=dbot_score, tags=indicator.get("tags", "").get("str", ""), ) table = { "Score": total_score, "Relevance:": human_readable_score, "Threat": ", ".join(threat for threat in indicator.get("threat", "")), "Last Seen": time.strftime("%Y-%m-%d", time.localtime(int(indicator.get("lseen", "")))), "Description": f"{string_to_context_key(indicator.get('description', ''))}\n", "Tags": ", ".join(tag for tag in indicator.get("tags", "").get("str", "")), } markdown_item += tableToMarkdown( f'RST Threat Feed Domain Reputation for: {indicator["ioc_value"]}\n', table, removeNull=True ) markdown.append(markdown_item) raw_results.append(parse_indicator_response(indicator, "Domain")) indicators.append(result) return markdown, raw_results, indicators def url_command(client: Client, args: dict[str, str]) -> tuple[list, list, list]: """ Executes URL enrichment against RST Threat Feed. Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the URL indicators. list(dict): the results to return into XSOAR's context. list(dict): the raw results to return into XSOAR's context. """ urls = argToList(args.get("url", "")) threshold = int(args.get("threshold", DEFAULT_THRESHOLD)) if check_arg_type("threshold_url", str(threshold)) != "": raise Exception(str(threshold) + ": threshold must be from 0 to 100") markdown = [] raw_results = [] indicators = [] for url in urls: markdown_item = "" urlregex = re.compile(URLREGEX) urlmatch = urlregex.fullmatch(url) if urlmatch: indicator = client.get_indicator(url) else: raise Exception("is not valid URL") if "error" in indicator: if indicator["error"] == "Not Found": score = Common.DBotScore( indicator=url, indicator_type=DBotScoreType.URL, integration_name="RST Cloud", score=Common.DBotScore.NONE, reliability=demisto.params().get("integrationReliability"), ) markdown_item += f"URL: {url} not found\n" raw_results.append(parse_indicator_response(indicator, "URL")) indicators.append(RSTUrl(url=indicator["ioc_value"], dbot_score=score)) markdown.append(markdown_item) continue else: raise Exception( f"RST Threat Feed API error while getting a response for {indicator['ioc_value']}: {indicator['error']}\n" ) else: total_score = int(indicator.get("score", {}).get("total")) calc_score = calculate_score(total_score, threshold, "url", int(indicator.get("lseen", ""))) dbot_score = Common.DBotScore( indicator=indicator["ioc_value"], indicator_type=DBotScoreType.URL, integration_name="RST Cloud", score=calc_score, malicious_description=indicator.get("description", ""), reliability=demisto.params().get("integrationReliability"), ) result = RSTUrl(url=indicator["ioc_value"], dbot_score=dbot_score, tags=indicator.get("tags", "").get("str", "")) human_readable_score = "" if calc_score == 3: human_readable_score = "Malicious" if calc_score == 2: human_readable_score = "Suspicious" table = { "Score": total_score, "Relevance": human_readable_score, "Threat": ", ".join(threat for threat in indicator.get("threat", "")), "Last Seen": time.strftime("%Y-%m-%d", time.localtime(int(indicator.get("lseen", "")))), "Description": f"{string_to_context_key(indicator.get('description', ''))}\n", "Tags": ", ".join(tag for tag in indicator.get("tags", "").get("str", "")), } markdown_item += tableToMarkdown( f'RST Threat Feed URL Reputation for: {indicator["ioc_value"]}\n', table, removeNull=True ) markdown.append(markdown_item) raw_results.append(parse_indicator_response(indicator, "URL")) indicators.append(result) return markdown, raw_results, indicators def file_command(client: Client, args: dict[str, str]) -> tuple[list, list, list]: """ Executes File enrichment against RST Threat Feed. Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the Hash indicators. list(dict): the results to return into XSOAR's context. list(dict): the raw results to return into XSOAR's context. """ hashes = argToList(args.get("file", "")) threshold = int(args.get("threshold", DEFAULT_THRESHOLD)) if check_arg_type("threshold_hash", str(threshold)) != "": raise Exception(str(threshold) + ": threshold must be from 0 to 100") markdown = [] raw_results = [] indicators = [] for hash in hashes: markdown_item = "" md5regex = re.compile(MD5REGEX) sha1regex = re.compile(SHA1REGEX) sha256regex = re.compile(SHA256REGEX) md5match = md5regex.fullmatch(hash) sha1match = sha1regex.fullmatch(hash) sha256match = sha256regex.fullmatch(hash) if md5match or sha1match or sha256match: indicator = client.get_indicator(hash) else: raise Exception("is not a valid Hash") if "error" in indicator: if indicator["error"] == "Not Found": score = Common.DBotScore( indicator=hash, indicator_type=DBotScoreType.FILE, integration_name="RST Cloud", score=Common.DBotScore.NONE, reliability=demisto.params().get("integrationReliability"), ) markdown_item += f"Hash: {hash} not found\n" raw_results.append(parse_indicator_response(indicator, "File")) if md5match: indicators.append(Common.File(md5=hash, sha1="", sha256="", dbot_score=score)) if sha1match: indicators.append(Common.File(md5="", sha1=hash, sha256="", dbot_score=score)) if sha256match: indicators.append(Common.File(md5="", sha1="", sha256=hash, dbot_score=score)) markdown.append(markdown_item) continue else: raise Exception( f"RST Threat Feed API error while getting a response for {indicator['ioc_value']}: {indicator['error']}\n" ) else: total_score = int(indicator.get("score", {}).get("total")) calc_score = calculate_score(total_score, threshold, "hash", int(indicator.get("lseen", ""))) dbot_score = Common.DBotScore( indicator=indicator["ioc_value"], indicator_type=DBotScoreType.FILE, integration_name="RST Cloud", score=calc_score, malicious_description=indicator.get("description", ""), reliability=demisto.params().get("integrationReliability"), ) result = Common.File(md5=indicator["md5"], sha1=indicator["sha1"], sha256=indicator["sha256"], dbot_score=dbot_score) human_readable_score = "" if calc_score == 3: human_readable_score = "Malicious" if calc_score == 2: human_readable_score = "Suspicious" table = { "Score": total_score, "Relevance": human_readable_score, "Threat": ", ".join(threat for threat in indicator.get("threat", "")), "Last Seen": time.strftime("%Y-%m-%d", time.localtime(int(indicator.get("lseen", 0)))), "Description": f"{string_to_context_key(indicator.get('description', ''))}\n", "Tags": ", ".join(tag for tag in indicator.get("tags", "").get("str", "")), } markdown_item += tableToMarkdown( f'RST Threat Feed File Reputation for: {indicator["ioc_value"]}\n', table, removeNull=True ) markdown.append(markdown_item) raw_results.append(parse_indicator_response(indicator, "File")) indicators.append(result) return markdown, raw_results, indicators def submit_command(client: Client, args: dict[str, str]) -> list: """ Submits a new indicator to RST Threat Feed via API Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the API response """ iocs = argToList(args.get("ioc", "")) description = argToList(args.get("description", "manual submission")) markdown = [] for i in range(len(iocs)): indicator = client.submit_indicator(iocs[i], description[i]) if "status" in indicator: markdown.append(f"Indicator: {iocs[i]} was submitted as a potential threat indicator to RST Cloud\n") elif "error" in indicator: raise Exception( f"Indicator: {iocs[i]} was not submitted successfully due to the following error: {indicator['error']}\n" ) return markdown def submitfp_command(client: Client, args: dict[str, str]) -> list: """ Submits a potential False Positive indicator to RST Threat Feed via API Args: client (Client): RST Threat Feed client. args (Dict[str, str]): the arguments for the command. Returns: list(str): human readable presentation of the API response """ iocs = argToList(args.get("ioc", "")) description = argToList(args.get("description", "manual submission")) markdown = [] for i in range(len(iocs)): indicator = client.submit_falsepositive(iocs[i], description[i]) if indicator["status"]: markdown.append(f"Indicator: {iocs[i]} was submitted as False Positive to RST Cloud\n") elif indicator["error"]: raise Exception( f"Indicator: {iocs[i]} was not submitted successfully due to the following error: {indicator['error']}\n" ) return markdown """ MAIN FUNCTION """ def main(): params = demisto.params() client = Client(params.get("apikey"), params.get("url"), params.get("insecure", False), params.get("proxy", False)) demisto.info("RST: Client initialised...") command = demisto.command() demisto.info(f"RST: Command being called is {command}") try: if command == "test-module": demisto.results(test_module(client)) elif command == "ip": markdown, raw_results, indicators = ip_command(client, demisto.args()) for i in range(len(raw_results)): output = CommandResults( readable_output=markdown[i], outputs_prefix="RST.IP", outputs_key_field="Address", outputs=raw_results[i], indicator=indicators[i], ) return_results(output) elif command == "domain": markdown, raw_results, indicators = domain_command(client, demisto.args()) for i in range(len(raw_results)): output = CommandResults( readable_output=markdown[i], outputs_prefix="RST.Domain", outputs_key_field="Name", outputs=raw_results[i], indicator=indicators[i], ) return_results(output) elif command == "url": markdown, raw_results, indicators = url_command(client, demisto.args()) for i in range(len(raw_results)): output = CommandResults( readable_output=markdown[i], outputs_prefix="RST.URL", outputs_key_field="Data", outputs=raw_results[i], indicator=indicators[i], ) return_results(output) elif command == "file": markdown, raw_results, indicators = file_command(client, demisto.args()) for i in range(len(raw_results)): output = CommandResults( readable_output=markdown[i], outputs_prefix="RST.File", outputs_key_field="UUID", outputs=raw_results[i], indicator=indicators[i], ) return_results(output) elif command == "rst-submit-new": markdown = submit_command(client, demisto.args()) for i in range(len(markdown)): output = CommandResults(readable_output=markdown[i]) return_results(output) elif command == "rst-submit-fp": markdown = submitfp_command(client, demisto.args()) for i in range(len(markdown)): output = CommandResults(readable_output=markdown[i]) return_results(output) else: raise Exception("Command not found.") except Exception as e: return_error(f"Failed to execute {command} command. Error: {e!s}") if __name__ in ("__main__", "__builtin__", "builtins"): main()