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()

❤️