codekingpro/portable-devtools

1# Copyright (c) Microsoft Corporation.2# All rights reserved.3#4# This code is licensed under the MIT License.5 6"""This module wraps Cloud Shell's IMDS-like interface inside an OAuth2-like helper"""7import base648import json9import logging10import os11import time12try:  # Python 213    from urlparse import urlparse14except:  # Python 315    from urllib.parse import urlparse16from .oauth2cli.oidc import decode_part17 18 19logger = logging.getLogger(__name__)20 21 22def _is_running_in_cloud_shell():23    return os.environ.get("AZUREPS_HOST_ENVIRONMENT", "").startswith("cloud-shell")24 25 26def _scope_to_resource(scope):  # This is an experimental reasonable-effort approach27    cloud_shell_supported_audiences = [28        "https://analysis.windows.net/powerbi/api",  # Came from https://msazure.visualstudio.com/One/_git/compute-CloudShell?path=/src/images/agent/env/envconfig.PROD.json29        "https://pas.windows.net/CheckMyAccess/Linux/.default",  # Cloud Shell accepts it as-is30        ]31    for a in cloud_shell_supported_audiences:32        if scope.startswith(a):33            return a34    u = urlparse(scope)35    if u.scheme:36        return "{}://{}".format(u.scheme, u.netloc)37    return scope  # There is no much else we can do here38 39 40def _obtain_token(http_client, scopes, client_id=None, data=None):41    resp = http_client.post(42        "http://localhost:50342/oauth2/token",43        data=dict(44            data or {},45            resource=" ".join(map(_scope_to_resource, scopes))),46        headers={"Metadata": "true"},47        )48    if resp.status_code >= 300:49        logger.debug("Cloud Shell IMDS error: %s", resp.text)50        cs_error = json.loads(resp.text).get("error", {})51        return {k: v for k, v in {52            "error": cs_error.get("code"),53            "error_description": cs_error.get("message"),54            }.items() if v}55    imds_payload = json.loads(resp.text)56    BEARER = "Bearer"57    oauth2_response = {58        "access_token": imds_payload["access_token"],59        "expires_in": int(imds_payload["expires_in"]),60        "token_type": imds_payload.get("token_type", BEARER),61        }62    expected_token_type = (data or {}).get("token_type", BEARER)63    if oauth2_response["token_type"] != expected_token_type:64        return {  # Generate a normal error (rather than an intrusive exception)65            "error": "broker_error",66            "error_description": "token_type {} is not supported by this version of Azure Portal".format(67                expected_token_type),68            }69    parts = imds_payload["access_token"].split(".")70 71    # The following default values are useful in SSH Cert scenario72    client_info = {  # Default value, in case the real value will be unavailable73        "uid": "user",74        "utid": "cloudshell",75        }76    now = time.time()77    preferred_username = "currentuser@cloudshell"78    oauth2_response["id_token_claims"] = {  # First 5 claims are required per OIDC79        "iss": "cloudshell",80        "sub": "user",81        "aud": client_id,82        "exp": now + 3600,83        "iat": now,84        "preferred_username": preferred_username,  # Useful as MSAL account's username85        }86 87    if len(parts) == 3:  # Probably a JWT. Use it to derive client_info and id token.88        try:89            # Data defined in https://docs.microsoft.com/en-us/azure/active-directory/develop/access-tokens#payload-claims90            jwt_payload = json.loads(decode_part(parts[1]))91            client_info = {92                # Mimic a real home_account_id,93                # so that this pseudo account and a real account would interop.94                "uid": jwt_payload.get("oid", "user"),95                "utid": jwt_payload.get("tid", "cloudshell"),96                }97            oauth2_response["id_token_claims"] = {98                "iss": jwt_payload["iss"],99                "sub": jwt_payload["sub"],  # Could use oid instead100                "aud": client_id,101                "exp": jwt_payload["exp"],102                "iat": jwt_payload["iat"],103                "preferred_username": jwt_payload.get("preferred_username")  # V2104                    or jwt_payload.get("unique_name")  # V1105                    or preferred_username,106                }107        except ValueError:108            logger.debug("Unable to decode jwt payload: %s", parts[1])109    oauth2_response["client_info"] = base64.b64encode(110        # Mimic a client_info, so that MSAL would create an account111        json.dumps(client_info).encode("utf-8")).decode("utf-8")112    oauth2_response["id_token_claims"]["tid"] = client_info["utid"]  # TBD113 114    ## Note: Decided to not surface resource back as scope,115    ##       because they would cause the downstream OAuth2 code path to116    ##       cache the token with a different scope and won't hit them later.117    #if imds_payload.get("resource"):118    #    oauth2_response["scope"] = imds_payload["resource"]119    if imds_payload.get("refresh_token"):120        oauth2_response["refresh_token"] = imds_payload["refresh_token"]121    return oauth2_response122 123