signature:
def _send ( self , cmd : str ) :
    """Send a string to the solver .
 : param cmd : a SMTLIBv2 command ( ex . ( check - sat ) )"""

implementation:
logger . debug ( '>%s' , cmd )
try :
    self . _proc . stdout . flush ( )
    self . _proc . stdin . write ( f'{cmd}\n' )
except IOError as e :
    raise SolverError ( str ( e ) )


signature:
def _recv ( self ) -> str :
    """Reads the response from the solver"""

implementation:
buf , left , right = self . __readline_and_count ( )
bufl = [ buf ]
while left != right :
    buf , l , r = self . __readline_and_count ( )
    bufl . append ( buf )
    left += l
    right += r
buf = '' . join ( bufl ) . strip ( )
logger . debug ( '<%s' , buf )
if '(error' in bufl [ 0 ] :
    raise Exception ( f"Error in smtlib: {bufl[0]}" )
return buf


signature:
def _is_sat ( self ) -> bool :
    """Check the satisfiability of the current state
 : return : whether current state is satisfiable or not ."""

implementation:
logger . debug ( "Solver.check() " )
start = time . time ( )
self . _send ( '(check-sat)' )
status = self . _recv ( )
logger . debug ( "Check took %s seconds (%s)" , time . time ( ) - start , status )
if status not in ( 'sat' , 'unsat' , 'unknown' ) :
    raise SolverError ( status )
if consider_unknown_as_unsat :
    if status == 'unknown' :
        logger . info ( 'Found an unknown core, probably a solver timeout' )
        status = 'unsat'
if status == 'unknown' :
    raise SolverUnknown ( status )
return status == 'sat'


signature:
def _assert ( self , expression : Bool ) :
    """Auxiliary method to send an assert"""

implementation:
assert isinstance ( expression , Bool )
smtlib = translate_to_smtlib ( expression )
self . _send ( '(assert %s)' % smtlib )


signature:
def _getvalue ( self , expression ) :
    """Ask the solver for one possible assignment for given expression using current set of constraints .
 The current set of expressions must be sat .
 NOTE : This is an internal method : it uses the current solver state ( set of constraints ! ) ."""

implementation:
if not issymbolic ( expression ) :
    return expression
assert isinstance ( expression , Variable )
if isinstance ( expression , Array ) :
    result = bytearray ( )
    for c in expression :
        expression_str = translate_to_smtlib ( c )
        self . _send ( '(get-value (%s))' % expression_str )
        response = self . _recv ( )
        result . append ( int ( '0x{:s}' . format ( response . split ( expression_str ) [ 1 ] [ 3 : - 2 ] ) , 16 ) )
    return bytes ( result )
else :
    self . _send ( '(get-value (%s))' % expression . name )
    ret = self . _recv ( )
    assert ret . startswith ( '((' ) and ret . endswith ( '))' ) , ret
    if isinstance ( expression , Bool ) :
        return { 'true' : True , 'false' : False } [ ret [ 2 : - 2 ] . split ( ' ' ) [ 1 ] ]
    elif isinstance ( expression , BitVec ) :
        pattern , base = self . _get_value_fmt
        m = pattern . match ( ret )
        expr , value = m . group ( 'expr' ) , m . group ( 'value' )
        return int ( value , base )
raise NotImplementedError ( "_getvalue only implemented for Bool and BitVec" )


signature:
def can_be_true ( self , constraints , expression ) :
    """Check if two potentially symbolic values can be equal"""

implementation:
if isinstance ( expression , bool ) :
    if not expression :
        return expression
    else : # if True check if constraints are feasible
        self . _reset ( constraints )
        return self . _is_sat ( )
assert isinstance ( expression , Bool )
with constraints as temp_cs :
    temp_cs . add ( expression )
    self . _reset ( temp_cs . to_string ( related_to = expression ) )
    return self . _is_sat ( )


signature:
def get_all_values ( self , constraints , expression , maxcnt = None , silent = False ) :
    """Returns a list with all the possible values for the symbol x"""

implementation:
if not isinstance ( expression , Expression ) :
    return [ expression ]
assert isinstance ( constraints , ConstraintSet )
assert isinstance ( expression , Expression )
expression = simplify ( expression )
if maxcnt is None :
    maxcnt = consts . maxsolutions
with constraints as temp_cs :
    if isinstance ( expression , Bool ) :
        var = temp_cs . new_bool ( )
    elif isinstance ( expression , BitVec ) :
        var = temp_cs . new_bitvec ( expression . size )
    elif isinstance ( expression , Array ) :
        var = temp_cs . new_array ( index_max = expression . index_max , value_bits = expression . value_bits , taint = expression . taint ) . array
    else :
        raise NotImplementedError ( f"get_all_values only implemented for {type(expression)} expression type." )
    temp_cs . add ( var == expression )
    self . _reset ( temp_cs . to_string ( related_to = var ) )
    result = [ ]
    while self . _is_sat ( ) :
        value = self . _getvalue ( var )
        result . append ( value )
        self . _assert ( var != value )
        if len ( result ) >= maxcnt :
            if silent : # do not throw an exception if set to silent
            # Default is not silent , assume user knows
            # what they are doing and will check the size
            # of returned vals list ( previous smtlib behavior )
                break
            else :
                raise TooManySolutions ( result )
    return result


signature:
def optimize ( self , constraints : ConstraintSet , x : BitVec , goal : str , M = 10000 ) :
    """Iteratively finds the maximum or minimum value for the operation
 ( Normally Operators . UGT or Operators . ULT )
 : param constraints : constraints to take into account
 : param x : a symbol or expression
 : param goal : goal to achieve , either ' maximize ' or ' minimize '
 : param M : maximum number of iterations allowed"""

implementation:
assert goal in ( 'maximize' , 'minimize' )
assert isinstance ( x , BitVec )
operation = { 'maximize' : Operators . UGE , 'minimize' : Operators . ULE } [ goal ]
with constraints as temp_cs :
    X = temp_cs . new_bitvec ( x . size )
    temp_cs . add ( X == x )
    aux = temp_cs . new_bitvec ( X . size , name = 'optimized_' )
    self . _reset ( temp_cs . to_string ( related_to = X ) )
    self . _send ( aux . declaration )
    if getattr ( self , f'support_{goal}' ) :
        self . _push ( )
        try :
            self . _assert ( operation ( X , aux ) )
            self . _send ( '(%s %s)' % ( goal , aux . name ) )
            self . _send ( '(check-sat)' )
            _status = self . _recv ( )
            if _status not in ( 'sat' , 'unsat' , 'unknown' ) : # Minimize ( or Maximize ) sometimes prints the objective before the status
            # This will be a line like NAME | - > VALUE
                maybe_sat = self . _recv ( )
                if maybe_sat == 'sat' :
                    m = RE_MIN_MAX_OBJECTIVE_EXPR_VALUE . match ( _status )
                    expr , value = m . group ( 'expr' ) , m . group ( 'value' )
                    assert expr == aux . name
                    return int ( value )
            elif _status == 'sat' :
                ret = self . _recv ( )
                if not ( ret . startswith ( '(' ) and ret . endswith ( ')' ) ) :
                    raise SolverError ( 'bad output on max, z3 may have been killed' )
                m = RE_OBJECTIVES_EXPR_VALUE . match ( ret )
                expr , value = m . group ( 'expr' ) , m . group ( 'value' )
                assert expr == aux . name
                return int ( value )
        finally :
            self . _pop ( )
            self . _reset ( temp_cs )
            self . _send ( aux . declaration )
    operation = { 'maximize' : Operators . UGT , 'minimize' : Operators . ULT } [ goal ]
    self . _assert ( aux == X )
    last_value = None
    i = 0
    while self . _is_sat ( ) :
        last_value = self . _getvalue ( aux )
        self . _assert ( operation ( aux , last_value ) )
        i = i + 1
        if i > M :
            raise SolverError ( "Optimizing error, maximum number of iterations was reached" )
    if last_value is not None :
        return last_value
    raise SolverError ( "Optimizing error, unsat or unknown core" )


signature:
def get_value ( self , constraints , expression ) :
    """Ask the solver for one possible result of given expression using given set of constraints ."""

implementation:
if not issymbolic ( expression ) :
    return expression
assert isinstance ( expression , ( Bool , BitVec , Array ) )
with constraints as temp_cs :
    if isinstance ( expression , Bool ) :
        var = temp_cs . new_bool ( )
    elif isinstance ( expression , BitVec ) :
        var = temp_cs . new_bitvec ( expression . size )
    elif isinstance ( expression , Array ) :
        var = [ ]
        result = [ ]
        for i in range ( expression . index_max ) :
            subvar = temp_cs . new_bitvec ( expression . value_bits )
            var . append ( subvar )
            temp_cs . add ( subvar == simplify ( expression [ i ] ) )
        self . _reset ( temp_cs )
        if not self . _is_sat ( ) :
            raise SolverError ( 'Model is not available' )
        for i in range ( expression . index_max ) :
            self . _send ( '(get-value (%s))' % var [ i ] . name )
            ret = self . _recv ( )
            assert ret . startswith ( '((' ) and ret . endswith ( '))' )
            pattern , base = self . _get_value_fmt
            m = pattern . match ( ret )
            expr , value = m . group ( 'expr' ) , m . group ( 'value' )
            result . append ( int ( value , base ) )
        return bytes ( result )
    temp_cs . add ( var == expression )
    self . _reset ( temp_cs )
if not self . _is_sat ( ) :
    raise SolverError ( 'Model is not available' )
self . _send ( '(get-value (%s))' % var . name )
ret = self . _recv ( )
if not ( ret . startswith ( '((' ) and ret . endswith ( '))' ) ) :
    raise SolverError ( 'SMTLIB error parsing response: %s' % ret )
if isinstance ( expression , Bool ) :
    return { 'true' : True , 'false' : False } [ ret [ 2 : - 2 ] . split ( ' ' ) [ 1 ] ]
if isinstance ( expression , BitVec ) :
    pattern , base = self . _get_value_fmt
    m = pattern . match ( ret )
    expr , value = m . group ( 'expr' ) , m . group ( 'value' )
    return int ( value , base )
raise NotImplementedError ( "get_value only implemented for Bool and BitVec" )


signature:
def summarized_name ( self , name ) :
    """Produce a summarized record name
 i . e . manticore . core . executor - > m . c . executor"""

implementation:
components = name . split ( '.' )
prefix = '.' . join ( c [ 0 ] for c in components [ : - 1 ] )
return f'{prefix}.{components[-1]}'


signature:
def colored_level_name ( self , levelname ) :
    """Colors the logging level in the logging record"""

implementation:
if self . colors_disabled :
    return self . plain_levelname_format . format ( levelname )
else :
    return self . colored_levelname_format . format ( self . color_map [ levelname ] , levelname )


signature:
def _find_zero ( cpu , constrs , ptr ) :
    """Helper for finding the closest NULL or , effectively NULL byte from a starting address .
 : param Cpu cpu :
 : param ConstraintSet constrs : Constraints for current ` State `
 : param int ptr : Address to start searching for a zero from
 : return : Offset from ` ptr ` to first byte that is 0 or an ` Expression ` that must be zero"""

implementation:
offset = 0
while True :
    byt = cpu . read_int ( ptr + offset , 8 )
    if issymbolic ( byt ) :
        if not solver . can_be_true ( constrs , byt != 0 ) :
            break
    else :
        if byt == 0 :
            break
    offset += 1
return offset


signature:
def strcmp ( state , s1 , s2 ) :
    """strcmp symbolic model .
 Algorithm : Walks from end of string ( minimum offset to NULL in either string )
 to beginning building tree of ITEs each time either of the
 bytes at current offset is symbolic .
 Points of Interest :
 - We ' ve been building up a symbolic tree but then encounter two
 concrete bytes that differ . We can throw away the entire symbolic
 tree !
 - If we ' ve been encountering concrete bytes that match
 at the end of the string as we walk forward , and then we encounter
 a pair where one is symbolic , we can forget about that 0 ` ret ` we ' ve
 been tracking and just replace it with the symbolic subtraction of
 the two
 : param State state : Current program state
 : param int s1 : Address of string 1
 : param int s2 : Address of string 2
 : return : Symbolic strcmp result
 : rtype : Expression or int"""

implementation:
cpu = state . cpu
if issymbolic ( s1 ) :
    raise ConcretizeArgument ( state . cpu , 1 )
