[snapshot] Add --publish so upload works from outside the cluster network

internal/snapshot/cmd/snapimport/import.go

@@ -35,12 +35,14 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/rest"
 
+	dataio "github.com/deckhouse/deckhouse-cli/internal/data"
 	"github.com/deckhouse/deckhouse-cli/internal/snapshot/aggapi"
 	snapshotapi "github.com/deckhouse/deckhouse-cli/internal/snapshot/api/v1alpha1"
 	"github.com/deckhouse/deckhouse-cli/internal/snapshot/progress"
 	"github.com/deckhouse/deckhouse-cli/internal/snapshot/snapimport"
 	"github.com/deckhouse/deckhouse-cli/internal/snapshot/transport"
 	systemflags "github.com/deckhouse/deckhouse-cli/internal/system/flags"
+	safeClient "github.com/deckhouse/deckhouse-cli/pkg/libsaferequest/client"
 )
 
 const (
@@ -55,6 +57,7 @@ const (
 	flagAllowExisting              = "allow-existing"
 	flagAllowUnauthenticatedLegacy = "allow-unauthenticated-legacy"
 	flagSkipUnsupportedFSEntries   = "skip-unsupported-fs-entries"
+	flagPublish                    = "publish"
 
 	defaultImportWorkers = 5
 
@@ -133,15 +136,29 @@ Scope and limitations:
     downgraded and tampered current archive.
   - Uploading requires RBAC to create DataImport (storage-volume-data-manager) and to call
     the manifests-and-children-refs-upload subresource (e.g. an admin kubeconfig); the
-    read-only snapshot admin role is not sufficient.`,
+    read-only snapshot admin role is not sufficient.
+
+--publish selects how each data leaf's bytes are streamed to its DataImport importer pod.
+With --publish=false (or when autodetection picks it), bytes go straight to the importer's
+in-cluster service, trusting only its internal CA (status.ca). With --publish=true, bytes go
+through the storage-foundation-published Ingress endpoint (status.publicURL) instead, so a
+kubeconfig without direct network access to the cluster's internal service network can still
+upload. If --publish is not given, the command probes whether the in-cluster importer endpoint
+is reachable and picks accordingly. IMPORTANT: the publish path works only with a kubeconfig
+authenticated by a bearer token. Ingress terminates TLS with its own certificate and does not
+forward the client's TLS certificate to the importer pod, so a certificate-based kubeconfig
+receives a 401 when --publish=true.`,
 		Example: `  # Upload the archive in ./out into namespace "restored"
   d8 snapshot upload -n restored -i ./out
 
   # Upload only a single VolumeSnapshot data leaf and its subtree
   d8 snapshot upload -n restored -i ./out --node VolumeSnapshot/pvc-1
 
   # Upload with a longer DataImport TTL and overall timeout
-  d8 snapshot upload -n restored -i ./out --ttl 4h --timeout 30m`,
+  d8 snapshot upload -n restored -i ./out --ttl 4h --timeout 30m
+
+  # Upload through the published Ingress endpoint (requires a bearer-token kubeconfig)
+  d8 snapshot upload -n restored -i ./out --publish=true`,
 		Args: cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return Run(log, cmd, args)
@@ -159,6 +176,8 @@ Scope and limitations:
 	cmd.Flags().Bool(flagAllowExisting, false, "downgrade namespace preflight conflict check to a warning (import-mode markers from a prior run are never conflicts regardless of this flag)")
 	cmd.Flags().Bool(flagAllowUnauthenticatedLegacy, false, "allow trusted pre-version archives whose snapshot.yaml metadata is unauthenticated (unsafe; explicit compatibility mode)")
 	cmd.Flags().Bool(flagSkipUnsupportedFSEntries, false, "skip unsupported filesystem entries and report them after upload (causes data loss for skipped paths)")
+	cmd.Flags().Bool(flagPublish, false, "upload volume data through the published (ingress) importer endpoint instead of the in-cluster service; "+
+		"if unset, the in-cluster endpoint's reachability is auto-detected")
 
 	return cmd
 }
@@ -279,6 +298,21 @@ func Run(log *slog.Logger, cmd *cobra.Command, _ []string) error {
 
 	dataPlaneClient := transport.NewClientForConfig(restConfig)
 
+	publishFlag, err := dataio.ParsePublishFlag(cmd.Flags())
+	if err != nil {
+		return fmt.Errorf("resolving --%s: %w", flagPublish, err)
+	}
+
+	// Probe from the command's already-resolved restConfig, not a fresh parse of
+	// --kubeconfig/--context: reparsing could target a different cluster than the one this
+	// command is actually uploading into.
+	probeClient := safeClient.NewSafeClientForConfig(restConfig)
+
+	publish, err := dataio.ResolvePublish(ctx, publishFlag, kubeClient, probeClient, log)
+	if err != nil {
+		return fmt.Errorf("resolving --%s: %w", flagPublish, err)
+	}
+
 	isTTY := term.IsTerminal(int(os.Stdout.Fd()))
 
 	// Upload shows Upload/Uploading/DataImport wording instead of progress.New's
@@ -298,7 +332,15 @@ func Run(log *slog.Logger, cmd *cobra.Command, _ []string) error {
 		runLog = slog.New(slog.NewTextHandler(sink.LogWriter(), &slog.HandlerOptions{Level: slog.LevelWarn}))
 	}
 
-	volumes := snapimport.NewClusterVolumeImporter(dynClient, dataPlaneClient, ttl, timeout, 3*time.Second, runLog)
+	volumes := snapimport.NewClusterVolumeImporter(snapimport.ClusterVolumeImporterOptions{
+		Dynamic:   dynClient,
+		Transport: dataPlaneClient,
+		TTL:       ttl,
+		Publish:   publish,
+		Wait:      timeout,
+		Poll:      3 * time.Second,
+		Log:       runLog,
+	})
 
 	cfg := snapimport.Config{
 		Namespace:                  namespace,

internal/snapshot/cmd/snapimport/import_test.go

@@ -477,3 +477,120 @@ func TestNewCommand_AllowExistingFlagDefault(t *testing.T) {
 		t.Fatalf("default --%s: got true, want false (opt-in flag)", flagAllowExisting)
 	}
 }
+
+// TestNewCommand_PublishFlagDefault verifies --publish defaults to false and, crucially, that
+// Changed stays false when the flag is never passed on the command line: Run's autodetection
+// path (dataio.ParsePublishFlag) distinguishes "not set" from "explicitly set to false" via
+// Changed, and that distinction must survive flag registration untouched.
+func TestNewCommand_PublishFlagDefault(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+	}{
+		{name: "success: default value is false and unset"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			cmd := NewCommand(slog.Default())
+
+			flag := cmd.Flags().Lookup(flagPublish)
+			if flag == nil {
+				t.Fatalf("--%s flag is not registered", flagPublish)
+			}
+
+			publish, err := cmd.Flags().GetBool(flagPublish)
+			if err != nil {
+				t.Fatalf("getting %s flag: %v", flagPublish, err)
+			}
+
+			if publish {
+				t.Fatalf("default --%s: got true, want false", flagPublish)
+			}
+
+			if flag.Changed {
+				t.Fatal("--publish.Changed = true without ever being set on the command line, want false (autodetection relies on this)")
+			}
+		})
+	}
+}
+
+// TestNewCommand_PublishFlagExplicitlySet verifies that explicitly passing --publish=true
+// marks the flag Changed, so Run's dataio.ParsePublishFlag sees an explicit override rather
+// than falling back to autodetection.
+func TestNewCommand_PublishFlagExplicitlySet(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name  string
+		value string
+		want  bool
+	}{
+		{name: "success: explicit true", value: "true", want: true},
+		{name: "success: explicit false", value: "false", want: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			cmd := NewCommand(slog.Default())
+
+			if err := cmd.Flags().Set(flagPublish, tt.value); err != nil {
+				t.Fatalf("setting --%s flag: %v", flagPublish, err)
+			}
+
+			flag := cmd.Flags().Lookup(flagPublish)
+			if flag == nil {
+				t.Fatalf("--%s flag is not registered", flagPublish)
+			}
+
+			if !flag.Changed {
+				t.Fatal("--publish.Changed = false after explicitly setting the flag, want true")
+			}
+
+			publish, err := cmd.Flags().GetBool(flagPublish)
+			if err != nil {
+				t.Fatalf("getting %s flag: %v", flagPublish, err)
+			}
+
+			if publish != tt.want {
+				t.Fatalf("--%s = %v, want %v", flagPublish, publish, tt.want)
+			}
+		})
+	}
+}
+
+// TestNewCommand_PublishDocumentation verifies the command's Long/Example text documents
+// --publish and its bearer-token requirement, since a certificate-based kubeconfig silently
+// fails (401) through the published path -- this must be discoverable from --help.
+func TestNewCommand_PublishDocumentation(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		fragment string
+	}{
+		{name: "flag mention", fragment: "--publish"},
+		{name: "ingress endpoint", fragment: "storage-foundation-published Ingress endpoint"},
+		{name: "bearer token requirement", fragment: "works only with a kubeconfig"},
+		{name: "certificate rejection", fragment: "receives a 401 when --publish=true"},
+		{name: "example usage", fragment: "--publish=true"},
+	}
+
+	cmd := NewCommand(slog.Default())
+	combined := cmd.Long + "\n" + cmd.Example
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			if !strings.Contains(combined, tc.fragment) {
+				t.Errorf("command Long/Example does not contain %q:\n%s", tc.fragment, combined)
+			}
+		})
+	}
+}

internal/snapshot/snapimport/fs.go

@@ -394,7 +394,8 @@ func headFileOffset(ctx context.Context, client httpDoer, fileURL string, totalS
 		return 0, false, 0, nil
 
 	default:
-		return 0, false, 0, fmt.Errorf("HEAD %s returned status %d (%s)", fileURL, resp.StatusCode, resp.Status)
+		return 0, false, 0, uploadStatusError(resp.StatusCode,
+			fmt.Errorf("HEAD %s returned status %d (%s)", fileURL, resp.StatusCode, resp.Status))
 	}
 }
 
@@ -445,7 +446,8 @@ func doFileChunk(client httpDoer, req *http.Request, offset, requestEnd, totalSi
 	}
 
 	if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusNoContent {
-		return 0, false, fmt.Errorf("server error at offset %d: status %d (%s)", offset, resp.StatusCode, resp.Status)
+		return 0, false, uploadStatusError(resp.StatusCode,
+			fmt.Errorf("server error at offset %d: status %d (%s)", offset, resp.StatusCode, resp.Status))
 	}
 
 	if err := bodyReport.validateExact(); err != nil {

internal/snapshot/snapimport/fs_test.go

@@ -621,6 +621,85 @@ func TestDoFileChunk_StrictStatusesAndOffsets(t *testing.T) {
 	}
 }
 
+// TestHeadFileOffset_ClassifiesUnauthorized verifies headFileOffset's default (non-OK/
+// non-NotFound) branch wraps errUploadUnauthorized only for 401/403 responses, mirroring
+// headBlockOffset's block-path classification.
+func TestHeadFileOffset_ClassifiesUnauthorized(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		statusCode int
+		wantWrap   bool
+	}{
+		{name: "success: 401 wraps sentinel", statusCode: http.StatusUnauthorized, wantWrap: true},
+		{name: "success: 403 wraps sentinel", statusCode: http.StatusForbidden, wantWrap: true},
+		{name: "success: 500 does not wrap sentinel", statusCode: http.StatusInternalServerError, wantWrap: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			doer := fileHTTPDoer(func(*http.Request) (*http.Response, error) {
+				return fileHTTPResponse(tc.statusCode, http.Header{}), nil
+			})
+
+			_, _, _, err := headFileOffset(context.Background(), doer, "https://import.example/file", 10)
+			if err == nil {
+				t.Fatal("headFileOffset unexpectedly returned nil error")
+			}
+
+			if got := errors.Is(err, errUploadUnauthorized); got != tc.wantWrap {
+				t.Errorf("errors.Is(err, errUploadUnauthorized) = %v, want %v (err=%v)", got, tc.wantWrap, err)
+			}
+		})
+	}
+}
+
+// TestDoFileChunk_ClassifiesUnauthorized verifies doFileChunk's non-Created/non-NoContent/
+// non-Conflict branch wraps errUploadUnauthorized only for 401/403 responses; the "want status
+// mismatch" branches below it can never observe 401/403 since they only run once the status is
+// already Created or NoContent.
+func TestDoFileChunk_ClassifiesUnauthorized(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		statusCode int
+		wantWrap   bool
+	}{
+		{name: "success: 401 wraps sentinel", statusCode: http.StatusUnauthorized, wantWrap: true},
+		{name: "success: 403 wraps sentinel", statusCode: http.StatusForbidden, wantWrap: true},
+		{name: "success: 500 does not wrap sentinel", statusCode: http.StatusInternalServerError, wantWrap: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			doer := fileHTTPDoer(func(*http.Request) (*http.Response, error) {
+				return fileHTTPResponse(tc.statusCode, http.Header{}), nil
+			})
+
+			req, err := http.NewRequest(http.MethodPut, "https://import.example/file", bytes.NewReader([]byte("x")))
+			if err != nil {
+				t.Fatalf("build request: %v", err)
+			}
+			req.ContentLength = 1
+
+			_, _, err = doFileChunk(doer, req, 0, 1, 1)
+			if err == nil {
+				t.Fatal("doFileChunk unexpectedly returned nil error")
+			}
+
+			if got := errors.Is(err, errUploadUnauthorized); got != tc.wantWrap {
+				t.Errorf("errors.Is(err, errUploadUnauthorized) = %v, want %v (err=%v)", got, tc.wantWrap, err)
+			}
+		})
+	}
+}
+
 func TestPutFile_SingleShotUpload_CorrectHeaders(t *testing.T) {
 	payload := []byte("hello, filesystem import")