背景

Cloud Composer の機密情報を扱い際にCloud Key Management Serviceを利用しようとしたが、scheduler上でうまく動かなかった。Secret Managerを使って機密情報管理をできるか検証したのでその時のメモ。

結論

(追記) Cloud Composer no shedulerで呼び出すように定義する場合、PERMISSIONエラーになるので注意。使う場合はworkerから呼び出されるように定義する。

Cloud Composerでは import secretmanager を使った実装がうまくいかないので、 import secretmanager_v1beta1 を使った実装をする

クイックスタートで雰囲気掴む

クイックスタート

コンソールで機密情報を secret_id と version と一緒に登録する。
手元のpythonでクイックスタート動かしたら、登録した機密情報をpythonコードで取得できた。
pip install google-cloud-secret-manager が必要。
ここでは、 import secretmanager で動かした。


from google.cloud import secretmanager
client = secretmanager.SecretManagerServiceClient()
name = "projects/PROJECT/secrets/SECRET_ID/versions/VERSION"
response = client.access_secret_version(request={"name": name})
payload = response.payload.data.decode("UTF-8")
print("Plaintext: {}".format(payload))

クイックスタートを参考にCloud Composerで実装すると動かない問題

クイックスタートのコードを参考にCloudComposerで実装したところ以下のようなエラーとなった。
Cloud Composerのイメージを composer-1.13.0-airflow-1.10.12 にあげてもダメ。
google-cloud-secret-manager ライブラリのバージョンも最新の 2.0.0だった。
謎。

[2020-11-27 07:12:07,912] {base_task_runner.py:113} INFO - Job 255765: Subtask test     response = client.access_secret_version(request={"name": name})
[2020-11-27 07:12:07,912] {base_task_runner.py:113} INFO - Job 255765: Subtask test TypeError: access_secret_version() got an unexpected keyword argument 'request'

import secretmanager じゃなくて import secretmanager_v1beta1 で実装すると動く

Client for Secret Manager API

ググったら import secretmanager_v1beta1 を使っているレファレンスが存在したので、そっちで実装したら動いた。謎。

実装

test.py

import airflow
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta
from dependencies import secrets

def test():
    print(secrets.get('secret_id', "1"))

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": airflow.utils.dates.days_ago(1),
    "execution_timeout": timedelta(minutes=30),
}

dag = DAG("test", default_args=default_args, catchup=False, schedule_interval="0 0 * * *")
PythonOperator(task_id='test', python_callable=test, dag=dag,)

dependencies/secrets.py

from google.cloud import secretmanager_v1beta1
from dependencies import const

def get(secret_id, version):
    client = secretmanager_v1beta1.SecretManagerServiceClient()
    name = client.secret_version_path(const.PROJECT, secret_id, version)
    response = client.access_secret_version(name)
    return response.payload.data.decode("UTF-8")