MISP
/
misp-modules
mirror of https://github.com/MISP/misp-modules
2
0
Fork 0
Code Issues Releases Wiki Activity
misp-modules/misp_modules/modules/expansion/google_threat_intelligence.py

323 lines
10 KiB
Python
Raw Blame History

#!/usr/local/bin/python
# Copyright © 2024 The Google Threat Intelligence authors. All Rights Reserved.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Google Threat Intelligence MISP expansion module."""
from urllib import parse
import vt
import pymisp
MISP_ATTRIBUTES = {
'input': [
'hostname',
'domain',
'ip-src',
'ip-dst',
'md5',
'sha1',
'sha256',
'url',
],
'format': 'misp_standard',
}
MODULE_INFO = {
'version': '1',
'author': 'Google Threat Intelligence team',
'description': ('An expansion module to have the observable\'s threat'
' score assessed by Google Threat Intelligence.'),
'module-type': ['expansion'],
'config': [
'apikey',
'event_limit',
'proxy_host',
'proxy_port',
'proxy_username',
'proxy_password'
]
}
DEFAULT_RESULTS_LIMIT = 10
class GoogleThreatIntelligenceParser:
"""Main parser class to create the MISP event."""
def __init__(self, client: vt.Client, limit: int) -> None:
self.client = client
self.limit = limit or DEFAULT_RESULTS_LIMIT
self.misp_event = pymisp.MISPEvent()
self.attribute = pymisp.MISPAttribute()
self.parsed_objects = {}
self.input_types_mapping = {
'ip-src': self.parse_ip,
'ip-dst': self.parse_ip,
'domain': self.parse_domain,
'hostname': self.parse_domain,
'md5': self.parse_hash,
'sha1': self.parse_hash,
'sha256': self.parse_hash,
'url': self.parse_url
}
self.proxies = None
def query_api(self, attribute: dict) -> None:
"""Get data from the API and parse it."""
self.attribute.from_dict(**attribute)
self.input_types_mapping[self.attribute.type](self.attribute.value)
def get_results(self) -> dict:
"""Serialize the MISP event."""
event = self.misp_event.to_dict()
results = {
key: event[key] for key in ('Attribute', 'Object') \
if (key in event and event[key])
}
return {'results': results}
def create_gti_report_object(self, report):
"""Create GTI report object."""
report = report.to_dict()
permalink = ('https://www.virustotal.com/gui/'
f"{report['type']}/{report['id']}")
report_object = pymisp.MISPObject('Google-Threat-Intel-report')
report_object.add_attribute('permalink', type='link', value=permalink)
report_object.add_attribute(
'Threat Score', type='text',
value=get_key(
report, 'attributes.gti_assessment.threat_score.value'))
report_object.add_attribute(
'Verdict', type='text',
value=get_key(
report, 'attributes.gti_assessment.verdict.value').replace(
'VERDICT_', ''))
report_object.add_attribute(
'Severity', type='text',
value=get_key(
report, 'attributes.gti_assessment.severity.value').replace(
'SEVERITY_', ''))
report_object.add_attribute(
'Threat Label', type='text',
value=get_key(
report, ('attributes.popular_threat_classification'
'.suggested_threat_label')))
self.misp_event.add_object(**report_object)
return report_object.uuid
def parse_domain(self, domain: str) -> str:
"""Create domain MISP object."""
domain_report = self.client.get_object(f'/domains/{domain}')
# DOMAIN
domain_object = pymisp.MISPObject('domain-ip')
domain_object.add_attribute(
'domain', type='domain', value=domain_report.id)
report_uuid = self.create_gti_report_object(domain_report)
domain_object.add_reference(report_uuid, 'analyzed-with')
self.misp_event.add_object(**domain_object)
return domain_object.uuid
def parse_hash(self, file_hash: str) -> str:
"""Create hash MISP object."""
file_report = self.client.get_object(f'/files/{file_hash}')
file_object = pymisp.MISPObject('file')
for hash_type in ('md5', 'sha1', 'sha256'):
file_object.add_attribute(
hash_type,
**{'type': hash_type, 'value': file_report.get(hash_type)})
report_uuid = self.create_gti_report_object(file_report)
file_object.add_reference(report_uuid, 'analyzed-with')
self.misp_event.add_object(**file_object)
return file_object.uuid
def parse_ip(self, ip: str) -> str:
"""Create ip MISP object."""
ip_report = self.client.get_object(f'/ip_addresses/{ip}')
# IP
ip_object = pymisp.MISPObject('domain-ip')
ip_object.add_attribute('ip', type='ip-dst', value=ip_report.id)
report_uuid = self.create_gti_report_object(ip_report)
ip_object.add_reference(report_uuid, 'analyzed-with')
self.misp_event.add_object(**ip_object)
return ip_object.uuid
def parse_url(self, url: str) -> str:
"""Create URL MISP object."""
url_id = vt.url_id(url)
url_report = self.client.get_object(f'/urls/{url_id}')
url_object = pymisp.MISPObject('url')
url_object.add_attribute('url', type='url', value=url_report.url)
report_uuid = self.create_gti_report_object(url_report)
url_object.add_reference(report_uuid, 'analyzed-with')
self.misp_event.add_object(**url_object)
return url_object.uuid
def get_key(dictionary, key, default_value=''):
"""Get value from nested dictionaries."""
dictionary = dictionary or {}
keys = key.split('.')
field_name = keys.pop()
for k in keys:
if k not in dictionary:
return default_value
dictionary = dictionary[k]
return dictionary.get(field_name, default_value)
def get_proxy_settings(config: dict) -> dict:
"""Returns proxy settings in the requests format or None if not set up."""
proxies = None
host = config.get('proxy_host')
port = config.get('proxy_port')
username = config.get('proxy_username')
password = config.get('proxy_password')
if host:
if not port:
raise KeyError(
('The google_threat_intelligence_proxy_host config is set, '
'please also set the virustotal_proxy_port.'))
parsed = parse.urlparse(host)
if 'http' in parsed.scheme:
scheme = 'http'
else:
scheme = parsed.scheme
netloc = parsed.netloc
host = f'{netloc}:{port}'
if username:
if not password:
raise KeyError(('The google_threat_intelligence_'
' proxy_host config is set, please also'
' set the virustotal_proxy_password.'))
auth = f'{username}:{password}'
host = auth + '@' + host
proxies = {
'http': f'{scheme}://{host}',
'https': f'{scheme}://{host}'
}
return proxies
def dict_handler(request: dict):
"""MISP entry point fo the module."""
if not request.get('config') or not request['config'].get('apikey'):
return {
'error': ('A Google Threat Intelligence api '
'key is required for this module.')
}
if not request.get('attribute'):
return {
'error': ('This module requires an "attribute" field as input,'
' which should contain at least a type, a value and an'
' uuid.')
}
if request['attribute']['type'] not in MISP_ATTRIBUTES['input']:
return {'error': 'Unsupported attribute type.'}
event_limit = request['config'].get('event_limit')
attribute = request['attribute']
try:
proxy_settings = get_proxy_settings(request.get('config'))
client = vt.Client(
request['config']['apikey'],
headers={
'x-tool': 'MISPModuleGTIExpansion',
},
proxy=proxy_settings['http'] if proxy_settings else None)
parser = GoogleThreatIntelligenceParser(
client, int(event_limit) if event_limit else None)
parser.query_api(attribute)
except vt.APIError as ex:
return {'error': ex.message}
except KeyError as ex:
return {'error': str(ex)}
return parser.get_results()
def introspection():
"""Returns the module input attributes required."""
return MISP_ATTRIBUTES
def version():
"""Returns the module metadata."""
return MODULE_INFO
if __name__ == '__main__':
# Testing/debug calls.
import os
api_key = os.getenv('GTI_API_KEY')
# File
request_data = {
'config': {'apikey': api_key},
'attribute': {
'type': 'sha256',
'value': ('ed01ebfbc9eb5bbea545af4d01bf5f10'
'71661840480439c6e5babe8e080e41aa')
}
}
response = dict_handler(request_data)
report_obj = response['results']['Object'][0]
print(report_obj.to_dict())
# URL
request_data = {
'config': {'apikey': api_key},
'attribute': {
'type': 'url',
'value': 'http://47.21.48.182:60813/Mozi.a'
}
}
response = dict_handler(request_data)
report_obj = response['results']['Object'][0]
print(report_obj.to_dict())
# Ip
request_data = {
'config': {'apikey': api_key},
'attribute': {
'type': 'ip-src',
'value': '180.72.148.38'
}
}
response = dict_handler(request_data)
report_obj = response['results']['Object'][0]
print(report_obj.to_dict())
# Domain
request_data = {
'config': {'apikey': api_key},
'attribute': {
'type': 'domain',
'value': 'qexyhuv.com'
}
}
response = dict_handler(request_data)
report_obj = response['results']['Object'][0]
print(report_obj.to_dict())
Reference in New Issue View Git Blame Copy Permalink