TheHive-Project / TheHive-Project/Cortex
Error with reponder Reporter_0_1
Open
Nobody has claimed this yet.
- Dominant language
- Scala
- Stars
- 1.6k
- Forks
- 264
- PR merge metrics
- No merged PRs in 30d
Description
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()
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
No project file or test is named. Start by reproducing the Reporter_0_1 responder failure in Cortex 3.1.3-1 on Ubuntu, using the listed TheHive case workflow, and capture the complete error rather than relying on the screenshot. Confirm whether the fault is in Cortex or the supplied Python responder script before defining what a fix should change.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100