-
Notifications
You must be signed in to change notification settings - Fork 982
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add Multiple Security Rule Checks for DeFi Projects #2388
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Large diffs are not rendered by default.
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,62 @@ | ||
import re | ||
from slither.detectors.abstract_detector import AbstractDetector, DetectorClassification | ||
|
||
|
||
class AMMKValueError(AbstractDetector): | ||
""" | ||
Module detecting potential errors in K-value calculation in Automated Market Makers (AMMs). | ||
""" | ||
|
||
ARGUMENT = "amm-k-value-error" | ||
HELP = "Potential K-value calculation errors" | ||
IMPACT = DetectorClassification.HIGH | ||
CONFIDENCE = DetectorClassification.HIGH | ||
|
||
WIKI = " " | ||
|
||
WIKI_TITLE = "AMM K-Value Error" | ||
WIKI_DESCRIPTION = ( | ||
"" | ||
) | ||
|
||
# region wiki_exploit_scenario | ||
WIKI_EXPLOIT_SCENARIO = """""" | ||
WIKI_RECOMMENDATION = "Ensure correct calculation of the K-value in Automated Market Makers (AMMs) to avoid potential trading issues." | ||
|
||
def _detect(self): | ||
""" | ||
Detect multiple constructor schemes in the same contract | ||
:return: Returns a list of contract JSON result, where each result contains all constructor definitions. | ||
""" | ||
results = [] | ||
matches0="" | ||
matches1="" | ||
matches2="" | ||
tainted_function_name="" | ||
tainted_nodes=[] | ||
|
||
for contract in self.contracts: | ||
# check if uniswap | ||
if "pair" in contract.name.lower() and any("swap" == f.name for f in contract.functions) and any("burn" == f.name for f in contract.functions) and any("mint" == f.name for f in contract.functions): | ||
print("found function") | ||
for f in contract.functions: | ||
if f.name=="swap": | ||
print("found swap") | ||
tainted_function_name=f.name | ||
for node in f.nodes: | ||
pattern = r'10+' | ||
if "balance0.mul(" in str(node): | ||
matches0 = re.findall(pattern, str(node)) | ||
tainted_nodes.append(node) | ||
if "balance1.mul(" in str(node): | ||
matches1 = re.findall(pattern, str(node)) | ||
tainted_nodes.append(node) | ||
if "require" in str(node) and ">=" in str(node): | ||
matches2 = re.findall(pattern, str(node)) | ||
tainted_nodes.append(node) | ||
if matches2!=matches0 or matches2!=matches1: | ||
info = [tainted_function_name, " has potential K Value Error in :\n",tainted_nodes[0],tainted_nodes[1],tainted_nodes[2]] | ||
res = self.generate_result(info) | ||
results.append(res) | ||
|
||
return results |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,255 @@ | ||
from typing import List | ||
from slither.detectors.abstract_detector import AbstractDetector, DetectorClassification | ||
from slither.utils.output import Output | ||
from slither.core.declarations import Contract | ||
from slither.core.variables.state_variable import StateVariable | ||
from slither.core.declarations import FunctionContract, Modifier | ||
from slither.core.cfg.node import NodeType, Node | ||
from slither.core.declarations.event import Event | ||
from slither.core.expressions import CallExpression, Identifier | ||
from slither.analyses.data_dependency.data_dependency import is_dependent | ||
|
||
from slither.core.declarations.solidity_variables import ( | ||
SolidityFunction, | ||
) | ||
from slither.core.variables.local_variable import LocalVariable | ||
from slither.core.variables.state_variable import StateVariable | ||
|
||
from slither.core.expressions import CallExpression | ||
from slither.core.expressions.assignment_operation import AssignmentOperation | ||
|
||
from slither.slithir.operations import ( | ||
EventCall, | ||
) | ||
from slither.detectors.defi.price_manipulation_tools import PriceManipulationTools | ||
|
||
class PriceManipulation(AbstractDetector): | ||
""" | ||
Detect when `msg.sender` is not used as `from` in transferFrom along with the use of permit. | ||
""" | ||
|
||
ARGUMENT = "price-manipulation-high" | ||
HELP = "transferFrom uses arbitrary from with permit" | ||
IMPACT = DetectorClassification.HIGH | ||
CONFIDENCE = DetectorClassification.MEDIUM | ||
|
||
WIKI = " https://metatrust.feishu.cn/wiki/wikcnley0RNMaoaSzdjcCpYxYoD" | ||
|
||
WIKI_TITLE = "The risk of price manipulation in DeFi projects" | ||
WIKI_DESCRIPTION = ( | ||
"Price manipulation is a common attack in DeFi projects. " | ||
) | ||
WIKI_EXPLOIT_SCENARIO = """""" | ||
|
||
WIKI_RECOMMENDATION = """""" | ||
# Functions that may return abnormal values due to price manipulation | ||
ERC20_FUNCTION = [ | ||
# "balanceOf", | ||
# "totalSupply", | ||
"getReserves", | ||
# "balance", | ||
"getAmountsOut", | ||
"getAmountOut" | ||
] | ||
# TODO: Problems only occur when the affected variables are subject to addition or multiplication operations, otherwise there will be no problem | ||
def checkIfHavePriceManipulation(self,contract:Contract): | ||
result_dependent_data=[] | ||
result_call_data=[] | ||
if contract.is_interface: | ||
return result_call_data,result_dependent_data | ||
for function in contract.functions: | ||
return_vars=[] | ||
# Collection 1: Get all sensitive functions and variables related to fund transfer in the function DANGEROUS_ERC20_FUNCTION | ||
dangerous_calls=self._get_all_dangerous_operation_variables(function) | ||
# Collection 4: All assigned variables and the underlying ERC20 operations involved in the function | ||
erc20_vars=[] | ||
erc20_calls=[] | ||
erc20_nodes=[] | ||
for node in function.nodes: | ||
# Get assigned variables and all involved underlying ERC20 operations in the node | ||
node_vars,node_calls=self._get_calls_and_var_recursively_node(node) | ||
if len(node_calls)>0: | ||
erc20_vars.append(node_vars) | ||
erc20_calls.append(node_calls) | ||
erc20_nodes.append(node) | ||
# Whether there is is_dependent between Collection 1 and Collection 4 variables | ||
# All sensitive variables in the function | ||
all_risk_vars = [arg.value for call in dangerous_calls for arg in call.arguments if isinstance(arg,Identifier) and (isinstance(arg.value,LocalVariable) or isinstance(arg.value,StateVariable))] | ||
for risk_var in all_risk_vars: | ||
for dangerous_erc20_vars,dangerous_erc20_calls,node in zip(erc20_vars,erc20_calls,erc20_nodes): | ||
for dangerous_erc20_var,dangerous_erc20_call in zip(dangerous_erc20_vars,dangerous_erc20_calls): | ||
if is_dependent(risk_var, dangerous_erc20_var, function): | ||
result_dependent_data.append([function,risk_var,dangerous_erc20_var,dangerous_erc20_call,node]) | ||
# print("risk variable in",function.name,":",risk_var.canonical_name,"rely on",dangerous_erc20_var.canonical_name,"with call:",dangerous_erc20_call) | ||
return result_dependent_data,result_call_data | ||
Comment on lines
+55
to
+84
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The method |
||
|
||
# Recursive retrieval of child calls from functions | ||
@staticmethod | ||
def _get_calls_recursively(func: FunctionContract, maxdepth=10): | ||
ret=[] | ||
if maxdepth<=0: | ||
return ret | ||
if hasattr(func,"calls_as_expressions"): | ||
if len(func.calls_as_expressions) > 0: | ||
for call in func.calls_as_expressions: | ||
if PriceManipulation._check_call_can_output(call): | ||
if str(call.called.value) in PriceManipulation.ERC20_FUNCTION: | ||
if not (len(call.arguments)==1 and str(call.arguments[0])=="address(this)"): | ||
ret.append(call) | ||
else: | ||
ret.extend(PriceManipulation._get_calls_recursively(call.called.value,maxdepth=maxdepth-1)) | ||
elif isinstance(call, CallExpression) and \ | ||
call.called and not hasattr(call.called, 'value'): | ||
# When there is an external call, only consider whether there is a balanceof and other ERC20 external calls, ignoring other calls | ||
# Other calls can be added here to ensure that the data returned by external projects will not have problems or check potential problems | ||
if hasattr(call.called,"member_name") and call.called.member_name in PriceManipulation.ERC20_FUNCTION: | ||
if not (len(call.arguments)==1 and str(call.arguments[0])=="address(this)"): | ||
ret.append(call) | ||
return ret | ||
|
||
@staticmethod | ||
def _check_if_can_output_call_info(call): | ||
argument=call.arguments[0] | ||
# balanceOf(a) | ||
if (hasattr(argument,"value") and (isinstance(argument.value,StateVariable)) or "pair" in str(argument).lower()) or (hasattr(argument,"expression") and hasattr(argument.expression,"value") and (isinstance(argument.expression.value,StateVariable)) or "pair" in str(argument.expression.value).lower()): | ||
return call | ||
|
||
# Get all sensitive operations related to transfer and minting in the function | ||
@staticmethod | ||
def _get_all_dangerous_operation_variables(func:FunctionContract): | ||
ret_calls=[] | ||
ret_vars=[] | ||
for call in func.calls_as_expressions: | ||
if (call.called and hasattr(call.called,"member_name") and call.called.member_name in PriceManipulationTools.DANGEROUS_ERC20_FUNCTION) or \ | ||
(call.called and hasattr(call.called,"value") and call.called.value.name in PriceManipulationTools.DANGEROUS_ERC20_FUNCTION): | ||
ret_calls.append(call) | ||
return ret_calls | ||
|
||
# Get all return variables operations in the function | ||
@staticmethod | ||
def _get_all_return_variables(func:FunctionContract): | ||
ret=[] | ||
for node in func.nodes: | ||
if node.will_return and len(node.variables_read)>0: | ||
ret.extend(node.variables_read) | ||
ret.extend(func.returns) | ||
return ret | ||
|
||
# Get all sensitive function return operations in the function | ||
@staticmethod | ||
def _get_all_return_calls(func:FunctionContract): | ||
ret_calls=[] | ||
for node in func.nodes: | ||
if node.will_return and "require" not in str(node) and hasattr(node,"calls_as_expression") and len(node.calls_as_expression)>0: | ||
_,calls=PriceManipulation._get_calls_and_var_recursively_node(node) | ||
|
||
for call in calls: | ||
if isinstance(call,SolidityFunction): | ||
ret_calls.append((call,node)) | ||
elif hasattr(call,"called") and \ | ||
((call.called and hasattr(call.called,"member_name") and call.called.member_name in PriceManipulation.ERC20_FUNCTION) or \ | ||
(call.called and hasattr(call.called,"value") and call.called.value.name in PriceManipulation.ERC20_FUNCTION)): | ||
ret_calls.append((call,node)) | ||
return ret_calls | ||
|
||
# Get all assignment operations from the function | ||
@staticmethod | ||
def _get_all_assignment_for_variables(func:FunctionContract): | ||
variable_assignment=[] | ||
for node in func.nodes: | ||
if isinstance(node.expression,AssignmentOperation): | ||
variable_assignment=node.variables_written | ||
if hasattr(node,"calls_as_expression") and len(node.calls_as_expression) > 0: | ||
pass | ||
|
||
|
||
|
||
# Get all child calls related to erc20 balance and getreserve from the node | ||
@staticmethod | ||
def _get_calls_and_var_recursively_node(node: NodeType): | ||
# Child calls | ||
ret_calls=[] | ||
# Variables related to balance | ||
ret_vars=[] | ||
variable_writtens=[] | ||
if isinstance(node.expression,AssignmentOperation): | ||
variable_writtens=node.variables_written # Save this variable if there is variable writing | ||
# If it is used to calculate token difference before and after, do not consider this case, return directly | ||
for var in variable_writtens: | ||
if var is None: | ||
continue | ||
if "before" in str(var.name).lower() or "after" in str(var.name).lower(): | ||
return [],[] | ||
# If the node writes variables using call, output all calls involved in this node, including erc20 and others | ||
if hasattr(node,"calls_as_expression") and len(node.calls_as_expression) > 0: | ||
for call in node.calls_as_expression: | ||
if PriceManipulation._check_call_can_output(call): | ||
if call.called.value.full_name in PriceManipulation.ERC20_FUNCTION: | ||
# Do not consider balanceOf(address(this)) | ||
if not (len(call.arguments)==1 and str(call.arguments[0])=="address(this)"): | ||
ret_calls.append(call) | ||
else: | ||
ret_calls.extend(PriceManipulation._get_calls_recursively(call.called.value)) | ||
if call.called and hasattr(call.called,"member_name") and call.called.member_name in PriceManipulation.ERC20_FUNCTION: | ||
# Do not consider balanceOf(address(this)) | ||
if not (len(call.arguments)==1 and str(call.arguments[0])=="address(this)"): | ||
ret_calls.append(call) | ||
return variable_writtens,ret_calls | ||
|
||
@staticmethod | ||
def _check_call_can_output(call): | ||
return isinstance(call, CallExpression) and \ | ||
call.called and hasattr(call.called, 'value') and \ | ||
isinstance(call.called.value, FunctionContract) and \ | ||
not isinstance(call.called.value,Modifier) and \ | ||
not isinstance(call.called.value, Event) | ||
|
||
|
||
def _check_contract_if_uniswap_fork(self,contract:Contract): | ||
if set(PriceManipulationTools.UNISWAP_PAIR_FUNCTION).issubset(set(contract.functions_declared)) or set(PriceManipulationTools.UNISWAP_ROUTER_FUNCTION).issubset(set(contract.functions_declared)): | ||
return True | ||
return False | ||
|
||
|
||
|
||
|
||
def _detect(self) -> List[Output]: | ||
"""""" | ||
results: List[Output] = [] | ||
result_dependent_data=[] | ||
result_call_data=[] | ||
info=[] | ||
for c in self.contracts: | ||
if c.name in PriceManipulationTools.SAFECONTRACTS: | ||
continue | ||
if c.is_interface: | ||
continue | ||
if self._check_contract_if_uniswap_fork(c): | ||
continue | ||
if any(router_name in c.name for router_name in ["Router","router"]): | ||
continue | ||
result_dependent_data,result_call_data=self.checkIfHavePriceManipulation(c) | ||
exist_node=[] | ||
if len(result_dependent_data)>0 or len(result_call_data)>0: | ||
info = ["Potential price manipulation risk:\n"] | ||
# print("risk variable in",function.name,":",risk_var.canonical_name,"rely on",dangerous_erc20_var.canonical_name,"with call:",dangerous_erc20_call) | ||
# data[4] is the node that will actually have a problem, deduplicate according to data[4] | ||
for data in result_dependent_data: | ||
if data[4] not in exist_node and not any(isinstance(ir,EventCall) for ir in data[4].irs): | ||
info += ["\t- In function ",str(data[0]),"\n", | ||
"\t\t-- ",data[4]," have potential price manipulated risk from ",str(data[2])," and call ",str(data[3])," which could influence variable:",str(data[1]),"\n" | ||
] | ||
exist_node.append(data[4]) | ||
|
||
# print("return call in",function.name,":",str(call[0]),"in return is dangerous") | ||
# Deduplicate according to call[2] | ||
for call in result_call_data: | ||
if call[2] not in exist_node and not any(isinstance(ir,EventCall) for ir in call[2].irs): | ||
info += ["\t- In function ",str(call[0]),"\n", | ||
"\t\t-- ",call[2],"have potential price manipulated risk in return call ",str(call[1])," could influence return value\n" | ||
] | ||
exist_node.append(call[2]) | ||
res=self.generate_result(info) | ||
results.append(res) | ||
return results | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Debug
print
statements should be removed or replaced with a logging mechanism suitable for the project's logging strategy.- print("found function")
Committable suggestion