proj: Restructure
This commit is contained in:
@@ -0,0 +1,333 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Optional, TypedDict, NotRequired
|
||||
from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn
|
||||
import os
|
||||
import re
|
||||
|
||||
|
||||
type Pointer = int
|
||||
type PointerTo[T] = tuple[Pointer, T]
|
||||
type Constant = tuple[int, str]
|
||||
type Flags = tuple[int, list[str]]
|
||||
StructTimeSpec = TypedDict('StructTimeSpec', {'tv_sec': int, 'tv_nsec': int})
|
||||
StructSigAction = TypedDict('StructSigAction', {'sa_flags': Flags, 'sa_handler': NotRequired[Pointer], 'sa_sigaction': NotRequired[Pointer], 'sa_mask': list[str]})
|
||||
|
||||
|
||||
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
|
||||
pass
|
||||
|
||||
|
||||
class Handler(StreamRequestHandler):
|
||||
pid: Optional[int]
|
||||
path: Optional[str]
|
||||
stack: list[tuple[int, str, tuple]]
|
||||
ret_addr: int
|
||||
|
||||
def before(self) -> None: pass
|
||||
def after(self) -> None: pass
|
||||
def before_fallback(self, func_name: str, *args) -> str: pass
|
||||
def after_fallback(self, func_name: str, *args, **kwargs) -> None: pass
|
||||
|
||||
def handle(self):
|
||||
first = self.rfile.readline()
|
||||
meta = {a[0]: a[1] for a in [tuple(p.decode('utf-8').split(':', 1)) for p in first.split(b' ', 1)[1].strip().split(b';')]}
|
||||
self.pid = int(meta['PID']) if 'PID' in meta else None
|
||||
self.path = meta['PATH'] if 'PATH' in meta else None
|
||||
self.stack = []
|
||||
print(f'Process with PID {self.pid} connected ({self.path})')
|
||||
self.before()
|
||||
try:
|
||||
while True:
|
||||
msg = self.rfile.readline()
|
||||
if not msg:
|
||||
return
|
||||
self.handle_msg(msg)
|
||||
finally:
|
||||
self.after()
|
||||
|
||||
@staticmethod
|
||||
def parse_str(argument: str) -> tuple[str or bytes, int]:
|
||||
if not ((len(argument) >= 2 and argument[0] == '"') or (len(argument) >= 3 and argument[0] == 'b' and argument[1] == '"')):
|
||||
raise ValueError()
|
||||
idx = 1
|
||||
esc, fin = False, False
|
||||
data = b'' if argument[0] == 'b' else ''
|
||||
tmp = None
|
||||
for ch in argument[1:]:
|
||||
idx += 1
|
||||
if fin:
|
||||
if ch in (' ', '\t'):
|
||||
continue
|
||||
elif ch in (',', ']'):
|
||||
idx -= 1
|
||||
break
|
||||
elif tmp:
|
||||
tmp += ch
|
||||
if len(tmp) == 2:
|
||||
data += bytes([int(tmp, 16)]) if argument[0] == 'b' else chr(int(tmp, 16))
|
||||
tmp = None
|
||||
elif esc:
|
||||
if ch in ('\\', '"'):
|
||||
data += ch.encode('ascii') if argument[0] == 'b' else ch
|
||||
elif ch == 'x':
|
||||
tmp = ''
|
||||
esc = False
|
||||
else:
|
||||
raise ValueError()
|
||||
elif ch == '"':
|
||||
fin = True
|
||||
elif ch == '\\':
|
||||
esc = True
|
||||
else:
|
||||
data += ch.encode('utf-8') if argument[0] == 'b' else ch
|
||||
if not fin:
|
||||
raise ValueError()
|
||||
return data, idx
|
||||
|
||||
@staticmethod
|
||||
def parse_arg(argument: str) -> tuple[any, int]:
|
||||
if argument == '':
|
||||
return None, 0
|
||||
m = re.match(r'\s*\(nil\)', argument)
|
||||
if m:
|
||||
return 0, len(m.group(0))
|
||||
m = re.match(r'^\s*(.*?)([,:]|$)', argument)
|
||||
a, e = m.group(1), m.group(2)
|
||||
idx = len(m.group(0))
|
||||
if a.startswith('0x'):
|
||||
val = int(a[2:], 16)
|
||||
elif a.startswith('0') and len(a) > 1:
|
||||
val = int(a[1:], 8)
|
||||
else:
|
||||
val = int(a, 10)
|
||||
if e in (',', ''):
|
||||
return val, idx
|
||||
if argument[idx] == '[':
|
||||
idx += 1
|
||||
l, i = Handler.parse_args(argument[idx:])
|
||||
idx += i
|
||||
if idx < len(argument) and argument[idx] == ',':
|
||||
idx += 1
|
||||
return (val, list(l)), idx
|
||||
elif argument[idx] == '|':
|
||||
m = re.match(r'^[| A-Za-z0-9_]*', argument[idx:])
|
||||
flags = m.group(0)
|
||||
if not flags.startswith('|') or not flags.endswith('|'):
|
||||
raise ValueError()
|
||||
idx += len(flags)
|
||||
if idx < len(argument) and argument[idx] == ',':
|
||||
idx += 1
|
||||
flags = [f.strip() for f in flags[1:-1].split('|') if len(f.strip()) > 0]
|
||||
return (val, flags), idx
|
||||
elif argument[idx] == '"':
|
||||
s, i = Handler.parse_str(argument[idx:])
|
||||
idx += i
|
||||
if idx < len(argument) and argument[idx] == ',':
|
||||
idx += 1
|
||||
return (val, s), idx
|
||||
elif argument[idx] == '{':
|
||||
m = re.match(r'^[^}]*', argument[idx:])
|
||||
value = m.group(0)
|
||||
if not value.startswith('{') or not value.endswith('}'):
|
||||
raise ValueError()
|
||||
idx += len(value)
|
||||
if idx < len(argument) and argument[idx] == ',':
|
||||
idx += 1
|
||||
entries = {}
|
||||
for e in [v.strip() for v in value[1:-1].split(',') if len(e.strip()) > 0]:
|
||||
k, v = e.split(':', 1)
|
||||
entries[k.strip()] = int(v.strip(), 0)
|
||||
return (val, entries), idx
|
||||
else:
|
||||
raise ValueError()
|
||||
|
||||
@staticmethod
|
||||
def parse_args(arguments: str) -> tuple[tuple, int]:
|
||||
args = []
|
||||
idx = 0
|
||||
while idx < len(arguments):
|
||||
if arguments[idx] == ']':
|
||||
idx += 1
|
||||
break
|
||||
val, i = Handler.parse_arg(arguments[idx:])
|
||||
args.append(val)
|
||||
idx += i
|
||||
return tuple(args), idx
|
||||
|
||||
def handle_msg(self, msg: bytes):
|
||||
timestamp, data = msg.rstrip(b'\n').split(b' ', 1)
|
||||
if not data.startswith(b'return ') and not data == b'return':
|
||||
call = data.decode('utf-8')
|
||||
print(f'[{self.pid}] {call}')
|
||||
func_name = call[:call.find('(')]
|
||||
self.ret_addr = int(call[call.rfind(':') + 1:], 0)
|
||||
args, _ = Handler.parse_args(call[call.find('(') + 1:call.rfind(':') - 1])
|
||||
self.stack.append((self.ret_addr, func_name, args))
|
||||
try:
|
||||
func = getattr(self, f'before_{func_name}')
|
||||
if not callable(func):
|
||||
func = None
|
||||
except AttributeError:
|
||||
func = None
|
||||
try:
|
||||
if func is None:
|
||||
raise NotImplementedError()
|
||||
command = func(*args) or self.before_fallback(func_name, *args) or 'ok'
|
||||
except NotImplementedError:
|
||||
command = self.before_fallback(func_name, *args) or 'ok'
|
||||
print(f'[{self.pid}] -> {command}')
|
||||
self.wfile.write(command.encode('utf-8') + b'\n')
|
||||
else:
|
||||
ret = data.decode('utf-8')
|
||||
ret_value, _ = Handler.parse_arg(ret[7:].split(';')[0])
|
||||
self.ret_addr, func_name, args = self.stack.pop()
|
||||
try:
|
||||
func = getattr(self, f'after_{func_name}')
|
||||
if not callable(func):
|
||||
func = None
|
||||
except AttributeError:
|
||||
func = None
|
||||
try:
|
||||
if func is None:
|
||||
raise NotImplementedError()
|
||||
if ret_value is None:
|
||||
func(*args)
|
||||
else:
|
||||
func(*args, ret_value)
|
||||
except NotImplementedError:
|
||||
if ret_value is None:
|
||||
self.after_fallback(func_name, *args)
|
||||
else:
|
||||
self.after_fallback(func_name, *args, ret_value)
|
||||
print(f'[{self.pid}] -> {ret}')
|
||||
|
||||
def before_malloc(self, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_malloc(self, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_calloc(self, nmemb: int, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_calloc(self, nmemb: int, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_realloc(self, ptr: Pointer, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_realloc(self, ptr: Pointer, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_reallocarray(self, ptr: Pointer, nmemb: int, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_reallocarray(self, ptr: Pointer, nmemb: int, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_free(self, ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_free(self, ptr: Pointer) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_getopt(self, argc: int, argv: PointerTo[list[PointerTo[bytes]]], optstring: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_getopt(self, argc: int, argv: PointerTo[list[PointerTo[bytes]]], optstring: PointerTo[bytes],
|
||||
ret_value: int) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_close(self, fildes: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_close(self, fildes: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sigaction(self, sig: Constant, act: PointerTo[StructSigAction], oact_ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sigaction(self, sig: Constant, act: PointerTo[StructSigAction], oact_ptr: Pointer,
|
||||
ret_value: int, errno: str = None, oact: StructSigAction = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_init(self, sem: Pointer, pshared: int, value: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_init(self, sem: Pointer, pshared: int, value: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int],
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_post(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_post(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_wait(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_wait(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_trywait(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_trywait(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_timedwait(self, sem: Pointer, abs_timeout: PointerTo[StructTimeSpec]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_timedwait(self, sem: Pointer, abs_timeout: PointerTo[StructTimeSpec],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_getvalue(self, sem: Pointer, value_ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_getvalue(self, sem: Pointer, value_ptr: Pointer,
|
||||
ret_value: int, errno: str = None, value: int = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_close(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_close(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_unlink(self, name: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_unlink(self, name: PointerTo[bytes],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_destroy(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_destroy(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_shm_open(self, name: PointerTo[bytes], oflag: Flags, mode: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_shm_open(self, name: PointerTo[bytes], oflag: Flags, mode: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_shm_unlink(self, name: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_shm_unlink(self, name: PointerTo[bytes],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_mmap(self, addr: Pointer, length: int, prot: Flags, flags: Flags, fildes: int, off: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_mmap(self, addr: Pointer, length: int, prot: Flags, flags: Flags, fildes: int, off: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_munmap(self, addr: Pointer, length: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_munmap(self, addr: Pointer, length: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_ftruncate(self, fildes: int, length: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_ftruncate(self, fildes: int, length: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
def intercept(socket: str, handler: type[Handler]) -> None:
|
||||
try:
|
||||
with ThreadedUnixStreamServer(socket, handler) as server:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print('\nBye')
|
||||
server.shutdown()
|
||||
finally:
|
||||
try:
|
||||
os.unlink(socket)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from intercept import *
|
||||
|
||||
|
||||
FUNCTION_ERRORS: dict[str, list[str]] = {
|
||||
'malloc': ['ENOMEM'],
|
||||
'calloc': ['ENOMEM'],
|
||||
'realloc': ['ENOMEM'],
|
||||
'reallocarray': ['ENOMEM'],
|
||||
'close': ['EBADF'], # EINTR, EIO
|
||||
'sigaction': ['EINVAL'],
|
||||
'sem_init': ['EINVAL', 'ENOSYS'],
|
||||
'sem_open': ['EACCES', 'EEXIST', 'EINVAL', 'EMFILE', 'ENAMETOOLONG', 'ENFILE', 'ENOENT', 'ENOMEM'],
|
||||
'sem_post': ['EINVAL', 'EOVERFLOW'],
|
||||
'sem_wait': ['EINTR', 'EINVAL'],
|
||||
'sem_trywait': ['EAGAIN', 'EINTR', 'EINVAL'],
|
||||
'sem_timedwait': ['EINTR', 'EINVAL', 'ETIMEDOUT'],
|
||||
'sem_getvalule': ['EINVAL'],
|
||||
'sem_close': ['EINVAL'],
|
||||
'sem_unlink': ['EACCES', 'ENAMETOOLONG', 'ENOENT'],
|
||||
'sem_destroy': ['EINVAL'],
|
||||
'shm_open': ['EACCES', 'EEXIST', 'EINVAL', 'EMFILE', 'ENAMETOOLONG', 'ENFILE', 'ENOENT'],
|
||||
'shm_unlink': ['EACCES', 'ENAMETOOLONG', 'ENOENT'],
|
||||
'mmap': ['EACCES', 'EBADF', 'EINVAL', 'EMFILE', 'ENODEV', 'ENOMEM', 'ENOTSUP', 'ENXIO', 'EOVERFLOW'], # EAGAIN
|
||||
'munmap': ['EINVAL'],
|
||||
'ftruncate': ['EINTR', 'EINVAL', 'EFBIG', 'EIO', 'EBADF'],
|
||||
}
|
||||
|
||||
SKIP_ERRORS: list[str] = ['EINTR']
|
||||
IGNORE_ERRORS: list[str] = ['EINVAL', 'EBADF', 'EOVERFLOW', 'ENAMETOOLONG']
|
||||
|
||||
|
||||
class MemoryAllocationTester(Handler):
|
||||
allocated: dict[int, tuple[str, int, int]]
|
||||
max_allocated: int
|
||||
num_malloc: int
|
||||
num_realloc: int
|
||||
num_free: int
|
||||
num_invalid_free: int
|
||||
|
||||
def before(self):
|
||||
self.allocated = {}
|
||||
self.max_allocated = 0
|
||||
self.num_malloc = 0
|
||||
self.num_realloc = 0
|
||||
self.num_free = 0
|
||||
self.num_invalid_free = 0
|
||||
|
||||
def after(self):
|
||||
if len(self.allocated) > 0:
|
||||
print("Not free'd:")
|
||||
for ptr, (func, ret, size) in self.allocated.items():
|
||||
print(f' 0x{ptr:x}: {size} bytes ({func}, return address 0x{ret:x})')
|
||||
else:
|
||||
print("All blocks free'd!")
|
||||
print(f'Max allocated: {self.max_allocated} bytes')
|
||||
|
||||
def update_max_allocated(self):
|
||||
total = sum(a[2] for a in self.allocated.values())
|
||||
if total > self.max_allocated:
|
||||
self.max_allocated = total
|
||||
|
||||
def after_malloc(self, size, ret_value, errno=None) -> None:
|
||||
self.num_malloc += 1
|
||||
if ret_value != 0:
|
||||
self.allocated[ret_value] = ('malloc', self.ret_addr, size)
|
||||
self.update_max_allocated()
|
||||
|
||||
def after_calloc(self, nmemb, size, ret_value, errno=None) -> None:
|
||||
self.num_malloc += 1
|
||||
if ret_value != 0:
|
||||
self.allocated[ret_value] = ('calloc', self.ret_addr, nmemb * size)
|
||||
self.update_max_allocated()
|
||||
|
||||
def after_realloc(self, ptr, size, ret_value, errno=None) -> None:
|
||||
self.num_realloc += 1
|
||||
if ptr != 0:
|
||||
if ret_value != 0:
|
||||
v = self.allocated[ptr]
|
||||
del self.allocated[ptr]
|
||||
self.allocated[ret_value] = (v[0], v[1], size)
|
||||
self.update_max_allocated()
|
||||
|
||||
def after_reallocarray(self, ptr, nmemb, size, ret_value, errno=None) -> None:
|
||||
self.num_realloc += 1
|
||||
if ptr != 0:
|
||||
if ret_value != 0:
|
||||
v = self.allocated[ptr]
|
||||
del self.allocated[ptr]
|
||||
self.allocated[ret_value] = (v[0], v[1], nmemb * size)
|
||||
self.update_max_allocated()
|
||||
|
||||
def after_free(self, ptr) -> None:
|
||||
self.num_free += 1
|
||||
if ptr != 0:
|
||||
if ptr in self.allocated:
|
||||
del self.allocated[ptr]
|
||||
else:
|
||||
self.num_free -= 1
|
||||
self.num_invalid_free += 1
|
||||
|
||||
|
||||
class InterruptedCheckTester(Handler):
|
||||
cycles: int = 50
|
||||
functions: dict[str, tuple[str or None, str]] = {
|
||||
fn: ('fail EINTR' if fn not in ('sem_post',) else None,
|
||||
'return 0' if fn.startswith('sem_') else 'ok')
|
||||
for fn, errors in FUNCTION_ERRORS.items()
|
||||
if 'EINTR' in errors or fn in ('sem_post',)
|
||||
}
|
||||
|
||||
counter: int = 0
|
||||
last_func_name: Optional[str] = None
|
||||
last_ret_addr: Optional[int] = None
|
||||
tested_functions: dict[tuple[str, int], str]
|
||||
|
||||
@property
|
||||
def while_testing(self) -> bool:
|
||||
return self.counter % self.cycles != 0
|
||||
|
||||
def before(self) -> None:
|
||||
self.tested_functions = {}
|
||||
|
||||
def after(self) -> None:
|
||||
if self.while_testing:
|
||||
self.error()
|
||||
for (name, ret_addr), status in self.tested_functions.items():
|
||||
print(f'{name} (0x{ret_addr:x}) -> {status}')
|
||||
|
||||
def error(self):
|
||||
print(f'Error: Return value and errno EINTR not handled correctly in {self.last_func_name} (return address 0x{self.last_ret_addr:x})')
|
||||
self.tested_functions[(self.last_func_name, self.last_ret_addr)] = 'failed'
|
||||
self.counter = 0
|
||||
self.last_func_name = None
|
||||
self.last_ret_addr = None
|
||||
|
||||
def before_fallback(self, func_name: str, *args) -> str:
|
||||
if self.while_testing and (self.last_func_name != func_name or self.last_ret_addr != self.ret_addr):
|
||||
self.error()
|
||||
return 'ok'
|
||||
elif func_name not in self.functions:
|
||||
return 'ok'
|
||||
elif self.functions[func_name][0] is None:
|
||||
return self.functions[func_name][1]
|
||||
self.counter += 1
|
||||
if self.while_testing:
|
||||
self.last_ret_addr = self.ret_addr
|
||||
self.last_func_name = func_name
|
||||
self.tested_functions[(self.last_func_name, self.last_ret_addr)] = 'running'
|
||||
return self.functions[func_name][0]
|
||||
else:
|
||||
self.tested_functions[(self.last_func_name, self.last_ret_addr)] = 'passed'
|
||||
self.last_ret_addr = None
|
||||
self.last_func_name = None
|
||||
return self.functions[func_name][1]
|
||||
|
||||
|
||||
class ReturnValueCheckTester(Handler):
|
||||
functions: dict[str, list[str]] = {
|
||||
fn: [e for e in errors if e not in SKIP_ERRORS and e not in IGNORE_ERRORS]
|
||||
for fn, errors in FUNCTION_ERRORS.items()
|
||||
if len(set(errors) - set(SKIP_ERRORS) - set(IGNORE_ERRORS)) > 0
|
||||
}
|
||||
Reference in New Issue
Block a user