Periodic maintenance PR

qiling/os/posix/const.py

@@ -241,13 +241,15 @@
 linux_arm_socket_options = {
     "SO_DEBUG"           : 0x0001,
     "SO_REUSEADDR"       : 0x0002,
-    "SO_KEEPALIVE"       : 0x0009,
+    "SO_TYPE"            : 0x0003,
+    "SO_ERROR"           : 0x0004,
     "SO_DONTROUTE"       : 0x0005,
     "SO_BROADCAST"       : 0x0006,
-    "SO_LINGER"          : 0x000d,
-    "SO_OOBINLINE"       : 0x000a,
     "SO_SNDBUF"          : 0x0007,
     "SO_RCVBUF"          : 0x0008,
+    "SO_KEEPALIVE"       : 0x0009,
+    "SO_OOBINLINE"       : 0x000a,
+    "SO_LINGER"          : 0x000d,
     "SO_REUSEPORT"       : 0x000f,
     "SO_SNDLOWAT"        : 0x0013,
     "SO_RCVLOWAT"        : 0x0012,

qiling/os/posix/const_mapping.py

@@ -1,40 +1,42 @@
 #!/usr/bin/env python3
-# 
+#
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
-from typing import Mapping, MutableSequence
+from typing import Mapping, TypeVar
 
 from qiling import Qiling
 from qiling.const import QL_ARCH, QL_OS
 
 from .const import *
 
-def _invert_dict(d: Mapping) -> Mapping:
-    return { v:k for k, v in d.items()}
+KT = TypeVar('KT')
+VT = TypeVar('VT')
+
+
+def __invert_dict(d: Mapping[KT, VT]) -> Mapping[VT, KT]:
+    return {v: k for k, v in d.items()}
 
 
-def _constant_mapping(bits: int, d_map: Mapping[str, int], ret: MutableSequence[str] = None, single_mapping: bool = False) -> str:
-    b_map = _invert_dict(d_map)
+def _constant_mapping(bits: int, consts_map: Mapping[str, int]) -> str:
+    return __invert_dict(consts_map)[bits]
 
-    if single_mapping:
-        return b_map[bits]
 
-    if ret is None:
-        ret = []
+def _flags_mapping(value: int, flags_map: Mapping[str, int]) -> str:
+    names = []
 
-    for val, sym in b_map.items():
-        if val & bits:
-            bits ^= val
-            ret.append(sym)
+    for name, flag in flags_map.items():
+        if value & flag:
+            value ^= flag
+            names.append(name)
 
-    if bits:
-        ret.append(f'{bits:#x}')
+    if value:
+        names.append(f'{value:#x}')
 
-    return " | ".join(ret)
+    return ' | '.join(names)
 
 
-def ql_open_flag_mapping(ql: Qiling, flags):
+def ql_open_flag_mapping(ql: Qiling, flags: int) -> int:
     def flag_mapping(flags, mapping_name, mapping_from, mapping_to, host_os, virt_os):
         ret = 0
         for n in mapping_name:
@@ -77,14 +79,12 @@ def flag_mapping(flags, mapping_name, mapping_from, mapping_to, host_os, virt_os
     elif virt_os == QL_OS.QNX:
         f = qnx_arm64_open_flags
 
-    if host_os == QL_OS.LINUX:
-        t = linux_x86_open_flags
-    elif host_os == QL_OS.MACOS:
-        t = macos_x86_open_flags
-    elif host_os == QL_OS.FREEBSD:
-        t = freebsd_x86_open_flags
-    elif host_os == QL_OS.WINDOWS:
-        t = windows_x86_open_flags
+    t = {
+        QL_OS.LINUX:   linux_x86_open_flags,
+        QL_OS.MACOS:   macos_x86_open_flags,
+        QL_OS.FREEBSD: freebsd_x86_open_flags,
+        QL_OS.WINDOWS: windows_x86_open_flags
+    }.get(host_os, {})
 
     if f == t:
         return flags
@@ -108,106 +108,83 @@ def mmap_flag_mapping(flags):
         'MAP_ANON'      : 0x00080000,
         'MAP_SYSRAM'    : 0x01000000
     }
-    return _constant_mapping(flags, mmap_flags)
 
+    return _flags_mapping(flags, mmap_flags)
 
-def mmap_prot_mapping(prots):
-    if prots == 0x0:
+
+def mmap_prot_mapping(prots: int) -> str:
+    if prots == 0:
         return 'PROT_NONE'
 
     mmap_prots = {
-        'PROT_READ' : 0x1,
-        'PROT_WRITE': 0x2,
-        'PROT_EXEC' : 0x4,
+        'PROT_READ' : 0b001,
+        'PROT_WRITE': 0b010,
+        'PROT_EXEC' : 0b100,
 
         # not supported by unicorn
         'PROT_GROWSDOWN' : 0x01000000,
         'PROT_GROWSUP'   : 0x02000000
     }
-    return _constant_mapping(prots, mmap_prots)
-
-
-def socket_type_mapping(t, archtype, ostype):
-    if ostype == QL_OS.MACOS:    
-        socket_type_map = {
-            QL_ARCH.X8664: linux_x86_socket_types,
-            QL_ARCH.ARM64: linux_arm_socket_types,
-        }[archtype]
-    else:
-        socket_type_map = {
-            QL_ARCH.X86: linux_x86_socket_types,
-            QL_ARCH.X8664: linux_x86_socket_types,
-            QL_ARCH.ARM: linux_arm_socket_types,
-            QL_ARCH.ARM64: linux_arm_socket_types,
-            QL_ARCH.MIPS: linux_mips_socket_types,
-        }[archtype]
+
+    return _flags_mapping(prots, mmap_prots)
+
+
+def socket_type_mapping(t: int, archtype: QL_ARCH) -> str:
+    socket_type_map = {
+        QL_ARCH.X86:   linux_x86_socket_types,
+        QL_ARCH.X8664: linux_x86_socket_types,
+        QL_ARCH.ARM:   linux_arm_socket_types,
+        QL_ARCH.ARM64: linux_arm_socket_types,
+        QL_ARCH.MIPS:  linux_mips_socket_types
+    }[archtype]
 
     # https://code.woboq.org/linux/linux/net/socket.c.html#1363
-    t &= SOCK_TYPE_MASK
-    return _constant_mapping(t, socket_type_map, single_mapping=True)
-
-def socket_domain_mapping(p, archtype, ostype):
-    if ostype == QL_OS.MACOS:
-        socket_domain_map = {
-            QL_ARCH.X8664: macos_x86_socket_domain,
-            QL_ARCH.ARM64: linux_arm_socket_domain,
-        }[archtype]        
-    else:     
-        socket_domain_map = {
-            QL_ARCH.X86: linux_x86_socket_domain,
-            QL_ARCH.X8664: linux_x86_socket_domain,
-            QL_ARCH.ARM: linux_arm_socket_domain,
-            QL_ARCH.ARM64: linux_arm_socket_domain,
-            QL_ARCH.MIPS: linux_mips_socket_domain,
-        }[archtype]
-    return _constant_mapping(p, socket_domain_map, single_mapping=True)
-
-def socket_level_mapping(t, archtype, ostype):
-    if ostype == QL_OS.MACOS:    
-        socket_level_map = {
-            QL_ARCH.X8664: linux_x86_socket_level,
-            QL_ARCH.ARM64: linux_arm_socket_level,
-        }[archtype]
-    else:    
-        socket_level_map = {
-            QL_ARCH.X86: linux_x86_socket_level,
-            QL_ARCH.X8664: linux_x86_socket_level,
-            QL_ARCH.ARM: linux_arm_socket_level,
-            QL_ARCH.ARM64: linux_arm_socket_level,
-            QL_ARCH.MIPS: linux_mips_socket_level,
-        }[archtype]
-    return _constant_mapping(t, socket_level_map, single_mapping=True)
-
-
-def socket_ip_option_mapping(t, archtype, ostype):
-    if ostype == QL_OS.MACOS: 
-        socket_option_map = {
-            QL_ARCH.X8664: macos_socket_ip_options,
-            QL_ARCH.ARM64: macos_socket_ip_options,
-        }[archtype]
-    else:      
-        socket_option_map = {
-            QL_ARCH.X86: linux_socket_ip_options,
-            QL_ARCH.X8664: linux_socket_ip_options,
-            QL_ARCH.ARM: linux_socket_ip_options,
-            QL_ARCH.ARM64: linux_socket_ip_options,
-            QL_ARCH.MIPS: linux_socket_ip_options,
-        }[archtype]
-    return _constant_mapping(t, socket_option_map, single_mapping=True)
-
-
-def socket_option_mapping(t, archtype, ostype):
-    if ostype == QL_OS.MACOS:
-        socket_option_map = {
-            QL_ARCH.X8664: linux_x86_socket_options,
-            QL_ARCH.ARM64: linux_arm_socket_options,
-        }[archtype]
-    else: 
-        socket_option_map = {
-            QL_ARCH.X86: linux_x86_socket_options,
-            QL_ARCH.X8664: linux_x86_socket_options,
-            QL_ARCH.ARM: linux_arm_socket_options,
-            QL_ARCH.ARM64: linux_arm_socket_options,
-            QL_ARCH.MIPS: linux_mips_socket_options,
-        }[archtype]
-    return _constant_mapping(t, socket_option_map, single_mapping=True)
+    return _constant_mapping(t & SOCK_TYPE_MASK, socket_type_map)
+
+
+def socket_domain_mapping(p: int, archtype: QL_ARCH, ostype: QL_OS) -> str:
+    socket_domain_map = {
+        QL_ARCH.X86:   linux_x86_socket_domain,
+        QL_ARCH.X8664: macos_x86_socket_domain if ostype == QL_OS.MACOS else linux_x86_socket_domain,
+        QL_ARCH.ARM:   linux_arm_socket_domain,
+        QL_ARCH.ARM64: linux_arm_socket_domain,
+        QL_ARCH.MIPS:  linux_mips_socket_domain
+    }[archtype]
+
+    return _constant_mapping(p, socket_domain_map)
+
+
+def socket_level_mapping(t: int, archtype: QL_ARCH) -> str:
+    socket_level_map = {
+        QL_ARCH.X86:   linux_x86_socket_level,
+        QL_ARCH.X8664: linux_x86_socket_level,
+        QL_ARCH.ARM:   linux_arm_socket_level,
+        QL_ARCH.ARM64: linux_arm_socket_level,
+        QL_ARCH.MIPS:  linux_mips_socket_level
+    }[archtype]
+
+    return _constant_mapping(t, socket_level_map)
+
+
+def socket_ip_option_mapping(t: int, archtype: QL_ARCH, ostype: QL_OS) -> str:
+    socket_option_map = {
+        QL_ARCH.X86:   linux_socket_ip_options,
+        QL_ARCH.X8664: macos_socket_ip_options if ostype == QL_OS.MACOS else linux_socket_ip_options,
+        QL_ARCH.ARM:   linux_socket_ip_options,
+        QL_ARCH.ARM64: macos_socket_ip_options if ostype == QL_OS.MACOS else linux_socket_ip_options,
+        QL_ARCH.MIPS:  linux_mips_socket_ip_options
+    }[archtype]
+
+    return _constant_mapping(t, socket_option_map)
+
+
+def socket_option_mapping(t: int, archtype: QL_ARCH) -> str:
+    socket_option_map = {
+        QL_ARCH.X86:   linux_x86_socket_options,
+        QL_ARCH.X8664: linux_x86_socket_options,
+        QL_ARCH.ARM:   linux_arm_socket_options,
+        QL_ARCH.ARM64: linux_arm_socket_options,
+        QL_ARCH.MIPS:  linux_mips_socket_options
+    }[archtype]
+
+    return _constant_mapping(t, socket_option_map)

qiling/os/posix/structs.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+#
+# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
+#
+
+import ctypes
+
+from qiling.const import QL_ENDIAN
+from qiling.os import struct
+
+
+def make_sockaddr(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class sockaddr(Struct):
+        _pack_ = 1
+
+        _fields_ = (
+            ('sa_family', ctypes.c_uint16),
+        )
+
+    return sockaddr
+
+
+def make_sockaddr_un(archbits: int, endian: QL_ENDIAN, pathlen: int):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class sockaddr_un(Struct):
+        _fields_ = (
+            ('sun_family', ctypes.c_int16),
+            ('sun_path',   ctypes.c_char * pathlen)
+        )
+
+    return sockaddr_un
+
+
+def make_sockaddr_in(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class in_addr(Struct):
+        _fields_ = (
+            ('s_addr', ctypes.c_uint32),
+        )
+
+    class sockaddr_in(Struct):
+        _fields_ = (
+            ('sin_family', ctypes.c_int16),
+            ('sin_port',   ctypes.c_uint16),
+            ('sin_addr',   in_addr),
+            ('sin_zero',   ctypes.c_byte * 8)
+        )
+
+    return sockaddr_in
+
+
+def make_sockaddr_in6(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class in6_addr(Struct):
+        _fields_ = (
+            ('s6_addr', ctypes.c_uint8 * 16),
+        )
+
+    class sockaddr_in6(Struct):
+        _fields_ = (
+            ('sin6_family',   ctypes.c_int16),
+            ('sin6_port',     ctypes.c_uint16),
+            ('sin6_flowinfo', ctypes.c_uint32),
+            ('sin6_addr',     in6_addr),
+            ('sin6_scope_id', ctypes.c_uint32)
+        )
+
+    return sockaddr_in6
+
+
+def make_msghdr(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class msghdr(Struct):
+        _fields_ = (
+            ('msg_name',       ctypes.c_uint64),
+            ('msg_namelen',    ctypes.c_int32),
+            ('msg_iov',        ctypes.c_uint64),
+            ('msg_iovlen',     ctypes.c_int32),
+            ('msg_control',    ctypes.c_uint64),
+            ('msg_controllen', ctypes.c_int32),
+            ('msg_flags',      ctypes.c_int32)
+        )
+
+    return msghdr
+
+
+def make_cmsghdr(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class cmsghdr(Struct):
+        _fields_ = (
+            ('cmsg_len',   ctypes.c_int32),
+            ('cmsg_level', ctypes.c_int32),
+            ('cmsg_type',  ctypes.c_int32)
+        )
+
+    return cmsghdr
+
+
+def make_iovec(archbits: int, endian: QL_ENDIAN):
+    Struct = struct.get_aligned_struct(archbits, endian)
+
+    class iovec(Struct):
+        _fields_ = (
+            ('iov_base', ctypes.c_uint64),
+            ('iov_len',  ctypes.c_uint64)
+        )
+
+    return iovec