<s> from boto . connection import AWSQueryConnection <EOL> from boto . provider import Provider , NO_CREDENTIALS_PROVIDED <EOL> from boto . regioninfo import RegionInfo <EOL> from boto . sts . credentials import Credentials , FederationToken , AssumedRole <EOL> from boto . sts . credentials import DecodeAuthorizationMessage <EOL> import boto <EOL> import boto . utils <EOL> import datetime <EOL> import threading <EOL> _session_token_cache = { } <EOL> class STSConnection ( AWSQueryConnection ) : <EOL> """<STR_LIT>""" <EOL> DefaultRegionName = '<STR_LIT>' <EOL> DefaultRegionEndpoint = '<STR_LIT>' <EOL> APIVersion = '<STR_LIT>' <EOL> def __init__ ( self , aws_access_key_id = None , aws_secret_access_key = None , <EOL> is_secure = True , port = None , proxy = None , proxy_port = None , <EOL> proxy_user = None , proxy_pass = None , debug = <NUM_LIT:0> , <EOL> https_connection_factory = None , region = None , path = '<STR_LIT:/>' , <EOL> converter = None , validate_certs = True , anon = False , <EOL> security_token = None , profile_name = None ) : <EOL> """<STR_LIT>""" <EOL> if not region : <EOL> region = RegionInfo ( self , self . DefaultRegionName , <EOL> self . DefaultRegionEndpoint , <EOL> connection_cls = STSConnection ) <EOL> self . region = region <EOL> self . anon = anon <EOL> self . _mutex = threading . Semaphore ( ) <EOL> provider = '<STR_LIT>' <EOL> if self . anon : <EOL> provider = Provider ( '<STR_LIT>' , NO_CREDENTIALS_PROVIDED , <EOL> NO_CREDENTIALS_PROVIDED , <EOL> NO_CREDENTIALS_PROVIDED ) <EOL> super ( STSConnection , self ) . __init__ ( aws_access_key_id , <EOL> aws_secret_access_key , <EOL> is_secure , port , proxy , proxy_port , <EOL> proxy_user , proxy_pass , <EOL> self . region . endpoint , debug , <EOL> https_connection_factory , path , <EOL> validate_certs = validate_certs , <EOL> security_token = security_token , <EOL> profile_name = profile_name , <EOL> provider = provider ) <EOL> def _required_auth_capability ( self ) : <EOL> if self . anon : <EOL> return [ '<STR_LIT>' ] <EOL> else : <EOL> return [ '<STR_LIT>' ] <EOL> def _check_token_cache ( self , token_key , duration = None , window_seconds = <NUM_LIT> ) : <EOL> token = _session_token_cache . get ( token_key , None ) <EOL> if token : <EOL> now = datetime . datetime . utcnow ( ) <EOL> expires = boto . utils . parse_ts ( token . expiration ) <EOL> delta = expires - now <EOL> if delta < datetime . timedelta ( seconds = window_seconds ) : <EOL> msg = '<STR_LIT>' % token_key <EOL> boto . log . debug ( msg ) <EOL> token = None <EOL> return token <EOL> def _get_session_token ( self , duration = None , <EOL> mfa_serial_number = None , mfa_token = None ) : <EOL> params = { } <EOL> if duration : <EOL> params [ '<STR_LIT>' ] = duration <EOL> if mfa_serial_number : <EOL> params [ '<STR_LIT>' ] = mfa_serial_number <EOL> if mfa_token : <EOL> params [ '<STR_LIT>' ] = mfa_token <EOL> return self . get_object ( '<STR_LIT>' , params , <EOL> Credentials , verb = '<STR_LIT:POST>' ) <EOL> def get_session_token ( self , duration = None , force_new = False , <EOL> mfa_serial_number = None , mfa_token = None ) : <EOL> """<STR_LIT>""" <EOL> token_key = '<STR_LIT>' % ( self . region . name , self . provider . access_key ) <EOL> token = self . _check_token_cache ( token_key , duration ) <EOL> if force_new or not token : <EOL> boto . log . debug ( '<STR_LIT>' % token_key ) <EOL> try : <EOL> self . _mutex . acquire ( ) <EOL> token = self . _get_session_token ( duration , <EOL> mfa_serial_number , <EOL> mfa_token ) <EOL> _session_token_cache [ token_key ] = token <EOL> finally : <EOL> self . _mutex . release ( ) <EOL> return token <EOL> def get_federation_token ( self , name , duration = None , policy = None ) : <EOL> """<STR_LIT>""" <EOL> params = { '<STR_LIT:Name>' : name } <EOL> if duration : <EOL> params [ '<STR_LIT>' ] = duration <EOL> if policy : <EOL> params [ '<STR_LIT>' ] = policy <EOL> return self . get_object ( '<STR_LIT>' , params , <EOL> FederationToken , verb = '<STR_LIT:POST>' ) <EOL> def assume_role ( self , role_arn , role_session_name , policy = None , <EOL> duration_seconds = None , external_id = None , <EOL> mfa_serial_number = None , <EOL> mfa_token = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : role_arn , <EOL> '<STR_LIT>' : role_session_name <EOL> } <EOL> if policy is not None : <EOL> params [ '<STR_LIT>' ] = policy <EOL> if duration_seconds is not None : <EOL> params [ '<STR_LIT>' ] = duration_seconds <EOL> if external_id is not None : <EOL> params [ '<STR_LIT>' ] = external_id <EOL> if mfa_serial_number is not None : <EOL> params [ '<STR_LIT>' ] = mfa_serial_number <EOL> if mfa_token is not None : <EOL> params [ '<STR_LIT>' ] = mfa_token <EOL> return self . get_object ( '<STR_LIT>' , params , AssumedRole , verb = '<STR_LIT:POST>' ) <EOL> def assume_role_with_saml ( self , role_arn , principal_arn , saml_assertion , <EOL> policy = None , duration_seconds = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : role_arn , <EOL> '<STR_LIT>' : principal_arn , <EOL> '<STR_LIT>' : saml_assertion , <EOL> } <EOL> if policy is not None : <EOL> params [ '<STR_LIT>' ] = policy <EOL> if duration_seconds is not None : <EOL> params [ '<STR_LIT>' ] = duration_seconds <EOL> return self . get_object ( '<STR_LIT>' , params , AssumedRole , <EOL> verb = '<STR_LIT:POST>' ) <EOL> def assume_role_with_web_identity ( self , role_arn , role_session_name , <EOL> web_identity_token , provider_id = None , <EOL> policy = None , duration_seconds = None ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : role_arn , <EOL> '<STR_LIT>' : role_session_name , <EOL> '<STR_LIT>' : web_identity_token , <EOL> } <EOL> if provider_id is not None : <EOL> params [ '<STR_LIT>' ] = provider_id <EOL> if policy is not None : <EOL> params [ '<STR_LIT>' ] = policy <EOL> if duration_seconds is not None : <EOL> params [ '<STR_LIT>' ] = duration_seconds <EOL> return self . get_object ( <EOL> '<STR_LIT>' , <EOL> params , <EOL> AssumedRole , <EOL> verb = '<STR_LIT:POST>' <EOL> ) <EOL> def decode_authorization_message ( self , encoded_message ) : <EOL> """<STR_LIT>""" <EOL> params = { <EOL> '<STR_LIT>' : encoded_message , <EOL> } <EOL> return self . get_object ( <EOL> '<STR_LIT>' , <EOL> params , <EOL> DecodeAuthorizationMessage , <EOL> verb = '<STR_LIT:POST>' <EOL> ) </s>
