Merge branch 'main' into houseme/health-readiness-contract-followup

This commit is contained in:
houseme
2026-08-25 09:13:33 +08:00
committed by GitHub
42 changed files with 3378 additions and 706 deletions
+1
View File
@@ -34,6 +34,7 @@ script-tests: ## Run shell script tests
./scripts/test_exact_1mib_handoff_abba.sh
./scripts/test_pinned_paired_abba_bench.sh
./scripts/test_manual_transition_runbooks.sh
./scripts/test_fuzz_runner.sh
./scripts/check_embedded_secrets.sh --self-test
python3 ./scripts/check_test_wiring.py --self-test
python3 ./scripts/check_security_coverage.py --self-test
+29 -42
View File
@@ -181,22 +181,14 @@ jobs:
needs: [ quick-checks ]
runs-on: sm-standard-4
timeout-minutes: 90
# Both lines are required. Job-level `permissions` replaces the workflow
# block rather than merging with it, so declaring only `actions: write`
# would drop `contents: read` and break this job's checkout and the
# repo-token the setup action hands to setup-protoc.
permissions:
contents: read
actions: write
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
# This job's token can cancel runs and delete Actions caches. Checkout
# otherwise writes it into .git/config, where a PR's own build.rs or
# proc-macro could read it back out.
# Checkout otherwise writes the token into .git/config, where a PR's
# own build.rs or proc-macro could read it back out.
persist-credentials: false
- name: Setup Rust environment
@@ -347,41 +339,36 @@ jobs:
- name: Run rebalance/decommission migration proofs
run: ./scripts/check_migration_gate_count.sh
# Early stop. Once this job has failed the PR cannot merge, so the sibling
# lanes are burning runners on a result nobody can act on: on run
# 30674613104 three lanes had already failed while Test and Lint and the
# rio-v2 variant kept going past 70 minutes.
#
# Only this job may cancel. The lanes that are NOT required checks
# (protocols, ILM, e2e, s3-tests) must never hold that power: a flake in
# one of them would turn the required "Test and Lint" into `cancelled`,
# which blocks the merge. Today a maintainer can merge with sftp red, and
# that has to stay true.
#
# These steps run last so the `if: always()` artifact upload above still
# captures logs and diagnostics before the run goes away.
# Record the reason before this job completes as FAILURE. A separate
# dependent job cancels sibling lanes only after GitHub has preserved this
# required check's failure verdict.
- name: Annotate early-stop reason
if: failure() && github.event_name == 'pull_request'
run: |
{
echo "## CI early-stop"
echo "Job \`${GITHUB_JOB}\` (Test and Lint) failed; cancelling run ${GITHUB_RUN_ID} to free runners."
echo "Sibling jobs showing **cancelled** were stopped by this job, not by their own failure."
} >> "$GITHUB_STEP_SUMMARY"
# curl rather than `gh`: every existing `gh` call in this repo runs on
# ubuntu-latest, and the sm-standard-* images are custom and trimmed (they
# ship no C toolchain, see the e2e job below), so `gh` is not known to
# exist here.
#
# Fork PRs are excluded explicitly instead of relying on the error path:
# their GITHUB_TOKEN is forced read-only and job-level permissions cannot
# raise it, so the call would always 403. Skipping keeps their logs clean.
- name: Cancel run on failure (same-repo PR only)
if: >-
failure() && github.event_name == 'pull_request'
&& github.event.pull_request.head.repo.full_name == github.repository
continue-on-error: true
run: |
{
echo "## CI early-stop"
echo "Job \`${GITHUB_JOB}\` (Test and Lint) failed; a follow-up job will cancel sibling lanes to free runners."
echo "Sibling jobs showing **cancelled** were stopped by the early-stop follow-up, not by their own failure."
} >> "$GITHUB_STEP_SUMMARY"
# Preserve the required Test and Lint FAILURE verdict before stopping sibling
# lanes. Cancelling from inside test-and-lint changed its own conclusion to
# CANCELLED and hid the actionable failure in the PR checks UI.
cancel-after-test-and-lint-failure:
name: Cancel siblings after Test and Lint failure
if: >-
failure() && needs.test-and-lint.result == 'failure'
&& github.event_name == 'pull_request'
&& github.event.pull_request.head.repo.full_name == github.repository
needs: [ test-and-lint ]
runs-on: ubuntu-latest
timeout-minutes: 5
permissions:
actions: write
steps:
- name: Cancel remaining jobs
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
@@ -389,7 +376,7 @@ jobs:
-H "Authorization: Bearer ${GH_TOKEN}" \
-H "Accept: application/vnd.github+json" \
-H "X-GitHub-Api-Version: 2022-11-28" \
"${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/cancel" || true
"${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/cancel"
# Dedicated serial lane for the ILM / lifecycle integration tests. These tests
# drive the object layer through process-global singletons (the GLOBAL_ENV
+105
View File
@@ -0,0 +1,105 @@
# Copyright 2024 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.
name: Upgrade Compatibility
on:
pull_request:
paths:
- ".github/workflows/e2e-upgrade.yml"
- "crates/e2e_test/src/common.rs"
- "crates/e2e_test/src/lib.rs"
- "crates/e2e_test/src/upgrade_compatibility_test.rs"
- "crates/ecstore/**"
- "crates/filemeta/**"
- "crates/kms/**"
- "crates/storage-api/**"
- "rustfs/**"
- "Cargo.lock"
push:
tags:
- "[0-9]*.[0-9]*.[0-9]*"
schedule:
- cron: "17 3 * * 1"
workflow_dispatch:
permissions:
contents: read
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
env:
CARGO_TERM_COLOR: always
RUST_BACKTRACE: 1
UPGRADE_SOURCE_VERSION: 1.0.0-rc.2
UPGRADE_SOURCE_ASSET: rustfs-linux-x86_64-gnu-v1.0.0-rc.2.zip
UPGRADE_SOURCE_SHA256: 7c789386bf85278f865b8e0d359bf4edb84d5aa408cc3fa54a18c25ca74cd6e7
jobs:
direct-upgrade:
name: Direct upgrade from rc.2
runs-on: ubuntu-latest
timeout-minutes: 60
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Setup Rust environment
uses: ./.github/actions/setup
with:
cache-shared-key: e2e-direct-upgrade
cache-save-if: ${{ github.ref == 'refs/heads/main' }}
install-build-packaging-tools: "false"
- name: Download pinned previous release
env:
SOURCE_DIR: ${{ runner.temp }}/rustfs-upgrade-source
run: |
set -euo pipefail
mkdir -p "$SOURCE_DIR"
archive="$SOURCE_DIR/$UPGRADE_SOURCE_ASSET"
curl --fail --location --retry 3 --output "$archive" \
"https://github.com/${GITHUB_REPOSITORY}/releases/download/${UPGRADE_SOURCE_VERSION}/${UPGRADE_SOURCE_ASSET}"
echo "$UPGRADE_SOURCE_SHA256 $archive" | sha256sum --check --strict
unzip -q "$archive" -d "$SOURCE_DIR"
chmod +x "$SOURCE_DIR/rustfs"
test -x "$SOURCE_DIR/rustfs"
echo "RUSTFS_UPGRADE_SOURCE_BINARY=$SOURCE_DIR/rustfs" >> "$GITHUB_ENV"
echo "RUSTFS_E2E_LOG_DIR=$RUNNER_TEMP/rustfs-upgrade-logs" >> "$GITHUB_ENV"
- name: Build current RustFS binary
run: |
cargo build --locked -p rustfs --bin rustfs
: > target/debug/rustfs.features
- name: Run direct-upgrade compatibility test
run: |
cargo test --locked -p e2e_test \
upgrade_compatibility_test::direct_upgrade_from_rc2_preserves_object_contracts \
-- --ignored --exact --nocapture
- name: Upload server logs
if: always()
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
with:
name: direct-upgrade-server-logs-${{ github.run_number }}
path: ${{ runner.temp }}/rustfs-upgrade-logs
if-no-files-found: warn
retention-days: 14
+2 -2
View File
@@ -173,7 +173,7 @@ jobs:
path: |
fuzz/artifacts/**
fuzz/corpus/${{ matrix.target }}/**
if-no-files-found: ignore
if-no-files-found: error
retention-days: 7
# ──────────────────────────────────────────────────────────────
@@ -227,7 +227,7 @@ jobs:
path: |
fuzz/artifacts/**
fuzz/corpus/${{ matrix.target }}/**
if-no-files-found: ignore
if-no-files-found: error
retention-days: 30
# ──────────────────────────────────────────────────────────────
+110
View File
@@ -0,0 +1,110 @@
# Copyright 2024 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.
name: OIDC Keycloak Live
on:
pull_request:
paths:
- ".github/workflows/oidc-keycloak.yml"
- "crates/config/src/constants/oidc.rs"
- "crates/iam/src/federation/**"
- "crates/iam/src/oidc.rs"
- "rustfs/src/admin/handlers/oidc.rs"
- "rustfs/src/admin/handlers/sts.rs"
- "scripts/test/oidc_keycloak_live.sh"
- "scripts/test/fixtures/keycloak-rustfs-ci-realm.json"
push:
branches: [main]
paths:
- ".github/workflows/oidc-keycloak.yml"
- "crates/config/src/constants/oidc.rs"
- "crates/iam/src/federation/**"
- "crates/iam/src/oidc.rs"
- "rustfs/src/admin/handlers/oidc.rs"
- "rustfs/src/admin/handlers/sts.rs"
- "scripts/test/oidc_keycloak_live.sh"
- "scripts/test/fixtures/keycloak-rustfs-ci-realm.json"
schedule:
- cron: "23 2 * * 1"
timezone: "Asia/Shanghai"
workflow_dispatch:
permissions:
contents: read
concurrency:
group: oidc-keycloak-live-${{ github.ref }}
cancel-in-progress: true
jobs:
oidc-keycloak-live:
name: OIDC Keycloak live gate
runs-on: ubuntu-latest
timeout-minutes: 60
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Setup Rust environment
uses: ./.github/actions/setup
with:
cache-shared-key: oidc-keycloak-live
cache-save-if: "true"
install-build-packaging-tools: "false"
install-test-tools: "false"
- name: Build RustFS
run: cargo build --locked -p rustfs --bin rustfs
- name: Install pinned request signer
run: |
python3 -m pip install --user --upgrade pip "awscurl==0.44"
echo "${HOME}/.local/bin" >> "${GITHUB_PATH}"
- name: Run live Keycloak discovery, JWT and STS checks
run: bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs
- name: Upload service logs
if: failure()
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
with:
name: oidc-keycloak-live-${{ github.run_number }}
path: ${{ runner.temp }}/rustfs-keycloak-live-*/**/*.log
if-no-files-found: ignore
retention-days: 3
alert-on-failure:
name: Alert on scheduled failure
needs: oidc-keycloak-live
if: >-
always() && github.event_name == 'schedule' &&
(needs.oidc-keycloak-live.result == 'failure' || needs.oidc-keycloak-live.result == 'cancelled')
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update failure-tracking issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
+2
View File
@@ -27,6 +27,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
| **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) |
| **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` |
| **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
| **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
## How to run
@@ -168,6 +169,7 @@ the same profile for membership and execution with one nightly worker.
| `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) |
| ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) |
| KMS suite | `e2e-full` job, merge queue + main | **Active** |
| Direct upgrade from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** |
| Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
| Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
| Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
+45 -91
View File
@@ -16,8 +16,10 @@
#[cfg(test)]
mod tests {
use std::borrow::Borrow;
use crate::common::{RustFSTestEnvironment, init_logging, signed_s3_request};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::types::{
AccelerateConfiguration, BucketAccelerateStatus, BucketLoggingStatus, IndexDocument, LoggingEnabled, Payer,
RequestPaymentConfiguration, WebsiteConfiguration,
@@ -26,6 +28,26 @@ mod tests {
use http::header::CONTENT_TYPE;
use tracing::info;
fn assert_s3_error<T, E, R>(result: Result<T, R>, expected_status: u16, expected_code: &str, context: &str)
where
T: std::fmt::Debug,
E: ProvideErrorMetadata + std::fmt::Debug,
R: Borrow<SdkError<E>> + std::fmt::Debug,
{
let error = result.expect_err(context);
let sdk_error = error.borrow();
assert_eq!(
sdk_error.raw_response().map(|response| response.status().as_u16()),
Some(expected_status),
"{context}: expected HTTP {expected_status}, got: {error:?}"
);
assert_eq!(
sdk_error.as_service_error().and_then(ProvideErrorMetadata::code),
Some(expected_code),
"{context}: expected {expected_code}, got: {error:?}"
);
}
#[tokio::test]
async fn test_dummy_bucket_compatibility_endpoints() {
init_logging();
@@ -217,17 +239,11 @@ mod tests {
.expect("DeleteBucketWebsite should return success");
let website_after_delete = client.get_bucket_website().bucket(bucket).send().await;
assert!(
website_after_delete.is_err(),
"GetBucketWebsite should return NoSuchWebsiteConfiguration after deletion"
);
let website_err = website_after_delete.err().unwrap();
let website_code = website_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(website_code, Some("NoSuchWebsiteConfiguration")),
"Unexpected GetBucketWebsite error code: {:?}, err: {:?}",
website_code,
website_err
assert_s3_error(
website_after_delete,
404,
"NoSuchWebsiteConfiguration",
"GetBucketWebsite after deleting the website configuration",
);
env.stop_server();
@@ -245,15 +261,7 @@ mod tests {
let missing_bucket = "test-dummy-bucket-missing";
let get_logging = client.get_bucket_logging().bucket(missing_bucket).send().await;
assert!(get_logging.is_err(), "GetBucketLogging should fail for missing bucket");
let get_logging_err = get_logging.err().unwrap();
let get_logging_code = get_logging_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(get_logging_code, Some("NoSuchBucket")),
"Unexpected GetBucketLogging error code: {:?}, err: {:?}",
get_logging_code,
get_logging_err
);
assert_s3_error(get_logging, 404, "NoSuchBucket", "GetBucketLogging for a missing bucket");
let put_logging = client
.put_bucket_logging()
@@ -261,41 +269,22 @@ mod tests {
.bucket_logging_status(BucketLoggingStatus::builder().build())
.send()
.await;
assert!(put_logging.is_err(), "PutBucketLogging should fail for missing bucket");
let put_logging_err = put_logging.err().unwrap();
let put_logging_code = put_logging_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(put_logging_code, Some("NoSuchBucket")),
"Unexpected PutBucketLogging error code: {:?}, err: {:?}",
put_logging_code,
put_logging_err
);
assert_s3_error(put_logging, 404, "NoSuchBucket", "PutBucketLogging for a missing bucket");
let get_accelerate = client
.get_bucket_accelerate_configuration()
.bucket(missing_bucket)
.send()
.await;
assert!(get_accelerate.is_err(), "GetBucketAccelerateConfiguration should fail for missing bucket");
let get_accelerate_err = get_accelerate.err().unwrap();
let get_accelerate_code = get_accelerate_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(get_accelerate_code, Some("NoSuchBucket")),
"Unexpected GetBucketAccelerateConfiguration error code: {:?}, err: {:?}",
get_accelerate_code,
get_accelerate_err
assert_s3_error(
get_accelerate,
404,
"NoSuchBucket",
"GetBucketAccelerateConfiguration for a missing bucket",
);
let get_request_payment = client.get_bucket_request_payment().bucket(missing_bucket).send().await;
assert!(get_request_payment.is_err(), "GetBucketRequestPayment should fail for missing bucket");
let get_request_payment_err = get_request_payment.err().unwrap();
let get_request_payment_code = get_request_payment_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(get_request_payment_code, Some("NoSuchBucket")),
"Unexpected GetBucketRequestPayment error code: {:?}, err: {:?}",
get_request_payment_code,
get_request_payment_err
);
assert_s3_error(get_request_payment, 404, "NoSuchBucket", "GetBucketRequestPayment for a missing bucket");
let put_accelerate = client
.put_bucket_accelerate_configuration()
@@ -307,14 +296,11 @@ mod tests {
)
.send()
.await;
assert!(put_accelerate.is_err(), "PutBucketAccelerateConfiguration should fail for missing bucket");
let put_accelerate_err = put_accelerate.err().unwrap();
let put_accelerate_code = put_accelerate_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(put_accelerate_code, Some("NoSuchBucket")),
"Unexpected PutBucketAccelerateConfiguration error code: {:?}, err: {:?}",
put_accelerate_code,
put_accelerate_err
assert_s3_error(
put_accelerate,
404,
"NoSuchBucket",
"PutBucketAccelerateConfiguration for a missing bucket",
);
let put_request_payment = client
@@ -328,15 +314,7 @@ mod tests {
)
.send()
.await;
assert!(put_request_payment.is_err(), "PutBucketRequestPayment should fail for missing bucket");
let put_request_payment_err = put_request_payment.err().unwrap();
let put_request_payment_code = put_request_payment_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(put_request_payment_code, Some("NoSuchBucket")),
"Unexpected PutBucketRequestPayment error code: {:?}, err: {:?}",
put_request_payment_code,
put_request_payment_err
);
assert_s3_error(put_request_payment, 404, "NoSuchBucket", "PutBucketRequestPayment for a missing bucket");
let put_website = client
.put_bucket_website()
@@ -353,37 +331,13 @@ mod tests {
)
.send()
.await;
assert!(put_website.is_err(), "PutBucketWebsite should fail for missing bucket");
let put_website_err = put_website.err().unwrap();
let put_website_code = put_website_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(put_website_code, Some("NoSuchBucket")),
"Unexpected PutBucketWebsite error code: {:?}, err: {:?}",
put_website_code,
put_website_err
);
assert_s3_error(put_website, 404, "NoSuchBucket", "PutBucketWebsite for a missing bucket");
let get_website = client.get_bucket_website().bucket(missing_bucket).send().await;
assert!(get_website.is_err(), "GetBucketWebsite should fail for missing bucket");
let get_website_err = get_website.err().unwrap();
let get_website_code = get_website_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(get_website_code, Some("NoSuchBucket")),
"Unexpected GetBucketWebsite error code: {:?}, err: {:?}",
get_website_code,
get_website_err
);
assert_s3_error(get_website, 404, "NoSuchBucket", "GetBucketWebsite for a missing bucket");
let delete_website = client.delete_bucket_website().bucket(missing_bucket).send().await;
assert!(delete_website.is_err(), "DeleteBucketWebsite should fail for missing bucket");
let delete_website_err = delete_website.err().unwrap();
let delete_website_code = delete_website_err.as_service_error().and_then(|e| e.code());
assert!(
matches!(delete_website_code, Some("NoSuchBucket")),
"Unexpected DeleteBucketWebsite error code: {:?}, err: {:?}",
delete_website_code,
delete_website_err
);
assert_s3_error(delete_website, 404, "NoSuchBucket", "DeleteBucketWebsite for a missing bucket");
env.stop_server();
}
@@ -17,6 +17,7 @@
use crate::common::{RustFSTestEnvironment, init_logging};
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::{Client, Config};
use tracing::info;
@@ -73,10 +74,14 @@ async fn test_bucket_policy_authenticated_user() -> Result<(), Box<dyn std::erro
let user_client = create_user_client(&env, user_access, user_secret);
// 4. Verify Access Denied initially (No Policy)
let result = user_client.list_objects_v2().bucket(bucket_name).send().await;
if result.is_ok() {
return Err("Should be Access Denied initially".into());
}
let denied = user_client
.list_objects_v2()
.bucket(bucket_name)
.send()
.await
.expect_err("a user without a bucket policy must be denied");
assert_eq!(denied.raw_response().map(|response| response.status().as_u16()), Some(403));
assert_eq!(denied.as_service_error().and_then(ProvideErrorMetadata::code), Some("AccessDenied"));
// 5. Apply Bucket Policy Allowed User
let policy_json = serde_json::json!({
+20 -14
View File
@@ -20,6 +20,7 @@ mod tests {
use crate::common::{RustFSTestEnvironment, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::config::{Credentials, Region, RequestChecksumCalculation};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart};
use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
@@ -186,16 +187,16 @@ mod tests {
.send()
.await;
assert!(
result.is_err(),
"PutObject with a mismatched SHA256 must be rejected, but it succeeded (issue #4341)"
let error = result.expect_err("PutObject with a mismatched SHA256 must be rejected (issue #4341)");
assert_eq!(
error.raw_response().map(|response| response.status().as_u16()),
Some(400),
"Mismatched SHA256 must return HTTP 400, got {error:?}"
);
let err = result.err().unwrap();
let msg = format!("{err:?}");
info!("PutObject correctly rejected mismatched checksum: {msg}");
assert!(
msg.contains("BadDigest") || msg.to_lowercase().contains("digest") || msg.to_lowercase().contains("checksum"),
"Expected a BadDigest/checksum error, got: {msg}"
assert_eq!(
error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("BadDigest"),
"Mismatched SHA256 must return BadDigest, got {error:?}"
);
// And the object must not have been stored.
@@ -556,11 +557,16 @@ mod tests {
})
.send()
.await;
assert!(put_bad.is_err(), "{header}: a mismatched checksum must be rejected");
let msg = format!("{:?}", put_bad.err().unwrap());
assert!(
msg.contains("BadDigest") || msg.to_lowercase().contains("digest") || msg.to_lowercase().contains("checksum"),
"{header}: expected a BadDigest/checksum error, got: {msg}"
let error = put_bad.expect_err("a mismatched checksum must be rejected");
assert_eq!(
error.raw_response().map(|response| response.status().as_u16()),
Some(400),
"{header}: mismatched checksum must return HTTP 400, got {error:?}"
);
assert_eq!(
error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("BadDigest"),
"{header}: mismatched checksum must return BadDigest, got {error:?}"
);
let error = client
.head_object()
+26 -2
View File
@@ -638,6 +638,18 @@ impl RustFSTestEnvironment {
extra_args: Vec<&str>,
extra_env: &[(&str, &str)],
cleanup_existing: bool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let binary_path = rustfs_binary_path();
self.start_rustfs_server_inner_with_binary(&binary_path, extra_args, extra_env, cleanup_existing)
.await
}
async fn start_rustfs_server_inner_with_binary(
&mut self,
binary_path: &Path,
extra_args: Vec<&str>,
extra_env: &[(&str, &str)],
cleanup_existing: bool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
if cleanup_existing {
self.cleanup_existing_processes().await?;
@@ -647,8 +659,7 @@ impl RustFSTestEnvironment {
info!("Starting RustFS server with args: {:?}", args);
let binary_path = rustfs_binary_path();
let mut command = Command::new(&binary_path);
let mut command = Command::new(binary_path);
command.env("RUST_LOG", "rustfs=info,rustfs_notify=debug");
// The embedded console would bind the fixed default port :9001, which
// collides with unrelated local services (e.g. Docker Desktop). Tests
@@ -668,6 +679,19 @@ impl RustFSTestEnvironment {
Ok(())
}
/// Start a specific RustFS binary against this environment's isolated
/// data directory. Upgrade tests use this to seed an old on-disk format
/// before restarting the same environment with the workspace binary.
pub async fn start_rustfs_server_from_binary(
&mut self,
binary_path: &Path,
extra_args: Vec<&str>,
extra_env: &[(&str, &str)],
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
self.start_rustfs_server_inner_with_binary(binary_path, extra_args, extra_env, true)
.await
}
/// Start RustFS server with basic configuration
pub async fn start_rustfs_server(&mut self, extra_args: Vec<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
self.start_rustfs_server_inner(extra_args, &[], true).await
+29 -1
View File
@@ -24,6 +24,7 @@
use crate::common::{RustFSTestEnvironment, awscurl_get, awscurl_post, init_logging as common_init_logging, local_http_client};
use aws_sdk_s3::Client;
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::ServerSideEncryption;
use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
@@ -50,6 +51,9 @@ pub const VAULT_TOKEN: &str = "dev-root-token";
pub const VAULT_TRANSIT_PATH: &str = "transit";
pub const VAULT_KEY_NAME: &str = "rustfs-master-key";
pub const ENV_TEST_VAULT_BIN: &str = "RUSTFS_TEST_VAULT_BIN";
pub const SSE_C_KEY_MISMATCH_MESSAGE: &str =
"The provided encryption parameters did not match the ones used originally to encrypt the object.";
pub const SSE_C_MISSING_PARAMETERS_MESSAGE: &str = "The object was stored using a form of Server Side Encryption. The correct parameters must be provided to retrieve the object.";
/// Initialize tracing for KMS tests with KMS-specific log levels
pub fn init_logging() {
@@ -63,6 +67,24 @@ pub fn sse_customer_key_md5_base64(key: &str) -> String {
BASE64.encode(hasher.finalize())
}
pub fn assert_s3_error<T, E>(result: Result<T, SdkError<E>>, status: u16, code: &str, message: &str, context: &str)
where
T: std::fmt::Debug,
E: ProvideErrorMetadata + std::fmt::Debug,
{
let error = result.expect_err(context);
assert_eq!(
error.raw_response().map(|response| response.status().as_u16()),
Some(status),
"{context}: unexpected HTTP status: {error:?}"
);
let service_error = error
.as_service_error()
.expect("request failure should retain an S3 service error");
assert_eq!(service_error.code(), Some(code), "{context}: unexpected error code: {error:?}");
assert_eq!(service_error.message(), Some(message), "{context}: unexpected error message: {error:?}");
}
pub async fn kms_admin_request(
base_url: &str,
method: http::Method,
@@ -559,7 +581,13 @@ pub async fn test_error_scenarios(s3_client: &Client, bucket: &str) -> Result<()
.send()
.await;
assert!(wrong_key_result.is_err(), "Download with wrong SSE-C key should fail");
assert_s3_error(
wrong_key_result,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"download with a wrong SSE-C key must be rejected",
);
info!("✅ Correctly rejected download with wrong SSE-C key");
info!("Error scenario tests completed successfully");
@@ -19,9 +19,9 @@
//! complex workflows.
use super::common::{
EncryptionType, LocalKMSTestEnvironment, MultipartTestConfig, create_sse_c_config, sse_customer_key_md5_base64,
test_all_multipart_encryption_types, test_kms_key_management, test_multipart_upload_with_config, test_sse_c_encryption,
test_sse_kms_encryption, test_sse_s3_encryption,
EncryptionType, LocalKMSTestEnvironment, MultipartTestConfig, SSE_C_KEY_MISMATCH_MESSAGE, assert_s3_error,
create_sse_c_config, sse_customer_key_md5_base64, test_all_multipart_encryption_types, test_kms_key_management,
test_multipart_upload_with_config, test_sse_c_encryption, test_sse_kms_encryption, test_sse_s3_encryption,
};
use crate::common::{TEST_BUCKET, init_logging};
use tracing::info;
@@ -191,7 +191,13 @@ async fn test_comprehensive_key_isolation() -> Result<(), Box<dyn std::error::Er
.send()
.await;
assert!(wrong_read_result.is_err(), "The encrypted file should not be readable with the wrong key");
assert_s3_error(
wrong_read_result,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"multipart SSE-C object GET with a wrong key must be rejected",
);
info!("✅ Confirm that key isolation is working correctly");
kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
+66 -15
View File
@@ -21,21 +21,14 @@
//! - Concurrent encryption operations
//! - Security validation tests
use super::common::{LocalKMSTestEnvironment, sse_customer_key_md5_base64};
use super::common::{LocalKMSTestEnvironment, SSE_C_KEY_MISMATCH_MESSAGE, assert_s3_error, sse_customer_key_md5_base64};
use crate::common::{TEST_BUCKET, init_logging};
use aws_sdk_s3::types::ServerSideEncryption;
use base64::Engine;
use md5::{Digest as Md5Digest, Md5};
use std::sync::Arc;
use tokio::sync::Semaphore;
use tracing::{info, warn};
fn md5_hex(input: impl AsRef<[u8]>) -> String {
let mut hasher = Md5::new();
hasher.update(input.as_ref());
hex::encode(hasher.finalize())
}
/// Test encryption of zero-byte files (empty files)
#[tokio::test]
async fn test_kms_zero_byte_file_encryption() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -295,7 +288,7 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
info!("🔍 Testing invalid SSE-C key length");
let invalid_short_key = "short"; // Too short
let invalid_key_b64 = base64::engine::general_purpose::STANDARD.encode(invalid_short_key);
let invalid_key_md5 = md5_hex(invalid_short_key);
let invalid_key_md5 = sse_customer_key_md5_base64(invalid_short_key);
let invalid_key_result = s3_client
.put_object()
@@ -308,14 +301,32 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
.send()
.await;
assert!(invalid_key_result.is_err(), "Should reject invalid key length");
assert_s3_error(
invalid_key_result,
400,
"InvalidRequest",
"SSE-C key must be 32 bytes (256 bits), got 5 bytes.",
"invalid SSE-C key length must be rejected",
);
assert_s3_error(
s3_client
.get_object()
.bucket(TEST_BUCKET)
.key("test-invalid-key-length")
.send()
.await,
404,
"NoSuchKey",
"The specified key does not exist.",
"rejected invalid-key PUT must not create an object",
);
info!("✅ Correctly rejected invalid key length");
// Test 2: Mismatched MD5 for SSE-C
info!("🔍 Testing mismatched MD5 for SSE-C key");
let valid_key = "01234567890123456789012345678901";
let valid_key_b64 = base64::engine::general_purpose::STANDARD.encode(valid_key);
let wrong_md5 = "wrongmd5hash12345678901234567890"; // Wrong MD5
let wrong_md5 = sse_customer_key_md5_base64("98765432109876543210987654321098");
let wrong_md5_result = s3_client
.put_object()
@@ -324,11 +335,24 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
.body(aws_sdk_s3::primitives::ByteStream::from(test_data.to_vec()))
.sse_customer_algorithm("AES256")
.sse_customer_key(&valid_key_b64)
.sse_customer_key_md5(wrong_md5)
.sse_customer_key_md5(&wrong_md5)
.send()
.await;
assert!(wrong_md5_result.is_err(), "Should reject mismatched MD5");
assert_s3_error(
wrong_md5_result,
400,
"InvalidRequest",
"The calculated MD5 hash of the key did not match the hash that was provided.",
"mismatched SSE-C key MD5 must be rejected",
);
assert_s3_error(
s3_client.get_object().bucket(TEST_BUCKET).key("test-wrong-md5").send().await,
404,
"NoSuchKey",
"The specified key does not exist.",
"rejected mismatched-MD5 PUT must not create an object",
);
info!("✅ Correctly rejected mismatched MD5");
// Test 3: Try to access SSE-C object without providing key
@@ -355,7 +379,28 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
.send()
.await;
assert!(no_key_result.is_err(), "Should require SSE-C key for access");
assert_s3_error(
no_key_result,
400,
"InvalidRequest",
"The object was stored using a form of Server Side Encryption. The correct parameters must be provided to retrieve the object.",
"SSE-C object GET without a customer key must be rejected",
);
let recovered = s3_client
.get_object()
.bucket(TEST_BUCKET)
.key("test-sse-c-no-key-access")
.sse_customer_algorithm("AES256")
.sse_customer_key(&valid_key_b64)
.sse_customer_key_md5(&valid_key_md5)
.send()
.await?
.body
.collect()
.await?
.into_bytes();
assert_eq!(recovered.as_ref(), test_data, "failed GET must not corrupt the SSE-C object");
info!("✅ Correctly required SSE-C key for access");
kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
@@ -563,7 +608,13 @@ async fn test_kms_key_validation_security() -> Result<(), Box<dyn std::error::Er
.send()
.await;
assert!(wrong_key_result.is_err(), "Should not be able to decrypt with wrong key");
assert_s3_error(
wrong_key_result,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"SSE-C object GET with the wrong customer key must be rejected",
);
info!("✅ Key isolation verified - wrong key cannot decrypt data");
kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
@@ -23,6 +23,7 @@
use super::common::LocalKMSTestEnvironment;
use crate::common::{TEST_BUCKET, init_logging};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::types::ServerSideEncryption;
use std::fs;
use std::time::Duration;
@@ -77,12 +78,25 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
.send()
.await;
// This should fail, but the server should still be responsive
if put_result2.is_err() {
info!("✅ Upload correctly failed when key directory unavailable");
} else {
warn!("⚠️ Upload succeeded despite unavailable key directory (may be using cached keys)");
}
let unavailable_error = put_result2.expect_err("a missing Local KMS key directory must reject encrypted writes");
assert_eq!(unavailable_error.raw_response().map(|response| response.status().as_u16()), Some(500));
assert_eq!(
unavailable_error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("InternalError")
);
let unavailable_absence = s3_client
.get_object()
.bucket(TEST_BUCKET)
.key(object_key2)
.send()
.await
.expect_err("a write rejected by unavailable KMS must not publish an object");
assert_eq!(unavailable_absence.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(
unavailable_absence.as_service_error().and_then(ProvideErrorMetadata::code),
Some("NoSuchKey")
);
info!("✅ Upload correctly failed when key directory unavailable");
// Restore the key directory
info!("🔧 Restoring key directory");
@@ -107,6 +121,11 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
assert_eq!(put_response3.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
let get_response3 = s3_client.get_object().bucket(TEST_BUCKET).key(object_key3).send().await?;
assert_eq!(get_response3.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
let downloaded_data3 = get_response3.body.collect().await?.into_bytes();
assert_eq!(downloaded_data3.as_ref(), test_data3);
// Verify we can still access the original file
info!("📥 Verifying access to original encrypted file");
let get_response = s3_client.get_object().bucket(TEST_BUCKET).key(object_key).send().await?;
@@ -174,12 +193,22 @@ async fn test_kms_corrupted_key_files() -> Result<(), Box<dyn std::error::Error
.send()
.await;
// This might succeed if KMS uses cached keys, but should eventually fail
if put_result2.is_err() {
info!("✅ Upload correctly failed with corrupted key");
} else {
warn!("⚠️ Upload succeeded despite corrupted key (likely using cached key)");
}
let corrupt_error = put_result2.expect_err("corrupt Local KMS key material must reject encrypted writes");
assert_eq!(corrupt_error.raw_response().map(|response| response.status().as_u16()), Some(500));
assert_eq!(
corrupt_error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("InternalError")
);
let corrupt_absence = s3_client
.get_object()
.bucket(TEST_BUCKET)
.key(object_key2)
.send()
.await
.expect_err("a write rejected by corrupt KMS material must not publish an object");
assert_eq!(corrupt_absence.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(corrupt_absence.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
info!("✅ Upload correctly failed with corrupted key");
// Restore the original key file
info!("🔧 Restoring original key file");
@@ -205,6 +234,11 @@ async fn test_kms_corrupted_key_files() -> Result<(), Box<dyn std::error::Error
assert_eq!(put_response3.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
let get_response3 = s3_client.get_object().bucket(TEST_BUCKET).key(object_key3).send().await?;
assert_eq!(get_response3.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
let downloaded_data3 = get_response3.body.collect().await?.into_bytes();
assert_eq!(downloaded_data3.as_ref(), test_data3);
kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
info!("✅ Corrupted key files test completed successfully");
Ok(())
@@ -280,18 +314,14 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
info!("🔧 Simulating upload interruption");
// Abort the multipart upload
let abort_result = s3_client
s3_client
.abort_multipart_upload()
.bucket(TEST_BUCKET)
.key(object_key)
.upload_id(upload_id)
.send()
.await;
match abort_result {
Ok(_) => info!("✅ Multipart upload aborted successfully"),
Err(e) => warn!("⚠️ Failed to abort multipart upload: {}", e),
}
.await?;
info!("✅ Multipart upload aborted successfully");
// Try to complete the aborted upload - this should fail
info!("🔍 Attempting to complete aborted upload");
@@ -310,18 +340,38 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
.set_parts(Some(completed_parts))
.build();
let complete_result = s3_client
let complete_error = s3_client
.complete_multipart_upload()
.bucket(TEST_BUCKET)
.key(object_key)
.upload_id(upload_id)
.multipart_upload(completed_multipart_upload)
.send()
.await;
assert!(complete_result.is_err(), "Should not be able to complete aborted upload");
.await
.expect_err("an aborted multipart upload must not be completable");
assert_eq!(complete_error.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(
complete_error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("NoSuchUpload")
);
assert_eq!(
complete_error.as_service_error().and_then(ProvideErrorMetadata::message),
Some(
"The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed."
)
);
info!("✅ Correctly failed to complete aborted upload");
let missing_object = s3_client
.get_object()
.bucket(TEST_BUCKET)
.key(object_key)
.send()
.await
.expect_err("aborting a multipart upload must not publish an object");
assert_eq!(missing_object.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(missing_object.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
// Start a new multipart upload and complete it successfully
info!("📤 Starting new multipart upload");
let create_multipart_output2 = s3_client
+9 -2
View File
@@ -20,7 +20,8 @@
//! - Complete encryption/decryption lifecycle
use super::common::{
LocalKMSTestEnvironment, get_kms_status, sse_customer_key_md5_base64, test_kms_key_management, test_sse_c_encryption,
LocalKMSTestEnvironment, SSE_C_KEY_MISMATCH_MESSAGE, assert_s3_error, get_kms_status, sse_customer_key_md5_base64,
test_kms_key_management, test_sse_c_encryption,
};
use crate::common::{TEST_BUCKET, init_logging};
use tracing::{error, info};
@@ -196,7 +197,13 @@ async fn test_local_kms_key_isolation() {
.send()
.await;
assert!(wrong_key_result.is_err(), "Should not be able to decrypt object1 with key2");
assert_s3_error(
wrong_key_result,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"local SSE-C object GET with a wrong key must be rejected",
);
kms_env
.base_env
+10 -4
View File
@@ -22,9 +22,9 @@ use crate::common::{TEST_BUCKET, init_logging};
use tracing::{error, info};
use super::common::{
VAULT_KEY_NAME, VaultTestEnvironment, get_kms_status, sse_customer_key_md5_base64, start_kms,
test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management, test_sse_c_encryption,
test_sse_kms_encryption, test_sse_s3_encryption,
SSE_C_KEY_MISMATCH_MESSAGE, VAULT_KEY_NAME, VaultTestEnvironment, assert_s3_error, get_kms_status,
sse_customer_key_md5_base64, start_kms, test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management,
test_sse_c_encryption, test_sse_kms_encryption, test_sse_s3_encryption,
};
/// Helper that brings up Vault, configures RustFS, and starts the KMS service.
@@ -182,7 +182,13 @@ async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error
.sse_customer_key_md5(&key2_md5)
.send()
.await;
assert!(wrong_key.is_err(), "Object1 should not decrypt with key2");
assert_s3_error(
wrong_key,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"Vault-backed SSE-C object GET with a wrong key must be rejected",
);
context
.base_env()
+4
View File
@@ -61,6 +61,10 @@ mod get_codec_streaming_compat_test;
#[cfg(test)]
mod version_id_regression_test;
// Pinned previous-release -> current-build on-disk compatibility.
#[cfg(test)]
mod upgrade_compatibility_test;
// Receiver-side replication LWW (rustfs/backlog#1953): stale inbound
// replication metadata must not overwrite a newer local category state.
#[cfg(test)]
+21 -5
View File
@@ -34,6 +34,7 @@
//! rejected header-SigV4 requests.
use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::request_signature_v4::{SIGN_V4_ALGORITHM, get_scope, get_signature, get_signing_key};
@@ -280,7 +281,8 @@ async fn tampered_payload_is_rejected() -> Result<(), Box<dyn std::error::Error
let mut env = RustFSTestEnvironment::new().await?;
setup(&mut env).await?;
let path = format!("/{BUCKET}/tampered-payload.txt");
let key = "tampered-payload.txt";
let path = format!("/{BUCKET}/{key}");
let claimed_body = b"the-body-i-claim-to-send";
let actual_body = b"the-body-i-really-send!!";
assert_eq!(claimed_body.len(), actual_body.len(), "keep content-length stable for the mismatch");
@@ -295,17 +297,31 @@ async fn tampered_payload_is_rejected() -> Result<(), Box<dyn std::error::Error
Ok(resp) => {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
assert_ne!(status.as_u16(), 200, "payload mismatch must not succeed, body:\n{body}");
assert!(
status.is_client_error() || status.is_server_error(),
"payload mismatch must be an error status, got {status}, body:\n{body}"
status.is_client_error(),
"payload mismatch must be rejected with a client error, got {status}, body:\n{body}"
);
info!(%status, "tampered payload rejected with error status");
}
// A mid-stream hash-mismatch abort surfacing as a transport error is
// also a valid rejection (definitely not a 200 success).
Err(err) => info!(%err, "tampered payload rejected via transport error"),
Err(err) => {
assert!(!err.is_connect(), "connection failure is not proof of payload rejection: {err}");
assert!(!err.is_timeout(), "request timeout is not proof of payload rejection: {err}");
info!(%err, "tampered payload rejected via mid-stream transport error");
}
}
let absent = env
.create_s3_client()
.get_object()
.bucket(BUCKET)
.key(key)
.send()
.await
.expect_err("a tampered payload must not publish an object");
assert_eq!(absent.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(absent.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
Ok(())
}
+4 -6
View File
@@ -23,6 +23,7 @@
use aws_sdk_s3::Client;
use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::operation::delete_object::DeleteObjectError;
use aws_sdk_s3::operation::put_object_retention::PutObjectRetentionError;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{
DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockMode,
@@ -182,11 +183,8 @@ pub async fn put_object_retention(
mode: ObjectLockRetentionMode,
retain_until: DateTime<Utc>,
bypass_governance: bool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// AWS SDK requires UTC time without timezone offset (e.g., "2026-01-24T11:20:14Z")
let retain_until_str = retain_until.format("%Y-%m-%dT%H:%M:%SZ").to_string();
let retain_until_datetime =
aws_sdk_s3::primitives::DateTime::from_str(&retain_until_str, aws_sdk_s3::primitives::DateTimeFormat::DateTime)?;
) -> Result<(), Box<SdkError<PutObjectRetentionError>>> {
let retain_until_datetime = aws_sdk_s3::primitives::DateTime::from_secs(retain_until.timestamp());
let retention = ObjectLockRetention::builder()
.mode(mode.clone())
@@ -204,7 +202,7 @@ pub async fn put_object_retention(
request = request.version_id(vid);
}
request.send().await?;
request.send().await.map_err(Box::new)?;
info!("Put object retention on {} with mode {:?}", key, mode);
Ok(())
}
@@ -1475,7 +1475,7 @@ async fn test_put_retention_compliance_cannot_shorten() {
)
.await;
assert!(shorten_result.is_err(), "Shortening COMPLIANCE retention should fail");
assert_access_denied(shorten_result, "Shortening COMPLIANCE retention should fail");
info!("✅ Test passed: Cannot shorten COMPLIANCE retention");
}
@@ -1598,10 +1598,7 @@ async fn test_put_retention_governance_shorten_requires_bypass() {
)
.await;
assert!(
shorten_without_bypass.is_err(),
"Shortening GOVERNANCE retention without bypass should fail"
);
assert_access_denied(shorten_without_bypass, "Shortening GOVERNANCE retention without bypass should fail");
// Shorten with bypass - should succeed
let shorten_with_bypass = put_object_retention(
+105 -54
View File
@@ -14,6 +14,7 @@
use crate::common::{RustFSTestEnvironment, admin_request, awscurl_delete, awscurl_get, awscurl_post, awscurl_put, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use http::{Method, StatusCode};
use tokio::time::{Duration, sleep, timeout};
use tracing::{debug, info};
@@ -132,19 +133,13 @@ impl QuotaTestEnv {
pub async fn object_exists(&self, key: &str) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
match self.client.head_object().bucket(&self.bucket_name).key(key).send().await {
Ok(_) => Ok(true),
Err(e) => {
// Check for any 404-related errors and return false instead of propagating
let error_str = e.to_string();
if error_str.contains("404") || error_str.contains("Not Found") || error_str.contains("NotFound") {
Err(error) => {
let status = error.raw_response().map(|response| response.status().as_u16());
let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
if status == Some(404) && matches!(code, Some("NotFound" | "NoSuchKey")) {
Ok(false)
} else {
// Also check the error code directly
if let Some(service_err) = e.as_service_error()
&& service_err.is_not_found()
{
return Ok(false);
}
Err(e.into())
Err(error.into())
}
}
}
@@ -278,7 +273,46 @@ impl QuotaTestEnv {
#[cfg(test)]
mod integration_tests {
use super::*;
use aws_sdk_s3::error::ProvideErrorMetadata;
fn assert_error_response(status: StatusCode, body: &str, expected_status: StatusCode, expected_code: &str) {
assert_eq!(status, expected_status, "unexpected error status: {status} {body}");
assert!(
body.contains(&format!("<Code>{expected_code}</Code>")),
"expected {expected_code}, got: {body}"
);
}
fn assert_quota_rejection<E>(status: Option<u16>, service_error: Option<&E>, error: &impl std::fmt::Debug)
where
E: ProvideErrorMetadata + std::fmt::Debug,
{
assert_eq!(status, Some(400), "quota rejection must return HTTP 400: {error:?}");
let service_error = service_error.expect("quota rejection must be an S3 service error");
assert_eq!(service_error.code(), Some("InvalidRequest"), "unexpected quota error: {error:?}");
assert!(
service_error
.message()
.is_some_and(|message| message.starts_with("Bucket quota exceeded")),
"operation must fail specifically at quota admission: {error:?}"
);
}
async fn assert_put_rejected_by_quota(env: &QuotaTestEnv, key: &str, size_bytes: usize) {
let error = env
.client
.put_object()
.bucket(&env.bucket_name)
.key(key)
.body(aws_sdk_s3::primitives::ByteStream::from(vec![0u8; size_bytes]))
.send()
.await
.expect_err("PUT above quota must be rejected");
assert_quota_rejection(
error.raw_response().map(|response| response.status().as_u16()),
error.as_service_error(),
&error,
);
}
#[tokio::test]
async fn test_quota_basic_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -304,8 +338,7 @@ mod integration_tests {
assert!(env.object_exists("test2.txt").await?);
// Try to upload 1KB more (should fail due to quota)
let upload_result = env.upload_object("test3.txt", 1024).await;
assert!(upload_result.is_err());
assert_put_rejected_by_quota(&env, "test3.txt", 1024).await;
assert!(!env.object_exists("test3.txt").await?);
// Clean up
@@ -356,10 +389,10 @@ mod integration_tests {
let err = put_aws_chunked("over-quota.bin", 16 * 1024)
.await
.expect_err("declared aws-chunked PUT over quota must be rejected");
let err_debug = format!("{err:?}");
assert!(
!err_debug.contains("UnexpectedContent"),
"over-quota rejection must be the quota error, not UnexpectedContent: {err_debug}"
assert_quota_rejection(
err.raw_response().map(|response| response.status().as_u16()),
err.as_service_error(),
&err,
);
assert!(!env.object_exists("over-quota.bin").await?);
@@ -581,24 +614,35 @@ mod integration_tests {
env.create_bucket().await?;
// Test invalid quota type
let url = format!("{}/rustfs/admin/v3/quota/{}", env.env.url, env.bucket_name);
let quota_path = format!("/rustfs/admin/v3/quota/{}", env.bucket_name);
let invalid_config = serde_json::json!({
"quota": 1024,
"quota_type": "SOFT" // Invalid type
});
let response = awscurl_put(&url, &invalid_config.to_string(), &env.env.access_key, &env.env.secret_key).await;
assert!(response.is_err());
let error_msg = response.unwrap_err().to_string();
assert!(error_msg.contains("InvalidArgument"));
let (status, body) = admin_request(
&env.env.url,
Method::PUT,
&quota_path,
Some(invalid_config.to_string()),
&env.env.access_key,
&env.env.secret_key,
)
.await?;
assert_error_response(status, &body, StatusCode::BAD_REQUEST, "InvalidArgument");
// Test operations on non-existent bucket
let url = format!("{}/rustfs/admin/v3/quota/non-existent-bucket", env.env.url);
let response = awscurl_get(&url, &env.env.access_key, &env.env.secret_key).await;
assert!(response.is_err());
let error_msg = response.unwrap_err().to_string();
assert!(error_msg.contains("NoSuchBucket"));
let (status, body) = admin_request(
&env.env.url,
Method::GET,
"/rustfs/admin/v3/quota/non-existent-bucket",
None,
&env.env.access_key,
&env.env.secret_key,
)
.await?;
assert_error_response(status, &body, StatusCode::NOT_FOUND, "NoSuchBucket");
env.cleanup_bucket().await?;
@@ -652,10 +696,16 @@ mod integration_tests {
"quota": 1024,
"quota_type": "SOFT"
});
let response = awscurl_put(&url, &invalid_config.to_string(), &env.env.access_key, &env.env.secret_key).await;
assert!(response.is_err());
let error_msg = response.unwrap_err().to_string();
assert!(error_msg.contains("InvalidArgument"));
let (status, body) = admin_request(
&env.env.url,
Method::PUT,
&format!("/rustfs/admin/v3/quota/{}", env.bucket_name),
Some(invalid_config.to_string()),
&env.env.access_key,
&env.env.secret_key,
)
.await?;
assert_error_response(status, &body, StatusCode::BAD_REQUEST, "InvalidArgument");
env.cleanup_bucket().await?;
@@ -698,26 +748,21 @@ mod integration_tests {
assert!(resp.contains("quota_limit"));
// Normal user sets quota — should be denied
let set_error = awscurl_put(
&get_url,
&serde_json::json!({"quota": 2048, "quota_type": "HARD"}).to_string(),
let quota_path = format!("/rustfs/admin/v3/quota/{}", env.bucket_name);
let (status, body) = admin_request(
&env.env.url,
Method::PUT,
&quota_path,
Some(serde_json::json!({"quota": 2048, "quota_type": "HARD"}).to_string()),
normal_ak,
normal_sk,
)
.await
.expect_err("normal user should not be able to set quota")
.to_string();
assert!(set_error.contains("AccessDenied"), "quota denial must return AccessDenied: {set_error}");
.await?;
assert_error_response(status, &body, StatusCode::FORBIDDEN, "AccessDenied");
// Normal user clears quota — should be denied
let delete_error = awscurl_delete(&get_url, normal_ak, normal_sk)
.await
.expect_err("normal user should not be able to clear quota")
.to_string();
assert!(
delete_error.contains("AccessDenied"),
"quota deletion denial must return AccessDenied: {delete_error}"
);
let (status, body) = admin_request(&env.env.url, Method::DELETE, &quota_path, None, normal_ak, normal_sk).await?;
assert_error_response(status, &body, StatusCode::FORBIDDEN, "AccessDenied");
env.cleanup_bucket().await?;
Ok(())
@@ -757,7 +802,12 @@ mod integration_tests {
.send()
.await;
assert!(copy_result.is_err());
let copy_error = copy_result.expect_err("copy above quota must be rejected");
assert_quota_rejection(
copy_error.raw_response().map(|response| response.status().as_u16()),
copy_error.as_service_error(),
&copy_error,
);
assert!(!env.object_exists("copy2.txt").await?);
env.cleanup_bucket().await?;
@@ -780,8 +830,7 @@ mod integration_tests {
env.upload_object("file2.txt", 1024 * 1024).await?;
// Verify quota is full
let upload_result = env.upload_object("file3.txt", 1024).await;
assert!(upload_result.is_err());
assert_put_rejected_by_quota(&env, "file3.txt", 1024).await;
// Delete multiple objects using batch delete
let objects = vec![
@@ -881,9 +930,7 @@ mod integration_tests {
// Test 2: Multipart upload exceeds quota (should fail)
// Upload 6MB filler (total now: 5MB + 6MB = 11MB > 10MB quota)
let upload_filler = env.upload_object("filler.txt", 6 * 1024 * 1024).await;
// This should fail due to quota
assert!(upload_filler.is_err());
assert_put_rejected_by_quota(&env, "filler.txt", 6 * 1024 * 1024).await;
// Verify filler doesn't exist
assert!(!env.object_exists("filler.txt").await?);
@@ -939,7 +986,11 @@ mod integration_tests {
.await;
let complete_error = complete_result.expect_err("multipart completion above quota must be rejected");
assert_eq!(complete_error.as_service_error().and_then(|error| error.code()), Some("InvalidRequest"));
assert_quota_rejection(
complete_error.raw_response().map(|response| response.status().as_u16()),
complete_error.as_service_error(),
&complete_error,
);
assert!(!env.object_exists("over_quota.txt").await?);
let staged_parts = env
@@ -20,7 +20,10 @@ use crate::fake_s3_target::{
FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation,
RequestRecord,
};
use crate::kms::common::{create_key_with_specific_id, sse_customer_key_md5_base64};
use crate::kms::common::{
SSE_C_KEY_MISMATCH_MESSAGE, SSE_C_MISSING_PARAMETERS_MESSAGE, assert_s3_error, create_key_with_specific_id,
sse_customer_key_md5_base64,
};
use crate::storage_api::replication_extension::BucketTargetSys;
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::error::ProvideErrorMetadata;
@@ -4375,7 +4378,13 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult {
// Without the customer key the replica must not be readable — the direct
// detection point for a silent-plaintext replica (backlog#1291).
let plain_read = target_client.get_object().bucket(&target_bucket).key(key).send().await;
assert!(plain_read.is_err(), "SSE-C replica must not be readable without the customer key");
assert_s3_error(
plain_read,
400,
"InvalidRequest",
SSE_C_MISSING_PARAMETERS_MESSAGE,
"SSE-C replica must not be readable without the customer key",
);
// A wrong customer key must fail too.
let wrong_key = BASE64_STANDARD.encode("99999999999999999999999999999999");
@@ -4389,7 +4398,13 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult {
.sse_customer_key_md5(&wrong_key_md5)
.send()
.await;
assert!(wrong_read.is_err(), "SSE-C replica must reject a wrong customer key");
assert_s3_error(
wrong_read,
400,
"InvalidRequest",
SSE_C_KEY_MISMATCH_MESSAGE,
"SSE-C replica must reject a wrong customer key",
);
Ok(())
}
@@ -4486,9 +4501,12 @@ async fn test_bucket_replication_sse_c_multipart_passthrough() -> TestResult {
assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), payload.as_slice());
let plain_read = target_client.get_object().bucket(&target_bucket).key(key).send().await;
assert!(
plain_read.is_err(),
"SSE-C multipart replica must not be readable without the customer key"
assert_s3_error(
plain_read,
400,
"InvalidRequest",
SSE_C_MISSING_PARAMETERS_MESSAGE,
"SSE-C multipart replica must not be readable without the customer key",
);
// Stability across scanner cycles: convergence must hold for passthrough.
@@ -4892,15 +4910,12 @@ async fn test_bucket_replication_sse_c_existing_object_resync() -> TestResult {
assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body.as_slice());
// No plaintext leak: the replica stays unreadable without the key.
assert!(
target_client
.get_object()
.bucket(target_bucket)
.key(key)
.send()
.await
.is_err(),
"SSE-C replica must not be readable without the customer key"
assert_s3_error(
target_client.get_object().bucket(target_bucket).key(key).send().await,
400,
"InvalidRequest",
SSE_C_MISSING_PARAMETERS_MESSAGE,
"SSE-C resynced replica must not be readable without the customer key",
);
Ok(())
+22 -16
View File
@@ -28,6 +28,7 @@
mod tests {
use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use http::StatusCode;
use http::header::HOST;
@@ -731,12 +732,7 @@ mod tests {
create_bucket(&client, bucket).await.expect("Failed to create bucket");
// Test that control characters are rejected
let invalid_keys = vec![
"file\0with\0null.txt",
"file\nwith\nnewline.txt",
"file\rwith\rcarriage.txt",
"file\twith\ttab.txt", // Tab might be allowed, but let's test
];
let invalid_keys = ["file\0with\0null.txt", "file\nwith\nnewline.txt", "file\rwith\rcarriage.txt"];
for key in invalid_keys {
info!("Testing rejection of control character in key: {:?}", key);
@@ -747,18 +743,28 @@ mod tests {
.key(key)
.body(ByteStream::from_static(b"test"))
.send()
.await;
// Note: The validation happens on the server side, so we expect an error
// For null byte, newline, and carriage return
if key.contains('\0') || key.contains('\n') || key.contains('\r') {
assert!(result.is_err(), "Control character should be rejected for key: {key:?}");
if let Err(e) = result {
info!("✅ Control character correctly rejected: {:?}", e);
}
}
.await
.expect_err("invalid control characters must be rejected by the server");
assert_eq!(
result.raw_response().map(|response| response.status().as_u16()),
Some(400),
"control character must return HTTP 400 for key {key:?}: {result:?}"
);
assert_eq!(
result.as_service_error().and_then(ProvideErrorMetadata::code),
Some("InvalidArgument"),
"control character must return InvalidArgument for key {key:?}: {result:?}"
);
}
let listed = client
.list_objects_v2()
.bucket(bucket)
.send()
.await
.expect("server must remain healthy after rejected requests");
assert!(listed.contents().is_empty(), "rejected requests must not create objects");
// Cleanup
env.stop_server();
info!("Test completed successfully");
@@ -0,0 +1,254 @@
// Copyright 2024 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 crate::common::{RustFSTestEnvironment, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{
BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, ServerSideEncryption, VersioningConfiguration,
};
use std::path::PathBuf;
type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
const SOURCE_BINARY_ENV: &str = "RUSTFS_UPGRADE_SOURCE_BINARY";
const SSE_MASTER_KEY_ENV: &str = "RUSTFS_SSE_S3_MASTER_KEY";
const SSE_MASTER_KEY: &str = "QkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkI=";
const PLAIN_BUCKET: &str = "upgrade-plain-data";
const VERSIONED_BUCKET: &str = "upgrade-versioned-data";
fn source_binary() -> Result<PathBuf, Box<dyn std::error::Error + Send + Sync>> {
let path = std::env::var_os(SOURCE_BINARY_ENV)
.map(PathBuf::from)
.ok_or("RUSTFS_UPGRADE_SOURCE_BINARY must point to the pinned previous release binary")?;
if !path.is_file() {
return Err(format!("upgrade source binary does not exist: {}", path.display()).into());
}
Ok(path)
}
async fn enable_versioning(client: &Client, bucket: &str) -> TestResult {
let configuration = VersioningConfiguration::builder()
.status(BucketVersioningStatus::Enabled)
.build();
client
.put_bucket_versioning()
.bucket(bucket)
.versioning_configuration(configuration)
.send()
.await?;
Ok(())
}
async fn read_object(
client: &Client,
bucket: &str,
key: &str,
version_id: Option<&str>,
) -> Result<(Option<ServerSideEncryption>, Vec<u8>), Box<dyn std::error::Error + Send + Sync>> {
let mut request = client.get_object().bucket(bucket).key(key);
if let Some(version_id) = version_id {
request = request.version_id(version_id);
}
let response = request.send().await?;
let encryption = response.server_side_encryption().cloned();
let body = response.body.collect().await?.into_bytes().to_vec();
Ok((encryption, body))
}
async fn write_multipart(client: &Client, bucket: &str, key: &str, parts: &[Vec<u8>]) -> TestResult {
let created = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
let upload_id = created.upload_id().ok_or("CreateMultipartUpload omitted upload ID")?;
let mut completed_parts = Vec::with_capacity(parts.len());
for (index, part) in parts.iter().enumerate() {
let part_number = i32::try_from(index + 1)?;
let uploaded = client
.upload_part()
.bucket(bucket)
.key(key)
.upload_id(upload_id)
.part_number(part_number)
.body(ByteStream::from(part.clone()))
.send()
.await?;
completed_parts.push(
CompletedPart::builder()
.part_number(part_number)
.e_tag(uploaded.e_tag().ok_or("UploadPart omitted ETag")?)
.build(),
);
}
client
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(upload_id)
.multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
.send()
.await?;
Ok(())
}
#[tokio::test]
#[ignore = "requires a pinned previous RustFS release binary"]
async fn direct_upgrade_from_rc2_preserves_object_contracts() -> TestResult {
init_logging();
let previous_binary = source_binary()?;
let mut env = RustFSTestEnvironment::new().await?;
let server_env = [(SSE_MASTER_KEY_ENV, SSE_MASTER_KEY)];
env.start_rustfs_server_from_binary(&previous_binary, vec![], &server_env)
.await?;
let old_client = env.create_s3_client();
env.create_test_bucket(PLAIN_BUCKET).await?;
env.create_test_bucket(VERSIONED_BUCKET).await?;
enable_versioning(&old_client, VERSIONED_BUCKET).await?;
let plain_key = "plain-object";
let plain_bytes = b"written by the previous RustFS release";
old_client
.put_object()
.bucket(PLAIN_BUCKET)
.key(plain_key)
.body(ByteStream::from_static(plain_bytes))
.send()
.await?;
let encrypted_key = "sse-s3-object";
let encrypted_bytes = b"encrypted by the previous RustFS release";
old_client
.put_object()
.bucket(PLAIN_BUCKET)
.key(encrypted_key)
.server_side_encryption(ServerSideEncryption::Aes256)
.body(ByteStream::from_static(encrypted_bytes))
.send()
.await?;
let multipart_key = "multipart-object";
let multipart_parts = vec![vec![b'a'; 5 * 1024 * 1024], b"final multipart bytes".to_vec()];
let multipart_bytes = multipart_parts.concat();
write_multipart(&old_client, PLAIN_BUCKET, multipart_key, &multipart_parts).await?;
let versioned_key = "versioned-object";
let version1_bytes = b"version one from the previous release";
let version1 = old_client
.put_object()
.bucket(VERSIONED_BUCKET)
.key(versioned_key)
.body(ByteStream::from_static(version1_bytes))
.send()
.await?
.version_id()
.ok_or("first versioned PUT omitted version ID")?
.to_string();
let version2_bytes = b"version two from the previous release";
let version2 = old_client
.put_object()
.bucket(VERSIONED_BUCKET)
.key(versioned_key)
.body(ByteStream::from_static(version2_bytes))
.send()
.await?
.version_id()
.ok_or("second versioned PUT omitted version ID")?
.to_string();
let deleted = old_client
.delete_object()
.bucket(VERSIONED_BUCKET)
.key(versioned_key)
.send()
.await?;
assert_eq!(deleted.delete_marker(), Some(true));
let delete_marker = deleted
.version_id()
.ok_or("versioned DELETE omitted delete marker version ID")?
.to_string();
env.restart_server_preserving_data(vec![], &server_env).await?;
let current_client = env.create_s3_client();
assert_eq!(read_object(&current_client, PLAIN_BUCKET, plain_key, None).await?.1, plain_bytes);
let (encryption, upgraded_encrypted_bytes) = read_object(&current_client, PLAIN_BUCKET, encrypted_key, None).await?;
assert_eq!(encryption, Some(ServerSideEncryption::Aes256));
assert_eq!(upgraded_encrypted_bytes, encrypted_bytes);
assert_eq!(read_object(&current_client, PLAIN_BUCKET, multipart_key, None).await?.1, multipart_bytes);
assert_eq!(
read_object(&current_client, VERSIONED_BUCKET, versioned_key, Some(&version1))
.await?
.1,
version1_bytes
);
assert_eq!(
read_object(&current_client, VERSIONED_BUCKET, versioned_key, Some(&version2))
.await?
.1,
version2_bytes
);
let current_read = current_client
.get_object()
.bucket(VERSIONED_BUCKET)
.key(versioned_key)
.send()
.await
.expect_err("the previous release's delete marker must remain current after upgrade");
assert_eq!(current_read.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(current_read.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
let listed = current_client
.list_object_versions()
.bucket(VERSIONED_BUCKET)
.prefix(versioned_key)
.send()
.await?;
assert_eq!(listed.versions().len(), 2);
assert!(
listed
.versions()
.iter()
.any(|version| version.version_id() == Some(version1.as_str()))
);
assert!(
listed
.versions()
.iter()
.any(|version| version.version_id() == Some(version2.as_str()))
);
assert_eq!(listed.delete_markers().len(), 1);
assert_eq!(listed.delete_markers()[0].version_id(), Some(delete_marker.as_str()));
assert_eq!(listed.delete_markers()[0].is_latest(), Some(true));
let post_upgrade_key = "written-after-upgrade";
let post_upgrade_bytes = b"written by the current RustFS build";
current_client
.put_object()
.bucket(PLAIN_BUCKET)
.key(post_upgrade_key)
.body(ByteStream::from_static(post_upgrade_bytes))
.send()
.await?;
assert_eq!(
read_object(&current_client, PLAIN_BUCKET, post_upgrade_key, None).await?.1,
post_upgrade_bytes
);
Ok(())
}
@@ -25,6 +25,7 @@
mod tests {
use crate::common::{RustFSTestEnvironment, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
use tracing::info;
@@ -285,7 +286,10 @@ mod tests {
let output = result.unwrap();
info!("📥 PutObject response - version_id: {:?}", output.version_id);
// version_id can be None or Some("null") for non-versioned buckets
assert!(
output.version_id().is_none() || output.version_id() == Some("null"),
"non-versioned PUT must omit version ID or return the S3 null version"
);
info!("✅ PASSED: PutObject works correctly without versioning");
}
@@ -317,7 +321,11 @@ mod tests {
.send()
.await;
assert!(put_result.is_ok(), "PUT operation failed");
let _version_id = put_result.unwrap().version_id;
let version_id = put_result
.unwrap()
.version_id()
.expect("versioned PUT should return a version ID")
.to_string();
// Test GET
info!("📥 Testing GET operation");
@@ -341,16 +349,46 @@ mod tests {
// Test DELETE
info!("🗑️ Testing DELETE operation");
let delete_result = client.delete_object().bucket(bucket).key(key).send().await;
assert!(delete_result.is_ok(), "DELETE operation failed");
let delete_result = client
.delete_object()
.bucket(bucket)
.key(key)
.send()
.await
.expect("DELETE operation failed");
assert_eq!(delete_result.delete_marker(), Some(true));
let delete_marker_version_id = delete_result
.version_id()
.expect("versioned DELETE should return a delete marker version ID")
.to_string();
// Verify object is deleted (should return NoSuchKey or version marker)
let get_after_delete = client.get_object().bucket(bucket).key(key).send().await;
assert!(
get_after_delete.is_err() || get_after_delete.unwrap().delete_marker == Some(true),
"Object should be deleted or have delete marker"
let get_after_delete = client
.get_object()
.bucket(bucket)
.key(key)
.send()
.await
.expect_err("the current delete marker must hide the object");
assert_eq!(get_after_delete.raw_response().map(|response| response.status().as_u16()), Some(404));
assert_eq!(
get_after_delete.as_service_error().and_then(ProvideErrorMetadata::code),
Some("NoSuchKey")
);
let versions = client
.list_object_versions()
.bucket(bucket)
.prefix(key)
.send()
.await
.expect("ListObjectVersions failed after DELETE");
assert_eq!(versions.versions().len(), 1);
assert_eq!(versions.versions()[0].version_id(), Some(version_id.as_str()));
assert_eq!(versions.versions()[0].is_latest(), Some(false));
assert_eq!(versions.delete_markers().len(), 1);
assert_eq!(versions.delete_markers()[0].version_id(), Some(delete_marker_version_id.as_str()));
assert_eq!(versions.delete_markers()[0].is_latest(), Some(true));
info!("✅ PASSED: All basic S3 operations work correctly");
}
@@ -417,31 +455,59 @@ mod tests {
let client = env.create_s3_client();
env.create_test_bucket(bucket).await?;
enable_versioning(&client, bucket).await?;
let key = "terraform.tfstate";
let response = client
let first_version = client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from(b"v1".to_vec()))
.send()
.await;
assert!(response.is_ok());
.await?
.version_id()
.ok_or("first Terraform state PUT omitted version ID")?
.to_string();
client.delete_object().bucket(bucket).key(key).send().await?;
let deleted = client.delete_object().bucket(bucket).key(key).send().await?;
assert_eq!(deleted.delete_marker(), Some(true));
let delete_marker = deleted
.version_id()
.ok_or("Terraform state DELETE omitted delete marker version ID")?
.to_string();
let response = client
let second_version = client
.put_object()
.bucket(bucket)
.key(key)
.body(ByteStream::from(b"v1".to_vec()))
.body(ByteStream::from(b"v2".to_vec()))
.send()
.await;
.await?
.version_id()
.ok_or("second Terraform state PUT omitted version ID")?
.to_string();
assert!(response.is_ok());
let get_response = client.get_object().bucket(bucket).key(key).send().await?;
let current_body = get_response.body.collect().await?.into_bytes();
assert_eq!(current_body.as_ref(), b"v2");
let get_response = client.get_object().bucket(bucket).key(key).send().await;
assert!(get_response.is_ok(), "Object should exist after PUT");
let listed = client.list_object_versions().bucket(bucket).prefix(key).send().await?;
assert_eq!(listed.versions().len(), 2);
assert!(
listed
.versions()
.iter()
.any(|version| version.version_id() == Some(first_version.as_str()) && version.is_latest() == Some(false))
);
assert!(
listed
.versions()
.iter()
.any(|version| version.version_id() == Some(second_version.as_str()) && version.is_latest() == Some(true))
);
assert_eq!(listed.delete_markers().len(), 1);
assert_eq!(listed.delete_markers()[0].version_id(), Some(delete_marker.as_str()));
assert_eq!(listed.delete_markers()[0].is_latest(), Some(false));
Ok(())
}
@@ -234,6 +234,7 @@ mod serial_tests {
create_versioned_bucket(&ecstore, bucket).await;
let v1 = put_versioned(&ecstore, bucket, object, &versioned_test_data(1)).await;
wait_for_object_copies(&disk_paths, bucket, object).await;
// Wipe the object entirely on disks 1..4, leaving it on ONLY disk[0]
// (1/4 disks < read-quorum 2). effective listing_quorum for 4 drives is
@@ -387,6 +388,7 @@ mod serial_tests {
let data_v1 = versioned_test_data(9);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await;
wait_for_object_copies(&disk_paths, bucket, object).await;
// EC4+4: wipe the object ENTIRELY on 4 disks, leaving full copies on the
// other 4 (== data_blocks). Meta quorum (4) still holds, so this heals via
@@ -500,6 +502,7 @@ mod serial_tests {
let data_v1 = versioned_test_data(11);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await;
wait_for_object_copies(&disk_paths, bucket, object).await;
// Drop the object entirely on ONE disk (its shard + meta gone), leaving it
// on 3/4 (>= data_blocks 2). The union must still enumerate it.
@@ -545,6 +548,7 @@ mod serial_tests {
let data_v1 = versioned_test_data(12);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await;
wait_for_object_copies(&disk_paths, bucket, object).await;
std::fs::remove_dir_all(object_dir(&disk_paths[3], bucket, object)).expect("wipe object on disk3");
for disk in &disk_paths[..3] {
+31 -18
View File
@@ -56,6 +56,32 @@ async fn wait_for_path_exists(path: &Path, timeout: Duration, interval: Duration
}
}
fn find_part_file(obj_dir: &Path) -> Option<PathBuf> {
WalkDir::new(obj_dir)
.min_depth(2)
.max_depth(2)
.into_iter()
.filter_map(Result::ok)
.find(|entry| entry.file_type().is_file() && entry.file_name().to_str().is_some_and(|name| name.starts_with("part.")))
.map(|entry| entry.into_path())
}
async fn wait_for_object_copies(disks: &[PathBuf], bucket: &str, object: &str) {
tokio::time::timeout(HEAL_FORMAT_WAIT_TIMEOUT, async {
loop {
if disks.iter().all(|disk| {
let obj_dir = disk.join(bucket).join(object);
obj_dir.join("xl.meta").exists() && find_part_file(&obj_dir).is_some()
}) {
break;
}
tokio::time::sleep(HEAL_FORMAT_WAIT_INTERVAL).await;
}
})
.await
.expect("PUT rename tails must converge before corrupting the disk fixture");
}
/// Test helper: build the shared 4-disk temp-dir ECStore environment
/// (rustfs-test-utils, backlog#1153 infra-1) and wrap it in the heal storage
/// layer. Port 0 + uuid temp dirs keep this parallel-safe under nextest.
@@ -103,18 +129,10 @@ mod serial_tests {
create_test_bucket(&ecstore, bucket_name).await;
upload_test_object(&ecstore, bucket_name, object_name, &test_data).await;
let _obj_dir = disk_paths[0].join(bucket_name).join(object_name);
wait_for_object_copies(&disk_paths, bucket_name, object_name).await;
// ─── 1️⃣ delete single data shard file ─────────────────────────────────────
let obj_dir = disk_paths[0].join(bucket_name).join(object_name);
// find part file at depth 2, e.g. .../<uuid>/part.1
let target_part = WalkDir::new(&obj_dir)
.min_depth(2)
.max_depth(2)
.into_iter()
.filter_map(Result::ok)
.find(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false))
.map(|e| e.into_path())
.expect("Failed to locate part file to delete");
let target_part = find_part_file(&obj_dir).expect("converged fixture must contain a part file");
std::fs::remove_file(&target_part).expect("failed to delete part file");
assert!(!target_part.exists());
@@ -171,6 +189,7 @@ mod serial_tests {
create_test_bucket(&ecstore, bucket_name).await;
upload_test_object(&ecstore, bucket_name, object_name, &test_data).await;
wait_for_object_copies(&disk_paths, bucket_name, object_name).await;
// Plant a leaked, unreferenced UUID data dir under the object on every disk
// that actually holds the object (i.e. has an `xl.meta`). Planting on a disk
@@ -381,15 +400,9 @@ mod serial_tests {
create_test_bucket(&ecstore, bucket_name).await;
upload_test_object(&ecstore, bucket_name, object_name, &test_data).await;
wait_for_object_copies(&disk_paths, bucket_name, object_name).await;
let obj_dir = disk_paths[0].join(bucket_name).join(object_name);
let target_part = WalkDir::new(&obj_dir)
.min_depth(2)
.max_depth(2)
.into_iter()
.filter_map(Result::ok)
.find(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false))
.map(|e| e.into_path())
.expect("Failed to locate part file to delete");
let target_part = find_part_file(&obj_dir).expect("converged fixture must contain a part file");
// ─── 1️⃣ delete format.json on one disk ──────────────
let format_path = disk_paths[0].join(".rustfs.sys").join("format.json");
+100 -4
View File
@@ -2273,6 +2273,7 @@ pub(crate) fn test_config(id: &str) -> OidcProviderConfig {
#[cfg(test)]
mod tests {
use super::*;
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use rustfs_utils::egress::OutboundDnsPolicyRejection;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -2641,12 +2642,15 @@ mod tests {
)
}
fn start_mock_oidc_discovery_server<F>(
fn start_mock_oidc_discovery_server_with_jwks<F, J>(
build_discovery_issuer: F,
max_requests: usize,
signing_alg: &'static str,
jwks_response: J,
) -> Option<(String, std::thread::JoinHandle<()>)>
where
F: Fn(&str) -> (String, String, String) + Send + 'static,
J: Fn(usize) -> String + Send + 'static,
{
use std::io::Write;
use std::net::{Shutdown, TcpListener};
@@ -2674,10 +2678,9 @@ mod tests {
"response_types_supported": ["code"],
"response_modes_supported": ["query"],
"subject_types_supported": ["public"],
"id_token_signing_alg_values_supported": ["RS256"],
"id_token_signing_alg_values_supported": [signing_alg],
})
.to_string();
let jwks_body = r#"{"keys":[]}"#;
let (ready_tx, ready_rx) = mpsc::channel();
let handle = std::thread::spawn(move || {
@@ -2687,6 +2690,7 @@ mod tests {
let _ = ready_tx.send(());
let mut seen = 0usize;
let mut jwks_fetches = 0usize;
let start = Instant::now();
let mut last_completed = Instant::now();
@@ -2719,7 +2723,11 @@ mod tests {
.expect("failed to set discovery mock read timeout");
let path = read_mock_oidc_request_path(&mut stream);
let response = mock_oidc_response(&path, &discovery_body, &expected_jwks_path, jwks_body);
let jwks_body = jwks_response(jwks_fetches);
if path == expected_jwks_path {
jwks_fetches += 1;
}
let response = mock_oidc_response(&path, &discovery_body, &expected_jwks_path, &jwks_body);
let _ = stream.write_all(response.as_bytes());
let _ = stream.flush();
let _ = stream.shutdown(Shutdown::Both);
@@ -2737,6 +2745,94 @@ mod tests {
Some((base, handle))
}
fn start_mock_oidc_discovery_server<F>(
build_discovery_issuer: F,
max_requests: usize,
) -> Option<(String, std::thread::JoinHandle<()>)>
where
F: Fn(&str) -> (String, String, String) + Send + 'static,
{
start_mock_oidc_discovery_server_with_jwks(build_discovery_issuer, max_requests, "RS256", |_| {
r#"{"keys":[]}"#.to_string()
})
}
fn oidc_es256_key_and_jwk(kid: &str) -> (EncodingKey, serde_json::Value) {
let certified =
rcgen::generate_simple_self_signed(vec![format!("{kid}.invalid")]).expect("OIDC signing key should generate");
let encoding_key =
EncodingKey::from_ec_pem(certified.signing_key.serialize_pem().as_bytes()).expect("OIDC signing key should encode");
let mut jwk = jsonwebtoken::jwk::Jwk::from_encoding_key(&encoding_key, Algorithm::ES256)
.expect("OIDC public JWK should derive from signing key");
jwk.common.key_id = Some(kid.to_string());
jwk.common.public_key_use = Some(jsonwebtoken::jwk::PublicKeyUse::Signature);
(encoding_key, serde_json::to_value(jwk).expect("OIDC JWK should serialize"))
}
#[tokio::test]
async fn web_identity_verification_refreshes_rotated_jwks() {
let (_, initial_jwk) = oidc_es256_key_and_jwk("initial");
let (rotated_key, rotated_jwk) = oidc_es256_key_and_jwk("rotated");
let initial_jwks = serde_json::json!({ "keys": [initial_jwk] }).to_string();
let rotated_jwks = serde_json::json!({ "keys": [rotated_jwk] }).to_string();
let Some((base, handle)) = start_mock_oidc_discovery_server_with_jwks(
|base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()),
4,
"ES256",
move |fetch| {
if fetch == 0 {
initial_jwks.clone()
} else {
rotated_jwks.clone()
}
},
) else {
return;
};
let config = build_mocked_oidc_provider_config("rotating", &base);
let policy = OutboundPolicy::from_allowed_origins(&base).expect("loopback origin should be allowed");
let http_client = ReqwestHttpClient::with_policy(policy);
let state = OidcSys::discover_provider(&config, &http_client)
.await
.expect("initial OIDC discovery should succeed");
let sys = OidcSys {
configs: HashMap::from([(config.id.clone(), config.clone())]),
provider_states: RwLock::new(HashMap::from([(config.id.clone(), state)])),
state_store: OidcStateStore::new(),
http_client,
};
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock should be after Unix epoch")
.as_secs();
let mut header = Header::new(Algorithm::ES256);
header.kid = Some("rotated".to_string());
let token = jsonwebtoken::encode(
&header,
&serde_json::json!({
"iss": base,
"sub": "rotated-user",
"aud": config.client_id,
"iat": now,
"exp": now + 300,
"groups": ["readwrite"],
}),
&rotated_key,
)
.expect("rotated OIDC token should sign");
let (claims, provider_id) = sys
.verify_web_identity_token(&token)
.await
.expect("verification should refresh JWKS and accept the rotated key");
assert_eq!(provider_id, "rotating");
assert_eq!(claims.sub, "rotated-user");
assert_eq!(claims.groups, vec!["readwrite"]);
handle.join().expect("rotating JWKS mock server should exit cleanly");
}
fn start_mock_oidc_tls_discovery_server<F>(
build_discovery_issuer: F,
max_requests: usize,
+8
View File
@@ -68,10 +68,18 @@ FUZZ_TARGET=path_containment ./scripts/fuzz/run.sh
# Nightly-style: 300s per target
MAX_TOTAL_TIME=300 ./scripts/fuzz/run.sh
# Replay a recorded libFuzzer seed
FUZZ_TARGET=path_containment FUZZ_SEED=123456789 ./scripts/fuzz/run.sh
# Skip build (use pre-built harness)
SKIP_BUILD=1 FUZZ_TARGET=local_metadata ./scripts/fuzz/run.sh
```
Each run writes `fuzz/artifacts/<target>/run-manifest.txt` with the target,
libFuzzer seed, time budget, Git revision and dirty state, and runner mode. CI
uploads that manifest with the corpus and any crash input so the exact run can
be replayed.
## CI Workflow
The GitHub Actions workflow (`.github/workflows/fuzz.yml`) uses a **build/run separation** pattern:
+14 -20
View File
@@ -62,6 +62,8 @@ pub struct ContainerResources {
pub cgroup_detected: bool,
/// Whether values were overridden by environment variables.
pub overridden: bool,
/// Pre-computed basis string for metrics ("cgroup" or "host").
pub basis: &'static str,
}
impl Default for ContainerResources {
@@ -71,6 +73,7 @@ impl Default for ContainerResources {
memory_bytes: 0,
cgroup_detected: false,
overridden: false,
basis: "host",
}
}
}
@@ -205,7 +208,7 @@ mod cgroup {
/// Detection priority:
/// 1. Environment variable overrides
/// 2. cgroup v1/v2 limits (Linux only)
/// 3. Host values from sysinfo
/// 3. Host values from sysinfo (single System instance for both CPU and memory)
fn detect_container_resources() -> ContainerResources {
// Check if cgroup detection is disabled
let cgroup_disabled = std::env::var(ENV_DISABLE_CGROUP_DETECTION)
@@ -234,30 +237,25 @@ fn detect_container_resources() -> ContainerResources {
let overridden = override_cores.is_some() || override_memory.is_some();
// Get host values for fallback
let host_cores = {
let mut sys =
sysinfo::System::new_with_specifics(sysinfo::RefreshKind::everything().without_memory().without_processes());
// Get host values from a single sysinfo::System instance (avoids double init)
let (host_cores, host_memory) = {
let mut sys = sysinfo::System::new_with_specifics(sysinfo::RefreshKind::everything().without_processes());
sys.refresh_cpu_all();
sys.cpus().len().max(1)
};
let host_memory = {
let mut sys = sysinfo::System::new();
sys.refresh_memory();
sys.total_memory()
(sys.cpus().len().max(1), sys.total_memory())
};
// Determine effective values: override > cgroup > host
let cpu_cores = override_cores.or(cgroup_cpus).unwrap_or(host_cores).max(1);
let memory_bytes = override_memory.or(cgroup_memory).unwrap_or(host_memory);
let basis = if cgroup_detected { "cgroup" } else { "host" };
ContainerResources {
cpu_cores,
memory_bytes,
cgroup_detected,
overridden,
basis,
}
}
@@ -273,12 +271,13 @@ pub fn container_resources() -> &'static ContainerResources {
/// Should be called once during startup to help operators verify detection.
pub fn log_container_resources() {
let res = container_resources();
let memory_mib = res.memory_bytes / (1024 * 1024);
if res.overridden {
tracing::info!(
cpu_cores = res.cpu_cores,
memory_bytes = res.memory_bytes,
memory_mib = res.memory_bytes / (1024 * 1024),
memory_mib,
cgroup_detected = res.cgroup_detected,
"container resources (overridden by environment variables)"
);
@@ -286,7 +285,7 @@ pub fn log_container_resources() {
tracing::info!(
cpu_cores = res.cpu_cores,
memory_bytes = res.memory_bytes,
memory_mib = res.memory_bytes / (1024 * 1024),
memory_mib,
"container resources (detected from cgroup)"
);
} else {
@@ -298,12 +297,6 @@ pub fn log_container_resources() {
}
}
/// Get the memory basis string for metrics.
pub fn memory_basis() -> &'static str {
let res = container_resources();
if res.cgroup_detected { "cgroup" } else { "host" }
}
// ============================================================================
// Tests
// ============================================================================
@@ -319,6 +312,7 @@ mod tests {
assert_eq!(resources.memory_bytes, 0);
assert!(!resources.cgroup_detected);
assert!(!resources.overridden);
assert_eq!(resources.basis, "host");
}
#[test]
+88 -7
View File
@@ -14,6 +14,7 @@
use std::env;
use std::ffi::OsString;
#[cfg(target_os = "linux")]
use std::fs;
use std::path::PathBuf;
use std::time::Duration;
@@ -63,16 +64,37 @@ impl HeartbeatConfig {
credential_store: CredentialStore,
state_path: impl Into<PathBuf>,
) -> Self {
let state_path = state_path.into();
Self {
endpoint: endpoint.into(),
root_ca_pem: root_ca_pem.into(),
identity_store,
credential_store,
state_path: state_path.into(),
state_path,
schedule: HeartbeatSchedule::default(),
}
}
#[cfg(any(target_os = "linux", test))]
pub(crate) fn state_only(state_root: PathBuf) -> Self {
Self {
endpoint: String::new(),
root_ca_pem: Vec::new(),
identity_store: IdentityStore::new(state_root.join("identity")),
credential_store: CredentialStore::new(state_root.join("credential")),
state_path: state_root.join("heartbeat/state.json"),
schedule: HeartbeatSchedule::default(),
}
}
pub(crate) fn transport_enabled(&self) -> bool {
!self.endpoint.is_empty()
}
pub(crate) fn state_root(&self) -> Option<&std::path::Path> {
self.state_path.parent().and_then(std::path::Path::parent)
}
pub fn from_env() -> Result<Option<Self>, HeartbeatConfigError> {
Self::from_env_values(
env::var_os(ENV_CONNECT_ENDPOINT),
@@ -90,19 +112,33 @@ impl HeartbeatConfig {
if !configured {
return Ok(None);
}
let (Some(endpoint), Some(root_ca_file), Some(state_dir)) = (endpoint, root_ca_file, state_dir) else {
let Some(state_dir) = state_dir else {
return Err(HeartbeatConfigError::Partial);
};
let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?;
let root_ca_file = PathBuf::from(root_ca_file);
let state_dir = PathBuf::from(state_dir);
if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() || state_dir.as_os_str().is_empty() {
if state_dir.as_os_str().is_empty() || endpoint.is_some() != root_ca_file.is_some() {
return Err(HeartbeatConfigError::Partial);
}
#[cfg(not(target_os = "linux"))]
return Err(HeartbeatConfigError::PlatformSecurity);
#[cfg(target_os = "linux")]
let (Some(endpoint), Some(root_ca_file)) = (endpoint, root_ca_file) else {
return Ok(Some(Self::state_only(state_dir)));
};
#[cfg(target_os = "linux")]
let endpoint = endpoint.into_string().map_err(|_| HeartbeatConfigError::EndpointEncoding)?;
#[cfg(target_os = "linux")]
let root_ca_file = PathBuf::from(root_ca_file);
#[cfg(target_os = "linux")]
if endpoint.is_empty() || root_ca_file.as_os_str().is_empty() {
return Err(HeartbeatConfigError::Partial);
}
#[cfg(target_os = "linux")]
let root_ca_pem = fs::read(&root_ca_file).map_err(|source| HeartbeatConfigError::RootCertificate {
path: root_ca_file,
source,
})?;
#[cfg(target_os = "linux")]
Ok(Some(Self::new(
endpoint,
root_ca_pem,
@@ -116,17 +152,19 @@ impl HeartbeatConfig {
#[derive(Debug, thiserror::Error)]
pub enum HeartbeatConfigError {
#[error(
"Connect heartbeat configuration requires RUSTFS_CONNECT_ENDPOINT, RUSTFS_CONNECT_ROOT_CA_FILE, and RUSTFS_CONNECT_STATE_DIR"
"Connect requires RUSTFS_CONNECT_STATE_DIR and either both or neither of RUSTFS_CONNECT_ENDPOINT and RUSTFS_CONNECT_ROOT_CA_FILE"
)]
Partial,
#[error("RUSTFS_CONNECT_ENDPOINT is not valid UTF-8")]
EndpointEncoding,
#[error("failed to read the Connect root CA at {path}: {source}")]
#[error("Connect root CA could not be read")]
RootCertificate {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("Connect inventory persistence requires Linux filesystem security guarantees")]
PlatformSecurity,
}
#[cfg(test)]
@@ -149,9 +187,34 @@ mod tests {
HeartbeatConfig::from_env_values(Some(OsString::from("https://connect.example/agent/")), None, None),
Err(HeartbeatConfigError::Partial)
));
assert!(matches!(
HeartbeatConfig::from_env_values(
Some(OsString::from("https://connect.example/agent/")),
Some(OsString::from("root.pem")),
None,
),
Err(HeartbeatConfigError::Partial)
));
assert!(matches!(
HeartbeatConfig::from_env_values(None, Some(OsString::from("root.pem")), Some(OsString::from("state"))),
Err(HeartbeatConfigError::Partial)
));
}
#[test]
#[cfg(target_os = "linux")]
fn state_directory_alone_enables_local_inventory_without_transport() {
let state = tempfile::tempdir().expect("tempdir").keep();
let config = HeartbeatConfig::from_env_values(None, None, Some(state.clone().into_os_string()))
.expect("state-only config")
.expect("enabled config");
assert_eq!(config.state_root(), Some(state.as_path()));
assert!(!config.transport_enabled());
}
#[test]
#[cfg(target_os = "linux")]
fn complete_environment_builds_the_durable_paths() {
let temp = tempfile::tempdir().expect("tempdir");
let root = temp.path().join("root.pem");
@@ -168,6 +231,24 @@ mod tests {
assert_eq!(config.endpoint, "https://connect.example/agent/");
assert_eq!(config.root_ca_pem, b"root certificate");
assert_eq!(config.state_path, state.join("heartbeat/state.json"));
assert_eq!(config.state_root(), Some(state.as_path()));
assert!(!state.exists(), "parsing configuration must not create state");
}
#[test]
#[cfg(not(target_os = "linux"))]
fn configured_inventory_fails_without_linux_filesystem_guarantees() {
assert!(matches!(
HeartbeatConfig::from_env_values(None, None, Some(OsString::from("state"))),
Err(HeartbeatConfigError::PlatformSecurity)
));
assert!(matches!(
HeartbeatConfig::from_env_values(
Some(OsString::from("https://connect.example/agent/")),
Some(OsString::from("missing-root.pem")),
Some(OsString::from("state")),
),
Err(HeartbeatConfigError::PlatformSecurity)
));
}
}
File diff suppressed because it is too large Load Diff
+163 -31
View File
@@ -32,7 +32,6 @@ pub struct HeartbeatRuntime {
shutdown: CancellationToken,
status: watch::Receiver<HeartbeatStatus>,
task: Option<JoinHandle<()>>,
inventory: Option<InventoryRuntime>,
}
impl HeartbeatRuntime {
@@ -40,22 +39,11 @@ impl HeartbeatRuntime {
self.status.clone()
}
pub(crate) fn with_inventory(mut self, inventory: Option<InventoryRuntime>) -> Self {
self.inventory = inventory;
self
}
pub async fn shutdown(mut self) {
self.shutdown.cancel();
if let Some(inventory) = self.inventory.as_ref() {
inventory.shutdown.cancel();
}
if let Some(task) = self.task.take() {
let _ = task.await;
}
if let Some(inventory) = self.inventory.take() {
inventory.shutdown().await;
}
}
}
@@ -90,6 +78,20 @@ impl Drop for InventoryRuntime {
}
}
pub(crate) async fn shutdown_connect_runtimes(heartbeat: Option<HeartbeatRuntime>, inventory: Option<InventoryRuntime>) {
let heartbeat = async move {
if let Some(runtime) = heartbeat {
runtime.shutdown().await;
}
};
let inventory = async move {
if let Some(runtime) = inventory {
runtime.shutdown().await;
}
};
tokio::join!(heartbeat, inventory);
}
pub fn spawn_heartbeat_runtime<F>(
config: Option<HeartbeatConfig>,
parent_shutdown: &CancellationToken,
@@ -101,6 +103,9 @@ where
let Some(config) = config else {
return Ok(None);
};
if !config.transport_enabled() {
return Ok(None);
}
let sender = HeartbeatSender::new(config.clone())?;
let store = HeartbeatStateStore::new(config.state_path.clone());
let lock = store.try_runtime_lock()?;
@@ -163,7 +168,6 @@ where
shutdown,
status: status_rx,
task: Some(task),
inventory: None,
}))
}
@@ -184,9 +188,14 @@ where
return Err(InventoryError::Schedule);
}
let retry_schedule = config.schedule;
let store = InventoryStateStore::from_heartbeat_path(&config.state_path)?;
let state_root = config.state_root().ok_or(InventoryError::StatePath)?;
let store = InventoryStateStore::from_state_root(state_root)?;
let lock = store.try_runtime_lock()?;
let sender = InventorySender::new(config)?;
let sender = if config.transport_enabled() {
Some(InventorySender::new(config)?)
} else {
None
};
let shutdown = parent_shutdown.child_token();
let task_shutdown = shutdown.clone();
let (status_tx, status_rx) = watch::channel(InventoryStatus::Starting);
@@ -197,8 +206,19 @@ where
if task_shutdown.is_cancelled() {
break;
}
let pending = match store.pending().await {
Ok(Some(pending)) => pending,
let pending = match if sender.is_some() { store.pending().await } else { Ok(None) } {
Ok(Some((pending, captured_at))) => {
if let Err(error) = store
.ensure_latest(pending.snapshot().clone(), captured_at, task_shutdown.clone())
.await
{
if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() {
break;
}
return failed_inventory(&status_tx, error);
}
pending
}
Ok(None) => {
let snapshot = match cancellable(&task_shutdown, sample()).await {
Some(Ok(snapshot)) => snapshot,
@@ -218,6 +238,27 @@ where
Ok(content_hash) => content_hash,
Err(error) => return failed_inventory(&status_tx, error),
};
let captured_at = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
if let Err(error) = store
.publish_latest(snapshot.clone(), captured_at, task_shutdown.clone())
.await
{
if matches!(error, InventoryError::Cancelled) && task_shutdown.is_cancelled() {
break;
}
return failed_inventory(&status_tx, error);
}
if task_shutdown.is_cancelled() {
break;
}
if sender.is_none() {
backoff = retry_schedule.initial_backoff;
let _ = status_tx.send(InventoryStatus::Unchanged { content_hash });
if sleep_or_cancel(&task_shutdown, schedule.cadence.saturating_add(jitter(schedule.jitter))).await {
break;
}
continue;
}
match store.prepare(snapshot).await {
Ok(Some(pending)) => pending,
Ok(None) => {
@@ -233,7 +274,15 @@ where
}
Err(error) => return failed_inventory(&status_tx, error),
};
let delivery = match cancellable(&task_shutdown, sender.send(&pending)).await {
let delivery = match cancellable(
&task_shutdown,
sender
.as_ref()
.expect("sender exists when delivery state is prepared")
.send(&pending),
)
.await
{
Some(Ok(delivery)) => delivery,
Some(Err(error)) => return failed_inventory(&status_tx, error),
None => break,
@@ -261,14 +310,13 @@ where
let _ = status_tx.send(InventoryStatus::BackingOff { delay });
delay
}
InventoryDelivery::AuthenticationStopped { status, reason } => {
let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason });
InventoryDelivery::AuthenticationStopped { status } => {
let _ = status_tx.send(InventoryStatus::AuthenticationStopped { status, reason: None });
return;
}
InventoryDelivery::Rejected { status, reason } => {
let suffix = reason.map_or_else(String::new, |reason| format!("; reason={reason}"));
InventoryDelivery::Rejected { status } => {
let _ = status_tx.send(InventoryStatus::Failed {
reason: format!("Connect rejected inventory with HTTP {status}{suffix}"),
reason: format!("connect_inventory_rejected_http_{status}"),
});
return;
}
@@ -292,6 +340,46 @@ fn failed(status: &watch::Sender<HeartbeatStatus>, error: HeartbeatError) {
});
}
pub(crate) fn heartbeat_failure_reason(error: &HeartbeatError) -> &'static str {
use super::registration::CredentialValidationError;
match error {
HeartbeatError::Endpoint => "connect_heartbeat_endpoint",
HeartbeatError::RootCertificate => "connect_heartbeat_root_certificate",
HeartbeatError::Schedule => "connect_heartbeat_schedule",
HeartbeatError::NotRegistered => "connect_heartbeat_not_registered",
HeartbeatError::IdentityMissing => "connect_heartbeat_identity_missing",
HeartbeatError::IdentityCertificate => "connect_heartbeat_identity_certificate",
HeartbeatError::CredentialName => "connect_heartbeat_credential_name",
HeartbeatError::CredentialExpired => "connect_heartbeat_credential_expired",
HeartbeatError::NodeSummary => "connect_heartbeat_node_summary",
HeartbeatError::SequenceExhausted => "connect_heartbeat_sequence_exhausted",
HeartbeatError::AlreadyRunning => "connect_heartbeat_already_running",
HeartbeatError::StateConflict => "connect_heartbeat_state_conflict",
HeartbeatError::StateIo { .. } => "connect_heartbeat_state_io",
HeartbeatError::StateInvalid { .. } => "connect_heartbeat_state_invalid",
HeartbeatError::StateCorrupt { .. } => "connect_heartbeat_state_corrupt",
#[cfg(unix)]
HeartbeatError::StatePermissions { .. } => "connect_heartbeat_state_permissions",
HeartbeatError::ResponseTooLarge => "connect_heartbeat_response_too_large",
HeartbeatError::Response => "connect_heartbeat_response",
HeartbeatError::Url(_) => "connect_heartbeat_url",
HeartbeatError::Transport(_) => "connect_heartbeat_transport",
HeartbeatError::Identity(_) => "connect_heartbeat_identity",
HeartbeatError::IdentityStore(_) => "connect_heartbeat_identity_store",
HeartbeatError::CredentialStore(_) => "connect_heartbeat_credential_store",
HeartbeatError::CredentialValidation(error) => match error {
CredentialValidationError::Certificate => "connect_heartbeat_credential_certificate",
CredentialValidationError::Chain => "connect_heartbeat_credential_chain",
CredentialValidationError::Identity => "connect_heartbeat_credential_identity",
CredentialValidationError::Key => "connect_heartbeat_credential_key",
CredentialValidationError::Validity => "connect_heartbeat_credential_validity",
CredentialValidationError::CertificateRequest => "connect_heartbeat_credential_request",
CredentialValidationError::RotationTranscript => "connect_heartbeat_credential_rotation_transcript",
},
}
}
fn failed_inventory(status: &watch::Sender<InventoryStatus>, error: InventoryError) {
let _ = status.send(InventoryStatus::Failed {
reason: error.to_string(),
@@ -327,7 +415,51 @@ mod tests {
use super::*;
#[tokio::test]
async fn heartbeat_shutdown_cancels_inventory_before_waiting_for_heartbeat() {
async fn state_only_configuration_does_not_start_heartbeat() {
let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory");
let config = HeartbeatConfig::state_only(temp.path().to_path_buf());
let shutdown = CancellationToken::new();
assert!(
spawn_heartbeat_runtime(Some(config), &shutdown, || { CoarseNodeSummary::new(1, 0, 0).expect("summary") })
.expect("disabled transport")
.is_none()
);
}
#[test]
fn runtimes_expose_only_stable_machine_reasons() {
let error = HeartbeatError::StateIo {
path: std::path::PathBuf::from("/private/connect/state.json"),
source: std::io::Error::other("transport.internal"),
};
assert_eq!(heartbeat_failure_reason(&error), "connect_heartbeat_state_io");
assert_eq!(
heartbeat_failure_reason(&HeartbeatError::StateCorrupt {
path: std::path::PathBuf::from("/private/connect/state.json"),
}),
"connect_heartbeat_state_corrupt"
);
assert_eq!(
heartbeat_failure_reason(&HeartbeatError::CredentialExpired),
"connect_heartbeat_credential_expired"
);
assert_eq!(
heartbeat_failure_reason(&HeartbeatError::CredentialValidation(
super::super::registration::CredentialValidationError::Identity
)),
"connect_heartbeat_credential_identity"
);
assert_eq!(
heartbeat_failure_reason(&HeartbeatError::CredentialValidation(
super::super::registration::CredentialValidationError::Key
)),
"connect_heartbeat_credential_key"
);
}
#[tokio::test]
async fn unified_shutdown_cancels_inventory_before_waiting_for_heartbeat() {
let heartbeat_shutdown = CancellationToken::new();
let inventory_shutdown = CancellationToken::new();
let task_inventory_shutdown = inventory_shutdown.clone();
@@ -342,18 +474,18 @@ mod tests {
task_inventory_shutdown.cancelled().await;
let _ = inventory_stopped.send(());
});
let runtime = HeartbeatRuntime {
let heartbeat = HeartbeatRuntime {
shutdown: heartbeat_shutdown,
status: heartbeat_status,
task: Some(heartbeat_task),
inventory: Some(InventoryRuntime {
shutdown: inventory_shutdown,
status: inventory_status,
task: Some(inventory_task),
}),
};
let inventory = InventoryRuntime {
shutdown: inventory_shutdown,
status: inventory_status,
task: Some(inventory_task),
};
let shutdown = tokio::spawn(runtime.shutdown());
let shutdown = tokio::spawn(shutdown_connect_runtimes(Some(heartbeat), Some(inventory)));
tokio::time::timeout(Duration::from_millis(250), stopped)
.await
.expect("inventory cancellation must not wait for heartbeat")
+2 -2
View File
@@ -391,8 +391,8 @@ pub fn memory_observability_controller_snapshot(ctx: &CancellationToken) -> Memo
/// Record the effective memory total and its basis (host or cgroup).
fn record_effective_memory(total_bytes: u64) {
let basis = crate::cgroup_resources::memory_basis();
metrics::gauge!("rustfs_memory_effective_total_bytes", "basis" => basis.to_string()).set(total_bytes as f64);
let basis = crate::cgroup_resources::container_resources().basis;
metrics::gauge!("rustfs_memory_effective_total_bytes", "basis" => basis).set(total_bytes as f64);
}
/// Record container resource detection results.
+3 -3
View File
@@ -14,6 +14,7 @@
use crate::storage_api::startup::lifecycle::ECStore;
use crate::{
connect::runtime::shutdown_connect_runtimes,
server::{ServiceStateManager, ShutdownHandle, start_persisted_event_notifier_reconciler, wait_for_shutdown},
startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap},
startup_runtime_sources,
@@ -129,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
let StartupServiceRuntime {
optional_runtimes,
heartbeat,
inventory,
iam_bootstrap,
enable_scanner,
} = service_runtime;
@@ -163,9 +165,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec
shutdown_token,
)
.await;
if let Some(heartbeat) = heartbeat {
heartbeat.shutdown().await;
}
shutdown_connect_runtimes(heartbeat, inventory).await;
if let Err(err) = event_notifier_reconciler.await {
tracing::warn!(
target: "rustfs::main::run",
+23 -4
View File
@@ -17,8 +17,9 @@ use crate::storage_api::startup::services::{ECStore, EndpointServerPools, Server
use crate::{
config::Config,
connect::{
CoarseNodeSummary, HeartbeatConfig, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime, InventorySchedule,
InventorySnapshot, spawn_heartbeat_runtime, spawn_inventory_runtime,
CoarseNodeSummary, HeartbeatConfig, HeartbeatError, HeartbeatRuntime, InventoryError, InventoryFlag, InventoryRuntime,
InventorySchedule, InventorySnapshot, runtime::heartbeat_failure_reason, spawn_heartbeat_runtime,
spawn_inventory_runtime,
},
init::{init_buffer_profile_system, init_kms_system},
server::ServiceStateManager,
@@ -40,6 +41,7 @@ use tokio_util::sync::CancellationToken;
pub(crate) struct StartupServiceRuntime {
pub(crate) optional_runtimes: OptionalRuntimeServices,
pub(crate) heartbeat: Option<HeartbeatRuntime>,
pub(crate) inventory: Option<InventoryRuntime>,
pub(crate) iam_bootstrap: IamBootstrapDisposition,
pub(crate) enable_scanner: bool,
}
@@ -104,11 +106,11 @@ pub(crate) async fn init_startup_runtime_services(
init_observability_runtime(store.clone(), ctx.clone()).await;
let heartbeat = start_heartbeat_runtime(heartbeat_config.clone(), heartbeat_nodes, &ctx)?;
let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?;
let heartbeat = heartbeat.map(|heartbeat| heartbeat.with_inventory(inventory));
Ok(StartupServiceRuntime {
optional_runtimes,
heartbeat,
inventory,
iam_bootstrap,
enable_scanner,
})
@@ -122,11 +124,18 @@ fn start_heartbeat_runtime(
let Some(config) = config else {
return Ok(None);
};
if !config.transport_enabled() {
return Ok(None);
}
let summary = u16::try_from(node_count.unwrap_or_default())
.ok()
.and_then(|total| CoarseNodeSummary::new(total, 0, 0).ok())
.ok_or_else(|| std::io::Error::other("Connect heartbeat node count is outside protocol bounds"))?;
spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(std::io::Error::other)
spawn_heartbeat_runtime(Some(config), shutdown, move || summary).map_err(startup_heartbeat_error)
}
fn startup_heartbeat_error(error: HeartbeatError) -> std::io::Error {
std::io::Error::other(heartbeat_failure_reason(&error))
}
fn start_inventory_runtime(
@@ -318,6 +327,16 @@ fn aggregate_inventory_capacity(
mod tests {
use super::*;
#[test]
fn heartbeat_startup_errors_expose_only_stable_codes() {
let error = startup_heartbeat_error(HeartbeatError::StateIo {
path: std::path::PathBuf::from("/private/connect/canary/state.json"),
source: std::io::Error::other("private-source-canary"),
});
assert_eq!(error.to_string(), "connect_heartbeat_state_io");
}
fn disk(state: &str, runtime_state: Option<&str>, disk_index: i32) -> rustfs_madmin::Disk {
rustfs_madmin::Disk {
state: state.to_owned(),
+131 -14
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#![cfg(target_os = "linux")]
use std::collections::VecDeque;
use std::fs;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -46,6 +48,10 @@ const CLUSTER_UID: &str = "0198f4b0-2b00-7d20-9e31-3f4a5b6c7d81";
const DEVICE_UID: &str = "0198f4b0-3c00-7e30-8f41-4a5b6c7d8e92";
const SNAPSHOT_UID: &str = "0198f4b0-4d00-7f40-9051-5b6c7d8e9fa3";
fn safe_tempdir() -> tempfile::TempDir {
tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe temporary directory")
}
struct TestPki {
root_params: CertificateParams,
root_key: KeyPair,
@@ -244,6 +250,7 @@ fn config(temp: &tempfile::TempDir, pki: &TestPki, server: &TestServer) -> Heart
if let Err(error) = fs::create_dir(temp.path().join("private-config-secret")) {
assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists, "Connect state root");
}
private_directory_mode(&temp.path().join("private-config-secret"));
HeartbeatConfig {
endpoint: server.endpoint.clone(),
root_ca_pem: pki.root_pem.as_bytes().to_vec(),
@@ -394,6 +401,38 @@ fn connect_inventory_bounds_fail_instead_of_truncating_or_inventing_values() {
));
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn connect_inventory_state_only_persists_without_constructing_transport() {
let temp = safe_tempdir();
let state = temp.path().join("state");
fs::create_dir(&state).expect("state root");
private_directory_mode(&state);
let config = HeartbeatConfig::new(
"",
Vec::new(),
IdentityStore::new(state.join("identity")),
CredentialStore::new(state.join("credential")),
state.join("heartbeat/state.json"),
);
let shutdown = CancellationToken::new();
let runtime = spawn_inventory_runtime(Some(config), schedule(), &shutdown, || std::future::ready(Ok(snapshot())))
.expect("state-only inventory")
.expect("configured inventory");
let mut status = runtime.status();
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::Unchanged { .. })).await,
InventoryStatus::Unchanged { .. }
));
let envelope: Value = serde_json::from_slice(&fs::read(state.join("inventory/latest.json")).expect("latest inventory"))
.expect("latest envelope");
assert_eq!(envelope["formatVersion"], "v1");
assert_eq!(envelope["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
assert_eq!(envelope.as_object().expect("envelope object").len(), 4);
runtime.shutdown().await;
}
#[tokio::test]
async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_or_network() {
let pki = TestPki::new();
@@ -409,7 +448,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o
"coarseFlags": []
}))
.expect("serde should not bypass the runtime validation boundary");
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, move || {
std::future::ready(Ok(invalid.clone()))
@@ -420,7 +459,7 @@ async fn connect_inventory_rejects_deserialized_invalid_snapshots_before_state_o
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await,
InventoryStatus::Failed { reason } if reason.contains("version is outside protocol bounds")
InventoryStatus::Failed { reason } if reason == "connect_inventory_snapshot_version"
));
assert!(!temp.path().join("private-config-secret/inventory/state.json").exists());
runtime.shutdown().await;
@@ -433,7 +472,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
let pki = TestPki::new();
let content_hash = snapshot().content_hash().expect("content hash");
let first_server = server(&pki, vec![Reply::error(StatusCode::SERVICE_UNAVAILABLE, "UNAVAILABLE")]).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let samples = Arc::new(AtomicUsize::new(0));
let sampled = samples.clone();
@@ -451,7 +490,40 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
));
assert_eq!(samples.load(Ordering::Relaxed), 1);
let original = first_server.seen.lock().expect("seen lock")[0].clone();
let latest: Value = serde_json::from_slice(
&fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("latest inventory"),
)
.expect("latest envelope");
assert_eq!(latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
for field in [
"rustfsVersion",
"osVersion",
"nodeCount",
"driveCount",
"capacityTotalBytes",
"capacityUsedBytes",
"coarseFlags",
] {
assert_eq!(latest["snapshot"][field], original[field]);
}
runtime.shutdown().await;
let state = temp.path().join("private-config-secret/inventory/state.json");
let legacy_persisted_at = std::time::SystemTime::now() - Duration::from_secs(60 * 60);
fs::File::options()
.write(true)
.open(&state)
.expect("pending state")
.set_times(std::fs::FileTimes::new().set_modified(legacy_persisted_at))
.expect("legacy pending timestamp");
let legacy_persisted_at = chrono::DateTime::<chrono::Utc>::from(
fs::metadata(&state)
.and_then(|metadata| metadata.modified())
.expect("persisted pending timestamp"),
)
.format("%Y-%m-%dT%H:%M:%SZ")
.to_string();
fs::remove_file(temp.path().join("private-config-secret/inventory/latest.json"))
.expect("simulate a pending snapshot created before local persistence");
let mut limited = Reply::error(StatusCode::TOO_MANY_REQUESTS, "RATE_LIMITED");
limited.retry_after = Some("0");
@@ -473,10 +545,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
if accepted == content_hash && received_at == "2026-08-22T01:02:03Z"
));
assert_eq!(restart_samples.load(Ordering::Relaxed), 0);
let restored_latest: Value = serde_json::from_slice(
&fs::read(temp.path().join("private-config-secret/inventory/latest.json")).expect("restored latest inventory"),
)
.expect("restored latest envelope");
assert_eq!(restored_latest["snapshot"], serde_json::to_value(snapshot()).expect("snapshot JSON"));
assert_eq!(restored_latest["capturedAt"], legacy_persisted_at);
let delivered = restart_server.seen.lock().expect("seen lock").clone();
assert_eq!(delivered, vec![original.clone(), original.clone()]);
assert_eq!(original["sequence"], 0);
let encoded = serde_json::to_string(&original).expect("request JSON");
let persisted = serde_json::to_string(&latest).expect("persisted JSON");
for forbidden in [
"private-config-secret",
"BEGIN CERTIFICATE",
@@ -486,9 +565,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
"path",
] {
assert!(!encoded.contains(forbidden), "request exposed {forbidden}");
assert!(!persisted.contains(forbidden), "persisted inventory exposed {forbidden}");
}
assert_eq!(original.as_object().expect("request object").len(), 10);
restart.shutdown().await;
#[cfg(target_os = "linux")]
let latest_before_unchanged = {
use std::os::unix::fs::MetadataExt as _;
fs::metadata(temp.path().join("private-config-secret/inventory/latest.json"))
.expect("latest metadata")
.ino()
};
let unchanged_samples = Arc::new(AtomicUsize::new(0));
let sampled = unchanged_samples.clone();
@@ -505,6 +592,17 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
));
assert_eq!(unchanged_samples.load(Ordering::Relaxed), 1);
assert_eq!(restart_server.seen.lock().expect("seen lock").len(), 2);
#[cfg(target_os = "linux")]
{
use std::os::unix::fs::MetadataExt as _;
assert_ne!(
fs::metadata(temp.path().join("private-config-secret/inventory/latest.json"))
.expect("refreshed latest metadata")
.ino(),
latest_before_unchanged,
"a complete unchanged sample must refresh the local envelope"
);
}
unchanged.shutdown().await;
}
@@ -512,7 +610,7 @@ async fn connect_inventory_restart_replays_the_pending_request_and_then_skips_un
async fn connect_inventory_disconnect_retries_without_resampling() {
let pki = TestPki::new();
let unavailable = server(&pki, Vec::new()).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let config = config(&temp, &pki, &unavailable);
drop(unavailable);
let shutdown = CancellationToken::new();
@@ -542,7 +640,7 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() {
let pki = TestPki::new();
let content_hash = snapshot().content_hash().expect("content hash");
let server = server(&pki, vec![Reply::ok(&content_hash)]).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let samples = Arc::new(AtomicUsize::new(0));
let sampled = samples.clone();
@@ -561,6 +659,11 @@ async fn connect_inventory_retries_an_incomplete_sample_before_delivery() {
.expect("configured inventory");
let mut status = runtime.status();
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::BackingOff { .. })).await,
InventoryStatus::BackingOff { .. }
));
assert!(!temp.path().join("private-config-secret/inventory/latest.json").exists());
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::Online { .. })).await,
InventoryStatus::Online { content_hash: accepted, .. } if accepted == content_hash
@@ -575,7 +678,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() {
let pki = TestPki::new();
let content_hash = snapshot().content_hash().expect("content hash");
let server = server(&pki, vec![Reply::ok(&content_hash)]).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let config = config(&temp, &pki, &server);
let seed = spawn_inventory_runtime(Some(config.clone()), schedule(), &shutdown, || std::future::ready(Ok(snapshot())))
@@ -645,7 +748,7 @@ async fn connect_inventory_unchanged_sample_resets_incomplete_backoff() {
async fn connect_inventory_revoked_device_stops_without_retrying() {
let pki = TestPki::new();
let server = server(&pki, vec![Reply::error(StatusCode::UNAUTHORIZED, "DEVICE_REVOKED")]).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let shutdown = CancellationToken::new();
let runtime = spawn_inventory_runtime(Some(config(&temp, &pki, &server)), schedule(), &shutdown, || {
std::future::ready(Ok(snapshot()))
@@ -656,7 +759,10 @@ async fn connect_inventory_revoked_device_stops_without_retrying() {
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::AuthenticationStopped { .. })).await,
InventoryStatus::AuthenticationStopped { status: 401, reason: Some(reason) } if reason == "DEVICE_REVOKED"
InventoryStatus::AuthenticationStopped {
status: 401,
reason: None
}
));
assert_eq!(server.seen.lock().expect("seen lock").len(), 1);
runtime.shutdown().await;
@@ -666,10 +772,11 @@ async fn connect_inventory_revoked_device_stops_without_retrying() {
async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_delivery() {
let pki = TestPki::new();
let server = server(&pki, Vec::new()).await;
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let config = config(&temp, &pki, &server);
let state = temp.path().join("private-config-secret/inventory/state.json");
fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory");
private_directory_mode(state.parent().expect("state directory"));
fs::write(
&state,
br#"{"nextSequence":9007199254740992,"pending":null,"lastAcceptedContentHash":null}"#,
@@ -689,7 +796,7 @@ async fn connect_inventory_sequence_overflow_fails_before_sampling_or_network_de
assert!(matches!(
wait_for(&mut status, |status| matches!(status, InventoryStatus::Failed { .. })).await,
InventoryStatus::Failed { reason } if reason.contains("sequence is exhausted")
InventoryStatus::Failed { reason } if reason == "connect_inventory_sequence_exhausted"
));
assert_eq!(samples.load(Ordering::Relaxed), 0);
assert!(server.seen.lock().expect("seen lock").is_empty());
@@ -702,10 +809,11 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
let server = server(&pki, Vec::new()).await;
for invalid_case in ["flags", "os-version"] {
let temp = tempfile::tempdir().expect("tempdir");
let temp = safe_tempdir();
let config = config(&temp, &pki, &server);
let state = temp.path().join("private-config-secret/inventory/state.json");
fs::create_dir_all(state.parent().expect("state directory")).expect("create state directory");
private_directory_mode(state.parent().expect("state directory"));
let mut pending = json!({
"protocolVersion": "v1",
"requestId": "00000000-0000-4000-8000-000000000001",
@@ -753,7 +861,7 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
matches!(status, InventoryStatus::Failed { .. } | InventoryStatus::BackingOff { .. })
})
.await,
InventoryStatus::Failed { reason } if reason.contains("violates the protocol invariants")
InventoryStatus::Failed { reason } if reason == "connect_inventory_state_corrupt"
));
assert_eq!(samples.load(Ordering::Relaxed), 0);
runtime.shutdown().await;
@@ -762,11 +870,20 @@ async fn connect_inventory_rejects_noncanonical_persisted_snapshots_before_deliv
assert!(server.seen.lock().expect("seen lock").is_empty());
}
#[cfg(unix)]
#[cfg(target_os = "linux")]
fn private_mode(path: &std::path::Path) {
use std::os::unix::fs::PermissionsExt as _;
fs::set_permissions(path, fs::Permissions::from_mode(0o600)).expect("private permissions");
}
#[cfg(not(unix))]
#[cfg(target_os = "linux")]
fn private_directory_mode(path: &std::path::Path) {
use std::os::unix::fs::PermissionsExt as _;
fs::set_permissions(path, fs::Permissions::from_mode(0o700)).expect("private directory permissions");
}
#[cfg(not(target_os = "linux"))]
fn private_mode(_path: &std::path::Path) {}
#[cfg(not(target_os = "linux"))]
fn private_directory_mode(_path: &std::path::Path) {}
+44 -4
View File
@@ -25,6 +25,7 @@
# Environment variables:
# FUZZ_TARGET — run only this target (default: all smoke targets)
# MAX_TOTAL_TIME — seconds to fuzz per target (default: 60)
# FUZZ_SEED — replay one libFuzzer seed (default: generate and record one per target)
# ARTIFACT_ROOT — artifact output directory (default: artifacts)
# BUILD_ONLY — set to 1 to skip fuzz runs (default: 0)
# SKIP_BUILD — set to 1 to skip build phase (default: 0)
@@ -37,6 +38,7 @@ SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
REPO_ROOT=$(CDPATH= cd -- "$SCRIPT_DIR/../.." && pwd)
FUZZ_DIR="$REPO_ROOT/fuzz"
MAX_TOTAL_TIME=${MAX_TOTAL_TIME:-60}
FUZZ_SEED=${FUZZ_SEED:-}
ARTIFACT_ROOT=${ARTIFACT_ROOT:-artifacts}
FUZZ_TARGET=${FUZZ_TARGET:-}
BUILD_ONLY=${BUILD_ONLY:-0}
@@ -44,6 +46,15 @@ SKIP_BUILD=${SKIP_BUILD:-0}
USE_PREBUILT_BINARY=${USE_PREBUILT_BINARY:-0}
PREBUILT_BINARY_DIR=${PREBUILT_BINARY_DIR:-}
if [ -n "$FUZZ_SEED" ]; then
case "$FUZZ_SEED" in
0*|*[!0-9]*)
echo "FUZZ_SEED must be a positive decimal integer without leading zeroes: $FUZZ_SEED" >&2
exit 1
;;
esac
fi
cd "$FUZZ_DIR"
mkdir -p "$ARTIFACT_ROOT"
@@ -75,6 +86,35 @@ for target in $targets; do
mkdir -p "$artifact_dir"
mkdir -p "$corpus_dir"
seed="$FUZZ_SEED"
if [ -z "$seed" ]; then
seed=$(printf '%s\n' "${GITHUB_RUN_ID:-local}:${GITHUB_RUN_ATTEMPT:-0}:$target:$(date +%s):$$" | cksum | awk '{print $1}')
if [ "$seed" = "0" ]; then
seed=1
fi
fi
revision=$(git -C "$REPO_ROOT" rev-parse HEAD 2>/dev/null || printf 'unknown')
if [ "$revision" = "unknown" ]; then
git_dirty="unknown"
elif [ -n "$(git -C "$REPO_ROOT" status --porcelain --untracked-files=normal 2>/dev/null)" ]; then
git_dirty="true"
else
git_dirty="false"
fi
if [ "$USE_PREBUILT_BINARY" = "1" ]; then
runner_mode="prebuilt"
else
runner_mode="cargo-fuzz"
fi
{
printf 'target=%s\n' "$target"
printf 'seed=%s\n' "$seed"
printf 'max_total_time=%s\n' "$MAX_TOTAL_TIME"
printf 'git_revision=%s\n' "$revision"
printf 'git_dirty=%s\n' "$git_dirty"
printf 'runner_mode=%s\n' "$runner_mode"
} > "$artifact_dir/run-manifest.txt"
if [ "$USE_PREBUILT_BINARY" = "1" ]; then
binary_dir="$PREBUILT_BINARY_DIR"
if [ -z "$binary_dir" ]; then
@@ -89,11 +129,11 @@ for target in $targets; do
echo "Missing executable prebuilt fuzz binary: $binary_path" >&2
exit 1
fi
echo "==> $binary_path (-max_total_time=$MAX_TOTAL_TIME, -artifact_prefix=$artifact_dir/, corpus=$corpus_dir)"
"$binary_path" -max_total_time="$MAX_TOTAL_TIME" -artifact_prefix="$artifact_dir/" "$corpus_dir"
echo "==> $binary_path (-max_total_time=$MAX_TOTAL_TIME, -seed=$seed, -artifact_prefix=$artifact_dir/, corpus=$corpus_dir)"
"$binary_path" -max_total_time="$MAX_TOTAL_TIME" -seed="$seed" -artifact_prefix="$artifact_dir/" "$corpus_dir"
continue
fi
echo "==> cargo +nightly fuzz run $target (-max_total_time=$MAX_TOTAL_TIME, -artifact_prefix=$artifact_dir/)"
cargo +nightly fuzz run "$target" -- -max_total_time="$MAX_TOTAL_TIME" -artifact_prefix="$artifact_dir/"
echo "==> cargo +nightly fuzz run $target (-max_total_time=$MAX_TOTAL_TIME, -seed=$seed, -artifact_prefix=$artifact_dir/)"
cargo +nightly fuzz run "$target" -- -max_total_time="$MAX_TOTAL_TIME" -seed="$seed" -artifact_prefix="$artifact_dir/"
done
@@ -6,8 +6,8 @@
# where it stays for the rest of the job. On this repository that matters more
# than usual: pull_request jobs run on self-hosted runners and execute the PR's
# own build.rs, proc-macros and tests, any of which can read that file. The
# Test and Lint job additionally holds actions: write, so its token can cancel
# runs and delete the Actions caches the whole pipeline depends on.
# post-failure cancellation job holds actions: write but never checks out
# repository code, so untrusted build scripts cannot read that token from Git.
#
# A checkout is exempt only when the token IS the credential the job needs —
# helm-package pushes to rustfs/helm with it. Mark those with a comment
+42
View File
@@ -0,0 +1,42 @@
{
"realm": "rustfs-ci",
"enabled": true,
"sslRequired": "none",
"accessTokenLifespan": 300,
"clients": [
{
"clientId": "rustfs-ci",
"secret": "rustfs-ci-secret",
"enabled": true,
"protocol": "openid-connect",
"publicClient": false,
"standardFlowEnabled": false,
"directAccessGrantsEnabled": true
},
{
"clientId": "wrong-audience",
"secret": "wrong-audience-secret",
"enabled": true,
"protocol": "openid-connect",
"publicClient": false,
"standardFlowEnabled": false,
"directAccessGrantsEnabled": true
}
],
"users": [
{
"id": "00000000-0000-0000-0000-000000000001",
"username": "alice",
"email": "alice@example.test",
"emailVerified": true,
"enabled": true,
"credentials": [
{
"type": "password",
"value": "alice-password",
"temporary": false
}
]
}
]
}
+241
View File
@@ -0,0 +1,241 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
RUSTFS_BINARY="${1:-${ROOT_DIR}/target/debug/rustfs}"
REALM_FIXTURE="${ROOT_DIR}/scripts/test/fixtures/keycloak-rustfs-ci-realm.json"
KEYCLOAK_IMAGE="${KEYCLOAK_IMAGE:-quay.io/keycloak/keycloak@sha256:6a7217a100bd3e5de4063a27a538ef999a3c5a88c4b4ec0ffc0a642aee7b2597}"
WORK_DIR="${RUNNER_TEMP:-/tmp}/rustfs-keycloak-live-${$}"
KEYCLOAK_CONTAINER="rustfs-keycloak-live-${$}"
RUSTFS_PID=""
mkdir -p "${WORK_DIR}"
cleanup() {
local status=$?
if [[ -n "${RUSTFS_PID}" ]] && kill -0 "${RUSTFS_PID}" 2>/dev/null; then
kill "${RUSTFS_PID}" 2>/dev/null || true
wait "${RUSTFS_PID}" 2>/dev/null || true
fi
if [[ "${status}" -ne 0 ]]; then
docker logs "${KEYCLOAK_CONTAINER}" >"${WORK_DIR}/keycloak.log" 2>&1 || true
echo "live-gate logs retained in ${WORK_DIR}" >&2
fi
docker rm -f "${KEYCLOAK_CONTAINER}" >/dev/null 2>&1 || true
if [[ "${status}" -eq 0 ]]; then
rm -rf "${WORK_DIR}"
fi
}
trap cleanup EXIT INT TERM
for command in curl docker python3 awscurl; do
command -v "${command}" >/dev/null || {
echo "missing required command: ${command}" >&2
exit 1
}
done
[[ -x "${RUSTFS_BINARY}" ]] || {
echo "RustFS binary is not executable: ${RUSTFS_BINARY}" >&2
exit 1
}
[[ -f "${REALM_FIXTURE}" ]] || {
echo "Keycloak realm fixture is missing: ${REALM_FIXTURE}" >&2
exit 1
}
free_port() {
python3 - <<'PY'
import socket
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
print(sock.getsockname()[1])
PY
}
KEYCLOAK_PORT="$(free_port)"
RUSTFS_PORT="$(free_port)"
KEYCLOAK_ORIGIN="http://127.0.0.1:${KEYCLOAK_PORT}"
ISSUER="${KEYCLOAK_ORIGIN}/realms/rustfs-ci"
DISCOVERY_URL="${ISSUER}/.well-known/openid-configuration"
RUSTFS_ORIGIN="http://127.0.0.1:${RUSTFS_PORT}"
docker run --detach --rm \
--name "${KEYCLOAK_CONTAINER}" \
--memory 1g \
--publish "127.0.0.1:${KEYCLOAK_PORT}:8080" \
--env KC_BOOTSTRAP_ADMIN_USERNAME=admin \
--env KC_BOOTSTRAP_ADMIN_PASSWORD=admin \
--env KC_HEALTH_ENABLED=true \
--env "KC_HOSTNAME=${KEYCLOAK_ORIGIN}" \
--volume "${REALM_FIXTURE}:/opt/keycloak/data/import/rustfs-ci-realm.json:ro" \
"${KEYCLOAK_IMAGE}" start-dev --import-realm >/dev/null
for _ in $(seq 1 120); do
if curl --noproxy '*' -fsS "${DISCOVERY_URL}" >"${WORK_DIR}/discovery.json" 2>/dev/null; then
break
fi
if [[ "$(docker inspect -f '{{.State.Running}}' "${KEYCLOAK_CONTAINER}" 2>/dev/null || true)" != "true" ]]; then
docker logs "${KEYCLOAK_CONTAINER}" >&2 || true
echo "Keycloak exited before discovery became ready" >&2
exit 1
fi
sleep 1
done
curl --noproxy '*' -fsS "${DISCOVERY_URL}" >"${WORK_DIR}/discovery.json"
python3 - "${WORK_DIR}/discovery.json" "${ISSUER}" <<'PY'
import json
import sys
with open(sys.argv[1], encoding="utf-8") as source:
discovery = json.load(source)
expected_issuer = sys.argv[2]
assert discovery.get("issuer") == expected_issuer, discovery
for field in ("token_endpoint", "jwks_uri"):
assert discovery.get(field), discovery
PY
mkdir -p "${WORK_DIR}/rustfs-data"
env \
NO_PROXY=127.0.0.1,localhost \
RUSTFS_ACCESS_KEY=rustfsadmin \
RUSTFS_SECRET_KEY=rustfsadmin \
RUSTFS_OUTBOUND_ALLOW_ORIGINS="${KEYCLOAK_ORIGIN}" \
RUSTFS_IDENTITY_OPENID_ENABLE=on \
RUSTFS_IDENTITY_OPENID_CONFIG_URL="${DISCOVERY_URL}" \
RUSTFS_IDENTITY_OPENID_ISSUER="${ISSUER}" \
RUSTFS_IDENTITY_OPENID_CLIENT_ID=rustfs-ci \
RUSTFS_IDENTITY_OPENID_CLIENT_SECRET=rustfs-ci-secret \
RUSTFS_IDENTITY_OPENID_SCOPES=openid,profile,email \
RUSTFS_IDENTITY_OPENID_ROLE_POLICY=consoleAdmin \
"${RUSTFS_BINARY}" --address "127.0.0.1:${RUSTFS_PORT}" "${WORK_DIR}/rustfs-data" \
>"${WORK_DIR}/rustfs.log" 2>&1 &
RUSTFS_PID=$!
for _ in $(seq 1 90); do
if curl --noproxy '*' -fsS "${RUSTFS_ORIGIN}/health/ready" >/dev/null 2>&1; then
break
fi
if ! kill -0 "${RUSTFS_PID}" 2>/dev/null; then
cat "${WORK_DIR}/rustfs.log" >&2
echo "RustFS exited before becoming ready" >&2
exit 1
fi
sleep 1
done
curl --noproxy '*' -fsS "${RUSTFS_ORIGIN}/health/ready" >/dev/null
token_for_client() {
local client_id="$1"
local client_secret="$2"
local response_file="${WORK_DIR}/token-${client_id}.json"
curl --noproxy '*' -fsS "${ISSUER}/protocol/openid-connect/token" \
--data-urlencode grant_type=password \
--data-urlencode "client_id=${client_id}" \
--data-urlencode "client_secret=${client_secret}" \
--data-urlencode username=alice \
--data-urlencode password=alice-password \
--data-urlencode scope=openid \
>"${response_file}"
python3 - "${response_file}" "${ISSUER}" "${client_id}" <<'PY'
import base64
import json
import sys
with open(sys.argv[1], encoding="utf-8") as source:
token = json.load(source)["id_token"]
payload = token.split(".")[1]
payload += "=" * (-len(payload) % 4)
claims = json.loads(base64.urlsafe_b64decode(payload))
assert claims["iss"] == sys.argv[2], claims
audience = claims["aud"]
if isinstance(audience, str):
audience = [audience]
assert sys.argv[3] in audience, claims
print(token)
PY
}
GOOD_TOKEN="$(token_for_client rustfs-ci rustfs-ci-secret)"
GOOD_STATUS="$(curl --noproxy '*' -sS \
-D "${WORK_DIR}/sts-good.headers" \
-o "${WORK_DIR}/sts-good.xml" \
-w '%{http_code}' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-X POST "${RUSTFS_ORIGIN}/" \
--data-urlencode Action=AssumeRoleWithWebIdentity \
--data-urlencode Version=2011-06-15 \
--data-urlencode DurationSeconds=900 \
--data-urlencode "WebIdentityToken=${GOOD_TOKEN}")"
[[ "${GOOD_STATUS}" == 200 ]] || {
cat "${WORK_DIR}/sts-good.xml" >&2
cat "${WORK_DIR}/rustfs.log" >&2
echo "expected valid Keycloak token to return HTTP 200, got ${GOOD_STATUS}" >&2
exit 1
}
grep -Eiq '^content-type: application/xml' "${WORK_DIR}/sts-good.headers"
IFS=$'\t' read -r STS_ACCESS_KEY STS_SECRET_KEY STS_SESSION_TOKEN < <(
python3 - "${WORK_DIR}/sts-good.xml" <<'PY'
import sys
import xml.etree.ElementTree as ET
root = ET.parse(sys.argv[1]).getroot()
values = {}
for element in root.iter():
values[element.tag.rsplit("}", 1)[-1]] = element.text or ""
for field in ("AccessKeyId", "SecretAccessKey", "SessionToken", "Expiration", "SubjectFromWebIdentityToken"):
assert values.get(field), values
print("\t".join(values[field] for field in ("AccessKeyId", "SecretAccessKey", "SessionToken")))
PY
)
awscurl --fail-with-body --service s3 --region us-east-1 \
--access_key "${STS_ACCESS_KEY}" \
--secret_key "${STS_SECRET_KEY}" \
--security_token "${STS_SESSION_TOKEN}" \
"${RUSTFS_ORIGIN}/" >"${WORK_DIR}/list-buckets.xml"
grep -q '<ListAllMyBucketsResult' "${WORK_DIR}/list-buckets.xml"
TAMPERED_TOKEN="$(python3 - "${GOOD_TOKEN}" <<'PY'
import sys
parts = sys.argv[1].split(".")
assert len(parts) == 3 and parts[2]
parts[2] = ("A" if parts[2][0] != "A" else "B") + parts[2][1:]
print(".".join(parts))
PY
)"
TAMPERED_STATUS="$(curl --noproxy '*' -sS \
-o "${WORK_DIR}/sts-tampered.xml" \
-w '%{http_code}' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-X POST "${RUSTFS_ORIGIN}/" \
--data-urlencode Action=AssumeRoleWithWebIdentity \
--data-urlencode Version=2011-06-15 \
--data-urlencode DurationSeconds=900 \
--data-urlencode "WebIdentityToken=${TAMPERED_TOKEN}")"
[[ "${TAMPERED_STATUS}" == 403 ]] || {
cat "${WORK_DIR}/sts-tampered.xml" >&2
echo "expected tampered token to return HTTP 403, got ${TAMPERED_STATUS}" >&2
exit 1
}
grep -q '<Code>AccessDenied</Code>' "${WORK_DIR}/sts-tampered.xml"
BAD_TOKEN="$(token_for_client wrong-audience wrong-audience-secret)"
BAD_STATUS="$(curl --noproxy '*' -sS \
-o "${WORK_DIR}/sts-bad.xml" \
-w '%{http_code}' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-X POST "${RUSTFS_ORIGIN}/" \
--data-urlencode Action=AssumeRoleWithWebIdentity \
--data-urlencode Version=2011-06-15 \
--data-urlencode DurationSeconds=900 \
--data-urlencode "WebIdentityToken=${BAD_TOKEN}")"
[[ "${BAD_STATUS}" == 403 ]] || {
cat "${WORK_DIR}/sts-bad.xml" >&2
echo "expected wrong-audience token to return HTTP 403, got ${BAD_STATUS}" >&2
exit 1
}
grep -q '<Code>AccessDenied</Code>' "${WORK_DIR}/sts-bad.xml"
echo "OIDC Keycloak live gate passed"
+90
View File
@@ -0,0 +1,90 @@
#!/bin/sh
set -eu
SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
REPO_ROOT=$(CDPATH= cd -- "$SCRIPT_DIR/.." && pwd)
TMP_ROOT=$(mktemp -d)
trap 'rm -rf "$TMP_ROOT"' EXIT HUP INT TERM
mkdir -p "$TMP_ROOT/bin"
cat > "$TMP_ROOT/bin/path_containment" <<'EOF'
#!/bin/sh
printf '%s\n' "$@" > "$FAKE_ARGS_FILE"
EOF
chmod +x "$TMP_ROOT/bin/path_containment"
EXPLICIT_ARTIFACTS="$TMP_ROOT/explicit-artifacts"
FAKE_ARGS_FILE="$TMP_ROOT/explicit-args" \
FUZZ_TARGET=path_containment \
FUZZ_SEED=123456789 \
MAX_TOTAL_TIME=7 \
ARTIFACT_ROOT="$EXPLICIT_ARTIFACTS" \
SKIP_BUILD=1 \
USE_PREBUILT_BINARY=1 \
PREBUILT_BINARY_DIR="$TMP_ROOT/bin" \
"$REPO_ROOT/scripts/fuzz/run.sh"
EXPLICIT_MANIFEST="$EXPLICIT_ARTIFACTS/path_containment/run-manifest.txt"
test -f "$EXPLICIT_MANIFEST"
grep -Fx -- '-seed=123456789' "$TMP_ROOT/explicit-args"
grep -Fx 'target=path_containment' "$EXPLICIT_MANIFEST"
grep -Fx 'seed=123456789' "$EXPLICIT_MANIFEST"
grep -Fx 'max_total_time=7' "$EXPLICIT_MANIFEST"
grep -Fx "git_revision=$(git -C "$REPO_ROOT" rev-parse HEAD)" "$EXPLICIT_MANIFEST"
grep -E '^git_dirty=(true|false)$' "$EXPLICIT_MANIFEST"
grep -Fx 'runner_mode=prebuilt' "$EXPLICIT_MANIFEST"
AUTO_ARTIFACTS="$TMP_ROOT/auto-artifacts"
FAKE_ARGS_FILE="$TMP_ROOT/auto-args" \
FUZZ_TARGET=path_containment \
MAX_TOTAL_TIME=1 \
ARTIFACT_ROOT="$AUTO_ARTIFACTS" \
SKIP_BUILD=1 \
USE_PREBUILT_BINARY=1 \
PREBUILT_BINARY_DIR="$TMP_ROOT/bin" \
"$REPO_ROOT/scripts/fuzz/run.sh"
AUTO_MANIFEST="$AUTO_ARTIFACTS/path_containment/run-manifest.txt"
auto_seed=$(sed -n 's/^seed=//p' "$AUTO_MANIFEST")
case "$auto_seed" in
''|*[!0-9]*)
echo "automatic seed was not recorded as an unsigned decimal integer: $auto_seed" >&2
exit 1
;;
esac
grep -Fx -- "-seed=$auto_seed" "$TMP_ROOT/auto-args"
cat > "$TMP_ROOT/bin/cargo" <<'EOF'
#!/bin/sh
printf '%s\n' "$@" > "$FAKE_CARGO_ARGS_FILE"
EOF
chmod +x "$TMP_ROOT/bin/cargo"
CARGO_ARTIFACTS="$TMP_ROOT/cargo-artifacts"
PATH="$TMP_ROOT/bin:$PATH" \
FAKE_CARGO_ARGS_FILE="$TMP_ROOT/cargo-args" \
FUZZ_TARGET=path_containment \
FUZZ_SEED=987654321 \
MAX_TOTAL_TIME=9 \
ARTIFACT_ROOT="$CARGO_ARTIFACTS" \
SKIP_BUILD=1 \
"$REPO_ROOT/scripts/fuzz/run.sh"
CARGO_MANIFEST="$CARGO_ARTIFACTS/path_containment/run-manifest.txt"
grep -Fx '+nightly' "$TMP_ROOT/cargo-args"
grep -Fx 'fuzz' "$TMP_ROOT/cargo-args"
grep -Fx 'run' "$TMP_ROOT/cargo-args"
grep -Fx 'path_containment' "$TMP_ROOT/cargo-args"
grep -Fx -- '-seed=987654321' "$TMP_ROOT/cargo-args"
grep -Fx 'seed=987654321' "$CARGO_MANIFEST"
grep -Fx 'runner_mode=cargo-fuzz' "$CARGO_MANIFEST"
for invalid_seed in 0 0123 not-a-number; do
if FUZZ_TARGET=path_containment FUZZ_SEED="$invalid_seed" SKIP_BUILD=1 "$REPO_ROOT/scripts/fuzz/run.sh" >/dev/null 2>&1; then
echo "invalid FUZZ_SEED was accepted: $invalid_seed" >&2
exit 1
fi
done
echo "fuzz runner seed manifest tests passed"