AWS Lambda から Google スプレッドシートに OAuth 2.0 で行を追記してみた

#!/bin/bash

# Google API の OAuth 2.0 リフレッシュトークンを取得する。
#
# ブラウザでのログイン・同意は自動化できないため、認可 URL を開いてもらい
# リダイレクト後の URL を貼り付けてもらう対話式にしている。
#
# Parameter Store への反映はこのスクリプトでは行わない。
# 出力されたリフレッシュトークンを手動で書き込むこと。
#
# 使い方:
#   scripts/rotate-refresh-token.sh <client_id>
#   （client_secret は実行後にプロンプトから入力する）

set -euo pipefail

redirect_uri="http://localhost:8080"
scope="https://www.googleapis.com/auth/spreadsheets"
parameter_name="/myapp/sheets/oauth/refresh-token"

client_id="${1:-}"

if [ -z "$client_id" ]; then
  echo "使い方: $0 <client_id>" >&2
  exit 1
fi

# client_secretはプロセス一覧やシェル履歴に残さないため、引数ではなく非エコー入力で受け取る
read -rsp "client_secretを入力してください: " client_secret
echo
if [ -z "$client_secret" ]; then
  echo "client_secretが入力されていません。" >&2
  exit 1
fi

function generate_state() {
  python3 -c 'import secrets; print(secrets.token_urlsafe(32), end="")'
}

function build_auth_url() {
  STATE="$1" CLIENT_ID="$client_id" REDIRECT_URI="$redirect_uri" SCOPE="$scope" python3 -c '
import os
from urllib.parse import urlencode

params = {
    "client_id": os.environ["CLIENT_ID"],
    "redirect_uri": os.environ["REDIRECT_URI"],
    "response_type": "code",
    "scope": os.environ["SCOPE"],
    "access_type": "offline",
    "prompt": "consent",
    "state": os.environ["STATE"],
}
print("https://accounts.google.com/o/oauth2/v2/auth?" + urlencode(params), end="")
'
}

# 貼り付けられたURLを検証してcodeのみを標準出力へ返す。
# codeは一度きりの値だがargvに載せないよう、URLは環境変数経由で渡す
function extract_code() {
  REDIRECTED_URL="$1" EXPECTED_STATE="$2" EXPECTED_REDIRECT_URI="$redirect_uri" python3 -c '
import os
import sys
from urllib.parse import parse_qs, urlparse

parsed = urlparse(os.environ["REDIRECTED_URL"])
expected = urlparse(os.environ["EXPECTED_REDIRECT_URI"])

if (parsed.scheme, parsed.hostname, parsed.port) != (expected.scheme, expected.hostname, expected.port):
    print(f"リダイレクト先が想定と異なります: {parsed.scheme}://{parsed.netloc}", file=sys.stderr)
    sys.exit(1)

qs = parse_qs(parsed.query)

error = qs.get("error", [""])[0]
if error:
    print(f"認可が拒否されました: {error}", file=sys.stderr)
    sys.exit(1)

# CSRF対策。認可URLに含めたstateと一致しないURLのcodeは信用しない
if qs.get("state", [""])[0] != os.environ["EXPECTED_STATE"]:
    print("stateが一致しません。認可URLを開き直してやり直してください。", file=sys.stderr)
    sys.exit(1)

code = qs.get("code", [""])[0]
if not code:
    print("URLからcodeを取得できませんでした。貼り付けたURLを確認してください。", file=sys.stderr)
    sys.exit(1)

print(code, end="")
'
}

function exchange_code_for_refresh_token() {
  local code="$1"
  local body response http_status response_body refresh_token

  # client_secret/codeをcurlのargvに載せないよう、リクエストボディを組み立てて標準入力から渡す
  body=$(CLIENT_ID="$client_id" CLIENT_SECRET="$client_secret" CODE="$code" REDIRECT_URI="$redirect_uri" python3 -c '
import os
from urllib.parse import urlencode

print(urlencode({
    "client_id": os.environ["CLIENT_ID"],
    "client_secret": os.environ["CLIENT_SECRET"],
    "code": os.environ["CODE"],
    "grant_type": "authorization_code",
    "redirect_uri": os.environ["REDIRECT_URI"],
}), end="")
')

  # codeは一度きりの値なので、通信停滞で無言のまま待ち続けないようタイムアウトを明示する
  response=$(printf '%s' "$body" | curl -sS -X POST https://oauth2.googleapis.com/token \
    --connect-timeout 10 --max-time 30 \
    -H "Content-Type: application/x-www-form-urlencoded" \
    --data-binary @- \
    -w $'\n%{http_code}') || {
    echo "トークンエンドポイントへのリクエストに失敗しました。" >&2
    return 1
  }

  http_status="${response##*$'\n'}"
  response_body="${response%$'\n'*}"

  if [ "$http_status" != "200" ]; then
    # レスポンス全文にはaccess_tokenが含まれうるため、errorとerror_descriptionのみ出す
    echo "トークンエンドポイントがHTTP ${http_status}を返しました。" >&2
    RESPONSE_BODY="$response_body" python3 -c '
import json
import os
import sys

try:
    data = json.loads(os.environ["RESPONSE_BODY"])
except json.JSONDecodeError:
    print("(レスポンスをJSONとして解釈できませんでした)", file=sys.stderr)
    sys.exit(0)

error = data.get("error")
error_description = data.get("error_description")
print(f"error: {error}, error_description: {error_description}", file=sys.stderr)
'
    return 1
  fi

  refresh_token=$(RESPONSE_BODY="$response_body" python3 -c '
import json
import os

print(json.loads(os.environ["RESPONSE_BODY"]).get("refresh_token", ""), end="")
')

  if [ -z "$refresh_token" ]; then
    echo "レスポンスにrefresh_tokenが含まれていませんでした。" >&2
    echo "既に同意済みの場合、https://myaccount.google.com/permissions でこのアプリへのアクセスを一度取り消してから再実行してください。" >&2
    return 1
  fi

  printf '%s' "$refresh_token"
}

function main() {
  local state auth_url redirected_url code refresh_token

  state=$(generate_state)
  auth_url=$(build_auth_url "$state")

  echo "以下のURLをブラウザで開き、対象のGoogleアカウントでログイン・同意してください。"
  echo "----------------------------"
  echo "$auth_url"
  echo "----------------------------"
  echo

  echo "同意後にリダイレクトされるURL（http://localhost:8080/?code=... 、ページ自体は表示エラーで構わない）を貼り付けてください:"
  read -r redirected_url

  if ! code=$(extract_code "$redirected_url" "$state"); then
    exit 1
  fi

  # コマンド置換は関数のexit statusを自動では伝播しないため、明示的に判定する
  if ! refresh_token=$(exchange_code_for_refresh_token "$code"); then
    exit 1
  fi

  echo "----------------------------"
  echo "取得したrefresh_token:"
  echo "$refresh_token"
  echo "----------------------------"
  echo
  echo "Parameter Storeへの反映は行っていません。以下のコマンドで手動更新してください:"
  echo "  aws ssm put-parameter --name ${parameter_name} \\"
  echo "    --value \"<refresh_token>\" --type SecureString --overwrite"
}

main