Python __builtin__ 模块，next() 实例源码

def search(cls, **type):
        """Search through all of the functions within the database and return the first result.
        Please review the help for functions.list for the definition of ``type``.
        """
        query_s = ', '.join("{:s}={!r}".format(k,v) for k,v in type.iteritems())

        res = __builtin__.list(cls.iterate(**type))
        if len(res) > 1:
            __builtin__.map(logging.info, (("[{:d}] {:s}".format(i, function.name(ea))) for i,ea in enumerate(res)))
            f = utils.compose(function.by,function.name)
            logging.warn("{:s}.search({:s}) : Found {:d} matching results, returning the first one. : {!r}".format('.'.join((__name__, cls.__name__)), query_s, len(res), f(res[0])))

        res = __builtin__.next(iter(res), None)
        if res is None:
            raise LookupError("{:s}.search({:s}) : Found 0 matching results.".format('.'.join((__name__, cls.__name__)), query_s))
        return res

def disasm(ea, **options):
    """Disassemble the instructions at the address ``ea``.
    If the integer ``count`` is specified, then return ``count`` number of instructions.
    If the bool ``comments`` is True, then return the comments for each instruction as well.
    """
    ea = interface.address.inside(ea)

    res,count = [], options.get('count',1)
    while count > 0:
        insn = idaapi.generate_disasm_line(ea)
        unformatted = idaapi.tag_remove(insn)
        nocomment = unformatted[:unformatted.rfind(';')] if ';' in unformatted and not options.get('comments',False) else unformatted
        res.append("{:x}: {:s}".format(ea, reduce(lambda t,x: t + (('' if t.endswith(' ') else ' ') if x == ' ' else x), nocomment, '')) )
        ea = address.next(ea)
        count -= 1
    return '\n'.join(res)

def search(cls, **type):
        """Search through all of the names within the database and return the first result.
        Please review the help for names.list for the definition of ``type``.
        """
        query_s = ', '.join("{:s}={!r}".format(k,v) for k,v in type.iteritems())

        res = __builtin__.list(cls.__iterate__(**type))
        if len(res) > 1:
            __builtin__.map(logging.info, (("[{:d}] {:x} {:s}".format(idx, idaapi.get_nlist_ea(idx), idaapi.get_nlist_name(idx))) for idx in res))
            f1, f2 = idaapi.get_nlist_ea, idaapi.get_nlist_name
            logging.warn("{:s}.search({:s}) : Found {:d} matching results, returning the first one. : {:x} {!r}".format('.'.join((__name__, cls.__name__)), query_s, len(res), f1(res[0]), f2(res[0])))

        res = __builtin__.next(iter(res), None)
        if res is None:
            raise LookupError("{:s}.search({:s}) : Found 0 matching results.".format('.'.join((__name__, cls.__name__)), query_s))
        return idaapi.get_nlist_ea(res)

def blocks(start, end):
    '''Returns each block between the addresses ``start`` and ``end``.'''
    block, _ = start, end = interface.address.head(start), address.tail(end)+1
    for ea in iterate(start, end):
        nextea = address.next(ea)

        if _instruction.is_call(ea):
            continue

        if _instruction.is_return(ea):
            yield block,nextea
            block = ea

        elif cxdown(ea):
            yield block,nextea
            block = nextea

        elif cxup(ea) and block != ea:
            yield block,ea
            block = ea
        continue
    return

# FIXME: The idaapi.is_basic_block_end api has got to be faster than doing it
#        with ida's xrefs in python..

def search(cls, **type):
        """Search through all of the entry-points within the database and return the first result.
        Please review the help for entry.list for the definition of ``type``.
        """
        query_s = ', '.join("{:s}={!r}".format(k,v) for k,v in type.iteritems())

        res = __builtin__.list(cls.__iterate__(**type))
        if len(res) > 1:
            __builtin__.map(logging.info, (("[{:d}] {:x} : ({:x}) {:s}".format(idx, cls.__address__(idx), cls.__entryordinal__(idx), cls.__entryname__(idx))) for idx in res))
            f = utils.compose(idaapi.get_entry_ordinal, idaapi.get_entry)
            logging.warn("{:s}.search({:s}) : Found {:d} matching results, returning the first one. : {:x}".format('.'.join((__name__,cls.__name__)), query_s, len(res), f(res[0])))

        res = __builtin__.next(iter(res), None)
        if res is None:
            raise LookupError("{:s}.search({:s}) : Found 0 matching results.".format('.'.join((__name__,cls.__name__)), query_s))
        return cls.__address__(res)

