Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
tx2 = self._stake(self._addr_array[23], stake_amount)
tx3 = self._stake(self._addr_array[24], stake_amount)
prev_block, tx_results, main_prep_list = self._make_and_req_block_for_prep_test([tx1, tx2, tx3])
self.assertIsNone(main_prep_list)
self._write_precommit_state(prev_block)
self._set_revision(REV_DECENTRALIZATION)
# register preps
reg_prep_tx_list = []
for i, address in enumerate(self._main_preps):
data: dict = {
ConstantKeys.NAME: "name",
ConstantKeys.EMAIL: "email",
ConstantKeys.WEBSITE: "website",
ConstantKeys.DETAILS: "json",
ConstantKeys.P2P_END_POINT: "ip",
ConstantKeys.PUBLIC_KEY: f'publicKey{i}'.encode(),
ConstantKeys.IREP: IISS_MIN_IREP
}
reg_prep_tx_list.append(self._reg_prep(address, data))
prev_block, tx_results, main_prep_list = self._make_and_req_block_for_prep_test(reg_prep_tx_list)
self.assertIsNone(main_prep_list)
self._write_precommit_state(prev_block)
# delegate
data: list = [
{
"address": str(address),
"value": hex(_DELEGATE_AMOUNT)
}for address in self._addr_array[:10]
invalid_website_list = [
"",
"invalid website",
"invalid.com",
"invalid_.com",
"c.com",
"http://c.com",
"https://c.com",
"ftp://caaa.com",
"http://valid.",
"https://valid." "https://target.asdf:7100",
]
for website in invalid_website_list:
reg_data: dict = deepcopy(prep_register_data)
reg_data[ConstantKeys.P2P_ENDPOINT] = website
tx = self.create_register_prep_tx(self._accounts[6], reg_data)
self.process_confirm_block_tx([tx], expected_status=False)
validate_endpoint = "20.20.7.8:8000"
reg_data: dict = deepcopy(prep_register_data)
reg_data[ConstantKeys.P2P_ENDPOINT] = validate_endpoint
tx = self.create_register_prep_tx(self._accounts[6], reg_data)
self.process_confirm_block_tx([tx])
def _reg_prep(self, address: 'Address', data: dict, _revision: int = REV_IISS):
data = deepcopy(data)
data[ConstantKeys.PUBLIC_KEY] = f"0x{data[ConstantKeys.PUBLIC_KEY].hex()}"
tx = self._make_score_call_tx(address, ZERO_SCORE_ADDRESS, 'registerPRep', data)
tx_list = [tx]
prev_block, tx_results = self._make_and_req_block(tx_list)
self.assertEqual(int(True), tx_results[0].status)
self.assertEqual('PRepRegistered(Address)', tx_results[0].event_logs[0].indexed[0])
self.assertEqual(address, tx_results[0].event_logs[0].data[0])
self._write_precommit_state(prev_block)
def _reg_prep(self, address: 'Address', data: dict):
data = deepcopy(data)
data[ConstantKeys.PUBLIC_KEY] = data[ConstantKeys.PUBLIC_KEY].hex()
tx = self._make_score_call_tx(address,
ZERO_SCORE_ADDRESS,
'registerPRep',
data)
prev_block, tx_results = self._make_and_req_block([tx])
self.assertEqual(int(True), tx_results[0].status)
self._write_precommit_state(prev_block)
def _validate_name(self):
reg_data: dict = deepcopy(prep_register_data)
reg_data[ConstantKeys.NAME] = ""
tx = self.create_register_prep_tx(self._accounts[0], reg_data)
self.process_confirm_block_tx([tx], expected_status=False)
reg_data[ConstantKeys.NAME] = "valid name"
tx = self.create_register_prep_tx(self._accounts[0], reg_data)
self.process_confirm_block_tx([tx])
ConstantKeys.VERSION: ValueType.INT,
ConstantKeys.TX_HASH: ValueType.BYTES,
ConstantKeys.FROM: ValueType.ADDRESS,
ConstantKeys.TO: ValueType.ADDRESS_OR_MALFORMED_ADDRESS,
ConstantKeys.VALUE: ValueType.HEXADECIMAL,
ConstantKeys.STEP_LIMIT: ValueType.INT,
ConstantKeys.FEE: ValueType.HEXADECIMAL,
ConstantKeys.TIMESTAMP: ValueType.INT,
ConstantKeys.NONCE: ValueType.INT,
ConstantKeys.SIGNATURE: ValueType.IGNORE,
ConstantKeys.DATA_TYPE: ValueType.STRING,
ConstantKeys.DATA: {
CONVERT_USING_SWITCH_KEY: {
SWITCH_KEY: ConstantKeys.DATA_TYPE,
ConstantKeys.CALL: type_convert_templates[ParamType.CALL_DATA],
ConstantKeys.DEPLOY: type_convert_templates[ParamType.DEPLOY_DATA],
ConstantKeys.BASE: type_convert_templates[ParamType.BASE_DATA]
}
},
KEY_CONVERTER: {
ConstantKeys.OLD_TX_HASH: ConstantKeys.TX_HASH
}
}
type_convert_templates[ParamType.INVOKE_TRANSACTION] = {
ConstantKeys.METHOD: ValueType.STRING,
ConstantKeys.PARAMS: type_convert_templates[ParamType.TRANSACTION_PARAMS_DATA],
ConstantKeys.GENESIS_DATA: {
ConstantKeys.ACCOUNTS: [
type_convert_templates[ParamType.ACCOUNT_DATA]
],
ConstantKeys.MESSAGE: ValueType.STRING
"stake": self._stake,
"delegated": self._delegated,
"totalBlocks": self._total_blocks,
"validatedBlocks": self._validated_blocks,
"unvalidatedSequenceBlocks": self._unvalidated_sequence_blocks,
"irep": self._irep,
"irepUpdateBlockHeight": self._irep_block_height,
"lastGenerateBlockHeight": self._last_generate_block_height,
"blockHeight": self._block_height,
"txIndex": self._tx_index,
"nodeAddress": self._node_address,
}
if dict_type == PRepDictType.FULL:
data[ConstantKeys.EMAIL] = self.email
data[ConstantKeys.WEBSITE] = self.website
data[ConstantKeys.DETAILS] = self.details
data[ConstantKeys.P2P_ENDPOINT] = self.p2p_endpoint
return data
ConstantKeys.PREV_BLOCK_HASH: ValueType.BYTES,
}
type_convert_templates[ParamType.ACCOUNT_DATA] = {
ConstantKeys.NAME: ValueType.STRING,
ConstantKeys.ADDRESS: ValueType.ADDRESS,
ConstantKeys.BALANCE: ValueType.INT
}
type_convert_templates[ParamType.CALL_DATA] = {
ConstantKeys.METHOD: ValueType.STRING,
ConstantKeys.PARAMS: ValueType.LATER
}
type_convert_templates[ParamType.DEPLOY_DATA] = {
ConstantKeys.CONTENT_TYPE: ValueType.STRING,
ConstantKeys.CONTENT: ValueType.IGNORE,
ConstantKeys.PARAMS: ValueType.LATER
}
type_convert_templates[ParamType.BASE_DATA] = {
ConstantKeys.PREP: {
ConstantKeys.PREP_INCENTIVE: ValueType.INT,
ConstantKeys.PREP_REWARD_RATE: ValueType.INT,
ConstantKeys.PREP_TOTAL_DELEGATION: ValueType.INT,
ConstantKeys.PREP_VALUE: ValueType.INT
},
ConstantKeys.EEP: {
ConstantKeys.EEP_INCENTIVE: ValueType.INT,
ConstantKeys.EEP_REWARD_RATE: ValueType.INT,
ConstantKeys.EEP_TOTAL_DELEGATION: ValueType.INT,
ConstantKeys.EEP_VALUE: ValueType.INT
def _rollback(self, request: dict) -> dict:
Logger.info(tag=_TAG, msg=f"ROLLBACK Request: {request}")
try:
converted_params = TypeConverter.convert(request, ParamType.ROLLBACK)
block_height: int = converted_params[ConstantKeys.BLOCK_HEIGHT]
block_hash: bytes = converted_params[ConstantKeys.BLOCK_HASH]
Logger.info(
tag=_TAG,
msg=f"ROLLBACK: BH={block_height} block_hash={bytes_to_hex(block_hash)}",
)
response: dict = self._icon_service_engine.rollback(
block_height, block_hash
)
response = MakeResponse.make_response(response)
except FatalException as e:
self._log_exception(e, _TAG)
response = MakeResponse.make_error_response(
ExceptionCode.SYSTEM_ERROR, str(e)
)
self._close()
except IconServiceBaseException as icon_e:
elif key == ConstantKeys.EMAIL:
_validate_email(context.revision, value)
elif key == ConstantKeys.COUNTRY:
_validate_country(value)
# node address validate
is_update_node_address: bool = ConstantKeys.NODE_ADDRESS in tx_data and tx_data[
ConstantKeys.NODE_ADDRESS
] is not None
if set_prep and not is_update_node_address:
# non changed skip
return
# register_prep or is_update_node_address is True
node_address: "Address" = tx_data[
ConstantKeys.NODE_ADDRESS
] if is_update_node_address else prep_address
# must prevent to change node address before divide node address revision.
if context.revision < Revision.DIVIDE_NODE_ADDRESS.value:
if prep_address != node_address:
raise InvalidParamsException(
f"nodeAddress not supported: revision={context.revision}"
)
context.prep_address_converter.validate_node_address(node_address)