diff --git a/crates/e2e_test/src/admin_iam_crud_test.rs b/crates/e2e_test/src/admin_iam_crud_test.rs index ac6a040f1..4d6754a42 100644 --- a/crates/e2e_test/src/admin_iam_crud_test.rs +++ b/crates/e2e_test/src/admin_iam_crud_test.rs @@ -26,75 +26,16 @@ //! Later batches tracked on backlog#1154: config get/set, info, pools status, //! group lifecycle, import/export IAM. -use crate::common::{RustFSTestEnvironment, init_logging, local_http_client}; +use crate::common::{RustFSTestEnvironment, admin_ok, admin_request, init_logging}; use aws_sdk_s3::config::{Credentials, Region}; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::{Client, Config}; -use http::header::{CONTENT_TYPE, HOST}; use reqwest::StatusCode; -use rustfs_signer::constants::UNSIGNED_PAYLOAD; -use rustfs_signer::sign_v4; -use s3s::Body; use serial_test::serial; use std::error::Error; use tokio::time::{Duration, sleep}; type TestResult = Result<(), Box>; -type BoxError = Box; - -/// Signs and sends an admin HTTP request with the given credential, returning -/// status and body. Native `/rustfs/admin/v3` requests and responses are plain -/// JSON (the MinIO-compat encryption applies only to `/minio/admin/v3` paths). -async fn admin_request( - base_url: &str, - method: http::Method, - path_and_query: &str, - body: Option, - access_key: &str, - secret_key: &str, -) -> Result<(StatusCode, String), BoxError> { - let url = format!("{base_url}{path_and_query}"); - let uri = url.parse::()?; - let authority = uri.authority().ok_or("admin URL missing authority")?.to_string(); - let mut builder = http::Request::builder() - .method(method.clone()) - .uri(uri) - .header(HOST, authority) - .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); - if body.is_some() { - builder = builder.header(CONTENT_TYPE, "application/json"); - } - - let content_len = body.as_ref().map(|b| b.len() as i64).unwrap_or_default(); - let signed = sign_v4(builder.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1"); - - let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; - let mut request = local_http_client().request(reqwest_method, &url); - for (name, value) in signed.headers() { - request = request.header(name, value); - } - if let Some(body) = body { - request = request.body(body); - } - let response = request.send().await?; - let status = response.status(); - let text = response.text().await.unwrap_or_default(); - Ok((status, text)) -} - -/// Root-credential admin request that must succeed; returns the response body. -async fn admin_ok( - env: &RustFSTestEnvironment, - method: http::Method, - path_and_query: &str, - body: Option, -) -> Result { - let (status, text) = admin_request(&env.url, method.clone(), path_and_query, body, &env.access_key, &env.secret_key).await?; - if !status.is_success() { - return Err(format!("{method} {path_and_query} failed: {status} {text}").into()); - } - Ok(text) -} fn build_s3_client(url: &str, access_key: &str, secret_key: &str) -> Client { let config = Config::builder() diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index ca8514421..0c4c4416f 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -24,7 +24,12 @@ use aws_sdk_s3::config::{Credentials, Region}; use aws_sdk_s3::{Client, Config}; use aws_smithy_http_client::Builder as SmithyHttpClientBuilder; +use http::header::{CONTENT_TYPE, HOST}; use reqwest::Client as HttpClient; +use reqwest::StatusCode; +use rustfs_signer::constants::UNSIGNED_PAYLOAD; +use rustfs_signer::sign_v4; +use s3s::Body; use std::ffi::OsStr; use std::fs as stdfs; use std::path::{Path, PathBuf}; @@ -75,6 +80,58 @@ pub fn local_http_client() -> HttpClient { .expect("failed to build local reqwest client") } +/// Signs and sends an admin HTTP request with the given credentials. +pub(crate) async fn admin_request( + base_url: &str, + method: http::Method, + path_and_query: &str, + body: Option, + access_key: &str, + secret_key: &str, +) -> Result<(StatusCode, String), Box> { + let url = format!("{base_url}{path_and_query}"); + let uri = url.parse::()?; + let authority = uri.authority().ok_or("admin URL missing authority")?.to_string(); + let mut request = http::Request::builder() + .method(method.clone()) + .uri(uri) + .header(HOST, authority) + .header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if body.is_some() { + request = request.header(CONTENT_TYPE, "application/json"); + } + + let content_length = i64::try_from(body.as_ref().map_or(0, String::len)).map_err(|_| "admin request body is too large")?; + let signed = sign_v4(request.body(Body::empty())?, content_length, access_key, secret_key, "", "us-east-1"); + + let mut request = local_http_client().request(method, &url); + for (name, value) in signed.headers() { + request = request.header(name, value); + } + if let Some(body) = body { + request = request.body(body); + } + let response = request.send().await?; + let status = response.status(); + let body = response.text().await?; + Ok((status, body)) +} + +/// Sends a root-credential admin request and returns its successful response body. +pub(crate) async fn admin_ok( + env: &RustFSTestEnvironment, + method: http::Method, + path_and_query: &str, + body: Option, +) -> Result> { + let (status, response_body) = + admin_request(&env.url, method.clone(), path_and_query, body, &env.access_key, &env.secret_key).await?; + if !status.is_success() { + return Err(format!("{method} {path_and_query} failed: {status} {response_body}").into()); + } + Ok(response_body) +} + /// Resolve the RustFS binary relative to the workspace. pub fn rustfs_binary_path() -> PathBuf { rustfs_binary_path_with_features(requested_rustfs_build_features().as_deref()) diff --git a/crates/e2e_test/src/sts_query_compat_test.rs b/crates/e2e_test/src/sts_query_compat_test.rs index 9637bc0e1..36484a050 100644 --- a/crates/e2e_test/src/sts_query_compat_test.rs +++ b/crates/e2e_test/src/sts_query_compat_test.rs @@ -12,22 +12,35 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::{RustFSTestEnvironment, init_logging, local_http_client}; +use crate::common::{RustFSTestEnvironment, admin_ok, init_logging}; use aws_sdk_sts::config::retry::RetryConfig; use aws_sdk_sts::config::{Credentials, Region}; use aws_sdk_sts::error::ProvideErrorMetadata; use aws_sdk_sts::operation::RequestId; use aws_sdk_sts::{Client, Config}; use aws_smithy_http_client::Builder as SmithyHttpClientBuilder; -use http::header::{CONTENT_TYPE, HOST}; -use rustfs_signer::constants::UNSIGNED_PAYLOAD; -use rustfs_signer::sign_v4; -use s3s::Body; +use bytes::Bytes; +use http::header::{AUTHORIZATION, CONTENT_TYPE}; +use http::{Request, Response}; +use http_body_util::{BodyExt, Full}; +use hyper::body::Incoming; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper_util::rt::TokioIo; +use serde_json::Value; use serial_test::serial; +use std::collections::BTreeSet; +use std::convert::Infallible; use std::error::Error; +use std::sync::Arc; +use tokio::net::TcpListener; +use tokio::sync::{Notify, mpsc}; +use tokio::task::{JoinHandle, JoinSet}; +use tokio::time::{Duration, timeout}; type BoxError = Box; type TestResult = Result<(), BoxError>; +const OPA_AUTH_TOKEN: &str = "sts-opa-token"; fn sts_client(url: &str, access_key: &str, secret_key: &str, session_token: Option<&str>) -> Client { let mut config = Config::builder() @@ -49,32 +62,14 @@ fn sts_client(url: &str, access_key: &str, secret_key: &str, session_token: Opti } async fn create_root_service_account(env: &RustFSTestEnvironment) -> Result<(String, String), BoxError> { - let path = "/rustfs/admin/v3/add-service-accounts"; - let url = format!("{}{path}", env.url); - let uri = url.parse::()?; - let authority = uri.authority().ok_or("admin URL missing authority")?.to_string(); - let body = serde_json::json!({ "targetUser": env.access_key.clone() }).to_string(); - let request = http::Request::builder() - .method(http::Method::PUT) - .uri(uri) - .header(HOST, authority) - .header(CONTENT_TYPE, "application/json") - .header("x-amz-content-sha256", UNSIGNED_PAYLOAD) - .body(Body::empty())?; - let content_length = i64::try_from(body.len()).map_err(|_| "service account request body is too large")?; - let signed = sign_v4(request, content_length, &env.access_key, &env.secret_key, "", "us-east-1"); - let mut request = local_http_client().put(&url); - for (name, value) in signed.headers() { - request = request.header(name, value); - } - let response = request.body(body).send().await?; - let status = response.status(); - let body = response.text().await?; - if !status.is_success() { - return Err(format!("create service account failed: {status} {body}").into()); - } - - let response: serde_json::Value = serde_json::from_str(&body)?; + let body = admin_ok( + env, + http::Method::PUT, + "/rustfs/admin/v3/add-service-accounts", + Some(serde_json::json!({ "targetUser": env.access_key.clone() }).to_string()), + ) + .await?; + let response: Value = serde_json::from_str(&body)?; let access_key = response["credentials"]["accessKey"] .as_str() .ok_or("service account response should contain credentials.accessKey")? @@ -86,28 +81,237 @@ async fn create_root_service_account(env: &RustFSTestEnvironment) -> Result<(Str Ok((access_key, secret_key)) } -async fn assert_chaining_denied(client: &Client, credential_kind: &str) -> TestResult { +async fn create_user_with_policy( + env: &RustFSTestEnvironment, + user: &str, + secret: &str, + policy_name: &str, + statements: Value, +) -> TestResult { + create_user(env, user, secret).await?; + admin_ok( + env, + http::Method::PUT, + &format!("/rustfs/admin/v3/add-canned-policy?name={policy_name}"), + Some( + serde_json::json!({ + "Version": "2012-10-17", + "Statement": statements, + }) + .to_string(), + ), + ) + .await?; + admin_ok( + env, + http::Method::POST, + "/rustfs/admin/v3/idp/builtin/policy/attach", + Some(serde_json::json!({ "policies": [policy_name], "user": user }).to_string()), + ) + .await?; + Ok(()) +} + +async fn create_user(env: &RustFSTestEnvironment, user: &str, secret: &str) -> TestResult { + admin_ok( + env, + http::Method::PUT, + &format!("/rustfs/admin/v3/add-user?accessKey={user}"), + Some(serde_json::json!({ "secretKey": secret, "status": "enabled" }).to_string()), + ) + .await?; + Ok(()) +} + +async fn assert_access_denied(client: &Client, context: &str) -> TestResult { let error = client .assume_role() .role_arn("arn:aws:iam::123456789012:role/test") .role_session_name("sts-query-compat-e2e") .send() .await - .expect_err("credential chaining must be denied"); + .expect_err("AssumeRole must be denied"); let service_error = error .as_service_error() - .ok_or_else(|| format!("{credential_kind} denial should deserialize as an STS service error: {error:?}"))?; + .ok_or_else(|| format!("{context} should deserialize as an STS service error: {error:?}"))?; assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(403)); assert_eq!(service_error.code(), Some("AccessDenied")); assert_eq!(service_error.message(), Some("Access Denied")); assert!( error.request_id().is_some_and(|request_id| !request_id.is_empty()), - "{credential_kind} denial should include a request ID" + "{context} should include a request ID" ); Ok(()) } +async fn handle_opa_request( + request: Request, + requests: mpsc::UnboundedSender, + validation_started: mpsc::UnboundedSender<()>, + validation_mode: OpaValidationMode, + expected_authorization: Option, +) -> Result>, Infallible> { + if let Some(expected_authorization) = expected_authorization + && request.headers().get(AUTHORIZATION).and_then(|value| value.to_str().ok()) != Some(expected_authorization.as_str()) + { + return Ok(Response::builder() + .status(401) + .body(Full::new(Bytes::new())) + .expect("static OPA unauthorized response must be valid")); + } + + let body = match request.into_body().collect().await { + Ok(body) => body.to_bytes(), + Err(error) => { + return Ok(Response::builder() + .status(400) + .body(Full::new(Bytes::from(error.to_string()))) + .expect("static OPA error response must be valid")); + } + }; + + let payload = if body.is_empty() { + None + } else { + match serde_json::from_slice::(&body) { + Ok(payload) => Some(payload), + Err(error) => { + return Ok(Response::builder() + .status(400) + .body(Full::new(Bytes::from(error.to_string()))) + .expect("static OPA error response must be valid")); + } + } + }; + if payload.is_none() { + let _ = validation_started.send(()); + if let OpaValidationMode::DelayedUnavailable(release) = validation_mode { + release.notified().await; + return Ok(Response::builder() + .status(503) + .body(Full::new(Bytes::new())) + .expect("static OPA unavailable response must be valid")); + } + } + let allow = match payload.as_ref().and_then(|value| value.pointer("/input/identity/account")) { + Some(Value::String(account)) if account == "opaallow" => payload + .as_ref() + .and_then(|value| value.pointer("/input/context/deny_only")) + .and_then(Value::as_bool) + .unwrap_or(false), + Some(Value::String(account)) if account == "opadeny" => false, + None => true, + _ => false, + }; + if let Some(payload) = payload { + let _ = requests.send(payload); + } + let body = + serde_json::to_vec(&serde_json::json!({ "result": { "allow": allow } })).expect("static OPA response must serialize"); + Ok(Response::builder() + .header(CONTENT_TYPE, "application/json") + .body(Full::new(Bytes::from(body))) + .expect("static OPA response must be valid")) +} + +#[derive(Clone)] +enum OpaValidationMode { + Ready, + DelayedUnavailable(Arc), +} + +struct OpaMock { + url: String, + requests: mpsc::UnboundedReceiver, + validation_started: mpsc::UnboundedReceiver<()>, + validation_release: Option>, + task: JoinHandle<()>, +} + +impl OpaMock { + async fn start() -> Result { + Self::start_with_mode(OpaValidationMode::Ready, Some(OPA_AUTH_TOKEN)).await + } + + async fn start_delayed_unavailable() -> Result { + let release = Arc::new(Notify::new()); + Self::start_with_mode(OpaValidationMode::DelayedUnavailable(release), None).await + } + + async fn start_with_mode(validation_mode: OpaValidationMode, auth_token: Option<&str>) -> Result { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let url = format!("http://{}/v1/data/rustfs/authz/allow", listener.local_addr()?); + let (requests_tx, requests) = mpsc::unbounded_channel(); + let (validation_started_tx, validation_started) = mpsc::unbounded_channel(); + let expected_authorization = auth_token.map(|token| format!("Bearer {token}")); + let validation_release = match &validation_mode { + OpaValidationMode::Ready => None, + OpaValidationMode::DelayedUnavailable(release) => Some(Arc::clone(release)), + }; + let task = tokio::spawn(async move { + let mut connections = JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((stream, _)) = accepted else { break }; + let requests = requests_tx.clone(); + let validation_started = validation_started_tx.clone(); + let validation_mode = validation_mode.clone(); + let expected_authorization = expected_authorization.clone(); + connections.spawn(async move { + let handler = service_fn(move |request| { + handle_opa_request( + request, + requests.clone(), + validation_started.clone(), + validation_mode.clone(), + expected_authorization.clone(), + ) + }); + let _ = http1::Builder::new() + .serve_connection(TokioIo::new(stream), handler) + .await; + }); + } + _ = connections.join_next(), if !connections.is_empty() => {} + } + } + }); + Ok(Self { + url, + requests, + validation_started, + validation_release, + task, + }) + } + + async fn next_request(&mut self) -> Result { + timeout(Duration::from_secs(5), self.requests.recv()) + .await? + .ok_or_else(|| "OPA request channel closed".into()) + } + + async fn wait_for_validation(&mut self) -> TestResult { + timeout(Duration::from_secs(5), self.validation_started.recv()) + .await? + .ok_or_else(|| "OPA validation channel closed".into()) + } + + fn release_validation(&self) { + if let Some(release) = &self.validation_release { + release.notify_one(); + } + } +} + +impl Drop for OpaMock { + fn drop(&mut self) { + self.task.abort(); + } +} + #[tokio::test] #[serial] async fn test_sts_query_responses_are_aws_sdk_compatible() -> TestResult { @@ -155,7 +359,7 @@ async fn test_sts_query_responses_are_aws_sdk_compatible() -> TestResult { "signature rejection should include a request ID" ); - assert_chaining_denied( + assert_access_denied( &sts_client( &env.url, temporary.access_key_id(), @@ -167,7 +371,187 @@ async fn test_sts_query_responses_are_aws_sdk_compatible() -> TestResult { .await?; let (service_access_key, service_secret_key) = create_root_service_account(&env).await?; - assert_chaining_denied(&sts_client(&env.url, &service_access_key, &service_secret_key, None), "service account").await?; + assert_access_denied( + &sts_client(&env.url, &service_access_key, &service_secret_key, None), + "service-account denial", + ) + .await?; + + let implicit_user = "stsimplicit"; + let explicit_allow_user = "stsallow"; + let explicit_deny_user = "stsdeny"; + let policyless_user = "stspolicyless"; + let secret = "stsAuthzSecret123"; + create_user(&env, policyless_user, secret).await?; + assert_access_denied(&sts_client(&env.url, policyless_user, secret, None), "policyless user").await?; + create_user_with_policy( + &env, + implicit_user, + secret, + "sts-implicit-policy", + serde_json::json!([{ + "Effect": "Allow", + "Action": ["s3:ListAllMyBuckets"], + "Resource": ["arn:aws:s3:::*"], + }]), + ) + .await?; + create_user_with_policy( + &env, + explicit_allow_user, + secret, + "sts-allow-policy", + serde_json::json!([{ + "Effect": "Allow", + "Action": ["sts:AssumeRole"], + "Resource": ["arn:aws:s3:::*"], + }]), + ) + .await?; + create_user_with_policy( + &env, + explicit_deny_user, + secret, + "sts-deny-policy", + serde_json::json!([ + { + "Effect": "Allow", + "Action": ["sts:AssumeRole"], + "Resource": ["arn:aws:s3:::*"], + }, + { + "Effect": "Deny", + "Action": ["sts:AssumeRole"], + "Resource": ["arn:aws:s3:::*"], + } + ]), + ) + .await?; + + for user in [implicit_user, explicit_allow_user] { + let output = sts_client(&env.url, user, secret, None) + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-authz-e2e") + .send() + .await + .map_err(|error| format!("{user} should be allowed to call AssumeRole: {error:?}"))?; + let credentials = output + .credentials() + .ok_or_else(|| format!("{user} AssumeRole response should contain credentials"))?; + assert!(!credentials.access_key_id().is_empty()); + assert!(!credentials.secret_access_key().is_empty()); + assert!(!credentials.session_token().is_empty()); + } + assert_access_denied(&sts_client(&env.url, explicit_deny_user, secret, None), "explicit sts:AssumeRole Deny").await?; + + env.stop_server(); + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_sts_assume_role_opa_contract() -> TestResult { + init_logging(); + + let mut opa = OpaMock::start().await?; + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_POLICY_PLUGIN_URL", opa.url.as_str()), + ("RUSTFS_POLICY_PLUGIN_AUTH_TOKEN", OPA_AUTH_TOKEN), + ], + ) + .await?; + + let secret = "stsOpaSecret123"; + create_user_with_policy( + &env, + "opaallow", + secret, + "sts-opa-local-deny-policy", + serde_json::json!([{ + "Effect": "Deny", + "Action": ["sts:AssumeRole"], + "Resource": ["arn:aws:s3:::*"], + }]), + ) + .await?; + create_user_with_policy( + &env, + "opadeny", + secret, + "sts-opa-local-allow-policy", + serde_json::json!([{ + "Effect": "Allow", + "Action": ["sts:AssumeRole"], + "Resource": ["arn:aws:s3:::*"], + }]), + ) + .await?; + + sts_client(&env.url, "opaallow", secret, None) + .assume_role() + .role_arn("arn:aws:iam::123456789012:role/test") + .role_session_name("sts-opa-contract") + .send() + .await + .map_err(|error| format!("OPA allow should override the local explicit Deny: {error:?}"))?; + assert_access_denied( + &sts_client(&env.url, "opadeny", secret, None), + "OPA denial despite local sts:AssumeRole Allow", + ) + .await?; + + let mut accounts = BTreeSet::new(); + for _ in 0..2 { + let request = opa.next_request().await?; + assert_eq!(request.pointer("/input/action").and_then(Value::as_str), Some("sts:AssumeRole")); + assert_eq!(request.pointer("/input/context/deny_only").and_then(Value::as_bool), Some(true)); + let account = request + .pointer("/input/identity/account") + .and_then(Value::as_str) + .ok_or("OPA input should include identity.account")?; + accounts.insert(account.to_owned()); + } + assert_eq!(accounts, BTreeSet::from(["opaallow".to_owned(), "opadeny".to_owned()])); + + env.stop_server(); + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_sts_assume_role_fails_closed_while_opa_is_unavailable() -> TestResult { + init_logging(); + + let mut opa = OpaMock::start_delayed_unavailable().await?; + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &[("RUSTFS_POLICY_PLUGIN_URL", opa.url.as_str())]) + .await?; + opa.wait_for_validation().await?; + + let user = "opaunavailable"; + let secret = "stsOpaUnavailableSecret123"; + create_user_with_policy( + &env, + user, + secret, + "sts-opa-unavailable-local-policy", + serde_json::json!([{ + "Effect": "Allow", + "Action": ["s3:ListAllMyBuckets"], + "Resource": ["arn:aws:s3:::*"], + }]), + ) + .await?; + + assert_access_denied(&sts_client(&env.url, user, secret, None), "configured OPA initialization").await?; + + opa.release_validation(); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_access_denied(&sts_client(&env.url, user, secret, None), "configured OPA validation failure").await?; env.stop_server(); Ok(()) diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 7a4822a1a..263e1f727 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -41,7 +41,7 @@ use rustfs_policy::policy::Args; use rustfs_policy::policy::opa; use rustfs_policy::policy::{Policy, PolicyDoc, iam_policy_claim_name_sa, policy_needs_existing_object_tag_for_args}; use serde_json::Value; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::sync::OnceLock; use time::OffsetDateTime; @@ -61,10 +61,45 @@ const VERIFIED_FEDERATED_POLICY_CLAIM: &str = "x-rustfs-internal-federated-polic const FEDERATED_POLICY_BOUNDARY_CLAIM: &str = "x-rustfs-internal-federated-boundary"; pub(crate) const SITE_REPLICATOR_CLAIM: &str = "site-replicator"; -static POLICY_PLUGIN_CLIENT: OnceLock>>> = OnceLock::new(); +#[derive(Clone)] +enum PolicyPluginState { + Disabled, + Initializing, + Ready(opa::AuthZPlugin), + Failed, +} -fn get_policy_plugin_client() -> Arc>> { - POLICY_PLUGIN_CLIENT.get_or_init(|| Arc::new(RwLock::new(None))).clone() +static POLICY_PLUGIN_STATE: OnceLock>> = OnceLock::new(); + +fn get_policy_plugin_state() -> Arc> { + POLICY_PLUGIN_STATE + .get_or_init(|| { + let configured = opa::is_configured(); + let state = Arc::new(RwLock::new(if configured { + PolicyPluginState::Initializing + } else { + PolicyPluginState::Disabled + })); + if configured { + let state = Arc::clone(&state); + tokio::spawn(async move { + let next_state = match opa::lookup_config().await { + Ok(conf) if conf.enable() => { + info!("OPA plugin enabled"); + PolicyPluginState::Ready(opa::AuthZPlugin::new(conf)) + } + Ok(_) => PolicyPluginState::Failed, + Err(e) => { + error!("Error loading OPA configuration err:{}", e); + PolicyPluginState::Failed + } + }; + *state.write().await = next_state; + }); + } + state + }) + .clone() } pub struct IamSys { @@ -173,19 +208,7 @@ impl IamSys { /// # Returns /// A new instance of IamSys pub fn new(store: Arc>) -> Self { - tokio::spawn(async move { - match opa::lookup_config().await { - Ok(conf) => { - if conf.enable() { - Self::set_policy_plugin_client(opa::AuthZPlugin::new(conf)).await; - info!("OPA plugin enabled"); - } - } - Err(e) => { - error!("Error loading OPA configuration err:{}", e); - } - }; - }); + get_policy_plugin_state(); Self { store, @@ -206,15 +229,22 @@ impl IamSys { } pub async fn set_policy_plugin_client(client: rustfs_policy::policy::opa::AuthZPlugin) { - let policy_plugin_client = get_policy_plugin_client(); - let mut guard = policy_plugin_client.write().await; - *guard = Some(client); + let policy_plugin_state = get_policy_plugin_state(); + let mut guard = policy_plugin_state.write().await; + *guard = PolicyPluginState::Ready(client); } pub async fn get_policy_plugin_client() -> Option { - let policy_plugin_client = get_policy_plugin_client(); - let guard = policy_plugin_client.read().await; - guard.clone() + let policy_plugin_state = get_policy_plugin_state(); + let guard = policy_plugin_state.read().await; + match &*guard { + PolicyPluginState::Ready(client) => Some(client.clone()), + PolicyPluginState::Disabled | PolicyPluginState::Initializing | PolicyPluginState::Failed => None, + } + } + + async fn policy_plugin_state() -> PolicyPluginState { + get_policy_plugin_state().read().await.clone() } pub async fn load_group(&self, name: &str) -> Result<()> { @@ -1060,11 +1090,20 @@ impl IamSys { }; } - if Self::get_policy_plugin_client().await.is_some() { - return PreparedIamAuth { - needs_existing_object_tag: false, - mode: PreparedIamMode::Opa, - }; + match Self::policy_plugin_state().await { + PolicyPluginState::Ready(_) => { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Opa, + }; + } + PolicyPluginState::Initializing | PolicyPluginState::Failed => { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + PolicyPluginState::Disabled => {} } let Ok((is_svc, parent_user)) = self.is_service_account(args.account).await else { @@ -1163,7 +1202,22 @@ impl IamSys { }; } - let combined_policy = self.get_combined_policy(&policies).await; + let (resolved_policies, combined_policy) = self.store.merge_policies(&policies.join(",")).await; + if args.deny_only { + let resolved_policy_names: HashSet<&str> = resolved_policies + .split(',') + .filter(|policy_name| !policy_name.trim().is_empty()) + .collect(); + if policies + .iter() + .any(|policy_name| !resolved_policy_names.contains(policy_name.as_str())) + { + return PreparedIamAuth { + needs_existing_object_tag: false, + mode: PreparedIamMode::Deny, + }; + } + } PreparedIamAuth { needs_existing_object_tag: policy_needs_existing_object_tag_for_args(&combined_policy, args).await, mode: PreparedIamMode::Regular { combined_policy }, @@ -1703,7 +1757,7 @@ mod tests { fn deprecated_list_polices_api_is_available() { let _ = IamSys::::list_polices; } - use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; + use rustfs_policy::policy::action::{Action, AdminAction, S3Action, StsAction}; use rustfs_policy::policy::policy_uses_existing_object_tag_conditions; use serde_json::Value; use std::{ @@ -3309,6 +3363,155 @@ mod tests { assert!(!iam_sys.eval_prepared(&prepared, &args).await); } + #[tokio::test] + async fn regular_deny_only_rejects_unresolved_mapped_policy() { + let iam_sys = test_iam_sys().await; + let user = "regular-unresolved-policy"; + let identity = UserIdentity::from(Credentials { + access_key: user.to_string(), + secret_key: "longenoughsecret".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + iam_sys.store.cache.with_write_lock(|cache| { + let now = OffsetDateTime::now_utc(); + cache.add_or_update_user(user, &identity, now); + cache.add_or_update_user_policy(user, &MappedPolicy::new("readwrite,missing-policy"), now); + }); + + let groups = None; + let claims = HashMap::new(); + let conditions = HashMap::new(); + let args = Args { + account: user, + groups: &groups, + action: Action::StsAction(StsAction::AssumeRoleAction), + bucket: "", + conditions: &conditions, + is_owner: false, + object: "", + claims: &claims, + deny_only: true, + }; + + let prepared = iam_sys.prepare_regular_auth(&args).await; + assert!(matches!(prepared.mode, PreparedIamMode::Deny)); + assert!( + !iam_sys.eval_prepared(&prepared, &args).await, + "deny-only evaluation must fail closed when a mapped policy document is unresolved" + ); + } + + #[tokio::test] + async fn regular_full_evaluation_keeps_resolved_part_of_partial_mapping() { + let iam_sys = test_iam_sys().await; + let user = "regular-partial-policy"; + let identity = UserIdentity::from(Credentials { + access_key: user.to_string(), + secret_key: "longenoughsecret".to_string(), + status: ACCOUNT_ON.to_string(), + ..Default::default() + }); + iam_sys.store.cache.with_write_lock(|cache| { + let now = OffsetDateTime::now_utc(); + cache.add_or_update_user(user, &identity, now); + cache.add_or_update_user_policy(user, &MappedPolicy::new("readwrite,missing-policy"), now); + }); + + let groups = None; + let claims = HashMap::new(); + let conditions = HashMap::new(); + let args = Args { + account: user, + groups: &groups, + action: Action::S3Action(S3Action::GetObjectAction), + bucket: "bucket", + conditions: &conditions, + is_owner: false, + object: "object", + claims: &claims, + deny_only: false, + }; + + let prepared = iam_sys.prepare_regular_auth(&args).await; + assert!(matches!(prepared.mode, PreparedIamMode::Regular { .. })); + assert!( + iam_sys.eval_prepared(&prepared, &args).await, + "full evaluation must preserve the historical partial-resolution behavior" + ); + } + + #[tokio::test] + async fn regular_deny_only_honors_group_derived_explicit_deny() { + let iam_sys = test_iam_sys().await; + let deny_policy = + Policy::parse_config(br#"{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["sts:AssumeRole"]}]}"#) + .expect("group deny policy should parse"); + let now = OffsetDateTime::now_utc(); + iam_sys + .store + .cache + .add_or_update_policy_doc("deny-assume-role", &PolicyDoc::new(deny_policy), now); + iam_sys + .store + .cache + .add_or_update_group_policy("testgroup", &MappedPolicy::new("deny-assume-role"), now); + + let groups = Some(vec!["testgroup".to_string()]); + let claims = HashMap::new(); + let conditions = HashMap::new(); + let args = Args { + account: "sts-fallback-test-parent", + groups: &groups, + action: Action::StsAction(StsAction::AssumeRoleAction), + bucket: "", + conditions: &conditions, + is_owner: false, + object: "", + claims: &claims, + deny_only: true, + }; + + let prepared = iam_sys.prepare_regular_auth(&args).await; + assert!(matches!(prepared.mode, PreparedIamMode::Regular { .. })); + assert!( + !iam_sys.eval_prepared(&prepared, &args).await, + "group-derived explicit Deny must remain authoritative" + ); + } + + #[tokio::test] + async fn regular_deny_only_rejects_unresolved_group_mapping() { + let iam_sys = test_iam_sys().await; + iam_sys.store.cache.add_or_update_group_policy( + "testgroup", + &MappedPolicy::new("readwrite,missing-policy"), + OffsetDateTime::now_utc(), + ); + + let groups = Some(vec!["testgroup".to_string()]); + let claims = HashMap::new(); + let conditions = HashMap::new(); + let args = Args { + account: "sts-fallback-test-parent", + groups: &groups, + action: Action::StsAction(StsAction::AssumeRoleAction), + bucket: "", + conditions: &conditions, + is_owner: false, + object: "", + claims: &claims, + deny_only: true, + }; + + let prepared = iam_sys.prepare_regular_auth(&args).await; + assert!(matches!(prepared.mode, PreparedIamMode::Deny)); + assert!( + !iam_sys.eval_prepared(&prepared, &args).await, + "deny-only evaluation must fail closed for an unresolved group-derived mapping" + ); + } + #[tokio::test] async fn test_sts_claim_policy_ignores_unsafe_and_missing_policy_names() { let store = StsTestMockStore::new(true); diff --git a/crates/policy/src/policy/opa.rs b/crates/policy/src/policy/opa.rs index 5ff6e3ea0..c1f70bcbb 100644 --- a/crates/policy/src/policy/opa.rs +++ b/crates/policy/src/policy/opa.rs @@ -30,6 +30,10 @@ impl Args { } } +pub fn is_configured() -> bool { + env::var_os(ENV_POLICY_PLUGIN_OPA_URL).is_some_and(|url| !url.is_empty()) +} + #[derive(Debug, Clone)] pub struct AuthZPlugin { client: OpaHttpClient, @@ -88,7 +92,12 @@ async fn validate(config: &Args) -> Result<(), OpaConfigError> { .build() .map_err(OpaConfigError::Connection)?; - match client.post(&config.url).send().await { + let mut request = client.post(&config.url); + if !config.auth_token.is_empty() { + request = request.header("Authorization", format!("Bearer {}", config.auth_token)); + } + + match request.send().await { Ok(resp) => { match resp.status() { reqwest::StatusCode::OK => { diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index ec71b2729..8b2192e35 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -9,7 +9,7 @@ > (`.config/nextest.toml`); admission criteria: `crates/e2e_test/README.md`. > ๐ŸŒ™ marks tests in the scheduled `e2e-repl-nightly` profile (backlog#1147 > repl-1): `replication_extension_test` splits 20 fast tests into the PR smoke -> lane and 27 slow / `_real_dual_node` / `_real_three_node` / `_real_single_node` tests into the +> lane and 28 slow / `_real_dual_node` / `_real_three_node` / `_real_single_node` tests into the > nightly lane (`.github/workflows/e2e-replication-nightly.yml`). > Note: counts exclude `#[ignore]`d tests (nextest lists them separately). > The SSE-S3 replication contract is ignored under backlog#1291 until its @@ -17,11 +17,11 @@ | module | tests | PR smoke | |---|---|---| -| admin_auth_test | 3 | โœ… | +| admin_auth_test | 4 | โœ… | | admin_iam_crud_test | 2 | โœ… | | admin_pools_test | 1 | โœ… | | admin_timeout_regression_test | 1 | | -| anonymous_access_test | 3 | โœ… | +| anonymous_access_test | 4 | โœ… | | api_rate_limit_test | 3 | | | archive_download_integrity_test | 13 | | | bucket_logging_test | 3 | | @@ -29,7 +29,7 @@ | checksum_upload_test | 7 | | | cluster_concurrency_test | 2 | | | cluster_multidrive_pool_test | 2 | | -| common | 10 | | +| common | 12 | | | compression_test | 1 | | | connection_cap_test | 2 | | | console_smoke_test | 1 | โœ… | @@ -51,7 +51,9 @@ | head_object_consistency_test | 1 | โœ… | | head_object_range_test | 1 | โœ… | | heal_erasure_disk_rebuild_test | 3 | | -| kms | 40 | | +| inline_fast_path_cluster_test | 14 | | +| internode_rpc_signature_e2e_test | 5 | | +| kms | 41 | | | leading_slash_key_test | 2 | โœ… | | list_buckets_double_slash_test | 3 | โœ… | | list_object_versions_metadata_extension_test | 1 | | @@ -72,7 +74,7 @@ | protocols | 16 | | | quota_test | 14 | | | reliability_disk_fault_test | 3 | | -| reliant | 10 | 5 โœ… | +| reliant | 24 | 18 โœ… | | replication_extension_test | 48 | 20 โœ… +28 ๐ŸŒ™ | | security_boundary_test | 4 | | | ssec_copy_test | 2 | โœ… | @@ -81,10 +83,11 @@ | special_chars_test | 14 | โœ… | | stale_multipart_cleanup_cluster_test | 1 | | | storage_class_capability_test | 4 | โœ… | +| sts_query_compat_test | 3 | โœ… | | tls_gen | 3 | | | tls_hot_reload_test | 1 | โœ… | | version_id_regression_test | 10 | โœ… | `notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above. -**Total listed: 486 tests across 67 modules ยท PR smoke subset: 129 tests / 32 modules** (30 full modules + 5 `reliant` tests + 20 of `replication_extension_test`) **ยท nightly `e2e-repl-nightly`: 28 tests** ยท generated 2026-07-26. +**Total listed: 527 tests across 70 modules ยท PR smoke subset: 147 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **ยท nightly `e2e-repl-nightly`: 28 tests** ยท generated 2026-07-30. diff --git a/rustfs/src/admin/handlers/sts.rs b/rustfs/src/admin/handlers/sts.rs index c3d83df26..60a1911c9 100644 --- a/rustfs/src/admin/handlers/sts.rs +++ b/rustfs/src/admin/handlers/sts.rs @@ -183,7 +183,7 @@ async fn handle_assume_role( conditions: &conditions, is_owner: owner, claims: cred.claims_or_empty(), - deny_only: false, + deny_only: true, bucket: "", object: "", })