if issymbolic ( s2 ) :
    raise ConcretizeArgument ( state . cpu , 2 )
s1_zero_idx = _find_zero ( cpu , state . constraints , s1 )
s2_zero_idx = _find_zero ( cpu , state . constraints , s2 )
min_zero_idx = min ( s1_zero_idx , s2_zero_idx )
ret = None
for offset in range ( min_zero_idx , - 1 , - 1 ) :
    s1char = ZEXTEND ( cpu . read_int ( s1 + offset , 8 ) , cpu . address_bit_size )
    s2char = ZEXTEND ( cpu . read_int ( s2 + offset , 8 ) , cpu . address_bit_size )
    if issymbolic ( s1char ) or issymbolic ( s2char ) :
        if ret is None or ( not issymbolic ( ret ) and ret == 0 ) :
            ret = s1char - s2char
        else :
            ret = ITEBV ( cpu . address_bit_size , s1char != s2char , s1char - s2char , ret )
    else :
        if s1char != s2char :
            ret = s1char - s2char
        elif ret is None :
            ret = 0
return ret


signature:
def strlen ( state , s ) :
    """strlen symbolic model .
 Algorithm : Walks from end of string not including NULL building ITE tree when current byte is symbolic .
 : param State state : current program state
 : param int s : Address of string
 : return : Symbolic strlen result
 : rtype : Expression or int"""

implementation:
cpu = state . cpu
if issymbolic ( s ) :
    raise ConcretizeArgument ( state . cpu , 1 )
zero_idx = _find_zero ( cpu , state . constraints , s )
ret = zero_idx
for offset in range ( zero_idx - 1 , - 1 , - 1 ) :
    byt = cpu . read_int ( s + offset , 8 )
    if issymbolic ( byt ) :
        ret = ITEBV ( cpu . address_bit_size , byt == 0 , offset , ret )
return ret


signature:
def all_events ( cls ) :
    """Return all events that all subclasses have so far registered to publish ."""

implementation:
all_evts = set ( )
for cls , evts in cls . __all_events__ . items ( ) :
    all_evts . update ( evts )
return all_evts


signature:
def forward_events_to ( self , sink , include_source = False ) :
    """This forwards signal to sink"""

implementation:
assert isinstance ( sink , Eventful ) , f'{sink.__class__.__name__} is not Eventful'
self . _forwards [ sink ] = include_source


signature:
def context ( self ) :
    """Convenient access to shared context"""

implementation:
if self . _context is not None :
    return self . _context
else :
    logger . warning ( "Using shared context without a lock" )
    return self . _executor . _shared_context


signature:
def locked_context ( self , key = None , value_type = list ) :
    """A context manager that provides safe parallel access to the global Manticore context .
 This should be used to access the global Manticore context
 when parallel analysis is activated . Code within the ` with ` block is executed
 atomically , so access of shared variables should occur within .
 Example use : :
 with m . locked _ context ( ) as context :
 visited = context [ ' visited ' ]
 visited . append ( state . cpu . PC )
 context [ ' visited ' ] = visited
 Optionally , parameters can specify a key and type for the object paired to this key . : :
 with m . locked _ context ( ' feature _ list ' , list ) as feature _ list :
 feature _ list . append ( 1)
 : param object key : Storage key
 : param value _ type : type of value associated with key
 : type value _ type : list or dict or set"""

implementation:
@ contextmanager
def _real_context ( ) :
    if self . _context is not None :
        yield self . _context
    else :
        with self . _executor . locked_context ( ) as context :
            yield context
with _real_context ( ) as context :
    if key is None :
        yield context
    else :
        assert value_type in ( list , dict , set )
        ctx = context . get ( key , value_type ( ) )
        yield ctx
        context [ key ] = ctx


signature:
def get_profiling_stats ( self ) :
    """Returns a pstat . Stats instance with profiling results if ` run ` was called with ` should _ profile = True ` .
 Otherwise , returns ` None ` ."""

implementation:
profile_file_path = os . path . join ( self . workspace , 'profiling.bin' )
try :
    return pstats . Stats ( profile_file_path )
except Exception as e :
    logger . debug ( f'Failed to get profiling stats: {e}' )
    return None


signature:
def run ( self , procs = 1 , timeout = None , should_profile = False ) :
    """Runs analysis .
 : param int procs : Number of parallel worker processes
 : param timeout : Analysis timeout , in seconds"""

implementation:
assert not self . running , "Manticore is already running."
self . _start_run ( )
self . _last_run_stats [ 'time_started' ] = time . time ( )
with self . shutdown_timeout ( timeout ) :
    self . _start_workers ( procs , profiling = should_profile )
    self . _join_workers ( )
self . _finish_run ( profiling = should_profile )


signature:
def main ( ) :
    """Dispatches execution into one of Manticore ' s engines : evm or native ."""

implementation:
args = parse_arguments ( )
if args . no_colors :
    log . disable_colors ( )
sys . setrecursionlimit ( consts . recursionlimit )
ManticoreBase . verbosity ( args . v )
if args . argv [ 0 ] . endswith ( '.sol' ) :
    ethereum_main ( args , logger )
else :
    install_helper . ensure_native_deps ( )
    native_main ( args , logger )


signature:
def GetNBits ( value , nbits ) :
    """Get the first ` nbits ` from ` value ` .
 : param value : Source value from which to extract
 : type value : int or long or BitVec
 : param int nbits : How many bits to extract
 : return : Low ` nbits ` bits of ` value ` .
 : rtype int or long or BitVec"""

implementation:
# NOP if sizes are the same
if isinstance ( value , int ) :
    return Operators . EXTRACT ( value , 0 , nbits )
elif isinstance ( value , BitVec ) :
    if value . size < nbits :
        return Operators . ZEXTEND ( value , nbits )
    else :
        return Operators . EXTRACT ( value , 0 , nbits )


signature:
def SInt ( value , width ) :
    """Convert a bitstring ` value ` of ` width ` bits to a signed integer
 representation .
 : param value : The value to convert .
 : type value : int or long or BitVec
 : param int width : The width of the bitstring to consider
 : return : The converted value
 : rtype int or long or BitVec"""

implementation: return Operators . ITEBV ( width , Bit ( value , width - 1 ) == 1 , GetNBits ( value , width ) - 2 ** width , GetNBits ( value , width ) )


signature:
def LSL_C ( value , amount , width ) :
    """The ARM LSL _ C ( logical left shift with carry ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value and the carry result
 : rtype tuple"""

implementation:
assert amount > 0
value = Operators . ZEXTEND ( value , width * 2 )
shifted = value << amount
result = GetNBits ( shifted , width )
carry = Bit ( shifted , width )
return ( result , carry )


signature:
def LSL ( value , amount , width ) :
    """The ARM LSL ( logical left shift ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value
 : rtype int or BitVec"""

implementation:
if amount == 0 :
    return value
result , _ = LSL_C ( value , amount , width )
return result


signature:
def LSR_C ( value , amount , width ) :
    """The ARM LSR _ C ( logical shift right with carry ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value and carry result
 : rtype tuple"""

implementation:
assert amount > 0
result = GetNBits ( value >> amount , width )
carry = Bit ( value >> ( amount - 1 ) , 0 )
return ( result , carry )


signature:
def LSR ( value , amount , width ) :
    """The ARM LSR ( logical shift right ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value
 : rtype int or BitVec"""

implementation:
if amount == 0 :
    return value
result , _ = LSR_C ( value , amount , width )
return result


signature:
def ASR_C ( value , amount , width ) :
    """The ARM ASR _ C ( arithmetic shift right with carry ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value and carry result
 : rtype tuple"""

implementation:
assert amount <= width
assert amount > 0
assert amount + width <= width * 2
value = Operators . SEXTEND ( value , width , width * 2 )
result = GetNBits ( value >> amount , width )
carry = Bit ( value , amount - 1 )
return ( result , carry )


signature:
def ASR ( value , amount , width ) :
    """The ARM ASR ( arithmetic shift right ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to shift it .
 : param int width : Width of the value
 : return : Resultant value
 : rtype int or BitVec"""

implementation:
if amount == 0 :
    return value
result , _ = ASR_C ( value , amount , width )
return result


signature:
def ROR_C ( value , amount , width ) :
    """The ARM ROR _ C ( rotate right with carry ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to rotate it .
 : param int width : Width of the value
 : return : Resultant value and carry result
 : rtype tuple"""

implementation:
assert amount <= width
assert amount > 0
m = amount % width
right , _ = LSR_C ( value , m , width )
left , _ = LSL_C ( value , width - m , width )
result = left | right
carry = Bit ( result , width - 1 )
return ( result , carry )


signature:
def ROR ( value , amount , width ) :
    """The ARM ROR ( rotate right ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to rotate it .
 : param int width : Width of the value
 : return : Resultant value
 : rtype int or BitVec"""

implementation:
if amount == 0 :
    return value
result , _ = ROR_C ( value , amount , width )
return result


signature:
def RRX_C ( value , carry , width ) :
    """The ARM RRX ( rotate right with extend and with carry ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to rotate it .
 : param int width : Width of the value
 : return : Resultant value and carry result
 : rtype tuple"""

implementation:
carry_out = Bit ( value , 0 )
result = ( value >> 1 ) | ( carry << ( width - 1 ) )
return ( result , carry_out )


signature:
def RRX ( value , carry , width ) :
    """The ARM RRX ( rotate right with extend ) operation .
 : param value : Value to shift
 : type value : int or long or BitVec
 : param int amount : How many bits to rotate it .
 : param int width : Width of the value
 : return : Resultant value
 : rtype int or BitVec"""

implementation:
result , _ = RRX_C ( value , carry , width )
return result


signature:
def locked_context ( self , key = None , default = dict ) :
    """Policy shared context dictionary"""

implementation:
keys = [ 'policy' ]
if key is not None :
    keys . append ( key )
with self . _executor . locked_context ( '.' . join ( keys ) , default ) as policy_context :
    yield policy_context


signature:
def _add_state_callback ( self , state_id , state ) :
    """Save summarize ( state ) on policy shared context before
 the state is stored"""

implementation:
summary = self . summarize ( state )
if summary is None :
    return
with self . locked_context ( 'summaries' , dict ) as ctx :
    ctx [ state_id ] = summary


signature:
def _visited_callback ( self , state , pc , instr ) :
    """Maintain our own copy of the visited set"""

implementation:
with self . locked_context ( 'visited' , set ) as ctx :
    ctx . add ( pc )


implementation:
pc = state . platform . current . PC
with self . locked_context ( 'visited' , dict ) as ctx :
    ctx [ pc ] = ctx . get ( pc , 0 ) + 1


signature:
def locked_context ( self , key = None , default = dict ) :
    """Executor context is a shared memory object . All workers share this .
 It needs a lock . Its used like this :
 with executor . context ( ) as context :
 visited = context [ ' visited ' ]
 visited . append ( state . cpu . PC )
 context [ ' visited ' ] = visited"""

implementation:
assert default in ( list , dict , set )
with self . _lock :
    if key is None :
        yield self . _shared_context
    else :
        sub_context = self . _shared_context . get ( key , None )
        if sub_context is None :
            sub_context = default ( )
        yield sub_context
        self . _shared_context [ key ] = sub_context


signature:
def enqueue ( self , state ) :
    """Enqueue state .
 Save state on storage , assigns an id to it , then add it to the
 priority queue"""

implementation:
# save the state to secondary storage
state_id = self . _workspace . save_state ( state )
self . put ( state_id )
self . _publish ( 'did_enqueue_state' , state_id , state )
return state_id


signature:
def put ( self , state_id ) :
    """Enqueue it for processing"""

implementation:
self . _states . append ( state_id )
self . _lock . notify_all ( )
return state_id


signature:
def get ( self ) :
    """Dequeue a state with the max priority"""

implementation:
# A shutdown has been requested
if self . is_shutdown ( ) :
    return None
# if not more states in the queue , let ' s wait for some forks
while len ( self . _states ) == 0 : # if no worker is running , bail out
    if self . running == 0 :
        return None
    # if a shutdown has been requested , bail out
    if self . is_shutdown ( ) :
        return None
    # if there ares actually some workers running , wait for state forks
    logger . debug ( "Waiting for available states" )
    self . _lock . wait ( )
