|
33 | 33 | LIST_IP = "explore/bulk/ip2asn"
|
34 | 34 | ASN_REPUTATION = "explore/ipreputation/history/asn"
|
35 | 35 | ASN_TAKEDOWN_REPUTATION = "explore/takedownreputation/asn"
|
| 36 | +IPV4_REPUTATION = "explore/ipreputation/history/ipv4" |
36 | 37 |
|
37 | 38 | ''' COMMANDS INPUTS '''
|
38 | 39 |
|
|
188 | 189 | InputArgument(name='limit',
|
189 | 190 | description='The maximum number of reputation history records to retrieve.')
|
190 | 191 | ]
|
| 192 | +IPV4_REPUTATION_INPUTS = [ |
| 193 | + InputArgument(name='ipv4', # option 1 |
| 194 | + description='IPv4 address for which information needs to be retrieved', |
| 195 | + required=True), |
| 196 | + InputArgument(name='explain', |
| 197 | + description='Show the information used to calculate the reputation score'), |
| 198 | + InputArgument(name='limit', |
| 199 | + description='The maximum number of reputation history to retrieve') |
| 200 | + ] |
191 | 201 |
|
192 | 202 |
|
193 | 203 |
|
|
417 | 427 | OutputArgument(name='Allocation_Date', output_type=int, description='The date when the ASN was allocated (YYYYMMDD).'),
|
418 | 428 | OutputArgument(name='Takedown_Reputation', output_type=int, description='The takedown reputation score for the ASN.')
|
419 | 429 | ]
|
| 430 | +IPV4_REPUTATION_OUTPUTS = [ |
| 431 | + OutputArgument(name='Date', output_type=int, description='Date when the reputation information was retrieved.'), |
| 432 | + OutputArgument(name='IP', output_type=str, description='IPv4 address for which the reputation is calculated.'), |
| 433 | + OutputArgument(name='Reputation.Score', output_type=int, description='Reputation score for the given IP address.') |
| 434 | + ] |
| 435 | + |
420 | 436 |
|
421 | 437 |
|
422 | 438 |
|
@@ -1025,6 +1041,26 @@ def get_asn_takedown_reputation(self, asn: str, limit: Optional[int] = None, exp
|
1025 | 1041 |
|
1026 | 1042 | return response.get('response', {}).get('takedown_reputation', {})
|
1027 | 1043 |
|
| 1044 | + def get_ipv4_reputation(self, ipv4: str, explain: bool = False, limit: int = None) -> List[Dict[str, Any]]: |
| 1045 | + """ |
| 1046 | + Retrieve reputation information for an IPv4 address. |
| 1047 | + """ |
| 1048 | + url_suffix = f"{IPV4_REPUTATION}/{ipv4}" |
| 1049 | + query_params = {} |
| 1050 | + |
| 1051 | + if explain: |
| 1052 | + query_params['explain'] = 'true' |
| 1053 | + if limit: |
| 1054 | + query_params['limit'] = limit |
| 1055 | + |
| 1056 | + raw_response = self._http_request( |
| 1057 | + method='GET', |
| 1058 | + url_suffix=url_suffix, |
| 1059 | + params=query_params |
| 1060 | + ) |
| 1061 | + ipv4_reputation = raw_response.get('response', {}).get('ip_reputation_history', []) |
| 1062 | + return ipv4_reputation |
| 1063 | + |
1028 | 1064 |
|
1029 | 1065 | ''' HELPER FUNCTIONS '''
|
1030 | 1066 | def filter_none_values(params: Dict[str, Any]) -> Dict[str, Any]:
|
@@ -1941,9 +1977,9 @@ def get_table_headers(explain: bool) -> list:
|
1941 | 1977 | @metadata_collector.command(
|
1942 | 1978 | command_name="silentpush-get-asn-takedown-reputation",
|
1943 | 1979 | inputs_list=ASN_TAKEDOWN_REPUTATION_INPUTS,
|
1944 |
| - outputs_prefix="SilentPush.", |
| 1980 | + outputs_prefix="SilentPush.ASNTakedownReputation", |
1945 | 1981 | outputs_list=ASN_TAKEDOWN_REPUTATION_OUTPUTS,
|
1946 |
| - description="This command Retrieve the takedown reputation information for an Autonomous System Number (ASN)." |
| 1982 | + description="This command retrieve the takedown reputation information for an Autonomous System Number (ASN)." |
1947 | 1983 | )
|
1948 | 1984 | def get_asn_takedown_reputation_command(client: Client, args: dict) -> CommandResults:
|
1949 | 1985 | """
|
@@ -2017,6 +2053,67 @@ def get_asn_takedown_reputation_command(client: Client, args: dict) -> CommandRe
|
2017 | 2053 | raw_response=response
|
2018 | 2054 | )
|
2019 | 2055 |
|
| 2056 | +@metadata_collector.command( |
| 2057 | + command_name="silentpush-get-ipv4-reputation", |
| 2058 | + inputs_list=IPV4_REPUTATION_INPUTS, |
| 2059 | + outputs_prefix="SilentPush.", |
| 2060 | + outputs_list=IPV4_REPUTATION_OUTPUTS, |
| 2061 | + description="This command retrieve the reputation information for an IPv4." |
| 2062 | +) |
| 2063 | +def get_ipv4_reputation_command(client: Client, args: Dict[str, Any]) -> CommandResults: |
| 2064 | + """ |
| 2065 | + Retrieves the reputation data for a given IPv4 address from the client. |
| 2066 | +
|
| 2067 | + Args: |
| 2068 | + client (Client): The client to interact with the reputation service. |
| 2069 | + args (Dict[str, Any]): Arguments passed to the command, including the IPv4 address, explain flag, and limit. |
| 2070 | +
|
| 2071 | + Returns: |
| 2072 | + CommandResults: The results of the command including the IPv4 reputation data. |
| 2073 | + """ |
| 2074 | + ipv4 = args.get('ipv4') |
| 2075 | + |
| 2076 | + if not ipv4: |
| 2077 | + raise DemistoException("IPv4 address is required") |
| 2078 | + |
| 2079 | + explain = argToBoolean(args.get('explain', "false")) |
| 2080 | + limit = arg_to_number(args.get('limit')) |
| 2081 | + |
| 2082 | + raw_response = client.get_ipv4_reputation(ipv4, explain, limit) |
| 2083 | + |
| 2084 | + # If no data is found for the provided IPv4 address, return a message |
| 2085 | + if not raw_response: |
| 2086 | + return CommandResults( |
| 2087 | + readable_output=f"No reputation data found for IPv4: {ipv4}", |
| 2088 | + outputs_prefix='SilentPush.IPv4Reputation', |
| 2089 | + outputs_key_field='ip', |
| 2090 | + outputs={'ip': ipv4}, |
| 2091 | + raw_response=raw_response |
| 2092 | + ) |
| 2093 | + |
| 2094 | + latest_reputation = raw_response[0] |
| 2095 | + |
| 2096 | + # Prepare reputation data for output |
| 2097 | + reputation_data = { |
| 2098 | + 'IP': latest_reputation.get('ip', ipv4), |
| 2099 | + 'Date': latest_reputation.get('date'), |
| 2100 | + 'Reputation Score': latest_reputation.get('ip_reputation') |
| 2101 | + } |
| 2102 | + |
| 2103 | + # Convert data to markdown table for readable output |
| 2104 | + readable_output = tableToMarkdown( |
| 2105 | + f'IPv4 Reputation Information for {ipv4}', |
| 2106 | + [reputation_data] |
| 2107 | + ) |
| 2108 | + |
| 2109 | + return CommandResults( |
| 2110 | + outputs_prefix='SilentPush.IPv4Reputation', |
| 2111 | + outputs_key_field='ip', |
| 2112 | + outputs=reputation_data, |
| 2113 | + readable_output=readable_output, |
| 2114 | + raw_response=raw_response |
| 2115 | + ) |
| 2116 | + |
2020 | 2117 |
|
2021 | 2118 | ''' MAIN FUNCTION '''
|
2022 | 2119 |
|
@@ -2084,6 +2181,9 @@ def main() -> None:
|
2084 | 2181 |
|
2085 | 2182 | elif demisto.command() == 'silentpush-get-asn-takedown-reputation':
|
2086 | 2183 | return_results(get_asn_takedown_reputation_command(client, demisto.args()))
|
| 2184 | + |
| 2185 | + elif demisto.command() == 'silentpush-get-ipv4-reputation': |
| 2186 | + return_results(get_ipv4_reputation_command(client, demisto.args())) |
2087 | 2187 |
|
2088 | 2188 | except Exception as e:
|
2089 | 2189 | demisto.error(traceback.format_exc()) # print the traceback
|
|
0 commit comments