Merge pull request #18278 from hakman/stream-container-images

k8s-ci-robot · web-flow · commit 15b43e1ee47e · 2026-05-06T21:22:23.000+05:30
nodeup: stream verified image bytes into ctr import
diff --git a/upup/pkg/fi/http.go b/upup/pkg/fi/http.go
@@ -23,13 +23,15 @@ import (
 	"net"
 	"net/http"
 	"os"
-	"path"
+	"path/filepath"
 	"time"
 
 	"k8s.io/klog/v2"
 	"k8s.io/kops/util/pkg/hashing"
 )
 
+const downloadTimeout = 3 * time.Minute
+
 // DownloadURL will download the file at the given url and store it as dest.
 // If hash is non-nil, it will also verify that it matches the hash of the downloaded file.
 func DownloadURL(url string, dest string, hash *hashing.Hash) (*hashing.Hash, error) {
@@ -43,47 +45,103 @@ func DownloadURL(url string, dest string, hash *hashing.Hash) (*hashing.Hash, er
 		}
 	}
 
-	dirMode := os.FileMode(0o755)
-	err := downloadURLAlways(url, dest, dirMode)
+	return downloadURLToFile(url, dest, hash)
+}
+
+func downloadURLToFile(url string, destPath string, hash *hashing.Hash) (*hashing.Hash, error) {
+	dir := filepath.Dir(destPath)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, fmt.Errorf("error creating directories for destination file %q: %v", destPath, err)
+	}
+
+	output, err := os.CreateTemp(dir, "."+filepath.Base(destPath)+".tmp")
+	if err != nil {
+		return nil, fmt.Errorf("error creating temporary file for download %q: %v", destPath, err)
+	}
+	tempPath := output.Name()
+	defer os.Remove(tempPath)
+
+	actual, err := downloadURLToWriter(url, output, hash)
+	if closeErr := output.Close(); closeErr != nil && err == nil {
+		err = closeErr
+	}
 	if err != nil {
 		return nil, err
 	}
+	if err := os.Chmod(tempPath, 0o644); err != nil {
+		return nil, fmt.Errorf("error setting mode on downloaded file %q: %v", tempPath, err)
+	}
+	if err := os.Rename(tempPath, destPath); err != nil {
+		return nil, fmt.Errorf("error moving downloaded file %q to %q: %v", tempPath, destPath, err)
+	}
+	return actual, nil
+}
 
+// downloadURLToWriter streams the file at the given url to dest.
+// If hash is non-nil, it will also verify that it matches the downloaded bytes.
+func downloadURLToWriter(url string, dest io.Writer, hash *hashing.Hash) (*hashing.Hash, error) {
+	responseBody, err := OpenURL(url)
+	if err != nil {
+		return nil, err
+	}
+	defer responseBody.Close()
+
+	start := time.Now()
+	defer func() {
+		klog.V(2).Infof("Downloading %q took %q", url, time.Since(start))
+	}()
+	klog.V(2).Infof("Downloading %q", url)
+
+	algorithm := hashing.HashAlgorithmSHA256
 	if hash != nil {
-		match, err := fileHasHash(dest, hash)
-		if err != nil {
-			return nil, err
-		}
-		if !match {
-			return nil, fmt.Errorf("downloaded from %q but hash did not match expected %q", url, hash)
-		}
-	} else {
-		hash, err = hashing.HashAlgorithmSHA256.HashFile(dest)
-		if err != nil {
-			return nil, err
-		}
+		algorithm = hash.Algorithm
 	}
+	hasher := algorithm.NewHasher()
+	writer := io.MultiWriter(dest, hasher)
 
-	return hash, nil
+	if _, err := io.Copy(writer, responseBody); err != nil {
+		return nil, fmt.Errorf("error downloading HTTP content from %q: %v", url, err)
+	}
+
+	actual := &hashing.Hash{
+		Algorithm: algorithm,
+		HashValue: hasher.Sum(nil),
+	}
+	if hash != nil && !actual.Equal(hash) {
+		return nil, fmt.Errorf("downloaded from %q but hash did not match expected %q", url, hash)
+	}
+	return actual, nil
 }
 
