Cortex
Cortex copied to clipboard
TheHive-Project
Error with reponder Reporter_0_1
Work Environment
| Question | Answer |
|---|---|
| OS version (server) | Ubuntu |
| OS version (client) | Ubuntu |
| Cortex version / git hash | 3.1.3-1 |
| Package Type | From source |
Problem Description
Error with responder Reporter_0_1.
Steps to Reproduce
- step 1 Creating a case in thehive
- step 2 Do "Run responders" in this case and select Reporter_0_1
Complementary information
Hello !
I recently had a problem with the Reporter answering machine. Here is the error displayed:
It was working fine, but I don't know why for a few days now it's showing me this error...
The script :
#!/usr/bin/env python3
# encoding: utf-8
#https://github.com/didix21/mdutils#writing-and-reading-files
import time
import json
import os
#import markdown
from cortexutils.responder import Responder
from mdutils import MdUtils
from thehive4py.api import TheHiveApi
from thehive4py.models import Case, CaseTask, CaseTaskLog, CustomFieldHelper
class Reporter(Responder):
def __init__(self):
Responder.__init__(self)
self.thehive_instance = self.get_param('config.thehive_instance', 'localhost:9000')
self.thehive_api = self.get_param('config.thehive_api', 'YOUR_KEY_HERE')
self.api = TheHiveApi(self.thehive_instance,self.thehive_api)
self.tmpPath = self.get_param('config.tmp_file_location')
def getSummary(self,severity):
# Summary Fields - Severity
if(severity == 1):
severity = "Low"
elif(severity == 2):
severity = "Medium"
elif(severity == 3):
severity = "High"
else:
severity = "unknown"
return severity
def getTLP(self,tlp):
# Summary Fields - TLP
if(tlp == 0):
tlp = ['', '**TLP:WHITE - Disclosure is not limited.**']
elif(tlp == 1):
tlp = ['', '**TLP:GREEN - Limited disclosure, restricted to the community.**']
elif(tlp == 2):
tlp = ['','**TLP:AMBER - Limited disclosure, restricted to participants’ organizations.**']
elif(tlp == 3):
tlp = ['', '**TLP:RED - Not for disclosure, restricted to participants only.**']
else:
tlp = "unknown"
return tlp
def getCaseSummary(self,data):
startDate = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.localtime(data['startDate'] / 1000)) #Convert epoch ms to sec then human readable
severity = self.getSummary(data['severity'])
if(data['tags'].__len__() == 0):
tags = ["No tags found"]
else:
tags = (data['tags'])
caseSummary = [' ',' ','**Severity** ', str(severity), '**Created By** ', str(data['createdBy']), '**Assignee** ', str(data['owner']), '**Tags** ', str(', '.join(tags))]
if data['status'] == 'Resolved':
closeDate = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.localtime(data['endDate'] / 1000)) #Convert epoch ms to sec then human readable
caseSummary.extend(['**Case status:** ', 'Closed', '**Start Date**', startDate, '**Close Date:** ', closeDate, '**Resolution:** ', data['resolutionStatus'], '**Summary:** <br>', data['summary']])
else:
caseSummary.extend(['**Case status:** ', 'Open', '**Start Date**', startDate])
return caseSummary
def getCaseObservables(self,case_observables):
case_observables_sorted = sorted(case_observables, key=lambda k: k['createdAt'])
caseObservables = ['Created At', 'Data Type', 'Data', 'Sighted', 'IOC', 'Tags']
for observable in case_observables_sorted:
createdAt = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.localtime(observable['createdAt'] / 1000)) #Convert epoch ms to sec then human readable
caseObservables.append(createdAt)
caseObservables.append(str(observable['dataType']))
if(observable['dataType'] == 'file'):
caseObservables.append(str(observable['attachment']['name']))
else:
caseObservables.append(str(observable['data'].replace('\n','<br>').replace('.','[.]').replace('http','hxxp')))
caseObservables.append(str(observable['sighted']))
caseObservables.append(str(observable['ioc']))
caseObservables.append(str(', '.join(observable['tags'])))
return caseObservables
def getCaseTasks(self,caseId):
response = self.api.get_case_tasks(caseId)
caseTasks = (json.dumps(response.json(), indent=4, sort_keys=True))
allTaskIds = {}
# Build a list of tasks that we want to get the details for
for task in json.loads(caseTasks):
if (task['title'] == 'Autogenerated Report') or (task['status'] == 'Cancel'):
continue
else:
taskId=task['id']
try:
if (task['description']):
allTaskIds[taskId] = {'taskGroup' : task['group'], 'taskTitle' : task['title'], 'createdAt' : task['createdAt'], 'createdBy' : task['createdBy'], 'owner' : task['owner'], 'status' : task['status'], 'description' : task['description']}
except KeyError:
allTaskIds[taskId] = {'taskGroup' : task['group'], 'taskTitle' : task['title'], 'createdAt' : task['createdAt'], 'createdBy' : task['createdBy'], 'owner' : task['owner'], 'status' : task['status'], 'description' : 'No description specified'}
return allTaskIds
def getCaseTaskLog(self,taskLogId):
response = self.api.get_task_logs(taskLogId)
caseTaskLog = (json.dumps(response.json(), indent=4, sort_keys=True))
return caseTaskLog
def getTlpFooter(self):
tlpFooter = ['Color','When should it be used?','How may it be shared',
'TLP:RED <br> ', "Sources may use TLP\:RED when information cannot be effectively acted upon by additional parties, and could lead to impacts on a party's privacy, reputation, or operations if misused.", "Recipients may not share TLP\:RED information with any parties outside of the specific exchange, meeting, or conversation in which it was originally disclosed. In the context of a meeting, for example, TLP\:RED information is limited to those present at the meeting. In most circumstances, TLP\:RED should be exchanged verbally or in person.",
'TLP:AMBER <br> ', "Sources may use TLP\:AMBER when information requires support to be effectively acted upon, yet carries risks to privacy, reputation, or operations if shared outside of the organizations involved.", "Recipients may only share TLP\:AMBER information with members of their own organization, and with clients or customers who need to know the information to protect themselves or prevent further harm. **Sources are at liberty to specify additional intended limits of the sharing\: these must be adhered to.**",
'TLP:GREEN <br> ' , "Sources may use TLP\:GREEN when information is useful for the awareness of all participating organizations as well as with peers within the broader community or sector.", "Recipients may share TLP\:GREEN information with peers and partner organizations within their sector or community, but not via publicly accessible channels. Information in this category can be circulated widely within a particular community. TLP\:GREEN information may not be released outside of the community.",
'TLP:WHITE <br> ', "Sources may use TLP\:WHITE when information carries minimal or no foreseeable risk of misuse, in accordance with applicable rules and procedures for public release.", "Subject to standard copyright rules, TLP\:WHITE information may be distributed without restriction."]
return tlpFooter
def addTask(self,caseId):
response = self.api.create_case_task(caseId, CaseTask(
title='Autogenerated Report',
startDate=int(time.time())*1000))
if response.status_code == 201:
return(json.dumps(response.json(), indent=4, sort_keys=True))
else:
self.error('ko: {}/{}'.format(response.status_code, response.text))
def addTaskLog(self,taskId,filename):
response = self.api.create_task_log(taskId, CaseTaskLog(
message='Autogenerated report',
file=filename))
if response.status_code == 201:
return(json.dumps(response.json(), indent=4, sort_keys=True))
else:
self.error('ko: {}/{}'.format(response.status_code, response.text))
def run(self):
Responder.run(self)
caseNumber = self.get_param('data.caseId') #Friendly case number
caseId = self.get_param('data.id') #Raw case number
case_observables = self.api.get_case_observables(caseId).json()
title = self.get_param('data.title', None, 'title is missing')
description = self.get_param('data.description', None, 'description is missing')
tags = self.get_param('data.tags')
data = self.get_param('data')
tlp = self.getTLP(data['tlp'])
# Title
#mdFile = MdUtils(file_name=str(caseNumber),title=tlp[0] + ' Case #' + str(caseNumber) + ': ' + title)
mdFile = MdUtils(file_name=str(self.tmpPath) + str(caseNumber),title=tlp[0] + ' Case #' + str(caseNumber) + ': ' + title)
# Case Summary
caseSummary = self.getCaseSummary(data)
mdFile.new_header(level=1, title='Case Summary')
mdFile.new_line(str(tlp[1]))
mdFile.new_table(columns=2, rows=int(caseSummary.__len__()/2), text=caseSummary, text_align='left')
# Case Description
mdFile.new_line('<div style="page-break-after: always;"></div>')
mdFile.new_line(' ')
mdFile.new_header(level=1, title='Case Description')
mdFile.new_line(str(data['description']))
mdFile.new_line(' ')
# Task Log
allTaskIds = self.getCaseTasks(caseId)
allTaskIds_sorted = sorted(allTaskIds.items(), key=lambda x: x[1]['createdAt'])
mdFile.new_header(level=1, title='Task Log Entries')
for task in allTaskIds_sorted:
title=str(task[1]['taskGroup'] + ' \: ' + task[1]['taskTitle'])
createdAt = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.localtime(task[1]['createdAt'] / 1000)) #Convert epoch ms to sec then human readable
mdFile.new_header(level=2, title=title)
mdFile.new_line(str('**Created At:** ') + str(createdAt))
mdFile.new_line(str('**Created By:** ') + str(task[1]['createdBy']))
mdFile.new_line(str('**Assigned To:** ') + str(task[1]['owner']))
mdFile.new_line(str('**Case Status:** ') + str(task[1]['status']))
mdFile.new_line(' ')
mdFile.new_line(str('**Description:** '))
mdFile.new_line(str(task[1]['description']))
mdFile.new_line(' ')
caseTaskLog = self.getCaseTaskLog(task[0])
caseTaskLogEntries = (json.loads(caseTaskLog))
caseTaskLogEntries_sorted = sorted(caseTaskLogEntries, key=lambda k: k['createdAt'])
for caseTaskLogEntry in caseTaskLogEntries_sorted:
createdAt = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.localtime(caseTaskLogEntry['createdAt'] / 1000)) #Convert epoch ms to sec then human readable
mdFile.new_line(str(createdAt) + ' : ' + str(caseTaskLogEntry['message']))
# Case Observables
mdFile.new_header(level=1, title='Case Observables')
caseObservables = self.getCaseObservables(case_observables)
mdFile.new_table(columns=6, rows=int(caseObservables.__len__()/6), text=caseObservables, text_align='left')
# TLP Protocol description
mdFile.new_line('<div style="page-break-after: always;"></div>')
mdFile.new_line(' ')
mdFile.new_header(level=1, title='Traffic Light Protocol (TLP) Definitions and Usage')
tlpFooter = self.getTlpFooter()
mdFile.new_table(columns=3, rows=5, text=tlpFooter, text_align='left')
# Build TOC
mdFile.new_table_of_contents(table_title='Table of Contents', depth=2)
# Compile the report
mdFile.create_md_file()
# Add the report to the case
addTask = json.loads(self.addTask(caseId))
taskId = addTask['_id']
# Add the MD file to the task
addTaskLog = json.loads(self.addTaskLog(taskId,str(self.tmpPath) + str(caseNumber)+'.md'))
# Cleanup the MD file
os.remove(str(self.tmpPath) + str(caseNumber)+'.md')
self.report({'report': 'created'})
if __name__ == '__main__':
Reporter().run()
