mirror of
https://github.com/trailofbits/manticore.git
synced 2026-10-12 02:23:44 +00:00
Switch to stable Black (#2536)
* Switch to stable Black version `black` has recently stabilized on version 22, which means that we can loosen the version restrictions a bit to get upstream bugfixes, without needing to worry about our formatting breaking unexpectedly. In 2023, they may issue a new release with breaking changes, but we won't be automatically upgraded. There have been a large number of formatting changes since 20.8.b1, which is the last version we standardized on. This will likely require rebase of several of the existing PRs, but with relatively few of them currently active, I think it's as good a time as any. More info about black's further development in their [stability policy](https://black.readthedocs.io/en/latest/the_black_code_style/index.html#stability-policy) * Add new formatting changes Mostly docstrings and exponentiation
This commit is contained in:
@@ -27,7 +27,7 @@ print("Source code:\n", source_code)
|
||||
|
||||
|
||||
class StopAtDepth(Detector):
|
||||
""" This just aborts explorations that are too deep """
|
||||
"""This just aborts explorations that are too deep"""
|
||||
|
||||
def will_run_callback(self, *args):
|
||||
with self.manticore.locked_context("seen_rep", dict) as reps:
|
||||
|
||||
@@ -155,7 +155,7 @@ def constraints_to_constraintset(constupl):
|
||||
|
||||
|
||||
def input_from_cons(constupl, datas):
|
||||
" solve bytes in |datas| based on "
|
||||
"solve bytes in |datas| based on"
|
||||
|
||||
def make_chr(c):
|
||||
try:
|
||||
|
||||
@@ -3,7 +3,7 @@ from manticore.core.plugin import Plugin
|
||||
|
||||
|
||||
def getchar(state):
|
||||
""" Symbolic `getchar` implementation. Returns an arbitrary single byte """
|
||||
"""Symbolic `getchar` implementation. Returns an arbitrary single byte"""
|
||||
res = state.new_symbolic_value(32, "getchar_res")
|
||||
state.constrain(0 < res)
|
||||
state.constrain(res < 256)
|
||||
@@ -11,7 +11,7 @@ def getchar(state):
|
||||
|
||||
|
||||
class PrintRetPlugin(Plugin):
|
||||
""" A plugin that looks for states that returned zero and solves for their inputs """
|
||||
"""A plugin that looks for states that returned zero and solves for their inputs"""
|
||||
|
||||
def will_terminate_state_callback(self, state, *args):
|
||||
retval = state.stack.peek()
|
||||
|
||||
@@ -202,7 +202,7 @@ def p_expression_or(p):
|
||||
|
||||
|
||||
def p_expression_neg(p):
|
||||
"expression : NEG expression "
|
||||
"expression : NEG expression"
|
||||
p[0] = ~p[1]
|
||||
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ logger = logging.getLogger(__name__)
|
||||
class DecorateAllMeta(type):
|
||||
@staticmethod
|
||||
def _if_enabled(f):
|
||||
""" decorator used to guard callbacks """
|
||||
"""decorator used to guard callbacks"""
|
||||
|
||||
@wraps(f)
|
||||
def g(self, *args, **kwargs):
|
||||
@@ -44,17 +44,17 @@ class Plugin(metaclass=DecorateAllMeta):
|
||||
self._plugin_context_name = f"{classname}_context_{hex(hash(self))}"
|
||||
|
||||
def enable(self):
|
||||
""" Enable all callbacks """
|
||||
"""Enable all callbacks"""
|
||||
with self.manticore.locked_context() as context:
|
||||
context[self._enabled_key] = True
|
||||
|
||||
def disable(self):
|
||||
""" Disable all callbacks """
|
||||
"""Disable all callbacks"""
|
||||
with self.manticore.locked_context() as context:
|
||||
context[self._enabled_key] = False
|
||||
|
||||
def is_enabled(self):
|
||||
""" True if callbacks are enabled """
|
||||
"""True if callbacks are enabled"""
|
||||
with self.manticore.locked_context() as context:
|
||||
return context.get(self._enabled_key, True)
|
||||
|
||||
@@ -85,18 +85,18 @@ class Plugin(metaclass=DecorateAllMeta):
|
||||
|
||||
@property
|
||||
def context(self):
|
||||
""" Convenient access to shared context """
|
||||
"""Convenient access to shared context"""
|
||||
plugin_context_name = self._plugin_context_name
|
||||
if plugin_context_name not in self.manticore.context:
|
||||
self.manticore.context[plugin_context_name] = {}
|
||||
return self.manticore.context[plugin_context_name]
|
||||
|
||||
def on_register(self):
|
||||
""" Called by parent manticore on registration """
|
||||
"""Called by parent manticore on registration"""
|
||||
pass
|
||||
|
||||
def on_unregister(self):
|
||||
""" Called be parent manticore on un-registration """
|
||||
"""Called be parent manticore on un-registration"""
|
||||
pass
|
||||
|
||||
def generate_testcase(self, state, testcase, message):
|
||||
@@ -332,7 +332,7 @@ class Profiler(Plugin):
|
||||
return ps
|
||||
|
||||
def save_profiling_data(self, stream=None):
|
||||
""":param stream: an output stream to write the profiling data """
|
||||
""":param stream: an output stream to write the profiling data"""
|
||||
ps = self.get_profiling_data()
|
||||
# XXX(yan): pstats does not support dumping to a file stream, only to a file
|
||||
# name. Below is essentially the implementation of pstats.dump_stats() without
|
||||
|
||||
@@ -112,7 +112,7 @@ class ConstraintSet:
|
||||
self._constraints.append(constraint)
|
||||
|
||||
def _get_sid(self) -> int:
|
||||
""" Returns a unique id. """
|
||||
"""Returns a unique id."""
|
||||
assert self._child is None
|
||||
self._sid += 1
|
||||
return self._sid
|
||||
@@ -232,23 +232,23 @@ class ConstraintSet:
|
||||
return result
|
||||
|
||||
def _declare(self, var):
|
||||
""" Declare the variable `var` """
|
||||
"""Declare the variable `var`"""
|
||||
if var.name in self._declarations:
|
||||
raise ValueError("Variable already declared")
|
||||
self._declarations[var.name] = var
|
||||
return var
|
||||
|
||||
def get_declared_variables(self):
|
||||
""" Returns the variable expressions of this constraint set """
|
||||
"""Returns the variable expressions of this constraint set"""
|
||||
return self._declarations.values()
|
||||
|
||||
def get_variable(self, name):
|
||||
""" Returns the variable declared under name or None if it does not exists """
|
||||
"""Returns the variable declared under name or None if it does not exists"""
|
||||
return self._declarations.get(name)
|
||||
|
||||
@property
|
||||
def declarations(self):
|
||||
""" Returns the variable expressions of this constraint set """
|
||||
"""Returns the variable expressions of this constraint set"""
|
||||
declarations = GetDeclarations()
|
||||
for a in self.constraints:
|
||||
try:
|
||||
@@ -279,11 +279,11 @@ class ConstraintSet:
|
||||
return iter(self.constraints)
|
||||
|
||||
def __str__(self):
|
||||
""" Returns a smtlib representation of the current state """
|
||||
"""Returns a smtlib representation of the current state"""
|
||||
return self.to_string()
|
||||
|
||||
def _make_unique_name(self, name="VAR"):
|
||||
""" Makes a unique variable name"""
|
||||
"""Makes a unique variable name"""
|
||||
# the while loop is necessary because appending the result of _get_sid()
|
||||
# is not guaranteed to make a unique name on the first try; a colliding
|
||||
# name could have been added previously
|
||||
@@ -292,7 +292,7 @@ class ConstraintSet:
|
||||
return name
|
||||
|
||||
def is_declared(self, expression_var) -> bool:
|
||||
""" True if expression_var is declared in this constraint set """
|
||||
"""True if expression_var is declared in this constraint set"""
|
||||
if not isinstance(expression_var, Variable):
|
||||
raise ValueError(f"Expression must be a Variable (not a {type(expression_var)})")
|
||||
return any(expression_var is x for x in self.get_declared_variables())
|
||||
|
||||
@@ -744,7 +744,7 @@ class Array(Expression, abstract=True):
|
||||
def __init__(self, *, index_bits: int, index_max: Optional[int], value_bits: int, **kwargs):
|
||||
assert index_bits in (32, 64, 256)
|
||||
assert value_bits in (8, 16, 32, 64, 256)
|
||||
assert index_max is None or index_max >= 0 and index_max < 2 ** index_bits
|
||||
assert index_max is None or index_max >= 0 and index_max < 2**index_bits
|
||||
self._index_bits = index_bits
|
||||
self._index_max = index_max
|
||||
self._value_bits = value_bits
|
||||
|
||||
@@ -221,7 +221,7 @@ class WorkerSingle(Worker):
|
||||
|
||||
|
||||
class WorkerThread(Worker):
|
||||
""" A worker thread """
|
||||
"""A worker thread"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
@@ -237,7 +237,7 @@ class WorkerThread(Worker):
|
||||
|
||||
|
||||
class WorkerProcess(Worker):
|
||||
""" A worker process """
|
||||
"""A worker process"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
@@ -278,21 +278,21 @@ class DaemonThread(WorkerThread):
|
||||
|
||||
|
||||
class DumpTCPHandler(socketserver.BaseRequestHandler):
|
||||
""" TCP Handler that calls the `dump` method bound to the server """
|
||||
"""TCP Handler that calls the `dump` method bound to the server"""
|
||||
|
||||
def handle(self):
|
||||
self.request.sendall(self.server.dump())
|
||||
|
||||
|
||||
class ReusableTCPServer(socketserver.TCPServer):
|
||||
""" Custom socket server that gracefully allows the address to be reused """
|
||||
"""Custom socket server that gracefully allows the address to be reused"""
|
||||
|
||||
allow_reuse_address = True
|
||||
dump: typing.Optional[typing.Callable] = None
|
||||
|
||||
|
||||
class LogCaptureWorker(DaemonThread):
|
||||
""" Extended DaemonThread that runs a TCP server that dumps the captured logs """
|
||||
"""Extended DaemonThread that runs a TCP server that dumps the captured logs"""
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
@@ -31,7 +31,7 @@ class ABI:
|
||||
|
||||
@staticmethod
|
||||
def _type_size(ty):
|
||||
""" Calculate `static` type size """
|
||||
"""Calculate `static` type size"""
|
||||
if ty[0] in ("int", "uint", "bytesM", "function"):
|
||||
return 32
|
||||
elif ty[0] in ("tuple"):
|
||||
|
||||
@@ -1075,8 +1075,8 @@ class ManticoreEVM(ManticoreBase):
|
||||
args=None,
|
||||
compile_args=None,
|
||||
):
|
||||
owner_account = self.create_account(balance=10 ** 10, name="owner", address=0x10000)
|
||||
attacker_account = self.create_account(balance=10 ** 10, name="attacker", address=0x20000)
|
||||
owner_account = self.create_account(balance=10**10, name="owner", address=0x10000)
|
||||
attacker_account = self.create_account(balance=10**10, name="attacker", address=0x20000)
|
||||
# Pretty print
|
||||
logger.info("Starting symbolic create contract")
|
||||
|
||||
|
||||
@@ -100,7 +100,7 @@ class FilterFunctions(Plugin):
|
||||
|
||||
|
||||
class LoopDepthLimiter(Plugin):
|
||||
""" This just aborts explorations that are too deep """
|
||||
"""This just aborts explorations that are too deep"""
|
||||
|
||||
def __init__(self, loop_count_threshold=5, **kwargs):
|
||||
super().__init__(**kwargs)
|
||||
|
||||
@@ -42,7 +42,7 @@ class SolidityMetadata:
|
||||
abi,
|
||||
warnings,
|
||||
):
|
||||
""" Contract metadata for Solidity-based contracts """
|
||||
"""Contract metadata for Solidity-based contracts"""
|
||||
self.name = name
|
||||
if isinstance(source_code, bytes):
|
||||
source_code = source_code.decode()
|
||||
|
||||
@@ -149,10 +149,10 @@ def manticore_verifier(
|
||||
user_accounts = []
|
||||
for n, address_i in enumerate(senders):
|
||||
user_accounts.append(
|
||||
m.create_account(balance=10 ** 10, address=address_i, name=f"sender_{n}")
|
||||
m.create_account(balance=10**10, address=address_i, name=f"sender_{n}")
|
||||
)
|
||||
# the address used for deployment
|
||||
owner_account = m.create_account(balance=10 ** 10, address=deployer, name="deployer")
|
||||
owner_account = m.create_account(balance=10**10, address=deployer, name="deployer")
|
||||
# the target contract account
|
||||
contract_account = m.solidity_create_contract(
|
||||
source_code,
|
||||
@@ -162,7 +162,7 @@ def manticore_verifier(
|
||||
name="contract_account",
|
||||
)
|
||||
# the address used for checking porperties
|
||||
checker_account = m.create_account(balance=10 ** 10, address=psender, name="psender")
|
||||
checker_account = m.create_account(balance=10**10, address=psender, name="psender")
|
||||
|
||||
print(f"# Owner account: 0x{int(owner_account):x}")
|
||||
print(f"# Contract account: 0x{int(contract_account):x}")
|
||||
|
||||
@@ -193,7 +193,7 @@ class Aarch64RegisterFile(RegisterFile):
|
||||
name = self._alias(register)
|
||||
parent, size = self._table[name]
|
||||
if isinstance(value, int):
|
||||
assert value <= 2 ** size - 1
|
||||
assert value <= 2**size - 1
|
||||
else:
|
||||
assert value.size == size
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ def _sig_is_varargs(sig: inspect.Signature) -> bool:
|
||||
|
||||
|
||||
class CpuException(Exception):
|
||||
""" Base cpu exception """
|
||||
"""Base cpu exception"""
|
||||
|
||||
|
||||
class DecodeException(CpuException):
|
||||
@@ -66,11 +66,11 @@ class InstructionEmulationError(CpuException):
|
||||
|
||||
|
||||
class DivideByZeroError(CpuException):
|
||||
""" A division by zero """
|
||||
"""A division by zero"""
|
||||
|
||||
|
||||
class Interruption(CpuException):
|
||||
""" A software interrupt. """
|
||||
"""A software interrupt."""
|
||||
|
||||
def __init__(self, N):
|
||||
super().__init__("CPU Software Interruption %08x" % N)
|
||||
@@ -191,7 +191,7 @@ class Operand:
|
||||
|
||||
@property
|
||||
def size(self):
|
||||
""" Return bit size of operand """
|
||||
"""Return bit size of operand"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
@@ -199,15 +199,15 @@ class Operand:
|
||||
return self._reg_name(self.op.reg)
|
||||
|
||||
def address(self):
|
||||
""" On a memory operand it returns the effective address """
|
||||
"""On a memory operand it returns the effective address"""
|
||||
raise NotImplementedError
|
||||
|
||||
def read(self):
|
||||
""" It reads the operand value from the registers or memory """
|
||||
"""It reads the operand value from the registers or memory"""
|
||||
raise NotImplementedError
|
||||
|
||||
def write(self, value):
|
||||
""" It writes the value of specific type to the registers or memory """
|
||||
"""It writes the value of specific type to the registers or memory"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
@@ -252,12 +252,12 @@ class RegisterFile:
|
||||
|
||||
@property
|
||||
def all_registers(self):
|
||||
""" Lists all possible register names (Including aliases) """
|
||||
"""Lists all possible register names (Including aliases)"""
|
||||
return tuple(self._aliases)
|
||||
|
||||
@property
|
||||
def canonical_registers(self):
|
||||
""" List the minimal most beautiful set of registers needed """
|
||||
"""List the minimal most beautiful set of registers needed"""
|
||||
raise NotImplementedError
|
||||
|
||||
def __contains__(self, register):
|
||||
@@ -602,7 +602,7 @@ class Cpu(Eventful):
|
||||
# Register access
|
||||
@property
|
||||
def regfile(self):
|
||||
""" The RegisterFile of this cpu """
|
||||
"""The RegisterFile of this cpu"""
|
||||
return self._regfile
|
||||
|
||||
@property
|
||||
|
||||
@@ -60,7 +60,7 @@ def SInt(value, width):
|
||||
return Operators.ITEBV(
|
||||
width,
|
||||
Bit(value, width - 1) == 1,
|
||||
GetNBits(value, width) - 2 ** width,
|
||||
GetNBits(value, width) - 2**width,
|
||||
GetNBits(value, width),
|
||||
)
|
||||
|
||||
|
||||
@@ -145,7 +145,7 @@ class Map(object, metaclass=ABCMeta):
|
||||
self._name = name
|
||||
|
||||
def _get_perms(self) -> str:
|
||||
""" Gets the access permissions of the map. """
|
||||
"""Gets the access permissions of the map."""
|
||||
return self._perms
|
||||
|
||||
def _set_perms(self, perms: str) -> None:
|
||||
@@ -165,7 +165,7 @@ class Map(object, metaclass=ABCMeta):
|
||||
perms = property(_get_perms, _set_perms)
|
||||
|
||||
def access_ok(self, access) -> bool:
|
||||
""" Check if there is enough permissions for access """
|
||||
"""Check if there is enough permissions for access"""
|
||||
for c in access:
|
||||
if c not in self.perms:
|
||||
return False
|
||||
@@ -184,7 +184,7 @@ class Map(object, metaclass=ABCMeta):
|
||||
return self._name
|
||||
|
||||
def __len__(self):
|
||||
"""Returns the current size in bytes. """
|
||||
"""Returns the current size in bytes."""
|
||||
return self._end - self._start
|
||||
|
||||
def __repr__(self):
|
||||
@@ -223,7 +223,7 @@ class Map(object, metaclass=ABCMeta):
|
||||
return object.__hash__(self)
|
||||
|
||||
def _in_range(self, index) -> bool:
|
||||
""" Returns True if index is in range """
|
||||
"""Returns True if index is in range"""
|
||||
if isinstance(index, slice):
|
||||
in_range = (
|
||||
index.start < index.stop and index.start >= self.start and index.stop <= self.end
|
||||
@@ -278,7 +278,7 @@ class Map(object, metaclass=ABCMeta):
|
||||
|
||||
|
||||
class AnonMap(Map):
|
||||
""" A concrete anonymous memory map """
|
||||
"""A concrete anonymous memory map"""
|
||||
|
||||
def __init__(self, start: int, size: int, perms: str, data_init=None, name=None):
|
||||
"""
|
||||
@@ -1350,7 +1350,7 @@ class SMemory(Memory):
|
||||
from ..core.state import Concretize
|
||||
|
||||
def setstate(state, _value):
|
||||
""" Roll back PC to redo last instruction """
|
||||
"""Roll back PC to redo last instruction"""
|
||||
state.cpu.PC = state.cpu._last_pc
|
||||
|
||||
raise Concretize(
|
||||
|
||||
@@ -865,7 +865,7 @@ class Decree(Platform):
|
||||
self.check_timers()
|
||||
|
||||
def awake(self, procid):
|
||||
""" Remove procid from waitlists and reestablish it in the running list """
|
||||
"""Remove procid from waitlists and reestablish it in the running list"""
|
||||
logger.info(
|
||||
"Remove procid:%d from waitlists and reestablish it in the running list", procid
|
||||
)
|
||||
@@ -889,21 +889,21 @@ class Decree(Platform):
|
||||
return fd - 1
|
||||
|
||||
def signal_receive(self, fd):
|
||||
""" Awake one process waiting to receive data on fd """
|
||||
"""Awake one process waiting to receive data on fd"""
|
||||
connections = self.connections
|
||||
if connections(fd) and self.twait[connections(fd)]:
|
||||
procid = random.sample(self.twait[connections(fd)], 1)[0]
|
||||
self.awake(procid)
|
||||
|
||||
def signal_transmit(self, fd):
|
||||
""" Awake one process waiting to transmit data on fd """
|
||||
"""Awake one process waiting to transmit data on fd"""
|
||||
connections = self.connections
|
||||
if connections(fd) and self.rwait[connections(fd)]:
|
||||
procid = random.sample(self.rwait[connections(fd)], 1)[0]
|
||||
self.awake(procid)
|
||||
|
||||
def check_timers(self):
|
||||
""" Awake process if timer has expired """
|
||||
"""Awake process if timer has expired"""
|
||||
if self._current is None:
|
||||
# Advance the clocks. Go to future!!
|
||||
advance = min([x for x in self.timers if x is not None]) + 1
|
||||
|
||||
@@ -110,10 +110,10 @@ consts.add(
|
||||
|
||||
|
||||
# Auxiliary constants and functions
|
||||
TT256 = 2 ** 256
|
||||
TT256M1 = 2 ** 256 - 1
|
||||
MASK160 = 2 ** 160 - 1
|
||||
TT255 = 2 ** 255
|
||||
TT256 = 2**256
|
||||
TT256M1 = 2**256 - 1
|
||||
MASK160 = 2**160 - 1
|
||||
TT255 = 2**255
|
||||
TOOHIGHMEM = 0x1000
|
||||
DEFAULT_FORK = "istanbul"
|
||||
|
||||
@@ -2165,7 +2165,7 @@ class EVM(Eventful):
|
||||
return tx.return_value
|
||||
|
||||
def CALL_gas(self, wanted_gas, address, value, in_offset, in_size, out_offset, out_size):
|
||||
""" Dynamic gas for CALL instruction. _arguably turing complete in itself_ """
|
||||
"""Dynamic gas for CALL instruction. _arguably turing complete in itself_"""
|
||||
GCALLVALUE = 9000
|
||||
GCALLNEW = 25000
|
||||
wanted_gas = Operators.ZEXTEND(wanted_gas, 512)
|
||||
|
||||
@@ -2118,7 +2118,7 @@ class Linux(Platform):
|
||||
def _do_epoll_wait(self, epfd, events, maxevents, to) -> int:
|
||||
# EP_MAX_EVENTS = (INT_MAX / sizeof(struct epoll_event))
|
||||
sizeof_epoll_event = 12
|
||||
EP_MAX_EVENTS = (2 ** 32 - 1) // sizeof_epoll_event
|
||||
EP_MAX_EVENTS = (2**32 - 1) // sizeof_epoll_event
|
||||
if maxevents <= 0 or maxevents > EP_MAX_EVENTS:
|
||||
return -errno.EINVAL
|
||||
|
||||
@@ -3095,9 +3095,9 @@ class Linux(Platform):
|
||||
error: Returns -1
|
||||
"""
|
||||
if tv != 0:
|
||||
microseconds = int(time.time() * 10 ** 6)
|
||||
microseconds = int(time.time() * 10**6)
|
||||
self.current.write_bytes(
|
||||
tv, struct.pack("L", microseconds // (10 ** 6)) + struct.pack("L", microseconds)
|
||||
tv, struct.pack("L", microseconds // (10**6)) + struct.pack("L", microseconds)
|
||||
)
|
||||
if tz != 0:
|
||||
logger.warning("No support for time zones in sys_gettimeofday")
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -28,7 +28,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def stub(arity, _state, *args):
|
||||
"""Default function used for hostfunc calls when a proper import wasn't provided """
|
||||
"""Default function used for hostfunc calls when a proper import wasn't provided"""
|
||||
logger.info("Called stub function with args: %s", args)
|
||||
return [0 for _ in range(arity)] # TODO: Return symbolic values
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ warnings.simplefilter("default", category=ManticoreDeprecationWarning)
|
||||
|
||||
|
||||
def deprecated(message: str):
|
||||
"""A decorator for marking functions as deprecated. """
|
||||
"""A decorator for marking functions as deprecated."""
|
||||
assert isinstance(message, str), "The deprecated decorator requires a message string argument."
|
||||
|
||||
def decorator(func):
|
||||
|
||||
@@ -204,7 +204,7 @@ def pretty_print_state_descriptors(desc: Dict):
|
||||
|
||||
|
||||
class deque(collections.deque):
|
||||
""" A wrapper around collections.deque that adds a few APIs present in SyncManager.Queue """
|
||||
"""A wrapper around collections.deque that adds a few APIs present in SyncManager.Queue"""
|
||||
|
||||
def empty(self) -> bool:
|
||||
return len(self) == 0
|
||||
|
||||
+12
-12
@@ -782,7 +782,7 @@ class Executor(Eventful):
|
||||
stack.has_type_on_top(I32, 2)
|
||||
c2 = stack.pop()
|
||||
c1 = stack.pop()
|
||||
stack.push(I32.cast((c1 - c2 + 2 ** 32) & MASK_32))
|
||||
stack.push(I32.cast((c1 - c2 + 2**32) & MASK_32))
|
||||
|
||||
def i32_mul(self, store, stack):
|
||||
stack.has_type_on_top(I32, 2)
|
||||
@@ -798,7 +798,7 @@ class Executor(Eventful):
|
||||
if self.check_zero_div(can_div_0):
|
||||
raise ZeroDivisionTrap()
|
||||
res = Operators.SDIV(c1, c2)
|
||||
can_overflow = res == 2 ** 31
|
||||
can_overflow = res == 2**31
|
||||
if self.check_overflow(can_overflow):
|
||||
raise OverflowDivisionTrap()
|
||||
stack.push(I32.cast(res))
|
||||
@@ -939,7 +939,7 @@ class Executor(Eventful):
|
||||
stack.has_type_on_top(I64, 2)
|
||||
c2 = stack.pop()
|
||||
c1 = stack.pop()
|
||||
stack.push(I64.cast((c1 - c2 + 2 ** 64) & MASK_64))
|
||||
stack.push(I64.cast((c1 - c2 + 2**64) & MASK_64))
|
||||
|
||||
def i64_mul(self, store, stack):
|
||||
stack.has_type_on_top(I64, 2)
|
||||
@@ -958,7 +958,7 @@ class Executor(Eventful):
|
||||
res = Operators.SDIV(c1, c2)
|
||||
else:
|
||||
res = int(math.trunc(Decimal(c1) / Decimal(c2)))
|
||||
can_overflow = res == 2 ** 63
|
||||
can_overflow = res == 2**63
|
||||
if self.check_overflow(can_overflow):
|
||||
raise OverflowDivisionTrap()
|
||||
stack.push(I64.cast(res))
|
||||
@@ -1075,7 +1075,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I32, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I32, "infinity")
|
||||
if c1 >= 2 ** 31 or c1 <= -(2 ** 31) - 1:
|
||||
if c1 >= 2**31 or c1 <= -(2**31) - 1:
|
||||
raise InvalidConversionTrap(I32, c1)
|
||||
stack.push(I32.cast(c1))
|
||||
|
||||
@@ -1088,7 +1088,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I32, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I32, "infinity")
|
||||
if c1 >= 2 ** 32 or c1 <= -1:
|
||||
if c1 >= 2**32 or c1 <= -1:
|
||||
raise InvalidConversionTrap(I32, c1)
|
||||
stack.push(I32.cast(c1))
|
||||
|
||||
@@ -1101,7 +1101,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I32, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I32, "infinity")
|
||||
if c1 >= 2 ** 31 or c1 <= -(2 ** 31) - 1:
|
||||
if c1 >= 2**31 or c1 <= -(2**31) - 1:
|
||||
raise InvalidConversionTrap(I32, c1)
|
||||
stack.push(I32.cast(c1))
|
||||
|
||||
@@ -1114,7 +1114,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I32, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I32, "infinity")
|
||||
if c1 >= 2 ** 32 or c1 <= -1:
|
||||
if c1 >= 2**32 or c1 <= -1:
|
||||
raise InvalidConversionTrap(I32, c1)
|
||||
stack.push(I32.cast(c1))
|
||||
|
||||
@@ -1143,7 +1143,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I64, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I64, "infinity")
|
||||
if c1 >= 2 ** 63 or c1 <= -(2 ** 63) - 1:
|
||||
if c1 >= 2**63 or c1 <= -(2**63) - 1:
|
||||
raise InvalidConversionTrap(I64, c1)
|
||||
stack.push(I64.cast(c1))
|
||||
|
||||
@@ -1156,7 +1156,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I64, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I64, "infinity")
|
||||
if c1 >= 2 ** 64 or c1 <= -1:
|
||||
if c1 >= 2**64 or c1 <= -1:
|
||||
raise InvalidConversionTrap(I64, c1)
|
||||
stack.push(I64.cast(c1))
|
||||
|
||||
@@ -1169,7 +1169,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I64, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I64, "infinity")
|
||||
if c1 >= 2 ** 63 or c1 <= -(2 ** 63) - 1:
|
||||
if c1 >= 2**63 or c1 <= -(2**63) - 1:
|
||||
raise InvalidConversionTrap(I64, c1)
|
||||
stack.push(I64.cast(c1))
|
||||
|
||||
@@ -1182,7 +1182,7 @@ class Executor(Eventful):
|
||||
raise InvalidConversionTrap(I64, "NaN")
|
||||
if math.isinf(c1):
|
||||
raise InvalidConversionTrap(I64, "infinity")
|
||||
if c1 >= 2 ** 64 or c1 <= -1:
|
||||
if c1 >= 2**64 or c1 <= -1:
|
||||
raise InvalidConversionTrap(I64, c1)
|
||||
stack.push(I64.cast(c1))
|
||||
|
||||
|
||||
@@ -79,7 +79,7 @@ consts.add(
|
||||
)
|
||||
|
||||
#: Size of a standard WASM memory page
|
||||
PAGESIZE = 2 ** 16
|
||||
PAGESIZE = 2**16
|
||||
|
||||
|
||||
# Wrappers around integers that we use for indexing the store.
|
||||
|
||||
@@ -325,7 +325,7 @@ ImmType = typing.Union[
|
||||
|
||||
|
||||
class Instruction:
|
||||
"""Internal instruction class that's pickle-friendly and works with the type system """
|
||||
"""Internal instruction class that's pickle-friendly and works with the type system"""
|
||||
|
||||
__slots__ = ["opcode", "mnemonic", "imm"]
|
||||
opcode: int #: Opcode, used for dispatching instructions
|
||||
@@ -443,7 +443,7 @@ class TypeMismatchTrap(Trap):
|
||||
|
||||
|
||||
class ConcretizeStack(Concretize):
|
||||
"""Tells Manticore to concretize the value `depth` values from the end of the stack. """
|
||||
"""Tells Manticore to concretize the value `depth` values from the end of the stack."""
|
||||
|
||||
def __init__(self, depth: int, ty: type, message: str, expression, policy=None, **kwargs):
|
||||
"""
|
||||
|
||||
@@ -18,7 +18,7 @@ def rtd_dependent_deps():
|
||||
# (we need to know how to import a given native dependency so we can check if native dependencies are installed)
|
||||
native_deps = ["capstone==4.0.2", "pyelftools", "unicorn==1.0.2"]
|
||||
|
||||
lint_deps = ["black==20.8b1", "mypy==0.790"]
|
||||
lint_deps = ["black~=22.0", "mypy==0.790"]
|
||||
|
||||
auto_test_deps = ["py-evm"]
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ real_open = open
|
||||
|
||||
|
||||
def fake_open(filename, mode="r", *args, **kwargs):
|
||||
""" Replace normal global open with this for a wuick dry run """
|
||||
"""Replace normal global open with this for a wuick dry run"""
|
||||
from io import StringIO
|
||||
|
||||
logging.info("Fake openning %r", (filename, mode) + args)
|
||||
|
||||
@@ -90,7 +90,7 @@ class EthDetectorTest(unittest.TestCase):
|
||||
|
||||
|
||||
class EthRetVal(EthDetectorTest):
|
||||
""" Detect when a return value of a low level transaction instruction is ignored """
|
||||
"""Detect when a return value of a low level transaction instruction is ignored"""
|
||||
|
||||
DETECTOR_CLASS = DetectUnusedRetVal
|
||||
|
||||
@@ -263,7 +263,7 @@ class EthEnvInstruction(EthDetectorTest):
|
||||
|
||||
|
||||
class EthDelegatecall(EthDetectorTest):
|
||||
""" Test the detection of funny delegatecalls """
|
||||
"""Test the detection of funny delegatecalls"""
|
||||
|
||||
DETECTOR_CLASS = DetectDelegatecall
|
||||
|
||||
@@ -403,7 +403,7 @@ class EthRaceCondition(EthDetectorTest):
|
||||
|
||||
|
||||
class EthBalance(EthDetectorTest):
|
||||
""" Test the detection of funny delegatecalls """
|
||||
"""Test the detection of funny delegatecalls"""
|
||||
|
||||
DETECTOR_CLASS = DetectManipulableBalance
|
||||
|
||||
|
||||
@@ -130,7 +130,7 @@ class EthAbiTests(unittest.TestCase):
|
||||
}
|
||||
}
|
||||
"""
|
||||
user_account = m.create_account(balance=10 ** 10, name="user_account")
|
||||
user_account = m.create_account(balance=10**10, name="user_account")
|
||||
contract_account = m.solidity_create_contract(
|
||||
source_code, owner=user_account, name="contract_account", gas=36225
|
||||
)
|
||||
@@ -179,7 +179,7 @@ class EthAbiTests(unittest.TestCase):
|
||||
d = b"".join(d)
|
||||
funcname, dynargs = ABI.deserialize(type_spec="func(uint256,uint256)", data=d)
|
||||
# self.assertEqual(funcname, 'AAAA')
|
||||
self.assertEqual(dynargs, (32, 2 ** 256 - 1))
|
||||
self.assertEqual(dynargs, (32, 2**256 - 1))
|
||||
|
||||
def test_simple_types1(self):
|
||||
d = [
|
||||
@@ -192,7 +192,7 @@ class EthAbiTests(unittest.TestCase):
|
||||
d = b"".join(d)
|
||||
funcname, dynargs = ABI.deserialize(type_spec="func(uint256,uint256,bool,address)", data=d)
|
||||
# self.assertEqual(funcname, 'AAAA')
|
||||
self.assertEqual(dynargs, (32, 2 ** 256 - 1, 0xFF, 0x424242))
|
||||
self.assertEqual(dynargs, (32, 2**256 - 1, 0xFF, 0x424242))
|
||||
|
||||
def test_simple_types_ints(self):
|
||||
d = [
|
||||
@@ -203,7 +203,7 @@ class EthAbiTests(unittest.TestCase):
|
||||
d = b"".join(d)
|
||||
func_id, dynargs = ABI.deserialize(type_spec="func(int256,int256)", data=d)
|
||||
self.assertEqual(func_id, b"AAAA")
|
||||
self.assertEqual(dynargs, (2 ** 255 - 1, -(2 ** 255)))
|
||||
self.assertEqual(dynargs, (2**255 - 1, -(2**255)))
|
||||
|
||||
def test_simple_types_ints_symbolic(self):
|
||||
cs = ConstraintSet()
|
||||
@@ -335,11 +335,11 @@ class EthAbiTests(unittest.TestCase):
|
||||
data = b"\xFF" * 32
|
||||
|
||||
parsed = ABI.deserialize("uint", data)
|
||||
self.assertEqual(parsed, 2 ** 256 - 1)
|
||||
self.assertEqual(parsed, 2**256 - 1)
|
||||
|
||||
for i in range(8, 257, 8):
|
||||
parsed = ABI.deserialize(f"uint{i}", data)
|
||||
self.assertEqual(parsed, 2 ** i - 1)
|
||||
self.assertEqual(parsed, 2**i - 1)
|
||||
|
||||
def test_empty_types(self):
|
||||
name, args = ABI.deserialize("func()", "\0" * 32)
|
||||
@@ -487,7 +487,7 @@ class EthTests(unittest.TestCase):
|
||||
|
||||
def test_solidity_create_contract_no_args(self):
|
||||
source_code = "contract A { constructor() {} }"
|
||||
owner = self.mevm.create_account(balance=10 ** 10)
|
||||
owner = self.mevm.create_account(balance=10**10)
|
||||
|
||||
# The default `args=()` makes it pass no arguments
|
||||
contract1 = self.mevm.solidity_create_contract(source_code, owner=owner)
|
||||
@@ -525,7 +525,7 @@ class EthTests(unittest.TestCase):
|
||||
|
||||
def test_solidity_create_contract_with_payable_constructor(self):
|
||||
source_code = "contract A { constructor() public payable {} }"
|
||||
owner = self.mevm.create_account(balance=10 ** 10)
|
||||
owner = self.mevm.create_account(balance=10**10)
|
||||
|
||||
contract = self.mevm.solidity_create_contract(source_code, owner=owner, balance=100)
|
||||
|
||||
@@ -613,7 +613,7 @@ class EthTests(unittest.TestCase):
|
||||
|
||||
}
|
||||
"""
|
||||
user_account = self.mevm.create_account(balance=10 ** 10)
|
||||
user_account = self.mevm.create_account(balance=10**10)
|
||||
contract_account = self.mevm.solidity_create_contract(source_code, owner=user_account)
|
||||
with self.assertRaises(EthereumError) as ctx:
|
||||
contract_account.ret(self.mevm.make_symbolic_value(), signature="(uint8)")
|
||||
@@ -674,7 +674,7 @@ class EthTests(unittest.TestCase):
|
||||
"""
|
||||
Tests issue 1325.
|
||||
"""
|
||||
owner = self.mevm.create_account(balance=10 ** 10)
|
||||
owner = self.mevm.create_account(balance=10**10)
|
||||
A = self.mevm.solidity_create_contract(
|
||||
"contract A { function foo() { revert(); } }", owner=owner
|
||||
)
|
||||
@@ -755,7 +755,7 @@ class EthTests(unittest.TestCase):
|
||||
"""
|
||||
consts = config.get_group("evm")
|
||||
consts.events = True
|
||||
user_account = self.mevm.create_account(balance=10 ** 10)
|
||||
user_account = self.mevm.create_account(balance=10**10)
|
||||
contract_account = self.mevm.solidity_create_contract(source_code, owner=user_account)
|
||||
input_sym = self.mevm.make_symbolic_value()
|
||||
contract_account.f(input_sym)
|
||||
@@ -813,7 +813,7 @@ class EthTests(unittest.TestCase):
|
||||
}
|
||||
}
|
||||
"""
|
||||
user_account = self.mevm.create_account(balance=10 ** 10)
|
||||
user_account = self.mevm.create_account(balance=10**10)
|
||||
contract_account = self.mevm.solidity_create_contract(source_code, owner=user_account)
|
||||
contract_account.ret(
|
||||
self.mevm.make_symbolic_value(),
|
||||
@@ -838,8 +838,8 @@ class EthTests(unittest.TestCase):
|
||||
}
|
||||
"""
|
||||
|
||||
owner_account = m.create_account(balance=10 ** 10)
|
||||
attacker_account = m.create_account(balance=10 ** 10)
|
||||
owner_account = m.create_account(balance=10**10)
|
||||
attacker_account = m.create_account(balance=10**10)
|
||||
contract_account = m.solidity_create_contract(contract_src, owner=owner_account, balance=0)
|
||||
|
||||
# Some global expression `sym_add1`
|
||||
@@ -886,7 +886,7 @@ class EthTests(unittest.TestCase):
|
||||
|
||||
def test_regression_internal_tx(self):
|
||||
m = self.mevm
|
||||
owner_account = m.create_account(balance=10 ** 10)
|
||||
owner_account = m.create_account(balance=10**10)
|
||||
c = """
|
||||
contract C1 {
|
||||
function g() returns (uint) {
|
||||
@@ -971,7 +971,7 @@ class EthTests(unittest.TestCase):
|
||||
}
|
||||
"""
|
||||
|
||||
owner = m.create_account(balance=20 ** 10)
|
||||
owner = m.create_account(balance=20**10)
|
||||
contract = m.solidity_create_contract(contract_src, owner=owner)
|
||||
receiver = m.create_account(balance=0)
|
||||
symbolic_address = m.make_symbolic_address()
|
||||
@@ -1169,7 +1169,7 @@ class EthTests(unittest.TestCase):
|
||||
"""
|
||||
|
||||
# Initiate the accounts
|
||||
user_account = m.create_account(balance=10 ** 10)
|
||||
user_account = m.create_account(balance=10**10)
|
||||
contract_account = m.solidity_create_contract(source_code, owner=user_account, balance=0)
|
||||
|
||||
contract_account.f(1) # it works
|
||||
@@ -1331,7 +1331,7 @@ class EthTests(unittest.TestCase):
|
||||
"""
|
||||
m: ManticoreEVM = self.mevm
|
||||
|
||||
creator_account = m.create_account(balance=10 ** 10)
|
||||
creator_account = m.create_account(balance=10**10)
|
||||
contract_account = m.solidity_create_contract(source_code, owner=creator_account, balance=0)
|
||||
|
||||
data = m.make_symbolic_buffer(320)
|
||||
|
||||
@@ -21,7 +21,7 @@ config.get_group("core").mprocessing = "single"
|
||||
|
||||
|
||||
class EthBenchmark(unittest.TestCase):
|
||||
""" https://consensys.net/diligence/evm-analyzer-benchmark-suite/ """
|
||||
"""https://consensys.net/diligence/evm-analyzer-benchmark-suite/"""
|
||||
|
||||
def setUp(self):
|
||||
self.mevm = ManticoreEVM()
|
||||
|
||||
@@ -2,11 +2,11 @@ import unittest
|
||||
|
||||
from manticore.native.cpu.aarch64 import Aarch64RegisterFile as RF
|
||||
|
||||
MAX_128 = 2 ** 128 - 1
|
||||
MAX_64 = 2 ** 64 - 1
|
||||
MAX_32 = 2 ** 32 - 1
|
||||
MAX_16 = 2 ** 16 - 1
|
||||
MAX_8 = 2 ** 8 - 1
|
||||
MAX_128 = 2**128 - 1
|
||||
MAX_64 = 2**64 - 1
|
||||
MAX_32 = 2**32 - 1
|
||||
MAX_16 = 2**16 - 1
|
||||
MAX_8 = 2**8 - 1
|
||||
|
||||
MAGIC_128 = 0x41424344454647484950515253545556
|
||||
MAGIC_64 = MAGIC_128 & MAX_64
|
||||
|
||||
@@ -557,7 +557,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("mov r0, r1")
|
||||
def test_mov_immreg1(self):
|
||||
self.rf.write("R1", 2 ** 32)
|
||||
self.rf.write("R1", 2**32)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R0"), 0)
|
||||
|
||||
@@ -634,7 +634,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
def test_movs_reg1(self):
|
||||
self.rf.write("R1", 2 ** 32)
|
||||
self.rf.write("R1", 2**32)
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
self.cpu.execute()
|
||||
@@ -643,11 +643,11 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
def test_movs_reg2(self):
|
||||
self.rf.write("R1", 2 ** 32 - 1)
|
||||
self.rf.write("R1", 2**32 - 1)
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R0"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R0"), 2**32 - 1)
|
||||
self._checkFlagsNZCV(1, 0, pre_c, pre_v)
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
@@ -699,7 +699,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("add r3, r1, 0x1")
|
||||
def test_add_imm_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 0x80000000)
|
||||
|
||||
@@ -733,7 +733,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("add r3, r1, r2")
|
||||
def test_add_reg_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.rf.write("R2", 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), (1 << 31))
|
||||
@@ -820,17 +820,17 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
def test_add_reg_sft_rrx(self):
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, rrx")
|
||||
def test_add_reg_sft_rrx2(self):
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, lsl r4")
|
||||
def test_add_reg_sft_lsl_reg(self):
|
||||
@@ -884,17 +884,17 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
def test_add_reg_sft_rrx_reg(self):
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, rrx")
|
||||
def test_add_reg_sft_rrx2_reg(self):
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
|
||||
# ADDS
|
||||
|
||||
@@ -935,7 +935,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("adds r3, r1, 0x1")
|
||||
def test_adds_imm_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 0x80000000)
|
||||
self._checkFlagsNZCV(1, 0, 0, 1)
|
||||
@@ -981,7 +981,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("adds r3, r1, r2")
|
||||
def test_adds_reg_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.rf.write("R2", 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), (1 << 31))
|
||||
@@ -1023,18 +1023,18 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
def test_adds_reg_sft_rrx(self):
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
self._checkFlagsNZCV(0, 0, 0, 0)
|
||||
|
||||
@itest_custom("adds r3, r1, r2, rrx")
|
||||
def test_adds_reg_sft_rrx2(self):
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
self.cpu.execute()
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
self._checkFlagsNZCV(1, 0, 0, 0)
|
||||
|
||||
@itest_setregs("R0=0")
|
||||
@@ -2240,7 +2240,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
// ovf should still be 1
|
||||
"""
|
||||
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self._setupCpu("adds r2, r1, #0x1")
|
||||
self.cpu.execute()
|
||||
self.rf.write("R1", 1)
|
||||
@@ -2321,7 +2321,7 @@ class Armv7CpuInstructions(unittest.TestCase):
|
||||
@itest_setregs("R1=0xfffffffe", "R2=0xfffffffe")
|
||||
@itest("UMULLS R1, R2, R1, R2")
|
||||
def test_umull_max(self):
|
||||
mul = 0xFFFFFFFE ** 2
|
||||
mul = 0xFFFFFFFE**2
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
self.assertEqual(self.rf.read("R1"), mul & Mask(32))
|
||||
|
||||
@@ -193,7 +193,7 @@ def assemble(asm: str, mode=CS_MODE_ARM) -> bytes:
|
||||
|
||||
|
||||
def emulate_next(cpu):
|
||||
"Read the next instruction and emulate it with Unicorn "
|
||||
"Read the next instruction and emulate it with Unicorn"
|
||||
cpu.decode_instruction(cpu.PC)
|
||||
emu = UnicornEmulator(cpu)
|
||||
emu.emulate(cpu.instruction)
|
||||
@@ -300,7 +300,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("mov r0, r1")
|
||||
def test_mov_immreg1(self):
|
||||
self.rf.write("R1", 2 ** 32)
|
||||
self.rf.write("R1", 2**32)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R0"), 0)
|
||||
|
||||
@@ -377,7 +377,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
def test_movs_reg1(self):
|
||||
self.rf.write("R1", 2 ** 32)
|
||||
self.rf.write("R1", 2**32)
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
emulate_next(self.cpu)
|
||||
@@ -386,11 +386,11 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
def test_movs_reg2(self):
|
||||
self.rf.write("R1", 2 ** 32 - 1)
|
||||
self.rf.write("R1", 2**32 - 1)
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R0"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R0"), 2**32 - 1)
|
||||
self._checkFlagsNZCV(1, 0, pre_c, pre_v)
|
||||
|
||||
@itest_custom("movs r0, r1")
|
||||
@@ -430,7 +430,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("add r3, r1, 0x1")
|
||||
def test_add_imm_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 0x80000000)
|
||||
|
||||
@@ -464,7 +464,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("add r3, r1, r2")
|
||||
def test_add_reg_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.rf.write("R2", 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), (1 << 31))
|
||||
@@ -532,17 +532,17 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
def test_add_reg_sft_rrx(self):
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, rrx")
|
||||
def test_add_reg_sft_rrx2(self):
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, lsl r4")
|
||||
def test_add_reg_sft_lsl_reg(self):
|
||||
@@ -596,17 +596,17 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
def test_add_reg_sft_rrx_reg(self):
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
|
||||
@itest_custom("add r3, r1, r2, rrx")
|
||||
def test_add_reg_sft_rrx2_reg(self):
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
|
||||
# ADDS
|
||||
|
||||
@@ -647,7 +647,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("adds r3, r1, 0x1")
|
||||
def test_adds_imm_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 0x80000000)
|
||||
self._checkFlagsNZCV(1, 0, 0, 1)
|
||||
@@ -693,7 +693,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
|
||||
@itest_custom("adds r3, r1, r2")
|
||||
def test_adds_reg_overflow(self):
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self.rf.write("R2", 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), (1 << 31))
|
||||
@@ -735,18 +735,18 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
def test_adds_reg_sft_rrx(self):
|
||||
self.rf.write("APSR_C", 0x0)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 31 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**31 - 1)
|
||||
self._checkFlagsNZCV(0, 0, 0, 0)
|
||||
|
||||
@itest_custom("adds r3, r1, r2, rrx")
|
||||
def test_adds_reg_sft_rrx2(self):
|
||||
self.rf.write("APSR_C", 0x1)
|
||||
self.rf.write("R1", 0x0)
|
||||
self.rf.write("R2", 2 ** 32 - 1)
|
||||
self.rf.write("R2", 2**32 - 1)
|
||||
emulate_next(self.cpu)
|
||||
self.assertEqual(self.rf.read("R3"), 2 ** 32 - 1)
|
||||
self.assertEqual(self.rf.read("R3"), 2**32 - 1)
|
||||
self._checkFlagsNZCV(1, 0, 0, 0)
|
||||
|
||||
# LDR imm
|
||||
@@ -1386,7 +1386,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
// ovf should still be 1
|
||||
"""
|
||||
|
||||
self.rf.write("R1", (2 ** 31 - 1))
|
||||
self.rf.write("R1", (2**31 - 1))
|
||||
self._setupCpu("adds r2, r1, #0x1")
|
||||
emulate_next(self.cpu)
|
||||
self.rf.write("R1", 1)
|
||||
@@ -1450,7 +1450,7 @@ class Armv7UnicornInstructions(unittest.TestCase):
|
||||
@itest_setregs("R1=0xfffffffe", "R2=0xfffffffe")
|
||||
@itest("UMULLS R1, R2, R1, R2")
|
||||
def test_umull_max(self):
|
||||
mul = 0xFFFFFFFE ** 2
|
||||
mul = 0xFFFFFFFE**2
|
||||
pre_c = self.rf.read("APSR_C")
|
||||
pre_v = self.rf.read("APSR_V")
|
||||
self.assertEqual(self.rf.read("R1"), mul & Mask(32))
|
||||
|
||||
@@ -10,7 +10,7 @@ class CPUTest(unittest.TestCase):
|
||||
_multiprocess_can_split_ = True
|
||||
|
||||
class ROOperand:
|
||||
""" Mocking class for operand ronly """
|
||||
"""Mocking class for operand ronly"""
|
||||
|
||||
def __init__(self, size, value):
|
||||
self.size = size
|
||||
@@ -20,7 +20,7 @@ class CPUTest(unittest.TestCase):
|
||||
return self.value & ((1 << self.size) - 1)
|
||||
|
||||
class RWOperand(ROOperand):
|
||||
""" Mocking class for operand rw """
|
||||
"""Mocking class for operand rw"""
|
||||
|
||||
def write(self, value):
|
||||
self.value = value & ((1 << self.size) - 1)
|
||||
|
||||
@@ -12,7 +12,7 @@ class CPUTest(unittest.TestCase):
|
||||
_multiprocess_can_split_ = True
|
||||
|
||||
class ROOperand:
|
||||
""" Mocking class for operand ronly """
|
||||
"""Mocking class for operand ronly"""
|
||||
|
||||
def __init__(self, size, value):
|
||||
self.size = size
|
||||
@@ -22,7 +22,7 @@ class CPUTest(unittest.TestCase):
|
||||
return self.value & ((1 << self.size) - 1)
|
||||
|
||||
class RWOperand(ROOperand):
|
||||
""" Mocking class for operand rw """
|
||||
"""Mocking class for operand rw"""
|
||||
|
||||
def write(self, value):
|
||||
self.value = value & ((1 << self.size) - 1)
|
||||
|
||||
@@ -1373,7 +1373,7 @@ class MemoryTest(unittest.TestCase):
|
||||
|
||||
mem.mprotect(addr, size, "w")
|
||||
with self.assertRaisesRegex(
|
||||
InvalidMemoryAccess, fr"Invalid memory access \(mode:.\) <{addr:x}>"
|
||||
InvalidMemoryAccess, rf"Invalid memory access \(mode:.\) <{addr:x}>"
|
||||
):
|
||||
_ = mem[addr]
|
||||
|
||||
@@ -1448,7 +1448,7 @@ class MemoryTest(unittest.TestCase):
|
||||
mem[addr] = "a"
|
||||
mem.mprotect(addr, size, "r")
|
||||
with self.assertRaisesRegex(
|
||||
InvalidMemoryAccess, fr"Invalid memory access \(mode:w\) <{addr:x}>"
|
||||
InvalidMemoryAccess, rf"Invalid memory access \(mode:w\) <{addr:x}>"
|
||||
):
|
||||
mem[addr] = "a"
|
||||
|
||||
@@ -1464,7 +1464,7 @@ class MemoryTest(unittest.TestCase):
|
||||
mem[addr] = "a"
|
||||
|
||||
with self.assertRaisesRegex(
|
||||
InvalidMemoryAccess, fr"Invalid memory access \(mode:r\) <{addr:x}>"
|
||||
InvalidMemoryAccess, rf"Invalid memory access \(mode:r\) <{addr:x}>"
|
||||
):
|
||||
_ = mem[addr]
|
||||
|
||||
|
||||
@@ -18,7 +18,7 @@ class RegisterTest(unittest.TestCase):
|
||||
|
||||
def test_truncate(self):
|
||||
r = Register(32)
|
||||
r.write(2 ** 32)
|
||||
r.write(2**32)
|
||||
self.assertEqual(r.read(), 0)
|
||||
|
||||
def test_largest_write(self):
|
||||
|
||||
@@ -12,7 +12,7 @@ class CPUTest(unittest.TestCase):
|
||||
_multiprocess_can_split_ = True
|
||||
|
||||
class ROOperand:
|
||||
""" Mocking class for operand ronly """
|
||||
"""Mocking class for operand ronly"""
|
||||
|
||||
def __init__(self, size, value):
|
||||
self.size = size
|
||||
@@ -22,7 +22,7 @@ class CPUTest(unittest.TestCase):
|
||||
return self.value & ((1 << self.size) - 1)
|
||||
|
||||
class RWOperand(ROOperand):
|
||||
""" Mocking class for operand rw """
|
||||
"""Mocking class for operand rw"""
|
||||
|
||||
def write(self, value):
|
||||
self.value = value & ((1 << self.size) - 1)
|
||||
|
||||
@@ -37,7 +37,7 @@ class CPUTest(unittest.TestCase):
|
||||
_multiprocess_can_split_ = True
|
||||
|
||||
class ROOperand:
|
||||
""" Mocking class for operand ronly """
|
||||
"""Mocking class for operand ronly"""
|
||||
|
||||
def __init__(self, size, value):
|
||||
self.size = size
|
||||
@@ -47,7 +47,7 @@ class CPUTest(unittest.TestCase):
|
||||
return self.value & ((1 << self.size) - 1)
|
||||
|
||||
class RWOperand(ROOperand):
|
||||
""" Mocking class for operand rw """
|
||||
"""Mocking class for operand rw"""
|
||||
|
||||
def write(self, value):
|
||||
self.value = value & ((1 << self.size) - 1)
|
||||
|
||||
@@ -37,7 +37,7 @@ class Module:
|
||||
|
||||
|
||||
def convert_types(to_convert):
|
||||
""" Convert unsigned ints from JSON into WASM Types (I32, F64, etc) """
|
||||
"""Convert unsigned ints from JSON into WASM Types (I32, F64, etc)"""
|
||||
out = []
|
||||
for item in to_convert:
|
||||
out.append(f"{item['type'].upper()}({item.get('value', 0)})")
|
||||
|
||||
@@ -80,7 +80,7 @@ env = Environment(loader=FileSystemLoader("."))
|
||||
|
||||
|
||||
def escape_null(in_str: str):
|
||||
""" Base-64 encode non-printable characters in test names so we can handle that obnoxious names module """
|
||||
"""Base-64 encode non-printable characters in test names so we can handle that obnoxious names module"""
|
||||
if in_str.isprintable() and not any((c in in_str) for c in {'"', "'", ";"}):
|
||||
return f'"{in_str}"'
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user