1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
# Copyright 2013 dotCloud inc.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import base64
import json
import os
import six
import docker.utils as utils
INDEX_URL = 'https://index.docker.io/v1/'
def swap_protocol(url):
if url.startswith('http://'):
return url.replace('http://', 'https://', 1)
if url.startswith('https://'):
return url.replace('https://', 'http://', 1)
return url
def expand_registry_url(hostname):
if hostname.startswith('http:') or hostname.startswith('https:'):
if '/' not in hostname[9:]:
hostname = hostname + '/v1/'
return hostname
if utils.ping('https://' + hostname + '_ping'):
return 'https://' + hostname + '/v1/'
return 'http://' + hostname + '/v1/'
def resolve_repository_name(repo_name):
if '://' in repo_name:
raise ValueError('Repository name cannot contain a '
'scheme ({0})'.format(repo_name))
parts = repo_name.split('/', 1)
if not '.' in parts[0] and not ':' in parts[0] and parts[0] != 'localhost':
# This is a docker index repo (ex: foo/bar or ubuntu)
return INDEX_URL, repo_name
if len(parts) < 2:
raise ValueError('Invalid repository name ({0})'.format(repo_name))
if 'index.docker.io' in parts[0]:
raise ValueError('Invalid repository name,'
'try "{0}" instead'.format(parts[1]))
return expand_registry_url(parts[0]), parts[1]
def resolve_authconfig(authconfig, registry):
default = {}
if registry == INDEX_URL or registry == '':
# default to the index server
return authconfig['Configs'].get(INDEX_URL, default)
# if its not the index server there are three cases:
#
# 1. this is a full config url -> it should be used as is
# 2. it could be a full url, but with the wrong protocol
# 3. it can be the hostname optionally with a port
#
# as there is only one auth entry which is fully qualified we need to start
# parsing and matching
if '/' not in registry:
registry = registry + '/v1/'
if not registry.startswith('http:') and not registry.startswith('https:'):
registry = 'https://' + registry
if registry in authconfig['Configs']:
return authconfig['Configs'][registry]
elif swap_protocol(registry) in authconfig['Configs']:
return authconfig['Configs'][swap_protocol(registry)]
return default
def decode_auth(auth):
if isinstance(auth, six.string_types):
auth = auth.encode('ascii')
s = base64.b64decode(auth)
login, pwd = s.split(b':')
return login, pwd
def encode_header(auth):
auth_json = json.dumps(auth)
return base64.b64encode(auth_json)
def load_config(root=None):
root = root or os.environ['HOME']
config = {
'Configs': {},
'rootPath': root
}
config_file = os.path.join(root, '.dockercfg')
if not os.path.exists(config_file):
return config
with open(config_file) as f:
try:
config['Configs'] = json.load(f)
for k, conf in six.iteritems(config['Configs']):
conf['Username'], conf['Password'] = decode_auth(conf['auth'])
del conf['auth']
config['Configs'][k] = conf
except Exception:
f.seek(0)
buf = []
for line in f:
k, v = line.split(' = ')
buf.append(v)
if len(buf) < 2:
raise Exception("The Auth config file is empty")
user, pwd = decode_auth(buf[0])
config['Configs'][INDEX_URL] = {
'Username': user,
'Password': pwd,
'Email': buf[1]
}
return config
|
