TheDude
/
mythril
mirror of https://github.com/ConsenSys/mythril
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Rewrite external_calls, refactor thief & extcalls to callbacks

Browse Source
pull/757/head
Bernhard Mueller 6 years ago
parent a0b1721989
commit e441907c38
3 changed files with 118 additions and 121 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 114
      mythril/analysis/modules/ether_thief.py
  2. 123
      mythril/analysis/modules/external_calls.py
  3. 2
      mythril/laser/ethereum/instructions.py

114
mythril/analysis/modules/ether_thief.py
Unescape View File

@ -3,6 +3,7 @@ from mythril.analysis import solver
from mythril.analysis.report import Issue
from mythril.analysis.swc_data import UNPROTECTED_ETHER_WITHDRAWAL
from mythril.analysis.modules.base import DetectionModule
from mythril.laser.ethereum.state.global_state import GlobalState
from mythril.exceptions import UnsatError
from z3 import BitVecVal, UGT, Sum
import logging
@ -22,80 +23,79 @@ An issue is reported if:
"""
class EtherThief(DetectionModule):
def __init__(self):
super().__init__(
name="Ether Thief",
swc_id=UNPROTECTED_ETHER_WITHDRAWAL,
hooks=["CALL"],
description=DESCRIPTION,
)
def _analyze_state(state):
instruction = state.get_current_instruction()
node = state.node
def execute(self, state_space):
logging.debug("Executing module: %s", self.name)
issues = []
if instruction["opcode"] != "CALL":
return []
for k in state_space.nodes:
node = state_space.nodes[k]
for state in node.states:
issues += self._analyze_state(state, node)
call_value = state.mstate.stack[-3]
target = state.mstate.stack[-2]
return issues
eth_sent_total = BitVecVal(0, 256)
@staticmethod
def _analyze_state(state, node):
issues = []
instruction = state.get_current_instruction()
constraints = node.constraints
if instruction["opcode"] != "CALL":
return []
for tx in state.world_state.transaction_sequence:
if tx.caller == 0xDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF:
call_value = state.mstate.stack[-3]
target = state.mstate.stack[-2]
# There's sometimes no overflow check on balances added.
# But we don't care about attacks that require more 2^^256 ETH to be sent.
eth_sent_total = BitVecVal(0, 256)
constraints += [BVAddNoOverflow(eth_sent_total, tx.call_value, False)]
eth_sent_total = Sum(eth_sent_total, tx.call_value)
constraints = node.constraints
constraints += [
UGT(call_value, eth_sent_total),
target == state.environment.sender,
]
for tx in state.world_state.transaction_sequence:
if tx.caller == 0xDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF:
try:
# There's sometimes no overflow check on balances added.
# But we don't care about attacks that require more 2^^256 ETH to be sent.
transaction_sequence = solver.get_transaction_sequence(state, constraints)
constraints += [BVAddNoOverflow(eth_sent_total, tx.call_value, False)]
eth_sent_total = Sum(eth_sent_total, tx.call_value)
debug = str(transaction_sequence)
constraints += [
UGT(call_value, eth_sent_total),
target == state.environment.sender,
]
issue = Issue(
contract=node.contract_name,
function_name=node.function_name,
address=instruction["address"],
swc_id=UNPROTECTED_ETHER_WITHDRAWAL,
title="Ether thief",
_type="Warning",
bytecode=state.environment.code.bytecode,
description="Arbitrary senders other than the contract creator can withdraw ETH from the contract"
+ " account without previously having sent an equivalent amount of ETH to it. This is likely to be"
+ " a vulnerability.",
debug=debug,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
except UnsatError:
logging.debug("[ETHER_THIEF] no model found")
return []
try:
return [issue]
transaction_sequence = solver.get_transaction_sequence(state, constraints)
debug = str(transaction_sequence)
class EtherThief(DetectionModule):
def __init__(self):
super().__init__(
name="Ether Thief",
swc_id=UNPROTECTED_ETHER_WITHDRAWAL,
hooks=["CALL"],
description=DESCRIPTION,
entrypoint="callback",
)
self._issues = []
issue = Issue(
contract=node.contract_name,
function_name=node.function_name,
address=instruction["address"],
swc_id=UNPROTECTED_ETHER_WITHDRAWAL,
title="Ether thief",
_type="Warning",
bytecode=state.environment.code.bytecode,
description="Arbitrary senders other than the contract creator can withdraw ETH from the contract"
+ " account without previously having sent an equivalent amount of ETH to it. This is likely to be"
+ " a vulnerability.",
debug=debug,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
issues.append(issue)
except UnsatError:
logging.debug("[ETHER_THIEF] no model found")
def execute(self, state: GlobalState):
self._issues.extend(_analyze_state(state))
return self.issues
return issues
@property
def issues(self):
return self._issues
detector = EtherThief()

123
mythril/analysis/modules/external_calls.py
Unescape View File

@ -1,88 +1,43 @@
from z3 import *
from mythril.analysis.ops import *
from mythril.analysis.report import Issue
from mythril.analysis import solver
from mythril.analysis.swc_data import REENTRANCY
from mythril.analysis.modules.base import DetectionModule
from mythril.laser.ethereum.state.global_state import GlobalState
from mythril.exceptions import UnsatError
import logging
DESCRIPTION = """
Search for low level calls (e.g. call.value()) that forward all gas to the callee.
Search for low level calls (e.g. call.value()) that forward all gas to the callee.
Report a warning if the callee address can be set by the sender, otherwise create
an informational issue.
"""
class ExternalCallModule(DetectionModule):
def __init__(self):
super().__init__(
name="External Calls",
swc_id=REENTRANCY,
hooks=["CALL", "DELEGATECALL", "STATICCALL", "CALLCODE"],
description="Check for call.value()() to external addresses",
)
def execute(self, state_space):
logging.debug("Executing module: %s", self.name)
issues = []
for k in state_space.nodes:
node = state_space.nodes[k]
for state in node.states:
issues += self._analyze_state(state, node)
return issues
def _analyze_state(state):
@staticmethod
def _analyze_state(state, node):
issues = []
instruction = state.get_current_instruction()
node = state.node
gas = state.mstate.stack[-1]
to = state.mstate.stack[-2]
if instruction["opcode"] not in ("CALL", "CALLCODE", "DELEGATECALL", "STATICCALL"):
return []
address = state.get_current_instruction()["address"]
gas = state.mstate.stack[-1]
to = state.mstate.stack[-2]
try:
constraints = node.constraints + [gas > 2300]
transaction_sequence = solver.get_transaction_sequence(state, constraints)
address = state.get_current_instruction()["address"]
# Check whether we can also set the callee address
try:
constraints = state.node.constraints + [gas > 2300]
constraints += [to == 0xDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF]
transaction_sequence = solver.get_transaction_sequence(state, constraints)
try:
constraints += [to == 0xDEADBEEFDEADBEEFDEADBEEFDEADBEEFDEADBEEF]
transaction_sequence = solver.get_transaction_sequence(state, constraints)
except UnsatError:
debug = str(transaction_sequence)
description = "The contract executes a function call to an external address. " \
"Verify that the code at this address is trusted and immutable."
issue = Issue(
contract=node.contract_name,
function_name=state.node.function_name,
address=address,
swc_id=REENTRANCY,
title="External call",
_type="Informational",
bytecode=state.environment.code.bytecode,
description=description,
debug=debug,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
issues.append(issue)
return issues
debug = str(transaction_sequence)
description = "The contract executes a function call with high gas to a user-supplied address. " \
"Note that the callee can contain arbitrary code and may re-enter any function in this contract. " \
"Review the business logic carefully to prevent unanticipated effects on the contract state."
"Note that the callee can contain arbitrary code and may re-enter any function in this contract. " \
"Review the business logic carefully to prevent unanticipated effects on the contract state."
issue = Issue(
contract=node.contract_name,
@ -96,11 +51,53 @@ class ExternalCallModule(DetectionModule):
debug=debug,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
issues.append(issue)
return issues
except UnsatError:
logging.debug("[EXTERNAL_CALLS] no model found for setting caller address.")
logging.debug("[EXTERNAL_CALLS] Callee address cannot be modified. Reporting informational issue.")
debug = str(transaction_sequence)
description = "The contract executes a function call to an external address. " \
"Verify that the code at this address is trusted and immutable."
issue = Issue(
contract=node.contract_name,
function_name=state.node.function_name,
address=address,
swc_id=REENTRANCY,
title="External call",
_type="Informational",
bytecode=state.environment.code.bytecode,
description=description,
debug=debug,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
except UnsatError:
logging.debug("[EXTERNAL_CALLS] No model found.")
return []
return [issue]
class ExternalCalls(DetectionModule):
def __init__(self):
super().__init__(
name="External calls",
swc_id=REENTRANCY,
hooks=["CALL"],
description=(DESCRIPTION),
entrypoint="callback",
)
self._issues = []
def execute(self, state: GlobalState):
self._issues.extend(_analyze_state(state))
return self.issues
@property
def issues(self):
return self._issues
detector = ExternalCallModule()
detector = ExternalCalls()

2
mythril/laser/ethereum/instructions.py
Unescape View File

@ -131,7 +131,7 @@ class Instruction:
def evaluate(self, global_state: GlobalState, post=False) -> List[GlobalState]:
""" Performs the mutation for this instruction """
# Generalize some ops
logging.debug("Evaluating {}".format(self.op_code))
# logging.debug("Evaluating {}".format(self.op_code))
op = self.op_code.lower()
if self.op_code.startswith("PUSH"):
op = "push"

Write Preview
Loading…
Cancel
Save