【Streamlit v1.58.0】認証モジュールのパスワードの変更

認証モジュールの仕様として「ユーザー名(username)」自体はユニークで同じusernameは使えません。これはユーザー名がデータのキーとして使われているためですが、この重複はシステムが自動で判断して弾いてくれません。

SQLite3と連携して新規ユーザー登録機能を自作する場合は、開発者がコード側で「すでに同じ username が存在するかどうか」を事前にチェックして弾くロジックを実装する必要がありました。(※前回【Streamlit v1.46.1】認証モジュール + SQLite3を使ってログイン機能を実装では、その部分がありません。)

今回は、新規ユーザー登録する前に、usernameの重複が無いか事前にチェックする機能とパスワードの変更機能を実装する場合のメモ書きです。

また、テーブル設計ではusernameを主キーにしないといけないのですが、いかにも重複しそうなカラムを主キーにして、エラーチェックで弾くのはどうかと感じましたので、主キーは変更せず「【Streamlit v1.58.0】ログインユーザーと個別のファイルとを紐付ける」で紹介したテーブル構造のまま実装しています。

usernameの重複を回避

SQLite3のテーブル構造
conn = sqlite3.connect("データベース名")
cur = conn.cursor()
cur.execute("""
CREATE TABLE users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    username TEXT ,
    password TEXT NOT NULL,
    role TEXT default('member')
)
""")
重複を回避する方法

新規登録を行う際は、データベースに INSERT する前に必ず SELECT 文で既存のユーザーを検索し、存在する場合は画面に警告を表示して処理をストップさせます。

重複を事前に弾くコード例

# -------------------------
#  新規メンバー登録
# -------------------------
def regist_user(conn, username, hashed_password):
    conn = sqlite3.connect("st-members.db") # 一旦conn.close()しているので再度接続する!
    # 🔽【最重要】同じusernameがすでに存在するかチェック
    cur = conn.cursor()
    cur.execute("select username from users where username = ?",(username,))
    if cur.fetchone():
        # データが見つかった(重複している)場合は、エラーを表示して処理を終了
        st.error(f"エラー: 『{username}』はすでに使用されています。別のユーザー名をお試しください。")
        conn.close()
        return False
    try:
        cur.execute("insert into users (username, password) values (?, ?)",(username, hashed_password))
        conn.commit()
        conn.close()
        return True
    except Exception as e:
        st.error(f"ユーザー登録エラー: {e}")
        return False

ログインパスワードの変更

変更はauthenticator.reset_password()を実行するだけです。引数に渡した mycredentials 辞書の中身は、自動的に新しいハッシュ化されたパスワードに書き換わります。しかしこれは一時的な変更でアプリを再起動すると元に戻ります。必ず(SQLite3やMySQLなどで)永続化しておく必要があります。

パスワードの変更には、セキュリティ対策上の制約条件が自動で追加されます。

エラーコード

  • Between 8 and 20 characters long ⇨ 8文字以上20文字以下にすること
  • Contain at least one digit ⇨ 数字を含むこと
  • Contain at least one lowercase letter ⇨ 小文字を含むこと
  • Contain at least one uppercase letter ⇨ 大文字を含むこと
  • Contain at least one special character (@$!%*?&) ⇨ 特殊文字を使うこと

変更と永続化のコード例

if authenticator.reset_password(st.session_state["name"]):
    st.success("パスワードが正常に変更されました。")
    new_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
    # 以下にSQLite3でパスワードを更新します。
    update_password_db(user_id,new_hashed_pw)

変更するコード例

# -------------------------------------
#  パスワードの変更をDBに反映
# -------------------------------------
def update_password_db(id,hashed_password):
    conn = sqlite3.connect("st-members.db")
    cur = conn.cursor()
    cur.execute(
        "update users set password = ? where id =?",(hashed_password, id)
    )
    conn.commit()
    conn.close()

パスワードの変更をDBに反映させるコード例

まとめ

app.pyの全コードとアプリの動作確認

app.pyのコード例
#  v1.58.0
import streamlit as st
# ↓画像を扱いたい方は以下もimportすると便利です。
from PIL import Image
import sqlite3
import os
import streamlit_authenticator as stauth # 認証
import pandas as pd
import io
from pathlib import Path # os.pathで出来ることは全てできる


