kraken

1
// Copyright (c) 2016-2019 Uber Technologies, Inc.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
package main
15

16
import (
17
	"bytes"
18
	"fmt"
19
	"io"
20
	"io/ioutil"
21
	"net/http"
22
	"net/http/httputil"
23
	"os/exec"
24
	"strings"
25
	"sync"
26
	"time"
27

28
	"github.com/uber/kraken/utils/errutil"
29
	"github.com/uber/kraken/utils/log"
30

31
	"github.com/docker/distribution"
32
	"github.com/docker/distribution/manifest/schema1"
33
	"github.com/docker/distribution/manifest/schema2"
34
	"github.com/opencontainers/go-digest"
35
)
36

37
// guessDigest returns digest from the URL.
38
// returns empty string if this push action does not look like a tag.
39
func guessDigest(url string, repo string) string {
40
	p := fmt.Sprintf("/v2/%s/manifests/", repo)
41
	idx := strings.Index(url, p)
42
	if idx < 0 {
43
		return ""
44
	}
45
	return url[idx+len(p):]
46
}
47

48
// PullImage pull images from source registry, it does not check if the file exits
49
func PullImage(source, repo, tag string, useDocker bool) error {
50
	t := time.Now()
51

52
	if useDocker {
53
		log.Info("pulling with docker daemon")
54
		err := exec.Command("docker", "pull", source+"/"+repo+":"+tag).Run()
55
		if err != nil {
56
			return fmt.Errorf("failed to pull image: %s:%s: %s", repo, tag, err.Error())
57
		}
58
		return nil
59
	}
60

61
	log.Info("pulling with http")
62
	manifest, err := pullManifest(http.Client{Timeout: transferTimeout}, source, repo, tag)
63
	if err != nil {
64
		return fmt.Errorf("failed to pull manifest %s:%s: %s", repo, tag, err)
65
	}
66

67
	layerDigests, err := getLayerDigestsFromManifest(&manifest)
68
	if err != nil {
69
		return fmt.Errorf("failed to get layer digests from manifest: %s", err)
70
	}
71

72
	var wg sync.WaitGroup
73
	var mu sync.Mutex
74
	var errs errutil.MultiError
75
	for _, d := range layerDigests {
76
		wg.Add(1)
77
		d := d
78
		go func() {
79
			defer wg.Done()
80
			err := pullLayer(http.Client{Timeout: transferTimeout}, source, repo, d)
81
			if err != nil {
82
				mu.Lock()
83
				defer mu.Unlock()
84
				errs = append(errs, err)
85
				return
86
			}
87
		}()
88
	}
89
	wg.Wait()
90

91
	if errs != nil {
92
		return fmt.Errorf("failed to pull image %s:%s: %s", repo, tag, errs)
93
	}
94

95
	log.Infof("finished pulling image %s:%s in %v", repo, tag, time.Since(t).Seconds())
96
	return nil
97
}
98

99
func pullManifest(client http.Client, source string, name string, reference string) (distribution.Manifest, error) {
100
	manifestURL := fmt.Sprintf(baseManifestQuery, source, name, reference)
101
	req, err := http.NewRequest("GET", manifestURL, bytes.NewReader([]byte{}))
102
	if err != nil {
103
		return nil, err
104
	}
105
	// Add `Accept` header to indicate schema2 is supported
106
	req.Header.Add("Accept", schema2.MediaTypeManifest)
107
	resp, err := client.Do(req)
108

109
	if err != nil {
110
		return nil, err
111
	}
112
	defer resp.Body.Close()
113

114
	if resp.StatusCode == 404 {
115
		return nil, fmt.Errorf("manifest not found")
116
	}
117

118
	if resp.StatusCode != 200 {
119
		return nil, fmt.Errorf("server returned %v", resp.Status)
120
	}
121

122
	version := resp.Header.Get("Content-Type")
123
	body, err := ioutil.ReadAll(resp.Body)
124
	if err != nil {
125
		return nil, err
126
	}
127
	manifest, _, err := distribution.UnmarshalManifest(version, body)
128
	if err != nil {
129
		return nil, err
130
	}
131

132
	return manifest, nil
133
}
134

135
func getLayerDigestsFromManifest(manifest *distribution.Manifest) ([]string, error) {
136
	var digests []string
137
	// Get layers from manifest
138
	switch (*manifest).(type) {
139
	case *schema1.SignedManifest:
140
		fsLayers := (*manifest).(*schema1.SignedManifest).FSLayers
141
		for _, fsLayer := range fsLayers {
142
			digests = append(digests, fsLayer.BlobSum.String())
143
		}
144
		break
145
	case *schema2.DeserializedManifest:
146
		layerDescriptors := (*manifest).(*schema2.DeserializedManifest).Layers
147
		for _, descriptor := range layerDescriptors {
148
			digests = append(digests, descriptor.Digest.String())
149
		}
150
		// for schema2, we also need a config layer
151
		config := (*manifest).(*schema2.DeserializedManifest).Config
152
		digests = append(digests, config.Digest.String())
153
		break
154
	default:
155
		mt, _, err := (*manifest).Payload()
156
		if err == nil {
157
			err = fmt.Errorf("manifest type %s is not supported", mt)
158
		}
159
		return nil, err
160
	}
161

162
	return digests, nil
163
}
164

165
func pullLayer(client http.Client, source, name string, layerDigest string) error {
166
	layerURL := fmt.Sprintf(baseLayerQuery, source, name, layerDigest)
167
	resp, err := client.Get(layerURL)
168
	if err != nil {
169
		return err
170
	}
171

172
	defer resp.Body.Close()
173

174
	if resp.StatusCode != 200 {
175
		respDump, errDump := httputil.DumpResponse(resp, true)
176
		if errDump != nil {
177
			return errDump
178
		}
179
		return fmt.Errorf("failed to pull layer: %s", respDump)
180
	}
181

182
	ok, err := verifyLayer(digest.Digest(layerDigest), resp.Body)
183
	if err != nil {
184
		return fmt.Errorf("failed to verfiy layer: %s", err)
185
	}
186

187
	if !ok {
188
		return fmt.Errorf("failed to verify layer: layer digest does not match to the content")
189
	}
190

191
	return nil
192
}
193

194
func verifyLayer(layerDigest digest.Digest, r io.Reader) (bool, error) {
195
	v := layerDigest.Verifier()
196

197
	if _, err := io.Copy(v, r); err != nil {
198
		return false, err
199
	}
200

201
	return v.Verified(), nil
202
}
203