protocol.rs 86.1%

1101 of 1279 lines run, 179 of 272 functions · run #407 on main · run missed
//! Transport-agnostic protocol (spec §16): what a client asks of a remote and
//! what a server does with a push. `LocalRemote` serves another repository
//! directory in-process, `HttpRemote` speaks to an `oxy-server` over HTTP(S),
//! and the server side reuses `Repo::receive_push` and friends unchanged.
//!
//! Wire messages are deterministic CBOR (`wire` module); over HTTP each is a
//! request or response body with content type `application/oxy-cbor`.

use crate::attest::Attest;
use crate::cbor::{self, Value};
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
use crate::identity::Signature;
use crate::objects::{self as o, Change, ObjectError};
use crate::policy::Policy;
use crate::repo::{Repo, RepoError, Result};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

pub const CONTENT_TYPE: &str = "application/oxy-cbor";
pub const PROTOCOL_VERSION: u64 = 1;

/// An attestation together with the signatures over its ObjectRef.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SignedAttest {
    pub attest: Attest,
    pub sigs: Vec<Signature>,
}

/// A policy together with the root signatures over its ObjectRef.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SignedPolicy {
    pub policy: Policy,
    pub sigs: Vec<Signature>,
}

impl SignedPolicy {
    /// The proposal-file encoding (`wire::policy_value`).
    pub fn encode(&self) -> Vec<u8> {
        crate::cbor::encode(&wire::policy_value(self))
    }
    pub fn decode(b: &[u8]) -> Result<SignedPolicy> {
        wire::policy_from(
            &crate::cbor::decode(b).map_err(|e| RepoError::Object(ObjectError::from(e)))?,
        )
    }
}

/// What `refs` returns: every bookmark, every tag, and the current policy.
#[derive(Clone, Debug, Default)]
pub struct RefsResponse {
    pub refs: Vec<RemoteRef>,
    pub policy: Option<SignedPolicy>,
    pub tags: Vec<TagRef>,
    /// Earlier policies, ascending by serial, so a client behind by more
    /// than one update can verify each step (spec §14.5).
    pub history: Vec<SignedPolicy>,
    /// The bookmark a clone checks out (spec §16.1), when the remote names one.
    pub default_bookmark: Option<String>,
    /// The host's latest transparency-log checkpoint, a signed note (spec
    /// §14.8), against which `RemoteRef::logged` proofs are made (key 6,
    /// only when the host keeps a log).
    pub log: Option<String>,
    /// The witnesses the host asks to cosign its checkpoints: each one's
    /// name in cosignatures and its identity's bytes (key 7, only when it
    /// has any), which a client pins (spec §14.8, *Integrated time*).
    pub witnesses: Vec<(String, Vec<u8>)>,
    /// Tags withdrawn here (spec §10.1; key 8, only when there are any):
    /// each with its `untag` attestation as `attestation`, and that
    /// attestation's acceptance as `accepted` when a short-lived key
    /// signed it. Older clients ignore them and keep the tag they have.
    pub withdrawn: Vec<TagRef>,
}

/// Where a reference attestation sits in the host's transparency log
/// (spec §14.8): its entry number, the entry, and the inclusion proof
/// against the checkpoint served beside it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Logged {
    pub index: u64,
    pub entry: Vec<u8>,
    pub proof: Vec<[u8; 32]>,
}

/// A checkpoint witnesses cosigned that includes a logged entry (spec
/// §14.8, *Integrated time*): the cosigned note, and the entry's inclusion
/// proof against it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Integration {
    pub note: String,
    pub proof: Vec<[u8; 32]>,
}

/// A tag as a remote reports it (spec §10.1).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TagRef {
    pub name: String,
    pub state: ObjectRef,
    pub heads: Vec<ChangeHash>,
    pub attestation: Option<SignedAttest>,
    /// The acceptance of `attestation` by its signer's certificate issuer
    /// when a short-lived key signed it (spec §14.7; key 4, only when there
    /// is one): the time a fetch checks the certificate at.
    pub accepted: Option<SignedAttest>,
    /// The attestation's place in the host's transparency log (key 5,
    /// only when it has one), as a bookmark's.
    pub logged: Option<Logged>,
    /// The cosigned checkpoints that first included `logged`'s entry (key
    /// 6, only when there are any).
    pub integration: Vec<Integration>,
}

impl TagRef {
    /// Whether this is a tag's withdrawal (its attestation an `untag`), as
    /// a push carries one (oxygenhub#301).
    pub fn is_withdrawal(&self) -> bool {
        self.attestation
            .as_ref()
            .is_some_and(|a| matches!(a.attest.body, crate::attest::AttestBody::Untag { .. }))
    }

    /// A tag with no acceptance or log place (yet).
    pub fn new(
        name: String,
        state: ObjectRef,
        heads: Vec<ChangeHash>,
        attestation: Option<SignedAttest>,
    ) -> TagRef {
        TagRef {
            name,
            state,
            heads,
            attestation,
            accepted: None,
            logged: None,
            integration: vec![],
        }
    }
}

/// One reference as a remote reports it.
#[derive(Clone, Debug)]
pub struct RemoteRef {
    pub name: String,
    pub state: ObjectRef,
    pub heads: Vec<ChangeHash>,
    /// The latest attestation, only when it describes `state`. An unattested
    /// push (accepted while no trust policy exists) leaves it `None`.
    pub attestation: Option<SignedAttest>,
    /// Serial of the latest attestation ever recorded, stale or not, so the
    /// next attestation continues the sequence.
    pub last_serial: Option<u64>,
    /// The host's acceptance of `attestation` when a short-lived key signed
    /// it (spec §14.7): the time its certificate is checked at.
    pub accepted: Option<SignedAttest>,
    /// The attestation's place in the host's transparency log (key 6, only
    /// when it has one).
    pub logged: Option<Logged>,
    /// The cosigned checkpoints that first included `logged`'s entry (key
    /// 7, only when there are any): what its integrated time is read from.
    pub integration: Vec<Integration>,
}

/// Everything needed to add a set of changes to a repository.
#[derive(Clone, Debug, Default)]
pub struct Bundle {
    /// In dependency order, each with its individual signatures.
    pub changes: Vec<(Change, Vec<Signature>)>,
    /// Seals covering the changes.
    pub attests: Vec<SignedAttest>,
    /// Chunks, blobs, identities, and state objects referenced by the changes.
    pub objects: Vec<(ObjectType, Vec<u8>)>,
    /// Policies between the receiver's and the one sent with the push,
    /// ascending by serial (spec §14.5).
    pub policies: Vec<SignedPolicy>,
    /// Each change's hash, in the order of `changes`, when whoever made
    /// the bundle knows it: the store it came from, or the bytes it was
    /// decoded from. Never read from the wire. Empty when unknown.
    pub hashes: Vec<ChangeHash>,
    /// Changes as stored, with their signatures, in dependency order, for a
    /// fetch between repositories on one machine: their text blobs and
    /// chunks are among `objects`. Never on the wire.
    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,
    /// Certificates of short-lived keys that signed what the bundle carries
    /// (spec §14.7), their issuers' identities among `objects`.
    pub certificates: Vec<crate::cert::SignedCertificate>,
}

