Configuring Ejabberd (XMPP Server) to use Akkoma for authentication

If you want to give your Akkoma users an XMPP (chat) account, you can configure Ejabberd to use your Akkoma server for user authentication, automatically giving every local user an XMPP account.

In general, you just have to follow the configuration described at https://docs.ejabberd.im/admin/configuration/authentication/#external-script. Please read this section carefully.

Copy the script below to suitable path on your system and set owner and permissions. Also do not forget adjusting AKKOMA_HOST and AKKOMA_PORT, if necessary.

cp akkoma_ejabberd_auth.py /etc/ejabberd/akkoma_ejabberd_auth.py
chown ejabberd /etc/ejabberd/akkoma_ejabberd_auth.py
chmod 700 /etc/ejabberd/akkoma_ejabberd_auth.py

Set external auth params in ejabberd.yaml file:

auth_method: [external]
extauth_program: "python3 /etc/ejabberd/akkoma_ejabberd_auth.py"
extauth_instances: 3
auth_use_cache: false

Restart / reload your ejabberd service.

After restarting your Ejabberd server, your users should now be able to connect with their Akkoma credentials.

import sys
import struct
import http.client
from base64 import b64encode
import logging


AKKOMA_HOST = "127.0.0.1"
AKKOMA_PORT = "4000"
AUTH_ENDPOINT = "/api/v1/accounts/verify_credentials"
USER_ENDPOINT = "/api/v1/accounts"
LOGFILE = "/var/log/ejabberd/akkoma_auth.log"

logging.basicConfig(filename=LOGFILE, level=logging.INFO)


# Akkoma functions
def create_connection():
    return http.client.HTTPConnection(AKKOMA_HOST, AKKOMA_PORT)


def verify_credentials(user: str, password: str) -> bool:
    user_pass_b64 = b64encode("{}:{}".format(
        user, password).encode('utf-8')).decode("ascii")
    params = {}
    headers = {
        "Authorization": "Basic {}".format(user_pass_b64)
    }

    try:
        conn = create_connection()
        conn.request("GET", AUTH_ENDPOINT, params, headers)

        response = conn.getresponse()
        if response.status == 200:
            return True

        return False
    except Exception as e:
        logging.info("Can not connect: %s", str(e))
        return False


def does_user_exist(user: str) -> bool:
    conn = create_connection()
    conn.request("GET", "{}/{}".format(USER_ENDPOINT, user))

    response = conn.getresponse()
    if response.status == 200:
        return True

    return False


def auth(username: str, server: str, password: str) -> bool:
    return verify_credentials(username, password)


def isuser(username, server):
    return does_user_exist(username)


def read():
    (pkt_size,) = struct.unpack('>H', bytes(sys.stdin.read(2), encoding='utf8'))
    pkt = sys.stdin.read(pkt_size)
    cmd = pkt.split(':')[0]
    if cmd == 'auth':
        username, server, password = pkt.split(':', 3)[1:]
        write(auth(username, server, password))
    elif cmd == 'isuser':
        username, server = pkt.split(':', 2)[1:]
        write(isuser(username, server))
    elif cmd == 'setpass':
        # u, s, p = pkt.split(':', 3)[1:]
        write(False)
    elif cmd == 'tryregister':
        # u, s, p = pkt.split(':', 3)[1:]
        write(False)
    elif cmd == 'removeuser':
        # u, s = pkt.split(':', 2)[1:]
        write(False)
    elif cmd == 'removeuser3':
        # u, s, p = pkt.split(':', 3)[1:]
        write(False)
    else:
        write(False)


def write(result):
    if result:
        sys.stdout.write('\x00\x02\x00\x01')
    else:
        sys.stdout.write('\x00\x02\x00\x00')
    sys.stdout.flush()


if __name__ == "__main__":
    logging.info("Starting akkoma ejabberd auth daemon...")
    while True:
        try:
            read()
        except Exception as e:
            logging.info(
                "Error while processing data from ejabberd %s", str(e))
            pass