Magic-Mirror/androguard
1
0
Fork 0
mirror of https://github.com/androguard/androguard.git synced 2025-02-18 12:47:47 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

improve wm

Browse Source
This commit is contained in:
Anthony Desnos 2010-11-19 11:47:01 +01:00
parent a78d7f9b8f
commit 010812fd28
4 changed files with 267 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
androguard.py
View File

@ -121,7 +121,7 @@ class WMCheck :
for method in andro.get_methods() :
_method, _vm = andro.get_method_descriptor(method.get_class_name(), method.get_name(), method.get_descriptor())
w_cmp = wm.WMCheck( w_orig, _vm, _method, a )
#w_cmp.show()

104
core/analysis/analysis.py
View File

@ -67,7 +67,7 @@ class ToString :
def push(self, name) :
for i in self.__tab :
for j in self.__re_tab[i] :
if j.match(name) :
if j.match(name) != None :
if len(self.__string) > 0 :
if i == 'O' and self.__string[-1] == 'O' :
continue
@ -76,17 +76,56 @@ class ToString :
def get_string(self) :
return self.__string
class BreakBlock(object) :
def __init__(self, _vm) :
self._ins = []
self._vm = _vm
self._ops = []
self._fields = {}
self._methods = {}
def get_ops(self) :
return self._ops
def get_fields(self) :
return self._fields
def get_methods(self) :
return self._methods
def push(self, ins) :
self._ins.append(ins)
def show(self) :
for i in self._ins :
print "\t\t",
i.show(0)
##### DVM ######
DVM_TOSTRING = { "O" : dvm.MATH_DVM_OPCODES,
MATH_DVM_RE = []
for i in dvm.MATH_DVM_OPCODES :
MATH_DVM_RE.append( (re.compile( i ), dvm.MATH_DVM_OPCODES[i]) )
DVM_TOSTRING = { "O" : dvm.MATH_DVM_OPCODES.keys(),
"I" : dvm.INVOKE_DVM_OPCODES,
"G" : dvm.FIELD_READ_DVM_OPCODES,
"P" : dvm.FIELD_WRITE_DVM_OPCODES,
}
class DVMBreakBlock :
class DVMBreakBlock(BreakBlock) :
def __init__(self, _vm) :
self.__ins = []
self.__vm = _vm
super(DVMBreakBlock, self).__init__(_vm)
def analyze(self) :
for i in self._ins :
for mre in MATH_DVM_RE :
if mre[0].match( i.get_name() ) :
self._ops.append( mre[1] )
break
##### JVM ######
FIELDS = {
@ -98,56 +137,38 @@ FIELDS = {
METHODS = [ "invokestatic", "invokevirtual", "invokespecial" ]
MATH_JVM_RE = []
for i in jvm.MATH_JVM_OPCODES :
MATH_JVM_RE.append( (re.compile( i ), jvm.MATH_JVM_OPCODES[i]) )
JVM_TOSTRING = { "O" : jvm.MATH_JVM_OPCODES.keys(),
"I" : jvm.INVOKE_JVM_OPCODES,
"G" : jvm.FIELD_READ_JVM_OPCODES,
"P" : jvm.FIELD_WRITE_JVM_OPCODES,
}
class JVMBreakBlock :
class JVMBreakBlock(BreakBlock) :
def __init__(self, _vm) :
self.__ins = []
self.__vm = _vm
self.__ops = []
self.__fields = {}
self.__methods = {}
super(JVMBreakBlock, self).__init__(_vm)
self.__info = {
"F" : [ "get_field_descriptor", self.__fields, ContextField ],
"M" : [ "get_method_descriptor", self.__methods, ContextMethod ],
"F" : [ "get_field_descriptor", self._fields, ContextField ],
"M" : [ "get_method_descriptor", self._methods, ContextMethod ],
}
def get_ops(self) :
return self.__ops
def get_fields(self) :
return self.__fields
def get_methods(self) :
return self.__methods
def push(self, ins) :
self.__ins.append(ins)
def analyze(self) :
ctt = []
MATH_RE = []
for i in jvm.MATH_JVM_OPCODES :
MATH_RE.append( (re.compile( i ), jvm.MATH_JVM_OPCODES[i]) )
for i in self.__ins :
for i in self._ins :
v = self.trans(i)
if v != None :
ctt.append( v )
t = ""
for mre in MATH_RE :
for mre in MATH_JVM_RE :
if mre[0].match( i.get_name() ) :
self.__ops.append( mre[1] )
self._ops.append( mre[1] )
break
# Woot it's a field !
@ -158,7 +179,7 @@ class JVMBreakBlock :
if t != "" :
o = i.get_operands()
desc = getattr(self.__vm, self.__info[t][0])( o[0], o[1], o[2] )
desc = getattr(self._vm, self.__info[t][0])( o[0], o[1], o[2] )
# It's an external
if desc == None :
@ -172,12 +193,12 @@ class JVMBreakBlock :
elif t == "M" :
self.__info[t][1][desc].append( self.__info[t][2]() )
for i in self.__fields :
for k in self.__fields[i] :
for i in self._fields :
for k in self._fields[i] :
k.set_details( ctt )
for i in self.__methods :
for k in self.__methods[i] :
for i in self._methods :
for k in self._methods[i] :
k.set_details( ctt )
def trans(self, i) :
@ -215,11 +236,6 @@ class JVMBreakBlock :
if "getfield" in i.get_name() :
return "F" + i.get_operands()[2]
def show(self) :
for i in self.__ins :
print "\t\t",
i.show(0)
class GVM_BCA :
def __init__(self, _vm, _method) :
self.__vm = _vm

158
core/bytecodes/dvm.py
View File

@ -346,14 +346,18 @@ MATH_DVM_OPCODES = { "add." : '+',
"mul." : '*',
"or." : '|',
"sub." : '-',
"and." : '&',
"xor." : '^',
"shl." : "<<",
"shr." : ">>",
}
INVOKE_DVM_OPCODES = [ "invoke." ]
FIELD_READ_DVM_OPCODES = [ ".get." ]
FIELD_WRITE_DVM_OPCODES = [ ".put." ]
FIELD_READ_DVM_OPCODES = [ ".get" ]
FIELD_WRITE_DVM_OPCODES = [ ".put" ]
BREAK_DVM_OPCODES = [ "invoke.", "move.", ".put.", "if." ]
BREAK_DVM_OPCODES = [ "invoke.", "move.", ".put", "if." ]
def readuleb128(buff) :
@ -1628,42 +1632,57 @@ class EncodedCatchHandlerList :
def get_raw(self) :
return writeuleb128( self.size ) + ''.join(i.get_raw() for i in self.list)
class DalvikCode :
def __init__(self, buff, cm) :
self.__CM = cm
off = buff.get_idx()
while off % 4 != 0 :
off += 1
class DBCSpe :
def __init__(self, x) :
pass
buff.set_idx( off )
def show(self) :
raise("oop")
self.__offset = self.__CM.add_offset( buff.get_idx(), self )
self.__off = buff.get_idx()
class DBC :
def __init__(self, class_manager, op_name, operands, raw_buff) :
self.__CM = class_manager
self.registers_size = SV( '<H', buff.read( 2 ) )
self.ins_size = SV( '<H', buff.read( 2 ) )
self.outs_size = SV( '<H', buff.read( 2 ) )
self.tries_size = SV( '<H', buff.read( 2 ) )
self.debug_info_off = SV( '<L', buff.read( 4 ) )
self.insns_size = SV( '<L', buff.read( 4 ) )
self.__op_name = op_name
self.__operands = operands
self.__raw_buff = raw_buff
self.__insn = buff.read( self.insns_size.get_value() * 2 )
def get_name(self) :
"""Return the name of the bytecode"""
return self.__op_name
def show(self, pos) :
print pos, self.__op_name, ' '.join(self._more_info(n[0], n[1]) for n in self.__operands)
def _more_info(self, c, v) :
if "string" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_string(v))
elif "meth" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_method(v))
elif "field" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_field(v))
elif "type" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_type(v))
return "%s%x" % (c, v)
class DCode :
def __init__(self, class_manager, size, buff) :
self.__CM = class_manager
self.__insn = buff
self.__h_special_bytecodes = {}
self.__bytecodes = []
ushort = calcsize( '<H' )
real_j = 0
j = 0
while j < (self.insns_size.get_value() * ushort) :
while j < (size * ushort) :
# handle special instructions
if real_j in self.__h_special_bytecodes :
special_e = self.__h_special_bytecodes[ real_j ]( self.__insn[j : ] )
self.__bytecodes.append( special_e )
self.__bytecodes.append( DBCSpe( self.__CM, special_e ) )
del self.__h_special_bytecodes[ real_j ]
j += special_e.get_size()
@ -1683,7 +1702,7 @@ class DalvikCode :
if special != None :
self.__h_special_bytecodes[ special[0] + real_j ] = special[1]
self.__bytecodes.append( [ DALVIK_OPCODES[ op_value ][1], repr( self.__insn[j : j + int( DALVIK_OPCODES[ op_value ][0][0] ) * ushort ] ), operands ] )
self.__bytecodes.append( DBC( self.__CM, DALVIK_OPCODES[ op_value ][1], operands, self.__insn[j : j + int( DALVIK_OPCODES[ op_value ][0][0] ) * ushort ] ) )
j += ( int( DALVIK_OPCODES[ op_value ][0][0] ) * ushort)
else :
@ -1691,18 +1710,6 @@ class DalvikCode :
real_j = j / 2
if (self.insns_size.get_value() % 2 == 1) :
self.__padding = SV( '<H', buff.read( 2 ) )
self.__tries = []
self.__handlers = []
if self.tries_size.get_value() > 0 :
for i in range(0, self.tries_size.get_value()) :
try_item = SVs( TRY_ITEM[0], TRY_ITEM[1], buff.read( calcsize(TRY_ITEM[0]) ) )
self.__tries.append( try_item )
self.__handlers.append( EncodedCatchHandlerList( buff ) )
def _analyze_mnemonic(self, buff_operands, mnemonic) :
operands = []
t_ops = mnemonic[3].split(' ')
@ -1780,9 +1787,59 @@ class DalvikCode :
else :
bytecode.Exit( "invalid size [ 0x%x ]" % size )
def get_bc(self) :
def get(self) :
return self.__bytecodes
def get_raw(self) :
return self.__insn
def show(self) :
nb = 0
for i in self.__bytecodes :
print nb,
i.show(nb)
nb += 1
class DalvikCode :
def __init__(self, buff, cm) :
self.__CM = cm
off = buff.get_idx()
while off % 4 != 0 :
off += 1
buff.set_idx( off )
self.__offset = self.__CM.add_offset( buff.get_idx(), self )
self.__off = buff.get_idx()
self.registers_size = SV( '<H', buff.read( 2 ) )
self.ins_size = SV( '<H', buff.read( 2 ) )
self.outs_size = SV( '<H', buff.read( 2 ) )
self.tries_size = SV( '<H', buff.read( 2 ) )
self.debug_info_off = SV( '<L', buff.read( 4 ) )
self.insns_size = SV( '<L', buff.read( 4 ) )
ushort = calcsize( '<H' )
self.__code = DCode( self.__CM, self.insns_size.get_value(), buff.read( self.insns_size.get_value() * ushort ) )
if (self.insns_size.get_value() % 2 == 1) :
self.__padding = SV( '<H', buff.read( 2 ) )
self.__tries = []
self.__handlers = []
if self.tries_size.get_value() > 0 :
for i in range(0, self.tries_size.get_value()) :
try_item = SVs( TRY_ITEM[0], TRY_ITEM[1], buff.read( calcsize(TRY_ITEM[0]) ) )
self.__tries.append( try_item )
self.__handlers.append( EncodedCatchHandlerList( buff ) )
def get_bc(self) :
return self.__code
def get_off(self) :
return self.__off
@ -1801,25 +1858,10 @@ class DalvikCode :
print ""
nb = 0
for i in self.__bytecodes :
if type(i).__name__ == 'list' :
print "\t", nb, i[0], ' '.join(self._more_info(n[0], n[1]) for n in i[-1])
else :
print "\t", nb, i.show()
nb += 1
self.__code.show()
print "*" * 80
def _more_info(self, c, v) :
if "string" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_string(v))
elif "meth" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_method(v))
elif "field" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_field(v))
elif "type" in c :
return "%s%x{%s}" % (c, v, self.__CM.get_type(v))
return "%s%x" % (c, v)
def get_obj(self) :
return [ i for i in self.__handlers ]
@ -1831,7 +1873,7 @@ class DalvikCode :
self.tries_size.get_value_buff() + \
self.debug_info_off.get_value_buff() + \
self.insns_size.get_value_buff() + \
self.__insn
self.__code.get_raw()
if (self.insns_size.get_value() % 2 == 1) :
buff += self.__padding.get_value_buff()
@ -2231,7 +2273,7 @@ class DalvikVMFormat(bytecode._Bytecode) :
if class_name == i.get_name() :
for j in i.get_methods() :
if method_name == j.get_name() and descriptor == j.get_descriptor() :
return i
return j
return None
def get_field_descriptor(self, class_name, field_name, descriptor) :

120
core/wm/wm.py
View File

@ -37,11 +37,11 @@ class WM :
self.__wms.append( (i, wb) )
print list_x
if list_x == [] :
raise("list is empty")
self.__ob = DWBO( hashlib.sha512( method.get_raw() ).hexdigest(), list_x )
self.__ob = None
if len(list_x) > 4 :
self.__ob = DWBO( "TOTO", list_x )#hashlib.sha512( method.get_raw() ).hexdigest(), list_x )
#for i in self.__a.get_bb() :
# print i
# i.show()
@ -59,14 +59,17 @@ class WM :
# print ob.verify_with_X( [ 1, 2, 3, 4, 5, 6 ] )
def save(self) :
buffer = "<method class=\"%s\" name=\"%s\" descriptor=\"%s\">\n" % ( self.__method.get_class_name(), escape( self.__method.get_name() ), self.__method.get_descriptor() )
buffer += "<sss>%s</sss>\n" % ( base64.b64encode( cPickle.dumps( self.__ob.get_y() ) ) )
buffer = ""
if self.__ob != None :
buffer += "<method class=\"%s\" name=\"%s\" descriptor=\"%s\">\n" % ( self.__method.get_class_name(), escape( self.__method.get_name() ), self.__method.get_descriptor() )
buffer += "<threshold>%d</threshold>\n" % ( self.__ob.get_threshold() )
buffer += "<sss>%s</sss>\n" % ( base64.b64encode( cPickle.dumps( self.__ob.get_y() ) ) )
for i in self.__wms :
buffer += "<wm type=\"%d\">%s</wm>\n" % ( i[0], base64.b64encode( cPickle.dumps( i[1].get_export_context() ) ) )
for i in self.__wms :
buffer += "<wm type=\"%d\">%s</wm>\n" % ( i[0], base64.b64encode( cPickle.dumps( i[1].get_export_context() ) ) )
buffer += "</method>\n"
buffer += "</method>\n"
return buffer
@ -78,8 +81,10 @@ class WMMLoad :
self.__wms = []
th = int( item.getElementsByTagName( 'threshold' )[0].firstChild.data )
x = base64.b64decode( item.getElementsByTagName( 'sss' )[0].firstChild.data )
# print cPickle.loads( x )
self.__dwbo = DWBOCheck( cPickle.loads( x ), th )
for s_item in item.getElementsByTagName( 'wm' ) :
@ -99,6 +104,9 @@ class WMMLoad :
def get_name(self) :
return self.__name
def get_dwbo(self) :
return self.__dwbo
class WMLoad :
def __init__(self, document) :
self.__methods = []
@ -111,7 +119,6 @@ class WMLoad :
class WMCheck :
def __init__(self, wm_orig, vm, method, analysis) :
print method.get_name()
for _method in wm_orig.get_methods() :
@ -129,7 +136,13 @@ class WMCheck :
list_x.append( i )
print "\t\t X :", list_x
#print wm_orig.get_dwbo().verify_with_X( list_x )
sols = _method.get_dwbo().verify_with_X( list_x )
print "\t\t SOL :", len(sols)
for i in sols :
if i > 0 :
print misc.long2str(i)
print ""
class Polynomial :
@ -192,7 +205,10 @@ class ShamirSecretScheme :
# print "THRESHOLD %d" % self.__threshold
self.poly = Polynomial(self.__threshold, self.__secret_long, len(self.__secret))
def get_threshold(self) :
return self.__threshold
def split(self) :
points = {}
for i in self.__pieces :
@ -286,6 +302,71 @@ class ShamirSecretScheme :
def get_secret_long(self) :
return self.__secret_long
class NevilleAlgorithm :
def __init__(self, th) :
self.__threshold = th
def interpolate(self, x0, y0, x1, y1, x) :
return (y0*(x-x1) - y1*(x-x0)) / (x0 - x1);
def neville_algorithm(self, xs, ys):
for i in range(1, len(xs)) :
for k in range(0, len(xs) - i) :
ys[k] = self.interpolate(xs[k], ys[k], xs[k+i], ys[k+1], 0)
return ys[0]
def run(self, coord_x, coord_y) :
sols = []
print self.__threshold, len(coord_x)
res = itertools.combinations( coord_x, self.__threshold + 1)
nb = 0
for i in res :
print nb, "/", len(coord_x) * (self.__threshold + 1)
nb += 1
# print "I", i
res2 = itertools.product( i, coord_y )
l = []
for j in res2 :
# print "\t res2 j", j
l.append( j )
print ""
res3 = itertools.combinations( l, self.__threshold + 1 )
for j in res3 :
# print "\t res3 j", j
d = []
oops = False
for v in j :
if v[0] not in d :
d.append(v[0])
else :
oops = True
break
if v[1] not in d :
d.append(v[1])
else :
oops = True
break
if oops == False :
# print oops, j
final_x = []
final_y = []
for v in j :
final_x.append(v[0])
final_y.append(v[1])
# print final_x, final_y
sol = self.neville_algorithm(final_x, final_y)
if sol > 0 and misc.long2str(sol) == "TOTO" :
print "laa"
sols.append( sol )
return sols
class DWBO :
def __init__(self, hash, val) :
self.__hash = hash
@ -294,7 +375,6 @@ class DWBO :
self.__sss = ShamirSecretScheme(self.__hash, self.__val, (len(self.__val) / 2) + 1)
self.__points = self.__sss.split()
def verify_with_X(self, coord_x) :
result, success = self.__sss.join( coord_x, self.__points.values() )
return result, success
@ -308,7 +388,19 @@ class DWBO :
def get_hash(self) :
return self.__hash
def get_threshold(self) :
return self.__sss.get_threshold()
def show(self) :
print self.__hash
for i in self.__points :
print i, self.__points[i]
class DWBOCheck :
def __init__(self, l_y, th) :
self.__l_y = l_y
self.__algo = NevilleAlgorithm( th )
def verify_with_X(self, l_x) :
# print "X :", l_x, "Y :", self.__l_y
return self.__algo.run( l_x, self.__l_y )