state_id = self . _policy . choice ( list ( self . _states ) )
if state_id is None :
    return None
del self . _states [ self . _states . index ( state_id ) ]
return state_id


signature:
def fork ( self , state , expression , policy = 'ALL' , setstate = None ) :
    """Fork state on expression concretizations .
 Using policy build a list of solutions for expression .
 For the state on each solution setting the new state with setstate
 For example if expression is a Bool it may have 2 solutions . True or False .
 Parent
 ( expression = ? ? )
 Child1 Child2
 ( expression = True ) ( expression = True )
 setstate ( True ) setstate ( False )
 The optional setstate ( ) function is supposed to set the concrete value
 in the child state ."""

implementation:
assert isinstance ( expression , Expression )
if setstate is None :
    setstate = lambda x , y : None
# Find a set of solutions for expression
solutions = state . concretize ( expression , policy )
if not solutions :
    raise ExecutorError ( "Forking on unfeasible constraint set" )
if len ( solutions ) == 1 :
    setstate ( state , solutions [ 0 ] )
    return state
logger . info ( "Forking. Policy: %s. Values: %s" , policy , ', ' . join ( f'0x{sol:x}' for sol in solutions ) )
self . _publish ( 'will_fork_state' , state , expression , solutions , policy )
# Build and enqueue a state for each solution
children = [ ]
for new_value in solutions :
    with state as new_state :
        new_state . constrain ( expression == new_value )
        # and set the PC of the new state to the concrete pc - dest
        # ( or other register or memory address to concrete )
        setstate ( new_state , new_value )
        self . _publish ( 'did_fork_state' , new_state , expression , new_value , policy )
        # enqueue new _ state
        state_id = self . enqueue ( new_state )
        # maintain a list of children for logging purpose
        children . append ( state_id )
logger . info ( "Forking current state into states %r" , children )
return None


signature:
def run ( self ) :
    """Entry point of the Executor ; called by workers to start analysis ."""

implementation:
# policy _ order = self . policy _ order
# policy = self . policy
current_state = None
current_state_id = None
with WithKeyboardInterruptAs ( self . shutdown ) : # notify siblings we are about to start a run
    self . _notify_start_run ( )
    logger . debug ( "Starting Manticore Symbolic Emulator Worker (pid %d)." , os . getpid ( ) )
    solver = Z3Solver ( )
    while not self . is_shutdown ( ) :
        try : # handle fatal errors : exceptions in Manticore
            try : # handle external ( e . g . solver ) errors , and executor control exceptions
            # select a suitable state to analyze
                if current_state is None :
                    with self . _lock : # notify siblings we are about to stop this run
                        self . _notify_stop_run ( )
                        try : # Select a single state _ id
                            current_state_id = self . get ( )
                            # load selected state from secondary storage
                            if current_state_id is not None :
                                self . _publish ( 'will_load_state' , current_state_id )
                                current_state = self . _workspace . load_state ( current_state_id )
                                self . forward_events_from ( current_state , True )
                                self . _publish ( 'did_load_state' , current_state , current_state_id )
                                logger . info ( "load state %r" , current_state_id )
                            # notify siblings we have a state to play with
                        finally :
                            self . _notify_start_run ( )
                # If current _ state is still None . We are done .
                if current_state is None :
                    logger . debug ( "No more states in the queue, byte bye!" )
                    break
                assert current_state is not None
                assert current_state . constraints is current_state . platform . constraints
                # Allows to terminate manticore worker on user request
                while not self . is_shutdown ( ) :
                    if not current_state . execute ( ) :
                        break
                else : # Notify this worker is done
                    self . _publish ( 'will_terminate_state' , current_state , current_state_id , TerminateState ( 'Shutdown' ) )
                    current_state = None
            # Handling Forking and terminating exceptions
            except Concretize as e : # expression
            # policy
            # setstate ( )
                logger . debug ( "Generic state fork on condition" )
                current_state = self . fork ( current_state , e . expression , e . policy , e . setstate )
            except TerminateState as e : # Notify this worker is done
                self . _publish ( 'will_terminate_state' , current_state , current_state_id , e )
                logger . debug ( "Generic terminate state" )
                if e . testcase :
                    self . _publish ( 'internal_generate_testcase' , current_state , message = str ( e ) )
                current_state = None
            except SolverError as e : # raise
                import traceback
                trace = traceback . format_exc ( )
                logger . error ( "Exception: %s\n%s" , str ( e ) , trace )
                # Notify this state is done
                self . _publish ( 'will_terminate_state' , current_state , current_state_id , e )
                if solver . check ( current_state . constraints ) :
                    self . _publish ( 'internal_generate_testcase' , current_state , message = "Solver failed" + str ( e ) )
                current_state = None
        except ( Exception , AssertionError ) as e : # raise
            import traceback
            trace = traceback . format_exc ( )
            logger . error ( "Exception: %s\n%s" , str ( e ) , trace )
            # Notify this worker is done
            self . _publish ( 'will_terminate_state' , current_state , current_state_id , e )
            current_state = None
    assert current_state is None or self . is_shutdown ( )
    # notify siblings we are about to stop this run
    self . _notify_stop_run ( )


signature:
def linux ( cls , path , argv = None , envp = None , entry_symbol = None , symbolic_files = None , concrete_start = '' , pure_symbolic = False , stdin_size = None , ** kwargs ) :
    """Constructor for Linux binary analysis .
 : param str path : Path to binary to analyze
 : param argv : Arguments to provide to the binary
 : type argv : list [ str ]
 : param envp : Environment to provide to the binary
 : type envp : dict [ str , str ]
 : param entry _ symbol : Entry symbol to resolve to start execution
 : type envp : str
 : param symbolic _ files : Filenames to mark as having symbolic input
 : type symbolic _ files : list [ str ]
 : param str concrete _ start : Concrete stdin to use before symbolic input
 : param int stdin _ size : symbolic stdin size to use
 : param kwargs : Forwarded to the Manticore constructor
 : return : Manticore instance , initialized with a Linux State
 : rtype : Manticore"""

implementation:
if stdin_size is None :
    stdin_size = consts . stdin_size
try :
    return cls ( _make_linux ( path , argv , envp , entry_symbol , symbolic_files , concrete_start , pure_symbolic , stdin_size ) , ** kwargs )
except elftools . common . exceptions . ELFError :
    raise Exception ( f'Invalid binary: {path}' )


signature:
def decree ( cls , path , concrete_start = '' , ** kwargs ) :
    """Constructor for Decree binary analysis .
 : param str path : Path to binary to analyze
 : param str concrete _ start : Concrete stdin to use before symbolic input
 : param kwargs : Forwarded to the Manticore constructor
 : return : Manticore instance , initialized with a Decree State
 : rtype : Manticore"""

implementation:
try :
    return cls ( _make_decree ( path , concrete_start ) , ** kwargs )
except KeyError : # FIXME ( mark ) magic parsing for DECREE should raise better error
    raise Exception ( f'Invalid binary: {path}' )


signature:
def init ( self , f ) :
    """A decorator used to register a hook function to run before analysis begins . Hook
 function takes one : class : ` ~ manticore . core . state . State ` argument ."""

implementation:
def callback ( manticore_obj , state ) :
    f ( state )
self . subscribe ( 'will_start_run' , types . MethodType ( callback , self ) )
return f


signature:
def hook ( self , pc ) :
    """A decorator used to register a hook function for a given instruction address .
 Equivalent to calling : func : ` ~ add _ hook ` .
 : param pc : Address of instruction to hook
 : type pc : int or None"""

implementation:
def decorator ( f ) :
    self . add_hook ( pc , f )
    return f
return decorator


signature:
def add_hook ( self , pc , callback ) :
    """Add a callback to be invoked on executing a program counter . Pass ` None `
 for pc to invoke callback on every instruction . ` callback ` should be a callable
 that takes one : class : ` ~ manticore . core . state . State ` argument .
 : param pc : Address of instruction to hook
 : type pc : int or None
 : param callable callback : Hook function"""

implementation:
if not ( isinstance ( pc , int ) or pc is None ) :
    raise TypeError ( f"pc must be either an int or None, not {pc.__class__.__name__}" )
else :
    self . _hooks . setdefault ( pc , set ( ) ) . add ( callback )
    if self . _hooks :
        self . _executor . subscribe ( 'will_execute_instruction' , self . _hook_callback )


signature:
def _hook_callback ( self , state , pc , instruction ) :
    'Invoke all registered generic hooks'

implementation:
# Ignore symbolic pc .
# TODO ( yan ) : Should we ask the solver if any of the hooks are possible ,
# and execute those that are ?
if issymbolic ( pc ) :
    return
# Invoke all pc - specific hooks
for cb in self . _hooks . get ( pc , [ ] ) :
    cb ( state )
# Invoke all pc - agnostic hooks
for cb in self . _hooks . get ( None , [ ] ) :
    cb ( state )


signature:
def resolve ( self , symbol ) :
    """A helper method used to resolve a symbol name into a memory address when
 injecting hooks for analysis .
 : param symbol : function name to be resolved
 : type symbol : string
 : param line : if more functions present , optional line number can be included
 : type line : int or None"""

implementation:
with open ( self . binary_path , 'rb' ) as f :
    elffile = ELFFile ( f )
    # iterate over sections and identify symbol table section
    for section in elffile . iter_sections ( ) :
        if not isinstance ( section , SymbolTableSection ) :
            continue
        # get list of symbols by name
        symbols = section . get_symbol_by_name ( symbol )
        if not symbols :
            continue
        # return first indexed memory address for the symbol ,
        return symbols [ 0 ] . entry [ 'st_value' ]
    raise ValueError ( f"The {self.binary_path} ELFfile does not contain symbol {symbol}" )


signature:
def binary_arch ( binary ) :
    """helper method for determining binary architecture
 : param binary : str for binary to introspect .
 : rtype bool : True for x86_64 , False otherwise"""

implementation:
with open ( binary , 'rb' ) as f :
    elffile = ELFFile ( f )
    if elffile [ 'e_machine' ] == 'EM_X86_64' :
        return True
    else :
        return False


signature:
def binary_symbols ( binary ) :
    """helper method for getting all binary symbols with SANDSHREW _ prepended .
 We do this in order to provide the symbols Manticore should hook on to
 perform main analysis .
 : param binary : str for binary to instrospect .
 : rtype list : list of symbols from binary"""

implementation:
def substr_after ( string , delim ) :
    return string . partition ( delim ) [ 2 ]
with open ( binary , 'rb' ) as f :
    elffile = ELFFile ( f )
    for section in elffile . iter_sections ( ) :
        if not isinstance ( section , SymbolTableSection ) :
            continue
        symbols = [ sym . name for sym in section . iter_symbols ( ) if sym ]
        return [ substr_after ( name , PREPEND_SYM ) for name in symbols if name . startswith ( PREPEND_SYM ) ]


signature:
def get_group ( name : str ) -> _Group :
    """Get a configuration variable group named | name |"""

implementation:
global _groups
if name in _groups :
    return _groups [ name ]
group = _Group ( name )
_groups [ name ] = group
return group


signature:
def save ( f ) :
    """Save current config state to an yml file stream identified by | f |
 : param f : where to write the config file"""

implementation:
global _groups
c = { }
for group_name , group in _groups . items ( ) :
    section = { var . name : var . value for var in group . updated_vars ( ) }
    if not section :
        continue
    c [ group_name ] = section
yaml . safe_dump ( c , f , line_break = True )


signature:
def parse_config ( f ) :
    """Load an yml - formatted configuration from file stream | f |
 : param file f : Where to read the config ."""

implementation:
try :
    c = yaml . safe_load ( f )
    for section_name , section in c . items ( ) :
        group = get_group ( section_name )
        for key , val in section . items ( ) :
            group . update ( key )
            setattr ( group , key , val )
# Any exception here should trigger the warning ; from not being able to parse yaml
# to reading poorly formatted values
except Exception :
    raise ConfigError ( "Failed reading config file. Do you have a local [.]manticore.yml file?" )


signature:
def load_overrides ( path = None ) :
    """Load config overrides from the yml file at | path | , or from default paths . If a path
 is provided and it does not exist , raise an exception
 Default paths : . / mcore . yml , . / . mcore . yml , . / manticore . yml , . / . manticore . yml ."""

implementation:
if path is not None :
    names = [ path ]
