lorenz.stechauner/BSc-Thesis
1
0
Fork 0
Code Activity

proj: Implement fork/exec/pipe

Browse Source
This commit is contained in:
Lorenz Stechauner
2025-04-02 10:37:49 +02:00
parent e9389d4187
commit d4575728b5
4 changed files with 824 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
proj/server/src/intercept/__init__.py
View File

@@ -90,7 +90,7 @@ class Handler(StreamRequestHandler):
def parse_arg(argument: str) -> tuple[any, int]:
if argument == '':
return None, 0
m = re.match(r'^\s*\(nil\)\s*(,|$)', argument)
m = re.match(r'^\s*(\(nil\)|NULL|null|nullptr)\s*(,|$)', argument)
if m:
return 0, len(m.group(0))
m = re.match(r'^\s*(.*?)([,:]|$)', argument)
@@ -232,6 +232,30 @@ class Handler(StreamRequestHandler):
def after_getopt(self, argc: int, argv: PointerTo[list[PointerTo[bytes]]], optstring: PointerTo[bytes],
ret_value: int) -> 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,
@@ -317,6 +341,81 @@ class Handler(StreamRequestHandler):
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: PointerTo[list[int]]) -> str:
raise NotImplementedError()
def after_pipe(self, fildes: PointerTo[list[int]],
ret_value: int, errno: str = 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 intercept(socket: str, handler: type[Handler]) -> None: