"""Switchboard helpers.""" import time from typing import Dict import boto3 from ddex_ingester_common.constants.switchboard import JWT_EXPIRATION_VALUE import jwt def generate_switchboard_jwt( swb_role_arn: str, swb_secret_arn: str, application_name: str) -> str: """Generate a Switchboard JWT.""" credentials = generate_switchboard_credentials(swb_role_arn, application_name) secret = get_secrets_manager_value(swb_secret_arn, credentials) token_payload = { 'systemName': 'ORCHARD', 'exp': time.time() + JWT_EXPIRATION_VALUE, 'iat': time.time(), 'secretId': secret['VersionId'], } return jwt.encode(token_payload, secret['SecretString']) def generate_switchboard_credentials(swb_role_arn: str, application_name: str) -> Dict: """Generate Switchboard AWS credentials.""" sts = boto3.client('sts') assumed_role = sts.assume_role( RoleArn=swb_role_arn, RoleSessionName=f'{application_name}-session' ) credentials = assumed_role['Credentials'] return { 'aws_access_key_id': credentials['AccessKeyId'], 'aws_secret_access_key': credentials['SecretAccessKey'], 'aws_session_token': credentials['SessionToken'], } def get_secrets_manager_value(arn: str, credentials: Dict) -> Dict: """Get Amazon Secrets Manager secret value. Args: arn (str): The ARN/name of the secrets manager secret credentials (dict): AWS credentials to use Returns: Secrets manager value. """ secretsmanager = boto3.client( 'secretsmanager', **credentials ) secret = secretsmanager.get_secret_value( SecretId=arn, ) return secret