diff --git a/cmd/satellite/flags.go b/cmd/satellite/flags.go new file mode 100644 index 00000000..2c974c30 --- /dev/null +++ b/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 +} diff --git a/cmd/satellite/flags_test.go b/cmd/satellite/flags_test.go new file mode 100644 index 00000000..ca6d55dc --- /dev/null +++ b/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) + } +} diff --git a/cmd/satellite/main.go b/cmd/satellite/main.go index 396d74bf..e96d6b28 100644 --- a/cmd/satellite/main.go +++ b/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 diff --git a/pkg/satellite/agent.go b/pkg/satellite/agent.go index 86c9447e..7d43c55e 100644 --- a/pkg/satellite/agent.go +++ b/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) } diff --git a/pkg/satellite/controllers/config.go b/pkg/satellite/controllers/config.go index 6ff27516..6c6fb49d 100644 --- a/pkg/satellite/controllers/config.go +++ b/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 } diff --git a/pkg/satellite/controllers/manager.go b/pkg/satellite/controllers/manager.go index 34b8ee41..1715b406 100644 --- a/pkg/satellite/controllers/manager.go +++ b/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), }) } diff --git a/pkg/satellite/controllers/snapshot_fetcher.go b/pkg/satellite/controllers/snapshot_fetcher.go index 204e7221..c5969a99 100644 --- a/pkg/satellite/controllers/snapshot_fetcher.go +++ b/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 diff --git a/pkg/satellite/controllers/snapshot_fetcher_test.go b/pkg/satellite/controllers/snapshot_fetcher_test.go new file mode 100644 index 00000000..2be164ae --- /dev/null +++ b/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) + } +} diff --git a/pkg/satellite/stream/client.go b/pkg/satellite/stream/client.go index 459b36e8..7fdea540 100644 --- a/pkg/satellite/stream/client.go +++ b/pkg/satellite/stream/client.go @@ -22,30 +22,39 @@ import ( "context" "fmt" "io" + "net" "net/http" "net/url" "strconv" + "time" "github.com/cockroachdb/errors" "github.com/cozystack/blockstor/pkg/storage" ) -// Client fetches a snapshot stream from a peer satellite. One Client -// instance is reused across requests — it just wraps an http.Client. +// ErrStreamDisabled is a fetch from a satellite whose stream TLS is not +// configured. It wraps storage.ErrNotFound, so the caller falls back to a +// blank volume and a DRBD resync, as it does when no peer has the snapshot. +var ErrStreamDisabled = errors.Wrap(storage.ErrNotFound, "the snapshot stream is disabled: no stream TLS configured") + +// The dial and handshake bounds http.DefaultTransport uses. The transfer +// itself is not bounded: a volume can take minutes. +const ( + dialTimeout = 30 * time.Second + tlsHandshakeTimeout = 10 * time.Second +) + +// Client fetches a snapshot stream from a peer satellite over mutual TLS. type Client struct { - http *http.Client + files TLSFiles } -// NewClient constructs a Client. Passing nil for http defaults to -// http.DefaultClient with no timeout — fine for the cross-node clone -// path where snapshot transfers can take minutes. -func NewClient(c *http.Client) *Client { - if c == nil { - c = &http.Client{} - } - - return &Client{http: c} +// NewClient builds a Client that presents and verifies certificates from +// files. With files not configured every Fetch answers ErrStreamDisabled: the +// stream carries raw volume data, and it is not sent in the clear. +func NewClient(files TLSFiles) *Client { + return &Client{files: files} } // Fetch opens a snapshot stream from peerAddr (host:port form) for @@ -60,8 +69,29 @@ func NewClient(c *http.Client) *Client { // snapshot doesn't live on that satellite — caller should try // another peer from snap.Nodes). func (c *Client) Fetch(ctx context.Context, peerAddr, rd, snap string, vol int32) (io.ReadCloser, error) { + if c == nil || !c.files.Configured() { + return nil, ErrStreamDisabled + } + + tlsConfig, err := c.files.clientConfig() + if err != nil { + return nil, errors.Wrap(err, "snapshot-fetch") + } + + // A transport per fetch, so the certificate and CA are the ones on + // disk now. Fetches are rare and each moves a whole volume. Without + // keep-alives the connection closes with the body; a fresh transport + // that pooled it would keep it, and its goroutines, open on both + // satellites for good. + httpClient := &http.Client{Transport: &http.Transport{ + TLSClientConfig: tlsConfig, + DisableKeepAlives: true, + DialContext: (&net.Dialer{Timeout: dialTimeout}).DialContext, + TLSHandshakeTimeout: tlsHandshakeTimeout, + }} + endpoint := url.URL{ - Scheme: "http", + Scheme: "https", Host: peerAddr, Path: PathPrefix + "/" + url.PathEscape(rd) + "/" + url.PathEscape(snap) + "/" + strconv.Itoa(int(vol)), } @@ -71,16 +101,16 @@ func (c *Client) Fetch(ctx context.Context, peerAddr, rd, snap string, vol int32 return nil, errors.Wrap(err, "build snapshot-fetch request") } - resp, err := c.http.Do(req) + resp, err := httpClient.Do(req) if err != nil { return nil, errors.Wrapf(err, "GET %s", endpoint.String()) } if resp.StatusCode == http.StatusNotFound { - // Body is small ("storage object not found\n"); drain to - // release the connection back to the pool, then surface as - // ErrNotFound so the caller can fall through to the next - // peer (or to DRBD resync) without parsing the message. + // Body is small ("storage object not found\n"); drain and + // close it, then surface as ErrNotFound so the caller can fall + // through to the next peer (or to DRBD resync) without parsing + // the message. _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() diff --git a/pkg/satellite/stream/redial_spec_test.go b/pkg/satellite/stream/redial_spec_test.go index 5b9d7429..85077396 100644 --- a/pkg/satellite/stream/redial_spec_test.go +++ b/pkg/satellite/stream/redial_spec_test.go @@ -29,8 +29,8 @@ package stream_test // Pinned current behaviour (DEFERRED — Outcome B): // // 1. blockstor's satellite has NO `--controller-bind-address` flag. -// The only flags on cmd/satellite/main.go are `--node-name` and -// `--state-dir`. Verified in TestSatelliteFlagsLackControllerBindAddress +// None of the flags in cmd/satellite/flags.go carries a controller +// address. Verified in TestSatelliteFlagsLackControllerBindAddress // below. // // 2. Phase 10.6 retired every satellite→controller gRPC wire — the @@ -108,25 +108,25 @@ import ( func TestSatelliteFlagsLackControllerBindAddress(t *testing.T) { t.Parallel() - mainPath := satelliteMainPath(t) + flagsPath := satelliteFlagsPath(t) fset := token.NewFileSet() - f, err := parser.ParseFile(fset, mainPath, nil, parser.SkipObjectResolution) + f, err := parser.ParseFile(fset, flagsPath, nil, parser.SkipObjectResolution) if err != nil { - t.Fatalf("parse %s: %v", mainPath, err) + t.Fatalf("parse %s: %v", flagsPath, err) } flags := collectFlagNames(f) - // Sanity: we actually parsed flags out of main.go. If main.go - // stops using flag.StringVar / flag.BoolVar entirely (e.g. + // Sanity: we actually parsed flags out of flags.go. If flags.go + // stops using flagSet.StringVar / flagSet.BoolVar entirely (e.g. // migrated to cobra) this slice will be empty and the test // becomes meaningless — fail loudly so a follow-up audit picks // it up. if len(flags) == 0 { - t.Fatalf("no flag.*Var calls found in %s — has the binary been ported to a new flag library? "+ - "Update this spec test to read whichever new mechanism declares CLI flags.", mainPath) + t.Fatalf("no flagSet.*Var calls found in %s: has the binary been ported to a new flag library? "+ + "Update this spec test to read whichever new mechanism declares CLI flags.", flagsPath) } // The forbidden-by-current-spec flag names. Any of these @@ -151,14 +151,19 @@ func TestSatelliteFlagsLackControllerBindAddress(t *testing.T) { // Belt-and-braces: pin the exact flag set we expect today, so // any drift (added or removed) trips review. The order matches - // the order declared in cmd/satellite/main.go. + // the order declared in cmd/satellite/flags.go. // Bug 207 added health-probe-bind-address for kubelet livenessProbe. // Bug 023 added controller-namespace — it names the NAMESPACE the // cluster master passphrase Secret lives in (read via the // kube-apiserver, the satellite's one and only wire); it carries // no controller ADDRESS and does not re-open the Outcome A // controller-wire question this spec guards. - want := []string{"node-name", "state-dir", "health-probe-bind-address", "controller-namespace"} + // The stream-tls-* files are the mutual TLS of the snapshot stream + // between satellites; they carry no controller address either. + want := []string{ + "node-name", "state-dir", "health-probe-bind-address", "controller-namespace", + "stream-tls-cert-file", "stream-tls-key-file", "stream-tls-ca-file", + } if !equalStringSlices(flags, want) { t.Fatalf("satellite flag set drifted: got %v, want %v. "+ "If this is intentional, update the want list AND update tests/scenarios/03-networking.md §3.10 "+ @@ -207,11 +212,11 @@ func TestNodeCRDHasNoActiveField(t *testing.T) { } } -// satelliteMainPath resolves cmd/satellite/main.go relative to the -// running test binary. The test runs under +// satelliteFlagsPath resolves cmd/satellite/flags.go, where parseFlags +// declares every satellite flag, relative to the running test binary. The test runs under // pkg/satellite/stream/, so we walk two directories up then // into cmd/satellite/. -func satelliteMainPath(t *testing.T) string { +func satelliteFlagsPath(t *testing.T) string { t.Helper() // runtime.Caller returns the path of this test file inside the @@ -219,21 +224,21 @@ func satelliteMainPath(t *testing.T) string { // os.Getwd which would be the per-test temp dir. _, thisFile, _, ok := runtime.Caller(0) if !ok { - t.Fatal("runtime.Caller(0) failed — cannot locate cmd/satellite/main.go") + t.Fatal("runtime.Caller(0) failed: cannot locate cmd/satellite/flags.go") } // thisFile = .../pkg/satellite/stream/redial_spec_test.go // Walk up to the module root (4 levels: stream → satellite → pkg → root). root := filepath.Dir(filepath.Dir(filepath.Dir(filepath.Dir(thisFile)))) - return filepath.Join(root, "cmd", "satellite", "main.go") + return filepath.Join(root, "cmd", "satellite", "flags.go") } -// collectFlagNames walks the parsed AST of cmd/satellite/main.go and -// returns every flag name passed to flag.StringVar / flag.BoolVar / -// flag.IntVar / flag.Float64Var / flag.DurationVar in declaration -// order. Pure AST inspection — does NOT execute main(), which would -// register the flags into flag.CommandLine and pollute other tests. +// collectFlagNames walks the parsed AST of cmd/satellite/flags.go and +// returns every flag name passed to flagSet.StringVar / flagSet.BoolVar / +// flagSet.IntVar / flagSet.Float64Var / flagSet.DurationVar (or the same on the flag +// package) in declaration order. Pure AST inspection, it does NOT +// execute main(). func collectFlagNames(f *ast.File) []string { var names []string @@ -249,11 +254,11 @@ func collectFlagNames(f *ast.File) []string { } pkg, ok := sel.X.(*ast.Ident) - if !ok || pkg.Name != "flag" { + if !ok || (pkg.Name != "flag" && pkg.Name != "flagSet") { return true } - // Match Var-style flag declarations only (the form main.go + // Match Var-style flag declarations only (the form flags.go // actually uses). Bare flag.String / flag.Bool return a // pointer and aren't used here; if a future change moves to // them, expand this list. diff --git a/pkg/satellite/stream/server.go b/pkg/satellite/stream/server.go index 1689dea3..73481db3 100644 --- a/pkg/satellite/stream/server.go +++ b/pkg/satellite/stream/server.go @@ -38,6 +38,10 @@ import ( // itself (the snapshot stream) is unbounded and can take minutes. const readHeaderTimeout = 10 * time.Second +// idleTimeout closes a connection a peer left open after its stream. The +// client closes its own; this bounds one that does not. +const idleTimeout = time.Minute + // PoolResolver returns the storage.Provider for a (rd, vol) pair on // this satellite. The satellite's Reconciler keeps a resource→pool // map populated by Apply; this is the gate that lets the stream @@ -147,14 +151,19 @@ func closeQuietly(c io.Closer) { _ = c.Close() } -// ListenAndServe binds an HTTP server on addr serving the snapshot -// stream endpoint. Caller is responsible for cancelling ctx to stop +// ListenAndServe binds an HTTPS server on addr serving the snapshot +// stream endpoint, with mutual TLS from files: a peer without a certificate +// signed by the cluster's CA is refused at the handshake. Caller is responsible for cancelling ctx to stop // the server. The Server type doesn't manage its own goroutine — // this helper is the canonical wiring for cmd/satellite. // // Returns http.ErrServerClosed on a clean ctx-cancel shutdown so // callers can distinguish that from a real bind error. -func ListenAndServe(ctx context.Context, addr string, s *Server) error { +func ListenAndServe(ctx context.Context, addr string, s *Server, files TLSFiles) error { + if !files.Configured() { + return ErrTLSIncomplete + } + mux := http.NewServeMux() s.Register(mux) @@ -162,6 +171,8 @@ func ListenAndServe(ctx context.Context, addr string, s *Server) error { Addr: addr, Handler: mux, ReadHeaderTimeout: readHeaderTimeout, // CWE-400: bound the slowloris window + IdleTimeout: idleTimeout, + TLSConfig: files.serverConfig(), } var ( @@ -170,7 +181,8 @@ func ListenAndServe(ctx context.Context, addr string, s *Server) error { ) wg.Go(func() { - serveErr = srv.ListenAndServe() + // The certificate comes from TLSConfig, read per handshake. + serveErr = srv.ListenAndServeTLS("", "") }) <-ctx.Done() diff --git a/pkg/satellite/stream/stream.go b/pkg/satellite/stream/stream.go index 3480cdf4..d5e39b76 100644 --- a/pkg/satellite/stream/stream.go +++ b/pkg/satellite/stream/stream.go @@ -30,9 +30,11 @@ limitations under the License. // → 404 when this satellite doesn't host the snapshot // → 501 when the resolved provider has no SnapshotShipper capability // -// Security: cluster-internal traffic only. The daemonset binds to -// hostPort outside the DRBD range; production deployments should -// front this with a NetworkPolicy or mTLS. No auth on the wire yet. +// Security: the stream carries raw volume data, so it runs over mutual +// TLS only (see TLSFiles). Each satellite presents a certificate for +// ServerName signed by the cluster's CA, and each side verifies the other. +// A satellite with no stream TLS configured serves nothing and fetches +// nothing; a clone or restore then falls back to a DRBD resync. package stream // Port is the satellite-to-satellite stream port. The DaemonSet binds it diff --git a/pkg/satellite/stream/tls.go b/pkg/satellite/stream/tls.go new file mode 100644 index 00000000..e96a5812 --- /dev/null +++ b/pkg/satellite/stream/tls.go @@ -0,0 +1,156 @@ +// 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 stream + +import ( + "crypto/tls" + "crypto/x509" + "os" + + "github.com/cockroachdb/errors" +) + +// ServerName is the name every satellite's stream certificate carries and +// every fetch verifies. Satellites dial each other by node address, which +// is not on any certificate, so the identity checked is "a satellite of this +// cluster": a certificate for this name signed by the cluster's CA. +const ServerName = "blockstor-satellite" + +// ErrTLSIncomplete refuses a partial TLS configuration: a listener with a +// certificate but no CA would serve without checking who asks. +var ErrTLSIncomplete = errors.New("stream TLS needs a certificate, a key and a CA bundle") + +// ErrNoClientCertificate is a stream handshake with no client certificate. +var ErrNoClientCertificate = errors.New("no client certificate on the stream handshake") + +// ErrNoCACertificate is a CA bundle with no certificate in it. +var ErrNoCACertificate = errors.New("no certificate in the stream CA bundle") + +// TLSFiles names the PEM files the stream's mutual TLS is built from: one +// certificate per satellite, usable for both server and client auth, and the +// CA that signed every satellite's certificate. cert-manager writes all three +// into one Secret. +// +// The files are read on every handshake rather than once. A stream transfer +// is one handshake for minutes of data, so this costs nothing, and a +// rotated certificate is picked up with no watcher and no restart. +type TLSFiles struct { + CertFile string + KeyFile string + CAFile string +} + +// Configured reports whether all three files are set. +func (f TLSFiles) Configured() bool { + return f.CertFile != "" && f.KeyFile != "" && f.CAFile != "" +} + +// Validate refuses a configuration with some of the files set and not all, +// and a full set whose files do not load: an unreadable file or a key that +// does not match the certificate would otherwise start fine and fail every +// cross-node restore on the node. The files are still read again on every +// handshake; this is a check at startup only. +func (f TLSFiles) Validate() error { + if f.CertFile == "" && f.KeyFile == "" && f.CAFile == "" { + return nil + } + + if !f.Configured() { + return ErrTLSIncomplete + } + + _, _, err := f.load() + + return err +} + +func (f TLSFiles) load() (tls.Certificate, *x509.CertPool, error) { + cert, err := tls.LoadX509KeyPair(f.CertFile, f.KeyFile) + if err != nil { + return tls.Certificate{}, nil, errors.Wrap(err, "load the stream certificate") + } + + caPEM, err := os.ReadFile(f.CAFile) + if err != nil { + return tls.Certificate{}, nil, errors.Wrap(err, "read the stream CA bundle") + } + + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + return tls.Certificate{}, nil, errors.Wrapf(ErrNoCACertificate, "%s", f.CAFile) + } + + return cert, pool, nil +} + +// serverConfig answers every handshake with the files as they are now, and +// requires a client certificate signed by the CA. +func (f TLSFiles) serverConfig() *tls.Config { + return &tls.Config{ + MinVersion: tls.VersionTLS12, + GetConfigForClient: func(*tls.ClientHelloInfo) (*tls.Config, error) { + cert, pool, err := f.load() + if err != nil { + return nil, err + } + + return &tls.Config{ + MinVersion: tls.VersionTLS12, + Certificates: []tls.Certificate{cert}, + ClientCAs: pool, + ClientAuth: tls.RequireAndVerifyClientCert, + VerifyConnection: verifyClientName, + }, nil + }, + } +} + +// verifyClientName refuses a client certificate that chains to the CA but +// is not issued for ServerName. The stream CA issues one certificate today; +// this keeps "a satellite of this cluster" true if it ever issues more. +// +//nolint:gocritic // hugeParam: tls.Config.VerifyConnection fixes the signature. +func verifyClientName(cs tls.ConnectionState) error { + if len(cs.PeerCertificates) == 0 { + return ErrNoClientCertificate + } + + err := cs.PeerCertificates[0].VerifyHostname(ServerName) + if err != nil { + return errors.Wrap(err, "stream client certificate") + } + + return nil +} + +// clientConfig presents this satellite's certificate and verifies the peer's +// against the CA and ServerName. +func (f TLSFiles) clientConfig() (*tls.Config, error) { + cert, pool, err := f.load() + if err != nil { + return nil, err + } + + return &tls.Config{ + MinVersion: tls.VersionTLS12, + Certificates: []tls.Certificate{cert}, + RootCAs: pool, + ServerName: ServerName, + }, nil +} diff --git a/pkg/satellite/stream/tls_test.go b/pkg/satellite/stream/tls_test.go new file mode 100644 index 00000000..81ad6586 --- /dev/null +++ b/pkg/satellite/stream/tls_test.go @@ -0,0 +1,457 @@ +// 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 stream + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "math/big" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strconv" + "strings" + "testing" + "time" + + "github.com/cozystack/blockstor/pkg/storage" +) + +// The stream returns raw volume data, and it was served on every node +// address with no authentication: any pod that could reach a node read any +// volume with a snapshot there. It now runs over mutual TLS only. + +type testCA struct { + cert *x509.Certificate + key *ecdsa.PrivateKey + pem []byte +} + +func newTestCA(t *testing.T) testCA { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("ca key: %v", err) + } + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign, + } + + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatalf("ca cert: %v", err) + } + + cert, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse ca: %v", err) + } + + return testCA{cert: cert, key: key, pem: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})} +} + +// issue writes a satellite certificate for dnsName signed by ca, with the CA +// bundle next to it, and returns the files. +func (ca testCA) issue(t *testing.T, dnsName string) TLSFiles { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("leaf key: %v", err) + } + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(time.Now().UnixNano()), + Subject: pkix.Name{CommonName: dnsName}, + DNSNames: []string{dnsName}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}, + } + + der, err := x509.CreateCertificate(rand.Reader, tmpl, ca.cert, &key.PublicKey, ca.key) + if err != nil { + t.Fatalf("leaf cert: %v", err) + } + + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatalf("marshal key: %v", err) + } + + dir := t.TempDir() + files := TLSFiles{ + CertFile: filepath.Join(dir, "tls.crt"), + KeyFile: filepath.Join(dir, "tls.key"), + CAFile: filepath.Join(dir, "ca.crt"), + } + + writeFile(t, files.CertFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})) + writeFile(t, files.KeyFile, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})) + writeFile(t, files.CAFile, ca.pem) + + return files +} + +func writeFile(t *testing.T, path string, data []byte) { + t.Helper() + + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} + +// noSnapshots answers every lookup with "not on this node", so a request +// that gets past the handshake comes back 404. +type noSnapshots struct{} + +func (noSnapshots) ProviderForResource(string) (storage.Provider, error) { + return nil, storage.ErrNotFound +} + +// serve starts the stream server with files on a free loopback port. +func serve(t *testing.T, files TLSFiles) string { + t.Helper() + + var lc net.ListenConfig + + probe, err := lc.Listen(t.Context(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("probe port: %v", err) + } + + addr := probe.Addr().String() + _ = probe.Close() + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + go func() { _ = ListenAndServe(ctx, addr, NewServer(noSnapshots{}), files) }() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + var dialer net.Dialer + + conn, dialErr := dialer.DialContext(t.Context(), "tcp", addr) + if dialErr == nil { + _ = conn.Close() + + return addr + } + + time.Sleep(20 * time.Millisecond) + } + + t.Fatalf("stream server did not come up on %s", addr) + + return "" +} + +func TestStreamFetchSucceedsBetweenSatellitesOfOneCA(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + + _, err := NewClient(ca.issue(t, ServerName)).Fetch(t.Context(), addr, "rd", "snap", 0) + if !errors.Is(err, storage.ErrNotFound) || !strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("fetch between two satellites of one CA = %v, want the server's 404", err) + } +} + +func TestStreamRefusesAPeerWithoutACertificate(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + + pool := x509.NewCertPool() + pool.AddCert(ca.cert) + + client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{ + MinVersion: tls.VersionTLS12, RootCAs: pool, ServerName: ServerName, + }}} + + resp, err := client.Do(newGet(t, "https://"+addr+PathPrefix+"/rd/snap/"+strconv.Itoa(0))) + if err == nil { + _ = resp.Body.Close() + t.Fatalf("a client with no certificate got %s", resp.Status) + } + + plain, err := http.DefaultClient.Do(newGet(t, "http://"+addr+PathPrefix+"/rd/snap/0")) + if err == nil { + defer func() { _ = plain.Body.Close() }() + + if plain.StatusCode == http.StatusNotFound || plain.StatusCode == http.StatusOK { + t.Fatalf("plain HTTP reached the handler: %s", plain.Status) + } + } +} + +func TestStreamRefusesAPeerFromAnotherCA(t *testing.T) { + t.Parallel() + + addr := serve(t, newTestCA(t).issue(t, ServerName)) + + _, err := NewClient(newTestCA(t).issue(t, ServerName)).Fetch(t.Context(), addr, "rd", "snap", 0) + if err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a peer from another CA reached the handler: %v", err) + } +} + +func TestStreamRefusesAServerCertificateForAnotherName(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, "not-a-satellite")) + + _, err := NewClient(ca.issue(t, ServerName)).Fetch(t.Context(), addr, "rd", "snap", 0) + if err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a server certificate for another name was accepted: %v", err) + } +} + +// Without TLS configured the stream is off, not open. +func TestStreamWithoutTLSIsOff(t *testing.T) { + t.Parallel() + + _, err := NewClient(TLSFiles{}).Fetch(t.Context(), "127.0.0.1:1", "rd", "snap", 0) + if !errors.Is(err, ErrStreamDisabled) { + t.Fatalf("fetch with no TLS = %v, want ErrStreamDisabled", err) + } + + // The caller falls back to a blank volume and a DRBD resync only on + // storage.ErrNotFound; any other error requeues the restore forever. + if !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("fetch with no TLS = %v, want it to match storage.ErrNotFound", err) + } + + if err := ListenAndServe(t.Context(), "127.0.0.1:0", NewServer(noSnapshots{}), TLSFiles{}); !errors.Is(err, ErrTLSIncomplete) { + t.Fatalf("server with no TLS = %v, want ErrTLSIncomplete", err) + } + + if err := (TLSFiles{CertFile: "c", KeyFile: "k"}).Validate(); !errors.Is(err, ErrTLSIncomplete) { + t.Fatalf("partial TLS flags = %v, want ErrTLSIncomplete", err) + } +} + +func newGet(t *testing.T, url string) *http.Request { + t.Helper() + + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, url, http.NoBody) + if err != nil { + t.Fatalf("build request: %v", err) + } + + return req +} + +// The server, not the client, has to reject a client certificate signed by +// another CA: here the client trusts the server, so only the server's check +// stands between a self-signed client and the stream. +func TestStreamServerRefusesAClientCertificateFromAnotherCA(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + + intruder := newTestCA(t).issue(t, ServerName) + writeFile(t, intruder.CAFile, ca.pem) + + _, err := NewClient(intruder).Fetch(t.Context(), addr, "rd", "snap", 0) + if err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a client certificate from another CA reached the handler: %v", err) + } +} + +// The files are read on every handshake: a certificate replaced on disk is +// served from the next connection on, with no restart. +func TestStreamPicksUpARotatedCertificate(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + served := ca.issue(t, "not-a-satellite") + addr := serve(t, served) + client := NewClient(ca.issue(t, ServerName)) + + if _, err := client.Fetch(t.Context(), addr, "rd", "snap", 0); err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a certificate for another name was accepted before the rotation: %v", err) + } + + rotated := ca.issue(t, ServerName) + for _, pair := range [][2]string{{rotated.CertFile, served.CertFile}, {rotated.KeyFile, served.KeyFile}} { + data, err := os.ReadFile(pair[0]) + if err != nil { + t.Fatalf("read rotated file: %v", err) + } + + writeFile(t, pair[1], data) + } + + _, err := client.Fetch(t.Context(), addr, "rd", "snap", 0) + if !strings.Contains(fmt.Sprint(err), "has no snapshot") { + t.Fatalf("the rotated certificate was not served: %v", err) + } +} + +// Every fetch built a transport that pooled its connection and was then +// dropped, leaving the connection and its goroutines open on both sides. +func TestStreamFetchLeavesNoConnectionOpen(t *testing.T) { + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + client := NewClient(ca.issue(t, ServerName)) + + _, _ = client.Fetch(t.Context(), addr, "rd", "snap", 0) + + before := runtime.NumGoroutine() + + for range 30 { + _, _ = client.Fetch(t.Context(), addr, "rd", "snap", 0) + } + + deadline := time.Now().Add(5 * time.Second) + for runtime.NumGoroutine() > before+5 && time.Now().Before(deadline) { + time.Sleep(50 * time.Millisecond) + } + + if after := runtime.NumGoroutine(); after > before+5 { + t.Fatalf("goroutines grew from %d to %d over 30 fetches", before, after) + } +} + +// The client side reads its files per fetch too: a client certificate +// replaced on disk is presented from the next fetch on. +func TestStreamClientPicksUpARotatedCertificate(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + + held := newTestCA(t).issue(t, ServerName) + writeFile(t, held.CAFile, ca.pem) + client := NewClient(held) + + if _, err := client.Fetch(t.Context(), addr, "rd", "snap", 0); err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a client certificate from another CA got through before the rotation: %v", err) + } + + rotated := ca.issue(t, ServerName) + for _, pair := range [][2]string{{rotated.CertFile, held.CertFile}, {rotated.KeyFile, held.KeyFile}} { + data, err := os.ReadFile(pair[0]) + if err != nil { + t.Fatalf("read rotated file: %v", err) + } + + writeFile(t, pair[1], data) + } + + _, err := client.Fetch(t.Context(), addr, "rd", "snap", 0) + if !strings.Contains(fmt.Sprint(err), "has no snapshot") { + t.Fatalf("the rotated client certificate was not presented: %v", err) + } +} + +// During a rolling upgrade a new satellite fetches from an old one that +// still serves the stream in plain HTTP. The old server would answer 404 +// for a snapshot it does not hold, and a 404 read as storage.ErrNotFound +// sends the restore to a blank volume. The fetch has to fail on the +// handshake instead, so the restore retries until the peer is upgraded. +func TestStreamFetchFromAPlainServerIsNotANotFound(t *testing.T) { + t.Parallel() + + old := httptest.NewServer(http.NotFoundHandler()) + t.Cleanup(old.Close) + + addr := strings.TrimPrefix(old.URL, "http://") + + ca := newTestCA(t) + + _, err := NewClient(ca.issue(t, ServerName)).Fetch(t.Context(), addr, "rd", "snap", 0) + if err == nil || errors.Is(err, storage.ErrNotFound) { + t.Fatalf("fetch from a plain HTTP server = %v, want an error that is not storage.ErrNotFound", err) + } +} + +// A full set of files that does not load refuses to start, as the +// apiserver does on bad TLS material, instead of failing every cross-node +// restore on the node later. +func TestStreamTLSValidateLoadsTheFiles(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + + good := ca.issue(t, ServerName) + if err := good.Validate(); err != nil { + t.Fatalf("a valid set = %v, want nil", err) + } + + mismatched := good + mismatched.KeyFile = ca.issue(t, ServerName).KeyFile + + missing := good + missing.CAFile = filepath.Join(t.TempDir(), "ca.crt") + + emptyCA := ca.issue(t, ServerName) + writeFile(t, emptyCA.CAFile, []byte("no certificate here\n")) + + for name, files := range map[string]TLSFiles{ + "a key for another certificate": mismatched, + "a missing CA bundle": missing, + "a CA bundle with no cert": emptyCA, + } { + if err := files.Validate(); err == nil { + t.Errorf("%s: Validate = nil, want an error", name) + } + } +} + +// The server checks the name on the client certificate as well as its CA: +// a certificate the stream CA issued for anything but ServerName is not a +// satellite's. +func TestStreamServerRefusesAClientCertificateForAnotherName(t *testing.T) { + t.Parallel() + + ca := newTestCA(t) + addr := serve(t, ca.issue(t, ServerName)) + + _, err := NewClient(ca.issue(t, "not-a-satellite")).Fetch(t.Context(), addr, "rd", "snap", 0) + if err == nil || strings.Contains(err.Error(), "has no snapshot") { + t.Fatalf("a client certificate for another name reached the handler: %v", err) + } +} diff --git a/stand/blockstor-apiserver-tls.yaml b/stand/blockstor-apiserver-tls.yaml index dcb8bb2d..63f2d9f6 100644 --- a/stand/blockstor-apiserver-tls.yaml +++ b/stand/blockstor-apiserver-tls.yaml @@ -7,6 +7,9 @@ # └─> blockstor-api-ca (CA Issuer) # ├─> blockstor-apiserver-server (serving cert, Service DNS SANs) # └─> blockstor-apiserver-client (client cert for in-cluster consumers) +# └─> blockstor-stream-ca (CA Certificate, 10y, kept apart from the API CA) +# └─> blockstor-stream-ca (CA Issuer) +# └─> blockstor-satellite-stream (one cert for every satellite's snapshot stream) # # The apiserver's TLS listener (RequireAndVerifyClientCert) verifies # every client cert against ca.crt of the server Secret, and serves @@ -114,3 +117,64 @@ spec: issuerRef: name: blockstor-api-ca kind: Issuer +--- +# The snapshot stream between satellites has a CA of its own. A stream +# certificate is a client certificate, and one from blockstor-api-ca would +# also be accepted by the apiserver's mutual-TLS listener; keeping the CAs +# apart keeps a satellite's stream identity off the REST API. +apiVersion: cert-manager.io/v1 +kind: Certificate +metadata: + name: blockstor-stream-ca + namespace: blockstor-system +spec: + commonName: blockstor-stream-ca + secretName: blockstor-stream-ca + duration: 87600h # 10 years + isCA: true + privateKey: + rotationPolicy: Never + usages: + - signing + - key encipherment + - cert sign + issuerRef: + name: ca-bootstrapper + kind: Issuer +--- +apiVersion: cert-manager.io/v1 +kind: Issuer +metadata: + name: blockstor-stream-ca + namespace: blockstor-system +spec: + ca: + secretName: blockstor-stream-ca +--- +# One certificate for every satellite, for both ends of a stream: the +# satellites dial each other by node address, so the name checked is the +# fixed blockstor-satellite (stream.ServerName), not an address. +apiVersion: cert-manager.io/v1 +kind: Certificate +metadata: + name: blockstor-satellite-stream + namespace: blockstor-system +spec: + secretName: blockstor-satellite-stream-tls + commonName: blockstor-satellite + duration: 2160h # 90 days + renewBefore: 360h # 15 days + privateKey: + algorithm: ECDSA + size: 256 + rotationPolicy: Always + usages: + - server auth + - client auth + - digital signature + - key encipherment + dnsNames: + - blockstor-satellite + issuerRef: + name: blockstor-stream-ca + kind: Issuer diff --git a/stand/blockstor-satellite-daemonset.yaml b/stand/blockstor-satellite-daemonset.yaml index 33a91a6c..fa185952 100644 --- a/stand/blockstor-satellite-daemonset.yaml +++ b/stand/blockstor-satellite-daemonset.yaml @@ -179,6 +179,15 @@ spec: # satellites instead of letting Node.Online lie via # stale heartbeat-reconciler timestamps. - --health-probe-bind-address=:9101 + # Mutual TLS of the snapshot stream between satellites. + # Every satellite presents the same certificate, issued by + # the stream CA in blockstor-apiserver-tls.yaml, and accepts + # only peers that present one from that CA. Without these + # the stream is off and cross-node clone falls back to a + # DRBD initial sync. + - --stream-tls-cert-file=/etc/blockstor/stream-tls/tls.crt + - --stream-tls-key-file=/etc/blockstor/stream-tls/tls.key + - --stream-tls-ca-file=/etc/blockstor/stream-tls/ca.crt env: - name: NODE_NAME valueFrom: {fieldRef: {fieldPath: spec.nodeName}} @@ -192,14 +201,16 @@ spec: - name: POD_NAMESPACE valueFrom: {fieldRef: {fieldPath: metadata.namespace}} ports: - # Satellite-to-satellite snapshot stream (cross-node clone). - # The receiving satellite GETs the raw byte stream from a - # peer that hosts the snapshot locally; without this port - # open peers see ECONNREFUSED and cross-node clone falls - # back to DRBD initial-sync. hostNetwork=true means this - # is the node's real port — must stay outside the DRBD - # range (7000-7999), every LINSTOR port and node-exporter's - # 9100 (see stream.Port). + # Satellite-to-satellite snapshot stream (cross-node clone), + # HTTPS with mutual TLS. The receiving satellite GETs the + # raw byte stream from a peer that hosts the snapshot + # locally. With this port closed a peer gets a connection + # error, tries the other nodes that hold the snapshot and + # retries the restore when none answers; only a satellite + # with no stream TLS falls back to a DRBD initial sync. + # hostNetwork=true means this is the node's real port, so it + # must stay outside the DRBD range (7000-7999), every LINSTOR + # port and node-exporter's 9100 (see stream.Port). - name: snap-stream containerPort: 3372 hostPort: 3372 @@ -311,6 +322,7 @@ spec: # same host directory piraeus's own satellite sees fixes the # capacity report and unlocks rwx-ganesha.sh. - {name: piraeus-pool, mountPath: /var/lib/piraeus/file-thin} + - {name: stream-tls, mountPath: /etc/blockstor/stream-tls, readOnly: true} resources: requests: {cpu: 50m, memory: 64Mi} limits: {cpu: 500m, memory: 512Mi} @@ -344,6 +356,11 @@ spec: # the upstream image + a /etc/drbd-reactor.d ConfigMap mount; # no other DaemonSet plumbing change is needed. volumes: + # Read on every stream handshake, so a certificate cert-manager + # rotates is picked up without a restart. + - name: stream-tls + secret: + secretName: blockstor-satellite-stream-tls - {name: dev, hostPath: {path: /dev, type: Directory}} - {name: modules, hostPath: {path: /lib/modules}} - {name: lvm-run, hostPath: {path: /run/lvm, type: DirectoryOrCreate}} diff --git a/tests/scenarios/03-networking.md b/tests/scenarios/03-networking.md index 0d30fbee..e21ed174 100644 --- a/tests/scenarios/03-networking.md +++ b/tests/scenarios/03-networking.md @@ -124,6 +124,8 @@ After `node interface modify ... --active`, satellite re-dials the controller vi Note on `--controller-namespace` (Bug 023): the satellite binary grew this flag to name the NAMESPACE the cluster master passphrase Secret (`linstor encryption create-passphrase`) lives in — defaulting to `$POD_NAMESPACE` then `blockstor-system`. It is read through the kube-apiserver like every other satellite input; it carries no controller ADDRESS and does not change the Outcome B posture above. +Note on `--stream-tls-cert-file`, `--stream-tls-key-file` and `--stream-tls-ca-file`: the PEM files of the mutual TLS on the satellite-to-satellite snapshot stream. They name local files, carry no controller ADDRESS, and do not change the Outcome B posture either. + --- ## End-to-end on dedicated replication network