DATA_DIR = "users_data"
DATABASE_NAME = "st-members.db"
# -------------------------
# 状態初期化
# -------------------------
def init_state():
    if "state" not in st.session_state:
        st.session_state["state"] = "wait"
    if "mypage" not in st.session_state:
        st.session_state["mypage"] = "Home"

# -------------------------
# ログイン後のメイン遷移
#  main画面
# -------------------------
def Home():
    st.header('ホーム画面')
    # 🔽画像が無いと動作しませんので、各自ご対応お願いします。
    img1="./images/公園の河.jpg"
    img2="./images/palazzina寺院.jpg"
    col1,col2 = st.columns(2)
    with col1:
        image1= Image.open(img1)
        st.image(image1)
    with col2:
        image2 = Image.open(img2)
        st.image(image2)

def info():
    st.header('お知らせ')
    st.write("〇〇公園のひまわりが見頃です。")
    st.write("〇〇町の祇園祭が無事終了しました。")

def inquiry():
    st.header('お問い合わせ')
    st.text_input("氏名")
    st.text_input("連絡先")

              
# -----------------------------------------
# ログインユーザーの権限取得
# usernameと一致する id,roleを返す 
# -----------------------------------------
def get_prop(username):
    conn = sqlite3.connect("st-members.db")
    cur=conn.cursor()
    cur.execute("select id, role from users where username = ? ",(username,))
    row = cur.fetchone()
    if row:
        return {"id": str(row[0]), "role":row[1]}
    return None


# ---------------------------------------------
#  credentials辞書を作成して返す処理 
# ---------------------------------------------
def fetch_users(conn):
    try:
        cur = conn.cursor()
        cur.execute("select username ,password from users")
        rows = cur.fetchall()

        credentials = {"usernames":{}}
        # credentials = {}
        for row in rows:
            username, hashed_password = row
            tmp_dict = {
                'name': username,   # ここでusernameをnameとして使用
                'password': hashed_password,
            }
            # 🔽この辞書形式で保存される
            credentials["usernames"][username] =tmp_dict
        conn.close()
        return credentials
    
    except Exception as e:
        st.error(f"ユーザー情報取得エラー: {e}")
        return {"usernames": {}}


# -------------------------
#  新規メンバー登録
# -------------------------
def regist_user(conn, username, hashed_password):
    conn = sqlite3.connect("st-members.db") # 一旦conn.close()しているので再度接続する!
    # 🔽【最重要】同じusernameがすでに存在するかチェック
    cur = conn.cursor()
    cur.execute("select username from users where username = ?",(username,))
    if cur.fetchone():
        # データが見つかった(重複している)場合は、エラーを表示して処理を終了
        st.error(f"エラー: 『{username}』はすでに使用されています。別のユーザー名をお試しください。")
        conn.close()
        return False
    try:
        cur.execute("insert into users (username, password) values (?, ?)",(username, hashed_password))
        conn.commit()
        conn.close()
        return True
    except Exception as e:
        st.error(f"ユーザー登録エラー: {e}")
        return False


# -------------------------------------
#  パスワードの変更をDBに反映
# パスワードをデータベースに上書き保存
# -------------------------------------
def update_password_db(id,hashed_password):
    conn = sqlite3.connect("st-members.db")
    cur = conn.cursor()
    cur.execute(
        "update users set password = ? where id =?",(hashed_password, id)
    )
    conn.commit()
    conn.close()


