Merge pull request #235 from JoranHonig/features/module_clean

Structure + test delegatecall

mythril/analysis/modules/delegatecall.py

@@ -12,62 +12,72 @@ Check for invocations of delegatecall(msg.data) in the fallback function.
 
 
 def execute(statespace):
-
+    """
+    Executes analysis module for delegate call analysis module
+    :param statespace: Statespace to analyse
+    :return: Found issues
+    """
     issues = []
 
     for call in statespace.calls:
 
-        if (call.type == "DELEGATECALL"):
-
-            state = call.state
-            address = state.get_current_instruction()['address']
+        if call.type is not "DELEGATECALL":
+            continue
+        if call.node.function_name is not "fallback":
+            continue
 
-            if (call.node.function_name == "fallback"):
+        state = call.state
+        address = state.get_current_instruction()['address']
+        meminstart = get_variable(state.mstate.stack[-3])
 
-                stack = state.mstate.stack
-                meminstart = get_variable(stack[-3])
+        if meminstart.type == VarType.CONCRETE:
+            issues += _concrete_call(call, state, address, meminstart)
 
-                if meminstart.type == VarType.CONCRETE:
+        if call.to.type == VarType.SYMBOLIC:
+            issues += _symbolic_call(call, state, address, statespace)
 
-                    if (re.search(r'calldata.*_0', str(state.mstate.memory[meminstart.val]))):
+    return issues
 
-                        issue = Issue(call.node.contract_name, call.node.function_name, address, "Call data forwarded with delegatecall()", "Informational")
 
-                        issue.description = \
-                            "This contract forwards its call data via DELEGATECALL in its fallback function. " \
-                            "This means that any function in the called contract can be executed. Note that the callee contract will have access to the storage of the calling contract.\n"   
+def _concrete_call(call, state, address, meminstart):
+    if not re.search(r'calldata.*_0', str(state.mstate.memory[meminstart.val])):
+        return []
 
-                        if (call.to.type == VarType.CONCRETE):
-                            issue.description += ("DELEGATECALL target: " + hex(call.to.val))
-                        else:
-                            issue.description += "DELEGATECALL target: " + str(call.to)
+    issue = Issue(call.node.contract_name, call.node.function_name, address,
+                  "Call data forwarded with delegatecall()", "Informational")
 
-                        issues.append(issue)
+    issue.description = \
+        "This contract forwards its call data via DELEGATECALL in its fallback function. " \
+        "This means that any function in the called contract can be executed. Note that the callee contract will have " \
+        "access to the storage of the calling contract.\n "
 
-                if (call.to.type == VarType.SYMBOLIC):
+    target = hex(call.to.val) if call.to.type == VarType.CONCRETE else str(call.to)
+    issue.description += "DELEGATECALL target: {}".format(target)
 
-                    issue = Issue(call.node.contract_name, call.node.function_name, address, call.type + " to a user-supplied address")
+    return [issue]
 
-                    if ("calldata" in str(call.to)):
 
-                        issue.description = \
-                            "This contract delegates execution to a contract address obtained from calldata. "
+def _symbolic_call(call, state, address, statespace):
+    issue = Issue(call.node.contract_name, call.node.function_name, address, call.type + " to a user-supplied address")
 
-                    else:
-                        m = re.search(r'storage_([a-z0-9_&^]+)', str(call.to))
+    if "calldata" in str(call.to):
+        issue.description = \
+            "This contract delegates execution to a contract address obtained from calldata. "
 
-                        if (m):
-                            idx = m.group(1)
+    else:
+        m = re.search(r'storage_([a-z0-9_&^]+)', str(call.to))
 
-                        func = statespace.find_storage_write(state.environment.active_account.address, idx)
+        if m:
+            idx = m.group(1)
 
-                        if (func):
-                            issue.description = "This contract delegates execution to a contract address in storage slot " + str(idx) + ". This storage slot can be written to by calling the function `" + func + "`. "
+        func = statespace.find_storage_write(state.environment.active_account.address, idx)
 
-                        else:
-                            logging.debug("[DELEGATECALL] No storage writes to index " + str(idx))
+        if (func):
+            issue.description = "This contract delegates execution to a contract address in storage slot " + str(
+                idx) + ". This storage slot can be written to by calling the function `" + func + "`. "
 
-                        issue.description += "Be aware that the called contract gets unrestricted access to this contract's state."
-                        issues.append(issue)
+        else:
+            logging.debug("[DELEGATECALL] No storage writes to index " + str(idx))
 
-    return issues
+    issue.description += "Be aware that the called contract gets unrestricted access to this contract's state."
+    return [issue]

