Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.10/site-packages/msal/managed_identity.py: 29%

1# Copyright (c) Microsoft Corporation. 

2# All rights reserved. 

3# 

4# This code is licensed under the MIT License. 

5import copy 

6import hashlib 

7import hmac 

8import json 

9import logging 

10import os 

11import ssl 

12import sys 

13import time 

14import uuid 

15from urllib.parse import urlparse # Python 3+ 

16from collections import UserDict # Python 3+ 

17from typing import List, Optional, Union # Needed in Python 3.7 & 3.8 

18import requests 

19from requests.adapters import HTTPAdapter 

20from urllib3.connection import HTTPSConnection 

21from urllib3.connectionpool import HTTPSConnectionPool 

22from .token_cache import TokenCache 

23from .individual_cache import _IndividualCache as IndividualCache 

24from .throttled_http_client import ThrottledHttpClientBase, RetryAfterParser 

25from .cloudshell import _is_running_in_cloud_shell 

26from .sku import SKU, __version__ 

27 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32class ManagedIdentityError(ValueError): 

33 pass 

34 

35 

36class ManagedIdentity(UserDict): 

37 """Feed an instance of this class to :class:`msal.ManagedIdentityClient` 

38 to acquire token for the specified managed identity. 

39 """ 

40 # The key names used in config dict 

41 ID_TYPE = "ManagedIdentityIdType" # Contains keyword ManagedIdentity so its json equivalent will be more readable 

42 ID = "Id" 

43 

44 # Valid values for key ID_TYPE 

45 CLIENT_ID = "ClientId" 

46 RESOURCE_ID = "ResourceId" 

47 OBJECT_ID = "ObjectId" 

48 SYSTEM_ASSIGNED = "SystemAssigned" 

49 

50 _types_mapping = { # Maps type name in configuration to type name on wire 

51 CLIENT_ID: "client_id", 

52 RESOURCE_ID: "msi_res_id", # VM's IMDS prefers msi_res_id https://github.com/Azure/azure-rest-api-specs/blob/dba6ed1f03bda88ac6884c0a883246446cc72495/specification/imds/data-plane/Microsoft.InstanceMetadataService/stable/2018-10-01/imds.json#L233-L239 

53 OBJECT_ID: "object_id", 

54 } 

55 

56 @classmethod 

57 def is_managed_identity(cls, unknown): 

58 return (isinstance(unknown, ManagedIdentity) 

59 or cls.is_system_assigned(unknown) 

60 or cls.is_user_assigned(unknown)) 

61 

62 @classmethod 

63 def is_system_assigned(cls, unknown): 

64 return isinstance(unknown, SystemAssignedManagedIdentity) or ( 

65 isinstance(unknown, dict) 

66 and unknown.get(cls.ID_TYPE) == cls.SYSTEM_ASSIGNED) 

67 

68 @classmethod 

69 def is_user_assigned(cls, unknown): 

70 return isinstance(unknown, UserAssignedManagedIdentity) or ( 

71 isinstance(unknown, dict) 

72 and unknown.get(cls.ID_TYPE) in cls._types_mapping 

73 and unknown.get(cls.ID)) 

74 

75 def __init__(self, identifier=None, id_type=None): 

76 # Undocumented. Use subclasses instead. 

77 super(ManagedIdentity, self).__init__({ 

78 self.ID_TYPE: id_type, 

79 self.ID: identifier, 

80 }) 

81 

82 

83class SystemAssignedManagedIdentity(ManagedIdentity): 

84 """Represent a system-assigned managed identity. 

85 

86 It is equivalent to a Python dict of:: 

87 

88 {"ManagedIdentityIdType": "SystemAssigned", "Id": None} 

89 

90 or a JSON blob of:: 

91 

92 {"ManagedIdentityIdType": "SystemAssigned", "Id": null} 

93 """ 

94 def __init__(self): 

95 super(SystemAssignedManagedIdentity, self).__init__(id_type=self.SYSTEM_ASSIGNED) 

96 

97 

98class UserAssignedManagedIdentity(ManagedIdentity): 

99 """Represent a user-assigned managed identity. 

100 

101 Depends on the id you provided, the outcome is equivalent to one of the below:: 

102 

103 {"ManagedIdentityIdType": "ClientId", "Id": "foo"} 

104 {"ManagedIdentityIdType": "ResourceId", "Id": "foo"} 

105 {"ManagedIdentityIdType": "ObjectId", "Id": "foo"} 

106 """ 

107 def __init__(self, *, client_id=None, resource_id=None, object_id=None): 

108 if client_id and not resource_id and not object_id: 

109 super(UserAssignedManagedIdentity, self).__init__( 

110 id_type=self.CLIENT_ID, identifier=client_id) 

111 elif not client_id and resource_id and not object_id: 

112 super(UserAssignedManagedIdentity, self).__init__( 

113 id_type=self.RESOURCE_ID, identifier=resource_id) 

114 elif not client_id and not resource_id and object_id: 

115 super(UserAssignedManagedIdentity, self).__init__( 

116 id_type=self.OBJECT_ID, identifier=object_id) 

117 else: 

118 raise ManagedIdentityError( 

119 "You shall specify one of the three parameters: " 

120 "client_id, resource_id, object_id") 

