win32

SimSystemWin32

SimSystemWin32(
    stdin=None,
    stdout=None,
    stderr=None,
    fd=None,
    sockets=None,
    socket_queue=None,
    argv=None,
    argc=None,
    environ=None,
    auxv=None,
    tls_modules=None,
    sigmask=None,
    pid=None,
    ppid=None,
    uid=None,
    gid=None,
    brk=None,
)

Bases: SimStatePlugin

Data storage and interaction mechanisms for states with an environment conforming to win32. Available as state.win32.

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def __init__(self,
        stdin=None,
        stdout=None,
        stderr=None,
        fd=None,
        sockets=None,
        socket_queue=None,
        argv=None,
        argc=None,
        environ=None,
        auxv=None,
        tls_modules=None,
        sigmask=None,
        pid=None,
        ppid=None,
        uid=None,
        gid=None,
        brk=None):
    super().__init__()

    # some limits and constants
    self.sigmask_bits = 1024
    self.maximum_symbolic_syscalls = 255
    self.max_length = 2 ** 16

    self.argc = argc
    self.argv = argv
    self.environ = environ
    self.auxv = auxv
    self.tls_modules = tls_modules if tls_modules is not None else {}
    self.brk = brk if brk is not None else 0x1b00000
    self._sigmask = sigmask
    self.pid = 1337 if pid is None else pid
    self.ppid = 1336 if ppid is None else ppid
    self.uid = 1000 if uid is None else uid
    self.gid = 1000 if gid is None else gid
    self.dev_fs = None
    self.proc_fs = None
    self.autotmp_counter = 0
    self._closed_fds = []

    self.sockets = sockets if sockets is not None else {}
    self.socket_queue = socket_queue if socket_queue is not None else []

    if stdin is None:
        stdin = SimPacketsStream('stdin', write_mode=False, writable=False, ident='stdin')
    if stdout is None:
        stdout = SimPacketsStream('stdout', write_mode=True, writable=True, ident='stdout')
    if stderr is None:
        stderr = SimPacketsStream('stderr', write_mode=True, writable=True, ident='stderr')

    if fd is None:
        fd = {}
        tty = SimFileDescriptorDuplex(stdin, stdout)

        # the initial fd layout just looks like this:
        # lrwx------ 1 audrey audrey 64 Jan 17 14:21 0 -> /dev/pts/4
        # lrwx------ 1 audrey audrey 64 Jan 17 14:21 1 -> /dev/pts/4
        # lrwx------ 1 audrey audrey 64 Jan 17 14:21 2 -> /dev/pts/4
        # but we want to distinguish the streams. we compromise by having 0 and 1 go to the "tty"
        # and stderr goes to a special stderr file
        fd[0] = tty
        fd[1] = tty
        fd[2] = SimFileDescriptor(stderr, 0)

    self.fd = fd
    # these are the storage mechanisms!
    self.stdin = stdin
    self.stdout = stdout
    self.stderr = stderr

close

close(fd)

Closes the given file descriptor (an AST). Returns whether the operation succeeded (a concrete boolean)

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def close(self, fd):
    """
    Closes the given file descriptor (an AST).
    Returns whether the operation succeeded (a concrete boolean)
    """
    try:
        fd = self.state.solver.eval_one(fd)
    except SimSolverError:
        l.error("Trying to close a symbolic file descriptor")
        return False

    if fd not in self.fd:
        l.info("Trying to close an unopened file descriptor")
        return False

    self.state.history.add_event('fs_close', fd=fd, close_idx=len(self.closed_fds))
    self.closed_fds.append((fd, self.fd[fd]))

    del self.fd[fd]
    return True

dump_file_by_path

dump_file_by_path(path, **kwargs)

Returns the concrete content for a file by path.

:param path: file path as string :param kwargs: passed to state.solver.eval :return: file contents as string

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def dump_file_by_path(self, path, **kwargs):
    """
    Returns the concrete content for a file by path.

    :param path: file path as string
    :param kwargs: passed to state.solver.eval
    :return: file contents as string
    """
    file = self.state.fs.get(path)
    if file is None:
        return None
    return file.concretize(**kwargs)

dumps

dumps(fd, **kwargs)

Returns the concrete content for a file descriptor.

BACKWARD COMPATIBILITY: if you ask for file descriptors 0 1 or 2, it will return the data from stdin, stdout, or stderr as a flat string.

:param fd: A file descriptor. :return: The concrete content. :rtype: str

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def dumps(self, fd, **kwargs):
    """
    Returns the concrete content for a file descriptor.

    BACKWARD COMPATIBILITY: if you ask for file descriptors 0 1 or 2, it will return the data from stdin, stdout,
    or stderr as a flat string.

    :param fd:  A file descriptor.
    :return:    The concrete content.
    :rtype:     str
    """
    if 0 <= fd <= 2:
        data = [self.stdin, self.stdout, self.stderr][fd].concretize(**kwargs)
        if type(data) is list:
            data = b''.join(data)
        return data
    return self.get_fd(fd).concretize(**kwargs)

get_fd

get_fd(fd)

