pythonsegmentation-faultctypespython-c-api

check if assessing an address will cause a segfault without crashing python


What I've tried:

  1. faulthandler is really useful to get a traceback where segfault occurred but it doesn't allow handling it properly.
import faulthandler
faulthandler.enable()

import ctypes
try:
    # Windows fatal exception: access violation
    # Current thread 0x00001334 (most recent call first):
    # File "c:\Users\Andrej\Desktop\pomoika\test.py", line 6 in <module>
    ctypes.c_byte.from_address(0).value
    print('never printed')
except:
    print('never printed')
  1. setting up a handler using singal.signal - it worked, as mentioned in the docs and in this question, handler's python code is never executed as handling segfault also causes a segfault recursively.
import ctypes
import signal

class AccessError(Exception):
    pass

def handler(signum, frame):
    raise AccessError("Memory access violation")

def is_safe(address: int) -> bool:
    try:
        # Set signal handler for segmentation faults
        signal.signal(signal.SIGSEGV, handler)
        v = ctypes.c_uint.from_address(address)
        a = v.value
        return True
    except AccessError:
        return False
    finally:
        # Reset signal handler to default behavior
        signal.signal(signal.SIGSEGV, signal.SIG_DFL)

# Test the function
print('before')
print(is_safe(0))  # Should print False or True depending on the address
print('after')
  1. Checking it in a separate thread. It kind of works but on Windows it returns False, True, False, so I guess it's not cross-platform.
import ctypes
import multiprocessing


def check_address(queue, address):
    try:
        ctypes.c_byte.from_address(address).value
        queue.put(True)
    except Exception:
        queue.put(False)


def is_safe(address: int, timeout: float = 1.0) -> bool:
    queue = multiprocessing.Queue()
    process = multiprocessing.Process(target=check_address, args=(queue, address))
    process.start()
    process.join(timeout)
    if process.exitcode is None:
        process.terminate()
        raise Exception(f"Process is stuck (it took longer than {timeout}).")
    elif process.exitcode == 0:
        process.terminate()
        process.join()
        v = queue.get()
        return v
    return False


if __name__ == "__main__":
    print(0, is_safe(0)) # False
    print("id(object)", is_safe(id(object))) # True
    a = object()
    print("a", is_safe(id(a))) # True

Solution

  • The simplest solution I've found is using os.write. I wonder if there's a simpler way to do it without setting up os.pipe and closing it later.

    import ctypes
    import os
    
    
    def is_safe(address: int, /) -> bool:
        # made by @chilaxan
        if address <= 0:
            return False
        r, w = os.pipe()
        try:
            return os.write(w, ctypes.c_char.from_address(address)) == 1
        except OSError as e:
            # [Errno 22] Invalid argument
            return False
        finally:
            os.close(r)
            os.close(w)
    
    
    if __name__ == "__main__":
        print(0, is_safe(0))  # False
        print(1, is_safe(1))  # False
        print("id(object)", is_safe(id(object)))  # True
        a = object()
        print("a", is_safe(id(a)))  # True