-
Notifications
You must be signed in to change notification settings - Fork 202
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Acquire SSH cert from Cloud Shell IMDS
Cloud Shell Detection PoC: Silent flow utilizes Cloud Shell IMDS Introduce get_accounts(username=msal.CURRENT_USER) A reasonable-effort to convert scope to resource Replace get_accounts(username=msal.CURRENT_USER) by acquire_token_interactive(..., prompt="none") Detect unsupported Portal so that AzCLI could fallback
- Loading branch information
Showing
4 changed files
with
175 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| # Copyright (c) Microsoft Corporation. | ||
| # All rights reserved. | ||
| # | ||
| # This code is licensed under the MIT License. | ||
|
|
||
| """This module wraps Cloud Shell's IMDS-like interface inside an OAuth2-like helper""" | ||
| import base64 | ||
| import json | ||
| import logging | ||
| import os | ||
| import time | ||
| try: # Python 2 | ||
| from urlparse import urlparse | ||
| except: # Python 3 | ||
| from urllib.parse import urlparse | ||
| from .oauth2cli.oidc import decode_part | ||
|
|
||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| def _is_running_in_cloud_shell(): | ||
| return os.environ.get("AZUREPS_HOST_ENVIRONMENT", "").startswith("cloud-shell") | ||
|
|
||
|
|
||
| def _scope_to_resource(scope): # This is an experimental reasonable-effort approach | ||
| cloud_shell_supported_audiences = [ | ||
| "https://analysis.windows.net/powerbi/api", # Came from https://msazure.visualstudio.com/One/_git/compute-CloudShell?path=/src/images/agent/env/envconfig.PROD.json | ||
| "https://pas.windows.net/CheckMyAccess/Linux/.default", # Cloud Shell accepts it as-is | ||
| ] | ||
| for a in cloud_shell_supported_audiences: | ||
| if scope.startswith(a): | ||
| return a | ||
| u = urlparse(scope) | ||
| if u.scheme: | ||
| return "{}://{}".format(u.scheme, u.netloc) | ||
| return scope # There is no much else we can do here | ||
|
|
||
|
|
||
| def _obtain_token(http_client, scopes, client_id=None, data=None): | ||
| resp = http_client.post( | ||
| "http://localhost:50342/oauth2/token", | ||
| data=dict( | ||
| data or {}, | ||
| resource=" ".join(map(_scope_to_resource, scopes))), | ||
| headers={"Metadata": "true"}, | ||
| ) | ||
| if resp.status_code >= 300: | ||
| logger.debug("Cloud Shell IMDS error: %s", resp.text) | ||
| cs_error = json.loads(resp.text).get("error", {}) | ||
| return {k: v for k, v in { | ||
| "error": cs_error.get("code"), | ||
| "error_description": cs_error.get("message"), | ||
| }.items() if v} | ||
| imds_payload = json.loads(resp.text) | ||
| BEARER = "Bearer" | ||
| oauth2_response = { | ||
| "access_token": imds_payload["access_token"], | ||
| "expires_in": int(imds_payload["expires_in"]), | ||
| "token_type": imds_payload.get("token_type", BEARER), | ||
| } | ||
| expected_token_type = (data or {}).get("token_type", BEARER) | ||
| if oauth2_response["token_type"] != expected_token_type: | ||
| return { # Generate a normal error (rather than an intrusive exception) | ||
| "error": "broker_error", | ||
| "error_description": "token_type {} is not supported by this version of Azure Portal".format( | ||
| expected_token_type), | ||
| } | ||
| parts = imds_payload["access_token"].split(".") | ||
|
|
||
| # The following default values are useful in SSH Cert scenario | ||
| client_info = { # Default value, in case the real value will be unavailable | ||
| "uid": "user", | ||
| "utid": "cloudshell", | ||
| } | ||
| now = time.time() | ||
| preferred_username = "currentuser@cloudshell" | ||
| oauth2_response["id_token_claims"] = { # First 5 claims are required per OIDC | ||
| "iss": "cloudshell", | ||
| "sub": "user", | ||
| "aud": client_id, | ||
| "exp": now + 3600, | ||
| "iat": now, | ||
| "preferred_username": preferred_username, # Useful as MSAL account's username | ||
| } | ||
|
|
||
| if len(parts) == 3: # Probably a JWT. Use it to derive client_info and id token. | ||
| try: | ||
| # Data defined in https://docs.microsoft.com/en-us/azure/active-directory/develop/access-tokens#payload-claims | ||
| jwt_payload = json.loads(decode_part(parts[1])) | ||
| client_info = { | ||
| # Mimic a real home_account_id, | ||
| # so that this pseudo account and a real account would interop. | ||
| "uid": jwt_payload.get("oid", "user"), | ||
| "utid": jwt_payload.get("tid", "cloudshell"), | ||
| } | ||
| oauth2_response["id_token_claims"] = { | ||
| "iss": jwt_payload["iss"], | ||
| "sub": jwt_payload["sub"], # Could use oid instead | ||
| "aud": client_id, | ||
| "exp": jwt_payload["exp"], | ||
| "iat": jwt_payload["iat"], | ||
| "preferred_username": jwt_payload.get("preferred_username") # V2 | ||
| or jwt_payload.get("unique_name") # V1 | ||
| or preferred_username, | ||
| } | ||
| except ValueError: | ||
| logger.debug("Unable to decode jwt payload: %s", parts[1]) | ||
| oauth2_response["client_info"] = base64.b64encode( | ||
| # Mimic a client_info, so that MSAL would create an account | ||
| json.dumps(client_info).encode("utf-8")).decode("utf-8") | ||
| oauth2_response["id_token_claims"]["tid"] = client_info["utid"] # TBD | ||
|
|
||
| ## Note: Decided to not surface resource back as scope, | ||
| ## because they would cause the downstream OAuth2 code path to | ||
| ## cache the token with a different scope and won't hit them later. | ||
| #if imds_payload.get("resource"): | ||
| # oauth2_response["scope"] = imds_payload["resource"] | ||
| if imds_payload.get("refresh_token"): | ||
| oauth2_response["refresh_token"] = imds_payload["refresh_token"] | ||
| return oauth2_response | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters