-
Notifications
You must be signed in to change notification settings - Fork 251
Open
Description
In the botocore folder located in the site packages of the ios replace the credenitals.py file's class of processprovider code with:
'''
class ProcessProvider(CredentialProvider):
METHOD = 'custom-process'
def __init__(self, profile_name, load_config, run_command=None):
self._profile_name = profile_name
self._load_config = load_config
self._loaded_config = None
self._run_command = run_command or self._default_run_command
def load(self):
credential_process = self._credential_process
if credential_process is None:
return
creds_dict = self._retrieve_credentials_using(credential_process)
if creds_dict.get('expiry_time') is not None:
return RefreshableCredentials.create_from_metadata(
creds_dict,
lambda: self._retrieve_credentials_using(credential_process),
self.METHOD,
)
return Credentials(
access_key=creds_dict['access_key'],
secret_key=creds_dict['secret_key'],
token=creds_dict.get('token'),
method=self.METHOD,
account_id=creds_dict.get('account_id'),
)
def _retrieve_credentials_using(self, credential_process):
process_list = compat_shell_split(credential_process)
result = self._run_command(process_list)
stdout = result['stdout']
stderr = result['stderr']
returncode = result['returncode']
if returncode != 0:
raise CredentialRetrievalError(
provider=self.METHOD, error_msg=stderr
)
parsed = botocore.compat.json.loads(stdout)
version = parsed.get('Version', '<Version key not provided>')
if version != 1:
raise CredentialRetrievalError(
provider=self.METHOD,
error_msg=(
f"Unsupported version '{version}' for credential process "
f"provider, supported versions: 1"
),
)
try:
return {
'access_key': parsed['AccessKeyId'],
'secret_key': parsed['SecretAccessKey'],
'token': parsed.get('SessionToken'),
'expiry_time': parsed.get('Expiration'),
'account_id': self._get_account_id(parsed),
}
except KeyError as e:
raise CredentialRetrievalError(
provider=self.METHOD,
error_msg=f"Missing required key in response: {e}",
)
@property
def _credential_process(self):
return self.profile_config.get('credential_process')
@property
def profile_config(self):
if self._loaded_config is None:
self._loaded_config = self._load_config()
profile_config = self._loaded_config.get('profiles', {}).get(
self._profile_name, {}
)
return profile_config
def _get_account_id(self, parsed):
account_id = parsed.get('AccountId')
return account_id or self.profile_config.get('aws_account_id')
def _default_run_command(self, process_list):
completed_process = subprocess.run(
process_list,
capture_output=True,
text=True,
check=False
)
return {
'stdout': completed_process.stdout,
'stderr': completed_process.stderr,
'returncode': completed_process.returncode
}
'''
And in the s3tranfer module, delete the last line(importing the basemanager from mutliprocessing) in the compat.py file
Finally, go to the auth.py file in botocore and at the top put, from collections import namedtuple, then go to the add)auth function around line 1010, and add this at the top of the function(replace access and sercet key with ur keys):
'''
Credentials = namedtuple('Credentials', ['access_key', 'secret_key', 'token', 'account_id'])
self.credentials = Credentials(
access_key='****',
secret_key='****',
token=None,
account_id=None
)
'''
Metadata
Metadata
Assignees
Labels
No labels