Backlogの課題にGitHubのコミットを連携する方法

このブログはBacklog Advent Calendar 2020 の7日目の記事です。 はじめてのAdvent Calendar参加でドキドキです。 adventar.org

Backlogの課題にGitHubのコミットやプルリクをコメントとして入れたい！

やりたいことは、「Backlogの課題にGitHubのコミットやプルリクをコメントとして入れたい」です。 BacklogのGitを使っていればコミットコメントに課題キーがあればその課題のコメントにコミットが連携されます。 なのでGitHubを使っていても同じようにしたい！

[GithubとBacklogの連携] Backlogでissue管理して、Githubへのコミット内容をBacklogにも反映させる様に連携する方法 - Qiitaを見て簡単にできる！とおもったら・・・GitHubのServiceはWebhookに統合されて消えていました。

We have deprecated GitHub Services in favor of integrating with webhooks. Replacing GitHub Services | GitHub Developer Guide

似たようなことをやっている人は世の中にいるのでいろいろ調べながら手作りすることにしました。

流れはこんなイメージ

GitHubとBacklogを連携するLambda関数を作る

1. BacklogでAPIキーを発行する

BacklogではAWSからやってくる処理を受け取るためのAPIキーを発行します。

絵だとこの辺のことです

1. [個人設定]の画面を開く

   • 

     [個人設定]は右上のメニューから

2. [API] > コメントを入力 > [登録]ボタンでAPIキーを発行する

   • 

     [登録]ボタンで一覧にAPIキーが追加される

2. AWSでLambdaとAPI Gatewayを作成する

AWSではGitHubからやってくる情報をAPI Gatewayで受け取ってLambda関数を呼び出して処理できるようにします。

絵だとこの辺のことです

IAM ポリシーを作成する

Lambda関数で使用する権限を作成します。

1. [AWS マネジメントコンソール]から[IAM]の画面を開く
2. サイドメニューの[ポリシー] > [ポリシーの作成]ボタンで作成画面を表示
3. [JSON]タブを開いて以下のJSONを設定 > [ポリシーの確認]ボタンで内容を確認
   • {アカウントID}に設定する値はAWS アカウント ID の確認方法 | AWSで確認する
4. [名前]に任意の値を設定 > [ポリシーの作成]ボタンで作成する

   • 

     Lambdaのログを作成する権限とSecrets Managerからシークレットを取得する権限を設定している

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:対象のリージョン:{アカウントID}:*"
        },
        {
            "Action": "secretsmanager:GetSecretValue",
            "Effect": "Allow",
            "Resource": "*"
        }
    ]
}

IAM ロールを作成する

Lambda関数にアタッチするロールを作成してさっき作ったポリシーを設定します。

1. [AWS マネジメントコンソール]から[IAM]の画面を開く
2. サイドメニューの[ロール] > [ロールの作成]ボタンで作成画面を表示
3. [AWSサービス] > [Lambda]を選択後に[次のステップ: アクセス権限]ボタンで次の画面を表示
4. [ポリシーのフィルタ]で作成したポリシーを検索して選択後に[次のステップ: タグ]ボタンで次の画面を表示
5. [タグの追加 (オプション)]は任意なので設定せずに[次のステップ: 確認]ボタンで次の画面を表示
6. [ロール名]を入力して[ロールの作成]ボタンでロールを作成する

Lambda関数をとりあえず作る

まずは、実装を後回しにして関数だけ作ります。

1. [AWS マネジメントコンソール]から[Lambda]の画面を開く
2. [関数の作成]ボタンで作成画面を表示する
3. [一から作成]を選択して以下を設定して[関数の作成]ボタンで関数を作成する
   • 関数名 : 任意の名前(今回はgithub_to_backlog)
   • ランタイム : Python3.7
   • 実行ロール : 既存のロールを使用する
   • 既存のロール : 作成したロールを選択

コードには初期コードがあるのでそのまま。実装は後でやります。

import json

def lambda_handler(event, context):
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

API Gatewayを作成する

• 参考
  • API Gateway での HTTP API の開発 - Amazon API Gateway
  • Amazon API Gatewayは「HTTP API」と「REST API」のどちらを選択すれば良いのか？ #reinvent | Developers.IO

今回は、[HTTP API]と[REST API]のどちらを使おうか迷ったけれど、「使ったことがない」「GitHubのWebhookを受け取りたいだけ」なんといっても「低コストらしいぞ」という理由で[HTTP API]にしました。

1. [AWS マネジメントコンソール]から[API Gateway]の画面を開く
2. [APIを作成]ボタンで[APIの作成]画面を表示する
3. [HTTP API]の[構築]ボタンで次の画面へ
4. [統合を追加] > [Lambda] > [Lambda 関数]で作成したLambda関数を選択
5. [API 名]に任意の名前を設定して[次へ]ボタンで[ルートを設定]画面へ
   • 
6. 以下を設定して[次へ]ボタンで[ステージを定義]画面へ
   • メソッド : POST
   • リソースパス : /{Lambda関数名}
   • 統合ターゲット : {Lambda関数名}
   • 
7. [ステージを追加]ボタンで以下を追加して[次へ]ボタンで[確認して作成]画面へ
   • ステージ名 : $default
   • 自動デプロイ : ON
   • 
8. [作成]ボタンで作成する
   • 
9. 「{Lambda関数名}のステージ」一覧の[URLを呼び出す]列に「呼び出しURL」が表示される
10. curlコマンドを使ってAPI Gatewayを呼び出してAPI GatewayがLambda関数を呼び出せることを確認する

curlコマンドのオプション	意味
-X	HTTPメソッドを指定する
-H	HTTPヘッダを指定する

# Lambda関数の初期コードに書いてある「Hello from Lambda!」が返却される
$ curl -X POST -H 'Content-Type:application/json' {呼び出しURL}/{リソースパス}
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
}
100    20  100    20    0     0     50      0 --:--:-- --:--:-- --:--:--    50"Hello from Lambda!"

# 失敗例) 「リソースパス」をくっつけ忘れると「Not Found」になるので注意
$ curl -X POST -H 'Content-Type:application/json' {呼び出しURL}
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100    23  100    23    0     0    287      0 --:--:-- --:--:-- --:--:--   291{"message":"Not Found"}

3. GithubでWebhook設定する

• 参考 : Github serviceをwebhookに変更した - アクトインディ開発者ブログ

GitHubでのプッシュやプルリクの情報がAPI Gatewayに送られるようにWebhookを設定します。

絵だとこの辺のことです

1. ブラウザでGitHubのリポジトリを表示する
2. [Settings] > [Webhooks] > [Add webhook]ボタン
3. 以下を設定して[Add webhook]ボタン
   • Payload URL : API Gatewayの「呼び出しURL/リソースパス」
   • Content type : application/json
   • Secret : 推測されにくい任意の文字列
     • 参考 : 暗号化されたシークレット - GitHub Docs
   • SSL verification : Enable SSL verification
   • Which events would you like... : 「Let me select individual events.」にして以下を選択
     • Pull requests
     • Pushes
   • 
4. 表示されたWebhookの横につくマークが緑チェックになるのを確認する
   • URLが間違っていたりすると赤バツが付くので、その場合は内容を確認する
   • 

4. AWSでSecrets Managerに情報を登録する

BacklogのAPIキーやGitHubのWebhookに設定したSecretは大切な情報なのでSecrets Managerに登録して、Lambda関数から取得して使うようにします。

絵だとこの辺のことです

1. [新しいシークレットを保存する]ボタンから作成画面を表示して以下を設定して[次]ボタン
   • シークレットの種類: その他のシークレット
   • シークレットのペア: 下記表参照

     • 

       Backlogの情報は1つのシークレットに2つのキーを設定する

   • 暗号化キー: DefaultEncryptionKey
2. 以下を設定して[次]ボタン
   • シークレットの名前: 下記表参照
   • 説明とタグ: 任意
