OXYGENHUB

Coverage / crates/oxy-core/src/protocol.rs 84.0%

857 of 1020 lines run, 144 of 223 functions · run #341 on main · run missed

1
//! Transport-agnostic protocol (spec §16): what a client asks of a remote and
2
//! what a server does with a push. `LocalRemote` serves another repository
3
//! directory in-process, `HttpRemote` speaks to an `oxy-server` over HTTP(S),
4
//! and the server side reuses `Repo::receive_push` and friends unchanged.
5
//!
6
//! Wire messages are deterministic CBOR (`wire` module); over HTTP each is a
7
//! request or response body with content type `application/oxy-cbor`.
8
9
use crate::attest::Attest;
10
use crate::cbor::{self, Value};
11
use crate::hash::{ChangeHash, ObjectRef, ObjectType};
12
use crate::identity::Signature;
13
use crate::objects::{self as o, Change, ObjectError};
14
use crate::policy::Policy;
15
use crate::repo::{Repo, RepoError, Result};
16
use std::path::{Path, PathBuf};
17
use std::sync::{Arc, Mutex};
18
19
pub const CONTENT_TYPE: &str = "application/oxy-cbor";
20
pub const PROTOCOL_VERSION: u64 = 1;
21
22
/// An attestation together with the signatures over its ObjectRef.
23
#[derive(Clone, Debug, PartialEq, Eq)]
24
pub struct SignedAttest {
25
    pub attest: Attest,
26
    pub sigs: Vec<Signature>,
27
}
28
29
/// A policy together with the root signatures over its ObjectRef.
30
#[derive(Clone, Debug, PartialEq, Eq)]
31
pub struct SignedPolicy {
32
    pub policy: Policy,
33
    pub sigs: Vec<Signature>,
34
}
35
36
impl SignedPolicy {
37
    /// The proposal-file encoding (`wire::policy_value`).
381×
    pub fn encode(&self) -> Vec<u8> {
391×
        crate::cbor::encode(&wire::policy_value(self))
401×
    }
411×
    pub fn decode(b: &[u8]) -> Result<SignedPolicy> {
421×
        wire::policy_from(
431×
            &crate::cbor::decode(b).map_err(|e| RepoError::Object(ObjectError::from(e)))?,
44
        )
451×
    }
46
}
47
48
/// What `refs` returns: every bookmark, every tag, and the current policy.
49
#[derive(Clone, Debug, Default)]
50
pub struct RefsResponse {
51
    pub refs: Vec<RemoteRef>,
52
    pub policy: Option<SignedPolicy>,
53
    pub tags: Vec<TagRef>,
54
    /// Earlier policies, ascending by serial, so a client behind by more
55
    /// than one update can verify each step (spec §14.5).
56
    pub history: Vec<SignedPolicy>,
57
    /// The bookmark a clone checks out (spec §16.1), when the remote names one.
58
    pub default_bookmark: Option<String>,
59
    /// The host's latest transparency-log checkpoint, a signed note (spec
60
    /// §14.8), against which `RemoteRef::logged` proofs are made (key 6,
61
    /// only when the host keeps a log).
62
    pub log: Option<String>,
63
}
64
65
/// Where a reference attestation sits in the host's transparency log
66
/// (spec §14.8): its entry number, the entry, and the inclusion proof
67
/// against the checkpoint served beside it.
68
#[derive(Clone, Debug, PartialEq, Eq)]
69
pub struct Logged {
70
    pub index: u64,
71
    pub entry: Vec<u8>,
72
    pub proof: Vec<[u8; 32]>,
73
}
74
75
/// A tag as a remote reports it (spec §10.1).
76
#[derive(Clone, Debug, PartialEq, Eq)]
77
pub struct TagRef {
78
    pub name: String,
79
    pub state: ObjectRef,
80
    pub heads: Vec<ChangeHash>,
81
    pub attestation: Option<SignedAttest>,
82
}
83
84
/// One reference as a remote reports it.
85
#[derive(Clone, Debug)]
86
pub struct RemoteRef {
87
    pub name: String,
88
    pub state: ObjectRef,
89
    pub heads: Vec<ChangeHash>,
90
    /// The latest attestation, only when it describes `state`. An unattested
91
    /// push (accepted while no trust policy exists) leaves it `None`.
92
    pub attestation: Option<SignedAttest>,
93
    /// Serial of the latest attestation ever recorded, stale or not, so the
94
    /// next attestation continues the sequence.
95
    pub last_serial: Option<u64>,
96
    /// The host's acceptance of `attestation` when a short-lived key signed
97
    /// it (spec §14.7): the time its certificate is checked at.
98
    pub accepted: Option<SignedAttest>,
99
    /// The attestation's place in the host's transparency log (key 6, only
100
    /// when it has one).
101
    pub logged: Option<Logged>,
102
}
103
104
/// Everything needed to add a set of changes to a repository.
105
#[derive(Clone, Debug, Default)]
106
pub struct Bundle {
107
    /// In dependency order, each with its individual signatures.
108
    pub changes: Vec<(Change, Vec<Signature>)>,
109
    /// Seals covering the changes.
110
    pub attests: Vec<SignedAttest>,
111
    /// Chunks, blobs, identities, and state objects referenced by the changes.
112
    pub objects: Vec<(ObjectType, Vec<u8>)>,
113
    /// Policies between the receiver's and the one sent with the push,
114
    /// ascending by serial (spec §14.5).
115
    pub policies: Vec<SignedPolicy>,
116
    /// Each change's hash, in the order of `changes`, when whoever made
117
    /// the bundle knows it: the store it came from, or the bytes it was
118
    /// decoded from. Never read from the wire. Empty when unknown.
119
    pub hashes: Vec<ChangeHash>,
120
    /// Changes as stored, with their signatures, in dependency order, for a
121
    /// fetch between repositories on one machine: their text blobs and
122
    /// chunks are among `objects`. Never on the wire.
123
    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,
124
    /// Certificates of short-lived keys that signed what the bundle carries
125
    /// (spec §14.7), their issuers' identities among `objects`.
126
    pub certificates: Vec<crate::cert::SignedCertificate>,
127
}
128
129
impl Bundle {
13027×
    pub fn is_empty(&self) -> bool {
13127×
        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()
13227×
    }
133
}
134
135
/// A requested bookmark move (spec §10.2, §16.3).
136
#[derive(Clone, Debug)]
137
pub struct RefUpdate {
138
    pub name: String,
139
    pub old_state: ObjectRef,
140
    pub new_state: ObjectRef,
141
    pub force: bool,
142
    pub attestation: Option<SignedAttest>,
143
}
144
145
#[derive(Clone, Debug, PartialEq, Eq)]
146
pub struct RefResult {
147
    pub name: String,
148
    pub ok: bool,
149
    pub reason: String,
150
}
151
152
pub trait Remote {
153
    fn repo_id(&self) -> Result<[u8; 16]>;
154
    fn refs(&self) -> Result<RefsResponse>;
155
    /// Changes in the closure of `want` but not of `have`, in dependency order.
156
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>>;
157
    /// The changes with their signatures, seals, identities, and blobs; blobs
158
    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).
159
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle>;
160
    /// Individual objects by reference (spec §16.1 `objects`); a blob comes
161
    /// with its chunks.
162
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>>;
163
    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the
164
    /// checkpoint object, its payload blob, and the blob's chunks.
165
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle>;
166
    /// The changes within `depth` dependency steps of `heads` (spec §17).
