coercer.core.Reporter
1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# File name : Reporter.py 4# Author : Podalirius (@podalirius_) 5# Date created : 17 Jul 2022 6 7import sqlite3 8import os 9import json 10import xlsxwriter 11import sys 12from coercer.structures.ReportingLevel import ReportingLevel 13from coercer.structures.TestResult import TestResult 14 15 16CURSOR_UP_ONE = '\x1b[1A' 17ERASE_LINE = '\x1b[2K' 18 19 20class Reporter(object): 21 """ 22 Documentation for class Reporter 23 """ 24 25 def __init__(self, options, verbose=False): 26 super(Reporter, self).__init__() 27 self.options = options 28 self.verbose = verbose 29 self.test_results = {} 30 31 def print_testing(self, msprotocol_rpc_instance): 32 print(" [>] (\x1b[93m%s\x1b[0m) %s " % ("-testing-", str(msprotocol_rpc_instance))) 33 sys.stdout.flush() 34 35 def print_info(self, message): 36 print("\x1b[1m[\x1b[92minfo\x1b[0m\x1b[1m]\x1b[0m %s" % message) 37 sys.stdout.flush() 38 39 def print_verbose(self, message): 40 print("[debug]",message) 41 42 def report_test_result(self, uuid, version, namedpipe, msprotocol_rpc_instance, result, exploitpath): 43 function_name = msprotocol_rpc_instance.function["name"] 44 if uuid not in self.test_results.keys(): 45 self.test_results[uuid] = {} 46 if version not in self.test_results[uuid].keys(): 47 self.test_results[uuid][version] = {} 48 if function_name not in self.test_results[uuid][version].keys(): 49 self.test_results[uuid][version][function_name] = {} 50 if namedpipe not in self.test_results[uuid][version][function_name].keys(): 51 self.test_results[uuid][version][function_name][namedpipe] = [] 52 53 # Save result to database 54 self.test_results[uuid][version][function_name][namedpipe].append({ 55 "function": msprotocol_rpc_instance.function, 56 "protocol": msprotocol_rpc_instance.protocol, 57 "testresult": result.name, 58 "exploitpath": exploitpath 59 }) 60 61 sys.stdout.write(CURSOR_UP_ONE) 62 sys.stdout.write(ERASE_LINE) 63 if self.options.mode in ["scan", "fuzz"]: 64 if result == TestResult.SMB_AUTH_RECEIVED: 65 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("SMB Auth", str(msprotocol_rpc_instance))) 66 sys.stdout.flush() 67 elif result == TestResult.HTTP_AUTH_RECEIVED: 68 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("HTTP Auth", str(msprotocol_rpc_instance))) 69 sys.stdout.flush() 70 elif result == TestResult.NCA_S_UNK_IF: 71 print(" [\x1b[1;95m-\x1b[0m] (\x1b[1;95m%s\x1b[0m) %s " % ("-No Func-", str(msprotocol_rpc_instance))) 72 sys.stdout.flush() 73 else: 74 if self.verbose: 75 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 76 sys.stdout.flush() 77 elif self.options.mode in ["coerce"]: 78 if result == TestResult.ERROR_BAD_NETPATH: 79 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("ERROR_BAD_NETPATH", str(msprotocol_rpc_instance))) 80 sys.stdout.flush() 81 else: 82 if self.verbose: 83 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 84 sys.stdout.flush() 85 86 def exportXLSX(self, filename): 87 basepath = os.path.dirname(filename) 88 filename = os.path.basename(filename) 89 if basepath not in [".", ""]: 90 if not os.path.exists(basepath): 91 os.makedirs(basepath) 92 path_to_file = basepath + os.path.sep + filename 93 else: 94 path_to_file = filename 95 # export 96 97 workbook = xlsxwriter.Workbook(path_to_file) 98 worksheet = workbook.add_worksheet() 99 100 header_format = workbook.add_format({'bold': 1}) 101 header_fields = ["Interface UUID", "Interface version", "SMB named pipe", "Protocol long name", "Protocol short name", "RPC function name", "Operation number", "Result", "Working path"] 102 for k in range(len(header_fields)): 103 worksheet.set_column(k, k + 1, len(header_fields[k]) + 3) 104 worksheet.set_row(0, 60, header_format) 105 worksheet.write_row(0, 0, header_fields) 106 107 row_id = 1 108 for uuid in self.test_results.keys(): 109 for version in self.test_results[uuid].keys(): 110 for function_name in self.test_results[uuid][version].keys(): 111 for namedpipe in self.test_results[uuid][version][function_name].keys(): 112 for test_result in self.test_results[uuid][version][function_name][namedpipe]: 113 data = [uuid, version, namedpipe, test_result["protocol"]["longname"], test_result["protocol"]["shortname"], test_result["function"]["name"], test_result["function"]["opnum"], test_result["testresult"], test_result["exploitpath"]] 114 worksheet.write_row(row_id, 0, data) 115 row_id += 1 116 worksheet.autofilter(0, 0, row_id, len(header_fields) - 1) 117 workbook.close() 118 self.print_info("Results exported to XLSX in '%s'" % path_to_file) 119 120 def exportJSON(self, filename): 121 basepath = os.path.dirname(filename) 122 filename = os.path.basename(filename) 123 if basepath not in [".", ""]: 124 if not os.path.exists(basepath): 125 os.makedirs(basepath) 126 path_to_file = basepath + os.path.sep + filename 127 else: 128 path_to_file = filename 129 # export 130 f = open(path_to_file, "w") 131 f.write(json.dumps(self.test_results, indent=4)) 132 f.close() 133 self.print_info("Results exported to JSON in '%s'" % path_to_file) 134 135 def exportSQLITE(self, target, filename): 136 basepath = os.path.dirname(filename) 137 filename = os.path.basename(filename) 138 if basepath not in [".", ""]: 139 if not os.path.exists(basepath): 140 os.makedirs(basepath) 141 path_to_file = basepath + os.path.sep + filename 142 else: 143 path_to_file = filename 144 # Exporting results 145 # Connecting to sqlite 146 conn = sqlite3.connect(path_to_file) 147 # Creating a cursor object using the cursor() method 148 cursor = conn.cursor() 149 cursor.execute("CREATE TABLE IF NOT EXISTS results(target VARCHAR(255), uuid VARCHAR(255), version VARCHAR(255), named_pipe VARCHAR(255), protocol_shortname VARCHAR(255), protocol_longname VARCHAR(512), function_name VARCHAR(255), result VARCHAR(255), path VARCHAR(512));") 150 for uuid in self.test_results.keys(): 151 for version in self.test_results[uuid].keys(): 152 for function_name in self.test_results[uuid][version].keys(): 153 for named_pipe in self.test_results[uuid][version][function_name].keys(): 154 for test_result in self.test_results[uuid][version][function_name][named_pipe]: 155 cursor.execute("INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( 156 target, 157 uuid, 158 version, 159 named_pipe, 160 test_result["protocol"]["shortname"], 161 test_result["protocol"]["longname"], 162 function_name, 163 test_result["testresult"], 164 str(bytes(test_result["exploitpath"], 'utf-8'))[2:-1].replace('\\\\', '\\') 165 ) 166 ) 167 # Commit your changes in the database 168 conn.commit() 169 # Closing the connection 170 conn.close() 171 self.print_info("Results exported to SQLITE3 db in '%s'" % path_to_file)
class
Reporter:
21class Reporter(object): 22 """ 23 Documentation for class Reporter 24 """ 25 26 def __init__(self, options, verbose=False): 27 super(Reporter, self).__init__() 28 self.options = options 29 self.verbose = verbose 30 self.test_results = {} 31 32 def print_testing(self, msprotocol_rpc_instance): 33 print(" [>] (\x1b[93m%s\x1b[0m) %s " % ("-testing-", str(msprotocol_rpc_instance))) 34 sys.stdout.flush() 35 36 def print_info(self, message): 37 print("\x1b[1m[\x1b[92minfo\x1b[0m\x1b[1m]\x1b[0m %s" % message) 38 sys.stdout.flush() 39 40 def print_verbose(self, message): 41 print("[debug]",message) 42 43 def report_test_result(self, uuid, version, namedpipe, msprotocol_rpc_instance, result, exploitpath): 44 function_name = msprotocol_rpc_instance.function["name"] 45 if uuid not in self.test_results.keys(): 46 self.test_results[uuid] = {} 47 if version not in self.test_results[uuid].keys(): 48 self.test_results[uuid][version] = {} 49 if function_name not in self.test_results[uuid][version].keys(): 50 self.test_results[uuid][version][function_name] = {} 51 if namedpipe not in self.test_results[uuid][version][function_name].keys(): 52 self.test_results[uuid][version][function_name][namedpipe] = [] 53 54 # Save result to database 55 self.test_results[uuid][version][function_name][namedpipe].append({ 56 "function": msprotocol_rpc_instance.function, 57 "protocol": msprotocol_rpc_instance.protocol, 58 "testresult": result.name, 59 "exploitpath": exploitpath 60 }) 61 62 sys.stdout.write(CURSOR_UP_ONE) 63 sys.stdout.write(ERASE_LINE) 64 if self.options.mode in ["scan", "fuzz"]: 65 if result == TestResult.SMB_AUTH_RECEIVED: 66 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("SMB Auth", str(msprotocol_rpc_instance))) 67 sys.stdout.flush() 68 elif result == TestResult.HTTP_AUTH_RECEIVED: 69 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("HTTP Auth", str(msprotocol_rpc_instance))) 70 sys.stdout.flush() 71 elif result == TestResult.NCA_S_UNK_IF: 72 print(" [\x1b[1;95m-\x1b[0m] (\x1b[1;95m%s\x1b[0m) %s " % ("-No Func-", str(msprotocol_rpc_instance))) 73 sys.stdout.flush() 74 else: 75 if self.verbose: 76 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 77 sys.stdout.flush() 78 elif self.options.mode in ["coerce"]: 79 if result == TestResult.ERROR_BAD_NETPATH: 80 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("ERROR_BAD_NETPATH", str(msprotocol_rpc_instance))) 81 sys.stdout.flush() 82 else: 83 if self.verbose: 84 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 85 sys.stdout.flush() 86 87 def exportXLSX(self, filename): 88 basepath = os.path.dirname(filename) 89 filename = os.path.basename(filename) 90 if basepath not in [".", ""]: 91 if not os.path.exists(basepath): 92 os.makedirs(basepath) 93 path_to_file = basepath + os.path.sep + filename 94 else: 95 path_to_file = filename 96 # export 97 98 workbook = xlsxwriter.Workbook(path_to_file) 99 worksheet = workbook.add_worksheet() 100 101 header_format = workbook.add_format({'bold': 1}) 102 header_fields = ["Interface UUID", "Interface version", "SMB named pipe", "Protocol long name", "Protocol short name", "RPC function name", "Operation number", "Result", "Working path"] 103 for k in range(len(header_fields)): 104 worksheet.set_column(k, k + 1, len(header_fields[k]) + 3) 105 worksheet.set_row(0, 60, header_format) 106 worksheet.write_row(0, 0, header_fields) 107 108 row_id = 1 109 for uuid in self.test_results.keys(): 110 for version in self.test_results[uuid].keys(): 111 for function_name in self.test_results[uuid][version].keys(): 112 for namedpipe in self.test_results[uuid][version][function_name].keys(): 113 for test_result in self.test_results[uuid][version][function_name][namedpipe]: 114 data = [uuid, version, namedpipe, test_result["protocol"]["longname"], test_result["protocol"]["shortname"], test_result["function"]["name"], test_result["function"]["opnum"], test_result["testresult"], test_result["exploitpath"]] 115 worksheet.write_row(row_id, 0, data) 116 row_id += 1 117 worksheet.autofilter(0, 0, row_id, len(header_fields) - 1) 118 workbook.close() 119 self.print_info("Results exported to XLSX in '%s'" % path_to_file) 120 121 def exportJSON(self, filename): 122 basepath = os.path.dirname(filename) 123 filename = os.path.basename(filename) 124 if basepath not in [".", ""]: 125 if not os.path.exists(basepath): 126 os.makedirs(basepath) 127 path_to_file = basepath + os.path.sep + filename 128 else: 129 path_to_file = filename 130 # export 131 f = open(path_to_file, "w") 132 f.write(json.dumps(self.test_results, indent=4)) 133 f.close() 134 self.print_info("Results exported to JSON in '%s'" % path_to_file) 135 136 def exportSQLITE(self, target, filename): 137 basepath = os.path.dirname(filename) 138 filename = os.path.basename(filename) 139 if basepath not in [".", ""]: 140 if not os.path.exists(basepath): 141 os.makedirs(basepath) 142 path_to_file = basepath + os.path.sep + filename 143 else: 144 path_to_file = filename 145 # Exporting results 146 # Connecting to sqlite 147 conn = sqlite3.connect(path_to_file) 148 # Creating a cursor object using the cursor() method 149 cursor = conn.cursor() 150 cursor.execute("CREATE TABLE IF NOT EXISTS results(target VARCHAR(255), uuid VARCHAR(255), version VARCHAR(255), named_pipe VARCHAR(255), protocol_shortname VARCHAR(255), protocol_longname VARCHAR(512), function_name VARCHAR(255), result VARCHAR(255), path VARCHAR(512));") 151 for uuid in self.test_results.keys(): 152 for version in self.test_results[uuid].keys(): 153 for function_name in self.test_results[uuid][version].keys(): 154 for named_pipe in self.test_results[uuid][version][function_name].keys(): 155 for test_result in self.test_results[uuid][version][function_name][named_pipe]: 156 cursor.execute("INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( 157 target, 158 uuid, 159 version, 160 named_pipe, 161 test_result["protocol"]["shortname"], 162 test_result["protocol"]["longname"], 163 function_name, 164 test_result["testresult"], 165 str(bytes(test_result["exploitpath"], 'utf-8'))[2:-1].replace('\\\\', '\\') 166 ) 167 ) 168 # Commit your changes in the database 169 conn.commit() 170 # Closing the connection 171 conn.close() 172 self.print_info("Results exported to SQLITE3 db in '%s'" % path_to_file)
Documentation for class Reporter
def
report_test_result( self, uuid, version, namedpipe, msprotocol_rpc_instance, result, exploitpath):
43 def report_test_result(self, uuid, version, namedpipe, msprotocol_rpc_instance, result, exploitpath): 44 function_name = msprotocol_rpc_instance.function["name"] 45 if uuid not in self.test_results.keys(): 46 self.test_results[uuid] = {} 47 if version not in self.test_results[uuid].keys(): 48 self.test_results[uuid][version] = {} 49 if function_name not in self.test_results[uuid][version].keys(): 50 self.test_results[uuid][version][function_name] = {} 51 if namedpipe not in self.test_results[uuid][version][function_name].keys(): 52 self.test_results[uuid][version][function_name][namedpipe] = [] 53 54 # Save result to database 55 self.test_results[uuid][version][function_name][namedpipe].append({ 56 "function": msprotocol_rpc_instance.function, 57 "protocol": msprotocol_rpc_instance.protocol, 58 "testresult": result.name, 59 "exploitpath": exploitpath 60 }) 61 62 sys.stdout.write(CURSOR_UP_ONE) 63 sys.stdout.write(ERASE_LINE) 64 if self.options.mode in ["scan", "fuzz"]: 65 if result == TestResult.SMB_AUTH_RECEIVED: 66 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("SMB Auth", str(msprotocol_rpc_instance))) 67 sys.stdout.flush() 68 elif result == TestResult.HTTP_AUTH_RECEIVED: 69 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("HTTP Auth", str(msprotocol_rpc_instance))) 70 sys.stdout.flush() 71 elif result == TestResult.NCA_S_UNK_IF: 72 print(" [\x1b[1;95m-\x1b[0m] (\x1b[1;95m%s\x1b[0m) %s " % ("-No Func-", str(msprotocol_rpc_instance))) 73 sys.stdout.flush() 74 else: 75 if self.verbose: 76 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 77 sys.stdout.flush() 78 elif self.options.mode in ["coerce"]: 79 if result == TestResult.ERROR_BAD_NETPATH: 80 print(" [\x1b[1;92m+\x1b[0m] (\x1b[1;92m%s\x1b[0m) %s " % ("ERROR_BAD_NETPATH", str(msprotocol_rpc_instance))) 81 sys.stdout.flush() 82 else: 83 if self.verbose: 84 print(" [\x1b[1;91m!\x1b[0m] (\x1b[1;91m%s\x1b[0m) %s " % (result.name, str(msprotocol_rpc_instance))) 85 sys.stdout.flush()
def
exportXLSX(self, filename):
87 def exportXLSX(self, filename): 88 basepath = os.path.dirname(filename) 89 filename = os.path.basename(filename) 90 if basepath not in [".", ""]: 91 if not os.path.exists(basepath): 92 os.makedirs(basepath) 93 path_to_file = basepath + os.path.sep + filename 94 else: 95 path_to_file = filename 96 # export 97 98 workbook = xlsxwriter.Workbook(path_to_file) 99 worksheet = workbook.add_worksheet() 100 101 header_format = workbook.add_format({'bold': 1}) 102 header_fields = ["Interface UUID", "Interface version", "SMB named pipe", "Protocol long name", "Protocol short name", "RPC function name", "Operation number", "Result", "Working path"] 103 for k in range(len(header_fields)): 104 worksheet.set_column(k, k + 1, len(header_fields[k]) + 3) 105 worksheet.set_row(0, 60, header_format) 106 worksheet.write_row(0, 0, header_fields) 107 108 row_id = 1 109 for uuid in self.test_results.keys(): 110 for version in self.test_results[uuid].keys(): 111 for function_name in self.test_results[uuid][version].keys(): 112 for namedpipe in self.test_results[uuid][version][function_name].keys(): 113 for test_result in self.test_results[uuid][version][function_name][namedpipe]: 114 data = [uuid, version, namedpipe, test_result["protocol"]["longname"], test_result["protocol"]["shortname"], test_result["function"]["name"], test_result["function"]["opnum"], test_result["testresult"], test_result["exploitpath"]] 115 worksheet.write_row(row_id, 0, data) 116 row_id += 1 117 worksheet.autofilter(0, 0, row_id, len(header_fields) - 1) 118 workbook.close() 119 self.print_info("Results exported to XLSX in '%s'" % path_to_file)
def
exportJSON(self, filename):
121 def exportJSON(self, filename): 122 basepath = os.path.dirname(filename) 123 filename = os.path.basename(filename) 124 if basepath not in [".", ""]: 125 if not os.path.exists(basepath): 126 os.makedirs(basepath) 127 path_to_file = basepath + os.path.sep + filename 128 else: 129 path_to_file = filename 130 # export 131 f = open(path_to_file, "w") 132 f.write(json.dumps(self.test_results, indent=4)) 133 f.close() 134 self.print_info("Results exported to JSON in '%s'" % path_to_file)
def
exportSQLITE(self, target, filename):
136 def exportSQLITE(self, target, filename): 137 basepath = os.path.dirname(filename) 138 filename = os.path.basename(filename) 139 if basepath not in [".", ""]: 140 if not os.path.exists(basepath): 141 os.makedirs(basepath) 142 path_to_file = basepath + os.path.sep + filename 143 else: 144 path_to_file = filename 145 # Exporting results 146 # Connecting to sqlite 147 conn = sqlite3.connect(path_to_file) 148 # Creating a cursor object using the cursor() method 149 cursor = conn.cursor() 150 cursor.execute("CREATE TABLE IF NOT EXISTS results(target VARCHAR(255), uuid VARCHAR(255), version VARCHAR(255), named_pipe VARCHAR(255), protocol_shortname VARCHAR(255), protocol_longname VARCHAR(512), function_name VARCHAR(255), result VARCHAR(255), path VARCHAR(512));") 151 for uuid in self.test_results.keys(): 152 for version in self.test_results[uuid].keys(): 153 for function_name in self.test_results[uuid][version].keys(): 154 for named_pipe in self.test_results[uuid][version][function_name].keys(): 155 for test_result in self.test_results[uuid][version][function_name][named_pipe]: 156 cursor.execute("INSERT INTO results VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ( 157 target, 158 uuid, 159 version, 160 named_pipe, 161 test_result["protocol"]["shortname"], 162 test_result["protocol"]["longname"], 163 function_name, 164 test_result["testresult"], 165 str(bytes(test_result["exploitpath"], 'utf-8'))[2:-1].replace('\\\\', '\\') 166 ) 167 ) 168 # Commit your changes in the database 169 conn.commit() 170 # Closing the connection 171 conn.close() 172 self.print_info("Results exported to SQLITE3 db in '%s'" % path_to_file)