获取 ACR 的 Azure Active Directory 刷新令牌

Get Azure Active Directory refresh token for ACR

我一直在尝试使用 AAD 交互式身份验证流程来列出给定 Azure 容器注册表实例中的图像(详见 here),这需要 AAD 访问令牌和 AAD刷新令牌以获取资源本身的 refresh/access 个令牌。

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

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

其中 returns 一个普通访问令牌(例如“eyJ0eX...”)。但是,我找不到任何以编程方式获取刷新令牌的参考资料;似乎所有 Azure SDK 都会在后台处理这个问题。有什么方法可以获取此令牌,因为流向资源本身 (ACR) 的身份验证流程需要两个令牌?

如果有任何其他编程方式可以在没有刷新令牌的情况下列出容器映像,我将非常乐意试一试。

同样重要的是要注意 - 我在 Azure CLI 无法正常工作的环境中工作,所以我必须使用 python SDK 或类似工具,而不是依赖系统命令,例如 'az login' 或'az get credentials'.

所以我对 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))