From 9a41a8b71ecadc8cbee9e168b197dfb763669b74 Mon Sep 17 00:00:00 2001 From: Konstantin Kozoriz Date: Tue, 25 Aug 2026 15:07:16 +0300 Subject: [PATCH 1/4] feat(snapshot): add publish-mode TLS primitives to transport and safe client Add ValidateHTTPSURL (transport) and NewSafeClientForConfig (safe client) as the foundational pieces needed to build an HTTPS client against a published (ingress) DataImport endpoint. Signed-off-by: Konstantin Kozoriz --- internal/snapshot/transport/http.go | 8 ++++ internal/snapshot/transport/http_test.go | 58 ++++++++++++++++++++++ pkg/libsaferequest/client/http.go | 7 +++ pkg/libsaferequest/client/http_test.go | 61 ++++++++++++++++++++++++ 4 files changed, 134 insertions(+) create mode 100644 pkg/libsaferequest/client/http_test.go diff --git a/internal/snapshot/transport/http.go b/internal/snapshot/transport/http.go index cd5ff0710..15793e15d 100644 --- a/internal/snapshot/transport/http.go +++ b/internal/snapshot/transport/http.go @@ -1221,6 +1221,14 @@ func (c *Client) SetTLSCAData(caData []byte) { } } +// ValidateHTTPSURL requires rawURL to be a well-formed HTTPS origin, without +// requiring a CA (unlike ValidateHTTPSIdentity, whose CA argument may be empty +// on the publish path). +func ValidateHTTPSURL(rawURL string) error { + _, err := parseHTTPSOrigin(rawURL) + return err +} + // ValidateHTTPSIdentity requires an HTTPS origin and a strictly parseable, // non-empty PEM certificate bundle suitable for endpoint-specific trust. func ValidateHTTPSIdentity(rawURL string, caData []byte) error { diff --git a/internal/snapshot/transport/http_test.go b/internal/snapshot/transport/http_test.go index 256059d00..31bad993e 100644 --- a/internal/snapshot/transport/http_test.go +++ b/internal/snapshot/transport/http_test.go @@ -1212,6 +1212,64 @@ func TestTLSIdentityClient_FailsClosed(t *testing.T) { } } +// TestValidateHTTPSURL verifies ValidateHTTPSURL accepts only a well-formed HTTPS origin, +// without requiring a CA argument (unlike ValidateHTTPSIdentity) -- the publish upload path +// (status.ca empty by design behind Ingress) relies on this weaker check. +func TestValidateHTTPSURL(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + rawURL string + wantErr bool + }{ + { + name: "success: well-formed HTTPS origin", + rawURL: "https://importer.example.test:8443", + }, + { + name: "error: plaintext HTTP origin", + rawURL: "http://importer.example.test", + wantErr: true, + }, + { + name: "error: empty string", + rawURL: "", + wantErr: true, + }, + { + name: "error: malformed URL", + rawURL: "://bad-url", + wantErr: true, + }, + { + name: "error: scheme-less host", + rawURL: "importer.example.test", + wantErr: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + err := ValidateHTTPSURL(tc.rawURL) + + if tc.wantErr { + if err == nil { + t.Fatalf("ValidateHTTPSURL(%q) = nil, want error", tc.rawURL) + } + + return + } + + if err != nil { + t.Fatalf("ValidateHTTPSURL(%q) = %v, want nil", tc.rawURL, err) + } + }) + } +} + func newPersistentTLSServer( t *testing.T, serial int64, diff --git a/pkg/libsaferequest/client/http.go b/pkg/libsaferequest/client/http.go index ef967a83c..a3c416ca7 100644 --- a/pkg/libsaferequest/client/http.go +++ b/pkg/libsaferequest/client/http.go @@ -62,6 +62,13 @@ func NewSafeClient(flags ...*pflag.FlagSet) (*SafeClient, error) { return &SafeClient{restConfig}, nil } +// NewSafeClientForConfig derives a SafeClient from an already-resolved REST +// configuration instead of re-parsing --kubeconfig/--context flags, so a probe +// built from it targets the same cluster the caller already resolved. +func NewSafeClientForConfig(config *rest.Config) *SafeClient { + return &SafeClient{restConfig: rest.CopyConfig(config)} +} + // SetProbeEndpoint configures host, TLS ServerName and timeout for probe requests. func (c *SafeClient) SetProbeEndpoint(timeout time.Duration, targetHost, kubeServiceServerName string) { c.restConfig.Host = targetHost diff --git a/pkg/libsaferequest/client/http_test.go b/pkg/libsaferequest/client/http_test.go new file mode 100644 index 000000000..f62b27bed --- /dev/null +++ b/pkg/libsaferequest/client/http_test.go @@ -0,0 +1,61 @@ +/* +Copyright 2026 Flant JSC + +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 client + +import ( + "testing" + "time" + + "k8s.io/client-go/rest" +) + +func TestNewSafeClientForConfig(t *testing.T) { + t.Parallel() + + t.Run("success: mutating the derived client leaves the original config untouched", func(t *testing.T) { + t.Parallel() + + original := &rest.Config{Host: "https://original.example:6443"} + + derived := NewSafeClientForConfig(original) + derived.SetProbeEndpoint(5*time.Second, "https://probe.example:443", "probe.example") + + if original.Host != "https://original.example:6443" { + t.Errorf("original.Host = %q, want unchanged", original.Host) + } + + if original.TLSClientConfig.ServerName != "" { + t.Errorf("original.TLSClientConfig.ServerName = %q, want unchanged (empty)", original.TLSClientConfig.ServerName) + } + + if derived.restConfig.Host != "https://probe.example:443" { + t.Errorf("derived.restConfig.Host = %q, want %q", derived.restConfig.Host, "https://probe.example:443") + } + }) + + t.Run("error: nil config panics inside rest.CopyConfig", func(t *testing.T) { + t.Parallel() + + defer func() { + if recover() == nil { + t.Fatal("NewSafeClientForConfig(nil) did not panic; rest.CopyConfig no longer dereferences a nil config") + } + }() + + NewSafeClientForConfig(nil) + }) +} From cd928cec63891dd5ac59e5dcae736175dcc434be Mon Sep 17 00:00:00 2001 From: Konstantin Kozoriz Date: Tue, 25 Aug 2026 15:07:33 +0300 Subject: [PATCH 2/4] feat(snapshot): support publish endpoint in DataImport upload Add spec.publish to the DataImport built by d8 snapshot upload, align it alongside spec.ttl on reuse, wait on status.publicURL when publish is enabled, and switch to a merged TLS trust pool with 401/403 diagnostics on that path. Signed-off-by: Konstantin Kozoriz --- internal/snapshot/snapimport/fs.go | 6 +- internal/snapshot/snapimport/fs_test.go | 79 ++ internal/snapshot/snapimport/volume.go | 189 ++++- internal/snapshot/snapimport/volume_test.go | 813 +++++++++++++++++++- 4 files changed, 1039 insertions(+), 48 deletions(-) diff --git a/internal/snapshot/snapimport/fs.go b/internal/snapshot/snapimport/fs.go index 07d940287..231d8bc0a 100644 --- a/internal/snapshot/snapimport/fs.go +++ b/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 { diff --git a/internal/snapshot/snapimport/fs_test.go b/internal/snapshot/snapimport/fs_test.go index 022798d95..72269c0c1 100644 --- a/internal/snapshot/snapimport/fs_test.go +++ b/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") diff --git a/internal/snapshot/snapimport/volume.go b/internal/snapshot/snapimport/volume.go index 6106a2ed9..a5f36a223 100644 --- a/internal/snapshot/snapimport/volume.go +++ b/internal/snapshot/snapimport/volume.go @@ -108,12 +108,26 @@ var dataImportGVR = schema.GroupVersionResource{Group: "storage-foundation.deckh var ErrForeignDataImport = errors.New("foreign DataImport collision") // errDataImportRecheck signals EnsureDataImport that the DataImport changed under our feet -// while alignDataImportTTL was retrying a conflicting update — it was deleted, or it +// while alignDataImportSpec was retrying a conflicting update — it was deleted, or it // transitioned to Ready=False/Expired — so the object must be re-evaluated from the top of // EnsureDataImport's reconcile loop (recreate or delete-and-recreate) rather than patched as // if it were still the healthy object the caller started with. var errDataImportRecheck = errors.New("DataImport changed during TTL alignment") +// errUploadUnauthorized marks an importer response that rejected the client's +// identity (401) or permissions (403). +var errUploadUnauthorized = errors.New("importer rejected the client identity") + +// uploadStatusError wraps err with errUploadUnauthorized when statusCode is 401 or 403, so +// callers can detect a rejected identity via errors.Is without parsing message text. +func uploadStatusError(statusCode int, err error) error { + if statusCode == http.StatusUnauthorized || statusCode == http.StatusForbidden { + return fmt.Errorf("%w: %w", err, errUploadUnauthorized) + } + + return err +} + // VolumeImporter imports a data leaf's volume bytes by creating an SVDM DataImport, // waiting for the importer to be ready, streaming the archive bytes, finalising the // upload, and waiting for the durable artifact to be produced. It is satisfied by @@ -150,6 +164,7 @@ type clusterVolumeImporter struct { sc *transport.Client newUploadClient func([]byte, string) (uploadHTTPClient, error) ttl string + publish bool poll time.Duration wait time.Duration requestTimeout time.Duration @@ -158,27 +173,42 @@ type clusterVolumeImporter struct { log *slog.Logger } -// NewClusterVolumeImporter builds the live VolumeImporter. ttl is the DataImport TTL, -// wait bounds the per-DataImport readiness/completion waits, and poll is the polling -// cadence. Block-volume uploads stream-decode directly into the PUT (see putBlock), so -// no scratch directory for decompressed temporary files is needed. -func NewClusterVolumeImporter( - dyn dynamic.Interface, - sc *transport.Client, - ttl string, - wait, poll time.Duration, - log *slog.Logger, -) VolumeImporter { +// ClusterVolumeImporterOptions configures NewClusterVolumeImporter. +type ClusterVolumeImporterOptions struct { + // Dynamic is the dynamic client used for DataImport CR lifecycle and status polling. + Dynamic dynamic.Interface + // Transport authenticates the HTTPS byte upload to the importer pod (or, when Publish + // is set, the published ingress endpoint). + Transport *transport.Client + // TTL is the idle TTL applied to every DataImport this importer creates or reuses. + TTL string + // Publish selects the published (ingress) upload endpoint instead of the in-cluster + // importer service. + Publish bool + // Wait bounds the per-DataImport readiness/completion waits. + Wait time.Duration + // Poll is the polling cadence used while waiting on DataImport status. + Poll time.Duration + // Log receives lifecycle events; a nil Log defaults to slog.Default(). + Log *slog.Logger +} + +// NewClusterVolumeImporter builds the live VolumeImporter. Block-volume uploads +// stream-decode directly into the PUT (see putBlock), so no scratch directory for +// decompressed temporary files is needed. +func NewClusterVolumeImporter(opts ClusterVolumeImporterOptions) VolumeImporter { + log := opts.Log if log == nil { log = slog.Default() } return &clusterVolumeImporter{ - dyn: dyn, - sc: sc, - ttl: ttl, - poll: poll, - wait: wait, + dyn: opts.Dynamic, + sc: opts.Transport, + ttl: opts.TTL, + publish: opts.Publish, + poll: opts.Poll, + wait: opts.Wait, requestTimeout: DefaultControlRequestTimeout, newRequestContext: context.WithTimeout, fsDecodeDeps: defaultFSDecodeDependencies(), @@ -243,8 +273,12 @@ func (c *clusterVolumeImporter) EnsureDataImport(ctx context.Context, leaf Plann } spec := map[string]interface{}{ - "ttl": c.ttl, - "mode": dataImportModePopulateData, + "ttl": c.ttl, + // Always set explicitly, including false: alignDataImportSpec compares this + // field against the server's returned value, and an implicit server-side + // default would make that comparison depend on a value we never sent. + "publish": c.publish, + "mode": dataImportModePopulateData, "snapshotRef": map[string]interface{}{ "apiVersion": leaf.APIVersion, "kind": leaf.Kind, @@ -274,9 +308,10 @@ func (c *clusterVolumeImporter) EnsureDataImport(ctx context.Context, leaf Plann } if !conditionFalseWithReason(existing, conditionReady, reasonExpired) { - // Align spec.ttl with the current run so retrying a stalled import with a - // longer --ttl is honoured instead of keeping the first create's value. - if tErr := c.alignDataImportTTL(ctx, ri, existing, leaf); tErr != nil { + // Align spec.ttl and spec.publish with the current run so retrying a stalled + // import with a longer --ttl, or a different --publish, is honoured instead + // of keeping the first create's values. + if tErr := c.alignDataImportSpec(ctx, ri, existing, leaf); tErr != nil { if errors.Is(tErr, errDataImportRecheck) { continue } @@ -333,6 +368,10 @@ func dataImportShortID(leaf PlannedNode) string { return leaf.DataImportIdentity[:dataImportIdentityIDLength] } +// dataImportAnnotations intentionally excludes spec.publish: publish is a property of +// how the CLI transports bytes to an existing DataImport, not of the archive content it +// identifies, so a --publish value change on a retry must not turn a reused DataImport +// into a foreign one (see validateDataImportSpec) and block resume. func dataImportAnnotations(leaf PlannedNode) map[string]string { return map[string]string{ dataImportIdentityVersionAnnotation: dataImportIdentityVersion, @@ -370,6 +409,9 @@ func validateDataImportMetadata(obj *unstructured.Unstructured, leaf PlannedNode return nil } +// validateDataImportSpec does not check spec.publish for the same reason +// dataImportAnnotations does not carry it: publish is a transport property, not part of +// the leaf's content identity. func validateDataImportSpec(obj *unstructured.Unstructured, leaf PlannedNode) error { mode, found, err := unstructured.NestedString(obj.Object, "spec", "mode") if err != nil || !found || mode != dataImportModePopulateData { @@ -413,17 +455,24 @@ func validateDataImportSpec(obj *unstructured.Unstructured, leaf PlannedNode) er return nil } -// alignDataImportTTL patches a reused DataImport's spec.ttl to the current run's TTL when it -// drifted, so increasing --ttl on a retry takes effect. No-op when already aligned. +// alignDataImportSpec patches a reused DataImport's spec.ttl and spec.publish to the +// current run's intent when either drifted, so increasing --ttl on a retry, or changing +// --publish between runs, takes effect. No-op when both are already aligned. publish is +// aligned in both directions (including true -> false): a prior --publish=true run must +// not keep exposing the upload publicly once a later run asks for the in-cluster path. +// +// Both fields share one Update because they land on the same object in the same +// reconcile pass; a second Update here would open a second, redundant conflict window on +// top of the one this function's own retry loop already handles. // // The whole body runs inside retry.RetryOnConflict, mirroring reconcileExistingMarker: a // conflicting Update forces a re-Get on the next attempt (current = nil), and the re-Get's // result is re-validated against leaf and re-checked for the Expired condition before any -// patch is attempted, so a concurrent run cannot cause a stale-revision TTL patch to land on +// patch is attempted, so a concurrent run cannot cause a stale-revision patch to land on // a foreign or already-expired object. errDataImportRecheck is returned as-is (never wrapped) // when the re-Get finds the object gone or expired, so EnsureDataImport's caller can tell // "retry from scratch" apart from a genuine patch failure via errors.Is. -func (c *clusterVolumeImporter) alignDataImportTTL(ctx context.Context, ri dynamic.ResourceInterface, existing *unstructured.Unstructured, leaf PlannedNode) error { +func (c *clusterVolumeImporter) alignDataImportSpec(ctx context.Context, ri dynamic.ResourceInterface, existing *unstructured.Unstructured, leaf PlannedNode) error { current := existing didUpdate := false @@ -452,8 +501,10 @@ func (c *clusterVolumeImporter) alignDataImportTTL(ctx context.Context, ri dynam return errDataImportRecheck } - cur, _, _ := unstructured.NestedString(current.Object, "spec", "ttl") - if cur == c.ttl { + curTTL, _, _ := unstructured.NestedString(current.Object, "spec", "ttl") + curPublish, _, _ := unstructured.NestedBool(current.Object, "spec", "publish") + + if curTTL == c.ttl && curPublish == c.publish { return nil } @@ -462,6 +513,10 @@ func (c *clusterVolumeImporter) alignDataImportTTL(ctx context.Context, ri dynam return fmt.Errorf("set DataImport ttl: %w", setErr) } + if setErr := unstructured.SetNestedField(candidate.Object, c.publish, "spec", "publish"); setErr != nil { + return fmt.Errorf("set DataImport publish: %w", setErr) + } + patched, updateErr := runControlRequest(ctx, c.requestTimeout, c.newRequestContext, func(requestCtx context.Context) (*unstructured.Unstructured, error) { return ri.Update(requestCtx, candidate, metav1.UpdateOptions{}) @@ -482,12 +537,13 @@ func (c *clusterVolumeImporter) alignDataImportTTL(ctx context.Context, ri dynam return err } - return fmt.Errorf("patch DataImport %s/%s ttl: %w", existing.GetNamespace(), existing.GetName(), err) + return fmt.Errorf("patch DataImport %s/%s spec: %w", existing.GetNamespace(), existing.GetName(), err) } if didUpdate { - c.log.Info("aligned DataImport ttl", - slog.String("namespace", existing.GetNamespace()), slog.String("name", existing.GetName()), slog.String("ttl", c.ttl)) + c.log.Info("aligned DataImport spec", + slog.String("namespace", existing.GetNamespace()), slog.String("name", existing.GetName()), + slog.String("ttl", c.ttl), slog.Bool("publish", c.publish)) } return nil @@ -563,7 +619,7 @@ func (c *clusterVolumeImporter) UploadVolumeData(ctx context.Context, leaf Plann return err } - url, _, _ := unstructured.NestedString(di.Object, "status", "url") + url := c.uploadBaseURL(di) volumeMode, _, _ := unstructured.NestedString(di.Object, "status", "volumeMode") caB64, _, _ := unstructured.NestedString(di.Object, "status", "ca") @@ -574,12 +630,30 @@ func (c *clusterVolumeImporter) UploadVolumeData(ctx context.Context, leaf Plann defer httpClient.CloseIdleConnections() if err := c.sendVolumeData(ctx, httpClient, url, volumeMode, leaf, namespace, diName, setTotal, onProgress, activate); err != nil { + if c.publish && errors.Is(err, errUploadUnauthorized) { + return fmt.Errorf("%w; ingress does not forward client TLS certificates — "+ + "use a kubeconfig with a bearer token, or --publish=false from inside the cluster", err) + } + return err } return c.waitDataImportCompleted(ctx, leaf, diName, namespace) } +// uploadBaseURL returns the endpoint clients must upload to: the published +// ingress URL when publish is enabled, otherwise the in-cluster importer URL. +func (c *clusterVolumeImporter) uploadBaseURL(di *unstructured.Unstructured) string { + field := "url" + if c.publish { + field = "publicURL" + } + + url, _, _ := unstructured.NestedString(di.Object, "status", field) + + return url +} + func verifyLeafPayloadCurrent(ctx context.Context, leaf PlannedNode) error { if leaf.archiveView == nil || leaf.payloadFile == nil { return nil @@ -761,8 +835,22 @@ func (c *clusterVolumeImporter) uploadClient(caB64, rawURL string) (uploadHTTPCl return nil, fmt.Errorf("decode DataImport status.ca: %w", err) } - if err := transport.ValidateHTTPSIdentity(rawURL, ca); err != nil { - return nil, fmt.Errorf("validate DataImport upload identity: %w", err) + // Through Ingress, TLS terminates at ingress-nginx's own certificate, which never chains + // to the importer pod's internal CA — endpoint-specific pinning is impossible there, so + // publish=true trades it for a merged trust pool (below). Confined to that branch; + // insecure-skip-tls-verify inherited from kubeconfig is untouched either way. + if c.publish { + if err := transport.ValidateHTTPSURL(rawURL); err != nil { + return nil, fmt.Errorf("validate DataImport publish upload URL: %w", err) + } + } + + // status.ca is normally empty on the publish path (no internal CA to pin to there), so + // identity pinning is skipped only in that case; it stays mandatory otherwise. + if !c.publish || len(ca) > 0 { + if err := transport.ValidateHTTPSIdentity(rawURL, ca); err != nil { + return nil, fmt.Errorf("validate DataImport upload identity: %w", err) + } } if c.newUploadClient != nil { @@ -776,8 +864,14 @@ func (c *clusterVolumeImporter) uploadClient(caB64, rawURL string) (uploadHTTPCl sub := c.sc.Copy() sub.SetRequestTimeout(0) - if err := sub.SetTLSIdentityCAData(ca); err != nil { - return nil, fmt.Errorf("configure DataImport upload TLS identity: %w", err) + if c.publish { + sub.SetTLSCAData(ca) + } + + if !c.publish { + if err := sub.SetTLSIdentityCAData(ca); err != nil { + return nil, fmt.Errorf("configure DataImport upload TLS identity: %w", err) + } } if err := sub.SetNetworkTimeouts(transport.NetworkTimeouts{ @@ -801,7 +895,9 @@ func (c *clusterVolumeImporter) uploadClient(caB64, rawURL string) (uploadHTTPCl } // waitDataImportReady blocks until the DataImport reports Ready=True with a populated -// status.url and volumeMode. +// volumeMode and upload endpoint — status.url normally, or status.publicURL when c.publish +// is set (the controller never revokes Ready once granted, so a DataImport republished after +// becoming Ready can sit at Ready=True with an empty publicURL until the ingress catches up). func (c *clusterVolumeImporter) waitDataImportReady( ctx context.Context, leaf PlannedNode, @@ -829,7 +925,7 @@ func (c *clusterVolumeImporter) waitDataImportReady( return nil, fmt.Errorf("data import %s/%s expired before becoming Ready (idle TTL elapsed); increase --ttl or retry", namespace, name) } - url, _, _ := unstructured.NestedString(di.Object, "status", "url") + url := c.uploadBaseURL(di) volumeMode, _, _ := unstructured.NestedString(di.Object, "status", "volumeMode") if conditionTrue(di, conditionReady) && url != "" && volumeMode != "" { @@ -837,6 +933,14 @@ func (c *clusterVolumeImporter) waitDataImportReady( } if time.Now().After(deadline) { + if c.publish && url == "" { + return nil, fmt.Errorf( + "timeout waiting for DataImport %s/%s to become Ready: status.publicURL is still empty; "+ + "the storage-foundation ingress may not be configured, or retry with --publish=false from inside the cluster", + namespace, name, + ) + } + return nil, fmt.Errorf("timeout waiting for DataImport %s/%s to become Ready", namespace, name) } @@ -2147,7 +2251,8 @@ func headBlockOffset(ctx context.Context, httpClient httpDoer, url string, total return 0, nil default: - return 0, fmt.Errorf("HEAD %s returned status %d (%s)", url, resp.StatusCode, resp.Status) + return 0, uploadStatusError(resp.StatusCode, + fmt.Errorf("HEAD %s returned status %d (%s)", url, resp.StatusCode, resp.Status)) } } @@ -2202,7 +2307,8 @@ func doBlockChunk(httpClient httpDoer, req *http.Request, offset, requestEnd, to } 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 { @@ -2277,7 +2383,8 @@ func postFinished(ctx context.Context, httpClient httpDoer, baseURL string) erro } if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { - return fmt.Errorf("finished returned status %d (%s)", resp.StatusCode, resp.Status) + return uploadStatusError(resp.StatusCode, + fmt.Errorf("finished returned status %d (%s)", resp.StatusCode, resp.Status)) } return nil diff --git a/internal/snapshot/snapimport/volume_test.go b/internal/snapshot/snapimport/volume_test.go index 77c834969..406fa0fa8 100644 --- a/internal/snapshot/snapimport/volume_test.go +++ b/internal/snapshot/snapimport/volume_test.go @@ -20,11 +20,15 @@ import ( gotar "archive/tar" "bytes" "context" + "crypto/ed25519" + "crypto/x509" + "crypto/x509/pkix" "encoding/base64" "encoding/pem" "errors" "fmt" "io" + "math/big" "math/rand" "net" "net/http" @@ -46,6 +50,7 @@ import ( k8sruntime "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" dynamicfake "k8s.io/client-go/dynamic/fake" + restclient "k8s.io/client-go/rest" clienttesting "k8s.io/client-go/testing" diapi "github.com/deckhouse/deckhouse-cli/internal/data/dataimport/api/v1alpha1" @@ -1637,6 +1642,170 @@ func TestUploadControlEndpoints_PropagateResponseByteLimit(t *testing.T) { } } +// TestUploadStatusError verifies the errUploadUnauthorized classifier: it wraps the sentinel +// only for HTTP 401/403 status codes (by code, never by message text), leaving every other +// status's error unwrapped so errors.Is(err, errUploadUnauthorized) stays false for them. +func TestUploadStatusError(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: 404 does not wrap sentinel", statusCode: http.StatusNotFound, wantWrap: false}, + {name: "success: 409 does not wrap sentinel", statusCode: http.StatusConflict, wantWrap: false}, + {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() + + base := fmt.Errorf("status %d", tc.statusCode) + + err := uploadStatusError(tc.statusCode, base) + + if got := errors.Is(err, errUploadUnauthorized); got != tc.wantWrap { + t.Errorf("errors.Is(err, errUploadUnauthorized) = %v, want %v (err=%v)", got, tc.wantWrap, err) + } + + if !errors.Is(err, base) && tc.wantWrap { + t.Errorf("wrapped error lost the original base error: %v", err) + } + }) + } +} + +// TestHeadBlockOffset_ClassifiesUnauthorized verifies headBlockOffset's default (non-OK/ +// non-NotFound) branch wraps errUploadUnauthorized only for 401/403 responses. +func TestHeadBlockOffset_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 := testHTTPDoer(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: tc.statusCode, + Status: fmt.Sprintf("%d %s", tc.statusCode, http.StatusText(tc.statusCode)), + Header: http.Header{}, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }) + + _, err := headBlockOffset(context.Background(), doer, "https://importer.test/api/v1/block", 10) + if err == nil { + t.Fatal("headBlockOffset 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) + } + }) + } +} + +// TestDoBlockChunk_ClassifiesUnauthorized verifies doBlockChunk's non-Created/non-NoContent/ +// non-Conflict branch wraps errUploadUnauthorized only for 401/403 responses; the "want status +// mismatch" branches below it can never see 401/403 (they only run once the status is already +// Created or NoContent), so this exercises the only reachable wrap site. +func TestDoBlockChunk_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 := testHTTPDoer(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: tc.statusCode, + Status: fmt.Sprintf("%d %s", tc.statusCode, http.StatusText(tc.statusCode)), + Header: http.Header{}, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }) + + req, err := http.NewRequestWithContext(context.Background(), http.MethodPut, "https://importer.test/api/v1/block", nil) + if err != nil { + t.Fatalf("build request: %v", err) + } + + _, _, err = doBlockChunk(doer, req, 0, 1, 1) + if err == nil { + t.Fatal("doBlockChunk 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) + } + }) + } +} + +// TestPostFinished_ClassifiesUnauthorized verifies postFinished's non-2xx branch wraps +// errUploadUnauthorized only for 401/403 responses. +func TestPostFinished_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 := testHTTPDoer(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: tc.statusCode, + Status: fmt.Sprintf("%d %s", tc.statusCode, http.StatusText(tc.statusCode)), + Header: http.Header{}, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }) + + err := postFinished(context.Background(), doer, "https://importer.test") + if err == nil { + t.Fatal("postFinished 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 TestSendVolumeData_WriteDeadlineLeavesResumeOffsetAndSkipsFinished(t *testing.T) { t.Parallel() @@ -4440,6 +4609,130 @@ func readyDataImportObj(leaf PlannedNode, rawURL, volumeMode, ca string) *unstru return obj } +// readyDataImportObjWithPublicURL builds a Ready DataImport with BOTH status.url and +// status.publicURL independently settable, for exercising uploadBaseURL/waitDataImportReady's +// publish-vs-non-publish branch selection precisely. +func readyDataImportObjWithPublicURL(leaf PlannedNode, url, publicURL, volumeMode, ca string) *unstructured.Unstructured { + obj := dataImportObjForLeaf(targetNS, leaf, false) + _ = unstructured.SetNestedSlice(obj.Object, readyConditions(conditionReady), "status", "conditions") + _ = unstructured.SetNestedField(obj.Object, volumeMode, "status", "volumeMode") + _ = unstructured.SetNestedField(obj.Object, ca, "status", "ca") + + if url != "" { + _ = unstructured.SetNestedField(obj.Object, url, "status", "url") + } + + if publicURL != "" { + _ = unstructured.SetNestedField(obj.Object, publicURL, "status", "publicURL") + } + + return obj +} + +// newTestVolumeImporterWithWait builds a clusterVolumeImporter like newTestVolumeImporter but +// with an explicit (short) wait budget, for tests that must observe a timeout without paying +// the default 2-second wait. +func newTestVolumeImporterWithWait(dyn *dynamicfake.FakeDynamicClient, publish bool, wait time.Duration) *clusterVolumeImporter { + imp := newTestVolumeImporter(dyn) + imp.publish = publish + imp.wait = wait + imp.poll = time.Millisecond + + return imp +} + +// TestWaitDataImportReady_PublishRequiresPublicURL is the most important regression guard for +// the publish readiness contract: the storage-foundation controller never lowers Ready back +// to False once it is True, so a DataImport that is Ready=True with status.url populated but +// status.publicURL still empty (Ingress wiring lagging behind the importer pod) must NOT be +// treated as ready when publish=true -- it must keep waiting until it times out. +func TestWaitDataImportReady_PublishRequiresPublicURL(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + di := readyDataImportObjWithPublicURL(leaf, "https://in-cluster.test", "", volumeModeBlock, "") + + dyn := newFakeDataImportDyn(di) + imp := newTestVolumeImporterWithWait(dyn, true, 60*time.Millisecond) + + start := time.Now() + _, err := imp.waitDataImportReady(context.Background(), leaf, imp.DataImportName(leaf), targetNS) + elapsed := time.Since(start) + + if err == nil { + t.Fatal("waitDataImportReady returned nil, want a timeout error (Ready=True but publicURL empty must not satisfy publish readiness)") + } + + if !strings.Contains(err.Error(), "publicURL") { + t.Errorf("error = %v, want it to mention publicURL", err) + } + + if elapsed < imp.wait { + t.Errorf("waitDataImportReady returned after %v, want it to wait out the full %v budget", elapsed, imp.wait) + } +} + +// TestWaitDataImportReady_PublishUsesPublicURL verifies that once status.publicURL is +// populated, publish=true returns immediately using it (not status.url). +func TestWaitDataImportReady_PublishUsesPublicURL(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + di := readyDataImportObjWithPublicURL(leaf, "https://in-cluster.test", "https://published.test", volumeModeBlock, "") + + dyn := newFakeDataImportDyn(di) + imp := newTestVolumeImporterWithWait(dyn, true, time.Second) + + got, err := imp.waitDataImportReady(context.Background(), leaf, imp.DataImportName(leaf), targetNS) + if err != nil { + t.Fatalf("waitDataImportReady: %v", err) + } + + if url := imp.uploadBaseURL(got); url != "https://published.test" { + t.Errorf("uploadBaseURL = %q, want the published URL https://published.test (not status.url)", url) + } +} + +// TestWaitDataImportReady_NonPublishIgnoresPublicURL is the mirror regression guard: with +// publish=false, a populated status.publicURL must never be mistaken for status.url. An empty +// status.url with a populated publicURL must still time out. +func TestWaitDataImportReady_NonPublishIgnoresPublicURL(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + di := readyDataImportObjWithPublicURL(leaf, "", "https://published.test", volumeModeBlock, "") + + dyn := newFakeDataImportDyn(di) + imp := newTestVolumeImporterWithWait(dyn, false, 60*time.Millisecond) + + _, err := imp.waitDataImportReady(context.Background(), leaf, imp.DataImportName(leaf), targetNS) + if err == nil { + t.Fatal("waitDataImportReady returned nil, want a timeout error (publish=false must never use status.publicURL)") + } +} + +// TestWaitDataImportReady_NonPublishUsesURLEvenWithPublicURLSet is a regression guard against +// crossed branches: when both status.url and status.publicURL are populated, publish=false +// must resolve to status.url. +func TestWaitDataImportReady_NonPublishUsesURLEvenWithPublicURLSet(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + di := readyDataImportObjWithPublicURL(leaf, "https://in-cluster.test", "https://published.test", volumeModeBlock, "") + + dyn := newFakeDataImportDyn(di) + imp := newTestVolumeImporterWithWait(dyn, false, time.Second) + + got, err := imp.waitDataImportReady(context.Background(), leaf, imp.DataImportName(leaf), targetNS) + if err != nil { + t.Fatalf("waitDataImportReady: %v", err) + } + + if url := imp.uploadBaseURL(got); url != "https://in-cluster.test" { + t.Errorf("uploadBaseURL = %q, want the in-cluster URL https://in-cluster.test (not publicURL)", url) + } +} + func TestUploadVolumeData_SkipsCompleted(t *testing.T) { // DataFile is set so the block-data preflight passes; the file is never opened because // the completed-import short-circuit returns before any upload. @@ -4519,6 +4812,85 @@ func TestUploadVolumeData_ClosesClientAfterRequestError(t *testing.T) { } } +// TestUploadVolumeData_UnauthorizedHintDependsOnPublish verifies UploadVolumeData's error +// wrapping around errUploadUnauthorized (a 401 from the importer's block HEAD probe): with +// publish=true it appends the bearer-token/kubeconfig hint explaining why a certificate-based +// kubeconfig fails through the published Ingress path; with publish=false the underlying +// error is returned unchanged, since that hint would be misleading for the in-cluster path. +func TestUploadVolumeData_UnauthorizedHintDependsOnPublish(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + publish bool + wantHint bool + }{ + {name: "success: publish=true appends the bearer-token hint", publish: true, wantHint: true}, + {name: "success: publish=false leaves the error unchanged", publish: false, wantHint: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + payload := []byte("unauthorized upload") + dataFile := filepath.Join(t.TempDir(), "data.bin") + if err := os.WriteFile(dataFile, payload, 0o600); err != nil { + t.Fatalf("write block payload: %v", err) + } + + leaf := volumeSnapshotLeaf("pvc-unauthorized") + leaf.DataFile = dataFile + leaf.Size = strconv.Itoa(len(payload)) + leaf.SizeBytes = int64(len(payload)) + leaf.DataImportIdentity = dataImportIdentity(leaf) + + ca := testUploadCA(t) + di := readyDataImportObjWithPublicURL(leaf, "https://importer.test", "https://importer.test", volumeModeBlock, base64.StdEncoding.EncodeToString(ca)) + + importer := newTestVolumeImporter(newFakeDataImportDyn(di)) + importer.publish = tc.publish + + importer.newUploadClient = func([]byte, string) (uploadHTTPClient, error) { + return &testUploadHTTPClient{ + do: func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Status: "401 Unauthorized", + Header: http.Header{}, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }, + close: func() {}, + }, nil + } + + err := importer.UploadVolumeData( + context.Background(), + leaf, + importer.DataImportName(leaf), + targetNS, + nil, + nil, + nil, + ) + if err == nil { + t.Fatal("UploadVolumeData unexpectedly returned nil for a 401 response") + } + + if !errors.Is(err, errUploadUnauthorized) { + t.Fatalf("UploadVolumeData error = %v, want errors.Is(errUploadUnauthorized)", err) + } + + hasHint := strings.Contains(err.Error(), "bearer token") + + if hasHint != tc.wantHint { + t.Errorf("error contains bearer-token hint = %v, want %v (err=%v)", hasHint, tc.wantHint, err) + } + }) + } +} + func TestUploadVolumeData_CancellationClosesOnlyAfterInFlightRequestReturns(t *testing.T) { payload := []byte("cancel upload") dataFile := filepath.Join(t.TempDir(), "data.bin") @@ -4658,6 +5030,211 @@ func testUploadCA(t *testing.T) []byte { return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate.Raw}) } +// TestUploadClientPublish_ValidatesBeforeFactory mirrors +// TestUploadClientRejectsInvalidIdentityBeforeFactory for the publish=true branch of +// uploadClient: ValidateHTTPSURL still requires HTTPS, and a non-empty malformed CA is still +// rejected (fail closed), but an EMPTY CA -- expected on the publish path, since Ingress +// terminates TLS with its own certificate rather than the importer's internal CA -- must be +// accepted and reach the client factory. +func TestUploadClientPublish_ValidatesBeforeFactory(t *testing.T) { + t.Parallel() + + validCA := base64.StdEncoding.EncodeToString(testUploadCA(t)) + + tests := []struct { + name string + rawURL string + ca string + wantErr bool + wantFactory bool + }{ + { + name: "error: plaintext URL rejected even under publish", + rawURL: "http://127.0.0.1:8443", + ca: validCA, + wantErr: true, + }, + { + name: "error: malformed non-empty CA is rejected fail-closed", + rawURL: "https://127.0.0.1:8443", + ca: base64.StdEncoding.EncodeToString([]byte("not PEM")), + wantErr: true, + }, + { + name: "error: invalid base64 CA is rejected", + rawURL: "https://127.0.0.1:8443", + ca: "%%%", + wantErr: true, + }, + { + name: "success: empty CA is accepted and reaches the factory", + rawURL: "https://127.0.0.1:8443", + ca: "", + wantFactory: true, + }, + { + name: "success: valid CA is still accepted and reaches the factory", + rawURL: "https://127.0.0.1:8443", + ca: validCA, + wantFactory: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var factoryCalls atomic.Int64 + + importer := &clusterVolumeImporter{ + publish: true, + newUploadClient: func([]byte, string) (uploadHTTPClient, error) { + factoryCalls.Add(1) + + return &testUploadHTTPClient{close: func() {}}, nil + }, + } + + _, err := importer.uploadClient(tc.ca, tc.rawURL) + + if tc.wantErr { + if err == nil { + t.Fatal("uploadClient unexpectedly accepted invalid identity/URL under publish") + } + } else if err != nil { + t.Fatalf("uploadClient: %v", err) + } + + wantCalls := int64(0) + if tc.wantFactory { + wantCalls = 1 + } + + if got := factoryCalls.Load(); got != wantCalls { + t.Fatalf("upload client factory calls = %d, want %d", got, wantCalls) + } + }) + } +} + +// unrelatedCertificatePEM returns a PEM-encoded self-signed certificate generated with its +// own independent key pair, standing in for a status.ca (or an operator-supplied CA) that does +// not chain to the real upload origin's certificate -- exercising the "foreign CA" side of the +// publish-vs-non-publish pinning tests below. It is deliberately NOT taken from a second +// httptest.NewTLSServer: httptest servers share one built-in default certificate unless +// explicitly configured otherwise, which would make the "unrelated" CA accidentally identical +// to the real origin's and silently defeat the test. +func unrelatedCertificatePEM(t *testing.T) []byte { + t.Helper() + + seed := bytes.Repeat([]byte{0x42}, ed25519.SeedSize) + privateKey := ed25519.NewKeyFromSeed(seed) + template := &x509.Certificate{ + SerialNumber: big.NewInt(9999), + Subject: pkix.Name{CommonName: "unrelated-test-ca"}, + NotBefore: time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC), + NotAfter: time.Date(2035, 1, 1, 0, 0, 0, 0, time.UTC), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + DNSNames: []string{"unrelated.invalid"}, + } + + der, err := x509.CreateCertificate(rand.New(rand.NewSource(1)), template, template, privateKey.Public(), privateKey) + if err != nil { + t.Fatalf("create unrelated test certificate: %v", err) + } + + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} + +// TestUploadClient_PinningDependsOnPublish is the critical regression guard for the +// publish-path trust model: with publish=true, uploadClient merges status.ca into the +// caller's already-configured (kubeconfig) trust pool via SetTLSCAData, so a request to the +// real origin succeeds even though the supplied CA itself is unrelated to it. With +// publish=false, uploadClient calls SetTLSIdentityCAData instead, which REPLACES trust with +// exactly the supplied CA -- so the same unrelated CA must cause the request to the real +// origin to fail. A regression here (e.g. always using SetTLSCAData) would silently widen the +// in-cluster upload path's pinning. +func TestUploadClient_PinningDependsOnPublish(t *testing.T) { + t.Parallel() + + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })) + t.Cleanup(srv.Close) + + certificate := srv.Certificate() + if certificate == nil { + t.Fatal("TLS test server has no certificate") + } + + trustedCA := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate.Raw}) + unrelatedCA := unrelatedCertificatePEM(t) + + tests := []struct { + name string + publish bool + wantErr bool + }{ + { + name: "success: publish merges the unrelated CA with the kubeconfig-trusted pool", + publish: true, + }, + { + name: "error: non-publish pins exclusively to the unrelated CA and rejects the real origin", + publish: false, + wantErr: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + config := &restclient.Config{ + Host: srv.URL, + BearerToken: "must-not-leak", + TLSClientConfig: restclient.TLSClientConfig{ + CAData: trustedCA, + }, + } + + importer := &clusterVolumeImporter{ + sc: transport.NewClientForConfig(config), + publish: tc.publish, + } + + httpClient, err := importer.uploadClient(base64.StdEncoding.EncodeToString(unrelatedCA), srv.URL) + if err != nil { + t.Fatalf("uploadClient: %v", err) + } + t.Cleanup(httpClient.CloseIdleConnections) + + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL, nil) + if err != nil { + t.Fatalf("build request: %v", err) + } + + resp, doErr := httpClient.HTTPDo(req) + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + + if tc.wantErr { + if doErr == nil { + t.Fatal("request unexpectedly succeeded against the real origin under strict (non-publish) pinning to an unrelated CA") + } + + return + } + + if doErr != nil { + t.Fatalf("request against the real origin unexpectedly failed under publish's merged trust pool: %v", doErr) + } + }) + } +} + func TestUploadVolumeData_CompletedReuseRejectsChangedVerifiedPayload(t *testing.T) { payload := []byte("verified completed payload") nodeSpec := archiveNode{ @@ -5006,7 +5583,7 @@ func TestEnsureDataImport_AlignsTTLRetryingConflict(t *testing.T) { // TestEnsureDataImport_TTLConflictRevalidatesFreshObject covers the race the retry loop must // close: if a concurrent writer replaces the DataImport with a foreign one in the window between -// the conflicting Update and alignDataImportTTL's re-Get, the re-Get's result must be +// the conflicting Update and alignDataImportSpec's re-Get, the re-Get's result must be // re-validated against leaf before any patch — never blindly reused from the pre-conflict // revision. func TestEnsureDataImport_TTLConflictRevalidatesFreshObject(t *testing.T) { @@ -5028,7 +5605,7 @@ func TestEnsureDataImport_TTLConflictRevalidatesFreshObject(t *testing.T) { updateCalls++ if updateCalls == 1 { // Simulate a concurrent writer swapping in a foreign DataImport in the exact - // window the conflict forces alignDataImportTTL to re-Get. + // window the conflict forces alignDataImportSpec to re-Get. if err := dyn.Tracker().Update(dataImportGVR, foreign, targetNS); err != nil { return true, nil, fmt.Errorf("swap in foreign DataImport: %w", err) } @@ -5054,7 +5631,7 @@ func TestEnsureDataImport_TTLConflictRevalidatesFreshObject(t *testing.T) { } // TestEnsureDataImport_TTLTargetVanishedRecreates covers a conflict whose re-Get finds the -// DataImport gone: alignDataImportTTL must surface errDataImportRecheck (not a hard failure) so +// DataImport gone: alignDataImportSpec must surface errDataImportRecheck (not a hard failure) so // EnsureDataImport's outer loop re-evaluates from scratch and creates a fresh DataImport. func TestEnsureDataImport_TTLTargetVanishedRecreates(t *testing.T) { leaf := volumeSnapshotLeaf("pvc-1") @@ -5109,7 +5686,7 @@ func TestEnsureDataImport_TTLTargetVanishedRecreates(t *testing.T) { // TestEnsureDataImport_TTLTargetExpiredDuringAlignmentRecreates mirrors // TestEnsureDataImport_RecreatesExpired for the conflict-retry path: if the re-Get after a -// conflicting TTL Update finds the DataImport now Ready=False/Expired, alignDataImportTTL must +// conflicting TTL Update finds the DataImport now Ready=False/Expired, alignDataImportSpec must // surface errDataImportRecheck so EnsureDataImport's outer loop deletes and recreates it instead // of patching a dying object. func TestEnsureDataImport_TTLTargetExpiredDuringAlignmentRecreates(t *testing.T) { @@ -5165,7 +5742,7 @@ func TestEnsureDataImport_TTLTargetExpiredDuringAlignmentRecreates(t *testing.T) } // TestEnsureDataImport_TTLAlreadyAlignedIssuesNoUpdate is a regression anchor for the ordering -// requirement in alignDataImportTTL: the ttl-equality check must run after any re-Get, but it +// requirement in alignDataImportSpec: the ttl-equality check must run after any re-Get, but it // must still short-circuit to zero Updates on the common already-aligned path. func TestEnsureDataImport_TTLAlreadyAlignedIssuesNoUpdate(t *testing.T) { leaf := volumeSnapshotLeaf("pvc-1") @@ -5182,6 +5759,232 @@ func TestEnsureDataImport_TTLAlreadyAlignedIssuesNoUpdate(t *testing.T) { } } +// TestEnsureDataImport_BuildsSpecPublishField verifies EnsureDataImport always sets +// spec.publish explicitly (including false), matching the importer's own Publish option -- +// so alignDataImportSpec's later comparison against the server's returned value never depends +// on an implicit server-side default. +func TestEnsureDataImport_BuildsSpecPublishField(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + publish bool + }{ + {name: "success: publish=false is explicitly set", publish: false}, + {name: "success: publish=true is explicitly set", publish: true}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + + dyn := newFakeDataImportDyn() + imp := newTestVolumeImporter(dyn) + imp.publish = tc.publish + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport: %v", err) + } + + diName := imp.DataImportName(leaf) + + got, err := dyn.Resource(dataImportGVR).Namespace(targetNS).Get(context.Background(), diName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("created DataImport not found: %v", err) + } + + publish, found, err := unstructured.NestedBool(got.Object, "spec", "publish") + if err != nil { + t.Fatalf("read spec.publish: %v", err) + } + + if !found { + t.Fatal("spec.publish was not set, want explicit value") + } + + if publish != tc.publish { + t.Errorf("spec.publish = %v, want %v", publish, tc.publish) + } + }) + } +} + +// TestEnsureDataImport_AlignsPublishOnReuse verifies alignDataImportSpec patches spec.publish +// on a reused DataImport when it drifts from the current run's --publish, in exactly one +// Update, leaving spec.ttl untouched when it already matched. +func TestEnsureDataImport_AlignsPublishOnReuse(t *testing.T) { + leaf := volumeSnapshotLeaf("pvc-1") + + // dataImportObj never sets spec.publish, so it is absent (equivalent to false). + existing := dataImportObj(targetNS, "pvc-1", false) + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) // ttl: "1h" + imp.publish = true + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport: %v", err) + } + + if u := countDataImportActions(dyn, "update"); u != 1 { + t.Errorf("update calls = %d, want exactly 1", u) + } + + diName := imp.DataImportName(leaf) + + got, err := dyn.Resource(dataImportGVR).Namespace(targetNS).Get(context.Background(), diName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get DataImport: %v", err) + } + + publish, _, _ := unstructured.NestedBool(got.Object, "spec", "publish") + if !publish { + t.Error("spec.publish = false, want true (aligned to the current run's --publish)") + } + + ttl, _, _ := unstructured.NestedString(got.Object, "spec", "ttl") + if ttl != "1h" { + t.Errorf("spec.ttl = %q, want unchanged 1h", ttl) + } +} + +// TestEnsureDataImport_AlignsBothTTLAndPublishInOneUpdate is a regression anchor for +// alignDataImportSpec's single-Update contract: when BOTH spec.ttl and spec.publish drift on +// the same reused object, they must be patched together in exactly one Update, never two +// separate ones (which would open a second, redundant conflict window). +func TestEnsureDataImport_AlignsBothTTLAndPublishInOneUpdate(t *testing.T) { + leaf := volumeSnapshotLeaf("pvc-1") + + existing := dataImportObj(targetNS, "pvc-1", false) + _ = unstructured.SetNestedField(existing.Object, "2m", "spec", "ttl") + _ = unstructured.SetNestedField(existing.Object, false, "spec", "publish") + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) // ttl: "1h" + imp.publish = true + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport: %v", err) + } + + if u := countDataImportActions(dyn, "update"); u != 1 { + t.Fatalf("update calls = %d, want exactly 1 (both fields patched together)", u) + } + + diName := imp.DataImportName(leaf) + + got, err := dyn.Resource(dataImportGVR).Namespace(targetNS).Get(context.Background(), diName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get DataImport: %v", err) + } + + ttl, _, _ := unstructured.NestedString(got.Object, "spec", "ttl") + if ttl != "1h" { + t.Errorf("spec.ttl = %q, want 1h", ttl) + } + + publish, _, _ := unstructured.NestedBool(got.Object, "spec", "publish") + if !publish { + t.Error("spec.publish = false, want true") + } +} + +// TestEnsureDataImport_PublishDowngradeInOneUpdate mirrors +// TestEnsureDataImport_AlignsBothTTLAndPublishInOneUpdate for the true->false direction: +// publish is aligned bidirectionally (unlike internal/data's upgrade-only semantics), so a +// prior --publish=true run must not keep exposing the upload publicly once a later run asks +// for the in-cluster path. +func TestEnsureDataImport_PublishDowngradeInOneUpdate(t *testing.T) { + leaf := volumeSnapshotLeaf("pvc-1") + + existing := dataImportObj(targetNS, "pvc-1", false) + _ = unstructured.SetNestedField(existing.Object, true, "spec", "publish") + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) // ttl: "1h", publish defaults false + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport: %v", err) + } + + if u := countDataImportActions(dyn, "update"); u != 1 { + t.Fatalf("update calls = %d, want exactly 1", u) + } + + diName := imp.DataImportName(leaf) + + got, err := dyn.Resource(dataImportGVR).Namespace(targetNS).Get(context.Background(), diName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get DataImport: %v", err) + } + + publish, _, _ := unstructured.NestedBool(got.Object, "spec", "publish") + if publish { + t.Error("spec.publish = true, want false (downgraded from a prior --publish=true run)") + } +} + +// TestEnsureDataImport_PublishAndTTLAlignedIssuesNoUpdate is a regression anchor for the +// ordering requirement in alignDataImportSpec: both the ttl- and publish-equality checks must +// run after any re-Get, but must still short-circuit to zero Updates when both are already +// aligned -- not just ttl alone, as covered by TestEnsureDataImport_TTLAlreadyAlignedIssuesNoUpdate. +func TestEnsureDataImport_PublishAndTTLAlignedIssuesNoUpdate(t *testing.T) { + leaf := volumeSnapshotLeaf("pvc-1") + + existing := dataImportObj(targetNS, "pvc-1", false) + _ = unstructured.SetNestedField(existing.Object, true, "spec", "publish") + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) // ttl: "1h" + imp.publish = true + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport: %v", err) + } + + if u := countDataImportActions(dyn, "update"); u != 0 { + t.Errorf("update calls = %d, want 0 (ttl and publish already aligned)", u) + } +} + +// TestEnsureDataImport_PublishChangeNeverForeign verifies the deliberate exclusion of +// spec.publish from dataImportAnnotations/validateDataImportSpec: reusing a DataImport whose +// spec.publish differs from the current run's --publish must NOT be treated as +// ErrForeignDataImport, since publish is a transport property of THIS run, not part of the +// leaf's content identity. +func TestEnsureDataImport_PublishChangeNeverForeign(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + existingPublish bool + runPublish bool + }{ + {name: "success: existing false, run requests true", existingPublish: false, runPublish: true}, + {name: "success: existing true, run requests false", existingPublish: true, runPublish: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + leaf := volumeSnapshotLeaf("pvc-1") + existing := dataImportObj(targetNS, "pvc-1", false) + _ = unstructured.SetNestedField(existing.Object, tc.existingPublish, "spec", "publish") + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) + imp.publish = tc.runPublish + + if _, err := imp.EnsureDataImport(context.Background(), leaf, targetNS); err != nil { + t.Fatalf("EnsureDataImport must not fail with ErrForeignDataImport on a publish-only change: %v", err) + } + }) + } +} + func TestEnsureDataImport_RecreatesExpired(t *testing.T) { leaf := volumeSnapshotLeaf("pvc-1") From 9f900c5aebc8672af5a10b164e723f888ff50421 Mon Sep 17 00:00:00 2001 From: Konstantin Kozoriz Date: Tue, 25 Aug 2026 15:07:49 +0300 Subject: [PATCH 3/4] feat(snapshot): add --publish flag to d8 snapshot upload Wire the flag through to the DataImport importer, auto-detecting the upload mode when unset, and document the bearer-token requirement for the published path. Signed-off-by: Konstantin Kozoriz --- internal/snapshot/cmd/snapimport/import.go | 48 ++++++- .../snapshot/cmd/snapimport/import_test.go | 117 ++++++++++++++++++ 2 files changed, 162 insertions(+), 3 deletions(-) diff --git a/internal/snapshot/cmd/snapimport/import.go b/internal/snapshot/cmd/snapimport/import.go index b49700910..60fc2a995 100644 --- a/internal/snapshot/cmd/snapimport/import.go +++ b/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,7 +136,18 @@ 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 @@ -141,7 +155,10 @@ Scope and limitations: 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, diff --git a/internal/snapshot/cmd/snapimport/import_test.go b/internal/snapshot/cmd/snapimport/import_test.go index af5d2aaa2..b596b0833 100644 --- a/internal/snapshot/cmd/snapimport/import_test.go +++ b/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) + } + }) + } +} From cf7d998f76fa5a347ea10cebbaa088fcfff9f013 Mon Sep 17 00:00:00 2001 From: Konstantin Kozoriz Date: Tue, 25 Aug 2026 19:54:13 +0300 Subject: [PATCH 4/4] fix(snapshot): enforce TLS verification on publish upload path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SetTLSCAData merged a trust pool but never reset the caller's inherited insecure-skip-tls-verify/tls-server-name, so Go skipped certificate verification entirely on that path — any endpoint could receive the real Kubernetes bearer token. Force verification on unconditionally, in both the rest.Config and the cloned transport, since client-go may already have baked the insecure flag into a base transport before this wrapper runs. Signed-off-by: Konstantin Kozoriz --- internal/snapshot/snapimport/volume.go | 7 +- internal/snapshot/snapimport/volume_test.go | 53 +++++++++-- internal/snapshot/transport/http.go | 20 +++++ internal/snapshot/transport/http_test.go | 99 +++++++++++++++++++++ 4 files changed, 169 insertions(+), 10 deletions(-) diff --git a/internal/snapshot/snapimport/volume.go b/internal/snapshot/snapimport/volume.go index a5f36a223..6416be889 100644 --- a/internal/snapshot/snapimport/volume.go +++ b/internal/snapshot/snapimport/volume.go @@ -837,8 +837,11 @@ func (c *clusterVolumeImporter) uploadClient(caB64, rawURL string) (uploadHTTPCl // Through Ingress, TLS terminates at ingress-nginx's own certificate, which never chains // to the importer pod's internal CA — endpoint-specific pinning is impossible there, so - // publish=true trades it for a merged trust pool (below). Confined to that branch; - // insecure-skip-tls-verify inherited from kubeconfig is untouched either way. + // publish=true trades it for a merged trust pool (below). Confined to that branch. Both + // branches force server certificate verification on regardless of what the caller's + // kubeconfig set for insecure-skip-tls-verify/tls-server-name — only the trust model + // differs: publish EXTENDS trust (system roots + kubeconfig CA + status.ca merged into one + // pool), non-publish REPLACES it (pinned exclusively to the importer pod's internal CA). if c.publish { if err := transport.ValidateHTTPSURL(rawURL); err != nil { return nil, fmt.Errorf("validate DataImport publish upload URL: %w", err) diff --git a/internal/snapshot/snapimport/volume_test.go b/internal/snapshot/snapimport/volume_test.go index 406fa0fa8..a4d4694c7 100644 --- a/internal/snapshot/snapimport/volume_test.go +++ b/internal/snapshot/snapimport/volume_test.go @@ -5172,9 +5172,10 @@ func TestUploadClient_PinningDependsOnPublish(t *testing.T) { unrelatedCA := unrelatedCertificatePEM(t) tests := []struct { - name string - publish bool - wantErr bool + name string + publish bool + insecure bool + wantErr bool }{ { name: "success: publish merges the unrelated CA with the kubeconfig-trusted pool", @@ -5185,17 +5186,49 @@ func TestUploadClient_PinningDependsOnPublish(t *testing.T) { publish: false, wantErr: true, }, + { + // Regression guard for the insecure-skip-tls-verify bypass: SetTLSCAData must force + // verification on even though the caller's kubeconfig inherited Insecure: true, so a + // server whose certificate is in NEITHER the system pool NOR the explicitly supplied + // CA must still be rejected -- and, crucially, the bearer token must never reach that + // untrusted server, since a failed handshake means the request body (headers + // included) was never sent. + name: "error: publish with inherited insecure-skip-tls-verify still enforces verification and never leaks the bearer token", + publish: true, + insecure: true, + wantErr: true, + }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { t.Parallel() + targetURL := srv.URL + configCAData := trustedCA + + var receivedAuthHeader string + + if tc.insecure { + // A dedicated server (rather than the shared srv above) guarantees its + // certificate is untrusted by construction, and lets this subtest read + // receivedAuthHeader without racing the other subtests' concurrent requests + // against the shared srv. + untrustedSrv := httptest.NewTLSServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + receivedAuthHeader = r.Header.Get("Authorization") + })) + t.Cleanup(untrustedSrv.Close) + + targetURL = untrustedSrv.URL + configCAData = nil + } + config := &restclient.Config{ - Host: srv.URL, + Host: targetURL, BearerToken: "must-not-leak", TLSClientConfig: restclient.TLSClientConfig{ - CAData: trustedCA, + Insecure: tc.insecure, + CAData: configCAData, }, } @@ -5204,13 +5237,13 @@ func TestUploadClient_PinningDependsOnPublish(t *testing.T) { publish: tc.publish, } - httpClient, err := importer.uploadClient(base64.StdEncoding.EncodeToString(unrelatedCA), srv.URL) + httpClient, err := importer.uploadClient(base64.StdEncoding.EncodeToString(unrelatedCA), targetURL) if err != nil { t.Fatalf("uploadClient: %v", err) } t.Cleanup(httpClient.CloseIdleConnections) - req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL, nil) + req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, targetURL, nil) if err != nil { t.Fatalf("build request: %v", err) } @@ -5222,7 +5255,11 @@ func TestUploadClient_PinningDependsOnPublish(t *testing.T) { if tc.wantErr { if doErr == nil { - t.Fatal("request unexpectedly succeeded against the real origin under strict (non-publish) pinning to an unrelated CA") + t.Fatal("request unexpectedly succeeded against an untrusted origin") + } + + if tc.insecure && receivedAuthHeader != "" { + t.Fatalf("Authorization header leaked to an untrusted server: %q", receivedAuthHeader) } return diff --git a/internal/snapshot/transport/http.go b/internal/snapshot/transport/http.go index 15793e15d..97de8e621 100644 --- a/internal/snapshot/transport/http.go +++ b/internal/snapshot/transport/http.go @@ -1179,6 +1179,22 @@ func (c *Client) NewRTClient(schemeFuncs ...func(s *apiruntime.Scheme) error) (c return NewRuntimeClient(c.restConfig, schemeFuncs...) } +// SetTLSCAData extends inherited server trust with a merged pool: system roots, +// the supplied caData, and any CA already configured on this Client's rest.Config +// (e.g. from kubeconfig). Unlike SetTLSIdentityCAData, which REPLACES trust with +// exactly one endpoint-specific CA, this widens it — appropriate only where +// endpoint-specific pinning is impossible (see the publish path in +// internal/snapshot/snapimport/volume.go). +// +// Because the merged pool only adds trust, it must never be reachable through a +// bypass: an inherited insecure-skip-tls-verify or tls-server-name from kubeconfig +// would make certificate verification a no-op regardless of how large RootCAs is, +// so both are forced off here — on the rest.Config itself and, since client-go may +// have already built a base *http.Transport with those values baked in before this +// WrapTransport runs, again on the transport clone that carries RootCAs. This must +// run unconditionally, not only when caData is non-empty: an empty caData is the +// normal case on the publish path (the ingress does not expose the importer pod's +// internal CA), and verification must stay on even then. func (c *Client) SetTLSCAData(caData []byte) { sysPool, err := x509.SystemCertPool() if err != nil || sysPool == nil { @@ -1195,6 +1211,8 @@ func (c *Client) SetTLSCAData(caData []byte) { c.restConfig.TLSClientConfig.CAData = nil c.restConfig.TLSClientConfig.CAFile = "" + c.restConfig.TLSClientConfig.Insecure = false + c.restConfig.TLSClientConfig.ServerName = "" prev := c.restConfig.WrapTransport c.restConfig.WrapTransport = func(rt http.RoundTripper) http.RoundTripper { @@ -1216,6 +1234,8 @@ func (c *Client) SetTLSCAData(caData []byte) { } clonedTransport.TLSClientConfig.RootCAs = sysPool + clonedTransport.TLSClientConfig.InsecureSkipVerify = false + clonedTransport.TLSClientConfig.ServerName = "" return clonedTransport } diff --git a/internal/snapshot/transport/http_test.go b/internal/snapshot/transport/http_test.go index 31bad993e..4f894bbf2 100644 --- a/internal/snapshot/transport/http_test.go +++ b/internal/snapshot/transport/http_test.go @@ -313,6 +313,93 @@ func TestClient_SetTLSCAData_PassThroughNonTransport(t *testing.T) { } } +// TestClient_SetTLSCAData_ForcesVerification is the regression guard for the +// insecure-skip-tls-verify/tls-server-name bypass: SetTLSCAData builds a merged +// trust pool, but an inherited Insecure or ServerName from kubeconfig can make +// verification a no-op (InsecureSkipVerify) or check the wrong hostname +// regardless of how large that pool is. SetTLSCAData must force both off — on +// the rest.Config itself AND on the transport clone that carries RootCAs, since +// client-go may already have baked the inherited values into a base +// *http.Transport before WrapTransport runs. +func TestClient_SetTLSCAData_ForcesVerification(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + caData []byte + }{ + { + name: "nil CA data", + caData: nil, + }, + { + name: "valid CA data", + caData: testCACertificatePEM(t), + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + sc := &Client{ + restConfig: &rest.Config{ + TLSClientConfig: rest.TLSClientConfig{ + Insecure: true, + ServerName: "api.example", + }, + }, + } + + sc.SetTLSCAData(tc.caData) + + if sc.restConfig.TLSClientConfig.Insecure { + t.Error("restConfig.TLSClientConfig.Insecure = true, want false after SetTLSCAData") + } + + if sc.restConfig.TLSClientConfig.ServerName != "" { + t.Errorf("restConfig.TLSClientConfig.ServerName = %q, want empty after SetTLSCAData", + sc.restConfig.TLSClientConfig.ServerName) + } + + orig := &http.Transport{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, // exercising the inherited-insecure bypass this test guards against + ServerName: "api.example", + }, + } + + got := sc.restConfig.WrapTransport(orig) + + wrapped, ok := got.(*http.Transport) + if !ok { + t.Fatalf("wrapped transport is not an *http.Transport: %T", got) + } + + if wrapped.TLSClientConfig.InsecureSkipVerify { + t.Error("cloned transport TLSClientConfig.InsecureSkipVerify = true, want false") + } + + if wrapped.TLSClientConfig.ServerName != "" { + t.Errorf("cloned transport TLSClientConfig.ServerName = %q, want empty", wrapped.TLSClientConfig.ServerName) + } + + if wrapped.TLSClientConfig.RootCAs == nil { + t.Error("cloned transport TLSClientConfig.RootCAs is nil") + } + + if !orig.TLSClientConfig.InsecureSkipVerify { + t.Error("original transport TLSClientConfig.InsecureSkipVerify was mutated, want it untouched") + } + + if orig.TLSClientConfig.ServerName != "api.example" { + t.Errorf("original transport TLSClientConfig.ServerName = %q, want it untouched (\"api.example\")", + orig.TLSClientConfig.ServerName) + } + }) + } +} + // TestClient_SetResponseHeaderTimeout_FailsFastOnHeaderStall asserts the // configured transport aborts a request whose server accepts the connection but // never sends response headers, within the response-header timeout. @@ -1306,6 +1393,18 @@ func newPersistentTLSServer( return server } +// testCACertificatePEM returns a PEM-encoded certificate suitable as a valid, +// well-formed CA bundle input, without requiring the caller to stand up and +// tear down a dedicated TLS server just to obtain one. +func testCACertificatePEM(t *testing.T) []byte { + t.Helper() + + srv := httptest.NewTLSServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + t.Cleanup(srv.Close) + + return certificatePEM(t, srv) +} + func certificatePEM(t *testing.T, server *httptest.Server) []byte { t.Helper()