167
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>>;
168
    /// Push changes, bookmark updates, new tags, and optionally a newer
169
    /// policy. The policy is applied first; a result named `policy` reports
170
    /// it, and each tag reports as `tags/<name>`.
171
    fn push(
172
        &self,
173
        bundle: Bundle,
174
        updates: Vec<RefUpdate>,
175
        policy: Option<SignedPolicy>,
176
        tags: Vec<TagRef>,
177
    ) -> Result<Vec<RefResult>>;
178
}
179
180
// ---------------------------------------------------------------- local
181
182
/// Another repository on this machine, served in-process.
183
pub struct LocalRemote {
184
    pub repo: Repo,
185
    /// The pushing repository's id, which binds an unbound target on the
186
    /// first push, as `Oxy-Repo-Id` does over HTTP.
187
    pub claimed_id: Option<[u8; 16]>,
188
}
189
190
impl LocalRemote {
1911085×
    pub fn open(path: &Path) -> Result<LocalRemote> {
192
        Ok(LocalRemote {
1931085×
            repo: Repo::open(path)?,
1941085×
            claimed_id: None,
195
        })
1961085×
    }
197
}
198
199
impl Remote for LocalRemote {
200853×
    fn repo_id(&self) -> Result<[u8; 16]> {
201853×
        self.repo.id()
202853×
    }
203852×
    fn refs(&self) -> Result<RefsResponse> {
204852×
        self.repo.remote_refs()
205852×
    }
206321×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
207321×
        self.repo.negotiate(have, want)
208321×
    }
209324×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
210324×
        self.repo.bundle_raw_for(changes, max_blob)
211324×
    }
2123×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
2133×
        self.repo.objects_for(refs)
2143×
    }
2151×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
2161×
        self.repo.checkpoint_bundle_for(heads)
2171×
    }
2183×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
2193×
        self.repo.shallow_changes(heads, depth)
2203×
    }
221251×
    fn push(
222251×
        &self,
223251×
        bundle: Bundle,
224251×
        updates: Vec<RefUpdate>,
225251×
        policy: Option<SignedPolicy>,
226251×
        tags: Vec<TagRef>,
227251×
    ) -> Result<Vec<RefResult>> {
228251×
        if self.repo.is_unbound() {
22911×
            match self.claimed_id {
23011×
                Some(id) => self.repo.bind_id(id)?,
231
                None => {
2320×
                    return Err(RepoError::Msg(
2330×
                        "the remote is unbound and this client did not say which repository it is"
2340×
                            .into(),
2350×
                    ))
236
                }
237
            }
238240×
        }
239251×
        self.repo.receive_push(bundle, updates, policy, tags)
240251×
    }
241
}
242
243
// ---------------------------------------------------------------- http
244
245
/// How to reach a remote: credentials and trust.
246
#[derive(Clone, Debug, Default)]
247
pub struct RemoteOptions {
248
    /// Bearer token for pushes.
249
    pub token: Option<String>,
250
    /// PEM file with extra root certificates to trust for `https://` remotes
251
    /// (self-hosted servers); the system web roots are always trusted.
252
    pub ca: Option<PathBuf>,
253
    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound
254
    /// hosted repository can adopt it on the first push (spec §16.1).
255
    pub repo_id: Option<[u8; 16]>,
256
    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its
257
    /// mirror and what it has seen: the remote's directory under
258
    /// `.oxy/remotes`. None (a clone, before there is a repository) means
259
    /// a temporary directory.
260
    pub cache_dir: Option<PathBuf>,
261
    /// Other tokens for the same hub, tried in turn when `token` is
262
    /// refused (401, 403 or 404): those of enclosing checkouts, when
263
    /// `token` came from the credential store rather than from this
264
    /// remote's own configuration (oxygen#112).
265
    pub fallback_tokens: Vec<String>,
266
}
267
268
/// Opens a store for a URL it understands, or passes (`None`). Stores that
269
/// need more than the standard library (the cloud ones) are added by the
270
/// program that links them, with [`register_store_opener`].
271
pub type StoreOpener =
272
    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;
273
274
static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());
275
276154×
pub fn register_store_opener(opener: StoreOpener) {
277154×
    STORE_OPENERS
278154×
        .lock()
279154×
        .unwrap_or_else(|p| p.into_inner())
280154×
        .push(opener);
281154×
}
282
283
/// A store for `url`, from `dir://` or any registered opener.
2841145×
pub fn open_store(
2851145×
    url: &str,
2861145×
    options: &RemoteOptions,
2871145×
) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {
2881145×
    if let Some(path) = url.strip_prefix("dir://") {
289
        return Some(
29059×
            crate::strata::LocalStore::open(path)
29159×
                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),
292
        );
2931086×
    }
2941086×
    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());
2951086×
    openers.iter().find_map(|open| open(url, options))
2961145×
}
297
298
/// The request header carrying the client's repository id.
299
pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";
300
301
/// The id of a hosted repository that has not adopted one yet.
302
pub const UNBOUND_ID: [u8; 16] = [0; 16];
303
304
/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.
305
pub struct HttpRemote {
306
    base: String,
307
    /// The token in use, and those still to try when it is refused.
308
    token: Mutex<Option<String>>,
309
    fallbacks: Mutex<Vec<String>>,
310
    repo_id: Option<[u8; 16]>,
311
    agent: ureq::Agent,
312
}
313
314
/// A TLS client configuration trusting the web roots plus `ca`, if given:
315
/// the trust every oxy connection uses, shared with tools that call hub APIs.
316157×
pub fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {
317157×
    let mut roots = rustls::RootCertStore::empty();
318157×
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
319157×
    if let Some(path) = ca {
32043×
        let pem = std::fs::read(path)
32143×
            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;
32243×
        let certs = rustls_pemfile::certs(&mut pem.as_slice())
32343×
            .collect::<std::result::Result<Vec<_>, _>>()
32443×
            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;
32543×
        if certs.is_empty() {
3260×
            return Err(RepoError::Msg(format!(
3270×
                "CA file {} holds no certificates",
3280×
                path.display()
3290×
            )));
33043×
        }
33143×
        for c in certs {
33243×
            roots
33343×
                .add(c)
33443×
                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;
335
        }
336114×
    }
337157×
    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
338157×
        rustls::crypto::ring::default_provider(),
339
    ))
340157×
    .with_safe_default_protocol_versions()
341157×
    .map_err(|e| RepoError::Msg(e.to_string()))?
342157×
    .with_root_certificates(roots)
343157×
    .with_no_client_auth();
344157×
    Ok(Arc::new(cfg))
345157×
}
346
347
impl HttpRemote {
348153×
    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {
349153×
        refuse_plain_token(base, options)?;
350153×
        let agent = ureq::AgentBuilder::new()
351153×
            .tls_config(tls_client_config(options.ca.as_deref())?)
352153×
            .build();
353153×
        Ok(HttpRemote {
354153×
            base: base.trim_end_matches('/').to_string(),
355153×
            token: Mutex::new(options.token.clone()),
356153×
            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),
357153×
            repo_id: options.repo_id,
358153×
            agent,
359153×
        })
360153×
    }
361
362
    /// A request, tried again with the next fallback token while the
363
    /// hub refuses the one in use (oxygen#112).
