Python signal 模块，SIGSEGV 实例源码

def test_run_async_exit_code(self):

        def run_with_exit_code_0(process_idx):
            sys.exit(0)

        def run_with_exit_code_11(process_idx):
            os.kill(os.getpid(), signal.SIGSEGV)

        with warnings.catch_warnings(record=True) as w:
            async.run_async(4, run_with_exit_code_0)
            # There should be no warnings
            assert len(w) == 0

        with warnings.catch_warnings(record=True) as w:
            async.run_async(4, run_with_exit_code_11)
            # There should be 4 warnings
            assert len(w) == 4

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def _default_handler(signum, *args):
        ''' The default signal handler. Don't register with built-in
        signal.signal! This needs to be used on the subprocess await
        death workaround.
        '''
        # All valid cpython windows signals
        sigs = {
            signal.SIGABRT: SIGABRT,
            # signal.SIGFPE: 'fpe', # Don't catch this
            # signal.SIGSEGV: 'segv', # Don't catch this
            # signal.SIGILL: 'illegal', # Don't catch this
            signal.SIGINT: SIGINT,
            signal.SIGTERM: SIGTERM,
            # Note that signal.CTRL_C_EVENT and signal.CTRL_BREAK_EVENT are
            # converted to SIGINT in _await_signal
        }

        try:
            exc = sigs[signum]
        except KeyError:
            exc = DaemonikerSignal

        _sketch_raise_in_main(exc)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        checked = set()
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows.
            # Issue #18396, only for signals without a C-level handler.
            if signal.getsignal(sig) is not None:
                signal.signal(sig, signal.signal(sig, handler))
                checked.add(sig)
        # Issue #18396: Ensure the above loop at least tested *something*
        self.assertTrue(checked)

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def _is_sigsegv(self, return_code):
        """
        Check return code against SIGSEGV
        """
        if return_code == -signal.SIGSEGV:
            return True
        return False

def _analyze(self):
        if self.signum == SIGSEGV:
            self.memoryFault()
        elif self.signum == SIGFPE:
            self.mathError()
        elif self.signum == SIGCHLD:
            self.childExit()
        elif self.signum == SIGABRT:
            self.error = Abort()
        return self.error

def _run_trace(self, stdout_file=None):
        """
        accumulate a basic block trace using qemu
        """

        timeout = 0.05
        if len(self.binaries) > 1:
            timeout = 0.25

        args  = ["timeout", "-k", str(timeout), str(timeout)]
        args += [os.path.join(self.base_dir, "bin", "fakesingle")]
        if self.use_alt_flag:
            args += ["-s", self.SEED_ALT]
        else:
            args += ["-s", self.SEED]
        args += self.binaries

        with open('/dev/null', 'wb') as devnull:
            stdout_f = devnull
            if stdout_file is not None:
                stdout_f = open(stdout_file, 'wb')

            l.debug("tracing as raw input")
            p = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=stdout_f, stderr=devnull)
            _, _ = p.communicate(self.payload)

            ret = p.wait()
            self.returncode = p.returncode

            # did a crash occur?
            if ret < 0 or ret == 139:
                if abs(ret) == signal.SIGSEGV or abs(ret) == signal.SIGILL or ret == 139:
                    l.info("input caused a crash (signal %d) during dynamic tracing", abs(ret))
                    l.debug("entering crash mode")
                    self.crash_mode = True

            if stdout_file is not None:
                stdout_f.close()

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def setup():
    for sig in (signal.SIGABRT, signal.SIGILL, signal.SIGINT, signal.SIGSEGV, signal.SIGTERM):
        signal.signal(sig, cleanup)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def crashes(self, signals=(signal.SIGSEGV, signal.SIGILL)):
        """
        Retrieve the crashes discovered by AFL. Since we are now detecting flag
        page leaks (via SIGUSR1) we will not return these leaks as crashes.
        Instead, these 'crashes' can be found with the leaks function.

        :param signals: list of valid kill signal numbers to override the default (SIGSEGV and SIGILL)
        :return: a list of strings which are crashing inputs
        """

        return self._get_crashing_inputs(signals)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def test_issue9324(self):
        # Updated for issue #10003, adding SIGBREAK
        handler = lambda x, y: None
        for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
                    signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
                    signal.SIGTERM):
            # Set and then reset a handler for signals that work on windows
            signal.signal(sig, signal.signal(sig, handler))

        with self.assertRaises(ValueError):
            signal.signal(-1, handler)

        with self.assertRaises(ValueError):
            signal.signal(7, handler)

def init():
    # exit handlers
    for sig in (SIGABRT, SIGILL, SIGINT, SIGSEGV, SIGTERM):
        signal(sig, cleanup)

