Simple Furcadia bot

furcadia-bot.py

#!/usr/bin/env python3

### Furcadia Login / Connection Example
#
# Last Updated: 2022-02-14
# Author: Artex <[email protected]>
#

import errno
import re
from socket import socket, AF_INET, SOCK_STREAM
from sys import stderr, exit
from typing import Union
from threading import Thread  # for console input thread


# (host, port)
G_FURC_ADDRESS = 'lightbringer.furcadia.com', 6500

# (username, password)
# NOTE: If your username has a space in it, write it as a pipe ('|') character instead of space!!
G_FURC_CREDENTIALS = 'CPU', 'password'  # (username, password)

# AFAIK, Furcadia won't send lines longer than this to you
# You will also not get less than 1 full line per packet/recv() call
G_BUFFER_SIZE = 4096  # bytes


def main():
    #---------------------------------------------------------------------------------------------#
    #--- CONNECTION STAGE ------------------------------------------------------------------------#
    #---------------------------------------------------------------------------------------------#

    print('[*] Connecting to %s:%d...' % G_FURC_ADDRESS, end=' ', flush=True)
    try:
        s = socket(AF_INET, SOCK_STREAM)
        s.connect(G_FURC_ADDRESS)
        print('OK')
    except OSError as ex:
        print('ERROR')
        stderr.write(f'Error connecting: {ex}\n')
        exit(1)
    
    # Set up a means for us to type commands into the console and send them to the server
    t = Thread(target=t_handle_console_input, args=(s,), daemon=True)
    t.start()

    #---------------------------------------------------------------------------------------------#
    #--- LOGIN STAGE -----------------------------------------------------------------------------#
    #---------------------------------------------------------------------------------------------#

    print('[*] LOGIN STAGE')
    try:
        stage_done = False
        while not stage_done:
            raw: bytes = s.recv(G_BUFFER_SIZE)

            # check for disconnection
            if not raw:
                print('[x] CONNECTION CLOSED (by server)')
                stderr.write(f'Login failed - server closed connection')
                exit(4)

            stage_done = handle_login_event(s, raw)
    except OSError as ex:
        stderr.write(f'Error logging in: {ex}\n')
        s.close()
        exit(2)
    except KeyboardInterrupt:
        s.close()
        print('[x] CONNECTION CLOSED (by us)')
        exit(3)

    #---------------------------------------------------------------------------------------------#
    #--- IN-GAME ---------------------------------------------------------------------------------#
    #---------------------------------------------------------------------------------------------#

    print('[*] IN-GAME STAGE')
    try:
        do_init_commands(s)
        while True:
            raw: bytes = s.recv(G_BUFFER_SIZE)

            # check for disconnection
            if not raw:
                print('[x] DISCONNECTED (by server)')
                exit(0)

            # handle message
            handle_game_event(s, raw)
    except OSError as ex:
        # s.close() was called - we expected that
        if ex.errno == errno.EBADF:  # "Bad file descriptor"
            print('[x] CONNECTION CLOSED (by us)')
            exit(0)
        # something else happened - we didn't expect that...
        else:
            raise
    except KeyboardInterrupt:
        s.close()
        print('[x] CONNECTION CLOSED (ctrl+c)')
        exit(3)


###############################################################################
# EVENT HANDLING FUNCTIONS
###############################################################################

def do_init_commands(s: socket):
    """
    Things you wish to do after fully logging in
    """
    # send_strln(s, 'wh Artex Rawr!')
    # s.close()  # if you wish to abruptly disconnect