121 

122 

123class _ThrottledHttpClient(ThrottledHttpClientBase): 

124 def __init__(self, *args, **kwargs): 

125 super(_ThrottledHttpClient, self).__init__(*args, **kwargs) 

126 self.get = IndividualCache( # All MIs (except Cloud Shell) use GETs 

127 mapping=self._expiring_mapping, 

128 key_maker=lambda func, args, kwargs: "REQ {} hash={} 429/5xx/Retry-After".format( 

129 args[0], # It is the endpoint, typically a constant per MI type 

130 self._hash( 

131 # Managed Identity flavors have inconsistent parameters. 

132 # We simply choose to hash them all. 

133 str(kwargs.get("params")) + str(kwargs.get("data"))), 

134 ), 

135 expires_in=RetryAfterParser(5).parse, # 5 seconds default for non-PCA 

136 )(self.get) # Note: Decorate the parent get(), not the http_client.get() 

137 

138 

139class ManagedIdentityClient(object): 

140 """This API encapsulates multiple managed identity back-ends: 

141 VM, App Service, Azure Automation (Runbooks), Azure Function, Service Fabric, 

142 and Azure Arc. 

143 

144 It also provides token cache support. 

145 

146 .. note:: 

147 

148 Cloud Shell support is NOT implemented in this class. 

149 Since MSAL Python 1.18 in May 2022, it has been implemented in 

150 :func:`PublicClientApplication.acquire_token_interactive` via calling pattern 

151 ``PublicClientApplication(...).acquire_token_interactive(scopes=[...], prompt="none")``. 

152 That is appropriate, because Cloud Shell yields a token with 

153 delegated permissions for the end user who has signed in to the Azure Portal 

154 (like what a ``PublicClientApplication`` does), 

155 not a token with application permissions for an app. 

156 """ 

157 __instance = "localhost" # We used to get this value from socket.getfqdn() 

158 # but it is unreliable because getfqdn() either hangs or returns empty value 

159 # on some misconfigured machines 

160 _tenant = "managed_identity" 

161 _TOKEN_SOURCE = "token_source" 

162 _TOKEN_SOURCE_IDP = "identity_provider" 

163 _TOKEN_SOURCE_CACHE = "cache" 

164 

165 def __init__( 

166 self, 

167 managed_identity: Union[ 

168 dict, 

169 ManagedIdentity, # Could use Type[ManagedIdentity] but it is deprecated in Python 3.9+ 

170 SystemAssignedManagedIdentity, 

171 UserAssignedManagedIdentity, 

172 ], 

173 *, 

174 http_client, 

175 token_cache=None, 

176 http_cache=None, 

177 client_capabilities: Optional[List[str]] = None, 

178 ): 

179 """Create a managed identity client. 

180 

181 :param managed_identity: 

182 It accepts an instance of :class:`SystemAssignedManagedIdentity` 

183 or :class:`UserAssignedManagedIdentity`. 

184 They are equivalent to a dict with a certain shape, 

185 which may be loaded from a JSON configuration file or an env var. 

186 

187 :param http_client: 

188 An http client object. For example, you can use ``requests.Session()``, 

189 optionally with exponential backoff behavior demonstrated in this recipe:: 

190 

191 import msal, requests 

192 from requests.adapters import HTTPAdapter, Retry 

193 s = requests.Session() 

194 retries = Retry(total=3, backoff_factor=0.1, status_forcelist=[ 

195 429, 500, 501, 502, 503, 504]) 

196 s.mount('https://', HTTPAdapter(max_retries=retries)) 

197 managed_identity = ... 

198 client = msal.ManagedIdentityClient(managed_identity, http_client=s) 

199 

200 For Service Fabric managed identity, ``http_client`` must be a 

201 ``requests.Session`` using ``requests.adapters.HTTPAdapter`` or a 

202 subclass for the Service Fabric endpoint. 

203 MSAL derives a separate session for the Service Fabric endpoint so that 

204 its certificate thumbprint can be validated before the Secret header is sent. 

205 Standard session, retry, and connection-pool settings are preserved, 

206 but custom adapter behavior is not used. 

207 

208 :param token_cache: 

209 Optional. It accepts a :class:`msal.TokenCache` instance to store tokens. 

210 It will use an in-memory token cache by default. 

211 

212 :param http_cache: 

213 Optional. It has the same characteristics as the 

214 :paramref:`msal.ClientApplication.http_cache`. 

215 

216 :param list[str] client_capabilities: (optional) 

217 Allows configuration of one or more client capabilities, e.g. ["CP1"]. 

218 

219 Client capability is meant to inform the Microsoft identity platform 

220 (STS) what this client is capable for, 

221 so STS can decide to turn on certain features. 

222 

223 Implementation details: 

224 Client capability in Managed Identity is relayed as-is 

225 via ``xms_cc`` parameter on the wire. 

226 

227 Recipe 1: Hard code a managed identity for your app:: 

228 

229 import msal, requests 

230 client = msal.ManagedIdentityClient( 

231 msal.UserAssignedManagedIdentity(client_id="foo"), 

232 http_client=requests.Session(), 

233 ) 

234 token = client.acquire_token_for_client("resource") 

235 

236 Recipe 2: Write once, run everywhere. 

237 If you use different managed identity on different deployment, 

238 you may use an environment variable (such as MY_MANAGED_IDENTITY_CONFIG) 

239 to store a json blob like 

240 ``{"ManagedIdentityIdType": "ClientId", "Id": "foo"}`` or 

241 ``{"ManagedIdentityIdType": "SystemAssigned", "Id": null}``. 

242 The following app can load managed identity configuration dynamically:: 

243 

244 import json, os, msal, requests 

245 config = os.getenv("MY_MANAGED_IDENTITY_CONFIG") 

246 assert config, "An ENV VAR with value should exist" 

247 client = msal.ManagedIdentityClient( 

248 json.loads(config), 

249 http_client=requests.Session(), 

250 ) 

251 token = client.acquire_token_for_client("resource") 

252 """ 

253 if not ManagedIdentity.is_managed_identity(managed_identity): 

254 raise ManagedIdentityError( 

255 f"Incorrect managed_identity: {managed_identity}") 

256 self._managed_identity = managed_identity 

257 self._http_client = _ThrottledHttpClient( 

258 # This class only throttles excess token acquisition requests. 

259 # It does not provide retry. 

260 # Retry is the http_client or caller's responsibility, not MSAL's. 

261 # 

262 # FWIW, here is the inconsistent retry recommendation. 

263 # 1. Only MI on VM defines exotic 404 and 410 retry recommendations 

264 # ( https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token#error-handling ) 

265 # (especially for 410 which was supposed to be a permanent failure). 

266 # 2. MI on Service Fabric specifically suggests to not retry on 404. 

267 # ( https://learn.microsoft.com/en-us/azure/service-fabric/how-to-managed-cluster-managed-identity-service-fabric-app-code#error-handling ) 

268 http_client, 

269 http_cache=http_cache, 

270 ) 

271 self._token_cache = token_cache or TokenCache() 

272 self._client_capabilities = client_capabilities 

273 

274 def acquire_token_for_client( 

275 self, 

276 *, 

277 resource: str, # If/when we support scope, resource will become optional 

278 claims_challenge: Optional[str] = None, 

279 ): 

280 """Acquire token for the managed identity. 

281 

282 The result will be automatically cached. 

283 Subsequent calls will automatically search from cache first. 

284 

285 :param resource: The resource for which the token is acquired. 

286 

287 :param claims_challenge: 

288 Optional. 

289 It is a string representation of a JSON object 

290 (which contains lists of claims being requested). 

291 

292 The tenant admin may choose to revoke all Managed Identity tokens, 

293 and then a *claims challenge* will be returned by the target resource, 

294 as a `claims_challenge` directive in the `www-authenticate` header, 

295 even if the app developer did not opt in for the "CP1" client capability. 

296 Upon receiving a `claims_challenge`, MSAL will attempt to acquire a new token. 

297 

298 .. note:: 

299 

300 Known issue: When an Azure VM has only one user-assigned managed identity, 

301 and your app specifies to use system-assigned managed identity, 

302 Azure VM may still return a token for your user-assigned identity. 

303 

304 This is a service-side behavior that cannot be changed by this library. 

305 `Azure VM docs <https://learn.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http>`_ 

306 """ 

307 access_token_to_refresh = None # This could become a public parameter in the future 

308 access_token_from_cache = None 

309 client_id_in_cache = self._managed_identity.get( 

310 ManagedIdentity.ID, "SYSTEM_ASSIGNED_MANAGED_IDENTITY") 

311 now = time.time() 

312 if True: # Attempt cache search even if receiving claims_challenge, 

313 # because we want to locate the existing token (if any) and refresh it 

314 matches = self._token_cache.search( 

315 self._token_cache.CredentialType.ACCESS_TOKEN, 

316 target=[resource], 

317 query=dict( 

318 client_id=client_id_in_cache, 

319 environment=self.__instance, 

320 realm=self._tenant, 

321 home_account_id=None, 

322 ), 

323 ) 

324 for entry in matches: 

325 expires_in = int(entry["expires_on"]) - now 

326 if expires_in < 5*60: # Then consider it expired 

327 continue # Removal is not necessary, it will be overwritten 

328 if claims_challenge and not access_token_to_refresh: 

329 # Since caller did not pinpoint the token causing claims challenge, 

330 # we have to assume it is the first token we found in cache. 

331 access_token_to_refresh = entry["secret"] 

332 break 

333 logger.debug("Cache hit an AT") 

334 access_token_from_cache = { # Mimic a real response 

335 "access_token": entry["secret"], 

336 "token_type": entry.get("token_type", "Bearer"), 

337 "expires_in": int(expires_in), # OAuth2 specs defines it as int 

338 self._TOKEN_SOURCE: self._TOKEN_SOURCE_CACHE, 

339 } 

340 if "refresh_on" in entry: 

341 access_token_from_cache["refresh_on"] = int(entry["refresh_on"]) 

342 if int(entry["refresh_on"]) < now: # aging 

343 break # With a fallback in hand, we break here to go refresh 

344 return access_token_from_cache # It is still good as new 

345 try: 

346 result = _obtain_token( 

347 self._http_client, self._managed_identity, resource, 

348 access_token_sha256_to_refresh=hashlib.sha256( 

349 access_token_to_refresh.encode("utf-8")).hexdigest() 

350 if access_token_to_refresh else None, 

351 client_capabilities=self._client_capabilities, 

352 ) 

353 if "access_token" in result: 

354 expires_in = result.get("expires_in", 3600) 

355 if "refresh_in" not in result and expires_in >= 7200: 

356 result["refresh_in"] = int(expires_in / 2) 

357 self._token_cache.add(dict( 

358 client_id=client_id_in_cache, 

359 scope=[resource], 

360 token_endpoint="https://{}/{}".format( 

361 self.__instance, self._tenant), 

362 response=result, 

363 params={}, 

364 data={}, 

365 )) 

366 if "refresh_in" in result: 

367 result["refresh_on"] = int(now + result["refresh_in"]) 

368 result[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP 

369 if (result and "error" not in result) or (not access_token_from_cache): 

370 return result 

371 except: # The exact HTTP exception is transportation-layer dependent 

372 # Typically network error. Potential AAD outage? 

373 if not access_token_from_cache: # It means there is no fall back option 

374 raise # We choose to bubble up the exception 

375 return access_token_from_cache 

376 

377 

378def _scope_to_resource(scope): # This is an experimental reasonable-effort approach 

379 u = urlparse(scope) 

380 if u.scheme: 

381 return "{}://{}".format(u.scheme, u.netloc) 

382 return scope # There is no much else we can do here 

383 

384 

385def _get_arc_endpoint(): 

386 if "IDENTITY_ENDPOINT" in os.environ and "IMDS_ENDPOINT" in os.environ: 

387 return os.environ["IDENTITY_ENDPOINT"] 

388 if ( # Defined in https://eng.ms/docs/cloud-ai-platform/azure-core/azure-management-and-platforms/control-plane-bburns/hybrid-resource-provider/azure-arc-for-servers/specs/extension_authoring 

389 sys.platform == "linux" and os.path.exists("/opt/azcmagent/bin/himds") 

390 or sys.platform == "win32" and os.path.exists(os.path.expandvars( 

391 # Avoid Windows-only "%EnvVar%" syntax so that tests can be run on Linux 

392 r"${ProgramFiles}\AzureConnectedMachineAgent\himds.exe" 

393 )) 

394 ): 

395 return "http://localhost:40342/metadata/identity/oauth2/token" 

396 

397 

398APP_SERVICE = object() 

399AZURE_ARC = object() 

400CLOUD_SHELL = object() # In MSAL Python, token acquisition was done by 

401 # PublicClientApplication(...).acquire_token_interactive(..., prompt="none") 

402MACHINE_LEARNING = object() 

403SERVICE_FABRIC = object() 

404DEFAULT_TO_VM = object() # Unknown environment; default to VM; you may want to probe 

405def get_managed_identity_source(): 

406 """Detect the current environment and return the likely identity source. 

407 

408 When this function returns ``CLOUD_SHELL``, you should use 

409 :func:`msal.PublicClientApplication.acquire_token_interactive` with ``prompt="none"`` 

410 to obtain a token. 

411 """ 

412 if ("IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ 

413 and "IDENTITY_SERVER_THUMBPRINT" in os.environ 

414 ): 

415 return SERVICE_FABRIC 

416 if "IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ: 

417 return APP_SERVICE 

418 if "MSI_ENDPOINT" in os.environ and "MSI_SECRET" in os.environ: 

419 return MACHINE_LEARNING 

420 if _get_arc_endpoint(): 

421 return AZURE_ARC 

422 if _is_running_in_cloud_shell(): 

423 return CLOUD_SHELL 

424 return DEFAULT_TO_VM 

425 

426 

427def _obtain_token( 

428 http_client, managed_identity, resource, 

429 *, 

430 access_token_sha256_to_refresh: Optional[str] = None, 

431 client_capabilities: Optional[List[str]] = None, 

432): 

433 if ("IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ 

434 and "IDENTITY_SERVER_THUMBPRINT" in os.environ 

435 ): 

436 if managed_identity: 

437 logger.debug( 

438 "Ignoring managed_identity parameter. " 

439 "Managed Identity in Service Fabric is configured in the cluster, " 

440 "not during runtime. See also " 

441 "https://learn.microsoft.com/en-us/azure/service-fabric/configure-existing-cluster-enable-managed-identity-token-service") 

442 return _obtain_token_on_service_fabric( 

443 http_client, 

444 os.environ["IDENTITY_ENDPOINT"], 

445 os.environ["IDENTITY_HEADER"], 

446 os.environ["IDENTITY_SERVER_THUMBPRINT"], 

447 resource, 

448 access_token_sha256_to_refresh=access_token_sha256_to_refresh, 

449 client_capabilities=client_capabilities, 

450 ) 

451 if "IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ: 

452 return _obtain_token_on_app_service( 

453 http_client, 

454 os.environ["IDENTITY_ENDPOINT"], 

455 os.environ["IDENTITY_HEADER"], 

456 managed_identity, 

457 resource, 

458 ) 

459 if "MSI_ENDPOINT" in os.environ and "MSI_SECRET" in os.environ: 

460 # Back ported from https://github.com/Azure/azure-sdk-for-python/blob/azure-identity_1.15.0/sdk/identity/azure-identity/azure/identity/_credentials/azure_ml.py 

461 return _obtain_token_on_machine_learning( 

462 http_client, 

463 os.environ["MSI_ENDPOINT"], 

464 os.environ["MSI_SECRET"], 

465 managed_identity, 

466 resource, 

467 ) 

468 arc_endpoint = _get_arc_endpoint() 

469 if arc_endpoint: 

470 return _obtain_token_on_arc( 

471 http_client, arc_endpoint, resource, managed_identity) 

472 return _obtain_token_on_azure_vm(http_client, managed_identity, resource) 

473 

474 

475def _adjust_param(params, managed_identity, types_mapping=None): 

476 # Modify the params dict in place 

477 id_name = (types_mapping or ManagedIdentity._types_mapping).get( 

478 managed_identity.get(ManagedIdentity.ID_TYPE)) 

479 if id_name: 

480 params[id_name] = managed_identity[ManagedIdentity.ID] 

481 

482def _obtain_token_on_azure_vm(http_client, managed_identity, resource): 

483 # Based on https://docs.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http 

484 logger.debug("Obtaining token via managed identity on Azure VM") 

485 params = { 

486 "api-version": "2018-02-01", 

487 "resource": resource, 

488 } 

489 _adjust_param(params, managed_identity) 

490 resp = http_client.get( 

491 os.getenv( 

492 "AZURE_POD_IDENTITY_AUTHORITY_HOST", "http://169.254.169.254" 

493 ).strip("/") + "/metadata/identity/oauth2/token", 

494 params=params, 

495 headers={ 

496 "Metadata": "true", 

497 "x-client-SKU": SKU, 

498 "x-client-Ver": __version__, 

499 "x-ms-client-request-id": str(uuid.uuid4()), 

500 }, 

501 ) 

502 try: 

503 payload = json.loads(resp.text) 

504 if payload.get("access_token") and payload.get("expires_in"): 

505 return { # Normalizing the payload into OAuth2 format 

506 "access_token": payload["access_token"], 

507 "expires_in": int(payload["expires_in"]), 

508 "resource": payload.get("resource"), 

509 "token_type": payload.get("token_type", "Bearer"), 

510 } 

511 return payload # It would be {"error": ..., "error_description": ...} according to https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token#error-handling 

512 except json.decoder.JSONDecodeError: 

513 logger.debug("IMDS emits unexpected payload: %s", resp.text) 

514 raise 

515 

516def _obtain_token_on_app_service( 

517 http_client, endpoint, identity_header, managed_identity, resource, 

518): 

519 """Obtains token for 

520 `App Service <https://learn.microsoft.com/en-us/azure/app-service/overview-managed-identity?tabs=portal%2Chttp#rest-endpoint-reference>`_, 

521 Azure Functions, and Azure Automation. 

522 """ 

523 # Prerequisite: Create your app service https://docs.microsoft.com/en-us/azure/app-service/quickstart-python 

524 # Assign it a managed identity https://docs.microsoft.com/en-us/azure/app-service/overview-managed-identity?tabs=portal%2Chttp 

525 # SSH into your container for testing https://docs.microsoft.com/en-us/azure/app-service/configure-linux-open-ssh-session 

526 logger.debug("Obtaining token via managed identity on Azure App Service") 

527 params = { 

528 "api-version": "2019-08-01", 

529 "resource": resource, 

530 } 

531 _adjust_param(params, managed_identity, types_mapping={ 

532 ManagedIdentity.CLIENT_ID: "client_id", 

533 ManagedIdentity.RESOURCE_ID: "mi_res_id", # App Service's resource id uses "mi_res_id" 

534 ManagedIdentity.OBJECT_ID: "object_id", 

535 }) 

536 

537 resp = http_client.get( 

538 endpoint, 

539 params=params, 

540 headers={ 

541 "X-IDENTITY-HEADER": identity_header, 

542 "Metadata": "true", # Unnecessary yet harmless for App Service, 

543 # It will be needed by Azure Automation 

544 # https://docs.microsoft.com/en-us/azure/automation/enable-managed-identity-for-automation#get-access-token-for-system-assigned-managed-identity-using-http-get 

545 }, 

546 ) 

547 try: 

548 payload = json.loads(resp.text) 

549 if payload.get("access_token") and payload.get("expires_on"): 

550 return { # Normalizing the payload into OAuth2 format 

551 "access_token": payload["access_token"], 

552 "expires_in": int(payload["expires_on"]) - int(time.time()), 

553 "resource": payload.get("resource"), 

554 "token_type": payload.get("token_type", "Bearer"), 

555 } 

556 return { 

557 "error": "invalid_scope", # Empirically, wrong resource ends up with a vague statusCode=500 

558 "error_description": "{}, {}".format( 

559 payload.get("statusCode"), payload.get("message")), 

560 } 

561 except json.decoder.JSONDecodeError: 

562 logger.debug("IMDS emits unexpected payload: %s", resp.text) 

563 raise 

564 

565def _obtain_token_on_machine_learning( 

566 http_client, endpoint, secret, managed_identity, resource, 

567): 

568 # Could not find protocol docs from https://docs.microsoft.com/en-us/azure/machine-learning 

569 # The following implementation is back ported from Azure Identity 1.15.0 

570 logger.debug("Obtaining token via managed identity on Azure Machine Learning") 

571 params = {"api-version": "2017-09-01", "resource": resource} 

572 _adjust_param(params, managed_identity) 

573 if params["api-version"] == "2017-09-01" and "client_id" in params: 

574 # Workaround for a known bug in Azure ML 2017 API 

575 params["clientid"] = params.pop("client_id") 

576 resp = http_client.get( 

577 endpoint, 

578 params=params, 

579 headers={"secret": secret}, 

580 ) 

581 try: 

582 payload = json.loads(resp.text) 

583 if payload.get("access_token") and payload.get("expires_on"): 

584 return { # Normalizing the payload into OAuth2 format 

585 "access_token": payload["access_token"], 

586 "expires_in": int(payload["expires_on"]) - int(time.time()), 

587 "resource": payload.get("resource"), 

588 "token_type": payload.get("token_type", "Bearer"), 

589 } 

590 return { 

591 "error": "invalid_scope", # TODO: To be tested 

592 "error_description": "{}".format(payload), 

593 } 

594 except json.decoder.JSONDecodeError: 

595 logger.debug("IMDS emits unexpected payload: %s", resp.text) 

596 raise 

597 

598 

599def _obtain_token_on_service_fabric( 

600 http_client, endpoint, identity_header, server_thumbprint, resource, 

601 *, 

602 access_token_sha256_to_refresh: str = None, 

603 client_capabilities: Optional[List[str]] = None, 

604): 

605 """Obtains token for 

606 `Service Fabric <https://learn.microsoft.com/en-us/azure/service-fabric/>`_ 

607 """ 

608 # Deployment https://learn.microsoft.com/en-us/azure/service-fabric/service-fabric-get-started-containers-linux 

609 # See also https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/identity/azure-identity/tests/managed-identity-live/service-fabric/service_fabric.md 

610 # Protocol https://learn.microsoft.com/en-us/azure/service-fabric/how-to-managed-identity-service-fabric-app-code#acquiring-an-access-token-using-rest-api 

611 logger.debug("Obtaining token via managed identity on Azure Service Fabric") 

612 parsed_endpoint = urlparse(endpoint) 

613 if parsed_endpoint.scheme.lower() != "https" or not parsed_endpoint.hostname: 

614 raise ManagedIdentityError( 

615 "Service Fabric managed identity endpoint must use HTTPS.") 

616 service_fabric_http_client = _create_service_fabric_http_client( 

617 http_client, endpoint, _normalize_service_fabric_thumbprint(server_thumbprint)) 

618 resp = service_fabric_http_client.get( 

619 endpoint, 

620 params={k: v for k, v in { 

621 "api-version": "2019-07-01-preview", 

622 "resource": resource, 

623 "token_sha256_to_refresh": access_token_sha256_to_refresh, 

624 "xms_cc": ",".join(client_capabilities) if client_capabilities else None, 

625 }.items() if v is not None}, 

626 headers={"Secret": identity_header}, 

627 ) 

628 try: 

629 payload = json.loads(resp.text) 

630 if payload.get("access_token") and payload.get("expires_on"): 

631 return { # Normalizing the payload into OAuth2 format 

632 "access_token": payload["access_token"], 

633 "expires_in": int( # Despite the example in docs shows an integer, 

634 payload["expires_on"] # Azure SDK team's test obtained a string. 

635 ) - int(time.time()), 

636 "resource": payload.get("resource"), 

637 "token_type": payload["token_type"], 

638 } 

639 error = payload.get("error", {}) # https://learn.microsoft.com/en-us/azure/service-fabric/how-to-managed-identity-service-fabric-app-code#error-handling 

640 error_mapping = { # Map Service Fabric errors into OAuth2 errors https://www.rfc-editor.org/rfc/rfc6749#section-5.2 

641 "SecretHeaderNotFound": "unauthorized_client", 

642 "ManagedIdentityNotFound": "invalid_client", 

643 "ArgumentNullOrEmpty": "invalid_scope", 

644 } 

645 return { 

646 "error": error_mapping.get(error.get("code"), "invalid_request"), 

647 "error_description": resp.text, 

648 } 

649 except json.decoder.JSONDecodeError: 

650 logger.debug("IMDS emits unexpected payload: %s", resp.text) 

651 raise 

652 

653 

654def _normalize_service_fabric_thumbprint(server_thumbprint): 

655 normalized = "".join( 

656 character for character in str(server_thumbprint) 

657 if character not in " \t\r\n:") 

658 if len(normalized) != 40 or any( 

659 character not in "0123456789abcdefABCDEF" 

660 for character in normalized): 

661 raise ManagedIdentityError( 

662 "IDENTITY_SERVER_THUMBPRINT must be a SHA-1 certificate thumbprint.") 

663 return normalized.lower() 

664 

665 

666class _ServiceFabricHTTPSConnection(HTTPSConnection): 

667 """An HTTPS connection that authenticates the Service Fabric endpoint certificate.""" 

668 _server_thumbprint = None 

669 

670 def connect(self): 

671 super(_ServiceFabricHTTPSConnection, self).connect() 

672 if getattr(self, "proxy_is_forwarding", False): 

673 self.close() 

674 raise ssl.SSLCertVerificationError( 

675 "Cannot validate the Service Fabric endpoint certificate through " 

676 "a forwarding proxy.") 

677 certificate = self.sock.getpeercert(binary_form=True) 

678 actual_thumbprint = hashlib.sha1(certificate).hexdigest() 

679 if not hmac.compare_digest(actual_thumbprint, self._server_thumbprint): 

680 self.close() 

681 raise ssl.SSLCertVerificationError( 

682 "Service Fabric endpoint certificate thumbprint does not match " 

683 "IDENTITY_SERVER_THUMBPRINT.") 

684 self.is_verified = True 

685 

686 

687class _ServiceFabricHTTPSConnectionPool(HTTPSConnectionPool): 

688 ConnectionCls = _ServiceFabricHTTPSConnection 

689 

690 

691class _ServiceFabricHTTPAdapter(HTTPAdapter): 

692 """Use certificate-thumbprint authentication for the Service Fabric endpoint.""" 

693 

694 def __init__(self, server_thumbprint, *args, **kwargs): 

695 connection_class = type( 

696 "_PinnedServiceFabricHTTPSConnection", 

697 (_ServiceFabricHTTPSConnection,), 

698 {"_server_thumbprint": server_thumbprint}, 

699 ) 

700 self._connection_pool_class = type( 

701 "_PinnedServiceFabricHTTPSConnectionPool", 

702 (_ServiceFabricHTTPSConnectionPool,), 

703 {"ConnectionCls": connection_class}, 

704 ) 

705 super(_ServiceFabricHTTPAdapter, self).__init__(*args, **kwargs) 

706 

707 def _configure_pool_manager(self, pool_manager): 

708 # PoolManager's mapping is module-global by default, so copy it before 

709 # replacing HTTPS only for this derived Service Fabric session. 

710 pool_manager.pool_classes_by_scheme = pool_manager.pool_classes_by_scheme.copy() 

711 pool_manager.pool_classes_by_scheme["https"] = self._connection_pool_class 

712 

713 def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs): 

714 super(_ServiceFabricHTTPAdapter, self).init_poolmanager( 

715 connections, maxsize, block=block, **pool_kwargs) 

716 self._configure_pool_manager(self.poolmanager) 

717 

718 def proxy_manager_for(self, proxy, **proxy_kwargs): 

719 pool_manager = super(_ServiceFabricHTTPAdapter, self).proxy_manager_for( 

720 proxy, **proxy_kwargs) 

721 self._configure_pool_manager(pool_manager) 

722 return pool_manager 

723 

724 def cert_verify(self, conn, url, verify, cert): 

725 # The exact Service Fabric certificate thumbprint is the trust anchor. 

726 # Do not inherit caller-provided verify=False or a custom CA configuration. 

727 super(_ServiceFabricHTTPAdapter, self).cert_verify( 

728 conn, url, verify=False, cert=cert) 

729 

730 

731def _create_service_fabric_http_client(http_client, endpoint, server_thumbprint): 

732 """Derive a Requests session with a pinning-only HTTPS transport. 

733 

734 HTTPAdapter subclasses are accepted as sources of retry and pool settings, 

735 but the derived session always uses MSAL's certificate-pinning adapter. 

736 Other HTTP clients and adapters are unsupported. 

737 """ 

738 if isinstance(http_client, ThrottledHttpClientBase): 

