Example Python Bot


This is just an example how i run my bot. Basically you can call any command which works on the same system (call webhooks via curl, run ansible playbooks, conquer the world, …).

#include "std_disclaimer.h"
 * Your warranty is now void.
 * I am not responsible for hacked machines, crashed containers,
 * thermonuclear war, or you getting fired because the coffe machine bot hook failed.
 * Please regard this as a proof of concept.


I've prepared a virtualenv to run a python bot as the same user which runs the signal gateway. I also wanted to run a non-blocking bot and as i'm not yet familiar enough with async programming, i'm using python-rq to take care of the jobs.

Example workflow:

sequenceDiagram Signal Gateway->>Hook Script:Message activate Hook Script Note over Hook Script:Check keywords Hook Script->>Redis:Schedule function to redis queue Worker-->Redis:Fetch and run job Worker->>Signal Gateway:Send message
apt-get -y install python3-virtualenv redis-server
sudo -u signal-gateway -H python3 -m virtualenv -p /usr/bin/python3 /home/signal-gateway/venv

Install the necessary python packages:

sudo -u signal-gateway -H /home/signal-gateway/venv/bin/pip install requests rq pyyaml


This is the hook script inside /home/signal-gateway/bot.py. This script gets called by the gateway with the message supplied as parameter. The script itself does some sanity checks (if sender is known) and has the mapping of slash commands to functions (functions are defined in in handler and are processed by the worker).


import os
import yaml
import re
import sys
import bot_handler
from rq import Queue
from rq.job import Job
from bot_worker import conn
q = Queue(connection=conn)

script_path = os.path.dirname(os.path.abspath( __file__ ))
regex_uuid = '[0-9a-f]{8}-([0-9a-f]{4}-){3}[0-9a-f]{12}'

help_message = """/cat : get random cat
/wanip : get current WAN IP"""

def bot_help(*args, **kwargs):
    job = q.enqueue(bot_handler.send_message, help_message, kwargs['sender'])

def cat(*args, **kwargs):
    job = q.enqueue(bot_handler.cat, kwargs['sender'])

def wanip(*args, **kwargs):
    job = q.enqueue(bot_handler.wanip, kwargs['sender'])

# regex match to get the function to run for a message
regex_handlers = [
    (r'/help', bot_help),
    (r'/cat', cat),
    (r'/wanip', wanip),

def message_handler(message, sender):
    try to find a function (in regex_handlers) to run for the message
    for regex, function in regex_handlers:
        if re.search(regex, message, re.IGNORECASE):
            function(message=message, sender=sender)

def get_contacts():
    load known contacts from file
    with open(script_path + '/.config/contacts.yml', 'r') as ymlfile:
        return yaml.safe_load(ymlfile)

def is_known_contact(contact):
    check if the sender is a known contact
    contacts = get_contacts()
    name = list(filter(lambda person: person['name'] == contact, contacts['contacts']))
    tel = list(filter(lambda person: person['tel'] == contact, contacts['contacts']))
    if name or tel:
        if name:
            recipient = name[0]['tel']
        elif tel:
            recipient = contact
        return (True, recipient)
        return (False, None)

if __name__ == "__main__":
    line = sys.argv[1].split()
    is_group = re.search(r".+(\[)(.+)(\]).*", line[2])
    if is_group:
        sender = is_group.group(2)
        sender = line[2]
    message = ' '.join(line[3:])
    friend = is_known_contact(sender)
    if friend[1]:
        sender = friend[1]
    if is_group or friend[0]:
        message_handler(message, sender)


This file in /home/signal-gateway/bot_handler.py contains all the functions to act on slash commands.

import re
import uuid
import requests
import os
import redis
from rq import Queue
from rq.job import Job
from bot_worker import conn
q = Queue(connection=conn)

redis_host = 'localhost'
cwd = os.getcwd()

def requests_get(url, data=None, user=None, password=None, filename=None, stream=False):
    result = requests.get(url, auth=(user, password), data=data, stream=stream, verify=True)
    if filename is not None:
        if result.status_code == 200:
                with open(filename, 'wb') as f:
                    return True
        return result

def requests_post(url, data=None, user=None, password=None, files=None, stream=False):
    result = requests.post(url, auth=(user, password), data=data, files=files, stream=stream)
    if result.status_code == 200:
        return result
    return False

def cleanup_attachments(filename):
    return os.remove(filename)

def get_group_id(recipient):
    try to find the groupid for a recipient
    for filename in os.listdir(cwd + '/.storage/groups'):
        textfile = open(cwd + '/.storage/groups/' + filename, 'r')
        filetext = textfile.read()
        matches = re.search('name: (' + re.escape(recipient) + ')', filetext)
        if matches:
            return filename
            return recipient

def send_message(message, recipient, filename=None):
    send a message using the signal gateway
    multipart_form_data = {
        'message': ('', message),
        'to': ('', get_group_id(recipient))
    if filename:
        multipart_form_data.update({ 'file': (open(filename, 'rb')) })
    result = requests_post('http://localhost:5000/', files=multipart_form_data)
    if filename:
        job = q.enqueue(cleanup_attachments, filename)
    return result

def cat(sender):
    fetch a random cat from thecatapi.com and send back
    filename = str(uuid.uuid4())
    result = requests_get('http://thecatapi.com/api/images/get?format=src&type=gif', None, None, None, filename, True)
    if result:
        send_message("Here's your random " + "\U0001F63B", sender, filename)

def wanip(sender):
    get current WAN ip address and send back
    result = requests.get('http://whatismyip.akamai.com')
    if result:
        send_message("current WAN ip address is " + str(result.text), sender)


The worker /home/signal-gateway/bot_worker.py subscribes to redis and picks up new jobs to work on.


import os
import re
import redis
from rq import Worker, Queue, Connection

listen = ['default']

redis_url = os.getenv('REDIS', 'redis://localhost:6379')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue, listen)))

The worker needs another systemd unit file to run:

cat /etc/systemd/system/signal-bot-worker.service << EOF
Description=Signal Bot Worker



Enable and start the worker:

systemctl enable signal-bot-worker
systemctl start signal-bot-worker