获取ACR的Azure Active Directory刷新令牌

2024-05-13 11:33:51 发布

您现在位置:Python中文网/ 问答频道 /正文

我一直在尝试使用AAD交互式身份验证流来列出给定Azure容器注册表实例中的图像(详细信息为here),该实例需要AAD访问令牌和AAD刷新令牌才能获得资源本身的刷新/访问令牌

我假设所需的AAD访问令牌是“management.azure.com”的简单访问令牌,我使用以下代码检索该令牌:

from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
credentials = InteractiveBrowserCredential()
aad_access_token = credentials.get_token('https://management.azure.com/.default')

返回正常访问令牌(例如“eyJ0eX…”)。但是,我找不到任何以编程方式获取刷新令牌的引用;似乎所有Azure SDK都会在后台处理这些问题。由于到资源本身(ACR)的身份验证流需要这两个令牌,是否有任何方法可以获取此令牌

如果存在任何其他编程方式来列出没有刷新标记的容器图像,我将非常乐意试一试

同样需要注意的是,我在Azure CLI无法正常工作的环境中工作,因此我必须使用python SDK或类似的工具,而不能使用诸如“az login”或“az get credentials”之类的系统命令


Tags: 实例图像comtoken身份验证get编程资源
1条回答
网友
1楼 · 发布于 2024-05-13 11:33:51

因此,对于使用AAD和Azure资源刷新和访问令牌的功能,我有一些误解。经过反复试验,我发现以下代码工作正常:

import requests
from azure.identity import InteractiveBrowserCredential
from pprint import pprint

CATALOG_SCOPE = "registry:catalog:*"
AZURE_MANAGEMENT_SCOPE_URL = 'https://management.azure.com/.default'
ACR_URL = "YOUR_REPO.azurecr.io"


def get_aad_access_token():
    cred = InteractiveBrowserCredential()
    aad_access_token = cred.get_token(AZURE_MANAGEMENT_SCOPE_URL)
    return aad_access_token.token


def get_acr_refresh_token(acr_url, aad_access_token):
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
    }

    data = {
        'grant_type': 'access_token',
        'service': acr_url,
        'access_token': aad_access_token
    }

    response = requests.post(f'https://{acr_url}/oauth2/exchange', headers=headers, data=data)
    acr_refresh_token = json.loads(response.content)
    # pprint(response.__dict__)
    return acr_refresh_token['refresh_token']


def get_acr_access_token(acr_refresh_token, acr_url, scope):
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
    }

    data = {
        'grant_type': 'refresh_token',
        'service': acr_url,
        'scope': scope,
        'refresh_token': acr_refresh_token
    }

    response = requests.post(f'https://{acr_url}/oauth2/token', headers=headers, data=data)
    # pprint(response.__dict__)
    acr_access_token = json.loads(response.content)
    return acr_access_token['access_token']


def list_acr_images(acr_url):
    operation = "/v2/_catalog/"
    aad_access_token = get_aad_access_token()
    acr_refresh_token = get_acr_refresh_token(acr_url, aad_access_token)
    acr_access_token = get_acr_access_token(acr_refresh_token, acr_url, CATALOG_SCOPE)

    headers = {
        'Authorization': f"Bearer {acr_access_token}",
    }

    response = requests.post(f'https://{acr_url}{operation}', headers=headers)
    return json.loads(response.content)


if __name__ == '__main__':
    pprint(list_acr_images(ACR_URL))

相关问题 更多 >