Python signal 模块，SIGILL 实例源码

def startProcess(self, iworker, testQueue, resultQueue, shouldStop, result):
        currentaddr = Value('c',bytes_(''))
        currentstart = Value('d',time.time())
        keyboardCaught = Event()
        p = Process(target=runner,
                   args=(iworker, testQueue,
                         resultQueue,
                         currentaddr,
                         currentstart,
                         keyboardCaught,
                         shouldStop,
                         self.loaderClass,
                         result.__class__,
                         pickle.dumps(self.config)))
        p.currentaddr = currentaddr
        p.currentstart = currentstart
        p.keyboardCaught = keyboardCaught
        old = signal.signal(signal.SIGILL, signalhandler)
        p.start()
        signal.signal(signal.SIGILL, old)
        return p

def test_catch_signals():
    print = lambda *args: None
    orig = signal.getsignal(signal.SIGILL)
    print(orig)
    with catch_signals([signal.SIGILL]) as queue:
        # Raise it a few times, to exercise signal coalescing, both at the
        # call_soon level and at the SignalQueue level
        signal_raise(signal.SIGILL)
        signal_raise(signal.SIGILL)
        await _core.wait_all_tasks_blocked()
        signal_raise(signal.SIGILL)
        await _core.wait_all_tasks_blocked()
        async for batch in queue:  # pragma: no branch
            assert batch == {signal.SIGILL}
            break
        signal_raise(signal.SIGILL)
        async for batch in queue:  # pragma: no branch
            assert batch == {signal.SIGILL}
            break
    with pytest.raises(RuntimeError):
        await queue.__anext__()
    assert signal.getsignal(signal.SIGILL) is orig

def startProcess(self, iworker, testQueue, resultQueue, shouldStop, result):
        currentaddr = Value('c',bytes_(''))
        currentstart = Value('d',time.time())
        keyboardCaught = Event()
        p = Process(target=runner,
                   args=(iworker, testQueue,
                         resultQueue,
                         currentaddr,
                         currentstart,
                         keyboardCaught,
                         shouldStop,
                         self.loaderClass,
                         result.__class__,
                         pickle.dumps(self.config)))
        p.currentaddr = currentaddr
        p.currentstart = currentstart
        p.keyboardCaught = keyboardCaught
        old = signal.signal(signal.SIGILL, signalhandler)
        p.start()
        signal.signal(signal.SIGILL, old)
        return p

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 kill_process_by_id(pid):
    pid = int(pid)
    os.kill(pid, signal.SIGILL)
    os.waitpid(pid, 0)

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 kill_process(pid):
    '??????'
    os.kill(pid, signal.SIGILL)

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