qilingframework/qiling
1
0
Fork 0
mirror of https://github.com/qilingframework/qiling.git synced 2026-10-12 02:21:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Re-implement POSIX shm

Browse Source
This commit is contained in:
elicn
2023-04-13 22:04:55 +03:00
parent 06cd3f09a9
commit 0dd8545e25
2 changed files with 138 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qiling/os/posix/posix.py
+45 -10
View File
@@ -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'
qiling/os/posix/syscall/shm.py
+93 -60
View File
@@ -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__ = [