This is a Authentik Expression Policy to assign groups based on slack channel membership.
Prerequisites:
id attribute containing a slack channel identifier.| from authentik.core.models import Group | |
| # Ensure flow is only run during OAuth logins via Slack | |
| if context["source"].provider_type != "slack": | |
| return True | |
| # Get the user-source connection object from the context, and get the access token | |
| connection = context.get("goauthentik.io/sources/connection") | |
| if not connection: | |
| return False | |
| access_token = connection.access_token | |
| channel_request = requests.get( | |
| "https://slack.com/api/conversations.list?exclude_archived=true&limit=500&types=public_channel%2Cprivate_channel", | |
| headers={ | |
| "Authorization": f"Bearer {access_token}" | |
| }, | |
| ) | |
| channel_info = channel_request.json() | |
| channels = list(filter(lambda e: e['is_member'], channel_info['channels'])) | |
| oauth_groups = list(map(lambda e: e['id'], channels)) | |
| # Get all groups | |
| slack_groups = Group.objects.filter(attributes__id__isnull=False) | |
| # Split user groups into slack groups and non slack groups | |
| user_groups_non_slack = request.user.ak_groups.exclude(pk__in=slack_groups.values_list("pk", flat=True)) | |
| user_groups_slack = list(request.user.ak_groups.filter(pk__in=slack_groups.values_list("pk", flat=True))) | |
| # Filter matching roles based on slack channels | |
| user_groups_slack_updated = slack_groups.filter(attributes__id__in=oauth_groups) | |
| # Combine user_groups_non_slack and matching_roles | |
| user_groups_updated = user_groups_non_slack.union(user_groups_slack_updated) | |
| # Update user's groups | |
| request.user.ak_groups.set(user_groups_updated) | |
| # Create event with roles changed | |
| ak_create_event( | |
| "slack_group_sync", | |
| user_slack_roles_before=", ".join(str(group) for group in user_groups_slack), | |
| user_slack_roles_after=", ".join(str(group) for group in user_groups_slack_updated), | |
| ) | |
| return True |