| 20 | |
| 21 | |
| 22 | def _get_azure_response( |
| 23 | resource: str, client_id: Optional[str] = None, timeout: float = 5 |
| 24 | ) -> dict[str, Any]: |
| 25 | # Deferred import to save overall import time. |
| 26 | from urllib.request import Request, urlopen |
| 27 | |
| 28 | url = "http://169.254.169.254/metadata/identity/oauth2/token" |
| 29 | url += "?api-version=2018-02-01" |
| 30 | url += f"&resource={resource}" |
| 31 | if client_id: |
| 32 | url += f"&client_id={client_id}" |
| 33 | headers = {"Metadata": "true", "Accept": "application/json"} |
| 34 | request = Request(url, headers=headers) # noqa: S310 |
| 35 | try: |
| 36 | with urlopen(request, timeout=timeout) as response: # noqa: S310 |
| 37 | status = response.status |
| 38 | body = response.read().decode("utf8") |
| 39 | except Exception as e: |
| 40 | msg = "Failed to acquire IMDS access token: %s" % e |
| 41 | raise ValueError(msg) from None |
| 42 | |
| 43 | if status != 200: |
| 44 | msg = "Failed to acquire IMDS access token." |
| 45 | raise ValueError(msg) |
| 46 | try: |
| 47 | data = json.loads(body) |
| 48 | except Exception: |
| 49 | raise ValueError("Azure IMDS response must be in JSON format") from None |
| 50 | |
| 51 | for key in ["access_token", "expires_in"]: |
| 52 | if not data.get(key): |
| 53 | msg = "Azure IMDS response must contain %s, but was %s." |
| 54 | msg = msg % (key, body) |
| 55 | raise ValueError(msg) |
| 56 | |
| 57 | return data |