go/runtime/bundle: Use indirection when downloading bundles

Author: peternose

.changelog/5962.cfg.md

@@ -20,9 +20,9 @@ The following configuration options have been added:
 - `runtime.runtimes.config` is the runtime local configuration,
 
 - `runtime.runtimes.repositories` is the list of runtime specific URLs
-   used to fetch runtime bundles,
+   used to fetch runtime bundle metadata,
 
 - `runtime.repositories` is the list of global URLs used to fetch
-   runtime bundles,
+   runtime bundle metadata,
 
 - `runtime.max_bundle_size` is the maximum allowed bundle size.

go/oasis-test-runner/scenario/e2e/runtime/keymanager_upgrade.go

@@ -80,7 +80,7 @@ func (sc *KmUpgradeImpl) Run(ctx context.Context, childEnv *env.Env) error {
 	port := parsedURL.Port()
 
 	// Start serving bundles.
-	server := newBundleServer(port, bundles)
+	server := newBundleServer(port, bundles, sc.Logger)
 	server.Start()
 	defer server.Stop()
 
@@ -90,7 +90,7 @@ func (sc *KmUpgradeImpl) Run(ctx context.Context, childEnv *env.Env) error {
 	}
 
 	// Verify that all key manager nodes requested bundle from the server.
-	n := len(sc.Net.Keymanagers())
+	n := 2 * len(sc.Net.Keymanagers())
 	if m := server.getRequestCount(); m != n {
 		return fmt.Errorf("invalid number of bundle requests (got: %d, expected: %d)", m, n)
 	}

go/oasis-test-runner/scenario/e2e/runtime/runtime_upgrade.go

@@ -14,6 +14,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	cmSync "github.com/oasisprotocol/oasis-core/go/common/sync"
 	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/env"
 	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/oasis"
@@ -87,7 +88,7 @@ func (sc *runtimeUpgradeImpl) Run(ctx context.Context, childEnv *env.Env) error
 	port := parsedURL.Port()
 
 	// Start serving bundles.
-	server := newBundleServer(port, bundles)
+	server := newBundleServer(port, bundles, sc.Logger)
 	server.Start()
 	defer server.Stop()
 
@@ -97,7 +98,7 @@ func (sc *runtimeUpgradeImpl) Run(ctx context.Context, childEnv *env.Env) error
 	}
 
 	// Verify that all client and compute nodes requested bundle from the server.
-	n := len(sc.Net.Clients()) + len(sc.Net.ComputeWorkers())
+	n := 2 * (len(sc.Net.Clients()) + len(sc.Net.ComputeWorkers()))
 	if m := server.getRequestCount(); m != n {
 		return fmt.Errorf("invalid number of bundle requests (got: %d, expected: %d)", m, n)
 	}
@@ -117,13 +118,16 @@ type bundleServer struct {
 	bundles map[string]string
 
 	requestCount uint64
+
+	logger *logging.Logger
 }
 
