1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
|
# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
"""This module contains the interface for controlling how configuration
is loaded.
"""
import copy
import logging
import os
from botocore import utils
from botocore.exceptions import InvalidConfigError
logger = logging.getLogger(__name__)
#: A default dictionary that maps the logical names for session variables
#: to the specific environment variables and configuration file names
#: that contain the values for these variables.
#: When creating a new Session object, you can pass in your own dictionary
#: to remap the logical names or to add new logical names. You can then
#: get the current value for these variables by using the
#: ``get_config_variable`` method of the :class:`botocore.session.Session`
#: class.
#: These form the keys of the dictionary. The values in the dictionary
#: are tuples of (<config_name>, <environment variable>, <default value>,
#: <conversion func>).
#: The conversion func is a function that takes the configuration value
#: as an argument and returns the converted value. If this value is
#: None, then the configuration value is returned unmodified. This
#: conversion function can be used to type convert config values to
#: values other than the default values of strings.
#: The ``profile`` and ``config_file`` variables should always have a
#: None value for the first entry in the tuple because it doesn't make
#: sense to look inside the config file for the location of the config
#: file or for the default profile to use.
#: The ``config_name`` is the name to look for in the configuration file,
#: the ``env var`` is the OS environment variable (``os.environ``) to
#: use, and ``default_value`` is the value to use if no value is otherwise
#: found.
#: NOTE: Fixing the spelling of this variable would be a breaking change.
#: Please leave as is.
BOTOCORE_DEFAUT_SESSION_VARIABLES = {
# logical: config_file, env_var, default_value, conversion_func
'profile': (None, ['AWS_DEFAULT_PROFILE', 'AWS_PROFILE'], None, None),
'region': ('region', 'AWS_DEFAULT_REGION', None, None),
'data_path': ('data_path', 'AWS_DATA_PATH', None, None),
'config_file': (None, 'AWS_CONFIG_FILE', '~/.aws/config', None),
'ca_bundle': ('ca_bundle', 'AWS_CA_BUNDLE', None, None),
'api_versions': ('api_versions', None, {}, None),
# This is the shared credentials file amongst sdks.
'credentials_file': (
None,
'AWS_SHARED_CREDENTIALS_FILE',
'~/.aws/credentials',
None,
),
# These variables only exist in the config file.
# This is the number of seconds until we time out a request to
# the instance metadata service.
'metadata_service_timeout': (
'metadata_service_timeout',
'AWS_METADATA_SERVICE_TIMEOUT',
1,
int,
),
# This is the number of request attempts we make until we give
# up trying to retrieve data from the instance metadata service.
'metadata_service_num_attempts': (
'metadata_service_num_attempts',
'AWS_METADATA_SERVICE_NUM_ATTEMPTS',
1,
int,
),
'ec2_metadata_service_endpoint': (
'ec2_metadata_service_endpoint',
'AWS_EC2_METADATA_SERVICE_ENDPOINT',
None,
None,
),
'ec2_metadata_service_endpoint_mode': (
'ec2_metadata_service_endpoint_mode',
'AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE',
None,
None,
),
'ec2_metadata_v1_disabled': (
'ec2_metadata_v1_disabled',
'AWS_EC2_METADATA_V1_DISABLED',
False,
utils.ensure_boolean,
),
'imds_use_ipv6': (
'imds_use_ipv6',
'AWS_IMDS_USE_IPV6',
False,
utils.ensure_boolean,
),
'use_dualstack_endpoint': (
'use_dualstack_endpoint',
'AWS_USE_DUALSTACK_ENDPOINT',
None,
utils.ensure_boolean,
),
'use_fips_endpoint': (
'use_fips_endpoint',
'AWS_USE_FIPS_ENDPOINT',
None,
utils.ensure_boolean,
),
'ignore_configured_endpoint_urls': (
'ignore_configured_endpoint_urls',
'AWS_IGNORE_CONFIGURED_ENDPOINT_URLS',
None,
utils.ensure_boolean,
),
'parameter_validation': ('parameter_validation', None, True, None),
# Client side monitoring configurations.
# Note: These configurations are considered internal to botocore.
# Do not use them until publicly documented.
'csm_enabled': (
'csm_enabled',
'AWS_CSM_ENABLED',
False,
utils.ensure_boolean,
),
'csm_host': ('csm_host', 'AWS_CSM_HOST', '127.0.0.1', None),
'csm_port': ('csm_port', 'AWS_CSM_PORT', 31000, int),
'csm_client_id': ('csm_client_id', 'AWS_CSM_CLIENT_ID', '', None),
# Endpoint discovery configuration
'endpoint_discovery_enabled': (
'endpoint_discovery_enabled',
'AWS_ENDPOINT_DISCOVERY_ENABLED',
'auto',
None,
),
'retry_mode': ('retry_mode', 'AWS_RETRY_MODE', 'legacy', None),
'defaults_mode': ('defaults_mode', 'AWS_DEFAULTS_MODE', 'legacy', None),
# We can't have a default here for v1 because we need to defer to
# whatever the defaults are in _retry.json.
'max_attempts': ('max_attempts', 'AWS_MAX_ATTEMPTS', None, int),
'user_agent_appid': ('sdk_ua_app_id', 'AWS_SDK_UA_APP_ID', None, None),
'request_min_compression_size_bytes': (
'request_min_compression_size_bytes',
'AWS_REQUEST_MIN_COMPRESSION_SIZE_BYTES',
10240,
None,
),
'disable_request_compression': (
'disable_request_compression',
'AWS_DISABLE_REQUEST_COMPRESSION',
False,
utils.ensure_boolean,
),
'sigv4a_signing_region_set': (
'sigv4a_signing_region_set',
'AWS_SIGV4A_SIGNING_REGION_SET',
None,
None,
),
'request_checksum_calculation': (
'request_checksum_calculation',
'AWS_REQUEST_CHECKSUM_CALCULATION',
"when_supported",
None,
),
'response_checksum_validation': (
'response_checksum_validation',
'AWS_RESPONSE_CHECKSUM_VALIDATION',
"when_supported",
None,
),
'account_id_endpoint_mode': (
'account_id_endpoint_mode',
'AWS_ACCOUNT_ID_ENDPOINT_MODE',
'preferred',
None,
),
'disable_host_prefix_injection': (
'disable_host_prefix_injection',
'AWS_DISABLE_HOST_PREFIX_INJECTION',
None,
utils.ensure_boolean,
),
}
# Evaluate AWS_STS_REGIONAL_ENDPOINTS settings
try:
# This is not a public interface and is subject to abrupt breaking changes.
# Any usage is not advised or supported in external code bases.
from botocore.customizations.sts import (
sts_default_setting as _sts_default_setting,
)
except ImportError:
_sts_default_setting = 'legacy'
_STS_DEFAULT_SETTINGS = {
'sts_regional_endpoints': (
'sts_regional_endpoints',
'AWS_STS_REGIONAL_ENDPOINTS',
_sts_default_setting,
None,
),
}
BOTOCORE_DEFAUT_SESSION_VARIABLES.update(_STS_DEFAULT_SETTINGS)
# A mapping for the s3 specific configuration vars. These are the configuration
# vars that typically go in the s3 section of the config file. This mapping
# follows the same schema as the previous session variable mapping.
DEFAULT_S3_CONFIG_VARS = {
'addressing_style': (('s3', 'addressing_style'), None, None, None),
'use_accelerate_endpoint': (
('s3', 'use_accelerate_endpoint'),
None,
None,
utils.ensure_boolean,
),
'use_dualstack_endpoint': (
('s3', 'use_dualstack_endpoint'),
None,
None,
utils.ensure_boolean,
),
'payload_signing_enabled': (
('s3', 'payload_signing_enabled'),
None,
None,
utils.ensure_boolean,
),
'use_arn_region': (
['s3_use_arn_region', ('s3', 'use_arn_region')],
'AWS_S3_USE_ARN_REGION',
None,
utils.ensure_boolean,
),
'us_east_1_regional_endpoint': (
[
's3_us_east_1_regional_endpoint',
('s3', 'us_east_1_regional_endpoint'),
],
'AWS_S3_US_EAST_1_REGIONAL_ENDPOINT',
None,
None,
),
's3_disable_multiregion_access_points': (
('s3', 's3_disable_multiregion_access_points'),
'AWS_S3_DISABLE_MULTIREGION_ACCESS_POINTS',
None,
utils.ensure_boolean,
),
}
# A mapping for the proxy specific configuration vars. These are
# used to configure how botocore interacts with proxy setups while
# sending requests.
DEFAULT_PROXIES_CONFIG_VARS = {
'proxy_ca_bundle': ('proxy_ca_bundle', None, None, None),
'proxy_client_cert': ('proxy_client_cert', None, None, None),
'proxy_use_forwarding_for_https': (
'proxy_use_forwarding_for_https',
None,
None,
utils.normalize_boolean,
),
}
def create_botocore_default_config_mapping(session):
chain_builder = ConfigChainFactory(session=session)
config_mapping = _create_config_chain_mapping(
chain_builder, BOTOCORE_DEFAUT_SESSION_VARIABLES
)
config_mapping['s3'] = SectionConfigProvider(
's3',
session,
_create_config_chain_mapping(chain_builder, DEFAULT_S3_CONFIG_VARS),
)
config_mapping['proxies_config'] = SectionConfigProvider(
'proxies_config',
session,
_create_config_chain_mapping(
chain_builder, DEFAULT_PROXIES_CONFIG_VARS
),
)
return config_mapping
def _create_config_chain_mapping(chain_builder, config_variables):
mapping = {}
for logical_name, config in config_variables.items():
mapping[logical_name] = chain_builder.create_config_chain(
instance_name=logical_name,
env_var_names=config[1],
config_property_names=config[0],
default=config[2],
conversion_func=config[3],
)
return mapping
class DefaultConfigResolver:
def __init__(self, default_config_data):
self._base_default_config = default_config_data['base']
self._modes = default_config_data['modes']
self._resolved_default_configurations = {}
def _resolve_default_values_by_mode(self, mode):
default_config = self._base_default_config.copy()
modifications = self._modes.get(mode)
for config_var in modifications:
default_value = default_config[config_var]
modification_dict = modifications[config_var]
modification = list(modification_dict.keys())[0]
modification_value = modification_dict[modification]
if modification == 'multiply':
default_value *= modification_value
elif modification == 'add':
default_value += modification_value
elif modification == 'override':
default_value = modification_value
default_config[config_var] = default_value
return default_config
def get_default_modes(self):
default_modes = ['legacy', 'auto']
default_modes.extend(self._modes.keys())
return default_modes
def get_default_config_values(self, mode):
if mode not in self._resolved_default_configurations:
defaults = self._resolve_default_values_by_mode(mode)
self._resolved_default_configurations[mode] = defaults
return self._resolved_default_configurations[mode]
class ConfigChainFactory:
"""Factory class to create our most common configuration chain case.
This is a convenience class to construct configuration chains that follow
our most common pattern. This is to prevent ordering them incorrectly,
and to make the config chain construction more readable.
"""
def __init__(self, session, environ=None):
"""Initialize a ConfigChainFactory.
:type session: :class:`botocore.session.Session`
:param session: This is the session that should be used to look up
values from the config file.
:type environ: dict
:param environ: A mapping to use for environment variables. If this
is not provided it will default to use os.environ.
"""
self._session = session
if environ is None:
environ = os.environ
self._environ = environ
def create_config_chain(
self,
instance_name=None,
env_var_names=None,
config_property_names=None,
default=None,
conversion_func=None,
):
"""Build a config chain following the standard botocore pattern.
In botocore most of our config chains follow the the precendence:
session_instance_variables, environment, config_file, default_value.
This is a convenience function for creating a chain that follow
that precendence.
:type instance_name: str
:param instance_name: This indicates what session instance variable
corresponds to this config value. If it is None it will not be
added to the chain.
:type env_var_names: str or list of str or None
:param env_var_names: One or more environment variable names to
search for this value. They are searched in order. If it is None
it will not be added to the chain.
:type config_property_names: str/tuple or list of str/tuple or None
:param config_property_names: One of more strings or tuples
representing the name of the key in the config file for this
config option. They are searched in order. If it is None it will
not be added to the chain.
:type default: Any
:param default: Any constant value to be returned.
:type conversion_func: None or callable
:param conversion_func: If this value is None then it has no effect on
the return type. Otherwise, it is treated as a function that will
conversion_func our provided type.
:rvalue: ConfigChain
:returns: A ConfigChain that resolves in the order env_var_names ->
config_property_name -> default. Any values that were none are
omitted form the chain.
"""
providers = []
if instance_name is not None:
providers.append(
InstanceVarProvider(
instance_var=instance_name, session=self._session
)
)
if env_var_names is not None:
providers.extend(self._get_env_providers(env_var_names))
if config_property_names is not None:
providers.extend(
self._get_scoped_config_providers(config_property_names)
)
if default is not None:
providers.append(ConstantProvider(value=default))
return ChainProvider(
providers=providers,
conversion_func=conversion_func,
)
def _get_env_providers(self, env_var_names):
env_var_providers = []
if not isinstance(env_var_names, list):
env_var_names = [env_var_names]
for env_var_name in env_var_names:
env_var_providers.append(
EnvironmentProvider(name=env_var_name, env=self._environ)
)
return env_var_providers
def _get_scoped_config_providers(self, config_property_names):
scoped_config_providers = []
if not isinstance(config_property_names, list):
config_property_names = [config_property_names]
for config_property_name in config_property_names:
scoped_config_providers.append(
ScopedConfigProvider(
config_var_name=config_property_name,
session=self._session,
)
)
return scoped_config_providers
class ConfigValueStore:
"""The ConfigValueStore object stores configuration values."""
def __init__(self, mapping=None):
"""Initialize a ConfigValueStore.
:type mapping: dict
:param mapping: The mapping parameter is a map of string to a subclass
of BaseProvider. When a config variable is asked for via the
get_config_variable method, the corresponding provider will be
invoked to load the value.
"""
self._overrides = {}
self._mapping = {}
if mapping is not None:
for logical_name, provider in mapping.items():
self.set_config_provider(logical_name, provider)
def __deepcopy__(self, memo):
config_store = ConfigValueStore(copy.deepcopy(self._mapping, memo))
for logical_name, override_value in self._overrides.items():
config_store.set_config_variable(logical_name, override_value)
return config_store
def __copy__(self):
config_store = ConfigValueStore(copy.copy(self._mapping))
for logical_name, override_value in self._overrides.items():
config_store.set_config_variable(logical_name, override_value)
return config_store
def get_config_variable(self, logical_name):
"""
Retrieve the value associated with the specified logical_name
from the corresponding provider. If no value is found None will
be returned.
:type logical_name: str
:param logical_name: The logical name of the session variable
you want to retrieve. This name will be mapped to the
appropriate environment variable name for this session as
well as the appropriate config file entry.
:returns: value of variable or None if not defined.
"""
if logical_name in self._overrides:
return self._overrides[logical_name]
if logical_name not in self._mapping:
return None
provider = self._mapping[logical_name]
return provider.provide()
def get_config_provider(self, logical_name):
"""
Retrieve the provider associated with the specified logical_name.
If no provider is found None will be returned.
:type logical_name: str
:param logical_name: The logical name of the session variable
you want to retrieve. This name will be mapped to the
appropriate environment variable name for this session as
well as the appropriate config file entry.
:returns: configuration provider or None if not defined.
"""
if (
logical_name in self._overrides
or logical_name not in self._mapping
):
return None
provider = self._mapping[logical_name]
return provider
def set_config_variable(self, logical_name, value):
"""Set a configuration variable to a specific value.
By using this method, you can override the normal lookup
process used in ``get_config_variable`` by explicitly setting
a value. Subsequent calls to ``get_config_variable`` will
use the ``value``. This gives you per-session specific
configuration values.
::
>>> # Assume logical name 'foo' maps to env var 'FOO'
>>> os.environ['FOO'] = 'myvalue'
>>> s.get_config_variable('foo')
'myvalue'
>>> s.set_config_variable('foo', 'othervalue')
>>> s.get_config_variable('foo')
'othervalue'
:type logical_name: str
:param logical_name: The logical name of the session variable
you want to set. These are the keys in ``SESSION_VARIABLES``.
:param value: The value to associate with the config variable.
"""
self._overrides[logical_name] = value
def clear_config_variable(self, logical_name):
"""Remove an override config variable from the session.
:type logical_name: str
:param logical_name: The name of the parameter to clear the override
value from.
"""
self._overrides.pop(logical_name, None)
def set_config_provider(self, logical_name, provider):
"""Set the provider for a config value.
This provides control over how a particular configuration value is
loaded. This replaces the provider for ``logical_name`` with the new
``provider``.
:type logical_name: str
:param logical_name: The name of the config value to change the config
provider for.
:type provider: :class:`botocore.configprovider.BaseProvider`
:param provider: The new provider that should be responsible for
providing a value for the config named ``logical_name``.
"""
self._mapping[logical_name] = provider
class SmartDefaultsConfigStoreFactory:
def __init__(self, default_config_resolver, imds_region_provider):
self._default_config_resolver = default_config_resolver
self._imds_region_provider = imds_region_provider
# Initializing _instance_metadata_region as None so we
# can fetch region in a lazy fashion only when needed.
self._instance_metadata_region = None
def merge_smart_defaults(self, config_store, mode, region_name):
if mode == 'auto':
mode = self.resolve_auto_mode(region_name)
default_configs = (
self._default_config_resolver.get_default_config_values(mode)
)
for config_var in default_configs:
config_value = default_configs[config_var]
method = getattr(self, f'_set_{config_var}', None)
if method:
method(config_store, config_value)
def resolve_auto_mode(self, region_name):
current_region = None
if os.environ.get('AWS_EXECUTION_ENV'):
default_region = os.environ.get('AWS_DEFAULT_REGION')
current_region = os.environ.get('AWS_REGION', default_region)
if not current_region:
if self._instance_metadata_region:
current_region = self._instance_metadata_region
else:
try:
current_region = self._imds_region_provider.provide()
self._instance_metadata_region = current_region
except Exception:
pass
if current_region:
if region_name == current_region:
return 'in-region'
else:
return 'cross-region'
return 'standard'
def _update_provider(self, config_store, variable, value):
original_provider = config_store.get_config_provider(variable)
default_provider = ConstantProvider(value)
if isinstance(original_provider, ChainProvider):
chain_provider_copy = copy.deepcopy(original_provider)
chain_provider_copy.set_default_provider(default_provider)
default_provider = chain_provider_copy
elif isinstance(original_provider, BaseProvider):
default_provider = ChainProvider(
providers=[original_provider, default_provider]
)
config_store.set_config_provider(variable, default_provider)
def _update_section_provider(
self, config_store, section_name, variable, value
):
section_provider_copy = copy.deepcopy(
config_store.get_config_provider(section_name)
)
section_provider_copy.set_default_provider(
variable, ConstantProvider(value)
)
config_store.set_config_provider(section_name, section_provider_copy)
def _set_retryMode(self, config_store, value):
self._update_provider(config_store, 'retry_mode', value)
def _set_stsRegionalEndpoints(self, config_store, value):
self._update_provider(config_store, 'sts_regional_endpoints', value)
def _set_s3UsEast1RegionalEndpoints(self, config_store, value):
self._update_section_provider(
config_store, 's3', 'us_east_1_regional_endpoint', value
)
def _set_connectTimeoutInMillis(self, config_store, value):
self._update_provider(config_store, 'connect_timeout', value / 1000)
class BaseProvider:
"""Base class for configuration value providers.
A configuration provider has some method of providing a configuration
value.
"""
def provide(self):
"""Provide a config value."""
raise NotImplementedError('provide')
class ChainProvider(BaseProvider):
"""This provider wraps one or more other providers.
Each provider in the chain is called, the first one returning a non-None
value is then returned.
"""
def __init__(self, providers=None, conversion_func=None):
"""Initalize a ChainProvider.
:type providers: list
:param providers: The initial list of providers to check for values
when invoked.
:type conversion_func: None or callable
:param conversion_func: If this value is None then it has no affect on
the return type. Otherwise, it is treated as a function that will
transform provided value.
"""
if providers is None:
providers = []
self._providers = providers
self._conversion_func = conversion_func
def __deepcopy__(self, memo):
return ChainProvider(
copy.deepcopy(self._providers, memo), self._conversion_func
)
def provide(self):
"""Provide the value from the first provider to return non-None.
Each provider in the chain has its provide method called. The first
one in the chain to return a non-None value is the returned from the
ChainProvider. When no non-None value is found, None is returned.
"""
for provider in self._providers:
value = provider.provide()
if value is not None:
return self._convert_type(value)
return None
def set_default_provider(self, default_provider):
if self._providers and isinstance(
self._providers[-1], ConstantProvider
):
self._providers[-1] = default_provider
else:
self._providers.append(default_provider)
num_of_constants = sum(
isinstance(provider, ConstantProvider)
for provider in self._providers
)
if num_of_constants > 1:
logger.info(
'ChainProvider object contains multiple '
'instances of ConstantProvider objects'
)
def _convert_type(self, value):
if self._conversion_func is not None:
return self._conversion_func(value)
return value
def __repr__(self):
return '[{}]'.format(', '.join([str(p) for p in self._providers]))
class InstanceVarProvider(BaseProvider):
"""This class loads config values from the session instance vars."""
def __init__(self, instance_var, session):
"""Initialize InstanceVarProvider.
:type instance_var: str
:param instance_var: The instance variable to load from the session.
:type session: :class:`botocore.session.Session`
:param session: The botocore session to get the loaded configuration
file variables from.
"""
self._instance_var = instance_var
self._session = session
def __deepcopy__(self, memo):
return InstanceVarProvider(
copy.deepcopy(self._instance_var, memo), self._session
)
def provide(self):
"""Provide a config value from the session instance vars."""
instance_vars = self._session.instance_variables()
value = instance_vars.get(self._instance_var)
return value
def __repr__(self):
return f'InstanceVarProvider(instance_var={self._instance_var}, session={self._session})'
class ScopedConfigProvider(BaseProvider):
def __init__(self, config_var_name, session):
"""Initialize ScopedConfigProvider.
:type config_var_name: str or tuple
:param config_var_name: The name of the config variable to load from
the configuration file. If the value is a tuple, it must only
consist of two items, where the first item represents the section
and the second item represents the config var name in the section.
:type session: :class:`botocore.session.Session`
:param session: The botocore session to get the loaded configuration
file variables from.
"""
self._config_var_name = config_var_name
self._session = session
def __deepcopy__(self, memo):
return ScopedConfigProvider(
copy.deepcopy(self._config_var_name, memo), self._session
)
def provide(self):
"""Provide a value from a config file property."""
scoped_config = self._session.get_scoped_config()
if isinstance(self._config_var_name, tuple):
section_config = scoped_config.get(self._config_var_name[0])
if not isinstance(section_config, dict):
return None
return section_config.get(self._config_var_name[1])
return scoped_config.get(self._config_var_name)
def __repr__(self):
return f'ScopedConfigProvider(config_var_name={self._config_var_name}, session={self._session})'
class EnvironmentProvider(BaseProvider):
"""This class loads config values from environment variables."""
def __init__(self, name, env):
"""Initialize with the keys in the dictionary to check.
:type name: str
:param name: The key with that name will be loaded and returned.
:type env: dict
:param env: Environment variables dictionary to get variables from.
"""
self._name = name
self._env = env
def __deepcopy__(self, memo):
return EnvironmentProvider(
copy.deepcopy(self._name, memo), copy.deepcopy(self._env, memo)
)
def provide(self):
"""Provide a config value from a source dictionary."""
if self._name in self._env:
return self._env[self._name]
return None
def __repr__(self):
return f'EnvironmentProvider(name={self._name}, env={self._env})'
class SectionConfigProvider(BaseProvider):
"""Provides a dictionary from a section in the scoped config
This is useful for retrieving scoped config variables (i.e. s3) that have
their own set of config variables and resolving logic.
"""
def __init__(self, section_name, session, override_providers=None):
self._section_name = section_name
self._session = session
self._scoped_config_provider = ScopedConfigProvider(
self._section_name, self._session
)
self._override_providers = override_providers
if self._override_providers is None:
self._override_providers = {}
def __deepcopy__(self, memo):
return SectionConfigProvider(
copy.deepcopy(self._section_name, memo),
self._session,
copy.deepcopy(self._override_providers, memo),
)
def provide(self):
section_config = self._scoped_config_provider.provide()
if section_config and not isinstance(section_config, dict):
logger.debug(
"The %s config key is not a dictionary type, "
"ignoring its value of: %s",
self._section_name,
section_config,
)
return None
for section_config_var, provider in self._override_providers.items():
provider_val = provider.provide()
if provider_val is not None:
if section_config is None:
section_config = {}
section_config[section_config_var] = provider_val
return section_config
def set_default_provider(self, key, default_provider):
provider = self._override_providers.get(key)
if isinstance(provider, ChainProvider):
provider.set_default_provider(default_provider)
return
elif isinstance(provider, BaseProvider):
default_provider = ChainProvider(
providers=[provider, default_provider]
)
self._override_providers[key] = default_provider
def __repr__(self):
return (
f'SectionConfigProvider(section_name={self._section_name}, '
f'session={self._session}, '
f'override_providers={self._override_providers})'
)
class ConstantProvider(BaseProvider):
"""This provider provides a constant value."""
def __init__(self, value):
self._value = value
def __deepcopy__(self, memo):
return ConstantProvider(copy.deepcopy(self._value, memo))
def provide(self):
"""Provide the constant value given during initialization."""
return self._value
def __repr__(self):
return f'ConstantProvider(value={self._value})'
class ConfiguredEndpointProvider(BaseProvider):
"""Lookup an endpoint URL from environment variable or shared config file.
NOTE: This class is considered private and is subject to abrupt breaking
changes or removal without prior announcement. Please do not use it
directly.
"""
_ENDPOINT_URL_LOOKUP_ORDER = [
'environment_service',
'environment_global',
'config_service',
'config_global',
]
def __init__(
self,
full_config,
scoped_config,
client_name,
environ=None,
):
"""Initialize a ConfiguredEndpointProviderChain.
:type full_config: dict
:param full_config: This is the dict representing the full
configuration file.
:type scoped_config: dict
:param scoped_config: This is the dict representing the configuration
for the current profile for the session.
:type client_name: str
:param client_name: The name used to instantiate a client using
botocore.session.Session.create_client.
:type environ: dict
:param environ: A mapping to use for environment variables. If this
is not provided it will default to use os.environ.
"""
self._full_config = full_config
self._scoped_config = scoped_config
self._client_name = client_name
self._transformed_service_id = self._get_snake_case_service_id(
self._client_name
)
if environ is None:
environ = os.environ
self._environ = environ
def provide(self):
"""Lookup the configured endpoint URL.
The order is:
1. The value provided by a service-specific environment variable.
2. The value provided by the global endpoint environment variable
(AWS_ENDPOINT_URL).
3. The value provided by a service-specific parameter from a services
definition section in the shared configuration file.
4. The value provided by the global parameter from a services
definition section in the shared configuration file.
"""
for location in self._ENDPOINT_URL_LOOKUP_ORDER:
logger.debug(
'Looking for endpoint for %s via: %s',
self._client_name,
location,
)
endpoint_url = getattr(self, f'_get_endpoint_url_{location}')()
if endpoint_url:
logger.info(
'Found endpoint for %s via: %s.',
self._client_name,
location,
)
return endpoint_url
logger.debug('No configured endpoint found.')
return None
def _get_snake_case_service_id(self, client_name):
# Get the service ID without loading the service data file, accounting
# for any aliases and standardizing the names with hyphens.
client_name = utils.SERVICE_NAME_ALIASES.get(client_name, client_name)
hyphenized_service_id = (
utils.CLIENT_NAME_TO_HYPHENIZED_SERVICE_ID_OVERRIDES.get(
client_name, client_name
)
)
return hyphenized_service_id.replace('-', '_')
def _get_service_env_var_name(self):
transformed_service_id_env = self._transformed_service_id.upper()
return f'AWS_ENDPOINT_URL_{transformed_service_id_env}'
def _get_services_config(self):
if 'services' not in self._scoped_config:
return {}
section_name = self._scoped_config['services']
services_section = self._full_config.get('services', {}).get(
section_name
)
if not services_section:
error_msg = (
f'The profile is configured to use the services '
f'section but the "{section_name}" services '
f'configuration does not exist.'
)
raise InvalidConfigError(error_msg=error_msg)
return services_section
def _get_endpoint_url_config_service(self):
snakecase_service_id = self._transformed_service_id.lower()
return (
self._get_services_config()
.get(snakecase_service_id, {})
.get('endpoint_url')
)
def _get_endpoint_url_config_global(self):
return self._scoped_config.get('endpoint_url')
def _get_endpoint_url_environment_service(self):
return EnvironmentProvider(
name=self._get_service_env_var_name(), env=self._environ
).provide()
def _get_endpoint_url_environment_global(self):
return EnvironmentProvider(
name='AWS_ENDPOINT_URL', env=self._environ
).provide()
|