739 http_client = http_client.http_client 

740 if not isinstance(http_client, requests.Session): 

741 raise ManagedIdentityError( 

742 "Service Fabric managed identity requires a requests.Session " 

743 "with an HTTPAdapter or subclass.") 

744 source_adapter = http_client.get_adapter(endpoint) 

745 if not isinstance(source_adapter, HTTPAdapter): 

746 raise ManagedIdentityError( 

747 "Service Fabric managed identity requires an HTTPAdapter or subclass.") 

748 

749 service_fabric_client = requests.Session() 

750 service_fabric_client.headers = http_client.headers.copy() 

751 service_fabric_client.cookies = http_client.cookies.copy() 

752 service_fabric_client.auth = http_client.auth 

753 service_fabric_client.params = copy.copy(http_client.params) 

754 service_fabric_client.hooks = { 

755 event: handlers[:] for event, handlers in http_client.hooks.items()} 

756 service_fabric_client.proxies = http_client.proxies.copy() 

757 service_fabric_client.stream = http_client.stream 

758 service_fabric_client.trust_env = http_client.trust_env 

759 service_fabric_client.max_redirects = http_client.max_redirects 

760 service_fabric_client.cert = http_client.cert 

761 service_fabric_client.verify = True 

762 service_fabric_client.adapters.clear() 

