IS-1120: prevalidate a tx more strictly 1

iconservice/base/type_converter_templates.py

@@ -107,6 +107,7 @@ class ConstantKeys:
     FROM = "from"
     TO = "to"
     VALUE = "value"
+    NID = "nid"
     STEP_LIMIT = "stepLimit"
     FEE = "fee"
     NONCE = "nonce"

iconservice/iconscore/icon_pre_validator.py

@@ -19,9 +19,10 @@
 from iconcommons.logger import Logger
 
 from .icon_score_step import get_input_data_size
-from ..base.address import Address, SYSTEM_SCORE_ADDRESS, generate_score_address
+from ..base.address import Address, SYSTEM_SCORE_ADDRESS, generate_score_address, is_icon_address_valid
 from ..base.exception import InvalidRequestException, InvalidParamsException, OutOfBalanceException
 from ..icon_constant import FIXED_FEE, MAX_DATA_SIZE, DEFAULT_BYTE_SIZE, DATA_BYTE_ORDER, Revision, DeployState
+from ..base.type_converter_templates import ConstantKeys
 from ..utils import is_lowercase_hex_string
 from ..utils.locked import is_address_locked
 
@@ -31,6 +32,23 @@
 
 
 TAG = "PV"
+REQUEST_PARAMS = (
+    ConstantKeys.VERSION,
+    ConstantKeys.STEP_LIMIT,
+    ConstantKeys.NID,
+    ConstantKeys.TIMESTAMP,
+    ConstantKeys.VALUE,
+    ConstantKeys.NONCE,
+    ConstantKeys.FROM,
+    ConstantKeys.TO,
+    ConstantKeys.SIGNATURE,
+    ConstantKeys.DATA_TYPE,
+    ConstantKeys.DATA,
+)
+
+REQUIRED_PARAMS = REQUEST_PARAMS[:4] + REQUEST_PARAMS[6:9]
+INT_PARAMS = REQUEST_PARAMS[:6]
+ADDR_PARAMS = REQUEST_PARAMS[6:8]
 
 
 class IconPreValidator:
@@ -44,6 +62,63 @@ def __init__(self) -> None:
         """
         pass
 
+    def origin_request_execute(self, request: dict):
+        self.origin_pre_validate_version(request)
+        self.origin_pre_validate_params(request)
+        self.origin_validate_fields(request)
+
+    def origin_validate_fields(self, params: dict):
+        for param, value in params.items():
+            self.origin_validate_param(param)
+            self.origin_validate_value(param, value)
+
+    @classmethod
+    def origin_pre_validate_version(cls, params: dict):
+        version: str = params.get(ConstantKeys.VERSION, None)
+        if version != '0x3':
+            raise InvalidRequestException(f'Invalid message version, got {version}')
+
+    @classmethod
+    def origin_pre_validate_params(cls, params: dict):
+        if len(params) > len(REQUEST_PARAMS):
+            raise InvalidRequestException('Unexpected Parameters')
+
+        required_results = [
+            required_key
+            for required_key
+            in REQUIRED_PARAMS
+            if required_key not in params
+        ]
+
+        if required_results:
+            raise InvalidRequestException(
+                f'Not included required parameters, missing parameters {required_results}'
+            )
+
+    @classmethod
+    def origin_validate_param(cls, param: str):
+        if param not in REQUEST_PARAMS:
+            raise InvalidParamsException(f'Unexpected Parameters, got {param}')
+
+    @classmethod
+    def origin_validate_value(cls, param: str, value: str):
+        if param in INT_PARAMS:
+            if not cls.is_integer_type(value):
+                raise InvalidRequestException(f'Unexpected INT Type, got {value}')
+        elif param in ADDR_PARAMS:
+            if not cls.is_address_type(value):
+                raise InvalidRequestException(f'Unexpected Address Type, got {value}')
+
+    @classmethod
+    def is_integer_type(cls, value: str) -> bool:
+        if value.startswith('0x'):
+            return value == hex(int(value, 16))
+        return False
+
+    @classmethod
+    def is_address_type(cls, value: str) -> bool:
+        return is_icon_address_valid(value)
+
     def execute(self, context: 'IconScoreContext', params: dict, step_price: int, minimum_step: int):
         """Validate a transaction on icx_sendTransaction
         If failed to validate a tx, raise an exception

tests/unit_test/iconscore/test_icon_origin_pre_validator.py

