-
-
Save ianchen06/2258dc3153c9e1ddeeb75884cbb9e442 to your computer and use it in GitHub Desktop.
Streamlit + Firebase authorization example
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import time | |
| import json | |
| from contextlib import suppress | |
| from datetime import datetime, timedelta | |
| from typing import Any, Dict, Final, Optional, Sequence, Union | |
| import extra_streamlit_components as stx | |
| import firebase_admin | |
| import jwt | |
| import requests | |
| import streamlit as st | |
| from email_validator import EmailNotValidError, validate_email | |
| from firebase_admin import auth | |
| from password_strength import PasswordPolicy | |
| TITLE: Final = "Example page" | |
| def pretty_title(title: str) -> None: | |
| """ | |
| Make a centered title, and give it a red line. Adapted from 'streamlit_extras.colored_headers' | |
| package. | |
| Parameters: | |
| ----------- | |
| title : str | |
| The title of your page. | |
| """ | |
| st.markdown( | |
| f"<h2 style='text-align: center'>{title}</h2>", | |
| unsafe_allow_html=True, | |
| ) | |
| st.markdown( | |
| '<hr style="background-color: #ff4b4b; margin-top: 0;' | |
| ' margin-bottom: 0; height: 3px; border: none; border-radius: 3px;">', | |
| unsafe_allow_html=True, | |
| ) | |
| def forgot_password_form() -> None: | |
| """Creates a Streamlit widget to reset a user's password. | |
| The function prompts the user to enter their email address, and | |
| generates a password reset link for the corresponding user. If the | |
| user has not verified their email address, a new verification link | |
| is generated and sent to the user. | |
| """ | |
| # Get the email and password from the user | |
| email = st.text_input("Email", key="email_password") | |
| # Attempt to log the user in | |
| if not st.button("Reset password"): | |
| return None | |
| user = auth.get_user_by_email(email) | |
| auth.generate_password_reset_link(user.email) | |
| if user.email_verified: | |
| return st.success(f"Password reset link has been sent to {email}", icon="✅") | |
| st.warning(f"{email} is not verified. Resending the verification link.") | |
| return auth.generate_email_verification_link(user.email) | |
| def register_user_form(preauthorized: Union[str, Sequence[str], None]) -> None: | |
| """Creates a Streamlit widget for user registration. | |
| The function prompts user to enter all the required information for registration through | |
| Firebase. Password strength is validated using entropy bits (the power of the password | |
| alphabet). Upon registration, a validation link is sent to the user's email address. | |
| Arguments: | |
| preauthorized (Union[str, Sequence[str], None]): An optional domain or a list of | |
| domains which are authorized to register. | |
| """ | |
| with st.form(key="register_form"): | |
| email, name, password, confirm_password, register_button = ( | |
| st.text_input("E-mail"), | |
| st.text_input("Name"), | |
| st.text_input("Password", type="password"), | |
| st.text_input("Confirm password", type="password"), | |
| st.form_submit_button(label="Submit"), | |
| ) | |
| if not register_button: | |
| return None | |
| # Below are some checks to ensure proper and secure registration | |
| if password != confirm_password: | |
| return st.error("Passwords do not match", icon="🚨") | |
| if not name: | |
| return st.error("Please enter your name", icon="🚨") | |
| if preauthorized and not email.endswith(preauthorized): | |
| return st.error("Domain not allowed", icon="🚨") | |
| try: | |
| # Check that the email address is valid. | |
| validate_email(email, check_deliverability=True) | |
| except EmailNotValidError as e: | |
| # Email is not valid. | |
| # The exception message is human-readable. | |
| return st.error(e, icon="🚨") | |
| # Need a password that has minimum 66 entropy bits (the power of its alphabet) | |
| # I multiply this number by 150 (100 * 1.5) to display password strength with st.progress | |
| strength = round(PasswordPolicy().password(password).strength() * 150) | |
| if strength < 100: | |
| st.progress(strength) | |
| return st.warning( | |
| "Password is too weak. Please choose a stronger password.", icon="⚠️" | |
| ) | |
| auth.create_user(email=email, password=password) | |
| st.success( | |
| "Your account has been created successfully. To complete the registration process, " | |
| "please verify your email address by clicking on the link we have sent to your inbox.", | |
| icon="✅", | |
| ) | |
| return st.balloons() | |
| def update_password_form() -> None: | |
| """Creates a Streamlit widget to update a user's password. | |
| The function prompts the user to enter a new password, and updates | |
| the corresponding user's password with the new value. | |
| """ | |
| # Get the email and password from the user | |
| new_password = st.text_input("New password", key="new_password") | |
| # Attempt to log the user in | |
| if not st.button("Update password"): | |
| return None | |
| user = auth.get_user_by_email(st.session_state["username"]) | |
| auth.update_user(user.uid, password=new_password) | |
| return st.success("Successfully updated user password.", icon="✅") | |
| def update_display_name_form() -> None: | |
| """Creates a Streamlit widget to update a user's display name. | |
| The function prompts the user to enter a new display name, and | |
| updates the corresponding user's display name with the new value. | |
| """ | |
| # Get the email and password from the user | |
| new_name = st.text_input("New name", key="new name") | |
| # Attempt to log the user in | |
| if not st.button("Update name"): | |
| return None | |
| user = auth.get_user_by_email(st.session_state["username"]) | |
| auth.update_user(user.display_name, display_name=new_name) | |
| return st.success("Successfully updated user display name.", icon="✅") | |
| def token_decode(token: str) -> Optional[Dict[str, Any]]: | |
| """Decodes a JSON Web Token (JWT) and returns the resulting dictionary. | |
| The function decodes the provided JWT using the secret key specified in the | |
| application's secrets. The decoded dictionary contains information required for | |
| passwordless reauthentication. | |
| Parameters | |
| ---------- | |
| token : str | |
| The JWT to decode. | |
| Returns | |
| ------- | |
| Dict[str, Any] or None | |
| The decoded dictionary, or None if an error occurred during decoding. | |
| """ | |
| with suppress(Exception): | |
| return jwt.decode(token, st.secrets["COOKIE_KEY"], algorithms=["HS256"]) | |
| def token_encode(exp_date: datetime) -> str: | |
| """Encodes a JSON Web Token (JWT) containing user session data for passwordless | |
| reauthentication. | |
| Parameters | |
| ---------- | |
| exp_date : datetime | |
| The expiration date of the JWT. | |
| Returns | |
| ------- | |
| str | |
| The encoded JWT cookie string for reauthentication. | |
| Notes | |
| ----- | |
| The JWT contains the user's name, username, and the expiration date of the JWT in | |
| timestamp format. The `st.secrets["COOKIE_KEY"]` value is used to sign the JWT with | |
| the HS256 algorithm. | |
| """ | |
| return jwt.encode( | |
| { | |
| "name": st.session_state["name"], | |
| "username": st.session_state["username"], | |
| "exp_date": exp_date.timestamp(), | |
| }, | |
| st.secrets["COOKIE_KEY"], | |
| algorithm="HS256", | |
| ) | |
| def check_cookie_is_valid(cookie_manager: stx.CookieManager, cookie_name: str) -> bool: | |
| """Check if the reauthentication cookie is valid and update the session state if it | |
| is valid. | |
| Parameters | |
| ---------- | |
| cookie_manager : stx.CookieManager | |
| A JWT cookie manager instance for Streamlit | |
| cookie_name : str | |
| The name of the reauthentication cookie. | |
| Returns | |
| ------- | |
| bool | |
| True if the cookie is valid and the session state is updated successfully; False otherwise. | |
| Notes | |
| ----- | |
| This function checks if the reauthentication cookie specified by `cookie_name` is present in | |
| the cookies stored by `cookie_manager`, and if it is valid, meaning that it is not expired, and | |
| that it contains the required fields "name" and "username". If the cookie is valid, this | |
| function updates the session state of the Streamlit app, setting the "name", "username", and | |
| "authentication_status" keys accordingly. If the cookie is not valid or cannot be decoded, the | |
| function does not modify the session state and the user does not get authenticated. | |
| """ | |
| token = cookie_manager.get(cookie_name) | |
| if token is None: | |
| return False | |
| token = token_decode(token) | |
| if ( | |
| token is not False | |
| and not st.session_state["logout"] | |
| and token["exp_date"] > datetime.utcnow().timestamp() | |
| and "name" in token | |
| and "username" in token | |
| ): | |
| st.session_state["name"] = token["name"] | |
| st.session_state["username"] = token["username"] | |
| st.session_state["authentication_status"] = True | |
| return True | |
| return False | |
| def login_form( | |
| cookie_manager: stx.CookieManager, cookie_name: str, cookie_expiry_days: int = 30 | |
| ) -> None: | |
| """Creates a login widget using Firebase REST API and a cookie manager. | |
| Arguments | |
| --------- | |
| - cookie_manager: a JWT cookie manager instance for Streamlit | |
| - cookie_name: a name of the reauthentication cookie | |
| - cookie_expiry_days: (optional) an integer representing the number of days until the | |
| cookie expires | |
| Notes | |
| ----- | |
| If the user has already been authenticated, this function does nothing. Otherwise, it displays | |
| a login form which prompts the user to enter their email and password. If the login credentials | |
| are valid and the user's email address has been verified, the user is authenticated and a | |
| reauthentication cookie is created with the specified expiration date. | |
| If the user's email address has not been verified or if there is an error with the | |
| authentication process, the user is not authenticated. | |
| """ | |
| if st.session_state["authentication_status"]: | |
| return None | |
| if check_cookie_is_valid(cookie_manager, cookie_name): | |
| return None | |
| with st.form("Login"): | |
| email = st.text_input("Email") | |
| st.session_state["username"] = email | |
| password = st.text_input("Password", type="password") | |
| if not st.form_submit_button("Login"): | |
| return None | |
| # Authenticate the user with Firebase REST API | |
| url = ( | |
| f"https://identitytoolkit.googleapis.com/v1/" | |
| f"accounts:signInWithPassword?key={st.secrets['FIREBASE_API_KEY']}" | |
| ) | |
| payload = { | |
| "email": email, | |
| "password": password, | |
| "returnSecureToken": True, | |
| "emailVerified": True, # Require email verification | |
| } | |
| response = requests.post(url, data=json.dumps(payload), timeout=10).json() | |
| if "idToken" not in response: | |
| return st.error("Invalid e-mail or password.", icon="🚨") | |
| try: | |
| decoded_token = auth.verify_id_token(response["idToken"]) | |
| user = auth.get_user(decoded_token["uid"]) | |
| if not user.email_verified: | |
| return st.error("Please verify your e-mail address.", icon="🚨") | |
| # At last, authenticate the user | |
| st.session_state["name"] = user.display_name | |
| exp_date = datetime.utcnow() + timedelta(days=cookie_expiry_days) | |
| cookie_manager.set( | |
| cookie_name, | |
| token_encode(exp_date), | |
| expires_at=exp_date, | |
| ) | |
| st.session_state["authentication_status"] = True | |
| except Exception as e: | |
| st.error(e) | |
| return None | |
| def login_panel(cookie_manager: stx.CookieManager, cookie_name: str) -> None: | |
| """Creates a side panel for logged in users, preventing the login menu from | |
| appearing. | |
| Arguments | |
| --------- | |
| - cookie_manager: a JWT cookie manager instance for Streamlit | |
| - cookie_name: a name of the reauthentication cookie | |
| Notes | |
| ----- | |
| If the user is logged in, this function displays two tabs for resetting the user's password | |
| and updating their display name. | |
| If the user clicks the "Logout" button, the reauthentication cookie and user-related information | |
| from the session state is deleted, and the user is logged out. | |
| """ | |
| if st.button("Logout"): | |
| cookie_manager.delete(cookie_name) | |
| st.session_state["logout"] = True | |
| st.session_state["name"] = None | |
| st.session_state["username"] = None | |
| st.session_state["authentication_status"] = None | |
| return None | |
| st.write(f"Welcome, *{st.session_state['name']}*!") | |
| tab1, tab2 = st.tabs(["Reset password", "Update user details"]) | |
| try: | |
| with tab1: | |
| update_password_form() | |
| with tab2: | |
| update_display_name_form() | |
| except Exception as e: | |
| st.error(e) | |
| return None | |
| def not_logged_in(preauthorized: Union[str, Sequence[str], None]) -> bool: | |
| """Creates a tab panel for unauthenticated, preventing the user control sidebar and | |
| the rest of the script from appearing until the user logs in. | |
| Arguments | |
| --- | |
| 'preauthorized' - an optional domain or a list of domains which are authorized to register | |
| Returns | |
| ------- | |
| Authentication status boolean. | |
| Notes | |
| ----- | |
| The function first sets the authentication status flag to True and pre-populates missing | |
| session state arguments for the first run. If the user is already authenticated, the login | |
| panel function is called to create a side panel for logged-in users. If the function call | |
| does not update the authentication status because the username/password does not exist in | |
| the Firebase database, the functions halts. If the user is not logged in, tab panel created | |
| for the Login, Register, and Forgot password forms, and the rest of the script does not get | |
| executed until the user logs in. | |
| """ | |
| early_return = True | |
| # In case of a first run, pre-populate missing session state arguments | |
| cookie_manager, cookie_name = stx.CookieManager(), "login_cookie" | |
| for key in {"name", "authentication_status", "username", "logout"}.difference( | |
| set(st.session_state) | |
| ): | |
| st.session_state[key] = None | |
| # Give a loading message when the website page restarts | |
| with st.spinner("Loading..."): | |
| while "authentication_status" not in st.session_state: | |
| time.sleep(0.1) | |
| if st.session_state["authentication_status"]: | |
| with st.sidebar: | |
| login_panel(cookie_manager, cookie_name) | |
| return not early_return | |
| if st.session_state["authentication_status"] is False: | |
| st.error("Username/password is incorrect", icon="🚨") | |
| tab1, tab2, tab3 = st.tabs(["Login", "Register", "Forgot password"]) | |
| with tab1: | |
| login_form(cookie_manager, cookie_name) | |
| try: | |
| with tab2: | |
| register_user_form(preauthorized) | |
| with tab3: | |
| forgot_password_form() | |
| except Exception as e: | |
| st.error(e) | |
| return early_return | |
| def app() -> None: | |
| """This is a part of a Streamlit app which is only visible if the user is logged in.""" | |
| st.subheader("Yay!!!") | |
| st.write("You are logged in!") | |
| st.write("Hello :sunglasses:") | |
| def main() -> None: | |
| """Launches a Streamlit example interface. | |
| The interface supports authentication through Firebase REST API and JSON Web Token (JWT) | |
| cookies. To use the portal, the user must be registered, optionally only with a preauthorized | |
| e-mail domain. The Firebase REST API and JWT cookies are used for authentication. If the user | |
| is not logged in, no content other than the login form gets shown. | |
| """ | |
| st.set_page_config(page_title=TITLE, layout="wide") | |
| # noinspection PyProtectedMember | |
| if not firebase_admin._apps: | |
| cred = firebase_admin.credentials.Certificate("firebase_auth_token.json") | |
| firebase_admin.initialize_app(cred) | |
| pretty_title(TITLE) | |
| if not_logged_in(preauthorized="gmail.com"): | |
| return | |
| app() | |
| # Run the Streamlit app | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment