698 lines
32 KiB
Python
698 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from __future__ import annotations
|
|
from typing import Optional, TypedDict, NamedTuple, NotRequired, BinaryIO
|
|
from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn
|
|
import os
|
|
import re
|
|
|
|
|
|
type Pointer = int
|
|
|
|
class PointerTo[T](NamedTuple):
|
|
ptr: Pointer
|
|
target: T
|
|
|
|
class Constant(NamedTuple):
|
|
raw: int
|
|
name: str
|
|
|
|
class Flags(NamedTuple):
|
|
bitfield: int
|
|
flags: 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[Constant]})
|
|
StructSockAddr = TypedDict('StructSockAddr', {'sa_family': Constant, 'sa_data': NotRequired[bytes],
|
|
'sun_path': NotRequired[bytes],
|
|
'sin_addr': NotRequired[bytes], 'sin_port': NotRequired[int],
|
|
'sin6_addr': NotRequired[bytes], 'sin6_port': NotRequired[int], 'sin6_scope_id': NotRequired[int]})
|
|
StructAddrInfo = TypedDict('StructAddrInfo', {'ai_flags': Flags, 'ai_family': Constant, 'ai_socktype': Constant, 'ai_protocol': int, 'ai_addrlen': int,
|
|
'ai_addr': PointerTo[StructSockAddr],
|
|
'ai_canonname': PointerTo[bytes], 'ai_next': Pointer})
|
|
StructMsgHdr = TypedDict('StructMsgHdr', {})
|
|
|
|
|
|
RET_ADDR_RE = re.compile(r': *((0x)?[0-9a-fA-Fx]+) *\((.+?)\+(.+?)(, *([^:]+?))?(, *(([^:]+?):([0-9]+)))?\)$')
|
|
|
|
|
|
class FunctionCallId:
|
|
func_name: str
|
|
ret_addr: int
|
|
obj_path: str
|
|
rel_ret_addr: int
|
|
sym_name: Optional[str] = None
|
|
src_file_name: Optional[str] = None
|
|
src_line_num: Optional[int] = None
|
|
|
|
@property
|
|
def obj_name(self) -> str:
|
|
return self.obj_path.split('/')[-1]
|
|
|
|
@staticmethod
|
|
def for_exit() -> FunctionCallId:
|
|
call_id = FunctionCallId()
|
|
call_id.func_name = 'exit'
|
|
call_id.obj_path = 'sys'
|
|
call_id.ret_addr = 0
|
|
call_id.rel_ret_addr = 0
|
|
return call_id
|
|
|
|
@property
|
|
def discriminator(self) -> str:
|
|
discr = [f'{self.obj_name}+0x{self.rel_ret_addr:x}']
|
|
if self.sym_name:
|
|
discr.append(self.sym_name)
|
|
if self.src_file_name and self.src_line_num:
|
|
discr.append(f'{self.src_file_name}:{self.src_line_num}')
|
|
return ', '.join(discr)
|
|
|
|
def __eq__(self, other) -> bool:
|
|
if not isinstance(other, FunctionCallId):
|
|
return False
|
|
return (self.func_name, self.obj_path, self.rel_ret_addr) == (other.func_name, other.obj_path, other.rel_ret_addr)
|
|
|
|
def __hash__(self) -> int:
|
|
return hash((self.func_name, self.obj_path, self.rel_ret_addr))
|
|
|
|
def __str__(self) -> str:
|
|
return self.func_name + ', ' + self.discriminator
|
|
|
|
def __repr__(self) -> str:
|
|
return f'<{self}>'
|
|
|
|
|
|
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
|
|
pass
|
|
|
|
|
|
class Parser:
|
|
rfile: BinaryIO
|
|
wfile: Optional[BinaryIO]
|
|
pid: Optional[int]
|
|
tid: Optional[int]
|
|
path: Optional[str]
|
|
stack: dict[tuple[int, int], list[tuple[int, int, str, Optional[str], Optional[str], Optional[int], str, tuple]]]
|
|
ret_addr: int
|
|
rel_ret_addr: int
|
|
dli_file_name: str
|
|
dli_sym_name: Optional[str]
|
|
src_file_name: Optional[str]
|
|
src_line_num: Optional[int]
|
|
|
|
def __init__(self, rfile: BinaryIO, wfile: BinaryIO = None):
|
|
self.rfile = rfile
|
|
self.wfile = wfile
|
|
self.stack = {}
|
|
self.pid = None
|
|
self.path = None
|
|
|
|
def get_call_id(self, func_name: str) -> FunctionCallId:
|
|
call_id = FunctionCallId()
|
|
call_id.func_name = func_name
|
|
call_id.ret_addr = self.ret_addr
|
|
call_id.obj_path = self.dli_file_name
|
|
call_id.rel_ret_addr = self.rel_ret_addr
|
|
call_id.sym_name = self.dli_sym_name
|
|
call_id.src_file_name = self.src_file_name
|
|
call_id.src_line_num = self.src_line_num
|
|
return call_id
|
|
|
|
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 parse(self):
|
|
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[bytes, int]:
|
|
if not (len(argument) >= 2 and argument[0] == '"'):
|
|
raise ValueError()
|
|
idx = 1
|
|
esc, fin = False, False
|
|
data = b''
|
|
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)])
|
|
tmp = None
|
|
elif esc:
|
|
if ch in ('\\', '"'):
|
|
data += ch.encode('ascii')
|
|
esc = False
|
|
elif ch == 'x':
|
|
tmp = ''
|
|
esc = False
|
|
elif ch in ('n', 'r'):
|
|
data += b'\n' if ch == 'n' else b'\r'
|
|
esc = False
|
|
else:
|
|
raise ValueError(ch)
|
|
elif ch == '"':
|
|
fin = True
|
|
elif ch == '\\':
|
|
esc = True
|
|
else:
|
|
data += ch.encode('utf-8')
|
|
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\)|NULL|null|nullptr)\s*(,|]|}|$)', argument)
|
|
if m:
|
|
return 0, len(m.group(0)) - (1 if m.group(2) in ('}', ']') else 0)
|
|
m = re.match(r'\s*"', argument)
|
|
if m:
|
|
idx = len(m.group(0)) - 1
|
|
s, i = Parser.parse_str(argument[idx:])
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] in ',;':
|
|
idx += 1
|
|
return s, idx
|
|
m = re.match(r'\s*\[', argument)
|
|
if m:
|
|
idx = len(m.group(0))
|
|
s, i = Parser.parse_args(argument[idx:])
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] in ',;':
|
|
idx += 1
|
|
return s, idx
|
|
m = re.match(r'\s*\{', argument)
|
|
if m:
|
|
idx = len(m.group(0))
|
|
s, i = Parser.parse_args(argument[idx:], named=True)
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] in ',;':
|
|
idx += 1
|
|
return s, idx
|
|
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)
|
|
elif a in ('(nil)', 'NULL', 'null', 'nullptr'):
|
|
val = 0
|
|
else:
|
|
val = int(a, 10)
|
|
if e in (',', ''):
|
|
return val, idx
|
|
elif e in ('}', ']'):
|
|
return val, idx - 1
|
|
if argument[idx] == '[':
|
|
idx += 1
|
|
l, i = Parser.parse_args(argument[idx:])
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] == ',':
|
|
idx += 1
|
|
return PointerTo(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 Flags(val, flags), idx
|
|
elif argument[idx] == '"':
|
|
s, i = Parser.parse_str(argument[idx:])
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] == ',':
|
|
idx += 1
|
|
return PointerTo(val, s), idx
|
|
elif argument[idx] == '{':
|
|
idx += 1
|
|
l, i = Parser.parse_args(argument[idx:], named=True)
|
|
idx += i
|
|
if idx < len(argument) and argument[idx] == ',':
|
|
idx += 1
|
|
return PointerTo(val, l), idx
|
|
else:
|
|
m = re.match(r'0x[0-9a-fA-F]+|[0-9]+|\(nil\)', argument[idx:])
|
|
if m is not None:
|
|
value = m.group(0)
|
|
idx += len(value)
|
|
if idx < len(argument) and argument[idx] == ',':
|
|
idx += 1
|
|
value = int(value, 0) if value != '(nil)' else 0
|
|
return PointerTo(val, value), idx
|
|
m = re.match(r'[A-Z0-9_]+|\?', argument[idx:])
|
|
if m is not None:
|
|
value = m.group(0)
|
|
idx += len(value)
|
|
if idx < len(argument) and argument[idx] == ',':
|
|
idx += 1
|
|
return Constant(val, value), idx
|
|
raise ValueError()
|
|
|
|
@staticmethod
|
|
def parse_args(arguments: str, named: bool = False, ret: bool = False) -> tuple[tuple or dict, int]:
|
|
args = [] if not named else {}
|
|
idx = 0
|
|
name = None
|
|
while idx < len(arguments):
|
|
if arguments[idx] == ']' and not named or arguments[idx] == '}' and named:
|
|
idx += 1
|
|
break
|
|
elif ret and named:
|
|
m = re.match(r'^\s*([^=]+)=', arguments[idx:])
|
|
idx += len(m.group(0))
|
|
name = m.group(1)
|
|
elif named:
|
|
m = re.match(r'^\s*([^:]+):', arguments[idx:])
|
|
idx += len(m.group(0))
|
|
name = m.group(1)
|
|
val, i = Parser.parse_arg(arguments[idx:])
|
|
if named:
|
|
args[name] = val
|
|
else:
|
|
args.append(val)
|
|
idx += i
|
|
return tuple(args) if not named else args, idx
|
|
|
|
def handle_msg(self, msg: bytes):
|
|
timestamp, pid, tid, data = msg.rstrip(b'\n').split(b' ', 3)
|
|
self.pid, self.tid = int(pid), int(tid)
|
|
if len(self.stack) == 0:
|
|
self.stack[(self.pid, self.tid)] = []
|
|
elif (self.pid, self.tid) not in self.stack:
|
|
self.stack[(self.pid, self.tid)] = []
|
|
if not data.startswith(b'return ') and not data == b'return':
|
|
call = data.decode('utf-8')
|
|
#print(f'[{self.pid}][{self.tid}] {call}')
|
|
func_name = call[:call.find('(')]
|
|
ret = call[call.rfind(': '):]
|
|
m = RET_ADDR_RE.match(ret)
|
|
g_ret, _, g_fname, g_rel, _, g_sym, _, _, src_fname, src_line = m.groups()
|
|
self.ret_addr = int(g_ret, 0)
|
|
self.rel_ret_addr = int(g_rel, 0)
|
|
self.dli_file_name = g_fname
|
|
self.dli_sym_name = g_sym
|
|
self.src_file_name = src_fname
|
|
self.src_line_num = int(src_line, 0) if src_line else None
|
|
args, _ = Parser.parse_args(call[call.find('(') + 1:call.rfind(': ') - 1])
|
|
self.stack[(self.pid, self.tid)].append(
|
|
(self.ret_addr, self.rel_ret_addr,
|
|
self.dli_file_name, self.dli_sym_name,
|
|
self.src_file_name, self.src_line_num,
|
|
func_name, args))
|
|
if func_name == 'fork':
|
|
self.stack[(0, 0)] = self.stack[(self.pid, self.tid)][:]
|
|
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'
|
|
if self.wfile:
|
|
#print(f'[{self.pid}][{self.tid}] -> {command}')
|
|
self.wfile.write(command.encode('utf-8') + b'\n')
|
|
else:
|
|
ret = data.decode('utf-8')
|
|
ret = ret.split(';', 1)
|
|
other_vals = ret[1].strip() if len(ret) > 1 else ''
|
|
ret_value, _ = Parser.parse_arg(ret[0][7:])
|
|
kwargs = {}
|
|
if other_vals.startswith('optind '):
|
|
ret = other_vals[7:].split(';', 1)
|
|
kwargs['optind'] = int(ret[0].strip())
|
|
other_vals = ret[1].strip() if len(ret) > 1 else ''
|
|
if other_vals.startswith('errno '):
|
|
ret = other_vals[6:].split(';', 1)
|
|
kwargs['errno'] = ret[0].strip()
|
|
other_vals = ret[1].strip() if len(ret) > 1 else ''
|
|
if len(other_vals) > 0:
|
|
kwargs, _ = Parser.parse_args(other_vals, named=True, ret=True)
|
|
if (self.pid, self.tid) not in self.stack:
|
|
self.stack[(self.pid, self.tid)] = self.stack[(0, 0)]
|
|
del self.stack[(0, 0)]
|
|
(self.ret_addr, self.rel_ret_addr,
|
|
self.dli_file_name, self.dli_sym_name,
|
|
self.src_file_name, self.src_line_num,
|
|
func_name, args) = self.stack[(self.pid, self.tid)].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, **kwargs)
|
|
else:
|
|
func(*args, ret_value=ret_value, **kwargs)
|
|
except NotImplementedError:
|
|
if ret_value is None:
|
|
self.after_fallback(func_name, *args, **kwargs)
|
|
else:
|
|
self.after_fallback(func_name, *args, ret_value=ret_value, **kwargs)
|
|
#print(f'[{self.pid}][{self.tid}] -> {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, optind: int = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_exit(self, status: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_exit(self, status: int) -> None:
|
|
raise NotImplementedError()
|
|
def before_read(self, fildes: int, buf_ptr: Pointer, nbyte: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_read(self, fildes: int, buf_ptr: Pointer, nbyte: int,
|
|
ret_value: int, errno: str = None, buf: PointerTo[bytes] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_pread(self, fildes: int, buf_ptr: Pointer, nbyte: int, offset: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_pread(self, fildes: int, buf_ptr: Pointer, nbyte: int, offset: int,
|
|
ret_value: int, errno: str = None, buf: PointerTo[bytes] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_write(self, fildes: int, buf: PointerTo[bytes], nbyte: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_write(self, fildes: int, buf: PointerTo[bytes], nbyte: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_pwrite(self, fildes: int, buf: PointerTo[bytes], nbyte: int, offste: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_pwrite(self, fildes: int, buf: PointerTo[bytes], nbyte: int, offste: int,
|
|
ret_value: int, errno: str = None) -> 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 before_fork(self) -> str:
|
|
raise NotImplementedError()
|
|
def after_fork(self,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_wait(self, stat_loc: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_wait(self, stat_loc: Pointer,
|
|
ret_value: int, errno: str = None, stat: int = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_waitpid(self, pid: int, stat_loc: Pointer, options: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_waitpid(self, pid: int, stat_loc: Pointer, options: Flags,
|
|
ret_value: int, errno: str = None, stat: int = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execl(self, pathname: PointerTo[bytes], *args: list[PointerTo[bytes]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execl(self, pathname: PointerTo[bytes], *args: list[PointerTo[bytes]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execlp(self, file: PointerTo[bytes], *args: list[PointerTo[bytes]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execlp(self, file: PointerTo[bytes], *args: list[PointerTo[bytes]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execle(self, pathname: PointerTo[bytes], *args: list[PointerTo[bytes]], envp: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execle(self, pathname: PointerTo[bytes], *args: list[PointerTo[bytes]], envp: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execv(self, pathname: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execv(self, pathname: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execvp(self, file: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execvp(self, file: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execvpe(self, file: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execvpe(self, file: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_execve(self, pathname: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_execve(self, pathname: PointerTo[bytes], argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_fexecve(self, fd: int, argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]]) -> str:
|
|
raise NotImplementedError()
|
|
def after_fexecve(self, fd: int, argv: PointerTo[list[PointerTo[bytes]]], envp: PointerTo[list[PointerTo[bytes]]],
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_pipe(self, fildes_ptr: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_pipe(self, fildes_ptr: Pointer,
|
|
ret_value: int, errno: str = None, fildes: list[int] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_dup(self, oldfd: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_dup(self, oldfd: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_dup2(self, oldfd: int, newfd: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_dup2(self, oldfd: int, newfd: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_dup3(self, oldfd: int, newfd: int, flags: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_dup3(self, oldfd: int, newfd: int, flags: Flags,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_socket(self, domain: Constant, socktype: Constant, protocol: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_socket(self, domain: Constant, socktype: Constant, protocol: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_bind(self, sockfd: int, address: PointerTo[StructSockAddr], address_len: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_bind(self, sockfd: int, address: PointerTo[StructSockAddr], address_len: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_listen(self, sockfd: int, backlog: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_listen(self, sockfd: int, backlog: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_accept(self, sockfd: int, address_ptr: Pointer, address_len_ptr: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_accept(self, sockfd: int, address_ptr: Pointer, address_len_ptr: Pointer,
|
|
ret_value: int, errno: str = None, address: PointerTo[StructSockAddr] = None, address_len: PointerTo[StructSockAddr] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_connect(self, sockfd: int, address: PointerTo[StructSockAddr], address_len: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_connect(self, sockfd: int, address: PointerTo[StructSockAddr], address_len: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_getaddrinfo(self, node: PointerTo[bytes], service: PointerTo[bytes], hints: PointerTo[list[StructAddrInfo]], res_ptr: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_getaddrinfo(self, node: PointerTo[bytes], service: PointerTo[bytes], hints: PointerTo[list[StructAddrInfo]], res_ptr: Pointer,
|
|
ret_value: Constant, errno: str = None, res: PointerTo[list[StructAddrInfo]] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_freeaddrinfo(self, res: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_freeaddrinfo(self, res: Pointer) -> None:
|
|
raise NotImplementedError()
|
|
def before_send(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_send(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_sendto(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags, dest_addr: PointerTo[StructSockAddr], addrlen: int) -> str:
|
|
raise NotImplementedError()
|
|
def after_sendto(self, sockfd: int, buf: PointerTo[bytes], length: int, flags: Flags, dest_addr: PointerTo[StructSockAddr], addrlen: int,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_sendmsg(self, sockfd: int, message: StructMsgHdr, flags: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_sendmsg(self, sockfd: int, message: StructMsgHdr, flags: Flags,
|
|
ret_value: int, errno: str = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_recv(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_recv(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags,
|
|
ret_value: int, errno: str = None, buf: PointerTo[bytes] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_recvfrom(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags, src_addr_ptr: Pointer, addrlen_ptr: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_recvfrom(self, sockfd: int, buf_ptr: Pointer, size: int, flags: Flags, src_addr_ptr: Pointer, addrlen_ptr: Pointer,
|
|
ret_value: int, errno: str = None, buf: PointerTo[bytes] = None, src_addr: PointerTo[StructSockAddr] = None, addrlen: int = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_recvmsg(self, sockfd: int, message_ptr: Pointer, flags: Flags) -> str:
|
|
raise NotImplementedError()
|
|
def after_recvmsg(self, sockfd: int, message_ptr: Pointer, flags: Flags,
|
|
ret_value: int, errno: str = None, message: PointerTo[StructMsgHdr] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_getline(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], stream: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_getline(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], stream: Pointer,
|
|
ret_value: int, errno: str = None, n: int = None, line: PointerTo[bytes] = None) -> None:
|
|
raise NotImplementedError()
|
|
def before_getdelim(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], delim: int, stream: Pointer) -> str:
|
|
raise NotImplementedError()
|
|
def after_getdelim(self, line_ptr: PointerTo[Pointer], n_ptr: PointerTo[int], delim: int, stream: Pointer,
|
|
ret_value: int, errno: str = None, n: int = None, line: PointerTo[bytes] = None) -> None:
|
|
raise NotImplementedError()
|
|
|
|
|
|
class Handler(StreamRequestHandler, Parser):
|
|
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' ', 3)[3].strip().split(b';')]}
|
|
self.pid = int(meta['PID']) if 'PID' in meta else None
|
|
self.path = meta['PATH'] if 'PATH' in meta else None
|
|
print(f'Process with PID {self.pid} connected ({self.path})')
|
|
self.stack = {}
|
|
self.parse()
|
|
|
|
|
|
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
|