Merge pull request #411 from cforger/master

Fix sample_network_auditing.py so it works with Python3
This commit is contained in:
Nguyen Anh Quynh 2016-02-01 10:42:30 +08:00
commit dad2baa92f

View file

@ -7,7 +7,6 @@ from unicorn import *
from unicorn.x86_const import *
import struct
import uuid
import random
SIZE_REG = 4
SOCKETCALL_MAX_ARGS = 3
@ -51,6 +50,7 @@ X86_REVERSE_TCP_2 = b"\x31\xc0\x31\xdb\x31\xc9\x31\xd2\xb0\x66\xb3\x01\x51\x6a\x
# memory address where emulation starts
ADDRESS = 0x1000000
# supported classes
class IdGenerator:
def __init__(self):
@ -63,6 +63,7 @@ class IdGenerator:
return next_id
class LogChain:
def __init__(self):
self.__chains = {}
@ -72,11 +73,11 @@ class LogChain:
self.__chains = {}
self.__linking_fds = {}
def create_chain(self, id):
if not self.__chains.has_key(id):
self.__chains[id] = []
def create_chain(self, my_id):
if not my_id in self.__chains:
self.__chains[my_id] = []
else:
print("LogChain: id %d existed" % id)
print("LogChain: id %d existed" % my_id)
def add_log(self, id, msg):
fd = self.get_original_fd(id)
@ -87,16 +88,16 @@ class LogChain:
print("LogChain: id %d doesn't exist" % id)
def link_fd(self, from_fd, to_fd):
if not self.__linking_fds.has_key(to_fd):
if not to_fd in self.__linking_fds:
self.__linking_fds[to_fd] = []
self.__linking_fds[to_fd].append(from_fd)
def get_original_fd(self, fd):
if self.__chains.has_key(fd):
if fd in self.__chains:
return fd
for orig_fd, links in self.__linking_fds.iteritems():
for orig_fd, links in self.__linking_fds.items():
if fd in links:
return orig_fd
@ -108,10 +109,11 @@ class LogChain:
| START REPORT |
----------------
""")
for id, logs in self.__chains.iteritems():
print("---- START FD(%d) ----" % id)
for my_id, logs in self.__chains.items():
print("---- START FD(%d) ----" % my_id)
print("\n".join(logs))
print("---- END FD(%d) ----" % id)
print("---- END FD(%d) ----" % my_id)
print("""
--------------
@ -119,10 +121,9 @@ class LogChain:
--------------
""")
# end supported classes
id_gen = IdGenerator()
fd_chains = LogChain()
# utilities
def bin_to_ipv4(ip):
@ -132,6 +133,7 @@ def bin_to_ipv4(ip):
(ip & 0xff00) >> 8,
(ip & 0xff))
def read_string(uc, addr):
ret = ""
@ -145,6 +147,7 @@ def read_string(uc, addr):
return ret
def parse_sock_address(sock_addr):
sin_family, = struct.unpack("<h", sock_addr[:2])
@ -154,8 +157,11 @@ def parse_sock_address(sock_addr):
elif sin_family == 6: # AF_INET6
return ""
def print_sockcall(msg):
print(">>> SOCKCALL %s" % msg)
# end utilities
# callback for tracing instructions
@ -168,8 +174,11 @@ def hook_code(uc, address, size, user_data):
print(" %x" % i, end="")
print("")
# callback for tracing Linux interrupt
def hook_intr(uc, intno, user_data):
global id_gen
# only handle Linux syscall
if intno != 0x80:
return
@ -350,8 +359,11 @@ def hook_intr(uc, intno, user_data):
fd_chains.add_log(fd, msg)
print_sockcall(msg)
# Test X86 32 bit
def test_i386(code):
global fd_chains
fd_chains.clean()
print("Emulate i386 code")
try:
@ -384,9 +396,13 @@ def test_i386(code):
fd_chains.print_report()
# Globals
fd_chains = LogChain()
id_gen = IdGenerator()
if __name__ == '__main__':
test_i386(X86_SEND_ETCPASSWD)
test_i386(X86_BIND_TCP)
test_i386(X86_REVERSE_TCP)
test_i386(X86_REVERSE_TCP_2)