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

fix: Pep8 related fixes.

Browse Source
pull/274/head
Raphaël Vinot 4 years ago
parent 7a7b7b109f
commit 454c9e0f43
8 changed files with 193 additions and 191 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      misp_modules/modules/expansion/circl_passivedns.py
  2. 207
      misp_modules/modules/expansion/xforceexchange.py
  3. 112
      misp_modules/modules/export_mod/liteexport.py
  4. 2
      misp_modules/modules/export_mod/nexthinkexport.py
  5. 2
      misp_modules/modules/export_mod/osqueryexport.py
  6. 53
      misp_modules/modules/export_mod/pdfexport.py
  7. 2
      misp_modules/modules/import_mod/openiocimport.py
  8. 4
      misp_modules/modules/import_mod/threatanalyzer_import.py

2
misp_modules/modules/expansion/circl_passivedns.py
Unescape View File

@ -32,7 +32,7 @@ def handler(q=False):
res = x.query(toquery)
out = ''
for v in res:
out = out + "{} ".format(v['rdata'])
out = out + "{} ".format(v['rdata'])
r = {'results': [{'types': mispattributes['output'], 'values': out}]}
return r

207
misp_modules/modules/expansion/xforceexchange.py
Unescape View File

@ -1,103 +1,104 @@
import requests
import json
import sys
BASEurl = "https://api.xforce.ibmcloud.com/"
extensions = {"ip1": "ipr/%s",
"ip2": "ipr/malware/%s",
"url": "url/%s",
"hash": "malware/%s",
"vuln": "/vulnerabilities/search/%s",
"dns": "resolve/%s"}
sys.path.append('./')
misperrors = {'error': 'Error'}
mispattributes = {'input': ['ip-src', 'ip-dst', 'vulnerability', 'md5', 'sha1', 'sha256'],
'output': ['ip-src', 'ip-dst', 'text', 'domain']}
# possible module-types: 'expansion', 'hover' or both
moduleinfo = {'version': '1', 'author': 'Joerg Stephan (@johest)',
'description': 'IBM X-Force Exchange expansion module',
'module-type': ['expansion', 'hover']}
# config fields that your code expects from the site admin
moduleconfig = ["apikey", "event_limit"]
limit = 5000 # Default
def MyHeader(key=False):
global limit
if key is False:
return None
return {"Authorization": "Basic %s " % key,
"Accept": "application/json",
'User-Agent': 'Mozilla 5.0'}
def handler(q=False):
global limit
if q is False:
return False
q = json.loads(q)
key = q["config"]["apikey"]
limit = int(q["config"].get("event_limit", 5))
r = {"results": []}
if "ip-src" in q:
r["results"] += apicall("dns", q["ip-src"], key)
if "ip-dst" in q:
r["results"] += apicall("dns", q["ip-dst"], key)
if "md5" in q:
r["results"] += apicall("hash", q["md5"], key)
if "sha1" in q:
r["results"] += apicall("hash", q["sha1"], key)
if "sha256" in q:
r["results"] += apicall("hash", q["sha256"], key)
if 'vulnerability' in q:
r["results"] += apicall("vuln", q["vulnerability"], key)
if "domain" in q:
r["results"] += apicall("dns", q["domain"], key)
uniq = []
for res in r["results"]:
if res not in uniq:
uniq.append(res)
r["results"] = uniq
return r
def apicall(indicator_type, indicator, key=False):
try:
myURL = BASEurl + (extensions[str(indicator_type)]) % indicator
jsondata = requests.get(myURL, headers=MyHeader(key)).json()
except Exception:
jsondata = None
redata = []
# print(jsondata)
if jsondata is not None:
if indicator_type is "hash":
if "malware" in jsondata:
lopointer = jsondata["malware"]
redata.append({"type": "text", "values": lopointer["risk"]})
if indicator_type is "dns":
if "records" in str(jsondata):
lopointer = jsondata["Passive"]["records"]
for dataset in lopointer:
redata.append({"type": "domain", "values": dataset["value"]})
return redata
def introspection():
return mispattributes
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo
import requests
import json
import sys
BASEurl = "https://api.xforce.ibmcloud.com/"
extensions = {"ip1": "ipr/%s",
"ip2": "ipr/malware/%s",
"url": "url/%s",
"hash": "malware/%s",
"vuln": "/vulnerabilities/search/%s",
"dns": "resolve/%s"}
sys.path.append('./')
misperrors = {'error': 'Error'}
mispattributes = {'input': ['ip-src', 'ip-dst', 'vulnerability', 'md5', 'sha1', 'sha256'],
'output': ['ip-src', 'ip-dst', 'text', 'domain']}
# possible module-types: 'expansion', 'hover' or both
moduleinfo = {'version': '1', 'author': 'Joerg Stephan (@johest)',
'description': 'IBM X-Force Exchange expansion module',
'module-type': ['expansion', 'hover']}
# config fields that your code expects from the site admin
moduleconfig = ["apikey", "event_limit"]
limit = 5000 # Default
def MyHeader(key=False):
global limit
if key is False:
return None
return {"Authorization": "Basic %s " % key,
"Accept": "application/json",
'User-Agent': 'Mozilla 5.0'}
def handler(q=False):
global limit
if q is False:
return False
q = json.loads(q)
key = q["config"]["apikey"]
limit = int(q["config"].get("event_limit", 5))
r = {"results": []}
if "ip-src" in q:
r["results"] += apicall("dns", q["ip-src"], key)
if "ip-dst" in q:
r["results"] += apicall("dns", q["ip-dst"], key)
if "md5" in q:
r["results"] += apicall("hash", q["md5"], key)
if "sha1" in q:
r["results"] += apicall("hash", q["sha1"], key)
if "sha256" in q:
r["results"] += apicall("hash", q["sha256"], key)
if 'vulnerability' in q:
r["results"] += apicall("vuln", q["vulnerability"], key)
if "domain" in q:
r["results"] += apicall("dns", q["domain"], key)
uniq = []
for res in r["results"]:
if res not in uniq:
uniq.append(res)
r["results"] = uniq
return r
def apicall(indicator_type, indicator, key=False):
try:
myURL = BASEurl + (extensions[str(indicator_type)]) % indicator
jsondata = requests.get(myURL, headers=MyHeader(key)).json()
except Exception:
jsondata = None
redata = []
# print(jsondata)
if jsondata is not None:
if indicator_type == "hash":
if "malware" in jsondata:
lopointer = jsondata["malware"]
redata.append({"type": "text", "values": lopointer["risk"]})
if indicator_type == "dns":
if "records" in str(jsondata):
lopointer = jsondata["Passive"]["records"]
for dataset in lopointer:
redata.append(
{"type": "domain", "values": dataset["value"]})
return redata
def introspection():
return mispattributes
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo

112
misp_modules/modules/export_mod/liteexport.py
Unescape View File

@ -16,73 +16,73 @@ responseType = "application/json"
def handler(q=False):
if q is False:
return False
if q is False:
return False
request = json.loads(q)
request = json.loads(q)
config = {}
if "config" in request:
config = request["config"]
else:
config = {"indent_json_export": None}
config = {}
if "config" in request:
config = request["config"]
else:
config = {"indent_json_export": None}
if config['indent_json_export'] is not None:
try:
config['indent_json_export'] = int(config['indent_json_export'])
except Exception:
config['indent_json_export'] = None
if config['indent_json_export'] is not None:
try:
config['indent_json_export'] = int(config['indent_json_export'])
except Exception:
config['indent_json_export'] = None
if 'data' not in request:
return False
if 'data' not in request:
return False
# ~ Misp json structur
liteEvent = {'Event': {}}
# ~ Misp json structur
liteEvent = {'Event': {}}
for evt in request['data']:
rawEvent = evt['Event']
liteEvent['Event']['info'] = rawEvent['info']
liteEvent['Event']['Attribute'] = []
for evt in request['data']:
rawEvent = evt['Event']
liteEvent['Event']['info'] = rawEvent['info']
liteEvent['Event']['Attribute'] = []
attrs = evt['Attribute']
for attr in attrs:
if 'Internal reference' not in attr['category']:
liteAttr = {}
liteAttr['category'] = attr['category']
liteAttr['type'] = attr['type']
liteAttr['value'] = attr['value']
liteEvent['Event']['Attribute'].append(liteAttr)
attrs = evt['Attribute']
for attr in attrs:
if 'Internal reference' not in attr['category']:
liteAttr = {}
liteAttr['category'] = attr['category']
liteAttr['type'] = attr['type']
liteAttr['value'] = attr['value']
liteEvent['Event']['Attribute'].append(liteAttr)
return {'response': [],
'data': str(base64.b64encode(bytes(
json.dumps(liteEvent, indent=config['indent_json_export']), 'utf-8')), 'utf-8')}
return {'response': [],
'data': str(base64.b64encode(bytes(
json.dumps(liteEvent, indent=config['indent_json_export']), 'utf-8')), 'utf-8')}
def introspection():
modulesetup = {}
try:
responseType
modulesetup['responseType'] = responseType
except NameError:
pass
try:
userConfig
modulesetup['userConfig'] = userConfig
except NameError:
pass
try:
outputFileExtension
modulesetup['outputFileExtension'] = outputFileExtension
except NameError:
pass
try:
inputSource
modulesetup['inputSource'] = inputSource
except NameError:
pass
return modulesetup
modulesetup = {}
try:
responseType
modulesetup['responseType'] = responseType
except NameError:
pass
try:
userConfig
modulesetup['userConfig'] = userConfig
except NameError:
pass
try:
outputFileExtension
modulesetup['outputFileExtension'] = outputFileExtension
except NameError:
pass
try:
inputSource
modulesetup['inputSource'] = inputSource
except NameError:
pass
return modulesetup
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo
moduleinfo['config'] = moduleconfig
return moduleinfo

