概要

タイトルで内容を示そうとした割に意味が分かりにくくて申し訳ありません。
実施したい内容は下記です。

SalesforceへAPIアクセスしたい
認証方法は OAuth2.0 クライアントログイン情報フローを使用
Pythonで実装
- simple-salesforce というモジュールを利用
クレデンシャルはAWS Secrets Managerのシークレットとして保存
取得したアクセストークンもシークレットとして保持、再利用する

ポイント

取得したアクセストークンを保持し、再利用する仕組みとします。

毎回認証した場合

毎回認証する場合下記のようになります。

ユーザーAからのリクエスト→認証を実行、トークン取得、API実行
ユーザーBからのリクエスト→認証を実行、トークン取得、API実行
ユーザーCからのリクエスト→認証を実行、トークン取得、API実行
...
ユーザーXからのリクエスト→認証を実行、トークン取得、API実行
ユーザーYからのリクエスト→認証を実行、トークン取得、API実行

これでもシンプルですし動作しますが、2つウィークポイントが考えられます。

毎回認証リクエストする処理コスト
ログイン回数制限によって処理実行不能になるリスク
- 1時間あたり3,600回を超えるログインリクエストはブロックされます
- サインインエラー: 「ログイン数超過」

要件次第ですが、多量のリクエスト数が見込まれる場合、2については致命的になる可能性があります。
そこでトークンを保持・再利用するように実装し、下記のような振る舞いとなるようにします。

トークンを再利用するようにした場合

ユーザーAからのリクエスト→保持トークン無し、認証を実行、トークン取得、トークン保持、API実行
ユーザーBからのリクエスト→保持トークン有り、API実行
ユーザーCからのリクエスト→保持トークン有り、API実行
...
ユーザーXからのリクエスト→保持トークン有り、API実行、トークン期限切れていた、認証を実行、トークン取得、トークン保持、API実行
ユーザーYからのリクエスト→保持トークン有り、API実行

ユーザーB,C については保持トークンが有効であるため認証不要でAPI実行可能です。
ただし、トークンの有効期限切れの場合は認証を再実行し、トークンを更新する処理を実行します。
ユーザーXの処理については初回APIリクエストが失敗するため、処理コストは大きくなりますが、システム全体としては適正化されていると考えます。

実装

最終的に simple-salesforceモジュールの Salesforce クラスを継承する形で実装してみました。

また、PyPIへも登録したので pip で利用可能です。

利用サンプル

準備

AWS Secrets Managerへクレデンシャルを登録します。

Domain キー
- アクセス対象セールスフォースのドメインのうち、末尾の .salesforce.com を除いた部分を指定します
- 私のドメイン(MyDomain)で確認すると確実だと思います
- 末尾を除外するのは simple-salesforceの仕様に拠ります
ConsumerKey ConsumerSecret キー
- セールスフォースの接続アプリケーション設定より取得します

AWSマネジメントコンソールで設定すると、下記のようになります（ドメインはDeveloperEdition組織の例）

コード実装

simple-salesforce-extends モジュールをインストールします。
simple-salesforce に依存しています。

$ pip install simple-salesforce-extends

boto3 モジュールをインストールします。

$ pip install boto3

コードを作成します。

上記手順で作成したシークレットARNを使用します。
Secrets Managerのboto3クライアントオブジェクトとシークレットARNを受け取り、それを使用してシークレットを取得するようになっています。

sample.py

import boto3
from simple_salesforce_extends import SalesforceClientCredential

sfdc_credential_secrets_arn = "arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:my-secret-arn"
secrets_client = boto3.client("secretsmanager")

sf = SalesforceClientCredential(
    secrets_client=secrets_client,
    credentials_secrets_manager_arn=sfdc_credential_secrets_arn,
)

soql = "SELECT Id, Name FROM Account LIMIT 1"
result = sf.query(soql)
print(result["records"][0])

サンプルを実行

正しく実行されました。

$ python sample.py
OrderedDict({'attributes': OrderedDict({'type': 'Account', 'url': '/services/data/v57.0/sobjects/Account/0015i000013ItW7AAK'}), 'Id': '0015i000013ItW7AAK', 'Name': 'Accountname'})

シークレットを確認

SessionId Instance キーが追加されます。
認証処理で取得できたトークンが SessionId キーの値、セールスフォースインスタンスが Instance キーの値として保持されます。

サンプルを再実行

何度かサンプルを実行してログイン履歴を確認し、ログインリクエストしていないことを確認します。
（処理側でログを出力して確認しても良いです）

最後に

サービス間インテグレーション処理を実装機会が増えていると思います。
ご参考になれば幸いです。

Salesforceアクセス時のクレデンシャルとトークンをAWS SecretsManagerに保持する

概要