def main():
    parser = argparse.ArgumentParser(description='Mock tool with control over its output & termination.')
    parser.add_argument('--print-args', action='store_true', default=False,
                        help='print the (non-option) arguments')
    parser.add_argument('--print-env', metavar='VAR', type=str, default=None,
                        help='print an environment variable')
    parser.add_argument('--echo-stdin', action='store_true', default=False,
                        help='echo the standard input')
    parser.add_argument('--to-stderr', action='store_true', default=False,
                        help='write to standard error instead of standard output')
    parser.add_argument('--crash', action='store_true', default=False,
                        help='crash process after output')
    parser.add_argument('--exit-code', metavar='N', type=int, default=0,
                        help='terminate process with given exit code (default: %(default)s)')
    parser.add_argument('args', nargs='*',
                        help='arbitrary command line arguments')
    args = parser.parse_args()

    out = sys.stderr if args.to_stderr else sys.stdout

    if args.print_args:
        for arg in args.args:
            print(arg, file=out, flush=True)

    if args.print_env is not None:
        print(os.getenv(args.print_env, ''), file=out, flush=True)

    if args.echo_stdin:
        for line in sys.stdin:
            print(line, file=out, end='', flush=True)

    if args.crash:
        os.kill(os.getpid(), signal.SIGSEGV)

    sys.exit(args.exit_code)

def _analyze(self):
        if self.signum in (SIGSEGV, SIGBUS):
            self.memoryFault()
        elif self.signum == SIGFPE:
            self.mathError()
        elif self.signum == SIGCHLD:
            self.childExit()
        elif self.signum == SIGABRT:
            self.reason = Abort()
        return self.reason

def platformProcessEvent(self, event):
        """
        Handle a mach exception message
        """
        #if self.attaching:
            #self.useptrace = True
            #return self.handlePosixSignal(event)

        #self.useptrace = False

        # Some event states that need to be reset
        self.softexc = False

        threadid, excode, codes = event

        # Set the thread that signaled.
        self.setMeta('ThreadId', threadid)
        self.setMeta('StoppedThreadId', threadid)

        self.setMeta('MachException', event)

        if excode == EXC_SOFTWARE:

            self.softexc = True

            assert( len(codes) == 2 )
            assert( codes[0] == EXC_SOFT_SIGNAL )

            sig = codes[1]
            self.handlePosixSignal(sig)

        elif excode == EXC_BAD_ACCESS:
            print('exc_bad_access',repr([hex(x) for x in codes ]))
            signo = signal.SIGSEGV
            #if codes[0] == KERN_INVALID_ADDRESS:
                #signo = signal.SIGBUS

            self._fireSignal(signo)

        elif excode == EXC_BAD_INSTRUCTION:
            print('exc_bad_instruction',repr([hex(x) for x in codes ]))
            self._fireSignal(signal.SIGILL)

        elif excode == EXC_CRASH:
            print('exc_crash')
            print('Crash:',repr([hex(x) for x in codes]))
            self._fireExit(0xffffffff)

        elif excode == EXC_BREAKPOINT:
            print('exc_breakpoint',codes)
            self.handlePosixSignal(signal.SIGTRAP)

        else:
            print('Unprocessed Exception Type: %d' % excode)
            self.fireNotifiers(vtrace.NOTIFY_SIGNAL)

        return

def do_monitor(args):
    """ Handle "monitor" mode.
    """

    # If we aren't running in monitor mode, then we are done.
    if not args.monitor:
        return

    # This is the main retry loop.
    while True:
        # Fork the process.
        logger.info('Forking child process.')
        pid = os.fork()

        # If we are the child, leave this function and work.
        if pid == 0:
            logger.debug('We are a newly spawned child process.')
            return

        logger.debug('Child process spawned: %d', pid)

        # Wait for the child to die. If we die first, kill the child.
        atexit.register(kill_process, pid)
        try:
            _, exit_status = os.waitpid(pid, 0)
        except KeyboardInterrupt:
            break
        atexit.unregister(kill_process)

        # Process the exit code.
        signal_number = exit_status & 0xFF
        exit_code = (exit_status >> 8) & 0xFF
        core_dump = bool(0x80 & signal_number)

        if signal_number == 0:
            logger.info('Child process exited with exit code: %d.', exit_code)
        else:
            logger.info('Child process exited with signal %d (core dump: %s).',
                signal_number, core_dump)

        retry = False
        if os.WIFSIGNALED(exit_status):
            if os.WTERMSIG(exit_status) == signal.SIGSEGV:
                logger.error('Child process seg faulted.')
                retry = True

        if not retry:
            break

    sys.exit(0)

###############################################################################

def do_monitor(args):
    """ Handle "monitor" mode.
    """

    # If we aren't running in monitor mode, then we are done.
    if not args.monitor:
        return

    # This is the main retry loop.
    while True:
        # Fork the process.
        logger.info('Forking child process.')
        pid = os.fork()

        # If we are the child, leave this function and work.
        if pid == 0:
            logger.debug('We are a newly spawned child process.')
            return

        logger.debug('Child process spawned: %d', pid)

        # Wait for the child to die. If we die first, kill the child.
        atexit.register(kill_process, pid)
        try:
            _, exit_status = os.waitpid(pid, 0)
        except KeyboardInterrupt:
            break
        atexit.unregister(kill_process)

        # Process the exit code.
        signal_number = exit_status & 0xFF
        exit_code = (exit_status >> 8) & 0xFF
        core_dump = bool(0x80 & signal_number)

        if signal_number == 0:
            logger.info('Child process exited with exit code: %d.', exit_code)
        else:
            logger.info('Child process exited with signal %d (core dump: %s).',
                signal_number, core_dump)

        retry = False
        if os.WIFSIGNALED(exit_status):
            if os.WTERMSIG(exit_status) == signal.SIGSEGV:
                logger.error('Child process seg faulted.')
                retry = True

        if not retry:
            break

    sys.exit(0)

###############################################################################