forge/authentik-listmonk-bridge
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

main.py: add capacity to update groups

Browse source
This commit is contained in:
Antoine Martin 2024-03-03 00:21:03 -05:00
parent 325512121c
commit 916e037b26
Signed by: forge
GPG key ID: D62A472A4AA7D541
2 changed files with 71 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
bridge.yml
View file

@ -1,5 +1,4 @@
debug: "false"
group: "user"
debug: False
authentik:
api_url: "https://authentik.company/api/v3"

79
main.py
View file

@ -12,7 +12,6 @@ with open ('/etc/authentik-listmonk-bridge/config.yml', 'r') as 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']
@ -20,29 +19,35 @@ 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
url = authentik_api_url + '/core/users/' + '?type=internal&type=external'
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):
def authentik_get_groups():
url = authentik_api_url + '/core/groups/'
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):
url = listmonk_api_url + '/subscribers'
payload = {"email": email, "name": username, "status": status, "lists": lists, "preconfirm_subscriptions": True}
payload = {"email": email, "name": username, "status": status, "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):
def listmonk_set_subscriber(subscriber_id, email, username, status):
url = listmonk_api_url + '/subscribers/' + str(subscriber_id)
payload = {"email": email, "name": username, "status": status, "lists": lists}
payload = {"email": email, "name": username, "status": status}
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 = {
@ -55,8 +60,33 @@ def listmonk_get_subscriber(username):
json_object = json_object['data']['results']
return json_object[0]
def listmonk_set_list(ids, action, target_list_ids, status):
url = listmonk_api_url + '/subscribers/lists'
payload = {"ids": ids, "action": action, "target_list_ids": target_list_ids, "status": status}
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_create_list(group):
url = listmonk_api_url + '/lists'
payload = {"name": group, "type": "private", "optin": "single"}
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_get_list(list):
url = listmonk_api_url + '/lists?page=1&per_page=100 + "&query=' + list
headers = {'accept': 'application/json'}
resp = requests.get(url, headers=headers, auth=(listmonk_api_usr, listmonk_api_key))
json_object = json.loads(resp._content)
json_object = json_object['data']['results']
return json_object[0]
# ensure that users are synced
for user_info in authentik_get_users():
print('>>> Checking ' + user_info['username'])
print('>>> Checking user ' + user_info['username'])
if user_info['is_active'] == True:
user_active = "enabled"
@ -72,18 +102,45 @@ for user_info in authentik_get_users():
# 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])
create_resp = listmonk_create_subscriber(user_info['email'], user_info['username'], status=user_active)
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])
set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], subscriber_info['status'])
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])
set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], status=user_active)
print(set_resp)
# creates list of lists_id
lists_id = []
for list_info in subscriber_info['lists']:
lists_id.append(list_info['id'])
# matches groups to list_id
groups_id = []
for group_info in user_info['groups_obj']:
list_info = listmonk_get_list(group_info['name'])
groups_id.append(list_info['id'])
# checks if groups_id is in list of actual lists_id
# this will fail if user is in a group, but subscriber is not in the respective list
for id in groups_id:
if id not in lists_id:
print('> User lists check failed, updating')
set_resp = listmonk_set_list([subscriber_info['id']], "add", groups_id, status="confirmed")
print(set_resp)
break
# checks if lists_id is in list of group id
# this will fail is user is in a subcriber list, but not in respective group
for id in lists_id:
if id not in groups_id:
print('> User groups check failed, updating')
set_resp = listmonk_set_list([subscriber_info['id']], "remove", [id], status="")