import json import requests import base64 from . import check_input_attribute, checking_error, standard_error_message from pymisp import MISPEvent, MISPObject 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 to_check = (('type', 'value'), ('type', 'value1')) if not j.get('attribute') or not any(check_input_attribute(j['attribute'], requirements=check) for check in to_check): return {'error': f'{standard_error_message}, {checking_error}.'} attribute = j['attribute'] if attribute['type'] not in misp_types_in: return {'error': 'Unsupported attribute type.'} client = SophosLabsApi(j['config']['client_id'], j['config']['client_secret']) mapping = { 'sha256': 'hash_lookup', 'ip-dst': 'ip_lookup', 'ip-src': 'ip_lookup', 'ip': 'ip_lookup', 'uri': 'url_lookup', 'url': 'url_lookup', 'domain': 'url_lookup', 'hostname': 'url_lookup' } attribute_value = attribute['value'] if 'value' in attribute else attribute['value1'] getattr(client, mapping[attribute['type']])(attribute_value) return client.get_result() def introspection(): return mispattributes def version(): moduleinfo['config'] = moduleconfig return moduleinfo