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

proj: Implement socket

Browse Source
This commit is contained in:
Lorenz Stechauner
2025-04-14 15:03:08 +02:00
parent d4575728b5
commit a738e3ee2b
3 changed files with 816 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -13,6 +13,12 @@ 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]})
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', {})
StructMsgHdr = TypedDict('StructMsgHdr', {})
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
@@ -396,10 +402,10 @@ class Handler(StreamRequestHandler):
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:
def before_pipe(self, fildes_ptr: Pointer) -> str:
raise NotImplementedError()
def after_pipe(self, fildes: PointerTo[list[int]],
ret_value: int, errno: str = None) -> None:
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()
@@ -416,6 +422,70 @@ class Handler(StreamRequestHandler):
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[StructAddrInfo], res: Pointer) -> str:
raise NotImplementedError()
def after_getaddrinfo(self, node: PointerTo[bytes], service: PointerTo[bytes], hints: PointerTo[StructAddrInfo], res: Pointer,
ret_value: int, errno: str = 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 intercept(socket: str, handler: type[Handler]) -> None: