upcloud_api/

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};

/// 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, agent: agent(options.call_timeout), upload: agent(options.upload_timeout) }))
}

struct Wire {
    base: String,
    account: bool,
    credential: Credential,
    retry: bool,
    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}"))?;
    let body = if text.trim().is_empty() { Value::Null } else { serde_json::from_str(&text).unwrap_or(Value::Null) };
    Ok(Reply { status, body, 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()
        };
        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,
            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());
    }
}