2
misp_modules/modules/export_mod/nexthinkexport.py
Unescape View File

@ -86,7 +86,7 @@ def handler(q=False):
for event in request["data"]:
for attribute in event["Attribute"]:
if attribute['type'] in types_to_use:
output = output + handlers[attribute['type']](attribute['value'], config['Period']) + '\n'
output = output + handlers[attribute['type']](attribute['value'], config['Period']) + '\n'
r = {"response": [], "data": str(base64.b64encode(bytes(output, 'utf-8')), 'utf-8')}
return r

2
misp_modules/modules/export_mod/osqueryexport.py
Unescape View File

@ -80,7 +80,7 @@ def handler(q=False):
for event in request["data"]:
for attribute in event["Attribute"]:
if attribute['type'] in types_to_use:
output = output + handlers[attribute['type']](attribute['value']) + '\n'
output = output + handlers[attribute['type']](attribute['value']) + '\n'
r = {"response": [], "data": str(base64.b64encode(bytes(output, 'utf-8')), 'utf-8')}
return r

53
misp_modules/modules/export_mod/pdfexport.py
Unescape View File

@ -152,36 +152,37 @@ def handler(q=False):
command_line = 'asciidoctor-pdf -'
args = shlex.split(command_line)
with subprocess.Popen(args, stdout=subprocess.PIPE, stdin=subprocess.PIPE) as process:
cmd_out, cmd_err = process.communicate(input=report.report.encode('utf-8'))
cmd_out, cmd_err = process.communicate(
input=report.report.encode('utf-8'))
return {'response': [], 'data': str(base64.b64encode(cmd_out), 'utf-8')}
def introspection():
modulesetup = {}
try:
responseType
modulesetup['responseType'] = responseType
except NameError:
pass
try:
userConfig
modulesetup['userConfig'] = userConfig
except NameError:
pass
try:
outputFileExtension
modulesetup['outputFileExtension'] = outputFileExtension
except NameError:
pass
try:
inputSource
modulesetup['inputSource'] = inputSource
except NameError:
pass
return modulesetup
modulesetup = {}
try:
responseType
modulesetup['responseType'] = responseType
except NameError:
pass
try:
userConfig
modulesetup['userConfig'] = userConfig
except NameError:
pass
try:
outputFileExtension
modulesetup['outputFileExtension'] = outputFileExtension
except NameError:
pass
try:
inputSource
modulesetup['inputSource'] = inputSource
except NameError:
pass
return modulesetup
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo
moduleinfo['config'] = moduleconfig
return moduleinfo

2
misp_modules/modules/import_mod/openiocimport.py
Unescape View File

@ -63,7 +63,7 @@ def handler(q=False):
"comment": getattr(attrib, 'comment', '')}
# add tag
if q.get('config') and q['config'].get('default tag') is not None:
toAppend["tags"] = q['config']['default tag'].split(",")
toAppend["tags"] = q['config']['default tag'].split(",")
r["results"].append(toAppend)
return r

4
misp_modules/modules/import_mod/threatanalyzer_import.py
Unescape View File

@ -325,7 +325,7 @@ def process_analysis_json(analysis_json):
for stored_created_file in process['stored_files']['stored_created_file']:
stored_created_file['@filename'] = cleanup_filepath(stored_created_file['@filename'])
if stored_created_file['@filename']:
if stored_created_file['@filesize'] is not '0':
if stored_created_file['@filesize'] != '0':
val = '{}|{}'.format(stored_created_file['@filename'], stored_created_file['@md5'])
# print("stored_created_file filename|md5: {}|{} IDS:yes".format(
# stored_created_file['@filename'], # filename
@ -346,7 +346,7 @@ def process_analysis_json(analysis_json):
for stored_modified_file in process['stored_files']['stored_modified_file']:
stored_modified_file['@filename'] = cleanup_filepath(stored_modified_file['@filename'])
if stored_modified_file['@filename']:
if stored_modified_file['@filesize'] is not '0':
if stored_modified_file['@filesize'] != '0':
val = '{}|{}'.format(stored_modified_file['@filename'], stored_modified_file['@md5'])
# print("stored_modified_file MODIFY FILE: {}\t{}".format(
# stored_modified_file['@filename'], # filename

Write Preview
Loading…
Cancel
Save