crates/upcloud-api/src/wire.rs
//! **The ONE wire.** Every request to the account and every request to a
//! `mock-upcloud` is built here, by the same code, from the same
//! [`crate::body`] spellings — only the base differs, and the base comes from an
//! [`Endpoint`] the caller could not have forged. That is what makes a run
//! against the mock a test of the code that runs against the account: the URL,
//! the body, the headers and the retry policy are not a parallel copy.
use std::path::Path;
use std::time::Duration;
use serde_json::Value;
use crate::net::{self, Budget};
use crate::over::{Call, Exchange, Method, Over};
use crate::{redact_upload_url, Endpoint, Reply, UpCloudApi, ACCOUNT_BASE};
/// A read that met the gateway (502/503/504) is asked again for up to this long
/// in all, the waits doubling from [`GATEWAY_FIRST`] up to [`GATEWAY_CAP`].
pub const GATEWAY_BUDGET: Duration = Duration::from_secs(180);
pub const GATEWAY_FIRST: Duration = Duration::from_secs(2);
pub const GATEWAY_CAP: Duration = Duration::from_secs(20);
/// How the wire authenticates. The account's clients use a token; monetize's
/// plugin can also be configured with a sub-account's username and password.
#[derive(Clone)]
pub enum Credential {
Token(String),
Basic { username: String, password: String },
}
impl std::fmt::Debug for Credential {
// Never printed, not even under `{:?}`.
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Credential::Token(_) => f.write_str("Credential::Token(…)"),
Credential::Basic { username, .. } => write!(f, "Credential::Basic({username}, …)"),
}
}
}
impl Credential {
fn header(&self) -> String {
match self {
Credential::Token(t) => format!("Bearer {t}"),
Credential::Basic { username, password } => format!("Basic {}", base64(format!("{username}:{password}").as_bytes())),
}
}
}
/// Timeouts. A single API call and the upload of a multi-hundred-MiB medium
/// cannot share one budget without one of them being wrong.
#[derive(Debug, Clone, Copy)]
pub struct Options {
/// One API call, not a procedure.
pub call_timeout: Duration,
/// One `PUT` of a medium.
pub upload_timeout: Duration,
/// Retry a TRANSPORT failure within [`net::Budget`] (reads always, writes
/// only when nothing left this box). `false` sends every call exactly once
/// — for a caller whose own contract is "one attempt, the caller retries",
/// which is monetize-cloud-impl's.
pub retry_transport: bool,
}
impl Default for Options {
fn default() -> Options {
Options { call_timeout: Duration::from_secs(60), upload_timeout: Duration::from_secs(30 * 60), retry_transport: true }
}
}
/// **The only constructor.** The implementation a run holds is a function of
/// the [`Endpoint`], and an `Endpoint` is only ever built from what the
/// operator typed.
pub fn connect(endpoint: &Endpoint, credential: Credential, options: Options) -> Box<dyn UpCloudApi + Send + Sync> {
let agent = |t: Duration| {
let cfg = ureq::Agent::config_builder()
// Every status the API composes — the `412 out_of_stock` included —
// arrives as an ordinary response and is judged by the caller.
.http_status_as_error(false)
.timeout_global(Some(t))
.build();
ureq::Agent::new_with_config(cfg)
};
let (base, account) = match endpoint {
Endpoint::Account => (ACCOUNT_BASE.to_string(), true),
Endpoint::Mock(b) => (b.as_str().to_string(), false),
};
Box::new(Over(Wire { base, account, credential, retry: options.retry_transport, gateway_first: GATEWAY_FIRST, agent: agent(options.call_timeout), upload: agent(options.upload_timeout) }))
}
struct Wire {
base: String,
account: bool,
credential: Credential,
retry: bool,
/// The first wait before a read that met a 502/503/504 is asked again
/// ([`GATEWAY_FIRST`]; a test shortens it).
gateway_first: Duration,
agent: ureq::Agent,
upload: ureq::Agent,
}
fn read(mut res: ureq::http::Response<ureq::Body>) -> Result<Reply, String> {
let status = res.status().as_u16();
let text = res.body_mut().read_to_string().map_err(|e| format!("reading the reply body of a {status}: {e}"))?;
Ok(Reply::from_text(status, text))
}
impl Wire {
fn who(&self) -> &'static str {
if self.account {
""
} else {
" (the fake)"
}
}
/// Reads are idempotent and retried on the wire; writes only when the
/// failure proves nothing left this box — one order must never become two.
fn call(&self, m: Method, path: &str, body: Option<&Value>) -> Result<Reply, String> {
let url = format!("{}{path}", self.base);
let what = format!("{} {path}", m.as_str());
let budget = if !self.retry {
Budget::once()
} else if m == Method::Get {
Budget::api()
} else {
Budget::write()
};
// A READ that met the gateway and not the API is asked again. MEASURED
// 2026-09-30, test2's install on the account: three `GET /server` polls in
// the boot lanes answered 502 ERROR_BAD_GATEWAY "Error accessing upstream
// service" within the same minute, and each lane gave up — main and the
// twin never got their cdrom and the install was lost. A 502/503/504 is the
// gateway saying the API behind it did not answer; for a GET nothing was
// done, so asking again cannot duplicate anything. Every other status is
// still DATA (net.rs), and a write is never retried on a status.
let mut wait = self.gateway_first;
let began = std::time::Instant::now();
loop {
let reply = self.call_once(m, body, &url, &what, budget)?;
let gateway = matches!(reply.status, 502..=504);
if !(gateway && m == Method::Get && self.retry && began.elapsed() < GATEWAY_BUDGET) {
return Ok(reply);
}
eprintln!(
" … {what}{}: the API gateway answered {} (the API behind it did not) — a read, asked again in {} s ({} of {} s)",
self.who(),
reply.status,
wait.as_secs(),
began.elapsed().as_secs(),
GATEWAY_BUDGET.as_secs()
);
std::thread::sleep(wait);
wait = (wait * 2).min(GATEWAY_CAP);
}
}
fn call_once(&self, m: Method, body: Option<&Value>, url: &str, what: &str, budget: Budget) -> Result<Reply, String> {
let url = url.to_string();
let auth = self.credential.header();
let res = net::call(what, budget, || match (m, body) {
(Method::Get, _) => self.agent.get(&url).header("Authorization", &auth).header("Accept", "application/json").call(),
(Method::Delete, _) => self.agent.delete(&url).header("Authorization", &auth).header("Accept", "application/json").call(),
(Method::Post, b) => self
.agent
.post(&url)
.header("Authorization", &auth)
.header("Accept", "application/json")
.header("Content-Type", "application/json")
.send(b.map(|v| v.to_string()).unwrap_or_else(|| "{}".to_string())),
(Method::Put, b) => self
.agent
.put(&url)
.header("Authorization", &auth)
.header("Accept", "application/json")
.header("Content-Type", "application/json")
.send(b.map(|v| v.to_string()).unwrap_or_else(|| "{}".to_string())),
})
.map_err(|e| format!("{what}{}: {e}", self.who()))?;
read(res)
}
/// The upload URL must belong to the cloud this wire talks to. A fake that
/// answered an import with the account's upload host, or an account whose
/// reply named loopback, is a reply this process will not act on.
fn upload_url_belongs(&self, url: &str) -> Result<(), String> {
let ok = if self.account {
url.starts_with("https://") && url.split('/').nth(2).map(|h| h.ends_with(".upcloud.com")).unwrap_or(false)
} else {
crate::is_loopback(url)
};
if ok {
Ok(())
} else {
Err(format!(
"REFUSED [upload-url-foreign] the import answered an upload URL {} that does not belong to {} — \
nothing is uploaded to a host the run was not aimed at",
redact_upload_url(url),
if self.account { "the account's upload hosts" } else { "the loopback fake" }
))
}
}
}
impl Exchange for Wire {
fn describe(&self) -> String {
if self.account {
format!("THE ACCOUNT — {}", self.base)
} else {
format!("MOCK_UPCLOUD at {} — a FAKE", self.base)
}
}
fn is_the_account(&self) -> bool {
self.account
}
fn exchange(&self, call: Call<'_>) -> Result<Reply, String> {
match call {
Call::Api { method, path, body } => self.call(method, &path, body.as_ref()),
Call::Upload { url, file } => self.upload(url, file),
}
}
}
impl Wire {
fn upload(&self, url: &str, file: &Path) -> Result<Reply, String> {
self.upload_url_belongs(url)?;
let shown = redact_upload_url(url);
let len = std::fs::metadata(file).map_err(|e| format!("{}: {e}", file.display()))?.len();
let f = std::fs::File::open(file).map_err(|e| format!("open {} for upload: {e}", file.display()))?;
// NOT retried: a medium PUT that failed mid-flight may have landed, and
// the caller verifies the digest the import reports anyway. NO bearer:
// the URL is itself the credential, and the API token has no business
// travelling to the upload host. Content-Length is set BY HAND and is
// load-bearing — the direct-upload endpoint answers a chunked PUT 411.
let res = self
.upload
.put(url)
.header("Content-Type", "application/octet-stream")
.header("Content-Length", &len.to_string())
.send(ureq::SendBody::from_owned_reader(f))
// The redacted form, on the transport-error path too: a hiccup
// part-way through the PUT once printed the session secret whole.
.map_err(|e| format!("PUT {shown}{}: {e}", self.who()))?;
read(res)
}
}
/// RFC 4648 base64 for one Basic header — hand-rolled so this leaf pulls no
/// crate for twelve lines.
fn base64(input: &[u8]) -> String {
const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
for c in input.chunks(3) {
let b = [c[0], *c.get(1).unwrap_or(&0), *c.get(2).unwrap_or(&0)];
let n = (b[0] as u32) << 16 | (b[1] as u32) << 8 | b[2] as u32;
out.push(T[(n >> 18) as usize & 63] as char);
out.push(T[(n >> 12) as usize & 63] as char);
out.push(if c.len() > 1 { T[(n >> 6) as usize & 63] as char } else { '=' });
out.push(if c.len() > 2 { T[n as usize & 63] as char } else { '=' });
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn basic_auth_is_rfc4648() {
assert_eq!(base64(b"user:pass"), "dXNlcjpwYXNz");
assert_eq!(base64(b"a"), "YQ==");
assert_eq!(base64(b"ab"), "YWI=");
}
/// **FAILS-BEFORE, BY NEUTRALISATION**: make `connect` ignore the endpoint
/// and hand every run the account's base, and this fails on the second
/// assertion — a fake-selected run would describe itself as the account.
#[test]
fn a_fake_run_never_holds_the_account_and_an_account_run_never_holds_a_fake() {
let real = connect(&Endpoint::Account, Credential::Token("t".into()), Options::default());
assert!(real.is_the_account() && real.describe().contains("THE ACCOUNT"));
let fake = connect(&Endpoint::mock("http://127.0.0.1:8099").unwrap(), Credential::Token("t".into()), Options::default());
assert!(!fake.is_the_account(), "a fake that says it is the account is the whole bug");
assert!(fake.describe().contains("FAKE") && !fake.describe().contains("api.upcloud.com"));
}
#[test]
fn an_upload_url_must_belong_to_the_cloud_the_run_is_aimed_at() {
let mk = |e: &Endpoint| Wire {
base: String::new(),
account: e.is_account(),
credential: Credential::Token("t".into()),
retry: true,
gateway_first: GATEWAY_FIRST,
agent: ureq::Agent::new_with_defaults(),
upload: ureq::Agent::new_with_defaults(),
};
let acct = mk(&Endpoint::Account);
let fake = mk(&Endpoint::mock("http://127.0.0.1:1").unwrap());
let real_url = "https://fi-hel1.img.upcloud.com/uploader/session/x";
let mock_url = "http://127.0.0.1:8099/uploader/session/x";
assert!(acct.upload_url_belongs(real_url).is_ok());
assert!(acct.upload_url_belongs(mock_url).is_err());
assert!(fake.upload_url_belongs(mock_url).is_ok());
let e = fake.upload_url_belongs(real_url).unwrap_err();
assert!(e.contains("upload-url-foreign") && !e.contains("/session/x"), "{e}");
assert!(acct.upload_url_belongs("https://evil.example/upcloud.com/x").is_err());
}
/// A tiny HTTP server that answers each connection with the next status in
/// `statuses` (the last one repeats) and counts the requests it saw.
fn serve(statuses: Vec<u16>) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
use std::io::{Read, Write};
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let base = format!("http://{}", l.local_addr().unwrap());
let seen = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let n = seen.clone();
std::thread::spawn(move || {
for c in l.incoming() {
let Ok(mut c) = c else { break };
// A client that never finishes its request must not hold this thread.
let _ = c.set_read_timeout(Some(Duration::from_secs(5)));
// Read the whole request (headers, then Content-Length bytes)
// before answering, or a POST's body meets a closed socket.
let mut got = Vec::new();
let mut buf = [0u8; 4096];
while let Ok(k) = c.read(&mut buf) {
if k == 0 { break }
got.extend_from_slice(&buf[..k]);
let text = String::from_utf8_lossy(&got).to_string();
if let Some(end) = text.find("\r\n\r\n") {
let len = text[..end].lines().find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))).unwrap_or(0);
if got.len() >= end + 4 + len { break }
}
}
let i = n.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let st = statuses[i.min(statuses.len() - 1)];
let body = if st == 200 { r#"{"server":{"state":"started"}}"# } else { "<html>bad gateway</html>" };
let _ = write!(c, "HTTP/1.1 {st} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len());
}
});
(base, seen)
}
fn wire_at(base: &str) -> Wire {
// Built as `connect` builds it: every status arrives as a response.
let agent = |t: u64| ureq::Agent::new_with_config(ureq::Agent::config_builder().http_status_as_error(false).timeout_global(Some(Duration::from_secs(t))).build());
Wire {
base: base.to_string(),
account: false,
credential: Credential::Token("t".into()),
retry: true,
gateway_first: Duration::from_millis(5),
agent: agent(5),
upload: agent(5),
}
}
/// **RED before the gateway retry**: test2's install lost main and the twin to
/// three `GET /server` polls that answered 502 once each. A READ that meets
/// the gateway is asked again and gets the API's answer.
#[test]
fn a_read_that_meets_a_bad_gateway_is_asked_again_and_gets_the_answer() {
let (base, seen) = serve(vec![502, 503, 504, 200]);
let r = wire_at(&base).call(Method::Get, "/server/x", None).unwrap();
assert_eq!(r.status, 200, "the read must reach the API's answer, not stop at the gateway");
assert_eq!(r.body["server"]["state"], "started");
assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 4);
}
/// A WRITE is never retried on a status: one order must never become two.
#[test]
fn a_write_that_meets_a_bad_gateway_is_not_asked_again() {
let (base, seen) = serve(vec![502, 200]);
let r = wire_at(&base).call(Method::Post, "/server/x/start", None).unwrap();
assert_eq!(r.status, 502);
assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
}
/// Any other status is DATA and answered at once (a 404 is an absence).
#[test]
fn a_read_that_is_refused_is_not_asked_again() {
let (base, seen) = serve(vec![404, 200]);
let r = wire_at(&base).call(Method::Get, "/server/x", None).unwrap();
assert_eq!(r.status, 404);
assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1);
}
}