-func downloadURLAlways(url string, destPath string, dirMode os.FileMode) error {
-	err := os.MkdirAll(path.Dir(destPath), dirMode)
+// OpenURL opens a hardened HTTP GET stream for url.
+func OpenURL(url string) (io.ReadCloser, error) {
+	httpClient := newDownloadHTTPClient()
+
+	ctx, cancel := context.WithTimeout(context.Background(), downloadTimeout)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
-		return fmt.Errorf("error creating directories for destination file %q: %v", destPath, err)
+		cancel()
+		return nil, fmt.Errorf("cannot create request: %v", err)
 	}
 
-	output, err := os.Create(destPath)
+	response, err := httpClient.Do(req)
 	if err != nil {
-		return fmt.Errorf("error creating file for download %q: %v", destPath, err)
+		cancel()
+		return nil, fmt.Errorf("error doing HTTP fetch of %q: %v", url, err)
 	}
-	defer output.Close()
 
-	klog.V(2).Infof("Downloading %q", url)
+	// http.Client follows 3xx automatically, so anything outside 2xx that reaches us is a bug or a missing Location.
+	if response.StatusCode < 200 || response.StatusCode > 299 {
+		response.Body.Close()
+		cancel()
+		return nil, fmt.Errorf("unexpected response from %q: HTTP %s", url, response.Status)
+	}
+
+	return &cancelOnCloseReadCloser{ReadCloser: response.Body, cancel: cancel}, nil
+}
 
