diff --git a/mythril/analysis/modules/ether_thief.py b/mythril/analysis/modules/ether_thief.py index 315c9e64..101b1b8b 100644 --- a/mythril/analysis/modules/ether_thief.py +++ b/mythril/analysis/modules/ether_thief.py @@ -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() diff --git a/mythril/analysis/modules/external_calls.py b/mythril/analysis/modules/external_calls.py index 0562ca9c..2476d5e0 100644 --- a/mythril/analysis/modules/external_calls.py +++ b/mythril/analysis/modules/external_calls.py @@ -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() diff --git a/mythril/laser/ethereum/instructions.py b/mythril/laser/ethereum/instructions.py index 7ade087d..16f2fc82 100644 --- a/mythril/laser/ethereum/instructions.py +++ b/mythril/laser/ethereum/instructions.py @@ -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"