#!/usr/bin/env python3 # ============================================================ # HOSTISH ACCOUNT # ============================================================ USERNAME = "your_username" PASSWORD = "your_password" # ============================================================ # LOCAL MINECRAFT SERVER # ============================================================ MINECRAFT_HOST = "127.0.0.1" MINECRAFT_PORT = 25565 # ============================================================ # OPTIONAL AUTOMATIC SERVER START # ============================================================ # # Leave blank to start Minecraft manually: # # START_BAT = r"" # # Or: # # START_BAT = r"C:\path\to\server\start.bat" # START_BAT = r"" AUTO_START_SERVER = True SERVER_START_TIMEOUT = 120 SERVER_START_CHECK_INTERVAL = 2 # ============================================================ # OPTIONAL WEB HOMEPAGE # ============================================================ # # Example: # # HOMEPAGE_FILE = r"C:\path\to\home.html" # # Leave blank to disable. # HOMEPAGE_FILE = r"" # ============================================================ # INTERNAL SETTINGS # ============================================================ import asyncio import json import os import ssl import subprocess import sys import time from datetime import datetime HOSTISH_URL = ( "wss://hostish.site/ws/minecraft" ) AGENT_VERSION = 4 RECONNECT_DELAY = 3 MAX_RECONNECT_DELAY = 15 TCP_CHUNK_SIZE = ( 64 * 1024 ) MAX_WEBSOCKET_MESSAGE = ( 4 * 1024 * 1024 ) MAX_HOMEPAGE_BYTES = ( 512 * 1024 ) MAGIC = b"MC" HEADER_SIZE = 18 # ============================================================ # DEPENDENCIES # ============================================================ _missing = [] try: import websockets except ImportError: websockets = None _missing.append( "websockets" ) try: import certifi except ImportError: certifi = None _missing.append( "certifi" ) if _missing: print() print( "Hostish Minecraft Agent" ) print() print( "Missing Python package(s): " + ", ".join( _missing ) ) print() print( "Install them with:" ) print() print( " python -m pip install " "websockets certifi" ) print() sys.exit( 1 ) # ============================================================ # LOGGING # ============================================================ def log( message: str, ) -> None: timestamp = ( datetime.now().strftime( "%H:%M:%S" ) ) print( f"[{timestamp}] {message}", flush=True, ) # ============================================================ # TLS # ============================================================ def create_ssl_context( ) -> ssl.SSLContext: context = ( ssl.create_default_context( cafile= certifi.where() ) ) context.check_hostname = ( True ) context.verify_mode = ( ssl.CERT_REQUIRED ) return context # ============================================================ # START.BAT PATH # ============================================================ def normalized_start_bat( ) -> str: configured = str( START_BAT or "" ).strip() if not configured: return "" return os.path.abspath( os.path.expanduser( configured ) ) # ============================================================ # VALIDATION # ============================================================ def validate_settings( ) -> bool: problems = [] username = str( USERNAME or "" ).strip() password = str( PASSWORD or "" ) if ( not username or username == "your_username" ): problems.append( 'Set USERNAME = ' '"your_username" ' "at the top of the file." ) if ( not password or password == "your_password" ): problems.append( 'Set PASSWORD = ' '"your_password" ' "at the top of the file." ) try: port = int( MINECRAFT_PORT ) if ( port < 1 or port > 65535 ): raise ValueError except Exception: problems.append( "MINECRAFT_PORT must be " "a number from 1 to 65535." ) start_bat = ( normalized_start_bat() ) if ( AUTO_START_SERVER and start_bat and not os.path.isfile( start_bat ) ): problems.append( "START_BAT does not exist:\n" f" {start_bat}" ) if problems: print() print( "Hostish Minecraft Agent " "is not configured correctly." ) print() for problem in problems: print( f" - {problem}" ) print() print( "Edit this Python file, " "save it, and run it again." ) print() return False return True # ============================================================ # HOMEPAGE # ============================================================ def load_homepage( ) -> tuple[ bool, str, str, ]: configured = str( HOMEPAGE_FILE or "" ).strip() if not configured: return ( False, "", "", ) path = os.path.abspath( os.path.expanduser( configured ) ) try: size = os.path.getsize( path ) except OSError as error: raise RuntimeError( "HOMEPAGE_FILE could not " f"be opened: {path} ({error})" ) from error if ( size > MAX_HOMEPAGE_BYTES ): raise RuntimeError( "HOMEPAGE_FILE is too large. " "Maximum size is " f"{MAX_HOMEPAGE_BYTES // 1024} KB." ) try: with open( path, "r", encoding="utf-8-sig", ) as file: html = ( file.read() ) except UnicodeError as error: raise RuntimeError( "HOMEPAGE_FILE must be UTF-8 HTML." ) from error except OSError as error: raise RuntimeError( "Could not read HOMEPAGE_FILE: " f"{error}" ) from error if not html.strip(): raise RuntimeError( "HOMEPAGE_FILE is empty." ) return ( True, html, path, ) # ============================================================ # LOCAL SERVER CHECK # ============================================================ async def test_local_server( timeout: float = 3, ) -> bool: try: ( _reader, writer, ) = await asyncio.wait_for( asyncio.open_connection( MINECRAFT_HOST, int( MINECRAFT_PORT ), ), timeout= timeout, ) writer.close() try: await writer.wait_closed() except Exception: pass return True except Exception: return False # ============================================================ # START.BAT # ============================================================ def start_minecraft_server( ) -> bool: start_bat = ( normalized_start_bat() ) if not start_bat: log( "START_BAT is blank; " "Minecraft will not be " "started automatically." ) return False if not os.path.isfile( start_bat ): log( "Could not start Minecraft " "server: START_BAT was not " f"found: {start_bat}" ) return False working_directory = ( os.path.dirname( start_bat ) ) log( "Starting Minecraft server..." ) log( f"Start script: {start_bat}" ) try: if os.name == "nt": subprocess.Popen( [ "cmd.exe", "/c", start_bat, ], cwd= working_directory, creationflags= subprocess.CREATE_NEW_CONSOLE, ) else: subprocess.Popen( [ "/bin/sh", start_bat, ], cwd= working_directory, start_new_session=True, ) return True except Exception as error: log( "Could not start Minecraft " "server: " f"{type(error).__name__}: " f"{error}" ) return False async def wait_for_minecraft_server( ) -> bool: timeout = max( 1, int( SERVER_START_TIMEOUT ), ) interval = max( 0.5, float( SERVER_START_CHECK_INTERVAL ), ) started_at = ( time.monotonic() ) log( "Waiting for Minecraft server " f"{MINECRAFT_HOST}:" f"{MINECRAFT_PORT}..." ) while True: if await test_local_server( timeout=2 ): log( "Minecraft server is online " f"after " f"{time.monotonic() - started_at:.1f} " "seconds." ) return True if ( time.monotonic() - started_at >= timeout ): log( "Minecraft server did not " "become available within " f"{timeout} seconds." ) return False await asyncio.sleep( interval ) async def ensure_minecraft_server( ) -> bool: log( "Checking local Minecraft server " f"{MINECRAFT_HOST}:" f"{MINECRAFT_PORT}..." ) if await test_local_server(): log( "Local Minecraft server detected." ) log( "Server is already running; " "START_BAT will not be launched." ) return True log( "No Minecraft server is currently " "responding on " f"{MINECRAFT_HOST}:" f"{MINECRAFT_PORT}." ) if not AUTO_START_SERVER: log( "Automatic server start is disabled." ) return False if not normalized_start_bat(): log( "START_BAT is blank. " "Start Minecraft manually " "before players join." ) return False if not start_minecraft_server(): return False return await wait_for_minecraft_server() # ============================================================ # PLAYER CONNECTION # ============================================================ class MinecraftConnection: def __init__( self, connection_id: str, websocket, send_lock: asyncio.Lock, ): self.id = ( connection_id ) self.websocket = ( websocket ) self.send_lock = ( send_lock ) self.reader = None self.writer = None self.read_task = None self.closed = False async def send_json( self, data: dict, ) -> None: async with self.send_lock: await self.websocket.send( json.dumps( data, separators=( ",", ":", ), ) ) async def send_binary( self, data: bytes, ) -> None: async with self.send_lock: await self.websocket.send( data ) async def open( self, ) -> bool: try: ( self.reader, self.writer, ) = await asyncio.wait_for( asyncio.open_connection( MINECRAFT_HOST, int( MINECRAFT_PORT ), ), timeout=8, ) await self.send_json( { "type": "tcp_open_result", "id": self.id, "ok": True, } ) log( "Player connected | " f"{self.id[:8]}" ) self.read_task = ( asyncio.create_task( self.minecraft_to_hostish() ) ) return True except Exception as error: try: await self.send_json( { "type": "tcp_open_result", "id": self.id, "ok": False, "error": str( error ), } ) except Exception: pass log( "Could not connect player " "to local Minecraft server: " f"{error}" ) return False async def minecraft_to_hostish( self, ) -> None: try: connection_bytes = ( bytes.fromhex( self.id ) ) if ( len( connection_bytes ) != 16 ): raise ValueError( "Hostish returned an " "invalid connection ID" ) prefix = ( MAGIC + connection_bytes ) while not self.closed: data = ( await self.reader.read( TCP_CHUNK_SIZE ) ) if not data: break await self.send_binary( prefix + data ) except asyncio.CancelledError: return except Exception as error: if not self.closed: log( "Player connection error | " f"{self.id[:8]} | " f"{error}" ) finally: await self.close( notify_hostish=True ) async def hostish_to_minecraft( self, data: bytes, ) -> None: if ( self.closed or self.writer is None or not data ): return try: self.writer.write( data ) await self.writer.drain() except Exception: await self.close( notify_hostish=True ) async def close( self, notify_hostish: bool = False, ) -> None: if self.closed: return self.closed = True if ( self.read_task is not None and self.read_task is not asyncio.current_task() ): self.read_task.cancel() if self.writer is not None: try: self.writer.close() await self.writer.wait_closed() except Exception: pass if notify_hostish: try: await self.send_json( { "type": "tcp_close", "id": self.id, } ) except Exception: pass log( "Player disconnected | " f"{self.id[:8]}" ) # ============================================================ # HOMEPAGE SEND # ============================================================ async def send_homepage_state( websocket, send_lock: asyncio.Lock, ) -> None: ( enabled, html, path, ) = load_homepage() async with send_lock: await websocket.send( json.dumps( { "type": "homepage_update", "enabled": enabled, "html": html if enabled else "", }, separators=( ",", ":", ), ) ) if enabled: log( "Browser homepage sent " f"to Hostish: {path}" ) else: log( "Browser homepage disabled; " "state sent to Hostish." ) # ============================================================ # HOSTISH TUNNEL # ============================================================ async def run_tunnel( ) -> None: ssl_context = ( create_ssl_context() ) log( "Connecting to Hostish..." ) async with websockets.connect( HOSTISH_URL, ssl= ssl_context, max_size= MAX_WEBSOCKET_MESSAGE, ping_interval= 20, ping_timeout= 20, open_timeout= 15, close_timeout= 5, ) as websocket: send_lock = ( asyncio.Lock() ) connections = {} # ---------------------------------------------------- # Authenticate # ---------------------------------------------------- await websocket.send( json.dumps( { "username": str( USERNAME ).strip(), "password": str( PASSWORD ), "agent": "minecraft", "agent_version": AGENT_VERSION, }, separators=( ",", ":", ), ) ) response_raw = ( await websocket.recv() ) if not isinstance( response_raw, str, ): raise RuntimeError( "Invalid authentication response" ) response = ( json.loads( response_raw ) ) if ( response.get( "type" ) != "auth" or not response.get( "ok" ) ): raise RuntimeError( response.get( "error", "Authentication failed", ) ) hostname = ( response.get( "hostname", f"{USERNAME}.hostish.site", ) ) port = int( response.get( "port", 25565, ) or 25565 ) log( f"Authenticated as {USERNAME}" ) if port == 25565: log( "Minecraft address: " f"{hostname}" ) else: log( "Minecraft address: " f"{hostname}:" f"{port}" ) log( "Local server: " f"{MINECRAFT_HOST}:" f"{MINECRAFT_PORT}" ) # ---------------------------------------------------- # Homepage update # ---------------------------------------------------- await send_homepage_state( websocket, send_lock, ) log( "Minecraft tunnel is online." ) # ---------------------------------------------------- # Main message loop # ---------------------------------------------------- try: async for message in websocket: # ============================================ # Raw TCP # ============================================ if isinstance( message, bytes, ): if ( len( message ) < HEADER_SIZE ): continue if ( message[:2] != MAGIC ): continue connection_id = ( message[ 2:18 ].hex() ) payload = ( message[ HEADER_SIZE: ] ) connection = ( connections.get( connection_id ) ) if ( connection is not None ): await ( connection .hostish_to_minecraft( payload ) ) continue # ============================================ # JSON message # ============================================ try: event = ( json.loads( message ) ) except Exception: continue event_type = ( event.get( "type" ) ) # -------------------------------------------- # Homepage acknowledgement # -------------------------------------------- if ( event_type == "homepage_update_result" ): if event.get( "ok" ): if ( event.get( "state" ) == "enabled" ): log( "Browser homepage " "accepted by Hostish " f"({event.get('bytes', 0)} bytes)." ) if event.get( "url" ): log( "Browser URL: " f"{event['url']}" ) else: log( "Browser homepage " "disabled on Hostish." ) else: log( "Browser homepage rejected " "by Hostish: " f"{event.get('state', 'unknown error')}" ) continue # -------------------------------------------- # New Minecraft connection # -------------------------------------------- if ( event_type == "tcp_open" ): connection_id = str( event.get( "id", "", ) ) if not connection_id: continue old = ( connections.pop( connection_id, None, ) ) if old is not None: await old.close() connection = ( MinecraftConnection( connection_id, websocket, send_lock, ) ) connections[ connection_id ] = connection opened = ( await connection.open() ) if not opened: connections.pop( connection_id, None, ) continue # -------------------------------------------- # Connection closed # -------------------------------------------- if ( event_type == "tcp_close" ): connection_id = str( event.get( "id", "", ) ) connection = ( connections.pop( connection_id, None, ) ) if ( connection is not None ): await connection.close() continue # -------------------------------------------- # Keepalive # -------------------------------------------- if ( event_type == "ping" ): async with send_lock: await websocket.send( json.dumps( { "type": "pong" }, separators=( ",", ":", ), ) ) continue finally: if connections: await asyncio.gather( *( connection.close() for connection in connections.values() ), return_exceptions=True, ) # ============================================================ # MAIN # ============================================================ async def main( ) -> None: if not validate_settings(): return try: ( homepage_enabled, _html, _path, ) = load_homepage() except RuntimeError as error: print() log( "Homepage configuration error: " f"{error}" ) print() return print() log( "Hostish Minecraft Agent" ) server_online = ( await ensure_minecraft_server() ) if not server_online: log( "WARNING: Minecraft is not " "currently responding." ) log( "Hostish will still connect, " "but players cannot join until " "the Minecraft server is running." ) if homepage_enabled: log( "Optional browser homepage " "is enabled." ) delay = ( RECONNECT_DELAY ) while True: try: await run_tunnel() delay = ( RECONNECT_DELAY ) except asyncio.CancelledError: return except KeyboardInterrupt: return except ssl.SSLCertVerificationError as error: log( "TLS certificate verification " f"failed: {error}" ) log( "Make sure this computer's " "date/time is correct and " "update certifi with: " "python -m pip install -U certifi" ) except Exception as error: log( "Disconnected: " f"{type(error).__name__}: " f"{error}" ) log( "Reconnecting in " f"{delay} seconds..." ) await asyncio.sleep( delay ) delay = min( delay + 2, MAX_RECONNECT_DELAY, ) # ============================================================ # ENTRY # ============================================================ if __name__ == "__main__": try: asyncio.run( main() ) except KeyboardInterrupt: print() log( "Stopped." )