else :
    possible_names = [ 'mcore.yml' , 'manticore.yml' ]
    names = [ os . path . join ( '.' , '' . join ( x ) ) for x in product ( [ '' , '.' ] , possible_names ) ]
for name in names :
    try :
        with open ( name , 'r' ) as yml_f :
            logger . info ( f'Reading configuration from {name}' )
            parse_config ( yml_f )
        break
    except FileNotFoundError :
        pass
else :
    if path is not None :
        raise FileNotFoundError ( f"'{path}' not found for config overrides" )


signature:
def add_config_vars_to_argparse ( args ) :
    """Import all defined config vars into | args | , for parsing command line .
 : param args : A container for argparse vars
 : type args : argparse . ArgumentParser or argparse . _ ArgumentGroup
 : return :"""

implementation:
global _groups
for group_name , group in _groups . items ( ) :
    for key in group :
        obj = group . _var_object ( key )
        args . add_argument ( f"--{group_name}.{key}" , type = type ( obj . default ) , default = obj . default , help = obj . description )


signature:
def process_config_values ( parser : argparse . ArgumentParser , args : argparse . Namespace ) :
    """Bring in provided config values to the args parser , and import entries to the config
 from all arguments that were actually passed on the command line
 : param parser : The arg parser
 : param args : The value that parser . parse _ args returned"""

implementation:
# First , load a local config file , if passed or look for one in pwd if it wasn ' t .
load_overrides ( args . config )
# Get a list of defined config vals . If these are passed on the command line ,
# update them in their correct group , not in the cli group
defined_vars = list ( get_config_keys ( ) )
command_line_args = vars ( args )
# Bring in the options keys into args
config_cli_args = get_group ( 'cli' )
# Place all command line args into the cli group ( for saving in the workspace ) . If
# the value is set on command line , then it takes precedence ; otherwise we try to
# read it from the config file ' s cli group .
for k in command_line_args :
    default = parser . get_default ( k )
    set_val = getattr ( args , k )
    if default is not set_val :
        if k not in defined_vars :
            config_cli_args . update ( k , value = set_val )
        else : # Update a var ' s native group
            group_name , key = k . split ( '.' )
            group = get_group ( group_name )
            setattr ( group , key , set_val )
    else :
        if k in config_cli_args :
            setattr ( args , k , getattr ( config_cli_args , k ) )


signature:
def add ( self , name : str , default = None , description : str = None ) :
    """Add a variable named | name | to this value group , optionally giving it a
 default value and a description .
 Variables must be added with this method before they can be set or read .
 Reading a variable replaces the variable that was defined previously , but
 updates the description if a new one is set ."""

implementation:
if name in self . _vars :
    raise ConfigError ( f"{self.name}.{name} already defined." )
if name == 'name' :
    raise ConfigError ( "'name' is a reserved name for a group." )
v = _Var ( name , description = description , default = default )
self . _vars [ name ] = v


signature:
def update ( self , name : str , value = None , default = None , description : str = None ) :
    """Like add , but can tolerate existing values ; also updates the value .
 Mostly used for setting fields from imported INI files and modified CLI flags ."""

implementation:
if name in self . _vars :
    description = description or self . _vars [ name ] . description
    default = default or self . _vars [ name ] . default
elif name == 'name' :
    raise ConfigError ( "'name' is a reserved name for a group." )
v = _Var ( name , description = description , default = default , defined = False )
v . value = value
self . _vars [ name ] = v


signature:
def get_description ( self , name : str ) -> str :
    """Return the description , or a help string of variable identified by | name | ."""

implementation:
if name not in self . _vars :
    raise ConfigError ( f"{self.name}.{name} not defined." )
return self . _vars [ name ] . description


signature:
def correspond ( text ) :
    """Communicate with the child process without closing stdin ."""

implementation:
subproc . stdin . write ( text )
subproc . stdin . flush ( )
return drain ( )


signature:
def function_signature_for_name_and_inputs ( name : str , inputs : Sequence [ Mapping [ str , Any ] ] ) -> str :
    """Returns the function signature for the specified name and Solidity JSON metadata inputs array .
 The ABI specification defines the function signature as the function name followed by the parenthesised list of
 parameter types separated by single commas and no spaces .
 See https : / / solidity . readthedocs . io / en / latest / abi - spec . html # function - selector"""

implementation: return name + SolidityMetadata . tuple_signature_for_components ( inputs )


signature:
def tuple_signature_for_components ( components : Sequence [ Mapping [ str , Any ] ] ) -> str :
    """Equivalent to ` ` function _ signature _ for _ name _ and _ inputs ( ' ' , components ) ` ` ."""

implementation:
ts = [ ]
for c in components :
    t : str = c [ 'type' ]
    if t . startswith ( 'tuple' ) :
        assert len ( t ) == 5 or t [ 5 ] == '['
        t = SolidityMetadata . tuple_signature_for_components ( c [ 'components' ] ) + t [ 5 : ]
    ts . append ( t )
return f'({",".join(ts)})'


signature:
def get_constructor_arguments ( self ) -> str :
    """Returns the tuple type signature for the arguments of the contract constructor ."""

implementation:
item = self . _constructor_abi_item
return '()' if item is None else self . tuple_signature_for_components ( item [ 'inputs' ] )


signature:
def get_source_for ( self , asm_offset , runtime = True ) :
    """Solidity source code snippet related to ` asm _ pos ` evm bytecode offset .
 If runtime is False , initialization bytecode source map is used"""

implementation:
srcmap = self . get_srcmap ( runtime )
try :
    beg , size , _ , _ = srcmap [ asm_offset ]
except KeyError : # asm _ offset pointing outside the known bytecode
    return ''
output = ''
nl = self . source_code [ : beg ] . count ( '\n' ) + 1
snippet = self . source_code [ beg : beg + size ]
for l in snippet . split ( '\n' ) :
    output += '    %s  %s\n' % ( nl , l )
    nl += 1
return output


signature:
def constructor_abi ( self ) -> Dict [ str , Any ] :
    """Returns a copy of the Solidity JSON ABI item for the contract constructor .
 The content of the returned dict is described at https : / / solidity . readthedocs . io / en / latest / abi - spec . html # json _"""

implementation:
item = self . _constructor_abi_item
if item :
    return dict ( item )
return { 'inputs' : [ ] , 'payable' : False , 'stateMutability' : 'nonpayable' , 'type' : 'constructor' }


signature:
def get_abi ( self , hsh : bytes ) -> Dict [ str , Any ] :
    """Returns a copy of the Solidity JSON ABI item for the function associated with the selector ` ` hsh ` ` .
 If no normal contract function has the specified selector , a dict describing the default or non - default
 fallback function is returned .
 The content of the returned dict is described at https : / / solidity . readthedocs . io / en / latest / abi - spec . html # json _"""

implementation:
if not isinstance ( hsh , ( bytes , bytearray ) ) :
    raise TypeError ( 'The selector argument must be a concrete byte array' )
sig = self . _function_signatures_by_selector . get ( hsh )
if sig is not None :
    return dict ( self . _function_abi_items_by_signature [ sig ] )
item = self . _fallback_function_abi_item
if item is not None :
    return dict ( item )
# An item describing the default fallback function .
return { 'payable' : False , 'stateMutability' : 'nonpayable' , 'type' : 'fallback' }


signature:
def get_func_argument_types ( self , hsh : bytes ) :
    """Returns the tuple type signature for the arguments of the function associated with the selector ` ` hsh ` ` .
 If no normal contract function has the specified selector ,
 the empty tuple type signature ` ` ' ( ) ' ` ` is returned ."""

implementation:
if not isinstance ( hsh , ( bytes , bytearray ) ) :
    raise TypeError ( 'The selector argument must be a concrete byte array' )
sig = self . _function_signatures_by_selector . get ( hsh )
return '()' if sig is None else sig [ sig . find ( '(' ) : ]


signature:
def get_func_return_types ( self , hsh : bytes ) -> str :
    """Returns the tuple type signature for the output values of the function
 associated with the selector ` ` hsh ` ` .
 If no normal contract function has the specified selector ,
 the empty tuple type signature ` ` ' ( ) ' ` ` is returned ."""

implementation:
if not isinstance ( hsh , ( bytes , bytearray ) ) :
    raise TypeError ( 'The selector argument must be a concrete byte array' )
abi = self . get_abi ( hsh )
outputs = abi . get ( 'outputs' )
return '()' if outputs is None else SolidityMetadata . tuple_signature_for_components ( outputs )


signature:
def get_func_name ( self , hsh : bytes ) -> str :
    """Returns the name of the normal function with the selector ` ` hsh ` ` ,
 or ` ` ' { fallback } ' ` ` if no such function exists ."""

implementation:
if not isinstance ( hsh , ( bytes , bytearray ) ) :
    raise TypeError ( 'The selector argument must be a concrete byte array' )
sig = self . _function_signatures_by_selector . get ( hsh )
return '{fallback}' if sig is None else sig [ : sig . find ( '(' ) ]


signature:
def get_func_signature ( self , hsh : bytes ) -> Optional [ str ] :
    """Returns the signature of the normal function with the selector ` ` hsh ` ` ,
 or ` ` None ` ` if no such function exists .
 This function returns ` ` None ` ` for any selector that will be dispatched to a fallback function ."""

implementation:
if not isinstance ( hsh , ( bytes , bytearray ) ) :
    raise TypeError ( 'The selector argument must be a concrete byte array' )
return self . _function_signatures_by_selector . get ( hsh )


signature:
def function_selectors ( self ) -> Iterable [ bytes ] :
    """The selectors of all normal contract functions ,
 plus ` ` self . fallback _ function _ selector ` ` if the contract has a non - default fallback function ."""

implementation:
selectors = self . _function_signatures_by_selector . keys ( )
if self . _fallback_function_abi_item is None :
    return tuple ( selectors )
return ( * selectors , self . fallback_function_selector )


signature:
def hashes ( self ) -> Tuple [ bytes , ... ] :
    """The selectors of all normal contract functions , plus ` ` self . fallback _ function _ selector ` ` ."""

implementation:
selectors = self . _function_signatures_by_selector . keys ( )
return ( * selectors , self . fallback_function_selector )


signature:
def convert_permissions ( m_perms ) :
    """Converts a Manticore permission string into a Unicorn permission
 : param m _ perms : Manticore perm string ( ' rwx ' )
 : return : Unicorn Permissions"""

implementation:
permissions = UC_PROT_NONE
if 'r' in m_perms :
    permissions |= UC_PROT_READ
if 'w' in m_perms :
    permissions |= UC_PROT_WRITE
if 'x' in m_perms :
    permissions |= UC_PROT_EXEC
return permissions


signature:
def hr_size ( num , suffix = 'B' ) -> str :
    """Human - readable data size
 From https : / / stackoverflow . com / a / 1094933
 : param num : number of bytes
 : param suffix : Optional size specifier
 : return : Formatted string"""

implementation:
for unit in ' KMGTPEZ' :
    if abs ( num ) < 1024.0 :
        return "%3.1f%s%s" % ( num , unit if unit != ' ' else '' , suffix )
    num /= 1024.0
return "%.1f%s%s" % ( num , 'Y' , suffix )


signature:
def copy_memory ( self , address , size ) :
    """Copy the bytes from address to address + size into Unicorn
 Used primarily for copying memory maps
 : param address : start of buffer to copy
 : param size : How many bytes to copy"""

implementation:
start_time = time . time ( )
map_bytes = self . _cpu . _raw_read ( address , size )
self . _emu . mem_write ( address , map_bytes )
if time . time ( ) - start_time > 3 :
    logger . info ( f"Copying {hr_size(size)} map at {hex(address)} took {time.time() - start_time} seconds" )


signature:
def map_memory_callback ( self , address , size , perms , name , offset , result ) :
    """Catches did _ map _ memory and copies the mapping into Manticore"""

implementation:
logger . info ( ' ' . join ( ( "Mapping Memory @" , hex ( address ) if type ( address ) is int else "0x??" , hr_size ( size ) , "-" , perms , "-" , f"{name}:{hex(offset) if name else ''}" , "->" , hex ( result ) ) ) )
self . _emu . mem_map ( address , size , convert_permissions ( perms ) )
self . copy_memory ( address , size )


signature:
def unmap_memory_callback ( self , start , size ) :
    """Unmap Unicorn maps when Manticore unmaps them"""

