feat(satellite): run the snapshot stream over mutual TLS

cmd/satellite/flags.go

@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: Apache-2.0
+
+/*
+Copyright 2026 Cozystack contributors.
+
+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 main
+
+import (
+	"flag"
+	"os"
+
+	"github.com/cockroachdb/errors"
+
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
+)
+
+// options are the satellite's command-line flags.
+type options struct {
+	nodeName            string
+	stateDir            string
+	probeAddr           string
+	controllerNamespace string
+	streamTLS           stream.TLSFiles
+}
+
+// parseFlags registers the satellite's flags on flagSet, parses args and
+// refuses a stream TLS configuration the satellite must not start with:
+// some of the three files set and not all would quietly turn the stream
+// off.
+func parseFlags(flagSet *flag.FlagSet, args []string) (options, error) {
+	var opts options
+
+	flagSet.StringVar(&opts.nodeName, "node-name", os.Getenv("NODE_NAME"),
+		"name this satellite registers under (defaults to NODE_NAME env)")
+	flagSet.StringVar(&opts.stateDir, "state-dir", "/var/lib/blockstor-satellite",
+		"directory the satellite uses to persist DRBD .res files and per-resource state")
+	flagSet.StringVar(&opts.probeAddr, "health-probe-bind-address", ":8081",
+		"The address the /healthz and /readyz probe endpoints bind to.")
+	flagSet.StringVar(&opts.controllerNamespace, "controller-namespace", "",
+		"Namespace where the controller's Secrets (cluster master passphrase) live (default: $POD_NAMESPACE, then 'blockstor-system').")
+
+	flagSet.StringVar(&opts.streamTLS.CertFile, "stream-tls-cert-file", "",
+		"PEM certificate this satellite presents on the snapshot stream, for both server and client auth (cert-manager tls.crt).")
+	flagSet.StringVar(&opts.streamTLS.KeyFile, "stream-tls-key-file", "",
+		"PEM key for --stream-tls-cert-file (cert-manager tls.key).")
+	flagSet.StringVar(&opts.streamTLS.CAFile, "stream-tls-ca-file", "",
+		"PEM CA bundle every peer satellite's stream certificate is verified against (cert-manager ca.crt). "+
+			"With none of the three set the snapshot stream is off and cross-node clone and restore fall back to a DRBD resync.")
+
+	err := flagSet.Parse(args)
+	if err != nil {
+		return opts, errors.Wrap(err, "parse flags")
+	}
+
+	err = opts.streamTLS.Validate()
+	if err != nil {
+		return opts, errors.Wrap(err, "snapshot stream TLS flags")
+	}
+
+	return opts, nil
+}

cmd/satellite/flags_test.go

@@ -0,0 +1,66 @@
+// SPDX-License-Identifier: Apache-2.0
+
+/*
+Copyright 2026 Cozystack contributors.
+
+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 main
+
+import (
+	"errors"
+	"flag"
+	"io"
+	"testing"
+
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
+)
+
+func parseTestFlags(args ...string) (options, error) {
+	fs := flag.NewFlagSet("satellite", flag.ContinueOnError)
+	fs.SetOutput(io.Discard)
+
+	return parseFlags(fs, args)
+}
+
+// A partial set of stream TLS flags refuses to start. Let through, it would
+// quietly turn the stream off and every cross-node restore on the node would
+// fall back to a resync.
+func TestSatelliteRefusesAPartialStreamTLSConfiguration(t *testing.T) {
+	t.Parallel()
+
+	for _, args := range [][]string{
+		{"--stream-tls-cert-file=/tls/tls.crt"},
+		{"--stream-tls-cert-file=/tls/tls.crt", "--stream-tls-key-file=/tls/tls.key"},
+		{"--stream-tls-ca-file=/tls/ca.crt"},
+	} {
+		_, err := parseTestFlags(args...)
+		if !errors.Is(err, stream.ErrTLSIncomplete) {
+			t.Errorf("flags %v = %v, want stream.ErrTLSIncomplete", args, err)
+		}
+	}
+}
+
+func TestSatelliteStartsWithTheStreamOff(t *testing.T) {
+	t.Parallel()
+
+	opts, err := parseTestFlags("--node-name=n1")
+	if err != nil {
+		t.Fatalf("no stream TLS flags = %v, want the stream off and no error", err)
+	}
+
+	if opts.nodeName != "n1" || opts.streamTLS.Configured() {
+		t.Fatalf("options = %+v, want node n1 and the stream off", opts)
+	}
+}

cmd/satellite/main.go

@@ -45,6 +45,7 @@ import (
 
 	"github.com/cozystack/blockstor/pkg/satellite"
 	"github.com/cozystack/blockstor/pkg/satellite/controllers"
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
 	"github.com/cozystack/blockstor/pkg/storage"
 	"github.com/cozystack/blockstor/pkg/uevent"
 )
@@ -57,25 +58,13 @@ func main() {
 // only ever calls os.Exit(run()) so there are no defers in the same frame
 // as the os.Exit call.
 func run() int {
-	var (
-		nodeName            string
-		stateDir            string
-		probeAddr           string
-		controllerNamespace string
-	)
-
-	flag.StringVar(&nodeName, "node-name", os.Getenv("NODE_NAME"),
-		"name this satellite registers under (defaults to NODE_NAME env)")
-	flag.StringVar(&stateDir, "state-dir", "/var/lib/blockstor-satellite",
-		"directory the satellite uses to persist DRBD .res files and per-resource state")
-	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081",
-		"The address the /healthz and /readyz probe endpoints bind to.")
-	flag.StringVar(&controllerNamespace, "controller-namespace", "",
-		"Namespace where the controller's Secrets (cluster master passphrase) live (default: $POD_NAMESPACE, then 'blockstor-system').")
-
-	flag.Parse()
-
-	controllerNamespace = resolveControllerNamespace(controllerNamespace)
+	opts, err := parseFlags(flag.CommandLine, os.Args[1:])
+
+	nodeName := opts.nodeName
+	stateDir := opts.stateDir
+	probeAddr := opts.probeAddr
+	controllerNamespace := resolveControllerNamespace(opts.controllerNamespace)
+	streamTLS := opts.streamTLS
 
 	logger := slog.New(slog.NewJSONHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
 	slog.SetDefault(logger)
@@ -88,6 +77,12 @@ func run() int {
 	// disappear).
 	ctrl.SetLogger(logr.FromSlogHandler(logger.Handler()))
 
+	if err != nil {
+		logger.Error("invalid flags", "err", err)
+
+		return 1
+	}
+
 	if nodeName == "" {
 		logger.Error("node-name is required (pass --node-name or set NODE_NAME)")
 
@@ -210,8 +205,9 @@ func run() int {
 		LocalAddress:           localAddress,
 		Logger:                 logger,
 		RESTConfig:             restCfg,
-		ManagerFactory:         mgrFactory(ready, logger, ueventListener, controllerNamespace),
+		ManagerFactory:         mgrFactory(ready, logger, ueventListener, controllerNamespace, streamTLS),
 		HealthProbeBindAddress: probeAddr,
+		StreamTLS:              streamTLS,
 		OnCacheSynced: func() {
 			logger.Info("satellite cache sync complete, marking ready")
 			ready.MarkReady()
@@ -257,7 +253,10 @@ func loadRESTConfig() (*rest.Config, error) {
 // The factory shape lets the agent stay ignorant of the readyState
 // + logger plumbing — it only sees the standard
 // satellite.ManagerFactory signature.
-func mgrFactory(ready *readyState, logger *slog.Logger, ueventListener controllers.UeventNotifier, controllerNamespace string) satellite.ManagerFactory {
+func mgrFactory(
+	ready *readyState, logger *slog.Logger, ueventListener controllers.UeventNotifier,
+	controllerNamespace string, streamTLS stream.TLSFiles,
+) satellite.ManagerFactory {
 	return func(restCfg *rest.Config, nodeName, probeAddr string, rec *satellite.Reconciler) (manager.Manager, error) {
 		mgr, err := controllers.NewManager(restCfg, &controllers.Config{
 			NodeName:               nodeName,
@@ -273,6 +272,7 @@ func mgrFactory(ready *readyState, logger *slog.Logger, ueventListener controlle
 			// PhysicalDeviceDiscoveryRunnable falls back to pure-
 			// polling discovery.
 			UeventListener: ueventListener,
+			StreamTLS:      streamTLS,
 		})
 		if err != nil {
 			return nil, err //nolint:wrapcheck // controllers.NewManager already wraps

pkg/satellite/agent.go

@@ -98,6 +98,11 @@ type Config struct {
 	// port-binding races between parallel runs.
 	HealthProbeBindAddress string
 
+	// StreamTLS is the mutual TLS of the satellite-to-satellite snapshot
+	// stream. Not configured, the stream server does not start and fetches
+	// answer stream.ErrStreamDisabled.
+	StreamTLS stream.TLSFiles
+
 	// OnCacheSynced fires once the controller-runtime manager's
 	// caches have completed their initial sync. cmd/satellite uses
 	// it to flip /readyz from 503 to 200 so kubelet only routes
@@ -206,19 +211,26 @@ func (a poolResolverAdapter) ProviderForResource(name string) (storage.Provider,
 }
 
 // startStreamServer launches the satellite-to-satellite snapshot
-// stream HTTP server in a goroutine. Bound to 0.0.0.0:stream.Port
+// stream HTTPS server in a goroutine, with mutual TLS. Bound to 0.0.0.0:stream.Port
 // — the DaemonSet runs on hostNetwork so this is the node's IP.
 //
-// A bind failure logs but does not abort the agent — without the
-// stream server, cross-node snapshot-restore on this satellite
-// falls back to "peer has no snapshot here" and the receiving
-// satellite tries other peers. Same-node clone keeps working.
+// A bind failure logs but does not abort the agent. A peer fetching from
+// this satellite then gets a connection error, tries the other nodes that
+// hold the snapshot, and retries the restore when none answers.
+// Same-node clone keeps working.
 func (a *Agent) startStreamServer(ctx context.Context, rec *Reconciler) {
+	if !a.cfg.StreamTLS.Configured() {
+		a.logger.Info("snapshot stream disabled: no stream TLS configured; " +
+			"cross-node clone and restore fall back to a DRBD resync")
+
+		return
+	}
+
 	addr := fmt.Sprintf("0.0.0.0:%d", stream.Port)
 	srv := stream.NewServer(poolResolverAdapter{rec: rec})
 
 	go func() {
-		err := stream.ListenAndServe(ctx, addr, srv)
+		err := stream.ListenAndServe(ctx, addr, srv, a.cfg.StreamTLS)
 		if err != nil && !errors.Is(err, context.Canceled) {
 			a.logger.Error("snapshot stream server exited", "addr", addr, "err", err)
 		}

pkg/satellite/controllers/config.go

@@ -23,6 +23,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/event"
 
 	"github.com/cozystack/blockstor/pkg/satellite"
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
 	"github.com/cozystack/blockstor/pkg/storage"
 )
 
@@ -106,4 +107,9 @@ type Config struct {
 	// pin the burst-debounce / reacts-within-deadline behaviour
 	// without opening a real netlink socket.
 	UeventListener UeventNotifier
+
+	// StreamTLS is the mutual TLS the cross-node snapshot fetcher dials
+	// peer satellites with. Not configured, nothing is fetched and a
+	// restore on this node falls back to a DRBD resync.
+	StreamTLS stream.TLSFiles
 }

pkg/satellite/controllers/manager.go

@@ -33,6 +33,7 @@ import (
 
 	blockstoriov1alpha1 "github.com/cozystack/blockstor/api/v1alpha1"
 	"github.com/cozystack/blockstor/pkg/drbd"
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
 )
 
 // reconcileTriggerBuffer bounds the observer → reconciler trigger
@@ -313,6 +314,7 @@ func wireCrossNodeFetcher(mgr manager.Manager, cfg *Config) {
 	cfg.Apply.SetCrossNodeFetcher(&SnapshotFetcher{
 		Client:   mgr.GetClient(),
 		NodeName: cfg.NodeName,
+		Stream:   stream.NewClient(cfg.StreamTLS),
 	})
 }
 

pkg/satellite/controllers/snapshot_fetcher.go

@@ -51,8 +51,8 @@ type SnapshotFetcher struct {
 	NodeName string
 
 	// Stream is the HTTP client that talks to peer satellites'
-	// snapshot-stream endpoints. Nil → NewClient(nil) is used at
-	// first Fetch call.
+	// snapshot-stream endpoints over mutual TLS. Nil fetches nothing
+	// (stream.ErrStreamDisabled), and the caller falls back to a resync.
 	Stream *stream.Client
 }
 
@@ -68,7 +68,7 @@ type SnapshotFetcher struct {
 //     through to a blank CreateVolume + DRBD resync.
 func (f *SnapshotFetcher) Fetch(ctx context.Context, srcRD, snapName string, vol int32) (io.ReadCloser, string, error) {
 	if f.Stream == nil {
-		f.Stream = stream.NewClient(nil)
+		f.Stream = stream.NewClient(stream.TLSFiles{})
 	}
 
 	var snap blockstoriov1alpha1.Snapshot

pkg/satellite/controllers/snapshot_fetcher_test.go

@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: Apache-2.0
+
+/*
+Copyright 2026 Cozystack contributors.
+
+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 controllers_test
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	blockstoriov1alpha1 "github.com/cozystack/blockstor/api/v1alpha1"
+	"github.com/cozystack/blockstor/pkg/satellite/controllers"
+	"github.com/cozystack/blockstor/pkg/satellite/stream"
+	"github.com/cozystack/blockstor/pkg/storage"
+	"github.com/cozystack/blockstor/pkg/store/k8s"
+)
+
+// A satellite with no stream TLS configured cannot fetch from a peer. The
+// fetcher has to answer storage.ErrNotFound then, so the restore falls back
+// to a blank volume and a DRBD resync. Any other error requeues the
+// restore, and with the stream off it never succeeds.
+func TestSnapshotFetcherWithTheStreamOffFallsBackToAResync(t *testing.T) {
+	t.Parallel()
+
+	snap := &blockstoriov1alpha1.Snapshot{
+		ObjectMeta: metav1.ObjectMeta{Name: k8s.Name("rd.snap")},
+		Spec: blockstoriov1alpha1.SnapshotSpec{
+			ResourceDefinitionName: "rd",
+			SnapshotName:           "snap",
+			Nodes:                  []string{"self", "peer"},
+		},
+	}
+	peer := &blockstoriov1alpha1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: "peer"},
+		Spec: blockstoriov1alpha1.NodeSpec{
+			NetInterfaces: []blockstoriov1alpha1.NodeNetInterface{{Name: "default", Address: "127.0.0.1"}},
+		},
+	}
+
+	fetcher := &controllers.SnapshotFetcher{
+		Client:   fake.NewClientBuilder().WithScheme(newSnapshotScheme(t)).WithObjects(snap, peer).Build(),
+		NodeName: "self",
+		Stream:   stream.NewClient(stream.TLSFiles{}),
+	}
+
+	body, _, err := fetcher.Fetch(t.Context(), "rd", "snap", 0)
+	if body != nil {
+		_ = body.Close()
+	}
+
+	if !errors.Is(err, storage.ErrNotFound) {
+		t.Fatalf("fetch with the stream off = %v, want storage.ErrNotFound", err)
+	}
+}