def search(cls, **type):
        """Search through all of the imports within the database and return the first result.
        Please review the help for imports.list for the definition of ``type``.
        """
        query_s = ', '.join("{:s}={!r}".format(k,v) for k,v in type.iteritems())

        res = __builtin__.list(cls.iterate(**type))
        if len(res) > 1:
            __builtin__.map(logging.info, ("{:x} {:s}<{:d}> {:s}".format(ea, module, ordinal, name) for ea,(module,name,ordinal) in res))
            f = utils.compose(utils.second, cls.__formatl__)
            logging.warn("{:s}.search({:s}) : Found {:d} matching results, returning the first one. : {!r}".format('.'.join((__name__,cls.__name__)), query_s, len(res), f(res[0])))

        res = __builtin__.next(iter(res), None)
        if res is None:
            raise LookupError("{:s}.search({:s}) : Found 0 matching results.".format('.'.join((__name__,cls.__name__)), query_s))
        return res[0]

def handle_nextarg(self, span, name):
        '''Should be called to signalize a nextarg directive.

        Args:
            span (tuple of int): Start and end line of the directive.
            name (str or None): Name of the argument following next or
                None if it should be the next positional argument.
        '''
        self._check_for_open_block(span, 'nextarg')
        block = self._open_blocks[-1]
        directive, fname, spans = block[0:3]
        self._check_if_matches_last(
            directive, 'call', spans[-1], span, 'nextarg')
        args, argnames = block[5:7]
        args.append(self._curnode)
        spans.append(span)
        if name is not None:
            argnames.append(name)
        elif argnames:
            msg = 'non-keyword argument following keyword argument'
            raise FyppFatalError(msg, fname, span)
        self._curnode = []

def next(itr, default=_undef):
        """compat wrapper for next()"""
        if default is _undef:
            return itr.next()
        try:
            return itr.next()
        except StopIteration:
            return default

def iterate(start, end, step=None):
    '''Iterate through all of the instruction and data boundaries from address ``start`` to ``end``.'''
    step = step or (address.prev if start > end else address.next)
    start, end = __builtin__.map(interface.address.head, (start, end))
    op = operator.gt if start > end else operator.lt
    while start != idaapi.BADADDR and op(start,end):
        yield start
        start = step(start)

    _, right = config.bounds()
    if end < right: yield end

def __index__(cls, ea):
        '''Returns the index of the entry-point at the specified ``address``.'''
        f = utils.compose(idaapi.get_entry_ordinal, idaapi.get_entry)
        iterable = itertools.imap(utils.compose(utils.fap(f, lambda n:n), __builtin__.tuple), __builtin__.range(idaapi.get_entry_qty()))
        filterable = itertools.ifilter(utils.compose(utils.first, functools.partial(operator.eq, ea)), iterable)
        result = itertools.imap(utils.second, filterable)
        return __builtin__.next(result, None)

def new(cls, name):
        '''Adds an entry-point to the database with the ``name`` using the next available index as the ordinal.'''
        return cls.new(ui.current.address(), name, idaapi.get_entry_qty())

def get(cls, ea):
        '''Return the import at the address ``ea``.'''
        ea = interface.address.inside(ea)
        res = itertools.ifilter(utils.compose(utils.first, functools.partial(operator.eq, ea)), cls.__iterate__())
        try:
            return utils.second(__builtin__.next(res))
        except StopIteration:
            pass
        raise LookupError("{:s}.get({:x}) : Unable to determine import at specified address.".format('.'.join((__name__, cls.__name__)), ea))

def iterate(cls):
        '''Return an iterator that walks forward through the database from the current address.'''
        return cls.iterate(ui.current.address(), cls.next)

def iterate(cls, ea):
        '''Return an iterator that walks forward through the database starting at the  address ``ea``.'''
        return cls.iterate(ea, cls.next)

def iterate(cls, ea, next):
        '''Return an iterator that walks through the database starting at the address ``ea``. Use ``next`` to determine the next address.'''
        ea = interface.address.inside(ea)
        while ea not in (None,idaapi.BADADDR):
            yield ea
            ea = next(ea)
        return

def next(cls):
        '''Return the next defined address from the current one.'''
        return cls.next(ui.current.address(), 1)