-	// Create a client with custom timeouts
-	// to avoid idle downloads to hang the program
-	httpClient := &http.Client{
+func newDownloadHTTPClient() *http.Client {
+	return &http.Client{
 		Transport: &http.Transport{
 			Proxy: http.ProxyFromEnvironment,
 			DialContext: (&net.Dialer{
@@ -95,35 +153,15 @@ func downloadURLAlways(url string, destPath string, dirMode os.FileMode) error {
 			IdleConnTimeout:       30 * time.Second,
 		},
 	}
+}
 
-	// this will stop slow downloads after 3 minutes
-	// and interrupt reading of the Response.Body
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
-	defer cancel()
-
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
-	if err != nil {
-		return fmt.Errorf("Cannot create request: %v", err)
-	}
-
-	response, err := httpClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("error doing HTTP fetch of %q: %v", url, err)
-	}
-	defer response.Body.Close()
-
-	if response.StatusCode >= 400 {
-		return fmt.Errorf("error response from %q: HTTP %v", url, response.StatusCode)
-	}
-
-	start := time.Now()
-	defer func() {
-		klog.V(2).Infof("Copying %q to %q took %q", url, destPath, time.Since(start))
-	}()
+type cancelOnCloseReadCloser struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
 
-	_, err = io.Copy(output, response.Body)
-	if err != nil {
-		return fmt.Errorf("error downloading HTTP content from %q: %v", url, err)
-	}
-	return nil
+func (r *cancelOnCloseReadCloser) Close() error {
+	err := r.ReadCloser.Close()
+	r.cancel()
+	return err
 }
diff --git a/upup/pkg/fi/http_test.go b/upup/pkg/fi/http_test.go
@@ -0,0 +1,95 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fi
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"k8s.io/kops/util/pkg/hashing"
+)
+
+func TestDownloadURLRejectsNon2xxAndPreservesDestination(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusFound)
+		_, _ = w.Write([]byte("redirect body"))
+	}))
+	defer server.Close()
+
+	dest := filepath.Join(t.TempDir(), "download")
+	if err := os.WriteFile(dest, []byte("original"), 0o644); err != nil {
+		t.Fatalf("WriteFile() error = %v", err)
+	}
+
+	if _, err := DownloadURL(server.URL, dest, nil); err == nil {
+		t.Fatalf("DownloadURL() expected error")
+	}
+
+	actual, err := os.ReadFile(dest)
+	if err != nil {
+		t.Fatalf("ReadFile() error = %v", err)
+	}
+	if string(actual) != "original" {
+		t.Fatalf("download destination = %q, expected original contents", actual)
+	}
+}
+
+func TestDownloadURLToWriterVerifiesHash(t *testing.T) {
+	body := []byte("payload")
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write(body)
+	}))
+	defer server.Close()
+
+	expectedHash, err := hashing.HashAlgorithmSHA256.Hash(bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("Hash() error = %v", err)
+	}
+
+	var output bytes.Buffer
+	actualHash, err := downloadURLToWriter(server.URL, &output, expectedHash)
+	if err != nil {
+		t.Fatalf("downloadURLToWriter() error = %v", err)
+	}
+	if !actualHash.Equal(expectedHash) {
+		t.Fatalf("downloadURLToWriter() hash = %v, expected %v", actualHash, expectedHash)
+	}
+	if !bytes.Equal(output.Bytes(), body) {
+		t.Fatalf("downloadURLToWriter() body = %q, expected %q", output.Bytes(), body)
+	}
+}
+
+func TestDownloadURLToWriterRejectsHashMismatch(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("payload"))
+	}))
+	defer server.Close()
+
+	wrongHash, err := hashing.HashAlgorithmSHA256.Hash(bytes.NewReader([]byte("different")))
+	if err != nil {
+		t.Fatalf("Hash() error = %v", err)
+	}
+
+	var output bytes.Buffer
+	if _, err := downloadURLToWriter(server.URL, &output, wrongHash); err == nil {
+		t.Fatalf("downloadURLToWriter() expected hash mismatch error")
+	}
+}
diff --git a/upup/pkg/fi/nodeup/command.go b/upup/pkg/fi/nodeup/command.go
@@ -28,6 +28,7 @@ import (
 	"net/url"
 	"os"
 	"os/exec"
+	"path"
 	"strconv"
 	"strings"
 	"time"
@@ -343,13 +344,23 @@ func (c *NodeUpCommand) Run(out io.Writer) error {
 		return fmt.Errorf("error building loader: %v", err)
 	}
 
-	for i, image := range nodeupConfig.Images[architecture] {
-		taskMap["LoadImage."+strconv.Itoa(i)] = &nodetasks.LoadImageTask{
+	for _, image := range nodeupConfig.Images[architecture] {
+		if len(image.Sources) == 0 {
+			return fmt.Errorf("image has no sources: %v", image)
+		}
+		u, err := url.Parse(image.Sources[0])
+		if err != nil {
+			return fmt.Errorf("invalid image source URL %q: %w", image.Sources[0], err)
+		}
+		key := "SideloadImage/" + path.Base(u.Path)
+		if _, ok := taskMap[key]; ok {
+			return fmt.Errorf("duplicate image task %q", key)
+		}
+		taskMap[key] = &nodetasks.LoadImageTask{
 			Sources: image.Sources,
 			Hash:    image.Hash,
 		}
 	}
-	// Protokube load image task is in ProtokubeBuilder
 
 	var target fi.NodeupTarget
 
diff --git a/upup/pkg/fi/nodeup/nodetasks/compression.go b/upup/pkg/fi/nodeup/nodetasks/compression.go
@@ -14,34 +14,25 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package utils
+package nodetasks
 
 import (
+	"bufio"
+	"bytes"
 	"compress/gzip"
 	"io"
-	"os"
 )
 
-// UngzipFile extracts a .gzip file
-func UngzipFile(gzipPath string, destPath string) error {
-	reader, err := os.Open(gzipPath)
-	if err != nil {
-		return err
-	}
-	defer reader.Close()
+var gzipMagic = []byte{0x1f, 0x8b}
 
-	writer, err := os.Create(destPath)
-	if err != nil {
-		return err
+func maybeGzipReader(r io.Reader) (io.ReadCloser, error) {
+	buffered := bufio.NewReader(r)
+	header, err := buffered.Peek(len(gzipMagic))
+	if err != nil && err != io.EOF {
+		return nil, err
 	}
-	defer writer.Close()
-
-	archive, err := gzip.NewReader(reader)
-	if err != nil {
-		return err
+	if len(header) == len(gzipMagic) && bytes.Equal(header, gzipMagic) {
+		return gzip.NewReader(buffered)
 	}
-	defer archive.Close()
-
-	_, err = io.Copy(writer, archive)
-	return err
+	return io.NopCloser(buffered), nil
 }
diff --git a/upup/pkg/fi/nodeup/nodetasks/load_image.go b/upup/pkg/fi/nodeup/nodetasks/load_image.go
diff --git a/upup/pkg/fi/nodeup/nodetasks/loadimage_test.go b/upup/pkg/fi/nodeup/nodetasks/loadimage_test.go
