<filename>ilf/fuzzers/ga/mutate_method_call.py
import random
from typing import List

import numpy as np

from .chromosome import MethodCall
from .obs_ga import ObsGA
from ..random import PolicyRandom
from ...ethereum.solidity.abi import SolType


def bytes_to_np_bin(bytes_val):
    size = len(bytes_val) * 8
    chrom = np.zeros((size,), dtype=np.byte)
    for i in range(size):
        byte = bytes_val[i >> 3]
        byte_idx = (size - 1 - i) % 8
        chrom[i] = (byte & (1 << byte_idx)) >> byte_idx
    return chrom


def np_bin_to_bytes(np_bin_val):
    bytes_size = len(np_bin_val) // 8
    bytes_val = np.zeros((bytes_size,), dtype=np.byte)
    for i in range(bytes_size):
        byte_val = 0
        for j in range(8):
            byte_val += np_bin_val[i * 8 + j] << (7 - j)
        bytes_val[i] = byte_val
    return bytearray(bytes_val.tobytes())


class MutateMethodCall:

    def __init__(self, policy_random: PolicyRandom, prob_mutate: float) -> None:
        self.policy_random = policy_random
        self.mutate_func = None
        self.prob_mutate = prob_mutate

    def set_mutate_func(self, func):
        self.mutate_func = func

    def mutate_method_call(self, method_call: MethodCall, obs: ObsGA):
        contract = method_call.chromosome.contract
        method = method_call.method
        sender = method_call.sender
        # mutate amount and timestamp
        # if method is payable, then mutate amount.
        if method_call.chromosome.contract.is_payable(method_call.method.name):
            amount_bytes = bytearray(method_call.amount.to_bytes(13, 'big'))
            self.mutate_func(self, amount_bytes)
            method_call.amount = int.from_bytes(
                amount_bytes, 'big') % self.policy_random.account_manager[sender].amount
        method_call.timestamp = self.policy_random.select_timestamp(obs)
        # mutate arguments
        origin_arguments = method_call.arguments
        new_arguments = []
        for arg, origin_argument in zip(method_call.method.inputs, origin_arguments):
            t = arg.evm_type.t
            if t == SolType.IntTy:
                new_arguments.append(self._mutate_int(
                    arg.evm_type.size, origin_argument))
            elif t == SolType.UintTy:
                new_arguments.append(self._mutate_uint(
                    arg.evm_type.size, origin_argument))
            elif t == SolType.BoolTy:
                new_arguments.append(self._mutate_bool(origin_argument))
            elif t == SolType.StringTy:
                new_arguments.append(self._mutate_string(origin_argument))
            elif t == SolType.SliceTy:
                new_arguments.append(self._mutate_slice(contract, method,
                                                        sender, arg.evm_type.elem, origin_argument, obs))
            elif t == SolType.ArrayTy:
                new_arguments.append(self._mutate_array(contract, method,
                                                        sender, arg.evm_type.elem, origin_argument, obs))
            elif t == SolType.AddressTy:
                new_arguments.append(
                    self._mutate_address(sender, origin_argument))
            elif t == SolType.FixedBytesTy:
                new_arguments.append(self._mutate_fixed_bytes(origin_argument))
            elif t == SolType.BytesTy:
                new_arguments.append(self._mutate_bytes(origin_argument, obs))
            else:
                assert False, 'type {} not supported'.format(t)
        method_call.arguments = new_arguments

    def _mutate_int(self, size, origin_int):
        chrom = bytearray(origin_int.to_bytes(size // 8, 'big', signed=True))
        self.mutate_func(self, chrom)
        return int.from_bytes(chrom, 'big', signed=True)

    def _mutate_uint(self, size, origin_uint: int):
        chrom = bytearray(origin_uint.to_bytes(size // 8, 'big', signed=False))
        self.mutate_func(self, chrom)
        return int.from_bytes(chrom, 'big', signed=False)

    def _mutate_string(self, origin_string):
        origin_string_bytes = bytearray(origin_string.encode('ascii'))
        origin_size = len(origin_string)
        size = random.randint(0, 40)
        if size <= origin_size:
            origin_string_bytes = origin_string_bytes[:size]
        else:
            for _ in range(size - origin_size):
                origin_string_bytes.append(random.randint(1, 127))
        self.mutate_func(self, origin_string_bytes)
        for i in range(size):
            origin_string_bytes[i] = 1 if origin_string_bytes[i] == 0 else origin_string_bytes[i] & 127
        return origin_string_bytes.decode('ascii')

    @staticmethod
    def _mutate_bool(origin_bool):
        return not origin_bool

    def _mutate_slice(self, contract, method, sender, typ, origin_slice: List, obs: ObsGA):
        origin_size = len(origin_slice)
        size = random.randint(1, 15)
        if size <= origin_size:
            origin_slice = origin_slice[:size]
        else:
            origin_slice.extend(
                self.policy_random.select_array(contract, method, sender, size - origin_size, typ, obs))
        return self._mutate_array(contract, method, sender, typ, origin_slice, obs)

    def _mutate_array(self, contract, method, sender, typ, origin_array, obs: ObsGA):
        t = typ.t
        for i in range(len(origin_array)):
            if t == SolType.IntTy:
                origin_array[i] = self._mutate_int(typ.size, origin_array[i])
            elif t == SolType.UintTy:
                origin_array[i] = self._mutate_uint(typ.size, origin_array[i])
            elif t == SolType.BoolTy:
                origin_array[i] = self._mutate_bool(origin_array[i])
            elif t == SolType.StringTy:
                origin_array[i] = self._mutate_string(origin_array[i])
            elif t == SolType.SliceTy:
                origin_array[i] = self._mutate_slice(
                    contract, method, sender, typ.elem, origin_array[i], obs)
            elif t == SolType.ArrayTy:
                origin_array[i] = self._mutate_array(
                    contract, method, sender, typ.elem, origin_array[i], obs)
            elif t == SolType.AddressTy:
                origin_array[i] = self._mutate_address(sender, origin_array[i])
            elif t == SolType.FixedBytesTy:
                origin_array[i] = self._mutate_fixed_bytes(origin_array[i])
            elif t == SolType.BytesTy:
                origin_array[i] = self._mutate_bytes(origin_array[i], obs)
            else:
                assert False, 'type not supported'
        return origin_array

    def _mutate_address(self, sender, origin_address):
        chosen_addresses = [
            address for address in self.policy_random.addresses if address != origin_address]
        if sender not in self.policy_random.account_manager.attacker_indices:
            chosen_addresses = [address for address in chosen_addresses if
                                address not in self.policy_random.account_manager.attacker_addresses]
        return random.choice(chosen_addresses)

    def _mutate_fixed_bytes(self, origin_fixed_bytes):
        chrom = bytearray(origin_fixed_bytes)
        self.mutate_func(self, chrom)
        return [x for x in chrom]

    def _mutate_bytes(self, origin_bytes: List, obs: ObsGA):
        origin_size = len(origin_bytes)
        size = random.randint(1, 15)
        if size <= origin_size:
            origin_bytes = origin_bytes[:size]
        else:
            origin_bytes.extend(
                self.policy_random.select_fixed_bytes(size - origin_size, obs))
        return self._mutate_fixed_bytes(origin_bytes)

<filename>ilf/fuzzers/ga/mutate_method_call.py
import random
from typing import List

import numpy as np

from .chromosome import MethodCall
from .obs_ga import ObsGA
from ..random import PolicyRandom
from ...ethereum.solidity.abi import SolType


def bytes_to_np_bin(bytes_val):
    size = len(bytes_val) * 8
    chrom = np.zeros((size,), dtype=np.byte)
    for i in range(size):
        byte = bytes_val[i >> 3]
        byte_idx = (size - 1 - i) % 8
        chrom[i] = (byte & (1 << byte_idx)) >> byte_idx
    return chrom


def np_bin_to_bytes(np_bin_val):
    bytes_size = len(np_bin_val) // 8
    bytes_val = np.zeros((bytes_size,), dtype=np.byte)
    for i in range(bytes_size):
        byte_val = 0
        for j in range(8):
            byte_val += np_bin_val[i * 8 + j] << (7 - j)
        bytes_val[i] = byte_val
    return bytearray(bytes_val.tobytes())


class MutateMethodCall:

    def __init__(self, policy_random: PolicyRandom, prob_mutate: float) -> None:
        self.policy_random = policy_random
        self.mutate_func = None
        self.prob_mutate = prob_mutate

    def set_mutate_func(self, func):
        self.mutate_func = func

    def mutate_method_call(self, method_call: MethodCall, obs: ObsGA):
        contract = method_call.chromosome.contract
        method = method_call.method
        sender = method_call.sender
        # mutate amount and timestamp
        # if method is payable, then mutate amount.
        if method_call.chromosome.contract.is_payable(method_call.method.name):
            amount_bytes = bytearray(method_call.amount.to_bytes(13, 'big'))
            self.mutate_func(self, amount_bytes)
            method_call.amount = int.from_bytes(
                amount_bytes, 'big') % self.policy_random.account_manager[sender].amount
        method_call.timestamp = self.policy_random.select_timestamp(obs)
        # mutate arguments
        origin_arguments = method_call.arguments
        new_arguments = []
        for arg, origin_argument in zip(method_call.method.inputs, origin_arguments):
            t = arg.evm_type.t
            if t == SolType.IntTy:
                new_arguments.append(self._mutate_int(
                    arg.evm_type.size, origin_argument))
            elif t == SolType.UintTy:
                new_arguments.append(self._mutate_uint(
                    arg.evm_type.size, origin_argument))
            elif t == SolType.BoolTy:
                new_arguments.append(self._mutate_bool(origin_argument))
            elif t == SolType.StringTy:
                new_arguments.append(self._mutate_string(origin_argument))
            elif t == SolType.SliceTy:
                new_arguments.append(self._mutate_slice(contract, method,
                                                        sender, arg.evm_type.elem, origin_argument, obs))
            elif t == SolType.ArrayTy:
                new_arguments.append(self._mutate_array(contract, method,
                                                        sender, arg.evm_type.elem, origin_argument, obs))
            elif t == SolType.AddressTy:
                new_arguments.append(
                    self._mutate_address(sender, origin_argument))
            elif t == SolType.FixedBytesTy:
                new_arguments.append(self._mutate_fixed_bytes(origin_argument))
            elif t == SolType.BytesTy:
                new_arguments.append(self._mutate_bytes(origin_argument, obs))
            else:
                assert False, 'type {} not supported'.format(t)
        method_call.arguments = new_arguments

    def _mutate_int(self, size, origin_int):
        chrom = bytearray(origin_int.to_bytes(size // 8, 'big', signed=True))
        self.mutate_func(self, chrom)
        return int.from_bytes(chrom, 'big', signed=True)

    def _mutate_uint(self, size, origin_uint: int):
        chrom = bytearray(origin_uint.to_bytes(size // 8, 'big', signed=False))
        self.mutate_func(self, chrom)
        return int.from_bytes(chrom, 'big', signed=False)

    def _mutate_string(self, origin_string):
        origin_string_bytes = bytearray(origin_string.encode('ascii'))
        origin_size = len(origin_string)
        size = random.randint(0, 40)
        if size <= origin_size:
            origin_string_bytes = origin_string_bytes[:size]
        else:
            for _ in range(size - origin_size):
                origin_string_bytes.append(random.randint(1, 127))
        self.mutate_func(self, origin_string_bytes)
        for i in range(size):
            origin_string_bytes[i] = 1 if origin_string_bytes[i] == 0 else origin_string_bytes[i] & 127
        return origin_string_bytes.decode('ascii')

    @staticmethod
    def _mutate_bool(origin_bool):
        return not origin_bool

    def _mutate_slice(self, contract, method, sender, typ, origin_slice: List, obs: ObsGA):
        origin_size = len(origin_slice)
        size = random.randint(1, 15)
        if size <= origin_size:
            origin_slice = origin_slice[:size]
        else:
            origin_slice.extend(
                self.policy_random.select_array(contract, method, sender, size - origin_size, typ, obs))
        return self._mutate_array(contract, method, sender, typ, origin_slice, obs)

    def _mutate_array(self, contract, method, sender, typ, origin_array, obs: ObsGA):
        t = typ.t
        for i in range(len(origin_array)):
            if t == SolType.IntTy:
                origin_array[i] = self._mutate_int(typ.size, origin_array[i])
            elif t == SolType.UintTy:
                origin_array[i] = self._mutate_uint(typ.size, origin_array[i])
            elif t == SolType.BoolTy:
                origin_array[i] = self._mutate_bool(origin_array[i])
            elif t == SolType.StringTy:
                origin_array[i] = self._mutate_string(origin_array[i])
            elif t == SolType.SliceTy:
                origin_array[i] = self._mutate_slice(
                    contract, method, sender, typ.elem, origin_array[i], obs)
            elif t == SolType.ArrayTy:
                origin_array[i] = self._mutate_array(
                    contract, method, sender, typ.elem, origin_array[i], obs)
            elif t == SolType.AddressTy:
                origin_array[i] = self._mutate_address(sender, origin_array[i])
            elif t == SolType.FixedBytesTy:
                origin_array[i] = self._mutate_fixed_bytes(origin_array[i])
            elif t == SolType.BytesTy:
                origin_array[i] = self._mutate_bytes(origin_array[i], obs)
            else:
                assert False, 'type not supported'
        return origin_array

    def _mutate_address(self, sender, origin_address):
        chosen_addresses = [
            address for address in self.policy_random.addresses if address != origin_address]
        if sender not in self.policy_random.account_manager.attacker_indices:
            chosen_addresses = [address for address in chosen_addresses if
                                address not in self.policy_random.account_manager.attacker_addresses]
        return random.choice(chosen_addresses)

    def _mutate_fixed_bytes(self, origin_fixed_bytes):
        chrom = bytearray(origin_fixed_bytes)
        self.mutate_func(self, chrom)
        return [x for x in chrom]

    def _mutate_bytes(self, origin_bytes: List, obs: ObsGA):
        origin_size = len(origin_bytes)
        size = random.randint(1, 15)
        if size <= origin_size:
            origin_bytes = origin_bytes[:size]
        else:
            origin_bytes.extend(
                self.policy_random.select_fixed_bytes(size - origin_size, obs))
        return self._mutate_fixed_bytes(origin_bytes)

# mutate amount and timestamp # if method is payable, then mutate amount. # mutate arguments