path: root/client/3rd-party/client.py

# 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