bench/aegir/src/lib.rs
//! Ægir — a tiny pure-Rust S3 client.
//!
//! HTTP + AWS Signature V4 (HMAC-SHA256). No `maybe-async`, no C, no OpenSSL —
//! so it compiles happily alongside `gix` where `rust-s3` can't (gix flips
//! `maybe-async/is_sync` globally and breaks rust-s3's async path). Path-style
//! addressing → works against RustFS / MinIO / real S3.
//!
//! Reads come back as [`bytes::Bytes`] so the body is shared, not re-copied, on
//! the way into arrow/parquet.
//!
//! ## Transports
//!
//! SigV4 signing and the request/response *semantics* are identical regardless
//! of how the bytes move. Only the byte transport is pluggable:
//!
//! - **`ureq`** (default) — pure-Rust HTTP over rustls. Works for TLS and plain.
//! - **`uring`** (opt-in feature) — an io_uring TCP transport for the
//! **plaintext** hot path (RustFS on `:9000`, no TLS). Built on the pure-Rust
//! `io-uring` crate (kernel-syscall wrappers; deps are only `libc`/`bitflags`/
//! `cfg-if` — no C, no bindgen). The socket is connected with `std`'s resolver
//! and the request/response bytes are pushed/pulled through a per-call
//! submission/completion loop (`Send`/`Recv` opcodes). TLS endpoints always
//! fall back to `ureq`, so `uring` is strictly additive.
//!
//! The transport is chosen per [`Client`] via [`Client::with_transport`] (or the
//! `AEGIR_TRANSPORT=uring` env). `ureq` stays the default.
use bytes::Bytes;
use chrono::Utc;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::io::Read;
#[cfg(feature = "uring")]
mod uring;
type HmacSha256 = Hmac<Sha256>;
/// The page ceiling [`Client::list_objects`] asks ListObjectsV2 for.
///
/// A CEILING, not a demand — AWS S3 clamps to 1 000 and paginates, so this cannot
/// change what a listing returns, only how many round trips it costs. It exists for
/// servers that honour it: see [`Client::list_objects`] for the measured 1.6 M-object
/// case where the per-page cost is a full walk and sort of the whole prefix.
const LIST_MAX_KEYS: &str = "1048576";
#[derive(Debug)]
pub struct Error(String);
impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "aegir: {}", self.0)
}
}
impl std::error::Error for Error {}
impl Error {
fn new(s: impl Into<String>) -> Self {
Error(s.into())
}
}
pub type Result<T> = std::result::Result<T, Error>;
/// Which byte transport a [`Client`] uses.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Transport {
/// Pure-Rust HTTP over rustls (`ureq`). The default; handles TLS + plaintext.
Ureq,
/// io_uring TCP for the plaintext hot path. Requires the `uring` feature;
/// TLS endpoints transparently fall back to `ureq`.
Uring,
}
impl Transport {
/// Default transport, honouring `AEGIR_TRANSPORT=uring|ureq` if set.
fn from_env() -> Self {
match std::env::var("AEGIR_TRANSPORT").ok().as_deref() {
Some("uring") => Transport::Uring,
_ => Transport::Ureq,
}
}
}
/// A SigV4-signed request, transport-agnostic. Built once by [`Client::signed`],
/// then handed to whichever transport executes it. Keeping this struct between
/// signing and sending is what lets `ureq` and `uring` share identical
/// request/response semantics.
struct SignedReq {
method: String,
/// Full URL (`{endpoint}{path}?{query}`).
url: String,
/// Header (name, value) pairs to send, in addition to `Host`. Includes the
/// SigV4 trio (`Authorization`, `x-amz-date`, `x-amz-content-sha256`) plus
/// any per-op extras (`Content-Type`, `Range`).
headers: Vec<(String, String)>,
/// Request body (empty for GET/HEAD/DELETE).
body: Vec<u8>,
}
impl SignedReq {
fn header(&mut self, name: &str, value: &str) {
self.headers.push((name.to_string(), value.to_string()));
}
}
/// A transport response: status + headers + body reader. Modeled on the subset
/// of `ureq::Response` the ops use, so both transports return the same shape.
struct Resp {
status: u16,
headers: Vec<(String, String)>,
body: Vec<u8>,
}
impl Resp {
fn header(&self, name: &str) -> Option<&str> {
self.headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case(name))
.map(|(_, v)| v.as_str())
}
}
/// An S3 client bound to one endpoint + credentials. Cheap to clone.
#[derive(Clone)]
pub struct Client {
// (Debug impl below — never prints the secret key.)
endpoint: String, // e.g. http://localhost:9000 (no trailing slash)
host: String, // host[:port] — must match the signed Host header
region: String,
access_key: String,
secret_key: String,
transport: Transport,
/// `true` when `endpoint` is `https://` (uring cannot speak TLS). Only read
/// on the `uring` code path.
#[cfg_attr(not(feature = "uring"), allow(dead_code))]
tls: bool,
agent: ureq::Agent,
/// Idle pool size set by [`Client::with_max_idle_per_host`] (`None` = ureq's default).
max_idle_per_host: Option<usize>,
/// Client TLS set by [`Client::with_tls_config`] (`None` = ureq's webpki roots, no client identity).
tls_config: Option<ureq::tls::TlsConfig>,
/// STS session token set by [`Client::with_session_token`]: signed and sent as
/// `x-amz-security-token` (temporary credentials, e.g. vended by a catalog).
session_token: Option<String>,
}
impl std::fmt::Debug for Client {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("aegir::Client")
.field("endpoint", &self.endpoint)
.field("region", &self.region)
.field("transport", &self.transport)
.finish_non_exhaustive()
}
}
impl Client {
pub fn new(
endpoint: impl Into<String>,
region: impl Into<String>,
access_key: impl Into<String>,
secret_key: impl Into<String>,
) -> Self {
let endpoint = endpoint.into().trim_end_matches('/').to_string();
let tls = endpoint.starts_with("https://");
let host = endpoint.split_once("://").map(|(_, h)| h).unwrap_or(&endpoint).to_string();
Self {
endpoint,
host,
region: region.into(),
access_key: access_key.into(),
secret_key: secret_key.into(),
transport: Transport::from_env(),
tls,
agent: build_agent(None, None),
max_idle_per_host: None,
tls_config: None,
session_token: None,
}
}
/// **Sign with temporary (STS) credentials**: every request carries — and
/// signs — `x-amz-security-token`. Needed when the keys came from an STS
/// `AssumeRole` or were vended by an Iceberg REST catalog. An empty token is
/// treated as none.
pub fn with_session_token(mut self, token: impl Into<String>) -> Self {
let token = token.into();
self.session_token = (!token.trim().is_empty()).then_some(token);
self
}
/// The agent for the current pool size and TLS config.
fn rebuild_agent(&mut self) {
self.agent = build_agent(self.max_idle_per_host, self.tls_config.clone());
}
/// **Use this TLS config for `https://`** — a private CA to verify the server
/// against, and/or a client certificate for mutual TLS (a MinIO that requires
/// one). Without it, ureq's webpki roots and no client identity. Composes with
/// [`with_max_idle_per_host`](Self::with_max_idle_per_host) in either order.
///
/// A `ureq::tls::TlsConfig`, not a rustls `ClientConfig`, since ureq 3 (2026-10-02):
/// ureq 3 owns its rustls setup and takes roots + identity as data.
pub fn with_tls_config(mut self, tls_config: ureq::tls::TlsConfig) -> Self {
self.tls_config = Some(tls_config);
self.rebuild_agent();
self
}
/// **Size the HTTP keep-alive pool.** `ureq::agent()` — what [`Client::new`]
/// builds — keeps `DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST = 1` idle connection
/// **per host**. That is invisible and free for a serial caller, and it is a
/// per-request TCP handshake for a concurrent one: with W threads sharing one
/// `Client`, exactly one finished connection is retained and the other W−1 are
/// closed, so W−1 of every W requests connect from scratch.
///
/// Measured on oden 2026-08-04 against the embedded RustFS: korp's tile upload
/// ran 32 threads through one shared `Client`, i.e. ~1.6 M PUTs over ~1 pooled
/// connection. Call this with at least the concurrency you intend to drive.
///
/// Additive: the default is unchanged, so no existing caller moves.
pub fn with_max_idle_per_host(mut self, max_idle_per_host: usize) -> Self {
self.max_idle_per_host = Some(max_idle_per_host.max(1));
self.rebuild_agent();
self
}
/// Select the byte transport. `ureq` is the default; `uring` opts into the
/// io_uring plaintext path (and is only honoured when the `uring` feature is
/// compiled — otherwise it's accepted but behaves as `ureq`).
pub fn with_transport(mut self, transport: Transport) -> Self {
self.transport = transport;
self
}
/// The transport actually in effect for this client (after feature/TLS
/// fallbacks). Useful for benchmark labelling.
pub fn effective_transport(&self) -> Transport {
#[cfg(feature = "uring")]
{
if self.transport == Transport::Uring && !self.tls {
return Transport::Uring;
}
}
Transport::Ureq
}
/// Build a SigV4-signed request. `canonical_uri` must be the exact, already
/// percent-encoded path that will be sent.
fn signed(
&self,
method: &str,
canonical_uri: &str,
query: &[(&str, &str)],
body: &[u8],
) -> SignedReq {
let canonical_query = sigv4::canonical_query(query);
let token: Vec<(&str, &str)> =
self.session_token.as_deref().map(|t| ("x-amz-security-token", t)).into_iter().collect();
let signed = sigv4::Credentials {
access_key: &self.access_key,
secret_key: &self.secret_key,
region: &self.region,
service: "s3",
}
.sign(
method,
&self.host,
canonical_uri,
&canonical_query,
&token,
&sigv4::payload_hash(body),
&sigv4::amz_date(Utc::now()),
);
let url = if canonical_query.is_empty() {
format!("{}{}", self.endpoint, canonical_uri)
} else {
format!("{}{}?{}", self.endpoint, canonical_uri, canonical_query)
};
SignedReq {
method: method.to_string(),
url,
headers: [
("Authorization".to_string(), signed.authorization),
("x-amz-date".to_string(), signed.amz_date),
("x-amz-content-sha256".to_string(), signed.payload_hash),
]
.into_iter()
.chain(token.iter().map(|(k, v)| (k.to_string(), v.to_string())))
.collect(),
body: body.to_vec(),
}
}
/// Execute a signed request through the effective transport. Returns the
/// parsed response on a 2xx, an [`Error`] otherwise — except this helper is
/// only used where any non-2xx is an error; callers that special-case a
/// status (404 on HEAD, 409 on create-bucket) use [`Self::send_raw`].
fn send(&self, req: SignedReq) -> Result<Resp> {
let resp = self.send_raw(req)?;
if (200..300).contains(&resp.status) {
Ok(resp)
} else {
Err(Error::new(format!("http status {}", resp.status)))
}
}
/// Execute a signed request, returning the response regardless of status
/// (so callers can branch on 404/409). Picks the io_uring transport only for
/// plaintext when the `uring` feature is on; otherwise `ureq`.
fn send_raw(&self, req: SignedReq) -> Result<Resp> {
#[cfg(feature = "uring")]
{
if self.transport == Transport::Uring && !self.tls {
return uring::send(&self.host, &req);
}
}
self.send_ureq(req)
}
/// The default `ureq` transport.
fn send_ureq(&self, req: SignedReq) -> Result<Resp> {
let mut b = ureq::http::Request::builder().method(req.method.as_str()).uri(&req.url);
for (k, v) in &req.headers {
b = b.header(k, v);
}
// The agent is built with `http_status_as_error(false)`, so a 4xx/5xx is an
// `Ok` response here — the ops read S3's error XML from it, as before.
let res = if req.body.is_empty() {
self.agent.run(b.body(()).map_err(|e| Error::new(e.to_string()))?)
} else {
self.agent.run(b.body(&req.body[..]).map_err(|e| Error::new(e.to_string()))?)
};
match res {
Ok(resp) => {
let status = resp.status().as_u16();
let headers = resp
.headers()
.iter()
.filter_map(|(n, v)| v.to_str().ok().map(|v| (n.as_str().to_string(), v.to_string())))
.collect();
let cap = resp
.headers()
.get("Content-Length")
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse().ok())
.unwrap_or(0);
let mut body = Vec::with_capacity(cap);
resp.into_body().into_reader().read_to_end(&mut body).map_err(|e| Error::new(e.to_string()))?;
Ok(Resp { status, headers, body })
}
Err(e) => Err(map_ureq(e)),
}
}
pub fn put_object(&self, bucket: &str, key: &str, body: &[u8], content_type: Option<&str>) -> Result<()> {
let uri = format!("/{bucket}/{}", encode_key(key));
let mut req = self.signed("PUT", &uri, &[], body);
if let Some(ct) = content_type {
req.header("Content-Type", ct);
}
self.send(req).map(|_| ())
}
/// Zero-copy-friendly read: the body is read once into a [`Bytes`] (shared
/// downstream, never re-copied).
pub fn get_object(&self, bucket: &str, key: &str) -> Result<Bytes> {
let uri = format!("/{bucket}/{}", encode_key(key));
let resp = self.send(self.signed("GET", &uri, &[], b""))?;
Ok(Bytes::from(resp.body))
}
pub fn head_object(&self, bucket: &str, key: &str) -> Result<bool> {
Ok(self.head_size(bucket, key)?.is_some())
}
/// HEAD → the object's size, or `None` if it doesn't exist.
pub fn head_size(&self, bucket: &str, key: &str) -> Result<Option<u64>> {
let uri = format!("/{bucket}/{}", encode_key(key));
let resp = self.send_raw(self.signed("HEAD", &uri, &[], b""))?;
match resp.status {
404 => Ok(None),
s if (200..300).contains(&s) => {
Ok(Some(resp.header("Content-Length").and_then(|s| s.parse().ok()).unwrap_or(0)))
}
s => Err(Error::new(format!("http status {s}"))),
}
}
/// Ranged GET (`bytes=start-end`, end inclusive) → [`Bytes`].
pub fn get_object_range(&self, bucket: &str, key: &str, start: u64, end_inclusive: u64) -> Result<Bytes> {
let uri = format!("/{bucket}/{}", encode_key(key));
let mut req = self.signed("GET", &uri, &[], b"");
req.header("Range", &format!("bytes={start}-{end_inclusive}"));
let resp = self.send(req)?;
Ok(Bytes::from(resp.body))
}
pub fn delete_object(&self, bucket: &str, key: &str) -> Result<()> {
let uri = format!("/{bucket}/{}", encode_key(key));
let resp = self.send_raw(self.signed("DELETE", &uri, &[], b""))?;
// S3 DELETE returns 204; some stores 200/404 (already gone) — all fine.
if resp.status == 404 || (200..300).contains(&resp.status) {
Ok(())
} else {
Err(Error::new(format!("http status {}", resp.status)))
}
}
pub fn create_bucket(&self, bucket: &str) -> Result<()> {
let uri = format!("/{bucket}");
let resp = self.send_raw(self.signed("PUT", &uri, &[], b""))?;
match resp.status {
// already owned / exists — both fine for idempotent ensure.
409 => Ok(()),
s if (200..300).contains(&s) => Ok(()),
s => Err(Error::new(format!("create_bucket http status {s}"))),
}
}
/// List **every** key under `prefix` (ListObjectsV2), following the continuation
/// token to the end. Hand-parses `<Key>` from the XML to stay dependency-light.
///
/// # Why it ASKS for a big page (`max-keys`), measured 2026-08-03
///
/// Page size is a request for a *ceiling*, not a demand: AWS S3 clamps
/// `max-keys` to 1 000 and keeps paginating, so asking for more can never return
/// more than a server is willing to give and never changes the result. What it
/// changes is the number of ROUND TRIPS — and against the embedded rustfs
/// (`s3s-fs` 0.14) the round-trip count is the whole cost.
///
/// `s3s-fs`'s `list_objects_v2` (`src/s3.rs`) recursively walks the ENTIRE prefix
/// and then sorts EVERY key, on every page, before discarding all but the 1 000
/// after the continuation marker. That is O(n² log n) in the object count, paid
/// per page. On oden, korp's `--tile-maps sweden` retire step listed
/// `sweden/tiles/` after the z15 ladder landed: **1 605 675 objects**, i.e. 1 606
/// pages at the old default, each one re-walking and re-sorting all 1.6 M keys.
/// It had not finished after 17 minutes and was on course for hours; the same
/// step over the previous 28 974-tile pyramid took seconds. Nothing was wrong
/// with the pyramid — the LIST could not deliver it.
///
/// One page of 1 048 576 collapses that to a single walk for any pyramid this
/// stack cuts, and the loop below is unchanged: a server that truncates still
/// gets followed to the end, so correctness does not depend on the ceiling being
/// honoured.
///
/// # This returned only the FIRST PAGE until 2026-08-02, and looked complete
/// S3 caps one ListObjectsV2 response at 1 000 keys and reports the cut with
/// `<IsTruncated>true</IsTruncated>` + a `<NextContinuationToken>`. This method
/// issued exactly one request and ignored both, so a prefix holding more than 1 000
/// objects silently returned 1 000 — a `Vec` shaped exactly like a complete listing.
///
/// Measured: korp's `--tile-maps sweden` retire step asked for every stale tile
/// under `sweden/tiles/`, was handed **999** of the **23 987** that were there, and
/// printed a confident `retired 999/999`. A caller cannot detect this from the
/// return value — which makes it the "empty and unreadable render identically"
/// failure, one page in.
pub fn list_objects(&self, bucket: &str, prefix: &str) -> Result<Vec<String>> {
let uri = format!("/{bucket}");
let mut out: Vec<String> = Vec::new();
let mut token: Option<String> = None;
loop {
let mut query: Vec<(&str, &str)> =
vec![("list-type", "2"), ("prefix", prefix), ("max-keys", LIST_MAX_KEYS)];
if let Some(t) = token.as_deref() {
query.push(("continuation-token", t));
}
let resp = self.send(self.signed("GET", &uri, &query, b""))?;
let body = String::from_utf8_lossy(&resp.body);
out.extend(
body.split("<Key>")
.skip(1)
.filter_map(|s| s.split("</Key>").next())
.map(xml_unescape),
);
// Only a truncated page carries a next token; its absence ends the walk,
// so a server that never truncates does exactly one request as before.
token = xml_tag(&body, "NextContinuationToken");
if token.is_none() {
return Ok(out);
}
}
}
/// **The immediate child "folders" under `prefix`** — a DELIMITED `ListObjectsV2`
/// (`delimiter=/`), which answers with `<CommonPrefixes>` instead of every key
/// beneath them. Returns the child NAMES: the query `prefix` and the trailing `/`
/// are stripped, so `list_prefixes(b, "")` over a bucket holding
/// `sweden/tiles/z15/…` answers `["sweden"]`, not 1.6 M keys.
///
/// # Why this exists next to [`Client::list_objects`]
///
/// "Which regions are in this bucket?" and "what is every object in it?" are
/// different questions, and asking the second to answer the first is what made
/// korp's map-asset roster cost a full store walk. MEASURED on oden 2026-08-04
/// against korp's embedded `s3s-fs`, twice, because another agent deleted a retired
/// pyramid between the readings — so both are on the record:
///
/// ```text
/// undelimited 3 211 352 keys 184.45 s wall 357.0 s server CPU (load avg 10.4)
/// undelimited 1 605 676 keys 29.91 s wall 36.1 s server CPU (load avg 2.3)
/// delimited ["sweden"] 2.13-5.21 s
/// ```
///
/// …for an answer that is one word long.
///
/// # HONEST about the backend
///
/// On AWS/MinIO a delimited listing is served from the index and is O(children).
/// `s3s-fs` 0.14.1 (`list_objects_with_delimiter`, `src/s3.rs:1266`) still walks
/// every directory — but it skips the per-file `metadata()` statx, the sort of the
/// whole key set, and the XML body carrying it, which is where that 357 s went.
/// So this is cheaper everywhere and asymptotically right on a real store; it is
/// NOT a claim that s3s-fs answers it in O(1).
///
/// Paginates exactly as [`Client::list_objects`] does: `CommonPrefixes` count
/// against `max-keys`, so a truncated page is followed to the end.
pub fn list_prefixes(&self, bucket: &str, prefix: &str) -> Result<Vec<String>> {
let uri = format!("/{bucket}");
let mut out: Vec<String> = Vec::new();
let mut token: Option<String> = None;
loop {
let mut query: Vec<(&str, &str)> = vec![
("list-type", "2"),
("prefix", prefix),
("delimiter", "/"),
("max-keys", LIST_MAX_KEYS),
];
if let Some(t) = token.as_deref() {
query.push(("continuation-token", t));
}
let resp = self.send(self.signed("GET", &uri, &query, b""))?;
let body = String::from_utf8_lossy(&resp.body);
out.extend(common_prefixes_in(&body, prefix));
token = xml_tag(&body, "NextContinuationToken");
if token.is_none() {
out.sort();
out.dedup();
return Ok(out);
}
}
}
}
/// **The CHILD NAMES in a delimited `ListObjectsV2` body**, relative to the request
/// `prefix` and with the trailing delimiter stripped. Pure, so the one thing that can
/// silently go wrong here is assertable without a store.
///
/// The one thing that can silently go wrong: the response carries its OWN `<Prefix>`
/// element — the echo of what was asked for — inside `<ListBucketResult>`, before any
/// `<CommonPrefixes>`. Taking every `<Prefix>` therefore yields one extra entry that is
/// either the empty string (a root listing) or the parent itself, and an empty region
/// name reaching a caller is exactly the "half-broken asset" that has no business
/// existing. A child always EXTENDS the request prefix and always ends in the delimiter,
/// so requiring both drops the echo without depending on element order or on a
/// `skip(1)` that a differently-shaped server would break.
fn common_prefixes_in(body: &str, prefix: &str) -> Vec<String> {
body.split("<Prefix>")
.skip(1)
.filter_map(|s| s.split("</Prefix>").next())
.map(xml_unescape)
.filter(|p| p.len() > prefix.len() && p.ends_with('/'))
.filter_map(|p| p.strip_prefix(prefix).map(|c| c.trim_end_matches('/').to_string()))
.filter(|c| !c.is_empty())
.collect()
}
/// The text content of the first `<tag>…</tag>` in `xml`, unescaped; `None` when the
/// tag is absent or unterminated.
fn xml_tag(xml: &str, tag: &str) -> Option<String> {
let open = format!("<{tag}>");
let rest = xml.split_once(&open)?.1;
let inner = rest.split_once(&format!("</{tag}>"))?.0;
Some(xml_unescape(inner))
}
/// **AWS Signature V4, for any service** — the one signer [`Client`] uses for
/// `s3`, public so a caller signing a *different* AWS-shaped endpoint reuses it
/// instead of growing a second copy.
///
/// The case that needed it: MinIO AIStor's Iceberg REST catalog (`/_iceberg`)
/// answers on the S3 port and wants SigV4 with the service **`s3tables`**, not
/// `s3`. Signed with `s3` it answers `400 AuthorizationParametersError`, which
/// reads like a licence refusal and is not one.
pub mod sigv4 {
use super::{Sha256, hmac, uri_encode};
use sha2::Digest;
/// Who signs, and for which region/service scope.
#[derive(Clone, Copy)]
pub struct Credentials<'a> {
pub access_key: &'a str,
pub secret_key: &'a str,
pub region: &'a str,
/// `s3` for object storage, `s3tables` for AIStor's Iceberg REST catalog.
pub service: &'a str,
}
// Hand-written: never print the secret.
impl std::fmt::Debug for Credentials<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "sigv4 {}/{}/{}", self.access_key, self.region, self.service)
}
}
/// The three headers a signed request carries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Signed {
pub authorization: String,
pub amz_date: String,
pub payload_hash: String,
}
/// `YYYYMMDDTHHMMSSZ`, the `x-amz-date` form.
pub fn amz_date(now: chrono::DateTime<chrono::Utc>) -> String {
now.format("%Y%m%dT%H%M%SZ").to_string()
}
/// [`amz_date`] of the current UTC time.
pub fn amz_date_now() -> String {
amz_date(chrono::Utc::now())
}
/// Lowercase hex SHA-256 of the body — `x-amz-content-sha256`.
pub fn payload_hash(body: &[u8]) -> String {
nornir_hash::hex_encode(&Sha256::digest(body))
}
/// The canonical query string: each key and value AWS-URI-encoded, pairs
/// sorted, joined with `&`. Also the query to put on the wire, so what is
/// signed is exactly what is sent.
pub fn canonical_query(pairs: &[(&str, &str)]) -> String {
let mut q: Vec<(String, String)> =
pairs.iter().map(|(k, v)| (uri_encode(k, true), uri_encode(v, true))).collect();
q.sort();
q.iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join("&")
}
/// AWS URI-encode (`A-Za-z0-9-_.~` kept, the rest `%XX`); `/` kept unless
/// `encode_slash`.
pub fn encode(s: &str, encode_slash: bool) -> String {
uri_encode(s, encode_slash)
}
impl Credentials<'_> {
/// Sign one request. `host`, `x-amz-content-sha256` and `x-amz-date` are
/// always signed; `extra` adds more (name, value) pairs to the signed set.
/// `canonical_uri` and `canonical_query` must be exactly what goes on the
/// wire (already encoded).
#[allow(clippy::too_many_arguments)]
pub fn sign(
&self,
method: &str,
host: &str,
canonical_uri: &str,
canonical_query: &str,
extra: &[(&str, &str)],
payload_hash: &str,
amz_date: &str,
) -> Signed {
let date_stamp = &amz_date[..8.min(amz_date.len())];
let mut headers: Vec<(String, String)> = vec![
("host".into(), host.trim().to_string()),
("x-amz-content-sha256".into(), payload_hash.to_string()),
("x-amz-date".into(), amz_date.to_string()),
];
headers.extend(extra.iter().map(|(k, v)| (k.to_ascii_lowercase(), v.trim().to_string())));
headers.sort();
let canonical_headers: String = headers.iter().map(|(k, v)| format!("{k}:{v}\n")).collect();
let signed_headers = headers.iter().map(|(k, _)| k.as_str()).collect::<Vec<_>>().join(";");
let canonical_request = format!(
"{method}\n{canonical_uri}\n{canonical_query}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"
);
let scope = format!("{date_stamp}/{}/{}/aws4_request", self.region, self.service);
let string_to_sign = format!(
"AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}",
nornir_hash::hex_encode(&Sha256::digest(canonical_request.as_bytes()))
);
let k_date = hmac(format!("AWS4{}", self.secret_key).as_bytes(), date_stamp.as_bytes());
let k_region = hmac(&k_date, self.region.as_bytes());
let k_service = hmac(&k_region, self.service.as_bytes());
let k_signing = hmac(&k_service, b"aws4_request");
let signature = nornir_hash::hex_encode(&hmac(&k_signing, string_to_sign.as_bytes()));
Signed {
authorization: format!(
"AWS4-HMAC-SHA256 Credential={}/{scope}, SignedHeaders={signed_headers}, Signature={signature}",
self.access_key
),
amz_date: amz_date.to_string(),
payload_hash: payload_hash.to_string(),
}
}
}
}
fn hmac(key: &[u8], data: &[u8]) -> Vec<u8> {
let mut m = <HmacSha256 as hmac::KeyInit>::new_from_slice(key).expect("HMAC accepts any key length");
m.update(data);
m.finalize().into_bytes().to_vec()
}
/// AWS URI-encode: unreserved = `A-Za-z0-9-_.~`; everything else `%XX` (upper).
/// `/` is kept when `encode_slash` is false (object keys are path-segmented).
fn uri_encode(s: &str, encode_slash: bool) -> String {
let mut out = String::with_capacity(s.len());
for b in s.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(b as char),
b'/' if !encode_slash => out.push('/'),
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
fn encode_key(key: &str) -> String {
uri_encode(key.trim_start_matches('/'), false)
}
fn xml_unescape(s: &str) -> String {
s.replace("&", "&").replace("<", "<").replace(">", ">").replace(""", "\"")
}
/// The ureq 3 agent with ureq 2's behaviour (ureq 2 -> 3, 2026-10-02): a 4xx/5xx is a
/// response, not an error (ureq 2 handed it back in `Error::Status`); no proxy from
/// the environment and at most 5 redirects (ureq 2 without `proxy-from-env`); and
/// ureq 2's pool of 100 idle connections, 1 per host, unless sized by
/// [`Client::with_max_idle_per_host`] (ureq 3 defaults to 10 / 3).
fn build_agent(max_idle_per_host: Option<usize>, tls: Option<ureq::tls::TlsConfig>) -> ureq::Agent {
let mut b = ureq::Agent::config_builder()
.http_status_as_error(false)
.proxy(None)
.max_redirects(5);
b = match max_idle_per_host {
Some(n) => b.max_idle_connections_per_host(n).max_idle_connections(n.saturating_mul(2).max(100)),
None => b.max_idle_connections_per_host(1).max_idle_connections(100),
};
if let Some(tls) = tls {
b = b.tls_config(tls);
}
b.build().new_agent()
}
fn map_ureq(e: ureq::Error) -> Error {
Error::new(e.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn uri_encode_keeps_slashes_in_keys() {
assert_eq!(encode_key("wh/data/file.parquet"), "wh/data/file.parquet");
assert_eq!(uri_encode("a b+c", true), "a%20b%2Bc");
}
#[test]
fn hex_is_lowercase() {
// The hex encoder is now the shared `nornir_hash::hex_encode` (dedup);
// keep a local guard that aegir's SigV4 hex is lowercase-hex.
assert_eq!(nornir_hash::hex_encode(&[0xde, 0xad, 0xbe, 0xef]), "deadbeef");
}
/// **REGRESSION: a truncated ListObjectsV2 page must be DETECTED, not returned as
/// if it were the whole listing.** `list_objects` issued one request and ignored
/// `<IsTruncated>` and `<NextContinuationToken>` entirely, so a prefix with more
/// than 1 000 objects came back as 1 000 — a `Vec` indistinguishable from a complete
/// one. Measured on korp's tile store: 999 keys returned of 23 987 present, and the
/// caller printed `retired 999/999`.
///
/// The paging loop hinges on reading the next token out of the page, so this pins
/// that reader against real S3 response shapes: a truncated page yields the token,
/// a final page yields `None` (which is what ENDS the walk — a parser that always
/// returned `Some` would loop forever, and one that always returned `None` would
/// restore the original bug).
#[test]
fn a_truncated_listing_page_yields_its_continuation_token() {
let truncated = "<?xml version=\"1.0\"?><ListBucketResult>\
<Name>nordisk-maps</Name><Prefix>sweden/tiles/</Prefix>\
<KeyCount>1000</KeyCount><IsTruncated>true</IsTruncated>\
<NextContinuationToken>1ueGcxLPRx1Tr/XYExHnhbYLgveDs2J/wm36Hy4vbOwM=</NextContinuationToken>\
<Contents><Key>sweden/tiles/z12/2174/947.parquet</Key></Contents>\
</ListBucketResult>";
assert_eq!(
xml_tag(truncated, "NextContinuationToken").as_deref(),
Some("1ueGcxLPRx1Tr/XYExHnhbYLgveDs2J/wm36Hy4vbOwM="),
"a truncated page MUST hand back its token, or the walk stops one page in"
);
assert_eq!(xml_tag(truncated, "IsTruncated").as_deref(), Some("true"));
// The FINAL page carries no token — this is what terminates the loop.
let last = "<?xml version=\"1.0\"?><ListBucketResult>\
<Name>nordisk-maps</Name><KeyCount>7</KeyCount><IsTruncated>false</IsTruncated>\
<Contents><Key>sweden/tiles/manifest.json</Key></Contents></ListBucketResult>";
assert_eq!(
xml_tag(last, "NextContinuationToken"),
None,
"a complete page must NOT report a next token, or the walk never ends"
);
// Entity-escaped tokens survive (`+` and `=` are common in real tokens, `&` in
// some implementations' opaque cursors).
let esc = "<NextContinuationToken>a&b=</NextContinuationToken>";
assert_eq!(xml_tag(esc, "NextContinuationToken").as_deref(), Some("a&b="));
// Unterminated / absent tags are `None`, never a partial token.
assert_eq!(xml_tag("<NextContinuationToken>oops", "NextContinuationToken"), None);
assert_eq!(xml_tag("<Other>x</Other>", "NextContinuationToken"), None);
}
// Known-answer SigV4 signing-key chain (AWS docs example).
#[test]
fn sigv4_signing_key_matches_aws_example() {
let secret = "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY";
let k_date = hmac(format!("AWS4{secret}").as_bytes(), b"20150830");
let k_region = hmac(&k_date, b"us-east-1");
let k_service = hmac(&k_region, b"iam");
let k_signing = hmac(&k_service, b"aws4_request");
assert_eq!(
nornir_hash::hex_encode(&k_signing),
"c4afb1cc5771d871763a393e44b703571b55cc28424d1a5e86da6ed3c154a4b9"
);
}
// Known-answer: AWS's own S3 SigV4 worked examples (header-based auth,
// "GET Object" with a signed Range, and "List Objects"). If the shared
// signer drifts by one byte these signatures change.
#[test]
fn sigv4_matches_aws_s3_worked_examples() {
let c = sigv4::Credentials {
access_key: "AKIAIOSFODNN7EXAMPLE",
secret_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
region: "us-east-1",
service: "s3",
};
let empty = sigv4::payload_hash(b"");
assert_eq!(empty, "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855");
let get = c.sign(
"GET",
"examplebucket.s3.amazonaws.com",
"/test.txt",
"",
&[("Range", "bytes=0-9")],
&empty,
"20130524T000000Z",
);
assert_eq!(
get.authorization,
"AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, \
SignedHeaders=host;range;x-amz-content-sha256;x-amz-date, \
Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"
);
let list = c.sign(
"GET",
"examplebucket.s3.amazonaws.com",
"/",
&sigv4::canonical_query(&[("prefix", "J"), ("max-keys", "2")]),
&[],
&empty,
"20130524T000000Z",
);
assert!(
list.authorization
.ends_with("Signature=34b48302e7b5fa45bde8084f4b7868a86f0a534bc59db6670ed5711ef69dc6f7"),
"{}",
list.authorization
);
}
// The service is part of the scope AND of the key chain: an `s3tables`
// signature must differ from an `s3` one over the same request.
#[test]
fn sigv4_service_is_in_the_scope() {
let mut c = sigv4::Credentials { access_key: "ak", secret_key: "sk", region: "us-east-1", service: "s3" };
let h = sigv4::payload_hash(b"");
let s3 = c.sign("GET", "minio:9000", "/_iceberg/v1/warehouses", "", &[], &h, "20260926T120000Z");
c.service = "s3tables";
let st = c.sign("GET", "minio:9000", "/_iceberg/v1/warehouses", "", &[], &h, "20260926T120000Z");
assert!(st.authorization.contains("/20260926/us-east-1/s3tables/aws4_request"));
assert_ne!(s3.authorization, st.authorization);
}
// SigV4 signing is identical regardless of transport: `signed()` produces the
// same Authorization header whether the bytes later go out via ureq or uring.
#[test]
fn signed_request_carries_sigv4_trio() {
let c = Client::new("http://localhost:9000", "us-east-1", "ak", "sk");
let req = c.signed("GET", "/warehouse/wh/x", &[], b"");
assert_eq!(req.method, "GET");
assert_eq!(c.host, "localhost:9000");
assert!(req.headers.iter().any(|(k, v)| k == "Authorization"
&& v.starts_with("AWS4-HMAC-SHA256 Credential=ak/")));
assert!(req.headers.iter().any(|(k, _)| k == "x-amz-date"));
assert!(req.headers.iter().any(|(k, _)| k == "x-amz-content-sha256"));
assert!(!req.headers.iter().any(|(k, _)| k == "x-amz-security-token"), "no token, no header");
}
/// Temporary credentials: the token is sent AND inside the signed set, else
/// S3/MinIO answers 403 SignatureDoesNotMatch.
#[test]
fn a_session_token_is_sent_and_signed() {
let c = Client::new("http://localhost:9000", "us-east-1", "ak", "sk").with_session_token("tok3n");
let req = c.signed("GET", "/warehouse/wh/x", &[], b"");
assert!(req.headers.iter().any(|(k, v)| k == "x-amz-security-token" && v == "tok3n"));
let auth = &req.headers.iter().find(|(k, _)| k == "Authorization").unwrap().1;
assert!(auth.contains("SignedHeaders=host;x-amz-content-sha256;x-amz-date;x-amz-security-token"), "{auth}");
assert!(Client::new("http://h", "r", "a", "s").with_session_token(" ").session_token.is_none());
}
#[test]
fn tls_endpoint_never_uses_uring() {
let c = Client::new("https://s3.amazonaws.com", "us-east-1", "ak", "sk")
.with_transport(Transport::Uring);
// https → uring must fall back to ureq even when requested.
assert_eq!(c.effective_transport(), Transport::Ureq);
}
/// **★ A delimited listing names the CHILDREN, never the echo of what was asked.**
///
/// `ListObjectsV2` with a delimiter answers with the request `<Prefix>` echoed back
/// inside `<ListBucketResult>` as well as one `<Prefix>` per `<CommonPrefixes>`. A
/// parser that takes them all hands the caller an extra entry — the empty string for
/// a root listing, the parent for a nested one. korp reads this to learn which map
/// REGIONS a bucket holds, so that entry becomes a region with no name.
///
/// RED before `common_prefixes_in` filtered the echo (i.e. taking every `<Prefix>`),
/// verbatim:
///
/// ```text
/// assertion `left == right` failed: a root listing names the top-level children and
/// NOTHING else — the echoed request prefix is not a child
/// left: ["", "sweden", "norway"]
/// right: ["sweden", "norway"]
/// ```
#[test]
fn a_delimited_listing_names_the_children_and_not_the_echoed_request_prefix() {
// Shaped exactly as s3s-fs / MinIO answer it: the echo first, then the children.
let root = "<?xml version=\"1.0\"?><ListBucketResult><Name>nordisk-maps</Name>\
<Prefix></Prefix><Delimiter>/</Delimiter><KeyCount>2</KeyCount>\
<CommonPrefixes><Prefix>sweden/</Prefix></CommonPrefixes>\
<CommonPrefixes><Prefix>norway/</Prefix></CommonPrefixes>\
</ListBucketResult>";
assert_eq!(
common_prefixes_in(root, ""),
["sweden", "norway"],
"a root listing names the top-level children and NOTHING else — the echoed \
request prefix is not a child",
);
// A NESTED listing: the echo is the parent, and it extends nothing, so it goes.
let nested = "<?xml version=\"1.0\"?><ListBucketResult>\
<Prefix>sweden/</Prefix><Delimiter>/</Delimiter>\
<CommonPrefixes><Prefix>sweden/tiles/</Prefix></CommonPrefixes>\
<CommonPrefixes><Prefix>sweden/tiles.old-positional/</Prefix></CommonPrefixes>\
<Contents><Key>sweden/ways.parquet</Key></Contents>\
</ListBucketResult>";
assert_eq!(
common_prefixes_in(nested, "sweden/"),
["tiles", "tiles.old-positional"],
"children are named RELATIVE to the request prefix, delimiter stripped, and \
an object at the current level is not a child",
);
// An empty bucket names nobody — never one blank child.
let empty = "<ListBucketResult><Prefix></Prefix><KeyCount>0</KeyCount></ListBucketResult>";
assert!(common_prefixes_in(empty, "").is_empty(), "an empty bucket has no children");
// XML-escaped names survive the trip.
let escaped = "<ListBucketResult><Prefix></Prefix>\
<CommonPrefixes><Prefix>a&b/</Prefix></CommonPrefixes></ListBucketResult>";
assert_eq!(common_prefixes_in(escaped, ""), ["a&b"], "the name is unescaped");
}
}