From ab147f339739323918cb4524404e674762966506 Mon Sep 17 00:00:00 2001
From: cxymds
Date: Wed, 11 Nov 2026 11:10:42 +0800
Subject: [PATCH 1/2] test(security): add root AssumeRole regression coverage
---
.github/workflows/security-script-tests.yml | 6 +-
README.md | 29 +-
rustfs-security-test.sh | 50 ++-
scripts/sts_root_regression.py | 277 ++++++++++++++
test_rustfs_security.py | 3 +
test_sts_root_regression.py | 389 ++++++++++++++++++++
6 files changed, 751 insertions(+), 3 deletions(-)
create mode 100644 scripts/sts_root_regression.py
create mode 100644 test_sts_root_regression.py
diff --git a/.github/workflows/security-script-tests.yml b/.github/workflows/security-script-tests.yml
index a210d52d..6117f555 100644
--- a/.github/workflows/security-script-tests.yml
+++ b/.github/workflows/security-script-tests.yml
@@ -5,12 +5,16 @@ on:
paths:
- rustfs-security-test.sh
- test_rustfs_security.py
+ - scripts/sts_root_regression.py
+ - test_sts_root_regression.py
- .github/workflows/security-script-tests.yml
push:
branches: [main]
paths:
- rustfs-security-test.sh
- test_rustfs_security.py
+ - scripts/sts_root_regression.py
+ - test_sts_root_regression.py
- .github/workflows/security-script-tests.yml
permissions:
@@ -27,4 +31,4 @@ jobs:
- name: Validate shell syntax
run: bash -n rustfs-security-test.sh
- name: Test security contracts without a server
- run: python3 -m unittest test_rustfs_security.py
+ run: python3 -B -m unittest test_rustfs_security.py test_sts_root_regression.py
diff --git a/README.md b/README.md
index 15e783f7..bda535c3 100644
--- a/README.md
+++ b/README.md
@@ -243,7 +243,34 @@ The default log filter is `error,rustfs::admin::auth=debug`; set
commit, while console output includes Bash, curl, and AWS CLI versions.
```bash
-python3 -B -m unittest -v test_rustfs_security.py test_sensitive_diagnostics.py
+python3 -B -m unittest -v test_rustfs_security.py test_sts_root_regression.py test_sensitive_diagnostics.py
+```
+
+### Root AssumeRole regression (#9279)
+
+The security suite runs STS-106 through STS-112 for [rustfs/rustfs#9279](https://github.com/rustfs/rustfs/issues/9279) in every selected topology. STS-106 reproduces the report's form-encoded POST with the `aws:amz:us-east-1:s3` signature scope, requires HTTP 200 and complete, unexpired temporary credentials, then signs ListBuckets with those credentials and the session token.
+
+| Case | Credentials / behavior | Required result |
+|---|---|---|
+| STS-106 | Configured root, original Query request | HTTP 200, usable STS credentials |
+| STS-107 | Root secret: 4 bytes (`root`, as reported) | Nonzero startup exit with a credential-length error |
+| STS-108 | Root secret: 7 bytes | Nonzero startup exit with a credential-length error |
+| STS-109 | Root access key: 2 bytes | Nonzero startup exit with a credential-length error |
+| STS-110 | Root access key: 3 bytes, secret: 8 bytes | Startup succeeds, usable STS credentials |
+| STS-111 | Access key `root`, secret: 8 bytes | Startup succeeds, usable STS credentials |
+| STS-112 | Root secret: 4 Unicode characters / 8 UTF-8 bytes | Startup succeeds, usable STS credentials |
+
+Startup cases run the installed candidate binary on the topology's first node in separate loopback processes, with fresh temporary volumes and disposable credentials. They leave the topology's system service and credentials unchanged. The helper requires Python 3 and curl with `--aws-sigv4` support on that node. Keep `scripts/sts_root_regression.py` with the suite when copying the runner.
+
+The assertions target the intended fix: older versions that accept short root credentials fail these startup cases, even if their STS request returns HTTP 200 or 403. If invalid credentials reach readiness, the report records the STS HTTP status and error code (including the reported `500 InternalError`) before failing. Unrelated startup errors, timeouts, malformed responses, and unusable credentials also fail. Only credential byte lengths and structured outcomes are printed; secret keys, session tokens, and raw server logs are never included.
+
+Run the same matrix against a local candidate binary, or check only the signed Query flow against a running endpoint:
+
+```bash
+python3 scripts/sts_root_regression.py --rustfs-bin /absolute/path/to/rustfs
+python3 scripts/sts_root_regression.py --rustfs-bin /absolute/path/to/rustfs --case secret-4
+# Uses RUSTFS_ACCESS_KEY and RUSTFS_SECRET_KEY from the environment.
+python3 scripts/sts_root_regression.py --endpoint http://127.0.0.1:9000
```
## Tier suite artifacts
diff --git a/rustfs-security-test.sh b/rustfs-security-test.sh
index 84beecb9..f231d190 100755
--- a/rustfs-security-test.sh
+++ b/rustfs-security-test.sh
@@ -17,7 +17,7 @@
# RustFS unified IAM / STS / OIDC security test suite (backlog #2024).
#
# Integrates the backlog security case matrix (IAM-102..114, SA-101..104,
-# STS-101..105, OIDC-101..108) with the request flows proven by
+# STS-101..112, OIDC-101..108) with the request flows proven by
# admin_iam_crud_test / sts_query_compat_test / oidc_keycloak_live.sh.
#
# Every case runs in three topologies, matching the other functional suites:
@@ -917,6 +917,46 @@ case_sts_105_revoke() {
esac
}
+# Issue #9279 uses an S3-scoped SigV4 POST, rather than the AWS CLI's STS
+# signer. Also verify every returned field and use the temporary credentials.
+case_sts_106_root_query() {
+ RUSTFS_ACCESS_KEY="${ACCESS_KEY}" RUSTFS_SECRET_KEY="${SECRET_KEY}" \
+ python3 "${SCRIPT_DIR}/scripts/sts_root_regression.py" --endpoint "${ENDPOINT}"
+}
+
+sts_root_startup_case() {
+ # The installed candidate runs on loopback with private temporary data and
+ # credentials. The helper always stops its own process; systemd and the
+ # topology's root credentials/data are never changed for these cases.
+ ssh "${SSH_OPTS[@]}" "${SSH_USER}@${NODES[0]}" \
+ "python3 - --rustfs-bin /usr/bin/rustfs --case $1" \
+ < "${SCRIPT_DIR}/scripts/sts_root_regression.py"
+}
+
+case_sts_107_short_root_secret() {
+ sts_root_startup_case secret-4
+}
+
+case_sts_108_seven_byte_root_secret() {
+ sts_root_startup_case secret-7
+}
+
+case_sts_109_short_root_access_key() {
+ sts_root_startup_case access-2
+}
+
+case_sts_110_minimum_root_credentials() {
+ sts_root_startup_case minimum
+}
+
+case_sts_111_eight_byte_root_secret() {
+ sts_root_startup_case secret-8
+}
+
+case_sts_112_utf8_root_secret() {
+ sts_root_startup_case secret-8-utf8
+}
+
# ==================== OIDC / SSO ====================
case_oidc_102_validate_rejects_bad_config() {
@@ -1093,6 +1133,13 @@ run_security_cases() {
run_test STS-102 "AssumeRole duration clamped to 12h" case_sts_102_duration_clamp
run_test STS-104 "temporary credentials respect identity scope" case_sts_104_temp_boundary
run_test STS-105 "revoke-tokens invalidates temporary credentials" case_sts_105_revoke
+ run_test STS-106 "root signed Query AssumeRole returns usable credentials (#9279)" case_sts_106_root_query
+ run_test STS-107 "startup rejects a 4-byte root secret (#9279)" case_sts_107_short_root_secret
+ run_test STS-108 "startup rejects a 7-byte root secret (#9279)" case_sts_108_seven_byte_root_secret
+ run_test STS-109 "startup rejects a 2-byte root access key (#9279)" case_sts_109_short_root_access_key
+ run_test STS-110 "3-byte access key and 8-byte secret support AssumeRole (#9279)" case_sts_110_minimum_root_credentials
+ run_test STS-111 "root with an 8-byte secret supports AssumeRole (#9279)" case_sts_111_eight_byte_root_secret
+ run_test STS-112 "root secret minimum counts UTF-8 bytes (#9279)" case_sts_112_utf8_root_secret
run_test OIDC-102 "OIDC validate rejects bad provider config" case_oidc_102_validate_rejects_bad_config
run_test OIDC-108 "STS rejects expired/garbage web identity token" case_oidc_108_reject_garbage_jwt
if [ "${OIDC_LIVE}" -eq 1 ] && [ "${OIDC_RUN}" -eq 0 ]; then
@@ -1109,6 +1156,7 @@ run_security_cases() {
main() {
parse_args "$@"
+ [ -f "${SCRIPT_DIR}/scripts/sts_root_regression.py" ] || die "missing scripts/sts_root_regression.py"
need_cmd ssh "openssh client"
need_cmd curl "curl"
need_cmd openssl "openssl"
diff --git a/scripts/sts_root_regression.py b/scripts/sts_root_regression.py
new file mode 100644
index 00000000..d549737d
--- /dev/null
+++ b/scripts/sts_root_regression.py
@@ -0,0 +1,277 @@
+#!/usr/bin/env python3
+"""Issue #9279: root credential startup boundaries and signed STS Query requests."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import xml.etree.ElementTree as ET
+
+
+ASSUME_ROLE_BODY = (
+ "Action=AssumeRole&Version=2011-06-15&"
+ "RoleArn=arn:aws:iam::000000000000:role/user&"
+ "RoleSessionName=repro&DurationSeconds=3600"
+)
+# Public, disposable fixtures; never use these credentials for a real deployment.
+# Lengths are UTF-8 bytes, matching Rust's str::len().
+CASES = {
+ "minimum": ("STS-110", "abc", "rootroot", None),
+ "secret-8": ("STS-111", "root", "rootroot", None),
+ "secret-8-utf8": ("STS-112", "root", "éééé", None),
+ "secret-4": ("STS-107", "root", "root", "secret"),
+ "secret-7": ("STS-108", "root", "root123", "secret"),
+ "access-2": ("STS-109", "ab", "rootroot", "access"),
+}
+
+
+class RegressionFailure(Exception):
+ """A bounded diagnostic that contains no response bodies or credentials."""
+
+
+def require(condition: bool, message: str) -> None:
+ if not condition:
+ raise RegressionFailure(message)
+
+
+def config_quote(value: str) -> str:
+ return '"' + value.replace("\\", "\\\\").replace('"', '\\"').replace(
+ "\n", "\\n").replace("\r", "\\r").replace("\t", "\\t") + '"'
+
+
+def http(endpoint: str, access: str = "", secret: str = "", token: str = "",
+ *, path: str = "/", body: str | None = None, timeout: float = 10) -> tuple[int, bytes]:
+ # curl's signer reproduces the report's aws:amz:us-east-1:s3 scope. Keep
+ # credentials in stdin, not argv, and never dump its raw response on failure.
+ config = ""
+ if access:
+ config = 'aws-sigv4 = "aws:amz:us-east-1:s3"\n'
+ config += "user = " + config_quote(access + ":" + secret) + "\n"
+ if token:
+ config += "header = " + config_quote("x-amz-security-token: " + token) + "\n"
+ with tempfile.TemporaryDirectory(prefix="rustfs-sts-http-") as directory:
+ output = Path(directory) / "response"
+ command = ["curl", "--disable", "--silent", "--show-error", "--noproxy", "*",
+ "--max-time", str(timeout), "--config", "-", "--output", str(output),
+ "--write-out", "%{http_code}"]
+ if body is not None:
+ command += ["--request", "POST", "--header", "Content-Type: application/x-www-form-urlencoded",
+ "--data-raw", body]
+ command += [endpoint.rstrip("/") + path]
+ try:
+ completed = subprocess.run(command, input=config, capture_output=True, text=True,
+ timeout=timeout + 5)
+ except (OSError, subprocess.TimeoutExpired) as exc:
+ raise RegressionFailure("curl could not complete the request") from exc
+ require(completed.returncode == 0, f"curl transport failed (exit {completed.returncode})")
+ require(bool(re.fullmatch(r"[1-5][0-9]{2}", completed.stdout)), "curl did not return an HTTP status")
+ require(output.is_file() and output.stat().st_size <= 2 * 1024 * 1024,
+ "HTTP response missing or larger than 2 MiB")
+ return int(completed.stdout), output.read_bytes()
+
+
+def xml(body: bytes) -> ET.Element:
+ try:
+ return ET.fromstring(body)
+ except ET.ParseError as exc:
+ raise RegressionFailure("response is not valid XML") from exc
+
+
+def tag(element: ET.Element) -> str:
+ return element.tag.rsplit("}", 1)[-1]
+
+
+def sts_response(status: int, body: bytes, observation: dict) -> ET.Element:
+ observation["sts_http_status"] = status
+ root = xml(body)
+ code = root.findtext(".//{*}Code", default="")
+ # Only copy structured, bounded identifiers, never untrusted message text.
+ if re.fullmatch(r"[A-Za-z][A-Za-z0-9]{0,63}", code):
+ observation["sts_error_code"] = code
+ request_id = root.findtext(".//{*}RequestId", default="")
+ if re.fullmatch(r"[A-Za-z0-9-]{1,128}", request_id):
+ observation["request_id"] = request_id
+ return root
+
+
+def check_assume_role(endpoint: str, access: str, secret: str, observation: dict) -> None:
+ status, body = http(endpoint, access, secret, body=ASSUME_ROLE_BODY)
+ root = sts_response(status, body, observation)
+ require(status == 200, f"AssumeRole expected HTTP 200, received {status}")
+ require(tag(root) == "AssumeRoleResponse" and root.find(".//{*}Error") is None,
+ "AssumeRole did not return a success envelope")
+ credentials = root.find("{*}AssumeRoleResult/{*}Credentials")
+ require(credentials is not None, "AssumeRole omitted Credentials")
+ fields = {name: credentials.findtext("{*}" + name, default="").strip()
+ for name in ("AccessKeyId", "SecretAccessKey", "SessionToken", "Expiration")}
+ require(all(fields.values()), "AssumeRole omitted a credential field or Expiration")
+ require(fields["AccessKeyId"] != access, "AssumeRole returned the parent access key")
+ try:
+ expiration = datetime.fromisoformat(fields["Expiration"].replace("Z", "+00:00"))
+ require(expiration.tzinfo is not None, "STS Expiration omitted its timezone")
+ remaining = (expiration - datetime.now(timezone.utc)).total_seconds()
+ except ValueError as exc:
+ raise RegressionFailure("STS Expiration is not an ISO-8601 timestamp") from exc
+ require(0 < remaining <= 3660, "STS Expiration is outside the requested 3600 second lifetime")
+ observation["session_lifetime_seconds"] = round(remaining)
+ status, body = http(endpoint, fields["AccessKeyId"], fields["SecretAccessKey"], fields["SessionToken"])
+ observation["s3_http_status"] = status
+ require(status == 200, f"temporary credentials could not ListBuckets (HTTP {status})")
+ require(tag(xml(body)) == "ListAllMyBucketsResult", "ListBuckets did not return an S3 success envelope")
+
+
+def credential_rejection(log: str, field: str) -> bool:
+ # An unrelated bind/configuration error or a crash must not pass a negative
+ # startup case. Accept spelling differences across the CLI and config layer.
+ key = re.compile(rf"{field}[_ -]?key", re.IGNORECASE)
+ invalid = re.compile(r"invalid|error|too short|at least|minimum", re.IGNORECASE)
+ length = re.compile(r"length|bytes?|too short|at least|minimum", re.IGNORECASE)
+ return any(key.search(line) and invalid.search(line) and length.search(line)
+ and not re.search(r"\bwarn(?:ing)?\b", line, re.IGNORECASE) for line in log.splitlines())
+
+
+def stop_process(process: subprocess.Popen) -> None:
+ if process.poll() is not None:
+ return
+ try:
+ os.killpg(process.pid, signal.SIGTERM)
+ process.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ os.killpg(process.pid, signal.SIGKILL)
+ process.wait(timeout=5)
+ except ProcessLookupError:
+ process.wait(timeout=5)
+
+
+def check_startup(binary: str, name: str, timeout: float, observation: dict) -> None:
+ case_id, access, secret, invalid_field = CASES[name]
+ observation.update(case_id=case_id, access_key_bytes=len(access.encode()), secret_key_bytes=len(secret.encode()),
+ expected="startup-rejected" if invalid_field else "usable-sts-session")
+ directory = Path(tempfile.mkdtemp(prefix="rustfs-sts-root-")).resolve()
+ process = None
+ try:
+ (directory / "data").mkdir()
+ with socket.socket() as listener:
+ listener.bind(("127.0.0.1", 0))
+ port = listener.getsockname()[1]
+ endpoint = f"http://127.0.0.1:{port}"
+ # Do not inherit credentials, credential files, volumes, TLS, KMS, or
+ # other cluster settings from the caller's environment.
+ environment = {
+ "PATH": os.environ.get("PATH", os.defpath),
+ "RUSTFS_ACCESS_KEY": access, "RUSTFS_SECRET_KEY": secret,
+ "RUSTFS_VOLUMES": str(directory / "data"), "RUSTFS_ADDRESS": f"127.0.0.1:{port}",
+ "RUSTFS_CONSOLE_ENABLE": "false", "RUSTFS_CONSOLE_ADDRESS": "127.0.0.1:0",
+ "RUSTFS_OBS_LOG_DIRECTORY": str(directory / "logs"), "RUSTFS_OBS_LOGGER_LEVEL": "warn",
+ }
+ log_path = directory / "server.log"
+ with log_path.open("wb") as log:
+ process = subprocess.Popen([binary], cwd=directory, env=environment, stdout=log, stderr=log,
+ start_new_session=True)
+ deadline = time.monotonic() + timeout
+ while time.monotonic() < deadline:
+ code = process.poll()
+ if code is not None:
+ observation["server_exit_code"] = code
+ require(invalid_field is not None, "RustFS exited before the valid-credential instance was ready")
+ require(0 < code < 128, "invalid credentials did not cause a normal nonzero startup exit")
+ with log_path.open("rb") as log:
+ diagnostic = log.read(256 * 1024).decode(errors="replace")
+ require(credential_rejection(diagnostic, invalid_field),
+ "startup failed without the expected credential-length diagnostic")
+ observation["credential_length_rejected"] = True
+ return
+ try:
+ status, _ = http(endpoint, path="/health/ready", timeout=1)
+ except RegressionFailure:
+ status = 0
+ observation["readiness_http_status"] = status
+ if status == 200 and process.poll() is None:
+ observation["server_ready"] = True
+ if invalid_field:
+ # Preserve the distinguishing 1.0.5 HTTP 500/InternalError
+ # evidence, but even HTTP 403 cannot excuse invalid startup.
+ try:
+ status, body = http(endpoint, access, secret, body=ASSUME_ROLE_BODY)
+ sts_response(status, body, observation)
+ except RegressionFailure:
+ observation["sts_probe_failed"] = True
+ raise RegressionFailure("RustFS accepted credentials below the startup minimum")
+ check_assume_role(endpoint, access, secret, observation)
+ require(process.poll() is None, "RustFS exited during the STS check")
+ return
+ time.sleep(0.2)
+ raise RegressionFailure("RustFS neither rejected the credentials nor became ready before the deadline")
+ finally:
+ primary_error = sys.exc_info()[1]
+ try:
+ if process is not None:
+ stop_process(process)
+ shutil.rmtree(directory)
+ observation["cleanup"] = "PASS"
+ except (OSError, subprocess.TimeoutExpired) as exc:
+ observation["cleanup"] = "FAIL"
+ observation["cleanup_reason"] = "could not stop the isolated process or remove its private data"
+ if primary_error is None:
+ raise RegressionFailure(observation["cleanup_reason"]) from exc
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ target = parser.add_mutually_exclusive_group(required=True)
+ target.add_argument("--endpoint", help="check the running server using RUSTFS_ACCESS_KEY/RUSTFS_SECRET_KEY")
+ target.add_argument("--rustfs-bin", help="start disposable loopback instances of this candidate binary")
+ parser.add_argument("--case", choices=["all", *CASES], default="all")
+ parser.add_argument("--startup-timeout", type=float, default=60)
+ args = parser.parse_args()
+ if args.startup_timeout <= 0:
+ parser.error("--startup-timeout must be positive")
+ if args.endpoint and args.case != "all":
+ parser.error("--case requires --rustfs-bin")
+ cases = ["configured-root"] if args.endpoint else list(CASES) if args.case == "all" else [args.case]
+ failed = False
+ for name in cases:
+ observation = {"issue": 9279, "case": name}
+ try:
+ if args.endpoint:
+ access = os.environ.get("RUSTFS_ACCESS_KEY", "")
+ secret = os.environ.get("RUSTFS_SECRET_KEY", "")
+ require(bool(access and secret), "RUSTFS_ACCESS_KEY and RUSTFS_SECRET_KEY are required")
+ observation.update(case_id="STS-106", access_key_bytes=len(access.encode()), secret_key_bytes=len(secret.encode()))
+ check_assume_role(args.endpoint, access, secret, observation)
+ else:
+ binary = shutil.which(args.rustfs_bin)
+ require(binary is not None, "candidate RustFS binary is unavailable")
+ check_startup(str(Path(binary).resolve()), name, args.startup_timeout, observation)
+ observation["verdict"] = "PASS"
+ except (RegressionFailure, OSError, subprocess.TimeoutExpired) as exc:
+ observation["verdict"] = "FAIL"
+ observation["reason"] = str(exc) if isinstance(exc, RegressionFailure) else "could not execute or clean up the probe"
+ failed = True
+ print(json.dumps(observation, sort_keys=True), flush=True)
+ return 1 if failed else 0
+
+
+if __name__ == "__main__":
+ # A runner cancellation still tears down the child process and its private
+ # data. No system service, shared volume, or unrelated process is touched.
+ def interrupted(_signum, _frame):
+ raise KeyboardInterrupt
+
+ for interrupt_signal in (signal.SIGTERM, signal.SIGHUP):
+ signal.signal(interrupt_signal, interrupted)
+ try:
+ sys.exit(main())
+ except KeyboardInterrupt:
+ sys.exit(130)
diff --git a/test_rustfs_security.py b/test_rustfs_security.py
index 2cb6500f..d9b07577 100644
--- a/test_rustfs_security.py
+++ b/test_rustfs_security.py
@@ -283,6 +283,9 @@ def test_preview_does_not_require_runtime_dependencies(self):
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertEqual(result.stdout.count("[DRY-RUN] STS-105"), 3)
+ for case_id in range(106, 113):
+ self.assertEqual(result.stdout.count(f"[DRY-RUN] STS-{case_id}"), 3)
+ self.assertEqual(report.read_text().count(f"| STS-{case_id} |"), 3)
for topology in ("single-single", "single-multi", "multi-multi"):
self.assertIn(f"## Topology: {topology}", report.read_text())
self.assertEqual(result.stdout, log_file.read_text())
diff --git a/test_sts_root_regression.py b/test_sts_root_regression.py
new file mode 100644
index 00000000..bd1ee134
--- /dev/null
+++ b/test_sts_root_regression.py
@@ -0,0 +1,389 @@
+#!/usr/bin/env python3
+"""Exercise #9279 oracles, real curl signing, and isolated-process cleanup."""
+
+from contextlib import contextmanager, redirect_stdout
+from datetime import datetime, timedelta, timezone
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import io
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+import xml.etree.ElementTree as ET
+
+from scripts import sts_root_regression as probe
+
+
+STS_NS = "https://sts.amazonaws.com/doc/2011-06-15/"
+CANARY = "PRIVATE_FIXTURE_SECRET_OR_TOKEN"
+LIST_BUCKETS = b''
+
+
+def credentials_xml(**overrides):
+ fields = {
+ "AccessKeyId": "temporary-access",
+ "SecretAccessKey": CANARY,
+ "SessionToken": CANARY,
+ "Expiration": (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat(),
+ **overrides,
+ }
+ root = ET.Element("AssumeRoleResponse", xmlns=STS_NS)
+ credentials = ET.SubElement(ET.SubElement(root, "AssumeRoleResult"), "Credentials")
+ for name, value in fields.items():
+ if value is not None:
+ ET.SubElement(credentials, name).text = value
+ ET.SubElement(ET.SubElement(root, "ResponseMetadata"), "RequestId").text = "fixture-request-id"
+ return ET.tostring(root)
+
+
+def error_xml(code="InternalError"):
+ return (f'Receiver{code}'
+ f'action not allowed {CANARY}repro-request'
+ '').encode()
+
+
+@contextmanager
+def server(sts_status=200, sts_body=None, s3_status=200, s3_body=LIST_BUCKETS):
+ requests = []
+
+ class Handler(BaseHTTPRequestHandler):
+ def respond(self, status, body):
+ data = self.rfile.read(int(self.headers.get("Content-Length", "0")))
+ requests.append((self.command, self.path, dict(self.headers), data))
+ self.send_response(status)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def do_POST(self):
+ self.respond(sts_status, credentials_xml() if sts_body is None else sts_body)
+
+ def do_GET(self):
+ self.respond(s3_status, s3_body)
+
+ def log_message(self, *_args):
+ pass
+
+ httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ worker = threading.Thread(target=lambda: httpd.serve_forever(poll_interval=0.01), daemon=True)
+ worker.start()
+ try:
+ yield f"http://127.0.0.1:{httpd.server_port}", requests
+ finally:
+ httpd.shutdown()
+ httpd.server_close()
+ worker.join()
+
+
+class QueryRequestTest(unittest.TestCase):
+ def assert_signature(self, request, access, secret):
+ method, path, headers, body = request
+ headers = {name.lower(): value for name, value in headers.items()}
+ match = re.fullmatch(r"AWS4-HMAC-SHA256 Credential=([^,]+), SignedHeaders=([^,]+), Signature=([a-f0-9]{64})",
+ headers["authorization"])
+ self.assertIsNotNone(match)
+ credential, signed_headers, actual = match.groups()
+ actual_access, scope = credential.split("/", 1)
+ self.assertEqual(actual_access, access)
+ date, region, service, terminal = scope.split("/")
+ self.assertEqual((region, service, terminal), ("us-east-1", "s3", "aws4_request"))
+ canonical_headers = "".join(f"{name}:{' '.join(headers[name].split())}\n" for name in signed_headers.split(";"))
+ canonical = "\n".join((method, path, "", canonical_headers, signed_headers, hashlib.sha256(body).hexdigest()))
+ to_sign = "\n".join(("AWS4-HMAC-SHA256", headers["x-amz-date"], scope, hashlib.sha256(canonical.encode()).hexdigest()))
+ key = ("AWS4" + secret).encode()
+ for value in (date, region, service, terminal):
+ key = hmac.new(key, value.encode(), hashlib.sha256).digest()
+ self.assertEqual(actual, hmac.new(key, to_sign.encode(), hashlib.sha256).hexdigest())
+
+ def test_original_query_and_temporary_s3_request_have_valid_s3_signatures(self):
+ for secret in ("rootroot", "éééé", 'fixture-"quote\\slash'):
+ with self.subTest(secret=secret), server() as (endpoint, requests):
+ observation = {}
+ # Parent AWS credentials/session tokens must not influence curl.
+ with patch.dict(os.environ, {"AWS_SESSION_TOKEN": "unrelated", "AWS_PROFILE": "unrelated"}):
+ probe.check_assume_role(endpoint, "root", secret, observation)
+ self.assertEqual(len(requests), 2)
+ self.assertEqual(requests[0][:2], ("POST", "/"))
+ self.assertEqual(requests[0][3], probe.ASSUME_ROLE_BODY.encode())
+ self.assertEqual(requests[0][2]["Content-Type"], "application/x-www-form-urlencoded")
+ self.assert_signature(requests[0], "root", secret)
+ self.assertEqual(requests[1][:2], ("GET", "/"))
+ self.assertEqual(requests[1][2]["x-amz-security-token"], CANARY)
+ self.assert_signature(requests[1], "temporary-access", CANARY)
+ self.assertEqual(observation["sts_http_status"], 200)
+ self.assertEqual(observation["s3_http_status"], 200)
+ self.assertNotIn(CANARY, json.dumps(observation))
+
+ def test_500_action_not_allowed_and_403_are_not_success(self):
+ for status, code in ((500, "InternalError"), (403, "AccessDenied")):
+ with self.subTest(status=status), server(status, error_xml(code)) as (endpoint, requests):
+ observation = {}
+ with self.assertRaises(probe.RegressionFailure) as caught:
+ probe.check_assume_role(endpoint, "root", "rootroot", observation)
+ self.assertEqual(observation["sts_http_status"], status)
+ self.assertEqual(observation["sts_error_code"], code)
+ self.assertEqual(observation["request_id"], "repro-request")
+ self.assertEqual(len(requests), 1)
+ self.assertNotIn(CANARY, str(caught.exception) + json.dumps(observation))
+
+ def test_incomplete_credentials_and_error_envelopes_cannot_pass_http_200(self):
+ bodies = [b"not XML", b"", error_xml(), b""]
+ for field in ("AccessKeyId", "SecretAccessKey", "SessionToken", "Expiration"):
+ bodies += [credentials_xml(**{field: None}), credentials_xml(**{field: " "})]
+ bodies += [credentials_xml(AccessKeyId="root")]
+ for body in bodies:
+ with self.subTest(body=body[:60]), server(sts_body=body) as (endpoint, requests):
+ with self.assertRaises(probe.RegressionFailure):
+ probe.check_assume_role(endpoint, "root", "rootroot", {})
+ self.assertEqual(len(requests), 1)
+
+ def test_expiration_must_be_valid_future_and_bounded(self):
+ now = datetime.now(timezone.utc)
+ for expiration in ("not-a-date", now.replace(tzinfo=None).isoformat(),
+ (now - timedelta(seconds=1)).isoformat(), (now + timedelta(hours=2)).isoformat()):
+ with self.subTest(expiration=expiration), server(sts_body=credentials_xml(Expiration=expiration)) as (endpoint, requests):
+ with self.assertRaises(probe.RegressionFailure):
+ probe.check_assume_role(endpoint, "root", "rootroot", {})
+ self.assertEqual(len(requests), 1)
+
+ def test_unusable_session_and_non_s3_success_bodies_fail(self):
+ for status, body in ((403, error_xml("AccessDenied")), (500, error_xml()),
+ (200, b""), (200, b"invalid XML")):
+ with self.subTest(status=status, body=body), server(s3_status=status, s3_body=body) as (endpoint, requests):
+ with self.assertRaises(probe.RegressionFailure):
+ probe.check_assume_role(endpoint, "root", "rootroot", {})
+ self.assertEqual(len(requests), 2)
+
+ def test_transport_failure_is_not_an_http_verdict(self):
+ with server() as (endpoint, _):
+ pass
+ with self.assertRaisesRegex(probe.RegressionFailure, "transport failed"):
+ probe.check_assume_role(endpoint, "root", "rootroot", {})
+
+ def test_cli_failure_keeps_status_without_disclosing_response(self):
+ with server(500, error_xml()) as (endpoint, _), io.StringIO() as output:
+ with patch.object(sys, "argv", ["probe", "--endpoint", endpoint]), redirect_stdout(output), \
+ patch.dict(os.environ, {"RUSTFS_ACCESS_KEY": "root", "RUSTFS_SECRET_KEY": CANARY}):
+ self.assertEqual(probe.main(), 1)
+ result = json.loads(output.getvalue())
+ self.assertEqual(result["verdict"], "FAIL")
+ self.assertEqual(result["sts_http_status"], 500)
+ self.assertNotIn(CANARY, output.getvalue())
+
+
+class StartupTest(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+ self.receipt = self.root / "receipt.json"
+
+ def binary(self, mode, diagnostic=""):
+ binary = self.root / "fake-rustfs"
+ # Real subprocesses/listeners exercise cleanup and readiness. Their
+ # HTTP results deliberately include success and the historical defect.
+ binary.write_text(f'''#!{sys.executable}
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+import os
+from pathlib import Path
+import signal
+import sys
+import time
+Path({str(self.receipt)!r}).write_text(json.dumps({{"pid": os.getpid(), "cwd": os.getcwd(), "env": dict(os.environ)}}))
+mode = {mode!r}
+if mode in ("exit-0", "exit-1", "crash"):
+ print({diagnostic!r}, flush=True)
+ if mode == "crash":
+ os.kill(os.getpid(), signal.SIGKILL)
+ sys.exit(0 if mode == "exit-0" else 1)
+if mode == "timeout":
+ time.sleep(600)
+class Handler(BaseHTTPRequestHandler):
+ def reply(self, status, body):
+ self.send_response(status)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+ def do_GET(self):
+ if self.path == "/health/ready":
+ self.reply(200, b'{{"ready":true}}')
+ elif self.path == "/":
+ self.reply(200, {LIST_BUCKETS!r})
+ else:
+ self.reply(403, {error_xml("AccessDenied")!r})
+ def do_POST(self):
+ self.rfile.read(int(self.headers.get("Content-Length", "0")))
+ if mode == "sts-500":
+ self.reply(500, {error_xml()!r})
+ elif mode == "sts-403":
+ self.reply(403, {error_xml("AccessDenied")!r})
+ else:
+ self.reply(200, {credentials_xml()!r})
+ def log_message(self, *args):
+ pass
+host, port = os.environ["RUSTFS_ADDRESS"].split(":")
+HTTPServer((host, int(port)), Handler).serve_forever()
+''')
+ binary.chmod(0o755)
+ return str(binary)
+
+ def assert_cleaned(self):
+ receipt = json.loads(self.receipt.read_text())
+ self.assertFalse(Path(receipt["cwd"]).exists(), receipt)
+ with self.assertRaises(ProcessLookupError):
+ os.kill(receipt["pid"], 0)
+ return receipt
+
+ def test_exact_short_secret_and_access_boundaries_require_length_errors(self):
+ for case, field, access_bytes, secret_bytes in (("secret-4", "secret", 4, 4),
+ ("secret-7", "secret", 4, 7),
+ ("access-2", "access", 2, 8)):
+ with self.subTest(case=case):
+ observation = {}
+ probe.check_startup(self.binary("exit-1", f"Error: invalid {field} key length"), case, 3, observation)
+ self.assertTrue(observation["credential_length_rejected"])
+ self.assertEqual((observation["access_key_bytes"], observation["secret_key_bytes"]),
+ (access_bytes, secret_bytes))
+ self.assertEqual(observation["cleanup"], "PASS")
+ self.assert_cleaned()
+
+ def test_unrelated_startup_errors_clean_exit_crashes_and_timeouts_fail(self):
+ scenarios = [("exit-1", "Error: Address already in use"),
+ ("exit-1", "WARNING: secret key minimum length is 8 bytes\nError: permission denied"),
+ ("exit-1", "Error: invalid access key length"),
+ ("exit-0", "Error: invalid secret key length"),
+ ("crash", "Error: invalid secret key length"), ("timeout", "")]
+ for mode, diagnostic in scenarios:
+ with self.subTest(mode=mode, diagnostic=diagnostic):
+ observation = {}
+ with self.assertRaises(probe.RegressionFailure):
+ probe.check_startup(self.binary(mode, diagnostic), "secret-4", 0.7, observation)
+ self.assertEqual(observation["cleanup"], "PASS")
+ self.assert_cleaned()
+
+ def test_accepting_short_credentials_fails_even_with_200_or_403_and_preserves_500(self):
+ for mode, status in (("success", 200), ("sts-403", 403), ("sts-500", 500)):
+ with self.subTest(mode=mode):
+ observation = {}
+ with self.assertRaisesRegex(probe.RegressionFailure, "accepted credentials below"):
+ probe.check_startup(self.binary(mode), "secret-4", 3, observation)
+ self.assertEqual(observation["sts_http_status"], status)
+ self.assertEqual(observation["cleanup"], "PASS")
+ if status == 500:
+ self.assertEqual(observation["sts_error_code"], "InternalError")
+ self.assertNotIn(CANARY, json.dumps(observation))
+ self.assert_cleaned()
+
+ def test_valid_and_utf8_byte_boundaries_require_usable_sessions_and_isolated_settings(self):
+ for case, access_bytes in (("minimum", 3), ("secret-8", 4), ("secret-8-utf8", 4)):
+ with self.subTest(case=case):
+ observation = {}
+ with patch.dict(os.environ, {"RUSTFS_VOLUMES": "/shared/do-not-touch", "RUSTFS_SECRET_KEY_FILE": "/private",
+ "RUSTFS_KMS_ENABLE": "true", "AWS_SESSION_TOKEN": CANARY}):
+ probe.check_startup(self.binary("success"), case, 3, observation)
+ self.assertEqual((observation["access_key_bytes"], observation["secret_key_bytes"]), (access_bytes, 8))
+ self.assertEqual(observation["s3_http_status"], 200)
+ receipt = self.assert_cleaned()
+ environment = receipt["env"]
+ self.assertEqual(environment["RUSTFS_VOLUMES"], str(Path(receipt["cwd"]) / "data"))
+ self.assertTrue(environment["RUSTFS_ADDRESS"].startswith("127.0.0.1:"))
+ for name in ("RUSTFS_SECRET_KEY_FILE", "RUSTFS_KMS_ENABLE", "AWS_SESSION_TOKEN"):
+ self.assertNotIn(name, environment)
+
+ def test_valid_credentials_do_not_hide_sts_500(self):
+ observation = {}
+ with self.assertRaisesRegex(probe.RegressionFailure, "expected HTTP 200, received 500"):
+ probe.check_startup(self.binary("sts-500"), "secret-8", 3, observation)
+ self.assertEqual(observation["sts_error_code"], "InternalError")
+ self.assert_cleaned()
+
+ def test_cancellation_and_ssh_hangup_clean_the_child_and_private_data(self):
+ for interrupt in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
+ with self.subTest(signal=interrupt):
+ self.receipt.unlink(missing_ok=True)
+ runner = subprocess.Popen([sys.executable, str(Path(probe.__file__).resolve()), "--rustfs-bin", self.binary("timeout"),
+ "--case", "secret-4"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ deadline = time.monotonic() + 5
+ receipt = None
+ while time.monotonic() < deadline:
+ try:
+ receipt = json.loads(self.receipt.read_text())
+ break
+ except (FileNotFoundError, json.JSONDecodeError):
+ time.sleep(0.02)
+ self.assertIsNotNone(receipt, "child did not start")
+ runner.send_signal(interrupt)
+ stdout, stderr = runner.communicate(timeout=10)
+ self.assertEqual(runner.returncode, 130, stdout + stderr)
+ self.assertNotIn("PASS", stdout)
+ self.assert_cleaned()
+ finally:
+ if runner.poll() is None:
+ runner.kill()
+ runner.communicate(timeout=5)
+
+ def test_cleanup_failure_cannot_pass_or_replace_the_sts_failure(self):
+ original_stop = probe.stop_process
+
+ def failed_cleanup(process):
+ original_stop(process)
+ raise OSError("fixture cleanup failure")
+
+ for mode, reason in (("success", "could not stop"), ("sts-500", "expected HTTP 200, received 500")):
+ with self.subTest(mode=mode):
+ observation = {}
+ try:
+ with patch.object(probe, "stop_process", side_effect=failed_cleanup):
+ with self.assertRaisesRegex(probe.RegressionFailure, reason):
+ probe.check_startup(self.binary(mode), "secret-8", 3, observation)
+ self.assertEqual(observation["cleanup"], "FAIL")
+ if mode == "sts-500":
+ self.assertEqual(observation["sts_error_code"], "InternalError")
+ finally:
+ receipt = json.loads(self.receipt.read_text())
+ shutil.rmtree(receipt["cwd"])
+ self.assert_cleaned()
+
+ def test_remote_stdin_entrypoint_runs_the_candidate_and_returns_its_verdict(self):
+ result = subprocess.run([sys.executable, "-", "--rustfs-bin", self.binary("success"), "--case", "minimum"],
+ input=Path(probe.__file__).read_text(), capture_output=True, text=True, timeout=10)
+ self.assertEqual(result.returncode, 0, result.stderr)
+ observation = json.loads(result.stdout)
+ self.assertEqual(observation["case_id"], "STS-110")
+ self.assertEqual(observation["verdict"], "PASS")
+ self.assertEqual(observation["cleanup"], "PASS")
+ self.assert_cleaned()
+
+ def test_matrix_retains_a_failed_case_and_continues_remaining_cases(self):
+ calls = []
+
+ def startup(_binary, name, _timeout, observation):
+ calls.append(name)
+ observation["cleanup"] = "PASS"
+ if name == "secret-4":
+ raise probe.RegressionFailure("fixture regression")
+
+ with patch.object(sys, "argv", ["probe", "--rustfs-bin", self.binary("success")]), \
+ patch.object(probe, "check_startup", side_effect=startup), io.StringIO() as output, redirect_stdout(output):
+ self.assertEqual(probe.main(), 1)
+ rows = [json.loads(line) for line in output.getvalue().splitlines()]
+ self.assertEqual(calls, list(probe.CASES))
+ self.assertEqual(len(rows), 6)
+ self.assertEqual([row["case"] for row in rows if row["verdict"] == "FAIL"], ["secret-4"])
+
+
+if __name__ == "__main__":
+ unittest.main()
From d700c2987eb27dbdc0e5be94f91a79e53582ffd0 Mon Sep 17 00:00:00 2001
From: cxymds
Date: Wed, 11 Nov 2026 11:32:11 +0800
Subject: [PATCH 2/2] fix(security): harden STS probe cleanup and startup
checks
---
README.md | 2 +
rustfs-security-test.sh | 8 +-
scripts/sts_root_regression.py | 126 ++++++++++++++++++++--
test_sts_root_regression.py | 188 ++++++++++++++++++++++++++++++++-
4 files changed, 307 insertions(+), 17 deletions(-)
diff --git a/README.md b/README.md
index bda535c3..e4cb49ea 100644
--- a/README.md
+++ b/README.md
@@ -262,6 +262,8 @@ The security suite runs STS-106 through STS-112 for [rustfs/rustfs#9279](https:/
Startup cases run the installed candidate binary on the topology's first node in separate loopback processes, with fresh temporary volumes and disposable credentials. They leave the topology's system service and credentials unchanged. The helper requires Python 3 and curl with `--aws-sigv4` support on that node. Keep `scripts/sts_root_regression.py` with the suite when copying the runner.
+The SSH runner keeps a heartbeat channel open after sending the helper source. Channel EOF or five seconds without a heartbeat requests remote cancellation without depending on a PTY or SIGHUP. Cancellation is checked between bounded requests; teardown completes its wait, forced termination if necessary, and private-directory removal before returning exit status 130. Repeated cancellation cannot interrupt teardown.
+
The assertions target the intended fix: older versions that accept short root credentials fail these startup cases, even if their STS request returns HTTP 200 or 403. If invalid credentials reach readiness, the report records the STS HTTP status and error code (including the reported `500 InternalError`) before failing. Unrelated startup errors, timeouts, malformed responses, and unusable credentials also fail. Only credential byte lengths and structured outcomes are printed; secret keys, session tokens, and raw server logs are never included.
Run the same matrix against a local candidate binary, or check only the signed Query flow against a running endpoint:
diff --git a/rustfs-security-test.sh b/rustfs-security-test.sh
index f231d190..51b455a3 100755
--- a/rustfs-security-test.sh
+++ b/rustfs-security-test.sh
@@ -928,9 +928,11 @@ sts_root_startup_case() {
# The installed candidate runs on loopback with private temporary data and
# credentials. The helper always stops its own process; systemd and the
# topology's root credentials/data are never changed for these cases.
- ssh "${SSH_OPTS[@]}" "${SSH_USER}@${NODES[0]}" \
- "python3 - --rustfs-bin /usr/bin/rustfs --case $1" \
- < "${SCRIPT_DIR}/scripts/sts_root_regression.py"
+ local option options=()
+ for option in "${SSH_OPTS[@]}"; do options+=("--ssh-option=${option}"); done
+ python3 "${SCRIPT_DIR}/scripts/sts_root_regression.py" \
+ --ssh-target "${SSH_USER}@${NODES[0]}" "${options[@]}" \
+ --rustfs-bin /usr/bin/rustfs --case "$1"
}
case_sts_107_short_root_secret() {
diff --git a/scripts/sts_root_regression.py b/scripts/sts_root_regression.py
index d549737d..8a8def2c 100644
--- a/scripts/sts_root_regression.py
+++ b/scripts/sts_root_regression.py
@@ -9,12 +9,15 @@
import os
from pathlib import Path
import re
+import select
+import shlex
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
+import threading
import time
import xml.etree.ElementTree as ET
@@ -34,6 +37,22 @@
"secret-7": ("STS-108", "root", "root123", "secret"),
"access-2": ("STS-109", "ab", "rootroot", "access"),
}
+SSH_HEARTBEAT_TIMEOUT = 5
+_cancel_requested = False
+
+# Read exactly the source frame; leave subsequent heartbeat bytes on fd 0.
+# Python's normal `-` mode consumes stdin through EOF before executing it.
+REMOTE_BOOTSTRAP = """import os, sys
+remaining = int(sys.argv.pop(1))
+chunks = []
+while remaining:
+ chunk = os.read(0, min(remaining, 65536))
+ if not chunk:
+ raise SystemExit(1)
+ chunks.append(chunk)
+ remaining -= len(chunk)
+exec(compile(b''.join(chunks), '', 'exec'))
+"""
class RegressionFailure(Exception):
@@ -45,6 +64,35 @@ def require(condition: bool, message: str) -> None:
raise RegressionFailure(message)
+def request_cancellation(*_args) -> None:
+ # Signal handlers only record cancellation. Raising asynchronously can
+ # interrupt child registration or the cleanup's wait/SIGKILL/rmtree path.
+ global _cancel_requested
+ _cancel_requested = True
+
+
+def check_cancellation() -> None:
+ if _cancel_requested:
+ raise KeyboardInterrupt
+
+
+def watch_connection() -> None:
+ deadline = time.monotonic() + SSH_HEARTBEAT_TIMEOUT
+ try:
+ while not _cancel_requested:
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ break
+ ready, _, _ = select.select([0], [], [], min(remaining, 0.2))
+ if ready:
+ if not os.read(0, 4096):
+ break
+ deadline = time.monotonic() + SSH_HEARTBEAT_TIMEOUT
+ except OSError:
+ pass
+ request_cancellation()
+
+
def config_quote(value: str) -> str:
return '"' + value.replace("\\", "\\\\").replace('"', '\\"').replace(
"\n", "\\n").replace("\r", "\\r").replace("\t", "\\t") + '"'
@@ -52,6 +100,7 @@ def config_quote(value: str) -> str:
def http(endpoint: str, access: str = "", secret: str = "", token: str = "",
*, path: str = "/", body: str | None = None, timeout: float = 10) -> tuple[int, bytes]:
+ check_cancellation()
# curl's signer reproduces the report's aws:amz:us-east-1:s3 scope. Keep
# credentials in stdin, not argv, and never dump its raw response on failure.
config = ""
@@ -74,6 +123,7 @@ def http(endpoint: str, access: str = "", secret: str = "", token: str = "",
timeout=timeout + 5)
except (OSError, subprocess.TimeoutExpired) as exc:
raise RegressionFailure("curl could not complete the request") from exc
+ check_cancellation()
require(completed.returncode == 0, f"curl transport failed (exit {completed.returncode})")
require(bool(re.fullmatch(r"[1-5][0-9]{2}", completed.stdout)), "curl did not return an HTTP status")
require(output.is_file() and output.stat().st_size <= 2 * 1024 * 1024,
@@ -134,11 +184,15 @@ def check_assume_role(endpoint: str, access: str, secret: str, observation: dict
def credential_rejection(log: str, field: str) -> bool:
# An unrelated bind/configuration error or a crash must not pass a negative
# startup case. Accept spelling differences across the CLI and config layer.
- key = re.compile(rf"{field}[_ -]?key", re.IGNORECASE)
- invalid = re.compile(r"invalid|error|too short|at least|minimum", re.IGNORECASE)
+ key = rf"(?:root[ _-]*|RUSTFS_)?{field}[_ -]?key"
+ invalid = re.compile(
+ rf"\binvalid[ _-]*{key}(?:[ _-]*length)?\b"
+ rf"|\b{key}(?:\s+length)?(?:\s+is)?\s*[:=-]?\s*(?:invalid|rejected|too short)\b"
+ rf"|\b(?:error|fatal)\s*:\s*[\"']?{key}\b[^\n]*\b(?:at least|minimum)\b", re.IGNORECASE)
length = re.compile(r"length|bytes?|too short|at least|minimum", re.IGNORECASE)
- return any(key.search(line) and invalid.search(line) and length.search(line)
- and not re.search(r"\bwarn(?:ing)?\b", line, re.IGNORECASE) for line in log.splitlines())
+ return any(invalid.search(line) and length.search(line)
+ and not re.search(r"\b(?:info|debug|trace|warn(?:ing)?)\b", line, re.IGNORECASE)
+ for line in log.splitlines())
def stop_process(process: subprocess.Popen) -> None:
@@ -154,7 +208,42 @@ def stop_process(process: subprocess.Popen) -> None:
process.wait(timeout=5)
+def run_ssh_probe(target: str, options: list[str], binary: str, case: str, timeout: float) -> int:
+ source = Path(__file__).read_bytes()
+ remote = shlex.join(["python3", "-c", REMOTE_BOOTSTRAP, str(len(source)), "--watch-stdin",
+ "--rustfs-bin", binary, "--case", case, "--startup-timeout", str(timeout)])
+ command = ["ssh", "-T", "-o", "ConnectTimeout=10", "-o", "ServerAliveInterval=2",
+ "-o", "ServerAliveCountMax=3", *options, "--", target, remote]
+ check_cancellation()
+ transport = subprocess.Popen(command, stdin=subprocess.PIPE, start_new_session=True)
+ try:
+ transport.stdin.write(source)
+ transport.stdin.flush()
+ while transport.poll() is None:
+ check_cancellation()
+ try:
+ transport.stdin.write(b"\n")
+ transport.stdin.flush()
+ except BrokenPipeError:
+ break
+ time.sleep(0.2)
+ return transport.wait(timeout=30)
+ finally:
+ # Closing stdin triggers the remote watchdog even without a PTY or
+ # SIGHUP. Give it time to finish an in-flight request and teardown.
+ try:
+ transport.stdin.close()
+ except BrokenPipeError:
+ pass
+ try:
+ transport.wait(timeout=30)
+ except subprocess.TimeoutExpired:
+ stop_process(transport)
+ check_cancellation()
+
+
def check_startup(binary: str, name: str, timeout: float, observation: dict) -> None:
+ check_cancellation()
case_id, access, secret, invalid_field = CASES[name]
observation.update(case_id=case_id, access_key_bytes=len(access.encode()), secret_key_bytes=len(secret.encode()),
expected="startup-rejected" if invalid_field else "usable-sts-session")
@@ -177,10 +266,11 @@ def check_startup(binary: str, name: str, timeout: float, observation: dict) ->
}
log_path = directory / "server.log"
with log_path.open("wb") as log:
- process = subprocess.Popen([binary], cwd=directory, env=environment, stdout=log, stderr=log,
+ process = subprocess.Popen([binary], cwd=directory, env=environment, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
start_new_session=True)
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
+ check_cancellation()
code = process.poll()
if code is not None:
observation["server_exit_code"] = code
@@ -225,6 +315,7 @@ def check_startup(binary: str, name: str, timeout: float, observation: dict) ->
observation["cleanup_reason"] = "could not stop the isolated process or remove its private data"
if primary_error is None:
raise RegressionFailure(observation["cleanup_reason"]) from exc
+ check_cancellation()
def main() -> int:
@@ -234,11 +325,28 @@ def main() -> int:
target.add_argument("--rustfs-bin", help="start disposable loopback instances of this candidate binary")
parser.add_argument("--case", choices=["all", *CASES], default="all")
parser.add_argument("--startup-timeout", type=float, default=60)
+ parser.add_argument("--ssh-target", help="run the candidate on this SSH destination with connection supervision")
+ parser.add_argument("--ssh-option", action="append", default=[], help="one SSH argument (use --ssh-option=-o for flags)")
+ parser.add_argument("--watch-stdin", action="store_true", help=argparse.SUPPRESS)
args = parser.parse_args()
if args.startup_timeout <= 0:
parser.error("--startup-timeout must be positive")
if args.endpoint and args.case != "all":
parser.error("--case requires --rustfs-bin")
+ if args.endpoint and (args.ssh_target or args.watch_stdin):
+ parser.error("connection supervision requires --rustfs-bin")
+ if args.ssh_option and not args.ssh_target:
+ parser.error("--ssh-option requires --ssh-target")
+ if args.ssh_target and args.watch_stdin:
+ parser.error("--ssh-target and --watch-stdin are mutually exclusive")
+ if args.ssh_target:
+ try:
+ return run_ssh_probe(args.ssh_target, args.ssh_option, args.rustfs_bin, args.case, args.startup_timeout)
+ except (OSError, subprocess.TimeoutExpired):
+ print("SSH probe transport failed", file=sys.stderr)
+ return 1
+ if args.watch_stdin:
+ threading.Thread(target=watch_connection, daemon=True).start()
cases = ["configured-root"] if args.endpoint else list(CASES) if args.case == "all" else [args.case]
failed = False
for name in cases:
@@ -259,6 +367,7 @@ def main() -> int:
observation["verdict"] = "FAIL"
observation["reason"] = str(exc) if isinstance(exc, RegressionFailure) else "could not execute or clean up the probe"
failed = True
+ check_cancellation()
print(json.dumps(observation, sort_keys=True), flush=True)
return 1 if failed else 0
@@ -266,11 +375,8 @@ def main() -> int:
if __name__ == "__main__":
# A runner cancellation still tears down the child process and its private
# data. No system service, shared volume, or unrelated process is touched.
- def interrupted(_signum, _frame):
- raise KeyboardInterrupt
-
- for interrupt_signal in (signal.SIGTERM, signal.SIGHUP):
- signal.signal(interrupt_signal, interrupted)
+ for interrupt_signal in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
+ signal.signal(interrupt_signal, request_cancellation)
try:
sys.exit(main())
except KeyboardInterrupt:
diff --git a/test_sts_root_regression.py b/test_sts_root_regression.py
index bd1ee134..4cd18fc5 100644
--- a/test_sts_root_regression.py
+++ b/test_sts_root_regression.py
@@ -188,6 +188,7 @@ def setUp(self):
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.receipt = self.root / "receipt.json"
+ self.shutdown_marker = self.root / "shutting-down"
def binary(self, mode, diagnostic=""):
binary = self.root / "fake-rustfs"
@@ -201,15 +202,21 @@ def binary(self, mode, diagnostic=""):
import signal
import sys
import time
-Path({str(self.receipt)!r}).write_text(json.dumps({{"pid": os.getpid(), "cwd": os.getcwd(), "env": dict(os.environ)}}))
mode = {mode!r}
+if mode in ("slow-cleanup", "cancelled-slow-cleanup"):
+ def defer_shutdown(*args):
+ Path({str(self.shutdown_marker)!r}).write_text("SIGTERM received")
+ signal.signal(signal.SIGTERM, defer_shutdown)
+Path({str(self.receipt)!r}).write_text(json.dumps({{"pid": os.getpid(), "cwd": os.getcwd(), "env": dict(os.environ)}}))
if mode in ("exit-0", "exit-1", "crash"):
print({diagnostic!r}, flush=True)
if mode == "crash":
os.kill(os.getpid(), signal.SIGKILL)
sys.exit(0 if mode == "exit-0" else 1)
-if mode == "timeout":
+if mode in ("timeout", "cancelled-slow-cleanup"):
time.sleep(600)
+if mode == "slow-start":
+ time.sleep({probe.SSH_HEARTBEAT_TIMEOUT + 1})
class Handler(BaseHTTPRequestHandler):
def reply(self, status, body):
self.send_response(status)
@@ -241,11 +248,62 @@ def log_message(self, *args):
def assert_cleaned(self):
receipt = json.loads(self.receipt.read_text())
- self.assertFalse(Path(receipt["cwd"]).exists(), receipt)
+ self.assertFalse(Path(receipt["cwd"]).exists(), f"private directory was not removed: {receipt['cwd']}")
with self.assertRaises(ProcessLookupError):
os.kill(receipt["pid"], 0)
return receipt
+ def wait_receipt(self):
+ deadline = time.monotonic() + 5
+ while time.monotonic() < deadline:
+ try:
+ return json.loads(self.receipt.read_text())
+ except (FileNotFoundError, json.JSONDecodeError):
+ time.sleep(0.02)
+ self.fail("child did not start")
+
+ def rescue_fixture(self):
+ # A failing cleanup regression test must not itself leave its fixture
+ # running. Only this test's recorded, isolated process group is owned.
+ if not self.receipt.exists():
+ return
+ receipt = json.loads(self.receipt.read_text())
+ if Path(receipt["cwd"]).exists():
+ try:
+ os.killpg(receipt["pid"], signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ shutil.rmtree(receipt["cwd"])
+
+ def remote_runner(self, binary):
+ source = Path(probe.__file__).read_bytes()
+ runner = subprocess.Popen([sys.executable, "-c", probe.REMOTE_BOOTSTRAP, str(len(source)), "--watch-stdin",
+ "--rustfs-bin", binary, "--case", "secret-8"],
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ runner.stdin.write(source)
+ runner.stdin.write(b"\n")
+ runner.stdin.flush()
+ return runner
+
+ def ssh_fixture(self, binary):
+ transport = self.root / "ssh"
+ transport.write_text(f'''#!{sys.executable}
+import json
+import os
+from pathlib import Path
+import shlex
+import sys
+assert not os.isatty(0), "fixture must exercise non-PTY transport"
+Path({str(self.root / 'ssh-args.json')!r}).write_text(json.dumps(sys.argv[1:]))
+command = shlex.split(sys.argv[-1])
+assert command[0] == "python3"
+command[0] = sys.executable
+command[command.index("--rustfs-bin") + 1] = {binary!r}
+os.execv(command[0], command)
+''')
+ transport.chmod(0o755)
+ return {**os.environ, "PATH": str(self.root) + os.pathsep + os.environ["PATH"]}
+
def test_exact_short_secret_and_access_boundaries_require_length_errors(self):
for case, field, access_bytes, secret_bytes in (("secret-4", "secret", 4, 4),
("secret-7", "secret", 4, 7),
@@ -259,9 +317,25 @@ def test_exact_short_secret_and_access_boundaries_require_length_errors(self):
self.assertEqual(observation["cleanup"], "PASS")
self.assert_cleaned()
+ def test_rejection_diagnostic_identifies_the_invalid_credential(self):
+ for field in ("access", "secret"):
+ with self.subTest(field=field):
+ for diagnostic in (f"Error: invalid {field} key length", f"Invalid{field.title()}KeyLength",
+ f"root {field} key is too short", f"Error: RUSTFS_{field.upper()}_KEY must be at least 8 bytes"):
+ self.assertTrue(probe.credential_rejection(diagnostic, field), diagnostic)
+ other = "secret" if field == "access" else "access"
+ for diagnostic in (f"Error: permission denied reading {field} key length configuration",
+ f"INFO: invalid {field} key length\nError: permission denied",
+ f"Error: invalid {other} key length\n{field} key minimum length is 8 bytes"):
+ self.assertFalse(probe.credential_rejection(diagnostic, field), diagnostic)
+
def test_unrelated_startup_errors_clean_exit_crashes_and_timeouts_fail(self):
scenarios = [("exit-1", "Error: Address already in use"),
("exit-1", "WARNING: secret key minimum length is 8 bytes\nError: permission denied"),
+ ("exit-1", "INFO: secret key minimum length is 8 bytes\nError: permission denied"),
+ ("exit-1", "DEBUG: secret key must be at least 8 bytes\nError: permission denied"),
+ ("exit-1", "TRACE: secret key minimum length is 8 bytes\nError: permission denied"),
+ ("exit-1", "secret key minimum length is 8 bytes\nError: permission denied"),
("exit-1", "Error: invalid access key length"),
("exit-0", "Error: invalid secret key length"),
("crash", "Error: invalid secret key length"), ("timeout", "")]
@@ -269,10 +343,116 @@ def test_unrelated_startup_errors_clean_exit_crashes_and_timeouts_fail(self):
with self.subTest(mode=mode, diagnostic=diagnostic):
observation = {}
with self.assertRaises(probe.RegressionFailure):
- probe.check_startup(self.binary(mode, diagnostic), "secret-4", 0.7, observation)
+ probe.check_startup(self.binary(mode, diagnostic), "secret-4", 0.7 if mode == "timeout" else 3, observation)
self.assertEqual(observation["cleanup"], "PASS")
self.assert_cleaned()
+ def test_cancellation_during_cleanup_waits_for_forced_shutdown_and_removal(self):
+ for mode in ("slow-cleanup", "cancelled-slow-cleanup"):
+ with self.subTest(mode=mode):
+ self.receipt.unlink(missing_ok=True)
+ self.shutdown_marker.unlink(missing_ok=True)
+ runner = subprocess.Popen([sys.executable, str(Path(probe.__file__).resolve()), "--rustfs-bin", self.binary(mode),
+ "--case", "secret-8"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ self.wait_receipt()
+ if mode == "cancelled-slow-cleanup":
+ runner.send_signal(signal.SIGTERM)
+ deadline = time.monotonic() + 5
+ while not self.shutdown_marker.exists() and time.monotonic() < deadline:
+ time.sleep(0.02)
+ self.assertTrue(self.shutdown_marker.exists(), "cleanup did not send SIGTERM to the child")
+ # This second cancellation arrives while stop_process is
+ # waiting on a child that ignores its first SIGTERM.
+ runner.send_signal(signal.SIGINT)
+ runner.send_signal(signal.SIGHUP)
+ stdout, stderr = runner.communicate(timeout=12)
+ self.assertEqual(runner.returncode, 130, stdout + stderr)
+ self.assertNotIn("PASS", stdout)
+ self.assert_cleaned()
+ finally:
+ if runner.poll() is None:
+ runner.kill()
+ runner.communicate(timeout=5)
+ self.rescue_fixture()
+
+ def test_channel_eof_cleans_the_remote_child_without_a_signal(self):
+ runner = self.remote_runner(self.binary("timeout"))
+ try:
+ self.wait_receipt()
+ # No signal is sent to either remote process. This is the stdin
+ # closure delivered by a non-PTY SSH channel disconnect.
+ runner.stdin.close()
+ runner.stdin = None
+ stdout, stderr = runner.communicate(timeout=5)
+ self.assertEqual(runner.returncode, 130, (stdout + stderr).decode())
+ self.assertNotIn(b"PASS", stdout)
+ self.assert_cleaned()
+ finally:
+ if runner.poll() is None:
+ runner.kill()
+ runner.communicate(timeout=5)
+ self.rescue_fixture()
+
+ def test_stalled_heartbeat_cleans_up_even_while_the_channel_is_open(self):
+ runner = self.remote_runner(self.binary("timeout"))
+ try:
+ self.wait_receipt()
+ # wait(), unlike communicate(), leaves the channel open. No EOF
+ # or SIGHUP can account for the watchdog's cancellation here.
+ self.assertEqual(runner.wait(timeout=probe.SSH_HEARTBEAT_TIMEOUT + 5), 130)
+ stdout, stderr = runner.communicate(timeout=5)
+ self.assertNotIn(b"PASS", stdout, stderr.decode())
+ self.assert_cleaned()
+ finally:
+ if runner.poll() is None:
+ runner.kill()
+ runner.communicate(timeout=5)
+ self.rescue_fixture()
+
+ def test_suite_ssh_entrypoint_keeps_heartbeats_and_preserves_the_remote_verdict(self):
+ environment = self.ssh_fixture(self.binary("slow-start"))
+ script = Path(probe.__file__).parent.parent / "rustfs-security-test.sh"
+ function = re.search(r"(?ms)^sts_root_startup_case\(\) \{.*?^\}", script.read_text()).group(0)
+ command = function + '''
+SSH_OPTS=(-o BatchMode=yes -p 2222)
+SSH_USER=fixture
+NODES=(node-first node-other)
+sts_root_startup_case minimum
+'''
+ result = subprocess.run(["bash", "-Eeu", "-o", "pipefail", "-c", command],
+ env={**environment, "SCRIPT_DIR": str(script.parent)},
+ capture_output=True, text=True, timeout=20)
+ self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+ observation = json.loads(result.stdout)
+ self.assertEqual(observation["case_id"], "STS-110")
+ self.assertEqual(observation["verdict"], "PASS")
+ arguments = json.loads((self.root / "ssh-args.json").read_text())
+ self.assertIn("-T", arguments)
+ self.assertIn("BatchMode=yes", arguments)
+ self.assertEqual(arguments[arguments.index("-p") + 1], "2222")
+ self.assertEqual(arguments[-2], "fixture@node-first")
+ self.assertIn("/usr/bin/rustfs", arguments[-1])
+ self.assert_cleaned()
+
+ def test_local_cancellation_closes_the_supervised_channel_and_waits_for_remote_cleanup(self):
+ environment = self.ssh_fixture(self.binary("timeout"))
+ runner = subprocess.Popen([sys.executable, str(Path(probe.__file__).resolve()), "--ssh-target", "fixture@node",
+ "--rustfs-bin", "/usr/bin/rustfs", "--case", "secret-8"],
+ env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+ try:
+ self.wait_receipt()
+ runner.send_signal(signal.SIGTERM)
+ stdout, stderr = runner.communicate(timeout=10)
+ self.assertEqual(runner.returncode, 130, stdout + stderr)
+ self.assertNotIn("PASS", stdout)
+ self.assert_cleaned()
+ finally:
+ if runner.poll() is None:
+ runner.kill()
+ runner.communicate(timeout=5)
+ self.rescue_fixture()
+
def test_accepting_short_credentials_fails_even_with_200_or_403_and_preserves_500(self):
for mode, status in (("success", 200), ("sts-403", 403), ("sts-500", 500)):
with self.subTest(mode=mode):