🔐

motoでATSエンドポイントへのアクセスをモックできるか？

2022/12/20に公開

AWS

結論：たぶん無理だと思います。テストコードをうまく書くのが良さそうです。（もし何か手段があれば教えて下さい）

以下、やりたかったことと、その背景、妥協策を書きます。

やりたかったこと

やりたかったことは、ATSエンドポイントでAWS IoTやShadowにアクセスするPythonアプリを、motoでテストしたい、ということです。

背景

certifiの更新をせざるを得なくなった

Pythonアプリなら大抵、boto3というモジュールを使うと思いますが、boto3をインストールすると、内部でcertifiというモジュールもインストールされます。certifiは、Mozillaが厳選したルート証明書の一覧を返すモジュールです。ルート証明書というのは、クライアントがサーバーにTLS通信する際に、接続先のサーバーホストの身元を保証する認証局 (CA: Certificate Authority)に問合せますがその認証局自体の正しさを保証する証明書です。

2020年11月頃までは、AWS IoTやShadowへのアクセスに、単に以下のようにすればよいだけでした。

import boto3


iotdata_client = boto3.client("iot-data")

# Shadowを更新
iotdata_client.update_thing_shadow(
    thingName="hogehoge",
    payload=json.dumps(
        {"state": {"desired": {"key": "value"}}}
    ),
)

しかし、2020年11月頃以降は、SSLの検証エラーが出ます。下のページによると、certifiが提供するルート証明書から、デフォルトのエンドポイントの証明書を検証していたSymantec社のルートCA証明書が、バージョン2020.12.05のcertifiから削除されているためだそうです。

現在ではATSエンドポイントでAWS IoTやShadowにアクセスするのが推奨ですが、これまでcertifiを2020.11.08に固定して使い続けてました。しかし、certifiのこのバージョンに脆弱性が見つかったため、この度最新版のcertifiを使うように修正を加えました。

ATSとは

Amazon Trust Sevicesという認証局(CA: certificate authorities)です。

認証局とは、サーバーなどの証明書が問題ないものかをチェックしてくれる機関で、ATSはAmazon自身が運営している認証局です。

ATSのエンドポイントはATSで検証でき、boto3でもこれまで通りSSLの検証が可能です。

ATSエンドポイントにアクセスするコード

現状このようにするのが良いのかなと思っています。

import boto3


# ATSエンドポイントのURLを取得する
iot_client = boto3.client("iot")
_res = _iot.describe_endpoint(endpointType="iot:Data-ATS")
_ats_endpoint_url = None
if ep := _res.get("endpointAddress", ""):
    _ats_endpoint_url = "https://" + ep

# endpoint_urlを指定して、Shadowを更新
iotdata_client.update_thing_shadow(
    thingName="hogehoge",
    payload=json.dumps(
        {"state": {"desired": {"key": "value"}}}
    ),
    endpoint_url=_ats_endpoint_url,  # ここ
)

新たに、"iot:DescribeEndpoint"のポリシーが必要になることに注意しましょう。また、endpoint_urlのキーワード引数にNoneが与えられると、従来のエンドポイントが利用されます(伏線)。

motoでテストコードを書くがエラーが出る

motoは、AWSのサービスのモックを作るテスト用のライブラリです。AWSサービスとの連携部分の処理が正しいかを確認する際に重宝されます。

https://github.com/spulec/moto

しかし、motoで、iotとiot-dataのモックを作成して、テストコードを書いても、update_thing_shadowメソッドを実行する所で、指定されたエンドポイントにアクセスできないことを示す「Could not connect to the endpoint URL: ~~~」というエラーが出ます。

motoのコードを読むと、describe_endpointメソッドでは、ATSのエンドポイントのURLが、ランダム関数で生成されるようです（こちら）。

そして、update_thing_shadowでそのランダムに生成されたURLをエンドポイントに指定しているのですが、ここはモック化されているわけではなく、motoであっても実際に指定したエンドポイントにアクセスします。恐らく、指定したエンドポイントのURLが正しいかを検証できるようにそうしているのだと思います。

よって、単にiotとiot-dataをmotoでモック化しただけでは、iot-dataのクライアントを利用する際に、エラーが出るのだと思われます。

妥協策

describe_endpointメソッドでATSエンドポイントのURLを取得する処理だけ、staticメソッドに外出しして、それをモック化する対応をしました。

class IoTDataClientFactory:
    _iot_data_client = None

    @staticmethod
    def describe_endpoint_url() -> Optional[str]:
        _iot = boto3.client("iot")
        _res = _iot.describe_endpoint(endpointType="iot:Data-ATS")
	if ep := _res.get("endpointAddress"):
            return "https://" + ep
	return None

    @classmethod
    def create(cls):
        if cls._iot_data_client:
            return cls._iot_client

        cls._iot_data_client = boto3.client(
            "iot-data",
            endpoint_url=IoTDataClientFactory.describe_endpoint_url(),
        )
        return cls._iot_data_client

あとは、このIoTDataClientFactoryクラスのcreateメソッドを使って、iot-dataのクライアントを取得して利用する(update_thing_shadowメソッドを使うなど)形です。

テストコードでは、IoTDataClientFactory.describe_endpoint_urlをモック化して、Noneを返すようにします。

@pytest.fixture(autouse=True)
def set_endpoint(self, mocker):
    mocker.patch("app.IoTClient.describe_endpoint_url", return_value=None)

テストコードのクラスに上のfixtureを書いても良いですし、conftest.pyを作ってその中に書いても良いと思います。

こうすることで、endpoint_urlには、Noneが指定され、デフォルトのエンドポイントが使われることになります。その場合は、従来どおり、motoでモック化されたiot-dataのクライアントでも、update_thing_shadowメソッドを実行しても、エラーになりません。

motoで取得したATSエンドポイントURLをそのまま使った形式は諦めましたが、妥当な妥協策かなと思います。