REVISIOのエンジニア植草です。
先日片岡から Streamlitについて投稿がございましたが、Snowflake上でStreamlitのコンテンツを 扱えるようになり、Snowflake・Streamlitのアップデートのスピード感を目の当たりにしているところです。
私もStreamlitを触り始めて早二ヶ月、いろんなコンテンツが簡単に作れて本当に優れたフレームワークだと日々感じています。
今回は開発したコンテンツを外部公開する場合を想定し、ログイン機能についてStreamlitとAWS Cognitoを使って実装してみました。
まずは作ったサンプル画面です。
3つの画面で構成されています。 ログインボタンがあるフロント画面→ログイン画面→コンテンツ画面というシンプルな構成です。
ログインボタンを押下するとログイン画面へ遷移
後述いたしますが、Cognitoで予め用意されたログイン画面
ログインするとグラフコンテンツを表示
イメージして頂けましたでしょうか？
こちらを以下手順に沿ってローカルで起動できるよう実装していきたいと思います。
- 1. AWS Cognitoのユーザープールの作成
- 2.Cognitoドメインの作成
- 3.ホストされたUIの設定
- 4. ユーザの作成
- 5. Streamlitでの画面実装
- 6. ローカルでコンテンツを起動する
1. AWS Cognitoのユーザープールの作成
ユーザ認証についてはAWS Cognitoに任せたいのでまずはユーザープールを作成していきます。
AWS Cognitoとは ウェブアプリとモバイルアプリ用のアイデンティティプラットフォームです。これは、ユーザーディレクトリ、認証サーバー、OAuth 2.0 アクセストークンと AWS 認証情報の承認サービスです。Amazon Cognito を使用すると、組み込みのユーザーディレクトリ、エンタープライズディレクトリ、Google や Facebook などのコンシューマー ID プロバイダーからユーザーを認証および承認できます。
AWS Cognitoで認証の流れは以下のとおりです
こちらのBlackBeltが分かりやすいので詳細を知りたい方は以下をご参照ください。 https://d1.awsstatic.com/webinars/jp/pdf/services/20200630_AWS_BlackBelt_Amazon%20Cognito.pdf
サインインエクスペリエンスを設定
Cognitoユーザープールの「サインインオプション」では「ユーザー名」、ユーザー名の要件で「ユーザーが任意のユーザー名をサインインすることを許可する」にチェックを入れて次へ
セキュリティ要件を設定
「パスワードポリシーモード」ではCognitoのデフォルトを選択、「多要素認証」では「MFAなし」を選択（セキュアにする場合はMFAを必須にしたほうがよいが、今回はなし）
サインアップエクスペリエンスを設定
「セルフサービスのサインアップ」では「自己登録を有効化」のチェックを外す（ユーザープールでユーザーサインアップをアクティブ化すると、インターネット上のあらゆるユーザーがアカウントにサインアップしてアプリケーションにサインインできてしまいますので基本はチェックを外す）
「検証とユーザーアカウントの確認」では検証する属性に「Eメールのメッセージを送信、Eメールアドレスを検証」を選択、それ以外はデフォルトのままで次へ
メッセージ配信を設定
「Eメールプロバイダー」は「CognitoでEメールを送信」を選択、それ以外はデフォルトままで次へ
アプリケーションを統合
「ユーザープール名」に今回は「Streamlit-Cognito-Test」という名前を設定（任意）、「ホストされた認証ページ」では「CognitoのホストされたUIを使用」にチェックを入れない（ホストされたUIを使用しますが、後で設定できますので一旦、チェックを入れないでおきます）
「最初のアプリケーションクライアント」の「アプリケーションタイプ」では「機密クライアント」を指定します。
「アプリケーションクライアント名」に今回は「streamlit-cognito-test」という名前を設定（任意）、「クライアントシークレット」は「クライアントのシークレットを生成する」を選択（トークンを取得する際にクライアントシークレットを使用するため）、それ以外はデフォルトのまま次へを押すと確認が画面が表示され、生成ボタンを押すと作成完了となります。
2.Cognitoドメインの作成
「Cognitoドメインの作成」を選択
ドメインに任意の名前を指定（cognitoなどの名前は使えない）
3.ホストされたUIの設定
「ホストされたUI」の編集画面へ
「ホストされたサインアップページとサインインページ」で「URL」に今回はローカルとCognitoを連携させるので「http://localhost:8501/」を設定、IDプロバイダーは「Cognitoユーザープール」を選択
「OAuth 2.0許可タイプ」では「認証コードを付与」を選択（Authorization code grantはログイン後に認可コードを取得し、その認可コードをもとにアクセストークンを取得して初めて認証される）、「OpenID Connect」のスコープでは「OpenID」を選択
4. ユーザの作成
本コンテンツにログインするユーザーを作成します。
「招待メッセージ」は「Eメールで招待を送信」にするとアカウント情報を付与してメールが送信されます。（メールの内容もカスタマイズ可能です）今回は「招待を送信しない」を選択。
「ユーザ名」、「Eメールアドレス」の入力、「Eメールアドレスを検証済みとしてマークする」にチェック、「仮パスワード」は招待メールを送信しないので「パスワードの設定」を選択し、任意のパスワードを入力し、ユーザを生成します。
コンテンツ作成後にこのユーザを使ってログインしていきます。
5. Streamlitでの画面実装
1〜4を通してAWS Cognitoにてユーザープール及びユーザの作成が完了いたしましたのでいよいよコンテンツを実装していきます。 前述の通りログインボタンがあるフロント画面→ログイン画面→AWS Cognitoへの認証→コンテンツ画面という流れになります。
プログラム構成は以下のようになります。Dockerで環境を構築します。
. ├── app │ ├── __init__.py │ ├── authenticate.py -- Cognitoへのアクセス、ログイン画面の生成など │ ├── constants.py -- .envから環境変数を設定するコンフィグ設定 │ ├── login.py -- フロント画面 │ ├── main.py -- グラフコンテンツを表示 │ └── static │ └── Logo_REVISIO_color.png -- フロント画面の画像 ├── .env 環境変数の設定 ├── docker-compose.yml └── Dockerfile
.envファイル:
環境変数となり、Cognitoへ認証する際に使用します。
COGNITO_DOMAIN=<2.で作成のCognitoドメイン> CLIENT_ID=<1.のアプリケーション統合で作成時に払い出されたクライアントID> CLIENT_SECRET=<アプリケーション統合で作成時に払い出されたクライアントシークレット> APP_URI=http://localhost:8501/
Dockerfile:
pipでstreamlit、pandas等をインストールします。
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update \
&& apt-get install --yes --no-install-recommends \
gcc \
g++ \
build-essential \
python3-dev \
vim
# ライブラリのインストール
RUN pip install --upgrade pip
RUN pip install streamlit pandas
# ソースのコピー
COPY app/ /app/
# コマンド実行
# 静的ファイルの使用するにはenableStaticServingオプションを有効にする
CMD ["streamlit", "run", "--server.port", "8501", "--server.enableStaticServing", "true", "/app/login.py"]
docker-compose.yml:
version: '3'
services:
streamlit:
build:
context: .
dockerfile: Dockerfile
ports:
- 8501:8501
volumes:
- ./app:/app
env_file:
- .env
login.py:
ログインボタンがあるフロント画面になります。
後述しますが、authenticate.pyからログインボタン、ログアウトボタンを生成。
認証が有効なら、main.pyにあるコンテンツを表示します。
import streamlit as st import authenticate as authenticate import main st.set_page_config(page_title='Streamlit-Cognito', page_icon=':chart_with_upwards_trend:', layout='wide', initial_sidebar_state="collapsed") # Cognitoユーザープールへのアクセス authenticate.set_st_state_vars() # ログイン画面のスタイル設定 authenticate.set_style_login() # 認証が有効 if st.session_state["authenticated"]: # ログアウト画面を表示（サイドバー） authenticate.button_logout() # メインページを表示 # コンテンツ内でエラーが発生した場合、生のエラーは表示せず、エラー文言のみ表示 try: main.show() except Exception as e: st.error('アプリケーションエラー！！') else: # ログイン画面表示 authenticate.button_login()
main.py:
コンテンツ画面となります。
st.area_chartを使って面グラフを表示します。
import streamlit as st import pandas as pd import numpy as np def show(): st.write('Streamlit-Cognitoアプリ') glagh = st.empty() with glagh.container(): chart_data = pd.DataFrame( np.random.randn(20, 3), columns=['A社', 'B社', 'C社'] ) st.area_chart(chart_data)
authenticate.py:
Cognito認証まわりの実装となります。
さきほどご紹介した認証フローを見ながら、コードを見て頂くと分かりやすいかもしれません。
import os import streamlit as st import requests import base64 import json from constants import Config import random import string @st.cache_data(ttl=86400, show_spinner=False, max_entries=100) def _generate_random_state(length=16): """ Cognitoのstate値を取得 ランダムな文字列を生成してstateパラメータとして使用する cache_dataとして保持しておくことでstateパラメータのチェックが可能 Returns: state値 """ letters_and_digits = string.ascii_letters + string.digits return ''.join(random.choice(letters_and_digits) for _ in range(length)) def _initialise_st_state_vars(): """ Cognito関連のセッション情報の初期化 Returns: Nothing. """ # 認可コード if "auth_code" not in st.session_state: st.session_state["auth_code"] = "" # Cognitoの認証チェック if "authenticated" not in st.session_state: st.session_state["authenticated"] = False # ユーザ情報 if "user_info" not in st.session_state: st.session_state["user_info"] = "" # Cognitoのstateパラメータを生成してセッションに保持 if "state" not in st.session_state: st.session_state['state'] = _generate_random_state() def _get_auth_code(): """ ログイン後にGETでパラメータから認可コード(code)を取得 Returns: auth_code: 認可コード """ # experimental_get_query_params(GET)でURLからパラメータを取得 auth_query_params = st.experimental_get_query_params() try: # 認可コードをパラメータから取得 auth_code = dict(auth_query_params)["code"][0] # stateパラメータを取得 received_state = dict(auth_query_params)["state"][0] # stateパラメータがセッションに保持したものと一致しているか if st.session_state['state'] != received_state: raise KeyError except (KeyError, TypeError): auth_code = "" return auth_code def _set_auth_code(): """ 認可コードをセッションに保持 Returns: Nothing. """ # セッションの初期化 _initialise_st_state_vars() # 認証コードを取得 auth_code = _get_auth_code() st.session_state["auth_code"] = auth_code def _get_user_tokens(auth_code): """ トークンURLに対しCoginitoのクライアントID、クライアントシークレット、取得した認可コードをもとに ユーザ認証してアクセストークンとIDトークンを取得 Args: auth_code: Cognitoサーバから取得した認可コード Returns: { 'access_token': ユーザ認証に成功した場合、cognitoサーバからアクセストークンを取得 'id_token': ユーザ認証に成功した場合、cognitoサーバからIDトークンを取得 } """ # トークンエンドポイント token_url = f"{Config.COGNITO_DOMAIN}/oauth2/token" client_secret_string = f"{Config.CLIENT_ID}:{Config.CLIENT_SECRET}" client_secret_encoded = str( base64.b64encode(client_secret_string.encode("utf-8")), "utf-8" ) headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {client_secret_encoded}", } body = { "grant_type": "authorization_code", "client_id": Config.CLIENT_ID, "code": auth_code, "redirect_uri": Config.APP_URI, } token_response = requests.post(token_url, headers=headers, data=body) try: access_token = token_response.json()["access_token"] id_token = token_response.json()["id_token"] except (KeyError, TypeError): access_token = "" id_token = "" return access_token, id_token def _get_user_info(access_token): """ 取得したアクセストークンでユーザ情報を取得 Args: access_token: cognitoのユーザープールから取得したアクセストークン Returns: userinfo_response: json object. """ # ユーザ情報エンドポイント userinfo_url = f"{Config.COGNITO_DOMAIN}/oauth2/userInfo" headers = { "Content-Type": "application/json;charset=UTF-8", "Authorization": f"Bearer {access_token}", } userinfo_response = requests.get(userinfo_url, headers=headers) return userinfo_response.json() def set_st_state_vars(): """ 取得した認証コード、アクセストークン、IDトークンをユーザ情報の取得、セッションへの保持 Returns: Nothing. """ # セッション情報の初期化 _initialise_st_state_vars() # ログイン前は認可コードは取得できない。 # ログイン後初めて認可コードをURLから取得 auth_code = _get_auth_code() # アクセストークン、IDトークンを取得 access_token, id_token = _get_user_tokens(auth_code) # アクセストークンを取得できたら、セッションに保持 if access_token != "": st.session_state["auth_code"] = auth_code st.session_state["authenticated"] = True # カスタム属性の取得には属性の読み取りおよび書き込み許可で許可する必要あり st.session_state["user_info"] = _get_user_info(access_token) def set_style_login(): """ ログイン、ログアウトコンテンツのスタイル設定 Returns: Nothing. """ st.markdown(""" <style> .button-login { font-family: 'メイリオ'; font-weight: bold; background-color: #797279; color: white !important; padding: 1em 1.5em; text-decoration: none; text-transform: uppercase; top: 40%; left: 50%; transform: translate(-50%, -50%); width: 150px; height: 80px; border-radius: 3px; } .button-logout { display: block !important; font-family: 'メイリオ'; font-weight: bold; background-color: #575057; color: white !important; padding: 10px 20px; text-decoration: none; text-transform: uppercase; transform: translate(-1%, -100%); margin: 0 auto; text-align: center; font-size: 12px; width: 100px; border-radius: 3px; } .inline-block { position: relative; height: 100vh; } .button-login:hover { background-color: #FF5B5A; text-decoration: none; } .button-login:active { background-color: black; } .button-logout:hover { background-color: #FF5B5A; text-decoration: none; } .button-logout:active { background-color: black; } .img-style { position: absolute; top: 30%; left: 50%; transform: translate(-50%, -50%); width: 15%; height: 5%; } </style> """, unsafe_allow_html=True) def button_login(): """ ログインコンテンツの出力 Returns: ログインコンテンツのHTML """ # Authorization Code Grant（認可コード許可） # ログインボタンにCognitoのホストUI（ログイン画面）のリンクを設定 login_link = f"{Config.COGNITO_DOMAIN}/login?client_id={Config.CLIENT_ID}&state={st.session_state['state']}&response_type=code&scope=openid&redirect_uri={Config.APP_URI}" html_button_login = f""" <div class='inline-block'> <img class='img-style' src='/app/static/Logo_REVISIO_color.png' > <a href='{login_link}' class='inline-block button-login' target='_self'>Log In</a> </div> """ return st.markdown(f"{html_button_login}", unsafe_allow_html=True) def button_logout(): """ ログアウトコンテンツの出力（サイドバー） Returns: ログアウトコンテンツのHTML """ logout_link = f"{Config.COGNITO_DOMAIN}/logout?client_id={Config.CLIENT_ID}&state={st.session_state['state']}&response_type=code&redirect_uri={Config.APP_URI}" html_button_logout = f""" <a href='{logout_link}' class='button-logout' target='_self'>Log Out</a> """ return st.sidebar.markdown(f"{html_button_logout}", unsafe_allow_html=True)
authenticate.pyについて主要な箇所を抜粋して説明していきます。
button_login:
ボタンのリンク先はCognitoのホストUI（ログイン画面）となります。
CognitoのホストUIは設定したCognitoドメインとなります。
Cognitoドメインに対しパラメータとしてclient_idはアプリケーション統合のクライアントID、stateパラメータを指定（CSRF対策）、response_typeはcode(Authorization Code Grant)、scopeはopenid、redirect_uriはログイン後にアクセスコンテンツのURLを指定します。
def button_login(): """ ログインコンテンツの出力 Returns: ログインコンテンツのHTML """ # Authorization Code Grant（認可コード許可） # ログインボタンにCognitoのホストUI（ログイン画面）のリンクを設定 login_link = f"{Config.COGNITO_DOMAIN}/login?client_id={Config.CLIENT_ID}&state={st.session_state['state']}&response_type=code&scope=openid&redirect_uri={Config.APP_URI}" html_button_login = f""" <div class='inline-block'> <img class='img-style' src='/app/static/Logo_REVISIO_color.png' > <a href='{login_link}' class='inline-block button-login' target='_self'>Log In</a> </div> """ return st.markdown(f"{html_button_login}", unsafe_allow_html=True)
set_st_state_vars:
CognitoのホストUIでログインすると認可コードが発行され、その認可コードをもとにアクセストークン、IDトークンを取得。
そのアクセストークンからアカウントのユーザ情報を取得します。
def set_st_state_vars(): """ 取得した認可コード、アクセストークン、IDトークンをユーザ情報の取得、セッションへの保持 Returns: Nothing. """ # セッション情報の初期化 _initialise_st_state_vars() # ログイン前は認可コードは取得できない。 # ログイン後初めて認可コードをURLから取得 auth_code = _get_auth_code() # アクセストークン、IDトークンを取得 access_token, id_token = _get_user_tokens(auth_code) # アクセストークンを取得できたら、セッションに保持 if access_token != "": st.session_state["auth_code"] = auth_code st.session_state["authenticated"] = True # カスタム属性の取得には属性の読み取りおよび書き込み許可で許可する必要あり st.session_state["user_info"] = _get_user_info(access_token)
_get_auth_code:
experimental_get_query_paramsを使ってURLから認可コードを取得します。
またstateパラメータも取得してセッション（session_state）に保持しているstateと一致しているチェック（CSRF対策）します。
セッションに保持しているstateは後述しますが、st.cache_dataにstreamlitのキャッシュ機能を使用して保持します。
本来stateをデータストアに保持しておいてstateパラメータにある値と検証させますが、そのデータストアをst.cache_dataを使って
キャッシュに保持して実現しています。
なおexperimental_get_query_paramsはExperimental featuresとなっておりますので使用する際はバージョンを固定するなどして
ご使用ください。
def _get_auth_code(): """ ログイン後にGETでパラメータから認可コード(code)を取得 Returns: auth_code: 認可コード """ # experimental_get_query_params(GET)でURLからパラメータを取得 auth_query_params = st.experimental_get_query_params() try: # 認可コードをパラメータから取得 auth_code = dict(auth_query_params)["code"][0] # stateパラメータを取得 received_state = dict(auth_query_params)["state"][0] # stateパラメータがセッションに保持したものと一致しているか if st.session_state['state'] != received_state: raise KeyError except (KeyError, TypeError): auth_code = "" return auth_code
_generate_random_state:
ランダムな値を生成してstateに設定します。
st.cache_dataでキャッシュに保持します。ttl（秒）でキャッシュの保持期間を指定。
キャッシュの仕様についてはこちらを参照ください。
@st.cache_data(ttl=86400, show_spinner=False, max_entries=100) def _generate_random_state(length=16): """ Cognitoのstate値を取得 ランダムな文字列を生成してstateパラメータとして使用する cache_dataとして保持しておくことでstateパラメータのチェックが可能 Returns: state値 """ letters_and_digits = string.ascii_letters + string.digits return ''.join(random.choice(letters_and_digits) for _ in range(length))
_get_user_tokens:
さきほど取得した認可コードとアプリケーション統合作成時に生成されたCoginitoのクライアントID、クライアントシークレットをもとに
設定したCoginitoドメインのトークンエンドポイントからアクセストークン、IDトークンを取得します。
def _get_user_tokens(auth_code): """ トークンURLに対しCoginitoのクライアントID、クライアントシークレット、取得した認可コードをもとに ユーザ認証してアクセストークンとIDトークンを取得 Args: auth_code: Cognitoサーバから取得した認可コード Returns: { 'access_token': ユーザ認証に成功した場合、cognitoサーバからアクセストークンを取得 'id_token': ユーザ認証に成功した場合、cognitoサーバからIDトークンを取得 } """ # トークンエンドポイント token_url = f"{Config.COGNITO_DOMAIN}/oauth2/token" client_secret_string = f"{Config.CLIENT_ID}:{Config.CLIENT_SECRET}" client_secret_encoded = str( base64.b64encode(client_secret_string.encode("utf-8")), "utf-8" ) headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {client_secret_encoded}", } body = { "grant_type": "authorization_code", "client_id": Config.CLIENT_ID, "code": auth_code, "redirect_uri": Config.APP_URI, } token_response = requests.post(token_url, headers=headers, data=body) try: access_token = token_response.json()["access_token"] id_token = token_response.json()["id_token"] except (KeyError, TypeError): access_token = "" id_token = "" return access_token, id_token
_get_user_info(access_token):
取得したアクセストークンをもとにCognitoのユーザ情報(メールアドレスなど)を取得します。
こちらはCognitoのさきほどCognitoのユーザ作成で作成したユーザ情報となります。
ユーザ情報はデフォルトのユーザ情報の他にカスタム属性として追加で付与でできる属性もありますので
追加することで権限管理なども柔軟に対応することが可能です。
def _get_user_info(access_token) """ 取得したアクセストークンでユーザ情報を取得 Args: access_token: cognitoのユーザープールから取得したアクセストークン Returns: userinfo_response: json object. """ # ユーザ情報エンドポイント userinfo_url = f"{Config.COGNITO_DOMAIN}/oauth2/userInfo" headers = { "Content-Type": "application/json;charset=UTF-8", "Authorization": f"Bearer {access_token}", } userinfo_response = requests.get(userinfo_url, headers=headers) return userinfo_response.json()
constants.py:
.envの情報を設定します。
import os class Config(): COGNITO_DOMAIN = os.getenv("COGNITO_DOMAIN", '') CLIENT_ID = os.getenv("CLIENT_ID", '') CLIENT_SECRET = os.getenv("CLIENT_SECRET", '') APP_URI = os.getenv("APP_URI", '')
6. ローカルでコンテンツを起動する
実装内容についてご理解頂けましたでしょうか？
このコンテンツについてDockerをビルドしてローカルで起動してみます。
docker-compose -f docker-compose.yml up -d --build
ブラウザでhttp://localhost:8501を開くと、以下のように冒頭でご紹介いたしました画面が表示されます。
4.で作成したユーザでログインしてみます。
コンテンツ画面が表示されました。
まとめ
Cognitoの認証まわりの実装が少し理解が必要ですが、ユーザ管理はすべてCognitoに任せることができるのでコンテンツ開発に注力できそうですね。
Streamlitのコンテンツを外部公開する際はCognitoを利用するのもいいかもしれません。