763 service_fabric_client.mount("https://", _ServiceFabricHTTPAdapter( 

764 server_thumbprint, 

765 max_retries=copy.deepcopy(source_adapter.max_retries), 

766 pool_connections=source_adapter._pool_connections, 

767 pool_maxsize=source_adapter._pool_maxsize, 

768 pool_block=source_adapter._pool_block, 

769 )) 

770 return service_fabric_client 

771 

772 

773_supported_arc_platforms_and_their_prefixes = { 

774 "linux": "/var/opt/azcmagent/tokens", 

775 "win32": os.path.expandvars(r"%ProgramData%\AzureConnectedMachineAgent\Tokens"), 

776} 

777 

778class ArcPlatformNotSupportedError(ManagedIdentityError): 

779 pass 

780 

781def _raise_if_arc_did_not_honor_user_assigned_identity(managed_identity, payload): 

782 """Fail closed when a user-assigned identity was requested but Azure Arc did not confirm it. 

783 

784 A legacy Azure Arc agent ignores the client_id / object_id / msi_res_id selector and silently 

785 returns the machine's system-assigned identity. An agent that supports user-assigned managed 

786 identity echoes the identity it used in the token response. When that echo is missing or does 

787 not match the requested selector, MSAL must not hand back a token for a different identity than 

788 the one that was requested. 

789 """ 