implementation:
logger . info ( f"Unmapping memory from {hex(start)} to {hex(start + size)}" )
mask = ( 1 << 12 ) - 1
if ( start & mask ) != 0 :
    logger . error ( "Memory to be unmapped is not aligned to a page" )
if ( size & mask ) != 0 :
    size = ( ( size >> 12 ) + 1 ) << 12
    logger . warning ( "Forcing unmap size to align to a page" )
self . _emu . mem_unmap ( start , size )


signature:
def protect_memory_callback ( self , start , size , perms ) :
    """Set memory protections in Unicorn correctly"""

implementation:
logger . info ( f"Changing permissions on {hex(start)}:{hex(start + size)} to {perms}" )
self . _emu . mem_protect ( start , size , convert_permissions ( perms ) )


signature:
def _hook_syscall ( self , uc , data ) :
    """Unicorn hook that transfers control to Manticore so it can execute the syscall"""

implementation:
logger . debug ( f"Stopping emulation at {hex(uc.reg_read(self._to_unicorn_id('RIP')))} to perform syscall" )
self . sync_unicorn_to_manticore ( )
from . . native . cpu . abstractcpu import Syscall
self . _to_raise = Syscall ( )
uc . emu_stop ( )


signature:
def _hook_write_mem ( self , uc , access , address , size , value , data ) :
    """Captures memory written by Unicorn"""

implementation:
self . _mem_delta [ address ] = ( value , size )
return True


signature:
def _hook_unmapped ( self , uc , access , address , size , value , data ) :
    """We hit an unmapped region ; map it into unicorn ."""

implementation:
try :
    self . sync_unicorn_to_manticore ( )
    logger . warning ( f"Encountered an operation on unmapped memory at {hex(address)}" )
    m = self . _cpu . memory . map_containing ( address )
    self . copy_memory ( m . start , m . end - m . start )
except MemoryException as e :
    logger . error ( "Failed to map memory {}-{}, ({}): {}" . format ( hex ( address ) , hex ( address + size ) , access , e ) )
    self . _to_raise = e
    self . _should_try_again = False
    return False
self . _should_try_again = True
return False


signature:
def emulate ( self , instruction ) :
    """Wrapper that runs the _ step function in a loop while handling exceptions"""

implementation:
# The emulation might restart if Unicorn needs to bring in a memory map
# or bring a value from Manticore state .
while True : # Try emulation
    self . _should_try_again = False
    self . _to_raise = None
    self . _step ( instruction )
    if not self . _should_try_again :
        break


signature:
def _step ( self , instruction , chunksize = 0 ) :
    """Execute a chunk fo instructions starting from instruction
 : param instruction : Where to start
 : param chunksize : max number of instructions to execute . Defaults to infinite ."""

implementation:
try :
    pc = self . _cpu . PC
    m = self . _cpu . memory . map_containing ( pc )
    if self . _stop_at :
        logger . info ( f"Emulating from {hex(pc)} to  {hex(self._stop_at)}" )
    self . _emu . emu_start ( pc , m . end if not self . _stop_at else self . _stop_at , count = chunksize )
except UcError : # We request re - execution by signaling error ; if we we didn ' t set
# _ should _ try _ again , it was likely an actual error
    if not self . _should_try_again :
        raise
if self . _should_try_again :
    return
# self . sync _ unicorn _ to _ manticore ( )
self . _cpu . PC = self . get_unicorn_pc ( )
if self . _cpu . PC == self . _stop_at :
    logger . info ( "Reached emulation target, switching to Manticore mode" )
    self . sync_unicorn_to_manticore ( )
    self . _stop_at = None
# Raise the exception from a hook that Unicorn would have eaten
if self . _to_raise :
    from . . native . cpu . abstractcpu import Syscall
    if type ( self . _to_raise ) is not Syscall :
        logger . info ( "Raising %s" , self . _to_raise )
    raise self . _to_raise
logger . info ( f"Exiting Unicorn Mode at {hex(self._cpu.PC)}" )
return


signature:
def sync_unicorn_to_manticore ( self ) :
    """Copy registers and written memory back into Manticore"""

implementation:
self . write_backs_disabled = True
for reg in self . registers :
    val = self . _emu . reg_read ( self . _to_unicorn_id ( reg ) )
    self . _cpu . write_register ( reg , val )
if len ( self . _mem_delta ) > 0 :
    logger . debug ( f"Syncing {len(self._mem_delta)} writes back into Manticore" )
for location in self . _mem_delta :
    value , size = self . _mem_delta [ location ]
    self . _cpu . write_int ( location , value , size * 8 )
self . write_backs_disabled = False
self . _mem_delta = { }


signature:
def write_back_memory ( self , where , expr , size ) :
    """Copy memory writes from Manticore back into Unicorn in real - time"""

implementation:
if self . write_backs_disabled :
    return
if type ( expr ) is bytes :
    self . _emu . mem_write ( where , expr )
else :
    if issymbolic ( expr ) :
        data = [ Operators . CHR ( Operators . EXTRACT ( expr , offset , 8 ) ) for offset in range ( 0 , size , 8 ) ]
        concrete_data = [ ]
        for c in data :
            if issymbolic ( c ) :
                c = chr ( solver . get_value ( self . _cpu . memory . constraints , c ) )
            concrete_data . append ( c )
        data = concrete_data
    else :
        data = [ Operators . CHR ( Operators . EXTRACT ( expr , offset , 8 ) ) for offset in range ( 0 , size , 8 ) ]
    logger . debug ( f"Writing back {hr_size(size // 8)} to {hex(where)}: {data}" )
    # TODO - the extra encoding is to handle null bytes output as strings when we concretize . That ' s probably a bug .
    self . _emu . mem_write ( where , b'' . join ( b . encode ( 'utf-8' ) if type ( b ) is str else b for b in data ) )


signature:
def write_back_register ( self , reg , val ) :
    """Sync register state from Manticore - > Unicorn"""

implementation:
if self . write_backs_disabled :
    return
if issymbolic ( val ) :
    logger . warning ( "Skipping Symbolic write-back" )
    return
if reg in self . flag_registers :
    self . _emu . reg_write ( self . _to_unicorn_id ( 'EFLAGS' ) , self . _cpu . read_register ( 'EFLAGS' ) )
    return
self . _emu . reg_write ( self . _to_unicorn_id ( reg ) , val )


signature:
def update_segment ( self , selector , base , size , perms ) :
    """Only useful for setting FS right now ."""

implementation:
logger . info ( "Updating selector %s to 0x%02x (%s bytes) (%s)" , selector , base , size , perms )
if selector == 99 :
    self . set_fs ( base )
else :
    logger . error ( "No way to write segment: %d" , selector )


signature:
def deprecated ( message : str ) :
    """A decorator for marking functions as deprecated ."""

implementation:
assert isinstance ( message , str ) , "The deprecated decorator requires a message string argument."
def decorator ( func ) :
    @ wraps ( func )
    def wrapper ( * args , ** kwargs ) :
        warnings . warn ( f"`{func.__qualname__}` is deprecated. {message}" , category = ManticoreDeprecationWarning , stacklevel = 2 )
        return func ( * args , ** kwargs )
    return wrapper
return decorator


signature:
def flip ( constraint ) :
    '''flips a constraint ( Equal )
 ( Equal ( BitVecITE Cond IfC ElseC ) IfC )
 ( Equal ( BitVecITE Cond IfC ElseC ) ElseC )'''

implementation:
equal = copy . copy ( constraint )
assert len ( equal . operands ) == 2
# assume they are the equal - > ite form that we produce on standard branches
ite , forcepc = equal . operands
assert isinstance ( ite , BitVecITE ) and isinstance ( forcepc , BitVecConstant )
assert len ( ite . operands ) == 3
cond , iifpc , eelsepc = ite . operands
assert isinstance ( iifpc , BitVecConstant ) and isinstance ( eelsepc , BitVecConstant )
equal . _operands = ( equal . operands [ 0 ] , eelsepc if forcepc . value == iifpc . value else iifpc )
return equal


signature:
def perm ( lst , func ) :
    '''Produce permutations of ` lst ` , where permutations are mutated by ` func ` . Used for flipping constraints . highly
 possible that returned constraints can be unsat this does it blindly , without any attention to the constraints
 themselves
 Considering lst as a list of constraints , e . g .
 [ C1 , C2 , C3 ]
 we ' d like to consider scenarios of all possible permutations of flipped constraints , excluding the original list .
 So we ' d like to generate :
 [ func ( C1 ) , C2 , C3 ] ,
 [ C1 , func ( C2 ) , C3 ] ,
 [ func ( C1 ) , func ( C2 ) , C3 ] ,
 [ C1 , C2 , func ( C3 ) ] ,
 . . etc
 This is effectively treating the list of constraints as a bitmask of width len ( lst ) and counting up , skipping the
 0th element ( unmodified array ) .
 The code below yields lists of constraints permuted as above by treating list indeces as bitmasks from 1 to
 2 * * len ( lst ) and applying func to all the set bit offsets .'''

implementation:
for i in range ( 1 , 2 ** len ( lst ) ) :
    yield [ func ( item ) if ( 1 << j ) & i else item for ( j , item ) in enumerate ( lst ) ]


signature:
def input_from_cons ( constupl , datas ) :
    'solve bytes in | datas | based on'

implementation:
def make_chr ( c ) :
    try :
        return chr ( c )
    except Exception :
        return c
newset = constraints_to_constraintset ( constupl )
ret = ''
for data in datas :
    for c in data :
        ret += make_chr ( solver . get_value ( newset , c ) )
return ret


signature:
def symbolic_run_get_cons ( trace ) :
    '''Execute a symbolic run that follows a concrete run ; return constraints generated
 and the stdin data produced'''

implementation:
m2 = Manticore . linux ( prog , workspace_url = 'mem:' )
f = Follower ( trace )
m2 . verbosity ( VERBOSITY )
m2 . register_plugin ( f )
def on_term_testcase ( mcore , state , stateid , err ) :
    with m2 . locked_context ( ) as ctx :
        readdata = [ ]
        for name , fd , data in state . platform . syscall_trace :
            if name in ( '_receive' , '_read' ) and fd == 0 :
                readdata . append ( data )
        ctx [ 'readdata' ] = readdata
        ctx [ 'constraints' ] = list ( state . constraints . constraints )
m2 . subscribe ( 'will_terminate_state' , on_term_testcase )
m2 . run ( )
constraints = m2 . context [ 'constraints' ]
datas = m2 . context [ 'readdata' ]
return constraints , datas


signature:
def seek ( self , offset , whence = os . SEEK_SET ) :
    """Repositions the file C { offset } according to C { whence } .
 Returns the resulting offset or - 1 in case of error .
 : rtype : int
 : return : the file offset ."""

implementation:
assert isinstance ( offset , int )
assert whence in ( os . SEEK_SET , os . SEEK_CUR , os . SEEK_END )
new_position = 0
if whence == os . SEEK_SET :
    new_position = offset
elif whence == os . SEEK_CUR :
    new_position = self . pos + offset
elif whence == os . SEEK_END :
    new_position = self . max_size + offset
if new_position < 0 :
    return - 1
self . pos = new_position
return self . pos


signature:
def read ( self , count ) :
    """Reads up to C { count } bytes from the file .
 : rtype : list
 : return : the list of symbolic bytes read"""

implementation:
if self . pos > self . max_size :
    return [ ]
else :
    size = min ( count , self . max_size - self . pos )
    ret = [ self . array [ i ] for i in range ( self . pos , self . pos + size ) ]
    self . pos += size
    return ret


signature:
def write ( self , data ) :
    """Writes the symbolic bytes in C { data } onto the file ."""

implementation:
size = min ( len ( data ) , self . max_size - self . pos )
for i in range ( self . pos , self . pos + size ) :
    self . array [ i ] = data [ i - self . pos ]


signature:
def empty_platform ( cls , arch ) :
    """Create a platform without an ELF loaded .
 : param str arch : The architecture of the new platform
 : rtype : Linux"""

implementation:
platform = cls ( None )
platform . _init_cpu ( arch )
platform . _init_std_fds ( )
return platform


signature:
def _execve ( self , program , argv , envp ) :
    """Load ` program ` and establish program state , such as stack and arguments .
 : param program str : The ELF binary to load
 : param argv list : argv array
 : param envp list : envp array"""

