Updates a Slack User Group with People that are on call in PagerDuty (updated for pagerduty v2 api and pull from env vars instead of KMS). Based on:https://gist.github.com/devdazed/473ab227c323fb01838f

slack-pagerduty-oncall.py

"""
Lambda Func to update slack usergroup based on pagerduty rotation
From: https://gist.github.com/devdazed/473ab227c323fb01838f


NOTE: If you get a permission denied while setting the usergroup it is because there’s a workspace preference in slack
that limits who can manage user groups. At the time of writing it was restricted to owners and admins so i had to get
an owner to install the app. First i added them as a collaborator and then had them re-install the app, and got the new
auth token and added that to param store.

Slack permissions required:
- Installer must be able to update user groups.
- usergroups:read
- usergroups:write
- users:read
- users:read.email

"""
# !/usr/bin/env python

from __future__ import print_function

import json
import os
import logging

from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

log = logging.getLogger(__name__)


class SlackOnCall(object):
    # The Slack API token to use for authentication to the Slack WebAPI
    slack_token = None

    # The Pager Duty API token to use for authentication into the PagerDuty API
    pager_duty_token = None

    # The Slack @user-group to update (Default: oncall)
    slack_user_group_handle = 'oncall'

    # The maximum escalation level to add to the group
    # (eg. if escalation level = 2, then levels 1 and 2 will be a part of the group
    # but not any levels 3 and above.
    escalation_level = 2

    def __init__(self, slack_token, pager_duty_token,
                 slack_user_group_handle=slack_user_group_handle, log_level='INFO',
                 escalation_level=escalation_level, escalation_policy_ids=None):

        self.slack_token = slack_token
        self.pager_duty_token = pager_duty_token
        self.slack_user_group_handle = slack_user_group_handle
        self.escalation_level = int(escalation_level)
        if not escalation_policy_ids:
            self.escalation_policy_ids = []
        else:
            self.escalation_policy_ids = [s.strip() for s in escalation_policy_ids.split(',')]

        self._slack_user_group = None
        self._on_call_email_addresses = None
        self._all_slack_users = None

        log.setLevel(log_level)

    def run(self):
        """
        Gets user group information and on-call information then updates the
        on-call user group in slack to be the on-call users for escalation
        levels 1 and 2.
        """
        slack_users = self.slack_users_by_email(self.on_call_email_addresses)
        if not slack_users:
            log.warning('No Slack users found for email addresses: %s', ','.join(self.on_call_email_addresses))
            return

        slack_user_ids = [u['id'] for u in slack_users]

        if set(slack_user_ids) == set(self.slack_user_group['users']):
            log.info('User group %s already set to %s', self.slack_user_group_handle, slack_user_ids)
            return

        self.update_on_call(slack_users)
        log.info('Job Complete')

    @staticmethod
    def _make_request(url, body=None, headers={}):
        req = Request(url, body, headers)
        log.info('Making request to %s', url)

        try:
            response = urlopen(req)
            body = response.read()
            try:
                body = json.loads(body)
                if 'error' in body:
                    msg = 'Error making request: {}'.format(body)
                    log.error(msg)
                    raise ValueError(msg)
                return body
            except ValueError:
                return body

        except HTTPError as e:
            log.error("Request failed: %d %s", e.code, e.reason)
        except URLError as e:
            log.error("Server connection failed: %s", e.reason)

    @property
    def slack_user_group(self):
        """
        :return: the Slack user group matching the slack_user_group_handle
        specified in the configuration
        """
        if self._slack_user_group is not None:
            return self._slack_user_group

        url = 'https://slack.com/api/usergroups.list?token={}&include_users=1'.format(self.slack_token)
        groups = self._make_request(url)['usergroups']
        for group in groups:
            if group['handle'] == self.slack_user_group_handle:
                self._slack_user_group = group
                return group

        raise ValueError('No user groups found that match {}'.format(self.slack_user_group_handle))

    @property
    def on_call_email_addresses(self):
        """
        Hits the PagerDuty API and gets level 1 and level 2 escalation
        on-call users and returns their email addresses
        :return: All on-call email addresses within the escalation bounds
        """
        if self._on_call_email_addresses is not None:
            return self._on_call_email_addresses
        url ='https://api.pagerduty.com/oncalls?time_zone=UTC&include%5B%5D=users'
        on_call = self._make_request(url, headers={
            'Authorization': 'Token token=' + self.pager_duty_token,
            'Accept': 'application/vnd.pagerduty+json;version=2'
        })
        users = set()  # users can be in multiple schedule, this will de-dupe

        for escalation_policy in on_call['oncalls']:
            if escalation_policy['escalation_level'] <= self.escalation_level:
                users.add(escalation_policy['user']['email'])

        log.info('Found %d users on-call', len(users))
        self._on_call_email_addresses = users
        return users

    @property
    def all_slack_users(self):
        if self._all_slack_users is not None:
            return self._all_slack_users

        url = 'https://slack.com/api/users.list?token={}'.format(self.slack_token)
        users = self._make_request(url)['members']
        log.info('Found %d total Slack users', len(users))
        self._all_slack_users = users
        return users

    def slack_users_by_email(self, emails):
        """
        Finds all slack users by their email address
        :param emails: List of email address to find users
        :return: List of Slack user objects found in :emails:
        """
        users = []
        for user in self.all_slack_users:
            if user['profile'].get('email') in emails:
                users.append(user)

        return users

    def update_on_call(self, slack_users):
        """
        Updates the specified user-group
        :param slack_users: Slack users to modify the group with
        """
        user_ids = [u['id'] for u in slack_users]
        url = 'https://slack.com/api/usergroups.users.update?token={0}&usergroup={1}&users={2}'.format(
            self.slack_token,
            self.slack_user_group['id'],
            ','.join(user_ids)
        )

        log.info('Updating user group %s from %s to %s',
                 self.slack_user_group_handle, self.slack_user_group['users'], user_ids)
        self._make_request(url)


