Skip to content

Manual analysis/some fixes #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Manual analysis/some fixes #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
198 changes: 171 additions & 27 deletions JsonParser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,13 @@
import json
import xlsxwriter
import r2pipe
import xlrd
from func_finder import _get_start, _get_rst
from collections import OrderedDict

# Predefined functions containing sensor addresses for comparision's
sensors = {
'batt_voltage': ['0x9a56', '0x9f5b', '0xa166', '0xa307', '-0xae2c', '0xd982', '0xe1cd'],
'batt_voltage': ['0x9a56', '0x9f5b', '0xa166', '0xa307', '0xae2c', '0xd982', '0xe1cd'],
'vehicle_speed': ['0x9be8', '0x9dce', '0xa59d', '0xa9a7', '0xafc6', '0xb960'],
'engine_speed': ['0xa59d', '0xa5ec', '0xa9a7', '0xafc6', '0xb5bf', '0xb960'],
'water_temp': ['0x9b46', '0xab56'],
Expand All @@ -21,7 +23,7 @@
hex_margin = 250

class EcuFile:
def __init__(self, file_name, functions):
def __init__(self, file_name, functions, gt=None):
"""
EcuFile constructor
:param file_name: File name of ECU binary
Expand All @@ -33,12 +35,18 @@ def __init__(self, file_name, functions):
self.file_name = split[len(split) - 1]
name = self.file_name.split('-')
self.name = name[0][4:] + '-' + name[1][2:] + '-' + name[4].split('.')[0]

local_gt = dict(gt)

r2 = r2pipe.open('./bins/' + self.file_name)
r2 = r2pipe.open('./bins/' + self.file_name, ["-2"])
start = _get_start(file_name) # using private functions here, but oh well
rst = _get_rst(r2)
r2.cmd('e asm.arch=m7700')
r2.cmd('e anal.limits=true')
r2.cmd('e anal.from=0x9000')
r2.cmd('e anal.to=0xffff')
r2.cmd('e anal.from={}'.format(start if rst > start else rst))
r2.cmd('e anal.to=0xfffe')
r2.cmd('s {}'.format(rst))
r2.cmd('aaa') # analyze all

for address, hashes in functions.items():
# Clean up hashes
Expand All @@ -49,18 +57,42 @@ def __init__(self, file_name, functions):
if hashes != [''] and int(address, 16) > 36864:
self.functions[address] = hashes

if self.name in local_gt.keys():
for sensor_type, l in dict(local_gt[self.name]).iteritems():
for addr in list(l):
r2.cmd("s {}".format(addr)) # seek to addr from XLSX
r2.cmd("sf.") # seek to beginning of function
gt[self.name][sensor_type][ # change addrs from file to point to start of function
gt[self.name][sensor_type].index(addr)] = r2.cmd("s")
r2.quit()
print('Created ECU file ' + self.file_name)

class Cell:

def __init__(self, func_addr, jaccard, flags=None):
self.func_addr = func_addr
self.jaccard = jaccard
if flags is not None:
self.flags = flags
else:
self.flags = { # more can be added in the future
'Guess_Candidate' : False, # is a guess candidate (mark yellow)
'Max_Row': False, # is highest jaccard value
'Control_Value': False # matches control address
}
self.row = None
self.col = None

class IndexTable:

def __init__(self, ecu_file_1, ecu_file_2):
"""
IndexTable constructor
:param ecu_file_1, ecu_file_2: ECU files used for this table
"""
self.indexes = OrderedDict()
self.tables = OrderedDict()

self.name = ecu_file_1.name + ' ' + ecu_file_2.name
self.test_name = ecu_file_2.file_name

Expand All @@ -69,6 +101,7 @@ def __init__(self, ecu_file_1, ecu_file_2):
self.purple_format = book.add_format({'font_color': 'white', 'bg_color': 'purple'})
self.blue_format = book.add_format({'font_color': 'white', 'bg_color': 'blue'})
self.yellow_format = book.add_format({'font_color': 'black', 'bg_color': 'yellow'})
self.red_format = book.add_format({'font_color': 'white', 'bg_color': 'red'})

print('Created index table ' + self.name)

Expand All @@ -78,20 +111,27 @@ def push_index(self, function_1, function_2, jaccard_index):
:param function_1, function_2: Header addresses
:param jaccard_index: Jaccard Index calculation
"""
self.indexes[function_1, function_2] = jaccard_index
self.indexes[function_1, function_2] = Cell(function_2, jaccard_index)

def _write_format(self, sheet, highest_index):
def _write_format(self, sheet, highest_index, sensor_addr):
"""
Format cells with result data
:param sheet: Excel sheet to write write results
:param highest_index: Highest jaccad index in row
"""
sheet.conditional_format(
highest_index[0], highest_index[1], highest_index[0], highest_index[1],
{'type': 'no_errors', 'format': self.blue_format}
)

def write_results(self, book):
if sensor_addr:
sheet.conditional_format(
highest_index[0], highest_index[1], highest_index[0], highest_index[1],
{'type': 'no_errors', 'format': self.purple_format}
) # match condition - color a match with man analysis purple
else:
sheet.conditional_format(
highest_index[0], highest_index[1], highest_index[0], highest_index[1],
{'type': 'no_errors', 'format': self.blue_format}
) # non-match - color it blue instead

def write_results(self, book, gt):
"""
Writes all results to Excel sheet
:param book: Excel sheet containing result data
Expand All @@ -103,49 +143,121 @@ def write_results(self, book):
sheet.freeze_panes(0, 1)
sheet.set_column(0, 0, 23)

global sensors

row, col = 0, 0
highest_index = [0, 0, 0]
left_margin, right_margin = 0, 0
tmp_key = ''
addr_list = None
sensor_addr = None
man_addrs = list()
max_rows = {}

sensor_name = table.name.split(" ")[1]
if sensor_name in gt.keys(): # pull ID'd addr row of manual analysis data from table
addr_list = gt[sensor_name]

# Write results to cells
for keys, jaccard_index in table.indexes.items():
for keys, cell in table.indexes.items():
# Switch to new row
if keys[0] != tmp_key:
jaccard_index = cell.jaccard
if keys[0] != tmp_key: # process first pos of new row
tmp_key = keys[0]
row = row + 1
col = 1

cell.row = row
cell.col = col
# Side header for each row
sheet.write(row, 0, keys[0], self.header_format)
print('\t Added row {}'.format(keys[0]))

if highest_index != [0, 0, 0]:
self._write_format(sheet, highest_index)
# Pull list of man analysis using keys
if addr_list is not None:
if "batt_voltage" in tmp_key:
man_addrs = addr_list["Battery Voltage"]
elif "vehicle_speed" in tmp_key:
man_addrs = addr_list["Vehicle Speed"]
elif "engine_speed" in tmp_key:
man_addrs = addr_list["Engine Speed"]
elif "water_temp" in tmp_key:
man_addrs = addr_list["Water Temp."]
elif "ignition_timing" in tmp_key:
man_addrs = addr_list["Vehicle Speed"]
elif "airflow" in tmp_key:
man_addrs = addr_list["Airflow Sensor"]
elif "throttle_position" in tmp_key:
man_addrs = addr_list["Throttle Position Sensors"]
elif "knock_correction" in tmp_key:
man_addrs = addr_list["Knock Correction"]

ctrl_addr = keys[0][keys[0].find("(")+1:keys[0].find(")")]
test = sensors[ctrl_addr].index(keys[0][0:6])
sensor_addr = man_addrs[test]

if highest_index != [0, 0, 0]:
max_rows[keys[0]] = cell
highest_index = [0, 0, 0]

# Set address range margins
hex_addr = int(keys[0].split(' ')[0], 16)
left_margin = hex_addr - hex_margin
right_margin = hex_addr + hex_margin
else:

else:# process rest of row

col = col + 1
cell.row = row
cell.col = col

if len(man_addrs) is not 0:
ctrl_addr = keys[0][keys[0].find("(")+1:keys[0].find(")")]
test = sensors[ctrl_addr].index(keys[0][0:6])
sensor_addr = man_addrs[test]

# Check if encountered higher Jaccard index
if jaccard_index > highest_index[2]:
highest_index = [row, col, jaccard_index]
max_rows[keys[0]] = cell

# Highlight address margins
# read the sample graph and write everything to spreadsheet
if (sensor_addr is not None) and (sensor_addr.rstrip() in keys[1]):
cell.flags["Control_Value"] = True

# Highlight "estimate" address margins
if int(keys[1], 16) >= left_margin and int(keys[1], 16) <= right_margin:
sheet.write(row, col, round(jaccard_index, 2), self.yellow_format)
else:
sheet.write(row, col, round(jaccard_index, 2))
cell.flags["Guess_Candidate"] = True

# write col headers (addresses)
sheet.write(0, col, keys[1], self.header_format)

self._write_format(sheet, highest_index)

# flag all of the "max" candidate items
for _, cell in max_rows.iteritems():
cell.flags["Max_Row"] = True

# write all results to spreadsheet
for _, cell in table.indexes.items():
self._write_cells(sheet, cell, cell.row, cell.col, keys, round(cell.jaccard , 2))

# Read the cell object, process the cells, write results to the sheet
# Object flags determine what color to add
# Add a new flag to the "flags" dict in the "cell" object if we need new
# colors
def _write_cells(self, sheet, cell, row, col, keys, jaccard_index):
# write header for row

# specify format for writing cell
write_format = None
if cell.flags["Max_Row"]:
write_format = self.blue_format
if cell.flags["Control_Value"]:
write_format = self.purple_format
elif cell.flags["Control_Value"]:
write_format = self.red_format
elif cell.flags["Guess_Candidate"]:
write_format = self.yellow_format

sheet.write(row, col, jaccard_index, write_format)

def _jaccard_index(list_1, list_2):
"""
Expand Down Expand Up @@ -190,21 +302,52 @@ def _create_tables(control_file, ecu_files):

return tables

# Process the ground truth/manual analysis data by loading it in
# This data is used to verify all of our nodes that are found using our
# automatic method
# Returns: Nested dictionary laid out as follows
# Binary name: Sensor name: [All sensor functions manually found]
# for each binary, sensor, and values in the provided file
def _process_gt(sheet):
ret = dict()
code_blocks = xlrd.open_workbook(sheet)
gt = code_blocks.sheet_by_index(0) # pull man analysis ground truth
i = 1
bin_nms = dict()

while i < 21: # we have 10 valid/working samples currently
bin_nms[i] = gt.cell(0, i).value
i += 2

sensor_nms = gt.col_values(0, 2)

for col, bin_nm in bin_nms.iteritems():
sensor_addr = gt.col_values(col, 2)
code_block_addr = gt.col_values(col + 1, 2)

tmp = {}
for _ in range(0, 8):
tmp[sensor_nms[_]] = code_block_addr[_].splitlines()

ret[bin_nm] = tmp
return ret

if __name__ == '__main__':
ecu_files = []
control_file = None
gt = None

if len(sys.argv) != 3:
if len(sys.argv) < 3:
print('Run \'python JsonParser.py file.json output.xlsx')
exit()

gt = _process_gt("code_blocks.xlsx")
# Open & parse JSON dump
with open(sys.argv[1]) as file:
json_data = json.load(file, object_pairs_hook=OrderedDict)

for file_name in json_data:
ecu_file = EcuFile(file_name, json_data[file_name])
ecu_file = EcuFile(file_name, json_data[file_name], gt)

# Pick out control file
if ecu_file.name == '27-93-EG33':
Expand All @@ -218,7 +361,8 @@ def _create_tables(control_file, ecu_files):

# Write all table data to sheets
for table in tables:
table.write_results(book)
table.write_results(book, gt)
# write GT data to sheets

book.close()

Expand Down
Loading