authentik-listmonk-bridge/main.py

#!/usr/bin/env python3

import requests
import json
import re
import yaml


from yaml.loader import SafeLoader
with open ('./bridge.yml', 'r') as file:
    config = yaml.safe_load(file)

# config parser
debug=config['debug']
group=config['group']
authentik_api_key=config['authentik']['api_key']
authentik_api_url=config['authentik']['api_url']
listmonk_api_key=config['listmonk']['api_key']
listmonk_api_url=config['listmonk']['api_url']
listmonk_api_usr=config['listmonk']['api_usr']

def authentik_get_users():
    url = authentik_api_url + '/core/users/' + '?type=internal&type=external' + '&groups_by_name=' + group
    headers = {'accept': 'application/json', 'Authorization': "Bearer " + authentik_api_key}
    resp = requests.get(url, headers=headers)
    json_object = json.loads(resp._content)
    return(json_object['results'])

def listmonk_create_subscriber(email, username, status, lists):
    url = listmonk_api_url + '/subscribers'
    payload = {"email": email, "name": username, "status": status, "lists": lists, "preconfirm_subscriptions": True}
    headers = {'Content-Type': 'application/json'}
    resp = requests.post(url, data=json.dumps(payload), headers=headers, auth=(listmonk_api_usr, listmonk_api_key))
    json_object = json.loads(resp._content)
    return json_object

def listmonk_set_subscriber(subscriber_id, email, username, status, lists):
    url = listmonk_api_url + '/subscribers/' + str(subscriber_id)
    payload = {"email": email, "name": username, "status": status, "lists": lists}
    headers = {'Content-Type': 'application/json'}
    resp = requests.put(url, data=json.dumps(payload), headers=headers, auth=(listmonk_api_usr, listmonk_api_key))
    json_object = json.loads(resp._content)
    return json_object


def listmonk_get_subscriber(username):
    url = listmonk_api_url + '/subscribers'
    params = {
        'page': '1',
        'per_page': '500',
        'query': "subscribers.name LIKE '" + username + "%'", 
    }
    resp = requests.get(url, params=params, auth=(listmonk_api_usr, listmonk_api_key))
    json_object = json.loads(resp._content)
    json_object = json_object['data']['results']
    return json_object[0]

for user_info in authentik_get_users():
    print('>>> Checking ' + user_info['username'])

    if user_info['is_active'] == True:
        user_active = "enabled"
    else:
        user_active = "disabled"

    try:
        subscriber_info = listmonk_get_subscriber(user_info['username'])
        print(subscriber_info)

    # if get user info fails, assume that we have to create a new one
    # TOOD: only create if the error is indeed user-existence related
    except Exception as e:
        print('> Failed to find username, creating subscriber')
        create_resp = listmonk_create_subscriber(user_info['email'], user_info['username'], status=user_active, lists=[1])
        print(create_resp)
        continue

    # checks if user email matches subscriber email
    if user_info['email'] != subscriber_info['email']:
        print('> Email check failed, updating')
        set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], subscriber_info['status'], lists=[1])
        print(set_resp)

    # check if user and subscriber info match as it relates to status
    if user_active != subscriber_info['status'] and subscriber_info['status'] != "blocklisted":
        print('> User status check failed, updating')
        set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], status=user_active, lists=[1])
        print(set_resp)