mirror of https://github.com/MISP/misp-modules
Modules for expansion services, import and export in MISP
http://misp.github.io/misp-modules
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
124 lines
5.7 KiB
124 lines
5.7 KiB
from pymisp import MISPEvent, MISPObject |
|
import json |
|
import requests |
|
import base64 |
|
from urllib.parse import quote |
|
|
|
moduleinfo = {'version': '1.0', |
|
'author': 'Ben Verschaeren', |
|
'description': 'SOPHOSLabs Intelix Integration', |
|
'module-type': ['expansion']} |
|
|
|
moduleconfig = ['client_id', 'client_secret'] |
|
|
|
misperrors = {'error': 'Error'} |
|
|
|
misp_types_in = ['sha256', 'ip', 'ip-src', 'ip-dst', 'uri', 'url', 'domain', 'hostname'] |
|
|
|
mispattributes = {'input': misp_types_in, |
|
'format': 'misp_standard'} |
|
|
|
|
|
class SophosLabsApi(): |
|
def __init__(self, client_id, client_secret): |
|
self.misp_event = MISPEvent() |
|
self.client_id = client_id |
|
self.client_secret = client_secret |
|
self.authToken = f"{self.client_id}:{self.client_secret}" |
|
self.baseurl = 'de.api.labs.sophos.com' |
|
d = {'grant_type': 'client_credentials'} |
|
h = {'Authorization': f"Basic {base64.b64encode(self.authToken.encode('UTF-8')).decode('ascii')}", |
|
'Content-Type': 'application/x-www-form-urlencoded'} |
|
r = requests.post('https://api.labs.sophos.com/oauth2/token', headers=h, data=d) |
|
if r.status_code == 200: |
|
j = json.loads(r.text) |
|
self.accessToken = j['access_token'] |
|
|
|
def get_result(self): |
|
event = json.loads(self.misp_event.to_json()) |
|
results = {key: event[key] for key in ('Attribute', 'Object') if (key in event and event[key])} |
|
return {'results': results} |
|
|
|
def hash_lookup(self, filehash): |
|
sophos_object = MISPObject('SOPHOSLabs Intelix SHA256 Report') |
|
h = {"Authorization": f"{self.accessToken}"} |
|
r = requests.get(f"https://{self.baseurl}/lookup/files/v1/{filehash}", headers=h) |
|
if r.status_code == 200: |
|
j = json.loads(r.text) |
|
if 'reputationScore' in j: |
|
sophos_object.add_attribute('Reputation Score', type='text', value=j['reputationScore']) |
|
if 0 <= j['reputationScore'] <= 19: |
|
sophos_object.add_attribute('Decision', type='text', value='This file is malicious') |
|
if 20 <= j['reputationScore'] <= 29: |
|
sophos_object.add_attribute('Decision', type='text', value='This file is potentially unwanted') |
|
if 30 <= j['reputationScore'] <= 69: |
|
sophos_object.add_attribute('Decision', type='text', value='This file is unknown and suspicious') |
|
if 70 <= j['reputationScore'] <= 100: |
|
sophos_object.add_attribute('Decision', type='text', value='This file is known good') |
|
if 'detectionName' in j: |
|
sophos_object.add_attribute('Detection Name', type='text', value=j['detectionName']) |
|
else: |
|
sophos_object.add_attribute('Detection Name', type='text', value='No name associated with this IoC') |
|
self.misp_event.add_object(**sophos_object) |
|
|
|
def ip_lookup(self, ip): |
|
sophos_object = MISPObject('SOPHOSLabs Intelix IP Category Lookup') |
|
h = {"Authorization": f"{self.accessToken}"} |
|
r = requests.get(f"https://{self.baseurl}/lookup/ips/v1/{ip}", headers=h) |
|
if r.status_code == 200: |
|
j = json.loads(r.text) |
|
if 'category' in j: |
|
for c in j['category']: |
|
sophos_object.add_attribute('IP Address Categorisation', type='text', value=c) |
|
else: |
|
sophos_object.add_attribute('IP Address Categorisation', type='text', value='No category assocaited with IoC') |
|
self.misp_event.add_object(**sophos_object) |
|
|
|
def url_lookup(self, url): |
|
sophos_object = MISPObject('SOPHOSLabs Intelix URL Lookup') |
|
h = {"Authorization": f"{self.accessToken}"} |
|
r = requests.get(f"https://{self.baseurl}/lookup/urls/v1/{quote(url, safe='')}", headers=h) |
|
if r.status_code == 200: |
|
j = json.loads(r.text) |
|
if 'productivityCategory' in j: |
|
sophos_object.add_attribute('URL Categorisation', type='text', value=j['productivityCategory']) |
|
else: |
|
sophos_object.add_attribute('URL Categorisation', type='text', value='No category assocaited with IoC') |
|
|
|
if 'riskLevel' in j: |
|
sophos_object.add_attribute('URL Risk Level', type='text', value=j['riskLevel']) |
|
else: |
|
sophos_object.add_attribute('URL Risk Level', type='text', value='No risk level associated with IoC') |
|
|
|
if 'securityCategory' in j: |
|
sophos_object.add_attribute('URL Security Category', type='text', value=j['securityCategory']) |
|
else: |
|
sophos_object.add_attribute('URL Security Category', type='text', value='No Security Category associated with IoC') |
|
self.misp_event.add_object(**sophos_object) |
|
|
|
|
|
def handler(q=False): |
|
if q is False: |
|
return False |
|
j = json.loads(q) |
|
if not j.get('config') or not j['config'].get('client_id') or not j['config'].get('client_secret'): |
|
misperrors['error'] = "Missing client_id or client_secret value for SOPHOSLabs Intelix. \ |
|
It's free to sign up here https://aws.amazon.com/marketplace/pp/B07SLZPMCS." |
|
return misperrors |
|
client = SophosLabsApi(j['config']['client_id'], j['config']['client_secret']) |
|
if j['attribute']['type'] == "sha256": |
|
client.hash_lookup(j['attribute']['value1']) |
|
if j['attribute']['type'] in ['ip-dst', 'ip-src', 'ip']: |
|
client.ip_lookup(j["attribute"]["value1"]) |
|
if j['attribute']['type'] in ['uri', 'url', 'domain', 'hostname']: |
|
client.url_lookup(j["attribute"]["value1"]) |
|
return client.get_result() |
|
|
|
|
|
def introspection(): |
|
return mispattributes |
|
|
|
|
|
def version(): |
|
moduleinfo['config'] = moduleconfig |
|
return moduleinfo
|
|
|