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