Abstract

file

Enterprise identity and access management (IAM) systems play a critical role in controlling user access and streamlining administrative workflows. Authentik is a modern, flexible IAM platform providing OAuth, LDAP-like directory services, group management, and more. Meanwhile, Mailcow is a widely adopted self-hosted email suite that offers secure email services with optional OIDC integration. Despite their capabilities, Mailcow does not natively inherit group structures from external IAM systems like Authentik, limiting automated email distribution to dynamically managed groups. This article presents a practical solution to synchronize Authentik user groups with Mailcow email aliases, detailing the design, implementation, encountered challenges, and best practices.

In large organizations, managing user accounts and email distribution lists manually is error-prone and inefficient. Authentik provides a centralized platform to manage user identities, authenticate via OAuth2/OIDC, and define group memberships that reflect organizational structures. These groups can represent functional teams, administrative roles, or stakeholder categories such as employees, administrators, or shareholders.

Mailcow, on the other hand, provides a comprehensive email server solution, supporting aliases, forwarding, and optional integration with external authentication sources. However, it does not automatically synchronize email aliases with external group definitions, creating friction in maintaining mailing lists corresponding to dynamic IAM groups.

Problem Statement

Without synchronization, email distribution to logical groups defined in Authentik must be managed manually in Mailcow. This process is labor-intensive and prone to errors: when users are added or removed from groups in Authentik, Mailcow aliases remain outdated, leading to misdirected emails, inconsistent communication, and administrative overhead.

To solve this, we propose an automated synchronization mechanism that:

  1. Queries Authentik via its API to retrieve all groups and their members.
  2. Updates Mailcow email aliases to reflect these group memberships.
  3. Ensures that additions, removals, and updates in Authentik are mirrored in Mailcow, preserving accurate group-based email communication.

Methodology

Authentik Group Retrieval

Authentik provides a REST API capable of returning groups and associated users. Critical considerations include:

  • Pagination: Large deployments may return results over multiple pages.
  • Hierarchical groups: Groups may have parent-child relationships, requiring recursive resolution to include members of subgroups.
  • Active status: Only active users with valid email addresses should be considered for alias synchronization.

Implementation is achieved using Python requests, recursively traversing groups and building a complete mapping of group names to member email addresses.

A prerequisite for automating synchronization between Authentik and Mailcow is the availability of valid API credentials. Both platforms require tokens to authenticate REST API requests securely.

Authentik provides dedicated Service Accounts for programmatic access. To obtain an API token suitable for group and user queries:

  1. Create a Service Account

    • Log in to the Authentik administration interface.
    • Navigate to Directory → Users, and click Create Service Account.
    • Assign a unique username for the service account. This account will be used solely for API interactions.

  2. Generate an API Token

    • Select the newly created service account and go to Directory → Tokens & App passwords.
    • Create a Token of type API Token.
    • Assign minimal permissions required for synchronization, such as view users and view groups.
    • Optionally, configure token expiration (default is 30 minutes) or leave it unset for indefinite validity.

  3. Use the Token in Scripts

    • Copy the generated token immediately.
    • In API requests, include it in the HTTP header:
Authorization: Bearer <AK_TOKEN>

This token allows scripts to safely query Authentik for group memberships without requiring a user password.

emails = effective_member_emails(group_uuid, groups, children_map)

This ensures that any subgroup members are correctly aggregated into parent groups.

We can use the following python to log the groups of Authentik:

import requests
import collections
import os

AK_BASE = os.environ.get("AK_BASE")  # e.g. "https://auth.example.com"
AK_TOKEN = os.environ.get("AK_TOKEN")

if not AK_BASE or not AK_TOKEN:
    raise SystemExit("Please setup AK_BASE 与 AK_TOKEN")

HEADERS = {"Authorization": f"Bearer {AK_TOKEN}", "Accept": "application/json"}
PAGE_SIZE = 200

