# ai generated (i don't like python) import ctypes import os import platform from ctypes import ( Structure, Union, POINTER, c_uint8, c_uint16, c_uint32, c_uint64, c_bool, c_int, c_char, c_void_p, c_size_t, CFUNCTYPE, byref, cast ) from typing import List, Optional, Tuple, Union as PyUnion from dataclasses import dataclass from enum import IntEnum import sys # Type aliases byte = c_uint8 hash_t = byte * 32 public_key = byte * 32 private_key = byte * 32 signature_t = byte * 64 ipv4 = byte * 4 ipv6 = byte * 16 # Constants MAX_RECIPIENTS = 32 MOTD_SIZE = 128 DATA_SIZE = 65536 SIGN_PUBLICKEYBYTES = 32 SIGN_SECRETKEYBYTES = 64 class ClientError(IntEnum): SUCCESS = 0 NETWORK_FAIL = 1 INVALID_ADDRESS = 2 CONNECTION_FAILED = 3 SEND_FAILED = 4 RECV_FAILED = 5 SERVER_BUSY = 6 INVALID_RESPONSE = 7 DESERIALIZE_FAILED = 8 ALLOC_FAILED = 9 FILE_NOT_FOUND = 10 FILE_READ_FAILED = 11 BAD_ARGUMENT = 12 # C Structure definitions class RelayNeighbor(Structure): _fields_ = [ ('active', c_bool), ('has_ipv4', c_bool), ('has_ipv6', c_bool), ('v4_addr', ipv4), ('v6_addr', ipv6) ] class RelayCapabilities(Structure): _fields_ = [ ('retention_hours', c_uint16), ('can_store', c_bool), ('leading_zeros', byte), ('accepts_plain', c_bool) ] class RelayStatistics(Structure): _fields_ = [ ('neighbors', c_uint16), ('held_packets', c_uint64), ('quick_connect', RelayNeighbor * 4) ] class RelayInfo(Structure): _fields_ = [ ('major_version', byte), ('minor_version', byte), ('motd', c_char * MOTD_SIZE), ('capabilities', RelayCapabilities), ('statistics', RelayStatistics) ] class DataPacketContents(Structure): _fields_ = [ ('encrypted', c_bool), ('created_time_ms', c_uint64), ('author', public_key), ('author_signing_pub', public_key), ('recipient_count', c_uint8), ('recipients', public_key * MAX_RECIPIENTS), ('data', byte * DATA_SIZE), ('signature', signature_t) ] class DataPacket(Structure): _fields_ = [ ('nonce', c_uint32), ('hash', hash_t), ('contents', DataPacketContents) ] class KeyPair(Structure): _fields_ = [ ('encryption_pub', public_key), ('encryption_priv', private_key), ('signing_pub', byte * SIGN_PUBLICKEYBYTES), ('signing_priv', byte * SIGN_SECRETKEYBYTES) ] class DataPacketList(Structure): _fields_ = [ ('packets', POINTER(DataPacket)), ('count', c_uint16), ('new_offset', c_uint64) ] # Python dataclasses for easier usage @dataclass class PyRelayNeighbor: active: bool has_ipv4: bool has_ipv6: bool v4_addr: bytes v6_addr: bytes @classmethod def from_c(cls, c_neighbor: RelayNeighbor) -> 'PyRelayNeighbor': return cls( active=c_neighbor.active, has_ipv4=c_neighbor.has_ipv4, has_ipv6=c_neighbor.has_ipv6, v4_addr=bytes(c_neighbor.v4_addr), v6_addr=bytes(c_neighbor.v6_addr) ) def format_ip(self) -> str: if self.has_ipv4: return '.'.join(str(b) for b in self.v4_addr) elif self.has_ipv6: parts = [] for i in range(0, 16, 2): word = (self.v6_addr[i] << 8) | self.v6_addr[i + 1] parts.append(f'{word:x}') return ':'.join(parts) return 'N/A' @dataclass class PyRelayCapabilities: retention_hours: int can_store: bool leading_zeros: int accepts_plain: bool @dataclass class PyRelayStatistics: neighbors: int held_packets: int quick_connect: List[PyRelayNeighbor] @dataclass class PyRelayInfo: major_version: int minor_version: int motd: str capabilities: PyRelayCapabilities statistics: PyRelayStatistics @classmethod def from_c(cls, c_info: RelayInfo) -> 'PyRelayInfo': return cls( major_version=c_info.major_version, minor_version=c_info.minor_version, motd=c_info.motd.decode('utf-8', errors='ignore').rstrip('\x00'), capabilities=PyRelayCapabilities( retention_hours=c_info.capabilities.retention_hours, can_store=c_info.capabilities.can_store, leading_zeros=c_info.capabilities.leading_zeros, accepts_plain=c_info.capabilities.accepts_plain ), statistics=PyRelayStatistics( neighbors=c_info.statistics.neighbors, held_packets=c_info.statistics.held_packets, quick_connect=[PyRelayNeighbor.from_c(n) for n in c_info.statistics.quick_connect] ) ) @dataclass class PyDataPacketContents: encrypted: bool created_time_ms: int author: bytes author_signing_pub: bytes recipient_count: int recipients: List[bytes] data: bytes signature: bytes @dataclass class PyDataPacket: nonce: int hash: bytes contents: PyDataPacketContents @classmethod def from_c(cls, c_packet: DataPacket) -> 'PyDataPacket': recipients = [] for i in range(c_packet.contents.recipient_count): recipients.append(bytes(c_packet.contents.recipients[i])) return cls( nonce=c_packet.nonce, hash=bytes(c_packet.hash), contents=PyDataPacketContents( encrypted=c_packet.contents.encrypted, created_time_ms=c_packet.contents.created_time_ms, author=bytes(c_packet.contents.author), author_signing_pub=bytes(c_packet.contents.author_signing_pub), recipient_count=c_packet.contents.recipient_count, recipients=recipients, data=bytes(c_packet.contents.data), signature=bytes(c_packet.contents.signature) ) ) def to_c(self) -> DataPacket: c_packet = DataPacket() c_packet.nonce = self.nonce for i, b in enumerate(self.hash[:32]): c_packet.hash[i] = b c_packet.contents.encrypted = self.contents.encrypted c_packet.contents.created_time_ms = self.contents.created_time_ms c_packet.contents.recipient_count = self.contents.recipient_count for i, b in enumerate(self.contents.author[:32]): c_packet.contents.author[i] = b for i, b in enumerate(self.contents.author_signing_pub[:32]): c_packet.contents.author_signing_pub[i] = b for i, recipient in enumerate(self.contents.recipients[:MAX_RECIPIENTS]): for j, b in enumerate(recipient[:32]): c_packet.contents.recipients[i][j] = b for i, b in enumerate(self.contents.data[:DATA_SIZE]): c_packet.contents.data[i] = b for i, b in enumerate(self.contents.signature[:64]): c_packet.contents.signature[i] = b return c_packet @dataclass class PyKeyPair: encryption_pub: bytes encryption_priv: bytes signing_pub: bytes signing_priv: bytes @classmethod def from_c(cls, c_keypair: KeyPair) -> 'PyKeyPair': return cls( encryption_pub=bytes(c_keypair.encryption_pub), encryption_priv=bytes(c_keypair.encryption_priv), signing_pub=bytes(c_keypair.signing_pub), signing_priv=bytes(c_keypair.signing_priv) ) def to_c(self) -> KeyPair: c_keypair = KeyPair() for i, b in enumerate(self.encryption_pub[:32]): c_keypair.encryption_pub[i] = b for i, b in enumerate(self.encryption_priv[:32]): c_keypair.encryption_priv[i] = b for i, b in enumerate(self.signing_pub[:32]): c_keypair.signing_pub[i] = b for i, b in enumerate(self.signing_priv[:64]): c_keypair.signing_priv[i] = b return c_keypair @dataclass class PyDataPacketList: packets: List[PyDataPacket] count: int new_offset: int class MoonpasteClientError(Exception): '''Exception raised for moonpaste client errors''' def __init__(self, error_code: ClientError, message: str = ''): self.error_code = error_code self.message = message or f'Client error: {error_code.name}' super().__init__(self.message) def find_library() -> str: '''Find the moonpaste client library''' system = platform.system().lower() if system == 'linux': lib_names = ['libmpclient.so'] elif system == 'darwin': lib_names = ['libmpclient.dylib'] elif system == 'windows': lib_names = ['libmpclient.dll'] else: lib_names = ['libmpclient.so', 'libmpclient.dylib', 'libmpclient.dll'] search_paths = [ './', './lib/', '../lib/', '/usr/local/lib/', '/usr/lib/', os.environ.get('MOONPASTE_LIB_PATH', '') ] search_paths = [p for p in search_paths if p] for search_path in search_paths: for lib_name in lib_names: full_path = os.path.join(search_path, lib_name) if os.path.exists(full_path): return full_path raise FileNotFoundError(f'Could not find moonpaste client library. Searched for: {lib_names}') class MoonpasteFFI: '''Low-level FFI wrapper for the moonpaste client library''' def __init__(self): lib_path = find_library() self.lib = ctypes.CDLL(lib_path) self._setup_function_signatures() def _setup_function_signatures(self): '''Setup C function signatures''' # mp_get_server_info self.lib.mp_get_server_info.argtypes = [ ctypes.c_char_p, # ip c_int, # port POINTER(RelayInfo) # out_info ] self.lib.mp_get_server_info.restype = c_int # mp_get_server_neighbors self.lib.mp_get_server_neighbors.argtypes = [ ctypes.c_char_p, # ip c_int, # port byte, # count POINTER(POINTER(RelayNeighbor)), # out_neighbors POINTER(byte) # out_count ] self.lib.mp_get_server_neighbors.restype = c_int # mp_push_packet self.lib.mp_push_packet.argtypes = [ ctypes.c_char_p, # ip c_int, # port DataPacket, # packet POINTER(c_bool) # out_success ] self.lib.mp_push_packet.restype = c_int # mp_get_packet self.lib.mp_get_packet.argtypes = [ ctypes.c_char_p, # ip c_int, # port hash_t, # packet_hash POINTER(DataPacket) # out_packet ] self.lib.mp_get_packet.restype = c_int # mp_get_firehose_packets self.lib.mp_get_firehose_packets.argtypes = [ ctypes.c_char_p, # ip c_int, # port c_uint64, # offset c_uint16, # count POINTER(DataPacketList) # out_list ] self.lib.mp_get_firehose_packets.restype = c_int # mp_create_packet_from_bytes self.lib.mp_create_packet_from_bytes.argtypes = [ POINTER(byte), # input c_size_t, # input_len byte, # recipient_count POINTER(hash_t), # recipients KeyPair # author_keys ] self.lib.mp_create_packet_from_bytes.restype = DataPacket # mp_create_packet_from_file self.lib.mp_create_packet_from_file.argtypes = [ ctypes.c_char_p, # filename byte, # recipient_count POINTER(hash_t), # recipients KeyPair, # author_keys POINTER(DataPacket) # out_packet ] self.lib.mp_create_packet_from_file.restype = c_int # mp_run_pow self.lib.mp_run_pow.argtypes = [ DataPacket, # packet byte # leading_zeros ] self.lib.mp_run_pow.restype = DataPacket # mp_generate_keypair self.lib.mp_generate_keypair.argtypes = [] self.lib.mp_generate_keypair.restype = KeyPair class MoonpasteClient: '''High-level Python client for moonpaste''' def __init__(self): self.ffi = MoonpasteFFI() def get_server_info(self, ip: str, port: int) -> PyRelayInfo: '''Get server information and capabilities''' info = RelayInfo() result = self.ffi.lib.mp_get_server_info( ip.encode('utf-8'), port, byref(info) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) return PyRelayInfo.from_c(info) def get_neighbors(self, ip: str, port: int, count: int = 5) -> List[PyRelayNeighbor]: '''Get a list of neighbors from the server''' neighbors_ptr = POINTER(RelayNeighbor)() out_count = byte() result = self.ffi.lib.mp_get_server_neighbors( ip.encode('utf-8'), port, count, byref(neighbors_ptr), byref(out_count) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) neighbors = [] if out_count.value > 0 and neighbors_ptr: for i in range(out_count.value): neighbors.append(PyRelayNeighbor.from_c(neighbors_ptr[i])) # Free the allocated memory ctypes.c_void_p.in_dll(self.ffi.lib, 'free').argtypes = [c_void_p] ctypes.c_void_p.in_dll(self.ffi.lib, 'free')(cast(neighbors_ptr, c_void_p)) return neighbors def push_packet(self, ip: str, port: int, packet: PyDataPacket) -> bool: '''Push a packet to the server''' c_packet = packet.to_c() success = c_bool() result = self.ffi.lib.mp_push_packet( ip.encode('utf-8'), port, c_packet, byref(success) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) return success.value def get_packet(self, ip: str, port: int, packet_hash: bytes) -> PyDataPacket: '''Get a specific packet by its hash''' if len(packet_hash) != 32: raise ValueError('Hash must be exactly 32 bytes') c_hash = hash_t() for i, b in enumerate(packet_hash): c_hash[i] = b packet = DataPacket() result = self.ffi.lib.mp_get_packet( ip.encode('utf-8'), port, c_hash, byref(packet) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) return PyDataPacket.from_c(packet) def get_firehose_packets(self, ip: str, port: int, offset: int = 0, count: int = 10) -> PyDataPacketList: '''Get a batch of packets from the server's firehose stream''' packet_list = DataPacketList() result = self.ffi.lib.mp_get_firehose_packets( ip.encode('utf-8'), port, offset, count, byref(packet_list) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) packets = [] if packet_list.count > 0 and packet_list.packets: for i in range(packet_list.count): packets.append(PyDataPacket.from_c(packet_list.packets[i])) # Free the allocated memory safely try: import ctypes.util libc_name = ctypes.util.find_library('c') if libc_name: libc = ctypes.CDLL(libc_name) libc.free.argtypes = [c_void_p] libc.free(cast(packet_list.packets, c_void_p)) else: # Platform-specific fallback try: if sys.platform.startswith('linux'): libc = ctypes.CDLL('libc.so.6') elif sys.platform == 'darwin': libc = ctypes.CDLL('libc.dylib') elif sys.platform == 'win32': libc = ctypes.CDLL('msvcrt.dll') else: libc = None if libc: libc.free.argtypes = [c_void_p] libc.free(cast(packet_list.packets, c_void_p)) except OSError: pass # Can't free, but better than crash except: pass # Memory will be freed when process exits return PyDataPacketList( packets=packets, count=packet_list.count, new_offset=packet_list.new_offset ) def create_packet(self, message: PyUnion[str, bytes], recipients: Optional[List[bytes]] = None, author_keys: Optional[PyKeyPair] = None) -> PyDataPacket: '''Create a data packet from message content''' if isinstance(message, str): data = message.encode('utf-8') else: data = message if author_keys is None: author_keys = self.generate_keypair() c_keys = author_keys.to_c() # Convert data to ctypes array data_len = min(len(data), DATA_SIZE) c_data = (byte * data_len)(*data[:data_len]) # Handle recipients recipient_count = 0 c_recipients = None if recipients: recipient_count = min(len(recipients), MAX_RECIPIENTS) c_recipients = (hash_t * recipient_count)() for i, recipient in enumerate(recipients[:recipient_count]): if len(recipient) != 32: raise ValueError(f'Recipient {i} must be exactly 32 bytes') for j, b in enumerate(recipient): c_recipients[i][j] = b packet = self.ffi.lib.mp_create_packet_from_bytes( cast(c_data, POINTER(byte)), data_len, recipient_count, cast(c_recipients, POINTER(hash_t)) if c_recipients else None, c_keys ) return PyDataPacket.from_c(packet) def create_packet_from_file(self, filename: str, recipients: Optional[List[bytes]] = None, author_keys: Optional[PyKeyPair] = None) -> PyDataPacket: '''Create a data packet from file contents''' if author_keys is None: author_keys = self.generate_keypair() c_keys = author_keys.to_c() # Handle recipients recipient_count = 0 c_recipients = None if recipients: recipient_count = min(len(recipients), MAX_RECIPIENTS) c_recipients = (hash_t * recipient_count)() for i, recipient in enumerate(recipients[:recipient_count]): if len(recipient) != 32: raise ValueError(f'Recipient {i} must be exactly 32 bytes') for j, b in enumerate(recipient): c_recipients[i][j] = b packet = DataPacket() result = self.ffi.lib.mp_create_packet_from_file( filename.encode('utf-8'), recipient_count, cast(c_recipients, POINTER(hash_t)) if c_recipients else None, c_keys, byref(packet) ) if result != ClientError.SUCCESS: raise MoonpasteClientError(ClientError(result)) return PyDataPacket.from_c(packet) def run_proof_of_work(self, packet: PyDataPacket, leading_zeros: int) -> PyDataPacket: '''Run Proof-of-Work on a packet''' c_packet = packet.to_c() result_packet = self.ffi.lib.mp_run_pow(c_packet, leading_zeros) return PyDataPacket.from_c(result_packet) def generate_keypair(self) -> PyKeyPair: '''Generate a new encryption and signing keypair''' c_keypair = self.ffi.lib.mp_generate_keypair() return PyKeyPair.from_c(c_keypair) # Utility methods @staticmethod def hash_to_hex(hash_bytes: bytes) -> str: '''Convert a hash to hexadecimal string''' return hash_bytes.hex() @staticmethod def hex_to_hash(hex_string: str) -> bytes: '''Convert hexadecimal string to hash bytes''' if len(hex_string) != 64: raise ValueError('Hash hex string must be exactly 64 characters') return bytes.fromhex(hex_string) def create_and_push_packet(self, ip: str, port: int, message: PyUnion[str, bytes], leading_zeros: int = 0, recipients: Optional[List[bytes]] = None, author_keys: Optional[PyKeyPair] = None) -> Tuple[PyDataPacket, bool]: '''Create a packet, run PoW, and push it to the server''' # Create the packet packet = self.create_packet(message, recipients, author_keys) # Run proof of work if required if leading_zeros > 0: packet = self.run_proof_of_work(packet, leading_zeros) # Push to server success = self.push_packet(ip, port, packet) return packet, success