1package registrypush
2
3import (
4 "bytes"
5 "context"
6 "crypto/sha256"
7 "crypto/tls"
8 "encoding/base64"
9 "errors"
10 "fmt"
11 "io"
12 "net/http"
13 "net/url"
14 "os"
15 "strings"
16 "time"
17
18 "github.com/google/go-containerregistry/pkg/authn"
19 "github.com/google/go-containerregistry/pkg/name"
20 v1 "github.com/google/go-containerregistry/pkg/v1"
21 "github.com/google/go-containerregistry/pkg/v1/tarball"
22 "github.com/moby/moby/client"
23)
24
25// DefaultChunkSize is 50MB - safely under Cloudflare's 100MB per-request limit
26// and Gordon's default max_blob_chunk_size of 95MB.
27const DefaultChunkSize = 50 * 1024 * 1024
28
29const defaultHTTPTimeout = 5 * time.Minute
30
31// ImageSource provides a v1.Image for a given ref.
32// The default reads from the local Docker/Podman daemon.
33// Override with WithImageSource for testing.
34type ImageSource func(ctx context.Context, ref string) (v1.Image, error)
35
36// Pusher uploads container images to a registry using chunked blob uploads.
37type Pusher struct {
38 chunkSize int64
39 timeout time.Duration
40 transport http.RoundTripper
41 insecureTLS bool
42 plainHTTP bool
43 imageSource func(ctx context.Context, ref string) (v1.Image, func(), error)
44 cleanupFn func()
45 progress io.Writer
46 authHeader string
47}
48
49// Option configures a Pusher.
50type Option func(*Pusher)
51
52// WithChunkSize sets the maximum size of each blob upload chunk.
53func WithChunkSize(size int64) Option {
54 return func(p *Pusher) { p.chunkSize = size }
55}
56
57// WithTransport sets a custom HTTP transport (for testing).
58func WithTransport(t http.RoundTripper) Option {
59 return func(p *Pusher) { p.transport = t }
60}
61
62// WithTimeout sets the HTTP client timeout for registry operations.
63func WithTimeout(d time.Duration) Option {
64 return func(p *Pusher) { p.timeout = d }
65}
66
67// WithInsecureTLS disables TLS certificate verification for registry requests.
68// This is ignored when a custom transport is provided via WithTransport.
69func WithInsecureTLS(insecure bool) Option {
70 return func(p *Pusher) { p.insecureTLS = insecure }
71}
72
73// WithPlainHTTP forces the registry base URL to use http:// instead of
74// https://. This is independent from WithInsecureTLS, which only disables
75// TLS certificate verification while still using https://.
76func WithPlainHTTP(plain bool) Option {
77 return func(p *Pusher) { p.plainHTTP = plain }
78}
79
80// WithImageSource overrides the default daemon-based image reader.
81func WithImageSource(src ImageSource) Option {
82 return func(p *Pusher) {
83 p.imageSource = func(ctx context.Context, ref string) (v1.Image, func(), error) {
84 img, err := src(ctx, ref)
85 return img, nil, err
86 }
87 }
88}
89
90// WithImageCleanup sets a function called after Push completes to clean up
91// resources created by the image source (e.g., temp tar files).
92func WithImageCleanup(fn func()) Option {
93 return func(p *Pusher) { p.cleanupFn = fn }
94}
95
96// WithProgress sets a writer for progress messages.
97func WithProgress(w io.Writer) Option {
98 return func(p *Pusher) { p.progress = w }
99}
100
101// WithAuth sets an explicit Authorization header value (e.g. "Bearer <token>")
102// to use for all registry requests. When set, the Docker credential keychain
103// is not consulted. This is useful when a pre-computed token is available
104// without requiring a local docker login.
105func WithAuth(header string) Option {
106 return func(p *Pusher) { p.authHeader = header }
107}
108
109// New creates a Pusher with the given options.
110func New(opts ...Option) *Pusher {
111 p := &Pusher{chunkSize: DefaultChunkSize, timeout: defaultHTTPTimeout}
112 for _, o := range opts {
113 o(p)
114 }
115 if p.transport == nil {
116 if p.insecureTLS {
117 p.transport = insecureTransport()
118 } else {
119 p.transport = http.DefaultTransport
120 }
121 }
122 if p.imageSource == nil {
123 p.imageSource = func(ctx context.Context, ref string) (v1.Image, func(), error) {
124 return defaultImageSourceWithCleanup(ctx, ref)
125 }
126 }
127 return p
128}
129
130// Push reads a local image (by ref) from the Docker/Podman daemon and uploads
131// it to the registry encoded in the ref, using chunked blob uploads.
132//
133// The ref must be fully qualified: registry.example.com/name:tag
134func (p *Pusher) Push(ctx context.Context, ref string) error {
135 parsedRef, err := name.ParseReference(ref)
136 if err != nil {
137 return fmt.Errorf("failed to parse image ref %s: %w", ref, err)
138 }
139
140 tag, ok := parsedRef.(name.Tag)
141 if !ok {
142 return fmt.Errorf("image ref %s must include a tag", ref)
143 }
144
145 var authHeader string
146 if p.authHeader != "" {
147 authHeader = p.authHeader
148 } else {
149 var resolveErr error
150 authHeader, resolveErr = resolveAuthHeader(parsedRef)
151 if resolveErr != nil {
152 return resolveErr
153 }
154 }
155
156 img, sourceCleanup, err := p.imageSource(ctx, ref)
157 if err != nil {
158 return err
159 }
160 userCleanup := p.cleanupFn
161 switch {
162 case sourceCleanup != nil && userCleanup != nil:
163 defer func() {
164 sourceCleanup()
165 userCleanup()
166 }()
167 case sourceCleanup != nil:
168 defer sourceCleanup()
169 case userCleanup != nil:
170 defer userCleanup()
171 }
172
173 baseURL := p.registryBaseURL(parsedRef.Context().RegistryStr())
174 repo := parsedRef.Context().RepositoryStr()
175
176 if err := p.uploadImageLayers(ctx, img, baseURL, repo, authHeader); err != nil {
177 return err
178 }
179
180 if err := p.uploadImageConfig(ctx, img, baseURL, repo, authHeader); err != nil {
181 return err
182 }
183
184 return p.uploadManifest(ctx, baseURL, repo, tag.Identifier(), img, authHeader)
185}
186
187// UploadBlob uploads a single blob to the registry using chunked uploads.
188// If the blob already exists, it is skipped.
189func (p *Pusher) UploadBlob(ctx context.Context, baseURL, repo, digest string, size int64, content io.Reader, auth string) error {
190 if err := p.checkBlobExists(ctx, baseURL, repo, digest, auth); err == nil {
191 return nil
192 } else if err != errBlobMissing {
193 return err
194 }
195
196 uploadURL, err := p.startBlobUpload(ctx, baseURL, repo, auth)
197 if err != nil {
198 return err
199 }
200
201 chunks := chunkCount(size, p.chunkSize)
202 if p.progress != nil {
203 if _, err := fmt.Fprintf(p.progress, "Uploading blob %s (%d bytes, %d chunks)\n", digest, size, chunks); err != nil {
204 return fmt.Errorf("failed to write upload progress: %w", err)
205 }
206 }
207
208 uploadURL, err = p.uploadBlobChunks(ctx, baseURL, uploadURL, content, auth)
209 if err != nil {
210 return err
211 }
212
213 return p.finalizeBlobUpload(ctx, baseURL, uploadURL, digest, auth)
214}
215
216// defaultImageSourceWithCleanup reads an image from the local Docker/Podman daemon.
217func defaultImageSourceWithCleanup(ctx context.Context, ref string) (v1.Image, func(), error) {
218 cli, err := client.New(client.FromEnv)
219 if err != nil {
220 return nil, nil, fmt.Errorf("failed to create Docker client: %w", err)
221 }
222 defer cli.Close()
223
224 tag, err := name.NewTag(ref)
225 if err != nil {
226 return nil, nil, fmt.Errorf("failed to parse image ref %s: %w", ref, err)
227 }
228
229 rc, err := cli.ImageSave(ctx, []string{ref})
230 if err != nil {
231 return nil, nil, fmt.Errorf("failed to export image %s: %w", ref, err)
232 }
233
234 tmpFile, err := os.CreateTemp("", "gordon-push-*.tar")
235 if err != nil {
236 if closeErr := rc.Close(); closeErr != nil {
237 return nil, nil, fmt.Errorf("failed to create temp file for image export: %w (also failed to close image export stream: %v)", err, closeErr)
238 }
239 return nil, nil, fmt.Errorf("failed to create temp file for image export: %w", err)
240 }
241 tmpPath := tmpFile.Name()
242
243 if _, err := io.Copy(tmpFile, rc); err != nil {
244 if closeErr := rc.Close(); closeErr != nil {
245 _ = tmpFile.Close()
246 _ = os.Remove(tmpPath)
247 return nil, nil, fmt.Errorf("failed to write image tar to temp file: %w (also failed to close image export stream: %v)", err, closeErr)
248 }
249 if closeErr := tmpFile.Close(); closeErr != nil {
250 _ = os.Remove(tmpPath)
251 return nil, nil, fmt.Errorf("failed to write image tar to temp file: %w (also failed to close temp file: %v)", err, closeErr)
252 }
253 if removeErr := os.Remove(tmpPath); removeErr != nil {
254 return nil, nil, fmt.Errorf("failed to write image tar to temp file: %w (also failed to remove temp file %s: %v)", err, tmpPath, removeErr)
255 }
256 return nil, nil, fmt.Errorf("failed to write image tar to temp file: %w", err)
257 }
258 if err := rc.Close(); err != nil {
259 _ = tmpFile.Close()
260 _ = os.Remove(tmpPath)
261 return nil, nil, fmt.Errorf("failed to close image export stream: %w", err)
262 }
263 if err := tmpFile.Close(); err != nil {
264 _ = os.Remove(tmpPath)
265 return nil, nil, fmt.Errorf("failed to close temp file %s: %w", tmpPath, err)
266 }
267
268 img, err := tarball.ImageFromPath(tmpPath, &tag)
269 if err != nil {
270 _ = os.Remove(tmpPath)
271 return nil, nil, fmt.Errorf("failed to parse image tar for %s: %w", ref, err)
272 }
273 cleanup := func() { _ = os.Remove(tmpPath) }
274 return img, cleanup, nil
275}
276
277func (p *Pusher) uploadImageLayers(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
278 layers, err := img.Layers()
279 if err != nil {
280 return fmt.Errorf("failed to read image layers: %w", err)
281 }
282
283 for _, layer := range layers {
284 if err := p.uploadLayer(ctx, baseURL, repo, layer, auth); err != nil {
285 return err
286 }
287 }
288
289 return nil
290}
291
292func (p *Pusher) uploadLayer(ctx context.Context, baseURL, repo string, layer v1.Layer, auth string) error {
293 digest, err := layer.Digest()
294 if err != nil {
295 return fmt.Errorf("failed to get layer digest: %w", err)
296 }
297
298 size, err := layer.Size()
299 if err != nil {
300 return fmt.Errorf("failed to get layer size for %s: %w", digest, err)
301 }
302
303 content, err := layer.Compressed()
304 if err != nil {
305 return fmt.Errorf("failed to read layer %s: %w", digest, err)
306 }
307 defer content.Close()
308
309 if err := p.UploadBlob(ctx, baseURL, repo, digest.String(), size, content, auth); err != nil {
310 return fmt.Errorf("failed to upload layer %s: %w", digest, err)
311 }
312
313 return nil
314}
315
316func (p *Pusher) uploadImageConfig(ctx context.Context, img v1.Image, baseURL, repo, auth string) error {
317 config, err := img.RawConfigFile()
318 if err != nil {
319 return fmt.Errorf("failed to read image config: %w", err)
320 }
321
322 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(config))
323 if err := p.UploadBlob(ctx, baseURL, repo, digest, int64(len(config)), bytesReader(config), auth); err != nil {
324 return fmt.Errorf("failed to upload image config %s: %w", digest, err)
325 }
326
327 return nil
328}
329
330func (p *Pusher) uploadManifest(ctx context.Context, baseURL, repo, tag string, img v1.Image, auth string) error {
331 manifest, err := img.RawManifest()
332 if err != nil {
333 return fmt.Errorf("failed to read image manifest: %w", err)
334 }
335
336 mediaType, err := img.MediaType()
337 if err != nil {
338 return fmt.Errorf("failed to read image media type: %w", err)
339 }
340
341 req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s/v2/%s/manifests/%s", baseURL, repo, tag), bytesReader(manifest))
342 if err != nil {
343 return fmt.Errorf("failed to create manifest upload request: %w", err)
344 }
345 setScopedAuth(req, auth, baseURL)
346 req.Header.Set("Content-Type", string(mediaType))
347 req.ContentLength = int64(len(manifest))
348
349 resp, err := p.httpClient().Do(req)
350 if err != nil {
351 return fmt.Errorf("failed to upload manifest: %w", err)
352 }
353 defer resp.Body.Close()
354
355 if resp.StatusCode != http.StatusCreated {
356 body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
357 return fmt.Errorf("manifest upload returned status %d: %s", resp.StatusCode, string(body))
358 }
359
360 return nil
361}
362
363func resolveAuthHeader(ref name.Reference) (string, error) {
364 authenticator, err := authn.DefaultKeychain.Resolve(ref.Context().Registry)
365 if err != nil {
366 return "", fmt.Errorf("failed to resolve registry credentials for %s: %w", ref.Context().RegistryStr(), err)
367 }
368
369 authConfig, err := authenticator.Authorization()
370 if err != nil {
371 return "", fmt.Errorf("failed to authorize for registry %s: %w", ref.Context().RegistryStr(), err)
372 }
373
374 if authConfig == nil {
375 return "", nil
376 }
377
378 if authConfig.RegistryToken != "" {
379 return "Bearer " + authConfig.RegistryToken, nil
380 }
381
382 if authConfig.Auth != "" {
383 return "Basic " + authConfig.Auth, nil
384 }
385
386 if authConfig.Username != "" || authConfig.Password != "" {
387 encoded := base64.StdEncoding.EncodeToString([]byte(authConfig.Username + ":" + authConfig.Password))
388 return "Basic " + encoded, nil
389 }
390
391 return "", nil
392}
393
394func (p *Pusher) registryBaseURL(host string) string {
395 if p.plainHTTP {
396 return "http://" + host
397 }
398 if strings.HasPrefix(host, "localhost:") || strings.HasPrefix(host, "127.0.0.1:") || host == "localhost" || host == "127.0.0.1" {
399 return "http://" + host
400 }
401 return "https://" + host
402}
403
404var errBlobMissing = errors.New("blob missing")
405
406func (p *Pusher) checkBlobExists(ctx context.Context, baseURL, repo, digest, auth string) error {
407 req, err := http.NewRequestWithContext(ctx, http.MethodHead, fmt.Sprintf("%s/v2/%s/blobs/%s", baseURL, repo, digest), nil)
408 if err != nil {
409 return fmt.Errorf("failed to create blob HEAD request: %w", err)
410 }
411 setScopedAuth(req, auth, baseURL)
412
413 resp, err := p.httpClient().Do(req)
414 if err != nil {
415 return fmt.Errorf("failed to check blob existence: %w", err)
416 }
417 defer resp.Body.Close()
418
419 switch resp.StatusCode {
420 case http.StatusOK:
421 return nil
422 case http.StatusNotFound:
423 return errBlobMissing
424 default:
425 body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
426 return fmt.Errorf("blob existence check returned status %d: %s", resp.StatusCode, string(body))
427 }
428}
429
430func (p *Pusher) startBlobUpload(ctx context.Context, baseURL, repo, auth string) (string, error) {
431 req, err := http.NewRequestWithContext(ctx, http.MethodPost, fmt.Sprintf("%s/v2/%s/blobs/uploads/", baseURL, repo), nil)
432 if err != nil {
433 return "", fmt.Errorf("failed to create blob upload start request: %w", err)
434 }
435 setScopedAuth(req, auth, baseURL)
436
437 resp, err := p.httpClient().Do(req)
438 if err != nil {
439 return "", fmt.Errorf("failed to start blob upload: %w", err)
440 }
441 defer resp.Body.Close()
442
443 if resp.StatusCode != http.StatusAccepted {
444 body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
445 return "", fmt.Errorf("blob upload start returned status %d: %s", resp.StatusCode, string(body))
446 }
447
448 location, err := resolveLocation(baseURL, resp.Header.Get("Location"))
449 if err != nil {
450 return "", fmt.Errorf("failed to resolve blob upload location: %w", err)
451 }
452
453 return location, nil
454}
455
456func (p *Pusher) uploadBlobChunks(ctx context.Context, baseURL, uploadURL string, content io.Reader, auth string) (string, error) {
457 buffer := make([]byte, p.chunkSize)
458 offset := int64(0)
459
460 for {
461 readBytes, err := io.ReadFull(io.LimitReader(content, p.chunkSize), buffer)
462 if err == io.EOF {
463 return uploadURL, nil
464 }
465 if err != nil && err != io.ErrUnexpectedEOF {
466 return "", fmt.Errorf("failed to read blob chunk: %w", err)
467 }
468
469 nextURL, patchErr := p.uploadBlobChunk(ctx, baseURL, uploadURL, buffer[:readBytes], offset, auth)
470 if patchErr != nil {
471 return "", patchErr
472 }
473
474 uploadURL = nextURL
475 offset += int64(readBytes)
476
477 if err == io.ErrUnexpectedEOF {
478 return uploadURL, nil
479 }
480 }
481}
482
483func (p *Pusher) uploadBlobChunk(ctx context.Context, baseURL, uploadURL string, chunk []byte, offset int64, auth string) (string, error) {
484 req, err := http.NewRequestWithContext(ctx, http.MethodPatch, uploadURL, bytesReader(chunk))
485 if err != nil {
486 return "", fmt.Errorf("failed to create blob chunk request: %w", err)
487 }
488 setScopedAuth(req, auth, baseURL)
489 req.Header.Set("Content-Type", "application/octet-stream")
490 req.Header.Set("Content-Range", fmt.Sprintf("%d-%d", offset, offset+int64(len(chunk))-1))
491 req.ContentLength = int64(len(chunk))
492
493 resp, err := p.httpClient().Do(req)
494 if err != nil {
495 return "", fmt.Errorf("failed to upload blob chunk: %w", err)
496 }
497 defer resp.Body.Close()
498
499 if resp.StatusCode != http.StatusAccepted {
500 body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
501 return "", fmt.Errorf("blob chunk upload returned status %d: %s", resp.StatusCode, string(body))
502 }
503
504 nextURL, err := resolveLocation(uploadURL, resp.Header.Get("Location"))
505 if err != nil {
506 return "", fmt.Errorf("failed to resolve next blob upload location: %w", err)
507 }
508
509 return nextURL, nil
510}
511
512func (p *Pusher) finalizeBlobUpload(ctx context.Context, baseURL, uploadURL, digest, auth string) error {
513 parsedURL, err := url.Parse(uploadURL)
514 if err != nil {
515 return fmt.Errorf("failed to parse blob upload location: %w", err)
516 }
517
518 query := parsedURL.Query()
519 query.Set("digest", digest)
520 parsedURL.RawQuery = query.Encode()
521
522 req, err := http.NewRequestWithContext(ctx, http.MethodPut, parsedURL.String(), nil)
523 if err != nil {
524 return fmt.Errorf("failed to create blob finalize request: %w", err)
525 }
526 setScopedAuth(req, auth, baseURL)
527
528 resp, err := p.httpClient().Do(req)
529 if err != nil {
530 return fmt.Errorf("failed to finalize blob upload: %w", err)
531 }
532 defer resp.Body.Close()
533
534 if resp.StatusCode != http.StatusCreated {
535 body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
536 return fmt.Errorf("blob upload finalize returned status %d: %s", resp.StatusCode, string(body))
537 }
538
539 return nil
540}
541
542func (p *Pusher) httpClient() *http.Client {
543 return &http.Client{Transport: p.transport, Timeout: p.timeout}
544}
545
546// setScopedAuth attaches the auth header only if the request URL matches the
547// registry origin (scheme + host). This prevents credential leakage if a
548// redirect somehow escapes origin pinning.
549func setScopedAuth(req *http.Request, auth, registryOrigin string) {
550 if auth == "" {
551 return
552 }
553 origin, err := url.Parse(registryOrigin)
554 if err != nil {
555 return
556 }
557 if req.URL.Scheme != origin.Scheme || req.URL.Host != origin.Host {
558 return
559 }
560 req.Header.Set("Authorization", auth)
561}
562
563func resolveLocation(baseURL, location string) (string, error) {
564 if location == "" {
565 return "", fmt.Errorf("missing Location header")
566 }
567
568 base, err := url.Parse(baseURL)
569 if err != nil {
570 return "", err
571 }
572
573 ref, err := url.Parse(location)
574 if err != nil {
575 return "", err
576 }
577
578 resolved := base.ResolveReference(ref)
579
580 if resolved.Scheme != base.Scheme || resolved.Host != base.Host {
581 return "", fmt.Errorf("registry returned cross-origin Location %q (expected %s://%s)", location, base.Scheme, base.Host)
582 }
583
584 return resolved.String(), nil
585}
586
587func chunkCount(size, chunkSize int64) int64 {
588 if size == 0 || chunkSize <= 0 {
589 return 0
590 }
591 return (size + chunkSize - 1) / chunkSize
592}
593
594func insecureTransport() http.RoundTripper {
595 var transport *http.Transport
596 if t, ok := http.DefaultTransport.(*http.Transport); ok {
597 transport = t.Clone()
598 } else {
599 transport = &http.Transport{}
600 }
601 if transport.TLSClientConfig == nil {
602 transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12}
603 }
604 //nolint:gosec // Explicit CLI opt-in via --insecure for self-signed/private cert deployments.
605 transport.TLSClientConfig.InsecureSkipVerify = true
606 return transport
607}
608
609func bytesReader(chunk []byte) io.Reader {
610 return bytes.NewReader(chunk)
611}