309 lines
12 KiB
Python
309 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from intercept import *
|
|
import sys
|
|
|
|
|
|
FUNCTION_ERRORS: dict[str, list[str]] = {
|
|
'malloc': ['ENOMEM'],
|
|
'calloc': ['ENOMEM'],
|
|
'realloc': ['ENOMEM'],
|
|
'reallocarray': ['ENOMEM'],
|
|
'read': ['EINTR', 'EIO', 'ECONNRESET', 'ENOTCONN', 'ETIMEDOUT'],
|
|
'pread': ['EINTR', 'EIO'],
|
|
'write': ['EINTR', 'EIO', 'EDQUOT', 'EFBIG', 'ENOSPC', 'EPIPE', 'ECONNRESET'],
|
|
'pwrite': ['EINTR', 'EIO', 'EDQUOT', 'EFBIG', 'ENOSPC', 'EPIPE'],
|
|
'close': [],
|
|
'sigaction': [],
|
|
'sem_init': ['ENOSYS'],
|
|
'sem_open': ['EACCES', 'EEXIST', 'EMFILE', 'ENFILE', 'ENOENT', 'ENOMEM'],
|
|
'sem_post': [],
|
|
'sem_wait': ['EINTR'],
|
|
'sem_trywait': ['EINTR'],
|
|
'sem_timedwait': ['EINTR', 'ETIMEDOUT'],
|
|
'sem_getvalule': [],
|
|
'sem_close': [],
|
|
'sem_unlink': ['EACCES', 'ENOENT'],
|
|
'sem_destroy': [],
|
|
'shm_open': ['EACCES', 'EEXIST', 'EMFILE', 'ENFILE', 'ENOENT'],
|
|
'shm_unlink': ['EACCES', 'ENOENT'],
|
|
'mmap': ['EACCES', 'EBADF', 'EMFILE', 'ENODEV', 'ENOMEM', 'ENOTSUP', 'ENXIO'],
|
|
'munmap': [],
|
|
'ftruncate': ['EINTR', 'EFBIG', 'EIO', 'EBADF'],
|
|
'fork': ['ENOMEM', 'ENOSYS'],
|
|
'wait': ['ECHILD', 'EINTR'],
|
|
'waitpid': ['ECHILD', 'EINTR'],
|
|
'execl': ['ECHILD', 'EINTR'],
|
|
'execlp': ['ECHILD', 'EINTR'],
|
|
'execle': ['ECHILD', 'EINTR'],
|
|
'execv': ['ECHILD', 'EINTR'],
|
|
'execvp': ['ECHILD', 'EINTR'],
|
|
'execvpe': ['ECHILD', 'EINTR'],
|
|
'execve': ['ECHILD', 'EINTR'],
|
|
'fexecve': ['ECHILD', 'EINTR'],
|
|
'pipe': ['EMFILE', 'ENFILE'],
|
|
'dup': ['EBADF', 'EMFILE'],
|
|
'dup2': ['EBADF', 'EBUSY', 'EINTR', 'EMFILE'],
|
|
'dup3': ['EBADF', 'EBUSY', 'EINTR', 'EMFILE'],
|
|
'send': ['EINTR'],
|
|
'recv': ['EINTR'],
|
|
}
|
|
|
|
SKIP_ERRORS: list[str] = ['EINTR']
|
|
|
|
|
|
class MemoryAllocationParser(Parser):
|
|
allocated: dict[int, tuple[str, int, str, str, int, str, int]]
|
|
invalid_frees: list[tuple[str, int, str, str, int, str, int]]
|
|
max_allocated: int
|
|
num_alloc: int
|
|
num_realloc: int
|
|
num_free: int
|
|
|
|
def before(self):
|
|
self.allocated = {}
|
|
self.invalid_frees = []
|
|
self.max_allocated = 0
|
|
self.num_alloc = 0
|
|
self.num_realloc = 0
|
|
self.num_free = 0
|
|
|
|
def update_max_allocated(self):
|
|
total = sum(a[1] 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_alloc += 1
|
|
if ret_value != 0:
|
|
self.allocated[ret_value] = ('malloc', size,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num)
|
|
self.update_max_allocated()
|
|
|
|
def after_calloc(self, nmemb, size, ret_value, errno=None) -> None:
|
|
self.num_alloc += 1
|
|
if ret_value != 0:
|
|
self.allocated[ret_value] = ('calloc', nmemb * size,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num)
|
|
self.update_max_allocated()
|
|
|
|
def after_realloc(self, ptr, size, ret_value, errno=None) -> None:
|
|
if ptr == 0:
|
|
self.num_alloc += 1
|
|
if ret_value != 0:
|
|
self.allocated[ret_value] = ('realloc', size,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num)
|
|
else:
|
|
self.num_realloc += 1
|
|
if ret_value != 0 and ptr in self.allocated:
|
|
v = self.allocated[ptr]
|
|
del self.allocated[ptr]
|
|
self.allocated[ret_value] = (v[0], size, v[2], v[3], v[4], v[5], v[6])
|
|
self.update_max_allocated()
|
|
|
|
def after_reallocarray(self, ptr, nmemb, size, ret_value, errno=None) -> None:
|
|
if ptr == 0:
|
|
self.num_alloc += 1
|
|
if ret_value != 0:
|
|
self.allocated[ret_value] = ('reallocarray', nmemb * size,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num)
|
|
else:
|
|
self.num_realloc += 1
|
|
if ret_value != 0:
|
|
v = self.allocated[ptr]
|
|
del self.allocated[ptr]
|
|
self.allocated[ret_value] = (v[0], nmemb * size, v[2], v[3], v[4], v[5], v[6])
|
|
self.update_max_allocated()
|
|
|
|
def after_getaddrinfo(self, node, service, hints, res_ptr, ret_value, errno=None, res=None) -> None:
|
|
self.num_alloc += 1
|
|
if ret_value[0] == 0 and res is not None:
|
|
size = sum(48 + r['ai_addrlen'] for r in res[1])
|
|
self.allocated[res[0]] = ('getaddrinfo', size,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num)
|
|
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.invalid_frees.append(('free', ptr,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num))
|
|
|
|
def after_freeaddrinfo(self, res: Pointer) -> None:
|
|
self.num_free += 1
|
|
if res != 0:
|
|
if res in self.allocated:
|
|
del self.allocated[res]
|
|
else:
|
|
self.num_free -= 1
|
|
self.invalid_frees.append(('freeaddrinfo', res,
|
|
self.dli_file_name, self.dli_sym_name, self.rel_ret_addr,
|
|
self.src_file_name, self.src_line_num))
|
|
|
|
|
|
class MemoryAllocationTester(MemoryAllocationParser, Handler):
|
|
pass
|
|
|
|
|
|
class InterruptedCheckParser(Parser):
|
|
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():
|
|
if status == 'passed':
|
|
print(f'\x1B[32;1m{name} (0x{ret_addr:x}) -> {status}\x1B[0m', file=sys.stderr)
|
|
else:
|
|
print(f'\x1B[31;1m{name} (0x{ret_addr:x}) -> {status}\x1B[0m', file=sys.stderr)
|
|
|
|
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})', file=sys.stderr)
|
|
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 InterruptedCheckTester(InterruptedCheckParser, Handler):
|
|
pass
|
|
|
|
|
|
def get_return_value_check_tester() -> tuple[dict, type[Parser]]:
|
|
ctx = {
|
|
'state': 'init',
|
|
'call_sequence': [],
|
|
'next': None,
|
|
'last_error': None,
|
|
'results': {},
|
|
}
|
|
|
|
class ReturnValueCheckTester(Parser):
|
|
functions: dict[str, list[str]] = {
|
|
fn: [e for e in errors if e not in SKIP_ERRORS]
|
|
for fn, errors in FUNCTION_ERRORS.items()
|
|
if len(set(errors) - set(SKIP_ERRORS)) > 0
|
|
}
|
|
context: dict
|
|
num: int = 0
|
|
|
|
@property
|
|
def call_sequence(self) -> list[str]:
|
|
return self.context['call_sequence']
|
|
|
|
@property
|
|
def is_init(self) -> bool:
|
|
return self.context['state'] == 'init'
|
|
|
|
@property
|
|
def is_testing(self) -> bool:
|
|
return self.context['state'] == 'testing'
|
|
|
|
@property
|
|
def is_waiting(self) -> bool:
|
|
return self.context['state'] == 'waiting'
|
|
|
|
@property
|
|
def is_failed(self) -> bool:
|
|
return self.context['state'] == 'failed'
|
|
|
|
def next(self, last: Optional[tuple[int, Optional[str]]]) -> tuple[int, Optional[str]]:
|
|
if last is None:
|
|
if len(self.call_sequence) == 0:
|
|
return 0, None
|
|
err = self.functions.get(self.call_sequence[0], [None])[0]
|
|
return (0, err) if err else self.next((0, err))
|
|
last_fn, last_er = last
|
|
if last_er is None:
|
|
if last_fn + 1 >= len(self.call_sequence):
|
|
return last_fn + 1, None
|
|
err = self.functions.get(self.call_sequence[last_fn + 1], [None])[0]
|
|
return (last_fn + 1, err) if err else self.next((last_fn + 1, err))
|
|
errors = self.functions.get(self.call_sequence[last_fn], [])
|
|
i = errors.index(last_er)
|
|
return (last_fn, errors[i + 1]) if i + 1 < len(errors) else self.next((last_fn, None))
|
|
|
|
def before(self) -> None:
|
|
self.context = ctx
|
|
if self.is_waiting or self.is_failed:
|
|
self.context['state'] = 'testing'
|
|
|
|
def after(self) -> None:
|
|
if self.is_init:
|
|
self.context['state'] = 'testing'
|
|
self.context['next'] = self.next(None)
|
|
elif self.is_testing and self.context['next'][0] > len(self.context['call_sequence']):
|
|
self.context['state'] = 'finished'
|
|
elif self.is_waiting:
|
|
self.context['results'][(self.num - 1, self.context['last_error'])] = 'passed'
|
|
|
|
def before_fallback(self, func_name: str, *args) -> str:
|
|
if self.is_init:
|
|
self.call_sequence.append(func_name)
|
|
return 'ok'
|
|
elif self.is_waiting or self.is_failed:
|
|
print(func_name)
|
|
if func_name not in ('malloc', 'free', 'close', 'exit'):
|
|
self.context['state'] = 'failed'
|
|
self.context['results'][(self.num - 1, self.context['last_error'])] = 'failed'
|
|
return 'ok'
|
|
self.num += 1
|
|
nxt = self.context['next']
|
|
if self.num - 1 != nxt[0] or nxt[1] is None:
|
|
return 'ok'
|
|
self.context['next'] = self.next(nxt)
|
|
self.context['state'] = 'waiting' if self.context['next'][1] else 'finished'
|
|
self.context['last_error'] = nxt[1]
|
|
return 'fail ' + nxt[1]
|
|
|
|
return ctx, ReturnValueCheckTester
|