3. [自動ローテーションを無効にする]を設定して[次]ボタン
4. [保存]ボタンでシークレットを作成する

シークレットの名前	シークレットキー	シークレットの値	使うところ
github/to/backlog	GITHUB_SECRET	GitHubのWebhookに 設定したSecret	GitHubから来た情報 をHMAC認証する時に使う
backlog/{GitHubのusername}	APIKEY	BacklogのAPIキー	Backlogに各ユーザで コメント追加するのに使う
同上	MAIL	Backlogに登録され ているメールアドレス (以下手順で確認) 1.Backlogの[個人設定]の画面を開く 2. [ユーザー情報] > [メールアドレス]	プルリクエストの通知を つけるためのid取得に使う

5. Lambda関数を実装する

環境変数を設定する

Backlogの課題にコメント追加するにはBacklog APIを使用します。 そのためにBacklog APIの情報としてLambdaの環境変数に以下を設定して使用します。

環境変数のキー	値	説明
BACKLOG_ENDPOINT	BacklogAPIのエンドポイント	https://BacklogのURLと同じ値/api/v2/ 参考:認証と認可 | Backlog Developer API | Nulab
PROJECT_KEY	Backlogのプロジェクトキー	参考 : プロジェクトの追加 – Backlog ヘルプセンター

Secrets Managerからシークレットの値を取得する

• 参考 : PythonでAWS Secrets ManagerからAPIキーを取得するときのちょっとしたポイント | Developers.IO

登録したBacklogのAPIキーやGitHubのWebhookに設定したSecretを取得できるようにします。 基本的な実装はシークレットを登録した際にSecrets Managerの画面に表示されるサンプルコードを使用しました。

def get_secrets_manager_dict(secret_name: str) -> dict:
    """Secrets Managerからシークレットのセットを辞書型で取得する"""
    secrets_dict = {}
    if not secret_name:
        print('シークレットの名前未設定')
    else:
        session = boto3.session.Session()
        client = session.client(
            service_name='secretsmanager',
            region_name='対象のリージョン'
        )
        try:
            get_secret_value_response = client.get_secret_value(
                SecretId=secret_name
            )
        except ClientError as e:
            print('シークレット取得失敗：シークレットの名前={}'.format(secret_name))
            print(e.response['Error'])
        else:
            if 'SecretString' in get_secret_value_response:
                secret = get_secret_value_response['SecretString']
            else:
                secret = base64.b64decode(get_secret_value_response['SecretBinary'])
            secrets_dict = ast.literal_eval(secret)
    return secrets_dict

GitHubのWebhookから送られてくる内容を取得する処理を作る

GitHubのWebhookから送られてくる内容は以下サイトに説明があります。

• Webhook event payloads | GitHub Developer Guide

まずは、GitHubから送られてきた情報が本当に設定したWebhookからなのを確認するためにHMAC認証します。 GitHubのWebhookに設定したSecretとGitHubから送られてきた情報で認証を行います。

参考 : GitHubのWebhookでプルリクエストをマージした際にツイートできるようしてみた - Qiita

def is_correct_signature(signature: str, body: dict) -> bool:
    """GitHubから送られてきた情報をHMAC認証する."""
    if signature and body:
        # GitHubのWebhookに設定したSecretをSecrets Managerから取得する
        secret = get_secrets_manager_key_value('github/to/backlog', 'GITHUB_SECRET')
        if secret:
            secret_bytes = bytes(secret, 'utf-8')
            body_bytes = bytes(body, 'utf-8')
            # Secretから16進数ダイジェストを作成する
            signedBody = "sha1=" + hmac.new(secret_bytes, body_bytes, hashlib.sha1).hexdigest()
            return signature == signedBody
    else:
        return False

プッシュとプルリクではGitHubからくる情報が異なるというのとBacklogの課題に追加するコメントをちょっと変えるために処理を切り分けます。

操作	action	pull_request	commits
Pull requests	o	o	x
Pushes	x	x	o

def lambda_handler(event, context):
    # GitHubから送られてきた情報をHMAC認証する
    if is_correct_signature(event['headers']['x-hub-signature'], event['body']):
        body = json.loads(event['body'])
        # プッシュとプルリクを識別して処理を切り分ける
        if 'pull_request' in body and 'action' in body:
            # Pull requestsの場合
            add_pull_request_comment(body['action'], body['pull_request'], body['sender']['login'])
        elif 'commits' in body:
            # Pushesの場合
            add_push_comment(body['commits'])
        else:
            print('処理対象外のリクエストなので処理しない' + json.dumps(body))
    else:
        print('認証できないGitHubのsignatureが送られてきた')

コメントから課題キーを検索する

コメント追加する課題を決めるためにコミットコメントやプルリクのコメントから課題キーを検索します。 複数の課題キーがあればそれぞれの課題にコメントが追加できるようにリストに課題キーを入れて返却します。

def get_issue_key(message: str) -> list:
    """コメントから課題キーを検索する."""
    issue_key = []
    # 「[プロジェクトキー] + [-] + [数字の繰り返し]を全て抜出
    key_format = '{}-[\d]+'.format(os.environ.get('PROJECT_KEY'))
    match_list = re.findall(key_format, message)
    if match_list:
        # 抜き出した課題キーのリストから重複を削除する
        issue_key = list(set(match_list))
    else:
        print('コメントにBacklogの課題キーが設定されていない')
    return issue_key

プルリクエストはレビューアーに通知をつけたいので通知リストを作る

通知リストは次の流れで作成します。

1. プルリクエストに設定されたレビューアーのGitHubユーザー名でSecrets ManagerからBacklogのメールアドレスを取得する
2. APIで取得したBacklogのユーザー一覧からメールアドレスでBacklogのユーザーIDを探す
   • ユーザー一覧の取得 | Backlog Developer API | Nulab
   • 通知に使うのはAPI取れたユーザー情報のidを使います。
3. 通知リストへ追加する

def create_notified_list(requested_reviewers: list, backlog_users: list) -> list:
    """レビューアーにBacklogの通知をつけるため、にコメント登録の通知を受け取るユーザーIDリストを作成する"""
    notified_list = []
    if requested_reviewers and backlog_users:

        for reviewer in requested_reviewers:
            # Secrets ManagerからBacklogのメールアドレスを取得する
            backlog_mail = get_secrets_manager_key_value('backlog/' + reviewer['login'], 'MAIL')

            if backlog_mail:
                for user in backlog_users:
                    # 取得したメールアドレスと同じメールアドレスのユーザーをBacklogユーザー一覧から探す
                    if backlog_mail == user['mailAddress']:
                        # ユーザーのIDをリストへ追加する
                        notified_list.append(user['id'])

    return notified_list

Backlogにコメントを追加する

• APIの仕様 : 課題コメントの追加 | Backlog Developer API | Nulab

def add_backlog_comment(api_key: str, issue_keys: list, comment: str, notified_list: list):
    """Backlogの課題にコメントを追加する."""
    if not api_key or not issue_keys:
        print('BacklogのAPIキーまたは課題キー未設定')
    else:
        params = {'apiKey': api_key}
        payload = {'content': comment}

        # コメント登録の通知を受け取るユーザーIDがある場合は設定する
        if notified_list:
            payload['notifiedUserId[]'] = notified_list

        # コメントにある課題すべてにコメントを追加する
        for issue_key in issue_keys:
            header = {'Content-Type': 'application/x-www-form-urlencoded'}
            api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['issues', issue_key, 'comments']))
            result = requests.post(api_path, headers=header, params=params, data=payload)

できた!!

今回はシンプルにコメント追加をしました。 今後、BacklogのGitみたいに課題のステータス変更をしたり、レビューコメントも追加したりしたら楽しそうです！

コードの全体

ここ以降は、コード全部を張っているだけなので興味のある人だけ見てください。

import json, os
import hmac, hashlib
import requests, urllib
import boto3
import base64
import ast, re
from botocore.exceptions import ClientError


def get_secrets_manager_dict(secret_name: str) -> dict:
    """Secrets Managerからシークレットのセットを辞書型で取得する"""
    secrets_dict = {}
    if not secret_name:
        print('シークレットの名前未設定')
    else:
        session = boto3.session.Session()
        client = session.client(
            service_name='secretsmanager',
            region_name='対象のリージョン'
        )
        try:
            get_secret_value_response = client.get_secret_value(
                SecretId=secret_name
            )
        except ClientError as e:
            print('シークレット取得失敗：シークレットの名前={}'.format(secret_name))
            print(e.response['Error'])
        else:
            if 'SecretString' in get_secret_value_response:
                secret = get_secret_value_response['SecretString']
            else:
                secret = base64.b64decode(get_secret_value_response['SecretBinary'])
            secrets_dict = ast.literal_eval(secret)
    return secrets_dict


def get_secrets_manager_key_value(secret_name: str, secret_key: str) -> str:
    """AWS Secrets Managerからシークレットキーの値を取得する."""
    value = ''
    secrets_dict = get_secrets_manager_dict(secret_name)
    if secrets_dict:
        if secret_key in secrets_dict:
            # secrets_dictが設定されていてsecret_keyがキーとして存在する場合
            value = secrets_dict[secret_key]
        else:
            print('シークレットキーの値取得失敗：シークレットの名前={}、シークレットキー={}'.format(secret_name, secret_key))
    return value


def is_correct_signature(signature: str, body: dict) -> bool:
    """GitHubから送られてきた情報をHMAC認証する."""
    if signature and body:
        # GitHubのWebhookに設定したSecretをSecrets Managerから取得する
        secret = get_secrets_manager_key_value('github/to/backlog', 'GITHUB_SECRET')
        if secret:
            secret_bytes = bytes(secret, 'utf-8')
            body_bytes = bytes(body, 'utf-8')
            # Secretから16進数ダイジェストを作成する
            signedBody = "sha1=" + hmac.new(secret_bytes, body_bytes, hashlib.sha1).hexdigest()
            return signature == signedBody
    else:
        return False


def get_backlog_api_key(github_username: str) -> str:
    """GitHubのユーザ名から該当ユーザのBacklogのAPIキーを取得する."""
    api_key = ''
    if github_username:
        api_key = get_secrets_manager_key_value('backlog/' + github_username, 'APIKEY')
    else:
        print('GitHubユーザー名未設定')
    return api_key


def add_backlog_comment(api_key: str, issue_keys: list, comment: str, notified_list: list):
    """Backlogの課題にコメントを追加する."""
    if not api_key or not issue_keys:
        print('BacklogのAPIキーまたは課題キー未設定')
    else:
        params = {'apiKey': api_key}
        payload = {'content': comment}

        # コメント登録の通知を受け取るユーザーIDがある場合は設定する
        if notified_list:
            payload['notifiedUserId[]'] = notified_list

        # コメントにある課題すべてにコメントを追加する
        for issue_key in issue_keys:
            header = {'Content-Type': 'application/x-www-form-urlencoded'}
            api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['issues', issue_key, 'comments']))
            result = requests.post(api_path, headers=header, params=params, data=payload)


def get_backlog_users(api_key: str) -> list:
    """Backlogユーザー一覧の取得"""
    users = []

    if not api_key:
        print('BacklogのAPIキー未設定')
    else:
        # ユーザーの取得対象はプロジェクト内に設定する
        api_path = urllib.parse.urljoin(os.environ.get('BACKLOG_ENDPOINT'), '/'.join(['projects', os.environ.get('PROJECT_KEY'), 'users']))
        api_result = requests.get(api_path, params={'apiKey': api_key}).json()

        if type(api_result) == list and api_result:
            # Backlogのユーザー一覧を取得する
            users = api_result
        else:
            print('Backlogプロジェクトのユーザー一覧取得失敗：{}'.format(json.dumps(api_result)))

    return users