Looks up the SimFileDescriptor associated with the given number (an AST). If the number is concrete and does not map to anything, return None. If the number is symbolic, constrain it to an open fd and create a new file for it.

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def get_fd(self, fd):
    """
    Looks up the SimFileDescriptor associated with the given number (an AST).
    If the number is concrete and does not map to anything, return None.
    If the number is symbolic, constrain it to an open fd and create a new file for it.
    """

    try:
        fd = self.state.solver.eval_one(fd)
        return self.fd.get(fd)
    except SimSolverError:
        pass

    ideal = self._pick_fd()
    self.state.add_constraints(fd == ideal)
    if not self.state.solver.satisfiable():
        raise SimWin32Error("Tried to do operation on symbolic but partially constrained file descriptor")
    fd = ideal
    new_filename = b'/tmp/angr_implicit_%d' % self.autotmp_counter
    l.warning("Tried to look up a symbolic fd - constrained to %d and opened %s", ideal, new_filename)
    self.autotmp_counter += 1
    if self.open(new_filename, Flags.O_RDWR, preferred_fd=fd) != fd:
        raise SimWin32Error("Something went wrong trying to open implicit temp")

    return self.fd.get(fd)

open

open(name, flags, preferred_fd=None)

Open a symbolic file. Basically open(2).

:param name: Path of the symbolic file, as a string or bytes. :type name: string or bytes :param flags: File operation flags, a bitfield of constants from open(2), as an AST :param preferred_fd: Assign this fd if it's not already claimed. :return: The file descriptor number allocated (maps through win32.get_fd to a SimFileDescriptor) or None if the open fails.

mode from open(2) is unsupported at present.

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def open(self, name, flags, preferred_fd=None):
    """
    Open a symbolic file. Basically open(2).

    :param name:            Path of the symbolic file, as a string or bytes.
    :type name:             string or bytes
    :param flags:           File operation flags, a bitfield of constants from open(2), as an AST
    :param preferred_fd:    Assign this fd if it's not already claimed.
    :return:                The file descriptor number allocated (maps through win32.get_fd to a SimFileDescriptor)
                            or None if the open fails.

    ``mode`` from open(2) is unsupported at present.
    """

    if len(name) == 0:
        return None
    if type(name) is str:
        name = name.encode()

    # FIXME: HACK
    if self.uid != 0 and name.startswith(b'/var/run'):
        return None

    # TODO: speed this up (editor's note: ...really? this is fine)
    fd = None
    if preferred_fd is not None and preferred_fd not in self.fd:
        fd = preferred_fd
    else:
        fd = self._pick_fd()

    flags = self.state.solver.eval(flags)
    writing = (flags & Flags.O_ACCMODE) in (Flags.O_RDWR, Flags.O_WRONLY)

    simfile = self.state.fs.get(name)
    if simfile is None:
        ident = SimFile.make_ident(name)
        if not writing:
            if options.ALL_FILES_EXIST not in self.state.options:
                return None
            l.warning("Trying to open unknown file %s - created a symbolic file since ALL_FILES_EXIST is set", name)
            simfile = SimFile(name, ident=ident, size=self.state.solver.BVS('filesize_%s' % ident, self.state.arch.bits, key=('file', ident, 'filesize'), eternal=True))
        else:
            simfile = SimFile(name, ident=ident)
        if not self.state.fs.insert(name, simfile):
            return None

    simfd = SimFileDescriptor(simfile, flags)
    simfd.set_state(self.state)
    self.fd[fd] = simfd
    return fd

sigmask

sigmask(sigsetsize=None)

Gets the current sigmask. If it's blank, a new one is created (of sigsetsize).

:param sigsetsize: the size (in bytes of the sigmask set) :return: the sigmask

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def sigmask(self, sigsetsize=None):
    """
    Gets the current sigmask. If it's blank, a new one is created (of sigsetsize).

    :param sigsetsize: the size (in *bytes* of the sigmask set)
    :return: the sigmask
    """
    if self._sigmask is None:
        if sigsetsize is not None:
            sc = self.state.solver.eval(sigsetsize)
            self.state.add_constraints(sc == sigsetsize)
            self._sigmask = self.state.solver.BVS('initial_sigmask', sc*self.state.arch.byte_width, key=('initial_sigmask',), eternal=True)
        else:
            self._sigmask = self.state.solver.BVS('initial_sigmask', self.sigmask_bits, key=('initial_sigmask',), eternal=True)
    return self._sigmask

sigprocmask

sigprocmask(how, new_mask, sigsetsize, valid_ptr=True)

Updates the signal mask.

:param how: the "how" argument of sigprocmask (see manpage) :param new_mask: the mask modification to apply :param sigsetsize: the size (in bytes of the sigmask set) :param valid_ptr: is set if the new_mask was not NULL

Source code in sema_toolchain/sema_scdg/application/plugin/win32.py

def sigprocmask(self, how, new_mask, sigsetsize, valid_ptr=True):
    """
    Updates the signal mask.

    :param how: the "how" argument of sigprocmask (see manpage)
    :param new_mask: the mask modification to apply
    :param sigsetsize: the size (in *bytes* of the sigmask set)
    :param valid_ptr: is set if the new_mask was not NULL
    """
    oldmask = self.sigmask(sigsetsize)
    self._sigmask = self.state.solver.If(valid_ptr,
        self.state.solver.If(how == self.SIG_BLOCK,
            oldmask | new_mask,
            self.state.solver.If(how == self.SIG_UNBLOCK,
                oldmask & (~new_mask),
                self.state.solver.If(how == self.SIG_SETMASK,
                    new_mask,
                    oldmask
                 )
            )
        ),
        oldmask
    )

Win32ProcFS

Bases: SimMount

The virtual file system mounted at /proc (as of now, on Linux).