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..cb988fb 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" @@ -1651,6 +1710,7 @@ dependencies = [ "rcgen", "reqwest", "ring", + "rustfs-admin", "rustls", "rustls-pemfile", "rustls-webpki", @@ -1667,7 +1727,7 @@ dependencies = [ "tokio-rustls", "tokio-stream", "tokio-util", - "tower", + "tower 0.5.2", "tower-http", "tracing", "tracing-subscriber", @@ -1773,6 +1833,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 +1905,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 +1924,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 +2053,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2", + "socket2 0.6.1", "thiserror 2.0.17", "tokio", "tracing", @@ -1873,7 +2069,7 @@ dependencies = [ "bytes", "getrandom 0.3.4", "lru-slab", - "rand", + "rand 0.9.4", "ring", "rustc-hash", "rustls", @@ -1894,7 +2090,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2", + "socket2 0.6.1", "tracing", "windows-sys 0.60.2", ] @@ -1914,14 +2110,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 +2148,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 +2268,7 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", - "tower", + "tower 0.5.2", "tower-http", "tower-service", "url", @@ -2106,6 +2332,57 @@ version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" +[[package]] +name = "rustfs-admin" +version = "0.1.0" +dependencies = [ + "chrono", + "hex", + "hmac", + "reqwest", + "serde", + "serde_json", + "sha2", + "url", +] + +[[package]] +name = "rustfs-cosi-driver" +version = "0.1.0" +dependencies = [ + "hex", + "k8s-openapi", + "kube", + "prost", + "protoc-bin-vendored", + "rand 0.8.7", + "rustfs-admin", + "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 +2638,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 +2651,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 +2747,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 +2837,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 +2968,7 @@ dependencies = [ "mio", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.1", "tokio-macros", "windows-sys 0.61.2", ] @@ -2722,6 +3022,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 +3127,7 @@ dependencies = [ "pin-project-lite", "tokio", "tokio-util", - "tower", + "tower 0.5.2", "tower-layer", "tower-service", "tracing", @@ -2947,7 +3311,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 +3857,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..5f7853f 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ url = "2.5" shadow-rs = "1.5.0" snafu = { version = "0.8.9", features = ["futures"] } kube-leader-election = { path = "crates/leader-election" } +rustfs-admin = { path = "crates/rustfs-admin" } hostname = "0.4" # Console dependencies @@ -63,7 +64,7 @@ shadow-rs = { version = "1.5.0", features = ["build"] } unused_variables = "allow" [workspace] -members = ["crates/leader-election"] +members = ["crates/leader-election", "crates/cosi-driver", "crates/rustfs-admin"] [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..f908e9a --- /dev/null +++ b/crates/cosi-driver/Cargo.toml @@ -0,0 +1,32 @@ +[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] +rustfs-admin = { path = "../rustfs-admin" } +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" +rand = "0.8" +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..25e545a --- /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 rustfs_admin::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] rustfs_admin::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/bucket.rs b/crates/cosi-driver/src/bucket.rs new file mode 100644 index 0000000..258cc78 --- /dev/null +++ b/crates/cosi-driver/src/bucket.rs @@ -0,0 +1,146 @@ +//! Bucket create/delete helpers with safe static-bucket defaults. + +use rustfs_admin::{CreateBucketResult, RustfsAdminClient}; +use thiserror::Error; +use tracing::info; + +use crate::parameters::BackendParameters; + +#[derive(Debug, Error)] +pub enum BucketError { + #[error("rustfs admin error: {0}")] + Admin(String), + #[error( + "refusing to delete bucket `{bucket_id}`: BucketClass uses static \ + bucketName/buckets override (adoption preview); delete is skipped \ + without ownership proof" + )] + StaticBucketDeleteRefused { bucket_id: String }, + #[error("no buckets to create (buckets/bucketName empty or only *)")] + NothingToCreate, +} + +#[derive(Debug, Clone)] +pub struct CreateBucketOutcome { + pub bucket_id: String, + pub region: String, + /// True when BAC/BC supplied bucketName/buckets (static adoption preview). + #[allow(dead_code)] + pub static_override: bool, +} + +/// Dynamic path: create the unique COSI request name. +/// Static override preview: create configured bucket names; do not treat as +/// fully owned for delete (see [`delete_bucket`]). +pub async fn create_bucket( + client: &RustfsAdminClient, + params: &BackendParameters, + cosi_name: &str, +) -> Result { + let static_override = params.bucket_name.is_some() || params.buckets.is_some(); + let targets = if static_override { + let list = params.buckets_to_create(cosi_name); + if list.is_empty() { + return Err(BucketError::NothingToCreate); + } + list + } else { + vec![cosi_name.to_string()] + }; + let bucket_id = if static_override { + params.primary_bucket_id(cosi_name) + } else { + cosi_name.to_string() + }; + + for bucket in &targets { + info!( + bucket = %bucket, + cosi_name = %cosi_name, + static_override, + "creating bucket" + ); + match client + .create_bucket(bucket, Some(params.region.as_str()), false) + .await + .map_err(|err| BucketError::Admin(err.to_string()))? + { + CreateBucketResult::Created | CreateBucketResult::AlreadyExists => {} + } + } + + Ok(CreateBucketOutcome { + bucket_id, + region: params.region.clone(), + static_override, + }) +} + +/// Delete only dynamically owned buckets. +/// +/// When `bucketName`/`buckets` is set on the class, refuse delete (FailedPrecondition +/// semantics at the gRPC layer) so shared/static buckets are not destroyed. +pub async fn delete_bucket( + client: &RustfsAdminClient, + params: &BackendParameters, + bucket_id: &str, +) -> Result<(), BucketError> { + if params.bucket_name.is_some() || params.buckets.is_some() { + return Err(BucketError::StaticBucketDeleteRefused { + bucket_id: bucket_id.to_string(), + }); + } + + info!(bucket = %bucket_id, "deleting dynamically owned bucket"); + client + .delete_bucket(bucket_id) + .await + .map_err(|err| BucketError::Admin(err.to_string()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + fn params(extra: &[(&str, &str)]) -> BackendParameters { + let mut map = HashMap::new(); + map.insert("endpoint".into(), "http://rustfs".into()); + map.insert("objectStoreUserSecretName".into(), "s".into()); + map.insert("objectStoreUserSecretNamespace".into(), "ns".into()); + for (k, v) in extra { + map.insert((*k).into(), (*v).into()); + } + BackendParameters::from_map(&map).unwrap() + } + + #[test] + fn dynamic_create_targets_cosi_name_only() { + let p = params(&[]); + assert!(p.bucket_name.is_none()); + assert!(p.buckets.is_none()); + // create_bucket itself needs admin client; unit-check targeting logic here + let targets = if p.bucket_name.is_some() || p.buckets.is_some() { + p.buckets_to_create("bc-1") + } else { + vec!["bc-1".to_string()] + }; + assert_eq!(targets, vec!["bc-1".to_string()]); + } + + #[test] + fn static_override_refuses_delete_without_admin() { + let p = params(&[("bucketName", "shared-mlflow")]); + let err = match ( + p.bucket_name.is_some() || p.buckets.is_some(), + "shared-mlflow", + ) { + (true, id) => BucketError::StaticBucketDeleteRefused { + bucket_id: id.to_string(), + }, + _ => unreachable!(), + }; + assert!(matches!(err, BucketError::StaticBucketDeleteRefused { .. })); + } +} diff --git a/crates/cosi-driver/src/credentials.rs b/crates/cosi-driver/src/credentials.rs new file mode 100644 index 0000000..b779cac --- /dev/null +++ b/crates/cosi-driver/src/credentials.rs @@ -0,0 +1,212 @@ +//! Durable random S3 credentials stored in Kubernetes Secrets. + +use k8s_openapi::api::core::v1::Secret; +use k8s_openapi::{ByteString, apimachinery::pkg::apis::meta::v1::ObjectMeta}; +use kube::api::{Patch, PatchParams}; +use kube::{Api, Client, Error as KubeError}; +use rand::{Rng, distributions::Alphanumeric}; +use sha2::{Digest, Sha256}; +use thiserror::Error; + +use crate::parameters::sanitize_policy_fragment; + +#[derive(Debug, Error)] +pub enum CredentialStoreError { + #[error("failed to read credential Secret {namespace}/{name}: {source}")] + Lookup { + namespace: String, + name: String, + #[source] + source: Box, + }, + #[error("failed to persist credential Secret {namespace}/{name}: {source}")] + Persist { + namespace: String, + name: String, + #[source] + source: Box, + }, + #[error("credential Secret {namespace}/{name} missing key `{key}`")] + MissingKey { + namespace: String, + name: String, + key: &'static str, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StoredCredentials { + pub access_key: String, + pub secret_key: String, + pub secret_name: String, +} + +pub fn credential_secret_name(grant_name: &str) -> String { + format!("cosi-cred-{}", sanitize_policy_fragment(grant_name)) +} + +pub fn access_key_hash(access_key: &str) -> String { + hex::encode(Sha256::digest(access_key.as_bytes())) +} + +pub fn random_secret_key(len: usize) -> String { + rand::thread_rng() + .sample_iter(&Alphanumeric) + .take(len) + .map(char::from) + .collect() +} + +fn decode_secret_key(secret: &Secret, key: &'static str) -> Result { + let namespace = secret.metadata.namespace.clone().unwrap_or_default(); + let name = secret.metadata.name.clone().unwrap_or_default(); + let value = secret.data.as_ref().and_then(|data| data.get(key)).ok_or( + CredentialStoreError::MissingKey { + namespace: namespace.clone(), + name: name.clone(), + key, + }, + )?; + String::from_utf8(value.0.clone()).map_err(|_| CredentialStoreError::MissingKey { + namespace, + name, + key, + }) +} + +/// Load existing credentials for a grant, or create a new random secret and persist it. +pub async fn load_or_create_credentials( + kube: &Client, + namespace: &str, + grant_name: &str, + access_key: &str, +) -> Result { + let secret_name = credential_secret_name(grant_name); + let api: Api = Api::namespaced(kube.clone(), namespace); + + match api.get(&secret_name).await { + Ok(existing) => { + let stored_access = decode_secret_key(&existing, "accessKeyID") + .or_else(|_| decode_secret_key(&existing, "AWS_ACCESS_KEY_ID"))?; + let secret_key = decode_secret_key(&existing, "accessSecretKey") + .or_else(|_| decode_secret_key(&existing, "AWS_SECRET_ACCESS_KEY"))?; + if stored_access != access_key { + // Access key changed for this grant — rotate secret material under same Secret. + let secret_key = random_secret_key(40); + persist_credentials(kube, namespace, &secret_name, access_key, &secret_key).await?; + return Ok(StoredCredentials { + access_key: access_key.to_string(), + secret_key, + secret_name, + }); + } + Ok(StoredCredentials { + access_key: stored_access, + secret_key, + secret_name, + }) + } + Err(KubeError::Api(err)) if err.code == 404 => { + let secret_key = random_secret_key(40); + persist_credentials(kube, namespace, &secret_name, access_key, &secret_key).await?; + Ok(StoredCredentials { + access_key: access_key.to_string(), + secret_key, + secret_name, + }) + } + Err(source) => Err(CredentialStoreError::Lookup { + namespace: namespace.to_string(), + name: secret_name, + source: Box::new(source), + }), + } +} + +async fn persist_credentials( + kube: &Client, + namespace: &str, + secret_name: &str, + access_key: &str, + secret_key: &str, +) -> Result<(), CredentialStoreError> { + let api: Api = Api::namespaced(kube.clone(), namespace); + let mut data = std::collections::BTreeMap::new(); + data.insert( + "accessKeyID".to_string(), + ByteString(access_key.as_bytes().to_vec()), + ); + data.insert( + "accessSecretKey".to_string(), + ByteString(secret_key.as_bytes().to_vec()), + ); + data.insert( + "AWS_ACCESS_KEY_ID".to_string(), + ByteString(access_key.as_bytes().to_vec()), + ); + data.insert( + "AWS_SECRET_ACCESS_KEY".to_string(), + ByteString(secret_key.as_bytes().to_vec()), + ); + + let secret = Secret { + metadata: ObjectMeta { + name: Some(secret_name.to_string()), + namespace: Some(namespace.to_string()), + labels: Some( + [ + ( + "app.kubernetes.io/name".to_string(), + "rustfs-cosi-driver".to_string(), + ), + ( + "rustfs.objectstorage.k8s.io/grant".to_string(), + sanitize_policy_fragment( + secret_name + .strip_prefix("cosi-cred-") + .unwrap_or(secret_name), + ), + ), + ] + .into_iter() + .collect(), + ), + ..ObjectMeta::default() + }, + type_: Some("Opaque".to_string()), + data: Some(data), + ..Secret::default() + }; + + api.patch( + secret_name, + &PatchParams::apply("rustfs-cosi-driver").force(), + &Patch::Apply(&secret), + ) + .await + .map_err(|source| CredentialStoreError::Persist { + namespace: namespace.to_string(), + name: secret_name.to_string(), + source: Box::new(source), + })?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::{access_key_hash, credential_secret_name, random_secret_key}; + + #[test] + fn secret_names_are_stable() { + assert_eq!(credential_secret_name("ba-abc.def"), "cosi-cred-ba-abc-def"); + } + + #[test] + fn random_secrets_are_not_derived_from_access_key() { + let a = random_secret_key(40); + let b = random_secret_key(40); + assert_ne!(a, b); + assert_ne!(a, access_key_hash("mlflow")); + assert!(a.len() >= 40); + } +} diff --git a/crates/cosi-driver/src/driver.rs b/crates/cosi-driver/src/driver.rs new file mode 100644 index 0000000..459c7f5 --- /dev/null +++ b/crates/cosi-driver/src/driver.rs @@ -0,0 +1,202 @@ +//! COSI Identity + Provisioner gRPC adapters (thin; logic lives in grant/bucket). + +use std::collections::HashMap; + +use kube::Client; +use tonic::{Request, Response, Status}; +use tracing::error; + +use crate::backend::{BackendError, admin_client_from_params}; +use crate::bucket::{self, BucketError}; +use crate::grant::{self, GrantError}; +use crate::parameters::{BackendParameters, DRIVER_NAME}; +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_bucket(err: BucketError) -> Status { + match err { + BucketError::StaticBucketDeleteRefused { .. } => { + Status::failed_precondition(err.to_string()) + } + BucketError::NothingToCreate => Status::invalid_argument(err.to_string()), + BucketError::Admin(msg) => { + error!(error = %msg, "rustfs admin error"); + Status::internal(msg) + } + } +} + +fn map_grant(err: GrantError) -> Status { + if err.is_conflict() { + return Status::already_exists(err.to_string()); + } + match err { + GrantError::MissingExternalPolicy(_) => Status::failed_precondition(err.to_string()), + GrantError::Credentials(e) => { + error!(error = %e, "credential store error"); + Status::internal(e.to_string()) + } + GrantError::Ownership(e) => { + error!(error = %e, "ownership store error"); + Status::internal(e.to_string()) + } + GrantError::Admin(msg) => { + error!(error = %msg, "rustfs admin error"); + Status::internal(msg) + } + other => Status::internal(other.to_string()), + } +} + +#[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 outcome = bucket::create_bucket(&client, ¶ms, &req.name) + .await + .map_err(map_bucket)?; + + Ok(Response::new(DriverCreateBucketResponse { + bucket_id: outcome.bucket_id, + bucket_info: Some(Protocol { + r#type: Some(crate::proto::cosi::v1alpha1::protocol::Type::S3(S3 { + region: outcome.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)?; + + bucket::delete_bucket(&client, ¶ms, &req.bucket_id) + .await + .map_err(map_bucket)?; + 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 result = + grant::grant_bucket_access(&self.kube, &client, ¶ms, &req.name, &req.bucket_id) + .await + .map_err(map_grant)?; + + let mut credentials = HashMap::new(); + credentials.insert( + "s3".to_string(), + CredentialDetails { + secrets: result.secrets, + }, + ); + + Ok(Response::new(DriverGrantBucketAccessResponse { + account_id: result.account_id, + 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)?; + + grant::revoke_bucket_access(&self.kube, &client, &req.account_id) + .await + .map_err(map_grant)?; + Ok(Response::new(DriverRevokeBucketAccessResponse {})) + } +} diff --git a/crates/cosi-driver/src/grant.rs b/crates/cosi-driver/src/grant.rs new file mode 100644 index 0000000..152ebb0 --- /dev/null +++ b/crates/cosi-driver/src/grant.rs @@ -0,0 +1,416 @@ +//! Grant provisioning state machine (durable secrets + CAS ownership). + +use std::collections::HashMap; + +use kube::Client; +use rustfs_admin::RustfsAdminClient; +use thiserror::Error; +use tracing::info; + +use crate::credentials::{ + CredentialStoreError, StoredCredentials, access_key_hash, credential_secret_name, + load_or_create_credentials, +}; +use crate::ownership::{GrantOwnershipState, OwnershipError, OwnershipStore}; +use crate::parameters::{BackendParameters, bucket_policy_document_for, grant_policy_name}; + +#[derive(Debug, Error)] +pub enum GrantError { + #[error(transparent)] + Ownership(#[from] OwnershipError), + #[error(transparent)] + Credentials(#[from] CredentialStoreError), + #[error("rustfs admin error: {0}")] + Admin(String), + #[error( + "preferredAccessKey `{account_id}` is already bound to another BucketAccess; \ + omit preferredAccessKey or choose a unique value" + )] + AccountConflict { account_id: String }, + #[error( + "access key `{account_id}` exists in RustFS without a matching Ready ownership \ + proof for grant `{grant_name}`" + )] + OrphanUserConflict { + account_id: String, + grant_name: String, + }, + #[error("external policy `{0}` does not exist")] + MissingExternalPolicy(String), +} + +impl GrantError { + pub fn is_conflict(&self) -> bool { + matches!( + self, + Self::AccountConflict { .. } + | Self::OrphanUserConflict { .. } + | Self::Ownership(OwnershipError::AccountConflict { .. }) + ) + } +} + +#[derive(Debug, Clone)] +pub struct GrantResult { + pub account_id: String, + #[allow(dead_code)] + pub secret_key: String, + pub secrets: HashMap, +} + +pub fn state_namespace(kube: &Client) -> String { + std::env::var("POD_NAMESPACE") + .or_else(|_| std::env::var("COSI_STATE_NAMESPACE")) + .unwrap_or_else(|_| kube.default_namespace().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 +} + +/// Resolve the policy name attached to this grant. +/// +/// - Generated policies are unique per grant (`cosi-pol-{grant}`). +/// - External BAC `policy` names are referenced only (never replaced). +pub fn resolve_policy_name(params: &BackendParameters, grant_name: &str) -> (String, bool) { + match params.policy.as_deref() { + Some(external) => (external.to_string(), true), + None => (grant_policy_name(grant_name), false), + } +} + +/// Whether the driver should call `add_canned_policy` for this grant. +/// +/// External BAC policy names are validate-only — never overwritten. +pub fn should_write_canned_policy(external_policy: bool) -> bool { + !external_policy +} + +async fn attach_policies( + client: &RustfsAdminClient, + access_key: &str, + grant_name: &str, + params: &BackendParameters, + policy_buckets: &[String], +) -> Result<(), GrantError> { + let (policy_name, external) = resolve_policy_name(params, grant_name); + if external { + // Validate-only: never add_canned_policy / replace. + client + .get_canned_policy(&policy_name) + .await + .map_err(|err| { + let msg = err.to_string(); + if msg.contains("not found") || msg.contains("NoSuch") { + GrantError::MissingExternalPolicy(policy_name.clone()) + } else { + GrantError::Admin(msg) + } + })?; + } else { + debug_assert!(should_write_canned_policy(false)); + let doc = bucket_policy_document_for(policy_buckets); + client + .add_canned_policy(&policy_name, &doc) + .await + .map_err(|err| GrantError::Admin(err.to_string()))?; + } + + client + .set_user_policy(access_key, &[policy_name]) + .await + .map_err(|err| GrantError::Admin(err.to_string()))?; + Ok(()) +} + +fn map_ownership_conflict(err: OwnershipError, account_id: &str) -> GrantError { + match err { + OwnershipError::AccountConflict { account_id, .. } => { + GrantError::AccountConflict { account_id } + } + other => { + let _ = account_id; + GrantError::Ownership(other) + } + } +} + +/// Provision or resume a grant with durable credentials and CAS ownership. +/// +/// Flow: PendingCreate (CAS) → durable Secret → add_user + policies → Ready. +/// Orphan RustFS users (no matching proof) are not adopted. +pub async fn grant_bucket_access( + kube: &Client, + client: &RustfsAdminClient, + params: &BackendParameters, + grant_name: &str, + bucket_id: &str, +) -> Result { + let namespace = state_namespace(kube); + let access_key = params + .preferred_access_key + .clone() + .unwrap_or_else(|| grant_name.to_string()); + let policy_buckets = params.buckets_for_policy(bucket_id); + let ak_hash = access_key_hash(&access_key); + let cred_name = credential_secret_name(grant_name); + let store = OwnershipStore::new(kube.clone(), namespace.clone()); + + let existing_proof = store.get(grant_name).await?; + let user_info = client + .get_user_info(&access_key) + .await + .map_err(|err| GrantError::Admin(err.to_string()))?; + + // Refuse silent adopt of orphan / foreign RustFS users. + if user_info.is_some() { + match &existing_proof { + Some(proof) + if proof.account_id == access_key + && matches!( + proof.state, + GrantOwnershipState::PendingCreate | GrantOwnershipState::Ready + ) => + { + // Crash resume or idempotent retry for this grant. + } + _ => { + if let Some(owner) = store.find_by_account_id(&access_key).await? { + if owner != grant_name { + return Err(GrantError::AccountConflict { + account_id: access_key, + }); + } + } else { + return Err(GrantError::OrphanUserConflict { + account_id: access_key, + grant_name: grant_name.to_string(), + }); + } + } + } + } + + let proof = store + .begin_or_resume(grant_name, &access_key, &ak_hash, &cred_name) + .await + .map_err(|err| map_ownership_conflict(err, &access_key))?; + + let StoredCredentials { + access_key, + secret_key, + secret_name: _, + } = load_or_create_credentials(kube, &namespace, grant_name, &proof.account_id).await?; + + info!( + grant = %grant_name, + account = %access_key, + state = ?proof.state, + buckets = %policy_buckets.join(","), + "granting bucket access" + ); + + match client + .get_user_info(&access_key) + .await + .map_err(|err| GrantError::Admin(err.to_string()))? + { + Some(_) => { + attach_policies(client, &access_key, grant_name, params, &policy_buckets).await?; + } + None => { + client + .add_user(&access_key, &secret_key) + .await + .map_err(|err| GrantError::Admin(err.to_string()))?; + attach_policies(client, &access_key, grant_name, params, &policy_buckets).await?; + } + } + + if proof.state != GrantOwnershipState::Ready { + store.mark_ready(grant_name).await?; + } + + Ok(GrantResult { + account_id: access_key.clone(), + secret_key: secret_key.clone(), + secrets: credential_map(&access_key, &secret_key, params, &policy_buckets), + }) +} + +pub async fn revoke_bucket_access( + kube: &Client, + client: &RustfsAdminClient, + account_id: &str, +) -> Result<(), GrantError> { + let namespace = state_namespace(kube); + let store = OwnershipStore::new(kube.clone(), namespace); + + if let Some(grant) = store.find_by_account_id(account_id).await? { + store.remove(&grant).await?; + } + + client + .remove_user(account_id) + .await + .map_err(|err| GrantError::Admin(err.to_string()))?; + Ok(()) +} + +/// Pure helpers for unit tests (no kube / RustFS). +#[cfg(test)] +pub mod logic { + use super::*; + + /// Whether an existing RustFS user may be resumed for this grant. + pub fn may_resume_existing_user( + grant_name: &str, + access_key: &str, + proof: Option<&crate::ownership::GrantOwnershipProof>, + account_owner: Option<&str>, + ) -> Result<(), GrantError> { + match proof { + Some(p) + if p.account_id == access_key + && matches!( + p.state, + GrantOwnershipState::PendingCreate | GrantOwnershipState::Ready + ) => + { + Ok(()) + } + _ => { + if let Some(owner) = account_owner { + if owner != grant_name { + return Err(GrantError::AccountConflict { + account_id: access_key.to_string(), + }); + } + Ok(()) + } else { + Err(GrantError::OrphanUserConflict { + account_id: access_key.to_string(), + grant_name: grant_name.to_string(), + }) + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::logic::may_resume_existing_user; + use super::{resolve_policy_name, should_write_canned_policy}; + use crate::ownership::{GrantOwnershipProof, GrantOwnershipState}; + use crate::parameters::BackendParameters; + use std::collections::HashMap; + + fn params_with_policy(policy: Option<&str>) -> BackendParameters { + let mut map = HashMap::new(); + map.insert("endpoint".into(), "http://rustfs".into()); + map.insert("objectStoreUserSecretName".into(), "s".into()); + map.insert("objectStoreUserSecretNamespace".into(), "ns".into()); + if let Some(p) = policy { + map.insert("policy".into(), p.into()); + } + BackendParameters::from_map(&map).unwrap() + } + + #[test] + fn generated_policy_is_unique_per_grant() { + let params = params_with_policy(None); + let (name, external) = resolve_policy_name(¶ms, "ba-1"); + assert!(!external); + assert_eq!(name, "cosi-pol-ba-1"); + let (other, _) = resolve_policy_name(¶ms, "ba-2"); + assert_ne!(name, other); + } + + #[test] + fn external_policy_is_reference_only() { + let params = params_with_policy(Some("shared-readonly")); + let (name, external) = resolve_policy_name(¶ms, "ba-1"); + assert!(external); + assert_eq!(name, "shared-readonly"); + } + + #[test] + fn pending_create_resumes_after_partial_failure() { + let proof = GrantOwnershipProof { + grant_name: "ba-1".into(), + account_id: "mlflow".into(), + access_key_hash: "h".into(), + cred_secret_name: "cosi-cred-ba-1".into(), + state: GrantOwnershipState::PendingCreate, + }; + assert!(may_resume_existing_user("ba-1", "mlflow", Some(&proof), Some("ba-1")).is_ok()); + } + + #[test] + fn concurrent_preferred_key_conflicts() { + let err = may_resume_existing_user("ba-2", "mlflow", None, Some("ba-1")).unwrap_err(); + assert!(matches!(err, super::GrantError::AccountConflict { .. })); + assert!(err.is_conflict()); + } + + #[test] + fn orphan_user_is_not_adopted() { + let err = may_resume_existing_user("ba-1", "mlflow", None, None).unwrap_err(); + assert!(matches!(err, super::GrantError::OrphanUserConflict { .. })); + } + + #[test] + fn ready_proof_allows_idempotent_retry() { + let proof = GrantOwnershipProof { + grant_name: "ba-1".into(), + account_id: "mlflow".into(), + access_key_hash: "h".into(), + cred_secret_name: "cosi-cred-ba-1".into(), + state: GrantOwnershipState::Ready, + }; + assert!(may_resume_existing_user("ba-1", "mlflow", Some(&proof), Some("ba-1")).is_ok()); + } + + #[test] + fn external_policy_is_never_written() { + let params = params_with_policy(Some("shared-readonly")); + let (_, external) = resolve_policy_name(¶ms, "ba-1"); + assert!(!should_write_canned_policy(external)); + assert!(should_write_canned_policy(false)); + } + + #[test] + fn retry_reuses_stable_credential_secret_name() { + use crate::credentials::credential_secret_name; + assert_eq!( + credential_secret_name("ba-1"), + credential_secret_name("ba-1") + ); + assert_ne!( + credential_secret_name("ba-1"), + credential_secret_name("ba-2") + ); + } +} diff --git a/crates/cosi-driver/src/main.rs b/crates/cosi-driver/src/main.rs new file mode 100644 index 0000000..87897c5 --- /dev/null +++ b/crates/cosi-driver/src/main.rs @@ -0,0 +1,102 @@ +//! RustFS COSI driver — serves Identity + Provisioner on a Unix socket. + +mod backend; +mod bucket; +mod credentials; +mod driver; +mod grant; +mod ownership; +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/ownership.rs b/crates/cosi-driver/src/ownership.rs new file mode 100644 index 0000000..64ea9b4 --- /dev/null +++ b/crates/cosi-driver/src/ownership.rs @@ -0,0 +1,281 @@ +//! Crash-safe grant ownership checkpoints (PendingCreate → Ready). +//! +//! Mirrors the Tenant user CAS pattern: persist PendingCreate before mutating +//! RustFS, then promote to Ready after success. Concurrent claims on the same +//! preferred access key conflict instead of silently adopting. + +use std::collections::BTreeMap; + +use k8s_openapi::api::core::v1::ConfigMap; +use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta; +use kube::Client; +use kube::api::{Api, PostParams}; +use serde::{Deserialize, Serialize}; +use thiserror::Error; +use tracing::{info, warn}; + +const OWNERSHIP_CM_NAME: &str = "rustfs-cosi-ownership"; +const PROOF_PREFIX: &str = "grant."; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum GrantOwnershipState { + PendingCreate, + Ready, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct GrantOwnershipProof { + pub grant_name: String, + pub account_id: String, + pub access_key_hash: String, + pub cred_secret_name: String, + pub state: GrantOwnershipState, +} + +#[derive(Debug, Error)] +pub enum OwnershipError { + #[error("kubernetes error: {0}")] + Kube(#[from] kube::Error), + #[error("invalid ownership proof for grant `{grant}`: {detail}")] + InvalidProof { grant: String, detail: String }, + #[error( + "ownership conflict: access key `{account_id}` is claimed by grant `{owner}`, \ + not `{requester}`" + )] + AccountConflict { + account_id: String, + owner: String, + requester: String, + }, + #[error("CAS conflict writing ownership for grant `{0}` (retry)")] + CasConflict(String), +} + +#[derive(Clone)] +pub struct OwnershipStore { + api: Api, +} + +impl OwnershipStore { + pub fn new(client: Client, namespace: String) -> Self { + Self { + api: Api::namespaced(client, &namespace), + } + } + + fn proof_key(grant_name: &str) -> String { + format!("{PROOF_PREFIX}{grant_name}") + } + + pub async fn get( + &self, + grant_name: &str, + ) -> Result, OwnershipError> { + let Some(cm) = self.get_cm().await? else { + return Ok(None); + }; + let Some(data) = cm.data.as_ref() else { + return Ok(None); + }; + let Some(raw) = data.get(&Self::proof_key(grant_name)) else { + return Ok(None); + }; + let proof: GrantOwnershipProof = + serde_json::from_str(raw).map_err(|err| OwnershipError::InvalidProof { + grant: grant_name.to_string(), + detail: err.to_string(), + })?; + Ok(Some(proof)) + } + + pub async fn find_by_account_id( + &self, + account_id: &str, + ) -> Result, OwnershipError> { + let Some(cm) = self.get_cm().await? else { + return Ok(None); + }; + let Some(data) = cm.data.as_ref() else { + return Ok(None); + }; + for (key, raw) in data { + if !key.starts_with(PROOF_PREFIX) { + continue; + } + let proof: GrantOwnershipProof = match serde_json::from_str(raw) { + Ok(p) => p, + Err(err) => { + warn!(key = %key, error = %err, "skipping corrupt ownership proof"); + continue; + } + }; + if proof.account_id == account_id { + return Ok(Some(proof.grant_name)); + } + } + Ok(None) + } + + /// CAS: create PendingCreate if absent, or resume existing proof for this grant. + /// Conflicts if another grant already owns the same account_id. + pub async fn begin_or_resume( + &self, + grant_name: &str, + account_id: &str, + access_key_hash: &str, + cred_secret_name: &str, + ) -> Result { + if let Some(owner) = self.find_by_account_id(account_id).await? + && owner != grant_name + { + return Err(OwnershipError::AccountConflict { + account_id: account_id.to_string(), + owner, + requester: grant_name.to_string(), + }); + } + + if let Some(existing) = self.get(grant_name).await? { + if existing.account_id != account_id { + return Err(OwnershipError::AccountConflict { + account_id: account_id.to_string(), + owner: existing.grant_name, + requester: grant_name.to_string(), + }); + } + return Ok(existing); + } + + let proof = GrantOwnershipProof { + grant_name: grant_name.to_string(), + account_id: account_id.to_string(), + access_key_hash: access_key_hash.to_string(), + cred_secret_name: cred_secret_name.to_string(), + state: GrantOwnershipState::PendingCreate, + }; + self.cas_put(&proof).await?; + info!( + grant = %grant_name, + account = %account_id, + "recorded PendingCreate ownership checkpoint" + ); + Ok(proof) + } + + pub async fn mark_ready(&self, grant_name: &str) -> Result<(), OwnershipError> { + let Some(mut proof) = self.get(grant_name).await? else { + return Err(OwnershipError::InvalidProof { + grant: grant_name.to_string(), + detail: "missing proof when promoting to Ready".into(), + }); + }; + proof.state = GrantOwnershipState::Ready; + self.cas_put(&proof).await?; + info!(grant = %grant_name, "promoted grant ownership to Ready"); + Ok(()) + } + + pub async fn remove(&self, grant_name: &str) -> Result<(), OwnershipError> { + let key = Self::proof_key(grant_name); + for _ in 0..8 { + let Some(mut cm) = self.get_cm().await? else { + return Ok(()); + }; + let rv = cm.metadata.resource_version.clone(); + let mut data = cm.data.take().unwrap_or_default(); + if data.remove(&key).is_none() { + return Ok(()); + } + cm.data = Some(data); + cm.metadata.resource_version = rv; + match self + .api + .replace(OWNERSHIP_CM_NAME, &PostParams::default(), &cm) + .await + { + Ok(_) => return Ok(()), + Err(kube::Error::Api(err)) if err.code == 409 => continue, + Err(err) => return Err(err.into()), + } + } + Err(OwnershipError::CasConflict(grant_name.to_string())) + } + + async fn get_cm(&self) -> Result, OwnershipError> { + match self.api.get(OWNERSHIP_CM_NAME).await { + Ok(cm) => Ok(Some(cm)), + Err(kube::Error::Api(err)) if err.code == 404 => Ok(None), + Err(err) => Err(err.into()), + } + } + + async fn ensure_cm(&self) -> Result { + if let Some(cm) = self.get_cm().await? { + return Ok(cm); + } + let cm = ConfigMap { + metadata: ObjectMeta { + name: Some(OWNERSHIP_CM_NAME.to_string()), + ..Default::default() + }, + data: Some(BTreeMap::new()), + ..Default::default() + }; + match self.api.create(&PostParams::default(), &cm).await { + Ok(created) => Ok(created), + Err(kube::Error::Api(err)) if err.code == 409 => self + .get_cm() + .await? + .ok_or_else(|| OwnershipError::CasConflict(OWNERSHIP_CM_NAME.to_string())), + Err(err) => Err(err.into()), + } + } + + async fn cas_put(&self, proof: &GrantOwnershipProof) -> Result<(), OwnershipError> { + let key = Self::proof_key(&proof.grant_name); + let value = serde_json::to_string(proof).map_err(|err| OwnershipError::InvalidProof { + grant: proof.grant_name.clone(), + detail: err.to_string(), + })?; + + for _ in 0..8 { + let mut cm = self.ensure_cm().await?; + let rv = cm.metadata.resource_version.clone(); + let mut data = cm.data.take().unwrap_or_default(); + data.insert(key.clone(), value.clone()); + cm.data = Some(data); + cm.metadata.resource_version = rv; + match self + .api + .replace(OWNERSHIP_CM_NAME, &PostParams::default(), &cm) + .await + { + Ok(_) => return Ok(()), + Err(kube::Error::Api(err)) if err.code == 409 => continue, + Err(err) => return Err(err.into()), + } + } + Err(OwnershipError::CasConflict(proof.grant_name.clone())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn proof_roundtrip() { + let proof = GrantOwnershipProof { + grant_name: "ba-1".into(), + account_id: "mlflow".into(), + access_key_hash: "abc".into(), + cred_secret_name: "cosi-cred-ba-1".into(), + state: GrantOwnershipState::PendingCreate, + }; + let raw = serde_json::to_string(&proof).unwrap(); + let back: GrantOwnershipProof = serde_json::from_str(&raw).unwrap(); + assert_eq!(proof, back); + } +} diff --git a/crates/cosi-driver/src/parameters.rs b/crates/cosi-driver/src/parameters.rs new file mode 100644 index 0000000..06cfa7a --- /dev/null +++ b/crates/cosi-driver/src/parameters.rs @@ -0,0 +1,163 @@ +//! 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, + /// Optional existing canned policy name. Referenced only — never replaced. + pub policy: Option, + pub tls_ca_configmap_name: Option, + pub tls_ca_configmap_namespace: Option, + /// Static/adoption preview: preferred S3 bucket name (overrides COSI name). + /// + /// Buckets created via this override are not deleted by DriverDeleteBucket + /// without ownership proof (PR A safe default). + pub bucket_name: Option, + /// Static/adoption preview: comma-separated bucket list (`*` = full access). + pub buckets: Option, + /// Preferred S3 access-key / account name for GrantBucketAccess. + /// + /// Must be unique per BucketAccess. Reusing the same value across claims is + /// rejected (ownership conflict). 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() +} + +/// Unique generated policy name per COSI grant (never shared across grants). +pub fn grant_policy_name(grant_name: &str) -> String { + format!("cosi-pol-{}", sanitize_policy_fragment(grant_name)) +} + +#[cfg(test)] +mod tests { + use super::grant_policy_name; + + #[test] + fn grant_policy_name_sanitizes() { + assert_eq!( + grant_policy_name("ba-81733d1a-ac7a-4759-96f3-fbcc07c0cee9"), + "cosi-pol-ba-81733d1a-ac7a-4759-96f3-fbcc07c0cee9" + ); + assert_eq!(grant_policy_name("ba/weird.name"), "cosi-pol-ba-weird-name"); + } +} diff --git a/crates/rustfs-admin/Cargo.toml b/crates/rustfs-admin/Cargo.toml new file mode 100644 index 0000000..7b27669 --- /dev/null +++ b/crates/rustfs-admin/Cargo.toml @@ -0,0 +1,30 @@ +# Copyright 2025 RustFS Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +[package] +name = "rustfs-admin" +version = "0.1.0" +edition = "2024" +license = "Apache-2.0" +publish = false + +[dependencies] +chrono = "0.4" +hex = "0.4" +hmac = "0.12" +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0" +sha2 = "0.10" +url = "2.5" diff --git a/src/sts/admin_ops.rs b/crates/rustfs-admin/src/admin_ops.rs similarity index 66% rename from src/sts/admin_ops.rs rename to crates/rustfs-admin/src/admin_ops.rs index 593d1a5..34df5b4 100644 --- a/src/sts/admin_ops.rs +++ b/crates/rustfs-admin/src/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,27 @@ impl RustfsAdminClient { .map_err(|_| RustfsClientError::RequestFailed)?; if response.status().is_success() { - return Ok(true); + let body = response + .text() + .await + .map_err(|_| RustfsClientError::RequestFailed)?; + // Existence probes (and some test fixtures) return an empty 200 body. + // Treat any successful response as "user exists"; parse policies when present. + if body.trim().is_empty() { + return Ok(Some(RustfsUserInfo { + policy_names: Vec::new(), + })); + } + let parsed: Value = serde_json::from_str(&body).unwrap_or(Value::Null); + 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 +238,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 +289,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/core_ops.rs b/crates/rustfs-admin/src/core_ops.rs similarity index 100% rename from src/sts/core_ops.rs rename to crates/rustfs-admin/src/core_ops.rs diff --git a/src/sts/helpers.rs b/crates/rustfs-admin/src/helpers.rs similarity index 73% rename from src/sts/helpers.rs rename to crates/rustfs-admin/src/helpers.rs index b0975bd..ec579d9 100644 --- a/src/sts/helpers.rs +++ b/crates/rustfs-admin/src/helpers.rs @@ -13,61 +13,13 @@ // limitations under the License. //! Internal helper duties: shared credential parsing, signature/hash utilities, and parsers. -use std::collections::BTreeMap; - use hmac::{Hmac, Mac}; -use k8s_openapi::ByteString; use reqwest::StatusCode; use serde_json::Value; use sha2::{Digest, Sha256}; use url::form_urlencoded; -use crate::Tenant; -use crate::sts::types::StsAssumeRoleCredentials; - -use super::{RustfsClientError, RustfsCredentials}; - -pub(super) fn extract_credentials( - data: Option<&BTreeMap>, -) -> Result { - let secret_data = data.ok_or(RustfsClientError::TenantSecretLookupFailed)?; - - Ok(RustfsCredentials { - access_key: get_secret_value(secret_data, "accesskey")?, - secret_key: get_secret_value(secret_data, "secretkey")?, - }) -} - -pub(super) fn tenant_tls_enabled(tenant: &Tenant) -> bool { - tenant.spec.tls.as_ref().is_some_and(|tls| tls.is_enabled()) -} - -pub(super) fn tenant_tls_client_certificate_required(tenant: &Tenant) -> bool { - tenant - .status - .as_ref() - .and_then(|status| status.certificates.tls.as_ref()) - .and_then(|tls| tls.client_ca_secret_ref.as_ref()) - .is_some() -} - -pub(super) fn get_secret_value( - data: &BTreeMap, - field: &'static str, -) -> Result { - let raw = data - .get(field) - .ok_or(RustfsClientError::MissingCredentialKey { key: field })?; - - let value = String::from_utf8(raw.0.clone()) - .map_err(|_| RustfsClientError::InvalidCredentialValue { key: field })?; - - if value.is_empty() { - return Err(RustfsClientError::EmptyCredentialValue { key: field }); - } - - Ok(value) -} +use super::RustfsClientError; /// Encode an `application/x-www-form-urlencoded` request body. pub(super) fn build_form_body(params: &[(&str, &str)]) -> String { @@ -199,20 +151,6 @@ pub(super) fn derive_signing_key( hmac_sha256(&k_service, "aws4_request") } -pub(super) fn parse_assume_role_response(body: &str) -> Option { - let access_key_id = extract_xml_tag(body, "AccessKeyId")?; - let secret_access_key = extract_xml_tag(body, "SecretAccessKey")?; - let session_token = extract_xml_tag(body, "SessionToken")?; - let expiration = extract_xml_tag(body, "Expiration")?; - - Some(StsAssumeRoleCredentials { - access_key_id, - secret_access_key, - session_token, - expiration, - }) -} - pub(super) fn extract_xml_tag(document: &str, tag: &str) -> Option { let open = format!("<{tag}>"); let close = format!(""); diff --git a/crates/rustfs-admin/src/lib.rs b/crates/rustfs-admin/src/lib.rs new file mode 100644 index 0000000..d55b4b9 --- /dev/null +++ b/crates/rustfs-admin/src/lib.rs @@ -0,0 +1,512 @@ +// Copyright 2025 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::{collections::BTreeMap, time::Duration}; + +use reqwest::{Certificate, Client as HttpClient, Response, StatusCode}; + +/// admin_ops: tenant admin operations (user/policy APIs). +#[path = "admin_ops.rs"] +mod admin_ops; +/// core_ops: shared request signing/dispatch internals. +#[path = "core_ops.rs"] +mod core_ops; +/// helpers: credential parsing, signing/hash utilities and parsers. +#[path = "helpers.rs"] +mod helpers; +/// pool_ops: pool lifecycle and status operations. +#[path = "pool_ops.rs"] +mod pool_ops; +/// s3_ops: bucket/object-lock operations for S3-compatible endpoints. +#[path = "s3_ops.rs"] +mod s3_ops; +/// sts_ops: temporary credential flows. +#[path = "sts_ops.rs"] +mod sts_ops; +const FORM_CONTENT_TYPE: &str = "application/x-www-form-urlencoded"; +const ASSUME_ROLE_PATH: &str = "/"; +const STS_SIGNING_SERVICE: &str = "sts"; +const JSON_CONTENT_TYPE: &str = "application/json"; +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"; +const ADD_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/add-canned-policy"; +const INFO_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/info-canned-policy"; +const SERVER_INFO_PATH: &str = "/rustfs/admin/v3/info"; +const POOLS_LIST_PATH: &str = "/rustfs/admin/v3/pools/list"; +const POOLS_STATUS_PATH: &str = "/rustfs/admin/v3/pools/status"; +const POOLS_DECOMMISSION_PATH: &str = "/rustfs/admin/v3/pools/decommission"; +const POOLS_CANCEL_PATH: &str = "/rustfs/admin/v3/pools/cancel"; +const ADMIN_SIGNING_SERVICE: &str = "s3"; +const ADMIN_HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); +const ADMIN_HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_UPSTREAM_ERROR_BODY_BYTES: usize = 8 * 1024; +const MAX_UPSTREAM_ERROR_DETAIL_CHARS: usize = 512; + +/// Credentials read from Tenant `.spec.credsSecret`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RustfsCredentials { + pub access_key: String, + pub secret_key: String, +} + +#[derive(Debug, Clone, serde::Deserialize, PartialEq)] +pub struct RustfsPoolListItem { + pub id: usize, + #[serde(rename = "cmdline")] + pub cmd_line: String, + #[serde(rename = "lastUpdate")] + pub last_update: String, + #[serde(rename = "totalSize")] + pub total_size: Option, + #[serde(rename = "currentSize")] + pub current_size: Option, + #[serde(rename = "usedSize")] + pub used_size: Option, + pub used: Option, + pub status: String, + #[serde(rename = "decommissionInfo")] + pub decommission: Option, +} + +#[derive(Debug, Clone, serde::Deserialize, PartialEq)] +pub struct RustfsPoolStatus { + pub id: usize, + #[serde(rename = "cmdline")] + pub cmd_line: String, + #[serde(rename = "lastUpdate")] + pub last_update: String, + #[serde(rename = "decommissionInfo")] + pub decommission: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CreateBucketResult { + Created, + 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")] + pub start_time: Option, + #[serde(rename = "startSize")] + pub start_size: Option, + #[serde(rename = "totalSize")] + pub total_size: Option, + #[serde(rename = "currentSize")] + pub current_size: Option, + pub complete: Option, + pub failed: Option, + pub canceled: Option, + #[serde(rename = "objectsDecommissioned")] + pub objects_decommissioned: Option, + #[serde(rename = "objectsDecommissionedFailed")] + pub objects_decommissioned_failed: Option, + #[serde(rename = "bytesDecommissioned")] + pub bytes_decommissioned: Option, + #[serde(rename = "bytesDecommissionedFailed")] + pub bytes_decommissioned_failed: Option, +} + +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsServerInfo { + #[serde(default)] + pub usage: Option, + #[serde(default)] + pub backend: Option, + #[serde(default)] + pub pools: Option>>, +} + +#[derive(Debug, Clone, serde::Deserialize, PartialEq)] +struct RustfsServerInfoResponse { + pub info: RustfsServerInfo, +} + +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsServerUsage { + #[serde(default)] + pub size: u64, +} + +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsErasureBackend { + #[serde(default, rename = "onlineDisks")] + pub online_disks: u64, + #[serde(default, rename = "offlineDisks")] + pub offline_disks: u64, + #[serde(default, rename = "standardSCParity", alias = "StandardSCParity")] + pub standard_sc_parity: Option, + #[serde(default, rename = "totalSets")] + pub total_sets: Vec, + #[serde(default, rename = "totalDrivesPerSet", alias = "drivesPerSet")] + pub drives_per_set: Vec, +} + +#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] +pub struct RustfsErasureSetInfo { + #[serde(default, rename = "rawUsage")] + pub raw_usage: u64, + #[serde(default, rename = "rawCapacity")] + pub raw_capacity: u64, + #[serde(default)] + pub usage: u64, + #[serde(default, rename = "objectsCount")] + pub objects_count: u64, + #[serde(default, rename = "healDisks")] + pub heal_disks: u64, +} + +/// Temporary credentials returned by RustFS STS AssumeRole. +#[derive(Debug, Clone)] +pub struct RustfsStsCredentials { + pub access_key_id: String, + pub secret_access_key: String, + pub session_token: String, + pub expiration: String, +} + +/// Error type for RustFS admin/STS client operations. +#[derive(Debug)] +pub enum RustfsClientError { + MissingTenantNamespace, + MissingCredsSecret, + MissingCredentialKey { + key: &'static str, + }, + EmptyCredentialValue { + key: &'static str, + }, + InvalidCredentialValue { + key: &'static str, + }, + TenantSecretLookupFailed, + InvalidPolicyName, + InvalidPolicyDocument, + TenantTlsRequired, + TenantTlsNotReady, + TenantTlsClientCertificateRequired, + MissingTenantTlsCaKey { + secret: String, + key: String, + }, + TenantTlsCaSecretLookupFailed { + secret: String, + }, + InvalidTenantTlsCa, + TlsClientBuildFailed, + RequestBuildFailed, + RequestFailed, + UnexpectedStatus { + status: StatusCode, + detail: Option, + }, + ParseResponseFailed, + SigningFailed, +} + +impl std::fmt::Display for RustfsClientError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::MissingTenantNamespace => write!(f, "tenant namespace is missing"), + Self::MissingCredsSecret => write!(f, "tenant credsSecret is missing"), + Self::MissingCredentialKey { key } => write!(f, "secret key missing: {key}"), + Self::EmptyCredentialValue { key } => write!(f, "secret key empty: {key}"), + Self::InvalidCredentialValue { key } => { + write!(f, "secret key is not valid utf8: {key}") + } + Self::TenantSecretLookupFailed => { + write!(f, "failed to load tenant credential secret") + } + Self::InvalidPolicyName => write!(f, "invalid policy name"), + Self::InvalidPolicyDocument => write!(f, "failed to parse canned policy response"), + Self::TenantTlsRequired => write!(f, "STS requires a TLS-enabled tenant"), + Self::TenantTlsNotReady => write!(f, "tenant TLS status is not ready"), + Self::TenantTlsClientCertificateRequired => { + write!(f, "tenant TLS requires a client certificate") + } + Self::MissingTenantTlsCaKey { secret, key } => { + write!(f, "tenant TLS CA secret {secret} missing key {key}") + } + Self::TenantTlsCaSecretLookupFailed { secret } => { + write!(f, "failed to load tenant TLS CA secret {secret}") + } + Self::InvalidTenantTlsCa => write!(f, "tenant TLS CA is not a valid PEM bundle"), + Self::TlsClientBuildFailed => write!(f, "failed to build TLS HTTP client"), + Self::RequestBuildFailed => write!(f, "failed to construct request"), + Self::RequestFailed => write!(f, "request failed"), + Self::UnexpectedStatus { status, detail } => { + write!(f, "upstream returned {status}")?; + if let Some(detail) = detail { + write!(f, ": {detail}")?; + } + Ok(()) + } + Self::ParseResponseFailed => write!(f, "failed to parse AssumeRole response"), + Self::SigningFailed => write!(f, "failed to compute request signature"), + } + } +} + +impl std::error::Error for RustfsClientError {} + +impl RustfsClientError { + async fn unexpected_response(response: Response) -> Self { + let status = response.status(); + let (body, truncated) = read_limited_response_body(response).await; + Self::unexpected_status_with_limited_body(status, &body, truncated) + } + + async fn limited_response_body(response: Response) -> (String, bool) { + read_limited_response_body(response).await + } + + fn unexpected_status_with_limited_body( + status: StatusCode, + body: &str, + body_truncated: bool, + ) -> Self { + Self::UnexpectedStatus { + status, + detail: summarize_upstream_error_body(body, body_truncated), + } + } +} + +async fn read_limited_response_body(mut response: Response) -> (String, bool) { + let mut body = Vec::new(); + let read_limit = MAX_UPSTREAM_ERROR_BODY_BYTES.saturating_add(1); + + loop { + let remaining = read_limit.saturating_sub(body.len()); + if remaining == 0 { + break; + } + + let chunk = match response.chunk().await { + Ok(Some(chunk)) => chunk, + Ok(None) => break, + Err(_) => break, + }; + if chunk.len() > remaining { + body.extend_from_slice(&chunk[..remaining]); + break; + } + body.extend_from_slice(&chunk); + } + + let truncated = body.len() > MAX_UPSTREAM_ERROR_BODY_BYTES; + if truncated { + body.truncate(MAX_UPSTREAM_ERROR_BODY_BYTES); + } + + (String::from_utf8_lossy(&body).into_owned(), truncated) +} + +fn summarize_upstream_error_body(body: &str, body_truncated: bool) -> Option { + let body = body.trim(); + if body.is_empty() { + return None; + } + + if let Some(message) = helpers::extract_xml_tag(body, "Message") { + let message = decode_basic_xml_entities(&message); + let detail = match helpers::extract_xml_tag(body, "Code") { + Some(code) if !code.trim().is_empty() => { + format!("{}: {message}", decode_basic_xml_entities(&code)) + } + _ => message, + }; + return Some(sanitize_error_detail(&detail)); + } + + if let Ok(value) = serde_json::from_str::(body) + && let Some(detail) = summarize_json_error(&value) + { + return Some(sanitize_error_detail(&detail)); + } + + if body_truncated { + return Some(format!( + "response body exceeded {MAX_UPSTREAM_ERROR_BODY_BYTES} bytes" + )); + } + + Some(sanitize_error_detail(body)) +} + +fn summarize_json_error(value: &serde_json::Value) -> Option { + if let Some(message) = value.as_str() { + return Some(message.to_string()); + } + + let object = value.as_object()?; + let message = ["message", "Message", "error", "Error"] + .iter() + .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str))?; + let code = ["code", "Code"] + .iter() + .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str)); + + Some(match code { + Some(code) if !code.trim().is_empty() => format!("{code}: {message}"), + _ => message.to_string(), + }) +} + +fn collapse_whitespace(value: &str) -> String { + value.split_whitespace().collect::>().join(" ") +} + +fn sanitize_error_detail(value: &str) -> String { + let detail = collapse_whitespace(value); + truncate_error_detail(detail) +} + +fn truncate_error_detail(value: String) -> String { + let mut truncated = String::new(); + for (index, ch) in value.chars().enumerate() { + if index >= MAX_UPSTREAM_ERROR_DETAIL_CHARS { + truncated.push_str("..."); + return truncated; + } + truncated.push(ch); + } + truncated +} + +fn decode_basic_xml_entities(value: &str) -> String { + value + .replace("<", "<") + .replace(">", ">") + .replace(""", "\"") + .replace("'", "'") + .replace("&", "&") +} + +#[derive(Debug)] +struct SignedRequest { + amz_date: String, + payload_hash: String, + authorization: String, +} + +/// RustFS admin/STS client. +pub struct RustfsAdminClient { + base_url: String, + access_key: String, + secret_key: String, + region: String, + http_client: HttpClient, +} + +fn default_http_client() -> HttpClient { + HttpClient::builder() + .connect_timeout(ADMIN_HTTP_CONNECT_TIMEOUT) + .timeout(ADMIN_HTTP_REQUEST_TIMEOUT) + .build() + .unwrap_or_else(|_| HttpClient::new()) +} + +impl RustfsAdminClient { + pub const STS_VERSION: &'static str = "2011-06-15"; + pub const STS_ACTION: &'static str = "AssumeRole"; + + pub fn new_with_base_url( + base_url: impl Into, + access_key: impl Into, + secret_key: impl Into, + ) -> Self { + Self::new_with_base_url_and_http_client( + base_url, + access_key, + secret_key, + default_http_client(), + ) + } + + pub fn new_with_base_url_and_ca_pem( + base_url: impl Into, + access_key: impl Into, + secret_key: impl Into, + ca_pem: &[u8], + ) -> Result { + let certs = Certificate::from_pem_bundle(ca_pem) + .map_err(|_| RustfsClientError::InvalidTenantTlsCa)?; + let mut builder = HttpClient::builder() + .connect_timeout(ADMIN_HTTP_CONNECT_TIMEOUT) + .timeout(ADMIN_HTTP_REQUEST_TIMEOUT); + for cert in certs { + builder = builder.add_root_certificate(cert); + } + let http_client = builder + .build() + .map_err(|_| RustfsClientError::TlsClientBuildFailed)?; + + Ok(Self::new_with_base_url_and_http_client( + base_url, + access_key, + secret_key, + http_client, + )) + } + + pub fn new_with_base_url_and_http_client( + base_url: impl Into, + access_key: impl Into, + secret_key: impl Into, + http_client: HttpClient, + ) -> Self { + Self { + base_url: base_url.into(), + access_key: access_key.into(), + secret_key: secret_key.into(), + region: "us-east-1".to_string(), + http_client, + } + } +} + +#[cfg(test)] +mod tests { + use super::helpers::build_canonical_query; + + #[test] + fn canonical_query_uses_sigv4_uri_encoding_and_encoded_sort_order() { + let query = build_canonical_query(&[ + ("z", "a b~c/雪"), + ("a~", "second"), + ("a ", "first"), + ("amp", "&="), + ("dup", "z"), + ("dup", "a"), + ("empty", ""), + ("雪", "key"), + ]); + + assert_eq!( + query, + "%E9%9B%AA=key&a%20=first&=%26%3D&a~=second&dup=a&dup=z&empty=&z=a%20b~c%2F%E9%9B%AA" + ); + } +} diff --git a/src/sts/pool_ops.rs b/crates/rustfs-admin/src/pool_ops.rs similarity index 100% rename from src/sts/pool_ops.rs rename to crates/rustfs-admin/src/pool_ops.rs diff --git a/src/sts/s3_ops.rs b/crates/rustfs-admin/src/s3_ops.rs similarity index 78% rename from src/sts/s3_ops.rs rename to crates/rustfs-admin/src/s3_ops.rs index 0df217f..023dc1a 100644 --- a/src/sts/s3_ops.rs +++ b/crates/rustfs-admin/src/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/sts_ops.rs b/crates/rustfs-admin/src/sts_ops.rs similarity index 75% rename from src/sts/sts_ops.rs rename to crates/rustfs-admin/src/sts_ops.rs index ba45bcc..c13e9b8 100644 --- a/src/sts/sts_ops.rs +++ b/crates/rustfs-admin/src/sts_ops.rs @@ -14,11 +14,11 @@ //! STS boundary: //! - temporary credentials and AssumeRole request composition/response parsing. -use super::helpers::{build_form_body, parse_assume_role_response}; +use super::helpers::{build_form_body, extract_xml_tag}; use super::{ - ASSUME_ROLE_PATH, FORM_CONTENT_TYPE, RustfsAdminClient, RustfsClientError, STS_SIGNING_SERVICE, + ASSUME_ROLE_PATH, FORM_CONTENT_TYPE, RustfsAdminClient, RustfsClientError, + RustfsStsCredentials, STS_SIGNING_SERVICE, }; -use crate::sts::types::StsAssumeRoleCredentials; impl RustfsAdminClient { // STS duties: temporary credentials and AssumeRole API call path. @@ -28,7 +28,7 @@ impl RustfsAdminClient { &self, policy: Option<&str>, duration_seconds: u64, - ) -> Result { + ) -> Result { let mut params = vec![ ("Version", Self::STS_VERSION.to_string()), ("Action", Self::STS_ACTION.to_string()), @@ -79,6 +79,19 @@ impl RustfsAdminClient { .await .map_err(|_| RustfsClientError::RequestFailed)?; - parse_assume_role_response(&body).ok_or(RustfsClientError::ParseResponseFailed) + let access_key_id = + extract_xml_tag(&body, "AccessKeyId").ok_or(RustfsClientError::ParseResponseFailed)?; + let secret_access_key = extract_xml_tag(&body, "SecretAccessKey") + .ok_or(RustfsClientError::ParseResponseFailed)?; + let session_token = + extract_xml_tag(&body, "SessionToken").ok_or(RustfsClientError::ParseResponseFailed)?; + let expiration = + extract_xml_tag(&body, "Expiration").ok_or(RustfsClientError::ParseResponseFailed)?; + Ok(RustfsStsCredentials { + access_key_id, + secret_access_key, + session_token, + expiration, + }) } } 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/rustfs_client.rs b/src/sts/rustfs_client.rs index 865c27f..a94204d 100644 --- a/src/sts/rustfs_client.rs +++ b/src/sts/rustfs_client.rs @@ -12,443 +12,49 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{collections::BTreeMap, time::Duration}; +use std::ops::{Deref, DerefMut}; use k8s_openapi::api::core::v1 as corev1; use kube::{Api, Client}; -use reqwest::{Certificate, Client as HttpClient, Response, StatusCode}; use crate::Tenant; use crate::cluster_dns; -use crate::utils::sanitize::redact_sensitive_pairs; -/// admin_ops: tenant admin operations (user/policy APIs). -#[path = "admin_ops.rs"] -mod admin_ops; -/// core_ops: shared request signing/dispatch internals. -#[path = "core_ops.rs"] -mod core_ops; -/// helpers: credential parsing, signing/hash utilities and parsers. -#[path = "helpers.rs"] -mod helpers; -/// pool_ops: pool lifecycle and status operations. -#[path = "pool_ops.rs"] -mod pool_ops; -/// s3_ops: bucket/object-lock operations for S3-compatible endpoints. -#[path = "s3_ops.rs"] -mod s3_ops; -/// sts_ops: temporary credential flows, AssumeRole request/response. -#[path = "sts_ops.rs"] -mod sts_ops; +pub use rustfs_admin::{ + CreateBucketResult, RustfsClientError, RustfsCredentials, RustfsErasureBackend, + RustfsErasureSetInfo, RustfsPoolDecommissionInfo, RustfsPoolListItem, RustfsPoolStatus, + RustfsServerInfo, RustfsServerUsage, RustfsUserInfo, +}; -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 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"; -const ADD_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/add-canned-policy"; -const INFO_CANNED_POLICY_PATH: &str = "/rustfs/admin/v3/info-canned-policy"; -const SERVER_INFO_PATH: &str = "/rustfs/admin/v3/info"; -const POOLS_LIST_PATH: &str = "/rustfs/admin/v3/pools/list"; -const POOLS_STATUS_PATH: &str = "/rustfs/admin/v3/pools/status"; -const POOLS_DECOMMISSION_PATH: &str = "/rustfs/admin/v3/pools/decommission"; -const POOLS_CANCEL_PATH: &str = "/rustfs/admin/v3/pools/cancel"; -const ADMIN_SIGNING_SERVICE: &str = "s3"; -const STS_SIGNING_SERVICE: &str = "sts"; -const ADMIN_HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); -const ADMIN_HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); -const MAX_UPSTREAM_ERROR_BODY_BYTES: usize = 8 * 1024; -const MAX_UPSTREAM_ERROR_DETAIL_CHARS: usize = 512; +/// Tenant-aware wrapper around the kube-agnostic RustFS admin client. +pub struct RustfsAdminClient(pub rustfs_admin::RustfsAdminClient); -/// Credentials read from Tenant `.spec.credsSecret`. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RustfsCredentials { - pub access_key: String, - pub secret_key: String, -} - -#[derive(Debug, Clone, serde::Deserialize, PartialEq)] -pub struct RustfsPoolListItem { - pub id: usize, - #[serde(rename = "cmdline")] - pub cmd_line: String, - #[serde(rename = "lastUpdate")] - pub last_update: String, - #[serde(rename = "totalSize")] - pub total_size: Option, - #[serde(rename = "currentSize")] - pub current_size: Option, - #[serde(rename = "usedSize")] - pub used_size: Option, - pub used: Option, - pub status: String, - #[serde(rename = "decommissionInfo")] - pub decommission: Option, -} - -#[derive(Debug, Clone, serde::Deserialize, PartialEq)] -pub struct RustfsPoolStatus { - pub id: usize, - #[serde(rename = "cmdline")] - pub cmd_line: String, - #[serde(rename = "lastUpdate")] - pub last_update: String, - #[serde(rename = "decommissionInfo")] - pub decommission: Option, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum CreateBucketResult { - Created, - AlreadyExists, -} - -#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] -pub struct RustfsPoolDecommissionInfo { - #[serde(rename = "startTime")] - pub start_time: Option, - #[serde(rename = "startSize")] - pub start_size: Option, - #[serde(rename = "totalSize")] - pub total_size: Option, - #[serde(rename = "currentSize")] - pub current_size: Option, - pub complete: Option, - pub failed: Option, - pub canceled: Option, - #[serde(rename = "objectsDecommissioned")] - pub objects_decommissioned: Option, - #[serde(rename = "objectsDecommissionedFailed")] - pub objects_decommissioned_failed: Option, - #[serde(rename = "bytesDecommissioned")] - pub bytes_decommissioned: Option, - #[serde(rename = "bytesDecommissionedFailed")] - pub bytes_decommissioned_failed: Option, -} - -#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] -pub struct RustfsServerInfo { - #[serde(default)] - pub usage: Option, - #[serde(default)] - pub backend: Option, - #[serde(default)] - pub pools: Option>>, -} - -#[derive(Debug, Clone, serde::Deserialize, PartialEq)] -pub(super) struct RustfsServerInfoResponse { - pub info: RustfsServerInfo, -} - -#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] -pub struct RustfsServerUsage { - #[serde(default)] - pub size: u64, -} - -#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] -pub struct RustfsErasureBackend { - #[serde(default, rename = "onlineDisks")] - pub online_disks: u64, - #[serde(default, rename = "offlineDisks")] - pub offline_disks: u64, - #[serde(default, rename = "standardSCParity", alias = "StandardSCParity")] - pub standard_sc_parity: Option, - #[serde(default, rename = "totalSets")] - pub total_sets: Vec, - #[serde(default, rename = "totalDrivesPerSet", alias = "drivesPerSet")] - pub drives_per_set: Vec, -} - -#[derive(Debug, Clone, Default, serde::Deserialize, PartialEq)] -pub struct RustfsErasureSetInfo { - #[serde(default, rename = "rawUsage")] - pub raw_usage: u64, - #[serde(default, rename = "rawCapacity")] - pub raw_capacity: u64, - #[serde(default)] - pub usage: u64, - #[serde(default, rename = "objectsCount")] - pub objects_count: u64, - #[serde(default, rename = "healDisks")] - pub heal_disks: u64, -} - -/// Error type for RustFS admin/STS client operations. -#[derive(Debug)] -pub enum RustfsClientError { - MissingTenantNamespace, - MissingCredsSecret, - MissingCredentialKey { - key: &'static str, - }, - EmptyCredentialValue { - key: &'static str, - }, - InvalidCredentialValue { - key: &'static str, - }, - TenantSecretLookupFailed, - InvalidPolicyName, - InvalidPolicyDocument, - TenantTlsRequired, - TenantTlsNotReady, - TenantTlsClientCertificateRequired, - MissingTenantTlsCaKey { - secret: String, - key: String, - }, - TenantTlsCaSecretLookupFailed { - secret: String, - }, - InvalidTenantTlsCa, - TlsClientBuildFailed, - RequestBuildFailed, - RequestFailed, - UnexpectedStatus { - status: StatusCode, - detail: Option, - }, - ParseResponseFailed, - SigningFailed, -} - -impl std::fmt::Display for RustfsClientError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::MissingTenantNamespace => write!(f, "tenant namespace is missing"), - Self::MissingCredsSecret => write!(f, "tenant credsSecret is missing"), - Self::MissingCredentialKey { key } => write!(f, "secret key missing: {key}"), - Self::EmptyCredentialValue { key } => write!(f, "secret key empty: {key}"), - Self::InvalidCredentialValue { key } => { - write!(f, "secret key is not valid utf8: {key}") - } - Self::TenantSecretLookupFailed => { - write!(f, "failed to load tenant credential secret") - } - Self::InvalidPolicyName => write!(f, "invalid policy name"), - Self::InvalidPolicyDocument => write!(f, "failed to parse canned policy response"), - Self::TenantTlsRequired => write!(f, "STS requires a TLS-enabled tenant"), - Self::TenantTlsNotReady => write!(f, "tenant TLS status is not ready"), - Self::TenantTlsClientCertificateRequired => { - write!(f, "tenant TLS requires a client certificate") - } - Self::MissingTenantTlsCaKey { secret, key } => { - write!(f, "tenant TLS CA secret {secret} missing key {key}") - } - Self::TenantTlsCaSecretLookupFailed { secret } => { - write!(f, "failed to load tenant TLS CA secret {secret}") - } - Self::InvalidTenantTlsCa => write!(f, "tenant TLS CA is not a valid PEM bundle"), - Self::TlsClientBuildFailed => write!(f, "failed to build TLS HTTP client"), - Self::RequestBuildFailed => write!(f, "failed to construct request"), - Self::RequestFailed => write!(f, "request failed"), - Self::UnexpectedStatus { status, detail } => { - write!(f, "upstream returned {status}")?; - if let Some(detail) = detail { - write!(f, ": {detail}")?; - } - Ok(()) - } - Self::ParseResponseFailed => write!(f, "failed to parse AssumeRole response"), - Self::SigningFailed => write!(f, "failed to compute request signature"), - } - } -} - -impl std::error::Error for RustfsClientError {} - -impl RustfsClientError { - pub(super) async fn unexpected_response(response: Response) -> Self { - let status = response.status(); - let (body, truncated) = read_limited_response_body(response).await; - Self::unexpected_status_with_limited_body(status, &body, truncated) - } - - pub(super) async fn limited_response_body(response: Response) -> (String, bool) { - read_limited_response_body(response).await - } +impl Deref for RustfsAdminClient { + type Target = rustfs_admin::RustfsAdminClient; - fn unexpected_status_with_limited_body( - status: StatusCode, - body: &str, - body_truncated: bool, - ) -> Self { - Self::UnexpectedStatus { - status, - detail: summarize_upstream_error_body(body, body_truncated), - } - } - - #[cfg(test)] - pub(super) fn unexpected_status_with_body(status: StatusCode, body: &str) -> Self { - Self::unexpected_status_with_limited_body(status, body, false) + fn deref(&self) -> &Self::Target { + &self.0 } } -async fn read_limited_response_body(mut response: Response) -> (String, bool) { - let mut body = Vec::new(); - let read_limit = MAX_UPSTREAM_ERROR_BODY_BYTES.saturating_add(1); - - loop { - let remaining = read_limit.saturating_sub(body.len()); - if remaining == 0 { - break; - } - - let chunk = match response.chunk().await { - Ok(Some(chunk)) => chunk, - Ok(None) => break, - Err(_) => break, - }; - if chunk.len() > remaining { - body.extend_from_slice(&chunk[..remaining]); - break; - } - body.extend_from_slice(&chunk); +impl DerefMut for RustfsAdminClient { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.0 } - - let truncated = body.len() > MAX_UPSTREAM_ERROR_BODY_BYTES; - if truncated { - body.truncate(MAX_UPSTREAM_ERROR_BODY_BYTES); - } - - (String::from_utf8_lossy(&body).into_owned(), truncated) -} - -fn summarize_upstream_error_body(body: &str, body_truncated: bool) -> Option { - let body = body.trim(); - if body.is_empty() { - return None; - } - - if let Some(message) = helpers::extract_xml_tag(body, "Message") { - let message = decode_basic_xml_entities(&message); - let detail = match helpers::extract_xml_tag(body, "Code") { - Some(code) if !code.trim().is_empty() => { - format!("{}: {message}", decode_basic_xml_entities(&code)) - } - _ => message, - }; - return Some(sanitize_error_detail(&detail)); - } - - if let Ok(value) = serde_json::from_str::(body) - && let Some(detail) = summarize_json_error(&value) - { - return Some(sanitize_error_detail(&detail)); - } - - if body_truncated { - return Some(format!( - "response body exceeded {MAX_UPSTREAM_ERROR_BODY_BYTES} bytes" - )); - } - - Some(sanitize_error_detail(body)) -} - -fn summarize_json_error(value: &serde_json::Value) -> Option { - if let Some(message) = value.as_str() { - return Some(message.to_string()); - } - - let object = value.as_object()?; - let message = ["message", "Message", "error", "Error"] - .iter() - .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str))?; - let code = ["code", "Code"] - .iter() - .find_map(|key| object.get(*key).and_then(serde_json::Value::as_str)); - - Some(match code { - Some(code) if !code.trim().is_empty() => format!("{code}: {message}"), - _ => message.to_string(), - }) -} - -fn collapse_whitespace(value: &str) -> String { - value.split_whitespace().collect::>().join(" ") -} - -fn sanitize_error_detail(value: &str) -> String { - let detail = collapse_whitespace(value); - let detail = redact_sensitive_pairs(&detail); - truncate_error_detail(detail) -} - -fn truncate_error_detail(value: String) -> String { - let mut truncated = String::new(); - for (index, ch) in value.chars().enumerate() { - if index >= MAX_UPSTREAM_ERROR_DETAIL_CHARS { - truncated.push_str("..."); - return truncated; - } - truncated.push(ch); - } - truncated -} - -fn decode_basic_xml_entities(value: &str) -> String { - value - .replace("<", "<") - .replace(">", ">") - .replace(""", "\"") - .replace("'", "'") - .replace("&", "&") -} - -#[derive(Debug)] -struct SignedRequest { - amz_date: String, - payload_hash: String, - authorization: String, -} - -/// RustFS admin/STS client. -pub struct RustfsAdminClient { - base_url: String, - access_key: String, - secret_key: String, - region: String, - http_client: HttpClient, -} - -fn default_http_client() -> HttpClient { - HttpClient::builder() - .connect_timeout(ADMIN_HTTP_CONNECT_TIMEOUT) - .timeout(ADMIN_HTTP_REQUEST_TIMEOUT) - .build() - .unwrap_or_else(|_| HttpClient::new()) -} - -fn tls_tenant_base_url(tenant: &Tenant, cluster_domain: &str) -> Result { - let namespace = tenant - .namespace() - .map_err(|_| RustfsClientError::MissingTenantNamespace)?; - let service_fqdn = - cluster_dns::service_fqdn(&tenant.headless_service_name(), &namespace, cluster_domain); - Ok(format!("https://{service_fqdn}:9000")) } impl RustfsAdminClient { - pub const STS_VERSION: &'static str = "2011-06-15"; - pub const STS_ACTION: &'static str = "AssumeRole"; + pub const STS_VERSION: &'static str = rustfs_admin::RustfsAdminClient::STS_VERSION; + pub const STS_ACTION: &'static str = rustfs_admin::RustfsAdminClient::STS_ACTION; pub fn new_with_base_url( base_url: impl Into, access_key: impl Into, secret_key: impl Into, ) -> Self { - Self::new_with_base_url_and_http_client( - base_url, - access_key, - secret_key, - default_http_client(), - ) + Self(rustfs_admin::RustfsAdminClient::new_with_base_url( + base_url, access_key, secret_key, + )) } pub fn new_with_base_url_and_ca_pem( @@ -457,39 +63,26 @@ impl RustfsAdminClient { secret_key: impl Into, ca_pem: &[u8], ) -> Result { - let certs = Certificate::from_pem_bundle(ca_pem) - .map_err(|_| RustfsClientError::InvalidTenantTlsCa)?; - let mut builder = HttpClient::builder() - .connect_timeout(ADMIN_HTTP_CONNECT_TIMEOUT) - .timeout(ADMIN_HTTP_REQUEST_TIMEOUT); - for cert in certs { - builder = builder.add_root_certificate(cert); - } - let http_client = builder - .build() - .map_err(|_| RustfsClientError::TlsClientBuildFailed)?; - - Ok(Self::new_with_base_url_and_http_client( - base_url, - access_key, - secret_key, - http_client, - )) + rustfs_admin::RustfsAdminClient::new_with_base_url_and_ca_pem( + base_url, access_key, secret_key, ca_pem, + ) + .map(Self) } pub fn new_with_base_url_and_http_client( base_url: impl Into, access_key: impl Into, secret_key: impl Into, - http_client: HttpClient, + http_client: reqwest::Client, ) -> Self { - Self { - base_url: base_url.into(), - access_key: access_key.into(), - secret_key: secret_key.into(), - region: "us-east-1".to_string(), - http_client, - } + Self( + rustfs_admin::RustfsAdminClient::new_with_base_url_and_http_client( + base_url, + access_key, + secret_key, + http_client, + ), + ) } pub fn from_tenant( @@ -518,15 +111,25 @@ impl RustfsAdminClient { credentials: RustfsCredentials, cluster_domain: &str, ) -> Result { - if !helpers::tenant_tls_enabled(tenant) { + if !tenant.spec.tls.as_ref().is_some_and(|tls| tls.is_enabled()) { return Err(RustfsClientError::TenantTlsRequired); } - if helpers::tenant_tls_client_certificate_required(tenant) { + if tenant + .status + .as_ref() + .and_then(|status| status.certificates.tls.as_ref()) + .and_then(|tls| tls.client_ca_secret_ref.as_ref()) + .is_some() + { return Err(RustfsClientError::TenantTlsClientCertificateRequired); } - let base_url = tls_tenant_base_url(tenant, cluster_domain)?; - + let namespace = tenant + .namespace() + .map_err(|_| RustfsClientError::MissingTenantNamespace)?; + let service_fqdn = + cluster_dns::service_fqdn(&tenant.headless_service_name(), &namespace, cluster_domain); + let base_url = format!("https://{service_fqdn}:9000"); match Self::load_tenant_tls_ca(kube_client, tenant).await? { Some(ca_pem) => Self::new_with_base_url_and_ca_pem( base_url, @@ -546,21 +149,18 @@ impl RustfsAdminClient { kube_client: &Client, tenant: &Tenant, ) -> Result>, RustfsClientError> { - if !helpers::tenant_tls_enabled(tenant) { + if !tenant.spec.tls.as_ref().is_some_and(|tls| tls.is_enabled()) { return Ok(None); } - let tls_status = tenant .status .as_ref() .and_then(|status| status.certificates.tls.as_ref()) .filter(|tls| tls.ready) .ok_or(RustfsClientError::TenantTlsNotReady)?; - let Some(ca_ref) = tls_status.ca_secret_ref.as_ref() else { return Ok(None); }; - let namespace = tenant .namespace() .map_err(|_| RustfsClientError::MissingTenantNamespace)?; @@ -581,11 +181,9 @@ impl RustfsAdminClient { secret: ca_ref.name.clone(), key: key.to_string(), })?; - Ok(Some(ca_pem)) } - /// Read Tenant credential Secret and return access/secret key pair. pub async fn load_tenant_credentials( kube_client: &Client, tenant: &Tenant, @@ -595,7 +193,6 @@ impl RustfsAdminClient { .creds_secret .as_ref() .ok_or(RustfsClientError::MissingCredsSecret)?; - let namespace = tenant .namespace() .map_err(|_| RustfsClientError::MissingTenantNamespace)?; @@ -604,11 +201,38 @@ impl RustfsAdminClient { .get(&reference.name) .await .map_err(|_| RustfsClientError::TenantSecretLookupFailed)?; - - helpers::extract_credentials(secret.data.as_ref()) + let data = secret + .data + .as_ref() + .ok_or(RustfsClientError::TenantSecretLookupFailed)?; + let credential = |key: &'static str| -> Result { + let value = data + .get(key) + .ok_or(RustfsClientError::MissingCredentialKey { key })?; + let value = String::from_utf8(value.0.clone()) + .map_err(|_| RustfsClientError::InvalidCredentialValue { key })?; + if value.is_empty() { + return Err(RustfsClientError::EmptyCredentialValue { key }); + } + Ok(value) + }; + Ok(RustfsCredentials { + access_key: credential("accesskey")?, + secret_key: credential("secretkey")?, + }) + } + + pub async fn assume_role( + &self, + policy: Option<&str>, + duration_seconds: u64, + ) -> Result { + let credentials = self.0.assume_role(policy, duration_seconds).await?; + Ok(crate::sts::types::StsAssumeRoleCredentials { + access_key_id: credentials.access_key_id, + secret_access_key: credentials.secret_access_key, + session_token: credentials.session_token, + expiration: credentials.expiration, + }) } } - -#[cfg(test)] -#[path = "tests.rs"] -mod tests; 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();