def create_notified_list(requested_reviewers: list, backlog_users: list) -> list:
    """レビューアーにBacklogの通知をつけるため、にコメント登録の通知を受け取るユーザーIDリストを作成する"""
    notified_list = []
    if requested_reviewers and backlog_users:

        for reviewer in requested_reviewers:
            # Secrets ManagerからBacklogのメールアドレスを取得する
            backlog_mail = get_secrets_manager_key_value('backlog/' + reviewer['login'], 'MAIL')

            if backlog_mail:
                for user in backlog_users:
                    # 取得したメールアドレスと同じメールアドレスのユーザーをBacklogユーザー一覧から探す
                    if backlog_mail == user['mailAddress']:
                        # ユーザーのIDをリストへ追加する
                        notified_list.append(user['id'])

    return notified_list


def get_issue_key(message: str) -> list:
    """コメントから課題キーを検索する."""
    issue_key = []
    # 「[プロジェクトキー] + [-] + [数字の繰り返し]を全て抜出
    key_format = '{}-[\d]+'.format(os.environ.get('PROJECT_KEY'))
    match_list = re.findall(key_format, message)
    if match_list:
        # 抜き出した課題キーのリストから重複を削除する
        issue_key = list(set(match_list))
    else:
        print('コメントにBacklogの課題キーが設定されていない')
    return issue_key


def add_push_comment(commits: list):
    print('プッシュの情報をBacklogの課題へコメント追加する')
    print(json.dumps(commits))
    # プッシュに含まれるコミットを1つずつ処理する
    for commit in commits:
        issue_key = []
        if 'message' in commit:
            issue_keys = get_issue_key(commit['message'])
        if issue_keys:
            backlog_api_key = get_backlog_api_key(commit['committer']['username'])
            if backlog_api_key:
                comment_format = '- URL：{}\n{}'
                comment = comment_format.format(commit['url'], commit['message'])
                add_backlog_comment(backlog_api_key, issue_keys, comment, [])


def add_pull_request_comment(action: str, pull_request: dict, username: str):
    print('プルリクエストの情報をBacklogの課題へコメント追加する\nアクション：{}\nマージ：{}'.format(action, str(pull_request['merged'])))
    print(json.dumps(pull_request))
    # プルリクのタイトルとコメントから課題キーを取得する
    issue_keys = get_issue_key(pull_request['title'] + pull_request['body'])
    if issue_keys:
        # BacklogのAPIキーを取得する
        backlog_api_key = get_backlog_api_key(username)

        if backlog_api_key:
            notified_list = []
            if 'requested_reviewers' in pull_request and pull_request['requested_reviewers']:
                # Backlogのユーザー一覧を取得する
                backlog_users = get_backlog_users(backlog_api_key)
                if backlog_users:
                    # レビューアーが設定されている場合は追加するコメント用の通知リストを作成する
                    notified_list = create_notified_list(pull_request['requested_reviewers'], backlog_users)

            # プルリクのアクションによってコメントを作成する
            comment = ''
            if action == 'opened' or action == 'reopened':
                comment = 'プルリクエストが作成されました'
            elif action == 'closed':
                if pull_request['merged']:
                    comment = 'プルリクエストがマージされました'
                else:
                    comment = 'プルリクエストが却下されました'
            else:
                comment = 'プルリクエストが変更されました'
            comment += '\n\n- URL：{}'.format(pull_request['html_url'])

            add_backlog_comment(backlog_api_key, issue_keys, comment, notified_list)


def lambda_handler(event, context):
    # GitHubから送られてきた情報をHMAC認証する
    if is_correct_signature(event['headers']['x-hub-signature'], event['body']):
        body = json.loads(event['body'])
        # プッシュとプルリクを識別して処理を切り分ける
        if 'pull_request' in body and 'action' in body:
            # Pull requestsの場合
            add_pull_request_comment(body['action'], body['pull_request'], body['sender']['login'])
        elif 'commits' in body:
            # Pushesの場合
            add_push_comment(body['commits'])
        else:
            print('処理対象外のリクエストなので処理しない' + json.dumps(body))
    else:
        print('認証できないGitHubのsignatureが送られてきた')