-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathssh_server.py
More file actions
805 lines (672 loc) · 30.1 KB
/
ssh_server.py
File metadata and controls
805 lines (672 loc) · 30.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
import asyncio
import sys
import os
from typing import Dict, Optional, Union, Any
try:
import asyncssh
SSH_AVAILABLE = True
except ImportError:
asyncssh = None
SSH_AVAILABLE = False
from colorama import Fore, Back, Style, init
import traceback
# Initialize colorama
init(autoreset=True)
class GameConnection:
"""Represents a connection to the game (SSH or direct)"""
def __init__(self, connection_type: str = "direct"):
self.connection_type = connection_type
self.user_id: Optional[str] = None
self.character = None
self.is_authenticated = False
self.is_in_character_creation = False
self.character_creation_session = None
self.ssh_process: Optional[Any] = None
self.reader: Optional[Any] = None # For TCP connections
self.writer: Optional[Any] = None # For TCP connections
self.should_disconnect = False # Flag to signal connection should be closed
self.has_entered_game = False # Flag to track if player has entered the game
self._just_entered_game = False
# Authentication state tracking
self.auth_state = "waiting_for_command" # waiting_for_command, waiting_for_username, waiting_for_password
self.auth_command: Optional[str] = None # "login" or "register"
self.username_buffer: Optional[str] = None
self.password_masking = False
async def send_message(self, message: str, color: str = "white"):
"""Send a message to the client with optional color"""
# Color mapping with black background
color_map = {
'red': Fore.RED + Back.BLACK,
'green': Fore.GREEN + Back.BLACK,
'blue': Fore.BLUE + Back.BLACK,
'yellow': Fore.YELLOW + Back.BLACK,
'cyan': Fore.CYAN + Back.BLACK,
'magenta': Fore.MAGENTA + Back.BLACK,
'white': Fore.WHITE + Back.BLACK,
'gold': Fore.YELLOW + Style.BRIGHT + Back.BLACK,
'bright_green': Fore.GREEN + Style.BRIGHT + Back.BLACK,
'bright_red': Fore.RED + Style.BRIGHT + Back.BLACK,
'dark_yellow': Fore.YELLOW + Style.DIM + Back.BLACK,
'light_green': Fore.GREEN + Style.BRIGHT + Back.BLACK
}
colored_message = color_map.get(color, Fore.WHITE) + message + Style.RESET_ALL
if self.connection_type == "ssh" and self.ssh_process:
self.ssh_process.stdout.write(colored_message + '\n')
else:
print(colored_message)
async def send_prompt(self, prompt: str):
"""Send a prompt without newline (placeholder)"""
pass
async def get_input(self, prompt: str = "") -> str:
"""Get input from the client"""
if self.connection_type == "ssh" and self.ssh_process:
if prompt:
self.ssh_process.stdout.write(prompt)
return await self.ssh_process.stdin.readline()
else:
return input(prompt)
class SSHGameSession:
"""SSH session handler for game connections"""
def __init__(self, game_server):
self.game_server = game_server
self.connection: Optional[GameConnection] = None
self.chan: Optional[Any] = None
def connection_made(self, chan):
"""Called when SSH connection is established"""
self.connection = GameConnection("ssh")
self.connection.ssh_process = self
self.chan = chan
print(f"SSH connection established from {chan.get_extra_info('peername')}")
# Start the game session
asyncio.create_task(self._start_game_session())
def data_received(self, data, datatype):
"""Handle incoming data from SSH client"""
if asyncssh and datatype == asyncssh.EXTENDED_DATA_STDERR:
return
# Check if connection should be disconnected
if self.connection and self.connection.should_disconnect:
if self.chan:
self.chan.close()
return
try:
text = data.decode('utf-8').strip()
if text:
asyncio.create_task(self._handle_input_and_check_disconnect(text))
except Exception as e:
print(f"Error processing SSH data: {e}")
async def _handle_input_and_check_disconnect(self, text):
"""Handle input and check for disconnect flag"""
try:
await self.game_server.handle_client_input(self.connection, text)
# Check if connection should be disconnected after processing input
if self.connection and self.connection.should_disconnect:
if self.chan:
self.chan.close()
except Exception as e:
print(f"Error handling SSH input: {e}")
def connection_lost(self, exc):
"""Called when SSH connection is lost"""
if self.connection and self.connection.user_id:
try:
asyncio.create_task(self.game_server.disconnect_player(self.connection.user_id))
except:
pass
print("SSH connection lost")
async def _start_game_session(self):
"""Start the game session for this SSH connection"""
try:
if self.connection:
await self.connection.send_message("=" * 60, "cyan")
await self.connection.send_message(" Welcome to SSH RPG - Text-Based MMORPG", "gold")
await self.connection.send_message("=" * 60, "cyan")
await self.connection.send_message("")
# Start authentication process
await self._handle_authentication()
except Exception as e:
if self.connection:
await self.connection.send_message(f"Error starting game session: {e}", "red")
print(f"Error in SSH game session: {e}")
traceback.print_exc()
async def _handle_authentication(self):
"""Handle user authentication"""
if self.connection:
await self.connection.send_message("Please login or create a new account.", "white")
await self.connection.send_message("Please enter username to login, otherwise type 'register' to create a new character", "yellow")
# SSH session interface methods
def pty_requested(self, term_type, term_size, term_modes):
"""Handle PTY request"""
return True
def shell_requested(self):
"""Handle shell request"""
return True
def exec_requested(self, command):
"""Handle exec request"""
return False
def subsystem_requested(self, subsystem):
"""Handle subsystem request"""
return False
def break_received(self, msec) -> bool:
"""Handle break signal"""
return True
def signal_received(self, signal):
"""Handle signal"""
if signal == 'INT':
if self.chan:
self.chan.close()
def terminal_size_changed(self, width, height, pixwidth, pixheight):
"""Handle terminal size change"""
pass
# Output methods for the game
def write(self, data):
"""Write data to SSH client"""
if isinstance(data, str):
data = data.encode('utf-8')
if self.chan:
self.chan.write(data)
def writelines(self, lines):
"""Write multiple lines to SSH client"""
for line in lines:
self.write(line + '\n')
def flush(self):
"""Flush output"""
pass
@property
def stdout(self):
"""Provide stdout-like interface"""
return self
@property
def stdin(self):
"""Provide stdin-like interface"""
return SSHInputStream(self.chan)
class SSHInputStream:
"""Provides stdin-like interface for SSH"""
def __init__(self, chan):
self.chan = chan
self._buffer = ""
self._lines = []
async def readline(self):
"""Read a line from SSH input"""
# This is a simplified implementation
# In a real implementation, you'd need to handle the SSH input stream properly
return ""
class SSHGameServer:
"""SSH server for the game"""
def __init__(self, game_server):
self.game_server = game_server
def session_requested(self):
"""Create new session for incoming SSH connection"""
return SSHGameSession(self.game_server)
def password_auth_supported(self):
"""Allow password authentication"""
return True
def validate_password(self, username, password):
"""Accept any username/password for game access"""
return True
def connection_requested(self, dest_host, dest_port, orig_host, orig_port):
"""Handle connection requests - reject all"""
return False
def server_requested(self, listen_host, listen_port):
"""Handle server requests - reject all"""
return False
async def unified_connection_handler(game_server, connection, input_reader_func, should_mask_password=False):
"""Unified connection handler for both SSH and TCP connections"""
try:
# Send welcome messages
await connection.send_message("=" * 60)
await connection.send_message(" Welcome to SSH RPG - Text-Based MMORPG")
await connection.send_message("=" * 60)
await connection.send_message("")
await connection.send_message("Please login or create a new account.")
await connection.send_message("Please enter username to login, otherwise type 'register' to create a new character")
# Handle client input loop with proper MUD prompt timing
while True:
try:
# Check if connection should be disconnected
if connection.should_disconnect:
break
# Get prompt if player is in game and authenticated
# But only if the game engine hasn't already sent one recently
prompt = None
if (connection.is_authenticated and
not connection.is_in_character_creation and
hasattr(connection, 'has_entered_game') and connection.has_entered_game and
not getattr(connection, 'password_masking', False) and
not getattr(connection, '_prompt_sent_by_engine', False)):
print(f"DEBUG {connection.connection_type.upper()}: Getting prompt for player {getattr(connection, 'user_id', 'unknown')}")
prompt = await game_server.get_player_prompt(connection)
# Clear the engine prompt flag after checking
if hasattr(connection, '_prompt_sent_by_engine'):
connection._prompt_sent_by_engine = False
# Read input using the provided reader function with prompt
command = await input_reader_func(prompt)
if command is None: # Connection closed
break
# Process the command immediately after receiving it
if command:
await game_server.handle_client_input(connection, command)
# Check again after processing command
if connection.should_disconnect:
break
except asyncio.CancelledError:
break
except Exception as e:
print(f"Error handling {connection.connection_type} client input: {e}")
break
except Exception as e:
print(f"{connection.connection_type} client error: {e}")
traceback.print_exc()
finally:
try:
if connection and connection.user_id:
await game_server.disconnect_player(int(connection.user_id))
except:
pass
def handle_ssh_client_with_server(game_server):
"""Create SSH client handler with game_server instance"""
async def handle_ssh_client(process):
"""Handle SSH client connections using process factory"""
connection = None
try:
# Get connection info
username = process.get_extra_info('username')
peername = process.get_extra_info('peername')
# Create a game connection wrapper for SSH
connection = SSHProcessConnection(process)
# Define SSH-specific input reader
async def ssh_input_reader(prompt=None):
# Use the connection's get_input method which handles prompts
if prompt:
return await connection.get_input(prompt)
else:
line = await process.stdin.readline()
if not line:
return None
return line.decode('utf-8').strip() if isinstance(line, bytes) else line.strip()
# Use unified handler
await unified_connection_handler(game_server, connection, ssh_input_reader)
except Exception as e:
print(f"SSH client error: {e}")
traceback.print_exc()
finally:
try:
if connection:
await connection.send_message("Connection closed.", "yellow")
except:
pass
try:
process.exit(0)
except:
pass
return handle_ssh_client
async def handle_ssh_client(process):
"""Handle SSH client connections using process factory (fallback)"""
connection = None
try:
# Get connection info
username = process.get_extra_info('username')
peername = process.get_extra_info('peername')
# Create a game connection wrapper for SSH
connection = SSHProcessConnection(process)
# Import here to avoid circular imports
from server import game_server
# Send welcome messages
await connection.send_message("=" * 60)
await connection.send_message(" Welcome to SSH RPG - Text-Based MMORPG")
await connection.send_message("=" * 60)
await connection.send_message("")
await connection.send_message("Please login or create a new account.")
await connection.send_message("Please enter username to login, otherwise type 'register' to create a new character")
# Handle client input loop
while True:
try:
# Check if connection should be disconnected
if connection.should_disconnect:
break
# Read input from SSH process
line = await process.stdin.readline()
if not line:
break
command = line.decode('utf-8').strip() if isinstance(line, bytes) else line.strip()
if command:
await game_server.handle_client_input(connection, command)
# Clear the just entered flag after any command (including empty ones)
if hasattr(connection, '_just_entered_game'):
connection._just_entered_game = False
# Note: Prompt sending is handled by handle_ssh_client_with_server
# This function is kept for compatibility but should not send prompts
# Check again after processing command
if connection.should_disconnect:
break
except asyncio.CancelledError:
break
except Exception as e:
print(f"Error handling SSH client input: {e}")
break
except Exception as e:
print(f"SSH client error: {e}")
traceback.print_exc()
finally:
try:
if connection and connection.user_id:
from server import game_server
await game_server.disconnect_player(int(connection.user_id))
except:
pass
try:
process.exit(0)
except:
pass
class SSHProcessConnection(GameConnection):
"""Game connection wrapper for SSH process"""
def __init__(self, process):
super().__init__("ssh")
self.process = process
self.ssh_process = process # Set this for base class compatibility
self._input_buffer = ""
async def send_message(self, message: str, color: str = "white"):
"""Send message to SSH client"""
try:
# Apply color formatting
colors = {
"red": Fore.RED,
"green": Fore.GREEN,
"blue": Fore.BLUE,
"yellow": Fore.YELLOW,
"magenta": Fore.MAGENTA,
"cyan": Fore.CYAN,
"white": Fore.WHITE,
"bright_red": Fore.LIGHTRED_EX,
"bright_green": Fore.LIGHTGREEN_EX,
"bright_blue": Fore.LIGHTBLUE_EX,
"bright_yellow": Fore.LIGHTYELLOW_EX,
"bright_magenta": Fore.LIGHTMAGENTA_EX,
"bright_cyan": Fore.LIGHTCYAN_EX,
"bright_white": Fore.LIGHTWHITE_EX
}
color_code = colors.get(color, Fore.WHITE)
formatted_message = f"{color_code}{message}{Style.RESET_ALL}\n"
self.process.stdout.write(formatted_message)
except Exception as e:
print(f"Error sending SSH message: {e}")
async def send_prompt(self, prompt: str):
"""Send a prompt without newline for bash-like behavior"""
try:
# Send prompt without newline to create proper prompt behavior
self.process.stdout.write(prompt)
await self.process.stdout.drain() # Ensure prompt is sent immediately
except Exception as e:
print(f"Error sending SSH prompt: {e}")
async def get_input(self, prompt: str = "") -> str:
"""Get input from SSH client"""
try:
if prompt:
self.process.stdout.write(prompt)
# Read line from stdin
line = await self.process.stdin.readline()
return line.strip() if line else ""
except Exception as e:
print(f"Error getting SSH input: {e}")
return ""
# Create SSH server class with proper inheritance
class SSHGameServerAuth:
"""SSH server with authentication"""
def __new__(cls):
if SSH_AVAILABLE and asyncssh is not None:
# Create a dynamic class that inherits from asyncssh.SSHServer
class _SSHGameServerAuth(asyncssh.SSHServer):
def password_auth_supported(self):
"""Allow password authentication"""
return True
def validate_password(self, username, password):
"""Accept any username/password for game access"""
return True
def connection_made(self, conn):
"""Called when a connection is made"""
pass
return _SSHGameServerAuth()
else:
# Return a basic object when asyncssh is not available
return super().__new__(cls)
def password_auth_supported(self):
"""Allow password authentication"""
return True
def validate_password(self, username, password):
"""Accept any username/password for game access"""
return True
def connection_made(self, conn):
"""Called when a connection is made"""
pass
async def start_ssh_server(game_server, host='localhost', port=2222):
"""Start the SSH server"""
if not SSH_AVAILABLE or asyncssh is None:
print("SSH support not available. Install asyncssh to enable SSH connections.")
return None
try:
# Generate or load persistent host key
host_key_file = "server_host_key"
if os.path.exists(host_key_file):
# Load existing host key
try:
with open(host_key_file, 'r') as f:
host_key_data = f.read()
host_key = asyncssh.import_private_key(host_key_data)
print(f"Loaded existing SSH host key from {host_key_file}")
except Exception as e:
print(f"Failed to load existing host key: {e}")
print("Generating new host key...")
host_key = asyncssh.generate_private_key('ssh-rsa')
# Save the new key
with open(host_key_file, 'w') as f:
f.write(host_key.export_private_key().decode())
print(f"New SSH host key saved to {host_key_file}")
else:
# Generate new host key and save it
host_key = asyncssh.generate_private_key('ssh-rsa')
with open(host_key_file, 'w') as f:
f.write(host_key.export_private_key().decode())
print(f"Generated and saved new SSH host key to {host_key_file}")
# Create and start SSH server using process factory
server = await asyncssh.create_server(
SSHGameServerAuth,
host=host,
port=port,
server_host_keys=[host_key],
process_factory=handle_ssh_client_with_server(game_server)
)
print(f"SSH server started on {host}:{port}")
return server
except Exception as e:
print(f"Failed to start SSH server: {e}")
traceback.print_exc()
return None
# Alternative simple SSH server using asyncio
class SimpleSSHServer:
"""Simple SSH-like server using raw TCP sockets"""
def __init__(self, game_server):
self.game_server = game_server
self.server = None
async def start(self, host='localhost', port=2223):
"""Start the simple TCP server"""
try:
self.server = await asyncio.start_server(
self._handle_client, host, port
)
print(f"Simple TCP server started on {host}:{port}")
print(f"Players can connect with: telnet {host} {port}")
except Exception as e:
print(f"Failed to start TCP server: {e}")
traceback.print_exc()
async def _handle_client(self, reader, writer):
"""Handle a new TCP client connection"""
connection = None
try:
connection = GameConnection("tcp")
connection.reader = reader
connection.writer = writer
# Override send_message for TCP with color support
async def tcp_send_message(message: str, color: str = "white"):
try:
# Color mapping with black background
color_map = {
'red': Fore.RED + Back.BLACK,
'green': Fore.GREEN + Back.BLACK,
'blue': Fore.BLUE + Back.BLACK,
'yellow': Fore.YELLOW + Back.BLACK,
'cyan': Fore.CYAN + Back.BLACK,
'magenta': Fore.MAGENTA + Back.BLACK,
'white': Fore.WHITE + Back.BLACK,
'gold': Fore.YELLOW + Style.BRIGHT + Back.BLACK,
'bright_green': Fore.GREEN + Style.BRIGHT + Back.BLACK,
'bright_red': Fore.RED + Style.BRIGHT + Back.BLACK,
'dark_yellow': Fore.YELLOW + Style.DIM + Back.BLACK,
'light_green': Fore.GREEN + Style.BRIGHT + Back.BLACK
}
colored_message = color_map.get(color, Fore.WHITE + Back.BLACK) + message + Style.RESET_ALL
writer.write((colored_message + '\n').encode('utf-8'))
await writer.drain()
except:
pass
# Override send_prompt for TCP
async def tcp_send_prompt(prompt: str):
try:
# Send prompt without newline for bash-like behavior
writer.write(prompt.encode('utf-8'))
await writer.drain()
except:
pass
connection.send_message = tcp_send_message
connection.send_prompt = tcp_send_prompt
# Define TCP-specific input reader
async def tcp_input_reader(prompt=None):
# Send prompt if provided
if prompt:
await connection.send_prompt(prompt)
if connection.password_masking:
# Handle password input with masking
command = await self._read_password_input(reader, writer)
connection.password_masking = False # Reset after reading
return command
else:
# Normal input handling
data = await reader.readline()
if not data:
return None
return data.decode('utf-8').strip()
# Use unified handler
await unified_connection_handler(self.game_server, connection, tcp_input_reader)
except Exception as e:
print(f"Error in TCP client handler: {e}")
traceback.print_exc()
finally:
try:
if connection and connection.user_id:
await self.game_server.disconnect_player(int(connection.user_id))
except:
pass
try:
if writer:
writer.close()
await writer.wait_closed()
except:
pass
async def _read_password_input(self, reader, writer):
"""Read password input with asterisk masking"""
password = ""
while True:
try:
# Read one character at a time
data = await reader.read(1)
if not data:
break
char = data.decode('utf-8')
# Handle different input characters
if char == '\r' or char == '\n':
# Enter pressed, finish input
writer.write(b'\n')
await writer.drain()
break
elif char == '\x7f' or char == '\x08': # Backspace or DEL
if password:
password = password[:-1]
# Move cursor back, write space, move back again
writer.write(b'\x08 \x08')
await writer.drain()
elif char.isprintable():
password += char
# Show asterisk instead of actual character
writer.write(b'*')
await writer.drain()
except Exception as e:
print(f"Error reading password input: {e}")
break
return password
async def stop(self):
"""Stop the server"""
if self.server:
self.server.close()
await self.server.wait_closed()
# For testing without SSH dependencies
class DirectConnection(GameConnection):
"""Direct connection for testing without SSH"""
def __init__(self):
super().__init__("direct")
async def send_message(self, message: str, color: str = "white"):
"""Send message to console"""
color_map = {
'red': Fore.RED + Back.BLACK,
'green': Fore.GREEN + Back.BLACK,
'blue': Fore.BLUE + Back.BLACK,
'yellow': Fore.YELLOW + Back.BLACK,
'cyan': Fore.CYAN + Back.BLACK,
'magenta': Fore.MAGENTA + Back.BLACK,
'white': Fore.WHITE + Back.BLACK,
'gold': Fore.YELLOW + Style.BRIGHT + Back.BLACK,
'bright_green': Fore.GREEN + Style.BRIGHT + Back.BLACK,
'bright_red': Fore.RED + Style.BRIGHT + Back.BLACK,
'dark_yellow': Fore.YELLOW + Style.DIM + Back.BLACK,
'light_green': Fore.GREEN + Style.BRIGHT + Back.BLACK
}
colored_message = color_map.get(color, Fore.WHITE) + message + Style.RESET_ALL
print(colored_message)
async def get_input(self, prompt: str = "") -> str:
"""Get input from console"""
if prompt:
print(prompt, end='')
return input()
async def test_direct_connection():
"""Test the game with a direct console connection"""
try:
from server import GameServer
# Create game server
game_server = GameServer()
await game_server.start()
# Create direct connection
connection = DirectConnection()
print("=" * 60)
print(" SSH RPG - Text-Based MMORPG (Direct Mode)")
print("=" * 60)
print()
print("Please login or create a new account.")
print("Type 'login <username> <password>' or 'register <username> <password>'")
print("Type 'quit' to exit.")
# Main input loop
while True:
try:
command = await connection.get_input("> ")
if command.lower() in ['quit', 'exit']:
break
await game_server.handle_client_input(connection, command)
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}")
# Cleanup
if connection.user_id:
await game_server.disconnect_player(int(connection.user_id))
await game_server.stop()
except Exception as e:
print(f"Error in direct connection test: {e}")
traceback.print_exc()
if __name__ == "__main__":
# Run direct connection test
asyncio.run(test_direct_connection())