diff --git a/qiling/os/posix/posix.py b/qiling/os/posix/posix.py index 1d639ce1..e7e609b9 100644 --- a/qiling/os/posix/posix.py +++ b/qiling/os/posix/posix.py @@ -4,7 +4,7 @@ # from inspect import signature, Parameter -from typing import Dict, NamedTuple, TextIO, Union, Callable, IO, List, Optional +from typing import Dict, TextIO, Tuple, Union, Callable, IO, List, Optional from unicorn.arm64_const import UC_ARM64_REG_X8, UC_ARM64_REG_X16 from unicorn.arm_const import ( @@ -91,14 +91,49 @@ class QlFileDes: self.__fds = fds -# vaguely reflects a shmid_ds structure -class QlShmId(NamedTuple): - segsz: int - uid: int - gid: int - # cuid: int - # cgid: int - mode: int +# vaguely reflects a shmid64_ds structure +class QlShmId: + + def __init__(self, key: int, uid: int, gid: int, mode: int, segsz: int) -> None: + # ipc64_perm + self.key = key + self.uid = uid + self.gid = gid + self.mode = mode + + self.segsz = segsz + + # track the memory locations this segment is currently attached to + self.attach: List[int] = [] + + +class QlShm: + def __init__(self) -> None: + self.__shm: Dict[int, QlShmId] = {} + self.__id: int = 0x0F000000 + + def __len__(self) -> int: + return len(self.__shm) + + def add(self, shm: QlShmId) -> int: + shmid = self.__id + self.__shm[shmid] = shm + + self.__id += 0x1000 + + return shmid + + def remove(self, shmid: int) -> None: + del self.__shm[shmid] + + def get_by_key(self, key: int) -> Tuple[int, Optional[QlShmId]]: + return next(((shmid, shmobj) for shmid, shmobj in self.__shm.items() if shmobj.key == key), (-1, None)) + + def get_by_id(self, shmid: int) -> Optional[QlShmId]: + return self.__shm.get(shmid, None) + + def get_by_attaddr(self, shmaddr: int) -> Optional[QlShmId]: + return next((shmobj for shmobj in self.__shm.values() if shmobj.attach.count(shmaddr) > 0), None) class QlOsPosix(QlOs): @@ -167,7 +202,7 @@ class QlOsPosix(QlOs): self.stdout = self._stdout self.stderr = self._stderr - self._shm: Dict[int, QlShmId] = {} + self._shm = QlShm() def __get_syscall_mapper(self, archtype: QL_ARCH): qlos_path = f'.os.{self.type.name.lower()}.map_syscall' diff --git a/qiling/os/posix/syscall/shm.py b/qiling/os/posix/syscall/shm.py index 4d2f5e16..76009442 100644 --- a/qiling/os/posix/syscall/shm.py +++ b/qiling/os/posix/syscall/shm.py @@ -3,107 +3,140 @@ # Cross Platform and Multi Architecture Advanced Binary Emulation Framework # -from unicorn.unicorn_const import UC_PROT_WRITE, UC_PROT_READ +from unicorn.unicorn_const import UC_PROT_READ, UC_PROT_WRITE, UC_PROT_EXEC from qiling import Qiling -from qiling.exception import QlOutOfMemory +from qiling.const import QL_ARCH +from qiling.exception import QlMemoryMappedError from qiling.os.posix.const import * from qiling.os.posix.posix import QlShmId def ql_syscall_shmget(ql: Qiling, key: int, size: int, shmflg: int): - def __create_shm(size: int, flags: int) -> int: - perms = flags & ((1 << 9) - 1) + def __create_shm(key: int, size: int, flags: int) -> int: + """Create a new shared memory segment for the specified key. - posix_to_uc = ( - (SHM_W, UC_PROT_WRITE), - (SHM_R, UC_PROT_READ) - ) + Returns: shmid of the newly created segment, -1 if an error has occured + """ - # convert posix permissions to unicorn memory access bits - uc_perms = sum(u_perm for p_perm, u_perm in posix_to_uc if perms & p_perm) + if len(ql.os.shm) >= SHMMNI: + return -1 # ENOSPC + + mode = flags & ((1 << 9) - 1) # determine size alignment: either normal or huge page if flags & SHM_HUGETLB: shiftsize = (flags >> HUGETLB_FLAG_ENCODE_SHIFT) & HUGETLB_FLAG_ENCODE_MASK - pagesize = (1 << shiftsize) + alignment = (1 << shiftsize) else: - pagesize = ql.mem.pagesize + alignment = ql.mem.pagesize - if len(ql.os.shm) < SHMMNI: - shm_size = ql.mem.align_up(size, pagesize) + shm_size = ql.mem.align_up(size, alignment) - try: - shm_addr = ql.mem.find_free_space(shm_size, ql.loader.mmap_address, align=pagesize) - except QlOutOfMemory: - return -1 # ENOMEM - else: - ql.mem.map(shm_addr, shm_size, uc_perms, '[shm]') + shmid = ql.os.shm.add(QlShmId(key, ql.os.uid, ql.os.gid, mode, shm_size)) - # for simplicity, the shm key is defined to be its base address - shm_key = shm_addr + ql.log.debug(f'created a new shm: key = {key:#x}, mode = 0{mode:o}, size = {shm_size:#x}. assigned id: {shmid:#x}') - ql.os.shm[shm_key] = QlShmId(shm_size, ql.os.uid, ql.os.gid, perms) - - else: - return -1 # ENOSPC - - return shm_key + return shmid # create new shared memory segment if key == IPC_PRIVATE: - key = __create_shm(size, shmflg) + shmid = __create_shm(key, size, shmflg) - # a shm with the specified key exists - elif key in ql.os.shm: - # user asked to create a new one? - if shmflg & (IPC_CREAT | IPC_EXCL): - return -1 # EEXIST - - shmid = ql.os.shm[key] - - # check whether the user has permissions to access this shm - # FIXME: should probably use ql.os.cuid instead, but we don't support it yet - if (ql.os.uid == shmid.uid) and (shmid.mode & (SHM_W | SHM_R)): - return key - - else: - return -1 # EACCES - - # a shm with the specified key does not exist else: - # user asked to create a new one? - if shmflg & IPC_CREAT: - key = __create_shm(size, shmflg) + shmid, shm = ql.os.shm.get_by_key(key) + # a shm with the specified key does not exist + if shm is None: + # the user asked to create a new one? + if shmflg & IPC_CREAT: + shmid = __create_shm(key, size, shmflg) + + else: + return -1 # ENOENT + + # a shm with the specified key exists else: - return -1 # ENOENT + # the user asked to create a new one? + if shmflg & (IPC_CREAT | IPC_EXCL): + return -1 # EEXIST - return key + # check whether the user has permissions to access this shm + # FIXME: should probably use ql.os.cuid instead, but we don't support it yet + if (ql.os.uid == shm.uid) and (shm.mode & (SHM_W | SHM_R)): + return shmid + + else: + return -1 # EACCES + + return shmid def ql_syscall_shmat(ql: Qiling, shmid: int, shmaddr: int, shmflg: int): - if shmid not in ql.os.shm: + shm = ql.os.shm.get_by_id(shmid) + + # a shm with the specified key does not exist + if shm is None: return -1 # EINVAL if shmaddr == 0: - # system may choose any suitable page-aligned address. since existing segments are - # guaranteed to be aligned and key is defined to be shm base address, we can just - # use the key - addr = shmid + # system may choose any suitable page-aligned address + attaddr = ql.mem.find_free_space(shm.segsz, ql.loader.mmap_address) elif shmflg & SHM_RND: - # note: should align to SHMLBA, but usually its value is just a page - addr = ql.mem.align(shmaddr) + # select the appropriate SHMLBA value, based on the platform + shmlba = { + QL_ARCH.MIPS: 0x40000, + QL_ARCH.ARM: ql.mem.pagesize * 4, + QL_ARCH.ARM64: ql.mem.pagesize * 4, + QL_ARCH.X86: ql.mem.pagesize, + QL_ARCH.X8664: ql.mem.pagesize + } + + # align the address specified by shmaddr to platform SHMLBA + attaddr = ql.mem.align(shmaddr, shmlba[ql.arch.type]) else: + # shmaddr is expected to be aligned if shmaddr & (ql.mem.pagesize - 1): return -1 # EINVAL - addr = shmaddr + attaddr = shmaddr - return addr + perms = UC_PROT_READ + + if shmflg & SHM_RDONLY == 0: + perms |= UC_PROT_WRITE + + if shmflg & SHM_EXEC: + perms |= UC_PROT_EXEC + + # user asked to attached the seg as readable; is it allowed? + if (perms & UC_PROT_READ) and not (shm.mode & SHM_R): + return -1 # EACCES + + # user asked to attached the seg as writable; is it allowed? + if (perms & UC_PROT_WRITE) and not (shm.mode & SHM_W): + return -1 # EACCES + + # TODO: if segment is already attached, there is no need to map another memory for it. + # if we do, data changes will not be reflected between the segment attachments. we could + # use a mmio map for additional attachments, and have writes and reads directed to the + # first attachment mapping + + try: + # attach the segment at shmaddr + ql.mem.map(attaddr, shm.segsz, perms, '[shm]') + except QlMemoryMappedError: + return -1 # EINVAL + + # track attachment + shm.attach.append(attaddr) + + ql.log.debug(f'shm {shmid:#x} attached at {attaddr:#010x}') + + return attaddr __all__ = [