141 lines
5.3 KiB
Python
141 lines
5.3 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
|
|
from minecraft.networking.connection import Connection
|
|
from minecraft.networking.modern import (
|
|
CancelPostjoinDialog, ChatCommand, PlayShowDialog,
|
|
PlayLogin, SubmitPostjoinLoginDialog, SystemChat, TeleportPosition,
|
|
)
|
|
|
|
|
|
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:
|
|
return component_text(fields['translate']) + component_text(fields.get('with', []))
|
|
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 = os.environ.get('SERVER', 'games.protospace.ca')
|
|
port = int(os.environ.get('PORT', '25565'))
|
|
username = os.environ.get('PLAYER_NAME', 'mosfet1')
|
|
password = os.environ.get('GAME_PASSWORD', 'mosfetbot')
|
|
|
|
connection = Connection(server, port, username=username,
|
|
allowed_versions=['26.3'])
|
|
connection.authme_password = password
|
|
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
|
|
|
|
@connection.listener(PlayLogin)
|
|
def on_play_login(packet):
|
|
if global_state is not None:
|
|
global_state.connected = True
|
|
global_state.connection_status = 'in_game'
|
|
|
|
|
|
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(TeleportPosition)
|
|
def on_spawn(packet):
|
|
if global_state is not None:
|
|
global_state.player_position = (packet.x, packet.y, packet.z)
|
|
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)
|
|
if global_state is not None:
|
|
global_state.last_chat = message
|
|
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)
|
|
try:
|
|
while connection.connected:
|
|
if global_state is not None and not global_state.get('running', True):
|
|
break
|
|
thread = connection.networking_thread
|
|
if thread is None:
|
|
break
|
|
thread.join(timeout=0.5)
|
|
if connection.connected:
|
|
connection.disconnect()
|
|
if connection.exception:
|
|
raise connection.exception
|
|
print('Connection ended.', flush=True)
|
|
except KeyboardInterrupt:
|
|
print('Disconnecting...', flush=True)
|
|
connection.disconnect()
|
|
finally:
|
|
if connection.connected:
|
|
connection.disconnect()
|
|
if global_state is not None:
|
|
global_state.connection = False
|
|
global_state.connected = False
|
|
global_state.connection_status = 'disconnected'
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|