<commit_before>
import logging
from collections import defaultdict

import ailment

from ...engines.light import SpOffset, ArithmeticExpression, SimEngineLight, SimEngineLightVEXMixin, SimEngineLightAILMixin
from ...errors import SimEngineError, AngrVariableRecoveryError
from ...knowledge_plugins import Function
from ...sim_variable import SimStackVariable, SimRegisterVariable
from ..code_location import CodeLocation
from ..forward_analysis import ForwardAnalysis, FunctionGraphVisitor
from .variable_recovery_base import VariableRecoveryBase, VariableRecoveryStateBase

l = logging.getLogger(name=__name__)


class ProcessorState:

    __slots__ = ['_arch', 'sp_adjusted', 'sp_adjustment', 'bp_as_base', 'bp']

    def __init__(self, arch):
        self._arch = arch
        # whether we have met the initial stack pointer adjustment
        self.sp_adjusted = None
        # how many bytes are subtracted from the stack pointer
        self.sp_adjustment = arch.bytes if arch.call_pushes_ret else 0
        # whether the base pointer is used as the stack base of the stack frame or not
        self.bp_as_base = None
        # content of the base pointer
        self.bp = None

    def copy(self):
        s = ProcessorState(self._arch)
        s.sp_adjusted = self.sp_adjusted
        s.sp_adjustment = self.sp_adjustment
        s.bp_as_base = self.bp_as_base
        s.bp = self.bp
        return s

    def merge(self, other):
        if not self == other:
            l.warning("Inconsistent merge: %s %s ", self, other)

        # FIXME: none of the following logic makes any sense...
        if other.sp_adjusted is True:
            self.sp_adjusted = True
        self.sp_adjustment = max(self.sp_adjustment, other.sp_adjustment)
        if other.bp_as_base is True:
            self.bp_as_base = True
        if self.bp is None:
            self.bp = other.bp
        elif other.bp is not None:  # and self.bp is not None
            if self.bp == other.bp:
                pass
            else:
                if type(self.bp) is int and type(other.bp) is int:
                    self.bp = max(self.bp, other.bp)
                else:
                    self.bp = None
        return self

    def __eq__(self, other):
        if not isinstance(other, ProcessorState):
            return False
        return (self.sp_adjusted == other.sp_adjusted and
                self.sp_adjustment == other.sp_adjustment and
                self.bp == other.bp and
                self.bp_as_base == other.bp_as_base)

    def __repr__(self):
        return "<ProcessorState %s%#x%s %s>" % (self.bp, self.sp_adjustment,
            " adjusted" if self.sp_adjusted else "", self.bp_as_base)


#
# The base engine
#