impl Bundle {
    pub fn is_empty(&self) -> bool {
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
    }
}

/// A requested bookmark move (spec §10.2, §16.3).
#[derive(Clone, Debug)]
pub struct RefUpdate {
    pub name: String,
    pub old_state: ObjectRef,
    pub new_state: ObjectRef,
    pub force: bool,
    pub attestation: Option<SignedAttest>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RefResult {
    pub name: String,
    pub ok: bool,
    pub reason: String,
}

pub trait Remote {
    fn repo_id(&self) -> Result<[u8; 16]>;
    fn refs(&self) -> Result<RefsResponse>;
    /// Changes in the closure of `want` but not of `have`, in dependency order.
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
    /// The changes with their signatures, seals, identities, and blobs; blobs
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
    /// with its chunks.
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the
    /// checkpoint object, its payload blob, and the blob's chunks.
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle>;
    /// A checkpoint of exactly `heads` without its graph blobs (spec §16.1
    /// `checkpoint-index`, oxygen#137), which the client then asks for in
    /// parts with `objects`. A remote that cannot answer it sends the
    /// whole checkpoint, which does as well.
    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        self.checkpoint(heads)
    }
    /// The changes within `depth` dependency steps of `heads` (spec §17).
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>>;
    /// Push changes, bookmark updates, new tags, and optionally a newer
    /// policy. The policy is applied first; a result named `policy` reports
    /// it, and each tag reports as `tags/<name>`.
    fn push(
        &self,
        bundle: Bundle,
        updates: Vec<RefUpdate>,
        policy: Option<SignedPolicy>,
        tags: Vec<TagRef>,
    ) -> Result<Vec<RefResult>>;
}

// ---------------------------------------------------------------- local

/// Another repository on this machine, served in-process.
pub struct LocalRemote {
    pub repo: Repo,
    /// The pushing repository's id, which binds an unbound target on the
    /// first push, as `Oxy-Repo-Id` does over HTTP.
    pub claimed_id: Option<[u8; 16]>,
}

impl LocalRemote {
    pub fn open(path: &Path) -> Result<LocalRemote> {
        Ok(LocalRemote {
            repo: Repo::open(path)?,
            claimed_id: None,
        })
    }
}

impl Remote for LocalRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        self.repo.id()
    }
    fn refs(&self) -> Result<RefsResponse> {
        self.repo.remote_refs()
    }
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        self.repo.negotiate(have, want)
    }
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
        self.repo.bundle_raw_for(changes, max_blob)
    }
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
        self.repo.objects_for(refs)
    }
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        self.repo.checkpoint_bundle_for(heads)
    }
    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        self.repo.checkpoint_index_for(heads)
    }
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
        self.repo.shallow_changes(heads, depth)
    }
    fn push(
        &self,
        bundle: Bundle,
        updates: Vec<RefUpdate>,
        policy: Option<SignedPolicy>,
        tags: Vec<TagRef>,
    ) -> Result<Vec<RefResult>> {
        if self.repo.is_unbound() {
            match self.claimed_id {
                Some(id) => self.repo.bind_id(id)?,
                None => {
                    return Err(RepoError::Msg(
                        "the remote is unbound and this client did not say which repository it is"
                            .into(),
                    ))
                }
            }
        }
        self.repo.receive_push(bundle, updates, policy, tags)
    }
}

// ---------------------------------------------------------------- http

/// How to reach a remote: credentials and trust.
#[derive(Clone, Debug, Default)]
pub struct RemoteOptions {
    /// Bearer token for pushes.
    pub token: Option<String>,
    /// PEM file with extra root certificates to trust for `https://` remotes
    /// (self-hosted servers); the system web roots are always trusted.
    pub ca: Option<PathBuf>,
    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound
    /// hosted repository can adopt it on the first push (spec §16.1).
    pub repo_id: Option<[u8; 16]>,
    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its
    /// mirror and what it has seen: the remote's directory under
    /// `.oxy/remotes`. None (a clone, before there is a repository) means
    /// a temporary directory.
    pub cache_dir: Option<PathBuf>,
    /// Other tokens for the same hub, tried in turn when `token` is
    /// refused (401, 403 or 404): the credential store's sign-in when
    /// `token` is the remote's own (oxygen#155), or those of enclosing
    /// checkouts when it came from the credential store (oxygen#112).
    pub fallback_tokens: Vec<String>,
}

/// Opens a store for a URL it understands, or passes (`None`). Stores that
/// need more than the standard library (the cloud ones) are added by the
/// program that links them, with [`register_store_opener`].
pub type StoreOpener =
    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;

static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());

pub fn register_store_opener(opener: StoreOpener) {
    STORE_OPENERS
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .push(opener);
}

/// A store for `url`, from `dir://` or any registered opener.
pub fn open_store(
    url: &str,
    options: &RemoteOptions,
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
    if let Some(path) = url.strip_prefix("dir://") {
        return Some(
            crate::strata::LocalStore::open(path)
                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),
        );
    }
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
    openers.iter().find_map(|open| open(url, options))
}

/// The request header carrying the client's repository id.
pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";

/// The id of a hosted repository that has not adopted one yet.
pub const UNBOUND_ID: [u8; 16] = [0; 16];

/// The request header in which a client declares the protocol features it
/// understands beyond the 0.5.0-era set: a comma-separated list, which a
/// server may tailor its answers to (docs/protocol.md, *Client
/// capabilities*). A request without it is a client of that era.
pub const FEATURES_HEADER: &str = "Oxy-Features";

/// The feature of a client that authorizes a tag signed by a short-lived,
/// certified key by its certificate at the time the host accepted it
/// (oxygen#129). A client without it refuses such a tag, and with it the
/// whole fetch, so servers leave those tags out of its `refs` (oxygen#131).
pub const FEATURE_CERTIFIED_TAGS: &str = "certified-tags";

/// The features this client declares in [`FEATURES_HEADER`].
pub const CLIENT_FEATURES: &[&str] = &[FEATURE_CERTIFIED_TAGS];

/// The `User-Agent` every request of this client sends: `oxy/<version>`.
pub fn user_agent() -> String {
    format!("oxy/{}", env!("CARGO_PKG_VERSION"))
}

/// Whether a [`FEATURES_HEADER`] value (absent: `None`) declares `feature`.
/// Entries are comma-separated, compared without surrounding blanks and
/// regardless of case; unknown ones are ignored.
pub fn declares_feature(header: Option<&str>, feature: &str) -> bool {
    header.is_some_and(|h| h.split(',').any(|f| f.trim().eq_ignore_ascii_case(feature)))
}

/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
pub struct HttpRemote {
    base: String,
    /// The token in use, and those still to try when it is refused.
    token: Mutex<Option<String>>,
    fallbacks: Mutex<Vec<String>>,
    repo_id: Option<[u8; 16]>,
    agent: ureq::Agent,
}

/// A TLS client configuration trusting the web roots plus `ca`, if given:
/// the trust every oxy connection uses, shared with tools that call hub APIs.
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
    let mut roots = rustls::RootCertStore::empty();
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
    if let Some(path) = ca {
        let pem = std::fs::read(path)
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
            .collect::<std::result::Result<Vec<_>, _>>()
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
        if certs.is_empty() {
            return Err(RepoError::Msg(format!(
                "CA file {} holds no certificates",
                path.display()
            )));
        }
        for c in certs {
            roots
                .add(c)
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
        }
    }
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
        rustls::crypto::ring::default_provider(),
    ))
    .with_safe_default_protocol_versions()
    .map_err(|e| RepoError::Msg(e.to_string()))?
    .with_root_certificates(roots)
    .with_no_client_auth();
    Ok(Arc::new(cfg))
}

impl HttpRemote {
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
        refuse_plain_token(base, options)?;
        let agent = ureq::AgentBuilder::new()
            .tls_config(tls_client_config(options.ca.as_deref())?)
            .user_agent(&user_agent())
            .build();
        Ok(HttpRemote {
            base: base.trim_end_matches('/').to_string(),
            token: Mutex::new(options.token.clone()),
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
            repo_id: options.repo_id,
            agent,
        })
    }

    /// A request, tried again with the next fallback token while the
    /// hub refuses the one in use (oxygen#112).
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
        loop {
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
            match self.call_once(post, endpoint, body, token.as_deref()) {
                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {
                    let next = self
                        .fallbacks
                        .lock()
                        .unwrap_or_else(|p| p.into_inner())
                        .pop();
                    let Some(next) = next else {
                        return Err(RepoError::Msg(m));
                    };
                    crate::credentials::warn_once(&format!(
                        "{}: the token in use was refused; trying another this computer has for the hub \
                         (a stale remote.NAME.token goes with `oxy config --unset remote.NAME.token`)",
                        self.base
                    ));
                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);
                }
                other => return other,
            }
        }
    }

    fn call_once(
        &self,
        post: bool,
        endpoint: &str,
        body: Option<&[u8]>,
        token: Option<&str>,
    ) -> Result<Vec<u8>> {
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
        let mut req = if post {
            self.agent.post(&url)
        } else {
            self.agent.get(&url)
        };
        req = req
            .set("Accept", CONTENT_TYPE)
            .set(FEATURES_HEADER, &CLIENT_FEATURES.join(","));
        if let Some(t) = token {
            req = req.set("Authorization", &format!("Bearer {t}"));
        }
        if let Some(id) = &self.repo_id {
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
        }
        let resp = match body {
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
            None => req.call(),
        };
        match resp {
            Ok(r) => {
                let mut out = Vec::new();
                r.into_reader()
                    .read_to_end(&mut out)
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
                Ok(out)
            }
            Err(ureq::Error::Status(code, r)) => {
                let text = r.into_string().unwrap_or_default();
                // The hint comes from the status alone: the server's text
                // (a host's reason for refusing a push, several lines
                // perhaps) is shown as it is, never read for hints. A
                // proxy's HTML error page is shown by its title only
                // (oxygen#133).
                let status = format!("{url}: HTTP {code}");
                let msg = format!("{status}: {}", plain_error_text(&text));
                Err(RepoError::Msg(match hint_for(&status) {
                    Some(h) => format!("{msg}\n  hint: {h}"),
                    None => msg,
                }))
            }
            // The transport error already names the URL.
            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
        }
    }
}

/// Whether a failed request's message says the hub refused its token:
/// HTTP 401, 403, or 404 (a hub hides a private repository from a token
/// that may not read it).
fn refused(message: &str) -> bool {
    [": HTTP 401", ": HTTP 403", ": HTTP 404"]
        .iter()
        .any(|c| message.contains(c))
}

/// Refuse to send a token where anyone on the way could read it: plain
/// `http` to another machine (see [`crate::credentials::token_allowed`]).
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
    if options.token.is_some() && !crate::credentials::token_allowed(url) {
        return Err(RepoError::Msg(format!(
            "{url}: refusing to send a token over plain http; use https"
        )));
    }
    Ok(())
}

/// What to do about a failed request, recognised from its message.
pub fn hint_for(message: &str) -> Option<&'static str> {
    let m = message.to_ascii_lowercase();
    let hint = if m.contains("http 401") {
        "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)"
    } else if m.contains("http 403") {
        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"
    } else if m.contains("http 413") {
        "the request was larger than the server, or a proxy in front of it, takes at once (Cloudflare's limit is 100 MB): push in smaller parts with `oxy config push.batch-bytes 33554432`, then push again"
    } else if m.contains("http 404") {
        "nothing at this URL that you can see: check the owner and repository name; a private repository needs you signed in (`oxy auth login HUB`)"
    } else if m.contains("unknownissuer")
        || m.contains("unknown issuer")
        || m.contains("certificate")
        || m.contains("invalidcertificate")
    {
        "the server's certificate is not trusted: for a hub with a self-signed certificate, download it from /.well-known/oxyhub-ca.pem on the hub and run `oxy config remote.<remote>.ca FILE`"
    } else if m.contains("connection refused") {
        "nothing is listening there: check the URL and port, and that the server is running"
    } else if m.contains("dns")
        || m.contains("failed to lookup")
        || m.contains("name or service not known")
    {
        "the host name does not resolve: check the URL"
    } else if m.contains("timed out") || m.contains("timeout") {
        "the server did not answer in time: check the network, or try again"
    } else {
        return None;
    };
    Some(hint)
}

/// An error response's text as it is, unless it is an HTML page (a proxy's,
/// such as Cloudflare's 413): then its title, or a note that it was one.
fn plain_error_text(text: &str) -> String {
    let t = text.trim();
    let lower = t.to_ascii_lowercase();
    if !(lower.starts_with("<!doctype html") || lower.starts_with("<html")) {
        return t.to_string();
    }
    let title = lower
        .find("<title>")
        .and_then(|i| {
            lower[i + 7..]
                .find("</title>")
                .map(|j| t[i + 7..i + 7 + j].trim())
        })
        .filter(|s| !s.is_empty());
    match title {
        Some(s) => format!("{s} (an HTML error page, from a proxy in front of the server perhaps)"),
        None => "an HTML error page (from a proxy in front of the server perhaps)".into(),
    }
}

/// `message`, followed by what to do about it when that is known.
pub fn with_hint(message: String) -> String {
    match hint_for(&message) {
        Some(h) => format!("{message}\n  hint: {h}"),
        None => message,
    }
}

/// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same
/// TLS trust and token rules as remotes. Returns the status and body, an
/// error status included; only transport failures are errors.
pub fn http_json(
    method: &str,
    url: &str,
    body: Option<&str>,
    options: &RemoteOptions,
) -> Result<(u16, String)> {
    refuse_plain_token(url, options)?;
    let agent = ureq::AgentBuilder::new()
        .tls_config(tls_client_config(options.ca.as_deref())?)
        .timeout(std::time::Duration::from_secs(60))
        .user_agent(&user_agent())
        .build();
    let mut req = agent.request(method, url).set("Accept", "application/json");
    if let Some(t) = &options.token {
        req = req.set("Authorization", &format!("Bearer {t}"));
    }
    let resp = match body {
        Some(b) => req.set("Content-Type", "application/json").send_string(b),
        None => req.call(),
    };
    match resp {
        Ok(r) => {
            let code = r.status();
            Ok((code, r.into_string().unwrap_or_default()))
        }
        Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())),
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
    }
}

/// An authenticated GET of any URL on a host, with the same TLS trust and
/// token as remotes use (for hub APIs: proposals). Returns the body.
pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {
    refuse_plain_token(url, options)?;
    let agent = ureq::AgentBuilder::new()
        .tls_config(tls_client_config(options.ca.as_deref())?)
        .user_agent(&user_agent())
        .build();
    let mut req = agent.get(url).set("Accept", "application/json");
    if let Some(t) = &options.token {
        req = req.set("Authorization", &format!("Bearer {t}"));
    }
    match req.call() {
        Ok(r) => {
            let mut out = Vec::new();
            r.into_reader()
                .read_to_end(&mut out)
                .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
            Ok(out)
        }
        Err(ureq::Error::Status(code, r)) => {
            let text = r.into_string().unwrap_or_default();
            Err(RepoError::Msg(with_hint(format!(
                "{url}: HTTP {code}: {}",
                text.trim()
            ))))
        }
        // The transport error already names the URL.
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
    }
}

impl Remote for HttpRemote {
    fn repo_id(&self) -> Result<[u8; 16]> {
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
    }
    fn refs(&self) -> Result<RefsResponse> {
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
    }
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
        wire::decode_hashes(&self.call(
            true,
            "negotiate",
            Some(&wire::encode_negotiate(have, want)),
        )?)
    }
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
        wire::decode_bundle(&self.call(
            true,
            "fetch",
            Some(&wire::encode_fetch(changes, max_blob)),
        )?)
    }
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
        Ok(
            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?
                .objects,
        )
    }
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        let body = {
            let _t = crate::timing::phase("remote: checkpoint answer");
            self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?
        };
        let _t = crate::timing::phase("remote: decode the checkpoint");
        wire::decode_bundle(&body)
    }
    fn checkpoint_index(&self, heads: &[ChangeHash]) -> Result<Bundle> {
        let body = {
            let _t = crate::timing::phase("remote: checkpoint index");
            match self.call(true, "checkpoint-index", Some(&wire::encode_hashes(heads))) {
                Ok(b) => b,
                // A server from before checkpoint-index: the whole answer.
                Err(RepoError::Msg(m)) if m.contains("HTTP 404") => return self.checkpoint(heads),
                Err(e) => return Err(e),
            }
        };
        wire::decode_bundle(&body)
    }
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
        wire::decode_hashes(&self.call(
            true,
            "shallow",
            Some(&wire::encode_shallow(heads, depth)),
        )?)
    }
    fn push(
        &self,
        bundle: Bundle,
        updates: Vec<RefUpdate>,
        policy: Option<SignedPolicy>,
        tags: Vec<TagRef>,
    ) -> Result<Vec<RefResult>> {
        wire::decode_push_response(&self.call(
            true,
            "push",
            Some(&wire::encode_push(
                &bundle,
                &updates,
                policy.as_ref(),
                &tags,
            )),
        )?)
    }
}

/// A remote's URL as it is stored: an `http(s)://` URL as given, a local
/// path made absolute (with its `file://` kept), so it still names the same
/// repository from inside the clone (oxygen#57).
pub fn stored_url(url: &str) -> String {
    // Only a path is made absolute: a plain one, or one behind file:// or
    // dir://. Every other URL (a hub, a bucket) is kept as written.
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
        (Some(p), _) => ("file://", p),
        (_, Some(p)) => ("dir://", p),
        _ if url.contains("://") => return url.to_string(),
        _ => ("", url),
    };
    match std::path::absolute(path) {
        Ok(abs) => format!("{prefix}{}", abs.display()),
        Err(_) => url.to_string(),
    }
}

/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`
/// (a hub), or a store (`dir://`, and whatever openers are registered).
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
    if url.starts_with("http://") || url.starts_with("https://") {
        return Ok(Box::new(HttpRemote::new(url, options)?));
    }
    if let Some(store) = open_store(url, options) {
        return crate::repo::strata::open_store_remote(store?, options);
    }
    let path = url.strip_prefix("file://").unwrap_or(url);
    if path.contains("://") {
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
    }
    let mut local = LocalRemote::open(Path::new(path))?;
    local.claimed_id = options.repo_id;
    Ok(Box::new(local))
}

// ---------------------------------------------------------------- wire

/// CBOR encodings of the protocol messages (spec §16.1).
pub mod wire {
    use super::*;

    fn schema(msg: &str) -> RepoError {
        RepoError::Object(ObjectError::Schema(msg.into()))
    }
    fn sigs_value(sigs: &[Signature]) -> Value {
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
    }
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
        Ok(o::array(v, "signatures")?
            .iter()
            .map(Signature::from_value)
            .collect::<std::result::Result<Vec<_>, _>>()?)
    }
    fn hashes_value(hs: &[ObjectRef]) -> Value {
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
    }
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
        Ok(o::array(v, "hashes")?
            .iter()
            .map(|h| o::oref(h, "hash"))
            .collect::<std::result::Result<Vec<_>, _>>()?)
    }
    fn attest_value(sa: &SignedAttest) -> Value {
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
    }
    fn attest_from(v: &Value) -> Result<SignedAttest> {
        let a = o::array(v, "signed attest")?;
        if a.len() != 2 {
            return Err(schema("signed attest: expected [attest, sigs]"));
        }
        Ok(SignedAttest {
            attest: Attest::decode(o::bytes(&a[0], "attest")?)?,
            sigs: sigs_from(&a[1])?,
        })
    }
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
    }
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
        if v.is_null() {
            Ok(None)
        } else {
            attest_from(v).map(Some)
        }
    }
    /// `[policy bytes, [signature]]`: also the format of a policy proposal
    /// file exchanged between root holders (spec §14.5).
    pub fn policy_value(sp: &SignedPolicy) -> Value {
        Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)])
    }
    pub fn policy_from(v: &Value) -> Result<SignedPolicy> {
        let a = o::array(v, "signed policy")?;
        if a.len() != 2 {
            return Err(schema("signed policy: expected [policy, sigs]"));
        }
        Ok(SignedPolicy {
            policy: Policy::decode(o::bytes(&a[0], "policy")?)?,
            sigs: sigs_from(&a[1])?,
        })
    }
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
        sp.map(policy_value).unwrap_or(Value::Null)
    }
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
        if v.is_null() {
            Ok(None)
        } else {
            policy_from(v).map(Some)
        }
    }
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
        Ok(o::map(v, what)?)
    }
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
        Ok(o::field(m, k, what)?)
    }

    /// A bundle's value. Each change's text blobs and their chunks (spec
    /// §6.1) are added to its objects when they are not already there, so a
    /// bundle can always be read on its own.
    pub fn bundle_value(b: &Bundle) -> Value {
        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());
        // Changes as stored are already their encoding, their texts among
        // the objects.
        for (bytes, s) in &b.raw_changes {
            changes.push(Value::Array(vec![
                Value::Bytes(bytes.clone()),
                sigs_value(s),
            ]));
        }
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
        for (c, s) in &b.changes {
            let (bytes, stored) = c.encode_parts();
            for t in stored {
                let listed = listed.get_or_insert_with(|| {
                    b.objects
                        .iter()
                        .map(|(ty, body)| crate::hash::hash_object(*ty, body))
                        .collect()
                });
                if !listed.insert(t.blob.hash()) {
                    continue;
                }
                for chunk in crate::chunk::chunks(&t.content) {
                    if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) {
                        texts.push((ObjectType::Chunk, chunk.to_vec()));
                    }
                }
                texts.push((ObjectType::Blob, t.blob.encode()));
            }
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
        }
        let mut v = Value::map([
            (0, Value::Array(changes)),
            (
                1,
                Value::Array(b.attests.iter().map(attest_value).collect()),
            ),
            (
                2,
                Value::Array(
                    b.objects
                        .iter()
                        .chain(&texts)
                        .map(|(t, body)| {
                            Value::Array(vec![
                                Value::Text(t.name().into()),
                                Value::Bytes(body.clone()),
                            ])
                        })
                        .collect(),
                ),
            ),
            (
                3,
                Value::Array(b.policies.iter().map(policy_value).collect()),
            ),
        ]);
        // Only when there are any; older peers ignore the field.
        if !b.certificates.is_empty() {
            if let Value::Map(m) = &mut v {
                m.insert(
                    4,
                    Value::Array(
                        b.certificates
                            .iter()
                            .map(|c| Value::Bytes(c.encode()))
                            .collect(),
                    ),
                );
            }
        }
        v
    }

    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
        let m = map(v, "bundle")?;
        let mut b = Bundle::default();
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
            let pair = o::array(ob, "object")?;
            if pair.len() != 2 {
                return Err(schema("object: expected [type, body]"));
            }
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
                .ok_or_else(|| schema("unknown object type"))?;
            b.objects
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
        }
        // Stored inserts are read from the bundle's own objects.
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
        for c in o::array(field(m, 0, "changes")?, "changes")? {
            let pair = o::array(c, "change entry")?;
            if pair.len() != 2 {
                return Err(schema("change entry: expected [change, sigs]"));
            }
            let encoded = o::bytes(&pair[0], "change")?;
            b.hashes
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
            let change = Change::decode_with(encoded, &mut |text, count, form| {
                let by_ref = by_ref.get_or_insert_with(|| {
                    b.objects
                        .iter()
                        .enumerate()
                        .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i))
                        .collect()
                });
                let get =
                    |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) {
                        Some((ty, body)) if *ty == want => Ok(body),
                        _ => Err(ObjectError::Schema(format!(
                            "the bundle lacks {} {}",
                            want.name(),
                            r.short(12)
                        ))),
                    };
                let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?;
                let mut content = Vec::with_capacity(blob.length as usize);
                for chunk in &blob.chunks {
                    content.extend_from_slice(get(chunk, ObjectType::Chunk)?);
                }
                crate::objects::text_lines(&content, count, form)
            })?;
            b.changes.push((change, sigs_from(&pair[1])?));
        }
        for a in o::array(field(m, 1, "attests")?, "attests")? {
            b.attests.push(attest_from(a)?);
        }
        if let Some(ps) = m.get(&3) {
            for p in o::array(ps, "policies")? {
                b.policies.push(policy_from(p)?);
            }
        }
        if let Some(cs) = m.get(&4) {
            for c in o::array(cs, "certificates")? {
                b.certificates.push(
                    crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?)
                        .map_err(|e| schema(&format!("certificate: {e}")))?,
                );
            }
        }
        Ok(b)
    }

    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
        cbor::encode(&bundle_value(b))
    }
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
    }

    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
            (1, Value::Bytes(id.to_vec())),
            (2, Value::Array(vec![Value::Text("seal".into())])),
        ]))
    }
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "capabilities")?;
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
        if !versions
            .iter()
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
        {
            return Err(RepoError::Msg(
                "remote does not speak protocol version 1".into(),
            ));
        }
        o::bytes(field(m, 1, "repo id")?, "repo id")?
            .try_into()
            .map_err(|_| schema("repo id: expected 16 bytes"))
    }

    fn tags_value(tags: &[TagRef]) -> Value {
        Value::Array(
            tags.iter()
                .map(|t| {
                    let mut v = Value::map([
                        (0, Value::Text(t.name.clone())),
                        (1, Value::Bytes(t.state.0.to_vec())),
                        (2, hashes_value(&t.heads)),
                        (3, opt_attest_value(&t.attestation)),
                    ]);
                    // Only when there is one; older peers ignore them
                    // (oxygen#129).
                    if let Value::Map(m) = &mut v {
                        if t.accepted.is_some() {
                            m.insert(4, opt_attest_value(&t.accepted));
                        }
                        if let Some(l) = &t.logged {
                            m.insert(5, logged_value(l));
                        }
                        if !t.integration.is_empty() {
                            m.insert(6, integration_value(&t.integration));
                        }
                    }
                    v
                })
                .collect(),
        )
    }
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
        let mut out = Vec::new();
        for t in o::array(v, "tags")? {
            let tm = map(t, "tag")?;
            out.push(TagRef {
                name: o::text(field(tm, 0, "name")?, "name")?.to_owned(),
                state: o::oref(field(tm, 1, "state")?, "state")?,
                heads: hashes_from(field(tm, 2, "heads")?)?,
                attestation: opt_attest_from(field(tm, 3, "attestation")?)?,
                accepted: match tm.get(&4) {
                    Some(a) => opt_attest_from(a)?,
                    None => None,
                },
                logged: tm.get(&5).map(logged_from).transpose()?,
                integration: match tm.get(&6) {
                    Some(l) => integration_from(l)?,
                    None => Vec::new(),
                },
            });
        }
        Ok(out)
    }

    /// The tags of a refs answer as oxy 0.5.0 read them (and the runners
    /// built with it), keys 0 to 3 only: what an older reader makes of an
    /// answer with keys 4 to 6.
    #[cfg(test)]
    pub(super) fn tags_as_0_5_read_them(bytes: &[u8]) -> Result<Vec<TagRef>> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "refs")?;
        let mut out = Vec::new();
        for t in o::array(field(m, 3, "tags")?, "tags")? {
            let tm = map(t, "tag")?;
            out.push(TagRef::new(
                o::text(field(tm, 0, "name")?, "name")?.to_owned(),
                o::oref(field(tm, 1, "state")?, "state")?,
                hashes_from(field(tm, 2, "heads")?)?,
                opt_attest_from(field(tm, 3, "attestation")?)?,
            ));
        }
        Ok(out)
    }

    /// A log place, `[entry number, entry bytes, [proof hash]]` (spec §14.8).
    fn logged_value(l: &Logged) -> Value {
        Value::Array(vec![
            Value::Uint(l.index),
            Value::Bytes(l.entry.clone()),
            proof_value(&l.proof),
        ])
    }
    fn logged_from(v: &Value) -> Result<Logged> {
        let a = o::array(v, "logged")?;
        if a.len() != 3 {
            return Err(schema("logged: expected [index, entry, proof]"));
        }
        Ok(Logged {
            index: o::uint(&a[0], "index")?,
            entry: o::bytes(&a[1], "entry")?.to_vec(),
            proof: proof_from(&a[2])?,
        })
    }

    /// Cosigned checkpoints holding an entry, `[[note, [proof hash]]]`.
    fn integration_value(integration: &[Integration]) -> Value {
        Value::Array(
            integration
                .iter()
                .map(|i| Value::Array(vec![Value::Text(i.note.clone()), proof_value(&i.proof)]))
                .collect(),
        )
    }
    fn integration_from(v: &Value) -> Result<Vec<Integration>> {
        let mut out = Vec::new();
        for i in o::array(v, "integration")? {
            let a = o::array(i, "integration")?;
            if a.len() != 2 {
                return Err(schema("integration: expected [note, proof]"));
            }
            out.push(Integration {
                note: o::text(&a[0], "note")?.to_owned(),
                proof: proof_from(&a[1])?,
            });
        }
        Ok(out)
    }

    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
        let mut v = Value::map([
            (
                0,
                Value::Array(
                    resp.refs
                        .iter()
                        .map(|r| {
                            let mut v = Value::map([
                                (0, Value::Text(r.name.clone())),
                                (1, Value::Bytes(r.state.0.to_vec())),
                                (2, hashes_value(&r.heads)),
                                (3, opt_attest_value(&r.attestation)),
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
                            ]);
                            // Only when there is one; older peers ignore it.
                            if let Value::Map(m) = &mut v {
                                if r.accepted.is_some() {
                                    m.insert(5, opt_attest_value(&r.accepted));
                                }
                                if let Some(l) = &r.logged {
                                    m.insert(6, logged_value(l));
                                }
                                if !r.integration.is_empty() {
                                    m.insert(7, integration_value(&r.integration));
                                }
                            }
                            v
                        })
                        .collect(),
                ),
            ),
            (1, Value::Bytes(id.to_vec())),
            (2, opt_policy_value(resp.policy.as_ref())),
            (3, tags_value(&resp.tags)),
            (
                4,
                Value::Array(resp.history.iter().map(policy_value).collect()),
            ),
            (
                5,
                resp.default_bookmark
                    .clone()
                    .map(Value::Text)
                    .unwrap_or(Value::Null),
            ),
        ]);
        if let Value::Map(m) = &mut v {
            if let Some(log) = &resp.log {
                m.insert(6, Value::Text(log.clone()));
            }
            if !resp.withdrawn.is_empty() {
                m.insert(8, tags_value(&resp.withdrawn));
            }
            if !resp.witnesses.is_empty() {
                m.insert(
                    7,
                    Value::Array(
                        resp.witnesses
                            .iter()
                            .map(|(name, id)| {
                                Value::Array(vec![
                                    Value::Text(name.clone()),
                                    Value::Bytes(id.clone()),
                                ])
                            })
                            .collect(),
                    ),
                );
            }
        }
        cbor::encode(&v)
    }
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "refs")?;
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
            .try_into()
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
        let mut refs = Vec::new();
        for r in o::array(field(m, 0, "refs")?, "refs")? {
            let rm = map(r, "ref")?;
            let ls = field(rm, 4, "last serial")?;
            refs.push(RemoteRef {
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
                state: o::oref(field(rm, 1, "state")?, "state")?,
                heads: hashes_from(field(rm, 2, "heads")?)?,
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
                last_serial: if ls.is_null() {
                    None
                } else {
                    Some(o::uint(ls, "last serial")?)
                },
                accepted: match rm.get(&5) {
                    Some(a) => opt_attest_from(a)?,
                    None => None,
                },
                logged: rm.get(&6).map(logged_from).transpose()?,
                integration: match rm.get(&7) {
                    Some(l) => integration_from(l)?,
                    None => Vec::new(),
                },
            });
        }
        let policy = match m.get(&2) {
            Some(p) => opt_policy_from(p)?,
            None => None,
        };
        let tags = match m.get(&3) {
            Some(t) => tags_from(t)?,
            None => Vec::new(),
        };
        let mut history = Vec::new();
        if let Some(h) = m.get(&4) {
            for p in o::array(h, "history")? {
                history.push(policy_from(p)?);
            }
        }
        let default_bookmark = match m.get(&5) {
            Some(Value::Text(t)) => Some(t.clone()),
            _ => None,
        };
        let log = match m.get(&6) {
            Some(Value::Text(t)) => Some(t.clone()),
            _ => None,
        };
        let mut witnesses = Vec::new();
        if let Some(w) = m.get(&7) {
            for x in o::array(w, "witnesses")? {
                let a = o::array(x, "witness")?;
                if a.len() != 2 {
                    return Err(schema("witness: expected [name, identity]"));
                }
                witnesses.push((
                    o::text(&a[0], "witness name")?.to_owned(),
                    o::bytes(&a[1], "witness identity")?.to_vec(),
                ));
            }
        }
        let withdrawn = match m.get(&8) {
            Some(t) => tags_from(t)?,
            None => Vec::new(),
        };
        Ok((
            id,
            RefsResponse {
                refs,
                policy,
                tags,
                history,
                default_bookmark,
                log,
                witnesses,
                withdrawn,
            },
        ))
    }

    /// A proof's hashes, from the leaf upwards (spec §14.8).
    fn proof_value(proof: &[[u8; 32]]) -> Value {
        Value::Array(proof.iter().map(|h| Value::Bytes(h.to_vec())).collect())
    }

    fn proof_from(v: &Value) -> Result<Vec<[u8; 32]>> {
        let mut proof = Vec::new();
        for h in o::array(v, "proof")? {
            proof.push(
                o::bytes(h, "proof hash")?
                    .try_into()
                    .map_err(|_| schema("proof hash: expected 32 bytes"))?,
            );
        }
        Ok(proof)
    }

    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, hashes_value(have)),
            (1, hashes_value(want)),
        ]))
    }
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "negotiate")?;
        Ok((
            hashes_from(field(m, 0, "have")?)?,
            hashes_from(field(m, 1, "want")?)?,
        ))
    }

    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, hashes_value(hs)),
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
        ]))
    }
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "fetch")?;
        let max = match m.get(&1) {
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
            _ => None,
        };
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
    }

    /// `shallow` request: `{0: [heads], 1: depth}`.
    pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, hashes_value(hs)),
            (1, Value::Uint(depth as u64)),
        ]))
    }
    pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec<ChangeHash>, u32)> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "shallow")?;
        Ok((
            hashes_from(field(m, 0, "heads")?)?,
            o::u32_(field(m, 1, "depth")?, "depth")?,
        ))
    }

    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
    }
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
    }

    pub fn encode_push(
        bundle: &Bundle,
        updates: &[RefUpdate],
        policy: Option<&SignedPolicy>,
        tags: &[TagRef],
    ) -> Vec<u8> {
        cbor::encode(&Value::map([
            (0, bundle_value(bundle)),
            (2, opt_policy_value(policy)),
            (3, tags_value(tags)),
            (
                1,
                Value::Array(
                    updates
                        .iter()
                        .map(|u| {
                            Value::map([
                                (0, Value::Text(u.name.clone())),
                                (1, Value::Bytes(u.old_state.0.to_vec())),
                                (2, Value::Bytes(u.new_state.0.to_vec())),
                                (3, Value::Bool(u.force)),
                                (4, opt_attest_value(&u.attestation)),
                            ])
                        })
                        .collect(),
                ),
            ),
        ]))
    }
    /// A decoded push: bundle, bookmark updates, policy, tags.
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);

    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "push")?;
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
        let mut updates = Vec::new();
        for u in o::array(field(m, 1, "updates")?, "updates")? {
            let um = map(u, "update")?;
            updates.push(RefUpdate {
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
                force: field(um, 3, "force")?
                    .as_bool()
                    .ok_or_else(|| schema("force: expected bool"))?,
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
            });
        }
        let policy = match m.get(&2) {
            Some(p) => opt_policy_from(p)?,
            None => None,
        };
        let tags = match m.get(&3) {
            Some(t) => tags_from(t)?,
            None => Vec::new(),
        };
        Ok((bundle, updates, policy, tags))
    }

    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
        cbor::encode(&Value::map([(
            0,
            Value::Array(
                results
                    .iter()
                    .map(|r| {
                        Value::map([
                            (0, Value::Text(r.name.clone())),
                            (1, Value::Bool(r.ok)),
                            (2, Value::Text(r.reason.clone())),
                        ])
                    })
                    .collect(),
            ),
        )]))
    }
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
        let m = map(&v, "push response")?;
        let mut out = Vec::new();
        for r in o::array(field(m, 0, "results")?, "results")? {
            let rm = map(r, "result")?;
            out.push(RefResult {
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
                ok: field(rm, 1, "ok")?
                    .as_bool()
                    .ok_or_else(|| schema("ok: expected bool"))?,
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
            });
        }
        Ok(out)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::attest::AttestBody;
    use crate::objects::Hlc;

    /// oxygen#133: a proxy's HTML error page is shown by its title, and a
    /// server's own text as it is; a 413 says to push in smaller parts.
    #[test]
    fn html_error_pages_are_shown_by_their_title() {
        let cloudflare = "<html>\r\n<head><title>413 Request Entity Too Large</title></head>\r\n<body><center><h1>413</h1></center><hr><center>cloudflare</center></body></html>";
        assert_eq!(
            plain_error_text(cloudflare),
            "413 Request Entity Too Large (an HTML error page, from a proxy in front of the server perhaps)"
        );
        assert_eq!(
            plain_error_text("  refused: a secret\nat a.txt:3 "),
            "refused: a secret\nat a.txt:3"
        );
        assert!(hint_for("https://hub/x: HTTP 413: too large")
            .unwrap()
            .contains("push.batch-bytes"));
    }

    #[test]
    fn client_features_are_declared_and_read() {
        assert_eq!(CLIENT_FEATURES.join(","), "certified-tags");
        assert!(user_agent().starts_with("oxy/"));
        assert!(declares_feature(
            Some("certified-tags"),
            FEATURE_CERTIFIED_TAGS
        ));
        assert!(declares_feature(
            Some("later-thing, Certified-Tags ,x"),
            FEATURE_CERTIFIED_TAGS
        ));
        assert!(!declares_feature(
            Some("certified-tags-v2"),
            FEATURE_CERTIFIED_TAGS
        ));
        assert!(!declares_feature(Some(""), FEATURE_CERTIFIED_TAGS));
        assert!(!declares_feature(None, FEATURE_CERTIFIED_TAGS));
    }

    #[test]
    fn wire_round_trips() {
        let c = crate::objects::tests::sample_change();
        let sk = crate::identity::SecretKey {
            ed_seed: [1; 32],
            ml_seed: [2; 32],
        };
        let id = sk.identity("a", "b", 0);
        let key = id.hash();
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
        let seal = Attest::seal([3; 16], vec![c.hash()]);
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
        let policy = Policy::single_root([3; 16], id.clone(), 99);
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
        let bundle = Bundle {
            changes: vec![(c.clone(), vec![sig.clone()])],
            attests: vec![SignedAttest {
                attest: seal.clone(),
                sigs: vec![seal_sig],
            }],
            objects: vec![
                (ObjectType::Identity, id.encode()),
                (ObjectType::Chunk, b"xyz".to_vec()),
            ],
            policies: vec![SignedPolicy {
                policy: policy.clone(),
                sigs: vec![psig.clone()],
            }],
            hashes: vec![],
            raw_changes: vec![],
            certificates: vec![],
        };
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
        assert_eq!(
            d.hashes,
            vec![c.hash()],
            "the receiver hashes what it was sent"
        );
        assert_eq!(d.changes, bundle.changes);
        assert_eq!(d.attests, bundle.attests);
        assert_eq!(d.objects, bundle.objects);
        assert_eq!(d.policies, bundle.policies);
        assert!(d.certificates.is_empty());
        // Certificates (keyless, oxygenhub#241) travel as field 4, only
        // when there are any: a bundle without them encodes as before.
        let plain = wire::encode_bundle(&bundle);
        let mut with = bundle.clone();
        let issuer_sk = crate::identity::SecretKey::generate();
        let issuer = issuer_sk.identity("hub", "h@x", 1);
        with.certificates.push(crate::cert::SignedCertificate::sign(
            crate::cert::Certificate {
                key: id.clone(),
                issuer: issuer.hash(),
                subject: "agent:a@b".into(),
                claims: vec![],
                scope: crate::identity::Scope::CHANGES,
                not_before: 1,
                expires: 2,
                grant: None,
            },
            &issuer,
            &issuer_sk,
        ));
        let encoded = wire::encode_bundle(&with);
        assert_ne!(encoded, plain);
        assert_eq!(
            wire::decode_bundle(&encoded).unwrap().certificates,
            with.certificates
        );
        with.certificates.clear();
        assert_eq!(wire::encode_bundle(&with), plain);

        let att = Attest {
            repo: [3; 16],
            hlc: Hlc::default(),
            body: AttestBody::Ref {
                name: "main".into(),
                old_state: ObjectRef([1; 32]),
                new_state: ObjectRef([2; 32]),
                serial: 4,
            },
        };
        let refs = vec![
            RemoteRef {
                name: "main".into(),
                state: ObjectRef([2; 32]),
                heads: vec![c.hash()],
                attestation: Some(SignedAttest {
                    attest: att.clone(),
                    sigs: vec![sig.clone()],
                }),
                last_serial: Some(4),
                accepted: None,
                logged: None,
                integration: vec![],
            },
            RemoteRef {
                name: "dev".into(),
                state: ObjectRef([5; 32]),
                heads: vec![],
                attestation: None,
                last_serial: None,
                accepted: None,
                logged: None,
                integration: vec![],
            },
        ];
        let tag = TagRef::new("v1".into(), ObjectRef([2; 32]), vec![c.hash()], None);
        let mut resp = RefsResponse {
            refs,
            policy: Some(SignedPolicy {
                policy: policy.clone(),
                sigs: vec![psig.clone()],
            }),
            tags: vec![tag.clone()],
            history: vec![SignedPolicy {
                policy: policy.clone(),
                sigs: vec![],
            }],
            default_bookmark: Some("develop".into()),
            log: Some("hub.example/log\n2\nroot\n\n- sig\n".into()),
            witnesses: vec![],
            withdrawn: vec![],
        };
        // Without integrated times (spec §14.8) nothing is added: an old
        // peer's answer, and what it reads.
        let old = wire::encode_refs([3; 16], &resp);
        let back = wire::decode_refs(&old).unwrap().1;
        assert!(back.witnesses.is_empty() && back.refs.iter().all(|r| r.integration.is_empty()));
        resp.refs[1].logged = Some(Logged {
            index: 41,
            entry: vec![1, 2, 3],
            proof: vec![[9; 32], [8; 32]],
        });
        resp.refs[1].integration = vec![Integration {
            note: "hub.example/log\n42\nroot\n\n- sig\n".into(),
            proof: vec![[7; 32]],
        }];
        resp.witnesses = vec![("w.example".into(), vec![4, 5, 6])];
        let new = wire::encode_refs([3; 16], &resp);
        let (id_back, back) = wire::decode_refs(&new).unwrap();
        assert_eq!(back.log, resp.log);
        assert_eq!(back.refs[1].logged, resp.refs[1].logged);
        assert_eq!(back.refs[1].integration, resp.refs[1].integration);
        assert_eq!(back.witnesses, resp.witnesses);
        assert_eq!(back.refs[0].logged, None);
        // The new keys are 7s an old reader never asks for.
        let crate::cbor::Value::Map(m) = crate::cbor::decode(&new).unwrap() else {
            panic!()
        };
        assert!(m.contains_key(&7));
        assert_eq!(id_back, [3; 16]);
        assert_eq!(back.refs.len(), 2);
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
        assert_eq!(back.refs[1].last_serial, None);
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
        assert_eq!(back.history, resp.history);
        assert_eq!(back.tags, vec![tag.clone()]);

        // A tag's acceptance and log place (oxygen#129): keys 4 to 6, only
        // when there are any, so a tag without them encodes as before.
        let crate::cbor::Value::Map(m) = crate::cbor::decode(&new).unwrap() else {
            panic!()
        };
        let crate::cbor::Value::Array(tags) = &m[&3] else {
            panic!()
        };
        let crate::cbor::Value::Map(plain) = &tags[0] else {
            panic!()
        };
        assert_eq!(plain.keys().copied().collect::<Vec<_>>(), vec![0, 1, 2, 3]);
        let mut full = tag.clone();
        full.name = "v2".into();
        full.attestation = Some(SignedAttest {
            attest: att.clone(),
            sigs: vec![sig.clone()],
        });
        full.accepted = full.attestation.clone();
        full.logged = Some(Logged {
            index: 7,
            entry: vec![4],
            proof: vec![[3; 32]],
        });
        full.integration = vec![Integration {
            note: "hub.example/log\n8\nroot\n\n- sig\n".into(),
            proof: vec![],
        }];
        resp.tags = vec![full.clone(), tag.clone()];
        let bytes = wire::encode_refs([3; 16], &resp);
        assert_eq!(wire::decode_refs(&bytes).unwrap().1.tags, resp.tags);
        // In a push too.
        let pushed = wire::encode_push(&Bundle::default(), &[], None, &resp.tags);
        assert_eq!(wire::decode_push(&pushed).unwrap().3, resp.tags);
        // A reader that knows only keys 0 to 3 (oxy 0.5.0 and older, and
        // the runners built with it) reads the same tags from the new answer.
        let stripped: Vec<TagRef> = resp
            .tags
            .iter()
            .map(|t| {
                TagRef::new(
                    t.name.clone(),
                    t.state,
                    t.heads.clone(),
                    t.attestation.clone(),
                )
            })
            .collect();
        assert_eq!(wire::tags_as_0_5_read_them(&bytes).unwrap(), stripped);

        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
            &[ObjectRef([1; 32])],
            &[ObjectRef([2; 32])],
        ))
        .unwrap();
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
        assert_eq!(
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
            vec![c.hash()]
        );
        assert_eq!(
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
            (vec![c.hash()], Some(4096))
        );
        assert_eq!(
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
            (vec![], None)
        );
        assert_eq!(
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
            (vec![c.hash()], 3)
        );
        // An old-style fetch request without the filter field still decodes.
        assert_eq!(
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
            (vec![c.hash()], None)
        );

        let up = RefUpdate {
            name: "main".into(),
            old_state: ObjectRef([1; 32]),
            new_state: ObjectRef([2; 32]),
            force: true,
            attestation: Some(SignedAttest {
                attest: att,
                sigs: vec![sig],
            }),
        };
        let sp = SignedPolicy {
            policy,
            sigs: vec![psig],
        };
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
            &bundle,
            std::slice::from_ref(&up),
            Some(&sp),
            std::slice::from_ref(&tag),
        ))
        .unwrap();
        assert_eq!(pol, Some(sp));
        assert_eq!(tags, vec![tag]);
        assert_eq!(b2.changes.len(), 1);
        assert_eq!(ups[0].name, "main");
        assert!(ups[0].force);
        assert_eq!(ups[0].attestation, up.attestation);
        let res = vec![RefResult {
            name: "main".into(),
            ok: false,
            reason: "stale".into(),
        }];
        assert_eq!(
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
            res
        );
        assert_eq!(
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
            [7; 16]
        );
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/oxy-core/src/protocol.rs 86.1%

Keyboard
