| 1178 | |
| 1179 | |
| 1180 | class EnvProvider(CredentialProvider): |
| 1181 | METHOD = 'env' |
| 1182 | CANONICAL_NAME = 'Environment' |
| 1183 | ACCESS_KEY = 'AWS_ACCESS_KEY_ID' |
| 1184 | SECRET_KEY = 'AWS_SECRET_ACCESS_KEY' |
| 1185 | # The token can come from either of these env var. |
| 1186 | # AWS_SESSION_TOKEN is what other AWS SDKs have standardized on. |
| 1187 | TOKENS = ['AWS_SECURITY_TOKEN', 'AWS_SESSION_TOKEN'] |
| 1188 | EXPIRY_TIME = 'AWS_CREDENTIAL_EXPIRATION' |
| 1189 | ACCOUNT_ID = 'AWS_ACCOUNT_ID' |
| 1190 | |
| 1191 | def __init__(self, environ=None, mapping=None): |
| 1192 | """ |
| 1193 | |
| 1194 | :param environ: The environment variables (defaults to |
| 1195 | ``os.environ`` if no value is provided). |
| 1196 | :param mapping: An optional mapping of variable names to |
| 1197 | environment variable names. Use this if you want to |
| 1198 | change the mapping of access_key->AWS_ACCESS_KEY_ID, etc. |
| 1199 | The dict can have up to 5 keys: |
| 1200 | * ``access_key`` |
| 1201 | * ``secret_key`` |
| 1202 | * ``token`` |
| 1203 | * ``expiry_time`` |
| 1204 | * ``account_id`` |
| 1205 | """ |
| 1206 | if environ is None: |
| 1207 | environ = os.environ |
| 1208 | self.environ = environ |
| 1209 | self._mapping = self._build_mapping(mapping) |
| 1210 | |
| 1211 | def _build_mapping(self, mapping): |
| 1212 | # Mapping of variable name to env var name. |
| 1213 | var_mapping = {} |
| 1214 | if mapping is None: |
| 1215 | # Use the class var default. |
| 1216 | var_mapping['access_key'] = self.ACCESS_KEY |
| 1217 | var_mapping['secret_key'] = self.SECRET_KEY |
| 1218 | var_mapping['token'] = self.TOKENS |
| 1219 | var_mapping['expiry_time'] = self.EXPIRY_TIME |
| 1220 | var_mapping['account_id'] = self.ACCOUNT_ID |
| 1221 | else: |
| 1222 | var_mapping['access_key'] = mapping.get( |
| 1223 | 'access_key', self.ACCESS_KEY |
| 1224 | ) |
| 1225 | var_mapping['secret_key'] = mapping.get( |
| 1226 | 'secret_key', self.SECRET_KEY |
| 1227 | ) |
| 1228 | var_mapping['token'] = mapping.get('token', self.TOKENS) |
| 1229 | if not isinstance(var_mapping['token'], list): |
| 1230 | var_mapping['token'] = [var_mapping['token']] |
| 1231 | var_mapping['expiry_time'] = mapping.get( |
| 1232 | 'expiry_time', self.EXPIRY_TIME |
| 1233 | ) |
| 1234 | var_mapping['account_id'] = mapping.get( |
| 1235 | 'account_id', self.ACCOUNT_ID |
| 1236 | ) |
| 1237 | return var_mapping |
no outgoing calls
no test coverage detected