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) + } + }) + } +} 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..6416be889 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,25 @@ 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. 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) + } + } + + // 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 +867,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 +898,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 +928,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 +936,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 +2254,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 +2310,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 +2386,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..a4d4694c7 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,248 @@ 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 + insecure 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, + }, + { + // 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: targetURL, + BearerToken: "must-not-leak", + TLSClientConfig: restclient.TLSClientConfig{ + Insecure: tc.insecure, + CAData: configCAData, + }, + } + + importer := &clusterVolumeImporter{ + sc: transport.NewClientForConfig(config), + publish: tc.publish, + } + + 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, targetURL, 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 an untrusted origin") + } + + if tc.insecure && receivedAuthHeader != "" { + t.Fatalf("Authorization header leaked to an untrusted server: %q", receivedAuthHeader) + } + + 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 +5620,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 +5642,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 +5668,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 +5723,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 +5779,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 +5796,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") diff --git a/internal/snapshot/transport/http.go b/internal/snapshot/transport/http.go index cd5ff0710..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,11 +1234,21 @@ func (c *Client) SetTLSCAData(caData []byte) { } clonedTransport.TLSClientConfig.RootCAs = sysPool + clonedTransport.TLSClientConfig.InsecureSkipVerify = false + clonedTransport.TLSClientConfig.ServerName = "" return clonedTransport } } +// 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..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. @@ -1212,6 +1299,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, @@ -1248,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() 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) + }) +}