From d11a28974f3b36e0f6db7afd89a33c8e2fb09d3c Mon Sep 17 00:00:00 2001 From: MaBing Date: Thu, 8 Oct 2026 16:02:08 +0800 Subject: [PATCH 1/2] feat(s3): support configurable HTTPS TLS --- api/dataset/v1alpha1/dataset_types.go | 5 +- .../bases/dataset.baizeai.io_datasets.yaml | 5 +- .../controller/dataset/dataset_controller.go | 5 + internal/controller/dataset/s3_tls.go | 42 +++++ internal/controller/dataset/s3_tls_test.go | 73 +++++++++ internal/pkg/datasources/datasource_s3.go | 81 +++++++++- .../datasources/s3_tls_integration_test.go | 147 ++++++++++++++++++ internal/pkg/datasources/s3_tls_test.go | 92 +++++++++++ manifests/dataset/values.yaml | 2 + pkg/s3tls/tls.go | 93 +++++++++++ pkg/s3tls/tls_test.go | 67 ++++++++ 11 files changed, 602 insertions(+), 10 deletions(-) create mode 100644 internal/controller/dataset/s3_tls.go create mode 100644 internal/controller/dataset/s3_tls_test.go create mode 100644 internal/pkg/datasources/s3_tls_integration_test.go create mode 100644 internal/pkg/datasources/s3_tls_test.go create mode 100644 pkg/s3tls/tls.go create mode 100644 pkg/s3tls/tls_test.go diff --git a/api/dataset/v1alpha1/dataset_types.go b/api/dataset/v1alpha1/dataset_types.go index 3fe5e80..9c4c5c8 100644 --- a/api/dataset/v1alpha1/dataset_types.go +++ b/api/dataset/v1alpha1/dataset_types.go @@ -87,7 +87,10 @@ type DatasetSource struct { // options is a map of key-value pairs that can be used to specify additional options for the dataset source, e.g. {"branch": "master"} // supported keys for each type of dataset source are: // - GIT: branch, commit, depth, submodules - // - S3: region, endpoint, provider, syncMode + // - S3: region, endpoint, provider, syncMode, tlsMode, tlsCASecretName + // tlsMode: system (default), customCA, insecureSkipVerify. + // customCA requires a same-namespace Secret reference (tlsCASecretName), key ca.crt. + // CA PEM and the internal loader path tlsCAFile must not be stored in options. // - HTTP: any key-value pair will be passed to the underlying http client as http headers // - PVC: // - NFS: diff --git a/config/crd/bases/dataset.baizeai.io_datasets.yaml b/config/crd/bases/dataset.baizeai.io_datasets.yaml index 2b50914..4408c9a 100644 --- a/config/crd/bases/dataset.baizeai.io_datasets.yaml +++ b/config/crd/bases/dataset.baizeai.io_datasets.yaml @@ -297,7 +297,10 @@ spec: options is a map of key-value pairs that can be used to specify additional options for the dataset source, e.g. {"branch": "master"} supported keys for each type of dataset source are: - GIT: branch, commit, depth, submodules - - S3: region, endpoint, provider, syncMode + - S3: region, endpoint, provider, syncMode, tlsMode, tlsCASecretName + tlsMode: system (default), customCA, insecureSkipVerify. + customCA requires a same-namespace Secret reference (tlsCASecretName), key ca.crt. + CA PEM and the internal loader path tlsCAFile must not be stored in options. - HTTP: any key-value pair will be passed to the underlying http client as http headers - PVC: - NFS: diff --git a/internal/controller/dataset/dataset_controller.go b/internal/controller/dataset/dataset_controller.go index 71fa2a4..77506d5 100644 --- a/internal/controller/dataset/dataset_controller.go +++ b/internal/controller/dataset/dataset_controller.go @@ -747,6 +747,11 @@ func (r *DatasetReconciler) reconcileJob(ctx context.Context, ds *datasetv1alpha } podSpec := &jobSpec.Template.Spec + if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeS3 { + if err := configureS3TLS(options, podSpec, container); err != nil { + return err + } + } // conda 类型需要将 ConfigMap mount 到容器 condaKeyItems := make([]corev1.KeyToPath, 0, 2) diff --git a/internal/controller/dataset/s3_tls.go b/internal/controller/dataset/s3_tls.go new file mode 100644 index 0000000..477e0c0 --- /dev/null +++ b/internal/controller/dataset/s3_tls.go @@ -0,0 +1,42 @@ +package dataset + +import ( + "fmt" + + corev1 "k8s.io/api/core/v1" + + "github.com/BaizeAI/dataset/pkg/s3tls" +) + +// configureS3TLS operates on the copied options only. Kubernetes reads Secrets. +func configureS3TLS(options map[string]string, pod *corev1.PodSpec, container *corev1.Container) error { + mode, err := s3tls.Mode(options["tlsMode"]) + if err != nil { + return err + } + name := options["tlsCASecretName"] + if mode == s3tls.CustomCA && name == "" { + return fmt.Errorf("customCA requires tlsCASecretName") + } + if mode != s3tls.CustomCA && name != "" { + return fmt.Errorf("tlsCASecretName is only allowed with customCA") + } + delete(options, "tlsCASecretName") + delete(options, "tlsCAFile") + options["tlsMode"] = mode + if mode != s3tls.CustomCA { + return nil + } + const volume = "dataset-s3-tls" + const dir = "/run/dataset/s3-tls" + pod.Volumes = append(pod.Volumes, corev1.Volume{ + Name: volume, + VolumeSource: corev1.VolumeSource{Secret: &corev1.SecretVolumeSource{ + SecretName: name, + Items: []corev1.KeyToPath{{Key: "ca.crt", Path: "ca.crt"}}, + }}, + }) + container.VolumeMounts = append(container.VolumeMounts, corev1.VolumeMount{Name: volume, MountPath: dir, ReadOnly: true}) + options["tlsCAFile"] = dir + "/ca.crt" + return nil +} diff --git a/internal/controller/dataset/s3_tls_test.go b/internal/controller/dataset/s3_tls_test.go new file mode 100644 index 0000000..13f291b --- /dev/null +++ b/internal/controller/dataset/s3_tls_test.go @@ -0,0 +1,73 @@ +package dataset + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/require" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + dsapi "github.com/BaizeAI/dataset/api/dataset/v1alpha1" +) + +func TestS3TLSJob(t *testing.T) { + for _, mode := range []string{"", "system", "customCA", "insecureSkipVerify"} { + t.Run(mode, func(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, dsapi.AddToScheme(scheme)) + require.NoError(t, batchv1.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + // No Secret is installed in the fake client. Job creation must not read it. + c := fake.NewClientBuilder().WithScheme(scheme).Build() + r := &DatasetReconciler{Client: c} + ds := &dsapi.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "s3", Namespace: "test", UID: "uid"}, Spec: dsapi.DatasetSpec{DataSyncRound: 1, SecretRef: "credentials", Source: dsapi.DatasetSource{Type: dsapi.DatasetTypeS3, URI: "s3://bucket", Options: map[string]string{"tlsMode": mode, "tlsCAFile": "/untrusted/path"}}}} + ds.Status.PVCName = "data" + if mode == "customCA" { + ds.Spec.Source.Options["tlsCASecretName"] = "ca" + } + before := ds.DeepCopy().Spec.Source.Options + require.NoError(t, r.reconcileJob(context.Background(), ds)) + require.Equal(t, before, ds.Spec.Source.Options) + jobs := &batchv1.JobList{} + require.NoError(t, c.List(context.Background(), jobs)) + require.Len(t, jobs.Items, 1) + pod := jobs.Items[0].Spec.Template.Spec + container := pod.Containers[0] + args := strings.Join(container.Args, " ") + require.NotContains(t, args, "tlsCASecretName") + require.NotContains(t, args, "/untrusted/path") + if mode == "customCA" { + require.Len(t, pod.Volumes, 3) + ca := pod.Volumes[0].Secret + require.NotNil(t, ca) + require.Equal(t, "ca", ca.SecretName) + require.Nil(t, ca.Optional) + require.Equal(t, []corev1.KeyToPath{{Key: "ca.crt", Path: "ca.crt"}}, ca.Items) + mount := container.VolumeMounts[0] + require.True(t, mount.ReadOnly) + require.Empty(t, mount.SubPath) + require.Equal(t, "/run/dataset/s3-tls", mount.MountPath) + require.Contains(t, args, "--options=tlsCAFile=/run/dataset/s3-tls/ca.crt") + } else { + require.Len(t, pod.Volumes, 2) + require.NotContains(t, args, "tlsCAFile") + } + require.Equal(t, "credentials", pod.Volumes[len(pod.Volumes)-2].Secret.SecretName) + require.Equal(t, "data", pod.Volumes[len(pod.Volumes)-1].PersistentVolumeClaim.ClaimName) + }) + } +} + +func TestS3TLSJobRejectsInvalidReferences(t *testing.T) { + for _, options := range []map[string]string{{"tlsMode": "unknown"}, {"tlsMode": "customCA"}, {"tlsMode": "system", "tlsCASecretName": "ca"}, {"tlsMode": "insecureSkipVerify", "tlsCASecretName": "ca"}} { + pod := &corev1.PodSpec{} + container := &corev1.Container{} + require.Error(t, configureS3TLS(options, pod, container)) + require.Empty(t, pod.Volumes) + } +} diff --git a/internal/pkg/datasources/datasource_s3.go b/internal/pkg/datasources/datasource_s3.go index 9009376..621c198 100644 --- a/internal/pkg/datasources/datasource_s3.go +++ b/internal/pkg/datasources/datasource_s3.go @@ -3,13 +3,17 @@ package datasources import ( "encoding/json" "fmt" - "github.com/samber/lo" + "io" "net/url" "os" "os/exec" "path/filepath" "strings" + "github.com/samber/lo" + + "github.com/BaizeAI/dataset/pkg/s3tls" + "github.com/sirupsen/logrus" "github.com/BaizeAI/dataset/pkg/log" @@ -37,7 +41,7 @@ func NewS3Loader(datasourceOptions map[string]string, options Options, secrets S s3.s3Options.secretAccessKey = secrets.AKSKSecretAccessKey s3.s3Options.SyncMode = lo.CoalesceOrEmpty(datasourceOptions["syncMode"], "sync") - err = s3.validateOptions(s3Options) + err = s3.validateOptions(s3.s3Options) if err != nil { return nil, err } @@ -46,10 +50,12 @@ func NewS3Loader(datasourceOptions map[string]string, options Options, secrets S } type S3LoaderOptions struct { - Provider string `json:"provider"` - Region string `json:"region"` - Endpoint string `json:"endpoint"` - SyncMode string `json:"syncMode"` + Provider string `json:"provider"` + Region string `json:"region"` + Endpoint string `json:"endpoint"` + SyncMode string `json:"syncMode"` + TLSMode string `json:"tlsMode"` + TLSCAFile string `json:"tlsCAFile"` accessKeyID string secretAccessKey string @@ -79,7 +85,36 @@ func (d *S3Loader) validateOptions(options S3LoaderOptions) error { return fmt.Errorf("invalid syncMode '%s', must be 'sync' or 'copy'", options.SyncMode) } - return nil + mode, err := s3tls.Mode(options.TLSMode) + if err != nil { + return err + } + if err := s3tls.ValidateEndpoint(mode, options.Provider, options.Endpoint); err != nil { + return err + } + if mode != s3tls.CustomCA { + if options.TLSCAFile != "" { + return fmt.Errorf("tlsCAFile is only allowed with customCA") + } + return nil + } + if options.TLSCAFile == "" { + return fmt.Errorf("tlsCAFile is required with customCA") + } + f, err := os.Open(options.TLSCAFile) + if err != nil { + return fmt.Errorf("cannot read TLS CA file: %w", err) + } + defer func() { + if err := f.Close(); err != nil { + log.Warnf("failed to close TLS CA file: %v", err) + } + }() + data, err := io.ReadAll(io.LimitReader(f, s3tls.MaxCABytes+1)) + if err != nil { + return fmt.Errorf("cannot read TLS CA file: %w", err) + } + return s3tls.ValidateCA(data) } func (d *S3Loader) mapProviderEnumStringToRCloneProvider(provider string) string { @@ -95,6 +130,7 @@ func (d *S3Loader) mapProviderEnumStringToRCloneProvider(provider string) string func (d *S3Loader) configTouch() error { cmd := exec.Command("rclone", "config", "touch") + cmd.Env = s3TLSEnvironment(os.Environ()) logger := log.WithField("command", cmd.String()) logger.Debug("executing command to touch rclone config") @@ -149,6 +185,7 @@ func (d *S3Loader) configCreate(configName string) error { } cmd := exec.Command("rclone", args...) + cmd.Env = s3TLSEnvironment(os.Environ()) logger = logger.WithField("command", cmd.String()) logger.Debug("executing command to create a new rclone config") @@ -165,6 +202,10 @@ func (d *S3Loader) configCreate(configName string) error { } func (d *S3Loader) Sync(fromURI string, toPath string) error { + // Validate again before launching rclone: projected Secret contents may change. + if err := d.validateOptions(d.s3Options); err != nil { + return err + } parsedURL, err := url.Parse(d.Options.URI) if err != nil { return err @@ -214,13 +255,14 @@ func (d *S3Loader) Sync(fromURI string, toPath string) error { } args = append(args, "-vvv") + args = append(args, d.tlsArgs()...) cmd := exec.Command("rclone", args...) cmd.Dir = d.Options.Root logger = logger.WithField("command", cmd.String()) logger.Debug("executing command to copy data") - cmd.Env = os.Environ() + cmd.Env = s3TLSEnvironment(os.Environ()) if accessKeyID != "" && secretAccessKey != "" { cmd.Env = append(cmd.Env, fmt.Sprintf("RCLONE_S3_ACCESS_KEY_ID=%s", accessKeyID)) @@ -237,3 +279,26 @@ func (d *S3Loader) Sync(fromURI string, toPath string) error { return nil } + +func (d *S3Loader) tlsArgs() []string { + if d.s3Options.TLSMode == s3tls.InsecureSkipVerify { + return []string{"--no-check-certificate=true"} + } + args := []string{"--no-check-certificate=false"} + if d.s3Options.TLSMode == s3tls.CustomCA { + args = append(args, "--ca-cert", d.s3Options.TLSCAFile) + } + return args +} + +func s3TLSEnvironment(env []string) []string { + result := make([]string, 0, len(env)) + for _, entry := range env { + key, _, _ := strings.Cut(entry, "=") + if strings.EqualFold(key, "RCLONE_NO_CHECK_CERTIFICATE") || strings.EqualFold(key, "RCLONE_CA_CERT") { + continue + } + result = append(result, entry) + } + return result +} diff --git a/internal/pkg/datasources/s3_tls_integration_test.go b/internal/pkg/datasources/s3_tls_integration_test.go new file mode 100644 index 0000000..9cdfc47 --- /dev/null +++ b/internal/pkg/datasources/s3_tls_integration_test.go @@ -0,0 +1,147 @@ +package datasources + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/md5" // Test object checksum for the S3 protocol. + "crypto/rand" + "crypto/tls" + "crypto/x509" + "encoding/pem" + "fmt" + "io" + "log" + "math/big" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// Run with S3_TLS_INTEGRATION=1 and the image's rclone v1.70.1 on PATH. +// This exercises actual HTTPS and S3 transfers without external credentials. +func TestS3TLSIntegration(t *testing.T) { + if os.Getenv("S3_TLS_INTEGRATION") != "1" { + t.Skip("set S3_TLS_INTEGRATION=1 to run the pinned rclone HTTPS fixture") + } + version, err := exec.Command("rclone", "version").Output() + require.NoError(t, err) + require.True(t, strings.HasPrefix(string(version), "rclone v1.70.1\n"), "use the data-loader's pinned rclone v1.70.1") + t.Setenv("RCLONE_CONFIG", filepath.Join(t.TempDir(), "rclone.conf")) + t.Setenv("RCLONE_RETRIES", "1") + t.Setenv("RCLONE_LOW_LEVEL_RETRIES", "1") + t.Setenv("RCLONE_RETRIES_SLEEP", "10ms") + t.Setenv("RCLONE_CONNECT_TIMEOUT", "2s") + t.Setenv("HTTPS_PROXY", "") + t.Setenv("HTTP_PROXY", "") + t.Setenv("RCLONE_NO_CHECK_CERTIFICATE", "true") + t.Setenv("RCLONE_CA_CERT", "/invalid/environment/ca") + for _, operation := range []string{"copy", "sync"} { + for _, scenario := range []struct { + name, mode string + wrongCA, badSAN, expired, systemTrust, ok bool + }{ + {name: "system trusted", mode: "system", systemTrust: true, ok: true}, + {name: "system untrusted", mode: "system"}, + {name: "custom trusted", mode: "customCA", ok: true}, + {name: "custom wrong CA", mode: "customCA", wrongCA: true}, + {name: "custom replaces system pool", mode: "customCA", wrongCA: true, systemTrust: true}, + {name: "custom SAN mismatch", mode: "customCA", badSAN: true}, + {name: "custom expired", mode: "customCA", expired: true}, + {name: "insecure explicit", mode: "insecureSkipVerify", badSAN: true, expired: true, ok: true}, + } { + t.Run(operation+"/"+scenario.name, func(t *testing.T) { + cert, bundle := s3IntegrationCertificate(t, scenario.badSAN, scenario.expired) + payload := "dataset HTTPS TLS fixture\n" + checksum := fmt.Sprintf("%x", md5.Sum([]byte(payload))) + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/bucket": + w.Header().Set("Content-Type", "application/xml") + if _, err := fmt.Fprintf(w, `bucket11000falseobject.txt2026-01-01T00:00:00.000Z"%s"%dSTANDARD`, checksum, len(payload)); err != nil { + t.Errorf("write S3 bucket listing: %v", err) + } + case "/bucket/object.txt": + w.Header().Set("Content-Length", fmt.Sprint(len(payload))) + w.Header().Set("ETag", `"`+checksum+`"`) + w.Header().Set("Last-Modified", "Thu, 01 Jan 2026 00:00:00 GMT") + if r.Method != "HEAD" { + if _, err := io.WriteString(w, payload); err != nil { + t.Errorf("write S3 object: %v", err) + } + } + default: + http.NotFound(w, r) + } + })) + server.Config.ErrorLog = log.New(io.Discard, "", 0) + server.TLS = &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12} + server.StartTLS() + defer server.Close() + root := t.TempDir() + ca := filepath.Join(root, "ca.crt") + trusted := filepath.Join(root, "system.crt") + require.NoError(t, os.WriteFile(trusted, bundle, 0600)) + if scenario.wrongCA { + _, bundle = s3IntegrationCertificate(t, false, false) + } + require.NoError(t, os.WriteFile(ca, bundle, 0600)) + if scenario.systemTrust { + t.Setenv("SSL_CERT_FILE", trusted) + } else { + t.Setenv("SSL_CERT_FILE", filepath.Join(root, "missing")) + } + t.Setenv("SSL_CERT_DIR", root) + // SSL_CERT_DIR must not contain the uploaded CA: that would trust it in SYSTEM. + emptySystem := filepath.Join(root, "system-dir") + require.NoError(t, os.Mkdir(emptySystem, 0700)) + t.Setenv("SSL_CERT_DIR", emptySystem) + options := map[string]string{"provider": "MINIO", "endpoint": server.URL, "syncMode": operation, "tlsMode": scenario.mode} + if scenario.mode == "customCA" { + options["tlsCAFile"] = ca + } + destination := filepath.Join(root, "data") + require.NoError(t, os.Mkdir(destination, 0700)) + loader, err := NewS3Loader(options, Options{URI: "s3://bucket", Root: root}, Secrets{AKSKAccessKeyID: "fixture-access", AKSKSecretAccessKey: "fixture-secret"}) + require.NoError(t, err) + err = loader.Sync("s3://bucket", destination) + if scenario.ok { + require.NoError(t, err) + data, err := os.ReadFile(filepath.Join(destination, "object.txt")) + require.NoError(t, err) + require.Equal(t, payload, string(data)) + } else { + require.Error(t, err) + _, err = os.Stat(filepath.Join(destination, "object.txt")) + require.True(t, os.IsNotExist(err)) + } + }) + } + } +} + +func s3IntegrationCertificate(t *testing.T, badSAN, expired bool) (tls.Certificate, []byte) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + cert := &x509.Certificate{SerialNumber: big.NewInt(time.Now().UnixNano()), NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour), KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, IsCA: true, BasicConstraintsValid: true} + if badSAN { + cert.DNSNames = []string{"wrong.example"} + } else { + cert.IPAddresses = []net.IP{net.ParseIP("127.0.0.1")} + } + if expired { + cert.NotBefore = time.Now().Add(-2 * time.Hour) + cert.NotAfter = time.Now().Add(-time.Hour) + } + der, err := x509.CreateCertificate(rand.Reader, cert, cert, &key.PublicKey, key) + require.NoError(t, err) + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} diff --git a/internal/pkg/datasources/s3_tls_test.go b/internal/pkg/datasources/s3_tls_test.go new file mode 100644 index 0000000..257f498 --- /dev/null +++ b/internal/pkg/datasources/s3_tls_test.go @@ -0,0 +1,92 @@ +package datasources + +import ( + "encoding/pem" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func tlsTestCA(t *testing.T) string { + t.Helper() + server := httptest.NewTLSServer(nil) + defer server.Close() + path := filepath.Join(t.TempDir(), "ca.crt") + require.NoError(t, os.WriteFile(path, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw}), 0600)) + return path +} + +func TestS3TLSValidation(t *testing.T) { + ca := tlsTestCA(t) + for _, c := range []struct { + mode, file, endpoint, provider string + ok bool + }{ + {"", "", "legacy-host", "MINIO", true}, {"system", "", "http://host", "MINIO", true}, + {"customCA", ca, "https://host", "MINIO", true}, {"customCA", ca, "", "AWS", true}, + {"insecureSkipVerify", "", "", "AWS", true}, {"insecureSkipVerify", "", "https://host", "MINIO", true}, + {"customCA", "", "https://host", "MINIO", false}, {"customCA", ca + ".missing", "https://host", "MINIO", false}, + {"customCA", ca, "http://host", "MINIO", false}, {"system", ca, "https://host", "MINIO", false}, + {"insecureSkipVerify", ca, "https://host", "MINIO", false}, {"unknown", "", "https://host", "MINIO", false}, + } { + _, err := NewS3Loader(map[string]string{"provider": c.provider, "region": "us-east-1", "endpoint": c.endpoint, "tlsMode": c.mode, "tlsCAFile": c.file}, Options{}, Secrets{}) + require.Equal(t, c.ok, err == nil, "%+v: %v", c, err) + } + require.NoError(t, os.WriteFile(ca, []byte("invalid"), 0600)) + _, err := NewS3Loader(map[string]string{"provider": "AWS", "region": "us-east-1", "tlsMode": "customCA", "tlsCAFile": ca}, Options{}, Secrets{}) + require.Error(t, err) +} + +func TestS3TLSTransferCommand(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("executable shell fixture requires Linux") + } + ca := tlsTestCA(t) + bin := t.TempDir() + capture := filepath.Join(bin, "capture") + script := "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$CAPTURE\"\nenv >> \"$CAPTURE\"\n" + require.NoError(t, os.WriteFile(filepath.Join(bin, "rclone"), []byte(script), 0700)) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("CAPTURE", capture) + t.Setenv("RCLONE_NO_CHECK_CERTIFICATE", "true") + t.Setenv("RCLONE_CA_CERT", "/bad/ca") + t.Setenv("HTTPS_PROXY", "http://proxy") + t.Setenv("SSL_CERT_FILE", "/system/ca") + for _, mode := range []string{"system", "customCA", "insecureSkipVerify"} { + for _, operation := range []string{"copy", "sync"} { + require.NoError(t, os.WriteFile(capture, nil, 0600)) + options := map[string]string{"provider": "AWS", "region": "us-east-1", "syncMode": operation, "tlsMode": mode} + if mode == "customCA" { + options["tlsCAFile"] = ca + } + loader, err := NewS3Loader(options, Options{URI: "s3://bucket/path", Root: bin}, Secrets{AKSKAccessKeyID: "access", AKSKSecretAccessKey: "secret"}) + require.NoError(t, err) + require.NoError(t, loader.Sync("s3://bucket/path", bin)) + bytes, err := os.ReadFile(capture) + require.NoError(t, err) + output := string(bytes) + require.Contains(t, output, operation+" baize-data-loader-copy-config-") + require.NotContains(t, output, "RCLONE_NO_CHECK_CERTIFICATE=") + require.NotContains(t, output, "RCLONE_CA_CERT=") + require.Contains(t, output, "HTTPS_PROXY=http://proxy") + require.Contains(t, output, "SSL_CERT_FILE=/system/ca") + require.Contains(t, output, "RCLONE_S3_ACCESS_KEY_ID=access") + if mode == "customCA" { + require.Contains(t, output, "--ca-cert "+ca) + } else { + require.NotContains(t, output, "--ca-cert ") + } + require.Contains(t, output, "--no-check-certificate="+map[bool]string{true: "true", false: "false"}[mode == "insecureSkipVerify"]) + } + } +} + +func TestS3TLSEnvironmentPreservesOtherSettings(t *testing.T) { + result := s3TLSEnvironment([]string{"RCLONE_NO_CHECK_CERTIFICATE=true", "RCLONE_CA_CERT=x", "HTTPS_PROXY=proxy", "AWS_ACCESS_KEY_ID=key", "SSL_CERT_FILE=system"}) + require.Equal(t, "HTTPS_PROXY=proxy,AWS_ACCESS_KEY_ID=key,SSL_CERT_FILE=system", strings.Join(result, ",")) +} diff --git a/manifests/dataset/values.yaml b/manifests/dataset/values.yaml index 8a7582b..4baea66 100644 --- a/manifests/dataset/values.yaml +++ b/manifests/dataset/values.yaml @@ -8,6 +8,7 @@ global: debug: false config: + # A custom loader image must implement the same S3 TLS contract as the controller. dataset_job_spec: {} # NFS protocol version used when creating new NFS PVs. Supported values are # 3, 4.0, 4.1, and 4.2. This value is passed to the controller through @@ -67,6 +68,7 @@ tolerations: [] affinity: {} +# Release CI pins both images to the same release tag. Upgrade both before enabling S3 TLS. controller: image: registry: '' diff --git a/pkg/s3tls/tls.go b/pkg/s3tls/tls.go new file mode 100644 index 0000000..10f7e06 --- /dev/null +++ b/pkg/s3tls/tls.go @@ -0,0 +1,93 @@ +// Package s3tls validates the S3 TLS contract without contacting object storage. +package s3tls + +import ( + "bytes" + "crypto/x509" + "encoding/pem" + "fmt" + "net/url" + "strconv" + "strings" + "unicode/utf8" +) + +const ( + System = "system" + CustomCA = "customCA" + InsecureSkipVerify = "insecureSkipVerify" + MaxCABytes = 64 * 1024 +) + +func Mode(mode string) (string, error) { + switch mode { + case "", System: + return System, nil + case CustomCA, InsecureSkipVerify: + return mode, nil + default: + return "", fmt.Errorf("tls.mode: unknown TLS mode") + } +} + +func ValidateEndpoint(mode, provider, endpoint string) error { + mode, err := Mode(mode) + if err != nil { + return err + } + if mode == System || (provider == "AWS" && endpoint == "") { + return nil + } + u, err := url.Parse(endpoint) + if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil { + return fmt.Errorf("tls.mode: non-default TLS requires a valid HTTPS endpoint (or an empty AWS endpoint)") + } + if strings.HasSuffix(u.Host, ":") { + return fmt.Errorf("endpoint: invalid port") + } + if port := u.Port(); port != "" { + n, err := strconv.Atoi(port) + if err != nil || n < 1 || n > 65535 { + return fmt.Errorf("endpoint: invalid port") + } + } + return nil +} + +// ValidateCA consumes every block. pem.Decode alone would skip malformed blocks +// and leading garbage, potentially accepting only part of an uploaded bundle. +func ValidateCA(data []byte) error { + if len(data) == 0 || len(data) > MaxCABytes || !utf8.Valid(data) { + return fmt.Errorf("tls.caCert: expected a UTF-8 PEM bundle of 1 to 65536 bytes") + } + rest := bytes.TrimSpace(data) + count := 0 + for len(rest) > 0 { + const begin = "-----BEGIN CERTIFICATE-----" + const end = "-----END CERTIFICATE-----" + if !bytes.HasPrefix(rest, []byte(begin)) { + return fmt.Errorf("tls.caCert: only CERTIFICATE PEM blocks and whitespace are allowed") + } + i := bytes.Index(rest, []byte(end)) + if i < 0 { + return fmt.Errorf("tls.caCert: malformed PEM block") + } + segment := rest[:i+len(end)] + if bytes.Contains(segment[len(begin):], []byte("-----BEGIN")) { + return fmt.Errorf("tls.caCert: malformed PEM block") + } + block, tail := pem.Decode(segment) + if block == nil || len(tail) != 0 || block.Type != "CERTIFICATE" || len(block.Headers) != 0 { + return fmt.Errorf("tls.caCert: invalid certificate PEM block") + } + if _, err := x509.ParseCertificate(block.Bytes); err != nil { + return fmt.Errorf("tls.caCert: invalid X.509 certificate") + } + count++ + rest = bytes.TrimSpace(rest[i+len(end):]) + } + if count == 0 { + return fmt.Errorf("tls.caCert: at least one certificate is required") + } + return nil +} diff --git a/pkg/s3tls/tls_test.go b/pkg/s3tls/tls_test.go new file mode 100644 index 0000000..7965755 --- /dev/null +++ b/pkg/s3tls/tls_test.go @@ -0,0 +1,67 @@ +package s3tls + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "encoding/pem" + "math/big" + "strings" + "testing" +) + +func TestValidateCA(t *testing.T) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + cert := &x509.Certificate{SerialNumber: big.NewInt(1)} + der, err := x509.CreateCertificate(rand.Reader, cert, cert, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + valid := string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})) + bad := "-----BEGIN CERTIFICATE-----\n@@\n-----END CERTIFICATE-----\n" + cases := []struct { + name, data string + ok bool + }{ + {"one", valid, true}, {"duplicate", valid + valid, true}, + {"CRLF", strings.ReplaceAll(valid, "\n", "\r\n"), true}, + {"boundary", valid + strings.Repeat(" ", MaxCABytes-len(valid)), true}, + {"too large", valid + strings.Repeat(" ", MaxCABytes-len(valid)+1), false}, + {"empty", "", false}, {"whitespace", " \n\t", false}, + {"private key", valid + "-----BEGIN PRIVATE KEY-----\nAQ==\n-----END PRIVATE KEY-----", false}, + {"leading garbage", "garbage" + valid, false}, {"trailing garbage", valid + "garbage", false}, + {"bad first", bad + valid, false}, {"bad middle", valid + bad + valid, false}, {"bad last", valid + bad, false}, + {"nested begin", "-----BEGIN CERTIFICATE-----\n" + valid, false}, + {"invalid DER", "-----BEGIN CERTIFICATE-----\nAQ==\n-----END CERTIFICATE-----", false}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if err := ValidateCA([]byte(c.data)); (err == nil) != c.ok { + t.Fatalf("valid=%v, error=%v", c.ok, err) + } + }) + } +} + +func TestEndpointContract(t *testing.T) { + for _, c := range []struct { + mode, provider, endpoint string + ok bool + }{ + {"", "MINIO", "old-endpoint:9000", true}, {System, "MINIO", "http://host", true}, + {CustomCA, "AWS", "", true}, {InsecureSkipVerify, "AWS", "", true}, + {CustomCA, "MINIO", "https://host:9443", true}, {CustomCA, "MINIO", "https://[::1]:443", true}, + {CustomCA, "MINIO", "", false}, {CustomCA, "MINIO", "http://host", false}, + {CustomCA, "MINIO", "https://", false}, {CustomCA, "MINIO", "https://host:abc", false}, + {CustomCA, "MINIO", "https://host:65536", false}, {CustomCA, "MINIO", "https://host:", false}, + {"unknown", "AWS", "", false}, + } { + if err := ValidateEndpoint(c.mode, c.provider, c.endpoint); (err == nil) != c.ok { + t.Errorf("%+v: %v", c, err) + } + } +} From 666dd6206242b92a9f3dca00917704e6935054fc Mon Sep 17 00:00:00 2001 From: MaBing Date: Wed, 14 Oct 2026 21:24:03 +0800 Subject: [PATCH 2/2] feat(nfs): support per-dataset versions --- README.md | 19 +++ api/dataset/v1alpha1/dataset_types.go | 5 +- .../v1alpha1/dataset_types_envtest_test.go | 48 ++++++ config/config.go | 13 +- .../bases/dataset.baizeai.io_datasets.yaml | 16 +- config/sample-config.yaml | 2 +- .../controller/dataset/dataset_controller.go | 24 ++- internal/controller/dataset/nfs_uri.go | 38 +++++ internal/controller/dataset/nfs_uri_test.go | 149 ++++++++++++++++++ .../controller/dataset/nfs_version_test.go | 56 +++++++ manifests/dataset/templates/deployment.yaml | 1 + manifests/dataset/values.yaml | 2 +- pkg/nfsconfig/version.go | 45 ++++++ pkg/nfsconfig/version_test.go | 32 ++++ 14 files changed, 435 insertions(+), 15 deletions(-) create mode 100644 internal/controller/dataset/nfs_uri.go create mode 100644 internal/controller/dataset/nfs_uri_test.go create mode 100644 internal/controller/dataset/nfs_version_test.go create mode 100644 pkg/nfsconfig/version.go create mode 100644 pkg/nfsconfig/version_test.go diff --git a/README.md b/README.md index dd9fa72..7f30904 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,12 @@ enable_cascading_deletion: true For `NFS` datasets, the controller sets the NFS mount option `nfsvers` when it creates a PersistentVolume. The supported versions are `3`, `4.0`, `4.1`, and `4.2`; the default is `4.1`. +An explicit port in the NFS URI (for example, `nfs://server:20490/export`) +sets only the NFS service port. NFSv3 still requires rpcbind, mountd, and +other auxiliary services to be correctly configured and reachable. The URI +does not support specifying their ports separately. When only a single +custom port is exposed, NFSv4 (4.0, 4.1, or 4.2) is recommended. + Set the version in the controller configuration file: ```yaml @@ -54,3 +60,16 @@ The `DATASET_NFS_VERSION` environment variable takes precedence over the configu config: dataset_nfs_version: "4.0" ``` + +Set a version for an individual NFS dataset to override the controller default: + +```yaml +spec: + source: + type: NFS + uri: nfs://server/export + options: + nfsVersion: "3" +``` + +Omit `nfsVersion` to use the controller default. An explicit version cannot be added, changed, or removed after creation. Changing the controller default does not update existing PVs. External PVCs (`volumeClaimRef`) use their existing PV options and cannot specify `nfsVersion`. Reference datasets inherit their source PV mount options. diff --git a/api/dataset/v1alpha1/dataset_types.go b/api/dataset/v1alpha1/dataset_types.go index 9c4c5c8..3995431 100644 --- a/api/dataset/v1alpha1/dataset_types.go +++ b/api/dataset/v1alpha1/dataset_types.go @@ -61,6 +61,8 @@ const ( AccessModeReadWrite AccessMode = "ReadWrite" ) +// +kubebuilder:validation:XValidation:rule="!has(self.options) || !('nfsVersion' in self.options) || (self.type == 'NFS' && self.options['nfsVersion'] in ['3', '4.0', '4.1', '4.2'])",message="nfsVersion must be 3, 4.0, 4.1 or 4.2 and is only supported for NFS datasets" +// +kubebuilder:validation:XValidation:rule="(has(self.options) && ('nfsVersion' in self.options)) ? (has(oldSelf.options) && ('nfsVersion' in oldSelf.options) && self.options['nfsVersion'] == oldSelf.options['nfsVersion']) : !(has(oldSelf.options) && ('nfsVersion' in oldSelf.options))",message="nfsVersion cannot be added, changed or removed after creation" type DatasetSource struct { // +kubebuilder:validation:Enum=GIT;S3;HTTP;PVC;NFS;CONDA;REFERENCE;HUGGING_FACE;MODEL_SCOPE;DATABASE;HADOOP;MANUAL // +kubebuilder:validation:XValidation:rule="self == oldSelf",message="Value is immutable" @@ -93,7 +95,7 @@ type DatasetSource struct { // CA PEM and the internal loader path tlsCAFile must not be stored in options. // - HTTP: any key-value pair will be passed to the underlying http client as http headers // - PVC: - // - NFS: + // - NFS: nfsVersion: 3, 4.0, 4.1 or 4.2. Omit to use the controller default. Immutable after creation. // - CONDA: requirements.txt, environment.yaml // - REFERENCE: // - HUGGING_FACE: repo, repoType, endpoint, include, exclude, revision @@ -131,6 +133,7 @@ type MountOptions struct { // DatasetSpec defines the desired state of Dataset // +kubebuilder:validation:XValidation:rule="!has(self.shareAccess) || size(self.shareAccess.rules) > 0",message="shareAccess.rules must contain at least one rule when shareAccess is configured" // +kubebuilder:validation:XValidation:rule="has(self.volumeClaimRef) || self.source.type != 'PVC' || (self.source.uri.matches('^pvc://[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?([.][a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?)*(/[^?#]*)?$') && size(self.source.uri.split('/')[2]) <= 253)",message="PVC dataset source URI must be pvc://[/] when volumeClaimRef is not set" +// +kubebuilder:validation:XValidation:rule="!has(self.volumeClaimRef) || !has(self.source.options) || !('nfsVersion' in self.source.options)",message="nfsVersion cannot be set when volumeClaimRef is used" type DatasetSpec struct { // Share indicates whether the model is shareable with others. // When set to true, the model can be shared according to the specified selector. diff --git a/api/dataset/v1alpha1/dataset_types_envtest_test.go b/api/dataset/v1alpha1/dataset_types_envtest_test.go index 3c867ec..2d2afc3 100644 --- a/api/dataset/v1alpha1/dataset_types_envtest_test.go +++ b/api/dataset/v1alpha1/dataset_types_envtest_test.go @@ -2,6 +2,7 @@ package v1alpha1_test import ( "context" + "fmt" "os" "path/filepath" goruntime "runtime" @@ -117,4 +118,51 @@ func TestShareAccessCRDValidation(t *testing.T) { immutable.Spec.ShareAccess.Rules[0].AccessMode = datasetv1alpha1.AccessModeReadWrite require.NoError(t, apiClient.Update(ctx, immutable)) }) + + t.Run("NFS version schema and transitions", func(t *testing.T) { + for index, options := range []map[string]string{nil, {}, {"other": "value"}, {"nfsVersion": "3"}, {"nfsVersion": "4.0"}, {"nfsVersion": "4.1"}, {"nfsVersion": "4.2"}} { + ds := newDataset(fmt.Sprintf("nfs-version-%d", index)) + ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: options} + require.NoError(t, apiClient.Create(ctx, ds)) + ds.Annotations = map[string]string{"description": "updated"} + require.NoError(t, apiClient.Update(ctx, ds)) + unchanged := ds.DeepCopy() + if unchanged.Spec.Source.Options == nil { + unchanged.Spec.Source.Options = map[string]string{} + } + unchanged.Spec.Source.Options["other"] = "updated" + require.NoError(t, apiClient.Update(ctx, unchanged)) + delete(unchanged.Spec.Source.Options, "other") + require.NoError(t, apiClient.Update(ctx, unchanged)) + changed := unchanged.DeepCopy() + if changed.Spec.Source.Options == nil { + changed.Spec.Source.Options = map[string]string{} + } + changed.Spec.Source.Options["nfsVersion"] = "3" + if options["nfsVersion"] == "3" { + changed.Spec.Source.Options["nfsVersion"] = "4.1" + } + require.Error(t, apiClient.Update(ctx, changed)) + if _, present := options["nfsVersion"]; present { + removed := unchanged.DeepCopy() + delete(removed.Spec.Source.Options, "nfsVersion") + require.Error(t, apiClient.Update(ctx, removed)) + removed.Spec.Source.Options = nil + require.Error(t, apiClient.Update(ctx, removed)) + } + } + for index, version := range []string{"", "4", "5", "auto"} { + ds := newDataset(fmt.Sprintf("nfs-invalid-%d", index)) + ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: map[string]string{"nfsVersion": version}} + require.Error(t, apiClient.Create(ctx, ds)) + } + ds := newDataset("nfs-version-wrong-type") + ds.Spec.Source.Options = map[string]string{"nfsVersion": "3"} + require.Error(t, apiClient.Create(ctx, ds)) + ds = newDataset("nfs-version-external") + ds.Spec.Source = datasetv1alpha1.DatasetSource{Type: datasetv1alpha1.DatasetTypeNFS, URI: "nfs://server/export", Options: map[string]string{"nfsVersion": "3"}} + ds.Spec.VolumeClaimRef = &datasetv1alpha1.VolumeClaimRef{Name: "external"} + require.Error(t, apiClient.Create(ctx, ds)) + }) + } diff --git a/config/config.go b/config/config.go index d1528aa..71b09d2 100644 --- a/config/config.go +++ b/config/config.go @@ -1,7 +1,7 @@ package config import ( - "fmt" + "github.com/BaizeAI/dataset/pkg/nfsconfig" "os" "strings" @@ -14,8 +14,8 @@ var ( ) const ( - defaultDatasetNFSVersion = "4.1" - datasetNFSVersionEnv = "DATASET_NFS_VERSION" + defaultDatasetNFSVersion = nfsconfig.DefaultVersion + datasetNFSVersionEnv = nfsconfig.EnvironmentKey ) type configuration struct { @@ -25,12 +25,7 @@ type configuration struct { } func validateDatasetNFSVersion(version string) error { - switch version { - case "3", "4.0", "4.1", "4.2": - return nil - default: - return fmt.Errorf("unsupported dataset NFS version %q, must be one of: 3, 4.0, 4.1, 4.2", version) - } + return nfsconfig.Validate(version) } // GetDatasetNFSVersion returns the NFS protocol version used for newly-created NFS PVs. diff --git a/config/crd/bases/dataset.baizeai.io_datasets.yaml b/config/crd/bases/dataset.baizeai.io_datasets.yaml index 4408c9a..8c2724d 100644 --- a/config/crd/bases/dataset.baizeai.io_datasets.yaml +++ b/config/crd/bases/dataset.baizeai.io_datasets.yaml @@ -303,7 +303,7 @@ spec: CA PEM and the internal loader path tlsCAFile must not be stored in options. - HTTP: any key-value pair will be passed to the underlying http client as http headers - PVC: - - NFS: + - NFS: nfsVersion: 3, 4.0, 4.1 or 4.2. Omit to use the controller default. Immutable after creation. - CONDA: requirements.txt, environment.yaml - REFERENCE: - HUGGING_FACE: repo, repoType, endpoint, include, exclude, revision @@ -356,6 +356,17 @@ spec: - type - uri type: object + x-kubernetes-validations: + - message: nfsVersion must be 3, 4.0, 4.1 or 4.2 and is only supported + for NFS datasets + rule: '!has(self.options) || !(''nfsVersion'' in self.options) || + (self.type == ''NFS'' && self.options[''nfsVersion''] in [''3'', + ''4.0'', ''4.1'', ''4.2''])' + - message: nfsVersion cannot be added, changed or removed after creation + rule: '(has(self.options) && (''nfsVersion'' in self.options)) ? + (has(oldSelf.options) && (''nfsVersion'' in oldSelf.options) && + self.options[''nfsVersion''] == oldSelf.options[''nfsVersion'']) + : !(has(oldSelf.options) && (''nfsVersion'' in oldSelf.options))' volumeClaimRef: description: volumeClaimRef is the reference to an existing PVC. properties: @@ -799,6 +810,9 @@ spec: when volumeClaimRef is not set rule: has(self.volumeClaimRef) || self.source.type != 'PVC' || (self.source.uri.matches('^pvc://[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?([.][a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?)*(/[^?#]*)?$') && size(self.source.uri.split('/')[2]) <= 253) + - message: nfsVersion cannot be set when volumeClaimRef is used + rule: '!has(self.volumeClaimRef) || !has(self.source.options) || !(''nfsVersion'' + in self.source.options)' status: description: DatasetStatus defines the observed state of Dataset properties: diff --git a/config/sample-config.yaml b/config/sample-config.yaml index ebbd71a..dada5e6 100644 --- a/config/sample-config.yaml +++ b/config/sample-config.yaml @@ -8,7 +8,7 @@ # Default: false (disabled for safety) enable_cascading_deletion: false -# NFS protocol version used when creating new NFS PVs (supported: 3, 4.0, 4.1, 4.2; default: 4.1). +# Default NFS version for new PVs when source.options.nfsVersion is omitted (supported: 3, 4.0, 4.1, 4.2; default: 4.1). # DATASET_NFS_VERSION takes precedence over this value. # dataset_nfs_version: "4.1" diff --git a/internal/controller/dataset/dataset_controller.go b/internal/controller/dataset/dataset_controller.go index 77506d5..a1543b9 100644 --- a/internal/controller/dataset/dataset_controller.go +++ b/internal/controller/dataset/dataset_controller.go @@ -27,6 +27,7 @@ import ( "github.com/BaizeAI/dataset/pkg/kubeutils" "github.com/BaizeAI/dataset/pkg/mountpolicy" + "github.com/BaizeAI/dataset/pkg/nfsconfig" "github.com/samber/lo" batchv1 "k8s.io/api/batch/v1" @@ -458,11 +459,22 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha if err != nil { return err } - pvTemp.Spec.MountOptions = []string{fmt.Sprintf("nfsvers=%s", config.GetDatasetNFSVersion())} + version, err := nfsconfig.Resolve(ds.Spec.Source.Options, config.GetDatasetNFSVersion()) + if err != nil { + return err + } + pvTemp.Spec.MountOptions = []string{fmt.Sprintf("nfsvers=%s", version)} u, err := url.Parse(ds.Spec.Source.URI) if err != nil { return err } + server, port, err := nfsServerAndPort(u) + if err != nil { + return err + } + if port != "" { + pvTemp.Spec.MountOptions = append(pvTemp.Spec.MountOptions, "port="+port) + } forceStorageClass = pvTemp.Spec.StorageClassName pv := &corev1.PersistentVolume{} @@ -488,7 +500,7 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha if pvTemp.Spec.CSI.VolumeAttributes == nil { pvTemp.Spec.CSI.VolumeAttributes = make(map[string]string) } - pvTemp.Spec.CSI.VolumeAttributes["server"] = u.Host + pvTemp.Spec.CSI.VolumeAttributes["server"] = server pvTemp.Spec.CSI.VolumeAttributes["share"] = "/" pvTemp.Spec.CSI.VolumeAttributes["subdir"] = u.Path pvTemp.Spec.CSI.VolumeAttributes["onDelete"] = "retain" @@ -500,6 +512,8 @@ func (r *DatasetReconciler) reconcilePVC(ctx context.Context, ds *datasetv1alpha if pvTemp.Spec.CSI.VolumeAttributes["mountPermissions"] == "" { pvTemp.Spec.CSI.VolumeAttributes["mountPermissions"] = ds.Spec.MountOptions.Mode } + // Preserve the existing volume identity, including the URI port. + // The mount server and NFS port are supplied separately above. pvTemp.Spec.CSI.VolumeHandle = fmt.Sprintf("%s#%s#%s#", u.Host, u.Path, pvName) if err := r.Create(ctx, &pvTemp); err != nil { @@ -1171,6 +1185,12 @@ func (r *DatasetReconciler) getSourceDataset(ctx context.Context, ds *datasetv1a } func (r *DatasetReconciler) validate(ctx context.Context, ds *datasetv1alpha1.Dataset) error { + if err := nfsconfig.ValidateOptions(string(ds.Spec.Source.Type), ds.Spec.Source.Options); err != nil { + return err + } + if _, present := ds.Spec.Source.Options[nfsconfig.VersionKey]; present && ds.Spec.VolumeClaimRef != nil { + return fmt.Errorf("nfsVersion cannot be set when volumeClaimRef is used") + } if ds.Spec.Source.Type == datasetv1alpha1.DatasetTypeManual && ds.Spec.Source.URI != "manual://" { return fmt.Errorf("MANUAL dataset source URI must be manual://") } diff --git a/internal/controller/dataset/nfs_uri.go b/internal/controller/dataset/nfs_uri.go new file mode 100644 index 0000000..8c0ec16 --- /dev/null +++ b/internal/controller/dataset/nfs_uri.go @@ -0,0 +1,38 @@ +package dataset + +import ( + "fmt" + "net/netip" + "net/url" + "strconv" + "strings" +) + +// nfsServerAndPort separates the mount server from the NFS service port. +// Keep the original host spelling (including IPv6 brackets) for compatibility. +func nfsServerAndPort(u *url.URL) (string, string, error) { + server := u.Host + if u.Hostname() == "" { + return "", "", fmt.Errorf("NFS URI must include a server") + } + // url.URL.Port treats the final numeric segment of an unbracketed IPv6 + // address as a port. Preserve legacy address-only URIs before splitting. + if addr, err := netip.ParseAddr(server); err == nil && addr.Is6() { + return server, "", nil + } + if strings.Count(server, ":") > 1 && !strings.HasPrefix(server, "[") { + return "", "", fmt.Errorf("invalid NFS server %q: IPv6 addresses with a port must be bracketed", server) + } + port := u.Port() + if port == "" { + if strings.HasSuffix(server, ":") { + return "", "", fmt.Errorf("invalid NFS port: port must not be empty") + } + return server, "", nil + } + number, err := strconv.ParseUint(port, 10, 16) + if err != nil || number == 0 { + return "", "", fmt.Errorf("invalid NFS port %q: must be between 1 and 65535", port) + } + return strings.TrimSuffix(server, ":"+port), strconv.FormatUint(number, 10), nil +} diff --git a/internal/controller/dataset/nfs_uri_test.go b/internal/controller/dataset/nfs_uri_test.go new file mode 100644 index 0000000..beae696 --- /dev/null +++ b/internal/controller/dataset/nfs_uri_test.go @@ -0,0 +1,149 @@ +package dataset + +import ( + "context" + "testing" + + api "github.com/BaizeAI/dataset/api/dataset/v1alpha1" + "github.com/BaizeAI/dataset/config" + "github.com/BaizeAI/dataset/internal/pkg/constants" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func TestReconcilePVCNFSURI(t *testing.T) { + tests := []struct { + name, uri, server, port, host, path, version string + }{ + {name: "reported IPv4 port", uri: "nfs://10.0.6.23:20490/exports", server: "10.0.6.23", port: "20490", host: "10.0.6.23:20490", path: "/exports", version: "4.0"}, + {name: "IPv4 default", uri: "nfs://192.0.2.1/exports", server: "192.0.2.1", host: "192.0.2.1", path: "/exports"}, + {name: "DNS default", uri: "nfs://nfs.example.test/exports", server: "nfs.example.test", host: "nfs.example.test", path: "/exports"}, + {name: "DNS port NFS 3", uri: "nfs://nfs.example.test:32049/exports", server: "nfs.example.test", port: "32049", host: "nfs.example.test:32049", path: "/exports", version: "3"}, + {name: "explicit standard port NFS 4.1", uri: "nfs://nfs.example.test:2049/exports", server: "nfs.example.test", port: "2049", host: "nfs.example.test:2049", path: "/exports", version: "4.1"}, + {name: "maximum port NFS 4.2", uri: "nfs://192.0.2.1:65535/exports", server: "192.0.2.1", port: "65535", host: "192.0.2.1:65535", path: "/exports", version: "4.2"}, + {name: "minimum port", uri: "nfs://nfs.example.test:1/exports", server: "nfs.example.test", port: "1", host: "nfs.example.test:1", path: "/exports"}, + {name: "decimal port with leading zeros", uri: "nfs://nfs.example.test:032049/exports", server: "nfs.example.test", port: "32049", host: "nfs.example.test:032049", path: "/exports"}, + {name: "IPv6 zone default", uri: "nfs://[fe80::1%25eth0]/exports", server: "[fe80::1%eth0]", host: "[fe80::1%eth0]", path: "/exports"}, + {name: "IPv6 zone port", uri: "nfs://[fe80::1%25eth0]:32049/exports", server: "[fe80::1%eth0]", port: "32049", host: "[fe80::1%eth0]:32049", path: "/exports"}, + {name: "IPv6 default", uri: "nfs://[2001:db8::1]/exports", server: "[2001:db8::1]", host: "[2001:db8::1]", path: "/exports"}, + {name: "IPv6 port", uri: "nfs://[2001:db8::1]:32049/exports", server: "[2001:db8::1]", port: "32049", host: "[2001:db8::1]:32049", path: "/exports"}, + {name: "legacy unbracketed IPv6", uri: "nfs://2001:db8::1/exports", server: "2001:db8::1", host: "2001:db8::1", path: "/exports"}, + {name: "legacy IPv6 numeric suffix is not a port", uri: "nfs://2001:db8::2049/exports", server: "2001:db8::2049", host: "2001:db8::2049", path: "/exports"}, + {name: "nested directory", uri: "nfs://nfs.example.test:32049/exports/data/", server: "nfs.example.test", port: "32049", host: "nfs.example.test:32049", path: "/exports/data/"}, + {name: "escaped directory", uri: "nfs://nfs.example.test:32049/exports/my%20data", server: "nfs.example.test", port: "32049", host: "nfs.example.test:32049", path: "/exports/my data"}, + {name: "root", uri: "nfs://nfs.example.test:32049/", server: "nfs.example.test", port: "32049", host: "nfs.example.test:32049", path: "/"}, + {name: "empty directory", uri: "nfs://nfs.example.test", server: "nfs.example.test", host: "nfs.example.test"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r, ds := newNFSURIReconciler(t, tt.uri) + version := "4.0" + if tt.version != "" { + ds.Spec.Source.Options = map[string]string{"nfsVersion": tt.version} + version = tt.version + } + require.NoError(t, r.reconcilePVC(context.Background(), ds)) + pv := &corev1.PersistentVolume{} + require.NoError(t, r.Get(context.Background(), client.ObjectKey{Name: "dataset-test-pvc-nfs"}, pv)) + options := []string{"nfsvers=" + version} + if tt.port != "" { + options = append(options, "port="+tt.port) + } + // A URI's NFS port must not imply mountd, lockd, or other service + // options, even for NFS 3. + require.Equal(t, options, pv.Spec.MountOptions) + require.Equal(t, tt.server, pv.Spec.CSI.VolumeAttributes["server"]) + require.Equal(t, "/", pv.Spec.CSI.VolumeAttributes["share"]) + require.Equal(t, tt.path, pv.Spec.CSI.VolumeAttributes["subdir"]) + require.Equal(t, "retain", pv.Spec.CSI.VolumeAttributes["onDelete"]) + require.Equal(t, tt.host+"#"+tt.path+"#"+pv.Name+"#", pv.Spec.CSI.VolumeHandle) + pvc := &corev1.PersistentVolumeClaim{} + require.NoError(t, r.Get(context.Background(), client.ObjectKeyFromObject(ds), pvc)) + require.Equal(t, pv.Name, pvc.Spec.VolumeName) + originalSpec := pv.Spec.DeepCopy() + require.NoError(t, r.reconcilePVC(context.Background(), ds)) + require.NoError(t, r.Get(context.Background(), client.ObjectKeyFromObject(pv), pv)) + require.Equal(t, *originalSpec, pv.Spec) + }) + } +} + +func TestReconcilePVCNFSURIInvalid(t *testing.T) { + for _, uri := range []string{ + "nfs://server:invalid/exports", + "nfs://server:-1/exports", + "nfs://server:0/exports", + "nfs://server:65536/exports", + "nfs://server:999999999999999999999999/exports", + "nfs://server:/exports", + "nfs://[2001:db8::1]:0/exports", + "nfs://[2001:db8::1]:65536/exports", + "nfs://[2001:db8::1]:/exports", + "nfs://server:123:456/exports", + "nfs:///exports", + "nfs://:32049/exports", + "nfs://[]/exports", + "%", + } { + t.Run(uri, func(t *testing.T) { + r, ds := newNFSURIReconciler(t, uri) + require.Error(t, r.reconcilePVC(context.Background(), ds)) + pvs := &corev1.PersistentVolumeList{} + require.NoError(t, r.List(context.Background(), pvs)) + require.Empty(t, pvs.Items) + pvcs := &corev1.PersistentVolumeClaimList{} + require.NoError(t, r.List(context.Background(), pvcs)) + require.Empty(t, pvcs.Items) + }) + } +} + +func TestReconcilePVCNFSURIDoesNotRewriteExistingPV(t *testing.T) { + r, ds := newNFSURIReconciler(t, "nfs://10.0.6.23:20490/exports") + // Correcting an old PV's immutable CSI source requires an explicit + // replacement by the operator. + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{ + Name: "dataset-test-pvc-nfs", + Labels: map[string]string{constants.DatasetNameLabel: ds.Name}, + }, + Spec: corev1.PersistentVolumeSpec{ + MountOptions: []string{"nfsvers=4.0"}, + PersistentVolumeSource: corev1.PersistentVolumeSource{ + CSI: &corev1.CSIPersistentVolumeSource{ + Driver: "nfs.csi.k8s.io", + VolumeHandle: "10.0.6.23:20490#/exports#dataset-test-pvc-nfs#", + VolumeAttributes: map[string]string{ + "server": "10.0.6.23:20490", "share": "/", "subdir": "/exports", + }, + }, + }, + }, + } + require.NoError(t, r.Create(context.Background(), pv)) + originalSpec := pv.Spec.DeepCopy() + require.NoError(t, r.reconcilePVC(context.Background(), ds)) + stored := &corev1.PersistentVolume{} + require.NoError(t, r.Get(context.Background(), client.ObjectKeyFromObject(pv), stored)) + require.Equal(t, *originalSpec, stored.Spec) +} + +func newNFSURIReconciler(t *testing.T, uri string) (*DatasetReconciler, *api.Dataset) { + t.Helper() + t.Setenv("DATASET_NFS_VERSION", "4.0") + require.NoError(t, config.ParseConfigFromFileContent("")) + scheme := runtime.NewScheme() + require.NoError(t, api.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + ds := &api.Dataset{ + ObjectMeta: metav1.ObjectMeta{Name: "nfs", Namespace: "test"}, + Spec: api.DatasetSpec{Source: api.DatasetSource{ + Type: api.DatasetTypeNFS, URI: uri, + }}, + } + return &DatasetReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).Build(), Scheme: scheme}, ds +} diff --git a/internal/controller/dataset/nfs_version_test.go b/internal/controller/dataset/nfs_version_test.go new file mode 100644 index 0000000..394f2d9 --- /dev/null +++ b/internal/controller/dataset/nfs_version_test.go @@ -0,0 +1,56 @@ +package dataset + +import ( + "context" + api "github.com/BaizeAI/dataset/api/dataset/v1alpha1" + "github.com/BaizeAI/dataset/config" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "testing" +) + +func TestNFSVersionPerDataset(t *testing.T) { + for _, version := range []string{"3", "4.0", "4.1", "4.2", "", "5"} { + t.Run(version, func(t *testing.T) { + t.Setenv("DATASET_NFS_VERSION", "4.2") + require.NoError(t, config.ParseConfigFromFileContent("")) + scheme := runtime.NewScheme() + require.NoError(t, api.AddToScheme(scheme)) + require.NoError(t, corev1.AddToScheme(scheme)) + ds := &api.Dataset{ObjectMeta: metav1.ObjectMeta{Name: "nfs", Namespace: "test"}, Spec: api.DatasetSpec{Source: api.DatasetSource{Type: api.DatasetTypeNFS, URI: "nfs://server/export", Options: map[string]string{"nfsVersion": version}}}} + cli := fake.NewClientBuilder().WithScheme(scheme).Build() + controller := &DatasetReconciler{Client: cli, Scheme: scheme} + err := controller.reconcilePVC(context.Background(), ds) + if version == "" || version == "5" { + require.Error(t, err) + pvs := &corev1.PersistentVolumeList{} + require.NoError(t, cli.List(context.Background(), pvs)) + require.Empty(t, pvs.Items) + return + } + require.NoError(t, err) + pv := &corev1.PersistentVolume{} + require.NoError(t, cli.Get(context.Background(), client.ObjectKey{Name: "dataset-test-pvc-nfs"}, pv)) + require.Equal(t, []string{"nfsvers=" + version}, pv.Spec.MountOptions) + t.Setenv("DATASET_NFS_VERSION", "4.0") + require.NoError(t, config.ParseConfigFromFileContent("")) + require.NoError(t, controller.reconcilePVC(context.Background(), ds)) + require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pv), pv)) + require.Equal(t, []string{"nfsvers=" + version}, pv.Spec.MountOptions) + }) + } +} + +func TestNFSVersionDefensiveValidation(t *testing.T) { + r := &DatasetReconciler{} + for _, sourceType := range []api.DatasetType{api.DatasetTypeS3, api.DatasetTypeReference, api.DatasetTypePVC} { + ds := &api.Dataset{Spec: api.DatasetSpec{Source: api.DatasetSource{Type: sourceType, Options: map[string]string{"nfsVersion": "3"}}}} + require.ErrorContains(t, r.validate(context.Background(), ds), "only supported for NFS") + } + ds := &api.Dataset{Spec: api.DatasetSpec{Source: api.DatasetSource{Type: api.DatasetTypeNFS, Options: map[string]string{"nfsVersion": "3"}}, VolumeClaimRef: &api.VolumeClaimRef{Name: "external"}}} + require.ErrorContains(t, r.validate(context.Background(), ds), "volumeClaimRef") +} diff --git a/manifests/dataset/templates/deployment.yaml b/manifests/dataset/templates/deployment.yaml index b0f132d..047f16f 100644 --- a/manifests/dataset/templates/deployment.yaml +++ b/manifests/dataset/templates/deployment.yaml @@ -3,6 +3,7 @@ kind: Deployment metadata: name: {{ include "dataset.fullname" . }} labels: + dataset.baizeai.io/controller: "true" {{- include "dataset.labels" . | nindent 4 }} spec: replicas: {{ .Values.replicaCount }} diff --git a/manifests/dataset/values.yaml b/manifests/dataset/values.yaml index 4baea66..d074c88 100644 --- a/manifests/dataset/values.yaml +++ b/manifests/dataset/values.yaml @@ -10,7 +10,7 @@ global: config: # A custom loader image must implement the same S3 TLS contract as the controller. dataset_job_spec: {} - # NFS protocol version used when creating new NFS PVs. Supported values are + # Default NFS version for new PVs when source.options.nfsVersion is omitted. Supported values are # 3, 4.0, 4.1, and 4.2. This value is passed to the controller through # DATASET_NFS_VERSION. dataset_nfs_version: "4.1" diff --git a/pkg/nfsconfig/version.go b/pkg/nfsconfig/version.go new file mode 100644 index 0000000..4049ff2 --- /dev/null +++ b/pkg/nfsconfig/version.go @@ -0,0 +1,45 @@ +// Package nfsconfig defines the NFS version contract shared by Dataset and Baize. +package nfsconfig + +import "fmt" + +const ( + DefaultVersion = "4.1" + VersionKey = "nfsVersion" + EnvironmentKey = "DATASET_NFS_VERSION" + ConfigKey = "dataset_nfs_version" +) + +// Validate accepts an explicit protocol version. Callers handle absent values. +func Validate(version string) error { + switch version { + case "3", "4.0", "4.1", "4.2": + return nil + default: + return fmt.Errorf("unsupported dataset NFS version %q, must be one of: 3, 4.0, 4.1, 4.2", version) + } +} + +// ValidateOptions distinguishes an absent version key from an explicitly empty one. +func ValidateOptions(sourceType string, options map[string]string) error { + version, present := options[VersionKey] + if !present { + return nil + } + if sourceType != "NFS" { + return fmt.Errorf("%s is only supported for NFS datasets", VersionKey) + } + return Validate(version) +} + +// Resolve applies the controller default only when the version key is absent. +func Resolve(options map[string]string, fallback string) (string, error) { + version, present := options[VersionKey] + if !present { + version = fallback + } + if err := Validate(version); err != nil { + return "", err + } + return version, nil +} diff --git a/pkg/nfsconfig/version_test.go b/pkg/nfsconfig/version_test.go new file mode 100644 index 0000000..edc663f --- /dev/null +++ b/pkg/nfsconfig/version_test.go @@ -0,0 +1,32 @@ +package nfsconfig + +import ( + "github.com/stretchr/testify/require" + "testing" +) + +func TestVersionContract(t *testing.T) { + for _, version := range []string{"3", "4.0", "4.1", "4.2"} { + t.Run(version, func(t *testing.T) { + options := map[string]string{VersionKey: version} + require.NoError(t, ValidateOptions("NFS", options)) + actual, err := Resolve(options, "4.2") + require.NoError(t, err) + require.Equal(t, version, actual) + require.Error(t, ValidateOptions("REFERENCE", options)) + }) + } + for _, version := range []string{"", "4", "5", "auto", " 4.1", "4.1 "} { + require.Error(t, ValidateOptions("NFS", map[string]string{VersionKey: version})) + _, err := Resolve(map[string]string{VersionKey: version}, DefaultVersion) + require.Error(t, err) + } + for _, options := range []map[string]string{nil, {}, {"other": "value"}} { + require.NoError(t, ValidateOptions("REFERENCE", options)) + actual, err := Resolve(options, "3") + require.NoError(t, err) + require.Equal(t, "3", actual) + } + _, err := Resolve(nil, "auto") + require.Error(t, err) +}