diff --git a/seaweed-vector/crates/core/src/main.rs b/seaweed-vector/crates/core/src/main.rs index 694b565ebe..687320d5ce 100644 --- a/seaweed-vector/crates/core/src/main.rs +++ b/seaweed-vector/crates/core/src/main.rs @@ -39,15 +39,6 @@ struct Args { )] filer_group: String, - /// Encrypt every chunk written for the volume servers, as `weed s3 - /// -encryptVolumeData` does. - #[arg( - long = "encryptVolumeData", - alias = "encrypt-volume-data", - env = "WEED_ENCRYPT_VOLUME_DATA" - )] - encrypt_volume_data: bool, - /// The `jwt.filer_signing.key` of the cluster, which the namespace /// authenticates cluster processes with; security.toml's by default. #[arg(long, env = "WEED_FILER_SIGNING_KEY")] @@ -208,9 +199,7 @@ async fn main() -> Result<()> { Some(masters) => { let cluster = weed_lance_worker::cluster::Cluster::new(masters, &args.filer_group, &security)?; - Storage(Some( - cluster.storage(&security, args.encrypt_volume_data).await?, - )) + Storage(Some(cluster.storage(&security).await?)) } None => Storage::default(), }; @@ -408,14 +397,16 @@ mod tests { "m:9333", "--filer-group", "eu", - "--encryptVolumeData", "describe", "t", ]) .unwrap(); assert_eq!(args.masters.as_deref(), Some("m:9333")); assert_eq!(args.filer_group, "eu"); - assert!(args.encrypt_volume_data); + assert!( + Args::try_parse_from(["weed-vector", "--encryptVolumeData", "describe", "t"]).is_err(), + "volume encryption follows the filer" + ); } #[test] diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index f244beb7c3..1144bfcd84 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -23,14 +23,6 @@ struct Args { /// through. #[arg(long, alias = "master", env = "WEED_VECTOR_MASTERS")] masters: String, - /// Encrypt every chunk written for the volume servers, as `weed s3 - /// -encryptVolumeData` does. - #[arg( - long = "encryptVolumeData", - alias = "encrypt-volume-data", - env = "WEED_VECTOR_ENCRYPT_VOLUME_DATA" - )] - encrypt_volume_data: bool, /// The filer group whose filers the engine keeps tables through. #[arg( long, @@ -135,7 +127,7 @@ async fn main() -> Result<()> { .await .with_context(|| format!("listen on {}", args.listen))?; let cluster = Cluster::new(&args.masters, &args.filer_group, &security)?; - let reads = cluster.storage(&security, args.encrypt_volume_data).await?; + let reads = cluster.storage(&security).await?; let mut service = VectorServer::new() .with_nprobes(args.nprobes.map(std::num::NonZeroUsize::get)) .with_storage(reads); diff --git a/seaweed-worker/README.md b/seaweed-worker/README.md index c92c89b3ba..c8b716f2c1 100644 --- a/seaweed-worker/README.md +++ b/seaweed-worker/README.md @@ -95,8 +95,9 @@ which `--masters` names the masters of (`--filer-group` picks the filers). What that takes comes from the same security.toml the weed processes use: `grpc.client` for the filers, `https.client` for volume servers under HTTPS, `jwt.signing.read.key` where volume reads are signed, and `s3.sse.kek` or -`s3.sse.key` where a bucket encrypts with SSE-S3. `--encryptVolumeData` encrypts -every chunk the worker writes for the volume servers, as the gateway's flag does. +`s3.sse.key` where a bucket encrypts with SSE-S3. Chunks are encrypted for the +volume servers when the filer is (`weed filer -encryptVolumeData`), as the S3 +gateway's are. The namespace authenticates the worker by the cluster token it derives from `jwt.filer_signing.key`; the same key signs the admin stream when admin requires diff --git a/seaweed-worker/crates/lance/src/cluster.rs b/seaweed-worker/crates/lance/src/cluster.rs index aa8f79ad99..f8978f5ae8 100644 --- a/seaweed-worker/crates/lance/src/cluster.rs +++ b/seaweed-worker/crates/lance/src/cluster.rs @@ -221,11 +221,9 @@ impl Cluster { } /// Connects to the cluster's storage and follows the masters' filer list. - pub async fn storage( - &self, - security: &Security, - encrypt_volume_data: bool, - ) -> Result { + /// Chunks are encrypted for the volume servers when the filer is + /// configured to (`weed filer -encryptVolumeData`), as `weed s3` does. + pub async fn storage(&self, security: &Security) -> Result { // The filers are dialed as the masters are: grpc.client's credentials // and server allowlist, through this crate's own connector. let filer_dialer: Option = security @@ -246,7 +244,6 @@ impl Cluster { grpc_tls: None, https_tls: security.https_client_tls()?, filer_dialer, - encrypt_volume_data, filer_group: self.filer_group.clone(), }; let storage = DirectReads::connect(&conn) diff --git a/seaweed-worker/crates/lance/src/direct_read.rs b/seaweed-worker/crates/lance/src/direct_read.rs index b1f09ea341..4019ff9173 100644 --- a/seaweed-worker/crates/lance/src/direct_read.rs +++ b/seaweed-worker/crates/lance/src/direct_read.rs @@ -472,9 +472,6 @@ pub struct Connection { /// Dials the filers in place of `grpc_tls`, for a TLS setup sw-client /// cannot express, such as a server allowlist. pub filer_dialer: Option, - /// Encrypt every chunk written for the volume servers, as `weed s3 - /// -encryptVolumeData` does; the filer's own setting turns it on too. - pub encrypt_volume_data: bool, /// The filer group, which the gateway prefixes a bucket's collection with. pub filer_group: String, } @@ -488,7 +485,6 @@ impl fmt::Debug for Connection { .field("grpc_tls", &self.grpc_tls) .field("https_tls", &self.https_tls) .field("filer_dialer", &self.filer_dialer.is_some()) - .field("encrypt_volume_data", &self.encrypt_volume_data) .field("filer_group", &self.filer_group) .finish() } @@ -548,7 +544,7 @@ impl DirectReads { { let shared = Arc::get_mut(&mut reads.shared).expect("just created"); shared.chunk_size = chunk_size; - shared.cipher = config.cipher || conn.encrypt_volume_data; + shared.cipher = config.cipher; shared.filer_group = conn.filer_group.clone(); } reads.report_every(Duration::from_secs(30)); diff --git a/seaweed-worker/crates/lance/src/main.rs b/seaweed-worker/crates/lance/src/main.rs index 5c58141c53..66c8aff5e7 100644 --- a/seaweed-worker/crates/lance/src/main.rs +++ b/seaweed-worker/crates/lance/src/main.rs @@ -48,14 +48,6 @@ struct Args { default_value = "" )] filer_group: String, - /// Encrypt every chunk written for the volume servers, as `weed s3 - /// -encryptVolumeData` does. - #[arg( - long = "encryptVolumeData", - alias = "encrypt-volume-data", - env = "WEED_ENCRYPT_VOLUME_DATA" - )] - encrypt_volume_data: bool, /// The cluster's security.toml; by default the file weed processes find /// in -config_dir, ., ~/.seaweedfs and /etc/seaweedfs. #[arg(long, env = "WEED_WORKER_SECURITY_TOML")] @@ -195,9 +187,7 @@ async fn main() -> Result<()> { let cluster = weed_lance_worker::cluster::Cluster::new(&args.masters, &args.filer_group, &security)?; - let storage = weed_lance_worker::dataset::Storage(Some( - cluster.storage(&security, args.encrypt_volume_data).await?, - )); + let storage = weed_lance_worker::dataset::Storage(Some(cluster.storage(&security).await?)); let metrics = Metrics::new(&options.worker_id, &options.worker_version)?; let lance_metrics = LanceMetrics::new(&metrics)?; diff --git a/seaweed-worker/crates/lance/tests/common/mod.rs b/seaweed-worker/crates/lance/tests/common/mod.rs index 510c5597f7..dc623ab09d 100644 --- a/seaweed-worker/crates/lance/tests/common/mod.rs +++ b/seaweed-worker/crates/lance/tests/common/mod.rs @@ -93,9 +93,7 @@ pub async fn storage() -> weed_lance_worker::dataset::Storage { let security = security(); let cluster = weed_lance_worker::cluster::Cluster::new(&masters, "", &security).expect("masters"); - weed_lance_worker::dataset::Storage(Some( - cluster.storage(&security, false).await.expect("storage"), - )) + weed_lance_worker::dataset::Storage(Some(cluster.storage(&security).await.expect("storage"))) } fn security() -> weed_lance_worker::security::Security { diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index bdea56ce31..0311d242d5 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -301,9 +301,13 @@ def run_cluster(param, tmp_path_factory, identities): "max": 64, "preStopSeconds": 0, }.items()]) wait_for(lambda: listening(vol_grpc), processes, "volume not ready") + # VECTOR_ENCRYPT_VOLUMES=1 turns on the filer's volume encryption, which + # the gateway, the engine and the worker all follow: every chunk they + # store is encrypted, and direct reads must undo it. + encrypt = ["-encryptVolumeData"] if os.environ.get("VECTOR_ENCRYPT_VOLUMES") else [] start("filer", [weed, "filer", f"-master={master_addr}", "-ip=127.0.0.1", "-ip.bind=127.0.0.1", f"-port={filer}", f"-port.grpc={filer_grpc}", - f"-defaultStoreDir={directory / 'filer'}"]) + f"-defaultStoreDir={directory / 'filer'}", *encrypt]) wait_for(lambda: listening(filer_grpc), processes, "filer not ready") endpoint = f"http://127.0.0.1:{s3}" # weed s3 turns AWS_ACCESS_KEY_ID into an admin identity. @@ -311,10 +315,7 @@ def run_cluster(param, tmp_path_factory, identities): config, unset = [f"-config={HERE / 's3.json'}"], () else: config, unset = [], ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY") - # VECTOR_ENCRYPT_VOLUMES=1 has the gateway encrypt every chunk it - # stores, which direct reads must undo. - encrypt = ["-encryptVolumeData"] if os.environ.get("VECTOR_ENCRYPT_VOLUMES") else [] - start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", *config, *encrypt, + start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", *config, "-ip=127.0.0.1", "-ip.bind=127.0.0.1", f"-port={s3}", f"-port.grpc={s3_grpc}", "-port.iceberg=0", f"-port.lance={lance}"], unset) wait_for(lambda: http_ready(f"http://127.0.0.1:{lance}/v1/namespace/root/list"), processes, "catalog not ready") @@ -324,7 +325,7 @@ def run_cluster(param, tmp_path_factory, identities): # The engine keeps tables on the filers and volume servers itself; it # has no S3 credentials to be given, and encrypts its chunks for the # volume servers as the gateway does. - start("vector", [engine, "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *encrypt]) + start("vector", [engine, "--masters", master_addr, "--listen", f"127.0.0.1:{vector}"]) wait_for(lambda: listening(vector), processes, "vector engine not ready") if os.environ.get("VECTOR_TLS"): wait_for(lambda: listening(s3_grpc), processes, "s3 gRPC not ready") @@ -362,7 +363,7 @@ def run_cluster(param, tmp_path_factory, identities): wait_for(lambda: http_ready(f"{admin_url}/api/plugin/job-types"), processes, "admin not ready") start("lance-worker", [lance_worker, "--admin", f"127.0.0.1:{admin}.{admin_grpc}", "--namespace", f"http://127.0.0.1:{lance}", "--id", "vectors-test-worker", - "--masters", master_addr, *(["--encryptVolumeData"] if encrypt else [])]) + "--masters", master_addr]) wait_for(lambda: "lance_vector_index" in [jt["job_type"] for jt in json.load(urllib.request.urlopen(f"{admin_url}/api/plugin/job-types", timeout=2))], processes, "the Lance worker never registered with admin") diff --git a/test/s3vectors/s3vectors_test.py b/test/s3vectors/s3vectors_test.py index cf0899ff76..feda356895 100644 --- a/test/s3vectors/s3vectors_test.py +++ b/test/s3vectors/s3vectors_test.py @@ -118,8 +118,9 @@ def index(vectors): def assert_engine_wrote_like_the_gateway(cluster, index): """The engine writes the table's files itself: the filer holds entries the - gateway could have written (an ETag, chunks, no inline content), and under - VECTOR_ENCRYPT_VOLUMES every chunk carries a cipher key.""" + gateway could have written (an ETag, chunks, no inline content), and every + chunk carries a cipher key exactly when VECTOR_ENCRYPT_VOLUMES encrypts at + the filer.""" import os import urllib.request @@ -135,6 +136,8 @@ def assert_engine_wrote_like_the_gateway(cluster, index): assert "Seaweed-X-Amz-ETag" in (meta.get("Extended") or {}), meta.get("Extended") if os.environ.get("VECTOR_ENCRYPT_VOLUMES"): assert all(chunk.get("cipher_key") for chunk in meta["chunks"]), meta["chunks"] + else: + assert not any(chunk.get("cipher_key") for chunk in meta["chunks"]), meta["chunks"] def run_cli(cluster, *args): diff --git a/weed/command/filer.go b/weed/command/filer.go index f791deacc8..f91bffaa81 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -184,7 +184,7 @@ func init() { filerS3Options.concurrentUploadLimitMB = cmdFiler.Flag.Int("s3.concurrentUploadLimitMB", 0, "limit total concurrent upload size for S3, 0 means unlimited") filerS3Options.concurrentFileUploadLimit = cmdFiler.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited") filerS3Options.enableIam = cmdFiler.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port") - filerS3Options.cipher = cmdFiler.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads") + filerS3Options.cipher = cmdFiler.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads; on whenever -encryptVolumeData is") filerS3Options.iamReadOnly = cmdFiler.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server") filerS3Options.portIceberg = cmdFiler.Flag.Int("s3.port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)") filerS3Options.portLance = cmdFiler.Flag.Int("s3.port.lance", 9101, "Lance Namespace server listen port (0 to disable)") diff --git a/weed/command/mini.go b/weed/command/mini.go index 2d963f7bda..5f906cdd6c 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -550,7 +550,7 @@ func initMiniS3Flags() { miniS3Options.enableIam = cmdMini.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same port") miniS3Options.iamReadOnly = miniS3IamReadOnly miniS3Options.dataCenter = cmdMini.Flag.String("s3.dataCenter", "", "prefer to read and write to volumes in this data center") - miniS3Options.cipher = cmdMini.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads") + miniS3Options.cipher = cmdMini.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads; on whenever -filer.encryptVolumeData is") miniS3Options.config = miniS3Config miniS3Options.iamConfig = miniIamConfig miniS3Options.auditLogConfig = cmdMini.Flag.String("s3.auditLogConfig", "", "path to the audit log config file") diff --git a/weed/command/s3.go b/weed/command/s3.go index b6eb942b7d..12b293d1de 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -143,7 +143,7 @@ func init() { s3StandaloneOptions.iamReadOnly = cmdS3.Flag.Bool("iam.readOnly", true, "disable IAM write operations on this server") s3StandaloneOptions.debug = cmdS3.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port") s3StandaloneOptions.debugPort = cmdS3.Flag.Int("debug.port", 6060, "http port for debugging") - s3StandaloneOptions.cipher = cmdS3.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers") + s3StandaloneOptions.cipher = cmdS3.Flag.Bool("encryptVolumeData", false, "encrypt data on volume servers; on whenever the filer's -encryptVolumeData is") s3StandaloneOptions.externalUrl = cmdS3.Flag.String("externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") s3StandaloneOptions.defaultFileMode = cmdS3.Flag.String("defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666") s3StandaloneOptions.cacheSizeMB = cmdS3.Flag.Int64("cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)") @@ -315,6 +315,8 @@ func (s3opt *S3Options) startS3Server() bool { // chunk size for S3 uploads, read from the filer's -maxMB var filerMaxMB int32 + // the filer's -encryptVolumeData, which turns on chunk encryption here too + var filerCipher bool for { err := pb.WithOneOfGrpcFilerClients(false, filerAddresses, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { @@ -328,6 +330,7 @@ func (s3opt *S3Options) startS3Server() bool { masterAddresses = pb.ServerAddresses(strings.Join(resp.Masters, ",")).ToAddresses() metricsAddress, metricsIntervalSec = resp.MetricsAddress, int(resp.MetricsIntervalSec) filerMaxMB = int32(resp.MaxMb) + filerCipher = resp.Cipher glog.V(0).Infof("S3 read filer buckets dir: %s", filerBucketsPath) glog.V(0).Infof("S3 read filer maxMB: %d", filerMaxMB) if len(masterAddresses) > 0 { @@ -406,7 +409,7 @@ func (s3opt *S3Options) startS3Server() bool { ConcurrentFileUploadLimit: int64(*s3opt.concurrentFileUploadLimit), EnableIam: *s3opt.enableIam, // Embedded IAM API (enabled by default) IamReadOnly: *s3opt.iamReadOnly, - Cipher: *s3opt.cipher, // encrypt data on volume servers + Cipher: *s3opt.cipher || filerCipher, // encrypt data on volume servers Ip: *s3opt.ip, BindIp: *s3opt.bindIp, GrpcPort: *s3opt.portGrpc, diff --git a/weed/command/server.go b/weed/command/server.go index d1af84da6e..6dd4cb3b53 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -201,7 +201,7 @@ func init() { s3Options.concurrentFileUploadLimit = cmdServer.Flag.Int("s3.concurrentFileUploadLimit", 0, "limit number of concurrent file uploads for S3, 0 means unlimited") s3Options.enableIam = cmdServer.Flag.Bool("s3.iam", true, "enable embedded IAM API on the same S3 port") s3Options.iamReadOnly = cmdServer.Flag.Bool("s3.iam.readOnly", true, "disable IAM write operations on this server") - s3Options.cipher = cmdServer.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads") + s3Options.cipher = cmdServer.Flag.Bool("s3.encryptVolumeData", false, "encrypt data on volume servers for S3 uploads; on whenever -filer.encryptVolumeData is") s3Options.externalUrl = cmdServer.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") s3Options.allowBackslashInKey = cmdServer.Flag.Bool("s3.allowBackslashInKey", false, allowBackslashInKeyUsage) s3Options.defaultFileMode = cmdServer.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666")