14
14
# See the License for the specific language governing permissions and
15
15
# limitations under the License.
16
16
17
+ import json
17
18
import logging
18
19
from typing import Optional
19
20
@@ -49,11 +50,15 @@ class VPCInstanceTokenManager(JWTTokenManager):
49
50
"""
50
51
51
52
METADATA_SERVICE_VERSION = '2021-09-20'
53
+ DEFAULT_IMS_ENDPOINT = 'http://169.254.169.254'
52
54
53
55
def __init__ (self ,
54
56
iam_profile_crn : Optional [str ] = None ,
55
57
iam_profile_id : Optional [str ] = None ,
56
- url : Optional [str ] = 'http://169.254.169.254' ) -> None :
58
+ url : Optional [str ] = None ) -> None :
59
+ if not url :
60
+ url = self .DEFAULT_IMS_ENDPOINT
61
+
57
62
super ().__init__ (url )
58
63
59
64
self .iam_profile_crn = iam_profile_crn
@@ -74,13 +79,11 @@ def request_token(self) -> dict:
74
79
url = self .url + '/instance_identity/v1/iam_token'
75
80
76
81
if self .iam_profile_crn :
77
- #pylint: disable=missing-format-argument-key
78
82
request_payload = {
79
- 'trusted_profile' : '{"crn": "{0}"}' .format (self .iam_profile_crn )}
83
+ 'trusted_profile' : '{{ "crn": "{0}"} }' .format (self .iam_profile_crn )}
80
84
if self .iam_profile_id :
81
- #pylint: disable=missing-format-argument-key
82
85
request_payload = {
83
- 'trusted_profile' : '{"id": "{0}"}' .format (self .iam_profile_id )}
86
+ 'trusted_profile' : '{{ "id": "{0}"} }' .format (self .iam_profile_id )}
84
87
85
88
headers = {
86
89
'Content-Type' : 'application/json' ,
@@ -92,11 +95,11 @@ def request_token(self) -> dict:
92
95
'Invoking VPC \' create_iam_token\' operation: %s' , url )
93
96
response = self ._request (
94
97
method = 'POST' ,
95
- url = ( self . url + self . OPERATION_PATH ) if self . url else self . url ,
98
+ url = url ,
96
99
headers = headers ,
97
100
params = {'version' : self .METADATA_SERVICE_VERSION },
98
- data = request_payload )
99
- logging .debug ('Returned from VPC \' create_access_token \' operation."' )
101
+ data = json . dumps ( request_payload ) )
102
+ logging .debug ('Returned from VPC \' create_iam_token \' operation."' )
100
103
101
104
return response
102
105
@@ -143,8 +146,7 @@ def retrieve_instance_identity_token(self) -> str:
143
146
method = 'PUT' ,
144
147
url = url ,
145
148
headers = headers ,
146
- data = request_body ,
147
- proxies = self .proxies )
149
+ data = json .dumps (request_body ))
148
150
logging .debug ('Returned from VPC \' create_access_token\' operation."' )
149
151
150
152
return response ['access_token' ]
0 commit comments