implementation:
argv = [ ] if argv is None else argv
envp = [ ] if envp is None else envp
logger . debug ( f"Loading {program} as a {self.arch} elf" )
self . load ( program , envp )
self . _arch_specific_init ( )
self . _stack_top = self . current . STACK
self . setup_stack ( [ program ] + argv , envp )
nprocs = len ( self . procs )
nfiles = len ( self . files )
assert nprocs > 0
self . running = list ( range ( nprocs ) )
# Each process can wait for one timeout
self . timers = [ None ] * nprocs
# each fd has a waitlist
self . rwait = [ set ( ) for _ in range ( nfiles ) ]
self . twait = [ set ( ) for _ in range ( nfiles ) ]
# Install event forwarders
for proc in self . procs :
    self . forward_events_from ( proc )


signature:
def _init_arm_kernel_helpers ( self ) :
    """ARM kernel helpers
 https : / / www . kernel . org / doc / Documentation / arm / kernel _ user _ helpers . txt"""

implementation:
page_data = bytearray ( b'\xf1\xde\xfd\xe7' * 1024 )
# Extracted from a RPi2
preamble = binascii . unhexlify ( 'ff0300ea' + '650400ea' + 'f0ff9fe5' + '430400ea' + '220400ea' + '810400ea' + '000400ea' + '870400ea' )
# XXX ( yan ) : The following implementations of cmpxchg and cmpxchg64 were
# handwritten to not use any exclusive instructions ( e . g . ldrexd ) or
# locking . For actual implementations , refer to
# arch / arm64 / kernel / kuser32 . S in the Linux source code .
__kuser_cmpxchg64 = binascii . unhexlify ( '30002de9' + # push { r4 , r5}
'08c09de5' + # ldr ip , [ sp , # 8]
'30009ce8' + # ldm ip , { r4 , r5}
'010055e1' + # cmp r5 , r1
'00005401' + # cmpeq r4 , r0
'0100a013' + # movne r0 , # 1
'0000a003' + # moveq r0 , # 0
'0c008c08' + # stmeq ip , { r2 , r3}
'3000bde8' + # pop { r4 , r5}
'1eff2fe1' # bx lr
)
__kuser_dmb = binascii . unhexlify ( '5bf07ff5' + # dmb ish
'1eff2fe1' # bx lr
)
__kuser_cmpxchg = binascii . unhexlify ( '003092e5' + # ldr r3 , [ r2]
'000053e1' + # cmp r3 , r0
'0000a003' + # moveq r0 , # 0
'00108205' + # streq r1 , [ r2]
'0100a013' + # movne r0 , # 1
'1eff2fe1' # bx lr
)
# Map a TLS segment
self . _arm_tls_memory = self . current . memory . mmap ( None , 4 , 'rw ' )
__kuser_get_tls = binascii . unhexlify ( '04009FE5' + # ldr r0 , [ pc , # 4]
'010090e8' + # ldm r0 , { r0}
'1eff2fe1' # bx lr
) + struct . pack ( '<I' , self . _arm_tls_memory )
tls_area = b'\x00' * 12
version = struct . pack ( '<I' , 5 )
def update ( address , code ) :
    page_data [ address : address + len ( code ) ] = code
# Offsets from Documentation / arm / kernel _ user _ helpers . txt in Linux
update ( 0x000 , preamble )
update ( 0xf60 , __kuser_cmpxchg64 )
update ( 0xfa0 , __kuser_dmb )
update ( 0xfc0 , __kuser_cmpxchg )
update ( 0xfe0 , __kuser_get_tls )
update ( 0xff0 , tls_area )
update ( 0xffc , version )
self . current . memory . mmap ( 0xffff0000 , len ( page_data ) , 'r x' , page_data )


signature:
def setup_stack ( self , argv , envp ) :
    """: param Cpu cpu : The cpu instance
 : param argv : list of parameters for the program to execute .
 : param envp : list of environment variables for the program to execute .
 http : / / www . phrack . org / issues . html ? issue = 58 & id = 5 # article
 position content size ( bytes ) + comment
 stack pointer - > [ argc = number of args ] 4
 [ argv [ 0 ] ( pointer ) ] 4 ( program name )
 [ argv [ 1 ] ( pointer ) ] 4
 [ argv [ . . ] ( pointer ) ] 4 * x
 [ argv [ n - 1 ] ( pointer ) ] 4
 [ argv [ n ] ( pointer ) ] 4 ( = NULL )
 [ envp [ 0 ] ( pointer ) ] 4
 [ envp [ 1 ] ( pointer ) ] 4
 [ envp [ . . ] ( pointer ) ] 4
 [ envp [ term ] ( pointer ) ] 4 ( = NULL )
 [ auxv [ 0 ] ( Elf32 _ auxv _ t ) ] 8
 [ auxv [ 1 ] ( Elf32 _ auxv _ t ) ] 8
 [ auxv [ . . ] ( Elf32 _ auxv _ t ) ] 8
 [ auxv [ term ] ( Elf32 _ auxv _ t ) ] 8 ( = AT _ NULL vector )
 [ padding ] 0 - 16
 [ argument ASCIIZ strings ] > = 0
 [ environment ASCIIZ str . ] > = 0
 (0xbfffffc ) [ end marker ] 4 ( = NULL )
 (0xc00000 ) < top of stack > 0 ( virtual )"""

implementation:
cpu = self . current
# In case setup _ stack ( ) is called again , we make sure we ' re growing the
# stack from the original top
cpu . STACK = self . _stack_top
auxv = self . auxv
logger . debug ( "Setting argv, envp and auxv." )
logger . debug ( f"\tArguments: {argv!r}" )
if envp :
    logger . debug ( "\tEnvironment:" )
    for e in envp :
        logger . debug ( f"\t\t{e!r}" )
logger . debug ( "\tAuxv:" )
for name , val in auxv . items ( ) :
    logger . debug ( f"\t\t{name}: 0x{val:x}" )
# We save the argument and environment pointers
argvlst = [ ]
envplst = [ ]
# end envp marker empty string
for evar in envp :
    cpu . push_bytes ( '\x00' )
    envplst . append ( cpu . push_bytes ( evar ) )
for arg in argv :
    cpu . push_bytes ( '\x00' )
    argvlst . append ( cpu . push_bytes ( arg ) )
# Put all auxv strings into the string stack area .
# And replace the value be its pointer
for name , value in auxv . items ( ) :
    if hasattr ( value , '__len__' ) :
        cpu . push_bytes ( value )
        auxv [ name ] = cpu . STACK
# The " secure execution " mode of secure _ getenv ( ) is controlled by the
# AT _ SECURE flag contained in the auxiliary vector passed from the
# kernel to user space .
auxvnames = { 'AT_IGNORE' : 1 , # Entry should be ignored
'AT_EXECFD' : 2 , # File descriptor of program
'AT_PHDR' : 3 , # Program headers for program
'AT_PHENT' : 4 , # Size of program header entry
'AT_PHNUM' : 5 , # Number of program headers
'AT_PAGESZ' : 6 , # System page size
'AT_BASE' : 7 , # Base address of interpreter
'AT_FLAGS' : 8 , # Flags
'AT_ENTRY' : 9 , # Entry point of program
'AT_NOTELF' : 10 , # Program is not ELF
'AT_UID' : 11 , # Real uid
'AT_EUID' : 12 , # Effective uid
'AT_GID' : 13 , # Real gid
'AT_EGID' : 14 , # Effective gid
'AT_CLKTCK' : 17 , # Frequency of times ( )
'AT_PLATFORM' : 15 , # String identifying platform .
'AT_HWCAP' : 16 , # Machine - dependent hints about processor capabilities .
'AT_FPUCW' : 18 , # Used FPU control word .
'AT_SECURE' : 23 , # Boolean , was exec setuid - like ?
'AT_BASE_PLATFORM' : 24 , # String identifying real platforms .
'AT_RANDOM' : 25 , # Address of 16 random bytes .
'AT_EXECFN' : 31 , # Filename of executable .
'AT_SYSINFO' : 32 , # Pointer to the global system page used for system calls and other nice things .
'AT_SYSINFO_EHDR' : 33 , # Pointer to the global system page used for system calls and other nice things .
}
# AT _ NULL
cpu . push_int ( 0 )
cpu . push_int ( 0 )
for name , val in auxv . items ( ) :
    cpu . push_int ( val )
    cpu . push_int ( auxvnames [ name ] )
# NULL ENVP
cpu . push_int ( 0 )
for var in reversed ( envplst ) : # ENVP n
    cpu . push_int ( var )
envp = cpu . STACK
# NULL ARGV
cpu . push_int ( 0 )
for arg in reversed ( argvlst ) : # Argv n
    cpu . push_int ( arg )
argv = cpu . STACK
# ARGC
cpu . push_int ( len ( argvlst ) )


signature:
def load ( self , filename , env ) :
    """Loads and an ELF program in memory and prepares the initial CPU state .
 Creates the stack and loads the environment variables and the arguments in it .
 : param filename : pathname of the file to be executed . ( used for auxv )
 : param list env : A list of env variables . ( used for extracting vars that control ld behavior )
 : raises error :
 - ' Not matching cpu ' : if the program is compiled for a different architecture
 - ' Not matching memory ' : if the program is compiled for a different address size
 : todo : define va _ randomize and read _ implies _ exec personality"""

implementation:
# load elf See binfmt _ elf . c
# read the ELF object file
cpu = self . current
elf = self . elf
arch = self . arch
env = dict ( var . split ( '=' ) for var in env if '=' in var )
addressbitsize = { 'x86' : 32 , 'x64' : 64 , 'ARM' : 32 } [ elf . get_machine_arch ( ) ]
logger . debug ( "Loading %s as a %s elf" , filename , arch )
assert elf . header . e_type in [ 'ET_DYN' , 'ET_EXEC' , 'ET_CORE' ]
# Get interpreter elf
interpreter = None
for elf_segment in elf . iter_segments ( ) :
    if elf_segment . header . p_type != 'PT_INTERP' :
        continue
    interpreter_filename = elf_segment . data ( ) [ : - 1 ]
    logger . info ( f'Interpreter filename: {interpreter_filename}' )
    if os . path . exists ( interpreter_filename . decode ( 'utf-8' ) ) :
        interpreter = ELFFile ( open ( interpreter_filename , 'rb' ) )
    elif 'LD_LIBRARY_PATH' in env :
        for mpath in env [ 'LD_LIBRARY_PATH' ] . split ( ":" ) :
            interpreter_path_filename = os . path . join ( mpath , os . path . basename ( interpreter_filename ) )
            logger . info ( f"looking for interpreter {interpreter_path_filename}" )
            if os . path . exists ( interpreter_path_filename ) :
                interpreter = ELFFile ( open ( interpreter_path_filename , 'rb' ) )
                break
    break
if interpreter is not None :
    assert interpreter . get_machine_arch ( ) == elf . get_machine_arch ( )
    assert interpreter . header . e_type in [ 'ET_DYN' , 'ET_EXEC' ]
# Stack Executability
executable_stack = False
for elf_segment in elf . iter_segments ( ) :
    if elf_segment . header . p_type != 'PT_GNU_STACK' :
        continue
    if elf_segment . header . p_flags & 0x01 :
        executable_stack = True
    else :
        executable_stack = False
    break
base = 0
elf_bss = 0
end_code = 0
end_data = 0
elf_brk = 0
self . load_addr = 0
for elf_segment in elf . iter_segments ( ) :
    if elf_segment . header . p_type != 'PT_LOAD' :
        continue
    align = 0x1000
    # elf _ segment . header . p _ align
    ELF_PAGEOFFSET = elf_segment . header . p_vaddr & ( align - 1 )
    flags = elf_segment . header . p_flags
    memsz = elf_segment . header . p_memsz + ELF_PAGEOFFSET
    offset = elf_segment . header . p_offset - ELF_PAGEOFFSET
    filesz = elf_segment . header . p_filesz + ELF_PAGEOFFSET
    vaddr = elf_segment . header . p_vaddr - ELF_PAGEOFFSET
    memsz = cpu . memory . _ceil ( memsz )
    if base == 0 and elf . header . e_type == 'ET_DYN' :
        assert vaddr == 0
        if addressbitsize == 32 :
            base = 0x56555000
        else :
            base = 0x555555554000
    perms = perms_from_elf ( flags )
    hint = base + vaddr
    if hint == 0 :
        hint = None
    logger . debug ( f"Loading elf offset: {offset:08x} addr:{base + vaddr:08x} {base + vaddr + memsz:08x} {perms}" )
    base = cpu . memory . mmapFile ( hint , memsz , perms , elf_segment . stream . name , offset ) - vaddr
    if self . load_addr == 0 :
        self . load_addr = base + vaddr
    k = base + vaddr + filesz
    if k > elf_bss :
        elf_bss = k
    if ( flags & 4 ) and end_code < k : # PF _ X
        end_code = k
    if end_data < k :
        end_data = k
    k = base + vaddr + memsz
    if k > elf_brk :
        elf_brk = k
