mirror of https://github.com/ConsenSys/mythril
Merge pull request #709 from ConsenSys/feature/detection-module
Detection Module Refactoringpull/713/head
Bernhard Mueller
6 years ago
committed by
GitHub
commit
0d736764f5
@ -0,0 +1,37 @@ |
||||
import logging |
||||
from typing import List |
||||
|
||||
|
||||
class DetectionModule: |
||||
def __init__( |
||||
self, |
||||
name: str, |
||||
swc_id: str, |
||||
hooks: List[str], |
||||
description: str, |
||||
entrypoint: str = "post", |
||||
): |
||||
self.name = name |
||||
self.swc_id = swc_id |
||||
self.hooks = hooks |
||||
self.description = description |
||||
if entrypoint not in ("post", "callback"): |
||||
logging.error( |
||||
"Invalid entrypoint in module %s, must be one of {post, callback}", |
||||
self.name, |
||||
) |
||||
self.entrypoint = entrypoint |
||||
|
||||
def execute(self, statespace): |
||||
raise NotImplementedError() |
||||
|
||||
def __repr__(self): |
||||
return ( |
||||
"<" |
||||
"DetectionModule " |
||||
"name={0.name} " |
||||
"swc_id={0.swc_id} " |
||||
"hooks={0.hooks} " |
||||
"description={0.description}" |
||||
">" |
||||
).format(self) |
||||
@ -1,83 +1,87 @@ |
||||
from mythril.analysis.report import Issue |
||||
from mythril.analysis.swc_data import * |
||||
from mythril.analysis.swc_data import MULTIPLE_SENDS |
||||
from mythril.analysis.modules.base import DetectionModule |
||||
from mythril.laser.ethereum.cfg import JumpType |
||||
|
||||
""" |
||||
MODULE DESCRIPTION: |
||||
|
||||
Check for multiple sends in a single transaction |
||||
""" |
||||
|
||||
|
||||
def execute(statespace): |
||||
issues = [] |
||||
|
||||
for call in statespace.calls: |
||||
findings = [] |
||||
# explore state |
||||
findings += _explore_states(call, statespace) |
||||
# explore nodes |
||||
findings += _explore_nodes(call, statespace) |
||||
|
||||
if len(findings) > 0: |
||||
node = call.node |
||||
instruction = call.state.get_current_instruction() |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
address=instruction["address"], |
||||
swc_id=MULTIPLE_SENDS, |
||||
bytecode=call.state.environment.code.bytecode, |
||||
title="Multiple Calls", |
||||
_type="Informational", |
||||
gas_used=( |
||||
call.state.mstate.min_gas_used, |
||||
call.state.mstate.max_gas_used, |
||||
), |
||||
) |
||||
|
||||
issue.description = ( |
||||
"Multiple sends are executed in a single transaction. " |
||||
"Try to isolate each external call into its own transaction," |
||||
" as external calls can fail accidentally or deliberately.\nConsecutive calls: \n" |
||||
) |
||||
class MultipleSendsModule(DetectionModule): |
||||
def __init__(self): |
||||
super().__init__( |
||||
name="Multiple Sends", |
||||
swc_id=MULTIPLE_SENDS, |
||||
hooks=["CALL", "DELEGATECALL", "STATICCALL", "CALLCODE"], |
||||
description="Check for multiple sends in a single transaction", |
||||
) |
||||
|
||||
for finding in findings: |
||||
issue.description += "Call at address: {}\n".format( |
||||
finding.state.get_current_instruction()["address"] |
||||
def execute(self, statespace): |
||||
issues = [] |
||||
|
||||
for call in statespace.calls: |
||||
findings = [] |
||||
# explore state |
||||
findings += self._explore_states(call, statespace) |
||||
# explore nodes |
||||
findings += self._explore_nodes(call, statespace) |
||||
|
||||
if len(findings) > 0: |
||||
node = call.node |
||||
instruction = call.state.get_current_instruction() |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
address=instruction["address"], |
||||
swc_id=MULTIPLE_SENDS, |
||||
bytecode=call.state.environment.code.bytecode, |
||||
title="Multiple Calls", |
||||
_type="Informational", |
||||
gas_used=( |
||||
call.state.mstate.min_gas_used, |
||||
call.state.mstate.max_gas_used, |
||||
), |
||||
) |
||||
|
||||
issues.append(issue) |
||||
return issues |
||||
|
||||
|
||||
def _explore_nodes(call, statespace): |
||||
children = _child_nodes(statespace, call.node) |
||||
sending_children = list(filter(lambda c: c.node in children, statespace.calls)) |
||||
return sending_children |
||||
|
||||
issue.description = ( |
||||
"Multiple sends are executed in a single transaction. " |
||||
"Try to isolate each external call into its own transaction," |
||||
" as external calls can fail accidentally or deliberately.\nConsecutive calls: \n" |
||||
) |
||||
|
||||
def _explore_states(call, statespace): |
||||
other_calls = list( |
||||
filter( |
||||
lambda other: other.node == call.node |
||||
and other.state_index > call.state_index, |
||||
statespace.calls, |
||||
for finding in findings: |
||||
issue.description += "Call at address: {}\n".format( |
||||
finding.state.get_current_instruction()["address"] |
||||
) |
||||
|
||||
issues.append(issue) |
||||
return issues |
||||
|
||||
def _explore_nodes(self, call, statespace): |
||||
children = self._child_nodes(statespace, call.node) |
||||
sending_children = list(filter(lambda c: c.node in children, statespace.calls)) |
||||
return sending_children |
||||
|
||||
def _explore_states(self, call, statespace): |
||||
other_calls = list( |
||||
filter( |
||||
lambda other: other.node == call.node |
||||
and other.state_index > call.state_index, |
||||
statespace.calls, |
||||
) |
||||
) |
||||
) |
||||
return other_calls |
||||
return other_calls |
||||
|
||||
def _child_nodes(self, statespace, node): |
||||
result = [] |
||||
children = [ |
||||
statespace.nodes[edge.node_to] |
||||
for edge in statespace.edges |
||||
if edge.node_from == node.uid and edge.type != JumpType.Transaction |
||||
] |
||||
|
||||
for child in children: |
||||
result.append(child) |
||||
result += self._child_nodes(statespace, child) |
||||
|
||||
def _child_nodes(statespace, node): |
||||
result = [] |
||||
children = [ |
||||
statespace.nodes[edge.node_to] |
||||
for edge in statespace.edges |
||||
if edge.node_from == node.uid and edge.type != JumpType.Transaction |
||||
] |
||||
return result |
||||
|
||||
for child in children: |
||||
result.append(child) |
||||
result += _child_nodes(statespace, child) |
||||
|
||||
return result |
||||
detector = MultipleSendsModule() |
||||
|
||||
@ -1,125 +1,127 @@ |
||||
from mythril.analysis.report import Issue |
||||
from mythril.analysis.swc_data import UNCHECKED_RET_VAL |
||||
from mythril.analysis.modules.base import DetectionModule |
||||
|
||||
from mythril.laser.ethereum.svm import NodeFlags |
||||
import logging |
||||
import re |
||||
|
||||
|
||||
""" |
||||
MODULE DESCRIPTION: |
||||
class UncheckedRetvalModule(DetectionModule): |
||||
def __init__(self): |
||||
super().__init__( |
||||
name="Unchecked Return Value", |
||||
swc_id=UNCHECKED_RET_VAL, |
||||
hooks=[], |
||||
description=( |
||||
"Test whether CALL return value is checked. " |
||||
"For direct calls, the Solidity compiler auto-generates this check. E.g.:\n" |
||||
" Alice c = Alice(address);\n" |
||||
" c.ping(42);\n" |
||||
"Here the CALL will be followed by IZSERO(retval), if retval = ZERO then state is reverted. " |
||||
"For low-level-calls this check is omitted. E.g.:\n" |
||||
' c.call.value(0)(bytes4(sha3("ping(uint256)")),1);' |
||||
), |
||||
) |
||||
|
||||
Test whether CALL return value is checked. |
||||
def execute(self, statespace): |
||||
|
||||
For direct calls, the Solidity compiler auto-generates this check. E.g.: |
||||
logging.debug("Executing module: UNCHECKED_RETVAL") |
||||
|
||||
Alice c = Alice(address); |
||||
c.ping(42); |
||||
issues = [] |
||||
|
||||
Here the CALL will be followed by IZSERO(retval), if retval = ZERO then state is reverted. |
||||
for k in statespace.nodes: |
||||
|
||||
For low-level-calls this check is omitted. E.g.: |
||||
node = statespace.nodes[k] |
||||
|
||||
c.call.value(0)(bytes4(sha3("ping(uint256)")),1); |
||||
if NodeFlags.CALL_RETURN in node.flags: |
||||
|
||||
""" |
||||
retval_checked = False |
||||
|
||||
for state in node.states: |
||||
|
||||
def execute(statespace): |
||||
instr = state.get_current_instruction() |
||||
|
||||
logging.debug("Executing module: UNCHECKED_RETVAL") |
||||
if instr["opcode"] == "ISZERO" and re.search( |
||||
r"retval", str(state.mstate.stack[-1]) |
||||
): |
||||
retval_checked = True |
||||
break |
||||
|
||||
issues = [] |
||||
if not retval_checked: |
||||
|
||||
for k in statespace.nodes: |
||||
address = state.get_current_instruction()["address"] |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
address=address, |
||||
bytecode=state.environment.code.bytecode, |
||||
title="Unchecked CALL return value", |
||||
swc_id=UNCHECKED_RET_VAL, |
||||
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used), |
||||
) |
||||
|
||||
node = statespace.nodes[k] |
||||
issue.description = ( |
||||
"The return value of an external call is not checked. " |
||||
"Note that execution continue even if the called contract throws." |
||||
) |
||||
|
||||
if NodeFlags.CALL_RETURN in node.flags: |
||||
issues.append(issue) |
||||
|
||||
retval_checked = False |
||||
else: |
||||
|
||||
for state in node.states: |
||||
n_states = len(node.states) |
||||
|
||||
instr = state.get_current_instruction() |
||||
for idx in range( |
||||
0, n_states - 1 |
||||
): # Ignore CALLs at last position in a node |
||||
|
||||
if instr["opcode"] == "ISZERO" and re.search( |
||||
r"retval", str(state.mstate.stack[-1]) |
||||
): |
||||
retval_checked = True |
||||
break |
||||
state = node.states[idx] |
||||
instr = state.get_current_instruction() |
||||
|
||||
if not retval_checked: |
||||
if instr["opcode"] == "CALL": |
||||
|
||||
address = state.get_current_instruction()["address"] |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
address=address, |
||||
bytecode=state.environment.code.bytecode, |
||||
title="Unchecked CALL return value", |
||||
swc_id=UNCHECKED_RET_VAL, |
||||
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used), |
||||
) |
||||
retval_checked = False |
||||
|
||||
issue.description = ( |
||||
"The return value of an external call is not checked. " |
||||
"Note that execution continue even if the called contract throws." |
||||
) |
||||
for _idx in range(idx, idx + 10): |
||||
|
||||
issues.append(issue) |
||||
try: |
||||
_state = node.states[_idx] |
||||
_instr = _state.get_current_instruction() |
||||
|
||||
else: |
||||
if _instr["opcode"] == "ISZERO" and re.search( |
||||
r"retval", str(_state.mstate.stack[-1]) |
||||
): |
||||
retval_checked = True |
||||
break |
||||
|
||||
n_states = len(node.states) |
||||
|
||||
for idx in range( |
||||
0, n_states - 1 |
||||
): # Ignore CALLs at last position in a node |
||||
except IndexError: |
||||
break |
||||
|
||||
state = node.states[idx] |
||||
instr = state.get_current_instruction() |
||||
if not retval_checked: |
||||
|
||||
if instr["opcode"] == "CALL": |
||||
address = instr["address"] |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
bytecode=state.environment.code.bytecode, |
||||
address=address, |
||||
title="Unchecked CALL return value", |
||||
swc_id=UNCHECKED_RET_VAL, |
||||
gas_used=( |
||||
state.mstate.min_gas_used, |
||||
state.mstate.max_gas_used, |
||||
), |
||||
) |
||||
|
||||
retval_checked = False |
||||
issue.description = ( |
||||
"The return value of an external call is not checked. " |
||||
"Note that execution continue even if the called contract throws." |
||||
) |
||||
|
||||
for _idx in range(idx, idx + 10): |
||||
issues.append(issue) |
||||
|
||||
try: |
||||
_state = node.states[_idx] |
||||
_instr = _state.get_current_instruction() |
||||
return issues |
||||
|
||||
if _instr["opcode"] == "ISZERO" and re.search( |
||||
r"retval", str(_state.mstate.stack[-1]) |
||||
): |
||||
retval_checked = True |
||||
break |
||||
|
||||
except IndexError: |
||||
break |
||||
|
||||
if not retval_checked: |
||||
|
||||
address = instr["address"] |
||||
issue = Issue( |
||||
contract=node.contract_name, |
||||
function_name=node.function_name, |
||||
bytecode=state.environment.code.bytecode, |
||||
address=address, |
||||
title="Unchecked CALL return value", |
||||
swc_id=UNCHECKED_RET_VAL, |
||||
gas_used=( |
||||
state.mstate.min_gas_used, |
||||
state.mstate.max_gas_used, |
||||
), |
||||
) |
||||
|
||||
issue.description = ( |
||||
"The return value of an external call is not checked. " |
||||
"Note that execution continue even if the called contract throws." |
||||
) |
||||
|
||||
issues.append(issue) |
||||
|
||||
return issues |
||||
detector = UncheckedRetvalModule() |
||||
|
||||
@ -1,22 +1,55 @@ |
||||
from mythril.analysis.report import Report |
||||
from collections import defaultdict |
||||
from ethereum.opcodes import opcodes |
||||
from mythril.analysis import modules |
||||
import pkgutil |
||||
import logging |
||||
|
||||
|
||||
def fire_lasers(statespace, module_names=None): |
||||
OPCODE_LIST = [c[0] for _, c in opcodes.items()] |
||||
|
||||
issues = [] |
||||
|
||||
def get_detection_module_hooks(): |
||||
hook_dict = defaultdict(list) |
||||
_modules = get_detection_modules(entrypoint="callback") |
||||
for module in _modules: |
||||
for op_code in map(lambda x: x.upper(), module.detector.hooks): |
||||
if op_code in OPCODE_LIST: |
||||
hook_dict[op_code].append(module.detector.execute) |
||||
elif op_code.endswith("*"): |
||||
to_register = filter(lambda x: x.startswith(op_code[:-1]), OPCODE_LIST) |
||||
for actual_hook in to_register: |
||||
hook_dict[actual_hook].append(module.detector.execute) |
||||
else: |
||||
logging.error( |
||||
"Encountered invalid hook opcode %s in module %s", |
||||
op_code, |
||||
module.detector.name, |
||||
) |
||||
return dict(hook_dict) |
||||
|
||||
|
||||
def get_detection_modules(entrypoint, except_modules=()): |
||||
except_modules = list(except_modules) + ["base"] |
||||
_modules = [] |
||||
|
||||
for loader, name, is_pkg in pkgutil.walk_packages(modules.__path__): |
||||
_modules.append(loader.find_module(name).load_module(name)) |
||||
for loader, name, _ in pkgutil.walk_packages(modules.__path__): |
||||
module = loader.find_module(name).load_module(name) |
||||
if ( |
||||
module.__name__ not in except_modules |
||||
and module.detector.entrypoint == entrypoint |
||||
): |
||||
_modules.append(module) |
||||
|
||||
logging.info("Found %s detection modules", len(_modules)) |
||||
return _modules |
||||
|
||||
|
||||
def fire_lasers(statespace, module_names=()): |
||||
logging.info("Starting analysis") |
||||
|
||||
for module in _modules: |
||||
if not module_names or module.__name__ in module_names: |
||||
logging.info("Executing " + str(module)) |
||||
issues += module.execute(statespace) |
||||
issues = [] |
||||
for module in get_detection_modules(entrypoint="post", except_modules=module_names): |
||||
logging.info("Executing " + module.detector.name) |
||||
issues += module.detector.execute(statespace) |
||||
|
||||
return issues |
||||
|
||||
Loading…
Reference in new issue