1// SPDX-License-Identifier: MIT
2
3package registryproxy
4
5import (
6 "context"
7 "errors"
8 "io"
9 "net/http"
10 "sync"
11 "time"
12)
13
14// upstreamClients returns the metadata client (auth, manifests: a total
15// timeout and no redirects) and the blob client (no total timeout, a response
16// header timeout, and redirects that drop the bearer when leaving the host).
17func upstreamClients(requestTimeout time.Duration) (metadata, blobs *http.Client) {
18 metadata = &http.Client{
19 Timeout: requestTimeout,
20 CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
21 }
22 blobs = &http.Client{
23 CheckRedirect: func(req *http.Request, via []*http.Request) error {
24 if len(via) >= 10 {
25 return errors.New("too many registry redirects")
26 }
27 if len(via) > 0 && req.URL.Host != via[0].URL.Host {
28 req.Header.Del("Authorization")
29 }
30 return nil
31 },
32 }
33 if tr, ok := http.DefaultTransport.(*http.Transport); ok {
34 tr = tr.Clone()
35 tr.ResponseHeaderTimeout = requestTimeout
36 blobs.Transport = tr
37 }
38 return metadata, blobs
39}
40
41// idleBody serializes timer reset/stop against expiry. Cancellation interrupts
42// even a blocked transport read; only completed, nonempty reads renew the lease.
43type idleBody struct {
44 io.ReadCloser
45 mu sync.Mutex
46 timer *time.Timer
47 cancel context.CancelFunc
48 idle time.Duration
49 expired bool
50}
51
52func (b *idleBody) Read(p []byte) (int, error) {
53 n, err := b.ReadCloser.Read(p)
54 if n > 0 {
55 b.mu.Lock()
56 if !b.expired {
57 b.timer.Reset(b.idle)
58 }
59 b.mu.Unlock()
60 }
61 return n, err
62}
63
64func (b *idleBody) Close() error {
65 b.mu.Lock()
66 b.expired = true
67 b.timer.Stop()
68 b.mu.Unlock()
69 b.cancel()
70 return b.ReadCloser.Close()
71}
72
73// blob GETs a blob. Each attempt has only a header deadline (RequestTimeout)
74// and then a body progress deadline (BlobIdleTimeout), and no total transfer
75// deadline: the response body is an idleBody that owns the attempt's context.
76func (u *upstream) blob(ctx context.Context, name, digest string) (*http.Response, error) {
77 return u.authorized(ctx, name, func(tok string) (*http.Response, error) {
78 bodyCtx, cancel := context.WithCancel(ctx)
79 lease := &idleBody{cancel: cancel, idle: u.blobIdle}
80 lease.timer = time.AfterFunc(u.requestTimeout, func() { lease.mu.Lock(); lease.expired = true; cancel(); lease.mu.Unlock() })
81 req, err := http.NewRequestWithContext(bodyCtx, http.MethodGet, u.base+"/v2/"+name+"/"+kindBlob.dir()+"/"+digest, nil)
82 if err != nil {
83 lease.timer.Stop()
84 cancel()
85 return nil, err
86 }
87 req.Header.Set("Authorization", "Bearer "+tok)
88 resp, err := u.blobClient.Do(req)
89 if err != nil {
90 lease.timer.Stop()
91 cancel()
92 return nil, err
93 }
94 lease.ReadCloser = resp.Body
95 resp.Body = lease
96 if resp.StatusCode == http.StatusOK {
97 lease.mu.Lock()
98 if !lease.expired {
99 lease.timer.Reset(u.blobIdle)
100 }
101 lease.mu.Unlock()
102 }
103 return resp, nil
104 })
105}