class SimEngineVRBase(SimEngineLight):
    def __init__(self, project):
        super().__init__()

        self.project = project
        self.processor_state = None
        self.variable_manager = None

    @property
    def func_addr(self):
        if self.state is None:
            return None
        return self.state.function.addr

    def process(self, state, *args, **kwargs):  # pylint:disable=unused-argument

        self.processor_state = state.processor_state
        self.variable_manager = state.variable_manager

        try:
            self._process(state, None, block=kwargs.pop('block', None))
        except SimEngineError as e:
            if kwargs.pop('fail_fast', False) is True:
                raise e

    def _process(self, state, successors, block=None, func_addr=None):  # pylint:disable=unused-argument,arguments-differ
        super()._process(state, successors, block=block)

    #
    # Logic
    #

    def _assign_to_register(self, offset, data, size, src=None, dst=None):
        """

        :param int offset:
        :param data:
        :param int size:
        :return:
        """

        codeloc = self._codeloc()  # type: CodeLocation

        if offset == self.arch.sp_offset:
            if type(data) is SpOffset:
                sp_offset = data.offset
                self.processor_state.sp_adjusted = True
                self.processor_state.sp_adjustment = sp_offset
                l.debug('Adjusting stack pointer at %#x with offset %+#x.', self.ins_addr, sp_offset)
            return

        if offset == self.arch.bp_offset:
            if data is not None:
                self.processor_state.bp = data
            else:
                self.processor_state.bp = None
            return

        # handle register writes
        if type(data) is SpOffset and isinstance(data.offset, int):
            # lea
            stack_offset = data.offset
            existing_vars = self.variable_manager[self.func_addr].find_variables_by_stmt(self.block.addr,<commit_msg>VariableRecoveryFast: Handle aligning stack pointers by masking. (#2083)<commit_after>
import logging
from collections import defaultdict

import ailment

from ...engines.light import SpOffset, ArithmeticExpression, SimEngineLight, SimEngineLightVEXMixin, SimEngineLightAILMixin
from ...errors import SimEngineError, AngrVariableRecoveryError
from ...knowledge_plugins import Function
from ...sim_variable import SimStackVariable, SimRegisterVariable
from ..code_location import CodeLocation
from ..forward_analysis import ForwardAnalysis, FunctionGraphVisitor
from .variable_recovery_base import VariableRecoveryBase, VariableRecoveryStateBase

l = logging.getLogger(name=__name__)


class ProcessorState:

    __slots__ = ['_arch', 'sp_adjusted', 'sp_adjustment', 'bp_as_base', 'bp']

    def __init__(self, arch):
        self._arch = arch
        # whether we have met the initial stack pointer adjustment
        self.sp_adjusted = None
        # how many bytes are subtracted from the stack pointer
        self.sp_adjustment = arch.bytes if arch.call_pushes_ret else 0
        # whether the base pointer is used as the stack base of the stack frame or not
        self.bp_as_base = None
        # content of the base pointer
        self.bp = None

    def copy(self):
        s = ProcessorState(self._arch)
        s.sp_adjusted = self.sp_adjusted
        s.sp_adjustment = self.sp_adjustment
        s.bp_as_base = self.bp_as_base
        s.bp = self.bp
        return s

    def merge(self, other):
        if not self == other:
            l.warning("Inconsistent merge: %s %s ", self, other)

        # FIXME: none of the following logic makes any sense...
        if other.sp_adjusted is True:
            self.sp_adjusted = True
        self.sp_adjustment = max(self.sp_adjustment, other.sp_adjustment)
        if other.bp_as_base is True:
            self.bp_as_base = True
        if self.bp is None:
            self.bp = other.bp
        elif other.bp is not None:  # and self.bp is not None
            if self.bp == other.bp:
                pass
            else:
                if type(self.bp) is int and type(other.bp) is int:
                    self.bp = max(self.bp, other.bp)
                else:
                    self.bp = None
        return self

    def __eq__(self, other):
        if not isinstance(other, ProcessorState):
            return False
        return (self.sp_adjusted == other.sp_adjusted and
                self.sp_adjustment == other.sp_adjustment and
                self.bp == other.bp and
                self.bp_as_base == other.bp_as_base)

    def __repr__(self):
        return "<ProcessorState %s%#x%s %s>" % (self.bp, self.sp_adjustment,
            " adjusted" if self.sp_adjusted else "", self.bp_as_base)


#
# The base engine
#

class SimEngineVRBase(SimEngineLight):
    def __init__(self, project):
        super().__init__()

        self.project = project
        self.processor_state = None
        self.variable_manager = None

    @property
    def func_addr(self):
        if self.state is None:
            return None
        return self.state.function.addr

    def process(self, state, *args, **kwargs):  # pylint:disable=unused-argument

        self.processor_state = state.processor_state
        self.variable_manager = state.variable_manager

        try:
            self._process(state, None, block=kwargs.pop('block', None))
        except SimEngineError as e:
            if kwargs.pop('fail_fast', False) is True:
                raise e

    def _process(self, state, successors, block=None, func_addr=None):  # pylint:disable=unused-argument,arguments-differ
        super()._process(state, successors, block=block)

    #
    # Logic
    #

    def _assign_to_register(self, offset, data, size, src=None, dst=None):
        """

        :param int offset:
        :param data:
        :param int size:
        :return:
        """

        codeloc = self._codeloc()  # type: CodeLocation

        if offset == self.arch.sp_offset:
            if type(data) is SpOffset:
                sp_offset = data.offset
                if isinstance(sp_offset, int):
                    self.processor_state.sp_adjusted = True
                    self.processor_state.sp_adjustment = sp_offset
                    l.debug('Adjusting stack pointer at %#x with offset %+#x.', self.ins_addr, sp_offset)
                elif (isinstance(sp_offset, ArithmeticExpression)
                      and sp_offset.op == ArithmeticExpression.And
                      and isinstance(sp_offset.operands[0], SpOffset)
                      and isinstance(sp_offset.operands[1], int)):
                    l.debug('Masking stack pointer at %#x with mask %#x.', self.ins_addr, sp_offset.operands[1])
                    # ignore masking
                else:
                    l.debug('An unsupported arithmetic expression %r is assigned to stack pointer at %#x. Ignore.',
                            sp_offset,
                            self.ins_addr,
                            )
                    # ignore unsupported arithmetic expressions.
            return

        if offset == self.arch.bp_offset:
            if data is not None:
                self.processor_state.bp = data
            else:
                self.processor_state.bp = None
            return

        # handle register writes
        if type(data) is SpOffset and isinstance(data.offset, int):
            # lea
            stack_offset = data.offset
            existing_vars = self.variable_manager[self.func_addr].find_variables_by_stmt(self.block.addr,
{'fraction_non_alphanumeric': 0.08385602624722571, 'fraction_numerical': 0.0008684743800057898, 'mean_word_length': 2.0891207153502234, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '674537', 'n_tokens_mistral': 2963, 'n_tokens_neox': 2823, 'n_words': 791}