diff --git a/.gitignore b/.gitignore index 183907c..c2e48fa 100755 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,4 @@ docs/* .claude/ .cursor/ .gemini/ +kubeconfig \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 201012d..0835f11 100755 --- a/Cargo.lock +++ b/Cargo.lock @@ -95,6 +95,12 @@ dependencies = [ "windows-sys 0.60.2", ] +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + [[package]] name = "arbitrary" version = "1.4.2" @@ -224,7 +230,7 @@ dependencies = [ "serde_urlencoded", "sync_wrapper", "tokio", - "tower", + "tower 0.5.2", "tower-layer", "tower-service", "tracing", @@ -654,6 +660,16 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "event-listener" version = "5.4.1" @@ -687,6 +703,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + [[package]] name = "flate2" version = "1.1.8" @@ -887,13 +909,19 @@ dependencies = [ "futures-core", "futures-sink", "http", - "indexmap", + "indexmap 2.12.0", "slab", "tokio", "tokio-util", "tracing", ] +[[package]] +name = "hashbrown" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" + [[package]] name = "hashbrown" version = "0.15.5" @@ -1077,7 +1105,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2", + "socket2 0.6.1", "tokio", "tower-service", "tracing", @@ -1215,6 +1243,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "indexmap" +version = "1.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" +dependencies = [ + "autocfg", + "hashbrown 0.12.3", +] + [[package]] name = "indexmap" version = "2.12.0" @@ -1255,6 +1293,15 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.15" @@ -1373,7 +1420,7 @@ dependencies = [ "thiserror 2.0.17", "tokio", "tokio-util", - "tower", + "tower 0.5.2", "tower-http", "tracing", ] @@ -1421,7 +1468,7 @@ dependencies = [ "futures", "k8s-openapi", "kube", - "rand", + "rand 0.9.4", "snafu", "tokio", "tokio-util", @@ -1491,6 +1538,12 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "linux-raw-sys" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039" + [[package]] name = "litemap" version = "0.8.1" @@ -1576,6 +1629,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "multimap" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1667,7 +1726,7 @@ dependencies = [ "tokio-rustls", "tokio-stream", "tokio-util", - "tower", + "tower 0.5.2", "tower-http", "tracing", "tracing-subscriber", @@ -1773,6 +1832,16 @@ dependencies = [ "sha2", ] +[[package]] +name = "petgraph" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3672b37090dbd86368a4145bc067582552b29c27377cad4e0a306c97f9bd7772" +dependencies = [ + "fixedbitset", + "indexmap 2.12.0", +] + [[package]] name = "pin-project" version = "1.1.10" @@ -1835,6 +1904,16 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn", +] + [[package]] name = "proc-macro2" version = "1.0.103" @@ -1844,6 +1923,122 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-build" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" +dependencies = [ + "heck", + "itertools", + "log", + "multimap", + "once_cell", + "petgraph", + "prettyplease", + "prost", + "prost-types", + "regex", + "syn", + "tempfile", +] + +[[package]] +name = "prost-derive" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "prost-types" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16" +dependencies = [ + "prost", +] + +[[package]] +name = "protoc-bin-vendored" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1c381df33c98266b5f08186583660090a4ffa0889e76c7e9a5e175f645a67fa" +dependencies = [ + "protoc-bin-vendored-linux-aarch_64", + "protoc-bin-vendored-linux-ppcle_64", + "protoc-bin-vendored-linux-s390_64", + "protoc-bin-vendored-linux-x86_32", + "protoc-bin-vendored-linux-x86_64", + "protoc-bin-vendored-macos-aarch_64", + "protoc-bin-vendored-macos-x86_64", + "protoc-bin-vendored-win32", +] + +[[package]] +name = "protoc-bin-vendored-linux-aarch_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c350df4d49b5b9e3ca79f7e646fde2377b199e13cfa87320308397e1f37e1a4c" + +[[package]] +name = "protoc-bin-vendored-linux-ppcle_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a55a63e6c7244f19b5c6393f025017eb5d793fd5467823a099740a7a4222440c" + +[[package]] +name = "protoc-bin-vendored-linux-s390_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1dba5565db4288e935d5330a07c264a4ee8e4a5b4a4e6f4e83fad824cc32f3b0" + +[[package]] +name = "protoc-bin-vendored-linux-x86_32" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8854774b24ee28b7868cd71dccaae8e02a2365e67a4a87a6cd11ee6cdbdf9cf5" + +[[package]] +name = "protoc-bin-vendored-linux-x86_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b38b07546580df720fa464ce124c4b03630a6fb83e05c336fea2a241df7e5d78" + +[[package]] +name = "protoc-bin-vendored-macos-aarch_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89278a9926ce312e51f1d999fee8825d324d603213344a9a706daa009f1d8092" + +[[package]] +name = "protoc-bin-vendored-macos-x86_64" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81745feda7ccfb9471d7a4de888f0652e806d5795b61480605d4943176299756" + +[[package]] +name = "protoc-bin-vendored-win32" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3" + [[package]] name = "quinn" version = "0.11.9" @@ -1857,7 +2052,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2", + "socket2 0.6.1", "thiserror 2.0.17", "tokio", "tracing", @@ -1873,7 +2068,7 @@ dependencies = [ "bytes", "getrandom 0.3.4", "lru-slab", - "rand", + "rand 0.9.4", "ring", "rustc-hash", "rustls", @@ -1894,7 +2089,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2", + "socket2 0.6.1", "tracing", "windows-sys 0.60.2", ] @@ -1914,14 +2109,35 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rand" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" +dependencies = [ + "libc", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + [[package]] name = "rand" version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ - "rand_chacha", - "rand_core", + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", ] [[package]] @@ -1931,7 +2147,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.16", ] [[package]] @@ -2042,7 +2267,7 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", - "tower", + "tower 0.5.2", "tower-http", "tower-service", "url", @@ -2106,6 +2331,42 @@ version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +[[package]] +name = "rustfs-cosi-driver" +version = "0.1.0" +dependencies = [ + "hex", + "k8s-openapi", + "kube", + "operator", + "prost", + "protoc-bin-vendored", + "rustls", + "serde", + "serde_json", + "sha2", + "thiserror 2.0.17", + "tokio", + "tokio-stream", + "tonic", + "tonic-build", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "rustix" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "146c9e247ccc180c1f61615433868c99f3de3ae256a30a43b49f67c2d9171f34" +dependencies = [ + "bitflags", + "errno", + "libc", + "linux-raw-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rustls" version = "0.23.35" @@ -2361,7 +2622,7 @@ version = "0.9.34+deprecated" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" dependencies = [ - "indexmap", + "indexmap 2.12.0", "itoa", "ryu", "serde", @@ -2374,7 +2635,7 @@ version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b4db627b98b36d4203a7b458cf3573730f2bb591b28871d916dfa9efabfd41f" dependencies = [ - "indexmap", + "indexmap 2.12.0", "itoa", "ryu", "serde", @@ -2470,6 +2731,16 @@ dependencies = [ "syn", ] +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "socket2" version = "0.6.1" @@ -2550,6 +2821,19 @@ dependencies = [ "syn", ] +[[package]] +name = "tempfile" +version = "3.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0136791f7c95b1f6dd99f9cc786b91bb81c3800b639b3478e561ddb7be95e5f1" +dependencies = [ + "fastrand", + "getrandom 0.3.4", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "thiserror" version = "1.0.69" @@ -2668,7 +2952,7 @@ dependencies = [ "mio", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.1", "tokio-macros", "windows-sys 0.61.2", ] @@ -2722,6 +3006,70 @@ dependencies = [ "tokio", ] +[[package]] +name = "tonic" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52" +dependencies = [ + "async-stream", + "async-trait", + "axum", + "base64", + "bytes", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "prost", + "socket2 0.5.10", + "tokio", + "tokio-stream", + "tower 0.4.13", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-build" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9557ce109ea773b399c9b9e5dca39294110b74f1f342cb347a80d1fce8c26a11" +dependencies = [ + "prettyplease", + "proc-macro2", + "prost-build", + "prost-types", + "quote", + "syn", +] + +[[package]] +name = "tower" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" +dependencies = [ + "futures-core", + "futures-util", + "indexmap 1.9.3", + "pin-project", + "pin-project-lite", + "rand 0.8.7", + "slab", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "tower" version = "0.5.2" @@ -2763,7 +3111,7 @@ dependencies = [ "pin-project-lite", "tokio", "tokio-util", - "tower", + "tower 0.5.2", "tower-layer", "tower-service", "tracing", @@ -2947,7 +3295,7 @@ version = "5.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2fcc29c80c21c31608227e0912b2d7fddba57ad76b606890627ba8ee7964e993" dependencies = [ - "indexmap", + "indexmap 2.12.0", "serde", "serde_json", "utoipa-gen", @@ -3493,7 +3841,7 @@ dependencies = [ "crossbeam-utils", "displaydoc", "flate2", - "indexmap", + "indexmap 2.12.0", "memchr", "thiserror 2.0.17", "zopfli", diff --git a/Cargo.toml b/Cargo.toml index 84c682e..db0d7e1 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,7 +63,7 @@ shadow-rs = { version = "1.5.0", features = ["build"] } unused_variables = "allow" [workspace] -members = ["crates/leader-election"] +members = ["crates/leader-election", "crates/cosi-driver"] [lints.clippy] unwrap_used = "deny" diff --git a/Dockerfile b/Dockerfile index 97e4b1b..a598152 100755 --- a/Dockerfile +++ b/Dockerfile @@ -17,6 +17,9 @@ ARG PNPM_VERSION=10.28.1 # Shared Cargo settings for slow / flaky networks (applies to all Rust stages) FROM ${RUST_BUILD_IMAGE} AS rust-base +RUN apt-get update \ + && apt-get install -y --no-install-recommends protobuf-compiler \ + && rm -rf /var/lib/apt/lists/* RUN mkdir -p /usr/local/cargo && \ printf '%s\n' \ '[http]' \ @@ -45,15 +48,15 @@ FROM rust-base AS cacher COPY --from=cargo-chef-installer /usr/local/cargo/bin/cargo-chef /usr/local/cargo/bin/cargo-chef WORKDIR /app COPY --from=planner /app/recipe.json recipe.json -RUN cargo chef cook --release --recipe-path recipe.json +RUN cargo chef cook --release --workspace --recipe-path recipe.json -# Stage 3: Build the binary +# Stage 3: Build the binaries (operator + COSI driver) FROM rust-base AS builder WORKDIR /app COPY . . COPY --from=cacher /app/target target COPY --from=cacher /usr/local/cargo /usr/local/cargo -RUN cargo build --release +RUN cargo build --release -p operator -p rustfs-cosi-driver # Stage 4: Build the static Console frontend FROM ${NODE_BUILD_IMAGE} AS console-web-builder @@ -72,5 +75,6 @@ FROM ${BASE_IMAGE} WORKDIR /app COPY --from=builder /app/target/release/operator . +COPY --from=builder /app/target/release/rustfs-cosi-driver . COPY --from=console-web-builder /app/console-web/out ./console-web ENTRYPOINT ["./operator"] diff --git a/crates/cosi-driver/Cargo.toml b/crates/cosi-driver/Cargo.toml new file mode 100644 index 0000000..5ef0848 --- /dev/null +++ b/crates/cosi-driver/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "rustfs-cosi-driver" +version = "0.1.0" +edition = "2024" +license = "Apache-2.0" +publish = false + +[[bin]] +name = "rustfs-cosi-driver" +path = "src/main.rs" + +[dependencies] +operator = { path = "../.." } +tokio = { version = "1.49.0", features = ["rt", "rt-multi-thread", "macros", "fs", "signal", "net"] } +tokio-stream = { version = "0.1", features = ["net"] } +tonic = "0.12" +prost = "0.13" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } +kube = { version = "2.0.1", features = ["client", "rustls-tls"] } +k8s-openapi = { version = "0.26.1", features = ["v1_30"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +sha2 = "0.10" +hex = "0.4" +thiserror = "2" +rustls = { version = "0.23", default-features = false, features = ["ring"] } + +[build-dependencies] +tonic-build = "0.12" +protoc-bin-vendored = "3" diff --git a/crates/cosi-driver/build.rs b/crates/cosi-driver/build.rs new file mode 100644 index 0000000..cce675e --- /dev/null +++ b/crates/cosi-driver/build.rs @@ -0,0 +1,16 @@ +fn main() -> Result<(), Box> { + // Prefer PATH `protoc` (Dockerfile installs protobuf-compiler); fall back to vendored. + if std::env::var_os("PROTOC").is_none() + && let Ok(protoc) = protoc_bin_vendored::protoc_bin_path() + { + // SAFETY: build script is single-threaded before codegen. + unsafe { + std::env::set_var("PROTOC", protoc); + } + } + tonic_build::configure() + .build_server(true) + .build_client(false) + .compile_protos(&["proto/cosi.proto"], &["proto"])?; + Ok(()) +} diff --git a/crates/cosi-driver/proto/cosi.proto b/crates/cosi-driver/proto/cosi.proto new file mode 100644 index 0000000..d26273d --- /dev/null +++ b/crates/cosi-driver/proto/cosi.proto @@ -0,0 +1,97 @@ +syntax = "proto3"; +package cosi.v1alpha1; + +option go_package = "sigs.k8s.io/container-object-storage-interface/proto;cosi"; + +service Identity { + rpc DriverGetInfo (DriverGetInfoRequest) returns (DriverGetInfoResponse) {} +} + +service Provisioner { + rpc DriverCreateBucket (DriverCreateBucketRequest) returns (DriverCreateBucketResponse) {} + rpc DriverDeleteBucket (DriverDeleteBucketRequest) returns (DriverDeleteBucketResponse) {} + rpc DriverGrantBucketAccess (DriverGrantBucketAccessRequest) returns (DriverGrantBucketAccessResponse); + rpc DriverRevokeBucketAccess (DriverRevokeBucketAccessRequest) returns (DriverRevokeBucketAccessResponse); +} + +enum S3SignatureVersion { + UnknownSignature = 0; + S3V2 = 1; + S3V4 = 2; +} + +enum AuthenticationType { + UnknownAuthenticationType = 0; + Key = 1; + IAM = 2; +} + +message S3 { + string region = 1; + S3SignatureVersion signature_version = 2; +} + +message AzureBlob { + string storage_account = 1; +} + +message GCS { + string private_key_name = 1; + string project_id = 2; + string service_account = 3; +} + +message Protocol { + oneof type { + S3 s3 = 1; + AzureBlob azureBlob = 2; + GCS gcs = 3; + } +} + +message CredentialDetails { + map secrets = 1; +} + +message DriverGetInfoRequest {} + +message DriverGetInfoResponse { + string name = 1; +} + +message DriverCreateBucketRequest { + string name = 1; + map parameters = 2; +} + +message DriverCreateBucketResponse { + string bucket_id = 1; + Protocol bucket_info = 2; +} + +message DriverDeleteBucketRequest { + string bucket_id = 1; + map delete_context = 2; +} + +message DriverDeleteBucketResponse {} + +message DriverGrantBucketAccessRequest { + string bucket_id = 1; + string name = 2; + AuthenticationType authentication_type = 3; + map parameters = 4; +} + +message DriverGrantBucketAccessResponse { + string account_id = 1; + map credentials = 2; +} + +message DriverRevokeBucketAccessRequest { + string bucket_id = 1; + string account_id = 2; + map revoke_access_context = 3; +} + +message DriverRevokeBucketAccessResponse {} diff --git a/crates/cosi-driver/src/backend.rs b/crates/cosi-driver/src/backend.rs new file mode 100644 index 0000000..b83e8da --- /dev/null +++ b/crates/cosi-driver/src/backend.rs @@ -0,0 +1,146 @@ +//! Admin credential lookup + RustFS admin client construction. + +use k8s_openapi::api::core::v1::{ConfigMap, Secret}; +use kube::{Api, Client}; +use operator::sts::rustfs_client::RustfsAdminClient; +use thiserror::Error; +use tracing::info; + +use crate::parameters::BackendParameters; + +#[derive(Debug, Error)] +pub enum BackendError { + #[error("failed to read Secret {namespace}/{name}: {source}")] + SecretLookup { + namespace: String, + name: String, + #[source] + source: kube::Error, + }, + #[error("secret key missing: {0}")] + MissingSecretKey(&'static str), + #[error("secret key empty: {0}")] + EmptySecretKey(&'static str), + #[error("secret key is not valid utf8: {0}")] + InvalidSecretKey(&'static str), + #[error("failed to read ConfigMap {namespace}/{name}: {source}")] + ConfigMapLookup { + namespace: String, + name: String, + #[source] + source: kube::Error, + }, + #[error("configmap key missing: {0}")] + MissingCaData(&'static str), + #[error(transparent)] + ClientBuild(#[from] operator::sts::rustfs_client::RustfsClientError), +} + +#[allow(clippy::result_large_err)] +fn secret_value<'a>(secret: &'a Secret, keys: &[&'static str]) -> Result<&'a str, BackendError> { + let data = secret + .data + .as_ref() + .ok_or(BackendError::MissingSecretKey(keys[0]))?; + for key in keys { + if let Some(bytes) = data.get(*key) { + let value = + std::str::from_utf8(&bytes.0).map_err(|_| BackendError::InvalidSecretKey(key))?; + if value.is_empty() { + return Err(BackendError::EmptySecretKey(key)); + } + return Ok(value); + } + } + Err(BackendError::MissingSecretKey(keys[0])) +} + +pub async fn admin_client_from_params( + kube: &Client, + params: &BackendParameters, +) -> Result { + let secrets: Api = + Api::namespaced(kube.clone(), ¶ms.object_store_user_secret_namespace); + let secret = secrets + .get(¶ms.object_store_user_secret_name) + .await + .map_err(|source| BackendError::SecretLookup { + namespace: params.object_store_user_secret_namespace.clone(), + name: params.object_store_user_secret_name.clone(), + source, + })?; + + let access_key = secret_value( + &secret, + &[ + "accesskey", + "accessKey", + "ACCESSKEY", + "AWS_ACCESS_KEY_ID", + "access_key", + "access-key", + "access_key_id", + "access-key-id", + "RUSTFS_ACCESS_KEY", + ], + )?; + let secret_key = secret_value( + &secret, + &[ + "secretkey", + "secretKey", + "SECRETKEY", + "AWS_SECRET_ACCESS_KEY", + "secret_key", + "secret-key", + "secret_access_key", + "secret-access-key", + "RUSTFS_SECRET_KEY", + ], + )?; + + info!( + endpoint = %params.endpoint, + secret = %params.object_store_user_secret_name, + "building RustFS admin client" + ); + + if let (Some(cm_name), Some(cm_ns)) = ( + params.tls_ca_configmap_name.as_ref(), + params + .tls_ca_configmap_namespace + .as_ref() + .or(Some(¶ms.object_store_user_secret_namespace)), + ) { + let cms: Api = Api::namespaced(kube.clone(), cm_ns); + let cm = cms + .get(cm_name) + .await + .map_err(|source| BackendError::ConfigMapLookup { + namespace: cm_ns.clone(), + name: cm_name.clone(), + source, + })?; + let ca = cm + .data + .as_ref() + .and_then(|d| { + d.get("ca.crt") + .or_else(|| d.get("tls.crt")) + .or_else(|| d.get("ca-bundle.crt")) + }) + .ok_or(BackendError::MissingCaData("ca.crt"))?; + return Ok(RustfsAdminClient::new_with_base_url_and_ca_pem( + params.endpoint.clone(), + access_key, + secret_key, + ca.as_bytes(), + )?); + } + + Ok(RustfsAdminClient::new_with_base_url( + params.endpoint.clone(), + access_key, + secret_key, + )) +} diff --git a/crates/cosi-driver/src/driver.rs b/crates/cosi-driver/src/driver.rs new file mode 100644 index 0000000..6a60632 --- /dev/null +++ b/crates/cosi-driver/src/driver.rs @@ -0,0 +1,349 @@ +//! COSI Identity + Provisioner gRPC services. + +use std::collections::HashMap; + +use kube::Client; +use operator::sts::rustfs_client::{CreateBucketResult, RustfsAdminClient}; +use tonic::{Request, Response, Status}; +use tracing::{error, info}; + +use crate::backend::{BackendError, admin_client_from_params}; +use crate::parameters::{ + BackendParameters, DRIVER_NAME, bucket_policy_document_for, credentials_for_account, + grant_owner_policy_document, grant_owner_policy_name, policy_name_for, +}; +use crate::proto::cosi::v1alpha1::{ + AuthenticationType, CredentialDetails, DriverCreateBucketRequest, DriverCreateBucketResponse, + DriverDeleteBucketRequest, DriverDeleteBucketResponse, DriverGetInfoRequest, + DriverGetInfoResponse, DriverGrantBucketAccessRequest, DriverGrantBucketAccessResponse, + DriverRevokeBucketAccessRequest, DriverRevokeBucketAccessResponse, Protocol, S3, + S3SignatureVersion, identity_server::Identity, provisioner_server::Provisioner, +}; + +#[derive(Clone)] +pub struct Driver { + kube: Client, +} + +impl Driver { + pub fn new(kube: Client) -> Self { + Self { kube } + } +} + +fn map_backend(err: BackendError) -> Status { + error!(error = %err, "backend error"); + Status::internal(err.to_string()) +} + +fn map_params(err: crate::parameters::ParameterError) -> Status { + Status::invalid_argument(err.to_string()) +} + +fn map_admin(err: operator::sts::rustfs_client::RustfsClientError) -> Status { + error!(error = %err, "rustfs admin error"); + Status::internal(err.to_string()) +} + +fn credential_map( + access_key: &str, + secret_key: &str, + params: &BackendParameters, + policy_buckets: &[String], +) -> HashMap { + let mut secrets = HashMap::new(); + secrets.insert("accessKeyID".to_string(), access_key.to_string()); + secrets.insert("accessSecretKey".to_string(), secret_key.to_string()); + secrets.insert("AWS_ACCESS_KEY_ID".to_string(), access_key.to_string()); + secrets.insert("AWS_SECRET_ACCESS_KEY".to_string(), secret_key.to_string()); + secrets.insert("ACCESSKEY".to_string(), access_key.to_string()); + secrets.insert("SECRETKEY".to_string(), secret_key.to_string()); + secrets.insert("endpoint".to_string(), params.endpoint.clone()); + secrets.insert("region".to_string(), params.region.clone()); + secrets.insert( + "BUCKETS".to_string(), + params + .buckets + .clone() + .unwrap_or_else(|| policy_buckets.join(",")), + ); + secrets +} + +fn user_owns_grant( + policy_names: &[String], + owner_policy: &str, + account_id: &str, + grant_name: &str, +) -> bool { + // Default Ceph-style path: account id is the COSI grant name itself. + if account_id == grant_name { + return true; + } + policy_names.iter().any(|name| name == owner_policy) +} + +async fn ensure_grant_policies( + client: &RustfsAdminClient, + access_key: &str, + bucket_policy_name: &str, + bucket_policy_doc: &str, + owner_policy_name: &str, +) -> Result<(), Status> { + client + .add_canned_policy(bucket_policy_name, bucket_policy_doc) + .await + .map_err(map_admin)?; + client + .add_canned_policy(owner_policy_name, &grant_owner_policy_document()) + .await + .map_err(map_admin)?; + client + .set_user_policy( + access_key, + &[ + bucket_policy_name.to_string(), + owner_policy_name.to_string(), + ], + ) + .await + .map_err(map_admin)?; + Ok(()) +} + +#[tonic::async_trait] +impl Identity for Driver { + async fn driver_get_info( + &self, + _request: Request, + ) -> Result, Status> { + Ok(Response::new(DriverGetInfoResponse { + name: DRIVER_NAME.to_string(), + })) + } +} + +#[tonic::async_trait] +impl Provisioner for Driver { + async fn driver_create_bucket( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.name.trim().is_empty() { + return Err(Status::invalid_argument("bucket name is required")); + } + let params = BackendParameters::from_map(&req.parameters).map_err(map_params)?; + let client = admin_client_from_params(&self.kube, ¶ms) + .await + .map_err(map_backend)?; + + let buckets = params.buckets_to_create(&req.name); + if buckets.is_empty() { + return Err(Status::invalid_argument( + "no buckets to create (buckets/bucketName empty or only *)", + )); + } + let bucket_id = params.primary_bucket_id(&req.name); + + for bucket in &buckets { + info!(bucket = %bucket, cosi_name = %req.name, "creating bucket"); + match client + .create_bucket(bucket, Some(params.region.as_str()), false) + .await + .map_err(map_admin)? + { + CreateBucketResult::Created | CreateBucketResult::AlreadyExists => {} + } + } + + Ok(Response::new(DriverCreateBucketResponse { + bucket_id, + bucket_info: Some(Protocol { + r#type: Some(crate::proto::cosi::v1alpha1::protocol::Type::S3(S3 { + region: params.region, + signature_version: S3SignatureVersion::S3v4 as i32, + })), + }), + })) + } + + async fn driver_delete_bucket( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.bucket_id.trim().is_empty() { + return Err(Status::invalid_argument("bucket_id is required")); + } + let params = BackendParameters::from_map(&req.delete_context).map_err(map_params)?; + let client = admin_client_from_params(&self.kube, ¶ms) + .await + .map_err(map_backend)?; + + let buckets = params.buckets_to_create(&req.bucket_id); + let targets = if buckets.is_empty() { + vec![req.bucket_id.clone()] + } else { + buckets + }; + + for bucket in &targets { + info!(bucket = %bucket, "deleting bucket"); + client.delete_bucket(bucket).await.map_err(map_admin)?; + } + Ok(Response::new(DriverDeleteBucketResponse {})) + } + + async fn driver_grant_bucket_access( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.bucket_id.trim().is_empty() { + return Err(Status::invalid_argument("bucket_id is required")); + } + if req.name.trim().is_empty() { + return Err(Status::invalid_argument("account name is required")); + } + if req.authentication_type != AuthenticationType::Key as i32 + && req.authentication_type != AuthenticationType::UnknownAuthenticationType as i32 + { + return Err(Status::invalid_argument( + "only KEY authentication is supported", + )); + } + + let params = BackendParameters::from_map(&req.parameters).map_err(map_params)?; + let client = admin_client_from_params(&self.kube, ¶ms) + .await + .map_err(map_backend)?; + + let grant_name = req.name.clone(); + let access_key = params + .preferred_access_key + .clone() + .unwrap_or_else(|| grant_name.clone()); + let secret_key = credentials_for_account(&access_key); + let policy_buckets = params.buckets_for_policy(&req.bucket_id); + let bucket_policy_name = params + .policy + .clone() + .unwrap_or_else(|| policy_name_for(&access_key)); + let bucket_policy_doc = bucket_policy_document_for(&policy_buckets); + let owner_policy_name = grant_owner_policy_name(&grant_name); + + info!( + bucket = %req.bucket_id, + account = %access_key, + grant = %grant_name, + policy = %bucket_policy_name, + owner_policy = %owner_policy_name, + buckets = %policy_buckets.join(","), + "granting bucket access" + ); + + match client.get_user_info(&access_key).await.map_err(map_admin)? { + Some(info) => { + if !user_owns_grant( + &info.policy_names, + &owner_policy_name, + &access_key, + &grant_name, + ) { + return Err(Status::already_exists(format!( + "preferredAccessKey `{access_key}` is already bound to another BucketAccess; \ + omit preferredAccessKey or choose a unique value" + ))); + } + // Same grant retry (or Ceph-style account == grant name): never rotate secret. + ensure_grant_policies( + &client, + &access_key, + &bucket_policy_name, + &bucket_policy_doc, + &owner_policy_name, + ) + .await?; + } + None => { + client + .add_user(&access_key, &secret_key) + .await + .map_err(map_admin)?; + ensure_grant_policies( + &client, + &access_key, + &bucket_policy_name, + &bucket_policy_doc, + &owner_policy_name, + ) + .await?; + } + } + + let secrets = credential_map(&access_key, &secret_key, ¶ms, &policy_buckets); + let mut credentials = HashMap::new(); + credentials.insert("s3".to_string(), CredentialDetails { secrets }); + + Ok(Response::new(DriverGrantBucketAccessResponse { + account_id: access_key, + credentials, + })) + } + + async fn driver_revoke_bucket_access( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + if req.bucket_id.trim().is_empty() { + return Err(Status::invalid_argument("bucket_id is required")); + } + if req.account_id.trim().is_empty() { + return Err(Status::invalid_argument("account_id is required")); + } + + let params = BackendParameters::from_map(&req.revoke_access_context).map_err(map_params)?; + let client = admin_client_from_params(&self.kube, ¶ms) + .await + .map_err(map_backend)?; + + info!( + bucket = %req.bucket_id, + account = %req.account_id, + "revoking bucket access" + ); + client + .remove_user(&req.account_id) + .await + .map_err(map_admin)?; + Ok(Response::new(DriverRevokeBucketAccessResponse {})) + } +} + +#[cfg(test)] +mod grant_tests { + use super::user_owns_grant; + + #[test] + fn same_grant_name_as_account_is_idempotent() { + assert!(user_owns_grant(&[], "cosi-grant-ba-1", "ba-1", "ba-1")); + } + + #[test] + fn preferred_key_requires_owner_marker() { + assert!(!user_owns_grant( + &["cosi-mlflow".to_string()], + "cosi-grant-ba-1", + "mlflow", + "ba-1" + )); + assert!(user_owns_grant( + &["cosi-mlflow".to_string(), "cosi-grant-ba-1".to_string()], + "cosi-grant-ba-1", + "mlflow", + "ba-1" + )); + } +} diff --git a/crates/cosi-driver/src/main.rs b/crates/cosi-driver/src/main.rs new file mode 100644 index 0000000..f39059c --- /dev/null +++ b/crates/cosi-driver/src/main.rs @@ -0,0 +1,98 @@ +//! RustFS COSI driver — serves Identity + Provisioner on a Unix socket. + +mod backend; +mod driver; +mod parameters; + +pub mod proto { + pub mod cosi { + pub mod v1alpha1 { + tonic::include_proto!("cosi.v1alpha1"); + } + } +} + +use std::path::PathBuf; + +use kube::Client; +use tokio::net::UnixListener; +use tokio_stream::wrappers::UnixListenerStream; +use tracing::{error, info}; + +use crate::driver::Driver; +use crate::parameters::DRIVER_NAME; +use crate::proto::cosi::v1alpha1::{ + identity_server::IdentityServer, provisioner_server::ProvisionerServer, +}; + +fn parse_unix_endpoint(raw: &str) -> Result { + let trimmed = raw.trim(); + let path = trimmed + .strip_prefix("unix://") + .ok_or_else(|| format!("unsupported COSI_ENDPOINT `{trimmed}`"))?; + if path.is_empty() { + return Err("COSI_ENDPOINT unix path is empty".to_string()); + } + Ok(PathBuf::from(path)) +} + +#[tokio::main] +async fn main() { + // Required for rustls 0.23 when multiple crypto backends may be linked via deps. + let _ = rustls::crypto::ring::default_provider().install_default(); + + let _ = tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_target(true) + .try_init(); + + let endpoint = std::env::var("COSI_ENDPOINT") + .unwrap_or_else(|_| "unix:///var/lib/cosi/cosi.sock".to_string()); + let sock_path = match parse_unix_endpoint(&endpoint) { + Ok(path) => path, + Err(err) => { + error!(error = %err, "invalid COSI_ENDPOINT"); + std::process::exit(2); + } + }; + + if let Some(parent) = sock_path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let _ = std::fs::remove_file(&sock_path); + + let kube = match Client::try_default().await { + Ok(client) => client, + Err(err) => { + error!(error = %err, "failed to create Kubernetes client"); + std::process::exit(1); + } + }; + + let listener = match UnixListener::bind(&sock_path) { + Ok(listener) => listener, + Err(err) => { + error!(error = %err, path = %sock_path.display(), "failed to bind COSI socket"); + std::process::exit(1); + } + }; + let incoming = UnixListenerStream::new(listener); + let driver = Driver::new(kube); + + info!( + driver = DRIVER_NAME, + endpoint = %endpoint, + "starting RustFS COSI driver" + ); + + let result = tonic::transport::Server::builder() + .add_service(IdentityServer::new(driver.clone())) + .add_service(ProvisionerServer::new(driver)) + .serve_with_incoming(incoming) + .await; + + if let Err(err) = result { + error!(error = %err, "RustFS COSI driver stopped"); + std::process::exit(1); + } +} diff --git a/crates/cosi-driver/src/parameters.rs b/crates/cosi-driver/src/parameters.rs new file mode 100644 index 0000000..f875921 --- /dev/null +++ b/crates/cosi-driver/src/parameters.rs @@ -0,0 +1,196 @@ +//! RustFS COSI driver parameters (BucketClass / BucketAccessClass). + +use std::collections::HashMap; + +use thiserror::Error; + +pub const DRIVER_NAME: &str = "rustfs.objectstorage.k8s.io"; + +#[derive(Debug, Clone)] +pub struct BackendParameters { + pub endpoint: String, + pub object_store_user_secret_name: String, + pub object_store_user_secret_namespace: String, + pub region: String, + pub policy: Option, + pub tls_ca_configmap_name: Option, + pub tls_ca_configmap_namespace: Option, + /// Preferred S3 bucket name (overrides COSI-generated CreateBucket name). + pub bucket_name: Option, + /// Comma-separated bucket list to create / authorize (`*` = full access). + pub buckets: Option, + /// Preferred access-key / account name for GrantBucketAccess. + /// + /// Must be unique per BucketAccess. Reusing the same value across claims is + /// rejected by the driver (Ceph-style isolation). Prefer omitting this so the + /// COSI grant name (`ba-`) is used as the account id. + pub preferred_access_key: Option, +} + +#[derive(Debug, Error)] +pub enum ParameterError { + #[error("missing required parameter `{0}`")] + MissingRequired(&'static str), + #[error("parameter `{0}` is empty")] + Empty(&'static str), +} + +fn required(map: &HashMap, key: &'static str) -> Result { + let value = map + .get(key) + .cloned() + .ok_or(ParameterError::MissingRequired(key))?; + if value.trim().is_empty() { + return Err(ParameterError::Empty(key)); + } + Ok(value) +} + +fn optional(map: &HashMap, key: &str) -> Option { + map.get(key) + .cloned() + .filter(|value| !value.trim().is_empty()) +} + +impl BackendParameters { + pub fn from_map(map: &HashMap) -> Result { + Ok(Self { + endpoint: required(map, "endpoint")?, + object_store_user_secret_name: required(map, "objectStoreUserSecretName")?, + object_store_user_secret_namespace: required(map, "objectStoreUserSecretNamespace")?, + region: optional(map, "region").unwrap_or_else(|| "us-east-1".to_string()), + policy: optional(map, "policy"), + tls_ca_configmap_name: optional(map, "tlsCAConfigMapName"), + tls_ca_configmap_namespace: optional(map, "tlsCAConfigMapNamespace"), + bucket_name: optional(map, "bucketName"), + buckets: optional(map, "buckets"), + preferred_access_key: optional(map, "preferredAccessKey") + .or_else(|| optional(map, "accessKey")), + }) + } + + /// Buckets to create (excludes `*`). Primary bucket_id is the first entry. + pub fn buckets_to_create(&self, fallback_name: &str) -> Vec { + let raw = self + .buckets + .as_deref() + .or(self.bucket_name.as_deref()) + .unwrap_or(fallback_name); + raw.split(',') + .map(str::trim) + .filter(|b| !b.is_empty() && *b != "*") + .map(ToOwned::to_owned) + .collect() + } + + /// Full bucket list for IAM policy (may include `*`). + pub fn buckets_for_policy(&self, fallback_name: &str) -> Vec { + let raw = self + .buckets + .as_deref() + .or(self.bucket_name.as_deref()) + .unwrap_or(fallback_name); + let list: Vec = raw + .split(',') + .map(str::trim) + .filter(|b| !b.is_empty()) + .map(ToOwned::to_owned) + .collect(); + if list.is_empty() { + vec![fallback_name.to_string()] + } else { + list + } + } + + pub fn primary_bucket_id(&self, cosi_name: &str) -> String { + self.bucket_name + .clone() + .or_else(|| self.buckets_to_create(cosi_name).into_iter().next()) + .unwrap_or_else(|| cosi_name.to_string()) + } +} + +pub fn bucket_policy_document_for(buckets: &[String]) -> String { + let has_wildcard = buckets.iter().any(|b| b == "*"); + let resources: Vec = if has_wildcard { + vec!["arn:aws:s3:::*".to_string(), "arn:aws:s3:::*/*".to_string()] + } else { + buckets + .iter() + .flat_map(|b| [format!("arn:aws:s3:::{b}"), format!("arn:aws:s3:::{b}/*")]) + .collect() + }; + serde_json::json!({ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": ["s3:*"], + "Resource": resources + }] + }) + .to_string() +} + +pub fn sanitize_policy_fragment(value: &str) -> String { + value + .chars() + .map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }) + .collect() +} + +pub fn policy_name_for(bucket: &str) -> String { + format!("cosi-{}", sanitize_policy_fragment(bucket)) +} + +/// Owner marker policy bound to a specific COSI grant `name` (`ba-`). +pub fn grant_owner_policy_name(grant_name: &str) -> String { + format!("cosi-grant-{}", sanitize_policy_fragment(grant_name)) +} + +/// Minimal canned policy used only as an ownership marker for a BucketAccess grant. +pub fn grant_owner_policy_document() -> String { + serde_json::json!({ + "Version": "2012-10-17", + "Statement": [{ + "Sid": "CosiGrantOwner", + "Effect": "Allow", + "Action": ["s3:ListAllMyBuckets"], + "Resource": ["arn:aws:s3:::*"] + }] + }) + .to_string() +} + +/// Deterministic secret so DriverGrantBucketAccess is idempotent across sidecar retries. +pub fn credentials_for_account(account_id: &str) -> String { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(format!("rustfs-cosi-v1:{account_id}").as_bytes()); + hex::encode(digest) +} + +#[cfg(test)] +mod tests { + use super::{credentials_for_account, grant_owner_policy_name}; + + #[test] + fn credentials_are_deterministic_and_long_enough() { + let a = credentials_for_account("ba-test-uid"); + let b = credentials_for_account("ba-test-uid"); + assert_eq!(a, b); + assert!(a.len() >= 8); + assert_ne!(a, credentials_for_account("other-account")); + } + + #[test] + fn grant_owner_policy_name_sanitizes() { + assert_eq!( + grant_owner_policy_name("ba-81733d1a-ac7a-4759-96f3-fbcc07c0cee9"), + "cosi-grant-ba-81733d1a-ac7a-4759-96f3-fbcc07c0cee9" + ); + assert_eq!( + grant_owner_policy_name("ba/weird.name"), + "cosi-grant-ba-weird-name" + ); + } +} diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md index 734d314..c7e0146 100644 --- a/docs/operator-user-guide.md +++ b/docs/operator-user-guide.md @@ -1097,6 +1097,14 @@ For the RustFS Tenant Console, use the Tenant admin credentials from `spec.creds - Keep Tenant examples under version control, but never commit raw Secret values. - Check `status.conditions` before debugging lower-level StatefulSets. +## 13.1 COSI `preferredAccessKey` + +When using the RustFS COSI driver (`rustfs.objectstorage.k8s.io`): + +- Prefer omitting `preferredAccessKey` so each `BucketAccess` gets a unique account id derived from the COSI grant name (`ba-`), matching Ceph COSI isolation. +- If you set `preferredAccessKey` (or `accessKey`), the value must be unique per `BucketAccess`. Reusing the same key across claims is rejected with `AlreadyExists` so credentials are never rotated out from under another workload. +- Grant retries for the same `BucketAccess` are idempotent and return the same secret; the driver does not overwrite an existing user's secret key. + ## 14. Related Documentation - [Project README](../README.md) diff --git a/src/sts/admin_ops.rs b/src/sts/admin_ops.rs index 593d1a5..e80cd34 100644 --- a/src/sts/admin_ops.rs +++ b/src/sts/admin_ops.rs @@ -23,12 +23,40 @@ use super::helpers::{ }; use super::{ ADD_CANNED_POLICY_PATH, ADD_USER_PATH, ADMIN_SIGNING_SERVICE, INFO_CANNED_POLICY_PATH, - JSON_CONTENT_TYPE, LIST_CANNED_POLICIES_PATH, RustfsAdminClient, RustfsClientError, - RustfsServerInfo, RustfsServerInfoResponse, SERVER_INFO_PATH, SET_POLICY_PATH, USER_INFO_PATH, + JSON_CONTENT_TYPE, LIST_CANNED_POLICIES_PATH, REMOVE_USER_PATH, RustfsAdminClient, + RustfsClientError, RustfsServerInfo, RustfsServerInfoResponse, RustfsUserInfo, + SERVER_INFO_PATH, SET_POLICY_PATH, USER_INFO_PATH, }; use reqwest::StatusCode; use serde_json::Value; +fn parse_user_info_policy_names(body: &Value) -> Vec { + let Some(field) = body + .get("policyName") + .or_else(|| body.get("policy_name")) + .or_else(|| body.get("PolicyName")) + else { + return Vec::new(); + }; + + match field { + Value::String(raw) => raw + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(ToOwned::to_owned) + .collect(), + Value::Array(items) => items + .iter() + .filter_map(|item| item.as_str()) + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(ToOwned::to_owned) + .collect(), + _ => Vec::new(), + } +} + impl RustfsAdminClient { // Admin duties: user and policy management APIs. // (Candidly scoped to tenant admin operations.) @@ -155,7 +183,11 @@ impl RustfsAdminClient { .map_err(|_| RustfsClientError::ParseResponseFailed) } - pub async fn user_exists(&self, access_key: &str) -> Result { + /// Fetch IAM user info. Returns `Ok(None)` when the user does not exist. + pub async fn get_user_info( + &self, + access_key: &str, + ) -> Result, RustfsClientError> { if access_key.trim().is_empty() { return Err(RustfsClientError::InvalidCredentialValue { key: "accesskey" }); } @@ -178,13 +210,21 @@ impl RustfsAdminClient { .map_err(|_| RustfsClientError::RequestFailed)?; if response.status().is_success() { - return Ok(true); + let body = response + .text() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + let parsed: Value = + serde_json::from_str(&body).map_err(|_| RustfsClientError::ParseResponseFailed)?; + return Ok(Some(RustfsUserInfo { + policy_names: parse_user_info_policy_names(&parsed), + })); } let status = response.status(); let (body, truncated) = RustfsClientError::limited_response_body(response).await; if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) { - return Ok(false); + return Ok(None); } Err(RustfsClientError::unexpected_status_with_limited_body( @@ -192,6 +232,10 @@ impl RustfsAdminClient { )) } + pub async fn user_exists(&self, access_key: &str) -> Result { + Ok(self.get_user_info(access_key).await?.is_some()) + } + pub async fn add_user( &self, access_key: &str, @@ -239,4 +283,71 @@ impl RustfsAdminClient { .await .map(|_| ()) } + + /// Remove a RustFS user. Missing users are treated as success (idempotent). + pub async fn remove_user(&self, access_key: &str) -> Result<(), RustfsClientError> { + if access_key.trim().is_empty() { + return Err(RustfsClientError::InvalidCredentialValue { key: "accesskey" }); + } + + let query = build_canonical_query(&[("accessKey", access_key)]); + let path = REMOVE_USER_PATH; + let url = format!("{}{}?{query}", self.base_url.trim_end_matches('/'), path); + let signed = self.sign_request("DELETE", path, &query, "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + + let response = self + .http_client + .delete(url) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + return Ok(()); + } + + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) { + return Ok(()); + } + + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } +} + +#[cfg(test)] +mod parse_tests { + use super::parse_user_info_policy_names; + use serde_json::json; + + #[test] + fn parses_comma_separated_policy_name() { + let body = json!({"policyName":"cosi-mlflow,cosi-grant-ba-1"}); + assert_eq!( + parse_user_info_policy_names(&body), + vec!["cosi-mlflow".to_string(), "cosi-grant-ba-1".to_string()] + ); + } + + #[test] + fn parses_policy_name_array_and_snake_case() { + let body = json!({"policy_name":["a","b"]}); + assert_eq!( + parse_user_info_policy_names(&body), + vec!["a".to_string(), "b".to_string()] + ); + } + + #[test] + fn missing_policy_field_yields_empty() { + assert!(parse_user_info_policy_names(&json!({"status":"enabled"})).is_empty()); + } } diff --git a/src/sts/rustfs_client.rs b/src/sts/rustfs_client.rs index 865c27f..9b51378 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -45,6 +45,7 @@ const FORM_CONTENT_TYPE: &str = "application/x-www-form-urlencoded"; const JSON_CONTENT_TYPE: &str = "application/json"; const ASSUME_ROLE_PATH: &str = "/"; const ADD_USER_PATH: &str = "/rustfs/admin/v3/add-user"; +const REMOVE_USER_PATH: &str = "/rustfs/admin/v3/remove-user"; const USER_INFO_PATH: &str = "/rustfs/admin/v3/user-info"; const SET_POLICY_PATH: &str = "/rustfs/admin/v3/set-policy"; const LIST_CANNED_POLICIES_PATH: &str = "/rustfs/admin/v3/list-canned-policies"; @@ -105,6 +106,12 @@ pub enum CreateBucketResult { AlreadyExists, } +/// Subset of `/rustfs/admin/v3/user-info` used by COSI grant ownership checks. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct RustfsUserInfo { + pub policy_names: Vec, +} + #[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] pub struct RustfsPoolDecommissionInfo { #[serde(rename = "startTime")] diff --git a/src/sts/s3_ops.rs b/src/sts/s3_ops.rs index 0df217f..023dc1a 100644 --- a/src/sts/s3_ops.rs +++ b/src/sts/s3_ops.rs @@ -91,6 +91,42 @@ impl RustfsAdminClient { )) } + /// Delete a bucket. Missing buckets are treated as success (idempotent). + pub async fn delete_bucket(&self, bucket: &str) -> Result<(), RustfsClientError> { + if bucket.trim().is_empty() { + return Err(RustfsClientError::RequestBuildFailed); + } + + let path = format!("/{bucket}"); + let signed = self.sign_request("DELETE", &path, "", "", None, ADMIN_SIGNING_SERVICE)?; + let host = self.host()?; + + let response = self + .http_client + .delete(format!("{}{}", self.base_url.trim_end_matches('/'), path)) + .header("x-amz-date", &signed.amz_date) + .header("x-amz-content-sha256", &signed.payload_hash) + .header("authorization", &signed.authorization) + .header("host", host) + .send() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + + if response.status().is_success() { + return Ok(()); + } + + let status = response.status(); + let (body, truncated) = RustfsClientError::limited_response_body(response).await; + if status == StatusCode::NOT_FOUND || body_mentions_not_found(&body) { + return Ok(()); + } + + Err(RustfsClientError::unexpected_status_with_limited_body( + status, &body, truncated, + )) + } + pub async fn bucket_object_lock_enabled( &self, bucket: &str, diff --git a/src/sts/tests.rs b/src/sts/tests.rs index 502a92a..bea76c4 100644 --- a/src/sts/tests.rs +++ b/src/sts/tests.rs @@ -1005,6 +1005,59 @@ async fn user_exists_limits_unexpected_error_response_body() { server.abort(); } +#[tokio::test] +async fn get_user_info_parses_comma_separated_policy_names() { + let router = Router::new().route( + USER_INFO_PATH, + get(|| async { + ( + StatusCode::OK, + r#"{"status":"enabled","policyName":"cosi-mlflow,cosi-grant-ba-1"}"#, + ) + }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + let info = client + .get_user_info("mlflow") + .await + .unwrap() + .expect("user should exist"); + assert_eq!( + info.policy_names, + vec!["cosi-mlflow".to_string(), "cosi-grant-ba-1".to_string()] + ); + assert!(client.user_exists("mlflow").await.unwrap()); + + server.abort(); +} + +#[tokio::test] +async fn get_user_info_returns_none_for_missing_user() { + let router = Router::new().route( + USER_INFO_PATH, + get(|| async { (StatusCode::NOT_FOUND, "NoSuchUser") }), + ); + + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + + let client = RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret"); + assert!(client.get_user_info("missing").await.unwrap().is_none()); + assert!(!client.user_exists("missing").await.unwrap()); + + server.abort(); +} + #[tokio::test] async fn set_user_policy_uses_single_authoritative_mapping_call() { let capture = Capture::default();