elf_entry = elf . header . e_entry
if elf . header . e_type == 'ET_DYN' :
    elf_entry += self . load_addr
entry = elf_entry
real_elf_brk = elf_brk
# We need to explicitly clear bss , as fractional pages will have data from the file
bytes_to_clear = elf_brk - elf_bss
if bytes_to_clear > 0 :
    logger . debug ( f"Zeroing main elf fractional pages. From bss({elf_bss:x}) to brk({elf_brk:x}), {bytes_to_clear} bytes." )
    cpu . write_bytes ( elf_bss , '\x00' * bytes_to_clear , force = True )
stack_size = 0x21000
if addressbitsize == 32 :
    stack_top = 0xc0000000
else :
    stack_top = 0x800000000000
stack_base = stack_top - stack_size
stack = cpu . memory . mmap ( stack_base , stack_size , 'rwx' , name = 'stack' ) + stack_size
assert stack_top == stack
reserved = cpu . memory . mmap ( base + vaddr + memsz , 0x1000000 , '   ' )
interpreter_base = 0
if interpreter is not None :
    base = 0
    elf_bss = 0
    end_code = 0
    end_data = 0
    elf_brk = 0
    entry = interpreter . header . e_entry
    for elf_segment in interpreter . iter_segments ( ) :
        if elf_segment . header . p_type != 'PT_LOAD' :
            continue
        align = 0x1000
        # elf _ segment . header . p _ align
        vaddr = elf_segment . header . p_vaddr
        filesz = elf_segment . header . p_filesz
        flags = elf_segment . header . p_flags
        offset = elf_segment . header . p_offset
        memsz = elf_segment . header . p_memsz
        ELF_PAGEOFFSET = ( vaddr & ( align - 1 ) )
        memsz = memsz + ELF_PAGEOFFSET
        offset = offset - ELF_PAGEOFFSET
        filesz = filesz + ELF_PAGEOFFSET
        vaddr = vaddr - ELF_PAGEOFFSET
        memsz = cpu . memory . _ceil ( memsz )
        if base == 0 and interpreter . header . e_type == 'ET_DYN' :
            assert vaddr == 0
            total_size = self . _interp_total_size ( interpreter )
            base = stack_base - total_size
        if base == 0 :
            assert vaddr == 0
        perms = perms_from_elf ( flags )
        hint = base + vaddr
        if hint == 0 :
            hint = None
        base = cpu . memory . mmapFile ( hint , memsz , perms , elf_segment . stream . name , offset )
        base -= vaddr
        logger . debug ( f"Loading interpreter offset: {offset:08x} " f"addr:{base + vaddr:08x} " f"{base + vaddr + memsz:08x} " f"{(flags & 1 and 'r' or ' ')}" f"{(flags & 2 and 'w' or ' ')}" f"{(flags & 4 and 'x' or ' ')}" )
        k = base + vaddr + filesz
        if k > elf_bss :
            elf_bss = k
        if ( flags & 4 ) and end_code < k : # PF _ X
            end_code = k
        if end_data < k :
            end_data = k
        k = base + vaddr + memsz
        if k > elf_brk :
            elf_brk = k
    if interpreter . header . e_type == 'ET_DYN' :
        entry += base
    interpreter_base = base
    bytes_to_clear = elf_brk - elf_bss
    if bytes_to_clear > 0 :
        logger . debug ( f"Zeroing interpreter elf fractional pages. From bss({elf_bss:x}) to brk({elf_brk:x}), {bytes_to_clear} bytes." )
        cpu . write_bytes ( elf_bss , '\x00' * bytes_to_clear , force = True )
# free reserved brk space
cpu . memory . munmap ( reserved , 0x1000000 )
# load vdso
# vdso _ addr = load _ vdso ( addressbitsize )
cpu . STACK = stack
cpu . PC = entry
logger . debug ( f"Entry point: {entry:016x}" )
logger . debug ( f"Stack start: {stack:016x}" )
logger . debug ( f"Brk: {real_elf_brk:016x}" )
logger . debug ( f"Mappings:" )
for m in str ( cpu . memory ) . split ( '\n' ) :
    logger . debug ( f"  {m}" )
self . base = base
self . elf_bss = elf_bss
self . end_code = end_code
self . end_data = end_data
self . elf_brk = real_elf_brk
self . brk = real_elf_brk
at_random = cpu . push_bytes ( 'A' * 16 )
at_execfn = cpu . push_bytes ( f'{filename}\x00' )
self . auxv = { 'AT_PHDR' : self . load_addr + elf . header . e_phoff , # Program headers for program
'AT_PHENT' : elf . header . e_phentsize , # Size of program header entry
'AT_PHNUM' : elf . header . e_phnum , # Number of program headers
'AT_PAGESZ' : cpu . memory . page_size , # System page size
'AT_BASE' : interpreter_base , # Base address of interpreter
'AT_FLAGS' : elf . header . e_flags , # Flags
'AT_ENTRY' : elf_entry , # Entry point of program
'AT_UID' : 1000 , # Real uid
'AT_EUID' : 1000 , # Effective uid
'AT_GID' : 1000 , # Real gid
'AT_EGID' : 1000 , # Effective gid
'AT_CLKTCK' : 100 , # Frequency of times ( )
'AT_HWCAP' : 0 , # Machine - dependent hints about processor capabilities .
'AT_RANDOM' : at_random , # Address of 16 random bytes .
'AT_EXECFN' : at_execfn , # Filename of executable .
}


signature:
def _open ( self , f ) :
    """Adds a file descriptor to the current file descriptor list
 : rtype : int
 : param f : the file descriptor to add .
 : return : the index of the file descriptor in the file descr . list"""

implementation:
if None in self . files :
    fd = self . files . index ( None )
    self . files [ fd ] = f
else :
    fd = len ( self . files )
    self . files . append ( f )
return fd


signature:
def _close ( self , fd ) :
    """Removes a file descriptor from the file descriptor list
 : rtype : int
 : param fd : the file descriptor to close .
 : return : C { 0 } on success ."""

implementation:
try :
    self . files [ fd ] . close ( )
    self . _closed_files . append ( self . files [ fd ] )
    # Keep track for SymbolicFile testcase generation
    self . files [ fd ] = None
except IndexError :
    raise FdError ( f"Bad file descriptor ({fd})" )


signature:
def _is_fd_open ( self , fd ) :
    """Determines if the fd is within range and in the file descr . list
 : param fd : the file descriptor to check ."""

implementation: return fd >= 0 and fd < len ( self . files ) and self . files [ fd ] is not None


signature:
def sys_chdir ( self , path ) :
    """chdir - Change current working directory
 : param int path : Pointer to path"""

implementation:
path_str = self . current . read_string ( path )
logger . debug ( f"chdir({path_str})" )
try :
    os . chdir ( path_str )
    return 0
except OSError as e :
    return e . errno


signature:
def sys_getcwd ( self , buf , size ) :
    """getcwd - Get the current working directory
 : param int buf : Pointer to dest array
 : param size : size in bytes of the array pointed to by the buf
 : return : buf ( Success ) , or 0"""

implementation:
try :
    current_dir = os . getcwd ( )
    length = len ( current_dir ) + 1
    if size > 0 and size < length :
        logger . info ( "GETCWD: size is greater than 0, but is smaller than the length" "of the path + 1. Returning ERANGE" )
        return - errno . ERANGE
    if not self . current . memory . access_ok ( slice ( buf , buf + length ) , 'w' ) :
        logger . info ( "GETCWD: buf within invalid memory. Returning EFAULT" )
        return - errno . EFAULT
    self . current . write_string ( buf , current_dir )
    logger . debug ( f"getcwd(0x{buf:08x}, {size}) -> <{current_dir}> (Size {length})" )
    return length
except OSError as e :
    return - e . errno


signature:
def sys_lseek ( self , fd , offset , whence ) :
    """lseek - reposition read / write file offset
 The lseek ( ) function repositions the file offset of the open file description associated
 with the file descriptor fd to the argument offset according to the directive whence
 : param fd : a valid file descriptor
 : param offset : the offset in bytes
 : param whence : SEEK _ SET : The file offset is set to offset bytes .
 SEEK _ CUR : The file offset is set to its current location plus offset bytes .
 SEEK _ END : The file offset is set to the size of the file plus offset bytes .
 : return : offset from file beginning , or EBADF ( fd is not a valid file descriptor or is not open )"""

implementation:
signed_offset = self . _to_signed_dword ( offset )
try :
    return self . _get_fd ( fd ) . seek ( signed_offset , whence )
except FdError as e :
    logger . info ( ( "LSEEK: Not valid file descriptor on lseek." "Fd not seekable. Returning EBADF" ) )
    return - e . err


signature:
def sys_write ( self , fd , buf , count ) :
    """write - send bytes through a file descriptor
 The write system call writes up to count bytes from the buffer pointed
 to by buf to the file descriptor fd . If count is zero , write returns 0
 and optionally sets * tx _ bytes to zero .
 : param fd a valid file descriptor
 : param buf a memory buffer
 : param count number of bytes to send
 : return : 0 Success
 EBADF fd is not a valid file descriptor or is not open .
 EFAULT buf or tx _ bytes points to an invalid address ."""

implementation:
data : bytes = bytes ( )
cpu = self . current
if count != 0 :
    try :
        write_fd = self . _get_fd ( fd )
    except FdError as e :
        logger . error ( f"WRITE: Not valid file descriptor ({fd}). Returning -{e.err}" )
        return - e . err
    # TODO check count bytes from buf
    if buf not in cpu . memory or buf + count not in cpu . memory :
        logger . debug ( "WRITE: buf points to invalid address. Returning EFAULT" )
        return - errno . EFAULT
    if fd > 2 and write_fd . is_full ( ) :
        cpu . PC -= cpu . instruction . size
        self . wait ( [ ] , [ fd ] , None )
        raise RestartSyscall ( )
    data : MixedSymbolicBuffer = cpu . read_bytes ( buf , count )
    data : bytes = self . _transform_write_data ( data )
    write_fd . write ( data )
    for line in data . split ( b'\n' ) :
        line = line . decode ( 'latin-1' )
        # latin - 1 encoding will happily decode any byte ( 0x00-0xff )
        logger . debug ( f"WRITE({fd}, 0x{buf:08x}, {count}) -> <{repr(line):48s}>" )
    self . syscall_trace . append ( ( "_write" , fd , data ) )
    self . signal_transmit ( fd )
return len ( data )


signature:
def sys_access ( self , buf , mode ) :
    """Checks real user ' s permissions for a file
 : rtype : int
 : param buf : a buffer containing the pathname to the file to check its permissions .
 : param mode : the access permissions to check .
 : return :
 - C { 0 } if the calling process can access the file in the desired mode .
 - C { - 1 } if the calling process can not access the file in the desired mode ."""

implementation:
filename = b''
for i in range ( 0 , 255 ) :
    c = Operators . CHR ( self . current . read_int ( buf + i , 8 ) )
    if c == b'\x00' :
        break
    filename += c
if os . access ( filename , mode ) :
    return 0
else :
    return - 1


signature:
def sys_newuname ( self , old_utsname ) :
    """Writes system information in the variable C { old _ utsname } .
 : rtype : int
 : param old _ utsname : the buffer to write the system info .
 : return : C { 0 } on success"""

implementation:
from datetime import datetime
def pad ( s ) :
    return s + '\x00' * ( 65 - len ( s ) )
