Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Sets the RPC mode to either of ganache or infura
:param rpc: either of the strings - ganache, infura-mainnet, infura-rinkeby, infura-kovan, infura-ropsten
"""
if rpc == "ganache":
rpcconfig = ("localhost", 7545, False)
else:
m = re.match(r"infura-(.*)", rpc)
if m and m.group(1) in ["mainnet", "rinkeby", "kovan", "ropsten"]:
rpcconfig = (m.group(1) + ".infura.io", 443, True)
else:
try:
host, port = rpc.split(":")
rpcconfig = (host, int(port), rpctls)
except ValueError:
raise CriticalError(
"Invalid RPC argument, use 'ganache', 'infura-[network]' or 'HOST:PORT'"
)
if rpcconfig:
log.info("Using RPC settings: %s" % str(rpcconfig))
self.eth = EthJsonRpc(rpcconfig[0], int(rpcconfig[1]), rpcconfig[2])
else:
raise CriticalError("Invalid RPC settings, check help for details.")
if m and m.group(1) in ["mainnet", "rinkeby", "kovan", "ropsten"]:
rpcconfig = (m.group(1) + ".infura.io", 443, True)
else:
try:
host, port = rpc.split(":")
rpcconfig = (host, int(port), rpctls)
except ValueError:
raise CriticalError(
"Invalid RPC argument, use 'ganache', 'infura-[network]' or 'HOST:PORT'"
)
if rpcconfig:
log.info("Using RPC settings: %s" % str(rpcconfig))
self.eth = EthJsonRpc(rpcconfig[0], int(rpcconfig[1]), rpcconfig[2])
else:
raise CriticalError("Invalid RPC settings, check help for details.")
solc_settings_json=self.solc_settings_json,
solc_binary=self.solc_binary,
)
self.contracts.append(contract)
contracts.append(contract)
else:
for contract in get_contracts_from_file(
input_file=file,
solc_settings_json=self.solc_settings_json,
solc_binary=self.solc_binary,
):
self.contracts.append(contract)
contracts.append(contract)
except FileNotFoundError:
raise CriticalError("Input file not found: " + file)
except CompilerError as e:
error_msg = str(e)
# Check if error is related to solidity version mismatch
if (
"Error: Source file requires different compiler version"
in error_msg
):
# Grab relevant line "pragma solidity ...", excluding any comments
solv_pragma_line = error_msg.split("\n")[-3].split("//")[0]
# Grab solidity version from relevant line
solv_match = re.findall(r"[0-9]+\.[0-9]+\.[0-9]+", solv_pragma_line)
error_suggestion = (
"" if len(solv_match) != 1 else solv_match[0]
)
error_msg = (
error_msg
solc_args=self.solc_args,
solc_binary=self.solc_binary,
)
self.contracts.append(contract)
contracts.append(contract)
else:
for contract in get_contracts_from_file(
input_file=file,
solc_args=self.solc_args,
solc_binary=self.solc_binary,
):
self.contracts.append(contract)
contracts.append(contract)
except FileNotFoundError:
raise CriticalError("Input file not found: " + file)
except CompilerError as e:
raise CriticalError(e)
except NoContractFoundError:
logging.error(
"The file " + file + " does not contain a compilable contract."
)
return address, contracts
"""
def search_callback(_, address, balance):
"""
:param _:
:param address: The address of the contract with the code in search
:param balance: The balance of the corresponding contract
"""
print("Address: " + address + ", balance: " + str(balance))
try:
self.leveldb.search(search, search_callback)
except SyntaxError:
raise CriticalError("Syntax error in search expression.")
def _init_solc_binary(version: str) -> str:
"""
Only proper versions are supported. No nightlies, commits etc (such as available in remix).
:param version: Version of the solc binary required
:return: The solc binary of the corresponding version
"""
if not version:
return os.environ.get("SOLC") or "solc"
# tried converting input to semver, seemed not necessary so just slicing for now
main_version = solc.main.get_solc_version_string()
main_version_number = re.match(r"\d+.\d+.\d+", main_version)
if main_version is None:
raise CriticalError(
"Could not extract solc version from string {}".format(main_version)
)
if version == main_version_number:
log.info("Given version matches installed version")
solc_binary = os.environ.get("SOLC") or "solc"
else:
solc_binary = util.solc_exists(version)
if solc_binary and solc_binary != util.solc_exists(
"default_ubuntu_version"
):
log.info("Given version is already installed")
else:
try:
if version.startswith("0.4"):
solc.install_solc("v" + version)
else:
if not re.match(r"0x[a-fA-F0-9]{40}", address):
raise CriticalError("Invalid contract address. Expected format is '0x...'.")
try:
code = self.eth.eth_getCode(address)
except FileNotFoundError as e:
raise CriticalError("IPC error: " + str(e))
except ConnectionError:
raise CriticalError(
"Could not connect to RPC server. Make sure that your node is running and that RPC parameters are set correctly."
)
except Exception as e:
raise CriticalError("IPC / RPC error: " + str(e))
if code == "0x" or code == "0x0":
raise CriticalError(
"Received an empty response from eth_getCode. Check the contract address and verify that you are on the correct chain."
)
else:
self.contracts.append(
EVMContract(
code, name=address, enable_online_lookup=self.enable_online_lookup
)
)
return address, self.contracts[-1] # return address and contract object
key = bytes(params[i], "utf8")
key_formatted = utils.rzpad(key, 32)
mappings.append(
int.from_bytes(
utils.sha3(key_formatted + position_formatted),
byteorder="big",
)
)
length = len(mappings)
if length == 1:
position = mappings[0]
else:
if len(params) >= 4:
raise CriticalError("Invalid number of parameters.")
if len(params) >= 1:
position = int(params[0])
if len(params) >= 2:
length = int(params[1])
if len(params) == 3 and params[2] == "array":
position_formatted = utils.zpad(
utils.int_to_big_endian(position), 32
)
position = int.from_bytes(
utils.sha3(position_formatted), byteorder="big"
)
except ValueError:
raise CriticalError(
"Invalid storage index. Please provide a numeric value."
def contract_hash_to_address(self, contract_hash):
"""
Returns address of the corresponding hash by searching the leveldb
:param contract_hash: Hash to be searched
"""
if not re.match(r"0x[a-fA-F0-9]{64}", contract_hash):
raise CriticalError("Invalid address hash. Expected format is '0x...'.")
print(self.leveldb.contract_hash_to_address(contract_hash))