203 lines
8.4 KiB
Python
203 lines
8.4 KiB
Python
"""Minimal current-protocol connection mode for AuthMe / offline servers.
|
|
|
|
Run with SERVER=... PLAYER_NAME=mosfet1 GAME_PASSWORD=... python -m mosfet.connect
|
|
"""
|
|
import os
|
|
import threading
|
|
from minecraft.networking.connection import Connection
|
|
from minecraft.networking.modern import (
|
|
CancelPostjoinDialog, ChatCommand, PlayGameEventPacket, PlayShowDialog,
|
|
PlayLogin, SubmitPostjoinLoginDialog, SystemChat, TeleportPosition,
|
|
)
|
|
from mosfet.protocol.managers import WorldState
|
|
from mosfet.vector import Point3D
|
|
|
|
|
|
def component_text(value):
|
|
"""Best-effort flattening of modern anonymous-NBT chat components."""
|
|
if value is None:
|
|
return ''
|
|
if isinstance(value, (str, int, float)):
|
|
return str(value)
|
|
if isinstance(value, (list, tuple)):
|
|
return ''.join(component_text(item) for item in value)
|
|
raw = getattr(value, 'value', None)
|
|
if isinstance(raw, (str, int, float)):
|
|
return str(raw)
|
|
if raw is not None and raw is not value and hasattr(raw, 'items'):
|
|
value = raw
|
|
tags = getattr(value, 'tags', None)
|
|
if tags is not None:
|
|
value = tags
|
|
if hasattr(value, 'items'):
|
|
fields = dict(value.items())
|
|
if 'text' in fields:
|
|
return component_text(fields['text']) + component_text(fields.get('extra', []))
|
|
if 'translate' in fields:
|
|
translation = component_text(fields['translate'])
|
|
arguments = fields.get('with', [])
|
|
if translation == 'commands.message.display.incoming' and len(arguments) >= 2:
|
|
return '{} whispers to you: {}'.format(
|
|
component_text(arguments[0]),
|
|
component_text(arguments[1]))
|
|
return translation + component_text(arguments)
|
|
return ''.join(component_text(v) for v in fields.values())
|
|
if isinstance(value, (list, tuple)):
|
|
return ''.join(component_text(item) for item in value)
|
|
return str(value)
|
|
|
|
|
|
def main(global_state=None, *, server=None, port=None, username=None,
|
|
password=None):
|
|
server = server or os.environ.get('SERVER', '127.0.0.1')
|
|
port = int(port or os.environ.get('PORT', '25565'))
|
|
username = username or os.environ.get('PLAYER_NAME', 'mosfet1')
|
|
password = password or os.environ.get('GAME_PASSWORD', 'mosfetbot')
|
|
|
|
connection = Connection(server, port, username=username,
|
|
allowed_versions=['26.3'],
|
|
handle_exception=False)
|
|
connection.authme_password = password
|
|
connection.world_state = WorldState()
|
|
connection.dimension_types_by_id = []
|
|
sent_auth_commands = set()
|
|
if global_state is not None:
|
|
global_state.connection = connection
|
|
global_state.name = username
|
|
global_state.connected = False
|
|
global_state.connection_status = 'connecting'
|
|
global_state.player_position = None
|
|
global_state.world_state = connection.world_state
|
|
global_state.chunks = connection.world_state
|
|
from mosfet.protocol.managers import DataManager
|
|
global_state.mcdata = DataManager()
|
|
|
|
@connection.listener(PlayLogin)
|
|
def on_play_login(packet):
|
|
if global_state is not None:
|
|
global_state.connected = True
|
|
global_state.connection_status = 'in_game'
|
|
global_state.player_id = packet.player_id
|
|
global_state.game_mode = packet.game_mode
|
|
global_state.dimension = packet.dimension.removeprefix('minecraft:')
|
|
packet.render_distance = packet.chunk_radius
|
|
global_state.info = packet
|
|
global_state.eid = packet.player_id
|
|
global_state.command_lock = False
|
|
from mosfet.world import World
|
|
from mosfet.commands import Commands
|
|
from mosfet.protocol.modern import ModernChatManager
|
|
from mosfet.protocol.managers import (
|
|
ChestOpenManager, EntityManager, InventoryManager,
|
|
)
|
|
from mosfet.bot import start_automation
|
|
global_state.world = World(global_state)
|
|
global_state.chat = ModernChatManager(global_state)
|
|
global_state.entities = EntityManager(global_state)
|
|
global_state.inventory = InventoryManager(global_state)
|
|
global_state.chest_open_manager = ChestOpenManager(global_state)
|
|
global_state.commands = Commands(global_state)
|
|
start_automation(global_state)
|
|
|
|
|
|
def send_auth_command(action):
|
|
if action in sent_auth_commands:
|
|
return
|
|
sent_auth_commands.add(action)
|
|
command = '/register {0} {0}'.format(password) if action == 'register' \
|
|
else '/login ' + password
|
|
print('[AuthMe] sending ' + command.split()[0], flush=True)
|
|
connection.write_packet(ChatCommand(command=command.lstrip('/')),
|
|
force=True)
|
|
|
|
@connection.listener(PlayGameEventPacket)
|
|
def on_game_event(packet):
|
|
if global_state is not None and packet.event == 3:
|
|
global_state.game_mode = int(packet.value)
|
|
|
|
@connection.listener(TeleportPosition)
|
|
def on_spawn(packet):
|
|
if global_state is not None:
|
|
position = getattr(connection, 'player_position', None)
|
|
if position is None:
|
|
position = (packet.x, packet.y, packet.z,
|
|
packet.yaw, packet.pitch)
|
|
x, y, z, yaw, pitch = position
|
|
global_state.player_position = (x, y, z)
|
|
global_state.pos = Point3D(global_state.player_position)
|
|
global_state.yaw = yaw
|
|
global_state.pitch = pitch
|
|
if getattr(connection, 'authme_mode', 'register') == 'register':
|
|
send_auth_command('register')
|
|
|
|
@connection.listener(PlayShowDialog)
|
|
def on_authme_dialog(packet):
|
|
is_login = b'login' in packet.payload.lower()
|
|
print('[AuthMe] dismissing the post-join dialog to use /commands.',
|
|
flush=True)
|
|
if is_login:
|
|
connection.write_packet(SubmitPostjoinLoginDialog(
|
|
password=password), force=True)
|
|
else:
|
|
connection.write_packet(CancelPostjoinDialog(), force=True)
|
|
send_auth_command('register')
|
|
|
|
@connection.listener(SystemChat)
|
|
def on_system_chat(packet):
|
|
message = component_text(packet.content)
|
|
chat_handled = False
|
|
if global_state is not None:
|
|
global_state.last_chat = message
|
|
chat = global_state.get('chat')
|
|
if chat is not None:
|
|
chat_handled = chat.handle_system_chat(packet, message)
|
|
if not chat_handled:
|
|
print('[server] ' + message, flush=True)
|
|
lowered = message.lower()
|
|
if 'already registered' in lowered:
|
|
send_auth_command('login')
|
|
elif 'register' in lowered and any(word in lowered for word in
|
|
('password', 'account', '/register')):
|
|
send_auth_command('register')
|
|
elif any(word in lowered for word in
|
|
('please login', 'please log in', 'log in', '/login',
|
|
'authenticate', 'authentication required')):
|
|
send_auth_command('login')
|
|
|
|
connection.connect()
|
|
print('Connected as {} to {}:{} using offline protocol 777.'.format(
|
|
username, server, port), flush=True)
|
|
shutdown_requested = False
|
|
try:
|
|
while connection.connected:
|
|
if global_state is not None and not global_state.get('running', True):
|
|
shutdown_requested = True
|
|
break
|
|
thread = connection.networking_thread
|
|
if thread is None:
|
|
break
|
|
thread.join(timeout=0.5)
|
|
if connection.connected:
|
|
connection.disconnect(immediate=True)
|
|
if connection.exception and not shutdown_requested:
|
|
raise connection.exception
|
|
print('Connection ended.', flush=True)
|
|
except KeyboardInterrupt:
|
|
shutdown_requested = True
|
|
print('Disconnecting...', flush=True)
|
|
connection.disconnect()
|
|
finally:
|
|
if connection.connected:
|
|
connection.disconnect(immediate=True)
|
|
if global_state is not None:
|
|
worker = global_state.get('movement_thread')
|
|
if worker and worker is not threading.current_thread():
|
|
worker.join(timeout=1.0)
|
|
global_state.connection = False
|
|
global_state.connected = False
|
|
global_state.connection_status = 'disconnected'
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|