Блог

Я пытался использовать поток интерактивной аутентификации AAD для перечисления изображений в данном экземпляре реестра контейнеров Azure (как описано здесь), для которого требуется как токен доступа 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…»). Однако я не смог найти никаких ссылок на получение токена обновления программно; Похоже, что все SDK Azure заботятся об этом в фоновом режиме. Есть ли какой-либо способ получить этот токен, поскольку для потока аутентификации самого ресурса (ACR) требуются оба токена?

Если существует какой-либо другой программный способ перечисления изображений контейнеров без маркера обновления, я буду более чем счастлив попробовать.

Также важно отметить — я работаю в среде, где Azure CLI не функционирует, поэтому я должен использовать SDK python или аналогичные инструменты и не рассчитывать на системные команды, такие как «вход в систему az» или «получение учетных данных az».

Поэтому у меня возникло несколько недоразумений относительно функциональности токенов обновления и доступа с ресурсами 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))