-func newBundleServer(port string, bundles map[string]string) *bundleServer {
+func newBundleServer(port string, bundles map[string]string, logger *logging.Logger) *bundleServer {
 	return &bundleServer{
 		startOne: cmSync.NewOne(),
 		port:     port,
 		bundles:  bundles,
+		logger:   logger,
 	}
 }
 
@@ -160,6 +164,30 @@ func (s *bundleServer) run(ctx context.Context) {
 }
 
 func (s *bundleServer) handleRequest(w http.ResponseWriter, r *http.Request) {
+	s.logger.Info("handling request",
+		"path", r.URL.Path,
+	)
+
+	if strings.HasSuffix(r.URL.Path, bundle.FileExtension) {
+		s.handleGetBundle(w, r)
+	} else {
+		s.handleGetMetadata(w, r)
+	}
+}
+
+func (s *bundleServer) handleGetMetadata(w http.ResponseWriter, r *http.Request) {
+	manifestHash := path.Base(r.URL.Path)
+	content := []byte(fmt.Sprintf("http://127.0.0.1:%s/%s%s\n", s.port, manifestHash, bundle.FileExtension))
+
+	w.Header().Set("Content-Disposition", "attachment; filename=metadata.txt")
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write(content)
+
+	atomic.AddUint64(&s.requestCount, 1)
+}
+
+func (s *bundleServer) handleGetBundle(w http.ResponseWriter, r *http.Request) {
 	filename := path.Base(r.URL.Path)
 
 	path, ok := s.bundles[filename]
@@ -174,7 +202,8 @@ func (s *bundleServer) handleRequest(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	w.Header().Set("Content-Type", "text/plain")
+	w.Header().Set("Content-Disposition", "attachment; filename=bundle.orc")
+	w.Header().Set("Content-Type", "application/octet-stream")
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(content)
 

go/runtime/bundle/discovery.go

@@ -1,6 +1,7 @@
 package bundle
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -29,6 +30,9 @@ const (
 	// requestTimeout is the time limit for http client requests.
 	requestTimeout = 10 * time.Second
 
+	// maxMetadataSizeBytes is the maximum allowed metadata size in bytes.
+	maxMetadataSizeBytes = 2 * 1024 // 2 KB
+
 	// maxDefaultBundleSizeBytes is the maximum allowed default bundle size
 	// in bytes.
 	maxDefaultBundleSizeBytes = 20 * 1024 * 1024 // 20 MB
@@ -295,7 +299,7 @@ func (d *Discovery) downloadBundle(runtimeID common.Namespace, manifestHash hash
 
 	for _, baseURLs := range [][]string{d.runtimeBaseURLs[runtimeID], d.globalBaseURLs} {
 		for _, baseURL := range baseURLs {
-			if err := d.tryDownloadBundle(runtimeID, manifestHash, baseURL); err != nil {
+			if err := d.tryDownloadBundle(manifestHash, baseURL); err != nil {
 				errs = errors.Join(errs, err)
 				continue
 			}
@@ -307,39 +311,49 @@ func (d *Discovery) downloadBundle(runtimeID common.Namespace, manifestHash hash
 	return errs
 }
 
-func (d *Discovery) tryDownloadBundle(runtimeID common.Namespace, manifestHash hash.Hash, baseURL string) error {
-	filename := fmt.Sprintf("%s%s", manifestHash.Hex(), FileExtension)
+func (d *Discovery) tryDownloadBundle(manifestHash hash.Hash, baseURL string) error {
+	metaURL, err := url.JoinPath(baseURL, manifestHash.Hex())
+	if err != nil {
+		d.logger.Error("failed to construct metadata URL",
+			"err", err,
+		)
+		return fmt.Errorf("failed to construct metadata URL: %w", err)
+	}
 
-	d.logger.Debug("downloading bundle",
-		"runtime_id", runtimeID,
-		"base_url", baseURL,
-		"filename", filename,
+	d.logger.Debug("downloading metadata",
+		"url", metaURL,
 	)
 
-	url, err := url.JoinPath(baseURL, filename)
+	bundleURL, err := d.fetchMetadata(metaURL)
 	if err != nil {
-		d.logger.Error("failed to construct URL",
+		d.logger.Error("failed to download metadata",
 			"err", err,
-			"base_url", baseURL,
-			"filename", filename,
+			"url", metaURL,
 		)
-		return fmt.Errorf("failed to construct URL: %w", err)
+		return fmt.Errorf("failed to download metadata: %w", err)
+	}
+
+	bundleURL, err = validateAndNormalizeURL(bundleURL)
+	if err != nil {
+		return err
 	}
 
-	src, err := d.fetchBundle(url)
+	d.logger.Debug("downloading bundle",
+		"url", bundleURL,
+	)
+
+	src, err := d.fetchBundle(bundleURL)
 	if err != nil {
 		d.logger.Error("failed to download bundle",
 			"err", err,
-			"url", url,
+			"url", metaURL,
 		)
 		return fmt.Errorf("failed to download bundle: %w", err)
 	}
 	defer os.Remove(src)
 
 	d.logger.Info("bundle downloaded",
-		"runtime_id", runtimeID,
-		"base_url", baseURL,
-		"filename", filename,
+		"url", bundleURL,
 	)
 
 	if err := d.registry.AddBundle(src, manifestHash); err != nil {
@@ -349,6 +363,7 @@ func (d *Discovery) tryDownloadBundle(runtimeID common.Namespace, manifestHash h
 		return fmt.Errorf("failed to add bundle: %w", err)
 	}
 
+	filename := fmt.Sprintf("%s%s", manifestHash.Hex(), FileExtension)
 	dst := filepath.Join(d.bundleDir, filename)
 	if err = os.Rename(src, dst); err != nil {
 		d.logger.Error("failed to move bundle",
@@ -357,9 +372,39 @@ func (d *Discovery) tryDownloadBundle(runtimeID common.Namespace, manifestHash h
 			"dst", dst,
 		)
 	}
+
+	d.logger.Debug("bundle stored",
+		"dst", dst,
+	)
+
 	return nil
 }
 
+func (d *Discovery) fetchMetadata(url string) (string, error) {
+	resp, err := d.client.Get(url)
+	if err != nil {
+		return "", fmt.Errorf("failed to fetch metadata: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("failed to fetch metadata: invalid status code %d", resp.StatusCode)
+	}
+
+	limitedReader := io.LimitedReader{
+		R: resp.Body,
+		N: maxMetadataSizeBytes,
+	}
+
+	var buffer bytes.Buffer
+	_, err = buffer.ReadFrom(&limitedReader)
+	if err != nil && err != io.EOF {
+		return "", fmt.Errorf("failed to read metadata content: %w", err)
+	}
+
+	return strings.TrimSpace(buffer.String()), nil
+}
+
 func (d *Discovery) fetchBundle(url string) (string, error) {
 	resp, err := d.client.Get(url)
 	if err != nil {
@@ -470,15 +515,23 @@ func (d *Discovery) copyBundle(src string) error {
 	return nil
 }
 
+func validateAndNormalizeURL(rawURL string) (string, error) {
+	parsedURL, err := url.Parse(rawURL)
+	if err != nil {
+		return "", fmt.Errorf("invalid URL '%s': %w", rawURL, err)
+	}
+	return parsedURL.String(), nil
+}
+
 func validateAndNormalizeURLs(rawURLs []string) ([]string, error) {
 	var normalizedURLs []string
 
 	for _, rawURL := range rawURLs {
-		parsedURL, err := url.Parse(rawURL)
+		normalizedURL, err := validateAndNormalizeURL(rawURL)
 		if err != nil {
-			return nil, fmt.Errorf("invalid URL '%s': %w", rawURL, err)
+			return nil, err
 		}
-		normalizedURLs = append(normalizedURLs, parsedURL.String())
+		normalizedURLs = append(normalizedURLs, normalizedURL)
 	}
 
 	return normalizedURLs, nil

go/runtime/config/config.go

@@ -115,7 +115,7 @@ type Config struct {
 	// LoadBalancer is the load balancer configuration.
 	LoadBalancer LoadBalancerConfig `yaml:"load_balancer,omitempty"`
 
-	// Repositories is the list of URLs used to fetch runtime bundles.
+	// Repositories is the list of URLs used to fetch runtime bundle metadata.
 	Repositories []string `yaml:"repositories,omitempty"`
 
 	// MaxBundleSize is the maximum allowed bundle size.
@@ -166,7 +166,7 @@ type RuntimeConfig struct {
 	// Config contains runtime local configuration.
 	Config map[string]interface{} `yaml:"config,omitempty"`
 
-	// Repositories is the list of URLs used to fetch runtime bundles.
+	// Repositories is the list of URLs used to fetch runtime bundle metadata.
 	Repositories []string `yaml:"repositories,omitempty"`
 }