def fetch_all_groups():
    url = f"{AK_BASE}/api/v3/core/groups/?include_users=true&page_size={PAGE_SIZE}"
    groups = []
    while url:
        resp = requests.get(url, headers=HEADERS, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        groups.extend(data.get("results", []))
        url = data.get("next")
    return groups

def build_children_map(groups):
    children = collections.defaultdict(list)
    for g in groups:
        parent = g.get("parent")
        if parent:
            children[parent].append(g["pk"])
    return children

def effective_member_emails(root_uuid, groups, children_map):
    by_uuid = {g["pk"]: g for g in groups}
    seen = set()
    stack = [root_uuid]
    emails = set()
    while stack:
        u = stack.pop()
        if u in seen:
            continue
        seen.add(u)
        grp = by_uuid.get(u)
        if not grp:
            continue
        for uo in grp.get("users_obj") or []:
            if uo.get("is_active") and uo.get("email"):
                emails.add(uo["email"].lower())
        stack.extend(children_map.get(u, []))
    return sorted(emails)

if __name__ == "__main__":
    groups = fetch_all_groups()
    children_map = build_children_map(groups)

    for g in groups:
        emails = effective_member_emails(g["pk"], groups, children_map)
        print(f"Group: {g['name']} ({g['pk']})")
        for e in emails:
            print(f"  - {e}")
        print()

Mailcow Alias Management

Mailcow provides an administrator API key for authorized REST operations. To obtain a key for alias management:

  1. Log in to the Mailcow Admin UI

    • Access the Mailcow installation via a web browser and log in with administrator credentials.

  2. Access API Settings

    • Navigate to Configuration → Access → Edit administrator details.
    • Expand the API section if it is collapsed.

  3. Configure API Access

    • Tick Activate API to enable API access for this account.
    • Optionally restrict access to specific IP addresses using the API allow from field to enhance security.
    • Ensure the key has Read/Write permissions for the intended resources, such as alias management.

  4. Save and Copy the API Key

    • Click Save to apply the configuration.
    • Copy the generated API key; this token will authenticate all REST API calls to Mailcow:
X-API-Key: <MAILCOW_API_KEY>
Content-Type: application/json
Accept: application/json

Mailcow exposes a JSON-based REST API for creating, editing, and deleting aliases. Key implementation details discovered during development include:

  1. Creating Aliases: Straightforward using /api/v1/add/alias. Must provide the address and a comma-separated list of goto email addresses.

  2. Updating Aliases: The /api/v1/edit/alias endpoint requires careful attention:

    • items must contain the alias ID as a list.
    • attr is a dictionary with values as arrays for boolean-like fields (e.g., "active": ["0","1"]).
    • goto must include the complete list of members; otherwise, omitted members are unintentionally removed.

  3. Deleting Aliases: Must match Authentik groups to ensure stale aliases are removed.

An example edit payload for updating a Mailcow alias:

{
  "items": ["40"],
  "attr": {
    "address": "gitea-admins@aiursoft.com",
    "goto": "anduin@aiursoft.com,dvorak@aiursoft.com",
    "active": ["0","1"],
    "sogo_visible": ["0","1"],
    "private_comment": "",
    "public_comment": ""
  }
}

This precise structure ensures Mailcow correctly updates the alias, preserving all members.

We can use the following code to create a new alias in Mailcow:

#!/usr/bin/env python3

import os, sys, json
import requests

BASE = os.environ.get("MAILCOW_API_URL", "").rstrip("/")
KEY  = os.environ.get("MAILCOW_API_KEY", "")
VERIFY = os.environ.get("MAILCOW_VERIFY_SSL", "1") not in ("0", "false", "False")

if not BASE or not KEY:
    print("Please set: MAILCOW_API_URL / MAILCOW_API_KEY", file=sys.stderr)
    sys.exit(2)

session = requests.Session()
session.headers.update({
    "X-API-Key": KEY,
    "Content-Type": "application/json",
    "Accept": "application/json",
})
session.verify = VERIFY

def create_alias(alias_addr, members):
    payload = {
        "address": alias_addr,
        "goto": ",".join(members),
        "active": True,
        "sogo_visible": True
    }
    r = session.post(f"{BASE}/api/v1/add/alias", json=payload, timeout=(10, 30))
    return r.json()

if __name__ == "__main__":
    alias = "setu-maintainer@aiursoft.com"
    members = ["anduin@aiursoft.com", "edgeneko@aiursoft.com"]
    resp = create_alias(alias, members)
    print(json.dumps(resp, ensure_ascii=False, indent=2))

Final code

Finally wokring code to sync Authentik groups to mailcow email groups:

#!/usr/bin/env python3
import os, sys, json
import requests, collections

# ====================== Env ======================
AK_BASE = os.environ.get("AK_BASE")
AK_TOKEN = os.environ.get("AK_TOKEN")
MC_BASE = os.environ.get("MAILCOW_API_URL", "").rstrip("/")
MC_KEY  = os.environ.get("MAILCOW_API_KEY")
VERIFY = os.environ.get("MAILCOW_VERIFY_SSL", "1") not in ("0", "false", "False")

if not (AK_BASE and AK_TOKEN and MC_BASE and MC_KEY):
    sys.exit("请设置 AK_BASE, AK_TOKEN, MAILCOW_API_URL, MAILCOW_API_KEY")

AK_HEADERS = {"Authorization": f"Bearer {AK_TOKEN}", "Accept": "application/json"}
PAGE_SIZE = 200

# ====================== Authentik ======================
def fetch_all_groups():
    url = f"{AK_BASE}/api/v3/core/groups/?include_users=true&page_size={PAGE_SIZE}"
    groups = []
    while url:
        r = requests.get(url, headers=AK_HEADERS, timeout=10); r.raise_for_status()
        data = r.json(); groups.extend(data.get("results", []))
        url = data.get("next")
    return groups

def build_children_map(groups):
    children = collections.defaultdict(list)
    for g in groups:
        if g.get("parent"):
            children[g["parent"]].append(g["pk"])
    return children

def effective_member_emails(root_uuid, groups, children_map):
    by_uuid = {g["pk"]: g for g in groups}
    seen, stack, emails = set(), [root_uuid], set()
    while stack:
        u = stack.pop()
        if u in seen: continue
        seen.add(u)
        grp = by_uuid.get(u)
        if grp:
            for uo in grp.get("users_obj") or []:
                if uo.get("is_active") and uo.get("email"):
                    emails.add(uo["email"].lower())
            stack.extend(children_map.get(u, []))
    return sorted(emails)

# ====================== Mailcow ======================
mc = requests.Session()
mc.headers.update({
    "X-API-Key": MC_KEY,
    "Content-Type": "application/json",
    "Accept": "application/json"
})
mc.verify = VERIFY

def mc_get_aliases():
    r = mc.get(f"{MC_BASE}/api/v1/get/alias/all", timeout=(10,30)); r.raise_for_status()
    return r.json()

def mc_create_alias(addr, members):
    payload = {"address":addr, "goto":",".join(members), "active":True, "sogo_visible":True}
    return mc.post(f"{MC_BASE}/api/v1/add/alias", json=payload, timeout=(10,30)).json()

def mc_edit_alias(addr, members, alias_id):
    payload = {
        "items": [str(alias_id)],
        "attr": {
            "address": addr,
            "goto": ",".join(members),
            "active": ["0","1"],
            "sogo_visible": ["0","1"],
            "private_comment": "",
            "public_comment": ""
        }
    }
    r = mc.post(f"{MC_BASE}/api/v1/edit/alias", json=payload, timeout=(10,30))
    r.raise_for_status()
    return r.json()

def mc_delete_alias(ids):
    return mc.post(f"{MC_BASE}/api/v1/delete/alias", json=ids, timeout=(10,30)).json()

# ====================== Sync ======================
def main():
    # 1. Get Authentik Groups
    groups = fetch_all_groups()
    children_map = build_children_map(groups)
    group_map = {g["name"]: effective_member_emails(g["pk"], groups, children_map) for g in groups}

    # 2. Get Mailcow alias
    mc_aliases = mc_get_aliases()
    alias_map = {a["address"]:a for a in mc_aliases}

    desired = set()
    for grp_name, users in group_map.items():
        alias_addr = f"{grp_name.lower()}@aiursoft.com"  # Use your domain name!
        desired.add(alias_addr)
        if not users:
            continue
        if alias_addr in alias_map:
            a = alias_map[alias_addr]
            existing = [e.strip().lower() for e in a.get("goto","").split(",") if e]
            if sorted(existing) != users:
                print(f" Update alias '{alias_addr}'")
                resp = mc_edit_alias(alias_addr, users, a["id"])
                print(json.dumps(resp, ensure_ascii=False, indent=2))
                # Verify
                refreshed = mc_get_aliases()
                updated = next((x for x in refreshed if x["address"] == alias_addr), None)
                if updated:
                    new_goto = sorted(e.strip().lower() for e in updated.get("goto","").split(",") if e)
                    if new_goto == users:
                        print("  ✔ Update success")
                    else:
                        print("  ⚠ WARNING: Updated goto:", new_goto, "But expected:", users)
        else:
            print(f" 创建 alias '{alias_addr}'")
            resp = mc_create_alias(alias_addr, users)
            print(json.dumps(resp, ensure_ascii=False, indent=2))

    # 3. Delete alias not exists on Authentik
    to_delete = [a["id"] for addr,a in alias_map.items() if addr not in desired]
    if to_delete:
        print(" Deleting alias IDs：", to_delete)
        resp = mc_delete_alias(to_delete)
        print(json.dumps(resp, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()

Automation and Scheduling

To maintain continuous synchronization, we implemented a shell wrapper run.sh and scheduled it via cron:

*/10 * * * * /root/syncer/run.sh >> /var/log/mailcow_sync.log 2>&1

Key lessons:

  • Environment variables (AK_BASE, AK_TOKEN, MAILCOW_API_URL, MAILCOW_API_KEY) must be explicitly exported in the shell script to ensure cron can access them.
  • Logging outputs to a file helps track synchronization success or failures.
  • A 10-minute interval balances timely updates with API load considerations.

Conclusion

By automating the synchronization between Authentik and Mailcow, organizations can maintain up-to-date mailing lists that accurately reflect group memberships, reducing administrative overhead and mitigating errors. This integration demonstrates the practical synergy of modern IAM systems with self-hosted email solutions, enabling secure, dynamic, and efficient communication across enterprise teams.

The final implementation is robust, handles nested group structures, ensures Mailcow aliases are consistent, and can be scheduled for automatic periodic execution. Future extensions may include support for multiple domains, advanced filtering, and audit logging for compliance purposes.