import streamlit_authenticator as stauth import yaml from yaml.loader import SafeLoader import streamlit as st import boto3 from botocore.exceptions import ClientError def get_authenticator(): # with open('.streamlit/users.yaml') as file: # cred_config = yaml.load(file, Loader=SafeLoader) file_content = read_text_file_from_s3('chatbot-editais-auth', 'config.yaml') # Parse the YAML content safely cred_config = yaml.safe_load(file_content) # Pre-hashing all plain text passwords once # hash_credentials = stauth.Hasher.hash_passwords(config['credentials']) authenticator = stauth.Authenticate( cred_config['credentials'], cred_config['cookie']['name'], cred_config['cookie']['key'], cred_config['cookie']['expiry_days'] ) return authenticator def st_authenticate(authenticator: stauth.Authenticate): # authenticator = get_authenticator() try: authenticator.login( location='sidebar', fields=dict(Username="Usuário", Password="Senha") ) except Exception as e: st.error(e) authenticator.cookie_controller.delete_cookie() st.warning('Caso o erro persistir, tente recarregar a página.') st.stop() if st.session_state['authentication_status']: with st.sidebar: st.write(f'Usuário: {st.session_state["name"]}') authenticator.logout() elif st.session_state['authentication_status'] is False: with st.sidebar: st.error('Usuário/senha incorreto') st.stop() elif st.session_state['authentication_status'] is None: st.warning('Por favor, informe o seu usuário e senha no painel lateral') st.stop() def read_text_file_from_s3(bucket, key): """ Read a YAML file from an S3 bucket using provided AWS credentials. Args: bucket (str): Name of the S3 bucket key (str): Path to the YAML file in the bucket aws_access_key_id (str): AWS Access Key ID aws_secret_access_key (str): AWS Secret Access Key Returns: dict: Parsed YAML content """ try: # Create an S3 client s3_client = boto3.client('s3') # Download the file from S3 response = s3_client.get_object(Bucket=bucket, Key=key) # Read the file content file_content = response['Body'].read().decode('utf-8') return file_content except Exception as e: st.error(f"Error reading file from S3: {e}") return None def get_secret(): secret_name = "assistente-produtos-servicos" region_name = "us-east-1" # Create a Secrets Manager client session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name=region_name ) try: get_secret_value_response = client.get_secret_value( SecretId=secret_name ) except ClientError as e: # For a list of exceptions thrown, see # https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html raise e secret = get_secret_value_response['SecretString'] return secret