def next(cls, ea):
        '''Return the next defined address from the address ``ea``.'''
        return cls.next(ea, 1)

def nextdata(cls):
        '''Returns the next address that has data referencing it.'''
        return cls.nextdata(ui.current.address(), 1)

def nextdata(cls, ea):
        '''Returns the next address from ``ea`` that has data referencing it.'''
        return cls.nextdata(ea, 1)

def nextdata(cls, ea, count):
        '''Returns the next address from ``ea`` that has data referencing it. Skip ``count`` results before returning.'''
        res = cls.walk(ea, cls.next, lambda n: len(xref.data_up(n)) == 0)
        return cls.nextdata(cls.next(res), count-1) if count > 1 else res

def nextcode(cls):
        '''Returns the next address that has code referencing it.'''
        return cls.nextcode(ui.current.address(), 1)

def nextcode(cls, ea):
        '''Returns the next address from ``ea`` that has code referencing it.'''
        return cls.nextcode(ea, 1)

def nextref(cls):
        '''Returns the next address that has anything referencing it.'''
        return cls.nextref(ui.current.address(), 1)

def nextref(cls, ea):
        '''Returns the next address from ``ea`` that has anything referencing it.'''
        return cls.nextref(ea, 1)

def nextref(cls, ea, count):
        '''Returns the next address from ``ea`` that has anything referencing it. Skip ``count`` references before returning.'''
        res = cls.walk(ea, cls.next, lambda n: len(xref.up(n)) == 0)
        return cls.nextref(cls.next(res), count-1) if count > 1 else res

def nextreg(cls, ea, reg, *regs, **modifiers):
        regs = (reg,) + regs
        count = modifiers.get('count',1)
        args = ', '.join(["{:x}".format(ea)] + __builtin__.map("\"{:s}\"".format, regs) + __builtin__.map(utils.unbox("{:s}={!r}".format), modifiers.items()))

        # generate each helper using the regmatch class
        iterops = interface.regmatch.modifier(**modifiers)
        uses_register = interface.regmatch.use(regs)

        # if within a function, then sure we're within the chunk's bounds.
        if function.within(ea):
            (_,end) = function.chunk(ea)
            fwithin = functools.partial(operator.gt, end)

        # otherwise ensure that we're not in a function and we're a code type.
        else:
            fwithin = utils.compose(utils.fap(utils.compose(function.within, operator.not_), type.is_code), all)

            end = cls.walk(ea, cls.next, fwithin)
            end = bottom() if end == idaapi.BADADDR else end

        # define a function for cls.walk to continue looping while
        F = lambda ea: fwithin(ea) and not any(uses_register(ea, opnum) for opnum in iterops(ea))

        # skip the current address
        nextea = cls.next(ea)
        if nextea is None:
            # FIXME: include registers in message
            logging.fatal("{:s}.nextreg({:s}) : Unable to start walking from next address. : {:x}".format('.'.join((__name__, cls.__name__)), args, ea))
            return ea

        # now walk while none of our registers match
        res = cls.walk(nextea, cls.next, F)
        if res == idaapi.BADADDR or (cls == address and res >= end):
            # FIXME: include registers in message
            raise ValueError("{:s}.nextreg({:s}, ...) : Unable to find register{:s} within chunk {:x}:{:x} : {:x}".format('.'.join((__name__, cls.__name__)), args, ('s','')[len(regs)>1], end, ea, res))

        # recurse if the user specified it
        modifiers['count'] = count - 1
        return cls.nextreg(cls.next(res), *regs, **modifiers) if count > 1 else res

def nextreg(cls, reg, *regs, **modifiers):
        '''Return the next address containing an instruction that uses one of the specified registers ``regs``.'''
        return cls.nextreg(ui.current.address(), reg, *regs, **modifiers)

def nextstack(cls, ea, delta):
        '''Return the next instruction from ``ea`` that is past the sp delta ``delta``.'''
        fn,sp = function.top(ea), function.get_spdelta(ea)
        _,end = function.chunk(ea)
        res = cls.walk(ea, cls.next, lambda ea: ea < end and abs(function.get_spdelta(ea) - sp) < delta)
        if res == idaapi.BADADDR or res >= end:
            raise ValueError("{:s}.nextstack({:x}, {:+x}) : Unable to locate instruction matching contraints due to walking outside the bounds of the function {:x} : {:x} >= {:x}".format('.'.join((__name__,cls.__name__)), ea, delta, fn, res, end))
        return res