requirements.txt

@@ -19,3 +19,5 @@ eth-tester>=0.1.0b21
 coverage
 jinja2
 pytest
+mock
+pytest_mock

tests/analysis/test_delegatecall.py

@@ -0,0 +1,250 @@
+from mythril.analysis.modules.delegatecall import execute, _concrete_call, _symbolic_call
+from mythril.analysis.ops import Call, Variable, VarType
+from mythril.analysis.symbolic import SymExecWrapper
+from laser.ethereum.svm import GlobalState, Node, Environment, Account
+import pytest
+import mock
+from mock import patch
+import pytest_mock
+
+
+def test_concrete_call():
+    # arrange
+    address = "0x10"
+
+    state = GlobalState(None, None)
+    state.mstate.memory = ["placeholder", "calldata_bling_0"]
+
+    node = Node("example")
+    node.contract_name = "the contract name"
+    node.function_name = "the function name"
+
+    to = Variable(1, VarType.CONCRETE)
+    meminstart = Variable(1, VarType.CONCRETE)
+    call = Call(node, state, None, None, to, None)
+
+    # act
+    issues = _concrete_call(call, state, address, meminstart)
+
+    # assert
+    issue = issues[0]
+    assert issue.address == address
+    assert issue.contract == node.contract_name
+    assert issue.function == node.function_name
+    assert issue.title == "Call data forwarded with delegatecall()"
+    assert issue.type == 'Informational'
+    assert issue.description == "This contract forwards its call data via DELEGATECALL in its fallback function." \
+                                " This means that any function in the called contract can be executed." \
+                                " Note that the callee contract will have access to the storage of the " \
+                                "calling contract.\n DELEGATECALL target: 0x1"
+
+
+def test_concrete_call_symbolic_to():
+    # arrange
+    address = "0x10"
+
+    state = GlobalState(None, None)
+    state.mstate.memory = ["placeholder", "calldata_bling_0"]
+
+    node = Node("example")
+    node.contract_name = "the contract name"
+    node.function_name = "the function name"
+
+    to = Variable("calldata_3", VarType.SYMBOLIC)
+    meminstart = Variable(1, VarType.CONCRETE)
+    call = Call(node, state, None, None, to, None)
+
+    # act
+    issues = _concrete_call(call, state, address, meminstart)
+
+    # assert
+    issue = issues[0]
+    assert issue.address == address
+    assert issue.contract == node.contract_name
+    assert issue.function == node.function_name
+    assert issue.title == "Call data forwarded with delegatecall()"
+    assert issue.type == 'Informational'
+    assert issue.description == "This contract forwards its call data via DELEGATECALL in its fallback function." \
+                                " This means that any function in the called contract can be executed." \
+                                " Note that the callee contract will have access to the storage of the " \
+                                "calling contract.\n DELEGATECALL target: calldata_3"
+
+
+def test_concrete_call_not_calldata():
+    # arrange
+    state = GlobalState(None, None)
+    state.mstate.memory = ["placeholder", "not_calldata"]
+    meminstart = Variable(1, VarType.CONCRETE)
+
+    # act
+    issues = _concrete_call(None, state, None, meminstart)
+
+    # assert
+    assert issues == []
+
+
+def test_symbolic_call_storage_to(mocker):
+    # arrange
+    address = "0x10"
+
+    active_account = Account(address)
+    environment = Environment(active_account, None, None, None, None, None)
+    state = GlobalState(None, environment)
+    state.mstate.memory = ["placeholder", "calldata_bling_0"]
+
+
+    node = Node("example")
+    node.contract_name = "the contract name"
+    node.function_name = "the function name"
+
+    to = Variable("storage_1", VarType.SYMBOLIC)
+    call = Call(node, state, None, "Type: ", to, None)
+
+
+    mocker.patch.object(SymExecWrapper, "__init__", lambda x, y: None)
+    statespace = SymExecWrapper(1)
+
+    mocker.patch.object(statespace, 'find_storage_write')
+    statespace.find_storage_write.return_value = "Function name"
+
+
+    # act
+    issues = _symbolic_call(call, state, address, statespace)
+
+    # assert
+    issue = issues[0]
+    assert issue.address == address
+    assert issue.contract == node.contract_name
+    assert issue.function == node.function_name
+    assert issue.title == 'Type:  to a user-supplied address'
+    assert issue.type == 'Informational'
+    assert issue.description == 'This contract delegates execution to a contract address in storage slot 1.' \
+                                ' This storage slot can be written to by calling the function `Function name`. ' \
+                                'Be aware that the called contract gets unrestricted access to this contract\'s state.'
+
+
+def test_symbolic_call_calldata_to(mocker):
+    # arrange
+    address = "0x10"
+
+    state = GlobalState(None, None)
+    state.mstate.memory = ["placeholder", "calldata_bling_0"]
+
+
+    node = Node("example")
+    node.contract_name = "the contract name"
+    node.function_name = "the function name"
+
+    to = Variable("calldata", VarType.SYMBOLIC)
+    call = Call(node, state, None, "Type: ", to, None)
+
+
+    mocker.patch.object(SymExecWrapper, "__init__", lambda x, y: None)
+    statespace = SymExecWrapper(1)
+
+    mocker.patch.object(statespace, 'find_storage_write')
+    statespace.find_storage_write.return_value = "Function name"
+
+
+    # act
+    issues = _symbolic_call(call, state, address, statespace)
+
+    # assert
+    issue = issues[0]
+    assert issue.address == address
+    assert issue.contract == node.contract_name
+    assert issue.function == node.function_name
+    assert issue.title == 'Type:  to a user-supplied address'
+    assert issue.type == 'Informational'
+    assert issue.description == 'This contract delegates execution to a contract address obtained from calldata. ' \
+                                'Be aware that the called contract gets unrestricted access to this contract\'s state.'
+
+
+@patch('laser.ethereum.svm.GlobalState.get_current_instruction')
+@patch('mythril.analysis.modules.delegatecall._concrete_call')
+@patch('mythril.analysis.modules.delegatecall._symbolic_call')
+def test_delegate_call(sym_mock, concrete_mock, curr_instruction):
+    # arrange
+    # sym_mock = mocker.patch.object(delegatecall, "_symbolic_call")
+    # concrete_mock = mocker.patch.object(delegatecall, "_concrete_call")
+    sym_mock.return_value = []
+    concrete_mock.return_value = []
+    curr_instruction.return_value = {'address': '0x10'}
+
+    active_account = Account('0x10')
+    environment = Environment(active_account, None, None, None, None, None)
+    state = GlobalState(None, environment)
+    state.mstate.memory = ["placeholder", "calldata_bling_0"]
+    state.mstate.stack = [1, 2, 3]
+    assert state.get_current_instruction() == {'address': '0x10'}
+
+    node = Node("example")
+    node.contract_name = "the contract name"
+    node.function_name = "fallback"
+
+    to = Variable("storage_1", VarType.SYMBOLIC)
+    call = Call(node, state, None, "DELEGATECALL", to, None)
+
+    statespace = mock.MagicMock()
+    statespace.calls = [call]
+
+    # act
+    issues = execute(statespace)
+
+    # assert
+    assert concrete_mock.call_count == 1
+    assert sym_mock.call_count == 1
+
+
+@patch('mythril.analysis.modules.delegatecall._concrete_call')
+@patch('mythril.analysis.modules.delegatecall._symbolic_call')
+def test_delegate_call_not_delegate(sym_mock, concrete_mock):
+    # arrange
+    # sym_mock = mocker.patch.object(delegatecall, "_symbolic_call")
+    # concrete_mock = mocker.patch.object(delegatecall, "_concrete_call")
+    sym_mock.return_value = []
+    concrete_mock.return_value = []
+
+    node = Node("example")
+    node.function_name = "fallback"
+
+    to = Variable("storage_1", VarType.SYMBOLIC)
+    call = Call(node, None, None, "NOT_DELEGATECALL", to, None)
+
+    statespace = mock.MagicMock()
+    statespace.calls = [call]
+
+    # act
+    issues = execute(statespace)
+
+    # assert
+    assert issues == []
+    assert concrete_mock.call_count == 0
+    assert sym_mock.call_count == 0
+
+
+@patch('mythril.analysis.modules.delegatecall._concrete_call')
+@patch('mythril.analysis.modules.delegatecall._symbolic_call')
+def test_delegate_call_not_fallback(sym_mock, concrete_mock):
+    # arrange
+    # sym_mock = mocker.patch.object(delegatecall, "_symbolic_call")
+    # concrete_mock = mocker.patch.object(delegatecall, "_concrete_call")
+    sym_mock.return_value = []
+    concrete_mock.return_value = []
+
+    node = Node("example")
+    node.function_name = "not_fallback"
+
+    to = Variable("storage_1", VarType.SYMBOLIC)
+    call = Call(node, None, None, "DELEGATECALL", to, None)
+
+    statespace = mock.MagicMock()
+    statespace.calls = [call]
+
+    # act
+    issues = execute(statespace)
+
+    # assert
+    assert issues == []
+    assert concrete_mock.call_count == 0
+    assert sym_mock.call_count == 0