download_docker_image.py

import json
from requests import get

REGISTRY_URL = 'https://registry.hub.docker.com/v2'
NAME = 'library/ubuntu'
REFERENCE = 'latest'

def parse_www_authenticate(s):
    objs = [obj.split('=') for obj in s.split()[1].split(',')]
    return {obj[0]: obj[1][1:-1] for obj in objs}

# ==============================================================================

print('=== Request manifests without token ===')
res = get(REGISTRY_URL + '/%s/manifests/%s' % (NAME, REFERENCE))

print(' => Resigry should return 401 Unauthorized')
assert res.status_code == 401
print(' => Docker-Distribution-Api-Version is registry/2.0')
assert res.headers['Docker-Distribution-Api-Version'] == 'registry/2.0'

params = parse_www_authenticate(res.headers['Www-Authenticate'])
realm = params['realm']
scope = params['scope']
service = params['service']
print(' => realm: %s' % realm)
print(' => scope: %s' % scope)
print(' => service: %s' % service)

# ==============================================================================

print('=== Get token from auth.docker.io ===')
res = get(realm + '?service=%s&scope=%s' % (service, scope))

print(' => Resigry should return 200 OK')
assert res.status_code == 200

token = res.json()['token']
print(' => token: %s' % (token[:32] + '...'))

headers = {'Authorization': 'Bearer %s' % token}

# ==============================================================================

print('=== Request manifests ===')
res = get(REGISTRY_URL + '/%s/manifests/%s' % (NAME, REFERENCE), headers=headers)

print(' => Resigry should return 200')
assert res.status_code == 200
print(' => Docker-Distribution-Api-Version is registry/2.0')
assert res.headers['Docker-Distribution-Api-Version'] == 'registry/2.0'

layers = res.json()['fsLayers']

# ==============================================================================

for layer in layers:
    shasum = layer['blobSum']

    print('=== Request fsLayers %s ===' % shasum)
    res = get(REGISTRY_URL + '/%s/blobs/%s' % (NAME, shasum), headers=headers)

    print(' => Resigry should return 200')
    assert res.status_code == 200
    print(' => Content-Type is application/octet-stream')
    assert res.headers['Content-Type'] == 'application/octet-stream'

    with open('%s.tar.gz' % shasum, 'wb') as f:
        f.write(res.content)