1// SPDX-License-Identifier: MIT
2
3// Package registryproxy provides an anonymous, pull-only Docker Hub mirror.
4// Its handler belongs on a sandbox's private runtime channel, not a public port.
5package registryproxy
6
7import (
8 "context"
9 "encoding/hex"
10 "errors"
11 "io"
12 "net/http"
13 "net/url"
14 "regexp"
15 "strconv"
16 "strings"
17 "time"
18
19 "github.com/bnema/zerowrap"
20)
21
22const manifestLimit = 4 << 20
23
24// Config configures a Proxy. Max must be positive. The directory is exclusively
25// owned by one Proxy; opening it clears incomplete downloads from tmp/.
26type Config struct {
27 Dir string
28 Max int64
29 Upstream string // default https://registry-1.docker.io
30 AuthURL string // default https://auth.docker.io/token
31 Log zerowrap.Logger
32 // RequestTimeout bounds auth, manifests and HEAD (default 10s).
33 RequestTimeout time.Duration
34 // BlobIdleTimeout bounds time without upstream body progress (default 30s).
35 // A blob body has no total timeout.
36 BlobIdleTimeout time.Duration
37 // KeepAliveInterval is how often a client waiting for a cold blob is kept
38 // alive while the download progresses (default 1s), when its
39 // ResponseWriter is a KeepAliver.
40 KeepAliveInterval time.Duration
41 // AbandonGrace is how long a download nobody waits for any more keeps
42 // running in case another request joins it; then it is cancelled and its
43 // temporary file removed (default 5s).
44 AbandonGrace time.Duration
45 // MaxDownloads caps the concurrent cold blob downloads across all jobs
46 // (default 16). A request may also carry a per-client Quota (WithQuota).
47 MaxDownloads int
48 // SlotWait is how long a request that would start a download waits for a
49 // free slot, global or of its quota, before it gets 503 (default 30s).
50 SlotWait time.Duration
51}
52
53// KeepAliver is a ResponseWriter that can move its connection's deadlines
54// without writing. While a request waits for a cold blob, the Proxy calls
55// KeepAlive every KeepAliveInterval as long as the download progresses, so
56// that a slow but moving download does not time the client out before the
57// first byte of the reply.
58type KeepAliver interface {
59 KeepAlive()
60}
61
62// Proxy implements http.Handler for GET and HEAD under /v2/.
63type Proxy struct {
64 s *store
65 u *upstream
66 log zerowrap.Logger
67 keepEvery time.Duration
68}
69
70// New opens the persistent cache. It starts no goroutines or listeners.
71func New(c Config) (*Proxy, error) {
72 if c.Upstream == "" {
73 c.Upstream = "https://registry-1.docker.io"
74 }
75 if c.AuthURL == "" {
76 c.AuthURL = "https://auth.docker.io/token"
77 }
78 for _, s := range []string{c.Upstream, c.AuthURL} {
79 u, err := url.Parse(s)
80 if err != nil || u.Host == "" || u.User != nil || (u.Scheme != "http" && u.Scheme != "https") {
81 return nil, errors.New("invalid registry upstream URL")
82 }
83 }
84 if c.RequestTimeout <= 0 {
85 c.RequestTimeout = 10 * time.Second
86 }
87 if c.BlobIdleTimeout <= 0 {
88 c.BlobIdleTimeout = 30 * time.Second
89 }
90 if c.KeepAliveInterval <= 0 {
91 c.KeepAliveInterval = time.Second
92 }
93 if c.AbandonGrace <= 0 {
94 c.AbandonGrace = 5 * time.Second
95 }
96 if c.MaxDownloads <= 0 {
97 c.MaxDownloads = 16
98 }
99 if c.SlotWait <= 0 {
100 c.SlotWait = 30 * time.Second
101 }
102 st, err := openStore(storeConfig{dir: c.Dir, max: c.Max, now: time.Now, grace: c.AbandonGrace,
103 maxDownloads: c.MaxDownloads, slotWait: c.SlotWait, log: c.Log})
104 if err != nil {
105 return nil, err
106 }
107 client, blobs := upstreamClients(c.RequestTimeout)
108 return &Proxy{s: st, u: &upstream{base: strings.TrimRight(c.Upstream, "/"), auth: c.AuthURL, client: client, blobClient: blobs,
109 requestTimeout: c.RequestTimeout, blobIdle: c.BlobIdleTimeout, tokens: map[string]bearer{}}, log: c.Log, keepEvery: c.KeepAliveInterval}, nil
110}
111
112var namePattern = regexp.MustCompile(`^[a-z0-9]+(?:[._-][a-z0-9]+)*(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*$`)
113var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`)
114
115func validDigest(s string) bool {
116 if len(s) != 71 || !strings.HasPrefix(s, "sha256:") {
117 return false
118 }
119 _, err := hex.DecodeString(s[7:])
120 return err == nil && s == strings.ToLower(s)
121}
122
123func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
124 if r.Method != http.MethodGet && r.Method != http.MethodHead {
125 w.WriteHeader(http.StatusMethodNotAllowed)
126 return
127 }
128 w.Header().Set("Docker-Distribution-Api-Version", "registry/2.0")
129 if r.URL.Path == "/v2/" {
130 w.Header().Set("Content-Type", "application/json")
131 w.Header().Set("Content-Length", "2")
132 if r.Method == http.MethodGet {
133 _, _ = io.WriteString(w, "{}")
134 }
135 return
136 }
137 if !strings.HasPrefix(r.URL.Path, "/v2/") {
138 http.NotFound(w, r)
139 return
140 }
141 path := strings.TrimPrefix(r.URL.Path, "/v2/")
142 i := strings.LastIndex(path, "/")
143 if i < 0 {
144 http.NotFound(w, r)
145 return
146 }
147 ref := path[i+1:]
148 path = path[:i]
149 i = strings.LastIndex(path, "/")
150 if i < 0 {
151 http.NotFound(w, r)
152 return
153 }
154 name, route := path[:i], path[i+1:]
155 // dockerd's legacy client sends one Accept line per media type.
156 accept := strings.Join(r.Header.Values("Accept"), ",")
157 if len(name) > 255 || !namePattern.MatchString(name) || len(accept) > 8192 {
158 http.NotFound(w, r)
159 return
160 }
161 if route == kindManifest.dir() && (validDigest(ref) || tagPattern.MatchString(ref)) {
162 p.manifest(w, r, name, ref, normalizeAccept(accept))
163 return
164 }
165 if route == kindBlob.dir() && validDigest(ref) {
166 p.blob(w, r, name, ref)
167 return
168 }
169 http.NotFound(w, r)
170}
171
172// logFor is the Proxy's logger, with the job_id of the request (WithJob).
173func (p *Proxy) logFor(ctx context.Context) zerowrap.Logger {
174 if job := jobFrom(ctx); job != "" {
175 return zerowrap.Logger{Logger: p.log.With().Str("job_id", job).Logger()}
176 }
177 return p.log
178}
179
180func serveFile(w http.ResponseWriter, r *http.Request, f *storedFile, media, digest string) {
181 defer func() { _ = f.Close() }()
182 fi, err := f.Stat()
183 if err != nil {
184 http.Error(w, "registry cache read failed", http.StatusBadGateway)
185 return
186 }
187 w.Header().Set("Content-Type", media)
188 w.Header().Set("Docker-Content-Digest", digest)
189 http.ServeContent(w, r, "", fi.ModTime(), f)
190}
191
192// serveStaleTag serves the manifest a tag pointed to when upstream could not
193// answer, if that manifest is still stored. It reports whether it served one.
194func (p *Proxy) serveStaleTag(w http.ResponseWriter, r *http.Request, name, tag, accept string) bool {
195 digest := p.s.tag(name, tag, accept)
196 if !validDigest(digest) {
197 return false
198 }
199 f, media, err := p.s.open(kindManifest, digest)
200 if err != nil {
201 return false
202 }
203 serveFile(w, r, f, media, digest)
204 return true
205}
206
207// manifest serves a manifest by tag or digest. accept is normalized.
208func (p *Proxy) manifest(w http.ResponseWriter, r *http.Request, name, ref, accept string) {
209 start := time.Now()
210 hit := false
211 log := p.logFor(r.Context())
212 defer func() {
213 log.Info().Str("repo", name).Str("ref", ref).Bool("hit", hit).Int64("ms", time.Since(start).Milliseconds()).Msg("registry manifest")
214 }()
215 ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
216 defer cancel()
217 digest := ref
218 tag := !validDigest(ref)
219 if tag {
220 resp, err := p.u.request(ctx, http.MethodHead, name, kindManifest, ref, accept)
221 if err != nil {
222 if unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
223 hit = true
224 return
225 }
226 http.Error(w, "registry upstream unavailable", proxyStatus(err))
227 return
228 }
229 digest = resp.Header.Get("Docker-Content-Digest")
230 _ = resp.Body.Close()
231 if !validDigest(digest) {
232 http.Error(w, "invalid upstream digest", http.StatusBadGateway)
233 return
234 }
235 }
236 // Shared flight, as for blobs: it must not end with the initiator's request.
237 f, media, cached, err := p.s.ensure(ctx, kindManifest, digest, func(flightCtx context.Context, dst io.Writer) (string, error) {
238 fillCtx, fillCancel := context.WithTimeout(flightCtx, p.u.requestTimeout)
239 defer fillCancel()
240 resp, err := p.u.request(fillCtx, http.MethodGet, name, kindManifest, digest, accept)
241 if err != nil {
242 return "", err
243 }
244 defer func() { _ = resp.Body.Close() }()
245 b, err := io.ReadAll(io.LimitReader(resp.Body, manifestLimit+1))
246 if err != nil {
247 return "", err
248 }
249 if len(b) > manifestLimit {
250 return "", errTooLarge
251 }
252 _, err = dst.Write(b)
253 return resp.Header.Get("Content-Type"), err
254 }, keepalive{})
255 hit = cached
256 if err != nil {
257 if tag && unavailable(err) && p.serveStaleTag(w, r, name, ref, accept) {
258 hit = true
259 return
260 }
261 http.Error(w, "registry fetch failed", proxyStatus(err))
262 return
263 }
264 if tag {
265 _ = p.s.setTag(name, ref, accept, digest)
266 }
267 serveFile(w, r, f, media, digest)
268}
269
270func (p *Proxy) blob(w http.ResponseWriter, r *http.Request, name, digest string) {
271 if f, media, err := p.s.open(kindBlob, digest); err == nil {
272 serveFile(w, r, f, media, digest)
273 return
274 }
275 // A cold HEAD is metadata-only and retains the short total timeout.
276 if r.Method == http.MethodHead {
277 ctx, cancel := context.WithTimeout(r.Context(), p.u.requestTimeout)
278 defer cancel()
279 resp, err := p.u.request(ctx, http.MethodHead, name, kindBlob, digest, "")
280 if err != nil {
281 http.Error(w, "registry fetch failed", proxyStatus(err))
282 return
283 }
284 defer func() { _ = resp.Body.Close() }()
285 if resp.ContentLength < 0 {
286 http.Error(w, "unknown blob length", http.StatusBadGateway)
287 return
288 }
289 w.Header().Set("Content-Type", "application/octet-stream")
290 w.Header().Set("Content-Length", strconv.FormatInt(resp.ContentLength, 10))
291 w.Header().Set("Docker-Content-Digest", digest)
292 return
293 }
294 // The flight is shared by every client that asks for this blob, so no client
295 // may own it: it downloads on a context of its own, into the temporary file
296 // only, and is bounded by Max and the upstream idle timeout. Every client,
297 // the initiator included, waits for the verified commit and then reads the
298 // stored file (simplest correct option: no client streams unverified bytes,
299 // and a client that disconnects fails only itself). The flight is cancelled
300 // when no client has waited for it for AbandonGrace. At most MaxDownloads
301 // run at once, and at most its Quota per client: a request over either
302 // waits up to SlotWait for a slot, then gets 503, which dockerd retries
303 // (about 50 s) before pulling from Docker Hub. While it waits, a
304 // KeepAliver writer (the cache channel's progress deadline) is kept alive
305 // as long as the download moves.
306 keep := keepalive{every: p.keepEvery}
307 if k, ok := w.(KeepAliver); ok {
308 keep.fn = k.KeepAlive
309 }
310 log := p.logFor(r.Context())
311 f, media, _, err := p.s.ensure(r.Context(), kindBlob, digest, func(downloadCtx context.Context, dst io.Writer) (string, error) {
312 start := time.Now()
313 var size int64
314 defer func() {
315 log.Debug().Str("digest", digest).Int64("bytes", size).Int64("ms", time.Since(start).Milliseconds()).Msg("registry blob")
316 }()
317 resp, err := p.u.blob(downloadCtx, name, digest)
318 if err != nil {
319 return "", err
320 }
321 defer func() { _ = resp.Body.Close() }()
322 if resp.ContentLength > p.s.max {
323 return "", errTooLarge
324 }
325 size, err = io.Copy(dst, resp.Body)
326 return "application/octet-stream", err
327 }, keep)
328 if errors.Is(err, errBusy) {
329 w.Header().Set("Retry-After", "1")
330 http.Error(w, "registry downloads busy", http.StatusServiceUnavailable)
331 return
332 }
333 if err != nil {
334 http.Error(w, "registry fetch failed", proxyStatus(err))
335 return
336 }
337 serveFile(w, r, f, media, digest)
338}