Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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__
29logger = logging.getLogger(__name__)
32class ManagedIdentityError(ValueError):
33 pass
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"
44 # Valid values for key ID_TYPE
45 CLIENT_ID = "ClientId"
46 RESOURCE_ID = "ResourceId"
47 OBJECT_ID = "ObjectId"
48 SYSTEM_ASSIGNED = "SystemAssigned"
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 }
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))
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)
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))
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 })
83class SystemAssignedManagedIdentity(ManagedIdentity):
84 """Represent a system-assigned managed identity.
86 It is equivalent to a Python dict of::
88 {"ManagedIdentityIdType": "SystemAssigned", "Id": None}
90 or a JSON blob of::
92 {"ManagedIdentityIdType": "SystemAssigned", "Id": null}
93 """
94 def __init__(self):
95 super(SystemAssignedManagedIdentity, self).__init__(id_type=self.SYSTEM_ASSIGNED)
98class UserAssignedManagedIdentity(ManagedIdentity):
99 """Represent a user-assigned managed identity.
101 Depends on the id you provided, the outcome is equivalent to one of the below::
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")
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()
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.
144 It also provides token cache support.
146 .. note::
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"
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.
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.
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::
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)
200 For Service Fabric managed identity, ``http_client`` must be a
201 ``requests.Session`` using the standard ``requests.adapters.HTTPAdapter``.
202 MSAL derives a separate session for the Service Fabric endpoint so that
203 its certificate thumbprint can be validated before the Secret header is sent.
205 :param token_cache:
206 Optional. It accepts a :class:`msal.TokenCache` instance to store tokens.
207 It will use an in-memory token cache by default.
209 :param http_cache:
210 Optional. It has the same characteristics as the
211 :paramref:`msal.ClientApplication.http_cache`.
213 :param list[str] client_capabilities: (optional)
214 Allows configuration of one or more client capabilities, e.g. ["CP1"].
216 Client capability is meant to inform the Microsoft identity platform
217 (STS) what this client is capable for,
218 so STS can decide to turn on certain features.
220 Implementation details:
221 Client capability in Managed Identity is relayed as-is
222 via ``xms_cc`` parameter on the wire.
224 Recipe 1: Hard code a managed identity for your app::
226 import msal, requests
227 client = msal.ManagedIdentityClient(
228 msal.UserAssignedManagedIdentity(client_id="foo"),
229 http_client=requests.Session(),
230 )
231 token = client.acquire_token_for_client("resource")
233 Recipe 2: Write once, run everywhere.
234 If you use different managed identity on different deployment,
235 you may use an environment variable (such as MY_MANAGED_IDENTITY_CONFIG)
236 to store a json blob like
237 ``{"ManagedIdentityIdType": "ClientId", "Id": "foo"}`` or
238 ``{"ManagedIdentityIdType": "SystemAssigned", "Id": null}``.
239 The following app can load managed identity configuration dynamically::
241 import json, os, msal, requests
242 config = os.getenv("MY_MANAGED_IDENTITY_CONFIG")
243 assert config, "An ENV VAR with value should exist"
244 client = msal.ManagedIdentityClient(
245 json.loads(config),
246 http_client=requests.Session(),
247 )
248 token = client.acquire_token_for_client("resource")
249 """
250 if not ManagedIdentity.is_managed_identity(managed_identity):
251 raise ManagedIdentityError(
252 f"Incorrect managed_identity: {managed_identity}")
253 self._managed_identity = managed_identity
254 self._http_client = _ThrottledHttpClient(
255 # This class only throttles excess token acquisition requests.
256 # It does not provide retry.
257 # Retry is the http_client or caller's responsibility, not MSAL's.
258 #
259 # FWIW, here is the inconsistent retry recommendation.
260 # 1. Only MI on VM defines exotic 404 and 410 retry recommendations
261 # ( https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token#error-handling )
262 # (especially for 410 which was supposed to be a permanent failure).
263 # 2. MI on Service Fabric specifically suggests to not retry on 404.
264 # ( https://learn.microsoft.com/en-us/azure/service-fabric/how-to-managed-cluster-managed-identity-service-fabric-app-code#error-handling )
265 http_client,
266 http_cache=http_cache,
267 )
268 self._token_cache = token_cache or TokenCache()
269 self._client_capabilities = client_capabilities
271 def acquire_token_for_client(
272 self,
273 *,
274 resource: str, # If/when we support scope, resource will become optional
275 claims_challenge: Optional[str] = None,
276 ):
277 """Acquire token for the managed identity.
279 The result will be automatically cached.
280 Subsequent calls will automatically search from cache first.
282 :param resource: The resource for which the token is acquired.
284 :param claims_challenge:
285 Optional.
286 It is a string representation of a JSON object
287 (which contains lists of claims being requested).
289 The tenant admin may choose to revoke all Managed Identity tokens,
290 and then a *claims challenge* will be returned by the target resource,
291 as a `claims_challenge` directive in the `www-authenticate` header,
292 even if the app developer did not opt in for the "CP1" client capability.
293 Upon receiving a `claims_challenge`, MSAL will attempt to acquire a new token.
295 .. note::
297 Known issue: When an Azure VM has only one user-assigned managed identity,
298 and your app specifies to use system-assigned managed identity,
299 Azure VM may still return a token for your user-assigned identity.
301 This is a service-side behavior that cannot be changed by this library.
302 `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>`_
303 """
304 access_token_to_refresh = None # This could become a public parameter in the future
305 access_token_from_cache = None
306 client_id_in_cache = self._managed_identity.get(
307 ManagedIdentity.ID, "SYSTEM_ASSIGNED_MANAGED_IDENTITY")
308 now = time.time()
309 if True: # Attempt cache search even if receiving claims_challenge,
310 # because we want to locate the existing token (if any) and refresh it
311 matches = self._token_cache.search(
312 self._token_cache.CredentialType.ACCESS_TOKEN,
313 target=[resource],
314 query=dict(
315 client_id=client_id_in_cache,
316 environment=self.__instance,
317 realm=self._tenant,
318 home_account_id=None,
319 ),
320 )
321 for entry in matches:
322 expires_in = int(entry["expires_on"]) - now
323 if expires_in < 5*60: # Then consider it expired
324 continue # Removal is not necessary, it will be overwritten
325 if claims_challenge and not access_token_to_refresh:
326 # Since caller did not pinpoint the token causing claims challenge,
327 # we have to assume it is the first token we found in cache.
328 access_token_to_refresh = entry["secret"]
329 break
330 logger.debug("Cache hit an AT")
331 access_token_from_cache = { # Mimic a real response
332 "access_token": entry["secret"],
333 "token_type": entry.get("token_type", "Bearer"),
334 "expires_in": int(expires_in), # OAuth2 specs defines it as int
335 self._TOKEN_SOURCE: self._TOKEN_SOURCE_CACHE,
336 }
337 if "refresh_on" in entry:
338 access_token_from_cache["refresh_on"] = int(entry["refresh_on"])
339 if int(entry["refresh_on"]) < now: # aging
340 break # With a fallback in hand, we break here to go refresh
341 return access_token_from_cache # It is still good as new
342 try:
343 result = _obtain_token(
344 self._http_client, self._managed_identity, resource,
345 access_token_sha256_to_refresh=hashlib.sha256(
346 access_token_to_refresh.encode("utf-8")).hexdigest()
347 if access_token_to_refresh else None,
348 client_capabilities=self._client_capabilities,
349 )
350 if "access_token" in result:
351 expires_in = result.get("expires_in", 3600)
352 if "refresh_in" not in result and expires_in >= 7200:
353 result["refresh_in"] = int(expires_in / 2)
354 self._token_cache.add(dict(
355 client_id=client_id_in_cache,
356 scope=[resource],
357 token_endpoint="https://{}/{}".format(
358 self.__instance, self._tenant),
359 response=result,
360 params={},
361 data={},
362 ))
363 if "refresh_in" in result:
364 result["refresh_on"] = int(now + result["refresh_in"])
365 result[self._TOKEN_SOURCE] = self._TOKEN_SOURCE_IDP
366 if (result and "error" not in result) or (not access_token_from_cache):
367 return result
368 except: # The exact HTTP exception is transportation-layer dependent
369 # Typically network error. Potential AAD outage?
370 if not access_token_from_cache: # It means there is no fall back option
371 raise # We choose to bubble up the exception
372 return access_token_from_cache
375def _scope_to_resource(scope): # This is an experimental reasonable-effort approach
376 u = urlparse(scope)
377 if u.scheme:
378 return "{}://{}".format(u.scheme, u.netloc)
379 return scope # There is no much else we can do here
382def _get_arc_endpoint():
383 if "IDENTITY_ENDPOINT" in os.environ and "IMDS_ENDPOINT" in os.environ:
384 return os.environ["IDENTITY_ENDPOINT"]
385 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
386 sys.platform == "linux" and os.path.exists("/opt/azcmagent/bin/himds")
387 or sys.platform == "win32" and os.path.exists(os.path.expandvars(
388 # Avoid Windows-only "%EnvVar%" syntax so that tests can be run on Linux
389 r"${ProgramFiles}\AzureConnectedMachineAgent\himds.exe"
390 ))
391 ):
392 return "http://localhost:40342/metadata/identity/oauth2/token"
395APP_SERVICE = object()
396AZURE_ARC = object()
397CLOUD_SHELL = object() # In MSAL Python, token acquisition was done by
398 # PublicClientApplication(...).acquire_token_interactive(..., prompt="none")
399MACHINE_LEARNING = object()
400SERVICE_FABRIC = object()
401DEFAULT_TO_VM = object() # Unknown environment; default to VM; you may want to probe
402def get_managed_identity_source():
403 """Detect the current environment and return the likely identity source.
405 When this function returns ``CLOUD_SHELL``, you should use
406 :func:`msal.PublicClientApplication.acquire_token_interactive` with ``prompt="none"``
407 to obtain a token.
408 """
409 if ("IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ
410 and "IDENTITY_SERVER_THUMBPRINT" in os.environ
411 ):
412 return SERVICE_FABRIC
413 if "IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ:
414 return APP_SERVICE
415 if "MSI_ENDPOINT" in os.environ and "MSI_SECRET" in os.environ:
416 return MACHINE_LEARNING
417 if _get_arc_endpoint():
418 return AZURE_ARC
419 if _is_running_in_cloud_shell():
420 return CLOUD_SHELL
421 return DEFAULT_TO_VM
424def _obtain_token(
425 http_client, managed_identity, resource,
426 *,
427 access_token_sha256_to_refresh: Optional[str] = None,
428 client_capabilities: Optional[List[str]] = None,
429):
430 if ("IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ
431 and "IDENTITY_SERVER_THUMBPRINT" in os.environ
432 ):
433 if managed_identity:
434 logger.debug(
435 "Ignoring managed_identity parameter. "
436 "Managed Identity in Service Fabric is configured in the cluster, "
437 "not during runtime. See also "
438 "https://learn.microsoft.com/en-us/azure/service-fabric/configure-existing-cluster-enable-managed-identity-token-service")
439 return _obtain_token_on_service_fabric(
440 http_client,
441 os.environ["IDENTITY_ENDPOINT"],
442 os.environ["IDENTITY_HEADER"],
443 os.environ["IDENTITY_SERVER_THUMBPRINT"],
444 resource,
445 access_token_sha256_to_refresh=access_token_sha256_to_refresh,
446 client_capabilities=client_capabilities,
447 )
448 if "IDENTITY_ENDPOINT" in os.environ and "IDENTITY_HEADER" in os.environ:
449 return _obtain_token_on_app_service(
450 http_client,
451 os.environ["IDENTITY_ENDPOINT"],
452 os.environ["IDENTITY_HEADER"],
453 managed_identity,
454 resource,
455 )
456 if "MSI_ENDPOINT" in os.environ and "MSI_SECRET" in os.environ:
457 # 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
458 return _obtain_token_on_machine_learning(
459 http_client,
460 os.environ["MSI_ENDPOINT"],
461 os.environ["MSI_SECRET"],
462 managed_identity,
463 resource,
464 )
465 arc_endpoint = _get_arc_endpoint()
466 if arc_endpoint:
467 return _obtain_token_on_arc(
468 http_client, arc_endpoint, resource, managed_identity)
469 return _obtain_token_on_azure_vm(http_client, managed_identity, resource)
472def _adjust_param(params, managed_identity, types_mapping=None):
473 # Modify the params dict in place
474 id_name = (types_mapping or ManagedIdentity._types_mapping).get(
475 managed_identity.get(ManagedIdentity.ID_TYPE))
476 if id_name:
477 params[id_name] = managed_identity[ManagedIdentity.ID]
479def _obtain_token_on_azure_vm(http_client, managed_identity, resource):
480 # 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
481 logger.debug("Obtaining token via managed identity on Azure VM")
482 params = {
483 "api-version": "2018-02-01",
484 "resource": resource,
485 }
486 _adjust_param(params, managed_identity)
487 resp = http_client.get(
488 os.getenv(
489 "AZURE_POD_IDENTITY_AUTHORITY_HOST", "http://169.254.169.254"
490 ).strip("/") + "/metadata/identity/oauth2/token",
491 params=params,
492 headers={
493 "Metadata": "true",
494 "x-client-SKU": SKU,
495 "x-client-Ver": __version__,
496 "x-ms-client-request-id": str(uuid.uuid4()),
497 },
498 )
499 try:
500 payload = json.loads(resp.text)
501 if payload.get("access_token") and payload.get("expires_in"):
502 return { # Normalizing the payload into OAuth2 format
503 "access_token": payload["access_token"],
504 "expires_in": int(payload["expires_in"]),
505 "resource": payload.get("resource"),
506 "token_type": payload.get("token_type", "Bearer"),
507 }
508 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
509 except json.decoder.JSONDecodeError:
510 logger.debug("IMDS emits unexpected payload: %s", resp.text)
511 raise
513def _obtain_token_on_app_service(
514 http_client, endpoint, identity_header, managed_identity, resource,
515):
516 """Obtains token for
517 `App Service <https://learn.microsoft.com/en-us/azure/app-service/overview-managed-identity?tabs=portal%2Chttp#rest-endpoint-reference>`_,
518 Azure Functions, and Azure Automation.
519 """
520 # Prerequisite: Create your app service https://docs.microsoft.com/en-us/azure/app-service/quickstart-python
521 # Assign it a managed identity https://docs.microsoft.com/en-us/azure/app-service/overview-managed-identity?tabs=portal%2Chttp
522 # SSH into your container for testing https://docs.microsoft.com/en-us/azure/app-service/configure-linux-open-ssh-session
523 logger.debug("Obtaining token via managed identity on Azure App Service")
524 params = {
525 "api-version": "2019-08-01",
526 "resource": resource,
527 }
528 _adjust_param(params, managed_identity, types_mapping={
529 ManagedIdentity.CLIENT_ID: "client_id",
530 ManagedIdentity.RESOURCE_ID: "mi_res_id", # App Service's resource id uses "mi_res_id"
531 ManagedIdentity.OBJECT_ID: "object_id",
532 })
534 resp = http_client.get(
535 endpoint,
536 params=params,
537 headers={
538 "X-IDENTITY-HEADER": identity_header,
539 "Metadata": "true", # Unnecessary yet harmless for App Service,
540 # It will be needed by Azure Automation
541 # https://docs.microsoft.com/en-us/azure/automation/enable-managed-identity-for-automation#get-access-token-for-system-assigned-managed-identity-using-http-get
542 },
543 )
544 try:
545 payload = json.loads(resp.text)
546 if payload.get("access_token") and payload.get("expires_on"):
547 return { # Normalizing the payload into OAuth2 format
548 "access_token": payload["access_token"],
549 "expires_in": int(payload["expires_on"]) - int(time.time()),
550 "resource": payload.get("resource"),
551 "token_type": payload.get("token_type", "Bearer"),
552 }
553 return {
554 "error": "invalid_scope", # Empirically, wrong resource ends up with a vague statusCode=500
555 "error_description": "{}, {}".format(
556 payload.get("statusCode"), payload.get("message")),
557 }
558 except json.decoder.JSONDecodeError:
559 logger.debug("IMDS emits unexpected payload: %s", resp.text)
560 raise
562def _obtain_token_on_machine_learning(
563 http_client, endpoint, secret, managed_identity, resource,
564):
565 # Could not find protocol docs from https://docs.microsoft.com/en-us/azure/machine-learning
566 # The following implementation is back ported from Azure Identity 1.15.0
567 logger.debug("Obtaining token via managed identity on Azure Machine Learning")
568 params = {"api-version": "2017-09-01", "resource": resource}
569 _adjust_param(params, managed_identity)
570 if params["api-version"] == "2017-09-01" and "client_id" in params:
571 # Workaround for a known bug in Azure ML 2017 API
572 params["clientid"] = params.pop("client_id")
573 resp = http_client.get(
574 endpoint,
575 params=params,
576 headers={"secret": secret},
577 )
578 try:
579 payload = json.loads(resp.text)
580 if payload.get("access_token") and payload.get("expires_on"):
581 return { # Normalizing the payload into OAuth2 format
582 "access_token": payload["access_token"],
583 "expires_in": int(payload["expires_on"]) - int(time.time()),
584 "resource": payload.get("resource"),
585 "token_type": payload.get("token_type", "Bearer"),
586 }
587 return {
588 "error": "invalid_scope", # TODO: To be tested
589 "error_description": "{}".format(payload),
590 }
591 except json.decoder.JSONDecodeError:
592 logger.debug("IMDS emits unexpected payload: %s", resp.text)
593 raise
596def _obtain_token_on_service_fabric(
597 http_client, endpoint, identity_header, server_thumbprint, resource,
598 *,
599 access_token_sha256_to_refresh: str = None,
600 client_capabilities: Optional[List[str]] = None,
601):
602 """Obtains token for
603 `Service Fabric <https://learn.microsoft.com/en-us/azure/service-fabric/>`_
604 """
605 # Deployment https://learn.microsoft.com/en-us/azure/service-fabric/service-fabric-get-started-containers-linux
606 # 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
607 # 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
608 logger.debug("Obtaining token via managed identity on Azure Service Fabric")
609 parsed_endpoint = urlparse(endpoint)
610 if parsed_endpoint.scheme.lower() != "https" or not parsed_endpoint.hostname:
611 raise ManagedIdentityError(
612 "Service Fabric managed identity endpoint must use HTTPS.")
613 service_fabric_http_client = _create_service_fabric_http_client(
614 http_client, endpoint, _normalize_service_fabric_thumbprint(server_thumbprint))
615 resp = service_fabric_http_client.get(
616 endpoint,
617 params={k: v for k, v in {
618 "api-version": "2019-07-01-preview",
619 "resource": resource,
620 "token_sha256_to_refresh": access_token_sha256_to_refresh,
621 "xms_cc": ",".join(client_capabilities) if client_capabilities else None,
622 }.items() if v is not None},
623 headers={"Secret": identity_header},
624 )
625 try:
626 payload = json.loads(resp.text)
627 if payload.get("access_token") and payload.get("expires_on"):
628 return { # Normalizing the payload into OAuth2 format
629 "access_token": payload["access_token"],
630 "expires_in": int( # Despite the example in docs shows an integer,
631 payload["expires_on"] # Azure SDK team's test obtained a string.
632 ) - int(time.time()),
633 "resource": payload.get("resource"),
634 "token_type": payload["token_type"],
635 }
636 error = payload.get("error", {}) # https://learn.microsoft.com/en-us/azure/service-fabric/how-to-managed-identity-service-fabric-app-code#error-handling
637 error_mapping = { # Map Service Fabric errors into OAuth2 errors https://www.rfc-editor.org/rfc/rfc6749#section-5.2
638 "SecretHeaderNotFound": "unauthorized_client",
639 "ManagedIdentityNotFound": "invalid_client",
640 "ArgumentNullOrEmpty": "invalid_scope",
641 }
642 return {
643 "error": error_mapping.get(error.get("code"), "invalid_request"),
644 "error_description": resp.text,
645 }
646 except json.decoder.JSONDecodeError:
647 logger.debug("IMDS emits unexpected payload: %s", resp.text)
648 raise
651def _normalize_service_fabric_thumbprint(server_thumbprint):
652 normalized = "".join(
653 character for character in str(server_thumbprint)
654 if character not in " \t\r\n:")
655 if len(normalized) != 40 or any(
656 character not in "0123456789abcdefABCDEF"
657 for character in normalized):
658 raise ManagedIdentityError(
659 "IDENTITY_SERVER_THUMBPRINT must be a SHA-1 certificate thumbprint.")
660 return normalized.lower()
663class _ServiceFabricHTTPSConnection(HTTPSConnection):
664 """An HTTPS connection that authenticates the Service Fabric endpoint certificate."""
665 _server_thumbprint = None
667 def connect(self):
668 super(_ServiceFabricHTTPSConnection, self).connect()
669 if getattr(self, "proxy_is_forwarding", False):
670 self.close()
671 raise ssl.SSLCertVerificationError(
672 "Cannot validate the Service Fabric endpoint certificate through "
673 "a forwarding proxy.")
674 certificate = self.sock.getpeercert(binary_form=True)
675 actual_thumbprint = hashlib.sha1(certificate).hexdigest()
676 if not hmac.compare_digest(actual_thumbprint, self._server_thumbprint):
677 self.close()
678 raise ssl.SSLCertVerificationError(
679 "Service Fabric endpoint certificate thumbprint does not match "
680 "IDENTITY_SERVER_THUMBPRINT.")
681 self.is_verified = True
684class _ServiceFabricHTTPSConnectionPool(HTTPSConnectionPool):
685 ConnectionCls = _ServiceFabricHTTPSConnection
688class _ServiceFabricHTTPAdapter(HTTPAdapter):
689 """Use certificate-thumbprint authentication for the Service Fabric endpoint."""
691 def __init__(self, server_thumbprint, *args, **kwargs):
692 connection_class = type(
693 "_PinnedServiceFabricHTTPSConnection",
694 (_ServiceFabricHTTPSConnection,),
695 {"_server_thumbprint": server_thumbprint},
696 )
697 self._connection_pool_class = type(
698 "_PinnedServiceFabricHTTPSConnectionPool",
699 (_ServiceFabricHTTPSConnectionPool,),
700 {"ConnectionCls": connection_class},
701 )
702 super(_ServiceFabricHTTPAdapter, self).__init__(*args, **kwargs)
704 def _configure_pool_manager(self, pool_manager):
705 # PoolManager's mapping is module-global by default, so copy it before
706 # replacing HTTPS only for this derived Service Fabric session.
707 pool_manager.pool_classes_by_scheme = pool_manager.pool_classes_by_scheme.copy()
708 pool_manager.pool_classes_by_scheme["https"] = self._connection_pool_class
710 def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
711 super(_ServiceFabricHTTPAdapter, self).init_poolmanager(
712 connections, maxsize, block=block, **pool_kwargs)
713 self._configure_pool_manager(self.poolmanager)
715 def proxy_manager_for(self, proxy, **proxy_kwargs):
716 pool_manager = super(_ServiceFabricHTTPAdapter, self).proxy_manager_for(
717 proxy, **proxy_kwargs)
718 self._configure_pool_manager(pool_manager)
719 return pool_manager
721 def cert_verify(self, conn, url, verify, cert):
722 # The exact Service Fabric certificate thumbprint is the trust anchor.
723 # Do not inherit caller-provided verify=False or a custom CA configuration.
724 super(_ServiceFabricHTTPAdapter, self).cert_verify(
725 conn, url, verify=False, cert=cert)
728def _create_service_fabric_http_client(http_client, endpoint, server_thumbprint):
729 """Clone a standard Requests session and attach a pinning-only HTTPS transport.
731 Custom HTTP clients and adapters are rejected because MSAL cannot prove that
732 they will validate the certificate before transmitting the Secret header.
733 """
734 if isinstance(http_client, ThrottledHttpClientBase):
735 http_client = http_client.http_client
736 if not isinstance(http_client, requests.Session):
737 raise ManagedIdentityError(
738 "Service Fabric managed identity requires a requests.Session "
739 "with the standard HTTPAdapter.")
740 source_adapter = http_client.get_adapter(endpoint)
741 if type(source_adapter) is not HTTPAdapter:
742 raise ManagedIdentityError(
743 "Service Fabric managed identity does not support custom HTTP adapters.")
745 service_fabric_client = requests.Session()
746 service_fabric_client.headers = http_client.headers.copy()
747 service_fabric_client.cookies = http_client.cookies.copy()
748 service_fabric_client.auth = http_client.auth
749 service_fabric_client.params = copy.copy(http_client.params)
750 service_fabric_client.hooks = {
751 event: handlers[:] for event, handlers in http_client.hooks.items()}
752 service_fabric_client.proxies = http_client.proxies.copy()
753 service_fabric_client.stream = http_client.stream
754 service_fabric_client.trust_env = http_client.trust_env
755 service_fabric_client.max_redirects = http_client.max_redirects
756 service_fabric_client.cert = http_client.cert
757 service_fabric_client.verify = True
758 service_fabric_client.adapters.clear()
759 service_fabric_client.mount("https://", _ServiceFabricHTTPAdapter(
760 server_thumbprint,
761 max_retries=copy.deepcopy(source_adapter.max_retries),
762 pool_connections=source_adapter._pool_connections,
763 pool_maxsize=source_adapter._pool_maxsize,
764 pool_block=source_adapter._pool_block,
765 ))
766 return service_fabric_client
769_supported_arc_platforms_and_their_prefixes = {
770 "linux": "/var/opt/azcmagent/tokens",
771 "win32": os.path.expandvars(r"%ProgramData%\AzureConnectedMachineAgent\Tokens"),
772}
774class ArcPlatformNotSupportedError(ManagedIdentityError):
775 pass
777def _raise_if_arc_did_not_honor_user_assigned_identity(managed_identity, payload):
778 """Fail closed when a user-assigned identity was requested but Azure Arc did not confirm it.
780 A legacy Azure Arc agent ignores the client_id / object_id / msi_res_id selector and silently
781 returns the machine's system-assigned identity. An agent that supports user-assigned managed
782 identity echoes the identity it used in the token response. When that echo is missing or does
783 not match the requested selector, MSAL must not hand back a token for a different identity than
784 the one that was requested.
785 """
786 if not ManagedIdentity.is_user_assigned(managed_identity):
787 return # System-assigned: there is no requested identity to confirm
788 requested = managed_identity.get(ManagedIdentity.ID)
789 echoed = {
790 ManagedIdentity.CLIENT_ID: payload.get("client_id"),
791 ManagedIdentity.OBJECT_ID: payload.get("object_id"),
792 # Azure Arc echoes msi_res_id; accept the mi_res_id spelling too as a safety net
793 ManagedIdentity.RESOURCE_ID: payload.get("msi_res_id") or payload.get("mi_res_id"),
794 }.get(managed_identity.get(ManagedIdentity.ID_TYPE))
795 # Compare case-insensitively: client_id / object_id are GUIDs, and an ARM resource id
796 # (msi_res_id) can legitimately differ in segment casing.
797 if not echoed or str(echoed).lower() != str(requested).lower():
798 raise ManagedIdentityError(
799 "Azure Arc did not confirm the requested user-assigned managed identity "
800 "in the token response. The agent likely does not support user-assigned "
801 "managed identities and returned the system-assigned identity.")
803def _obtain_token_on_arc(http_client, endpoint, resource, managed_identity=None):
804 # https://learn.microsoft.com/en-us/azure/azure-arc/servers/managed-identity-authentication
805 logger.debug("Obtaining token via managed identity on Azure Arc")
806 params = {"api-version": "2020-06-01", "resource": resource}
807 if managed_identity:
808 _adjust_param(params, managed_identity, types_mapping={
809 ManagedIdentity.CLIENT_ID: "client_id",
810 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
811 ManagedIdentity.OBJECT_ID: "object_id",
812 })
813 resp = http_client.get(
814 endpoint,
815 params=params.copy(),
816 headers={"Metadata": "true"},
817 )
818 www_auth = "www-authenticate" # Header in lower case
819 challenge = {
820 # Normalized to lowercase, because header names are case-insensitive
821 # https://datatracker.ietf.org/doc/html/rfc7230#section-3.2
822 k.lower(): v for k, v in resp.headers.items() if k.lower() == www_auth
823 }.get(www_auth, "").split("=") # Output will be ["Basic realm", "content"]
824 if not ( # https://datatracker.ietf.org/doc/html/rfc7617#section-2
825 len(challenge) == 2 and challenge[0].lower() == "basic realm"):
826 raise ManagedIdentityError(
827 "Unrecognizable WWW-Authenticate header: {}".format(resp.headers))
828 if sys.platform not in _supported_arc_platforms_and_their_prefixes:
829 raise ArcPlatformNotSupportedError(
830 f"Platform {sys.platform} was undefined and unsupported")
831 filename = os.path.join(
832 # 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
833 _supported_arc_platforms_and_their_prefixes[sys.platform],
834 os.path.splitext(os.path.basename(challenge[1]))[0] + ".key")
835 if os.stat(filename).st_size > 4096: # Check size BEFORE loading its content
836 raise ManagedIdentityError("Local key file shall not be larger than 4KB")
837 with open(filename) as f:
838 secret = f.read()
839 response = http_client.get(
840 endpoint,
841 params=params.copy(),
842 headers={"Metadata": "true", "Authorization": "Basic {}".format(secret)},
843 )
844 try:
845 payload = json.loads(response.text)
846 if payload.get("access_token") and payload.get("expires_in"):
847 # Example: https://learn.microsoft.com/en-us/azure/azure-arc/servers/media/managed-identity-authentication/bash-token-output-example.png
848 _raise_if_arc_did_not_honor_user_assigned_identity(
849 managed_identity, payload)
850 return {
851 "access_token": payload["access_token"],
852 "expires_in": int(payload["expires_in"]),
853 "token_type": payload.get("token_type", "Bearer"),
854 "resource": payload.get("resource"),
855 }
856 except json.decoder.JSONDecodeError:
857 pass
858 return {
859 "error": "invalid_request",
860 "error_description": response.text,
861 }