Add additional analysis modules (#1836)

* Update z3

* Add analysis modules

* Modify test
add/feats
Nikhil Parasaram 9 months ago committed by GitHub
parent 57dbd0cfd6
commit 13969337fd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
  1. 14
      mythril/analysis/module/loader.py
  2. 85
      mythril/analysis/module/modules/requirements_violation.py
  3. 138
      mythril/analysis/module/modules/transaction_order_dependence.py
  4. 142
      mythril/analysis/module/modules/unexpected_ether.py
  5. 5
      mythril/laser/ethereum/call.py
  6. 17
      tests/integration_tests/analysis_tests.py
  7. 1
      tests/integration_tests/safe_functions_test.py
  8. 15
      tests/testdata/input_contracts/requirements_violation_neg.sol
  9. 15
      tests/testdata/input_contracts/requirements_violation_pos.sol
  10. 28
      tests/testdata/input_contracts/tx.sol
  11. 16
      tests/testdata/input_contracts/unexpected_ether_neg.sol
  12. 16
      tests/testdata/input_contracts/unexpected_ether_pos.sol

@ -14,12 +14,17 @@ from mythril.analysis.module.modules.exceptions import Exceptions
from mythril.analysis.module.modules.external_calls import ExternalCalls from mythril.analysis.module.modules.external_calls import ExternalCalls
from mythril.analysis.module.modules.integer import IntegerArithmetics from mythril.analysis.module.modules.integer import IntegerArithmetics
from mythril.analysis.module.modules.multiple_sends import MultipleSends from mythril.analysis.module.modules.multiple_sends import MultipleSends
from mythril.analysis.module.modules.requirements_violation import RequirementsViolation
from mythril.analysis.module.modules.state_change_external_calls import ( from mythril.analysis.module.modules.state_change_external_calls import (
StateChangeAfterCall, StateChangeAfterCall,
) )
from mythril.analysis.module.modules.suicide import AccidentallyKillable from mythril.analysis.module.modules.suicide import AccidentallyKillable
from mythril.analysis.module.modules.transaction_order_dependence import (
TransactionOrderDependence,
)
from mythril.analysis.module.modules.unchecked_retval import UncheckedRetval from mythril.analysis.module.modules.unchecked_retval import UncheckedRetval
from mythril.analysis.module.modules.user_assertions import UserAssertions from mythril.analysis.module.modules.user_assertions import UserAssertions
from mythril.analysis.module.modules.unexpected_ether import UnexpectedEther
from mythril.analysis.module.base import EntryPoint from mythril.analysis.module.base import EntryPoint
@ -90,19 +95,22 @@ class ModuleLoader(object, metaclass=Singleton):
def _register_mythril_modules(self): def _register_mythril_modules(self):
self._modules.extend( self._modules.extend(
[ [
AccidentallyKillable(),
ArbitraryJump(), ArbitraryJump(),
ArbitraryStorage(), ArbitraryStorage(),
ArbitraryDelegateCall(), ArbitraryDelegateCall(),
PredictableVariables(),
TxOrigin(),
EtherThief(), EtherThief(),
Exceptions(), Exceptions(),
ExternalCalls(), ExternalCalls(),
IntegerArithmetics(), IntegerArithmetics(),
MultipleSends(), MultipleSends(),
PredictableVariables(),
RequirementsViolation(),
StateChangeAfterCall(), StateChangeAfterCall(),
AccidentallyKillable(), TransactionOrderDependence(),
TxOrigin(),
UncheckedRetval(), UncheckedRetval(),
UnexpectedEther(),
UserAssertions(), UserAssertions(),
] ]
) )

@ -0,0 +1,85 @@
"""This module contains the detection code for requirement violations in a call"""
import logging
from mythril.analysis import solver
from mythril.analysis.module.base import DetectionModule
from mythril.analysis.report import Issue
from mythril.analysis.issue_annotation import IssueAnnotation
from mythril.laser.smt import And
from mythril.analysis.swc_data import REQUIREMENT_VIOLATION
from mythril.exceptions import UnsatError
from mythril.laser.ethereum.state.global_state import GlobalState
from typing import List
log = logging.getLogger(__name__)
class RequirementsViolation(DetectionModule):
"""This module detects transaction order dependence"""
name = "Requirement Violation"
swc_id = REQUIREMENT_VIOLATION
description = "Checks whether any requirements violate in a call."
entrypoint = "callback"
pre_hooks = ["REVERT"]
plugin_default_enabled = True
def _execute(self, state: GlobalState) -> List[Issue]:
"""
:param state:
:return:
"""
issues = self._analyze_state(state)
for issue in issues:
self.cache.add((issue.source_location, issue.bytecode_hash))
return issues
def _analyze_state(self, state) -> List[Issue]:
"""
:param state:
:return:
"""
if len(state.transaction_stack) < 2:
return []
try:
address = state.get_current_instruction()["address"]
description_head = "A requirement was violated in a nested call and the call was reverted as a result."
description_tail = "Make sure valid inputs are provided to the nested call (for instance, via passed arguments)."
transaction_sequence = solver.get_transaction_sequence(
state, state.world_state.constraints
)
issue = Issue(
contract=state.environment.active_account.contract_name,
function_name=state.environment.active_function_name,
address=address,
swc_id=REQUIREMENT_VIOLATION,
title="requirement violation",
severity="Medium",
description_head=description_head,
description_tail=description_tail,
bytecode=state.environment.code.bytecode,
transaction_sequence=transaction_sequence,
gas_used=(state.mstate.min_gas_used, state.mstate.max_gas_used),
)
state.annotate(
IssueAnnotation(
conditions=[And(*state.world_state.constraints)],
issue=issue,
detector=self,
)
)
return [issue]
except UnsatError:
log.debug("no model found")
return []
detector = RequirementsViolation()

@ -0,0 +1,138 @@
"""This module contains the detection code for transaction order dependence."""
from mythril.analysis import solver
from mythril.analysis.potential_issues import (
PotentialIssue,
get_potential_issues_annotation,
)
from mythril.analysis.swc_data import TX_ORDER_DEPENDENCE
from mythril.laser.ethereum.transaction.symbolic import ACTORS
from mythril.analysis.module.base import DetectionModule
from mythril.laser.smt import Or, Bool
from mythril.laser.ethereum.state.global_state import GlobalState
from mythril.exceptions import UnsatError
import logging
log = logging.getLogger(__name__)
DESCRIPTION = """
Search for calls whose value depends on balance or storage.
"""
class BalanceAnnotation:
def __init__(self, caller):
self.caller = caller
class StorageAnnotation:
def __init__(self, caller):
self.caller = caller
class TransactionOrderDependence(DetectionModule):
"""This module detects transaction order dependence."""
name = "Transaction Order Dependence"
swc_id = TX_ORDER_DEPENDENCE
description = DESCRIPTION
entrypoint = "callback"
pre_hooks = ["CALL"]
post_hooks = ["BALANCE", "SLOAD"]
plugin_default_enabled = True
def _execute(self, state: GlobalState) -> None:
"""
:param state:
:return:
"""
potential_issues = self._analyze_state(state)
annotation = get_potential_issues_annotation(state)
annotation.potential_issues.extend(potential_issues)
@staticmethod
def annotate_balance_storage_vals(state, opcode):
val = state.mstate.stack[-1]
if opcode == "BALANCE":
annotation = BalanceAnnotation
else:
annotation = StorageAnnotation
if len(val.get_annotations(annotation)) == 0:
state.mstate.stack[-1].annotate(annotation(state.environment.sender))
return []
def _analyze_state(self, state: GlobalState):
"""
:param state:
:return:
"""
opcode = state.get_current_instruction()["opcode"]
if opcode != "CALL":
opcode = state.environment.code.instruction_list[state.mstate.pc - 1][
"opcode"
]
if opcode in ("BALANCE", "SLOAD"):
self.annotate_balance_storage_vals(state, opcode)
return []
value = state.mstate.stack[-3]
if (
len(value.get_annotations(StorageAnnotation))
+ len(value.get_annotations(BalanceAnnotation))
== 0
):
return []
callers = []
storage_annotations = value.get_annotations(StorageAnnotation)
if len(storage_annotations) == 1:
callers.append(storage_annotations[0].caller)
balance_annotations = value.get_annotations(BalanceAnnotation)
if len(balance_annotations) == 1:
callers.append(balance_annotations[0].caller)
address = state.get_current_instruction()["address"]
call_constraint = Bool(False)
for caller in callers:
call_constraint = Or(call_constraint, ACTORS.attacker == caller)
try:
constraints = state.world_state.constraints + [call_constraint]
solver.get_transaction_sequence(state, constraints)
description_head = (
"The value of the call is dependent on balance or storage write"
)
description_tail = (
"This can lead to race conditions. An attacker may be able to run a transaction after our transaction "
"which can change the value of the call"
)
issue = PotentialIssue(
contract=state.environment.active_account.contract_name,
function_name=state.environment.active_function_name,
address=address,
swc_id=TX_ORDER_DEPENDENCE,
title="Transaction Order Dependence",
bytecode=state.environment.code.bytecode,
severity="Medium",
description_head=description_head,
description_tail=description_tail,
constraints=constraints,
detector=self,
)
except UnsatError:
log.debug("[Transaction Order Dependence] No model found.")
return []
return [issue]
detector = TransactionOrderDependence()

@ -0,0 +1,142 @@
"""This module contains the detection code for unexpected ether balance."""
from mythril.analysis.report import Issue
from mythril.analysis.issue_annotation import IssueAnnotation
from mythril.analysis.swc_data import UNEXPECTED_ETHER_BALANCE
from mythril.analysis.module.base import DetectionModule
from mythril.analysis.module.module_helpers import is_prehook
from mythril.laser.smt import BitVec, And
from mythril.analysis.solver import UnsatError, get_transaction_sequence
from mythril.laser.ethereum.state.annotation import StateAnnotation
from mythril.laser.ethereum.state.global_state import GlobalState
import logging
log = logging.getLogger(__name__)
DESCRIPTION = """
Check for strict equality checks with contract balance
"""
class EtherBalanceCheckAnnotation(StateAnnotation):
"""State Annotation used if an overflow is both possible and used in the annotated path"""
def __init__(self, balance: BitVec) -> None:
self.balance = balance
class BalanceConditionAnnotation:
""""""
def __init__(self, address=None) -> None:
self.address = address
class UnexpectedEther(DetectionModule):
"""This module checks for strict equality checks with contract balance."""
author = "MythX Team"
plugin_license = "All rights reserved - ConsenSys"
plugin_type = "Detection Module"
plugin_description = DESCRIPTION
plugin_default_enabled = True
name = "Unexpected Ether Balance"
swc_id = "132"
description = DESCRIPTION
pre_hooks = ["INVALID", "EQ", "RETURN", "STOP"]
post_hooks = ["BALANCE"]
def _execute(self, state: GlobalState) -> None:
"""
:param state:
:return:
"""
return self._analyze_state(state)
def _analyze_state(self, state):
"""
:param state:
:return:
"""
node = state.node
instruction = state.get_current_instruction()
if is_prehook() is False:
balance = state.mstate.stack[-1]
annotations = state.get_annotations(EtherBalanceCheckAnnotation)
for annotation in annotations:
if annotation.balance == balance:
return []
state.annotate(EtherBalanceCheckAnnotation(balance))
return []
if instruction["opcode"] == "EQ":
op1, op2 = state.mstate.stack[-2:]
annotations = list(state.get_annotations(EtherBalanceCheckAnnotation))
op = None
for annotation in annotations:
if hash(annotation.balance) != hash(op1) and hash(
annotation.balance
) != hash(op2):
continue
if hash(annotation.balance) == hash(op1):
op = op1
else:
op = op2
break
if op is None:
return []
op.annotate(
BalanceConditionAnnotation(
address=state.get_current_instruction()["address"]
)
)
log.debug(
"Equality check for contract balance in function " + node.function_name
)
return []
for constraint in state.world_state.constraints:
for annotation in constraint.get_annotations(BalanceConditionAnnotation):
if annotation.address in self.cache:
continue
try:
transaction_sequence = get_transaction_sequence(
state, state.world_state.constraints
)
except UnsatError:
continue
log.debug("Detected a strict equality check")
title = "Strict Ether balance check"
description_head = "Use of strict ether balance checking"
description_tail = "Ether can be forcefully sent to this contract, This may make the contract unusable."
issue = Issue(
contract=state.environment.active_account.contract_name,
function_name=state.environment.active_function_name,
address=annotation.address,
title=title,
bytecode=state.environment.code.bytecode,
swc_id=UNEXPECTED_ETHER_BALANCE,
severity="Low",
description_head=description_head,
description_tail=description_tail,
transaction_sequence=transaction_sequence,
)
state.annotate(
IssueAnnotation(
conditions=[And(*state.world_state.constraints)],
issue=issue,
detector=self,
)
)
return [issue]
return []
detector = UnexpectedEther()

@ -206,16 +206,17 @@ def native_call(
if isinstance(callee_address, BitVec) or not ( if isinstance(callee_address, BitVec) or not (
0 < int(callee_address, 16) <= PRECOMPILE_COUNT 0 < int(callee_address, 16) <= PRECOMPILE_COUNT
or hevm_cheat_code.is_cheat_address(callee_address)
): ):
return None return None
# Disabled due to issues with how multi contract calls are handled with some older solc versions
"""
if hevm_cheat_code.is_cheat_address(callee_address): if hevm_cheat_code.is_cheat_address(callee_address):
log.info("HEVM cheat code address triggered") log.info("HEVM cheat code address triggered")
handle_cheat_codes( handle_cheat_codes(
global_state, callee_address, call_data, memory_out_offset, memory_out_size global_state, callee_address, call_data, memory_out_offset, memory_out_size
) )
return [global_state] return [global_state]
"""
log.debug("Native contract called: " + callee_address) log.debug("Native contract called: " + callee_address)
try: try:
mem_out_start = util.get_concrete_int(memory_out_offset) mem_out_start = util.get_concrete_int(memory_out_offset)

@ -80,3 +80,20 @@ def test_analysis_pending(file_name, tx_data, calldata):
0 0
] ]
assert test_case["steps"][tx_data["TX_OUTPUT"]]["input"] == calldata assert test_case["steps"][tx_data["TX_OUTPUT"]]["input"] == calldata
test_data_code = [
("114", f"{TESTDATA}/input_contracts/tx.sol", True),
("123", f"{TESTDATA}/input_contracts/requirements_violation_pos.sol", True),
("123", f"{TESTDATA}/input_contracts/requirements_violation_neg.sol", False),
("132", f"{TESTDATA}/input_contracts/unexpected_ether_pos.sol", True),
("132", f"{TESTDATA}/input_contracts/unexpected_ether_neg.sol", False),
]
@pytest.mark.parametrize("swc, file_path, exists", test_data_code)
def test_analysis_code(swc, file_path, exists):
assert (
"SWC ID: {}".format(swc)
in output_of(f"python3 {MYTH} a {file_path} --solver-timeout 5000")
) == exists

