Django + google-authで、Google Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイントを作成する

Djangoアプリでのジョブをスケジューリングしたいと思い、Google Cloud Schedulerのドキュメントを読んだところ

Cloud Scheduler では、作業単位のスケジュールを設定して、定義した回数または一定の間隔で実行できます。これらの作業単位は、一般的に cron ジョブと呼ばれています。代表的な使い方としては、レポートメールを毎日送信する、10 分間隔でキャッシュ データを更新する、1 時間に 1 回要約情報を更新する、などがあります。

Cloud Scheduler を使用して作成された各 cron ジョブは、指定のスケジュールに従ってターゲットに送信されます。ターゲットはタスクが処理される場所です。ターゲットは、次のいずれかのタイプでなければなりません。

• 一般公開されている HTTP/S エンドポイント

Cloud Scheduler について  |  Cloud Scheduler のドキュメント  |  Google Cloud

との記載がありました。

これにより、Djangoアプリ側でGoogle Cloud Scheduler から呼び出されたことが分かれば、ジョブのスケジューリングができそうでした。

　
次にどのような特徴を持つリクエストが飛んでくるかを調べてみたところ

HTTP リクエストの Authorization ヘッダに Service Account の情報を含んだ OpenID Connect の ID Token が渡ってくるので、その ID Token を検証すれば ok という形です。

GCP からの HTTP リクエストをセキュアに認証する. はじめに | by Yuki Furuyama | google-cloud-jp | Medium

とありました。

検証方法は

ターゲットが Google Cloud の外部にある場合、受信サービスはトークンを手動で確認する必要があります。

HTTP ターゲットで認証を使用する  |  Cloud Scheduler のドキュメント  |  Google Cloud

とありました。

ここには 手動で確認 とありますが、別のドキュメントには

Using one of the Google API Client Libraries (e.g. Java, Node.js, PHP, Python) is the recommended way to validate Google ID tokens in a production environment.

Using a Google API Client Library | バックエンドサーバーで認証する  |  Google Sign-In for Websites  |  Google Developers

とあり、Pythonクライアントライブラリを使えばそこまで苦労せずに確認できそうでした。

　
そこで今回、Django + google-authで、Google Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイントを作成してみましたので、メモを残します。

　
目次

環境

• Google Cloud Scheduler
• Google Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイント
  • Python 3.10.7
  • Django 4.1.1
  • google-auth 2.12.0
    • なお、IDトークン検証時に id_token.verify_oauth2_token(received_id_token, requests.Request(), CLIENT_ID) というエラーになるため、 requests も必要
    • requests 2.28.1
  • PyJWT 2.5.0
    • IDトークンをデコードするために使用
    • 今回、動作確認のためインストールしているだけなので、実運用上は無しで良い
  • WSL2上のUbuntuに構築

IDトークンの中身を確認できるDjangoアプリを作成

まずは、Cloud Schedulerからどんな値を持つIDトークンが送られてくるかを確認するためのDjangoアプリを作成します。

DjangoでAPIエンドポイントを作る

HTTPリクエストを受け取れれば良いので、ふつうのDjangoアプリをWSL2上に構築します。

実装の詳細は記事の末尾にあるソースコードに譲り、ここでは概要だけ記載します。

　
プロジェクトの urls.py は

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('api.urls'))
]

として、アプリのurls.pyは

from django.urls import path

from api.views import CommandView, CommandWithAuthView

app_name = 'api'

urlpatterns = [
    path('health-check', HealthCheckView.as_view(), name='health-check'),
    path('command', CommandView.as_view(), name='command'),
]

とします。

　
次に、ViewではHTTPリクエストヘッダから Authorization ヘッダを取り出し、どんなIDトークンなのか見て見る感じにします。

ここで、IDトークンはエンコードされているため、 PyJWT を使ってヘッダとペイロードをデコードします。

• Reading Headers without Validation | Usage Examples — PyJWT 2.5.0 documentation
• Reading the Claimset without Validation | Usage Examples — PyJWT 2.5.0 documentation

なお、この時点ではIDトークンのデコードができれば良いことにして、中身の正しさは検証しません。

class CommandView(View):
    def get(self, request, *args, **kwargs):
        authz_header = request.headers.get('Authorization')  # => Bearer ***
        received_id_token = authz_header.replace('Bearer', '').lstrip()  # => IDトークンだけになる
        print(received_id_token)

        # JWTのヘッダを検証なしでdecode
        print(jwt.get_unverified_header(received_id_token))

        # JWTのペイロードを検証なしでdecode
        print(jwt.decode(received_id_token, options={"verify_signature": False}))

        return JsonResponse({
            'foo': 'bar'
        })

　
他に、リクエストが届くかどうか確認するためだけのViewも用意します。

class HealthCheckView(View):
    def get(self, request, *args, **kwargs):
        return JsonResponse({
            'ham': 'spam'
        })

Djangoアプリを作ったWSL2上に、ngrokを構築

上記のDjangoアプリをインターネット上にデプロイして確認することも考えましたが、環境構築をするのが手間でした。

そこで今回は、ngrokを使ってインターネットからDjangoアプリにHTTPリクエストが届くようにしてみます。
ngrok - Online in One Line

　
まずはngrokのサイトに行き、サインアップを行い、トークンを取得します。

次に、WSL2に ngrok をインストールします。

今回はWSL2上のUbuntuなので、 Install ngrok via Apt の方法でインストールします。

$ curl -s https://ngrok-agent.s3.amazonaws.com/ngrok.asc | sudo tee /etc/apt/trusted.gpg.d/ngrok.asc >/dev/null && echo "deb https://ngrok-agent.s3.amazonaws.com buster main" | sudo tee /etc/apt/sources.list.d/ngrok.list && sudo apt update && sudo apt install ngrok

　
実行後、インストールされたかを確認します。

$ ngrok version 3.1.0

　
続いて、 Add authtoken に従い、tokenを設定します。

$ ngrok config add-authtoken ***

Authtoken saved to configuration file: /home/user/.config/ngrok/ngrok.yml

　
最後に、ngrokを起動すると公開URLが表示されます。

$ ngrok http 8000
...
Forwarding                    https://***.jp.ngrok.io -> http://localhost:8000

　
アクセスできるかどうか、Windows上のターミナルからアクセスします。

> curl https://***.jp.ngrok.io/api/health-check

<!doctype html>
...

アクセスできたものの、ngrokのBrowser Warningページを取得していたようでした。

　
Browser WarningページではなくDjangoアプリのレスポンスを得たいため、リクエストヘッダに ngrok-skip-browser-warning を追加します。
ngrok - Online in One Line

再度 curl を実行すると、Djangoからのレスポンスがありました。

>curl -H 'ngrok-skip-browser-warning:hoge' https://***.jp.ngrok.io/api/health-check
{"ham": "spam"}

Cloud Scheduler のセットアップ

Django側の準備はできたため、次にCloud Schedulerのセットアップを行います。
https://console.cloud.google.com/cloudscheduler

ジョブを作成 をクリックし、新規ジョブを作成します。

まずは スケジュールを定義する ページです。

項目	値
名前	任意
リージョン	asia-northeast1 (東京)
説明	任意
頻度	任意 (今回は都度実行するため)
タイムゾーン	日本標準時(JST)

　
実行内容を構成する ページです。

項目	値
ターゲットタイプ	HTTP
URL	https://***.jp.ngrok.io/api/command (IDトークンを解析するURL)
HTTPメソッド	GET
HTTPヘッダ - ngrok-skip-browser-warning	任意の値
Authヘッダー	OIDCトークンを追加
サービスアカウント	新規作成
対象	(空白)

　
なお、今回作成するサービスアカウントはGoogle Cloudへアクセスしないため、ロールを設定せずに作成します。

　
オプションの設定 はデフォルトのままです。

今回はコンソールからジョブを強制実行するだけなので、最大再試行回数も 0 で良いです。

ジョブの強制実行

Cloud Schedulerのページに先ほど作成したジョブが表示されます。

右側のドットから ジョブを強制実行する をクリックしてジョブを実行します。

　
すると、Djangoの実行ログにIDトークンの情報が表示されます。

# IDトークンのヘッダ
{'alg': 'RS256', 'kid': '***', 'typ': 'JWT'}

# IDトークンのペイロード
{
  'aud': 'https://***.jp.ngrok.io/api/command',
  'azp': '***',
  'email': '***@***.iam.gserviceaccount.com',
  'email_verified': True,
  'exp': 1664324413,
  'iat': 1664320813,
  'iss': 'https://accounts.google.com',
  'sub': '***'
}

　
それぞれ

項目	値
aud	Cloud Schedulerで指定した 対象
email	Cloud Schedulerで指定したサービスアカウントのemail形式
iss	https://accounts.google.com で固定

なことが分かりました。

Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイントを作成

ここまで確認してきたことにより、 aud にはCloud Schedulerで指定した 対象 が入ると分かりました。

そのため、IDトークンを検証している以下のサンプルコードの CLIENT_ID は、 aud の値であるGoogle Schedulerで指定した 対象 の値をセットすれば良さそうでした。
https://developers.google.com/identity/sign-in/web/backend-auth?hl=ja#using-a-google-api-client-library

　
そこで、google-authを使って、Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイントを作成します。

Djangoアプリにエンドポイントを追加

Viewとurls.pyに修正を加えます。

CLIENT_ID = 'https://***.jp.ngrok.io/api/command-with-auth'

class CommandWithAuthView(View):
    def get(self, request, *args, **kwargs):
        authz_header = request.headers.get('Authorization')
        received_id_token = authz_header.replace('Bearer', '').lstrip()

        try:
            id_info = id_token.verify_oauth2_token(received_id_token, requests.Request(), CLIENT_ID)
            print(id_info)

            return JsonResponse({
                'status': 'success'
            })
        except ValueError as e:
            print(e)
            return JsonResponse({
                'status': 'unauthorized'
            })

　
urls.pyにも追加します。

urlpatterns = [
    # ...
    path('command-with-auth', CommandWithAuthView.as_view(), name='auth')  # 追加
]

Cloud Schedulerの設定変更

ジョブが動作した時にHTTPリクエストを飛ばす先を変更します。

実行内容を構成する の URL に、先ほど作成したエンドポイントのURL https://***.jp.ngrok.io/api/command-with-auth を指定します。

また、 対象 にデフォルト値として

オーディエンスは、OIDC トークンの受信者を制限します。通常は、ジョブのターゲット URL（URL パラメータなし）です。指定しなかった場合、Cloud Scheduler はデフォルトで、リクエスト パラメータを含む URL 全体をオーディエンスとして使用します。

が指定されているため、空白に戻してデフォルト値の設定とします。

Cloud Schedulerの実行とDjangoのログを確認

Cloud Schedulerを再実行すると、ログにIDトークンのペイロードが表示されました。 id_token.verify_oauth2_token の戻り値は、IDトークンのペイロードのようです。

[28/Sep/2022 12:31:54] "GET /api/command-with-auth HTTP/1.1" 200 21
{
  'aud': 'https://***.jp.ngrok.io/api/command-with-auth',
  'azp': '***',
  'email': '***@***.iam.gserviceaccount.com',
  'email_verified': True,
  'exp': 1664324413,
  'iat': 1664320813,
  'iss': 'https://accounts.google.com',
  'sub': '***'
}

　
次に、IDトークンの検証が失敗する場合の挙動を確認してみます。

今回はお手軽な方法として、定数 CLIENT_ID を aud とは異なる値にしてみます。

CLIENT_ID = 'https://***.jp.ngrok.io/api/command'

　
その状態で Cloud Scheduler を再実行すると、Djangoのログに以下が出力されました。

Token has wrong audience https://***.jp.ngrok.io/api/command-with-auth, expected one of ['https://***.jp.ngrok.io/api/command']
[28/Sep/2022 12:36:17] "GET /api/command-with-auth HTTP/1.1" 200 26

　
以上より、IDトークンを検証することで、Cloud SchedulerからのHTTPリクエストのみ受け付けるAPIエンドポイントを作成することができました。

ソースコード

Githubに上げました。
https://github.com/thinkAmi-sandbox/cloud_scheduler_handler_sample