def registry_get(self, api):
url = "https://%s/v2/%s" % (self.registry, api)
response = get(url, auth=(self.user, self.password),
headers={"Accept": 'application/vnd.docker.distribution.manifest.v2+json'})
if response.status_code == 401:
challenge = response.headers['Www-Authenticate']
if challenge.startswith("Bearer "):
challenge = challenge[7:]
opts = urllib2.parse_keqv_list(urllib2.parse_http_list(challenge))
authresp = get("{realm}?service={service}&scope={scope}".format(**opts), auth=(self.user, self.password))
if authresp.ok:
token = authresp.json()['token']
response = get(url, headers={'Authorization': 'Bearer %s' % token})
else:
raise TaskError("problem authenticating with docker registry: [%s] %s" % (authresp.status_code,
authresp.content))
return response
python类parse_keqv_list()的实例源码
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_keqv_list(l):
"""A unicode-safe version of urllib2.parse_keqv_list"""
# With Python 2.6, parse_http_list handles unicode fine
return urllib2.parse_keqv_list(l)
def parse_authorization_header(authorization_header):
"""Parse an OAuth authorization header into a list of 2-tuples"""
auth_scheme = 'OAuth '.lower()
if authorization_header[:len(auth_scheme)].lower().startswith(auth_scheme):
items = parse_http_list(authorization_header[len(auth_scheme):])
try:
return list(parse_keqv_list(items).items())
except (IndexError, ValueError):
pass
raise ValueError('Malformed authorization header')
def parse_digest_challenge(self, challenge_string):
return urllib2.parse_keqv_list(urllib2.parse_http_list(challenge_string))
