-
Notifications
You must be signed in to change notification settings - Fork 466
/
asmblock.py
1645 lines (1340 loc) · 56.3 KB
/
asmblock.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#-*- coding:utf-8 -*-
from builtins import map
from builtins import range
import logging
import warnings
from collections import namedtuple
from builtins import int as int_types
from future.utils import viewitems, viewvalues
from miasm.expression.expression import ExprId, ExprInt, get_expr_locs
from miasm.expression.expression import LocKey
from miasm.expression.simplifications import expr_simp
from miasm.expression.modint import moduint, modint
from miasm.core.utils import Disasm_Exception, pck
from miasm.core.graph import DiGraph, DiGraphSimplifier, MatchGraphJoker
from miasm.core.interval import interval
from miasm.core.locationdb import LocationDB
log_asmblock = logging.getLogger("asmblock")
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter("[%(levelname)-8s]: %(message)s"))
log_asmblock.addHandler(console_handler)
log_asmblock.setLevel(logging.WARNING)
def is_int(a):
return isinstance(a, (modint, moduint, int_types))
class AsmRaw(object):
def __init__(self, raw=b""):
self.raw = raw
def __str__(self):
return repr(self.raw)
def to_string(self, loc_db):
return str(self)
class asm_raw(AsmRaw):
def __init__(self, raw=b""):
warnings.warn('DEPRECATION WARNING: use "AsmRaw" instead of "asm_raw"')
super(asm_label, self).__init__(raw)
class AsmConstraint(object):
c_to = "c_to"
c_next = "c_next"
def __init__(self, loc_key, c_t=c_to):
# Sanity check
assert isinstance(loc_key, LocKey)
self.loc_key = loc_key
self.c_t = c_t
def get_label(self):
warnings.warn('DEPRECATION WARNING: use ".loc_key" instead of ".label"')
return self.loc_key
def set_label(self, loc_key):
warnings.warn('DEPRECATION WARNING: use ".loc_key" instead of ".label"')
self.loc_key = loc_key
label = property(get_label, set_label)
def to_string(self, loc_db=None):
if loc_db is None:
return "%s:%s" % (self.c_t, self.loc_key)
else:
return "%s:%s" % (
self.c_t,
loc_db.pretty_str(self.loc_key)
)
def __str__(self):
return self.to_string()
class asm_constraint(AsmConstraint):
def __init__(self, loc_key, c_t=AsmConstraint.c_to):
warnings.warn('DEPRECATION WARNING: use "AsmConstraint" instead of "asm_constraint"')
super(asm_constraint, self).__init__(loc_key, c_t)
class AsmConstraintNext(AsmConstraint):
def __init__(self, loc_key):
super(AsmConstraintNext, self).__init__(
loc_key,
c_t=AsmConstraint.c_next
)
class asm_constraint_next(AsmConstraint):
def __init__(self, loc_key):
warnings.warn('DEPRECATION WARNING: use "AsmConstraintNext" instead of "asm_constraint_next"')
super(asm_constraint_next, self).__init__(loc_key)
class AsmConstraintTo(AsmConstraint):
def __init__(self, loc_key):
super(AsmConstraintTo, self).__init__(
loc_key,
c_t=AsmConstraint.c_to
)
class asm_constraint_to(AsmConstraint):
def __init__(self, loc_key):
warnings.warn('DEPRECATION WARNING: use "AsmConstraintTo" instead of "asm_constraint_to"')
super(asm_constraint_to, self).__init__(loc_key)
class AsmBlock(object):
def __init__(self, loc_key, alignment=1):
assert isinstance(loc_key, LocKey)
self.bto = set()
self.lines = []
self._loc_key = loc_key
self.alignment = alignment
def get_label(self):
warnings.warn('DEPRECATION WARNING: use ".loc_key" instead of ".label"')
return self.loc_key
loc_key = property(lambda self:self._loc_key)
label = property(get_label)
def to_string(self, loc_db=None):
out = []
if loc_db is None:
out.append(str(self.loc_key))
else:
out.append(loc_db.pretty_str(self.loc_key))
for instr in self.lines:
out.append(instr.to_string(loc_db))
if self.bto:
lbls = ["->"]
for dst in self.bto:
if dst is None:
lbls.append("Unknown? ")
else:
lbls.append(dst.to_string(loc_db) + " ")
lbls = '\t'.join(sorted(lbls))
out.append(lbls)
return '\n'.join(out)
def __str__(self):
return self.to_string()
def addline(self, l):
self.lines.append(l)
def addto(self, c):
assert isinstance(self.bto, set)
self.bto.add(c)
def split(self, loc_db, offset):
loc_key = loc_db.get_or_create_offset_location(offset)
log_asmblock.debug('split at %x', offset)
offsets = [x.offset for x in self.lines]
offset = loc_db.get_location_offset(loc_key)
if offset not in offsets:
log_asmblock.warning(
'cannot split block at %X ' % offset +
'middle instruction? default middle')
offsets.sort()
return None
new_block = AsmBlock(loc_key)
i = offsets.index(offset)
self.lines, new_block.lines = self.lines[:i], self.lines[i:]
flow_mod_instr = self.get_flow_instr()
log_asmblock.debug('flow mod %r', flow_mod_instr)
c = AsmConstraint(loc_key, AsmConstraint.c_next)
# move dst if flowgraph modifier was in original block
# (usecase: split delayslot block)
if flow_mod_instr:
for xx in self.bto:
log_asmblock.debug('lbl %s', xx)
c_next = set(
x for x in self.bto if x.c_t == AsmConstraint.c_next
)
c_to = [x for x in self.bto if x.c_t != AsmConstraint.c_next]
self.bto = set([c] + c_to)
new_block.bto = c_next
else:
new_block.bto = self.bto
self.bto = set([c])
return new_block
def get_range(self):
"""Returns the offset hull of an AsmBlock"""
if len(self.lines):
return (self.lines[0].offset,
self.lines[-1].offset + self.lines[-1].l)
else:
return 0, 0
def get_offsets(self):
return [x.offset for x in self.lines]
def add_cst(self, loc_key, constraint_type):
"""
Add constraint between current block and block at @loc_key
@loc_key: LocKey instance of constraint target
@constraint_type: AsmConstraint c_to/c_next
"""
assert isinstance(loc_key, LocKey)
c = AsmConstraint(loc_key, constraint_type)
self.bto.add(c)
def get_flow_instr(self):
if not self.lines:
return None
for i in range(-1, -1 - self.lines[0].delayslot - 1, -1):
if not 0 <= i < len(self.lines):
return None
l = self.lines[i]
if l.splitflow() or l.breakflow():
raise NotImplementedError('not fully functional')
def get_subcall_instr(self):
if not self.lines:
return None
delayslot = self.lines[0].delayslot
end_index = len(self.lines) - 1
ds_max_index = max(end_index - delayslot, 0)
for i in range(end_index, ds_max_index - 1, -1):
l = self.lines[i]
if l.is_subcall():
return l
return None
def get_next(self):
for constraint in self.bto:
if constraint.c_t == AsmConstraint.c_next:
return constraint.loc_key
return None
@staticmethod
def _filter_constraint(constraints):
"""Sort and filter @constraints for AsmBlock.bto
@constraints: non-empty set of AsmConstraint instance
Always the same type -> one of the constraint
c_next and c_to -> c_next
"""
# Only one constraint
if len(constraints) == 1:
return next(iter(constraints))
# Constraint type -> set of corresponding constraint
cbytype = {}
for cons in constraints:
cbytype.setdefault(cons.c_t, set()).add(cons)
# Only one type -> any constraint is OK
if len(cbytype) == 1:
return next(iter(constraints))
# At least 2 types -> types = {c_next, c_to}
# c_to is included in c_next
return next(iter(cbytype[AsmConstraint.c_next]))
def fix_constraints(self):
"""Fix next block constraints"""
# destination -> associated constraints
dests = {}
for constraint in self.bto:
dests.setdefault(constraint.loc_key, set()).add(constraint)
self.bto = set(
self._filter_constraint(constraints)
for constraints in viewvalues(dests)
)
class asm_bloc(object):
def __init__(self, loc_key, alignment=1):
warnings.warn('DEPRECATION WARNING: use "AsmBlock" instead of "asm_bloc"')
super(asm_bloc, self).__init__(loc_key, alignment)
class AsmBlockBad(AsmBlock):
"""Stand for a *bad* ASM block (malformed, unreachable,
not disassembled, ...)"""
ERROR_UNKNOWN = -1
ERROR_CANNOT_DISASM = 0
ERROR_NULL_STARTING_BLOCK = 1
ERROR_FORBIDDEN = 2
ERROR_IO = 3
ERROR_TYPES = {
ERROR_UNKNOWN: "Unknown error",
ERROR_CANNOT_DISASM: "Unable to disassemble",
ERROR_NULL_STARTING_BLOCK: "Null starting block",
ERROR_FORBIDDEN: "Address forbidden by dont_dis",
ERROR_IO: "IOError",
}
def __init__(self, loc_key=None, alignment=1, errno=ERROR_UNKNOWN, *args, **kwargs):
"""Instantiate an AsmBlock_bad.
@loc_key, @alignment: same as AsmBlock.__init__
@errno: (optional) specify a error type associated with the block
"""
super(AsmBlockBad, self).__init__(loc_key, alignment, *args, **kwargs)
self._errno = errno
errno = property(lambda self: self._errno)
def __str__(self):
error_txt = self.ERROR_TYPES.get(self._errno, self._errno)
return "%s\n\tBad block: %s" % (
self.loc_key,
error_txt
)
def addline(self, *args, **kwargs):
raise RuntimeError("An AsmBlockBad cannot have line")
def addto(self, *args, **kwargs):
raise RuntimeError("An AsmBlockBad cannot have bto")
def split(self, *args, **kwargs):
raise RuntimeError("An AsmBlockBad cannot be split")
class asm_block_bad(AsmBlockBad):
def __init__(self, loc_key=None, alignment=1, errno=-1, *args, **kwargs):
warnings.warn('DEPRECATION WARNING: use "AsmBlockBad" instead of "asm_block_bad"')
super(asm_block_bad, self).__init__(loc_key, alignment, *args, **kwargs)
class AsmSymbolPool(LocationDB):
"""[DEPRECATED API] use 'LocationDB' instead"""
def __init__(self, *args, **kwargs):
warnings.warn("Deprecated API, use 'LocationDB' instead")
super(AsmSymbolPool, self).__init__(*args, **kwargs)
class asm_symbol_pool(AsmSymbolPool):
def __init__(self):
warnings.warn('DEPRECATION WARNING: use "LocationDB" instead of "asm_symbol_pool"')
super(asm_symbol_pool, self).__init__()
class AsmCFG(DiGraph):
"""Directed graph standing for a ASM Control Flow Graph with:
- nodes: AsmBlock
- edges: constraints between blocks, synchronized with AsmBlock's "bto"
Specialized the .dot export and force the relation between block to be uniq,
and associated with a constraint.
Offer helpers on AsmCFG management, such as research by loc_key, sanity
checking and mnemonic size guessing.
"""
# Internal structure for pending management
AsmCFGPending = namedtuple("AsmCFGPending",
["waiter", "constraint"])
def __init__(self, loc_db=None, *args, **kwargs):
super(AsmCFG, self).__init__(*args, **kwargs)
# Edges -> constraint
self.edges2constraint = {}
# Expected LocKey -> set( (src, dst), constraint )
self._pendings = {}
# Loc_Key2block built on the fly
self._loc_key_to_block = {}
# loc_db
self.loc_db = loc_db
def copy(self):
"""Copy the current graph instance"""
graph = self.__class__(self.loc_db)
return graph + self
# Compatibility with old list API
def append(self, *args, **kwargs):
raise DeprecationWarning("AsmCFG is a graph, use add_node")
def remove(self, *args, **kwargs):
raise DeprecationWarning("AsmCFG is a graph, use del_node")
def __getitem__(self, *args, **kwargs):
raise DeprecationWarning("Order of AsmCFG elements is not reliable")
def __contains__(self, _):
"""
DEPRECATED. Use:
- loc_key in AsmCFG.nodes() to test loc_key existence
"""
raise RuntimeError("DEPRECATED")
def __iter__(self):
"""
DEPRECATED. Use:
- AsmCFG.blocks() to iter on blocks
- loc_key in AsmCFG.nodes() to test loc_key existence
"""
raise RuntimeError("DEPRECATED")
def __len__(self):
"""Return the number of blocks in AsmCFG"""
return len(self._nodes)
@property
def blocks(self):
return viewvalues(self._loc_key_to_block)
# Manage graph with associated constraints
def add_edge(self, src, dst, constraint):
"""Add an edge to the graph
@src: LocKey instance, source
@dst: LocKey instance, destination
@constraint: constraint associated to this edge
"""
# Sanity check
assert isinstance(src, LocKey)
assert isinstance(dst, LocKey)
known_cst = self.edges2constraint.get((src, dst), None)
if known_cst is not None:
assert known_cst == constraint
return
# Add the edge to src.bto if needed
block_src = self.loc_key_to_block(src)
if block_src:
if dst not in [cons.loc_key for cons in block_src.bto]:
block_src.bto.add(AsmConstraint(dst, constraint))
# Add edge
self.edges2constraint[(src, dst)] = constraint
super(AsmCFG, self).add_edge(src, dst)
def add_uniq_edge(self, src, dst, constraint):
"""
Synonym for `add_edge`
"""
self.add_edge(src, dst, constraint)
def del_edge(self, src, dst):
"""Delete the edge @src->@dst and its associated constraint"""
src_blk = self.loc_key_to_block(src)
dst_blk = self.loc_key_to_block(dst)
assert src_blk is not None
assert dst_blk is not None
# Delete from src.bto
to_remove = [cons for cons in src_blk.bto if cons.loc_key == dst]
if to_remove:
assert len(to_remove) == 1
src_blk.bto.remove(to_remove[0])
# Del edge
del self.edges2constraint[(src, dst)]
super(AsmCFG, self).del_edge(src, dst)
def del_block(self, block):
super(AsmCFG, self).del_node(block.loc_key)
del self._loc_key_to_block[block.loc_key]
def add_node(self, node):
assert isinstance(node, LocKey)
return super(AsmCFG, self).add_node(node)
def add_block(self, block):
"""
Add the block @block to the current instance, if it is not already in
@block: AsmBlock instance
Edges will be created for @block.bto, if destinations are already in
this instance. If not, they will be resolved when adding these
aforementioned destinations.
`self.pendings` indicates which blocks are not yet resolved.
"""
status = super(AsmCFG, self).add_node(block.loc_key)
if not status:
return status
# Update waiters
if block.loc_key in self._pendings:
for bblpend in self._pendings[block.loc_key]:
self.add_edge(bblpend.waiter.loc_key, block.loc_key, bblpend.constraint)
del self._pendings[block.loc_key]
# Synchronize edges with block destinations
self._loc_key_to_block[block.loc_key] = block
for constraint in block.bto:
dst = self._loc_key_to_block.get(constraint.loc_key,
None)
if dst is None:
# Block is yet unknown, add it to pendings
to_add = self.AsmCFGPending(waiter=block,
constraint=constraint.c_t)
self._pendings.setdefault(constraint.loc_key,
set()).add(to_add)
else:
# Block is already in known nodes
self.add_edge(block.loc_key, dst.loc_key, constraint.c_t)
return status
def merge(self, graph):
"""Merge with @graph, taking in account constraints"""
# Add known blocks
for block in graph.blocks:
self.add_block(block)
# Add nodes not already in it (ie. not linked to a block)
for node in graph.nodes():
self.add_node(node)
# -> add_edge(x, y, constraint)
for edge in graph._edges:
# May fail if there is an incompatibility in edges constraints
# between the two graphs
self.add_edge(*edge, constraint=graph.edges2constraint[edge])
def node2lines(self, node):
if self.loc_db is None:
loc_key_name = node
else:
loc_key_name = self.loc_db.pretty_str(node)
yield self.DotCellDescription(text=loc_key_name,
attr={'align': 'center',
'colspan': 2,
'bgcolor': 'grey'})
block = self._loc_key_to_block.get(node, None)
if block is None:
return
if isinstance(block, AsmBlockBad):
yield [
self.DotCellDescription(
text=block.ERROR_TYPES.get(block._errno,
block._errno
),
attr={})
]
return
for line in block.lines:
if self._dot_offset:
yield [self.DotCellDescription(text="%.8X" % line.offset,
attr={}),
self.DotCellDescription(text=line.to_string(self.loc_db), attr={})]
else:
yield self.DotCellDescription(text=line.to_string(self.loc_db), attr={})
def node_attr(self, node):
block = self._loc_key_to_block.get(node, None)
if isinstance(block, AsmBlockBad):
return {'style': 'filled', 'fillcolor': 'red'}
return {}
def edge_attr(self, src, dst):
cst = self.edges2constraint.get((src, dst), None)
edge_color = "blue"
if len(self.successors(src)) > 1:
if cst == AsmConstraint.c_next:
edge_color = "red"
else:
edge_color = "limegreen"
return {"color": edge_color}
def dot(self, offset=False):
"""
@offset: (optional) if set, add the corresponding offsets in each node
"""
self._dot_offset = offset
return super(AsmCFG, self).dot()
# Helpers
@property
def pendings(self):
"""Dictionary of loc_key -> set(AsmCFGPending instance) indicating
which loc_key are missing in the current instance.
A loc_key is missing if a block which is already in nodes has constraints
with him (thanks to its .bto) and the corresponding block is not yet in
nodes
"""
return self._pendings
def label2block(self, loc_key):
"""
DEPRECATED: Use "loc_key_to_block" instead of "label2block"
Return the block corresponding to loc_key @loc_key
@loc_key: LocKey instance
"""
warnings.warn('DEPRECATION WARNING: use "loc_key_to_block" instead of "label2block"')
return self.loc_key_to_block(loc_key)
def rebuild_edges(self):
"""Consider blocks '.bto' and rebuild edges according to them, ie:
- update constraint type
- add missing edge
- remove no more used edge
This method should be called if a block's '.bto' in nodes have been
modified without notifying this instance to resynchronize edges.
"""
for block in self.blocks:
edges = []
# Rebuild edges from bto
for constraint in block.bto:
dst = self._loc_key_to_block.get(constraint.loc_key,
None)
if dst is None:
# Missing destination, add to pendings
self._pendings.setdefault(
constraint.loc_key,
set()
).add(
self.AsmCFGPending(
block,
constraint.c_t
)
)
continue
edge = (block.loc_key, dst.loc_key)
edges.append(edge)
if edge in self._edges:
# Already known edge, constraint may have changed
self.edges2constraint[edge] = constraint.c_t
else:
# An edge is missing
self.add_edge(edge[0], edge[1], constraint.c_t)
# Remove useless edges
for succ in self.successors(block.loc_key):
edge = (block.loc_key, succ)
if edge not in edges:
self.del_edge(*edge)
def get_bad_blocks(self):
"""Iterator on AsmBlockBad elements"""
# A bad asm block is always a leaf
for loc_key in self.leaves():
block = self._loc_key_to_block.get(loc_key, None)
if isinstance(block, AsmBlockBad):
yield block
def get_bad_blocks_predecessors(self, strict=False):
"""Iterator on loc_keys with an AsmBlockBad destination
@strict: (optional) if set, return loc_key with only bad
successors
"""
# Avoid returning the same block
done = set()
for badblock in self.get_bad_blocks():
for predecessor in self.predecessors_iter(badblock.loc_key):
if predecessor not in done:
if (strict and
not all(isinstance(self._loc_key_to_block.get(block, None), AsmBlockBad)
for block in self.successors_iter(predecessor))):
continue
yield predecessor
done.add(predecessor)
def getby_offset(self, offset):
"""Return asmblock containing @offset"""
for block in self.blocks:
if block.lines[0].offset <= offset < \
(block.lines[-1].offset + block.lines[-1].l):
return block
return None
def loc_key_to_block(self, loc_key):
"""
Return the asmblock corresponding to loc_key @loc_key, None if unknown
loc_key
@loc_key: LocKey instance
"""
return self._loc_key_to_block.get(loc_key, None)
def sanity_check(self):
"""Do sanity checks on blocks' constraints:
* no pendings
* no multiple next constraint to same block
* no next constraint to self
"""
if len(self._pendings) != 0:
raise RuntimeError(
"Some blocks are missing: %s" % list(
map(
str,
self._pendings
)
)
)
next_edges = {
edge: constraint
for edge, constraint in viewitems(self.edges2constraint)
if constraint == AsmConstraint.c_next
}
for loc_key in self._nodes:
if loc_key not in self._loc_key_to_block:
raise RuntimeError("Not supported yet: every node must have a corresponding AsmBlock")
# No next constraint to self
if (loc_key, loc_key) in next_edges:
raise RuntimeError('Bad constraint: self in next')
# No multiple next constraint to same block
pred_next = list(ploc_key
for (ploc_key, dloc_key) in next_edges
if dloc_key == loc_key)
if len(pred_next) > 1:
raise RuntimeError("Too many next constraints for block %r"
"(%s)" % (loc_key,
pred_next))
def guess_blocks_size(self, mnemo):
"""Asm and compute max block size
Add a 'size' and 'max_size' attribute on each block
@mnemo: metamn instance"""
for block in self.blocks:
size = 0
for instr in block.lines:
if isinstance(instr, AsmRaw):
# for special AsmRaw, only extract len
if isinstance(instr.raw, list):
data = None
if len(instr.raw) == 0:
l = 0
else:
l = (instr.raw[0].size // 8) * len(instr.raw)
elif isinstance(instr.raw, str):
data = instr.raw.encode()
l = len(data)
elif isinstance(instr.raw, bytes):
data = instr.raw
l = len(data)
else:
raise NotImplementedError('asm raw')
else:
# Assemble the instruction to retrieve its len.
# If the instruction uses symbol it will fail
# In this case, the max_instruction_len is used
try:
candidates = mnemo.asm(instr)
l = len(candidates[-1])
except:
l = mnemo.max_instruction_len
data = None
instr.data = data
instr.l = l
size += l
block.size = size
block.max_size = size
log_asmblock.info("size: %d max: %d", block.size, block.max_size)
def apply_splitting(self, loc_db, dis_block_callback=None, **kwargs):
"""Consider @self' bto destinations and split block in @self if one of
these destinations jumps in the middle of this block.
In order to work, they must be only one block in @self per loc_key in
@loc_db (which is true if @self come from the same disasmEngine).
@loc_db: LocationDB instance associated with @self'loc_keys
@dis_block_callback: (optional) if set, this callback will be called on
new block destinations
@kwargs: (optional) named arguments to pass to dis_block_callback
"""
# Get all possible destinations not yet resolved, with a resolved
# offset
block_dst = []
for loc_key in self.pendings:
offset = loc_db.get_location_offset(loc_key)
if offset is not None:
block_dst.append(offset)
todo = set(self.blocks)
rebuild_needed = False
while todo:
# Find a block with a destination inside another one
cur_block = todo.pop()
range_start, range_stop = cur_block.get_range()
for off in block_dst:
if not (off > range_start and off < range_stop):
continue
# `cur_block` must be split at offset `off`from miasm.core.locationdb import LocationDB
new_b = cur_block.split(loc_db, off)
log_asmblock.debug("Split block %x", off)
if new_b is None:
log_asmblock.error("Cannot split %x!!", off)
continue
# Remove pending from cur_block
# Links from new_b will be generated in rebuild_edges
for dst in new_b.bto:
if dst.loc_key not in self.pendings:
continue
self.pendings[dst.loc_key] = set(pending for pending in self.pendings[dst.loc_key]
if pending.waiter != cur_block)
# The new block destinations may need to be disassembled
if dis_block_callback:
offsets_to_dis = set(
self.loc_db.get_location_offset(constraint.loc_key)
for constraint in new_b.bto
)
dis_block_callback(cur_bloc=new_b,
offsets_to_dis=offsets_to_dis,
loc_db=loc_db, **kwargs)
# Update structure
rebuild_needed = True
self.add_block(new_b)
# The new block must be considered
todo.add(new_b)
range_start, range_stop = cur_block.get_range()
# Rebuild edges to match new blocks'bto
if rebuild_needed:
self.rebuild_edges()
def __str__(self):
out = []
for block in self.blocks:
out.append(str(block))
for loc_key_a, loc_key_b in self.edges():
out.append("%s -> %s" % (loc_key_a, loc_key_b))
return '\n'.join(out)
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, hex(id(self)))
# Out of _merge_blocks to be computed only once
_acceptable_block = lambda graph, loc_key: (not isinstance(graph.loc_key_to_block(loc_key), AsmBlockBad) and
len(graph.loc_key_to_block(loc_key).lines) > 0)
_parent = MatchGraphJoker(restrict_in=False, filt=_acceptable_block)
_son = MatchGraphJoker(restrict_out=False, filt=_acceptable_block)
_expgraph = _parent >> _son
def _merge_blocks(dg, graph):
"""Graph simplification merging AsmBlock with one and only one son with this
son if this son has one and only one parent"""
# Blocks to ignore, because they have been removed from the graph
to_ignore = set()
for match in _expgraph.match(graph):
# Get matching blocks
lbl_block, lbl_succ = match[_parent], match[_son]
block = graph.loc_key_to_block(lbl_block)
succ = graph.loc_key_to_block(lbl_succ)
# Ignore already deleted blocks
if (block in to_ignore or
succ in to_ignore):
continue
# Remove block last instruction if needed
last_instr = block.lines[-1]
if last_instr.delayslot > 0:
# TODO: delayslot
raise RuntimeError("Not implemented yet")
if last_instr.is_subcall():
continue
if last_instr.breakflow() and last_instr.dstflow():
block.lines.pop()
# Merge block
block.lines += succ.lines
for nextb in graph.successors_iter(lbl_succ):
graph.add_edge(lbl_block, nextb, graph.edges2constraint[(lbl_succ, nextb)])
graph.del_block(succ)
to_ignore.add(lbl_succ)
bbl_simplifier = DiGraphSimplifier()
bbl_simplifier.enable_passes([_merge_blocks])
def conservative_asm(mnemo, instr, symbols, conservative):
"""
Asm instruction;
Try to keep original instruction bytes if it exists
"""
candidates = mnemo.asm(instr, symbols)
if not candidates:
raise ValueError('cannot asm:%s' % str(instr))
if not hasattr(instr, "b"):
return candidates[0], candidates
if instr.b in candidates:
return instr.b, candidates
if conservative:
for c in candidates:
if len(c) == len(instr.b):
return c, candidates
return candidates[0], candidates
def fix_expr_val(expr, symbols):
"""Resolve an expression @expr using @symbols"""
def expr_calc(e):
if isinstance(e, ExprId):
# Example:
# toto:
# .dword label
loc_key = symbols.get_name_location(e.name)
offset = symbols.get_location_offset(loc_key)
e = ExprInt(offset, e.size)
return e
result = expr.visit(expr_calc)
result = expr_simp(result)
if not isinstance(result, ExprInt):
raise RuntimeError('Cannot resolve symbol %s' % expr)
return result
def fix_loc_offset(loc_db, loc_key, offset, modified):
"""
Fix the @loc_key offset to @offset. If the @offset has changed, add @loc_key
to @modified
@loc_db: current loc_db
"""
loc_offset = loc_db.get_location_offset(loc_key)
if loc_offset == offset:
return
loc_db.set_location_offset(loc_key, offset, force=True)
modified.add(loc_key)
class BlockChain(object):
"""Manage blocks linked with an asm_constraint_next"""
def __init__(self, loc_db, blocks):
self.loc_db = loc_db
self.blocks = blocks
self.place()
@property
def pinned(self):
"""Return True iff at least one block is pinned"""
return self.pinned_block_idx is not None
def _set_pinned_block_idx(self):
self.pinned_block_idx = None
for i, block in enumerate(self.blocks):
loc_key = block.loc_key
if self.loc_db.get_location_offset(loc_key) is not None:
if self.pinned_block_idx is not None:
raise ValueError("Multiples pinned block detected")
self.pinned_block_idx = i
def place(self):
"""Compute BlockChain min_offset and max_offset using pinned block and
blocks' size
"""
self._set_pinned_block_idx()
self.max_size = 0
for block in self.blocks:
self.max_size += block.max_size + block.alignment - 1
# Check if chain has one block pinned
if not self.pinned: