Automating an SSH session with Python

A reader of my blog recently sent me a question, and I’m going to answer them today. They wanted to get some guidance on using iTerm 2 to automate a MUD, which is basically a text-based MMORPG.

Hi Ryan, I found you through your post about iTerm2 python scripting and would like to get your input on a project I’ve been thinking through:

The project is a helper-bot for an ancient 80’s text-based adventure game that’s run over an ssh session. The game supports multiple concurrent sessions with each session controlling a character in the game.

The character commands are relatively basic: say, kill, heal, cure, get, drop, follow, etc..

I’m looking for a script that runs as a daemon that can react to commands from one player character and execute them bot-style on a second character-session.

For example: a player (IronMan) would type something in session 1 that would appear in the “bot session” as:

IronMan says “GameBot, Heal”

The bot reads this and types into its own session:

heal IronMan

I’m looking to support a few different kinds of interactions, with this “heal” example being the most basic, but getting somewhat more complex for other types of interactions:

• Execute a command once
• Execute a command until told to stop
• Execute a command until a condition is met
• Any of the above, but accepting arguments to alter the bot output

I have a background in software development, although it’s been some time since I’ve been in the thick of it (and never with Python). I’m looking for a working skeleton that I can use to further develop for this game on my own.

Let me know what you think, and I’m happy to answer any further questions.

Thank you, Billy

Thanks for the question, Billy! I think that automating programs is a widely applicable skill, both for games as well as more serious tasks. Let’s take a look at what it would take to accomplish this. We don’t have an actual MUD to test with, but we can use ssh-chat to create an SSH-based chat room that will allow us to simulate the core idea. I just downloaded this program, and connected to it from two separate terminals to do all my testing. Here’s what it will look like:

Using iTerm 2 Triggers

The absolute simplest thing you can do actually gets you pretty close without any coding at all. iTerm Triggers can automatically send text to a terminal when they receive text. Let’s create a trigger so that when we see the text “Player said: Robot, heal”, the session will automatically respond with “do action heal”. We’ll use regular expressions to make “heal” be any command. The trigger will look like this:

• Regular Expression: ^player: robot, (\w+)$
• Action: Send Text...
• Parameters: do action \1\r

The regular expression will match entire lines and the command can be any single word. The parameters include \1, which will be replaced with the single word command, and the \r is the escape sequence for the return key (if you omit this, the command will be typed but not sent).

Note: I recommend adding this trigger to a secondary profile, rather than your default one, so that you don’t accidentally run the trigger when you aren’t using the automation.

To test it, I’ve got ssh-chat opened twice, side by side, with the “robot” profile on the left and my normal one on the right. When I type “robot, heal” on the right, the left terminal detects it and automatically responds.

And we’re done! If you use your new bot profile and connect to the game, whenever it sees a string matching the regular expression, it will respond with the desired command.

Unfortunately, this approach can’t be expanded upon. We can’t repeat the command, or add delays, and the arguments are limited to what we capture with the regular expression. To go further, we’ll switch to Python.

Writing a bot

To take this to the next level, let’s write a bot. It will be a Python script that receives the game’s output, and sends commands back to the game. Here’s an initial version:

#!/usr/bin/env python3
import sys, re


def clean_line(line):
    """
    Clean up escape codes and extra spaces in the line.
    >>> clean_line('\x1b[38;05;173mplayer\x1b[0m: robot, heal ')
    'player: robot, heal'
    """
    return re.sub(r"\x1b\[[0-9;]*[a-zA-Z]|\x07", "", line).strip()


def send_command(line):
    "Send a command to the SSH session."
    # \r is the character code for a return. The default end is \n.
    print(line, end="\r")
    sys.stdout.flush()


def main():
    try:
        while True:
            line = clean_line(input())
            # For debugging, print every line exactly as it is received.
            print(repr(line), file=sys.stderr)
            m = re.match(r"^\[robot\] player: robot, (\w+)$", line)
            if m is not None:
                send_command(f"do action {m[1]}")
    except EOFError:
        pass


if __name__ == "__main__":
    main()

We can test it by running it ourselves:

When we type “player: robot, heal” (which is what the robot would see if we typed it into the game), the robot responds by writing “do action heal” on a new line. Note that because the robot ends commands with \r, the cursor will be on the same line as the robot’s output and you’ll type over it, but this is not an issue. How do we hook this up to SSH?

Standard UNIX tools

We actually don’t need to write an iTerm 2 plugin to finish this automation. In fact, we don’t need to use iTerm2 at all, although iTerm2 does provide a nice feature that we will use later on. For now, what we want to do is write a program that will read from the game and send commands back to it. We’ll need some way to tie the two programs together. There’s lots of ways to do this, but the easiest way is to use a named pipe, aka a FIFO (which stands for “first-in, first-out”). What’s a named pipe? We use anonymous pipes all the time, for example echo | cat creates an anonymous pipe that takes the output of echo and sends it to cat. Our bot could work like that as well: if we ran ssh | bot.py then the bot would receive all of the game’s output. But to send the game input, we need the pipe to go both ways, and anonymous pipes can’t do that. A named pipe is a special file that works just like a pipe: nothing can be written to it until another program is reading from it, and the written data will immediately be sent to the reading program.

Let’s create a named pipe called bot-output, which will receive the bot’s output, and then send that named pipe to SSH. It looks something like this, where we send the bot’s output into SSH and SSH’s output into the bot. We have to use >> to avoid deleting the FIFO and replacing it with a regular file.

$ mkfifo bot-output
$ cat bot-output | ssh ... | ./bot.py >> bot-output

Here’s how it works:

As you can see, the robot doesn’t see the same lines that we do, because of terminal escape codes. We have to filter them out, but sometimes lines come in jumbled and duplicated. Notably, the robot’s prompt ([robot]) is always at the start of every input line.

More advanced automation

Now that we have a custom Python program, we can take this however far we want. I’m going to show a simple example of repeating a line until told to stop. But there’s an important problem here: we need to do multiple things at once. Namely, we need to output lines on a schedule, but we also need to read lines to identify the stop command. There are two main ways we can approach this: threads, and asynchronous events. Threads are very useful, but come with many subtle caveats, and Python’s asynchronous framework is quite good, so we will use that.

#!/usr/bin/env python3
import sys, re, asyncio

# Omitting clean_line and send_command for brevity, they have not changed.

# Adapted from https://stackoverflow.com/a/64317899/123899
async def connect_stdin():
    loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    return reader


async def do_heal(person):
    while True:
        send_command(f"heal {person}")
        await asyncio.sleep(2)


async def main():
    stdin = await connect_stdin()
    task = None
    try:
        while True:
            # Our stream returns bytes, so we will decode it to a str now.
            line = clean_line((await stdin.readline()).decode("utf-8"))
            # For debugging, print every line exactly as it is received.
            print(repr(line), file=sys.stderr)
            if re.match(r"^\[robot\] player: robot, heal me$", line) is not None:
                task = asyncio.create_task(do_heal("player"))
            elif re.match(r"^\[robot\] player: robot, stop$", line) is not None:
                if task is not None:
                    task.cancel()
                    task = None
    except EOFError:
        pass


if __name__ == "__main__":
    asyncio.run(main())

By using asyncio.create_task, we can have the robot run a process while it’s waiting for more commands, like repeating a command. Here is a demo of our new bot in action:

OK Billy, hopefully this is a good skeleton for you to take and run with! This technique is useful for a wide variety of automations, and there are a lot of tools and libraries that provide different approaches. Here are a few:

• If you have a simple script that you need to send to a program, the expect program can handle it. It has a simple scripting language included, but it’s not as advanced as Python.
• If you’re automating SSH specifically, you can use Paramiko to handle the SSH connection directly in Python instead of using a FIFO as we did here. This allows a deeper integration (for example, reconnecting automatically), but means you have to configure the entire SSH connection from Python, which may be more difficult to use.
• You can also use tmux pipe-pane to connect an existing tmux pane to a shell script, which might be easier than manually creating a FIFO.

Finally, since I promised more iTerm2 goodness, let’s look at how we can use the bot we wrote without manually creating FIFOs using another iTerm 2 feature.

Using iTerm2 Coprocesses

iTerm2 supports Coprocesses, which basically does exactly what we did with pipes, except built in to iTerm2. It’s a little bit harder to debug a coprocess than a simple shell script, so I recommend using this once you already have your script working.

To actually use the coprocess, we’ll take the exact bot script we wrote above, and find its full path. Then we put that path into Session -> Run Coprocess. iTerm2 shows an icon in the affected terminal, but otherwise it’s invisible. Here’s what it looks like:

You can see the icon in the corner of the robot’s terminal that shows that iTerm2 is running the coprocess. The output is certainly easier to read when using a coprocess, since you see the screen normally, but there is no error output (unless the coprocess crashes), so I find it harder to debug. If you combined this with a log file and automatic re-running, it might be better than other methods.

Future automation idea: It’s also possible to set up a trigger to automatically start a coprocess. You could use this to complete a lengthy login step, by having the trigger detect when you are attempting to log in to a service.

Creating my personal cloud with HashiCorp

Stop using em in media queries

---

Hi, I’m Ryan. I’ve been a full-stack developer for over 19 years and work on every part of the modern tech stack. Whenever I encounter something interesting in my work, I write about it here. Thanks for reading this post, I hope you enjoyed it!