def nextcall(cls):
        '''Return the next call instruction.'''
        return cls.nextcall(ui.current.address(), 1)

def nextcall(cls, ea):
        '''Return the next call instruction from the address ``ea``.'''
        return cls.nextcall(ea, 1)

def nextcall(cls, ea, count):
        res = cls.walk(ea, cls.next, lambda n: not _instruction.is_call(n))
        return cls.nextcall(cls.next(res), count-1) if count > 1 else res

def nextbranch(cls):
        '''Return the next branch instruction.'''
        return cls.nextbranch(ui.current.address(), 1)

def nextbranch(cls, ea, count):
        res = cls.walk(ea, cls.next, lambda n: _instruction.is_call(n) or not _instruction.is_branch(n))
        return cls.nextbranch(cls.next(res), count-1) if count > 1 else res

def nexttag(cls, **tagname):
        '''Return the next address that contains a tag.'''
        return cls.nexttag(ui.current.address(), 1, **tagname)

def nexttag(cls, ea, **tagname):
        """Returns the next address from ``ea`` that contains a tag.
        If the str ``tag`` is specified, then only return the address if the specified tag is defined.
        """
        return cls.nexttag(ea, 1, **tagname)

def nexttag(cls, ea, count, **tagname):
        tagname = tagname.get('tagname', None)
        res = cls.walk(ea, cls.next, lambda n: not (type.has_comment(n) if tagname is None else tagname in tag(n)))
        return cls.nextbranch(cls.next(res), count-1) if count > 1 else res

def walk(ea, next, match):
        '''Used internally. Please see .iterate() instead.'''
        ea = interface.address.inside(ea)
        res = __builtin__.set()
        while ea not in (None,idaapi.BADADDR) and is_code(ea) and ea not in res and match(ea):
            res.add(ea)
            ea = next(ea)
        return ea

def next(cls, ea):
        '''Emulate the instruction at ``ea`` and return the next address that would be executed.'''
        return cls.next(ea, 1)

def next(cls, ea, count):
        ea = interface.address.within(ea)
        isStop = lambda ea: _instruction.feature(ea) & idaapi.CF_STOP == idaapi.CF_STOP
        invalidQ = utils.compose(utils.fap(utils.compose(type.is_code, operator.not_), isStop), any)
        refs = filter(type.is_code, xref.down(ea))
        if len(refs) > 1:
            logging.fatal("{:s}.next({:x}, count={:d}) : Unable to determine next address due to multiple xrefs being available : {:s}".format('.'.join((__name__, cls.__name__)), ea, count, ', '.join(__builtin__.map("{:x}".format,refs))))
            return None
        if invalidQ(ea) and not _instruction.is_jmp(ea):
#            logging.fatal("{:s}.next({:x}, count={:d}) : Unable to move to next address. Flow has stopped.".format('.'.join((__name__, cls.__name__)), ea, count))
            return None
        res = refs[0] if _instruction.is_jmp(ea) else address.next(ea)
        return cls.next(res, count-1) if count > 1 else res

def iterate(ea, start, next):
        ea = interface.address.inside(ea)
        ea = ea if type.flags(ea, idaapi.FF_DATA) else idaapi.prev_head(ea,0)

        addr = start(ea)
        while addr != idaapi.BADADDR:
            yield addr
            addr = next(ea, addr)
        return

def code(ea, descend):
        """Return all of the code xrefs that refer to the address ``ea``.
        If the bool ``descend`` is defined, then return only code refs that are referred by the specified address.
        """
        if descend:
            start,next = idaapi.get_first_cref_from, idaapi.get_next_cref_from
        else:
            start,next = idaapi.get_first_cref_to, idaapi.get_next_cref_to

        for addr in xref.iterate(ea, start, next):
            yield addr
        return

def data(ea, descend):
        """Return all of the data xrefs that refer to the address ``ea``.
        If the bool ``descend`` is defined, then return only the data refs that are referred by the specified address.
        """
        if descend:
            start,next = idaapi.get_first_dref_from, idaapi.get_next_dref_from
        else:
            start,next = idaapi.get_first_dref_to, idaapi.get_next_dref_to

        for addr in xref.iterate(ea, start, next):
            yield addr
        return