@@ -0,0 +1,185 @@
+import pytest
+import os
+import random
+
+from iconservice.iconscore.icon_pre_validator import IconPreValidator
+from iconservice.base.address import (
+    Address,
+    AddressPrefix
+)
+from iconservice.base.exception import (
+    InvalidRequestException,
+    InvalidParamsException
+)
+
+
+def make_required_parameters(**kwargs) -> dict:
+    return {
+        'version': kwargs['version'],
+        'from': kwargs['_from'],
+        'to': kwargs['to'],
+        'stepLimit': kwargs['stepLimit'],
+        'nid': '0x1',
+        'timestamp': kwargs['timestamp'],
+        'signature': kwargs['signature'],
+    }
+
+
+def make_origin_parameters(option: dict = None) -> dict:
+    params = make_required_parameters(
+        version=hex(3),
+        _from=str(Address.from_data(AddressPrefix.EOA, os.urandom(20))),
+        to=str(Address.from_data(random.choice([AddressPrefix.EOA, AddressPrefix.CONTRACT]), os.urandom(20))),
+        stepLimit=hex(random.randint(10, 5000)),
+        timestamp=hex(random.randint(10, 5000)),
+        signature='VAia7YZ2Ji6igKWzjR2YsGa2m53nKPrfK7uXYW78QLE+ATehAVZPC40szvAiA6NEU5gCYB4c4qaQzqDh2ugcHgA='
+    )
+
+    if option:
+        params.update(option)
+    return params
+
+
+@pytest.fixture
+def validator():
+    validator = IconPreValidator()
+    return validator
+
+
+@pytest.mark.parametrize(
+    'malformed_value, expected',
+    [
+        ('', False),
+        ('0x0001', False),
+        ('0x0000001', False),
+        ('100', False),
+        ('0x1e', True)
+    ]
+)
+def test_malformed_number_string(validator, malformed_value, expected):
+    assert validator.is_integer_type(malformed_value) == expected
+
+
+@pytest.mark.parametrize(
+    'malformed_value, expected',
+    [
+        ('', False),
+        ('0x1A', False),
+        ('0x3E', False),
+        ('0xBE', False),
+        ('0Xff', False),
+        ('0xff', True)
+    ]
+)
+def test_malformed_uppercase_string(validator, malformed_value, expected):
+    assert validator.is_integer_type(malformed_value) == expected
+
+
+@pytest.mark.parametrize(
+    'malformed_value, expected',
+    [
+        ('', False),
+        ('0xqw1234', False),
+        ('hxEfe3', False),
+        ('hx1234', False),
+        ('cx1234', False),
+        ('cxEab1', False),
+        ('1234', False),
+        ('hx3f945d146a87552487ad70a050eebfa2564e8e5c', True)
+    ]
+)
+def test_malformed_address(validator, malformed_value, expected):
+    assert validator.is_address_type(malformed_value) == expected
+
+
+@pytest.mark.parametrize(
+    'msg',
+    [
+        {},
+        make_origin_parameters({'version': '0x1'}),
+        make_origin_parameters({'version': '0x2'}),
+    ]
+)
+def test_pre_validate_version(validator, msg):
+    with pytest.raises(InvalidRequestException):
+        validator.origin_pre_validate_version(msg)
+
+
+@pytest.mark.parametrize(
+    'msg',
+    [
+        {},
+        {'version': '0x3'},
+        {'version': '0x3', 'from': 'temp', 'to': 'temp', 'stepLimit': 'temp', 'timestamp': 'test'},
+        make_origin_parameters({
+            'value': '',
+            'nonce': '',
+            'data_type': '',
+            'data': '',
+            'item': '',
+            'test': '',
+            'failed': '',
+            'failure': ''
+        })
+    ]
+)
+def test_pre_validate_params(validator, msg):
+    with pytest.raises(InvalidRequestException):
+        validator.origin_pre_validate_params(msg)
+
+
+@pytest.mark.parametrize(
+    'param',
+    [
+        '',
+        None,
+        'foo',
+        'bar'
+    ]
+)
+def test_validate_param(validator, param):
+    with pytest.raises(InvalidParamsException):
+        validator.origin_validate_param(param)
+
+
+@pytest.mark.parametrize(
+    'param, value',
+    [
+        ('stepLimit', ''),
+        ('stepLimit', 'Fx35'),
+        ('to', '0xFF'),
+        ('value', '0X1e'),
+        ('to', 'cX3F9e')
+    ]
+)
+def test_validate_value(validator, param, value):
+    with pytest.raises(InvalidRequestException):
+        validator.origin_validate_value(param, value)
+
+
+@pytest.mark.parametrize(
+    'msg',
+    [
+        make_origin_parameters({'test': 'foo'}),
+        make_origin_parameters({'to': 'hxEE'}),
+        make_origin_parameters({'stepLimit': '0xEf'}),
+    ]
+)
+def test_validate_fields(validator, msg):
+    with pytest.raises((InvalidRequestException, InvalidParamsException)):
+        validator.origin_validate_fields(msg)
+
+
+@pytest.mark.parametrize(
+    'msg',
+    [
+        {},
+        make_origin_parameters({'version': '0x1'}),
+        make_origin_parameters({'to': 'hxFF'}),
+        make_origin_parameters({'value': '0x001'})
+    ]
+)
+def test_validate_request_execute(validator, msg):
+    with pytest.raises((InvalidParamsException, InvalidRequestException)):
+        validator.origin_request_execute(msg)
+