server/lib/src/offsite.rs
//! **Holger's data, off-site** (lane M5, the pattern's off-site stream): the main
//! streams what it must not lose to a store outside the estate's provider, sealed for
//! the operator, on a jera job — and a copy is RESTORABLE into a fresh box.
//!
//! What a copy holds, in Holger's own format ([`FORMAT`]):
//!
//! * the ESTATE — every file of [`crate::replication::ESTATE`] under the data root (the
//! config, audit, attestations, properties, the PROD store, passkeys, entitlements,
//! meters, purge records …), exactly the set a twin pulls over `holger.estate.v1`;
//! * every ARTIFACT of every repository that lists (the sealed `monetize-backup` among
//! them: it is the estate's own data, never a tenant), read through its backend.
//!
//! The plaintext is frames, `[u32 BE len][record]`, a HEAD record first and one
//! zero-length frame at the END: a copy without its end marker is never restored. A
//! record is one line of JSON (what it is, its size and hash) and then its bytes.
//!
//! **Sealed for the operator**: X25519 to the fed `offsite_recipient` (the box needs no
//! secret to seal), ChaCha20-Poly1305 in 1 MiB segments with a final-segment flag and
//! the header as associated data, so a cut copy does not open. Only the operator's
//! private key opens it ([`open`]).
//!
//! **Where it goes** ([`Remote`]): an S3 store (`offsite_r2_endpoint`, `offsite_bucket`,
//! the fed `secret.offsite_r2_*` pair; SigV4, path style) or a mounted directory
//! (`offsite_dir`), under `offsite_prefix`. A copy is shipped when the data changed (or
//! a day passed), and only counts once the store holds every byte.
//!
//! **The way back** ([`restore`]): decrypt, check EVERY record against its hash and the
//! end marker first, then lay the estate files into a FRESH data root and put every
//! artifact through the fresh box's own backends.
//!
//! The run is a jera ChainQueue job ([`start`], house rule), on the MAIN only
//! (`backup_roles = [main]`). Its status is `.holger-offsite/status.json` under the data
//! root (a dot directory: never part of the estate, never what makes a box un-fresh).
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use crate::lock::LockOrRecover as _;
use std::time::Duration;
use base64::Engine as _;
use jera::jobs::{ChainQueue, ChainSpec, JobSink, Priority};
use serde::{Deserialize, Serialize};
use traits::{ArtifactId, RepositoryBackendTrait};
/// The format word of the plaintext and of the sealed header.
pub const FORMAT: &str = "holger-offsite-1";
/// Plaintext bytes per sealed segment.
pub const SEGMENT: usize = 1 << 20;
/// The largest copy a box builds in memory.
pub const MAX_PLAIN_BYTES: usize = 1 << 30;
/// The box-own directory of the copy's status and job ledger (a dot directory).
pub const OWN_DIR: &str = ".holger-offsite";
/// The status of the last attempt.
pub const STATUS_REL: &str = ".holger-offsite/status.json";
/// The job ledger of the copy.
pub const JOBS_REL: &str = ".holger-offsite/jobs";
/// Object names: `<prefix>holger-offsite-<unix ms, 13 digits>-<digest, 16 hex>.holger-offsite`.
pub const OBJECT_PREFIX: &str = "holger-offsite-";
pub const OBJECT_SUFFIX: &str = ".holger-offsite";
/// An unchanged copy is shipped again after a day all the same.
pub const RESEND_AFTER_MS: u64 = 24 * 3600 * 1000;
pub const DEFAULT_EVERY_MINUTES: u64 = 360;
pub const MIN_EVERY_MINUTES: u64 = 15;
/// The feed keys (the installer's feed, as the ONE box's own off-box copy is fed).
pub const FEED_RECIPIENT: &str = "offsite_recipient";
pub const FEED_DIR: &str = "offsite_dir";
pub const FEED_ENDPOINT: &str = "offsite_r2_endpoint";
pub const FEED_BUCKET: &str = "offsite_bucket";
pub const FEED_REGION: &str = "offsite_region";
pub const FEED_PREFIX: &str = "offsite_prefix";
pub const FEED_EVERY: &str = "offsite_every_minutes";
/// The store's key pair: secrets of the feed, never said.
pub const FEED_ACCESS_KEY: &str = "secret.offsite_r2_access_key";
pub const FEED_SECRET_KEY: &str = "secret.offsite_r2_secret_key";
/// The jera job: its ledger kind and the one key every round is submitted under.
pub const KIND: &str = "offsite.copy";
pub const KEY: &str = "offsite-copy";
/// A repository and its backend, as the copy reads and the restore writes them.
pub type Repos = Vec<(String, Arc<dyn RepositoryBackendTrait>)>;
fn b64() -> base64::engine::GeneralPurpose {
base64::engine::general_purpose::STANDARD
}
fn hex(b: &[u8]) -> String {
b.iter().map(|x| format!("{x:02x}")).collect()
}
fn sha256_hex(b: &[u8]) -> String {
hex(ring::digest::digest(&ring::digest::SHA256, b).as_ref())
}
/// A 32-byte X25519 key from its base64; the value is never said.
pub fn key32(text: &str, what: &str) -> Result<[u8; 32], String> {
let raw = b64()
.decode(text.trim())
.map_err(|_| format!("[offsite-key] {what} is not base64 (the value is not shown)"))?;
raw.try_into()
.map_err(|_| format!("[offsite-key] {what} is not 32 bytes (the value is not shown)"))
}
/// **The operator's key pair**, base64 `(private, public)`: the public half is fed as
/// `offsite_recipient`, the private half stays with the operator.
pub fn keygen() -> Result<(String, String), String> {
let mut raw = [0u8; 32];
getrandom::fill(&mut raw).map_err(|e| format!("[offsite-key] no randomness: {e}"))?;
let secret = x25519_dalek::StaticSecret::from(raw);
let public = x25519_dalek::PublicKey::from(&secret);
Ok((
b64().encode(secret.to_bytes()),
b64().encode(public.as_bytes()),
))
}
// ── the plaintext: frames of records ────────────────────────────────────────
/// What one record is.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Record {
/// The first record: the format and when the copy was taken.
Head {
format: String,
taken_unix_ms: u64,
files: u64,
artifacts: u64,
},
/// One estate file, relative to the data root.
Estate {
path: String,
size: u64,
blake3: String,
},
/// One artifact of one repository.
Artifact {
repository: String,
namespace: Option<String>,
name: String,
version: String,
size: u64,
sha256: String,
},
}
fn frame(out: &mut Vec<u8>, rec: &Record, data: &[u8]) -> Result<(), String> {
let line = serde_json::to_vec(rec).map_err(|e| e.to_string())?;
let len = line.len() + 1 + data.len();
if out.len() + 4 + len > MAX_PLAIN_BYTES {
return Err(format!(
"[offsite-too-big] the copy would pass {} MiB: a bigger estate needs a streamed copy",
MAX_PLAIN_BYTES >> 20
));
}
out.extend_from_slice(&(len as u32).to_be_bytes());
out.extend_from_slice(&line);
out.push(b'\n');
out.extend_from_slice(data);
Ok(())
}
/// What [`pack`] made.
#[derive(Debug, Clone, Default)]
pub struct Packed {
pub plain: Vec<u8>,
pub files: u64,
pub artifacts: u64,
/// blake3 of everything after the head: the same data packs to the same digest.
pub digest: String,
/// Repositories the copy could not read whole, with why (said, never dropped).
pub skipped: Vec<String>,
}
/// **Pack the data root and every repository** into the plaintext frames.
pub fn pack(
root: &Path,
repos: &[(String, Arc<dyn RepositoryBackendTrait>)],
taken_unix_ms: u64,
) -> Result<Packed, String> {
let m = crate::estate_wire::manifest(root)?;
let mut body = Vec::new();
let mut p = Packed::default();
for f in &m.files {
let at = root.join(crate::estate_wire::safe_path(&f.path)?);
// The bytes as READ now (a store may have moved since the manifest).
let bytes = std::fs::read(&at).map_err(|e| format!("{}: {e}", at.display()))?;
frame(
&mut body,
&Record::Estate {
path: f.path.clone(),
size: bytes.len() as u64,
blake3: blake3::hash(&bytes).to_hex().to_string(),
},
&bytes,
)?;
p.files += 1;
}
let mut repos: Vec<_> = repos.iter().collect();
repos.sort_by(|a, b| a.0.cmp(&b.0));
for (name, backend) in repos {
let caps = backend.caps();
if caps.group {
continue; // a view over other repositories: its members are copied
}
if !caps.listing {
p.skipped.push(format!(
"{name}: its backend does not list, so its artifacts cannot be copied"
));
continue;
}
let entries = backend
.list(None, usize::MAX)
.map_err(|e| format!("{name}: list: {e:#}"))?;
for e in entries {
let Some(bytes) = backend
.fetch(&e.id)
.map_err(|err| format!("{name}: fetch {}: {err:#}", crate::purge::label(&e.id)))?
else {
p.skipped.push(format!(
"{name}: {} was listed and is gone",
crate::purge::label(&e.id)
));
continue;
};
let rec = Record::Artifact {
repository: name.clone(),
namespace: e.id.namespace.clone(),
name: e.id.name.clone(),
version: e.id.version.clone(),
size: bytes.len() as u64,
sha256: sha256_hex(&bytes),
};
frame(&mut body, &rec, &bytes)?;
p.artifacts += 1;
}
}
p.digest = blake3::hash(&body).to_hex().to_string();
frame(
&mut p.plain,
&Record::Head {
format: FORMAT.into(),
taken_unix_ms,
files: p.files,
artifacts: p.artifacts,
},
&[],
)?;
p.plain.extend_from_slice(&body);
p.plain.extend_from_slice(&[0, 0, 0, 0]);
Ok(p)
}
/// One record read back, its bytes checked.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Entry {
pub record: Record,
pub data: Vec<u8>,
}
/// **Read the plaintext whole, or refuse it**: the head first, every record's bytes
/// against its size and hash, and the end marker last with nothing after it.
pub fn unpack(plain: &[u8]) -> Result<(u64, Vec<Entry>), String> {
let mut at = 0usize;
let mut out = Vec::new();
let mut head: Option<u64> = None;
loop {
if plain.len() - at < 4 {
return Err(
"[offsite-truncated] the copy ends without its end marker: it is not restored"
.into(),
);
}
let n =
u32::from_be_bytes([plain[at], plain[at + 1], plain[at + 2], plain[at + 3]]) as usize;
at += 4;
if n == 0 {
if at != plain.len() {
return Err(format!(
"[offsite-trailing] {} byte(s) follow the end marker",
plain.len() - at
));
}
break;
}
if plain.len() - at < n {
return Err(format!(
"[offsite-truncated] a record of {n} bytes is cut at byte {at}"
));
}
let rec = &plain[at..at + n];
at += n;
let nl = rec
.iter()
.position(|b| *b == b'\n')
.ok_or("[offsite-malformed] a record has no line")?;
let record: Record = serde_json::from_slice(&rec[..nl])
.map_err(|e| format!("[offsite-malformed] a record: {e}"))?;
let data = rec[nl + 1..].to_vec();
match (&record, head) {
(
Record::Head {
format,
taken_unix_ms,
..
},
None,
) => {
if format != FORMAT {
return Err(format!("[offsite-format] format `{format}`, not {FORMAT}"));
}
head = Some(*taken_unix_ms);
continue;
}
(_, None) => return Err("[offsite-malformed] the first record is not the head".into()),
(Record::Head { .. }, Some(_)) => {
return Err("[offsite-malformed] a second head".into())
}
(
Record::Estate {
path,
size,
blake3: want,
},
_,
) => {
crate::estate_wire::safe_path(path)?;
if data.len() as u64 != *size || blake3::hash(&data).to_hex().as_str() != want {
return Err(format!(
"[offsite-forged] the estate file `{path}` does not match its hash"
));
}
}
(
Record::Artifact {
repository,
name,
size,
sha256,
..
},
_,
) => {
if data.len() as u64 != *size || &sha256_hex(&data) != sha256 {
return Err(format!(
"[offsite-forged] the artifact {repository}/{name} does not match its hash"
));
}
}
}
out.push(Entry { record, data });
}
head.map(|t| (t, out))
.ok_or_else(|| "[offsite-malformed] no head".into())
}
/// What [`restore`] laid down.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Restored {
pub taken_unix_ms: u64,
pub files: u64,
pub artifacts: u64,
/// One line per artifact that could not be put (the restore is then NOT whole).
pub failed: Vec<String>,
}
/// **Restore a copy into a FRESH box**: the whole copy is checked first ([`unpack`]),
/// a data root that already holds estate files is refused, then the estate is laid and
/// every artifact put through the fresh box's own backend for its repository.
pub fn restore(
plain: &[u8],
root: &Path,
repos: &[(String, Arc<dyn RepositoryBackendTrait>)],
) -> Result<Restored, String> {
let (taken_unix_ms, entries) = unpack(plain)?;
if let Some(f) = crate::estate_wire::manifest(root)?.files.first() {
return Err(format!(
"[offsite-not-fresh] {} already holds `{}`: a copy is restored into a fresh box only",
root.display(),
f.path
));
}
let mut r = Restored {
taken_unix_ms,
..Restored::default()
};
for e in entries {
match e.record {
Record::Estate { path, size, blake3 } => {
crate::estate_wire::write_fetched(
root,
&crate::estate_wire::EstateFile { path, size, blake3 },
&e.data,
)?;
r.files += 1;
}
Record::Artifact {
repository,
namespace,
name,
version,
..
} => {
let id = ArtifactId {
namespace,
name,
version,
};
match repos.iter().find(|(n, _)| *n == repository) {
None => r.failed.push(format!(
"{repository}: this box serves no such repository, so {} is not restored",
crate::purge::label(&id)
)),
Some((_, b)) => match b.put(&id, &e.data) {
Ok(()) => r.artifacts += 1,
Err(err) => r.failed.push(format!(
"{repository}: {}: {err:#}",
crate::purge::label(&id)
)),
},
}
}
Record::Head { .. } => {}
}
}
Ok(r)
}
// ── sealing ─────────────────────────────────────────────────────────────────
/// The sealed copy's header (a line of JSON, the associated data of every segment).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Header {
pub format: String,
/// base64 X25519: whom it is for, and the sender's ephemeral half.
pub recipient: String,
pub ephemeral: String,
}
fn segment_key(
shared: &[u8; 32],
ephemeral: &[u8; 32],
recipient: &[u8; 32],
) -> ring::aead::LessSafeKey {
let mut ctx = ring::digest::Context::new(&ring::digest::SHA256);
for part in [FORMAT.as_bytes(), shared, ephemeral, recipient] {
ctx.update(part);
}
let k = ctx.finish();
ring::aead::LessSafeKey::new(
ring::aead::UnboundKey::new(&ring::aead::CHACHA20_POLY1305, k.as_ref())
.expect("a 32-byte key"),
)
}
fn nonce(counter: u64, last: bool) -> ring::aead::Nonce {
let mut n = [0u8; 12];
n[0] = u8::from(last);
n[4..].copy_from_slice(&counter.to_be_bytes());
ring::aead::Nonce::assume_unique_for_key(n)
}
/// **Seal** `plain` for `recipient`: the header line, then `[u32 BE len][ciphertext]`
/// segments, the last one flagged.
pub fn seal(recipient: &[u8; 32], plain: &[u8]) -> Result<Vec<u8>, String> {
let mut raw = [0u8; 32];
getrandom::fill(&mut raw).map_err(|e| format!("[offsite-seal] no randomness: {e}"))?;
let secret = x25519_dalek::StaticSecret::from(raw);
let ephemeral = x25519_dalek::PublicKey::from(&secret);
let shared = secret.diffie_hellman(&x25519_dalek::PublicKey::from(*recipient));
let h = Header {
format: FORMAT.into(),
recipient: b64().encode(recipient),
ephemeral: b64().encode(ephemeral.as_bytes()),
};
let line = serde_json::to_string(&h).map_err(|e| e.to_string())? + "\n";
let key = segment_key(shared.as_bytes(), ephemeral.as_bytes(), recipient);
let mut out = line.clone().into_bytes();
let chunks: Vec<&[u8]> = if plain.is_empty() {
vec![&[][..]]
} else {
plain.chunks(SEGMENT).collect()
};
for (i, c) in chunks.iter().enumerate() {
let mut buf = c.to_vec();
key.seal_in_place_append_tag(
nonce(i as u64, i + 1 == chunks.len()),
ring::aead::Aad::from(line.as_bytes()),
&mut buf,
)
.map_err(|_| "[offsite-seal] sealing a segment failed".to_string())?;
out.extend_from_slice(&(buf.len() as u32).to_be_bytes());
out.extend_from_slice(&buf);
}
Ok(out)
}
/// **Open** a sealed copy with the operator's private key: every segment authenticated,
/// a copy without its final segment refused, as is one sealed for another key.
pub fn open(private: &[u8; 32], sealed: &[u8]) -> Result<Vec<u8>, String> {
let nl = sealed
.iter()
.position(|&b| b == b'\n')
.ok_or("[offsite-not-a-copy] no header line")?;
let line = &sealed[..=nl];
let h: Header = serde_json::from_slice(&sealed[..nl])
.map_err(|e| format!("[offsite-not-a-copy] the header: {e}"))?;
if h.format != FORMAT {
return Err(format!(
"[offsite-not-a-copy] format `{}`, not {FORMAT}",
h.format
));
}
let secret = x25519_dalek::StaticSecret::from(*private);
let recipient = key32(&h.recipient, "the copy's recipient")?;
if x25519_dalek::PublicKey::from(&secret).as_bytes() != &recipient {
return Err("[offsite-not-yours] this copy is sealed for another key".into());
}
let ephemeral = key32(&h.ephemeral, "the copy's ephemeral key")?;
let shared = secret.diffie_hellman(&x25519_dalek::PublicKey::from(ephemeral));
let key = segment_key(shared.as_bytes(), &ephemeral, &recipient);
let (mut at, mut counter, mut plain) = (nl + 1, 0u64, Vec::new());
loop {
if sealed.len() - at < 4 {
return Err(format!(
"[offsite-truncated] the copy ends after segment {counter} with no final segment"
));
}
let n = u32::from_be_bytes([sealed[at], sealed[at + 1], sealed[at + 2], sealed[at + 3]])
as usize;
at += 4;
if sealed.len() - at < n {
return Err(format!("[offsite-truncated] segment {counter} is cut"));
}
let mut buf = sealed[at..at + n].to_vec();
at += n;
let last = at == sealed.len();
match key.open_in_place(nonce(counter, last), ring::aead::Aad::from(line), &mut buf) {
Ok(p) => plain.extend_from_slice(p),
Err(_) => return Err(format!("[offsite-forged] segment {counter} does not authenticate (a changed byte, a lost final segment, or another key)")),
}
if last {
return Ok(plain);
}
counter += 1;
}
}
// ── the store ───────────────────────────────────────────────────────────────
/// **Where the copy goes**: written once, its size read back, read for a restore.
pub trait Remote: Send + Sync {
/// Where, for the lines that say so (never a credential).
fn describe(&self) -> String;
fn put(&self, key: &str, bytes: &[u8]) -> Result<(), String>;
/// The size held under `key`, `None` when nothing is.
fn size(&self, key: &str) -> Result<Option<u64>, String>;
fn get(&self, key: &str) -> Result<Option<Vec<u8>>, String>;
/// Every key under `prefix`.
fn list(&self, prefix: &str) -> Result<Vec<String>, String>;
}
/// **A mounted directory** (another disk, a synced directory): objects are files.
pub struct DirRemote(pub PathBuf);
impl DirRemote {
fn at(&self, key: &str) -> Result<PathBuf, String> {
if key
.split('/')
.any(|c| c.is_empty() || c == "." || c == "..")
{
return Err(format!("[offsite-key] `{key}` is not a plain object key"));
}
Ok(self.0.join(key))
}
}
impl Remote for DirRemote {
fn describe(&self) -> String {
format!("the directory {}", self.0.display())
}
fn put(&self, key: &str, bytes: &[u8]) -> Result<(), String> {
let at = self.at(key)?;
if let Some(d) = at.parent() {
std::fs::create_dir_all(d).map_err(|e| format!("{}: {e}", d.display()))?;
}
let tmp = at.with_extension("part");
{
use std::io::Write as _;
let mut w =
std::fs::File::create(&tmp).map_err(|e| format!("{}: {e}", tmp.display()))?;
w.write_all(bytes)
.and_then(|_| w.sync_all())
.map_err(|e| format!("{}: {e}", tmp.display()))?;
}
std::fs::rename(&tmp, &at).map_err(|e| format!("{}: {e}", at.display()))
}
fn size(&self, key: &str) -> Result<Option<u64>, String> {
Ok(std::fs::metadata(self.at(key)?).ok().map(|m| m.len()))
}
fn get(&self, key: &str) -> Result<Option<Vec<u8>>, String> {
match std::fs::read(self.at(key)?) {
Ok(b) => Ok(Some(b)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(format!("{key}: {e}")),
}
}
fn list(&self, prefix: &str) -> Result<Vec<String>, String> {
let (dir, start) = match prefix.rsplit_once('/') {
Some((d, s)) => (self.0.join(d), (format!("{d}/"), s.to_string())),
None => (self.0.clone(), (String::new(), prefix.to_string())),
};
let mut out: Vec<String> = match std::fs::read_dir(&dir) {
Ok(rd) => rd
.flatten()
.filter_map(|e| e.file_name().into_string().ok())
.filter(|n| n.starts_with(&start.1) && !n.ends_with(".part"))
.map(|n| format!("{}{n}", start.0))
.collect(),
Err(_) => Vec::new(),
};
out.sort();
Ok(out)
}
}
/// Pure: an S3 key, URI-encoded as SigV4 wants it (`/` kept).
pub fn uri_encode(key: &str) -> String {
key.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-._~/".contains(&b) {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect()
}
/// Pure: `YYYYMMDDTHHMMSSZ` of a unix time.
pub fn amz_date(unix: u64) -> String {
let days = (unix / 86_400) as i64;
let s = unix % 86_400;
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z - era * 146_097;
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = yoe + era * 400 + i64::from(m <= 2);
format!(
"{y:04}{m:02}{d:02}T{:02}{:02}{:02}Z",
s / 3600,
s % 3600 / 60,
s % 60
)
}
/// One request to sign (AWS Signature Version 4).
pub struct SigV4<'a> {
pub method: &'a str,
/// Already URI-encoded.
pub path: &'a str,
pub query: &'a str,
/// Lower-case names; every one is signed.
pub headers: &'a [(&'a str, &'a str)],
pub payload_sha256: &'a str,
pub amz_date: &'a str,
pub region: &'a str,
pub service: &'a str,
}
fn hmac(key: &[u8], msg: &[u8]) -> Vec<u8> {
ring::hmac::sign(&ring::hmac::Key::new(ring::hmac::HMAC_SHA256, key), msg)
.as_ref()
.to_vec()
}
/// Pure: the `Authorization` header of `req`.
pub fn authorization(req: &SigV4<'_>, access_key: &str, secret_key: &str) -> String {
let mut hs: Vec<(String, String)> = req
.headers
.iter()
.map(|(n, v)| (n.to_ascii_lowercase(), v.trim().to_string()))
.collect();
hs.sort();
let canonical: String = hs.iter().map(|(n, v)| format!("{n}:{v}\n")).collect();
let signed = hs
.iter()
.map(|(n, _)| n.as_str())
.collect::<Vec<_>>()
.join(";");
let creq = format!(
"{}\n{}\n{}\n{canonical}\n{signed}\n{}",
req.method, req.path, req.query, req.payload_sha256
);
let date = req.amz_date.get(..8).unwrap_or(req.amz_date);
let scope = format!("{date}/{}/{}/aws4_request", req.region, req.service);
let to_sign = format!(
"AWS4-HMAC-SHA256\n{}\n{scope}\n{}",
req.amz_date,
sha256_hex(creq.as_bytes())
);
let mut k = hmac(format!("AWS4{secret_key}").as_bytes(), date.as_bytes());
for part in [req.region, req.service, "aws4_request"] {
k = hmac(&k, part.as_bytes());
}
let sig = hex(&hmac(&k, to_sign.as_bytes()));
format!(
"AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed}, Signature={sig}"
)
}
/// **An S3 store** (Cloudflare R2 and the like), path style, SigV4-signed.
#[cfg(feature = "online")]
pub struct S3Remote {
agent: ureq::Agent,
endpoint: String,
host: String,
bucket: String,
region: String,
access_key: String,
secret_key: String,
}
#[cfg(feature = "online")]
impl S3Remote {
pub fn new(
endpoint: &str,
bucket: &str,
region: &str,
access_key: &str,
secret_key: &str,
) -> S3Remote {
// A box behind its front (the lean estate) reaches the store through the front's door.
let agent = ureq::Agent::config_builder()
.http_status_as_error(false)
.timeout_global(Some(Duration::from_secs(1800)))
.proxy(crate::egress::ureq_proxy())
.build()
.new_agent();
let endpoint = endpoint.trim_end_matches('/').to_string();
let host = endpoint
.split_once("://")
.map(|(_, h)| h)
.unwrap_or(&endpoint)
.to_string();
S3Remote {
agent,
endpoint,
host,
bucket: bucket.into(),
region: region.into(),
access_key: access_key.into(),
secret_key: secret_key.into(),
}
}
fn signed(&self, method: &str, path: &str, query: &str, payload: &str) -> (String, String) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or_default();
let date = amz_date(now);
let req = SigV4 {
method,
path,
query,
headers: &[
("host", &self.host),
("x-amz-content-sha256", payload),
("x-amz-date", &date),
],
payload_sha256: payload,
amz_date: &date,
region: &self.region,
service: "s3",
};
(
authorization(&req, &self.access_key, &self.secret_key),
date,
)
}
fn call(
&self,
method: &str,
key: &str,
query: &str,
body: &[u8],
) -> Result<(u16, Vec<u8>, Option<u64>), String> {
crate::egress::check("offsite copy", &self.endpoint).map_err(|e| e.to_string())?;
let path = if key.is_empty() {
format!("/{}", self.bucket)
} else {
format!("/{}/{}", self.bucket, uri_encode(key))
};
let payload = sha256_hex(body);
let (auth, date) = self.signed(method, &path, query, &payload);
let url = if query.is_empty() {
format!("{}{path}", self.endpoint)
} else {
format!("{}{path}?{query}", self.endpoint)
};
let r = match method {
"PUT" => self
.agent
.put(&url)
.header("x-amz-content-sha256", &payload)
.header("x-amz-date", &date)
.header("authorization", &auth)
.send(body),
"HEAD" => self
.agent
.head(&url)
.header("x-amz-content-sha256", &payload)
.header("x-amz-date", &date)
.header("authorization", &auth)
.call(),
_ => self
.agent
.get(&url)
.header("x-amz-content-sha256", &payload)
.header("x-amz-date", &date)
.header("authorization", &auth)
.call(),
};
let mut r = r.map_err(|e| format!("[offsite-store] {method} {}: {e}", self.describe()))?;
let status = r.status().as_u16();
let len = r
.headers()
.get("content-length")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok());
let body = if method == "HEAD" {
Vec::new()
} else {
r.body_mut()
.with_config()
.limit(MAX_PLAIN_BYTES as u64 * 2)
.read_to_vec()
.map_err(|e| format!("[offsite-store] the body: {e}"))?
};
Ok((status, body, len))
}
}
/// `HTTP <status> <Code>` of a refused request: the S3 error's code, never its body.
#[cfg(feature = "online")]
fn refused(status: u16, body: &[u8]) -> String {
let text = String::from_utf8_lossy(body);
let code = text
.split_once("<Code>")
.and_then(|(_, t)| t.split_once("</Code>"))
.map(|(c, _)| {
c.chars()
.filter(|c| c.is_ascii_alphanumeric())
.take(64)
.collect::<String>()
})
.unwrap_or_default();
format!("[offsite-store] HTTP {status} {code}")
.trim_end()
.to_string()
}
#[cfg(feature = "online")]
impl Remote for S3Remote {
fn describe(&self) -> String {
format!("the bucket {} at {}", self.bucket, self.host)
}
fn put(&self, key: &str, bytes: &[u8]) -> Result<(), String> {
let (s, b, _) = self.call("PUT", key, "", bytes)?;
if (200..300).contains(&s) {
Ok(())
} else {
Err(refused(s, &b))
}
}
fn size(&self, key: &str) -> Result<Option<u64>, String> {
match self.call("HEAD", key, "", &[])? {
(404, _, _) => Ok(None),
(s, _, Some(n)) if (200..300).contains(&s) => Ok(Some(n)),
(s, b, _) => Err(refused(s, &b)),
}
}
fn get(&self, key: &str) -> Result<Option<Vec<u8>>, String> {
match self.call("GET", key, "", &[])? {
(404, _, _) => Ok(None),
(s, b, _) if (200..300).contains(&s) => Ok(Some(b)),
(s, b, _) => Err(refused(s, &b)),
}
}
fn list(&self, prefix: &str) -> Result<Vec<String>, String> {
let query = format!(
"list-type=2&prefix={}",
uri_encode(prefix).replace('/', "%2F")
);
let (s, b, _) = self.call("GET", "", &query, &[])?;
if !(200..300).contains(&s) {
return Err(refused(s, &b));
}
Ok(keys_of_listing(&String::from_utf8_lossy(&b)))
}
}
/// Pure: the `<Key>` values of a ListObjectsV2 answer, sorted.
pub fn keys_of_listing(xml: &str) -> Vec<String> {
let mut out: Vec<String> = xml
.split("<Key>")
.skip(1)
.filter_map(|t| t.split_once("</Key>").map(|(k, _)| k.replace("&", "&")))
.collect();
out.sort();
out
}
// ── the config, from the feed ───────────────────────────────────────────────
/// Where the copy goes, as fed.
#[derive(Clone, PartialEq, Eq)]
pub enum Target {
Dir(PathBuf),
S3 {
endpoint: String,
bucket: String,
region: String,
access_key: String,
secret_key: String,
},
}
impl std::fmt::Debug for Target {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Target::Dir(d) => write!(f, "Dir({})", d.display()),
Target::S3 {
endpoint,
bucket,
region,
..
} => write!(
f,
"S3 {{ endpoint: {endpoint}, bucket: {bucket}, region: {region}, keys: <secret> }}"
),
}
}
}
/// **The off-site copy's settings**, from the installer's feed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Config {
pub recipient: [u8; 32],
pub target: Target,
/// `holger/<estate>/`, always ending in `/` (or empty).
pub prefix: String,
pub every: Duration,
}
impl Config {
/// `Ok(None)`: no `offsite_recipient` fed — the copy is off and the box says so.
/// A recipient with no store, or half a key pair, is refused by name (no value said).
pub fn of(feed: &BTreeMap<String, String>) -> Result<Option<Config>, String> {
let get = |k: &str| {
feed.get(k)
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
};
let Some(r) = get(FEED_RECIPIENT) else {
return Ok(None);
};
let recipient = key32(&r, FEED_RECIPIENT)?;
let target = match (get(FEED_DIR), get(FEED_ENDPOINT)) {
(Some(_), Some(_)) => return Err(format!("[offsite-config] both `{FEED_DIR}` and `{FEED_ENDPOINT}` are fed: name one store")),
(Some(d), None) => Target::Dir(PathBuf::from(d)),
(None, Some(endpoint)) => {
if !endpoint.starts_with("https://") && !endpoint.starts_with("http://127.0.0.1") && !endpoint.starts_with("http://localhost") {
return Err(format!("[offsite-config] `{FEED_ENDPOINT}` must be https:// (a copy never leaves in the clear)"));
}
let (Some(access_key), Some(secret_key)) = (get(FEED_ACCESS_KEY), get(FEED_SECRET_KEY)) else {
return Err(format!("[offsite-config] `{FEED_ENDPOINT}` is fed without both `{FEED_ACCESS_KEY}` and `{FEED_SECRET_KEY}` (no value is shown)"));
};
let bucket = get(FEED_BUCKET).ok_or_else(|| format!("[offsite-config] `{FEED_ENDPOINT}` is fed without `{FEED_BUCKET}`"))?;
Target::S3 { endpoint, bucket, region: get(FEED_REGION).unwrap_or_else(|| "auto".into()), access_key, secret_key }
}
(None, None) => return Err(format!("[offsite-config] `{FEED_RECIPIENT}` is fed but no store: feed `{FEED_DIR}` or `{FEED_ENDPOINT}`")),
};
let mut prefix = get(FEED_PREFIX).unwrap_or_else(|| "holger/".into());
if prefix.split('/').any(|c| c == "..") {
return Err(format!("[offsite-config] `{FEED_PREFIX}` may not climb"));
}
if !prefix.is_empty() && !prefix.ends_with('/') {
prefix.push('/');
}
let minutes = match get(FEED_EVERY) {
None => DEFAULT_EVERY_MINUTES,
Some(v) => v
.parse::<u64>()
.map_err(|_| format!("[offsite-config] `{FEED_EVERY}` = {v:?} is not minutes"))?
.max(MIN_EVERY_MINUTES),
};
Ok(Some(Config {
recipient,
target,
prefix,
every: Duration::from_secs(minutes * 60),
}))
}
/// The store this config names.
pub fn remote(&self) -> Result<Box<dyn Remote>, String> {
match &self.target {
Target::Dir(d) => Ok(Box::new(DirRemote(d.clone()))),
#[cfg(feature = "online")]
Target::S3 { endpoint, bucket, region, access_key, secret_key } => Ok(Box::new(S3Remote::new(endpoint, bucket, region, access_key, secret_key))),
#[cfg(not(feature = "online"))]
Target::S3 { .. } => Err("[offsite-config] this build has no outbound HTTP (`online` is off): feed `offsite_dir` instead".into()),
}
}
}
// ── one attempt ─────────────────────────────────────────────────────────────
/// What the last attempts did ([`STATUS_REL`]). Never a credential.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Status {
pub last_attempt_unix_ms: u64,
pub last_ok_unix_ms: u64,
pub last_object: String,
pub last_digest: String,
pub last_bytes: u64,
pub failures_in_a_row: u32,
pub last_error: String,
}
/// The status under `root` (default when there is none yet).
pub fn status(root: &Path) -> Status {
std::fs::read(root.join(STATUS_REL))
.ok()
.and_then(|b| serde_json::from_slice(&b).ok())
.unwrap_or_default()
}
fn write_status(root: &Path, st: &Status) {
let _ = std::fs::create_dir_all(root.join(OWN_DIR));
if let Ok(b) = serde_json::to_vec_pretty(st) {
let _ = std::fs::write(root.join(STATUS_REL), b);
}
}
/// Pure: the object key of a copy — sortable by time, named for its content.
pub fn object_key(prefix: &str, taken_unix_ms: u64, digest: &str) -> String {
format!(
"{prefix}{OBJECT_PREFIX}{taken_unix_ms:013}-{}{OBJECT_SUFFIX}",
&digest[..digest.len().min(16)]
)
}
/// What one attempt did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
Shipped {
object: String,
bytes: u64,
files: u64,
artifacts: u64,
},
/// The data is the copy last shipped, within [`RESEND_AFTER_MS`].
Unchanged { object: String },
}
/// **One copy, now**: pack, seal, put, read the size back; the status says what happened.
pub fn backup_once(
root: &Path,
repos: &[(String, Arc<dyn RepositoryBackendTrait>)],
cfg: &Config,
remote: &dyn Remote,
now_ms: u64,
) -> Result<Outcome, String> {
let mut st = status(root);
st.last_attempt_unix_ms = now_ms;
let out = (|| -> Result<Outcome, String> {
let p = pack(root, repos, now_ms)?;
if p.digest == st.last_digest && now_ms.saturating_sub(st.last_ok_unix_ms) < RESEND_AFTER_MS
{
return Ok(Outcome::Unchanged {
object: st.last_object.clone(),
});
}
let sealed = seal(&cfg.recipient, &p.plain)?;
let key = object_key(&cfg.prefix, now_ms, &p.digest);
remote.put(&key, &sealed)?;
match remote.size(&key)? {
Some(n) if n == sealed.len() as u64 => {}
other => return Err(format!("[offsite-short] {} holds {other:?} bytes of {key}, {} were sent: the copy does not count", remote.describe(), sealed.len())),
}
for s in &p.skipped {
log::warn!("offsite: {s}");
}
st.last_digest = p.digest.clone();
Ok(Outcome::Shipped {
object: key,
bytes: sealed.len() as u64,
files: p.files,
artifacts: p.artifacts,
})
})();
match &out {
Ok(Outcome::Shipped { object, bytes, .. }) => {
st.last_ok_unix_ms = now_ms;
st.last_object = object.clone();
st.last_bytes = *bytes;
st.failures_in_a_row = 0;
st.last_error.clear();
}
Ok(Outcome::Unchanged { .. }) => st.failures_in_a_row = 0,
Err(e) => {
st.failures_in_a_row += 1;
st.last_error = e.clone();
}
}
write_status(root, &st);
out
}
/// The newest copy under `prefix`, by name (the names sort by time).
pub fn latest(remote: &dyn Remote, prefix: &str) -> Result<Option<String>, String> {
let start = format!("{prefix}{OBJECT_PREFIX}");
Ok(remote
.list(&start)?
.into_iter()
.filter(|k| k.ends_with(OBJECT_SUFFIX))
.max())
}
/// **The way back, whole**: the newest copy under `prefix`, opened with the operator's
/// private key and restored into the fresh box at `root`.
pub fn restore_latest(
remote: &dyn Remote,
prefix: &str,
private: &[u8; 32],
root: &Path,
repos: &[(String, Arc<dyn RepositoryBackendTrait>)],
) -> Result<(String, Restored), String> {
let key = latest(remote, prefix)?.ok_or_else(|| {
format!(
"[offsite-none] {} holds no copy under `{prefix}`",
remote.describe()
)
})?;
let sealed = remote
.get(&key)?
.ok_or_else(|| format!("[offsite-none] {key} went away"))?;
let plain = open(private, &sealed)?;
Ok((key, restore(&plain, root, repos)?))
}
/// Where a run reads the repositories from, asked anew every run.
pub type RepoSource = Arc<dyn Fn() -> Repos + Send + Sync>;
/// Every repository of a wired holger with its backend, at boot.
pub fn repos_of(holger: &crate::Holger) -> Repos {
holger
.repositories
.iter()
.filter_map(|r| {
r.backend_repository
.clone()
.map(|b| (r.ron_name.clone(), b))
})
.collect()
}
/// **The repositories as each run finds them**: the boot's, and every repository a door
/// serves now — one made at run time (a paid signup's, lane M4b) is copied from the next
/// run on, with no restart.
pub fn repo_source(holger: &crate::Holger) -> RepoSource {
let boot = repos_of(holger);
let live = holger.live_routes_cell.get().cloned();
Arc::new(move || {
let mut v = boot.clone();
if let Some(l) = &live {
for (_, table) in &l.doors {
for (name, b) in table.all_repos() {
if !v.iter().any(|(n, _)| *n == name) {
v.push((name, b));
}
}
}
}
v
})
}
// ── the jera job ────────────────────────────────────────────────────────────
/// Everything one copy needs.
pub struct Duty {
pub root: PathBuf,
pub repos: RepoSource,
pub cfg: Config,
pub remote: Box<dyn Remote>,
}
impl Duty {
/// **One copy**, and what it did (the job row's detail).
pub fn round(&self) -> anyhow::Result<serde_json::Value> {
let now = crate::clock::now_unix_ms().max(0) as u64;
match backup_once(
&self.root,
&(self.repos)(),
&self.cfg,
self.remote.as_ref(),
now,
) {
Ok(Outcome::Shipped {
object,
bytes,
files,
artifacts,
}) => {
log::info!("offsite: a copy of {files} estate file(s) and {artifacts} artifact(s) ({bytes} B sealed) is in {} as {object}", self.remote.describe());
Ok(
serde_json::json!({"shipped": object, "bytes": bytes, "files": files, "artifacts": artifacts}),
)
}
Ok(Outcome::Unchanged { object }) => Ok(serde_json::json!({"unchanged": object})),
Err(e) => {
log::error!("offsite: the copy failed: {e}");
Err(anyhow::anyhow!(e))
}
}
}
}
/// The spec of one copy.
pub fn spec(generation: i64) -> ChainSpec {
ChainSpec::new(
KIND,
KEY,
"offsite",
generation,
"holger",
serde_json::json!({}),
)
.with_target("holger's off-site copy")
.with_priority(Priority::Low)
}
/// **Run the copy for the life of the box** on `rt`: a ChainQueue of its own and the
/// ticker that submits one copy every `cfg.every` (the first a minute after start).
pub fn start(rt: &tokio::runtime::Handle, sink: JobSink, duty: Arc<Duty>) -> Arc<ChainQueue> {
let q = ChainQueue::new(sink);
let d = Arc::clone(&duty);
// A32-14, stays a task: the off-box copy's jera worker.
rt.spawn(Arc::clone(&q).run(move |_spec| d.round()));
let tick = Arc::clone(&q);
let every = duty.cfg.every;
// A32-14, stays a task: the off-box copy's ticker (it submits the jera rounds).
rt.spawn(async move {
tokio::time::sleep(Duration::from_secs(60).min(every)).await;
let mut generation = 0i64;
loop {
generation += 1;
let _ = tick.submit(spec(generation));
tokio::time::sleep(every).await;
}
});
q
}
/// **Start the off-site copy on the box's feed bus** (A7 row 8): the copy starts the moment
/// a feed names `offsite_recipient` and its store, and follows a later change of either from
/// the next copy on, without a restart. `is_twin` is asked at every copy: a box ELECTED twin
/// after its start copies nothing (only the main copies, `backup_roles = [main]`).
pub fn start_on(
feeds: &dyn crate::late::Feeds,
root: PathBuf,
repos: RepoSource,
is_twin: Arc<dyn Fn() -> bool + Send + Sync>,
) -> Result<(), String> {
let rt = tokio::runtime::Handle::try_current()
.map_err(|_| "offsite: no runtime to run the copy on".to_string())?;
let current: Arc<std::sync::Mutex<Option<Arc<Duty>>>> = Arc::new(std::sync::Mutex::new(None));
let queue: Arc<std::sync::OnceLock<Arc<ChainQueue>>> = Arc::new(std::sync::OnceLock::new());
crate::late::on_change(
feeds,
"offsite: the recipient and its store",
Config::of,
move |cfg: &Result<Option<Config>, String>| {
let cfg = match cfg {
Ok(Some(c)) => c.clone(),
Ok(None) => {
if current.lock_or_recover().is_none() {
log::info!("offsite: no `{FEED_RECIPIENT}` is fed: nothing of this box is copied off-site");
}
return;
}
Err(why) => return log::error!("offsite: the fed off-site copy is refused: {why}"),
};
let remote = match cfg.remote() {
Ok(r) => r,
Err(why) => {
return log::error!("offsite: the fed off-site store is refused: {why}")
}
};
let was = current.lock_or_recover().replace(Arc::new(Duty {
root: root.clone(),
repos: Arc::clone(&repos),
cfg: cfg.clone(),
remote,
}));
log::info!(
"offsite: every {} min under `{}`, sealed for the operator{}",
cfg.every.as_secs() / 60,
cfg.prefix,
if was.is_some() {
" (the feed changed it: from the next copy on)"
} else {
""
}
);
if queue.get().is_some() {
return;
}
let jobs = match jera::jobs::JobStore::open(&root.join(JOBS_REL)) {
Ok(j) => j,
Err(e) => return log::error!("offsite: the job ledger: {e}"),
};
let q = ChainQueue::new(jobs.sink());
let (now, twin) = (Arc::clone(¤t), Arc::clone(&is_twin));
// A32-14, stays a task: the twin copy's jera worker.
rt.spawn(Arc::clone(&q).run(move |_spec| {
if twin() {
return Ok(serde_json::json!({ "shipped": false, "why": "this box is a twin: only the main copies off-site" }));
}
let duty = now.lock_or_recover().clone();
match duty {
Some(d) => d.round(),
None => Ok(serde_json::json!({ "shipped": false, "why": "no off-site copy is fed" })),
}
}));
let (tick, now) = (Arc::clone(&q), Arc::clone(¤t));
// A32-14, stays a task: the twin copy's ticker (it submits the jera rounds).
rt.spawn(async move {
let every = |n: &Arc<std::sync::Mutex<Option<Arc<Duty>>>>| {
n.lock_or_recover()
.as_ref()
.map(|d| d.cfg.every)
.unwrap_or(Duration::from_secs(DEFAULT_EVERY_MINUTES * 60))
};
tokio::time::sleep(Duration::from_secs(60).min(every(&now))).await;
let mut generation = 0i64;
loop {
generation += 1;
let _ = tick.submit(spec(generation));
tokio::time::sleep(every(&now)).await;
}
});
let _ = queue.set(q);
},
);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// The feed's words: off without a recipient, refused by name when half fed, and a
/// credential never shown.
#[test]
fn the_feed_names_the_copy_or_it_is_off() {
let feed = |pairs: &[(&str, &str)]| {
pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect::<BTreeMap<_, _>>()
};
assert_eq!(Config::of(&feed(&[])).unwrap(), None, "no recipient: off");
let (_, public) = keygen().unwrap();
assert!(Config::of(&feed(&[(FEED_RECIPIENT, &public)]))
.unwrap_err()
.contains("no store"));
let half = Config::of(&feed(&[
(FEED_RECIPIENT, &public),
(FEED_ENDPOINT, "https://acct.r2.cloudflarestorage.com"),
(FEED_BUCKET, "b"),
(FEED_ACCESS_KEY, "AKIA-SECRET-VALUE"),
]))
.unwrap_err();
assert!(
half.contains(FEED_SECRET_KEY) && !half.contains("AKIA-SECRET-VALUE"),
"{half}"
);
assert!(Config::of(&feed(&[
(FEED_RECIPIENT, &public),
(FEED_ENDPOINT, "http://r2.example"),
(FEED_BUCKET, "b"),
(FEED_ACCESS_KEY, "a"),
(FEED_SECRET_KEY, "s")
]))
.unwrap_err()
.contains("https://"));
let c = Config::of(&feed(&[
(FEED_RECIPIENT, &public),
(FEED_ENDPOINT, "https://acct.r2.cloudflarestorage.com"),
(FEED_BUCKET, "b"),
(FEED_ACCESS_KEY, "AKIA-SECRET-VALUE"),
(FEED_SECRET_KEY, "TOP-SECRET"),
(FEED_PREFIX, "holger/test2"),
(FEED_EVERY, "1"),
]))
.unwrap()
.unwrap();
assert_eq!(
(c.prefix.as_str(), c.every),
("holger/test2/", Duration::from_secs(MIN_EVERY_MINUTES * 60))
);
let shown = format!("{c:?}");
assert!(
!shown.contains("AKIA-SECRET-VALUE") && !shown.contains("TOP-SECRET"),
"{shown}"
);
let d = Config::of(&feed(&[
(FEED_RECIPIENT, &public),
(FEED_DIR, "/mnt/offsite"),
]))
.unwrap()
.unwrap();
assert_eq!(
(d.target, d.prefix.as_str(), d.every),
(
Target::Dir("/mnt/offsite".into()),
"holger/",
Duration::from_secs(DEFAULT_EVERY_MINUTES * 60)
)
);
assert!(Config::of(&feed(&[(FEED_RECIPIENT, "not base64!")]))
.unwrap_err()
.contains("[offsite-key]"));
}
/// The AWS SigV4 reference (S3 docs, "GET Object" example).
#[test]
fn the_signature_is_the_aws_reference_one() {
let empty = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
let req = SigV4 {
method: "GET",
path: "/test.txt",
query: "",
headers: &[
("host", "examplebucket.s3.amazonaws.com"),
("range", "bytes=0-9"),
("x-amz-content-sha256", empty),
("x-amz-date", "20130524T000000Z"),
],
payload_sha256: empty,
amz_date: "20130524T000000Z",
region: "us-east-1",
service: "s3",
};
assert_eq!(
authorization(&req, "AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"),
"AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, SignedHeaders=host;range;x-amz-content-sha256;x-amz-date, Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"
);
assert_eq!(amz_date(1_369_353_600), "20130524T000000Z");
assert_eq!(
uri_encode("holger/x/a b+c~.holger-offsite"),
"holger/x/a%20b%2Bc~.holger-offsite"
);
assert_eq!(keys_of_listing("<ListBucketResult><Contents><Key>p/b</Key></Contents><Contents><Key>p/a</Key></Contents></ListBucketResult>"), ["p/a", "p/b"]);
}
}