feat(storage): redirect blob pulls to backend URLs (#4092)

* feat(storage): redirect blob pulls to backend URLs

* fix: rebase conflicts

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* refactor: rename redirect field

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* test: relax brittle TestPeriodicGC substore log assertion

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* feat(storage): improve blob redirect config handling and validation

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix(storage): address PR review feedback for blob redirect

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* feat(storage): apply latest PR review fixes for blob redirect

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* test: fix blob redirect and verify test regressions

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix(storage): enforce redirectBlobURL validation and add redirect tests

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix(storage): fix err113/noctx lint errors in storage driver tests

- Replace httptest.NewRequest with httptest.NewRequestWithContext in
  s3, gcs, and imagestore driver tests (noctx)
- Replace dynamic errors.New in s3 driver test with a package-level
  static sentinel error (err113)

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* test(storage): use temp dirs in imagestore redirect tests

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix: handle ranged blob redirects and add regression tests

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix: validate blob digest consistently in GetBlob

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* test: fix GetBlobPartialFn mock return values for range requests

The test 'does not redirect ranged blob requests' was failing because the mock
was returning incorrect length values. For a range request 'bytes=0-0' (1 byte),
it was returning 4 bytes, which caused a length mismatch check in GetBlob to
return HTTP 500.

Fix the mock to dynamically calculate the correct length: to - from + 1

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix(storage): preserve signed URL bytes in normalizeBlobRedirectURL

Preserve the original URL bytes from backend storage drivers (important
for signed/presigned URLs) while only lowercasing the scheme prefix.
URL re-serialization via net/url can invalidate signatures through path
escaping or canonicalization.

Add regression tests covering signed URL query parameters and mixed-case
scheme handling.

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix(storage): address PR review comments for blob redirect

- Return signed redirect URLs unchanged; validate scheme/CRLF/host only,
  no URL normalization that would corrupt signed URL bytes
- Add inline comments for all non-obvious decisions: range bypass, soft
  fallback on invalid URL, local driver empty return, subpath resolution,
  redirectBlobURL config constraint on local/empty driver
- Expand TestNormalizeBlobRedirectURL to cover allowed schemes (http/https),
  parse failure, missing host, and CRLF injection cases
- Add TestIsBlobRedirectEnabled covering subpath-only enablement with
  default store disabled

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* test(storage): address remaining blob redirect review comments

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

* fix: gofumpt formatting in routes_test.go

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>

---------

Signed-off-by: Ramkumar Chinchani <rchincha.dev@gmail.com>
Co-authored-by: Akash Kumar <meakash7902@gmail.com>

examples/README.md

@@ -1099,6 +1099,7 @@ The following AWS policy is required by zot for push and pull. Make sure to repl
     "storage": {
         "rootDirectory": "/tmp/zot",  # local path used to store dedupe cache database
         "dedupe": true,
+        "redirectBlobURL": true,
         "storageDriver": {
             "name": "s3",
             "rootdirectory": "/zot",  # this is a prefix that is applied to all S3 keys to allow you to segment data in your bucket if necessary.
@@ -1113,6 +1114,8 @@ The following AWS policy is required by zot for push and pull. Make sure to repl
     }
 ```
 
+Blob pull redirects are disabled by default. With S3 or GCS storage, set `redirectBlobURL` to `true` under `storage` or under a `subPaths` entry to return a `307 Temporary Redirect` to the storage driver's signed URL after zot authorization. If the storage driver does not return a redirect URL, zot proxies the blob as before.
+
 There are multiple ways to specify S3 credentials besides config file:
 
 - Environment variables:

pkg/api/config/config.go

@@ -27,17 +27,18 @@ var (
 )
 
 type StorageConfig struct {
-	RootDirectory string
-	MaxRepos      int
-	Dedupe        bool
-	RemoteCache   bool
-	GC            bool
-	Commit        bool
-	GCDelay       time.Duration // applied for blobs
-	GCInterval    time.Duration
-	Retention     ImageRetention
-	StorageDriver map[string]any `mapstructure:",omitempty"`
-	CacheDriver   map[string]any `mapstructure:",omitempty"`
+	RootDirectory   string
+	MaxRepos        int
+	Dedupe          bool
+	RemoteCache     bool
+	RedirectBlobURL bool
+	GC              bool
+	Commit          bool
+	GCDelay         time.Duration // applied for blobs
+	GCInterval      time.Duration
+	Retention       ImageRetention
+	StorageDriver   map[string]any `mapstructure:",omitempty"`
+	CacheDriver     map[string]any `mapstructure:",omitempty"`
 
 	// GCMaxSchedulerDelay is the maximum random delay for GC task scheduling
 	// This field is not configurable by the end user
@@ -803,7 +804,8 @@ func New() *Config {
 
 func (expConfig StorageConfig) ParamsEqual(actConfig StorageConfig) bool {
 	return expConfig.GC == actConfig.GC && expConfig.Dedupe == actConfig.Dedupe &&
-		expConfig.GCDelay == actConfig.GCDelay && expConfig.GCInterval == actConfig.GCInterval
+		expConfig.RedirectBlobURL == actConfig.RedirectBlobURL && expConfig.GCDelay == actConfig.GCDelay &&
+		expConfig.GCInterval == actConfig.GCInterval
 }
 
 // isRetentionEnabledInternal checks if retention is enabled without acquiring a lock (internal use only).
@@ -1009,6 +1011,7 @@ func (c *Config) UpdateReloadableConfig(newConfig *Config) {
 	// Update storage configuration
 	c.Storage.GC = newConfig.Storage.GC
 	c.Storage.Dedupe = newConfig.Storage.Dedupe
+	c.Storage.RedirectBlobURL = newConfig.Storage.RedirectBlobURL
 	c.Storage.GCDelay = newConfig.Storage.GCDelay
 	c.Storage.GCInterval = newConfig.Storage.GCInterval
 
@@ -1026,6 +1029,7 @@ func (c *Config) UpdateReloadableConfig(newConfig *Config) {
 
 		subPathConfig.GC = storageConfig.GC
 		subPathConfig.Dedupe = storageConfig.Dedupe
+		subPathConfig.RedirectBlobURL = storageConfig.RedirectBlobURL
 		subPathConfig.GCDelay = storageConfig.GCDelay
 		subPathConfig.GCInterval = storageConfig.GCInterval
 
@@ -1156,6 +1160,25 @@ func (c *Config) CopyStorageConfig() GlobalStorageConfig {
 	return storageCopy
 }
 
+// IsBlobRedirectEnabled returns whether blob redirect is enabled for a store path.
+// If a matching subpath exists, its setting takes precedence over the global one.
+func (c *Config) IsBlobRedirectEnabled(storePath string) bool {
+	if c == nil {
+		return false
+	}
+
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if storePath != "/" {
+		if subPathConfig, ok := c.Storage.SubPaths[storePath]; ok {
+			return subPathConfig.RedirectBlobURL
+		}
+	}
+
+	return c.Storage.RedirectBlobURL
+}
+
 // CopyExtensionsConfig returns a copy of the extensions config if it exists.
 func (c *Config) CopyExtensionsConfig() *extconf.ExtensionConfig {
 	if c == nil {

pkg/api/config/config_test.go

@@ -118,6 +118,14 @@ func TestConfig(t *testing.T) {
 
 		So(firstStorageConfig.ParamsEqual(secondStorageConfig), ShouldBeTrue)
 
+		firstStorageConfig.RedirectBlobURL = true
+
+		So(firstStorageConfig.ParamsEqual(secondStorageConfig), ShouldBeFalse)
+
+		firstStorageConfig.RedirectBlobURL = false
+
+		So(firstStorageConfig.ParamsEqual(secondStorageConfig), ShouldBeTrue)
+
 		isSame, err := config.SameFile("test-config", "test")
 		So(err, ShouldNotBeNil)
 		So(isSame, ShouldBeFalse)
@@ -1593,6 +1601,38 @@ func TestConfig(t *testing.T) {
 			})
 		})
 
+		Convey("Test IsBlobRedirectEnabled()", func() {
+			Convey("returns global setting for default store path", func() {
+				cfg := &config.Config{
+					Storage: config.GlobalStorageConfig{
+						StorageConfig: config.StorageConfig{RedirectBlobURL: true},
+					},
+				}
+
+				So(cfg.IsBlobRedirectEnabled("/"), ShouldBeTrue)
+			})
+
+			Convey("returns subpath setting when subpath exists", func() {
+				cfg := &config.Config{
+					Storage: config.GlobalStorageConfig{
+						StorageConfig: config.StorageConfig{RedirectBlobURL: false},
+						SubPaths: map[string]config.StorageConfig{
+							"/a": {RedirectBlobURL: true},
+						},
+					},
+				}
+
+				So(cfg.IsBlobRedirectEnabled("/a"), ShouldBeTrue)
+				So(cfg.IsBlobRedirectEnabled("/b"), ShouldBeFalse)
+			})
+
+			Convey("nil config returns false", func() {
+				var nilCfg *config.Config
+
+				So(nilCfg.IsBlobRedirectEnabled("/"), ShouldBeFalse)
+			})
+		})
+
 		Convey("Test CopyLogConfig()", func() {
 			Convey("Test with non-nil Log", func() {
 				cfg := &config.Config{
@@ -2300,6 +2340,39 @@ func TestConfig(t *testing.T) {
 			So(cfg.CopyExtensionsConfig().IsSearchEnabled(), ShouldBeTrue)
 		})
 
+		Convey("Test with Storage update", func() {
+			cfg := &config.Config{
+				Storage: config.GlobalStorageConfig{
+					StorageConfig: config.StorageConfig{
+						GC:              true,
+						Dedupe:          false,
+						RedirectBlobURL: false,
+						GCDelay:         time.Hour,
+						GCInterval:      2 * time.Hour,
+					},
+				},
+			}
+			newConfig := &config.Config{
+				Storage: config.GlobalStorageConfig{
+					StorageConfig: config.StorageConfig{
+						GC:              false,
+						Dedupe:          true,
+						RedirectBlobURL: true,
+						GCDelay:         3 * time.Hour,
+						GCInterval:      4 * time.Hour,
+					},
+				},
+			}
+
+			cfg.UpdateReloadableConfig(newConfig)
+
+			So(cfg.Storage.GC, ShouldBeFalse)
+			So(cfg.Storage.Dedupe, ShouldBeTrue)
+			So(cfg.Storage.RedirectBlobURL, ShouldBeTrue)
+			So(cfg.Storage.GCDelay, ShouldEqual, 3*time.Hour)
+			So(cfg.Storage.GCInterval, ShouldEqual, 4*time.Hour)
+		})
+
 		Convey("Test search CVE config removal when new config has nil Search.CVE", func() {
 			// First set up a config with search enabled and CVE config
 			enabled := true
@@ -3223,21 +3296,24 @@ func TestConfig(t *testing.T) {
 			cfg := &config.Config{
 				Storage: config.GlobalStorageConfig{
 					StorageConfig: config.StorageConfig{
-						GC:     true,
-						Dedupe: false,
+						GC:              true,
+						Dedupe:          false,
+						RedirectBlobURL: false,
 					},
 					SubPaths: map[string]config.StorageConfig{
 						"/path1": {
-							GC:         true,
-							Dedupe:     false,
-							GCDelay:    time.Hour,
-							GCInterval: time.Hour * 24,
+							GC:              true,
+							Dedupe:          false,
+							RedirectBlobURL: false,
+							GCDelay:         time.Hour,
+							GCInterval:      time.Hour * 24,
 						},
 						"/path2": {
-							GC:         false,
-							Dedupe:     true,
-							GCDelay:    time.Hour * 2,
-							GCInterval: time.Hour * 48,
+							GC:              false,
+							Dedupe:          true,
+							RedirectBlobURL: true,
+							GCDelay:         time.Hour * 2,
+							GCInterval:      time.Hour * 48,
 						},
 					},
 				},
@@ -3247,21 +3323,24 @@ func TestConfig(t *testing.T) {
 			newConfig := &config.Config{
 				Storage: config.GlobalStorageConfig{
 					StorageConfig: config.StorageConfig{
-						GC:     true,
-						Dedupe: false,
+						GC:              true,
+						Dedupe:          false,
+						RedirectBlobURL: true,
 					},
 					SubPaths: map[string]config.StorageConfig{
 						"/path1": {
-							GC:         false,          // Changed
-							Dedupe:     true,           // Changed
-							GCDelay:    time.Hour * 2,  // Changed
-							GCInterval: time.Hour * 12, // Changed
+							GC:              false,          // Changed
+							Dedupe:          true,           // Changed
+							RedirectBlobURL: true,           // Changed
+							GCDelay:         time.Hour * 2,  // Changed
+							GCInterval:      time.Hour * 12, // Changed
 						},
 						"/path2": {
-							GC:         true,           // Changed
-							Dedupe:     false,          // Changed
-							GCDelay:    time.Hour * 3,  // Changed
-							GCInterval: time.Hour * 36, // Changed
+							GC:              true,           // Changed
+							Dedupe:          false,          // Changed
+							RedirectBlobURL: false,          // Changed
+							GCDelay:         time.Hour * 3,  // Changed
+							GCInterval:      time.Hour * 36, // Changed
 						},
 					},
 				},
@@ -3277,6 +3356,7 @@ func TestConfig(t *testing.T) {
 			path1Config := cfg.Storage.SubPaths["/path1"]
 			So(path1Config.GC, ShouldBeFalse)
 			So(path1Config.Dedupe, ShouldBeTrue)
+			So(path1Config.RedirectBlobURL, ShouldBeTrue)
 			So(path1Config.GCDelay, ShouldEqual, time.Hour*2)
 			So(path1Config.GCInterval, ShouldEqual, time.Hour*12)
 
@@ -3284,6 +3364,7 @@ func TestConfig(t *testing.T) {
 			path2Config := cfg.Storage.SubPaths["/path2"]
 			So(path2Config.GC, ShouldBeTrue)
 			So(path2Config.Dedupe, ShouldBeFalse)
+			So(path2Config.RedirectBlobURL, ShouldBeFalse)
 			So(path2Config.GCDelay, ShouldEqual, time.Hour*3)
 			So(path2Config.GCInterval, ShouldEqual, time.Hour*36)
 		})

pkg/api/controller_test.go

@@ -12194,7 +12194,7 @@ func TestPeriodicGC(t *testing.T) {
 
 		// periodic GC is enabled for sub store
 		So(string(data), ShouldContainSubstring,
-			fmt.Sprintf("\"SubPaths\":{\"/a\":{\"RootDirectory\":\"%s\",\"MaxRepos\":0,\"Dedupe\":false,\"RemoteCache\":false,\"GC\":true,\"Commit\":false,\"GCDelay\":1000000000,\"GCInterval\":86400000000000", subDir)) //nolint:lll // gofumpt conflicts with lll
+			fmt.Sprintf("\"SubPaths\":{\"/a\":{\"RootDirectory\":\"%s\",\"MaxRepos\":0,\"Dedupe\":false,\"RemoteCache\":false,\"RedirectBlobURL\":false,\"GC\":true,\"Commit\":false,\"GCDelay\":1000000000,\"GCInterval\":86400000000000", subDir)) //nolint:lll // gofumpt conflicts with lll
 	})
 
 	Convey("Periodic gc error", t, func() {

pkg/api/routes.go

@@ -1418,6 +1418,35 @@ func writeMultipartRanges(
 	}
 }
 
+func (rh *RouteHandler) isBlobRedirectEnabled(repo string) bool {
+	storePath := rh.c.StoreController.GetStorePath(repo)
+
+	return rh.c.Config.IsBlobRedirectEnabled(storePath)
+}
+
+func normalizeBlobRedirectURL(rawURL string) (string, bool) {
+	if strings.ContainsAny(rawURL, "\r\n") {
+		return "", false
+	}
+
+	parsedURL, err := url.Parse(rawURL)
+	if err != nil {
+		return "", false
+	}
+
+	if parsedURL.Scheme == "" || parsedURL.Host == "" {
+		return "", false
+	}
+
+	scheme := strings.ToLower(parsedURL.Scheme)
+	if scheme != constants.SchemeHTTP && scheme != constants.SchemeHTTPS {
+		return "", false
+	}
+
+	// Keep the exact signed URL returned by the backend; only validate safety here.
+	return rawURL, true
+}
+
 // GetBlob godoc
 // @Summary Get image blob/layer
 // @Description Get an image's blob/layer given a digest
@@ -1473,6 +1502,34 @@ func (rh *RouteHandler) GetBlob(response http.ResponseWriter, request *http.Requ
 		}
 	}
 
+	if err := digest.Validate(); err != nil {
+		writeBlobError(zerr.ErrBadBlobDigest)
+
+		return
+	}
+
+	// Keep ranged pulls on the proxy path so zot can preserve 206 semantics.
+	if !rangeHeaderPresent && rh.isBlobRedirectEnabled(name) {
+		redirectURL, err := imgStore.GetBlobRedirectURL(request, name, digest)
+		if err != nil {
+			writeBlobError(err)
+
+			return
+		} else if redirectURL != "" {
+			if normalizedURL, ok := normalizeBlobRedirectURL(redirectURL); ok {
+				response.Header().Set(constants.DistContentDigestKey, digest.String())
+				response.Header().Set("Location", normalizedURL)
+				response.WriteHeader(http.StatusTemporaryRedirect)
+
+				return
+			}
+
+			// Invalid redirect URLs are treated as a soft failure to keep pulls working.
+			rh.c.Log.Warn().Str("repo", name).Str("digest", digest.String()).
+				Msg("ignoring invalid blob redirect URL and falling back to proxy")
+		}
+	}
+
 	if rangeHeaderPresent {
 		ok, bsize, err := imgStore.CheckBlob(name, digest)
 		if err != nil {

pkg/api/routes_internal_test.go

@@ -4,6 +4,10 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+
+	"zotregistry.dev/zot/v2/pkg/api/config"
+	"zotregistry.dev/zot/v2/pkg/storage"
+	storageTypes "zotregistry.dev/zot/v2/pkg/storage/types"
 )
 
 func TestParseRangeHeader(t *testing.T) {
@@ -102,3 +106,99 @@ func TestParseRangeHeader(t *testing.T) {
 		})
 	}
 }
+
+func TestNormalizeBlobRedirectURL(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		rawURL  string
+		wantURL string
+		wantOK  bool
+	}{
+		{
+			name:    "preserves signed url bytes unchanged",
+			rawURL:  "HTTPS://storage.example.com/blob?X-Amz-Signature=a%2Fb%2Bc",
+			wantURL: "HTTPS://storage.example.com/blob?X-Amz-Signature=a%2Fb%2Bc",
+			wantOK:  true,
+		},
+		{
+			name:    "allows http scheme",
+			rawURL:  "http://storage.example.com/blob",
+			wantURL: "http://storage.example.com/blob",
+			wantOK:  true,
+		},
+		{
+			name:   "rejects disallowed scheme",
+			rawURL: "javascript:alert(1)",
+			wantOK: false,
+		},
+		{
+			name:   "rejects parse failure",
+			rawURL: "https://storage.example.com/%zz",
+			wantOK: false,
+		},
+		{
+			name:   "rejects missing host",
+			rawURL: "https:///blob",
+			wantOK: false,
+		},
+		{
+			name:   "rejects crlf injection",
+			rawURL: "https://storage.example.com/blob?sig=abc\r\nX-Test: y",
+			wantOK: false,
+		},
+	}
+
+	for _, test := range tests {
+		test := test
+
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			gotURL, gotOK := normalizeBlobRedirectURL(test.rawURL)
+			if gotOK != test.wantOK {
+				t.Fatalf("expected ok=%v, got %v", test.wantOK, gotOK)
+			}
+
+			if gotURL != test.wantURL {
+				t.Fatalf("expected url %q, got %q", test.wantURL, gotURL)
+			}
+		})
+	}
+}
+
+func TestIsBlobRedirectEnabled(t *testing.T) {
+	t.Parallel()
+
+	routeHandler := &RouteHandler{
+		c: &Controller{
+			Config: &config.Config{
+				Storage: config.GlobalStorageConfig{
+					StorageConfig: config.StorageConfig{
+						RedirectBlobURL: false,
+					},
+					SubPaths: map[string]config.StorageConfig{
+						"/a": {
+							RedirectBlobURL: true,
+						},
+					},
+				},
+			},
+			StoreController: storage.StoreController{
+				SubStore: map[string]storageTypes.ImageStore{
+					"/a": nil,
+				},
+			},
+		},
+	}
+
+	if !routeHandler.isBlobRedirectEnabled("a/repo") {
+		t.Fatal("expected redirect to be enabled for /a subpath repo")
+	}
+
+	// Default storage remains disabled even when a specific subpath enables redirect.
+	if routeHandler.isBlobRedirectEnabled("b/repo") {
+		t.Fatal("expected redirect to be disabled for default storage")
+	}
+}

pkg/api/routes_test.go

@@ -933,6 +933,283 @@ func TestRoutes(t *testing.T) {
 					},
 				})
 			So(statusCode, ShouldEqual, http.StatusBadRequest)
+
+			Convey("redirects blob pulls when storage redirect is enabled", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				redirectURL := "https://storage.example.com/zot/repo/blobs/sha256/layer"
+				getBlobCalled := false
+
+				ctlr.Config.Storage.RedirectBlobURL = true
+				defer func() {
+					ctlr.Config.Storage.RedirectBlobURL = false
+				}()
+
+				ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						So(r.Method, ShouldEqual, http.MethodGet)
+						So(repo, ShouldEqual, "repo")
+						So(digest.String(), ShouldEqual, blobDigest)
+
+						return redirectURL, nil
+					},
+					GetBlobFn: func(repo string, digest godigest.Digest, mediaType string) (io.ReadCloser, int64, error) {
+						getBlobCalled = true
+
+						return io.NopCloser(bytes.NewBufferString("")), 0, nil
+					},
+				}
+
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusTemporaryRedirect)
+				So(resp.Header.Get(constants.DistContentDigestKey), ShouldEqual, blobDigest)
+				So(resp.Header.Get("Location"), ShouldEqual, redirectURL)
+				So(getBlobCalled, ShouldBeFalse)
+			})
+
+			Convey("rejects invalid blob digests before redirect or proxy execution", func() {
+				invalidDigest := "sha256:bad"
+
+				Convey("with redirect enabled", func() {
+					ctlr.Config.Storage.RedirectBlobURL = true
+					defer func() {
+						ctlr.Config.Storage.RedirectBlobURL = false
+					}()
+
+					ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+						GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+							t.Fatal("GetBlobRedirectURL should not run for an invalid digest")
+
+							return "", nil
+						},
+					}
+
+					request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+					request = mux.SetURLVars(request, map[string]string{
+						"name":   "repo",
+						"digest": invalidDigest,
+					})
+					response := httptest.NewRecorder()
+
+					rthdlr.GetBlob(response, request)
+
+					resp := response.Result()
+					defer resp.Body.Close()
+
+					So(resp.StatusCode, ShouldEqual, http.StatusBadRequest)
+				})
+			})
+
+			Convey("does not redirect ranged blob requests", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				redirectCalled := false
+				proxyCalled := false
+
+				ctlr.Config.Storage.RedirectBlobURL = true
+				defer func() {
+					ctlr.Config.Storage.RedirectBlobURL = false
+				}()
+
+				ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						redirectCalled = true
+
+						return "https://storage.example.com/zot/repo/blobs/sha256/layer", nil
+					},
+					CheckBlobFn: func(repo string, digest godigest.Digest) (bool, int64, error) {
+						return true, 4, nil
+					},
+					GetBlobPartialFn: func(repo string, digest godigest.Digest, mediaType string,
+						from, to int64,
+					) (io.ReadCloser, int64, int64, error) {
+						proxyCalled = true
+
+						// Return the correct length: to - from + 1
+						length := to - from + 1
+
+						return io.NopCloser(strings.NewReader("b")), length, length, nil
+					},
+				}
+
+				// Range requests must stay on proxy path to preserve partial-content behavior.
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request.Header.Set("Range", "bytes=0-0")
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusPartialContent)
+				So(redirectCalled, ShouldBeFalse)
+				So(proxyCalled, ShouldBeTrue)
+			})
+
+			Convey("falls back to proxying when redirect URL is unavailable", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				getBlobCalled := false
+
+				ctlr.Config.Storage.RedirectBlobURL = true
+				defer func() {
+					ctlr.Config.Storage.RedirectBlobURL = false
+				}()
+
+				ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						return "", nil
+					},
+					GetBlobFn: func(repo string, digest godigest.Digest, mediaType string) (io.ReadCloser, int64, error) {
+						getBlobCalled = true
+
+						return io.NopCloser(bytes.NewBufferString("blob")), 4, nil
+					},
+				}
+
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusOK)
+				So(getBlobCalled, ShouldBeTrue)
+			})
+
+			Convey("falls back to proxying when redirect URL is invalid", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				getBlobCalled := false
+
+				ctlr.Config.Storage.RedirectBlobURL = true
+				defer func() {
+					ctlr.Config.Storage.RedirectBlobURL = false
+				}()
+
+				ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						return "javascript:alert(1)", nil
+					},
+					GetBlobFn: func(repo string, digest godigest.Digest, mediaType string) (io.ReadCloser, int64, error) {
+						getBlobCalled = true
+
+						return io.NopCloser(bytes.NewBufferString("blob")), 4, nil
+					},
+				}
+
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusOK)
+				So(resp.Header.Get("Location"), ShouldEqual, "")
+				So(getBlobCalled, ShouldBeTrue)
+			})
+
+			Convey("uses subpath redirect config", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				redirectURL := "https://storage.example.com/zot-a/repo/blobs/sha256/layer"
+				getBlobCalled := false
+				subStore := &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						So(repo, ShouldEqual, "a/repo")
+
+						return redirectURL, nil
+					},
+					GetBlobFn: func(repo string, digest godigest.Digest, mediaType string) (io.ReadCloser, int64, error) {
+						getBlobCalled = true
+
+						return io.NopCloser(bytes.NewBufferString("")), 0, nil
+					},
+				}
+
+				ctlr.Config.Storage.RedirectBlobURL = false
+				// Redirect enablement is resolved from matched store path, not only global storage.
+				ctlr.Config.Storage.SubPaths = map[string]config.StorageConfig{
+					"/a": {RedirectBlobURL: true},
+				}
+				ctlr.StoreController.SubStore = map[string]storageTypes.ImageStore{
+					"/a": subStore,
+				}
+				defer func() {
+					ctlr.Config.Storage.SubPaths = nil
+					ctlr.StoreController.SubStore = nil
+				}()
+
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "a/repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusTemporaryRedirect)
+				So(resp.Header.Get(constants.DistContentDigestKey), ShouldEqual, blobDigest)
+				So(resp.Header.Get("Location"), ShouldEqual, redirectURL)
+				So(getBlobCalled, ShouldBeFalse)
+			})
+
+			Convey("returns registry errors from redirect lookup", func() {
+				blobDigest := "sha256:7b8437f04f83f084b7ed68ad8c4a4947e12fc4e1b006b38129bac89114ec3621"
+				getBlobCalled := false
+
+				ctlr.Config.Storage.RedirectBlobURL = true
+				defer func() {
+					ctlr.Config.Storage.RedirectBlobURL = false
+				}()
+
+				ctlr.StoreController.DefaultStore = &mocks.MockedImageStore{
+					GetBlobRedirectURLFn: func(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+						return "", zerr.ErrBlobNotFound
+					},
+					GetBlobFn: func(repo string, digest godigest.Digest, mediaType string) (io.ReadCloser, int64, error) {
+						getBlobCalled = true
+
+						return io.NopCloser(bytes.NewBufferString("")), 0, nil
+					},
+				}
+
+				request, _ := http.NewRequestWithContext(context.TODO(), http.MethodGet, baseURL, nil)
+				request = mux.SetURLVars(request, map[string]string{
+					"name":   "repo",
+					"digest": blobDigest,
+				})
+				response := httptest.NewRecorder()
+
+				rthdlr.GetBlob(response, request)
+
+				resp := response.Result()
+				defer resp.Body.Close()
+				So(resp.StatusCode, ShouldEqual, http.StatusNotFound)
+				So(getBlobCalled, ShouldBeFalse)
+			})
 		})
 
 		Convey("CreateBlobUpload", func() {

pkg/cli/server/root.go

@@ -618,6 +618,16 @@ func validateExtensionsConfig(cfg *config.Config, logger zlog.Logger) error {
 func validateStorageConfigSection(
 	cfg *config.Config, logger zlog.Logger, storageConfig config.GlobalStorageConfig,
 ) error {
+	// Redirect mode requires a backend capable of issuing external signed URLs.
+	// With local storage there is no target URL to redirect clients to.
+	if storageConfig.RedirectBlobURL && len(storageConfig.StorageDriver) == 0 {
+		msg := "invalid storage config, redirectBlobURL is supported only for s3/gcs storage"
+		logger.Error().Err(zerr.ErrBadConfig).Bool("redirectBlobURL", storageConfig.RedirectBlobURL).
+			Str("storageDriver", storageConstants.LocalStorageDriverName).Msg(msg)
+
+		return fmt.Errorf("%w: %s", zerr.ErrBadConfig, msg)
+	}
+
 	if len(storageConfig.StorageDriver) != 0 {
 		// enforce s3/gcs driver in case of using storage driver
 		if storageConfig.StorageDriver["name"] != storageConstants.S3StorageDriverName &&
@@ -641,6 +651,17 @@ func validateStorageConfigSection(
 	// enforce s3/gcs driver on subpaths in case of using storage driver
 	if len(storageConfig.SubPaths) > 0 {
 		for route, subStorageConfig := range storageConfig.SubPaths {
+			// Apply the same redirect precondition per subpath because subpaths can
+			// override driver config independently from the default store.
+			if subStorageConfig.RedirectBlobURL && len(subStorageConfig.StorageDriver) == 0 {
+				msg := "invalid storage config, redirectBlobURL is supported only for s3/gcs storage"
+				logger.Error().Err(zerr.ErrBadConfig).Str("subpath", route).
+					Bool("redirectBlobURL", subStorageConfig.RedirectBlobURL).
+					Str("storageDriver", storageConstants.LocalStorageDriverName).Msg(msg)
+
+				return fmt.Errorf("%w: %s", zerr.ErrBadConfig, msg)
+			}
+
 			if len(subStorageConfig.StorageDriver) != 0 {
 				if subStorageConfig.StorageDriver["name"] != storageConstants.S3StorageDriverName &&
 					subStorageConfig.StorageDriver["name"] != storageConstants.GCSStorageDriverName {

pkg/cli/server/root_test.go

@@ -1570,6 +1570,67 @@ storage:
 			"invalid storage config, default storage root directory cannot be inside substore (route: /a) root directory")
 	})
 
+	Convey("Test redirectBlobURL error for local storage", t, func(c C) {
+		// Redirect requires a remote driver that can generate an external URL.
+		content := `{"storage":{"rootDirectory":"/tmp/zot","redirectBlobURL":true,
+							"subPaths": {"/a": {"rootDirectory": "/tmp/zot-a","redirectBlobURL":true}}},
+							"http":{"address":"127.0.0.1","port":"8080"}}`
+
+		tmpfile := MakeTempFileWithContent(t, "zot-test.json", content)
+
+		cfg := config.New()
+		err := cli.LoadConfiguration(cfg, tmpfile)
+		So(err, ShouldNotBeNil)
+		So(err.Error(), ShouldContainSubstring,
+			"invalid storage config, redirectBlobURL is supported only for s3/gcs storage")
+	})
+
+	Convey("Test redirectBlobURL error for empty storageDriver map", t, func(c C) {
+		// An empty driver map is equivalent to local storage for redirect validation.
+		content := `{"storage":{"rootDirectory":"/tmp/zot","redirectBlobURL":true,
+							"storageDriver": {},
+							"subPaths": {"/a": {"rootDirectory": "/tmp/zot-a","redirectBlobURL":true,"storageDriver": {}}}},
+							"http":{"address":"127.0.0.1","port":"8080"}}`
+
+		tmpfile := MakeTempFileWithContent(t, "zot-test-empty-map.json", content)
+
+		cfg := config.New()
+		err := cli.LoadConfiguration(cfg, tmpfile)
+		So(err, ShouldNotBeNil)
+		So(err.Error(), ShouldContainSubstring,
+			"invalid storage config, redirectBlobURL is supported only for s3/gcs storage")
+	})
+
+	Convey("Test redirectBlobURL error for subpath local storage", t, func(c C) {
+		// Global storage has no redirect; only the subpath enables it without a driver.
+		content := `{"storage":{"rootDirectory":"/tmp/zot",
+							"subPaths": {"/a": {"rootDirectory": "/tmp/zot-a","redirectBlobURL":true}}},
+							"http":{"address":"127.0.0.1","port":"8080"}}`
+
+		tmpfile := MakeTempFileWithContent(t, "zot-test-subpath-local.json", content)
+
+		cfg := config.New()
+		err := cli.LoadConfiguration(cfg, tmpfile)
+		So(err, ShouldNotBeNil)
+		So(err.Error(), ShouldContainSubstring,
+			"invalid storage config, redirectBlobURL is supported only for s3/gcs storage")
+	})
+
+	Convey("Test redirectBlobURL error for subpath empty storageDriver map", t, func(c C) {
+		// Global storage has no redirect; subpath has an empty driver map.
+		content := `{"storage":{"rootDirectory":"/tmp/zot",
+							"subPaths": {"/a": {"rootDirectory": "/tmp/zot-a","redirectBlobURL":true,"storageDriver": {}}}},
+							"http":{"address":"127.0.0.1","port":"8080"}}`
+
+		tmpfile := MakeTempFileWithContent(t, "zot-test-subpath-empty-map.json", content)
+
+		cfg := config.New()
+		err := cli.LoadConfiguration(cfg, tmpfile)
+		So(err, ShouldNotBeNil)
+		So(err.Error(), ShouldContainSubstring,
+			"invalid storage config, redirectBlobURL is supported only for s3/gcs storage")
+	})
+
 	Convey("Test verify w/ authorization and w/o authentication", t, func(c C) {
 		content := `{"storage":{"rootDirectory":"/tmp/zot"},
 		 					"http":{"address":"127.0.0.1","port":"8080","realm":"zot",

pkg/storage/gcs/driver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net/http"
 	"strings"
 
 	// Add gcs support.
@@ -176,6 +177,12 @@ func (driver *Driver) Link(src, dest string) error {
 	return driver.formatErr(driver.store.PutContent(context.Background(), dest, []byte{}), dest)
 }
 
+func (driver *Driver) RedirectURL(r *http.Request, path string) (string, error) {
+	redirectURL, err := driver.store.RedirectURL(r, path)
+
+	return redirectURL, driver.formatErr(err, path)
+}
+
 // formatErr converts GCS-specific 404/not found errors to PathNotFoundError.
 func (driver *Driver) formatErr(err error, path string) error {
 	switch actual := err.(type) { //nolint: errorlint

pkg/storage/gcs/driver_test.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
@@ -383,6 +385,38 @@ func TestDriver(t *testing.T) {
 			err := gcsDriver.Link("/src", "/dst")
 			So(err, ShouldBeNil)
 		})
+
+		Convey("RedirectURL", func() {
+			req := httptest.NewRequestWithContext(context.Background(), http.MethodGet,
+				"http://localhost/v2/repo/blobs/sha256:abc", nil)
+
+			Convey("Success", func() {
+				storeMock.RedirectURLFn = func(r *http.Request, path string) (string, error) {
+					So(r, ShouldEqual, req)
+					So(path, ShouldEqual, "/blob/path")
+
+					return "https://example.com/signed", nil
+				}
+
+				url, err := gcsDriver.RedirectURL(req, "/blob/path")
+				So(err, ShouldBeNil)
+				So(url, ShouldEqual, "https://example.com/signed")
+			})
+
+			Convey("Error", func() {
+				storeMock.RedirectURLFn = func(_ *http.Request, _ string) (string, error) {
+					return "", errTest
+				}
+
+				url, err := gcsDriver.RedirectURL(req, "/blob/path")
+				So(url, ShouldEqual, "")
+				So(err, ShouldNotBeNil)
+
+				var storageErr driver.Error
+				So(errors.As(err, &storageErr), ShouldBeTrue)
+				So(storageErr.DriverName, ShouldEqual, "gcs")
+			})
+		})
 	})
 }
 

pkg/storage/imagestore/imagestore.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/http"
 	"path"
 	"path/filepath"
 	"slices"
@@ -1673,6 +1674,29 @@ func (is *ImageStore) GetBlob(repo string, digest godigest.Digest, mediaType str
 	return blobReadCloser, binfo.Size(), nil
 }
 
+func (is *ImageStore) GetBlobRedirectURL(r *http.Request, repo string, digest godigest.Digest) (string, error) {
+	var lockLatency time.Time
+
+	if err := digest.Validate(); err != nil {
+		return "", zerr.ErrBadBlobDigest
+	}
+
+	// Local storage has no external signed URL endpoint; proxy path is expected.
+	if is.storeDriver.Name() == storageConstants.LocalStorageDriverName {
+		return "", nil
+	}
+
+	is.RLock(&lockLatency)
+	defer is.RUnlock(&lockLatency)
+
+	binfo, err := is.originalBlobInfo(repo, digest)
+	if err != nil {
+		return "", err
+	}
+
+	return is.storeDriver.RedirectURL(r, binfo.Path())
+}
+
 // GetBlobContent returns blob contents, the caller function MUST lock from outside.
 // Should be used for small files(manifests/config blobs).
 func (is *ImageStore) GetBlobContent(repo string, digest godigest.Digest) ([]byte, error) {

pkg/storage/imagestore/imagestore_test.go

@@ -0,0 +1,100 @@
+package imagestore_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/distribution/distribution/v3/registry/storage/driver"
+	godigest "github.com/opencontainers/go-digest"
+	. "github.com/smartystreets/goconvey/convey"
+
+	zerr "zotregistry.dev/zot/v2/errors"
+	"zotregistry.dev/zot/v2/pkg/extensions/monitoring"
+	zlog "zotregistry.dev/zot/v2/pkg/log"
+	"zotregistry.dev/zot/v2/pkg/storage/gcs"
+	"zotregistry.dev/zot/v2/pkg/storage/imagestore"
+	"zotregistry.dev/zot/v2/pkg/storage/local"
+	"zotregistry.dev/zot/v2/pkg/test/mocks"
+)
+
+func TestGetBlobRedirectURL(t *testing.T) {
+	Convey("GetBlobRedirectURL", t, func() {
+		log := zlog.NewTestLogger()
+		metrics := monitoring.NewMetricsServer(false, log)
+
+		Convey("returns bad digest for invalid digest", func() {
+			store := imagestore.NewImageStore(t.TempDir(), "", false, false, log, metrics, nil,
+				local.New(true), nil, nil, nil)
+
+			url, err := store.GetBlobRedirectURL(nil, "repo", godigest.Digest("not-a-digest"))
+			So(url, ShouldEqual, "")
+			So(errors.Is(err, zerr.ErrBadBlobDigest), ShouldBeTrue)
+		})
+
+		Convey("returns empty URL for local storage", func() {
+			store := imagestore.NewImageStore(t.TempDir(), "", false, false, log, metrics, nil,
+				local.New(true), nil, nil, nil)
+
+			digest := godigest.FromString("blob-content")
+			// Local driver has no external signed URL endpoint, so redirect is intentionally empty.
+			url, err := store.GetBlobRedirectURL(nil, "repo", digest)
+			So(err, ShouldBeNil)
+			So(url, ShouldEqual, "")
+		})
+
+		Convey("returns redirect URL for remote storage", func() {
+			rootDir := t.TempDir()
+			storeMock := &mocks.StorageDriverMock{}
+			remoteDriver := gcs.New(storeMock)
+			store := imagestore.NewImageStore(rootDir, "", false, false, log, metrics, nil,
+				remoteDriver, nil, nil, nil)
+
+			repo := "repo"
+			digest := godigest.FromString("blob-content")
+			expectedBlobPath := store.BlobPath(repo, digest)
+			expectedURL := "https://example.com/signed/blob"
+
+			storeMock.StatFn = func(_ context.Context, path string) (driver.FileInfo, error) {
+				So(path, ShouldEqual, expectedBlobPath)
+
+				return &mocks.FileInfoMock{
+					PathFn: func() string { return path },
+					SizeFn: func() int64 { return 42 },
+				}, nil
+			}
+
+			storeMock.RedirectURLFn = func(_ *http.Request, path string) (string, error) {
+				So(path, ShouldEqual, expectedBlobPath)
+
+				return expectedURL, nil
+			}
+
+			req := httptest.NewRequestWithContext(context.Background(), http.MethodGet,
+				"http://localhost/v2/repo/blobs/sha256:deadbeef", nil)
+
+			url, err := store.GetBlobRedirectURL(req, repo, digest)
+			So(err, ShouldBeNil)
+			So(url, ShouldEqual, expectedURL)
+		})
+
+		Convey("returns blob not found when blob path does not exist", func() {
+			rootDir := t.TempDir()
+			storeMock := &mocks.StorageDriverMock{}
+			remoteDriver := gcs.New(storeMock)
+			store := imagestore.NewImageStore(rootDir, "", false, false, log, metrics, nil,
+				remoteDriver, nil, nil, nil)
+
+			storeMock.StatFn = func(_ context.Context, path string) (driver.FileInfo, error) {
+				return nil, driver.PathNotFoundError{Path: path}
+			}
+
+			digest := godigest.FromString("blob-content")
+			url, err := store.GetBlobRedirectURL(nil, "repo", digest)
+			So(url, ShouldEqual, "")
+			So(errors.Is(err, zerr.ErrBlobNotFound), ShouldBeTrue)
+		})
+	})
+}

pkg/storage/local/driver.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net/http"
 	"os"
 	"path"
 	"sort"
@@ -293,6 +294,10 @@ func (driver *Driver) Link(src, dest string) error {
 	return nil
 }
 
+func (driver *Driver) RedirectURL(_ *http.Request, _ string) (string, error) {
+	return "", nil
+}
+
 func (driver *Driver) formatErr(err error) error {
 	switch actual := err.(type) { //nolint: errorlint
 	case nil:

pkg/storage/local/driver_test.go

@@ -222,6 +222,16 @@ func TestMove(t *testing.T) {
 	})
 }
 
+func TestRedirectURL(t *testing.T) {
+	Convey("Test RedirectURL returns empty URL for local driver", t, func() {
+		driver := local.New(true)
+
+		url, err := driver.RedirectURL(nil, "/repo/blobs/sha256/abc")
+		So(err, ShouldBeNil)
+		So(url, ShouldEqual, "")
+	})
+}
+
 func TestValidateHardLink(t *testing.T) {
 	Convey("Test ValidateHardLink functionality", t, func() {
 		rootDir := t.TempDir()

pkg/storage/s3/driver.go

@@ -3,6 +3,7 @@ package s3
 import (
 	"context"
 	"io"
+	"net/http"
 
 	// Add s3 support.
 	"github.com/distribution/distribution/v3/registry/storage/driver"
@@ -110,3 +111,7 @@ func (driver *Driver) SameFile(path1, path2 string) bool {
 func (driver *Driver) Link(src, dest string) error {
 	return driver.store.PutContent(context.Background(), dest, []byte{})
 }
+
+func (driver *Driver) RedirectURL(r *http.Request, path string) (string, error) {
+	return driver.store.RedirectURL(r, path)
+}

pkg/storage/s3/driver_test.go

@@ -0,0 +1,48 @@
+package s3_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	. "github.com/smartystreets/goconvey/convey"
+
+	"zotregistry.dev/zot/v2/pkg/storage/s3"
+	"zotregistry.dev/zot/v2/pkg/test/mocks"
+)
+
+var errRedirect = errors.New("redirect error")
+
+func TestDriverRedirectURL(t *testing.T) {
+	Convey("S3 Driver RedirectURL", t, func() {
+		storeMock := &mocks.StorageDriverMock{}
+		s3Driver := s3.New(storeMock)
+		req := httptest.NewRequestWithContext(context.Background(), http.MethodGet,
+			"http://localhost/v2/repo/blobs/sha256:abc", nil)
+
+		Convey("Success", func() {
+			storeMock.RedirectURLFn = func(r *http.Request, path string) (string, error) {
+				So(r, ShouldEqual, req)
+				So(path, ShouldEqual, "/blob/path")
+
+				return "https://example.com/signed", nil
+			}
+
+			url, err := s3Driver.RedirectURL(req, "/blob/path")
+			So(err, ShouldBeNil)
+			So(url, ShouldEqual, "https://example.com/signed")
+		})
+
+		Convey("Error", func() {
+			storeMock.RedirectURLFn = func(_ *http.Request, _ string) (string, error) {
+				return "", errRedirect
+			}
+
+			url, err := s3Driver.RedirectURL(req, "/blob/path")
+			So(url, ShouldEqual, "")
+			So(errors.Is(err, errRedirect), ShouldBeTrue)
+		})
+	})
+}

pkg/storage/types/types.go

@@ -3,6 +3,7 @@ package types
 import (
 	"context"
 	"io"
+	"net/http"
 	"time"
 
 	storagedriver "github.com/distribution/distribution/v3/registry/storage/driver"
@@ -70,6 +71,7 @@ type ImageStore interface { //nolint:interfacebloat
 	PopulateStorageMetrics(interval time.Duration, sch *scheduler.Scheduler)
 	VerifyBlobDigestValue(repo string, digest godigest.Digest) error
 	GetAllDedupeReposCandidates(digest godigest.Digest) ([]string, error)
+	GetBlobRedirectURL(r *http.Request, repo string, digest godigest.Digest) (string, error)
 }
 
 type Driver interface { //nolint:interfacebloat
@@ -87,4 +89,5 @@ type Driver interface { //nolint:interfacebloat
 	Move(sourcePath string, destPath string) error
 	SameFile(path1, path2 string) bool
 	Link(src, dest string) error
+	RedirectURL(r *http.Request, path string) (string, error)
 }

pkg/test/mocks/image_store_mock.go

@@ -3,6 +3,7 @@ package mocks
 import (
 	"context"
 	"io"
+	"net/http"
 	"time"
 
 	godigest "github.com/opencontainers/go-digest"
@@ -62,6 +63,7 @@ type MockedImageStore struct {
 	StatIndexFn                   func(repo string) (bool, int64, time.Time, error)
 	VerifyBlobDigestValueFn       func(repo string, digest godigest.Digest) error
 	GetAllDedupeReposCandidatesFn func(digest godigest.Digest) ([]string, error)
+	GetBlobRedirectURLFn          func(r *http.Request, repo string, digest godigest.Digest) (string, error)
 }
 
 func (is MockedImageStore) StatIndex(repo string) (bool, int64, time.Time, error) {
@@ -340,6 +342,16 @@ func (is MockedImageStore) GetBlob(repo string, digest godigest.Digest, mediaTyp
 	return io.NopCloser(&io.LimitedReader{}), 0, nil
 }
 
+func (is MockedImageStore) GetBlobRedirectURL(
+	r *http.Request, repo string, digest godigest.Digest,
+) (string, error) {
+	if is.GetBlobRedirectURLFn != nil {
+		return is.GetBlobRedirectURLFn(r, repo, digest)
+	}
+
+	return "", nil
+}
+
 func (is MockedImageStore) DeleteBlobUpload(repo string, uuid string) error {
 	if is.DeleteBlobUploadFn != nil {
 		return is.DeleteBlobUploadFn(repo, uuid)