def handle_login_event(s: socket, raw: bytes) -> bool:
    """
    This function handles LOGIN STAGE messages

    Note:
      Each message may consist of one or more lines!

    :param s: client socket
    :param raw: raw message (including \n characters)
    :return: True if login stage complete; false otherwise
    """

    # Banner Example:
    #   b'#568 576\n\n\n\n\n                              Good morning, Dave.\n\n\n\n\nDragonroar\n\n'
    #     ||   '-- max. seen usercount                                                 '-- we may login!
    #     |'------ current usercount
    #     '------- usercount prefix (separates usercount from regular newstext for backwards compatibility)

    print('[>][%4d] %r' % (len(raw), raw), flush=True)

    for line in raw.split(b'\n')[:-1]:
        # usercount line - usually the first line sent from the server
        # example: '#568 576\n'
        if line.startswith(b'#'):
            num_users_str, max_users_str = line[1:].decode('ascii').split(' ')  # ['568', '576']
            print(f'[#] usercount: {num_users_str}')
            print(f'[#] max.users: {max_users_str}')
        
        # we may log in after we notice this
        elif line == b'Dragonroar':
            username, password = G_FURC_CREDENTIALS
            print(f'[=] LOGGING IN AS {username}...')
            send_strln(s, f'connect {username} {password}')
            # new format:
            # send_strln(s, f'account {email} {username} {password}')

            # If you look at this string in Wireshark, you will notice additional arguments
            # used in these commands. This is known as "MachineID hash". You don't need to
            # provide it. It only exists to "unlock" the official client.
            # 
            # The official client will refuse to work without the following \PW response
            # to the provided machineid hash.
        
        elif line.startswith(b'&&&&&&&&&'):
            print(f'[@] LOGGED IN!')

            # You used to be able to (and had to) specify colors and description immediately
            # after logging in, but now apparently the system works with "costumes" which it
            # keeps on the web...
            #
            # send_strln(s, f'desc //Central Processing Unit//')
            # send_strln(s, f'color ]t#############')  # sending color would complete the signin stage
            send_strln(s, f'costume %-1')  # does this replace the `color line?
            send_strln(s, f'unafk')
            return True  # login stage complete
        
        elif line.startswith(b']]'):
            stderr.write('LOGIN FAILED\n')
            # from this point, we just wait for the server to terminate the connection

    return False


def handle_game_event(s: socket, raw: bytes):
    """
    This function handles IN-GAME STAGE events

    :param s: client socket
    :param raw: raw message (including \n characters)
    """
    # just print it
    for line in raw.split(b'\n')[:-1]:
        print_raw = True

        # take out the HTML tags out of printed lines and print them to the screen
        if line.startswith(b'('):
            text = clean_html(line.decode('ascii')[1:])
            text = text.replace('&lt;', '<').replace('&gt;', '>')  # convert < and > out of the HTML form
            print(f'[....] {text}')
            handle_game_text(s, text)
            print_raw = False

        # server wants us to download map (]r means it wants us to load from disk, but that's rarely used)
        elif line.startswith(b']q '):
            _ = line[3:].decode('ascii').split(' ')
            map_name  = _[0]
            map_crc32 = _[1]
            # there are other arguments there that I don't recognize
            print(f'[----] Download Map: {map_name} (crc: {map_crc32}')
            send_strln(s, 'vascodagama')  # confirms that we finished downloading the dream

        if print_raw:
            print_with_size(line)
    
    # s.close()  # if you wish to abruptly disconnect


def handle_game_text(s: socket, line: str):
    """
    This function handles IN-GAME STAGE text

    :param s: client socket
    :param line: line of text coming from Furcadia server (sanitized for HTML and <> characters)
    """
    m = re.match(r'^(\S+) asks you to join their company', line)
    if m:
        name = m[1]
        send_strln(s, 'join')
        return

    m = re.match(r'^(\S+) requests permission to lead you.', line)
    if m:
        name = m[1]
        send_strln(s, 'follow')
        return


def t_handle_console_input(s: socket):
    """
    This function runs in a separate thread and sends anything you type into
    the console directly to the server
    """
    print('[T] CONSOLE THREAD READY')
    while True:
        cmd = input()
        send_strln(s, cmd)


###############################################################################
# UTILITY FUNCTIONS
###############################################################################

def clean_html(s: str) -> str:
    """Clean HTML tags from within a string (the naive way)"""
    tags = []

    in_tag = False
    for i, ch in enumerate(s):
        if ch == '<':
            i_tag_start = i
            in_tag = True
        if ch == '>' and in_tag:
            tags.append((i_tag_start, i))
            in_tag = False
    
    if tags:
        i = 0
        segments = []
        for i_start, i_end in tags:
            segments.append(s[i:i_start])
            i = i_end + 1
        segments.append(s[i:])
        return ''.join(segments)
    else:
        return s  # no tags present


def send_str(sock: socket, s: str):
    """Send a string through a socket that only accepts bytes"""
    sock.send(s.encode('ascii'))


def send_strln(sock: socket, s: str):
    send_str(sock, s+'\n')


def recv_str(sock: socket, size: int = G_BUFFER_SIZE) -> str:
    """Receive a string from a socket that only provides bytes"""
    raise NotImplementedError("Furcadia uses non-printable characters! Use bytes instead...")


def print_with_size(s: Union[str,bytes], *args, **kwargs):
    if type(s) == bytes:
        s = repr(s)[2:-1]  # print them like Python would normally print bytes, without the "b'...'"

    print('[%4d] %s' % (len(s), s), *args, **kwargs)


###############################################################################
# INIT
###############################################################################

if __name__ == '__main__':
    main()