def find_slotaddress(cls, ea):
            # FIXME: figure out how to fail if this address isn't found
            res = itertools.islice(itertools.count(), cls.MAX_SLOT_COUNT)
            res, iterable = itertools.tee(itertools.imap(cls.get_slotaddress, res))
            try:
                count = len(__builtin__.list(itertools.takewhile(lambda n: n != ea, res)))
            except IndexError:
                raise KeyError("{:s}.find_slotaddress({:x}) : Unable to find specified slot address.".format('.'.join((__name__,cls.__name__)), ea))
            __builtin__.list(itertools.islice(iterable, count))
            if __builtin__.next(iterable) != ea:
                raise KeyError("{:s}.find_slotaddress({:x}) : Unable to find specified slot address.".format('.'.join((__name__,cls.__name__)), ea))
            return count

def find_slotaddress(cls, ea):
            res = itertools.islice(itertools.count(), cls.MAX_SLOT_COUNT)
            res, iterable = itertools.tee(itertools.imap(cls.get_slotaddress, res))
            try:
                count = len(__builtin__.list(itertools.takewhile(lambda n: n != ea, res)))
            except IndexError:
                raise KeyError("{:s}.find_slotaddress({:x}) : Unable to find specified slot address.".format('.'.join((__name__,cls.__name__)), ea))
            __builtin__.list(itertools.islice(iterable, count))
            if __builtin__.next(iterable) != ea:
                raise KeyError("{:s}.find_slotaddress({:x}) : Unable to find specified slot address.".format('.'.join((__name__,cls.__name__)), ea))
            return count

def free_slotindex(cls):
            res = __builtin__.next((i for i in xrange(cls.MAX_SLOT_COUNT) if idaapi.get_marked_pos(i) == idaapi.BADADDR), None)
            if res is None:
                raise ValueError("{:s}.free_slotindex : No free slots available for mark.".format('.'.join((__name__, 'bookmarks', cls.__name__))))
            return res

def align(cls, ea, **alignment):
        '''Set the data at address ``ea`` as aligned with the specified ``alignment``.'''
        if not type.is_unknown(ea):
            raise TypeError("{:s}.set.align({:x}, ...) : Data at specified address has already been defined.".format('.'.join((__name__,cls.__name__)), ea))
        res = alignment.get('alignment', alignment.get('size', address.next(ea)-ea))
        return cls.data(ea, res, type=idaapi.FF_ALIGN)

def handle_nextarg(self, span, name):
        '''Called when parser encounters a nextarg directive.

        It is a dummy method and should be overriden for actual use.

        Args:
            span (tuple of int): Start and end line of the directive.
            name (str or None): Name of the argument following next or
                None if it should be the next positional argument.
        '''
        self._log_event('nextarg', span, name=name)

def _postprocess_eval_line(self, evalline, fname, span):
        lines = evalline.split('\n')
        # If line ended on '\n', last element is ''. We remove it and
        # add the trailing newline later manually.
        trailing_newline = (lines[-1] == '')
        if trailing_newline:
            del lines[-1]
        lnum = linenumdir(span[0], fname) if self._linenums else ''
        clnum = lnum if self._contlinenums else ''
        linenumsep = '\n' + lnum
        clinenumsep = '\n' + clnum
        foldedlines = [self._foldline(line) for line in lines]
        outlines = [clinenumsep.join(lines) for lines in foldedlines]
        result = linenumsep.join(outlines)
        # Add missing trailing newline
        if trailing_newline:
            trailing = '\n'
            if self._linenums:
                # Last line was folded, but no linenums were generated for
                # the continuation lines -> current line position is not
                # in sync with the one calculated from the last line number
                unsync = (
                    len(foldedlines) and len(foldedlines[-1]) > 1
                    and not self._contlinenums)
                # Eval directive in source consists of more than one line
                multiline = span[1] - span[0] > 1
                if unsync or multiline:
                    # For inline eval directives span[0] == span[1]
                    # -> next line is span[0] + 1 and not span[1] as for
                    # line eval directives
                    nextline = max(span[1], span[0] + 1)
                    trailing += linenumdir(nextline, fname)
        else:
            trailing = ''
        return result + trailing

def next(itr, default=_undef):
        "compat wrapper for next()"
        if default is _undef:
            return itr.next()
        try:
            return itr.next()
        except StopIteration:
            return default

def next(itr, default=_undef):
        """compat wrapper for next()"""
        if default is _undef:
            return itr.next()
        try:
            return itr.next()
        except StopIteration:
            return default