now = datetime ( 2017 , 8 , 0o1 ) . strftime ( "%a %b %d %H:%M:%S ART %Y" )
info = ( ( 'sysname' , 'Linux' ) , ( 'nodename' , 'ubuntu' ) , ( 'release' , '4.4.0-77-generic' ) , ( 'version' , '#98 SMP ' + now ) , ( 'machine' , self . _uname_machine ) , ( 'domainname' , '' ) )
uname_buf = '' . join ( pad ( pair [ 1 ] ) for pair in info )
self . current . write_bytes ( old_utsname , uname_buf )
return 0


signature:
def sys_brk ( self , brk ) :
    """Changes data segment size ( moves the C { brk } to the new address )
 : rtype : int
 : param brk : the new address for C { brk } .
 : return : the value of the new C { brk } .
 : raises error :
 - " Error in brk ! " if there is any error allocating the memory"""

implementation:
if brk != 0 and brk > self . elf_brk :
    mem = self . current . memory
    size = brk - self . brk
    if brk > mem . _ceil ( self . brk ) :
        perms = mem . perms ( self . brk - 1 )
        addr = mem . mmap ( mem . _ceil ( self . brk ) , size , perms )
        if not mem . _ceil ( self . brk ) == addr :
            logger . error ( f"Error in brk: ceil: {hex(mem._ceil(self.brk))} brk: {hex(brk)} self.brk: {hex(self.brk)} addr: {hex(addr)}" )
            return self . brk
    self . brk += size
return self . brk


signature:
def sys_arch_prctl ( self , code , addr ) :
    """Sets architecture - specific thread state
 : rtype : int
 : param code : must be C { ARCH _ SET _ FS } .
 : param addr : the base address of the FS segment .
 : return : C { 0 } on success
 : raises error :
 - if C { code } is different to C { ARCH _ SET _ FS }"""

implementation:
ARCH_SET_GS = 0x1001
ARCH_SET_FS = 0x1002
ARCH_GET_FS = 0x1003
ARCH_GET_GS = 0x1004
if code not in { ARCH_SET_GS , ARCH_SET_FS , ARCH_GET_FS , ARCH_GET_GS } :
    logger . debug ( "code not in expected options ARCH_GET/SET_FS/GS" )
    return - errno . EINVAL
if code != ARCH_SET_FS :
    raise NotImplementedError ( "Manticore supports only arch_prctl with code=ARCH_SET_FS (0x1002) for now" )
self . current . FS = 0x63
self . current . set_descriptor ( self . current . FS , addr , 0x4000 , 'rw' )
return 0


signature:
def sys_open ( self , buf , flags , mode ) :
    """: param buf : address of zero - terminated pathname
 : param flags : file access bits
 : param mode : file permission mode"""

implementation:
filename = self . current . read_string ( buf )
try :
    f = self . _sys_open_get_file ( filename , flags )
    logger . debug ( f"Opening file {filename} for real fd {f.fileno()}" )
except IOError as e :
    logger . warning ( f"Could not open file {filename}. Reason: {e!s}" )
    return - e . errno if e . errno is not None else - errno . EINVAL
return self . _open ( f )


signature:
def sys_openat ( self , dirfd , buf , flags , mode ) :
    """Openat SystemCall - Similar to open system call except dirfd argument
 when path contained in buf is relative , dirfd is referred to set the relative path
 Special value AT _ FDCWD set for dirfd to set path relative to current directory
 : param dirfd : directory file descriptor to refer in case of relative path at buf
 : param buf : address of zero - terminated pathname
 : param flags : file access bits
 : param mode : file permission mode"""

implementation:
filename = self . current . read_string ( buf )
dirfd = ctypes . c_int32 ( dirfd ) . value
if os . path . isabs ( filename ) or dirfd == self . FCNTL_FDCWD :
    return self . sys_open ( buf , flags , mode )
try :
    dir_entry = self . _get_fd ( dirfd )
except FdError as e :
    logger . info ( "openat: Not valid file descriptor. Returning EBADF" )
    return - e . err
if not isinstance ( dir_entry , Directory ) :
    logger . info ( "openat: Not directory descriptor. Returning ENOTDIR" )
    return - errno . ENOTDIR
dir_path = dir_entry . name
filename = os . path . join ( dir_path , filename )
try :
    f = self . _sys_open_get_file ( filename , flags )
    logger . debug ( f"Opening file {filename} for real fd {f.fileno()}" )
except IOError as e :
    logger . info ( f"Could not open file {filename}. Reason: {e!s}" )
    return - e . errno if e . errno is not None else - errno . EINVAL
return self . _open ( f )


signature:
def sys_rename ( self , oldnamep , newnamep ) :
    """Rename filename ` oldnamep ` to ` newnamep ` .
 : param int oldnamep : pointer to oldname
 : param int newnamep : pointer to newname"""

implementation:
oldname = self . current . read_string ( oldnamep )
newname = self . current . read_string ( newnamep )
ret = 0
try :
    os . rename ( oldname , newname )
except OSError as e :
    ret = - e . errno
return ret


signature:
def sys_fsync ( self , fd ) :
    """Synchronize a file ' s in - core state with that on disk ."""

implementation:
ret = 0
try :
    self . files [ fd ] . sync ( )
except IndexError :
    ret = - errno . EBADF
except FdError :
    ret = - errno . EINVAL
return ret


signature:
def sys_rt_sigaction ( self , signum , act , oldact ) :
    """Wrapper for sys _ sigaction"""

implementation: return self . sys_sigaction ( signum , act , oldact )


signature:
def sys_rt_sigprocmask ( self , cpu , how , newset , oldset ) :
    """Wrapper for sys _ sigprocmask"""

implementation: return self . sys_sigprocmask ( cpu , how , newset , oldset )


signature:
def sys_dup ( self , fd ) :
    """Duplicates an open file descriptor
 : rtype : int
 : param fd : the open file descriptor to duplicate .
 : return : the new file descriptor ."""

implementation:
if not self . _is_fd_open ( fd ) :
    logger . info ( "DUP: Passed fd is not open. Returning EBADF" )
    return - errno . EBADF
newfd = self . _dup ( fd )
return newfd


signature:
def sys_dup2 ( self , fd , newfd ) :
    """Duplicates an open fd to newfd . If newfd is open , it is first closed
 : rtype : int
 : param fd : the open file descriptor to duplicate .
 : param newfd : the file descriptor to alias the file described by fd .
 : return : newfd ."""

implementation:
try :
    file = self . _get_fd ( fd )
except FdError as e :
    logger . info ( "DUP2: Passed fd is not open. Returning EBADF" )
    return - e . err
soft_max , hard_max = self . _rlimits [ self . RLIMIT_NOFILE ]
if newfd >= soft_max :
    logger . info ( "DUP2: newfd is above max descriptor table size" )
    return - errno . EBADF
if self . _is_fd_open ( newfd ) :
    self . _close ( newfd )
if newfd >= len ( self . files ) :
    self . files . extend ( [ None ] * ( newfd + 1 - len ( self . files ) ) )
self . files [ newfd ] = self . files [ fd ]
logger . debug ( 'sys_dup2(%d,%d) -> %d' , fd , newfd , newfd )
return newfd


signature:
def sys_chroot ( self , path ) :
    """An implementation of chroot that does perform some basic error checking ,
 but does not actually chroot .
 : param path : Path to chroot"""

implementation:
if path not in self . current . memory :
    return - errno . EFAULT
path_s = self . current . read_string ( path )
if not os . path . exists ( path_s ) :
    return - errno . ENOENT
if not os . path . isdir ( path_s ) :
    return - errno . ENOTDIR
return - errno . EPERM


signature:
def sys_close ( self , fd ) :
    """Closes a file descriptor
 : rtype : int
 : param fd : the file descriptor to close .
 : return : C { 0 } on success ."""

implementation:
if self . _is_fd_open ( fd ) :
    self . _close ( fd )
else :
    return - errno . EBADF
logger . debug ( f'sys_close({fd})' )
return 0


signature:
def sys_readlink ( self , path , buf , bufsize ) :
    """Read
 : rtype : int
 : param path : the " link path id "
 : param buf : the buffer where the bytes will be putted .
 : param bufsize : the max size for read the link .
 : todo : Out eax number of bytes actually sent | EAGAIN | EBADF | EFAULT | EINTR | errno . EINVAL | EIO | ENOSPC | EPIPE"""

implementation:
if bufsize <= 0 :
    return - errno . EINVAL
filename = self . current . read_string ( path )
if filename == '/proc/self/exe' :
    data = os . path . abspath ( self . program )
else :
    data = os . readlink ( filename ) [ : bufsize ]
self . current . write_bytes ( buf , data )
return len ( data )


signature:
def sys_mmap_pgoff ( self , address , size , prot , flags , fd , offset ) :
    """Wrapper for mmap2"""

implementation: return self . sys_mmap2 ( address , size , prot , flags , fd , offset )


signature:
def sys_mmap ( self , address , size , prot , flags , fd , offset ) :
    """Creates a new mapping in the virtual address space of the calling process .
 : rtype : int
 : param address : the starting address for the new mapping . This address is used as hint unless the
 flag contains C { MAP _ FIXED } .
 : param size : the length of the mapping .
 : param prot : the desired memory protection of the mapping .
 : param flags : determines whether updates to the mapping are visible to other
 processes mapping the same region , and whether updates are carried
 through to the underlying file .
 : param fd : the contents of a file mapping are initialized using C { size } bytes starting at
 offset C { offset } in the file referred to by the file descriptor C { fd } .
 : param offset : the contents of a file mapping are initialized using C { size } bytes starting at
 offset C { offset } in the file referred to by the file descriptor C { fd } .
 : return :
 - C { - 1 } in case you use C { MAP _ FIXED } in the flags and the mapping can not be place at the desired address .
 - the address of the new mapping ( that must be the same as address in case you included C { MAP _ FIXED } in flags ) .
 : todo : handle exception ."""

implementation:
if address == 0 :
    address = None
cpu = self . current
if flags & 0x10 :
    cpu . memory . munmap ( address , size )
perms = perms_from_protflags ( prot )
if flags & 0x20 :
    result = cpu . memory . mmap ( address , size , perms )
elif fd == 0 :
    assert offset == 0
    result = cpu . memory . mmap ( address , size , perms )
    data = self . files [ fd ] . read ( size )
    cpu . write_bytes ( result , data )
else : # FIXME Check if file should be symbolic input and do as with fd0
    result = cpu . memory . mmapFile ( address , size , perms , self . files [ fd ] . name , offset )
actually_mapped = f'0x{result:016x}'
if address is None or result != address :
    address = address or 0
    actually_mapped += f' [requested: 0x{address:016x}]'
if flags & 0x10 != 0 and result != address :
    cpu . memory . munmap ( result , size )
    result = - 1
return result


signature:
def sys_mprotect ( self , start , size , prot ) :
    """Sets protection on a region of memory . Changes protection for the calling process ' s
 memory page ( s ) containing any part of the address range in the interval [ C { start } , C { start } + C { size } - 1 ] .
 : rtype : int
 : param start : the starting address to change the permissions .
 : param size : the size of the portion of memory to change the permissions .
 : param prot : the new access permission for the memory .
 : return : C { 0 } on success ."""

implementation:
perms = perms_from_protflags ( prot )
ret = self . current . memory . mprotect ( start , size , perms )
return 0


signature:
def sys_munmap ( self , addr , size ) :
    """Unmaps a file from memory . It deletes the mappings for the specified address range
 : rtype : int
 : param addr : the starting address to unmap .
 : param size : the size of the portion to unmap .
 : return : C { 0 } on success ."""

implementation:
self . current . memory . munmap ( addr , size )
return 0


signature:
def sys_readv ( self , fd , iov , count ) :
    """Works just like C { sys _ read } except that data is read into multiple buffers .
 : rtype : int
 : param fd : the file descriptor of the file to read .
 : param iov : the buffer where the the bytes to read are stored .
 : param count : amount of C { iov } buffers to read from the file .
 : return : the amount of bytes read in total ."""

implementation:
cpu = self . current
ptrsize = cpu . address_bit_size
sizeof_iovec = 2 * ( ptrsize // 8 )
total = 0
for i in range ( 0 , count ) :
    buf = cpu . read_int ( iov + i * sizeof_iovec , ptrsize )
    size = cpu . read_int ( iov + i * sizeof_iovec + ( sizeof_iovec // 2 ) , ptrsize )
    data = self . files [ fd ] . read ( size )
    total += len ( data )
    cpu . write_bytes ( buf , data )
    self . syscall_trace . append ( ( "_read" , fd , data ) )
return total