@ -14,7 +14,6 @@ test_data = (
"ether_send.sol", "ether_send.sol",
[ [
"crowdfunding()", "crowdfunding()",
"withdrawfunds()",
"owner()", "owner()",
"balances(address)", "balances(address)",
"getBalance()", "getBalance()",

@ -0,0 +1,15 @@
pragma solidity ^0.4.25;
contract Bar {
Foo private f = new Foo();
function doubleBaz() public view returns (int256) {
return 2 * f.baz(1); //Changes the external contract to not hit the overly strong requirement.
}
}
contract Foo {
function baz(int256 x) public pure returns (int256) {
require(0 < x); //You can also fix the contract by changing the input to the uint type and removing the require
return 42;
}
}

@ -0,0 +1,15 @@
pragma solidity ^0.4.25;
contract Bar {
Foo private f = new Foo();
function doubleBaz() public view returns (int256) {
return 2 * f.baz(0);
}
}
contract Foo {
function baz(int256 x) public pure returns (int256) {
require(0 < x);
return 42;
}
}

@ -0,0 +1,28 @@
pragma solidity ^0.4.16;
contract EthTxOrderDependenceMinimal {
address public owner;
bool public claimed;
uint public reward;
function EthTxOrderDependenceMinimal() public {
owner = msg.sender;
}
function setReward() public payable {
require (!claimed);
require(msg.sender == owner);
owner.transfer(reward);
reward = msg.value;
}
function claimReward(uint256 submission) {
require (!claimed);
require(submission < 10);
msg.sender.transfer(reward);
claimed = true;
}
}

@ -0,0 +1,16 @@
pragma solidity ^0.5.0;
contract Lockdrop {
function lock()
external
payable
{
uint256 eth = msg.value;
address owner = msg.sender;
assert(address(0x0).balance > msg.value);
}
}

@ -0,0 +1,16 @@
pragma solidity ^0.5.0;
contract Lockdrop {
function lock()
external
payable
{
uint256 eth = msg.value;
address owner = msg.sender;
assert(address(0x0).balance == msg.value);
}
}
Loading…
Cancel
Save