def lambda_handler(*_):
    """
    Main entry point for AWS Lambda.

    Variables can not be passed in to AWS Lambda, the configuration
    parameters below are encrypted using AWS IAM Keys.
    """

    # Boto is always available in AWS lambda, but may not be available in
    # standalone mode
    import boto3

    # To generate the encrypted values, go to AWS IAM Keys and Generate a key
    # Then grant decryption using the key to the IAM Role used for your lambda
    # function.
    #
    # Use the command `aws kms encrypt --key-id alias/<key-alias> --plaintext <value-to-encrypt>
    # Put the encrypted value in the configuration dictionary below
    config = {
        'slack_token': os.environ['SLACK_API_TOKEN'],
        'slack_user_group_handle': os.environ['SLACK_USER_GROUP_HANDLE'],
        'pager_duty_token': os.environ['PAGERDUTY_API_TOKEN'],
        'escalation_level': os.environ['ESCALATION_LEVEL'],
        'escalation_policy_ids': os.environ['ESCALATION_POLICY_IDS']
    }

    return SlackOnCall(**config).run()

Installation notes

I set this up as a lambda in AWS with serverless using the following serverless.yaml:

service: pagerduty-slack-usergroup-updater

custom:
  pythonRequirements:
    dockerizePip: true

provider:
  name: aws
  runtime: python3.6
  region: us-west-1
  stage: prod
  timeout: 30
  role: <arn-for-lambda-iam-role>
  environment:
    SLACK_API_TOKEN: ${ssm:/PAGERDUTY_SLACK_USERGROUP_UPDATER_SLACK_API_TOKEN~true}
    PAGERDUTY_API_TOKEN: ${ssm:/PAGERDUTY_SLACK_USERGROUP_UPDATER_PAGERDUTY_API_TOKEN~true}
    SLACK_USER_GROUP_HANDLE: <user-group-name>
    # comma separated string
    ESCALATION_POLICY_IDS: <escalation-policy-id1>,<escalation-policy-id2>
    ESCALATION_LEVEL: 1

functions:
  update:
    handler: slack-pagerduty-oncall.lambda_handler
    events:
      - schedule:
          # every 30 minutes
          rate: cron(30 * * * ? *)
package:
  exclude:
    - tests/**
    - .pytest_cache/**
    - .test_venv/**
    - __pycache__/**

plugins:
  - serverless-python-requirements

Notes:

• I keep my api tokens in AWS parameter store and have serverless pull them down and make them lambda environment vars with those ssm lines. just add params to parameter store for PAGERDUTY_SLACK_USERGROUP_UPDATER_SLACK_API_TOKEN and PAGERDUTY_SLACK_USERGROUP_UPDATER_PAGERDUTY_API_TOKEN, maybe choose a shorter name if you want. I can't recommend this if you really want to keep those API keys super secret. In that caseyou should probably use the kms method in the original version but ¯\(ツ)/¯.
• Replace the <...> bits with your info
• The interval can be adjusted by changing the cron(...) line as described in aws docs.
• I use the serverless serverless-python-requirements plugin which you can install with:
  sls plugin install -n serverless-python-requirements

Had a problem with large slack orgs (pagination) and auth (new oath token).
I've update this to work with slack_sdk and used a different method to find users.

https://gist.github.com/tr-fteixeira/25317bb6adaa1a5d223e33044029dc4a

❤️