# -------------------------
#   メイン関数
# -------------------------
def main():
    conn = sqlite3.connect("st-members.db")
    if conn is None:
        st.error("データベースへの接続に失敗しました。アプリケーションを終了します。")
        st.stop() # データベース接続ができない場合はここで停止

    mycredentials = fetch_users(conn)

    # ↓本番環境ではしてはいけない。
    mysignature_key = 'super_secret34567815!#$_verylonglong-hot'
    # 認証オブジェクトの初期化
    authenticator = stauth.Authenticate(
        mycredentials,
        'streamlit_v1.47_login_cookie',  # 任意のクッキー名
        mysignature_key,                 # 任意の署名キー
        cookie_expiry_days=30,          # クッキーの有効期限(日数)
    )

    # サイドバーの設計
    with st.sidebar:
        # 状態初期化
        init_state()
        # 待機状態
        if st.session_state["state"]== "wait":
            if st.button("新規登録はこちらから"):
                st.session_state["state"] = "signin_req"
                st.rerun()
            st.divider()
            
            # st.header('ログイン') 
            authenticator.login(location="sidebar",fields={
                "Form name": "ログイン",
                "Username": "ユーザー名",
                "Password": "パスワード",
                "Login": "ログイン"
            })

            # if auth_status:
            if st.session_state["authentication_status"] :
                st.session_state["state"] = "logined"
                
            elif st.session_state["authentication_status"] is False:
                st.error("ユーザー名またはパスワードが間違っています")
            elif st.session_state["authentication_status"] is None:
                st.warning("ユーザー名とパスワードを入力してください")
        # ログイン成功
        elif st.session_state["state"]== "logined":
            if st.session_state["authentication_status"]:
                st.write("ログインに成功しました。")
                st.success(f"ようこそ {st.session_state['name']} さん!")
            
            # ----------------------
            # selectboxで遷移先選択
            # ----------------------
            dash_page = st.selectbox('選択してください',['Home','お知らせ', 'お問い合わせ'],key="sidebar_page")
            st.session_state["mypage"] = dash_page

            # ---------------------------
            # パスワード更新 @2026/09/10
            # ---------------------------
            st.divider()
            st.subheader("⚙️ パスワードの変更")
            try:
                # Between 8 and 20 characters long
                # Contain at least one uppercase letter
                # Contain at least one special character (@$!%*?&)
                old_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
                if authenticator.reset_password(st.session_state["name"]):
                    st.success("パスワードが正常に変更されました。")
                    new_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
                    user_name = st.session_state["name"]
                    row = get_prop(user_name)
                    user_id = row["id"]
                    update_password_db(user_id,new_hashed_pw)

            except Exception as e:
                st.error(e)

            st.divider()
            st.subheader("ログアウトしますか")
            if st.session_state["authentication_status"] is True :
                authenticator.logout("ログアウト", location="sidebar")

            if st.session_state["authentication_status"] is None :
                st.session_state["state"] = "wait"
                
                st.warning("ログアウトしました。再度ログインしてください。")
                st.rerun()
        # 新規登録
        elif st.session_state["state"] == "signin_req":
            st.header('登録フォーム')
            st.write("アカウントをお持ちでない方はこちらから登録してください。")

            with st.form("register_user_form"):
                username = st.text_input("ユーザー名")
                new_password = st.text_input("パスワード", type="password")
                confirm_password = st.text_input("パスワードの確認", type="password")
                
                submit_registration = st.form_submit_button("登録する")
                if submit_registration:
                    if username and new_password and confirm_password:
                        if new_password == confirm_password :
                            try:
                                hashed_password = stauth.Hasher.hash(new_password)
                                if regist_user(conn, username, hashed_password):
                                    st.session_state["state"] = "wait"
                                    st.markdown("<span style='color:blue;'>ご登録ありがとうございました。</span>", unsafe_allow_html=True) 
                                    st.rerun() 

                            except Exception as e:
                                st.error(f"パスワードのハッシュ化または登録中にエラーが発生しました: {e}")
                        else:
                            st.error("パスワードが一致しません。再度確認してください。")
                    else :
                        st.error("全ての登録項目を入力してください。")

            st.divider()
            if st.button("ログイン画面に戻る"):
                st.session_state["state"] = "wait"
                st.rerun()

    # -------------------------
    # メイン画面の切り替え処理
    # -------------------------
    if st.session_state["state"] == "logined":
        page = st.session_state.get("mypage", "Home")
        # st.header('ダッシュボード')
        # 画面遷移
        if page == "Home":
            Home()
        elif page == "お知らせ":
            info()
        elif page == "お問い合わせ":
            inquiry()

    
if __name__ == '__main__':
    main()
アプリの動作確認

①同じusernameでの登録を弾く処理

②ログインパスワードの変更処理

以上です😁

Follow me!