(http_client, scopes, client_id=None, data=None)
| 42 | |
| 43 | |
| 44 | def _obtain_token(http_client, scopes, client_id=None, data=None): |
| 45 | resp = http_client.post( |
| 46 | "http://localhost:50342/oauth2/token", |
| 47 | data=dict( |
| 48 | data or {}, |
| 49 | resource=" ".join(map(_scope_to_resource, scopes))), |
| 50 | headers={"Metadata": "true"}, |
| 51 | ) |
| 52 | if resp.status_code >= 300: |
| 53 | logger.debug("Cloud Shell IMDS error: %s", resp.text) |
| 54 | cs_error = json.loads(resp.text).get("error", {}) |
| 55 | return {k: v for k, v in { |
| 56 | "error": cs_error.get("code"), |
| 57 | "error_description": cs_error.get("message"), |
| 58 | }.items() if v} |
| 59 | imds_payload = json.loads(resp.text) |
| 60 | BEARER = "Bearer" |
| 61 | oauth2_response = { |
| 62 | "access_token": imds_payload["access_token"], |
| 63 | "expires_in": int(imds_payload["expires_in"]), |
| 64 | "token_type": imds_payload.get("token_type", BEARER), |
| 65 | } |
| 66 | expected_token_type = (data or {}).get("token_type", BEARER) |
| 67 | if oauth2_response["token_type"] != expected_token_type: |
| 68 | return { # Generate a normal error (rather than an intrusive exception) |
| 69 | "error": "broker_error", |
| 70 | "error_description": "token_type {} is not supported by this version of Azure Portal".format( |
| 71 | expected_token_type), |
| 72 | } |
| 73 | parts = imds_payload["access_token"].split(".") |
| 74 | |
| 75 | # The following default values are useful in SSH Cert scenario |
| 76 | client_info = { # Default value, in case the real value will be unavailable |
| 77 | "uid": "user", |
| 78 | "utid": "cloudshell", |
| 79 | } |
| 80 | now = time.time() |
| 81 | preferred_username = "currentuser@cloudshell" |
| 82 | oauth2_response["id_token_claims"] = { # First 5 claims are required per OIDC |
| 83 | "iss": "cloudshell", |
| 84 | "sub": "user", |
| 85 | "aud": client_id, |
| 86 | "exp": now + 3600, |
| 87 | "iat": now, |
| 88 | "preferred_username": preferred_username, # Useful as MSAL account's username |
| 89 | } |
| 90 | |
| 91 | if len(parts) == 3: # Probably a JWT. Use it to derive client_info and id token. |
| 92 | try: |
| 93 | # Data defined in https://docs.microsoft.com/en-us/azure/active-directory/develop/access-tokens#payload-claims |
| 94 | jwt_payload = json.loads(decode_part(parts[1])) |
| 95 | client_info = { |
| 96 | # Mimic a real home_account_id, |
| 97 | # so that this pseudo account and a real account would interop. |
| 98 | "uid": jwt_payload.get("oid", "user"), |
| 99 | "utid": jwt_payload.get("tid", "cloudshell"), |
| 100 | } |
| 101 | oauth2_response["id_token_claims"] = { |
no test coverage detected