790 if not ManagedIdentity.is_user_assigned(managed_identity): 

791 return # System-assigned: there is no requested identity to confirm 

792 requested = managed_identity.get(ManagedIdentity.ID) 

793 echoed = { 

794 ManagedIdentity.CLIENT_ID: payload.get("client_id"), 

795 ManagedIdentity.OBJECT_ID: payload.get("object_id"), 

796 # Azure Arc echoes msi_res_id; accept the mi_res_id spelling too as a safety net 

797 ManagedIdentity.RESOURCE_ID: payload.get("msi_res_id") or payload.get("mi_res_id"), 

798 }.get(managed_identity.get(ManagedIdentity.ID_TYPE)) 

799 # Compare case-insensitively: client_id / object_id are GUIDs, and an ARM resource id 

800 # (msi_res_id) can legitimately differ in segment casing. 

801 if not echoed or str(echoed).lower() != str(requested).lower(): 

802 raise ManagedIdentityError( 

803 "Azure Arc did not confirm the requested user-assigned managed identity " 

804 "in the token response. The agent likely does not support user-assigned " 

805 "managed identities and returned the system-assigned identity.") 

806 

807def _obtain_token_on_arc(http_client, endpoint, resource, managed_identity=None): 

808 # https://learn.microsoft.com/en-us/azure/azure-arc/servers/managed-identity-authentication 

809 logger.debug("Obtaining token via managed identity on Azure Arc") 

810 params = {"api-version": "2020-06-01", "resource": resource} 

811 if managed_identity: 

812 _adjust_param(params, managed_identity, types_mapping={ 

813 ManagedIdentity.CLIENT_ID: "client_id", 

814 ManagedIdentity.RESOURCE_ID: "msi_res_id", # Azure Arc honors the IMDS msi_res_id spelling; mi_res_id is ignored and returns the system-assigned identity 

815 ManagedIdentity.OBJECT_ID: "object_id", 

816 }) 

817 resp = http_client.get( 

818 endpoint, 

819 params=params.copy(), 

820 headers={"Metadata": "true"}, 

821 ) 

822 if resp.status_code != 401: 

823 return { 

824 "error": "invalid_request", 

825 "error_description": resp.text, 

826 } 

827 www_auth = "www-authenticate" # Header in lower case 

828 challenge = { 

829 # Normalized to lowercase, because header names are case-insensitive 

830 # https://datatracker.ietf.org/doc/html/rfc7230#section-3.2 

831 k.lower(): v for k, v in resp.headers.items() if k.lower() == www_auth 

832 }.get(www_auth, "").split("=") # Output will be ["Basic realm", "content"] 

833 if not ( # https://datatracker.ietf.org/doc/html/rfc7617#section-2 

834 len(challenge) == 2 and challenge[0].lower() == "basic realm"): 

835 raise ManagedIdentityError( 

836 "Unrecognizable WWW-Authenticate header: {}".format(resp.headers)) 

837 if sys.platform not in _supported_arc_platforms_and_their_prefixes: 

838 raise ArcPlatformNotSupportedError( 

839 f"Platform {sys.platform} was undefined and unsupported") 

840 filename = os.path.join( 

841 # This algorithm is documented in an internal doc https://msazure.visualstudio.com/One/_wiki/wikis/One.wiki/233012/VM-Extension-Authoring-for-Arc?anchor=2.-obtaining-tokens 

842 _supported_arc_platforms_and_their_prefixes[sys.platform], 

843 os.path.splitext(os.path.basename(challenge[1]))[0] + ".key") 

844 if os.stat(filename).st_size > 4096: # Check size BEFORE loading its content 

845 raise ManagedIdentityError("Local key file shall not be larger than 4KB") 

846 with open(filename) as f: 

847 secret = f.read() 

848 response = http_client.get( 

849 endpoint, 

850 params=params.copy(), 

851 headers={"Metadata": "true", "Authorization": "Basic {}".format(secret)}, 

852 ) 

853 try: 

854 payload = json.loads(response.text) 

855 if payload.get("access_token") and payload.get("expires_in"): 

856 # Example: https://learn.microsoft.com/en-us/azure/azure-arc/servers/media/managed-identity-authentication/bash-token-output-example.png 

857 _raise_if_arc_did_not_honor_user_assigned_identity( 

858 managed_identity, payload)