import os
import streamlit as st
from dotenv import load_dotenv
import requests
import base64
import json
load_dotenv()
COGNITO_DOMAIN = os.environ.get("COGNITO_DOMAIN")
CLIENT_ID = os.environ.get("CLIENT_ID")
CLIENT_SECRET = os.environ.get("CLIENT_SECRET")
APP_URI = os.environ.get("APP_URI")
def init_state():
if "auth_code" not in st.session_state:
st.session_state["auth_code"] = ""
if "authenticated" not in st.session_state:
st.session_state["authenticated"] = False
if "user_cognito_groups" not in st.session_state:
st.session_state["user_cognito_groups"] = []
# Get the authorization code after the user has logged in
def get_auth_code():
auth_query_params = st.experimental_get_query_params()
try:
auth_code = dict(auth_query_params)["code"][0]
except (KeyError, TypeError):
auth_code = ""
return auth_code
# Set the authorization code after the user has logged in
def set_auth_code():
init_state()
auth_code = get_auth_code()
st.session_state["auth_code"] = auth_code
# Get the access token from the authorization code
def get_user_tokens(auth_code):
# Variables to make a post request
token_url = f"{COGNITO_DOMAIN}/oauth2/token"
client_secret_string = f"{CLIENT_ID}:{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": CLIENT_ID,
"code": auth_code,
"redirect_uri": 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
# Use access token to retrieve user info
def get_user_info(access_token):
userinfo_url = f"{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()
# Decode access token to JWT to get user's cognito groups
def pad_base64(data):
missing_padding = len(data) % 4
if missing_padding != 0:
data += "=" * (4 - missing_padding)
return data
def get_user_cognito_groups(id_token):
user_cognito_groups = []
if id_token != "":
header, payload, signature = id_token.split(".")
printable_payload = base64.urlsafe_b64decode(pad_base64(payload))
payload_dict = json.loads(printable_payload)
try:
user_cognito_groups = list(dict(payload_dict)["cognito:groups"])
except (KeyError, TypeError):
pass
return user_cognito_groups
# Set streamlit state variables
def set_st_state_vars():
init_state()
auth_code = get_auth_code()
access_token, id_token = get_user_tokens(auth_code)
user_cognito_groups = get_user_cognito_groups(id_token)
if access_token != "":
st.session_state["auth_code"] = auth_code
st.session_state["authenticated"] = True
st.session_state["user_cognito_groups"] = user_cognito_groups
# Login/ Logout HTML components
login_link = f"{COGNITO_DOMAIN}/login?client_id={CLIENT_ID}&response_type=code&scope=email+openid&redirect_uri={APP_URI}"
logout_link = f"{COGNITO_DOMAIN}/logout?client_id={CLIENT_ID}&logout_uri={APP_URI}"
html_css_login = """
"""
html_button_login = (
html_css_login
+ f"Log In"
)
html_button_logout = (
html_css_login
+ f"Log Out"
)
def button_login():
return st.sidebar.markdown(f"{html_button_login}", unsafe_allow_html=True)
def button_logout():
return st.sidebar.markdown(f"{html_button_logout}", unsafe_allow_html=True)