364388×
    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {
365
        loop {
366389×
            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();
367389×
            match self.call_once(post, endpoint, body, token.as_deref()) {
36811×
                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {
3693×
                    let next = self
3703×
                        .fallbacks
3713×
                        .lock()
3723×
                        .unwrap_or_else(|p| p.into_inner())
3733×
                        .pop();
3743×
                    let Some(next) = next else {
3752×
                        return Err(RepoError::Msg(m));
376
                    };
3771×
                    crate::credentials::warn_once(&format!(
3781×
                        "{}: the stored token was refused; using the one an enclosing checkout has for this hub",
3791×
                        self.base
3801×
                    ));
3811×
                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);
382
                }
383386×
                other => return other,
384
            }
385
        }
386388×
    }
387
388389×
    fn call_once(
389389×
        &self,
390389×
        post: bool,
391389×
        endpoint: &str,
392389×
        body: Option<&[u8]>,
393389×
        token: Option<&str>,
394389×
    ) -> Result<Vec<u8>> {
395389×
        let url = format!("{}/oxy/v1/{endpoint}", self.base);
396389×
        let mut req = if post {
397134×
            self.agent.post(&url)
398
        } else {
399255×
            self.agent.get(&url)
400
        };
401389×
        req = req.set("Accept", CONTENT_TYPE);
402389×
        if let Some(t) = token {
40322×
            req = req.set("Authorization", &format!("Bearer {t}"));
404367×
        }
405389×
        if let Some(id) = &self.repo_id {
406364×
            req = req.set(REPO_ID_HEADER, &hex::encode(id));
407364×
        }
408389×
        let resp = match body {
409134×
            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),
410255×
            None => req.call(),
411
        };
41211×
        match resp {
413378×
            Ok(r) => {
414378×
                let mut out = Vec::new();
415378×
                r.into_reader()
416378×
                    .read_to_end(&mut out)
417378×
                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
418378×
                Ok(out)
419
            }
4208×
            Err(ureq::Error::Status(code, r)) => {
4218×
                let text = r.into_string().unwrap_or_default();
4228×
                let msg = format!("{url}: HTTP {code}: {}", text.trim());
4238×
                Err(RepoError::Msg(with_hint(msg)))
424
            }
425
            // The transport error already names the URL.
4263×
            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
427
        }
428389×
    }
429
}
430
431
/// Whether a failed request's message says the hub refused its token:
432
/// HTTP 401, 403, or 404 (a hub hides a private repository from a token
433
/// that may not read it).
4344×
fn refused(message: &str) -> bool {
4354×
    [": HTTP 401", ": HTTP 403", ": HTTP 404"]
4364×
        .iter()
4376×
        .any(|c| message.contains(c))
4384×
}
439
440
/// Refuse to send a token where anyone on the way could read it: plain
441
/// `http` to another machine (see [`crate::credentials::token_allowed`]).
442153×
fn refuse_plain_token(url: &str, options: &RemoteOptions) -> Result<()> {
443153×
    if options.token.is_some() && !crate::credentials::token_allowed(url) {
4440×
        return Err(RepoError::Msg(format!(
4450×
            "{url}: refusing to send a token over plain http; use https"
4460×
        )));
447153×
    }
448153×
    Ok(())
449153×
}
450
451
/// What to do about a failed request, recognised from its message.
45219×
pub fn hint_for(message: &str) -> Option<&'static str> {
45319×
    let m = message.to_ascii_lowercase();
45419×
    let hint = if m.contains("http 401") {
4556×
        "the hub needs you signed in, or did not accept the token: run `oxy auth login HUB` (pushing needs repo:write)"
45613×
    } else if m.contains("http 403") {
4571×
        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"
45812×
    } else if m.contains("http 404") {
4594×
        "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`)"
4608×
    } else if m.contains("unknownissuer")
4616×
        || m.contains("unknown issuer")
4626×
        || m.contains("certificate")
4635×
        || m.contains("invalidcertificate")
464
    {
4653×
        "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`"
4665×
    } else if m.contains("connection refused") {
4673×
        "nothing is listening there: check the URL and port, and that the server is running"
4682×
    } else if m.contains("dns")
4691×
        || m.contains("failed to lookup")
4701×
        || m.contains("name or service not known")
471
    {
4721×
        "the host name does not resolve: check the URL"
4731×
    } else if m.contains("timed out") || m.contains("timeout") {
4740×
        "the server did not answer in time: check the network, or try again"
475
    } else {
4761×
        return None;
477
    };
47818×
    Some(hint)
47919×
}
480
481
/// `message`, followed by what to do about it when that is known.
48211×
pub fn with_hint(message: String) -> String {
48311×
    match hint_for(&message) {
48411×
        Some(h) => format!("{message}\n  hint: {h}"),
4850×
        None => message,
486
    }
48711×
}
488
489
/// A JSON request to a hub API (`GET`, `POST` or `DELETE`), with the same
490
/// TLS trust and token rules as remotes. Returns the status and body, an
491
/// error status included; only transport failures are errors.
4920×
pub fn http_json(
4930×
    method: &str,
4940×
    url: &str,
4950×
    body: Option<&str>,
4960×
    options: &RemoteOptions,
4970×
) -> Result<(u16, String)> {
4980×
    refuse_plain_token(url, options)?;
4990×
    let agent = ureq::AgentBuilder::new()
5000×
        .tls_config(tls_client_config(options.ca.as_deref())?)
5010×
        .timeout(std::time::Duration::from_secs(60))
5020×
        .build();
5030×
    let mut req = agent.request(method, url).set("Accept", "application/json");
5040×
    if let Some(t) = &options.token {
5050×
        req = req.set("Authorization", &format!("Bearer {t}"));
5060×
    }
5070×
    let resp = match body {
5080×
        Some(b) => req.set("Content-Type", "application/json").send_string(b),
5090×
        None => req.call(),
510
    };
5110×
    match resp {
5120×
        Ok(r) => {
5130×
            let code = r.status();
5140×
            Ok((code, r.into_string().unwrap_or_default()))
515
        }
5160×
        Err(ureq::Error::Status(code, r)) => Ok((code, r.into_string().unwrap_or_default())),
5170×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
518
    }
5190×
}
520
521
/// An authenticated GET of any URL on a host, with the same TLS trust and
522
/// token as remotes use (for hub APIs: proposals). Returns the body.
5230×
pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {
5240×
    refuse_plain_token(url, options)?;
5250×
    let agent = ureq::AgentBuilder::new()
5260×
        .tls_config(tls_client_config(options.ca.as_deref())?)
5270×
        .build();
5280×
    let mut req = agent.get(url).set("Accept", "application/json");
5290×
    if let Some(t) = &options.token {
5300×
        req = req.set("Authorization", &format!("Bearer {t}"));
5310×
    }
5320×
    match req.call() {
5330×
        Ok(r) => {
5340×
            let mut out = Vec::new();
5350×
            r.into_reader()
5360×
                .read_to_end(&mut out)
5370×
                .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;
5380×
            Ok(out)
539
        }
5400×
        Err(ureq::Error::Status(code, r)) => {
5410×
            let text = r.into_string().unwrap_or_default();
5420×
            Err(RepoError::Msg(with_hint(format!(
5430×
                "{url}: HTTP {code}: {}",
5440×
                text.trim()
5450×
            ))))
546
        }
547
        // The transport error already names the URL.
5480×
        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),
549
    }
5500×
}
551
552
impl Remote for HttpRemote {
553132×
    fn repo_id(&self) -> Result<[u8; 16]> {
554132×
        wire::decode_capabilities(&self.call(false, "capabilities", None)?)
555132×
    }
556123×
    fn refs(&self) -> Result<RefsResponse> {
557123×
        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)
558123×
    }
55949×
    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {
56049×
        wire::decode_hashes(&self.call(
561
            true,
56249×
            "negotiate",
56349×
            Some(&wire::encode_negotiate(have, want)),
5640×
        )?)
56549×
    }
56649×
    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {
56749×
        wire::decode_bundle(&self.call(
568
            true,
56949×
            "fetch",
57049×
            Some(&wire::encode_fetch(changes, max_blob)),
5710×
        )?)
57249×
    }
5731×
    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {
574
        Ok(
5751×
            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?
576
                .objects,
577
        )
5781×
    }
5790×
    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {
5800×
        wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?)
5810×
    }
5820×
    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {
5830×
        wire::decode_hashes(&self.call(
584
            true,
5850×
            "shallow",
5860×
            Some(&wire::encode_shallow(heads, depth)),
5870×
        )?)
5880×
    }
58934×
    fn push(
59034×
        &self,
59134×
        bundle: Bundle,
59234×
        updates: Vec<RefUpdate>,
59334×
        policy: Option<SignedPolicy>,
59434×
        tags: Vec<TagRef>,
59534×
    ) -> Result<Vec<RefResult>> {
59634×
        wire::decode_push_response(&self.call(
597
            true,
59834×
            "push",
59934×
            Some(&wire::encode_push(
60034×
                &bundle,
60134×
                &updates,
60234×
                policy.as_ref(),
60334×
                &tags,
60434×
            )),
6054×
        )?)
60634×
    }
607
}
608
609
/// A remote's URL as it is stored: an `http(s)://` URL as given, a local
610
/// path made absolute (with its `file://` kept), so it still names the same
611
/// repository from inside the clone (oxygen#57).
612263×
pub fn stored_url(url: &str) -> String {
613
    // Only a path is made absolute: a plain one, or one behind file:// or
614
    // dir://. Every other URL (a hub, a bucket) is kept as written.
615263×
    let (prefix, path) = match (url.strip_prefix("file://"), url.strip_prefix("dir://")) {
6161×
        (Some(p), _) => ("file://", p),
61711×
        (_, Some(p)) => ("dir://", p),
618251×
        _ if url.contains("://") => return url.to_string(),
619228×
        _ => ("", url),
620
    };
621240×
    match std::path::absolute(path) {
622240×
        Ok(abs) => format!("{prefix}{}", abs.display()),
6230×
        Err(_) => url.to_string(),
624
    }
625263×
}
626
627
/// Open a remote by URL: a local path (optionally `file://`), `http(s)://`
628
/// (a hub), or a store (`dir://`, and whatever openers are registered).
6291286×
pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {
6301286×
    if url.starts_with("http://") || url.starts_with("https://") {
631153×
        return Ok(Box::new(HttpRemote::new(url, options)?));
6321133×
    }
6331133×
    if let Some(store) = open_store(url, options) {
63448×
        return crate::repo::strata::open_store_remote(store?, options);
6351085×
    }
6361085×
    let path = url.strip_prefix("file://").unwrap_or(url);
6371085×
    if path.contains("://") {
6380×
        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));
6391085×
    }
6401085×
    let mut local = LocalRemote::open(Path::new(path))?;
6411085×
    local.claimed_id = options.repo_id;
6421085×
    Ok(Box::new(local))
6431286×
}
644
645
// ---------------------------------------------------------------- wire
646
647
/// CBOR encodings of the protocol messages (spec §16.1).
648
pub mod wire {
649
    use super::*;
650
6510×
    fn schema(msg: &str) -> RepoError {
6520×
        RepoError::Object(ObjectError::Schema(msg.into()))
6530×
    }
654147×
    fn sigs_value(sigs: &[Signature]) -> Value {
655147×
        Value::Array(sigs.iter().map(|s| s.to_value()).collect())
656147×
    }
657142×
    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {
658142×
        Ok(o::array(v, "signatures")?
659142×
            .iter()
660142×
            .map(Signature::from_value)
661142×
            .collect::<std::result::Result<Vec<_>, _>>()?)
662142×
    }
663366×
    fn hashes_value(hs: &[ObjectRef]) -> Value {
664535×
        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())
665366×
    }
666425×
    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {
667425×
        Ok(o::array(v, "hashes")?
668425×
            .iter()
669595×
            .map(|h| o::oref(h, "hash"))
670425×
            .collect::<std::result::Result<Vec<_>, _>>()?)
671425×
    }
67224×
    fn attest_value(sa: &SignedAttest) -> Value {
67324×
        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])
67424×
    }
67538×
    fn attest_from(v: &Value) -> Result<SignedAttest> {
67638×
        let a = o::array(v, "signed attest")?;
67738×
        if a.len() != 2 {
6780×
            return Err(schema("signed attest: expected [attest, sigs]"));
67938×
        }
680
        Ok(SignedAttest {
68138×
            attest: Attest::decode(o::bytes(&a[0], "attest")?)?,
68238×
            sigs: sigs_from(&a[1])?,
683
        })
68438×
    }
685197×
    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {
686197×
        sa.as_ref().map(attest_value).unwrap_or(Value::Null)
687197×
    }
688252×
    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {
689252×
        if v.is_null() {
690225×
            Ok(None)
691
        } else {
69227×
            attest_from(v).map(Some)
693
        }
694252×
    }
695
    /// `[policy bytes, [signature]]`: also the format of a policy proposal
696
    /// file exchanged between root holders (spec §14.5).
69710×
    pub fn policy_value(sp: &SignedPolicy) -> Value {
69810×
        Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)])
69910×
    }
7008×
    pub fn policy_from(v: &Value) -> Result<SignedPolicy> {
7018×
        let a = o::array(v, "signed policy")?;
7028×
        if a.len() != 2 {
7030×
            return Err(schema("signed policy: expected [policy, sigs]"));
7048×
        }
705
        Ok(SignedPolicy {
7068×
            policy: Policy::decode(o::bytes(&a[0], "policy")?)?,
7078×
            sigs: sigs_from(&a[1])?,
708
        })
7098×
    }
710186×
    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {
711186×
        sp.map(policy_value).unwrap_or(Value::Null)
712186×
    }
713225×
    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {
714225×
        if v.is_null() {
715222×
            Ok(None)
716
        } else {
7173×
            policy_from(v).map(Some)
718
        }
719225×
    }
720931×
    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {
721931×
        Ok(o::map(v, what)?)
722931×
    }
7232603×
    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {
7242603×
        Ok(o::field(m, k, what)?)
7252603×
    }
726
727
    /// A bundle's value. Each change's text blobs and their chunks (spec
728
    /// §6.1) are added to its objects when they are not already there, so a
729
    /// bundle can always be read on its own.
730120×
    pub fn bundle_value(b: &Bundle) -> Value {
731120×
        let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len());
732
        // Changes as stored are already their encoding, their texts among
733
        // the objects.
734120×
        for (bytes, s) in &b.raw_changes {
7350×
            changes.push(Value::Array(vec![
7360×
                Value::Bytes(bytes.clone()),
7370×
                sigs_value(s),
7380×
            ]));
7390×
        }
740120×
        let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new();
741120×
        let mut listed: Option<std::collections::HashSet<ObjectRef>> = None;
742120×
        for (c, s) in &b.changes {
743113×
            let (bytes, stored) = c.encode_parts();
744113×
            for t in stored {
7450×
                let listed = listed.get_or_insert_with(|| {
7460×
                    b.objects
7470×
                        .iter()
7480×
                        .map(|(ty, body)| crate::hash::hash_object(*ty, body))
7490×
                        .collect()
7500×
                });
7510×
                if !listed.insert(t.blob.hash()) {
7520×
                    continue;
7530×
                }
7540×
                for chunk in crate::chunk::chunks(&t.content) {
7550×
                    if listed.insert(crate::hash::hash_object(ObjectType::Chunk, chunk)) {
7560×
                        texts.push((ObjectType::Chunk, chunk.to_vec()));
7570×
                    }
758
                }
7590×
                texts.push((ObjectType::Blob, t.blob.encode()));
760
            }
761113×
            changes.push(Value::Array(vec![Value::Bytes(bytes), sigs_value(s)]));
762
        }
763120×
        let mut v = Value::map([
764120×
            (0, Value::Array(changes)),
765120×
            (
766120×
                1,
767120×
                Value::Array(b.attests.iter().map(attest_value).collect()),
768120×
            ),
769
            (
770
                2,
771
                Value::Array(
772120×
                    b.objects
773120×
                        .iter()
774120×
                        .chain(&texts)
775120×
                        .map(|(t, body)| {
77688×
                            Value::Array(vec![
77788×
                                Value::Text(t.name().into()),
77888×
                                Value::Bytes(body.clone()),
77988×
                            ])
78088×
                        })
781120×
                        .collect(),
782
                ),
783
            ),
784120×
            (
785120×
                3,
786120×
                Value::Array(b.policies.iter().map(policy_value).collect()),
787120×
            ),
788
        ]);
789
        // Only when there are any; older peers ignore the field.
790120×
        if !b.certificates.is_empty() {
7911×
            if let Value::Map(m) = &mut v {
7921×
                m.insert(
793
                    4,
794
                    Value::Array(
7951×
                        b.certificates
7961×
                            .iter()
7971×
                            .map(|c| Value::Bytes(c.encode()))
7981×
                            .collect(),
799
                    ),
800
                );
8010×
            }
802119×
        }
803120×
        v
804120×
    }
805
806110×
    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {
807110×
        let m = map(v, "bundle")?;
808110×
        let mut b = Bundle::default();
809110×
        for ob in o::array(field(m, 2, "objects")?, "objects")? {
81077×
            let pair = o::array(ob, "object")?;
81177×
            if pair.len() != 2 {
8120×
                return Err(schema("object: expected [type, body]"));
81377×
            }
81477×
            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?)
81577×
                .ok_or_else(|| schema("unknown object type"))?;
81677×
            b.objects
81777×
                .push((ty, o::bytes(&pair[1], "object body")?.to_vec()));
818
        }
819
        // Stored inserts are read from the bundle's own objects.
820110×
        let mut by_ref: Option<std::collections::HashMap<ObjectRef, usize>> = None;
821110×
        for c in o::array(field(m, 0, "changes")?, "changes")? {
82296×
            let pair = o::array(c, "change entry")?;
82396×
            if pair.len() != 2 {
8240×
                return Err(schema("change entry: expected [change, sigs]"));
82596×
            }
82696×
            let encoded = o::bytes(&pair[0], "change")?;
82796×
            b.hashes
82896×
                .push(crate::hash::hash_object(ObjectType::Change, encoded));
82996×
            let change = Change::decode_with(encoded, &mut |text, count, form| {
8300×
                let by_ref = by_ref.get_or_insert_with(|| {
8310×
                    b.objects
8320×
                        .iter()
8330×
                        .enumerate()
8340×
                        .map(|(i, (ty, body))| (crate::hash::hash_object(*ty, body), i))
8350×
                        .collect()
8360×
                });
8370×
                let get =
8380×
                    |r: &ObjectRef, want: ObjectType| match by_ref.get(r).map(|i| &b.objects[*i]) {
8390×
                        Some((ty, body)) if *ty == want => Ok(body),
8400×
                        _ => Err(ObjectError::Schema(format!(
8410×
                            "the bundle lacks {} {}",
8420×
                            want.name(),
8430×
                            r.short(12)
8440×
                        ))),
8450×
                    };
8460×
                let blob = crate::objects::Blob::decode(get(text, ObjectType::Blob)?)?;
8470×
                let mut content = Vec::with_capacity(blob.length as usize);
8480×
                for chunk in &blob.chunks {
8490×
                    content.extend_from_slice(get(chunk, ObjectType::Chunk)?);
850
                }
8510×
                crate::objects::text_lines(&content, count, form)
8520×
            })?;
85396×
            b.changes.push((change, sigs_from(&pair[1])?));
854
        }
855110×
        for a in o::array(field(m, 1, "attests")?, "attests")? {
85611×
            b.attests.push(attest_from(a)?);
857
        }
858110×
        if let Some(ps) = m.get(&3) {
859110×
            for p in o::array(ps, "policies")? {
8603×
                b.policies.push(policy_from(p)?);
861
            }
8620×
        }
863110×
        if let Some(cs) = m.get(&4) {
8641×
            for c in o::array(cs, "certificates")? {
8651×
                b.certificates.push(
8661×
                    crate::cert::SignedCertificate::decode(o::bytes(c, "certificate")?)
8671×
                        .map_err(|e| schema(&format!("certificate: {e}")))?,
868
                );
869
            }
870109×
        }
871110×
        Ok(b)
872110×
    }
873
87485×
    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {
87585×
        cbor::encode(&bundle_value(b))
87685×
    }
87779×
    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {
87879×
        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)
87979×
    }
880
881127×
    pub fn encode_capabilities(id: [u8; 16]) -> Vec<u8> {
882127×
        cbor::encode(&Value::map([
883127×
            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),
884127×
            (1, Value::Bytes(id.to_vec())),
885127×
            (2, Value::Array(vec![Value::Text("seal".into())])),
886127×
        ]))
887127×
    }
888127×
    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {
889127×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
890127×
        let m = map(&v, "capabilities")?;
891127×
        let versions = o::array(field(m, 0, "versions")?, "versions")?;
892127×
        if !versions
893127×
            .iter()
894127×
            .any(|x| x.as_uint() == Some(PROTOCOL_VERSION))
895
        {
8960×
            return Err(RepoError::Msg(
8970×
                "remote does not speak protocol version 1".into(),
8980×
            ));
899127×
        }
900127×
        o::bytes(field(m, 1, "repo id")?, "repo id")?
901127×
            .try_into()
902127×
            .map_err(|_| schema("repo id: expected 16 bytes"))
903127×
    }
904
905186×
    fn tags_value(tags: &[TagRef]) -> Value {
906
        Value::Array(
907186×
            tags.iter()
908186×
                .map(|t| {
9097×
                    Value::map([
9107×
                        (0, Value::Text(t.name.clone())),
9117×
                        (1, Value::Bytes(t.state.0.to_vec())),
9127×
                        (2, hashes_value(&t.heads)),
9137×
                        (3, opt_attest_value(&t.attestation)),
9147×
                    ])
9157×
                })
916186×
                .collect(),
917
        )
918186×
    }
919225×
    fn tags_from(v: &Value) -> Result<Vec<TagRef>> {
920225×
        let mut out = Vec::new();
921225×
        for t in o::array(v, "tags")? {
92223×
            let tm = map(t, "tag")?;
92323×
            out.push(TagRef {
92423×
                name: o::text(field(tm, 0, "name")?, "name")?.to_owned(),
92523×
                state: o::oref(field(tm, 1, "state")?, "state")?,
92623×
                heads: hashes_from(field(tm, 2, "heads")?)?,
92723×
                attestation: opt_attest_from(field(tm, 3, "attestation")?)?,
928
            });
929
        }
930225×
        Ok(out)
931225×
    }
932
933151×
    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {
934151×
        let mut v = Value::map([
935
            (
936
                0,
937
                Value::Array(
938151×
                    resp.refs
939151×
                        .iter()
940155×
                        .map(|r| {
941155×
                            let mut v = Value::map([
942155×
                                (0, Value::Text(r.name.clone())),
943155×
                                (1, Value::Bytes(r.state.0.to_vec())),
944155×
                                (2, hashes_value(&r.heads)),
945155×
                                (3, opt_attest_value(&r.attestation)),
946155×
                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),
947155×
                            ]);
948
                            // Only when there is one; older peers ignore it.
949155×
                            if let Value::Map(m) = &mut v {
950155×
                                if r.accepted.is_some() {
9510×
                                    m.insert(5, opt_attest_value(&r.accepted));
952155×
                                }
953155×
                                if let Some(l) = &r.logged {
9541×
                                    m.insert(
955
                                        6,
9561×
                                        Value::Array(vec![
9571×
                                            Value::Uint(l.index),
9581×
                                            Value::Bytes(l.entry.clone()),
959
                                            Value::Array(
9601×
                                                l.proof
9611×
                                                    .iter()
9622×
                                                    .map(|h| Value::Bytes(h.to_vec()))
9631×
                                                    .collect(),
964
                                            ),
965
                                        ]),
966
                                    );
967154×
                                }
9680×
                            }
969155×
                            v
970155×
                        })
971151×
                        .collect(),
972
                ),
973
            ),
974151×
            (1, Value::Bytes(id.to_vec())),
975151×
            (2, opt_policy_value(resp.policy.as_ref())),
976151×
            (3, tags_value(&resp.tags)),
977151×
            (
978151×
                4,
979151×
                Value::Array(resp.history.iter().map(policy_value).collect()),
980151×
            ),
981151×
            (
982151×
                5,
983151×
                resp.default_bookmark
984151×
                    .clone()
985151×
                    .map(Value::Text)
986151×
                    .unwrap_or(Value::Null),
987151×
            ),
988
        ]);
989151×
        if let (Some(log), Value::Map(m)) = (&resp.log, &mut v) {
9901×
            m.insert(6, Value::Text(log.clone()));
991150×
        }
992151×
        cbor::encode(&v)
993151×
    }
994194×
    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {
995194×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
996194×
        let m = map(&v, "refs")?;
997194×
        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?
998194×
            .try_into()
999194×
            .map_err(|_| schema("repo id: expected 16 bytes"))?;
1000194×
        let mut refs = Vec::new();
1001198×
        for r in o::array(field(m, 0, "refs")?, "refs")? {
1002198×
            let rm = map(r, "ref")?;
1003198×
            let ls = field(rm, 4, "last serial")?;
1004198×
            refs.push(RemoteRef {
1005198×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
1006198×
                state: o::oref(field(rm, 1, "state")?, "state")?,
1007198×
                heads: hashes_from(field(rm, 2, "heads")?)?,
1008198×
                attestation: opt_attest_from(field(rm, 3, "attestation")?)?,
1009198×
                last_serial: if ls.is_null() {
1010182×
                    None
1011
                } else {
101216×
                    Some(o::uint(ls, "last serial")?)
1013
                },
1014198×
                accepted: match rm.get(&5) {
10150×
                    Some(a) => opt_attest_from(a)?,
1016198×
                    None => None,
1017
                },
1018198×
                logged: match rm.get(&6) {
10191×
                    Some(l) => {
10201×
                        let a = o::array(l, "logged")?;
10211×
                        if a.len() != 3 {
10220×
                            return Err(schema("logged: expected [index, entry, proof]"));
10231×
                        }
10241×
                        let mut proof = Vec::new();
10252×
                        for h in o::array(&a[2], "proof")? {
10262×
                            proof.push(
10272×
                                o::bytes(h, "proof hash")?
10282×
                                    .try_into()
10292×
                                    .map_err(|_| schema("proof hash: expected 32 bytes"))?,
1030
                            );
1031
                        }
1032
                        Some(Logged {
10331×
                            index: o::uint(&a[0], "index")?,
10341×
                            entry: o::bytes(&a[1], "entry")?.to_vec(),
10351×
                            proof,
1036
                        })
1037
                    }
1038197×
                    None => None,
1039
                },
1040
            });
1041
        }
1042194×
        let policy = match m.get(&2) {
1043194×
            Some(p) => opt_policy_from(p)?,
10440×
            None => None,
1045
        };
1046194×
        let tags = match m.get(&3) {
1047194×
            Some(t) => tags_from(t)?,
10480×
            None => Vec::new(),
1049
        };
1050194×
        let mut history = Vec::new();
1051194×
        if let Some(h) = m.get(&4) {
1052194×
            for p in o::array(h, "history")? {
10531×
                history.push(policy_from(p)?);
1054
            }
10550×
        }
1056194×
        let default_bookmark = match m.get(&5) {
1057125×
            Some(Value::Text(t)) => Some(t.clone()),
105869×
            _ => None,
1059
        };
1060194×
        let log = match m.get(&6) {
10611×
            Some(Value::Text(t)) => Some(t.clone()),
1062193×
            _ => None,
1063
        };
1064194×
        Ok((
1065194×
            id,
1066194×
            RefsResponse {
1067194×
                refs,
1068194×
                policy,
1069194×
                tags,
1070194×
                history,
1071194×
                default_bookmark,
1072194×
                log,
1073194×
            },
1074194×
        ))
1075194×
    }
1076
107750×
    pub fn encode_negotiate(have: &[ChangeHash], want: &[ChangeHash]) -> Vec<u8> {
107850×
        cbor::encode(&Value::map([
107950×
            (0, hashes_value(have)),
108050×
            (1, hashes_value(want)),
108150×
        ]))
108250×
    }
108350×
    pub fn decode_negotiate(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Vec<ChangeHash>)> {
108450×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
108550×
        let m = map(&v, "negotiate")?;
1086
        Ok((
108750×
            hashes_from(field(m, 0, "have")?)?,
108850×
            hashes_from(field(m, 1, "want")?)?,
1089
        ))
109050×
    }
1091
1092
    /// `fetch` request: `{0: [changes], 1: max blob size or null}`.
109351×
    pub fn encode_fetch(hs: &[ChangeHash], max_blob: Option<u64>) -> Vec<u8> {
109451×
        cbor::encode(&Value::map([
109551×
            (0, hashes_value(hs)),
109651×
            (1, max_blob.map(Value::Uint).unwrap_or(Value::Null)),
109751×
        ]))
109851×
    }
109952×
    pub fn decode_fetch(bytes: &[u8]) -> Result<(Vec<ChangeHash>, Option<u64>)> {
110052×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
110152×
        let m = map(&v, "fetch")?;
110252×
        let max = match m.get(&1) {
110351×
            Some(x) if !x.is_null() => Some(o::uint(x, "max blob")?),
110450×
            _ => None,
1105
        };
110652×
        Ok((hashes_from(field(m, 0, "changes")?)?, max))
110752×
    }
1108
1109
    /// `shallow` request: `{0: [heads], 1: depth}`.
11101×
    pub fn encode_shallow(hs: &[ChangeHash], depth: u32) -> Vec<u8> {
11111×
        cbor::encode(&Value::map([
11121×
            (0, hashes_value(hs)),
11131×
            (1, Value::Uint(depth as u64)),
11141×
        ]))
11151×
    }
11161×
    pub fn decode_shallow(bytes: &[u8]) -> Result<(Vec<ChangeHash>, u32)> {
11171×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
11181×
        let m = map(&v, "shallow")?;
1119
        Ok((
11201×
            hashes_from(field(m, 0, "heads")?)?,
11211×
            o::u32_(field(m, 1, "depth")?, "depth")?,
1122
        ))
11231×
    }
1124
112552×
    pub fn encode_hashes(hs: &[ChangeHash]) -> Vec<u8> {
112652×
        cbor::encode(&Value::map([(0, hashes_value(hs))]))
112752×
    }
112851×
    pub fn decode_hashes(bytes: &[u8]) -> Result<Vec<ChangeHash>> {
112951×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
113051×
        hashes_from(field(map(&v, "hashes")?, 0, "hashes")?)
113151×
    }
1132
113335×
    pub fn encode_push(
113435×
        bundle: &Bundle,
113535×
        updates: &[RefUpdate],
113635×
        policy: Option<&SignedPolicy>,
113735×
        tags: &[TagRef],
113835×
    ) -> Vec<u8> {
113935×
        cbor::encode(&Value::map([
114035×
            (0, bundle_value(bundle)),
114135×
            (2, opt_policy_value(policy)),
114235×
            (3, tags_value(tags)),
1143
            (
1144
                1,
1145
                Value::Array(
114635×
                    updates
114735×
                        .iter()
114835×
                        .map(|u| {
114935×
                            Value::map([
115035×
                                (0, Value::Text(u.name.clone())),
115135×
                                (1, Value::Bytes(u.old_state.0.to_vec())),
115235×
                                (2, Value::Bytes(u.new_state.0.to_vec())),
115335×
                                (3, Value::Bool(u.force)),
115435×
                                (4, opt_attest_value(&u.attestation)),
115535×
                            ])
115635×
                        })
115735×
                        .collect(),
1158
                ),
1159
            ),
1160
        ]))
116135×
    }
1162
    /// A decoded push: bundle, bookmark updates, policy, tags.
1163
    pub type PushParts = (Bundle, Vec<RefUpdate>, Option<SignedPolicy>, Vec<TagRef>);
1164
116531×
    pub fn decode_push(bytes: &[u8]) -> Result<PushParts> {
116631×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
116731×
        let m = map(&v, "push")?;
116831×
        let bundle = bundle_from_value(field(m, 0, "bundle")?)?;
116931×
        let mut updates = Vec::new();
117031×
        for u in o::array(field(m, 1, "updates")?, "updates")? {
117131×
            let um = map(u, "update")?;
117231×
            updates.push(RefUpdate {
117331×
                name: o::text(field(um, 0, "name")?, "name")?.to_owned(),
117431×
                old_state: o::oref(field(um, 1, "old_state")?, "old_state")?,
117531×
                new_state: o::oref(field(um, 2, "new_state")?, "new_state")?,
117631×
                force: field(um, 3, "force")?
117731×
                    .as_bool()
117831×
                    .ok_or_else(|| schema("force: expected bool"))?,
117931×
                attestation: opt_attest_from(field(um, 4, "attestation")?)?,
1180
            });
1181
        }
118231×
        let policy = match m.get(&2) {
118331×
            Some(p) => opt_policy_from(p)?,
11840×
            None => None,
1185
        };
118631×
        let tags = match m.get(&3) {
118731×
            Some(t) => tags_from(t)?,
11880×
            None => Vec::new(),
1189
        };
119031×
        Ok((bundle, updates, policy, tags))
119131×
    }
1192
119331×
    pub fn encode_push_response(results: &[RefResult]) -> Vec<u8> {
119431×
        cbor::encode(&Value::map([(
1195
            0,
1196
            Value::Array(
119731×
                results
119831×
                    .iter()
119932×
                    .map(|r| {
120032×
                        Value::map([
120132×
                            (0, Value::Text(r.name.clone())),
120232×
                            (1, Value::Bool(r.ok)),
120332×
                            (2, Value::Text(r.reason.clone())),
120432×
                        ])
120532×
                    })
120631×
                    .collect(),
1207
            ),
1208
        )]))
120931×
    }
121031×
    pub fn decode_push_response(bytes: &[u8]) -> Result<Vec<RefResult>> {
121131×
        let v = cbor::decode(bytes).map_err(ObjectError::from)?;
121231×
        let m = map(&v, "push response")?;
121331×
        let mut out = Vec::new();
121432×
        for r in o::array(field(m, 0, "results")?, "results")? {
121532×
            let rm = map(r, "result")?;
121632×
            out.push(RefResult {
121732×
                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),
121832×
                ok: field(rm, 1, "ok")?
121932×
                    .as_bool()
122032×
                    .ok_or_else(|| schema("ok: expected bool"))?,
122132×
                reason: o::text(field(rm, 2, "reason")?, "reason")?.to_owned(),
1222
            });
1223
        }
122431×
        Ok(out)
122531×
    }
1226
}
1227
1228
#[cfg(test)]
1229
mod tests {
1230
    use super::*;
1231
    use crate::attest::AttestBody;
1232
    use crate::objects::Hlc;
1233
1234
    #[test]
12351×
    fn wire_round_trips() {
12361×
        let c = crate::objects::tests::sample_change();
12371×
        let sk = crate::identity::SecretKey {
12381×
            ed_seed: [1; 32],
12391×
            ml_seed: [2; 32],
12401×
        };
12411×
        let id = sk.identity("a", "b", 0);
12421×
        let key = id.hash();
12431×
        let sig = sk.sign(crate::identity::Domain::Change, &key, &c.hash());
12441×
        let seal = Attest::seal([3; 16], vec![c.hash()]);
12451×
        let seal_sig = sk.sign(crate::identity::Domain::Attest, &key, &seal.hash());
12461×
        let policy = Policy::single_root([3; 16], id.clone(), 99);
12471×
        let psig = sk.sign(crate::identity::Domain::Policy, &key, &policy.hash());
12481×
        let bundle = Bundle {
12491×
            changes: vec![(c.clone(), vec![sig.clone()])],
12501×
            attests: vec![SignedAttest {
12511×
                attest: seal.clone(),
12521×
                sigs: vec![seal_sig],
12531×
            }],
12541×
            objects: vec![
12551×
                (ObjectType::Identity, id.encode()),
12561×
                (ObjectType::Chunk, b"xyz".to_vec()),
12571×
            ],
12581×
            policies: vec![SignedPolicy {
12591×
                policy: policy.clone(),
12601×
                sigs: vec![psig.clone()],
12611×
            }],
12621×
            hashes: vec![],
12631×
            raw_changes: vec![],
12641×
            certificates: vec![],
12651×
        };
12661×
        let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap();
12671×
        assert_eq!(
1268
            d.hashes,
12691×
            vec![c.hash()],
1270
            "the receiver hashes what it was sent"
1271
        );
12721×
        assert_eq!(d.changes, bundle.changes);
12731×
        assert_eq!(d.attests, bundle.attests);
12741×
        assert_eq!(d.objects, bundle.objects);
12751×
        assert_eq!(d.policies, bundle.policies);
12761×
        assert!(d.certificates.is_empty());
1277
        // Certificates (keyless, oxygenhub#241) travel as field 4, only
1278
        // when there are any: a bundle without them encodes as before.
12791×
        let plain = wire::encode_bundle(&bundle);
12801×
        let mut with = bundle.clone();
12811×
        let issuer_sk = crate::identity::SecretKey::generate();
12821×
        let issuer = issuer_sk.identity("hub", "h@x", 1);
12831×
        with.certificates.push(crate::cert::SignedCertificate::sign(
12841×
            crate::cert::Certificate {
12851×
                key: id.clone(),
12861×
                issuer: issuer.hash(),
12871×
                subject: "agent:a@b".into(),
12881×
                claims: vec![],
12891×
                scope: crate::identity::Scope::CHANGES,
12901×
                not_before: 1,
12911×
                expires: 2,
12921×
                grant: None,
12931×
            },
12941×
            &issuer,
12951×
            &issuer_sk,
1296
        ));
12971×
        let encoded = wire::encode_bundle(&with);
12981×
        assert_ne!(encoded, plain);
12991×
        assert_eq!(
13001×
            wire::decode_bundle(&encoded).unwrap().certificates,
1301
            with.certificates
1302
        );
13031×
        with.certificates.clear();
13041×
        assert_eq!(wire::encode_bundle(&with), plain);
1305
13061×
        let att = Attest {
13071×
            repo: [3; 16],
13081×
            hlc: Hlc::default(),
13091×
            body: AttestBody::Ref {
13101×
                name: "main".into(),
13111×
                old_state: ObjectRef([1; 32]),
13121×
                new_state: ObjectRef([2; 32]),
13131×
                serial: 4,
13141×
            },
13151×
        };
13161×
        let refs = vec![
13171×
            RemoteRef {
13181×
                name: "main".into(),
13191×
                state: ObjectRef([2; 32]),
13201×
                heads: vec![c.hash()],
13211×
                attestation: Some(SignedAttest {
13221×
                    attest: att.clone(),
13231×
                    sigs: vec![sig.clone()],
13241×
                }),
13251×
                last_serial: Some(4),
13261×
                accepted: None,
13271×
                logged: None,
13281×
            },
13291×
            RemoteRef {
13301×
                name: "dev".into(),
13311×
                state: ObjectRef([5; 32]),
13321×
                heads: vec![],
13331×
                attestation: None,
13341×
                last_serial: None,
13351×
                accepted: None,
13361×
                logged: None,
13371×
            },
1338
        ];
13391×
        let tag = TagRef {
13401×
            name: "v1".into(),
13411×
            state: ObjectRef([2; 32]),
13421×
            heads: vec![c.hash()],
13431×
            attestation: None,
13441×
        };
13451×
        let mut resp = RefsResponse {
13461×
            refs,
13471×
            policy: Some(SignedPolicy {
13481×
                policy: policy.clone(),
13491×
                sigs: vec![psig.clone()],
13501×
            }),
13511×
            tags: vec![tag.clone()],
13521×
            history: vec![SignedPolicy {
13531×
                policy: policy.clone(),
13541×
                sigs: vec![],
13551×
            }],
13561×
            default_bookmark: Some("develop".into()),
13571×
            log: Some("hub.example/log\n2\nroot\n\n- sig\n".into()),
13581×
        };
13591×
        resp.refs[1].logged = Some(Logged {
13601×
            index: 41,
13611×
            entry: vec![1, 2, 3],
13621×
            proof: vec![[9; 32], [8; 32]],
13631×
        });
13641×
        let (id_back, back) = wire::decode_refs(&wire::encode_refs([3; 16], &resp)).unwrap();
13651×
        assert_eq!(back.log, resp.log);
13661×
        assert_eq!(back.refs[1].logged, resp.refs[1].logged);
13671×
        assert_eq!(back.refs[0].logged, None);
13681×
        assert_eq!(id_back, [3; 16]);
13691×
        assert_eq!(back.refs.len(), 2);
13701×
        assert_eq!(back.refs[0].attestation.as_ref().unwrap().attest, att);
13711×
        assert_eq!(back.refs[1].last_serial, None);
13721×
        assert_eq!(back.policy.as_ref().unwrap().policy, policy);
13731×
        assert_eq!(back.history, resp.history);
13741×
        assert_eq!(back.tags, vec![tag.clone()]);
1375
13761×
        let (h, w) = wire::decode_negotiate(&wire::encode_negotiate(
13771×
            &[ObjectRef([1; 32])],
13781×
            &[ObjectRef([2; 32])],
13791×
        ))
13801×
        .unwrap();
13811×
        assert_eq!((h, w), (vec![ObjectRef([1; 32])], vec![ObjectRef([2; 32])]));
13821×
        assert_eq!(
13831×
            wire::decode_hashes(&wire::encode_hashes(&[c.hash()])).unwrap(),
13841×
            vec![c.hash()]
1385
        );
13861×
        assert_eq!(
13871×
            wire::decode_fetch(&wire::encode_fetch(&[c.hash()], Some(4096))).unwrap(),
13881×
            (vec![c.hash()], Some(4096))
1389
        );
13901×
        assert_eq!(
13911×
            wire::decode_fetch(&wire::encode_fetch(&[], None)).unwrap(),
13921×
            (vec![], None)
1393
        );
13941×
        assert_eq!(
13951×
            wire::decode_shallow(&wire::encode_shallow(&[c.hash()], 3)).unwrap(),
13961×
            (vec![c.hash()], 3)
1397
        );
1398
        // An old-style fetch request without the filter field still decodes.
13991×
        assert_eq!(
14001×
            wire::decode_fetch(&wire::encode_hashes(&[c.hash()])).unwrap(),
14011×
            (vec![c.hash()], None)
1402
        );
1403
14041×
        let up = RefUpdate {
14051×
            name: "main".into(),
14061×
            old_state: ObjectRef([1; 32]),
14071×
            new_state: ObjectRef([2; 32]),
14081×
            force: true,
14091×
            attestation: Some(SignedAttest {
14101×
                attest: att,
14111×
                sigs: vec![sig],
14121×
            }),
14131×
        };
14141×
        let sp = SignedPolicy {
14151×
            policy,
14161×
            sigs: vec![psig],
14171×
        };
14181×
        let (b2, ups, pol, tags) = wire::decode_push(&wire::encode_push(
14191×
            &bundle,
14201×
            std::slice::from_ref(&up),
14211×
            Some(&sp),
14221×
            std::slice::from_ref(&tag),
14231×
        ))
14241×
        .unwrap();
14251×
        assert_eq!(pol, Some(sp));
14261×
        assert_eq!(tags, vec![tag]);
14271×
        assert_eq!(b2.changes.len(), 1);
14281×
        assert_eq!(ups[0].name, "main");
14291×
        assert!(ups[0].force);
14301×
        assert_eq!(ups[0].attestation, up.attestation);
14311×
        let res = vec![RefResult {
14321×
            name: "main".into(),
14331×
            ok: false,
14341×
            reason: "stale".into(),
14351×
        }];
14361×
        assert_eq!(
14371×
            wire::decode_push_response(&wire::encode_push_response(&res)).unwrap(),
1438
            res
1439
        );
14401×
        assert_eq!(
14411×
            wire::decode_capabilities(&wire::encode_capabilities([7; 16])).unwrap(),
1442
            [7; 16]
1443
        );
14441×
    }
1445
}